From b3b7d6b5e1ff162fc2b40e3d99b4f33edbe75ae4 Mon Sep 17 00:00:00 2001 From: FilipBl4gojevic <84703395+FilipBl4gojevic@users.noreply.github.com> Date: Sun, 16 Aug 2026 09:56:41 +0200 Subject: [PATCH] fix(gateway): per-run deadline + queue bound on Mutex - 08-12 freeze fix One chain call that never settled, run under the per-key Mutex with no deadline, convoyed every subsequent request on that key silently while /models and /account stayed healthy (2026-08-12 incident). The per-fetch 30s aborts in the Ogmios providers predate the freeze and did not prevent it; the unbounded thing was the composition under the mutex. Mutex.run() now races each fn against a deadline (default 180s): on expiry the run rejects with MutexTimeoutError and the chain advances, so queued work proceeds instead of freezing. A zombie that settles later can at worst lose a UTxO-contention race with the run that replaced it - strictly better than a permanent convoy. New work is refused with MutexQueueFullError past a queue bound (default 64) instead of queueing without limit. Serialization and rejection-tolerance semantics are unchanged (covered by tests). Session locks in transcripts.ts construct the same class and inherit both protections. Co-Authored-By: Claude Fable 5 --- gateway/src/sdk/registry.ts | 83 +++++++++++++++++++-- tests/unit/gateway-mutex.test.ts | 121 +++++++++++++++++++++++++++++++ 2 files changed, 197 insertions(+), 7 deletions(-) create mode 100644 tests/unit/gateway-mutex.test.ts diff --git a/gateway/src/sdk/registry.ts b/gateway/src/sdk/registry.ts index e4cb172..ddb8765 100644 --- a/gateway/src/sdk/registry.ts +++ b/gateway/src/sdk/registry.ts @@ -15,19 +15,88 @@ import { open as unseal } from "../crypto/seal.js"; import { deriveWalletKey } from "../wallet.js"; import type { ApiKeyRow } from "../db/store.js"; -/** Tiny promise-chain mutex: run(fn) executes fns strictly one at a time. */ +/** Default per-run deadline. Chain reads and tx submits settle in seconds; + * anything past this is the 2026-08-12 freeze mode (a fetch that never + * settles), not a slow success. */ +export const DEFAULT_MUTEX_TIMEOUT_MS = 180_000; + +/** Default queue bound per mutex. A healthy key drains its queue in seconds; + * a queue this deep means the head is stuck and callers should fail fast. */ +export const DEFAULT_MUTEX_MAX_QUEUE = 64; + +/** A run() exceeded its deadline and was abandoned so queued work can proceed. */ +export class MutexTimeoutError extends Error { + constructor(label: string, timeoutMs: number) { + super(`mutex run '${label}' exceeded ${timeoutMs}ms and was abandoned; queued work proceeds`); + this.name = "MutexTimeoutError"; + } +} + +/** run() was refused because the queue is at its bound (head likely stuck). */ +export class MutexQueueFullError extends Error { + constructor(label: string, depth: number) { + super(`mutex queue full (${depth} pending) — refusing '${label}'; the head of the queue is likely stuck`); + this.name = "MutexQueueFullError"; + } +} + +function runWithDeadline(fn: () => Promise, timeoutMs: number, label: string): Promise { + return new Promise((resolvePromise, rejectPromise) => { + const timer = setTimeout(() => rejectPromise(new MutexTimeoutError(label, timeoutMs)), timeoutMs); + // Node-only nicety: don't hold the process open for a pending deadline. + timer.unref?.(); + fn().then( + (v) => { + clearTimeout(timer); + resolvePromise(v); + }, + (e) => { + clearTimeout(timer); + rejectPromise(e); + }, + ); + }); +} + +/** + * Tiny promise-chain mutex: run(fn) executes fns strictly one at a time — + * now with a per-run deadline and a queue bound (2026-08-12 freeze fix). + * + * On deadline the stuck fn is NOT cancelled (promises aren't); it is + * abandoned: its run() rejects with MutexTimeoutError and the chain advances. + * A zombie that settles later can at worst lose a UTxO-contention race with + * the run that replaced it — strictly better than the alternative, where one + * stuck chain call convoys every subsequent request on the key forever. + */ export class Mutex { private tail: Promise = Promise.resolve(); + private depth = 0; - run(fn: () => Promise): Promise { - const result = this.tail.then(fn, fn); - // Keep the chain alive even if fn rejects; swallow here so the next - // run() isn't poisoned by a prior rejection. - this.tail = result.then( + constructor(private readonly opts: { timeoutMs?: number; maxQueue?: number } = {}) {} + + run(fn: () => Promise, label = "op"): Promise { + const timeoutMs = this.opts.timeoutMs ?? DEFAULT_MUTEX_TIMEOUT_MS; + const maxQueue = this.opts.maxQueue ?? DEFAULT_MUTEX_MAX_QUEUE; + + if (this.depth >= maxQueue) { + return Promise.reject(new MutexQueueFullError(label, this.depth)); + } + this.depth++; + + // Keep the chain alive even if fn rejects or times out; swallow here so + // the next run() isn't poisoned by a prior rejection. + const result = this.tail.then( + () => runWithDeadline(fn, timeoutMs, label), + () => runWithDeadline(fn, timeoutMs, label), + ); + const settled = result.finally(() => { + this.depth--; + }); + this.tail = settled.then( () => undefined, () => undefined, ); - return result; + return settled; } } diff --git a/tests/unit/gateway-mutex.test.ts b/tests/unit/gateway-mutex.test.ts new file mode 100644 index 0000000..4599d84 --- /dev/null +++ b/tests/unit/gateway-mutex.test.ts @@ -0,0 +1,121 @@ +/** + * gateway-mutex.test.ts — Mutex deadline + queue bound (2026-08-12 freeze fix). + * + * Incident: one chain call that never settled, run under a per-key Mutex with + * no deadline, convoyed every subsequent request on that key silently while + * /models and /account stayed healthy. The fix gives run() a deadline that + * abandons the stuck operation (advancing the chain) and a queue bound that + * rejects new work instead of queueing without limit. + * + * Serialization semantics of the original class must be preserved. + */ + +import { describe, it, expect, vi, afterEach } from "vitest"; +import { + Mutex, + MutexTimeoutError, + MutexQueueFullError, +} from "../../gateway/src/sdk/registry.js"; + +afterEach(() => { + vi.useRealTimers(); +}); + +// ─── Preserved semantics ───────────────────────────────────────────────────── + +describe("Mutex — serialization (preserved semantics)", () => { + it("runs fns strictly one at a time, in order", async () => { + const mutex = new Mutex(); + const order: number[] = []; + let firstRunning = false; + + const first = mutex.run(async () => { + firstRunning = true; + await new Promise((r) => setTimeout(r, 20)); + order.push(1); + firstRunning = false; + }); + const second = mutex.run(async () => { + expect(firstRunning).toBe(false); + order.push(2); + }); + + await Promise.all([first, second]); + expect(order).toEqual([1, 2]); + }); + + it("a rejecting fn does not poison the chain", async () => { + const mutex = new Mutex(); + + await expect(mutex.run(async () => Promise.reject(new Error("boom")))).rejects.toThrow("boom"); + await expect(mutex.run(async () => "alive")).resolves.toBe("alive"); + }); +}); + +// ─── Deadline (the freeze fix) ─────────────────────────────────────────────── + +describe("Mutex — per-run deadline", () => { + it("rejects a stuck run with MutexTimeoutError once the deadline passes", async () => { + vi.useFakeTimers(); + const mutex = new Mutex({ timeoutMs: 1_000 }); + + const stuck = mutex.run(() => new Promise(() => {}), "stuck-chain-read"); + // Attach the rejection expectation BEFORE advancing time so the rejection + // is not treated as unhandled. + const assertion = expect(stuck).rejects.toBeInstanceOf(MutexTimeoutError); + + await vi.advanceTimersByTimeAsync(1_001); + await assertion; + }); + + it("queued work proceeds after a stuck run is abandoned (no convoy)", async () => { + vi.useFakeTimers(); + const mutex = new Mutex({ timeoutMs: 1_000 }); + + const stuck = mutex.run(() => new Promise(() => {}), "stuck"); + const stuckAssertion = expect(stuck).rejects.toBeInstanceOf(MutexTimeoutError); + const queued = mutex.run(async () => "unblocked"); + + await vi.advanceTimersByTimeAsync(1_001); + await stuckAssertion; + await expect(queued).resolves.toBe("unblocked"); + }); + + it("a run that finishes in time is unaffected by the deadline", async () => { + vi.useFakeTimers(); + const mutex = new Mutex({ timeoutMs: 1_000 }); + + const quick = mutex.run(async () => "done"); + await vi.advanceTimersByTimeAsync(1); + await expect(quick).resolves.toBe("done"); + }); +}); + +// ─── Queue bound ───────────────────────────────────────────────────────────── + +describe("Mutex — queue bound", () => { + it("rejects immediately with MutexQueueFullError when the queue is full", async () => { + const mutex = new Mutex({ maxQueue: 2 }); + let releaseFirst!: () => void; + const gate = new Promise((r) => { + releaseFirst = r; + }); + + const first = mutex.run(() => gate.then(() => "first")); + const second = mutex.run(async () => "second"); + const third = mutex.run(async () => "third"); + + await expect(third).rejects.toBeInstanceOf(MutexQueueFullError); + + releaseFirst(); + await expect(first).resolves.toBe("first"); + await expect(second).resolves.toBe("second"); + }); + + it("slots free up once runs settle", async () => { + const mutex = new Mutex({ maxQueue: 1 }); + + await mutex.run(async () => "a"); + await expect(mutex.run(async () => "b")).resolves.toBe("b"); + }); +});