diff --git a/src/Swoole/Database/DatabaseManager.php b/src/Swoole/Database/DatabaseManager.php index ca1de9f..695eb47 100644 --- a/src/Swoole/Database/DatabaseManager.php +++ b/src/Swoole/Database/DatabaseManager.php @@ -41,9 +41,59 @@ public function connection($name = null) Context::set($contextKey, $connection); Context::set("{$contextKey}.pool", $pool); + $this->armCoroutineExitRelease(); + return $connection; } + /** + * Release whatever this coroutine still holds when it ends. + * + * Worker::handle() releases the HTTP request coroutine's connections + * itself (with garbage-collected-first ordering), so this defer is a + * no-op there: the context is already empty when it fires. It exists for + * every OTHER borrower — child coroutines spawned by app code and + * coroutine-based daemons — whose borrows the Worker never sees. Without + * it those connections bypass release(), the pool's counter drifts up + * one slot per borrow until "Connection pool exhausted", and a long-lived + * borrower accumulates leaked server-side prepared statements that + * max_lifetime recycling can never reach. + * + * The defer collects the coroutine's cycle garbage before releasing: + * locals are already destroyed when defers run, so any PDOStatement the + * coroutine still references lives in a cycle, and collecting now closes + * it against a connection that is still idle and unborrowable — the same + * safe-point ordering the Worker uses. Garbage that only becomes + * collectable later stays bounded by the pool's max_lifetime. + * + * If Context::clear() runs mid-coroutine (the Worker does this once per + * request), the armed flag is lost and a later borrow arms a second + * defer. Both run at exit; the second walk finds an empty context and is + * a cheap no-op, so the stacking is bounded and harmless for the request + * path. Daemon loops that clear context repeatedly should release + * explicitly via releaseConnections() instead of relying on this hook. + */ + protected function armCoroutineExitRelease(): void + { + if (Context::get('db.exit_release_armed')) { + return; + } + + Context::set('db.exit_release_armed', true); + + \Swoole\Coroutine::defer(function (): void { + try { + if (gc_status()['roots'] > 0) { + gc_collect_cycles(); + } + + $this->releaseConnections(); + } catch (\Throwable $e) { + error_log('⚠️ Failed to release DB connections at coroutine exit: '.$e->getMessage()); + } + }); + } + protected function getPool($name) { $this->syncApplication(); @@ -136,6 +186,8 @@ public function releaseConnections() return; } + $released = 0; + // Get all context keys and release connections $allContext = Context::all(); foreach ($allContext as $key => $value) { @@ -143,19 +195,26 @@ public function releaseConnections() // Get the connection $connectionKey = substr($key, 0, -strlen('.pool')); $connection = Context::get($connectionKey); - + if ($connection && $value instanceof DatabasePool) { // Release connection back to pool $value->release($connection); + $released++; } - + // Clean up context Context::delete($key); Context::delete($connectionKey); } } - $this->pruneIdleConnections(); + // Pruning drains and refills every pool's channel; skip it when this + // walk found nothing - the request path already released and pruned + // through the Worker, and its exit defer would otherwise pay a full + // no-op prune on every request. + if ($released > 0) { + $this->pruneIdleConnections(); + } } public function pruneIdleConnections(): int diff --git a/src/Swoole/Database/DatabasePool.php b/src/Swoole/Database/DatabasePool.php index 57a421b..5b726ca 100644 --- a/src/Swoole/Database/DatabasePool.php +++ b/src/Swoole/Database/DatabasePool.php @@ -29,6 +29,20 @@ class DatabasePool protected array $connectionConfig; protected array $idleSince = []; protected array $createdAt = []; + protected array $borrowedSince = []; + + /** + * Weak references to every connection this pool created and has not + * closed. A borrower that drops its connection without release() (a + * child coroutine ending, app code deleting the context entry) leaves + * the counter incremented for a connection that no longer exists; + * reconcileVanishedConnections() detects the dead reference and heals + * the slot instead of letting the pool drift toward false exhaustion. + * + * @var array> + */ + protected array $liveConnections = []; + protected ?int $idlePruneTimerId = null; /** @@ -209,6 +223,8 @@ public function release($connection): void */ public function pruneIdleConnections(?float $now = null): int { + $this->reconcileVanishedConnections(); + $maxIdleTime = (float) ($this->config['max_idle_time'] ?? 60.0); $minConnections = (int) ($this->config['min_connections'] ?? 1); @@ -340,12 +356,61 @@ protected function resetConnection($connection): void } /** - * Safely close a connection + * Start tracking a freshly created connection. + * + * PHP reuses object ids. If the previous occupant of this id is a + * tracked connection that died while factory->make() was connecting + * (allocation-triggered gc can run inside make, and the freed slot is + * handed to the next allocation), overwriting its weak reference would + * strand that slot's counter increment forever. Heal it on overwrite. + */ + protected function trackConnection(object $connection): void + { + $id = spl_object_id($connection); + + if (isset($this->liveConnections[$id]) && $this->liveConnections[$id]->get() === null) { + unset($this->liveConnections[$id], $this->idleSince[$id], $this->borrowedSince[$id], $this->createdAt[$id]); + $this->currentConnections--; + error_log('⚠️ DB pool healed a connection slot dropped without release (object id reused)'); + } + + $this->createdAt[$id] = microtime(true); + $this->liveConnections[$id] = \WeakReference::create($connection); + } + + /** + * Decrement the counter for connections that were garbage-collected + * without passing through release()/closeConnection(). + */ + public function reconcileVanishedConnections(): int + { + $healed = 0; + + foreach ($this->liveConnections as $id => $ref) { + if ($ref->get() !== null) { + continue; + } + + unset($this->liveConnections[$id], $this->idleSince[$id], $this->createdAt[$id], $this->borrowedSince[$id]); + $this->currentConnections--; + $healed++; + } + + if ($healed > 0) { + error_log("⚠️ DB pool healed {$healed} connection slot(s) dropped without release"); + } + + return $healed; + } + + /** + * Safely close a connection and drop its tracking state. */ protected function closeConnection($connection): void { if (is_object($connection)) { - unset($this->createdAt[spl_object_id($connection)]); + $id = spl_object_id($connection); + unset($this->createdAt[$id], $this->liveConnections[$id], $this->borrowedSince[$id]); } try { @@ -410,13 +475,15 @@ protected function hasOutlivedMaxLifetime($connection, ?float $now = null): bool */ protected function createConnection() { + $this->reconcileVanishedConnections(); + $this->currentConnections++; try { $connection = $this->factory->make($this->connectionConfig, $this->name); if (is_object($connection)) { - $this->createdAt[spl_object_id($connection)] = microtime(true); + $this->trackConnection($connection); } if ($connection instanceof Connection) { @@ -447,14 +514,18 @@ protected function createConnection() protected function markIdle($connection): void { if (is_object($connection)) { - $this->idleSince[spl_object_id($connection)] = microtime(true); + $id = spl_object_id($connection); + $this->idleSince[$id] = microtime(true); + unset($this->borrowedSince[$id]); } } protected function markBorrowed($connection): void { if (is_object($connection)) { - unset($this->idleSince[spl_object_id($connection)]); + $id = spl_object_id($connection); + unset($this->idleSince[$id]); + $this->borrowedSince[$id] = microtime(true); } } @@ -514,6 +585,10 @@ public function flush(): void */ public function getStats(): array { + // Heal first so a quiet pool does not report vanished borrowers as + // live or long-borrowed connections. + $this->reconcileVanishedConnections(); + return [ 'current_connections' => $this->currentConnections, 'available_connections' => $this->channel->length(), @@ -522,9 +597,34 @@ public function getStats(): array 'min_connections' => $this->config['min_connections'] ?? 1, 'max_idle_time' => $this->config['max_idle_time'] ?? 60.0, 'max_lifetime' => $this->config['max_lifetime'] ?? self::DEFAULT_MAX_LIFETIME, + 'tracked_connections' => count($this->liveConnections), + 'long_borrowed_connections' => $this->longBorrowedCount(), ]; } + /** + * Connections held by a borrower for longer than max_lifetime. The pool + * cannot recycle these (recycling happens at release), so a non-zero + * value points at a coroutine or daemon that never releases — the + * borrower should end its coroutine or call + * DatabaseManager::releaseConnections() periodically. + */ + protected function longBorrowedCount(?float $now = null): int + { + $maxLifetime = (float) ($this->config['max_lifetime'] ?? self::DEFAULT_MAX_LIFETIME); + + if ($maxLifetime <= 0) { + return 0; + } + + $now ??= microtime(true); + + return count(array_filter( + $this->borrowedSince, + static fn (float $since): bool => ($now - $since) >= $maxLifetime + )); + } + protected function startIdlePruner(): void { $heartbeat = (float) ($this->config['heartbeat'] ?? -1); diff --git a/tests/Feature/DaemonConnectionReleaseTest.php b/tests/Feature/DaemonConnectionReleaseTest.php new file mode 100644 index 0000000..5966ce2 --- /dev/null +++ b/tests/Feature/DaemonConnectionReleaseTest.php @@ -0,0 +1,195 @@ +markTestSkipped('Swoole coroutine support is required.'); + } + + $events = new ArrayObject(); + + \Swoole\Coroutine\run(function () use ($events) { + $done = new Channel(1); + + Coroutine::create(function () use ($events, $done) { + app('db')->connection()->select('select 1'); // arms the exit hook + + Context::set('db.connection.probe', new \stdClass()); + Context::set('db.connection.probe.pool', new ReleaseOrderRecordingPool($events)); + + $probe = new ReleaseOrderDestructProbe($events); + $probe->self = $probe; + unset($probe); + + $done->push(true); + }); + + $done->pop(); + Coroutine::sleep(0.05); + }); + + $this->assertSame( + ['destruct', 'release'], + iterator_to_array($events), + 'Cycle garbage must die (destruct) before the exit hook releases the connection (release).' + ); + } + + /** + * The exit hook's release walk runs on every DB-using request coroutine + * after the Worker has already emptied the context; it must not pay a + * full all-pool prune for that no-op. + * + * @requires extension swoole + */ + public function test_release_walk_skips_pruning_when_nothing_was_held(): void + { + if (! class_exists(Coroutine::class) || ! function_exists('Swoole\\Coroutine\\run')) { + $this->markTestSkipped('Swoole coroutine support is required.'); + } + + $events = new ArrayObject(); + $pool = new ReleaseOrderRecordingPool($events); + + $manager = app('db'); + $this->assertInstanceOf(DatabaseManager::class, $manager); + + $pools = new \ReflectionProperty(DatabaseManager::class, 'pools'); + $originalPools = $pools->getValue($manager); + $pools->setValue($manager, ['recording' => $pool]); + + try { + \Swoole\Coroutine\run(function () use ($manager, $events, $pool) { + $manager->releaseConnections(); + $events['after_empty'] = $pool->pruneCalls; + + Context::set('db.connection.probe', new \stdClass()); + Context::set('db.connection.probe.pool', $pool); + $manager->releaseConnections(); + $events['after_held'] = $pool->pruneCalls; + }); + } finally { + $pools->setValue($manager, $originalPools); + } + + $this->assertSame(0, $events['after_empty'], 'An empty context walk must not prune every pool.'); + $this->assertGreaterThan(0, $events['after_held'], 'A walk that released something must still prune.'); + } + + /** + * Connections borrowed by coroutines the Worker never tears down (child + * coroutines spawned by app code, coroutine-based daemons) must return + * to the pool when their coroutine ends. Without that they bypass + * release() entirely: the pool counter drifts one slot per borrow until + * false "Connection pool exhausted", and long-lived borrowers accumulate + * leaked server-side prepared statements that max_lifetime recycling can + * never reach. + * + * @requires extension swoole + */ + public function test_connection_borrowed_by_plain_coroutine_returns_to_pool_at_exit(): void + { + if (! class_exists(Coroutine::class) || ! function_exists('Swoole\\Coroutine\\run')) { + $this->markTestSkipped('Swoole coroutine support is required.'); + } + + $stats = null; + + \Swoole\Coroutine\run(function () use (&$stats) { + $done = new Channel(1); + + Coroutine::create(function () use ($done) { + $connection = app('db')->connection(); + $connection->select('select 1'); + $done->push(true); + // Ends without releasing - the exit hook must return the + // borrow to the pool. + }); + + $done->pop(); + Coroutine::sleep(0.05); + + $stats = app('db')->poolStats(); + }); + + $this->assertNotEmpty($stats); + $pool = array_values($stats)[0]; + + $this->assertSame(1, $pool['current_connections'], 'The borrow must survive as a pooled connection, not vanish.'); + $this->assertSame( + 1, + $pool['available_connections'], + 'The connection must be RETURNED to the pool at coroutine exit - an empty channel means the borrow leaked.' + ); + } + + /** + * Many short-lived borrowing coroutines must not drift the pool counter + * toward exhaustion. Before the exit hook existed, each iteration leaked + * one slot and the pool died after max_connections coroutines. + * + * @requires extension swoole + */ + public function test_many_borrowing_coroutines_do_not_exhaust_the_pool(): void + { + if (! class_exists(Coroutine::class) || ! function_exists('Swoole\\Coroutine\\run')) { + $this->markTestSkipped('Swoole coroutine support is required.'); + } + + $failures = 0; + $stats = null; + + \Swoole\Coroutine\run(function () use (&$failures, &$stats) { + // Default pool max_connections is 10; 30 sequential borrowing + // coroutines exhaust a drifting pool three times over. + for ($i = 0; $i < 30; $i++) { + $done = new Channel(1); + + Coroutine::create(function () use ($done) { + try { + app('db')->connection()->select('select 1'); + $done->push(true); + } catch (\Throwable $e) { + $done->push(false); + } + }); + + if ($done->pop() !== true) { + $failures++; + } + + Coroutine::sleep(0.001); + } + + $stats = app('db')->poolStats(); + }); + + $this->assertSame(0, $failures, 'Every borrowing coroutine must get a connection - drift means later ones starve.'); + + $pool = array_values($stats)[0]; + $this->assertLessThanOrEqual( + 2, + $pool['current_connections'], + 'Connections must be reused across coroutines, not leaked one per coroutine.' + ); + } +} diff --git a/tests/Feature/WorkerPooledConnectionReleaseOrderTest.php b/tests/Feature/WorkerPooledConnectionReleaseOrderTest.php index f33cf0c..233db14 100644 --- a/tests/Feature/WorkerPooledConnectionReleaseOrderTest.php +++ b/tests/Feature/WorkerPooledConnectionReleaseOrderTest.php @@ -124,6 +124,8 @@ public function __destruct() class ReleaseOrderRecordingPool extends DatabasePool { + public int $pruneCalls = 0; + public function __construct(protected ArrayObject $events) { } @@ -135,6 +137,8 @@ public function release($connection): void public function pruneIdleConnections(?float $now = null): int { + $this->pruneCalls++; + return 0; } } diff --git a/tests/Unit/DatabasePoolTest.php b/tests/Unit/DatabasePoolTest.php index 099996d..df6f120 100644 --- a/tests/Unit/DatabasePoolTest.php +++ b/tests/Unit/DatabasePoolTest.php @@ -608,6 +608,109 @@ public function test_prune_closes_over_age_connections_even_at_min_connections() $this->assertSame(0, $stats['current_connections']); } + public function test_reconcile_heals_slots_for_connections_dropped_without_release(): void + { + $this->skipIfNoSwooleCoroutine(); + + if (! extension_loaded('pdo_sqlite')) { + $this->markTestSkipped('PDO SQLite is required.'); + } + + $factory = Mockery::mock(ConnectionFactory::class); + $factory->shouldReceive('make') + ->once() + ->withAnyArgs() + ->andReturnUsing(fn () => new Connection(new PDO('sqlite::memory:'), 'database', '', [])); + + $healed = null; + $stats = null; + + \Swoole\Coroutine\run(function () use ($factory, &$healed, &$stats) { + $pool = new DatabasePool([ + 'min_connections' => 0, + 'max_connections' => 2, + 'wait_timeout' => 0.1, + 'max_lifetime' => 60.0, + ], [], 'sqlite', $factory); + + $connection = $pool->get(); + $this->assertSame(1, $pool->getStats()['current_connections']); + + // A borrower that vanishes without release() - e.g. a child + // coroutine whose context died with the connection inside it. + unset($connection); + gc_collect_cycles(); + + $healed = $pool->reconcileVanishedConnections(); + $stats = $pool->getStats(); + }); + + $this->assertSame(1, $healed, 'The vanished connection must be detected via its dead weak reference.'); + $this->assertSame(0, $stats['current_connections'], 'The slot must be reclaimed so the pool cannot drift into false exhaustion.'); + $this->assertSame(0, $stats['tracked_connections']); + } + + public function test_reconcile_leaves_live_connections_alone(): void + { + $this->skipIfNoSwooleCoroutine(); + + if (! extension_loaded('pdo_sqlite')) { + $this->markTestSkipped('PDO SQLite is required.'); + } + + $factory = Mockery::mock(ConnectionFactory::class); + $factory->shouldReceive('make') + ->once() + ->withAnyArgs() + ->andReturnUsing(fn () => new Connection(new PDO('sqlite::memory:'), 'database', '', [])); + + \Swoole\Coroutine\run(function () use ($factory) { + $pool = new DatabasePool([ + 'min_connections' => 0, + 'max_connections' => 2, + 'wait_timeout' => 0.1, + 'max_lifetime' => 60.0, + ], [], 'sqlite', $factory); + + $connection = $pool->get(); + gc_collect_cycles(); + + $this->assertSame(0, $pool->reconcileVanishedConnections(), 'A held connection must never be treated as vanished.'); + $this->assertSame(1, $pool->getStats()['current_connections']); + + $pool->release($connection); + $this->assertSame(1, $pool->getStats()['current_connections']); + }); + } + + public function test_tracking_a_new_connection_heals_a_reused_object_id(): void + { + $pool = $this->newPoolWithoutConstructor(); + + $abandoned = new \stdClass(); + $reusedId = spl_object_id($abandoned); + + $live = new \ReflectionProperty(DatabasePool::class, 'liveConnections'); + $live->setValue($pool, [$reusedId => \WeakReference::create($abandoned)]); + $this->setPoolCurrentConnections($pool, 1); + + // The abandoned connection dies while factory->make() is connecting; + // PHP hands its object id to the next same-shape allocation. + unset($abandoned); + $fresh = new \stdClass(); + + if (spl_object_id($fresh) !== $reusedId) { + $this->markTestSkipped('Allocator did not reuse the object id on this build.'); + } + + $track = new ReflectionMethod(DatabasePool::class, 'trackConnection'); + $track->invoke($pool, $fresh); + + $current = new \ReflectionProperty(DatabasePool::class, 'currentConnections'); + $this->assertSame(0, $current->getValue($pool), 'Overwriting a dead weak reference must heal the abandoned slot, or the drift becomes permanently unhealable.'); + $this->assertSame($fresh, $live->getValue($pool)[$reusedId]->get()); + } + protected function newPoolWithoutConstructor(): DatabasePool { $reflection = new \ReflectionClass(DatabasePool::class);