|
| 1 | +# rbs_inline: enabled |
| 2 | + |
| 3 | +require "database_test_helper" |
| 4 | +require "timeout" |
| 5 | + |
| 6 | +class CallerLifecycleTest < ActiveSupport::TestCase |
| 7 | + class Counter < SolidObjects::Actor |
| 8 | + actor_type "caller-lifecycle-counter" |
| 9 | + attribute :count, default: 0 |
| 10 | + |
| 11 | + class << self |
| 12 | + attr_accessor :started, :release |
| 13 | + end |
| 14 | + |
| 15 | + # @rbs (label: String) -> Integer |
| 16 | + def increment(label:) |
| 17 | + self.count += 1 |
| 18 | + emit(:caller_effect, label:) |
| 19 | + if self.class.started |
| 20 | + self.class.started << label |
| 21 | + self.class.release.pop |
| 22 | + end |
| 23 | + count |
| 24 | + end |
| 25 | + end |
| 26 | + |
| 27 | + setup do |
| 28 | + SolidObjects.configuration.max_attempts = 1 |
| 29 | + end |
| 30 | + |
| 31 | + teardown do |
| 32 | + Counter.started = nil |
| 33 | + Counter.release = nil |
| 34 | + end |
| 35 | + |
| 36 | + test "concurrent callers preserve staged effects and ordered state" do |
| 37 | + Counter.started = Queue.new |
| 38 | + Counter.release = Queue.new |
| 39 | + enqueued = Queue.new |
| 40 | + subscription = ActiveSupport::Notifications.subscribe("solid_objects.message.enqueued") { enqueued << true } |
| 41 | + reference = Counter.ref("concurrent") |
| 42 | + |
| 43 | + first = Thread.new { reference.increment(label: "first") } |
| 44 | + assert_equal "first", Timeout.timeout(10) { Counter.started.pop } |
| 45 | + second = Thread.new { reference.increment(label: "second") } |
| 46 | + 2.times { Timeout.timeout(10) { enqueued.pop } } |
| 47 | + Counter.release << true |
| 48 | + assert_equal 1, Timeout.timeout(10) { first.value } |
| 49 | + assert_equal "second", Timeout.timeout(10) { Counter.started.pop } |
| 50 | + Counter.release << true |
| 51 | + assert_equal 2, Timeout.timeout(10) { second.value } |
| 52 | + |
| 53 | + instance = SolidObjects::Instance.find_by!(actor_type: Counter.actor_type, actor_id: "concurrent") |
| 54 | + assert_equal({ "count" => 2 }, instance.state) |
| 55 | + assert_equal [ "first", "second" ], SolidObjects::Effect.all.map { |effect| effect.arguments.fetch("label") }.sort |
| 56 | + assert_equal [ 1, 1 ], SolidObjects::Message.order(:sequence).pluck(:attempt_count) |
| 57 | + assert_empty SolidObjects::ClaimedMessage.all |
| 58 | + assert_empty SolidObjects::DeadLetter.all |
| 59 | + ensure |
| 60 | + Counter.release << true if first&.alive? |
| 61 | + Counter.release << true if second&.alive? |
| 62 | + first&.join(10) |
| 63 | + second&.join(10) |
| 64 | + ActiveSupport::Notifications.unsubscribe(subscription) if subscription |
| 65 | + end |
| 66 | + |
| 67 | + test "caller re-registers after stale cleanup and process retention" do |
| 68 | + reference = Counter.ref("retained-process") |
| 69 | + assert_equal 1, reference.increment(label: "before") |
| 70 | + original_process = SolidObjects.caller_process.process_registry.process_record |
| 71 | + original_process.update!(last_heartbeat_at: 2.minutes.ago) |
| 72 | + assert_equal 1, SolidObjects::ProcessRegistry.cleanup_dead |
| 73 | + assert_equal "stopped", original_process.reload.shutdown_state |
| 74 | + |
| 75 | + assert_equal 2, reference.increment(label: "after-cleanup") |
| 76 | + stopped_process = SolidObjects.caller_process.process_registry.process_record |
| 77 | + refute_equal original_process.id, stopped_process.id |
| 78 | + assert_equal "running", stopped_process.shutdown_state |
| 79 | + stopped_process.update!(last_heartbeat_at: 2.minutes.ago) |
| 80 | + assert_equal 1, SolidObjects::ProcessRegistry.cleanup_dead |
| 81 | + SolidObjects::Process.where(id: [ original_process.id, stopped_process.id ]).update_all(stopped_at: 8.days.ago) |
| 82 | + assert_equal 2, SolidObjects::ProcessPruner.new.prune |
| 83 | + refute SolidObjects::Process.exists?(stopped_process.id) |
| 84 | + |
| 85 | + assert_equal 3, reference.increment(label: "after-retention") |
| 86 | + replacement = SolidObjects.caller_process.process_registry.process_record |
| 87 | + refute_equal stopped_process.id, replacement.id |
| 88 | + assert_equal "running", replacement.shutdown_state |
| 89 | + assert_equal [ 1, 1, 1 ], SolidObjects::Message.order(:sequence).pluck(:attempt_count) |
| 90 | + assert_empty SolidObjects::DeadLetter.all |
| 91 | + end |
| 92 | +end |
0 commit comments