From e00f35009eba79be4eff9988b87015a33062cc74 Mon Sep 17 00:00:00 2001 From: Sagar Kharal Date: Sun, 6 Sep 2026 17:05:32 +0530 Subject: [PATCH 1/2] fix(import): CRC-validate reused landing episodes before stamping receipts The #390 resume path stamped conversion-grade receipts over reused bytes it never integrity-checked: _episode_identity_matches compared seven metadata fields and never read the message payloads, so a landing file with payload damage (bit rot, partial external copy) that left the container and metadata intact was reused and given a fresh receipt. In the resume scenario the prior run published no manifest (manifest-last), so nothing recorded the damage. The identity check now ends with a CRC-validated full message pass (the same standard hflow doctor applies). A file that fails falls through to re-conversion: damaged work is not reuse, it is work to redo. The reuse log line names the verification. tests/reuse_test_helpers.py adds flip_chunk_payload_bytes, a surgical corruption helper that flips bytes inside a chunk's records region while leaving the container and metadata intact, shared so the #423-family tests can reuse it. The identity fixture now writes one chunked message with CompressionType.NONE so the corruption is addressable in place. Controlled result on current main: 400 payload bytes flipped, size and metadata unchanged, identity still matched without the fix, and the reuse receipt carried a different content id (ce5f28aa -> 28ef502d). With the fix, the damaged file is refused and re-converted. Refs #426 --- src/hflow/importers/lerobot.py | 31 +++++++++-- tests/reuse_test_helpers.py | 49 +++++++++++++++++ tests/test_lerobot_converter.py | 93 ++++++++++++++++++++++++++++++++- 3 files changed, 167 insertions(+), 6 deletions(-) create mode 100644 tests/reuse_test_helpers.py diff --git a/src/hflow/importers/lerobot.py b/src/hflow/importers/lerobot.py index 36decc9f..f1bdaae8 100644 --- a/src/hflow/importers/lerobot.py +++ b/src/hflow/importers/lerobot.py @@ -29,6 +29,7 @@ from typing import NotRequired, TypedDict from urllib.parse import urlsplit +from mcap.reader import make_reader from mcap.writer import Writer as McapWriter from hflow.catalog import content_episode_id @@ -155,7 +156,13 @@ def _episode_identity_matches( episode_index: int, camera_keys: tuple[str, ...], ) -> bool: - """True when a published landing file belongs to this exact import.""" + """True when a published landing file belongs to this exact import. + + Identity over the metadata records, then a CRC-validated full message + pass: a reused episode must be whole, not merely labeled. Metadata-only + reads never touch chunk payloads, so without this pass a payload-damaged + file would match identity and be stamped with a fresh receipt. + """ from mcap.exceptions import McapError reader = None @@ -180,7 +187,7 @@ def _episode_identity_matches( if recorded_camera_keys is None: return False expected_gop = f"{IMPORT_GOP_SECONDS:g}" - return ( + if not ( episode_metadata.get("source_dataset") == dataset_source.repo_id and episode_metadata.get("source_revision") == dataset_source.revision and episode_metadata.get("source_episode_index") == str(episode_index) @@ -189,7 +196,19 @@ def _episode_identity_matches( and recorded_camera_keys == camera_keys and episode_metadata.get("gop_seconds") == expected_gop and provenance_metadata.get("gop_seconds") == expected_gop - ) + ): + return False + # Metadata matching is not integrity: the metadata records live outside + # the chunks, so payload damage never reaches them. Reuse must hold the + # file to the same standard hflow doctor applies to any canonical file. + try: + with local_episode_path.open("rb") as stream: + validated_reader = make_reader(stream, validate_crcs=True) + for _ in validated_reader.iter_messages(log_time_order=False): + pass + except (OSError, McapError, ValueError): + return False + return True def _try_reuse_completed_episode( @@ -923,7 +942,11 @@ def import_lerobot_dataset( camera_keys=resolved_camera_keys, ) if reused_episode is not None: - logger.info("reusing completed LeRobot episode %s", reused_episode["uri"]) + logger.info( + "reusing verified completed LeRobot episode %s (content_id %s)", + reused_episode["uri"], + reused_episode["content_id"], + ) published_episodes.append(reused_episode) continue published_episodes.append( diff --git a/tests/reuse_test_helpers.py b/tests/reuse_test_helpers.py new file mode 100644 index 00000000..a8c69fa9 --- /dev/null +++ b/tests/reuse_test_helpers.py @@ -0,0 +1,49 @@ +"""Shared helpers for reusing-episode and delivery-corruption tests.""" + +from __future__ import annotations + +import io +import struct +from pathlib import Path + +from hflow.catalog import content_episode_id + + +def flip_chunk_payload_bytes(episode_path: Path, *, count: int = 4) -> None: + """Corrupt message-data bytes in place, leaving the file valid. + + Locates the first chunk record and flips ``count`` bytes inside its + message-data records (the tail of the chunk's uncompressed records + region). The MCAP summary, indexes, and metadata records stay intact; + the chunk CRC no longer matches the damaged bytes, which is exactly what + a CRC-validated read catches and what on-disk bit rot or a partial + external copy looks like when the container survives. The fixture + writer emits uncompressed chunks (``compression=""``), so the records + region is plaintext. + """ + from mcap.reader import make_reader + + data = bytearray(episode_path.read_bytes()) + summary = make_reader(io.BytesIO(bytes(data))).get_summary() + if summary is None or not summary.chunk_indexes: + raise ValueError(f"{episode_path} has no chunk records to corrupt") + chunk_index_record = summary.chunk_indexes[0] + chunk_start = chunk_index_record.chunk_start_offset + # Chunk record per the MCAP spec: opcode(1) length(8) then + # message_start_time(8) message_end_time(8) uncompressed_size(8) + # uncompressed_crc(4) compression_length(4) compression records_length(8) + # records. + compression_length = struct.unpack_from(" bool: + """True when the file on disk no longer matches the recorded content id.""" + return content_episode_id(episode_path) != receipt_content_id diff --git a/tests/test_lerobot_converter.py b/tests/test_lerobot_converter.py index 5b145c6d..60c19ef8 100755 --- a/tests/test_lerobot_converter.py +++ b/tests/test_lerobot_converter.py @@ -1499,14 +1499,24 @@ def _write_identity_matching_landing_mcap( field and leave the rest matching, which is what separates the individual comparisons in ``_episode_identity_matches`` from each other. """ - from mcap.writer import Writer + from mcap.writer import CompressionType, Writer from hflow.format import METADATA_RECORD_EPISODE, METADATA_RECORD_PROVENANCE destination.parent.mkdir(parents=True, exist_ok=True) with destination.open("wb") as stream: - writer = Writer(stream) + # Uncompressed chunks: the payload-corruption test flips bytes inside + # the chunk records region, which is only addressable in place when + # the records are plaintext. + writer = Writer(stream, compression=CompressionType.NONE) writer.start(profile="", library="test-lerobot-resume") + 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.add_metadata( METADATA_RECORD_EPISODE, { @@ -2110,3 +2120,82 @@ def test_converter_version_bumped_with_the_label_support() -> None: """The label changes episode/v1 bytes, which content_episode_id hashes: the converter version moves with the change, not after it.""" assert prep.CONVERTER_VERSION == "lerobot-converter-v7" + + +def test_reuse_refuses_a_landing_episode_with_damaged_payload( + tmp_path: Path, +) -> None: + """Payload damage that leaves the MCAP structure and metadata intact is + still damaged work: the reuse path CRC-validates before stamping the + receipt, and refuses rather than laundering the bytes (#426).""" + from reuse_test_helpers import flip_chunk_payload_bytes + + data_root = LocalStorageRoot(tmp_path / "out") + landing = tmp_path / "out" / "landing" / "lerobot_episode_0001.mcap" + _write_identity_matching_landing_mcap( + landing, + dataset_source=_MATCHING_SOURCE, + episode_index=0, + camera_keys=_MATCHING_CAMERA_KEYS, + marker="payload-damaged", + ) + receipt_before = prep._try_reuse_completed_episode( + data_root, + dataset_source=_MATCHING_SOURCE, + episode_index=0, + camera_keys=_MATCHING_CAMERA_KEYS, + ) + assert receipt_before is not None + intact_content_id = receipt_before["content_id"] + + flip_chunk_payload_bytes(landing) + + assert ( + prep._try_reuse_completed_episode( + data_root, + dataset_source=_MATCHING_SOURCE, + episode_index=0, + camera_keys=_MATCHING_CAMERA_KEYS, + ) + is None + ) + # The damaged bytes must not be laundered: the damaged file's hash + # differs from the receipt the intact file earned. + from reuse_test_helpers import content_id_differs_from_delivery_receipt + + assert content_id_differs_from_delivery_receipt(landing, intact_content_id) + + +def test_reuse_accepts_an_intact_episode_after_the_crc_pass( + tmp_path: Path, +) -> None: + """The control: the CRC pass adds no refusal for undamaged bytes.""" + from reuse_test_helpers import flip_chunk_payload_bytes + + data_root = LocalStorageRoot(tmp_path / "out") + landing = tmp_path / "out" / "landing" / "lerobot_episode_0001.mcap" + _write_identity_matching_landing_mcap( + landing, + dataset_source=_MATCHING_SOURCE, + episode_index=0, + camera_keys=_MATCHING_CAMERA_KEYS, + marker="intact", + ) + + reused = prep._try_reuse_completed_episode( + data_root, + dataset_source=_MATCHING_SOURCE, + episode_index=0, + camera_keys=_MATCHING_CAMERA_KEYS, + ) + assert reused is not None + + flip_chunk_payload_bytes(landing) + + damaged = prep._try_reuse_completed_episode( + data_root, + dataset_source=_MATCHING_SOURCE, + episode_index=0, + camera_keys=_MATCHING_CAMERA_KEYS, + ) + assert damaged is None From cea1a03812d9825075fc6637ae1c72460d026f1c Mon Sep 17 00:00:00 2001 From: Kingston Date: Sun, 6 Sep 2026 06:51:32 -0700 Subject: [PATCH 2/2] feat(import): warn when a reused episode fails CRC validation --- src/hflow/importers/lerobot.py | 12 +++++++++++- tests/test_lerobot_converter.py | 25 +++++++++++++++++-------- 2 files changed, 28 insertions(+), 9 deletions(-) diff --git a/src/hflow/importers/lerobot.py b/src/hflow/importers/lerobot.py index f1bdaae8..1f9e0ec4 100644 --- a/src/hflow/importers/lerobot.py +++ b/src/hflow/importers/lerobot.py @@ -206,7 +206,17 @@ def _episode_identity_matches( validated_reader = make_reader(stream, validate_crcs=True) for _ in validated_reader.iter_messages(log_time_order=False): pass - except (OSError, McapError, ValueError): + except (OSError, McapError, ValueError) as error: + # Reached only when identity already matched, so this is our own prior + # output found damaged, not a stranger's file. Re-conversion repairs it + # silently otherwise, which would hide bit rot in a landing tree for as + # long as the imports keep succeeding. + logger.warning( + "re-converting landing episode %s: identity matches but the payload " + "failed CRC validation (%s)", + local_episode_path, + error, + ) return False return True diff --git a/tests/test_lerobot_converter.py b/tests/test_lerobot_converter.py index 60c19ef8..caa260e8 100755 --- a/tests/test_lerobot_converter.py +++ b/tests/test_lerobot_converter.py @@ -7,6 +7,7 @@ import io import json +import logging import shutil import subprocess import urllib.request @@ -2124,6 +2125,7 @@ def test_converter_version_bumped_with_the_label_support() -> None: def test_reuse_refuses_a_landing_episode_with_damaged_payload( tmp_path: Path, + caplog: pytest.LogCaptureFixture, ) -> None: """Payload damage that leaves the MCAP structure and metadata intact is still damaged work: the reuse path CRC-validates before stamping the @@ -2150,15 +2152,22 @@ def test_reuse_refuses_a_landing_episode_with_damaged_payload( flip_chunk_payload_bytes(landing) - assert ( - prep._try_reuse_completed_episode( - data_root, - dataset_source=_MATCHING_SOURCE, - episode_index=0, - camera_keys=_MATCHING_CAMERA_KEYS, + with caplog.at_level(logging.WARNING, logger="hflow.importers.lerobot"): + assert ( + prep._try_reuse_completed_episode( + data_root, + dataset_source=_MATCHING_SOURCE, + episode_index=0, + camera_keys=_MATCHING_CAMERA_KEYS, + ) + is None ) - is None - ) + # Silent re-conversion would repair the damage and hide it. The operator + # gets told which file failed, so bit rot in a landing tree is visible + # rather than absorbed by the next successful import. + warning_messages = [record.getMessage() for record in caplog.records] + assert any("failed CRC validation" in message for message in warning_messages), warning_messages + assert any(str(landing) in message for message in warning_messages), warning_messages # The damaged bytes must not be laundered: the damaged file's hash # differs from the receipt the intact file earned. from reuse_test_helpers import content_id_differs_from_delivery_receipt