From e840eb0e3c6df004e13accd30adebc0c09f05088 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcus=20Nerl=C3=B8e?= Date: Fri, 7 Aug 2026 16:51:01 +0200 Subject: [PATCH] fix(core): give each run its own external fallback trace id Runs that carry no external trace context (schedules, task-to-task triggers) fall back to a trace id generated once in the TracingSDK constructor. With `experimental_processKeepAlive` the TracingSDK outlives the run, so every run on a warm process was exported to the external OTLP endpoint under that one id, merging unrelated runs into a single trace. Across our production traces, 80.3% contained spans from more than one run, worst case 25. This is the same warm-start hazard c043c4a6a fixed for the external context path, which read the context live but deliberately left the fallback captured at construction. Key the fallback off the internal trace id that every span and log record of a run already carries, rather than off ambient state. Batch processors drain asynchronously, so a run's records are routinely exported after the next run has started; deciding the id at export time from whatever run is current would stamp the earlier run's records with the later run's id. Letting the record decide sidesteps the timing entirely, and makes a run's spans and logs agree without coordinating. The map is bounded, since a warm process serves unboundedly many runs and only the in-flight ones can still have records to export. An empty configured id still means external export is off. Co-Authored-By: Claude Opus 5 (1M context) --- .changeset/external-trace-id-per-run.md | 5 + packages/core/src/v3/otel/tracingSDK.ts | 106 ++++++++-- .../test/externalSpanExporterWrapper.test.ts | 190 +++++++++++++++++- 3 files changed, 281 insertions(+), 20 deletions(-) create mode 100644 .changeset/external-trace-id-per-run.md diff --git a/.changeset/external-trace-id-per-run.md b/.changeset/external-trace-id-per-run.md new file mode 100644 index 00000000000..65d0ca98b8e --- /dev/null +++ b/.changeset/external-trace-id-per-run.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/core": patch +--- + +Runs that don't continue an incoming trace are no longer merged into one trace when they execute on the same warm worker process. Each run now appears as its own trace in your external observability tool, so per-run cost and latency attribution works again. diff --git a/packages/core/src/v3/otel/tracingSDK.ts b/packages/core/src/v3/otel/tracingSDK.ts index 0b3a66a87b4..2fb5785f87d 100644 --- a/packages/core/src/v3/otel/tracingSDK.ts +++ b/packages/core/src/v3/otel/tracingSDK.ts @@ -162,12 +162,13 @@ export class TracingSDK { ) ); - const externalTraceId = idGenerator.generateTraceId(); + // Shared by every wrapper below so a run's spans and logs agree on the id. + const fallbackTraceId = new FallbackExternalTraceId(idGenerator.generateTraceId()); for (const exporter of config.exporters ?? []) { spanProcessors.push( getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1" - ? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId), { + ? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, fallbackTraceId), { maxExportBatchSize: parseInt( getEnvVar("TRIGGER_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE") ?? "64" ), @@ -179,7 +180,7 @@ export class TracingSDK { ), maxQueueSize: parseInt(getEnvVar("TRIGGER_OTEL_SPAN_MAX_QUEUE_SIZE") ?? "512"), }) - : new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId)) + : new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, fallbackTraceId)) ); } @@ -231,7 +232,7 @@ export class TracingSDK { logProcessors.push( getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1" ? new BatchLogRecordProcessor( - new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId), + new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceId), { maxExportBatchSize: parseInt( getEnvVar("TRIGGER_OTEL_LOG_MAX_EXPORT_BATCH_SIZE") ?? "64" @@ -246,7 +247,7 @@ export class TracingSDK { } ) : new SimpleLogRecordProcessor( - new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId) + new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceId) ) ); } @@ -393,10 +394,77 @@ function setLogLevel(level: TracingDiagnosticLogLevel) { diag.setLogger(new DiagConsoleLogger(), diagLogLevel); } +/** + * A warm process serves runs one at a time, so this is only ever asked about + * the current run and the tail of recently ended ones still draining. + */ +const MAX_TRACKED_INTERNAL_TRACES = 64; + +/** + * External trace ids for runs that carry no external trace context, one per + * run. + * + * There has to be one per run because with `processKeepAlive` the `TracingSDK` + * — and so the wrappers — outlive the run, so an id captured at construction + * merges every run on the process into a single trace. + * + * Which id a record gets is decided by the record's own internal trace id + * rather than by whatever run is current when the exporter is called. Batch + * processors drain asynchronously, so a run's spans and logs are routinely + * exported after the next run has already started; reading ambient state at + * that point would stamp them with the wrong run's id. Keying off the record + * also means spans and logs agree without having to coordinate. + */ +export class FallbackExternalTraceId { + private readonly byInternalTrace = new Map(); + + constructor( + private seed: string, + private traceIdGenerator: Pick = idGenerator + ) {} + + /** False when no external trace id was configured, i.e. external export is off. */ + get enabled(): boolean { + return !!this.seed; + } + + forInternalTrace(internalTraceId: string): string { + // An empty seed means external export is disabled — leave it that way + // rather than minting an id and switching the feature on. + if (!this.seed) { + return this.seed; + } + + const known = this.byInternalTrace.get(internalTraceId); + + if (known) { + return known; + } + + // The first run reuses the id generated at construction, so the configured + // seed is not thrown away. + const traceId = + this.byInternalTrace.size === 0 ? this.seed : this.traceIdGenerator.generateTraceId(); + + this.byInternalTrace.set(internalTraceId, traceId); + + if (this.byInternalTrace.size > MAX_TRACKED_INTERNAL_TRACES) { + // Map iterates in insertion order, so this drops the oldest run. + const oldest = this.byInternalTrace.keys().next().value; + + if (oldest !== undefined) { + this.byInternalTrace.delete(oldest); + } + } + + return traceId; + } +} + export class ExternalSpanExporterWrapper { constructor( private underlyingExporter: SpanExporter, - private externalTraceId: string + private fallback: FallbackExternalTraceId ) {} private transformSpan(span: ReadableSpan): ReadableSpan | undefined { @@ -407,7 +475,7 @@ export class ExternalSpanExporterWrapper { const isExternallySampled = externalTraceContext ? isTraceFlagSampled(externalTraceContext.traceFlags) - : !!this.externalTraceId; + : this.fallback.enabled; if (!isExternallySampled) { return; @@ -419,7 +487,7 @@ export class ExternalSpanExporterWrapper { const externalTraceId = externalTraceContext ? externalTraceContext.traceId - : this.externalTraceId; + : this.fallback.forInternalTrace(span.spanContext().traceId); const isAttemptSpan = span.attributes[SemanticInternalAttributes.SPAN_ATTEMPT]; @@ -477,10 +545,10 @@ export class ExternalSpanExporterWrapper { } } -class ExternalLogRecordExporterWrapper { +export class ExternalLogRecordExporterWrapper { constructor( private underlyingExporter: LogRecordExporter, - private externalTraceId: string + private fallback: FallbackExternalTraceId ) {} export(logs: any[], resultCallback: (result: any) => void): void { @@ -488,7 +556,7 @@ class ExternalLogRecordExporterWrapper { const isExternallySampled = externalTraceContext ? isTraceFlagSampled(externalTraceContext.traceFlags) - : !!this.externalTraceId; + : this.fallback.enabled; if (!isExternallySampled) { this.underlyingExporter.export([], resultCallback); @@ -519,14 +587,20 @@ class ExternalLogRecordExporterWrapper { | { traceId: string; spanId: string; tracestate?: string; traceFlags: number } | undefined ): ReadableLogRecord { - // Capture externalTraceId for use within the proxy's scope. - // Use externalTraceContext.traceId if available, otherwise fall back to generated externalTraceId + // Without a spanContext there is no internal trace id to key the fallback + // on, and nothing to rewrite. + if (!logRecord.spanContext) { + return logRecord; + } + + // Capture externalTraceId for use within the proxy's scope. Use + // externalTraceContext.traceId if available, otherwise the id belonging to + // the run this record came from. const externalTraceId = externalTraceContext ? externalTraceContext.traceId - : this.externalTraceId; + : this.fallback.forInternalTrace(logRecord.spanContext.traceId); - // If there's no spanContext, or if the externalTraceId is not set, return the original logRecord. - if (!logRecord.spanContext || !externalTraceId) { + if (!externalTraceId) { return logRecord; } diff --git a/packages/core/test/externalSpanExporterWrapper.test.ts b/packages/core/test/externalSpanExporterWrapper.test.ts index 9b51653a1ec..10ae3d14894 100644 --- a/packages/core/test/externalSpanExporterWrapper.test.ts +++ b/packages/core/test/externalSpanExporterWrapper.test.ts @@ -1,17 +1,26 @@ import { SpanKind, SpanStatusCode, TraceFlags } from "@opentelemetry/api"; +import type { LogRecordExporter, ReadableLogRecord } from "@opentelemetry/sdk-logs"; import type { ReadableSpan, SpanExporter } from "@opentelemetry/sdk-trace-node"; import { beforeEach, describe, expect, it } from "vitest"; -import { ExternalSpanExporterWrapper } from "../src/v3/otel/tracingSDK.js"; +import { + ExternalLogRecordExporterWrapper, + ExternalSpanExporterWrapper, + FallbackExternalTraceId, +} from "../src/v3/otel/tracingSDK.js"; import { SemanticInternalAttributes } from "../src/v3/semanticInternalAttributes.js"; import { traceContext } from "../src/v3/trace-context-api.js"; import { StandardTraceContextManager } from "../src/v3/traceContext/manager.js"; const TRACEPARENT_RUN_A = "00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-1111111111111111-01"; const TRACEPARENT_RUN_B = "00-bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-2222222222222222-01"; +const SEED = "ffffffffffffffffffffffffffffffff"; +// Every span and log record of one run shares the run's internal trace id. +const INTERNAL_TRACE_RUN_A = "cccccccccccccccccccccccccccccccc"; +const INTERNAL_TRACE_RUN_B = "dddddddddddddddddddddddddddddddd"; -function createAttemptSpan(): ReadableSpan { +function createAttemptSpan(internalTraceId = INTERNAL_TRACE_RUN_A): ReadableSpan { const spanCtx = { - traceId: "cccccccccccccccccccccccccccccccc", + traceId: internalTraceId, spanId: "3333333333333333", traceFlags: TraceFlags.SAMPLED, }; @@ -36,6 +45,18 @@ function createAttemptSpan(): ReadableSpan { } as unknown as ReadableSpan; } +function createLogRecord(internalTraceId = INTERNAL_TRACE_RUN_A): ReadableLogRecord { + return { + body: "hello", + attributes: {}, + spanContext: { + traceId: internalTraceId, + spanId: "3333333333333333", + traceFlags: TraceFlags.SAMPLED, + }, + } as unknown as ReadableLogRecord; +} + function makeCapturingExporter(): { exporter: SpanExporter; captured: ReadableSpan[][] } { const captured: ReadableSpan[][] = []; const exporter: SpanExporter = { @@ -49,10 +70,40 @@ function makeCapturingExporter(): { exporter: SpanExporter; captured: ReadableSp return { exporter, captured }; } +function makeCapturingLogExporter(): { + exporter: LogRecordExporter; + captured: ReadableLogRecord[][]; +} { + const captured: ReadableLogRecord[][] = []; + const exporter: LogRecordExporter = { + export: (records, cb) => { + captured.push(records); + cb({ code: 0 } as any); + }, + shutdown: () => Promise.resolve(), + }; + return { exporter, captured }; +} + +/** Yields 000…001, 000…002, … so a reminted id is identifiable by its ordinal. */ +function makeIdGenerator() { + let generated = 0; + return { + generateTraceId: () => `${++generated}`.padStart(32, "0"), + get count() { + return generated; + }, + }; +} + describe("ExternalSpanExporterWrapper warm-start regression", () => { let manager: StandardTraceContextManager; beforeEach(() => { + // `setGlobalManager` delegates to `registerGlobal`, which ignores a second + // registration — without disabling first, every test after the first would + // keep mutating the first test's manager. + traceContext.disable(); manager = new StandardTraceContextManager(); traceContext.setGlobalManager(manager); }); @@ -62,7 +113,7 @@ describe("ExternalSpanExporterWrapper warm-start regression", () => { manager.traceContext = { external: { traceparent: TRACEPARENT_RUN_A } }; - const wrapper = new ExternalSpanExporterWrapper(exporter, "ffffffffffffffffffffffffffffffff"); + const wrapper = new ExternalSpanExporterWrapper(exporter, new FallbackExternalTraceId(SEED)); manager.traceContext = { external: { traceparent: TRACEPARENT_RUN_B } }; @@ -77,4 +128,135 @@ describe("ExternalSpanExporterWrapper warm-start regression", () => { expect(span.parentSpanContext?.traceId).toBe("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"); expect(span.spanContext().traceId).toBe("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"); }); + + // Runs triggered internally — a schedule, or one task triggering another — + // carry no external trace context and so take the generated fallback. That id + // was captured at construction, which on a warm-started worker meant every run + // on the process shared a single trace id. + it("gives each run its own fallback trace id when there is no external context", () => { + const { exporter, captured } = makeCapturingExporter(); + const idGenerator = makeIdGenerator(); + + const wrapper = new ExternalSpanExporterWrapper( + exporter, + new FallbackExternalTraceId(SEED, idGenerator) + ); + + wrapper.export([createAttemptSpan(INTERNAL_TRACE_RUN_A)], () => {}); + // A second run on the same warm process. + wrapper.export([createAttemptSpan(INTERNAL_TRACE_RUN_B)], () => {}); + + const runATraceId = captured[0]![0]!.spanContext().traceId; + const runBTraceId = captured[1]![0]!.spanContext().traceId; + + expect(runATraceId).toBe(SEED); + expect(runBTraceId).not.toBe(runATraceId); + expect(runBTraceId).toBe("00000000000000000000000000000001"); + }); + + it("keeps one fallback trace id across every export within a run", () => { + const { exporter, captured } = makeCapturingExporter(); + const idGenerator = makeIdGenerator(); + + const wrapper = new ExternalSpanExporterWrapper( + exporter, + new FallbackExternalTraceId(SEED, idGenerator) + ); + + wrapper.export([createAttemptSpan()], () => {}); + wrapper.export([createAttemptSpan()], () => {}); + + expect(captured[1]![0]!.spanContext().traceId).toBe(captured[0]![0]!.spanContext().traceId); + expect(idGenerator.count).toBe(0); + }); + + // Batch processors drain asynchronously, so a run's records are routinely + // exported after the next run has already started. Deciding the id from + // ambient state at that moment would stamp the earlier run's records with the + // later run's id, merging exactly the traces this is meant to separate. + it("stamps records with their own run's id even when exported after the next run started", () => { + const spans = makeCapturingExporter(); + const logs = makeCapturingLogExporter(); + const idGenerator = makeIdGenerator(); + + const fallback = new FallbackExternalTraceId(SEED, idGenerator); + const spanWrapper = new ExternalSpanExporterWrapper(spans.exporter, fallback); + const logWrapper = new ExternalLogRecordExporterWrapper(logs.exporter, fallback); + + // Run B is underway and has already exported. + spanWrapper.export([createAttemptSpan(INTERNAL_TRACE_RUN_B)], () => {}); + manager.traceContext = { traceparent: TRACEPARENT_RUN_B }; + + // Run A's queued records only drain now. + spanWrapper.export([createAttemptSpan(INTERNAL_TRACE_RUN_A)], () => {}); + logWrapper.export([createLogRecord(INTERNAL_TRACE_RUN_A)], () => {}); + + const runBTraceId = spans.captured[0]![0]!.spanContext().traceId; + const lateRunASpanId = spans.captured[1]![0]!.spanContext().traceId; + const lateRunALogId = logs.captured[0]![0]!.spanContext!.traceId; + + expect(lateRunASpanId).not.toBe(runBTraceId); + expect(lateRunALogId).toBe(lateRunASpanId); + }); + + // The TracingSDK shares one FallbackExternalTraceId across its span and log + // wrappers, so a run's spans and logs land on the same external trace. + it("keeps a run's spans and logs on the same id", () => { + const spans = makeCapturingExporter(); + const logs = makeCapturingLogExporter(); + const idGenerator = makeIdGenerator(); + + const fallback = new FallbackExternalTraceId(SEED, idGenerator); + const spanWrapper = new ExternalSpanExporterWrapper(spans.exporter, fallback); + const logWrapper = new ExternalLogRecordExporterWrapper(logs.exporter, fallback); + + spanWrapper.export([createAttemptSpan(INTERNAL_TRACE_RUN_B)], () => {}); + logWrapper.export([createLogRecord(INTERNAL_TRACE_RUN_B)], () => {}); + + expect(logs.captured[0]![0]!.spanContext!.traceId).toBe( + spans.captured[0]![0]!.spanContext().traceId + ); + }); + + it("leaves external export off when no external trace id was configured", () => { + const { exporter, captured } = makeCapturingExporter(); + const idGenerator = makeIdGenerator(); + + const wrapper = new ExternalSpanExporterWrapper( + exporter, + new FallbackExternalTraceId("", idGenerator) + ); + + wrapper.export([createAttemptSpan()], () => {}); + + // Minting an id here would switch external export on for a deployment that + // never asked for it. + expect(captured[0]).toHaveLength(0); + }); + + // A warm process is long-lived, so the map that remembers each run's id has + // to be bounded rather than growing for the life of the worker. + it("bounds how many runs it remembers", () => { + const { exporter, captured } = makeCapturingExporter(); + const idGenerator = makeIdGenerator(); + + const wrapper = new ExternalSpanExporterWrapper( + exporter, + new FallbackExternalTraceId(SEED, idGenerator) + ); + + const firstRun = "aa000000000000000000000000000000"; + wrapper.export([createAttemptSpan(firstRun)], () => {}); + + for (let i = 0; i < 64; i++) { + wrapper.export([createAttemptSpan(`bb${`${i}`.padStart(30, "0")}`)], () => {}); + } + + // Evicted, so it is treated as a run never seen before. + wrapper.export([createAttemptSpan(firstRun)], () => {}); + + expect(captured.at(-1)![0]!.spanContext().traceId).not.toBe( + captured[0]![0]!.spanContext().traceId + ); + }); });