Fix ffmpeg respawn churn and ~1s video latency on live H.264 streams

Cargo.lock

@@ -3043,7 +3043,7 @@ dependencies = [
 
 [[package]]
 name = "dimos-viewer"
-version = "0.32.0-alpha.1"
+version = "0.32.0-alpha.2"
 dependencies = [
  "bincode",
  "clap",

crates/utils/re_video/src/decode/ffmpeg_cli/ffmpeg.rs

@@ -245,6 +245,7 @@ impl FFmpegProcessAndListener {
         encoding_details: Option<&VideoEncodingDetails>,
         ffmpeg_path: Option<&std::path::Path>,
         codec: &crate::VideoCodec,
+        low_latency: bool,
     ) -> Result<Self, Error> {
         re_tracing::profile_function!();
 
@@ -311,6 +312,15 @@ impl FFmpegProcessAndListener {
                 "-analyzeduration",
                 "0",
             ])
+            // Frame-multithreading adds roughly one frame of output delay per thread and
+            // ffmpeg defaults to the core count. For live streams a single decode thread
+            // cuts the holdback from ~17 frames to ~3 (measured, 720p H.264), and decode
+            // throughput of one thread is ample for typical stream resolutions.
+            .args(if low_latency {
+                &["-threads", "1"][..]
+            } else {
+                &[][..]
+            })
             // Keep in mind that all arguments that are about the input, need to go before!
             .format(codec_str) // TODO(andreas): should we check ahead of time whether this is available?
             //.fps_mode("0")
@@ -899,6 +909,7 @@ pub struct FFmpegCliDecoder {
     output_sender: Sender<FrameResult>,
     ffmpeg_path: Option<std::path::PathBuf>,
     codec: crate::VideoCodec,
+    low_latency: bool,
 }
 
 impl FFmpegCliDecoder {
@@ -908,6 +919,7 @@ impl FFmpegCliDecoder {
         output_sender: Sender<FrameResult>,
         ffmpeg_path: Option<std::path::PathBuf>,
         codec: &crate::VideoCodec,
+        low_latency: bool,
     ) -> Result<Self, Error> {
         re_tracing::profile_function!();
 
@@ -926,6 +938,7 @@ impl FFmpegCliDecoder {
             encoding_details,
             ffmpeg_path.as_deref(),
             codec,
+            low_latency,
         )?;
 
         Ok(Self {
@@ -934,6 +947,7 @@ impl FFmpegCliDecoder {
             output_sender,
             ffmpeg_path,
             codec: codec.clone(),
+            low_latency,
         })
     }
 }
@@ -990,13 +1004,14 @@ impl AsyncDecoder for FFmpegCliDecoder {
 
     fn reset(&mut self, video_descr: &VideoDataDescription) -> crate::decode::Result<()> {
         re_tracing::profile_function!();
-        re_log::trace!("Resetting ffmpeg decoder {}", self.debug_name);
+        re_log::debug!("Resetting ffmpeg decoder {}", self.debug_name);
         self.ffmpeg = FFmpegProcessAndListener::new(
             &self.debug_name,
             self.output_sender.clone(),
             video_descr.encoding_details.as_ref(),
             self.ffmpeg_path.as_deref(),
             &self.codec,
+            self.low_latency,
         )?;
         Ok(())
     }

crates/utils/re_video/src/decode/mod.rs

@@ -310,6 +310,14 @@ pub fn new_decoder(
             output_sender,
             decode_settings.ffmpeg_path.clone(),
             &video.codec,
+            // Live streams want minimal decode latency: ffmpeg's frame-multithreading
+            // adds roughly one frame of delay per thread (measured: 17 frames held back
+            // at default/auto threads vs 3 with a single thread). For static assets
+            // keep the default for decode throughput.
+            matches!(
+                video.delivery_method,
+                crate::VideoDeliveryMethod::Stream { .. }
+            ),
         )?)),
 
         crate::VideoCodec::ImageSequence(codec) => {

crates/utils/re_video/src/player/mod.rs

@@ -655,17 +655,38 @@ impl<T: Default> VideoPlayer<T> {
             && let Some(keyframe_idx) = video_description.sample_keyframe_idx(last_enqueued)
         {
             if keyframe_idx < requested_keyframe_idx {
-                // Need to reset if we're skipping frames.
-                self.reset(video_description)?;
-                // Skip forward and just enqueue the requested keyframe.
-                self.enqueue_keyframe_range(
-                    video_description,
-                    requested_keyframe_idx,
-                    requested_sample_idx,
-                    get_video_buffer,
-                )?;
-
-                requested_keyframe_idx
+                let requested_gop_start = video_description
+                    .keyframe_indices
+                    .get(requested_keyframe_idx)
+                    .copied();
+
+                if requested_gop_start.is_some_and(|gop_start| last_enqueued + 1 >= gop_start) {
+                    // The enqueued samples lead contiguously into the requested keyframe's GOP.
+                    // This is the normal per-GOP rollover on live streams (the next GOP can never
+                    // be pre-enqueued there, since its samples don't exist yet): the decoder can
+                    // decode straight through the new IDR, no reset needed. The enqueue loop
+                    // below picks up the new keyframe range.
+                    re_log::debug!(
+                        "Continuing video decoder across GOP boundary (last enqueued {last_enqueued}, requested keyframe {requested_keyframe_idx})"
+                    );
+                    keyframe_idx
+                } else {
+                    // Actual gap between the enqueued samples and the requested keyframe's GOP —
+                    // we're skipping frames, so reset.
+                    re_log::debug!(
+                        "Resetting video decoder: gap between enqueued samples and requested keyframe (last enqueued {last_enqueued}, GOP start {requested_gop_start:?})"
+                    );
+                    self.reset(video_description)?;
+                    // Skip forward and just enqueue the requested keyframe.
+                    self.enqueue_keyframe_range(
+                        video_description,
+                        requested_keyframe_idx,
+                        requested_sample_idx,
+                        get_video_buffer,
+                    )?;
+
+                    requested_keyframe_idx
+                }
             } else {
                 keyframe_idx
             }
@@ -768,8 +789,19 @@ impl<T: Default> VideoPlayer<T> {
 
         // Signal the end of the video if we reached it.
         // This is important for some decoders to flush out all the frames.
+        //
+        // Never signal it for stream-delivered video: a live stream has no end, and a
+        // brief ingest gap (>250ms, easily hit under load with a 200ms batcher tick)
+        // would otherwise be mistaken for the end of the video. Once new samples then
+        // arrive, the player has to fully reset the decoder — which for the ffmpeg CLI
+        // decoder means killing and respawning the process, once per gap.
+        // The tradeoff is that during a stall the last few frames stay unflushed in the
+        // decoder, so the frozen image is slightly staler.
         if !self.signaled_end_of_video
-            && !treat_video_as_live_stream(&self.config, video_description)
+            && matches!(
+                video_description.delivery_method,
+                crate::VideoDeliveryMethod::Static { .. }
+            )
             && self.enqueued_last_sample_of_video(video_description)
         {
             re_log::trace!("Signaling end of video");
@@ -797,6 +829,7 @@ impl<T: Default> VideoPlayer<T> {
     ) -> Result<(), VideoPlayerError> {
         // If we haven't decoded anything at all yet, reset the decoder.
         let Some(last_requested) = self.last_requested else {
+            re_log::debug!("Resetting video decoder: nothing requested yet");
             return self.reset(video_description);
         };
 
@@ -812,6 +845,13 @@ impl<T: Default> VideoPlayer<T> {
 
             // For each new (!) error after entering the error state, we reset the decoder.
             // This way, it might later recover from the error as we progress in the video.
+            re_log::debug!(
+                "Resetting video decoder after decode error: {}",
+                self.last_error
+                    .as_ref()
+                    .map(|e| e.latest_error.to_string())
+                    .unwrap_or_default()
+            );
             self.reset(video_description)?;
         }
         // Reset if seeking forward by more than one GOP
@@ -821,6 +861,10 @@ impl<T: Default> VideoPlayer<T> {
                 enqueued_idx < video_description.keyframe_indices[last_keyframe]
             })
         {
+            re_log::debug!(
+                "Resetting video decoder: seeking forward by more than one GOP (last enqueued {:?}, requested sample {requested}, requested keyframe {requested_keyframe})",
+                self.last_enqueued
+            );
             self.reset(video_description)?;
         }
         // Previously signaled the end of the video, but encountering frames that are newer than the last enqueued.

crates/viewer/re_renderer/src/video/mod.rs

@@ -299,6 +299,9 @@ impl Video {
             };
 
             if needs_reset {
+                re_log::debug!(
+                    "Resetting video decoder: out-of-order sample insertion within the active GOP (change {change_start}..{change_end}, delta {size_delta})"
+                );
                 player.reset(&self.video_description).ok_or_log_error_once();
             } else {
                 // Shift player indices to the new index space.

crates/viewer/re_viewer_context/src/cache/video_stream_cache.rs

@@ -333,9 +333,7 @@ impl VideoStreamCache {
 
                 match result {
                     Err(VideoStreamProcessingError::OutOfOrderSamples) => {
-                        re_log::trace_once!(
-                            "Out-of-order chunk detected, performing delta re-merge"
-                        );
+                        re_log::debug!("Out-of-order chunk detected, performing delta re-merge");
                         match handle_out_of_order_chunk(
                             video_data,
                             &mut entry.known_chunk_ranges,

dimos/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dimos-viewer"
-version = "0.32.0-alpha.1"
+version = "0.32.0-alpha.2"
 edition = "2024"
 rust-version = "1.92"
 license = "MIT OR Apache-2.0"

dimos/pyproject.toml

@@ -4,22 +4,22 @@ build-backend = "maturin"
 
 [project]
 name = "dimos-viewer"
-version = "0.32.0a1"
+version = "0.32.0a2"
 description = "Interactive Rerun viewer for DimOS with click-to-navigate support"
 readme = "README.md"
 requires-python = ">=3.10"
-license = {text = "MIT OR Apache-2.0"}
-authors = [{name = "Dimensional Inc.", email = "engineering@dimensional.com"}]
+license = { text = "MIT OR Apache-2.0" }
+authors = [{ name = "Dimensional Inc.", email = "engineering@dimensional.com" }]
 classifiers = [
-    "Development Status :: 4 - Beta",
-    "Intended Audience :: Developers",
-    "Topic :: Scientific/Engineering",
-    "Topic :: Scientific/Engineering :: Visualization",
-    "Programming Language :: Rust",
-    "License :: OSI Approved :: MIT License",
-    "License :: OSI Approved :: Apache Software License",
-    "Operating System :: POSIX :: Linux",
-    "Operating System :: MacOS",
+  "Development Status :: 4 - Beta",
+  "Intended Audience :: Developers",
+  "Topic :: Scientific/Engineering",
+  "Topic :: Scientific/Engineering :: Visualization",
+  "Programming Language :: Rust",
+  "License :: OSI Approved :: MIT License",
+  "License :: OSI Approved :: Apache Software License",
+  "Operating System :: POSIX :: Linux",
+  "Operating System :: MacOS",
 ]
 
 [project.urls]

dimos/src/interaction/handle.rs

@@ -34,8 +34,8 @@ impl InteractionHandle {
             is_2d,
         };
 
-        if let Err(e) = self.tx.send(event) {
-            eprintln!("Failed to send click event: {}", e);
+        if let Err(err) = self.tx.send(event) {
+            eprintln!("Failed to send click event: {err}");
         }
     }
 }

dimos/src/interaction/keyboard.rs

@@ -93,8 +93,8 @@ impl KeyboardHandler {
         // If not engaged, don't capture any keys
         if !self.engaged {
             if self.was_active {
-                if let Err(e) = self.publish_stop() {
-                    re_log::warn!("Failed to send stop on disengage: {e}");
+                if let Err(err) = self.publish_stop() {
+                    re_log::warn!("Failed to send stop on disengage: {err}");
                 }
                 self.was_active = false;
             }
@@ -107,8 +107,8 @@ impl KeyboardHandler {
         // Check for emergency stop (Space key pressed - one-shot action)
         if ctx.input(|i| i.key_pressed(egui::Key::Space)) {
             self.state.reset();
-            if let Err(e) = self.publish_stop() {
-                re_log::warn!("Failed to send emergency stop: {e}");
+            if let Err(err) = self.publish_stop() {
+                re_log::warn!("Failed to send emergency stop: {err}");
             }
             self.was_active = false;
             self.estop_flash = true;
@@ -117,13 +117,13 @@ impl KeyboardHandler {
 
         // Publish twist command if keys are active, or stop if just released
         if self.state.any_active() {
-            if let Err(e) = self.publish_twist() {
-                re_log::warn!("Failed to publish twist command: {e}");
+            if let Err(err) = self.publish_twist() {
+                re_log::warn!("Failed to publish twist command: {err}");
             }
             self.was_active = true;
         } else if self.was_active {
-            if let Err(e) = self.publish_stop() {
-                re_log::warn!("Failed to send stop on key release: {e}");
+            if let Err(err) = self.publish_stop() {
+                re_log::warn!("Failed to send stop on key release: {err}");
             }
             self.was_active = false;
         }

dimos/src/interaction/ws.rs

@@ -24,6 +24,7 @@ use tokio::sync::mpsc;
 pub enum SendError {
     /// The send queue is full; the event was dropped.
     QueueFull,
+
     /// Failed to serialize the event to JSON.
     Serialize(String),
 }
@@ -136,7 +137,7 @@ impl WsPublisher {
     }
 
     fn broadcast(&self, event: WsEvent) -> Result<(), SendError> {
-        let json = serde_json::to_string(&event).map_err(|e| SendError::Serialize(e.to_string()))?;
+        let json = serde_json::to_string(&event).map_err(|err| SendError::Serialize(err.to_string()))?;
         // Non-blocking: error if the channel is full rather than block the UI thread.
         self.tx.try_send(json).map_err(|_| SendError::QueueFull)
     }

docs/websockets.md

@@ -1,4 +1,4 @@
-# dimos-viewer WebSocket Event Stream
+# Dimos-viewer WebSocket event stream
 
 When `dimos-viewer` is started with `--connect`, LCM multicast is not available
 (LCM uses UDP multicast which is limited to the local machine or subnet). Instead,