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 == []