fix(linux): retry on shared memory when every DMA-BUF frame arrives corrupted

electron/native/pipewire-capture/csrc/pw_shim.c

@@ -233,6 +233,8 @@ struct osc_pw_session {
     int import_dmabuf;
     struct osc_dmabuf_map dmabuf_maps[OSC_MAX_DMABUF_MAPS];
     int frame_drop_reports;
+    /* Own budget, so other drops cannot starve the corrupted-frame retry. */
+    int corrupted_frame_reports;
     int capture_issue_reports;
     /* fd whose DMA_BUF_SYNC_START has not been closed by its END yet, or -1.
      * The bracket has to span the on_frame callback, not just osc_read_frame,
@@ -740,6 +742,7 @@ static void osc_on_param_changed(void *userdata, uint32_t id, const struct spa_p
      * set actually in use rather than a set that no longer exists. */
     session->buffer_info_reports = 0;
     session->frame_drop_reports = 0;
+    session->corrupted_frame_reports = 0;
     session->capture_issue_reports = 0;
 
     api.stream_update_params(session->stream, params, SPA_N_ELEMENTS(params));
@@ -1033,11 +1036,13 @@ static void osc_report_frame_drop(struct osc_pw_session *session, const struct s
                                   enum osc_frame_bounds_error error)
 {
     const char *reason = osc_frame_bounds_error_name(error);
+    int *reports = error == OSC_FRAME_BOUNDS_CORRUPTED ? &session->corrupted_frame_reports
+                                                       : &session->frame_drop_reports;
 
-    if (session->frame_drop_reports >= OSC_FRAME_DROP_REPORTS) {
+    if (*reports >= OSC_FRAME_DROP_REPORTS) {
         return;
     }
-    session->frame_drop_reports++;
+    (*reports)++;
 
     /*
      * On the event stream, not only in a debug log. Every reason this reports is
@@ -1463,10 +1468,16 @@ static int osc_read_frame(struct osc_pw_session *session, const struct spa_buffe
     stride = data->chunk->stride;
 
     if (is_dmabuf_import) {
+        /* Reported like the CPU path does, so a stream that only ever delivers
+         * corrupted frames can be renegotiated on shared memory (issue #1503). */
+        if (SPA_FLAG_IS_SET(chunk_flags, SPA_CHUNK_FLAG_CORRUPTED)) {
+            osc_report_frame_drop(session, data, mapped_len, available, chunk_offset,
+                                  chunk_size, chunk_flags, stride, OSC_FRAME_BOUNDS_CORRUPTED);
+            return 0;
+        }
         /* A zero-sized chunk is how a compositor ships a cursor update with no new
          * frame attached. Not an error, just not a frame. */
-        if (chunk_size == 0 || height <= 0 ||
-            SPA_FLAG_IS_SET(chunk_flags, SPA_CHUNK_FLAG_CORRUPTED)) {
+        if (chunk_size == 0 || height <= 0) {
             return 0;
         }
         /*

electron/native/pipewire-capture/src/capture.rs

@@ -43,6 +43,7 @@ use crate::ffmpeg as ff;
 use crate::shim::{self, AudioRing};
 
 /// An audio capture to mux alongside the video.
+#[derive(Clone)]
 pub struct AudioSource {
     /// "system" or "microphone" — the label a warning names.
     pub label: &'static str,
@@ -1159,6 +1160,35 @@ mod tests {
         let _ = std::fs::remove_file(&output);
     }
 
+    #[test]
+    fn a_paused_capture_resumed_before_it_is_dropped_leaves_its_rings_open() {
+        // The shared-memory retry drops an unstarted capture but keeps its
+        // rings for the replacement. If that capture was paused, it must be
+        // resumed first, or the rings keep refusing samples for the next one.
+        let ring = Arc::new(AudioRing::new(1, AUDIO_SAMPLE_RATE as usize, AUDIO_CHANNELS));
+        let output = std::env::temp_dir().join("openscreen-capture-retire-paused.mp4");
+        let (mut capture, _) = Capture::start(
+            &output,
+            320,
+            240,
+            30,
+            Some(1_000_000),
+            Some(Backend::Software),
+            vec![AudioSource { label: "system", ring: ring.clone(), gain: 1.0, bitrate: 128_000 }],
+            None,
+        )
+        .expect("start");
+        capture.pause();
+        capture.resume();
+        drop(capture);
+
+        ring.push_for_test(&[0.5, -0.5]);
+        let mut drained = Vec::new();
+        ring.drain_into(&mut drained);
+        assert!(!drained.is_empty(), "the ring must take samples after the capture is retired");
+        let _ = std::fs::remove_file(&output);
+    }
+
     #[test]
     fn microphone_gain_clamps_instead_of_wrapping() {
         // A boosted microphone that clips must clip flat. An out-of-range float

electron/native/pipewire-capture/src/main.rs

@@ -42,7 +42,7 @@ use std::collections::HashSet;
 use std::io::{BufRead, Write};
 use std::os::fd::OwnedFd;
 use std::path::PathBuf;
-use std::sync::mpsc::{self, RecvTimeoutError, Sender};
+use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -656,6 +656,74 @@ fn should_retry_shared_memory(
         && error.is_some_and(|message| message.contains("alloc buffers"))
 }
 
+/// niri on the NVIDIA driver negotiates DMA-BUF, then marks every chunk
+/// corrupted with size 0 (issue #1503), so no frame ever reaches the encoder.
+/// Under the shim's five drop reports per stream, so the count is reached.
+const CORRUPTED_FRAMES_BEFORE_RETRY: u32 = 3;
+
+fn is_corrupted_dmabuf_frame(code: &str, detail: &str) -> bool {
+    code == "frame-dropped"
+        && detail.contains("(producer-marked-frame-corrupted)")
+        && detail.contains("dataType=DmaBuf")
+}
+
+/// Drops the failed stream and reconnects with the portal's independent
+/// remote, offering the shm format alone; that attempt arms no further retry.
+#[allow(clippy::too_many_arguments)]
+fn retry_on_shared_memory<W: Write>(
+    emitter: &mut Emitter<W>,
+    receiver: &Receiver<Message>,
+    sender: &Sender<Message>,
+    frames: &Option<Arc<shim::FrameMailbox>>,
+    session: &mut Option<shim::Session>,
+    portal_stream: &mut Option<StreamInfo>,
+    granted_kind: &mut Option<portal::SourceKind>,
+    fallback: SharedMemoryFallback,
+    shared_memory_fallback: &mut Option<SharedMemoryFallback>,
+    reason: &str,
+) -> Result<(), ()> {
+    // Joined before reconnecting, so everything the old stream sent is queued.
+    // Its events are dropped, or a late error would end the new stream; the
+    // rest goes back on the channel.
+    drop(session.take());
+    // A frame still in the mailbox, and the buffers queued for return, belong
+    // to the stream just dropped: neither may reach the new one.
+    if let Some(mailbox) = frames.as_ref() {
+        let _ = mailbox.take();
+        let _ = mailbox.drain_requeue();
+    }
+    for message in receiver.try_iter().collect::<Vec<_>>() {
+        if !matches!(message, Message::Stream(_)) {
+            let _ = sender.send(message);
+        }
+    }
+    *portal_stream = None;
+    let _ = emitter.emit(&Event::Warning {
+        code: "capture-retry-shared-memory".to_owned(),
+        message: reason.to_owned(),
+    });
+    let retry = portal::PortalStream {
+        fd: fallback.fd,
+        fallback_fd: None,
+        node_id: fallback.node_id,
+        position: fallback.position,
+        size: fallback.size,
+        source_kind: fallback.source_kind,
+    };
+    begin_stream(
+        emitter,
+        sender,
+        frames,
+        session,
+        portal_stream,
+        granted_kind,
+        retry,
+        false,
+        true,
+        shared_memory_fallback,
+    )
+}
+
 /// A stream error ends the helper only before capture has started, or the app
 /// waits on a recording that never starts. "Started" is the first staged frame
 /// in video mode, and `streaming` in cursor-only mode, which has no `Capture`.
@@ -695,6 +763,8 @@ fn run<W: Write>(
     // Retain the portal grant while the preferred DMA-BUF format is starting,
     // so a buffer-allocation failure can retry shared-memory-first once.
     let mut shared_memory_fallback: Option<SharedMemoryFallback> = None;
+    // DMA-BUF frames the producer marked corrupted before any usable one.
+    let mut corrupted_frames = 0;
     // `record` has been received. Latched, because it can arrive before the
     // picker has been answered.
     let mut armed = false;
@@ -740,7 +810,7 @@ fn run<W: Write>(
     // has settled by the time the first video frame arrives. Everything they
     // record before that first frame is discarded — see `Capture::stage`.
     // `_audio_sessions` is bound, not dropped: dropping one stops its thread.
-    let (_audio_sessions, mut audio_sources) = start_audio(emitter, &config.audio);
+    let (_audio_sessions, audio_sources) = start_audio(emitter, &config.audio);
     // Buffered until the format is negotiated: the encoder cannot be opened
     // before the frame size is known, and `pause` can arrive first.
     let mut paused = false;
@@ -937,7 +1007,9 @@ fn run<W: Write>(
                                 config.fps,
                                 config.bitrate,
                                 config.forced_encoder,
-                                std::mem::take(&mut audio_sources),
+                                // Cloned, not taken: an unstarted capture is rebuilt
+                                // when the stream falls back to shared memory.
+                                audio_sources.clone(),
                                 frame.dmabuf.as_ref(),
                             ) {
                                 Ok((started, selection)) => {
@@ -970,6 +1042,7 @@ fn run<W: Write>(
                     let Some(capture) = capture.as_mut() else {
                         // Cursor-only session: no encoder, nothing to stage. The
                         // recycle below still has to run.
+                        shared_memory_fallback = None;
                         mailbox.recycle(frame.pixels);
                         continue;
                     };
@@ -1041,6 +1114,9 @@ fn run<W: Write>(
                     // Only once a frame has actually staged — a first frame that
                     // dropped leaves capture unstarted, so this waits for a real one.
                     if first && capture.started() {
+                        // A frame staged: the corrupted-frame retry is moot. Not
+                        // earlier, as a first frame can still fail its GPU import.
+                        shared_memory_fallback = None;
                         let _ = emitter.emit(&Event::CaptureStarted {
                             timestamp_ms: timestamp_ms(),
                             path: config
@@ -1253,10 +1329,52 @@ fn run<W: Write>(
             // Without this the only honest symptom is a file with no frames in
             // it and nothing anywhere saying why.
             Ok(Message::Stream(StreamEvent::CaptureIssue { code, detail })) => {
+                let corrupted = is_corrupted_dmabuf_frame(&code, &detail);
                 let _ = emitter.emit(&Event::Warning {
                     code,
                     message: detail,
                 });
+                if corrupted && shared_memory_fallback.is_some() {
+                    corrupted_frames += 1;
+                    if corrupted_frames >= CORRUPTED_FRAMES_BEFORE_RETRY {
+                        let fallback = shared_memory_fallback.take().expect("checked above");
+                        // Nothing of the retired stream carries over: a frame
+                        // held for its crop, its crop wait, its streaming edge
+                        // (the click gate), and an encoder that one of its
+                        // frames opened but never started, which may be bound to
+                        // the DMA-BUF importer the shm stream cannot feed.
+                        if let Some(held) = held_frame.take() {
+                            if let Some(mailbox) = frames.as_ref() {
+                                mailbox.recycle(held.pixels);
+                            }
+                        }
+                        crop_wait_since = None;
+                        streaming_since = None;
+                        // Its audio rings outlive it through `audio_sources`. A
+                        // pause closed them; reopen them before it goes, or a
+                        // Resume that lands before the replacement opens has no
+                        // capture to resume and the new one records no audio.
+                        // The replacement re-pauses them if `paused` still holds.
+                        if let Some(mut retired) = capture.take() {
+                            retired.resume();
+                        }
+                        if let Err(()) = retry_on_shared_memory(
+                            emitter,
+                            &receiver,
+                            &sender,
+                            &frames,
+                            &mut session,
+                            &mut portal_stream,
+                            &mut granted_kind,
+                            fallback,
+                            &mut shared_memory_fallback,
+                            "The compositor marks every DMA-BUF frame corrupted; retrying screen capture with shared-memory buffers.",
+                        ) {
+                            exit_code = 1;
+                            break;
+                        }
+                    }
+                }
             }
 
             Ok(Message::Stream(StreamEvent::State { state, error })) => {
@@ -1281,9 +1399,12 @@ fn run<W: Write>(
                     if streaming_since.is_none() {
                         streaming_since = Some(timestamp_ms());
                     }
-                    // No retry is appropriate after the compositor accepted
-                    // the stream; release the extra portal fd immediately.
-                    shared_memory_fallback = None;
+                    // No allocation retry is appropriate after the compositor
+                    // accepted the stream. A video session keeps the grant until
+                    // its first frame, for the corrupted-frame retry (#1503).
+                    if frames.is_none() {
+                        shared_memory_fallback = None;
+                    }
                 } else if state == "unconnected" {
                     streaming_since = None;
                 }
@@ -1296,42 +1417,21 @@ fn run<W: Write>(
                     shared_memory_fallback.is_some(),
                 ) {
                     let fallback = shared_memory_fallback.take().expect("checked above");
-                    // Drop and join the failed stream before reconnecting with
-                    // the portal's independent remote. The second attempt
-                    // offers the shm format alone; it will not arm another retry.
-                    drop(session.take());
-                    portal_stream = None;
-                    let _ = emitter.emit(&Event::Warning {
-                        code: "capture-retry-shared-memory".to_owned(),
-                        message: "PipeWire buffer allocation failed; retrying screen capture with shared-memory buffers."
-                            .to_owned(),
-                    });
-                    let retry = portal::PortalStream {
-                        fd: fallback.fd,
-                        fallback_fd: None,
-                        node_id: fallback.node_id,
-                        position: fallback.position,
-                        size: fallback.size,
-                        source_kind: fallback.source_kind,
-                    };
-                    if let Err(()) = begin_stream(
+                    if let Err(()) = retry_on_shared_memory(
                         emitter,
+                        &receiver,
                         &sender,
                         &frames,
                         &mut session,
                         &mut portal_stream,
                         &mut granted_kind,
-                        retry,
-                        false,
-                        true,
+                        fallback,
                         &mut shared_memory_fallback,
+                        "PipeWire buffer allocation failed; retrying screen capture with shared-memory buffers.",
                     ) {
                         exit_code = 1;
                         break;
                     }
-                    // `begin_stream` already replaced the session and info;
-                    // any queued messages from the old stream precede events
-                    // from this new PipeWire thread.
                     continue;
                 }
                 if let Some(error) = error {
@@ -1905,7 +2005,8 @@ mod microphone_resolution_tests {
 #[cfg(test)]
 mod capture_fallback_tests {
     use super::{
-        holds_for_crop, should_retry_shared_memory, stream_error_is_fatal, MAX_CROP_WAIT,
+        holds_for_crop, is_corrupted_dmabuf_frame, should_retry_shared_memory,
+        stream_error_is_fatal, MAX_CROP_WAIT,
     };
     use std::time::Duration;
 
@@ -1960,6 +2061,24 @@ mod capture_fallback_tests {
         ));
     }
 
+    #[test]
+    fn a_corrupted_dmabuf_frame_counts_toward_the_shared_memory_retry() {
+        // The shim's report for niri on NVIDIA (issue #1503).
+        let corrupted = |data_type: &str| {
+            format!(
+                "the compositor's buffer failed validation (producer-marked-frame-corrupted),                  so this frame was dropped: dataType={data_type} stride=10240 width=2560                  height=1080 chunkOffset=0 chunkSize=0 available=11059200"
+            )
+        };
+        assert!(is_corrupted_dmabuf_frame("frame-dropped", &corrupted("DmaBuf")));
+        // Shared memory already is the fallback.
+        assert!(!is_corrupted_dmabuf_frame("frame-dropped", &corrupted("MemFd")));
+        assert!(!is_corrupted_dmabuf_frame(
+            "frame-dropped",
+            "the compositor's buffer failed validation (stride-shorter-than-row), so this              frame was dropped: dataType=DmaBuf",
+        ));
+        assert!(!is_corrupted_dmabuf_frame("dmabuf-import-failed", &corrupted("DmaBuf")));
+    }
+
     #[test]
     fn does_not_retry_unrelated_pipewire_errors_or_non_error_states() {
         assert!(!should_retry_shared_memory(