From 1a9cd877484d438fb9a2a1efeb41add8e2865c0d Mon Sep 17 00:00:00 2001 From: Luc Leray Date: Tue, 7 Jul 2026 18:33:20 +0200 Subject: [PATCH 1/2] fix(otel): ship each trace's spans in one report through its owning request context Long-running invocations (e.g. Vercel Workflows) lose most spans from the first part of the run: the trace looks blank early on, with all step spans bunched into the last few seconds. Root cause, in layers: 1. The BatchSpanProcessor's scheduled flush runs in a bare setTimeout, outside the request's AsyncLocalStorage, so the runtime exporter resolved no request context and silently dropped every mid-run batch. 2. The BSP queue is shared by all concurrent requests on the instance, so even an in-context flush ships other invocations' spans through a foreign telemetry channel, where they are not reliably persisted (verified empirically: per-trace re-routing alone still lost most spans). 3. Only a single report delivered near the end of the request is reliably persisted end to end (verified empirically: mid-run reports through the correct owning context still did not land; a single report at the end always does). Fix: - Capture the request context per trace id at root-span start (always in-request), bounded by a hard cap against never-ending traces. - The exporter ACCUMULATES spans of registered traces into per-trace buffers instead of reporting mid-run; untracked traces keep shipping immediately through the ambient context; unattributable spans are retained and re-attempted instead of silently acked. - When the trace's final waitUntil flush runs (inside the owning request), finalizeTrace ships the whole buffer in a single report through the owning context and drops the registration. - The composite processor drains the BSP queue in-context on span end (debounced on OTEL_BSP_MAX_EXPORT_BATCH_SIZE / OTEL_BSP_SCHEDULE_DELAY) so long runs cannot overflow its maxQueueSize. Off-Vercel behavior is unchanged. Verified on a long-running workflow reproduction: with this change all step spans consistently survive across repeated runs (previously only the last stretch of the trace did). --- .changeset/fix-span-context-routing.md | 5 + packages/otel/build.ts | 4 +- .../composite-span-processor.test.ts | 179 +++++++++++++++++ .../src/processor/composite-span-processor.ts | 61 +++++- .../otel/src/vercel-request-context/api.ts | 83 ++++++++ .../vercel-request-context/exporter.test.ts | 190 ++++++++++++++++++ .../src/vercel-request-context/exporter.ts | 175 ++++++++++++++-- 7 files changed, 676 insertions(+), 21 deletions(-) create mode 100644 .changeset/fix-span-context-routing.md create mode 100644 packages/otel/src/processor/composite-span-processor.test.ts create mode 100644 packages/otel/src/vercel-request-context/exporter.test.ts diff --git a/.changeset/fix-span-context-routing.md b/.changeset/fix-span-context-routing.md new file mode 100644 index 0000000..8179130 --- /dev/null +++ b/.changeset/fix-span-context-routing.md @@ -0,0 +1,5 @@ +--- +"@vercel/otel": patch +--- + +Fix long-running invocations (e.g. Vercel Workflows) losing most spans from the first part of the run. The runtime only reliably persists one `reportSpans` payload per invocation, delivered through that invocation's own request context, while the BatchSpanProcessor flushed many times mid-run (from a bare `setTimeout` outside the request's AsyncLocalStorage) and shipped whole batches through whichever context was ambient. Spans are now attributed to the request context that owns their trace (captured at root-span start), accumulated per trace, and shipped in a single report when the trace is finalized from its own request's `waitUntil`. The processor also drains the BatchSpanProcessor queue in-context on span end so long runs cannot overflow it, and the exporter retains spans it cannot attribute instead of silently dropping them. diff --git a/packages/otel/build.ts b/packages/otel/build.ts index 4b7bc1e..db23ed6 100644 --- a/packages/otel/build.ts +++ b/packages/otel/build.ts @@ -6,8 +6,8 @@ const MINIFY = true; const SOURCEMAP = true; const MAX_SIZES = { - "dist/node/index.js": 300_000, // Increased from original 217KB limit - "dist/edge/index.js": 191_000, // Increased from original 185KB limit + "dist/node/index.js": 305_000, // Increased from 300KB for the span-attribution fix (main was within ~2KB of the limit) + "dist/edge/index.js": 193_000, // Increased from 191KB for the span-attribution fix (main was within ~100 bytes of the limit) }; type ExternalPluginFactory = (external: string[]) => Plugin; diff --git a/packages/otel/src/processor/composite-span-processor.test.ts b/packages/otel/src/processor/composite-span-processor.test.ts new file mode 100644 index 0000000..7eb05b7 --- /dev/null +++ b/packages/otel/src/processor/composite-span-processor.test.ts @@ -0,0 +1,179 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { + ReadableSpan, + Span, + SpanProcessor, +} from "@opentelemetry/sdk-trace-base"; +import { hrTime } from "@opentelemetry/core"; +import { ROOT_CONTEXT, TraceFlags, SpanKind, SpanStatusCode } from "@opentelemetry/api"; +import { resourceFromAttributes } from "@opentelemetry/resources"; +import type { VercelRequestContext } from "../vercel-request-context/api"; +import { hasVercelRequestContextForTrace } from "../vercel-request-context/api"; +import { CompositeSpanProcessor } from "./composite-span-processor"; + +const TRACE_ID = "ee75cd9e534ff5e9ed78b4a0c706f0f2"; +const VRC_SYMBOL = Symbol.for("@vercel/request-context"); +type GlobalWithReader = Record; + +function installAmbient(ctx: VercelRequestContext | undefined): void { + (globalThis as GlobalWithReader)[VRC_SYMBOL] = { get: () => ctx }; +} + +function fakeVrc(): VercelRequestContext & { + flushes: Array<() => Promise>; +} { + const flushes: Array<() => Promise> = []; + return { + flushes, + waitUntil: (promiseOrFunc) => { + if (typeof promiseOrFunc === "function") { + flushes.push(promiseOrFunc as () => Promise); + } + }, + headers: {}, + url: "https://example.com", + telemetry: { reportSpans: () => undefined }, + }; +} + +function readableSpanFields(spanId: string, parent: boolean) { + const spanContext = { + traceId: TRACE_ID, + spanId, + traceFlags: TraceFlags.SAMPLED, + isRemote: false, + }; + return { + name: `span-${spanId}`, + kind: SpanKind.INTERNAL, + spanContext: () => spanContext, + parentSpanContext: parent + ? { + spanId: "7e2a325411bdc191", + traceId: TRACE_ID, + traceFlags: TraceFlags.SAMPLED, + } + : undefined, + startTime: hrTime(1), + endTime: hrTime(2), + status: { code: SpanStatusCode.UNSET }, + attributes: {}, + links: [], + events: [], + duration: hrTime(1), + ended: true, + resource: resourceFromAttributes({}), + instrumentationScope: { name: "default" }, + droppedAttributesCount: 0, + droppedEventsCount: 0, + droppedLinksCount: 0, + }; +} + +function createRootSpan(spanId: string): Span { + return { + ...readableSpanFields(spanId, false), + setAttribute: () => undefined, + setAttributes: () => undefined, + addEvent: () => undefined, + addLink: () => undefined, + addLinks: () => undefined, + setStatus: () => undefined, + updateName: () => undefined, + end: () => undefined, + isRecording: () => true, + recordException: () => undefined, + } as unknown as Span; +} + +function createChildEnd(spanId: string): ReadableSpan { + return readableSpanFields(spanId, true) as unknown as ReadableSpan; +} + +function fakeProcessor(): { + processor: SpanProcessor; + forceFlush: ReturnType; +} { + const forceFlush = vi.fn().mockResolvedValue(undefined); + const processor: SpanProcessor = { + onStart: vi.fn(), + onEnd: vi.fn(), + forceFlush, + shutdown: vi.fn().mockResolvedValue(undefined), + }; + return { processor, forceFlush }; +} + +describe("CompositeSpanProcessor", () => { + afterEach(() => { + installAmbient(undefined); + vi.unstubAllEnvs(); + vi.restoreAllMocks(); + }); + + it("registers the request context for the trace at root start and finalizes it after the final flush", async () => { + const vrc = fakeVrc(); + installAmbient(vrc); + const { processor: downstream, forceFlush } = fakeProcessor(); + const processor = new CompositeSpanProcessor([downstream], undefined); + + const root = createRootSpan("7e2a325411bdc191"); + processor.onStart(root, ROOT_CONTEXT); + + // The trace's owning context is now captured, independent of the ambient one. + installAmbient(undefined); + expect(hasVercelRequestContextForTrace(TRACE_ID)).toBe(true); + + // Root ends; the waitUntil-registered final flush drains the processors + // and finalizes (ships + unregisters) the trace. + processor.onEnd(root as unknown as ReadableSpan); + expect(vrc.flushes).toHaveLength(1); + await vrc.flushes[0]?.(); + expect(forceFlush).toHaveBeenCalled(); + expect(hasVercelRequestContextForTrace(TRACE_ID)).toBe(false); + }); + + it("flushes in-context once the batch-size threshold of ended spans is reached", () => { + installAmbient(fakeVrc()); + vi.stubEnv("OTEL_BSP_MAX_EXPORT_BATCH_SIZE", "3"); + const { processor: downstream, forceFlush } = fakeProcessor(); + const processor = new CompositeSpanProcessor([downstream], undefined); + + processor.onEnd(createChildEnd("0000000000000001")); + processor.onEnd(createChildEnd("0000000000000002")); + expect(forceFlush).toHaveBeenCalledTimes(0); + + processor.onEnd(createChildEnd("0000000000000003")); + expect(forceFlush).toHaveBeenCalledTimes(1); + }); + + it("flushes in-context once the schedule delay elapses between ended spans", () => { + installAmbient(fakeVrc()); + vi.stubEnv("OTEL_BSP_MAX_EXPORT_BATCH_SIZE", "1000000"); + vi.stubEnv("OTEL_BSP_SCHEDULE_DELAY", "5000"); + const { processor: downstream, forceFlush } = fakeProcessor(); + const processor = new CompositeSpanProcessor([downstream], undefined); + + const now = vi.spyOn(Date, "now"); + now.mockReturnValue(1000); + processor.onEnd(createChildEnd("0000000000000001")); + processor.onEnd(createChildEnd("0000000000000002")); + expect(forceFlush).toHaveBeenCalledTimes(0); + + now.mockReturnValue(7000); + processor.onEnd(createChildEnd("0000000000000003")); + expect(forceFlush).toHaveBeenCalledTimes(1); + }); + + it("does not flush off-Vercel (no request context)", () => { + installAmbient(undefined); + vi.stubEnv("OTEL_BSP_MAX_EXPORT_BATCH_SIZE", "2"); + const { processor: downstream, forceFlush } = fakeProcessor(); + const processor = new CompositeSpanProcessor([downstream], undefined); + + for (let i = 1; i <= 5; i++) { + processor.onEnd(createChildEnd(`00000000000000${i}0`)); + } + expect(forceFlush).toHaveBeenCalledTimes(0); + }); +}); diff --git a/packages/otel/src/processor/composite-span-processor.ts b/packages/otel/src/processor/composite-span-processor.ts index 7ca514a..cd94041 100644 --- a/packages/otel/src/processor/composite-span-processor.ts +++ b/packages/otel/src/processor/composite-span-processor.ts @@ -5,7 +5,12 @@ import type { ReadableSpan, SpanProcessor, } from "@opentelemetry/sdk-trace-base"; -import { getVercelRequestContext } from "../vercel-request-context/api"; +import { getNumberFromEnv } from "@opentelemetry/core"; +import { + finalizeTrace, + getVercelRequestContext, + registerVercelRequestContextForTrace, +} from "../vercel-request-context/api"; import { getVercelRequestContextAttributes } from "../vercel-request-context/attributes"; import { isSampled } from "../util/sampled"; import type { AttributesFromHeaders } from "../types"; @@ -18,6 +23,13 @@ export class CompositeSpanProcessor implements SpanProcessor { >(); private readonly waitSpanEnd = new Map void>(); + private readonly flushBatchSize = + getNumberFromEnv("OTEL_BSP_MAX_EXPORT_BATCH_SIZE") ?? 512; + private readonly flushDelayMs = + getNumberFromEnv("OTEL_BSP_SCHEDULE_DELAY") ?? 5000; + private spansSinceFlush = 0; + private lastFlushMs = 0; + constructor( private processors: SpanProcessor[], private attributesFromHeaders: AttributesFromHeaders | undefined, @@ -58,8 +70,16 @@ export class CompositeSpanProcessor implements SpanProcessor { span.setAttributes(vercelRequestContextAttrs); } - // Flush the streams to avoid data loss. if (vrc) { + // Capture the request context that owns this trace. The exporter + // accumulates this trace's spans (mid-run flushes only drain the + // BatchSpanProcessor's queue, they don't report) and ships them in a + // single report when the trace is finalized below: the runtime only + // reliably persists one report per invocation, and only through the + // invocation's own telemetry channel. + registerVercelRequestContextForTrace(traceId, vrc); + + // Flush the streams to avoid data loss. vrc.waitUntil(async () => { if (this.rootSpanIds.has(traceId)) { // Not root has not completed yet, so no point in flushing. @@ -81,7 +101,15 @@ export class CompositeSpanProcessor implements SpanProcessor { clearTimeout(timer); } } - return this.forceFlush(); + try { + // Drain the processors so every one of this trace's remaining + // spans reaches the exporter's per-trace buffer... + return await this.forceFlush(); + } finally { + // ...then ship the buffer in one report and drop the captured + // context. + finalizeTrace(traceId); + } }); } } @@ -135,6 +163,33 @@ export class CompositeSpanProcessor implements SpanProcessor { this.waitSpanEnd.delete(traceId); pending(); } + } else if (sampled && getVercelRequestContext()) { + // Periodic drain, Vercel-only. Long-running invocations produce more + // spans than the BatchSpanProcessor's queue holds (it silently drops + // past maxQueueSize), so drain it regularly into the exporter's + // per-trace buffers. Draining doesn't report anything: the trace's + // spans ship in a single report at finalizeTrace. Off-Vercel this is + // skipped and the stock timer behaves as before. The root span's final + // flush is handled by the waitUntil registered in onStart. + this.maybeFlush(); + } + } + + private maybeFlush(): void { + this.spansSinceFlush += 1; + const now = Date.now(); + if (this.lastFlushMs === 0) { + this.lastFlushMs = now; + } + if ( + this.spansSinceFlush >= this.flushBatchSize || + now - this.lastFlushMs >= this.flushDelayMs + ) { + this.spansSinceFlush = 0; + this.lastFlushMs = now; + void this.forceFlush().catch((e) => { + diag.error("@vercel/otel: periodic flush failed:", e); + }); } } } diff --git a/packages/otel/src/vercel-request-context/api.ts b/packages/otel/src/vercel-request-context/api.ts index f173fc8..36adc80 100644 --- a/packages/otel/src/vercel-request-context/api.ts +++ b/packages/otel/src/vercel-request-context/api.ts @@ -35,3 +35,86 @@ export function getVercelRequestContext(): VercelRequestContext | undefined { const reader = (globalThis as GlobalWithReader)[symbol]; return reader?.get(); } + +/** + * Registry of request contexts keyed by trace id, so span exports can be + * attributed to the request that OWNS each span instead of whichever request + * happens to be ambient at flush time. + * + * Why this exists: the BatchSpanProcessor's queue is shared by every request + * running on the function instance, and its scheduled flush runs in a bare + * `setTimeout` OUTSIDE any request's AsyncLocalStorage. The runtime only + * reliably persists spans reported through their own invocation's + * `telemetry.reportSpans` channel, and only ONE report per invocation is + * dependable (mid-run reports have been observed not to land). Long-running + * invocations, e.g. Vercel Workflows, used to lose every span that was + * flushed mid-run: their traces looked blank except for the last few seconds. + * + * The context is captured per trace while inside the owning request (root + * span `onStart`), spans for registered traces are accumulated by the + * exporter, and the trace's whole buffer is shipped in a single report by + * `finalizeTrace` (invoked from the request's `waitUntil`, after the final + * flush). A hard cap bounds the registry against traces whose root span + * never ends. + */ +const traceContextRegistry = new Map(); + +const MAX_REGISTRY_SIZE = 1024; + +type TraceFinalizer = (traceId: string) => void; + +/** Callbacks (e.g. the runtime span exporter) shipping a trace's buffer. */ +const traceFinalizers = new Set(); + +/** @internal Register a callback invoked when a trace is finalized. */ +export function onTraceFinalize(finalizer: TraceFinalizer): void { + traceFinalizers.add(finalizer); +} + +/** @internal Capture the request context that owns a given trace. */ +export function registerVercelRequestContextForTrace( + traceId: string, + context: VercelRequestContext, +): void { + if ( + traceContextRegistry.size >= MAX_REGISTRY_SIZE && + !traceContextRegistry.has(traceId) + ) { + // Evict the oldest entry (Map preserves insertion order) so a leak of + // never-ending traces cannot grow the registry unboundedly. + const oldest: string | undefined = traceContextRegistry.keys().next() + .value as string | undefined; + if (oldest !== undefined) { + traceContextRegistry.delete(oldest); + } + } + traceContextRegistry.set(traceId, context); +} + +/** @internal Whether a trace has a captured owning context. */ +export function hasVercelRequestContextForTrace(traceId: string): boolean { + return traceContextRegistry.has(traceId); +} + +/** @internal Resolve the request context that owns a trace. */ +export function getVercelRequestContextForTrace( + traceId: string, +): VercelRequestContext | undefined { + return traceContextRegistry.get(traceId); +} + +/** + * @internal Ship a trace's accumulated spans (via the registered finalizers) + * and drop its captured context. Called from the owning request's + * `waitUntil` after the final flush, so the single report this produces is + * sent while the invocation is still able to deliver telemetry. + */ +export function finalizeTrace(traceId: string): void { + try { + for (const finalizer of traceFinalizers) { + finalizer(traceId); + } + } finally { + traceContextRegistry.delete(traceId); + } +} \ No newline at end of file diff --git a/packages/otel/src/vercel-request-context/exporter.test.ts b/packages/otel/src/vercel-request-context/exporter.test.ts new file mode 100644 index 0000000..f326b22 --- /dev/null +++ b/packages/otel/src/vercel-request-context/exporter.test.ts @@ -0,0 +1,190 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { ReadableSpan } from "@opentelemetry/sdk-trace-base"; +import { ExportResultCode, hrTime } from "@opentelemetry/core"; +import { TraceFlags, SpanKind, SpanStatusCode } from "@opentelemetry/api"; +import type { IExportTraceServiceRequest } from "@opentelemetry/otlp-transformer/build/src/trace/internal-types"; +import { resourceFromAttributes } from "@opentelemetry/resources"; +import type { VercelRequestContext } from "./api"; +import { finalizeTrace, registerVercelRequestContextForTrace } from "./api"; +import { VercelRuntimeSpanExporter } from "./exporter"; + +const TRACE_A = "aa75cd9e534ff5e9ed78b4a0c706f0f2"; +const TRACE_B = "bb75cd9e534ff5e9ed78b4a0c706f0f2"; +const VRC_SYMBOL = Symbol.for("@vercel/request-context"); + +type GlobalWithReader = Record; + +function createSpan(spanId: string, traceId = TRACE_A): ReadableSpan { + const spanContext = { + traceId, + spanId, + traceFlags: TraceFlags.SAMPLED, + isRemote: false, + }; + return { + name: `span-${spanId}`, + kind: SpanKind.INTERNAL, + spanContext: () => spanContext, + parentSpanContext: undefined, + startTime: hrTime(1), + endTime: hrTime(2), + status: { code: SpanStatusCode.UNSET }, + attributes: {}, + links: [], + events: [], + duration: hrTime(1), + ended: true, + resource: resourceFromAttributes({}), + instrumentationScope: { name: "default" }, + droppedAttributesCount: 0, + droppedEventsCount: 0, + droppedLinksCount: 0, + }; +} + +function makeContext(): { + ctx: VercelRequestContext; + reportSpans: ReturnType; +} { + const reportSpans = vi.fn(); + const ctx: VercelRequestContext = { + waitUntil: () => undefined, + headers: {}, + url: "https://example.com", + telemetry: { reportSpans }, + }; + return { ctx, reportSpans }; +} + +function installAmbient(ctx: VercelRequestContext | undefined): void { + (globalThis as GlobalWithReader)[VRC_SYMBOL] = { get: () => ctx }; +} + +function shippedSpanIds( + reportSpans: ReturnType, + call = 0, +): string[] { + const data = reportSpans.mock.calls[call]?.[0] as IExportTraceServiceRequest; + return (data.resourceSpans ?? []).flatMap((rs) => + rs.scopeSpans.flatMap((ss) => + (ss.spans ?? []).map((s) => String(s.spanId)), + ), + ); +} + +describe("VercelRuntimeSpanExporter", () => { + afterEach(() => { + installAmbient(undefined); + // Clear any registrations left behind by a test. + finalizeTrace(TRACE_A); + finalizeTrace(TRACE_B); + vi.restoreAllMocks(); + }); + + it("accumulates spans of a registered trace and ships them once, through the owning context, at finalizeTrace", () => { + const owner = makeContext(); + const ambient = makeContext(); + registerVercelRequestContextForTrace(TRACE_A, owner.ctx); + installAmbient(ambient.ctx); + + const exporter = new VercelRuntimeSpanExporter(); + const result = vi.fn(); + // Two mid-run flushes: nothing must be reported yet. + exporter.export([createSpan("0000000000000001", TRACE_A)], result); + exporter.export([createSpan("0000000000000002", TRACE_A)], result); + expect(result).toHaveBeenCalledWith({ + code: ExportResultCode.SUCCESS, + error: undefined, + }); + expect(owner.reportSpans).toHaveBeenCalledTimes(0); + expect(ambient.reportSpans).toHaveBeenCalledTimes(0); + + // Finalize: exactly ONE report, through the owner, with all spans. + finalizeTrace(TRACE_A); + expect(owner.reportSpans).toHaveBeenCalledTimes(1); + expect(ambient.reportSpans).toHaveBeenCalledTimes(0); + expect(shippedSpanIds(owner.reportSpans)).toHaveLength(2); + + // Finalizing again reports nothing further. + finalizeTrace(TRACE_A); + expect(owner.reportSpans).toHaveBeenCalledTimes(1); + }); + + it("keeps concurrent traces separate: each finalize ships only its own spans", () => { + const ownerA = makeContext(); + const ownerB = makeContext(); + registerVercelRequestContextForTrace(TRACE_A, ownerA.ctx); + registerVercelRequestContextForTrace(TRACE_B, ownerB.ctx); + + const exporter = new VercelRuntimeSpanExporter(); + exporter.export( + [ + createSpan("0000000000000001", TRACE_A), + createSpan("0000000000000002", TRACE_B), + createSpan("0000000000000003", TRACE_A), + ], + vi.fn(), + ); + + finalizeTrace(TRACE_A); + expect(ownerA.reportSpans).toHaveBeenCalledTimes(1); + expect(shippedSpanIds(ownerA.reportSpans)).toHaveLength(2); + expect(ownerB.reportSpans).toHaveBeenCalledTimes(0); + + finalizeTrace(TRACE_B); + expect(ownerB.reportSpans).toHaveBeenCalledTimes(1); + expect(shippedSpanIds(ownerB.reportSpans)).toHaveLength(1); + }); + + it("ships spans of untracked traces immediately through the ambient context", () => { + const ambient = makeContext(); + installAmbient(ambient.ctx); + + const exporter = new VercelRuntimeSpanExporter(); + exporter.export([createSpan("0000000000000001", TRACE_B)], vi.fn()); + + expect(ambient.reportSpans).toHaveBeenCalledTimes(1); + expect(shippedSpanIds(ambient.reportSpans)).toHaveLength(1); + }); + + it("retains unattributable spans and re-attempts them on a later flush", () => { + const exporter = new VercelRuntimeSpanExporter(); + + installAmbient(undefined); + const firstResult = vi.fn(); + exporter.export([createSpan("0000000000000001", TRACE_B)], firstResult); + expect(firstResult).toHaveBeenCalledWith({ + code: ExportResultCode.SUCCESS, + error: undefined, + }); + + const { ctx, reportSpans } = makeContext(); + installAmbient(ctx); + exporter.export([createSpan("0000000000000002", TRACE_B)], vi.fn()); + + // Retained + incoming ship together via the now-available ambient context. + expect(reportSpans).toHaveBeenCalledTimes(1); + expect(shippedSpanIds(reportSpans)).toHaveLength(2); + }); + + it("retains a finalized trace's spans when its context fails, without losing them", () => { + const failing = makeContext(); + failing.reportSpans.mockImplementation(() => { + throw new Error("boom"); + }); + registerVercelRequestContextForTrace(TRACE_A, failing.ctx); + installAmbient(undefined); + + const exporter = new VercelRuntimeSpanExporter(); + exporter.export([createSpan("0000000000000001", TRACE_A)], vi.fn()); + finalizeTrace(TRACE_A); + expect(failing.reportSpans).toHaveBeenCalledTimes(1); + + // The retained spans ship on the next flush that has a working context. + const healthy = makeContext(); + installAmbient(healthy.ctx); + exporter.export([], vi.fn()); + expect(healthy.reportSpans).toHaveBeenCalledTimes(1); + expect(shippedSpanIds(healthy.reportSpans)).toHaveLength(1); + }); +}); \ No newline at end of file diff --git a/packages/otel/src/vercel-request-context/exporter.ts b/packages/otel/src/vercel-request-context/exporter.ts index ff9f959..f66b727 100644 --- a/packages/otel/src/vercel-request-context/exporter.ts +++ b/packages/otel/src/vercel-request-context/exporter.ts @@ -3,32 +3,100 @@ import type { ReadableSpan, SpanExporter } from "@opentelemetry/sdk-trace-base"; import { ExportResultCode, type ExportResult } from "@opentelemetry/core"; import { JsonTraceSerializer } from "@opentelemetry/otlp-transformer/build/src/trace/json/trace"; import type { IExportTraceServiceRequest } from "@opentelemetry/otlp-transformer/build/src/trace/internal-types"; -import { getVercelRequestContext } from "./api"; +import { + getVercelRequestContext, + getVercelRequestContextForTrace, + hasVercelRequestContextForTrace, + onTraceFinalize, + type VercelRequestContext, +} from "./api"; +/** + * Upper bound on spans accumulated per trace while its invocation runs. + * A long-running invocation (e.g. a Vercel Workflow) can produce thousands + * of spans; when the cap is hit the oldest spans are dropped with a warning + * rather than growing memory unboundedly. + */ +const MAX_TRACE_BUFFER_SIZE = 8192; + +/** + * Ships spans to the Vercel runtime's telemetry channel. + * + * The runtime only reliably persists ONE `reportSpans` payload per + * invocation, and only when reported through that invocation's own request + * context. So instead of shipping every batch as it is flushed (which loses + * every mid-run batch of a long invocation, and mis-routes spans of + * concurrent requests sharing the BatchSpanProcessor queue), this exporter: + * + * - ACCUMULATES spans whose trace has a captured owning context (see + * `registerVercelRequestContextForTrace`) into a per-trace buffer, and + * ships the whole buffer in a single report when the trace is finalized + * from its own request's `waitUntil` (see `finalizeTrace`). + * - Ships spans of untracked traces immediately through the ambient request + * context, preserving the previous behavior for spans created outside any + * instrumented request. + * - Retains spans it cannot attribute at all and re-attempts them on the + * next flush, instead of silently acking and dropping them. + */ export class VercelRuntimeSpanExporter implements SpanExporter { + private readonly traceBuffers = new Map(); + private pendingSpans: ReadableSpan[] = []; + private droppedSpansCount = 0; + + constructor() { + onTraceFinalize((traceId) => this.shipTrace(traceId)); + } + export( spans: ReadableSpan[], resultCallback: (result: ExportResult) => void, ): void { - const context = getVercelRequestContext(); - if (!context?.telemetry) { - diag.warn("@vercel/otel: no telemetry context found"); - resultCallback({ code: ExportResultCode.SUCCESS, error: undefined }); - return; - } + const immediate = new Map(); + const unattributable: ReadableSpan[] = []; - try { - const serializedData = JsonTraceSerializer.serializeRequest(spans); + // Previously-retained spans get another attribution attempt first. + const candidates = + this.pendingSpans.length > 0 ? [...this.pendingSpans, ...spans] : spans; + this.pendingSpans = []; - if (!serializedData) { - throw new Error("Failed to serialize spans"); + for (const span of candidates) { + const traceId = span.spanContext().traceId; + if (hasVercelRequestContextForTrace(traceId)) { + this.bufferForTrace(traceId, span); + continue; } - // Convert back to object format for the Vercel telemetry API - const data = JSON.parse( - new TextDecoder().decode(serializedData), - ) as IExportTraceServiceRequest; + const ambient = getVercelRequestContext(); + if (ambient?.telemetry) { + const group = immediate.get(ambient); + if (group) { + group.push(span); + } else { + immediate.set(ambient, [span]); + } + continue; + } + unattributable.push(span); + } - context.telemetry.reportSpans(data); + if (unattributable.length > 0) { + // No owning or ambient context (e.g. a bare-setTimeout timer flush for + // an untracked trace). Retain and re-attempt on the next flush instead + // of silently dropping. + diag.debug( + `@vercel/otel: no telemetry context for ${unattributable.length} span(s); retaining for the next flush`, + ); + this.retain(unattributable); + } + + try { + immediate.forEach((contextSpans, context) => { + try { + reportSpans(context.telemetry, contextSpans); + } catch (e) { + this.retain(contextSpans); + diag.warn("@vercel/otel: failed to report spans, retained:", e); + } + }); resultCallback({ code: ExportResultCode.SUCCESS, error: undefined }); } catch (e) { resultCallback({ @@ -38,6 +106,63 @@ export class VercelRuntimeSpanExporter implements SpanExporter { } } + /** + * Ship a trace's accumulated spans in a single report through the context + * that owns the trace. Invoked via `finalizeTrace` from the owning + * request's `waitUntil`, after the final flush drained the processor. + */ + private shipTrace(traceId: string): void { + const spans = this.traceBuffers.get(traceId); + this.traceBuffers.delete(traceId); + if (!spans || spans.length === 0) { + return; + } + const context = + getVercelRequestContextForTrace(traceId) ?? getVercelRequestContext(); + if (!context?.telemetry) { + diag.warn( + `@vercel/otel: no telemetry context to finalize trace ${traceId}; retaining ${spans.length} span(s)`, + ); + this.retain(spans); + return; + } + try { + reportSpans(context.telemetry, spans); + } catch (e) { + this.retain(spans); + diag.warn("@vercel/otel: failed to report finalized trace, retained:", e); + } + } + + private bufferForTrace(traceId: string, span: ReadableSpan): void { + let buffer = this.traceBuffers.get(traceId); + if (!buffer) { + buffer = []; + this.traceBuffers.set(traceId, buffer); + } + buffer.push(span); + const overflow = buffer.length - MAX_TRACE_BUFFER_SIZE; + if (overflow > 0) { + buffer.splice(0, overflow); + this.droppedSpansCount += overflow; + diag.warn( + `@vercel/otel: trace span buffer full, dropped ${this.droppedSpansCount} span(s) total`, + ); + } + } + + private retain(spans: ReadableSpan[]): void { + this.pendingSpans.push(...spans); + const overflow = this.pendingSpans.length - MAX_TRACE_BUFFER_SIZE; + if (overflow > 0) { + this.pendingSpans.splice(0, overflow); + this.droppedSpansCount += overflow; + diag.warn( + `@vercel/otel: retained span buffer full, dropped ${this.droppedSpansCount} span(s) total`, + ); + } + } + shutdown(): Promise { return Promise.resolve(); } @@ -46,3 +171,21 @@ export class VercelRuntimeSpanExporter implements SpanExporter { return Promise.resolve(); } } + +function reportSpans( + telemetry: NonNullable | undefined, + spans: ReadableSpan[], +): void { + if (!telemetry) { + throw new Error("no telemetry channel"); + } + const serializedData = JsonTraceSerializer.serializeRequest(spans); + if (!serializedData) { + throw new Error("Failed to serialize spans"); + } + // Convert back to object format for the Vercel telemetry API + const data = JSON.parse( + new TextDecoder().decode(serializedData), + ) as IExportTraceServiceRequest; + telemetry.reportSpans(data); +} \ No newline at end of file From f7114056b34a3512fbc7994695a7ce8549d549a6 Mon Sep 17 00:00:00 2001 From: Luc Leray Date: Wed, 8 Jul 2026 12:11:49 +0200 Subject: [PATCH 2/2] fix(otel): stream spans as strict per-trace payloads through owning contexts Variation of the accumulate-and-ship-once approach (#214) that also supports exporting spans DURING the invocation, not only at its end. Empirical finding behind the design: payloads composed of a SINGLE trace that belongs to an active request are the shape the delivery pipeline handles reliably. What poisoned mid-run delivery before was payload mixing: the BatchSpanProcessor's queue is shared by all concurrent requests, so its flushes shipped mixed-trace payloads whose delivery was non-deterministic, and stale retained spans leaking into active payloads recreated the same poisoning even with client-side routing. The exporter now enforces strict per-trace payload discipline: - Every flush groups incoming spans by their own trace id and ships each trace's spans as their own single-trace payload, through the telemetry channel captured for that trace at root-span start (ambient fallback for untracked traces). Traces never share a payload. - Spans with no reachable channel are buffered per trace (bounded, oldest dropped with a warning) and retried on the next flush; the trace's finalize (from the owning request's waitUntil) ships whatever is left. - The composite processor still drains the BSP queue in-context on span end (debounced on OTEL_BSP_MAX_EXPORT_BATCH_SIZE / OTEL_BSP_SCHEDULE_DELAY) so long runs cannot overflow maxQueueSize, and registers/finalizes the per-trace context as in #214. Streaming is opt-in via VERCEL_OTEL_STREAM_SPANS=1 and not recommended yet: parts of the platform-side delivery pipeline currently only reliably persist batches delivered near the end of the request, so the default remains accumulate-and-ship-once. Once that is addressed, flipping the flag gives mid-run streaming without loss; spans then leave the process every flush interval, bounding client memory on long runs. --- .changeset/fix-span-context-routing.md | 5 - .changeset/fix-span-streaming.md | 5 + .../src/processor/composite-span-processor.ts | 23 ++- .../otel/src/vercel-request-context/api.ts | 37 ++-- .../vercel-request-context/exporter.test.ts | 172 +++++++++++++----- .../src/vercel-request-context/exporter.ts | 166 +++++++++-------- 6 files changed, 241 insertions(+), 167 deletions(-) delete mode 100644 .changeset/fix-span-context-routing.md create mode 100644 .changeset/fix-span-streaming.md diff --git a/.changeset/fix-span-context-routing.md b/.changeset/fix-span-context-routing.md deleted file mode 100644 index 8179130..0000000 --- a/.changeset/fix-span-context-routing.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"@vercel/otel": patch ---- - -Fix long-running invocations (e.g. Vercel Workflows) losing most spans from the first part of the run. The runtime only reliably persists one `reportSpans` payload per invocation, delivered through that invocation's own request context, while the BatchSpanProcessor flushed many times mid-run (from a bare `setTimeout` outside the request's AsyncLocalStorage) and shipped whole batches through whichever context was ambient. Spans are now attributed to the request context that owns their trace (captured at root-span start), accumulated per trace, and shipped in a single report when the trace is finalized from its own request's `waitUntil`. The processor also drains the BatchSpanProcessor queue in-context on span end so long runs cannot overflow it, and the exporter retains spans it cannot attribute instead of silently dropping them. diff --git a/.changeset/fix-span-streaming.md b/.changeset/fix-span-streaming.md new file mode 100644 index 0000000..81570d6 --- /dev/null +++ b/.changeset/fix-span-streaming.md @@ -0,0 +1,5 @@ +--- +"@vercel/otel": patch +--- + +Fix long-running invocations (e.g. Vercel Workflows) losing most spans from the first part of the run. The exporter now enforces strict per-trace `reportSpans` payloads: spans are grouped by their own trace id, buffered per trace, and shipped in one payload at the trace's final flush through the request context captured for that trace at root-span start. Untracked traces keep shipping immediately through the ambient context, unattributable spans are retained and retried instead of silently dropped, and the processor drains the BatchSpanProcessor queue in-context on span end so long runs cannot overflow its maxQueueSize. An opt-in streaming mode (`VERCEL_OTEL_STREAM_SPANS=1`) ships every trace's spans on every flush instead. diff --git a/packages/otel/src/processor/composite-span-processor.ts b/packages/otel/src/processor/composite-span-processor.ts index cd94041..30c1cb9 100644 --- a/packages/otel/src/processor/composite-span-processor.ts +++ b/packages/otel/src/processor/composite-span-processor.ts @@ -72,11 +72,10 @@ export class CompositeSpanProcessor implements SpanProcessor { if (vrc) { // Capture the request context that owns this trace. The exporter - // accumulates this trace's spans (mid-run flushes only drain the - // BatchSpanProcessor's queue, they don't report) and ships them in a - // single report when the trace is finalized below: the runtime only - // reliably persists one report per invocation, and only through the - // invocation's own telemetry channel. + // streams this trace's spans as single-trace payloads through this + // context's telemetry channel on every flush (the runtime only + // reliably persists payloads composed of one active trace), and the + // finalize below ships whatever is left when the request ends. registerVercelRequestContextForTrace(traceId, vrc); // Flush the streams to avoid data loss. @@ -164,13 +163,13 @@ export class CompositeSpanProcessor implements SpanProcessor { pending(); } } else if (sampled && getVercelRequestContext()) { - // Periodic drain, Vercel-only. Long-running invocations produce more - // spans than the BatchSpanProcessor's queue holds (it silently drops - // past maxQueueSize), so drain it regularly into the exporter's - // per-trace buffers. Draining doesn't report anything: the trace's - // spans ship in a single report at finalizeTrace. Off-Vercel this is - // skipped and the stock timer behaves as before. The root span's final - // flush is handled by the waitUntil registered in onStart. + // Periodic in-context drain, Vercel-only. Long-running invocations + // produce more spans than the BatchSpanProcessor's queue holds (it + // silently drops past maxQueueSize), so drain it regularly into the + // exporter, which streams each trace's spans out as single-trace + // payloads. Off-Vercel this is skipped and the stock timer behaves as + // before. The root span's final flush is handled by the waitUntil + // registered in onStart. this.maybeFlush(); } } diff --git a/packages/otel/src/vercel-request-context/api.ts b/packages/otel/src/vercel-request-context/api.ts index 36adc80..2057b25 100644 --- a/packages/otel/src/vercel-request-context/api.ts +++ b/packages/otel/src/vercel-request-context/api.ts @@ -37,25 +37,26 @@ export function getVercelRequestContext(): VercelRequestContext | undefined { } /** - * Registry of request contexts keyed by trace id, so span exports can be - * attributed to the request that OWNS each span instead of whichever request - * happens to be ambient at flush time. + * Registry of request contexts keyed by trace id, so span exports can reach + * the telemetry channel of the request that OWNS each span even when the + * flush runs outside any request context. * * Why this exists: the BatchSpanProcessor's queue is shared by every request * running on the function instance, and its scheduled flush runs in a bare - * `setTimeout` OUTSIDE any request's AsyncLocalStorage. The runtime only - * reliably persists spans reported through their own invocation's - * `telemetry.reportSpans` channel, and only ONE report per invocation is - * dependable (mid-run reports have been observed not to land). Long-running - * invocations, e.g. Vercel Workflows, used to lose every span that was - * flushed mid-run: their traces looked blank except for the last few seconds. + * `setTimeout` OUTSIDE any request's AsyncLocalStorage, where the ambient + * context resolves to nothing (the exporter used to silently drop those + * batches). The runtime also only reliably persists payloads composed of a + * SINGLE trace belonging to an active request; mixed-queue payloads used to + * be dropped whole or misfiled depending on which span sat first in the + * batch. Long-running invocations, e.g. Vercel Workflows, used to lose every + * span flushed mid-run: their traces looked blank except for the last few + * seconds. * * The context is captured per trace while inside the owning request (root - * span `onStart`), spans for registered traces are accumulated by the - * exporter, and the trace's whole buffer is shipped in a single report by - * `finalizeTrace` (invoked from the request's `waitUntil`, after the final - * flush). A hard cap bounds the registry against traces whose root span - * never ends. + * span `onStart`); the exporter streams each trace's spans as single-trace + * payloads through it, and `finalizeTrace` (invoked from the request's + * `waitUntil`, after the final flush) ships whatever is left. A hard cap + * bounds the registry against traces whose root span never ends. */ const traceContextRegistry = new Map(); @@ -104,10 +105,10 @@ export function getVercelRequestContextForTrace( } /** - * @internal Ship a trace's accumulated spans (via the registered finalizers) - * and drop its captured context. Called from the owning request's - * `waitUntil` after the final flush, so the single report this produces is - * sent while the invocation is still able to deliver telemetry. + * @internal Ship a trace's remaining buffered spans (via the registered + * finalizers) and drop its captured context. Called from the owning + * request's `waitUntil` after the final flush, so the last payload is sent + * while the invocation is still able to deliver telemetry. */ export function finalizeTrace(traceId: string): void { try { diff --git a/packages/otel/src/vercel-request-context/exporter.test.ts b/packages/otel/src/vercel-request-context/exporter.test.ts index f326b22..abcac0d 100644 --- a/packages/otel/src/vercel-request-context/exporter.test.ts +++ b/packages/otel/src/vercel-request-context/exporter.test.ts @@ -60,19 +60,35 @@ function installAmbient(ctx: VercelRequestContext | undefined): void { (globalThis as GlobalWithReader)[VRC_SYMBOL] = { get: () => ctx }; } -function shippedSpanIds( +function shippedTraceIds( reportSpans: ReturnType, call = 0, ): string[] { const data = reportSpans.mock.calls[call]?.[0] as IExportTraceServiceRequest; return (data.resourceSpans ?? []).flatMap((rs) => rs.scopeSpans.flatMap((ss) => - (ss.spans ?? []).map((s) => String(s.spanId)), + (ss.spans ?? []).map((s) => String(s.traceId)), ), ); } -describe("VercelRuntimeSpanExporter", () => { +function shippedSpanCount( + reportSpans: ReturnType, + call = 0, +): number { + return shippedTraceIds(reportSpans, call).length; +} + +function makeStreamingExporter(): VercelRuntimeSpanExporter { + process.env.VERCEL_OTEL_STREAM_SPANS = "1"; + try { + return new VercelRuntimeSpanExporter(); + } finally { + delete process.env.VERCEL_OTEL_STREAM_SPANS; + } +} + +describe("VercelRuntimeSpanExporter (streaming mode, VERCEL_OTEL_STREAM_SPANS=1)", () => { afterEach(() => { installAmbient(undefined); // Clear any registrations left behind by a test. @@ -81,42 +97,37 @@ describe("VercelRuntimeSpanExporter", () => { vi.restoreAllMocks(); }); - it("accumulates spans of a registered trace and ships them once, through the owning context, at finalizeTrace", () => { + it("streams a registered trace's spans on every flush, as single-trace payloads", () => { const owner = makeContext(); - const ambient = makeContext(); registerVercelRequestContextForTrace(TRACE_A, owner.ctx); - installAmbient(ambient.ctx); + installAmbient(undefined); - const exporter = new VercelRuntimeSpanExporter(); + const exporter = makeStreamingExporter(); const result = vi.fn(); - // Two mid-run flushes: nothing must be reported yet. exporter.export([createSpan("0000000000000001", TRACE_A)], result); exporter.export([createSpan("0000000000000002", TRACE_A)], result); + expect(result).toHaveBeenCalledWith({ code: ExportResultCode.SUCCESS, error: undefined, }); - expect(owner.reportSpans).toHaveBeenCalledTimes(0); - expect(ambient.reportSpans).toHaveBeenCalledTimes(0); - - // Finalize: exactly ONE report, through the owner, with all spans. - finalizeTrace(TRACE_A); - expect(owner.reportSpans).toHaveBeenCalledTimes(1); - expect(ambient.reportSpans).toHaveBeenCalledTimes(0); - expect(shippedSpanIds(owner.reportSpans)).toHaveLength(2); + // Two mid-run flushes -> two reports, each shipped immediately. + expect(owner.reportSpans).toHaveBeenCalledTimes(2); + expect(shippedSpanCount(owner.reportSpans, 0)).toBe(1); + expect(shippedSpanCount(owner.reportSpans, 1)).toBe(1); - // Finalizing again reports nothing further. + // Finalize with an empty buffer reports nothing further. finalizeTrace(TRACE_A); - expect(owner.reportSpans).toHaveBeenCalledTimes(1); + expect(owner.reportSpans).toHaveBeenCalledTimes(2); }); - it("keeps concurrent traces separate: each finalize ships only its own spans", () => { + it("never mixes traces into one payload: each trace ships separately to its own context", () => { const ownerA = makeContext(); const ownerB = makeContext(); registerVercelRequestContextForTrace(TRACE_A, ownerA.ctx); registerVercelRequestContextForTrace(TRACE_B, ownerB.ctx); - const exporter = new VercelRuntimeSpanExporter(); + const exporter = makeStreamingExporter(); exporter.export( [ createSpan("0000000000000001", TRACE_A), @@ -126,29 +137,39 @@ describe("VercelRuntimeSpanExporter", () => { vi.fn(), ); - finalizeTrace(TRACE_A); expect(ownerA.reportSpans).toHaveBeenCalledTimes(1); - expect(shippedSpanIds(ownerA.reportSpans)).toHaveLength(2); - expect(ownerB.reportSpans).toHaveBeenCalledTimes(0); - - finalizeTrace(TRACE_B); expect(ownerB.reportSpans).toHaveBeenCalledTimes(1); - expect(shippedSpanIds(ownerB.reportSpans)).toHaveLength(1); + expect(new Set(shippedTraceIds(ownerA.reportSpans))).toEqual( + new Set([TRACE_A]), + ); + expect(shippedSpanCount(ownerA.reportSpans)).toBe(2); + expect(new Set(shippedTraceIds(ownerB.reportSpans))).toEqual( + new Set([TRACE_B]), + ); + expect(shippedSpanCount(ownerB.reportSpans)).toBe(1); }); - it("ships spans of untracked traces immediately through the ambient context", () => { + it("ships untracked traces through the ambient context, still one payload per trace", () => { const ambient = makeContext(); installAmbient(ambient.ctx); - const exporter = new VercelRuntimeSpanExporter(); - exporter.export([createSpan("0000000000000001", TRACE_B)], vi.fn()); + const exporter = makeStreamingExporter(); + exporter.export( + [ + createSpan("0000000000000001", TRACE_A), + createSpan("0000000000000002", TRACE_B), + ], + vi.fn(), + ); - expect(ambient.reportSpans).toHaveBeenCalledTimes(1); - expect(shippedSpanIds(ambient.reportSpans)).toHaveLength(1); + // Two payloads (one per trace), never a mixed one. + expect(ambient.reportSpans).toHaveBeenCalledTimes(2); + expect(new Set(shippedTraceIds(ambient.reportSpans, 0)).size).toBe(1); + expect(new Set(shippedTraceIds(ambient.reportSpans, 1)).size).toBe(1); }); - it("retains unattributable spans and re-attempts them on a later flush", () => { - const exporter = new VercelRuntimeSpanExporter(); + it("buffers spans with no reachable channel and ships them on a later flush", () => { + const exporter = makeStreamingExporter(); installAmbient(undefined); const firstResult = vi.fn(); @@ -162,29 +183,84 @@ describe("VercelRuntimeSpanExporter", () => { installAmbient(ctx); exporter.export([createSpan("0000000000000002", TRACE_B)], vi.fn()); - // Retained + incoming ship together via the now-available ambient context. + // Buffered + incoming ship together (same trace, one payload). expect(reportSpans).toHaveBeenCalledTimes(1); - expect(shippedSpanIds(reportSpans)).toHaveLength(2); + expect(shippedSpanCount(reportSpans)).toBe(2); + }); + + it("ships leftovers at finalizeTrace through the owning context", () => { + const exporter = makeStreamingExporter(); + + // No channel at flush time: spans stay buffered. + installAmbient(undefined); + exporter.export([createSpan("0000000000000001", TRACE_A)], vi.fn()); + + // The trace's request registers + finalizes (waitUntil path). + const owner = makeContext(); + registerVercelRequestContextForTrace(TRACE_A, owner.ctx); + finalizeTrace(TRACE_A); + + expect(owner.reportSpans).toHaveBeenCalledTimes(1); + expect(shippedSpanCount(owner.reportSpans)).toBe(1); }); - it("retains a finalized trace's spans when its context fails, without losing them", () => { - const failing = makeContext(); - failing.reportSpans.mockImplementation(() => { + it("retains a trace's spans when its channel throws and retries on the next flush", () => { + const owner = makeContext(); + owner.reportSpans.mockImplementationOnce(() => { throw new Error("boom"); }); - registerVercelRequestContextForTrace(TRACE_A, failing.ctx); + registerVercelRequestContextForTrace(TRACE_A, owner.ctx); + + const exporter = makeStreamingExporter(); + const result = vi.fn(); + exporter.export([createSpan("0000000000000001", TRACE_A)], result); + expect(result).toHaveBeenCalledWith({ + code: ExportResultCode.SUCCESS, + error: undefined, + }); + expect(owner.reportSpans).toHaveBeenCalledTimes(1); + + // Next flush retries the retained span together with the new one. + exporter.export([createSpan("0000000000000002", TRACE_A)], vi.fn()); + expect(owner.reportSpans).toHaveBeenCalledTimes(2); + expect(shippedSpanCount(owner.reportSpans, 1)).toBe(2); + }); +}); + +describe("VercelRuntimeSpanExporter (default accumulate mode)", () => { + afterEach(() => { installAmbient(undefined); + finalizeTrace(TRACE_A); + finalizeTrace(TRACE_B); + vi.restoreAllMocks(); + }); + + it("accumulates a registered trace's spans and ships them once at finalizeTrace", () => { + const owner = makeContext(); + const ambient = makeContext(); + registerVercelRequestContextForTrace(TRACE_A, owner.ctx); + installAmbient(ambient.ctx); const exporter = new VercelRuntimeSpanExporter(); exporter.export([createSpan("0000000000000001", TRACE_A)], vi.fn()); + exporter.export([createSpan("0000000000000002", TRACE_A)], vi.fn()); + // Nothing ships mid-run in accumulate mode. + expect(owner.reportSpans).toHaveBeenCalledTimes(0); + expect(ambient.reportSpans).toHaveBeenCalledTimes(0); + finalizeTrace(TRACE_A); - expect(failing.reportSpans).toHaveBeenCalledTimes(1); - - // The retained spans ship on the next flush that has a working context. - const healthy = makeContext(); - installAmbient(healthy.ctx); - exporter.export([], vi.fn()); - expect(healthy.reportSpans).toHaveBeenCalledTimes(1); - expect(shippedSpanIds(healthy.reportSpans)).toHaveLength(1); + expect(owner.reportSpans).toHaveBeenCalledTimes(1); + expect(shippedSpanCount(owner.reportSpans)).toBe(2); + }); + + it("still ships untracked traces immediately via the ambient context", () => { + const ambient = makeContext(); + installAmbient(ambient.ctx); + + const exporter = new VercelRuntimeSpanExporter(); + exporter.export([createSpan("0000000000000001", TRACE_B)], vi.fn()); + + expect(ambient.reportSpans).toHaveBeenCalledTimes(1); + expect(shippedSpanCount(ambient.reportSpans)).toBe(1); }); -}); \ No newline at end of file +}); diff --git a/packages/otel/src/vercel-request-context/exporter.ts b/packages/otel/src/vercel-request-context/exporter.ts index f66b727..3fd1b33 100644 --- a/packages/otel/src/vercel-request-context/exporter.ts +++ b/packages/otel/src/vercel-request-context/exporter.ts @@ -6,42 +6,59 @@ import type { IExportTraceServiceRequest } from "@opentelemetry/otlp-transformer import { getVercelRequestContext, getVercelRequestContextForTrace, - hasVercelRequestContextForTrace, onTraceFinalize, type VercelRequestContext, } from "./api"; /** - * Upper bound on spans accumulated per trace while its invocation runs. - * A long-running invocation (e.g. a Vercel Workflow) can produce thousands - * of spans; when the cap is hit the oldest spans are dropped with a warning - * rather than growing memory unboundedly. + * Upper bound on spans buffered per trace while no telemetry channel is + * available to ship them. When the cap is hit the oldest spans are dropped + * with a warning rather than growing memory unboundedly. */ const MAX_TRACE_BUFFER_SIZE = 8192; /** * Ships spans to the Vercel runtime's telemetry channel. * - * The runtime only reliably persists ONE `reportSpans` payload per - * invocation, and only when reported through that invocation's own request - * context. So instead of shipping every batch as it is flushed (which loses - * every mid-run batch of a long invocation, and mis-routes spans of - * concurrent requests sharing the BatchSpanProcessor queue), this exporter: + * The runtime attributes a `reportSpans` payload from its content, and only + * payloads composed of a SINGLE trace that belongs to a currently-active + * request are reliably persisted end to end. Mixed-trace payloads (the OTel + * BatchSpanProcessor queue is shared by every concurrent request on the + * instance, so its flushes ship whatever is queued) used to be dropped whole + * or misfiled depending on which span happened to sit first in the batch. * - * - ACCUMULATES spans whose trace has a captured owning context (see - * `registerVercelRequestContextForTrace`) into a per-trace buffer, and - * ships the whole buffer in a single report when the trace is finalized - * from its own request's `waitUntil` (see `finalizeTrace`). - * - Ships spans of untracked traces immediately through the ambient request - * context, preserving the previous behavior for spans created outside any - * instrumented request. - * - Retains spans it cannot attribute at all and re-attempts them on the - * next flush, instead of silently acking and dropping them. + * So this exporter enforces strict per-trace payload discipline, streaming + * spans out as they are flushed: + * + * - Every incoming span is grouped by ITS OWN trace id, and each trace's + * spans ship as their own single-trace payload. Traces are never mixed + * into one payload, so one stale span can no longer poison other traces' + * delivery. + * - Spans ship EAGERLY on every flush, through the telemetry channel of the + * context captured for their trace (see + * `registerVercelRequestContextForTrace`), falling back to the ambient + * context for untracked traces. This bounds client memory on long + * invocations and gets spans out of the process while the run is still + * executing. + * - Spans with no reachable telemetry channel are buffered per trace + * (bounded) and retried on the next flush; the trace's final flush + * (`finalizeTrace`, from the owning request's `waitUntil`) ships whatever + * is left instead of silently dropping it. */ export class VercelRuntimeSpanExporter implements SpanExporter { private readonly traceBuffers = new Map(); - private pendingSpans: ReadableSpan[] = []; private droppedSpansCount = 0; + /** + * Opt-in streaming mode (`VERCEL_OTEL_STREAM_SPANS=1`): ship every trace's + * spans on every flush instead of accumulating until the trace's final + * flush. NOTE: today the platform drops full-size trace chunks that the + * runtime forwards mid-response, so streamed spans of long traces are + * lost past the runtime's chunk threshold; keep this OFF until that is + * fixed. Default (off) accumulates per trace and ships once at + * `finalizeTrace`, which is reliable end to end. + */ + private readonly streamSpans = + process.env.VERCEL_OTEL_STREAM_SPANS === "1"; constructor() { onTraceFinalize((traceId) => this.shipTrace(traceId)); @@ -51,52 +68,22 @@ export class VercelRuntimeSpanExporter implements SpanExporter { spans: ReadableSpan[], resultCallback: (result: ExportResult) => void, ): void { - const immediate = new Map(); - const unattributable: ReadableSpan[] = []; - - // Previously-retained spans get another attribution attempt first. - const candidates = - this.pendingSpans.length > 0 ? [...this.pendingSpans, ...spans] : spans; - this.pendingSpans = []; - - for (const span of candidates) { - const traceId = span.spanContext().traceId; - if (hasVercelRequestContextForTrace(traceId)) { - this.bufferForTrace(traceId, span); - continue; - } - const ambient = getVercelRequestContext(); - if (ambient?.telemetry) { - const group = immediate.get(ambient); - if (group) { - group.push(span); - } else { - immediate.set(ambient, [span]); - } - continue; - } - unattributable.push(span); - } - - if (unattributable.length > 0) { - // No owning or ambient context (e.g. a bare-setTimeout timer flush for - // an untracked trace). Retain and re-attempt on the next flush instead - // of silently dropping. - diag.debug( - `@vercel/otel: no telemetry context for ${unattributable.length} span(s); retaining for the next flush`, - ); - this.retain(unattributable); + // Group incoming spans per trace (strict: traces never share a payload). + for (const span of spans) { + this.bufferForTrace(span.spanContext().traceId, span); } try { - immediate.forEach((contextSpans, context) => { - try { - reportSpans(context.telemetry, contextSpans); - } catch (e) { - this.retain(contextSpans); - diag.warn("@vercel/otel: failed to report spans, retained:", e); + for (const traceId of Array.from(this.traceBuffers.keys())) { + if (this.streamSpans || !getVercelRequestContextForTrace(traceId)) { + // Streaming mode ships everything on every flush. Accumulate mode + // (default) keeps registered traces buffered until their + // finalizeTrace, and only ships untracked traces (no owning + // request captured, e.g. spans created outside an instrumented + // request) now via the ambient context. + this.shipTrace(traceId); } - }); + } resultCallback({ code: ExportResultCode.SUCCESS, error: undefined }); } catch (e) { resultCallback({ @@ -107,30 +94,36 @@ export class VercelRuntimeSpanExporter implements SpanExporter { } /** - * Ship a trace's accumulated spans in a single report through the context - * that owns the trace. Invoked via `finalizeTrace` from the owning - * request's `waitUntil`, after the final flush drained the processor. + * Ship a trace's buffered spans as a single-trace payload through the + * telemetry channel of the context that owns the trace (captured at root + * span start), falling back to the ambient context. Leaves the spans + * buffered when no channel is reachable (they retry on the next flush and + * at `finalizeTrace`). */ private shipTrace(traceId: string): void { const spans = this.traceBuffers.get(traceId); - this.traceBuffers.delete(traceId); if (!spans || spans.length === 0) { + this.traceBuffers.delete(traceId); return; } const context = getVercelRequestContextForTrace(traceId) ?? getVercelRequestContext(); if (!context?.telemetry) { - diag.warn( - `@vercel/otel: no telemetry context to finalize trace ${traceId}; retaining ${spans.length} span(s)`, + // No reachable channel right now (e.g. a bare-setTimeout timer flush + // for an untracked trace). Keep the spans buffered for a later flush + // instead of silently dropping them. + diag.debug( + `@vercel/otel: no telemetry context for trace ${traceId}; keeping ${spans.length} span(s) buffered`, ); - this.retain(spans); return; } + this.traceBuffers.delete(traceId); try { reportSpans(context.telemetry, spans); } catch (e) { - this.retain(spans); - diag.warn("@vercel/otel: failed to report finalized trace, retained:", e); + // Put them back so the next flush or the trace's finalize retries. + this.retainForTrace(traceId, spans); + diag.warn("@vercel/otel: failed to report spans, retained:", e); } } @@ -141,24 +134,29 @@ export class VercelRuntimeSpanExporter implements SpanExporter { this.traceBuffers.set(traceId, buffer); } buffer.push(span); - const overflow = buffer.length - MAX_TRACE_BUFFER_SIZE; - if (overflow > 0) { - buffer.splice(0, overflow); - this.droppedSpansCount += overflow; - diag.warn( - `@vercel/otel: trace span buffer full, dropped ${this.droppedSpansCount} span(s) total`, - ); + this.enforceCap(buffer); + } + + private retainForTrace(traceId: string, spans: ReadableSpan[]): void { + const buffer = this.traceBuffers.get(traceId); + if (buffer) { + // Newer spans may have been buffered while we were shipping; keep + // start-order by prepending the failed batch. + buffer.unshift(...spans); + this.enforceCap(buffer); + } else { + this.traceBuffers.set(traceId, spans); + this.enforceCap(spans); } } - private retain(spans: ReadableSpan[]): void { - this.pendingSpans.push(...spans); - const overflow = this.pendingSpans.length - MAX_TRACE_BUFFER_SIZE; + private enforceCap(buffer: ReadableSpan[]): void { + const overflow = buffer.length - MAX_TRACE_BUFFER_SIZE; if (overflow > 0) { - this.pendingSpans.splice(0, overflow); + buffer.splice(0, overflow); this.droppedSpansCount += overflow; diag.warn( - `@vercel/otel: retained span buffer full, dropped ${this.droppedSpansCount} span(s) total`, + `@vercel/otel: trace span buffer full, dropped ${this.droppedSpansCount} span(s) total`, ); } } @@ -188,4 +186,4 @@ function reportSpans( new TextDecoder().decode(serializedData), ) as IExportTraceServiceRequest; telemetry.reportSpans(data); -} \ No newline at end of file +}