fix: Handle batch size too large in kinesis batch sender (#64)

* handle batch size too large errors

* rubocop fixes

* fix bug in metric for failed events

* fix test data

Author: effron

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    journaled (6.2.4)
+    journaled (6.2.5)
       activejob
       activerecord
       activesupport
@@ -231,7 +231,7 @@ GEM
     uncruft (0.5.0)
       railties (>= 6.1.0)
     unicode-display_width (2.6.0)
-    uri (1.0.4)
+    uri (1.1.1)
     useragent (0.16.11)
     webmock (3.24.0)
       addressable (>= 2.8.0)

app/models/journaled/outbox/event.rb

@@ -19,6 +19,7 @@ class Event < Journaled.outbox_base_class_name.constantize
       attribute :event_data, :json
 
       validates :event_type, :event_data, :partition_key, :stream_name, presence: true
+      validate :failed_at_and_failure_reason_must_be_consistent
 
       scope :ready_to_process, -> {
         where(failed_at: nil)
@@ -63,6 +64,14 @@ def requeue!
       def self.oldest_non_failed_timestamp
         ready_to_process.order(:id).limit(1).pick(:created_at)
       end
+
+      private
+
+      def failed_at_and_failure_reason_must_be_consistent
+        if failed_at.present? != failure_reason.present?
+          errors.add(:base, 'failed_at and failure_reason must both be present or both be absent')
+        end
+      end
     end
   end
 end

gemfiles/rails_7_2.gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: ..
   specs:
-    journaled (6.2.4)
+    journaled (6.2.5)
       activejob
       activerecord
       activesupport

gemfiles/rails_8_0.gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: ..
   specs:
-    journaled (6.2.4)
+    journaled (6.2.5)
       activejob
       activerecord
       activesupport

lib/journaled/kinesis_batch_sender.rb

@@ -15,6 +15,8 @@ class KinesisBatchSender
       'ValidationException',
     ].freeze
 
+    BATCH_TOO_LARGE_PATTERN = /too large/i
+
     # Send a batch of database events to Kinesis
     #
     # Uses put_records batch API. Groups events by stream and sends each group as a batch.
@@ -40,11 +42,10 @@ def send_stream_batch(stream_name, stream_events)
       begin
         response = kinesis_client.put_records(stream_name:, records:)
         process_response(response, stream_events)
-      rescue Aws::Kinesis::Errors::ValidationException
-        # Re-raise batch-level validation errors (configuration issues)
-        # These indicate invalid stream name, batch too large, etc.
-        # Not event data problems - requires manual intervention
-        raise
+      rescue Aws::Kinesis::Errors::ValidationException => e
+        raise unless e.message.match?(BATCH_TOO_LARGE_PATTERN)
+
+        handle_batch_too_large(stream_name, stream_events)
       rescue StandardError => e
         # Handle transient errors (throttling, network issues, service unavailable)
         handle_transient_batch_error(e, stream_events)
@@ -93,6 +94,36 @@ def create_failed_event(event, error_code:, error_message:, transient:)
       )
     end
 
+    def handle_batch_too_large(stream_name, stream_events)
+      if stream_events.size <= 1
+        # Single event exceeds payload limit — treat as permanent failure
+        return {
+          succeeded: [],
+          failed: stream_events.map do |event|
+            create_failed_event(
+              event,
+              error_code: 'ValidationException',
+              error_message: 'Record exceeds Kinesis payload limit',
+              transient: false,
+            )
+          end,
+        }
+      end
+
+      Rails.logger.warn(
+        "[journaled] Batch too large for Kinesis (#{stream_events.size} events), splitting in half and retrying",
+      )
+
+      mid = stream_events.size / 2
+      first_half = send_stream_batch(stream_name, stream_events[...mid])
+      second_half = send_stream_batch(stream_name, stream_events[mid...])
+
+      {
+        succeeded: first_half[:succeeded] + second_half[:succeeded],
+        failed: first_half[:failed] + second_half[:failed],
+      }
+    end
+
     def handle_transient_batch_error(error, stream_events)
       Rails.logger.error("Kinesis batch send failed (transient): #{error.class} - #{error.message}")
 

lib/journaled/outbox/metric_emitter.rb

@@ -73,7 +73,7 @@ def calculate_queue_metrics
             Event.select(
               'COUNT(*) AS total_count',
               'COUNT(*) FILTER (WHERE failed_at IS NULL) AS workable_count',
-              'COUNT(*) FILTER (WHERE failure_reason IS NOT NULL AND failed_at IS NULL) AS failed_count',
+              'COUNT(*) FILTER (WHERE failed_at IS NOT NULL) AS failed_count',
               'MIN(created_at) FILTER (WHERE failed_at IS NULL) AS oldest_non_failed_timestamp',
             ).to_sql,
           )

lib/journaled/version.rb

@@ -1,5 +1,5 @@
 # frozen_string_literal: true
 
 module Journaled
-  VERSION = "6.2.4"
+  VERSION = "6.2.5"
 end

spec/lib/journaled/kinesis_batch_sender_spec.rb

@@ -125,6 +125,75 @@
       end
     end
 
+    context 'when batch exceeds total payload limit' do
+      before do
+        skip "Database event tests require PostgreSQL" unless ActiveRecord::Base.connection.adapter_name == 'PostgreSQL'
+      end
+
+      let(:large_data) { { event_type: 'test', payload: 'x' * 3_000_000 } }
+      let(:event_1) { create_database_event(stream_name: 'stream1', event_data: large_data) }
+      let(:event_2) { create_database_event(stream_name: 'stream1', event_data: large_data) }
+      let(:events) { [event_1, event_2] }
+
+      before do
+        allow(Rails.logger).to receive(:warn)
+
+        call_count = 0
+        allow(kinesis_client).to receive(:put_records) do |args|
+          call_count += 1
+          if call_count == 1
+            raise Aws::Kinesis::Errors::ValidationException.new(nil, 'Request Payload is too large')
+          end
+
+          mock_put_records_response(
+            args[:records].map { { error_code: nil, error_message: nil } },
+          )
+        end
+      end
+
+      it 'splits the batch and sends each half separately' do
+        result = subject.send_batch(events)
+
+        expect(result[:succeeded]).to contain_exactly(event_1, event_2)
+        expect(result[:failed]).to be_empty
+        # 1 failed call + 2 successful half-batch calls
+        expect(kinesis_client).to have_received(:put_records).exactly(3).times
+      end
+
+      it 'logs a warning about the split' do
+        subject.send_batch(events)
+
+        expect(Rails.logger).to have_received(:warn).with(/Batch too large.*splitting in half/)
+      end
+    end
+
+    context 'when a single event exceeds the payload limit' do
+      before do
+        skip "Database event tests require PostgreSQL" unless ActiveRecord::Base.connection.adapter_name == 'PostgreSQL'
+      end
+
+      let(:event) { create_database_event(stream_name: 'stream1') }
+      let(:events) { [event] }
+
+      before do
+        allow(kinesis_client).to receive(:put_records)
+          .and_raise(Aws::Kinesis::Errors::ValidationException.new(nil, 'Request Payload is too large'))
+      end
+
+      it 'marks the event as a permanent failure' do
+        result = subject.send_batch(events)
+
+        expect(result[:succeeded]).to be_empty
+        expect(result[:failed].length).to eq(1)
+
+        failure = result[:failed].first
+        expect(failure.event).to eq(event)
+        expect(failure.error_code).to eq('ValidationException')
+        expect(failure.error_message).to eq('Record exceeds Kinesis payload limit')
+        expect(failure.permanent?).to be true
+      end
+    end
+
     context 'with permanent error' do
       before do
         skip "Database event tests require PostgreSQL" unless ActiveRecord::Base.connection.adapter_name == 'PostgreSQL'

spec/lib/journaled/outbox/worker_spec.rb

@@ -341,6 +341,7 @@
           partition_key: 'key2',
           stream_name: 'test_stream',
           failure_reason: 'Some error',
+          failed_at: Time.current,
         )
 
         # Start worker, then travel forward in time
@@ -405,6 +406,7 @@
           event_data: { test: 'data' },
           partition_key: 'key2',
           stream_name: 'test_stream',
+          failure_reason: 'Some error',
           failed_at: Time.current,
         )
 
@@ -433,21 +435,20 @@
         expect(emitted['journaled.worker.queue_workable_count'][:value]).to eq(1) # Only non-failed events
       end
 
-      it 'counts erroring events correctly' do
+      it 'counts failed events correctly' do
         Journaled::Outbox::Event.create!(
           event_type: 'test_event',
           event_data: { test: 'data' },
           partition_key: 'key1',
           stream_name: 'test_stream',
-          failure_reason: 'Error but not failed',
         )
 
         Journaled::Outbox::Event.create!(
           event_type: 'test_event',
           event_data: { test: 'data' },
           partition_key: 'key2',
           stream_name: 'test_stream',
-          failure_reason: 'Error and failed',
+          failure_reason: 'Permanent failure',
           failed_at: Time.current,
         )
 
@@ -472,7 +473,7 @@
           sleep 0.1 until emitted.key?('journaled.worker.queue_failed_count') || Time.current > timeout
         end
 
-        expect(emitted['journaled.worker.queue_failed_count'][:value]).to eq(1) # Only events with error but not failed
+        expect(emitted['journaled.worker.queue_failed_count'][:value]).to eq(1)
       end
     end
   end

spec/models/journaled/outbox/event_spec.rb

@@ -66,6 +66,7 @@
         partition_key: 'key1',
         stream_name: 'test_stream',
         failed_at: Time.current,
+        failure_reason: 'Some error',
       )
 
       # Create a non-failed event