From 9a80def5f9c58db2e20f85d7fe2a64d99beca274 Mon Sep 17 00:00:00 2001 From: Sravan1011 Date: Mon, 7 Sep 2026 23:48:22 +0530 Subject: [PATCH] fix(ingest): CRC-validate the primary ingest read, not just the resume path #429 fixed the LeRobot resume path; the same hole was open one stage earlier, on the primary path every ingest takes. transform.py's read of the source file (open_reader) never asked for CRC validation, so a structurally valid MCAP with a damaged chunk payload transcoded without complaint and got a fresh, true receipt over corrupt bytes. open_reader and PythonMcapEpisodeReader gain validate_crcs (default False); only the ingest read in transform.py opts in, so the other three call sites reading HFlow's own canonical output are unaffected. The transcode already decodes every chunk in one pass, so this piggybacks a check on a read that already happens: measured 1.113x on a 30MB source and 1.035x on a 75MB one. CRCValidationError subclasses ValueError, not McapError, so classify_ingest_failure needed its own branch for it -- without one it silently fell through to INFRASTRUCTURE, blaming the platform for a damaged recording. It classifies to the same SOURCE_UNREADABLE kind as the not-MCAP case; error_type (InvalidMagic vs CRCValidationError) keeps the two distinguishable in the ledger without a new enum member. Refs #431 Co-Authored-By: Claude Sonnet 5 --- src/hflow/ingest_ledger.py | 12 ++++++++ src/hflow/reader.py | 18 ++++++++---- src/hflow/transform.py | 6 +++- tests/test_ingest_in_process.py | 48 +++++++++++++++++++++++++++++++ tests/test_ingest_ledger.py | 50 +++++++++++++++++++++++++++++++++ 5 files changed, 128 insertions(+), 6 deletions(-) diff --git a/src/hflow/ingest_ledger.py b/src/hflow/ingest_ledger.py index 439579cd..a3141fda 100644 --- a/src/hflow/ingest_ledger.py +++ b/src/hflow/ingest_ledger.py @@ -100,6 +100,7 @@ def classify_ingest_failure(error: BaseException) -> IngestFailureKind: return IngestFailureKind.SOURCE_UNSUPPORTED try: from mcap.exceptions import McapError + from mcap.stream_reader import CRCValidationError except ImportError: # pragma: no cover - mcap is a hard dependency return IngestFailureKind.INFRASTRUCTURE if isinstance(error, McapError): @@ -107,6 +108,17 @@ def classify_ingest_failure(error: BaseException) -> IngestFailureKind: # and friends: the file is not a readable MCAP, which is a fact about # the recording rather than about this machine. return IngestFailureKind.SOURCE_UNREADABLE + if isinstance(error, CRCValidationError): + # Raised by a CRC-validated read (transform.py's ingest read passes + # validate_crcs=True) on a structurally valid MCAP whose chunk payload + # does not match its recorded checksum. mcap does not root this on + # McapError -- it subclasses ValueError instead -- so it needs its own + # branch or it silently falls to INFRASTRUCTURE below, blaming the + # platform for a damaged recording. Same failure kind as McapError: + # the file is a fact about the recording, not the machine. error_type + # still distinguishes "not MCAP" (InvalidMagic) from "MCAP with a + # damaged payload" (CRCValidationError) in the stored row. + return IngestFailureKind.SOURCE_UNREADABLE return IngestFailureKind.INFRASTRUCTURE diff --git a/src/hflow/reader.py b/src/hflow/reader.py index 75d38008..29f2d77e 100644 --- a/src/hflow/reader.py +++ b/src/hflow/reader.py @@ -140,11 +140,11 @@ def close(self) -> None: class PythonMcapEpisodeReader: """Pure-Python :class:`EpisodeReader` backend over the stock ``mcap`` package.""" - def __init__(self, path: Path | str) -> None: + def __init__(self, path: Path | str, *, validate_crcs: bool = False) -> None: self.path = Path(path) self._stream: IO[bytes] = self.path.open("rb") try: - self._reader: McapReader = make_reader(self._stream) + self._reader: McapReader = make_reader(self._stream, validate_crcs=validate_crcs) except BaseException: # make_reader validates the magic bytes; don't leak the handle # when it rejects a non-MCAP, empty, or truncated file. @@ -293,6 +293,14 @@ def __exit__( self.close() -def open_reader(path: Path | str) -> EpisodeReader: - """Open an episode file with the default (pure-Python) reader backend.""" - return PythonMcapEpisodeReader(path) +def open_reader(path: Path | str, *, validate_crcs: bool = False) -> EpisodeReader: + """Open an episode file with the default (pure-Python) reader backend. + + ``validate_crcs`` checks each chunk's CRC as it is decoded, catching + payload damage that magic-byte and summary checks alone cannot see. It + defaults to ``False`` because most callers re-read a canonical file HFlow + already produced and already identifies by content hash; pass ``True`` + only when reading a source that has not been trusted yet (see + ``hflow.transform``). + """ + return PythonMcapEpisodeReader(path, validate_crcs=validate_crcs) diff --git a/src/hflow/transform.py b/src/hflow/transform.py index 035e95dc..aa1791cb 100644 --- a/src/hflow/transform.py +++ b/src/hflow/transform.py @@ -573,7 +573,11 @@ def write_canonical_episode( {derived_topic: version for derived_topic, _series, version in derived_channels}, ) - reader = open_reader(source_path) + # validate_crcs=True: this is a source nobody has trusted yet, unlike the + # canonical-file reads elsewhere in the tree. The transcode below already + # decodes every chunk over the full iteration, so this piggybacks a CRC + # check on a pass that was already happening rather than adding one. + reader = open_reader(source_path, validate_crcs=True) try: # Keyed by CHANNEL id: several channels may legally share a topic and # each must survive the transform as its own output channel. diff --git a/tests/test_ingest_in_process.py b/tests/test_ingest_in_process.py index 1ab2ae5f..0b85cd55 100644 --- a/tests/test_ingest_in_process.py +++ b/tests/test_ingest_in_process.py @@ -201,6 +201,54 @@ def test_a_failed_source_is_recorded_where_it_can_be_found( assert rows == [("episodes-in/corrupt.mcap", "sync", "source-unreadable", "InvalidMagic")] +def test_a_payload_damaged_source_is_classified_the_same_as_unreadable( + project: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Chunk-payload damage inside an otherwise valid MCAP (bit rot, a + partial copy) must fail ingest the same way a not-MCAP file does (#431), + not transcode quietly into a canonical episode with a receipt over + corrupt bytes. Distinct from the not-MCAP case only in ``error_type``.""" + from mcap.writer import CompressionType + from mcap.writer import Writer as StockWriter + from reuse_test_helpers import flip_chunk_payload_bytes + + from hflow.curation import open_catalog_connection + + monkeypatch.delenv("HFLOW_DATA_ROOT", raising=False) + monkeypatch.delenv("HFLOW_AIRFLOW_URL", raising=False) + + damaged = project / "data" / "episodes-in" / "payload-damaged.mcap" + with damaged.open("wb") as stream: + # Uncompressed chunk: flip_chunk_payload_bytes corrupts the records + # region in place, which is only addressable when it is plaintext. + writer = StockWriter(stream, compression=CompressionType.NONE) + writer.start(profile="", library="test") + schema_id = writer.register_schema( + name="test.Pointer", encoding="ros2msg", data=b"int32 x\n" + ) + channel_id = writer.register_channel( + topic="/pointer", message_encoding="ros2msg", schema_id=schema_id + ) + writer.add_message(channel_id, log_time=10**9, data=b"\x01\x00\x00\x00", publish_time=10**9) + writer.finish() + flip_chunk_payload_bytes(damaged) + monkeypatch.chdir(project) + + assert cli_main(["ingest", "episodes-in/payload-damaged.mcap"]) == 1 + assert "ingest_failures" in capsys.readouterr().err + + connection = open_catalog_connection(project / "data" / "catalog") + try: + rows = connection.execute( + "SELECT source_uri, stage, failure_kind, error_type FROM ingest_failures" + ).fetchall() + finally: + connection.close() + assert rows == [ + ("episodes-in/payload-damaged.mcap", "sync", "source-unreadable", "CRCValidationError") + ] + + def test_a_source_that_is_not_there_is_not_blamed_on_the_data( project: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/test_ingest_ledger.py b/tests/test_ingest_ledger.py index 17b79d3c..7bcb2e7f 100644 --- a/tests/test_ingest_ledger.py +++ b/tests/test_ingest_ledger.py @@ -5,6 +5,8 @@ import numpy as np import pytest from mcap.exceptions import InvalidMagic +from mcap.stream_reader import CRCValidationError +from mcap.writer import CompressionType from mcap.writer import Writer as StockWriter from hflow import transform @@ -26,6 +28,24 @@ def test_classify_mcap_error_as_source_unreadable() -> None: assert classify_ingest_failure(error) == IngestFailureKind.SOURCE_UNREADABLE +def test_classify_crc_validation_error_as_source_unreadable() -> None: + """``CRCValidationError`` subclasses ``ValueError``, not ``McapError`` (#431): + without its own branch it would fall through to ``INFRASTRUCTURE`` and + blame the platform for a damaged recording.""" + from mcap.records import Chunk + + chunk = Chunk( + compression="", + data=b"", + message_end_time=0, + message_start_time=0, + uncompressed_crc=1, + uncompressed_size=0, + ) + error = CRCValidationError(expected=1, actual=2, record=chunk) + assert classify_ingest_failure(error) == IngestFailureKind.SOURCE_UNREADABLE + + def test_classify_source_not_conforming_as_source_unsupported() -> None: error = SourceNotConforming("x") assert classify_ingest_failure(error) == IngestFailureKind.SOURCE_UNSUPPORTED @@ -36,6 +56,36 @@ def test_classify_unrecognized_error_as_infrastructure() -> None: assert classify_ingest_failure(error) == IngestFailureKind.INFRASTRUCTURE +def test_ingest_refuses_a_source_with_a_damaged_chunk_payload(tmp_path: Path) -> None: + """A structurally valid MCAP whose chunk payload does not match its + recorded CRC must not transcode quietly into a canonical episode with a + fresh receipt over corrupt bytes (#431). ``open_reader`` only checks CRCs + when told to; ingest's read now asks for it.""" + from reuse_test_helpers import flip_chunk_payload_bytes + + source = tmp_path / "payload-damaged.mcap" + with source.open("wb") as stream: + # Uncompressed chunk: flip_chunk_payload_bytes corrupts the records + # region in place, which is only addressable when it is plaintext. + writer = StockWriter(stream, compression=CompressionType.NONE) + writer.start(profile="", library="test") + schema_id = writer.register_schema( + name="test.Pointer", encoding="ros2msg", data=b"int32 x\n" + ) + channel_id = writer.register_channel( + topic="/pointer", message_encoding="ros2msg", schema_id=schema_id + ) + writer.add_message(channel_id, log_time=10**9, data=b"\x01\x00\x00\x00", publish_time=10**9) + writer.finish() + flip_chunk_payload_bytes(source) + + output = tmp_path / "out.mcap" + with pytest.raises(CRCValidationError): + write_canonical_episode(source, output) + + assert not output.exists() + + def test_unsupported_compressed_image_format_classifies_as_source_unsupported() -> None: with pytest.raises(SourceNotConforming) as raised: transform._input_codec_for_image_format("bogus", "/cam")