Release the connection when the statement pool declines to retain a statement

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;
         }
 

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);
+    }
 }