Serialize concurrent increments and unless_exist writes

app/models/solid_cache/entry.rb

@@ -70,9 +70,17 @@ def clear_delete
       def lock_and_write(key, &block)
         transaction do
           without_query_cache do
-            result = lock.where(key_hash: key_hash_for(key)).pick(:key, :value)
-            new_value = block.call(result&.first == key ? result[1] : nil)
-            write(key, new_value) if new_value
+            # A zero byte_size marks a placeholder row, which ensures missing keys can be locked.
+            key_hash = key_hash_for(key)
+            insert_all [ { key: key, value: "", key_hash: key_hash, byte_size: 0 } ], unique_by: upsert_unique_by
+
+            result = lock.where(key_hash: key_hash).pick(:key, :value, :byte_size)
+            new_value = block.call(result&.first == key && result[2] > 0 ? result[1] : nil)
+            if new_value
+              write(key, new_value)
+            elsif result&.last == 0
+              delete_by_key(key)
+            end
             new_value
           end
         end

test/models/solid_cache/entry_test.rb

@@ -14,6 +14,29 @@ class EntryTest < ActiveSupport::TestCase
       assert_equal({ "foo" => "bar", "hello" => "there" }, Entry.read_multi([ "hello".b, "foo".b, "bar".b ]))
     end
 
+    test "lock and write without a new value leaves a missing key absent" do
+      result = Entry.lock_and_write("hello".b) do |value|
+        assert_nil value
+        nil
+      end
+
+      assert_nil result
+      assert_nil Entry.read("hello".b)
+      assert_equal 0, uncached_entry_count
+    end
+
+    test "lock and write preserves an existing empty value" do
+      Entry.write("hello".b, "")
+
+      result = Entry.lock_and_write("hello".b) do |value|
+        assert_equal "", value
+        "there"
+      end
+
+      assert_equal "there", result
+      assert_equal "there", Entry.read("hello".b)
+    end
+
     test "id range" do
       assert_equal 0, Entry.id_range
 

test/unit/solid_cache_test.rb

@@ -65,6 +65,70 @@ def test_write_expired_value_with_unless_exist
     travel 2.seconds
     assert_equal true, @cache.write(1, "bbbb", expires_in: 1.second, unless_exist: true)
   end
+
+  test "concurrent writes with unless_exist to a missing key" do
+    10.times do |i|
+      key = "foo#{i}"
+      results = concurrently(key) { @cache.write(key, 1, unless_exist: true) }
+
+      assert_equal 1, results.count(true)
+      assert_equal 7, results.count(false)
+      assert_equal 1, @cache.read(key)
+    end
+  end
+
+  test "concurrent increments of a missing key" do
+    10.times do |i|
+      key = "foo#{i}"
+      results = concurrently(key) { @cache.increment(key) }
+
+      assert_equal (1..8).to_a, results.sort
+      assert_equal 8, @cache.read(key)
+    end
+  end
+
+  private
+    def concurrently(key, &block)
+      key = @cache.send(:normalize_key, key, {})
+      # Keep schema discovery out of the concurrent SQLite transactions.
+      @cache.connections.with_connection_for(key) do
+        SolidCache::Entry.connection_pool.schema_cache.add(SolidCache::Entry.table_name)
+      end
+
+      ready = Queue.new
+      start = Queue.new
+      threads = 8.times.map do
+        Thread.new do
+          ready << true
+          start.pop
+          Rails.application.executor.wrap do
+            @cache.connections.with_connection_for(key) do
+              with_sqlite_busy_handler(&block)
+            end
+          end
+        end
+      end
+
+      8.times { ready.pop }
+      8.times { start << true }
+      threads.map(&:value)
+    ensure
+      threads&.each(&:kill)
+      threads&.each(&:join)
+    end
+
+    def with_sqlite_busy_handler(&block)
+      SolidCache::Record.with_connection do |connection|
+        if connection.adapter_name == "SQLite" && Rails::VERSION::MAJOR == 7
+          # Rails 7.2's busy_timeout holds the GVL while waiting for other writers.
+          timeout = connection.pool.db_config.configuration_hash.fetch(:timeout)
+          connection.raw_connection.busy_handler_timeout = timeout
+        end
+        block.call
+      ensure
+        connection.raw_connection.busy_timeout(timeout) if timeout
+      end
+    end
 end
 
 class SolidCacheFailsafeTest < ActiveSupport::TestCase