Skip to content

Commit 682c863

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 240b957 commit 682c863

7 files changed

Lines changed: 485 additions & 10 deletions

File tree

‎src/Database/Adapter/Postgres.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;
@@ -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
/**
@@ -2764,12 +2764,12 @@ protected function getOperatorSQL(string $column, Operator $operator, int &$bind
27642764
* Bind operator parameters to statement
27652765
* Override to handle PostgreSQL-specific JSON binding
27662766
*
2767-
* @param \PDOStatement|PDOStatementProxy $stmt
2767+
* @param \PDOStatement|PDOStatement $stmt
27682768
* @param Operator $operator
27692769
* @param int &$bindIndex
27702770
* @return void
27712771
*/
2772-
protected function bindOperatorParams(\PDOStatement|PDOStatementProxy $stmt, Operator $operator, int &$bindIndex): void
2772+
protected function bindOperatorParams(\PDOStatement|PDOStatement $stmt, Operator $operator, int &$bindIndex): void
27732773
{
27742774
$method = $operator->getMethod();
27752775
$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
@@ -1978,12 +1978,12 @@ abstract protected function getOperatorSQL(string $column, Operator $operator, i
19781978
/**
19791979
* Bind operator parameters to prepared statement
19801980
*
1981-
* @param \PDOStatement|PDOStatementProxy $stmt
1981+
* @param \PDOStatement|PDOStatement $stmt
19821982
* @param \Utopia\Database\Operator $operator
19831983
* @param int &$bindIndex
19841984
* @return void
19851985
*/
1986-
protected function bindOperatorParams(\PDOStatement|PDOStatementProxy $stmt, Operator $operator, int &$bindIndex): void
1986+
protected function bindOperatorParams(\PDOStatement|PDOStatement $stmt, Operator $operator, int &$bindIndex): void
19871987
{
19881988
$method = $operator->getMethod();
19891989
$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;
@@ -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
/**
@@ -2002,12 +2002,12 @@ private function getSupportForMathFunctions(): bool
20022002
* Bind operator parameters to statement
20032003
* Override to handle SQLite-specific operator bindings
20042004
*
2005-
* @param \PDOStatement|PDOStatementProxy $stmt
2005+
* @param \PDOStatement|PDOStatement $stmt
20062006
* @param Operator $operator
20072007
* @param int &$bindIndex
20082008
* @return void
20092009
*/
2010-
protected function bindOperatorParams(\PDOStatement|PDOStatementProxy $stmt, Operator $operator, int &$bindIndex): void
2010+
protected function bindOperatorParams(\PDOStatement|PDOStatement $stmt, Operator $operator, int &$bindIndex): void
20112011
{
20122012
$method = $operator->getMethod();
20132013

‎src/Database/PDO.php‎

Lines changed: 48 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,52 @@ 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, $options);
50+
}
51+
52+
/**
53+
* Prepare a raw \PDOStatement on the underlying connection, reconnecting
54+
* once if a stale connection surfaces during prepare. Used by
55+
* {@see PDOStatement} to re-prepare after a reconnect without re-wrapping.
56+
*
57+
* Under emulated prepares this never reaches the server (so the loss
58+
* surfaces at execution time instead and is recovered by {@see PDOStatement});
59+
* with native prepares the server is contacted here, so a lost connection
60+
* outside a transaction is reconnected and retried, matching __call().
61+
*
62+
* @param array<mixed> $options
63+
* @throws \Throwable
64+
*/
65+
public function prepareNative(string $query, array $options = []): \PDOStatement
66+
{
67+
try {
68+
$statement = $this->pdo->prepare($query, $options);
69+
} catch (\Throwable $e) {
70+
if (!Connection::hasError($e) || $this->pdo->inTransaction()) {
71+
throw $e;
72+
}
73+
74+
$this->reconnect();
75+
$statement = $this->pdo->prepare($query, $options);
76+
}
77+
78+
if ($statement === false) {
79+
throw new \PDOException("Failed to prepare statement: {$query}");
80+
}
81+
82+
return $statement;
83+
}
84+
3785
/**
3886
* @param string $method
3987
* @param array<mixed> $args

‎src/Database/PDOStatement.php‎

Lines changed: 205 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,205 @@
1+
<?php
2+
3+
namespace Utopia\Database;
4+
5+
use Utopia\Console;
6+
7+
/**
8+
* Wraps a \PDOStatement so a connection lost during execution is recovered
9+
* transparently: the owning PDO reconnects, the statement is re-prepared
10+
* against the fresh connection, previously bound parameters/columns/attributes
11+
* are replayed, and the failed execute() is retried.
12+
*
13+
* Recovery is attempted only for execute(), and only outside a transaction:
14+
* re-running any other method (fetch, rowCount, ...) without a fresh execute
15+
* would return data from an unexecuted statement, and a connection cannot be
16+
* healed in place mid-transaction (the uncommitted state is gone, so the call
17+
* is rethrown for Adapter::withTransaction to roll back and replay).
18+
*
19+
* @mixin \PDOStatement
20+
* @implements \IteratorAggregate<int, mixed>
21+
*/
22+
class PDOStatement implements \IteratorAggregate
23+
{
24+
/**
25+
* @var array<int|string, array{mixed, int}>
26+
*/
27+
private array $values = [];
28+
29+
/**
30+
* @var array<int|string, array{mixed, int, int, mixed}>
31+
*/
32+
private array $params = [];
33+
34+
/**
35+
* @var array<int|string, array{mixed, int, ?int, ?int, mixed}>
36+
*/
37+
private array $columns = [];
38+
39+
/**
40+
* @var array<int, mixed>
41+
*/
42+
private array $attributes = [];
43+
44+
/**
45+
* @var array<int|string, mixed>|null
46+
*/
47+
private ?array $fetchMode = null;
48+
49+
/**
50+
* @param array<mixed> $options
51+
*/
52+
public function __construct(
53+
private readonly PDO $pdo,
54+
private \PDOStatement $statement,
55+
private readonly string $query,
56+
private readonly array $options = [],
57+
) {
58+
}
59+
60+
public function __get(string $name): mixed
61+
{
62+
return $this->statement->{$name};
63+
}
64+
65+
public function __set(string $name, mixed $value): void
66+
{
67+
$this->statement->{$name} = $value;
68+
}
69+
70+
public function __isset(string $name): bool
71+
{
72+
return isset($this->statement->{$name});
73+
}
74+
75+
public function __unset(string $name): void
76+
{
77+
unset($this->statement->{$name});
78+
}
79+
80+
public function __clone(): void
81+
{
82+
throw new \Error('Trying to clone an uncloneable PDOStatement');
83+
}
84+
85+
/**
86+
* Preserve \PDOStatement's native iterability (foreach over rows), which
87+
* does not route through __call().
88+
*/
89+
public function getIterator(): \Traversable
90+
{
91+
return $this->statement;
92+
}
93+
94+
/**
95+
* @param array<mixed> $args
96+
* @throws \Throwable
97+
*/
98+
public function __call(string $method, array $args): mixed
99+
{
100+
try {
101+
return $this->statement->{$method}(...$args);
102+
} catch (\Throwable $e) {
103+
if (
104+
\strcasecmp($method, 'execute') !== 0
105+
|| $this->pdo->inTransaction()
106+
|| !Connection::hasError($e)
107+
) {
108+
throw $e;
109+
}
110+
111+
Console::warning('[Database] ' . $e->getMessage());
112+
Console::warning('[Database] Lost connection detected. Re-preparing statement...');
113+
114+
$this->reprepare();
115+
116+
return $this->statement->{$method}(...$args);
117+
}
118+
}
119+
120+
public function getStatement(): \PDOStatement
121+
{
122+
return $this->statement;
123+
}
124+
125+
public function setAttribute(int $attribute, mixed $value): bool
126+
{
127+
$this->attributes[$attribute] = $value;
128+
129+
return $this->statement->setAttribute($attribute, $value);
130+
}
131+
132+
public function setFetchMode(int $mode, mixed ...$args): bool
133+
{
134+
$this->fetchMode = [$mode, ...$args];
135+
136+
return $this->statement->setFetchMode($mode, ...$args);
137+
}
138+
139+
public function bindValue(int|string $param, mixed $value, int $type = \PDO::PARAM_STR): bool
140+
{
141+
$this->values[$param] = [$value, $type];
142+
143+
return $this->statement->bindValue($param, $value, $type);
144+
}
145+
146+
public function bindParam(int|string $param, mixed &$variable, int $type = \PDO::PARAM_STR, int $maxLength = 0, mixed $driverOptions = null): bool
147+
{
148+
// Store the variable by reference so a value changed between bind and
149+
// execute is the value replayed after a reconnect (PDO binds late).
150+
$this->params[$param] = [&$variable, $type, $maxLength, $driverOptions];
151+
152+
return $this->statement->bindParam($param, $variable, $type, $maxLength, $driverOptions);
153+
}
154+
155+
public function bindColumn(int|string $column, mixed &$variable, ?int $type = null, ?int $maxLength = null, mixed $driverOptions = null): bool
156+
{
157+
// Record how many optional arguments were actually supplied so omitted
158+
// ones keep PDO's real defaults instead of being replayed as explicit
159+
// nulls (which would change the call contract / emit deprecations).
160+
$arity = \func_num_args();
161+
$this->columns[$column] = [&$variable, $arity, $type, $maxLength, $driverOptions];
162+
163+
return $this->bindColumnTo($this->statement, $column, $variable, $arity, $type, $maxLength, $driverOptions);
164+
}
165+
166+
private function reprepare(): void
167+
{
168+
$this->pdo->reconnect();
169+
$this->statement = $this->pdo->prepareNative($this->query, $this->options);
170+
171+
foreach ($this->attributes as $attribute => $value) {
172+
$this->statement->setAttribute($attribute, $value);
173+
}
174+
175+
if ($this->fetchMode !== null) {
176+
$this->statement->setFetchMode(...$this->fetchMode);
177+
}
178+
179+
foreach ($this->params as $param => $bind) {
180+
$this->statement->bindParam($param, $bind[0], $bind[1], $bind[2], $bind[3]);
181+
}
182+
183+
foreach ($this->columns as $column => $bind) {
184+
$this->bindColumnTo($this->statement, $column, $bind[0], $bind[1], $bind[2], $bind[3], $bind[4]);
185+
}
186+
187+
foreach ($this->values as $param => [$value, $type]) {
188+
$this->statement->bindValue($param, $value, $type);
189+
}
190+
}
191+
192+
/**
193+
* Forward bindColumn passing only the optional arguments the caller
194+
* supplied ($arity counts column + variable + supplied options).
195+
*/
196+
private function bindColumnTo(\PDOStatement $statement, int|string $column, mixed &$variable, int $arity, ?int $type = null, ?int $maxLength = null, mixed $driverOptions = null): bool
197+
{
198+
return match (true) {
199+
$arity <= 2 => $statement->bindColumn($column, $variable),
200+
$arity === 3 => $statement->bindColumn($column, $variable, $type ?? \PDO::PARAM_STR),
201+
$arity === 4 => $statement->bindColumn($column, $variable, $type ?? \PDO::PARAM_STR, $maxLength ?? 0),
202+
default => $statement->bindColumn($column, $variable, $type ?? \PDO::PARAM_STR, $maxLength ?? 0, $driverOptions),
203+
};
204+
}
205+
}

0 commit comments

Comments
 (0)