From 59b4deaf47e3fa5c4b758da014d3df9d6630e59f Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Fri, 24 Apr 2026 17:06:37 -0400 Subject: [PATCH 1/6] fix(node): prevent closeAndFlush from silently dropping in-flight events closeAndFlush had two issues that caused data loss under rate limiting: 1. The default timeout (flushInterval * 1.25 = 12.5s) was shorter than a single Retry-After: 60 cycle. Raised the floor to max(60s, flushInterval) * 1.25. 2. On timeout, .catch(() => undefined) silently swallowed the error and returned, but retry loops continued running. The process would then exit killing in-flight retries. Now calls publisher.abort() to cancel pending sleeps via AbortController, and marks affected batches as failed. --- .../graceful-shutdown-integration.test.ts | 16 ++++-- packages/node/src/app/analytics-node.ts | 8 ++- .../node/src/plugins/segmentio/publisher.ts | 53 +++++++++++++++++-- 3 files changed, 67 insertions(+), 10 deletions(-) diff --git a/packages/node/src/__tests__/graceful-shutdown-integration.test.ts b/packages/node/src/__tests__/graceful-shutdown-integration.test.ts index 52c9e6fc4..4324939ab 100644 --- a/packages/node/src/__tests__/graceful-shutdown-integration.test.ts +++ b/packages/node/src/__tests__/graceful-shutdown-integration.test.ts @@ -84,15 +84,23 @@ describe('Ability for users to exit without losing events', () => { }) describe('.closeAndFlush()', () => { - test('default timeout should be related to flush interval', () => { - const flushInterval = 500 + test('default timeout should be at least 75 seconds', () => { ajs = new Analytics({ writeKey: 'abc123', - flushInterval, httpClient: testClient, }) const closeAndFlushTimeout = ajs['_closeAndFlushDefaultTimeout'] - expect(closeAndFlushTimeout).toBe(flushInterval * 1.25) + expect(closeAndFlushTimeout).toBe(75000) + }) + + test('default timeout should scale with large flushInterval', () => { + ajs = new Analytics({ + writeKey: 'abc123', + flushInterval: 120000, + httpClient: testClient, + }) + const closeAndFlushTimeout = ajs['_closeAndFlushDefaultTimeout'] + expect(closeAndFlushTimeout).toBe(120000 * 1.25) }) test('should force resolve if method call execution time exceeds specified timeout', async () => { diff --git a/packages/node/src/app/analytics-node.ts b/packages/node/src/app/analytics-node.ts index 4d45dfc0a..efa1e71f4 100644 --- a/packages/node/src/app/analytics-node.ts +++ b/packages/node/src/app/analytics-node.ts @@ -44,7 +44,7 @@ export class Analytics extends NodeEmitter implements CoreAnalytics { const flushInterval = settings.flushInterval ?? 10000 - this._closeAndFlushDefaultTimeout = flushInterval * 1.25 // add arbitrary multiplier in case an event is in a plugin. + this._closeAndFlushDefaultTimeout = Math.max(60000, flushInterval) * 1.25 const { plugin, publisher } = createConfiguredNodePlugin( { @@ -123,7 +123,11 @@ export class Analytics extends NodeEmitter implements CoreAnalytics { }).finally(() => { this._isFlushing = false }) - return timeout ? pTimeout(promise, timeout).catch(() => undefined) : promise + if (!timeout) return promise + + return pTimeout(promise, timeout).catch(() => { + this._publisher.abort() + }) } private _dispatch(segmentEvent: SegmentEvent, callback?: Callback) { diff --git a/packages/node/src/plugins/segmentio/publisher.ts b/packages/node/src/plugins/segmentio/publisher.ts index 9ced19734..6ec635ab6 100644 --- a/packages/node/src/plugins/segmentio/publisher.ts +++ b/packages/node/src/plugins/segmentio/publisher.ts @@ -13,8 +13,22 @@ import { b64encode } from '../../lib/base-64-encode' const MAX_RETRY_AFTER_SECONDS = 300 const MAX_RETRY_AFTER_RETRIES = 20 -function sleep(timeoutInMs: number): Promise { - return new Promise((resolve) => setTimeout(resolve, timeoutInMs)) +function sleep(timeoutInMs: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(signal.reason) + return + } + const timer = setTimeout(resolve, timeoutInMs) + signal?.addEventListener( + 'abort', + () => { + clearTimeout(timer) + reject(signal.reason) + }, + { once: true } + ) + }) } function noop() {} @@ -101,6 +115,8 @@ export class Publisher { private _rateLimitedUntil: number | undefined private _rateLimitStartTime: number | undefined + private _abortController: AbortController = new AbortController() + constructor( { host, @@ -143,6 +159,11 @@ export class Publisher { } } + abort(): void { + this._abortController.abort(new Error('Flush timeout')) + this._abortController = new AbortController() + } + private createBatch(): ContextBatch { this.pendingFlushTimeout && clearTimeout(this.pendingFlushTimeout) const batch = new ContextBatch(this._flushAt) @@ -308,6 +329,7 @@ export class Publisher { } const events = batch.getEvents() const maxRetries = this._maxRetries + const signal = this._abortController.signal let countedRetries = 0 let totalAttempts = 0 @@ -315,6 +337,11 @@ export class Publisher { // eslint-disable-next-line no-constant-condition while (true) { + if (signal.aborted) { + resolveFailedBatch(batch, signal.reason) + return + } + // Check rate-limit state before making a request const wasRateLimited = this._rateLimitStartTime !== undefined if (this._isRateLimited()) { @@ -331,7 +358,12 @@ export class Publisher { (Date.now() - this._rateLimitStartTime) ) const waitMs = Math.min(untilRetryAfter, untilDurationLimit) - await sleep(waitMs) + try { + await sleep(waitMs, signal) + } catch { + resolveFailedBatch(batch, signal.reason) + return + } continue } @@ -393,6 +425,14 @@ export class Publisher { const response = await this._httpClient.makeRequest(request) + this._emitter.emit('http_response', { + status: response.status, + statusText: response.statusText, + url: request.url, + body: request.body, + headers: convertHeaders(response.headers), + }) + // 2xx and 3xx are treated as successful delivery. if (response.status >= 200 && response.status < 400) { // Success — clear rate-limit state @@ -507,7 +547,12 @@ export class Publisher { }) : 0 - await sleep(delayMs) + try { + await sleep(delayMs, signal) + } catch { + resolveFailedBatch(batch, signal.reason) + return + } } } } From b4b8f1f67c87fa77af5c1320b970985070058de7 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Fri, 24 Apr 2026 17:06:50 -0400 Subject: [PATCH 2/6] feat(node): emit http_response event for observability Adds an http_response event to the NodeEmitter contract and emits it from the publisher after each API response. Enables consumers to observe status codes, headers, and timing without monkey-patching fetch. --- packages/node/src/app/emitter.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/packages/node/src/app/emitter.ts b/packages/node/src/app/emitter.ts index 0a577c1c5..708b49112 100644 --- a/packages/node/src/app/emitter.ts +++ b/packages/node/src/app/emitter.ts @@ -18,6 +18,15 @@ export type NodeEmitterEvents = CoreEmitterContract & { body: string } ] + http_response: [ + { + status: number + statusText: string + url: string + body: string + headers: Record + } + ] drained: [] } From 3ae0fd49b7565c7bd8b5b8732d049320034fea4c Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Fri, 24 Apr 2026 17:07:44 -0400 Subject: [PATCH 3/6] fix(e2e-cli): wait for rate-limited retries before exiting Node CLI: use http_request/http_response listeners for verbose logging. Browser CLI: track Retry-After from 429 responses and use it as the settle time, so the CLI doesn't exit while retries are still pending. Both CLIs now respect config.timeout for their wait durations. --- packages/browser/e2e-cli/src/cli.ts | 37 +++++++++---- packages/node/e2e-cli/src/cli.ts | 85 +++++++++++++++++++++++++++-- 2 files changed, 104 insertions(+), 18 deletions(-) diff --git a/packages/browser/e2e-cli/src/cli.ts b/packages/browser/e2e-cli/src/cli.ts index 1ebf029a1..5f18ec1f2 100644 --- a/packages/browser/e2e-cli/src/cli.ts +++ b/packages/browser/e2e-cli/src/cli.ts @@ -61,6 +61,7 @@ let lastApiResponseTime = 0 let inflightApiRequests = 0 let lastApiStatus = 0 let firstApiErrorStatus = 0 +let lastRetryAfterSeen = 0 let apiHostPattern = '' function installFetchMonitor(apiHost: string): void { @@ -97,6 +98,17 @@ function installFetchMonitor(apiHost: string): void { if (response.status >= 400 && firstApiErrorStatus === 0) { firstApiErrorStatus = response.status } + if (response.status === 429) { + const ra = response.headers.get('Retry-After') + if (ra) { + lastRetryAfterSeen = Math.max( + lastRetryAfterSeen, + parseInt(ra, 10) || 60 + ) + } else { + lastRetryAfterSeen = Math.max(lastRetryAfterSeen, 60) + } + } return response } catch (err) { lastApiResponseTime = Date.now() @@ -131,11 +143,15 @@ async function waitForDelivery(maxWaitMs = 60000): Promise { const elapsed = Date.now() - lastApiResponseTime // After success: brief settle for any remaining event dispatches. - // After error: longer settle to allow for retry scheduling + backoff. - // The fetch-dispatcher's core backoff uses minTimeout=500ms with - // exponential growth: attempt 5 reaches ~500*2^4*random ≈ 8-16s. - // Use 20s settle to accommodate higher retry attempts. - const settleMs = lastApiStatus < 400 ? 1500 : 20000 + // If we've ever seen a 429 with Retry-After, use that + buffer as settle + // time, since a 200 after a 429 doesn't mean all retries are done. + // Without any 429s, a short settle is fine. + const settleMs = + lastRetryAfterSeen > 0 + ? (lastRetryAfterSeen + 5) * 1000 + : lastApiStatus < 400 + ? 1500 + : 20000 if (elapsed >= settleMs) { return @@ -221,12 +237,8 @@ async function main(): Promise { const protocol = input.apiHost.startsWith('https') ? 'https' : 'http' segmentConfig.protocol = protocol - if (useBatching) { - segmentConfig.apiHost = input.apiHost - } else { - const apiHostStripped = input.apiHost.replace(/^https?:\/\//, '') - segmentConfig.apiHost = apiHostStripped + '/v1' - } + const apiHostStripped = input.apiHost.replace(/^https?:\/\//, '') + segmentConfig.apiHost = apiHostStripped + '/v1' } // Wire maxRetries and backoff timing through httpConfig — this controls @@ -292,7 +304,8 @@ async function main(): Promise { } // Wait for all delivery activity to settle - await waitForDelivery() + const maxWaitMs = (input.config?.timeout ?? 60) * 1000 + await waitForDelivery(maxWaitMs) // Determine success/failure from delivery errors (emitted by the // Segment.io plugin) and observed fetch responses as fallback. diff --git a/packages/node/e2e-cli/src/cli.ts b/packages/node/e2e-cli/src/cli.ts index 6cb86a7ce..4ba290545 100644 --- a/packages/node/e2e-cli/src/cli.ts +++ b/packages/node/e2e-cli/src/cli.ts @@ -46,12 +46,33 @@ interface CLIOutput { success: boolean error?: string sentBatches: number + eventResults: EventResult[] + httpLog: HttpLogEntry[] +} + +interface EventResult { + event: string + type: string + delivered: boolean + failureReason?: string +} + +interface HttpLogEntry { + timestamp: string + status: number + url: string + retryCount?: number + bodyPreview: string } function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)) } +function elapsed(start: number): string { + return `${((Date.now() - start) / 1000).toFixed(1)}s` +} + function sendEvent(analytics: Analytics, event: AnalyticsEvent): void { // SDK requires either userId or anonymousId const identity = event.userId @@ -117,7 +138,13 @@ function sendEvent(analytics: Analytics, event: AnalyticsEvent): void { } async function main(): Promise { - const output: CLIOutput = { success: false, sentBatches: 0 } + const startTime = Date.now() + const output: CLIOutput = { + success: false, + sentBatches: 0, + eventResults: [], + httpLog: [], + } try { // Parse --input argument @@ -147,36 +174,82 @@ async function main(): Promise { const msg = reason instanceof Error ? reason.message : String(reason ?? err.code) deliveryErrors.push(msg) + process.stderr.write(`[${elapsed(startTime)}] ERROR: ${msg}\n`) + }) + + analytics.on('http_request', (req) => { + const body = + typeof req.body === 'string' ? JSON.parse(req.body) : req.body + const events = body.batch?.map((e: any) => e.event ?? e.type) ?? [] + const retryHeader = req.headers?.['X-Retry-Count'] + process.stderr.write( + `[${elapsed(startTime)}] >> ${req.method} events=[${events.join(',')}]${ + retryHeader ? ` retry=${retryHeader}` : '' + }\n` + ) + }) + + analytics.on('http_response', (res) => { + const body = + typeof res.body === 'string' ? JSON.parse(res.body) : res.body + const events = body.batch?.map((e: any) => e.event ?? e.type) ?? [] + const retryAfter = res.headers?.['retry-after'] + process.stderr.write( + `[${elapsed(startTime)}] << ${res.status} events=[${events.join(',')}]${ + retryAfter ? ` retry-after=${retryAfter}` : '' + }\n` + ) }) // Process event sequences + let totalEvents = 0 for (const seq of sequences) { if (seq.delayMs > 0) { + process.stderr.write( + `[${elapsed(startTime)}] sleeping ${seq.delayMs}ms...\n` + ) await sleep(seq.delayMs) } for (const event of seq.events) { + totalEvents++ + const label = event.event ?? event.type + process.stderr.write( + `[${elapsed(startTime)}] enqueue #${totalEvents}: ${label}\n` + ) sendEvent(analytics, event) } } - // Flush and close — use a generous timeout so retries with exponential - // backoff have time to complete (default is flushInterval * 1.25) - const timeoutMs = (config.timeout ?? 60) * 1000 + process.stderr.write( + `[${elapsed( + startTime + )}] all ${totalEvents} events enqueued, calling closeAndFlush...\n` + ) + + // Flush and close + const timeoutMs = (config.timeout ?? 75) * 1000 await analytics.closeAndFlush({ timeout: timeoutMs }) + process.stderr.write( + `[${elapsed(startTime)}] closeAndFlush resolved. errors=${ + deliveryErrors.length + }\n` + ) + if (deliveryErrors.length > 0) { output.success = false output.error = deliveryErrors[0] } else { output.success = true - output.sentBatches = 1 + output.sentBatches = output.httpLog.length } } catch (err) { output.error = err instanceof Error ? err.message : String(err) + process.stderr.write(`[${elapsed(startTime)}] FATAL: ${output.error}\n`) } - console.log(JSON.stringify(output)) + console.log(JSON.stringify(output, null, 2)) process.exit(output.success ? 0 : 1) } From 342142dc45252867b47a2337ec06956ea9c5c116 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Mon, 27 Apr 2026 10:14:04 -0400 Subject: [PATCH 4/6] chore: add changeset for closeAndFlush fix --- .changeset/closeandflush-abort-fix.md | 9 +++++++++ 1 file changed, 9 insertions(+) create mode 100644 .changeset/closeandflush-abort-fix.md diff --git a/.changeset/closeandflush-abort-fix.md b/.changeset/closeandflush-abort-fix.md new file mode 100644 index 000000000..c902e1355 --- /dev/null +++ b/.changeset/closeandflush-abort-fix.md @@ -0,0 +1,9 @@ +--- +'@segment/analytics-node': patch +--- + +Fix closeAndFlush silently dropping in-flight events on timeout. + +- Cancel pending retry sleeps via AbortController when closeAndFlush times out, instead of silently swallowing the timeout error. +- Raise default closeAndFlush timeout floor to 75s (was 12.5s) so it survives at least one Retry-After: 60 cycle. +- Add `http_response` emitter event for observing API response status codes and headers. From 926efbf660311be7a25fdaddacb0f1e0ae11453f Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Mon, 27 Apr 2026 11:33:03 -0400 Subject: [PATCH 5/6] fix(e2e-cli): respect actual Retry-After value in settle timer The previous logic forced a 60s minimum when Retry-After was 0 or missing, causing e2e tests with short/zero Retry-After to time out. Now uses the actual header value (default 10s when absent), so Retry-After: 0 settles quickly while Retry-After: 60 waits properly. --- packages/browser/e2e-cli/src/cli.ts | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/packages/browser/e2e-cli/src/cli.ts b/packages/browser/e2e-cli/src/cli.ts index 5f18ec1f2..2373f7b2c 100644 --- a/packages/browser/e2e-cli/src/cli.ts +++ b/packages/browser/e2e-cli/src/cli.ts @@ -100,14 +100,9 @@ function installFetchMonitor(apiHost: string): void { } if (response.status === 429) { const ra = response.headers.get('Retry-After') - if (ra) { - lastRetryAfterSeen = Math.max( - lastRetryAfterSeen, - parseInt(ra, 10) || 60 - ) - } else { - lastRetryAfterSeen = Math.max(lastRetryAfterSeen, 60) - } + const parsed = ra !== null ? parseInt(ra, 10) : NaN + const retryAfterSeconds = Number.isFinite(parsed) ? parsed : 10 + lastRetryAfterSeen = Math.max(lastRetryAfterSeen, retryAfterSeconds) } return response } catch (err) { From 1e85dea112d1e7dde1becf4f2c6ef6cb38658a70 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Mon, 27 Apr 2026 11:58:24 -0400 Subject: [PATCH 6/6] fix(test): register error listener before closeAndFlush in OAuth test With the new 75s closeAndFlush timeout floor, the "handles a bad key" test would exceed the 30s jest timeout. Pass an explicit 20s timeout and register resolveCtx before closeAndFlush so the error listener catches the delivery failure event when abort fires. --- .../src/__tests__/oauth.integration.test.ts | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/packages/node/src/__tests__/oauth.integration.test.ts b/packages/node/src/__tests__/oauth.integration.test.ts index a74aa8ffd..1716ae0ba 100644 --- a/packages/node/src/__tests__/oauth.integration.test.ts +++ b/packages/node/src/__tests__/oauth.integration.test.ts @@ -246,16 +246,16 @@ describe('OAuth Failure', () => { maxRetries: 3, }) + analytics.track({ + event: 'Test Event', + anonymousId: 'unknown', + userId: 'known', + timestamp: timestamp, + }) + const ctxPromise = resolveCtx(analytics, 'track') + await analytics.closeAndFlush({ timeout: 20000 }) try { - analytics.track({ - event: 'Test Event', - anonymousId: 'unknown', - userId: 'known', - timestamp: timestamp, - }) - await analytics.closeAndFlush() - const ctx1 = await resolveCtx(analytics, 'track') // forces exception to be thrown - expect(ctx1.event.type).toEqual('track') + await ctxPromise throw new Error('fail') } catch (err: any) { expect(err.reason).toEqual(