feat(events): Redeliver subset of failed events from a batch (#176)

* wip

* feat(events): Redeliver subset of failed events from a batch

* improve retries

Author: felipecsl

src/events/batch-event-processor.spec.ts

@@ -1,6 +1,6 @@
 import ArrayBackedNamedEventQueue from './array-backed-named-event-queue';
 import BatchEventProcessor from './batch-event-processor';
-import { Event } from './event-dispatcher';
+import Event from './event';
 
 describe('BatchEventProcessor', () => {
   describe('nextBatch', () => {

src/events/batch-event-processor.ts

@@ -1,4 +1,4 @@
-import { Event } from './event-dispatcher';
+import Event from './event';
 import NamedEventQueue from './named-event-queue';
 
 export default class BatchEventProcessor {

src/events/batch-retry-manager.spec.ts

@@ -0,0 +1,77 @@
+import BatchRetryManager from './batch-retry-manager';
+import Event from './event';
+
+describe('BatchRetryManager', () => {
+  const mockDelivery = { deliver: jest.fn() };
+  const retryIntervalMs = 100;
+  const maxRetryDelayMs = 1000;
+  const maxRetries = 3;
+  const mockConfig = { retryIntervalMs, maxRetryDelayMs, maxRetries };
+
+  let batchRetryManager: BatchRetryManager;
+  let mockBatch: Event[];
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    batchRetryManager = new BatchRetryManager(mockDelivery, mockConfig);
+    mockBatch = [{ uuid: 'event1' }, { uuid: 'event2' }] as Event[];
+  });
+
+  it('should successfully retry and deliver a batch with no failures', async () => {
+    mockDelivery.deliver.mockResolvedValueOnce({ failedEvents: [] });
+    const result = await batchRetryManager.retry(mockBatch);
+    expect(result).toEqual([]);
+    expect(mockDelivery.deliver).toHaveBeenCalledTimes(1);
+    expect(mockDelivery.deliver).toHaveBeenCalledWith(mockBatch);
+  });
+
+  it('should retry failed deliveries up to maxRetries times and return last failed batch', async () => {
+    mockDelivery.deliver.mockResolvedValue({ failedEvents: [{ id: 'event1' }] });
+    const result = await batchRetryManager.retry(mockBatch);
+    expect(result).toEqual([{ id: 'event1' }]);
+    expect(mockDelivery.deliver).toHaveBeenCalledTimes(maxRetries);
+  });
+
+  it('should exponentially delay retries up to maxRetryDelayMs', async () => {
+    mockDelivery.deliver
+      .mockResolvedValueOnce({ failedEvents: [{ id: 'event1' }] })
+      .mockResolvedValueOnce({ failedEvents: [{ id: 'event1' }] })
+      .mockResolvedValueOnce({ failedEvents: [] });
+
+    jest.useFakeTimers();
+
+    const retryPromise = batchRetryManager.retry(mockBatch);
+
+    // 1st retry: 100ms
+    // 2nd retry: 200ms
+    // 3rd retry: 400ms
+    await jest.advanceTimersByTimeAsync(100 + 200 + 400);
+
+    const result = await retryPromise;
+    expect(result).toEqual([]);
+    expect(mockDelivery.deliver).toHaveBeenCalledTimes(3);
+
+    jest.useRealTimers();
+  });
+
+  it('should not exceed maxRetryDelayMs for delays', async () => {
+    batchRetryManager = new BatchRetryManager(mockDelivery, {
+      ...mockConfig,
+      maxRetryDelayMs: 300,
+    });
+    mockDelivery.deliver
+      .mockResolvedValueOnce({ failedEvents: mockBatch })
+      .mockResolvedValueOnce({ failedEvents: mockBatch })
+      .mockResolvedValueOnce({ failedEvents: mockBatch })
+      .mockResolvedValueOnce({ failedEvents: mockBatch });
+
+    jest.useFakeTimers();
+
+    const retryPromise = batchRetryManager.retry(mockBatch);
+    // 100ms + 200ms + 300ms (maxRetryDelayMs) = 600ms
+    await jest.advanceTimersByTimeAsync(600);
+    const result = await retryPromise;
+    expect(result).toEqual(mockBatch);
+    jest.useRealTimers();
+  });
+});

src/events/batch-retry-manager.ts

@@ -1,7 +1,11 @@
 import { logger } from '../application-logger';
 
-import EventDelivery from './event-delivery';
-import { Event } from './event-dispatcher';
+import Event from './event';
+import { EventDeliveryResult } from './event-delivery';
+
+export type IEventDelivery = {
+  deliver(batch: Event[]): Promise<EventDeliveryResult>;
+};
 
 /**
  * Attempts to retry delivering a batch of events to the ingestionUrl up to `maxRetries` times
@@ -14,34 +18,34 @@ export default class BatchRetryManager {
    * @param config.maxRetries - The maximum number of retries
    */
   constructor(
-    private readonly delivery: EventDelivery,
+    private readonly delivery: IEventDelivery,
     private readonly config: {
       retryIntervalMs: number;
       maxRetryDelayMs: number;
       maxRetries: number;
     },
   ) {}
 
-  /** Re-attempts delivery of the provided batch, returns whether the retry succeeded. */
-  async retry(batch: Event[], attempt = 0): Promise<boolean> {
+  /** Re-attempts delivery of the provided batch, returns the UUIDs of events that failed retry. */
+  async retry(batch: Event[], attempt = 0): Promise<Event[]> {
     const { retryIntervalMs, maxRetryDelayMs, maxRetries } = this.config;
     const delay = Math.min(retryIntervalMs * Math.pow(2, attempt), maxRetryDelayMs);
-    logger.info(`[BatchRetryManager] Retrying batch delivery in ${delay}ms...`);
+    logger.info(
+      `[BatchRetryManager] Retrying batch delivery of ${batch.length} events in ${delay}ms...`,
+    );
     await new Promise((resolve) => setTimeout(resolve, delay));
 
-    const success = await this.delivery.deliver(batch);
-    if (success) {
-      logger.info(`[BatchRetryManager] Batch delivery successfully after ${attempt} retries.`);
-      return true;
+    const { failedEvents } = await this.delivery.deliver(batch);
+    if (failedEvents.length === 0) {
+      logger.info(`[BatchRetryManager] Batch delivery successfully after ${attempt + 1} tries.`);
+      return [];
     }
     // attempts are zero-indexed while maxRetries is not
     if (attempt < maxRetries - 1) {
-      return this.retry(batch, attempt + 1);
+      return this.retry(failedEvents, attempt + 1);
     } else {
-      logger.warn(
-        `[BatchRetryManager] Failed to deliver batch after ${maxRetries} retries, bailing`,
-      );
-      return false;
+      logger.warn(`[BatchRetryManager] Failed to deliver batch after ${maxRetries} tries, bailing`);
+      return batch;
     }
   }
 }

src/events/default-event-dispatcher.spec.ts

@@ -4,7 +4,7 @@ import DefaultEventDispatcher, {
   EventDispatcherConfig,
   newDefaultEventDispatcher,
 } from './default-event-dispatcher';
-import { Event } from './event-dispatcher';
+import Event from './event';
 import NetworkStatusListener from './network-status-listener';
 import NoOpEventDispatcher from './no-op-event-dispatcher';
 
@@ -100,7 +100,7 @@ describe('DefaultEventDispatcher', () => {
       });
 
       const fetch = global.fetch as jest.Mock;
-      fetch.mockResolvedValue({ ok: true });
+      fetch.mockResolvedValue({ ok: true, json: () => Promise.resolve([]) });
 
       await new Promise((resolve) => setTimeout(resolve, 100));
 
@@ -207,7 +207,7 @@ describe('DefaultEventDispatcher', () => {
       });
 
       const fetch = global.fetch as jest.Mock;
-      fetch.mockResolvedValue({ ok: true });
+      fetch.mockResolvedValue({ ok: true, json: () => Promise.resolve([]) });
 
       isOffline = true;
       // simulate the network going offline
@@ -242,7 +242,7 @@ describe('DefaultEventDispatcher', () => {
       // Simulate fetch failure on the first attempt
       (global.fetch as jest.Mock)
         .mockResolvedValueOnce({ ok: false }) // First attempt fails
-        .mockResolvedValueOnce({ ok: true }); // Second attempt succeeds
+        .mockResolvedValueOnce({ ok: true, json: () => Promise.resolve([]) }); // Second attempt succeeds
 
       // Fast-forward to trigger the first attempt
       await new Promise((resolve) => setTimeout(resolve, 100));

src/events/default-event-dispatcher.ts

@@ -2,8 +2,9 @@ import { logger } from '../application-logger';
 
 import BatchEventProcessor from './batch-event-processor';
 import BatchRetryManager from './batch-retry-manager';
+import Event from './event';
 import EventDelivery from './event-delivery';
-import EventDispatcher, { Event } from './event-dispatcher';
+import EventDispatcher from './event-dispatcher';
 import NamedEventQueue from './named-event-queue';
 import NetworkStatusListener from './network-status-listener';
 import NoOpEventDispatcher from './no-op-event-dispatcher';
@@ -93,13 +94,13 @@ export default class DefaultEventDispatcher implements EventDispatcher {
       return;
     }
 
-    const success = await this.eventDelivery.deliver(batch);
-    if (!success) {
-      logger.warn('[EventDispatcher] Failed to deliver batch, retrying...');
-      const retrySucceeded = await this.retryManager.retry(batch);
-      if (!retrySucceeded) {
+    const { failedEvents } = await this.eventDelivery.deliver(batch);
+    if (failedEvents.length > 0) {
+      logger.warn('[EventDispatcher] Failed to deliver some events from batch, retrying...');
+      const failedRetry = await this.retryManager.retry(failedEvents);
+      if (failedRetry.length > 0) {
         // re-enqueue events that failed to retry
-        this.batchProcessor.push(...batch);
+        this.batchProcessor.push(...failedRetry);
       }
     }
     logger.debug(`[EventDispatcher] Delivered batch of ${batch.length} events.`);

src/events/event-delivery.spec.ts

@@ -0,0 +1,59 @@
+import Event from './event';
+import EventDelivery from './event-delivery';
+
+describe('EventDelivery', () => {
+  global.fetch = jest.fn();
+  const sdkKey = 'test-sdk-key';
+  const ingestionUrl = 'https://test-ingestion.url';
+  const testBatch: Event[] = [
+    { uuid: '1', timestamp: Date.now(), type: 'test_event', payload: { key: 'value1' } },
+    { uuid: '2', timestamp: Date.now(), type: 'test_event', payload: { key: 'value2' } },
+    { uuid: '3', timestamp: Date.now(), type: 'test_event', payload: { key: 'value3' } },
+  ];
+  let eventDelivery: EventDelivery;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    eventDelivery = new EventDelivery(sdkKey, ingestionUrl);
+  });
+
+  it('should deliver events successfully when response is OK', async () => {
+    const mockResponse = { ok: true, json: async () => ({}) };
+    (global.fetch as jest.Mock).mockResolvedValue(mockResponse);
+    const result = await eventDelivery.deliver(testBatch);
+    expect(result).toEqual({ failedEvents: [] });
+    expect(global.fetch).toHaveBeenCalledWith(ingestionUrl, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json', 'x-eppo-token': sdkKey },
+      body: JSON.stringify({ eppo_events: testBatch }),
+    });
+  });
+
+  it('should return failed result if response is not OK', async () => {
+    const mockResponse = { ok: false };
+    (global.fetch as jest.Mock).mockResolvedValue(mockResponse);
+    const result = await eventDelivery.deliver(testBatch);
+    expect(result).toEqual({ failedEvents: testBatch });
+  });
+
+  it('should return failed events when response includes failed events', async () => {
+    const failedEvents = ['1', '2'];
+    const mockResponse = { ok: true, json: async () => ({ failed_events: failedEvents }) };
+    (global.fetch as jest.Mock).mockResolvedValue(mockResponse);
+    const result = await eventDelivery.deliver(testBatch);
+    expect(result).toEqual({ failedEvents: [testBatch[0], testBatch[1]] });
+  });
+
+  it('should return success=true if no failed events in the response', async () => {
+    const mockResponse = { ok: true, json: async () => ({}) };
+    (global.fetch as jest.Mock).mockResolvedValue(mockResponse);
+    const result = await eventDelivery.deliver(testBatch);
+    expect(result).toEqual({ failedEvents: [] });
+  });
+
+  it('should handle fetch errors gracefully', async () => {
+    (global.fetch as jest.Mock).mockRejectedValue(new Error('Network error'));
+    const result = await eventDelivery.deliver(testBatch);
+    expect(result).toEqual({ failedEvents: testBatch });
+  });
+});

src/events/event-delivery.ts

@@ -1,11 +1,19 @@
 import { logger } from '../application-logger';
 
-import { Event } from './event-dispatcher';
+import Event from './event';
+
+export type EventDeliveryResult = {
+  failedEvents: Event[];
+};
 
 export default class EventDelivery {
   constructor(private readonly sdkKey: string, private readonly ingestionUrl: string) {}
 
-  async deliver(batch: Event[]): Promise<boolean> {
+  /**
+   * Delivers a batch of events to the ingestion URL endpoint. Returns the UUIDs of any events from
+   * the batch that failed ingestion.
+   */
+  async deliver(batch: Event[]): Promise<EventDeliveryResult> {
     try {
       logger.info(
         `[EventDispatcher] Delivering batch of ${batch.length} events to ${this.ingestionUrl}...`,
@@ -15,11 +23,34 @@ export default class EventDelivery {
         headers: { 'Content-Type': 'application/json', 'x-eppo-token': this.sdkKey },
         body: JSON.stringify({ eppo_events: batch }),
       });
-      // TODO: Parse response to check `failed_event_uploads` for any failed event ingestions in the batch
-      return response.ok;
-    } catch {
-      logger.warn('Failed to upload event batch');
-      return false;
+      if (response.ok) {
+        return await this.parseFailedEvents(response, batch);
+      } else {
+        return { failedEvents: batch };
+      }
+    } catch (e: any) {
+      logger.warn(`Failed to upload event batch`, e);
+      return { failedEvents: batch };
+    }
+  }
+
+  private async parseFailedEvents(
+    response: Response,
+    batch: Event[],
+  ): Promise<EventDeliveryResult> {
+    logger.info('[EventDispatcher] Batch delivered successfully.');
+    const responseBody = (await response.json()) as { failed_events?: string[] };
+    const failedEvents = new Set(responseBody?.failed_events || []);
+    if (failedEvents.size > 0) {
+      logger.warn(
+        `[EventDispatcher] ${failedEvents.size}/${batch.length} events failed ingestion.`,
+      );
+      // even though some events may have failed to successfully deliver, we'll still consider
+      // the batch as a whole to have been delivered successfully and just re-enqueue the failed
+      // events for retry later
+      return { failedEvents: batch.filter(({ uuid }) => failedEvents.has(uuid)) };
+    } else {
+      return { failedEvents: [] };
     }
   }
 }

src/events/event-dispatcher.ts

@@ -1,9 +1,4 @@
-export type Event = {
-  uuid: string;
-  timestamp: number;
-  type: string;
-  payload: Record<string, unknown>;
-};
+import Event from './event';
 
 export default interface EventDispatcher {
   /** Dispatches (enqueues) an event for eventual delivery. */

src/events/event.ts

@@ -0,0 +1,8 @@
+type Event = {
+  uuid: string;
+  timestamp: number;
+  type: string;
+  payload: Record<string, unknown>;
+};
+
+export default Event;