fix(gateway): per-run deadline + queue bound on Mutex - 08-12 freeze fix
Some checks failed
CI / typecheck + vitest (Node 20) (push) Has been cancelled
Some checks failed
CI / typecheck + vitest (Node 20) (push) Has been cancelled
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 <noreply@anthropic.com>
This commit is contained in:
@@ -15,19 +15,88 @@ import { open as unseal } from "../crypto/seal.js";
|
|||||||
import { deriveWalletKey } from "../wallet.js";
|
import { deriveWalletKey } from "../wallet.js";
|
||||||
import type { ApiKeyRow } from "../db/store.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<T>(fn: () => Promise<T>, timeoutMs: number, label: string): Promise<T> {
|
||||||
|
return new Promise<T>((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 {
|
export class Mutex {
|
||||||
private tail: Promise<unknown> = Promise.resolve();
|
private tail: Promise<unknown> = Promise.resolve();
|
||||||
|
private depth = 0;
|
||||||
|
|
||||||
run<T>(fn: () => Promise<T>): Promise<T> {
|
constructor(private readonly opts: { timeoutMs?: number; maxQueue?: number } = {}) {}
|
||||||
const result = this.tail.then(fn, fn);
|
|
||||||
// Keep the chain alive even if fn rejects; swallow here so the next
|
run<T>(fn: () => Promise<T>, label = "op"): Promise<T> {
|
||||||
// run() isn't poisoned by a prior rejection.
|
const timeoutMs = this.opts.timeoutMs ?? DEFAULT_MUTEX_TIMEOUT_MS;
|
||||||
this.tail = result.then(
|
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,
|
||||||
() => undefined,
|
() => undefined,
|
||||||
);
|
);
|
||||||
return result;
|
return settled;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
121
tests/unit/gateway-mutex.test.ts
Normal file
121
tests/unit/gateway-mutex.test.ts
Normal file
@@ -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<never>(() => {}), "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<never>(() => {}), "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<void>((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");
|
||||||
|
});
|
||||||
|
});
|
||||||
Reference in New Issue
Block a user