From 3f00661974c50b7cb33ef2bf46b32d92419db68c Mon Sep 17 00:00:00 2001 From: Sagar Kharal Date: Fri, 11 Sep 2026 14:58:57 +0530 Subject: [PATCH] fix(checks): validate canonical CRCs once at lane entry with memoization Every post-sync read ran with chunk CRC validation off, so a canonical that decayed on disk after sync was re-certified by the lanes that exist to judge it: fresh measured findings stamped over bytes the file's own integrity stamp refuses (#474). App.process now verifies the canonical's chunk CRCs once per run at the front door every consuming lane shares, before any step runs. A file that fails its own stamp is refused with the named reason canonical-crc-mismatch: one report field, one framework-owned check_runs row (critical, so the curation views read the episode as unverified), and no check, enrichment, or media step. The refusal covers the meta lane, the relabel lane, and the production meta task an online re-check flows through; a run with no step work to do pays no strict read. The verdict is memoized on the App keyed by path plus a size+mtime witness, so repeated stage runs and retried batches in one process pay for one strict read, and bytes that change under the cache are re-validated rather than trusted. Replays of the same refusal dedupe through the existing run fingerprint, keeping one record per episode. Episode._reader's default and the snapshot verifier are untouched. --- src/hflow/app.py | 141 +++++++++++++++++---- src/hflow/reader.py | 34 +++++ tests/test_canonical_integrity.py | 201 ++++++++++++++++++++++++++++++ 3 files changed, 351 insertions(+), 25 deletions(-) create mode 100644 tests/test_canonical_integrity.py diff --git a/src/hflow/app.py b/src/hflow/app.py index de141d01..89f4512f 100644 --- a/src/hflow/app.py +++ b/src/hflow/app.py @@ -54,7 +54,7 @@ PipelineManifest, StepManifest, ) -from hflow.reader import open_reader +from hflow.reader import open_reader, verify_canonical_integrity from hflow.resample import DerivedSeries from hflow.step_selection import ( ALL_REGISTERED_STEPS, @@ -228,6 +228,12 @@ def _resolve_data_root(data_root: "Path | str | StorageRoot | None") -> "Path | # Published artifacts are recorded as measurements under this prefix, so a # reader can tell "here is where the file went" from an ordinary label. ARTIFACT_MEASUREMENT_KEY_PREFIX = "artifact/" +# The framework-owned step that records a check-lane refusal: one row naming +# why the lane stood down, queryable by its error value. Not registered on +# the App; it exists only on the refusal record (see +# :func:`_canonical_integrity_refusal_run`). +CANONICAL_INTEGRITY_STEP_NAME = "integrity/canonical" +CANONICAL_INTEGRITY_STEP_VERSION = parse_step_version("1") _MEDIA_CONTACT_SHEET_FPS = 0.5 _SYNC_COMPLETION_MARKER_NAME = ".sync-complete.json" @@ -897,11 +903,26 @@ def _check_run_rows(report: "ProcessReport") -> list[CheckRunRow]: enrichment's labels and published artifact keys. One owner, so the collision guard below and the catalog append can never - disagree about what would be written. Artifact keys come from - ``artifact_uris`` rather than the result's declared artifacts: a step whose - artifact failed to publish contributes no key. + disagree about what would be written. A refused episode contributes one + framework-owned row naming the refusal (its ``critical`` flag is what + makes the curation views read the episode as unverified rather than ok). + Artifact keys come from ``artifact_uris`` rather than the result's + declared artifacts: a step whose artifact failed to publish contributes + no key. """ - check_rows = [ + check_rows: list[CheckRunRow] = [] + if report.refusal_reason is not None: + check_rows.append( + CheckRunRow( + check_name=CANONICAL_INTEGRITY_STEP_NAME, + check_version=CANONICAL_INTEGRITY_STEP_VERSION, + critical=True, + status=CheckStatus.ERROR, + duration_s=0.0, + error=report.refusal_reason, + ) + ) + check_rows.extend( CheckRunRow.from_result( check_name=run.check.name, check_version=run.check.version, @@ -912,7 +933,7 @@ def _check_run_rows(report: "ProcessReport") -> list[CheckRunRow]: result=run.result, ) for run in report.checks - ] + ) _raise_if_measurement_keys_claim_artifact_namespace( (row.check_name, key) for row in check_rows for key in row.measurements ) @@ -1092,6 +1113,11 @@ class ProcessReport: # a reused run and a transcoded run are otherwise indistinguishable # without comparing file timestamps. sync_reused: bool = False + # The named reason the canonical episode was refused at the check lane + # entry, or None when it entered clean. Mirrored verbatim on the refusal + # row's error field in the catalog, where downstream tooling filters for + # it by equality. + refusal_reason: str | None = None def check(self, name: str) -> CheckRunReport: """Return the run report for the uniquely named check. @@ -1125,8 +1151,10 @@ def quarantined(self) -> bool: @property def has_errors(self) -> bool: """Whether any enabled check or enrichment failed to execute correctly.""" - return any(run.status is CheckStatus.ERROR for run in self.checks) or any( - run.status is CheckStatus.ERROR for run in self.enrichments + return ( + self.refusal_reason is not None + or any(run.status is CheckStatus.ERROR for run in self.checks) + or any(run.status is CheckStatus.ERROR for run in self.enrichments) ) def _stages_line(self) -> str: @@ -1152,6 +1180,11 @@ def summary(self) -> str: ] if self.sync_reused: lines.append("sync: reused the existing canonical episode (source unchanged)") + if self.refusal_reason is not None: + lines.append( + f"REFUSED: {self.refusal_reason} -- the canonical episode failed its " + "integrity stamp; no checks ran" + ) if self.catalog_entry is not None: record_verb = "recorded" if self.catalog_entry.written else "already recorded" lines.append( @@ -1291,6 +1324,16 @@ def __init__( self.enrichments: list[RegisteredEnrichment] = [] self.derived: list[DerivedChannel] = [] self.transform_override: TransformFunction | None = None + # Front-door integrity verdicts for canonical files, keyed by path and + # held with a size+mtime witness of the bytes the verdict describes. + # Lives on the App because the App is the run context: the same + # episode is opened again by later stage runs and retries in this + # process. The witness matters as much as the cache: decay can happen + # between two opens in one process, and a verdict must never outlive + # the file state it was read from. + self._canonical_integrity_cache: dict[ + str, tuple[tuple[int, int], tuple[bool, str | None]] + ] = {} # Which registrations came from ``default_checks`` rather than from # the pipeline: registering one of these yourself replaces it (that # is how a default gets a gate or a bound parameter), while two USER @@ -1343,6 +1386,28 @@ def _yield_defaults_superseded_by_the_pipeline(self, report: "ProcessReport") -> continue run.outcome = NotRun(SupersededByPipeline(superseded_keys=tuple(superseded_keys))) + def _canonical_integrity_verdict(self, canonical_path: Path) -> "tuple[bool, str | None]": + """The check lane's front-door verdict for one canonical file. + + At most one full CRC read per file state per run: the same episode is + re-opened by later stage runs and retried stage batches in this + process (a metadata backfill after a full run, a replayed META lane), + and re-validating unchanged bytes would pay the pass again for the + same answer. The cached verdict is keyed to the file's size and + mtime, so bytes that changed under the cache are re-validated rather + than trusted -- a verdict describes a file state, not a path. Sync + rewriting the file drops the cached entry outright. + """ + cache_key = str(canonical_path) + file_state = canonical_path.stat() + witness = (file_state.st_size, file_state.st_mtime_ns) + cached = self._canonical_integrity_cache.get(cache_key) + if cached is not None and cached[0] == witness: + return cached[1] + verdict = verify_canonical_integrity(canonical_path) + self._canonical_integrity_cache[cache_key] = (witness, verdict) + return verdict + def _reusable_canonical_episode( self, run_storage_root: StorageRoot, @@ -2324,6 +2389,9 @@ def process( # ones from a different source episode sharing this run dir's stem. if scratch_dir.exists(): shutil.rmtree(scratch_dir) + # So is any cached integrity verdict: it describes bytes that no + # longer exist. + self._canonical_integrity_cache.pop(str(canonical_path), None) else: try: canonical_path = run_storage_root.fetch(canonical_file_name) @@ -2417,6 +2485,43 @@ def process( if Stage.META in enabled_stages else [] ) + enrichments_to_run = ( + [ + registered + for registered in self._ordered_enrichments() + if registered_step_is_selected(registered_step_selection, registered.name) + ] + if Stage.LABELS in enabled_stages + else [] + ) + # The media stage is silently absent on a camera-less episode: + # there is nothing to render, so no row claims otherwise. + media_will_run = ( + Stage.MEDIA in enabled_stages + and bool(canonical_episode.cameras) + and registered_step_is_selected( + registered_step_selection, MEDIA_CONTACT_SHEET_STEP_NAME + ) + ) + if checks_to_run or enrichments_to_run or media_will_run: + # The front door every consuming lane shares: whatever is about + # to spend step work over the canonical's bytes pays one strict + # read first, so a canonical that decayed on disk after sync is + # refused with a named reason instead of measured, labeled, or + # rendered over (#474). A run with no step work to do pays + # nothing. + is_intact, refusal_reason = self._canonical_integrity_verdict(canonical_path) + if not is_intact: + # Refuse the episode with ONE diagnosis: the named reason + # lands on the report and on one catalog row (built with + # the other check rows in _check_run_rows), and no step + # runs. Exact replays of the refusal dedupe through the + # run fingerprint, so retries stay one record. + assert refusal_reason is not None + report.refusal_reason = refusal_reason + checks_to_run = [] + enrichments_to_run = [] + media_will_run = False # Keys already emitted by the pipeline's own steps in this run. # A default that has any key in common with what is here can be # superseded at the top of the loop, before paying its ffmpeg @@ -2452,7 +2557,7 @@ def process( # A default with a registered key pattern: if any pipeline # step has already emitted a key the default would emit, # the default's measurement would be a duplicate and would - # be thrown away by ``_yield_defaults_superseded_by_the_…`` + # be thrown away by ``_yield_defaults_superseded_by_the_...`` # anyway. Skip the ffmpeg work entirely and record the same # superseded reason, with the same key list, as the # post-execution path. Same-parameter wrappers and steps @@ -2543,28 +2648,14 @@ def process( ) if Stage.LABELS in enabled_stages: - for registered_enrichment in self._ordered_enrichments(): - if not registered_step_is_selected( - registered_step_selection, registered_enrichment.name - ): - continue + for registered_enrichment in enrichments_to_run: report.enrichments.append( _execute_enrichment( registered_enrichment, canonical_episode, quarantine_skip ) ) - # The media stage is silently absent on a camera-less episode: - # there is nothing to render, so no row claims otherwise. - if ( - Stage.MEDIA in enabled_stages - and canonical_episode.cameras - and ( - registered_step_is_selected( - registered_step_selection, MEDIA_CONTACT_SHEET_STEP_NAME - ) - ) - ): + if media_will_run: media_directory = run_dir / "media" def render_contact_sheets(media_episode: Episode) -> EnrichmentResult: diff --git a/src/hflow/reader.py b/src/hflow/reader.py index 29f2d77e..d4706ccf 100644 --- a/src/hflow/reader.py +++ b/src/hflow/reader.py @@ -28,12 +28,18 @@ import numpy as np from mcap.reader import McapReader, make_reader from mcap.records import Attachment +from mcap.stream_reader import CRCValidationError logger = logging.getLogger(__name__) DEFAULT_BATCH_MAX_MESSAGES = 1024 DEFAULT_BATCH_MAX_BYTES = 32 * 1024 * 1024 +# The named reason a file fails its own integrity stamp, returned by +# :func:`verify_canonical_integrity` and recorded on the check lane's refusal +# row, so downstream tooling can filter for damaged canonicals by exact value. +CANONICAL_CRC_MISMATCH_REASON = "canonical-crc-mismatch" + @dataclass(frozen=True) class TopicInfo: @@ -304,3 +310,31 @@ def open_reader(path: Path | str, *, validate_crcs: bool = False) -> EpisodeRead ``hflow.transform``). """ return PythonMcapEpisodeReader(path, validate_crcs=validate_crcs) + + +def verify_canonical_integrity(path: Path | str) -> tuple[bool, str | None]: + """Validate one episode file's chunk CRCs with a strict full read. + + The check lane's front door. ``Episode`` reads run with CRC validation + off (the reader docstring's trust argument covers bytes identified by + content hash at sync time), so a canonical that decayed on disk after + sync would otherwise be measured by checks as if it were intact. This + re-opens the file the strict way and reads every message, which forces + the chunk CRC pass over exactly the bytes the checks are about to + certify. + + Returns ``(is_valid, reason)``: ``(True, None)`` when every chunk + matches its stored CRC, and ``(False, CANONICAL_CRC_MISMATCH_REASON)`` + when the file refuses its own integrity stamp. ``CRCValidationError`` + is caught by its precise type -- it subclasses ``ValueError``, and the + broader type would also swallow unrelated boundary errors this function + must not answer for. + """ + with Path(path).open("rb") as stream: + try: + reader = make_reader(stream, validate_crcs=True) + for _schema, _channel, _message in reader.iter_messages(log_time_order=False): + pass + except CRCValidationError: + return (False, CANONICAL_CRC_MISMATCH_REASON) + return (True, None) diff --git a/tests/test_canonical_integrity.py b/tests/test_canonical_integrity.py new file mode 100644 index 00000000..093221f7 --- /dev/null +++ b/tests/test_canonical_integrity.py @@ -0,0 +1,201 @@ +"""The check lane's front door: canonical episodes are integrity-checked once. + +#474: every post-sync read ran with chunk CRC validation off, so a canonical +that decayed on disk after sync was re-certified by the very lanes that exist +to judge it -- fresh ``measured`` findings over bytes the file's own stamp +refuses. The guard refuses the episode once, with a named reason, before any +check runs, and memoizes the verdict so one run pays for one strict read. +""" + +import io +from pathlib import Path + +import pytest +from mcap.reader import make_reader + +import hflow +import hflow.app +from hflow.app import CANONICAL_INTEGRITY_STEP_NAME +from hflow.curation import open_catalog_connection +from hflow.reader import CANONICAL_CRC_MISMATCH_REASON, verify_canonical_integrity +from hflow.stage_execution import process_stage_batch +from hflow.testing import SyntheticEpisodeSpec, synthesize_episode + +SPEC = SyntheticEpisodeSpec(duration_s=2.0, cameras=()) + + +def _app_with_probe_check( + data_root: Path, +) -> tuple[hflow.App, list[int], list[int]]: + """An app whose one check and one enrichment count their invocations.""" + app = hflow.App("integrity", data_root=data_root, default_checks=()) + probe_runs: list[int] = [] + caption_runs: list[int] = [] + + @app.check(version="1") + def probe(ep: hflow.Episode) -> hflow.CheckResult: + probe_runs.append(1) + return hflow.CheckResult(measurements={"probe": 1}) + + @app.enrich(version="1") + def caption(ep: hflow.Episode) -> hflow.EnrichmentResult: + caption_runs.append(1) + return hflow.EnrichmentResult(labels={"caption": "a robot arm moves"}) + + return app, probe_runs, caption_runs + + +def _corrupt_first_chunk_crc(canonical_path: Path) -> None: + """Flip one bit of the first chunk's stored CRC: header rot, not payload + damage -- the payload still decompresses, so only the CRC knows.""" + data = bytearray(canonical_path.read_bytes()) + summary = make_reader(io.BytesIO(bytes(data))).get_summary() + assert summary is not None and summary.chunk_indexes + crc_offset = summary.chunk_indexes[0].chunk_start_offset + 33 + data[crc_offset] ^= 0x01 + canonical_path.write_bytes(bytes(data)) + + +def test_a_decayed_canonical_is_refused_once_with_a_named_reason(tmp_path: Path) -> None: + data_root = tmp_path / "data" + app, probe_runs, caption_runs = _app_with_probe_check(data_root) + source = synthesize_episode(tmp_path / "episode.mcap", SPEC) + + healthy = app.process(source, stages="full") + assert healthy.refusal_reason is None + probe_runs_after_healthy = len(probe_runs) + caption_runs_after_healthy = len(caption_runs) + + _corrupt_first_chunk_crc(healthy.canonical_path) + + refused = app.process(source, stages="metadata_backfill") + assert refused.refusal_reason == CANONICAL_CRC_MISMATCH_REASON + # ONE diagnosis: the refusal lives on the report field and on one + # framework-owned catalog row; report.checks is empty because no check + # ran, so zero check tracebacks is structural. + assert refused.checks == [] + assert "REFUSED: canonical-crc-mismatch" in refused.summary() + assert len(probe_runs) == probe_runs_after_healthy + assert refused.has_errors + + # The refusal covers every consuming lane. The relabel lane declines to + # spend enrichments on the damaged bytes ... + relabel_refused = app.process(source, stages="relabel") + assert relabel_refused.refusal_reason == CANONICAL_CRC_MISMATCH_REASON + assert relabel_refused.enrichments == [] + assert len(caption_runs) == caption_runs_after_healthy + + # ... and the production meta task (the generated DAG's own batch entry, + # the lane an online re-check flows through) refuses with it. + recheck_counts = process_stage_batch(app, [str(source)], "meta") + assert recheck_counts == {"processed": 0, "quarantined": 0, "errors": 1} + + # The reason is a queryable value, not just a log string: one refusal row + # filtered by exact error equality, and an episode the curation views no + # longer read as ok. + connection = open_catalog_connection(data_root / "catalog") + try: + refusal_rows = connection.execute( + "SELECT check_name, status, error FROM check_runs WHERE error = ?", + [CANONICAL_CRC_MISMATCH_REASON], + ).fetchall() + episode_status = connection.execute("SELECT status FROM episodes").fetchall() + finally: + connection.close() + assert refusal_rows == [(CANONICAL_INTEGRITY_STEP_NAME, "error", CANONICAL_CRC_MISMATCH_REASON)] + assert episode_status == [("unverified",)] + + +def test_a_healthy_canonical_runs_clean(tmp_path: Path) -> None: + data_root = tmp_path / "data" + app, probe_runs, _caption_runs = _app_with_probe_check(data_root) + source = synthesize_episode(tmp_path / "episode.mcap", SPEC) + + report = app.process(source, stages="full") + assert report.refusal_reason is None + assert [run.status for run in report.checks] == [hflow.CheckStatus.MEASURED] + assert not report.has_errors + assert len(probe_runs) == 1 + assert verify_canonical_integrity(report.canonical_path) == (True, None) + + connection = open_catalog_connection(data_root / "catalog") + try: + refusal_row_count = connection.execute( + "SELECT count(*) FROM check_runs WHERE status = 'error'" + ).fetchone() + finally: + connection.close() + assert refusal_row_count is not None + assert int(refusal_row_count[0]) == 0 + + +def test_one_corrupt_episode_is_strict_read_once_per_run( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Two check-lane runs over the same damaged bytes pay for one strict read. + + The spy wraps the real validator -- the pass still happens -- because the + thing being pinned is the cost the memo exists to avoid: a second full + CRC read of bytes already judged in this run. + """ + data_root = tmp_path / "data" + app, _probe_runs, _caption_runs = _app_with_probe_check(data_root) + source = synthesize_episode(tmp_path / "episode.mcap", SPEC) + + synced = app.process(source, stages={hflow.Stage.SYNC}, record=False) + _corrupt_first_chunk_crc(synced.canonical_path) + + strict_reads: list[Path] = [] + real_verify = hflow.app.verify_canonical_integrity + + def counting_verify(path: Path | str) -> tuple[bool, str | None]: + strict_reads.append(Path(path)) + return real_verify(path) + + monkeypatch.setattr(hflow.app, "verify_canonical_integrity", counting_verify) + + first = app.process(source, stages="metadata_backfill") + second = app.process(source, stages="metadata_backfill") + + assert len(strict_reads) == 1 + assert first.refusal_reason == second.refusal_reason == CANONICAL_CRC_MISMATCH_REASON + + # Both runs observed the same outcome, so the catalog keeps ONE refusal + # record: the exact replay deduped through the run fingerprint. + connection = open_catalog_connection(data_root / "catalog") + try: + refusal_row_count = connection.execute( + "SELECT count(*) FROM check_runs WHERE error = ?", [CANONICAL_CRC_MISMATCH_REASON] + ).fetchone() + finally: + connection.close() + assert refusal_row_count is not None + assert int(refusal_row_count[0]) == 1 + + +def test_a_run_with_no_step_work_pays_no_strict_read( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The guard protects step work over the canonical bytes; a stage + selection with none to do pays no strict read.""" + data_root = tmp_path / "data" + app = hflow.App("no-steps", data_root=data_root, default_checks=()) + source = synthesize_episode(tmp_path / "episode.mcap", SPEC) + + strict_reads: list[Path] = [] + real_verify = hflow.app.verify_canonical_integrity + + def counting_verify(path: Path | str) -> tuple[bool, str | None]: + strict_reads.append(Path(path)) + return real_verify(path) + + monkeypatch.setattr(hflow.app, "verify_canonical_integrity", counting_verify) + + # Sync owns the canonical and just wrote it; a relabel with no + # enrichments registered and a meta with no checks registered have no + # step work to protect. + app.process(source, stages={hflow.Stage.SYNC}, record=False) + app.process(source, stages="relabel", record=False) + app.process(source, stages="metadata_backfill", record=False) + + assert strict_reads == []