fix: Heartbeat while effect handlers run

Author: cardmagic

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## 0.15.0 - 2026-09-15
 
+- Maintain effect-owner heartbeats during long-running handlers, completion, and
+  failure handling, so healthy external I/O cannot trigger abandoned recovery.
 - Return a stable effect handle from every `emit`. Wrappers must return it;
   operations relying on an implicit `nil` result should return `nil` explicitly.
 - Add abandoned effect recovery with `on_recovery`, optional `on_status`, staged

docs/effect-recovery.md

@@ -118,6 +118,12 @@ floored at that runtime threshold; changing configuration changes the effective
 floor even for existing effects. Database lookup errors surface as errors,
 never as missing/stale observations.
 
+Effect workers maintain their process heartbeat while the handler waits on
+external I/O and while committing success or failure. A long-running healthy
+handler therefore remains protected beyond the recovery timeout. This requires
+an available database connection for the heartbeat, as well as runtime threads
+that can continue running.
+
 ## Compatibility and installation
 
 Upgrade all effect workers and process cleanup roles before emitting effects

docs/roadmap.md

@@ -16,7 +16,8 @@
   originally staged arguments for callback correlation
 - Stable `emit` handles and opted-in abandoned effect retirement coordinated
   with effect claims, with atomic recovery/status mailbox notifications and
-  per-effect extending heartbeat timeouts. See [effect recovery](effect-recovery.md)
+  per-effect extending heartbeat timeouts and heartbeats throughout long-running
+  effect handlers. See [effect recovery](effect-recovery.md)
   for the SQL lock protocol, retention boundary, and external idempotency limit.
 - Public RBS effect success/failure envelopes and error records, checked against
   the runtime constructors and a packaged consumer with strict Steep diagnostics

lib/solid_objects.rb

@@ -57,6 +57,7 @@
 require "solid_objects/effect_registry"
 require "solid_objects/effect_payload"
 require "solid_objects/effect_recovery_coordinator"
+require "solid_objects/process_heartbeat"
 require "solid_objects/commit_action_registry"
 require "solid_objects/lease"
 require "solid_objects/lease_renewer"

lib/solid_objects/effect_executor.rb

@@ -45,12 +45,16 @@ def run_once
       effect = claim_next
       return false unless effect
 
+      heartbeat = ProcessHeartbeat.new(process_registry:)
+      heartbeat.start
       result = deliver(effect)
       complete(effect, result)
       true
     rescue => error
       fail_effect(effect, error) if effect
       false
+    ensure
+      heartbeat&.stop
     end
 
     # @rbs () -> void

lib/solid_objects/process_heartbeat.rb

@@ -0,0 +1,53 @@
+# rbs_inline: enabled
+
+module SolidObjects
+  class ProcessHeartbeat
+    # @rbs @process_registry: ProcessRegistry
+    # @rbs @mutex: Thread::Mutex
+    # @rbs @condition: Thread::ConditionVariable
+    # @rbs @stopped: bool
+    # @rbs @thread: Thread?
+
+    # @rbs (process_registry: ProcessRegistry) -> void
+    def initialize(process_registry:)
+      @process_registry = process_registry
+      @mutex = Thread::Mutex.new
+      @condition = Thread::ConditionVariable.new
+      @stopped = false
+      @thread = nil
+    end
+
+    # @rbs () -> void
+    def start
+      @thread = Thread.new do
+        Thread.current.report_on_exception = false
+        loop do
+          break if wait_for_interval
+
+          Record.connection_pool.with_connection { process_registry.heartbeat }
+        end
+      end
+    end
+
+    # @rbs () -> void
+    def stop
+      mutex.synchronize do
+        @stopped = true
+        condition.broadcast
+      end
+      @thread&.join
+    end
+
+    private
+
+    attr_reader :process_registry, :mutex, :condition
+
+    # @rbs () -> bool
+    def wait_for_interval
+      mutex.synchronize do
+        condition.wait(mutex, SolidObjects.configuration.process_heartbeat_interval) unless @stopped
+        @stopped
+      end
+    end
+  end
+end

sig/generated/lib/solid_objects/process_heartbeat.rbs

@@ -0,0 +1,35 @@
+# Generated from lib/solid_objects/process_heartbeat.rb with RBS::Inline
+
+module SolidObjects
+  class ProcessHeartbeat
+    @process_registry: ProcessRegistry
+
+    @mutex: Thread::Mutex
+
+    @condition: Thread::ConditionVariable
+
+    @stopped: bool
+
+    @thread: Thread?
+
+    # @rbs (process_registry: ProcessRegistry) -> void
+    def initialize: (process_registry: ProcessRegistry) -> void
+
+    # @rbs () -> void
+    def start: () -> void
+
+    # @rbs () -> void
+    def stop: () -> void
+
+    private
+
+    attr_reader process_registry: untyped
+
+    attr_reader mutex: untyped
+
+    attr_reader condition: untyped
+
+    # @rbs () -> bool
+    def wait_for_interval: () -> bool
+  end
+end

test/integration/effect_recovery_test.rb

@@ -60,6 +60,78 @@ def checked(effect_id:, outcome:, arguments: nil, result: nil)
     assert_empty actor.send(:drain_effect_intents)
   end
 
+  test "a running effect keeps its process heartbeat fresh" do
+    SolidObjects.configuration.process_heartbeat_interval = 0.02
+    SolidObjects.configuration.process_alive_threshold = 0.1
+    ExportActor.ref("long-handler").async.start_recoverable_export
+    worker = SolidObjects::Worker.new
+    worker.run_until_idle
+    entered = Queue.new
+    release = Queue.new
+    heartbeats = Queue.new
+    executing = nil
+    registry = SolidObjects::ProcessRegistry.new
+    registry.define_singleton_method(:heartbeat) do
+      updated = super()
+      heartbeats << Thread.current if updated && executing
+      updated
+    end
+    SolidObjects.register_effect(:build_report) do
+      executing = Thread.current
+      entered << true
+      release.pop
+      { "artifact_key" => "report.pdf" }
+    end
+    executor = SolidObjects::EffectExecutor.new(process_registry: registry)
+    effect_thread = Thread.new do
+      SolidObjects::Record.connection_pool.with_connection { executor.run_once }
+    end
+    Timeout.timeout(5) { entered.pop }
+    heartbeat_thread = Timeout.timeout(2) { heartbeats.pop }
+    Timeout.timeout(2) { 7.times { heartbeats.pop } }
+    SolidObjects::EffectRecoveryCoordinator.new.recover_available
+    assert_equal "processing", SolidObjects::Effect.find_by!(name: "build_report").status
+    assert_empty SolidObjects::Message.where(operation: "retired")
+    release << true
+    assert effect_thread.value
+    refute heartbeat_thread.alive?
+    assert_equal "completed", SolidObjects::Effect.find_by!(name: "build_report").status
+  ensure
+    release&.push(true)
+    effect_thread&.join(5)
+    executor&.stop
+    worker&.stop
+  end
+
+  test "a failed handler stops its heartbeat before scheduling a retry" do
+    SolidObjects.configuration.process_heartbeat_interval = 0.01
+    ExportActor.ref("failing-handler").async.start_recoverable_export
+    worker = SolidObjects::Worker.new
+    worker.run_until_idle
+    heartbeats = Queue.new
+    registry = SolidObjects::ProcessRegistry.new
+    registry.define_singleton_method(:heartbeat) do
+      updated = super()
+      heartbeats << Thread.current if updated
+      updated
+    end
+    heartbeat_thread = nil
+    SolidObjects.register_effect(:build_report) do
+      heartbeats.pop(true) until heartbeats.empty?
+      heartbeat_thread = Timeout.timeout(2) { heartbeats.pop }
+      raise "remote request failed"
+    end
+    executor = SolidObjects::EffectExecutor.new(process_registry: registry)
+    refute executor.run_once
+    refute_nil heartbeat_thread
+    refute heartbeat_thread.alive?
+    assert_equal "pending", SolidObjects::Effect.find_by!(name: "build_report").status
+    assert_empty SolidObjects::Message.where(operation: "retired")
+  ensure
+    executor&.stop
+    worker&.stop
+  end
+
   test "a longer recovery timeout survives ordinary process cleanup" do
     ExportActor.ref("long-grace").async.start_with_timeout(timeout: 120)
     worker = SolidObjects::Worker.new