Fix pool cleanup and keepalive reconnection races

src/connection-pool/src/KeepaliveConnection.php

@@ -5,6 +5,7 @@
 namespace Hypervel\ConnectionPool;
 
 use Closure;
+use Hypervel\ConnectionPool\Exceptions\ConnectionException;
 use Hypervel\ConnectionPool\Exceptions\InvalidArgumentException;
 use Hypervel\ConnectionPool\Exceptions\SocketPopException;
 use Hypervel\Contracts\ConnectionPool\Connection as ConnectionContract;
@@ -126,7 +127,7 @@ public function reconnect(): bool
             $previousChannel?->close();
         }
 
-        return true;
+        return $this->isConnected();
     }
 
     /**
@@ -136,8 +137,8 @@ public function reconnect(): bool
      */
     public function call(Closure $closure, bool $refresh = true): mixed
     {
-        if (! $this->isConnected()) {
-            $this->reconnect();
+        if (! $this->isConnected() && ! $this->reconnect()) {
+            throw new ConnectionException(sprintf('Socket of %s could not be reconnected.', $this->name));
         }
 
         $channel = $this->channel;

src/contracts/src/ConnectionPool/Connection.php

@@ -13,6 +13,8 @@ public function getConnection(): mixed;
 
     /**
      * Reconnect the connection.
+     *
+     * Return true if the connection is available for use when reconnection completes.
      */
     public function reconnect(): bool;
 

src/docs/pools.md

@@ -374,6 +374,8 @@ The connection class is responsible for translating connection options into the
 
 If a protocol needs to keep one socket alive with a periodic heartbeat, you may extend `KeepaliveConnection`. This connection type exposes a `call()` method for working with its socket and does not allow direct `getConnection()` access. Your subclass should create the socket through `getActiveConnection()` and may override `heartbeat()` and `sendClose()` for the protocol.
 
+If reconnection completes without an available socket, `call()` immediately throws `Hypervel\ConnectionPool\Exceptions\ConnectionException`. A failed socket wait throws `Hypervel\ConnectionPool\Exceptions\SocketPopException`; `wait_timeout` limits how long the call waits.
+
 If a connection is closed or replaced while a call is running, the call's socket is dropped when it finishes instead of being returned for reuse. This cleanup does not send a protocol close message. Return a socket resource or client whose release or destructor closes the underlying connection.
 
 Connection pools are worker-lifetime services. A package should keep them in a manager that returns the current pool for each operation instead of retaining a borrowed connection or a pool that has been removed.

src/object-pool/src/ObjectPool.php

@@ -293,13 +293,16 @@ protected function ensureBorrowed(object $object): int
     }
 
     /**
-     * Return an object to the idle channel without recording user activity.
+     * Return an object to the idle channel without recording activity, or destroy it when the pool has closed.
      *
      * @param T $object
      */
     protected function requeue(object $object): void
     {
-        $this->channel->push($object);
+        // Maintenance may resume while a concurrent close is still draining the channel.
+        if (! $this->channel->push($object)) {
+            $this->destroyObject($object);
+        }
     }
 
     /**

tests/ConnectionPool/KeepaliveConnectionTest.php

@@ -4,6 +4,7 @@
 
 namespace Hypervel\Tests\ConnectionPool;
 
+use Hypervel\ConnectionPool\Exceptions\ConnectionException;
 use Hypervel\ConnectionPool\Exceptions\SocketPopException;
 use Hypervel\Container\Container;
 use Hypervel\Context\CoroutineContext;
@@ -468,6 +469,83 @@ public static function cleanupFailures(): array
         ];
     }
 
+    public function testCallFailsWithoutWaitingWhenTheWinnerClosesDuringLosingReconnectCleanup(): void
+    {
+        $pool = new HeartbeatPoolStub(new Container, 'test', [
+            'heartbeat_interval' => 3600.0,
+            'wait_timeout' => 0.05,
+        ]);
+        $connection = $pool->borrow();
+        $firstReady = new Channel(1);
+        $loserClosing = new Channel(1);
+        $finishLoserClose = new Channel(1);
+        $completed = new Channel(1);
+        $created = [];
+        $closed = [];
+        $called = false;
+        $connection->createCallback = function () use (&$created, $firstReady): object {
+            $socket = new stdClass;
+            $created[] = $socket;
+
+            if (count($created) === 1) {
+                $firstReady->pop(1.0);
+            }
+
+            return $socket;
+        };
+        $connection->closeCallback = function (object $socket) use (
+            &$created,
+            &$closed,
+            $loserClosing,
+            $finishLoserClose,
+        ): void {
+            $closed[] = $socket;
+
+            if ($socket === $created[0]) {
+                $loserClosing->push(true);
+                $finishLoserClose->pop(1.0);
+            }
+        };
+        $coroutineId = Coroutine::create(function () use ($connection, &$called, $completed): void {
+            try {
+                $connection->call(function () use (&$called): void {
+                    $called = true;
+                });
+                $completed->push(true);
+            } catch (Throwable $exception) {
+                $completed->push($exception);
+            }
+        });
+
+        try {
+            $winner = $connection->call(static fn ($socket) => $socket);
+            $this->assertSame($created[1], $winner);
+            $firstReady->push(true);
+            $this->assertTrue($loserClosing->pop(1.0));
+
+            $connection->close();
+            $finishLoserClose->push(true);
+
+            $this->assertFalse(Coroutine::exists($coroutineId));
+            $failure = $completed->pop(1.0);
+            $this->assertInstanceOf(ConnectionException::class, $failure);
+            $this->assertSame('Socket of keepalive.connection could not be reconnected.', $failure->getMessage());
+            $this->assertFalse($called);
+            $this->assertFalse($connection->isConnected());
+            $this->assertCount(2, $created);
+            $this->assertSame($created, $closed);
+            $this->assertSame(2, $connection->closeCount);
+            $this->assertNull((new ClassInvoker($connection))->timerId);
+            $this->assertSame([], (new ClassInvoker($connection->timer))->coroutines);
+        } finally {
+            $firstReady->close();
+            $finishLoserClose->close();
+            Coroutine::join([$coroutineId], 1.0);
+            $connection->discard();
+            $connection->timer->clearAll();
+        }
+    }
+
     #[DataProvider('heartbeatModes')]
     public function testCanceledCloseClearsStateAndReconnectWakesOldWaiters(?float $heartbeatInterval): void
     {
@@ -585,12 +663,14 @@ public function testLateCloseDoesNotClearAReplacement(): void
         }
     }
 
-    public function testCloseDuringTimerCreationDoesNotRetainTheTimer(): void
+    #[DataProvider('reconnectOperations')]
+    public function testCloseDuringTimerCreationFailsReconnectionWithoutRetainingTheTimer(bool $throughCall): void
     {
         $pool = new HeartbeatPoolStub(new Container, 'test', ['heartbeat_interval' => 3600.0]);
         $connection = $pool->borrow();
         $connection->setActiveConnection(new stdClass);
         $closed = false;
+        $called = false;
         Coroutine::afterCreated(function () use ($connection, &$closed): void {
             if (! $closed) {
                 $closed = true;
@@ -599,10 +679,23 @@ public function testCloseDuringTimerCreationDoesNotRetainTheTimer(): void
         });
 
         try {
-            $connection->reconnect();
+            if ($throughCall) {
+                try {
+                    $connection->call(function () use (&$called): void {
+                        $called = true;
+                    });
+                    $this->fail('Expected reconnection to fail after timer creation closed the socket.');
+                } catch (ConnectionException $exception) {
+                    $this->assertSame('Socket of keepalive.connection could not be reconnected.', $exception->getMessage());
+                }
+            } else {
+                $this->assertFalse($connection->reconnect());
+            }
 
             $this->assertTrue($closed);
+            $this->assertFalse($called);
             $this->assertFalse($connection->isConnected());
+            $this->assertSame(1, $connection->closeCount);
             $this->assertNull((new ClassInvoker($connection))->timerId);
             $this->assertSame([], (new ClassInvoker($connection->timer))->coroutines);
         } finally {
@@ -611,6 +704,17 @@ public function testCloseDuringTimerCreationDoesNotRetainTheTimer(): void
         }
     }
 
+    /**
+     * Provide direct and caller-initiated reconnection.
+     */
+    public static function reconnectOperations(): array
+    {
+        return [
+            'direct reconnect' => [false],
+            'reconnect before call' => [true],
+        ];
+    }
+
     public function testReconnectPublishesIfAnEarlierWinnerHasAlreadyClosed(): void
     {
         $pool = new HeartbeatPoolStub(new Container, 'test', ['heartbeat_interval' => 3600.0]);

tests/ObjectPool/ObjectPoolTest.php

@@ -9,15 +9,18 @@
 use Hypervel\Contracts\Debug\ExceptionHandler;
 use Hypervel\Coroutine\Coroutine;
 use Hypervel\Coroutine\PoolChannel;
+use Hypervel\Engine\Channel;
 use Hypervel\ObjectPool\Exceptions\PoolClosedException;
 use Hypervel\ObjectPool\Exceptions\PoolExhaustedException;
 use Hypervel\ObjectPool\ObjectPool;
 use Hypervel\ObjectPool\PoolOptions;
 use Hypervel\Tests\TestCase;
 use Mockery as m;
+use PHPUnit\Framework\Attributes\DataProvider;
 use RuntimeException;
 use stdClass;
 use Swoole\Coroutine\CanceledException;
+use Throwable;
 
 use function Hypervel\Coroutine\parallel;
 
@@ -525,6 +528,106 @@ public function testMaintenanceRequeuesPreserveReleaseTimestamps(): void
         $pool->close();
     }
 
+    #[DataProvider('maintenanceMethods')]
+    public function testMaintenanceDestroysHealthyObjectsWhileCloseIsStillDraining(string $method): void
+    {
+        $maintenanceStarted = new Channel(1);
+        $closeStarted = new Channel(1);
+        $resumeMaintenance = new Channel(1);
+        $resumeClose = new Channel(1);
+        $maintenanceDone = new Channel(1);
+        $closeDone = new Channel(1);
+        $objects = [];
+        $destroyed = [];
+        $pool = $this->pool(
+            ['max_objects' => 3, 'min_retained_objects' => 1, 'max_lifetime' => 60, 'max_idle_time' => 60],
+            destroyCallback: function (object $object) use (
+                &$objects,
+                &$destroyed,
+                $maintenanceStarted,
+                $closeStarted,
+                $resumeMaintenance,
+                $resumeClose,
+            ): void {
+                $destroyed[] = $object;
+
+                if ($object === $objects[0]) {
+                    $maintenanceStarted->push(true);
+                    $resumeMaintenance->pop(1.0);
+                } elseif ($object === $objects[1]) {
+                    $closeStarted->push(true);
+                    $resumeClose->pop(1.0);
+                }
+            },
+        );
+        $objects = [$pool->borrow(), $pool->borrow(), $pool->borrow()];
+
+        foreach ($objects as $object) {
+            $pool->release($object);
+        }
+
+        if ($method === 'sweepExpired') {
+            $pool->ageCreation($objects[0], 120.0);
+        } else {
+            $pool->ageRelease($objects[0], 120.0);
+        }
+
+        $coroutineIds = [];
+
+        try {
+            $coroutineIds[] = Coroutine::create(function () use ($pool, $method, $maintenanceDone): void {
+                try {
+                    $pool->{$method}();
+                    $maintenanceDone->push(true);
+                } catch (Throwable $exception) {
+                    $maintenanceDone->push($exception);
+                }
+            });
+            $this->assertTrue($maintenanceStarted->pop(1.0));
+
+            $coroutineIds[] = Coroutine::create(function () use ($pool, $closeDone): void {
+                try {
+                    $pool->close();
+                    $closeDone->push(true);
+                } catch (Throwable $exception) {
+                    $closeDone->push($exception);
+                }
+            });
+            $this->assertTrue($closeStarted->pop(1.0));
+            $this->assertTrue($pool->isClosed());
+            $this->assertTrue($closeDone->isEmpty());
+
+            $resumeMaintenance->push(true);
+            $this->assertTrue($maintenanceDone->pop(1.0));
+            $this->assertTrue($closeDone->isEmpty());
+            $this->assertSame($objects, $destroyed);
+            $this->assertSame(1, $pool->getManagedCount());
+
+            $resumeClose->push(true);
+            $this->assertTrue($closeDone->pop(1.0));
+            $this->assertSame([
+                'managed' => 0, 'borrowed' => 0, 'idle' => 0, 'waiting' => 0, 'closed' => true,
+            ], $pool->getStats());
+            $this->assertSame($objects, $destroyed);
+        } finally {
+            $resumeMaintenance->close();
+            $resumeClose->close();
+            Coroutine::join($coroutineIds, 1.0);
+            $pool->close();
+        }
+    }
+
+    /**
+     * Provide maintenance operations that requeue healthy objects.
+     */
+    public static function maintenanceMethods(): array
+    {
+        return [
+            'lifetime sweep' => ['sweepExpired'],
+            'idle trim' => ['trimIdle'],
+        ];
+    }
+
     public function testPoolIdleTimeoutRequiresNoBorrowedOrInFlightObjects(): void
     {
         $pool = $this->pool(['pool_idle_timeout' => 0.001]);