diff --git a/libraries/dagster-async-executor/dagster_async_executor/executor.py b/libraries/dagster-async-executor/dagster_async_executor/executor.py index 0cee5c09..213e7402 100644 --- a/libraries/dagster-async-executor/dagster_async_executor/executor.py +++ b/libraries/dagster-async-executor/dagster_async_executor/executor.py @@ -238,10 +238,20 @@ async def _async_execution_iterator( send_stream.clone(), ) - try: + # When the only remaining work is parked in ActiveExecution + # (a step in _waiting_to_retry under a delayed RetryPolicy, or + # a pending concurrency claim), no worker is alive to emit an + # event, so receive() would block forever while is_complete + # stays False. Bound the wait by sleep_interval() and re-poll, + # mirroring stock executors' sleep_til_ready() behavior, so the + # retry-ready step gets promoted by get_steps_to_execute(). + sleep_interval = active.sleep_interval() + with anyio.move_on_after( + sleep_interval if sleep_interval > 0 else None + ) as scope: event = await recv_stream.receive() - except anyio.EndOfStream: - raise + if scope.cancelled_caught: + continue yield event active.handle_event(event) diff --git a/libraries/dagster-async-executor/dagster_async_executor_tests/test_retry_policy.py b/libraries/dagster-async-executor/dagster_async_executor_tests/test_retry_policy.py new file mode 100644 index 00000000..cd1bd976 --- /dev/null +++ b/libraries/dagster-async-executor/dagster_async_executor_tests/test_retry_policy.py @@ -0,0 +1,56 @@ +import tempfile + +import dagster as dg +import pytest + +from dagster_async_executor import async_executor + + +def fail_once_then_succeed_job() -> dg.JobDefinition: + @dg.op( + retry_policy=dg.RetryPolicy(max_retries=1, delay=0.5), + config_schema={"attempts_file": str}, + ) + def fail_once_then_succeed(context: dg.OpExecutionContext) -> int: + attempts_file: str = context.op_config["attempts_file"] + with open(attempts_file) as f: + attempts = int(f.read() or "0") + with open(attempts_file, "w") as f: + f.write(str(attempts + 1)) + if attempts == 0: + raise RuntimeError("intentional first-attempt failure") + return attempts + + @dg.job(executor_def=async_executor) + def retry_job() -> None: + fail_once_then_succeed() + + return retry_job + + +@pytest.mark.timeout(30) +def test_delayed_retry_policy_does_not_hang() -> None: + with tempfile.TemporaryDirectory() as tmpdir: + attempts_file = f"{tmpdir}/attempts" + with open(attempts_file, "w") as f: + f.write("0") + with ( + dg.instance_for_test() as instance, + dg.execute_job( + dg.reconstructable(fail_once_then_succeed_job), + instance=instance, + run_config={ + "ops": { + "fail_once_then_succeed": { + "config": {"attempts_file": attempts_file} + } + } + }, + ) as result, + ): + assert result.success + # Prove the delayed retry actually ran (not a first-attempt success): + # the op returns `attempts`, which is 1 only on the retried attempt. + assert result.output_for_node("fail_once_then_succeed") == 1 + with open(attempts_file) as f: + assert f.read() == "2" # op compute ran exactly twice diff --git a/libraries/dagster-async-executor/pyproject.toml b/libraries/dagster-async-executor/pyproject.toml index 80e97620..12bc06d9 100644 --- a/libraries/dagster-async-executor/pyproject.toml +++ b/libraries/dagster-async-executor/pyproject.toml @@ -13,8 +13,8 @@ dynamic = ["version"] dev = [ "ruff", "pytest", + "pytest-timeout", "ty==0.0.37", - "pytest", ] [build-system] diff --git a/libraries/dagster-async-executor/uv.lock b/libraries/dagster-async-executor/uv.lock index f440dc04..d0e36897 100644 --- a/libraries/dagster-async-executor/uv.lock +++ b/libraries/dagster-async-executor/uv.lock @@ -252,6 +252,7 @@ dependencies = [ [package.dev-dependencies] dev = [ { name = "pytest" }, + { name = "pytest-timeout" }, { name = "ruff" }, { name = "ty" }, ] @@ -265,6 +266,7 @@ requires-dist = [ [package.metadata.requires-dev] dev = [ { name = "pytest" }, + { name = "pytest-timeout" }, { name = "ruff" }, { name = "ty", specifier = "==0.0.37" }, ] @@ -309,7 +311,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -870,6 +872,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" }, ] +[[package]] +name = "pytest-timeout" +version = "2.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ac/82/4c9ecabab13363e72d880f2fb504c5f750433b2b6f16e99f4ec21ada284c/pytest_timeout-2.4.0.tar.gz", hash = "sha256:7e68e90b01f9eff71332b25001f85c75495fc4e3a836701876183c4bcfd0540a", size = 17973, upload-time = "2025-05-05T19:44:34.99Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fa/b6/3127540ecdf1464a00e5a01ee60a1b09175f6913f0644ac748494d9c4b21/pytest_timeout-2.4.0-py3-none-any.whl", hash = "sha256:c42667e5cdadb151aeb5b26d114aff6bdf5a907f176a007a30b940d3d865b5c2", size = 14382, upload-time = "2025-05-05T19:44:33.502Z" }, +] + [[package]] name = "python-dotenv" version = "1.2.2"