| @@ -22,6 +22,7 @@ · 21 unmodified lines |
| 22 | 22 | import { expect, layer } from "@effect/vitest"; |
| 23 | 23 | import { resolve } from "node:path"; |
| 24 | 24 | import { createServer } from "node:http"; |
| 25 | +import type { AddressInfo } from "node:net"; |
| 25 | 26 | |
| 26 | 27 | import { |
| 27 | 28 | HttpApi, |
| @@ -90,6 +91,23 @@ class Worker extends Context.Tag("MiniflareE2E/Worker")< · 62 unmodified lines |
| 90 | 91 | } |
| 91 | 92 | >() {} |
| 92 | 93 | |
| 94 | +type CapturedSpan = { |
| 95 | + readonly name: string; |
| 96 | + readonly attributes: Record<string, unknown>; |
| 97 | +}; |
| 98 | + |
| 99 | +class TelemetryReceiver extends Context.Tag("MiniflareE2E/TelemetryReceiver")< |
| 100 | + TelemetryReceiver, |
| 101 | + { |
| 102 | + readonly tracesUrl: string; |
| 103 | + readonly spans: () => ReadonlyArray<CapturedSpan>; |
| 104 | + readonly waitForSpan: ( |
| 105 | + predicate: (span: CapturedSpan) => boolean, |
| 106 | + timeoutMs?: number, |
| 107 | + ) => Promise<CapturedSpan>; |
| 108 | + } |
| 109 | +>() {} |
| 110 | + |
| 93 | 111 | const UpstreamLive = Layer.effect( |
| 94 | 112 | Upstream, |
| 95 | 113 | Effect.gen(function* () { |
| @@ -107,36 +125,160 @@ const UpstreamLive = Layer.effect( · 11 unmodified lines |
| 107 | 125 | }), |
| 108 | 126 | ).pipe(Layer.provide(UpstreamServeLayer)); |
| 109 | 127 | |
| 110 | | −const WorkerLive = Layer.scoped( |
| 111 | | − Worker, |
| 128 | +// --------------------------------------------------------------------------- |
| 129 | +// Telemetry receiver — a node HTTP server on a random port that speaks |
| 130 | +// OTLP/JSON. The Effect OTLPTraceExporter in `services/telemetry.ts` |
| 131 | +// posts JSON bodies to it (confirmed via |
| 132 | +// `@opentelemetry/exporter-trace-otlp-http` — `Content-Type: |
| 133 | +// application/json` + `JsonTraceSerializer`). We parse resourceSpans → |
| 134 | +// scopeSpans → spans → attributes so tests can assert the DO actually |
| 135 | +// reported the expected spans, not just that the exporter was called. |
| 136 | +// --------------------------------------------------------------------------- |
| 137 | + |
| 138 | +type OtlpAttributeValue = { |
| 139 | + readonly stringValue?: string; |
| 140 | + readonly intValue?: string | number; |
| 141 | + readonly doubleValue?: number; |
| 142 | + readonly boolValue?: boolean; |
| 143 | +}; |
| 144 | +type OtlpAttribute = { readonly key: string; readonly value?: OtlpAttributeValue }; |
| 145 | +type OtlpSpan = { readonly name: string; readonly attributes?: ReadonlyArray<OtlpAttribute> }; |
| 146 | +type OtlpPayload = { |
| 147 | + readonly resourceSpans?: ReadonlyArray<{ |
| 148 | + readonly scopeSpans?: ReadonlyArray<{ readonly spans?: ReadonlyArray<OtlpSpan> }>; |
| 149 | + }>; |
| 150 | +}; |
| 151 | + |
| 152 | +const unwrapAttrValue = (v?: OtlpAttributeValue): unknown => { |
| 153 | + if (!v) return undefined; |
| 154 | + if (v.stringValue !== undefined) return v.stringValue; |
| 155 | + if (v.intValue !== undefined) return Number(v.intValue); |
| 156 | + if (v.doubleValue !== undefined) return v.doubleValue; |
| 157 | + if (v.boolValue !== undefined) return v.boolValue; |
| 158 | + return undefined; |
| 159 | +}; |
| 160 | + |
| 161 | +const TelemetryReceiverLive = Layer.scoped( |
| 162 | + TelemetryReceiver, |
| 112 | 163 | Effect.acquireRelease( |
| 113 | | − Effect.promise(() => |
| 114 | | − unstable_dev(resolve(__dirname, "./test-worker.ts"), { |
| 115 | | − config: resolve(__dirname, "../wrangler.miniflare.jsonc"), |
| 116 | | − experimental: { disableExperimentalWarning: true }, |
| 117 | | − ip: "127.0.0.1", |
| 118 | | − logLevel: "info", |
| 119 | | − }), |
| 120 | | − ), |
| 121 | | − (w) => Effect.promise(() => w.stop()), |
| 122 | | − ).pipe( |
| 123 | | − Effect.map((w: Unstable_DevWorker) => ({ |
| 124 | | − baseUrl: new URL(`http://${w.address}:${w.port}`), |
| 125 | | − seedOrg: async (id: string, name: string) => { |
| 126 | | − const res = await w.fetch("/__test__/seed-org", { |
| 127 | | − method: "POST", |
| 128 | | − headers: { "content-type": "application/json" }, |
| 129 | | − body: JSON.stringify({ id, name }), |
| 164 | + Effect.async< |
| 165 | + { |
| 166 | + readonly tracesUrl: string; |
| 167 | + readonly spans: ReadonlyArray<CapturedSpan>; |
| 168 | + readonly store: Array<CapturedSpan>; |
| 169 | + readonly close: () => Promise<void>; |
| 170 | + }, |
| 171 | + never |
| 172 | + >((resume) => { |
| 173 | + const store: Array<CapturedSpan> = []; |
| 174 | + const server = createServer((req, res) => { |
| 175 | + if (req.method !== "POST" || !req.url?.endsWith("/v1/traces")) { |
| 176 | + res.statusCode = 404; |
| 177 | + res.end(); |
| 178 | + return; |
| 179 | + } |
| 180 | + let body = ""; |
| 181 | + req.on("data", (chunk) => { body += chunk; }); |
| 182 | + req.on("end", () => { |
| 183 | + try { |
| 184 | + const payload = JSON.parse(body) as OtlpPayload; |
| 185 | + for (const rs of payload.resourceSpans ?? []) { |
| 186 | + for (const ss of rs.scopeSpans ?? []) { |
| 187 | + for (const sp of ss.spans ?? []) { |
| 188 | + const attrs: Record<string, unknown> = {}; |
| 189 | + for (const a of sp.attributes ?? []) { |
| 190 | + attrs[a.key] = unwrapAttrValue(a.value); |
| 191 | + } |
| 192 | + store.push({ name: sp.name, attributes: attrs }); |
| 193 | + } |
| 194 | + } |
| 195 | + } |
| 196 | + } catch { |
| 197 | + // ignore malformed payloads |
| 198 | + } |
| 199 | + res.writeHead(200, { "content-type": "application/json" }); |
| 200 | + res.end("{}"); |
| 130 | 201 | }); |
| 131 | | − if (res.status !== 204) { |
| 132 | | − throw new Error(`seed-org failed: ${res.status} ${await res.text()}`); |
| 202 | + }); |
| 203 | + server.listen(0, "127.0.0.1", () => { |
| 204 | + const addr = server.address() as AddressInfo; |
| 205 | + resume( |
| 206 | + Effect.succeed({ |
| 207 | + tracesUrl: `http://127.0.0.1:${addr.port}/v1/traces`, |
| 208 | + spans: store, |
| 209 | + store, |
| 210 | + close: () => new Promise<void>((r) => server.close(() => r())), |
| 211 | + }), |
| 212 | + ); |
| 213 | + }); |
| 214 | + }), |
| 215 | + (t) => Effect.promise(() => t.close()), |
| 216 | + ).pipe( |
| 217 | + Effect.map((t) => ({ |
| 218 | + tracesUrl: t.tracesUrl, |
| 219 | + spans: () => [...t.store], |
| 220 | + waitForSpan: async ( |
| 221 | + predicate: (s: CapturedSpan) => boolean, |
| 222 | + timeoutMs = 5_000, |
| 223 | + ) => { |
| 224 | + const deadline = Date.now() + timeoutMs; |
| 225 | + for (;;) { |
| 226 | + const hit = t.store.find(predicate); |
| 227 | + if (hit) return hit; |
| 228 | + if (Date.now() > deadline) { |
| 229 | + throw new Error( |
| 230 | + `Timed out waiting for span. Captured ${t.store.length}: ${t.store.map((s) => s.name).join(", ") || "<none>"}`, |
| 231 | + ); |
| 232 | + } |
| 233 | + await new Promise((r) => setTimeout(r, 50)); |
| 133 | 234 | } |
| 134 | 235 | }, |
| 135 | 236 | })), |
| 136 | 237 | ), |
| 137 | 238 | ); |
| 138 | 239 | |
| 139 | | −const TestEnv = Layer.mergeAll(UpstreamLive, WorkerLive); |
| 240 | +const WorkerLive = Layer.scoped( |
| 241 | + Worker, |
| 242 | + Effect.gen(function* () { |
| 243 | + const receiver = yield* TelemetryReceiver; |
| 244 | + // AXIOM_TOKEN activates DoTelemetryLive inside the worker; AXIOM_TRACES_URL |
| 245 | + // redirects the exporter at our in-process OTLP/JSON receiver so spans |
| 246 | + // become observable in the test process. |
| 247 | + return yield* Effect.acquireRelease( |
| 248 | + Effect.promise(() => |
| 249 | + unstable_dev(resolve(__dirname, "./test-worker.ts"), { |
| 250 | + config: resolve(__dirname, "../wrangler.miniflare.jsonc"), |
| 251 | + experimental: { disableExperimentalWarning: true }, |
| 252 | + ip: "127.0.0.1", |
| 253 | + logLevel: "info", |
| 254 | + vars: { |
| 255 | + AXIOM_TOKEN: "test-token", |
| 256 | + AXIOM_TRACES_URL: receiver.tracesUrl, |
| 257 | + }, |
| 258 | + }), |
| 259 | + ), |
| 260 | + (w) => Effect.promise(() => w.stop()), |
| 261 | + ).pipe( |
| 262 | + Effect.map((w: Unstable_DevWorker) => ({ |
| 263 | + baseUrl: new URL(`http://${w.address}:${w.port}`), |
| 264 | + seedOrg: async (id: string, name: string) => { |
| 265 | + const res = await w.fetch("/__test__/seed-org", { |
| 266 | + method: "POST", |
| 267 | + headers: { "content-type": "application/json" }, |
| 268 | + body: JSON.stringify({ id, name }), |
| 269 | + }); |
| 270 | + if (res.status !== 204) { |
| 271 | + throw new Error(`seed-org failed: ${res.status} ${await res.text()}`); |
| 272 | + } |
| 273 | + }, |
| 274 | + })), |
| 275 | + ); |
| 276 | + }), |
| 277 | +); |
| 278 | + |
| 279 | +const TestEnv = Layer.mergeAll(UpstreamLive, WorkerLive).pipe( |
| 280 | + Layer.provideMerge(TelemetryReceiverLive), |
| 281 | +); |
| 140 | 282 | |
| 141 | 283 | // --------------------------------------------------------------------------- |
| 142 | 284 | // Client helpers |
| @@ -258,4 +400,42 @@ layer(TestEnv, { timeout: 60_000 })("cloud MCP over real HTTP (miniflare)", (it) · 115 unmodified lines |
| 258 | 400 | yield* Effect.promise(() => client.close()); |
| 259 | 401 | }), 30_000, |
| 260 | 402 | ); |
| 403 | + |
| 404 | + it.effect("reports the McpSessionDO.handleRequest span via the OTLP exporter", () => |
| 405 | + Effect.gen(function* () { |
| 406 | + const { baseUrl, seedOrg } = yield* Worker; |
| 407 | + const receiver = yield* TelemetryReceiver; |
| 408 | + const orgId = nextOrgId(); |
| 409 | + yield* Effect.promise(() => seedOrg(orgId, "Telemetry Org")); |
| 410 | + const client = yield* Effect.promise(() => |
| 411 | + connectClient(baseUrl, makeTestBearer(nextAccountId(), orgId)), |
| 412 | + ); |
| 413 | + // Trigger the DO through a multi-step flow so we can assert that |
| 414 | + // handleRequest spans are reported for every DO hit, not just init. |
| 415 | + yield* Effect.promise(() => client.listTools()); |
| 416 | + yield* Effect.promise(() => |
| 417 | + client.callTool({ name: "execute", arguments: { code: "return 1 + 2" } }), |
| 418 | + ); |
| 419 | + yield* Effect.promise(() => client.close()); |
| 420 | + |
| 421 | + // The initialize POST carries no session-id; subsequent requests do. |
| 422 | + // Assert on one of the session-id'd handleRequest spans so we verify |
| 423 | + // attribute propagation beyond the degenerate init case. |
| 424 | + const handleSpan = yield* Effect.promise(() => |
| 425 | + receiver.waitForSpan( |
| 426 | + (s) => |
| 427 | + s.name === "McpSessionDO.handleRequest" && |
| 428 | + s.attributes["mcp.request.session_id_present"] === true, |
| 429 | + ), |
| 430 | + ); |
| 431 | + expect(handleSpan.attributes["mcp.request.method"]).toBeDefined(); |
| 432 | + // 200 for normal POSTs, 202 for notifications/initialized. |
| 433 | + expect([200, 202]).toContain(handleSpan.attributes["mcp.response.status_code"]); |
| 434 | + |
| 435 | + // init runs once per new session and should appear on the initialize POST. |
| 436 | + yield* Effect.promise(() => |
| 437 | + receiver.waitForSpan((s) => s.name === "McpSessionDO.init"), |
| 438 | + ); |
| 439 | + }), 30_000, |
| 440 | + ); |
| 261 | 441 | }); |