Fixed iteration tests

README.md

@@ -207,6 +207,8 @@ foreach ($results as $result) {
 }
 ```
 
+`peek()` returns the next row without moving the cursor (or `null` if there is none). It can pull the next record from the server into the buffer if needed, but does not advance past it—unlike `next()` in a `foreach`, which always consumes. Repeated calls to `peek()` return the same value until you advance the iterator (for example with `next()` or by continuing a `foreach`).
+
 Cypher values and types map to these php types and classes:
 
 | Cypher         | Php                                           |

composer.json

@@ -28,7 +28,7 @@
         "psr/http-factory": "^1.0",
         "psr/http-client": "^1.0",
         "php-http/message": "^1.0",
-        "stefanak-michal/bolt": "^7.1.4",
+        "stefanak-michal/bolt": "^7.4",
         "symfony/polyfill-php80": "^1.2",
         "psr/simple-cache": ">=2.0",
         "ext-json": "*",

src/Bolt/BoltConnection.php

@@ -71,6 +71,18 @@ class BoltConnection implements ConnectionInterface
 
     private ?float $originalTimeout = null;
 
+    /**
+     * When one PULL yields RECORD(s) then FAILURE, {@see assertNoFailure()} runs RESET on the FAILURE before we
+     * can finish yielding those rows. {@see pull()} therefore defers the {@see Neo4jException} to the next
+     * {@see BoltResult::fetchResults()} so buffered RECORDs are delivered first (TestKit pull_2_end_error.script).
+     *
+     * This is not peek-specific: any consumption path that pulls (including {@see CypherList::peek()} calling
+     * {@see Iterator::current()} on a lazy {@see BoltResult}) relies on the same ordering. Official Neo4j drivers
+     * treat peek at end-of-stream as non-throwing (null / absent record); for a FAILURE that arrives after RECORDs
+     * in the same PULL response, surfacing the error only after those rows matches the same stream semantics.
+     */
+    private ?Neo4jException $deferredPullFailure = null;
+
     /**
      * @return array{0: V4_4|V5|V5_1|V5_2|V5_3|V5_4|null, 1: Connection}
      */
@@ -217,6 +229,7 @@ public function consumeResults(): void
      */
     public function reset(): void
     {
+        $this->deferredPullFailure = null;
         $message = $this->messageFactory->createResetMessage();
         $response = $message->send()->getResponse();
         $this->assertNoFailure($response);
@@ -271,6 +284,7 @@ public function run(
         ?iterable $tsxMetadata,
     ): array {
         $extra = $this->buildRunExtra($database, $timeout, $holder, $mode, $tsxMetadata);
+
         $message = $this->messageFactory->createRunMessage($text, $parameters, $extra);
         $response = $message->send()->getResponse();
         $this->assertNoFailure($response);
@@ -337,6 +351,23 @@ public function pull(?int $qid, ?int $fetchSize): array
             return $tbr;
         } catch (Throwable $e) {
             $this->restoreOriginalTimeout();
+            if ($e instanceof Neo4jException) {
+                // RECORD(s) then FAILURE in one PULL: RESET already ran in assertNoFailure — return rows and
+                // defer the exception to the next fetchResults() so the last record is yielded first and no
+                // extra PULL is sent (pull_2_end_error.script).
+                if (!empty($tbr)) {
+                    $this->deferredPullFailure = $e;
+                    $tbr[] = ['has_more' => true];
+
+                    /** @var non-empty-list<list> */
+                    return $tbr;
+                }
+                throw $e;
+            }
+            // If we've received some records before the disconnect, return them so first next() succeeds.
+            // Second next() must pull again and fail with a connection error (TestKit exit_after_record scripts).
+            // Do not append []: BoltResult treats trailing empty SUCCESS as stream completion, so the iterator
+            // would stop cleanly instead of surfacing the disconnect. A synthetic has_more:true means "not done".
             // If we've received some records before the disconnect, return them so first next() succeeds.
             // Second next() must pull again and fail with a connection error (TestKit exit_after_record scripts).
             // Do not append []: BoltResult treats trailing empty SUCCESS as stream completion, so the iterator
@@ -363,6 +394,7 @@ public function close(): void
         // Other exceptions (Neo4jException, TypeError, etc.) should propagate.
         try {
             if ($this->isOpen()) {
+                $this->deferredPullFailure = null;
                 if ($this->isStreaming()) {
                     $this->discardUnconsumedResults();
                 }
@@ -390,11 +422,23 @@ public function close(): void
      */
     public function invalidate(): void
     {
+        $this->deferredPullFailure = null;
         $this->subscribedResults = [];
         $this->connection->disconnect();
         unset($this->boltProtocol);
     }
 
+    /**
+     * Consumes a FAILURE that was deferred from the previous {@see pull()} (RECORD(s) then FAILURE).
+     */
+    public function takeDeferredPullFailure(): ?Neo4jException
+    {
+        $e = $this->deferredPullFailure;
+        $this->deferredPullFailure = null;
+
+        return $e;
+    }
+
     private function buildRunExtra(?string $database, ?float $timeout, ?BookmarkHolder $holder, ?AccessMode $mode, ?iterable $metadata): array
     {
         $extra = [];
@@ -431,6 +475,10 @@ private function buildResultExtra(?int $fetchSize, ?int $qid): array
         }
 
         if ($qid !== null && $qid >= 0) {
+            // Always send explicit qid (including 0). Omitting qid defaults to the "current" stream; with
+            // multiple concurrent RUN streams in a transaction, PULL must target the correct stream or Neo4j
+            // returns e.g. "No such statement: N" (Neo.ClientError.Request.InvalidFormat). The Bolt library's
+            // openStreams counter is not reliable for this across all server versions and message orderings.
             $extra['qid'] = $qid;
         }
 

src/Bolt/BoltResult.php

@@ -145,6 +145,13 @@ private function fetchResults(): void
     {
         $this->networkPullOccurred = true;
 
+        $deferred = $this->connection->takeDeferredPullFailure();
+        if ($deferred !== null) {
+            throw $deferred;
+        }
+
+        $this->networkPullOccurred = true;
+
         try {
             $meta = $this->connection->pull($this->qid, $this->effectivePullSize());
         } catch (BoltConnectException|BoltException $e) {

src/Bolt/BoltUnmanagedTransaction.php

@@ -13,7 +13,6 @@
 
 namespace Laudis\Neo4j\Bolt;
 
-use Bolt\enum\ServerState;
 use Laudis\Neo4j\Contracts\ConnectionPoolInterface;
 use Laudis\Neo4j\Contracts\UnmanagedTransactionInterface;
 use Laudis\Neo4j\Databags\BookmarkHolder;
@@ -86,6 +85,8 @@ public function commit(iterable $statements = []): CypherList
 
         $this->ensureBeginSent();
 
+        $this->connection->consumeResults();
+
         // Force the results to pull all the results.
         // After a commit, the connection will be in the ready state, making it impossible to use PULL
         $tbr = $this->runStatements($statements)->each(static function (CypherList $list) {
@@ -101,6 +102,11 @@ public function commit(iterable $statements = []): CypherList
     public function rollback(): void
     {
         if ($this->isFinished()) {
+            if ($this->state === TransactionState::TERMINATED) {
+                // Run/pull already failed; connection may have been RESET — nothing to send.
+                return;
+            }
+
             if ($this->state === TransactionState::COMMITTED) {
                 throw new TransactionException("Can't rollback a committed transaction.");
             }
@@ -112,6 +118,14 @@ public function rollback(): void
 
         $this->ensureBeginSent();
 
+        // FAILURE on PULL triggers RESET in {@see BoltConnection::assertNoFailure()}; server has no open tx.
+        // TestKit stubs (e.g. tx_error_on_pull.script) expect no ROLLBACK after that RESET.
+        if ($this->connection->getServerState() === 'READY') {
+            $this->state = TransactionState::ROLLED_BACK;
+
+            return;
+        }
+
         $this->messageFactory->createRollbackMessage()->send();
         $this->state = TransactionState::ROLLED_BACK;
     }
@@ -146,8 +160,10 @@ public function runStatement(Statement $statement): SummarizedResult
         $parameters = ParameterHelper::formatParameters($statement->getParameters(), $this->connection->getProtocol());
         $start = microtime(true);
 
-        $serverState = $this->connection->protocol()->serverState;
-        if ($serverState === ServerState::STREAMING) {
+        // Only drain an outstanding autocommit result (STREAMING). In an explicit transaction (TX_STREAMING)
+        // several RUN streams may be open; consumeResults() would preload other streams and reorder PULLs
+        // vs RUN (TestKit tx_pull_1_nested*, Neo4j parallel/nested tx tests).
+        if ($this->connection->getServerState() === 'STREAMING') {
             $this->connection->consumeResults();
         }
 
@@ -217,7 +233,15 @@ public function isFinished(): bool
 
     private function ensureBeginSent(): void
     {
-        if ($this->isInstantTransaction || $this->beginSent) {
+        if ($this->isInstantTransaction) {
+            return;
+        }
+        // FAILURE on PULL triggers RESET in BoltConnection — server is READY with no tx, but we may still
+        // have beginSent=true (e.g. execute_read retry). Must send BEGIN again before RUN.
+        if ($this->beginSent && $this->state === TransactionState::ACTIVE && $this->connection->getServerState() === 'READY') {
+            $this->beginSent = false;
+        }
+        if ($this->beginSent) {
             return;
         }
         try {

src/Bolt/ConnectionPool.php

@@ -112,15 +112,10 @@ public function acquire(SessionConfiguration $config): Generator
 
     public function release(ConnectionInterface $connection): void
     {
+        // Return a permit only — keep the connection in {@see $activeConnections} so it stays
+        // pooled for reuse and is still closed by {@see close()}. Removing it here orphaned
+        // sockets (no GOODBYE on driver close), which breaks TestKit stubs after errors.
         $this->semaphore->post();
-
-        foreach ($this->activeConnections as $i => $activeConnection) {
-            if ($connection === $activeConnection) {
-                array_splice($this->activeConnections, $i, 1);
-
-                return;
-            }
-        }
     }
 
     public function getLogger(): ?Neo4jLogger

src/Bolt/ProtocolFactory.php

@@ -33,12 +33,11 @@ public function createProtocol(IConnection $connection): V4_4|V5|V5_1|V5_2|V5_3|
         }
 
         $bolt = new Bolt($connection);
-        // Offer protocol versions from newest to oldest (only 4.4 and above are supported)
         $bolt->setProtocolVersions('5.4.4', 4.4);
         $protocol = $bolt->build();
 
         if (!($protocol instanceof V4_4 || $protocol instanceof V5 || $protocol instanceof V5_1 || $protocol instanceof V5_2 || $protocol instanceof V5_3 || $protocol instanceof V5_4)) {
-            throw new RuntimeException('Client only supports bolt version 4.4 to 5.4');
+            throw new RuntimeException('Client only supports Bolt protocol 4.4 through 5.4');
         }
 
         return $protocol;

src/BoltFactory.php

@@ -30,7 +30,7 @@
 use Laudis\Neo4j\Enum\SocketType;
 
 /**
- * Small wrapper around the bolt library to easily guarantee only bolt version 3 and up will be created and authenticated.
+ * Small wrapper around the bolt library to create and authenticate Bolt connections (protocol 4.4+).
  */
 class BoltFactory
 {

src/Enum/ConnectionProtocol.php

@@ -13,7 +13,6 @@
 
 namespace Laudis\Neo4j\Enum;
 
-use Bolt\protocol\V3;
 use Bolt\protocol\V4;
 use Bolt\protocol\V4_1;
 use Bolt\protocol\V4_2;
@@ -67,7 +66,7 @@ final class ConnectionProtocol extends TypedEnum implements JsonSerializable
      *
      * @psalm-suppress ImpureMethodCall
      */
-    public static function determineBoltVersion(V3|V4|V4_1|V4_2|V4_3|V4_4|V5|V5_1|V5_2|V5_3|V5_4 $bolt): self
+    public static function determineBoltVersion(V4|V4_1|V4_2|V4_3|V4_4|V5|V5_1|V5_2|V5_3|V5_4 $bolt): self
     {
         $version = self::resolve($bolt->getVersion());
 

src/Formatter/Specialised/BoltOGMTranslator.php

@@ -298,13 +298,13 @@ private function makeFromBoltPath(BoltPath $path): Path
         foreach ($rels as $rel) {
             $relationships[] = $this->makeFromBoltUnboundRelationship($rel);
         }
-        /** @var list<int> $ids */
-        $ids = $path->ids;
+        /** @var list<int> $indices */
+        $indices = $path->indices;
 
         return new Path(
             new CypherList($nodes),
             new CypherList($relationships),
-            new CypherList($ids),
+            new CypherList($indices),
         );
     }
 

src/Types/CypherList.php

@@ -308,6 +308,24 @@ public function key(): int
         return $this->cacheKey();
     }
 
+    /**
+     * Returns the next record without consuming it (the iterator position is unchanged).
+     *
+     * This may pull and buffer the next row from the server when the result is lazy, the same as
+     * {@see current()} on a non-empty result. When there is no next record, returns `null` without
+     * erroring; calling {@see current()} on an exhausted result is not safe.
+     *
+     * @return TValue|null
+     */
+    public function peek()
+    {
+        if (!$this->valid()) {
+            return null;
+        }
+
+        return $this->current();
+    }
+
     /**
      * @return array<int, TValue>
      */

testkit-backend/features.php

@@ -99,7 +99,7 @@
     // notified when the server reports a token expired.
     'Feature:Auth:Managed' => false,
     // The driver supports Bolt protocol version 3
-    'Feature:Bolt:3.0' => true,
+    'Feature:Bolt:3.0' => false,
     // The driver supports Bolt protocol version 4.1
     'Feature:Bolt:4.1' => true,
     // The driver supports Bolt protocol version 4.2

testkit-backend/src/Handlers/ResultList.php

@@ -67,22 +67,28 @@ public function handle($request): TestkitResponseInterface
 
             return new RecordListResponse($rows);
         } catch (Neo4jException $e) {
-            $this->repository->removeRecords($request->getResultId());
+            $response = new DriverErrorResponse($request->getResultId(), $e);
+            // Keep error for RetryableNegative lookup by result id (execute_read tx_func tests).
+            $this->repository->addRecords($request->getResultId(), $response);
 
-            return new DriverErrorResponse($request->getResultId(), $e);
+            return $response;
         } catch (BoltException $e) {
             $this->repository->removeRecords($request->getResultId());
             $neo4jError = Neo4jError::fromMessageAndCode('Neo.ClientError.General.ConnectionError', $e->getMessage());
             $wrapped = new Neo4jException([$neo4jError], $e);
+            $response = new DriverErrorResponse($request->getResultId(), $wrapped);
+            $this->repository->addRecords($request->getResultId(), $response);
 
-            return new DriverErrorResponse($request->getResultId(), $wrapped);
+            return $response;
         } catch (Throwable $e) {
             $this->repository->removeRecords($request->getResultId());
             if ($this->isConnectionOrSocketError($e)) {
                 $neo4jError = Neo4jError::fromMessageAndCode('Neo.ClientError.General.ConnectionError', $e->getMessage());
                 $wrapped = new Neo4jException([$neo4jError], $e);
+                $response = new DriverErrorResponse($request->getResultId(), $wrapped);
+                $this->repository->addRecords($request->getResultId(), $response);
 
-                return new DriverErrorResponse($request->getResultId(), $wrapped);
+                return $response;
             }
             throw $e;
         }

testkit-backend/src/Handlers/ResultNext.php

@@ -73,22 +73,25 @@ public function handle($request): TestkitResponseInterface
 
             return new RecordResponse($values);
         } catch (Neo4jException $e) {
-            $this->repository->removeRecords($request->getResultId());
+            $response = new DriverErrorResponse($request->getResultId(), $e);
+            $this->repository->addRecords($request->getResultId(), $response);
 
-            return new DriverErrorResponse($request->getResultId(), $e);
+            return $response;
         } catch (BoltException $e) {
-            $this->repository->removeRecords($request->getResultId());
             $neo4jError = Neo4jError::fromMessageAndCode('Neo.ClientError.General.ConnectionError', $e->getMessage());
             $wrapped = new Neo4jException([$neo4jError], $e);
+            $response = new DriverErrorResponse($request->getResultId(), $wrapped);
+            $this->repository->addRecords($request->getResultId(), $response);
 
-            return new DriverErrorResponse($request->getResultId(), $wrapped);
+            return $response;
         } catch (Throwable $e) {
-            $this->repository->removeRecords($request->getResultId());
             if ($this->isConnectionOrSocketError($e)) {
                 $neo4jError = Neo4jError::fromMessageAndCode('Neo.ClientError.General.ConnectionError', $e->getMessage());
                 $wrapped = new Neo4jException([$neo4jError], $e);
+                $response = new DriverErrorResponse($request->getResultId(), $wrapped);
+                $this->repository->addRecords($request->getResultId(), $response);
 
-                return new DriverErrorResponse($request->getResultId(), $wrapped);
+                return $response;
             }
             throw $e;
         }