diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 8c66760..270837f 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -153,7 +153,7 @@ func runWorker(cfg *config.Config) { taskHandler.RegisterHandlers() registerWeComApprovalTasks(taskHandler.GetMux(), runtime, cfg, appLogger) registerAgentRechargeRecoveryTask(taskHandler.GetMux(), runtime, appLogger) - registerAuditArchiveTask(taskHandler.GetMux(), runtime, cfg.Worker.AuditRetentionCleanupEnabled, appLogger, retentionLogger) + registerAuditArchiveTask(taskHandler.GetMux(), runtime, cfg.Worker.AuditRetentionCleanupEnabled, cfg.Worker.AuditArchiveTasksEnabled, appLogger, retentionLogger) outboxHandler := outbox.NewHandler(runtime.outboxConsumers) taskHandler.GetMux().HandleFunc(constants.TaskTypeOutboxDeliver, outboxHandler.Handle) startOutboxRelay(ctx, runtime, cfg.Worker.InstanceName, appLogger) @@ -286,6 +286,7 @@ func initWorkerRuntime(ctx context.Context, cfg *config.Config, appLogger *zap.L pollingIotCardStore, appLogger, cfg.Polling.VerboseLog, + cfg.Worker.PollingTotalMaxConcurrency, ) pollingDeviceSimBindingStore := postgres.NewDeviceSimBindingStore(db, redisClient) @@ -693,7 +694,7 @@ func startAsynqScheduler(cfg *config.Config, redisAddr string, appLogger *zap.Lo &asynq.SchedulerOpts{Location: time.Local}, ) - if err := registerAsynqScheduleTasks(asynqScheduler); err != nil { + if err := registerAsynqScheduleTasks(asynqScheduler, cfg.Worker.AuditArchiveTasksEnabled); err != nil { appLogger.Fatal("注册 Asynq 定时任务失败", zap.Error(err)) } @@ -704,12 +705,14 @@ func startAsynqScheduler(cfg *config.Config, redisAddr string, appLogger *zap.Lo }() appLogger.Info("Asynq Scheduler 已启动", - zap.Bool("audit_retention_cleanup_enabled", cfg.Worker.AuditRetentionCleanupEnabled)) + zap.Bool("audit_retention_cleanup_enabled", cfg.Worker.AuditRetentionCleanupEnabled), + zap.Bool("audit_archive_tasks_enabled", cfg.Worker.AuditArchiveTasksEnabled), + zap.Int("polling_total_max_concurrency", cfg.Worker.PollingTotalMaxConcurrency)) return asynqScheduler } // registerAsynqScheduleTasks 注册 Worker 入口需要的全部定时任务。 -func registerAsynqScheduleTasks(asynqScheduler *asynq.Scheduler) error { +func registerAsynqScheduleTasks(asynqScheduler *asynq.Scheduler, auditArchiveEnabled bool) error { if _, err := asynqScheduler.Register("@every 1m", asynq.NewTask( constants.TaskTypeAgentRechargeRecovery, nil, @@ -779,6 +782,9 @@ func registerAsynqScheduleTasks(asynqScheduler *asynq.Scheduler) error { ); err != nil { return fmt.Errorf("注册每日流量落盘定时任务失败: %w", err) } + if !auditArchiveEnabled { + return nil + } if _, err := asynqScheduler.Register("CRON_TZ=Asia/Shanghai 0 4 * * *", asynq.NewTask( constants.TaskTypeAuditDailyArchive, nil, @@ -813,13 +819,13 @@ func registerAsynqScheduleTasks(asynqScheduler *asynq.Scheduler) error { } // registerAuditArchiveTask 注册 Audit 与 Integration 冷归档任务处理器。 -func registerAuditArchiveTask(mux *asynq.ServeMux, runtime *workerRuntime, cleanupEnabled bool, appLogger, retentionLogger *zap.Logger) { +func registerAuditArchiveTask(mux *asynq.ServeMux, runtime *workerRuntime, cleanupEnabled, enabled bool, appLogger, retentionLogger *zap.Logger) { if runtime.storageSvc == nil { appLogger.Warn("对象存储未配置，审计归档任务将在执行时重试") - mux.HandleFunc(constants.TaskTypeAuditDailyArchive, task.NewAuditDailyArchiveHandler(nil, appLogger).Handle) - integrationHandler := task.NewIntegrationArchiveHandler(nil, appLogger) + mux.HandleFunc(constants.TaskTypeAuditDailyArchive, task.NewAuditDailyArchiveHandler(nil, appLogger, enabled).Handle) + integrationHandler := task.NewIntegrationArchiveHandler(nil, appLogger, enabled) mux.HandleFunc(constants.TaskTypeIntegrationDailyArchive, integrationHandler.HandleDaily) - mux.HandleFunc(constants.TaskTypeAuditDailyRetention, task.NewAuditRetentionHandler(nil, retentionLogger, cleanupEnabled).Handle) + mux.HandleFunc(constants.TaskTypeAuditDailyRetention, task.NewAuditRetentionHandler(nil, retentionLogger, cleanupEnabled, enabled).Handle) return } auditWriter, ok := runtime.workerResult.Services.PaymentAudit.(*auditInfra.Writer) @@ -830,15 +836,15 @@ func registerAuditArchiveTask(mux *asynq.ServeMux, runtime *workerRuntime, clean if err != nil { appLogger.Fatal("初始化统一审计归档服务失败", zap.Error(err)) } - mux.HandleFunc(constants.TaskTypeAuditDailyArchive, task.NewAuditDailyArchiveHandler(service, appLogger).Handle) - integrationHandler := task.NewIntegrationArchiveHandler(service, appLogger) + mux.HandleFunc(constants.TaskTypeAuditDailyArchive, task.NewAuditDailyArchiveHandler(service, appLogger, enabled).Handle) + integrationHandler := task.NewIntegrationArchiveHandler(service, appLogger, enabled) mux.HandleFunc(constants.TaskTypeIntegrationDailyArchive, integrationHandler.HandleDaily) - mux.HandleFunc(constants.TaskTypeAuditDailyRetention, task.NewAuditRetentionHandler(service, retentionLogger, cleanupEnabled).Handle) + mux.HandleFunc(constants.TaskTypeAuditDailyRetention, task.NewAuditRetentionHandler(service, retentionLogger, cleanupEnabled, enabled).Handle) appLogger.Info("注册审计归档任务处理器", zap.String("audit_task_type", constants.TaskTypeAuditDailyArchive), zap.String("integration_daily_task_type", constants.TaskTypeIntegrationDailyArchive), zap.String("retention_task_type", constants.TaskTypeAuditDailyRetention), - zap.Bool("retention_cleanup_enabled", cleanupEnabled)) + zap.Bool("retention_cleanup_enabled", cleanupEnabled), zap.Bool("archive_tasks_enabled", enabled)) } // createTaskHandler 创建并返回包含全部任务处理器的 Asynq Handler。 diff --git a/internal/application/auditarchive/retention.go b/internal/application/auditarchive/retention.go index 0b64553..35c935c 100644 --- a/internal/application/auditarchive/retention.go +++ b/internal/application/auditarchive/retention.go @@ -49,7 +49,7 @@ func (e *RetentionBlockedError) Error() string { } func (e *RetentionBlockedError) Unwrap() error { return e.Err } -// RetainPendingDays 从最早仍在线的日期连续处理至昨天。cleanup 为 false 时只读校验。 +// RetainPendingDays 每次最多处理一个最早待处理的上海自然日，避免单任务跨历史日期长时间占用数据库。 func (s *Service) RetainPendingDays(ctx context.Context, cleanup bool) ([]RetentionResult, error) { if s.db == nil || s.store == nil { return nil, fmt.Errorf("日志日留存数据库或对象存储未配置") @@ -58,42 +58,14 @@ func (s *Service) RetainPendingDays(ctx context.Context, cleanup bool) ([]Retent if err != nil { return nil, err } - results := make([]RetentionResult, 0) - var auditBlocked, integrationBlocked *RetentionBlockedError - for date := start; date.Before(end); date = date.AddDate(0, 0, 1) { - 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 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 !start.Before(end) { + return nil, nil } - if auditBlocked != nil { - return results, auditBlocked + result, err := s.retainDate(ctx, start, cleanup) + if err != nil { + return nil, &RetentionBlockedError{ArchiveDate: start.Format(time.DateOnly), Source: constants.AuditArchiveSource, Err: err} } - if integrationBlocked != nil { - return results, integrationBlocked - } - return results, nil + return []RetentionResult{result}, nil } // RetainDate 校验并按需清理指定已结束自然日，供受控演练使用。 diff --git a/internal/infrastructure/integrationlog/repository.go b/internal/infrastructure/integrationlog/repository.go index 8ffa159..9ec7157 100644 --- a/internal/infrastructure/integrationlog/repository.go +++ b/internal/infrastructure/integrationlog/repository.go @@ -45,6 +45,7 @@ type Attempt struct { CorrelationID *string AuditEventID *uint InitialResult string + StateChanged bool RecoveryStrategy *string } @@ -135,7 +136,7 @@ func (r *Repository) Start(ctx context.Context, input Attempt) (*model.Integrati Operation: input.Operation, ExternalID: sanitizedOptionalText(input.ExternalID), ResourceType: resourceType, ResourceID: input.ResourceID, ResourceKey: sanitizedOptionalText(input.ResourceKey), TriggerSource: input.TriggerSource, TriggerScene: sanitizedOptionalText(input.TriggerScene), TriggerSeries: input.TriggerSeries, ScheduledAt: input.ScheduledAt, - StartedAt: input.StartedAt, Attempt: input.Attempt, Result: result, + StartedAt: input.StartedAt, Attempt: input.Attempt, Result: result, StateChanged: input.StateChanged, RequestSummary: requestSummary, Metadata: metadata, RequestID: input.RequestID, CorrelationID: input.CorrelationID, AuditEventID: input.AuditEventID, RecoveryStrategy: sanitizedOptionalText(input.RecoveryStrategy), @@ -416,8 +417,8 @@ func validateAttempt(input Attempt) error { if input.Direction != constants.IntegrationDirectionInbound && input.Direction != constants.IntegrationDirectionOutbound { return pkgerrors.New(pkgerrors.CodeInvalidParam, "Integration Log 方向无效") } - if input.InitialResult != "" && input.InitialResult != constants.IntegrationResultPending && !isUnsentResult(input.InitialResult) { - return pkgerrors.New(pkgerrors.CodeInvalidParam, "Integration Log 初始结果只能是待处理或未发送终态") + if input.InitialResult != "" && input.InitialResult != constants.IntegrationResultPending && !isTerminalResult(input.InitialResult) { + return pkgerrors.New(pkgerrors.CodeInvalidParam, "Integration Log 初始结果无效") } if !validGeneratedString(input.IntegrationID, constants.IntegrationIDMaxLength) || !validOptionalString(input.TriggerSeries, constants.IntegrationTriggerSeriesMaxLength) || diff --git a/internal/service/polling/concurrency_service.go b/internal/service/polling/concurrency_service.go index 83037ab..6188d82 100644 --- a/internal/service/polling/concurrency_service.go +++ b/internal/service/polling/concurrency_service.go @@ -127,8 +127,8 @@ func (s *ConcurrencyService) GetByTaskType(ctx context.Context, taskType string) // UpdateMaxConcurrency 更新最大并发数 func (s *ConcurrencyService) UpdateMaxConcurrency(ctx context.Context, taskType string, maxConcurrency int, updatedBy uint) error { // 验证参数 - if maxConcurrency < 1 { - return errors.New(errors.CodeInvalidParam, "并发数必须为正整数") + if maxConcurrency < 1 || maxConcurrency > constants.PollingMaxConcurrencyLimit { + return errors.New(errors.CodeInvalidParam, "并发数必须为 1-1000") } // 验证任务类型存在 diff --git a/internal/task/audit_daily_archive.go b/internal/task/audit_daily_archive.go index ea9d626..e80131f 100644 --- a/internal/task/audit_daily_archive.go +++ b/internal/task/audit_daily_archive.go @@ -22,15 +22,20 @@ type AuditDailyArchivePayload struct { type AuditDailyArchiveHandler struct { service *auditarchive.Service logger *zap.Logger + enabled bool } // NewAuditDailyArchiveHandler 创建统一审计每日冷归档任务处理器。 -func NewAuditDailyArchiveHandler(service *auditarchive.Service, logger *zap.Logger) *AuditDailyArchiveHandler { - return &AuditDailyArchiveHandler{service: service, logger: logger} +func NewAuditDailyArchiveHandler(service *auditarchive.Service, logger *zap.Logger, enabled bool) *AuditDailyArchiveHandler { + return &AuditDailyArchiveHandler{service: service, logger: logger, enabled: enabled} } // Handle 执行前一完整自然日归档，或按任务载荷补档指定自然日。 func (h *AuditDailyArchiveHandler) Handle(ctx context.Context, task *asynq.Task) error { + if !h.enabled { + h.logger.Info("统一审计日归档已停用，跳过任务") + return nil + } if h.service == nil { return fmt.Errorf("统一审计归档服务未配置") } diff --git a/internal/task/audit_monthly_retention.go b/internal/task/audit_monthly_retention.go index e9ec063..1ab0d27 100644 --- a/internal/task/audit_monthly_retention.go +++ b/internal/task/audit_monthly_retention.go @@ -16,15 +16,20 @@ type AuditRetentionHandler struct { service *auditarchive.Service logger *zap.Logger cleanupEnabled bool + enabled bool } // NewAuditRetentionHandler 创建日留存处理器。 -func NewAuditRetentionHandler(service *auditarchive.Service, logger *zap.Logger, cleanupEnabled bool) *AuditRetentionHandler { - return &AuditRetentionHandler{service: service, logger: logger, cleanupEnabled: cleanupEnabled} +func NewAuditRetentionHandler(service *auditarchive.Service, logger *zap.Logger, cleanupEnabled, enabled bool) *AuditRetentionHandler { + return &AuditRetentionHandler{service: service, logger: logger, cleanupEnabled: cleanupEnabled, enabled: enabled} } // Handle 在关闭清理开关时只读校验，开启后从最早未完成日连续删除。 func (h *AuditRetentionHandler) Handle(ctx context.Context, _ *asynq.Task) error { + if !h.enabled { + h.logger.Info("日志日留存已停用，跳过任务") + return nil + } if h.service == nil { return fmt.Errorf("日志日留存服务未配置") } diff --git a/internal/task/integration_archive.go b/internal/task/integration_archive.go index 298ff40..2ff21e9 100644 --- a/internal/task/integration_archive.go +++ b/internal/task/integration_archive.go @@ -22,15 +22,20 @@ type IntegrationDailyArchivePayload struct { type IntegrationArchiveHandler struct { service *auditarchive.Service logger *zap.Logger + enabled bool } // NewIntegrationArchiveHandler 创建 Integration Log 归档任务处理器。 -func NewIntegrationArchiveHandler(service *auditarchive.Service, logger *zap.Logger) *IntegrationArchiveHandler { - return &IntegrationArchiveHandler{service: service, logger: logger} +func NewIntegrationArchiveHandler(service *auditarchive.Service, logger *zap.Logger, enabled bool) *IntegrationArchiveHandler { + return &IntegrationArchiveHandler{service: service, logger: logger, enabled: enabled} } // HandleDaily 执行前一完整自然日归档，或按任务载荷补档指定自然日。 func (h *IntegrationArchiveHandler) HandleDaily(ctx context.Context, task *asynq.Task) error { + if !h.enabled { + h.logger.Info("Integration Log 日归档已停用，跳过任务") + return nil + } if h.service == nil { return fmt.Errorf("Integration Log 归档服务未配置") } diff --git a/internal/task/polling_base.go b/internal/task/polling_base.go index 0d8d4b0..2411a7b 100644 --- a/internal/task/polling_base.go +++ b/internal/task/polling_base.go @@ -16,28 +16,32 @@ import ( // acquireConcurrencyScript 原子获取并发信号量的 Lua 脚本 // INCR + EXPIRE 合并为单个服务端操作，消除二者之间的崩溃窗口： // 若 Worker 在 INCR 后、EXPIRE 前崩溃，key 将永久留在 Redis 导致计数器卡死。 -// KEYS[1]: 当前并发计数 key -// ARGV[1]: 最大并发数；ARGV[2]: key TTL（秒） -// 返回 -1 表示超额拒绝，>0 表示成功获取后的计数值 +// KEYS[1]: 全部轮询计数；KEYS[2]: 分类轮询计数 +// ARGV[1]: 总量上限；ARGV[2]: 分类上限；ARGV[3]: key TTL（秒） +// 返回 -1 表示任一上限超额，>0 表示成功获取后的总计数值 var acquireConcurrencyScript = redis.NewScript(` - local current = redis.call('INCR', KEYS[1]) - if tonumber(current) > tonumber(ARGV[1]) then + local total = redis.call('INCR', KEYS[1]) + local kind = redis.call('INCR', KEYS[2]) + if tonumber(total) > tonumber(ARGV[1]) or tonumber(kind) > tonumber(ARGV[2]) then redis.call('DECR', KEYS[1]) + redis.call('DECR', KEYS[2]) return -1 end - redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2])) - return current + redis.call('EXPIRE', KEYS[1], tonumber(ARGV[3])) + redis.call('EXPIRE', KEYS[2], tonumber(ARGV[3])) + return total `) var releaseConcurrencyScript = redis.NewScript(` - local current = tonumber(redis.call('GET', KEYS[1]) or '0') or 0 - if current <= 0 then - if redis.call('EXISTS', KEYS[1]) == 1 then - redis.call('SET', KEYS[1], 0, 'KEEPTTL') + for _, key in ipairs(KEYS) do + local current = tonumber(redis.call('GET', key) or '0') or 0 + if current <= 0 then + if redis.call('EXISTS', key) == 1 then redis.call('SET', key, 0, 'KEEPTTL') end + else + redis.call('DECR', key) end - return 0 end - return redis.call('DECR', KEYS[1]) + return 0 `) const pollingFallbackOperationTimeout = 5 * time.Second @@ -52,13 +56,14 @@ func pollingFallbackContext() (context.Context, context.CancelFunc) { // PollingBase 轮询共享基类 // 封装并发控制、卡缓存、重入队、配置间隔查询等公共方法，所有 Handler 共享 type PollingBase struct { - redis *redis.Client - queueMgr *polling.PollingQueueManager - configMgr *polling.PollingConfigManager - iotCardStore *postgres.IotCardStore - logger *zap.Logger - verboseLog bool - trafficLock *cardtrafficlock.Lock + redis *redis.Client + queueMgr *polling.PollingQueueManager + configMgr *polling.PollingConfigManager + iotCardStore *postgres.IotCardStore + logger *zap.Logger + verboseLog bool + totalMaxConcurrency int + trafficLock *cardtrafficlock.Lock } // NewPollingBase 创建轮询共享基类 @@ -69,15 +74,17 @@ func NewPollingBase( iotCardStore *postgres.IotCardStore, logger *zap.Logger, verboseLog bool, + totalMaxConcurrency int, ) *PollingBase { return &PollingBase{ - redis: redisClient, - queueMgr: queueMgr, - configMgr: configMgr, - iotCardStore: iotCardStore, - logger: logger, - verboseLog: verboseLog, - trafficLock: cardtrafficlock.New(redisClient), + redis: redisClient, + queueMgr: queueMgr, + configMgr: configMgr, + iotCardStore: iotCardStore, + logger: logger, + verboseLog: verboseLog, + totalMaxConcurrency: totalMaxConcurrency, + trafficLock: cardtrafficlock.New(redisClient), } } @@ -88,22 +95,33 @@ func (b *PollingBase) acquireConcurrency(ctx context.Context, taskType string) b shortType := shortTaskType(taskType) configKey := constants.RedisPollingConcurrencyConfigKey(shortType) currentKey := constants.RedisPollingConcurrencyCurrentKey(taskType) + totalKey := constants.RedisPollingConcurrencyTotalCurrentKey() maxConcurrency, err := b.redis.Get(ctx, configKey).Int() - if err != nil { + if err != nil || maxConcurrency < 1 || maxConcurrency > constants.PollingMaxConcurrencyLimit { maxConcurrency = constants.PollingDefaultMaxConcurrency } + totalMax := b.totalMaxConcurrency + if totalMax < 1 || totalMax > constants.PollingMaxConcurrencyLimit { + totalMax = constants.PollingDefaultTotalMaxConcurrency + } result, err := acquireConcurrencyScript.Run( - ctx, b.redis, []string{currentKey}, - maxConcurrency, constants.PollingConcurrencyKeyTTL, + ctx, b.redis, []string{totalKey, currentKey}, + totalMax, maxConcurrency, constants.PollingConcurrencyKeyTTL, ).Int64() if err != nil { b.logger.Warn("获取并发计数失败，放行任务", zap.Error(err)) return true } - return result != -1 + if result == -1 { + b.logger.Info("轮询因并发令牌不足延后", zap.String("task_type", taskType), + zap.Int("max_concurrency", maxConcurrency), zap.Int("total_max_concurrency", totalMax), + zap.String("metric", "polling.deferred.concurrency_limit")) + return false + } + return true } // releaseConcurrency 释放并发信号量 @@ -112,7 +130,7 @@ func (b *PollingBase) releaseConcurrency(_ context.Context, taskType string) { defer cancel() currentKey := constants.RedisPollingConcurrencyCurrentKey(taskType) - if err := releaseConcurrencyScript.Run(ctx, b.redis, []string{currentKey}).Err(); err != nil { + if err := releaseConcurrencyScript.Run(ctx, b.redis, []string{constants.RedisPollingConcurrencyTotalCurrentKey(), currentKey}).Err(); err != nil { b.logger.Warn("释放并发计数失败", zap.String("task_type", taskType), zap.Error(err)) } } diff --git a/internal/task/polling_carddata_handler.go b/internal/task/polling_carddata_handler.go index d0e9179..f069f7c 100644 --- a/internal/task/polling_carddata_handler.go +++ b/internal/task/polling_carddata_handler.go @@ -60,27 +60,16 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e if h.gateway == nil { return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata) } - attemptStartedAt := time.Now() - attempt, err := startGatewayAttempt(ctx, h.integration, cardID, constants.IntegrationOperationGatewayTraffic, constants.CardObservationSceneTrafficPolling) - if err != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "建立 Gateway 流量 Integration Log 失败", err) - } + attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayTraffic, constants.CardObservationSceneTrafficPolling) result, err := h.gateway.QueryFlow(ctx, &gateway.FlowQueryReq{CardNo: card.ICCID}) if err != nil { - if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 流量 Integration Log 失败", logErr) - } + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询流量失败") return h.failAndRequeue(ctx, cardID, startedAt, "查询流量失败", err) } if strings.TrimSpace(result.ICCID) == "" { - if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 流量 Integration Log 失败", logErr) - } + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "流量查询响应缺少 ICCID") return h.failAndRequeue(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil) } - if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 流量 Integration Log 失败", logErr) - } ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingCarddata, "卡流量轮询任务", attempt.IntegrationID) if h.observation == nil || h.carrier == nil { return h.failAndRequeue(ctx, cardID, startedAt, "卡流量观测能力未配置", nil) @@ -94,13 +83,23 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e }, }) if err != nil { + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultUnknown, false, "应用流量观测失败") return h.failAndRequeue(ctx, cardID, startedAt, "应用流量观测失败", err) } + changed := decision.IncrementMB > 0 || decision.CrossMonth + if changed { + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultSuccess, true, "流量业务事实发生变化") + } if h.base.verboseLog { h.base.logger.Info("流量轮询详情", zap.Uint("card_id", cardID), zap.String("iccid", card.ICCID), zap.Float64("gateway_flow_mb", float64(result.Used)), zap.Float64("increment_mb", decision.IncrementMB), zap.Bool("is_cross_month", decision.CrossMonth), zap.Bool("reading_accepted", decision.ReadingAccepted)) } + metric := "polling.observation.unchanged" + if changed { + metric = "polling.observation.persisted" + } + h.base.logger.Info("流量轮询观测完成", zap.Uint("card_id", cardID), zap.Bool("changed", changed), zap.String("metric", metric)) h.base.updateStats(ctx, constants.TaskTypePollingCarddata, true, time.Since(startedAt)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata) } diff --git a/internal/task/polling_cardstatus_handler.go b/internal/task/polling_cardstatus_handler.go index 33764f6..9c64beb 100644 --- a/internal/task/polling_cardstatus_handler.go +++ b/internal/task/polling_cardstatus_handler.go @@ -59,27 +59,16 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task) if h.gateway == nil { return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) } - attemptStartedAt := time.Now() - attempt, err := startGatewayAttempt(ctx, h.integration, cardID, constants.IntegrationOperationGatewayNetwork, constants.CardObservationSceneNetworkPolling) - if err != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "建立 Gateway 网络 Integration Log 失败", err) - } + attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayNetwork, constants.CardObservationSceneNetworkPolling) result, err := h.gateway.QueryCardStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID}) if err != nil { - if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 网络 Integration Log 失败", logErr) - } + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询卡状态失败") return h.failAndRequeue(ctx, cardID, startedAt, "查询卡状态失败", err) } if strings.TrimSpace(result.ICCID) == "" { - if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 网络 Integration Log 失败", logErr) - } + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "卡状态查询响应缺少 ICCID") return h.failAndRequeue(ctx, cardID, startedAt, "卡状态查询响应缺少 ICCID", nil) } - if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 网络 Integration Log 失败", logErr) - } ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingCardStatus, "卡网络状态轮询任务", attempt.IntegrationID) if h.observation == nil { return h.failAndRequeue(ctx, cardID, startedAt, "卡网络观测能力未配置", nil) @@ -93,14 +82,24 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task) }, }) if err != nil { + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultUnknown, false, "应用网络状态观测失败") return h.failAndRequeue(ctx, cardID, startedAt, "应用网络状态观测失败", err) } + changed := decision.StatusChanged || decision.StopReasonChanged || decision.StopPolling + if changed { + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultSuccess, true, "网络状态发生变化") + } if h.base.verboseLog || !decision.StatusKnown { h.base.logger.Info("卡状态轮询详情", zap.Uint("card_id", cardID), zap.String("iccid", card.ICCID), zap.String("card_status", result.CardStatus), zap.String("extend", strings.TrimSpace(result.Extend)), zap.Int("new_network_status", decision.AfterStatus), zap.Bool("status_known", decision.StatusKnown), zap.Bool("changed", decision.StatusChanged), zap.Bool("stop_polling", decision.StopPolling)) } + metric := "polling.observation.unchanged" + if changed { + metric = "polling.observation.persisted" + } + h.base.logger.Info("卡状态轮询观测完成", zap.Uint("card_id", cardID), zap.Bool("changed", changed), zap.String("metric", metric)) h.base.updateStats(ctx, constants.TaskTypePollingCardStatus, true, time.Since(startedAt)) if decision.StopPolling { h.base.logger.Info("独立卡命中风险状态，已关闭轮询", zap.Uint("card_id", cardID), zap.String("gateway_extend", decision.GatewayExtend)) diff --git a/internal/task/polling_integration_log.go b/internal/task/polling_integration_log.go index 6564ffa..1b11c36 100644 --- a/internal/task/polling_integration_log.go +++ b/internal/task/polling_integration_log.go @@ -13,24 +13,30 @@ import ( "github.com/break/junhong_cmp_fiber/pkg/errors" ) -// startGatewayAttempt 在实际调用 Gateway 前建立 Integration Log 尝试事实。 -func startGatewayAttempt(ctx context.Context, repository *integrationlog.Repository, cardID uint, operation, scene string) (*model.IntegrationLog, error) { - if repository == nil { - return nil, errors.New(errors.CodeInternalError, "Gateway Integration Log 未配置") - } +// newGatewayAttempt 只在内存中准备稳定关联标识；成功无变化的轮询不落库。 +func newGatewayAttempt(cardID uint, operation, scene string) integrationlog.Attempt { resourceID := strconv.FormatUint(uint64(cardID), 10) integrationID := uuid.NewString() triggerSource := constants.CardObservationSourcePolling triggerScene := scene - return repository.Start(ctx, integrationlog.Attempt{ - IntegrationID: integrationID, - Provider: constants.IntegrationProviderGateway, Direction: constants.IntegrationDirectionOutbound, + return integrationlog.Attempt{IntegrationID: integrationID, + Provider: constants.IntegrationProviderGateway, Direction: constants.IntegrationDirectionOutbound, Operation: operation, ResourceType: constants.AssetTypeIotCard, ResourceID: &resourceID, TriggerSource: &triggerSource, TriggerScene: &triggerScene, TriggerSeries: &integrationID, - CorrelationID: &integrationID, - RequestSummary: map[string]any{"card_id": cardID}, Metadata: map[string]any{"scene": scene}, - InitialResult: constants.IntegrationResultPending, - }) + CorrelationID: &integrationID, RequestSummary: map[string]any{"card_id": cardID}, + Metadata: map[string]any{"scene": scene}} +} + +// recordGatewayAttempt 在结果明确后才写入外部交互日志。 +func recordGatewayAttempt(ctx context.Context, repository *integrationlog.Repository, attempt integrationlog.Attempt, result string, changed bool, summary string) error { + if repository == nil { + return errors.New(errors.CodeInternalError, "Gateway Integration Log 未配置") + } + attempt.InitialResult = result + attempt.StateChanged = changed + attempt.Metadata = map[string]any{"scene": attempt.TriggerScene, "summary": summary} + _, err := repository.Start(ctx, attempt) + return err } // completeGatewayAttempt 记录 Gateway 查询成功或明确失败，不把响应正文写入日志。 diff --git a/internal/task/polling_realname_handler.go b/internal/task/polling_realname_handler.go index 0cbe203..cc3a7dc 100644 --- a/internal/task/polling_realname_handler.go +++ b/internal/task/polling_realname_handler.go @@ -51,27 +51,16 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) e if card.CardCategory == constants.CardCategoryIndustry || h.gateway == nil { return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) } - attemptStartedAt := time.Now() - attempt, err := startGatewayAttempt(ctx, h.integration, cardID, constants.IntegrationOperationGatewayRealname, constants.CardObservationSceneRealnamePolling) - if err != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "建立 Gateway 实名 Integration Log 失败", err) - } + attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayRealname, constants.CardObservationSceneRealnamePolling) result, err := h.gateway.QueryRealnameStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID}) if err != nil { - if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 实名 Integration Log 失败", logErr) - } + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询实名状态失败") return h.failAndRequeue(ctx, cardID, startedAt, "查询实名状态失败", err) } if strings.TrimSpace(result.ICCID) == "" { - if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 实名 Integration Log 失败", logErr) - } + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "实名查询响应缺少 ICCID") return h.failAndRequeue(ctx, cardID, startedAt, "实名查询响应缺少 ICCID", nil) } - if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 实名 Integration Log 失败", logErr) - } ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingRealname, "实名状态轮询任务", attempt.IntegrationID) if h.observation == nil { return h.failAndRequeue(ctx, cardID, startedAt, "卡实名观测能力未配置", nil) @@ -85,13 +74,23 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) e }, }) if err != nil { + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultUnknown, false, "应用实名观测失败") return h.failAndRequeue(ctx, cardID, startedAt, "应用实名观测失败", err) } + changed := decision.StatusChanged + if changed { + _ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultSuccess, true, "实名状态发生变化") + } if h.base.verboseLog { h.base.logger.Info("实名状态轮询详情", zap.Uint("card_id", cardID), zap.String("iccid", card.ICCID), zap.Bool("real_status", result.RealStatus), zap.Int("new_status", decision.AfterStatus), zap.Bool("changed", decision.StatusChanged), zap.Bool("reversal_pending", decision.ReversalPending)) } + metric := "polling.observation.unchanged" + if changed { + metric = "polling.observation.persisted" + } + h.base.logger.Info("实名轮询观测完成", zap.Uint("card_id", cardID), zap.Bool("changed", changed), zap.String("metric", metric)) h.base.updateStats(ctx, constants.TaskTypePollingRealname, true, time.Since(startedAt)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) } diff --git a/pkg/config/config.go b/pkg/config/config.go index 0526678..ba698e6 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -169,7 +169,9 @@ type GatewayConfig struct { type WorkerConfig struct { Role string `mapstructure:"role"` // Worker 运行角色：all、leader、consumer InstanceName string `mapstructure:"instance_name"` // Worker 实例名称，用于多实例日志区分 + PollingTotalMaxConcurrency int `mapstructure:"polling_total_max_concurrency"` // 全部轮询总并发上限 AuditRetentionCleanupEnabled bool `mapstructure:"audit_retention_cleanup_enabled"` // 是否启用审计日志物理清理 + AuditArchiveTasksEnabled bool `mapstructure:"audit_archive_tasks_enabled"` // 是否启用审计归档与日留存任务 } // ApprovalConfig 审批新旧入口切换配置。 @@ -382,6 +384,10 @@ func (c *Config) Validate() error { constants.WorkerRoleLeader: true, constants.WorkerRoleConsumer: true, } + if c.Worker.PollingTotalMaxConcurrency < 1 || c.Worker.PollingTotalMaxConcurrency > constants.PollingMaxConcurrencyLimit { + return fmt.Errorf("invalid configuration: worker.polling_total_max_concurrency: must be 1-%d (current value: %d)", constants.PollingMaxConcurrencyLimit, c.Worker.PollingTotalMaxConcurrency) + } + if !validWorkerRoles[c.Worker.Role] { return fmt.Errorf( "invalid configuration: worker.role: invalid worker role (current value: %s, expected: %s, %s, %s)", diff --git a/pkg/config/defaults/config.yaml b/pkg/config/defaults/config.yaml index 1fc0b30..24da7f4 100644 --- a/pkg/config/defaults/config.yaml +++ b/pkg/config/defaults/config.yaml @@ -143,8 +143,11 @@ polling_auto_trigger: worker: role: "all" instance_name: "" + polling_total_max_concurrency: 100 # 日留存只读演练验收通过前必须保持关闭 audit_retention_cleanup_enabled: false + # 审计归档、Integration 归档与日留存默认关闭，低峰期按单日推进 + audit_archive_tasks_enabled: false # 审批新旧入口切换配置 approval: diff --git a/pkg/config/loader.go b/pkg/config/loader.go index 9f81f4d..0775d23 100644 --- a/pkg/config/loader.go +++ b/pkg/config/loader.go @@ -135,7 +135,9 @@ func bindEnvVariables(v *viper.Viper) { "polling_auto_trigger.auto_trigger_system_user_id", "worker.role", "worker.instance_name", + "worker.polling_total_max_concurrency", "worker.audit_retention_cleanup_enabled", + "worker.audit_archive_tasks_enabled", "approval.legacy_refund_manual_enabled", "approval.legacy_offline_recharge_pay_enabled", "wecom.base_url", diff --git a/pkg/constants/polling.go b/pkg/constants/polling.go index 246951c..ab1280c 100644 --- a/pkg/constants/polling.go +++ b/pkg/constants/polling.go @@ -18,6 +18,12 @@ const PollingDequeueMaxBatchSize = 7000 // 300：支撑 10M 卡、2小时 carddata 间隔（Gateway 200ms → 300并发 = 1500任务/秒/类型） const PollingDefaultMaxConcurrency = 300 +// PollingDefaultTotalMaxConcurrency 全部轮询任务共享的默认总并发上限。 +const PollingDefaultTotalMaxConcurrency = 100 + +// PollingMaxConcurrencyLimit 轮询并发配置的代码安全上限。 +const PollingMaxConcurrencyLimit = 1000 + // PollingConcurrencyKeyTTL 并发计数 key 的过期时间（秒） // 保证 Worker 意外崩溃（defer Decr 未执行）时计数器在此时间内自动归零 const PollingConcurrencyKeyTTL = 600 // 10 分钟 diff --git a/pkg/constants/redis.go b/pkg/constants/redis.go index 7c8ed14..9c40667 100644 --- a/pkg/constants/redis.go +++ b/pkg/constants/redis.go @@ -289,6 +289,11 @@ func RedisPollingConcurrencyCurrentKey(taskType string) string { return fmt.Sprintf("polling:concurrency:current:%s", taskType) } +// RedisPollingConcurrencyTotalCurrentKey 返回全部轮询共享的并发计数键。 +func RedisPollingConcurrencyTotalCurrentKey() string { + return "polling:concurrency:current:total" +} + // RedisPollingManualQueueKey 生成手动触发队列的 Redis 键 // 用途：List 存储手动触发的卡 ID（FIFO 队列） // 过期时间：无（临时队列）