PYTHON-4600 Handle round trip time being negative when time.monotonic() is not monotonic (#1758)

Co-authored-by: rmorotti <romain.morotti@man.com>

doc/contributors.rst

@@ -101,3 +101,4 @@ The following is a list of people who have contributed to
 - Casey Clements (caseyclements)
 - Ivan Lukyanchikov (ilukyanchikov)
 - Terry Patterson
+- Romain Morotti

pymongo/_csot.py

@@ -149,10 +149,7 @@ class MovingMinimum:
 
     def add_sample(self, sample: float) -> None:
         if sample < 0:
-            # Likely system time change while waiting for hello response
-            # and not using time.monotonic. Ignore it, the next one will
-            # probably be valid.
-            return
+            raise ValueError(f"duration cannot be negative {sample}")
         self.samples.append(sample)
 
     def get(self) -> float:

pymongo/asynchronous/monitor.py

@@ -48,6 +48,15 @@ def _sanitize(error: Exception) -> None:
     error.__cause__ = None
 
 
+def _monotonic_duration(start: float) -> float:
+    """Return the duration since the given start time.
+
+    Accounts for buggy platforms where time.monotonic() is not monotonic.
+    See PYTHON-4600.
+    """
+    return max(0.0, time.monotonic() - start)
+
+
 class MonitorBase:
     def __init__(self, topology: Topology, name: str, interval: int, min_interval: float):
         """Base class to do periodic work on a background thread.
@@ -247,7 +256,7 @@ class Monitor(MonitorBase):
             _sanitize(error)
             sd = self._server_description
             address = sd.address
-            duration = time.monotonic() - start
+            duration = _monotonic_duration(start)
             if self._publish:
                 awaited = bool(self._stream and sd.is_server_type_known and sd.topology_version)
                 assert self._listeners is not None
@@ -317,7 +326,8 @@ class Monitor(MonitorBase):
         else:
             # New connection handshake or polling hello (MongoDB <4.4).
             response = await conn._hello(cluster_time, None, None)
-        return response, time.monotonic() - start
+        duration = _monotonic_duration(start)
+        return response, duration
 
 
 class SrvMonitor(MonitorBase):
@@ -441,7 +451,7 @@ class _RttMonitor(MonitorBase):
                 raise Exception("_RttMonitor closed")
             start = time.monotonic()
             await conn.hello()
-            return time.monotonic() - start
+            return _monotonic_duration(start)
 
 
 # Close monitors to cancel any in progress streaming checks before joining

pymongo/read_preferences.py

@@ -607,10 +607,7 @@ class MovingAverage:
 
     def add_sample(self, sample: float) -> None:
         if sample < 0:
-            # Likely system time change while waiting for hello response
-            # and not using time.monotonic. Ignore it, the next one will
-            # probably be valid.
-            return
+            raise ValueError(f"duration cannot be negative {sample}")
         if self.average is None:
             self.average = sample
         else:

pymongo/synchronous/monitor.py

@@ -48,6 +48,15 @@ def _sanitize(error: Exception) -> None:
     error.__cause__ = None
 
 
+def _monotonic_duration(start: float) -> float:
+    """Return the duration since the given start time.
+
+    Accounts for buggy platforms where time.monotonic() is not monotonic.
+    See PYTHON-4600.
+    """
+    return max(0.0, time.monotonic() - start)
+
+
 class MonitorBase:
     def __init__(self, topology: Topology, name: str, interval: int, min_interval: float):
         """Base class to do periodic work on a background thread.
@@ -247,7 +256,7 @@ class Monitor(MonitorBase):
             _sanitize(error)
             sd = self._server_description
             address = sd.address
-            duration = time.monotonic() - start
+            duration = _monotonic_duration(start)
             if self._publish:
                 awaited = bool(self._stream and sd.is_server_type_known and sd.topology_version)
                 assert self._listeners is not None
@@ -317,7 +326,8 @@ class Monitor(MonitorBase):
         else:
             # New connection handshake or polling hello (MongoDB <4.4).
             response = conn._hello(cluster_time, None, None)
-        return response, time.monotonic() - start
+        duration = _monotonic_duration(start)
+        return response, duration
 
 
 class SrvMonitor(MonitorBase):
@@ -441,7 +451,7 @@ class _RttMonitor(MonitorBase):
                 raise Exception("_RttMonitor closed")
             start = time.monotonic()
             conn.hello()
-            return time.monotonic() - start
+            return _monotonic_duration(start)
 
 
 # Close monitors to cancel any in progress streaming checks before joining