fix(#311): don't apply DownloadResult.resource on rehydrate (cross-resource cache leak)

[jqnatividad]

Summary

Closes #311.

When the same file bytes are uploaded as two different CKAN resources, the second flow's database_task crashed with:

Postgres COPY failed: relation "<new_resource_id>" does not exist

Caught during PR #309's post-merge smoke test, filed as #311, now fixed and end-to-end verified.

Root cause

analyze_task / validate_task / format_convert_task cache by file content (dpp:{task_name}:{file_hash}), so the second flow cache-hits and returns the first flow's cached result chain — which still points all the way back to the first flow's DownloadResult carrying the first flow's full resource dict (with the OLD id).

database_task runs fresh (mutating stage, no cache). But _stage_run → rehydrate walks the cached chain, and the DownloadResult branch in _apply_result did ctx.resource = dict(result.resource) — wholesale replacing the current flow's resource dict (set by _build_runtime_context from CKAN) with the cached one. From that point:

• DatabaseStage._create_datastore_table used context.resource["id"] (OLD resource_id, from the cached dict) for datastore_create — no-op, the table already exists from the first run.
• DatabaseStage._copy_data used context.resource_id (set fresh by _build_runtime_context, NOT touched by rehydrate) for TRUNCATE — the NEW resource_id's table doesn't exist, TRUNCATE fails with UndefinedTable.

Fix

Remove ctx.resource = dict(result.resource) from the DownloadResult branch of _apply_result. Resource identity is per-flow-run state, not per-file-content state — _build_runtime_context already fetched the current flow's resource from CKAN, and that's the source of truth.

What still works correctly:

• ctx.file_hash / ctx.resource_url / ctx.content_length / ctx.tmp ARE still applied — those are genuinely content-derived.
• The download stage's only mutation of ctx.resource (["hash"] = file_hash, download.py:129) happens for the current resource dict when the body runs. On a download_task cache hit (same resource_id), the body is skipped, but the metadata stage drops file hash on re-fetch — persisted resource hash is always empty #310 fix in MetadataStage already restores ctx.resource["hash"] from ctx.file_hash after its own re-fetch.

Test plan

• 213/213 Python unit tests pass on dpp-test (was 209; +4 new + 1 existing updated for the new contract).
• New tests/test_rehydrate_resource_identity.py regression suite:
  • Central Per-stage file cache skips Analysis on identical content but Database expects datastore_create's table #311 contract (ctx.resource not overwritten).
  • Content fields still propagated (the cache's reason for existing).
  • Multi-layer chain (the realistic bug case via AnalyzeResult.upstream).
  • file_hash propagation route via ctx.file_hash (not the resource dict).
• Existing test_rehydrate_reconstitutes_context_from_nested_results updated to assert the new contract with documentation explaining why.
• End-to-end smoke on the integration stack: uploaded the same 8-row CSV as two different resources back-to-back. First flow took 18s (full pipeline), second took 10s (cache hits on validate/analyze, no crash). Both ended with datastore_active=True, identical BLAKE3 hash, and queryable datastore data with correct rows. Pre-fix, the second flow crashed with relation does not exist.
• Integration test in CI.

🤖 Generated with Claude Code

[Copilot]

Pull request overview

Fixes a cross-resource cache leak during Prefect result-chain rehydration by ensuring cached DownloadResult.resource does not overwrite the current flow run’s CKAN resource identity (preventing wrong resource_id from propagating into database_task).

Changes:

• Stop applying DownloadResult.resource onto ctx.resource during rehydrate, preserving per-flow-run resource identity.
• Update existing rehydration unit test to assert the new “resource identity is not overwritten” contract.
• Add a focused regression test suite covering the #311 contract and confirming content-derived fields still propagate through nested cached result chains.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

File	Description
ckanext/datapusher_plus/jobs/runtime_context.py	Prevents rehydrate from clobbering ctx.resource from cached DownloadResult across flows.
tests/test_prefect_flow.py	Updates rehydration test expectations to reflect the new resource-identity contract.
tests/test_rehydrate_resource_identity.py	Adds regression coverage for cross-resource cache-hit rehydration behavior (#311).

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.