Cap event capture requests at 100 events

src/events.ts

@@ -3,8 +3,12 @@ import { EventCaptureClient } from "./event-capture";
 import { ConsoleLogger, Logger } from "./logger";
 
 const DEFAULT_FLUSH_INTERVAL = 1000; // 1 second
-const DEFAULT_MAX_SIZE = 1000; // 1000 items
+const DEFAULT_MAX_SIZE = 100; // 100 items
 const DEFAULT_MAX_RETRIES = 3;
+// The capture service rejects any batch larger than this with
+// `400 {"error": "batch too large", "max_size": 100}`, so a flush is split
+// into chunks of at most this many events regardless of how many are buffered.
+const MAX_EVENTS_PER_REQUEST = 100;
 const DEFAULT_INITIAL_RETRY_DELAY = 1000; // 1 second in milliseconds
 
 interface EventBufferOptions {
@@ -60,52 +64,71 @@ class EventBuffer {
             const events = [...this.events];
             this.events = [];
 
-            // 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})`);
-                    }
+            // The buffer can hold more than one request's worth of events: a
+            // caller that does not await `push` keeps appending while a flush
+            // is in flight, since `push` skips its size check whenever
+            // `flushing` is set. Send in chunks so an oversized buffer is
+            // never turned into an oversized request.
+            //
+            // Each chunk is retried independently. Retrying the whole drained
+            // set together would resend chunks that already succeeded.
+            for (let i = 0; i < events.length; i += MAX_EVENTS_PER_REQUEST) {
+                await this.sendChunk(events.slice(i, i + MAX_EVENTS_PER_REQUEST));
+            }
+        } finally {
+            this.flushing = false;
+        }
+    }
 
-                    // Attempt to send events
-                    await this.captureClient.sendBatch(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));
-                        }
+    /**
+     * Sends a single request's worth of events, retrying with exponential
+     * backoff. Failures are logged and the chunk is dropped, matching the
+     * buffer's contract that tracking never throws to the caller.
+     */
+    private async sendChunk(events: CreateEventRequestBody[]): Promise<void> {
+        // 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.captureClient.sendBatch(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`);
-            }
-        } finally {
-            this.flushing = false;
+        // 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`);
         }
     }
 

tests/unit/events.test.ts

@@ -97,7 +97,7 @@ describe("EventBuffer", () => {
 
         expect(mockLogger.error).toHaveBeenCalledWith(
             "Event batch submission failed after 1 retries:",
-            expect.any(Error)
+            expect.any(Error),
         );
     });
 
@@ -259,4 +259,113 @@ describe("EventBuffer", () => {
 
         expect(mockLogger.info).toHaveBeenCalledWith("Event batch submission succeeded after 1 retries");
     });
+
+    describe("batch size cap", () => {
+        const makeEvent = (n: number): CreateEventRequestBody => ({
+            body: {
+                company: { id: "test-company" },
+                event: `test-event-${n}`,
+                user: { id: "test-user" },
+            },
+            eventType: "track",
+            sentAt: new Date(),
+        });
+
+        const batchSizes = () => mockCaptureClient.sendBatch.mock.calls.map((call) => call[0].length);
+
+        it("splits a drained buffer into requests of at most 100 events", async () => {
+            const buffer = new EventBuffer(mockCaptureClient, {
+                logger: mockLogger,
+                // Deliberately larger than the server's cap, to prove the cap
+                // is enforced at send time rather than by the buffer size.
+                maxSize: 1000,
+                interval: 1000,
+            });
+
+            for (let i = 0; i < 250; i++) {
+                await buffer.push(makeEvent(i));
+            }
+            await buffer.flush();
+
+            expect(batchSizes()).toEqual([100, 100, 50]);
+        });
+
+        it("never sends more than 100 events when pushes are not awaited", async () => {
+            // Reproduces the original failure. `push` skips its size check
+            // while a flush is in flight, so a caller that fires `track()`
+            // without awaiting keeps appending for the whole duration of the
+            // in-flight request. Hold the first request open so the buffer
+            // grows well past the cap before it is drained again.
+            let releaseFirstSend: () => void = () => undefined;
+            const firstSendHeld = new Promise<void>((resolve) => {
+                releaseFirstSend = resolve;
+            });
+            mockCaptureClient.sendBatch.mockImplementationOnce(() => firstSendHeld);
+
+            const buffer = new EventBuffer(mockCaptureClient, {
+                logger: mockLogger,
+                maxSize: 100,
+                interval: 1000,
+            });
+
+            const pushes = Promise.all(Array.from({ length: 300 }, (_, i) => buffer.push(makeEvent(i))));
+            // Let every push run up to the point where it appends or blocks.
+            await Promise.resolve();
+            releaseFirstSend();
+            await pushes;
+            await buffer.stop();
+
+            expect(mockCaptureClient.sendBatch).toHaveBeenCalled();
+            for (const size of batchSizes()) {
+                expect(size).toBeLessThanOrEqual(100);
+            }
+            const total = batchSizes().reduce((sum, size) => sum + size, 0);
+            expect(total).toBe(300);
+        });
+
+        it("does not resend a delivered chunk when a later chunk fails", async () => {
+            mockCaptureClient.sendBatch
+                .mockResolvedValueOnce(undefined) // chunk 1 delivered
+                .mockRejectedValue(new Error("boom")); // chunk 2 fails every attempt
+
+            const buffer = new EventBuffer(mockCaptureClient, {
+                logger: mockLogger,
+                maxSize: 1000,
+                interval: 1000,
+                maxRetries: 2,
+                initialRetryDelay: 1,
+            });
+
+            for (let i = 0; i < 150; i++) {
+                await buffer.push(makeEvent(i));
+            }
+            await buffer.flush();
+
+            // 1 delivery for chunk 1, then 1 + 2 retries for chunk 2.
+            expect(mockCaptureClient.sendBatch).toHaveBeenCalledTimes(4);
+
+            const firstChunk = mockCaptureClient.sendBatch.mock.calls[0][0];
+            const resends = mockCaptureClient.sendBatch.mock.calls.filter((call) => call[0] === firstChunk);
+            expect(resends).toHaveLength(1);
+        });
+
+        it("still attempts later chunks after an earlier chunk fails", async () => {
+            mockCaptureClient.sendBatch.mockRejectedValueOnce(new Error("boom")).mockResolvedValue(undefined);
+
+            const buffer = new EventBuffer(mockCaptureClient, {
+                logger: mockLogger,
+                maxSize: 1000,
+                interval: 1000,
+                maxRetries: 0,
+                initialRetryDelay: 1,
+            });
+
+            for (let i = 0; i < 150; i++) {
+                await buffer.push(makeEvent(i));
+            }
+            await buffer.flush();
+
+            expect(batchSizes()).toEqual([100, 50]);
+        });
+    });
 });