Skip to content

Commit 4ae9c4a

Browse files
committed
fix: report lease loss in all turn outcomes
Catch lost fences around the full executor path, including failure and rejection persistence. Preserve recovery and report each lost attempt so instrumentation can stop a repeatedly interrupted worker. Before the fix, two new recovery tests missed activation.lost and two callback shutdown cases required their outer time limit. All four pass now. Full rake passes: 940 tests, 3940 assertions, 28 conditional skips.
1 parent 2b41a93 commit 4ae9c4a

4 files changed

Lines changed: 82 additions & 38 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,8 @@
1111
can proceed. Lease loss alone still does not dead-letter a message; handler
1212
failure attempt limits, setting names, defaults, and schema stay unchanged.
1313
- Emit `activation.lost` with the message ID and total attempt count when a
14-
turn loses its fence, matching JavaScript instrumentation. Shutdown requested
14+
turn loses its fence, including while saving a handler failure or rejection,
15+
matching JavaScript instrumentation. Shutdown requested
1516
by that callback stops the worker before another pass. Skip a cached actor
1617
whose first message is delayed so later ready messages cannot starve peers.
1718

‎docs/roadmap.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,8 @@
2828
fenced commits. Interrupted first attempts recover immediately; repeated
2929
interruptions use durable retry delays from total claim attempts minus one,
3030
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;
31+
`activation.lost` with the message ID and attempt, including losses while
32+
saving a handler failure or rejection. There is no hard retry cap;
3233
applications needing one must impose a stopping policy.
3334
- Bounded activation passes, idle cache, hot-actor yield, and process records
3435
- At-least-once retries, terminal domain rejection, strict poison ordering,

‎lib/solid_objects/executor.rb‎

Lines changed: 13 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,18 @@ def initialize(activation:, message:)
1616

1717
# @rbs () -> bool
1818
def call
19+
execute
20+
rescue LostActivation
21+
SolidObjects.instrument(:"activation.lost", **instrumentation_payload, duration_milliseconds: elapsed_milliseconds)
22+
raise
23+
end
24+
25+
private
26+
27+
attr_reader :activation, :message
28+
29+
# @rbs () -> bool
30+
def execute
1931
state_before = actor.state.to_h
2032
observables_before = actor.observable_values
2133
message_context = MessageContext.new(
@@ -39,10 +51,7 @@ def call
3951
state_changed: state_after.value != state_before
4052
)
4153
true
42-
rescue CommittedTransactionError
43-
raise
44-
rescue LostActivation
45-
SolidObjects.instrument(:"activation.lost", **instrumentation_payload, duration_milliseconds: elapsed_milliseconds)
54+
rescue CommittedTransactionError, LostActivation
4655
raise
4756
rescue Rejected => rejection
4857
activation.restore_state(state_before) if state_before
@@ -56,10 +65,6 @@ def call
5665
false
5766
end
5867

59-
private
60-
61-
attr_reader :activation, :message
62-
6368
# @rbs () -> Actor
6469
def actor
6570
activation.actor

‎test/integration/activation_recovery_backoff_test.rb‎

Lines changed: 65 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -117,37 +117,74 @@ def increment(label:)
117117
end
118118
end
119119

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"
120+
%w[completion failure rejection].each do |outcome|
121+
test "reports each lost #{outcome} and stops from its instrumentation callback" do
122+
worker = new_worker
123+
reference = Counter.ref("one").async.increment(label: "slow")
124+
Counter.before_commit = ->(_) do
125+
@clock[:now] += 0.201
126+
raise SolidObjects::Rejected.new(code: "declined", message: "declined") if outcome == "rejection"
127+
raise "handler failed" if outcome == "failure"
128+
end
129+
observed = []
130+
SolidObjects.configuration.retry_delay = ->(_) { 0 }
131+
SolidObjects.configuration.instrumentation = ->(event) do
132+
next unless event.fetch("name") == "solid_objects.activation.lost"
127133

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
134+
observed << event
135+
worker.request_shutdown if observed.length == 3
136+
end
137+
forced_shutdown = false
138+
guard = Thread.new do
139+
sleep 2
140+
forced_shutdown = true
141+
worker.request_shutdown
142+
end
143+
begin
144+
worker.run
145+
ensure
146+
guard.kill
147+
guard.join
148+
end
149+
assert_not forced_shutdown
150+
assert_predicate worker, :stopped?
151+
assert_equal [ 1, 2, 3 ], observed.map { |event| event.fetch("attempt") }
152+
assert observed.all? { |event| event.fetch("messageId") == reference.id.to_s }
153+
observed.each { |event| assert_portable_attributes(event) }
154+
assert worker.shutdown_requested?
155+
assert_empty stored_message(reference).instance.state
156+
assert_empty SolidObjects::DeadLetter.all
136157
end
137-
begin
138-
worker.run
139-
ensure
140-
guard.kill
141-
guard.join
158+
end
159+
160+
%w[failure rejection].each do |outcome|
161+
test "reports lease loss while saving a handler #{outcome} and recovers" do
162+
worker = new_worker
163+
reference = Counter.ref("one").async.increment(label: "slow")
164+
Counter.before_commit = ->(_) do
165+
@clock[:now] += 0.201
166+
raise SolidObjects::Rejected.new(code: "declined", message: "declined") if outcome == "rejection"
167+
168+
raise "handler failed"
169+
end
170+
assert_equal 0, worker.run_once
171+
losses = @events.select { |event| event.fetch("name") == "solid_objects.activation.lost" }
172+
assert_equal [ 1 ], losses.map { |event| event.fetch("attempt") }
173+
assert_equal [ reference.id.to_s ], losses.map { |event| event.fetch("messageId") }
174+
losses.each { |event| assert_portable_attributes(event) }
175+
assert_empty stored_message(reference).instance.state
176+
assert_nil stored_message(reference).error
177+
assert_nil stored_message(reference).rejection
178+
assert_empty SolidObjects::DeadLetter.all
179+
assert_empty SolidObjects::Effect.all
180+
assert_not @events.any? { |event| %w[solid_objects.message.failed solid_objects.message.rejected].include?(event.fetch("name")) }
181+
182+
Counter.before_commit = nil
183+
assert_equal 1, worker.run_once
184+
assert_equal 1, stored_message(reference).result
185+
assert_equal 2, stored_message(reference).attempt_count
186+
assert_equal 1, SolidObjects::Effect.count
142187
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
151188
end
152189

153190
test "keeps a handler failure terminal with one allowed attempt" do

0 commit comments

Comments
 (0)