fixed configuration_hints testkit tests

p123-stack · p123-stack · commit 976497d576d4 · 2025-12-29T15:46:35.000+05:30
diff --git a/src/Bolt/Session.php b/src/Bolt/Session.php
@@ -33,6 +33,7 @@
 use Laudis\Neo4j\Neo4j\Neo4jConnectionPool;
 use Laudis\Neo4j\Types\CypherList;
 use Psr\Log\LogLevel;
+use Throwable;
 
 /**
  * A session using bolt connections.
@@ -73,8 +74,56 @@ public function runStatements(iterable $statements, ?TransactionConfiguration $c
 
         $this->getLogger()?->log(LogLevel::INFO, 'Running statements', ['statements' => $statements]);
         $config = $this->mergeTsxConfig($config);
+
         foreach ($statements as $statement) {
-            $tbr[] = $this->beginInstantTransaction($this->config, $config)->runStatement($statement);
+            // Wrap in retry logic for connection errors
+            $retries = 0;
+            $maxRetries = 3;
+
+            while ($retries < $maxRetries) {
+                try {
+                    $tbr[] = $this->beginInstantTransaction($this->config, $config)->runStatement($statement);
+                    break; // Success, exit retry loop
+                } catch (Neo4jException $e) {
+                    if ($this->shouldClearRoutingTable($e)) {
+                        $this->getLogger()?->log(LogLevel::WARNING, 'Connection error in instant transaction, retrying', [
+                            'error' => $e->getMessage(),
+                            'retry' => $retries + 1,
+                        ]);
+
+                        if ($this->pool instanceof Neo4jConnectionPool) {
+                            $this->pool->clearRoutingTable($this->config);
+                        }
+                        $this->pool->close();
+
+                        ++$retries;
+                        if ($retries >= $maxRetries) {
+                            throw $e;
+                        }
+                    } else {
+                        throw $e;
+                    }
+                } catch (Throwable $e) {
+                    if ($this->isConnectionError($e)) {
+                        $this->getLogger()?->log(LogLevel::WARNING, 'Connection error in instant transaction, retrying', [
+                            'error' => $e->getMessage(),
+                            'retry' => $retries + 1,
+                        ]);
+
+                        if ($this->pool instanceof Neo4jConnectionPool) {
+                            $this->pool->clearRoutingTable($this->config);
+                        }
+                        $this->pool->close();
+
+                        ++$retries;
+                        if ($retries >= $maxRetries) {
+                            throw $e;
+                        }
+                    } else {
+                        throw $e;
+                    }
+                }
+            }
         }
 
         return new CypherList($tbr);
@@ -141,16 +190,74 @@ private function retry(callable $tsxHandler, bool $read, TransactionConfiguratio
                     $transaction->rollback();
                 }
 
-                if ($e->getTitle() === 'NotALeader') {
+                // ADD THIS SECTION - Handle connection timeouts and routing failures
+                if ($e->getTitle() === 'NotALeader'
+                    || $e->getNeo4jCode() === 'Neo.ClientError.Cluster.NotALeader'
+                    || $this->isConnectionError($e)) {
+                    // Clear routing table before closing pool to force fresh ROUTE request on retry
+                    if ($this->pool instanceof Neo4jConnectionPool) {
+                        $this->pool->clearRoutingTable($this->config);
+                    }
                     // By closing the pool, we force the connection to be re-acquired and the routing table to be refetched
                     $this->pool->close();
                 } elseif ($e->getClassification() !== 'TransientError') {
                     throw $e;
                 }
+            } catch (Exception $e) {
+                if ($this->isConnectionError($e)) {
+                    if ($this->pool instanceof Neo4jConnectionPool) {
+                        $this->pool->clearRoutingTable($this->config);
+                    }
+                    $this->pool->close();
+                } else {
+                    throw $e;
+                }
             }
         }
     }
 
+    /**
+     * Check if the exception is a connection-related error.
+     */
+    private function isConnectionError(Throwable $e): bool
+    {
+        $message = strtolower($e->getMessage());
+
+        // Check for common connection error messages
+        if (str_contains($message, 'interrupted system call')
+            || str_contains($message, 'broken pipe')
+            || str_contains($message, 'connection reset')
+            || str_contains($message, 'connection timeout')
+            || str_contains($message, 'connection closed')) {
+            return true;
+        }
+
+        // Check for Neo4jException-specific codes
+        if ($e instanceof Neo4jException) {
+            return $e->getNeo4jCode() === 'Neo.ClientError.Cluster.NotALeader';
+        }
+
+        return false;
+    }
+
+    /**
+     * Check if the exception should trigger a routing table clear.
+     */
+    private function shouldClearRoutingTable(Neo4jException $e): bool
+    {
+        $message = strtolower($e->getMessage());
+        $title = $e->getTitle();
+
+        // Clear routing table for timeout, connection, and cluster errors
+        return str_contains($message, 'interrupted system call')
+            || str_contains($message, 'broken pipe')
+            || str_contains($message, 'connection reset')
+            || str_contains($message, 'connection timeout')
+            || str_contains($message, 'connection closed')
+            || $e->getNeo4jCode() === 'Neo.ClientError.Cluster.NotALeader'
+            || $title === 'NotALeader';
+    }
+
     private static function triggerLazyResult(mixed $tbr): void
     {
         if ($tbr instanceof CypherSequence) {
diff --git a/src/Common/SysVSemaphore.php b/src/Common/SysVSemaphore.php
@@ -29,8 +29,11 @@
 
 class SysVSemaphore implements SemaphoreInterface
 {
+    /**
+     * @psalm-suppress UndefinedClass
+     */
     private function __construct(
-        private readonly \SysvSemaphore $semaphore,
+        private readonly mixed $semaphore,
     ) {
     }
 
diff --git a/src/Neo4j/Neo4jConnectionPool.php b/src/Neo4j/Neo4jConnectionPool.php
@@ -134,6 +134,8 @@ public function acquire(SessionConfiguration $config): Generator
         $latestError = null;
 
         if ($table == null) {
+            $this->getLogger()?->log(LogLevel::DEBUG, 'Routing table not found in cache, fetching new routing table');
+
             $addresses = $this->getAddresses($this->data->getUri()->getHost());
             foreach ($addresses as $address) {
                 $triedAddresses[] = $address;
@@ -150,8 +152,18 @@ public function acquire(SessionConfiguration $config): Generator
                      */
                     $connection = GeneratorHelper::getReturnFromGenerator($pool->acquire($config));
                     $table = $this->routingTable($connection, $config);
+
+                    $this->getLogger()?->log(LogLevel::DEBUG, 'Successfully fetched routing table', [
+                        'ttl' => $table->getTtl(),
+                        'leaders' => $table->getWithRole(RoutingRoles::LEADER()),
+                        'followers' => $table->getWithRole(RoutingRoles::FOLLOWER()),
+                        'routers' => $table->getWithRole(RoutingRoles::ROUTE()),
+                    ]);
                 } catch (ConnectException $e) {
-                    // todo - once client side logging is implemented it must be conveyed here.
+                    $this->getLogger()?->log(LogLevel::WARNING, 'Failed to connect to address', [
+                        'address' => $address,
+                        'error' => $e->getMessage(),
+                    ]);
                     $latestError = $e;
                     continue; // We continue if something is wrong with the current server
                 }
@@ -174,6 +186,11 @@ public function acquire(SessionConfiguration $config): Generator
             $server = $server->withScheme($this->data->getUri()->getScheme());
         }
 
+        $this->getLogger()?->log(LogLevel::DEBUG, 'Acquiring connection from server', [
+            'server' => (string) $server,
+            'access_mode' => $config->getAccessMode()?->getValue(),
+        ]);
+
         return $this->createOrGetPool($this->data->getUri()->getHost(), $server)->acquire($config);
     }
 
@@ -182,6 +199,89 @@ public function getLogger(): ?Neo4jLogger
         return $this->logger;
     }
 
+    /**
+     * Get the current routing table from cache for the given session configuration.
+     *
+     * @return RoutingTable|null The cached routing table, or null if not yet initialized
+     */
+    public function getRoutingTable(SessionConfiguration $config): ?RoutingTable
+    {
+        $key = $this->createKey($this->data, $config);
+        /** @var RoutingTable|null $table */
+        $table = $this->cache->get($key);
+
+        return $table;
+    }
+
+    /**
+     * Clear the cached routing table for the given session configuration.
+     * This forces a new routing table to be fetched on the next acquire() call.
+     */
+    public function clearRoutingTable(SessionConfiguration $config): void
+    {
+        $key = $this->createKey($this->data, $config);
+        $deleted = $this->cache->delete($key);
+
+        $this->getLogger()?->log(LogLevel::INFO, 'Cleared routing table from cache', [
+            'key' => $key,
+            'deleted' => $deleted,
+        ]);
+    }
+
+    /**
+     * Remove a failed server from the routing table.
+     * This removes the server from all roles (leader, follower, router) and updates the cache.
+     *
+     * @param SessionConfiguration $config        The session configuration
+     * @param string               $serverAddress The address of the failed server (e.g., "172.18.0.3:9010")
+     */
+    public function removeFailedServer(SessionConfiguration $config, string $serverAddress): void
+    {
+        $key = $this->createKey($this->data, $config);
+        /** @var RoutingTable|null $table */
+        $table = $this->cache->get($key);
+
+        if ($table !== null) {
+            $this->getLogger()?->log(LogLevel::WARNING, 'Removing failed server from routing table', [
+                'server' => $serverAddress,
+            ]);
+
+            // Remove the server and update cache
+            $updatedTable = $table->removeServer($serverAddress);
+
+            // Only update cache if the table actually changed
+            if ($updatedTable !== $table) {
+                $this->cache->set($key, $updatedTable, $updatedTable->getTtl());
+
+                $this->getLogger()?->log(LogLevel::INFO, 'Updated routing table after removing failed server', [
+                    'server' => $serverAddress,
+                    'remaining_leaders' => $updatedTable->getWithRole(RoutingRoles::LEADER()),
+                    'remaining_followers' => $updatedTable->getWithRole(RoutingRoles::FOLLOWER()),
+                    'remaining_routers' => $updatedTable->getWithRole(RoutingRoles::ROUTE()),
+                ]);
+            }
+        }
+    }
+
+    /**
+     * Check if a server exists in the routing table.
+     *
+     * @param SessionConfiguration $config        The session configuration
+     * @param string               $serverAddress The address of the server to check
+     *
+     * @return bool True if the server exists in the routing table, false otherwise
+     */
+    public function hasServer(SessionConfiguration $config, string $serverAddress): bool
+    {
+        $table = $this->getRoutingTable($config);
+
+        if ($table === null) {
+            return false;
+        }
+
+        return $table->hasServer($serverAddress);
+    }
+
     /**
      * @throws Exception
      */
@@ -193,6 +293,10 @@ private function getNextServer(RoutingTable $table, ?AccessMode $mode): Uri
             $servers = $table->getWithRole(RoutingRoles::FOLLOWER());
         }
 
+        if (count($servers) === 0) {
+            throw new RuntimeException(sprintf('No servers available for access mode: %s', $mode?->getValue() ?? 'WRITE'));
+        }
+
         return Uri::create($servers[random_int(0, count($servers) - 1)]);
     }
 
@@ -252,6 +356,8 @@ private function createKey(ConnectionRequestData $data, ?SessionConfiguration $c
 
     public function close(): void
     {
+        $this->getLogger()?->log(LogLevel::INFO, 'Closing all connection pools');
+
         foreach (self::$pools as $pool) {
             $pool->close();
         }
diff --git a/src/Neo4j/RoutingTable.php b/src/Neo4j/RoutingTable.php
@@ -60,4 +60,51 @@ public function getWithRole(?RoutingRoles $role = null): array
 
         return array_values(array_unique($tbr));
     }
+
+    /**
+     * Remove a server from the routing table.
+     *
+     * @param string $serverAddress The address to remove
+     *
+     * @return RoutingTable A new routing table with the server removed
+     */
+    public function removeServer(string $serverAddress): RoutingTable
+    {
+        /** @var list<array{addresses: list<string>, role: string}> $updatedServers */
+        $updatedServers = [];
+
+        foreach ($this->servers as $server) {
+            $updatedAddresses = array_filter(
+                $server['addresses'],
+                static fn (string $address): bool => $address !== $serverAddress
+            );
+
+            if (!empty($updatedAddresses)) {
+                $updatedServers[] = [
+                    'addresses' => array_values($updatedAddresses),
+                    'role' => $server['role'],
+                ];
+            }
+        }
+
+        return new self($updatedServers, $this->ttl);
+    }
+
+    /**
+     * Check if a server exists in the routing table.
+     *
+     * @param string $serverAddress The address to check
+     *
+     * @return bool True if the server exists, false otherwise
+     */
+    public function hasServer(string $serverAddress): bool
+    {
+        foreach ($this->servers as $server) {
+            if (in_array($serverAddress, $server['addresses'], true)) {
+                return true;
+            }
+        }
+
+        return false;
+    }
 }
diff --git a/testkit-backend/src/Handlers/ForcedRoutingTableUpdate.php b/testkit-backend/src/Handlers/ForcedRoutingTableUpdate.php
@@ -15,6 +15,7 @@
 
 use Exception;
 use Laudis\Neo4j\Contracts\ConnectionPoolInterface;
+use Laudis\Neo4j\Databags\SessionConfiguration;
 use Laudis\Neo4j\Neo4j\Neo4jConnectionPool;
 use Laudis\Neo4j\Neo4j\Neo4jDriver;
 use Laudis\Neo4j\TestkitBackend\Contracts\RequestHandlerInterface;
@@ -53,11 +54,14 @@ public function handle($request): TestkitResponseInterface
             /** @var ConnectionPoolInterface $pool */
             $pool = $poolProperty->getValue($driver);
 
-            $tableProperty = (new ReflectionClass(Neo4jConnectionPool::class))->getProperty('table');
-            $tableProperty->setAccessible(true);
-            $tableProperty->setValue($pool, null);
+            if ($pool instanceof Neo4jConnectionPool) {
+                // Clear the cached routing table to force a refresh
+                $config = new SessionConfiguration();
+                $pool->clearRoutingTable($config);
+            }
         }
 
+        // Running a query will trigger a new routing table fetch if needed
         $driver->createSession()->run('RETURN 1 AS x');
 
         return new DriverResponse($request->getDriverId());
diff --git a/testkit-backend/src/Handlers/GetRoutingTable.php b/testkit-backend/src/Handlers/GetRoutingTable.php
diff --git a/testkit-backend/testkit.sh b/testkit-backend/testkit.sh

Original file line number	Diff line number	Diff line change
`@@ -29,8 +29,11 @@`
`29`	`29`
`30`	`30`	`class SysVSemaphore implements SemaphoreInterface`
`31`	`31`	`{`
	`32`	`+ /**`
	`33`	`+ * @psalm-suppress UndefinedClass`
	`34`	`+ */`
`32`	`35`	`private function __construct(`
`33`		`- private readonly \SysvSemaphore $semaphore,`
	`36`	`+ private readonly mixed $semaphore,`
`34`	`37`	`) {`
`35`	`38`	`}`
`36`	`39`