fix(lock): jitter + tunable retry in distributed acquire (F9d)

.changeset/fix-f9d-acquire-jitter.md

@@ -0,0 +1,15 @@
+---
+"sql-fs-api": minor
+---
+
+fix(lock): add bounded jitter + tunable retry to the distributed acquire loops (F9d, #141)
+
+The distributed exec lock and RW lock polled Redis on a flat `acquireRetryMs`
+(default 50 ms) interval, leaving competing replicas phase-aligned so a
+cross-replica writer could be repeatedly passed over (bounded by
+`acquireTimeoutMs`, then 503). Every acquire/drain poll now sleeps a jittered
+`retryMs/2 + random()*retryMs/2` (range `[retryMs/2, retryMs]`) to
+de-synchronize pollers. The retry interval is now configurable via
+`REDIS_EXEC_LOCK_ACQUIRE_RETRY_MS` (previously hardcoded — `server.ts` omitted
+it). Circuit-breaker / error-budget behavior is unchanged. The FIFO ZSET ticket
+queue is deferred as a follow-up.

CLAUDE.md

@@ -211,6 +211,7 @@ const TABLE = Object.assign(Object.create(null) as Record<string, string>, {
 | `REDIS_EXEC_LOCK_LEASE_MS` | No (default: 60000) | Distributed exec lock lease duration (ms). Lock auto-expires if the holder dies. Must be > `REDIS_EXEC_LOCK_RENEW_MS`. |
 | `REDIS_EXEC_LOCK_RENEW_MS` | No (default: 20000) | Heartbeat interval for exec lock renewal (ms). Must be strictly less than `REDIS_EXEC_LOCK_LEASE_MS` to guarantee renewal fires before expiry. |
 | `REDIS_EXEC_LOCK_ACQUIRE_TIMEOUT_MS` | No (default: 300000) | Max time to wait to acquire the exec lock before returning 503 (ms). |
+| `REDIS_EXEC_LOCK_ACQUIRE_RETRY_MS` | No (default: 50) | Base poll interval (ms) for the distributed acquire/drain retry loops. Each sleep is jittered to `[retryMs/2, retryMs]` to de-synchronize competing replicas and reduce cross-replica writer starvation (F9d). Must be > 0. |
 | `REDIS_BLOB_CACHE_ENABLED` | No (default: true) | Set to `false` to disable the Redis blob cache. Blob reads always fall through to Postgres when disabled. |
 | `REDIS_BLOB_CACHE_TTL_MS` | No (default: 86400000) | TTL for blob cache entries (ms, default 24h). |
 | `REDIS_BLOB_MAX_BYTES` | No (default: 8388608) | Max blob size cached in Redis (bytes, default 8 MB). Blobs larger than this bypass Redis entirely. |

src/api/distributed-lock.ts

@@ -76,6 +76,20 @@ function assertLockOptions(opts: DistributedLockOptions): void {
 	}
 }
 
+/**
+ * F9d: bounded jitter for acquire-loop sleeps. Returns a delay in
+ * `[retryMs/2, retryMs]` (`retryMs/2 + random()*retryMs/2`). A flat `retryMs`
+ * keeps competing replicas phase-aligned, so a cross-replica writer can be
+ * repeatedly passed over by a peer that polls a hair earlier each cycle.
+ * Jittering each sleep de-synchronizes the pollers, spreading acquire chances
+ * fairly without a stateful ticket queue. The lower bound stays at `retryMs/2`
+ * so we never busy-poll Redis harder than ~2x the configured rate.
+ */
+export function jitteredDelayMs(retryMs: number): number {
+	const half = retryMs / 2;
+	return half + Math.random() * half;
+}
+
 export class LockAcquireTimeoutError extends Error {
 	readonly code = "ELOCKTIMEOUT";
 	constructor(key: string) {
@@ -145,7 +159,8 @@ export async function withDistributedLock<T>(
 		}
 		if (acquired) break;
 		if (Date.now() >= deadline) throw new LockAcquireTimeoutError(key);
-		await new Promise((r) => setTimeout(r, acquireRetryMs));
+		// F9d: jittered sleep to de-synchronize cross-replica pollers.
+		await new Promise((r) => setTimeout(r, jitteredDelayMs(acquireRetryMs)));
 	}
 
 	// ── Heartbeat ──

src/api/distributed-rw-lock.ts

@@ -18,7 +18,7 @@ import {
 	getRedisCircuitBreaker,
 } from "../redis/circuit-breaker.js";
 export { LockAcquireTimeoutError, LockLostError } from "./distributed-lock.js";
-import { LockAcquireTimeoutError, LockLostError } from "./distributed-lock.js";
+import { LockAcquireTimeoutError, LockLostError, jitteredDelayMs } from "./distributed-lock.js";
 
 // ── Lua scripts ──────────────────────────────────────────────────────────────
 
@@ -199,7 +199,8 @@ async function acquireShared(
 		}
 		if (acquired) return;
 		if (Date.now() >= deadline) throw new LockAcquireTimeoutError(keys.readers);
-		await sleep(acquireRetryMs);
+		// F9d: jittered sleep to de-synchronize cross-replica pollers.
+		await sleep(jitteredDelayMs(acquireRetryMs));
 	}
 }
 
@@ -288,7 +289,8 @@ async function acquireExclusive(
 		}
 		if (flagAcquired) break;
 		if (Date.now() >= deadline) throw new LockAcquireTimeoutError(keys.writer);
-		await sleep(acquireRetryMs);
+		// F9d: jittered sleep to de-synchronize cross-replica pollers.
+		await sleep(jitteredDelayMs(acquireRetryMs));
 	}
 }
 
@@ -324,7 +326,8 @@ async function waitReadersDrained(
 		if (count === -1) throw new LockLostError(keys.writer);
 		if (count === 0) return;
 		if (Date.now() >= deadline) throw new LockAcquireTimeoutError(keys.writer);
-		await sleep(acquireRetryMs);
+		// F9d: jittered sleep to de-synchronize cross-replica pollers.
+		await sleep(jitteredDelayMs(acquireRetryMs));
 	}
 }
 

src/api/server.ts

@@ -9,7 +9,7 @@ import { Hono } from "hono";
 import { bodyLimit } from "hono/body-limit";
 import type { ContentfulStatusCode } from "hono/utils/http-status";
 import { closeRedisClient, getRedisClient } from "../redis/client.js";
-import { parseNonNegativeInt } from "../redis/config.js";
+import { parseNonNegativeInt, parsePositiveInt } from "../redis/config.js";
 import { PostgresDialect } from "../sql-fs/dialects/postgres.js";
 import { translateSqlError } from "../sql-fs/errors.js";
 import { RedisBlobCache } from "../sql-fs/redis-blob-cache.js";
@@ -44,6 +44,8 @@ const execLockOptions = redisClient
 			leaseMs: parseNonNegativeInt("REDIS_EXEC_LOCK_LEASE_MS", 60_000),
 			renewMs: parseNonNegativeInt("REDIS_EXEC_LOCK_RENEW_MS", 20_000),
 			acquireTimeoutMs: parseNonNegativeInt("REDIS_EXEC_LOCK_ACQUIRE_TIMEOUT_MS", 300_000),
+			// F9d: tunable acquire poll interval (jittered to [retryMs/2, retryMs]).
+			acquireRetryMs: parsePositiveInt("REDIS_EXEC_LOCK_ACQUIRE_RETRY_MS", 50),
 			readerLeaseMs: parseNonNegativeInt("REDIS_RWLOCK_READER_LEASE_MS", 60_000),
 		}
 	: undefined;

src/api/tests/unit/distributed-acquire-jitter.test.ts

@@ -0,0 +1,171 @@
+/**
+ * F9d unit tests: bounded acquire jitter + tunable retry interval.
+ *
+ * Covers:
+ *  - `jitteredDelayMs` stays within `[retryMs/2, retryMs]` across many samples.
+ *  - A configured `acquireRetryMs` is honored: acquire still succeeds when the
+ *    lock frees, and still times out at `acquireTimeoutMs` when it does not.
+ *  - The circuit-breaker fast-fail path is unaffected by jitter.
+ */
+
+import type { Redis } from "ioredis";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { resetRedisCircuitBreakerForTest } from "../../../redis/circuit-breaker.js";
+import { LockAcquireTimeoutError, execLockKey, jitteredDelayMs, withDistributedLock } from "../../distributed-lock.js";
+
+interface Entry {
+	value: string;
+	expiresAt: number;
+}
+
+/** Minimal token-aware Redis fake (mirrors distributed-lock.test.ts). */
+class FakeRedis {
+	store = new Map<string, Entry>();
+	/** When true, every `set` (acquire) throws — simulates a connection-class error. */
+	failSet = false;
+
+	private gc(): void {
+		const now = Date.now();
+		for (const [k, e] of this.store) {
+			if (e.expiresAt <= now) this.store.delete(k);
+		}
+	}
+
+	async set(key: string, value: string, _px: "PX", ms: number, _nx: "NX"): Promise<"OK" | null> {
+		if (this.failSet) throw new Error("redis set failed");
+		this.gc();
+		if (this.store.has(key)) return null;
+		this.store.set(key, { value, expiresAt: Date.now() + ms });
+		return "OK";
+	}
+
+	async eval(script: string, _numKeys: number, key: string, token: string, ms?: string): Promise<number> {
+		this.gc();
+		if (script.includes("del")) {
+			const entry = this.store.get(key);
+			if (entry?.value === token) {
+				this.store.delete(key);
+				return 1;
+			}
+			return 0;
+		}
+		const entry = this.store.get(key);
+		if (entry?.value === token && ms !== undefined) {
+			entry.expiresAt = Date.now() + Number(ms);
+			return 1;
+		}
+		return 0;
+	}
+}
+
+function asRedis(f: FakeRedis): Redis {
+	return f as unknown as Redis;
+}
+
+const KEY = execLockKey("default", "sbx-jitter");
+
+describe("jitteredDelayMs", () => {
+	it("stays within [retryMs/2, retryMs] across many samples", () => {
+		const retryMs = 50;
+		for (let i = 0; i < 10_000; i++) {
+			const d = jitteredDelayMs(retryMs);
+			expect(d).toBeGreaterThanOrEqual(retryMs / 2);
+			expect(d).toBeLessThanOrEqual(retryMs);
+		}
+	});
+
+	it("hits the lower bound (retryMs/2) when random() returns 0", () => {
+		const spy = vi.spyOn(Math, "random").mockReturnValue(0);
+		expect(jitteredDelayMs(50)).toBe(25);
+		spy.mockRestore();
+	});
+
+	it("approaches the upper bound (retryMs) as random() → 1", () => {
+		const spy = vi.spyOn(Math, "random").mockReturnValue(0.999999);
+		expect(jitteredDelayMs(50)).toBeCloseTo(50, 2);
+		expect(jitteredDelayMs(50)).toBeLessThanOrEqual(50);
+		spy.mockRestore();
+	});
+
+	it("scales with retryMs (different base → different bounds)", () => {
+		const spy = vi.spyOn(Math, "random").mockReturnValue(0);
+		expect(jitteredDelayMs(200)).toBe(100);
+		spy.mockRestore();
+	});
+});
+
+describe("acquireRetryMs is honored under jitter", () => {
+	afterEach(() => {
+		vi.useRealTimers();
+		resetRedisCircuitBreakerForTest();
+	});
+
+	it("acquire still succeeds once the lock frees (jittered retries)", async () => {
+		const r = new FakeRedis();
+		let releaseFirst!: () => void;
+		const hold = new Promise<void>((resolve) => {
+			releaseFirst = resolve;
+		});
+
+		const first = withDistributedLock(asRedis(r), KEY, async () => hold, {
+			leaseMs: 5_000,
+			renewMs: 4_000,
+			acquireTimeoutMs: 5_000,
+			acquireRetryMs: 20,
+		});
+		await new Promise((res) => setTimeout(res, 20));
+
+		const second = withDistributedLock(asRedis(r), KEY, async () => "got-it", {
+			leaseMs: 5_000,
+			renewMs: 4_000,
+			acquireTimeoutMs: 5_000,
+			acquireRetryMs: 20,
+		});
+
+		// Let the second caller poll a few jittered cycles, then release the first.
+		await new Promise((res) => setTimeout(res, 80));
+		releaseFirst();
+		await first;
+		await expect(second).resolves.toBe("got-it");
+		expect(r.store.has(KEY)).toBe(false);
+	});
+
+	it("acquire still times out at acquireTimeoutMs when the lock never frees", async () => {
+		const r = new FakeRedis();
+		// Plant a foreign, long-lived lock so the caller can never acquire.
+		r.store.set(KEY, { value: "stranger", expiresAt: Date.now() + 60_000 });
+
+		const start = Date.now();
+		await expect(
+			withDistributedLock(asRedis(r), KEY, async () => "nope", {
+				leaseMs: 5_000,
+				renewMs: 4_000,
+				acquireTimeoutMs: 120,
+				acquireRetryMs: 20,
+			}),
+		).rejects.toBeInstanceOf(LockAcquireTimeoutError);
+		// Should not return materially before the deadline (jitter only reshapes
+		// the poll cadence, it does not shorten the overall timeout).
+		expect(Date.now() - start).toBeGreaterThanOrEqual(110);
+	});
+});
+
+describe("circuit-breaker path unaffected by jitter", () => {
+	afterEach(() => {
+		resetRedisCircuitBreakerForTest();
+	});
+
+	it("fast-fails on persistent thrown acquire errors via the error budget", async () => {
+		const r = new FakeRedis();
+		r.failSet = true; // every acquire attempt throws (connection-class error)
+		await expect(
+			withDistributedLock(asRedis(r), KEY, async () => "nope", {
+				leaseMs: 5_000,
+				renewMs: 4_000,
+				acquireTimeoutMs: 60_000, // huge — only the error budget should cut it short
+				acquireRetryMs: 5,
+				errorBudgetMs: 50,
+			}),
+		).rejects.toBeInstanceOf(LockAcquireTimeoutError);
+	});
+});

src/redis/config.ts

@@ -12,3 +12,13 @@ export function parseNonNegativeInt(envVar: string, defaultValue: number): numbe
 	}
 	return parsed;
 }
+
+export function parsePositiveInt(envVar: string, defaultValue: number): number {
+	const raw = process.env[envVar];
+	if (raw === undefined || raw === "") return defaultValue;
+	const parsed = Number(raw);
+	if (!Number.isInteger(parsed) || parsed <= 0) {
+		throw new Error(`Invalid value for ${envVar}: "${raw}" — expected a positive integer (got ${parsed}).`);
+	}
+	return parsed;
+}

src/redis/tests/unit/config.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, describe, expect, it } from "vitest";
-import { parseNonNegativeInt } from "../../config.js";
+import { parseNonNegativeInt, parsePositiveInt } from "../../config.js";
 
 const TEST_VAR = "VFS_TEST_ENV_VAR_DO_NOT_SET";
 
@@ -57,3 +57,29 @@ describe("parseNonNegativeInt", () => {
 		expect(() => parseNonNegativeInt(TEST_VAR, 0)).toThrow(/expected a non-negative integer/);
 	});
 });
+
+describe("parsePositiveInt", () => {
+	it("returns the default when the env var is unset", () => {
+		expect(parsePositiveInt(TEST_VAR, 50)).toBe(50);
+	});
+
+	it("parses a valid positive integer", () => {
+		process.env[TEST_VAR] = "42";
+		expect(parsePositiveInt(TEST_VAR, 1)).toBe(42);
+	});
+
+	it("rejects zero", () => {
+		process.env[TEST_VAR] = "0";
+		expect(() => parsePositiveInt(TEST_VAR, 1)).toThrow(/expected a positive integer/);
+	});
+
+	it("rejects negative numbers", () => {
+		process.env[TEST_VAR] = "-1";
+		expect(() => parsePositiveInt(TEST_VAR, 1)).toThrow(/expected a positive integer/);
+	});
+
+	it("rejects non-numeric input", () => {
+		process.env[TEST_VAR] = "abc";
+		expect(() => parsePositiveInt(TEST_VAR, 1)).toThrow(/Invalid value for VFS_TEST_ENV_VAR_DO_NOT_SET/);
+	});
+});

thoughts/shared/plans/2026-06-13_f9d-acquire-jitter.md

@@ -0,0 +1,64 @@
+# F9d — Distributed acquire fairness: jitter + tunable retry
+
+## Context
+
+GitHub issue #141 (F9d, severity:low). Distributed acquire loops poll Redis on a
+flat `acquireRetryMs` (default 50 ms) `setTimeout`/`sleep`. Across replicas the
+sleeps stay phase-aligned, so a cross-replica writer can be repeatedly passed
+over by another replica that happens to poll a hair earlier each cycle — bounded
+by `acquireTimeoutMs` (300 s) then a 503, never infinite.
+
+This lands on top of the just-merged F5 circuit breaker (#134), which sits in the
+same acquire loops. Jitter/retry must not regress breaker behavior.
+
+### Current state (verified by symbol)
+- `src/api/distributed-lock.ts` — legacy single-key acquire loop: `await new Promise((r) => setTimeout(r, acquireRetryMs))` at the bottom of the `while (true)` acquire loop. `acquireRetryMs` already on `DistributedLockOptions` (default 50, asserted `> 0`).
+- `src/api/distributed-rw-lock.ts` — three poll loops use `await sleep(acquireRetryMs)`: `acquireShared`, `acquireExclusive` (Phase A flag), `waitReadersDrained` (drain). `acquireRetryMs` already on `DistributedRWLockOptions` (default 50, asserted `> 0`). A module-local `sleep()` helper exists.
+- `src/api/server.ts:42-49` — builds `execLockOptions` (`Partial<DistributedRWLockOptions>`) from env but **omits `acquireRetryMs`**, so 50 ms is effectively hardcoded in prod. Threaded straight into `withDistributedLock`/`withDistributedRWLock` via `session-manager.ts`.
+- `parseNonNegativeInt` (`src/redis/config.ts`) is the env parser used for the other lock knobs. Consistent to reuse (0 is caught lazily by the lock's `assert*Options`, same as `leaseMs`/`renewMs`).
+
+## Desired end state
+
+- Every acquire/drain poll sleeps a **bounded jittered** interval `retryMs/2 + random()*retryMs/2` (range `[retryMs/2, retryMs]`) instead of a flat `retryMs`, de-synchronizing competing replicas.
+- `acquireRetryMs` is configurable via `REDIS_EXEC_LOCK_ACQUIRE_RETRY_MS` (wired through `server.ts`) and documented in CLAUDE.md.
+- Circuit-breaker / error-budget behavior unchanged.
+- ZSET ticket queue is explicitly **deferred** as a follow-up (only needed if multi-writer-per-sandbox becomes real).
+
+## What we are NOT doing
+- No ZSET/LIST FIFO ticket queue (deferred; would need TTL-reaping to stay crash-safe).
+- No change to lease/renew/heartbeat logic, breaker, or error budget.
+- No change to acquire success/timeout contract (`LockAcquireTimeoutError` → 503).
+
+## Implementation
+
+### Phase 1 — jitter helper + apply in both lock files
+- Add `jitteredDelayMs(retryMs)` to `distributed-lock.ts`, exported, returning `retryMs / 2 + Math.random() * (retryMs / 2)`. Use it in the legacy acquire loop's sleep.
+- Import it into `distributed-rw-lock.ts`; replace the flat `sleep(acquireRetryMs)` in `acquireShared`, `acquireExclusive`, `waitReadersDrained` with `sleep(jitteredDelayMs(acquireRetryMs))`. Leave the renew-retry timers (heartbeat) untouched — those are not the contention loops F9d targets.
+
+**Automated criteria**
+- [ ] `pnpm typecheck` passes.
+- [ ] `pnpm lint:fix` clean.
+- [ ] `pnpm test -- distributed-lock distributed-rw-lock` green.
+
+**Manual criteria**
+- [ ] Jitter only appears in acquire/drain loops, not heartbeat renewal.
+
+### Phase 2 — wire `REDIS_EXEC_LOCK_ACQUIRE_RETRY_MS` through server.ts
+- Add `acquireRetryMs: parseNonNegativeInt("REDIS_EXEC_LOCK_ACQUIRE_RETRY_MS", 50)` to `execLockOptions`.
+- Document the env var in CLAUDE.md env table.
+
+**Automated criteria**
+- [ ] `pnpm typecheck` passes.
+
+**Manual criteria**
+- [ ] Env var documented; default 50 matches the lock default.
+
+### Phase 3 — regression tests
+- New `distributed-acquire-jitter.test.ts`: assert `jitteredDelayMs(retryMs)` stays in `[retryMs/2, retryMs]` across many samples; assert configured `acquireRetryMs` is honored (acquire still succeeds; timeout still fires at `acquireTimeoutMs`); breaker path unaffected.
+
+**Automated criteria**
+- [ ] `pnpm test:unit` green, existing tests unchanged.
+
+## Discoveries
+- `acquireRetryMs` was already a first-class option on both lock types and threaded end-to-end via `session-manager.ts`; only the `server.ts` env wiring was missing — matches the issue's "already threaded" note.
+- Jitter narrows-or-equals the flat interval (max = `retryMs`), so existing tests with generous `acquireTimeoutMs` remain valid.