diff --git a/src/events.ts b/src/events.ts index fb21dfa6..52b0ebc3 100644 --- a/src/events.ts +++ b/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 { + // 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`); } } diff --git a/tests/unit/events.test.ts b/tests/unit/events.test.ts index 10f017ee..4a185b32 100644 --- a/tests/unit/events.test.ts +++ b/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((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]); + }); + }); });