From b4102da3ea64153b48add8b074f622ecdd04afee Mon Sep 17 00:00:00 2001
From: ryan echternacht <ryan@echternacht.org>
Date: Wed, 19 Aug 2026 13:59:50 -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 both send sites
handed it the entire drained backlog.

@max_batch_size does not bound that backlog. push appends
unconditionally, and the flush it triggers returns immediately while
another flush is in flight:

    def flush
      @mutex.synchronize do
        return if @flushing || @events.empty?

so every event pushed during an in-flight request accumulates, and
drain_pending then sends the whole pile as one request. With two
producer threads and a slow send this is unbounded, growing with request
duration times push rate.

Route both flush and drain_pending through send_capped, which slices the
drained set into requests of at most 100 events. Each chunk retries on
its own, since retrying the whole set would resend chunks that had
already been delivered.

This matches schematic-go, schematic-java, and schematic-csharp, which
all bound the drain rather than the buffer.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 lib/schematic/event_buffer.rb | 18 +++++++-
 test/custom.test.rb           | 77 +++++++++++++++++++++++++++++++++++
 2 files changed, 93 insertions(+), 2 deletions(-)

diff --git a/lib/schematic/event_buffer.rb b/lib/schematic/event_buffer.rb
index b482ce9..91de348 100644
--- a/lib/schematic/event_buffer.rb
+++ b/lib/schematic/event_buffer.rb
@@ -8,6 +8,10 @@ module Schematic
   class EventBuffer
     DEFAULT_FLUSH_INTERVAL = 5.0 # seconds (canonical Go value)
     DEFAULT_MAX_BATCH_SIZE = 100
+    # The capture service rejects any batch larger than this with
+    # `400 {"error": "batch too large", "max_size": 100}`. @max_batch_size is
+    # only a flush trigger, so the cap is enforced again at send time.
+    MAX_EVENTS_PER_REQUEST = 100
     DEFAULT_MAX_RETRIES = 3
     DEFAULT_INITIAL_RETRY_DELAY = 1.0 # seconds
     JITTER_FACTOR = 0.25
@@ -59,7 +63,7 @@ def flush
 
       return unless events_to_send&.any?
 
-      send_batch(events_to_send)
+      send_capped(events_to_send)
 
       # Events may have accumulated while we were sending. Drain them so
       # a size-triggered flush that lost the race doesn't have to wait for
@@ -122,7 +126,17 @@ def drain_pending
         batch
       end
 
-      send_batch(events_to_send) if events_to_send&.any?
+      send_capped(events_to_send) if events_to_send&.any?
+    end
+
+    # A flush drains however much has piled up, which is not bounded by
+    # @max_batch_size: push appends unconditionally and its size-triggered
+    # flush is a no-op while @flushing is set, so every event pushed during an
+    # in-flight send accumulates. Split the drained set so an oversized buffer
+    # is never turned into an oversized request. Each chunk retries on its own,
+    # since retrying the whole set would resend chunks already delivered.
+    def send_capped(events)
+      events.each_slice(MAX_EVENTS_PER_REQUEST) { |chunk| send_batch(chunk) }
     end
 
     def send_batch(events)
diff --git a/test/custom.test.rb b/test/custom.test.rb
index 413398b..d5de91e 100644
--- a/test/custom.test.rb
+++ b/test/custom.test.rb
@@ -3445,3 +3445,80 @@ def build_ds_client(flag_cache: nil, company_cache: nil, user_cache: nil, rules_
     WebMock.reset!
   end
 end
+
+# =============================================================================
+# EventBuffer batch size cap
+# =============================================================================
+describe "EventBuffer batch size cap" do
+  after do
+    WebMock.reset!
+  end
+
+  def build_buffer
+    Schematic::EventBuffer.new(
+      api_key: "test_key",
+      logger: Schematic::ConsoleLogger.new(level: :error),
+      interval: 3600, # never fires; these tests drive the flush themselves
+      offline: false
+    )
+  end
+
+  it "splits a drained backlog into requests of at most 100 events" do
+    sizes = []
+    stub_request(:post, CAPTURE_URL).to_return do |req|
+      sizes << JSON.parse(req.body)["events"].size
+      { status: 200 }
+    end
+
+    buffer = build_buffer
+    # Seeded directly rather than pushed: pushing would flush at every 100th
+    # event, which is exactly the backlog this test needs to already exist.
+    buffer.instance_variable_set(
+      :@events,
+      (1..250).map { |i| { event_type: "track", body: { event: "e#{i}" } } }
+    )
+    buffer.flush
+    buffer.stop
+
+    assert_equal [100, 100, 50], sizes
+  end
+
+  it "never sends more than 100 events when events pile up behind an in-flight send" do
+    # push appends unconditionally and its size-triggered flush is a no-op
+    # while @flushing is set, so everything pushed during a send accumulates
+    # and drain_pending would send it as one oversized request.
+    sizes = []
+    sizes_mutex = Mutex.new
+    started = Queue.new
+    release = Queue.new
+
+    stub_request(:post, CAPTURE_URL).to_return do |req|
+      is_first = sizes_mutex.synchronize do
+        sizes << JSON.parse(req.body)["events"].size
+        sizes.size == 1
+      end
+      if is_first
+        started << true
+        release.pop
+      end
+      { status: 200 }
+    end
+
+    buffer = build_buffer
+
+    # The 100th push triggers a flush; hold that request open.
+    producer = Thread.new do
+      100.times { |i| buffer.push({ event_type: "track", body: { event: "a#{i}" } }) }
+    end
+
+    started.pop
+    150.times { |i| buffer.push({ event_type: "track", body: { event: "b#{i}" } }) }
+    release << true
+    producer.join
+    buffer.stop
+
+    assert_equal 250, sizes.sum
+    assert sizes.all? { |size| size <= 100 },
+           "expected every request to carry at most 100 events, got #{sizes.inspect}"
+  end
+end