From 6c943539c4282e74b465e606df69c82a2e4ee540 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 11:33:04 -0700 Subject: [PATCH 1/5] feat: Coordinate abandoned effect recovery Return stable emit handles and retire abandoned effects atomically with durable recovery callbacks. Preserve owner heartbeat grace across cleanup and fence late completion using instance-before-effect locks. --- .rubocop.yml | 3 + CHANGELOG.md | 11 + README.md | 1 + app/models/solid_objects/effect_recovery.rb | 10 + ...000_add_solid_objects_effect_recoveries.rb | 17 + docs/effect-recovery.md | 185 ++++++++++ docs/roadmap.md | 4 + lib/solid_objects.rb | 1 + lib/solid_objects/actor.rb | 42 ++- lib/solid_objects/actor_signatures.rb | 2 +- lib/solid_objects/database_adapter.rb | 13 + lib/solid_objects/doctor.rb | 1 + lib/solid_objects/effect_executor.rb | 3 + lib/solid_objects/effect_payload.rb | 15 + .../effect_recovery_coordinator.rb | 125 +++++++ lib/solid_objects/executor.rb | 16 +- lib/solid_objects/process_pruner.rb | 2 +- lib/solid_objects/process_registry.rb | 15 +- lib/solid_objects/test_helper.rb | 1 + sig/generated/lib/solid_objects/actor.rbs | 43 ++- .../lib/solid_objects/database_adapter.rbs | 3 + .../lib/solid_objects/effect_payload.rbs | 9 + .../effect_recovery_coordinator.rbs | 38 ++ .../models/solid_objects/effect_recovery.rbs | 6 + sig/public/effect_payload.rbs | 14 + test/database_test_helper.rb | 3 + test/dummy/prepare_cli_reminder.rb | 2 + test/dummy/prepare_cli_worker.rb | 2 + test/dummy/web_mount_check.rb | 2 + test/integration/actor_signatures_test.rb | 2 + test/integration/effect_payload_types_test.rb | 16 +- test/integration/effect_recovery_test.rb | 346 ++++++++++++++++++ test/integration/separate_database_test.rb | 3 +- test/types/effect_payloads.rb | 22 ++ test/types/effect_payloads.rbs | 6 + 35 files changed, 961 insertions(+), 23 deletions(-) create mode 100644 app/models/solid_objects/effect_recovery.rb create mode 100644 db/migrate/20260915000000_add_solid_objects_effect_recoveries.rb create mode 100644 docs/effect-recovery.md create mode 100644 lib/solid_objects/effect_recovery_coordinator.rb create mode 100644 sig/generated/lib/solid_objects/effect_recovery_coordinator.rbs create mode 100644 sig/generated/models/solid_objects/effect_recovery.rbs create mode 100644 test/integration/effect_recovery_test.rb diff --git a/.rubocop.yml b/.rubocop.yml index 7299253..b44eb63 100644 --- a/.rubocop.yml +++ b/.rubocop.yml @@ -3,6 +3,9 @@ inherit_gem: { rubocop-rails-omakase: rubocop.yml } +Layout/LeadingCommentSpace: + AllowRBSInlineAnnotation: true + AllCops: TargetRubyVersion: 3.3 Exclude: diff --git a/CHANGELOG.md b/CHANGELOG.md index a7d6ae0..95deb9d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## Unreleased + +- 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 + `request_effect_recovery`, and an extending `recovery_timeout` in seconds. + Retirement and durable callbacks share the claim-locking transaction. Add + frozen outcome constants and public RBS envelopes. Install the new recovery + binding migration before upgrading runtime processes. External actions still + require idempotency; retirement does not cancel an old handler or remote call. + ## 0.14.7 - 2026-09-14 - Publish RBS contracts for effect callback envelopes and Ruby error summaries. diff --git a/README.md b/README.md index 2657fc5..7322961 100644 --- a/README.md +++ b/README.md @@ -149,6 +149,7 @@ SQL and should be allowed to enjoy that. - One successful turn commits actor state and staged reminders, messages, effects, commit actions, and broadcasts together. - Fencing prevents stale Ruby code from committing, but it cannot stop that code from continuing to run. - External effects can repeat and must deduplicate with the stable effect ID or another durable idempotency key. +- [Effect recovery](docs/effect-recovery.md) uses stable `emit` handles and `on_recovery` to retire abandoned work atomically with a durable actor callback. - Actor handlers may read application records but cannot write them directly. Use `commit_action` for bounded same-database writes and `emit` for external I/O. - `async`, reminders, effects, and broadcasts need `bundle exec solid_objects start`. Pending work remains in SQL while it is down. - One hot identity is intentionally sequential. There are no transactions across actor identities. diff --git a/app/models/solid_objects/effect_recovery.rb b/app/models/solid_objects/effect_recovery.rb new file mode 100644 index 0000000..d5fb2dd --- /dev/null +++ b/app/models/solid_objects/effect_recovery.rb @@ -0,0 +1,10 @@ +# rbs_inline: enabled + +module SolidObjects + class EffectRecovery < Record + self.table_name = SolidObjects.table_name(:effect_recoveries) + self.primary_key = "effect_id" + + belongs_to :instance, class_name: "SolidObjects::Instance" + end +end diff --git a/db/migrate/20260915000000_add_solid_objects_effect_recoveries.rb b/db/migrate/20260915000000_add_solid_objects_effect_recoveries.rb new file mode 100644 index 0000000..bb2cd5e --- /dev/null +++ b/db/migrate/20260915000000_add_solid_objects_effect_recoveries.rb @@ -0,0 +1,17 @@ +# rbs_inline: enabled + +class AddSolidObjectsEffectRecoveries < ActiveRecord::Migration[7.1] + # @rbs () -> void + def change + create_table SolidObjects.table_name(:effect_recoveries), id: :string, limit: 36, primary_key: :effect_id do |definition| + definition.references :instance, null: false, + foreign_key: { to_table: SolidObjects.table_name(:instances), on_delete: :cascade, name: "fk_so_effect_recoveries_instance" } + definition.string :recovery_operation, limit: 191 + definition.string :status_operation, limit: 191 + definition.float :recovery_timeout + definition.datetime :retired_at, precision: 6 + definition.timestamps precision: 6, null: false + definition.check_constraint "recovery_timeout IS NULL OR recovery_timeout > 0", name: "chk_so_effect_recoveries_timeout" + end + end +end diff --git a/docs/effect-recovery.md b/docs/effect-recovery.md new file mode 100644 index 0000000..f388780 --- /dev/null +++ b/docs/effect-recovery.md @@ -0,0 +1,185 @@ +# Effect recovery coordination + +`emit` returns a JSON-serializable handle containing the public `effect_id`. +Registering `on_recovery` opts the effect into retirement when its owner has +stopped heartbeating. `on_status` is optional and receives responses only to +explicit `request_effect_recovery(handle)` intents. Normal success and failure +retain their existing callbacks. + +## Watchdog using supported APIs + +```ruby +class ReportExport < SolidObjects::Actor + attribute :revision, default: 0 + attribute :export_effect, default: nil + attribute :artifact_key, default: "" + attribute :applied_effect_id, default: nil + + def start + self.revision += 1 + self.export_effect = emit(:build_report, + revision: revision, + on_success: :export_finished, + on_failure: :export_failed, + on_recovery: :recover_export, + on_status: :inspect_export, + recovery_timeout: 120) + schedule(at: Time.now + 30, key: "export-watchdog").watchdog + nil + end + + def watchdog + request_effect_recovery(export_effect) if export_effect + end + + def recover_export(effect_id:, arguments:, outcome:) + return unless effect_id == export_effect&.fetch("effect_id") + return unless arguments.fetch("revision") == revision + + start + end + + def export_finished(effect_id:, arguments:, result:) + apply_export_result(effect_id:, arguments:, result:) + end + + def export_failed(effect_id:, arguments:, error:) + end + + def inspect_export(effect_id:, outcome:, arguments: nil, result: nil) + return unless effect_id == export_effect&.fetch("effect_id") + + case outcome + when SolidObjects::EffectRecoveryOutcome::COMPLETED + apply_export_result(effect_id:, arguments:, result:) + when SolidObjects::EffectRecoveryOutcome::DEFERRED, SolidObjects::EffectRecoveryOutcome::PENDING + schedule(at: Time.now + 30, key: "export-watchdog").watchdog + end + end + + private + + def apply_export_result(effect_id:, arguments:, result:) + return unless effect_id == export_effect&.fetch("effect_id") + return unless arguments.fetch("revision") == revision + return if applied_effect_id == effect_id + + self.artifact_key = result.fetch("artifact_key") + self.applied_effect_id = effect_id + end +end +``` + +Register `build_report` through the ordinary effect registry. Its successful +result in this example is `{ "artifact_key" => "reports/example.pdf" }`. +The library builds the retirement payload, including `"outcome" => "retired"`; +the effect handler does not return that outcome itself. Ruby actor operations +receive keywords. `recover_export` needs handle/revision guards but no outcome +guard because only a new retirement invokes it. The same guarded result helper +handles success and completed-status repair, preventing duplicate application. +Only recovery emits a replacement; status observations never do. + +The watchdog is optional: `on_recovery` alone enables automatic retirement. +`on_status` alone does not enable retirement, polling, or subscriptions. Explicit +checks require both bindings persisted by `emit` and cannot replace either. + +## Public envelopes and timeout + +`SolidObjects::effect_handle` describes `{ "effect_id" => String }`. +`SolidObjects::effect_retired_payload[Arguments]` requires the effect ID, original +arguments, and `"outcome" => "retired"`. Status uses +`SolidObjects::effect_recovery_payload[Arguments, Result]`, a record union. +Every variant has an effect ID; retired and completed require original arguments; +only completed has a recorded result (including `nil`). Other Ruby observations +contain only the effect ID and outcome. + +Strict packaged-consumer tests verify the constant literals and individual +records. Steep 2.0 does not narrow this string-keyed record union after comparing +`payload["outcome"]` with `COMPLETED`; accessing `result` through the union still +fails its return-type check. Use the concrete completed/retired record in typed +helpers after validating the discriminator, with an explicit type assertion if +needed. The library retains precise records rather than weakening them to an +untyped hash. Ordinary Ruby keyword dispatch needs no payload hydration. + +| Frozen `SolidObjects::EffectRecoveryOutcome` constant | Wire value | Meaning | +| --- | --- | --- | +| `RETIRED` | `"retired"` | This check retired the effect; separate recovery owns replacement. | +| `DEFERRED` | `"deferred"` | Fresh owner; preserve its claim and attempts. | +| `PENDING` | `"pending"` | Initial execution or retry remains with the scheduler. | +| `COMPLETED` | `"completed"` | Original arguments and recorded result are available. | +| `DEAD` | `"dead"` | Preserve terminal failure and its existing callback. | +| `ALREADY_RETIRED` | `"already_retired"` | Earlier retirement; no additional recovery notification. | +| `MISSING` | `"missing"` | Owned binding exists but effect data was pruned. | + +`recovery_timeout` must be positive finite seconds and requires `on_recovery`. +Fractional durations are allowed. Omission uses the current runtime +`process_alive_threshold`, normally 60 seconds. Smaller positive values are +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. + +## Compatibility and installation + +Upgrade all effect workers and process cleanup roles before emitting effects +with recovery enabled. Older runtimes do not honor the persisted bindings or +the new lock protocol. + +Run `solid_objects:install:migrations` and your application's normal migration +process before starting upgraded workers. The additive migration creates the +durable binding table; it does not change existing effect status constraints. + +`emit` now returns its handle, including without recovery options. Callers may +ignore it. Wrappers must return `super`; operations whose last expression used +to be `emit` may now return the handle to callers. End those operations with +`nil` if their previous result must remain unchanged. The return-value change is +intentional and is not strictly backward compatible. + +## Transaction and lock protocol + +Emission, actor state, the effect, and its recovery binding share the actor's +fenced commit. An explicit check executes on that same connection. Automatic +recovery performs one independent library transaction per candidate, with no +application transaction waiting on a second connection. + +The lock order is originating instance, effect rows ordered by public effect +ID, recovery binding rows in the same order, then owner processes ordered by ID. +Completion and failure must acquire the instance before the effect. Pending +claims lock only their effect and do not subsequently acquire an instance lock. +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. + +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 +the cutoff is fresh; equality is stale. A stopped or draining process with +fresh heartbeat evidence still protects an opted-in effect until that timeout. +Cleanup preserves opted-in claims, and process pruning excludes processes that +still own effects. Later effect polling and process cleanup revisit deferred +effects without resetting their last heartbeat. + +Retirement stores a durable `retired_at` in `effect_recoveries` and moves the +effect into the existing terminal `completed` storage state, clearing its +claim. The recovery record distinguishes retirement from successful completion; +no success callback is generated. All recovery observations consult that record +before interpreting the effect row. A late completion or failure is rejected by +the existing processing/claim fence. This representation avoids rewriting the +existing effect-status constraint across adapters. + +The retirement record and the recovery mailbox message commit atomically. A +winning explicit check additionally enqueues its status response after the +retirement notification. Failure to insert either message rolls back the whole +decision. Retirement is deduplicated per effect; check responses use separate +per-request idempotency keys. Wake-up signals are delivery hints after commit. + +Recovery bindings survive effect/message pruning and remain until the originating +instance is destroyed or pruned. They do not prevent normal message or instance +retention. Within that lifetime, a removed non-retired effect reports `missing` +and a retirement record reports `already_retired`. A handle without an owned +binding raises an error, without disclosing another actor's state or recreating +a destroyed actor. Status-response message idempotency follows normal mailbox +retention; callers cannot supply or reuse internal check request IDs. + +An owner heartbeat measures process liveness, not effect progress. Retirement +does not cancel the old handler or prove a remote request stopped. External +actions still require idempotency across retries and replacement generations. diff --git a/docs/roadmap.md b/docs/roadmap.md index 839d9ce..efa8e98 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -14,6 +14,10 @@ dead letters, and tail retry - Transactional effects with success/failure actor messages carrying the 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) + 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 - Actor-to-actor asynchronous outbox delivery. Effects and broadcasts use diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb index dcb48a7..d27f6b5 100644 --- a/lib/solid_objects.rb +++ b/lib/solid_objects.rb @@ -56,6 +56,7 @@ require "solid_objects/polling_backoff" require "solid_objects/effect_registry" require "solid_objects/effect_payload" +require "solid_objects/effect_recovery_coordinator" require "solid_objects/commit_action_registry" require "solid_objects/lease" require "solid_objects/lease_renewer" diff --git a/lib/solid_objects/actor.rb b/lib/solid_objects/actor.rb index 63f9e15..2c40a15 100644 --- a/lib/solid_objects/actor.rb +++ b/lib/solid_objects/actor.rb @@ -2,7 +2,9 @@ module SolidObjects class Actor - EffectIntent = Data.define(:name, :arguments, :success_operation, :failure_operation) + EffectIntent = Data.define(:effect_id, :name, :arguments, :success_operation, :failure_operation, + :recovery_operation, :status_operation, :recovery_timeout) + EffectRecoveryIntent = Data.define(:effect_id, :request_id) CommitActionIntent = Data.define(:name, :arguments) # The reminders table holds a name in 191 characters. REMINDER_NAME_LIMIT = 191 @@ -143,6 +145,7 @@ def default_actor_type # @rbs @actor_id: String # @rbs @state: State # @rbs @effect_intents: Array[EffectIntent] + # @rbs @effect_recovery_intents: Array[EffectRecoveryIntent] # @rbs @commit_action_intents: Array[CommitActionIntent] # @rbs @reminder_intents: Array[ReminderIntent] # @rbs @outbound_message_intents: Array[OutboundMessageIntent] @@ -154,6 +157,7 @@ def initialize(actor_id:, state:) @actor_id = actor_id @state = state @effect_intents = [] + @effect_recovery_intents = [] @commit_action_intents = [] @reminder_intents = [] @outbound_message_intents = [] @@ -175,18 +179,41 @@ def reject(code, message, details: {}) raise Rejected.new(code: rejection_code, message:, details:) end - # @rbs (Symbol | String, ?on_success: Symbol | String?, ?on_failure: Symbol | String?, **untyped) -> nil - def emit(name, on_success: nil, on_failure: nil, **arguments) + # @rbs (Symbol | String, ?on_success: Symbol | String?, ?on_failure: Symbol | String?, ?on_recovery: Symbol | String?, ?on_status: Symbol | String?, ?recovery_timeout: Numeric?, **untyped) -> effect_handle + def emit(name, on_success: nil, on_failure: nil, on_recovery: nil, on_status: nil, recovery_timeout: nil, **arguments) validate_effect_callback!(on_success) 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 + effect_id = SecureRandom.uuid EffectIntent.new( + effect_id:, name: name.to_s, arguments: Serialization.dump(arguments), success_operation: on_success&.to_s, - failure_operation: on_failure&.to_s + failure_operation: on_failure&.to_s, + recovery_operation: on_recovery&.to_s, + status_operation: on_status&.to_s, + recovery_timeout: recovery_timeout&.to_f ).tap do |intent| effect_intents << intent end + { "effect_id" => effect_id } + end + + # @rbs (effect_handle) -> nil + def request_effect_recovery(handle) + unless handle.is_a?(Hash) && handle["effect_id"].is_a?(String) && !handle.fetch("effect_id").empty? + raise InvalidPayload, "expected an effect handle returned by emit" + end + + effect_recovery_intents << EffectRecoveryIntent.new(effect_id: handle.fetch("effect_id"), request_id: SecureRandom.uuid) nil end @@ -360,6 +387,11 @@ def drain_effect_intents effect_intents.shift(effect_intents.length) end + # @rbs () -> Array[EffectRecoveryIntent] + def drain_effect_recovery_intents + effect_recovery_intents.shift(effect_recovery_intents.length) + end + # @rbs () -> Array[CommitActionIntent] def drain_commit_action_intents commit_action_intents.shift(commit_action_intents.length) @@ -378,6 +410,7 @@ def drain_outbound_message_intents # @rbs () -> void def discard_intents effect_intents.clear + effect_recovery_intents.clear commit_action_intents.clear reminder_intents.clear outbound_message_intents.clear @@ -386,6 +419,7 @@ def discard_intents private attr_reader :effect_intents, + :effect_recovery_intents, :commit_action_intents, :reminder_intents, :outbound_message_intents diff --git a/lib/solid_objects/actor_signatures.rb b/lib/solid_objects/actor_signatures.rb index dde82ae..1aaf279 100644 --- a/lib/solid_objects/actor_signatures.rb +++ b/lib/solid_objects/actor_signatures.rb @@ -60,7 +60,7 @@ def public_send: (Symbol | String, **untyped) -> nil def schedule: (at: Time, ?every: Numeric?, ?missed: Symbol | String, ?key: (String | Symbol | Integer)?) -> #{name}::_SolidObjectsOperations def transmit: () -> #{name}::_SolidObjectsOperations - def emit: (Symbol | String, ?on_success: (#{callbacks}), ?on_failure: (#{callbacks}), **untyped) -> nil + def emit: (Symbol | String, ?on_success: (#{callbacks}), ?on_failure: (#{callbacks}), ?on_recovery: (#{callbacks}), ?on_status: (#{callbacks}), ?recovery_timeout: Numeric?, **untyped) -> SolidObjects::effect_handle end RBS end diff --git a/lib/solid_objects/database_adapter.rb b/lib/solid_objects/database_adapter.rb index 26cfa92..6ea5a3a 100644 --- a/lib/solid_objects/database_adapter.rb +++ b/lib/solid_objects/database_adapter.rb @@ -111,6 +111,19 @@ def database_now ActiveSupport::IsolatedExecutionState[TRANSACTION_CLOCK] ||= read_database_now end + # @rbs () -> Time + def database_clock_now + value = with_connection do |connection| + expression = case self.class.family(connection) + when :postgresql then "clock_timestamp()" + when :mysql then "CURRENT_TIMESTAMP(6)" + else "STRFTIME('%Y-%m-%d %H:%M:%f', 'now')" + end + connection.select_value("SELECT #{expression}") + end + value.is_a?(Time) ? value.utc : Time.parse("#{value} UTC").utc + end + # @rbs () { () -> untyped } -> untyped def with_lock_retry yield diff --git a/lib/solid_objects/doctor.rb b/lib/solid_objects/doctor.rb index 54f490f..2b54f1e 100644 --- a/lib/solid_objects/doctor.rb +++ b/lib/solid_objects/doctor.rb @@ -78,6 +78,7 @@ def to_s ], reminders: %w[id instance_id operation next_run_at status], effects: %w[id message_id instance_id effect_id status available_at], + effect_recoveries: %w[effect_id instance_id recovery_operation status_operation recovery_timeout retired_at], broadcasts: %w[id message_id instance_id broadcast_id status available_at], dead_letters: %w[id message_id instance_id actor_type actor_id attempts] }.freeze diff --git a/lib/solid_objects/effect_executor.rb b/lib/solid_objects/effect_executor.rb index 30c1273..592c6ac 100644 --- a/lib/solid_objects/effect_executor.rb +++ b/lib/solid_objects/effect_executor.rb @@ -113,6 +113,7 @@ def current_polling_interval # @rbs () -> Effect? def claim_next + EffectRecoveryCoordinator.new.recover_available database_adapter.transaction do now = database_adapter.database_now effect = database_adapter.lock_candidates( @@ -178,6 +179,7 @@ def complete(effect, result) ) result_message = nil database_adapter.transaction do + Instance.lock.find(effect.instance_id) locked_effect = Effect.lock.find(effect.id) verify_claim!(locked_effect) result_message = enqueue_result_message( @@ -213,6 +215,7 @@ def complete(effect, result) def fail_effect(effect, error) result_message = nil database_adapter.transaction do + Instance.lock.find(effect.instance_id) locked_effect = Effect.lock.find(effect.id) verify_claim!(locked_effect) dead = locked_effect.attempt_count >= locked_effect.max_attempts diff --git a/lib/solid_objects/effect_payload.rb b/lib/solid_objects/effect_payload.rb index ef7986c..249dfb2 100644 --- a/lib/solid_objects/effect_payload.rb +++ b/lib/solid_objects/effect_payload.rb @@ -3,6 +3,21 @@ module SolidObjects module EffectPayload class << self + # @rbs [Arguments] (effect_id: String, arguments: Arguments) -> effect_retired_payload[Arguments] + def retired(effect_id:, arguments:) + { "effect_id" => effect_id, "arguments" => arguments, "outcome" => EffectRecoveryOutcome::RETIRED } + end + + # @rbs [Arguments, Result] (effect_id: String, arguments: Arguments, result: Result) -> effect_completed_recovery_payload[Arguments, Result] + def recovery_completed(effect_id:, arguments:, result:) + { "effect_id" => effect_id, "arguments" => arguments, "outcome" => EffectRecoveryOutcome::COMPLETED, "result" => result } + end + + # @rbs (effect_id: String, outcome: effect_observation_outcome) -> effect_observation_payload + def recovery_observation(effect_id:, outcome:) + { "effect_id" => effect_id, "outcome" => outcome } + end + # @rbs [Arguments, Result] (effect_id: String, arguments: Arguments, result: Result) -> effect_success_payload[Arguments, Result] def success(effect_id:, arguments:, result:) { "effect_id" => effect_id, "arguments" => arguments, "result" => result } diff --git a/lib/solid_objects/effect_recovery_coordinator.rb b/lib/solid_objects/effect_recovery_coordinator.rb new file mode 100644 index 0000000..dea8600 --- /dev/null +++ b/lib/solid_objects/effect_recovery_coordinator.rb @@ -0,0 +1,125 @@ +# rbs_inline: enabled + +module SolidObjects + module EffectRecoveryOutcome + RETIRED = "retired".freeze #: "retired" + DEFERRED = "deferred".freeze #: "deferred" + PENDING = "pending".freeze #: "pending" + COMPLETED = "completed".freeze #: "completed" + DEAD = "dead".freeze #: "dead" + ALREADY_RETIRED = "already_retired".freeze #: "already_retired" + MISSING = "missing".freeze #: "missing" + end + + class EffectRecoveryCoordinator + # @rbs (instance: Instance, intents: Array[Actor::EffectRecoveryIntent]) -> void + def check(instance:, intents:) + return if intents.empty? + + effect_ids = intents.map(&:effect_id).uniq.sort + effects = Effect.where(instance_id: instance.id, effect_id: effect_ids).order(:effect_id).lock.to_a.index_by(&:effect_id) + recoveries = EffectRecovery.where(instance_id: instance.id, effect_id: effect_ids).order(:effect_id).lock.to_a.index_by(&:effect_id) + effect_ids.each do |effect_id| + recovery = recoveries[effect_id] + unless recovery&.recovery_operation && recovery.status_operation + raise InvalidPayload, "effect recovery requires an owned handle with on_recovery and on_status" + end + end + owner_ids = effects.values.filter_map(&:claimed_by).uniq.sort + owners = Process.where(id: owner_ids).order(:id).lock.to_a.index_by(&:id) + now = SolidObjects.database_adapter.database_clock_now + intents.each do |intent| + check_one(instance:, intent:, recovery: recoveries.fetch(intent.effect_id), effect: effects[intent.effect_id], owners:, now:) + end + end + + # @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| + notification = SolidObjects.database_adapter.transaction do + instance = Instance.lock.find_by(id: candidate.instance_id) + next unless instance + + effect = Effect.lock.find_by(effect_id: candidate.effect_id, instance_id: instance.id) + recovery = EffectRecovery.lock.find_by(effect_id: candidate.effect_id, instance_id: instance.id) + next unless effect && recovery + next if recovery.retired_at || effect.status != "processing" + + owner = Process.lock.find_by(id: effect.claimed_by) if effect.claimed_by + now = SolidObjects.database_adapter.database_clock_now + timeout = [ SolidObjects.configuration.process_alive_threshold, recovery.recovery_timeout || 0 ].max + next if owner && owner.last_heartbeat_at > now - timeout + + retire(instance:, effect:, recovery:, now:) + end + Mailbox.new.announce(notification) if notification + end + end + + private + + # @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}" + return if Message.where(instance_id: instance.id, idempotency_key: key).exists? + + outcome = observe(effect:, recovery:, owners:, now:) + retire(instance:, effect:, recovery:, now:) if outcome == EffectRecoveryOutcome::RETIRED + operation = recovery.status_operation + unless operation && SolidObjects.registry.fetch(instance.actor_type).definition.messages.key?(operation.to_sym) + raise UnknownMessage, "unknown effect status operation #{operation.inspect}" + end + arguments = case outcome + when EffectRecoveryOutcome::RETIRED + EffectPayload.retired(effect_id: intent.effect_id, arguments: effect.arguments) + when EffectRecoveryOutcome::COMPLETED + EffectPayload.recovery_completed(effect_id: intent.effect_id, arguments: effect.arguments, result: effect.result) + else + EffectPayload.recovery_observation(effect_id: intent.effect_id, outcome:) + end + Mailbox.new.enqueue_in_transaction( + reference: Reference.new(actor_type: instance.actor_type, actor_id: instance.actor_id), + operation:, + arguments:, + delivery_mode: "internal", + idempotency_key: key + ) + end + + # @rbs (effect: Effect?, recovery: EffectRecovery, owners: Hash[String, Process], now: Time) -> String + def observe(effect:, recovery:, owners:, now:) + return EffectRecoveryOutcome::ALREADY_RETIRED if recovery.retired_at + return EffectRecoveryOutcome::MISSING unless effect + return EffectRecoveryOutcome::PENDING if effect.status == "pending" + return EffectRecoveryOutcome::COMPLETED if effect.status == "completed" + return EffectRecoveryOutcome::DEAD if effect.status == "dead" + + owner = owners[effect.claimed_by] + timeout = [ SolidObjects.configuration.process_alive_threshold, recovery.recovery_timeout || 0 ].max + return EffectRecoveryOutcome::DEFERRED if owner && owner.last_heartbeat_at > now - timeout + + EffectRecoveryOutcome::RETIRED + end + + # @rbs (instance: Instance, effect: Effect, recovery: EffectRecovery, now: Time) -> Message + def retire(instance:, effect:, recovery:, now:) + actor_class = SolidObjects.registry.fetch(instance.actor_type) + operation = recovery.recovery_operation + unless operation && actor_class.definition.messages.key?(operation.to_sym) + raise UnknownMessage, "unknown effect recovery operation #{operation.inspect}" + end + + effect.update!(status: "completed", completed_at: now, claimed_by: nil, claimed_at: nil) + recovery.update!(retired_at: now) + Mailbox.new.enqueue_in_transaction( + reference: Reference.new(actor_type: instance.actor_type, actor_id: instance.actor_id), + operation:, + arguments: EffectPayload.retired(effect_id: effect.effect_id, arguments: effect.arguments), + delivery_mode: "internal", + idempotency_key: "effect:#{effect.effect_id}:recovery" + ) + end + end +end diff --git a/lib/solid_objects/executor.rb b/lib/solid_objects/executor.rb index 4c652f2..553008e 100644 --- a/lib/solid_objects/executor.rb +++ b/lib/solid_objects/executor.rb @@ -90,6 +90,7 @@ def complete(result, observable_changes, state_after:, state_changed:) max_bytes: SolidObjects.configuration.max_result_bytes ) effect_intents = actor.drain_effect_intents + recovery_intents = actor.drain_effect_recovery_intents commit_action_intents = actor.drain_commit_action_intents reminder_intents = actor.drain_reminder_intents outbound_message_intents = actor.drain_outbound_message_intents @@ -130,6 +131,7 @@ def complete(result, observable_changes, state_after:, state_changed:) observable_changes:, state_changed: ) + EffectRecoveryCoordinator.new.check(instance:, intents: recovery_intents) claimed_message.destroy! end @@ -241,10 +243,10 @@ def ensure_application_database_is_shared! # @rbs (message: Message, instance: Instance, intents: Array[Actor::EffectIntent]) -> Array[Effect] def enqueue_effects(message:, instance:, intents:) intents.map do |intent| - Effect.create!( + effect = Effect.create!( message:, instance:, - effect_id: SecureRandom.uuid, + effect_id: intent.effect_id, name: intent.name, arguments: intent.arguments, success_operation: intent.success_operation, @@ -253,6 +255,16 @@ def enqueue_effects(message:, instance:, intents:) max_attempts: SolidObjects.configuration.max_attempts, available_at: SolidObjects.database_adapter.database_now ) + if intent.recovery_operation || intent.status_operation + EffectRecovery.create!( + effect_id: intent.effect_id, + instance:, + recovery_operation: intent.recovery_operation, + status_operation: intent.status_operation, + recovery_timeout: intent.recovery_timeout + ) + end + effect end end diff --git a/lib/solid_objects/process_pruner.rb b/lib/solid_objects/process_pruner.rb index 574c874..53898ee 100644 --- a/lib/solid_objects/process_pruner.rb +++ b/lib/solid_objects/process_pruner.rb @@ -43,7 +43,7 @@ def prunable Process.where( shutdown_state: "stopped", stopped_at: ...(now - SolidObjects.configuration.process_retention) - ) + ).where.not(id: Effect.where.not(claimed_by: nil).select(:claimed_by)) end end end diff --git a/lib/solid_objects/process_registry.rb b/lib/solid_objects/process_registry.rb index 4a69f3d..6f76f26 100644 --- a/lib/solid_objects/process_registry.rb +++ b/lib/solid_objects/process_registry.rb @@ -10,6 +10,7 @@ class ProcessRegistry class << self # @rbs (?now: Time) -> Integer def cleanup_dead(now: SolidObjects.database_adapter.database_now) + EffectRecoveryCoordinator.new.recover_available stale_at = now - SolidObjects.configuration.process_alive_threshold dead_processes = Process .where.not(shutdown_state: "stopped") @@ -32,12 +33,14 @@ def deregister(process_record, now: SolidObjects.database_adapter.database_now) process_id: nil, activation_token: nil ) - Effect.where(claimed_by: process_record.id).update_all( - status: "pending", - claimed_by: nil, - claimed_at: nil, - available_at: now - ) + Effect.where(claimed_by: process_record.id) + .where.not(effect_id: EffectRecovery.where.not(recovery_operation: nil).select(:effect_id)) + .update_all( + status: "pending", + claimed_by: nil, + claimed_at: nil, + available_at: now + ) Reminder.where(claimed_by: process_record.id).update_all( claimed_by: nil, claimed_at: nil diff --git a/lib/solid_objects/test_helper.rb b/lib/solid_objects/test_helper.rb index 49676b3..804469c 100644 --- a/lib/solid_objects/test_helper.rb +++ b/lib/solid_objects/test_helper.rb @@ -35,6 +35,7 @@ def actor_owned_models ClaimedMessage, ReadyMessage, Broadcast, + EffectRecovery, Effect, Reminder, Message, diff --git a/sig/generated/lib/solid_objects/actor.rbs b/sig/generated/lib/solid_objects/actor.rbs index 3a7bfdf..d123d8f 100644 --- a/sig/generated/lib/solid_objects/actor.rbs +++ b/sig/generated/lib/solid_objects/actor.rbs @@ -3,6 +3,8 @@ module SolidObjects class Actor class EffectIntent < Data + attr_reader effect_id(): untyped + attr_reader name(): untyped attr_reader arguments(): untyped @@ -11,12 +13,31 @@ module SolidObjects attr_reader failure_operation(): untyped - def self.new: (untyped name, untyped arguments, untyped success_operation, untyped failure_operation) -> instance - | (name: untyped, arguments: untyped, success_operation: untyped, failure_operation: untyped) -> instance + attr_reader recovery_operation(): untyped + + attr_reader status_operation(): untyped + + attr_reader recovery_timeout(): untyped + + def self.new: (untyped effect_id, untyped name, untyped arguments, untyped success_operation, untyped failure_operation, untyped recovery_operation, untyped status_operation, untyped recovery_timeout) -> instance + | (effect_id: untyped, name: untyped, arguments: untyped, success_operation: untyped, failure_operation: untyped, recovery_operation: untyped, status_operation: untyped, recovery_timeout: untyped) -> instance + + def self.members: () -> [ :effect_id, :name, :arguments, :success_operation, :failure_operation, :recovery_operation, :status_operation, :recovery_timeout ] + + def members: () -> [ :effect_id, :name, :arguments, :success_operation, :failure_operation, :recovery_operation, :status_operation, :recovery_timeout ] + end + + class EffectRecoveryIntent < Data + attr_reader effect_id(): untyped + + attr_reader request_id(): untyped - def self.members: () -> [ :name, :arguments, :success_operation, :failure_operation ] + def self.new: (untyped effect_id, untyped request_id) -> instance + | (effect_id: untyped, request_id: untyped) -> instance - def members: () -> [ :name, :arguments, :success_operation, :failure_operation ] + def self.members: () -> [ :effect_id, :request_id ] + + def members: () -> [ :effect_id, :request_id ] end class CommitActionIntent < Data @@ -139,6 +160,8 @@ module SolidObjects @commit_action_intents: Array[CommitActionIntent] + @effect_recovery_intents: Array[EffectRecoveryIntent] + @effect_intents: Array[EffectIntent] @state: State @@ -158,8 +181,11 @@ module SolidObjects # @rbs (Symbol | String, String, ?details: Hash[String | Symbol, untyped]) -> bot def reject: (Symbol | String, String, ?details: Hash[String | Symbol, untyped]) -> bot - # @rbs (Symbol | String, ?on_success: Symbol | String?, ?on_failure: Symbol | String?, **untyped) -> nil - def emit: (Symbol | String, ?on_success: Symbol | String?, ?on_failure: Symbol | String?, **untyped) -> nil + # @rbs (Symbol | String, ?on_success: Symbol | String?, ?on_failure: Symbol | String?, ?on_recovery: Symbol | String?, ?on_status: Symbol | String?, ?recovery_timeout: Numeric?, **untyped) -> effect_handle + def emit: (Symbol | String, ?on_success: Symbol | String?, ?on_failure: Symbol | String?, ?on_recovery: Symbol | String?, ?on_status: Symbol | String?, ?recovery_timeout: Numeric?, **untyped) -> effect_handle + + # @rbs (effect_handle) -> nil + def request_effect_recovery: (effect_handle) -> nil # @rbs () -> OperationDispatcher def transmit: () -> OperationDispatcher @@ -216,6 +242,9 @@ module SolidObjects # @rbs () -> Array[EffectIntent] def drain_effect_intents: () -> Array[EffectIntent] + # @rbs () -> Array[EffectRecoveryIntent] + def drain_effect_recovery_intents: () -> Array[EffectRecoveryIntent] + # @rbs () -> Array[CommitActionIntent] def drain_commit_action_intents: () -> Array[CommitActionIntent] @@ -232,6 +261,8 @@ module SolidObjects attr_reader effect_intents: untyped + attr_reader effect_recovery_intents: untyped + attr_reader commit_action_intents: untyped attr_reader reminder_intents: untyped diff --git a/sig/generated/lib/solid_objects/database_adapter.rbs b/sig/generated/lib/solid_objects/database_adapter.rbs index 82bddcc..f429e0b 100644 --- a/sig/generated/lib/solid_objects/database_adapter.rbs +++ b/sig/generated/lib/solid_objects/database_adapter.rbs @@ -55,6 +55,9 @@ module SolidObjects # @rbs () -> Time def database_now: () -> Time + # @rbs () -> Time + def database_clock_now: () -> Time + # @rbs () { () -> untyped } -> untyped def with_lock_retry: () { () -> untyped } -> untyped diff --git a/sig/generated/lib/solid_objects/effect_payload.rbs b/sig/generated/lib/solid_objects/effect_payload.rbs index e1c13c5..74258c7 100644 --- a/sig/generated/lib/solid_objects/effect_payload.rbs +++ b/sig/generated/lib/solid_objects/effect_payload.rbs @@ -2,6 +2,15 @@ module SolidObjects module EffectPayload + # @rbs [Arguments] (effect_id: String, arguments: Arguments) -> effect_retired_payload[Arguments] + def self.retired: [Arguments] (effect_id: String, arguments: Arguments) -> effect_retired_payload[Arguments] + + # @rbs [Arguments, Result] (effect_id: String, arguments: Arguments, result: Result) -> effect_completed_recovery_payload[Arguments, Result] + def self.recovery_completed: [Arguments, Result] (effect_id: String, arguments: Arguments, result: Result) -> effect_completed_recovery_payload[Arguments, Result] + + # @rbs (effect_id: String, outcome: effect_observation_outcome) -> effect_observation_payload + def self.recovery_observation: (effect_id: String, outcome: effect_observation_outcome) -> effect_observation_payload + # @rbs [Arguments, Result] (effect_id: String, arguments: Arguments, result: Result) -> effect_success_payload[Arguments, Result] def self.success: [Arguments, Result] (effect_id: String, arguments: Arguments, result: Result) -> effect_success_payload[Arguments, Result] diff --git a/sig/generated/lib/solid_objects/effect_recovery_coordinator.rbs b/sig/generated/lib/solid_objects/effect_recovery_coordinator.rbs new file mode 100644 index 0000000..75d4188 --- /dev/null +++ b/sig/generated/lib/solid_objects/effect_recovery_coordinator.rbs @@ -0,0 +1,38 @@ +# Generated from lib/solid_objects/effect_recovery_coordinator.rb with RBS::Inline + +module SolidObjects + module EffectRecoveryOutcome + RETIRED: "retired" + + DEFERRED: "deferred" + + PENDING: "pending" + + COMPLETED: "completed" + + DEAD: "dead" + + ALREADY_RETIRED: "already_retired" + + MISSING: "missing" + end + + class EffectRecoveryCoordinator + # @rbs (instance: Instance, intents: Array[Actor::EffectRecoveryIntent]) -> void + def check: (instance: Instance, intents: Array[Actor::EffectRecoveryIntent]) -> void + + # @rbs () -> void + def recover_available: () -> void + + private + + # @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 + + # @rbs (effect: Effect?, recovery: EffectRecovery, owners: Hash[String, Process], now: Time) -> String + def observe: (effect: Effect?, recovery: EffectRecovery, owners: Hash[String, Process], now: Time) -> String + + # @rbs (instance: Instance, effect: Effect, recovery: EffectRecovery, now: Time) -> Message + def retire: (instance: Instance, effect: Effect, recovery: EffectRecovery, now: Time) -> Message + end +end diff --git a/sig/generated/models/solid_objects/effect_recovery.rbs b/sig/generated/models/solid_objects/effect_recovery.rbs new file mode 100644 index 0000000..116e442 --- /dev/null +++ b/sig/generated/models/solid_objects/effect_recovery.rbs @@ -0,0 +1,6 @@ +# Generated from app/models/solid_objects/effect_recovery.rb with RBS::Inline + +module SolidObjects + class EffectRecovery < Record + end +end diff --git a/sig/public/effect_payload.rbs b/sig/public/effect_payload.rbs index caa4958..d001e2d 100644 --- a/sig/public/effect_payload.rbs +++ b/sig/public/effect_payload.rbs @@ -1,4 +1,18 @@ module SolidObjects + type effect_handle = { "effect_id" => String } + + type effect_retired_payload[Arguments] = { + "effect_id" => String, "arguments" => Arguments, "outcome" => "retired" + } + + type effect_completed_recovery_payload[Arguments, Result] = { + "effect_id" => String, "arguments" => Arguments, "outcome" => "completed", "result" => Result + } + + type effect_observation_outcome = "deferred" | "pending" | "dead" | "already_retired" | "missing" + type effect_observation_payload = { "effect_id" => String, "outcome" => effect_observation_outcome } + type effect_recovery_payload[Arguments, Result] = effect_retired_payload[Arguments] | effect_completed_recovery_payload[Arguments, Result] | effect_observation_payload + type effect_error = { "class" => String?, "message" => String, diff --git a/test/database_test_helper.rb b/test/database_test_helper.rb index 611d748..56b42b5 100644 --- a/test/database_test_helper.rb +++ b/test/database_test_helper.rb @@ -24,10 +24,12 @@ 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) ActiveRecord::Base.connection.create_table(:solid_objects_test_domain_records) do |table| table.string :name, null: false @@ -42,6 +44,7 @@ require_relative "../app/models/solid_objects/claimed_message" require_relative "../app/models/solid_objects/reminder" require_relative "../app/models/solid_objects/effect" +require_relative "../app/models/solid_objects/effect_recovery" require_relative "../app/models/solid_objects/broadcast" require_relative "../app/models/solid_objects/dead_letter" diff --git a/test/dummy/prepare_cli_reminder.rb b/test/dummy/prepare_cli_reminder.rb index 9f472ff..23570dd 100644 --- a/test/dummy/prepare_cli_reminder.rb +++ b/test/dummy/prepare_cli_reminder.rb @@ -6,10 +6,12 @@ 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) # A reminder that is already due, so the scheduler claims and enqueues it on # its first pass rather than waiting. diff --git a/test/dummy/prepare_cli_worker.rb b/test/dummy/prepare_cli_worker.rb index ec7f0f1..0893b40 100644 --- a/test/dummy/prepare_cli_worker.rb +++ b/test/dummy/prepare_cli_worker.rb @@ -6,10 +6,12 @@ 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) now = Time.current instance = SolidObjects::Instance.create!( diff --git a/test/dummy/web_mount_check.rb b/test/dummy/web_mount_check.rb index 1aea6f3..9e5f258 100644 --- a/test/dummy/web_mount_check.rb +++ b/test/dummy/web_mount_check.rb @@ -14,11 +14,13 @@ 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" ActiveRecord::Migration.verbose = false CreateSolidObjectsTables.new.migrate(:up) AddStateRevisionToSolidObjectsInstances.new.migrate(:up) RenameMessageDispatchColumns.new.migrate(:up) +AddSolidObjectsEffectRecoveries.new.migrate(:up) instance = SolidObjects::Instance.create!( actor_type: "MountCheckActor", diff --git a/test/integration/actor_signatures_test.rb b/test/integration/actor_signatures_test.rb index 1841ef9..0843586 100644 --- a/test/integration/actor_signatures_test.rb +++ b/test/integration/actor_signatures_test.rb @@ -62,6 +62,8 @@ class ActorSignaturesTest < ActiveSupport::TestCase emit :run_model, on_failure: :helper emit :run_model, on_success: :status emit :run_model, on_failure: :schedule + emit :build_report, on_recovery: :recvoer + emit :build_report, on_status: :inspec RUBY File.write(consumer_path, consumer.sub(" commit_action :global_action, generation: 1", invalid_calls)) output, status = typecheck(directory) diff --git a/test/integration/effect_payload_types_test.rb b/test/integration/effect_payload_types_test.rb index fef86da..c15fd2b 100644 --- a/test/integration/effect_payload_types_test.rb +++ b/test/integration/effect_payload_types_test.rb @@ -36,10 +36,20 @@ class EffectPayloadTypesTest < ActiveSupport::TestCase assert status.success?, output File.write(configuration_path, configuration) + signature_path = File.join(project, "sig/consumer.rbs") + signatures = File.read(signature_path) + File.write(signature_path, signatures.sub("effect_completed_recovery_payload[report_arguments, String]", "effect_recovery_payload[report_arguments, String]")) + output, status = typecheck(project) + refute status.success?, "Steep narrowing changed; update the documented record-union limitation" + assert_includes output, "Ruby::ReturnTypeMismatch" + File.write(signature_path, signatures) + [ [ '"effect_id" => "effect-1"', '"effect_identifier" => "effect-1"' ], [ '"message" => "failed"', '"message" => 42' ], - [ 'arguments["generation"]', 'arguments["generation"].to_s' ] + [ 'arguments["generation"]', 'arguments["generation"].to_s' ], + [ "EffectRecoveryOutcome::RETIRED", "EffectRecoveryOutcome::PENDING" ], + [ 'payload["arguments"]["revision"]', 'payload["arguments"]["revision"].to_s' ] ].each do |original, invalid| File.write(consumer_path, consumer.sub(original, invalid)) output, status = typecheck(project) @@ -53,7 +63,9 @@ class EffectPayloadTypesTest < ActiveSupport::TestCase [ [ '"result" => result', '"outcome" => result' ], [ '"error" => error', '"failure" => error' ], - [ '"backtrace" => Array(exception.backtrace).first(50)', '"backtrace" => [42]' ] + [ '"backtrace" => Array(exception.backtrace).first(50)', '"backtrace" => [42]' ], + [ '"outcome" => EffectRecoveryOutcome::RETIRED', '"outcome" => "recovered"' ], + [ '"outcome" => EffectRecoveryOutcome::COMPLETED, "result" => result', '"outcome" => EffectRecoveryOutcome::COMPLETED' ] ].each do |original, invalid| File.write(constructor_path, constructors.sub(original, invalid)) output, status = typecheck(project) diff --git a/test/integration/effect_recovery_test.rb b/test/integration/effect_recovery_test.rb new file mode 100644 index 0000000..bbaaef5 --- /dev/null +++ b/test/integration/effect_recovery_test.rb @@ -0,0 +1,346 @@ +# frozen_string_literal: true + +require "database_test_helper" +require "timeout" + +class EffectRecoveryTest < ActiveSupport::TestCase + class ExportActor < SolidObjects::Actor + actor_type "effect-recovery-export" + attribute :effect_handle, default: nil + attribute :notifications, default: -> { [] } + + def start_export + self.effect_handle = emit(:build_report, revision: 1) + end + + def start_recoverable_export + self.effect_handle = emit(:build_report, revision: 1, on_recovery: :retired) + end + + def retired(effect_id:, arguments:, outcome:) + self.notifications += [ { "effect_id" => effect_id, "arguments" => arguments, "outcome" => outcome } ] + end + + def start_checked_export + self.effect_handle = emit(:build_report, revision: 1, on_recovery: :retired, on_status: :checked, on_success: :completed) + end + + def start_with_timeout(timeout:) + self.effect_handle = emit(:build_report, revision: 1, on_recovery: :retired, on_status: :checked, recovery_timeout: timeout) + end + + def completed(effect_id:, arguments:, result:) + self.notifications += [ { "effect_id" => effect_id, "result" => result } ] + end + + def check_export + request_effect_recovery(effect_handle) + end + + def checked(effect_id:, outcome:, arguments: nil, result: nil) + self.notifications += [ { "effect_id" => effect_id, "outcome" => outcome } ] + end + end + + test "emit returns the persisted public effect identity" do + reference = ExportActor.ref("export") + reference.async.start_export + worker = SolidObjects::Worker.new + worker.run_until_idle + + effect = SolidObjects::Effect.find_by!(name: "build_report") + assert_equal({ "effect_id" => effect.effect_id }, effect.instance.state.fetch("effect_handle")) + ensure + worker&.stop + end + + test "false is not a recovery timeout" do + actor = ExportActor.new(actor_id: "invalid", state: SolidObjects::State.new(ExportActor.definition.state_definition)) + assert_raises(ArgumentError) { actor.start_with_timeout(timeout: false) } + assert_empty actor.send(:drain_effect_intents) + 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 + worker.run_until_idle + effect_executor = SolidObjects::EffectExecutor.new + effect = effect_executor.send(:claim_next) + owner = SolidObjects::Process.find(effect.claimed_by) + owner.update!(last_heartbeat_at: SolidObjects.database_adapter.database_clock_now - 75) + + SolidObjects::ProcessRegistry.cleanup_dead + + assert_equal "processing", effect.reload.status + assert_equal owner.id, effect.claimed_by + assert_empty SolidObjects::Message.where(operation: "retired") + owner.update!(stopped_at: Time.at(0)) + SolidObjects::ProcessPruner.new.prune + assert SolidObjects::Process.exists?(owner.id) + assert_nil effect_executor.send(:claim_next) + owner.update!(last_heartbeat_at: SolidObjects.database_adapter.database_clock_now - 125) + assert_nil effect_executor.send(:claim_next) + assert_equal 1, SolidObjects::Message.where(operation: "retired").count + ensure + effect_executor&.stop + worker&.stop + end + + test "explicit checks distinguish completed dead missing and already retired effects" do + reference = ExportActor.ref("outcomes") + reference.async.start_checked_export + worker = SolidObjects::Worker.new + worker.run_until_idle + effect = SolidObjects::Effect.find_by!(name: "build_report") + coordinator = SolidObjects::EffectRecoveryCoordinator.new + check = -> do + SolidObjects.database_adapter.transaction do + instance = SolidObjects::Instance.lock.find(effect.instance_id) + coordinator.check(instance:, intents: [ SolidObjects::Actor::EffectRecoveryIntent.new(effect_id: effect.effect_id, request_id: SecureRandom.uuid) ]) + end + SolidObjects::Message.where(operation: "checked").order(:sequence).last.arguments + end + + effect.update!(status: "completed", result: nil) + payload = check.call + assert_equal "completed", payload.fetch("outcome") + assert_nil payload.fetch("result") + assert_equal({ "revision" => 1 }, payload.fetch("arguments")) + effect.update!(status: "dead") + assert_equal "dead", check.call.fetch("outcome") + effect.destroy! + assert_equal "missing", check.call.fetch("outcome") + SolidObjects::EffectRecovery.find(effect.effect_id).update!(retired_at: Time.now.utc) + assert_equal "already_retired", check.call.fetch("outcome") + assert_empty SolidObjects::Message.where(operation: "retired") + ensure + worker&.stop + end + + test "rollback undoes retirement and both callback messages" do + ExportActor.ref("rollback").async.start_checked_export + worker = SolidObjects::Worker.new + worker.run_until_idle + effect_executor = SolidObjects::EffectExecutor.new + effect = effect_executor.send(:claim_next) + SolidObjects::Process.find(effect.claimed_by).update!(last_heartbeat_at: Time.at(0)) + + 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: SecureRandom.uuid) ]) + raise ActiveRecord::Rollback + 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 + effect_executor&.stop + worker&.stop + end + + test "cleanup retires abandoned opted-in effects and delivers one durable notification" do + ExportActor.ref("abandoned").async.start_recoverable_export + worker = SolidObjects::Worker.new + worker.run_until_idle + effect_executor = SolidObjects::EffectExecutor.new + effect = effect_executor.send(:claim_next) + process = SolidObjects::Process.find(effect.claimed_by) + process.update!(last_heartbeat_at: SolidObjects.database_adapter.database_now - 70) + + SolidObjects::ProcessRegistry.cleanup_dead + + assert_equal 1, SolidObjects::Message.where(operation: "retired").count + assert_nil effect_executor.send(:claim_next) + worker.run_until_idle + assert_equal [ { "effect_id" => effect.effect_id, "arguments" => { "revision" => 1 }, "outcome" => "retired" } ], + effect.instance.reload.state.fetch("notifications") + SolidObjects::ProcessRegistry.cleanup_dead + assert_equal 1, SolidObjects::Message.where(operation: "retired").count + ensure + effect_executor&.stop + worker&.stop + end + + test "an explicit pending check delivers status without retiring or changing attempts" do + reference = ExportActor.ref("checked") + reference.async.start_checked_export + worker = SolidObjects::Worker.new + worker.run_until_idle + effect = SolidObjects::Effect.find_by!(name: "build_report") + + reference.async.check_export + worker.run_until_idle + + assert_equal [ { "effect_id" => effect.effect_id, "outcome" => "pending" } ], + effect.instance.reload.state.fetch("notifications") + assert_equal "pending", effect.reload.status + assert_equal 0, effect.attempt_count + assert_empty SolidObjects::Message.where(operation: "retired") + ensure + worker&.stop + end + + test "retirement and late completion share instance before effect lock order" do + skip "PostgreSQL lock observation" unless database_family == :postgresql + + ExportActor.ref("lock-order").async.start_checked_export + worker = SolidObjects::Worker.new + worker.run_until_idle + effect_executor = SolidObjects::EffectExecutor.new + effect = effect_executor.send(:claim_next) + SolidObjects::Process.find(effect.claimed_by).update!(last_heartbeat_at: SolidObjects.database_adapter.database_now - 70) + completion_pid = Queue.new + completion_result = Queue.new + completer = nil + + SolidObjects.database_adapter.transaction do + instance = SolidObjects::Instance.lock.find(effect.instance_id) + blocker_pid = SolidObjects::Record.connection.select_value("SELECT pg_backend_pid()").to_i + completer = Thread.new do + SolidObjects::Record.connection_pool.with_connection do |connection| + completion_pid << connection.select_value("SELECT pg_backend_pid()").to_i + begin + effect_executor.send(:complete, effect, { "artifact_key" => "late" }) + completion_result << :completed + rescue => error + completion_result << error + end + end + end + pid = Timeout.timeout(5) { completion_pid.pop } + Timeout.timeout(5) do + until SolidObjects::Record.connection.select_value("SELECT #{blocker_pid} = ANY(pg_blocking_pids(#{pid}))") + Thread.pass + end + end + SolidObjects::EffectRecoveryCoordinator.new.check( + instance:, + intents: [ SolidObjects::Actor::EffectRecoveryIntent.new(effect_id: effect.effect_id, request_id: SecureRandom.uuid) ] + ) + end + + result = Timeout.timeout(5) { completion_result.pop } + assert_instance_of SolidObjects::LostActivation, result + assert SolidObjects::EffectRecovery.find(effect.effect_id).retired_at + assert_equal [ "retired", "checked" ], SolidObjects::Message.where(operation: [ "retired", "checked" ]).order(:sequence).pluck(:operation) + ensure + completer&.join(5) + effect_executor&.stop + worker&.stop + end + + test "recovery rechecks a heartbeat refreshed while waiting for its owner" do + skip "PostgreSQL lock observation" unless database_family == :postgresql + + worker, effect_executor, effect = processing_export("refreshed") + results = Queue.new + process_ids = Queue.new + recovery_thread = nil + SolidObjects.database_adapter.transaction do + owner = SolidObjects::Process.lock.find(effect.claimed_by) + recovery_thread = spawn_recovery(results:, process_ids:) + wait_for_blocked_process(Timeout.timeout(5) { process_ids.pop }) + owner.update!(last_heartbeat_at: SolidObjects.database_adapter.database_clock_now) + end + assert_equal :done, Timeout.timeout(5) { results.pop } + assert_equal "processing", effect.reload.status + assert_empty SolidObjects::Message.where(operation: "retired") + ensure + recovery_thread&.join(5) + effect_executor&.stop + worker&.stop + end + + test "two blocked automatic recovery passes enqueue one callback" do + skip "PostgreSQL lock observation" unless database_family == :postgresql + + worker, effect_executor, effect = processing_export("simultaneous") + results = Queue.new + process_ids = Queue.new + recovery_threads = [] + SolidObjects.database_adapter.transaction do + SolidObjects::Instance.lock.find(effect.instance_id) + 2.times { recovery_threads << spawn_recovery(results:, process_ids:) } + 2.times { wait_for_blocked_process(Timeout.timeout(5) { process_ids.pop }) } + end + 2.times { assert_equal :done, Timeout.timeout(5) { results.pop } } + assert_equal 1, SolidObjects::Message.where(operation: "retired").count + worker.run_until_idle + assert_equal 1, effect.instance.reload.state.fetch("notifications").length + ensure + recovery_threads&.each { |thread| thread.join(5) } + effect_executor&.stop + worker&.stop + end + + test "late failure leaves retirement and the existing recovery callback intact" do + worker, effect_executor, effect = processing_export("late-failure") + SolidObjects::EffectRecoveryCoordinator.new.recover_available + effect_executor.send(:fail_effect, effect, RuntimeError.new("late")) + assert_equal "completed", effect.reload.status + assert_equal 1, SolidObjects::Message.where(operation: "retired").count + assert_empty SolidObjects::Message.where(operation: "completed") + ensure + 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 + + worker, effect_executor, effect = processing_export("new-claimant") + replacement = SolidObjects::ProcessRegistry.new.register(kind: "effect") + results = Queue.new + process_ids = Queue.new + recovery_thread = nil + SolidObjects.database_adapter.transaction do + SolidObjects::Instance.lock.find(effect.instance_id) + locked = SolidObjects::Effect.lock.find(effect.id) + recovery_thread = spawn_recovery(results:, process_ids:) + wait_for_blocked_process(Timeout.timeout(5) { process_ids.pop }) + locked.update!(claimed_by: replacement.id) + end + assert_equal :done, Timeout.timeout(5) { results.pop } + assert_equal replacement.id, effect.reload.claimed_by + assert_empty SolidObjects::Message.where(operation: "retired") + ensure + recovery_thread&.join(5) + effect_executor&.stop + worker&.stop + end + + private + + def processing_export(actor_id) + ExportActor.ref(actor_id).async.start_checked_export + worker = SolidObjects::Worker.new + worker.run_until_idle + effect_executor = SolidObjects::EffectExecutor.new + effect = effect_executor.send(:claim_next) + SolidObjects::Process.find(effect.claimed_by).update!(last_heartbeat_at: Time.at(0)) + [ worker, effect_executor, effect ] + end + + def spawn_recovery(results:, process_ids:) + Thread.new do + SolidObjects::Record.connection_pool.with_connection do |connection| + process_ids << connection.select_value("SELECT pg_backend_pid()").to_i + begin + SolidObjects::EffectRecoveryCoordinator.new.recover_available + results << :done + rescue => error + results << error + end + end + end + end + + def wait_for_blocked_process(process_id) + Timeout.timeout(5) do + until SolidObjects::Record.connection.select_value("SELECT cardinality(pg_blocking_pids(#{process_id})) > 0") + Thread.pass + end + end + end +end diff --git a/test/integration/separate_database_test.rb b/test/integration/separate_database_test.rb index be6d899..16cf46c 100644 --- a/test/integration/separate_database_test.rb +++ b/test/integration/separate_database_test.rb @@ -55,7 +55,8 @@ def connect_solid_objects_to(database) [ CreateSolidObjectsTables, AddStateRevisionToSolidObjectsInstances, - RenameMessageDispatchColumns + RenameMessageDispatchColumns, + AddSolidObjectsEffectRecoveries ].each do |migration_class| migration = migration_class.new migration.define_singleton_method(:connection) { SolidObjects::Record.connection } diff --git a/test/types/effect_payloads.rb b/test/types/effect_payloads.rb index b65f7f3..bb3077c 100644 --- a/test/types/effect_payloads.rb +++ b/test/types/effect_payloads.rb @@ -1,6 +1,28 @@ # rbs_inline: enabled class EffectPayloadConsumer < SolidObjects::Actor + def recovery_result(payload) + return payload["result"] if payload["outcome"] == SolidObjects::EffectRecoveryOutcome::COMPLETED + + nil + end + + def retired(arguments) + { "effect_id" => "effect-1", "arguments" => arguments, "outcome" => SolidObjects::EffectRecoveryOutcome::RETIRED } + end + + def completed_recovery(arguments) + { "effect_id" => "effect-1", "arguments" => arguments, "outcome" => SolidObjects::EffectRecoveryOutcome::COMPLETED, "result" => nil } + end + + def retired_revision(payload) + payload["arguments"]["revision"] + end + + def retired_outcome + SolidObjects::EffectRecoveryOutcome::RETIRED + end + def fail_turn(effect_id:, arguments:, error:) arguments["generation"] end diff --git a/test/types/effect_payloads.rbs b/test/types/effect_payloads.rbs index 1c26d34..aefac3a 100644 --- a/test/types/effect_payloads.rbs +++ b/test/types/effect_payloads.rbs @@ -1,5 +1,11 @@ class EffectPayloadConsumer < SolidObjects::Actor type run_arguments = { "generation" => Integer } + type report_arguments = { "revision" => Integer } + def recovery_result: (SolidObjects::effect_completed_recovery_payload[report_arguments, String]) -> String? + def retired: (report_arguments) -> SolidObjects::effect_retired_payload[report_arguments] + def completed_recovery: (report_arguments) -> SolidObjects::effect_completed_recovery_payload[report_arguments, nil] + def retired_revision: (SolidObjects::effect_retired_payload[report_arguments]) -> Integer + def retired_outcome: () -> "retired" def fail_turn: (effect_id: String, arguments: run_arguments, error: SolidObjects::effect_error) -> Integer def success_value: (SolidObjects::effect_success_payload[run_arguments, String]) -> String From 09e05f9c187d1a427c8043cb39f352ce638651b4 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 12:04:47 -0700 Subject: [PATCH 2/5] fix: Bound recovery scans and verify crash safety --- benchmark/support.rb | 3 + docs/effect-recovery.md | 7 + examples/at_least_once/boot.rb | 4 +- lib/solid_objects/actor.rb | 17 +- .../effect_recovery_coordinator.rb | 24 +- sig/generated/lib/solid_objects/actor.rbs | 3 + .../effect_recovery_coordinator.rbs | 3 + test/integration/effect_recovery_test.rb | 221 ++++++++++++++++++ 8 files changed, 272 insertions(+), 10 deletions(-) diff --git a/benchmark/support.rb b/benchmark/support.rb index 2c15641..30136e4 100644 --- a/benchmark/support.rb +++ b/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| diff --git a/docs/effect-recovery.md b/docs/effect-recovery.md index f388780..6481485 100644 --- a/docs/effect-recovery.md +++ b/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 diff --git a/examples/at_least_once/boot.rb b/examples/at_least_once/boot.rb index c66c443..cd68a4c 100644 --- a/examples/at_least_once/boot.rb +++ b/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 diff --git a/lib/solid_objects/actor.rb b/lib/solid_objects/actor.rb index 2c40a15..50f89b2 100644 --- a/lib/solid_objects/actor.rb +++ b/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 diff --git a/lib/solid_objects/effect_recovery_coordinator.rb b/lib/solid_objects/effect_recovery_coordinator.rb index dea8600..9d0e398 100644 --- a/lib/solid_objects/effect_recovery_coordinator.rb +++ b/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}" diff --git a/sig/generated/lib/solid_objects/actor.rbs b/sig/generated/lib/solid_objects/actor.rbs index d123d8f..e64d41c 100644 --- a/sig/generated/lib/solid_objects/actor.rbs +++ b/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 diff --git a/sig/generated/lib/solid_objects/effect_recovery_coordinator.rbs b/sig/generated/lib/solid_objects/effect_recovery_coordinator.rbs index 75d4188..86abe4e 100644 --- a/sig/generated/lib/solid_objects/effect_recovery_coordinator.rbs +++ b/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 diff --git a/test/integration/effect_recovery_test.rb b/test/integration/effect_recovery_test.rb index bbaaef5..accf91f 100644 --- a/test/integration/effect_recovery_test.rb +++ b/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) From 2fdf5d09b1e984cecebc3a06122984b3e6098d4f Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 12:15:46 -0700 Subject: [PATCH 3/5] chore: Prepare version 0.15.0 --- CHANGELOG.md | 2 +- Gemfile.lock | 4 ++-- lib/solid_objects/version.rb | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 95deb9d..0a8e0ce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog -## Unreleased +## 0.15.0 - 2026-09-15 - Return a stable effect handle from every `emit`. Wrappers must return it; operations relying on an implicit `nil` result should return `nil` explicitly. diff --git a/Gemfile.lock b/Gemfile.lock index d99d59a..e641844 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - solid_objects (0.14.7) + solid_objects (0.15.0) actioncable (>= 7.1) actionpack (>= 7.1) actionview (>= 7.1) @@ -384,7 +384,7 @@ CHECKSUMS rubocop-rails-omakase (1.1.0) sha256=2af73ac8ee5852de2919abbd2618af9c15c19b512c4cfc1f9a5d3b6ef009109d ruby-progressbar (1.13.0) sha256=80fc9c47a9b640d6834e0dc7b3c94c9df37f08cb072b7761e4a71e22cff29b33 securerandom (0.4.1) sha256=cc5193d414a4341b6e225f0cb4446aceca8e50d5e1888743fac16987638ea0b1 - solid_objects (0.14.7) + solid_objects (0.15.0) sqlite3 (2.9.5-aarch64-linux-gnu) sha256=78075b6337d3d182c6d2b4691049ed45cd220826160c9ea18946bf6a1de200dc sqlite3 (2.9.5-aarch64-linux-musl) sha256=18c801185deb4adc01ddb281e8f672a39e3d1729979ca91e39439cd3eac0402d sqlite3 (2.9.5-arm-linux-gnu) sha256=1bdfca0c7d63998c60b0f4a8e3c8df2d33800ccc4abd2d612eddbbbc92a4c48b diff --git a/lib/solid_objects/version.rb b/lib/solid_objects/version.rb index ddd4d5a..1d09f5e 100644 --- a/lib/solid_objects/version.rb +++ b/lib/solid_objects/version.rb @@ -1,5 +1,5 @@ # rbs_inline: enabled module SolidObjects - VERSION = "0.14.7" + VERSION = "0.15.0" end From 15d2237942fb6c0b08ae5c8ed98d002e0f74ecc2 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 12:50:01 -0700 Subject: [PATCH 4/5] fix: Heartbeat while effect handlers run --- CHANGELOG.md | 2 + docs/effect-recovery.md | 6 ++ docs/roadmap.md | 3 +- lib/solid_objects.rb | 1 + lib/solid_objects/effect_executor.rb | 4 ++ lib/solid_objects/process_heartbeat.rb | 53 ++++++++++++++ .../lib/solid_objects/process_heartbeat.rbs | 35 +++++++++ test/integration/effect_recovery_test.rb | 72 +++++++++++++++++++ 8 files changed, 175 insertions(+), 1 deletion(-) create mode 100644 lib/solid_objects/process_heartbeat.rb create mode 100644 sig/generated/lib/solid_objects/process_heartbeat.rbs diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a8e0ce..5b1d994 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/effect-recovery.md b/docs/effect-recovery.md index 6481485..5282301 100644 --- a/docs/effect-recovery.md +++ b/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 diff --git a/docs/roadmap.md b/docs/roadmap.md index efa8e98..0261cfa 100644 --- a/docs/roadmap.md +++ b/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 diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb index d27f6b5..fcdb0a4 100644 --- a/lib/solid_objects.rb +++ b/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" diff --git a/lib/solid_objects/effect_executor.rb b/lib/solid_objects/effect_executor.rb index 592c6ac..e9d089e 100644 --- a/lib/solid_objects/effect_executor.rb +++ b/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 diff --git a/lib/solid_objects/process_heartbeat.rb b/lib/solid_objects/process_heartbeat.rb new file mode 100644 index 0000000..df0f6ed --- /dev/null +++ b/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 diff --git a/sig/generated/lib/solid_objects/process_heartbeat.rbs b/sig/generated/lib/solid_objects/process_heartbeat.rbs new file mode 100644 index 0000000..5eda6b7 --- /dev/null +++ b/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 diff --git a/test/integration/effect_recovery_test.rb b/test/integration/effect_recovery_test.rb index accf91f..5bccc55 100644 --- a/test/integration/effect_recovery_test.rb +++ b/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 From db979521bce16c1ea960122564090e0e3c0ad48a Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 13:08:09 -0700 Subject: [PATCH 5/5] fix: Resume heartbeats after database errors --- CHANGELOG.md | 2 ++ docs/effect-recovery.md | 5 ++++ lib/solid_objects/process_heartbeat.rb | 11 +++++++ .../lib/solid_objects/process_heartbeat.rbs | 3 ++ test/integration/effect_recovery_test.rb | 30 +++++++++++++++++++ 5 files changed, 51 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b1d994..0f5fc4a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/effect-recovery.md b/docs/effect-recovery.md index 5282301..e268f5a 100644 --- a/docs/effect-recovery.md +++ b/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 diff --git a/lib/solid_objects/process_heartbeat.rb b/lib/solid_objects/process_heartbeat.rb index df0f6ed..972b154 100644 --- a/lib/solid_objects/process_heartbeat.rb +++ b/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 diff --git a/sig/generated/lib/solid_objects/process_heartbeat.rbs b/sig/generated/lib/solid_objects/process_heartbeat.rbs index 5eda6b7..ccd37ef 100644 --- a/sig/generated/lib/solid_objects/process_heartbeat.rbs +++ b/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 diff --git a/test/integration/effect_recovery_test.rb b/test/integration/effect_recovery_test.rb index 5bccc55..4ff4bfc 100644 --- a/test/integration/effect_recovery_test.rb +++ b/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