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