mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-21 05:08:32 +00:00
test(scheduler): update CN platform expectations
This commit is contained in:
wucm667
@@ -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)
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user