Recover Counterparty read slots after cancelled worker runs

apps/api/src/integrations/cooldown.ts

@@ -3,34 +3,54 @@
  * it does not claim to coordinate every Worker in the account. */
 let notBefore = 0;
 const DEFAULT_COOLDOWN_MS = 30_000;
+export const COUNTERPARTY_READ_TIMEOUT_MS = 15_000;
 const READ_LIMIT = 2;
-let active = 0;
+const active = new Map<() => void, number>();
 const queued: Array<{ admit: () => void; reject: (error: unknown) => void }> = [];
 
 function drain(): void {
+  // Workers can cancel an invocation without running its timers or finally
+  // blocks. A later invocation must not wait forever on its abandoned slots.
+  // Reap by wall time as well as listening for aborts in a live invocation.
+  const now = Date.now();
+  let expired = 0;
+  for (const [cleanup, deadline] of active) {
+    if (deadline <= now) {
+      active.delete(cleanup);
+      cleanup();
+      expired++;
+    }
+  }
+  if (expired > 0) console.warn({ event: "counterparty_read_slots_expired", count: expired });
   if (Date.now() < notBefore) {
     for (const waiting of queued.splice(0)) waiting.reject(new CounterpartyReadDeferred(notBefore));
     return;
   }
-  while (active < READ_LIMIT && queued.length > 0) queued.shift()!.admit();
-}
-
-function release(): void {
-  active--;
-  drain();
+  while (active.size < READ_LIMIT && queued.length > 0) queued.shift()!.admit();
 }
 
 /** The existing read timeout includes admission, so a queue cannot add a
  * second timeout window to work running under the indexer's lease. Only
  * permits are shared; responses and fetch promises belong to their caller. */
-export function acquireCounterpartyRead(signal: AbortSignal): Promise<() => void> {
+export function acquireCounterpartyRead(signal: AbortSignal, deadline: number): Promise<() => void> {
   assertCounterpartyReadAllowed();
   signal.throwIfAborted();
   return new Promise((resolve, reject) => {
     const waiting = {
       admit() {
         signal.removeEventListener("abort", abort);
-        active++;
+        if (signal.aborted || deadline <= Date.now()) {
+          reject(signal.reason ?? new DOMException("Counterparty read deadline expired", "TimeoutError"));
+          return;
+        }
+        const cleanup = () => signal.removeEventListener("abort", release);
+        const release = () => {
+          cleanup();
+          // A late finally must not release a replacement read's slot.
+          if (active.delete(cleanup)) drain();
+        };
+        active.set(cleanup, deadline);
+        signal.addEventListener("abort", release, { once: true });
         resolve(release);
       },
       reject(error: unknown) {

apps/api/src/integrations/counterparty.ts

@@ -10,6 +10,7 @@ import type { CounterpartyEvent } from "@launchpad/xcp69/trades";
 import {
   acquireCounterpartyRead,
   assertCounterpartyReadAllowed,
+  COUNTERPARTY_READ_TIMEOUT_MS,
   CounterpartyReadDeferred,
   recordCounterpartyCooldown,
 } from "#api/integrations/cooldown";
@@ -45,14 +46,20 @@ export function counterpartyApiBase(): string {
 }
 
 async function get<T>(path: string): Promise<T> {
-  const signal = AbortSignal.timeout(15_000);
-  const release = await acquireCounterpartyRead(signal);
+  const startedAt = Date.now();
+  const deadline = startedAt + COUNTERPARTY_READ_TIMEOUT_MS;
+  const signal = AbortSignal.timeout(COUNTERPARTY_READ_TIMEOUT_MS);
+  let release: (() => void) | undefined;
+  let stage = "admission";
   try {
+    release = await acquireCounterpartyRead(signal, deadline);
     // A concurrent response may establish a cooldown after this read joined
     // the queue. Check again immediately before starting its own request.
     assertCounterpartyReadAllowed();
     signal.throwIfAborted();
+    stage = "fetch";
     const res = await fetch(freshCounterpartyUrl(`${base}${path}`), { signal });
+    stage = "body";
     if (!res.ok) {
       if (res.status === 429) recordCounterpartyCooldown(res.headers.get("retry-after"));
       // A Worker holds six outbound connections, and a Response whose body is
@@ -68,8 +75,14 @@ async function get<T>(path: string): Promise<T> {
     // hard cap sits above. Parse losslessly so unsafe magnitudes survive as
     // strings instead of silently drifting, the same way the web app does.
     return parseJsonLossless<T>(await res.text());
+  } catch (error) {
+    if (signal.aborted) console.warn({
+      event: "counterparty_read_timeout", path: path.split("?")[0], stage,
+      duration_ms: Date.now() - startedAt,
+    });
+    throw error;
   } finally {
-    release();
+    release?.();
   }
 }
 

tests/api-counterparty-cooldown.test.ts

@@ -165,4 +165,38 @@ describe("the API refuses reads during Counterparty's cooldown without waiting",
     expect(timeout.mock.calls).toEqual([[15_000], [15_000], [15_000]]);
     timeout.mockRestore();
   });
+
+  it("reclaims abandoned slots without letting late cleanup release their replacements", async () => {
+    const { acquireCounterpartyRead } = await import("#api/integrations/cooldown");
+    const abandoned = await Promise.all(Array.from({ length: 2 }, () =>
+      acquireCounterpartyRead(new AbortController().signal, NOW + 1_000)));
+    // An invocation ended: neither its abort timer nor its finally will run.
+    vi.setSystemTime(NOW + 1_001);
+    const replacements = await Promise.all(Array.from({ length: 2 }, () =>
+      acquireCounterpartyRead(new AbortController().signal, NOW + 2_000)));
+    let admitted = false;
+    const next = acquireCounterpartyRead(new AbortController().signal, NOW + 2_000)
+      .then(release => { admitted = true; return release; });
+    abandoned.forEach(release => release());
+    await Promise.resolve();
+    expect(admitted).toBe(false);
+    replacements[0]!();
+    const release = await next;
+    expect(admitted).toBe(true);
+    release();
+    replacements[1]!();
+  });
+
+  it("releases an admitted read on abort even if its body never settles", async () => {
+    const { acquireCounterpartyRead } = await import("#api/integrations/cooldown");
+    const controller = new AbortController();
+    const first = await acquireCounterpartyRead(controller.signal, NOW + 1_000);
+    const second = await acquireCounterpartyRead(new AbortController().signal, NOW + 1_000);
+    const next = acquireCounterpartyRead(new AbortController().signal, NOW + 1_000);
+    controller.abort();
+    const release = await next;
+    first();
+    second();
+    release();
+  });
 });

tests/api-counterparty-lifecycle.test.ts

@@ -0,0 +1,72 @@
+import { buildSync } from "esbuild";
+import { createRequire } from "node:module";
+import { fileURLToPath } from "node:url";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+const require = createRequire(import.meta.url);
+const { Miniflare, convertV4MiniflareOptions } = require("miniflare");
+
+// Exercise the real client across separate workerd invocations. Node promises
+// alone do not reproduce Workers cancelling timers and unfinished work when
+// a request ends, while keeping module-level semaphore state for later runs.
+const bundled = buildSync({
+  stdin: {
+    resolveDir: fileURLToPath(new URL("..", import.meta.url)),
+    contents: `
+      import { fetchNodeStatus } from './apps/api/src/integrations/counterparty.ts';
+      const realNow = Date.now;
+      let offset = 0;
+      Date.now = () => realNow() + offset;
+      const realTimeout = AbortSignal.timeout;
+      AbortSignal.timeout = () => realTimeout(100);
+      let hanging = false;
+      let started = 0;
+      let completed = 0;
+      globalThis.fetch = async () => {
+        started++;
+        if (hanging) await new Promise(resolve => setTimeout(resolve, 60_000));
+        return Response.json({ result: { counterparty_height: 123, version: '11.5.0' } });
+      };
+      export default {
+        async fetch(req) {
+          if (new URL(req.url).pathname === '/abandon') {
+            hanging = true;
+            for (let i = 0; i < 2; i++) {
+              fetchNodeStatus().finally(() => completed++).catch(() => {});
+            }
+            await new Promise(resolve => setTimeout(resolve, 10));
+            return Response.json({ started, completed });
+          }
+          hanging = false;
+          // Advance past the read deadline without waiting fifteen seconds.
+          offset = 16_000;
+          try {
+            return Response.json({ node: await fetchNodeStatus(), started, completed });
+          } catch (error) {
+            return Response.json({ error: error.name, started, completed }, { status: 503 });
+          }
+        }
+      };
+    `,
+  },
+  bundle: true, write: false, format: "esm", platform: "node", target: "es2022",
+});
+const options = {
+  modules: true, script: bundled.outputFiles[0].text,
+  compatibilityDate: "2025-01-01", compatibilityFlags: ["nodejs_compat"],
+};
+const mf = new Miniflare(convertV4MiniflareOptions ? convertV4MiniflareOptions(options) : options);
+beforeAll(async () => { await mf.ready; });
+afterAll(async () => { await mf.dispose(); });
+
+describe("Counterparty reads after a cancelled Worker invocation", () => {
+  it("can reach the node after an earlier invocation abandons both read slots", async () => {
+    const abandoned = await mf.dispatchFetch("https://test/abandon");
+    expect(await abandoned.json()).toEqual({ started: 2, completed: 0 });
+    const recovered = await mf.dispatchFetch("https://test/read");
+    expect(recovered.status).toBe(200);
+    expect(await recovered.json()).toEqual({
+      node: { height: 123, version: "11.5.0" }, started: 3, completed: 0,
+    });
+  });
+});