1import * as Effect from "effect/Effect";
2import { createServer, type Server } from "node:http";
3import { spawn } from "node:child_process";
4
5export interface OtlpCollector {
6 readonly server: Server;
7 /** Base URL (`http://127.0.0.1:<port>`); append the OTLP signal path. */
8 readonly url: string;
9 /** Responses fully written to the client. */
10 readonly completedRequests: { value: number };
11 /** Received batches, including exports whose client disconnected. */
12 readonly requests: Array<{
13 body: string;
14 receivedAt: number;
15 responseStartedAt?: number;
16 closedAt?: number;
17 completed: boolean;
18 aborted: boolean;
19 }>;
20 /** Re-arm the response gate for subsequent matching batches. */
21 readonly holdResponses: () => void;
22 /** Acknowledge held batches and leave the gate open until re-armed. */
23 readonly releaseResponses: () => void;
24}
25
26/**
27 * Starts a local OTLP endpoint that records exports and completed responses.
28 * Matching `holdResponse` batches wait for explicit release, so tests can
29 * assert response/export ordering without timer races. The Node server is
30 * a test adapter only.
31 */
32export const startOtlpCollector = (
33 options: {
34 holdResponse?: (body: string) => boolean;
35 } = {},
36) =>
37 Effect.acquireRelease(
38 Effect.callback<OtlpCollector, Error>((resume) => {
39 const completedRequests = { value: 0 };
40 const requests: OtlpCollector["requests"] = [];
41 const pending = new Set<() => void>();
42 let held = true;
43 const holdResponses = () => {
44 held = true;
45 };
46 const releaseResponses = () => {
47 held = false;
48 for (const send of pending) send();
49 pending.clear();
50 };
51 const server = createServer((request, response) => {
52 let body = "";
53 request.setEncoding("utf8");
54 request.on("data", (chunk: string) => {
55 body += chunk;
56 });
57 request.once("end", () => {
58 const batch: OtlpCollector["requests"][number] = {
59 body,
60 receivedAt: Date.now(),
61 completed: false,
62 aborted: false,
63 };
64 requests.push(batch);
65 const send = () => {
66 batch.responseStartedAt = Date.now();
67 response.writeHead(200, { "content-type": "application/json" });
68 response.end('{"partialSuccess":{}}');
69 };
70 response.once("finish", () => {
71 batch.completed = true;
72 completedRequests.value += 1;
73 });
74 response.once("close", () => {
75 batch.closedAt = Date.now();
76 batch.aborted = !batch.completed;
77 pending.delete(send);
78 });
79 if (held && options.holdResponse?.(body)) {
80 pending.add(send);
81 } else {
82 send();
83 }
84 });
85 });
86 const onError = (error: Error) => resume(Effect.fail(error));
87 server.once("error", onError);
88 server.listen(0, "127.0.0.1", () => {
89 server.off("error", onError);
90 const address = server.address();
91 if (address === null || typeof address === "string") {
92 resume(
93 Effect.fail(new Error("OTLP test collector address unavailable")),
94 );
95 return;
96 }
97 resume(
98 Effect.succeed({
99 server,
100 completedRequests,
101 requests,
102 holdResponses,
103 releaseResponses,
104 url: `http://127.0.0.1:${address.port}`,
105 }),
106 );
107 });
108 return Effect.sync(() => server.close());
109 }),
110 ({ server, releaseResponses }) =>
111 Effect.callback<void, Error>((resume) => {
112 releaseResponses();
113 server.close((error) =>
114 resume(error === undefined ? Effect.void : Effect.fail(error)),
115 );
116 }).pipe(Effect.orDie),
117 );
118
119/** Node's native HTTP close events expose cancellations that Bun's adapter omits. */
120export const startDelayedOtlpCollector = () =>
121 Effect.acquireRelease(
122 Effect.callback<
123 Pick<OtlpCollector, "url" | "requests" | "completedRequests"> & {
124 stop: () => void;
125 },
126 Error
127 >((resume) => {
128 const requests: OtlpCollector["requests"] = [];
129 const completedRequests = { value: 0 };
130 const child = spawn(
131 "node",
132 [
133 "-e",
134 `
135 const { createServer } = require("node:http");
136 const emit = (event) => process.stdout.write(JSON.stringify(event) + "\\n");
137 let nextId = 0;
138 const server = createServer((request, response) => {
139 let body = "";
140 request.setEncoding("utf8");
141 request.on("data", (chunk) => { body += chunk; });
142 request.once("end", () => {
143 const id = nextId++;
144 let completed = false;
145 emit({ type: "received", id, body, at: Date.now() });
146 const send = () => {
147 emit({ type: "responding", id, at: Date.now() });
148 response.writeHead(200, { "content-type": "application/json" });
149 response.end('{"partialSuccess":{}}');
150 };
151 const delayed = body.includes('"name":"otel-event-flush.rpc"');
152 const timer = delayed ? setTimeout(send, 4000) : undefined;
153 response.once("finish", () => {
154 completed = true;
155 emit({ type: "finished", id, at: Date.now() });
156 });
157 response.once("close", () => {
158 clearTimeout(timer);
159 emit({ type: "closed", id, at: Date.now(), aborted: !completed });
160 });
161 if (!delayed) send();
162 });
163 });
164 server.listen(0, "127.0.0.1", () => emit({ type: "ready", port: server.address().port }));
165 `,
166 ],
167 { stdio: ["ignore", "pipe", "pipe"] },
168 );
169 const stop = () => {
170 child.kill();
171 };
172 child.once("error", (error) => resume(Effect.fail(error)));
173 child.stderr.on("data", (data) => console.error(String(data)));
174 let pending = "";
175 child.stdout.on("data", (data) => {
176 pending += String(data);
177 let newline: number;
178 while ((newline = pending.indexOf("\n")) !== -1) {
179 const message = JSON.parse(pending.slice(0, newline)) as
180 | { type: "ready"; port: number }
181 | { type: "received"; id: number; body: string; at: number }
182 | { type: "responding" | "finished"; id: number; at: number }
183 | { type: "closed"; id: number; at: number; aborted: boolean };
184 pending = pending.slice(newline + 1);
185 if (message.type === "ready") {
186 resume(
187 Effect.succeed({
188 url: `http://127.0.0.1:${message.port}`,
189 requests,
190 completedRequests,
191 stop,
192 }),
193 );
194 } else if (message.type === "received") {
195 requests[message.id] = {
196 body: message.body,
197 receivedAt: message.at,
198 completed: false,
199 aborted: false,
200 };
201 } else {
202 const batch = requests[message.id]!;
203 if (message.type === "responding")
204 batch.responseStartedAt = message.at;
205 else if (message.type === "finished") {
206 batch.completed = true;
207 completedRequests.value += 1;
208 } else if (message.type === "closed") {
209 batch.closedAt = message.at;
210 batch.aborted = message.aborted;
211 }
212 }
213 }
214 });
215 return Effect.sync(stop);
216 }).pipe(Effect.timeout("5 seconds")),
217 ({ stop }) => Effect.sync(stop),
218 );