feat: run retention on the supervisor

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+- Run retention on the supervisor rather than leaving it configured but
+  unscheduled. Every actor call writes a durable message row, so a policy that
+  nothing invokes let history grow without bound until an application scheduled
+  its own job. `retention_interval` defaults to one hour, and zero disables it.
+  Retention runs on its own thread, so a slow pass cannot delay replacing a
+  crashed role, and a failed pass retries at monitor cadence with a doubling
+  backoff rather than deferring for the whole interval.
 - Add Ruby 4.0 to the compatibility matrix, which now covers Ruby 3.3, 3.4, and
   4.0 against Rails 8.0 and 8.1.
 

docs/roadmap.md

@@ -35,7 +35,9 @@
   graceful caller shutdown, committed state snapshots, and an opt-in Minitest
   helper
 - Supervisor role replacement: a role whose thread dies is restarted until
-  shutdown is requested, and dead process records are pruned on an interval
+  shutdown is requested, and dead process records plus expired message and
+  process history are pruned on their own intervals without an application
+  scheduling its own job
 - SQLite, PostgreSQL, and MySQL integration suites
 - Opt-in cross-process wake-up on PostgreSQL through `WakeUpAdapters.for`, with
   a listening connection per waiting thread and release on supervisor shutdown
@@ -82,13 +84,12 @@
 ## Next milestones
 
 1. Add result lookup by request ID and broader deadlock retry classification.
-2. Add scheduled retention and stale-process maintenance.
-3. Add Turbo append intents and expand reconnect coverage in a full browser.
-4. Add distributed rate limits, global admission hooks, and cache-capacity
+2. Add Turbo append intents and expand reconnect coverage in a full browser.
+3. Add distributed rate limits, global admission hooks, and cache-capacity
    eviction.
-5. Expand security scanning beyond the Brakeman scan, such as dependency
+4. Expand security scanning beyond the Brakeman scan, such as dependency
    auditing and secret scanning.
-6. Benchmark all workloads under documented hardware/database settings and
+5. Benchmark all workloads under documented hardware/database settings and
    publish adapter-specific adoption measurements. Throughput, synchronous
    latency, query counts, and the three reactive delivery paths are measured on
    SQLite; adapter-specific and end-to-end browser measurements are not.

lib/solid_objects/configuration.rb

@@ -22,6 +22,7 @@ class Configuration
     # @rbs @process_alive_threshold: Float
     # @rbs @shutdown_timeout: Float
     # @rbs @supervisor_monitor_interval: Float
+    # @rbs @retention_interval: Float
     # @rbs @dead_process_cleanup_interval: Float
     # @rbs @message_retention: Numeric
     # @rbs @message_retention_by_actor_type: Hash[String, Numeric]
@@ -65,6 +66,7 @@ class Configuration
       :process_alive_threshold,
       :shutdown_timeout,
       :supervisor_monitor_interval,
+      :retention_interval,
       :dead_process_cleanup_interval,
       :message_retention,
       :message_retention_by_actor_type,
@@ -108,6 +110,7 @@ def initialize
       @lock_retry_attempts = 10
       @supervisor_monitor_interval = 1.0
       @dead_process_cleanup_interval = 60.0
+      @retention_interval = 3600.0
       @process_heartbeat_interval = 15.0
       @process_alive_threshold = 60.0
       @shutdown_timeout = 15.0
@@ -144,6 +147,10 @@ def validate!
         raise ArgumentError, "table_name_prefix must contain lowercase letters, digits, and underscores"
       end
 
+      if retention_interval.negative?
+        raise ArgumentError, "retention_interval must not be negative"
+      end
+
       unless supervisor_monitor_interval.positive?
         raise ArgumentError, "supervisor_monitor_interval must be positive"
       end

lib/solid_objects/supervisor.rb

@@ -2,11 +2,14 @@
 
 module SolidObjects
   class Supervisor
+    MAXIMUM_RETENTION_BACKOFF_DOUBLINGS = 16
+
     # @rbs @components: Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor]
     # @rbs @threads: Array[Thread]
     # @rbs @monitor: Thread?
     # @rbs @started: bool
     # @rbs @cleaned_up_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
@@ -26,6 +29,7 @@ def initialize(
       @monitor = nil
       @started = false
       @cleaned_up_at = nil
+      @retention = nil
       @lifecycle = Thread::Mutex.new
     end
 
@@ -44,6 +48,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
 
@@ -57,6 +62,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)
@@ -133,6 +139,76 @@ 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.
+    # @rbs () -> void
+    def retention_loop
+      failures = 0
+      while @started
+        begin
+          prune_expired_records
+          failures = 0
+        rescue => error
+          failures += 1
+          SolidObjects.instrument(
+            :"supervisor.retention_failed",
+            error_class: error.class.name,
+            error_message: error.message
+          )
+        end
+        wait_for_next_retention(failures)
+      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 (Integer) -> void
+    def wait_for_next_retention(failures)
+      deadline = monotonic_now + retention_pause(failures)
+      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
+      return unless SolidObjects.configuration.retention_interval.positive?
+
+      MessagePruner.new.prune
+      ProcessPruner.new.prune
+    end
+
+    # A transient lock or connection error must not defer retention for the
+    # whole interval, so a failed pass retries at monitor cadence. The pause
+    # then doubles per consecutive failure, capped by the interval, so a
+    # database that stays down is not polled once a second forever.
+    # @rbs (Integer) -> Float
+    def retention_pause(failures)
+      interval = SolidObjects.configuration.retention_interval
+      interval = SolidObjects.configuration.supervisor_monitor_interval unless interval.positive?
+      return interval if failures.zero?
+
+      backoff = SolidObjects.configuration.supervisor_monitor_interval *
+        (2**[ failures - 1, MAXIMUM_RETENTION_BACKOFF_DOUBLINGS ].min)
+      [ backoff, interval ].min
+    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/configuration.rbs

@@ -2,10 +2,12 @@
 
 module SolidObjects
   class Configuration
-    @shutdown_timeout: Float
+    @table_name_prefix: String
 
     @supervisor_monitor_interval: Float
 
+    @retention_interval: Float
+
     @dead_process_cleanup_interval: Float
 
     @message_retention: Numeric
@@ -50,8 +52,6 @@ module SolidObjects
 
     @authorize_administration: Proc
 
-    @table_name_prefix: String
-
     @polling_interval: Float
 
     @sync_polling_interval: Float
@@ -86,6 +86,8 @@ module SolidObjects
 
     @process_alive_threshold: Float
 
+    @shutdown_timeout: Float
+
     attr_accessor table_name_prefix: untyped
 
     attr_accessor polling_interval: untyped
@@ -126,6 +128,8 @@ module SolidObjects
 
     attr_accessor supervisor_monitor_interval: untyped
 
+    attr_accessor retention_interval: untyped
+
     attr_accessor dead_process_cleanup_interval: untyped
 
     attr_accessor message_retention: untyped

sig/generated/lib/solid_objects/supervisor.rbs

@@ -2,8 +2,12 @@
 
 module SolidObjects
   class Supervisor
+    MAXIMUM_RETENTION_BACKOFF_DOUBLINGS: ::Integer
+
     @lifecycle: Thread::Mutex
 
+    @retention: Thread?
+
     @cleaned_up_at: Float
 
     @started: bool
@@ -52,6 +56,34 @@ 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.
+    # @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 (Integer) -> void
+    def wait_for_next_retention: (Integer) -> 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
+
+    # A transient lock or connection error must not defer retention for the
+    # whole interval, so a failed pass retries at monitor cadence. The pause
+    # then doubles per consecutive failure, capped by the interval, so a
+    # database that stays down is not polled once a second forever.
+    # @rbs (Integer) -> Float
+    def retention_pause: (Integer) -> Float
+
+    # @rbs () -> void
+    def stop_retention: () -> void
+
     # @rbs () -> void
     def cleanup_dead_processes: () -> void