feat(node): link captured exceptions to active OpenTelemetry spans

.changeset/node-exception-trace-context.md

@@ -0,0 +1,5 @@
+---
+'posthog-node': patch
+---
+
+Link captured exceptions to active OpenTelemetry traces when the API is available.

packages/node/package.json

@@ -47,6 +47,8 @@
   },
   "devDependencies": {
     "@edge-runtime/vm": "^5.0.0",
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/context-async-hooks": "^2.2.0",
     "@posthog-tooling/tsconfig-base": "workspace:*",
     "@posthog/types": "workspace:^",
     "@rslib/core": "catalog:",
@@ -58,9 +60,13 @@
     "typescript": "catalog:native"
   },
   "peerDependencies": {
+    "@opentelemetry/api": "^1.2.0",
     "rxjs": "^7.0.0"
   },
   "peerDependenciesMeta": {
+    "@opentelemetry/api": {
+      "optional": true
+    },
     "rxjs": {
       "optional": true
     }

packages/node/src/__tests__/evaluate-flags.spec.ts

@@ -466,13 +466,7 @@ describe('evaluateFlags', () => {
 
       posthog.captureException(new Error('boom'), 'user-1', undefined, undefined, flags.onlyAccessed())
 
-      // captureException → addPendingPromise(buildEventMessage().then(msg => capture(...)))
-      // → capture itself queues async work via prepareEventMessage. The 'capture' event
-      // fires inside captureStateless before the network flush, so we just need enough
-      // microtask cycles to let the chain resolve.
-      await waitForPromises()
-      await waitForPromises()
-      await waitForPromises()
+      await posthog.flush()
 
       const exception = captures.find((m) => m.event === '$exception')
       expect(exception).toBeDefined()

packages/node/src/__tests__/extensions/exception-trace-context.spec.ts

@@ -0,0 +1,142 @@
+import { context, trace, TraceFlags, type Span, type SpanContext } from '@opentelemetry/api'
+import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks'
+import { PostHog } from '@/entrypoints/index.node'
+
+const traceId = '4bf92f3577b34da6a3ce929d0e0e4736'
+const spanId = '00f067aa0ba902b7'
+
+let client: PostHog
+
+function capture(properties?: Record<string, string>) {
+  return client.prepareEventMessage({ event: '$exception', distinctId: 'user-1', properties })
+}
+
+function withSpan(spanContext: SpanContext, callback: () => ReturnType<typeof capture>) {
+  return context.with(trace.setSpan(context.active(), trace.wrapSpanContext(spanContext)), callback)
+}
+
+describe('exception trace context', () => {
+  const manager = new AsyncLocalStorageContextManager()
+
+  beforeAll(() => {
+    context.setGlobalContextManager(manager.enable())
+    client = new PostHog('test-token', {
+      host: 'https://example.test',
+      fetchRetryCount: 0,
+      fetch: async () => ({ status: 200, text: async () => 'ok', json: async () => ({}) }),
+    })
+  })
+
+  afterAll(async () => {
+    await client.shutdown()
+    context.disable()
+    manager.disable()
+  })
+
+  it.each([TraceFlags.SAMPLED, TraceFlags.NONE])('adds active IDs with trace flags %i', async (traceFlags) => {
+    const event = await withSpan({ traceId, spanId, traceFlags }, () => capture())
+
+    expect(event.properties).toMatchObject({ $trace_id: traceId, $span_id: spanId })
+  })
+
+  it('keeps leading zeros', async () => {
+    const event = await withSpan(
+      { traceId: '00000000000000000000000000000001', spanId: '0000000000000002', traceFlags: TraceFlags.NONE },
+      () => capture()
+    )
+
+    expect(event.properties).toMatchObject({
+      $trace_id: '00000000000000000000000000000001',
+      $span_id: '0000000000000002',
+    })
+  })
+
+  it.each([
+    { traceId: '00000000000000000000000000000000', spanId },
+    { traceId, spanId: '0000000000000000' },
+  ])('omits both IDs when one ID is zero', async ({ traceId: invalidTraceId, spanId: invalidSpanId }) => {
+    const event = await withSpan(
+      { traceId: invalidTraceId, spanId: invalidSpanId, traceFlags: TraceFlags.SAMPLED },
+      () => capture()
+    )
+
+    expect(event.properties).not.toHaveProperty('$trace_id')
+    expect(event.properties).not.toHaveProperty('$span_id')
+  })
+
+  it('omits trace IDs when no span is active', async () => {
+    const event = await capture()
+
+    expect(event.properties).not.toHaveProperty('$trace_id')
+    expect(event.properties).not.toHaveProperty('$span_id')
+  })
+
+  it.each([
+    { $trace_id: 'custom-trace', $span_id: 'custom-span' },
+    { $trace_id: 'custom-trace' },
+    { $span_id: 'custom-span' },
+  ])('preserves caller supplied IDs without mixing trace contexts', async (properties) => {
+    const event = await withSpan({ traceId, spanId, traceFlags: TraceFlags.SAMPLED }, () => capture(properties))
+
+    expect(
+      Object.fromEntries(Object.entries(event.properties).filter(([key]) => key === '$trace_id' || key === '$span_id'))
+    ).toEqual(properties)
+  })
+
+  it.each([
+    { $trace_id: 'context-trace', $span_id: 'context-span' },
+    { $trace_id: 'context-trace' },
+    { $span_id: 'context-span' },
+  ])('preserves request context IDs without mixing trace contexts', async (properties) => {
+    const event = await client.withContext({ properties }, () =>
+      withSpan({ traceId, spanId, traceFlags: TraceFlags.SAMPLED }, () => capture())
+    )
+
+    expect(
+      Object.fromEntries(Object.entries(event.properties).filter(([key]) => key === '$trace_id' || key === '$span_id'))
+    ).toEqual(properties)
+  })
+
+  it('preserves caller supplied IDs when there is no span', async () => {
+    const event = await capture({ $trace_id: 'custom-trace', $span_id: 'custom-span' })
+
+    expect(event.properties).toMatchObject({ $trace_id: 'custom-trace', $span_id: 'custom-span' })
+  })
+
+  it('does not drop an exception when reading the span fails', async () => {
+    const span = {
+      spanContext: () => {
+        throw new Error('span failed')
+      },
+    } as unknown as Span
+    const event = await context.with(trace.setSpan(context.active(), span), () => capture())
+
+    expect(event.event).toBe('$exception')
+    expect(event.properties).not.toHaveProperty('$trace_id')
+    expect(event.properties).not.toHaveProperty('$span_id')
+  })
+
+  it('adds IDs to an exception captured through the public API', async () => {
+    const events: any[] = []
+    client.on('capture', (event) => events.push(event))
+
+    await withSpan({ traceId, spanId, traceFlags: TraceFlags.SAMPLED }, async () => {
+      client.captureException(new Error('boom'), 'user-1')
+      await client.flush()
+    })
+
+    expect(events.find((event) => event.event === '$exception')?.properties).toMatchObject({
+      $trace_id: traceId,
+      $span_id: spanId,
+    })
+  })
+
+  it('does not add trace IDs to other events', async () => {
+    const event = await withSpan({ traceId, spanId, traceFlags: TraceFlags.SAMPLED }, () =>
+      client.prepareEventMessage({ event: 'purchase', distinctId: 'user-1' })
+    )
+
+    expect(event.properties).not.toHaveProperty('$trace_id')
+    expect(event.properties).not.toHaveProperty('$span_id')
+  })
+})

packages/node/src/__tests__/extensions/sentry-integration.spec.ts

@@ -4,7 +4,6 @@ import {
   sentryIntegration,
   type SentryIntegrationOptions,
 } from '@/extensions/sentry-integration'
-import { waitForPromises } from '../utils'
 
 vi.mock('../../version', () => ({ version: '1.2.3' }))
 
@@ -115,9 +114,7 @@ describe('PostHogSentryIntegration', () => {
 
     processorFunction(createMockSentryException())
 
-    await waitForPromises() // First flush
-    vi.runOnlyPendingTimers() // Flush timer
-    await waitForPromises() // Second flush
+    await posthog.flush()
     const batchEvents = getLastBatchEvents()
 
     expect(batchEvents).toEqual([

packages/node/src/client.ts

@@ -62,6 +62,7 @@ import {
   RequiresServerEvaluation,
 } from './extensions/feature-flags/feature-flags'
 import ErrorTracking from './extensions/error-tracking'
+import { getActiveTraceProperties } from './extensions/error-tracking/trace-context'
 import { PostHogMemoryStorage } from './storage-memory'
 import { ContextData, ContextOptions, IPostHogContext } from './extensions/context/types'
 import { type CaptureMode, resolveCaptureMode } from './capture-v1/config'
@@ -3261,6 +3262,10 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen
         }
       : { ...(properties || {}) }
 
+    if (event === '$exception' && !('$trace_id' in mergedProperties || '$span_id' in mergedProperties)) {
+      Object.assign(mergedProperties, await getActiveTraceProperties())
+    }
+
     if (!mergedDistinctId) {
       mergedDistinctId = uuidv7()
       mergedProperties.$process_person_profile = false

packages/node/src/extensions/error-tracking/trace-context.ts

@@ -0,0 +1,28 @@
+const TRACE_ID_PATTERN = /^[0-9a-f]{32}$/i
+const SPAN_ID_PATTERN = /^[0-9a-f]{16}$/i
+
+let otelApi: Promise<typeof import('@opentelemetry/api')> | undefined
+
+export async function getActiveTraceProperties(): Promise<Record<string, string>> {
+  try {
+    const api = await (otelApi ??= import('@opentelemetry/api'))
+    const spanContext = api.trace.getActiveSpan()?.spanContext()
+    const traceId = spanContext?.traceId
+    const spanId = spanContext?.spanId
+
+    if (
+      typeof traceId === 'string' &&
+      typeof spanId === 'string' &&
+      TRACE_ID_PATTERN.test(traceId) &&
+      SPAN_ID_PATTERN.test(spanId) &&
+      !/^0+$/.test(traceId) &&
+      !/^0+$/.test(spanId)
+    ) {
+      return { $trace_id: traceId.toLowerCase(), $span_id: spanId.toLowerCase() }
+    }
+  } catch {
+    return {}
+  }
+
+  return {}
+}