fix: lock the actor instance by primary key

An enqueue used create_or_find_by!, which inserts first. For an actor
that already exists, every enqueue paid an insert that failed on the
unique key, a transaction restart, and two locking reads. Find the row
first, then lock it by its primary key. A steady-state enqueue now
issues 10 statements instead of 12, and holds the row for one statement
less.

This also removes a deadlock. A failed insert leaves a shared lock on
the identity index, and MySQL keeps that lock across a savepoint
rollback. The old code escaped this only when the insert was the first
statement of the transaction, because Active Record then restarts the
transaction instead of the savepoint. Callers that already wrote, such
as the executor and the reminder scheduler, got a real savepoint and
deadlocked when they created the same actor at the same time. Four of
eight concurrent callers failed that way before this change.

The mailbox now reads the winning row in shared mode after a duplicate
key, and locks it by primary key. MySQL needs the shared read because
it defaults to repeatable read, and a consistent read cannot see the
winning row. PostgreSQL and SQLite take no shared lock, because a share
lock there creates the same upgrade deadlock it prevents on MySQL.

Validate with bundle exec rake test on SQLite, PostgreSQL, mysql2, and
Trilogy, and with bundle exec rake standard rubocop rbs steep security.

Author: cardmagic

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## Unreleased
+
+- Find the actor instance before the insert when an enqueue starts, and lock
+  that row by its primary key. A steady-state enqueue now writes no instance
+  row and issues 10 statements instead of 12.
+- Stop the deadlock between concurrent enqueues that create the same actor
+  inside a transaction that already wrote. MySQL keeps the shared lock of a
+  failed insert across a savepoint rollback, so the mailbox reads the winning
+  row in shared mode and never asks to upgrade that lock.
+
 ## 0.15.1 - 2026-09-16
 
 - Use the existing cleanup index when finding expired actor instances. Preserve

docs/roadmap.md

@@ -6,7 +6,12 @@
 - Explicit actor registry, references, JSON state, and state migrations
 - Fluent direct synchronous RPC, configured `sync`, and durable `async`
 - Durable message history plus ready/claimed membership tables
-- Concurrent sequence allocation and actor creation
+- Concurrent sequence allocation and actor creation. An enqueue finds the
+  instance row with an unlocked read, then locks that row by its primary key.
+  A steady-state enqueue writes no instance row, and issues 10 statements
+  instead of 12. Concurrent creation causes no deadlock on SQLite, PostgreSQL,
+  or MySQL. MySQL needs a shared read after a duplicate key, because it uses
+  repeatable read. The tests count statements, and do not measure latency.
 - Activation leases, renewal, unique activation tokens, generations, and
   fenced commits
 - Bounded activation passes, idle cache, hot-actor yield, and process records

lib/solid_objects/database_adapter.rb

@@ -99,6 +99,11 @@ def claim_lock
       nil
     end
 
+    # @rbs () -> String?
+    def shared_lock
+      nil
+    end
+
     # @rbs () -> String
     def current_time_expression
       "CURRENT_TIMESTAMP"
@@ -163,6 +168,11 @@ def lock_candidates(relation)
       claim_lock ? relation.lock(claim_lock) : relation
     end
 
+    # @rbs (ActiveRecord::Relation[untyped]) -> ActiveRecord::Relation[untyped]
+    def share_locked(relation)
+      shared_lock ? relation.lock(shared_lock) : relation
+    end
+
     private
 
     attr_reader :connection_pool, :fixed_connection

lib/solid_objects/database_adapters/mysql.rb

@@ -15,6 +15,11 @@ def claim_lock
         "FOR UPDATE SKIP LOCKED"
       end
 
+      # @rbs () -> String
+      def shared_lock
+        "FOR SHARE"
+      end
+
       # A non-transactional engine would silently break fenced commits, so the
       # storage engine is verified rather than assumed.
       # @rbs () -> Array[String]

lib/solid_objects/mailbox.rb

@@ -52,7 +52,6 @@ def enqueue_in_transaction(
         max_bytes: SolidObjects.configuration.max_payload_bytes
       )
       instance = find_or_create_instance(reference, actor_class)
-      instance.lock!
 
       existing = find_idempotent_message(instance, idempotency_key)
       if existing
@@ -108,13 +107,46 @@ def with_instance_retry
 
     # @rbs (Reference, Class) -> Instance
     def find_or_create_instance(reference, actor_class)
-      Instance.create_or_find_by!(
-        actor_type: reference.actor_type,
-        actor_id: reference.actor_id
-      ) do |instance|
-        instance.state = {}
-        instance.state_version = actor_class.state_version
+      identifier = instance_identifier(reference)
+      return lock_instance!(identifier) if identifier
+
+      create_locked_instance(reference, actor_class)
+    end
+
+    # @rbs (Reference) -> Integer?
+    def instance_identifier(reference)
+      Instance
+        .where(actor_type: reference.actor_type, actor_id: reference.actor_id)
+        .pick(:id)
+    end
+
+    # @rbs (Reference, Class) -> Instance
+    def create_locked_instance(reference, actor_class)
+      Instance.transaction(requires_new: true) do
+        Instance.create!(
+          actor_type: reference.actor_type,
+          actor_id: reference.actor_id,
+          state: {},
+          state_version: actor_class.state_version
+        )
       end
+    rescue ActiveRecord::RecordNotUnique
+      lock_instance!(committed_instance_identifier(reference))
+    end
+
+    # @rbs (Reference) -> Integer?
+    def committed_instance_identifier(reference)
+      database_adapter.share_locked(
+        Instance.where(actor_type: reference.actor_type, actor_id: reference.actor_id)
+      ).pick(:id)
+    end
+
+    # @rbs (Integer?) -> Instance
+    def lock_instance!(identifier)
+      instance = identifier && Instance.lock.find_by(id: identifier)
+      return instance if instance
+
+      raise ActiveRecord::RecordNotFound, "actor instance disappeared while enqueueing"
     end
 
     # @rbs (Instance, String?) -> Message?

sig/generated/lib/solid_objects/database_adapter.rbs

@@ -49,6 +49,9 @@ module SolidObjects
     # @rbs () -> String?
     def claim_lock: () -> String?
 
+    # @rbs () -> String?
+    def shared_lock: () -> String?
+
     # @rbs () -> String
     def current_time_expression: () -> String
 
@@ -70,6 +73,9 @@ module SolidObjects
     # @rbs (ActiveRecord::Relation[untyped]) -> ActiveRecord::Relation[untyped]
     def lock_candidates: (ActiveRecord::Relation[untyped]) -> ActiveRecord::Relation[untyped]
 
+    # @rbs (ActiveRecord::Relation[untyped]) -> ActiveRecord::Relation[untyped]
+    def share_locked: (ActiveRecord::Relation[untyped]) -> ActiveRecord::Relation[untyped]
+
     private
 
     attr_reader connection_pool: untyped

sig/generated/lib/solid_objects/database_adapters/mysql.rbs

@@ -11,6 +11,9 @@ module SolidObjects
       # @rbs () -> String
       def claim_lock: () -> String
 
+      # @rbs () -> String
+      def shared_lock: () -> String
+
       # A non-transactional engine would silently break fenced commits, so the
       # storage engine is verified rather than assumed.
       # @rbs () -> Array[String]

sig/generated/lib/solid_objects/mailbox.rbs

@@ -28,6 +28,18 @@ module SolidObjects
     # @rbs (Reference, Class) -> Instance
     def find_or_create_instance: (Reference, Class) -> Instance
 
+    # @rbs (Reference) -> Integer?
+    def instance_identifier: (Reference) -> Integer?
+
+    # @rbs (Reference, Class) -> Instance
+    def create_locked_instance: (Reference, Class) -> Instance
+
+    # @rbs (Reference) -> Integer?
+    def committed_instance_identifier: (Reference) -> Integer?
+
+    # @rbs (Integer?) -> Instance
+    def lock_instance!: (Integer?) -> Instance
+
     # @rbs (Instance, String?) -> Message?
     def find_idempotent_message: (Instance, String?) -> Message?
 

test/integration/enqueue_statement_count_test.rb

@@ -0,0 +1,78 @@
+# frozen_string_literal: true
+
+require "database_test_helper"
+
+class EnqueueStatementCountTest < ActiveSupport::TestCase
+  class CartActor < SolidObjects::Actor
+    actor_type "enqueue-statement-count-cart"
+
+    attribute :items, default: -> { [] }
+
+    def add(product_id:)
+      self.items += [ product_id ]
+    end
+  end
+
+  STEADY_STATE_STATEMENT_COUNT = 10
+
+  setup { CartActor.ensure_registered! }
+
+  test "a steady-state enqueue never inserts the instance row" do
+    reference = CartActor.ref("alice")
+    reference.async.add(product_id: "shirt")
+
+    statements = capture_statements { reference.async.add(product_id: "pants") }
+
+    assert_empty instance_statements(statements).grep(/\AINSERT/i)
+  end
+
+  test "a steady-state enqueue touches the instance row three times" do
+    reference = CartActor.ref("alice")
+    reference.async.add(product_id: "shirt")
+
+    statements = instance_statements(capture_statements { reference.async.add(product_id: "pants") })
+
+    assert_equal 2, statements.grep(/\ASELECT/i).length, statements.inspect
+    assert_equal 1, statements.grep(/\AUPDATE/i).length, statements.inspect
+    assert_equal 3, statements.length, statements.inspect
+  end
+
+  test "a steady-state enqueue opens one transaction and never restarts it" do
+    reference = CartActor.ref("alice")
+    reference.async.add(product_id: "shirt")
+
+    statements = capture_statements { reference.async.add(product_id: "pants") }
+    control = statements.grep(/\A(?:BEGIN|COMMIT|ROLLBACK|SAVEPOINT|RELEASE)/i)
+
+    assert_empty control.grep(/ROLLBACK|SAVEPOINT/i), statements.inspect
+    assert_equal 2, control.length, statements.inspect
+  end
+
+  test "a steady-state enqueue issues a fixed number of statements" do
+    reference = CartActor.ref("alice")
+    reference.async.add(product_id: "shirt")
+
+    statements = capture_statements { reference.async.add(product_id: "pants") }
+
+    assert_equal STEADY_STATE_STATEMENT_COUNT, statements.length, statements.inspect
+  end
+
+  private
+
+  def instance_statements(statements)
+    statements.grep(/solid_objects_instances/)
+  end
+
+  def capture_statements
+    statements = []
+    subscriber = lambda do |*arguments|
+      payload = arguments.last
+      next if payload[:name] == "SCHEMA"
+      next if payload[:cached]
+
+      statements << payload.fetch(:sql).to_s.strip
+    end
+    ActiveSupport::Notifications.subscribed(subscriber, "sql.active_record") { yield }
+    statements
+  end
+end

test/integration/enqueue_test.rb

@@ -58,6 +58,81 @@ def add(product_id:)
 
     assert_empty errors.size.times.map { errors.pop }
     assert_equal (1..8).to_a, results.size.times.map { results.pop }.sort
+    assert_equal 1, SolidObjects::Instance.where(actor_type: "enqueue-carts", actor_id: "alice").count
+  end
+
+  test "allocates unique sequences under concurrent enqueue to an existing actor" do
+    reference = CartActor.ref("alice")
+    reference.async.add(product_id: "first")
+    start = Queue.new
+    results = Queue.new
+    errors = Queue.new
+
+    threads = 8.times.map do |index|
+      Thread.new do
+        SolidObjects::Record.connection_pool.with_connection do
+          start.pop
+          results << reference.async.add(product_id: "product-#{index}").sequence
+        rescue => error
+          errors << error
+        end
+      end
+    end
+
+    threads.length.times { start << true }
+    threads.each(&:join)
+
+    assert_empty errors.size.times.map { errors.pop }
+    assert_equal (2..9).to_a, results.size.times.map { results.pop }.sort
+    assert_equal 1, SolidObjects::Instance.where(actor_type: "enqueue-carts", actor_id: "alice").count
+  end
+
+  test "creates the instance once when concurrent callers already hold a dirty transaction" do
+    CartActor.ensure_registered!
+    reference = SolidObjects::Reference.new(actor_type: "enqueue-carts", actor_id: "alice")
+    mailbox = SolidObjects::Mailbox.new
+    start = Queue.new
+    sequences = Queue.new
+    errors = Queue.new
+
+    threads = 8.times.map do |index|
+      Thread.new do
+        SolidObjects::Record.connection_pool.with_connection do
+          start.pop
+          SolidObjects.database_adapter.transaction do
+            SolidObjectsTestDomainRecord.create!(name: "dirty-#{index}")
+            sequences << mailbox.enqueue_in_transaction(
+              reference:,
+              operation: :add,
+              arguments: { product_id: "product-#{index}" },
+              delivery_mode: "async",
+              idempotency_key: nil
+            ).sequence
+          end
+        rescue => error
+          errors << error
+        end
+      end
+    end
+
+    threads.length.times { start << true }
+    threads.each { |thread| thread.join(30) }
+
+    assert_empty errors.size.times.map { errors.pop }
+    assert_equal (1..8).to_a, sequences.size.times.map { sequences.pop }.sort
+    assert_equal 1, SolidObjects::Instance.where(actor_type: "enqueue-carts", actor_id: "alice").count
+  end
+
+  test "gives up when the instance keeps disappearing between the lookup and the insert" do
+    SolidObjects::Instance.singleton_class.define_method(:create!) do |*, **|
+      raise ActiveRecord::RecordNotUnique, "simulated create race"
+    end
+
+    assert_raises(SolidObjects::ActorDestroyed) do
+      CartActor.ref("ghost").async.add(product_id: "shirt")
+    end
+  ensure
+    SolidObjects::Instance.singleton_class.send(:remove_method, :create!)
   end
 
   test "deduplicates the same idempotent enqueue" do