PYTHON-3388 Propagate Original Error for Write Errors Labeled NoWritesPerformed

pymongo/mongo_client.py

@@ -81,6 +81,7 @@
     PyMongoError,
     ServerSelectionTimeoutError,
     WaitQueueTimeoutError,
+    WriteConcernError,
 )
 from pymongo.lock import _HAS_REGISTER_AT_FORK, _create_lock, _release_locks
 from pymongo.pool import ConnectionClosedReason
@@ -836,6 +837,8 @@ def __init__(
             # This will be used later if we fork.
             MongoClient._clients[self._topology._topology_id] = self
 
+        self._indefinite_error: Optional[WriteConcernError] = None
+
     def _init_background(self):
         self._topology = Topology(self._topology_settings)
 
@@ -1405,6 +1408,10 @@ def is_retrying():
                 # Add the RetryableWriteError label, if applicable.
                 _add_retryable_write_error(exc, max_wire_version)
                 retryable_error = exc.has_error_label("RetryableWriteError")
+                if self._indefinite_error and exc.has_error_label("NoWritesPerformed"):
+                    raise self._indefinite_error from exc
+                if isinstance(exc, WriteConcernError):
+                    self._indefinite_error = exc
                 if retryable_error:
                     session._unpin()
                 if not retryable_error or (is_retrying() and not multiple_retries):

test/test_retryable_writes.py

@@ -26,9 +26,11 @@
 from test.utils import (
     CMAPListener,
     DeprecationFilter,
+    EventListener,
     OvertCommandListener,
     TestCreator,
     rs_or_single_client,
+    wait_until,
 )
 from test.utils_spec_runner import SpecRunner
 from test.version import Version
@@ -45,6 +47,7 @@
 )
 from pymongo.mongo_client import MongoClient
 from pymongo.monitoring import (
+    CommandSucceededEvent,
     ConnectionCheckedOutEvent,
     ConnectionCheckOutFailedEvent,
     ConnectionCheckOutFailedReason,
@@ -64,6 +67,24 @@
 _TEST_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "retryable_writes", "legacy")
 
 
+class InsertEventListener(EventListener):
+    def __init__(self):
+        self._retried = False
+        super().__init__()
+
+    def reset(self) -> None:
+        self._retried = False
+        super().reset()
+
+    def succeeded(self, event: CommandSucceededEvent):
+        if (
+            event.command_name == "insert"
+            and event.reply.get("writeConcernError", {}).get("code", None) == 91
+        ):
+            self._retried = True
+        super(InsertEventListener, self).succeeded(event)
+
+
 class TestAllScenarios(SpecRunner):
     RUN_ON_LOAD_BALANCER = True
     RUN_ON_SERVERLESS = True
@@ -581,8 +602,56 @@ def test_pool_paused_error_is_retryable(self):
         failed = cmd_listener.failed_events
         self.assertEqual(1, len(failed), msg)
 
+    @client_context.require_failCommand_fail_point
+    @client_context.require_replica_set
+    @client_knobs(heartbeat_frequency=0.05, min_heartbeat_interval=0.05)
+    def test_returns_original_error_code(
+        self,
+    ):  # TODO: Make this a real integration test where we stepdown the primary.
+        cmd_listener = InsertEventListener()
+        client = rs_or_single_client(retryWrites=True, event_listeners=[cmd_listener])
+        client.test.test.drop()
+        client.test.another_coll.drop()
+        self.addCleanup(client.close)
+        cmd_listener.reset()
+        client.admin.command(
+            {
+                "configureFailPoint": "failCommand",
+                "mode": {"times": 1},
+                "data": {
+                    "writeConcernError": {
+                        "code": 91,
+                        "errorLabels": ["RetryableWriteError"],
+                    },
+                    "failCommands": ["insert"],
+                },
+            }
+        )
+        client.test.test.insert_one({"_id": 1})
+        wait_until(lambda: cmd_listener._retried, "ready for second failpoint")
+        client.test.test.drop()
+        client.admin.command(
+            {
+                "configureFailPoint": "failCommand",
+                "mode": {"times": 1},
+                "data": {
+                    "errorCode": 10107,
+                    "errorLabels": ["RetryableWriteError", "NoWritesPerformed"],
+                    "failCommands": ["insert"],
+                },
+            }
+        )
+        with self.assertRaises(WriteConcernError) as exc:
+            client.test.test.insert_one({"_id": 1})
+        assert exc.exception.code == 91
+        client.admin.command(
+            {
+                "configureFailPoint": "failCommand",
+                "mode": "off",
+            }
+        )
+
 
-# TODO: Make this a real integration test where we stepdown the primary.
 class TestRetryableWritesTxnNumber(IgnoreDeprecationsTest):
     @client_context.require_replica_set
     @client_context.require_no_mmap