playback issues

sam-tool-tauri/src-tauri/src/decoder.rs

@@ -309,6 +309,20 @@ impl Drop for DecoderSession {
     }
 }
 
+/// Pack a frame for the JS wire: 8-byte little-endian i64 frame index
+/// followed by the raw JPEG. The backend is authoritative on which frame
+/// actually decoded — JS reads the prefix to bind paints to the real frame
+/// number rather than the optimistic request, which matters when codec
+/// keyframe-snap or seek-ahead returns something other than the requested
+/// index. (Today our ffmpeg path is exact-seek so they always match, but
+/// the prefix makes the system robust to that ever changing.)
+pub fn pack_frame(idx: u64, jpeg: &[u8]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(8 + jpeg.len());
+    out.extend_from_slice(&(idx as i64).to_le_bytes());
+    out.extend_from_slice(jpeg);
+    out
+}
+
 /// Pick a VAAPI render node. `SAM_TOOL_VAAPI_DEVICE` overrides; otherwise we
 /// probe `/dev/dri/renderD128..199` for the first that exists. Multi-GPU
 /// systems can end up with the integrated card on D128 and a discrete card

sam-tool-tauri/src-tauri/src/lib.rs

@@ -6,6 +6,7 @@ mod extract;
 mod sam;
 mod settings;
 mod state;
+mod stream;
 mod user;
 mod video;
 
@@ -62,12 +63,15 @@ fn current_video(state: State<'_, AppState>) -> Result<Option<VideoInfo>, String
 
 #[tauri::command]
 fn decode_frame(state: State<'_, AppState>, frame_idx: u64) -> Result<Response, String> {
-    let bytes = {
+    let jpeg = {
         let mut guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?;
         let decoder = guard.as_mut().ok_or("no video loaded")?;
         decoder.decode_at(frame_idx)?
     };
-    Ok(Response::new(bytes))
+    // Wire format: 8-byte LE frame number + JPEG bytes. JS unpacks the
+    // prefix in the worker so paints bind to the backend-authoritative
+    // index, not the optimistic request.
+    Ok(Response::new(decoder::pack_frame(frame_idx, &jpeg)))
 }
 
 #[tauri::command]
@@ -309,6 +313,10 @@ pub fn run() {
             sam_health_check,
             sam_list_models,
             sam_segment_bbox,
+            stream::start_stream,
+            stream::set_stream_state,
+            stream::stop_stream,
+            stream::stream_current_frame,
         ])
         .run(tauri::generate_context!())
         .expect("error while running tauri application");

sam-tool-tauri/src-tauri/src/state.rs

@@ -1,14 +1,22 @@
 use crate::decoder::DecoderSession;
-use std::sync::Mutex;
+use crate::stream::StreamControl;
+use std::sync::{Arc, Mutex};
 
 pub struct AppState {
-    pub decoder: Mutex<Option<DecoderSession>>,
+    /// Wrapped in Arc so the stream producer thread can hold its own
+    /// handle and reach back into the decoder without going through Tauri.
+    pub decoder: Arc<Mutex<Option<DecoderSession>>>,
+    /// Atomic control struct read by the stream producer each tick. JS
+    /// mutates via tauri commands; the producer polls it for play / pause /
+    /// speed / target / stride.
+    pub stream_control: Arc<StreamControl>,
 }
 
 impl AppState {
     pub fn new() -> Self {
         Self {
-            decoder: Mutex::new(None),
+            decoder: Arc::new(Mutex::new(None)),
+            stream_control: Arc::new(StreamControl::new()),
         }
     }
 }

sam-tool-tauri/src-tauri/src/stream.rs

@@ -0,0 +1,208 @@
+//! Forward playback transport. A Rust producer thread decodes via the
+//! existing DecoderSession and pushes packed frames to the JS side via a
+//! Tauri Channel — eliminates the per-frame IPC round-trip that pure
+//! `invoke()` playback paid for. Backward / step / scrub / pause stay on
+//! the invoke-pull path (see lib.rs::decode_frame).
+
+use crate::decoder::{pack_frame, DecoderSession};
+use crate::state::AppState;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::{Duration, Instant};
+use tauri::ipc::{Channel, InvokeResponseBody};
+use tauri::State;
+
+pub struct StreamControl {
+    /// Bumped on each `start_stream`. Producer threads check this and exit
+    /// when it changes — guarantees only one producer at a time.
+    pub session_id: AtomicU64,
+    /// True while the producer should keep emitting. JS clears via
+    /// `set_stream_state` / `stop_stream` to pause.
+    pub playing: AtomicBool,
+    /// Speed multiplier × 100 (atomic u64; floats aren't atomic).
+    pub speed_x100: AtomicU64,
+    /// Next frame to emit. JS may overwrite this for in-stream seeks.
+    pub current_frame: AtomicU64,
+    /// Advance step (1 = every frame, N = every Nth — frame_skip).
+    pub stride: AtomicU64,
+    /// Stop when current_frame >= total_frames.
+    pub total_frames: AtomicU64,
+}
+
+impl StreamControl {
+    pub fn new() -> Self {
+        Self {
+            session_id: AtomicU64::new(0),
+            playing: AtomicBool::new(false),
+            speed_x100: AtomicU64::new(100),
+            current_frame: AtomicU64::new(0),
+            stride: AtomicU64::new(1),
+            total_frames: AtomicU64::new(0),
+        }
+    }
+}
+
+#[tauri::command]
+pub fn start_stream(
+    state: State<'_, AppState>,
+    channel: Channel<InvokeResponseBody>,
+    from_idx: u64,
+    speed: f64,
+    stride: u64,
+) -> Result<u64, String> {
+    let (total, fps) = {
+        let guard = state.decoder.lock().map_err(|e| format!("lock: {e}"))?;
+        let info = guard.as_ref().ok_or("no video loaded")?.info();
+        (info.total_frames, info.fps)
+    };
+    if from_idx >= total {
+        return Err("from_idx beyond total_frames".into());
+    }
+
+    let control = state.stream_control.clone();
+    // fetch_add returns the previous value; +1 = our new id. Any existing
+    // producer with the prior id will exit on its next iteration.
+    let sid = control.session_id.fetch_add(1, Ordering::SeqCst) + 1;
+    control
+        .speed_x100
+        .store((speed.max(0.01) * 100.0) as u64, Ordering::SeqCst);
+    control.current_frame.store(from_idx, Ordering::SeqCst);
+    control.stride.store(stride.max(1), Ordering::SeqCst);
+    control.total_frames.store(total, Ordering::SeqCst);
+    // playing must be the LAST flag we set — once it's true the producer
+    // can start consuming the other fields.
+    control.playing.store(true, Ordering::SeqCst);
+
+    let decoder = state.decoder.clone();
+    thread::spawn(move || {
+        producer_loop(decoder, control, channel, sid, fps);
+    });
+
+    Ok(sid)
+}
+
+#[tauri::command]
+pub fn set_stream_state(
+    state: State<'_, AppState>,
+    playing: Option<bool>,
+    speed: Option<f64>,
+    target_frame: Option<u64>,
+    stride: Option<u64>,
+) -> Result<(), String> {
+    let c = &state.stream_control;
+    if let Some(s) = speed {
+        c.speed_x100
+            .store((s.max(0.01) * 100.0) as u64, Ordering::SeqCst);
+    }
+    if let Some(t) = target_frame {
+        c.current_frame.store(t, Ordering::SeqCst);
+    }
+    if let Some(s) = stride {
+        c.stride.store(s.max(1), Ordering::SeqCst);
+    }
+    if let Some(p) = playing {
+        c.playing.store(p, Ordering::SeqCst);
+    }
+    Ok(())
+}
+
+#[tauri::command]
+pub fn stop_stream(state: State<'_, AppState>) -> Result<(), String> {
+    let c = &state.stream_control;
+    c.playing.store(false, Ordering::SeqCst);
+    // Bump session_id too so a producer that misses the playing flag still
+    // exits on its session check.
+    c.session_id.fetch_add(1, Ordering::SeqCst);
+    Ok(())
+}
+
+#[tauri::command]
+pub fn stream_current_frame(state: State<'_, AppState>) -> Result<u64, String> {
+    Ok(state.stream_control.current_frame.load(Ordering::Relaxed))
+}
+
+fn producer_loop(
+    decoder: Arc<Mutex<Option<DecoderSession>>>,
+    control: Arc<StreamControl>,
+    channel: Channel<InvokeResponseBody>,
+    my_sid: u64,
+    fps: f64,
+) {
+    let mut next_tick = Instant::now();
+    loop {
+        if control.session_id.load(Ordering::Relaxed) != my_sid {
+            return;
+        }
+        if !control.playing.load(Ordering::Relaxed) {
+            return;
+        }
+
+        let speed = (control.speed_x100.load(Ordering::Relaxed) as f64 / 100.0).max(0.01);
+        let stride = control.stride.load(Ordering::Relaxed).max(1);
+        let total = control.total_frames.load(Ordering::Relaxed);
+        let frame = control.current_frame.load(Ordering::Relaxed);
+
+        if total == 0 || frame >= total {
+            control.playing.store(false, Ordering::SeqCst);
+            return;
+        }
+
+        // Decode this frame via the existing DecoderSession (preserves
+        // hwaccel fallback + stderr capture + frame cache).
+        let jpeg = {
+            let mut guard = match decoder.lock() {
+                Ok(g) => g,
+                Err(_) => return,
+            };
+            match guard.as_mut() {
+                Some(d) => d.decode_at(frame),
+                None => Err("no video loaded".into()),
+            }
+        };
+
+        let send_ok = match jpeg {
+            Ok(bytes) => {
+                let packed = pack_frame(frame, &bytes);
+                channel.send(InvokeResponseBody::Raw(packed)).is_ok()
+            }
+            Err(_) => {
+                // Surface decode errors as an empty packed frame so JS knows
+                // to stop. With just an 8-byte prefix and no JPEG it's
+                // unambiguous on the receive side.
+                let packed = pack_frame(frame, &[]);
+                let _ = channel.send(InvokeResponseBody::Raw(packed));
+                false
+            }
+        };
+        if !send_ok {
+            // Channel closed (frontend tore down) — stop cleanly.
+            control.playing.store(false, Ordering::SeqCst);
+            return;
+        }
+
+        // Advance the cursor. JS may have already moved it (an in-stream
+        // seek) — in that case the next iteration picks up the new value.
+        let next_frame = frame.saturating_add(stride);
+        // Only advance if no one else moved it (compare_exchange would be
+        // strictly correct, but a plain store is fine: a race here just
+        // means we emit the seeked-to frame on the next iteration).
+        let _ = control.current_frame.compare_exchange(
+            frame,
+            next_frame,
+            Ordering::SeqCst,
+            Ordering::Relaxed,
+        );
+
+        // Pace: stride frames per (fps × speed) per second.
+        let frame_interval = Duration::from_secs_f64(stride as f64 / (fps * speed));
+        next_tick += frame_interval;
+        let now = Instant::now();
+        if next_tick > now {
+            thread::sleep(next_tick - now);
+        } else {
+            // Behind real-time — reset baseline instead of accumulating debt.
+            next_tick = now;
+        }
+    }
+}