From 9a0c5e7dcbc5e6c3cdd6a27adff2d43ab1fea5ca Mon Sep 17 00:00:00 2001
From: Nick Pezza <pezza@hey.com>
Date: Fri, 14 Aug 2026 11:53:37 -0400
Subject: [PATCH 1/5] Batch writes together

---
 app/models/solid_cable/message.rb             |  7 ++
 .../subscription_adapter/solid_cable.rb       | 14 +++-
 .../install/templates/config/cable.yml        |  2 +
 lib/solid_cable.rb                            |  4 +-
 lib/solid_cable/batched_broadcaster.rb        | 73 +++++++++++++++++++
 lib/solid_cable/configuration.rb              | 11 ++-
 .../subscription_adapter/solid_cable_test.rb  |  9 +++
 .../solid_cable/batched_broadcaster_test.rb   | 50 +++++++++++++
 8 files changed, 166 insertions(+), 4 deletions(-)
 create mode 100644 lib/solid_cable/batched_broadcaster.rb
 create mode 100644 test/lib/solid_cable/batched_broadcaster_test.rb

diff --git a/app/models/solid_cable/message.rb b/app/models/solid_cable/message.rb
index 398af1b..4a16d70 100644
--- a/app/models/solid_cable/message.rb
+++ b/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)
diff --git a/lib/action_cable/subscription_adapter/solid_cable.rb b/lib/action_cable/subscription_adapter/solid_cable.rb
index e1b9f43..df14d0b 100644
--- a/lib/action_cable/subscription_adapter/solid_cable.rb
+++ b/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)
diff --git a/lib/generators/solid_cable/install/templates/config/cable.yml b/lib/generators/solid_cable/install/templates/config/cable.yml
index b9adc5a..8db79af 100644
--- a/lib/generators/solid_cable/install/templates/config/cable.yml
+++ b/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
diff --git a/lib/solid_cable.rb b/lib/solid_cable.rb
index 2694896..03c434f 100644
--- a/lib/solid_cable.rb
+++ b/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"))
diff --git a/lib/solid_cable/batched_broadcaster.rb b/lib/solid_cable/batched_broadcaster.rb
new file mode 100644
index 0000000..fe3a1fa
--- /dev/null
+++ b/lib/solid_cable/batched_broadcaster.rb
@@ -0,0 +1,73 @@
+# frozen_string_literal: true
+
+module SolidCable
+  class BatchedBroadcaster
+    Stopped = Class.new(StandardError)
+    Message = Struct.new(:channel, :payload, :enqueued_at, 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:, enqueued_at: monotonic_time)
+
+      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
+
+      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 = first_message.enqueued_at + batch_delay
+
+        while batch.size < batch_size && (remaining = deadline - monotonic_time).positive?
+          message = queue.pop(timeout: remaining)
+          break if message.nil?
+
+          batch << message
+        end
+
+        flush(batch)
+      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
diff --git a/lib/solid_cable/configuration.rb b/lib/solid_cable/configuration.rb
index 236f640..f29922a 100644
--- a/lib/solid_cable/configuration.rb
+++ b/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
 
diff --git a/test/lib/action_cable/subscription_adapter/solid_cable_test.rb b/test/lib/action_cable/subscription_adapter/solid_cable_test.rb
index 4e1e0c3..87e7899 100644
--- a/test/lib/action_cable/subscription_adapter/solid_cable_test.rb
+++ b/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
diff --git a/test/lib/solid_cable/batched_broadcaster_test.rb b/test/lib/solid_cable/batched_broadcaster_test.rb
new file mode 100644
index 0000000..03edaef
--- /dev/null
+++ b/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

From dda6908b4b374c93d901087a7ae3179695bf9eab Mon Sep 17 00:00:00 2001
From: Nick Pezza <pezza@hey.com>
Date: Fri, 14 Aug 2026 21:57:48 -0400
Subject: [PATCH 2/5] use sized queue

---
 lib/solid_cable.rb                     | 2 +-
 lib/solid_cable/batched_broadcaster.rb | 4 ++--
 lib/solid_cable/configuration.rb       | 4 ++++
 3 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/lib/solid_cable.rb b/lib/solid_cable.rb
index 03c434f..40b9d07 100644
--- a/lib/solid_cable.rb
+++ b/lib/solid_cable.rb
@@ -11,7 +11,7 @@ class << self
     delegate :connects_to, :silence_polling?, :polling_interval,
       :message_retention, :autotrim?, :trim_batch_size, :use_skip_locked,
       :trim_chance, :reconnect_attempts, :writer_batch_size, :writer_batch_delay,
-      to: :configuration
+      :writer_queue_size, to: :configuration
 
     def configuration
       @configuration ||= Configuration.new(**Rails.application.config_for("cable"))
diff --git a/lib/solid_cable/batched_broadcaster.rb b/lib/solid_cable/batched_broadcaster.rb
index fe3a1fa..4061242 100644
--- a/lib/solid_cable/batched_broadcaster.rb
+++ b/lib/solid_cable/batched_broadcaster.rb
@@ -5,10 +5,10 @@ class BatchedBroadcaster
     Stopped = Class.new(StandardError)
     Message = Struct.new(:channel, :payload, :enqueued_at, keyword_init: true)
 
-    def initialize(batch_size: SolidCable.writer_batch_size, batch_delay: SolidCable.writer_batch_delay)
+    def initialize(queue_size: SolidCable.writer_queue_size, batch_size: SolidCable.writer_batch_size, batch_delay: SolidCable.writer_batch_delay)
       @batch_size = batch_size
       @batch_delay = batch_delay
-      @queue = Queue.new
+      @queue = SizedQueue.new(queue_size)
 
       @thread = Thread.new do
         Thread.current.name = "solid_cable_writer"
diff --git a/lib/solid_cable/configuration.rb b/lib/solid_cable/configuration.rb
index f29922a..13146ae 100644
--- a/lib/solid_cable/configuration.rb
+++ b/lib/solid_cable/configuration.rb
@@ -75,6 +75,10 @@ def writer_batch_delay
         [ parse_duration(options.writer_batch_delay, default: 0.001.seconds), 0 ].max
     end
 
+    def writer_queue_size
+      @writer_queue_size ||= options.writer_queue_size.to_i
+    end
+
     private
       attr_reader :options
 

From 5e9fb087920512fa91a371e2cfdd9b9da6765f5e Mon Sep 17 00:00:00 2001
From: Nick Pezza <pezza@hey.com>
Date: Sat, 15 Aug 2026 09:11:10 -0400
Subject: [PATCH 3/5] switch to sync when needed

---
 lib/solid_cable/batched_broadcaster.rb | 15 ++++++++++++---
 1 file changed, 12 insertions(+), 3 deletions(-)

diff --git a/lib/solid_cable/batched_broadcaster.rb b/lib/solid_cable/batched_broadcaster.rb
index 4061242..16a00e8 100644
--- a/lib/solid_cable/batched_broadcaster.rb
+++ b/lib/solid_cable/batched_broadcaster.rb
@@ -3,12 +3,13 @@
 module SolidCable
   class BatchedBroadcaster
     Stopped = Class.new(StandardError)
-    Message = Struct.new(:channel, :payload, :enqueued_at, keyword_init: true)
+    Message = Struct.new(:channel, :payload, :enqueued_at, :completed, keyword_init: true)
 
     def initialize(queue_size: SolidCable.writer_queue_size, batch_size: SolidCable.writer_batch_size, batch_delay: SolidCable.writer_batch_delay)
       @batch_size = batch_size
       @batch_delay = batch_delay
-      @queue = SizedQueue.new(queue_size)
+      @queue_size = queue_size
+      @queue = Queue.new
 
       @thread = Thread.new do
         Thread.current.name = "solid_cable_writer"
@@ -19,8 +20,12 @@ def initialize(queue_size: SolidCable.writer_queue_size, batch_size: SolidCable.
 
     def broadcast(channel, payload)
       message = Message.new(channel:, payload:, enqueued_at: monotonic_time)
+      wait_for_commit = queue.size + 1 > queue_size
+      message.completed = Concurrent::Event.new if wait_for_commit
 
       queue.enq message
+
+      message.completed.wait if wait_for_commit
     rescue ClosedQueueError
       raise Stopped, "Solid Cable writer has stopped"
     end
@@ -31,7 +36,7 @@ def shutdown
     end
 
     private
-      attr_reader :batch_size, :batch_delay, :queue, :thread
+      attr_reader :batch_size, :batch_delay, :queue, :thread, :queue_size
 
       def listen_for_initial_messages
         loop do
@@ -64,6 +69,10 @@ def flush(batch)
         end
       rescue StandardError => error
         Rails.error.report(error)
+      ensure
+        batch.each do |message|
+          message.completed&.set
+        end
       end
 
       def monotonic_time

From 7471a7cd0c922883892507c2dc3ea56c72293aa0 Mon Sep 17 00:00:00 2001
From: Nick Pezza <pezza@hey.com>
Date: Sat, 15 Aug 2026 09:42:54 -0400
Subject: [PATCH 4/5] be greedy

---
 lib/solid_cable/batched_broadcaster.rb | 21 ++++++++-------------
 1 file changed, 8 insertions(+), 13 deletions(-)

diff --git a/lib/solid_cable/batched_broadcaster.rb b/lib/solid_cable/batched_broadcaster.rb
index 16a00e8..4470cdd 100644
--- a/lib/solid_cable/batched_broadcaster.rb
+++ b/lib/solid_cable/batched_broadcaster.rb
@@ -3,12 +3,11 @@
 module SolidCable
   class BatchedBroadcaster
     Stopped = Class.new(StandardError)
-    Message = Struct.new(:channel, :payload, :enqueued_at, :completed, keyword_init: true)
+    Message = Struct.new(:channel, :payload, keyword_init: true)
 
-    def initialize(queue_size: SolidCable.writer_queue_size, batch_size: SolidCable.writer_batch_size, batch_delay: SolidCable.writer_batch_delay)
+    def initialize(batch_size: SolidCable.writer_batch_size, batch_delay: SolidCable.writer_batch_delay)
       @batch_size = batch_size
       @batch_delay = batch_delay
-      @queue_size = queue_size
       @queue = Queue.new
 
       @thread = Thread.new do
@@ -19,13 +18,9 @@ def initialize(queue_size: SolidCable.writer_queue_size, batch_size: SolidCable.
     end
 
     def broadcast(channel, payload)
-      message = Message.new(channel:, payload:, enqueued_at: monotonic_time)
-      wait_for_commit = queue.size + 1 > queue_size
-      message.completed = Concurrent::Event.new if wait_for_commit
+      message = Message.new(channel:, payload:)
 
       queue.enq message
-
-      message.completed.wait if wait_for_commit
     rescue ClosedQueueError
       raise Stopped, "Solid Cable writer has stopped"
     end
@@ -50,7 +45,11 @@ def listen_for_initial_messages
 
       def collect_batch(first_message)
         batch = [ first_message ]
-        deadline = first_message.enqueued_at + batch_delay
+        deadline = monotonic_time + batch_delay
+
+        while batch.size < batch_size && (message = queue.pop(timeout: 0))
+          batch << message
+        end
 
         while batch.size < batch_size && (remaining = deadline - monotonic_time).positive?
           message = queue.pop(timeout: remaining)
@@ -69,10 +68,6 @@ def flush(batch)
         end
       rescue StandardError => error
         Rails.error.report(error)
-      ensure
-        batch.each do |message|
-          message.completed&.set
-        end
       end
 
       def monotonic_time

From 17539f0b46e5180f8764c9fa3d7e77f2ea9ac507 Mon Sep 17 00:00:00 2001
From: Nick Pezza <pezza@hey.com>
Date: Sat, 15 Aug 2026 10:45:36 -0400
Subject: [PATCH 5/5] cleanup

---
 lib/solid_cable.rb                     |  2 +-
 lib/solid_cable/batched_broadcaster.rb | 11 +++++++++--
 lib/solid_cable/configuration.rb       |  4 ----
 3 files changed, 10 insertions(+), 7 deletions(-)

diff --git a/lib/solid_cable.rb b/lib/solid_cable.rb
index 40b9d07..03c434f 100644
--- a/lib/solid_cable.rb
+++ b/lib/solid_cable.rb
@@ -11,7 +11,7 @@ class << self
     delegate :connects_to, :silence_polling?, :polling_interval,
       :message_retention, :autotrim?, :trim_batch_size, :use_skip_locked,
       :trim_chance, :reconnect_attempts, :writer_batch_size, :writer_batch_delay,
-      :writer_queue_size, to: :configuration
+      to: :configuration
 
     def configuration
       @configuration ||= Configuration.new(**Rails.application.config_for("cable"))
diff --git a/lib/solid_cable/batched_broadcaster.rb b/lib/solid_cable/batched_broadcaster.rb
index 4470cdd..53f2ec8 100644
--- a/lib/solid_cable/batched_broadcaster.rb
+++ b/lib/solid_cable/batched_broadcaster.rb
@@ -47,18 +47,25 @@ 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
-
-        flush(batch)
       end
 
       def flush(batch)
diff --git a/lib/solid_cable/configuration.rb b/lib/solid_cable/configuration.rb
index 13146ae..f29922a 100644
--- a/lib/solid_cable/configuration.rb
+++ b/lib/solid_cable/configuration.rb
@@ -75,10 +75,6 @@ def writer_batch_delay
         [ parse_duration(options.writer_batch_delay, default: 0.001.seconds), 0 ].max
     end
 
-    def writer_queue_size
-      @writer_queue_size ||= options.writer_queue_size.to_i
-    end
-
     private
       attr_reader :options