From 7204bf350d36236d97a868e6fbb56e4ab89790a7 Mon Sep 17 00:00:00 2001 From: Nick Pezza Date: Wed, 19 Aug 2026 19:42:07 -0400 Subject: [PATCH] Bring autotrimming inline with solid cache Since we are now batching writes we can effectively copy solid caches expiry logic. --- README.md | 2 +- app/jobs/solid_cable/trim_job.rb | 16 +------- .../subscription_adapter/solid_cable.rb | 2 - lib/solid_cable.rb | 3 +- lib/solid_cable/batched_broadcaster.rb | 18 ++++++++- lib/solid_cable/configuration.rb | 12 +----- lib/solid_cable/trimming.rb | 23 +++++++++++ test/jobs/trim_job_test.rb | 32 +++++++-------- .../subscription_adapter/solid_cable_test.rb | 18 ++++----- .../solid_cable/batched_broadcaster_test.rb | 39 +++++++++++++++++++ 10 files changed, 106 insertions(+), 59 deletions(-) create mode 100644 lib/solid_cable/trimming.rb diff --git a/README.md b/README.md index d5dc919..8a2032c 100644 --- a/README.md +++ b/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. diff --git a/app/jobs/solid_cable/trim_job.rb b/app/jobs/solid_cable/trim_job.rb index 2b5a183..1950362 100644 --- a/app/jobs/solid_cable/trim_job.rb +++ b/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 diff --git a/lib/action_cable/subscription_adapter/solid_cable.rb b/lib/action_cable/subscription_adapter/solid_cable.rb index df14d0b..2cff1ac 100644 --- a/lib/action_cable/subscription_adapter/solid_cable.rb +++ b/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) diff --git a/lib/solid_cable.rb b/lib/solid_cable.rb index 46737ef..0d62bc8 100644 --- a/lib/solid_cable.rb +++ b/lib/solid_cable.rb @@ -3,6 +3,7 @@ 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" @@ -10,7 +11,7 @@ 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 diff --git a/lib/solid_cable/batched_broadcaster.rb b/lib/solid_cable/batched_broadcaster.rb index 26ec960..3055076 100644 --- a/lib/solid_cable/batched_broadcaster.rb +++ b/lib/solid_cable/batched_broadcaster.rb @@ -1,7 +1,11 @@ # frozen_string_literal: true +require "concurrent" + module SolidCable class BatchedBroadcaster + include Trimming + Stopped = Class.new(StandardError) Message = Struct.new(:channel, :payload, keyword_init: true) @@ -9,6 +13,7 @@ def initialize(batch_size: SolidCable.writer_batch_size, batch_delay: SolidCable @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 diff --git a/lib/solid_cable/configuration.rb b/lib/solid_cable/configuration.rb index 9d36b8d..4b8107b 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, :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 diff --git a/lib/solid_cable/trimming.rb b/lib/solid_cable/trimming.rb new file mode 100644 index 0000000..745ad8a --- /dev/null +++ b/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 diff --git a/test/jobs/trim_job_test.rb b/test/jobs/trim_job_test.rb index e274065..3a59d63 100644 --- a/test/jobs/trim_job_test.rb +++ b/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 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 4b11763..b19bcb2 100644 --- a/test/lib/action_cable/subscription_adapter/solid_cable_test.rb +++ b/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 diff --git a/test/lib/solid_cable/batched_broadcaster_test.rb b/test/lib/solid_cable/batched_broadcaster_test.rb index 03edaef..38ecf52 100644 --- a/test/lib/solid_cable/batched_broadcaster_test.rb +++ b/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")