Add transaction support via sessions

.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:

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

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" }
       ]
     },
     {

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)

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
 
 		-

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<string, mixed> $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);

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

src/Exception/TransactionException.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\Rango\Exception;
+
+use RuntimeException;
+
+// phpcs:disable SlevomatCodingStandard.Classes.SuperfluousExceptionNaming.SuperfluousSuffix
+final class TransactionException extends RuntimeException implements Exception
+{
+    public static function alreadyInProgress(): self
+    {
+        return new self('A transaction is already in progress for this session');
+    }
+
+    public static function noTransactionStarted(): self
+    {
+        return new self('There is no transaction started for this session');
+    }
+
+    public static function sessionEnded(): self
+    {
+        return new self('The session has already been ended');
+    }
+
+    public static function connectionBusy(): self
+    {
+        return new self('The underlying PDO connection is already running a transaction');
+    }
+}
+// phpcs:enable SlevomatCodingStandard.Classes.SuperfluousExceptionNaming.SuperfluousSuffix

src/Operation/BulkWrite.php

@@ -40,7 +40,11 @@ public function execute(PDO $pdo, QueryBuilder $queryBuilder): BulkWriteResult
         $insertedIds = [];
         $upsertedIds = [];
 
-        $pdo->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;
         }