refactor(prefect): runtime-resolve flow-timeout + download retries#285
Merged
Conversation
Addresses the "import-time constants" follow-up on #282: two of the three knobs that were baked in at module import are now re-resolved per flow run, so env / ckan.ini changes take effect on the very next run even on work-pool types that reuse a long-lived interpreter (the default `process` worker re-imports per run, so behavior there is unchanged). **Download retries (`DATAPUSHER_PLUS_DOWNLOAD_RETRIES` / `ckanext.datapusher_plus.download_retries`)**: removed `retries=` from the `@task` decorator on `download_task`; the flow body now applies it via `download_task.with_options(retries=download_retries)(...)`, resolving the count via the existing `_resolve_int` helper at flow start. The `_TASK_RETRY_DOWNLOAD` module constant is gone. **Flow timeout (`DATAPUSHER_PLUS_FLOW_TIMEOUT_SECONDS` / `ckanext.datapusher_plus.flow_timeout`)**: kept the `@flow` decorator's `timeout_seconds=_FLOW_TIMEOUT_SECONDS` as Prefect's hard-stop ceiling (necessary because Prefect's decorator-based timeout is the only thing that can kill a stage hung mid-execution). Added a complementary soft-deadline check `_check_deadline()` that runs between stages using a runtime-resolved value: so a *shorter* deadline picks up immediately on any worker type, while raising the ceiling above the import-time value still requires a restart on interpreter-reusing pools. **Database retries**: intentionally *not* moved. `Task.with_options()` in Prefect 3.x preserves `on_completion` / `on_failure` / `on_running` but does *not* preserve `on_rollback_hooks`. `database_task` carries the rollback hook that drops the partial datastore table on transactional failure, and silently breaking that contract would be worse than the import-time limitation. Both the constants block and the `@task` decorator comment now spell this out. The full audit: with_options(retries=N) call site -> task | safe? --------------------------------------------+------ download_task | yes (no on_rollback) database_task | no (has on_rollback) indexing/formula/metadata_task | no (have on_rollback, | not retry-eligible | anyway) Tests: two new cases in `test_prefect_flow.py` — 1. `test_flow_resolves_retry_counts_per_run`: spies on `download_task.with_options` and asserts it's called with the env- resolved retry count. 2. `test_flow_aborts_when_soft_deadline_exceeded`: sets the deadline env var to "0" and asserts `mark_job_as_errored` fires with the deadline message before the flow propagates the JobError. (Uses 0 instead of patching `time.monotonic` because Prefect's flow infrastructure also calls `monotonic` during setup, which exhausts a small fake-iter before the flow body runs.) Verified locally in `dpp-test`: 52/52 (50 prior + 2 new). Tracking: #282. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This was referenced May 16, 2026
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
1 participant
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
Addresses the "import-time constants" follow-up on #282. Two of the three knobs that were baked into the
@flow/@taskdecorators at module import are now re-resolved per flow run, so env /ckan.inichanges take effect on the very next run — even on work-pool types that reuse a long-lived interpreter. (The defaultprocessworker re-imports per run anyway, so behavior there is unchanged.)What moved to runtime resolution
DATAPUSHER_PLUS_DOWNLOAD_RETRIES/ckanext.datapusher_plus.download_retries): the flow body now applies the count viadownload_task.with_options(retries=download_retries)(...)instead of the@task(retries=…)decorator.DATAPUSHER_PLUS_FLOW_TIMEOUT_SECONDS/ckanext.datapusher_plus.flow_timeout): the@flowdecorator'stimeout_secondsstays as Prefect's hard-stop ceiling (the only thing that can kill a stage hung mid-execution). Added a soft_check_deadline()between stages that uses a runtime-resolved value. A shorter deadline takes effect immediately on any worker type; raising it above the import-time ceiling still requires a restart on interpreter-reusing pools.What stayed import-time, and why
_TASK_RETRY_DATABASE):Task.with_options()in Prefect 3.x preserveson_completion/on_failure/on_runninghooks but does not preserveon_rollback_hooks.database_taskcarries the rollback hook that drops the partial datastore table on transactional failure (@database_task.on_rollback). Silently breaking that contract for a per-run config refresh would be a net loss. Both the constants block and the@taskdecorator comment now spell this out.Verified empirically in
dpp-test:Tests
Two new cases in
tests/test_prefect_flow.py:test_flow_resolves_retry_counts_per_run— spies ondownload_task.with_optionsand asserts it's called with the env-resolved retry count.test_flow_aborts_when_soft_deadline_exceeded— sets the deadline env var to"0"and assertsmark_job_as_erroredfires with the deadline message before the flow propagatesJobError.Verified locally in the
dpp-testckan-dev container: 52/52 unit tests pass (50 prior + 2 new).Test plan
ci.yml(DataPusher+ Integration CI) — 7-file quick-dir matrix still passes; in particular the rollback path ondatabase_taskstill fires (regression test for the asymmetry that made me revert the database-retries change).🤖 Generated with Claude Code