Fix callback accumulation in SqlStatementPool

Fixes amphp/postgres#67.

Author: trowski

src/SqlCommonConnectionPool.php

@@ -51,6 +51,9 @@ abstract class SqlCommonConnectionPool implements SqlConnectionPool
 
     private readonly DeferredFuture $onClose;
 
+    /** @var \WeakMap<TStatement, true> */
+    private \WeakMap $statements;
+
     /**
      * Creates a Statement of the appropriate type using the Statement object returned by the Link object and the
      * given release callable.
@@ -114,6 +117,10 @@ public function __construct(
         }
 
         $this->connections = $connections = new \SplObjectStorage();
+
+        /** @var \WeakMap<TStatement, true> For Psalm. */
+        $this->statements = new \WeakMap();
+
         $this->idle = $idle = new \SplQueue();
         $this->onClose = new DeferredFuture();
 
@@ -207,6 +214,10 @@ public function close(): void
             async(fn () => $connection->close())->ignore();
         }
 
+        foreach ($this->statements as $statement => $_) {
+            $statement->close();
+        }
+
         $this->onClose->complete();
 
         $this->awaitingConnection?->error(new SqlException("Connection pool closed"));
@@ -359,7 +370,12 @@ public function execute(string $sql, array $params = []): SqlResult
     public function prepare(string $sql): SqlStatement
     {
         /** @psalm-suppress InvalidArgument Psalm is not properly detecting the templated return type. */
-        return $this->createStatementPool($sql, $this->prepareStatement(...));
+        $statement = $this->createStatementPool($sql, $this->prepareStatement(...));
+
+        $this->statements[$statement] = true;
+
+        /** @var TStatement $statement Psalm is not properly detecting the templated type. */
+        return $statement;
     }
 
     /**

src/SqlStatementPool.php

@@ -79,8 +79,6 @@ public function __construct(SqlConnectionPool $pool, string $sql, \Closure $prep
 
         EventLoop::unreference($timeoutWatcher);
         $this->onClose(static fn () => EventLoop::cancel($timeoutWatcher));
-
-        $this->pool->onClose(static fn () => $onClose->isComplete() || $onClose->complete());
     }
 
     public function __destruct()