feat(video-streamer): normalize recording sessions

Cargo.toml

@@ -32,6 +32,7 @@ codegen-units = 1
 lto = true
 
 [patch.crates-io]
+cadeau = { git = "https://github.com/Devolutions/cadeau", rev = "d990b0a96e60226f5107e7838dde2487139c2207" }
 ebml-iterable = { git = "https://github.com/irvingoujAtDevolution/ebml-iterable", tag = "v0.6.3-devo1" }
 tracing-appender = { git = "https://github.com/CBenoit/tracing.git", rev = "42097daf92e683cf18da7639ddccb056721a796c" }
 

crates/video-streamer/Cargo.toml

@@ -13,6 +13,7 @@ bench = ["perf-diagnostics"]
 
 [dependencies]
 anyhow = "1.0"
+bytes = "1"
 futures-util = { version = "0.3", features = ["sink"] }
 tokio = { version = "1.52", features = [
     "io-util",

crates/video-streamer/README.md

@@ -1,58 +1,78 @@
 # video-streamer
 
-This crate takes an unseekable WebM recording (typically from Chrome CaptureStream) and rewrites it into a “fresh” WebM stream that can start playing immediately.
-It does this by parsing the incoming WebM, finding the correct cut point, and re-encoding frames.
-The output stream begins with a keyframe and valid headers.
+`video-streamer` converts one logical recording session into a pull-driven stream of independent VP8 WebM segments.
 
-## Prerequisites
-
-This crate relies on `cadeau` and its XMF backend for VP8/VP9 decode+encode.
-To override which XMF implementation is used at runtime, set `DGATEWAY_LIB_XMF_PATH` to an `xmf.dll` path before running tests or benches.
+The input may contain several append-only clips.
+Each clip may contain VP8 or VP9 and may change resolution.
+The output keeps one transport connection, always uses VP8, and starts a new fixed-size WebM segment at every clip or resolution boundary.
 
-Example:
+## Interface
 
-`$env:DGATEWAY_LIB_XMF_PATH = 'D:\library\cadeau\xmf.dll'`
+Call `stream_session` with a recording event stream and a message transport.
+The transport must implement `Stream<Item = Result<Bytes, E>> + Sink<Bytes, Error = E>`.
 
-## Tests
+```rust
+stream_session(recording_events, transport, SessionConfig::default()).await?;
+```
 
-Run all tests:
+The input must follow this grammar:
 
-`cargo test -p video-streamer`
+```text
+(ClipStarted Bytes* CaughtUp Bytes* ClipEnded)* SessionEnded
+```
 
-Run the WebM streaming correctness suite:
+Use `StartAt::LiveEdge` for the clip that was already growing when a consumer joined.
+The streamer retains only that clip's latest group of pictures until `CaughtUp` arrives.
+Use `StartAt::Beginning` for clips that start after the consumer joins.
 
-`cargo test -p video-streamer --test webm_stream_correctness -- --nocapture`
+The incremental decoder owns incomplete EBML bytes between `Bytes` events.
+The caller never seeks, rolls back, or retries an incomplete element.
+The decoder limits one buffered EBML element and one retained group of pictures to 64 MiB each.
 
-Some tests are marked `#[ignore]` because they require large local assets or are intended for local investigation.
-Run ignored tests with:
+## Wire protocol
 
-`cargo test -p video-streamer -- --ignored --nocapture`
+Each transport item is one complete protocol message.
+For WebSocket use, one item maps to one binary WebSocket message.
+The first byte is its type code.
 
-Test assets live under `testing-assets\`.
+Client messages:
 
-## Logging and diagnostics
+| Code | Message | Payload |
+| ---: | --- | --- |
+| `0` | Start | Empty |
+| `1` | Pull | Empty |
 
-Most detailed diagnostics are compiled out by default to keep production logs clean.
-To include extra diagnostics, build with `perf-diagnostics`:
+Server messages:
 
-`cargo test -p video-streamer --features perf-diagnostics -- --nocapture`
+| Code | Message | Payload |
+| ---: | --- | --- |
+| `0` | Chunk | WebM bytes |
+| `1` | Segment started | `{"codec":"vp8","sequence":N,"width":W,"height":H}` |
+| `2` | Error | `{"error":"UnexpectedError"}` |
+| `3` | Stream ended | Empty |
 
-Then set `RUST_LOG` as needed.
-Example:
+`Start` requests the first `Segment started` message.
+Each `Pull` requests exactly one later server message.
+The next `Segment started` message ends the previous segment implicitly.
+`Stream ended` ends the final segment and the session.
 
-`$env:RUST_LOG = 'video_streamer=trace'`
+Every segment has its own EBML and Tracks headers.
+Every segment begins with a keyframe and keeps one resolution.
 
-## Benchmarks
-
-The main benchmark is `benches\vpx_reencode.rs`.
-Run it with:
+## Prerequisites
 
-`cargo bench -p video-streamer --bench vpx_reencode --features bench -- --nocapture`
+This crate uses `cadeau` and its XMF backend for VP8 and VP9 decoding and VP8 encoding.
+The streamer reads the dimensions of every decoded image so source resolution changes do not depend on codec header parsing.
+Set `DGATEWAY_LIB_XMF_PATH` when the default XMF library is unavailable.
 
-Benchmark output is intentionally quiet by default.
-To print detailed per-run results, set `VIDEO_STREAMER_BENCH_VERBOSE`:
+```powershell
+$env:DGATEWAY_LIB_XMF_PATH = 'D:\library\cadeau\xmf.dll'
+```
 
-`$env:VIDEO_STREAMER_BENCH_VERBOSE = '1'`
+## Checks
 
-To correlate benchmark results with internal timing, also enable `perf-diagnostics` (the `bench` feature enables it).
-This is intentionally a build-time gate so production logs stay clean.
+```powershell
+cargo +nightly fmt --all
+cargo check -p video-streamer --tests
+cargo clippy -p video-streamer --tests -- -D warnings
+```

crates/video-streamer/src/decoder.rs

@@ -0,0 +1,55 @@
+use anyhow::Context as _;
+use cadeau::xmf::vpx::{VpxCodec, VpxDecoder, VpxImage};
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) struct Dimensions {
+    pub width: u32,
+    pub height: u32,
+}
+
+pub(crate) struct DecodedFrame<'decoder> {
+    pub image: VpxImage<'decoder>,
+    pub dimensions: Dimensions,
+}
+
+pub(crate) struct InputDecoder {
+    codec: VpxCodec,
+    threads: u32,
+    decoder: Option<VpxDecoder>,
+}
+
+impl InputDecoder {
+    pub(crate) fn new(codec: VpxCodec, threads: u32) -> Self {
+        Self {
+            codec,
+            threads,
+            decoder: None,
+        }
+    }
+
+    pub(crate) fn decode<'decoder>(&'decoder mut self, data: &[u8]) -> anyhow::Result<DecodedFrame<'decoder>> {
+        if self.decoder.is_none() {
+            self.decoder = Some(
+                VpxDecoder::builder()
+                    .threads(self.threads)
+                    .width(0)
+                    .height(0)
+                    .codec(self.codec)
+                    .build()?,
+            );
+        }
+
+        let decoder = self.decoder.as_mut().context("input decoder is missing")?;
+        decoder.decode(data)?;
+        let image = decoder.next_frame()?;
+        let dimensions = Dimensions {
+            width: image.width(),
+            height: image.height(),
+        };
+        anyhow::ensure!(
+            dimensions.width > 0 && dimensions.height > 0,
+            "decoder returned invalid frame dimensions"
+        );
+        Ok(DecodedFrame { image, dimensions })
+    }
+}

crates/video-streamer/src/lib.rs

@@ -25,7 +25,11 @@ macro_rules! perf_debug {
 
 pub mod config;
 pub mod debug;
+mod decoder;
+mod normalizer;
+mod protocol;
 pub mod reopenable;
+mod session;
 pub(crate) mod streamer;
 
 #[macro_use]
@@ -39,6 +43,8 @@ pub use streamer::reopenable_file::ReOpenableFile;
 pub use streamer::signal_writer::SignalWriter;
 #[rustfmt::skip]
 pub use streamer::webm_stream;
+#[rustfmt::skip]
+pub use session::{RecordingEvent, SessionConfig, StartAt, stream_session};
 
 #[cfg(feature = "bench")]
 pub mod bench_support;