Only retry transaction once

doctrine · Jan 9, 2024 · e756e3f · e756e3f
1 parent 6126986
commit e756e3f
Show file tree

Hide file tree

Showing 3 changed files with 128 additions and 5 deletions.
diff --git a/lib/Doctrine/ODM/MongoDB/UnitOfWork.php b/lib/Doctrine/ODM/MongoDB/UnitOfWork.php
@@ -25,10 +25,12 @@
 use Doctrine\Persistence\PropertyChangedListener;
 use InvalidArgumentException;
 use MongoDB\BSON\UTCDateTime;
+use MongoDB\Driver\Exception\RuntimeException;
 use MongoDB\Driver\Session;
 use MongoDB\Driver\WriteConcern;
 use ProxyManager\Proxy\GhostObjectInterface;
 use ReflectionProperty;
+use Throwable;
 use UnexpectedValueException;
 
 use function array_diff_key;
@@ -37,13 +39,13 @@
 use function array_key_exists;
 use function array_merge;
 use function assert;
+use function call_user_func;
 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;
@@ -460,7 +462,7 @@ public function commit(array $options = []): void
 
                 $this->lifecycleEventManager->enableTransactionalMode($session);
 
-                with_transaction(
+                $this->withTransaction(
                     $session,
                     function (Session $session) use ($options): void {
                         $this->doCommit(['session' => $session] + $this->stripTransactionOptions($options));
@@ -3177,4 +3179,77 @@ private function getTransactionOptions(array $options): array
             self::TRANSACTION_OPTIONS,
         );
     }
+
+    /**
+     * This following method was taken from the MongoDB Library and adapted to not use the default 120 seconds timeout.
+     * The code within this method is licensed under the Apache License. Copyright belongs to MongoDB, Inc.
+     */
+    private function withTransaction(Session $session, callable $callback, array $transactionOptions = []): void
+    {
+        $closureAttempts = 0;
+
+        while (true) {
+            $session->startTransaction($transactionOptions);
+
+            try {
+                $closureAttempts++;
+                call_user_func($callback, $session);
+            } catch (Throwable $e) {
+                if ($session->isInTransaction()) {
+                    $session->abortTransaction();
+                }
+
+                if (
+                    $e instanceof RuntimeException &&
+                    $e->hasErrorLabel('TransientTransactionError') &&
+                    ! $this->shouldAbortWithTransaction($closureAttempts)
+                ) {
+                    continue;
+                }
+
+                throw $e;
+            }
+
+            if (! $session->isInTransaction()) {
+                // Assume callback intentionally ended the transaction
+                return;
+            }
+
+            while (true) {
+                try {
+                    $session->commitTransaction();
+                } catch (RuntimeException $e) {
+                    if (
+                        $e->getCode() !== 50 /* MaxTimeMSExpired */ &&
+                        $e->hasErrorLabel('UnknownTransactionCommitResult') &&
+                        ! $this->shouldAbortWithTransaction($closureAttempts)
+                    ) {
+                        // Retry committing the transaction
+                        continue;
+                    }
+
+                    if (
+                        $e->hasErrorLabel('TransientTransactionError') &&
+                        ! $this->shouldAbortWithTransaction($closureAttempts)
+                    ) {
+                        // Restart the transaction, invoking the callback again
+                        continue 2;
+                    }
+
+                    throw $e;
+                }
+
+                // Commit was successful
+                break;
+            }
+
+            // Transaction was successful
+            break;
+        }
+    }
+
+    private function shouldAbortWithTransaction(int $closureAttempts): bool
+    {
+        return $closureAttempts >= 2;
+    }
 }
diff --git a/phpstan-baseline.neon b/phpstan-baseline.neon
@@ -620,6 +620,11 @@ parameters:
 			count: 1
 			path: lib/Doctrine/ODM/MongoDB/UnitOfWork.php
 
+		-
+			message: "#^Method Doctrine\\\\ODM\\\\MongoDB\\\\UnitOfWork\\:\\:withTransaction\\(\\) has parameter \\$transactionOptions with 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

diff --git a/tests/Doctrine/ODM/MongoDB/Tests/UnitOfWorkTransactionalCommitConsistencyTest.php b/tests/Doctrine/ODM/MongoDB/Tests/UnitOfWorkTransactionalCommitConsistencyTest.php
@@ -115,6 +115,50 @@ public function testTransientInsertError(): void
         self::assertEquals([], $this->uow->getDocumentChangeSet($friendUser));
     }
 
+    public function testMultipleTransientErrors(): void
+    {
+        $firstUser           = new ForumUser();
+        $firstUser->username = 'alcaeus';
+        $this->uow->persist($firstUser);
+
+        $secondUser           = new ForumUser();
+        $secondUser->username = 'jmikola';
+        $this->uow->persist($secondUser);
+
+        $friendUser = new FriendUser('GromNaN');
+        $this->uow->persist($friendUser);
+
+        // Add a failpoint that triggers multiple transient errors. The transaction is expected to fail
+        $this->createTransientFailPoint('insert', 2);
+
+        try {
+            $this->uow->commit();
+            self::fail('Expected exception when committing');
+        } catch (Throwable $e) {
+            self::assertInstanceOf(BulkWriteException::class, $e);
+            self::assertSame(192, $e->getCode());
+        }
+
+        self::assertSame(
+            0,
+            $this->dm->getDocumentCollection(ForumUser::class)->countDocuments(),
+        );
+
+        self::assertSame(
+            0,
+            $this->dm->getDocumentCollection(FriendUser::class)->countDocuments(),
+        );
+
+        self::assertTrue($this->uow->isScheduledForInsert($firstUser));
+        self::assertNotEquals([], $this->uow->getDocumentChangeSet($firstUser));
+
+        self::assertTrue($this->uow->isScheduledForInsert($secondUser));
+        self::assertNotEquals([], $this->uow->getDocumentChangeSet($secondUser));
+
+        self::assertTrue($this->uow->isScheduledForInsert($friendUser));
+        self::assertNotEquals([], $this->uow->getDocumentChangeSet($friendUser));
+    }
+
     public function testDuplicateKeyError(): void
     {
         // Create a unique index on the collection to let the second insert fail
@@ -534,12 +578,11 @@ protected static function getConfiguration(): Configuration
         return $configuration;
     }
 
-    private function createTransientFailPoint(string $failCommand): void
+    private function createTransientFailPoint(string $failCommand, int $times = 1): void
     {
         $this->dm->getClient()->selectDatabase('admin')->command([
             'configureFailPoint' => 'failCommand',
-            // Trigger the error twice, working around retryable writes
-            'mode' => ['times' => 2],
+            'mode' => ['times' => $times],
             'data' => [
                 'errorCode' => 192, // FailPointEnabled
                 'errorLabels' => ['TransientTransactionError'],