Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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")
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion packages/api/internal/sandbox/storage/redis/heal.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Expand Down
12 changes: 6 additions & 6 deletions packages/api/internal/sandbox/storage/redis/operations.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
})
Expand Down Expand Up @@ -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
})
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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 {
Expand All @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
78 changes: 64 additions & 14 deletions packages/api/internal/sandbox/storage/redis/scripts.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
`)

Expand All @@ -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 + `
Expand Down Expand Up @@ -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
`)

Expand All @@ -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
Expand Down Expand Up @@ -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])
Expand All @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand Down
6 changes: 3 additions & 3 deletions packages/api/internal/sandbox/storage/redis/scripts_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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')`)
Expand Down
6 changes: 3 additions & 3 deletions packages/api/internal/sandbox/storage/redis/state_change.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
}
Expand Down
Loading