From 334538ab1eb3bdaa94917035361c98fd859566f5 Mon Sep 17 00:00:00 2001 From: Bryan English Date: Tue, 3 Nov 2026 16:44:13 -0500 Subject: [PATCH] feat: let flush() stop waiting on an AbortSignal Previously, a caller could not bound how long flush() waited. Now flush({ signal }) takes an AbortSignal, such as AbortSignal.timeout(ms), the way core APIs take one, and rejects with an AbortError shaped like core's when it aborts. Export failures still resolve the promise. Aborting stops the waiting, not the export, which keeps running, bounded by exportTimeout. With a signal that has already aborted, flush() rejects right away and starts no export, so the spans stay buffered for the next flush. The abort listener runs even when an earlier listener stops the event's propagation, as core's do. validateAbortSignal is ported from core into lib/validate.js, with parity tests. Signed-off-by: Bryan English --- README.md | 21 ++- lib/flush.js | 54 +++++++- lib/validate.js | 17 ++- test/common/validate-battery.js | 8 ++ test/fixtures/flush-abort-child.js | 63 +++++++++ test/fixtures/parity-child.js | 2 + test/test-flush-promise.js | 215 ++++++++++++++++++++++++++++- 7 files changed, 365 insertions(+), 15 deletions(-) create mode 100644 test/fixtures/flush-abort-child.js diff --git a/README.md b/README.md index e8c5e63..f0d505c 100644 --- a/README.md +++ b/README.md @@ -59,19 +59,32 @@ 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 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 diff --git a/lib/flush.js b/lib/flush.js index 670c375..0c9520a 100644 --- a/lib/flush.js +++ b/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 { @@ -13,6 +14,7 @@ const { getSdkName, getSdkVersion, } = require('./core'); +const { validateAbortSignal, validateObject } = require('./validate'); const kDefaultMaxBufferSize = 100; const kDefaultFlushIntervalMs = 10_000; @@ -174,16 +176,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; @@ -239,6 +254,33 @@ 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(); + }); + }); +} + // Counts failures and emits an OTelExportWarning, throttled to at most one // per kWarningThrottleMs. function warnExportFailure(message) { @@ -360,7 +402,7 @@ function startFlusher(options) { 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 diff --git a/lib/validate.js b/lib/validate.js index 7323818..4451fd7 100644 --- a/lib/validate.js +++ b/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, diff --git a/test/common/validate-battery.js b/test/common/validate-battery.js index e9190f9..ea923a3 100644 --- a/test/common/validate-battery.js +++ b/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]], diff --git a/test/fixtures/flush-abort-child.js b/test/fixtures/flush-abort-child.js new file mode 100644 index 0000000..c2b068e --- /dev/null +++ b/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')); + }); diff --git a/test/fixtures/parity-child.js b/test/fixtures/parity-child.js index 5f98d6a..2c0d18d 100644 --- a/test/fixtures/parity-child.js +++ b/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) => diff --git a/test/test-flush-promise.js b/test/test-flush-promise.js index 74ac974..daddfd1 100644 --- a/test/test-flush-promise.js +++ b/test/test-flush-promise.js @@ -2,10 +2,12 @@ const assert = require('node:assert'); const http = require('node:http'); +const { getEventListeners } = require('node:events'); const { after, before, describe, it } = require('node:test'); const { flush } = require('../lib/flush'); const { fixture, + getClientSpan, getServerSpan, getSpans, httpGet, @@ -16,9 +18,11 @@ const { // flush() returns a promise that resolves once every span buffered when // it was called has been exported or dropped, and every export already in -// flight has finished. It resolves, never rejects. Tracing activates once -// per process, so the in-process tests share one collector and the other -// configurations run in child processes through fixtures. +// flight has finished. Export failures resolve it, since they are already +// reported as warnings. Only an abort the caller asked for through +// options.signal rejects it. Tracing activates once per process, so the +// in-process tests share one collector and the other configurations run +// in child processes through fixtures. describe('flush() promise', () => { let collector; @@ -201,4 +205,209 @@ describe('flush() promise', () => { collector.close(); } }); + + it('exports as before with no signal and with empty options', async () => { + const server = await startHTTPServer(); + await httpGet(server.port, '/flush-no-signal'); + await server.close(); + + await flush(); + await flush({}); + + const allSpans = collector.payloads.flatMap(getSpans); + assert.ok(getServerSpan(allSpans, '/flush-no-signal')); + }); + + it('throws ERR_INVALID_ARG_TYPE on invalid options and signal', () => { + assert.throws(() => flush(null), { + name: 'TypeError', + code: 'ERR_INVALID_ARG_TYPE', + }); + assert.throws(() => flush('options'), { + name: 'TypeError', + code: 'ERR_INVALID_ARG_TYPE', + }); + assert.throws(() => flush({ signal: null }), { + name: 'TypeError', + code: 'ERR_INVALID_ARG_TYPE', + }); + assert.throws(() => flush({ signal: 'signal' }), { + name: 'TypeError', + code: 'ERR_INVALID_ARG_TYPE', + }); + assert.throws(() => flush({ signal: {} }), { + name: 'TypeError', + code: 'ERR_INVALID_ARG_TYPE', + }); + }); + + it('rejects with an AbortError and keeps the spans when the signal is already aborted', async () => { + const server = await startHTTPServer(); + await httpGet(server.port, '/flush-already-aborted'); + await server.close(); + + const controller = new AbortController(); + controller.abort(); + const before = collector.requests.length; + + await assert.rejects(flush({ signal: controller.signal }), { + name: 'AbortError', + code: 'ABORT_ERR', + message: 'The operation was aborted', + cause: controller.signal.reason, + }); + + // No export was started, so the spans are still buffered. + assert.strictEqual(collector.requests.length, before); + + await flush(); + assert.ok(collector.requests.length > before); + const allSpans = collector.payloads.flatMap(getSpans); + assert.ok(getServerSpan(allSpans, '/flush-already-aborted')); + }); + + it('rejects right away when aborted mid-export, and the export still finishes', async () => { + // The collector holds the export's response until the second release, + // which the child sends only after the rejection, so the abort finds + // the export in flight and the rejection cannot wait for it. + let exportArrived = false; + let exportBody = ''; + let pendingExportRes; + let pendingReleaseRes; + const collector = http.createServer((req, res) => { + if (req.url === '/release-1' || req.url === '/release-2') { + req.resume(); + req.on('end', () => { + if (req.url === '/release-2' && pendingExportRes !== undefined) { + pendingExportRes.writeHead(200); + pendingExportRes.end(); + pendingExportRes = undefined; + } + if (exportArrived || req.url === '/release-2') { + res.writeHead(200); + res.end(); + } else { + pendingReleaseRes = res; + } + }); + return; + } + req.on('data', (chunk) => { + exportBody += chunk; + }); + req.on('end', () => { + exportArrived = true; + if (pendingReleaseRes !== undefined) { + pendingReleaseRes.writeHead(200); + pendingReleaseRes.end(); + pendingReleaseRes = undefined; + } + pendingExportRes = res; + }); + }); + await new Promise((resolve) => collector.listen(0, '127.0.0.1', resolve)); + + try { + const { code, stdout } = await spawnMicrotel( + [ + fixture('flush-abort-child.js'), + JSON.stringify({ + endpoint: `http://127.0.0.1:${collector.address().port}`, + }), + ], + {}, + { timeout: 5_000 }, + ); + + assert.strictEqual(code, 0); + assert.strictEqual( + stdout.trim(), + 'rejected AbortError ABORT_ERR\nlisteners 0\n' + + 'second flush pending\nsecond flush settled', + ); + assert.ok( + getClientSpan(getSpans(JSON.parse(exportBody)), '127.0.0.1:1/'), + ); + } finally { + collector.close(); + } + }); + + it('rejects even when an earlier abort listener stops propagation', async () => { + // Core's own abort listeners run regardless, and so does this one, or + // a caller's listener could hold the flush open. + const controller = new AbortController(); + controller.signal.addEventListener('abort', (event) => + event.stopImmediatePropagation(), + ); + const flushed = flush({ signal: controller.signal }); + controller.abort(); + await assert.rejects(flushed, { name: 'AbortError', code: 'ABORT_ERR' }); + }); + + it('rejects with the timeout reason as cause', async () => { + // The collector accepts the connection but never answers, so only the + // signal can settle the flush first. + const collector = http.createServer(() => {}); + await new Promise((resolve) => collector.listen(0, '127.0.0.1', resolve)); + + try { + const { code, stdout } = await spawnMicrotel( + [ + '-e', + ` + const http = require('node:http'); + const { init, flush } = require(process.env.MICROTEL_PATH); + init({ + endpoint: 'http://127.0.0.1:${collector.address().port}', + exportTimeout: 250, + }); + http.get('http://127.0.0.1:1/', () => {}) + .on('error', async () => { + try { + await flush({ signal: AbortSignal.timeout(50) }); + console.log('flush resolved'); + } catch (err) { + console.log(JSON.stringify({ + name: err.name, + code: err.code, + message: err.message, + cause: err.cause.name, + })); + } + }); + `, + ], + {}, + { timeout: 5_000 }, + ); + + assert.strictEqual(code, 0); + assert.deepStrictEqual(JSON.parse(stdout), { + name: 'AbortError', + code: 'ABORT_ERR', + message: 'The operation was aborted', + cause: 'TimeoutError', + }); + } finally { + collector.close(); + } + }); + + it('drops the abort listener once flush() settles', async () => { + const server = await startHTTPServer(); + await httpGet(server.port, '/flush-listener'); + await server.close(); + + const controller = new AbortController(); + const signal = controller.signal; + const wait = flush({ signal }); + assert.strictEqual(getEventListeners(signal, 'abort').length, 1); + await wait; + assert.strictEqual(getEventListeners(signal, 'abort').length, 0); + + // A long-lived signal reused across flushes collects no listeners. + for (let i = 0; i < 3; i++) await flush({ signal }); + assert.strictEqual(getEventListeners(signal, 'abort').length, 0); + }); });