From c9de7928c80e859f57de6b96e2cf9795a9761db5 Mon Sep 17 00:00:00 2001 From: hallelx2 Date: Mon, 12 Oct 2026 00:24:32 +0100 Subject: [PATCH 1/2] fix(relay): keep stream notifies alive with waitUntil Every Durable Object notify was a floating fetch, so a Worker could cancel it once the handler returned its 202 and the live webhook frame never reached the extension, the dashboard or the CLI; they then waited for their next poll. The production CLI e2e lost the frame in about half its runs. The channel host (.bridgehook.dev, which real senders use) had no ExecutionContext at all. A background function (ctx.waitUntil) now runs from the Worker entry and Hono contexts into intake, claim, response and replay notifies. A guard test fails if a do/notify fetch is left floating again. --- relay/src/event-store.ts | 5 +- relay/src/index.ts | 117 +++++++++++++++++++++++------- relay/src/notify-lifetime.test.ts | 36 +++++++++ relay/src/routes/me.ts | 2 + 4 files changed, 134 insertions(+), 26 deletions(-) create mode 100644 relay/src/notify-lifetime.test.ts diff --git a/relay/src/event-store.ts b/relay/src/event-store.ts index ae39a62..bb374bc 100644 --- a/relay/src/event-store.ts +++ b/relay/src/event-store.ts @@ -13,6 +13,8 @@ export const MAX_REPLAY_BODY_BYTES = 1_048_576; // 1 MB export interface ReplayNotifier { getChannelDO(channelId: string): { fetch(req: Request): Promise }; notifyUser(userId: string | null, payload: string): void; + /** Keeps a notify alive past the response (ctx.waitUntil). */ + background?(p: Promise): void; } export function safeJsonObject(raw: string | null): Record { @@ -154,10 +156,11 @@ export async function queueReplay( kind: "replay", replayOf: source.id, }); - notifier + const notified = notifier .getChannelDO(source.channelId) .fetch(new Request("https://do/notify", { method: "POST", body: ssePayload })) .catch((err) => console.error("DO notify failed:", err)); + notifier.background?.(notified); notifier.notifyUser(userId, ssePayload); return { diff --git a/relay/src/index.ts b/relay/src/index.ts index db13e07..fd7a1a4 100644 --- a/relay/src/index.ts +++ b/relay/src/index.ts @@ -192,12 +192,56 @@ function getUserDO(env: Env, userId: string) { * Best-effort fan-out to a user's UserDO. Failures are logged and swallowed * — the SSE push is a UX nicety, never load-bearing. */ -function notifyUserDO(env: Env, userId: string | null, payload: string): void { +/** + * Keeps a promise alive after the response is sent. A notify left floating + * can be cancelled the moment the handler returns, and then live clients + * (extension, dashboard, CLI) miss the webhook frame and wait for a poll. + */ +type Background = (p: Promise) => void; + +/** waitUntil from an ExecutionContext; without one (tests), let it float. */ +function backgroundOf(ctx: Pick | undefined): Background { + return (p) => { + if (ctx) ctx.waitUntil(p); + }; +} + +/** waitUntil from a Hono context; `executionCtx` throws when there is none. */ +function honoBackground(c: { executionCtx: ExecutionContext }): Background { + let ctx: ExecutionContext | undefined; + try { + ctx = c.executionCtx; + } catch { + ctx = undefined; + } + return backgroundOf(ctx); +} + +function notifyChannelDO( + env: Env, + channelId: string, + payload: string, + background: Background, +): void { + background( + getChannelDO(env, channelId) + .fetch(new Request("https://do/notify", { method: "POST", body: payload })) + .catch((err) => console.error("DO notify failed:", err)), + ); +} + +function notifyUserDO( + env: Env, + userId: string | null, + payload: string, + background: Background, +): void { if (!userId) return; - const stub = getUserDO(env, userId); - stub - .fetch(new Request("https://do/notify", { method: "POST", body: payload })) - .catch((err) => console.error("UserDO notify failed:", err)); + background( + getUserDO(env, userId) + .fetch(new Request("https://do/notify", { method: "POST", body: payload })) + .catch((err) => console.error("UserDO notify failed:", err)), + ); } function jsonResponse(status: number, body: unknown): Response { @@ -224,6 +268,7 @@ async function handleWebhookIntake( forwardPath: string, request: Request, env: Env, + background: Background, ): Promise { if (!CHANNEL_ID_RE.test(channelId)) { return jsonResponse(404, { error: "Channel not found" }); @@ -388,12 +433,14 @@ async function handleWebhookIntake( }), ) : null; - stub - .fetch(new Request("https://do/notify", { method: "POST", body: ssePayload })) - .catch((err) => console.error("DO notify failed:", err)); + background( + stub + .fetch(new Request("https://do/notify", { method: "POST", body: ssePayload })) + .catch((err) => console.error("DO notify failed:", err)), + ); - // Cross-channel dashboard fan-out: only when the channel has an owner. - notifyUserDO(env, channel.userId, ssePayload); + // Cross-channel fan-out (dashboard, extension, CLI): owned channels only. + notifyUserDO(env, channel.userId, ssePayload, background); if (waiting) { let outcome: SyncOutcome = { kind: "timeout" }; @@ -617,7 +664,14 @@ app.route( db: getDb(env), notifier: { getChannelDO: (channelId: string) => getChannelDO(env, channelId), - notifyUser: (userId: string | null, payload: string) => notifyUserDO(env, userId, payload), + notifyUser: (userId: string | null, payload: string) => + notifyUserDO( + env, + userId, + payload, + honoBackground(c as unknown as { executionCtx: ExecutionContext }), + ), + background: honoBackground(c as unknown as { executionCtx: ExecutionContext }), }, tunnelDomain: env.TUNNEL_DOMAIN ?? null, }; @@ -717,10 +771,12 @@ app.all("/mcp", async (c) => { tunnelDomain: env.TUNNEL_DOMAIN, notifier: { getChannelDO: (channelId: string) => getChannelDO(env, channelId), - notifyUser: (uid: string | null, payload: string) => notifyUserDO(env, uid, payload), + notifyUser: (uid: string | null, payload: string) => + notifyUserDO(env, uid, payload, honoBackground(c)), + background: honoBackground(c), }, deliver: (channelId: string, forwardPath: string, request: Request) => - handleWebhookIntake(channelId, forwardPath, request, env), + handleWebhookIntake(channelId, forwardPath, request, env, honoBackground(c)), waitUntil: (p: Promise) => c.executionCtx.waitUntil(p), }); }); @@ -1091,7 +1147,7 @@ app.get("/api/channels/:channelId/events", async (c) => { // The canonical `/[/path]` form is registered last, after every // first-party route, so it can never shadow one. See ./webhook-url.ts. app.on([...WEBHOOK_METHODS, "GET", "HEAD"], "/hook/:channelId", async (c) => { - return handleWebhookIntake(c.req.param("channelId"), "/", c.req.raw, c.env); + return handleWebhookIntake(c.req.param("channelId"), "/", c.req.raw, c.env, honoBackground(c)); }); // ── Claim event for executor (auth) ── @@ -1172,11 +1228,9 @@ app.post("/hook/:channelId/claim", async (c) => { claimerId: clientId, claimedAt: claimedAt.toISOString(), }); - const stub = getChannelDO(c.env, channelId); - stub - .fetch(new Request("https://do/notify", { method: "POST", body: payload })) - .catch((err) => console.error("DO notify (claim) failed:", err)); - notifyUserDO(c.env, channel.userId, payload); + const background = honoBackground(c); + notifyChannelDO(c.env, channelId, payload, background); + notifyUserDO(c.env, channel.userId, payload, background); return c.json({ claimed: true, claimerId: clientId, claimedAt: claimedAt.toISOString() }); } @@ -1305,7 +1359,7 @@ app.post("/hook/:channelId/response", async (c) => { await stub .fetch(new Request("https://do/notify", { method: "POST", body: doPayload })) .catch((err) => console.error("DO notify failed:", err)); - notifyUserDO(c.env, channel.userId, responsePayload); + notifyUserDO(c.env, channel.userId, responsePayload, honoBackground(c)); return c.json({ ok: true }); }); @@ -1319,7 +1373,13 @@ app.all("*", async (c, next) => { if (!parsed) return next(); const m = c.req.method.toUpperCase(); if (isWebhookMethod(m) || m === "GET" || m === "HEAD") { - return handleWebhookIntake(parsed.channelId, parsed.forwardPath, c.req.raw, c.env); + return handleWebhookIntake( + parsed.channelId, + parsed.forwardPath, + c.req.raw, + c.env, + honoBackground(c), + ); } return c.json({ error: "Method not allowed" }, 405, { Allow: [...WEBHOOK_METHODS, "GET"].join(", "), @@ -1369,10 +1429,15 @@ async function retentionSweep(db: DB): Promise { * auth and the executor endpoints are not reachable on channel hosts, and no * response carries credentials. */ -async function handleChannelHost(channelId: string, request: Request, env: Env): Promise { +async function handleChannelHost( + channelId: string, + request: Request, + env: Env, + background: Background, +): Promise { let res: Response; try { - res = await channelHostResponse(channelId, request, env); + res = await channelHostResponse(channelId, request, env, background); } catch (err) { // Outside Hono, so app.onError never sees this; answer the same way. console.error("Channel host error:", err); @@ -1391,11 +1456,12 @@ async function channelHostResponse( channelId: string, request: Request, env: Env, + background: Background, ): Promise { const url = new URL(request.url); const method = request.method.toUpperCase(); if (isWebhookMethod(method) || method === "GET" || method === "HEAD") { - return handleWebhookIntake(channelId, url.pathname || "/", request, env); + return handleWebhookIntake(channelId, url.pathname || "/", request, env, background); } if (method === "OPTIONS") { return new Response(null, { @@ -1422,7 +1488,8 @@ async function channelHostResponse( export default { async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise { const host = classifyHost(request.headers.get("host"), env.TUNNEL_DOMAIN); - if (host.kind === "channel") return handleChannelHost(host.channelId, request, env); + if (host.kind === "channel") + return handleChannelHost(host.channelId, request, env, backgroundOf(ctx)); // Only relay. serves the API, auth and executor endpoints. // Every other in-zone host (www, api, foo-bar, a.b, …) is a 404; // app/docs never reach the Worker (Pages, script-less routes). diff --git a/relay/src/notify-lifetime.test.ts b/relay/src/notify-lifetime.test.ts new file mode 100644 index 0000000..d0637e9 --- /dev/null +++ b/relay/src/notify-lifetime.test.ts @@ -0,0 +1,36 @@ +/** + * A Durable Object notify left floating can be cancelled once the handler + * returns its response, and live clients then miss the frame (HAL-2721). Every + * `https://do/notify` fetch must be awaited, or handed to ctx.waitUntil + * (`background(...)` / `notifier.background`). + */ +import { readFileSync, readdirSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; + +const SRC = __dirname; +const KEPT_ALIVE = /(await\s|background\(\s*$|const notified = )/m; + +function sourceFiles(dir: string): string[] { + return readdirSync(dir, { withFileTypes: true }).flatMap((e) => { + const p = join(dir, e.name); + if (e.isDirectory()) return sourceFiles(p); + return e.name.endsWith(".ts") && !e.name.endsWith(".test.ts") ? [p] : []; + }); +} + +describe("Durable Object notifies outlive the response", () => { + it("awaits every do/notify fetch or passes it to waitUntil", () => { + const floating: string[] = []; + for (const file of sourceFiles(SRC)) { + const lines = readFileSync(file, "utf8").split("\n"); + lines.forEach((line, i) => { + if (!line.includes('"https://do/notify"')) return; + // The statement that starts the fetch chain: up to 4 lines above. + const head = lines.slice(Math.max(0, i - 4), i + 1).join("\n"); + if (!KEPT_ALIVE.test(head)) floating.push(`${file.slice(SRC.length + 1)}:${i + 1}`); + }); + } + expect(floating).toEqual([]); + }); +}); diff --git a/relay/src/routes/me.ts b/relay/src/routes/me.ts index b406981..91e7f90 100644 --- a/relay/src/routes/me.ts +++ b/relay/src/routes/me.ts @@ -43,6 +43,8 @@ export interface ChannelNotifier { fetch(req: Request): Promise; }; notifyUser(userId: string | null, payload: string): void; + /** Keeps a notify alive past the response (ctx.waitUntil). */ + background?(p: Promise): void; } export interface MeEnv { From 987a97fd2a2d1380bdfd3e4e8bd85bc0e829b471 Mon Sep 17 00:00:00 2001 From: hallelx2 Date: Mon, 12 Oct 2026 00:30:54 +0100 Subject: [PATCH 2/2] test(relay): check the statement that opens each notify chain Review: the guard passed if any await sat in the four lines above, so a floating notify right after an awaited call slipped through. It now finds the line that opens the fetch chain and requires that line to await or store the promise, or to sit inside a background( call; it matches any do/notify string, and tests its own loophole cases. --- relay/src/notify-lifetime.test.ts | 45 ++++++++++++++++++++++++++----- 1 file changed, 39 insertions(+), 6 deletions(-) diff --git a/relay/src/notify-lifetime.test.ts b/relay/src/notify-lifetime.test.ts index d0637e9..6aaf039 100644 --- a/relay/src/notify-lifetime.test.ts +++ b/relay/src/notify-lifetime.test.ts @@ -1,7 +1,7 @@ /** * A Durable Object notify left floating can be cancelled once the handler * returns its response, and live clients then miss the frame (HAL-2721). Every - * `https://do/notify` fetch must be awaited, or handed to ctx.waitUntil + * `do/notify` fetch must be awaited, or handed to ctx.waitUntil * (`background(...)` / `notifier.background`). */ import { readFileSync, readdirSync } from "node:fs"; @@ -9,7 +9,6 @@ import { join } from "node:path"; import { describe, expect, it } from "vitest"; const SRC = __dirname; -const KEPT_ALIVE = /(await\s|background\(\s*$|const notified = )/m; function sourceFiles(dir: string): string[] { return readdirSync(dir, { withFileTypes: true }).flatMap((e) => { @@ -19,18 +18,52 @@ function sourceFiles(dir: string): string[] { }); } +/** + * The statement that opens the fetch chain containing line `i`: the nearest + * line at or above it that is not a `.method(` continuation. It is kept alive + * when it awaits or stores the promise for waitUntil, or when the line before + * it opens a `background(` call. + */ +export function keptAlive(lines: string[], i: number): boolean { + let open = i; + while (open > 0 && lines[open].trim().startsWith(".")) open--; + const head = lines[open].trim(); + if (/^(return\s+)?await\s/.test(head) || /^const notified = /.test(head)) return true; + if (/background\??\.?\($/.test(head)) return true; + return open > 0 && /background\??\.?\(\s*$/.test(lines[open - 1]); +} + +const NOTIFY = /do\/notify/; + describe("Durable Object notifies outlive the response", () => { it("awaits every do/notify fetch or passes it to waitUntil", () => { const floating: string[] = []; for (const file of sourceFiles(SRC)) { const lines = readFileSync(file, "utf8").split("\n"); lines.forEach((line, i) => { - if (!line.includes('"https://do/notify"')) return; - // The statement that starts the fetch chain: up to 4 lines above. - const head = lines.slice(Math.max(0, i - 4), i + 1).join("\n"); - if (!KEPT_ALIVE.test(head)) floating.push(`${file.slice(SRC.length + 1)}:${i + 1}`); + if (NOTIFY.test(line) && !keptAlive(lines, i)) + floating.push(`${file.slice(SRC.length + 1)}:${i + 1}`); }); } expect(floating).toEqual([]); }); + + it("flags a floating fetch even right after an awaited statement", () => { + const src = [ + "\tawait touchDevice(db, id);", + "\tstub", + '\t\t.fetch(new Request("https://do/notify", { method: "POST" }))', + "\t\t.catch(() => {});", + "\tvoid stub.fetch(`https://do/notify`);", + "\tbackground(", + "\t\tstub", + '\t\t\t.fetch(new Request("https://do/notify"))', + "\t);", + '\tawait stub.fetch(new Request("https://do/notify"));', + ]; + expect(keptAlive(src, 2)).toBe(false); + expect(keptAlive(src, 4)).toBe(false); + expect(keptAlive(src, 7)).toBe(true); + expect(keptAlive(src, 9)).toBe(true); + }); });