fix: Linux PipeWire engine two-level Frame enum + bounded frame channel (all platforms)

src/capturer/engine/linux/mod.rs

@@ -5,7 +5,7 @@ use std::{
         mpsc::{self, sync_channel, SyncSender},
     },
     thread::JoinHandle,
-    time::Duration,
+    time::{Duration, SystemTime},
 };
 
 use pipewire as pw;
@@ -31,7 +31,7 @@ use pw::{
 
 use crate::{
     capturer::Options,
-    frame::{BGRxFrame, Frame, RGBFrame, RGBxFrame, XBGRFrame},
+    frame::{BGRxFrame, Frame, RGBFrame, RGBxFrame, VideoFrame, XBGRFrame},
 };
 
 use self::{error::LinCapError, portal::ScreenCastPortal};
@@ -40,11 +40,16 @@ mod error;
 mod portal;
 
 static CAPTURER_STATE: AtomicU8 = AtomicU8::new(0);
-static STREAM_STATE_CHANGED_TO_ERROR: AtomicBool = AtomicBool::new(false);
+// NOTE: these are process-wide statics; the Linux backend assumes a single active capturer per process.
+// Signals pipewire_capturer's main loop to stop early. Set on both a
+// genuine PipeWire stream error and (see process_callback) when the frame
+// receiver has disconnected -- both are "the loop should end now"
+// conditions, so one flag models the intent accurately rather than two.
+static STREAM_SHOULD_EXIT: AtomicBool = AtomicBool::new(false);
 
 #[derive(Clone)]
 struct ListenerUserData {
-    pub tx: mpsc::Sender<Frame>,
+    pub tx: mpsc::SyncSender<Frame>,
     pub format: spa::param::video::VideoInfoRaw,
 }
 
@@ -85,7 +90,7 @@ fn state_changed_callback(
     match new {
         StreamState::Error(e) => {
             eprintln!("pipewire: State changed to error({e})");
-            STREAM_STATE_CHANGED_TO_ERROR.store(true, std::sync::atomic::Ordering::Relaxed);
+            STREAM_SHOULD_EXIT.store(true, std::sync::atomic::Ordering::Relaxed);
         }
         _ => {}
     }
@@ -118,6 +123,14 @@ fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) {
             if buffer.is_null() {
                 break 'outside;
             }
+            // Once STREAM_SHOULD_EXIT is set (stream error or disconnected
+            // receiver), the outer loop in pipewire_capturer is about to
+            // stop anyway -- skip the frame_data copy (the actual
+            // per-frame cost here) and everything after it, and just
+            // requeue the buffer below.
+            if STREAM_SHOULD_EXIT.load(std::sync::atomic::Ordering::Relaxed) {
+                break 'outside;
+            }
             let timestamp = unsafe { get_timestamp(buffer) };
 
             let n_datas = unsafe { (*buffer).n_datas };
@@ -133,35 +146,85 @@ fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) {
                 .to_vec()
             };
 
-            if let Err(e) = match user_data.format.format() {
-                VideoFormat::RGBx => user_data.tx.send(Frame::RGBx(RGBxFrame {
-                    display_time: timestamp as u64,
-                    width: frame_size.width as i32,
-                    height: frame_size.height as i32,
-                    data: frame_data,
-                })),
-                VideoFormat::RGB => user_data.tx.send(Frame::RGB(RGBFrame {
-                    display_time: timestamp as u64,
-                    width: frame_size.width as i32,
-                    height: frame_size.height as i32,
-                    data: frame_data,
-                })),
-                VideoFormat::xBGR => user_data.tx.send(Frame::XBGR(XBGRFrame {
-                    display_time: timestamp as u64,
-                    width: frame_size.width as i32,
-                    height: frame_size.height as i32,
-                    data: frame_data,
-                })),
-                VideoFormat::BGRx => user_data.tx.send(Frame::BGRx(BGRxFrame {
-                    display_time: timestamp as u64,
-                    width: frame_size.width as i32,
-                    height: frame_size.height as i32,
-                    data: frame_data,
-                })),
-                _ => panic!("Unsupported frame format received"),
-            } {
-                eprintln!("{e}");
+            // `timestamp` (from spa_meta_header.pts) is a PipeWire monotonic
+            // nanosecond count since an arbitrary reference — not wall-clock.
+            // display_time's SystemTime contract is wall-clock, so we use
+            // SystemTime::now() here (matches what the macOS and Windows
+            // engines do today).  Relative frame ordering survives via
+            // channel-send order; sub-millisecond buffer timing is lost.
+            let _ = timestamp; // suppress "unused" warning until we wire pts elsewhere
+            let display_time = SystemTime::now();
+
+            // `try_send`, not `send`: this callback runs on the same thread
+            // that `pipewire_capturer`'s loop uses to poll CAPTURER_STATE
+            // between `pw_loop.iterate()` calls. A blocking `send()` on a
+            // bounded channel would stall this thread whenever the consumer
+            // falls behind, and since `LinuxCapturer::stop_capture()` joins
+            // this exact thread, a full channel + a paused consumer would
+            // make `stop_capture()` hang forever. Dropping the frame under
+            // backpressure (rather than blocking the producer) keeps both
+            // the bounded-memory guarantee and a responsive stop path.
+            let send_result = match user_data.format.format() {
+                VideoFormat::RGBx => {
+                    user_data
+                        .tx
+                        .try_send(Frame::Video(VideoFrame::RGBx(RGBxFrame {
+                            display_time,
+                            width: frame_size.width as i32,
+                            height: frame_size.height as i32,
+                            data: frame_data,
+                        })))
+                }
+                VideoFormat::RGB => {
+                    user_data
+                        .tx
+                        .try_send(Frame::Video(VideoFrame::RGB(RGBFrame {
+                            display_time,
+                            width: frame_size.width as i32,
+                            height: frame_size.height as i32,
+                            data: frame_data,
+                        })))
+                }
+                VideoFormat::xBGR => {
+                    user_data
+                        .tx
+                        .try_send(Frame::Video(VideoFrame::XBGR(XBGRFrame {
+                            display_time,
+                            width: frame_size.width as i32,
+                            height: frame_size.height as i32,
+                            data: frame_data,
+                        })))
+                }
+                VideoFormat::BGRx => {
+                    user_data
+                        .tx
+                        .try_send(Frame::Video(VideoFrame::BGRx(BGRxFrame {
+                            display_time,
+                            width: frame_size.width as i32,
+                            height: frame_size.height as i32,
+                            data: frame_data,
+                        })))
+                }
+                _ => {
+                    if !STREAM_SHOULD_EXIT.swap(true, std::sync::atomic::Ordering::Relaxed) {
+                        eprintln!("Unsupported frame format received");
+                    }
+                    Ok(())
+                }
+            };
+
+            if let Err(mpsc::TrySendError::Disconnected(_)) = send_result {
+                // The consumer (Capturer/rx) is gone, e.g. dropped without
+                // calling stop_capture(). Signal the main loop above to
+                // exit instead of spinning pw_loop.iterate() forever on a
+                // stream nobody will ever read from, and log it once (swap
+                // instead of store) rather than once per incoming frame.
+                if !STREAM_SHOULD_EXIT.swap(true, std::sync::atomic::Ordering::Relaxed) {
+                    eprintln!("Frame receiver disconnected");
+                }
             }
+            // TrySendError::Full is a silent intentional drop under
+            // backpressure — see comment above.
         }
     } else {
         eprintln!("Out of buffers");
@@ -173,10 +236,24 @@ fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) {
 // TODO: Format negotiation
 fn pipewire_capturer(
     options: Options,
-    tx: mpsc::Sender<Frame>,
+    tx: mpsc::SyncSender<Frame>,
     ready_sender: &SyncSender<bool>,
     stream_id: u32,
 ) -> Result<(), LinCapError> {
+    // STREAM_SHOULD_EXIT is a process-wide static; the only other writer
+    // is process_callback/state_changed_callback (both set it to `true`).
+    // Reset it here, before stream.connect() or any loop iteration, not
+    // just before the main capture loop below: connect()'s handshake can
+    // plausibly dispatch pipewire callbacks on this same thread before we
+    // ever call pw_loop.iterate() ourselves (MainLoop wraps pw_main_loop
+    // in an Rc, not Arc -- there is no separate background dispatch
+    // thread, so any dispatch that happens before our own iterate() calls
+    // must be happening synchronously inside calls like connect()).
+    // Resetting only right before the main loop (as an earlier commit did)
+    // would silently clear a real pre-start error or disconnect instead of
+    // letting it end the loop immediately, as it should.
+    STREAM_SHOULD_EXIT.store(false, std::sync::atomic::Ordering::Relaxed);
+
     pw::init();
 
     let mainloop = MainLoop::new(None)?;
@@ -299,8 +376,8 @@ fn pipewire_capturer(
 
     // User has called Capturer::start() and we start the main loop
     while CAPTURER_STATE.load(std::sync::atomic::Ordering::Relaxed) == 1
-        && /* If the stream state got changed to `Error`, we exit. TODO: tell user that we exited */
-          !STREAM_STATE_CHANGED_TO_ERROR.load(std::sync::atomic::Ordering::Relaxed)
+        && /* Exit early on a PipeWire stream error or a disconnected frame receiver. TODO: tell user that we exited */
+          !STREAM_SHOULD_EXIT.load(std::sync::atomic::Ordering::Relaxed)
     {
         pw_loop.iterate(Duration::from_millis(100));
     }
@@ -317,7 +394,20 @@ pub struct LinuxCapturer {
 
 impl LinuxCapturer {
     // TODO: Error handling
-    pub fn new(options: &Options, tx: mpsc::Sender<Frame>) -> Self {
+    pub fn new(options: &Options, tx: mpsc::SyncSender<Frame>) -> Self {
+        // Same class of bug as STREAM_SHOULD_EXIT (see the reset in
+        // pipewire_capturer): CAPTURER_STATE is a process-wide static, and
+        // the only other writer is stop_capture(), which never runs if a
+        // previous LinuxCapturer's receiver was dropped instead of properly
+        // stopped. Left stale at 1 (or 2), a new instance's background
+        // thread would skip its "wait for start_capture()" gate below and
+        // start iterating immediately -- capturing before the caller ever
+        // called start_capture() on *this* instance. Reset synchronously
+        // here, before the background thread exists and before the caller
+        // can possibly call start_capture() on the handle this returns, so
+        // there's no race with either.
+        CAPTURER_STATE.store(0, std::sync::atomic::Ordering::Relaxed);
+
         let connection =
             dbus::blocking::Connection::new_session().expect("Failed to create dbus connection");
         let stream_id = ScreenCastPortal::new(&connection)
@@ -360,10 +450,10 @@ impl LinuxCapturer {
             }
         }
         CAPTURER_STATE.store(0, std::sync::atomic::Ordering::Relaxed);
-        STREAM_STATE_CHANGED_TO_ERROR.store(false, std::sync::atomic::Ordering::Relaxed);
+        STREAM_SHOULD_EXIT.store(false, std::sync::atomic::Ordering::Relaxed);
     }
 }
 
-pub fn create_capturer(options: &Options, tx: mpsc::Sender<Frame>) -> LinuxCapturer {
+pub fn create_capturer(options: &Options, tx: mpsc::SyncSender<Frame>) -> LinuxCapturer {
     LinuxCapturer::new(options, tx)
 }

src/capturer/engine/mac/mod.rs

@@ -53,7 +53,14 @@ impl sc::stream::DelegateImpl for ErrorHandler {
 
 #[repr(C)]
 pub struct CapturerInner {
-    pub tx: mpsc::Sender<ChannelItem>,
+    pub tx: mpsc::SyncSender<ChannelItem>,
+    // Set once tx.try_send() observes a disconnected receiver, so later
+    // callbacks can skip the retain+send work entirely instead of repeating
+    // it with nowhere for the result to go. There's no return-an-error path
+    // here (unlike Windows' on_frame_arrived) to actually stop the
+    // sc::Stream from this callback, so this is the same "flip a shared
+    // flag" compromise used for the equivalent case on Linux/Windows.
+    disconnected: AtomicBool,
 }
 
 define_obj_type!(pub Capturer + StreamOutputImpl, CapturerInner, CAPTURER);
@@ -69,7 +76,27 @@ impl sc::stream::OutputImpl for Capturer {
         sample_buf: &mut cm::SampleBuf,
         kind: sc::OutputType,
     ) {
-        let _ = self.inner_mut().tx.send((sample_buf.retained(), kind));
+        let inner = self.inner_mut();
+        if inner
+            .disconnected
+            .load(std::sync::atomic::Ordering::Relaxed)
+        {
+            return;
+        }
+
+        // try_send, not send: don't block ScreenCaptureKit's delivery queue
+        // when the consumer is behind — drop the sample buffer instead
+        // (same rationale as the Linux/Windows engines).
+        if let Err(mpsc::TrySendError::Disconnected(_)) =
+            inner.tx.try_send((sample_buf.retained(), kind))
+        {
+            if !inner
+                .disconnected
+                .swap(true, std::sync::atomic::Ordering::Relaxed)
+            {
+                eprintln!("Frame receiver disconnected");
+            }
+        }
     }
 }
 
@@ -85,7 +112,7 @@ pub(crate) enum CreateCapturerError {
 
 pub(crate) fn create_capturer(
     options: &Options,
-    tx: mpsc::Sender<ChannelItem>,
+    tx: mpsc::SyncSender<ChannelItem>,
     error_flag: Arc<AtomicBool>,
 ) -> Result<(arc::R<Capturer>, arc::R<ErrorHandler>, arc::R<sc::Stream>), CreateCapturerError> {
     // If no target is specified, capture the main display
@@ -187,7 +214,10 @@ pub(crate) fn create_capturer(
     let error_handler = ErrorHandler::with(ErrorHandlerInner { error_flag });
     let stream = sc::Stream::with_delegate(&filter, &stream_config, error_handler.as_ref());
 
-    let capturer = CapturerInner { tx };
+    let capturer = CapturerInner {
+        tx,
+        disconnected: AtomicBool::new(false),
+    };
 
     let queue = dispatch::Queue::serial_with_ar_pool();
 

src/capturer/engine/mod.rs

@@ -58,7 +58,7 @@ pub struct Engine {
 }
 
 impl Engine {
-    pub fn new(options: &Options, tx: mpsc::Sender<ChannelItem>) -> Engine {
+    pub fn new(options: &Options, tx: mpsc::SyncSender<ChannelItem>) -> Engine {
         #[cfg(target_os = "macos")]
         {
             let error_flag = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));