PYTHON-1894: Client-side errors in in-progress transactions must not change transaction state

.evergreen/remove-unimplemented-tests.sh

@@ -1,7 +1,6 @@
 #!/bin/bash
 PYMONGO=$(dirname "$(cd "$(dirname "$0")" || exit; pwd)")
 
-rm $PYMONGO/test/transactions/legacy/errors-client.json  # PYTHON-1894
 rm $PYMONGO/test/connection_monitoring/wait-queue-fairness.json  # PYTHON-1873
 rm $PYMONGO/test/discovery_and_monitoring/unified/pool-clear-application-error.json  # PYTHON-4918
 rm $PYMONGO/test/discovery_and_monitoring/unified/pool-clear-checkout-error.json  # PYTHON-4918

pymongo/asynchronous/bulk.py

@@ -36,7 +36,11 @@
 from bson.objectid import ObjectId
 from bson.raw_bson import RawBSONDocument
 from pymongo import _csot, common
-from pymongo.asynchronous.client_session import AsyncClientSession, _validate_session_write_concern
+from pymongo.asynchronous.client_session import (
+    AsyncClientSession,
+    _TxnState,
+    _validate_session_write_concern,
+)
 from pymongo.asynchronous.helpers import _handle_reauth
 from pymongo.bulk_shared import (
     _COMMANDS,
@@ -271,6 +275,11 @@ async def write_command(
         if bwc.publish:
             bwc._start(cmd, request_id, docs)
         try:
+            if bwc.session is not None and bwc.session._starting_transaction:
+                # Mark the transaction as in progress once the first
+                # transactional bulk message is about to go on the wire.
+                bwc.session._transaction.has_sent_command = True
+                bwc.session._transaction.state = _TxnState.IN_PROGRESS
             reply = await bwc.conn.write_command(request_id, msg, bwc.codec)  # type: ignore[misc]
             duration = datetime.datetime.now() - bwc.start_time
             if _COMMAND_LOGGER.isEnabledFor(logging.DEBUG):

pymongo/asynchronous/client_bulk.py

@@ -35,7 +35,11 @@
 from bson.objectid import ObjectId
 from bson.raw_bson import RawBSONDocument
 from pymongo import _csot, common
-from pymongo.asynchronous.client_session import AsyncClientSession, _validate_session_write_concern
+from pymongo.asynchronous.client_session import (
+    AsyncClientSession,
+    _TxnState,
+    _validate_session_write_concern,
+)
 from pymongo.asynchronous.collection import AsyncCollection
 from pymongo.asynchronous.command_cursor import AsyncCommandCursor
 from pymongo.asynchronous.database import AsyncDatabase
@@ -258,6 +262,9 @@ async def write_command(
         if bwc.publish:
             bwc._start(cmd, request_id, op_docs, ns_docs)
         try:
+            if bwc.session is not None and bwc.session._starting_transaction:
+                bwc.session._transaction.has_sent_command = True
+                bwc.session._transaction.state = _TxnState.IN_PROGRESS
             reply = await bwc.conn.write_command(request_id, msg, bwc.codec)  # type: ignore[misc, arg-type]
             duration = datetime.datetime.now() - bwc.start_time
             if _COMMAND_LOGGER.isEnabledFor(logging.DEBUG):

pymongo/asynchronous/client_session.py

@@ -433,6 +433,7 @@ def __init__(self, opts: Optional[TransactionOptions], client: AsyncMongoClient[
         self.attempt = 0
         self.client = client
         self.has_completed_command = False
+        self.has_sent_command = False
 
     def active(self) -> bool:
         return self.state in (_TxnState.STARTING, _TxnState.IN_PROGRESS)
@@ -469,6 +470,7 @@ async def reset(self) -> None:
         self.recovery_token = None
         self.attempt = 0
         self.has_completed_command = False
+        self.has_sent_command = False
 
     def __del__(self) -> None:
         if self.conn_mgr:
@@ -1135,7 +1137,6 @@ def _apply_to(
 
             if self._transaction.state == _TxnState.STARTING:
                 # First command begins a new transaction.
-                self._transaction.state = _TxnState.IN_PROGRESS
                 command["startTransaction"] = True
 
                 assert self._transaction.opts

pymongo/asynchronous/mongo_client.py

@@ -2870,8 +2870,8 @@ async def run(self) -> T:
                         self._last_error = exc
                         self._attempt_number += 1
 
-                        # Revert back to starting state if we're in a transaction but haven't completed the first
-                        # command.
+                        # Revert back to starting state only if the first
+                        # transactional command was never completed.
                         if (
                             overloaded
                             and self._session is not None
@@ -2921,8 +2921,8 @@ async def run(self) -> T:
                         self._last_error = exc
                     if self._last_error is None:
                         self._last_error = exc
-                    # Revert back to starting state if we're in a transaction but haven't completed the first
-                    # command.
+                    # Revert back to starting state only if the first
+                    # transactional command was never completed.
                     if overloaded and self._session is not None and self._session.in_transaction:
                         transaction = self._session._transaction
                         if not transaction.has_completed_command:

pymongo/asynchronous/pool.py

@@ -38,7 +38,7 @@
 
 from bson import DEFAULT_CODEC_OPTIONS
 from pymongo import _csot, helpers_shared
-from pymongo.asynchronous.client_session import _validate_session_write_concern
+from pymongo.asynchronous.client_session import _TxnState, _validate_session_write_concern
 from pymongo.asynchronous.helpers import _handle_reauth
 from pymongo.asynchronous.network import command
 from pymongo.common import (
@@ -400,6 +400,9 @@ async def command(
         if self.op_msg_enabled:
             self._raise_if_not_writable(unacknowledged)
         try:
+            if session is not None and session._starting_transaction:
+                session._transaction.has_sent_command = True
+                session._transaction.state = _TxnState.IN_PROGRESS
             return await command(
                 self,
                 dbname,

pymongo/asynchronous/server.py

@@ -27,6 +27,7 @@
 )
 
 from bson import _decode_all_selective
+from pymongo.asynchronous.client_session import _TxnState
 from pymongo.asynchronous.helpers import _handle_reauth
 from pymongo.errors import NotPrimaryError, OperationFailure
 from pymongo.helpers_shared import _check_command_response
@@ -204,6 +205,11 @@ async def run_operation(
             if more_to_come:
                 reply = await conn.receive_message(None)
             else:
+                # Mark the transaction as in progress once the first transactional message is about to be sent,
+                # so local validation errors keep the session in STARTING, but post-send failures do not.
+                if operation.session is not None and operation.session._starting_transaction:
+                    operation.session._transaction.has_sent_command = True
+                    operation.session._transaction.state = _TxnState.IN_PROGRESS
                 await conn.send_message(data, max_doc_size)
                 reply = await conn.receive_message(request_id)
 

pymongo/synchronous/bulk.py

@@ -67,7 +67,11 @@
     _randint,
 )
 from pymongo.read_preferences import ReadPreference
-from pymongo.synchronous.client_session import ClientSession, _validate_session_write_concern
+from pymongo.synchronous.client_session import (
+    ClientSession,
+    _TxnState,
+    _validate_session_write_concern,
+)
 from pymongo.synchronous.helpers import _handle_reauth
 from pymongo.write_concern import WriteConcern
 
@@ -271,6 +275,11 @@ def write_command(
         if bwc.publish:
             bwc._start(cmd, request_id, docs)
         try:
+            if bwc.session is not None and bwc.session._starting_transaction:
+                # Mark the transaction as in progress once the first
+                # transactional bulk message is about to go on the wire.
+                bwc.session._transaction.has_sent_command = True
+                bwc.session._transaction.state = _TxnState.IN_PROGRESS
             reply = bwc.conn.write_command(request_id, msg, bwc.codec)  # type: ignore[misc]
             duration = datetime.datetime.now() - bwc.start_time
             if _COMMAND_LOGGER.isEnabledFor(logging.DEBUG):

pymongo/synchronous/client_bulk.py

@@ -35,7 +35,11 @@
 from bson.objectid import ObjectId
 from bson.raw_bson import RawBSONDocument
 from pymongo import _csot, common
-from pymongo.synchronous.client_session import ClientSession, _validate_session_write_concern
+from pymongo.synchronous.client_session import (
+    ClientSession,
+    _TxnState,
+    _validate_session_write_concern,
+)
 from pymongo.synchronous.collection import Collection
 from pymongo.synchronous.command_cursor import CommandCursor
 from pymongo.synchronous.database import Database
@@ -258,6 +262,9 @@ def write_command(
         if bwc.publish:
             bwc._start(cmd, request_id, op_docs, ns_docs)
         try:
+            if bwc.session is not None and bwc.session._starting_transaction:
+                bwc.session._transaction.has_sent_command = True
+                bwc.session._transaction.state = _TxnState.IN_PROGRESS
             reply = bwc.conn.write_command(request_id, msg, bwc.codec)  # type: ignore[misc, arg-type]
             duration = datetime.datetime.now() - bwc.start_time
             if _COMMAND_LOGGER.isEnabledFor(logging.DEBUG):

pymongo/synchronous/client_session.py

@@ -431,6 +431,7 @@ def __init__(self, opts: Optional[TransactionOptions], client: MongoClient[Any])
         self.attempt = 0
         self.client = client
         self.has_completed_command = False
+        self.has_sent_command = False
 
     def active(self) -> bool:
         return self.state in (_TxnState.STARTING, _TxnState.IN_PROGRESS)
@@ -467,6 +468,7 @@ def reset(self) -> None:
         self.recovery_token = None
         self.attempt = 0
         self.has_completed_command = False
+        self.has_sent_command = False
 
     def __del__(self) -> None:
         if self.conn_mgr:
@@ -1131,7 +1133,6 @@ def _apply_to(
 
             if self._transaction.state == _TxnState.STARTING:
                 # First command begins a new transaction.
-                self._transaction.state = _TxnState.IN_PROGRESS
                 command["startTransaction"] = True
 
                 assert self._transaction.opts

pymongo/synchronous/mongo_client.py

@@ -2860,8 +2860,8 @@ def run(self) -> T:
                         self._last_error = exc
                         self._attempt_number += 1
 
-                        # Revert back to starting state if we're in a transaction but haven't completed the first
-                        # command.
+                        # Revert back to starting state only if the first
+                        # transactional command was never completed.
                         if (
                             overloaded
                             and self._session is not None
@@ -2911,8 +2911,8 @@ def run(self) -> T:
                         self._last_error = exc
                     if self._last_error is None:
                         self._last_error = exc
-                    # Revert back to starting state if we're in a transaction but haven't completed the first
-                    # command.
+                    # Revert back to starting state only if the first
+                    # transactional command was never completed.
                     if overloaded and self._session is not None and self._session.in_transaction:
                         transaction = self._session._transaction
                         if not transaction.has_completed_command:

pymongo/synchronous/pool.py

@@ -87,7 +87,7 @@
 from pymongo.server_api import _add_to_command
 from pymongo.server_type import SERVER_TYPE
 from pymongo.socket_checker import SocketChecker
-from pymongo.synchronous.client_session import _validate_session_write_concern
+from pymongo.synchronous.client_session import _TxnState, _validate_session_write_concern
 from pymongo.synchronous.helpers import _handle_reauth
 from pymongo.synchronous.network import command
 
@@ -400,6 +400,9 @@ def command(
         if self.op_msg_enabled:
             self._raise_if_not_writable(unacknowledged)
         try:
+            if session is not None and session._starting_transaction:
+                session._transaction.has_sent_command = True
+                session._transaction.state = _TxnState.IN_PROGRESS
             return command(
                 self,
                 dbname,

pymongo/synchronous/server.py

@@ -38,6 +38,7 @@
 )
 from pymongo.message import _convert_exception, _GetMore, _OpMsg, _Query
 from pymongo.response import PinnedResponse, Response
+from pymongo.synchronous.client_session import _TxnState
 from pymongo.synchronous.helpers import _handle_reauth
 
 if TYPE_CHECKING:
@@ -204,6 +205,11 @@ def run_operation(
             if more_to_come:
                 reply = conn.receive_message(None)
             else:
+                # Mark the transaction as in progress once the first transactional message is about to be sent,
+                # so local validation errors keep the session in STARTING, but post-send failures do not.
+                if operation.session is not None and operation.session._starting_transaction:
+                    operation.session._transaction.has_sent_command = True
+                    operation.session._transaction.state = _TxnState.IN_PROGRESS
                 conn.send_message(data, max_doc_size)
                 reply = conn.receive_message(request_id)
 

test/asynchronous/test_unified_format.py

@@ -39,9 +39,6 @@
         os.path.join(TEST_PATH, "valid-pass"),
         module=__name__,
         class_name_prefix="UnifiedTestFormat",
-        expected_failures=[
-            "Client side error in command starting transaction",  # PYTHON-1894
-        ],
     )
 )
 

test/asynchronous/unified_format.py

@@ -569,8 +569,6 @@ def maybe_skip_test(self, spec):
         class_name = self.__class__.__name__.lower()
         description = spec["description"].lower()
 
-        if "client side error in command starting transaction" in description:
-            self.skipTest("Implement PYTHON-1894")
         if "type=symbol" in description:
             self.skipTest("PyMongo does not support the symbol type")
         if "timeoutms applied to entire download" in description:

test/test_unified_format.py

@@ -39,9 +39,6 @@
         os.path.join(TEST_PATH, "valid-pass"),
         module=__name__,
         class_name_prefix="UnifiedTestFormat",
-        expected_failures=[
-            "Client side error in command starting transaction",  # PYTHON-1894
-        ],
     )
 )
 

test/unified_format.py

@@ -568,8 +568,6 @@ def maybe_skip_test(self, spec):
         class_name = self.__class__.__name__.lower()
         description = spec["description"].lower()
 
-        if "client side error in command starting transaction" in description:
-            self.skipTest("Implement PYTHON-1894")
         if "type=symbol" in description:
             self.skipTest("PyMongo does not support the symbol type")
         if "timeoutms applied to entire download" in description: