Merge pull request #15 from Apex-Fusion/fix/gateway-mutex-deadline

fix(gateway): per-run deadline + queue bound on Mutex (08-12 freeze fix)

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<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 {
   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,
     );
-    return result;
+    return settled;
   }
 }
 

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<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");
+  });
+});