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. diff --git a/packages/browser/e2e-cli/src/cli.ts b/packages/browser/e2e-cli/src/cli.ts index 1ebf029a1..2373f7b2c 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,12 @@ 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') + const parsed = ra !== null ? parseInt(ra, 10) : NaN + const retryAfterSeconds = Number.isFinite(parsed) ? parsed : 10 + lastRetryAfterSeen = Math.max(lastRetryAfterSeen, retryAfterSeconds) + } return response } catch (err) { lastApiResponseTime = Date.now() @@ -131,11 +138,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 +232,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 +299,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) } 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/__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( 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/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: [] } 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 + } } } }