diff --git a/backend/internal/service/scheduler_snapshot_outbox_cleanup_test.go b/backend/internal/service/scheduler_snapshot_outbox_cleanup_test.go index 9a7d360e2e..2a476ff077 100644 --- a/backend/internal/service/scheduler_snapshot_outbox_cleanup_test.go +++ b/backend/internal/service/scheduler_snapshot_outbox_cleanup_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "reflect" + "sync" "testing" "time" @@ -14,6 +15,8 @@ type outboxCleanupCache struct { watermark int64 setWatermarks []int64 updateErr error + listBucketErr error + listBuckets []SchedulerBucket listBucketCalls int } @@ -71,7 +74,7 @@ func (c *outboxCleanupCache) UnlockBucket(ctx context.Context, bucket SchedulerB func (c *outboxCleanupCache) ListBuckets(ctx context.Context) ([]SchedulerBucket, error) { c.listBucketCalls++ - return nil, nil + return c.listBuckets, c.listBucketErr } func (c *outboxCleanupCache) GetOutboxWatermark(ctx context.Context) (int64, error) { @@ -92,6 +95,8 @@ type outboxCleanupDeleteCall struct { type outboxCleanupRepo struct { events []SchedulerOutboxEvent rows []int64 + maxIDCalls int + maxIDErr error lockAcquired bool lockAttempts int releaseCount int @@ -99,6 +104,40 @@ type outboxCleanupRepo struct { firstCreatedAfterID []int64 } +type outboxCleanupAccountRepo struct { + AccountRepository +} + +func (r *outboxCleanupAccountRepo) ListSchedulableUngroupedByPlatform(context.Context, string) ([]Account, error) { + return nil, nil +} + +type blockingOutboxCleanupCache struct { + *outboxCleanupCache + mu sync.Mutex + calls int + started chan struct{} + release chan struct{} +} + +func (c *blockingOutboxCleanupCache) ListBuckets(context.Context) ([]SchedulerBucket, error) { + c.mu.Lock() + c.calls++ + call := c.calls + c.mu.Unlock() + if call == 1 { + close(c.started) + <-c.release + } + return c.listBuckets, c.listBucketErr +} + +func (c *blockingOutboxCleanupCache) listCalls() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.calls +} + func (r *outboxCleanupRepo) ListAfterAndReleaseDedup(ctx context.Context, afterID int64, limit int) ([]SchedulerOutboxEvent, error) { events := make([]SchedulerOutboxEvent, 0, len(r.events)) for _, event := range r.events { @@ -124,6 +163,10 @@ func (r *outboxCleanupRepo) FirstCreatedAtAfter(ctx context.Context, afterID int } func (r *outboxCleanupRepo) MaxID(ctx context.Context) (int64, error) { + r.maxIDCalls++ + if r.maxIDErr != nil { + return 0, r.maxIDErr + } var maxID int64 for _, id := range r.rows { if id > maxID { @@ -313,6 +356,515 @@ func TestSchedulerSnapshotServicePollOutboxDoesNotUseConsumedEventForLag(t *test } } +func TestSchedulerSnapshotServiceCheckOutboxLagLatchesPersistentDegradation(t *testing.T) { + tests := []struct { + name string + createdAt time.Time + rows []int64 + lagSeconds int + backlogThreshold int + }{ + { + name: "lag", + createdAt: time.Now().Add(-time.Hour), + rows: []int64{1}, + lagSeconds: 1, + }, + { + name: "backlog", + createdAt: time.Now(), + rows: []int64{100}, + backlogThreshold: 50, + }, + { + name: "lag_and_backlog", + createdAt: time.Now().Add(-time.Hour), + rows: []int64{100}, + lagSeconds: 1, + backlogThreshold: 50, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cache := &outboxCleanupCache{listBuckets: []SchedulerBucket{{Platform: PlatformOpenAI, Mode: SchedulerModeSingle}}} + repo := &outboxCleanupRepo{ + events: []SchedulerOutboxEvent{{ID: 1, CreatedAt: tt.createdAt}}, + rows: tt.rows, + } + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxLagRebuildSeconds: tt.lagSeconds, + OutboxLagRebuildFailures: 1, + OutboxBacklogRebuildRows: tt.backlogThreshold, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, &outboxCleanupAccountRepo{}, nil, cfg) + + for range 3 { + svc.checkOutboxLag(context.Background(), 0) + } + + if cache.listBucketCalls != 1 { + t.Fatalf("expected one rebuild attempt during a persistent degraded episode, got %d", cache.listBucketCalls) + } + }) + } +} + +func TestSchedulerSnapshotServiceCheckOutboxLagFailedRebuildRearmsAfterRecovery(t *testing.T) { + cache := &outboxCleanupCache{listBucketErr: errors.New("list buckets failed")} + repo := &outboxCleanupRepo{ + events: []SchedulerOutboxEvent{{ID: 1, CreatedAt: time.Now().Add(-time.Hour)}}, + rows: []int64{1}, + } + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxLagRebuildSeconds: 1, + OutboxLagRebuildFailures: 1, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, nil, nil, cfg) + + svc.checkOutboxLag(context.Background(), 0) + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 1 { + t.Fatalf("expected a failed rebuild to stay bounded within the episode, got %d attempts", cache.listBucketCalls) + } + + svc.checkOutboxLag(context.Background(), 1) + repo.events = append(repo.events, SchedulerOutboxEvent{ID: 2, CreatedAt: time.Now().Add(-time.Hour)}) + repo.rows = []int64{2} + svc.checkOutboxLag(context.Background(), 1) + + if cache.listBucketCalls != 2 { + t.Fatalf("expected recovery to rearm a failed rebuild for the next episode, got %d attempts", cache.listBucketCalls) + } +} + +func TestSchedulerSnapshotServiceCheckOutboxLagFailedRebuildRetriesAfterCooldownWithoutRecovery(t *testing.T) { + cache := &outboxCleanupCache{ + listBucketErr: errors.New("list buckets failed"), + listBuckets: []SchedulerBucket{{Platform: PlatformOpenAI, Mode: SchedulerModeSingle}}, + } + repo := &outboxCleanupRepo{ + events: []SchedulerOutboxEvent{{ID: 1, CreatedAt: time.Now().Add(-time.Hour)}}, + rows: []int64{1}, + } + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxLagRebuildSeconds: 1, + OutboxLagRebuildFailures: 1, + FullRebuildIntervalSeconds: 0, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, &outboxCleanupAccountRepo{}, nil, cfg) + + svc.checkOutboxLag(context.Background(), 0) + for range 3 { + svc.checkOutboxLag(context.Background(), 0) + } + if cache.listBucketCalls != 1 { + t.Fatalf("expected failed rebuild polls to be rate limited, got %d attempts", cache.listBucketCalls) + } + + svc.lagMu.Lock() + if !svc.outboxRebuildRetryAt.After(time.Now()) { + t.Fatal("expected failed rebuild to schedule a future retry") + } + svc.outboxRebuildRetryAt = time.Now().Add(-time.Second) + svc.lagMu.Unlock() + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 2 { + t.Fatalf("expected persistent degradation to retry after cooldown, got %d attempts", cache.listBucketCalls) + } + + for range 3 { + svc.checkOutboxLag(context.Background(), 0) + } + if cache.listBucketCalls != 2 { + t.Fatalf("expected repeated rebuild failures to stay rate limited, got %d attempts", cache.listBucketCalls) + } + + svc.lagMu.Lock() + svc.outboxRebuildRetryAt = time.Now().Add(-time.Second) + svc.lagMu.Unlock() + cache.listBucketErr = nil + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 3 { + t.Fatalf("expected degraded episode to retry after cooldown, got %d attempts", cache.listBucketCalls) + } + + for range 3 { + svc.checkOutboxLag(context.Background(), 0) + } + if cache.listBucketCalls != 3 { + t.Fatalf("expected successful retry to latch the degraded episode, got %d attempts", cache.listBucketCalls) + } +} + +func TestSchedulerSnapshotServiceCheckOutboxLagBacklogRetryDoesNotBypassNewLagThreshold(t *testing.T) { + cache := &outboxCleanupCache{listBucketErr: errors.New("list buckets failed")} + repo := &outboxCleanupRepo{ + events: []SchedulerOutboxEvent{{ID: 1, CreatedAt: time.Now()}}, + rows: []int64{100}, + } + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxLagRebuildSeconds: 1, + OutboxLagRebuildFailures: 3, + OutboxBacklogRebuildRows: 50, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, nil, nil, cfg) + + // Start with backlog-only degradation and leave its failed rebuild retry due. + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 1 { + t.Fatalf("expected the backlog degradation to attempt one rebuild, got %d", cache.listBucketCalls) + } + svc.lagMu.Lock() + svc.outboxRebuildRetryAt = time.Now().Add(-time.Second) + svc.lagMu.Unlock() + + // The backlog recovers while lag becomes newly degraded. The stale backlog + // retry must not make the first lag observation bypass its failure threshold. + repo.rows = []int64{1} + repo.events[0].CreatedAt = time.Now().Add(-time.Hour) + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 1 { + t.Fatalf("expected the new lag episode to start at its own threshold, got %d rebuild attempts", cache.listBucketCalls) + } + + svc.checkOutboxLag(context.Background(), 0) + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 2 { + t.Fatalf("expected lag rebuild only after three lag observations, got %d attempts", cache.listBucketCalls) + } +} + +func TestSchedulerSnapshotServiceCheckOutboxLagLagRetryDoesNotDelayOrEscalateNewBacklog(t *testing.T) { + cache := &outboxCleanupCache{listBucketErr: errors.New("list buckets failed")} + repo := &outboxCleanupRepo{ + events: []SchedulerOutboxEvent{{ID: 1, CreatedAt: time.Now().Add(-time.Hour)}}, + rows: []int64{1}, + } + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxLagRebuildSeconds: 1, + OutboxLagRebuildFailures: 1, + OutboxBacklogRebuildRows: 50, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, nil, nil, cfg) + + // Start with lag-only degradation and a failed rebuild in cooldown. + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 1 { + t.Fatalf("expected the lag degradation to attempt one rebuild, got %d", cache.listBucketCalls) + } + + // Lag recovers while backlog becomes newly degraded. It must start immediately + // and its first failure must use the base retry generation, not lag's count. + repo.events[0].CreatedAt = time.Now() + repo.rows = []int64{100} + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 2 { + t.Fatalf("expected the new backlog degradation not to inherit lag cooldown, got %d rebuild attempts", cache.listBucketCalls) + } + svc.lagMu.Lock() + failures := svc.outboxRebuildFailures + svc.lagMu.Unlock() + if failures != 1 { + t.Fatalf("expected backlog retry failures to restart at one, got %d", failures) + } +} + +func TestSchedulerSnapshotServiceCheckOutboxLagBacklogRetrySurvivesUnknownBacklog(t *testing.T) { + cache := &outboxCleanupCache{listBucketErr: errors.New("list buckets failed")} + repo := &outboxCleanupRepo{ + events: []SchedulerOutboxEvent{{ID: 1, CreatedAt: time.Now()}}, + rows: []int64{100}, + } + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxBacklogRebuildRows: 50, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, nil, nil, cfg) + + // A failed backlog rebuild starts a reason-scoped cooldown. + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 1 { + t.Fatalf("expected one initial backlog rebuild, got %d", cache.listBucketCalls) + } + svc.lagMu.Lock() + retryAt := svc.outboxRebuildRetryAt + svc.lagMu.Unlock() + if !retryAt.After(time.Now()) { + t.Fatalf("expected a future backlog retry, got %s", retryAt) + } + + // A temporary MaxID failure makes backlog health unknown, not recovered. + repo.maxIDErr = errors.New("max id unavailable") + svc.checkOutboxLag(context.Background(), 0) + svc.lagMu.Lock() + retryReason := svc.outboxRebuildRetryReason + failures := svc.outboxRebuildFailures + retryAtAfterUnknown := svc.outboxRebuildRetryAt + svc.lagMu.Unlock() + if retryReason != "outbox_backlog" || failures != 1 || !retryAtAfterUnknown.Equal(retryAt) { + t.Fatalf("expected unknown backlog to preserve retry state, got reason=%q failures=%d retry_at=%s", retryReason, failures, retryAtAfterUnknown) + } + + // When MaxID recovers and backlog remains degraded, the original cooldown + // still applies; only an expired cooldown may trigger the retry. + repo.maxIDErr = nil + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 1 { + t.Fatalf("expected backlog recovery before cooldown to stay rate limited, got %d attempts", cache.listBucketCalls) + } + svc.lagMu.Lock() + svc.outboxRebuildRetryAt = time.Now().Add(-time.Second) + svc.lagMu.Unlock() + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 2 { + t.Fatalf("expected backlog retry after cooldown expiry, got %d attempts", cache.listBucketCalls) + } +} + +func TestSchedulerSnapshotServiceCheckOutboxLagPreemptsUnknownBacklogRetryAtThreshold(t *testing.T) { + cache := &outboxCleanupCache{listBucketErr: errors.New("list buckets failed")} + repo := &outboxCleanupRepo{ + events: []SchedulerOutboxEvent{{ID: 1, CreatedAt: time.Now()}}, + rows: []int64{100}, + } + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxLagRebuildSeconds: 1, + OutboxLagRebuildFailures: 3, + OutboxBacklogRebuildRows: 50, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, nil, nil, cfg) + + // Backlog starts the first failed rebuild generation and remains unknown. + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 1 { + t.Fatalf("expected one initial backlog rebuild, got %d", cache.listBucketCalls) + } + repo.maxIDErr = errors.New("max id unavailable") + repo.events[0].CreatedAt = time.Now().Add(-time.Hour) + + // A known lag degradation must keep accumulating independently of the active + // backlog cooldown and preempt it only after reaching its own threshold. + for observation := 1; observation <= 2; observation++ { + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 1 { + t.Fatalf("expected lag observation %d to stay below threshold, got %d rebuild attempts", observation, cache.listBucketCalls) + } + } + svc.checkOutboxLag(context.Background(), 0) + if cache.listBucketCalls != 2 { + t.Fatalf("expected lag to preempt backlog cooldown at its threshold, got %d attempts", cache.listBucketCalls) + } + + svc.lagMu.Lock() + retryReason := svc.outboxRebuildRetryReason + failures := svc.outboxRebuildFailures + retryAt := svc.outboxRebuildRetryAt + svc.lagMu.Unlock() + if retryReason != "outbox_lag" || failures != 1 || !retryAt.After(time.Now()) { + t.Fatalf("expected a fresh lag retry generation, got reason=%q failures=%d retry_at=%s", retryReason, failures, retryAt) + } +} + +func TestOutboxRebuildRetryDelayIsExponentiallyBounded(t *testing.T) { + previous := time.Duration(0) + for failures := 1; failures <= 20; failures++ { + delay := outboxRebuildRetryDelay(failures) + if delay < previous { + t.Fatalf("expected retry delay to be monotonic, failure %d produced %s after %s", failures, delay, previous) + } + if delay > outboxRebuildRetryMaxDelay { + t.Fatalf("expected retry delay to stay bounded, got %s", delay) + } + previous = delay + } + if previous != outboxRebuildRetryMaxDelay { + t.Fatalf("expected repeated failures to reach max delay %s, got %s", outboxRebuildRetryMaxDelay, previous) + } +} + +func TestSchedulerSnapshotServicePollOutboxEmptyBatchClearsDegradedEpisode(t *testing.T) { + cache := &outboxCleanupCache{listBuckets: []SchedulerBucket{{Platform: PlatformOpenAI, Mode: SchedulerModeSingle}}} + repo := &outboxCleanupRepo{ + events: []SchedulerOutboxEvent{{ID: 1, CreatedAt: time.Now().Add(-time.Hour)}}, + rows: []int64{1}, + } + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxLagRebuildSeconds: 1, + OutboxLagRebuildFailures: 1, + OutboxBacklogRebuildRows: 1, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, &outboxCleanupAccountRepo{}, nil, cfg) + + svc.checkOutboxLag(context.Background(), 0) + cache.watermark = 1 + svc.pollOutbox() + + if !reflect.DeepEqual(repo.firstCreatedAfterID, []int64{0}) { + t.Fatalf("expected empty poll to use the empty batch as recovery evidence, got watermarks %#v", repo.firstCreatedAfterID) + } + if repo.maxIDCalls != 1 { + t.Fatalf("expected empty poll to skip a redundant backlog query, got %d health checks", repo.maxIDCalls) + } + + repo.events = append(repo.events, SchedulerOutboxEvent{ID: 2, CreatedAt: time.Now().Add(-time.Hour)}) + repo.rows = []int64{2} + svc.checkOutboxLag(context.Background(), 1) + if cache.listBucketCalls != 2 { + t.Fatalf("expected empty-poll recovery to rearm the next degraded episode, got %d attempts", cache.listBucketCalls) + } +} + +func TestSchedulerSnapshotServiceOutboxLagWarningIsTransitionLimited(t *testing.T) { + svc := NewSchedulerSnapshotService(nil, nil, nil, nil, nil) + + if !svc.shouldLogOutboxLagWarning(true) { + t.Fatal("expected the initial degraded transition to log") + } + if svc.shouldLogOutboxLagWarning(true) { + t.Fatal("expected persistent degradation to suppress repeated warnings") + } + if svc.shouldLogOutboxLagWarning(true) { + t.Fatal("expected persistent degradation to suppress repeated warnings") + } + if svc.shouldLogOutboxLagWarning(false) { + t.Fatal("expected recovery not to emit a lag warning") + } + if !svc.shouldLogOutboxLagWarning(true) { + t.Fatal("expected renewed degradation to log after recovery") + } +} + +func TestSchedulerSnapshotServiceCheckOutboxLagSamplesMaxIDErrors(t *testing.T) { + svc := NewSchedulerSnapshotService(nil, nil, nil, nil, nil) + now := time.Now() + + if !svc.shouldLogOutboxMaxIDError(now) { + t.Fatal("expected the first MaxID error to log") + } + if svc.shouldLogOutboxMaxIDError(now.Add(outboxMaxIDErrorLogSampleInterval / 2)) { + t.Fatal("expected MaxID errors inside the sample interval to be suppressed") + } + if !svc.shouldLogOutboxMaxIDError(now.Add(outboxMaxIDErrorLogSampleInterval)) { + t.Fatal("expected MaxID error logging to rearm after the sample interval") + } +} + +func TestSchedulerSnapshotServicePollOutboxHealthyEmptyBatchSkipsLagHealthQueries(t *testing.T) { + cache := &outboxCleanupCache{} + repo := &outboxCleanupRepo{} + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxLagRebuildSeconds: 1, + OutboxLagRebuildFailures: 1, + OutboxBacklogRebuildRows: 1, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, nil, nil, cfg) + + svc.pollOutbox() + + if len(repo.firstCreatedAfterID) != 0 { + t.Fatalf("expected healthy empty poll to skip lag query, got watermarks %#v", repo.firstCreatedAfterID) + } + if repo.maxIDCalls != 0 { + t.Fatalf("expected healthy empty poll to skip backlog query, got %d calls", repo.maxIDCalls) + } +} + +func TestSchedulerSnapshotServiceEmptyPollDoesNotReleaseRunningRebuild(t *testing.T) { + baseCache := &outboxCleanupCache{ + watermark: 1, + listBuckets: []SchedulerBucket{{Platform: PlatformOpenAI, Mode: SchedulerModeSingle}}, + } + cache := &blockingOutboxCleanupCache{ + outboxCleanupCache: baseCache, + started: make(chan struct{}), + release: make(chan struct{}), + } + repo := &outboxCleanupRepo{ + events: []SchedulerOutboxEvent{{ID: 1, CreatedAt: time.Now().Add(-time.Hour)}}, + rows: []int64{1}, + } + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxLagRebuildSeconds: 1, + OutboxLagRebuildFailures: 1, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, &outboxCleanupAccountRepo{}, nil, cfg) + + firstDone := make(chan struct{}) + go func() { + svc.checkOutboxLag(context.Background(), 0) + close(firstDone) + }() + select { + case <-cache.started: + case <-time.After(time.Second): + t.Fatal("first rebuild did not start") + } + + // The empty batch proves recovery for episode/retry state, but it must not + // release ownership of the still-running rebuild. + svc.pollOutbox() + + secondDone := make(chan struct{}) + go func() { + svc.checkOutboxLag(context.Background(), 0) + close(secondDone) + }() + select { + case <-secondDone: + case <-time.After(200 * time.Millisecond): + close(cache.release) + <-firstDone + <-secondDone + t.Fatal("second lag check queued another rebuild while the first was running") + } + + close(cache.release) + <-firstDone + if calls := cache.listCalls(); calls != 1 { + t.Fatalf("expected one rebuild generation, got %d", calls) + } +} + func TestSchedulerSnapshotServiceCleanupSkipsNonPositiveWatermark(t *testing.T) { repo := &outboxCleanupRepo{ rows: []int64{1, 2, 3}, diff --git a/backend/internal/service/scheduler_snapshot_service.go b/backend/internal/service/scheduler_snapshot_service.go index cc8befacde..a5bccdb265 100644 --- a/backend/internal/service/scheduler_snapshot_service.go +++ b/backend/internal/service/scheduler_snapshot_service.go @@ -28,6 +28,9 @@ const ( schedulerGroupLifecycleTimeout = 30 * time.Second schedulerGroupLifecycleLeaseTTL = 60 * time.Second schedulerGroupLifecycleReleaseTimeout = 2 * time.Second + outboxRebuildRetryBaseDelay = 5 * time.Second + outboxRebuildRetryMaxDelay = 5 * time.Minute + outboxMaxIDErrorLogSampleInterval = time.Minute ) // batchSeenKey tracks completed per-platform rebuilds and group lifecycle work @@ -105,17 +108,24 @@ type schedulerActiveGroupIDLister interface { } type SchedulerSnapshotService struct { - cache SchedulerCache - outboxRepo SchedulerOutboxRepository - accountRepo AccountRepository - groupRepo GroupRepository - cfg *config.Config - stopCh chan struct{} - stopOnce sync.Once - wg sync.WaitGroup - fallbackLimit *fallbackLimiter - lagMu sync.Mutex - lagFailures int + cache SchedulerCache + outboxRepo SchedulerOutboxRepository + accountRepo AccountRepository + groupRepo GroupRepository + cfg *config.Config + stopCh chan struct{} + stopOnce sync.Once + wg sync.WaitGroup + fallbackLimit *fallbackLimiter + lagMu sync.Mutex + lagFailures int + outboxRebuildLatched bool + outboxRebuildRunning bool + outboxRebuildFailures int + outboxRebuildRetryAt time.Time + outboxRebuildRetryReason string + outboxLagWarningActive bool + outboxMaxIDErrorLastLoggedAt time.Time fullRebuildRunMu sync.Mutex fullRebuildStateMu sync.Mutex @@ -340,6 +350,10 @@ func (s *SchedulerSnapshotService) pollOutbox() { return } if len(events) == 0 { + // The outbox query itself proves there is no event after the watermark. + // Clear degraded/retry state without adding two more repository queries to + // the healthy one-second poll path. + s.clearOutboxDegradedEpisode() return } @@ -1126,58 +1140,184 @@ func (s *SchedulerSnapshotService) checkOutboxLag(ctx context.Context, watermark if s.cfg == nil || s.outboxRepo == nil { return } + now := time.Now() oldestCreatedAt, ok, err := s.outboxRepo.FirstCreatedAtAfter(ctx, watermark) if err != nil { logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox pending event read failed: %v", err) return } - if !ok || oldestCreatedAt.IsZero() { - s.lagMu.Lock() + var lag time.Duration + if ok && !oldestCreatedAt.IsZero() { + lag = now.Sub(oldestCreatedAt) + } + lagSeconds := int(lag.Seconds()) + lagWarning := ok && !oldestCreatedAt.IsZero() && + s.cfg.Gateway.Scheduling.OutboxLagWarnSeconds > 0 && + lagSeconds >= s.cfg.Gateway.Scheduling.OutboxLagWarnSeconds + + lagDegraded := ok && !oldestCreatedAt.IsZero() && + s.cfg.Gateway.Scheduling.OutboxLagRebuildSeconds > 0 && + lagSeconds >= s.cfg.Gateway.Scheduling.OutboxLagRebuildSeconds + + backlogThreshold := s.cfg.Gateway.Scheduling.OutboxBacklogRebuildRows + backlogKnown := true + var backlog int64 + if backlogThreshold > 0 { + maxID, maxErr := s.outboxRepo.MaxID(ctx) + if maxErr != nil { + backlogKnown = false + if s.shouldLogOutboxMaxIDError(now) { + logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox max id read failed: %v", maxErr) + } + } else { + backlog = maxID - watermark + } + } + backlogDegraded := backlogKnown && backlogThreshold > 0 && backlog >= int64(backlogThreshold) + + // A successful rebuild latches the degraded episode until recovery. A failed + // rebuild remains retryable, but only after an exponentially backed-off + // cooldown so a one-second poll cannot create a rebuild storm. + logLagWarning := s.shouldLogOutboxLagWarning(lagWarning) + s.lagMu.Lock() + fullyRecovered := !lagDegraded && backlogKnown && !backlogDegraded + if fullyRecovered { s.lagFailures = 0 - s.lagMu.Unlock() - return + s.outboxRebuildLatched = false + s.outboxRebuildFailures = 0 + s.outboxRebuildRetryAt = time.Time{} + s.outboxRebuildRetryReason = "" } - lag := time.Since(oldestCreatedAt) - if lagSeconds := int(lag.Seconds()); lagSeconds >= s.cfg.Gateway.Scheduling.OutboxLagWarnSeconds && s.cfg.Gateway.Scheduling.OutboxLagWarnSeconds > 0 { + if s.outboxRebuildRetryReason != "" { + retryReasonActive := (s.outboxRebuildRetryReason == "outbox_lag" && lagDegraded) || + (s.outboxRebuildRetryReason == "outbox_backlog" && (!backlogKnown || backlogDegraded)) + if !retryReasonActive { + s.outboxRebuildFailures = 0 + s.outboxRebuildRetryAt = time.Time{} + s.outboxRebuildRetryReason = "" + } + } + + lagRetryPending := s.outboxRebuildRetryReason == "outbox_lag" && !s.outboxRebuildRetryAt.IsZero() + if lagDegraded { + if !s.outboxRebuildLatched && !s.outboxRebuildRunning && !lagRetryPending { + s.lagFailures++ + } + } else { + s.lagFailures = 0 + } + failures := s.lagFailures + lagReady := lagDegraded && failures >= s.cfg.Gateway.Scheduling.OutboxLagRebuildFailures + retryDue := s.outboxRebuildRetryReason != "" && + !s.outboxRebuildRetryAt.IsZero() && !now.Before(s.outboxRebuildRetryAt) + + reason := "" + lagCanPreemptRetry := lagReady && s.outboxRebuildRetryReason != "outbox_lag" + if !s.outboxRebuildLatched && !s.outboxRebuildRunning && + (s.outboxRebuildRetryAt.IsZero() || retryDue || lagCanPreemptRetry) { + switch { + case lagReady || (retryDue && s.outboxRebuildRetryReason == "outbox_lag" && lagDegraded): + if s.outboxRebuildRetryReason != "" && s.outboxRebuildRetryReason != "outbox_lag" { + s.outboxRebuildFailures = 0 + s.outboxRebuildRetryAt = time.Time{} + s.outboxRebuildRetryReason = "" + } + reason = "outbox_lag" + s.lagFailures = 0 + case backlogDegraded && (s.outboxRebuildRetryReason == "" || + (retryDue && s.outboxRebuildRetryReason == "outbox_backlog")): + reason = "outbox_backlog" + } + if reason != "" { + s.outboxRebuildRunning = true + } + } + s.lagMu.Unlock() + + if logLagWarning { logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox lag warning: %ds", lagSeconds) } - if s.cfg.Gateway.Scheduling.OutboxLagRebuildSeconds > 0 && int(lag.Seconds()) >= s.cfg.Gateway.Scheduling.OutboxLagRebuildSeconds { - s.lagMu.Lock() - s.lagFailures++ - failures := s.lagFailures - s.lagMu.Unlock() + if reason == "" { + return + } - if failures >= s.cfg.Gateway.Scheduling.OutboxLagRebuildFailures { - logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox lag rebuild triggered: lag=%s failures=%d", lag, failures) - s.lagMu.Lock() - s.lagFailures = 0 - s.lagMu.Unlock() - if err := s.triggerFullRebuild("outbox_lag"); err != nil { - logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox lag rebuild failed: %v", err) - } - } + var rebuildErr error + switch reason { + case "outbox_lag": + logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox lag rebuild triggered: lag=%s failures=%d", lag, failures) + rebuildErr = s.triggerFullRebuild(reason) + case "outbox_backlog": + logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox backlog rebuild triggered: backlog=%d", backlog) + rebuildErr = s.triggerFullRebuild(reason) + } + + s.lagMu.Lock() + s.outboxRebuildRunning = false + if rebuildErr == nil { + s.outboxRebuildLatched = true + s.outboxRebuildFailures = 0 + s.outboxRebuildRetryAt = time.Time{} + s.outboxRebuildRetryReason = "" } else { - s.lagMu.Lock() - s.lagFailures = 0 - s.lagMu.Unlock() + s.outboxRebuildLatched = false + s.outboxRebuildFailures++ + s.outboxRebuildRetryAt = time.Now().Add(outboxRebuildRetryDelay(s.outboxRebuildFailures)) + s.outboxRebuildRetryReason = reason } + s.lagMu.Unlock() - threshold := s.cfg.Gateway.Scheduling.OutboxBacklogRebuildRows - if threshold <= 0 { - return + if rebuildErr != nil { + logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] %s rebuild failed: %v", reason, rebuildErr) } - maxID, err := s.outboxRepo.MaxID(ctx) - if err != nil { - return - } - if maxID-watermark >= int64(threshold) { - logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox backlog rebuild triggered: backlog=%d", maxID-watermark) - if err := s.triggerFullRebuild("outbox_backlog"); err != nil { - logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox backlog rebuild failed: %v", err) +} + +func outboxRebuildRetryDelay(failures int) time.Duration { + delay := outboxRebuildRetryBaseDelay + for i := 1; i < failures && delay < outboxRebuildRetryMaxDelay; i++ { + delay *= 2 + if delay >= outboxRebuildRetryMaxDelay { + return outboxRebuildRetryMaxDelay } } + return delay +} + +func (s *SchedulerSnapshotService) clearOutboxDegradedEpisode() { + if s == nil { + return + } + s.lagMu.Lock() + if s.lagFailures != 0 || s.outboxRebuildLatched || s.outboxRebuildRunning || + s.outboxRebuildFailures != 0 || !s.outboxRebuildRetryAt.IsZero() || + s.outboxRebuildRetryReason != "" || s.outboxLagWarningActive { + s.lagFailures = 0 + s.outboxRebuildLatched = false + s.outboxRebuildFailures = 0 + s.outboxRebuildRetryAt = time.Time{} + s.outboxRebuildRetryReason = "" + s.outboxLagWarningActive = false + } + s.lagMu.Unlock() +} + +func (s *SchedulerSnapshotService) shouldLogOutboxMaxIDError(now time.Time) bool { + s.lagMu.Lock() + defer s.lagMu.Unlock() + if !s.outboxMaxIDErrorLastLoggedAt.IsZero() && now.Sub(s.outboxMaxIDErrorLastLoggedAt) < outboxMaxIDErrorLogSampleInterval { + return false + } + s.outboxMaxIDErrorLastLoggedAt = now + return true +} + +func (s *SchedulerSnapshotService) shouldLogOutboxLagWarning(active bool) bool { + s.lagMu.Lock() + defer s.lagMu.Unlock() + shouldLog := active && !s.outboxLagWarningActive + s.outboxLagWarningActive = active + return shouldLog } func (s *SchedulerSnapshotService) loadAccountsFromDB(ctx context.Context, bucket SchedulerBucket, useMixed bool) ([]Account, error) {