Skip to content

Commit f2f8a74

Browse files
abnegateclaude
andcommitted
refactor(transaction): own connection recovery in the adapter, drop Swoole PDOProxy dependency
The Swoole PDOProxy keeps its own transaction counter that is incremented on beginTransaction but only decremented on a *successful* commit/rollback, and is never reset on reconnect or on pool checkin (utopia-php/pools does not call its reset() hook). A connection lost mid-transaction therefore leaks the counter and poisons the pooled connection: every later startTransaction trusts the stale counter, rolls back a transaction the real connection no longer holds, and fails with "There is no active transaction". This produced a sustained write outage across all projects on cloud nyc3. Make the library self-sufficient for connection-loss recovery so consumers no longer need to wrap connections in a Swoole PDOProxy: - PDOStatement wraps prepared statements and transparently re-prepares on the reconnected PDO when the connection is lost at execution time, replaying bound params/attributes. Recovery is skipped inside a transaction, where it rethrows so withTransaction can replay the whole transaction from the start. - PDO::prepare() returns the wrapper; prepareNative() re-prepares raw on the reconnected connection. ERRMODE_EXCEPTION is enforced by default. - withTransaction() reconnects on a lost connection before replaying, so the retry runs on a fresh, transaction-less connection. - Transaction state now has a single source of truth (the real PDO via Utopia\Database\PDO::inTransaction()); there is no separate counter to desync. - Replace the Swoole\Database\PDOStatementProxy type hints in the SQL adapters. Stacked on #895. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent d9ed545 commit f2f8a74

9 files changed

Lines changed: 442 additions & 15 deletions

File tree

src/Database/Adapter.php

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -434,6 +434,17 @@ public function withTransaction(callable $callback): mixed
434434
}
435435

436436
if ($attempts < $retries) {
437+
// A lost connection cannot be healed in place mid-transaction;
438+
// reconnect so the retry replays the whole transaction from a
439+
// fresh, transaction-less connection.
440+
if (Connection::hasError($action)) {
441+
try {
442+
$this->reconnect();
443+
} catch (\Throwable) {
444+
// Best effort; the next startTransaction surfaces a fresh failure.
445+
}
446+
}
447+
437448
\usleep($sleep * ($attempts + 1));
438449
continue;
439450
}

src/Database/Adapter/Postgres.php

Lines changed: 15 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
use Exception;
66
use PDO;
77
use PDOException;
8-
use Swoole\Database\PDOStatementProxy;
98
use Utopia\Database\Database;
109
use Utopia\Database\Document;
1110
use Utopia\Database\Exception as DatabaseException;
@@ -18,6 +17,7 @@
1817
use Utopia\Database\Exception\Truncate as TruncateException;
1918
use Utopia\Database\Helpers\ID;
2019
use Utopia\Database\Operator;
20+
use Utopia\Database\PDOStatement;
2121
use Utopia\Database\Query;
2222

2323
/**
@@ -38,11 +38,18 @@ public function startTransaction(): bool
3838
{
3939
try {
4040
if ($this->inTransaction === 0) {
41-
if ($this->getPDO()->inTransaction()) {
42-
$this->getPDO()->rollBack();
43-
} else {
44-
// If no active transaction, this has no effect.
45-
$this->getPDO()->prepare('ROLLBACK')->execute();
41+
try {
42+
if ($this->getPDO()->inTransaction()) {
43+
$this->getPDO()->rollBack();
44+
} else {
45+
// If no active transaction, this has no effect.
46+
$this->getPDO()->prepare('ROLLBACK')->execute();
47+
}
48+
} catch (PDOException) {
49+
// A pooled connection can report a transaction it no longer
50+
// holds after a reconnect (e.g. Swoole PDOProxy keeps its own
51+
// counter), making this cleanup rollback throw. It is best
52+
// effort; swallow it and begin a fresh transaction below.
4653
}
4754

4855
$result = $this->getPDO()->beginTransaction();
@@ -2792,12 +2799,12 @@ protected function getOperatorSQL(string $column, Operator $operator, int &$bind
27922799
* Bind operator parameters to statement
27932800
* Override to handle PostgreSQL-specific JSON binding
27942801
*
2795-
* @param \PDOStatement|PDOStatementProxy $stmt
2802+
* @param \PDOStatement|PDOStatement $stmt
27962803
* @param Operator $operator
27972804
* @param int &$bindIndex
27982805
* @return void
27992806
*/
2800-
protected function bindOperatorParams(\PDOStatement|PDOStatementProxy $stmt, Operator $operator, int &$bindIndex): void
2807+
protected function bindOperatorParams(\PDOStatement|PDOStatement $stmt, Operator $operator, int &$bindIndex): void
28012808
{
28022809
$method = $operator->getMethod();
28032810
$values = $operator->getValues();

src/Database/Adapter/SQL.php

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@
44

55
use Exception;
66
use PDOException;
7-
use Swoole\Database\PDOStatementProxy;
87
use Utopia\Database\Adapter;
98
use Utopia\Database\Change;
109
use Utopia\Database\Database;
@@ -16,6 +15,7 @@
1615
use Utopia\Database\Exception\Timeout as TimeoutException;
1716
use Utopia\Database\Exception\Transaction as TransactionException;
1817
use Utopia\Database\Operator;
18+
use Utopia\Database\PDOStatement;
1919
use Utopia\Database\Query;
2020

2121
abstract class SQL extends Adapter
@@ -1919,12 +1919,12 @@ abstract protected function getOperatorSQL(string $column, Operator $operator, i
19191919
/**
19201920
* Bind operator parameters to prepared statement
19211921
*
1922-
* @param \PDOStatement|PDOStatementProxy $stmt
1922+
* @param \PDOStatement|PDOStatement $stmt
19231923
* @param \Utopia\Database\Operator $operator
19241924
* @param int &$bindIndex
19251925
* @return void
19261926
*/
1927-
protected function bindOperatorParams(\PDOStatement|PDOStatementProxy $stmt, Operator $operator, int &$bindIndex): void
1927+
protected function bindOperatorParams(\PDOStatement|PDOStatement $stmt, Operator $operator, int &$bindIndex): void
19281928
{
19291929
$method = $operator->getMethod();
19301930
$values = $operator->getValues();

src/Database/Adapter/SQLite.php

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
use Exception;
66
use PDO;
77
use PDOException;
8-
use Swoole\Database\PDOStatementProxy;
98
use Utopia\Database\Database;
109
use Utopia\Database\Document;
1110
use Utopia\Database\Exception as DatabaseException;
@@ -17,6 +16,7 @@
1716
use Utopia\Database\Exception\Transaction as TransactionException;
1817
use Utopia\Database\Helpers\ID;
1918
use Utopia\Database\Operator;
19+
use Utopia\Database\PDOStatement;
2020

2121
/**
2222
* Main differences from MariaDB and MySQL:
@@ -1423,12 +1423,12 @@ private function getSupportForMathFunctions(): bool
14231423
* Bind operator parameters to statement
14241424
* Override to handle SQLite-specific operator bindings
14251425
*
1426-
* @param \PDOStatement|PDOStatementProxy $stmt
1426+
* @param \PDOStatement|PDOStatement $stmt
14271427
* @param Operator $operator
14281428
* @param int &$bindIndex
14291429
* @return void
14301430
*/
1431-
protected function bindOperatorParams(\PDOStatement|PDOStatementProxy $stmt, Operator $operator, int &$bindIndex): void
1431+
protected function bindOperatorParams(\PDOStatement|PDOStatement $stmt, Operator $operator, int &$bindIndex): void
14321432
{
14331433
$method = $operator->getMethod();
14341434

src/Database/PDO.php

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@ public function __construct(
2626
protected ?string $password,
2727
protected array $config = []
2828
) {
29+
$this->config[\PDO::ATTR_ERRMODE] ??= \PDO::ERRMODE_EXCEPTION;
30+
2931
$this->pdo = new \PDO(
3032
$this->dsn,
3133
$this->username,
@@ -34,6 +36,47 @@ public function __construct(
3436
);
3537
}
3638

39+
/**
40+
* Prepare a statement, returning a wrapper that transparently re-prepares
41+
* itself on the underlying connection if that connection is lost before the
42+
* statement is executed.
43+
*
44+
* @param array<mixed> $options
45+
* @throws \Throwable
46+
*/
47+
public function prepare(string $query, array $options = []): PDOStatement
48+
{
49+
return new PDOStatement($this, $this->prepareNative($query, $options), $query);
50+
}
51+
52+
/**
53+
* Prepare a raw \PDOStatement on the underlying connection, reconnecting
54+
* once if the connection has been lost. Used by {@see PDOStatement} to
55+
* re-prepare after a reconnect without re-wrapping the result.
56+
*
57+
* @param array<mixed> $options
58+
* @throws \Throwable
59+
*/
60+
public function prepareNative(string $query, array $options = []): \PDOStatement
61+
{
62+
try {
63+
$statement = $this->pdo->prepare($query, $options);
64+
} catch (\Throwable $e) {
65+
if (!Connection::hasError($e) || $this->pdo->inTransaction()) {
66+
throw $e;
67+
}
68+
69+
$this->reconnect();
70+
$statement = $this->pdo->prepare($query, $options);
71+
}
72+
73+
if ($statement === false) {
74+
throw new \PDOException("Failed to prepare statement: {$query}");
75+
}
76+
77+
return $statement;
78+
}
79+
3780
/**
3881
* @param string $method
3982
* @param array<mixed> $args

src/Database/PDOStatement.php

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
<?php
2+
3+
namespace Utopia\Database;
4+
5+
use Utopia\CLI\Console;
6+
7+
/**
8+
* Wraps a \PDOStatement so a connection lost at execution time is recovered
9+
* transparently: the owning PDO reconnects, the statement is re-prepared
10+
* against the fresh connection, previously bound parameters and attributes are
11+
* replayed, and the failed call is retried.
12+
*
13+
* Recovery is skipped while a transaction is open, because the uncommitted
14+
* state lives on the dead connection and cannot survive a reconnect. There the
15+
* call is rethrown so the transaction can be rolled back and replayed from the
16+
* start by the caller.
17+
*
18+
* @mixin \PDOStatement
19+
*/
20+
class PDOStatement
21+
{
22+
/**
23+
* @var array<int|string, array{mixed, int}>
24+
*/
25+
private array $values = [];
26+
27+
/**
28+
* @var array<int|string, array{mixed, int, int, mixed}>
29+
*/
30+
private array $params = [];
31+
32+
/**
33+
* @var array<int|string, array{mixed, ?int, ?int, mixed}>
34+
*/
35+
private array $columns = [];
36+
37+
/**
38+
* @var array<int, mixed>
39+
*/
40+
private array $attributes = [];
41+
42+
/**
43+
* @var array<int|string, mixed>|null
44+
*/
45+
private ?array $fetchMode = null;
46+
47+
public function __construct(
48+
private readonly PDO $pdo,
49+
private \PDOStatement $statement,
50+
private readonly string $query,
51+
) {
52+
}
53+
54+
public function __get(string $name): mixed
55+
{
56+
return $this->statement->{$name};
57+
}
58+
59+
public function __set(string $name, mixed $value): void
60+
{
61+
$this->statement->{$name} = $value;
62+
}
63+
64+
public function __isset(string $name): bool
65+
{
66+
return isset($this->statement->{$name});
67+
}
68+
69+
public function __unset(string $name): void
70+
{
71+
unset($this->statement->{$name});
72+
}
73+
74+
public function __clone(): void
75+
{
76+
throw new \Error('Trying to clone an uncloneable PDOStatement');
77+
}
78+
79+
/**
80+
* @param array<mixed> $args
81+
* @throws \Throwable
82+
*/
83+
public function __call(string $method, array $args): mixed
84+
{
85+
try {
86+
return $this->statement->{$method}(...$args);
87+
} catch (\Throwable $e) {
88+
if ($this->pdo->inTransaction() || !Connection::hasError($e)) {
89+
throw $e;
90+
}
91+
92+
Console::warning('[Database] ' . $e->getMessage());
93+
Console::warning('[Database] Lost connection detected. Re-preparing statement...');
94+
95+
$this->reprepare();
96+
97+
return $this->statement->{$method}(...$args);
98+
}
99+
}
100+
101+
public function getStatement(): \PDOStatement
102+
{
103+
return $this->statement;
104+
}
105+
106+
public function setAttribute(int $attribute, mixed $value): bool
107+
{
108+
$this->attributes[$attribute] = $value;
109+
110+
return $this->statement->setAttribute($attribute, $value);
111+
}
112+
113+
public function setFetchMode(int $mode, mixed ...$args): bool
114+
{
115+
$this->fetchMode = [$mode, ...$args];
116+
117+
return $this->statement->setFetchMode($mode, ...$args);
118+
}
119+
120+
public function bindValue(int|string $param, mixed $value, int $type = \PDO::PARAM_STR): bool
121+
{
122+
$this->values[$param] = [$value, $type];
123+
124+
return $this->statement->bindValue($param, $value, $type);
125+
}
126+
127+
public function bindParam(int|string $param, mixed &$variable, int $type = \PDO::PARAM_STR, int $maxLength = 0, mixed $driverOptions = null): bool
128+
{
129+
$this->params[$param] = [$variable, $type, $maxLength, $driverOptions];
130+
131+
return $this->statement->bindParam($param, $variable, $type, $maxLength, $driverOptions);
132+
}
133+
134+
public function bindColumn(int|string $column, mixed &$variable, ?int $type = null, ?int $maxLength = null, mixed $driverOptions = null): bool
135+
{
136+
$this->columns[$column] = [$variable, $type, $maxLength, $driverOptions];
137+
138+
return $this->statement->bindColumn($column, $variable, $type, $maxLength, $driverOptions);
139+
}
140+
141+
private function reprepare(): void
142+
{
143+
$this->pdo->reconnect();
144+
$this->statement = $this->pdo->prepareNative($this->query);
145+
146+
foreach ($this->attributes as $attribute => $value) {
147+
$this->statement->setAttribute($attribute, $value);
148+
}
149+
150+
if ($this->fetchMode !== null) {
151+
$this->statement->setFetchMode(...$this->fetchMode);
152+
}
153+
154+
foreach ($this->params as $param => [$variable, $type, $maxLength, $driverOptions]) {
155+
$this->statement->bindParam($param, $variable, $type, $maxLength, $driverOptions);
156+
}
157+
158+
foreach ($this->columns as $column => [$variable, $type, $maxLength, $driverOptions]) {
159+
$this->statement->bindColumn($column, $variable, $type, $maxLength, $driverOptions);
160+
}
161+
162+
foreach ($this->values as $param => [$value, $type]) {
163+
$this->statement->bindValue($param, $value, $type);
164+
}
165+
}
166+
}

0 commit comments

Comments
 (0)