From 1bc1fb0f28d08ccbfc2c09e84ed6d030aa403216 Mon Sep 17 00:00:00 2001 From: lyy0709 <65712338+lyy0709@users.noreply.github.com> Date: Sat, 15 Aug 2026 03:40:49 +0800 Subject: [PATCH] fix(scheduler): preserve pre-hydration account metadata --- .../internal/repository/scheduler_cache.go | 69 +++++++--- .../repository/scheduler_cache_unit_test.go | 127 ++++++++++++++++++ 2 files changed, 181 insertions(+), 15 deletions(-) diff --git a/backend/internal/repository/scheduler_cache.go b/backend/internal/repository/scheduler_cache.go index b03e4e66371..ed3a52741b0 100644 --- a/backend/internal/repository/scheduler_cache.go +++ b/backend/internal/repository/scheduler_cache.go @@ -15,18 +15,22 @@ import ( ) const ( - schedulerBucketSetKey = "sched:buckets" - schedulerOutboxWatermarkKey = "sched:outbox:watermark" - schedulerAccountPrefix = "sched:acc:" - schedulerAccountMetaPrefix = "sched:meta:" - schedulerAccountLastUsedPrefix = "sched:acc:last_used:" - schedulerActivePrefix = "sched:active:" - schedulerReadyPrefix = "sched:ready:" - schedulerVersionPrefix = "sched:ver:" - schedulerEpochPrefix = "sched:epoch:" - schedulerRetiredPrefix = "sched:retired:" - schedulerSnapshotPrefix = "sched:" - schedulerLockPrefix = "sched:lock:" + schedulerBucketSetKey = "sched:buckets" + schedulerOutboxWatermarkKey = "sched:outbox:watermark" + schedulerAccountPrefix = "sched:acc:" + // schedulerAccountMetaPrefix is schema-versioned because metadata is read + // before the selected account is hydrated. Bump the namespace whenever a + // projection change affects pre-hydration scheduling behavior. + schedulerAccountMetaPrefix = "sched:meta:v2:" + schedulerLegacyAccountMetaPrefix = "sched:meta:" + schedulerAccountLastUsedPrefix = "sched:acc:last_used:" + schedulerActivePrefix = "sched:active:" + schedulerReadyPrefix = "sched:ready:" + schedulerVersionPrefix = "sched:ver:" + schedulerEpochPrefix = "sched:epoch:" + schedulerRetiredPrefix = "sched:retired:" + schedulerSnapshotPrefix = "sched:" + schedulerLockPrefix = "sched:lock:" defaultSchedulerSnapshotMGetChunkSize = 128 defaultSchedulerSnapshotWriteChunkSize = 256 @@ -604,7 +608,12 @@ func (c *schedulerCache) DeleteAccount(ctx context.Context, accountID int64) err return nil } id := strconv.FormatInt(accountID, 10) - return c.rdb.Del(ctx, schedulerAccountKey(id), schedulerAccountMetaKey(id), schedulerLastUsedKey(id)).Err() + return c.rdb.Del(ctx, + schedulerAccountKey(id), + schedulerAccountMetaKey(id), + schedulerLegacyAccountMetaKey(id), + schedulerLastUsedKey(id), + ).Err() } func (c *schedulerCache) UpdateLastUsed(ctx context.Context, updates map[int64]time.Time) error { @@ -636,7 +645,12 @@ func (c *schedulerCache) UpdateLastUsed(ctx context.Context, updates map[int64]t "error", err, ) idText := strconv.FormatInt(id, 10) - pipe.Del(ctx, schedulerAccountKey(idText), schedulerAccountMetaKey(idText), schedulerLastUsedKey(idText)) + pipe.Del(ctx, + schedulerAccountKey(idText), + schedulerAccountMetaKey(idText), + schedulerLegacyAccountMetaKey(idText), + schedulerLastUsedKey(idText), + ) queued++ continue } @@ -720,6 +734,10 @@ func schedulerAccountMetaKey(id string) string { return schedulerAccountMetaPrefix + id } +func schedulerLegacyAccountMetaKey(id string) string { + return schedulerLegacyAccountMetaPrefix + id +} + func schedulerLastUsedKey(id string) string { return schedulerAccountLastUsedPrefix + id } @@ -809,6 +827,7 @@ func (c *schedulerCache) writeAccountIDs(ctx context.Context, accounts []service id := strconv.FormatInt(account.ID, 10) pipe.Set(ctx, schedulerAccountKey(id), fullPayload, 0) pipe.Set(ctx, schedulerAccountMetaKey(id), metaPayload, 0) + pipe.Del(ctx, schedulerLegacyAccountMetaKey(id)) // Keep the hot LastUsedAt side key untouched: a lagging snapshot rebuild // must not overwrite a newer scheduler update. accountIDs = append(accountIDs, account.ID) @@ -954,7 +973,17 @@ func filterSchedulerCredentials(credentials map[string]any) map[string]any { if len(credentials) == 0 { return nil } - keys := []string{"model_mapping", "compact_model_mapping", "api_key", "project_id", "oauth_type", "plan_type"} + keys := []string{ + "model_mapping", + "compact_model_mapping", + "api_key", + "project_id", + "oauth_type", + "plan_type", + "account_scheduling_threshold", + "subscription_tier", + "team_id", + } filtered := make(map[string]any) for _, key := range keys { if value, ok := credentials[key]; ok && value != nil { @@ -1000,6 +1029,16 @@ func filterSchedulerExtra(extra map[string]any) map[string]any { "openai_ws_force_http", "openai_responses_mode", "openai_responses_supported", + "openai_passthrough", + "openai_oauth_passthrough", + "openai_compact_mode", + "openai_compact_supported", + "privacy_mode", + "session_window_utilization", + "passive_usage_7d_utilization", + "passive_usage_7d_reset", + "grok_sched_utilization", + "grok_sched_reset_at", "codex_5h_used_percent", "codex_7d_used_percent", "codex_5h_reset_at", diff --git a/backend/internal/repository/scheduler_cache_unit_test.go b/backend/internal/repository/scheduler_cache_unit_test.go index d0373f16c7b..ce6dd2a61e1 100644 --- a/backend/internal/repository/scheduler_cache_unit_test.go +++ b/backend/internal/repository/scheduler_cache_unit_test.go @@ -75,6 +75,8 @@ func TestSchedulerCacheUpdateLastUsedClearsUnencodablePayload(t *testing.T) { cache := newSchedulerCacheUnit(t) account := service.Account{ID: 114, Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey} require.NoError(t, cache.SetAccount(ctx, &account)) + legacyKey := schedulerLegacyAccountMetaKey(strconv.FormatInt(account.ID, 10)) + require.NoError(t, cache.rdb.Set(ctx, legacyKey, `{}`, 0).Err()) invalidTime := time.Date(10000, time.January, 1, 0, 0, 0, 0, time.UTC) require.NoError(t, cache.UpdateLastUsed(ctx, map[int64]time.Time{account.ID: invalidTime})) @@ -82,6 +84,7 @@ func TestSchedulerCacheUpdateLastUsedClearsUnencodablePayload(t *testing.T) { cached, err := cache.GetAccount(ctx, account.ID) require.NoError(t, err) require.Nil(t, cached) + require.EqualValues(t, 0, cache.rdb.Exists(ctx, legacyKey).Val()) } func TestSchedulerCacheSnapshotAccountIDReusePreservesPayloadAndMembers(t *testing.T) { @@ -369,6 +372,130 @@ func TestBuildSchedulerMetadataAccount_KeepsGrokMediaEligibility(t *testing.T) { }) } +func TestBuildSchedulerMetadataAccount_KeepsPreHydrationSchedulingFields(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + + t.Run("anthropic threshold override and windows", func(t *testing.T) { + windowEnd := now.Add(2 * time.Hour) + account := service.Account{ + ID: 90, + Platform: service.PlatformAnthropic, + Type: service.AccountTypeOAuth, + Status: service.StatusActive, + Schedulable: true, + SessionWindowEnd: &windowEnd, + Credentials: map[string]any{ + "account_scheduling_threshold": 80, + "access_token": "must-not-enter-metadata", + }, + Extra: map[string]any{ + "session_window_utilization": 0.85, + "passive_usage_7d_utilization": 0.90, + "passive_usage_7d_reset": now.Add(7 * 24 * time.Hour).Format(time.RFC3339), + "unrelated": "drop-me", + }, + } + + got := buildSchedulerMetadataAccount(account) + decision := service.EvaluateAccountSchedulingThreshold(&got, map[string]int{ + service.PlatformAnthropic: 95, + }, now) + + require.True(t, decision.ShouldPause) + require.Equal(t, 80, decision.ThresholdPercent) + require.Equal(t, "7d", decision.Window) + require.Empty(t, got.GetCredential("access_token")) + require.NotContains(t, got.Extra, "unrelated") + }) + + t.Run("grok threshold and scheduling identity", func(t *testing.T) { + account := service.Account{ + ID: 91, + Platform: service.PlatformGrok, + Type: service.AccountTypeOAuth, + Status: service.StatusActive, + Schedulable: true, + Credentials: map[string]any{ + "account_scheduling_threshold": 80, + "subscription_tier": "free", + "team_id": "team-a", + "refresh_token": "must-not-enter-metadata", + }, + Extra: map[string]any{ + "grok_sched_utilization": 90.0, + "grok_sched_reset_at": now.Add(time.Hour).Format(time.RFC3339), + }, + } + + got := buildSchedulerMetadataAccount(account) + decision := service.EvaluateAccountSchedulingThreshold(&got, map[string]int{ + service.PlatformGrok: 95, + }, now) + + require.True(t, decision.ShouldPause) + require.Equal(t, 80, decision.ThresholdPercent) + require.Equal(t, "quota", decision.Window) + require.Equal(t, "free", got.GetCredential("subscription_tier")) + require.Equal(t, "team-a", got.GetCredential("team_id")) + require.Empty(t, got.GetCredential("refresh_token")) + }) + + t.Run("openai privacy passthrough and compact routing", func(t *testing.T) { + account := service.Account{ + ID: 92, + Platform: service.PlatformOpenAI, + Type: service.AccountTypeOAuth, + Credentials: map[string]any{ + "model_mapping": map[string]any{"known-model": "known-model"}, + }, + Extra: map[string]any{ + "privacy_mode": service.PrivacyModeTrainingOff, + "openai_passthrough": true, + "openai_compact_mode": service.OpenAICompactModeForceOn, + "openai_compact_supported": true, + }, + } + + got := buildSchedulerMetadataAccount(account) + supported, known := got.OpenAICompactSupportKnown() + + require.True(t, got.IsPrivacySet()) + require.True(t, got.IsOpenAIPassthroughEnabled()) + require.True(t, got.IsModelSupported("unmapped-upstream-model")) + require.True(t, known) + require.True(t, supported) + }) +} + +func TestSchedulerMetadataSchemaVersionRejectsAndCleansLegacyPayload(t *testing.T) { + ctx := context.Background() + cache := newSchedulerCacheUnit(t) + account := service.Account{ID: 93, Platform: service.PlatformGrok, Type: service.AccountTypeOAuth} + bucket := service.SchedulerBucket{GroupID: 93, Platform: service.PlatformGrok, Mode: service.SchedulerModeSingle} + token, err := cache.CaptureBucketWriteToken(ctx, bucket) + require.NoError(t, err) + require.NoError(t, cache.SetSnapshot(ctx, bucket, token, []service.Account{account})) + + id := strconv.FormatInt(account.ID, 10) + metadata, err := cache.rdb.Get(ctx, schedulerAccountMetaKey(id)).Bytes() + require.NoError(t, err) + require.NoError(t, cache.rdb.Del(ctx, schedulerAccountMetaKey(id)).Err()) + require.NoError(t, cache.rdb.Set(ctx, schedulerLegacyAccountMetaKey(id), metadata, 0).Err()) + + snapshot, hit, err := cache.GetSnapshot(ctx, bucket) + require.NoError(t, err) + require.False(t, hit, "legacy metadata must force a database rebuild") + require.Nil(t, snapshot) + + require.NoError(t, cache.SetAccount(ctx, &account)) + require.EqualValues(t, 0, cache.rdb.Exists(ctx, schedulerLegacyAccountMetaKey(id)).Val()) + require.EqualValues(t, 1, cache.rdb.Exists(ctx, schedulerAccountMetaKey(id)).Val()) + + require.NoError(t, cache.rdb.Set(ctx, schedulerLegacyAccountMetaKey(id), metadata, 0).Err()) + require.NoError(t, cache.DeleteAccount(ctx, account.ID)) + require.EqualValues(t, 0, cache.rdb.Exists(ctx, schedulerLegacyAccountMetaKey(id)).Val()) +} + func TestBuildSchedulerMetadataAccount_KeepsSlimGroupMembership(t *testing.T) { account := service.Account{ ID: 42,