fix: one Bolt connection per thread, not one per pool

lib/active_cypher/bolt/transaction.rb

@@ -77,7 +77,7 @@ def run(query, parameters = {})
                 @state = :failed
                 code = msg.metadata['code']
                 message = msg.metadata['message']
-                raise QueryError, "Query execution failed: #{code} - #{message}"
+                raise query_error_for(code), "Query execution failed: #{code} - #{message}"
               else
                 raise ProtocolError, "Unexpected message type: #{msg.class}"
               end
@@ -90,7 +90,7 @@ def run(query, parameters = {})
             @state = :failed
             code = response.metadata['code']
             message = response.metadata['message']
-            raise QueryError, "Query execution failed: #{code} - #{message}"
+            raise query_error_for(code), "Query execution failed: #{code} - #{message}"
           else
             raise ProtocolError, "Unexpected response to RUN: #{response.class}"
           end
@@ -204,6 +204,18 @@ def rolled_back?
       def failed?
         @state == :failed
       end
+
+      private
+
+      # Server error code -> exception class. Substring match covers both
+      # Neo4j and Memgraph spellings of the same codes.
+      def query_error_for(code)
+        c = code.to_s
+        return ConstraintError if c.include?('Constraint')
+        return TransientError  if c.include?('TransientError')
+
+        QueryError
+      end
     end
   end
 end

lib/active_cypher/connection_adapters/memgraph_adapter.rb

@@ -112,6 +112,8 @@ def execute_ddl(cypher, params = {})
         end
       end
 
+      TRANSIENT_RETRIES = 3
+
       # Override run to execute queries using auto-commit mode.
       # Memgraph auto-commits each query, so we send RUN + PULL directly
       # without BEGIN/COMMIT wrapper. This avoids transaction state issues.
@@ -120,7 +122,26 @@ def run(cypher, params = {}, context: 'Query', db: nil, access_mode: :write)
         logger.debug { "[#{context}] #{cypher} #{params.inspect}" }
 
         instrument_query(cypher, params, context: context, metadata: { db: db, access_mode: access_mode }) do
-          run_auto_commit(cypher, prepare_params(params))
+          with_transient_retry(context) { run_auto_commit(cypher, prepare_params(params)) }
+        end
+      end
+
+      # Retry write/write conflicts the server flagged as transient. Safe only
+      # because this path is auto-commit (the whole query rolled back); do NOT
+      # lift into the explicit-transaction path, where earlier statements may
+      # already have applied.
+      def with_transient_retry(context)
+        attempts = 0
+        begin
+          yield
+        rescue ActiveCypher::TransientError => e
+          attempts += 1
+          raise if attempts > TRANSIENT_RETRIES
+
+          # Full jitter, so conflicting writers don't back off in step.
+          sleep(rand * 0.05 * (2**(attempts - 1)))
+          logger.debug { "[#{context}] transient conflict, retry #{attempts}/#{TRANSIENT_RETRIES}: #{e.message}" }
+          retry
         end
       end
 

lib/active_cypher/connection_pool.rb

@@ -25,27 +25,40 @@ def initialize(spec)
         @spec = resolved_config.merge(@spec.except(:url))
       end
 
-      @conn_ref = nil # holds the adapter instance
-      @creation_mutex = Mutex.new # prevents multiple threads from creating connections simultaneously
+      # One connection per thread: Bolt is a stateful, ordered protocol, so
+      # sharing a socket across threads corrupts the stream. @connections
+      # exists only so disconnect can close them all; the thread-local is
+      # what the hot path reads.
+      @connections = {}
+      @creation_mutex = Mutex.new
     end
 
-    # Returns a live adapter, initializing it once in a thread‑safe way.
+    # Returns a live adapter belonging to the calling thread.
     def connection
-      # Fast path — already connected and alive
-      conn = @conn_ref
+      conn = Thread.current[thread_key]
       return conn if conn&.active?
 
-      # Use mutex for the slow path to prevent thundering herd
-      @creation_mutex.synchronize do
-        # Check again inside the mutex in case another thread created it
-        conn = @conn_ref
-        return conn if conn&.active?
+      # Built outside the mutex: connecting does network IO.
+      new_conn = build_connection
+      Thread.current[thread_key] = new_conn
+
+      # Reap dead threads' connections, or each short-lived thread leaks a
+      # socket the server counts against max_connections.
+      orphaned = @creation_mutex.synchronize do
+        dead = @connections.reject { |t, _| t.alive? }
+        dead.each_key { |t| @connections.delete(t) }
+        @connections[Thread.current] = new_conn
+        dead.values
+      end
 
-        # Create a new connection
-        new_conn = build_connection
-        @conn_ref = new_conn
-        return new_conn
+      # Closed outside the mutex: disconnecting does IO too.
+      orphaned.each do |conn|
+        conn.disconnect
+      rescue StandardError => e
+        puts "Warning: Error disconnecting orphaned connection: #{e.message}" if ENV['DEBUG']
       end
+
+      new_conn
     end
     alias checkout connection
 
@@ -54,23 +67,31 @@ def troubled?
       @retry_count >= @spec[:max_retries]
     end
 
-    # Explicitly close and reset the connection
+    # Explicitly close every connection this pool handed out.
     def disconnect
-      conn = @conn_ref
-      return unless conn
+      conns = @creation_mutex.synchronize do
+        taken = @connections.values
+        @connections.clear
+        taken
+      end
 
-      begin
+      conns.each do |conn|
         conn.disconnect
       rescue StandardError => e
         # Log but don't raise to ensure cleanup continues
         puts "Warning: Error disconnecting: #{e.message}" if ENV['DEBUG']
-      ensure
-        @conn_ref = nil
       end
+
+      Thread.current[thread_key] = nil
     end
 
     private
 
+    # Namespaced per pool instance so multiple databases don't share.
+    def thread_key
+      @thread_key ||= :"active_cypher_connection_#{object_id}"
+    end
+
     def build_connection
       adapter_name = @spec[:adapter]
       raise ArgumentError, 'Missing adapter name in connection specification' unless adapter_name

lib/activecypher.rb

@@ -52,6 +52,14 @@ class ProtocolError < ConnectionError; end
   # Could be you. Could be Cypher. Could be fate.
   class QueryError < Error; end
 
+  # A uniqueness or existence constraint said no.
+  # Rescue this to ignore duplicates without swallowing every QueryError.
+  class ConstraintError < QueryError; end
+
+  # Two writers touched the same node. Nothing is wrong with your query.
+  # The server is politely asking you to try again.
+  class TransientError < QueryError; end
+
   # Your Cypher syntax is... interpretive.
   # Unfortunately, the parser isn’t in the mood for interpretive dance.
   class CypherSyntaxError < QueryError; end

test/active_cypher/connection_pool_threading_test.rb

@@ -0,0 +1,84 @@
+# frozen_string_literal: true
+
+require 'minitest/autorun'
+require 'active_cypher/connection_pool'
+
+# The pool must hand each thread its own connection.
+#
+# Bolt is a stateful, ordered protocol over one socket. Sharing an adapter
+# across threads lets their reads and writes interleave, and the failure is not
+# a clean error -- it is a corrupted stream:
+#
+#   ActiveCypher::ProtocolError: Failed to decode message:
+#     FrozenError - can't modify frozen IO::Stream::StringBuffer
+#
+# Observed with six threads against Memgraph, plus spurious
+# Memgraph.ExecutionException from replies matched to the wrong request. Any
+# threaded job runner hits it: GoodJob defaults to 5 threads, Sidekiq to 10.
+class ConnectionPoolThreadingTest < Minitest::Test
+  # A pool whose build step is stubbed, so this tests the handing-out and not
+  # the network.
+  class FakeAdapter
+    def active? = true
+    def disconnect = nil
+  end
+
+  def pool
+    p = ActiveCypher::ConnectionPool.allocate
+    p.instance_variable_set(:@spec, { adapter: 'fake' })
+    p.instance_variable_set(:@connections, {})
+    p.instance_variable_set(:@creation_mutex, Mutex.new)
+    def p.build_connection = FakeAdapter.new
+    p
+  end
+
+  def test_each_thread_gets_its_own_connection
+    p = pool
+    conns = 8.times.map { Thread.new { p.connection } }.map(&:value)
+
+    assert_equal 8, conns.map(&:object_id).uniq.size,
+                 'threads shared a connection; the Bolt stream will corrupt'
+  end
+
+  def test_the_same_thread_reuses_its_connection
+    p = pool
+
+    assert_same p.connection, p.connection
+  end
+
+  def test_disconnect_closes_every_live_thread_s_connection
+    p = pool
+    gate = Queue.new
+    # Threads held ALIVE, so their connections are genuinely tracked rather
+    # than reaped as orphans while the test is still setting up.
+    ts = 4.times.map { Thread.new { p.connection; gate.pop } }
+    sleep 0.05 until p.instance_variable_get(:@connections).size == 4
+
+    p.disconnect
+
+    assert_empty p.instance_variable_get(:@connections)
+    4.times { gate << :go }
+    ts.each(&:join)
+  end
+
+  # A connection whose thread has exited must be CLOSED, not merely forgotten.
+  # Otherwise a process that spawns short-lived threads leaks a Bolt socket per
+  # thread, and the server counts them against max_connections.
+  def test_connections_of_dead_threads_are_disconnected
+    closed = []
+    p = pool
+    p.define_singleton_method(:build_connection) do
+      a = FakeAdapter.new
+      a.define_singleton_method(:disconnect) { closed << object_id }
+      a
+    end
+
+    dead = Thread.new { p.connection }
+    dead.join
+    refute dead.alive?
+
+    p.connection # triggers the reap
+
+    assert_equal 1, closed.size, "the dead thread's connection was leaked"
+  end
+end