From e59abddd8ada205cd769b36c3dd9a37f630d9ec1 Mon Sep 17 00:00:00 2001 From: David Badura Date: Sat, 29 Aug 2026 11:06:37 +0200 Subject: [PATCH] Add transaction support via sessions --- .github/workflows/integration.yml | 16 +- README.md | 1 + docs/project.json | 3 +- docs/transactions.md | 75 ++++++++++ phpstan-baseline.neon | 2 +- src/Client.php | 10 ++ src/Exception/QueryException.php | 28 +++- src/Exception/TransactionException.php | 32 ++++ src/Operation/BulkWrite.php | 14 +- src/Session.php | 194 +++++++++++++++++++++++++ src/SqlRunner.php | 4 +- tests/IntegrationTest.php | 91 ++++++++++++ tests/PostgresIntegrationTest.php | 77 ++++++++++ 13 files changed, 529 insertions(+), 18 deletions(-) create mode 100644 docs/transactions.md create mode 100644 src/Exception/TransactionException.php create mode 100644 src/Session.php diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 67a5a08..70bda0f 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -71,14 +71,6 @@ jobs: runs-on: ${{ matrix.operating-system }} - services: - mongodb: - image: "mongo:${{ matrix.mongodb-version }}" - options: >- - --health-cmd "mongosh --eval 'db.runCommand({ ping: 1 })' --quiet" - ports: - - "27017:27017" - strategy: matrix: dependencies: @@ -93,12 +85,18 @@ jobs: - "8.0" env: - MONGODB_URI: 'mongodb://localhost:27017' + MONGODB_URI: 'mongodb://localhost:27017/?replicaSet=rs0' steps: - name: "Checkout" uses: actions/checkout@v6 + - name: "Start MongoDB" + uses: supercharge/mongodb-github-action@1.12.0 + with: + mongodb-version: "${{ matrix.mongodb-version }}" + mongodb-replica-set: rs0 + - name: "Install PHP" uses: "shivammathur/setup-php@2.36.0" with: diff --git a/README.md b/README.md index f45826e..8121635 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ ecosystem without giving up the flexible document-based development experience o * [Projection and sorting](https://patchlevel.dev/docs/rango/latest/querying#projection) with dot-notation support * [Aggregation pipelines](https://patchlevel.dev/docs/rango/latest/aggregation) with `$match`, `$group`, `$unwind`, and `$lookup` * [Bulk writes](https://patchlevel.dev/docs/rango/latest/crud-operations#bulk-writes) wrapped in a single transaction +* [Transactions](https://patchlevel.dev/docs/rango/latest/transactions) via `startSession` and `withTransaction` * [Index management](https://patchlevel.dev/docs/rango/latest/indexes) backed by native PostgreSQL indexes ## Installation diff --git a/docs/project.json b/docs/project.json index 76efcf8..3c99e66 100644 --- a/docs/project.json +++ b/docs/project.json @@ -15,7 +15,8 @@ "title": "Advanced", "subEntries": [ { "title": "Aggregation", "file": "aggregation.md" }, - { "title": "Indexes", "file": "indexes.md" } + { "title": "Indexes", "file": "indexes.md" }, + { "title": "Transactions", "file": "transactions.md" } ] }, { diff --git a/docs/transactions.md b/docs/transactions.md new file mode 100644 index 0000000..d236f46 --- /dev/null +++ b/docs/transactions.md @@ -0,0 +1,75 @@ +# Transactions + +A session groups several operations into a single transaction: either every write is applied, or none of them are. The API mirrors `mongodb/mongodb`, so code written against `startSession`, `startTransaction`, and `commitTransaction` keeps working unchanged. + +Because a [client](connection.md) owns exactly one PostgreSQL connection, a transaction wraps a plain `BEGIN` / `COMMIT` / `ROLLBACK` on that connection. + +## Running a transaction + +Start a session on the client, open a transaction, and commit it once every operation succeeded: + +```php +$session = $client->startSession(); +$session->startTransaction(); + +try { + $accounts->updateOne(['_id' => 'alice'], ['$inc' => ['balance' => -100]], ['session' => $session]); + $accounts->updateOne(['_id' => 'bob'], ['$inc' => ['balance' => 100]], ['session' => $session]); + + $session->commitTransaction(); +} catch (Throwable $e) { + $session->abortTransaction(); + + throw $e; +} finally { + $session->endSession(); +} +``` +`abortTransaction` rolls everything back. `endSession` releases the session and rolls back a transaction that is still open; the underlying connection stays open and can start a new session. + +## The callback style + +`withTransaction` takes care of commit and rollback for you. It commits when the callback returns and rolls back when it throws, then re-throws the exception: + +```php +$session = $client->startSession(); + +$session->withTransaction(static function ($session) use ($accounts): void { + $accounts->updateOne(['_id' => 'alice'], ['$inc' => ['balance' => -100]], ['session' => $session]); + $accounts->updateOne(['_id' => 'bob'], ['$inc' => ['balance' => 100]], ['session' => $session]); +}); +``` +If PostgreSQL rejects the transaction with a serialization failure or a deadlock, `withTransaction` retries the callback a few times before giving up. Make sure the callback has no side effects outside the database, since it may run more than once. + +## The session option + +Every collection method accepts a `session` option, just like the MongoDB driver. Passing it keeps your code portable. Rango runs all operations on the one connection the client holds, so once a transaction is open on that connection, every following operation takes part in it whether or not you pass the option. + +:::note +Only pass a session that came from the same client. A session from another client points at a different connection and its transaction would not cover the operation. +::: + +## Isolation level + +By default a transaction runs at PostgreSQL's `READ COMMITTED` level. Pass a `readConcern` to raise it: + +```php +$session->startTransaction(['readConcern' => 'snapshot']); +``` +`snapshot` maps to `REPEATABLE READ` and `linearizable` maps to `SERIALIZABLE`. A `SERIALIZABLE` transaction is the case that can fail with a serialization error under concurrency, which is why `withTransaction` retries. + +## Bulk writes + +`bulkWrite` already runs in its own transaction. Inside a session transaction it joins the surrounding one instead of opening a nested transaction, so a failing bulk write aborts the whole transaction. + +## Limitations + +* A session runs one transaction at a time. Calling `startTransaction` again before committing or aborting throws. +* Read the results of `find` and `aggregate` inside the transaction. A cursor that is still open when you commit may not see a consistent snapshot afterwards. +* Sessions are not causally consistent across connections the way MongoDB sessions are; there is only ever the single client connection. + +## Learn more + +* [How a client maps to a PostgreSQL connection](connection.md) +* [How CRUD operations translate to SQL](how-it-works.md) +* [Bulk writes](crud-operations.md) diff --git a/phpstan-baseline.neon b/phpstan-baseline.neon index 97dd980..785c7fc 100644 --- a/phpstan-baseline.neon +++ b/phpstan-baseline.neon @@ -597,7 +597,7 @@ parameters: - message: '#^Cannot access offset ''name'' on array\|object\|null\.$#' identifier: offsetAccess.nonOffsetAccessible - count: 9 + count: 10 path: tests/IntegrationTest.php - diff --git a/src/Client.php b/src/Client.php index ab07677..5af4738 100644 --- a/src/Client.php +++ b/src/Client.php @@ -30,6 +30,16 @@ public function __construct(string|PDO $uri) $this->queryBuilder = new QueryBuilder($this->pdo); } + /** + * Start a session that can group operations into a single transaction. + * + * @param array $options reserved for MongoDB compatibility, currently unused + */ + public function startSession(array $options = []): Session + { + return new Session($this->pdo); + } + public function getDatabase(string $name): Database { return new Database($this, $name); diff --git a/src/Exception/QueryException.php b/src/Exception/QueryException.php index a7482ab..c2160d8 100644 --- a/src/Exception/QueryException.php +++ b/src/Exception/QueryException.php @@ -4,21 +4,45 @@ namespace Patchlevel\Rango\Exception; +use PDOException; use RuntimeException; use Throwable; +use function is_string; use function sprintf; // phpcs:disable SlevomatCodingStandard.Classes.SuperfluousExceptionNaming.SuperfluousSuffix final class QueryException extends RuntimeException implements Exception { - public function __construct(string $query, string $error, int $code = 0, Throwable|null $previous = null) - { + public function __construct( + string $query, + string $error, + int $code = 0, + Throwable|null $previous = null, + private readonly string|null $sqlState = null, + ) { parent::__construct( sprintf("Query failed: %s\nError: %s", $query, $error), $code, $previous, ); } + + public static function fromPdo(string $query, PDOException $e): self + { + $sqlState = $e->errorInfo[0] ?? null; + if (!is_string($sqlState) || $sqlState === '') { + $code = $e->getCode(); + $sqlState = is_string($code) && $code !== '' ? $code : null; + } + + return new self($query, $e->getMessage(), (int)$e->getCode(), $e, $sqlState); + } + + /** The five-character SQLSTATE returned by PostgreSQL, if available. */ + public function sqlState(): string|null + { + return $this->sqlState; + } } // phpcs:enable SlevomatCodingStandard.Classes.SuperfluousExceptionNaming.SuperfluousSuffix diff --git a/src/Exception/TransactionException.php b/src/Exception/TransactionException.php new file mode 100644 index 0000000..507dd2c --- /dev/null +++ b/src/Exception/TransactionException.php @@ -0,0 +1,32 @@ +beginTransaction(); + $ownsTransaction = !$pdo->inTransaction(); + + if ($ownsTransaction) { + $pdo->beginTransaction(); + } try { foreach ($this->operations as $operation) { @@ -86,9 +90,13 @@ public function execute(PDO $pdo, QueryBuilder $queryBuilder): BulkWriteResult } } - $pdo->commit(); + if ($ownsTransaction) { + $pdo->commit(); + } } catch (Throwable $e) { - $pdo->rollBack(); + if ($ownsTransaction && $pdo->inTransaction()) { + $pdo->rollBack(); + } throw $e; } diff --git a/src/Session.php b/src/Session.php new file mode 100644 index 0000000..bba68ae --- /dev/null +++ b/src/Session.php @@ -0,0 +1,194 @@ + $options */ + public function startTransaction(array $options = []): void + { + if ($this->ended) { + throw TransactionException::sessionEnded(); + } + + if ($this->transactionState === self::STATE_IN_PROGRESS) { + throw TransactionException::alreadyInProgress(); + } + + if ($this->pdo->inTransaction()) { + throw TransactionException::connectionBusy(); + } + + $this->pdo->beginTransaction(); + + $isolationLevel = self::isolationLevelFor($options); + if ($isolationLevel !== null) { + SqlRunner::exec($this->pdo, 'SET TRANSACTION ISOLATION LEVEL ' . $isolationLevel); + } + + $this->transactionState = self::STATE_IN_PROGRESS; + } + + public function commitTransaction(): void + { + $this->assertInProgress(); + + try { + $this->pdo->commit(); + } catch (PDOException $e) { + // A failed COMMIT ends the transaction on the server side. PostgreSQL + // reports serialization failures here, so surface it like any query. + $this->transactionState = self::STATE_ABORTED; + + throw QueryException::fromPdo('COMMIT', $e); + } + + $this->transactionState = self::STATE_COMMITTED; + } + + public function abortTransaction(): void + { + $this->assertInProgress(); + + $this->rollBack(); + } + + /** + * Run the callback inside a transaction, committing on success and rolling + * back on failure. Transient errors (serialization failures, deadlocks) are + * retried a few times before the exception is re-thrown. + * + * @param callable(self): void $callback + * @param array $options + */ + public function withTransaction(callable $callback, array $options = []): void + { + $attempt = 0; + + while (true) { + $attempt++; + + $this->startTransaction($options); + + try { + $callback($this); + $this->commitTransaction(); + + return; + } catch (Throwable $e) { + if ($this->pdo->inTransaction()) { + $this->rollBack(); + } else { + $this->transactionState = self::STATE_ABORTED; + } + + if (self::isRetryable($e) && $attempt <= self::MAX_RETRIES) { + continue; + } + + throw $e; + } + } + } + + public function isInTransaction(): bool + { + return $this->transactionState === self::STATE_IN_PROGRESS; + } + + /** + * End the session. A still-open transaction is rolled back. The underlying + * PDO connection stays open and can be reused for a new session. + */ + public function endSession(): void + { + if ($this->transactionState === self::STATE_IN_PROGRESS) { + $this->rollBack(); + } + + $this->ended = true; + } + + /** @internal Used by the client to make sure a passed session belongs to it. */ + public function pdo(): PDO + { + return $this->pdo; + } + + private function rollBack(): void + { + $this->pdo->rollBack(); + $this->transactionState = self::STATE_ABORTED; + } + + private function assertInProgress(): void + { + if ($this->transactionState !== self::STATE_IN_PROGRESS) { + throw TransactionException::noTransactionStarted(); + } + } + + private static function isRetryable(Throwable $e): bool + { + return $e instanceof QueryException + && in_array($e->sqlState(), self::RETRYABLE_SQL_STATES, true); + } + + /** @param array $options */ + private static function isolationLevelFor(array $options): string|null + { + $readConcern = $options['readConcern'] ?? null; + + if (is_array($readConcern)) { + $readConcern = $readConcern['level'] ?? null; + } + + if (!is_string($readConcern)) { + return null; + } + + return match ($readConcern) { + 'snapshot' => 'REPEATABLE READ', + 'linearizable' => 'SERIALIZABLE', + default => null, + }; + } +} diff --git a/src/SqlRunner.php b/src/SqlRunner.php index 7325be5..902d0b1 100644 --- a/src/SqlRunner.php +++ b/src/SqlRunner.php @@ -20,7 +20,7 @@ public static function exec(PDO $pdo, string $sql): int try { $rowCount = $pdo->exec($sql); } catch (PDOException $e) { - throw new QueryException($sql, $e->getMessage(), (int)$e->getCode(), $e); + throw QueryException::fromPdo($sql, $e); } if ($rowCount === false) { @@ -35,7 +35,7 @@ public static function query(PDO $pdo, string $sql): PDOStatement try { $statement = $pdo->query($sql); } catch (PDOException $e) { - throw new QueryException($sql, $e->getMessage(), (int)$e->getCode(), $e); + throw QueryException::fromPdo($sql, $e); } if ($statement === false) { diff --git a/tests/IntegrationTest.php b/tests/IntegrationTest.php index 83b8a2a..0bd80d3 100644 --- a/tests/IntegrationTest.php +++ b/tests/IntegrationTest.php @@ -28,6 +28,8 @@ abstract class IntegrationTest extends TestCase { protected MongoDbCollection|RangoCollection $collection; + protected MongoDbClient|RangoClient $client; + abstract protected function getClient(): MongoDbClient|RangoClient; abstract protected function getCollection(): MongoDbCollection|RangoCollection; @@ -38,6 +40,7 @@ protected function setUp(): void { parent::setUp(); + $this->client = $this->getClient(); $this->collection = $this->getCollection(); $this->collection->drop(); } @@ -1342,4 +1345,92 @@ public function testSelectAliases(): void self::assertEquals('alias', $doc['name']); } + + /** + * MongoDB only allows transactions on a replica set or sharded cluster, so + * the parity run against a standalone server skips these cases. + */ + private function requireTransactionSupport(): void + { + $probe = $this->client->selectCollection('test', 'tx_probe'); + + try { + $session = $this->client->startSession(); + $session->startTransaction(); + $probe->insertOne(['_id' => 'probe'], ['session' => $session]); + $session->abortTransaction(); + $session->endSession(); + } catch (RuntimeException $e) { + self::markTestSkipped('backend does not support transactions: ' . $e->getMessage()); + } + } + + public function testTransactionCommitPersistsEveryWrite(): void + { + $this->requireTransactionSupport(); + + $items = $this->client->selectCollection('test', 'tx_commit'); + $items->drop(); + + $session = $this->client->startSession(); + $session->startTransaction(); + + $items->insertOne(['_id' => '1', 'name' => 'foo'], ['session' => $session]); + $items->insertOne(['_id' => '2', 'name' => 'bar'], ['session' => $session]); + + $session->commitTransaction(); + $session->endSession(); + + self::assertEquals(2, $items->countDocuments()); + } + + public function testTransactionAbortDiscardsEveryWrite(): void + { + $this->requireTransactionSupport(); + + $items = $this->client->selectCollection('test', 'tx_abort'); + $items->drop(); + + $session = $this->client->startSession(); + $session->startTransaction(); + + $items->insertOne(['_id' => '1', 'name' => 'foo'], ['session' => $session]); + $items->insertOne(['_id' => '2', 'name' => 'bar'], ['session' => $session]); + + $session->abortTransaction(); + $session->endSession(); + + self::assertEquals(0, $items->countDocuments()); + } + + public function testTransactionRollsBackWhenAnOperationFails(): void + { + $this->requireTransactionSupport(); + + $items = $this->client->selectCollection('test', 'tx_error'); + $items->drop(); + $items->insertOne(['_id' => 'existing', 'name' => 'old']); + + $session = $this->client->startSession(); + $session->startTransaction(); + + $failed = false; + + try { + $items->insertOne(['_id' => 'fresh', 'name' => 'new'], ['session' => $session]); + $items->insertOne(['_id' => 'existing', 'name' => 'dupe'], ['session' => $session]); + } catch (RuntimeException) { + $failed = true; + } + + if ($session->isInTransaction()) { + $session->abortTransaction(); + } + + $session->endSession(); + + self::assertTrue($failed, 'the duplicate insert should have raised an error'); + self::assertEquals(1, $items->countDocuments()); + self::assertEquals('old', $items->findOne(['_id' => 'existing'])['name']); + } } diff --git a/tests/PostgresIntegrationTest.php b/tests/PostgresIntegrationTest.php index e695c9b..79b47a3 100644 --- a/tests/PostgresIntegrationTest.php +++ b/tests/PostgresIntegrationTest.php @@ -7,6 +7,10 @@ use Patchlevel\Rango\Client; use Patchlevel\Rango\Collection; use Patchlevel\Rango\Database; +use Patchlevel\Rango\Exception\QueryException; +use Patchlevel\Rango\Exception\TransactionException; +use Patchlevel\Rango\Session; +use Patchlevel\Rango\SqlRunner; use RuntimeException; use function getenv; @@ -47,4 +51,77 @@ protected function getDatabase(): Database { return $this->getClient()->getDatabase('test'); } + + public function testWithTransactionCommitsOnSuccess(): void + { + $client = $this->getClient(); + $items = $client->getDatabase('test')->getCollection('with_tx_ok'); + $items->drop(); + + $session = $client->startSession(); + $session->withTransaction(static function (Session $session) use ($items): void { + $items->insertOne(['_id' => '1'], ['session' => $session]); + $items->insertOne(['_id' => '2'], ['session' => $session]); + }); + + self::assertFalse($session->isInTransaction()); + self::assertSame(2, $items->countDocuments()); + } + + public function testWithTransactionRollsBackAndRethrowsOnError(): void + { + $client = $this->getClient(); + $items = $client->getDatabase('test')->getCollection('with_tx_err'); + $items->drop(); + $items->insertOne(['_id' => 'x']); + + $session = $client->startSession(); + + try { + $session->withTransaction(static function (Session $session) use ($items): void { + $items->insertOne(['_id' => 'y'], ['session' => $session]); + $items->insertOne(['_id' => 'x'], ['session' => $session]); + }); + + self::fail('expected the duplicate insert to bubble up'); + } catch (QueryException) { + // expected + } + + self::assertFalse($session->isInTransaction()); + self::assertSame(1, $items->countDocuments()); + } + + public function testStartingASecondTransactionThrows(): void + { + $session = $this->getClient()->startSession(); + $session->startTransaction(); + + try { + $this->expectException(TransactionException::class); + $session->startTransaction(); + } finally { + $session->abortTransaction(); + } + } + + public function testAbortWithoutTransactionThrows(): void + { + $session = $this->getClient()->startSession(); + + $this->expectException(TransactionException::class); + $session->abortTransaction(); + } + + public function testReadConcernRaisesTheIsolationLevel(): void + { + $session = $this->getClient()->startSession(); + $session->startTransaction(['readConcern' => 'snapshot']); + + $level = SqlRunner::query($session->pdo(), 'SHOW transaction_isolation')->fetchColumn(); + + $session->abortTransaction(); + + self::assertSame('repeatable read', $level); + } }