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..0f5fc4a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,20 @@ # Changelog +## 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. + 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 + `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/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/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/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/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..e268f5a --- /dev/null +++ b/docs/effect-recovery.md @@ -0,0 +1,203 @@ +# 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. + +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. + +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 +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. + +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 +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..0261cfa 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -14,6 +14,11 @@ 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 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 - Actor-to-actor asynchronous outbox delivery. Effects and broadcasts use 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.rb b/lib/solid_objects.rb index dcb48a7..fcdb0a4 100644 --- a/lib/solid_objects.rb +++ b/lib/solid_objects.rb @@ -56,6 +56,8 @@ require "solid_objects/polling_backoff" 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/actor.rb b/lib/solid_objects/actor.rb index 63f9e15..50f89b2 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,36 @@ 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) + validate_recovery_timeout!(timeout: recovery_timeout, operation: on_recovery) + 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 +382,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 +405,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 +414,7 @@ def discard_intents private attr_reader :effect_intents, + :effect_recovery_intents, :commit_action_intents, :reminder_intents, :outbound_message_intents @@ -412,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/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..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 @@ -113,6 +117,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 +183,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 +219,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..9d0e398 --- /dev/null +++ b/lib/solid_objects/effect_recovery_coordinator.rb @@ -0,0 +1,143 @@ +# 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 + recovery_candidates.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 () -> 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}" + 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_heartbeat.rb b/lib/solid_objects/process_heartbeat.rb new file mode 100644 index 0000000..972b154 --- /dev/null +++ b/lib/solid_objects/process_heartbeat.rb @@ -0,0 +1,64 @@ +# 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 } + rescue => error + report_failure(error) + 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 (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 + condition.wait(mutex, SolidObjects.configuration.process_heartbeat_interval) unless @stopped + @stopped + end + end + 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/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 diff --git a/sig/generated/lib/solid_objects/actor.rbs b/sig/generated/lib/solid_objects/actor.rbs index 3a7bfdf..e64d41c 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: () -> [ :name, :arguments, :success_operation, :failure_operation ] + def self.members: () -> [ :effect_id, :name, :arguments, :success_operation, :failure_operation, :recovery_operation, :status_operation, :recovery_timeout ] - def members: () -> [ :name, :arguments, :success_operation, :failure_operation ] + 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.new: (untyped effect_id, untyped request_id) -> instance + | (effect_id: untyped, request_id: untyped) -> instance + + 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 @@ -246,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/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..86abe4e --- /dev/null +++ b/sig/generated/lib/solid_objects/effect_recovery_coordinator.rbs @@ -0,0 +1,41 @@ +# 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 () -> 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 + + # @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/lib/solid_objects/process_heartbeat.rbs b/sig/generated/lib/solid_objects/process_heartbeat.rbs new file mode 100644 index 0000000..ccd37ef --- /dev/null +++ b/sig/generated/lib/solid_objects/process_heartbeat.rbs @@ -0,0 +1,38 @@ +# 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 (Exception) -> void + def report_failure: (Exception) -> void + + # @rbs () -> bool + def wait_for_interval: () -> bool + 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..4ff4bfc --- /dev/null +++ b/test/integration/effect_recovery_test.rb @@ -0,0 +1,669 @@ +# 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 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 "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 + 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 "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 + + 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 + + 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) + 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