fix(dagster-async-executor): bound event receive so delayed retries don't hang (#327)

* chore(dagster-async-executor): add pytest-timeout dev dep, de-dupe pytest

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

* 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

* fix(dagster-async-executor): resolve delayed-retry hang via bounded receive()

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

* 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

Author: Brandon-Peebles-Zocdoc

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)

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

libraries/dagster-async-executor/pyproject.toml

@@ -13,8 +13,8 @@ dynamic = ["version"]
 dev = [
     "ruff",
     "pytest",
+    "pytest-timeout",
     "ty==0.0.37",
-    "pytest",
 ]
 
 [build-system]