Skip to content

Commit ef1e9a3

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 7b390e5 commit ef1e9a3

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
@@ -162,12 +162,13 @@ export class TracingSDK {
162162
)
163163
);
164164

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

167168
for (const exporter of config.exporters ?? []) {
168169
spanProcessors.push(
169170
getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
170-
? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId), {
171+
? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, fallbackTraceIds), {
171172
maxExportBatchSize: parseInt(
172173
getEnvVar("TRIGGER_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE") ?? "64"
173174
),
@@ -179,7 +180,7 @@ export class TracingSDK {
179180
),
180181
maxQueueSize: parseInt(getEnvVar("TRIGGER_OTEL_SPAN_MAX_QUEUE_SIZE") ?? "512"),
181182
})
182-
: new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId))
183+
: new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, fallbackTraceIds))
183184
);
184185
}
185186

@@ -231,7 +232,7 @@ export class TracingSDK {
231232
logProcessors.push(
232233
getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
233234
? new BatchLogRecordProcessor(
234-
new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId),
235+
new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceIds),
235236
{
236237
maxExportBatchSize: parseInt(
237238
getEnvVar("TRIGGER_OTEL_LOG_MAX_EXPORT_BATCH_SIZE") ?? "64"
@@ -246,7 +247,7 @@ export class TracingSDK {
246247
}
247248
)
248249
: new SimpleLogRecordProcessor(
249-
new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId)
250+
new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceIds)
250251
)
251252
);
252253
}
@@ -393,10 +394,81 @@ function setLogLevel(level: TracingDiagnosticLogLevel) {
393394
diag.setLogger(new DiagConsoleLogger(), diagLogLevel);
394395
}
395396

397+
/** Only the current run and the tail of recently ended ones can still export. */
398+
export const MAX_TRACKED_INTERNAL_TRACES = 64;
399+
400+
/**
401+
* External trace ids for runs that carry no external trace context — with
402+
* `processKeepAlive` the `TracingSDK` outlives the run, so an id captured at
403+
* construction merges every run on the process into one trace.
404+
*
405+
* A record's id comes from its own internal trace id rather than from whatever
406+
* run is current when the exporter is called. Batch processors drain
407+
* asynchronously, so a run's records are routinely exported after the next run
408+
* has started, and reading ambient state then would stamp them with the wrong
409+
* run's id. It also makes a run's spans and logs agree without coordinating.
410+
*
411+
* Granularity therefore follows the internal trace, not the run: a run tree
412+
* shares one internal trace, so a parent and the runs it triggers land on one
413+
* external trace together, which is the grouping you want.
414+
*/
415+
export class FallbackExternalTraceIds {
416+
private readonly byInternalTrace = new Map<string, string>();
417+
418+
constructor(
419+
private seed: string,
420+
private traceIdGenerator: Pick<RandomIdGenerator, "generateTraceId"> = idGenerator
421+
) {}
422+
423+
/** False when no external trace id was configured, i.e. external export is off. */
424+
get enabled(): boolean {
425+
return !!this.seed;
426+
}
427+
428+
forInternalTrace(internalTraceId: string): string {
429+
// An empty seed means external export is disabled — leave it that way
430+
// rather than minting an id and switching the feature on.
431+
if (!this.seed) {
432+
return this.seed;
433+
}
434+
435+
const known = this.byInternalTrace.get(internalTraceId);
436+
437+
if (known) {
438+
// Re-insert so the map is ordered by last use rather than first. A run
439+
// that is still exporting keeps its id even if enough unrelated traces
440+
// appear alongside it to fill the map, which would otherwise split it
441+
// across two external traces.
442+
this.byInternalTrace.delete(internalTraceId);
443+
this.byInternalTrace.set(internalTraceId, known);
444+
445+
return known;
446+
}
447+
448+
// The first run reuses the id generated at construction, so the configured
449+
// seed is not thrown away.
450+
const traceId =
451+
this.byInternalTrace.size === 0 ? this.seed : this.traceIdGenerator.generateTraceId();
452+
453+
this.byInternalTrace.set(internalTraceId, traceId);
454+
455+
if (this.byInternalTrace.size > MAX_TRACKED_INTERNAL_TRACES) {
456+
// Map iterates in insertion order, so this drops the least recently used.
457+
const stalest = this.byInternalTrace.keys().next().value;
458+
459+
if (stalest !== undefined) {
460+
this.byInternalTrace.delete(stalest);
461+
}
462+
}
463+
464+
return traceId;
465+
}
466+
}
467+
396468
export class ExternalSpanExporterWrapper {
397469
constructor(
398470
private underlyingExporter: SpanExporter,
399-
private externalTraceId: string
471+
private fallback: FallbackExternalTraceIds
400472
) {}
401473

402474
private transformSpan(span: ReadableSpan): ReadableSpan | undefined {
@@ -407,7 +479,7 @@ export class ExternalSpanExporterWrapper {
407479

408480
const isExternallySampled = externalTraceContext
409481
? isTraceFlagSampled(externalTraceContext.traceFlags)
410-
: !!this.externalTraceId;
482+
: this.fallback.enabled;
411483

412484
if (!isExternallySampled) {
413485
return;
@@ -419,7 +491,7 @@ export class ExternalSpanExporterWrapper {
419491

420492
const externalTraceId = externalTraceContext
421493
? externalTraceContext.traceId
422-
: this.externalTraceId;
494+
: this.fallback.forInternalTrace(span.spanContext().traceId);
423495

424496
const isAttemptSpan = span.attributes[SemanticInternalAttributes.SPAN_ATTEMPT];
425497

@@ -477,18 +549,18 @@ export class ExternalSpanExporterWrapper {
477549
}
478550
}
479551

480-
class ExternalLogRecordExporterWrapper {
552+
export class ExternalLogRecordExporterWrapper {
481553
constructor(
482554
private underlyingExporter: LogRecordExporter,
483-
private externalTraceId: string
555+
private fallback: FallbackExternalTraceIds
484556
) {}
485557

486558
export(logs: any[], resultCallback: (result: any) => void): void {
487559
const externalTraceContext = traceContext.getExternalTraceContext();
488560

489561
const isExternallySampled = externalTraceContext
490562
? isTraceFlagSampled(externalTraceContext.traceFlags)
491-
: !!this.externalTraceId;
563+
: this.fallback.enabled;
492564

493565
if (!isExternallySampled) {
494566
this.underlyingExporter.export([], resultCallback);
@@ -519,14 +591,20 @@ class ExternalLogRecordExporterWrapper {
519591
| { traceId: string; spanId: string; tracestate?: string; traceFlags: number }
520592
| undefined
521593
): ReadableLogRecord {
522-
// Capture externalTraceId for use within the proxy's scope.
523-
// Use externalTraceContext.traceId if available, otherwise fall back to generated externalTraceId
594+
// Without a spanContext there is no internal trace id to key the fallback
595+
// on, and nothing to rewrite.
596+
if (!logRecord.spanContext) {
597+
return logRecord;
598+
}
599+
600+
// Capture externalTraceId for use within the proxy's scope. Use
601+
// externalTraceContext.traceId if available, otherwise the id belonging to
602+
// the run this record came from.
524603
const externalTraceId = externalTraceContext
525604
? externalTraceContext.traceId
526-
: this.externalTraceId;
605+
: this.fallback.forInternalTrace(logRecord.spanContext.traceId);
527606

528-
// If there's no spanContext, or if the externalTraceId is not set, return the original logRecord.
529-
if (!logRecord.spanContext || !externalTraceId) {
607+
if (!externalTraceId) {
530608
return logRecord;
531609
}
532610

0 commit comments

Comments
 (0)