From 81e7397770c001100dcdddae80177f6f72cfb667 Mon Sep 17 00:00:00 2001 From: superman2003 <2112076433zcr@gmail.com> Date: Tue, 21 Jul 2026 13:27:23 +0800 Subject: [PATCH] fix(scheduler): isolate LastUsedAt cache writes --- .../internal/repository/scheduler_cache.go | 167 +++++++++++----- .../scheduler_cache_last_used_unit_test.go | 183 ++++++++++++++++++ 2 files changed, 306 insertions(+), 44 deletions(-) create mode 100644 backend/internal/repository/scheduler_cache_last_used_unit_test.go diff --git a/backend/internal/repository/scheduler_cache.go b/backend/internal/repository/scheduler_cache.go index 13bf106df6..746cc51204 100644 --- a/backend/internal/repository/scheduler_cache.go +++ b/backend/internal/repository/scheduler_cache.go @@ -15,20 +15,22 @@ import ( ) const ( - schedulerBucketSetKey = "sched:buckets" - schedulerOutboxWatermarkKey = "sched:outbox:watermark" - schedulerAccountPrefix = "sched:acc:" - schedulerAccountMetaPrefix = "sched:meta:" - 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 = "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 + schedulerLastUsedUpdateChunkSize = 256 // snapshotGraceTTLSeconds 旧快照过期的宽限期（秒）。 // 替代立即 DEL，让正在读取旧版本的 reader 有足够时间完成 ZRANGE。 @@ -40,6 +42,25 @@ const ( schedulerGroupLifecycleOwnerTokenBytes = 16 ) +var updateSchedulerLastUsedScript = redis.NewScript(` +local updated = 0 +for index = 1, #ARGV do + local key_index = (index - 1) * 2 + 1 + local candidate = tonumber(ARGV[index]) + if candidate == nil then + return redis.error_reply('invalid last_used value') + end + if redis.call('EXISTS', KEYS[key_index]) == 1 then + local current = tonumber(redis.call('GET', KEYS[key_index + 1])) + if current == nil or candidate > current then + redis.call('SET', KEYS[key_index + 1], ARGV[index]) + updated = updated + 1 + end + end +end +return updated +`) + var ( // epoch 标识 bucket writer 的代际，retired key 是持久退休标记。 // Capture、allocate、activate 都在 Lua 内同时校验两者：-1 表示已退休，-2 表示 epoch 无效或与 token 代际不匹配； @@ -256,16 +277,22 @@ func (c *schedulerCache) GetSnapshot(ctx context.Context, bucket service.Schedul } keys := make([]string, 0, len(ids)) + lastUsedKeys := make([]string, 0, len(ids)) for _, id := range ids { keys = append(keys, schedulerAccountMetaKey(id)) + lastUsedKeys = append(lastUsedKeys, schedulerLastUsedKey(id)) } values, err := c.mgetChunked(ctx, keys) if err != nil { return nil, false, err } + lastUsedValues, err := c.mgetChunked(ctx, lastUsedKeys) + if err != nil { + return nil, false, err + } accounts := make([]*service.Account, 0, len(values)) - for _, val := range values { + for i, val := range values { if val == nil { return nil, false, nil } @@ -273,6 +300,9 @@ func (c *schedulerCache) GetSnapshot(ctx context.Context, bucket service.Schedul if err != nil { return nil, false, err } + if err := applySchedulerLastUsed(account, lastUsedValues[i]); err != nil { + return nil, false, err + } accounts = append(accounts, account) } @@ -537,15 +567,22 @@ func schedulerBucketWriteResultError(result int64, bucket service.SchedulerBucke } func (c *schedulerCache) GetAccount(ctx context.Context, accountID int64) (*service.Account, error) { - key := schedulerAccountKey(strconv.FormatInt(accountID, 10)) - val, err := c.rdb.Get(ctx, key).Result() - if err == redis.Nil { - return nil, nil - } + id := strconv.FormatInt(accountID, 10) + values, err := c.rdb.MGet(ctx, schedulerAccountKey(id), schedulerLastUsedKey(id)).Result() if err != nil { return nil, err } - return decodeCachedAccount(val) + if len(values) != 2 || values[0] == nil { + return nil, nil + } + account, err := decodeCachedAccount(values[0]) + if err != nil { + return nil, err + } + if err := applySchedulerLastUsed(account, values[1]); err != nil { + return nil, err + } + return account, nil } func (c *schedulerCache) SetAccount(ctx context.Context, account *service.Account) error { @@ -567,7 +604,7 @@ 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)).Err() + return c.rdb.Del(ctx, schedulerAccountKey(id), schedulerAccountMetaKey(id), schedulerLastUsedKey(id)).Err() } func (c *schedulerCache) UpdateLastUsed(ctx context.Context, updates map[int64]time.Time) error { @@ -575,41 +612,46 @@ func (c *schedulerCache) UpdateLastUsed(ctx context.Context, updates map[int64]t return nil } - keys := make([]string, 0, len(updates)) - ids := make([]int64, 0, len(updates)) - for id := range updates { - keys = append(keys, schedulerAccountKey(strconv.FormatInt(id, 10))) - ids = append(ids, id) - } - - values, err := c.mgetChunked(ctx, keys) - if err != nil { - return err - } - pipe := c.rdb.Pipeline() - for i, val := range values { - if val == nil { + queued := 0 + keys := make([]string, 0, schedulerLastUsedUpdateChunkSize*2) + args := make([]any, 0, schedulerLastUsedUpdateChunkSize) + queueBatch := func() { + if len(args) == 0 { + return + } + updateSchedulerLastUsedScript.Eval(ctx, pipe, keys, args...) + queued++ + keys = make([]string, 0, schedulerLastUsedUpdateChunkSize*2) + args = make([]any, 0, schedulerLastUsedUpdateChunkSize) + } + for id, usedAt := range updates { + if id <= 0 { continue } - account, err := decodeCachedAccount(val) - if err != nil { - return err - } - account.LastUsedAt = ptrTime(updates[ids[i]]) - updated, metaPayload, err := marshalSchedulerCacheAccount(*account) + millis, err := schedulerLastUsedMillis(usedAt) if err != nil { slog.Warn("scheduler cache removes account with unencodable payload", - "account_id", ids[i], + "account_id", id, "error", err, ) - pipe.Del(ctx, keys[i], schedulerAccountMetaKey(strconv.FormatInt(ids[i], 10))) + idText := strconv.FormatInt(id, 10) + pipe.Del(ctx, schedulerAccountKey(idText), schedulerAccountMetaKey(idText), schedulerLastUsedKey(idText)) + queued++ continue } - pipe.Set(ctx, keys[i], updated, 0) - pipe.Set(ctx, schedulerAccountMetaKey(strconv.FormatInt(ids[i], 10)), metaPayload, 0) + idText := strconv.FormatInt(id, 10) + keys = append(keys, schedulerAccountKey(idText), schedulerLastUsedKey(idText)) + args = append(args, millis) + if len(args) >= schedulerLastUsedUpdateChunkSize { + queueBatch() + } } - _, err = pipe.Exec(ctx) + queueBatch() + if queued == 0 { + return nil + } + _, err := pipe.Exec(ctx) return err } @@ -678,10 +720,45 @@ func schedulerAccountMetaKey(id string) string { return schedulerAccountMetaPrefix + id } +func schedulerLastUsedKey(id string) string { + return schedulerAccountLastUsedPrefix + id +} + func ptrTime(t time.Time) *time.Time { return &t } +func schedulerLastUsedMillis(value time.Time) (int64, error) { + if _, err := value.MarshalJSON(); err != nil { + return 0, err + } + return value.UTC().UnixMilli(), nil +} + +func applySchedulerLastUsed(account *service.Account, value any) error { + if account == nil || value == nil { + return nil + } + var raw string + switch typed := value.(type) { + case string: + raw = typed + case []byte: + raw = string(typed) + default: + return fmt.Errorf("unexpected last_used cache type: %T", value) + } + millis, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return fmt.Errorf("invalid last_used cache value %q: %w", raw, err) + } + lastUsedAt := time.UnixMilli(millis).UTC() + if account.LastUsedAt == nil || lastUsedAt.After(*account.LastUsedAt) { + account.LastUsedAt = ptrTime(lastUsedAt) + } + return nil +} + func decodeCachedAccount(val any) (*service.Account, error) { var payload []byte switch raw := val.(type) { @@ -732,6 +809,8 @@ 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) + // Keep the hot LastUsedAt side key untouched: a lagging snapshot rebuild + // must not overwrite a newer scheduler update. accountIDs = append(accountIDs, account.ID) pending++ if pending >= c.writeChunkSize { diff --git a/backend/internal/repository/scheduler_cache_last_used_unit_test.go b/backend/internal/repository/scheduler_cache_last_used_unit_test.go new file mode 100644 index 0000000000..06810da9db --- /dev/null +++ b/backend/internal/repository/scheduler_cache_last_used_unit_test.go @@ -0,0 +1,183 @@ +//go:build unit + +package repository + +import ( + "context" + "strconv" + "strings" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/require" +) + +func TestSchedulerCacheUpdateLastUsedUsesSideKeyWithoutRewritingPayloads(t *testing.T) { + ctx := context.Background() + cache := newSchedulerCacheUnit(t) + bucket := service.SchedulerBucket{ + GroupID: 9, + Platform: service.PlatformGrok, + Mode: service.SchedulerModeSingle, + } + initial := time.Now().UTC().Truncate(time.Millisecond).Add(-time.Hour) + account := service.Account{ + ID: 9201, + Name: "grok-large-oauth", + Platform: service.PlatformGrok, + Type: service.AccountTypeOAuth, + Status: service.StatusActive, + Schedulable: true, + LastUsedAt: &initial, + Credentials: map[string]any{ + "access_token": strings.Repeat("a", 4096), + "refresh_token": strings.Repeat("r", 4096), + }, + Extra: map[string]any{"large": strings.Repeat("x", 4096)}, + } + 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) + fullBefore, err := cache.rdb.Get(ctx, schedulerAccountKey(id)).Bytes() + require.NoError(t, err) + metaBefore, err := cache.rdb.Get(ctx, schedulerAccountMetaKey(id)).Bytes() + require.NoError(t, err) + + latest := initial.Add(37 * time.Second) + require.NoError(t, cache.UpdateLastUsed(ctx, map[int64]time.Time{account.ID: latest})) + + fullAfter, err := cache.rdb.Get(ctx, schedulerAccountKey(id)).Bytes() + require.NoError(t, err) + metaAfter, err := cache.rdb.Get(ctx, schedulerAccountMetaKey(id)).Bytes() + require.NoError(t, err) + require.Equal(t, fullBefore, fullAfter) + require.Equal(t, metaBefore, metaAfter) + require.Equal(t, strconv.FormatInt(latest.UnixMilli(), 10), cache.rdb.Get(ctx, schedulerLastUsedKey(id)).Val()) + + cached, err := cache.GetAccount(ctx, account.ID) + require.NoError(t, err) + require.NotNil(t, cached) + require.NotNil(t, cached.LastUsedAt) + require.Equal(t, latest, *cached.LastUsedAt) + + snapshot, hit, err := cache.GetSnapshot(ctx, bucket) + require.NoError(t, err) + require.True(t, hit) + require.Len(t, snapshot, 1) + require.NotNil(t, snapshot[0].LastUsedAt) + require.Equal(t, latest, *snapshot[0].LastUsedAt) +} + +func TestSchedulerCacheLastUsedSideKeyIsMonotonicAndRequiresAccount(t *testing.T) { + ctx := context.Background() + cache := newSchedulerCacheUnit(t) + account := service.Account{ID: 9202, Platform: service.PlatformGrok, Type: service.AccountTypeOAuth} + require.NoError(t, cache.SetAccount(ctx, &account)) + + newer := time.Now().UTC().Truncate(time.Millisecond) + older := newer.Add(-time.Minute) + require.NoError(t, cache.UpdateLastUsed(ctx, map[int64]time.Time{account.ID: newer})) + require.NoError(t, cache.UpdateLastUsed(ctx, map[int64]time.Time{account.ID: older})) + + id := strconv.FormatInt(account.ID, 10) + require.Equal(t, strconv.FormatInt(newer.UnixMilli(), 10), cache.rdb.Get(ctx, schedulerLastUsedKey(id)).Val()) + cached, err := cache.GetAccount(ctx, account.ID) + require.NoError(t, err) + require.NotNil(t, cached) + require.Equal(t, newer, *cached.LastUsedAt) + + const missingID int64 = 9299 + require.NoError(t, cache.UpdateLastUsed(ctx, map[int64]time.Time{missingID: newer})) + _, err = cache.rdb.Get(ctx, schedulerLastUsedKey(strconv.FormatInt(missingID, 10))).Result() + require.ErrorIs(t, err, redis.Nil) + + require.NoError(t, cache.DeleteAccount(ctx, account.ID)) + _, err = cache.rdb.Get(ctx, schedulerLastUsedKey(id)).Result() + require.ErrorIs(t, err, redis.Nil) +} + +func TestSchedulerCacheLastUsedSideKeyFallsBackToNewerEmbeddedValue(t *testing.T) { + ctx := context.Background() + cache := newSchedulerCacheUnit(t) + embedded := time.Now().UTC().Truncate(time.Millisecond) + account := service.Account{ + ID: 9203, + Platform: service.PlatformGrok, + Type: service.AccountTypeOAuth, + LastUsedAt: &embedded, + } + require.NoError(t, cache.SetAccount(ctx, &account)) + + id := strconv.FormatInt(account.ID, 10) + require.NoError(t, cache.rdb.Set(ctx, schedulerLastUsedKey(id), embedded.Add(-time.Hour).UnixMilli(), 0).Err()) + cached, err := cache.GetAccount(ctx, account.ID) + require.NoError(t, err) + require.NotNil(t, cached) + require.Equal(t, embedded, *cached.LastUsedAt) +} + +func TestSchedulerCacheLastUsedSideKeySurvivesStaleAccountAndSnapshotWrites(t *testing.T) { + ctx := context.Background() + cache := newSchedulerCacheUnit(t) + bucket := service.SchedulerBucket{ + GroupID: 10, + Platform: service.PlatformGrok, + Mode: service.SchedulerModeSingle, + } + embedded := time.Now().UTC().Truncate(time.Millisecond).Add(-time.Minute) + latest := embedded.Add(30 * time.Second) + account := service.Account{ + ID: 9204, + Platform: service.PlatformGrok, + Type: service.AccountTypeOAuth, + Schedulable: true, + LastUsedAt: &embedded, + } + require.NoError(t, cache.SetAccount(ctx, &account)) + require.NoError(t, cache.UpdateLastUsed(ctx, map[int64]time.Time{account.ID: latest})) + + require.NoError(t, cache.SetAccount(ctx, &account)) + 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) + require.Equal(t, strconv.FormatInt(latest.UnixMilli(), 10), cache.rdb.Get(ctx, schedulerLastUsedKey(id)).Val()) + cached, err := cache.GetAccount(ctx, account.ID) + require.NoError(t, err) + require.NotNil(t, cached) + require.Equal(t, latest, *cached.LastUsedAt) + snapshot, hit, err := cache.GetSnapshot(ctx, bucket) + require.NoError(t, err) + require.True(t, hit) + require.Len(t, snapshot, 1) + require.Equal(t, latest, *snapshot[0].LastUsedAt) +} + +func TestSchedulerCacheUpdateLastUsedChunksLargeBatches(t *testing.T) { + ctx := context.Background() + cache := newSchedulerCacheUnit(t) + total := schedulerLastUsedUpdateChunkSize + 1 + accounts := make([]service.Account, 0, total) + updates := make(map[int64]time.Time, total) + base := time.Now().UTC().Truncate(time.Millisecond) + for i := 0; i < total; i++ { + id := int64(9300 + i) + accounts = append(accounts, service.Account{ID: id, Platform: service.PlatformGrok}) + updates[id] = base.Add(time.Duration(i) * time.Millisecond) + } + + written, err := cache.writeAccountIDs(ctx, accounts) + require.NoError(t, err) + require.Len(t, written, total) + require.NoError(t, cache.UpdateLastUsed(ctx, updates)) + + for id, usedAt := range updates { + key := schedulerLastUsedKey(strconv.FormatInt(id, 10)) + require.Equal(t, strconv.FormatInt(usedAt.UnixMilli(), 10), cache.rdb.Get(ctx, key).Val()) + } +}