diff --git a/packages/api/internal/sandbox/storage/redis/execution_pin_test.go b/packages/api/internal/sandbox/storage/redis/execution_pin_test.go index 9d95dd04a2..fba4e8f99b 100644 --- a/packages/api/internal/sandbox/storage/redis/execution_pin_test.go +++ b/packages/api/internal/sandbox/storage/redis/execution_pin_test.go @@ -172,9 +172,15 @@ func transitionKeysFor(t *testing.T, client redis.UniversalClient, sbx sandboxty transitionKeyFor(sbx.SandboxID), transitionResultKeyFor(sbx.SandboxID, transitionID), lockKey, + metaKeyFor(key), } } +// metaOf packs a record's execution ID into the meta value for tests (no hold). +func metaOf(eid string) string { + return eid + metaFieldSep + metaFieldSep + "0" +} + const transitionLockValue = "token" func TestStartTransitionScript_RefusesAStaleIncarnation(t *testing.T) { @@ -192,7 +198,7 @@ func TestStartTransitionScript_RefusesAStaleIncarnation(t *testing.T) { written, err := startTransitionScript.Run(ctx, client, keys, `{"executionID":"clobbered"}`, transitionID, 60, 60, uuid.NewString(), // the incarnation the caller pinned, no longer stored - transitionLockValue, + transitionLockValue, metaOf("clobbered"), ).Int64() require.NoError(t, err) assert.Equal(t, int64(0), written, "the script must refuse a stale pin") @@ -223,7 +229,7 @@ func TestStartTransitionScript_ProceedsOnTheMatchingIncarnation(t *testing.T) { require.NoError(t, err) written, err := startTransitionScript.Run(ctx, client, keys, - data, transitionID, 60, 60, sbx.ExecutionID, transitionLockValue, + data, transitionID, 60, 60, sbx.ExecutionID, transitionLockValue, metaOf(sbx.ExecutionID), ).Int64() require.NoError(t, err) assert.Equal(t, int64(1), written) @@ -250,7 +256,7 @@ func TestStartTransitionScript_RefusesADeletedRecord(t *testing.T) { keys := transitionKeysFor(t, client, sbx, transitionID) written, err := startTransitionScript.Run(ctx, client, keys, - `{"executionID":"resurrected"}`, transitionID, 60, 60, sbx.ExecutionID, transitionLockValue, + `{"executionID":"resurrected"}`, transitionID, 60, 60, sbx.ExecutionID, transitionLockValue, metaOf("resurrected"), ).Int64() require.NoError(t, err) assert.Equal(t, int64(0), written) @@ -278,7 +284,7 @@ func TestStartTransitionScript_UnpinnedWritesUnconditionally(t *testing.T) { require.NoError(t, err) written, err := startTransitionScript.Run(ctx, client, keys, - data, transitionID, 60, 60, "", transitionLockValue, + data, transitionID, 60, 60, "", transitionLockValue, metaOf(sbx.ExecutionID), ).Int64() require.NoError(t, err) assert.Equal(t, int64(1), written) @@ -308,7 +314,7 @@ func TestStartTransitionScript_RefusesALostLock(t *testing.T) { require.NoError(t, err) written, err := startTransitionScript.Run(ctx, client, keys, - data, transitionID, 60, 60, sbx.ExecutionID, transitionLockValue, + data, transitionID, 60, 60, sbx.ExecutionID, transitionLockValue, metaOf(sbx.ExecutionID), ).Int64() require.NoError(t, err) assert.Equal(t, int64(3), written) diff --git a/packages/api/internal/sandbox/storage/redis/heal.go b/packages/api/internal/sandbox/storage/redis/heal.go index d4cf891662..f09487e3a5 100644 --- a/packages/api/internal/sandbox/storage/redis/heal.go +++ b/packages/api/internal/sandbox/storage/redis/heal.go @@ -507,7 +507,7 @@ func (s *Storage) healUnindexed(ctx context.Context, lock *storageLock, ref reco if now.Sub(sbx.EndTime) <= sandboxtypes.StaleCutoff { return "", nil } - deleted, _, _, err := parseRemoveReply(RunScript(ctx, s.redisClient, "remove_record", removeSandboxScript, []string{key, goneKeyFor(key), lock.Key()}, ref.SandboxID, sbx.ExecutionID, int(goneExecutionsTTL.Seconds()), "", lockValueOf(lock))) + deleted, _, _, err := parseRemoveReply(RunScript(ctx, s.redisClient, "remove_record", removeSandboxScript, []string{key, goneKeyFor(key), lock.Key(), metaKeyFor(key)}, ref.SandboxID, sbx.ExecutionID, int(goneExecutionsTTL.Seconds()), "", lockValueOf(lock))) if err != nil { return "", fmt.Errorf("remove_record: %w", err) } diff --git a/packages/api/internal/sandbox/storage/redis/lifecycle_hold.go b/packages/api/internal/sandbox/storage/redis/lifecycle_hold.go index 7ed5a6ff1c..2122970bdb 100644 --- a/packages/api/internal/sandbox/storage/redis/lifecycle_hold.go +++ b/packages/api/internal/sandbox/storage/redis/lifecycle_hold.go @@ -175,7 +175,7 @@ func (s *Storage) writeHeld(ctx context.Context, lock *storageLock, sbx sandboxt if err != nil { return fmt.Errorf("failed to marshal sandbox: %w", err) } - if err := s.setUnderLock(ctx, lock, recordKey(sbx.SandboxID), data); err != nil { + if err := s.setUnderLock(ctx, lock, recordKey(sbx.SandboxID), data, packMeta(sbx)); err != nil { return fmt.Errorf("failed to store sandbox in Redis: %w", err) } diff --git a/packages/api/internal/sandbox/storage/redis/lifecycle_hold_test.go b/packages/api/internal/sandbox/storage/redis/lifecycle_hold_test.go index 358a8452d9..94e0f0752d 100644 --- a/packages/api/internal/sandbox/storage/redis/lifecycle_hold_test.go +++ b/packages/api/internal/sandbox/storage/redis/lifecycle_hold_test.go @@ -134,7 +134,7 @@ func TestLifecycleHold(t *testing.T) { // the hold taken since, which the Go check before it could not see. transitionID := uuid.NewString() keys := transitionKeysFor(t, client, sbx, transitionID) - written, err := RunScript(ctx, client, "start_transition", startTransitionScript, keys, `{"executionID":"stale"}`, transitionID, 60, 60, "", transitionLockValue).Int64() + written, err := RunScript(ctx, client, "start_transition", startTransitionScript, keys, `{"executionID":"stale"}`, transitionID, 60, 60, "", transitionLockValue, metaOf("stale")).Int64() require.NoError(t, err) assert.Equal(t, int64(2), written) assert.Zero(t, client.Exists(ctx, keys[1]).Val(), "no transition started") @@ -294,7 +294,7 @@ func TestLifecycleHoldExpires(t *testing.T) { until := time.Now().Add(300 * time.Millisecond) require.NoError(t, holdExecution(t, storage, short, "op", until)) lockRecord(t, client, shortKey) - removeKeys := []string{shortKey, goneKeyFor(shortKey), redis_utils.GetLockKey(shortKey)} + removeKeys := []string{shortKey, goneKeyFor(shortKey), redis_utils.GetLockKey(shortKey), metaKeyFor(shortKey)} deleted, _, _, err := parseRemoveReply(RunScript(ctx, client, "remove_record", removeSandboxScript, removeKeys, short.SandboxID, short.ExecutionID, 60, "", testLockValue)) require.NoError(t, err) assert.False(t, deleted, "refused while held") diff --git a/packages/api/internal/sandbox/storage/redis/operations.go b/packages/api/internal/sandbox/storage/redis/operations.go index 1963952b6f..e026d32e70 100644 --- a/packages/api/internal/sandbox/storage/redis/operations.go +++ b/packages/api/internal/sandbox/storage/redis/operations.go @@ -94,14 +94,14 @@ func (s *Storage) add(ctx context.Context, sbx sandboxtypes.Sandbox, replace boo // The commit point, the gone set beside the record in its slot; the // index entry lives in another slot and follows. - keys := []string{key, goneKeyFor(key)} + keys := []string{key, goneKeyFor(key), metaKeyFor(key)} replaceFlag := "0" if replace { replaceFlag = "1" } var written int64 err = s.run(faultAddBeforeRecordWrite, func() (runErr error) { - written, runErr = RunScript(ctx, s.redisClient, "add_sandbox", addSandboxScript, keys, data, sbx.SandboxID, sbx.ExecutionID, replaceFlag, int(goneExecutionsTTL.Seconds())).Int64() + written, runErr = RunScript(ctx, s.redisClient, "add_sandbox", addSandboxScript, keys, data, sbx.SandboxID, sbx.ExecutionID, replaceFlag, int(goneExecutionsTTL.Seconds()), packMeta(sbx)).Int64() return runErr }) @@ -426,7 +426,7 @@ func (s *Storage) removeSandboxRecord(ctx context.Context, key, teamKey string, var raw string for range removeRecordRetries + 1 { err = s.run(faultRemoveAfterSREM, func() (runErr error) { - deleted, raw, absent, runErr = parseRemoveReply(RunScript(ctx, s.redisClient, "remove_record", removeSandboxScript, []string{key, goneKeyFor(key), redis_utils.GetLockKey(key)}, sandboxID, expectedExecutionID, int(goneExecutionsTTL.Seconds()), holdID, lockValueOf(lock))) + deleted, raw, absent, runErr = parseRemoveReply(RunScript(ctx, s.redisClient, "remove_record", removeSandboxScript, []string{key, goneKeyFor(key), redis_utils.GetLockKey(key), metaKeyFor(key)}, sandboxID, expectedExecutionID, int(goneExecutionsTTL.Seconds()), holdID, lockValueOf(lock))) return runErr }) @@ -521,7 +521,7 @@ func (s *Storage) Update(ctx context.Context, teamID uuid.UUID, sandboxID string // The update may have acted outside the store already (a network update // applies on the node first), so the write is retried, and fenced by the lock. - err = s.setUnderLock(ctx, lock, key, newData) + err = s.setUnderLock(ctx, lock, key, newData, packMeta(updatedSbx)) if err != nil { return sandboxtypes.Sandbox{}, fmt.Errorf("failed to store sandbox in Redis: %w", err) } @@ -612,7 +612,7 @@ const updateWriteBackoff = 50 * time.Millisecond // caller's token, checked and written in one script, so a write that lands // late cannot overwrite what a later lock holder stored. A failed write is // retried a bounded number of times; a lost lock is not. -func (s *Storage) setUnderLock(ctx context.Context, lock *storageLock, key string, data []byte) error { +func (s *Storage) setUnderLock(ctx context.Context, lock *storageLock, key string, data []byte, meta string) error { var err error for attempt := range updateWriteAttempts { if attempt > 0 { @@ -623,7 +623,7 @@ func (s *Storage) setUnderLock(ctx context.Context, lock *storageLock, key strin } } var written int64 - written, err = RunScript(ctx, s.redisClient, "fenced_set", fencedSetScript, []string{lock.Key(), key}, lockValueOf(lock), data).Int64() + written, err = RunScript(ctx, s.redisClient, "fenced_set", fencedSetScript, []string{lock.Key(), key, metaKeyFor(key)}, lockValueOf(lock), data, meta).Int64() if err == nil { if written == 1 { return nil diff --git a/packages/api/internal/sandbox/storage/redis/reinstall_test.go b/packages/api/internal/sandbox/storage/redis/reinstall_test.go index 1773fa5e3c..9ecf9d2379 100644 --- a/packages/api/internal/sandbox/storage/redis/reinstall_test.go +++ b/packages/api/internal/sandbox/storage/redis/reinstall_test.go @@ -94,9 +94,9 @@ func reinstallBeforeAssert(ctx context.Context, client redis.UniversalClient, sb } key := recordKey(sbx.SandboxID) - keys := []string{key, goneKeyFor(key)} + keys := []string{key, goneKeyFor(key), metaKeyFor(key)} - return RunScript(ctx, client, "add_sandbox", addSandboxScript, keys, data, sbx.SandboxID, sbx.ExecutionID, "1", int(goneExecutionsTTL.Seconds())).Err() + return RunScript(ctx, client, "add_sandbox", addSandboxScript, keys, data, sbx.SandboxID, sbx.ExecutionID, "1", int(goneExecutionsTTL.Seconds()), packMeta(sbx)).Err() } // reinstalls are the two ways a Replace can have landed when a removal of diff --git a/packages/api/internal/sandbox/storage/redis/scripts.go b/packages/api/internal/sandbox/storage/redis/scripts.go index f0b820605a..89c891da5c 100644 --- a/packages/api/internal/sandbox/storage/redis/scripts.go +++ b/packages/api/internal/sandbox/storage/redis/scripts.go @@ -34,10 +34,35 @@ const isHeldLua = ` end ` +// parseMetaLua defines parse_meta, which reads the fields the hot path needs — +// execution ID and whether the record is held — from packMeta's string without +// decoding the record JSON. Layout: executionID US lifecycleHold US +// lifecycleHoldUntil(ms). Returns (executionID, held). A hold whose deadline +// Redis's clock has reached is released, matching is_held. +const parseMetaLua = ` + local function parse_meta(s) + local sep = string.char(31) + local i1 = string.find(s, sep, 1, true) + if not i1 then return s, false end + local eid = string.sub(s, 1, i1 - 1) + local i2 = string.find(s, sep, i1 + 1, true) + if not i2 then return eid, false end + local hold = string.sub(s, i1 + 1, i2 - 1) + if hold == '' then return eid, false end + local until_ms = tonumber(string.sub(s, i2 + 1)) + if not until_ms then return eid, false end + local now = redis.call('TIME') + return eid, until_ms > tonumber(now[1]) * 1000 + math.floor(tonumber(now[2]) / 1000) + end +` + // Lua scripts for atomic operations. // These scripts ensure true atomicity in Redis cluster mode var ( // A write past an expired lock can replace the record; compare it atomically before restoring it. + // KEYS[1] = record key, KEYS[2] = meta key + // ARGV[1] = expected current record, ARGV[2] = updated record, + // ARGV[3] = packed meta of the updated record (see packMeta) restoreSandboxScript = redis.NewScript(` local current = redis.call('GET', KEYS[1]) if current == ARGV[2] then @@ -47,19 +72,22 @@ var ( return 0 end redis.call('SET', KEYS[1], ARGV[2], 'KEEPTTL') + redis.call('SET', KEYS[2], ARGV[3]) return 1 `) // fencedSetScript writes a record, keeping its TTL, only while the lock // key still holds the writer's value. The keys share a slot: the lock key // is the record key with a prefix. - // KEYS[1] = lock key, KEYS[2] = record key - // ARGV[1] = lock value (token and metadata), ARGV[2] = serialized record + // KEYS[1] = lock key, KEYS[2] = record key, KEYS[3] = meta key + // ARGV[1] = lock value (token and metadata), ARGV[2] = serialized record, + // ARGV[3] = packed meta (see packMeta) fencedSetScript = redis.NewScript(` if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end redis.call('SET', KEYS[2], ARGV[2], 'KEEPTTL') + redis.call('SET', KEYS[3], ARGV[3]) return 1 `) @@ -81,10 +109,11 @@ var ( // (KEYS[2], in the record's slot), which holds the executions removed or // replaced under this sandbox ID for ARGV[5] seconds. A write that // replaces another execution's record puts that execution in the gone set. - // KEYS[1] = sandbox key, KEYS[2] = gone set + // KEYS[1] = sandbox key, KEYS[2] = gone set, KEYS[3] = meta key // ARGV[1] = serialized sandbox data, ARGV[2] = sandbox ID, // ARGV[3] = execution ID, ARGV[4] = '1' to replace a stored record of the - // same execution, ARGV[5] = gone set TTL in seconds + // same execution, ARGV[5] = gone set TTL in seconds, + // ARGV[6] = packed meta (see packMeta) // Replies: 1 written, 0 already stored for this execution, 2 refused as // gone, 3 refused as held. addSandboxScript = redis.NewScript(isHeldLua + ` @@ -114,6 +143,7 @@ var ( redis.call('EXPIRE', KEYS[2], ARGV[5]) end redis.call('SET', KEYS[1], ARGV[1]) + redis.call('SET', KEYS[3], ARGV[6]) return 1 `) @@ -134,6 +164,7 @@ var ( // removes nothing. // KEYS[1] = sandbox key, KEYS[2] = gone set, both in the record's slot // KEYS[3] = the record's lock key, in the record's slot + // KEYS[4] = meta key, in the record's slot // ARGV[1] = sandbox ID // ARGV[2] = expected execution ID, or "" to remove any execution // ARGV[3] = gone set TTL in seconds @@ -164,6 +195,7 @@ var ( return {0, data} end redis.call('DEL', KEYS[1]) + redis.call('DEL', KEYS[4]) if ok and decoded['executionID'] then redis.call('SADD', KEYS[2], decoded['executionID']) redis.call('EXPIRE', KEYS[2], ARGV[3]) @@ -188,33 +220,51 @@ var ( // KEYS[2] = transition key // KEYS[3] = transition result key // KEYS[4] = the record's lock key, in the record's slot + // KEYS[5] = meta key, in the record's slot // ARGV[1] = new sandbox JSON data // ARGV[2] = transition ID (UUID) // ARGV[3] = transition key TTL in seconds // ARGV[4] = result key TTL in seconds // ARGV[5] = expected execution ID, or "" to write unconditionally // ARGV[6] = lock value (token and metadata) - startTransitionScript = redis.NewScript(isHeldLua + ` + // ARGV[7] = packed meta of the new record (see packMeta) + // + // The held check and execution-ID guard read the meta key (KEYS[5]) with a + // single GET instead of cjson.decode of the record JSON, so the hot path + // (the evictor's StartRemoving, calling this with ARGV[5] set under + // concurrent expiry) no longer parses JSON on Redis's event loop (#3602). + // Migration: a record written before the meta key existed has no meta; the + // script then falls back to decoding the record, exactly as before, so + // pre-deploy sandboxes stay correct until they are next written (which sets + // the meta key) or expire. + startTransitionScript = redis.NewScript(isHeldLua + parseMetaLua + ` if redis.call('GET', KEYS[4]) ~= ARGV[6] then return 3 end - local current = redis.call('GET', KEYS[1]) - local ok, decoded = false, nil - if current then - ok, decoded = pcall(cjson.decode, current) + local meta = redis.call('GET', KEYS[5]) + local stored, held + if meta then + stored, held = parse_meta(meta) + else + local current = redis.call('GET', KEYS[1]) + if current then + local ok, decoded = pcall(cjson.decode, current) + if ok then + stored = decoded['executionID'] + held = is_held(decoded) + end + end end - if ok and is_held(decoded) then + if held then return 2 end if ARGV[5] ~= '' then - if not current then - return 0 - end - if not ok or decoded['executionID'] ~= ARGV[5] then + if not stored or stored ~= ARGV[5] then return 0 end end redis.call('SET', KEYS[1], ARGV[1]) + redis.call('SET', KEYS[5], ARGV[7]) redis.call('SET', KEYS[2], ARGV[2], 'EX', ARGV[3]) redis.call('SET', KEYS[3], '', 'EX', ARGV[4]) return 1 diff --git a/packages/api/internal/sandbox/storage/redis/scripts_remove_test.go b/packages/api/internal/sandbox/storage/redis/scripts_remove_test.go index 319cba565f..fb20e6fde8 100644 --- a/packages/api/internal/sandbox/storage/redis/scripts_remove_test.go +++ b/packages/api/internal/sandbox/storage/redis/scripts_remove_test.go @@ -109,7 +109,7 @@ func newRemoveScript(client redis.UniversalClient) removeScript { as := func(ctx context.Context, sandboxID, expected, holdID, lockValue string) *redis.Cmd { key := recordKey(sandboxID) - return RunScript(ctx, client, "remove_record", removeSandboxScript, []string{key, goneKeyFor(key), redis_utils.GetLockKey(key)}, sandboxID, expected, 60, holdID, lockValue) + return RunScript(ctx, client, "remove_record", removeSandboxScript, []string{key, goneKeyFor(key), redis_utils.GetLockKey(key), metaKeyFor(key)}, sandboxID, expected, 60, holdID, lockValue) } return removeScript{ diff --git a/packages/api/internal/sandbox/storage/redis/scripts_test.go b/packages/api/internal/sandbox/storage/redis/scripts_test.go index e91fcc72e3..8d07eb9b8c 100644 --- a/packages/api/internal/sandbox/storage/redis/scripts_test.go +++ b/packages/api/internal/sandbox/storage/redis/scripts_test.go @@ -31,15 +31,15 @@ func TestRunScript_RecordsCallsAndDurationByScript(t *testing.T) { key := recordKey("sbx") goneKey := goneKeyFor(key) - require.NoError(t, metrics.run(t.Context(), client, "add_sandbox", addSandboxScript, []string{key, goneKey}, "{}", "sbx", "exec", "0", 60).Err()) + require.NoError(t, metrics.run(t.Context(), client, "add_sandbox", addSandboxScript, []string{key, goneKey, metaKeyFor(key)}, "{}", "sbx", "exec", "0", 60, metaOf("exec")).Err()) lockRecord(t, client, key) - deleted, raw, _, err := parseRemoveReply(metrics.run(t.Context(), client, "remove_record", removeSandboxScript, []string{key, goneKey, redis_utils.GetLockKey(key)}, "sbx", "", 60, "", testLockValue)) + deleted, raw, _, err := parseRemoveReply(metrics.run(t.Context(), client, "remove_record", removeSandboxScript, []string{key, goneKey, redis_utils.GetLockKey(key), metaKeyFor(key)}, "sbx", "", 60, "", testLockValue)) require.NoError(t, err) require.True(t, deleted) require.Equal(t, "{}", raw) - err = metrics.run(t.Context(), client, "remove_record", removeSandboxScript, []string{key, goneKey, redis_utils.GetLockKey(key)}, "sbx", "", 60, "", testLockValue).Err() + err = metrics.run(t.Context(), client, "remove_record", removeSandboxScript, []string{key, goneKey, redis_utils.GetLockKey(key), metaKeyFor(key)}, "sbx", "", 60, "", testLockValue).Err() require.ErrorIs(t, err, redis.Nil) failing := redis.NewScript(`return redis.call('NOSUCHCOMMAND')`) diff --git a/packages/api/internal/sandbox/storage/redis/state_change.go b/packages/api/internal/sandbox/storage/redis/state_change.go index 49e6381052..0615200a3e 100644 --- a/packages/api/internal/sandbox/storage/redis/state_change.go +++ b/packages/api/internal/sandbox/storage/redis/state_change.go @@ -158,8 +158,8 @@ func (s *Storage) StartRemoving(ctx context.Context, teamID uuid.UUID, sandboxID resultTtlSeconds := int(transitionResultKeyTTL.Seconds()) written, err := RunScript(ctx, s.redisClient, "start_transition", startTransitionScript, - []string{key, transitionKey, resultKey, lock.Key()}, - newData, transitionID, ttlSeconds, resultTtlSeconds, opts.ExpectExecutionID, lockValueOf(lock), + []string{key, transitionKey, resultKey, lock.Key(), metaKeyFor(key)}, + newData, transitionID, ttlSeconds, resultTtlSeconds, opts.ExpectExecutionID, lockValueOf(lock), packMeta(updated), ).Int64() if err != nil { return transition, false, nil, fmt.Errorf("failed to update sandbox state: %w", err) @@ -355,7 +355,7 @@ func (s *Storage) compareAndRestore(ctx context.Context, key string, transition if err != nil { return sandboxtypes.Sandbox{}, err } - written, err := RunScript(ctx, s.redisClient, "restore_sandbox", restoreSandboxScript, []string{key}, data, updated).Int64() + written, err := RunScript(ctx, s.redisClient, "restore_sandbox", restoreSandboxScript, []string{key, metaKeyFor(key)}, data, updated, packMeta(restored)).Int64() if err != nil { return sandboxtypes.Sandbox{}, fmt.Errorf("failed to restore sandbox: %w", err) } diff --git a/packages/api/internal/sandbox/storage/redis/update_write_retry_test.go b/packages/api/internal/sandbox/storage/redis/update_write_retry_test.go index fab4c4cba9..7c8fb2d47e 100644 --- a/packages/api/internal/sandbox/storage/redis/update_write_retry_test.go +++ b/packages/api/internal/sandbox/storage/redis/update_write_retry_test.go @@ -44,7 +44,7 @@ func isFencedWriteOf(cmd redis.Cmder, key string) bool { if (name != "evalsha" && name != "eval") || len(args) < 5 { return false } - if n, ok := args[2].(int); !ok || n != 2 { + if n, ok := args[2].(int); !ok || n != 3 { return false } k, ok := args[4].(string) @@ -144,7 +144,7 @@ func TestFencedSetScript(t *testing.T) { require.NoError(t, client.Set(ctx, key, "v1", time.Hour).Err()) require.NoError(t, client.Set(ctx, lockKey, "token-a", time.Minute).Err()) - written, err := fencedSetScript.Run(ctx, client, []string{lockKey, key}, "token-a", "v2").Int64() + written, err := fencedSetScript.Run(ctx, client, []string{lockKey, key, metaKeyFor(key)}, "token-a", "v2", metaOf("e2")).Int64() require.NoError(t, err) assert.Equal(t, int64(1), written) value, err := client.Get(ctx, key).Result() @@ -154,7 +154,7 @@ func TestFencedSetScript(t *testing.T) { require.NoError(t, err) assert.Greater(t, ttl, 50*time.Minute, "KEEPTTL keeps the record's expiry") - written, err = fencedSetScript.Run(ctx, client, []string{lockKey, key}, "token-b", "v3").Int64() + written, err = fencedSetScript.Run(ctx, client, []string{lockKey, key, metaKeyFor(key)}, "token-b", "v3", metaOf("e3")).Int64() require.NoError(t, err) assert.Equal(t, int64(0), written, "another holder's token writes nothing") value, err = client.Get(ctx, key).Result() @@ -162,7 +162,7 @@ func TestFencedSetScript(t *testing.T) { assert.Equal(t, "v2", value) require.NoError(t, client.Del(ctx, lockKey).Err()) - written, err = fencedSetScript.Run(ctx, client, []string{lockKey, key}, "token-a", "v4").Int64() + written, err = fencedSetScript.Run(ctx, client, []string{lockKey, key, metaKeyFor(key)}, "token-a", "v4", metaOf("e4")).Int64() require.NoError(t, err) assert.Equal(t, int64(0), written, "an expired lock writes nothing") } diff --git a/packages/api/internal/sandbox/storage/redis/utils.go b/packages/api/internal/sandbox/storage/redis/utils.go index 233b0beae5..dffd91122d 100644 --- a/packages/api/internal/sandbox/storage/redis/utils.go +++ b/packages/api/internal/sandbox/storage/redis/utils.go @@ -17,8 +17,13 @@ const ( notifySuffix = "notify" indexKey = "index" recordSuffix = "record" + metaSuffix = "meta" ) +// metaFieldSep separates the packed meta fields. US (0x1f) cannot occur in a +// UUID execution ID or in a lifecycle-hold token, so a plain split is safe. +const metaFieldSep = "\x1f" + var ( // globalStorageNotifyChannel is the single Redis PubSub channel used by // storage notifications. The per-event routing key is embedded in the message @@ -83,6 +88,28 @@ func goneKeyFor(recordKey string) string { return redis_utils.CreateKey(recordKey, "gone") } +// metaKeyFor names the meta key beside a record key, in the record's slot. It +// holds packMeta's string — the execution ID and lifecycle hold of the record +// — so the hot path can read them with a single GET instead of decoding the +// record JSON on Redis's event loop (see packMeta and startTransitionScript). +func metaKeyFor(recordKey string) string { + return redis_utils.CreateKey(recordKey, metaSuffix) +} + +// packMeta serializes the fields the hot path needs from a record — its +// execution ID and lifecycle hold — into the meta key's value, so they can be +// read without decoding the record JSON. Layout: executionID, lifecycleHold, +// lifecycleHoldUntil (Unix ms), US-separated. Kept in sync with the record by +// every script that writes KEYS[1] (each writes the meta key in the same +// atomic script); removeSandboxScript deletes it with the record. +func packMeta(sbx sandboxtypes.Sandbox) string { + return strings.Join([]string{ + sbx.ExecutionID, + sbx.LifecycleHold, + fmt.Sprintf("%d", sbx.LifecycleHoldUntil), + }, metaFieldSep) +} + // GetSandboxStorageTeamIndexKey returns the storage team index key for external packages (e.g. reservations). func GetSandboxStorageTeamIndexKey(teamID string) string { return redis_utils.CreateKey(GetTeamPrefix(teamID), indexKey)