Bring autotrimming inline with solid cache

README.md

@@ -126,7 +126,7 @@ with Rails 7. Solid Cable raises during boot for that unsupported combination.
 
 ## Trimming
 
-Messages are autotrimmed based upon the `message_retention` setting to determine how long messages are to be kept around. If no `message_retention` is given or parsing fails, it defaults to `1.day`. Messages are trimmed when a messsage is broadcast.
+Messages are autotrimmed based upon the `message_retention` setting to determine how long messages are to be kept around. If no `message_retention` is given or parsing fails, it defaults to `1.day`. For every message written, Solid Cable attempts to trim twice as many expired messages.
 
 Autotrimming can negatively impact performance slightly depending on your workload because it is potentially doing a delete on broadcast. If
 you would prefer, you can disable autotrimming by setting `autotrim: false` and you can manually enqueue the job later, `SolidCable::TrimJob.perform_later`, or run it on a recurring interval out of band.

app/jobs/solid_cable/trim_job.rb

@@ -2,26 +2,12 @@
 
 module SolidCable
   class TrimJob < ActiveJob::Base
-    def perform
-      return unless trim?
-
+    def perform(trim_batch_size: ::SolidCable.trim_batch_size)
       ::SolidCable::Message.transaction do
         ids = ::SolidCable::Message.trimmable.non_blocking_lock.
               limit(trim_batch_size).pluck(:id)
         ::SolidCable::Message.where(id: ids).delete_all
       end
     end
-
-    private
-      def trim_batch_size
-        ::SolidCable.trim_batch_size
-      end
-
-      def trim?
-        expires_per_write = (1 / trim_batch_size.to_f) * ::SolidCable.trim_chance
-
-        !::SolidCable.autotrim? ||
-          rand < (expires_per_write - expires_per_write.floor)
-      end
   end
 end

lib/action_cable/subscription_adapter/solid_cable.rb

@@ -25,8 +25,6 @@ def initialize(*)
 
       def broadcast(channel, payload)
         broadcaster.broadcast(channel, payload)
-
-        ::SolidCable::TrimJob.perform_now if ::SolidCable.autotrim?
       end
 
       def subscribe(channel, callback, success_callback = nil)

lib/solid_cable.rb

@@ -3,14 +3,15 @@
 require "solid_cable/version"
 require "solid_cable/engine"
 require "solid_cable/configuration"
+require "solid_cable/trimming"
 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, :writer_batch_size, :writer_batch_delay,
+      :reconnect_attempts, :writer_batch_size, :writer_batch_delay,
       :encrypt?, :encryption_context_properties,
       to: :configuration
 

lib/solid_cable/batched_broadcaster.rb

@@ -1,14 +1,19 @@
 # frozen_string_literal: true
 
+require "concurrent"
+
 module SolidCable
   class BatchedBroadcaster
+    include Trimming
+
     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
+      @background = Concurrent::FixedThreadPool.new(1, max_queue: 100, fallback_policy: :discard)
 
       @thread = Thread.new do
         Thread.current.name = "solid_cable_writer"
@@ -28,10 +33,12 @@ def broadcast(channel, payload)
     def shutdown
       queue.close
       thread.join
+      background.shutdown
+      background.wait_for_termination
     end
 
     private
-      attr_reader :batch_size, :batch_delay, :queue, :thread
+      attr_reader :batch_size, :batch_delay, :queue, :thread, :background
 
       def listen_for_initial_messages
         loop do
@@ -72,11 +79,20 @@ def flush(batch)
         Rails.application.executor.wrap do
           SolidCable::Message.
             broadcast_batch(batch.map { |message| [ message.channel, message.payload ] })
+          track_writes(batch.size) if SolidCable.autotrim?
         end
       rescue StandardError => error
         Rails.error.report(error)
       end
 
+      def async(&block)
+        background << -> do
+          Rails.application.executor.wrap(&block)
+        rescue Exception => error # rubocop:disable Lint/RescueException
+          Rails.error.report(error)
+        end
+      end
+
       def monotonic_time
         Process.clock_gettime(Process::CLOCK_MONOTONIC)
       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, :writer_batch_size, :writer_batch_delay,
+      :reconnect_attempts, :writer_batch_size, :writer_batch_delay,
       :encrypt, :encryption_context_properties
 
     def connects_to
@@ -49,16 +49,6 @@ def use_skip_locked
       @use_skip_locked = options.use_skip_locked != false
     end
 
-    # For every write that we do, we attempt to delete trim_chance times as
-    # many records. This ensures there is downward pressure on the cache size
-    # while there is valid data to delete. Read this as 'every time the trim job
-    # runs theres a trim_multiplier chance this trims'. Adjust number to make it
-    # more or less likely to trim. Only works like this if trim_batch_size is
-    # 100
-    def trim_chance
-      2
-    end
-
     def reconnect_attempts
       @reconnect_attempts ||= begin
         attempts = options[:reconnect_attempts] || 1

lib/solid_cable/trimming.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+module SolidCable
+  module Trimming
+    # For every write that we do, we attempt to delete TRIM_MULTIPLIER times as
+    # many records. This ensures there is downward pressure on the message count
+    # while there is old data to delete.
+    TRIM_MULTIPLIER = 2
+
+    private
+      def track_writes(count)
+        trim_batches(count).times { async { TrimJob.perform_now } }
+      end
+
+      def trim_batches(count)
+        trims_per_write = (1 / SolidCable.trim_batch_size.to_f) * TRIM_MULTIPLIER
+        batches = (count * trims_per_write).floor
+        overflow_batch_chance = count * trims_per_write - batches
+        batches += 1 if rand < overflow_batch_chance
+        batches
+      end
+  end
+end

test/jobs/trim_job_test.rb

@@ -7,31 +7,27 @@ class TrimJobTest < ActiveJob::TestCase
   include ConfigStubs
 
   test "trims a limited number of messages" do
-    SolidCable.stub(:trim_chance, 99.999) do
-      with_cable_config trim_batch_size: 2, message_rention: "1.second" do
-        4.times do
-          SolidCable::Message.broadcast("foo", "bar")
-          SolidCable::Message.update_all(created_at: 2.days.ago)
-        end
+    with_cable_config trim_batch_size: 2, message_retention: "1.second" do
+      4.times do
+        SolidCable::Message.broadcast("foo", "bar")
+        SolidCable::Message.update_all(created_at: 2.days.ago)
+      end
 
-        assert_difference -> { SolidCable::Message.count }, -2 do
-          SolidCable::TrimJob.perform_now
-        end
+      assert_difference -> { SolidCable::Message.count }, -2 do
+        SolidCable::TrimJob.perform_now
       end
     end
   end
 
   test "trims when out of band with autotrim disabled" do
-    SolidCable.stub(:trim_chance, 0) do
-      with_cable_config autotrim: false, trim_batch_size: 2, message_rention: "1.second" do
-        4.times do
-          SolidCable::Message.broadcast("foo", "bar")
-          SolidCable::Message.update_all(created_at: 2.days.ago)
-        end
+    with_cable_config autotrim: false, trim_batch_size: 2, message_retention: "1.second" do
+      4.times do
+        SolidCable::Message.broadcast("foo", "bar")
+        SolidCable::Message.update_all(created_at: 2.days.ago)
+      end
 
-        assert_difference -> { SolidCable::Message.count }, -2 do
-          SolidCable::TrimJob.perform_now
-        end
+      assert_difference -> { SolidCable::Message.count }, -2 do
+        SolidCable::TrimJob.perform_now
       end
     end
   end

test/lib/action_cable/subscription_adapter/solid_cable_test.rb

@@ -92,18 +92,16 @@ class ActionCable::SubscriptionAdapter::SolidCableTest < ActionCable::TestCase
   end
 
   test "trims_after_unsubscribe" do
-    SolidCable.stub(:trim_chance, 99.999999) do
-      with_cable_config message_retention: "2.seconds", trim_batch_size: 2 do
-        subscribe_as_queue("channel") do |queue|
-          4.times do
-            @tx_adapter.broadcast("channel", "hello world")
-            sleep 3
-          end
-
-          queue.clear
+    with_cable_config message_retention: "2.seconds", trim_batch_size: 2 do
+      subscribe_as_queue("channel") do |queue|
+        4.times do
+          @tx_adapter.broadcast("channel", "hello world")
+          sleep 3
         end
-        assert_equal 1, SolidCable::Message.where(channel: "channel").count
+
+        queue.clear
       end
+      assert_equal 1, SolidCable::Message.where(channel: "channel").count
     end
   end
 

test/lib/solid_cable/batched_broadcaster_test.rb

@@ -1,8 +1,11 @@
 # frozen_string_literal: true
 
 require "test_helper"
+require "config_stubs"
 
 class SolidCable::BatchedBroadcasterTest < ActiveSupport::TestCase
+  include ConfigStubs
+
   teardown do
     @broadcaster&.shutdown
   end
@@ -33,6 +36,42 @@ class SolidCable::BatchedBroadcasterTest < ActiveSupport::TestCase
     end
   end
 
+  test "trims in proportion to the number of messages written" do
+    trims = Queue.new
+
+    with_cable_config trim_batch_size: 2 do
+      @broadcaster = SolidCable::BatchedBroadcaster.new(batch_size: 2, batch_delay: 1)
+
+      SolidCable::Message.stub(:broadcast_batch, nil) do
+        SolidCable::TrimJob.stub(:perform_now, -> { trims << true }) do
+          @broadcaster.broadcast("one", "first")
+          @broadcaster.broadcast("two", "second")
+          @broadcaster.shutdown
+        end
+      end
+    end
+
+    assert_equal 2, trims.size
+  end
+
+  test "trims asynchronously" do
+    write_threads = Queue.new
+    trim_threads = Queue.new
+
+    with_cable_config trim_batch_size: 2 do
+      @broadcaster = SolidCable::BatchedBroadcaster.new(batch_size: 1, batch_delay: 0)
+
+      SolidCable::Message.stub(:broadcast_batch, ->(*) { write_threads << Thread.current }) do
+        SolidCable::TrimJob.stub(:perform_now, -> { trim_threads << Thread.current }) do
+          @broadcaster.broadcast("channel", "payload")
+          @broadcaster.shutdown
+        end
+      end
+    end
+
+    assert_not_same write_threads.pop, trim_threads.pop
+  end
+
   test "reports write errors" do
     @broadcaster = SolidCable::BatchedBroadcaster.new(batch_size: 1, batch_delay: 0)
     write_error = RuntimeError.new("write failed")