From cde1c560ffe402b7765e169066f1daa0204393a3 Mon Sep 17 00:00:00 2001 From: brandon-peebles-zocdoc Date: Mon, 22 Jun 2026 18:06:59 -0400 Subject: [PATCH 1/4] chore(dagster-async-executor): add pytest-timeout dev dep, de-dupe pytest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove the duplicate `pytest` entry from `[dependency-groups].dev` and add `pytest-timeout` (needed by upcoming regression test for the delayed retry hang). Regenerated `uv.lock` — resolves to pytest-timeout v2.4.0. Generated with AI Co-Authored-By: Claude Code --- libraries/dagster-async-executor/pyproject.toml | 2 +- libraries/dagster-async-executor/uv.lock | 16 +++++++++++++++- 2 files changed, 16 insertions(+), 2 deletions(-) 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" From 6ad2c8295904a076b9cca31f9c471c99e7ef1d5b Mon Sep 17 00:00:00 2001 From: brandon-peebles-zocdoc Date: Mon, 22 Jun 2026 18:10:57 -0400 Subject: [PATCH 2/4] test(dagster-async-executor): add regression test for delayed-retry hang Add test_retry_policy.py with a 30-second pytest-timeout guard that reproduces the hang caused by recv_stream.receive() blocking forever when the last step is parked in _waiting_to_retry under a delayed RetryPolicy. Pre-fix: test times out via pytest-timeout (RED). Generated with AI Co-Authored-By: Claude Code --- .../test_retry_policy.py | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 libraries/dagster-async-executor/dagster_async_executor_tests/test_retry_policy.py 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..2c2347da --- /dev/null +++ b/libraries/dagster-async-executor/dagster_async_executor_tests/test_retry_policy.py @@ -0,0 +1,46 @@ +import os +import tempfile + +import dagster as dg +import pytest + +from dagster_async_executor import async_executor + +_ATTEMPTS_FILE = os.path.join(tempfile.gettempdir(), "async_executor_retry_attempts") + + +def fail_once_then_succeed_job() -> dg.JobDefinition: + @dg.op(retry_policy=dg.RetryPolicy(max_retries=1, delay=0.5)) + def fail_once_then_succeed() -> int: + 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(): + fail_once_then_succeed() + + return retry_job + + +@pytest.mark.timeout(30) +def test_delayed_retry_policy_does_not_hang(): + 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, + ) 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 From 29ce25acfc3e9665d395dd69a0648a5726de5072 Mon Sep 17 00:00:00 2001 From: brandon-peebles-zocdoc Date: Mon, 22 Jun 2026 18:11:09 -0400 Subject: [PATCH 3/4] fix(dagster-async-executor): resolve delayed-retry hang via bounded receive() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the last in-flight step emits STEP_UP_FOR_RETRY under a RetryPolicy with delay > 0, the step is parked in ActiveExecution._waiting_to_retry. No worker is alive to emit an event, so recv_stream.receive() blocks forever while active.is_complete stays False — the run wedges in STARTED. Fix: wrap receive() in anyio.move_on_after(sleep_interval()) so the wait is bounded by the retry backoff. When the timeout fires (scope.cancelled_caught), continue to re-poll, which runs get_steps_to_execute() → _update() to promote the now-ready retry step. When sleep_interval() == 0 (hot path), we pass None to move_on_after, preserving exact current behavior. The removed try/except anyio.EndOfStream: raise was a no-op; EndOfStream is still handled by _drain_events after the loop. Mirroring stock in_process/multiprocess executors' sleep_til_ready() pattern. Generated with AI Co-Authored-By: Claude Code --- .../dagster_async_executor/executor.py | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) 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) From e410b49f15386b9713fcc162364de45b3b2bc410 Mon Sep 17 00:00:00 2001 From: brandon-peebles-zocdoc Date: Mon, 22 Jun 2026 18:31:30 -0400 Subject: [PATCH 4/4] test(dagster-async-executor): make retry test xdist-safe via config_schema Replace fixed global temp filename with per-test TemporaryDirectory and thread the attempts-file path into the op through config_schema + run_config, eliminating shared-state races under pytest-xdist parallel runs. Generated with AI Co-Authored-By: Claude Code --- .../test_retry_policy.py | 58 +++++++++++-------- 1 file changed, 34 insertions(+), 24 deletions(-) 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 index 2c2347da..cd1bd976 100644 --- 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 @@ -1,4 +1,3 @@ -import os import tempfile import dagster as dg @@ -6,41 +5,52 @@ from dagster_async_executor import async_executor -_ATTEMPTS_FILE = os.path.join(tempfile.gettempdir(), "async_executor_retry_attempts") - def fail_once_then_succeed_job() -> dg.JobDefinition: - @dg.op(retry_policy=dg.RetryPolicy(max_retries=1, delay=0.5)) - def fail_once_then_succeed() -> int: - with open(_ATTEMPTS_FILE) as f: + @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: + 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(): + def retry_job() -> None: fail_once_then_succeed() return retry_job @pytest.mark.timeout(30) -def test_delayed_retry_policy_does_not_hang(): - 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, - ) 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 +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