Detect the DMA-BUF crash bug by Hyprland version, add a capture-stall watchdog

The 1080p DMA-BUF pipeline crashed the user's entire Hyprland session
tonight. Root cause identified: hyprwm/Hyprland PR #15167 (fixed
2026-06-18, first shipped v0.56.0) -- CGLRenderbuffer's destructor
unconditionally dereferences m_framebuffer, which is left null when
createEGLImage() fails to import the DMA-BUF. This machine runs
v0.55.4, a week before the fix. The crash happens inside Hyprland's
own Screenshare::CScreenshareFrame::copyDmabuf(), confirmed against
the coredump; it is not a breadcast, GStreamer, or PipeWire bug, and
it is not resolution-dependent (upstream's own reports span unrelated
triggers -- touchpad gestures, tab switching, a Discord stream -- not
capture geometry), so 1080p vs 720p was never the actual variable.

choose_capture_backend() now reads the running compositor's version
over its own IPC socket and only uses the DMA-BUF path on Hyprland
>= 0.56.0 (or non-Hyprland sessions, unaffected by this bug). Below
that, or if the version can't be determined, it falls back to the
plain system-memory/wl_shm pipeline -- slower and still subject to
xdg-desktop-portal-hyprland's separate "Out of buffers" stall bug
(also confirmed via journalctl, and also not fixed in the installed
xdpw 1.3.12), but that failure mode is a stall, not a compositor-wide
abort.

That stall used to be silent forever: xdpw stops requesting frames
after 10 failed retries and never signals PipeWire, so GStreamer's own
bus reports nothing -- no error, no EOS -- and the frame pump just
blocks. pull_encoded_frame now bails after 10s of a Playing pipeline
producing nothing, converting an indefinite silent freeze into a real,
reported session failure (still correctly distinguishing a genuine
stall from ordinary EOS/teardown, so a normal stop() doesn't trip it).

VideoParams is no longer a hand-synced constant: build_video_pipeline_for_streaming
now returns the geometry/frame-rate it actually chose alongside the
pipeline, and both breadcastd::cast_mirror and cast_stream_test thread
that straight into the OFFER instead of a separately-maintained
default. Keeping two copies in sync by hand is exactly how the
resolution mismatch bug happened earlier tonight; returning the real
value makes that class of bug unrepresentable rather than just fixed
once.

Verified without touching the real compositor: cargo build --workspace
--examples, clippy, and both new unit tests (version parsing, the
0.55.4/0.56.0 backend-selection boundary) are clean. The watchdog's
firing path and the DMA-BUF path post-Hyprland-update are not yet
validated against real hardware -- deliberately, given what the last
live test cost. Recommended order: update Hyprland
(pacman -Syu hyprland xdg-desktop-portal-hyprland gets 0.56.1 + xdpw
1.4.1, which also picks up upstream fixes for the exact copy-fence and
SHM-handling bugs hit tonight) and confirm `hyprctl version` reports
>= 0.56.0 before testing DMA-BUF again. Without updating, this commit
still helps: the wl_shm path is selected automatically and the stall
is now bounded instead of indefinite.

.gitignore

@@ -31,3 +31,7 @@ logs/
 *.pid
 
 # Local hygiene notes (not for commit)
+CLAUDE.md
+
+# graphify knowledge-graph output (local tool cache, not for commit)
+graphify-out/

breadcast-caststream-sys/src/facade.cc

@@ -1,6 +1,7 @@
 #include "facade.h"
 
 #include <atomic>
+#include <chrono>
 #include <future>
 #include <memory>
 #include <string>
@@ -44,6 +45,34 @@ struct CastStreamSender {
   bool have_origin = false;
   int64_t origin_capture_time_us = 0;
 
+  // The capture timestamp of the last frame actually handed to
+  // Sender::EnqueueFrame. openscreen enforces strictly-increasing RTP
+  // timestamps with a *fatal* OSP_CHECK_GT (sender_impl.cc), not an error
+  // return -- so a single frame arriving with a non-increasing capture time
+  // (a buffer with no PTS, which the GStreamer side substitutes 0 for; a
+  // clock reset on portal source change; any encoder that ever reorders
+  // output) would abort the whole process. Dropping such a frame instead
+  // costs at most one frame of video. Only touched on the TaskRunner
+  // thread.
+  bool have_last_capture_time = false;
+  int64_t last_capture_time_us = 0;
+
+  // Set by breadcast_caststream_sender_destroy *before* it posts its
+  // teardown task, and checked by the self-rescheduling poll below.
+  //
+  // Without this, the poll task (posted with a 100ms delay, so it is the one
+  // task that can be scheduled to run *after* an already-queued teardown
+  // task) dereferences `environment` after teardown has reset it -- a null
+  // `unique_ptr<Environment>`, whose `task_runner()` accessor immediately
+  // dereferences a member -- i.e. a hard SIGSEGV on openscreen's TaskRunner
+  // thread. If it lands even later it is a use-after-free of `this`, since
+  // destroy() `delete`s this struct once the teardown task completes.
+  // TaskRunnerImpl's shutdown has an explicit "flushing phase" that keeps
+  // running runnable tasks, and PlatformClientPosix::ShutDown()'s quit task
+  // is queued *behind* whatever is already pending, so this is a race the
+  // teardown path can and does lose.
+  std::atomic<bool> shutting_down{false};
+
   // negotiated uses acquire/release so that once
   // breadcast_caststream_sender_enqueue_frame observes it true (from an
   // arbitrary caller thread), `session->video_sender()` is guaranteed
@@ -52,6 +81,41 @@ struct CastStreamSender {
   std::atomic<bool> needs_key_frame{true};
   std::atomic<int32_t> estimated_bandwidth_bps{kDefaultBandwidthEstimateBps};
 
+  // Set (never cleared except by the one consuming read, see
+  // breadcast_caststream_sender_needs_key_frame below) whenever a frame gets
+  // silently dropped after already being encoded -- either
+  // Sender::EnqueueFrame() rejecting it (e.g. MAX_DURATION_IN_FLIGHT, the
+  // in-flight budget openscreen enforces) or the non-monotonic-capture-time
+  // guard below. Either way, vah264enc already encoded the *next* frame as a
+  // P-slice depending on the one that just got dropped -- the encoder has no
+  // idea the drop happened, since it happens downstream of encoding, at this
+  // FFI boundary. Left alone, that reference is now dangling: the receiver
+  // decodes it against whatever picture it last successfully received,
+  // producing a stuck or corrupted frame that only resolves at the next
+  // regularly-scheduled key frame (key-int-max=60 -- up to ~2s, longer still
+  // if that key frame is itself dropped the same way). Forcing a key frame
+  // on the very next enqueue turns "up to several seconds of corruption"
+  // into "one dropped frame, then a clean resync" -- a separate flag rather
+  // than reusing `needs_key_frame` directly because SchedulePoll's 100ms
+  // timer unconditionally overwrites that one with the Sender's own
+  // (unrelated) NeedsKeyFrame() reading, which would silently clobber this
+  // signal before frame_pump_loop ever observed it.
+  std::atomic<bool> frame_chain_broken{false};
+
+  // See BreadcastEnqueueStats in facade.h. Counters are reset by the reading
+  // call; gauges are overwritten at every enqueue attempt. All relaxed --
+  // these are diagnostics, and a torn read across two of them costs nothing
+  // but a slightly inconsistent log line.
+  std::atomic<int32_t> enqueue_ok{0};
+  std::atomic<int32_t> enqueue_payload_too_large{0};
+  std::atomic<int32_t> enqueue_id_span_limit{0};
+  std::atomic<int32_t> enqueue_max_duration_in_flight{0};
+  std::atomic<int32_t> dropped_non_monotonic{0};
+  std::atomic<int32_t> in_flight_frames{0};
+  std::atomic<int32_t> in_flight_ms{0};
+  std::atomic<int32_t> max_in_flight_ms{0};
+  std::atomic<int32_t> round_trip_time_ms{0};
+
   // The caller's own user_data + callbacks, as passed to `_create`. Not
   // called directly -- session.h/message_port_bridge.h are instead given
   // trampolines below (with `this` as their user_data) so this struct can
@@ -62,8 +126,14 @@ struct CastStreamSender {
   BreadcastOnPictureLostFn rust_on_picture_lost = nullptr;
 
   void SchedulePoll() {
+    if (shutting_down.load(std::memory_order_acquire) || !environment) {
+      return;
+    }
     environment->task_runner().PostTaskWithDelay(
         [this] {
+          if (shutting_down.load(std::memory_order_acquire)) {
+            return;
+          }
           if (session && session->video_sender()) {
             needs_key_frame.store(session->video_sender()->NeedsKeyFrame(),
                                   std::memory_order_relaxed);
@@ -186,6 +256,10 @@ CastStreamSender* breadcast_caststream_sender_create(
 
 void breadcast_caststream_sender_negotiate(CastStreamSender* sender) {
   sender->environment->task_runner().PostTask([sender] {
+    if (sender->shutting_down.load(std::memory_order_acquire) ||
+        !sender->session) {
+      return;
+    }
     sender->session->Negotiate();
     sender->SchedulePoll();
   });
@@ -202,6 +276,10 @@ void breadcast_caststream_sender_on_message(CastStreamSender* sender,
   auto ns = std::make_shared<std::string>(message_namespace, message_namespace_len);
   auto body = std::make_shared<std::string>(message, message_len);
   sender->environment->task_runner().PostTask([sender, source, ns, body] {
+    if (sender->shutting_down.load(std::memory_order_acquire) ||
+        !sender->message_port) {
+      return;
+    }
     sender->message_port->DeliverMessage(*source, *ns, *body);
   });
 }
@@ -225,15 +303,32 @@ int32_t breadcast_caststream_sender_enqueue_frame(CastStreamSender* sender,
     using namespace openscreen;
     using namespace openscreen::cast;
 
+    if (sender->shutting_down.load(std::memory_order_acquire) ||
+        !sender->session) {
+      return;
+    }
     Sender* video_sender = sender->session->video_sender();
     if (!video_sender) {
       return;
     }
 
+    // See `last_capture_time_us`: openscreen aborts the process (fatal
+    // OSP_CHECK, not an error return) if RTP timestamps ever fail to
+    // strictly increase, so a non-monotonic capture time has to be dropped
+    // here rather than passed through.
+    if (sender->have_last_capture_time &&
+        capture_time_us <= sender->last_capture_time_us) {
+      sender->frame_chain_broken.store(true, std::memory_order_relaxed);
+      sender->dropped_non_monotonic.fetch_add(1, std::memory_order_relaxed);
+      return;
+    }
+
     if (!sender->have_origin) {
       sender->have_origin = true;
       sender->origin_capture_time_us = capture_time_us;
     }
+    sender->have_last_capture_time = true;
+    sender->last_capture_time_us = capture_time_us;
 
     const FrameId frame_id = video_sender->GetNextFrameId();
     const FrameId referenced_frame_id =
@@ -254,17 +349,74 @@ int32_t breadcast_caststream_sender_enqueue_frame(CastStreamSender* sender,
                        ByteView(owned_data->data(), owned_data->size()));
 
     // EnqueueFrame()'s result (e.g. MAX_DURATION_IN_FLIGHT under backpressure)
-    // isn't propagated to the caller: by the time this runs, enqueue_frame()
-    // has already returned 0 synchronously (this call is posted, not
-    // immediate -- see facade.h's threading contract). Backpressure here just
-    // means this one frame is dropped; the encoder finds out indirectly via
-    // needs_key_frame()/estimated_bandwidth_bps() polling.
-    (void)video_sender->EnqueueFrame(frame);
+    // can't be propagated to enqueue_frame()'s caller synchronously -- by the
+    // time this runs, that call has already returned 0 (this call is posted,
+    // not immediate -- see facade.h's threading contract). What it *can* do
+    // is flag the drop so the next frame comes in clean -- see
+    // `frame_chain_broken`'s doc comment on why that matters here, not just
+    // for the encoder's bitrate.
+    // Sampled *before* the enqueue attempt, so these describe the state the
+    // Sender used to make its accept/reject decision below rather than the
+    // state after it. See BreadcastEnqueueStats in facade.h.
+    const auto to_ms = [](Clock::duration d) {
+      return static_cast<int32_t>(
+          std::chrono::duration_cast<std::chrono::milliseconds>(d).count());
+    };
+    sender->in_flight_frames.store(
+        static_cast<int32_t>(video_sender->GetInFlightFrameCount()),
+        std::memory_order_relaxed);
+    sender->in_flight_ms.store(
+        to_ms(video_sender->GetInFlightMediaDuration(rtp_timestamp)),
+        std::memory_order_relaxed);
+    sender->max_in_flight_ms.store(
+        to_ms(video_sender->GetMaxInFlightMediaDuration()),
+        std::memory_order_relaxed);
+    sender->round_trip_time_ms.store(
+        to_ms(video_sender->GetCurrentRoundTripTime()),
+        std::memory_order_relaxed);
+
+    switch (video_sender->EnqueueFrame(frame)) {
+      case Sender::OK:
+        sender->enqueue_ok.fetch_add(1, std::memory_order_relaxed);
+        break;
+      case Sender::PAYLOAD_TOO_LARGE:
+        sender->enqueue_payload_too_large.fetch_add(1, std::memory_order_relaxed);
+        sender->frame_chain_broken.store(true, std::memory_order_relaxed);
+        break;
+      case Sender::REACHED_ID_SPAN_LIMIT:
+        sender->enqueue_id_span_limit.fetch_add(1, std::memory_order_relaxed);
+        sender->frame_chain_broken.store(true, std::memory_order_relaxed);
+        break;
+      case Sender::MAX_DURATION_IN_FLIGHT:
+        sender->enqueue_max_duration_in_flight.fetch_add(1, std::memory_order_relaxed);
+        sender->frame_chain_broken.store(true, std::memory_order_relaxed);
+        break;
+    }
   });
 
   return 0;
 }
 
+void breadcast_caststream_sender_take_stats(CastStreamSender* sender,
+                                            BreadcastEnqueueStats* out) {
+  if (!sender || !out) {
+    return;
+  }
+  out->enqueue_ok = sender->enqueue_ok.exchange(0, std::memory_order_relaxed);
+  out->enqueue_payload_too_large =
+      sender->enqueue_payload_too_large.exchange(0, std::memory_order_relaxed);
+  out->enqueue_id_span_limit =
+      sender->enqueue_id_span_limit.exchange(0, std::memory_order_relaxed);
+  out->enqueue_max_duration_in_flight =
+      sender->enqueue_max_duration_in_flight.exchange(0, std::memory_order_relaxed);
+  out->dropped_non_monotonic =
+      sender->dropped_non_monotonic.exchange(0, std::memory_order_relaxed);
+  out->in_flight_frames = sender->in_flight_frames.load(std::memory_order_relaxed);
+  out->in_flight_ms = sender->in_flight_ms.load(std::memory_order_relaxed);
+  out->max_in_flight_ms = sender->max_in_flight_ms.load(std::memory_order_relaxed);
+  out->round_trip_time_ms = sender->round_trip_time_ms.load(std::memory_order_relaxed);
+}
+
 int32_t breadcast_caststream_sender_needs_key_frame(CastStreamSender* sender) {
   return sender->needs_key_frame.load(std::memory_order_relaxed) ? 1 : 0;
 }
@@ -277,6 +429,12 @@ void breadcast_caststream_sender_destroy(CastStreamSender* sender) {
   if (!sender) {
     return;
   }
+  // Latched *before* the teardown task is posted so the self-rescheduling
+  // poll (see CastStreamSender::shutting_down) stops re-arming itself and
+  // no longer touches `environment`/`session` -- both of which the teardown
+  // task below is about to reset out from under it.
+  sender->shutting_down.store(true, std::memory_order_release);
+
   // These must be torn down on the TaskRunner thread (they hold raw
   // references into it and into `environment`), so hop over there and block
   // until it's done before shutting the TaskRunner itself down.

breadcast-caststream-sys/src/facade.h

@@ -103,6 +103,59 @@ int32_t breadcast_caststream_sender_enqueue_frame(CastStreamSender* sender,
                                                   int32_t is_key_frame,
                                                   int64_t capture_time_us);
 
+// A snapshot of why frames are (or aren't) making it into the Sender.
+//
+// The `enqueue_frame` entry point above cannot report this: it returns as
+// soon as the frame is *posted* to openscreen's TaskRunner, long before
+// Sender::EnqueueFrame actually runs and decides. So a caller watching only
+// its return value sees a 100% success rate even while every frame is being
+// rejected downstream -- which is exactly the blind spot that made a
+// multi-second picture freeze look, from the sender's own counters, like a
+// perfectly healthy 30fps stream.
+//
+// The four `enqueue_*` counters are cumulative-since-last-read: reading
+// them resets them to zero, so a caller polling once a second gets per-second
+// rates directly. The remaining fields are instantaneous gauges, sampled on
+// the TaskRunner thread at the moment of the most recent enqueue attempt.
+typedef struct BreadcastEnqueueStats {
+  // Sender::EnqueueFrame returned OK -- the frame is genuinely in flight.
+  int32_t enqueue_ok;
+  // Sender::PAYLOAD_TOO_LARGE -- the encoded access unit needs more RTP
+  // packets than the packetizer allows.
+  int32_t enqueue_payload_too_large;
+  // Sender::REACHED_ID_SPAN_LIMIT -- more than kMaxUnackedFrames (120)
+  // frames have gone unacknowledged.
+  int32_t enqueue_id_span_limit;
+  // Sender::MAX_DURATION_IN_FLIGHT -- the in-flight media window
+  // (see `in_flight_ms`/`max_in_flight_ms`) is full. The expected symptom
+  // of the receiver's acknowledgements stalling.
+  int32_t enqueue_max_duration_in_flight;
+  // Frames dropped by the facade before ever reaching EnqueueFrame, by the
+  // non-monotonic-capture-time guard in enqueue_frame.
+  int32_t dropped_non_monotonic;
+
+  // Sender::GetInFlightFrameCount() at the last enqueue attempt.
+  int32_t in_flight_frames;
+  // Sender::GetInFlightMediaDuration() at the last enqueue attempt, in ms --
+  // i.e. the media timespan between the oldest unacknowledged frame and the
+  // one being enqueued. Note this is a *timespan*, not a byte count: frame
+  // size has no bearing on it, so a large key frame is neither more nor less
+  // likely to be rejected than a small P-frame.
+  int32_t in_flight_ms;
+  // Sender::GetMaxInFlightMediaDuration() at the last enqueue attempt, in ms.
+  // A frame is rejected when `in_flight_ms` would exceed this. openscreen
+  // computes it as clamp(2*RTT, kMinSenderInFlight, playout_delay/3), so on
+  // a low-latency LAN it sits at the kMinSenderInFlight floor.
+  int32_t max_in_flight_ms;
+  // Sender::GetCurrentRoundTripTime() at the last enqueue attempt, in ms.
+  int32_t round_trip_time_ms;
+} BreadcastEnqueueStats;
+
+// Fills `out` with the current stats and resets the counters. Safe to call
+// from any thread; cheap, non-blocking, lock-free.
+void breadcast_caststream_sender_take_stats(CastStreamSender* sender,
+                                            BreadcastEnqueueStats* out);
+
 // True (nonzero) if the receiver wants a key frame as soon as possible.
 // Safe to poll frequently; cheap, non-blocking, lock-free.
 int32_t breadcast_caststream_sender_needs_key_frame(CastStreamSender* sender);

breadcast-caststream-sys/src/lib.rs

@@ -45,6 +45,29 @@ pub type OnErrorFn =
 
 pub type OnPictureLostFn = extern "C" fn(user_data: *mut c_void);
 
+/// Mirrors `BreadcastEnqueueStats` in `facade.h` -- see that struct's doc
+/// comment for what each field means and why they exist at all (short
+/// version: `sender_enqueue_frame`'s return value reports only that the
+/// frame was *posted* to openscreen's TaskRunner, never whether
+/// `Sender::EnqueueFrame` subsequently accepted it, so it reads 100% success
+/// even while every frame is being rejected).
+///
+/// The `enqueue_*`/`dropped_*` fields are counts since the previous
+/// `sender_take_stats` call; the rest are instantaneous gauges.
+#[repr(C)]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct EnqueueStats {
+    pub enqueue_ok: i32,
+    pub enqueue_payload_too_large: i32,
+    pub enqueue_id_span_limit: i32,
+    pub enqueue_max_duration_in_flight: i32,
+    pub dropped_non_monotonic: i32,
+    pub in_flight_frames: i32,
+    pub in_flight_ms: i32,
+    pub max_in_flight_ms: i32,
+    pub round_trip_time_ms: i32,
+}
+
 unsafe extern "C" {
     /// Returns null on failure (e.g. an unparseable `remote_ip`, or the
     /// local UDP socket failed to bind).
@@ -113,6 +136,16 @@ unsafe extern "C" {
         capture_time_us: i64,
     ) -> i32;
 
+    /// Fills `out` with the current enqueue stats and resets the counters.
+    ///
+    /// # Safety
+    /// `sender` must be live and `out` must be a valid, writable pointer to
+    /// an `EnqueueStats` for the duration of this call.
+    pub fn breadcast_caststream_sender_take_stats(
+        sender: *mut CastStreamSender,
+        out: *mut EnqueueStats,
+    );
+
     /// # Safety
     /// `sender` must be live.
     pub fn breadcast_caststream_sender_needs_key_frame(sender: *mut CastStreamSender) -> i32;

breadcast-caststream-sys/src/session.cc

@@ -39,6 +39,33 @@ using openscreen::cast::VideoStream;
 // audio-then-video streams collapses to just "index 0 is the video stream."
 constexpr int kVideoStreamIndex = 0;
 
+// The playout delay breadcast asks the receiver for -- the window between
+// capture here and presentation there. Deliberately *not*
+// openscreen::cast::kDefaultTargetPlayoutDelay (400ms), because that value
+// turned out to be the binding constraint on throughput, not just on latency.
+//
+// SenderImpl::GetMaxInFlightMediaDuration() computes the sender's send window
+// as clamp(2*RTT, kMinSenderInFlight, target_playout_delay/3). At a 400ms
+// target that ceiling is 133ms -- about four frames at 30 FPS. Instrumented
+// measurement against real hardware (see BreadcastEnqueueStats in facade.h)
+// found the round-trip time to a Chromecast over Wi-Fi sitting at 57-145ms,
+// i.e. 2*RTT of 114-290ms: consistently *above* that 133ms ceiling. The
+// window was therefore pinned at the ceiling and 3-40% of frames were being
+// rejected with MAX_DURATION_IN_FLIGHT every second, each one silently
+// breaking the H.264 reference chain and freezing the picture until the next
+// key frame.
+//
+// Raising this to 1200ms lifts the ceiling to 400ms, so 2*RTT lands inside
+// the clamp and the window tracks measured network conditions the way
+// openscreen intended, instead of being capped below one round trip. The cost
+// is ~800ms of additional end-to-end latency, which is unnoticeable for
+// screen mirroring to a TV and a straight trade against multi-second freezes.
+//
+// Note this is only the *sender's* half of the fix: it must stay paired with
+// the kMinSenderInFlight patch in vendor/openscreen (see PATCHES.md), which
+// raises the floor of that same clamp for the moments when RTT dips.
+constexpr std::chrono::milliseconds kTargetPlayoutDelay(1200);
+
 VideoStream BuildVideoStream(const VideoParams& params,
                              bool use_android_rtp_hack) {
   Stream stream;
@@ -47,7 +74,7 @@ VideoStream BuildVideoStream(const VideoParams& params,
   stream.channels = 1;
   stream.rtp_payload_type = GetPayloadType(VideoCodec::kH264, use_android_rtp_hack);
   stream.ssrc = GenerateSsrc(/*higher_priority=*/false);
-  stream.target_delay = openscreen::cast::kDefaultTargetPlayoutDelay;
+  stream.target_delay = kTargetPlayoutDelay;
   stream.aes_key = GenerateRandomBytes16();
   stream.aes_iv_mask = GenerateRandomBytes16();
   stream.receiver_rtcp_event_log = true;

breadcast-caststream-sys/vendor/openscreen/PATCHES.md

@@ -62,3 +62,25 @@ There are no release/API-stability guarantees upstream. To update:
    (pulled in separately via gclient/DEPS in a full Chromium checkout).
    Reimplemented on `EVP_EncodeBlock`/`EVP_DecodeBlock` from system OpenSSL
    instead, same public interface.
+6. **`cast/streaming/impl/sender_impl.cc`** — raised `kMinSenderInFlight`
+   from upstream's 66ms to 200ms. This is a behaviour patch, not a
+   portability one, and is the only one here that changes what goes on the
+   wire — so unlike the others it should *not* be silently re-applied when
+   rolling the pin without re-measuring first.
+
+   `GetMaxInFlightMediaDuration()` sizes the sender's send window as
+   `clamp(2*RTT, kMinSenderInFlight, target_playout_delay/3)`. Upstream's
+   66ms floor assumes the RTT is negligible, which holds for Chrome's own
+   usage but not for breadcast's measured case: instrumenting real
+   `Sender::EnqueueFrame` result codes (see `BreadcastEnqueueStats` in
+   `../../src/facade.h`) against a Chromecast over Wi-Fi showed RTT of
+   57-145ms and a steady 3-40% of frames per second rejected with
+   `MAX_DURATION_IN_FLIGHT`. Because breadcast enqueues already-encoded
+   frames, each such rejection silently breaks the H.264 reference chain
+   rather than merely dropping a frame, which is what the user-visible
+   multi-second picture freezes turned out to be.
+
+   Pairs with `kTargetPlayoutDelay` in `../../src/session.cc`, which raises
+   the *ceiling* of that same clamp (the ceiling, not this floor, was the
+   binding constraint at 400ms playout delay). Both are needed: this floor
+   covers RTT dips, that ceiling covers the normal case.

breadcast-caststream-sys/vendor/openscreen/cast/streaming/impl/sender_impl.cc

@@ -24,10 +24,18 @@ namespace {
 
 // The minimum amount of media the Sender keeps in-flight, regardless of the
 // measured network round-trip time. This keeps the encoder pipeline flowing on
-// low-latency networks (roughly two video frames at 30 FPS). See
-// crbug.com/498035450.
+// low-latency networks. See crbug.com/498035450.
+//
+// LOCAL PATCH (breadcast): upstream is 66ms, roughly two video frames at
+// 30 FPS. That is only enough when the round-trip time is genuinely
+// negligible. Instrumented measurement against real hardware (a Chromecast
+// over Wi-Fi) showed RTT swinging between 57ms and 145ms, so a 66ms floor
+// leaves the send window narrower than a single round trip -- frames are
+// rejected with MAX_DURATION_IN_FLIGHT faster than acknowledgements can
+// free the window back up. A 200ms floor is ~6 frames at 30 FPS, enough to
+// cover one round trip at the worst observed RTT. See PATCHES.md.
 constexpr Clock::duration kMinSenderInFlight =
-    Clock::to_duration(milliseconds(66));
+    Clock::to_duration(milliseconds(200));
 
 }  // namespace
 

breadcast-core/src/cast_sender.rs

@@ -326,10 +326,20 @@ fn run_io_loop(
                 // doc comment for why that's fine for this project's usage.
             }
             Ok(_) => {}
-            Err(e) => {
-                tracing::debug!(error = %e, "cast session io loop ending: receive error");
+            // Only end the loop on an error that means the connection
+            // itself is gone -- a single malformed/unexpected message (e.g.
+            // a MEDIA_STATUS missing a field this crate's struct treats as
+            // required) is a `Serialization`/`Parsing` error, not a dead
+            // socket, and used to take the whole receiver-status/control
+            // channel down with it for the rest of the session even though
+            // the RTP stream itself was unaffected.
+            Err(e @ (rust_cast::errors::Error::Io(_) | rust_cast::errors::Error::Tls(_) | rust_cast::errors::Error::Dns(_))) => {
+                tracing::debug!(error = %e, "cast session io loop ending: connection error");
                 return;
             }
+            Err(e) => {
+                tracing::warn!(error = %e, "cast session: ignoring unparseable/unexpected message");
+            }
         }
     }
 }

breadcast-core/src/caststream.rs

@@ -24,7 +24,7 @@ use breadcast_caststream_sys::{
     self as sys, breadcast_caststream_sender_create, breadcast_caststream_sender_destroy,
     breadcast_caststream_sender_enqueue_frame, breadcast_caststream_sender_estimated_bandwidth_bps,
     breadcast_caststream_sender_needs_key_frame, breadcast_caststream_sender_negotiate,
-    breadcast_caststream_sender_on_message,
+    breadcast_caststream_sender_on_message, breadcast_caststream_sender_take_stats,
 };
 
 /// The Cast Streaming ("Mirroring") receiver app id, pre-installed on every
@@ -48,10 +48,44 @@ pub struct VideoParams {
 impl Default for VideoParams {
     fn default() -> Self {
         Self {
+            // NOT the value that goes on the wire. `build_video_pipeline_for_streaming`
+            // returns the geometry it will really encode, and both
+            // `breadcastd::cast_mirror` and `cast_stream_test` pass *that*
+            // to `CastStreamSender::start` -- this default only supplies the
+            // fields that don't vary with the capture path (bitrate, frame
+            // rate denominator).
+            //
+            // It has to work that way because the pipeline picks its capture
+            // backend at runtime, and the two backends differ in resolution
+            // and frame rate. This OFFER's resolution is what the receiver
+            // allocates its decoder/output surface for, so advertising
+            // anything other than what's actually sent is a real protocol
+            // mismatch -- one that plausibly explains a receiver decoder
+            // corrupting/freezing rather than just looking soft. Keeping the
+            // two in sync by hand across two files is exactly how that got
+            // out of step before; returning it from the pipeline builder
+            // makes the mismatch unrepresentable. The values below are the
+            // DMA-BUF path's, kept only as a sane standalone default.
             width: 1920,
             height: 1080,
-            max_bitrate_bps: 8_000_000,
-            max_frame_rate_numerator: 30,
+            // Kept equal to `breadcastd::cast_mirror::MAX_BITRATE_KBPS *
+            // 1000` -- see that constant's doc comment for why 8 Mbps
+            // (this struct's previous value) isn't used here: real hardware
+            // testing showed the AIMD probe pinning to whatever this
+            // ceiling is for the entire session once the estimator reports
+            // (unreliably) that there's room, and 8 Mbps sustained was more
+            // than the previous network+receiver could actually hold,
+            // producing repeated multi-second freezes rather than just
+            // softer video.
+            max_bitrate_bps: 6_000_000,
+            // A *ceiling*, not a promise -- which is what makes it safe for
+            // the DMA-BUF path, where nothing caps the rate (`vapostproc` is
+            // a per-frame transform and can't do temporal conversion, see
+            // `build_video_pipeline_for_streaming`) and frames arrive at
+            // whatever the compositor delivers, up to this machine's 60Hz
+            // refresh. The wl_shm path does have a `videorate` capping it
+            // hard, and overrides this with the rate it actually enforces.
+            max_frame_rate_numerator: 60,
             max_frame_rate_denominator: 1,
         }
     }
@@ -217,6 +251,22 @@ impl CastStreamSender {
         Ok(())
     }
 
+    /// A snapshot of how the *underlying* `Sender::EnqueueFrame` has been
+    /// answering, plus its in-flight window gauges. Reading resets the
+    /// counters, so polling once a second yields per-second rates.
+    ///
+    /// [`Self::enqueue_frame`] deliberately cannot report any of this: it
+    /// returns as soon as the frame is posted to openscreen's TaskRunner,
+    /// before the real accept/reject happens. Any "frames enqueued per
+    /// second" figure derived from its return value is therefore a count of
+    /// *attempts*, and stays pinned at the capture rate even while every
+    /// frame is being rejected downstream and the picture is frozen.
+    pub fn enqueue_stats(&self) -> sys::EnqueueStats {
+        let mut stats = sys::EnqueueStats::default();
+        unsafe { breadcast_caststream_sender_take_stats(self.raw, &mut stats) };
+        stats
+    }
+
     /// True if the receiver wants a key frame as soon as possible. Cheap to
     /// poll frequently (e.g. once per captured frame, before encoding it).
     pub fn needs_key_frame(&self) -> bool {

breadcast-core/src/examples/cast_stream_test.rs

@@ -10,7 +10,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
-use breadcast_core::caststream::{CastStreamEvent, VideoParams, WEBRTC_NAMESPACE};
+use breadcast_core::caststream::{CastStreamEvent, WEBRTC_NAMESPACE};
 use breadcast_core::net::local_lan_ip;
 use breadcast_core::pipeline::{
     build_video_pipeline_for_streaming, pull_encoded_frame, request_key_frame, set_video_bitrate_kbps,
@@ -58,7 +58,7 @@ async fn main() -> anyhow::Result<()> {
     let capture = breadcast_core::CaptureSession::start().await?;
     println!("Got PipeWire video node id: {}", capture.video_node_id());
 
-    let (pipeline, appsink, encoder) = build_video_pipeline_for_streaming(capture.video_node_id())?;
+    let (pipeline, appsink, encoder, video_params) = build_video_pipeline_for_streaming(capture.video_node_id())?;
 
     // Watch the encode pipeline's own bus in the background -- see
     // mirror_test.rs's identical block for why this matters.
@@ -81,7 +81,7 @@ async fn main() -> anyhow::Result<()> {
         &device.host,
         "sender-0",
         session.transport_id(),
-        VideoParams::default(),
+        video_params,
     )?;
     let sender = Arc::new(sender);
 

breadcast-core/src/pipeline/mod.rs

@@ -1,4 +1,6 @@
+use std::io::{Read as _, Write as _};
 use std::os::unix::fs::DirBuilderExt;
+use std::os::unix::net::UnixStream;
 use std::path::{Path, PathBuf};
 use std::time::Duration;
 
@@ -8,6 +10,8 @@ use gstreamer::prelude::*;
 use gstreamer_app as gst_app;
 use gstreamer_video as gst_video;
 
+use crate::caststream::VideoParams;
+
 /// Builds (but doesn't start) the capture → encode → mux → HLS pipeline for
 /// a single video source. `output_dir` is created if it doesn't exist;
 /// `hlssink3` writes `segment%05d.ts` files and `playlist.m3u8` there.
@@ -62,15 +66,37 @@ pub fn build_video_pipeline(video_node_id: u32, output_dir: &Path) -> Result<gst
     // (stretches to 16:9) — correctness/compatibility first, an
     // aspect-preserving scale (letterbox via `videoscale
     // add-borders=true`) is a follow-up, not a blocker.
+    // HLS segment sizing is *the* dominant term in this path's end-to-end
+    // latency, and the two knobs are coupled: a segment can only be cut on a
+    // key frame, so the real segment duration is `max(target-duration,
+    // GOP length)` no matter what `target-duration` says. With the previous
+    // `target-duration=2` + `key-int-max=60` (60 frames / 30fps = a 2s GOP),
+    // segments were 2s, and a renderer that buffers the customary three of
+    // them before starting playback sits ~6s behind live -- on top of
+    // however much of the playlist it decides to start from. `dlna_mirror.rs`
+    // then waited for 3 segments to exist before even handing over the URL,
+    // adding another ~6s of already-stale content.
+    //
+    // 1s segments (GOP dropped to 30 frames to make that actually
+    // achievable) roughly halve that. Going below 1s is not worth it here:
+    // MPEG-TS + a per-segment key frame means shorter segments cost real
+    // bitrate, and classic (non-LL) HLS clients don't reliably honour
+    // sub-second target durations anyway. Genuinely low latency on this path
+    // needs LL-HLS, which `hlssink3` does not implement -- the Cast
+    // Streaming path (`build_video_pipeline_for_streaming`) is the
+    // low-latency answer, and this one is the compatibility answer.
+    //
+    // `playlist-length`/`max-files` shrink to match so the playlist doesn't
+    // advertise a long backlog of stale segments for a client to start from.
     let pipeline_str = format!(
         "pipewiresrc path={video_node_id} do-timestamp=true ! \
          videoconvert ! videoscale ! videorate ! \
          video/x-raw,format=NV12,width=1280,height=720,framerate=30/1 ! \
-         vah264enc bitrate=4000 key-int-max=60 rate-control=cbr ! \
+         vah264enc bitrate=4000 key-int-max=30 rate-control=cbr ! \
          video/x-h264,profile=main ! \
          h264parse config-interval=1 ! \
          hlssink.video \
-         hlssink3 name=hlssink target-duration=2 playlist-length=6 max-files=10"
+         hlssink3 name=hlssink target-duration=1 playlist-length=3 max-files=6"
     );
 
     let element = gst::parse::launch(&pipeline_str).context("failed to parse GStreamer pipeline")?;
@@ -123,23 +149,243 @@ pub fn build_video_pipeline(video_node_id: u32, output_dir: &Path) -> Result<gst
 /// `CastStreamSender::estimated_bandwidth_bps()` — see
 /// [`request_key_frame`]/`set_video_bitrate_kbps` for the two knobs a
 /// congestion-control loop needs).
+/// Which capture-side memory path [`build_video_pipeline_for_streaming`] asks
+/// PipeWire (and through it, the compositor's portal implementation) to hand
+/// this pipeline. The two are not interchangeable: they have *different*
+/// known failure modes on different compositor versions, which is why the
+/// choice is made at runtime rather than baked into one pipeline string.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum CaptureBackend {
+    /// Zero-copy: `pipewiresrc` hands DMA-BUFs straight to `vapostproc`,
+    /// which hands VA-memory straight to `vah264enc`. Nothing ever touches
+    /// system memory, and the portal never falls back to `wl_shm`.
+    Dmabuf,
+    /// `videoconvert ! videoscale ! videorate` on plain system memory, which
+    /// makes PipeWire request `wl_shm` buffers from the portal. CPU-costly
+    /// and stall-prone (see [`choose_capture_backend`]), but it does not go
+    /// anywhere near a compositor's DMA-BUF render-into-client-buffer path.
+    Shm,
+}
+
+/// First Hyprland release containing `renderer/rbo: avoid nullptr deref`
+/// (upstream commit `ae1690c2`, PR #15167, 2026-06-18), tagged in v0.56.0 on
+/// 2026-07-20.
+///
+/// Below this, asking Hyprland for a DMA-BUF screencast can **kill the
+/// user's entire compositor session**, and there is nothing this pipeline
+/// can do about it from the client side. The mechanism, confirmed against a
+/// real coredump on v0.55.4 plus upstream's own source:
+/// `Screenshare::CScreenshareFrame::copyDmabuf()` renders the monitor into
+/// the client-supplied DMA-BUF via `IHyprRenderer::beginRender` ->
+/// `getOrCreateRenderbuffer`. If `CGLRenderbuffer`'s constructor fails to
+/// import the buffer (`createEGLImage` returns `EGL_NO_IMAGE_KHR`) it
+/// early-returns leaving `m_framebuffer` null -- and v0.55.4's destructor
+/// then unconditionally does `unbind(); m_framebuffer->release();` on that
+/// null pointer while the failed renderbuffer is being torn down. The
+/// resulting abort takes down Hyprland, every window, and (separately,
+/// same instant) `xdg-desktop-portal-hyprland`. Upstream's fix is a one-line
+/// `if (m_framebuffer)` guard; it converts the abort into a dropped frame.
+///
+/// Reported upstream at least three times (hyprwm/Hyprland #13487, #13543,
+/// #13653, all v0.54.x, all on AMD) and auto-closed unread by the
+/// issues-are-disabled bot rather than triaged, so the "is it fixed?"
+/// question can only be answered from the commit log, not the tracker. The
+/// reported triggers (window-group tab switching, touchpad gestures, an
+/// emulator in a Discord stream) have nothing in common with each other or
+/// with resolution -- treat the import failure as intermittent, not as
+/// something a particular capture geometry or DRM modifier provokes.
+///
+/// Note in particular that this is *not* avoidable by requesting a
+/// "simpler" buffer layout. `vapostproc` advertises exactly one AMD DRM
+/// modifier on its `video/x-raw(memory:DMABuf)` pads -- `0x0200000008401b04`
+/// = GFX11, 64K_R_X tiling, `DCC=0` (verified with `gst-inspect-1.0
+/// vapostproc` and `drm_fourcc.h`'s field shifts). It offers no LINEAR
+/// alternative, and the one modifier it does offer is already uncompressed,
+/// so there is no tiling/compression hazard left to negotiate away.
+const HYPRLAND_MIN_SAFE_DMABUF: (u32, u32, u32) = (0, 56, 0);
+
+/// Reads the running Hyprland's version over its own IPC socket (the same
+/// `j/version` request `hyprctl version -j` makes) without spawning
+/// `hyprctl`, which needn't be installed. `None` if this isn't a Hyprland
+/// session at all, or if the version can't be determined.
+fn hyprland_version() -> Option<(u32, u32, u32)> {
+    let signature = std::env::var("HYPRLAND_INSTANCE_SIGNATURE").ok()?;
+    let runtime_dir = std::env::var("XDG_RUNTIME_DIR").ok()?;
+
+    let mut socket = UnixStream::connect(format!("{runtime_dir}/hypr/{signature}/.socket.sock")).ok()?;
+    // Bounded on both halves: this runs on the way into starting a mirror
+    // session, and a wedged compositor must not be able to hang that.
+    socket.set_write_timeout(Some(Duration::from_secs(1))).ok()?;
+    socket.set_read_timeout(Some(Duration::from_secs(1))).ok()?;
+    socket.write_all(b"j/version").ok()?;
+
+    let mut response = String::new();
+    socket.read_to_string(&mut response).ok()?;
+    let parsed: serde_json::Value = serde_json::from_str(&response).ok()?;
+
+    // `version` is the plain "0.55.4"; `tag` is "v0.55.4" and is what older
+    // Hyprlands report, so accept either.
+    let raw = parsed.get("version").or_else(|| parsed.get("tag"))?.as_str()?;
+    parse_hyprland_version(raw)
+}
+
+/// Splits a Hyprland version string into comparable components. Strips a
+/// leading `v` (`tag` carries one, `version` doesn't) and anything from the
+/// first `-` (a git build's tag looks like `v0.55.4-123-gdeadbee`).
+fn parse_hyprland_version(raw: &str) -> Option<(u32, u32, u32)> {
+    let mut parts = raw.trim().trim_start_matches('v').split('-').next()?.split('.');
+    let major = parts.next()?.parse().ok()?;
+    let minor = parts.next()?.parse().ok()?;
+    // A two-component "0.56" is treated as 0.56.0 rather than rejected --
+    // erring toward *parsing* here is safe, since the comparison against
+    // `HYPRLAND_MIN_SAFE_DMABUF` is what decides anything.
+    let patch = parts.next().unwrap_or("0").parse().ok()?;
+    Some((major, minor, patch))
+}
+
+/// Picks the capture path, trading two *different* real bugs off against
+/// each other rather than pretending either one is hypothetical.
+///
+/// [`CaptureBackend::Dmabuf`] is the better path and the default: it is
+/// genuinely zero-copy, and it sidesteps `xdg-desktop-portal-hyprland`'s
+/// `wl_shm` stall entirely. That stall is not a hiccup -- it is terminal.
+/// In xdpw's `src/portals/Screencopy.cpp`, when the PipeWire consumer is
+/// holding every buffer, the portal logs "Out of buffers" and re-queues a
+/// frame only while `copyRetries++ < MAX_RETRIES` (10); `copyRetries` is
+/// reset to 0 *only* on a successful copy. So ten consecutive misses and
+/// the portal stops requesting frames forever, without sending an error to
+/// PipeWire -- which is exactly why a 45-second freeze showed up in
+/// `journalctl` and nowhere on this pipeline's own GStreamer bus. (Worth
+/// keeping in mind that "the consumer is holding every buffer" means the
+/// stall can *originate* downstream: a brief encoder or RTP-send stall stops
+/// buffers being recycled, and the portal's give-up logic then makes it
+/// permanent. [`pull_encoded_frame`]'s watchdog is the backstop for both.)
+///
+/// But on Hyprland older than [`HYPRLAND_MIN_SAFE_DMABUF`] the DMA-BUF path
+/// can abort the compositor outright, which is a categorically worse outcome
+/// than a stalled cast -- so there, fall back to `wl_shm` and let the
+/// watchdog bound the damage. Non-Hyprland sessions are unaffected by that
+/// bug and keep DMA-BUF.
+fn choose_capture_backend() -> CaptureBackend {
+    let Some(version) = hyprland_version() else {
+        // Either not Hyprland (so the Hyprland-specific crash can't apply),
+        // or Hyprland with an unreadable version. The latter is the
+        // ambiguous case; prefer the path that cannot take the desktop down.
+        if std::env::var_os("HYPRLAND_INSTANCE_SIGNATURE").is_some() {
+            tracing::warn!(
+                "running under Hyprland but could not read its version; using the slower wl_shm \
+                 capture path, since DMA-BUF screencast aborts the compositor before v{}.{}.{}",
+                HYPRLAND_MIN_SAFE_DMABUF.0,
+                HYPRLAND_MIN_SAFE_DMABUF.1,
+                HYPRLAND_MIN_SAFE_DMABUF.2,
+            );
+            return CaptureBackend::Shm;
+        }
+        return CaptureBackend::Dmabuf;
+    };
+
+    if version < HYPRLAND_MIN_SAFE_DMABUF {
+        tracing::warn!(
+            hyprland = format!("{}.{}.{}", version.0, version.1, version.2),
+            "this Hyprland predates the fix for the DMA-BUF screencast compositor crash \
+             (upstream PR #15167, released in v0.56.0) -- falling back to the slower, \
+             stall-prone wl_shm capture path. Updating Hyprland restores zero-copy capture."
+        );
+        return CaptureBackend::Shm;
+    }
+
+    CaptureBackend::Dmabuf
+}
+
 pub fn build_video_pipeline_for_streaming(
     video_node_id: u32,
-) -> Result<(gst::Pipeline, gst_app::AppSink, gst::Element)> {
+) -> Result<(gst::Pipeline, gst_app::AppSink, gst::Element, VideoParams)> {
     gst::init().context("failed to initialize GStreamer")?;
 
-    // Same 1280x720@30 Main-profile baseline as build_video_pipeline, for
-    // the same reason (see its doc comment) -- broad decoder compatibility
-    // first, revisit upward once a specific device's real ceiling is known.
-    let pipeline_str = "pipewiresrc path=%VIDEO_NODE_ID% do-timestamp=true ! \
-         videoconvert ! videoscale ! videorate ! \
-         video/x-raw,format=NV12,width=1280,height=720,framerate=30/1 ! \
-         vah264enc name=venc bitrate=4000 key-int-max=60 rate-control=cbr ! \
-         video/x-h264,profile=main ! \
-         h264parse name=h264parse config-interval=-1 ! \
-         video/x-h264,stream-format=byte-stream,alignment=au ! \
-         appsink name=appsink emit-signals=false sync=false max-buffers=4 drop=true"
-        .replace("%VIDEO_NODE_ID%", &video_node_id.to_string());
+    // 1920x1080@native-rate Main profile. Went 1080p -> 720p -> 1080p again
+    // tonight: the first 1080p attempt froze near-instantly, but that had
+    // nothing to do with resolution -- it was the `xdg-desktop-portal-hyprland`
+    // wl_shm buffer-exhaustion bug below (real, structural, whatever the
+    // resolution) compounded by openscreen's in-flight RTP budget being too
+    // tight for this receiver's actual RTT (see `frame_chain_broken` in
+    // `breadcast-caststream-sys/src/facade.cc`, and the playout-delay tuning
+    // in `breadcast-caststream-sys/src/session.cc`). With both of those
+    // fixed -- confirmed via a real, freeze-free 720p session -- the
+    // packet-count increase 1080p brings back is no longer landing on an
+    // already-struggling budget, so it's worth trying again on its own
+    // merits.
+    //
+    // On the DMA-BUF path `pipewiresrc` deliberately does *not* go through
+    // `videoconvert ! videoscale ! videorate ! video/x-raw,...` (plain
+    // system-memory caps) -- doing so forces PipeWire to hand the
+    // compositor's portal implementation a `wl_shm` (shared-memory) buffer
+    // request, and on this system (`xdg-desktop-portal-hyprland`) that path
+    // is real-world buggy: journalctl during a live freeze showed it
+    // repeatedly logging "Asked for a wl_shm buffer which is legacy" / "Out
+    // of buffers" / "Retrying screencopy" in a tight loop that never
+    // actually delivered a frame -- multi-second (once 45+ second) stalls
+    // with *zero* signal on breadcast's own GStreamer bus, since nothing
+    // here was erroring, it was just starved waiting on a buffer the
+    // portal's legacy path never produced. See `choose_capture_backend` for
+    // why that stall is permanent rather than transient, and for the one
+    // case where it's still the lesser evil.
+    //
+    // `vapostproc` (VA-API postprocessor -- confirmed present via
+    // `gst-inspect-1.0 vapostproc`, ships in `gst-plugins-bad`'s `va`
+    // plugin) accepts `video/x-raw(memory:DMABuf)` directly from
+    // `pipewiresrc` and outputs `video/x-raw(memory:VAMemory)`, which
+    // `vah264enc` also accepts natively -- a fully zero-copy DMA-BUF path
+    // from portal to hardware encoder that never touches the legacy wl_shm
+    // fallback at all. No `videorate` in this path: `vapostproc` is a
+    // per-frame transform (scale/convert), not a temporal one, so it can't
+    // do frame-rate reduction the way `videorate` does on raw memory --
+    // frames flow at whatever rate PipeWire actually delivers rather than a
+    // forced 30fps. This is fine for RTP: `facade.cc` derives RTP timestamps
+    // from each frame's real capture time regardless of the nominal rate,
+    // and openscreen's own frame pacing doesn't assume a fixed source rate
+    // either. The returned `VideoParams` advertises 60 as a *ceiling*
+    // (`max_frame_rate_*`), which stays truthful whether the compositor
+    // actually delivers 60 or fewer; the wl_shm path's `videorate` does cap
+    // hard, so it advertises the rate it really enforces.
+    //
+    // Both branches return the geometry they actually encode, and the caller
+    // hands that straight to the Cast OFFER. That coupling is deliberate:
+    // advertising a resolution other than what's really sent is a genuine
+    // protocol mismatch this project has already been bitten by once, and
+    // keeping two constants manually in sync across two files is how that
+    // happened. Returning it makes the mismatch unrepresentable.
+    let (pipeline_str, params) = match choose_capture_backend() {
+        CaptureBackend::Dmabuf => (
+            "pipewiresrc path=%VIDEO_NODE_ID% do-timestamp=true ! \
+             video/x-raw(memory:DMABuf),format=DMA_DRM ! \
+             vapostproc ! \
+             video/x-raw(memory:VAMemory),format=NV12,width=1920,height=1080 ! \
+             vah264enc name=venc bitrate=4000 key-int-max=60 rate-control=cbr ! \
+             video/x-h264,profile=main ! \
+             h264parse name=h264parse config-interval=-1 ! \
+             video/x-h264,stream-format=byte-stream,alignment=au ! \
+             appsink name=appsink emit-signals=false sync=false max-buffers=4 drop=true",
+            VideoParams { width: 1920, height: 1080, max_frame_rate_numerator: 60, ..VideoParams::default() },
+        ),
+        // 720p30 rather than 1080p60 on this path on purpose: every frame is
+        // a CPU convert + scale here, and CPU cost is precisely what makes
+        // the portal's "Out of buffers" give-up more likely, since the
+        // portal runs out exactly when the consumer is slow to recycle
+        // buffers. The lighter shape is also what was last known to work on
+        // real hardware before the DMA-BUF switch.
+        CaptureBackend::Shm => (
+            "pipewiresrc path=%VIDEO_NODE_ID% do-timestamp=true ! \
+             videoconvert ! videoscale ! videorate ! \
+             video/x-raw,format=NV12,width=1280,height=720,framerate=30/1 ! \
+             vah264enc name=venc bitrate=4000 key-int-max=60 rate-control=cbr ! \
+             video/x-h264,profile=main ! \
+             h264parse name=h264parse config-interval=-1 ! \
+             video/x-h264,stream-format=byte-stream,alignment=au ! \
+             appsink name=appsink emit-signals=false sync=false max-buffers=4 drop=true",
+            VideoParams { width: 1280, height: 720, max_frame_rate_numerator: 30, ..VideoParams::default() },
+        ),
+    };
+    let pipeline_str = pipeline_str.replace("%VIDEO_NODE_ID%", &video_node_id.to_string());
 
     let element = gst::parse::launch(&pipeline_str).context("failed to parse GStreamer pipeline")?;
     let Ok(pipeline) = element.downcast::<gst::Pipeline>() else {
@@ -154,28 +400,95 @@ pub fn build_video_pipeline_for_streaming(
 
     let encoder = pipeline.by_name("venc").context("parsed pipeline has no element named 'venc'")?;
 
-    Ok((pipeline, appsink, encoder))
+    Ok((pipeline, appsink, encoder, params))
 }
 
+/// How long [`pull_encoded_frame`] waits per `try_pull_sample` call. Short
+/// enough that a teardown from another thread is noticed promptly, long
+/// enough not to spin.
+const CAPTURE_STALL_POLL: Duration = Duration::from_millis(250);
+
+/// How long [`pull_encoded_frame`] tolerates a *playing* pipeline producing
+/// no frames at all before declaring the capture dead.
+///
+/// This exists because the failure it catches is otherwise completely
+/// silent. `xdg-desktop-portal-hyprland` stops requesting frames after ten
+/// consecutive "Out of buffers" misses and never sends an error to PipeWire
+/// (see [`choose_capture_backend`]); a compositor that fails to import a
+/// capture buffer likewise just drops the frame. In both cases GStreamer has
+/// nothing to report -- no bus error, no EOS, no flow-return failure -- so
+/// without a timeout here the frame pump blocks in `pull_sample` forever and
+/// the mirror session appears frozen with nothing anywhere saying why. That
+/// is precisely the 45-second freeze that took a `journalctl` dig to
+/// explain.
+///
+/// Bailing propagates out of `breadcastd`'s frame-pump thread, which already
+/// reports `DaemonCommand::SessionEnded` on exit, so the session tears down
+/// and the failure surfaces as a real event instead of a hang. Generous
+/// enough (10s) that a merely slow moment -- a heavy compositor frame, a
+/// bitrate renegotiation -- doesn't trip it; anything longer than this is
+/// not a hiccup, since neither of the known failure modes recovers.
+const CAPTURE_STALL_TIMEOUT: Duration = Duration::from_secs(10);
+
 /// Pulls one complete Annex-B H.264 access unit from `appsink`, blocking
 /// until one is available. Returns `None` once the pipeline reaches EOS or
-/// the sink otherwise stops (e.g. pipeline torn down from another thread).
+/// the sink otherwise stops (e.g. pipeline torn down from another thread),
+/// and errors if the pipeline is still playing but has gone
+/// [`CAPTURE_STALL_TIMEOUT`] without producing a frame.
+///
+/// A buffer with no PTS is skipped (this pulls the next one instead) rather
+/// than reported with a substituted timestamp, as an earlier version did
+/// with `.unwrap_or(0)`. That substitution was actively dangerous rather
+/// than merely imprecise: openscreen derives the frame's RTP timestamp from
+/// this value and enforces strict monotonicity with a *fatal* `OSP_CHECK`
+/// (`sender_impl.cc`'s `OSP_CHECK_GT(frame.rtp_timestamp, ...)`), not an
+/// error return -- so a single PTS-less buffer part-way into a session would
+/// abort the whole daemon. `facade.cc` independently drops non-monotonic
+/// capture times as a second line of defence; neither guard makes the other
+/// redundant, since `enqueue_frame` is a public FFI entry point that has to
+/// hold up against any caller.
 pub fn pull_encoded_frame(appsink: &gst_app::AppSink) -> Result<Option<(Vec<u8>, bool, i64)>> {
-    let sample = match appsink.pull_sample() {
-        Ok(sample) => sample,
-        Err(_) if appsink.is_eos() => return Ok(None),
-        Err(e) => bail!("appsink pull_sample failed: {e}"),
-    };
-    let buffer = sample.buffer().context("pulled sample had no buffer")?;
-    let map = buffer.map_readable().context("failed to map sample buffer readable")?;
-    let is_key_frame = !buffer.flags().contains(gst::BufferFlags::DELTA_UNIT);
-    // `.unwrap_or(0)` rather than propagating a missing PTS as an error:
-    // CastStreamSender::enqueue_frame only needs monotonically-increasing,
-    // real-elapsed-time-proportional values (see its doc comment) -- an
-    // occasional buffer with no PTS shouldn't abort an otherwise-live
-    // stream over it.
-    let capture_time_us = buffer.pts().map(|t| t.useconds() as i64).unwrap_or(0);
-    Ok(Some((map.as_slice().to_vec(), is_key_frame, capture_time_us)))
+    let poll = gst::ClockTime::from_mseconds(CAPTURE_STALL_POLL.as_millis() as u64);
+    let mut stalled_for = Duration::ZERO;
+    loop {
+        let Some(sample) = appsink.try_pull_sample(Some(poll)) else {
+            // EOS is the ordinary end: the user hit "Stop sharing" in the
+            // portal, or the source went away.
+            if appsink.is_eos() {
+                return Ok(None);
+            }
+            // Teardown from another thread (`CastMirrorSession::stop` sets
+            // the pipeline to Null) makes the sink flush, and a flushing
+            // sink returns `None` *immediately* rather than after the
+            // timeout. Treat that as a clean end too -- otherwise this would
+            // busy-spin for the whole stall budget and then report a
+            // spurious "capture stalled" on every normal stop.
+            if !matches!(appsink.current_state(), gst::State::Playing | gst::State::Paused) {
+                return Ok(None);
+            }
+
+            stalled_for += CAPTURE_STALL_POLL;
+            if stalled_for < CAPTURE_STALL_TIMEOUT {
+                continue;
+            }
+            bail!(
+                "capture stalled: no encoded frame for {}s while the pipeline was still \
+                 playing (no GStreamer error, no EOS). This is the shape of a portal-side \
+                 give-up -- see `choose_capture_backend` -- rather than a pipeline fault, \
+                 and it will not recover on its own",
+                CAPTURE_STALL_TIMEOUT.as_secs()
+            );
+        };
+        stalled_for = Duration::ZERO;
+        let buffer = sample.buffer().context("pulled sample had no buffer")?;
+        let Some(capture_time_us) = buffer.pts().map(|t| t.useconds() as i64) else {
+            tracing::debug!("skipped an encoded frame with no PTS");
+            continue;
+        };
+        let map = buffer.map_readable().context("failed to map sample buffer readable")?;
+        let is_key_frame = !buffer.flags().contains(gst::BufferFlags::DELTA_UNIT);
+        return Ok(Some((map.as_slice().to_vec(), is_key_frame, capture_time_us)));
+    }
 }
 
 /// Sends an upstream "force key unit" event from `appsink`, propagating to
@@ -310,3 +623,33 @@ pub async fn wait_for_playlist_segments(playlist_path: &Path, min_segments: usiz
         tokio::time::sleep(Duration::from_millis(250)).await;
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_both_shapes_hyprland_reports() {
+        // `version` (plain) and `tag` (v-prefixed) from the same running
+        // compositor, plus the `-N-gSHA` suffix a git build's tag carries.
+        assert_eq!(parse_hyprland_version("0.55.4"), Some((0, 55, 4)));
+        assert_eq!(parse_hyprland_version("v0.55.4"), Some((0, 55, 4)));
+        assert_eq!(parse_hyprland_version("v0.56.0-123-gdeadbee"), Some((0, 56, 0)));
+        assert_eq!(parse_hyprland_version("0.56"), Some((0, 56, 0)));
+        assert_eq!(parse_hyprland_version(" v0.56.2\n"), Some((0, 56, 2)));
+
+        assert_eq!(parse_hyprland_version(""), None);
+        assert_eq!(parse_hyprland_version("unknown"), None);
+    }
+
+    #[test]
+    fn straddles_the_dmabuf_crash_fix_correctly() {
+        // The whole point of the constant: v0.55.4 aborts the compositor on
+        // a DMA-BUF screencast, v0.56.0 is the first release with the fix.
+        assert!(parse_hyprland_version("0.55.4").unwrap() < HYPRLAND_MIN_SAFE_DMABUF);
+        assert!(parse_hyprland_version("0.55.99").unwrap() < HYPRLAND_MIN_SAFE_DMABUF);
+        assert!(parse_hyprland_version("0.56.0").unwrap() >= HYPRLAND_MIN_SAFE_DMABUF);
+        assert!(parse_hyprland_version("0.56.1").unwrap() >= HYPRLAND_MIN_SAFE_DMABUF);
+        assert!(parse_hyprland_version("1.0.0").unwrap() >= HYPRLAND_MIN_SAFE_DMABUF);
+    }
+}

breadcastd/src/cast_mirror.rs

@@ -13,7 +13,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 
 use anyhow::{Context, Result};
-use breadcast_core::caststream::{CastStreamEvent, VideoParams, WEBRTC_NAMESPACE};
+use breadcast_core::caststream::{CastStreamEvent, WEBRTC_NAMESPACE};
 use breadcast_core::pipeline::{
     build_video_pipeline_for_streaming, pull_encoded_frame, request_key_frame, set_video_bitrate_kbps,
 };
@@ -24,11 +24,53 @@ use rust_cast::channels::receiver::CastDeviceApp;
 
 use crate::daemon::DaemonCommand;
 
+/// The encoder's starting target bitrate, in kbps -- must match the
+/// `bitrate=` property `build_video_pipeline_for_streaming` builds the
+/// `vah264enc` with, since [`bitrate_control_step`] treats it as the value
+/// already in effect at t=0.
+const INITIAL_BITRATE_KBPS: u32 = 4000;
+/// Never encode below this. 720p30 below roughly 1.5 Mbps is a wall of
+/// blocking artifacts -- if the link genuinely can't carry that, dropping
+/// frames is a better failure mode than shipping unwatchable video.
+const MIN_BITRATE_KBPS: u32 = 1500;
+/// Never encode above this, regardless of how much headroom the estimator
+/// reports. Matches `VideoParams::default().max_bitrate_bps`, i.e. what the
+/// OFFER told the receiver to expect -- see that constant's doc comment for
+/// why this is 6 Mbps and not higher: real hardware testing showed the AIMD
+/// probe below pinning to whatever this ceiling is for the *entire* session
+/// (the estimator it trusts read a suspiciously flat ~20 Mbps almost the
+/// whole time), and 8 Mbps sustained was more than the previous
+/// network+receiver could actually hold without repeated multi-second
+/// freezes -- though the deeper cause of those freezes turned out to be
+/// `frame_chain_broken` in `facade.cc`, not bitrate on its own. 6 Mbps is
+/// still a very generous ceiling for 720p30; revisit only with real
+/// evidence this specific link+receiver can sustain more, not just because
+/// the estimator claims there's headroom.
+const MAX_BITRATE_KBPS: u32 = 6000;
+
 pub struct CastMirrorSession {
     pipeline: gst::Pipeline,
     session: CastSession,
     capture: Option<CaptureSession>,
-    threads: Vec<std::thread::JoinHandle<()>>,
+    /// The FFI Cast Streaming session. Held here (rather than only inside
+    /// the pump-thread closures, as an earlier version did) so its
+    /// `Drop` -- which calls `breadcast_caststream_sender_destroy` and
+    /// blocks until openscreen's threads stop -- happens at an explicit,
+    /// deterministic point in [`Self::stop`], instead of "whichever
+    /// detached pump thread happened to drop the last `Arc`."
+    sender: Option<Arc<CastStreamSender>>,
+    /// Forwards inbound CASTV2 `urn:x-cast:com.google.cast.webrtc` messages
+    /// (the ANSWER) into the FFI session. Ends when [`CastSession::stop`]
+    /// closes the raw-message channel. Holds an `Arc<CastStreamSender>`.
+    message_pump: Option<std::thread::JoinHandle<()>>,
+    /// Forwards outbound FFI events (the OFFER) onto the CASTV2 connection.
+    /// Ends only once the `CastStreamSender` itself is dropped (that is what
+    /// closes the event channel), so it must be joined *after* `sender` is
+    /// dropped, not before -- joining it first would deadlock.
+    event_pump: Option<std::thread::JoinHandle<()>>,
+    /// Pulls encoded frames from the appsink into the FFI session. Ends on
+    /// pipeline EOS/flush. Holds an `Arc<CastStreamSender>`.
+    frame_pump: Option<std::thread::JoinHandle<()>>,
 }
 
 impl CastMirrorSession {
@@ -45,7 +87,13 @@ impl CastMirrorSession {
         let capture = CaptureSession::start().await.context("failed to start portal screen capture")?;
         let video_node_id = capture.video_node_id();
 
-        let (pipeline, appsink, encoder) =
+        // `video_params` describes what this pipeline will *actually* encode
+        // -- it isn't a constant, because the pipeline picks its capture path
+        // at runtime (see `build_video_pipeline_for_streaming`) and the two
+        // paths differ in resolution and frame rate. It's threaded into the
+        // OFFER below rather than re-derived there, so the advertised stream
+        // and the encoded stream cannot drift apart.
+        let (pipeline, appsink, encoder, video_params) =
             build_video_pipeline_for_streaming(video_node_id).context("failed to build the encode pipeline")?;
 
         {
@@ -74,13 +122,11 @@ impl CastMirrorSession {
         .context("failed to connect and launch the Mirroring receiver")?;
 
         let (sender, stream_events) =
-            CastStreamSender::start(&device.host, "sender-0", session.transport_id(), VideoParams::default())
+            CastStreamSender::start(&device.host, "sender-0", session.transport_id(), video_params)
                 .context("failed to start the Cast Streaming session")?;
         let sender = Arc::new(sender);
 
-        let mut threads = Vec::new();
-
-        threads.push({
+        let message_pump = {
             let sender = sender.clone();
             std::thread::spawn(move || {
                 while let Some(msg) = raw_messages.recv() {
@@ -89,10 +135,10 @@ impl CastMirrorSession {
                     }
                 }
             })
-        });
+        };
 
         let negotiated = Arc::new(AtomicBool::new(false));
-        threads.push({
+        let event_pump = {
             let session = session.clone();
             let negotiated = negotiated.clone();
             std::thread::spawn(move || {
@@ -109,7 +155,7 @@ impl CastMirrorSession {
                     }
                 }
             })
-        });
+        };
 
         tracing::info!(device = %device.name, "sending Cast Streaming OFFER");
         sender.negotiate();
@@ -119,16 +165,22 @@ impl CastMirrorSession {
             tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
         }
         if !negotiated.load(Ordering::Acquire) {
-            let _ = session.stop();
-            capture.close().await.ok();
+            // Bounded for the same reason `Self::stop`'s calls are -- an
+            // unresponsive receiver (which is exactly what "negotiation
+            // timed out" implies) can wedge either of these forever
+            // otherwise, taking the whole single-threaded daemon actor
+            // down with it before this even gets to return an error.
+            stop_session_bounded(&session).await;
+            close_capture_bounded(capture).await;
             anyhow::bail!("never received an ANSWER from {} (negotiation timed out)", device.name);
         }
 
         pipeline.set_state(gst::State::Playing).context("failed to start the encode pipeline")?;
         tracing::info!(device = %device.name, "mirroring started");
 
-        threads.push({
+        let frame_pump = {
             let device_name = device.name.clone();
+            let sender = sender.clone();
             std::thread::spawn(move || {
                 let result = frame_pump_loop(&appsink, &encoder, &sender);
                 if let Err(e) = result {
@@ -140,63 +192,256 @@ impl CastMirrorSession {
                 // thread than drop the notification.
                 let _ = daemon_tx.blocking_send(DaemonCommand::SessionEnded);
             })
-        });
+        };
 
-        Ok(Self { pipeline, session, capture: Some(capture), threads })
+        Ok(Self {
+            pipeline,
+            session,
+            capture: Some(capture),
+            sender: Some(sender),
+            message_pump: Some(message_pump),
+            event_pump: Some(event_pump),
+            frame_pump: Some(frame_pump),
+        })
     }
 
-    /// Tears down the session: stops the pipeline (which unblocks the frame
-    /// pump thread's blocking `appsink.pull_sample()` call), stops the
-    /// CASTV2 session (which ends its io thread, closing the channels the
-    /// other two pump threads block on), then joins every thread.
+    /// Tears down the session. Order matters and is not interchangeable:
+    ///
+    /// 1. Pipeline to `Null` -- unblocks the frame pump's blocking
+    ///    `appsink.pull_sample()`, so it can exit and release its
+    ///    `Arc<CastStreamSender>`.
+    /// 2. Stop the CASTV2 session -- ends its io thread, closing the
+    ///    raw-message channel the message pump blocks on, so it too can exit
+    ///    and release its `Arc`.
+    /// 3. Join those two. After this, no thread is calling into the FFI
+    ///    session and `self.sender` holds the only remaining `Arc`.
+    /// 4. Drop `self.sender` -- runs `breadcast_caststream_sender_destroy`
+    ///    (blocking until openscreen's threads stop) at a point where
+    ///    nothing else can be mid-call into it, and closes the FFI event
+    ///    channel.
+    /// 5. Only *then* join the event pump, which blocks on that channel and
+    ///    would deadlock if joined before step 4.
     pub async fn stop(mut self) {
         if let Err(e) = self.pipeline.set_state(gst::State::Null) {
             tracing::warn!(error = ?e, "failed to stop the encode pipeline cleanly");
         }
-        if let Err(e) = self.session.stop() {
-            tracing::warn!(error = ?e, "failed to cleanly stop the cast session");
-        }
+        stop_session_bounded(&self.session).await;
         if let Some(capture) = self.capture.take() {
-            if let Err(e) = capture.close().await {
-                tracing::warn!(error = ?e, "failed to cleanly close the portal capture session");
-            }
+            close_capture_bounded(capture).await;
         }
-        for thread in self.threads.drain(..) {
-            // These threads all end once the pipeline/session teardown
-            // above propagates to them (see this method's own doc comment)
-            // -- `spawn_blocking` just keeps `.join()`'s wait off the async
-            // runtime's worker threads.
-            if let Err(panic) = tokio::task::spawn_blocking(move || thread.join()).await {
-                tracing::warn!(error = ?panic, "mirror session pump thread join task panicked");
-            }
+
+        join_pump(self.frame_pump.take(), "frame").await;
+        join_pump(self.message_pump.take(), "message").await;
+
+        // Step 4: the blocking FFI teardown, kept off the async runtime's
+        // worker threads for the same reason the joins are.
+        if let Some(sender) = self.sender.take() {
+            let _ = tokio::task::spawn_blocking(move || drop(sender)).await;
         }
+
+        join_pump(self.event_pump.take(), "event").await;
     }
 }
 
+/// `CastSession::stop` blocks on a round trip the receiver has to answer,
+/// over a `rust_cast` connection that offers no read timeout -- if the
+/// receiver has gone unresponsive (wedged decoder, dropped off the
+/// network, etc.) that round trip never returns. Both callers run inside
+/// the single-threaded daemon actor, so an unbounded wait here doesn't
+/// just fail one stop -- it permanently freezes the entire daemon (every
+/// future IPC request hangs too), recoverable only by killing the
+/// process. Bound it: if the receiver hasn't answered in 5s, give up on a
+/// graceful stop and let teardown continue anyway. The io thread may leak
+/// (still blocked in that same call), but a single leaked thread beats an
+/// unrecoverable daemon.
+async fn stop_session_bounded(session: &CastSession) {
+    let session = session.clone();
+    match tokio::time::timeout(std::time::Duration::from_secs(5), tokio::task::spawn_blocking(move || session.stop())).await {
+        Ok(Ok(Err(e))) => tracing::warn!(error = ?e, "failed to cleanly stop the cast session"),
+        Ok(Err(panic)) => tracing::warn!(error = ?panic, "cast session stop task panicked"),
+        Err(_) => tracing::warn!("cast session did not acknowledge stop within 5s (receiver unresponsive?) -- tearing down anyway"),
+        Ok(Ok(Ok(()))) => {}
+    }
+}
+
+/// Same reasoning as [`stop_session_bounded`]: `capture.close()` is a
+/// `Session::close()` D-Bus call to the xdg-desktop-portal backend, which
+/// this project has observed to be flaky (the "Failed to populate
+/// properties cache... UnknownMethod" warnings logged on every portal
+/// session) -- an unbounded `.await` here is just as capable of freezing
+/// the whole daemon actor if that call never gets a reply.
+async fn close_capture_bounded(capture: CaptureSession) {
+    match tokio::time::timeout(std::time::Duration::from_secs(5), capture.close()).await {
+        Ok(Err(e)) => tracing::warn!(error = ?e, "failed to cleanly close the portal capture session"),
+        Err(_) => tracing::warn!("portal capture session did not close within 5s -- abandoning it"),
+        Ok(Ok(())) => {}
+    }
+}
+
+/// Bounded to 5s for the same reason [`CastMirrorSession::stop`]'s own wait
+/// on `CastSession::stop` is: `message_pump` blocks on a channel only the
+/// (possibly wedged, per that comment) cast session io thread ever closes,
+/// so an unbounded join here is exactly as capable of freezing the whole
+/// daemon actor forever. A timed-out thread is abandoned rather than
+/// joined -- it may still be running, but nothing here waits on it again.
+async fn join_pump(handle: Option<std::thread::JoinHandle<()>>, what: &str) {
+    let Some(handle) = handle else { return };
+    match tokio::time::timeout(std::time::Duration::from_secs(5), tokio::task::spawn_blocking(move || handle.join())).await {
+        Ok(Err(panic)) => tracing::warn!(pump = what, error = ?panic, "mirror session pump thread join task panicked"),
+        Err(_) => tracing::warn!(pump = what, "mirror session pump thread did not exit within 5s -- abandoning it"),
+        Ok(Ok(_)) => {}
+    }
+}
+
+/// One step of the encoder-bitrate congestion-control loop: given the
+/// currently-applied target and openscreen's latest bandwidth estimate,
+/// returns the new target in kbps.
+///
+/// openscreen's `BandwidthEstimator` deliberately *under*-estimates capacity
+/// whenever the transmit rate is below it (see its class comment in
+/// `vendor/openscreen/cast/streaming/impl/bandwidth_estimator.h`), and
+/// prescribes a TCP-like response: cut hard when the estimate is below the
+/// current target, ramp back up *gradually* when it's above. An earlier
+/// version of this loop instead did `target = 0.85 * estimate` every second
+/// unconditionally, which multiplies the target by <= 0.85 once a second
+/// with no way back up -- 4000 kbps collapses past 1500 within ~6 seconds
+/// and pins at the floor, which is exactly the "low quality / compression
+/// artifacts" symptom, on a perfectly healthy LAN.
+///
+/// An estimate of 0 means "not enough recent data to say" (documented
+/// return value), and must leave the target alone rather than be treated as
+/// a zero-bandwidth link.
+fn bitrate_control_step(current_kbps: u32, estimate_bps: i32) -> u32 {
+    if estimate_bps <= 0 {
+        return current_kbps;
+    }
+    let estimate_kbps = (estimate_bps / 1000) as u32;
+    let next = if estimate_kbps < current_kbps {
+        // Below target: back off immediately to just under the estimate.
+        ((estimate_kbps as f64) * 0.85) as u32
+    } else {
+        // Headroom: probe upward by 10% per second, not straight to the
+        // estimate -- the estimate is a lower bound, and jumping to it
+        // oscillates.
+        current_kbps + current_kbps / 10
+    };
+    next.clamp(MIN_BITRATE_KBPS, MAX_BITRATE_KBPS)
+}
+
 fn frame_pump_loop(
     appsink: &gstreamer_app::AppSink,
     encoder: &gst::Element,
     sender: &CastStreamSender,
 ) -> Result<()> {
     let mut last_bitrate_update = std::time::Instant::now();
+    let mut current_kbps = INITIAL_BITRATE_KBPS;
+    // `needs_key_frame()` is a snapshot of an atomic the C++ side only
+    // refreshes every 100ms, so it stays true for several frames after a
+    // request has already been sent upstream. Firing a force-key-unit event
+    // per frame in that window makes the encoder emit a burst of IDRs, which
+    // under CBR eats the whole bitrate budget and produces a visible quality
+    // dip on every picture-loss report. One request per refresh window is
+    // enough.
+    let mut last_key_frame_request: Option<std::time::Instant> = None;
+    // Pulled-frame and successful-enqueue counters, logged once/sec
+    // alongside the bitrate step below -- otherwise a stalled pipeline
+    // (upstream not producing samples) and a stalled sender (producing
+    // samples nobody can get rid of) are both silent: this is a per-frame
+    // hot loop, so anything more than a periodic summary would flood the
+    // log rather than help debug either case.
+    let mut pulled_since_log: u32 = 0;
+    let mut enqueued_since_log: u32 = 0;
     loop {
         let Some((data, is_key_frame, capture_time_us)) = pull_encoded_frame(appsink)? else {
             return Ok(()); // EOS -- pipeline was set to Null, or the portal source ended
         };
+        pulled_since_log += 1;
 
-        if sender.needs_key_frame() && !is_key_frame {
+        if sender.needs_key_frame()
+            && !is_key_frame
+            && last_key_frame_request.is_none_or(|t| t.elapsed() >= std::time::Duration::from_millis(250))
+        {
             request_key_frame(appsink);
+            last_key_frame_request = Some(std::time::Instant::now());
+        }
+        if is_key_frame {
+            last_key_frame_request = None;
         }
 
-        if let Err(e) = sender.enqueue_frame(&data, is_key_frame, capture_time_us) {
-            tracing::debug!(error = ?e, "dropped a frame (not negotiated yet or backpressure)");
+        match sender.enqueue_frame(&data, is_key_frame, capture_time_us) {
+            Ok(()) => enqueued_since_log += 1,
+            Err(e) => tracing::debug!(error = ?e, "dropped a frame (not negotiated yet or backpressure)"),
         }
 
         if last_bitrate_update.elapsed() >= std::time::Duration::from_secs(1) {
-            let bps = sender.estimated_bandwidth_bps();
-            let target_kbps = ((bps as f64 * 0.85) / 1000.0).max(500.0) as u32;
-            set_video_bitrate_kbps(encoder, target_kbps);
+            let next_kbps = bitrate_control_step(current_kbps, sender.estimated_bandwidth_bps());
+            if next_kbps != current_kbps {
+                current_kbps = next_kbps;
+                set_video_bitrate_kbps(encoder, current_kbps);
+            }
+            // `enqueued_fps` counts *posted* frames, not accepted ones (see
+            // `CastStreamSender::enqueue_stats`) -- it is the `accepted_fps`
+            // and `rejected_*` fields below that say whether video is
+            // actually reaching the receiver. A run where `enqueued_fps`
+            // holds at 30 while `accepted_fps` drops to 0 is a frozen
+            // picture, and nothing else logged here would show it.
+            let stats = sender.enqueue_stats();
+            tracing::debug!(
+                pulled_fps = pulled_since_log,
+                enqueued_fps = enqueued_since_log,
+                accepted_fps = stats.enqueue_ok,
+                rejected_in_flight = stats.enqueue_max_duration_in_flight,
+                rejected_id_span = stats.enqueue_id_span_limit,
+                rejected_too_large = stats.enqueue_payload_too_large,
+                dropped_non_monotonic = stats.dropped_non_monotonic,
+                in_flight_frames = stats.in_flight_frames,
+                in_flight_ms = stats.in_flight_ms,
+                max_in_flight_ms = stats.max_in_flight_ms,
+                rtt_ms = stats.round_trip_time_ms,
+                bitrate_kbps = current_kbps,
+                estimated_bandwidth_bps = sender.estimated_bandwidth_bps(),
+                "frame pump rate"
+            );
+            pulled_since_log = 0;
+            enqueued_since_log = 0;
             last_bitrate_update = std::time::Instant::now();
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_zero_estimate_leaves_the_target_alone() {
+        assert_eq!(bitrate_control_step(4000, 0), 4000);
+        assert_eq!(bitrate_control_step(4000, -1), 4000);
+    }
+
+    #[test]
+    fn headroom_ramps_up_gradually_and_is_capped() {
+        assert_eq!(bitrate_control_step(4000, 20_000_000), 4400);
+        assert_eq!(bitrate_control_step(MAX_BITRATE_KBPS, 20_000_000), MAX_BITRATE_KBPS);
+    }
+
+    #[test]
+    fn a_low_estimate_backs_off_but_not_below_the_floor() {
+        assert_eq!(bitrate_control_step(4000, 2_000_000), 1700);
+        assert_eq!(bitrate_control_step(4000, 100_000), MIN_BITRATE_KBPS);
+    }
+
+    /// The regression this loop exists to prevent: a *steady* estimate at
+    /// roughly the current encode rate must hold the target there (AIMD
+    /// oscillates a little around it, which is fine), not ratchet it down
+    /// once per second the way `target = 0.85 * estimate` did -- that
+    /// reached the floor in about a dozen iterations.
+    #[test]
+    fn a_steady_estimate_does_not_spiral_downward() {
+        let mut kbps = 4000;
+        for _ in 0..60 {
+            kbps = bitrate_control_step(kbps, 4_000_000);
+            assert!(kbps >= 3000, "target spiralled down to {kbps} kbps on a steady 4 Mbps estimate");
+        }
+    }
+}

breadcastd/src/daemon.rs

@@ -73,6 +73,12 @@ impl ActiveSession {
     }
 }
 
+/// `host` strings come straight from mDNS resolution, so this just checks
+/// the address, not whether a zone id is attached (mDNS never gives us one).
+fn is_link_local_v6(host: &str) -> bool {
+    matches!(host.parse::<std::net::IpAddr>(), Ok(std::net::IpAddr::V6(v6)) if (v6.segments()[0] & 0xffc0) == 0xfe80)
+}
+
 struct Daemon {
     cast_devices: HashMap<String, CastDevice>,
     /// Keyed by `DlnaDevice::url`, the closest thing DLNA has to a stable
@@ -128,7 +134,20 @@ impl Daemon {
                 }
             }
             DaemonCommand::CastDeviceFound(device) => {
-                self.cast_devices.insert(device.id.clone(), device);
+                // mDNS resolves one physical device on every local address
+                // it has -- typically a private IPv4 and a link-local IPv6
+                // -- as separate events carrying the same id. A bare
+                // `fe80::` address has no interface scope attached, so
+                // `TcpStream::connect`ing to it fails outright; don't let
+                // one clobber an already-usable host just because it
+                // happened to resolve more recently.
+                let should_replace = !matches!(
+                    self.cast_devices.get(&device.id),
+                    Some(existing) if is_link_local_v6(&device.host) && !is_link_local_v6(&existing.host)
+                );
+                if should_replace {
+                    self.cast_devices.insert(device.id.clone(), device);
+                }
                 self.broadcast_devices();
             }
             DaemonCommand::CastDeviceLost(id) => {
@@ -163,8 +182,11 @@ impl Daemon {
                     let _ = reply.send(Ok(()));
                 }
                 Err(e) => {
-                    bread_events::emit_mirroring_failed(&self.bread_client, &device.id, &e.to_string());
-                    let _ = reply.send(Err(e.to_string()));
+                    // `{e:#}` (not `{e}`/`to_string()`) so the full anyhow
+                    // context chain reaches the caller/GUI instead of just
+                    // the outermost ".context()" message.
+                    bread_events::emit_mirroring_failed(&self.bread_client, &device.id, &format!("{e:#}"));
+                    let _ = reply.send(Err(format!("{e:#}")));
                 }
             }
             return;
@@ -184,8 +206,8 @@ impl Daemon {
                     let _ = reply.send(Ok(()));
                 }
                 Err(e) => {
-                    bread_events::emit_mirroring_failed(&self.bread_client, &device.url, &e.to_string());
-                    let _ = reply.send(Err(e.to_string()));
+                    bread_events::emit_mirroring_failed(&self.bread_client, &device.url, &format!("{e:#}"));
+                    let _ = reply.send(Err(format!("{e:#}")));
                 }
             }
             return;

breadcastd/src/dlna_mirror.rs

@@ -81,7 +81,13 @@ impl DlnaMirrorSession {
             .context("failed to start the HLS HTTP server")?;
         let stream_url = http.url(lan_ip, "playlist.m3u8");
 
-        wait_for_playlist_segments(&output_dir.join("playlist.m3u8"), 3, Duration::from_secs(20))
+        // Two segments, not three: this is a "don't hand the renderer a 404
+        // playlist" guard, and every segment waited for here is a segment of
+        // already-stale video sitting between the renderer and live (see
+        // `build_video_pipeline`'s note on HLS latency). Two is the minimum
+        // that still proves the encoder is genuinely producing output rather
+        // than having emitted one segment and stalled.
+        wait_for_playlist_segments(&output_dir.join("playlist.m3u8"), 2, Duration::from_secs(20))
             .await
             .context("encode pipeline never produced playable HLS segments")?;