From bd6ed0770ae126f2073f45f1cb0dc9557520b525 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 21 Sep 2026 17:41:09 -0700 Subject: [PATCH] fix: read message and snapshot state uncached MessageReference#status, MessageReference#result, and ActorSnapshot read through the Active Record query cache. A caller that polls holds one cache for the whole poll, and the worker that finishes the message is a different process, so its write cannot clear that cache. The reader saw its first answer forever. This reaches any poll inside one executor block: a controller action, an Active Job, or a rails runner script. A real application reproduced it, reporting "ready" twenty times over six seconds while the row had completed. Read those three uncached, as SynchronousInvocation and SyncDiagnostics already do for the same reason. The suite never caught it because the query cache is off in tests. The new test opens one, then writes from a second connection so the write cannot clear it, which is what a worker process does. --- CHANGELOG.md | 9 +++ lib/solid_objects/actor_snapshot.rb | 10 ++-- lib/solid_objects/message_reference.rb | 20 ++++--- test/integration/query_cache_reads_test.rb | 69 ++++++++++++++++++++++ 4 files changed, 95 insertions(+), 13 deletions(-) create mode 100644 test/integration/query_cache_reads_test.rb 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