feat: let flush() stop waiting on an AbortSignal

README.md

@@ -58,7 +58,31 @@ started. Invalid options throw.
 
 * `init(options)` activates tracing. Returns nothing. Throws on invalid options.
 * `isActive()` returns whether tracing was activated.
-* `flush()` exports any buffered spans immediately. 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, bounded by a signal so a stuck collector cannot hold the exit:
+
+```js
+process.on('SIGTERM', async () => {
+  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
 
@@ -240,7 +264,7 @@ done while the requests run.
 
 Smaller benchmarks time pieces of that work on their own: building client
 and server spans, generating IDs into a `traceparent` value, and exporting
-batches of spans until the collector has received the last one.
+batches of spans, with each flush awaited through the end of its export.
 
 ## Current limitations
 
@@ -266,7 +290,7 @@ Inbound HTTP/2 context is fully parsed; only the client side is affected.
 Buffered spans are exported periodically, when the buffer fills, and when the
 event loop is about to drain. Spans that are still buffered when `process.exit()`
 is called explicitly are lost, because explicit exits do not run the exit-time
-flush.
+flush. Awaiting `flush()` before the exit sends them.
 
 ## License
 

benchmark/export.bench.js

@@ -7,54 +7,33 @@ const { Span, SPAN_KIND_CLIENT } = require('../lib/span');
 
 const kBatchesPerSample = 20;
 const kSpansPerBatch = 100;
-// A sample takes a few milliseconds. Failed exports only warn, so a lost
-// batch would otherwise leave the sample waiting forever.
-const kSampleTimeoutMs = 5000;
 
 let collector;
-let received = 0;
-let onReceived = () => {};
 
 before(async () => {
   collector = http.createServer((req, res) => {
     req.resume();
-    req.on('end', () => {
-      res.end();
-      received++;
-      onReceived();
-    });
+    req.on('end', () => res.end());
   });
   await new Promise((resolve) => collector.listen(0, '127.0.0.1', resolve));
   microtel.init({
     endpoint: `http://127.0.0.1:${collector.address().port}`,
     // Each batch is flushed by hand, so the buffer never fills on its own.
     maxBufferSize: kSpansPerBatch + 1,
+    // Every flush resolves by this deadline, so a stalled export slows one
+    // sample instead of hanging the run.
+    exportTimeout: 250,
   });
 });
 
 after(() => new Promise((resolve) => collector.close(resolve)));
 
-// Covers ending spans, serializing each batch, and sending it, up to the
-// collector receiving the last batch. flush() does not report when its
-// export finishes, so the collector counts the batches instead.
+// Covers ending spans, serializing each batch, and sending it, through
+// the end of the last export. Each flush resolves once its own export,
+// and any still in flight, has finished.
 bench('export spans to a collector', { samples: 30 }, async (b) => {
-  const target = received + kBatchesPerSample;
-  let watchdog;
-  const done = new Promise((resolve, reject) => {
-    onReceived = () => {
-      if (received >= target) resolve();
-    };
-    watchdog = setTimeout(() => {
-      const missing = target - received;
-      reject(
-        new Error(
-          `${missing} of ${kBatchesPerSample} batches never reached the collector`,
-        ),
-      );
-    }, kSampleTimeoutMs);
-  });
-
   b.start();
+  const flushes = [];
   for (let batch = 0; batch < kBatchesPerSample; batch++) {
     for (let i = 0; i < kSpansPerBatch; i++) {
       const span = new Span('GET', SPAN_KIND_CLIENT);
@@ -64,12 +43,8 @@ bench('export spans to a collector', { samples: 30 }, async (b) => {
       span.setAttribute('server.port', 80);
       span.end();
     }
-    microtel.flush();
-  }
-  try {
-    await done;
-  } finally {
-    clearTimeout(watchdog);
+    flushes.push(microtel.flush());
   }
+  await Promise.all(flushes);
   b.end(kBatchesPerSample * kSpansPerBatch);
 });

index.js

@@ -3,7 +3,8 @@
 // @nodejs/microtel: OpenTelemetry tracing for Node.js core protocols.
 //
 // init() activates tracing (at most once per process, and it cannot be
-// stopped), isActive() reports state, and flush() forces an export.
+// stopped), isActive() reports state, and flush() forces an export whose
+// returned promise resolves once it has finished.
 
 const { start, isActive } = require('./lib/core');
 const { flush } = require('./lib/flush');

lib/flush.js

@@ -3,6 +3,7 @@
 const { Buffer } = require('node:buffer');
 const http = require('node:http');
 const https = require('node:https');
+const { addAbortListener } = require('node:events');
 const { clearTimeout, setImmediate, setTimeout } = require('node:timers');
 
 const {
@@ -12,6 +13,7 @@ const {
   getSdkName,
   getSdkVersion,
 } = require('./core');
+const { validateAbortSignal, validateObject } = require('./validate');
 
 const kDefaultMaxBufferSize = 100;
 const kDefaultFlushIntervalMs = 10_000;
@@ -39,6 +41,9 @@ let maxBufferSize = kDefaultMaxBufferSize;
 let exportTimeoutMs = kDefaultExportTimeoutMs;
 let exportFailureCount = 0;
 let lastExportWarningTime = 0;
+// Exports whose requests have not settled yet. flush() snapshots it so a
+// later flush also waits for the exports an earlier one started.
+const inFlightExports = new Set();
 
 let cachedResource = null;
 let cachedScope = null;
@@ -170,13 +175,41 @@ function addSpan(span) {
   }
 }
 
-function flush() {
+// 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(() => {});
+}
+
+// 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;
   }
-  if (spanBuffer.length === 0) return;
-  if (getEndpoint() == null) return;
+
+  // The returned promise covers exactly the spans buffered and the
+  // exports in flight when flush() runs: spans ending later belong to a
+  // later flush.
+  const pending = [...inFlightExports];
+
+  if (spanBuffer.length === 0 || getEndpoint() == null)
+    return settleAll(pending);
 
   const spans = spanBuffer;
   spanBuffer = [];
@@ -192,7 +225,7 @@ function flush() {
     }
   }
 
-  if (otlpSpans.length === 0) return;
+  if (otlpSpans.length === 0) return settleAll(pending);
 
   let payload;
   try {
@@ -213,10 +246,38 @@ function flush() {
     warnExportFailure(
       `Failed to serialize ${otlpSpans.length} spans: ${err.message}`,
     );
-    return;
+    return settleAll(pending);
   }
 
-  sendToCollector(payload);
+  pending.push(sendToCollector(payload));
+  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();
+    });
+  });
 }
 
 // Counts failures and emits an OTelExportWarning, throttled to at most one
@@ -236,6 +297,20 @@ function sendToCollector(body) {
   const endpoint = getEndpoint();
   if (endpoint == null) return;
 
+  // Settled by done() from whichever comes first: the response ending, a
+  // request or response error, or the export deadline. Settling twice is
+  // harmless, since a promise keeps its first resolution.
+  let settle;
+  const exportDone = new Promise((resolve) => {
+    settle = resolve;
+  });
+  const done = () => {
+    inFlightExports.delete(exportDone);
+    settle();
+  };
+  inFlightExports.add(exportDone);
+
+  let req;
   try {
     // The endpoint is fixed after start, so core resolved the export URL
     // once at activation.
@@ -248,7 +323,7 @@ function sendToCollector(body) {
     // encoded again on write.
     const encoded = Buffer.from(body);
 
-    const req = transport.request(
+    req = transport.request(
       {
         hostname: parsed.hostname,
         port: parsed.port,
@@ -273,12 +348,14 @@ function sendToCollector(body) {
               `OTLP collector responded with HTTP ${res.statusCode}`,
             );
           }
+          done();
         });
         res.on('error', (err) => {
           clearDeadline();
           warnExportFailure(
             `OTLP export response stream error: ${err.message}`,
           );
+          done();
         });
       },
     );
@@ -298,20 +375,32 @@ function sendToCollector(body) {
       warnExportFailure(
         `Failed to export spans to ${endpoint}: ${err.message}`,
       );
+      done();
     });
 
     req.end(encoded);
   } catch (err) {
-    warnExportFailure(`Failed to export spans to ${endpoint}: ${err.message}`);
+    // Once the request exists, destroying it releases its socket and runs
+    // its 'error' listener, which clears the deadline, reports the
+    // failure, and settles the export.
+    if (req !== undefined) {
+      req.destroy(err);
+    } else {
+      done();
+      warnExportFailure(
+        `Failed to export spans to ${endpoint}: ${err.message}`,
+      );
+    }
   }
+  return exportDone;
 }
 
 function startFlusher(options) {
   maxBufferSize = options?.maxBufferSize ?? kDefaultMaxBufferSize;
   flushIntervalMs = options?.flushInterval ?? kDefaultFlushIntervalMs;
   exportTimeoutMs = options?.exportTimeout ?? kDefaultExportTimeoutMs;
 
-  process.on('beforeExit', flush);
+  process.on('beforeExit', () => flush());
 }
 
 // Resets per-run state for the test suite; tracing itself never resets
@@ -330,6 +419,7 @@ function resetCaches() {
   lastExportWarningTime = 0;
   maxBufferSize = kDefaultMaxBufferSize;
   spanBuffer = [];
+  inFlightExports.clear();
 }
 
 module.exports = {

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/otel.js

@@ -85,9 +85,11 @@ async function startTracing(options) {
   return collector;
 }
 
-// Flushes the span buffer and resolves with the spans of the next export.
+// Flushes the span buffer and resolves with the spans of the next
+// export. The payload has reached the collector by the time flush()
+// resolves.
 async function flushAndCollect(collector) {
-  flush();
+  await flush();
   return getSpans(await collector.next());
 }