diff --git a/CHANGELOG.md b/CHANGELOG.md index c87adbe..d62f12d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## Unreleased + +- Read the durable row rather than the query cache in `MessageReference#status`, + `MessageReference#result`, and an actor snapshot. A caller that polls holds one + query cache for the whole poll, and the worker that finishes the message is + another process, so its write cannot clear that cache. A poll inside a request, + a job, or `rails runner` reported the first answer forever. The synchronous + wait already read uncached. + ## 0.15.2 - 2026-09-21 - Find the actor instance before the insert when an enqueue starts, and lock diff --git a/lib/solid_objects/actor_snapshot.rb b/lib/solid_objects/actor_snapshot.rb index 43fa307..2b277d0 100644 --- a/lib/solid_objects/actor_snapshot.rb +++ b/lib/solid_objects/actor_snapshot.rb @@ -16,10 +16,12 @@ class ActorSnapshot def initialize(reference) @reference = reference @actor_class = SolidObjects.registry.fetch(reference.actor_type) - @instance = Instance.find_by( - actor_type: reference.actor_type, - actor_id: reference.actor_id - ) + @instance = Instance.uncached do + Instance.find_by( + actor_type: reference.actor_type, + actor_id: reference.actor_id + ) + end @instance_id = @instance&.id || 0 @revision = @instance&.state_revision || 0 @actor = build_actor diff --git a/lib/solid_objects/message_reference.rb b/lib/solid_objects/message_reference.rb index 7f960fc..d015cb5 100644 --- a/lib/solid_objects/message_reference.rb +++ b/lib/solid_objects/message_reference.rb @@ -35,19 +35,21 @@ def initialize(id:, request_id:, actor_type:, actor_id:, sequence:) # @rbs () -> String def status - message = Message.find(id) - return "rejected" if message.rejected? - return "completed" if message.completed? - return "dead" if message.dead? - return "claimed" if message.claimed? - return "ready" if message.ready? - - "unknown" + Message.uncached do + message = Message.find(id) + return "rejected" if message.rejected? + return "completed" if message.completed? + return "dead" if message.dead? + return "claimed" if message.claimed? + return "ready" if message.ready? + + "unknown" + end end # @rbs () -> untyped def result - Message.find(id).result + Message.uncached { Message.find(id).result } end # @rbs (?timeout: Numeric, ?authorization_context: untyped) -> untyped diff --git a/test/integration/query_cache_reads_test.rb b/test/integration/query_cache_reads_test.rb new file mode 100644 index 0000000..e66d27d --- /dev/null +++ b/test/integration/query_cache_reads_test.rb @@ -0,0 +1,69 @@ +# frozen_string_literal: true + +require "database_test_helper" + +class QueryCacheReadsTest < ActiveSupport::TestCase + class CounterActor < SolidObjects::Actor + actor_type "query-cache-counter" + + attribute :count, default: 0 + + def increment + self.count += 1 + end + end + + setup { CounterActor.ensure_registered! } + + def from_another_connection + Thread.new do + SolidObjects::Record.connection_pool.with_connection { yield } + end.join + end + + test "message status refreshes while a query cache is open" do + message = CounterActor.ref("alice").async.increment + + ActiveRecord::Base.cache do + assert_equal "ready", message.status + + from_another_connection do + SolidObjects::ReadyMessage.where(message_id: message.id).delete_all + SolidObjects::Message.find(message.id).update!(completed_at: Time.current) + end + + assert_equal "completed", message.status + end + end + + test "message result refreshes while a query cache is open" do + message = CounterActor.ref("alice").async.increment + + ActiveRecord::Base.cache do + assert_nil message.result + + from_another_connection do + SolidObjects::Message.find(message.id).update!(result: { "value" => 7 }) + end + + assert_equal({ "value" => 7 }, message.result) + end + end + + test "an actor snapshot refreshes while a query cache is open" do + reference = CounterActor.ref("alice") + reference.async.increment + SolidObjects::Worker.new.run_until_idle + + ActiveRecord::Base.cache do + assert_equal 1, reference.snapshot.count + + from_another_connection do + instance = SolidObjects::Instance.find_by!(actor_type: "query-cache-counter", actor_id: "alice") + instance.update!(state: instance.state.merge("count" => 9)) + end + + assert_equal 9, reference.snapshot.count + end + end +end