Skip to content

Commit a0e1988

Browse files
committed
fix: back off interrupted message recovery
Independently reported lease loss could repeatedly reclaim a message without a delay. Recover attempt one immediately, then persist retry_delay for total claim attempts minus one. Keep fences, mailbox order, and handler failure limits unchanged. Emit activation.lost with message identity and attempt. Skip cached actors whose mailbox head is delayed so other actors can progress. Document the remaining need for an application hard bound and match the JS contract. Seven focused regressions fail on the prior head and pass with the fix; the handler-failure control stays green. Validate SQLite, PostgreSQL, mysql2, Trilogy, and the full lint, type, and security checks.
1 parent 639efef commit a0e1988

10 files changed

Lines changed: 273 additions & 4 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,19 @@
22

33
## Unreleased
44

5+
- Back off repeated interrupted-message recovery using `retry_delay`.
6+
The first claimed attempt recovers immediately; after attempts 2, 3, 4,
7+
and onward, the delay callback receives 1, 2, 3, and onward (defaults:
8+
1, 2, 4 seconds, capped at 60 seconds). Total persisted claim attempts,
9+
including earlier handler failures, determine the delay. The ready timestamp
10+
survives worker replacement and preserves mailbox order while other actors
11+
can proceed. Lease loss alone still does not dead-letter a message; handler
12+
failure attempt limits, setting names, defaults, and schema stay unchanged.
13+
- Emit `activation.lost` with the message ID and total attempt count when a
14+
turn loses its fence, matching JavaScript instrumentation. Shutdown requested
15+
by that callback stops the worker before another pass. Skip a cached actor
16+
whose first message is delayed so later ready messages cannot starve peers.
17+
518
- Preserve caller lifecycle guarantees with regression tests for concurrent
619
synchronous calls retaining their staged effects and for caller process
720
re-registration after stale-process cleanup and retention. Both cases

‎compatibility/telemetry-events.json‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,21 @@
8484
"sequence"
8585
]
8686
},
87+
{
88+
"name": "activation.lost",
89+
"attributes": [
90+
"actorId",
91+
"actorType",
92+
"attempt",
93+
"deliveryMode",
94+
"durationMilliseconds",
95+
"instanceId",
96+
"messageId",
97+
"operation",
98+
"requestId",
99+
"sequence"
100+
]
101+
},
87102
{
88103
"name": "activation.started",
89104
"attributes": ["actorId", "actorType", "generation", "instanceId", "ownerId"],

‎docs/agents.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,9 @@ Obey these rules in actor code:
173173
write in the same database.
174174
- Do not call an external API in a handler. Use `emit` and an effect handler.
175175
- Write each handler so that it can run again. Delivery is at least once.
176+
- Repeated lease losses back off but do not become dead letters. Observe
177+
`activation.lost` and set an application stopping policy if a hard bound is
178+
required; increasing the handler-failure attempt limit does not bound lease loss.
176179
- A reminder changes state only when it runs, and it runs only while
177180
`solid_objects start` runs. Do not compute expiry from the clock in a query;
178181
read the state that the reminder committed.

‎docs/observability.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ JavaScript does not await exporters. Telemetry is not a durable audit trail.
4545
| ------------------------------------------- | ---------------------------------------------------------------------------- |
4646
| `activation.started/completed/failed` | Local actor activation hook lifecycle |
4747
| `message.started/completed/failed/rejected` | One attempt's execution outcome |
48+
| `activation.lost` | A message turn lost its activation fence; includes message ID and total attempt count |
4849
| `message.retry` | Failed attempt durably queued for another attempt |
4950
| `dead_letter.created` | Message exhausted retries or failed permanently |
5051
| `commit_action.started/completed/failed` | One registered commit action inside the commit transaction |

‎docs/operations.md‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -606,3 +606,24 @@ heartbeat cleanup.
606606
Back up actor tables with the same consistency guarantees as application data.
607607
Restoring only instances without their mailboxes/outboxes, or vice versa, can
608608
violate application expectations.
609+
610+
## Interrupted-message recovery
611+
612+
An interrupted first claimed attempt recovers immediately. Recovery after
613+
persisted attempts 2, 3, 4, ... calls `retry_delay` with 1, 2, 3, ... . The
614+
callback returns seconds; defaults grow from 1 to 2 to 4 seconds up to 60 seconds.
615+
This uses total claim attempts, including earlier handler failures, not a
616+
separate lease-loss counter. The delay is stored in ready-message availability
617+
when a stale claim is recovered, survives worker replacement, and keeps later
618+
messages behind that message. Other actors can still run. Custom delays,
619+
including zero, remain authoritative.
620+
621+
Lease loss alone does not create a dead letter, even with `max_attempts: 1`.
622+
Handler failures still use the full attempt count for `max_attempts` and
623+
`retry_delay`. An always-losing message remains pending indefinitely. Observe
624+
`activation.lost` (message ID and total attempt count) and impose an application
625+
stopping policy if a hard bound is required. Calling `request_shutdown` from
626+
instrumentation stops the worker before its next pass. Ruby uses a separate
627+
renewal thread and waits after lost turns; it does not have the JavaScript
628+
microtask starvation pattern. A handler must still return before its worker
629+
can finish shutting down.

‎docs/roadmap.md‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,11 @@
2525
repeatable read. The tests count statements, and do not measure latency.
2626
- Activation leases, renewal between turns as well as in the renewal thread,
2727
unique activation tokens, generations, and
28-
fenced commits
28+
fenced commits. Interrupted first attempts recover immediately; repeated
29+
interruptions use durable retry delays from total claim attempts minus one,
30+
preserving mailbox order and progress for other actors. Lease loss emits
31+
`activation.lost` with the message ID and attempt and has no hard retry cap;
32+
applications needing one must impose a stopping policy.
2933
- Bounded activation passes, idle cache, hot-actor yield, and process records
3034
- At-least-once retries, terminal domain rejection, strict poison ordering,
3135
dead letters, and tail retry

‎docs/virtual-actors.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,10 @@ again without harm.
170170
a crash or a lost lease.
171171
- Ordered turns do not cancel stale Ruby code. Fencing stops a stale activation
172172
from a commit, but that code can continue to run.
173+
- An interrupted first attempt recovers immediately. Repeated interruptions
174+
use growing, durable retry delays while keeping mailbox order. Lease loss
175+
alone has no attempt limit; use instrumentation and an application stopping
176+
policy when a hard bound is required.
173177
- External effects can run more than once. Use the stable effect ID, or
174178
another durable key, as the idempotency key at the provider.
175179
- There are no transactions across actor identities, and there is no replay of

‎lib/solid_objects/activation.rb‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,8 @@ def drain_until(message_id:, deadline:)
5252
# @rbs () -> bool
5353
def ready?
5454
now = SolidObjects.database_adapter.database_now
55-
ReadyMessage.where(instance_id: lease.instance_id, available_at: ..now).exists? ||
55+
ready_message = ReadyMessage.where(instance_id: lease.instance_id).order(:sequence).first
56+
(ready_message && ready_message.available_at <= now) ||
5657
ClaimedMessage.where(
5758
instance_id: lease.instance_id,
5859
process_id: lease.owner_id,
@@ -227,11 +228,12 @@ def recover_stale_claim(instance)
227228
claimed_message.destroy!
228229
return if message.completed? || message.dead?
229230

231+
delay = (message.attempt_count <= 1) ? 0 : SolidObjects.configuration.retry_delay.call(message.attempt_count - 1)
230232
ReadyMessage.create!(
231233
message:,
232234
instance:,
233235
sequence: message.sequence,
234-
available_at: SolidObjects.database_adapter.database_now
236+
available_at: SolidObjects.database_adapter.database_now + delay
235237
)
236238
end
237239

‎lib/solid_objects/executor.rb‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,10 @@ def call
3939
state_changed: state_after.value != state_before
4040
)
4141
true
42-
rescue CommittedTransactionError, LostActivation
42+
rescue CommittedTransactionError
43+
raise
44+
rescue LostActivation
45+
SolidObjects.instrument(:"activation.lost", **instrumentation_payload, duration_milliseconds: elapsed_milliseconds)
4346
raise
4447
rescue Rejected => rejection
4548
activation.restore_state(state_before) if state_before
Lines changed: 203 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,203 @@
1+
# rbs_inline: enabled
2+
3+
require "database_test_helper"
4+
require "portable_telemetry_assertions"
5+
6+
class ActivationRecoveryBackoffTest < ActiveSupport::TestCase
7+
include PortableTelemetryAssertions
8+
9+
class Counter < SolidObjects::Actor
10+
actor_type "activation-recovery-backoff"
11+
attribute :count, default: 0
12+
13+
class << self
14+
attr_accessor :before_commit
15+
end
16+
17+
def increment(label:)
18+
self.count += 1
19+
emit(:applied, label:)
20+
self.class.before_commit&.call(label)
21+
count
22+
end
23+
end
24+
25+
setup do
26+
SolidObjects.configuration.lease_duration = 0.2
27+
SolidObjects.configuration.lease_renewal_interval = 0.05
28+
SolidObjects.configuration.max_attempts = 1
29+
@clock = { now: SolidObjects.database_adapter.database_now }
30+
clock = @clock
31+
SolidObjects.database_adapter.define_singleton_method(:database_now) { clock.fetch(:now) }
32+
Counter.before_commit = ->(label) { clock[:now] += 0.201 if label == "slow" }
33+
@events = []
34+
SolidObjects.configuration.instrumentation = ->(event) { @events << event }
35+
@workers = []
36+
end
37+
38+
teardown do
39+
@workers.each(&:stop)
40+
SolidObjects.database_adapter.singleton_class.remove_method(:database_now)
41+
Counter.before_commit = nil
42+
end
43+
44+
test "recovers once immediately then delays without blocking other actors or reordering" do
45+
worker = new_worker
46+
first = Counter.ref("one").async.increment(label: "slow")
47+
later = Counter.ref("one").async.increment(label: "later")
48+
assert_equal 0, worker.run_once
49+
assert_equal 0, worker.run_once
50+
assert_equal 2, stored_message(first).attempt_count
51+
assert_equal 0, worker.run_once
52+
assert_in_delta 1, remaining_delay(first)
53+
other = Counter.ref("other").async.increment(label: "other")
54+
assert_equal 1, worker.run_once
55+
assert_equal 1, stored_message(other).result
56+
assert_equal 0, stored_message(later).attempt_count
57+
assert_empty stored_message(first).instance.state
58+
59+
@clock[:now] += 0.999
60+
assert_equal 0, worker.run_once
61+
@clock[:now] += 0.001
62+
assert_equal 0, worker.run_once
63+
assert_equal 3, stored_message(first).attempt_count
64+
assert_equal 0, worker.run_once
65+
assert_in_delta 2, remaining_delay(first)
66+
assert_empty SolidObjects::DeadLetter.all
67+
68+
Counter.before_commit = nil
69+
@clock[:now] += 2
70+
assert_equal 2, worker.run_once
71+
assert_equal 1, stored_message(first).result
72+
assert_equal 2, stored_message(later).result
73+
assert_equal %w[later other slow], SolidObjects::Effect.all.map { |effect| effect.arguments.fetch("label") }.sort
74+
end
75+
76+
%w[stop cleanup replacement].each do |recovery|
77+
test "preserves custom recovery waits through #{recovery}" do
78+
delay_arguments = []
79+
SolidObjects.configuration.retry_delay = ->(attempt) {
80+
delay_arguments << attempt
81+
attempt * 0.025
82+
}
83+
worker = new_worker
84+
reference = Counter.ref("one").async.increment(label: "slow")
85+
2.times { worker.run_once }
86+
worker.stop if recovery == "stop"
87+
if recovery == "cleanup"
88+
@clock[:now] += SolidObjects.configuration.process_alive_threshold + 1
89+
SolidObjects::ProcessRegistry.cleanup_dead
90+
end
91+
replacement = new_worker
92+
assert_equal 0, replacement.run_once
93+
assert_equal [ 1 ], delay_arguments
94+
assert_in_delta 0.025, remaining_delay(reference)
95+
worker.stop
96+
assert_in_delta 0.025, remaining_delay(reference)
97+
98+
@clock[:now] += 0.025
99+
replacement.run_once
100+
assert_equal 3, stored_message(reference).attempt_count
101+
assert_equal 0, replacement.run_once
102+
assert_equal [ 1, 2 ], delay_arguments
103+
assert_in_delta 0.05, remaining_delay(reference)
104+
end
105+
end
106+
107+
test "caps default recovery waits without dead lettering lease loss" do
108+
worker = new_worker
109+
reference = Counter.ref("one").async.increment(label: "slow")
110+
2.times { worker.run_once }
111+
[ 1, 2, 4, 8, 16, 32, 60, 60 ].each do |delay|
112+
worker.run_once
113+
assert_in_delta delay, remaining_delay(reference)
114+
assert_empty SolidObjects::DeadLetter.all
115+
@clock[:now] += delay
116+
worker.run_once
117+
end
118+
end
119+
120+
test "emits each lost message attempt and stops from its instrumentation callback" do
121+
worker = new_worker
122+
reference = Counter.ref("one").async.increment(label: "slow")
123+
observed = []
124+
SolidObjects.configuration.retry_delay = ->(_) { 0 }
125+
SolidObjects.configuration.instrumentation = ->(event) do
126+
next unless event.fetch("name") == "solid_objects.activation.lost"
127+
128+
observed << event
129+
worker.request_shutdown if observed.length == 3
130+
end
131+
forced_shutdown = false
132+
guard = Thread.new do
133+
sleep 2
134+
forced_shutdown = true
135+
worker.request_shutdown
136+
end
137+
begin
138+
worker.run
139+
ensure
140+
guard.kill
141+
guard.join
142+
end
143+
assert_not forced_shutdown
144+
assert_predicate worker, :stopped?
145+
assert_equal [ 1, 2, 3 ], observed.map { |event| event.fetch("attempt") }
146+
assert observed.all? { |event| event.fetch("messageId") == reference.id.to_s }
147+
observed.each { |event| assert_portable_attributes(event) }
148+
assert worker.shutdown_requested?
149+
assert_empty stored_message(reference).instance.state
150+
assert_empty SolidObjects::DeadLetter.all
151+
end
152+
153+
test "keeps a handler failure terminal with one allowed attempt" do
154+
Counter.before_commit = ->(_) { raise "handler failure" }
155+
worker = new_worker
156+
reference = Counter.ref("one").async.increment(label: "fail")
157+
worker.run_once
158+
assert_predicate stored_message(reference), :dead?
159+
assert_equal 1, stored_message(reference).attempt_count
160+
end
161+
162+
test "uses total claims for recovery after an earlier handler failure" do
163+
delay_arguments = []
164+
SolidObjects.configuration.max_attempts = 3
165+
SolidObjects.configuration.retry_delay = ->(attempt) {
166+
delay_arguments << attempt
167+
attempt * 0.025
168+
}
169+
worker = new_worker
170+
reference = Counter.ref("mixed").async.increment(label: "slow")
171+
Counter.before_commit = ->(_) { raise "handler failure" }
172+
worker.run_once
173+
assert_equal [ 1 ], delay_arguments
174+
@clock[:now] += 0.025
175+
Counter.before_commit = ->(_) { @clock[:now] += 0.201 }
176+
worker.run_once
177+
assert_equal 0, worker.run_once
178+
assert_equal [ 1, 1 ], delay_arguments
179+
assert_in_delta 0.025, remaining_delay(reference)
180+
assert_equal 2, stored_message(reference).attempt_count
181+
@clock[:now] += 0.025
182+
Counter.before_commit = ->(_) { raise "handler failed again" }
183+
worker.run_once
184+
assert_predicate stored_message(reference), :dead?
185+
assert_equal 3, stored_message(reference).attempt_count
186+
end
187+
188+
private
189+
190+
def new_worker
191+
SolidObjects::Worker.new.tap { |worker| @workers << worker }
192+
end
193+
194+
def stored_message(reference)
195+
SolidObjects::Message.find(reference.id)
196+
end
197+
198+
def remaining_delay(reference)
199+
ready = SolidObjects::ReadyMessage.find_by(message_id: reference.id)
200+
assert ready, "expected recovered claim to wait in ready membership"
201+
ready.available_at - @clock.fetch(:now)
202+
end
203+
end

0 commit comments

Comments
 (0)