diff --git a/CHANGELOG.md b/CHANGELOG.md index 56cabce..ca75bce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,28 @@ # Changelog +## 0.17.5 - 2026-10-12 + +- Back off repeated interrupted-message recovery using `retry_delay`. + The first claimed attempt recovers immediately; after attempts 2, 3, 4, + and onward, the delay callback receives 1, 2, 3, and onward (defaults: + 1, 2, 4 seconds, capped at 60 seconds). Total persisted claim attempts, + including earlier handler failures, determine the delay. The ready timestamp + survives worker replacement and preserves mailbox order while other actors + can proceed. Lease loss alone still does not dead-letter a message; handler + failure attempt limits, setting names, defaults, and schema stay unchanged. +- Emit `activation.lost` with the message ID and total attempt count when a + turn loses its fence, including while saving a handler failure or rejection, + matching JavaScript instrumentation. Shutdown requested + by that callback stops the worker before another pass. Skip a cached actor + whose first message is delayed so later ready messages cannot starve peers. + +- Preserve caller lifecycle guarantees with regression tests for concurrent + synchronous calls retaining their staged effects and for caller process + re-registration after stale-process cleanup and retention. Both cases + already work in Ruby; no runtime behavior or configuration changes. + The concurrency test waits for the second caller's busy-actor wait before + allowing the first caller to commit its staged effect. + ## 0.17.4 - 2026-10-12 - Renew activations between turns in a drain pass when renewal is due, diff --git a/Gemfile.lock b/Gemfile.lock index 50fa84c..28eea9e 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - solid_objects (0.17.4) + solid_objects (0.17.5) actioncable (>= 7.1) actionpack (>= 7.1) actionview (>= 7.1) @@ -392,7 +392,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.17.4) + solid_objects (0.17.5) 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/compatibility/telemetry-events.json b/compatibility/telemetry-events.json index 19669db..fb5ccea 100644 --- a/compatibility/telemetry-events.json +++ b/compatibility/telemetry-events.json @@ -84,6 +84,21 @@ "sequence" ] }, + { + "name": "activation.lost", + "attributes": [ + "actorId", + "actorType", + "attempt", + "deliveryMode", + "durationMilliseconds", + "instanceId", + "messageId", + "operation", + "requestId", + "sequence" + ] + }, { "name": "activation.started", "attributes": ["actorId", "actorType", "generation", "instanceId", "ownerId"], diff --git a/docs/agents.md b/docs/agents.md index 8fce4a6..5fffef7 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -173,6 +173,9 @@ Obey these rules in actor code: write in the same database. - Do not call an external API in a handler. Use `emit` and an effect handler. - Write each handler so that it can run again. Delivery is at least once. +- Repeated lease losses back off but do not become dead letters. Observe + `activation.lost` and set an application stopping policy if a hard bound is + required; increasing the handler-failure attempt limit does not bound lease loss. - A reminder changes state only when it runs, and it runs only while `solid_objects start` runs. Do not compute expiry from the clock in a query; read the state that the reminder committed. diff --git a/docs/observability.md b/docs/observability.md index 3892f2d..e732b16 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -45,6 +45,7 @@ JavaScript does not await exporters. Telemetry is not a durable audit trail. | ------------------------------------------- | ---------------------------------------------------------------------------- | | `activation.started/completed/failed` | Local actor activation hook lifecycle | | `message.started/completed/failed/rejected` | One attempt's execution outcome | +| `activation.lost` | A message turn lost its activation fence; includes message ID and total attempt count | | `message.retry` | Failed attempt durably queued for another attempt | | `dead_letter.created` | Message exhausted retries or failed permanently | | `commit_action.started/completed/failed` | One registered commit action inside the commit transaction | diff --git a/docs/operations.md b/docs/operations.md index 12aaaf6..137e9b1 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -606,3 +606,24 @@ heartbeat cleanup. Back up actor tables with the same consistency guarantees as application data. Restoring only instances without their mailboxes/outboxes, or vice versa, can violate application expectations. + +## Interrupted-message recovery + +An interrupted first claimed attempt recovers immediately. Recovery after +persisted attempts 2, 3, 4, ... calls `retry_delay` with 1, 2, 3, ... . The +callback returns seconds; defaults grow from 1 to 2 to 4 seconds up to 60 seconds. +This uses total claim attempts, including earlier handler failures, not a +separate lease-loss counter. The delay is stored in ready-message availability +when a stale claim is recovered, survives worker replacement, and keeps later +messages behind that message. Other actors can still run. Custom delays, +including zero, remain authoritative. + +Lease loss alone does not create a dead letter, even with `max_attempts: 1`. +Handler failures still use the full attempt count for `max_attempts` and +`retry_delay`. An always-losing message remains pending indefinitely. Observe +`activation.lost` (message ID and total attempt count) and impose an application +stopping policy if a hard bound is required. Calling `request_shutdown` from +instrumentation stops the worker before its next pass. Ruby uses a separate +renewal thread and waits after lost turns; it does not have the JavaScript +microtask starvation pattern. A handler must still return before its worker +can finish shutting down. diff --git a/docs/roadmap.md b/docs/roadmap.md index 5d771d1..aac489e 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -25,7 +25,12 @@ repeatable read. The tests count statements, and do not measure latency. - Activation leases, renewal between turns as well as in the renewal thread, unique activation tokens, generations, and - fenced commits + fenced commits. Interrupted first attempts recover immediately; repeated + interruptions use durable retry delays from total claim attempts minus one, + preserving mailbox order and progress for other actors. Lease loss emits + `activation.lost` with the message ID and attempt, including losses while + saving a handler failure or rejection. There is no hard retry cap; + applications needing one must impose a stopping policy. - Bounded activation passes, idle cache, hot-actor yield, and process records - At-least-once retries, terminal domain rejection, strict poison ordering, dead letters, and tail retry diff --git a/docs/virtual-actors.md b/docs/virtual-actors.md index 6097796..8970c0a 100644 --- a/docs/virtual-actors.md +++ b/docs/virtual-actors.md @@ -170,6 +170,10 @@ again without harm. a crash or a lost lease. - Ordered turns do not cancel stale Ruby code. Fencing stops a stale activation from a commit, but that code can continue to run. +- An interrupted first attempt recovers immediately. Repeated interruptions + use growing, durable retry delays while keeping mailbox order. Lease loss + alone has no attempt limit; use instrumentation and an application stopping + policy when a hard bound is required. - External effects can run more than once. Use the stable effect ID, or another durable key, as the idempotency key at the provider. - There are no transactions across actor identities, and there is no replay of diff --git a/lib/solid_objects/activation.rb b/lib/solid_objects/activation.rb index e663177..c4964d1 100644 --- a/lib/solid_objects/activation.rb +++ b/lib/solid_objects/activation.rb @@ -52,7 +52,8 @@ def drain_until(message_id:, deadline:) # @rbs () -> bool def ready? now = SolidObjects.database_adapter.database_now - ReadyMessage.where(instance_id: lease.instance_id, available_at: ..now).exists? || + ready_message = ReadyMessage.where(instance_id: lease.instance_id).order(:sequence).first + (ready_message && ready_message.available_at <= now) || ClaimedMessage.where( instance_id: lease.instance_id, process_id: lease.owner_id, @@ -227,11 +228,12 @@ def recover_stale_claim(instance) claimed_message.destroy! return if message.completed? || message.dead? + delay = (message.attempt_count <= 1) ? 0 : SolidObjects.configuration.retry_delay.call(message.attempt_count - 1) ReadyMessage.create!( message:, instance:, sequence: message.sequence, - available_at: SolidObjects.database_adapter.database_now + available_at: SolidObjects.database_adapter.database_now + delay ) end diff --git a/lib/solid_objects/executor.rb b/lib/solid_objects/executor.rb index b93d2d7..bb92af8 100644 --- a/lib/solid_objects/executor.rb +++ b/lib/solid_objects/executor.rb @@ -16,6 +16,18 @@ def initialize(activation:, message:) # @rbs () -> bool def call + execute + rescue LostActivation + SolidObjects.instrument(:"activation.lost", **instrumentation_payload, duration_milliseconds: elapsed_milliseconds) + raise + end + + private + + attr_reader :activation, :message + + # @rbs () -> bool + def execute state_before = actor.state.to_h observables_before = actor.observable_values message_context = MessageContext.new( @@ -53,10 +65,6 @@ def call false end - private - - attr_reader :activation, :message - # @rbs () -> Actor def actor activation.actor diff --git a/lib/solid_objects/version.rb b/lib/solid_objects/version.rb index 9816c7d..cfc0521 100644 --- a/lib/solid_objects/version.rb +++ b/lib/solid_objects/version.rb @@ -1,5 +1,5 @@ # rbs_inline: enabled module SolidObjects - VERSION = "0.17.4" + VERSION = "0.17.5" end diff --git a/sig/generated/lib/solid_objects/executor.rbs b/sig/generated/lib/solid_objects/executor.rbs index f4250b9..d9c51d9 100644 --- a/sig/generated/lib/solid_objects/executor.rbs +++ b/sig/generated/lib/solid_objects/executor.rbs @@ -22,6 +22,9 @@ module SolidObjects attr_reader message: untyped + # @rbs () -> bool + def execute: () -> bool + # @rbs () -> Actor def actor: () -> Actor diff --git a/test/integration/activation_recovery_backoff_test.rb b/test/integration/activation_recovery_backoff_test.rb new file mode 100644 index 0000000..2a3d6a2 --- /dev/null +++ b/test/integration/activation_recovery_backoff_test.rb @@ -0,0 +1,240 @@ +# rbs_inline: enabled + +require "database_test_helper" +require "portable_telemetry_assertions" + +class ActivationRecoveryBackoffTest < ActiveSupport::TestCase + include PortableTelemetryAssertions + + class Counter < SolidObjects::Actor + actor_type "activation-recovery-backoff" + attribute :count, default: 0 + + class << self + attr_accessor :before_commit + end + + def increment(label:) + self.count += 1 + emit(:applied, label:) + self.class.before_commit&.call(label) + count + end + end + + setup do + SolidObjects.configuration.lease_duration = 0.2 + SolidObjects.configuration.lease_renewal_interval = 0.05 + SolidObjects.configuration.max_attempts = 1 + @clock = { now: SolidObjects.database_adapter.database_now } + clock = @clock + SolidObjects.database_adapter.define_singleton_method(:database_now) { clock.fetch(:now) } + Counter.before_commit = ->(label) { clock[:now] += 0.201 if label == "slow" } + @events = [] + SolidObjects.configuration.instrumentation = ->(event) { @events << event } + @workers = [] + end + + teardown do + @workers.each(&:stop) + SolidObjects.database_adapter.singleton_class.remove_method(:database_now) + Counter.before_commit = nil + end + + test "recovers once immediately then delays without blocking other actors or reordering" do + worker = new_worker + first = Counter.ref("one").async.increment(label: "slow") + later = Counter.ref("one").async.increment(label: "later") + assert_equal 0, worker.run_once + assert_equal 0, worker.run_once + assert_equal 2, stored_message(first).attempt_count + assert_equal 0, worker.run_once + assert_in_delta 1, remaining_delay(first) + other = Counter.ref("other").async.increment(label: "other") + assert_equal 1, worker.run_once + assert_equal 1, stored_message(other).result + assert_equal 0, stored_message(later).attempt_count + assert_empty stored_message(first).instance.state + + @clock[:now] += 0.999 + assert_equal 0, worker.run_once + @clock[:now] += 0.001 + assert_equal 0, worker.run_once + assert_equal 3, stored_message(first).attempt_count + assert_equal 0, worker.run_once + assert_in_delta 2, remaining_delay(first) + assert_empty SolidObjects::DeadLetter.all + + Counter.before_commit = nil + @clock[:now] += 2 + assert_equal 2, worker.run_once + assert_equal 1, stored_message(first).result + assert_equal 2, stored_message(later).result + assert_equal %w[later other slow], SolidObjects::Effect.all.map { |effect| effect.arguments.fetch("label") }.sort + end + + %w[stop cleanup replacement].each do |recovery| + test "preserves custom recovery waits through #{recovery}" do + delay_arguments = [] + SolidObjects.configuration.retry_delay = ->(attempt) { + delay_arguments << attempt + attempt * 0.025 + } + worker = new_worker + reference = Counter.ref("one").async.increment(label: "slow") + 2.times { worker.run_once } + worker.stop if recovery == "stop" + if recovery == "cleanup" + @clock[:now] += SolidObjects.configuration.process_alive_threshold + 1 + SolidObjects::ProcessRegistry.cleanup_dead + end + replacement = new_worker + assert_equal 0, replacement.run_once + assert_equal [ 1 ], delay_arguments + assert_in_delta 0.025, remaining_delay(reference) + worker.stop + assert_in_delta 0.025, remaining_delay(reference) + + @clock[:now] += 0.025 + replacement.run_once + assert_equal 3, stored_message(reference).attempt_count + assert_equal 0, replacement.run_once + assert_equal [ 1, 2 ], delay_arguments + assert_in_delta 0.05, remaining_delay(reference) + end + end + + test "caps default recovery waits without dead lettering lease loss" do + worker = new_worker + reference = Counter.ref("one").async.increment(label: "slow") + 2.times { worker.run_once } + [ 1, 2, 4, 8, 16, 32, 60, 60 ].each do |delay| + worker.run_once + assert_in_delta delay, remaining_delay(reference) + assert_empty SolidObjects::DeadLetter.all + @clock[:now] += delay + worker.run_once + end + end + + %w[completion failure rejection].each do |outcome| + test "reports each lost #{outcome} and stops from its instrumentation callback" do + worker = new_worker + reference = Counter.ref("one").async.increment(label: "slow") + Counter.before_commit = ->(_) do + @clock[:now] += 0.201 + raise SolidObjects::Rejected.new(code: "declined", message: "declined") if outcome == "rejection" + raise "handler failed" if outcome == "failure" + end + observed = [] + SolidObjects.configuration.retry_delay = ->(_) { 0 } + SolidObjects.configuration.instrumentation = ->(event) do + next unless event.fetch("name") == "solid_objects.activation.lost" + + observed << event + worker.request_shutdown if observed.length == 3 + end + forced_shutdown = false + guard = Thread.new do + sleep 2 + forced_shutdown = true + worker.request_shutdown + end + begin + worker.run + ensure + guard.kill + guard.join + end + assert_not forced_shutdown + assert_predicate worker, :stopped? + assert_equal [ 1, 2, 3 ], observed.map { |event| event.fetch("attempt") } + assert observed.all? { |event| event.fetch("messageId") == reference.id.to_s } + observed.each { |event| assert_portable_attributes(event) } + assert worker.shutdown_requested? + assert_empty stored_message(reference).instance.state + assert_empty SolidObjects::DeadLetter.all + end + end + + %w[failure rejection].each do |outcome| + test "reports lease loss while saving a handler #{outcome} and recovers" do + worker = new_worker + reference = Counter.ref("one").async.increment(label: "slow") + Counter.before_commit = ->(_) do + @clock[:now] += 0.201 + raise SolidObjects::Rejected.new(code: "declined", message: "declined") if outcome == "rejection" + + raise "handler failed" + end + assert_equal 0, worker.run_once + losses = @events.select { |event| event.fetch("name") == "solid_objects.activation.lost" } + assert_equal [ 1 ], losses.map { |event| event.fetch("attempt") } + assert_equal [ reference.id.to_s ], losses.map { |event| event.fetch("messageId") } + losses.each { |event| assert_portable_attributes(event) } + assert_empty stored_message(reference).instance.state + assert_nil stored_message(reference).error + assert_nil stored_message(reference).rejection + assert_empty SolidObjects::DeadLetter.all + assert_empty SolidObjects::Effect.all + assert_not @events.any? { |event| %w[solid_objects.message.failed solid_objects.message.rejected].include?(event.fetch("name")) } + + Counter.before_commit = nil + assert_equal 1, worker.run_once + assert_equal 1, stored_message(reference).result + assert_equal 2, stored_message(reference).attempt_count + assert_equal 1, SolidObjects::Effect.count + end + end + + test "keeps a handler failure terminal with one allowed attempt" do + Counter.before_commit = ->(_) { raise "handler failure" } + worker = new_worker + reference = Counter.ref("one").async.increment(label: "fail") + worker.run_once + assert_predicate stored_message(reference), :dead? + assert_equal 1, stored_message(reference).attempt_count + end + + test "uses total claims for recovery after an earlier handler failure" do + delay_arguments = [] + SolidObjects.configuration.max_attempts = 3 + SolidObjects.configuration.retry_delay = ->(attempt) { + delay_arguments << attempt + attempt * 0.025 + } + worker = new_worker + reference = Counter.ref("mixed").async.increment(label: "slow") + Counter.before_commit = ->(_) { raise "handler failure" } + worker.run_once + assert_equal [ 1 ], delay_arguments + @clock[:now] += 0.025 + Counter.before_commit = ->(_) { @clock[:now] += 0.201 } + worker.run_once + assert_equal 0, worker.run_once + assert_equal [ 1, 1 ], delay_arguments + assert_in_delta 0.025, remaining_delay(reference) + assert_equal 2, stored_message(reference).attempt_count + @clock[:now] += 0.025 + Counter.before_commit = ->(_) { raise "handler failed again" } + worker.run_once + assert_predicate stored_message(reference), :dead? + assert_equal 3, stored_message(reference).attempt_count + end + + private + + def new_worker + SolidObjects::Worker.new.tap { |worker| @workers << worker } + end + + def stored_message(reference) + SolidObjects::Message.find(reference.id) + end + + def remaining_delay(reference) + ready = SolidObjects::ReadyMessage.find_by(message_id: reference.id) + assert ready, "expected recovered claim to wait in ready membership" + ready.available_at - @clock.fetch(:now) + end +end diff --git a/test/integration/caller_lifecycle_test.rb b/test/integration/caller_lifecycle_test.rb new file mode 100644 index 0000000..4017653 --- /dev/null +++ b/test/integration/caller_lifecycle_test.rb @@ -0,0 +1,120 @@ +# rbs_inline: enabled + +require "database_test_helper" +require "timeout" + +class CallerLifecycleTest < ActiveSupport::TestCase + class BlockingWakeUp + # @rbs @waiting: Thread::Queue + # @rbs @release: Thread::Queue + + attr_reader :waiting, :release + + # @rbs () -> void + def initialize + @waiting = Queue.new + @release = Queue.new + end + + # @rbs () -> void + def signal + end + + # @rbs (timeout: Numeric) -> void + def wait(timeout:) + waiting << timeout + release.pop + end + end + + class Counter < SolidObjects::Actor + actor_type "caller-lifecycle-counter" + attribute :count, default: 0 + + class << self + attr_accessor :started, :release + end + + # @rbs (label: String) -> Integer + def increment(label:) + self.count += 1 + emit(:caller_effect, label:) + if self.class.started + self.class.started << label + self.class.release.pop + end + count + end + end + + setup do + SolidObjects.configuration.max_attempts = 1 + end + + teardown do + Counter.started = nil + Counter.release = nil + end + + test "concurrent callers preserve staged effects and ordered state" do + Counter.started = Queue.new + Counter.release = Queue.new + wake_up = BlockingWakeUp.new + SolidObjects.configuration.wake_up_adapter = wake_up + reference = Counter.ref("concurrent") + + first = Thread.new { reference.increment(label: "first") } + assert_equal "first", Timeout.timeout(10) { Counter.started.pop } + second = Thread.new { reference.increment(label: "second") } + Timeout.timeout(10) { wake_up.waiting.pop } + assert_raises(ThreadError) { Counter.started.pop(true) } + assert_equal 1, SolidObjects::ClaimedMessage.count + assert_equal 1, SolidObjects::ReadyMessage.count + assert_empty SolidObjects::Effect.all + Counter.release << true + assert_equal 1, Timeout.timeout(10) { first.value } + wake_up.release << true + assert_equal "second", Timeout.timeout(10) { Counter.started.pop } + Counter.release << true + assert_equal 2, Timeout.timeout(10) { second.value } + + instance = SolidObjects::Instance.find_by!(actor_type: Counter.actor_type, actor_id: "concurrent") + assert_equal({ "count" => 2 }, instance.state) + assert_equal [ "first", "second" ], SolidObjects::Effect.all.map { |effect| effect.arguments.fetch("label") }.sort + assert_equal [ 1, 1 ], SolidObjects::Message.order(:sequence).pluck(:attempt_count) + assert_empty SolidObjects::ClaimedMessage.all + assert_empty SolidObjects::DeadLetter.all + ensure + Counter.release << true if first&.alive? + Counter.release << true if second&.alive? + wake_up&.release&.push(true) if second&.alive? + first&.join(10) + second&.join(10) + end + + test "caller re-registers after stale cleanup and process retention" do + reference = Counter.ref("retained-process") + assert_equal 1, reference.increment(label: "before") + original_process = SolidObjects.caller_process.process_registry.process_record + original_process.update!(last_heartbeat_at: 2.minutes.ago) + assert_equal 1, SolidObjects::ProcessRegistry.cleanup_dead + assert_equal "stopped", original_process.reload.shutdown_state + + assert_equal 2, reference.increment(label: "after-cleanup") + stopped_process = SolidObjects.caller_process.process_registry.process_record + refute_equal original_process.id, stopped_process.id + assert_equal "running", stopped_process.shutdown_state + stopped_process.update!(last_heartbeat_at: 2.minutes.ago) + assert_equal 1, SolidObjects::ProcessRegistry.cleanup_dead + SolidObjects::Process.where(id: [ original_process.id, stopped_process.id ]).update_all(stopped_at: 8.days.ago) + assert_equal 2, SolidObjects::ProcessPruner.new.prune + refute SolidObjects::Process.exists?(stopped_process.id) + + assert_equal 3, reference.increment(label: "after-retention") + replacement = SolidObjects.caller_process.process_registry.process_record + refute_equal stopped_process.id, replacement.id + assert_equal "running", replacement.shutdown_state + assert_equal [ 1, 1, 1 ], SolidObjects::Message.order(:sequence).pluck(:attempt_count) + assert_empty SolidObjects::DeadLetter.all + end +end