fix(otel): strict per-trace span payloads with opt-in mid-run streaming

.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.

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;

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<symbol, unknown>;
+
+function installAmbient(ctx: VercelRequestContext | undefined): void {
+  (globalThis as GlobalWithReader)[VRC_SYMBOL] = { get: () => ctx };
+}
+
+function fakeVrc(): VercelRequestContext & {
+  flushes: Array<() => Promise<unknown>>;
+} {
+  const flushes: Array<() => Promise<unknown>> = [];
+  return {
+    flushes,
+    waitUntil: (promiseOrFunc) => {
+      if (typeof promiseOrFunc === "function") {
+        flushes.push(promiseOrFunc as () => Promise<unknown>);
+      }
+    },
+    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<typeof vi.fn>;
+} {
+  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);
+  });
+});

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<string, () => 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,15 @@ 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
+        // 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.
         vrc.waitUntil(async () => {
           if (this.rootSpanIds.has(traceId)) {
             // Not root has not completed yet, so no point in flushing.
@@ -81,7 +100,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 +162,33 @@ export class CompositeSpanProcessor implements SpanProcessor {
         this.waitSpanEnd.delete(traceId);
         pending();
       }
+    } else if (sampled && getVercelRequestContext()) {
+      // 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();
+    }
+  }
+
+  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);
+      });
     }
   }
 }

packages/otel/src/vercel-request-context/api.ts

@@ -35,3 +35,87 @@ 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 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, 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`); 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<string, VercelRequestContext>();
+
+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<TraceFinalizer>();
+
+/** @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 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 {
+    for (const finalizer of traceFinalizers) {
+      finalizer(traceId);
+    }
+  } finally {
+    traceContextRegistry.delete(traceId);
+  }
+}