diff --git a/changelog/fixed/patient-namespace-newt.md b/changelog/fixed/patient-namespace-newt.md new file mode 100644 index 000000000..f76c41cfe --- /dev/null +++ b/changelog/fixed/patient-namespace-newt.md @@ -0,0 +1 @@ +Fix `Worker.shutdown()` and `async with Worker(...)` hanging forever when worker validation fails, for example on a namespace that does not exist. `async with` now raises the validation error. diff --git a/temporalio/worker/_worker.py b/temporalio/worker/_worker.py index 3c4c47996..2d14098a6 100644 --- a/temporalio/worker/_worker.py +++ b/temporalio/worker/_worker.py @@ -829,7 +829,14 @@ def make_lambda(plugin: Plugin, next: Callable[[Worker], Awaitable[None]]): async def _run(self): # Eagerly validate which will do a namespace check in Core. - await self._bridge_worker.validate() + try: + await self._bridge_worker.validate() + except BaseException: + # Nothing was started, so there is nothing to shut down. Mark + # shutdown complete or shutdown() and async with would wait forever. + if not self._started: + self._shutdown_complete_event.set() + raise if self._started: raise RuntimeError("Already started") diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index a7fef48c9..eca0b8840 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -283,6 +283,36 @@ async def test_worker_validate_fail(client: Client, env: WorkflowEnvironment): assert str(err.value).startswith("Worker validation failed") +async def test_worker_validate_fail_does_not_hang_shutdown( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping_v1: + pytest.skip("Java test server does not appear to fail on invalid namespace") + config = client.config() + config["namespace"] = "does-not-exist" + client = Client(**config) + + # Calling shutdown() after run() raised should return right away + worker = Worker( + client, task_queue=f"tq-{uuid.uuid4()}", workflows=[NeverRunWorkflow] + ) + with pytest.raises(RuntimeError, match="^Worker validation failed"): + await worker.run() + await asyncio.wait_for(worker.shutdown(), 10) + assert worker.is_shutdown + + # async with should raise the validation error instead of hanging + async def run_in_context() -> None: + async with Worker( + client, task_queue=f"tq-{uuid.uuid4()}", workflows=[NeverRunWorkflow] + ): + # The worker cancels this when it fails + await asyncio.Future() + + with pytest.raises(RuntimeError, match="^Worker validation failed"): + await asyncio.wait_for(run_in_context(), 10) + + async def test_can_run_resource_based_worker(client: Client): tuner = WorkerTuner.create_resource_based( target_memory_usage=0.5,