Merge branch 'cmc/rrd_footers_1_framing' into cmc/rrd_footers_2_rrd_manifests

teh-cmc · teh-cmc · commit eae5d66b1a77 · 2025-12-03T09:59:19.000+01:00
diff --git a/Cargo.lock b/Cargo.lock
@@ -9085,6 +9085,7 @@ dependencies = [
  "re_protos",
  "re_smart_channel",
  "re_sorbet",
+ "re_span",
  "re_tracing",
  "re_types",
  "re_types_core",
diff --git a/crates/store/re_log_encoding/Cargo.toml b/crates/store/re_log_encoding/Cargo.toml
@@ -50,6 +50,7 @@ re_log.workspace = true
 re_protos.workspace = true
 re_smart_channel.workspace = true
 re_sorbet.workspace = true
+re_span.workspace = true
 re_tracing.workspace = true
 re_types_core.workspace = true
 
diff --git a/crates/store/re_log_encoding/src/rrd/decoder/state_machine.rs b/crates/store/re_log_encoding/src/rrd/decoder/state_machine.rs
@@ -340,15 +340,12 @@ impl<T: DecoderEntrypoint> Decoder<T> {
                             let StreamFooter {
                                 fourcc: _,
                                 identifier: _,
-                                rrd_footer_byte_offset_from_start_excluding_header,
-                                rrd_footer_byte_size_excluding_header,
+                                rrd_footer_byte_span_from_start_excluding_header,
                                 crc_excluding_header: _,
                             } = footer;
 
-                            let rrd_footer_start =
-                                rrd_footer_byte_offset_from_start_excluding_header;
                             let rrd_footer_end =
-                                rrd_footer_start + rrd_footer_byte_size_excluding_header;
+                                rrd_footer_byte_span_from_start_excluding_header.end();
 
                             if rrd_footer_end > position as u64 {
                                 // The RRD footer cannot possibly end after the stream footer starts, since it must
diff --git a/crates/store/re_log_encoding/src/rrd/encoder.rs b/crates/store/re_log_encoding/src/rrd/encoder.rs
@@ -109,8 +109,7 @@ impl FooterState {
     #[expect(clippy::unnecessary_wraps)] // won't stay for long
     fn append(
         &mut self,
-        _byte_offset: u64,
-        _byte_size: u64,
+        _byte_span_excluding_header: re_span::Span<u64>,
         msg: &re_log_types::LogMsg,
     ) -> Result<(), EncodeError> {
         match msg {
@@ -226,12 +225,13 @@ impl<W: std::io::Write> Encoder<W> {
 
         let byte_size_excluding_header = n - crate::MessageHeader::ENCODED_SIZE_BYTES as u64;
 
+        let byte_span_excluding_header = re_span::Span {
+            start: byte_offset_excluding_header,
+            len: byte_size_excluding_header,
+        };
+
         if let Some(footer_state) = self.footer_state.as_mut() {
-            footer_state.append(
-                byte_offset_excluding_header,
-                byte_size_excluding_header,
-                message,
-            )?;
+            footer_state.append(byte_span_excluding_header, message)?;
         }
 
         Ok(n)
diff --git a/crates/store/re_log_encoding/src/rrd/frames.rs b/crates/store/re_log_encoding/src/rrd/frames.rs
@@ -229,41 +229,39 @@ pub struct StreamFooter {
     /// Always set to [`Self::RRD_IDENTIFIER`].
     pub identifier: [u8; 4], // FOOT
 
-    /// The position in bytes where the serialized [`RrdFooter`] payload starts, excluding the
-    /// message header.
+    /// The span in bytes where the serialized [`RrdFooter`] payload starts end ends, excluding
+    /// the message header.
     ///
     /// I.e. a transport-level [`RrdFooter`] can be decoded from the bytes at (pseudo-code):
     /// ```text
-    /// let start = stream_footer.rrd_footer_byte_offset_from_start_excluding_header;
-    /// let end = start + stream_footer.rrd_footer_byte_size_excluding_header;
+    /// let start = stream_footer.rrd_footer_byte_span_from_start_excluding_header.start;
+    /// let end = stream_footer.rrd_footer_byte_span_from_start_excluding_header.end();
     /// let bytes = &file[start..end];
     /// let rrd_footer = re_protos::RrdFooter::decode(bytes)?;
     /// let rrd_footer = rrd_footer.to_application()?;
     /// ```
     ///
     /// [`RrdFooter`]: [crate::RrdFooter]
-    pub rrd_footer_byte_offset_from_start_excluding_header: u64,
-
-    /// The size in bytes of the serialized [`RrdFooter`] payload, excluding the message header.
-    ///
-    /// This is guaranteed to be the same value as the `len` found in the associated message
-    /// header, but duplicating it here makes it possible for decoders to get everything they
-    /// need using a single IO.
-    ///
-    /// [`RrdFooter`]: [crate::RrdFooter]
-    pub rrd_footer_byte_size_excluding_header: u64,
+    pub rrd_footer_byte_span_from_start_excluding_header: re_span::Span<u64>,
 
     /// Checksum for the [`RrdFooter`] payload.
     ///
     /// The footer is most often accessed by jumping straight to it, so this is a nice extra safety
     /// to make sure that we didn't just get "lucky" (or unlucky, rather) when jumping around and
     /// parsing random bytes.
     ///
+    /// For now, the checksum algorithm is hardcoded to `xxh32`, the 32bit variant of the
+    /// [`xxhash` family of hashing algorithms](https://xxhash.com/).
+    /// This is fast, HW-accelerated, non-cryptographic hash that is perfect when needing to hash
+    /// RRD footers, which can potentially get very, very large.
+    ///
     /// [`RrdFooter`]: [crate::RrdFooter]
     //
     // TODO(cmc): It shouldn't be the job of the StreamFooter to carry checksums for a specific
     // message's payload. All frames should have identifiers and CRCs for both themselves and their
     // payloads, in which case this CRC would belong in the MessageHeader.
+    // TODO(cmc): In a potential future RRF3, we might make the choice of checksum algorithm
+    // configurable via flag.
     pub crc_excluding_header: u32,
 }
 
@@ -273,15 +271,13 @@ impl StreamFooter {
     pub const RRD_IDENTIFIER: [u8; 4] = *b"FOOT";
 
     pub fn new(
-        rrd_footer_byte_offset_from_start_excluding_header: u64,
-        rrd_footer_byte_size_excluding_header: u64,
+        rrd_footer_byte_span_from_start_excluding_header: re_span::Span<u64>,
         crc_excluding_header: u32,
     ) -> Self {
         Self {
             fourcc: crate::RRD_FOURCC,
             identifier: Self::RRD_IDENTIFIER,
-            rrd_footer_byte_offset_from_start_excluding_header,
-            rrd_footer_byte_size_excluding_header,
+            rrd_footer_byte_span_from_start_excluding_header,
             crc_excluding_header,
         }
     }
@@ -291,11 +287,14 @@ impl StreamFooter {
         rrd_footer_bytes: &[u8],
     ) -> Self {
         let crc_excluding_header = xxhash_rust::xxh32::xxh32(rrd_footer_bytes, Self::CRC_SEED);
+        let rrd_footer_byte_span_from_start_excluding_header = re_span::Span {
+            start: rrd_footer_byte_offset_from_start_excluding_header,
+            len: rrd_footer_bytes.len() as u64,
+        };
         Self {
             fourcc: crate::RRD_FOURCC,
             identifier: Self::RRD_IDENTIFIER,
-            rrd_footer_byte_offset_from_start_excluding_header,
-            rrd_footer_byte_size_excluding_header: rrd_footer_bytes.len() as u64,
+            rrd_footer_byte_span_from_start_excluding_header,
             crc_excluding_header,
         }
     }
@@ -306,18 +305,25 @@ impl Encodable for StreamFooter {
         let Self {
             fourcc,
             identifier,
-            rrd_footer_byte_offset_from_start_excluding_header,
-            rrd_footer_byte_size_excluding_header,
-            crc_excluding_header: crc,
+            rrd_footer_byte_span_from_start_excluding_header,
+            crc_excluding_header,
         } = self;
 
         let before = out.len() as u64;
 
         out.extend_from_slice(fourcc);
         out.extend_from_slice(identifier);
-        out.extend_from_slice(&rrd_footer_byte_offset_from_start_excluding_header.to_le_bytes());
-        out.extend_from_slice(&rrd_footer_byte_size_excluding_header.to_le_bytes());
-        out.extend_from_slice(&crc.to_le_bytes());
+        out.extend_from_slice(
+            &rrd_footer_byte_span_from_start_excluding_header
+                .start
+                .to_le_bytes(),
+        );
+        out.extend_from_slice(
+            &rrd_footer_byte_span_from_start_excluding_header
+                .len
+                .to_le_bytes(),
+        );
+        out.extend_from_slice(&crc_excluding_header.to_le_bytes());
 
         let n = out.len() as u64 - before;
         assert_eq!(Self::ENCODED_SIZE_BYTES as u64, n);
@@ -356,17 +362,16 @@ impl Decodable for StreamFooter {
             )));
         }
 
-        let rrd_footer_byte_offset_from_start_excluding_header =
-            u64::from_le_bytes(data[8..16].try_into().expect("cannot fail, checked above"));
-        let rrd_footer_byte_size_excluding_header =
-            u64::from_le_bytes(data[16..24].try_into().expect("cannot fail, checked above"));
+        let rrd_footer_byte_span_from_start_excluding_header = re_span::Span {
+            start: u64::from_le_bytes(data[8..16].try_into().expect("cannot fail, checked above")),
+            len: u64::from_le_bytes(data[16..24].try_into().expect("cannot fail, checked above")),
+        };
         let crc = u32::from_le_bytes(data[24..28].try_into().expect("cannot fail, checked above"));
 
         Ok(Self {
             fourcc,
             identifier,
-            rrd_footer_byte_offset_from_start_excluding_header,
-            rrd_footer_byte_size_excluding_header,
+            rrd_footer_byte_span_from_start_excluding_header,
             crc_excluding_header: crc,
         })
     }
diff --git a/crates/store/re_log_encoding/tests/footers_and_manifests.rs b/crates/store/re_log_encoding/tests/footers_and_manifests.rs
@@ -64,16 +64,18 @@ fn footer_empty() {
         re_log_encoding::StreamFooter::from_rrd_bytes(&msgs_encoded[stream_footer_start..])
             .unwrap();
 
-    let rrd_footer_start =
-        stream_footer.rrd_footer_byte_offset_from_start_excluding_header as usize;
-    let rrd_footer_end = rrd_footer_start
-        .checked_add(stream_footer.rrd_footer_byte_size_excluding_header as usize)
-        .unwrap();
-    let rrd_footer_bytes = &msgs_encoded[rrd_footer_start..rrd_footer_end];
+    let rrd_footer_range = stream_footer
+        .rrd_footer_byte_span_from_start_excluding_header
+        .try_cast::<usize>()
+        .unwrap()
+        .range();
+    let rrd_footer_bytes = &msgs_encoded[rrd_footer_range];
 
     {
         let crc = re_log_encoding::StreamFooter::from_rrd_footer_bytes(
-            rrd_footer_start as u64,
+            stream_footer
+                .rrd_footer_byte_span_from_start_excluding_header
+                .start,
             rrd_footer_bytes,
         )
         .crc_excluding_header;
