fix(database): drop poisoned pooled connection on any lost-connection error (DAT-1904)

A pooled DB connection is checked out per operation from a worker-shared,
per-host pool. The reconnect wrapper in PDOStatement::__call() only self-healed
a lost connection for execute() outside a transaction. Every other
lost-connection error (fetch, fetchAll, closeCursor, or execute inside a
transaction) was rethrown while the underlying connection was left untouched and
reclaimed to the pool as-is.

A connection lost mid-result is wire-desynced: a partial or stale result frame
can remain buffered on the socket. Returning that connection to the pool poisons
it. The next coroutine to pop it runs its own query and reads the leftover frame
instead, returning a different row (cross-document, and cross-tenant on shared
shards) and corrupting writes (the "no active transaction" class).

The fetch phase must reconnect because it cannot be retried in place: re-running
fetch on a fresh statement would read no rows, so the only safe recovery is to
discard the desynced socket and let the caller replay. This change extends the
recovery so that on any detected lost-connection error outside the
execute-outside-transaction fast path, PDOStatement::__call() calls
PDO::reconnect() to drop the socket before rethrowing. The existing transparent
retry is unchanged. Non-connection errors still rethrow untouched, so a healthy
connection is never thrashed on ordinary query errors. Callers and
Adapter::withTransaction still see the rethrow and roll back or replay on the now
fresh connection.

Adds a regression test proving reconnect() is invoked before the exception
propagates for a fetch-phase lost connection and for execute inside a
transaction, and is not invoked for a non-connection error.

Refs: #895, #896,
appwrite-labs/cloud#4827 (downstream containment),
utopia-php/pools#33 (pool-layer defense in depth)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: abnegate

src/Database/PDOStatement.php

@@ -10,11 +10,19 @@
  * against the fresh connection, previously bound parameters/columns/attributes
  * are replayed, and the failed execute() is retried.
  *
- * Recovery is attempted only for execute(), and only outside a transaction:
- * re-running any other method (fetch, rowCount, ...) without a fresh execute
- * would return data from an unexecuted statement, and a connection cannot be
- * healed in place mid-transaction (the uncommitted state is gone, so the call
- * is rethrown for Adapter::withTransaction to roll back and replay).
+ * Transparent retry is attempted only for execute(), and only outside a
+ * transaction: re-running any other method (fetch, rowCount, ...) without a
+ * fresh execute would return data from an unexecuted statement, and a
+ * connection cannot be healed in place mid-transaction (the uncommitted state
+ * is gone, so the call is rethrown for Adapter::withTransaction to roll back
+ * and replay).
+ *
+ * In every lost-connection case that cannot be retried, the poisoned connection
+ * is still discarded (reconnect) before the rethrow: the failed fetch/execute
+ * leaves the socket wire-desynced, and a pooled connection reclaimed with a
+ * partial result frame still buffered would bleed that frame into the next
+ * borrower's query. Reconnecting guarantees the connection returned to the pool
+ * is always clean.
  *
  * @mixin \PDOStatement
  * @implements \IteratorAggregate<int, mixed>
@@ -108,20 +116,28 @@ public function __call(string $method, array $args): mixed
         try {
             return $this->statement->{$method}(...$args);
         } catch (\Throwable $e) {
-            if (
-                \strcasecmp($method, 'execute') !== 0
-                || $this->pdo->inTransaction()
-                || !Connection::hasError($e)
-            ) {
+            if (!Connection::hasError($e)) {
                 throw $e;
             }
 
-            Console::warning('[Database] ' . $e->getMessage());
-            Console::warning('[Database] Lost connection detected. Re-preparing statement...');
+            if (\strcasecmp($method, 'execute') === 0 && !$this->pdo->inTransaction()) {
+                Console::warning('[Database] ' . $e->getMessage());
+                Console::warning('[Database] Lost connection detected. Re-preparing statement...');
 
-            $this->reprepare();
+                $this->reprepare();
 
-            return $this->statement->{$method}(...$args);
+                return $this->statement->{$method}(...$args);
+            }
+
+            // A lost connection during fetch/fetchAll/closeCursor, or during
+            // execute inside a transaction, cannot be retried in place -- but the
+            // socket is now wire-desynced and may still hold a partial result
+            // frame. Drop the connection so it can never be reclaimed to the pool
+            // and bleed that frame into the next borrower's query; the rethrow
+            // still lets withTransaction roll back and replay on a fresh one.
+            $this->pdo->reconnect();
+
+            throw $e;
         }
     }
 

tests/unit/PDOStatementTest.php

@@ -69,10 +69,12 @@ public function testExecuteReconnectsRePreparesAndReplaysWhenNotInTransaction():
         $this->assertTrue($statement->execute());
     }
 
-    public function testExecuteRethrowsAndDoesNotReconnectInsideTransaction(): void
+    public function testExecuteInsideTransactionReconnectsThenRethrows(): void
     {
         $pdo = $this->pdoMock(inTransaction: true);
-        $pdo->expects($this->never())->method('reconnect');
+        // Cannot retry in place (the uncommitted state is gone), but the poisoned
+        // socket must still be dropped so it is not reclaimed to the pool.
+        $pdo->expects($this->once())->method('reconnect');
         $pdo->expects($this->never())->method('prepareNative');
 
         $statement = $this->statementMock();
@@ -133,23 +135,50 @@ public function testIsIterableAndDelegatesIterationToTheStatement(): void
         $this->assertSame($statement, $wrapper->getIterator());
     }
 
-    public function testDoesNotReconnectForNonExecuteMethods(): void
+    public function testFetchPhaseLostConnectionReconnectsBeforeRethrowing(): void
     {
+        // DAT-1904: a lost connection during the fetch phase leaves the socket
+        // wire-desynced (a partial result frame may remain buffered). The poisoned
+        // connection must be dropped (reconnect) before the rethrow so it can never
+        // be reclaimed to the pool and bled into the next borrower's query.
         $pdo = $this->pdoMock(inTransaction: false);
-        $pdo->expects($this->never())->method('reconnect');
+        $pdo->expects($this->once())->method('reconnect');
+        // Not re-prepared/retried in place: a fresh fetch would read no rows.
         $pdo->expects($this->never())->method('prepareNative');
 
         $statement = $this->statementMock();
         $statement->expects($this->once())
-            ->method('fetch')
+            ->method('fetchAll')
             ->willThrowException(new PDOException('server has gone away'));
 
         $wrapper = new PDOStatement($pdo, $statement, 'SELECT 1');
 
         $this->expectException(PDOException::class);
         $this->expectExceptionMessage('server has gone away');
 
-        $wrapper->fetch();
+        $wrapper->fetchAll();
+    }
+
+    public function testDoesNotReconnectForNonExecuteNonConnectionError(): void
+    {
+        // A business-logic error (not a lost connection) leaves the pooled
+        // connection untouched: reconnecting a healthy connection would thrash the
+        // pool on every ordinary query error.
+        $pdo = $this->pdoMock(inTransaction: false);
+        $pdo->expects($this->never())->method('reconnect');
+        $pdo->expects($this->never())->method('prepareNative');
+
+        $statement = $this->statementMock();
+        $statement->expects($this->once())
+            ->method('fetchAll')
+            ->willThrowException(new PDOException('SQLSTATE[42000]: Syntax error'));
+
+        $wrapper = new PDOStatement($pdo, $statement, 'SELECT 1');
+
+        $this->expectException(PDOException::class);
+        $this->expectExceptionMessage('Syntax error');
+
+        $wrapper->fetchAll();
     }
 
     public function testBindParamReplaysCurrentValueAfterReconnect(): void