fix(node): prevent closeAndFlush from dropping in-flight events (#1365)

Author: MichaelGHSeg

.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.

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<void> {
 
     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<void> {
       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<void> {
     }
 
     // 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.

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<void> {
   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<void> {
-  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<void> {
       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)
 }
 

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 () => {

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(

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) {

packages/node/src/app/emitter.ts

@@ -18,6 +18,15 @@ export type NodeEmitterEvents = CoreEmitterContract<Context> & {
       body: string
     }
   ]
+  http_response: [
+    {
+      status: number
+      statusText: string
+      url: string
+      body: string
+      headers: Record<string, string>
+    }
+  ]
   drained: []
 }