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

dev-packages/node-integration-tests/suites/tracing/flush-on-exit/scenario.ts

@@ -0,0 +1,13 @@
+import * as Sentry from '@sentry/node';
+import { loggingTransport } from '@sentry-internal/node-integration-tests';
+
+Sentry.init({
+  dsn: 'https://public@dsn.ingest.sentry.io/1337',
+  release: '1.0',
+  tracesSampleRate: 1.0,
+  transport: loggingTransport,
+});
+
+Sentry.startSpan({ name: 'cron-job' }, () => {
+  Sentry.startSpan({ name: 'step' }, () => {});
+});

dev-packages/node-integration-tests/suites/tracing/flush-on-exit/test.ts

@@ -0,0 +1,14 @@
+import { expect, test } from 'vitest';
+import { createRunner } from '../../../utils/runner';
+
+test('sends buffered spans when the process exits on its own', async () => {
+  await createRunner(__dirname, 'scenario.ts')
+    .withoutAutoFlush()
+    .expect({
+      span: container => {
+        expect(container.items.map(span => span.name).sort()).toEqual(['cron-job', 'step']);
+      },
+    })
+    .start()
+    .completed();
+});

dev-packages/node-integration-tests/utils/runner/createRunner.ts

@@ -180,6 +180,7 @@ export function createRunner(...paths: string[]) {
   let withEnv: Record<string, string | undefined> = {};
   let withSentryServer = false;
   let ensureNoErrorOutput = false;
+  let autoFlush = true;
   // When set, the test using this runner expects `completed()` to reject (e.g. `test.fails` variants
   // created via `createEsmAndCjsTests` with `failsOnEsm`/`failsOnCjs`). We suppress the captured-log
   // dump in that case, since the failure is expected and the output would just be noise.
@@ -330,6 +331,11 @@ export function createRunner(...paths: string[]) {
       unordered = true;
       return this;
     },
+    /** Skip the injected `beforeExit` flush, so the test sees only what the SDK sends on its own. */
+    withoutAutoFlush: function () {
+      autoFlush = false;
+      return this;
+    },
     ensureNoErrorOutput: function () {
       if (expectedEnvelopes.length > 0) {
         throw new Error('You should not use `ensureNoErrorOutput` when using `expect`!');
@@ -630,7 +636,7 @@ export function createRunner(...paths: string[]) {
           // `should-exit`, `ensureNoErrorOutput`) verify the child exits
           // naturally and auto-flush would delay that with retrying HTTP
           // requests to the fake DSN.
-          const wantsAutoFlush = !ensureNoErrorOutput && hasExpectations;
+          const wantsAutoFlush = autoFlush && !ensureNoErrorOutput && hasExpectations;
           const runtime = getRuntime();
           const childFlags = wantsAutoFlush ? [...buildAutoFlushFlags(flags, testPath, runtime), ...flags] : flags;
           const entryPath = buildScenario(runtime, testPath);

packages/deno/src/client.ts

@@ -1,5 +1,5 @@
 import type { ServerRuntimeClientOptions } from '@sentry/core/server';
-import { _INTERNAL_flushLogsBuffer, _INTERNAL_flushMetricsBuffer, SDK_VERSION } from '@sentry/core';
+import { SDK_VERSION } from '@sentry/core';
 import { ServerRuntimeClient } from '@sentry/core/server';
 import { registerDiagnosticsChannelInjection } from '@sentry/server-runtime-injection/register';
 import { setAsyncLocalStorageAsyncContextStrategy } from '@sentry/server-utils';
@@ -25,8 +25,7 @@ function getHostName(): string | undefined {
  * @see SentryClient for usage documentation.
  */
 export class DenoClient extends ServerRuntimeClient<DenoClientOptions> {
-  private _logOnExitFlushListener: (() => void) | undefined;
-  private _metricsOnExitFlushListener: (() => void) | undefined;
+  private _onExitFlushListener: (() => void) | undefined;
 
   /**
    * Creates a new Deno SDK instance.
@@ -56,12 +55,10 @@ export class DenoClient extends ServerRuntimeClient<DenoClientOptions> {
 
     super(clientOptions);
 
-    this._logOnExitFlushListener = () => {
-      _INTERNAL_flushLogsBuffer(this);
-    };
-
-    this._metricsOnExitFlushListener = () => {
-      _INTERNAL_flushMetricsBuffer(this);
+    // The log, metric and span buffers drain on `flush`. Their timers are unref'd, so a natural exit
+    // would drop what they hold.
+    this._onExitFlushListener = () => {
+      this.emit('flush');
     };
 
     if (serverName) {
@@ -74,10 +71,8 @@ export class DenoClient extends ServerRuntimeClient<DenoClientOptions> {
     }
 
     // Unlike unload, beforeExit lets the transport finish asynchronous sends.
-    process.on('beforeExit', this._logOnExitFlushListener);
-    process.on('beforeExit', this._metricsOnExitFlushListener);
-    globalThis.addEventListener('unload', this._logOnExitFlushListener);
-    globalThis.addEventListener('unload', this._metricsOnExitFlushListener);
+    process.on('beforeExit', this._onExitFlushListener);
+    globalThis.addEventListener('unload', this._onExitFlushListener);
 
     // In the constructor, as in `NodeClient`, so a directly constructed client installs the hooks
     // too. Registration is idempotent.
@@ -102,14 +97,9 @@ export class DenoClient extends ServerRuntimeClient<DenoClientOptions> {
   /** @inheritDoc */
   // @ts-expect-error - PromiseLike is a subset of Promise
   public async close(timeout?: number | undefined): PromiseLike<boolean> {
-    if (this._logOnExitFlushListener) {
-      process.off('beforeExit', this._logOnExitFlushListener);
-      globalThis.removeEventListener('unload', this._logOnExitFlushListener);
-    }
-
-    if (this._metricsOnExitFlushListener) {
-      process.off('beforeExit', this._metricsOnExitFlushListener);
-      globalThis.removeEventListener('unload', this._metricsOnExitFlushListener);
+    if (this._onExitFlushListener) {
+      process.off('beforeExit', this._onExitFlushListener);
+      globalThis.removeEventListener('unload', this._onExitFlushListener);
     }
 
     return super.close(timeout);

packages/deno/test/mod.test.ts

@@ -4,7 +4,15 @@ import { createStackParser, forEachEnvelopeItem } from '@sentry/core';
 import { nodeStackLineParser } from '@sentry/core/server';
 import { assertEquals } from 'https://deno.land/std@0.202.0/assert/assert_equals.ts';
 import { assertSnapshot } from 'https://deno.land/std@0.202.0/testing/snapshot.ts';
-import { DenoClient, getCurrentScope, getDefaultIntegrations, logger, metrics, Scope } from '../build/esm/index.js';
+import {
+  DenoClient,
+  getCurrentScope,
+  getDefaultIntegrations,
+  logger,
+  metrics,
+  Scope,
+  startSpan,
+} from '../build/esm/index.js';
 import { getNormalizedEvent } from './normalize.ts';
 import { makeTestTransport } from './transport.ts';
 
@@ -70,15 +78,15 @@ function expectCaptureExceptionEvent(event: Event | undefined): void {
         filename: 'app:///test/mod.test.ts',
         function: '?',
         in_app: true,
-        lineno: 46,
+        lineno: 54,
       },
       {
         colno: 12,
         context_line: "    return new Error('Some unhandled error');",
         filename: 'app:///test/mod.test.ts',
         function: 'something',
         in_app: true,
-        lineno: 43,
+        lineno: 51,
       },
     ],
   );
@@ -232,7 +240,36 @@ Deno.test('preserves existing log attributes when adding server.address', () =>
   assertEquals(log.attributes?.['server.address'], 'test-server');
 });
 
-Deno.test('close() removes log and metric exit listeners', async () => {
+Deno.test('sends buffered streamed spans before the process exits', async () => {
+  const itemTypes: string[] = [];
+  const client = new DenoClient({
+    dsn: 'https://233a45e5efe34c47a3536797ce15dafa@nothing.here/5650507',
+    integrations: [],
+    stackParser: createStackParser(nodeStackLineParser()),
+    tracesSampleRate: 1,
+    traceLifecycle: 'stream',
+    transport: makeTestTransport(envelope => {
+      forEachEnvelopeItem(envelope, (_, type) => {
+        itemTypes.push(type);
+      });
+    }),
+  });
+  client.init();
+  getCurrentScope().setClient(client);
+
+  startSpan({ name: 'cron-job' }, () => {});
+  await delay(0);
+  assertEquals(itemTypes, []);
+
+  process.emit('beforeExit', 0);
+  await delay(0);
+
+  assertEquals(itemTypes, ['span']);
+
+  await client.close();
+});
+
+Deno.test('close() removes the exit flush listener', async () => {
   const originalListeners = process.listeners('beforeExit');
   const removeEventListenerCalls: Array<string> = [];
   const originalRemoveEventListener = globalThis.removeEventListener;
@@ -254,11 +291,11 @@ Deno.test('close() removes log and metric exit listeners', async () => {
 
     await client.close();
 
-    assertEquals(addedListeners.length, 2);
+    assertEquals(addedListeners.length, 1);
     assertEquals(process.listeners('beforeExit'), originalListeners);
     assertEquals(
       removeEventListenerCalls.filter(event => event === 'unload'),
-      ['unload', 'unload'],
+      ['unload'],
     );
   } finally {
     globalThis.removeEventListener = originalRemoveEventListener;

packages/node/src/sdk/client.ts

@@ -4,8 +4,6 @@ import { trace } from '@opentelemetry/api';
 import type { ServerRuntimeClientOptions } from '@sentry/core/server';
 import {
   _INTERNAL_clearAiProviderSkips,
-  _INTERNAL_flushLogsBuffer,
-  _INTERNAL_flushMetricsBuffer,
   _INTERNAL_setDeferSegmentSpanCapture,
   applySdkMetadata,
   debug,
@@ -38,8 +36,7 @@ export class NodeClient extends ServerRuntimeClient<NodeClientOptions> {
   private _tracer: Tracer | undefined;
   private _clientReportInterval: NodeJS.Timeout | undefined;
   private _clientReportOnExitFlushListener: (() => void) | undefined;
-  private _logOnExitFlushListener: (() => void) | undefined;
-  private _metricsOnExitFlushListener: (() => void) | undefined;
+  private _onExitFlushListener: (() => void) | undefined;
 
   public constructor(options: NodeClientOptions) {
     const serverName =
@@ -61,12 +58,10 @@ export class NodeClient extends ServerRuntimeClient<NodeClientOptions> {
 
     super(clientOptions);
 
-    this._logOnExitFlushListener = () => {
-      _INTERNAL_flushLogsBuffer(this);
-    };
-
-    this._metricsOnExitFlushListener = () => {
-      _INTERNAL_flushMetricsBuffer(this);
+    // The log, metric and span buffers drain on `flush`. Their timers are unref'd, so a natural exit
+    // would drop what they hold.
+    this._onExitFlushListener = () => {
+      this.emit('flush');
     };
 
     if (serverName) {
@@ -78,8 +73,7 @@ export class NodeClient extends ServerRuntimeClient<NodeClientOptions> {
       });
     }
 
-    process.on('beforeExit', this._logOnExitFlushListener);
-    process.on('beforeExit', this._metricsOnExitFlushListener);
+    process.on('beforeExit', this._onExitFlushListener);
 
     // 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
@@ -161,12 +155,8 @@ export class NodeClient extends ServerRuntimeClient<NodeClientOptions> {
       process.off('beforeExit', this._clientReportOnExitFlushListener);
     }
 
-    if (this._logOnExitFlushListener) {
-      process.off('beforeExit', this._logOnExitFlushListener);
-    }
-
-    if (this._metricsOnExitFlushListener) {
-      process.off('beforeExit', this._metricsOnExitFlushListener);
+    if (this._onExitFlushListener) {
+      process.off('beforeExit', this._onExitFlushListener);
     }
 
     const allEventsSent = await super.close(timeout);

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 the exit flush listener 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(1);
       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);