Skip to content

Commit cca52e2

Browse files
committed
fix: renew leases and preserve claimed fences
Apply the parity repair for independently reported lease defects. Renew between drain turns when due, without relying on the renewal thread, and keep the activation fence while a message claim remains. Before the fix, regressions reported an expired lease during draining and a cleared expiry with an outstanding claim. Confirm renewal, recovery with max_attempts: 1, and replacement-owner fencing on SQLite, PostgreSQL, mysql2, and Trilogy. Preserve existing cross-pass renewal and NULL-expiry claim recovery behavior.
1 parent 8c6f32c commit cca52e2

5 files changed

Lines changed: 168 additions & 1 deletion

File tree

‎CHANGELOG.md‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,14 @@
11
# Changelog
22

3+
## Unreleased
4+
5+
- Renew activations between turns in a drain pass when renewal is due,
6+
including when the renewal thread has not run.
7+
- Keep the activation fence and expiry while an outstanding message claim
8+
remains. The next worker pass recovers an expired turn in mailbox order,
9+
including with `max_attempts: 1`. Reclaiming increments `attempt_count`;
10+
lease loss alone does not create a dead letter.
11+
312
## 0.17.3 - 2026-10-09
413

514
- Add four problem guides in `docs/guides/`: preventing race conditions in

‎docs/roadmap.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,8 @@
2323
instead of 12. Concurrent creation causes no deadlock on SQLite, PostgreSQL,
2424
or MySQL. MySQL needs a shared read after a duplicate key, because it uses
2525
repeatable read. The tests count statements, and do not measure latency.
26-
- Activation leases, renewal, unique activation tokens, generations, and
26+
- Activation leases, renewal between turns as well as in the renewal thread,
27+
unique activation tokens, generations, and
2728
fenced commits
2829
- Bounded activation passes, idle cache, hot-actor yield, and process records
2930
- At-least-once retries, terminal domain rejection, strict poison ordering,

‎lib/solid_objects/activation.rb‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -139,6 +139,7 @@ def drain_messages(message_id: nil, deadline: nil)
139139
break
140140
end
141141

142+
renew_lease if processed_count.positive? && lease_renewal_due?
142143
message = claim_next_message
143144
break unless message
144145

‎lib/solid_objects/lease.rb‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -123,6 +123,7 @@ def release
123123
instance = Instance.lock.find_by(id: instance_id)
124124
next false unless instance
125125
next false unless owned_generation?(instance)
126+
next false if ClaimedMessage.where(instance_id:).exists?
126127

127128
instance.update!(
128129
activation_owner_id: nil,
Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
# rbs_inline: enabled
2+
3+
require "database_test_helper"
4+
5+
class ActivationLeaseRecoveryTest < ActiveSupport::TestCase
6+
class Counter < SolidObjects::Actor
7+
actor_type "activation-lease-recovery"
8+
attribute :count, default: 0
9+
10+
class << self
11+
attr_accessor :before_commit
12+
end
13+
14+
def increment
15+
self.count += 1
16+
self.class.before_commit&.call
17+
count
18+
end
19+
end
20+
21+
setup do
22+
SolidObjects.configuration.lease_duration = 0.2
23+
SolidObjects.configuration.lease_renewal_interval = 0.05
24+
SolidObjects.configuration.max_attempts = 1
25+
@clock = { now: SolidObjects.database_adapter.database_now }
26+
clock = @clock
27+
SolidObjects.database_adapter.define_singleton_method(:database_now) { clock.fetch(:now) }
28+
Counter.before_commit = nil
29+
@events = []
30+
@subscription = ActiveSupport::Notifications.subscribe(/solid_objects\.(message|activation|recovery)\./) do |event|
31+
@events << event.name
32+
end
33+
end
34+
35+
teardown do
36+
ActiveSupport::Notifications.unsubscribe(@subscription)
37+
SolidObjects.database_adapter.singleton_class.remove_method(:database_now)
38+
Counter.before_commit = nil
39+
end
40+
41+
test "keeps renewing across short worker passes for five lease durations" do
42+
worker = SolidObjects::Worker.new
43+
Counter.before_commit = -> { @clock[:now] += 0.03 }
44+
messages = []
45+
first_expiration = nil
46+
47+
34.times do
48+
messages << Counter.ref("continuous").async.increment
49+
assert_equal 1, worker.run_once
50+
first_expiration ||= instance.activation_expires_at
51+
end
52+
53+
assert_operator instance.activation_expires_at, :>, first_expiration + 0.8
54+
assert_equal 1, instance.activation_generation
55+
messages.each_with_index do |reference, index|
56+
message = SolidObjects::Message.find(reference.id)
57+
assert_predicate message, :completed?
58+
assert_equal index + 1, message.result
59+
assert_equal 1, message.attempt_count
60+
end
61+
assert_equal 34, @events.count("solid_objects.message.completed")
62+
ensure
63+
worker&.stop
64+
end
65+
66+
test "renews between short turns while draining without a renewal thread" do
67+
messages = 34.times.map { Counter.ref("draining").async.increment }
68+
registry = SolidObjects::ProcessRegistry.new
69+
process = registry.register
70+
activation = SolidObjects::ActivationManager.new(owner_id: process.id).claim_next
71+
first_expiration = instance.activation_expires_at
72+
Counter.before_commit = -> { @clock[:now] += 0.03 }
73+
74+
assert_equal 34, activation.drain
75+
assert_operator instance.activation_expires_at, :>, first_expiration + 0.8
76+
assert_equal 1, instance.activation_generation
77+
assert messages.all? { |reference| SolidObjects::Message.find(reference.id).completed? }
78+
ensure
79+
activation&.deactivate
80+
registry&.stop
81+
end
82+
83+
test "keeps the expired lease fence while a message claim remains" do
84+
worker = SolidObjects::Worker.new
85+
first = Counter.ref("expired").async.increment
86+
second = Counter.ref("expired").async.increment
87+
Counter.before_commit = -> { @clock[:now] += 0.22 }
88+
89+
assert_equal 0, worker.run_once
90+
assert_empty instance.state
91+
assert SolidObjects::ClaimedMessage.exists?(message_id: first.id)
92+
assert_not_nil instance.activation_expires_at
93+
assert_operator instance.activation_expires_at, :<=, @clock.fetch(:now)
94+
Counter.before_commit = nil
95+
96+
assert_equal 2, worker.run_once
97+
assert_equal 1, SolidObjects::Message.find(first.id).result
98+
assert_equal 2, SolidObjects::Message.find(first.id).attempt_count
99+
assert_equal 2, SolidObjects::Message.find(second.id).result
100+
assert_equal 1, SolidObjects::Message.find(second.id).attempt_count
101+
assert_empty SolidObjects::DeadLetter.all
102+
assert_equal 2, @events.count("solid_objects.message.completed")
103+
ensure
104+
worker&.stop
105+
end
106+
107+
test "recovers on the same worker and preserves mailbox order with one allowed attempt" do
108+
worker = SolidObjects::Worker.new
109+
first = Counter.ref("recoverable").async.increment
110+
second = Counter.ref("recoverable").async.increment
111+
Counter.before_commit = -> { @clock[:now] += 0.22 }
112+
113+
assert_equal 0, worker.run_once
114+
assert_empty instance.state
115+
Counter.before_commit = nil
116+
117+
assert_equal 2, worker.run_once
118+
assert_equal 1, SolidObjects::Message.find(first.id).result
119+
assert_equal 2, SolidObjects::Message.find(second.id).result
120+
assert_equal 2, SolidObjects::Message.find(first.id).attempt_count
121+
assert_empty SolidObjects::DeadLetter.all
122+
ensure
123+
worker&.stop
124+
end
125+
126+
test "a stale worker cannot release a second worker's activation or repeat its commits" do
127+
first_worker = SolidObjects::Worker.new
128+
second_worker = SolidObjects::Worker.new
129+
first = Counter.ref("takeover").async.increment
130+
second = Counter.ref("takeover").async.increment
131+
Counter.before_commit = -> do
132+
Counter.before_commit = nil
133+
@clock[:now] += 0.22
134+
assert_equal 2, Thread.new { second_worker.run_once }.value
135+
end
136+
137+
assert_equal 0, first_worker.run_once
138+
assert_equal 1, SolidObjects::Message.find(first.id).result
139+
assert_equal 2, SolidObjects::Message.find(second.id).result
140+
assert_equal 2, SolidObjects::Message.find(first.id).attempt_count
141+
assert_equal 2, @events.count("solid_objects.message.completed")
142+
third = Counter.ref("takeover").async.increment
143+
assert_equal 1, second_worker.run_once
144+
assert_equal 3, SolidObjects::Message.find(third.id).result
145+
ensure
146+
first_worker&.stop
147+
second_worker&.stop
148+
end
149+
150+
private
151+
152+
def instance
153+
SolidObjects::Instance.find_by!(actor_type: Counter.actor_type)
154+
end
155+
end

0 commit comments

Comments
 (0)