From 820fef1a37ee4c25bde4a7d380c5eb0feee7c00f Mon Sep 17 00:00:00 2001 From: Daniil Gentili Date: Mon, 19 Oct 2026 19:47:25 +0200 Subject: [PATCH] Complete PgSqlHandle operations only once the connection is idle The poll callback completed an operation after its first result, and the caller then fetched the remaining results with the blocking pg_get_result. In between, another fiber could see no pending operation and send its own query on the same handle; pg_get_result then blocked on that query and returned its result as the previous query's next result, so the other query's result was lost and its caller waited forever. This happens with statement deallocation, which runs a DEALLOCATE query in its own fiber while the pool considers the connection idle: a query sent from {main} (whose suspension resumes before queued fiber resumptions) while the DEALLOCATE is in flight hangs. The poll callback now collects all results of an operation, until pg_get_result returns false, before completing it, and the results are built from that list, without blocking calls on the connection. --- src/Internal/PgSqlHandle.php | 82 ++++++++++++++------------- test/PgSqlStatementDeallocateTest.php | 47 +++++++++++++++ 2 files changed, 89 insertions(+), 40 deletions(-) create mode 100644 test/PgSqlStatementDeallocateTest.php diff --git a/src/Internal/PgSqlHandle.php b/src/Internal/PgSqlHandle.php index 022a09a..a859e74 100644 --- a/src/Internal/PgSqlHandle.php +++ b/src/Internal/PgSqlHandle.php @@ -77,9 +77,11 @@ public function __construct( $deferred = &$this->pendingOperation; $listeners = &$this->listeners; $onClose = new DeferredFuture(); + $results = []; $poll = EventLoop::onReadable($socket, static function (string $watcher) use ( &$deferred, + &$results, &$lastUsedAt, &$listeners, &$connection, @@ -118,15 +120,32 @@ public function __construct( return; // No active query, only notification listeners. } - if (\pg_connection_busy($connection)) { - return; - } + // Collect every result of the operation before completing it: the operation only ends once + // the connection is idle again. Otherwise, another operation could be sent in between, and + // fetching the remaining results of this one would block on and consume its results instead. + $operation = $deferred; + while (!\pg_connection_busy($connection)) { + $result = \pg_get_result($connection); + if ($result !== false) { + $results[] = $result; + $status = \pg_result_status($result); + if ($status !== \PGSQL_COPY_IN && $status !== \PGSQL_COPY_OUT) { + continue; + } + // COPY is not supported, the result is an error and closes the connection. + } - $deferred->complete(\pg_get_result($connection)); - $deferred = null; + $operationResults = $results; + $results = []; + + $deferred = null; + $operation->complete($operationResults); + + if (empty($listeners)) { + EventLoop::unreference($watcher); + } - if (empty($listeners)) { - EventLoop::unreference($watcher); + return; } } catch (SqlConnectionException $exception) { $connection = null; // Marks connection as dead. @@ -209,7 +228,7 @@ private function fetchTypes(): Future EventLoop::queue(function () use ($queryDeferred, $typesDeferred): void { try { - $result = $queryDeferred->getFuture()->await(); + [$result] = $queryDeferred->getFuture()->await(); if (\pg_result_status($result) !== \PGSQL_TUPLES_OK) { throw new SqlException(\pg_result_error($result) ?: 'Unknown result error'); } @@ -265,9 +284,11 @@ public function isClosed(): bool * @param \Closure $function Function to execute. * @param mixed ...$args Arguments to pass to function. * + * @return non-empty-list<\PgSql\Result> All results of the operation. + * * @throws SqlException */ - private function send(\Closure $function, mixed ...$args): mixed + private function send(\Closure $function, mixed ...$args): array { $this->types ??= $this->fetchTypes()->await(); @@ -305,18 +326,20 @@ private function send(\Closure $function, mixed ...$args): mixed } /** - * @param \PgSql\Result $result PostgreSQL result resource. + * @param non-empty-list<\PgSql\Result> $results All results of the operation. * @param string $sql Query SQL. * * @throws SqlException * @throws SqlQueryError */ - private function createResult(\PgSql\Result $result, string $sql): PostgresResult + private function createResult(array $results, string $sql): PostgresResult { if ($this->handle === null) { throw new \Error("The connection to the database has been closed"); } + $result = \array_shift($results); + \assert($this->types !== null, 'Expected type array to be populated before creating a result'); switch ($status = \pg_result_status($result)) { @@ -326,11 +349,15 @@ private function createResult(\PgSql\Result $result, string $sql): PostgresResul case \PGSQL_COMMAND_OK: return new PostgresCommandResult( \pg_affected_rows($result), - Future::complete($this->fetchNextResult($sql)), + Future::complete($results ? $this->createResult($results, $sql) : null), ); case \PGSQL_TUPLES_OK: - return new PgSqlResultSet($result, $this->types, Future::complete($this->fetchNextResult($sql))); + return new PgSqlResultSet( + $result, + $this->types, + Future::complete($results ? $this->createResult($results, $sql) : null), + ); case \PGSQL_NONFATAL_ERROR: case \PGSQL_FATAL_ERROR: @@ -338,16 +365,7 @@ private function createResult(\PgSql\Result $result, string $sql): PostgresResul foreach (self::DIAGNOSTIC_CODES as $fieldCode => $description) { $diagnostics[$description] = \pg_result_error_field($result, $fieldCode); } - $message = \pg_result_error($result) ?: 'Unknown result error'; - \set_error_handler(self::getErrorHandler()); - try { - while (\pg_connection_busy($this->handle) && \pg_get_result($this->handle)) { - // Clear all outstanding result rows from the connection - } - } finally { - \restore_error_handler(); - throw new PostgresQueryError($message, $diagnostics, $sql); - } + throw new PostgresQueryError(\pg_result_error($result) ?: 'Unknown result error', $diagnostics, $sql); case \PGSQL_BAD_RESPONSE: $this->close(); @@ -365,22 +383,6 @@ private function createResult(\PgSql\Result $result, string $sql): PostgresResul } } - /** - * @throws SqlException - */ - private function fetchNextResult(string $sql): ?PostgresResult - { - if ($this->handle === null) { - throw new \Error("The connection to the database has been closed"); - } - - if ($result = \pg_get_result($this->handle)) { - return $this->createResult($result, $sql); - } - - return null; - } - #[\Override] public function statementExecute(string $name, array $params): PostgresResult { @@ -482,7 +484,7 @@ public function prepare(string $sql): PostgresStatement } $future = async(function () use ($name, $modifiedSql, $sql): string { - $result = $this->send(\pg_send_prepare(...), $name, $modifiedSql); + [$result] = $this->send(\pg_send_prepare(...), $name, $modifiedSql); switch ($status = \pg_result_status($result)) { case \PGSQL_COMMAND_OK: diff --git a/test/PgSqlStatementDeallocateTest.php b/test/PgSqlStatementDeallocateTest.php new file mode 100644 index 0000000..0a8478a --- /dev/null +++ b/test/PgSqlStatementDeallocateTest.php @@ -0,0 +1,47 @@ +getHandle() instanceof \EvLoop) { + $this->markTestSkipped("ext-pgsql is not compatible with pecl-ev"); + } + + $timeout = EventLoop::delay(5, static fn () => throw new \Error('The result of a query was lost')); + + try { + $connection = PgSqlConnection::connect( + PostgresConfig::fromString('host=localhost user=postgres password=postgres'), + ); + + for ($i = 0; $i < 20; ++$i) { + $statement = $connection->prepare("SELECT $i AS value"); + self::assertSame($i, $statement->execute()->fetchRow()['value']); + unset($statement); // Deallocates the statement in another fiber. + + delay(0); // Lets the DEALLOCATE be sent. + + self::assertSame($i, $connection->query("SELECT $i AS value")->fetchRow()['value']); + } + + $connection->close(); + } finally { + EventLoop::cancel($timeout); + } + } +}