diff --git a/cmd/audit-retention-simulate/main.go b/cmd/audit-retention-simulate/main.go index 3923cea..f627774 100644 --- a/cmd/audit-retention-simulate/main.go +++ b/cmd/audit-retention-simulate/main.go @@ -141,9 +141,24 @@ func run(ctx context.Context) error { if err := collectAfterCleanup(ctx, db, monthStart, monthEnd, &summary); err != nil { return err } - if summary.TargetRowsAfterCleanup != 0 || summary.BoundaryRowsAfterCleanup != 6 || summary.CleanupMarkersAfter != int64(summary.Days*2) { + if summary.TargetRowsAfterCleanup != 1 || summary.BoundaryRowsAfterCleanup != 6 || summary.CleanupMarkersAfter != int64(summary.Days*2) { return fmt.Errorf("清理范围复核失败: target=%d boundary=%d markers=%d", summary.TargetRowsAfterCleanup, summary.BoundaryRowsAfterCleanup, summary.CleanupMarkersAfter) } + if err := db.WithContext(ctx).Model(&model.IntegrationLog{}). + Where("integration_id = ?", "int_"+simulationPrefix+"-"+monthStart.Format("20060102")). + Updates(map[string]any{"result": constants.IntegrationResultSuccess, "updated_at": time.Now()}).Error; err != nil { + return err + } + if _, err := service.RetainDate(ctx, monthStart, true); err != nil { + return fmt.Errorf("pending 终结后的续跑清理演练失败: %w", err) + } + var remaining int64 + if err := db.WithContext(ctx).Model(&model.IntegrationLog{}).Where("created_at >= ? AND created_at < ?", monthStart, monthEnd).Count(&remaining).Error; err != nil { + return err + } + if remaining != 0 { + return fmt.Errorf("pending 终结后的续跑清理未完成，剩余 %d 条记录", remaining) + } if err := syncRetentionLogger(); err != nil { return fmt.Errorf("刷新日留存仿真日志失败: %w", err) } @@ -270,16 +285,6 @@ func archiveMonth(ctx context.Context, db *gorm.DB, service *auditarchive.Servic if err := service.ArchiveDate(ctx, start); err != nil { return fmt.Errorf("Audit 故障重试演练失败: %w", err) } - if err := db.WithContext(ctx).Model(&model.IntegrationLog{}). - Where("integration_id = ?", "int_"+simulationPrefix+"-"+start.Format("20060102")). - Updates(map[string]any{"result": constants.IntegrationResultSuccess, "updated_at": time.Now()}).Error; err != nil { - return err - } - for date := start; date.Before(end); date = date.AddDate(0, 0, 1) { - if err := service.FinalizeIntegrationDate(ctx, date); err != nil { - return err - } - } return nil } diff --git a/internal/application/auditarchive/integration.go b/internal/application/auditarchive/integration.go index 0bcdf79..c57f6de 100644 --- a/internal/application/auditarchive/integration.go +++ b/internal/application/auditarchive/integration.go @@ -77,20 +77,20 @@ func (s *Service) archiveIntegrationDate(ctx context.Context, archiveDate time.T if err != nil { return err } + if final && run.Status == constants.ArchiveStatusSuccess && run.IsFinal && run.CleanedAt != nil { + terminalCount, countErr := s.integrationTerminalCount(ctx, start, end) + if countErr != nil { + return countErr + } + if terminalCount == 0 { + return nil + } + } file, err := s.buildIntegrationArchiveFile(ctx, start, end) if err != nil { return err } defer os.Remove(file.path) - if final { - pending, pendingErr := s.integrationPendingCount(ctx, start, end) - if pendingErr != nil { - return pendingErr - } - if pending > 0 { - return fmt.Errorf("仍有 %d 条 pending Integration Log，无法形成最终归档", pending) - } - } if run.Status == constants.ArchiveStatusSuccess && run.RecordCount == file.recordCount && run.SHA256 == file.sha256 { valid, validateErr := s.validateIntegrationRun(ctx, run) if validateErr == nil && valid && (!final || run.IsFinal) { @@ -146,7 +146,8 @@ func (s *Service) acquireIntegrationRun(ctx context.Context, run *model.LogArchi Where("id = ? AND (status <> ? OR updated_at < ?)", run.ID, constants.ArchiveStatusRunning, now.Add(-3*time.Hour)). Updates(map[string]any{ "status": constants.ArchiveStatusRunning, "revision": revision, "is_final": false, - "attempt_count": gorm.Expr("attempt_count + 1"), "error_summary": "", "completed_at": nil, "updated_at": now, + "attempt_count": gorm.Expr("attempt_count + 1"), "error_summary": "", "completed_at": nil, + "cleanup_started_at": nil, "cleaned_at": nil, "updated_at": now, }) if result.Error != nil { return false, fmt.Errorf("锁定 Integration Log 归档任务失败: %w", result.Error) @@ -301,16 +302,6 @@ func (s *Service) integrationRecordCount(ctx context.Context, start, end time.Ti return count, nil } -func (s *Service) integrationPendingCount(ctx context.Context, start, end time.Time) (int64, error) { - var count int64 - if err := s.db.WithContext(ctx).Model(&model.IntegrationLog{}). - Where("created_at >= ? AND created_at < ? AND result = ?", start, end, constants.IntegrationResultPending). - Count(&count).Error; err != nil { - return 0, fmt.Errorf("统计 pending Integration Log 失败: %w", err) - } - return count, nil -} - func integrationObjectKeys(date time.Time, revision int) (string, string) { prefix := fmt.Sprintf("audit-archive/v1/%04d/%02d/%02d", date.Year(), date.Month(), date.Day()) name := fmt.Sprintf("integration-logs-%s-r%d", date.Format(time.DateOnly), revision) diff --git a/internal/application/auditarchive/retention.go b/internal/application/auditarchive/retention.go index a35294a..0b64553 100644 --- a/internal/application/auditarchive/retention.go +++ b/internal/application/auditarchive/retention.go @@ -7,7 +7,6 @@ import ( "io" "os" "strconv" - "strings" "time" "github.com/bytedance/sonic" @@ -50,11 +49,6 @@ func (e *RetentionBlockedError) Error() string { } func (e *RetentionBlockedError) Unwrap() error { return e.Err } -type retentionRuns struct { - audit *model.LogArchiveRun - integration *model.LogArchiveRun -} - // RetainPendingDays 从最早仍在线的日期连续处理至昨天。cleanup 为 false 时只读校验。 func (s *Service) RetainPendingDays(ctx context.Context, cleanup bool) ([]RetentionResult, error) { if s.db == nil || s.store == nil { @@ -65,19 +59,39 @@ func (s *Service) RetainPendingDays(ctx context.Context, cleanup bool) ([]Retent return nil, err } results := make([]RetentionResult, 0) + var auditBlocked, integrationBlocked *RetentionBlockedError for date := start; date.Before(end); date = date.AddDate(0, 0, 1) { - result, err := s.retainDate(ctx, date, cleanup) - if err != nil { - source := "retention" - if strings.HasPrefix(err.Error(), "Audit") { - source = constants.AuditArchiveSource + startedAt := time.Now() + result := RetentionResult{ArchiveDate: date.Format(time.DateOnly)} + if auditBlocked == nil { + auditResult, auditErr := s.retainAuditDate(ctx, date, cleanup) + if auditErr != nil { + auditBlocked = &RetentionBlockedError{ArchiveDate: result.ArchiveDate, Source: constants.AuditArchiveSource, Err: auditErr} + } else { + result.EventCount, result.ResourceCount = auditResult.EventCount, auditResult.ResourceCount + result.ManifestKeys = append(result.ManifestKeys, auditResult.ManifestKeys...) } - if strings.HasPrefix(err.Error(), "Integration") { - source = constants.IntegrationArchiveSource - } - return results, &RetentionBlockedError{ArchiveDate: date.Format(time.DateOnly), Source: source, Err: err} } - results = append(results, result) + if integrationBlocked == nil { + integrationResult, integrationErr := s.retainIntegrationDate(ctx, date, cleanup) + if integrationErr != nil { + integrationBlocked = &RetentionBlockedError{ArchiveDate: result.ArchiveDate, Source: constants.IntegrationArchiveSource, Err: integrationErr} + } else { + result.IntegrationCount = integrationResult.IntegrationCount + result.ManifestKeys = append(result.ManifestKeys, integrationResult.ManifestKeys...) + } + } + result.EstimatedBatches = estimatedRetentionBatches(result) + result.Duration = time.Since(startedAt) + if auditBlocked == nil || integrationBlocked == nil { + results = append(results, result) + } + } + if auditBlocked != nil { + return results, auditBlocked + } + if integrationBlocked != nil { + return results, integrationBlocked } return results, nil } @@ -91,9 +105,16 @@ func (s *Service) pendingRetentionRange(ctx context.Context) (time.Time, time.Ti today := time.Now().In(s.location) end := time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, s.location) var earliest *time.Time - for _, item := range []struct{ table, column string }{{"tb_audit_event", "created_at"}, {"tb_integration_log", "created_at"}} { + for _, item := range []struct{ table, column, condition string }{ + {"tb_audit_event", "created_at", ""}, + {"tb_integration_log", "created_at", "result <> 'pending'"}, + } { + query := s.db.WithContext(ctx).Table(item.table).Select("MIN(" + item.column + ")") + if item.condition != "" { + query = query.Where(item.condition) + } var value *time.Time - if err := s.db.WithContext(ctx).Table(item.table).Select("MIN(" + item.column + ")").Scan(&value).Error; err != nil { + if err := query.Scan(&value).Error; err != nil { return time.Time{}, time.Time{}, fmt.Errorf("查询日留存起点失败: %w", err) } if value != nil && (earliest == nil || value.Before(*earliest)) { @@ -110,20 +131,52 @@ func (s *Service) pendingRetentionRange(ctx context.Context) (time.Time, time.Ti func (s *Service) retainDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) { startedAt := time.Now() - var existing []model.LogArchiveRun - if err := s.db.WithContext(ctx).Where("source IN ? AND archive_date = ? AND instance_id = ?", []string{constants.AuditArchiveSource, constants.IntegrationArchiveSource}, date.Format(time.DateOnly), s.instanceID).Find(&existing).Error; err != nil { - return RetentionResult{}, fmt.Errorf("读取日归档账本失败: %w", err) + auditResult, err := s.retainAuditDate(ctx, date, cleanup) + if err != nil { + return RetentionResult{}, err } - statuses := map[string]string{} - for _, run := range existing { - statuses[run.Source] = run.Status + integrationResult, err := s.retainIntegrationDate(ctx, date, cleanup) + if err != nil { + return RetentionResult{}, err } - if statuses[constants.AuditArchiveSource] != constants.ArchiveStatusSuccess { + result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), EventCount: auditResult.EventCount, ResourceCount: auditResult.ResourceCount, IntegrationCount: integrationResult.IntegrationCount, ManifestKeys: append(auditResult.ManifestKeys, integrationResult.ManifestKeys...)} + result.EstimatedBatches = estimatedRetentionBatches(result) + result.Duration = time.Since(startedAt) + return result, nil +} + +func (s *Service) retainAuditDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) { + run, err := s.retentionRun(ctx, date, constants.AuditArchiveSource) + if err != nil { + return RetentionResult{}, err + } + if run == nil || run.Status != constants.ArchiveStatusSuccess { if err := s.ArchiveDate(ctx, date); err != nil { return RetentionResult{}, fmt.Errorf("Audit 归档失败: %w", err) } + run, err = s.retentionRun(ctx, date, constants.AuditArchiveSource) + if err != nil { + return RetentionResult{}, err + } } - if statuses[constants.IntegrationArchiveSource] != constants.ArchiveStatusSuccess { + if err := s.validateAuditRetentionDay(ctx, date, run); err != nil { + return RetentionResult{}, fmt.Errorf("Audit 完整性校验失败: %w", err) + } + result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), EventCount: run.EventCount, ResourceCount: run.ResourceCount, ManifestKeys: []string{run.ManifestKey}} + if cleanup { + if err := s.cleanupAuditDate(ctx, date, run); err != nil { + return result, err + } + } + return result, nil +} + +func (s *Service) retainIntegrationDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) { + run, err := s.retentionRun(ctx, date, constants.IntegrationArchiveSource) + if err != nil { + return RetentionResult{}, err + } + if run == nil || run.Status != constants.ArchiveStatusSuccess { if err := s.ArchiveIntegrationDate(ctx, date); err != nil { return RetentionResult{}, fmt.Errorf("Integration Log 归档失败: %w", err) } @@ -131,48 +184,31 @@ func (s *Service) retainDate(ctx context.Context, date time.Time, cleanup bool) if err := s.FinalizeIntegrationDate(ctx, date); err != nil { return RetentionResult{}, fmt.Errorf("Integration Log 最终归档失败: %w", err) } - runs, err := s.loadRetentionRuns(ctx, date) + run, err = s.retentionRun(ctx, date, constants.IntegrationArchiveSource) if err != nil { return RetentionResult{}, err } - if err := s.validateAuditRetentionDay(ctx, date, runs.audit); err != nil { - return RetentionResult{}, fmt.Errorf("Audit 完整性校验失败: %w", err) - } - if err := s.validateIntegrationRetentionDay(ctx, date, runs.integration); err != nil { + if err := s.validateIntegrationRetentionDay(ctx, date, run); err != nil { return RetentionResult{}, fmt.Errorf("Integration Log 完整性校验失败: %w", err) } - result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), EventCount: runs.audit.EventCount, ResourceCount: runs.audit.ResourceCount, IntegrationCount: runs.integration.RecordCount, ManifestKeys: []string{runs.audit.ManifestKey, runs.integration.ManifestKey}} - result.EstimatedBatches = estimatedRetentionBatches(result) + result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), IntegrationCount: run.RecordCount, ManifestKeys: []string{run.ManifestKey}} if cleanup { - if err := s.cleanupAuditDate(ctx, date, runs.audit); err != nil { - return result, err - } - if err := s.cleanupIntegrationDate(ctx, date, runs.integration); err != nil { + if err := s.cleanupIntegrationDate(ctx, date, run); err != nil { return result, err } } - result.Duration = time.Since(startedAt) return result, nil } -func (s *Service) loadRetentionRuns(ctx context.Context, date time.Time) (retentionRuns, error) { - var rows []model.LogArchiveRun - if err := s.db.WithContext(ctx).Where("source IN ? AND archive_date = ? AND instance_id = ?", []string{constants.AuditArchiveSource, constants.IntegrationArchiveSource}, date.Format(time.DateOnly), s.instanceID).Find(&rows).Error; err != nil { - return retentionRuns{}, fmt.Errorf("读取日归档账本失败: %w", err) +func (s *Service) retentionRun(ctx context.Context, date time.Time, source string) (*model.LogArchiveRun, error) { + var runs []model.LogArchiveRun + if err := s.db.WithContext(ctx).Where("source = ? AND archive_date = ? AND instance_id = ?", source, date.Format(time.DateOnly), s.instanceID).Find(&runs).Error; err != nil { + return nil, fmt.Errorf("读取日归档账本失败: %w", err) } - runs := retentionRuns{} - for index := range rows { - switch rows[index].Source { - case constants.AuditArchiveSource: - runs.audit = &rows[index] - case constants.IntegrationArchiveSource: - runs.integration = &rows[index] - } + if len(runs) == 0 { + return nil, nil } - if runs.audit == nil || runs.integration == nil { - return retentionRuns{}, fmt.Errorf("Audit 或 Integration Log 归档账本缺失") - } - return runs, nil + return &runs[0], nil } func (s *Service) validateAuditRetentionDay(ctx context.Context, date time.Time, run *model.LogArchiveRun) error { @@ -200,13 +236,17 @@ func (s *Service) validateIntegrationRetentionDay(ctx context.Context, date time if err != nil { return err } - if run.CleanedAt != nil && count != 0 { - return fmt.Errorf("已标记清理完成但数据库仍有 %d 条记录", count) + terminalCount, err := s.integrationTerminalCount(ctx, run.RangeStart, run.RangeEnd) + if err != nil { + return err + } + if run.CleanedAt != nil && terminalCount != 0 { + return fmt.Errorf("已标记清理完成但数据库仍有 %d 条终态记录", terminalCount) } if run.CleanupStartedAt != nil && count > run.RecordCount { return fmt.Errorf("续跑窗口记录数超过最终归档数量") } - if run.CleanupStartedAt == nil { + if terminalCount > 0 && run.CleanupStartedAt == nil { file, err := s.buildIntegrationArchiveFile(ctx, run.RangeStart, run.RangeEnd) if err != nil { return err @@ -348,7 +388,11 @@ func (s *Service) cleanupAuditDate(ctx context.Context, date time.Time, run *mod return s.markCleaned(ctx, constants.AuditArchiveSource, date) } func (s *Service) cleanupIntegrationDate(ctx context.Context, date time.Time, run *model.LogArchiveRun) error { - if run.CleanedAt != nil { + terminalCount, err := s.integrationTerminalCount(ctx, date, date.AddDate(0, 0, 1)) + if err != nil { + return err + } + if run.CleanedAt != nil && terminalCount == 0 { return nil } if err := s.markCleanupStarted(ctx, constants.IntegrationArchiveSource, date); err != nil { @@ -356,7 +400,7 @@ func (s *Service) cleanupIntegrationDate(ctx context.Context, date time.Time, ru } end := date.AddDate(0, 0, 1) for { - subquery := s.db.Model(&model.IntegrationLog{}).Select("id").Where("created_at >= ? AND created_at < ?", date, end).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize) + subquery := s.db.Model(&model.IntegrationLog{}).Select("id").Where("created_at >= ? AND created_at < ? AND result <> ?", date, end, constants.IntegrationResultPending).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize) deleted := s.db.WithContext(ctx).Where("id IN (?)", subquery).Delete(&model.IntegrationLog{}) if deleted.Error != nil { return fmt.Errorf("分批物理删除 Integration Log 失败: %w", deleted.Error) @@ -367,6 +411,16 @@ func (s *Service) cleanupIntegrationDate(ctx context.Context, date time.Time, ru } return s.markCleaned(ctx, constants.IntegrationArchiveSource, date) } +func (s *Service) integrationTerminalCount(ctx context.Context, start, end time.Time) (int64, error) { + var count int64 + if err := s.db.WithContext(ctx).Model(&model.IntegrationLog{}). + Where("created_at >= ? AND created_at < ? AND result <> ?", start, end, constants.IntegrationResultPending). + Count(&count).Error; err != nil { + return 0, fmt.Errorf("统计终态 Integration Log 失败: %w", err) + } + return count, nil +} + func (s *Service) deleteAuditResources(ctx context.Context, start, end time.Time) error { for { subquery := s.db.Model(&model.AuditEventResource{}).Select("tb_audit_event_resource.id").Joins("JOIN tb_audit_event ON tb_audit_event.id = tb_audit_event_resource.audit_event_id").Where("tb_audit_event.created_at >= ? AND tb_audit_event.created_at < ?", start, end).Order("tb_audit_event_resource.id ASC").Limit(constants.AuditRetentionDeleteBatchSize) diff --git a/internal/task/asset_package_batch_order.go b/internal/task/asset_package_batch_order.go index 9f13102..d837361 100644 --- a/internal/task/asset_package_batch_order.go +++ b/internal/task/asset_package_batch_order.go @@ -121,19 +121,13 @@ func (h *AssetPackageBatchOrderHandler) finishBatchOrderTask(ctx context.Context return err } rootID := audit.TaskEventID(constants.AuditResourceAssetPackageBatchOrderTask, taskRecord.ID, "completed") - var childCount int64 - if err := tx.WithContext(ctx).Model(&model.AuditEvent{}). - Where("parent_event_id = ? AND action_code = ?", rootID, constants.AuditActionOrderCreated). - Count(&childCount).Error; err != nil { - return err - } - result := batchAuditResult(int(childCount), failCount) + result := batchAuditResult(successCount, failCount) return h.auditWriter.WriteTask(ctx, tx, audit.TaskInput{ EventID: rootID, ActionCode: constants.AuditActionAssetPackageBatchOrderTaskCompleted, Summary: "完成资产套餐批量订购任务", TaskID: taskRecord.ID, TaskNo: taskRecord.TaskNo, Result: result, CorrelationID: taskRecord.TaskNo, ParentEventID: audit.TaskEventID(constants.AuditResourceAssetPackageBatchOrderTask, taskRecord.ID, "created"), - BatchTotal: len(items), SuccessCount: int(childCount), FailCount: failCount, + BatchTotal: len(items), SuccessCount: successCount, FailCount: failCount, IdentitySnapshot: map[string]any{ "id": taskRecord.ID, "task_no": taskRecord.TaskNo, "file_name": taskRecord.FileName, "package_id": taskRecord.PackageID, "package_code": taskRecord.PackageCode, diff --git a/internal/task/polling_carddata_handler.go b/internal/task/polling_carddata_handler.go index 331437a..d0e9179 100644 --- a/internal/task/polling_carddata_handler.go +++ b/internal/task/polling_carddata_handler.go @@ -67,11 +67,15 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e } result, err := h.gateway.QueryFlow(ctx, &gateway.FlowQueryReq{CardNo: card.ICCID}) if err != nil { - _ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt) + if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { + return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 流量 Integration Log 失败", logErr) + } return h.failAndRequeue(ctx, cardID, startedAt, "查询流量失败", err) } if strings.TrimSpace(result.ICCID) == "" { - _ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt) + if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { + return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 流量 Integration Log 失败", logErr) + } return h.failAndRequeue(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil) } if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil { diff --git a/internal/task/polling_cardstatus_handler.go b/internal/task/polling_cardstatus_handler.go index 4dcc863..37b6c40 100644 --- a/internal/task/polling_cardstatus_handler.go +++ b/internal/task/polling_cardstatus_handler.go @@ -66,11 +66,15 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task) } result, err := h.gateway.QueryCardStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID}) if err != nil { - _ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt) + if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { + return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 网络 Integration Log 失败", logErr) + } return h.failAndRequeue(ctx, cardID, startedAt, "查询卡状态失败", err) } if strings.TrimSpace(result.ICCID) == "" { - _ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt) + if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { + return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 网络 Integration Log 失败", logErr) + } return h.failAndRequeue(ctx, cardID, startedAt, "卡状态查询响应缺少 ICCID", nil) } if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil { diff --git a/internal/task/polling_integration_log.go b/internal/task/polling_integration_log.go index 18dfbf2..6564ffa 100644 --- a/internal/task/polling_integration_log.go +++ b/internal/task/polling_integration_log.go @@ -38,11 +38,13 @@ func completeGatewayAttempt(ctx context.Context, repository *integrationlog.Repo if repository == nil || attempt == nil { return errors.New(errors.CodeInternalError, "Gateway Integration Log 尝试不存在") } + completionCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer cancel() result := constants.IntegrationResultFailed if success { result = constants.IntegrationResultSuccess } - _, err := repository.Complete(ctx, attempt.IntegrationID, integrationlog.Completion{ + _, err := repository.Complete(completionCtx, attempt.IntegrationID, integrationlog.Completion{ Result: result, DurationMS: time.Since(startedAt).Milliseconds(), StateChanged: false, ResponseSummary: map[string]any{"success": success}, }) diff --git a/internal/task/polling_realname_handler.go b/internal/task/polling_realname_handler.go index 4761f57..0cbe203 100644 --- a/internal/task/polling_realname_handler.go +++ b/internal/task/polling_realname_handler.go @@ -58,11 +58,15 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) e } result, err := h.gateway.QueryRealnameStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID}) if err != nil { - _ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt) + if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { + return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 实名 Integration Log 失败", logErr) + } return h.failAndRequeue(ctx, cardID, startedAt, "查询实名状态失败", err) } if strings.TrimSpace(result.ICCID) == "" { - _ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt) + if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { + return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 实名 Integration Log 失败", logErr) + } return h.failAndRequeue(ctx, cardID, startedAt, "实名查询响应缺少 ICCID", nil) } if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil { diff --git a/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/.openspec.yaml b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/.openspec.yaml new file mode 100644 index 0000000..d160e09 --- /dev/null +++ b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-21 diff --git a/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/design.md b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/design.md new file mode 100644 index 0000000..b864c53 --- /dev/null +++ b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/design.md @@ -0,0 +1,53 @@ +## Context + +见 proposal.md。现有实现以 Audit 和 Integration 归档对作为原子清理单元，且在生成 Integration 最终 revision 前要求整日不存在 `pending`。2026-08-18 的 458 条 Gateway 同步轮询遗留 pending 因而阻断约 314 万条 Integration Log 和后续日期的清理。 + +## Goals / Non-Goals + +**Goals:** + +- 只保留 pending Integration Log 在线，尽快释放已终态日志和审计数据。 +- 保持删除前可恢复归档和逐批续跑语义。 +- 防止旧 Gateway 轮询处理器静默制造新的永久 pending。 + +**Non-Goals:** + +- 不直接修改历史 pending 的结果或删除它们。 +- 不改变 Integration Log 的写入频率、JSONL 格式、对象存储格式或数据库 Schema。 +- 不在 Worker 内执行 VACUUM 或表重写。 + +## Decisions + +### 归档覆盖全日，删除只匹配终态 + +仍按创建日生成包含全部记录的归档 revision；移除“存在 pending 即不得最终归档”的门禁。留存校验确认当前在线集合与该 revision 一致后，删除条件限定为 `result <> 'pending'`。这样每条已删除数据都在对象存储中存在恢复副本，而 pending 永不被删除。 + +不按结果分别创建归档文件：会改变既有归档格式和恢复语义，且没有必要。 + +### Audit 与 Integration 分来源推进 + +Audit 归档校验成功后独立完成 Audit 资源和事件删除；Integration 的 pending 或残留不再使 Audit 退化为未清理。每个来源仍按自身失败日阻断该来源后续日期，避免跨越该来源的归档或校验故障。 + +不继续要求两来源同日原子清理：该设计正是容量积压的根因，且两类数据已有独立的归档账本和清理断点。 + +### 以剩余可删除记录决定 Integration 续跑 + +已经删除过终态记录的日期不因只剩 pending 而反复阻断日期扫描。若旧 pending 后续进入终态，则该日期重新生成 revision、验证当前在线集合，并删除新增终态记录。查询仅检查可删除终态记录和对应清理断点，避免对只含 pending 的大历史范围反复做全表工作。 + +### Gateway 失败分支不吞终结错误 + +三个旧轮询 Handler 对已创建的尝试使用受控收口：终结写入失败返回任务错误并保留重试信号；不再使用 `_ = completeGatewayAttempt(...)` 静默丢弃错误。任务取消时仍尝试使用不受取消影响的短生命周期上下文终结该条日志。 + +## Risks / Trade-offs + +- [终态日志在归档和删除间新增或变化] → 归档后重新统计和校验；不一致时不删除并在下次重建 revision。 +- [pending 永久不终结] → 仅保留这些记录，不能阻止终态数据和审计数据释放；通过调查接口和留存日志排查。 +- [PostgreSQL 文件空间不立即缩小] → DELETE 回收空间供后续写入复用；维护者按运行手册评估 VACUUM，不由 Worker 自动执行。 +- [旧 pending 终结后遗漏清理] → 续跑选择条件识别清理后进入终态的记录，并重建该日 revision。 + +## Migration Plan + +1. 发布 Worker 二进制，保持既有物理清理开关状态。 +2. 若当前开关关闭，维护者在低峰期启用后重启调度 Worker。 +3. 观察 `logs/audit-retention.log`、`tb_log_archive_run` 清理断点和数据库磁盘指标；8 月 18 日的已终态记录应先被清理，458 条 pending 保留。 +4. 若归档或删除异常，关闭清理开关并重启 Worker；已删除记录通过对应归档 revision 恢复，未删除记录仍在线。 diff --git a/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/proposal.md b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/proposal.md new file mode 100644 index 0000000..dfd8aea --- /dev/null +++ b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/proposal.md @@ -0,0 +1,26 @@ +## Why + +日留存将某日存在的任意 `pending` Integration Log 视为整日删除的阻断条件。Gateway 同步轮询的遗留 pending 记录使 2026-08-18 及所有后续日期无法清理，在线日志持续积压并带来数据库容量风险。 + +## What Changes + +- 日留存继续归档全天 Integration Log，但只将已进入公开终态的记录纳入可清理集合。 +- `pending` 记录保留在线，且不得阻断同日终态记录和后续日期的归档、验证与清理。 +- 已完成的 Gateway 轮询尝试在终结日志写入失败时不得静默忽略错误，避免新增永久 pending 记录。 + +## Capabilities + +### New Capabilities + +无。 + +### Modified Capabilities + +- `external-integration`: 外部交互日志日留存从“整日无 pending 才能清理”改为“仅清理已终态记录并保留 pending”。 +- `operations-audit`: 日留存不因 Integration Log 的 pending 记录阻断审计事件和资源快照的连续清理。 + +## Impact + +- 代码：`internal/application/auditarchive/`、旧 Gateway 轮询任务处理器。 +- Worker：日留存会释放已终态的在线日志；pending 继续可调查和补偿。 +- 数据库：无迁移，不执行直接 SQL 数据修复。 diff --git a/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/specs/external-integration/spec.md b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/specs/external-integration/spec.md new file mode 100644 index 0000000..7da9466 --- /dev/null +++ b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/specs/external-integration/spec.md @@ -0,0 +1,59 @@ +## MODIFIED Requirements + +### Requirement: 外部交互日志逐日物理留存 + +系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位物理留存 `tb_integration_log`。删除一条已终态在线外部交互日志前，系统 MUST 已为其创建日生成可读取、可验证的归档对象和清单；该归档 MUST 包含该日当时全部外部交互日志，并验证日期范围、记录数量和校验摘要可作为恢复凭证。`pending` 记录 MUST 保留在线，不得被物理删除，且不得阻断同日已终态记录或更晚日期的归档、验证与物理清理。每次执行 SHALL 清理所有满足归档校验条件的已终态记录；某一来源的归档或校验失败时，系统 MUST 保留该来源当天及更晚日期的在线外部交互日志。 + +#### Scenario: 最终归档通过后清理在线外部交互日志 +- **GIVEN** 某已结束自然日的外部交互归档对象和清单校验成功，且存在已终态在线外部交互日志 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统分批删除该日已终态的在线外部交互日志并保留归档对象 + +#### Scenario: 存在待完成外部交互日志 +- **GIVEN** 某归档日仍存在待完成的外部交互日志 +- **WHEN** 日留存任务尝试处理该日期 +- **THEN** 系统保留该 pending 日志，但不得停止同日终态记录和更晚日期的清理 + +#### Scenario: 含 pending 的日期清理终态日志 +- **GIVEN** 某已结束自然日同时存在已终态和 `pending` 的外部交互日志，且该日归档对象和清单校验成功 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统分批删除该日已终态的在线外部交互日志，保留 `pending` 记录，并保留归档对象 + +#### Scenario: 历史积压按日期连续补清 +- **GIVEN** 存在多个昨天及更早日期尚未完成物理清理，且这些日期各自存在满足归档校验条件的已终态外部交互日志 +- **WHEN** 日留存任务执行 +- **THEN** 系统按日期从早到晚清理全部符合条件的终态日志 + +#### Scenario: pending 不阻断后续日期 +- **GIVEN** 较早归档日只剩 `pending` 外部交互日志，且更晚归档日存在满足归档校验条件的已终态外部交互日志 +- **WHEN** 日留存任务执行 +- **THEN** 系统继续处理更晚归档日的已终态外部交互日志，不因较早日期的 pending 停止日期推进 + +#### Scenario: pending 后续终结 +- **GIVEN** 某归档日的 pending 外部交互日志在该日首次留存后进入公开终态 +- **WHEN** 后续日留存任务执行 +- **THEN** 系统为包含该记录的当前在线集合重新验证归档，并删除该已终态记录 + +#### Scenario: 归档或校验失败保留同来源后续数据 +- **GIVEN** 某归档日的外部交互归档缺失、归档校验失败或清单与在线数据不一致 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统不删除该日及更晚日期的在线外部交互日志，并将失败信息写入独立日留存日志 + +### Requirement: 外部失败边界 + +系统 SHALL 将第三方超时、渠道错误和无效响应转换为当前稳定的系统错误；已接入外部交互日志的渠道同时保留脱敏结果。Gateway 同步轮询在已建立 pending 外部交互日志后，无论查询成功、失败或任务上下文取消，MUST 尝试将该尝试终结为公开终态；终结失败 MUST 作为任务失败返回或记录为可重试错误，不得静默忽略。 + +#### Scenario: 外部失败边界 +- **GIVEN** 外部系统超时或返回失败 +- **WHEN** 调用依赖该系统的操作 +- **THEN** 客户端收到当前稳定错误；已接入外部交互日志的调用记录脱敏渠道结果 + +#### Scenario: Gateway 轮询查询失败 +- **GIVEN** Gateway 同步轮询已建立 pending 外部交互日志且查询返回错误 +- **WHEN** 轮询任务处理该错误 +- **THEN** 系统将该日志终结为失败或结果未知的公开终态，并按既有策略重新入队 + +#### Scenario: Gateway 轮询终结日志失败 +- **GIVEN** Gateway 同步轮询已建立 pending 外部交互日志且终结写入失败 +- **WHEN** 轮询任务结束该次尝试 +- **THEN** 系统记录并返回该终结失败，不得把该错误静默丢弃 diff --git a/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/specs/operations-audit/spec.md b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/specs/operations-audit/spec.md new file mode 100644 index 0000000..e70714e --- /dev/null +++ b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/specs/operations-audit/spec.md @@ -0,0 +1,25 @@ +## MODIFIED Requirements + +### Requirement: 审计在线数据逐日物理留存 + +系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位处理统一审计事件及其资源快照的在线留存。每次留存执行 SHALL 从最早尚未完成清理的归档日开始，连续处理至昨天；Integration Log 的 `pending` 记录不得阻断已通过完整性校验的审计事件及资源快照清理。任一审计归档日未通过完整性校验时，系统 MUST 保留该日及其后续日期的在线审计数据，且不得将它们标记为已清理。系统 MUST 先删除该日的审计资源快照，再删除该日的审计事件，并保留对象存储归档对象及清单作为恢复凭证。 + +#### Scenario: 已验证日期完成物理清理 +- **GIVEN** 某已结束自然日的审计归档成功，归档对象和清单可读取且与在线记录范围、数量和校验摘要一致 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统分批删除该日在线审计资源快照和审计事件，并将该归档日标记为已清理 + +#### Scenario: Integration Log pending 不阻断审计清理 +- **GIVEN** 某已结束自然日的审计归档已通过完整性校验，且同日存在 pending Integration Log +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统仍完成该日在线审计资源快照和审计事件的物理清理 + +#### Scenario: 归档校验失败阻断连续清理 +- **GIVEN** 某尚未清理的审计归档日缺少归档、归档校验失败或清单与在线数据不一致 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统不删除该日或更晚日期的在线审计数据，并将失败信息写入独立日留存日志 + +#### Scenario: 失败后从未完成日期续跑 +- **GIVEN** 日留存任务在某个审计归档日失败，较早日期已经标记为已清理 +- **WHEN** 后续日留存任务再次执行 +- **THEN** 系统从最早未完成清理的审计归档日继续处理，不重复删除已完成日期的数据 diff --git a/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/tasks.md b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/tasks.md new file mode 100644 index 0000000..a045060 --- /dev/null +++ b/openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/tasks.md @@ -0,0 +1,15 @@ +## 1. 日留存按终态清理 + +- [x] 1.1 调整 Integration Log 归档与留存校验，使归档保留全日快照、pending 不再阻断终态记录清理。 +- [x] 1.2 将 Integration Log 删除条件限制为公开终态，并支持同一日期遗留 pending 后续终结时重建归档 revision 和续跑删除。 +- [x] 1.3 让 Audit 留存独立于同日 Integration pending 完成归档校验和物理清理，并保持各来源的失败断点。 + +## 2. Gateway 轮询收口 + +- [x] 2.1 修改旧卡状态、流量和实名轮询 Handler，使已创建尝试的终结失败不被忽略，并在取消上下文下仍尝试终结。 + +## 3. 验证与运行说明 + +- [x] 3.1 格式化修改的 Go 文件并构建 `./cmd/worker`。 +- [ ] 3.2 以可控演练或最小验证确认：含 pending 的日期删除终态日志并继续后续日期，pending 本身保留在线。 +- [x] 3.3 更新生产运行说明，记录发布 Worker、启用既有清理开关、观察留存日志与 PostgreSQL 空间复用的步骤。 diff --git a/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/.openspec.yaml b/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/.openspec.yaml new file mode 100644 index 0000000..701445b --- /dev/null +++ b/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-26 diff --git a/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/design.md b/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/design.md new file mode 100644 index 0000000..e9543e6 --- /dev/null +++ b/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/design.md @@ -0,0 +1,32 @@ +## Context + +批量订购逐行调用后台订单创建，并在全部行处理结束后于同一事务保存任务结果和完成审计。当前完成审计以关联订单审计事件数量充当成功行数；该数量会包含失败记录和重试记录，可能大于输入总行数，违反审计表的批量统计约束并回滚任务完成状态。 + +## Goals / Non-Goals + +**Goals:** + +- 使任务完成审计使用本次处理结果的业务成功数和失败数。 +- 保持任务结果、汇总计数和完成审计在同一事务提交。 + +**Non-Goals:** + +- 不改变单笔订单、钱包余额不足或重复输入的业务规则。 +- 不恢复或重放历史任务；历史数据由维护者按生产运维流程处置。 + +## Decisions + +- 以 `processRows` 返回的 `successCount` 和 `failCount` 作为任务完成审计的唯一批量统计来源。这些值与将写入任务记录的逐行结果来自同一次处理，满足审计表的计数不变量。 +- 移除对关联订单审计事件数量的查询。审计事件数量会因失败记录和重试而膨胀，不能表示批量业务成功数。 + +## Risks / Trade-offs + +- [历史待处理任务不自动恢复] → 保持现有状态，避免对已扣款订单重复执行；由维护者核对后受控结案。 +- [完成审计仍依赖事务写入成功] → 使用与任务结果一致的统计值，消除本次约束失败原因，同时保留既有原子性。 + +## Migration Plan + +1. 发布 API/Worker 中的修复后二进制，重点更新 Worker。 +2. 在隔离环境验证部分成功任务返回完成状态与一致计数。 +3. 生产历史任务由维护者在核对订单和源文件后手工结案；不重投原 Asynq 消息。 +4. 若发布异常，回滚 Worker 二进制；该修复不含迁移。 diff --git a/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/proposal.md b/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/proposal.md new file mode 100644 index 0000000..83fba50 --- /dev/null +++ b/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/proposal.md @@ -0,0 +1,23 @@ +## Why + +资产套餐批量订购在部分行因钱包余额不足或输入重复而失败时，完成审计将审计事件条数误作成功行数，可能违反审计批量统计约束并回滚任务终态。已实际创建的订单因此在任务查询中显示为待处理且无结果，既不能反映业务结果，也有误重试风险。 + +## What Changes + +- 批量订购任务完成时，使用本次逐行处理得到的成功数和失败数写入完成审计。 +- 确保部分成功、部分失败的批量订购任务能持久化为已完成，并可查询真实汇总和逐行结果。 + +## Capabilities + +### New Capabilities + +无。 + +### Modified Capabilities + +- `package-lifecycle`: 异步批量订购任务必须持久化并返回逐行处理后的终态统计，即使部分资产因余额不足或输入校验失败而未能购包。 + +## Impact + +- `internal/task/asset_package_batch_order.go` 中的批量订购完成与审计统计。 +- 不新增 API、数据表或依赖；既有批量订购详情将正确展示部分完成结果。 diff --git a/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/specs/package-lifecycle/spec.md b/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/specs/package-lifecycle/spec.md new file mode 100644 index 0000000..e100fe5 --- /dev/null +++ b/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/specs/package-lifecycle/spec.md @@ -0,0 +1,17 @@ +## MODIFIED Requirements + +### Requirement: 批量操作可追踪 + +系统 SHALL 为同步批量分配和调价直接返回处理结果；对异步批量订购返回任务标识并提供状态查询。异步批量订购完成后，系统 MUST 持久化每个输入行的成功或失败结果及与其一致的总数、成功数和失败数；部分资产因余额不足、资产校验或重复输入失败不得阻止任务进入完成终态。 + +#### Scenario: 批量操作可追踪 + +- **GIVEN** 操作者提交非空且有权处理的资源集合 +- **WHEN** 创建批量操作 +- **THEN** 同步操作直接返回结果；异步订购返回任务标识且可查询处理状态 + +#### Scenario: 批量订购部分失败后查询结果 + +- **GIVEN** 异步批量订购中的部分资产已成功创建订单，其他资产因钱包余额不足或输入重复失败 +- **WHEN** Worker 完成全部输入行的处理 +- **THEN** 任务状态为已完成，逐行结果保留成功订单与失败原因，且总数等于成功数与失败数之和 diff --git a/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/tasks.md b/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/tasks.md new file mode 100644 index 0000000..743eea0 --- /dev/null +++ b/openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/tasks.md @@ -0,0 +1,8 @@ +## 1. 批量订购完成统计修复 + +- [x] 1.1 修改批量订购完成审计，使其使用逐行处理产生的成功数和失败数，不再统计关联审计事件数量。 +- [x] 1.2 格式化修改的 Go 文件并构建 Worker，验证部分成功任务的完成审计统计满足任务总行数约束。 + +## 2. 规格验证 + +- [x] 2.1 运行 OpenSpec 校验，确认变更工件一致。 diff --git a/openspec/specs/external-integration/spec.md b/openspec/specs/external-integration/spec.md index c057f1e..7f682f5 100644 --- a/openspec/specs/external-integration/spec.md +++ b/openspec/specs/external-integration/spec.md @@ -28,7 +28,7 @@ ### Requirement: 外部失败边界 -系统 SHALL 将第三方超时、渠道错误和无效响应转换为当前稳定的系统错误；已接入外部交互日志的渠道同时保留脱敏结果。 +系统 SHALL 将第三方超时、渠道错误和无效响应转换为当前稳定的系统错误；已接入外部交互日志的渠道同时保留脱敏结果。Gateway 同步轮询在已建立 pending 外部交互日志后，无论查询成功、失败或任务上下文取消，MUST 尝试将该尝试终结为公开终态；终结失败 MUST 作为任务失败返回或记录为可重试错误，不得静默忽略。 #### Scenario: 外部失败边界 @@ -36,6 +36,18 @@ - **WHEN** 调用依赖该系统的操作 - **THEN** 客户端收到当前稳定错误；已接入外部交互日志的调用记录脱敏渠道结果 +#### Scenario: Gateway 轮询查询失败 + +- **GIVEN** Gateway 同步轮询已建立 pending 外部交互日志且查询返回错误 +- **WHEN** 轮询任务处理该错误 +- **THEN** 系统将该日志终结为失败或结果未知的公开终态，并按既有策略重新入队 + +#### Scenario: Gateway 轮询终结日志失败 + +- **GIVEN** Gateway 同步轮询已建立 pending 外部交互日志且终结写入失败 +- **WHEN** 轮询任务结束该次尝试 +- **THEN** 系统记录并返回该终结失败，不得把该错误静默丢弃 + ### Requirement: 外部调用重试边界 系统 SHALL 仅对 Gateway 客户端超时、连接失败和 DNS 失败自动重试，默认最多重试两次且每次重新签名；HTTP 非 200、响应解析失败、Gateway 业务失败和调用方取消不重试。企业微信审批只有确认尚未调用提交接口的失败可释放后重试，提交结果未知时不得盲目重建审批。 @@ -103,22 +115,42 @@ ### Requirement: 外部交互日志逐日物理留存 -系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位物理留存 `tb_integration_log`。删除某日在线外部交互日志前，系统 MUST 为该日生成最终归档版本，确认不存在待完成记录，并验证归档对象、清单、日期范围、记录数量和校验摘要可作为恢复凭证。每次执行 SHALL 从最早尚未完成清理的归档日连续处理至昨天；任一日期不能形成或验证最终归档时，系统 MUST 不删除该日及更晚日期的在线外部交互日志。 +系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位物理留存 `tb_integration_log`。删除一条已终态在线外部交互日志前，系统 MUST 已为其创建日生成可读取、可验证的归档对象和清单；该归档 MUST 包含该日当时全部外部交互日志，并验证日期范围、记录数量和校验摘要可作为恢复凭证。`pending` 记录 MUST 保留在线，不得被物理删除，且不得阻断同日已终态记录或更晚日期的归档、验证与物理清理。每次执行 SHALL 清理所有满足归档校验条件的已终态记录；某一来源的归档或校验失败时，系统 MUST 保留该来源当天及更晚日期的在线外部交互日志。 #### Scenario: 最终归档通过后清理在线外部交互日志 -- **GIVEN** 某已结束自然日的外部交互日志均已进入终态，且该日最终归档对象和清单校验成功 +- **GIVEN** 某已结束自然日的外部交互归档对象和清单校验成功，且存在已终态在线外部交互日志 - **WHEN** 日留存任务处理该日期 -- **THEN** 系统分批删除该日的在线外部交互日志并将该归档日标记为已清理，归档对象保持可用 +- **THEN** 系统分批删除该日已终态的在线外部交互日志并保留归档对象 #### Scenario: 存在待完成外部交互日志 - **GIVEN** 某归档日仍存在待完成的外部交互日志 - **WHEN** 日留存任务尝试处理该日期 -- **THEN** 系统不删除该日及更晚日期的在线外部交互日志，并在独立日留存日志中记录该日期未形成最终归档的原因 +- **THEN** 系统保留该 pending 日志，但不得停止同日终态记录和更晚日期的清理 + +#### Scenario: 含 pending 的日期清理终态日志 +- **GIVEN** 某已结束自然日同时存在已终态和 `pending` 的外部交互日志，且该日归档对象和清单校验成功 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统分批删除该日已终态的在线外部交互日志，保留 `pending` 记录，并保留归档对象 #### Scenario: 历史积压按日期连续补清 -- **GIVEN** 存在多个昨天及更早日期尚未完成物理清理 -- **WHEN** 日留存任务执行且这些日期依次通过最终归档和完整性校验 -- **THEN** 系统按日期从早到晚清理全部连续合格日期 +- **GIVEN** 存在多个昨天及更早日期尚未完成物理清理，且这些日期各自存在满足归档校验条件的已终态外部交互日志 +- **WHEN** 日留存任务执行 +- **THEN** 系统按日期从早到晚清理全部符合条件的终态日志 + +#### Scenario: pending 不阻断后续日期 +- **GIVEN** 较早归档日只剩 `pending` 外部交互日志，且更晚归档日存在满足归档校验条件的已终态外部交互日志 +- **WHEN** 日留存任务执行 +- **THEN** 系统继续处理更晚归档日的已终态外部交互日志，不因较早日期的 pending 停止日期推进 + +#### Scenario: pending 后续终结 +- **GIVEN** 某归档日的 pending 外部交互日志在该日首次留存后进入公开终态 +- **WHEN** 后续日留存任务执行 +- **THEN** 系统为包含该记录的当前在线集合重新验证归档，并删除该已终态记录 + +#### Scenario: 归档或校验失败保留同来源后续数据 +- **GIVEN** 某归档日的外部交互归档缺失、归档校验失败或清单与在线数据不一致 +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统不删除该日及更晚日期的在线外部交互日志，并将失败信息写入独立日留存日志 ## 可达操作索引 diff --git a/openspec/specs/operations-audit/spec.md b/openspec/specs/operations-audit/spec.md index dd53ece..de90aba 100644 --- a/openspec/specs/operations-audit/spec.md +++ b/openspec/specs/operations-audit/spec.md @@ -48,13 +48,18 @@ ### Requirement: 审计在线数据逐日物理留存 -系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位处理统一审计事件及其资源快照的在线留存。每次留存执行 SHALL 从最早尚未完成清理的归档日开始，连续处理至昨天；任一归档日未通过完整性校验时，系统 MUST 保留该日及其后续日期的在线审计数据，且不得将它们标记为已清理。系统 MUST 先删除该日的审计资源快照，再删除该日的审计事件，并保留对象存储归档对象及清单作为恢复凭证。 +系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位处理统一审计事件及其资源快照的在线留存。每次留存执行 SHALL 从最早尚未完成清理的归档日开始，连续处理至昨天；Integration Log 的 `pending` 记录不得阻断已通过完整性校验的审计事件及资源快照清理。任一审计归档日未通过完整性校验时，系统 MUST 保留该日及其后续日期的在线审计数据，且不得将它们标记为已清理。系统 MUST 先删除该日的审计资源快照，再删除该日的审计事件，并保留对象存储归档对象及清单作为恢复凭证。 #### Scenario: 已验证日期完成物理清理 - **GIVEN** 某已结束自然日的审计归档成功，归档对象和清单可读取且与在线记录范围、数量和校验摘要一致 - **WHEN** 日留存任务处理该日期 - **THEN** 系统分批删除该日在线审计资源快照和审计事件，并将该归档日标记为已清理 +#### Scenario: Integration Log pending 不阻断审计清理 +- **GIVEN** 某已结束自然日的审计归档已通过完整性校验，且同日存在 pending Integration Log +- **WHEN** 日留存任务处理该日期 +- **THEN** 系统仍完成该日在线审计资源快照和审计事件的物理清理 + #### Scenario: 归档校验失败阻断连续清理 - **GIVEN** 某尚未清理的归档日缺少归档、归档校验失败或清单与在线数据不一致 - **WHEN** 日留存任务处理该日期 diff --git a/openspec/specs/package-lifecycle/spec.md b/openspec/specs/package-lifecycle/spec.md index 6e1a3b8..db05e73 100644 --- a/openspec/specs/package-lifecycle/spec.md +++ b/openspec/specs/package-lifecycle/spec.md @@ -18,7 +18,7 @@ ### Requirement: 批量操作可追踪 -系统 SHALL 为同步批量分配和调价直接返回处理结果；对异步批量订购返回任务标识并提供状态查询。 +系统 SHALL 为同步批量分配和调价直接返回处理结果；对异步批量订购返回任务标识并提供状态查询。异步批量订购完成后，系统 MUST 持久化每个输入行的成功或失败结果及与其一致的总数、成功数和失败数；部分资产因余额不足、资产校验或重复输入失败不得阻止任务进入完成终态。 #### Scenario: 批量操作可追踪 @@ -26,6 +26,12 @@ - **WHEN** 创建批量操作 - **THEN** 同步操作直接返回结果；异步订购返回任务标识且可查询处理状态 +#### Scenario: 批量订购部分失败后查询结果 + +- **GIVEN** 异步批量订购中的部分资产已成功创建订单，其他资产因钱包余额不足或输入重复失败 +- **WHEN** Worker 完成全部输入行的处理 +- **THEN** 任务状态为已完成，逐行结果保留成功订单与失败原因，且总数等于成功数与失败数之和 + ### Requirement: 授权页面禁止重复选择套餐 系统 SHALL 使代理系列授权页面能够区分目标店铺已授权和未授权套餐；已授权套餐 MUST 以不可新增的状态返回，首次创建系列授权和既有系列新增套餐均适用。