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

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]