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")