Connection Receive Timeout Configuration Hint Implementation (#279)

- Removed conversion of connection/timeout errors to NotALeader in BoltResult.php and BoltMessage.php; original exceptions are rethrown after invalidating the connection.
 - Updated Session.isConnectionError() to check for BoltConnectException and removed NotALeader from it.
 - Replaced catch (Throwable) with catch (BoltConnectException) in BoltConnection.close() and discardUnconsumedResults().
 - Added a BoltConnectException handler in the testkit backend AbstractRunner that wraps connection/timeout errors in Neo4jException when building DriverErrorResponse.

Author: p123-stack

.github/workflows/testkit.yml

@@ -13,6 +13,8 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
+        with:
+          submodules: recursive
 
       - name: Restore Neo4j Image Cache if it exists
         id: cache-docker-neo4j

.gitmodules

@@ -1,5 +1,3 @@
 [submodule "testkit"]
 	path = testkit
-	url = https://github.com/neo4j-drivers/testkit.git
-	branch = 5.0
-
+	url = git@github.com:nagels-tech/testkit.git

src/Bolt/BoltConnection.php

@@ -15,6 +15,7 @@
 
 use Bolt\enum\ServerState;
 use Bolt\enum\Signature;
+use Bolt\error\ConnectException as BoltConnectException;
 use Bolt\protocol\Response;
 use Bolt\protocol\V4_4;
 use Bolt\protocol\V5;
@@ -315,6 +316,9 @@ public function __destruct()
 
     public function close(): void
     {
+        // Graceful cleanup: GOODBYE/DISCARD may fail if connection already broken.
+        // Only catch network/connection failures - if connection is broken we can't send anyway.
+        // Other exceptions (Neo4jException, TypeError, etc.) should propagate.
         try {
             if ($this->isOpen()) {
                 if ($this->isStreaming()) {
@@ -326,10 +330,29 @@ public function close(): void
 
                 unset($this->boltProtocol); // has to be set to null as the sockets don't recover nicely contrary to what the underlying code might lead you to believe;
             }
-        } catch (Throwable) {
+        } catch (BoltConnectException $e) {
+            $this->logger?->log(LogLevel::WARNING, 'Failed to close connection gracefully', [
+                'exception' => $e->getMessage(),
+            ]);
         }
     }
 
+    /**
+     * Invalidates the connection without sending GOODBYE message.
+     *
+     * This method closes the Bolt protocol and socket connection WITHOUT
+     * sending a GOODBYE message, which is essential when handling timeout
+     * exceptions or when the connection is already broken. Sending GOODBYE
+     * on a broken connection can interfere with the server's expected
+     * message sequence.
+     */
+    public function invalidate(): void
+    {
+        $this->subscribedResults = [];
+        $this->connection->disconnect();
+        unset($this->boltProtocol);
+    }
+
     private function buildRunExtra(?string $database, ?float $timeout, ?BookmarkHolder $holder, ?AccessMode $mode, ?iterable $metadata): array
     {
         $extra = [];
@@ -416,31 +439,31 @@ public function assertNoFailure(Response $response): void
 
     /**
      * Discard unconsumed results - sends DISCARD to server for each subscribed result.
+     * Try-catch prevents DISCARD failures from breaking cleanup chain in Session.close().
      */
     public function discardUnconsumedResults(): void
     {
-        if (!in_array($this->protocol()->serverState, [ServerState::STREAMING, ServerState::TX_STREAMING], true)) {
-            return;
-        }
-
-        $this->logger?->log(LogLevel::DEBUG, 'Discarding unconsumed results');
-
-        $this->subscribedResults = array_values(array_filter(
-            $this->subscribedResults,
-            static fn (WeakReference $ref): bool => $ref->get() !== null
-        ));
-
-        if (!empty($this->subscribedResults)) {
-            try {
-                $this->discard(null);
-                $this->logger?->log(LogLevel::DEBUG, 'Sent DISCARD ALL for unconsumed results');
-            } catch (Throwable $e) {
-                $this->logger?->log(LogLevel::ERROR, 'Failed to discard results', [
-                    'exception' => $e->getMessage(),
-                ]);
+        if ($this->isOpen() && in_array($this->protocol()->serverState, [ServerState::STREAMING, ServerState::TX_STREAMING], true)) {
+            $this->logger?->log(LogLevel::DEBUG, 'Discarding unconsumed results');
+
+            $this->subscribedResults = array_values(array_filter(
+                $this->subscribedResults,
+                static fn (WeakReference $ref): bool => $ref->get() !== null
+            ));
+
+            if (!empty($this->subscribedResults)) {
+                try {
+                    $this->discard(null);
+                    $this->logger?->log(LogLevel::DEBUG, 'Sent DISCARD ALL for unconsumed results');
+                } catch (BoltConnectException $e) {
+                    // Connection already broken - can't send DISCARD. Log and continue cleanup.
+                    $this->logger?->log(LogLevel::ERROR, 'Failed to discard results', [
+                        'exception' => $e->getMessage(),
+                    ]);
+                }
             }
-        }
 
-        $this->subscribedResults = [];
+            $this->subscribedResults = [];
+        }
     }
 }

src/Bolt/BoltResult.php

@@ -14,6 +14,9 @@
 namespace Laudis\Neo4j\Bolt;
 
 use function array_splice;
+
+use Bolt\error\ConnectException as BoltConnectException;
+
 use function count;
 
 use Generator;
@@ -100,7 +103,14 @@ public function consume(): array
 
     private function fetchResults(): void
     {
-        $meta = $this->connection->pull($this->qid, $this->fetchSize);
+        try {
+            $meta = $this->connection->pull($this->qid, $this->fetchSize);
+        } catch (BoltConnectException $e) {
+            // Invalidate connection on socket/network errors so pool does not reuse it.
+            // Rethrow as-is - Session retry logic inspects the actual exception via isConnectionError().
+            $this->connection->invalidate();
+            throw $e;
+        }
 
         /** @var list<list> $rows */
         $rows = array_splice($meta, 0, count($meta) - 1);
@@ -154,6 +164,13 @@ public function __destruct()
 
     public function discard(): void
     {
-        $this->connection->discard($this->qid === -1 ? null : $this->qid);
+        try {
+            $this->connection->discard($this->qid === -1 ? null : $this->qid);
+        } catch (BoltConnectException $e) {
+            // Connection already broken if DISCARD fails. Invalidate to prevent pool from reusing it.
+            // Don't rethrow: this is called from __destruct() where exceptions don't propagate properly.
+            // Connection will be detected as broken on next operation when pool tries to reuse it.
+            $this->connection->invalidate();
+        }
     }
 }

src/Bolt/BoltUnmanagedTransaction.php

@@ -14,6 +14,7 @@
 namespace Laudis\Neo4j\Bolt;
 
 use Bolt\enum\ServerState;
+use Laudis\Neo4j\Contracts\ConnectionPoolInterface;
 use Laudis\Neo4j\Contracts\UnmanagedTransactionInterface;
 use Laudis\Neo4j\Databags\BookmarkHolder;
 use Laudis\Neo4j\Databags\SessionConfiguration;
@@ -53,6 +54,7 @@ public function __construct(
         private readonly BookmarkHolder $bookmarkHolder,
         private readonly BoltMessageFactory $messageFactory,
         private readonly bool $isInstantTransaction,
+        private readonly ?ConnectionPoolInterface $pool = null,
     ) {
     }
 
@@ -149,11 +151,14 @@ public function runStatement(Statement $statement): SummarizedResult
                 $this->database,
                 $this->tsxConfig->getTimeout(),
                 $this->isInstantTransaction ? $this->bookmarkHolder : null, // let the begin transaction pass the bookmarks if it is a managed transaction
-                $this->isInstantTransaction ? $this->config->getAccessMode() : null, // let the begin transaction decide if it is a managed transaction
+                null, // mode is never sent in RUN messages - it comes from session configuration
                 $this->tsxConfig->getMetaData()
             );
         } catch (Throwable $e) {
             $this->state = TransactionState::TERMINATED;
+            if ($this->pool !== null) {
+                $this->pool->release($this->connection);
+            }
             throw $e;
         }
         $run = microtime(true);