Add logic for transactional commits (#2589)

* Extract commit logic

* Support transactional commit operations

* Always use transactional flush if supported

With this commit, all tests using the document manager use transactional flush as long as transactions are supported. Certain tests can use the static $allowsTransactions variable to disable this behaviour.

* Test with MongoDB 7.0

* Update test names

* Update phpstan baseline

* Fix query selection in shard key tests

* Flip transaction options constant by default

* Use supportsTransaction method when skipping tests

* Apply review feedback to tests

* Add separate test to check write concern in commit options

* Strip write options when in transaction

* Use majority write concern in test

.github/workflows/continuous-integration.yml

@@ -22,6 +22,7 @@ jobs:
           - "8.2"
           - "8.3"
         mongodb-version:
+          - "7.0"
           - "6.0"
           - "5.0"
           - "4.4"
@@ -34,24 +35,34 @@ jobs:
         symfony-version:
           - "stable"
         include:
+          # Test against lowest dependencies
           - dependencies: "lowest"
             php-version: "8.1"
             mongodb-version: "4.4"
             driver-version: "1.11.0"
             topology: "server"
             symfony-version: "stable"
-          - topology: "sharded_cluster"
+          # Test with highest dependencies
+          - topology: "server"
+            php-version: "8.2"
+            mongodb-version: "6.0"
+            driver-version: "stable"
+            dependencies: "highest"
+            symfony-version: "7"
+          # Test with a 4.4 replica set
+          - topology: "replica_set"
             php-version: "8.2"
             mongodb-version: "4.4"
             driver-version: "stable"
             dependencies: "highest"
             symfony-version: "stable"
-          - topology: "server"
+          # Test with a 4.4 sharded cluster
+          - topology: "sharded_cluster"
             php-version: "8.2"
-            mongodb-version: "6.0"
+            mongodb-version: "4.4"
             driver-version: "stable"
             dependencies: "highest"
-            symfony-version: "7"
+            symfony-version: "stable"
 
     steps:
       - name: "Checkout"

lib/Doctrine/ODM/MongoDB/Persisters/DocumentPersister.php

@@ -35,6 +35,7 @@
 use MongoDB\Driver\CursorInterface;
 use MongoDB\Driver\Exception\Exception as DriverException;
 use MongoDB\Driver\Exception\WriteException;
+use MongoDB\Driver\Session;
 use MongoDB\Driver\WriteConcern;
 use MongoDB\GridFS\Bucket;
 use stdClass;
@@ -1580,7 +1581,23 @@ private function getWriteOptions(array $options = []): array
             unset($writeOptions['w']);
         }
 
-        return $writeOptions;
+        return $this->isInTransaction($options)
+            ? $this->uow->stripTransactionOptions($writeOptions)
+            : $writeOptions;
+    }
+
+    private function isInTransaction(array $options): bool
+    {
+        if (! isset($options['session'])) {
+            return false;
+        }
+
+        $session = $options['session'];
+        if (! $session instanceof Session) {
+            return false;
+        }
+
+        return $session->isInTransaction();
     }
 
     /**

lib/Doctrine/ODM/MongoDB/UnitOfWork.php

@@ -25,20 +25,25 @@
 use Doctrine\Persistence\PropertyChangedListener;
 use InvalidArgumentException;
 use MongoDB\BSON\UTCDateTime;
+use MongoDB\Driver\Session;
 use MongoDB\Driver\WriteConcern;
 use ProxyManager\Proxy\GhostObjectInterface;
 use ReflectionProperty;
 use UnexpectedValueException;
 
+use function array_diff_key;
 use function array_filter;
+use function array_intersect_key;
 use function array_key_exists;
+use function array_merge;
 use function assert;
 use function count;
 use function get_class;
 use function in_array;
 use function is_array;
 use function is_object;
 use function method_exists;
+use function MongoDB\with_transaction;
 use function preg_match;
 use function serialize;
 use function spl_object_hash;
@@ -61,6 +66,7 @@
  *      fsync?: bool,
  *      safe?: int,
  *      w?: int,
+ *      withTransaction?: bool,
  *      writeConcern?: WriteConcern
  * }
  */
@@ -92,6 +98,12 @@ final class UnitOfWork implements PropertyChangedListener
 
     /** @internal */
     public const DEPRECATED_WRITE_OPTIONS = ['fsync', 'safe', 'w'];
+    private const TRANSACTION_OPTIONS     = [
+        'maxCommitTimeMS' => 1,
+        'readConcern' => 1,
+        'readPreference' => 1,
+        'writeConcern' => 1,
+    ];
 
     /**
      * The identity map holds references to all managed documents.
@@ -441,27 +453,18 @@ public function commit(array $options = []): void
                 }
             }
 
-            // Raise onFlush
             $this->evm->dispatchEvent(Events::onFlush, new Event\OnFlushEventArgs($this->dm));
 
-            foreach ($this->getClassesForCommitAction($this->scheduledDocumentUpserts) as $classAndDocuments) {
-                [$class, $documents] = $classAndDocuments;
-                $this->executeUpserts($class, $documents, $options);
-            }
-
-            foreach ($this->getClassesForCommitAction($this->scheduledDocumentInsertions) as $classAndDocuments) {
-                [$class, $documents] = $classAndDocuments;
-                $this->executeInserts($class, $documents, $options);
-            }
-
-            foreach ($this->getClassesForCommitAction($this->scheduledDocumentUpdates) as $classAndDocuments) {
-                [$class, $documents] = $classAndDocuments;
-                $this->executeUpdates($class, $documents, $options);
-            }
-
-            foreach ($this->getClassesForCommitAction($this->scheduledDocumentDeletions, true) as $classAndDocuments) {
-                [$class, $documents] = $classAndDocuments;
-                $this->executeDeletions($class, $documents, $options);
+            if ($this->useTransaction($options)) {
+                with_transaction(
+                    $this->dm->getClient()->startSession(),
+                    function (Session $session) use ($options): void {
+                        $this->doCommit(['session' => $session] + $this->stripTransactionOptions($options));
+                    },
+                    $this->getTransactionOptions($options),
+                );
+            } else {
+                $this->doCommit($options);
             }
 
             // Raise postFlush
@@ -3110,8 +3113,63 @@ public function isUninitializedObject(object $obj): bool
         };
     }
 
+    /** @internal */
+    public function stripTransactionOptions(array $options): array
+    {
+        return array_diff_key(
+            $options,
+            self::TRANSACTION_OPTIONS,
+        );
+    }
+
     private function objToStr(object $obj): string
     {
         return method_exists($obj, '__toString') ? (string) $obj : $obj::class . '@' . spl_object_hash($obj);
     }
+
+    /** @psalm-param CommitOptions $options */
+    private function doCommit(array $options): void
+    {
+        foreach ($this->getClassesForCommitAction($this->scheduledDocumentUpserts) as $classAndDocuments) {
+            [$class, $documents] = $classAndDocuments;
+            $this->executeUpserts($class, $documents, $options);
+        }
+
+        foreach ($this->getClassesForCommitAction($this->scheduledDocumentInsertions) as $classAndDocuments) {
+            [$class, $documents] = $classAndDocuments;
+            $this->executeInserts($class, $documents, $options);
+        }
+
+        foreach ($this->getClassesForCommitAction($this->scheduledDocumentUpdates) as $classAndDocuments) {
+            [$class, $documents] = $classAndDocuments;
+            $this->executeUpdates($class, $documents, $options);
+        }
+
+        foreach ($this->getClassesForCommitAction($this->scheduledDocumentDeletions, true) as $classAndDocuments) {
+            [$class, $documents] = $classAndDocuments;
+            $this->executeDeletions($class, $documents, $options);
+        }
+    }
+
+    /** @psalm-param CommitOptions $options */
+    private function useTransaction(array $options): bool
+    {
+        if (isset($options['withTransaction'])) {
+            return $options['withTransaction'];
+        }
+
+        return $this->dm->getConfiguration()->isTransactionalFlushEnabled();
+    }
+
+    /** @psalm-param CommitOptions $options */
+    private function getTransactionOptions(array $options): array
+    {
+        return array_intersect_key(
+            array_merge(
+                $this->dm->getConfiguration()->getDefaultCommitOptions(),
+                $options,
+            ),
+            self::TRANSACTION_OPTIONS,
+        );
+    }
 }

phpstan-baseline.neon

@@ -535,6 +535,11 @@ parameters:
 			count: 1
 			path: lib/Doctrine/ODM/MongoDB/PersistentCollection.php
 
+		-
+			message: "#^Method Doctrine\\\\ODM\\\\MongoDB\\\\Persisters\\\\DocumentPersister\\:\\:isInTransaction\\(\\) has parameter \\$options with no value type specified in iterable type array\\.$#"
+			count: 1
+			path: lib/Doctrine/ODM/MongoDB/Persisters/DocumentPersister.php
+
 		-
 			message: "#^Method Doctrine\\\\ODM\\\\MongoDB\\\\Proxy\\\\Factory\\\\StaticProxyFactory\\:\\:createInitializer\\(\\) should return Closure\\(ProxyManager\\\\Proxy\\\\GhostObjectInterface\\<TDocument\\>&TDocument\\=, string\\=, array\\<string, mixed\\>\\=, Closure\\|null\\=, array\\<string, mixed\\>\\=\\)\\: bool but returns Closure\\(ProxyManager\\\\Proxy\\\\GhostObjectInterface, string, array, mixed, array\\)\\: true\\.$#"
 			count: 1
@@ -590,6 +595,21 @@ parameters:
 			count: 1
 			path: lib/Doctrine/ODM/MongoDB/Types/DateType.php
 
+		-
+			message: "#^Method Doctrine\\\\ODM\\\\MongoDB\\\\UnitOfWork\\:\\:getTransactionOptions\\(\\) return type has no value type specified in iterable type array\\.$#"
+			count: 1
+			path: lib/Doctrine/ODM/MongoDB/UnitOfWork.php
+
+		-
+			message: "#^Method Doctrine\\\\ODM\\\\MongoDB\\\\UnitOfWork\\:\\:stripTransactionOptions\\(\\) has parameter \\$options with no value type specified in iterable type array\\.$#"
+			count: 1
+			path: lib/Doctrine/ODM/MongoDB/UnitOfWork.php
+
+		-
+			message: "#^Method Doctrine\\\\ODM\\\\MongoDB\\\\UnitOfWork\\:\\:stripTransactionOptions\\(\\) return type has no value type specified in iterable type array\\.$#"
+			count: 1
+			path: lib/Doctrine/ODM/MongoDB/UnitOfWork.php
+
 		-
 			message: "#^Unable to resolve the template type T in call to method Doctrine\\\\ODM\\\\MongoDB\\\\DocumentManager\\:\\:getClassMetadata\\(\\)$#"
 			count: 1

tests/Doctrine/ODM/MongoDB/Tests/BaseTestCase.php

@@ -11,6 +11,8 @@
 use Doctrine\ODM\MongoDB\UnitOfWork;
 use Doctrine\Persistence\Mapping\Driver\MappingDriver;
 use MongoDB\Client;
+use MongoDB\Driver\Manager;
+use MongoDB\Driver\Server;
 use MongoDB\Model\DatabaseInfo;
 use PHPUnit\Framework\TestCase;
 
@@ -34,6 +36,8 @@
 
 abstract class BaseTestCase extends TestCase
 {
+    protected static ?bool $supportsTransactions;
+    protected static bool $allowsTransactions = true;
     protected ?DocumentManager $dm;
     protected UnitOfWork $uow;
 
@@ -87,6 +91,9 @@ protected static function getConfiguration(): Configuration
         $config->addFilter('testFilter', Filter::class);
         $config->addFilter('testFilter2', Filter::class);
 
+        // Enable transactions if supported
+        $config->setUseTransactionalFlush(static::$allowsTransactions && self::supportsTransactions());
+
         return $config;
     }
 
@@ -127,6 +134,32 @@ protected function getServerVersion(): string
         return $result['version'];
     }
 
+    protected function getPrimaryServer(): Server
+    {
+        return $this->dm->getClient()->getManager()->selectServer();
+    }
+
+    protected function skipTestIfNoTransactionSupport(): void
+    {
+        if (! self::supportsTransactions()) {
+            $this->markTestSkipped('Test requires a topology that supports transactions');
+        }
+    }
+
+    protected function skipTestIfTransactionalFlushDisabled(): void
+    {
+        if (! $this->dm?->getConfiguration()->isTransactionalFlushEnabled()) {
+            $this->markTestSkipped('Test only applies when transactional flush is enabled');
+        }
+    }
+
+    protected function skipTestIfTransactionalFlushEnabled(): void
+    {
+        if ($this->dm?->getConfiguration()->isTransactionalFlushEnabled()) {
+            $this->markTestSkipped('Test is not compatible with transactional flush');
+        }
+    }
+
     /** @psalm-param class-string $className */
     protected function skipTestIfNotSharded(string $className): void
     {
@@ -208,4 +241,16 @@ protected static function removeMultipleHosts(string $uri): string
 
         return substr_replace($uri, $singleHost, $pos, strlen($multipleHosts));
     }
+
+    protected static function supportsTransactions(): bool
+    {
+        return self::$supportsTransactions ??= self::detectTransactionSupport();
+    }
+
+    private static function detectTransactionSupport(): bool
+    {
+        $manager = new Manager(self::getUri());
+
+        return $manager->selectServer()->getType() !== Server::TYPE_STANDALONE;
+    }
 }

tests/Doctrine/ODM/MongoDB/Tests/Functional/AtomicSetTest.php

@@ -25,6 +25,9 @@
  */
 class AtomicSetTest extends BaseTestCase
 {
+    // This test counts executed commands and thus doesn't work with transactions
+    protected static bool $allowsTransactions = false;
+
     private CommandLogger $logger;
 
     public function setUp(): void

tests/Doctrine/ODM/MongoDB/Tests/Functional/CollectionPersisterTest.php

@@ -15,6 +15,9 @@
 
 class CollectionPersisterTest extends BaseTestCase
 {
+    // This test counts executed commands and thus doesn't work with transactions
+    protected static bool $allowsTransactions = false;
+
     private CommandLogger $logger;
 
     public function setUp(): void

tests/Doctrine/ODM/MongoDB/Tests/Functional/CommitImprovementTest.php

@@ -18,6 +18,9 @@
 
 class CommitImprovementTest extends BaseTestCase
 {
+    // This test counts executed commands and thus doesn't work with transactions
+    protected static bool $allowsTransactions = false;
+
     private CommandLogger $logger;
 
     public function setUp(): void