fix(core): external trace context leaks across warm-started runs (#3768)

On warm-started workers with `processKeepAlive` enabled, every run's
attempt span was exported with the first run's `traceId` and
`parentSpanId`. `ExternalSpanExporterWrapper` and
`ExternalLogRecordExporterWrapper` captured `externalTraceContext` at
`TracingSDK` construction, and the SDK is memoized for the worker's
lifetime - so the per-run reset of
`StandardTraceContextManager.traceContext` never reached the wrappers.

Reported by a customer running v4.4.x: 33 distinct runs on the same
host/pid showed up in their APM as siblings of one parent span.

Fix: the wrappers now read external context live from the trace context
manager per export. Runs without an external trace context fall through
to the unchanged `externalTraceId` fallback - no behaviour change for
them.

Regression test in
`packages/core/test/externalSpanExporterWrapper.test.ts` asserts that
reassigning the manager between exports produces correctly-parented
spans.
This commit is contained in:
nicktrn
2026-05-28 11:43:39 +01:00
committed by GitHub
parent 816986d44e
commit c043c4a6ad
3 changed files with 132 additions and 49 deletions
@@ -0,0 +1,5 @@
---
"@trigger.dev/core": patch
---
Fix external trace context leaking across runs on warm-started workers with `processKeepAlive` enabled. Every subsequent run's attempt span was being exported with the first run's `traceId` and `parentSpanId`, breaking causal-chain navigation in external APM tools. Runs without an external trace context are unaffected.
+44 -49
View File
@@ -171,13 +171,12 @@ export class TracingSDK {
); );
const externalTraceId = idGenerator.generateTraceId(); const externalTraceId = idGenerator.generateTraceId();
const externalTraceContext = traceContext.getExternalTraceContext();
for (const exporter of config.exporters ?? []) { for (const exporter of config.exporters ?? []) {
spanProcessors.push( spanProcessors.push(
getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1" getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
? new BatchSpanProcessor( ? new BatchSpanProcessor(
new ExternalSpanExporterWrapper(exporter, externalTraceId, externalTraceContext), new ExternalSpanExporterWrapper(exporter, externalTraceId),
{ {
maxExportBatchSize: parseInt( maxExportBatchSize: parseInt(
getEnvVar("TRIGGER_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE") ?? "64" getEnvVar("TRIGGER_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE") ?? "64"
@@ -192,7 +191,7 @@ export class TracingSDK {
} }
) )
: new SimpleSpanProcessor( : new SimpleSpanProcessor(
new ExternalSpanExporterWrapper(exporter, externalTraceId, externalTraceContext) new ExternalSpanExporterWrapper(exporter, externalTraceId)
) )
); );
} }
@@ -245,11 +244,7 @@ export class TracingSDK {
logProcessors.push( logProcessors.push(
getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1" getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
? new BatchLogRecordProcessor( ? new BatchLogRecordProcessor(
new ExternalLogRecordExporterWrapper( new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId),
externalLogExporter,
externalTraceId,
externalTraceContext
),
{ {
maxExportBatchSize: parseInt( maxExportBatchSize: parseInt(
getEnvVar("TRIGGER_OTEL_LOG_MAX_EXPORT_BATCH_SIZE") ?? "64" getEnvVar("TRIGGER_OTEL_LOG_MAX_EXPORT_BATCH_SIZE") ?? "64"
@@ -264,11 +259,7 @@ export class TracingSDK {
} }
) )
: new SimpleLogRecordProcessor( : new SimpleLogRecordProcessor(
new ExternalLogRecordExporterWrapper( new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId)
externalLogExporter,
externalTraceId,
externalTraceContext
)
) )
); );
} }
@@ -417,23 +408,23 @@ function setLogLevel(level: TracingDiagnosticLogLevel) {
diag.setLogger(new DiagConsoleLogger(), diagLogLevel); diag.setLogger(new DiagConsoleLogger(), diagLogLevel);
} }
class ExternalSpanExporterWrapper { export class ExternalSpanExporterWrapper {
private readonly _isExternallySampled: boolean;
constructor( constructor(
private underlyingExporter: SpanExporter, private underlyingExporter: SpanExporter,
private externalTraceId: string, private externalTraceId: string
private externalTraceContext: ) {}
| { traceId: string; spanId: string; traceFlags: number; tracestate?: string }
| undefined
) {
this._isExternallySampled = externalTraceContext
? isTraceFlagSampled(externalTraceContext.traceFlags)
: !!externalTraceId;
}
private transformSpan(span: ReadableSpan): ReadableSpan | undefined { private transformSpan(span: ReadableSpan): ReadableSpan | undefined {
if (!this._isExternallySampled) { // Read external context live, so per-run reassignment of
// standardTraceContextManager.traceContext is honoured on warm-started
// workers that reuse a single TracingSDK across runs.
const externalTraceContext = traceContext.getExternalTraceContext();
const isExternallySampled = externalTraceContext
? isTraceFlagSampled(externalTraceContext.traceFlags)
: !!this.externalTraceId;
if (!isExternallySampled) {
return; return;
} }
@@ -441,8 +432,8 @@ class ExternalSpanExporterWrapper {
return; return;
} }
const externalTraceId = this.externalTraceContext const externalTraceId = externalTraceContext
? this.externalTraceContext.traceId ? externalTraceContext.traceId
: this.externalTraceId; : this.externalTraceId;
const isAttemptSpan = span.attributes[SemanticInternalAttributes.SPAN_ATTEMPT]; const isAttemptSpan = span.attributes[SemanticInternalAttributes.SPAN_ATTEMPT];
@@ -457,15 +448,15 @@ class ExternalSpanExporterWrapper {
}; };
} }
if (isAttemptSpan && this.externalTraceContext) { if (isAttemptSpan && externalTraceContext) {
parentSpanContext = { parentSpanContext = {
...parentSpanContext, ...parentSpanContext,
traceId: externalTraceId, traceId: externalTraceId,
spanId: this.externalTraceContext.spanId, spanId: externalTraceContext.spanId,
traceState: this.externalTraceContext.tracestate traceState: externalTraceContext.tracestate
? new TraceState(this.externalTraceContext.tracestate) ? new TraceState(externalTraceContext.tracestate)
: undefined, : undefined,
traceFlags: this.externalTraceContext.traceFlags, traceFlags: externalTraceContext.traceFlags,
}; };
} else if (isAttemptSpan) { } else if (isAttemptSpan) {
parentSpanContext = undefined; parentSpanContext = undefined;
@@ -502,28 +493,27 @@ class ExternalSpanExporterWrapper {
} }
class ExternalLogRecordExporterWrapper { class ExternalLogRecordExporterWrapper {
private readonly _isExternallySampled: boolean;
constructor( constructor(
private underlyingExporter: LogRecordExporter, private underlyingExporter: LogRecordExporter,
private externalTraceId: string, private externalTraceId: string
private externalTraceContext: ) {}
| { traceId: string; spanId: string; tracestate?: string; traceFlags: number }
| undefined
) {
this._isExternallySampled = externalTraceContext
? isTraceFlagSampled(externalTraceContext.traceFlags)
: !!externalTraceId;
}
export(logs: any[], resultCallback: (result: any) => void): void { export(logs: any[], resultCallback: (result: any) => void): void {
if (!this._isExternallySampled) { const externalTraceContext = traceContext.getExternalTraceContext();
const isExternallySampled = externalTraceContext
? isTraceFlagSampled(externalTraceContext.traceFlags)
: !!this.externalTraceId;
if (!isExternallySampled) {
this.underlyingExporter.export([], resultCallback); this.underlyingExporter.export([], resultCallback);
return; return;
} }
const modifiedLogs = logs.map(this.transformLogRecord.bind(this)); const modifiedLogs = logs.map((log) =>
this.transformLogRecord(log, externalTraceContext)
);
this.underlyingExporter.export(modifiedLogs, resultCallback); this.underlyingExporter.export(modifiedLogs, resultCallback);
} }
@@ -532,11 +522,16 @@ class ExternalLogRecordExporterWrapper {
return this.underlyingExporter.shutdown(); return this.underlyingExporter.shutdown();
} }
transformLogRecord(logRecord: ReadableLogRecord): ReadableLogRecord { transformLogRecord(
logRecord: ReadableLogRecord,
externalTraceContext:
| { traceId: string; spanId: string; tracestate?: string; traceFlags: number }
| undefined
): ReadableLogRecord {
// Capture externalTraceId for use within the proxy's scope. // Capture externalTraceId for use within the proxy's scope.
// Use externalTraceContext.traceId if available, otherwise fall back to generated externalTraceId // Use externalTraceContext.traceId if available, otherwise fall back to generated externalTraceId
const externalTraceId = this.externalTraceContext const externalTraceId = externalTraceContext
? this.externalTraceContext.traceId ? externalTraceContext.traceId
: this.externalTraceId; : this.externalTraceId;
// If there's no spanContext, or if the externalTraceId is not set, return the original logRecord. // If there's no spanContext, or if the externalTraceId is not set, return the original logRecord.
@@ -0,0 +1,83 @@
import { SpanKind, SpanStatusCode, TraceFlags } from "@opentelemetry/api";
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 { 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";
function createAttemptSpan(): ReadableSpan {
const spanCtx = {
traceId: "cccccccccccccccccccccccccccccccc",
spanId: "3333333333333333",
traceFlags: TraceFlags.SAMPLED,
};
return {
name: "Attempt 1",
kind: SpanKind.CONSUMER,
spanContext: () => spanCtx,
parentSpanContext: undefined,
startTime: [0, 0],
endTime: [0, 0],
status: { code: SpanStatusCode.UNSET },
attributes: { [SemanticInternalAttributes.SPAN_ATTEMPT]: true },
links: [],
events: [],
duration: [0, 0],
ended: true,
resource: {} as any,
instrumentationLibrary: { name: "test" } as any,
droppedAttributesCount: 0,
droppedEventsCount: 0,
droppedLinksCount: 0,
} as unknown as ReadableSpan;
}
function makeCapturingExporter(): { exporter: SpanExporter; captured: ReadableSpan[][] } {
const captured: ReadableSpan[][] = [];
const exporter: SpanExporter = {
export: (spans, cb) => {
captured.push(spans);
cb({ code: 0 } as any);
},
shutdown: () => Promise.resolve(),
forceFlush: () => Promise.resolve(),
};
return { exporter, captured };
}
describe("ExternalSpanExporterWrapper warm-start regression", () => {
let manager: StandardTraceContextManager;
beforeEach(() => {
manager = new StandardTraceContextManager();
traceContext.setGlobalManager(manager);
});
it("rewrites attempt spans using the manager's current external context, not the value captured at construction", () => {
const { exporter, captured } = makeCapturingExporter();
manager.traceContext = { external: { traceparent: TRACEPARENT_RUN_A } };
const wrapper = new ExternalSpanExporterWrapper(
exporter,
"ffffffffffffffffffffffffffffffff"
);
manager.traceContext = { external: { traceparent: TRACEPARENT_RUN_B } };
wrapper.export([createAttemptSpan()], () => {});
expect(captured).toHaveLength(1);
expect(captured[0]).toHaveLength(1);
const span = captured[0]![0]!;
expect(span.parentSpanContext?.spanId).toBe("2222222222222222");
expect(span.parentSpanContext?.spanId).not.toBe("1111111111111111");
expect(span.parentSpanContext?.traceId).toBe("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
expect(span.spanContext().traceId).toBe("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
});
});