diff --git a/src/SqlStatementPool.php b/src/SqlStatementPool.php index e3e7baf..faae7c4 100644 --- a/src/SqlStatementPool.php +++ b/src/SqlStatementPool.php @@ -108,12 +108,24 @@ public function execute(array $params = []): SqlResult throw $exception; } - return $this->createResult($result, fn () => $this->push($statement)); + // Drop the closure's statement reference once released: if push() declines to + // retain the statement, a lingering reference here would keep the underlying + // connection checked out for as long as the result object stays alive. + $release = function () use (&$statement): void { + if ($statement !== null) { + $this->push($statement); + $statement = null; + } + }; + + return $this->createResult($result, $release); } /** * Only retains statements if less than 10% of the pool is consumed by this statement and the pool has - * available connections. + * available connections. A statement that is not retained is closed so that its connection returns to + * the pool immediately; otherwise a saturated pool (e.g. a single-connection pool) deadlocks, since + * the next execute() waits for a connection that only this statement's destruction would release. * * @param TStatement $statement */ @@ -122,10 +134,12 @@ protected function push(SqlStatement $statement): void $maxConnections = $this->pool->getConnectionLimit(); if ($this->statements->count() > ($maxConnections / 10)) { + $statement->close(); return; } if ($maxConnections === $this->pool->getConnectionCount() && $this->pool->getIdleConnectionCount() === 0) { + $statement->close(); return; } diff --git a/test/SqlStatementPoolTest.php b/test/SqlStatementPoolTest.php index 9b52256..6914d4f 100644 --- a/test/SqlStatementPoolTest.php +++ b/test/SqlStatementPoolTest.php @@ -4,7 +4,10 @@ use Amp\PHPUnit\AsyncTestCase; use Amp\Sql\Common\SqlStatementPool; +use Amp\Sql\Common\Test\Stub\StubSqlPooledResult; +use Amp\Sql\Common\Test\Stub\StubSqlResult; use Amp\Sql\SqlConnectionPool; +use Amp\Sql\SqlResult; use Amp\Sql\SqlStatement; use function Amp\delay; @@ -81,4 +84,102 @@ public function testIdleStatementsRemovedAfterTimeout() self::assertFalse($statementPool->isClosed()); self::assertSame(\time(), $statementPool->getLastUsedAt()); } + + private function createStatementPool(SqlConnectionPool $pool, \Closure $prepare): SqlStatementPool + { + return new class($pool, 'SELECT 1', $prepare) extends SqlStatementPool { + protected function createResult(SqlResult $result, \Closure $release): SqlResult + { + return new StubSqlPooledResult($result, $release); + } + }; + } + + public function testDeclinedStatementIsClosedSoItsConnectionIsReleased() + { + $pool = $this->createMock(SqlConnectionPool::class); + $pool->method('isClosed') + ->willReturn(false); + $pool->method('getIdleTimeout') + ->willReturn(60); + $pool->method('getConnectionLimit') + ->willReturn(1); + $pool->method('getConnectionCount') + ->willReturn(1); + $pool->method('getIdleConnectionCount') + ->willReturn(0); + + $createStatement = function (): SqlStatement { + $statement = $this->createMock(SqlStatement::class); + $statement->method('isClosed') + ->willReturn(false); + $statement->method('getQuery') + ->willReturn('SELECT 1'); + $statement->method('getLastUsedAt') + ->willReturn(\time()); + $statement->method('execute') + ->willReturn(new StubSqlResult([])); + $statement->expects($this->once()) + ->method('close'); + + return $statement; + }; + + // The pool is saturated (limit 1, all connections busy), so push() declines to + // retain the statement. It must then close the statement and drop its reference, + // otherwise the checked-out connection is never returned and the pool deadlocks + // on the next execute() while the first result is still referenced. + $statementPool = $this->createStatementPool($pool, $this->createCallback(2, $createStatement)); + + $result = $statementPool->execute(); + \iterator_to_array($result); + + delay(0.1); // Allow the queued release to run. + + // The first result remains referenced; a fresh statement must be prepared. + $secondResult = $statementPool->execute(); + \iterator_to_array($secondResult); + + delay(0.1); + } + + public function testStatementIsRetainedWhenThePoolHasCapacity() + { + $pool = $this->createMock(SqlConnectionPool::class); + $pool->method('isClosed') + ->willReturn(false); + $pool->method('getIdleTimeout') + ->willReturn(60); + $pool->method('getConnectionLimit') + ->willReturn(20); + $pool->method('getConnectionCount') + ->willReturn(2); + $pool->method('getIdleConnectionCount') + ->willReturn(1); + + $statement = $this->createMock(SqlStatement::class); + $statement->method('isClosed') + ->willReturn(false); + $statement->method('getQuery') + ->willReturn('SELECT 1'); + $statement->method('getLastUsedAt') + ->willReturn(\time()); + $statement->method('execute') + ->willReturnCallback(static fn () => new StubSqlResult([])); + $statement->expects($this->never()) + ->method('close'); + + $statementPool = $this->createStatementPool($pool, $this->createCallback(1, fn () => $statement)); + + $result = $statementPool->execute(); + \iterator_to_array($result); + + delay(0.1); // Allow the queued release to run. + + // The retained statement is reused instead of preparing a new one. + $secondResult = $statementPool->execute(); + \iterator_to_array($secondResult); + + delay(0.1); + } }