1/**
2 * Tests for streaming tail workers (`RuntimeWorker.streamingTails`, the
3 * deployed `streaming_tail_consumers` setting).
4 *
5 * Contract delta from plain tails (`test/Tails.test.ts`): a plain tail
6 * consumer exports `tail(items)` and receives completed `TraceItem`s after
7 * the producer's invocation finishes, while a streaming tail consumer exports
8 * `tailStream(event)`, which is invoked with the invocation's `onset` event
9 * while the producer is still executing and returns a handler that receives
10 * every subsequent event of the session, ending with the terminal `outcome`.
11 *
12 * Topology: every worker runs in its own workerd instance, so
13 * `streamingTails` designators point at the registry proxy's
14 * `ExternalService` entrypoint, whose `tailStream()` forwards the session to
15 * the consumer's process via the workerd debug port (the remote handler comes
16 * back as an RPC stub). A consumer that is not running yet is not a config
17 * error: sessions are dropped with a warning until the consumer registers
18 * itself, which the second test pins.
19 */
20import * as NodeServices from "@effect/platform-node/NodeServices";
21import { describe, expect, it, layer } from "@effect/vitest";
22import * as Effect from "effect/Effect";
23import * as Layer from "effect/Layer";
24import * as Schedule from "effect/Schedule";
25import * as FetchHttpClient from "effect/http/FetchHttpClient";
26import * as Docker from "../Docker.ts";
27import * as Globals from "../globals/Globals.ts";
28import * as Internet from "../globals/Internet.ts";
29import * as Storage from "../globals/Storage.ts";
30import * as Paths from "../internal/Paths.ts";
31import * as Runtime from "../Runtime.ts";
32import * as Workerd from "../workerd/Workerd.ts";
33import {
34 configProvider,
35 localRuntimeLayer,
36 makeTempDirectory,
37 poll,
38 PredicateFailed,
39 startTestWorker,
40 waitForRegistryEntry,
41} from "./helpers/runtime.ts";
42
43/** The shape of a JSON-serialized streaming `TailEvent` as recorded by the consumer. */
44interface RecordedTailEvent {
45 invocationId?: string;
46 timestamp?: string;
47 sequence?: number;
48 event?: {
49 type?: string;
50 outcome?: string;
51 level?: string;
52 message?: unknown;
53 info?: { type?: string; url?: string; method?: string };
54 };
55}
56
57/**
58 * Records every streamed event — the `onset` handed to `tailStream()` plus
59 * every event delivered to the returned handler — and serves them back via
60 * `GET /events`.
61 */
62const CONSUMER_SCRIPT = `
63const events = [];
64export default {
65 fetch(request) {
66 const url = new URL(request.url);
67 if (url.pathname === "/events") return Response.json(events);
68 return new Response("consumer");
69 },
70 tailStream(event) {
71 events.push(event);
72 return (tailEvent) => {
73 events.push(tailEvent);
74 };
75 },
76};
77`;
78
79const producerScript = (marker: string) => `
80export default {
81 fetch() {
82 console.log(${JSON.stringify(marker)});
83 return new Response("hello from producer");
84 },
85};
86`;
87
88const baseWorker = {
89 compatibilityDate: "2026-03-10",
90 compatibilityFlags: [],
91 bindings: [],
92} as const;
93
94const consumerWorker = (name: string) => ({
95 ...baseWorker,
96 name,
97 modules: [
98 { name: "main.js", type: "ESModule", content: CONSUMER_SCRIPT } as const,
99 ],
100});
101
102const producerWorker = (name: string, consumer: string, marker: string) => ({
103 ...baseWorker,
104 name,
105 streamingTails: [consumer],
106 modules: [
107 {
108 name: "main.js",
109 type: "ESModule",
110 content: producerScript(marker),
111 } as const,
112 ],
113});
114
115const eventTypes = (events: ReadonlyArray<RecordedTailEvent>) =>
116 events.map((tailEvent) => tailEvent.event?.type);
117
118layer(localRuntimeLayer, { excludeTestServices: true })(
119 "Streaming tail workers",
120 (it) => {
121 it.effect(
122 "streams onset and outcome events from a producer to its streaming tail consumer",
123 () =>
124 Effect.gen(function* () {
125 const consumer = yield* startTestWorker(
126 consumerWorker("streaming-tails-consumer"),
127 );
128 yield* waitForRegistryEntry({
129 kind: "worker",
130 scriptName: "streaming-tails-consumer",
131 });
132
133 const producer = yield* startTestWorker(
134 producerWorker(
135 "streaming-tails-producer",
136 "streaming-tails-consumer",
137 "streaming-tails-basic-marker",
138 ),
139 );
140 expect(yield* producer.fetchText("/hello")).toBe(
141 "hello from producer",
142 );
143
144 // The outcome event lands after the response completes; poll until
145 // the consumer has recorded a full session (onset through outcome).
146 const events = yield* poll<Array<RecordedTailEvent>>(
147 consumer,
148 "/events",
149 (recorded) => eventTypes(recorded).includes("outcome"),
150 20_000,
151 );
152
153 const onset = events.find(
154 (tailEvent) => tailEvent.event?.type === "onset",
155 );
156 expect(onset).toBeDefined();
157 expect(onset?.event?.info?.type).toBe("fetch");
158 expect(onset?.event?.info?.url).toContain("/hello");
159 expect(onset?.invocationId).toBeTruthy();
160
161 const log = events.find(
162 (tailEvent) => tailEvent.event?.type === "log",
163 );
164 expect(log?.event?.message).toEqual(["streaming-tails-basic-marker"]);
165
166 const outcome = events.find(
167 (tailEvent) => tailEvent.event?.type === "outcome",
168 );
169 expect(outcome?.event?.outcome).toBe("ok");
170
171 // Every streamed event belongs to the same invocation as the onset.
172 for (const tailEvent of events) {
173 expect(tailEvent.invocationId).toBe(onset?.invocationId);
174 }
175 }),
176 { timeout: 60_000 },
177 );
178
179 it.effect(
180 "drops sessions while the consumer is down and streams once it starts",
181 () =>
182 Effect.gen(function* () {
183 // The consumer is not running yet: starting the producer and serving
184 // requests must both succeed — sessions are dropped with a warning.
185 const producer = yield* startTestWorker(
186 producerWorker(
187 "streaming-tails-late-producer",
188 "streaming-tails-late-consumer",
189 "streaming-tails-late-marker",
190 ),
191 );
192 expect(yield* producer.fetchText("/hello")).toBe(
193 "hello from producer",
194 );
195
196 const consumer = yield* startTestWorker(
197 consumerWorker("streaming-tails-late-consumer"),
198 );
199 yield* waitForRegistryEntry({
200 kind: "worker",
201 scriptName: "streaming-tails-late-consumer",
202 });
203
204 // The producer's registry proxy picks the consumer up asynchronously;
205 // keep producing until a full session lands.
206 const events = yield* producer.fetch("/hello").pipe(
207 Effect.andThen(
208 consumer.fetchJson<Array<RecordedTailEvent>>("/events"),
209 ),
210 Effect.flatMap((recorded) =>
211 eventTypes(recorded).includes("outcome")
212 ? Effect.succeed(recorded)
213 : Effect.fail(new PredicateFailed({ value: recorded })),
214 ),
215 Effect.retry({
216 while: (error) => error._tag === "PredicateFailed",
217 schedule: Schedule.spaced("250 millis"),
218 times: 120,
219 }),
220 );
221 const log = events.find(
222 (tailEvent) => tailEvent.event?.type === "log",
223 );
224 expect(log?.event?.message).toEqual(["streaming-tails-late-marker"]);
225 }),
226 { timeout: 90_000 },
227 );
228 },
229);
230
231// A runtime composed without the registry proxy plugin (`streamingTails` has
232// nothing to resolve through) must fail fast with a typed `ConfigError`
233// instead of silently starting a worker whose streaming tail sessions go
234// nowhere. This lives outside the `layer(localRuntimeLayer)` suite:
235// `PluginContext.make` also picks plugins up from the ambient effect context,
236// so the suite's registry proxy would leak into the deliberately
237// registry-free runtime below.
238describe("Streaming tail workers without a registry proxy", () => {
239 const noRegistryRuntimeLayer = Runtime.RuntimeLive.pipe(
240 Layer.provide(Globals.GlobalsLive),
241 Layer.provide(
242 Layer.effect(
243 Storage.Storage,
244 Effect.suspend(() => makeTempDirectory()).pipe(
245 Effect.map(Storage.make),
246 ),
247 ),
248 ),
249 Layer.provide(Internet.InternetLive),
250 Layer.provideMerge(Paths.PathsLive),
251 Layer.provideMerge(Docker.DockerLive),
252 Layer.provide(Workerd.WorkerdLive),
253 Layer.provide(configProvider()),
254 Layer.provideMerge(
255 Layer.mergeAll(NodeServices.layer, FetchHttpClient.layer),
256 ),
257 );
258
259 it.effect(
260 "fails with ConfigError when streamingTails are declared without a registry proxy",
261 () =>
262 Effect.gen(function* () {
263 const runtime = yield* Runtime.Runtime;
264 const error = yield* runtime
265 .start(
266 producerWorker(
267 "streaming-tails-no-registry",
268 "streaming-tails-nowhere",
269 "unused",
270 ),
271 )
272 .pipe(Effect.flip);
273 expect(error._tag).toBe("ConfigError");
274 expect(error.subtag).toBe("PluginNotFound");
275 }).pipe(Effect.provide(noRegistryRuntimeLayer), Effect.scoped),
276 { timeout: 30_000 },
277 );
278});