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..6aaf039 --- /dev/null +++ b/relay/src/notify-lifetime.test.ts @@ -0,0 +1,69 @@ +/** + * 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 + * `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; + +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] : []; + }); +} + +/** + * 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 (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); + }); +}); 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 {