feat: Add ability to configure Outbox processor to use batches or guaranteed ordering (#58)

* use parallelizable outbox using batch api

* add back sequential mode as library level config

* clean up tests, reset class level config in test helper, pull out KinesisFailedEvent struct

* add back accidentally deleted event specs

* update readme, bump version

* update readme to acknowledge possible future optimizations to guaranteed_order mode

* clarify readme

* use 500 as default worker batch size

* shorten default sleep, always sleep between batches to prevent excessive database polling

Author: effron

Gemfile.lock

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

README.md

@@ -164,6 +164,25 @@ Journaling provides a number of different configuation options that can be set i
   Journaled.outbox_base_class_name = 'EventsRecord'
   ```
 
+#### `Journaled.outbox_processing_mode` (default: `:batch`)
+
+  **Only relevant when using `Journaled::Outbox::Adapter`.**
+
+  Controls how events are sent to Kinesis. Two modes are available:
+
+  - **`:batch`** (default) - Uses the Kinesis `put_records` batch API for high throughput. Events are sent in parallel batches, allowing multiple workers to run concurrently. Best for most use cases where strict ordering is not required.
+
+  - **`:guaranteed_order`** - Uses the Kinesis `put_record` single-event API to send events sequentially. Events are processed one at a time in order, stopping on the first transient failure to preserve ordering. Use this when you need strict ordering guarantees per partition key. Note: The current implementation requires single-threaded processing, but future optimizations may support batching and multi-threading by partition key.
+
+  Example:
+  ```ruby
+  # For high throughput (default)
+  Journaled.outbox_processing_mode = :batch
+
+  # For guaranteed ordering
+  Journaled.outbox_processing_mode = :guaranteed_order
+  ```
+
 #### ActiveJob `set` options
 
 Both model-level directives accept additional options to be passed into ActiveJob's `set` method:
@@ -182,6 +201,8 @@ journal_attributes :email, enqueue_with: { priority: 20, queue: 'journaled' }
 
 Journaled includes a built-in Outbox-style delivery adapter with horizontally scalable workers.
 
+By default, the Outbox adapter uses the Kinesis `put_records` batch API for high-throughput event processing, allowing multiple workers to process events in parallel. If you require strict ordering guarantees per partition key, you can configure sequential processing mode (see configuration options below).
+
 **Setup:**
 
 This feature requires creating database tables and is completely optional. Existing users are unaffected.
@@ -207,6 +228,16 @@ Journaled.delivery_adapter = Journaled::Outbox::Adapter
 # Optional: Customize worker behavior (these are the defaults)
 Journaled.worker_batch_size = 500        # Max events per Kinesis batch (Kinesis API limit)
 Journaled.worker_poll_interval = 5       # Seconds between polls
+
+# Optional: Configure processing mode (default: :batch)
+# - :batch - Uses Kinesis put_records batch API for high throughput (default)
+#            Events are sent in parallel batches. Multiple workers can run concurrently.
+# - :guaranteed_order - Uses Kinesis put_record single-event API for sequential processing
+#                       Events are sent one at a time in order. Use this if you need
+#                       strict ordering guarantees per partition key. The current
+#                       implementation processes events single-threaded, though future
+#                       optimizations may support batching/multi-threading by partition key.
+Journaled.outbox_processing_mode = :batch
 ```
 
 **Note:** When using the Outbox adapter, you do **not** need to configure an ActiveJob queue adapter (skip step 1 of Installation). The Outbox adapter uses the `journaled_outbox_events` table for event storage and its own worker daemons for processing, making it independent of ActiveJob. Transactional batching still works seamlessly with the Outbox adapter.
@@ -217,6 +248,8 @@ Journaled.worker_poll_interval = 5       # Seconds between polls
 bundle exec rake journaled_worker:work
 ```
 
+**Note:** In `:batch` mode (the default), you can run multiple worker processes concurrently for horizontal scaling. In `:guaranteed_order` mode, the current implementation is optimized for running a single worker to maintain ordering guarantees.
+
 4. **Monitoring:**
 
 The system emits `ActiveSupport::Notifications` events:

app/models/journaled/outbox/event.rb

@@ -29,12 +29,20 @@ class Event < Journaled.outbox_base_class_name.constantize
 
       # Fetch a batch of events for processing using SELECT FOR UPDATE
       #
+      # In :guaranteed_order mode, uses blocking lock to ensure sequential processing.
+      # In :batch mode, uses SKIP LOCKED to allow parallel workers.
+      #
       # @return [Array<Journaled::Outbox::Event>] Events locked for processing
       def self.fetch_batch_for_update
-        ready_to_process
-          .limit(Journaled.worker_batch_size)
-          .lock
-          .to_a
+        query = ready_to_process.limit(Journaled.worker_batch_size)
+
+        lock_clause = if Journaled.outbox_processing_mode == :guaranteed_order
+          'FOR UPDATE'
+        else
+          'FOR UPDATE SKIP LOCKED'
+        end
+
+        query.lock(lock_clause).to_a
       end
 
       # Requeue a failed event for processing

gemfiles/rails_7_2.gemfile.lock

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

gemfiles/rails_8_0.gemfile.lock

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

lib/journaled.rb

@@ -12,7 +12,9 @@
 require 'journaled/delivery_adapters/active_job_adapter'
 require 'journaled/outbox/adapter'
 require 'journaled/kinesis_client_factory'
+require 'journaled/kinesis_failed_event'
 require 'journaled/kinesis_batch_sender'
+require 'journaled/kinesis_sequential_sender'
 require 'journaled/outbox/batch_processor'
 require 'journaled/outbox/metric_emitter'
 require 'journaled/outbox/worker'
@@ -31,8 +33,9 @@ module Journaled
   mattr_writer(:transactional_batching_enabled) { true }
 
   # Worker configuration (for Outbox-style event processing)
-  mattr_accessor(:worker_batch_size) { 1000 }
-  mattr_accessor(:worker_poll_interval) { 1 } # seconds
+  mattr_accessor(:worker_batch_size) { 500 }
+  mattr_accessor(:worker_poll_interval) { 0.5 } # seconds
+  mattr_accessor(:outbox_processing_mode) { :batch } # :batch or :guaranteed_order
 
   def self.transactional_batching_enabled?
     Thread.current[:journaled_transactional_batching_enabled] || @@transactional_batching_enabled

lib/journaled/kinesis_batch_sender.rb

@@ -1,98 +1,108 @@
 # frozen_string_literal: true
 
 module Journaled
-  # Sends batches of events to Kinesis using the PutRecord single-event API
+  # Sends batches of events to Kinesis using the PutRecords batch API
   #
   # This class handles:
-  # - Sending events individually to support guaranteed ordering
+  # - Sending events in batches to improve throughput
   # - Handling failures on a per-event basis
   # - Classifying errors as transient vs permanent
   #
   # Returns structured results for the caller to handle event state management.
   class KinesisBatchSender
-    FailedEvent = Struct.new(:event, :error_code, :error_message, :transient, keyword_init: true) do
-      def transient?
-        transient
-      end
-
-      def permanent?
-        !transient
-      end
-    end
-
-    PERMANENT_ERROR_CLASSES = [
-      Aws::Kinesis::Errors::ValidationException,
+    # Per-record error codes that indicate permanent failures (bad event data)
+    PERMANENT_ERROR_CODES = [
+      'ValidationException',
     ].freeze
 
     # Send a batch of database events to Kinesis
     #
-    # Sends events one at a time to guarantee ordering. Stops on first transient failure.
+    # Uses put_records batch API. Groups events by stream and sends each group as a batch.
     #
     # @param events [Array<Journaled::Outbox::Event>] Events to send
     # @return [Hash] Result with:
     #   - succeeded: Array of successfully sent events
-    #   - failed: Array of FailedEvent structs (only permanent failures)
+    #   - failed: Array of FailedEvent structs (both transient and permanent failures)
     def send_batch(events)
-      result = { succeeded: [], failed: [] }
+      # Group events by stream since put_records requires all records to go to the same stream
+      events.group_by(&:stream_name).each_with_object({ succeeded: [], failed: [] }) do |(stream_name, stream_events), result|
+        batch_result = send_stream_batch(stream_name, stream_events)
+        result[:succeeded].concat(batch_result[:succeeded])
+        result[:failed].concat(batch_result[:failed])
+      end
+    end
 
-      events.each do |event|
-        event_result = send_event(event)
-        if event_result.is_a?(FailedEvent)
-          if event_result.transient?
-            emit_transient_failure_metric
-            break
-          else
-            result[:failed] << event_result
-          end
-        else
-          result[:succeeded] << event_result
-        end
+    private
+
+    def send_stream_batch(stream_name, stream_events)
+      records = build_records(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 StandardError => e
+        # Handle transient errors (throttling, network issues, service unavailable)
+        handle_transient_batch_error(e, stream_events)
       end
+    end
 
-      result
+    def build_records(stream_events)
+      stream_events.map do |event|
+        {
+          data: event.event_data.merge(id: event.id).to_json,
+          partition_key: event.partition_key,
+        }
+      end
     end
 
-    private
+    def process_response(response, stream_events)
+      succeeded = []
+      failed = []
 
-    # Send a single event to Kinesis
-    #
-    # @param event [Journaled::Outbox::Event] Event to send
-    # @return [Journaled::Outbox::Event, FailedEvent] The event on success, or FailedEvent on failure
-    def send_event(event)
-      # Merge the DB-generated ID into the event data before sending to Kinesis
-      event_data_with_id = event.event_data.merge(id: event.id)
+      response.records.each_with_index do |record_result, index|
+        event = stream_events[index]
 
-      kinesis_client.put_record(
-        stream_name: event.stream_name,
-        data: event_data_with_id.to_json,
-        partition_key: event.partition_key,
-      )
+        if record_result.error_code
+          failed << create_failed_event(event, record_result)
+        else
+          succeeded << event
+        end
+      end
 
-      event
-    rescue *PERMANENT_ERROR_CLASSES => e
-      Rails.logger.error("Kinesis event send failed (permanent): #{e.class} - #{e.message}")
-      FailedEvent.new(
-        event:,
-        error_code: e.class.to_s,
-        error_message: e.message,
-        transient: false,
-      )
-    rescue StandardError => e
-      Rails.logger.error("Kinesis event send failed (transient): #{e.class} - #{e.message}")
-      FailedEvent.new(
+      { succeeded:, failed: }
+    end
+
+    def create_failed_event(event, record_result)
+      Journaled::KinesisFailedEvent.new(
         event:,
-        error_code: e.class.to_s,
-        error_message: e.message,
-        transient: true,
+        error_code: record_result.error_code,
+        error_message: record_result.error_message,
+        transient: PERMANENT_ERROR_CODES.exclude?(record_result.error_code),
       )
     end
 
-    def kinesis_client
-      @kinesis_client ||= KinesisClientFactory.build
+    def handle_transient_batch_error(error, stream_events)
+      Rails.logger.error("Kinesis batch send failed (transient): #{error.class} - #{error.message}")
+
+      failed = stream_events.map do |event|
+        Journaled::KinesisFailedEvent.new(
+          event:,
+          error_code: error.class.to_s,
+          error_message: error.message,
+          transient: true,
+        )
+      end
+
+      { succeeded: [], failed: }
     end
 
-    def emit_transient_failure_metric
-      ActiveSupport::Notifications.instrument('journaled.kinesis_batch_sender.transient_failure')
+    def kinesis_client
+      @kinesis_client ||= KinesisClientFactory.build
     end
   end
 end

lib/journaled/kinesis_failed_event.rb

@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+module Journaled
+  # Represents a failed event from Kinesis send operations
+  #
+  # Used by both KinesisBatchSender and KinesisSequentialSender to represent
+  # events that failed to send to Kinesis, along with error details and whether
+  # the failure is transient (retriable) or permanent.
+  KinesisFailedEvent = Struct.new(:event, :error_code, :error_message, :transient, keyword_init: true) do
+    def transient?
+      transient
+    end
+
+    def permanent?
+      !transient
+    end
+  end
+end