diff --git a/.changeset/tail-forwarding-speaks-up.md b/.changeset/tail-forwarding-speaks-up.md new file mode 100644 index 00000000000..2a8b773c4a0 --- /dev/null +++ b/.changeset/tail-forwarding-speaks-up.md @@ -0,0 +1,7 @@ +--- +"miniflare": patch +--- + +Report failures to forward tail events between local dev sessions + +When a Worker's tail consumer runs in a separate local dev session and that session becomes unreachable, the failure to deliver tail events was discarded silently. It is now reported as a warning. diff --git a/fixtures/dev-registry/package.json b/fixtures/dev-registry/package.json index c3a3b5c9a0d..c55322d5430 100644 --- a/fixtures/dev-registry/package.json +++ b/fixtures/dev-registry/package.json @@ -5,7 +5,7 @@ "scripts": { "check:types": "tsc --build", "test": "vitest", - "test:ci": "vitest run", + "test:ci": "vitest run tests/crash-bisect.test.ts", "vite": "vite dev", "wrangler": "wrangler dev" }, diff --git a/fixtures/dev-registry/tests/crash-bisect.test.ts b/fixtures/dev-registry/tests/crash-bisect.test.ts new file mode 100644 index 00000000000..1969b25b6bf --- /dev/null +++ b/fixtures/dev-registry/tests/crash-bisect.test.ts @@ -0,0 +1,273 @@ +/** + * TEMPORARY EXPERIMENT — not for merge. + * + * Windows CI results so far: + * 1 4 arms run concurrently with dev-registry.test.ts assets 3/8, plain 0/8 + * 2 same arms isolated, + order swap, + wrangler arm 0/24 + * 3 2/3/4 concurrent idle sessions, + port fallback 0/17 + * 4 long-lived victim + one stressor each 11/12 + * 5 peer startup vs cpu starvation (rewritten harness) 0/12 + * 6 round 4's harness verbatim, context captured 13/17 + * peer churn 4/5, churn+traffic 4/5, traffic 1/4, cpu 0/3 + * 7 what exactly is needed, merged timestamped logs + * alpha 7-10 forced restarts, no peer 0/3 + * beta registry entries written/deleted by hand 0/3 + * gamma real peer, SIGKILLed 3/3 + * delta untouched 0/3 + * + * So neither restarting the victim nor feeding it registry churn does anything. + * It takes a real peer process that starts, gets wired up, and then dies + * abruptly. This round pins down which part of that matters: + * + * epsilon peer killed forcefully, assets + tail consumer 4/5 + * zeta "graceful" stop 1/3 (void: + * taskkill /T never actually stopped vite on Windows, + * cleanExits was 0, so this arm just re-ran epsilon) + * eta peer without assets, killed 0/3 (11 cycles) + * theta unrelated assets peer, killed 0/3 (14 cycles) + * + * An assets peer on its own is not enough (theta) and a bound peer is not + * enough (eta). The only peer that kills the victim is the one that is both an + * assets worker and the victim's tail consumer - so the tail path through the + * assets RPC proxy hop. Every round so far ran without the fix on + * https://github.com/cloudflare/workers-sdk/pull/14993, which makes exactly + * that path stop dropping rejections on the floor. + * + * THIS RUN SITS ON TOP OF THAT FIX. Baseline to beat: epsilon crashed 4/5 reps, + * 17 kill cycles. Same arm, 8 reps, plus both negative controls. + */ +import childProcess from "node:child_process"; +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import { resolve } from "node:path"; +import kill from "tree-kill"; +/* eslint-disable workers-sdk/no-vitest-import-expect -- module-scope helpers */ +import { afterAll, describe, onTestFinished, test } from "vitest"; +/* eslint-enable workers-sdk/no-vitest-import-expect */ + +const cwd = resolve(__dirname, ".."); +const tmpPathBase = path.join(os.tmpdir(), "wrangler-tests"); +const REPS = 8; +const OBSERVE_MS = 30_000; +const CRASH = + /std::terminate|Fatal uncaught|Received signal|crashed unexpectedly/; + +const VICTIM = "exported-handler"; +const PEER = "worker-entrypoint-with-assets"; +const PEER_NO_ASSETS = "worker-entrypoint"; +const PEER_UNRELATED = "exported-handler-with-assets"; + +// eslint-disable-next-line no-control-regex -- strip terminal colour codes +const ANSI = /\u001B\[[0-9;]*[A-Za-z]/g; + +function stripAnsi(value: string) { + return value.replace(ANSI, ""); +} + +const results: { + arm: string; + rep: number; + atMs: number | null; + cycles: number; + cleanExits: number; +}[] = []; +let dumped = 0; + +function spawnMerged(config: string, registry: string, clock: () => number) { + const proc = childProcess.spawn( + "pnpm", + ["vite", "--config", `vite.${config}.config.ts`], + { + cwd, + shell: true, + env: { ...process.env, MINIFLARE_REGISTRY_PATH: registry }, + } + ); + let merged = ""; + let raw = ""; + const attach = (stream: "out" | "err", s: NodeJS.ReadableStream | null) => { + s?.setEncoding("utf8"); + s?.on("data", (chunk: string) => { + raw += chunk; + for (const line of chunk.split(/\r?\n/)) { + if (line.trim() !== "") { + merged += `+${clock()}ms [${stream}] ${stripAnsi(line)}\n`; + } + } + }); + }; + attach("out", proc.stdout); + attach("err", proc.stderr); + + let dead = false; + const closed = new Promise((res) => + proc.once("close", () => { + dead = true; + res(); + }) + ); + + function force() { + return new Promise((res) => { + if (dead || !proc.pid) { + return res(); + } + kill(proc.pid, "SIGKILL", () => res()); + }); + } + + /** + * Ask the tree to close rather than shooting it: tree-kill always passes + * /T /F on Windows, which is exactly the abruptness under test. + */ + function requestStop() { + if (dead || !proc.pid) { + return; + } + if (process.platform === "win32") { + childProcess.spawn("taskkill", ["/pid", String(proc.pid), "/T"], { + stdio: "ignore", + }); + } else { + kill(proc.pid, "SIGTERM", () => {}); + } + } + + onTestFinished(force); + + return { + force, + requestStop, + exited: () => dead, + log: () => merged, + async waitForExit(timeoutMs: number) { + await Promise.race([ + closed, + new Promise((r) => setTimeout(r, timeoutMs)), + ]); + return dead; + }, + async ready(timeoutMs = 40_000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline && !/Local:/.test(stripAnsi(raw))) { + await new Promise((r) => setTimeout(r, 200)); + } + return /Local:/.test(stripAnsi(raw)); + }, + }; +} + +async function arm( + name: string, + rep: number, + opts: { peer: string; graceful?: boolean } +) { + const registry = await fs.realpath(await fs.mkdtemp(tmpPathBase)); + onTestFinished(async () => { + await fs.rm(registry, { recursive: true, maxRetries: 10 }); + }); + + let t0 = Date.now(); + const victim = spawnMerged(VICTIM, registry, () => Date.now() - t0); + if (!(await victim.ready())) { + throw new Error("victim never became ready"); + } + t0 = Date.now(); + + let atMs: number | null = null; + let watching = true; + const watcher = (async () => { + while (watching) { + if (atMs === null && CRASH.test(victim.log())) { + atMs = Date.now() - t0; + } + await new Promise((r) => setTimeout(r, 200)); + } + })(); + + let cycles = 0; + let cleanExits = 0; + try { + while (Date.now() < t0 + OBSERVE_MS) { + const peer = spawnMerged(opts.peer, registry, () => 0); + await peer.ready(); + cycles++; + await new Promise((r) => setTimeout(r, 2000)); + + if (opts.graceful) { + peer.requestStop(); + // A graceful stop that silently fails would leave the peer alive and + // make this arm look clean for the wrong reason, so verify, and only + // then escalate. + if (await peer.waitForExit(10_000)) { + cleanExits++; + } else { + await peer.force(); + } + } else { + await peer.force(); + } + await new Promise((r) => setTimeout(r, 1000)); + } + } finally { + watching = false; + await watcher; + } + + const log = victim.log(); + results.push({ arm: name, rep, atMs, cycles, cleanExits }); + console.log( + `[${name}#${rep}] victim=${ + atMs === null ? "ok" : `CRASHED at +${atMs}ms` + } cycles=${cycles} cleanExits=${cleanExits} restarts=${ + (log.match(/server restarted/g) ?? []).length + }` + ); + + if (atMs !== null && dumped < 2) { + dumped++; + const lines = log.split("\n"); + const i = lines.findIndex((l) => CRASH.test(l)); + console.log(`::group::[VICTIM LOG] ${name}#${rep}`); + console.log(lines.slice(Math.max(0, i - 18), i + 4).join("\n")); + console.log("::endgroup::"); + } +} + +describe.sequential("crash bisect round 8: how the peer dies", () => { + afterAll(() => { + console.log("\n::group::[BISECT SUMMARY]"); + for (const a of [...new Set(results.map((r) => r.arm))]) { + const rows = results.filter((r) => r.arm === a); + const hits = rows.filter((r) => r.atMs !== null); + const cycles = rows.reduce((n, r) => n + r.cycles, 0); + const clean = rows.reduce((n, r) => n + r.cleanExits, 0); + console.log( + ` ${a.padEnd(24)} crashed ${hits.length}/${rows.length} ` + + `cycles=${cycles} cleanExits=${clean}` + + (hits.length + ? ` at=[${hits.map((h) => `+${h.atMs}ms`).join(", ")}]` + : "") + ); + } + console.log("::endgroup::"); + }); + + for (let rep = 1; rep <= REPS; rep++) { + test(`epsilon forceful kill #${rep}`, async () => { + await arm("epsilon force", rep, { peer: PEER }); + }); + + // Controls only need enough reps to stay comparable with round 8. + if (rep <= 3) { + test(`eta peer without assets #${rep}`, async () => { + await arm("eta no-assets peer", rep, { peer: PEER_NO_ASSETS }); + }); + + test(`theta unrelated peer #${rep}`, async () => { + await arm("theta unrelated peer", rep, { peer: PEER_UNRELATED }); + }); + } + } +}); diff --git a/packages/miniflare/src/workers/core/dev-registry-proxy.worker.ts b/packages/miniflare/src/workers/core/dev-registry-proxy.worker.ts index 8fc05d10d87..64e592d5199 100644 --- a/packages/miniflare/src/workers/core/dev-registry-proxy.worker.ts +++ b/packages/miniflare/src/workers/core/dev-registry-proxy.worker.ts @@ -161,7 +161,7 @@ export class ExternalServiceProxy extends WorkerEntrypoint { // Forward tail events to the remote worker via RPC. // Events with rpcMethod==="tail" are filtered out to prevent infinite // recursion (the remote tail() call would itself produce a tail event). - tail(events: TraceItem[]) { + async tail(events: TraceItem[]) { if (!this._fetcher) { return; } @@ -176,8 +176,12 @@ export class ExternalServiceProxy extends WorkerEntrypoint { JSON.stringify(filtered, tailEventsReplacer), tailEventsReviver ); + // `await` rather than `return`: the RPC rejects when the peer's debug + // port has gone away, and returning the promise leaves that rejection + // outside this `try`, so it escapes as an unhandled rejection instead of + // being reported. // @ts-expect-error .tail is not in the `Fetcher` type but it's a valid RPC call - return this._fetcher.tail(serializedEvents); + await this._fetcher.tail(serializedEvents); } catch (e) { console.warn( `[dev-registry] Failed to forward tail events to "${ diff --git a/packages/miniflare/test/dev-registry.spec.ts b/packages/miniflare/test/dev-registry.spec.ts index 5f7714e07f6..7ebfee39158 100644 --- a/packages/miniflare/test/dev-registry.spec.ts +++ b/packages/miniflare/test/dev-registry.spec.ts @@ -1787,4 +1787,70 @@ describe.sequential("DevRegistry", () => { { timeout: 10_000, interval: 100 } ); }); + test("reports a failure to forward tail events to a departed peer", async ({ + expect, + }) => { + const unsafeDevRegistryPath = await useTmp(); + + const remote = new Miniflare({ + name: "remote-worker", + unsafeDevRegistryPath, + compatibilityFlags: ["experimental"], + modules: true, + script: ` + import { WorkerEntrypoint } from "cloudflare:workers"; + export default class extends WorkerEntrypoint { + tail() {} + } + `, + }); + useDispose(remote); + await remote.ready; + + const logs: string[] = []; + const local = new Miniflare({ + name: "local-worker", + unsafeDevRegistryPath, + tails: ["remote-worker"], + compatibilityFlags: ["experimental"], + modules: true, + handleStructuredLogs: ({ message }) => void logs.push(message), + script: ` + export default { + fetch() { + console.log("tail me"); + return new Response("ok"); + } + } + `, + }); + useDispose(local); + await local.ready; + + // Establish the tail forwarding path while the peer is alive. + await vi.waitFor( + async () => { + await (await local.dispatchFetch("http://placeholder")).text(); + expect(logs.join("\n")).toContain("tail me"); + }, + { timeout: 10_000, interval: 100 } + ); + + // Drop the peer without letting it deregister, so `local` keeps a registry + // entry pointing at a debug port that is no longer accepting connections. + // The forwarding RPC now rejects; that rejection must be reported rather + // than escaping as an unhandled rejection. + await remote.dispose(); + logs.length = 0; + + await vi.waitFor( + async () => { + await (await local.dispatchFetch("http://placeholder")).text(); + expect(logs.join("\n")).toContain( + `[dev-registry] Failed to forward tail events to "remote-worker"` + ); + }, + { timeout: 10_000, interval: 100 } + ); + }); });