Cap event capture requests at 100 events

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)

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