fix(prefect): address roborev review on #286#287
Merged
Conversation
Roborev #2160 raised three findings on PR #286 / commit 6cdb76f. Two are addressed here; the third is a documented false-positive skip. **Medium — caching.py: contradictory TASK_SOURCE-rationale comment blocks (lines 157-171 vs 173-188).** The pre-existing "Composed cache policies" block explained WHY we compose `CacheKeyFnPolicy` with `TASK_SOURCE`. The new block immediately below said we deliberately do NOT compose. Consolidated into a single block under "Cache policies" that explains both the empirical hazard with `+ TASK_SOURCE` and the trade-off (no auto-invalidation on task body changes; 24h TTL bounds the staleness window). **Medium — file_persistence.py OOM risk on multi-GB files.** `persist_file` reads the full file into RAM before handing it to the Prefect block API. Prefect 3's `LocalFileSystem.write_path` / `read_path` are bytes-in / bytes-out (no streaming variant), so on a multi-GB CSV ingestion a single task completion could OOM the worker. Added `DATAPUSHER_PLUS_MAX_PERSIST_FILE_MB` (default 512 MiB); files above the cap skip persistence with a debug log and fall back to the pre-caching same-run-only behaviour (cross-run cache hits surface a clear `FileNotFoundError` from the downstream stage rather than OOM). Set the env var to `0` to disable the cap entirely. Two new tests in `test_file_persistence.py` cover the cap-trips-and-skips path and the cap=0-disabled-and-persists path. **Low — analyze_task csv_path_key double-write (skipped, false positive).** Reviewer claimed `analyze_task`'s `ctx.tmp` is the same file `validate_task` already persisted, making the analyze-stage persist a wasted duplicate. This is incorrect: `AnalysisStage` transforms `ctx.tmp` via `context.update_tmp(...)` at three sites in `stages/analysis.py` (lines 180 `qsv_safenames_csv`, 526 `qsv_slice_csv`, 563 `qsv_applydp_csv`). The analyze:csv slot holds genuinely different bytes from validate:csv and is needed so a cross-run cache hit on `AnalyzeResult` has a real key to restore from. Logged this in the roborev close comment. Verified in `dpp-test`: 62/62 pass (60 prior + 2 new size-cap tests). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
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 roborev review #2160 on the just-merged PR #286 (commit
6cdb76f). Two of the three findings are fixed; the third is a documented false-positive skip.Fixes
Medium —
caching.py: contradictory comment blocks.The pre-existing "Composed cache policies" comment block explained WHY we compose
CacheKeyFnPolicywithTASK_SOURCE. The new block right below said we deliberately do NOT compose. Consolidated into a single block that explains both the empirical hazard with+ TASK_SOURCE(Prefect 3.7'sCompoundCachePolicy.compute_keydroppingNonecontributions and collapsing to a TASK_SOURCE-only key) and the trade-off.Medium —
file_persistence.py: OOM risk on multi-GB files.persist_filereads the full file into RAM before handing it to the Prefect block (Prefect 3'sLocalFileSystem.write_path/read_pathare bytes-in / bytes-out, no streaming variant), so on multi-GB CSV ingestions a single task completion could OOM the worker. AddedDATAPUSHER_PLUS_MAX_PERSIST_FILE_MB(default 512 MiB). Files above the cap skip persistence with a debug log and fall back to pre-caching same-run-only behaviour (cross-run cache hits would surface a clearFileNotFoundErrorinstead of OOM). Set env var to0to disable the cap entirely.Skipped (false positive)
Low —
analyze_taskcsv_path_key"double-write". Reviewer claimedanalyze_task'sctx.tmpis the same filevalidate_taskalready persisted, so persisting under…:analyze:csvis wasted. This is incorrect:AnalysisStagetransformsctx.tmpviacontext.update_tmp(...)at three sites instages/analysis.py(line 180qsv_safenames_csv, line 526qsv_slice_csv, line 563qsv_applydp_csv). The analyze:csv slot holds genuinely different bytes from validate:csv and is needed so a cross-run cache hit onAnalyzeResulthas a real key to restore from. Logged in the roborev close comment.Tests
Two new cases in
tests/test_file_persistence.py:test_persist_skips_files_above_size_cap— file above cap returnsNone, never reaches the block.test_persist_cap_zero_disables_size_gate— cap=0 disables the gate.Verified in the
dpp-testckan-dev container: 62/62 unit tests pass (60 prior + 2 new).Test plan
ci.yml(DataPusher+ Integration CI) — quick-dir matrix still green (the size cap is well above the test files; no behaviour change for normal-sized inputs).🤖 Generated with Claude Code