Skip to content

Commit 6efeb47

Browse files
NERLOEclaude
andcommitted
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 c043c4a 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) <noreply@anthropic.com>
1 parent 4c21af8 commit 6efeb47

3 files changed

Lines changed: 297 additions & 20 deletions

File tree

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"@trigger.dev/core": patch
3+
---
4+
5+
Unrelated runs are no longer merged into a single trace in your external observability tool when they happen to execute on the same warm worker process. A run and the runs it triggers still share one trace, so a run tree stays together.

packages/core/src/v3/otel/tracingSDK.ts

Lines changed: 94 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -163,12 +163,13 @@ export class TracingSDK {
163163
)
164164
);
165165

166-
const externalTraceId = idGenerator.generateTraceId();
166+
// Shared by every wrapper below so a run's spans and logs agree on the id.
167+
const fallbackTraceIds = new FallbackExternalTraceIds(idGenerator.generateTraceId());
167168

168169
for (const exporter of config.exporters ?? []) {
169170
spanProcessors.push(
170171
getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
171-
? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId), {
172+
? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, fallbackTraceIds), {
172173
maxExportBatchSize: parseInt(
173174
getEnvVar("TRIGGER_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE") ?? "64"
174175
),
@@ -180,7 +181,7 @@ export class TracingSDK {
180181
),
181182
maxQueueSize: parseInt(getEnvVar("TRIGGER_OTEL_SPAN_MAX_QUEUE_SIZE") ?? "512"),
182183
})
183-
: new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId))
184+
: new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, fallbackTraceIds))
184185
);
185186
}
186187

@@ -232,7 +233,7 @@ export class TracingSDK {
232233
logProcessors.push(
233234
getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
234235
? new BatchLogRecordProcessor(
235-
new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId),
236+
new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceIds),
236237
{
237238
maxExportBatchSize: parseInt(
238239
getEnvVar("TRIGGER_OTEL_LOG_MAX_EXPORT_BATCH_SIZE") ?? "64"
@@ -247,7 +248,7 @@ export class TracingSDK {
247248
}
248249
)
249250
: new SimpleLogRecordProcessor(
250-
new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId)
251+
new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceIds)
251252
)
252253
);
253254
}
@@ -424,10 +425,81 @@ function setLogLevel(level: TracingDiagnosticLogLevel) {
424425
diag.setLogger(new DiagConsoleLogger(), diagLogLevel);
425426
}
426427

428+
/** Only the current run and the tail of recently ended ones can still export. */
429+
export const MAX_TRACKED_INTERNAL_TRACES = 64;
430+
431+
/**
432+
* External trace ids for runs that carry no external trace context — with
433+
* `processKeepAlive` the `TracingSDK` outlives the run, so an id captured at
434+
* construction merges every run on the process into one trace.
435+
*
436+
* A record's id comes from its own internal trace id rather than from whatever
437+
* run is current when the exporter is called. Batch processors drain
438+
* asynchronously, so a run's records are routinely exported after the next run
439+
* has started, and reading ambient state then would stamp them with the wrong
440+
* run's id. It also makes a run's spans and logs agree without coordinating.
441+
*
442+
* Granularity therefore follows the internal trace, not the run: a run tree
443+
* shares one internal trace, so a parent and the runs it triggers land on one
444+
* external trace together, which is the grouping you want.
445+
*/
446+
export class FallbackExternalTraceIds {
447+
private readonly byInternalTrace = new Map<string, string>();
448+
449+
constructor(
450+
private seed: string,
451+
private traceIdGenerator: Pick<RandomIdGenerator, "generateTraceId"> = idGenerator
452+
) {}
453+
454+
/** False when no external trace id was configured, i.e. external export is off. */
455+
get enabled(): boolean {
456+
return !!this.seed;
457+
}
458+
459+
forInternalTrace(internalTraceId: string): string {
460+
// An empty seed means external export is disabled — leave it that way
461+
// rather than minting an id and switching the feature on.
462+
if (!this.seed) {
463+
return this.seed;
464+
}
465+
466+
const known = this.byInternalTrace.get(internalTraceId);
467+
468+
if (known) {
469+
// Re-insert so the map is ordered by last use rather than first. A run
470+
// that is still exporting keeps its id even if enough unrelated traces
471+
// appear alongside it to fill the map, which would otherwise split it
472+
// across two external traces.
473+
this.byInternalTrace.delete(internalTraceId);
474+
this.byInternalTrace.set(internalTraceId, known);
475+
476+
return known;
477+
}
478+
479+
// The first run reuses the id generated at construction, so the configured
480+
// seed is not thrown away.
481+
const traceId =
482+
this.byInternalTrace.size === 0 ? this.seed : this.traceIdGenerator.generateTraceId();
483+
484+
this.byInternalTrace.set(internalTraceId, traceId);
485+
486+
if (this.byInternalTrace.size > MAX_TRACKED_INTERNAL_TRACES) {
487+
// Map iterates in insertion order, so this drops the least recently used.
488+
const stalest = this.byInternalTrace.keys().next().value;
489+
490+
if (stalest !== undefined) {
491+
this.byInternalTrace.delete(stalest);
492+
}
493+
}
494+
495+
return traceId;
496+
}
497+
}
498+
427499
export class ExternalSpanExporterWrapper {
428500
constructor(
429501
private underlyingExporter: SpanExporter,
430-
private externalTraceId: string
502+
private fallback: FallbackExternalTraceIds
431503
) {}
432504

433505
private transformSpan(span: ReadableSpan): ReadableSpan | undefined {
@@ -438,7 +510,7 @@ export class ExternalSpanExporterWrapper {
438510

439511
const isExternallySampled = externalTraceContext
440512
? isTraceFlagSampled(externalTraceContext.traceFlags)
441-
: !!this.externalTraceId;
513+
: this.fallback.enabled;
442514

443515
if (!isExternallySampled) {
444516
return;
@@ -450,7 +522,7 @@ export class ExternalSpanExporterWrapper {
450522

451523
const externalTraceId = externalTraceContext
452524
? externalTraceContext.traceId
453-
: this.externalTraceId;
525+
: this.fallback.forInternalTrace(span.spanContext().traceId);
454526

455527
const isAttemptSpan = span.attributes[SemanticInternalAttributes.SPAN_ATTEMPT];
456528

@@ -508,18 +580,18 @@ export class ExternalSpanExporterWrapper {
508580
}
509581
}
510582

511-
class ExternalLogRecordExporterWrapper {
583+
export class ExternalLogRecordExporterWrapper {
512584
constructor(
513585
private underlyingExporter: LogRecordExporter,
514-
private externalTraceId: string
586+
private fallback: FallbackExternalTraceIds
515587
) {}
516588

517589
export(logs: any[], resultCallback: (result: any) => void): void {
518590
const externalTraceContext = traceContext.getExternalTraceContext();
519591

520592
const isExternallySampled = externalTraceContext
521593
? isTraceFlagSampled(externalTraceContext.traceFlags)
522-
: !!this.externalTraceId;
594+
: this.fallback.enabled;
523595

524596
if (!isExternallySampled) {
525597
this.underlyingExporter.export([], resultCallback);
@@ -550,14 +622,20 @@ class ExternalLogRecordExporterWrapper {
550622
| { traceId: string; spanId: string; tracestate?: string; traceFlags: number }
551623
| undefined
552624
): ReadableLogRecord {
553-
// Capture externalTraceId for use within the proxy's scope.
554-
// Use externalTraceContext.traceId if available, otherwise fall back to generated externalTraceId
625+
// Without a spanContext there is no internal trace id to key the fallback
626+
// on, and nothing to rewrite.
627+
if (!logRecord.spanContext) {
628+
return logRecord;
629+
}
630+
631+
// Capture externalTraceId for use within the proxy's scope. Use
632+
// externalTraceContext.traceId if available, otherwise the id belonging to
633+
// the run this record came from.
555634
const externalTraceId = externalTraceContext
556635
? externalTraceContext.traceId
557-
: this.externalTraceId;
636+
: this.fallback.forInternalTrace(logRecord.spanContext.traceId);
558637

559-
// If there's no spanContext, or if the externalTraceId is not set, return the original logRecord.
560-
if (!logRecord.spanContext || !externalTraceId) {
638+
if (!externalTraceId) {
561639
return logRecord;
562640
}
563641

0 commit comments

Comments
 (0)