Batch writes together

app/models/solid_cable/message.rb

@@ -15,6 +15,13 @@ def broadcast(channel, payload)
           channel_hash: channel_hash_for(channel) })
       end
 
+      def broadcast_batch(broadcasts)
+        created_at = Time.current
+        insert_all broadcasts.map { |channel, payload|
+          { created_at:, channel:, payload:, channel_hash: channel_hash_for(channel) }
+        }
+      end
+
       # Need to unpack this as a signed integer since Postgresql and SQLite
       # don't support unsigned integers
       def channel_hash_for(channel)

lib/action_cable/subscription_adapter/solid_cable.rb

@@ -20,10 +20,11 @@ def initialize(*)
           end
 
         @listener = nil
+        @broadcaster = nil
       end
 
       def broadcast(channel, payload)
-        ::SolidCable::Message.broadcast(channel, payload)
+        broadcaster.broadcast(channel, payload)
 
         ::SolidCable::TrimJob.perform_now if ::SolidCable.autotrim?
       end
@@ -36,7 +37,10 @@ def unsubscribe(channel, callback)
         listener.remove_subscriber(channel, callback)
       end
 
-      delegate :shutdown, to: :listener
+      def shutdown
+        @broadcaster&.shutdown
+        @listener&.shutdown
+      end
 
       private
         def listener
@@ -45,6 +49,12 @@ def listener
           end
         end
 
+        def broadcaster
+          @broadcaster || @mutex.synchronize do
+            @broadcaster ||= ::SolidCable::BatchedBroadcaster.new
+          end
+        end
+
         def pubsub_executor
           @pubsub_executor ||=
             if respond_to?(:executor, true)

lib/generators/solid_cable/install/templates/config/cable.yml

@@ -15,3 +15,5 @@ production:
       writing: cable
   polling_interval: 0.1.seconds
   message_retention: 1.day
+  writer_batch_size: 4
+  writer_batch_delay: 0.001.seconds

lib/solid_cable.rb

@@ -3,13 +3,15 @@
 require "solid_cable/version"
 require "solid_cable/engine"
 require "solid_cable/configuration"
+require "solid_cable/batched_broadcaster"
 require "action_cable/subscription_adapter/solid_cable"
 
 module SolidCable
   class << self
     delegate :connects_to, :silence_polling?, :polling_interval,
       :message_retention, :autotrim?, :trim_batch_size, :use_skip_locked,
-      :trim_chance, :reconnect_attempts, to: :configuration
+      :trim_chance, :reconnect_attempts, :writer_batch_size, :writer_batch_delay,
+      to: :configuration
 
     def configuration
       @configuration ||= Configuration.new(**Rails.application.config_for("cable"))

lib/solid_cable/batched_broadcaster.rb

@@ -0,0 +1,84 @@
+# frozen_string_literal: true
+
+module SolidCable
+  class BatchedBroadcaster
+    Stopped = Class.new(StandardError)
+    Message = Struct.new(:channel, :payload, keyword_init: true)
+
+    def initialize(batch_size: SolidCable.writer_batch_size, batch_delay: SolidCable.writer_batch_delay)
+      @batch_size = batch_size
+      @batch_delay = batch_delay
+      @queue = Queue.new
+
+      @thread = Thread.new do
+        Thread.current.name = "solid_cable_writer"
+        Thread.current.abort_on_exception = true
+        listen_for_initial_messages
+      end
+    end
+
+    def broadcast(channel, payload)
+      message = Message.new(channel:, payload:)
+
+      queue.enq message
+    rescue ClosedQueueError
+      raise Stopped, "Solid Cable writer has stopped"
+    end
+
+    def shutdown
+      queue.close
+      thread.join
+    end
+
+    private
+      attr_reader :batch_size, :batch_delay, :queue, :thread, :queue_size
+
+      def listen_for_initial_messages
+        loop do
+          message = queue.pop
+
+          break if message.nil?
+
+          collect_batch(message)
+        end
+      end
+
+      def collect_batch(first_message)
+        batch = [ first_message ]
+        deadline = monotonic_time + batch_delay
+
+        drain_queue_into(batch)
+        wait_for_messages_until(batch, deadline)
+
+        flush(batch)
+      end
+
+      def drain_queue_into(batch)
+        while batch.size < batch_size && (message = queue.pop(timeout: 0))
+          batch << message
+        end
+      end
+
+      def wait_for_messages_until(batch, deadline)
+        while batch.size < batch_size && (remaining = deadline - monotonic_time).positive?
+          message = queue.pop(timeout: remaining)
+          break if message.nil?
+
+          batch << message
+        end
+      end
+
+      def flush(batch)
+        Rails.application.executor.wrap do
+          SolidCable::Message.
+            broadcast_batch(batch.map { |message| [ message.channel, message.payload ] })
+        end
+      rescue StandardError => error
+        Rails.error.report(error)
+      end
+
+      def monotonic_time
+        Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      end
+  end
+end

lib/solid_cable/configuration.rb

@@ -6,7 +6,7 @@ def initialize(**options)
 
     attr_writer :connects_to, :silence_polling, :polling_interval,
       :message_retention, :autotrim, :trim_batch_size, :use_skip_locked,
-      :trim_chance, :reconnect_attempts
+      :trim_chance, :reconnect_attempts, :writer_batch_size, :writer_batch_delay
 
     def connects_to
       @connects_to ||= options.connects_to.to_h.deep_transform_values(&:to_sym)
@@ -66,6 +66,15 @@ def reconnect_attempts
       end
     end
 
+    def writer_batch_size
+      @writer_batch_size ||= [ (options.writer_batch_size || 4).to_i, 1 ].max
+    end
+
+    def writer_batch_delay
+      @writer_batch_delay ||=
+        [ parse_duration(options.writer_batch_delay, default: 0.001.seconds), 0 ].max
+    end
+
     private
       attr_reader :options
 

test/lib/action_cable/subscription_adapter/solid_cable_test.rb

@@ -151,6 +151,7 @@ class ActionCable::SubscriptionAdapter::SolidCableTest < ActionCable::TestCase
   test "does not send old messages" do
     @tx_adapter.broadcast("channel", "channel1")
     @tx_adapter.broadcast("channel", "channel2")
+    wait_for_messages("channel1", "channel2")
 
     subscribe_as_queue("channel") do |queue|
       assert_empty queue
@@ -159,6 +160,7 @@ class ActionCable::SubscriptionAdapter::SolidCableTest < ActionCable::TestCase
       @tx_adapter.broadcast("channel", "channel4")
       @tx_adapter.broadcast("other", "other1")
       @tx_adapter.broadcast("other", "other2")
+      wait_for_messages("channel3", "channel4", "other1", "other2")
 
       subscribe_as_queue("other") do |other_queue|
         assert_empty other_queue
@@ -169,6 +171,7 @@ class ActionCable::SubscriptionAdapter::SolidCableTest < ActionCable::TestCase
 
     @tx_adapter.broadcast("channel", "channel5")
     @tx_adapter.broadcast("channel", "channel6")
+    wait_for_messages("channel5", "channel6")
 
     subscribe_as_queue("channel") do |queue|
       assert_empty queue
@@ -258,4 +261,10 @@ def with_active_record_logger(logger)
     def next_message_in_queue(queue)
       Timeout.timeout(5, nil, "Failed to get next item in queue") { queue.pop }
     end
+
+    def wait_for_messages(*payloads)
+      Timeout.timeout(5, nil, "Failed to persist broadcasts") do
+        sleep 0.001 until SolidCable::Message.where(payload: payloads).count == payloads.size
+      end
+    end
 end

test/lib/solid_cable/batched_broadcaster_test.rb

@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class SolidCable::BatchedBroadcasterTest < ActiveSupport::TestCase
+  teardown do
+    @broadcaster&.shutdown
+  end
+
+  test "batches concurrent broadcasts" do
+    @broadcaster = SolidCable::BatchedBroadcaster.new(batch_size: 2, batch_delay: 1)
+    batches = Queue.new
+
+    SolidCable::Message.stub(:broadcast_batch, ->(batch) { batches << batch }) do
+      threads = [
+        Thread.new { @broadcaster.broadcast("one", "first") },
+        Thread.new { @broadcaster.broadcast("two", "second") }
+      ]
+      threads.each(&:join)
+      @broadcaster.shutdown
+    end
+
+    assert_equal 1, batches.size
+    assert_equal [ [ "one", "first" ], [ "two", "second" ] ], batches.pop.sort
+  end
+
+  test "rejects broadcasts after shutdown" do
+    @broadcaster = SolidCable::BatchedBroadcaster.new(batch_size: 2, batch_delay: 0)
+    @broadcaster.shutdown
+
+    assert_raises(SolidCable::BatchedBroadcaster::Stopped) do
+      @broadcaster.broadcast("channel", "payload")
+    end
+  end
+
+  test "reports write errors" do
+    @broadcaster = SolidCable::BatchedBroadcaster.new(batch_size: 1, batch_delay: 0)
+    write_error = RuntimeError.new("write failed")
+    reported_errors = Queue.new
+
+    Rails.error.stub(:report, ->(error, **) { reported_errors << error }) do
+      SolidCable::Message.stub(:broadcast_batch, ->(*) { raise write_error }) do
+        @broadcaster.broadcast("channel", "payload")
+        @broadcaster.shutdown
+      end
+    end
+
+    assert_same write_error, reported_errors.pop
+  end
+end