fixed connection reuse bug during tests

Author: transistive

src/Bolt/BoltConnection.php

@@ -18,16 +18,14 @@
 use Bolt\error\MessageException;
 use Bolt\protocol\V3;
 use Bolt\protocol\V4;
-use Laudis\Neo4j\Types\CypherList;
-use function count;
 use Laudis\Neo4j\BoltFactory;
 use Laudis\Neo4j\Common\ConnectionConfiguration;
 use Laudis\Neo4j\Contracts\ConnectionInterface;
 use Laudis\Neo4j\Databags\DatabaseInfo;
 use Laudis\Neo4j\Databags\DriverConfiguration;
-use Laudis\Neo4j\Databags\SummarizedResult;
 use Laudis\Neo4j\Enum\AccessMode;
 use Laudis\Neo4j\Enum\ConnectionProtocol;
+use Laudis\Neo4j\Types\CypherList;
 use LogicException;
 use Psr\Http\Message\UriInterface;
 use RuntimeException;
@@ -127,7 +125,7 @@ public function getDatabaseInfo(): ?DatabaseInfo
      */
     public function isOpen(): bool
     {
-        return $this->boltProtocol !== null;
+        return $this->serverState !== 'DISCONNECTED' && $this->serverState !== 'DEFUNCT';
     }
 
     public function open(): void
@@ -157,6 +155,7 @@ public function close(): void
         $this->protocol()->goodbye();
 
         $this->serverState = 'DEFUNCT';
+        $this->boltProtocol = null; // has to be set to null as the sockets don't recover nicely contrary to what the underlying code might lead you to believe
     }
 
     private function consumeResults(): void
@@ -214,6 +213,13 @@ public function begin(?string $database, ?float $timeout): void
 
             throw $e;
         }
+
+        $this->serverState = 'TX_READY';
+    }
+
+    public function getFactory(): BoltFactory
+    {
+        return $this->factory;
     }
 
     /**
@@ -360,7 +366,7 @@ public function pull(?int $qid, ?int $fetchSize): array
             throw $e;
         }
 
-        $this->interpretResult($tbr);
+        $this->interpretResult($tbr[count($tbr) - 1]);
 
         return $tbr;
     }
@@ -375,7 +381,7 @@ public function getDriverConfiguration(): DriverConfiguration
 
     public function __destruct()
     {
-        if ($this->serverState !== 'DISCONNECTED' && $this->serverState !== 'DEFUNCT') {
+        if ($this->isOpen()) {
             $this->close();
         }
     }
@@ -396,11 +402,11 @@ private function buildRunExtra(?string $database, ?float $timeout): array
     private function buildResultExtra(?int $fetchSize, ?int $qid): array
     {
         $extra = [];
-        if ($fetchSize) {
+        if ($fetchSize !== null) {
             $extra['n'] = $fetchSize;
         }
 
-        if ($qid) {
+        if ($qid !== null) {
             $extra['qid'] = $qid;
         }
 
@@ -429,7 +435,7 @@ private function protocol(): V3
     private function interpretResult(array $result): void
     {
         if (str_starts_with($this->serverState, 'TX_')) {
-            if ($result['has_more'] ?? count($this->subscribedResults) === 1) {
+            if ($result['has_more'] ?? ($this->countResults() > 1)) {
                 $this->serverState = 'TX_STREAMING';
             } else {
                 $this->serverState = 'TX_READY';
@@ -440,4 +446,16 @@ private function interpretResult(array $result): void
             $this->serverState = 'READY';
         }
     }
+
+    private function countResults(): int
+    {
+        $ctr = 0;
+        foreach ($this->subscribedResults as $result) {
+            if ($result->get() !== null) {
+                ++$ctr;
+            }
+        }
+
+        return $ctr;
+    }
 }

src/Bolt/BoltConnectionPool.php

@@ -65,11 +65,7 @@ public function acquire(
 
         foreach (self::$connectionCache[$key] as $i => $connection) {
             if (!$connection->isOpen()) {
-                $sslConfig = $connection->getDriverConfiguration()->getSslConfiguration();
-                $newSslConfig = $this->driverConfig->getSslConfiguration();
-                if ($sslConfig->getMode() !== $newSslConfig->getMode() ||
-                    $sslConfig->isVerifyPeer() === $newSslConfig->isVerifyPeer()
-                ) {
+                if ($this->compare($connection, $authenticate)) {
                     $connection = $this->getConnection($connectingTo, $authenticate, $config);
 
                     /** @psalm-suppress PropertyTypeCoercion */
@@ -82,6 +78,9 @@ public function acquire(
 
                 return $connection;
             }
+            if ($connection->getServerState() === 'READY' && !$this->compare($connection, $authenticate)) {
+                return $connection;
+            }
         }
 
         $connection = $this->getConnection($connectingTo, $authenticate, $config);
@@ -124,4 +123,14 @@ private function getConnection(
 
         return new BoltConnection($factory, $bolt, $config);
     }
+
+    private function compare(BoltConnection $connection, AuthenticateInterface $authenticate): bool
+    {
+        $sslConfig = $connection->getDriverConfiguration()->getSslConfiguration();
+        $newSslConfig = $this->driverConfig->getSslConfiguration();
+
+        return $sslConfig->getMode() !== $newSslConfig->getMode() ||
+            $sslConfig->isVerifyPeer() === $newSslConfig->isVerifyPeer() ||
+            $authenticate !== $connection->getFactory()->getAuth();
+    }
 }

src/Bolt/BoltResult.php

@@ -146,7 +146,7 @@ public function rewind(): void
 
     public function __destruct()
     {
-        if (in_array($this->connection->getServerState(), ['STREAMING', 'TX_STREAMING', 'FAILED', 'INTERRUPTED'], true)) {
+        if (in_array($this->connection->getServerState(), ['STREAMING', 'TX_STREAMING'], true)) {
             $this->discard();
         }
     }

src/Bolt/Session.php

@@ -177,7 +177,7 @@ private function startTransaction(TransactionConfiguration $config, SessionConfi
 
             $connection->begin($this->config->getDatabase(), $config->getTimeout());
         } catch (MessageException $e) {
-            if (isset($connection)) {
+            if (isset($connection) && $connection->getServerState() === 'FAILED') {
                 $connection->reset();
             }
             throw Neo4jException::fromMessageException($e);

src/BoltFactory.php

@@ -113,4 +113,9 @@ private static function configureSsl(UriInterface $uri, StreamSocket $socket, Dr
             $socket->setSslContextOptions($options);
         }
     }
+
+    public function getAuth(): AuthenticateInterface
+    {
+        return $this->auth;
+    }
 }

src/Formatter/SummarizedResultFormatter.php

@@ -197,13 +197,9 @@ public function formatBoltResult(array $meta, BoltResult $result, BoltConnection
         /**
          * @psalm-suppress MixedArgument
          *
-         * @var SummarizedResult<CypherMap<OGMTypes>> $result
+         * @var SummarizedResult<CypherMap<OGMTypes>>
          */
-        $result = (new SummarizedResult($summary, $formattedResult))->withCacheLimit($result->getFetchSize());
-
-        $connection->subscribeResult($result);
-
-        return $result;
+        return (new SummarizedResult($summary, $formattedResult))->withCacheLimit($result->getFetchSize());
     }
 
     /**

tests/Integration/BasicDriverTest.php

@@ -44,7 +44,7 @@ public function getConnections(): array
 
         $tbr = [];
         foreach (explode(',', $connections) as $connection) {
-            $tbr[] = [$connection];
+            $tbr[$connection] = [$connection];
         }
 
         return $tbr;

tests/Integration/ComplexQueryTest.php

@@ -409,11 +409,7 @@ public function testDiscardAfterTimeout(string $alias): void
         $result = $this->getClient()
             ->getDriver($alias)
             ->createSession()
-            ->run(
-                "MATCH (n:Node) SET n.testing = 'hello' WITH * CALL apoc.util.sleep(2000000)",
-                [],
-                TransactionConfiguration::default()->withTimeout(150)
-            );
+            ->run("CALL apoc.util.sleep(2000000)", [], TransactionConfiguration::default()->withTimeout(150));
 
         unset($result);
     }