From ebe3c11047d5f1ad6413b675a976057d921cf23f Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 18 Oct 2026 17:26:18 -0700 Subject: [PATCH] test: make the heartbeat recovery test independent of write speed The test set a 0.1 second alive threshold and ran the recovery pass right after the 8th heartbeat. One slow heartbeat commit on a busy CI runner made the stored heartbeat older than the threshold, so the pass retired the running effect as designed, and the test read "completed". It failed three times on MySQL in the last 30 failed CI runs. The test now waits until the stored heartbeat is more than one threshold past the claim, and runs the recovery pass with its clock set half a threshold after that heartbeat. A slow write can only make the test slower. A new case adds a 0.15 second pause after each heartbeat write: it failed with the CI message before this change and passes after it. Both cases still fail when the heartbeat thread is off. --- test/integration/effect_recovery_test.rb | 100 ++++++++++++++--------- 1 file changed, 60 insertions(+), 40 deletions(-) diff --git a/test/integration/effect_recovery_test.rb b/test/integration/effect_recovery_test.rb index 4ff4bfc..cd2889c 100644 --- a/test/integration/effect_recovery_test.rb +++ b/test/integration/effect_recovery_test.rb @@ -61,46 +61,11 @@ def checked(effect_id:, outcome:, arguments: nil, result: nil) 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 + assert_running_effect_keeps_heartbeat_fresh(heartbeat_delay: 0) + end + + test "a running effect keeps its process heartbeat fresh when heartbeat writes are slow" do + assert_running_effect_keeps_heartbeat_fresh(heartbeat_delay: 0.15) end test "a failed handler stops its heartbeat before scheduling a retry" do @@ -635,6 +600,61 @@ def retired(effect_id:, arguments:, outcome:) private + def assert_running_effect_keeps_heartbeat_fresh(heartbeat_delay:) + 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() + sleep(heartbeat_delay) if updated && executing + 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 } + effect = SolidObjects::Effect.find_by!(name: "build_report") + owner = SolidObjects::Process.find(effect.claimed_by) + threshold = SolidObjects.configuration.process_alive_threshold + heartbeat_thread = Timeout.timeout(2) { heartbeats.pop } + Timeout.timeout(5) { heartbeats.pop until owner.reload.last_heartbeat_at - effect.claimed_at > threshold } + recover_at(owner.last_heartbeat_at + (threshold / 2)) + assert_equal "processing", effect.reload.status + assert_empty SolidObjects::Message.where(operation: "retired") + release << true + assert effect_thread.value + refute heartbeat_thread.alive? + assert_equal "completed", effect.reload.status + ensure + release&.push(true) + effect_thread&.join(5) + executor&.stop + worker&.stop + end + + def recover_at(database_now) + adapter = SolidObjects.database_adapter + adapter.define_singleton_method(:database_clock_now) { database_now } + SolidObjects::EffectRecoveryCoordinator.new.recover_available + ensure + adapter.singleton_class.send(:remove_method, :database_clock_now) + end + def processing_export(actor_id) ExportActor.ref(actor_id).async.start_checked_export worker = SolidObjects::Worker.new