PYTHON-4579 Stop gossiping $clusterTime on SDAM connections

Author: ShaneHarvey

pymongo/asynchronous/monitor.py

@@ -21,11 +21,11 @@
 import logging
 import time
 import weakref
-from typing import TYPE_CHECKING, Any, Mapping, Optional, cast
+from typing import TYPE_CHECKING, Any, Optional
 
 from pymongo import common, periodic_executor
 from pymongo._csot import MovingMinimum
-from pymongo.errors import NetworkTimeout, NotPrimaryError, OperationFailure, _OperationCancelled
+from pymongo.errors import NetworkTimeout, _OperationCancelled
 from pymongo.hello import Hello
 from pymongo.lock import _async_create_lock
 from pymongo.logger import _SDAM_LOGGER, _debug_log, _SDAMStatusMessage
@@ -250,13 +250,7 @@ async def _check_server(self) -> ServerDescription:
         self._conn_id = None
         start = time.monotonic()
         try:
-            try:
-                return await self._check_once()
-            except (OperationFailure, NotPrimaryError) as exc:
-                # Update max cluster time even when hello fails.
-                details = cast(Mapping[str, Any], exc.details)
-                await self._topology.receive_cluster_time(details.get("$clusterTime"))
-                raise
+            return await self._check_once()
         except asyncio.CancelledError:
             raise
         except ReferenceError:
@@ -355,7 +349,6 @@ async def _check_with_socket(self, conn: AsyncConnection) -> tuple[Hello, float]
 
         Can raise ConnectionFailure or OperationFailure.
         """
-        cluster_time = self._topology.max_cluster_time()
         start = time.monotonic()
         if conn.more_to_come:
             # Read the next streaming hello (MongoDB 4.4+).
@@ -365,13 +358,13 @@ async def _check_with_socket(self, conn: AsyncConnection) -> tuple[Hello, float]
         ):
             # Initiate streaming hello (MongoDB 4.4+).
             response = await conn._hello(
-                cluster_time,
+                None,
                 self._server_description.topology_version,
                 self._settings.heartbeat_frequency,
             )
         else:
             # New connection handshake or polling hello (MongoDB <4.4).
-            response = await conn._hello(cluster_time, None, None)
+            response = await conn._hello(None, None, None)
         duration = _monotonic_duration(start)
         return response, duration
 

pymongo/asynchronous/topology.py

@@ -493,7 +493,8 @@ async def _process_change(
 
         self._description = new_td
         await self._update_servers()
-        self._receive_cluster_time_no_lock(server_description.cluster_time)
+        # TODO: Verify that app errors update the $clusterTime.
+        # self._receive_cluster_time_no_lock(server_description.cluster_time)
 
         if self._publish_tp and not suppress_event:
             assert self._events is not None

pymongo/synchronous/monitor.py

@@ -21,11 +21,11 @@
 import logging
 import time
 import weakref
-from typing import TYPE_CHECKING, Any, Mapping, Optional, cast
+from typing import TYPE_CHECKING, Any, Optional
 
 from pymongo import common, periodic_executor
 from pymongo._csot import MovingMinimum
-from pymongo.errors import NetworkTimeout, NotPrimaryError, OperationFailure, _OperationCancelled
+from pymongo.errors import NetworkTimeout, _OperationCancelled
 from pymongo.hello import Hello
 from pymongo.lock import _create_lock
 from pymongo.logger import _SDAM_LOGGER, _debug_log, _SDAMStatusMessage
@@ -250,6 +250,7 @@ def _check_server(self) -> ServerDescription:
         self._conn_id = None
         start = time.monotonic()
         try:
+<<<<<<< HEAD
             try:
                 return self._check_once()
             except (OperationFailure, NotPrimaryError) as exc:
@@ -259,6 +260,17 @@ def _check_server(self) -> ServerDescription:
                 raise
         except asyncio.CancelledError:
             raise
+||||||| parent of 14c8432bc (PYTHON-4579 Stop gossiping $clusterTime on SDAM connections)
+            try:
+                return self._check_once()
+            except (OperationFailure, NotPrimaryError) as exc:
+                # Update max cluster time even when hello fails.
+                details = cast(Mapping[str, Any], exc.details)
+                self._topology.receive_cluster_time(details.get("$clusterTime"))
+                raise
+=======
+            return self._check_once()
+>>>>>>> 14c8432bc (PYTHON-4579 Stop gossiping $clusterTime on SDAM connections)
         except ReferenceError:
             raise
         except Exception as error:
@@ -355,7 +367,6 @@ def _check_with_socket(self, conn: Connection) -> tuple[Hello, float]:
 
         Can raise ConnectionFailure or OperationFailure.
         """
-        cluster_time = self._topology.max_cluster_time()
         start = time.monotonic()
         if conn.more_to_come:
             # Read the next streaming hello (MongoDB 4.4+).
@@ -365,13 +376,13 @@ def _check_with_socket(self, conn: Connection) -> tuple[Hello, float]:
         ):
             # Initiate streaming hello (MongoDB 4.4+).
             response = conn._hello(
-                cluster_time,
+                None,
                 self._server_description.topology_version,
                 self._settings.heartbeat_frequency,
             )
         else:
             # New connection handshake or polling hello (MongoDB <4.4).
-            response = conn._hello(cluster_time, None, None)
+            response = conn._hello(None, None, None)
         duration = _monotonic_duration(start)
         return response, duration
 

pymongo/synchronous/topology.py

@@ -493,7 +493,8 @@ def _process_change(
 
         self._description = new_td
         self._update_servers()
-        self._receive_cluster_time_no_lock(server_description.cluster_time)
+        # TODO: Verify that app errors update the $clusterTime.
+        # self._receive_cluster_time_no_lock(server_description.cluster_time)
 
         if self._publish_tp and not suppress_event:
             assert self._events is not None

test/test_discovery_and_monitoring.py

@@ -244,7 +244,7 @@ class TestClusterTimeComparison(unittest.TestCase):
     def test_cluster_time_comparison(self):
         t = create_mock_topology("mongodb://host")
 
-        def send_cluster_time(time, inc, should_update):
+        def send_cluster_time(time, inc):
             old = t.max_cluster_time()
             new = {"clusterTime": Timestamp(time, inc)}
             got_hello(
@@ -259,16 +259,14 @@ def send_cluster_time(time, inc, should_update):
             )
 
             actual = t.max_cluster_time()
-            if should_update:
-                self.assertEqual(actual, new)
-            else:
-                self.assertEqual(actual, old)
-
-        send_cluster_time(0, 1, True)
-        send_cluster_time(2, 2, True)
-        send_cluster_time(2, 1, False)
-        send_cluster_time(1, 3, False)
-        send_cluster_time(2, 3, True)
+            # We never update $clusterTime from monitoring connections.
+            self.assertEqual(actual, old)
+
+        send_cluster_time(0, 1)
+        send_cluster_time(2, 2)
+        send_cluster_time(2, 1)
+        send_cluster_time(1, 3)
+        send_cluster_time(2, 3)
 
 
 class TestIgnoreStaleErrors(IntegrationTest):