fix(ingest): CRC-validate the primary ingest read, not just the resume path

src/hflow/ingest_ledger.py

@@ -100,13 +100,25 @@ 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):
         # The single base of InvalidMagic, EndOfFile, RecordLengthLimitExceeded
         # 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
 
 

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)

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.

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:

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