worker: infinitely wait for start (#983)

* worker: infinitely wait for start

Instead of returning error on worker start after 10s, wait indefinitely
until the worker can start or until shutdown has been signalled. This is
important for environments whereby you spin up 100-500 workers at the
same time and it takes some time for the cluster to settle.

Signed-off-by: joshvanl <me@joshvanl.dev>

* Address comments

Signed-off-by: joshvanl <me@joshvanl.dev>

* review comments

Signed-off-by: joshvanl <me@joshvanl.dev>

* Adds unit test for change

Signed-off-by: joshvanl <me@joshvanl.dev>

* lint

Signed-off-by: joshvanl <me@joshvanl.dev>

---------

Signed-off-by: joshvanl <me@joshvanl.dev>

Author: JoshVanL

ext/dapr-ext-workflow/dapr/ext/workflow/_durabletask/worker.py

@@ -331,6 +331,7 @@ def __init__(
         self._current_channel: Optional[grpc.Channel] = None  # Store channel reference for cleanup
         self._channel_cleanup_threads: list[threading.Thread] = []  # Deferred channel close threads
         self._stream_ready = threading.Event()
+        self._runLoop: Optional[Thread] = None
         # Use provided concurrency options or create default ones
         self._concurrency_options = (
             concurrency_options if concurrency_options is not None else ConcurrencyOptions()
@@ -387,8 +388,13 @@ def run_loop():
         self._logger.info(f'Starting gRPC worker that connects to {self._host_address}')
         self._runLoop = Thread(target=run_loop, name='WorkerRunLoop')
         self._runLoop.start()
-        if not self._stream_ready.wait(timeout=10):
-            raise RuntimeError('Failed to establish work item stream connection within 10 seconds')
+        while not self._stream_ready.wait(timeout=1):
+            if self._shutdown.is_set():
+                raise RuntimeError('Worker was stopped before the work item stream was established')
+            if not self._runLoop.is_alive():
+                raise RuntimeError(
+                    'Worker run loop exited before the work item stream was established'
+                )
         self._is_running = True
 
     async def _keepalive_loop(self, stub):
@@ -801,7 +807,9 @@ def _deferred_close():
 
     def stop(self):
         """Stops the worker and waits for any pending work items to complete."""
-        if not self._is_running:
+        # Guards on _runLoop rather than _is_running so stop() can unblock a start()
+        # that is still waiting for the work item stream to be established.
+        if self._runLoop is None:
             return
 
         self._logger.info('Stopping gRPC worker...')
@@ -833,6 +841,7 @@ def stop(self):
         self._async_worker_manager.shutdown()
         self._logger.info('Worker shutdown completed')
         self._is_running = False
+        self._runLoop = None
 
     # TODO: This should be removed in the future as we do handle grpc errs
     def _handle_grpc_execution_error(self, rpc_error: grpc.RpcError, request_type: str):

ext/dapr-ext-workflow/tests/durabletask/test_worker_stop.py

@@ -1,6 +1,10 @@
+import asyncio
+import threading
+import time
 from unittest.mock import MagicMock, patch
 
 import grpc
+import pytest
 from dapr.ext.workflow._durabletask.worker import TaskHubGrpcWorker
 
 
@@ -146,3 +150,65 @@ def test_deferred_close_prunes_finished_threads():
     worker._channel_cleanup_threads[-1].join(timeout=2)
     # Only the still-alive (or just-finished ch2) thread remains; ch1's was pruned
     assert len(worker._channel_cleanup_threads) <= 1
+
+
+def test_stop_before_start_is_noop():
+    """stop() is safe to call before start() — _runLoop is None, no AttributeError."""
+    worker = TaskHubGrpcWorker()
+    with patch.object(worker._shutdown, 'set') as shutdown_set:
+        worker.stop()
+        shutdown_set.assert_not_called()
+
+
+def test_stop_is_idempotent():
+    """A second stop() returns early because _runLoop was cleared by the first."""
+    worker = _make_running_worker()
+    worker._current_channel = MagicMock()
+    worker.stop()
+    assert worker._runLoop is None
+    with patch.object(worker._shutdown, 'set') as shutdown_set:
+        worker.stop()
+        shutdown_set.assert_not_called()
+
+
+def test_start_raises_when_run_loop_exits_early():
+    """start() raises RuntimeError if the run loop thread exits before _stream_ready is set."""
+    worker = TaskHubGrpcWorker()
+
+    async def fast_exit():
+        return
+
+    with patch.object(worker, '_async_run_loop', side_effect=fast_exit):
+        with pytest.raises(RuntimeError, match='Worker run loop exited'):
+            worker.start()
+
+
+def test_start_raises_when_stopped_during_startup():
+    """stop() unblocks a start() that is waiting for _stream_ready; start() raises."""
+    worker = TaskHubGrpcWorker()
+
+    async def wait_for_shutdown():
+        # Block without setting _stream_ready so start() stays in its wait loop.
+        while not worker._shutdown.is_set():
+            await asyncio.sleep(0.05)
+
+    errors = []
+
+    def _start():
+        try:
+            worker.start()
+        except Exception as e:  # noqa: BLE001
+            errors.append(e)
+
+    with patch.object(worker, '_async_run_loop', side_effect=wait_for_shutdown):
+        t = threading.Thread(target=_start)
+        t.start()
+        # Let start() enter its wait loop (timeout=1 per iteration).
+        time.sleep(1.2)
+        worker.stop()
+        t.join(timeout=5)
+
+    assert not t.is_alive(), 'start() did not return after stop()'
+    assert len(errors) == 1, f'Expected exactly one error, got: {errors}'
+    assert isinstance(errors[0], RuntimeError)
+    assert 'Worker was stopped' in str(errors[0])