fix: Bound recovery scans and verify crash safety

Author: cardmagic

benchmark/support.rb

@@ -428,9 +428,11 @@ def migrate
       require_relative "../db/migrate/20260805000000_create_solid_objects_tables"
       require_relative "../db/migrate/20260806000000_add_state_revision_to_solid_objects_instances"
       require_relative "../db/migrate/20260813000000_rename_message_dispatch_columns"
+      require_relative "../db/migrate/20260915000000_add_solid_objects_effect_recoveries"
       CreateSolidObjectsTables.new.migrate(:up)
       AddStateRevisionToSolidObjectsInstances.new.migrate(:up)
       RenameMessageDispatchColumns.new.migrate(:up)
+      AddSolidObjectsEffectRecoveries.new.migrate(:up)
     end
 
     # @rbs () -> void
@@ -444,6 +446,7 @@ def load_models
         claimed_message
         reminder
         effect
+        effect_recovery
         broadcast
         dead_letter
       ].each do |model|

docs/effect-recovery.md

@@ -149,6 +149,13 @@ Mailbox insertion reuses the instance lock already held by the decision.
 Multiple checks in one actor commit lock all their effects and bindings before
 locking any processes. Unlocked candidate reads are hints, never decisions.
 
+Automatic passes prefilter owner freshness and the effective per-effect timeout
+using database time, and visit at most `claim_scan_limit` stale candidates. Fresh
+actors and owners are not locked, including owners protected by extended grace.
+Remaining stale effects are revisited on later polls. Every candidate still
+undergoes the authoritative locked recheck, and successful retirement announces
+the committed mailbox work through the existing wake-up mechanism.
+
 The decision samples database wall time after obtaining the owner lock. The
 effective timeout is the larger of the runtime's `process_alive_threshold` and
 the effect's persisted `recovery_timeout`, in seconds. A heartbeat newer than

examples/at_least_once/boot.rb

@@ -23,7 +23,7 @@ def self.call(database_path)
     require "solid_objects/database_adapter"
     %w[
       record process instance message ready_message claimed_message
-      reminder effect broadcast dead_letter
+      reminder effect effect_recovery broadcast dead_letter
     ].each { |model| require File.join(ROOT, "app/models/solid_objects", model) }
 
     SolidObjects.configuration.authorize_message = ->(**) { true }
@@ -40,8 +40,10 @@ def self.migrate
     require File.join(ROOT, "db/migrate/20260805000000_create_solid_objects_tables")
     require File.join(ROOT, "db/migrate/20260806000000_add_state_revision_to_solid_objects_instances")
     require File.join(ROOT, "db/migrate/20260813000000_rename_message_dispatch_columns")
+    require File.join(ROOT, "db/migrate/20260915000000_add_solid_objects_effect_recoveries")
     CreateSolidObjectsTables.new.migrate(:up)
     AddStateRevisionToSolidObjectsInstances.new.migrate(:up)
     RenameMessageDispatchColumns.new.migrate(:up)
+    AddSolidObjectsEffectRecoveries.new.migrate(:up)
   end
 end

lib/solid_objects/actor.rb

@@ -185,12 +185,7 @@ def emit(name, on_success: nil, on_failure: nil, on_recovery: nil, on_status: ni
       validate_effect_callback!(on_failure)
       validate_effect_callback!(on_recovery)
       validate_effect_callback!(on_status)
-      unless recovery_timeout.nil?
-        unless recovery_timeout.is_a?(Numeric) && recovery_timeout.real? && recovery_timeout.to_f.finite? && recovery_timeout.positive?
-          raise ArgumentError, "recovery_timeout must be a positive finite duration in seconds"
-        end
-        raise ArgumentError, "recovery_timeout requires on_recovery" unless on_recovery
-      end
+      validate_recovery_timeout!(timeout: recovery_timeout, operation: on_recovery)
       effect_id = SecureRandom.uuid
       EffectIntent.new(
         effect_id:,
@@ -446,5 +441,15 @@ def validate_effect_callback!(operation)
 
       raise UnknownMessage, "unknown effect callback operation #{operation.inspect}"
     end
+
+    # @rbs (timeout: Numeric?, operation: String | Symbol?) -> void
+    def validate_recovery_timeout!(timeout:, operation:)
+      return if timeout.nil?
+
+      unless timeout.is_a?(Numeric) && timeout.real? && timeout.to_f.finite? && timeout.to_f.positive?
+        raise ArgumentError, "recovery_timeout must be a positive finite duration in seconds"
+      end
+      raise ArgumentError, "recovery_timeout requires on_recovery" unless operation
+    end
   end
 end

lib/solid_objects/effect_recovery_coordinator.rb

@@ -35,9 +35,7 @@ def check(instance:, intents:)
 
     # @rbs () -> void
     def recover_available
-      candidates = EffectRecovery.where(retired_at: nil).where.not(recovery_operation: nil)
-        .where(effect_id: Effect.where(status: "processing").select(:effect_id))
-      candidates.find_each do |candidate|
+      recovery_candidates.each do |candidate|
         notification = SolidObjects.database_adapter.transaction do
           instance = Instance.lock.find_by(id: candidate.instance_id)
           next unless instance
@@ -60,6 +58,26 @@ def recover_available
 
     private
 
+    # @rbs () -> ActiveRecord::Relation[EffectRecovery]
+    def recovery_candidates
+      effects = Effect.table_name
+      owners = Process.table_name
+      bindings = EffectRecovery.table_name
+      heartbeat = case DatabaseAdapter.family(Record.connection)
+      when :postgresql then "EXTRACT(EPOCH FROM #{owners}.last_heartbeat_at)"
+      when :mysql then "UNIX_TIMESTAMP(#{owners}.last_heartbeat_at)"
+      else "CAST(STRFTIME('%s', #{owners}.last_heartbeat_at) AS REAL)"
+      end
+      now = SolidObjects.database_adapter.database_clock_now.to_f
+      threshold = SolidObjects.configuration.process_alive_threshold
+      EffectRecovery.joins("INNER JOIN #{effects} ON #{effects}.effect_id = #{bindings}.effect_id")
+        .joins("LEFT JOIN #{owners} ON #{owners}.id = #{effects}.claimed_by")
+        .where(retired_at: nil).where.not(recovery_operation: nil)
+        .where("#{effects}.status = ?", "processing")
+        .where("#{owners}.id IS NULL OR #{heartbeat} <= ? - CASE WHEN #{bindings}.recovery_timeout > ? THEN #{bindings}.recovery_timeout ELSE ? END", now, threshold, threshold)
+        .order(:effect_id).limit(SolidObjects.configuration.claim_scan_limit)
+    end
+
     # @rbs (instance: Instance, intent: Actor::EffectRecoveryIntent, recovery: EffectRecovery, effect: Effect?, owners: Hash[String, Process], now: Time) -> void
     def check_one(instance:, intent:, recovery:, effect:, owners:, now:)
       key = "effect:#{intent.effect_id}:check:#{intent.request_id}"

sig/generated/lib/solid_objects/actor.rbs

@@ -277,5 +277,8 @@ module SolidObjects
 
     # @rbs (Symbol | String?) -> void
     def validate_effect_callback!: (Symbol | String?) -> void
+
+    # @rbs (timeout: Numeric?, operation: String | Symbol?) -> void
+    def validate_recovery_timeout!: (timeout: Numeric?, operation: String | Symbol?) -> void
   end
 end

sig/generated/lib/solid_objects/effect_recovery_coordinator.rbs

@@ -26,6 +26,9 @@ module SolidObjects
 
     private
 
+    # @rbs () -> ActiveRecord::Relation[EffectRecovery]
+    def recovery_candidates: () -> ActiveRecord::Relation[EffectRecovery]
+
     # @rbs (instance: Instance, intent: Actor::EffectRecoveryIntent, recovery: EffectRecovery, effect: Effect?, owners: Hash[String, Process], now: Time) -> void
     def check_one: (instance: Instance, intent: Actor::EffectRecoveryIntent, recovery: EffectRecovery, effect: Effect?, owners: Hash[String, Process], now: Time) -> void
 

test/integration/effect_recovery_test.rb

@@ -286,6 +286,57 @@ def checked(effect_id:, outcome:, arguments: nil, result: nil)
     worker&.stop
   end
 
+  test "one remaining mailbox slot cannot partially commit retirement" do
+    worker, effect_executor, effect = processing_export("full-mailbox")
+    original_limit = SolidObjects.configuration.max_mailbox_length
+    SolidObjects.configuration.max_mailbox_length = 1
+    assert_raises(SolidObjects::MailboxFull) do
+      SolidObjects.database_adapter.transaction do
+        instance = SolidObjects::Instance.lock.find(effect.instance_id)
+        SolidObjects::EffectRecoveryCoordinator.new.check(instance:, intents: [ SolidObjects::Actor::EffectRecoveryIntent.new(effect_id: effect.effect_id, request_id: "full") ])
+      end
+    end
+    assert_equal "processing", effect.reload.status
+    assert_nil SolidObjects::EffectRecovery.find(effect.effect_id).retired_at
+    assert_empty SolidObjects::Message.where(operation: [ "retired", "checked" ])
+  ensure
+    SolidObjects.configuration.max_mailbox_length = original_limit if original_limit
+    effect_executor&.stop
+    worker&.stop
+  end
+
+  test "runtime floor changes and missing owners are rechecked for existing effects" do
+    worker, effect_executor, effect = processing_export("runtime-floor")
+    SolidObjects::EffectRecovery.find(effect.effect_id).update!(recovery_timeout: 1)
+    SolidObjects::Process.find(effect.claimed_by).update!(last_heartbeat_at: SolidObjects.database_adapter.database_clock_now - 70)
+    SolidObjects.configuration.process_alive_threshold = 120
+    SolidObjects::EffectRecoveryCoordinator.new.recover_available
+    assert_equal "processing", effect.reload.status
+    assert_empty SolidObjects::Message.where(operation: "retired")
+    effect.update!(claimed_by: nil)
+    SolidObjects::EffectRecoveryCoordinator.new.recover_available
+    assert_equal 1, SolidObjects::Message.where(operation: "retired").count
+  ensure
+    effect_executor&.stop
+    worker&.stop
+  end
+
+  test "database time lookup failure rolls back without an abandonment outcome" do
+    worker, effect_executor, effect = processing_export("lookup-error")
+    adapter = SolidObjects.database_adapter
+    adapter.define_singleton_method(:database_clock_now) do
+      SolidObjects::Record.connection.select_value("SELECT absent_recovery_column FROM #{SolidObjects.table_name(:processes)}")
+    end
+    assert_raises(ActiveRecord::StatementInvalid) { SolidObjects::EffectRecoveryCoordinator.new.recover_available }
+    assert_equal "processing", effect.reload.status
+    assert_nil SolidObjects::EffectRecovery.find(effect.effect_id).retired_at
+    assert_empty SolidObjects::Message.where(operation: [ "retired", "checked" ])
+  ensure
+    adapter&.singleton_class&.remove_method(:database_clock_now)
+    effect_executor&.stop
+    worker&.stop
+  end
+
   test "a changed claimant is rechecked after the effect lock becomes available" do
     skip "PostgreSQL lock observation" unless database_family == :postgresql
 
@@ -310,6 +361,176 @@ def checked(effect_id:, outcome:, arguments: nil, result: nil)
     worker&.stop
   end
 
+  test "pending work does not wait on a recovery candidate within its extended grace" do
+    skip "PostgreSQL independent claims" unless database_family == :postgresql
+
+    worker, effect_executor, effect = processing_export("fresh-candidate")
+    SolidObjects::EffectRecovery.find(effect.effect_id).update!(recovery_timeout: 120)
+    SolidObjects::Process.find(effect.claimed_by).update!(last_heartbeat_at: SolidObjects.database_adapter.database_clock_now - 75)
+    ExportActor.ref("other").async.start_export
+    worker.run_until_idle
+    claimant = SolidObjects::EffectExecutor.new
+    results = Queue.new
+    claim_thread = nil
+    SolidObjects.database_adapter.transaction do
+      SolidObjects::Instance.lock.find(effect.instance_id)
+      claim_thread = Thread.new do
+        SolidObjects::Record.connection_pool.with_connection do
+          results << claimant.send(:claim_next)
+        end
+      end
+      selected = Timeout.timeout(2) { results.pop }
+      assert_equal "other", selected.instance.actor_id
+    end
+  ensure
+    claim_thread&.join(5)
+    claimant&.stop
+    effect_executor&.stop
+    worker&.stop
+  end
+
+  test "recovery notification survives a process crash after retirement" do
+    worker, effect_executor, effect = processing_export("crash")
+    worker.stop
+    configuration = SolidObjects::Record.connection_db_config.configuration_hash
+    script = <<~RUBY
+      require "solid_objects"
+      ActiveRecord::Base.establish_connection(JSON.parse(ENV.fetch("RECOVERY_DATABASE_CONFIGURATION")))
+      %w[record process instance message ready_message claimed_message reminder effect effect_recovery broadcast dead_letter].each do |model|
+        require File.expand_path("app/models/solid_objects/\#{model}")
+      end
+      class RecoveryExport < SolidObjects::Actor
+        actor_type "effect-recovery-export"
+        def retired(effect_id:, arguments:, outcome:)
+        end
+      end
+      SolidObjects::EffectRecoveryCoordinator.new.recover_available
+      ::Process.kill("KILL", ::Process.pid)
+    RUBY
+    process_id = ::Process.spawn({ "RECOVERY_DATABASE_CONFIGURATION" => JSON.generate(configuration) }, Gem.ruby, "-Ilib", "-e", script)
+    _, status = ::Process.wait2(process_id)
+    assert status.signaled?
+    assert_equal Signal.list.fetch("KILL"), status.termsig
+    assert_equal 1, SolidObjects::Message.where(operation: "retired").count
+    assert_empty effect.instance.reload.state.fetch("notifications")
+    worker = SolidObjects::Worker.new
+    worker.run_until_idle
+    assert_equal 1, effect.instance.reload.state.fetch("notifications").length
+    assert_nil effect_executor.send(:claim_next)
+  ensure
+    effect_executor&.stop
+    worker&.stop
+  end
+
+  test "a pending claimant wins before the explicit check and is rechecked" do
+    skip "PostgreSQL independent claims" unless database_family == :postgresql
+
+    ExportActor.ref("claim-first").async.start_checked_export
+    worker = SolidObjects::Worker.new
+    worker.run_until_idle
+    effect = SolidObjects::Effect.find_by!(name: "build_report")
+    claimant = SolidObjects::EffectExecutor.new
+    adapter = SolidObjects.database_adapter
+    original_lock = adapter.method(:lock_candidates)
+    locked = Queue.new
+    release = Queue.new
+    results = Queue.new
+    origin_locked = Queue.new
+    adapter.define_singleton_method(:lock_candidates) do |scope|
+      relation = original_lock.call(scope).load
+      locked << true
+      release.pop
+      relation
+    end
+    claim_thread = Thread.new do
+      SolidObjects::Record.connection_pool.with_connection { results << claimant.send(:claim_next) }
+    end
+    Timeout.timeout(5) { locked.pop }
+    check_thread = Thread.new do
+      SolidObjects::Record.connection_pool.with_connection do
+        SolidObjects.database_adapter.transaction do
+          instance = SolidObjects::Instance.lock.find(effect.instance_id)
+          origin_locked << true
+          SolidObjects::EffectRecoveryCoordinator.new.check(instance:, intents: [ SolidObjects::Actor::EffectRecoveryIntent.new(effect_id: effect.effect_id, request_id: "claim-first") ])
+        end
+      end
+    end
+    Timeout.timeout(5) { origin_locked.pop }
+    release << true
+    assert_equal effect.id, Timeout.timeout(5) { results.pop }.id
+    check_thread.join(5)
+    assert_equal "deferred", SolidObjects::Message.find_by!(operation: "checked").arguments.fetch("outcome")
+    assert_empty SolidObjects::Message.where(operation: "retired")
+  ensure
+    release&.push(true)
+    claim_thread&.join(5)
+    check_thread&.join(5)
+    adapter&.singleton_class&.remove_method(:lock_candidates)
+    claimant&.stop
+    worker&.stop
+  end
+
+  test "an explicit check can win and leave pending work for the claimant" do
+    skip "PostgreSQL independent claims" unless database_family == :postgresql
+
+    ExportActor.ref("check-first").async.start_checked_export
+    worker = SolidObjects::Worker.new
+    worker.run_until_idle
+    effect = SolidObjects::Effect.find_by!(name: "build_report")
+    claimant = SolidObjects::EffectExecutor.new
+    results = Queue.new
+    claim_thread = nil
+    SolidObjects.database_adapter.transaction do
+      instance = SolidObjects::Instance.lock.find(effect.instance_id)
+      SolidObjects::EffectRecoveryCoordinator.new.check(instance:, intents: [ SolidObjects::Actor::EffectRecoveryIntent.new(effect_id: effect.effect_id, request_id: "check-first") ])
+      claim_thread = Thread.new do
+        SolidObjects::Record.connection_pool.with_connection { results << claimant.send(:claim_next) }
+      end
+      assert_nil Timeout.timeout(5) { results.pop }
+    end
+    assert_equal effect.id, claimant.send(:claim_next).id
+    assert_equal "pending", SolidObjects::Message.find_by!(operation: "checked").arguments.fetch("outcome")
+    assert_empty SolidObjects::Message.where(operation: "retired")
+  ensure
+    claim_thread&.join(5)
+    claimant&.stop
+    worker&.stop
+  end
+
+  test "concurrent explicit checks and a replay share one retirement" do
+    skip "PostgreSQL independent checks" unless database_family == :postgresql
+
+    worker, effect_executor, effect = processing_export("explicit-race")
+    process_ids = Queue.new
+    check = ->(request_id) do
+      SolidObjects.database_adapter.transaction do
+        instance = SolidObjects::Instance.lock.find(effect.instance_id)
+        SolidObjects::EffectRecoveryCoordinator.new.check(instance:, intents: [ SolidObjects::Actor::EffectRecoveryIntent.new(effect_id: effect.effect_id, request_id:) ])
+      end
+    end
+    threads = []
+    SolidObjects.database_adapter.transaction do
+      SolidObjects::Instance.lock.find(effect.instance_id)
+      %w[one two].each do |request_id|
+        threads << Thread.new do
+          SolidObjects::Record.connection_pool.with_connection do |connection|
+            process_ids << connection.select_value("SELECT pg_backend_pid()").to_i
+            check.call(request_id)
+          end
+        end
+      end
+      2.times { wait_for_blocked_process(Timeout.timeout(5) { process_ids.pop }) }
+    end
+    threads.each { |thread| thread.join(5) }
+    check.call("one")
+    assert_equal 1, SolidObjects::Message.where(operation: "retired").count
+    assert_equal %w[retired already_retired], SolidObjects::Message.where(operation: "checked").order(:sequence).map { |message| message.arguments.fetch("outcome") }
+  ensure
+    threads&.each { |thread| thread.join(5) }
+    effect_executor&.stop
+    worker&.stop
+  end
+
   private
 
   def processing_export(actor_id)