Address review

Author: NoahStapp

pymongo/asynchronous/pool.py

@@ -863,7 +863,8 @@ async def _reset(
         if close:
             if not _IS_SYNC:
                 await asyncio.gather(
-                    *[conn.close_conn(ConnectionClosedReason.POOL_CLOSED) for conn in sockets]
+                    *[conn.close_conn(ConnectionClosedReason.POOL_CLOSED) for conn in sockets],
+                    return_exceptions=True,
                 )
             else:
                 for conn in sockets:
@@ -899,7 +900,8 @@ async def _reset(
                     )
             if not _IS_SYNC:
                 await asyncio.gather(
-                    *[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets]
+                    *[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets],
+                    return_exceptions=True,
                 )
             else:
                 for conn in sockets:
@@ -951,7 +953,8 @@ async def remove_stale_sockets(self, reference_generation: int) -> None:
                     close_conns.append(self.conns.pop())
             if not _IS_SYNC:
                 await asyncio.gather(
-                    *[conn.close_conn(ConnectionClosedReason.IDLE) for conn in close_conns]
+                    *[conn.close_conn(ConnectionClosedReason.IDLE) for conn in close_conns],
+                    return_exceptions=True,
                 )
             else:
                 for conn in close_conns:

pymongo/asynchronous/topology.py

@@ -552,7 +552,7 @@ async def on_change(
             if self._opened and self._description.has_server(server_description.address):
                 await self._process_change(server_description, reset_pool, interrupt_connections)
         # Clear the pool from a failed heartbeat, done outside the lock to avoid blocking on connection close.
-        if self._opened and self._description.has_server(server_description.address) and reset_pool:
+        if reset_pool:
             server = self._servers.get(server_description.address)
             if server:
                 await server.pool.reset(interrupt_connections=interrupt_connections)

pymongo/synchronous/pool.py

@@ -861,7 +861,8 @@ def _reset(
         if close:
             if not _IS_SYNC:
                 asyncio.gather(
-                    *[conn.close_conn(ConnectionClosedReason.POOL_CLOSED) for conn in sockets]
+                    *[conn.close_conn(ConnectionClosedReason.POOL_CLOSED) for conn in sockets],
+                    return_exceptions=True,
                 )
             else:
                 for conn in sockets:
@@ -896,7 +897,10 @@ def _reset(
                         serviceId=service_id,
                     )
             if not _IS_SYNC:
-                asyncio.gather(*[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets])
+                asyncio.gather(
+                    *[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets],
+                    return_exceptions=True,
+                )
             else:
                 for conn in sockets:
                     conn.close_conn(ConnectionClosedReason.STALE)
@@ -945,7 +949,8 @@ def remove_stale_sockets(self, reference_generation: int) -> None:
                     close_conns.append(self.conns.pop())
             if not _IS_SYNC:
                 asyncio.gather(
-                    *[conn.close_conn(ConnectionClosedReason.IDLE) for conn in close_conns]
+                    *[conn.close_conn(ConnectionClosedReason.IDLE) for conn in close_conns],
+                    return_exceptions=True,
                 )
             else:
                 for conn in close_conns:

pymongo/synchronous/topology.py

@@ -552,7 +552,7 @@ def on_change(
             if self._opened and self._description.has_server(server_description.address):
                 self._process_change(server_description, reset_pool, interrupt_connections)
         # Clear the pool from a failed heartbeat, done outside the lock to avoid blocking on connection close.
-        if self._opened and self._description.has_server(server_description.address) and reset_pool:
+        if reset_pool:
             server = self._servers.get(server_description.address)
             if server:
                 server.pool.reset(interrupt_connections=interrupt_connections)