diff --git a/.changeset/lucky-pandas-wander.md b/.changeset/lucky-pandas-wander.md new file mode 100644 index 00000000000..0cdf692ebec --- /dev/null +++ b/.changeset/lucky-pandas-wander.md @@ -0,0 +1,9 @@ +--- +"miniflare": patch +--- + +Stop deleting and recreating every dev registry entry on each config update + +Applying options rewrote this instance's dev registry entries by removing them and putting them straight back. Other dev sessions find Workers by watching that directory, so each update briefly looked to them like every Worker in the session had gone away — and a session that had already resolved one of those Workers could be left acting on that, up to and including tearing down a binding to a Worker that never actually stopped running. + +Entries are now reconciled instead: Workers that are still present are updated in place, and only the ones that have genuinely gone are removed. Switching to a different registry path still clears the entries from the directory being left behind. diff --git a/.github/workflows/test-and-check.yml b/.github/workflows/test-and-check.yml index ad538d90fc4..5bc09fa3833 100644 --- a/.github/workflows/test-and-check.yml +++ b/.github/workflows/test-and-check.yml @@ -82,7 +82,8 @@ jobs: run: node packages/wrangler/src/__tests__/test-old-node-version.js error test: - timeout-minutes: 30 + # TEMPORARY VALIDATION — not for merge. + timeout-minutes: 60 concurrency: group: ${{ github.workflow }}-${{ github.ref }}-${{ matrix.os }}-${{ matrix.suite }}-test cancel-in-progress: ${{ github.head_ref != 'changeset-release/main' }} @@ -187,7 +188,9 @@ jobs: # Since the dev registry is now file-based (not network-based), fixture tests can safely run in parallel. # Concurrency is capped at 2 to avoid CPU starvation on CI runners when multiple fixtures # spawn workerd processes simultaneously (Windows runners are especially slow under load). - run: pnpm run test:ci --concurrency=2 --log-order=stream --filter="./fixtures/*" ${{ matrix.os == 'ubuntu-latest' && '--filter="!./fixtures/browser-run"' || '' }} + # TEMPORARY VALIDATION — not for merge. Isolated so an unrelated fixture + # flake can't abort the job before dev-registry runs. + run: pnpm run test:ci --log-order=stream --filter="@fixture/dev-registry" env: NODE_OPTIONS: "--max_old_space_size=8192" WRANGLER_LOG_PATH: ${{ runner.temp }}/wrangler-debug-logs/ diff --git a/fixtures/dev-registry/tests/dev-registry.test.ts b/fixtures/dev-registry/tests/dev-registry.test.ts index c60c9e7ac41..4d0e27b151e 100644 --- a/fixtures/dev-registry/tests/dev-registry.test.ts +++ b/fixtures/dev-registry/tests/dev-registry.test.ts @@ -18,10 +18,9 @@ import { } from "../../../packages/vite-plugin-cloudflare/e2e/helpers"; import { runWranglerDev as baseRunWranglerDev } from "../../shared/src/run-wrangler-long-lived"; -// TODO: These tests are consistently failing on Windows in CI and are blocking -// other work. Skipping them there as a temporary measure until the underlying -// issue is fixed. There's still value in running them on macOS and Linux. -const describe = baseDescribe.skipIf(process.platform === "win32"); +// TEMPORARY VALIDATION — not for merge. #15018's Windows skip is lifted so the +// fix can be measured against the suite it is meant to help. +const describe = baseDescribe; const waitForTimeout = 20_000; const cwd = resolve(__dirname, ".."); @@ -46,11 +45,20 @@ async function runViteDev( }); const url = await waitForReady(proc); - onTestFailed(() => { - console.log(`::group::Vite dev session (${config})`); + // TEMPORARY VALIDATION — not for merge. Dump on finish rather than only on + // failure: dumping only failed tests biases the record, and a buffer that + // ends mid-recovery then looks identical to a genuine stall. + onTestFinished(() => { + console.log( + `::group::Vite dev session (${config}) captured-at ${new Date().toISOString()}` + ); console.log(proc.stdout); console.log(proc.stderr); console.log("::endgroup::"); + const output = proc.stdout + proc.stderr; + if (/std::terminate|crashed unexpectedly/.test(output)) { + console.log(`CRASH-DETECTED ${config}`); + } }); // Wait for the dev session to be ready @@ -517,298 +525,302 @@ describe("Dev Registry: wrangler dev <-> wrangler dev", () => { }); }); -describe("Dev Registry: vite dev <-> vite dev", () => { - it("supports exported handler fetch over service binding", async ({ - devRegistryPath, - }) => { - const workerEntrypointWithAssets = await runViteDev( - "vite.worker-entrypoint-with-assets.config.ts", - devRegistryPath - ); - await runViteDev("vite.worker-entrypoint.config.ts", devRegistryPath); - - // Test fallback before exported-handler is started - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "exported-handler", - "test-method": "fetch", - }); - const response = await fetch( - `${workerEntrypointWithAssets}?${searchParams}` - ); - - expect(response.status).toBe(503); - expect(await response.text()).toEqual( - `Worker "exported-handler" not found. Make sure it is running locally.` +for (const validationRound of [1, 2, 3, 4]) { + describe(`Dev Registry: vite dev <-> vite dev [round ${validationRound}]`, () => { + it("supports exported handler fetch over service binding", async ({ + devRegistryPath, + }) => { + const workerEntrypointWithAssets = await runViteDev( + "vite.worker-entrypoint-with-assets.config.ts", + devRegistryPath ); - }, waitForTimeout); + await runViteDev("vite.worker-entrypoint.config.ts", devRegistryPath); - const exportedHandler = await runViteDev( - "vite.exported-handler.config.ts", - devRegistryPath - ); - - // Test exported-handler -> worker-entrypoint-with-assets - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "worker-entrypoint-with-assets", - "test-method": "fetch", - }); - const response = await fetch(`${exportedHandler}?${searchParams}`); + // Test fallback before exported-handler is started + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "exported-handler", + "test-method": "fetch", + }); + const response = await fetch( + `${workerEntrypointWithAssets}?${searchParams}` + ); - expect(await response.text()).toBe("Hello from Worker Entrypoint!"); - expect(response.status).toBe(200); + expect(response.status).toBe(503); + expect(await response.text()).toEqual( + `Worker "exported-handler" not found. Make sure it is running locally.` + ); + }, waitForTimeout); - // Test fetching asset from "worker-entrypoint-with-assets" over service binding - // Exported handler has no assets, so it will hit the user worker and - // forward the request to "worker-entrypoint-with-assets" with the asset path - const assetResponse = await fetch( - `${exportedHandler}/example.txt?${searchParams}` + const exportedHandler = await runViteDev( + "vite.exported-handler.config.ts", + devRegistryPath ); - expect(await assetResponse.text()).toBe("This is an example asset file"); - }, waitForTimeout); - // Test worker-entrypoint-with-assets -> exported-handler - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "exported-handler", - "test-method": "fetch", - }); - const response = await fetch( - `${workerEntrypointWithAssets}?${searchParams}` - ); + // Test exported-handler -> worker-entrypoint-with-assets + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "worker-entrypoint-with-assets", + "test-method": "fetch", + }); + const response = await fetch(`${exportedHandler}?${searchParams}`); - expect(await response.text()).toEqual("Hello from exported handler!"); - expect(response.status).toBe(200); - }, waitForTimeout); + expect(await response.text()).toBe("Hello from Worker Entrypoint!"); + expect(response.status).toBe(200); - // Test exported-handler -> named-entrypoint - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "named-entrypoint", - "test-method": "fetch", - }); - const response = await fetch(`${exportedHandler}?${searchParams}`); + // Test fetching asset from "worker-entrypoint-with-assets" over service binding + // Exported handler has no assets, so it will hit the user worker and + // forward the request to "worker-entrypoint-with-assets" with the asset path + const assetResponse = await fetch( + `${exportedHandler}/example.txt?${searchParams}` + ); + expect(await assetResponse.text()).toBe( + "This is an example asset file" + ); + }, waitForTimeout); - expect(await response.text()).toEqual("Hello from Named Entrypoint!"); - expect(response.status).toBe(200); - }, waitForTimeout); + // Test worker-entrypoint-with-assets -> exported-handler + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "exported-handler", + "test-method": "fetch", + }); + const response = await fetch( + `${workerEntrypointWithAssets}?${searchParams}` + ); - // Test exported-handler -> named-entrypoint-with-assets - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "named-entrypoint-with-assets", - "test-method": "fetch", - }); - const response = await fetch(`${exportedHandler}?${searchParams}`); + expect(await response.text()).toEqual("Hello from exported handler!"); + expect(response.status).toBe(200); + }, waitForTimeout); - expect(await response.text()).toEqual("Hello from Named Entrypoint!"); - expect(response.status).toBe(200); - }, waitForTimeout); - }); + // Test exported-handler -> named-entrypoint + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "named-entrypoint", + "test-method": "fetch", + }); + const response = await fetch(`${exportedHandler}?${searchParams}`); - it("supports RPC over service binding", async ({ devRegistryPath }) => { - const exportedHandler = await runViteDev( - "vite.exported-handler.config.ts", - devRegistryPath - ); - await runViteDev("vite.worker-entrypoint.config.ts", devRegistryPath); + expect(await response.text()).toEqual("Hello from Named Entrypoint!"); + expect(response.status).toBe(200); + }, waitForTimeout); - // Test fallback before worker-entrypoint-with-assets is started - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "worker-entrypoint-with-assets", - "test-method": "rpc", - }); - const response = await fetch(`${exportedHandler}?${searchParams}`); + // Test exported-handler -> named-entrypoint-with-assets + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "named-entrypoint-with-assets", + "test-method": "fetch", + }); + const response = await fetch(`${exportedHandler}?${searchParams}`); - expect(response.status).toBe(500); - expect(await response.text()).toEqual( - `Worker "worker-entrypoint-with-assets" not found. Make sure it is running locally.` - ); - }, waitForTimeout); + expect(await response.text()).toEqual("Hello from Named Entrypoint!"); + expect(response.status).toBe(200); + }, waitForTimeout); + }); - await runViteDev( - "vite.worker-entrypoint-with-assets.config.ts", - devRegistryPath - ); + it("supports RPC over service binding", async ({ devRegistryPath }) => { + const exportedHandler = await runViteDev( + "vite.exported-handler.config.ts", + devRegistryPath + ); + await runViteDev("vite.worker-entrypoint.config.ts", devRegistryPath); - // Test exported-handler -> worker-entrypoint RPC - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "worker-entrypoint", - "test-method": "rpc", - }); - const response = await fetch(`${exportedHandler}?${searchParams}`); + // Test fallback before worker-entrypoint-with-assets is started + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "worker-entrypoint-with-assets", + "test-method": "rpc", + }); + const response = await fetch(`${exportedHandler}?${searchParams}`); - expect(response.status).toBe(200); - expect(await response.text()).toEqual("Pong"); - }, waitForTimeout); + expect(response.status).toBe(500); + expect(await response.text()).toEqual( + `Worker "worker-entrypoint-with-assets" not found. Make sure it is running locally.` + ); + }, waitForTimeout); - // Test exported-handler -> worker-entrypoint-with-assets RPC - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "worker-entrypoint-with-assets", - "test-method": "rpc", - }); - const response = await fetch(`${exportedHandler}?${searchParams}`); + await runViteDev( + "vite.worker-entrypoint-with-assets.config.ts", + devRegistryPath + ); - expect(response.status).toBe(200); - expect(await response.text()).toEqual("Pong"); - }, waitForTimeout); + // Test exported-handler -> worker-entrypoint RPC + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "worker-entrypoint", + "test-method": "rpc", + }); + const response = await fetch(`${exportedHandler}?${searchParams}`); - // Test exported-handler -> named-entrypoint RPC - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "named-entrypoint", - "test-method": "rpc", - }); - const response = await fetch(`${exportedHandler}?${searchParams}`); + expect(response.status).toBe(200); + expect(await response.text()).toEqual("Pong"); + }, waitForTimeout); - expect(response.status).toBe(200); - expect(await response.text()).toEqual("Pong from Named Entrypoint"); - }, waitForTimeout); + // Test exported-handler -> worker-entrypoint-with-assets RPC + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "worker-entrypoint-with-assets", + "test-method": "rpc", + }); + const response = await fetch(`${exportedHandler}?${searchParams}`); - // Test exported-handler -> named-entrypoint-with-assets RPC - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "named-entrypoint-with-assets", - "test-method": "rpc", - }); - const response = await fetch(`${exportedHandler}?${searchParams}`); + expect(response.status).toBe(200); + expect(await response.text()).toEqual("Pong"); + }, waitForTimeout); - expect(response.status).toBe(200); - expect(await response.text()).toEqual("Pong from Named Entrypoint"); - }, waitForTimeout); - }); + // Test exported-handler -> named-entrypoint RPC + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "named-entrypoint", + "test-method": "rpc", + }); + const response = await fetch(`${exportedHandler}?${searchParams}`); - it("supports WebSocket upgrade over service binding", async ({ - devRegistryPath, - }) => { - const exportedHandler = await runViteDev( - "vite.exported-handler.config.ts", - devRegistryPath - ); - await runViteDev("vite.worker-entrypoint.config.ts", devRegistryPath); + expect(response.status).toBe(200); + expect(await response.text()).toEqual("Pong from Named Entrypoint"); + }, waitForTimeout); - // Test exported-handler -> worker-entrypoint WebSocket proxy - await vi.waitFor(async () => { - const searchParams = new URLSearchParams({ - "test-service": "worker-entrypoint", - "test-method": "websocket-proxy", - }); - const wsUrl = `${exportedHandler.replace("http", "ws")}?${searchParams}`; - const ws = new WebSocket(wsUrl); - - const message = await new Promise((resolve, reject) => { - ws.addEventListener("open", () => ws.send("hello")); - ws.addEventListener("message", (event) => { - resolve(String(event.data)); - ws.close(); + // Test exported-handler -> named-entrypoint-with-assets RPC + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "named-entrypoint-with-assets", + "test-method": "rpc", }); - ws.addEventListener("error", () => - reject(new Error("WebSocket connection failed")) - ); - }); + const response = await fetch(`${exportedHandler}?${searchParams}`); - expect(message).toBe("echo:hello"); - }, waitForTimeout); - }); + expect(response.status).toBe(200); + expect(await response.text()).toEqual("Pong from Named Entrypoint"); + }, waitForTimeout); + }); - it("supports tail handler", async ({ devRegistryPath }) => { - const exportedHandler = await runViteDev( - "vite.exported-handler.config.ts", - devRegistryPath - ); - const workerEntrypointWithAssets = await runViteDev( - "vite.worker-entrypoint-with-assets.config.ts", - devRegistryPath - ); + it("supports WebSocket upgrade over service binding", async ({ + devRegistryPath, + }) => { + const exportedHandler = await runViteDev( + "vite.exported-handler.config.ts", + devRegistryPath + ); + await runViteDev("vite.worker-entrypoint.config.ts", devRegistryPath); - const searchParams = new URLSearchParams({ - "test-method": "tail", - }); + // Test exported-handler -> worker-entrypoint WebSocket proxy + await vi.waitFor(async () => { + const searchParams = new URLSearchParams({ + "test-service": "worker-entrypoint", + "test-method": "websocket-proxy", + }); + const wsUrl = `${exportedHandler.replace("http", "ws")}?${searchParams}`; + const ws = new WebSocket(wsUrl); + + const message = await new Promise((resolve, reject) => { + ws.addEventListener("open", () => ws.send("hello")); + ws.addEventListener("message", (event) => { + resolve(String(event.data)); + ws.close(); + }); + ws.addEventListener("error", () => + reject(new Error("WebSocket connection failed")) + ); + }); - await vi.waitFor(async () => { - // Trigger tail handler of worker-entrypoint via exported-handler - await fetch(`${exportedHandler}?${searchParams}`, { - method: "POST", - body: JSON.stringify(["hello world", "this is the 2nd log"]), - }); - await fetch(`${exportedHandler}?${searchParams}`, { - method: "POST", - body: JSON.stringify(["some other log"]), - }); + expect(message).toBe("echo:hello"); + }, waitForTimeout); + }); - const response = await fetch( - `${workerEntrypointWithAssets}?${searchParams}` + it("supports tail handler", async ({ devRegistryPath }) => { + const exportedHandler = await runViteDev( + "vite.exported-handler.config.ts", + devRegistryPath + ); + const workerEntrypointWithAssets = await runViteDev( + "vite.worker-entrypoint-with-assets.config.ts", + devRegistryPath ); - expect(await response.json()).toEqual({ - worker: "Worker Entrypoint", - tailEvents: expect.arrayContaining([ - [["[exported-handler]"], ["hello world", "this is the 2nd log"]], - [["[exported-handler]"], ["some other log"]], - ]), + const searchParams = new URLSearchParams({ + "test-method": "tail", }); - }, waitForTimeout); - await vi.waitFor(async () => { - // Trigger tail handler of exported-handler via worker-entrypoint - await fetch(`${workerEntrypointWithAssets}?${searchParams}`, { - method: "POST", - body: JSON.stringify(["hello from test"]), - }); - await fetch(`${workerEntrypointWithAssets}?${searchParams}`, { - method: "POST", - body: JSON.stringify(["yet another log", "and another one"]), - }); + await vi.waitFor(async () => { + // Trigger tail handler of worker-entrypoint via exported-handler + await fetch(`${exportedHandler}?${searchParams}`, { + method: "POST", + body: JSON.stringify(["hello world", "this is the 2nd log"]), + }); + await fetch(`${exportedHandler}?${searchParams}`, { + method: "POST", + body: JSON.stringify(["some other log"]), + }); - const response = await fetch(`${exportedHandler}?${searchParams}`); + const response = await fetch( + `${workerEntrypointWithAssets}?${searchParams}` + ); - expect(await response.json()).toEqual({ - worker: "exported-handler", - tailEvents: expect.arrayContaining([ - [["[Worker Entrypoint]"], ["hello from test"]], - [["[Worker Entrypoint]"], ["yet another log", "and another one"]], - ]), - }); - }, waitForTimeout); - }); + expect(await response.json()).toEqual({ + worker: "Worker Entrypoint", + tailEvents: expect.arrayContaining([ + [["[exported-handler]"], ["hello world", "this is the 2nd log"]], + [["[exported-handler]"], ["some other log"]], + ]), + }); + }, waitForTimeout); - it("supports queues across dev sessions", async ({ devRegistryPath }) => { - const exportedHandler = await runViteDev( - "vite.exported-handler.config.ts", - devRegistryPath - ); - const workerEntrypoint = await runViteDev( - "vite.worker-entrypoint.config.ts", - devRegistryPath - ); + await vi.waitFor(async () => { + // Trigger tail handler of exported-handler via worker-entrypoint + await fetch(`${workerEntrypointWithAssets}?${searchParams}`, { + method: "POST", + body: JSON.stringify(["hello from test"]), + }); + await fetch(`${workerEntrypointWithAssets}?${searchParams}`, { + method: "POST", + body: JSON.stringify(["yet another log", "and another one"]), + }); - await vi.waitFor(async () => { - const sendParams = new URLSearchParams({ - "test-method": "queue-send", - }); - const sendResponse = await fetch(`${exportedHandler}?${sendParams}`, { - method: "POST", - body: "hello from vite producer", - }); - expect(await sendResponse.text()).toBe("Queued"); - expect(sendResponse.status).toBe(200); + const response = await fetch(`${exportedHandler}?${searchParams}`); - const receivedParams = new URLSearchParams({ - "test-method": "queue-received", - }); - const receivedResponse = await fetch( - `${workerEntrypoint}?${receivedParams}` + expect(await response.json()).toEqual({ + worker: "exported-handler", + tailEvents: expect.arrayContaining([ + [["[Worker Entrypoint]"], ["hello from test"]], + [["[Worker Entrypoint]"], ["yet another log", "and another one"]], + ]), + }); + }, waitForTimeout); + }); + + it("supports queues across dev sessions", async ({ devRegistryPath }) => { + const exportedHandler = await runViteDev( + "vite.exported-handler.config.ts", + devRegistryPath ); - expect(await receivedResponse.json()).toContain( - "hello from vite producer" + const workerEntrypoint = await runViteDev( + "vite.worker-entrypoint.config.ts", + devRegistryPath ); - }, waitForTimeout); + + await vi.waitFor(async () => { + const sendParams = new URLSearchParams({ + "test-method": "queue-send", + }); + const sendResponse = await fetch(`${exportedHandler}?${sendParams}`, { + method: "POST", + body: "hello from vite producer", + }); + expect(await sendResponse.text()).toBe("Queued"); + expect(sendResponse.status).toBe(200); + + const receivedParams = new URLSearchParams({ + "test-method": "queue-received", + }); + const receivedResponse = await fetch( + `${workerEntrypoint}?${receivedParams}` + ); + expect(await receivedResponse.json()).toContain( + "hello from vite producer" + ); + }, waitForTimeout); + }); }); -}); +} describe("Dev Registry: vite dev <-> wrangler dev", () => { it("uses the same dev registry path by default", async () => { diff --git a/packages/miniflare/src/index.ts b/packages/miniflare/src/index.ts index 87314a6b275..eab29cc8882 100644 --- a/packages/miniflare/src/index.ts +++ b/packages/miniflare/src/index.ts @@ -2684,6 +2684,9 @@ export class Miniflare { this.#log.warn( "Debug port not available — skipping dev registry registration" ); + // Nothing we advertised before is reachable, so withdraw it rather than + // leaving peers pointed at an address we can no longer serve. + this.#devRegistry.register({}); return; } const debugPortAddress = `127.0.0.1:${debugPort}`; diff --git a/packages/miniflare/src/shared/dev-registry.ts b/packages/miniflare/src/shared/dev-registry.ts index c74995ae68d..4fdb6c0cc80 100644 --- a/packages/miniflare/src/shared/dev-registry.ts +++ b/packages/miniflare/src/shared/dev-registry.ts @@ -137,11 +137,17 @@ export class DevRegistry { registryPath: string | undefined, onUpdate?: (registry: WorkerRegistry) => void ): Promise { - // Unregister all registered workers - this.unregisterWorkers(); this.onUpdate = onUpdate; if (registryPath !== this.registryPath) { + // Our entries live in the directory we are leaving, so they have to be + // removed from there before we switch. When the path is unchanged we + // deliberately keep them: `register()` reconciles the set instead, so a + // config update no longer deletes and recreates every entry. Other dev + // sessions watch this directory, and a deletion is visible to them even + // if we put the file straight back. + this.unregisterWorkers(); + // Close the existing watcher if it exists. // It will watch the new path if there is any dependent services in a later step await this.watcher?.close(); @@ -159,6 +165,17 @@ export class DevRegistry { // Make sure the registry path exists mkdirSync(this.registryPath, { recursive: true }); + // Drop the entries for Workers this instance no longer has. Workers that + // remain are overwritten in place below instead of being deleted and + // recreated, so a peer never observes one of its service binding or + // `tail_consumers` targets disappearing during a routine config update. + for (const name of [...this.registeredWorkers]) { + if (!(name in workers)) { + this.unregister(name); + this.registeredWorkers.delete(name); + } + } + for (const [name, definition] of Object.entries(workers)) { const definitionPath = path.join(this.registryPath, name); const existingHeartbeat = this.heartbeats.get(name); diff --git a/packages/miniflare/test/dev-registry.spec.ts b/packages/miniflare/test/dev-registry.spec.ts index 5f7714e07f6..e169908fb58 100644 --- a/packages/miniflare/test/dev-registry.spec.ts +++ b/packages/miniflare/test/dev-registry.spec.ts @@ -1,3 +1,5 @@ +import path from "node:path"; +import { watch } from "chokidar"; import { getWorkerRegistry, Miniflare } from "miniflare"; import { describe, onTestFinished, test, vi } from "vitest"; import { useDispose, useTmp } from "./test-shared"; @@ -1788,3 +1790,108 @@ describe.sequential("DevRegistry", () => { ); }); }); + +describe("registry churn across config updates", () => { + const script = (body: string) => + `export default { async fetch() { return new Response("${body}"); } }`; + + test("does not withdraw a Worker's entry during an unrelated config update", async ({ + expect, + }) => { + const unsafeDevRegistryPath = await useTmp(); + const mf = new Miniflare({ + name: "stable-worker", + unsafeDevRegistryPath, + modules: true, + script: script("before"), + }); + useDispose(mf); + await mf.ready; + + await vi.waitFor( + () => { + expect( + getWorkerRegistry(unsafeDevRegistryPath)["stable-worker"] + ).toBeDefined(); + }, + { timeout: 10_000, interval: 100 } + ); + + // Other dev sessions learn about us by watching this directory, so watch it + // the same way and record what a peer would actually see. + const events: string[] = []; + const watcher = watch(unsafeDevRegistryPath, { + ignoreInitial: true, + }); + onTestFinished(() => watcher.close()); + await new Promise((resolve) => watcher.once("ready", resolve)); + watcher.on("unlink", (file) => + events.push(`unlink:${path.basename(file)}`) + ); + watcher.on("add", (file) => events.push(`add:${path.basename(file)}`)); + watcher.on("change", (file) => + events.push(`change:${path.basename(file)}`) + ); + + await mf.setOptions({ + name: "stable-worker", + unsafeDevRegistryPath, + modules: true, + script: script("after"), + }); + + // Wait until the update has definitely reached the directory, so that an + // empty event list can't be mistaken for a passing assertion. + await vi.waitFor( + () => { + expect(events.some((event) => event.endsWith(":stable-worker"))).toBe( + true + ); + }, + { timeout: 10_000, interval: 100 } + ); + + expect(events).not.toContain("unlink:stable-worker"); + expect( + getWorkerRegistry(unsafeDevRegistryPath)["stable-worker"] + ).toBeDefined(); + }); + + test("withdraws the entry for a Worker removed from the config", async ({ + expect, + }) => { + const unsafeDevRegistryPath = await useTmp(); + const mf = new Miniflare({ + unsafeDevRegistryPath, + workers: [ + { name: "kept-worker", modules: true, script: script("kept") }, + { name: "dropped-worker", modules: true, script: script("dropped") }, + ], + }); + useDispose(mf); + await mf.ready; + + await vi.waitFor( + () => { + const registry = getWorkerRegistry(unsafeDevRegistryPath); + expect(registry["kept-worker"]).toBeDefined(); + expect(registry["dropped-worker"]).toBeDefined(); + }, + { timeout: 10_000, interval: 100 } + ); + + await mf.setOptions({ + unsafeDevRegistryPath, + workers: [{ name: "kept-worker", modules: true, script: script("kept") }], + }); + + await vi.waitFor( + () => { + const registry = getWorkerRegistry(unsafeDevRegistryPath); + expect(registry["kept-worker"]).toBeDefined(); + expect(registry["dropped-worker"]).toBeUndefined(); + }, + { timeout: 10_000, interval: 100 } + ); + }); +});