break
b3499adfca
Constraint: 切换 main 前必须保存当前七月分支全部项目进展，套餐生效提案仅属于 Iteration/7-11。 Rejected: 将七月套餐修复直接移植到 main | 两个分支的可靠投递架构不同。 Confidence: medium Scope-risk: broad Directive: 不得将本提交整体 cherry-pick 到 main；main 套餐热修必须基于其纯 Asynq 代码独立实施。 Tested: git diff --check；openspec validate fix-package-activation-starvation --strict。 Not-tested: 按用户要求未运行自动化测试；go build ./... 因当前审计改造中的 Enterprise 模型字面量和 role.recordFailure 参数类型错误未通过。
256 lines
9.6 KiB
Go
256 lines
9.6 KiB
Go
// Package outbox 提供公共 Outbox 的受控人工恢复用例。
|
|
package outbox
|
|
|
|
import (
|
|
"context"
|
|
stderrors "errors"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
|
|
"github.com/break/junhong_cmp_fiber/internal/model"
|
|
"github.com/break/junhong_cmp_fiber/pkg/auditfailure"
|
|
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
|
pkgerrors "github.com/break/junhong_cmp_fiber/pkg/errors"
|
|
)
|
|
|
|
// Operator 表示人工恢复操作者的授权快照。
|
|
type Operator struct {
|
|
ID uint
|
|
SuperAdmin bool
|
|
RequestID string
|
|
CorrelationID string
|
|
}
|
|
|
|
// RecoveryAudit 是交给统一 Audit Port 的安全恢复事实。
|
|
type RecoveryAudit struct {
|
|
OperatorID uint
|
|
OperationType string
|
|
Description string
|
|
EventIDs []string
|
|
Reason string
|
|
BatchID string
|
|
RequestID string
|
|
CorrelationID string
|
|
Events []RecoveryEventAudit
|
|
Result string
|
|
ErrorCode string
|
|
ErrorSummary string
|
|
}
|
|
|
|
// RecoveryEventAudit 是一次人工恢复中单个 Outbox 事件的身份和状态变化。
|
|
type RecoveryEventAudit struct {
|
|
ID uint
|
|
EventID string
|
|
EventType string
|
|
AggregateType string
|
|
AggregateID string
|
|
ResourceType string
|
|
ResourceID string
|
|
BusinessKey string
|
|
BeforeStatus int
|
|
AfterStatus int
|
|
BeforeNextAttempt time.Time
|
|
AfterNextAttempt time.Time
|
|
BeforeLeaseOwner *string
|
|
BeforeLeaseExpires *time.Time
|
|
AfterLeaseOwner *string
|
|
AfterLeaseExpires *time.Time
|
|
}
|
|
|
|
// AuditWriter 是 tech-global-audit 提供实现的统一审计接缝。
|
|
type AuditWriter interface {
|
|
WriteRecovery(ctx context.Context, tx *gorm.DB, audit RecoveryAudit) error
|
|
}
|
|
|
|
// RecoveryService 执行选择性重放和过期租约释放。
|
|
type RecoveryService struct {
|
|
db *gorm.DB
|
|
audit AuditWriter
|
|
now func() time.Time
|
|
}
|
|
|
|
// NewRecoveryService 创建受控恢复用例；审计接缝不可缺失。
|
|
func NewRecoveryService(db *gorm.DB, audit AuditWriter, now func() time.Time) (*RecoveryService, error) {
|
|
if db == nil || audit == nil {
|
|
return nil, stderrors.New("Outbox 恢复必须配置数据库和统一审计接缝")
|
|
}
|
|
if now == nil {
|
|
now = time.Now
|
|
}
|
|
return &RecoveryService{db: db, audit: audit, now: now}, nil
|
|
}
|
|
|
|
// Replay 只重放明确选择的最终失败或租约过期事件，并保留原始内容和身份。
|
|
func (s *RecoveryService) Replay(ctx context.Context, operator Operator, ids []uint, reason string) (string, error) {
|
|
if err := validateCommand(operator, ids, reason); err != nil {
|
|
return "", err
|
|
}
|
|
batchID := uuid.NewString()
|
|
now := s.now().UTC()
|
|
failureAudit := RecoveryAudit{
|
|
OperatorID: operator.ID, OperationType: constants.AuditOperationOutboxReplay,
|
|
Description: "人工重放 Outbox 事件失败", Reason: reason, BatchID: batchID,
|
|
RequestID: operator.RequestID, CorrelationID: operator.CorrelationID,
|
|
}
|
|
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
events, err := loadSelectedForUpdate(tx, ids)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
failureAudit.EventIDs, failureAudit.Events = unchangedRecoveryAudit(events)
|
|
if len(events) != len(ids) {
|
|
failureAudit.Events = nil
|
|
return pkgerrors.New(pkgerrors.CodeInvalidStatus, "选择的事件不存在或状态不允许重放")
|
|
}
|
|
eventIDs := make([]string, 0, len(events))
|
|
auditEvents := make([]RecoveryEventAudit, 0, len(events))
|
|
for _, event := range events {
|
|
allowed := event.Status == constants.OutboxStatusFailed ||
|
|
(event.Status == constants.OutboxStatusDelivering && event.LeaseExpiresAt != nil && !event.LeaseExpiresAt.After(now))
|
|
if !allowed {
|
|
failureAudit.Result = constants.AuditResultDenied
|
|
failureAudit.ErrorCode = strconv.Itoa(pkgerrors.CodeInvalidStatus)
|
|
failureAudit.ErrorSummary = "选择的事件不存在或状态不允许重放"
|
|
return pkgerrors.New(pkgerrors.CodeInvalidStatus, "选择的事件不存在或状态不允许重放")
|
|
}
|
|
eventIDs = append(eventIDs, event.EventID)
|
|
auditEvents = append(auditEvents, recoveryEventAudit(event, now))
|
|
}
|
|
failureAudit.Result = constants.AuditResultFailed
|
|
failureAudit.ErrorCode = strconv.Itoa(pkgerrors.CodeDatabaseError)
|
|
failureAudit.ErrorSummary = "Outbox 人工重放事务已回滚"
|
|
result := tx.Model(&model.OutboxEvent{}).Where("id IN ?", ids).Updates(map[string]any{
|
|
"status": constants.OutboxStatusPending, "next_attempt_at": now,
|
|
"lease_owner": nil, "lease_expires_at": nil, "updated_at": now,
|
|
})
|
|
if result.Error != nil {
|
|
return result.Error
|
|
}
|
|
return s.audit.WriteRecovery(ctx, tx, RecoveryAudit{
|
|
OperatorID: operator.ID, OperationType: constants.AuditOperationOutboxReplay, Description: "人工重放 Outbox 事件",
|
|
EventIDs: eventIDs, Reason: reason, BatchID: batchID,
|
|
RequestID: operator.RequestID, CorrelationID: operator.CorrelationID, Events: auditEvents,
|
|
})
|
|
})
|
|
if err != nil {
|
|
s.recordFailure(ctx, failureAudit)
|
|
}
|
|
return batchID, err
|
|
}
|
|
|
|
// ReleaseExpiredLeases 只释放明确选择且已经过期的投递租约。
|
|
func (s *RecoveryService) ReleaseExpiredLeases(ctx context.Context, operator Operator, ids []uint, reason string) (string, error) {
|
|
if err := validateCommand(operator, ids, reason); err != nil {
|
|
return "", err
|
|
}
|
|
batchID := uuid.NewString()
|
|
now := s.now().UTC()
|
|
failureAudit := RecoveryAudit{
|
|
OperatorID: operator.ID, OperationType: constants.AuditOperationOutboxReleaseExpiredLease,
|
|
Description: "人工释放 Outbox 过期租约失败", Reason: reason, BatchID: batchID,
|
|
RequestID: operator.RequestID, CorrelationID: operator.CorrelationID,
|
|
}
|
|
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
events, err := loadSelectedForUpdate(tx, ids)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
failureAudit.EventIDs, failureAudit.Events = unchangedRecoveryAudit(events)
|
|
if len(events) != len(ids) {
|
|
failureAudit.Events = nil
|
|
return pkgerrors.New(pkgerrors.CodeInvalidStatus, "选择的租约不存在或仍然有效")
|
|
}
|
|
eventIDs := make([]string, 0, len(events))
|
|
auditEvents := make([]RecoveryEventAudit, 0, len(events))
|
|
for _, event := range events {
|
|
if event.Status != constants.OutboxStatusDelivering || event.LeaseExpiresAt == nil || event.LeaseExpiresAt.After(now) {
|
|
failureAudit.Result = constants.AuditResultDenied
|
|
failureAudit.ErrorCode = strconv.Itoa(pkgerrors.CodeInvalidStatus)
|
|
failureAudit.ErrorSummary = "选择的租约不存在或仍然有效"
|
|
return pkgerrors.New(pkgerrors.CodeInvalidStatus, "选择的租约不存在或仍然有效")
|
|
}
|
|
eventIDs = append(eventIDs, event.EventID)
|
|
auditEvents = append(auditEvents, recoveryEventAudit(event, now))
|
|
}
|
|
failureAudit.Result = constants.AuditResultFailed
|
|
failureAudit.ErrorCode = strconv.Itoa(pkgerrors.CodeDatabaseError)
|
|
failureAudit.ErrorSummary = "Outbox 过期租约释放事务已回滚"
|
|
result := tx.Model(&model.OutboxEvent{}).Where("id IN ? AND status = ? AND lease_expires_at <= ?", ids, constants.OutboxStatusDelivering, now).
|
|
Updates(map[string]any{
|
|
"status": constants.OutboxStatusPending, "next_attempt_at": now,
|
|
"lease_owner": nil, "lease_expires_at": nil, "updated_at": now,
|
|
})
|
|
if result.Error != nil {
|
|
return result.Error
|
|
}
|
|
return s.audit.WriteRecovery(ctx, tx, RecoveryAudit{
|
|
OperatorID: operator.ID, OperationType: constants.AuditOperationOutboxReleaseExpiredLease, Description: "人工释放 Outbox 过期租约",
|
|
EventIDs: eventIDs, Reason: reason, BatchID: batchID,
|
|
RequestID: operator.RequestID, CorrelationID: operator.CorrelationID, Events: auditEvents,
|
|
})
|
|
})
|
|
if err != nil {
|
|
s.recordFailure(ctx, failureAudit)
|
|
}
|
|
return batchID, err
|
|
}
|
|
|
|
func (s *RecoveryService) recordFailure(ctx context.Context, audit RecoveryAudit) {
|
|
if len(audit.Events) == 0 || audit.Result == "" {
|
|
return
|
|
}
|
|
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
return s.audit.WriteRecovery(ctx, tx, audit)
|
|
})
|
|
if err != nil {
|
|
auditfailure.RecordSecondaryWriteFailure(
|
|
audit.OperationType, audit.Events[0].EventID, audit.RequestID, audit.CorrelationID, audit.ErrorCode, err,
|
|
)
|
|
}
|
|
}
|
|
|
|
func validateCommand(operator Operator, ids []uint, reason string) error {
|
|
if !operator.SuperAdmin {
|
|
return pkgerrors.New(pkgerrors.CodeForbidden)
|
|
}
|
|
if operator.ID == 0 || len(ids) == 0 || reason == "" {
|
|
return pkgerrors.New(pkgerrors.CodeInvalidParam)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func loadSelectedForUpdate(tx *gorm.DB, ids []uint) ([]model.OutboxEvent, error) {
|
|
var events []model.OutboxEvent
|
|
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id IN ?", ids).Order("id ASC").Find(&events).Error
|
|
return events, err
|
|
}
|
|
|
|
func recoveryEventAudit(event model.OutboxEvent, nextAttempt time.Time) RecoveryEventAudit {
|
|
return RecoveryEventAudit{
|
|
ID: event.ID, EventID: event.EventID, EventType: event.EventType,
|
|
AggregateType: event.AggregateType, AggregateID: event.AggregateID,
|
|
ResourceType: event.ResourceType, ResourceID: event.ResourceID, BusinessKey: event.BusinessKey,
|
|
BeforeStatus: event.Status, AfterStatus: constants.OutboxStatusPending,
|
|
BeforeNextAttempt: event.NextAttemptAt, AfterNextAttempt: nextAttempt,
|
|
BeforeLeaseOwner: event.LeaseOwner, BeforeLeaseExpires: event.LeaseExpiresAt,
|
|
}
|
|
}
|
|
|
|
func unchangedRecoveryAudit(events []model.OutboxEvent) ([]string, []RecoveryEventAudit) {
|
|
eventIDs := make([]string, 0, len(events))
|
|
auditEvents := make([]RecoveryEventAudit, 0, len(events))
|
|
for _, event := range events {
|
|
eventIDs = append(eventIDs, event.EventID)
|
|
auditEvent := recoveryEventAudit(event, event.NextAttemptAt)
|
|
auditEvent.AfterStatus = event.Status
|
|
auditEvent.AfterLeaseOwner = event.LeaseOwner
|
|
auditEvent.AfterLeaseExpires = event.LeaseExpiresAt
|
|
auditEvents = append(auditEvents, auditEvent)
|
|
}
|
|
return eventIDs, auditEvents
|
|
}
|