fix(node): Flush buffered streamed spans on process exit

A process that exits on its own lost every span still in the span buffer,
since its flush timers are unref'd. Flush it on `beforeExit`, like logs
and metrics.

Author: logaretm

packages/node/src/sdk/client.ts

@@ -40,6 +40,7 @@ export class NodeClient extends ServerRuntimeClient<NodeClientOptions> {
   private _clientReportOnExitFlushListener: (() => void) | undefined;
   private _logOnExitFlushListener: (() => void) | undefined;
   private _metricsOnExitFlushListener: (() => void) | undefined;
+  private _spanOnExitFlushListener: (() => void) | undefined;
 
   public constructor(options: NodeClientOptions) {
     const serverName =
@@ -69,6 +70,12 @@ export class NodeClient extends ServerRuntimeClient<NodeClientOptions> {
       _INTERNAL_flushMetricsBuffer(this);
     };
 
+    // The span buffer drains on the client's `flush` hook. Its timers are unref'd, so without this a
+    // process that exits on its own drops every span still waiting in the buffer.
+    this._spanOnExitFlushListener = () => {
+      this.emit('flush');
+    };
+
     if (serverName) {
       this.on('beforeCaptureLog', log => {
         log.attributes = {
@@ -80,6 +87,7 @@ export class NodeClient extends ServerRuntimeClient<NodeClientOptions> {
 
     process.on('beforeExit', this._logOnExitFlushListener);
     process.on('beforeExit', this._metricsOnExitFlushListener);
+    process.on('beforeExit', this._spanOnExitFlushListener);
 
     // Enable deferred segment-span transaction capture here, in the constructor, rather than in
     // `initOtel`. Every client runs its constructor exactly once, whereas `initOtel` only runs on
@@ -169,6 +177,10 @@ export class NodeClient extends ServerRuntimeClient<NodeClientOptions> {
       process.off('beforeExit', this._metricsOnExitFlushListener);
     }
 
+    if (this._spanOnExitFlushListener) {
+      process.off('beforeExit', this._spanOnExitFlushListener);
+    }
+
     const allEventsSent = await super.close(timeout);
     if (this.traceProvider) {
       await this.traceProvider.shutdown();

packages/node/test/sdk/client.test.ts

@@ -1,7 +1,16 @@
 import { AsyncLocalStorage } from 'node:async_hooks';
 import { ProxyTracer } from '@opentelemetry/api';
 import type { Event, EventHint, Log } from '@sentry/core';
-import { getAsyncContextStrategy, getMainCarrier, Scope, SDK_VERSION } from '@sentry/core';
+import {
+  createTransport,
+  getAsyncContextStrategy,
+  getClient,
+  getMainCarrier,
+  resolvedSyncPromise,
+  Scope,
+  SDK_VERSION,
+  startSpan,
+} from '@sentry/core';
 import type { SentryTracerProvider } from '@sentry/opentelemetry';
 import { setOpenTelemetryContextAsyncContextStrategy } from '@sentry/opentelemetry';
 import * as SentryOpentelemetry from '@sentry/opentelemetry';
@@ -10,7 +19,7 @@ import * as os from 'os';
 import { afterEach, beforeEach, describe, expect, it, test, vi } from 'vitest';
 import { NodeClient } from '../../src';
 import { getDefaultNodeClientOptions } from '../helpers/getDefaultNodeClientOptions';
-import { cleanupOtel } from '../helpers/mockSdkInit';
+import { cleanupOtel, mockSdkInit } from '../helpers/mockSdkInit';
 
 describe('NodeClient', () => {
   beforeEach(() => {
@@ -420,20 +429,40 @@ describe('NodeClient', () => {
       expect(clearIntervalSpy).toHaveBeenCalledWith(intervalResult?.value);
     });
 
-    it('removes log and metric exit listeners when closed', async () => {
+    it('removes log, metric and span exit listeners when closed', async () => {
       const originalListeners = process.listeners('beforeExit');
       const client = new NodeClient(getDefaultNodeClientOptions({ sendClientReports: false }));
 
       const addedListeners = process.listeners('beforeExit').filter(listener => !originalListeners.includes(listener));
 
       await client.close();
 
-      expect(addedListeners).toHaveLength(2);
+      expect(addedListeners).toHaveLength(3);
       expect(process.listeners('beforeExit')).toEqual(originalListeners);
     });
   });
 
   describe('flush', () => {
+    it('sends buffered streamed spans before the process exits', async () => {
+      const send = vi.fn(() => resolvedSyncPromise({}));
+      mockSdkInit({
+        tracesSampleRate: 1,
+        traceLifecycle: 'stream',
+        transport: () => createTransport({ recordDroppedEvent: () => undefined }, send),
+      });
+
+      startSpan({ name: 'cron-job' }, () => {});
+      await new Promise(resolve => setImmediate(resolve));
+      expect(send).not.toHaveBeenCalled();
+
+      process.emit('beforeExit', 0);
+
+      expect(send).toHaveBeenCalledTimes(1);
+      expect(send).toHaveBeenCalledWith(expect.objectContaining({ body: expect.stringContaining('"type":"span"') }));
+
+      await getClient()?.close();
+    });
+
     it('flush returns immediately when nothing is processing', async () => {
       const options = getDefaultNodeClientOptions();
       const client = new NodeClient(options);