fix: lock the actor instance by primary key

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## 0.15.2 - 2026-09-21
+
+- 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

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    solid_objects (0.15.1)
+    solid_objects (0.15.2)
       actioncable (>= 7.1)
       actionpack (>= 7.1)
       actionview (>= 7.1)
@@ -384,7 +384,7 @@ CHECKSUMS
   rubocop-rails-omakase (1.1.0) sha256=2af73ac8ee5852de2919abbd2618af9c15c19b512c4cfc1f9a5d3b6ef009109d
   ruby-progressbar (1.13.0) sha256=80fc9c47a9b640d6834e0dc7b3c94c9df37f08cb072b7761e4a71e22cff29b33
   securerandom (0.4.1) sha256=cc5193d414a4341b6e225f0cb4446aceca8e50d5e1888743fac16987638ea0b1
-  solid_objects (0.15.1)
+  solid_objects (0.15.2)
   sqlite3 (2.9.5-aarch64-linux-gnu) sha256=78075b6337d3d182c6d2b4691049ed45cd220826160c9ea18946bf6a1de200dc
   sqlite3 (2.9.5-aarch64-linux-musl) sha256=18c801185deb4adc01ddb281e8f672a39e3d1729979ca91e39439cd3eac0402d
   sqlite3 (2.9.5-arm-linux-gnu) sha256=1bdfca0c7d63998c60b0f4a8e3c8df2d33800ccc4abd2d612eddbbbc92a4c48b

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,23 @@ 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
+      identity = { actor_type: reference.actor_type, actor_id: reference.actor_id }
+      identifier = Instance.where(identity).pick(:id)
+      return lock_instance!(identifier) if identifier
+
+      Instance.transaction(requires_new: true) do
+        Instance.create!(**identity, state: {}, state_version: actor_class.state_version)
       end
+    rescue ActiveRecord::RecordNotUnique
+      lock_instance!(database_adapter.share_locked(Instance.where(identity)).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?

lib/solid_objects/version.rb

@@ -1,5 +1,5 @@
 # rbs_inline: enabled
 
 module SolidObjects
-  VERSION = "0.15.1"
+  VERSION = "0.15.2"
 end

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,9 @@ module SolidObjects
     # @rbs (Reference, Class) -> Instance
     def find_or_create_instance: (Reference, Class) -> Instance
 
+    # @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

@@ -3,6 +3,8 @@
 require "database_test_helper"
 
 class EnqueueTest < ActiveSupport::TestCase
+  WORKER_HANG_TIMEOUT = 180
+
   class CartActor < SolidObjects::Actor
     actor_type "enqueue-carts"
 
@@ -58,6 +60,83 @@ 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 }
+    unfinished = threads.reject { |thread| thread.join(WORKER_HANG_TIMEOUT) }
+    unfinished.each(&:kill).each(&:join)
+
+    assert_empty unfinished, "an enqueue hung for over #{WORKER_HANG_TIMEOUT} seconds"
+    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