Skip to content

Commit 12c3bf1

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 12c3bf1

7 files changed

Lines changed: 536 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: 219 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,219 @@
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+
* The order bindValue()/bindParam() were called, so a placeholder rebound
36+
* across methods replays with the last binding winning, as PDO applies it.
37+
*
38+
* @var array<int, array{string, int|string}>
39+
*/
40+
private array $bindOrder = [];
41+
42+
/**
43+
* @var array<int|string, array{mixed, int, ?int, ?int, mixed}>
44+
*/
45+
private array $columns = [];
46+
47+
/**
48+
* @var array<int, mixed>
49+
*/
50+
private array $attributes = [];
51+
52+
/**
53+
* @var array<int|string, mixed>|null
54+
*/
55+
private ?array $fetchMode = null;
56+
57+
/**
58+
* @param array<mixed> $options
59+
*/
60+
public function __construct(
61+
private readonly PDO $pdo,
62+
private \PDOStatement $statement,
63+
private readonly string $query,
64+
private readonly array $options = [],
65+
) {
66+
}
67+
68+
public function __get(string $name): mixed
69+
{
70+
return $this->statement->{$name};
71+
}
72+
73+
public function __set(string $name, mixed $value): void
74+
{
75+
$this->statement->{$name} = $value;
76+
}
77+
78+
public function __isset(string $name): bool
79+
{
80+
return isset($this->statement->{$name});
81+
}
82+
83+
public function __unset(string $name): void
84+
{
85+
unset($this->statement->{$name});
86+
}
87+
88+
public function __clone(): void
89+
{
90+
throw new \Error('Trying to clone an uncloneable PDOStatement');
91+
}
92+
93+
/**
94+
* Preserve \PDOStatement's native iterability (foreach over rows), which
95+
* does not route through __call().
96+
*/
97+
public function getIterator(): \Traversable
98+
{
99+
return $this->statement;
100+
}
101+
102+
/**
103+
* @param array<mixed> $args
104+
* @throws \Throwable
105+
*/
106+
public function __call(string $method, array $args): mixed
107+
{
108+
try {
109+
return $this->statement->{$method}(...$args);
110+
} catch (\Throwable $e) {
111+
if (
112+
\strcasecmp($method, 'execute') !== 0
113+
|| $this->pdo->inTransaction()
114+
|| !Connection::hasError($e)
115+
) {
116+
throw $e;
117+
}
118+
119+
Console::warning('[Database] ' . $e->getMessage());
120+
Console::warning('[Database] Lost connection detected. Re-preparing statement...');
121+
122+
$this->reprepare();
123+
124+
return $this->statement->{$method}(...$args);
125+
}
126+
}
127+
128+
public function getStatement(): \PDOStatement
129+
{
130+
return $this->statement;
131+
}
132+
133+
public function setAttribute(int $attribute, mixed $value): bool
134+
{
135+
$this->attributes[$attribute] = $value;
136+
137+
return $this->statement->setAttribute($attribute, $value);
138+
}
139+
140+
public function setFetchMode(int $mode, mixed ...$args): bool
141+
{
142+
$this->fetchMode = [$mode, ...$args];
143+
144+
return $this->statement->setFetchMode($mode, ...$args);
145+
}
146+
147+
public function bindValue(int|string $param, mixed $value, int $type = \PDO::PARAM_STR): bool
148+
{
149+
$this->values[$param] = [$value, $type];
150+
$this->bindOrder[] = ['value', $param];
151+
152+
return $this->statement->bindValue($param, $value, $type);
153+
}
154+
155+
public function bindParam(int|string $param, mixed &$variable, int $type = \PDO::PARAM_STR, int $maxLength = 0, mixed $driverOptions = null): bool
156+
{
157+
// Store the variable by reference so a value changed between bind and
158+
// execute is the value replayed after a reconnect (PDO binds late).
159+
$this->params[$param] = [&$variable, $type, $maxLength, $driverOptions];
160+
$this->bindOrder[] = ['param', $param];
161+
162+
return $this->statement->bindParam($param, $variable, $type, $maxLength, $driverOptions);
163+
}
164+
165+
public function bindColumn(int|string $column, mixed &$variable, ?int $type = null, ?int $maxLength = null, mixed $driverOptions = null): bool
166+
{
167+
// Record how many optional arguments were actually supplied so omitted
168+
// ones keep PDO's real defaults instead of being replayed as explicit
169+
// nulls (which would change the call contract / emit deprecations).
170+
$arity = \func_num_args();
171+
$this->columns[$column] = [&$variable, $arity, $type, $maxLength, $driverOptions];
172+
173+
return $this->bindColumnTo($this->statement, $column, $variable, $arity, $type, $maxLength, $driverOptions);
174+
}
175+
176+
private function reprepare(): void
177+
{
178+
$this->pdo->reconnect();
179+
$this->statement = $this->pdo->prepareNative($this->query, $this->options);
180+
181+
foreach ($this->attributes as $attribute => $value) {
182+
$this->statement->setAttribute($attribute, $value);
183+
}
184+
185+
if ($this->fetchMode !== null) {
186+
$this->statement->setFetchMode(...$this->fetchMode);
187+
}
188+
189+
// Replay value/param bindings in the original call order so a placeholder
190+
// rebound across methods ends up with the binding the caller applied last.
191+
foreach ($this->bindOrder as [$kind, $key]) {
192+
if ($kind === 'value') {
193+
[$value, $type] = $this->values[$key];
194+
$this->statement->bindValue($key, $value, $type);
195+
} else {
196+
$bind = $this->params[$key];
197+
$this->statement->bindParam($key, $bind[0], $bind[1], $bind[2], $bind[3]);
198+
}
199+
}
200+
201+
foreach ($this->columns as $column => $bind) {
202+
$this->bindColumnTo($this->statement, $column, $bind[0], $bind[1], $bind[2], $bind[3], $bind[4]);
203+
}
204+
}
205+
206+
/**
207+
* Forward bindColumn passing only the optional arguments the caller
208+
* supplied ($arity counts column + variable + supplied options).
209+
*/
210+
private function bindColumnTo(\PDOStatement $statement, int|string $column, mixed &$variable, int $arity, ?int $type = null, ?int $maxLength = null, mixed $driverOptions = null): bool
211+
{
212+
return match (true) {
213+
$arity <= 2 => $statement->bindColumn($column, $variable),
214+
$arity === 3 => $statement->bindColumn($column, $variable, $type ?? \PDO::PARAM_STR),
215+
$arity === 4 => $statement->bindColumn($column, $variable, $type ?? \PDO::PARAM_STR, $maxLength ?? 0),
216+
default => $statement->bindColumn($column, $variable, $type ?? \PDO::PARAM_STR, $maxLength ?? 0, $driverOptions),
217+
};
218+
}
219+
}

0 commit comments

Comments
 (0)