diff --git a/apps/ai/src/chat/turn-runner.ts b/apps/ai/src/chat/turn-runner.ts index 73f5a028c..9c116f452 100644 --- a/apps/ai/src/chat/turn-runner.ts +++ b/apps/ai/src/chat/turn-runner.ts @@ -54,6 +54,50 @@ const telemetry = MapleCloudflareSDK.make( }), ) +/** + * How often running turns ship the spans they have finished so far. + * + * 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 +/** + * 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 + +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. */ readonly session: ChatSession @@ -423,6 +467,7 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis }), ) + retainFlushTimer(input.env) try { await runtime.runPromise(program) } catch { @@ -437,7 +482,16 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis }) } } finally { - await runtime.dispose().catch(() => undefined) - await telemetry.flush(input.env).catch(() => undefined) + 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. + // `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 Effect.runPromise( + Effect.promise(() => runtime.dispose()).pipe(Effect.timeout(DISPOSE_TIMEOUT), Effect.ignoreCause), + ) + 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 f6e7d7b63..817a44503 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 "./test-span.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, 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, + }) + }) + 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..9fe0c359a --- /dev/null +++ b/apps/slack-agent/agent/lib/openrouter-trace.test.ts @@ -0,0 +1,68 @@ +import { afterEach, describe, expect, test } from "bun:test" +import { installFetchStub, type FetchStub } from "./fetch-stub.js" +import { openRouterFetch } from "./openrouter-trace.js" +import { withActiveSpan } from "./test-span.js" + +let stub: FetchStub | undefined + +afterEach(() => { + stub?.restore() + stub = undefined +}) + +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 +} + +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 post(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("each call carries its own span, not the first one's", async () => { + stub = installFetchStub(() => new Response("{}")) + 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 a JSON object through untouched", async () => { + stub = installFetchStub(() => new Response("{}")) + await withActiveSpan("ai.streamText.doStream", async () => { + 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 new file mode 100644 index 000000000..253faad87 --- /dev/null +++ b/apps/slack-agent/agent/lib/openrouter-trace.ts @@ -0,0 +1,50 @@ +import { isSpanContextValid, trace, TraceFlags } 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. + * + * 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) || + (spanContext.traceFlags & TraceFlags.SAMPLED) === 0 || + 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 && !Array.isArray(request.trace) + ? 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/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/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/README.md b/packages/effect-sdk/README.md index 196a6343c..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 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 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 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 f5f518d0a..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("keeps older failed telemetry and discards newest overflow", () => + 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, "http.server GET") - assert.strictEqual(restored.at(-1)?.name, "newest") + assert.isUndefined(restored.find((span) => span.name === "oldest")) }), ) }) @@ -370,3 +385,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..7e67ecec4 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.ts @@ -44,6 +44,16 @@ 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. + * + * 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 export interface SpanBufferOptions { @@ -101,7 +111,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 +153,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