diff --git a/internal/config/config.go b/internal/config/config.go index 98a0949..1186828 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -439,7 +439,7 @@ func Load() Config { SecretID: getEnvAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_COS_SECRET_ID", "TASK_CENTER_ARCHIVE_COS_SECRET_ID", "LIKEI_TENCENT_COS_SECRET_ID"}, ""), SecretKey: getEnvAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_COS_SECRET_KEY", "TASK_CENTER_ARCHIVE_COS_SECRET_KEY", "LIKEI_TENCENT_COS_SECRET_KEY"}, ""), Prefix: getEnvAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_COS_PREFIX", "TASK_CENTER_ARCHIVE_COS_PREFIX"}, "task-center-events/v1"), - BatchSize: getEnvIntAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_BATCH_SIZE", "TASK_CENTER_ARCHIVE_BATCH_SIZE"}, 10000), + BatchSize: getEnvIntAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_BATCH_SIZE", "TASK_CENTER_ARCHIVE_BATCH_SIZE"}, 500), FlushSeconds: getEnvIntAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_FLUSH_SECONDS", "TASK_CENTER_ARCHIVE_FLUSH_SECONDS"}, 60), Workers: getEnvIntAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_WORKERS", "TASK_CENTER_ARCHIVE_WORKERS"}, 2), }, diff --git a/internal/model/taskcenter_models.go b/internal/model/taskcenter_models.go index 83c78f4..2a3aef1 100644 --- a/internal/model/taskcenter_models.go +++ b/internal/model/taskcenter_models.go @@ -116,19 +116,19 @@ func (TaskCenterEventDedup) TableName() string { return "task_center_event_dedup // TaskCenterEventArchiveOutbox 保存待归档到 COS 的原始事件。 type TaskCenterEventArchiveOutbox struct { - ID int64 `gorm:"column:id;primaryKey"` + ID int64 `gorm:"column:id;primaryKey;index:idx_task_center_archive_claim_create,priority:3;index:idx_task_center_archive_claim_stale,priority:3"` SysOrigin string `gorm:"column:sys_origin;size:32;uniqueIndex:uk_task_center_archive_event,priority:1"` EventID string `gorm:"column:event_id;size:128;uniqueIndex:uk_task_center_archive_event,priority:2"` EventType string `gorm:"column:event_type;size:64"` UserID int64 `gorm:"column:user_id"` OccurredAt time.Time `gorm:"column:occurred_at"` RawJSON string `gorm:"column:raw_json;type:text"` - Status string `gorm:"column:status;size:32;index:idx_task_center_archive_status_time,priority:1"` + Status string `gorm:"column:status;size:32;index:idx_task_center_archive_status_time,priority:1;index:idx_task_center_archive_claim_create,priority:1;index:idx_task_center_archive_claim_stale,priority:1"` RetryCount int `gorm:"column:retry_count"` COSKey string `gorm:"column:cos_key;size:512"` FailureReason string `gorm:"column:failure_reason;size:512"` - CreateTime time.Time `gorm:"column:create_time;index:idx_task_center_archive_status_time,priority:2"` - UpdateTime time.Time `gorm:"column:update_time"` + CreateTime time.Time `gorm:"column:create_time;index:idx_task_center_archive_status_time,priority:2;index:idx_task_center_archive_claim_create,priority:2"` + UpdateTime time.Time `gorm:"column:update_time;index:idx_task_center_archive_claim_stale,priority:2"` ArchivedAt *time.Time `gorm:"column:archived_at"` } diff --git a/internal/service/taskcenter/archive.go b/internal/service/taskcenter/archive.go index 4c82603..89187bb 100644 --- a/internal/service/taskcenter/archive.go +++ b/internal/service/taskcenter/archive.go @@ -20,6 +20,11 @@ import ( "gorm.io/gorm/clause" ) +const ( + defaultTaskCenterArchiveBatchSize = 500 + maxTaskCenterArchiveBatchSize = 1000 +) + // StartArchiveWorker 启动任务中心事件 COS 归档 worker。 func (s *Service) StartArchiveWorker(ctx context.Context) error { if !s.cfg.TaskCenter.Archive.Enabled { @@ -68,10 +73,7 @@ func (s *Service) runArchiveWorker(ctx context.Context, workerIndex int) { } func (s *Service) archivePendingEvents(ctx context.Context) error { - batchSize := s.cfg.TaskCenter.Archive.BatchSize - if batchSize <= 0 { - batchSize = 10000 - } + batchSize := normalizeTaskCenterArchiveBatchSize(s.cfg.TaskCenter.Archive.BatchSize) rows, err := s.claimArchiveRows(ctx, batchSize) if err != nil { return err @@ -97,16 +99,30 @@ func (s *Service) claimArchiveRows(ctx context.Context, batchSize int) ([]model. var rows []model.TaskCenterEventArchiveOutbox err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { staleProcessingBefore := time.Now().Add(-10 * time.Minute) - if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}). - Where("status IN ? OR (status = ? AND update_time < ?)", - []string{ArchiveStatusPending, ArchiveStatusFailed}, - ArchiveStatusProcessing, - staleProcessingBefore, - ). - Order("create_time asc, id asc"). - Limit(batchSize). - Find(&rows).Error; err != nil { - return err + staleLimit, _ := taskCenterArchiveClaimLimits(batchSize) + if staleLimit > 0 { + var staleRows []model.TaskCenterEventArchiveOutbox + // PROCESSING 超时行保留固定接管额度，避免 PENDING/FAILED 持续堆积时旧锁永久饥饿。 + if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}). + Where("status = ? AND update_time < ?", ArchiveStatusProcessing, staleProcessingBefore). + Order("update_time asc, id asc"). + Limit(staleLimit). + Find(&staleRows).Error; err != nil { + return err + } + rows = append(rows, staleRows...) + } + if remaining := batchSize - len(rows); remaining > 0 { + var retryableRows []model.TaskCenterEventArchiveOutbox + // PENDING/FAILED 只按状态和创建时间抢占，避免和 PROCESSING 超时接管揉成 OR 后放大扫描。 + if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}). + Where("status IN ?", []string{ArchiveStatusPending, ArchiveStatusFailed}). + Order("create_time asc, id asc"). + Limit(remaining). + Find(&retryableRows).Error; err != nil { + return err + } + rows = append(rows, retryableRows...) } if len(rows) == 0 { return nil @@ -124,6 +140,30 @@ func (s *Service) claimArchiveRows(ctx context.Context, batchSize int) ([]model. return rows, nil } +func normalizeTaskCenterArchiveBatchSize(batchSize int) int { + if batchSize <= 0 { + return defaultTaskCenterArchiveBatchSize + } + if batchSize > maxTaskCenterArchiveBatchSize { + return maxTaskCenterArchiveBatchSize + } + return batchSize +} + +func taskCenterArchiveClaimLimits(batchSize int) (int, int) { + if batchSize <= 1 { + return batchSize, 0 + } + staleLimit := batchSize / 10 + if staleLimit < 1 { + staleLimit = 1 + } + if staleLimit > 100 { + staleLimit = 100 + } + return staleLimit, batchSize - staleLimit +} + func (s *Service) newCOSClient() (*cos.Client, error) { bucketURL, err := url.Parse(fmt.Sprintf("https://%s.cos.%s.myqcloud.com", strings.TrimSpace(s.cfg.TaskCenter.Archive.Bucket), diff --git a/internal/service/taskcenter/taskcenter_test.go b/internal/service/taskcenter/taskcenter_test.go index de438c5..85a6ac2 100644 --- a/internal/service/taskcenter/taskcenter_test.go +++ b/internal/service/taskcenter/taskcenter_test.go @@ -42,6 +42,51 @@ func newTestService(t *testing.T, java taskCenterJavaGateway) (*Service, *gorm.D return service, db } +func TestNormalizeTaskCenterArchiveBatchSize(t *testing.T) { + tests := []struct { + name string + input int + want int + }{ + {name: "default", input: 0, want: defaultTaskCenterArchiveBatchSize}, + {name: "negative default", input: -1, want: defaultTaskCenterArchiveBatchSize}, + {name: "keeps configured value", input: 200, want: 200}, + {name: "caps large batch", input: 10000, want: maxTaskCenterArchiveBatchSize}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := normalizeTaskCenterArchiveBatchSize(tt.input); got != tt.want { + t.Fatalf("normalizeTaskCenterArchiveBatchSize(%d) = %d, want %d", tt.input, got, tt.want) + } + }) + } +} + +func TestTaskCenterArchiveClaimLimitsReserveStaleRows(t *testing.T) { + tests := []struct { + name string + batchSize int + wantStale int + wantRetryable int + }{ + {name: "single row", batchSize: 1, wantStale: 1, wantRetryable: 0}, + {name: "small batch keeps one stale slot", batchSize: 5, wantStale: 1, wantRetryable: 4}, + {name: "normal batch uses ten percent", batchSize: 500, wantStale: 50, wantRetryable: 450}, + {name: "large batch caps stale slot", batchSize: 1000, wantStale: 100, wantRetryable: 900}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotStale, gotRetryable := taskCenterArchiveClaimLimits(tt.batchSize) + if gotStale != tt.wantStale || gotRetryable != tt.wantRetryable { + t.Fatalf("taskCenterArchiveClaimLimits(%d) = (%d, %d), want (%d, %d)", + tt.batchSize, gotStale, gotRetryable, tt.wantStale, tt.wantRetryable) + } + }) + } +} + func TestDailyTaskConfigEventListAndClaim(t *testing.T) { java := &taskCenterTestGateway{events: map[string]bool{}} service, db := newTestService(t, java) diff --git a/migrations/056_task_center_archive_outbox_claim_indexes.sql b/migrations/056_task_center_archive_outbox_claim_indexes.sql new file mode 100644 index 0000000..b76f4bb --- /dev/null +++ b/migrations/056_task_center_archive_outbox_claim_indexes.sql @@ -0,0 +1,29 @@ +SET @current_schema := DATABASE(); + +SET @add_task_center_archive_claim_create_sql := IF( + NOT EXISTS ( + SELECT 1 FROM INFORMATION_SCHEMA.STATISTICS + WHERE table_schema = @current_schema + AND table_name = 'task_center_event_archive_outbox' + AND index_name = 'idx_task_center_archive_claim_create' + ), + 'ALTER TABLE `task_center_event_archive_outbox` ADD INDEX `idx_task_center_archive_claim_create` (`status`, `create_time`, `id`)', + 'SELECT 1' +); +PREPARE add_task_center_archive_claim_create_stmt FROM @add_task_center_archive_claim_create_sql; +EXECUTE add_task_center_archive_claim_create_stmt; +DEALLOCATE PREPARE add_task_center_archive_claim_create_stmt; + +SET @add_task_center_archive_claim_stale_sql := IF( + NOT EXISTS ( + SELECT 1 FROM INFORMATION_SCHEMA.STATISTICS + WHERE table_schema = @current_schema + AND table_name = 'task_center_event_archive_outbox' + AND index_name = 'idx_task_center_archive_claim_stale' + ), + 'ALTER TABLE `task_center_event_archive_outbox` ADD INDEX `idx_task_center_archive_claim_stale` (`status`, `update_time`, `id`)', + 'SELECT 1' +); +PREPARE add_task_center_archive_claim_stale_stmt FROM @add_task_center_archive_claim_stale_sql; +EXECUTE add_task_center_archive_claim_stale_stmt; +DEALLOCATE PREPARE add_task_center_archive_claim_stale_stmt;