From 1997ba2e12f28747a961aff96c239bb643747ffc Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 9 Sep 2026 11:41:21 +0000 Subject: [PATCH 1/2] Fix object cleanup when pool maintenance overlaps closure A lifetime sweep or idle trim can resume while close is suspended in another destroy callback. The closed channel still contains objects for draining, so maintenance can remove a healthy object and lose it when the channel rejects its requeue. Destroy rejected requeues through the existing cleanup path. Cover both maintenance operations with a controlled overlap, asserting that the healthy object is destroyed before close resumes and every object is cleaned up exactly once. --- src/object-pool/src/ObjectPool.php | 7 +- tests/ObjectPool/ObjectPoolTest.php | 103 ++++++++++++++++++++++++++++ 2 files changed, 108 insertions(+), 2 deletions(-) diff --git a/src/object-pool/src/ObjectPool.php b/src/object-pool/src/ObjectPool.php index 9c5ecfa6b..020460802 100644 --- a/src/object-pool/src/ObjectPool.php +++ b/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); + } } /** diff --git a/tests/ObjectPool/ObjectPoolTest.php b/tests/ObjectPool/ObjectPoolTest.php index f4bd38b44..a940fcafa 100644 --- a/tests/ObjectPool/ObjectPoolTest.php +++ b/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]); From 50de10ed90398ecf7c309dfcf7a3fcfc7ff3ca30 Mon Sep 17 00:00:00 2001 From: Raj Siva-Rajah <5361908+binaryfire@users.noreply.github.com> Date: Wed, 9 Sep 2026 11:41:33 +0000 Subject: [PATCH 2/2] Fail keepalive calls immediately when reconnection does not succeed Concurrent reconnect cleanup can finish after the winning socket has closed. Check the false reconnect result before attempting a socket wait, and raise ConnectionException immediately instead of waiting until timeout and reporting socket exhaustion. Return the actual connected state after heartbeat activation and previous-channel wakeup, including closure during timer creation. Document the reconnect result and keepalive exception behavior, and cover concurrent cleanup and timer-creation closure without adding retries or shared state. --- .../src/KeepaliveConnection.php | 7 +- .../src/ConnectionPool/Connection.php | 2 + src/docs/pools.md | 2 + .../KeepaliveConnectionTest.php | 108 +++++++++++++++++- 4 files changed, 114 insertions(+), 5 deletions(-) diff --git a/src/connection-pool/src/KeepaliveConnection.php b/src/connection-pool/src/KeepaliveConnection.php index bdd670b52..f54519aa1 100644 --- a/src/connection-pool/src/KeepaliveConnection.php +++ b/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; diff --git a/src/contracts/src/ConnectionPool/Connection.php b/src/contracts/src/ConnectionPool/Connection.php index ccc40cc7c..022cfa265 100644 --- a/src/contracts/src/ConnectionPool/Connection.php +++ b/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; diff --git a/src/docs/pools.md b/src/docs/pools.md index 4bf785d7c..150528593 100644 --- a/src/docs/pools.md +++ b/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. diff --git a/tests/ConnectionPool/KeepaliveConnectionTest.php b/tests/ConnectionPool/KeepaliveConnectionTest.php index c15450283..9225cc27b 100644 --- a/tests/ConnectionPool/KeepaliveConnectionTest.php +++ b/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]);