Close worker background event loops (#256)

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: adb27bb3-082b-4553-9941-7672749d7424

Author: andystaples

CHANGELOG.md

@@ -54,6 +54,9 @@ import paths, `__all__`, `dir()`, and star-imports behave exactly as before.
 
 FIXED
 
+- Fixed `TaskHubGrpcWorker` leaving its background event loop unclosed after
+shutdown, which could retain resources and emit delayed `ResourceWarning`
+messages.
 - Fixed `AsyncTaskHubGrpcClient` failing during construction when no current
 event loop was set. SDK-owned async gRPC channels are now created on first use,
 binding them to the event loop that performs the RPC.

durabletask/worker.py

@@ -760,9 +760,7 @@ def start(self):
         self._shutdown.clear()
 
         def run_loop() -> None:
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            loop.run_until_complete(self._async_run_loop())
+            asyncio.run(self._async_run_loop())
 
         self._logger.info(f"Starting gRPC worker that connects to {self._host_address}")
         self._runLoop = Thread(target=run_loop)

tests/durabletask/test_worker_resiliency.py

@@ -180,6 +180,23 @@ async def fake_run_loop():
     worker.stop()
 
 
+def test_worker_start_closes_background_event_loop():
+    worker = TaskHubGrpcWorker()
+    event_loops = []
+
+    async def fake_run_loop():
+        event_loops.append(asyncio.get_running_loop())
+
+    worker._async_run_loop = fake_run_loop
+    worker.start()
+    worker._runLoop.join(timeout=1.0)
+
+    assert len(event_loops) == 1
+    assert event_loops[0].is_closed() is True
+
+    worker.stop()
+
+
 def test_worker_classifies_graceful_close_before_first_message():
     worker = TaskHubGrpcWorker(
         resiliency_options=GrpcWorkerResiliencyOptions(silent_disconnect_timeout_seconds=5.0)