Address review

Author: NoahStapp

pymongo/asynchronous/mongo_client.py

@@ -1566,14 +1566,9 @@ async def close(self) -> None:
             await self._encrypter.close()
         self._closed = True
         if not _IS_SYNC:
-            join_tasks = [self._kill_cursors_executor]
-            try:
-                while self._topology._monitor_tasks:
-                    join_tasks.append(self._topology._monitor_tasks.pop())
-            except IndexError:
-                pass
-            join_tasks = [t.join() for t in join_tasks]  # type: ignore[func-returns-value]
-            await asyncio.gather(*join_tasks)
+            await asyncio.gather(
+                *[self._topology.cleanup_monitors(), self._kill_cursors_executor.join()]  # type: ignore[func-returns-value]
+            )
 
     if not _IS_SYNC:
         # Add support for contextlib.aclosing.

pymongo/asynchronous/monitor.py

@@ -190,7 +190,7 @@ def gc_safe_close(self) -> None:
         self.cancel_check()
 
     async def join(self) -> None:
-        await asyncio.gather(*[self._executor.join(), self._rtt_monitor.join()])
+        await asyncio.gather(*[self._executor.join(), self._rtt_monitor.join()])  # type: ignore[func-returns-value]
 
     async def close(self) -> None:
         self.gc_safe_close()

pymongo/asynchronous/topology.py

@@ -289,27 +289,19 @@ async def select_servers(
         else:
             server_timeout = server_selection_timeout
 
+        # Cleanup any completed monitor tasks safely
+        if not _IS_SYNC and self._monitor_tasks:
+            await self.cleanup_monitors()
+
         async with self._lock:
             server_descriptions = await self._select_servers_loop(
                 selector, server_timeout, operation, operation_id, address
             )
 
-            servers = [
+            return [
                 cast(Server, self.get_server_by_address(sd.address)) for sd in server_descriptions
             ]
 
-        if not _IS_SYNC and self._monitor_tasks:
-            join_tasks = []
-            try:
-                while self._monitor_tasks:
-                    join_tasks.append(self._monitor_tasks.pop())
-            except IndexError:
-                pass
-            join_tasks = [t.join() for t in join_tasks]  # type: ignore[func-returns-value]
-            await asyncio.gather(*join_tasks)
-
-        return servers
-
     async def _select_servers_loop(
         self,
         selector: Callable[[Selection], Selection],
@@ -1057,6 +1049,15 @@ def _error_message(self, selector: Callable[[Selection], Selection]) -> str:
             else:
                 return ",".join(str(server.error) for server in servers if server.error)
 
+    async def cleanup_monitors(self) -> None:
+        tasks = []
+        try:
+            while self._monitor_tasks:
+                tasks.append(self._monitor_tasks.pop())
+        except IndexError:
+            pass
+        await asyncio.gather(*[t.join() for t in tasks])  # type: ignore[func-returns-value]
+
     def __repr__(self) -> str:
         msg = ""
         if not self._opened:

pymongo/synchronous/mongo_client.py

@@ -1560,14 +1560,9 @@ def close(self) -> None:
             self._encrypter.close()
         self._closed = True
         if not _IS_SYNC:
-            join_tasks = [self._kill_cursors_executor]
-            try:
-                while self._topology._monitor_tasks:
-                    join_tasks.append(self._topology._monitor_tasks.pop())
-            except IndexError:
-                pass
-            join_tasks = [t.join() for t in join_tasks]  # type: ignore[func-returns-value]
-            asyncio.gather(*join_tasks)
+            asyncio.gather(
+                *[self._topology.cleanup_monitors(), self._kill_cursors_executor.join()]  # type: ignore[func-returns-value]
+            )
 
     if not _IS_SYNC:
         # Add support for contextlib.closing.

pymongo/synchronous/monitor.py

@@ -190,7 +190,7 @@ def gc_safe_close(self) -> None:
         self.cancel_check()
 
     def join(self) -> None:
-        asyncio.gather(*[self._executor.join(), self._rtt_monitor.join()])
+        asyncio.gather(*[self._executor.join(), self._rtt_monitor.join()])  # type: ignore[func-returns-value]
 
     def close(self) -> None:
         self.gc_safe_close()

pymongo/synchronous/topology.py

@@ -289,27 +289,19 @@ def select_servers(
         else:
             server_timeout = server_selection_timeout
 
+        # Cleanup any completed monitor tasks safely
+        if not _IS_SYNC and self._monitor_tasks:
+            self.cleanup_monitors()
+
         with self._lock:
             server_descriptions = self._select_servers_loop(
                 selector, server_timeout, operation, operation_id, address
             )
 
-            servers = [
+            return [
                 cast(Server, self.get_server_by_address(sd.address)) for sd in server_descriptions
             ]
 
-        if not _IS_SYNC and self._monitor_tasks:
-            join_tasks = []
-            try:
-                while self._monitor_tasks:
-                    join_tasks.append(self._monitor_tasks.pop())
-            except IndexError:
-                pass
-            join_tasks = [t.join() for t in join_tasks]  # type: ignore[func-returns-value]
-            asyncio.gather(*join_tasks)
-
-        return servers
-
     def _select_servers_loop(
         self,
         selector: Callable[[Selection], Selection],
@@ -1055,6 +1047,15 @@ def _error_message(self, selector: Callable[[Selection], Selection]) -> str:
             else:
                 return ",".join(str(server.error) for server in servers if server.error)
 
+    def cleanup_monitors(self) -> None:
+        tasks = []
+        try:
+            while self._monitor_tasks:
+                tasks.append(self._monitor_tasks.pop())
+        except IndexError:
+            pass
+        asyncio.gather(*[t.join() for t in tasks])  # type: ignore[func-returns-value]
+
     def __repr__(self) -> str:
         msg = ""
         if not self._opened: