diff --git a/internal/service/taskcenter/archive.go b/internal/service/taskcenter/archive.go index 89187bb..05e04cb 100644 --- a/internal/service/taskcenter/archive.go +++ b/internal/service/taskcenter/archive.go @@ -113,16 +113,32 @@ func (s *Service) claimArchiveRows(ctx context.Context, batchSize int) ([]model. 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 + pendingLimit := taskCenterArchivePendingClaimLimit(remaining) + for _, retryable := range []struct { + status string + limit int + }{ + {status: ArchiveStatusPending, limit: pendingLimit}, + {status: ArchiveStatusFailed, limit: remaining}, + } { + if retryable.limit <= 0 || remaining <= 0 { + continue + } + if retryable.limit > remaining { + retryable.limit = remaining + } + var retryableRows []model.TaskCenterEventArchiveOutbox + // PENDING 和 FAILED 分开抢占,避免 IN 跨 status 排序触发 filesort,同时给 FAILED 保留重试额度。 + if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}). + Where("status = ?", retryable.status). + Order("create_time asc, id asc"). + Limit(retryable.limit). + Find(&retryableRows).Error; err != nil { + return err + } + rows = append(rows, retryableRows...) + remaining -= len(retryableRows) } - rows = append(rows, retryableRows...) } if len(rows) == 0 { return nil @@ -164,6 +180,20 @@ func taskCenterArchiveClaimLimits(batchSize int) (int, int) { return staleLimit, batchSize - staleLimit } +func taskCenterArchivePendingClaimLimit(retryableLimit int) int { + if retryableLimit <= 1 { + return retryableLimit + } + failedReserve := retryableLimit / 10 + if failedReserve < 1 { + failedReserve = 1 + } + if failedReserve > 100 { + failedReserve = 100 + } + return retryableLimit - failedReserve +} + 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 85a6ac2..b3908fa 100644 --- a/internal/service/taskcenter/taskcenter_test.go +++ b/internal/service/taskcenter/taskcenter_test.go @@ -87,6 +87,28 @@ func TestTaskCenterArchiveClaimLimitsReserveStaleRows(t *testing.T) { } } +func TestTaskCenterArchivePendingClaimLimitReservesFailedRows(t *testing.T) { + tests := []struct { + name string + retryableLimit int + want int + }{ + {name: "empty", retryableLimit: 0, want: 0}, + {name: "single row stays pending first", retryableLimit: 1, want: 1}, + {name: "small retry batch keeps one failed slot", retryableLimit: 4, want: 3}, + {name: "normal retry batch reserves ten percent", retryableLimit: 450, want: 405}, + {name: "large retry batch caps failed reserve", retryableLimit: 900, want: 810}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := taskCenterArchivePendingClaimLimit(tt.retryableLimit); got != tt.want { + t.Fatalf("taskCenterArchivePendingClaimLimit(%d) = %d, want %d", tt.retryableLimit, got, tt.want) + } + }) + } +} + func TestDailyTaskConfigEventListAndClaim(t *testing.T) { java := &taskCenterTestGateway{events: map[string]bool{}} service, db := newTestService(t, java)