fix(dagster-async-executor): bound event receive so delayed retries don't hang#327
Merged
cmpadden merged 4 commits intoJun 29, 2026
Conversation
…test 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
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
…eceive() 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
…chema 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
cmpadden
approved these changes
Jun 29, 2026
Contributor
There was a problem hiding this comment.
Thank you @Brandon-Peebles-Zocdoc
Released here - https://github.com/dagster-io/community-integrations/actions/runs/28377658665
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
AsyncExecutorcan cause a run to be stuck inSTARTEDindefinitely when the last in-flightstep is parked for a delayed retry.
_async_execution_iteratordrives scheduling off the event stream and blocksunconditionally on
recv_stream.receive(). When the final in-flight step emitsSTEP_UP_FOR_RETRYunder aRetryPolicy(delay > 0), Dagster parks it inActiveExecution._waiting_to_retryuntil the backoff elapses. During that window:get_steps_to_execute()returns[](the step is promoted to executable inside_update()only oncetime.time() >= at_time), so no worker is started;to send another event;
recv_stream.receive()blocks forever, whileactive.is_completestaysFalsebecause it counts
_waiting_to_retry.The stock
in_process/multiprocessexecutors avoid this by polling withactive.sleep_til_ready()/sleep_interval()instead of blocking on an event.Fix
Bound the receive with
anyio.move_on_after(active.sleep_interval())when somethingtime-based is pending, and re-poll on timeout so
get_steps_to_execute()→_update()promotes the now-ready retry step. Whensleep_interval() == 0(nothingparked),
move_on_after(None)is an unbounded wait — identical to the previousbehavior on the hot path.
This mirrors the stock executors'
sleep_til_ready()polling. The timeout also boundsreceive()on the pending concurrency-claim path, which is likewise strictly closer tothe stock executors' behavior. The removed
try/except anyio.EndOfStream: raisewas ano-op (
EndOfStreamcannot fire while the orchestrator'ssend_streamis open); it isstill handled by
_drain_eventsafter the loop.Test
test_retry_policy.py: a single-op job that fails on its first attempt and succeeds onthe retry under
RetryPolicy(max_retries=1, delay=0.5). A single op is the minimalreproducer because it is by definition the last in-flight step. Guarded with
@pytest.mark.timeout(30)(viapytest-timeout, added as a dev dependency) so thepre-fix hang surfaces as a fast failure instead of wedging the suite. The test asserts
positive evidence that the retry actually ran (the op's returned attempt count and a
per-test attempts file), not merely
result.success.Changes
executor.py— boundrecv_stream.receive()withanyio.move_on_after(sleep_interval()), re-poll on timeout.test_retry_policy.py— new timeout-guarded regression test.pyproject.toml/uv.lock— addpytest-timeoutdev dependency (and de-dupe a duplicatepytestentry).