feat(prefect): re-enable content caching with stable file persistence#286
Merged
Conversation
Re-enables `cache_policy=...` on the read-only stages (download, format_convert, validate, analyze) and fixes the file-path blocker that forced caching off in the first place (the "next step" called out in `caching.py`'s pre-existing docstring). ## Why it was off Each per-stage result dataclass carries working file paths (`downloaded_path`, `csv_path`, `quarantine_csv_path`) that point into a per-flow-run `TemporaryDirectory`. Without persistence, a cached result from an earlier run referenced a tempdir that no longer exists, so a cache hit would either error out or silently mis-read stale data downstream. ## What this PR adds * **`jobs/file_persistence.py`** (new): `persist_file(local_path, key)` writes a local file to the result-storage block under a stable key; `restore_file(key, dest_path)` reads it back. Both degrade gracefully (return `None` / `False`) when the block is unavailable. * **Result dataclasses gain `*_path_key` fields** (`DownloadResult.downloaded_path_key`, `ConvertResult.csv_path_key`, `ValidateResult.csv_path_key`, `ValidateResult.quarantine_csv_path_key`, `AnalyzeResult.csv_path_key`). Optional + defaulting to `None`, so cached results from older runs predating this field deserialize cleanly and just behave as "no cross-run rehydration". * **Task bodies persist their working file at completion** via a tiny `_persist_stage_file(local_path, file_hash, stage, slot)` helper. Skips silently when the storage block is unavailable; same-run chains continue to work either way. * **`_apply_result` routes every file path through a new `_resolve_or_restore(local_path, storage_key, dest_dir)` helper**: use the local path if it exists (same-run); otherwise restore into the current run's tempdir from the persisted key; otherwise return the original path unchanged so the downstream stage raises a clear `FileNotFoundError` instead of appearing to succeed on stale data. * **Re-enabled `cache_policy=DOWNLOAD_CACHE_POLICY` / `CONTENT_CACHE_POLICY`** on the four read-only tasks, with `cache_expiration=DEFAULT_CACHE_EXPIRATION` (24 hours by default). The mutating tasks (database, indexing, formula, metadata) intentionally still do NOT cache — caching their results would skip the side effect. ## A correctness fix in caching.py The pre-existing `CacheKeyFnPolicy(cache_key_fn=...) + TASK_SOURCE` composition has a hazard verified empirically against Prefect 3.7: Prefect's `CompoundCachePolicy.compute_key` drops `None` contributions and hashes the rest, so when `cache_key_fn` returns `None` (no `file_hash` yet, or operator set `ignore_hash=True`), the compound key collapses to just the TASK_SOURCE hash — identical for every invocation of the same task source regardless of parameters. Two unrelated runs with no content key would hit each other's cache. Fix: drop the `+ TASK_SOURCE` composition. Cost is that a DP+ upgrade changing a task body no longer auto-invalidates older cached output; the 24-hour default TTL bounds the staleness window, and operators wanting fresh output post-upgrade can clear the storage block or shorten `DATAPUSHER_PLUS_CACHE_TTL_HOURS`. Documented. ## Tests New `tests/test_file_persistence.py` covers `persist_file`, `restore_file`, and `_resolve_or_restore`: - Round-trip persistence + restore - Graceful degradation when block unavailable - Graceful degradation when local file missing - `_resolve_or_restore` returns local path when file exists - `_resolve_or_restore` restores from storage when path missing - `_resolve_or_restore` returns original (broken) path when neither works, so downstream gets a clear FileNotFoundError - Pass-through for None / empty inputs Verified in `dpp-test`: **60/60 unit tests pass** (52 prior + 8 new). Tracking: closes the result-caching item on #282. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This was referenced May 16, 2026
jqnatividad
added a commit
that referenced
this pull request
May 16, 2026
fix(prefect): address roborev review on #286
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
Re-enables
cache_policy=on the read-only stages and clears the file-path blocker that forced caching off in the first place (the "next step" called out incaching.py's pre-existing docstring). Closes the result-caching item on #282.Why caching was off
Each per-stage result dataclass carries working file paths that point into a per-run
TemporaryDirectory. Without persistence, a cached result from an earlier run referenced a tempdir that no longer exists, so a cache hit would either error or silently misread stale data downstream.What this PR adds
jobs/file_persistence.py—persist_file(local_path, key)+restore_file(key, dest_path), both with graceful no-op fallback when the storage block isn't available.*_path_keyfields (back-compat: cached results predating the field deserialize cleanly)._persist_stage_file(...)helper using the content-hash + stage + slot as a stable key._apply_resultroutes every file path through a new_resolve_or_restore(...)helper: local path if it still exists (same-run); otherwise restore from storage into the current tempdir (cross-run cache hit); otherwise pass the original path through so the downstream gets a clearFileNotFoundErrorinstead of silent staleness.cache_policy=re-enabled ondownload_task,format_convert_task,validate_task,analyze_taskwithcache_expiration=DEFAULT_CACHE_EXPIRATION(24h default). Mutating tasks (database, indexing, formula, metadata) intentionally still do not cache — caching the result would skip the side effect.Correctness fix in
caching.pyThe pre-existing
CacheKeyFnPolicy(cache_key_fn=...) + TASK_SOURCEcomposition has a subtle hazard verified empirically against Prefect 3.7: whencache_key_fnreturnsNone(nofile_hashyet, or operator passedignore_hash=True), Prefect'sCompoundCachePolicy.compute_keydrops theNonecontribution and hashes the rest — so the compound key collapses to just the TASK_SOURCE hash, identical for every invocation of the same task source regardless of parameters. Two unrelated runs with no content key would hit each other's cache.Fix: drop the
+ TASK_SOURCEcomposition. The cost is that a DP+ upgrade changing a task body no longer auto-invalidates older cached output; the 24-hour default TTL bounds the staleness window. Operators wanting fresh output post-upgrade can clear the storage block or shortenDATAPUSHER_PLUS_CACHE_TTL_HOURS. Documented incaching.py.Tests
New
tests/test_file_persistence.pywith 8 cases covering persist/restore round-trip, graceful degradation paths, and the three_resolve_or_restorebranches (local-exists / storage-restore / fall-through to clear error).Verified locally in the
dpp-testckan-dev container: 60/60 unit tests pass (52 prior + 8 new).Test plan
ci.yml(DataPusher+ Integration CI) — the 7-file quick-dir matrix exercises the same task chain end-to-end with a real Prefect server + worker. Caching hits should be harmless on the first run (cache empty) and should make subsequent runs of the same files faster.🤖 Generated with Claude Code