diff --git a/packages/api/internal/sandbox/storage/redis/items.go b/packages/api/internal/sandbox/storage/redis/items.go index 7d65111228..4d86d612d8 100644 --- a/packages/api/internal/sandbox/storage/redis/items.go +++ b/packages/api/internal/sandbox/storage/redis/items.go @@ -38,6 +38,7 @@ func (s *Storage) ExpiredItems(ctx context.Context) ([]sandboxtypes.Sandbox, err // Group by team for per-team MGET (Redis Cluster slot compatibility). type memberRef struct { member string + teamID string sandboxID string executionID string } @@ -67,7 +68,7 @@ func (s *Storage) ExpiredItems(ctx context.Context) ([]sandboxtypes.Sandbox, err teamSandboxes[teamID] = entry } - entry.refs = append(entry.refs, memberRef{member: member, sandboxID: sandboxID, executionID: executionID}) + entry.refs = append(entry.refs, memberRef{member: member, teamID: teamID, sandboxID: sandboxID, executionID: executionID}) } pipe := s.redisClient.Pipeline() @@ -101,13 +102,20 @@ func (s *Storage) ExpiredItems(ctx context.Context) ([]sandboxtypes.Sandbox, err for i, raw := range batch.cmd.Val() { ref := batch.refs[i] - // Sandbox key gone but ZSET entry remains — orphaned. Members are - // execution-scoped, so this removal can never unindex a fresh - // execution concurrently re-added under the same sandbox ID. + // Sandbox key gone but ZSET entry remains — orphaned. + // MGET confirmed the key is absent; SREM the stale team index entry + // so it does not accumulate indefinitely. Sandbox IDs are randomly + // generated and never reused, so a concurrent Add cannot write the + // same sandboxID back after this SREM. if raw == nil { staleMembers = append(staleMembers, ref.member) orphanCount++ + teamIndexKey := GetSandboxStorageTeamIndexKey(ref.teamID) + if sremErr := s.redisClient.SRem(ctx, teamIndexKey, ref.sandboxID).Err(); sremErr != nil { + logger.L().Warn(ctx, "Failed to remove orphan from team index", zap.Error(sremErr), logger.WithSandboxID(ref.sandboxID)) + } + continue } diff --git a/packages/api/internal/sandbox/storage/redis/operations.go b/packages/api/internal/sandbox/storage/redis/operations.go index 7fa0574b69..e1152820ff 100644 --- a/packages/api/internal/sandbox/storage/redis/operations.go +++ b/packages/api/internal/sandbox/storage/redis/operations.go @@ -38,9 +38,13 @@ func (s *Storage) Add(ctx context.Context, sbx sandboxtypes.Sandbox) error { return fmt.Errorf("failed to add sandbox to global expiration index: %w", err) } - // Execute Lua script for atomic SET + SADD - err = addSandboxScript.Run(ctx, s.redisClient, []string{key, teamKey}, data, sbx.SandboxID).Err() - if err != nil { + // MULTI/EXEC: SET sandboxKey + SADD teamIndex — no conditional logic needed, + // both keys share the {teamID} hash tag so they land on the same cluster slot. + if _, err = s.redisClient.TxPipelined(ctx, func(pipe redis.Pipeliner) error { + pipe.Set(ctx, key, data, 0) + pipe.SAdd(ctx, teamKey, sbx.SandboxID) + return nil + }); err != nil { return fmt.Errorf("failed to store sandbox in Redis: %w", err) } @@ -93,13 +97,19 @@ func (s *Storage) Remove(ctx context.Context, teamID uuid.UUID, sandboxID string } }() - // Execute Lua script for atomic DEL + SREM; it returns the deleted JSON - // so the expiration-index cleanup below is scoped to the execution we - // actually removed. - raw, err := removeSandboxScript.Run(ctx, s.redisClient, []string{key, teamKey}, sandboxID).Text() - if err != nil && !errors.Is(err, redis.Nil) { + // MULTI/EXEC: GET + DEL sandboxKey + SREM teamIndex atomically. + // GET is first so we can scope expiration-index cleanup to the execution we + // actually removed. Both keys share the {teamID} hash tag (same cluster slot). + var getCmd *redis.StringCmd + if _, err = s.redisClient.TxPipelined(ctx, func(pipe redis.Pipeliner) error { + getCmd = pipe.Get(ctx, key) + pipe.Del(ctx, key) + pipe.SRem(ctx, teamKey, sandboxID) + return nil + }); err != nil && !errors.Is(err, redis.Nil) { return fmt.Errorf("failed to remove sandbox from Redis: %w", err) } + raw := getCmd.Val() // Clean up from the global expiration index. // Do it after the removal to prevent leaking expired sandboxes. diff --git a/packages/api/internal/sandbox/storage/redis/scripts.go b/packages/api/internal/sandbox/storage/redis/scripts.go index 2786d1322a..6ce04d2b16 100644 --- a/packages/api/internal/sandbox/storage/redis/scripts.go +++ b/packages/api/internal/sandbox/storage/redis/scripts.go @@ -2,31 +2,7 @@ package redis import "github.com/redis/go-redis/v9" -// Lua scripts for atomic operations. -// These scripts ensure true atomicity in Redis cluster mode var ( - // addSandboxScript atomically stores a sandbox and adds it to the team index. - // KEYS[1] = sandbox key, KEYS[2] = team index key - // ARGV[1] = serialized sandbox data, ARGV[2] = sandbox ID - addSandboxScript = redis.NewScript(` - redis.call('SET', KEYS[1], ARGV[1]) - redis.call('SADD', KEYS[2], ARGV[2]) - return 1 - `) - - // removeSandboxScript atomically removes a sandbox and its team index entry. - // It returns the stored JSON (or nil if the key was already gone) so the - // caller knows exactly which execution it removed and can scope the - // expiration-index cleanup to that execution. - // KEYS[1] = sandbox key, KEYS[2] = team index key - // ARGV[1] = sandbox ID - removeSandboxScript = redis.NewScript(` - local data = redis.call('GET', KEYS[1]) - redis.call('DEL', KEYS[1]) - redis.call('SREM', KEYS[2], ARGV[1]) - return data - `) - // startTransitionScript atomically updates sandbox and sets transition key with UUID. // This is called AFTER Go code has validated the transition and prepared the new sandbox data. //