csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

缓解io压力

Browse Source
This commit is contained in:
break
2026-09-02 16:14:27 +08:00
parent 395e5fb47c
commit dbfeeee253
17 changed files with 181 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
cmd/worker/main.go
View File

@@ -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。

42
internal/application/auditarchive/retention.go
View File

@@ -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 校验并按需清理指定已结束自然日，供受控演练使用。

7
internal/infrastructure/integrationlog/repository.go
View File

@@ -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) ||

4
internal/service/polling/concurrency_service.go
View File

@@ -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")
}
// 验证任务类型存在

9
internal/task/audit_daily_archive.go
View File

@@ -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("统一审计归档服务未配置")
}

9
internal/task/audit_monthly_retention.go
View File

@@ -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("日志日留存服务未配置")
}

9
internal/task/integration_archive.go
View File

@@ -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 归档服务未配置")
}

82
internal/task/polling_base.go
View File

@@ -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))
}
}

27
internal/task/polling_carddata_handler.go
View File

@@ -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)
}

27
internal/task/polling_cardstatus_handler.go
View File

@@ -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))

30
internal/task/polling_integration_log.go
View File

@@ -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 查询成功或明确失败，不把响应正文写入日志。

27
internal/task/polling_realname_handler.go
View File

@@ -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)
}

6
pkg/config/config.go
View File

@@ -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)",

3
pkg/config/defaults/config.yaml
View File

@@ -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:

2
pkg/config/loader.go
View File

@@ -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",

6
pkg/constants/polling.go
View File

@@ -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 分钟

5
pkg/constants/redis.go
View File

@@ -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 队列）
// 过期时间：无（临时队列）