PYTHON-5517 Updates to connection pool backoff

Author: blink1073

pymongo/asynchronous/pool.py

@@ -52,6 +52,7 @@
     DocumentTooLarge,
     ExecutionTimeout,
     InvalidOperation,
+    NetworkTimeout,
     NotPrimaryError,
     OperationFailure,
     PyMongoError,
@@ -723,6 +724,7 @@ class PoolState:
     PAUSED = 1
     READY = 2
     CLOSED = 3
+    BACKOFF = 4
 
 
 # Do *not* explicitly inherit from object or Jython won't call __del__
@@ -791,6 +793,7 @@ def __init__(
         self._pending = 0
         self._client_id = client_id
         self._backoff = 0
+        self._backoff_connection_time = -1
         if self.enabled_for_cmap:
             assert self.opts._event_listeners is not None
             self.opts._event_listeners.publish_pool_created(
@@ -817,6 +820,9 @@ def __init__(
     async def ready(self) -> None:
         # Take the lock to avoid the race condition described in PYTHON-2699.
         async with self.lock:
+            # Do not set the pool as ready if in backoff.
+            if self._backoff:
+                return
             if self.state != PoolState.READY:
                 self.state = PoolState.READY
                 if self.enabled_for_cmap:
@@ -846,7 +852,7 @@ async def _reset(
         async with self.size_cond:
             if self.closed:
                 return
-            # Clear the backoff state.
+            # Clear the backoff amount.
             self._backoff = 0
             if self.opts.pause_enabled and pause and not self.opts.load_balanced:
                 old_state, self.state = self.state, PoolState.PAUSED
@@ -1029,26 +1035,34 @@ async def remove_stale_sockets(self, reference_generation: int) -> None:
                     self.requests -= 1
                     self.size_cond.notify()
 
-    def _handle_connection_error(self, error: BaseException, phase: str, conn_id: int) -> None:
+    def _handle_connection_error(self, error: BaseException, phase: str) -> None:
         # Handle system overload condition for non-sdam pools.
-        # Look for an AutoReconnect error raised from a ConnectionResetError with
-        # errno == errno.ECONNRESET or raised from an OSError that we've created due to
-        # a closed connection.
+        # Look for an AutoReconnect or NetworkTimeout error.
         # If found, set backoff and add error labels.
-        if self.is_sdam or type(error) != AutoReconnect:
+        if self.is_sdam or type(error) not in (AutoReconnect, NetworkTimeout):
             return
-        self._backoff += 1
         error._add_error_label("SystemOverloadedError")
         error._add_error_label("RetryableError")
+        self.backoff()
+
+    def backoff(self):
+        """Set/increase backoff mode."""
+        self._backoff += 1
+        if self.state != PoolState.BACKOFF:
+            self.state = PoolState.BACKOFF
+        if self.enabled_for_cmap:
+            assert self.opts._event_listeners is not None
+            self.opts._event_listeners.publish_pool_backoff(self.address, self._backoff)
+        self._backoff_connection_time = _backoff(self._backoff) + time.monotonic()
+
         # Log the pool backoff message.
         if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
             _debug_log(
                 _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.POOL_BACKOFF,
+                message=_ConnectionStatusMessage.POOL_BACKOFF % self._backoff,
                 clientId=self._client_id,
                 serverHost=self.address[0],
                 serverPort=self.address[1],
-                driverConnectionId=conn_id,
                 reason=_verbose_connection_error_reason(ConnectionClosedReason.POOL_BACKOFF),
                 error=ConnectionClosedReason.POOL_BACKOFF,
             )
@@ -1082,10 +1096,6 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
                 driverConnectionId=conn_id,
             )
 
-        # Apply backoff if applicable.
-        if self._backoff:
-            await asyncio.sleep(_backoff(self._backoff))
-
         # Pass a context to determine if we successfully create a configured socket.
         context = dict(has_created_socket=False)
 
@@ -1114,7 +1124,7 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
                     error=ConnectionClosedReason.ERROR,
                 )
             if context["has_created_socket"]:
-                self._handle_connection_error(error, "handshake", conn_id)
+                self._handle_connection_error(error, "handshake")
             if isinstance(error, (IOError, OSError, *SSLErrors)):
                 details = _get_timeout_details(self.opts)
                 _raise_connection_failure(self.address, error, timeout_details=details)
@@ -1138,15 +1148,18 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
         except BaseException as e:
             async with self.lock:
                 self.active_contexts.discard(conn.cancel_context)
-            self._handle_connection_error(e, "hello", conn_id)
+            self._handle_connection_error(e, "hello")
             await conn.close_conn(ConnectionClosedReason.ERROR)
             raise
 
         if handler:
             await handler.client._topology.receive_cluster_time(conn._cluster_time)
 
         # Clear the backoff state.
-        self._backoff = 0
+        if self._backoff:
+            self._backoff = 0
+            await self.ready()
+
         return conn
 
     @contextlib.asynccontextmanager
@@ -1342,6 +1355,9 @@ async def _get_conn(
                     if await self._perished(conn):
                         conn = None
                         continue
+                # See if we need to wait for the backoff period.
+                elif self._backoff and (self._backoff_connection_time < time.monotonic()):
+                    continue
                 else:  # We need to create a new connection
                     try:
                         conn = await self.connect(handler=handler)

pymongo/logger.py

@@ -42,7 +42,7 @@ class _ConnectionStatusMessage(str, enum.Enum):
     POOL_READY = "Connection pool ready"
     POOL_CLOSED = "Connection pool closed"
     POOL_CLEARED = "Connection pool cleared"
-    POOL_BACKOFF = "Connection pool backoff"
+    POOL_BACKOFF = "Connection pool backoff attempt number {%s}"
 
     CONN_CREATED = "Connection created"
     CONN_READY = "Connection ready"

pymongo/monitoring.py

@@ -914,6 +914,28 @@ class PoolClosedEvent(_PoolEvent):
     __slots__ = ()
 
 
+class PoolBackoffEvent(_PoolEvent):
+    """Published when a Connection Pool is backing off.
+
+    :param address: The address (host, port) pair of the server this Pool is
+       attempting to connect to.
+    :param attempt: The backoff attempt number.
+
+    .. versionadded:: 4.16
+    """
+
+    __slots__ = ("__attempt",)
+
+    def __init__(self, address: _Address, attempt: int) -> None:
+        super().__init__(address)
+        self.__attempt = attempt
+
+    @property
+    def attempt(self) -> Optional[ObjectId]:
+        """The backoff attempt number."""
+        return self.__attempt
+
+
 class ConnectionClosedReason:
     """An enum that defines values for `reason` on a
     :class:`ConnectionClosedEvent`.
@@ -1830,6 +1852,15 @@ def publish_pool_closed(self, address: _Address) -> None:
             except Exception:
                 _handle_exception()
 
+    def publish_pool_backoutt(self, address: _Address, attempt: int) -> None:
+        """Publish a :class:`PoolBackoffEvent` to all pool listeners."""
+        event = PoolBackoffEvent(address, attempt)
+        for subscriber in self.__cmap_listeners:
+            try:
+                subscriber.pool_closed(event)
+            except Exception:
+                _handle_exception()
+
     def publish_connection_created(self, address: _Address, connection_id: int) -> None:
         """Publish a :class:`ConnectionCreatedEvent` to all connection
         listeners.

pymongo/synchronous/pool.py

@@ -49,6 +49,7 @@
     DocumentTooLarge,
     ExecutionTimeout,
     InvalidOperation,
+    NetworkTimeout,
     NotPrimaryError,
     OperationFailure,
     PyMongoError,
@@ -721,6 +722,7 @@ class PoolState:
     PAUSED = 1
     READY = 2
     CLOSED = 3
+    BACKOFF = 4
 
 
 # Do *not* explicitly inherit from object or Jython won't call __del__
@@ -789,6 +791,7 @@ def __init__(
         self._pending = 0
         self._client_id = client_id
         self._backoff = 0
+        self._backoff_connection_time = -1
         if self.enabled_for_cmap:
             assert self.opts._event_listeners is not None
             self.opts._event_listeners.publish_pool_created(
@@ -815,6 +818,9 @@ def __init__(
     def ready(self) -> None:
         # Take the lock to avoid the race condition described in PYTHON-2699.
         with self.lock:
+            # Do not set the pool as ready if in backoff.
+            if self._backoff:
+                return
             if self.state != PoolState.READY:
                 self.state = PoolState.READY
                 if self.enabled_for_cmap:
@@ -844,7 +850,7 @@ def _reset(
         with self.size_cond:
             if self.closed:
                 return
-            # Clear the backoff state.
+            # Clear the backoff amount.
             self._backoff = 0
             if self.opts.pause_enabled and pause and not self.opts.load_balanced:
                 old_state, self.state = self.state, PoolState.PAUSED
@@ -1025,26 +1031,34 @@ def remove_stale_sockets(self, reference_generation: int) -> None:
                     self.requests -= 1
                     self.size_cond.notify()
 
-    def _handle_connection_error(self, error: BaseException, phase: str, conn_id: int) -> None:
+    def _handle_connection_error(self, error: BaseException, phase: str) -> None:
         # Handle system overload condition for non-sdam pools.
-        # Look for an AutoReconnect error raised from a ConnectionResetError with
-        # errno == errno.ECONNRESET or raised from an OSError that we've created due to
-        # a closed connection.
+        # Look for an AutoReconnect or NetworkTimeout error.
         # If found, set backoff and add error labels.
-        if self.is_sdam or type(error) != AutoReconnect:
+        if self.is_sdam or type(error) not in (AutoReconnect, NetworkTimeout):
             return
-        self._backoff += 1
         error._add_error_label("SystemOverloadedError")
         error._add_error_label("RetryableError")
+        self.backoff()
+
+    def backoff(self):
+        """Set/increase backoff mode."""
+        self._backoff += 1
+        if self.state != PoolState.BACKOFF:
+            self.state = PoolState.BACKOFF
+        if self.enabled_for_cmap:
+            assert self.opts._event_listeners is not None
+            self.opts._event_listeners.publish_pool_backoff(self.address, self._backoff)
+        self._backoff_connection_time = _backoff(self._backoff) + time.monotonic()
+
         # Log the pool backoff message.
         if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
             _debug_log(
                 _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.POOL_BACKOFF,
+                message=_ConnectionStatusMessage.POOL_BACKOFF % self._backoff,
                 clientId=self._client_id,
                 serverHost=self.address[0],
                 serverPort=self.address[1],
-                driverConnectionId=conn_id,
                 reason=_verbose_connection_error_reason(ConnectionClosedReason.POOL_BACKOFF),
                 error=ConnectionClosedReason.POOL_BACKOFF,
             )
@@ -1078,10 +1092,6 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
                 driverConnectionId=conn_id,
             )
 
-        # Apply backoff if applicable.
-        if self._backoff:
-            time.sleep(_backoff(self._backoff))
-
         # Pass a context to determine if we successfully create a configured socket.
         context = dict(has_created_socket=False)
 
@@ -1110,7 +1120,7 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
                     error=ConnectionClosedReason.ERROR,
                 )
             if context["has_created_socket"]:
-                self._handle_connection_error(error, "handshake", conn_id)
+                self._handle_connection_error(error, "handshake")
             if isinstance(error, (IOError, OSError, *SSLErrors)):
                 details = _get_timeout_details(self.opts)
                 _raise_connection_failure(self.address, error, timeout_details=details)
@@ -1134,15 +1144,18 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
         except BaseException as e:
             with self.lock:
                 self.active_contexts.discard(conn.cancel_context)
-            self._handle_connection_error(e, "hello", conn_id)
+            self._handle_connection_error(e, "hello")
             conn.close_conn(ConnectionClosedReason.ERROR)
             raise
 
         if handler:
             handler.client._topology.receive_cluster_time(conn._cluster_time)
 
         # Clear the backoff state.
-        self._backoff = 0
+        if self._backoff:
+            self._backoff = 0
+            self.ready()
+
         return conn
 
     @contextlib.contextmanager
@@ -1338,6 +1351,9 @@ def _get_conn(
                     if self._perished(conn):
                         conn = None
                         continue
+                # See if we need to wait for the backoff period.
+                elif self._backoff and (self._backoff_connection_time < time.monotonic()):
+                    continue
                 else:  # We need to create a new connection
                     try:
                         conn = self.connect(handler=handler)

test/connection_logging/connection-logging.json

@@ -331,9 +331,7 @@
                   "uriOptions": {
                     "retryReads": false,
                     "appname": "clientAppName",
-                    "heartbeatFrequencyMS": 10000,
-                    "socketTimeoutMS": 500,
-                    "connectTimeoutMS": 500
+                    "heartbeatFrequencyMS": 10000
                   },
                   "observeLogMessages": {
                     "connection": "debug"
@@ -357,9 +355,7 @@
                 "failCommands": [
                   "saslContinue"
                 ],
-                "closeConnection": false,
-                "blockConnection": true,
-                "blockTimeMS": 1000,
+                "errorCode": 18,
                 "appName": "clientAppName"
               }
             }
@@ -372,7 +368,7 @@
             "filter": {}
           },
           "expectError": {
-            "isClientError": true
+            "isError": true
           }
         }
       ],

test/connection_monitoring/pool-create-min-size-error.json

@@ -15,17 +15,13 @@
         "isMaster",
         "hello"
       ],
-      "closeConnection": false,
-      "blockConnection": true,
-      "blockTimeMS": 1000,
+      "errorCode": 18,
       "appName": "poolCreateMinSizeErrorTest"
     }
   },
   "poolOptions": {
     "minPoolSize": 1,
     "backgroundThreadIntervalMS": 50,
-    "socketTimeoutMS": 500,
-    "connectTimeoutMS": 500,
     "appName": "poolCreateMinSizeErrorTest"
   },
   "operations": [

test/discovery_and_monitoring/unified/auth-network-error.json

@@ -53,9 +53,7 @@
                 "failCommands": [
                   "saslContinue"
                 ],
-                "closeConnection": false,
-                "blockConnection": true,
-                "blockTimeMS": 1000,
+                "errorCode": 18,
                 "appName": "authNetworkErrorTest"
               }
             }