固化七月迭代审计治理进展以隔离线上热修
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 参数类型错误未通过。
This commit is contained in:
@@ -4,6 +4,7 @@ package outbox
|
||||
import (
|
||||
"context"
|
||||
stderrors "errors"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
@@ -11,6 +12,7 @@ import (
|
||||
"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"
|
||||
)
|
||||
@@ -33,6 +35,30 @@ type RecoveryAudit struct {
|
||||
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 提供实现的统一审计接缝。
|
||||
@@ -65,23 +91,38 @@ func (s *RecoveryService) Replay(ctx context.Context, operator Operator, ids []u
|
||||
}
|
||||
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,
|
||||
@@ -90,11 +131,14 @@ func (s *RecoveryService) Replay(ctx context.Context, operator Operator, ids []u
|
||||
return result.Error
|
||||
}
|
||||
return s.audit.WriteRecovery(ctx, tx, RecoveryAudit{
|
||||
OperatorID: operator.ID, OperationType: "outbox_replay", Description: "人工重放 Outbox 事件",
|
||||
OperatorID: operator.ID, OperationType: constants.AuditOperationOutboxReplay, Description: "人工重放 Outbox 事件",
|
||||
EventIDs: eventIDs, Reason: reason, BatchID: batchID,
|
||||
RequestID: operator.RequestID, CorrelationID: operator.CorrelationID,
|
||||
RequestID: operator.RequestID, CorrelationID: operator.CorrelationID, Events: auditEvents,
|
||||
})
|
||||
})
|
||||
if err != nil {
|
||||
s.recordFailure(ctx, failureAudit)
|
||||
}
|
||||
return batchID, err
|
||||
}
|
||||
|
||||
@@ -105,21 +149,36 @@ func (s *RecoveryService) ReleaseExpiredLeases(ctx context.Context, operator Ope
|
||||
}
|
||||
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,
|
||||
@@ -129,14 +188,31 @@ func (s *RecoveryService) ReleaseExpiredLeases(ctx context.Context, operator Ope
|
||||
return result.Error
|
||||
}
|
||||
return s.audit.WriteRecovery(ctx, tx, RecoveryAudit{
|
||||
OperatorID: operator.ID, OperationType: "outbox_release_expired_lease", Description: "人工释放 Outbox 过期租约",
|
||||
OperatorID: operator.ID, OperationType: constants.AuditOperationOutboxReleaseExpiredLease, Description: "人工释放 Outbox 过期租约",
|
||||
EventIDs: eventIDs, Reason: reason, BatchID: batchID,
|
||||
RequestID: operator.RequestID, CorrelationID: operator.CorrelationID,
|
||||
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)
|
||||
@@ -152,3 +228,28 @@ func loadSelectedForUpdate(tx *gorm.DB, ids []uint) ([]model.OutboxEvent, error)
|
||||
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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user