From 60eadff404547c695e09afd596e91427ecec0faa Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Thu, 17 Sep 2026 20:42:54 +0200 Subject: [PATCH 1/3] fix(telemetry): keep OpenRouter Broadcast traces attached to their Maple session Three ways an OpenRouter mirror trace ended up with no Maple parent in prod: - The chat turn buffered every span in the Durable Object and flushed once, at the end, as background work no request keeps alive. 22 of 101 investigation passes on 2026-09-17 lost that flush entirely; OpenRouter's spans arrived nested under span ids the warehouse never saw. The turn now flushes every 10 s while it runs, and once more before `dispose()` rather than only after. - The SDK's span buffer refused new spans past 10,000, which drops the root that ends last; it now evicts the oldest, and `restore` keeps the newest. - The Slack agent sent `session_id` and `trace_name` but no span ids, so every call was a root trace, and the follow-up gate sent no session at all. The provider now goes through a `fetch` that stamps the active span's ids per request, on both models. --- apps/ai/src/chat/turn-runner.ts | 19 +++++ .../slack-agent/agent/lib/agent-model.test.ts | 13 ++++ apps/slack-agent/agent/lib/agent-model.ts | 5 +- .../agent/lib/follow-up-relevance.ts | 2 + .../agent/lib/openrouter-trace.test.ts | 74 +++++++++++++++++++ .../slack-agent/agent/lib/openrouter-trace.ts | 41 ++++++++++ apps/slack-agent/bun.lock | 1 + apps/slack-agent/package.json | 1 + .../src/shared/flushable-tracer.test.ts | 22 +++++- .../effect-sdk/src/shared/flushable-tracer.ts | 10 ++- 10 files changed, 183 insertions(+), 5 deletions(-) create mode 100644 apps/slack-agent/agent/lib/openrouter-trace.test.ts create mode 100644 apps/slack-agent/agent/lib/openrouter-trace.ts diff --git a/apps/ai/src/chat/turn-runner.ts b/apps/ai/src/chat/turn-runner.ts index 73f5a028c..6e0e9bb6a 100644 --- a/apps/ai/src/chat/turn-runner.ts +++ b/apps/ai/src/chat/turn-runner.ts @@ -54,6 +54,18 @@ const telemetry = MapleCloudflareSDK.make( }), ) +/** + * How often a running turn ships the spans it has finished so far. + * + * The turn is background work on the Durable Object: no request holds the isolate open for it, and + * an investigation has no subscriber at all. A single flush at the end of the turn was therefore + * one fetch that had to survive the object's whole remaining life — and on 2026-09-17 it did not + * for 22 of 101 investigation passes: OpenRouter's Broadcast mirror arrived nested under span ids + * the warehouse never saw, and the Agent Sessions list showed the pass as an OpenRouter-only + * session with no agent. Flushing on an interval bounds a lost flush to the tail of the turn. + */ +const FLUSH_INTERVAL_MS = 10_000 + export interface RunChatSessionTurnInput { /** The Durable Object itself. Appends are direct calls, not stub RPC. */ readonly session: ChatSession @@ -423,6 +435,9 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis }), ) + // `flush` never rejects and serializes overlapping calls, so a tick landing on the final flush + // queues behind it rather than racing it. + const flushTimer = setInterval(() => void telemetry.flush(input.env), FLUSH_INTERVAL_MS) try { await runtime.runPromise(program) } catch { @@ -437,6 +452,10 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis }) } } finally { + clearInterval(flushTimer) + // The turn's own spans have ended by now; ship them before `dispose`, whose finalizers (the + // Postgres connection, the model client) are the one part of the turn that can still hang. + await telemetry.flush(input.env).catch(() => undefined) await runtime.dispose().catch(() => undefined) await telemetry.flush(input.env).catch(() => undefined) } diff --git a/apps/slack-agent/agent/lib/agent-model.test.ts b/apps/slack-agent/agent/lib/agent-model.test.ts index f6e7d7b63..fa5ee2f30 100644 --- a/apps/slack-agent/agent/lib/agent-model.test.ts +++ b/apps/slack-agent/agent/lib/agent-model.test.ts @@ -2,6 +2,7 @@ import { afterEach, describe, expect, test } from "bun:test" import { generateText, type LanguageModel } from "ai" import { agentModel } from "./agent-model.js" import { installFetchStub, type FetchStub } from "./fetch-stub.js" +import { withActiveSpan } from "./openrouter-trace.test.js" type StepStarted = NonNullable<(typeof agentModel.events)["step.started"]> @@ -48,6 +49,18 @@ describe("agentModel", () => { expect(body.usage).toEqual({ include: true }) }) + test("a call made under a span nests OpenRouter's mirror under it", async () => { + const ids = await withActiveSpan("ai.streamText.doStream", async (ids) => { + await capturedBody(agentModel.fallback) + return ids + }) + const raw = stub?.calls[0]?.body + if (typeof raw !== "string") throw new Error("no request reached the transport") + // SAFETY: the stub captured the JSON object the provider serialised; `trace` is read as one key. + const body = JSON.parse(raw) as Record + expect(body.trace).toEqual({ trace_name: "slack", trace_id: ids.traceId, parent_span_id: ids.spanId }) + }) + test("the fallback, used before any session exists, carries none", async () => { const body = await capturedBody(agentModel.fallback) expect(body).not.toHaveProperty("session_id") diff --git a/apps/slack-agent/agent/lib/agent-model.ts b/apps/slack-agent/agent/lib/agent-model.ts index e1042a9b5..b6c764d00 100644 --- a/apps/slack-agent/agent/lib/agent-model.ts +++ b/apps/slack-agent/agent/lib/agent-model.ts @@ -1,5 +1,6 @@ import { createOpenRouter } from "@openrouter/ai-sdk-provider" import { defineDynamic } from "eve" +import { openRouterFetch } from "./openrouter-trace.js" /** * OpenRouter over its REST API. @@ -8,12 +9,14 @@ import { defineDynamic } from "eve" * traffic to Maple's app page on openrouter.ai. Same URL and title as `apps/api` on purpose: the * referer is the app's identity, so a different one here would mint a second app entry and split * the rankings. Surfaces are told apart by `trace.trace_name` instead — static, because this - * process only ever is the Slack agent. + * process only ever is the Slack agent. `fetch` adds the calling span's ids per request, which is + * what nests OpenRouter's Broadcast mirror of the call under this agent's own trace. */ const openrouter = createOpenRouter({ apiKey: process.env.OPENROUTER_API_KEY ?? "", appUrl: "https://maple.dev", appName: "Maple", + fetch: openRouterFetch, extraBody: { trace: { trace_name: "slack" } }, }) diff --git a/apps/slack-agent/agent/lib/follow-up-relevance.ts b/apps/slack-agent/agent/lib/follow-up-relevance.ts index e936c2234..148bc2abd 100644 --- a/apps/slack-agent/agent/lib/follow-up-relevance.ts +++ b/apps/slack-agent/agent/lib/follow-up-relevance.ts @@ -1,5 +1,6 @@ import { generateText } from "ai" import { createOpenRouter } from "@openrouter/ai-sdk-provider" +import { openRouterFetch } from "./openrouter-trace.js" import type { SlackThreadMessage } from "eve/channels/slack" import { formatContextBlock, @@ -104,6 +105,7 @@ function buildDefaultDeps(): FollowUpRelevanceDeps { apiKey: process.env.OPENROUTER_API_KEY ?? "", appUrl: "https://maple.dev", appName: "Maple", + fetch: openRouterFetch, extraBody: { trace: { trace_name: "slack" } }, }) const model = openrouter(gateModelId()) diff --git a/apps/slack-agent/agent/lib/openrouter-trace.test.ts b/apps/slack-agent/agent/lib/openrouter-trace.test.ts new file mode 100644 index 000000000..f28322965 --- /dev/null +++ b/apps/slack-agent/agent/lib/openrouter-trace.test.ts @@ -0,0 +1,74 @@ +import { afterEach, beforeAll, describe, expect, test } from "bun:test" +import { context, trace } from "@opentelemetry/api" +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks" +import { BasicTracerProvider } from "@opentelemetry/sdk-trace-base" +import { installFetchStub, type FetchStub } from "./fetch-stub.js" +import { openRouterFetch } from "./openrouter-trace.js" + +/** A tracer whose spans are active across `await`, the way the OTel Node SDK's are in production. */ +export const withActiveSpan = async ( + name: string, + fn: (ids: { traceId: string; spanId: string }) => Promise, +) => + new BasicTracerProvider().getTracer("test").startActiveSpan(name, async (span) => { + try { + return await fn(span.spanContext()) + } finally { + span.end() + } + }) + +let stub: FetchStub | undefined + +beforeAll(() => { + // A second registration in the same process is refused, harmlessly: the first one is ours too. + context.setGlobalContextManager(new AsyncLocalStorageContextManager().enable()) +}) + +afterEach(() => { + stub?.restore() + stub = undefined +}) + +const sent = (): Record => { + const body = stub?.calls[0]?.body + if (typeof body !== "string") throw new Error("no request reached the transport") + // SAFETY: the test posted a JSON object below and reads its keys one at a time. + return JSON.parse(body) as Record +} + +describe("openRouterFetch", () => { + test("stamps the active span's ids into the request's trace object", async () => { + stub = installFetchStub(() => new Response("{}")) + const ids = await withActiveSpan("ai.streamText.doStream", async (ids) => { + await openRouterFetch("https://openrouter.ai/api/v1/chat/completions", { + method: "POST", + body: JSON.stringify({ model: "m", trace: { trace_name: "slack" } }), + }) + return ids + }) + expect(sent().trace).toEqual({ + trace_name: "slack", + trace_id: ids.traceId, + parent_span_id: ids.spanId, + }) + expect(sent().model).toBe("m") + }) + + test("leaves the request alone when no span is active", async () => { + stub = installFetchStub(() => new Response("{}")) + await openRouterFetch("https://openrouter.ai/api/v1/chat/completions", { + method: "POST", + body: JSON.stringify({ model: "m", trace: { trace_name: "slack" } }), + }) + expect(sent().trace).toEqual({ trace_name: "slack" }) + }) + + test("passes a body that is not JSON through untouched", async () => { + stub = installFetchStub(() => new Response("{}")) + await withActiveSpan("ai.streamText.doStream", async () => { + await openRouterFetch("https://example.com", { method: "POST", body: "not json" }) + }) + expect(stub.calls[0]?.body).toBe("not json") + }) +}) diff --git a/apps/slack-agent/agent/lib/openrouter-trace.ts b/apps/slack-agent/agent/lib/openrouter-trace.ts new file mode 100644 index 000000000..4faced98e --- /dev/null +++ b/apps/slack-agent/agent/lib/openrouter-trace.ts @@ -0,0 +1,41 @@ +import { isSpanContextValid, trace } from "@opentelemetry/api" + +/** + * The `fetch` the OpenRouter provider sends its requests through: it stamps the calling span's + * W3C ids onto the request as `trace.trace_id` / `trace.parent_span_id`. + * + * OpenRouter's Broadcast export uses them verbatim, so the `LLM Generation` trace it sends Maple + * for every call (provider attempts, fallbacks, router latency) nests under the AI SDK span that + * made the call. Without them each call arrived as a root trace of its own — the session still + * held them, through `session_id`, but as one detached trace per model call. The provider's + * `extraBody` cannot carry the ids: it is fixed when the model is built, and the ids are per call. + * + * `globalThis.fetch` is read per call, not captured: the tests stub it, and undici's + * instrumentation patches it after this module loads. + */ +// SAFETY: Bun's `typeof fetch` also declares `preconnect`, a warm-up hint the provider never calls; +// the call signature is the whole contract here. +export const openRouterFetch = ((input: RequestInfo | URL, init?: RequestInit) => { + const spanContext = trace.getActiveSpan()?.spanContext() + if (spanContext === undefined || !isSpanContextValid(spanContext) || typeof init?.body !== "string") { + return globalThis.fetch(input, init) + } + return globalThis.fetch(input, { ...init, body: withTraceIds(init.body, spanContext) }) +}) as typeof fetch + +const withTraceIds = (body: string, span: { traceId: string; spanId: string }): string => { + try { + // SAFETY: the provider serialised a JSON object; anything else is passed through untouched. + const parsed = JSON.parse(body) as unknown + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) return body + const request = parsed as Record + const existing = + typeof request.trace === "object" && request.trace !== null ? request.trace : undefined + return JSON.stringify({ + ...request, + trace: { ...existing, trace_id: span.traceId, parent_span_id: span.spanId }, + }) + } catch { + return body + } +} diff --git a/apps/slack-agent/bun.lock b/apps/slack-agent/bun.lock index 51d68bb1a..25bd439f6 100644 --- a/apps/slack-agent/bun.lock +++ b/apps/slack-agent/bun.lock @@ -25,6 +25,7 @@ "zod": "4.4.3", }, "devDependencies": { + "@opentelemetry/context-async-hooks": "^2.1.0", "@types/bun": "^1.3.14", "@types/node": "24.x", "typescript": "7.0.2", diff --git a/apps/slack-agent/package.json b/apps/slack-agent/package.json index cbdfee19c..06ed2dfac 100644 --- a/apps/slack-agent/package.json +++ b/apps/slack-agent/package.json @@ -36,6 +36,7 @@ "zod": "4.4.3" }, "devDependencies": { + "@opentelemetry/context-async-hooks": "^2.1.0", "@types/bun": "^1.3.14", "@types/node": "24.x", "typescript": "7.0.2" diff --git a/packages/effect-sdk/src/shared/flushable-tracer.test.ts b/packages/effect-sdk/src/shared/flushable-tracer.test.ts index f5f518d0a..8aa3cc10c 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.test.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.test.ts @@ -292,7 +292,7 @@ describe("makeSpanBuffer rendered 5xx responses", () => { }) describe("makeSpanBuffer restore", () => { - it.effect("keeps older failed telemetry and discards newest overflow", () => + it.effect("puts failed telemetry back ahead of what arrived since, and trims the oldest", () => Effect.gen(function* () { const buffer = makeSpanBuffer() yield* runSpan(buffer, Effect.succeed("old")) @@ -308,7 +308,7 @@ describe("makeSpanBuffer restore", () => { buffer.restore([oldest!, ...Array.from({ length: 10_000 }, () => newest!)]) const restored = buffer.drain() assert.strictEqual(restored.length, 10_000) - assert.strictEqual(restored[0]?.name, "http.server GET") + assert.strictEqual(restored[0]?.name, "newest") assert.strictEqual(restored.at(-1)?.name, "newest") }), ) @@ -370,3 +370,21 @@ describe("makeSpanBuffer captureException", () => { assert.strictEqual(buffer.size(), 0) }) }) + +describe("makeSpanBuffer capacity", () => { + it.effect("keeps the newest spans when full, so the root that ends last survives", () => + Effect.gen(function* () { + const buffer = makeSpanBuffer() + yield* Effect.forEach( + Array.from({ length: 10_000 }, (_, index) => index), + (index) => Effect.void.pipe(Effect.withSpan(`leaf ${index}`)), + { discard: true }, + ).pipe(Effect.withSpan("chat.turn"), Effect.provide(buffer.tracerLayer)) + + const spans = buffer.drain() + assert.strictEqual(spans.length, 10_000) + assert.strictEqual(spans.at(-1)?.name, "chat.turn") + assert.isUndefined(spans.find((span) => span.name === "leaf 0")) + }), + ) +}) diff --git a/packages/effect-sdk/src/shared/flushable-tracer.ts b/packages/effect-sdk/src/shared/flushable-tracer.ts index 2da9bcd10..90b717fd8 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.ts @@ -44,6 +44,12 @@ export interface SpanBuffer { readonly size: () => number } +/** + * Spans held between flushes. Past this the OLDEST span goes, not the newest: a span ends after its + * children, so the newest spans are the ones that close a unit of work — the request or turn root + * that carries its outcome — and a buffer that refused them kept 10,000 leaves and lost the root. + * Seen on 2026-09-17: investigation turns past the cap exported without their `chat.turn` span. + */ const MAX_BUFFER = 10_000 export interface SpanBufferOptions { @@ -101,7 +107,7 @@ export const makeSpanBuffer = (options: SpanBufferOptions = {}): SpanBuffer => { if (!span.sampled) return if (dropSpan !== undefined && dropSpan(span.name)) return if (isIgnoredSpan(span)) return - if (buffer.length >= MAX_BUFFER) return + if (buffer.length >= MAX_BUFFER) buffer.shift() buffer.push(makeOtlpSpan(span, anticipatedErrorIdentifiers)) } @@ -143,7 +149,7 @@ export const makeSpanBuffer = (options: SpanBufferOptions = {}): SpanBuffer => { }, restore: (items) => { if (disabled || items.length === 0) return - buffer = [...items, ...buffer].slice(0, MAX_BUFFER) + buffer = [...items, ...buffer].slice(-MAX_BUFFER) }, setDisabled: (value) => { disabled = value From fd7323ce686dbe1f0ad7a88f5fe33b1cac173660 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Thu, 17 Sep 2026 22:11:30 +0200 Subject: [PATCH 2/3] fix(telemetry): review follow-ups for the turn flush and the Slack trace stamp - `flush(env, { force: true })` posts through a signal's 60 s cooldown; the turn's two final flushes use it, so a tick's failed POST can no longer leave them skipped. OTLP POSTs now abort after 15 s, so a stalled collector fails into the cooldown instead of holding the turn slot. - One flush timer per isolate, refcounted across turns, instead of one per turn against the shared buffer. - The log buffer evicts the oldest record like the span buffer does; README and the server preset's option doc describe the new policy. - Slack agent: only sampled spans are stamped; the test helper for an active span lives in its own module instead of a sibling test file. --- apps/ai/src/chat/turn-runner.ts | 43 +++++++++--- .../slack-agent/agent/lib/agent-model.test.ts | 18 ++--- .../agent/lib/openrouter-trace.test.ts | 62 ++++++++--------- .../slack-agent/agent/lib/openrouter-trace.ts | 19 +++-- apps/slack-agent/agent/lib/test-span.ts | 27 ++++++++ packages/effect-sdk/README.md | 2 +- packages/effect-sdk/src/cloudflare/index.ts | 69 +++++++++++-------- packages/effect-sdk/src/server/flushable.ts | 4 +- .../effect-sdk/src/shared/flush-core.test.ts | 53 ++++++++++++++ packages/effect-sdk/src/shared/flush-core.ts | 26 ++++++- .../effect-sdk/src/shared/flushable-logger.ts | 5 +- .../src/shared/flushable-tracer.test.ts | 31 ++++++--- .../effect-sdk/src/shared/flushable-tracer.ts | 4 ++ 13 files changed, 263 insertions(+), 100 deletions(-) create mode 100644 apps/slack-agent/agent/lib/test-span.ts diff --git a/apps/ai/src/chat/turn-runner.ts b/apps/ai/src/chat/turn-runner.ts index 6e0e9bb6a..64663de08 100644 --- a/apps/ai/src/chat/turn-runner.ts +++ b/apps/ai/src/chat/turn-runner.ts @@ -55,16 +55,42 @@ const telemetry = MapleCloudflareSDK.make( ) /** - * How often a running turn ships the spans it has finished so far. + * How often running turns ship the spans they have finished so far. * - * The turn is background work on the Durable Object: no request holds the isolate open for it, and + * A turn is background work on the Durable Object: no request holds the isolate open for it, and * an investigation has no subscriber at all. A single flush at the end of the turn was therefore * one fetch that had to survive the object's whole remaining life — and on 2026-09-17 it did not * for 22 of 101 investigation passes: OpenRouter's Broadcast mirror arrived nested under span ids * the warehouse never saw, and the Agent Sessions list showed the pass as an OpenRouter-only * session with no agent. Flushing on an interval bounds a lost flush to the tail of the turn. + * + * One timer per isolate, not per turn: `telemetry` and its buffers are shared by every + * `ChatSession` in the isolate, so a timer per turn would drain the same buffer N times a window + * and re-export the cumulative metric snapshot each time. The SDK resolves its endpoint from the + * first `env` it sees, so which turn's `env` the timer captured does not matter. + * + * The trade: a POST that fails after the collector persisted it is retried by the next tick, so + * a span can now land twice. Session usage nets by response id; raw span counts do not. */ const FLUSH_INTERVAL_MS = 10_000 +let liveTurns = 0 +let flushTimer: ReturnType | undefined + +const retainFlushTimer = (env: Record): void => { + liveTurns += 1 + if (flushTimer !== undefined) return + // Host timer on purpose: it must outlive every turn's Effect runtime, and `flush` is a Promise + // API that never rejects (`guardFlush`). + // oxlint-disable-next-line effecttsgo/global-timers + flushTimer = setInterval(() => void telemetry.flush(env), FLUSH_INTERVAL_MS) +} + +const releaseFlushTimer = (): void => { + liveTurns -= 1 + if (liveTurns > 0 || flushTimer === undefined) return + clearInterval(flushTimer) + flushTimer = undefined +} export interface RunChatSessionTurnInput { /** The Durable Object itself. Appends are direct calls, not stub RPC. */ @@ -435,9 +461,7 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis }), ) - // `flush` never rejects and serializes overlapping calls, so a tick landing on the final flush - // queues behind it rather than racing it. - const flushTimer = setInterval(() => void telemetry.flush(input.env), FLUSH_INTERVAL_MS) + retainFlushTimer(input.env) try { await runtime.runPromise(program) } catch { @@ -452,11 +476,14 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis }) } } finally { - clearInterval(flushTimer) + releaseFlushTimer() // The turn's own spans have ended by now; ship them before `dispose`, whose finalizers (the // Postgres connection, the model client) are the one part of the turn that can still hang. - await telemetry.flush(input.env).catch(() => undefined) + // `force`: these two are the last flushes this turn makes, so a tick's failed POST must not + // leave them skipped inside the cooldown. `flush` never rejects and serializes overlapping + // calls — a tick in flight finishes (or times out) first, then this one drains. + await telemetry.flush(input.env, { force: true }) await runtime.dispose().catch(() => undefined) - await telemetry.flush(input.env).catch(() => undefined) + await telemetry.flush(input.env, { force: true }) } } diff --git a/apps/slack-agent/agent/lib/agent-model.test.ts b/apps/slack-agent/agent/lib/agent-model.test.ts index fa5ee2f30..817a44503 100644 --- a/apps/slack-agent/agent/lib/agent-model.test.ts +++ b/apps/slack-agent/agent/lib/agent-model.test.ts @@ -2,7 +2,7 @@ import { afterEach, describe, expect, test } from "bun:test" import { generateText, type LanguageModel } from "ai" import { agentModel } from "./agent-model.js" import { installFetchStub, type FetchStub } from "./fetch-stub.js" -import { withActiveSpan } from "./openrouter-trace.test.js" +import { withActiveSpan } from "./test-span.js" type StepStarted = NonNullable<(typeof agentModel.events)["step.started"]> @@ -50,15 +50,15 @@ describe("agentModel", () => { }) test("a call made under a span nests OpenRouter's mirror under it", async () => { - const ids = await withActiveSpan("ai.streamText.doStream", async (ids) => { - await capturedBody(agentModel.fallback) - return ids + const { ids, body } = await withActiveSpan("ai.streamText.doStream", async (ids) => ({ + ids, + body: await capturedBody(agentModel.fallback), + })) + expect(body.trace).toEqual({ + trace_name: "slack", + trace_id: ids.traceId, + parent_span_id: ids.spanId, }) - const raw = stub?.calls[0]?.body - if (typeof raw !== "string") throw new Error("no request reached the transport") - // SAFETY: the stub captured the JSON object the provider serialised; `trace` is read as one key. - const body = JSON.parse(raw) as Record - expect(body.trace).toEqual({ trace_name: "slack", trace_id: ids.traceId, parent_span_id: ids.spanId }) }) test("the fallback, used before any session exists, carries none", async () => { diff --git a/apps/slack-agent/agent/lib/openrouter-trace.test.ts b/apps/slack-agent/agent/lib/openrouter-trace.test.ts index f28322965..9fe0c359a 100644 --- a/apps/slack-agent/agent/lib/openrouter-trace.test.ts +++ b/apps/slack-agent/agent/lib/openrouter-trace.test.ts @@ -1,37 +1,20 @@ -import { afterEach, beforeAll, describe, expect, test } from "bun:test" -import { context, trace } from "@opentelemetry/api" -import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks" -import { BasicTracerProvider } from "@opentelemetry/sdk-trace-base" +import { afterEach, describe, expect, test } from "bun:test" import { installFetchStub, type FetchStub } from "./fetch-stub.js" import { openRouterFetch } from "./openrouter-trace.js" - -/** A tracer whose spans are active across `await`, the way the OTel Node SDK's are in production. */ -export const withActiveSpan = async ( - name: string, - fn: (ids: { traceId: string; spanId: string }) => Promise, -) => - new BasicTracerProvider().getTracer("test").startActiveSpan(name, async (span) => { - try { - return await fn(span.spanContext()) - } finally { - span.end() - } - }) +import { withActiveSpan } from "./test-span.js" let stub: FetchStub | undefined -beforeAll(() => { - // A second registration in the same process is refused, harmlessly: the first one is ours too. - context.setGlobalContextManager(new AsyncLocalStorageContextManager().enable()) -}) - afterEach(() => { stub?.restore() stub = undefined }) -const sent = (): Record => { - const body = stub?.calls[0]?.body +const URL = "https://openrouter.ai/api/v1/chat/completions" +const post = (body: string) => openRouterFetch(URL, { method: "POST", body }) + +const sent = (index = 0): Record => { + const body = stub?.calls[index]?.body if (typeof body !== "string") throw new Error("no request reached the transport") // SAFETY: the test posted a JSON object below and reads its keys one at a time. return JSON.parse(body) as Record @@ -41,10 +24,7 @@ describe("openRouterFetch", () => { test("stamps the active span's ids into the request's trace object", async () => { stub = installFetchStub(() => new Response("{}")) const ids = await withActiveSpan("ai.streamText.doStream", async (ids) => { - await openRouterFetch("https://openrouter.ai/api/v1/chat/completions", { - method: "POST", - body: JSON.stringify({ model: "m", trace: { trace_name: "slack" } }), - }) + await post(JSON.stringify({ model: "m", trace: { trace_name: "slack" } })) return ids }) expect(sent().trace).toEqual({ @@ -55,20 +35,34 @@ describe("openRouterFetch", () => { expect(sent().model).toBe("m") }) - test("leaves the request alone when no span is active", async () => { + test("each call carries its own span, not the first one's", async () => { stub = installFetchStub(() => new Response("{}")) - await openRouterFetch("https://openrouter.ai/api/v1/chat/completions", { - method: "POST", - body: JSON.stringify({ model: "m", trace: { trace_name: "slack" } }), + const first = await withActiveSpan("call 1", async (ids) => { + await post(JSON.stringify({ model: "m" })) + return ids }) + const second = await withActiveSpan("call 2", async (ids) => { + await post(JSON.stringify({ model: "m" })) + return ids + }) + expect(first.spanId).not.toBe(second.spanId) + expect(sent(0).trace).toEqual({ trace_id: first.traceId, parent_span_id: first.spanId }) + expect(sent(1).trace).toEqual({ trace_id: second.traceId, parent_span_id: second.spanId }) + }) + + test("leaves the request alone when no span is active", async () => { + stub = installFetchStub(() => new Response("{}")) + await post(JSON.stringify({ model: "m", trace: { trace_name: "slack" } })) expect(sent().trace).toEqual({ trace_name: "slack" }) }) - test("passes a body that is not JSON through untouched", async () => { + test("passes a body that is not a JSON object through untouched", async () => { stub = installFetchStub(() => new Response("{}")) await withActiveSpan("ai.streamText.doStream", async () => { - await openRouterFetch("https://example.com", { method: "POST", body: "not json" }) + await post("not json") + await post("[1]") }) expect(stub.calls[0]?.body).toBe("not json") + expect(stub.calls[1]?.body).toBe("[1]") }) }) diff --git a/apps/slack-agent/agent/lib/openrouter-trace.ts b/apps/slack-agent/agent/lib/openrouter-trace.ts index 4faced98e..253faad87 100644 --- a/apps/slack-agent/agent/lib/openrouter-trace.ts +++ b/apps/slack-agent/agent/lib/openrouter-trace.ts @@ -1,4 +1,4 @@ -import { isSpanContextValid, trace } from "@opentelemetry/api" +import { isSpanContextValid, trace, TraceFlags } from "@opentelemetry/api" /** * The `fetch` the OpenRouter provider sends its requests through: it stamps the calling span's @@ -10,14 +10,21 @@ import { isSpanContextValid, trace } from "@opentelemetry/api" * held them, through `session_id`, but as one detached trace per model call. The provider's * `extraBody` cannot carry the ids: it is fixed when the model is built, and the ids are per call. * - * `globalThis.fetch` is read per call, not captured: the tests stub it, and undici's - * instrumentation patches it after this module loads. + * Only a sampled span is stamped: a sampled-out one has valid-looking ids that Maple will never + * receive, and a mirror nested under those would be an orphan rather than a root trace. + * + * `globalThis.fetch` is read per call, not captured, so the tests' stub sees the request. */ // SAFETY: Bun's `typeof fetch` also declares `preconnect`, a warm-up hint the provider never calls; // the call signature is the whole contract here. export const openRouterFetch = ((input: RequestInfo | URL, init?: RequestInit) => { const spanContext = trace.getActiveSpan()?.spanContext() - if (spanContext === undefined || !isSpanContextValid(spanContext) || typeof init?.body !== "string") { + if ( + spanContext === undefined || + !isSpanContextValid(spanContext) || + (spanContext.traceFlags & TraceFlags.SAMPLED) === 0 || + typeof init?.body !== "string" + ) { return globalThis.fetch(input, init) } return globalThis.fetch(input, { ...init, body: withTraceIds(init.body, spanContext) }) @@ -30,7 +37,9 @@ const withTraceIds = (body: string, span: { traceId: string; spanId: string }): if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) return body const request = parsed as Record const existing = - typeof request.trace === "object" && request.trace !== null ? request.trace : undefined + typeof request.trace === "object" && request.trace !== null && !Array.isArray(request.trace) + ? request.trace + : undefined return JSON.stringify({ ...request, trace: { ...existing, trace_id: span.traceId, parent_span_id: span.spanId }, diff --git a/apps/slack-agent/agent/lib/test-span.ts b/apps/slack-agent/agent/lib/test-span.ts new file mode 100644 index 000000000..d6b327580 --- /dev/null +++ b/apps/slack-agent/agent/lib/test-span.ts @@ -0,0 +1,27 @@ +/** + * Test-only: run a function under an active, sampled OTel span — the way the Node SDK's spans are + * active in production. Not imported by any production module; lives beside `fetch-stub.ts` for + * the same reason it does (a shared seam that no `*.test.ts` should have to import from another). + */ +import { context, trace } from "@opentelemetry/api" +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks" +import { BasicTracerProvider } from "@opentelemetry/sdk-trace-base" + +// Without a context manager `startActiveSpan` never makes the span visible to `getActiveSpan()`. +// A second registration in one process is refused; `withActiveSpan` checks the outcome instead. +context.setGlobalContextManager(new AsyncLocalStorageContextManager().enable()) + +const tracer = new BasicTracerProvider().getTracer("test") + +export const withActiveSpan = ( + name: string, + fn: (ids: { traceId: string; spanId: string }) => Promise, +): Promise => + tracer.startActiveSpan(name, async (span) => { + try { + if (trace.getActiveSpan() !== span) throw new Error("no active-span context manager") + return await fn(span.spanContext()) + } finally { + span.end() + } + }) diff --git a/packages/effect-sdk/README.md b/packages/effect-sdk/README.md index 196a6343c..0afb71621 100644 --- a/packages/effect-sdk/README.md +++ b/packages/effect-sdk/README.md @@ -73,7 +73,7 @@ export default { `telemetry.layer` MUST live in the same runtime as your routes — provide it to the layer composition you hand to `HttpRouter.toWebHandler`, not a separate per-request runtime, or your spans won't pick up the Tracer reference. -When `MAPLE_INGEST_KEY` is unset, the SDK runs in no-op mode: buffers are drained so they don't grow across the isolate's lifetime, but no requests are made. After a flush failure, the failed batch is restored ahead of newer telemetry and that signal sleeps 60s before retrying. Trace and log cooldowns are independent. +When `MAPLE_INGEST_KEY` is unset, the SDK runs in no-op mode: buffers are drained so they don't grow across the isolate's lifetime, but no requests are made. After a flush failure, the failed batch is put back ahead of newer telemetry and that signal sleeps 60s before retrying; buffers hold 10k items per signal and evict the oldest past that, so a long outage loses the earliest telemetry first. Trace and log cooldowns are independent. `flush(env, { force: true })` posts through a cooldown — for the last flush a piece of background work will make. ### Cloudflare-specific options diff --git a/packages/effect-sdk/src/cloudflare/index.ts b/packages/effect-sdk/src/cloudflare/index.ts index 5f7344355..33c55bb56 100644 --- a/packages/effect-sdk/src/cloudflare/index.ts +++ b/packages/effect-sdk/src/cloudflare/index.ts @@ -149,7 +149,16 @@ export interface Telemetry { * - Errors are caught and logged to `console.error`; cooldown of 60s * per signal before next attempt after a failure. */ - flush(env: Record): Promise + flush(env: Record, options?: FlushOptions): Promise +} + +export interface FlushOptions { + /** + * POST even while a signal is in its 60 s cooldown after a failed flush. For the last flush + * a piece of background work will make: nothing drains these buffers after it, so a skipped + * flush there is a lost one, not a deferred one. + */ + readonly force?: boolean | undefined } const resolveOnce = (env: Record, config: Config): Resolved => { @@ -192,35 +201,39 @@ export const make = (config: Config = {}): Telemetry => { // Never rejects: this runs inside `ctx.waitUntil`, where a rejection would // surface as an unhandled Worker error caused purely by telemetry. const flush = makeSerializedFlush( - guardFlush("[MapleCloudflareSDK]", async (env: Record): Promise => { - // Effect defers work onto the scheduler's next macrotask - // (`scheduleTask(task, 0)`) — including `HttpMiddleware.tracer`'s - // `span.end` and `withSpan` finalizers — while the drain below is - // synchronous. Flushing in the same task therefore misses exactly the - // spans the request just produced, and an isolated request (e.g. a lone - // webhook) can freeze the isolate before a later flush rescues them. - // Yield one macrotask so those tasks run first. This sits INSIDE the - // serialized body, so overlapping flushes still queue rather than - // interleave. - await new Promise((resolve) => setTimeout(resolve, 0)) + guardFlush( + "[MapleCloudflareSDK]", + async (env: Record, options?: FlushOptions): Promise => { + // Effect defers work onto the scheduler's next macrotask + // (`scheduleTask(task, 0)`) — including `HttpMiddleware.tracer`'s + // `span.end` and `withSpan` finalizers — while the drain below is + // synchronous. Flushing in the same task therefore misses exactly the + // spans the request just produced, and an isolated request (e.g. a lone + // webhook) can freeze the isolate before a later flush rescues them. + // Yield one macrotask so those tasks run first. This sits INSIDE the + // serialized body, so overlapping flushes still queue rather than + // interleave. + await new Promise((resolve) => setTimeout(resolve, 0)) - if (resolved === undefined) { - resolved = resolveOnce(env, config) - } + if (resolved === undefined) { + resolved = resolveOnce(env, config) + } - await runFlush({ - resolved, - spans, - logs, - metrics, - tracesState, - logsState, - metricsState, - transport: fetchTransport, - logPrefix: "[MapleCloudflareSDK]", - onNoOp: noOpNotice, - }) - }), + await runFlush({ + resolved, + spans, + logs, + metrics, + tracesState, + logsState, + metricsState, + transport: fetchTransport, + logPrefix: "[MapleCloudflareSDK]", + onNoOp: noOpNotice, + force: options?.force, + }) + }, + ), { coalesceSameArguments: true }, ) diff --git a/packages/effect-sdk/src/server/flushable.ts b/packages/effect-sdk/src/server/flushable.ts index fb512d512..3d4d44380 100644 --- a/packages/effect-sdk/src/server/flushable.ts +++ b/packages/effect-sdk/src/server/flushable.ts @@ -80,8 +80,8 @@ export interface MapleFlushableConfig { /** * Background auto-flush cadence in milliseconds. Default `5000`. Set to `0` * or `false` to disable and flush purely on demand (note: the in-memory - * buffer caps at 10k items and drops new spans past that, so a long-running - * process that never flushes will lose data). + * buffer caps at 10k items and evicts the oldest past that, so a long-running + * process that never flushes will lose its earliest data). */ readonly autoFlushInterval?: number | false | undefined } diff --git a/packages/effect-sdk/src/shared/flush-core.test.ts b/packages/effect-sdk/src/shared/flush-core.test.ts index a2024b82b..9da31055b 100644 --- a/packages/effect-sdk/src/shared/flush-core.test.ts +++ b/packages/effect-sdk/src/shared/flush-core.test.ts @@ -105,6 +105,59 @@ describe("runFlush", () => { }), ) + it.live("`force` posts through a signal's cooldown and still re-arms it on failure", () => + Effect.gen(function* () { + vi.useFakeTimers() + vi.setSystemTime(new Date("2026-01-01T00:00:00Z")) + const spans = makeSpanBuffer() + const logs = makeLogBuffer() + const metrics = makeMetricBuffer() + const tracesState: SignalState = { disabledUntil: 0 } + const logsState: SignalState = { disabledUntil: 0 } + const metricsState: SignalState = { disabledUntil: 0 } + let attempts = 0 + let failTraces = true + const transport: FlushTransport = { + post: async (url) => { + if (!url.endsWith("/v1/traces")) return + attempts += 1 + if (failTraces) throw new Error("collector unavailable") + }, + } + const flush = (force?: boolean) => + runFlush({ + resolved, + spans, + logs, + metrics, + tracesState, + logsState, + metricsState, + transport, + logPrefix: "[test]", + onNoOp: () => undefined, + force, + }) + + yield* recordSpan(spans, "first") + yield* Effect.promise(() => flush()) + expect(attempts).toBe(1) + // Inside the cooldown a plain flush is skipped, a forced one is attempted. + yield* Effect.promise(() => flush()) + expect(attempts).toBe(1) + yield* Effect.promise(() => flush(true)) + expect(attempts).toBe(2) + expect(spans.size()).toBe(1) + // The forced failure re-armed the cooldown for the next plain flush. + yield* Effect.promise(() => flush()) + expect(attempts).toBe(2) + failTraces = false + yield* Effect.promise(() => flush(true)) + expect(attempts).toBe(3) + expect(spans.size()).toBe(0) + }), + ) + vitestIt("serializes overlapping flush calls", async () => { let active = 0 let peak = 0 diff --git a/packages/effect-sdk/src/shared/flush-core.ts b/packages/effect-sdk/src/shared/flush-core.ts index 828f48796..2ba853375 100644 --- a/packages/effect-sdk/src/shared/flush-core.ts +++ b/packages/effect-sdk/src/shared/flush-core.ts @@ -13,6 +13,12 @@ import type { OtlpSpan, SpanBuffer } from "./flushable-tracer.js" /** Disable a signal for this long after a failed POST so a broken collector isn't hammered. */ const COOLDOWN_MS = 60_000 +/** + * Abort a POST that has not completed by then. A flush is awaited at boundaries that hold real + * resources — a Worker's `waitUntil`, a Durable Object turn's slot — so a stalled collector must + * fail (and enter the cooldown) rather than hold them open. + */ +const POST_TIMEOUT_MS = 15_000 /** * Minimal resource shape consumed by {@link buildResolved}. Structurally @@ -135,7 +141,12 @@ const anyValue = (value: unknown): unknown => { /** Plain `fetch` POST. Throws on non-2xx so {@link flushSignal} records a cooldown. */ const post = async (url: string, headers: Record, body: unknown): Promise => { - const res = await fetch(url, { method: "POST", headers, body: JSON.stringify(body) }) + const res = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify(body), + signal: AbortSignal.timeout(POST_TIMEOUT_MS), + }) if (!res.ok) { throw new Error(`OTLP ${res.status} ${res.statusText}`) } @@ -153,9 +164,10 @@ const flushSignal = async (args: { readonly signal: string readonly transport: FlushTransport readonly logPrefix: string + readonly force: boolean }): Promise => { - const { url, headers, buffer, body, state, signal, transport, logPrefix } = args - if (state.disabledUntil && Date.now() < state.disabledUntil) { + const { url, headers, buffer, body, state, signal, transport, logPrefix, force } = args + if (!force && state.disabledUntil && Date.now() < state.disabledUntil) { console.warn( `${logPrefix} ${signal} flush skipped (cooldown ${state.disabledUntil - Date.now()}ms remaining)`, ) @@ -236,6 +248,9 @@ export const makeSerializedFlush = >( * - `noOp`: drain so the buffers don't grow unbounded, fire `onNoOp` (one-shot * "telemetry disabled" notice), never POST. * - empty buffers: short-circuit without a request. + * - `force`: POST even inside a signal's cooldown. For the last flush a unit of work will ever + * make — after it nothing else drains these buffers, so the cooldown's "try again later" has no + * later. A failure still arms the cooldown for whoever flushes next. */ export const runFlush = async (args: { readonly resolved: Resolved @@ -248,6 +263,7 @@ export const runFlush = async (args: { readonly transport: FlushTransport readonly logPrefix: string readonly onNoOp: () => void + readonly force?: boolean | undefined }): Promise => { const { resolved: r, @@ -261,6 +277,7 @@ export const runFlush = async (args: { logPrefix, onNoOp, } = args + const force = args.force === true if (r.noOp) { spans.drain() @@ -280,6 +297,7 @@ export const runFlush = async (args: { signal: "traces", transport, logPrefix, + force, }), flushSignal({ url: r.logsUrl, @@ -290,6 +308,7 @@ export const runFlush = async (args: { signal: "logs", transport, logPrefix, + force, }), flushSignal({ url: r.metricsUrl, @@ -300,6 +319,7 @@ export const runFlush = async (args: { signal: "metrics", transport, logPrefix, + force, }), ]) } diff --git a/packages/effect-sdk/src/shared/flushable-logger.ts b/packages/effect-sdk/src/shared/flushable-logger.ts index cb531c3a9..8f3d79086 100644 --- a/packages/effect-sdk/src/shared/flushable-logger.ts +++ b/packages/effect-sdk/src/shared/flushable-logger.ts @@ -14,6 +14,7 @@ export interface LogBuffer { readonly size: () => number } +/** Same cap and eviction as the span buffer: when full the OLDEST record goes, so the line that closes a unit of work survives and lines keep pointing at spans that were kept. */ const MAX_BUFFER = 10_000 export const makeLogBuffer = (options: { readonly excludeLogSpans?: boolean } = {}): LogBuffer => { @@ -23,7 +24,7 @@ export const makeLogBuffer = (options: { readonly excludeLogSpans?: boolean } = const logger = Logger.make((logOptions) => { if (disabled) return - if (buffer.length >= MAX_BUFFER) return + if (buffer.length >= MAX_BUFFER) buffer.shift() buffer.push(makeLogRecord(logOptions, excludeLogSpans)) }) @@ -36,7 +37,7 @@ export const makeLogBuffer = (options: { readonly excludeLogSpans?: boolean } = }, restore: (items) => { if (disabled || items.length === 0) return - buffer = [...items, ...buffer].slice(0, MAX_BUFFER) + buffer = [...items, ...buffer].slice(-MAX_BUFFER) }, setDisabled: (value) => { disabled = value diff --git a/packages/effect-sdk/src/shared/flushable-tracer.test.ts b/packages/effect-sdk/src/shared/flushable-tracer.test.ts index 8aa3cc10c..7b367945f 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.test.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.test.ts @@ -292,15 +292,31 @@ describe("makeSpanBuffer rendered 5xx responses", () => { }) describe("makeSpanBuffer restore", () => { - it.effect("puts failed telemetry back ahead of what arrived since, and trims the oldest", () => + const record = (buffer: ReturnType, name: string) => + Effect.void.pipe(Effect.withSpan(name), Effect.provide(buffer.tracerLayer)) + + it.effect("puts failed telemetry back ahead of what arrived since", () => Effect.gen(function* () { const buffer = makeSpanBuffer() - yield* runSpan(buffer, Effect.succeed("old")) - const [oldest] = buffer.drain() - yield* Effect.succeed(undefined).pipe( - Effect.withSpan("newest"), - Effect.provide(buffer.tracerLayer), + yield* record(buffer, "failed") + const failed = buffer.drain() + yield* record(buffer, "since 1") + yield* record(buffer, "since 2") + + buffer.restore(failed) + assert.deepStrictEqual( + buffer.drain().map((span) => span.name), + ["failed", "since 1", "since 2"], ) + }), + ) + + it.effect("trims the oldest when the restored batch overflows the cap", () => + Effect.gen(function* () { + const buffer = makeSpanBuffer() + yield* record(buffer, "oldest") + const [oldest] = buffer.drain() + yield* record(buffer, "newest") const [newest] = buffer.drain() assert.isDefined(oldest) assert.isDefined(newest) @@ -308,8 +324,7 @@ describe("makeSpanBuffer restore", () => { buffer.restore([oldest!, ...Array.from({ length: 10_000 }, () => newest!)]) const restored = buffer.drain() assert.strictEqual(restored.length, 10_000) - assert.strictEqual(restored[0]?.name, "newest") - assert.strictEqual(restored.at(-1)?.name, "newest") + assert.isUndefined(restored.find((span) => span.name === "oldest")) }), ) }) diff --git a/packages/effect-sdk/src/shared/flushable-tracer.ts b/packages/effect-sdk/src/shared/flushable-tracer.ts index 90b717fd8..7e67ecec4 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.ts @@ -49,6 +49,10 @@ export interface SpanBuffer { * children, so the newest spans are the ones that close a unit of work — the request or turn root * that carries its outcome — and a buffer that refused them kept 10,000 leaves and lost the root. * Seen on 2026-09-17: investigation turns past the cap exported without their `chat.turn` span. + * + * The same rule governs `restore`: a batch put back after a failed POST is older than what + * arrived meanwhile, so at the cap it is what gets trimmed. A collector that stays down for a + * signal's cooldown therefore loses whole earlier traces rather than every trace's root. */ const MAX_BUFFER = 10_000 From d2f6afbddfaf9173415efcda78a6321639faaea3 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Thu, 17 Sep 2026 22:29:54 +0200 Subject: [PATCH 3/3] fix(ai): bound runtime.dispose() so the last flush always runs A hung finalizer kept `runChatSessionTurn` pending, which skipped the second forced flush and the session's `endTurn`. Disposal is now given five seconds, the same bound the metering call already has. Also covers the Cloudflare preset's public `flush(env, { force: true })` in its cooldown test and tightens the README sentence describing it. --- apps/ai/src/chat/turn-runner.ts | 10 +++++++++- packages/effect-sdk/README.md | 2 +- packages/effect-sdk/src/cloudflare/index.test.ts | 8 ++++++++ 3 files changed, 18 insertions(+), 2 deletions(-) diff --git a/apps/ai/src/chat/turn-runner.ts b/apps/ai/src/chat/turn-runner.ts index 64663de08..9c116f452 100644 --- a/apps/ai/src/chat/turn-runner.ts +++ b/apps/ai/src/chat/turn-runner.ts @@ -73,6 +73,12 @@ const telemetry = MapleCloudflareSDK.make( * a span can now land twice. Session usage nets by response id; raw span counts do not. */ const FLUSH_INTERVAL_MS = 10_000 +/** + * How long the turn waits for `runtime.dispose()`. Its finalizers — the Postgres connection, the + * model client — are the one part of the turn that can hang, and past this the second flush and + * the session's `endTurn` matter more than a clean close. + */ +const DISPOSE_TIMEOUT = "5 seconds" let liveTurns = 0 let flushTimer: ReturnType | undefined @@ -483,7 +489,9 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis // leave them skipped inside the cooldown. `flush` never rejects and serializes overlapping // calls — a tick in flight finishes (or times out) first, then this one drains. await telemetry.flush(input.env, { force: true }) - await runtime.dispose().catch(() => undefined) + await Effect.runPromise( + Effect.promise(() => runtime.dispose()).pipe(Effect.timeout(DISPOSE_TIMEOUT), Effect.ignoreCause), + ) await telemetry.flush(input.env, { force: true }) } } diff --git a/packages/effect-sdk/README.md b/packages/effect-sdk/README.md index 0afb71621..668599789 100644 --- a/packages/effect-sdk/README.md +++ b/packages/effect-sdk/README.md @@ -73,7 +73,7 @@ export default { `telemetry.layer` MUST live in the same runtime as your routes — provide it to the layer composition you hand to `HttpRouter.toWebHandler`, not a separate per-request runtime, or your spans won't pick up the Tracer reference. -When `MAPLE_INGEST_KEY` is unset, the SDK runs in no-op mode: buffers are drained so they don't grow across the isolate's lifetime, but no requests are made. After a flush failure, the failed batch is put back ahead of newer telemetry and that signal sleeps 60s before retrying; buffers hold 10k items per signal and evict the oldest past that, so a long outage loses the earliest telemetry first. Trace and log cooldowns are independent. `flush(env, { force: true })` posts through a cooldown — for the last flush a piece of background work will make. +When `MAPLE_INGEST_KEY` is unset, the SDK runs in no-op mode: buffers are drained so they don't grow across the isolate's lifetime, but no requests are made. After a flush failure, the failed batch is put back ahead of newer telemetry and that signal sleeps 60s before retrying; buffers hold 10k items per signal and evict the oldest past that, so a long outage loses the earliest telemetry first. Trace and log cooldowns are independent. `flush(env, { force: true })` posts even inside that cooldown: use it for the last flush of a unit of work that nothing will flush after (a Durable Object's background turn, say), and run it under `ctx.waitUntil()` like any other flush so the isolate stays up until it completes. ### Cloudflare-specific options diff --git a/packages/effect-sdk/src/cloudflare/index.test.ts b/packages/effect-sdk/src/cloudflare/index.test.ts index d5d79f724..5bcef3543 100644 --- a/packages/effect-sdk/src/cloudflare/index.test.ts +++ b/packages/effect-sdk/src/cloudflare/index.test.ts @@ -299,6 +299,14 @@ describe("MapleCloudflareSDK.make", () => { await telemetry.flush(env) // second flush — within cooldown, should be a no-op expect(calls.length).toBe(failedCount) expect(consoleErrorSpy).toHaveBeenCalled() + + // A forced flush is the last one a piece of background work makes: it posts through the + // cooldown (and, failing again here, re-arms it). + await telemetry.flush(env, { force: true }) + expect(calls.length).toBe(failedCount + 1) + expect(calls.at(-1)?.url).toMatch(/\/v1\/traces$/) + await telemetry.flush(env) + expect(calls.length).toBe(failedCount + 1) }) // Effect defers `span.end` and `withSpan` finalizers onto the scheduler's