diff --git a/backend/internal/service/scheduler_snapshot_full_rebuild_lifecycle_test.go b/backend/internal/service/scheduler_snapshot_full_rebuild_lifecycle_test.go index fa918a0862..e52adbb4c8 100644 --- a/backend/internal/service/scheduler_snapshot_full_rebuild_lifecycle_test.go +++ b/backend/internal/service/scheduler_snapshot_full_rebuild_lifecycle_test.go @@ -290,13 +290,13 @@ func TestSchedulerFullRebuildActiveTombstoneDoesNotBlockFollowingGroupEvent(t *t require.Equal(t, 1, activeCalls) require.Zero(t, fallbackCalls) require.Equal(t, []int64{groupID, groupID}, freshCalls) - require.Len(t, cache.tokens(), 24, "full rebuild and the following group event must each run fresh authority") + require.Len(t, cache.tokens(), 36, "full rebuild and the following group event must each run fresh authority") _, reopenHeld := cache.lifecycleMutationLeaseStates() - require.Len(t, reopenHeld, 24) + require.Len(t, reopenHeld, 36) for _, held := range reopenHeld { require.True(t, held) } - require.Equal(t, 21, accounts.callCount()) + require.Equal(t, 30, accounts.callCount()) } func TestSchedulerFullRebuildGlobalReadErrorsFailBeforeMutationOrDB(t *testing.T) { @@ -367,15 +367,15 @@ func TestSchedulerFullRebuildFreshActivePreparesEveryTokenBeforeFirstDB(t *testi capturesAtFirstDB = cache.captureAttemptCount() held, reopenCount := cache.leaseHeldAndTokenCount() require.False(t, held) - require.Equal(t, 12, reopenCount) - require.Equal(t, 13, capturesAtFirstDB, "C(0) and the historical bucket must be captured before DB") + require.Equal(t, 18, reopenCount) + require.Equal(t, 19, capturesAtFirstDB, "C(0) and the historical bucket must be captured before DB") } svc := newFullRebuildLifecycleService(cache, nil, accounts, groups, config.RunModeStandard) require.NoError(t, svc.rebuildFullSnapshot(context.Background(), "test")) require.Equal(t, capturesAtFirstDB, cache.captureAttemptCount()) - require.Equal(t, 15, accounts.callCount()) - require.Equal(t, 8, accounts.groupCallCount(groupID)) + require.Equal(t, 21, accounts.callCount()) + require.Equal(t, 11, accounts.groupCallCount(groupID)) _, historicalPublished := cache.counts(historical) require.Equal(t, 1, historicalPublished) activeCalls, fallbackCalls, freshCalls := groups.stats() @@ -398,7 +398,7 @@ func TestSchedulerFullRebuildOrdinaryCaptureErrorReturnsBeforeFirstDB(t *testing err := svc.rebuildFullSnapshot(context.Background(), "test") require.ErrorIs(t, err, wantErr) - require.Equal(t, 14, cache.captureAttemptCount(), "all canonical and ordinary captures must be attempted before returning") + require.Equal(t, 20, cache.captureAttemptCount(), "all canonical and ordinary captures must be attempted before returning") require.Zero(t, accounts.callCount()) require.Zero(t, cache.totalSetAttempts()) } @@ -416,8 +416,8 @@ func TestSchedulerFullRebuildPreservesGroupZeroActiveHistoricalAndInvalidRegistr svc := newFullRebuildLifecycleService(cache, nil, accounts, groups, config.RunModeStandard) require.NoError(t, svc.rebuildFullSnapshot(context.Background(), "test")) - require.Equal(t, 27, cache.captureAttemptCount()) - require.Equal(t, 17, accounts.callCount()) + require.Equal(t, 39, cache.captureAttemptCount()) + require.Equal(t, 23, accounts.callCount()) groups.mu.Lock() require.Equal(t, 1, groups.listCalls) groups.mu.Unlock() @@ -456,7 +456,7 @@ func TestSchedulerFullRebuildActiveTombstoneFreshInactiveOrMissingFiltersAllGrou require.NoError(t, svc.rebuildFullSnapshot(context.Background(), "test")) require.Zero(t, accounts.groupCallCount(groupID)) - require.Equal(t, 7, accounts.groupCallCount(0)) + require.Equal(t, 10, accounts.groupCallCount(0)) require.Empty(t, cache.tokens()) require.Equal(t, bucketStrings(append(canonical, historical)), bucketStrings(cache.retiredBuckets())) for _, bucket := range append(canonical, historical) { @@ -528,7 +528,7 @@ func TestSchedulerFullRebuildPartialLifecycleFailureReturnsBeforeDBAndRetries(t require.Equal(t, []int64{1, 2}, freshCalls) require.Zero(t, accounts.callCount()) require.Zero(t, cache.totalSetAttempts()) - require.Equal(t, 13, len(cache.retiredBuckets())) + require.Equal(t, 19, len(cache.retiredBuckets())) groups.mu.Lock() delete(groups.freshErr, 2) @@ -537,8 +537,8 @@ func TestSchedulerFullRebuildPartialLifecycleFailureReturnsBeforeDBAndRetries(t require.NoError(t, svc.triggerFullRebuild("retry")) _, _, freshCalls = groups.stats() require.Equal(t, []int64{1, 2, 2, 3}, freshCalls) - require.Equal(t, 39, len(cache.retiredBuckets())) - require.Equal(t, 7, accounts.callCount()) + require.Equal(t, 57, len(cache.retiredBuckets())) + require.Equal(t, 10, accounts.callCount()) require.Empty(t, cache.tokens()) } @@ -561,14 +561,14 @@ func TestSchedulerFullRebuildActiveTombstoneLazyRecoveryDiscardsPartialCaptureTa capturesAtFirstDB = cache.captureAttemptCount() held, reopenCount := cache.leaseHeldAndTokenCount() require.False(t, held) - require.Equal(t, 12, reopenCount) - require.Equal(t, 19, capturesAtFirstDB) + require.Equal(t, 18, reopenCount) + require.Equal(t, 25, capturesAtFirstDB) } svc := newFullRebuildLifecycleService(cache, nil, accounts, groups, config.RunModeStandard) require.NoError(t, svc.rebuildFullSnapshot(context.Background(), "test")) require.Equal(t, capturesAtFirstDB, cache.captureAttemptCount()) - require.Equal(t, 15, accounts.callCount()) + require.Equal(t, 21, accounts.callCount()) for _, bucket := range canonical { attempts, published := cache.counts(bucket) require.Equal(t, 1, attempts, "discarded pre-recovery tokens must never publish: %s", bucket.String()) @@ -600,9 +600,9 @@ func TestSchedulerFullRebuildSimpleModePreservesRegistryWithoutLifecycleAuthorit require.Zero(t, activeCalls) require.Zero(t, fallbackCalls) require.Empty(t, freshCalls) - require.Equal(t, 15, cache.captureAttemptCount()) - require.Equal(t, 10, accounts.callCount()) - require.Equal(t, 10, accounts.groupCallCount(0)) + require.Equal(t, 21, cache.captureAttemptCount()) + require.Equal(t, 13, accounts.callCount()) + require.Equal(t, 13, accounts.groupCallCount(0)) require.Empty(t, cache.retiredBuckets()) require.Empty(t, cache.tokens()) for _, bucket := range registered { @@ -632,11 +632,11 @@ func TestSchedulerFullRebuildFreshReopenLockBusyRetriesWithoutBlockingOrdinaryTa require.Zero(t, cache.currentWatermark()) _, groupZeroPublished := cache.counts(schedulerCanonicalBuckets(0)[0]) require.Equal(t, 1, groupZeroPublished, "ordinary tasks must still run when one strict Reopen task is busy") - require.Equal(t, 14, accounts.callCount()) + require.Equal(t, 20, accounts.callCount()) svc.pollOutbox() require.Equal(t, int64(1), cache.currentWatermark()) - require.Equal(t, 28, accounts.callCount()) + require.Equal(t, 40, accounts.callCount()) _, busyBucketPublished := cache.counts(canonical[0]) require.Equal(t, 1, busyBucketPublished) activeCalls, fallbackCalls, freshCalls := groups.stats() @@ -654,7 +654,7 @@ func TestSchedulerFullRebuildOrdinaryLockBusyKeepsExistingSkipSemantics(t *testi svc := newFullRebuildLifecycleService(cache, nil, accounts, groups, config.RunModeStandard) require.NoError(t, svc.rebuildFullSnapshot(context.Background(), "test")) - require.Equal(t, 7, accounts.callCount()) + require.Equal(t, 10, accounts.callCount()) attempts, published := cache.counts(busyBucket) require.Zero(t, attempts) require.Zero(t, published) diff --git a/backend/internal/service/scheduler_snapshot_group_lifecycle_test.go b/backend/internal/service/scheduler_snapshot_group_lifecycle_test.go index d088f3df7a..f9e75ed158 100644 --- a/backend/internal/service/scheduler_snapshot_group_lifecycle_test.go +++ b/backend/internal/service/scheduler_snapshot_group_lifecycle_test.go @@ -441,7 +441,7 @@ func TestSchedulerGroupLifecycleActiveReopensAndRebuildsAllCurrentBuckets(t *tes accounts.beforeLoad = func() { held, tokenCount := cache.leaseHeldAndTokenCount() require.False(t, held, "the group lifecycle lease must be released before the first account query") - require.Equal(t, 12, tokenCount, "all reopen tokens must be prepared before the first account query") + require.Equal(t, 18, tokenCount, "all reopen tokens must be prepared before the first account query") } svc := newGroupLifecycleTestService(cache, accounts, groups, config.RunModeStandard) seen := make(map[batchSeenKey]struct{}) @@ -453,8 +453,8 @@ func TestSchedulerGroupLifecycleActiveReopensAndRebuildsAllCurrentBuckets(t *tes registered, err := cache.retirementRaceCache.ListBuckets(context.Background()) require.NoError(t, err) require.Contains(t, bucketStrings(registered), historical.String()) - require.Len(t, cache.tokens(), 12) - require.Equal(t, 7, accounts.callCount()) + require.Len(t, cache.tokens(), 18) + require.Equal(t, 10, accounts.callCount()) require.Equal(t, 1, accounts.platformCallCount(PlatformOpenAI)) for _, bucket := range current { _, published := cache.counts(bucket) @@ -472,16 +472,16 @@ func TestSchedulerGroupLifecycleActiveReopensAndRebuildsAllCurrentBuckets(t *tes require.True(t, cache.releaseDeadline) require.NoError(t, cache.releaseCtxErr) _, reopenHeld := cache.lifecycleMutationLeaseStates() - require.Len(t, reopenHeld, 12) + require.Len(t, reopenHeld, 18) for _, held := range reopenHeld { require.True(t, held) } lockTTLs, unlockCalls := cache.lockStats() - require.Len(t, lockTTLs, 12) + require.Len(t, lockTTLs, 18) for _, ttl := range lockTTLs { require.Equal(t, 30*time.Second, ttl) } - require.Equal(t, 12, unlockCalls) + require.Equal(t, 18, unlockCalls) requireLifecycleSeen(t, seen, groupID) } @@ -497,8 +497,8 @@ func TestSchedulerGroupLifecycleInactiveThenActiveAuthoritativelyReopens(t *test groups.set(&Group{ID: groupID, Status: StatusActive, Hydrated: true}, nil) require.NoError(t, svc.handleGroupEvent(context.Background(), ptrInt64(groupID), make(map[batchSeenKey]struct{}))) - require.Len(t, cache.tokens(), 12) - require.Equal(t, 7, accounts.callCount()) + require.Len(t, cache.tokens(), 18) + require.Equal(t, 10, accounts.callCount()) for _, bucket := range expectedGroupLifecycleBuckets(groupID) { _, published := cache.counts(bucket) require.Equal(t, 1, published, bucket.String()) @@ -546,15 +546,15 @@ func TestSchedulerGroupLifecycleEpochPreventsABA(t *testing.T) { groups.set(&Group{ID: groupID, Status: StatusActive, Hydrated: true}, nil) require.NoError(t, svc.handleGroupEvent(context.Background(), ptrInt64(groupID), make(map[batchSeenKey]struct{}))) firstActiveTokens := cache.tokens() - require.Len(t, firstActiveTokens, 12) + require.Len(t, firstActiveTokens, 18) groups.set(&Group{ID: groupID, Status: StatusDisabled, Hydrated: true}, nil) require.NoError(t, svc.handleGroupEvent(context.Background(), ptrInt64(groupID), make(map[batchSeenKey]struct{}))) groups.set(&Group{ID: groupID, Status: StatusActive, Hydrated: true}, nil) require.NoError(t, svc.handleGroupEvent(context.Background(), ptrInt64(groupID), make(map[batchSeenKey]struct{}))) allTokens := cache.tokens() - require.Len(t, allTokens, 24) - require.Greater(t, allTokens[12].Epoch, firstActiveTokens[0].Epoch) + require.Len(t, allTokens, 36) + require.Greater(t, allTokens[18].Epoch, firstActiveTokens[0].Epoch) require.ErrorIs(t, cache.SetSnapshot(context.Background(), firstActiveTokens[0].Bucket, firstActiveTokens[0], nil), ErrSchedulerBucketWriteFenced) } @@ -571,11 +571,11 @@ func TestSchedulerGroupLifecycleSeenIsIndependentAndDeduplicatesGroupEvents(t *t require.NoError(t, svc.handleGroupEvent(context.Background(), ptrInt64(groupID), seen)) require.Equal(t, 1, groups.callCount()) - require.Equal(t, 7, accounts.callCount()) + require.Equal(t, 10, accounts.callCount()) requireLifecycleSeen(t, seen, groupID) require.NoError(t, svc.handleGroupEvent(context.Background(), ptrInt64(groupID), seen)) require.Equal(t, 1, groups.callCount()) - require.Equal(t, 7, accounts.callCount()) + require.Equal(t, 10, accounts.callCount()) } func TestSchedulerGroupLifecycleFailuresDoNotMarkSeen(t *testing.T) { diff --git a/backend/internal/service/scheduler_snapshot_retirement_test.go b/backend/internal/service/scheduler_snapshot_retirement_test.go index 3d24179758..6fa2b7ddc5 100644 --- a/backend/internal/service/scheduler_snapshot_retirement_test.go +++ b/backend/internal/service/scheduler_snapshot_retirement_test.go @@ -176,7 +176,7 @@ func TestSchedulerFullRebuildCapturesAllRegistryTokensBeforeDBLoad(t *testing.T) } captures, reopens := cache.captureAndReopenCounts() - require.Equal(t, 24, captures, "group0 and active-group canonical tokens must be captured before the first DB load") + require.Equal(t, 36, captures, "group0 and active-group canonical tokens must be captured before the first DB load") require.Zero(t, reopens) require.NoError(t, cache.RetireBucket(context.Background(), queued)) _, err := cache.ReopenBucket(context.Background(), queued)