fix: Resume heartbeats after database errors

Author: cardmagic

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 - Maintain effect-owner heartbeats during long-running handlers, completion, and
   failure handling, so healthy external I/O cannot trigger abandoned recovery.
+  Report failed heartbeat updates and retry on the next configured interval
+  without consuming effect attempts.
 - 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

@@ -124,6 +124,11 @@ 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.
 
+Failed updates emit `solid_objects.process.heartbeat_failed` and retry at the
+configured heartbeat interval without consuming effect attempts. If an outage
+lasts beyond the freshness window, recovery can still be permitted; retries do
+not cancel external work or extend the configured window.
+
 ## Compatibility and installation
 
 Upgrade all effect workers and process cleanup roles before emitting effects

lib/solid_objects/process_heartbeat.rb

@@ -25,6 +25,8 @@ def start
           break if wait_for_interval
 
           Record.connection_pool.with_connection { process_registry.heartbeat }
+        rescue => error
+          report_failure(error)
         end
       end
     end
@@ -42,6 +44,15 @@ def stop
 
     attr_reader :process_registry, :mutex, :condition
 
+    # @rbs (Exception) -> void
+    def report_failure(error)
+      payload = { process_id: process_registry.process_record&.id, error_class: error.class.name }
+      SolidObjects.configuration.logger.warn({ event: "solid_objects.process.heartbeat_failed", **payload })
+      SolidObjects.instrument(:"process.heartbeat_failed", **payload)
+    rescue
+      nil
+    end
+
     # @rbs () -> bool
     def wait_for_interval
       mutex.synchronize do

sig/generated/lib/solid_objects/process_heartbeat.rbs

@@ -29,6 +29,9 @@ module SolidObjects
 
     attr_reader condition: untyped
 
+    # @rbs (Exception) -> void
+    def report_failure: (Exception) -> void
+
     # @rbs () -> bool
     def wait_for_interval: () -> bool
   end

test/integration/effect_recovery_test.rb

@@ -132,6 +132,36 @@ def checked(effect_id:, outcome:, arguments: nil, result: nil)
     worker&.stop
   end
 
+  test "heartbeat maintenance resumes after a database error" do
+    SolidObjects.configuration.process_heartbeat_interval = 0.01
+    registry = SolidObjects::ProcessRegistry.new
+    registry.register(kind: "effect")
+    resumed = Queue.new
+    attempts = 0
+    registry.define_singleton_method(:heartbeat) do
+      attempts += 1
+      if attempts == 1
+        SolidObjects::Record.connection.select_value("SELECT absent_heartbeat_column FROM #{SolidObjects.table_name(:processes)}")
+      end
+      updated = super()
+      resumed << true if updated
+      updated
+    end
+    events = []
+    subscriber = ActiveSupport::Notifications.subscribe("solid_objects.process.heartbeat_failed") { |event| events << event.payload }
+    heartbeat = SolidObjects::ProcessHeartbeat.new(process_registry: registry)
+    heartbeat.start
+    Timeout.timeout(2) { resumed.pop }
+    heartbeat.stop
+    assert_operator attempts, :>=, 2
+    assert_equal 1, events.length
+    assert_equal registry.process_record.id, events.first.fetch(:process_id)
+  ensure
+    heartbeat&.stop
+    ActiveSupport::Notifications.unsubscribe(subscriber) if subscriber
+    registry&.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