Retry event buffer submissions (#31)

* Retry event buffer submissions

* Retry event buffer submissions

* test

Author: bpapillon

src/events.test.ts

@@ -5,20 +5,32 @@ import { Events } from "./api/resources/events/client/Client";
 import { CreateEventRequestBody } from "./api";
 import { Logger } from "./logger";
 
+process.env.NODE_ENV = "test";
+
 jest.useFakeTimers();
 
 describe("EventBuffer", () => {
     let mockEventsApi: jest.Mocked<Events>;
     let mockLogger: jest.Mocked<Logger>;
 
     beforeEach(() => {
+        const mockResponse = {
+            data: {
+                events: [],
+            },
+            params: {},
+        };
+
         mockEventsApi = {
-            createEventBatch: jest.fn().mockResolvedValue(undefined),
+            createEventBatch: jest.fn().mockResolvedValue(mockResponse),
         } as any;
 
         mockLogger = {
             error: jest.fn(),
             log: jest.fn(),
+            warn: jest.fn(),
+            info: jest.fn(),
+            debug: jest.fn(),
         } as any;
     });
 
@@ -70,9 +82,12 @@ describe("EventBuffer", () => {
     // The rest of the tests remain unchanged as they don't directly test the maxSize behavior
     it("should log error if flushing fails", async () => {
         mockEventsApi.createEventBatch.mockRejectedValue(new Error("Flush error"));
+
         const buffer = new EventBuffer(mockEventsApi, {
             logger: mockLogger,
             interval: 1000,
+            maxRetries: 1,
+            initialRetryDelay: 1,
         });
 
         const event: CreateEventRequestBody = {
@@ -87,10 +102,14 @@ describe("EventBuffer", () => {
         await buffer.push(event);
         await buffer.push(event);
 
-        // Manually trigger flush
+        // Since we're skipping delays in test environment,
+        // we can just call flush directly
         await buffer.flush();
 
-        expect(mockLogger.error).toHaveBeenCalledWith("Failed to flush events", expect.any(Error));
+        expect(mockLogger.error).toHaveBeenCalledWith(
+            "Event batch submission failed after 1 retries:",
+            expect.any(Error)
+        );
     });
 
     it("should stop accepting events after stop is called", async () => {
@@ -166,4 +185,45 @@ describe("EventBuffer", () => {
 
         expect(mockEventsApi.createEventBatch).not.toHaveBeenCalled();
     });
+
+    it("should retry and succeed after a failure", async () => {
+        const mockResponse = {
+            data: {
+                events: [],
+            },
+            params: {},
+        };
+
+        // First call fails, second succeeds
+        mockEventsApi.createEventBatch
+            .mockRejectedValueOnce(new Error("Temporary failure"))
+            .mockResolvedValueOnce(mockResponse);
+
+        const buffer = new EventBuffer(mockEventsApi, {
+            logger: mockLogger,
+            interval: 1000,
+            maxRetries: 3,
+            initialRetryDelay: 1,
+        });
+
+        const event: CreateEventRequestBody = {
+            body: {
+                company: { id: "test-company" },
+                event: "test-event",
+                user: { id: "test-user" },
+            },
+            eventType: "track",
+            sentAt: new Date(),
+        };
+        await buffer.push(event);
+
+        // Since we're skipping delays in test environment,
+        // we can just call flush directly
+        await buffer.flush();
+
+        // Verify that the createEventBatch was called twice (once failed, once succeeded)
+        expect(mockEventsApi.createEventBatch).toHaveBeenCalledTimes(2);
+
+        expect(mockLogger.info).toHaveBeenCalledWith("Event batch submission succeeded after 1 retries");
+    });
 });

src/events.ts

@@ -4,12 +4,16 @@ import { ConsoleLogger, Logger } from "./logger";
 
 const DEFAULT_FLUSH_INTERVAL = 1000; // 1 second
 const DEFAULT_MAX_SIZE = 1000; // 1000 items
+const DEFAULT_MAX_RETRIES = 3;
+const DEFAULT_INITIAL_RETRY_DELAY = 1000; // 1 second in milliseconds
 
 interface EventBufferOptions {
     interval?: number;
     logger?: Logger;
     maxSize?: number;
     offline?: boolean;
+    maxRetries?: number;
+    initialRetryDelay?: number;
 }
 
 class EventBuffer {
@@ -20,6 +24,8 @@ class EventBuffer {
     private logger: Logger;
     private maxSize: number;
     private offline: boolean;
+    private maxRetries: number;
+    private initialRetryDelay: number;
     private shutdown: boolean = false;
     private stopped: boolean = false;
 
@@ -29,12 +35,16 @@ class EventBuffer {
             maxSize = DEFAULT_MAX_SIZE,
             interval = DEFAULT_FLUSH_INTERVAL,
             offline = false,
+            maxRetries = DEFAULT_MAX_RETRIES,
+            initialRetryDelay = DEFAULT_INITIAL_RETRY_DELAY,
         } = opts || {};
         this.eventsApi = eventsApi;
         this.interval = interval;
         this.logger = logger;
         this.maxSize = maxSize;
         this.offline = offline;
+        this.maxRetries = maxRetries;
+        this.initialRetryDelay = initialRetryDelay;
 
         this.startPeriodicFlush();
     }
@@ -47,10 +57,49 @@ class EventBuffer {
         const events = [...this.events];
         this.events = [];
 
-        try {
-            await this.eventsApi.createEventBatch({ events });
-        } catch (err) {
-            this.logger.error("Failed to flush events", err);
+        // Initialize retry counter and success flag
+        let retryCount = 0;
+        let success = false;
+        let lastError: any = null;
+
+        // Try with retries and exponential backoff
+        while (retryCount <= this.maxRetries && !success) {
+            try {
+                if (retryCount > 0) {
+                    // Log retry attempt
+                    this.logger.info(`Retrying event batch submission (attempt ${retryCount} of ${this.maxRetries})`);
+                }
+
+                // Attempt to send events
+                await this.eventsApi.createEventBatch({ events });
+                success = true;
+            } catch (err) {
+                lastError = err;
+                retryCount++;
+
+                if (retryCount <= this.maxRetries) {
+                    // Calculate backoff with jitter
+                    const delay = this.initialRetryDelay * Math.pow(2, retryCount - 1);
+                    const jitter = Math.random() * 0.1 * delay; // 10% jitter
+                    const waitTime = delay + jitter;
+
+                    this.logger.warn(
+                        `Event batch submission failed: ${err}. Retrying in ${(waitTime / 1000).toFixed(2)} seconds...`
+                    );
+
+                    // Wait before retry
+                    if (process.env.NODE_ENV !== "test") {
+                        await new Promise((resolve) => setTimeout(resolve, waitTime));
+                    }
+                }
+            }
+        }
+
+        // After all retries, if still not successful, log the error
+        if (!success) {
+            this.logger.error(`Event batch submission failed after ${this.maxRetries} retries:`, lastError);
+        } else if (retryCount > 0) {
+            this.logger.info(`Event batch submission succeeded after ${retryCount} retries`);
         }
     }