Merge pull request #4305 from wp-a/fix/outbox-degraded-rebuild-latch

fix(scheduler): latch degraded outbox rebuilds

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},

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