refactor(mcap): integrate mcap crate for S3 streaming

[zhexuany]

Description

Integrate Foxglove's mcap crate for MCAP record parsing while keeping robocodec's custom Rayon-based parallel decompression for 6-8x performance on local files.

Type of Change

• Code cleanup/refactoring
• Performance improvement

Related Issues

Related to #52 (enhancement for S3MessageStream metadata)

Changes Made

• Add McapS3Adapter (src/io/formats/mcap/s3_adapter.rs, ~460 lines)

  • Wraps mcap::LinearReader for event-driven MCAP parsing
  • Process chunks incrementally as data arrives from S3
  • Parse Schema, Channel, and Message records
  • Crate-private (not part of public API)

• Update S3Reader (src/io/s3/reader.rs)

  • Replace StreamingMcapParser with McapS3Adapter
  • Fix silent failures: add logging for MCAP parse errors
  • Extract chunk parsing logic into helper method

• Deprecate StreamingMcapParser (src/io/formats/mcap/stream.rs)

  • Mark as deprecated with recommendation to use McapS3Adapter

• Update module exports (src/io/formats/mcap/mod.rs)

  • Make s3_adapter crate-private

Reduces custom MCAP parsing code from ~1600 lines to ~460 lines.

Testing

• Unit tests added/updated
• All existing tests pass (1840 tests)

Test Commands Used:

cargo test
cargo clippy --all-features -- -D warnings
cargo fmt --check

Checklist

• My code follows the project's style guidelines
• I have performed a self-review of my code
• I have commented my code where necessary
• My changes generate no new warnings
• New and existing tests pass locally

Additional Notes

Hybrid Architecture

This PR uses a hybrid approach:

• S3 streaming: Uses mcap::LinearReader for robust event-driven parsing
• Local files: Keeps custom Rayon-based parallel decompression (6-8x speedup)

Silent Failure Fixes

Three locations where parse errors were silently discarded now log explicitly:

1. scan_mcap_for_metadata() - logs MCAP parse errors during channel discovery
2. parse_mcap_header() - logs MCAP parse errors during header parsing
3. S3MessageStream::parse_chunk() - logs MCAP parse errors during streaming

All use tracing::warn! with context (location, offset, error details).

[codecov]

Codecov Report

❌ Patch coverage is 65.11205% with 576 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/io/formats/mcap/transport_reader.rs	9.02%	121 Missing ⚠️
src/io/s3/reader.rs	40.99%	95 Missing ⚠️
src/io/reader/mod.rs	52.51%	66 Missing ⚠️
src/io/traits.rs	13.46%	45 Missing ⚠️
src/io/formats/mcap/s3_adapter.rs	86.34%	40 Missing ⚠️
src/io/formats/bag/stream.rs	21.62%	29 Missing ⚠️
src/io/formats/mcap/two_pass.rs	0.00%	23 Missing ⚠️
src/io/formats/mcap/parallel.rs	55.81%	19 Missing ⚠️
src/io/formats/rrd/reader.rs	46.87%	17 Missing ⚠️
src/io/formats/mcap/reader.rs	38.09%	13 Missing ⚠️
... and 16 more

📢 Thoughts on this report? Let us know!

[greptile-apps]

Greptile Overview

Greptile Summary

Refactored MCAP S3 streaming to use Foxglove's mcap crate's LinearReader instead of custom parsing, reducing code from ~1600 to ~460 lines while maintaining the high-performance Rayon-based parallel decompression for local files.

Key Changes:

• Added McapS3Adapter wrapping mcap::LinearReader for event-driven MCAP parsing on S3 streams
• Updated S3Reader to use McapS3Adapter instead of deprecated StreamingMcapParser
• Fixed silent failure issues by adding explicit error logging with tracing::warn! in three locations
• Deprecated StreamingMcapParser with migration guidance
• Made s3_adapter module crate-private (not exposed in public API)
• Refactored chunk parsing into dedicated helper method for better code organization

Critical Issue Found:

• McapS3Adapter::parse_message() has a buffer bounds bug where MESSAGE_HEADER_LEN is 20 but the code accesses body[18..26] (requires 26 bytes) and uses body[20..] for data instead of body[26..]

Positive Improvements:

• Better architecture leveraging battle-tested mcap crate for streaming scenarios
• Cleaner error handling with proper logging instead of silent failures
• Improved code maintainability with constants for buffer sizes
• Good test coverage for the new adapter module

Confidence Score: 2/5

• Critical buffer bounds bug in message parsing will cause runtime errors when processing MCAP messages
• The buffer bounds issue in McapS3Adapter::parse_message() (checking for 20 bytes but accessing 26 bytes) is a critical logic error that will cause out-of-bounds panics or incorrect data parsing during S3 streaming. While the refactoring approach is sound and improves code quality, this bug must be fixed before merging.
• src/io/formats/mcap/s3_adapter.rs requires immediate attention for the buffer bounds bug in parse_message()

Important Files Changed

Filename	Overview
src/io/formats/mcap/s3_adapter.rs	New adapter integrating mcap::LinearReader for S3 streaming, but contains critical buffer bounds bug in parse_message() (line 276-290)
src/io/s3/reader.rs	Replaces StreamingMcapParser with McapS3Adapter, adds proper error logging for silent failures, refactors parsing logic with cleaner structure
src/io/formats/mcap/stream.rs	Adds deprecation warning for StreamingMcapParser directing users to McapS3Adapter
src/io/formats/mcap/mod.rs	Adds crate-private s3_adapter module and updates documentation to reflect hybrid architecture

Sequence Diagram

sequenceDiagram
    participant S3 as S3 Storage
    participant Reader as S3Reader
    participant Adapter as McapS3Adapter
    participant LinearReader as mcap::LinearReader
    participant Stream as S3MessageStream

    Note over Reader: Initialization Phase
    Reader->>S3: fetch_range(footer)
    S3-->>Reader: footer data
    Reader->>Reader: parse_mcap_footer()
    
    alt Has Summary Section
        Reader->>S3: fetch_range(summary)
        S3-->>Reader: summary data
        Reader->>Reader: scan_summary_for_channels()
    else No Summary
        Reader->>S3: fetch_range(0, 10MB)
        S3-->>Reader: initial data
        Reader->>Adapter: new()
        Adapter->>LinearReader: new()
        Reader->>Adapter: process_chunk(data)
        Adapter->>LinearReader: insert() + notify_read()
        loop Process Events
            LinearReader-->>Adapter: Record{opcode, data}
            alt Schema Record
                Adapter->>Adapter: parse_schema()
            else Channel Record
                Adapter->>Adapter: parse_channel()
            end
        end
        Adapter-->>Reader: channels discovered
    end

    Note over Stream: Streaming Phase
    Stream->>Adapter: new()
    loop While not EOF
        Stream->>S3: fetch_range(position, chunk_size)
        S3-->>Stream: chunk data
        Stream->>Adapter: process_chunk(chunk_data)
        Adapter->>LinearReader: insert() + notify_read()
        loop Process Events
            LinearReader-->>Adapter: Record{opcode, data}
            alt Message Record
                Adapter->>Adapter: parse_message()
                Adapter-->>Stream: MessageRecord
            end
        end
        Stream-->>Stream: pending_messages.push()
        Stream-->>Reader: (ChannelInfo, data)
    end

Loading

[greptile-apps]

_{4 files reviewed, 1 comment}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

MESSAGE_HEADER_LEN is 20 but line 288 accesses body[18..26] which requires 26 bytes minimum, and line 290 should use body[26..] not body[20..] for data

Suggested change

	const MESSAGE_HEADER_LEN: usize = 20;
	if body.len() < MESSAGE_HEADER_LEN {
	return Err(FatalError::invalid_format(
	"MCAP Message record",
	body[..body.len().min(10)].to_vec(),
	));
	}
	let channel_id = u16::from_le_bytes(body[0..2].try_into().unwrap());
	let sequence = u64::from_le_bytes(body[2..10].try_into().unwrap());
	let log_time = u64::from_le_bytes(body[10..18].try_into().unwrap());
	let publish_time = u64::from_le_bytes(body[18..26].try_into().unwrap());
	let data = body[20..].to_vec();
	const MESSAGE_HEADER_LEN: usize = 26;
	if body.len() < MESSAGE_HEADER_LEN {
	return Err(FatalError::invalid_format(
	"MCAP Message record",
	body[..body.len().min(10)].to_vec(),
	));
	}
	let channel_id = u16::from_le_bytes(body[0..2].try_into().unwrap());
	let sequence = u64::from_le_bytes(body[2..10].try_into().unwrap());
	let log_time = u64::from_le_bytes(body[10..18].try_into().unwrap());
	let publish_time = u64::from_le_bytes(body[18..26].try_into().unwrap());
	let data = body[26..].to_vec();

Prompt To Fix With AI

This is a comment left during a code review.
Path: src/io/formats/mcap/s3_adapter.rs
Line: 276:290

Comment:
`MESSAGE_HEADER_LEN` is 20 but line 288 accesses `body[18..26]` which requires 26 bytes minimum, and line 290 should use `body[26..]` not `body[20..]` for data

```suggestion
        const MESSAGE_HEADER_LEN: usize = 26;

        if body.len() < MESSAGE_HEADER_LEN {
            return Err(FatalError::invalid_format(
                "MCAP Message record",
                body[..body.len().min(10)].to_vec(),
            ));
        }

        let channel_id = u16::from_le_bytes(body[0..2].try_into().unwrap());
        let sequence = u64::from_le_bytes(body[2..10].try_into().unwrap());
        let log_time = u64::from_le_bytes(body[10..18].try_into().unwrap());
        let publish_time = u64::from_le_bytes(body[18..26].try_into().unwrap());

        let data = body[26..].to_vec();
```

How can I resolve this? If you propose a fix, please make it concise.