diff --git a/apps/api/src/integrations/cooldown.ts b/apps/api/src/integrations/cooldown.ts index 35e0fd3..f41590b 100644 --- a/apps/api/src/integrations/cooldown.ts +++ b/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) { diff --git a/apps/api/src/integrations/counterparty.ts b/apps/api/src/integrations/counterparty.ts index 316e176..fe858ec 100644 --- a/apps/api/src/integrations/counterparty.ts +++ b/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(path: string): Promise { - 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(path: string): Promise { // 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(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?.(); } } diff --git a/tests/api-counterparty-cooldown.test.ts b/tests/api-counterparty-cooldown.test.ts index ea5662c..892371e 100644 --- a/tests/api-counterparty-cooldown.test.ts +++ b/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(); + }); }); diff --git a/tests/api-counterparty-lifecycle.test.ts b/tests/api-counterparty-lifecycle.test.ts new file mode 100644 index 0000000..9221292 --- /dev/null +++ b/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, + }); + }); +});