fix: S3 BAG/RRD/MCAP transport reading and modularize tests

.github/workflows/ci.yml

@@ -243,4 +243,4 @@ jobs:
           MINIO_ENDPOINT: http://127.0.0.1:9000
           MINIO_BUCKET: test-bucket
           MINIO_REGION: us-east-1
-        run: cargo test --features remote -- s3_integration_tests
+        run: 'cargo test --features remote --test s3_tests s3::'

.github/workflows/test-s3.yml

@@ -11,7 +11,7 @@ on:
   pull_request:
     paths:
       - 'src/io/s3/**'
-      - 'tests/s3_integration_test.rs'
+      - 'tests/s3/**'
       - 'tests/s3_tests.rs'
       - 'docker-compose.yml'
       - '.github/workflows/test-s3.yml'
@@ -58,30 +58,39 @@ jobs:
 
       - name: Wait for MinIO to be healthy (bucket created)
         run: |
-        # Wait for MinIO healthcheck to pass (this means bucket exists)
-        for i in {1..60}; do
-          if docker compose ps | grep "robocodec-minio" | grep -q "healthy"; then
-            echo "MinIO is healthy and bucket is ready"
-            docker compose ps
-            break
+          # Wait for MinIO healthcheck to pass (this means bucket exists)
+          for i in {1..60}; do
+            if docker compose ps | grep "robocodec-minio" | grep -q "healthy"; then
+              echo "MinIO is healthy and bucket is ready"
+              docker compose ps
+              break
+            fi
+            echo "Waiting for MinIO to be healthy... ($i/60)"
+            sleep 2
+          done
+
+          # Verify bucket exists
+          if ! curl -f http://localhost:9000/test-fixtures 2>/dev/null; then
+            echo "Bucket 'test-fixtures' not found"
+            docker compose logs minio minio-init
+            exit 1
           fi
-          echo "Waiting for MinIO to be healthy... ($i/60)"
-          sleep 2
-        done
-
-        # Verify bucket exists
-        if ! curl -f http://localhost:9000/test-fixtures 2>/dev/null; then
-          echo "Bucket 'test-fixtures' not found"
-          docker compose logs minio minio-init
-          exit 1
-        fi
-        echo "Bucket 'test-fixtures' verified"
+          echo "Bucket 'test-fixtures' verified"
 
       - name: Run S3 unit tests
-        run: cargo test --package robocodec --lib io::s3
+        run: 'cargo test --package robocodec --lib io::s3'
 
       - name: Run S3 integration tests (with live MinIO)
-        run: cargo test --test s3_tests s3_integration
+        run: 'cargo test --features remote --test s3_tests s3::integration::'
+
+      - name: Run S3 RoboReader fail-fast tests
+        run: 'cargo test --features remote --test s3_tests s3::roboreader::'
+
+      - name: Run S3 parity fail-fast tests
+        run: 'cargo test --features remote --test s3_tests s3::parity::'
+
+      - name: Run S3 performance guardrail fail-fast tests
+        run: 'cargo test --features remote --test s3_tests s3::performance::'
 
       - name: Run clippy on S3 module
         run: cargo clippy --package robocodec -- -D warnings -D clippy::all

CLAUDE.md

@@ -136,7 +136,7 @@ The library exports these key types at the top level:
 - **S3**: `s3://bucket/path/file.mcap` (with optional `?endpoint=` and `?region=` query params)
 - **HTTP/HTTPS**: `https://example.com/file.mcap` (via HttpTransport)
 
-Transport-based reading uses `McapTransportReader` internally for streaming from remote sources.
+Transport-based reading dispatches to format readers via `FormatReader::open_from_transport`.
 
 - **`RoboWriter`** - Unified writer with format auto-detection
   - `create(path)` - Create writer based on extension

Cargo.toml

@@ -60,7 +60,7 @@ zstd = "0.13"
 lz4_flex = "0.11"
 bzip2 = "0.4"
 crc32fast = "1.4"
-mcap = "0.24"
+mcap = { version = "0.24", features = ["zstd", "lz4"] }
 rosbag = "0.6"
 bytemuck = "1.15"
 chrono = "0.4"

docs/adr-004-real-s3-streaming-minimal-api.md

@@ -0,0 +1,164 @@
+# ADR-004: Real S3 Streaming Reads with Minimal Public API
+
+**Author**: ArcheBase Team  
+**Date**: 2026-02-27  
+**Status**: Accepted  
+
+## Context
+
+ADR-002 and ADR-003 added transport readers for BAG and RRD, bringing all formats onto `RoboReader::open("s3://...")`. This closed functional gaps, but current behavior is still not fully aligned with true incremental remote streaming.
+
+Key gaps motivating this ADR:
+
+- Transport readers currently read the entire object before parse completes.
+- Retry configuration exists but is not enforced in request paths.
+- Range response validation is weak (status/header/length checks are incomplete).
+
+These gaps create correctness and resiliency risk for large remote objects and unstable networks, and they blur the API contract between public reader semantics and internal transport mechanics.
+
+## Decision
+
+Implement real S3 incremental reads behind the existing unified reader API, while freezing and minimizing the public surface.
+
+Decision points:
+
+- Keep the user-facing contract centered on `RoboReader`, unified decoded message types, and `ReaderConfig`.
+- Enforce strict HTTP range semantics for S3 reads, including validation and retry behavior.
+- Remove full-object preload behavior from transport reader paths; parsing must advance incrementally from fetched ranges/chunks.
+- Preserve format-specific parser implementations internally, but unify streaming behavior at iterator level (`decoded()` and raw iteration) across MCAP/BAG/RRD.
+
+## Phased Execution Plan
+
+### Phase 0: API boundary freeze
+
+- Goal: lock public API shape before internal refactor.
+- Exit criteria:
+  - Public API inventory documented (`RoboReader`, unified result/metadata types, `ReaderConfig`).
+  - No new public transport- or S3-specific reader types exported.
+
+### Phase 1: strict S3 range semantics + retries
+
+- Goal: make network fetch semantics correct and deterministic.
+- Exit criteria:
+  - Range request paths validate HTTP status (`206` for ranged responses where applicable), `Content-Range`, and payload length consistency.
+  - Retry policy from S3 config is actually applied in request execution paths.
+  - Retry classification cleanly separates recoverable vs fatal errors.
+
+### Phase 2: real incremental parsing (remove full-object preload)
+
+- Goal: ensure remote reads are truly streaming.
+- Exit criteria:
+  - Transport readers no longer require loading full object before parse completion.
+  - Parsing progresses in bounded-memory chunks and yields messages as data arrives.
+  - End-of-stream and partial-chunk edge cases are covered by tests.
+
+### Phase 3: unified iterator-level streaming via RoboReader
+
+- Goal: standardize observable streaming behavior at the unified API.
+- Exit criteria:
+  - `RoboReader::decoded()` behaves consistently for local and S3 sources across MCAP/BAG/RRD.
+  - Raw and decoded iterators share the same incremental consumption semantics.
+  - Format dispatch in `RoboReader` remains unchanged from a caller perspective.
+
+### Phase 4: local-vs-S3 parity correctness suite
+
+- Goal: verify remote behavior matches local correctness.
+- Exit criteria:
+  - Fixture-driven tests compare local and S3/transport outputs for channels, message payloads, timestamps, and ordering.
+  - Error path tests cover short reads, invalid range headers, and retriable transport failures.
+  - Parity suite runs for MCAP, BAG, and RRD.
+
+### Phase 5: performance hardening + CI guardrails
+
+- Goal: prevent regressions in memory profile and throughput.
+- Exit criteria:
+  - Benchmarks capture latency/throughput for representative object sizes and network conditions.
+  - CI gate tracks bounded-memory behavior and fails on major regression thresholds.
+  - Retry/backoff behavior validated under fault-injection scenarios.
+
+### Phase 6: docs finalization + API stabilization
+
+- Goal: finalize contract and migration guidance.
+- Exit criteria:
+  - Rustdoc and architecture docs reflect real streaming semantics and internal/public boundaries.
+  - ADR status reviewed for promotion from Proposed when all gates pass.
+  - Release notes document behavior guarantees and non-goals.
+
+## Public API Boundary (Minimal Surface)
+
+Public (stable contract):
+
+- `RoboReader` (`open`, `open_with_config`, iterator-facing methods).
+- Unified types such as `DecodedMessageResult` and `ChannelInfo`.
+- `ReaderConfig` (and builder) as the reader configuration surface.
+
+Internal (not public contract):
+
+- `Transport` trait and concrete transport types.
+- S3 client implementations and authentication plumbing.
+- Range fetch/retry internals (request policy, backoff, validation details).
+- Format-specific remote readers (`*TransportReader`) and parser state machines.
+
+This boundary preserves a small, format-agnostic API while allowing internal transport/parser evolution without downstream breakage.
+
+## Consequences
+
+Positive:
+
+- Stronger correctness guarantees for remote reads.
+- Better resiliency on transient network and object-store failures.
+- Predictable memory behavior for large S3 objects.
+- No public API expansion despite substantial internal improvements.
+
+Trade-offs:
+
+- Increased internal complexity in transport execution and parser coordination.
+- More integration and fault-injection test maintenance.
+- Potential short-term throughput variance while strict validation and retry logic are tuned.
+
+## Testing and Performance Gates
+
+- Correctness parity tests: local file vs S3 transport for MCAP/BAG/RRD outputs.
+- Protocol validation tests: status code, `Content-Range`, and body-length invariants.
+- Resilience tests: retry/backoff behavior across recoverable and fatal failure classes.
+- Resource gates: bounded-memory checks and regression thresholds in CI.
+- Compatibility checks: existing public `RoboReader` usage patterns compile and behave consistently.
+
+## Rollout and Compatibility
+
+- Rollout is internal-first and incremental by phase, with no new public entry points.
+- Existing callers using `RoboReader::open("s3://...")` remain source-compatible.
+- Behavior changes are semantic hardening (true streaming, stricter validation, retry enforcement), not API shape changes.
+- If regressions appear in a format path, rollback is scoped to internal transport/reader strategy without public API breakage.
+
+## Implementation Status (Current)
+
+- [x] **Phase 0: API boundary freeze** - **Completed**
+  - Public API surface remains centered on `RoboReader`, unified metadata/result types, and `ReaderConfig`; no new public S3 transport types were introduced.
+- [x] **Phase 1: strict S3 range semantics + retries** - **Completed**
+  - Strict S3 range validation and retry application are implemented in request paths.
+- [x] **Phase 2: real incremental parsing (remove full-object preload)** - **Completed**
+  - Transport reader paths no longer rely on full-object preload before parse completion, and incremental parsing behavior is validated across format paths.
+- [x] **Phase 3: unified iterator-level streaming via RoboReader** - **Completed**
+  - S3 raw and decoded iterator support is implemented with incremental, fail-fast behavior.
+- [x] **Phase 4: local-vs-S3 parity correctness suite** - **Completed**
+  - Fail-fast local-vs-S3 parity tests are in place for MCAP, BAG, and RRD via `RoboReader` public API.
+- [x] **Phase 5: performance hardening + CI guardrails** - **Completed**
+  - Fail-fast S3 performance guardrail tests enforce coarse latency/throughput thresholds in CI.
+- [x] **Phase 6: docs finalization + API stabilization** - **Completed**
+  - ADR status is promoted to `Accepted`, implementation status is finalized, and release notes capture guarantees and non-goals.
+
+## Behavior Guarantees
+
+- `RoboReader::open("s3://...")` resolves to the incremental S3 reader path and supports streaming consumption through `iter_raw()` and `decoded()`.
+- S3 range handling enforces strict status/header/length validation with configured retry behavior on recoverable failures.
+- CI includes fail-fast parity and performance guardrail gates for S3 paths to catch correctness and major regression issues early.
+- The public API remains minimal and stable (`RoboReader`, unified metadata/result types, `ReaderConfig`) with no new public S3-specific reader surface.
+
+## References
+
+- Existing ADRs: `docs/adr-002-bag-s3-streaming.md`, `docs/adr-003-rrd-s3-streaming.md`
+- Public API surface: `src/lib.rs`, `src/io/reader/mod.rs`, `src/io/reader/config.rs`, `src/io/metadata.rs`
+- Current transport readers: `src/io/formats/mcap/transport_reader.rs`, `src/io/formats/bag/transport_reader.rs`, `src/io/formats/rrd/transport_reader.rs`
+- Transport abstraction: `src/io/transport/core.rs`, `src/io/transport/s3/transport.rs`
+- S3 request and retry internals: `src/io/s3/client.rs`, `src/io/s3/config.rs`, `src/io/s3/error.rs`

src/io/formats/bag/mod.rs

@@ -22,10 +22,6 @@ pub mod sequential;
 #[cfg(feature = "remote")]
 pub mod stream;
 
-// Transport-based reader (S3, HTTP support)
-#[cfg(feature = "remote")]
-pub mod transport_reader;
-
 // Writer implementation
 pub mod writer;
 
@@ -40,6 +36,4 @@ pub use stream::{
     BAG_MAGIC_PREFIX, BagMessageRecord, BagRecord, BagRecordFields, BagRecordHeader,
     StreamingBagParser,
 };
-#[cfg(feature = "remote")]
-pub use transport_reader::BagTransportReader;
 pub use writer::{BagMessage, BagWriter};

src/io/formats/bag/parallel.rs

@@ -62,6 +62,59 @@ impl BagFormat {
         let writer = BagWriter::create(path)?;
         Ok(Box::new(writer))
     }
+
+    /// Open a BAG reader from a transport source.
+    #[cfg(feature = "remote")]
+    pub fn open_from_transport(
+        mut transport: Box<dyn crate::io::transport::Transport>,
+        path: String,
+    ) -> Result<ParallelBagReader> {
+        use std::pin::Pin;
+        use std::task::{Context, Poll, Waker};
+
+        let mut data = Vec::new();
+        let mut buffer = vec![0u8; 64 * 1024];
+        let waker = Waker::noop();
+        let mut cx = Context::from_waker(waker);
+        let mut pinned_transport = unsafe { Pin::new_unchecked(transport.as_mut()) };
+
+        loop {
+            match pinned_transport.as_mut().poll_read(&mut cx, &mut buffer) {
+                Poll::Ready(Ok(0)) => break,
+                Poll::Ready(Ok(n)) => data.extend_from_slice(&buffer[..n]),
+                Poll::Ready(Err(e)) => {
+                    return Err(CodecError::encode(
+                        "Transport",
+                        format!("Failed to read from {path}: {e}"),
+                    ));
+                }
+                Poll::Pending => std::thread::yield_now(),
+            }
+        }
+
+        let unique = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_nanos();
+        let temp_path = std::env::temp_dir().join(format!(
+            "robocodec_bag_transport_{}_{}.bag",
+            std::process::id(),
+            unique
+        ));
+
+        std::fs::write(&temp_path, &data).map_err(|e| {
+            CodecError::encode(
+                "BAG",
+                format!("Failed to write temporary BAG data to {:?}: {e}", temp_path),
+            )
+        })?;
+
+        let mut reader = ParallelBagReader::open(&temp_path)?;
+        reader.path = path;
+
+        let _ = std::fs::remove_file(&temp_path);
+        Ok(reader)
+    }
 }
 
 /// Parallel BAG reader with memory-mapped file access.

src/io/formats/bag/stream.rs

@@ -117,6 +117,8 @@ pub struct StreamingBagParser {
     version: Option<String>,
     /// Cached channel map (converted from connections)
     cached_channels: HashMap<u16, ChannelInfo>,
+    /// Message counts per connection ID
+    connection_message_counts: HashMap<u32, u64>,
 }
 
 impl StreamingBagParser {
@@ -132,6 +134,7 @@ impl StreamingBagParser {
             buffer_pos: 0,
             version: None,
             cached_channels: HashMap::new(),
+            connection_message_counts: HashMap::new(),
         }
     }
 
@@ -169,6 +172,12 @@ impl StreamingBagParser {
         }
 
         self.message_count += messages.len() as u64;
+        for msg in &messages {
+            *self
+                .connection_message_counts
+                .entry(msg.conn_id)
+                .or_insert(0) += 1;
+        }
         Ok(messages)
     }
 
@@ -625,7 +634,11 @@ impl StreamingBagParser {
                     schema: Some(conn.message_definition.clone()),
                     schema_data: None,
                     schema_encoding: Some("ros1msg".to_string()),
-                    message_count: 0,
+                    message_count: self
+                        .connection_message_counts
+                        .get(conn_id)
+                        .copied()
+                        .unwrap_or(0),
                     callerid: if conn.caller_id.is_empty() {
                         None
                     } else {