From 5ef6f99344e8fb550e9a53badb806b395a08861c Mon Sep 17 00:00:00 2001 From: QuangNguyen2609 Date: Sun, 14 Jun 2026 19:58:30 +0930 Subject: [PATCH 1/2] fix(lock): add jitter + tunable retry to distributed acquire to reduce writer starvation (#141) --- .changeset/fix-f9d-acquire-jitter.md | 15 ++ CLAUDE.md | 1 + src/api/distributed-lock.ts | 17 +- src/api/distributed-rw-lock.ts | 11 +- src/api/server.ts | 2 + .../unit/distributed-acquire-jitter.test.ts | 171 ++++++++++++++++++ .../plans/2026-06-13_f9d-acquire-jitter.md | 64 +++++++ 7 files changed, 276 insertions(+), 5 deletions(-) create mode 100644 .changeset/fix-f9d-acquire-jitter.md create mode 100644 src/api/tests/unit/distributed-acquire-jitter.test.ts create mode 100644 thoughts/shared/plans/2026-06-13_f9d-acquire-jitter.md diff --git a/.changeset/fix-f9d-acquire-jitter.md b/.changeset/fix-f9d-acquire-jitter.md new file mode 100644 index 0000000..d42b652 --- /dev/null +++ b/.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. diff --git a/CLAUDE.md b/CLAUDE.md index d964c0e..5a6c693 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -211,6 +211,7 @@ const TABLE = Object.assign(Object.create(null) as Record, { | `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. | diff --git a/src/api/distributed-lock.ts b/src/api/distributed-lock.ts index d4abcfa..e5b8a77 100644 --- a/src/api/distributed-lock.ts +++ b/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( } 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 ── diff --git a/src/api/distributed-rw-lock.ts b/src/api/distributed-rw-lock.ts index 9056d11..593043e 100644 --- a/src/api/distributed-rw-lock.ts +++ b/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)); } } diff --git a/src/api/server.ts b/src/api/server.ts index e0f2e52..95c0a9e 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -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: parseNonNegativeInt("REDIS_EXEC_LOCK_ACQUIRE_RETRY_MS", 50), readerLeaseMs: parseNonNegativeInt("REDIS_RWLOCK_READER_LEASE_MS", 60_000), } : undefined; diff --git a/src/api/tests/unit/distributed-acquire-jitter.test.ts b/src/api/tests/unit/distributed-acquire-jitter.test.ts new file mode 100644 index 0000000..007dded --- /dev/null +++ b/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(); + /** 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 { + 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((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); + }); +}); diff --git a/thoughts/shared/plans/2026-06-13_f9d-acquire-jitter.md b/thoughts/shared/plans/2026-06-13_f9d-acquire-jitter.md new file mode 100644 index 0000000..343887c --- /dev/null +++ b/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`) 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. From a8bccb309474c8455640640d40aef065e5c4bee2 Mon Sep 17 00:00:00 2001 From: QuangNguyen2609 Date: Sat, 20 Jun 2026 16:24:55 +0930 Subject: [PATCH 2/2] fix(lock): reject acquireRetryMs=0 at startup instead of on first lock use parseNonNegativeInt accepted 0 for REDIS_EXEC_LOCK_ACQUIRE_RETRY_MS, but both distributed lock validators require > 0. Add parsePositiveInt and use it so bad config fails clearly at boot. --- src/api/server.ts | 4 ++-- src/redis/config.ts | 10 ++++++++++ src/redis/tests/unit/config.test.ts | 28 +++++++++++++++++++++++++++- 3 files changed, 39 insertions(+), 3 deletions(-) diff --git a/src/api/server.ts b/src/api/server.ts index 95c0a9e..df60984 100644 --- a/src/api/server.ts +++ b/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"; @@ -45,7 +45,7 @@ const execLockOptions = redisClient 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: parseNonNegativeInt("REDIS_EXEC_LOCK_ACQUIRE_RETRY_MS", 50), + acquireRetryMs: parsePositiveInt("REDIS_EXEC_LOCK_ACQUIRE_RETRY_MS", 50), readerLeaseMs: parseNonNegativeInt("REDIS_RWLOCK_READER_LEASE_MS", 60_000), } : undefined; diff --git a/src/redis/config.ts b/src/redis/config.ts index d9eeaa3..ca29a84 100644 --- a/src/redis/config.ts +++ b/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; +} diff --git a/src/redis/tests/unit/config.test.ts b/src/redis/tests/unit/config.test.ts index 05d8f2e..f41180d 100644 --- a/src/redis/tests/unit/config.test.ts +++ b/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/); + }); +});