fix(core): mint the fallback external trace id per run

.changeset/external-trace-id-per-run.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+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.

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

@@ -163,12 +163,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 fallbackTraceIds = new FallbackExternalTraceIds(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, fallbackTraceIds), {
               maxExportBatchSize: parseInt(
                 getEnvVar("TRIGGER_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE") ?? "64"
               ),
@@ -180,7 +181,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, fallbackTraceIds))
       );
     }
 
@@ -232,7 +233,7 @@ export class TracingSDK {
       logProcessors.push(
         getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
           ? new BatchLogRecordProcessor(
-              new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId),
+              new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceIds),
               {
                 maxExportBatchSize: parseInt(
                   getEnvVar("TRIGGER_OTEL_LOG_MAX_EXPORT_BATCH_SIZE") ?? "64"
@@ -247,7 +248,7 @@ export class TracingSDK {
               }
             )
           : new SimpleLogRecordProcessor(
-              new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId)
+              new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceIds)
             )
       );
     }
@@ -424,10 +425,81 @@ function setLogLevel(level: TracingDiagnosticLogLevel) {
   diag.setLogger(new DiagConsoleLogger(), diagLogLevel);
 }
 
+/** Only the current run and the tail of recently ended ones can still export. */
+export const MAX_TRACKED_INTERNAL_TRACES = 64;
+
+/**
+ * External trace ids for runs that carry no external trace context — with
+ * `processKeepAlive` the `TracingSDK` outlives the run, so an id captured at
+ * construction merges every run on the process into one trace.
+ *
+ * A record's id comes from its own internal trace id rather than from whatever
+ * run is current when the exporter is called. Batch processors drain
+ * asynchronously, so a run's records are routinely exported after the next run
+ * has started, and reading ambient state then would stamp them with the wrong
+ * run's id. It also makes a run's spans and logs agree without coordinating.
+ *
+ * Granularity therefore follows the internal trace, not the run: a run tree
+ * shares one internal trace, so a parent and the runs it triggers land on one
+ * external trace together, which is the grouping you want.
+ */
+export class FallbackExternalTraceIds {
+  private readonly byInternalTrace = new Map<string, string>();
+
+  constructor(
+    private seed: string,
+    private traceIdGenerator: Pick<RandomIdGenerator, "generateTraceId"> = 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) {
+      // Re-insert so the map is ordered by last use rather than first. A run
+      // that is still exporting keeps its id even if enough unrelated traces
+      // appear alongside it to fill the map, which would otherwise split it
+      // across two external traces.
+      this.byInternalTrace.delete(internalTraceId);
+      this.byInternalTrace.set(internalTraceId, 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 least recently used.
+      const stalest = this.byInternalTrace.keys().next().value;
+
+      if (stalest !== undefined) {
+        this.byInternalTrace.delete(stalest);
+      }
+    }
+
+    return traceId;
+  }
+}
+
 export class ExternalSpanExporterWrapper {
   constructor(
     private underlyingExporter: SpanExporter,
-    private externalTraceId: string
+    private fallback: FallbackExternalTraceIds
   ) {}
 
   private transformSpan(span: ReadableSpan): ReadableSpan | undefined {
@@ -438,7 +510,7 @@ export class ExternalSpanExporterWrapper {
 
     const isExternallySampled = externalTraceContext
       ? isTraceFlagSampled(externalTraceContext.traceFlags)
-      : !!this.externalTraceId;
+      : this.fallback.enabled;
 
     if (!isExternallySampled) {
       return;
@@ -450,7 +522,7 @@ export class ExternalSpanExporterWrapper {
 
     const externalTraceId = externalTraceContext
       ? externalTraceContext.traceId
-      : this.externalTraceId;
+      : this.fallback.forInternalTrace(span.spanContext().traceId);
 
     const isAttemptSpan = span.attributes[SemanticInternalAttributes.SPAN_ATTEMPT];
 
@@ -508,18 +580,18 @@ export class ExternalSpanExporterWrapper {
   }
 }
 
-class ExternalLogRecordExporterWrapper {
+export class ExternalLogRecordExporterWrapper {
   constructor(
     private underlyingExporter: LogRecordExporter,
-    private externalTraceId: string
+    private fallback: FallbackExternalTraceIds
   ) {}
 
   export(logs: any[], resultCallback: (result: any) => void): void {
     const externalTraceContext = traceContext.getExternalTraceContext();
 
     const isExternallySampled = externalTraceContext
       ? isTraceFlagSampled(externalTraceContext.traceFlags)
-      : !!this.externalTraceId;
+      : this.fallback.enabled;
 
     if (!isExternallySampled) {
       this.underlyingExporter.export([], resultCallback);
@@ -550,14 +622,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;
     }