1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Test from "@/Test/Alchemy";
3import { expect } from "alchemy-test";
4import * as ConfigProvider from "effect/ConfigProvider";
5import * as Effect from "effect/Effect";
6import * as Fiber from "effect/Fiber";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import OtelEventFlushWorker from "./fixtures/otel-event-flush-worker.ts";
10import {
11 startDelayedOtlpCollector,
12 startOtlpCollector,
13} from "../Utils/OtlpCollector.ts";
14
15const { test } = Test.make({ providers: Cloudflare.providers(), dev: true });
16
17const traceSpanNames = (body: string) => {
18 const payload = JSON.parse(body) as {
19 resourceSpans: { scopeSpans: { spans: { name: string }[] }[] }[];
20 };
21 return payload.resourceSpans.flatMap((resource) =>
22 resource.scopeSpans.flatMap((scope) =>
23 scope.spans.map((span) => span.name),
24 ),
25 );
26};
27
28test.provider(
29 "delivers Worker and Durable Object OTLP batches without delaying the Worker response",
30 (stack) =>
31 Effect.gen(function* () {
32 // Hold only Worker exports: the DO may flush in either foreground or background.
33 const collector = yield* startOtlpCollector({
34 holdResponse: (body) =>
35 body.includes('"name":"otel-event-flush.worker"'),
36 });
37 const currentConfig = yield* ConfigProvider.ConfigProvider;
38 yield* stack.destroy();
39 const deployment = yield* stack
40 .deploy(
41 Effect.gen(function* () {
42 const worker = yield* OtelEventFlushWorker;
43 return { url: worker.url };
44 }),
45 )
46 .pipe(
47 Effect.provideService(
48 ConfigProvider.ConfigProvider,
49 ConfigProvider.orElse(
50 ConfigProvider.fromUnknown({
51 OTLP_EVENT_FLUSH_URL: `${collector.url}/v1/traces`,
52 }),
53 currentConfig,
54 ),
55 ),
56 );
57
58 if (deployment.url === undefined) {
59 return yield* Effect.die(
60 "OTLP event flush test Worker URL unavailable",
61 );
62 }
63 const client = yield* HttpClient.HttpClient;
64 const response = yield* client.get(deployment.url);
65 expect(response.status).toBe(200);
66 expect(yield* response.text).toBe("worker-saw:durable-object-ok");
67
68 // Release only after the response, not on a timer that can outlive
69 // the exporter's shutdown deadline under runner load.
70 expect(collector.completedRequests.value).toBeLessThanOrEqual(1);
71 yield* Effect.sync(collector.releaseResponses);
72
73 // Both batches must be acknowledged before the next event.
74 yield* Effect.sync(() => collector.completedRequests.value).pipe(
75 Effect.repeat({
76 schedule: Schedule.spaced("100 millis"),
77 until: (completed) => completed >= 2,
78 times: 30,
79 }),
80 );
81 expect(collector.completedRequests.value).toBe(2);
82
83 // Same contract for the Durable Object RPC event path: the Worker's
84 // own batch (the 4th) must not be complete at response time.
85 yield* Effect.sync(collector.holdResponses);
86 const rpcResponse = yield* client.get(`${deployment.url}/rpc`);
87 expect(rpcResponse.status).toBe(200);
88 expect(yield* rpcResponse.text).toBe("worker-saw:durable-object-rpc-ok");
89 expect(collector.completedRequests.value).toBeLessThanOrEqual(3);
90 yield* Effect.sync(collector.releaseResponses);
91
92 yield* Effect.sync(() => collector.completedRequests.value).pipe(
93 Effect.repeat({
94 schedule: Schedule.spaced("100 millis"),
95 until: (completed) => completed >= 4,
96 times: 30,
97 }),
98 );
99 expect(collector.completedRequests.value).toBe(4);
100 expect(collector.requests).toHaveLength(4);
101 expect(
102 collector.requests.every((batch) => batch.completed && !batch.aborted),
103 ).toBe(true);
104 for (const [name, count] of [
105 ["otel-event-flush.worker", 2],
106 ["otel-event-flush.child", 1],
107 ["otel-event-flush.rpc", 1],
108 ["http.server GET", 3],
109 ] as const) {
110 expect(
111 collector.requests.filter((batch) =>
112 batch.body.includes(`"name":"${name}"`),
113 ),
114 ).toHaveLength(count);
115 }
116
117 yield* stack.destroy();
118 expect(collector.completedRequests.value).toBe(4);
119 }),
120 {
121 tags: ["provider:cloudflare", "provider:cloudflare:worker", "local"],
122 timeout: 120_000,
123 },
124);
125
126for (const scenario of [
127 { name: "bound default", shutdownTimeout: 0, delivered: true },
128 { name: "custom 3 seconds", shutdownTimeout: 3_000, delivered: false },
129 { name: "custom 10 seconds", shutdownTimeout: 10_000, delivered: true },
130]) {
131 test.provider(
132 `DO RPC exporter deadline with a four-second acknowledgement (${scenario.name})`,
133 (stack) =>
134 Effect.gen(function* () {
135 yield* stack.destroy();
136 const collector = yield* startDelayedOtlpCollector();
137 const currentConfig = yield* ConfigProvider.ConfigProvider;
138 const deployment = yield* stack
139 .deploy(
140 Effect.gen(function* () {
141 const worker = yield* OtelEventFlushWorker;
142 return { url: worker.url };
143 }),
144 )
145 .pipe(
146 Effect.provideService(
147 ConfigProvider.ConfigProvider,
148 ConfigProvider.orElse(
149 ConfigProvider.fromUnknown({
150 OTLP_EVENT_FLUSH_URL: `${collector.url}/v1/traces`,
151 OTLP_EVENT_FLUSH_SHUTDOWN_TIMEOUT: `${scenario.shutdownTimeout} millis`,
152 }),
153 currentConfig,
154 ),
155 ),
156 );
157 if (deployment.url === undefined) {
158 return yield* Effect.fail(
159 new Error("OTLP deadline Worker URL unavailable"),
160 );
161 }
162 const client = yield* HttpClient.HttpClient;
163 const startedAt = yield* Effect.sync(() => Date.now());
164 const responseFiber = yield* Effect.gen(function* () {
165 const response = yield* client.get(`${deployment.url}/rpc`);
166 const body = yield* response.text;
167 const receivedAt = yield* Effect.sync(() => Date.now());
168 return { status: response.status, body, receivedAt };
169 }).pipe(Effect.forkChild());
170
171 const rpcBatch = yield* Effect.sync(() =>
172 collector.requests.find((batch) =>
173 traceSpanNames(batch.body).includes("otel-event-flush.rpc"),
174 ),
175 ).pipe(
176 Effect.repeat({
177 schedule: Schedule.spaced("500 millis"),
178 until: (batch) => batch !== undefined,
179 times: 10,
180 }),
181 );
182 if (rpcBatch === undefined) {
183 const response = yield* Fiber.join(responseFiber).pipe(
184 Effect.timeout("10 seconds"),
185 );
186 yield* Effect.logError("Missing DO RPC export", {
187 scenario,
188 response,
189 batches: collector.requests,
190 });
191 return yield* Effect.fail(
192 new Error("The collector never received the DO RPC batch"),
193 );
194 }
195 const response = yield* Fiber.join(responseFiber).pipe(
196 Effect.timeout("10 seconds"),
197 );
198 expect(response.status).toBe(200);
199 expect(response.body).toBe("worker-saw:durable-object-rpc-ok");
200 yield* Effect.sync(
201 () =>
202 collector.requests.length === 2 &&
203 collector.requests.every((batch) => batch.closedAt !== undefined),
204 ).pipe(
205 Effect.repeat({
206 schedule: Schedule.spaced("1 second"),
207 until: (settled) => settled,
208 times: 10,
209 }),
210 );
211 yield* Effect.log("Native DO RPC exporter deadline", {
212 scenario: scenario.name,
213 workerResponseMillis: response.receivedAt - startedAt,
214 workerResponseAfterRpcReceiptMillis:
215 response.receivedAt - rpcBatch.receivedAt,
216 rpcSocketLifetimeMillis:
217 rpcBatch.closedAt === undefined
218 ? undefined
219 : rpcBatch.closedAt - rpcBatch.receivedAt,
220 batches: collector.requests.map((batch) => ({
221 spans: traceSpanNames(batch.body),
222 receivedAt: batch.receivedAt,
223 responseStartedAt: batch.responseStartedAt,
224 closedAt: batch.closedAt,
225 completed: batch.completed,
226 aborted: batch.aborted,
227 })),
228 });
229 expect(collector.requests).toHaveLength(2);
230 const names = yield* Effect.sync(() =>
231 collector.requests
232 .flatMap((batch) => traceSpanNames(batch.body))
233 .sort(),
234 );
235 expect(names).toEqual([
236 "http.server GET",
237 "otel-event-flush.rpc",
238 "otel-event-flush.worker",
239 ]);
240 expect(rpcBatch.closedAt).toBeTypeOf("number");
241 expect(response.receivedAt).toBeLessThan(rpcBatch.closedAt!);
242 expect(rpcBatch.completed).toBe(scenario.delivered);
243 expect(rpcBatch.aborted).toBe(!scenario.delivered);
244 expect(collector.completedRequests.value).toBe(
245 scenario.delivered ? 2 : 1,
246 );
247 const socketLifetime = rpcBatch.closedAt! - rpcBatch.receivedAt;
248 if (scenario.delivered) {
249 expect(rpcBatch.responseStartedAt).toBeTypeOf("number");
250 expect(response.receivedAt).toBeLessThan(rpcBatch.responseStartedAt!);
251 expect(socketLifetime).toBeGreaterThanOrEqual(3_900);
252 expect(socketLifetime).toBeLessThan(10_000);
253 } else {
254 expect(rpcBatch.responseStartedAt).toBeUndefined();
255 expect(socketLifetime).toBeGreaterThanOrEqual(2_500);
256 expect(socketLifetime).toBeLessThan(4_000);
257 }
258 const workerBatch = collector.requests.find(
259 (batch) => batch !== rpcBatch,
260 )!;
261 expect(workerBatch.completed).toBe(true);
262 expect(workerBatch.aborted).toBe(false);
263 yield* stack.destroy();
264 }).pipe(Effect.scoped),
265 {
266 tags: ["provider:cloudflare", "provider:cloudflare:worker", "local"],
267 timeout: 120_000,
268 },
269 );
270}