feat: let flush() stop waiting on an AbortSignal

README.md

@@ -72,19 +72,32 @@ can replace any of microtel's values except `service.name` and the
 
 * `init(options)` activates tracing. Returns nothing. Throws on invalid options.
 * `isActive()` returns whether tracing was activated.
-* `flush()` exports any buffered spans immediately and returns a promise. The promise resolves once those spans and any export already in flight are done, whether or not they reached the collector. It never rejects, since failed exports are already reported as `OTelExportWarning` warnings. Also runs on `beforeExit`.
+* `flush([options])` exports any buffered spans immediately and returns a promise. The promise resolves once those spans and any export already in flight are done, whether or not they reached the collector. Export failures resolve it rather than reject it, since they are already reported as `OTelExportWarning` warnings. Also runs on `beforeExit`.
+
+`flush({ signal })` takes an `AbortSignal` that bounds the wait, such as
+`AbortSignal.timeout(ms)`. When it aborts, the promise rejects with an
+`AbortError` shaped like the one Node.js core APIs reject with: its code is
+`ABORT_ERR`, and its `cause` is the signal's reason. Aborting stops the waiting,
+not the export, which keeps running, bounded by `exportTimeout`. With a
+signal that has already aborted, the promise rejects right away and no export
+starts, so the spans stay buffered for a later flush.
 
 Explicit exits skip the `beforeExit` flush, and so does a serverless platform
 that freezes the process after each invocation. Await `flush()` first in either
-case:
+case, bounded by a signal so a stuck collector cannot hold the exit:
 
 ```js
 process.on('SIGTERM', async () => {
-  await otel.flush();
-  process.exit(0);
+  try {
+    await otel.flush({ signal: AbortSignal.timeout(5000) });
+  } finally {
+    process.exit(0);
+  }
 });
 ```
 
+A flush that times out rejects, and the `finally` exits anyway.
+
 ## Instrumented operations
 
 When tracing is active, spans are created automatically for the following

lib/flush.js

@@ -4,6 +4,7 @@ const { Buffer } = require('node:buffer');
 const http = require('node:http');
 const https = require('node:https');
 const { hostname } = require('node:os');
+const { addAbortListener } = require('node:events');
 const { clearTimeout, setImmediate, setTimeout } = require('node:timers');
 
 const { guard } = require('./guard');
@@ -18,6 +19,7 @@ const {
   getSdkName,
   getSdkVersion,
 } = require('./core');
+const { validateAbortSignal, validateObject } = require('./validate');
 
 const kDefaultMaxBufferSize = 100;
 const kDefaultFlushIntervalMs = 10_000;
@@ -200,16 +202,29 @@ function addSpan(span) {
   }
 }
 
-// Resolves once every export in the list has settled. flush() resolves
-// rather than rejects: failures warn through warnExportFailure, and a
-// rejected promise would surface as an unhandled rejection in every
-// caller that ignores the return value.
+// Resolves once every export in the list has settled. It resolves
+// rather than rejects even for export failures: they warn through
+// warnExportFailure, and a rejection would surface as an unhandled
+// rejection in every caller that ignores the return value.
 function settleAll(pending) {
   if (pending.length === 0) return Promise.resolve();
   return Promise.all(pending).then(() => {});
 }
 
-function flush() {
+// Node core builds its AbortError from an internal class this package
+// cannot import, so flush() builds the same shape itself.
+function abortError(signal) {
+  const err = new Error('The operation was aborted', {
+    cause: signal.reason,
+  });
+  err.code = 'ABORT_ERR';
+  err.name = 'AbortError';
+  return err;
+}
+
+// Starts an export of every buffered span and resolves once it and every
+// export already in flight have settled.
+function exportBuffered() {
   if (flushTimer != null) {
     clearTimeout(flushTimer);
     flushTimer = null;
@@ -265,14 +280,43 @@ function flush() {
   return settleAll(pending);
 }
 
+function flush(options = { __proto__: null }) {
+  validateObject(options, 'options');
+  const { signal } = options;
+  validateAbortSignal(signal, 'options.signal');
+
+  // With a signal that has already aborted, the promise rejects before any
+  // export starts, the way core's APIs check signal.aborted first. The
+  // spans stay buffered for a later flush.
+  if (signal?.aborted) return Promise.reject(abortError(signal));
+
+  const settled = exportBuffered();
+  if (signal === undefined) return settled;
+
+  return new Promise((resolve, reject) => {
+    // addAbortListener's listener runs even when an earlier listener stops
+    // the event's propagation, as core's own abort listeners do.
+    const listener = addAbortListener(signal, () => reject(abortError(signal)));
+    // Aborting stops the waiting, not the export: the export it took
+    // keeps running in the background, bounded by exportTimeout, and its
+    // spans still reach the collector if it succeeds.
+    settled.then(() => {
+      listener[Symbol.dispose]();
+      resolve();
+    });
+  });
+}
+
 function deferredFlush() {
   flushScheduled = false;
   flush();
 }
 
 const guardedDeferredFlush = guard('deferred flush', deferredFlush);
 const guardedTimerFlush = guard('flush timer', flush);
-const guardedBeforeExitFlush = guard('beforeExit flush', flush);
+// Node passes the exit code to beforeExit listeners, which flush() would
+// reject as its options.
+const guardedBeforeExitFlush = guard('beforeExit flush', () => flush());
 
 // Counts failures and emits an OTelExportWarning, throttled to at most one
 // per kWarningThrottleMs.

lib/validate.js

@@ -179,8 +179,20 @@ function outOfRangeError(str, range, input) {
 }
 
 // Validators, from lib/internal/validators.js. Only the default behavior
-// is ported: validateObject without option flags, validateInteger with an
-// optional minimum, and validateNumber with optional bounds.
+// is ported: validateObject without option flags, validateAbortSignal,
+// validateInteger with an optional minimum, and validateNumber with
+// optional bounds.
+
+// From lib/internal/validators.js: an AbortSignal is anything with an
+// 'aborted' property, the same duck-typed check core performs.
+function validateAbortSignal(signal, name) {
+  if (
+    signal !== undefined &&
+    (signal === null || typeof signal !== 'object' || !('aborted' in signal))
+  ) {
+    throw invalidArgTypeError(name, 'AbortSignal', signal);
+  }
+}
 
 function validateString(value, name) {
   if (typeof value !== 'string')
@@ -227,6 +239,7 @@ function validateNumber(value, name, min = undefined, max) {
 module.exports = {
   validateString,
   validateObject,
+  validateAbortSignal,
   validateInteger,
   validateNumber,
   invalidArgTypeError,

test/common/validate-battery.js

@@ -43,6 +43,14 @@ const kValidatorCases = [
   ['validateObject', [123, 'x']],
   ['validateObject', [() => {}, 'x']],
   ['validateObject', [{}, 'x']],
+  ['validateAbortSignal', [123, 'x']],
+  ['validateAbortSignal', [null, 'x']],
+  ['validateAbortSignal', ['str', 'x']],
+  ['validateAbortSignal', [true, 'x']],
+  ['validateAbortSignal', [() => {}, 'x']],
+  ['validateAbortSignal', [undefined, 'x']],
+  ['validateAbortSignal', [{ aborted: false }, 'x']],
+  ['validateAbortSignal', [new AbortController().signal, 'x']],
   ['validateInteger', [0, 'x', 1]],
   ['validateInteger', ['abc', 'x', 1]],
   ['validateInteger', [1.5, 'x', 1]],

test/fixtures/flush-abort-child.js

@@ -0,0 +1,63 @@
+'use strict';
+
+// Spawned with JSON init options as argv[2]: buffers one span, flushes
+// with an abort signal, and tells the collector through raw release
+// requests (which the HTTP client instrumentation does not trace) when
+// to answer the export. The printed order proves the flush rejected
+// while the export was still in flight, and that a later flush still
+// waits for that export.
+
+const net = require('node:net');
+const http = require('node:http');
+const { getEventListeners } = require('node:events');
+const { init, flush } = require(process.env.MICROTEL_PATH);
+
+const config = JSON.parse(process.argv[2]);
+const port = new URL(config.endpoint).port;
+
+function rawRelease(port, path) {
+  return new Promise((resolve) => {
+    const socket = net.connect(port, '127.0.0.1', () => {
+      socket.end(
+        `GET ${path} HTTP/1.1\r\n` +
+          `Host: 127.0.0.1\r\n` +
+          `Connection: close\r\n` +
+          `\r\n`,
+      );
+      socket.on('data', () => {});
+      socket.on('end', resolve);
+    });
+  });
+}
+
+init(config);
+
+http
+  .get('http://127.0.0.1:1/', () => {})
+  .on('error', async () => {
+    const controller = new AbortController();
+    const order = [];
+    const rejected = flush({ signal: controller.signal }).then(
+      () => 'resolved',
+      (err) => `rejected ${err.name} ${err.code}`,
+    );
+
+    // The response proves the export has reached the collector, so the
+    // abort below finds it in flight.
+    await rawRelease(port, '/release-1');
+    controller.abort();
+    order.push(await rejected);
+    order.push(
+      `listeners ${getEventListeners(controller.signal, 'abort').length}`,
+    );
+    // A flush started after the abort still waits for the held export: it
+    // is pending a turn later, and settles once the export is answered.
+    let later = 'pending';
+    const second = flush().then(() => (later = 'settled'));
+    await new Promise((resolve) => setImmediate(resolve));
+    order.push(`second flush ${later}`);
+    await rawRelease(port, '/release-2');
+    await second;
+    order.push(`second flush ${later}`);
+    console.log(order.join('\n'));
+  });

test/fixtures/parity-child.js

@@ -7,6 +7,7 @@
 const {
   validateString,
   validateObject,
+  validateAbortSignal,
   validateInteger,
   validateNumber,
 } = require('internal/validators');
@@ -18,6 +19,7 @@ const { runBattery } = require(process.env.BATTERY_MODULE);
 const impl = {
   validateString,
   validateObject,
+  validateAbortSignal,
   validateInteger,
   validateNumber,
   invalidArgTypeError: (name, expected, actual) =>

test/test-beforeexit-flush.js

@@ -16,18 +16,22 @@ describe('otel beforeExit flush', () => {
   it('flushes buffered spans on process beforeExit', async () => {
     const collector = await startOTelCollector();
 
-    // The child makes a traced request and exits without flushing.
-    const { code } = await spawnMicrotel([
-      fixture('beforeexit-flush-child.js'),
-      JSON.stringify({ endpoint: `http://127.0.0.1:${collector.port}` }),
-    ]);
-
-    assert.strictEqual(code, 0);
-
-    const spans = getSpans(await collector.next());
-
-    await collector.close();
-
-    assert.ok(spans.length > 0, 'Expected spans to be flushed on beforeExit');
+    try {
+      // The child makes a traced request and exits without flushing.
+      const { code } = await spawnMicrotel([
+        fixture('beforeexit-flush-child.js'),
+        JSON.stringify({ endpoint: `http://127.0.0.1:${collector.port}` }),
+      ]);
+
+      assert.strictEqual(code, 0);
+
+      // The export keeps the child alive until it finishes, so its payload
+      // is already here once the child exits, and a missing one fails
+      // rather than waits.
+      const spans = collector.payloads.flatMap(getSpans);
+      assert.ok(spans.length > 0, 'Expected spans to be flushed on beforeExit');
+    } finally {
+      await collector.close();
+    }
   });
 });