diff --git a/packages/core/src/__tests__/api.test.ts b/packages/core/src/__tests__/api.test.ts index 86a378752..485b105fa 100644 --- a/packages/core/src/__tests__/api.test.ts +++ b/packages/core/src/__tests__/api.test.ts @@ -25,7 +25,11 @@ describe('#sendEvents', () => { .mockReturnValue('2001-01-01T00:00:00.000Z'); }); - async function sendAnEventPer(writeKey: string, toUrl: string) { + async function sendAnEventPer( + writeKey: string, + toUrl: string, + retryCount?: number + ) { const mockResponse = Promise.resolve('MANOS'); // eslint-disable-next-line @typescript-eslint/ban-ts-comment // @ts-ignore @@ -60,9 +64,19 @@ describe('#sendEvents', () => { writeKey: writeKey, url: toUrl, events: [event], + retryCount, }); - expect(fetch).toHaveBeenCalledWith(toUrl, { + return event; + } + + it('sends an event', async () => { + const toSegmentBatchApi = 'https://api.segment.io/v1.b'; + const writeKey = 'SEGMENT_KEY'; + + const event = await sendAnEventPer(writeKey, toSegmentBatchApi); + + expect(fetch).toHaveBeenCalledWith(toSegmentBatchApi, { method: 'POST', body: JSON.stringify({ batch: [event], @@ -71,21 +85,68 @@ describe('#sendEvents', () => { }), headers: { 'Content-Type': 'application/json; charset=utf-8', + 'X-Retry-Count': '0', }, + keepalive: true, }); - } - - it('sends an event', async () => { - const toSegmentBatchApi = 'https://api.segment.io/v1.b'; - const writeKey = 'SEGMENT_KEY'; - - await sendAnEventPer(writeKey, toSegmentBatchApi); }); it('sends an event to proxy', async () => { const toProxyUrl = 'https://myprox.io/b'; const writeKey = 'SEGMENT_KEY'; - await sendAnEventPer(writeKey, toProxyUrl); + const event = await sendAnEventPer(writeKey, toProxyUrl); + + expect(fetch).toHaveBeenCalledWith(toProxyUrl, { + method: 'POST', + body: JSON.stringify({ + batch: [event], + sentAt: '2001-01-01T00:00:00.000Z', + writeKey: 'SEGMENT_KEY', + }), + headers: { + 'Content-Type': 'application/json; charset=utf-8', + 'X-Retry-Count': '0', + }, + keepalive: true, + }); + }); + + it('sends X-Retry-Count header with default value 0', async () => { + const url = 'https://api.segment.io/v1.b'; + await sendAnEventPer('KEY', url); + + expect(fetch).toHaveBeenCalledWith( + url, + expect.objectContaining({ + headers: expect.objectContaining({ + 'X-Retry-Count': '0', + }), + }) + ); + }); + + it('sends X-Retry-Count header with provided retry count', async () => { + const url = 'https://api.segment.io/v1.b'; + await sendAnEventPer('KEY', url, 5); + + expect(fetch).toHaveBeenCalledWith( + url, + expect.objectContaining({ + headers: expect.objectContaining({ + 'X-Retry-Count': '5', + }), + }) + ); + }); + + it('sends X-Retry-Count as string format', async () => { + const url = 'https://api.segment.io/v1.b'; + await sendAnEventPer('KEY', url, 42); + + const callArgs = (fetch as jest.Mock).mock.calls[0]; + const headers = callArgs[1].headers; + expect(typeof headers['X-Retry-Count']).toBe('string'); + expect(headers['X-Retry-Count']).toBe('42'); }); }); diff --git a/packages/core/src/api.ts b/packages/core/src/api.ts index 6aa2851a7..8853234e7 100644 --- a/packages/core/src/api.ts +++ b/packages/core/src/api.ts @@ -4,10 +4,12 @@ export const uploadEvents = async ({ writeKey, url, events, + retryCount = 0, }: { writeKey: string; url: string; events: SegmentEvent[]; + retryCount?: number; }) => { return await fetch(url, { method: 'POST', @@ -19,6 +21,7 @@ export const uploadEvents = async ({ }), headers: { 'Content-Type': 'application/json; charset=utf-8', + 'X-Retry-Count': retryCount.toString(), }, }); }; diff --git a/packages/core/src/plugins/QueueFlushingPlugin.ts b/packages/core/src/plugins/QueueFlushingPlugin.ts index 1580ee288..19e6df24a 100644 --- a/packages/core/src/plugins/QueueFlushingPlugin.ts +++ b/packages/core/src/plugins/QueueFlushingPlugin.ts @@ -15,12 +15,12 @@ export class QueueFlushingPlugin extends UtilityPlugin { type = PluginType.after; private storeKey: string; - private isPendingUpload = false; private queueStore: Store<{ events: SegmentEvent[] }> | undefined; private onFlush: (events: SegmentEvent[]) => Promise; private isRestoredResolve: () => void; private isRestored: Promise; private timeoutWarned = false; + private flushPromise?: Promise; /** * @param onFlush callback to execute when the queue is flushed (either by reaching the limit or manually) e.g. code to upload events to your destination @@ -63,16 +63,36 @@ export class QueueFlushingPlugin extends UtilityPlugin { async execute(event: SegmentEvent): Promise { await this.queueStore?.dispatch((state) => { - const events = [...state.events, event]; + const stampedEvent = { ...event, _queuedAt: Date.now() }; + const events = [...state.events, stampedEvent]; return { events }; }); return event; } /** - * Calls the onFlush callback with the events in the queue + * Calls the onFlush callback with the events in the queue. + * Ensures only one flush operation runs at a time. */ async flush() { + // Safety: prevent concurrent flush operations + if (this.flushPromise) { + this.analytics?.logger.info( + 'Flush already in progress, waiting for completion' + ); + await this.flushPromise; + return; + } + + this.flushPromise = this._doFlush(); + try { + await this.flushPromise; + } finally { + this.flushPromise = undefined; + } + } + + private async _doFlush(): Promise { // Wait for the queue to be restored try { await this.isRestored; @@ -103,14 +123,7 @@ export class QueueFlushingPlugin extends UtilityPlugin { } const events = (await this.queueStore?.getState(true))?.events ?? []; - if (!this.isPendingUpload) { - try { - this.isPendingUpload = true; - await this.onFlush(events); - } finally { - this.isPendingUpload = false; - } - } + await this.onFlush(events); } /** @@ -130,6 +143,26 @@ export class QueueFlushingPlugin extends UtilityPlugin { return { events: filteredEvents }; }); } + + /** + * Removes events from the queue by their messageId + * @param messageIds array of messageId strings to remove + */ + async dequeueByMessageIds(messageIds: string[]): Promise { + await this.queueStore?.dispatch((state) => { + if (messageIds.length === 0 || state.events.length === 0) { + return state; + } + + const idsToRemove = new Set(messageIds); + const filteredEvents = state.events.filter( + (e) => e.messageId == null || !idsToRemove.has(e.messageId) + ); + + return { events: filteredEvents }; + }); + } + /** * Clear all events from the queue */ diff --git a/packages/core/src/plugins/SegmentDestination.ts b/packages/core/src/plugins/SegmentDestination.ts index cc7e911e6..0807f80b7 100644 --- a/packages/core/src/plugins/SegmentDestination.ts +++ b/packages/core/src/plugins/SegmentDestination.ts @@ -1,5 +1,6 @@ import { DestinationPlugin } from '../plugin'; import { + HttpConfig, PluginType, SegmentAPIIntegration, SegmentAPISettings, @@ -11,20 +12,44 @@ import { uploadEvents } from '../api'; import type { SegmentClient } from '../analytics'; import { DestinationMetadataEnrichment } from './DestinationMetadataEnrichment'; import { QueueFlushingPlugin } from './QueueFlushingPlugin'; -import { defaultApiHost } from '../constants'; -import { checkResponseForErrors, translateHTTPError } from '../errors'; -import { defaultConfig } from '../constants'; +import { defaultApiHost, defaultConfig } from '../constants'; +import { translateHTTPError, classifyError, parseRetryAfter } from '../errors'; +import { RetryManager } from '../backoff/RetryManager'; const MAX_EVENTS_PER_BATCH = 100; const MAX_PAYLOAD_SIZE_IN_KB = 500; export const SEGMENT_DESTINATION_KEY = 'Segment.io'; +/** + * Result of uploading a single batch + */ +type BatchResult = { + batch: SegmentEvent[]; + messageIds: string[]; + status: 'success' | '429' | 'transient' | 'permanent' | 'network_error'; + statusCode?: number; + retryAfterSeconds?: number; +}; + +/** + * Aggregated error information from parallel batch uploads + */ +type ErrorAggregation = { + successfulMessageIds: string[]; + has429: boolean; + longestRetryAfter: number; + hasTransientError: boolean; + permanentErrorMessageIds: string[]; +}; + export class SegmentDestination extends DestinationPlugin { type = PluginType.destination; key = SEGMENT_DESTINATION_KEY; private apiHost?: string; + private httpConfig?: HttpConfig; private settingsResolve: () => void; private settingsPromise: Promise; + private retryManager?: RetryManager; constructor() { super(); @@ -34,6 +59,144 @@ export class SegmentDestination extends DestinationPlugin { this.settingsResolve = resolve; } + /** + * Upload a single batch and return structured result + */ + private async uploadBatch(batch: SegmentEvent[]): Promise { + const config = this.analytics?.getConfig() ?? defaultConfig; + const messageIds = batch + .map((e) => e.messageId) + .filter((id): id is string => id !== undefined && id !== ''); + + const retryCount = this.retryManager + ? await this.retryManager.getRetryCount() + : 0; + + // Strip internal metadata before sending upstream + const cleanedBatch = batch.map(({ _queuedAt, ...event }) => event); + + try { + const res = await uploadEvents({ + writeKey: config.writeKey, + url: this.getEndpoint(), + events: cleanedBatch as SegmentEvent[], + retryCount, + }); + + if (res.status === 200) { + return { + batch, + messageIds, + status: 'success', + statusCode: 200, + }; + } + + // Parse retry-after for 429 + const retryAfterSeconds = + res.status === 429 + ? parseRetryAfter( + res.headers.get('Retry-After'), + this.httpConfig?.rateLimitConfig?.maxRetryInterval + ) + : undefined; + + // Classify error + const classification = classifyError(res.status, { + default4xxBehavior: this.httpConfig?.backoffConfig?.default4xxBehavior, + default5xxBehavior: this.httpConfig?.backoffConfig?.default5xxBehavior, + statusCodeOverrides: + this.httpConfig?.backoffConfig?.statusCodeOverrides, + rateLimitEnabled: this.httpConfig?.rateLimitConfig?.enabled, + }); + + if (classification.errorType === 'rate_limit') { + return { + batch, + messageIds, + status: '429', + statusCode: res.status, + retryAfterSeconds: + retryAfterSeconds !== undefined && retryAfterSeconds > 0 + ? retryAfterSeconds + : 60, // Default to 60s if not provided + }; + } else if (classification.errorType === 'transient') { + return { + batch, + messageIds, + status: 'transient', + statusCode: res.status, + }; + } else { + // Permanent error + return { + batch, + messageIds, + status: 'permanent', + statusCode: res.status, + }; + } + } catch (e) { + // Network error + this.analytics?.reportInternalError(translateHTTPError(e)); + return { + batch, + messageIds, + status: 'network_error', + }; + } + } + + /** + * Aggregate errors from parallel batch results + */ + private aggregateErrors(results: BatchResult[]): ErrorAggregation { + const aggregation: ErrorAggregation = { + successfulMessageIds: [], + has429: false, + longestRetryAfter: 0, + hasTransientError: false, + permanentErrorMessageIds: [], + }; + + for (const result of results) { + switch (result.status) { + case 'success': + aggregation.successfulMessageIds.push(...result.messageIds); + break; + + case '429': + aggregation.has429 = true; + if ( + result.retryAfterSeconds !== undefined && + result.retryAfterSeconds > 0 + ) { + aggregation.longestRetryAfter = Math.max( + aggregation.longestRetryAfter, + result.retryAfterSeconds + ); + } + break; + + case 'transient': + aggregation.hasTransientError = true; + break; + + case 'permanent': + aggregation.permanentErrorMessageIds.push(...result.messageIds); + break; + + case 'network_error': + // Treat as transient + aggregation.hasTransientError = true; + break; + } + } + + return aggregation; + } + private sendEvents = async (events: SegmentEvent[]): Promise => { if (events.length === 0) { return Promise.resolve(); @@ -44,43 +207,114 @@ export class SegmentDestination extends DestinationPlugin { const config = this.analytics?.getConfig() ?? defaultConfig; - const chunkedEvents: SegmentEvent[][] = chunk( + // Prune events that have exceeded maxTotalBackoffDuration + const maxAge = this.httpConfig?.backoffConfig?.maxTotalBackoffDuration ?? 0; + if (maxAge > 0) { + const now = Date.now(); + const maxAgeMs = maxAge * 1000; + const expiredMessageIds: string[] = []; + const freshEvents: SegmentEvent[] = []; + + for (const event of events) { + if (event._queuedAt !== undefined && now - event._queuedAt > maxAgeMs) { + if (event.messageId !== undefined && event.messageId !== '') { + expiredMessageIds.push(event.messageId); + } + } else { + freshEvents.push(event); + } + } + + if (expiredMessageIds.length > 0) { + await this.queuePlugin.dequeueByMessageIds(expiredMessageIds); + this.analytics?.logger.warn( + `Pruned ${expiredMessageIds.length} events older than ${maxAge}s` + ); + } + + events = freshEvents; + + if (events.length === 0) { + return Promise.resolve(); + } + } + + // Check if blocked by rate limit or backoff + if (this.retryManager) { + const canRetry = await this.retryManager.canRetry(); + if (!canRetry) { + this.analytics?.logger.info('Upload blocked by retry manager'); + return; + } + } + + // Chunk events into batches + const batches: SegmentEvent[][] = chunk( events, config.maxBatchSize ?? MAX_EVENTS_PER_BATCH, MAX_PAYLOAD_SIZE_IN_KB ); - let sentEvents: SegmentEvent[] = []; - let numFailedEvents = 0; - - await Promise.all( - chunkedEvents.map(async (batch: SegmentEvent[]) => { - try { - const res = await uploadEvents({ - writeKey: config.writeKey, - url: this.getEndpoint(), - events: batch, - }); - checkResponseForErrors(res); - sentEvents = sentEvents.concat(batch); - } catch (e) { - this.analytics?.reportInternalError(translateHTTPError(e)); - this.analytics?.logger.warn(e); - numFailedEvents += batch.length; - } finally { - await this.queuePlugin.dequeue(sentEvents); - } - }) + // Upload all batches in parallel + const results: BatchResult[] = await Promise.all( + batches.map((batch) => this.uploadBatch(batch)) ); - if (sentEvents.length) { + // Aggregate errors + const aggregation = this.aggregateErrors(results); + + // Handle 429 - ONCE per flush with longest retry-after + if (aggregation.has429 && this.retryManager) { + await this.retryManager.handle429(aggregation.longestRetryAfter); + this.analytics?.logger.warn( + `Rate limited (429): waiting ${aggregation.longestRetryAfter}s before retry` + ); + // Events stay in queue + } + + // Handle transient errors - ONCE per flush + if (aggregation.hasTransientError && this.retryManager) { + await this.retryManager.handleTransientError(); + // Events stay in queue + } + + // Handle successes - dequeue + if (aggregation.successfulMessageIds.length > 0) { + await this.queuePlugin.dequeueByMessageIds( + aggregation.successfulMessageIds + ); + + // Reset retry manager on success + if (this.retryManager) { + await this.retryManager.reset(); + } + if (config.debug === true) { - this.analytics?.logger.info(`Sent ${sentEvents.length} events`); + this.analytics?.logger.info( + `Sent ${aggregation.successfulMessageIds.length} events` + ); } } - if (numFailedEvents) { - this.analytics?.logger.error(`Failed to send ${numFailedEvents} events.`); + // Handle permanent errors - dequeue (drop) + if (aggregation.permanentErrorMessageIds.length > 0) { + await this.queuePlugin.dequeueByMessageIds( + aggregation.permanentErrorMessageIds + ); + this.analytics?.logger.error( + `Dropped ${aggregation.permanentErrorMessageIds.length} events due to permanent errors` + ); + } + + // Log summary + const failedCount = + events.length - + aggregation.successfulMessageIds.length - + aggregation.permanentErrorMessageIds.length; + if (failedCount > 0) { + this.analytics?.logger.warn( + `${failedCount} events will retry (429: ${aggregation.has429}, transient: ${aggregation.hasTransientError})` + ); } return Promise.resolve(); @@ -95,7 +329,7 @@ export class SegmentDestination extends DestinationPlugin { let baseURL = ''; let endpoint = ''; if (hasProxy) { - //baseURL is always config?.proxy if hasProxy + //baseURL is always config?.proxy if hasProxy baseURL = config?.proxy ?? ''; if (useSegmentEndpoints) { const isProxyEndsWithSlash = baseURL.endsWith('/'); @@ -111,12 +345,15 @@ export class SegmentDestination extends DestinationPlugin { return defaultApiHost; } } + configure(analytics: SegmentClient): void { super.configure(analytics); + const config = analytics.getConfig(); + // If the client has a proxy we don't need to await for settings apiHost, we can send events directly // Important! If new settings are required in the future you probably want to change this! - if (analytics.getConfig().proxy !== undefined) { + if (config.proxy !== undefined) { this.settingsResolve(); } @@ -137,6 +374,34 @@ export class SegmentDestination extends DestinationPlugin { //assign the api host from segment settings (domain/v1) this.apiHost = `https://${segmentSettings.apiHost}/b`; } + + // Initialize httpConfig and retry manager from server-side CDN config + const httpConfig = this.analytics?.getHttpConfig(); + if (httpConfig) { + this.httpConfig = httpConfig; + + if ( + !this.retryManager && + (httpConfig.rateLimitConfig || httpConfig.backoffConfig) + ) { + const config = this.analytics?.getConfig(); + this.retryManager = new RetryManager( + config?.writeKey ?? '', + config?.storePersistor, + httpConfig.rateLimitConfig, + httpConfig.backoffConfig, + this.analytics?.logger, + config?.retryStrategy ?? 'lazy' + ); + + if (config?.autoFlushOnRetryReady === true) { + this.retryManager.setAutoFlushCallback(() => { + void this.flush(); + }); + } + } + } + this.settingsResolve(); } diff --git a/packages/core/src/plugins/__tests__/QueueFlushingPlugin.test.ts b/packages/core/src/plugins/__tests__/QueueFlushingPlugin.test.ts index b3b02d802..1a483f73d 100644 --- a/packages/core/src/plugins/__tests__/QueueFlushingPlugin.test.ts +++ b/packages/core/src/plugins/__tests__/QueueFlushingPlugin.test.ts @@ -60,6 +60,7 @@ describe('QueueFlushingPlugin', () => { const event: SegmentEvent = { type: EventType.TrackEvent, event: 'test2', + messageId: 'msg-dequeue-1', properties: { test: 'test2', }, @@ -72,7 +73,7 @@ describe('QueueFlushingPlugin', () => { // eslint-disable-next-line @typescript-eslint/ban-ts-comment // @ts-ignore expect(queuePlugin.queueStore?.getState().events).toHaveLength(1); - await queuePlugin.dequeue(event); + await queuePlugin.dequeueByMessageIds(['msg-dequeue-1']); // eslint-disable-next-line @typescript-eslint/ban-ts-comment // @ts-ignore expect(queuePlugin.queueStore?.getState().events).toHaveLength(0); @@ -111,6 +112,7 @@ describe('QueueFlushingPlugin', () => { const event1: SegmentEvent = { type: EventType.TrackEvent, event: 'test1', + messageId: 'msg-count-1', properties: { test: 'test1', }, @@ -119,6 +121,7 @@ describe('QueueFlushingPlugin', () => { const event2: SegmentEvent = { type: EventType.TrackEvent, event: 'test2', + messageId: 'msg-count-2', properties: { test: 'test2', }, @@ -130,7 +133,7 @@ describe('QueueFlushingPlugin', () => { let eventsCount = await queuePlugin.pendingEvents(); expect(eventsCount).toBe(2); - await queuePlugin.dequeue(event1); + await queuePlugin.dequeueByMessageIds(['msg-count-1']); eventsCount = await queuePlugin.pendingEvents(); expect(eventsCount).toBe(1); diff --git a/packages/core/src/plugins/__tests__/SegmentDestination.test.ts b/packages/core/src/plugins/__tests__/SegmentDestination.test.ts index 885097fd9..23f4d410c 100644 --- a/packages/core/src/plugins/__tests__/SegmentDestination.test.ts +++ b/packages/core/src/plugins/__tests__/SegmentDestination.test.ts @@ -9,6 +9,7 @@ import { import { Config, EventType, + HttpConfig, SegmentAPIIntegration, SegmentEvent, TrackEventType, @@ -260,10 +261,12 @@ describe('SegmentDestination', () => { config, settings, events, + httpConfig, }: { config?: Config; settings?: SegmentAPIIntegration; events: SegmentEvent[]; + httpConfig?: HttpConfig; }) => { const plugin = new SegmentDestination(); @@ -278,6 +281,13 @@ describe('SegmentDestination', () => { }); plugin.configure(analytics); + + if (httpConfig !== undefined) { + // eslint-disable-next-line @typescript-eslint/ban-ts-comment + // @ts-ignore + plugin.httpConfig = httpConfig; + } + // The settings store won't match but that's ok, the plugin should rely only on the settings it receives during update plugin.update( { @@ -322,6 +332,7 @@ describe('SegmentDestination', () => { expect(sendEventsSpy).toHaveBeenCalledWith({ url: getURL(defaultApiHost, ''), // default api already appended with '/b' writeKey: '123-456', + retryCount: 0, events: events.slice(0, 2).map((e) => ({ ...e, })), @@ -329,6 +340,7 @@ describe('SegmentDestination', () => { expect(sendEventsSpy).toHaveBeenCalledWith({ url: getURL(defaultApiHost, ''), // default api already appended with '/b' writeKey: '123-456', + retryCount: 0, events: events.slice(2, 4).map((e) => ({ ...e, })), @@ -356,6 +368,7 @@ describe('SegmentDestination', () => { expect(sendEventsSpy).toHaveBeenCalledWith({ url: getURL(customEndpoint, '/b'), writeKey: '123-456', + retryCount: 0, events: events.slice(0, 2).map((e) => ({ ...e, })), @@ -407,13 +420,129 @@ describe('SegmentDestination', () => { expect(sendEventsSpy).toHaveBeenCalledWith({ url: expectedUrl, writeKey: '123-456', + retryCount: 0, events: events.map((e) => ({ ...e, })), }); } ); + + describe('event age pruning', () => { + it('prunes events older than maxTotalBackoffDuration', async () => { + const now = Date.now(); + const events = [ + { messageId: 'old-1', _queuedAt: now - 50000 * 1000 }, + { messageId: 'fresh-1' }, + { messageId: 'fresh-2' }, + ] as SegmentEvent[]; + + const { plugin, sendEventsSpy } = createTestWith({ + events, + httpConfig: { + backoffConfig: { + enabled: true, + maxRetryCount: 100, + baseBackoffInterval: 0.5, + maxBackoffInterval: 300, + maxTotalBackoffDuration: 43200, + jitterPercent: 10, + default4xxBehavior: 'drop', + default5xxBehavior: 'retry', + statusCodeOverrides: {}, + }, + }, + }); + + await plugin.flush(); + + expect(sendEventsSpy).toHaveBeenCalledTimes(1); + const sentEvents = sendEventsSpy.mock.calls[0][0].events; + expect(sentEvents).toHaveLength(2); + expect(sentEvents.map((e: SegmentEvent) => e.messageId)).toEqual([ + 'fresh-1', + 'fresh-2', + ]); + }); + + it('does not prune when maxTotalBackoffDuration is 0', async () => { + const now = Date.now(); + const events = [ + { messageId: 'old-1', _queuedAt: now - 50000 * 1000 }, + { messageId: 'fresh-1' }, + ] as SegmentEvent[]; + + const { plugin, sendEventsSpy } = createTestWith({ + events, + httpConfig: { + backoffConfig: { + enabled: true, + maxRetryCount: 100, + baseBackoffInterval: 0.5, + maxBackoffInterval: 300, + maxTotalBackoffDuration: 0, + jitterPercent: 10, + default4xxBehavior: 'drop', + default5xxBehavior: 'retry', + statusCodeOverrides: {}, + }, + }, + }); + + await plugin.flush(); + + expect(sendEventsSpy).toHaveBeenCalledTimes(1); + const sentEvents = sendEventsSpy.mock.calls[0][0].events; + expect(sentEvents).toHaveLength(2); + }); + + it('does not prune events without _queuedAt', async () => { + const events = [ + { messageId: 'old-1' }, + { messageId: 'fresh-1' }, + ] as SegmentEvent[]; + + const { plugin, sendEventsSpy } = createTestWith({ + events, + httpConfig: { + backoffConfig: { + enabled: true, + maxRetryCount: 100, + baseBackoffInterval: 0.5, + maxBackoffInterval: 300, + maxTotalBackoffDuration: 43200, + jitterPercent: 10, + default4xxBehavior: 'drop', + default5xxBehavior: 'retry', + statusCodeOverrides: {}, + }, + }, + }); + + await plugin.flush(); + + expect(sendEventsSpy).toHaveBeenCalledTimes(1); + const sentEvents = sendEventsSpy.mock.calls[0][0].events; + expect(sentEvents).toHaveLength(2); + }); + + it('strips _queuedAt before upload', async () => { + const now = Date.now(); + const events = [ + { messageId: 'msg-1', _queuedAt: now - 1000 }, + ] as SegmentEvent[]; + + const { plugin, sendEventsSpy } = createTestWith({ events }); + + await plugin.flush(); + + expect(sendEventsSpy).toHaveBeenCalledTimes(1); + const sentEvents = sendEventsSpy.mock.calls[0][0].events; + expect(sentEvents[0]).not.toHaveProperty('_queuedAt'); + }); + }); }); + describe('getEndpoint', () => { it.each([ ['example.com/v1/', 'https://example.com/v1/'],