Support Tail Workers in local dev (#9083)

* Tail Workers support in local ded

* tests

* fix snapshots

* Create wild-boxes-invite.md

* Apply suggestions from code review

Co-authored-by: Carmen Popoviciu <cpopoviciu@cloudflare.com>

* fix snapshots

* tails -> tailConsumers

* Update packages/wrangler/e2e/__snapshots__/pages-dev.test.ts.snap

Co-authored-by: Carmen Popoviciu <cpopoviciu@cloudflare.com>

* fix snapshots

* fix snapshots

---------

Co-authored-by: Carmen Popoviciu <cpopoviciu@cloudflare.com>

Author: penalosa

.changeset/wild-boxes-invite.md

@@ -0,0 +1,6 @@
+---
+"miniflare": minor
+"wrangler": minor
+---
+
+Support Tail Workers in local dev

fixtures/worker-app/src/index.js

@@ -101,4 +101,7 @@ export default {
 		ctx.waitUntil(Promise.resolve(event.scheduledTime));
 		ctx.waitUntil(Promise.resolve(event.cron));
 	},
+	tail(events) {
+		console.log("tails", { events });
+	},
 };

packages/miniflare/src/plugins/core/index.ts

@@ -155,6 +155,8 @@ const CoreOptionsSchemaInput = z.intersection(
 		 Router Worker)
 		 */
 		hasAssetsAndIsVitest: z.boolean().optional(),
+
+		tails: z.array(ServiceDesignatorSchema).optional(),
 	})
 );
 export const CoreOptionsSchema = CoreOptionsSchemaInput.transform((value) => {
@@ -590,6 +592,7 @@ export const CORE_PLUGIN: Plugin<
 
 		const services: Service[] = [];
 		const extensions: Extension[] = [];
+
 		if (isWrappedBinding) {
 			const stringName = JSON.stringify(name);
 			function invalidWrapped(reason: string): never {
@@ -700,6 +703,19 @@ export const CORE_PLUGIN: Plugin<
 						sharedOptions.unsafeModuleFallbackService !== undefined
 							? `localhost:${loopbackPort}`
 							: undefined,
+					tails:
+						options.tails === undefined
+							? undefined
+							: options.tails.map<ServiceDesignator>((service) => {
+									return getCustomServiceDesignator(
+										/* referrer */ options.name,
+										workerIndex,
+										CustomServiceKind.UNKNOWN,
+										name,
+										service,
+										options.hasAssetsAndIsVitest
+									);
+								}),
 				},
 			});
 		}
@@ -716,6 +732,18 @@ export const CORE_PLUGIN: Plugin<
 				if (maybeService !== undefined) services.push(maybeService);
 			}
 		}
+
+		if (options.tails !== undefined) {
+			for (const service of options.tails) {
+				const maybeService = maybeGetCustomServiceService(
+					workerIndex,
+					CustomServiceKind.UNKNOWN,
+					name,
+					service
+				);
+				if (maybeService !== undefined) services.push(maybeService);
+			}
+		}
 		if (options.outboundService !== undefined) {
 			const maybeService = maybeGetCustomServiceService(
 				workerIndex,

packages/miniflare/src/runtime/config/workerd.ts

@@ -63,6 +63,7 @@ export type Worker = (
 	durableObjectUniqueKeyModifier?: string;
 	durableObjectStorage?: Worker_DurableObjectStorage;
 	moduleFallback?: string;
+	tails?: ServiceDesignator[];
 };
 
 export type Worker_DurableObjectStorage =

packages/miniflare/test/index.spec.ts

@@ -921,6 +921,58 @@ test("Miniflare: service binding to named entrypoint that implements a method re
 	t.deepEqual(rpcTarget.id, "test-id");
 });
 
+test("Miniflare: tail consumer called", async (t) => {
+	const mf = new Miniflare({
+		workers: [
+			{
+				name: "a",
+				tails: ["b"],
+				compatibilityDate: "2025-04-28",
+				modules: true,
+				script: `
+
+				export default {
+					async fetch(request, env) {
+						if(request.url.includes("b")) { return env.B.fetch(request)}
+						console.log("log event")
+
+						return new Response("hello from a");
+					}
+				}
+				`,
+				serviceBindings: {
+					B: "b",
+				},
+			},
+			{
+				name: "b",
+				modules: true,
+				compatibilityDate: "2025-04-28",
+
+				script: `
+				let event;
+				export default {
+					fetch() {return Response.json(event)},
+					tail(e) {event = e }
+				};
+				`,
+			},
+		],
+	});
+	t.teardown(() => mf.dispose());
+
+	const res = await mf.dispatchFetch("http://placeholder");
+	t.deepEqual(await res.text(), "hello from a");
+	t.deepEqual(
+		(
+			(await (await mf.dispatchFetch("http://placeholder/b")).json()) as {
+				logs: { message: string[] }[];
+			}[]
+		)[0].logs[0].message,
+		["log event"]
+	);
+});
+
 test("Miniflare: custom outbound service", async (t) => {
 	const mf = new Miniflare({
 		workers: [

packages/wrangler/e2e/__snapshots__/pages-dev.test.ts.snap

@@ -3,7 +3,7 @@
 exports[`Pages 'wrangler pages dev' > should merge (with override) \`wrangler.toml\` configuration with configuration provided via the command line, with command line args taking precedence 1`] = `
 "✨ Compiled Worker successfully
 Your Worker and resources are simulated locally via Miniflare. For more information, see: https://developers.cloudflare.com/workers/testing/local-development.
-Your worker has access to the following bindings:
+Your Worker has access to the following bindings:
 - Durable Objects:
   - DO_BINDING_1_TOML: NEW_DO_1 (defined in NEW_DO_SCRIPT_1 [not connected])
   - DO_BINDING_2_TOML: DO_2_TOML (defined in DO_SCRIPT_2_TOML [not connected])
@@ -30,7 +30,7 @@ Your worker has access to the following bindings:
   - VAR1: "(hidden)"
   - VAR2: "VAR_2_TOML"
   - VAR3: "(hidden)"
-Service bindings & durable object bindings connect to other \`wrangler dev\` processes running locally, with their connection status indicated by [connected] or [not connected]. For more details, refer to https://developers.cloudflare.com/workers/runtime-apis/bindings/service-bindings/#local-development
+Service bindings, Durable Object bindings, and Tail consumers connect to other \`wrangler dev\` processes running locally, with their connection status indicated by [connected] or [not connected]. For more details, refer to https://developers.cloudflare.com/workers/runtime-apis/bindings/service-bindings/#local-development
 ▲ [WARNING] Using Workers AI always accesses your Cloudflare account in order to run AI models, and so will incur usage charges even in local development.
 "
 `;

packages/wrangler/e2e/deployments.test.ts

@@ -373,7 +373,7 @@ Uploaded 2 of 3 assets
 Uploaded 3 of 3 assets
 ✨ Success! Uploaded 3 files (TIMINGS)
 Total Upload: xx KiB / gzip: xx KiB
-Your worker has access to the following bindings:
+Your Worker has access to the following bindings:
 - Assets:
   - Binding: ASSETS
 Uploaded tmp-e2e-worker-00000000-0000-0000-0000-000000000000 (TIMINGS)
@@ -540,7 +540,7 @@ Uploaded 2 of 3 assets
 Uploaded 3 of 3 assets
 ✨ Success! Uploaded 3 files (TIMINGS)
 Total Upload: xx KiB / gzip: xx KiB
-Your worker has access to the following bindings:
+Your Worker has access to the following bindings:
 - Assets:
   - Binding: ASSETS
 Uploaded tmp-e2e-worker-00000000-0000-0000-0000-000000000000 (TIMINGS)
@@ -662,7 +662,7 @@ Current Version ID: 00000000-0000-0000-0000-000000000000`);
 			);
 			normalizedStdout = normalize(output.stdout);
 			expect(normalizedStdout).toEqual(`Total Upload: xx KiB / gzip: xx KiB
-Your worker has access to the following bindings:
+Your Worker has access to the following bindings:
 - Dispatch Namespaces:
   - DISPATCH: tmp-e2e-dispatch-00000000-0000-0000-0000-000000000000
 Uploaded tmp-e2e-worker-00000000-0000-0000-0000-000000000000 (TIMINGS)
@@ -762,7 +762,7 @@ Uploaded 2 of 3 assets
 Uploaded 3 of 3 assets
 ✨ Success! Uploaded 3 files (TIMINGS)
 Total Upload: xx KiB / gzip: xx KiB
-Your worker has access to the following bindings:
+Your Worker has access to the following bindings:
 - Assets:
   - Binding: ASSETS
 Uploaded tmp-e2e-worker-00000000-0000-0000-0000-000000000000 (TIMINGS)
@@ -775,7 +775,7 @@ Current Version ID: 00000000-0000-0000-0000-000000000000`);
 			);
 			normalizedStdout = normalize(output.stdout);
 			expect(normalizedStdout).toEqual(`Total Upload: xx KiB / gzip: xx KiB
-Your worker has access to the following bindings:
+Your Worker has access to the following bindings:
 - Dispatch Namespaces:
   - DISPATCH: tmp-e2e-dispatch-00000000-0000-0000-0000-000000000000
 Uploaded tmp-e2e-worker-00000000-0000-0000-0000-000000000000 (TIMINGS)
@@ -873,7 +873,7 @@ Uploaded 2 of 3 assets
 Uploaded 3 of 3 assets
 ✨ Success! Uploaded 3 files (TIMINGS)
 Total Upload: xx KiB / gzip: xx KiB
-Your worker has access to the following bindings:
+Your Worker has access to the following bindings:
 - Assets:
   - Binding: ASSETS
 Uploaded tmp-e2e-worker-00000000-0000-0000-0000-000000000000 (TIMINGS)
@@ -886,7 +886,7 @@ Current Version ID: 00000000-0000-0000-0000-000000000000`);
 			);
 			normalizedStdout = normalize(output.stdout);
 			expect(normalizedStdout).toEqual(`Total Upload: xx KiB / gzip: xx KiB
-Your worker has access to the following bindings:
+Your Worker has access to the following bindings:
 - Dispatch Namespaces:
   - DISPATCH: tmp-e2e-dispatch-00000000-0000-0000-0000-000000000000
 Uploaded tmp-e2e-worker-00000000-0000-0000-0000-000000000000 (TIMINGS)

packages/wrangler/e2e/dev-registry.test.ts

@@ -302,7 +302,7 @@ describe.each([{ cmd: "wrangler dev" }])("dev registry $cmd", ({ cmd }) => {
 			);
 
 			expect(normalizeOutput(workerA.currentOutput)).toContain(
-				"bindings connect to other `wrangler dev` processes running locally"
+				"connect to other `wrangler dev` processes running locally"
 			);
 		});
 
@@ -375,6 +375,69 @@ describe.each([{ cmd: "wrangler dev" }])("dev registry $cmd", ({ cmd }) => {
 		);
 	});
 
+	describe("Tail consumers", () => {
+		beforeEach(async () => {
+			await baseSeed(a, {
+				"wrangler.toml": dedent`
+							name = "${workerName}"
+							main = "src/index.ts"
+							compatibility_date = "2025-04-28"
+
+							[[tail_consumers]]
+							service = "${workerName2}"
+					`,
+				"src/index.ts": dedent/* javascript */ `
+						export default {
+							async fetch(req, env) {
+								console.log("log something")
+								return new Response("hello from a")
+							},
+						};
+						`,
+			});
+
+			b = await makeRoot();
+			await baseSeed(b, {
+				"wrangler.toml": dedent`
+							name = "${workerName2}"
+							main = "src/index.ts"
+							compatibility_date = "2025-04-28"
+					`,
+				"src/index.ts": dedent/* javascript */ `
+						export default {
+							async tail(event) {
+								console.log("received tail event", event)
+							},
+						};
+					`,
+			});
+		});
+
+		it("can fetch a without b running", async () => {
+			const workerA = helper.runLongLived(cmd, { cwd: a });
+			const { url } = await workerA.waitForReady(5_000);
+
+			await expect(fetchText(`${url}`)).resolves.toBe("hello from a");
+		});
+
+		it("tail event sent to b", async () => {
+			const workerA = helper.runLongLived(cmd, { cwd: a });
+			const { url } = await workerA.waitForReady(5_000);
+
+			const workerB = helper.runLongLived(cmd, { cwd: b });
+
+			await workerA.readUntil(/connected/);
+
+			await expect(fetchText(`${url}`)).resolves.toBe("hello from a");
+
+			await vi.waitFor(
+				async () =>
+					expect(workerB.currentOutput).includes("received tail event"),
+				{ interval: 1000, timeout: 10_000 }
+			);
+		});
+	});
+
 	describe("durable objects", () => {
 		beforeEach(async () => {
 			await baseSeed(a, {
@@ -526,7 +589,7 @@ describe.each([{ cmd: "wrangler dev" }])("dev registry $cmd", ({ cmd }) => {
 			);
 
 			expect(normalizeOutput(workerA.currentOutput)).toContain(
-				"bindings connect to other `wrangler dev` processes running locally"
+				"connect to other `wrangler dev` processes running locally"
 			);
 		});
 

packages/wrangler/e2e/multiworker-dev.test.ts

@@ -389,6 +389,69 @@ describe("multiworker", () => {
 		});
 	});
 
+	describe("Tail consumers", () => {
+		beforeEach(async () => {
+			await baseSeed(a, {
+				"wrangler.toml": dedent`
+						name = "${workerName}"
+						main = "src/index.ts"
+						compatibility_date = "2025-04-28"
+
+						[[tail_consumers]]
+						service = "${workerName2}"
+				`,
+				"src/index.ts": dedent/* javascript */ `
+					export default {
+						async fetch(req, env) {
+							console.log("log something")
+							return new Response("hello from a")
+						},
+					};
+					`,
+			});
+
+			b = await makeRoot();
+			await baseSeed(b, {
+				"wrangler.toml": dedent`
+						name = "${workerName2}"
+						main = "src/index.ts"
+						compatibility_date = "2025-04-28"
+				`,
+				"src/index.ts": dedent/* javascript */ `
+					export default {
+						async tail(event) {
+							console.log("received tail event", event)
+						},
+					};
+				`,
+			});
+		});
+
+		it("can fetch a without b running", async () => {
+			const worker = helper.runLongLived(`wrangler dev`, { cwd: a });
+
+			const { url } = await worker.waitForReady(5_000);
+
+			await expect(fetchText(`${url}`)).resolves.toBe("hello from a");
+		});
+
+		it("tail event sent to b", async () => {
+			const worker = helper.runLongLived(
+				`wrangler dev -c wrangler.toml -c ${b}/wrangler.toml`,
+				{ cwd: a }
+			);
+			const { url } = await worker.waitForReady(5_000);
+
+			await expect(fetchText(`${url}`)).resolves.toBe("hello from a");
+
+			await vi.waitFor(
+				async () =>
+					expect(worker.currentOutput).includes("received tail event"),
+				{ interval: 1000, timeout: 10_000 }
+			);
+		});
+	});
+
 	describe("pages", () => {
 		beforeEach(async () => {
 			await baseSeed(a, {

packages/wrangler/e2e/pages-dev.test.ts

@@ -119,7 +119,7 @@ describe.sequential.each([{ cmd: "wrangler pages dev" }])(
 			);
 			await worker.waitForReady();
 			expect(normalizeOutput(worker.currentOutput)).toContain(
-				dedent`Your worker has access to the following bindings:
+				dedent`Your Worker has access to the following bindings:
 					- Durable Objects:
 					  - TEST_DO: TestDurableObject (defined in a [not connected])
 					- KV Namespaces:
@@ -325,7 +325,7 @@ describe.sequential.each([{ cmd: "wrangler pages dev" }])(
 
 			expect(text).toBe("⚡️ Pages ⚡️ supports wrangler.toml");
 			expect(normalizeOutput(worker.currentOutput)).toContain(
-				dedent`Your worker has access to the following bindings:
+				dedent`Your Worker has access to the following bindings:
 					- KV Namespaces:
 					  - KV_BINDING_TOML: KV_ID_TOML [simulated locally]
 					- Vars: