fix: run retention on its own thread

Greptile flagged two defects in the scheduled retention pass.

A failed pass suppressed retries: the timestamp was recorded before
pruning ran, so a transient lock or connection error deferred retention
for the full interval, up to an hour by default. Retention now retries on
the next tick, and the failure is instrumented as supervisor.retention_failed.

Retention also shared the monitor thread with role replacement. Pruning
is unbounded work against a table that only grows, so a slow pass held
back the supervision that keeps a crashed role alive. It now runs on a
dedicated thread. The pause between passes is taken in monitor-interval
steps so shutdown does not wait out an hour-long sleep.

Author: cardmagic

lib/solid_objects/supervisor.rb

@@ -7,7 +7,7 @@ class Supervisor
     # @rbs @monitor: Thread?
     # @rbs @started: bool
     # @rbs @cleaned_up_at: Float
-    # @rbs @pruned_at: Float
+    # @rbs @retention: Thread?
     # @rbs @lifecycle: Thread::Mutex
 
     # @rbs (?worker_count: Integer, ?effect_worker_count: Integer, ?broadcast_worker_count: Integer, ?reminder_scheduler_count: Integer) -> void
@@ -27,7 +27,7 @@ def initialize(
       @monitor = nil
       @started = false
       @cleaned_up_at = nil
-      @pruned_at = nil
+      @retention = nil
       @lifecycle = Thread::Mutex.new
     end
 
@@ -46,6 +46,7 @@ def start
       @started = true
       @threads = components.map { |component| supervise(component) }
       @monitor = Thread.new { monitor_loop }
+      @retention = Thread.new { retention_loop }
       SolidObjects.instrument(:"supervisor.started", component_count: components.length)
     end
 
@@ -59,6 +60,7 @@ def stop
         # list, or never starts.
         @lifecycle.synchronize { @started = false }
         stop_monitor
+        stop_retention
         components.each(&:request_shutdown)
         join_until_timeout
         components.reject(&:stopped?).each(&:stop)
@@ -86,7 +88,6 @@ def monitor_loop
         begin
           replace_dead_roles
           cleanup_dead_processes
-          prune_expired_records
         rescue => error
           SolidObjects.instrument(
             :"supervisor.monitor_failed",
@@ -136,21 +137,67 @@ def thread_error(thread)
       error.class.name
     end
 
+    # Retention gets its own thread rather than sharing the monitor's. A large
+    # backlog or a lock wait can make a pass slow, and role replacement must not
+    # wait behind housekeeping. A failed pass retries on the next tick instead of
+    # deferring for the whole interval.
+    # @rbs () -> void
+    def retention_loop
+      while @started
+        begin
+          prune_expired_records
+        rescue => error
+          SolidObjects.instrument(
+            :"supervisor.retention_failed",
+            error_class: error.class.name,
+            error_message: error.message
+          )
+        end
+        wait_for_next_retention
+      end
+    end
+
+    # Sleeping the whole interval would make shutdown wait out an hour-long
+    # nap, so the pause is taken in short steps that notice a stop request.
+    # @rbs () -> void
+    def wait_for_next_retention
+      deadline = monotonic_now + retention_pause
+      step = SolidObjects.configuration.supervisor_monitor_interval
+      while @started && monotonic_now < deadline
+        sleep [ step, deadline - monotonic_now ].min
+      end
+    end
+
     # Every actor call writes a durable message row, so retention that is only
     # configured and never run leaves those rows to grow without bound. The
     # supervisor runs it rather than requiring every application to schedule
     # its own job.
     # @rbs () -> void
     def prune_expired_records
-      interval = SolidObjects.configuration.retention_interval
-      return unless interval.positive?
-      return if @pruned_at && monotonic_now - @pruned_at < interval
+      return unless SolidObjects.configuration.retention_interval.positive?
 
-      @pruned_at = monotonic_now
       MessagePruner.new.prune
       ProcessPruner.new.prune
     end
 
+    # @rbs () -> Float
+    def retention_pause
+      interval = SolidObjects.configuration.retention_interval
+      return interval if interval.positive?
+
+      SolidObjects.configuration.supervisor_monitor_interval
+    end
+
+    # @rbs () -> void
+    def stop_retention
+      retention = @retention
+      @retention = nil
+      return unless retention
+
+      retention.join(SolidObjects.configuration.shutdown_timeout)
+      retention.kill if retention.alive?
+    end
+
     # @rbs () -> void
     def cleanup_dead_processes
       interval = SolidObjects.configuration.dead_process_cleanup_interval

sig/generated/lib/solid_objects/supervisor.rbs

@@ -4,7 +4,7 @@ module SolidObjects
   class Supervisor
     @lifecycle: Thread::Mutex
 
-    @pruned_at: Float
+    @retention: Thread?
 
     @cleaned_up_at: Float
 
@@ -54,13 +54,31 @@ module SolidObjects
     # @rbs (Thread?) -> String?
     def thread_error: (Thread?) -> String?
 
+    # Retention gets its own thread rather than sharing the monitor's. A large
+    # backlog or a lock wait can make a pass slow, and role replacement must not
+    # wait behind housekeeping. A failed pass retries on the next tick instead of
+    # deferring for the whole interval.
+    # @rbs () -> void
+    def retention_loop: () -> void
+
+    # Sleeping the whole interval would make shutdown wait out an hour-long
+    # nap, so the pause is taken in short steps that notice a stop request.
+    # @rbs () -> void
+    def wait_for_next_retention: () -> void
+
     # Every actor call writes a durable message row, so retention that is only
     # configured and never run leaves those rows to grow without bound. The
     # supervisor runs it rather than requiring every application to schedule
     # its own job.
     # @rbs () -> void
     def prune_expired_records: () -> void
 
+    # @rbs () -> Float
+    def retention_pause: () -> Float
+
+    # @rbs () -> void
+    def stop_retention: () -> void
+
     # @rbs () -> void
     def cleanup_dead_processes: () -> void
 

test/integration/scheduled_retention_test.rb

@@ -16,7 +16,32 @@ def stop = @stopped = true
     def stopped? = @stopped
   end
 
+  class CrashOnceRole < IdleRole
+    class << self
+      attr_accessor :shared_runs
+    end
+
+    def initialize
+      self.class.shared_runs ||= 0
+    end
+
+    def runs = self.class.shared_runs
+
+    # Crashes only after the first monitor pass has begun, so replacement is
+    # needed while a slow retention pass is already running.
+    def run
+      self.class.shared_runs += 1
+      if self.class.shared_runs == 1
+        sleep 0.1
+        raise "role crashed"
+      end
+
+      super
+    end
+  end
+
   setup do
+    CrashOnceRole.shared_runs = nil
     SolidObjects.configuration.supervisor_monitor_interval = 0.02
     SolidObjects.configuration.dead_process_cleanup_interval = 0
     SolidObjects.configuration.retention_interval = 0.05
@@ -83,25 +108,41 @@ def stopped? = @stopped
     ActiveSupport::Notifications.unsubscribe(subscription) if subscription
   end
 
-  test "a failing retention pass does not stop the supervisor" do
+  test "a failing retention pass retries rather than deferring for the interval" do
     failures = []
-    subscription = ActiveSupport::Notifications.subscribe("solid_objects.supervisor.monitor_failed") do
+    subscription = ActiveSupport::Notifications.subscribe("solid_objects.supervisor.retention_failed") do
       failures << true
     end
     role = IdleRole.new
     @supervisor = supervisor(role)
     @supervisor.define_singleton_method(:prune_expired_records) { raise "boom" }
 
     @supervisor.start
-    Timeout.timeout(10) { sleep 0.02 until failures.any? }
-    sleep 0.1
+    Timeout.timeout(10) { sleep 0.02 until failures.length >= 2 }
 
     assert @supervisor.instance_variable_get(:@monitor).alive?,
       "the monitor should survive a failing retention pass"
+    assert_operator failures.length, :>=, 2,
+      "a failed pass should retry on the next tick, not wait out the interval"
   ensure
     ActiveSupport::Notifications.unsubscribe(subscription) if subscription
   end
 
+  # Role replacement must not wait behind housekeeping.
+  test "a slow retention pass does not block role replacement" do
+    SolidObjects.configuration.retention_interval = 0.02
+    role = CrashOnceRole.new
+    @supervisor = supervisor(role)
+    @supervisor.define_singleton_method(:prune_expired_records) { sleep 10 }
+
+    @supervisor.start
+
+    # Shorter than the pruning pass, so sharing a thread with it fails here.
+    Timeout.timeout(3) { sleep 0.02 until role.runs >= 2 }
+    assert_operator role.runs, :>=, 2,
+      "a crashed role should be replaced while retention is still running"
+  end
+
   test "rejects a negative retention interval" do
     SolidObjects.configuration.retention_interval = -1