fix(scheduler): isolate LastUsedAt cache writes

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 {

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())
+	}
+}