Stream .rrd files (#2412)

<!--
Open the PR up as a draft until you feel it is ready for a proper
review.

Do not make PR:s from your own `main` branch, as that makes it difficult
for reviewers to add their own fixes.

Add any improvements to the branch as new commits to make it easier for
reviewers to follow the progress. All commits will be squashed to a
single commit once the PR is merged into `main`.

Make sure you mention any issues that this PR closes in the description,
as well as any other related issues.

To get an auto-generated PR description you can put "copilot:summary" or
"copilot:walkthrough" anywhere.
-->

### What

Closes #2262

Instead of waiting for the `.rrd` file to finish downloading,
`stream_rrd_from_http` now uses `ehttp::streaming` and parses messages
as data is streamed in.

As part of this change, compression now uses the LZ4 block format
instead of the LZ4 frame format. **This is a breaking change to the RRD
format**.

| example                | `main`     | `jan/stream-rrd` | change  |
| ---------------------- | ---------- | ---------------- | ------- |
| signed_distance_fields | 21.328 MB  | 21.420 MB        | +0.43%  |
| deep_sdf               | 21.329 MB  | 21.329 MB        | 0.00%   |
| dicom                  | 67.657 MB  | 67.657 MB        | 0.00%   |
| raw_mesh               | 109.070 KB | 110.960 KB       | +1.73%  |
| clock                  | 54.901 KB  | 80.439 KB        | +46.52% |
| dicom_mri              | 67.656 MB  | 66.434 MB        | -1.81%  |
| api_demo               | 320.549 KB | 388.021 KB       | +21.05% |
| structure_from_motion  | 252.978 MB | 253.735 MB       | +0.30%  |
| car                    | 402.542 KB | 391.056 KB       | -2.85%  |
| rgbd                   | 168.420 MB | 167.382 MB       | -0.62%  |
| colmap                 | 253.054 MB | 253.054 MB       | 0.00%   |
| plots                  | 149.360 KB | 211.155 KB       | +41.37% |
| nyud                   | 168.467 MB | 168.467 MB       | 0.00%   |
| text_logging           | 2.732 KB   | 3.091 KB         | +13.14% |

### Checklist
* [ ] I have read and agree to [Contributor
Guide](https://github.com/rerun-io/rerun/blob/main/CONTRIBUTING.md) and
the [Code of
Conduct](https://github.com/rerun-io/rerun/blob/main/CODE_OF_CONDUCT.md)
* [ ] I've included a screenshot or gif (if applicable)

<!-- This line will get updated when the PR build summary job finishes.
-->
PR Build Summary: https://build.rerun.io/pr/2412

<!-- pr-link-docs:start -->
Docs preview: https://rerun.io/preview/0c82335/docs
Examples preview: https://rerun.io/preview/0c82335/examples
<!-- pr-link-docs:end -->

---------

Co-authored-by: Emil Ernerfeldt <[email protected]>

.github/workflows/reusable_build_web_demo.yml

@@ -89,7 +89,7 @@ jobs:
         env:
           COMMIT_HASH: ${{ env.SHORT_SHA }}
         run: |
-          python3 scripts/build_demo_app.py --skip-wasm-build
+          python3 scripts/build_demo_app.py --skip-build
 
       - name: Upload web demo assets
         uses: actions/upload-artifact@v3

.vscode/launch.json

@@ -4,6 +4,44 @@
     // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
     "version": "0.2.0",
     "configurations": [
+        {
+            "name": "Launch tests",
+            "type": "lldb",
+            "request": "launch",
+            "cargo": {
+                "args": [
+                    "test",
+                    "-p=re_log_encoding",
+                    "--no-run",
+                    "--lib",
+                    "--all-features"
+                ],
+                "filter": {
+                    "kind": "lib"
+                }
+            },
+            "cwd": "${workspaceFolder}"
+        },
+        {
+            "name": "Debug 'rerun' colmap.rrd from url",
+            "type": "lldb",
+            "request": "launch",
+            "cargo": {
+                "args": [
+                    "build",
+                    "--package=rerun-cli",
+                    "--features=native_viewer"
+                ],
+                "filter": {
+                    "name": "rerun",
+                    "kind": "bin"
+                }
+            },
+            "args": [
+                "https://demo.rerun.io/commit/0f89b62/examples/colmap/data.rrd"
+            ],
+            "cwd": "${workspaceFolder}"
+        },
         {
             "name": "Debug 'rerun' data.rrd",
             "type": "lldb",

crates/re_log_encoding/Cargo.toml

@@ -37,7 +37,7 @@ re_smart_channel.workspace = true
 re_tracing.workspace = true
 
 # External:
-ehttp = "0.2"
+ehttp = { version = "0.3", features = ["streaming"] }
 parking_lot.workspace = true
 thiserror.workspace = true
 web-time.workspace = true

crates/re_log_encoding/src/decoder.rs

@@ -1,7 +1,12 @@
 //! Decoding [`LogMsg`]:es from `.rrd` files/streams.
 
+pub mod stream;
+
 use re_log_types::LogMsg;
 
+use crate::FileHeader;
+use crate::MessageHeader;
+use crate::OLD_RRD_HEADERS;
 use crate::{Compression, EncodingOptions, Serializer};
 
 // ----------------------------------------------------------------------------
@@ -31,7 +36,7 @@ pub enum DecodeError {
     #[error("Not an .rrd file")]
     NotAnRrd,
 
-    #[error("Found an .rrd file from a Rerun version from 0.5.1 or earlier")]
+    #[error("Found an .rrd file from an old, incompatible Rerun version")]
     OldRrdVersion,
 
     #[error("Failed to decode the options: {0}")]
@@ -41,7 +46,7 @@ pub enum DecodeError {
     Read(std::io::Error),
 
     #[error("lz4 error: {0}")]
-    Lz4(std::io::Error),
+    Lz4(lz4_flex::block::DecompressError),
 
     #[error("MsgPack error: {0}")]
     MsgPack(#[from] rmp_serde::decode::Error),
@@ -60,71 +65,50 @@ pub fn decode_bytes(bytes: &[u8]) -> Result<Vec<LogMsg>, DecodeError> {
 
 // ----------------------------------------------------------------------------
 
-enum Decompressor<R: std::io::Read> {
-    Uncompressed(R),
-    Lz4(lz4_flex::frame::FrameDecoder<R>),
-}
+pub fn read_options(bytes: &[u8]) -> Result<EncodingOptions, DecodeError> {
+    let mut read = std::io::Cursor::new(bytes);
 
-impl<R: std::io::Read> Decompressor<R> {
-    fn new(compression: Compression, read: R) -> Self {
-        match compression {
-            Compression::Off => Self::Uncompressed(read),
-            Compression::LZ4 => Self::Lz4(lz4_flex::frame::FrameDecoder::new(read)),
-        }
+    let FileHeader {
+        magic,
+        version,
+        options,
+    } = FileHeader::decode(&mut read)?;
+
+    if OLD_RRD_HEADERS.contains(&magic) {
+        return Err(DecodeError::OldRrdVersion);
+    } else if &magic != crate::RRD_HEADER {
+        return Err(DecodeError::NotAnRrd);
     }
 
-    pub fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), DecodeError> {
-        use std::io::Read as _;
+    warn_on_version_mismatch(version);
 
-        match self {
-            Decompressor::Uncompressed(read) => read.read_exact(buf).map_err(DecodeError::Read),
-            Decompressor::Lz4(lz4) => lz4.read_exact(buf).map_err(DecodeError::Lz4),
-        }
+    match options.serializer {
+        Serializer::MsgPack => {}
     }
-}
 
-// ----------------------------------------------------------------------------
+    Ok(options)
+}
 
 pub struct Decoder<R: std::io::Read> {
-    decompressor: Decompressor<R>,
-    buffer: Vec<u8>,
+    compression: Compression,
+    read: R,
+    uncompressed: Vec<u8>, // scratch space
+    compressed: Vec<u8>,   // scratch space
 }
 
 impl<R: std::io::Read> Decoder<R> {
     pub fn new(mut read: R) -> Result<Self, DecodeError> {
         re_tracing::profile_function!();
 
-        {
-            let mut header = [0_u8; 4];
-            read.read_exact(&mut header).map_err(DecodeError::Read)?;
-            if &header == b"RRF0" {
-                return Err(DecodeError::OldRrdVersion);
-            } else if &header != crate::RRD_HEADER {
-                return Err(DecodeError::NotAnRrd);
-            }
-        }
-
-        {
-            let mut version_bytes = [0_u8; 4];
-            read.read_exact(&mut version_bytes)
-                .map_err(DecodeError::Read)?;
-            warn_on_version_mismatch(version_bytes);
-        }
-
-        let options = {
-            let mut options_bytes = [0_u8; 4];
-            read.read_exact(&mut options_bytes)
-                .map_err(DecodeError::Read)?;
-            EncodingOptions::from_bytes(options_bytes)?
-        };
-
-        match options.serializer {
-            Serializer::MsgPack => {}
-        }
+        let mut data = [0_u8; FileHeader::SIZE];
+        read.read_exact(&mut data).map_err(DecodeError::Read)?;
+        let compression = read_options(&data)?.compression;
 
         Ok(Self {
-            decompressor: Decompressor::new(options.compression, read),
-            buffer: vec![],
+            compression,
+            read,
+            uncompressed: vec![],
+            compressed: vec![],
         })
     }
 }
@@ -135,21 +119,43 @@ impl<R: std::io::Read> Iterator for Decoder<R> {
     fn next(&mut self) -> Option<Self::Item> {
         re_tracing::profile_function!();
 
-        let mut len = [0_u8; 8];
-        self.decompressor.read_exact(&mut len).ok()?;
-        let len = u64::from_le_bytes(len) as usize;
-
-        self.buffer.resize(len, 0);
+        let header = match MessageHeader::decode(&mut self.read) {
+            Ok(header) => header,
+            Err(err) => match err {
+                DecodeError::Read(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                    return None
+                }
+                other => return Some(Err(other)),
+            },
+        };
 
-        {
-            re_tracing::profile_scope!("lz4");
-            if let Err(err) = self.decompressor.read_exact(&mut self.buffer) {
-                return Some(Err(err));
+        match self.compression {
+            Compression::Off => {
+                self.uncompressed
+                    .resize(header.uncompressed_len as usize, 0);
+                if let Err(err) = self.read.read_exact(&mut self.uncompressed) {
+                    return Some(Err(DecodeError::Read(err)));
+                }
+            }
+            Compression::LZ4 => {
+                self.compressed.resize(header.compressed_len as usize, 0);
+                if let Err(err) = self.read.read_exact(&mut self.compressed) {
+                    return Some(Err(DecodeError::Read(err)));
+                }
+                self.uncompressed
+                    .resize(header.uncompressed_len as usize, 0);
+
+                re_tracing::profile_scope!("lz4");
+                if let Err(err) =
+                    lz4_flex::block::decompress_into(&self.compressed, &mut self.uncompressed)
+                {
+                    return Some(Err(DecodeError::Lz4(err)));
+                }
             }
         }
 
         re_tracing::profile_scope!("MsgPack deser");
-        match rmp_serde::from_read(&mut self.buffer.as_slice()) {
+        match rmp_serde::from_slice(&self.uncompressed) {
             Ok(msg) => Some(Ok(msg)),
             Err(err) => Some(Err(err.into())),
         }