From 62fbea2e82ea34d5ef25f71b039afd0813850c10 Mon Sep 17 00:00:00 2001 From: ryan echternacht Date: Wed, 19 Aug 2026 13:54:24 -0400 Subject: [PATCH] Cap event capture requests at 100 events The capture service rejects any batch over 100 events with `400 {"error": "batch too large", "max_size": 100}`, but EventBuffer drained its entire backlog into a single sendBatch call. Any caller with more than 100 buffered events lost the whole batch, with only a log line to show for it. The buffer's maxSize did not bound this. It gates the push-triggered flush, and only when a flush is not already running: if (this.events.length >= this.maxSize && !this.flushing) { await this.flush(); } this.events.push(event); so while a request is in flight the size check is skipped entirely and callers keep appending for its full duration, retry backoff included. A caller that does not await track() sees the same thing, since pushes interleave across the await inside push. Split each flush into chunks of at most 100 events, retrying each chunk independently. Per-chunk retries matter: retrying the whole drained set would resend chunks that had already been delivered. Also lower the default maxSize from 1000 to 100 so the common path is one request per flush. The 1000 default predates the move to the capture service in 1.5.0, which is when this became reachable. Before that the buffer flushed to the REST createEventBatch endpoint and the mismatch was latent. Reported by a daily cron that tracked 113 companies then called close(). All 113 events went out as one request and 400'd, losing a full day of usage data. Co-Authored-By: Claude Opus 5 (1M context) --- src/events.ts | 107 +++++++++++++++++++++--------------- tests/unit/events.test.ts | 111 +++++++++++++++++++++++++++++++++++++- 2 files changed, 175 insertions(+), 43 deletions(-) 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]); + }); + }); });