fix: Close polling transport in a finally so a cancelled stop() does not leak it

AsyncPollingUpdateProcessor.stop() awaited wait_stopped() and only then closed
the requester. If the caller of stop() was cancelled during that wait, close()
never ran and an owned aiohttp transport could leak. Move the close into a
finally so it still runs on cancellation, while letting CancelledError propagate.
Adds a regression test that cancels stop() mid-wait and asserts the transport is
still closed.

Author: jsonbailey

ldclient/impl/datasource/async_polling.py

@@ -44,11 +44,13 @@ def initialized(self):
 
     async def stop(self):
         self.__stop_with_error_info(None)
-        # Wait for the current poll to finish before closing the transport. This way
-        # stop() does not return until background work has stopped, so we never close
-        # the transport while a request is still using it.
-        await self._task.wait_stopped()
-        await self._requester.close()
+        # Wait for the current poll to finish before closing the transport, so we do
+        # not close it while a request is still using it. The close is in a finally
+        # so an owned transport is still released if stop() is cancelled mid-wait.
+        try:
+            await self._task.wait_stopped()
+        finally:
+            await self._requester.close()
 
     def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]):
         log.info("Stopping AsyncPollingUpdateProcessor")

ldclient/testing/impl/datasource/test_async_polling.py

@@ -438,3 +438,30 @@ async def test_second_start_call_raises(self, mock_interval):
             processor.start()
 
         await processor.stop()
+
+    @pytest.mark.asyncio
+    async def test_stop_closes_transport_when_cancelled_mid_wait(self):
+        # If the caller of stop() is cancelled while it waits for the poll to
+        # finish, the owned transport must still be closed (the close is in a
+        # finally), rather than leaking.
+        requester = MagicMock()
+        requester.close = AsyncMock()
+        processor = make_processor(requester=requester)
+
+        # Replace the repeating task so wait_stopped() hangs until we cancel stop().
+        waiting = asyncio.Event()
+
+        async def hang():
+            waiting.set()
+            await asyncio.Event().wait()
+
+        processor._task = MagicMock()
+        processor._task.wait_stopped = hang
+
+        stop_task = asyncio.create_task(processor.stop())
+        await asyncio.wait_for(waiting.wait(), timeout=2.0)
+        stop_task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await stop_task
+
+        requester.close.assert_awaited_once()