Fix mirroring host-selection race, control-channel fragility, teardown

deadlock, and OFFER/encode resolution mismatch

Found live-testing on real hardware after the first round of fixes:

1. mDNS resolves one Chromecast on every local address it has (a private
   IPv4 and a link-local IPv6 in the common case), as separate
   CastDeviceFound events for the same id. The device map was a plain
   HashMap::insert, so whichever address resolved last won -- and a bare
   fe80:: address has no interface scope attached, so connecting to it
   fails outright. This is what "failed to connect and launch the
   Mirroring receiver" actually was; the message just didn't say why,
   since daemon.rs was converting the anyhow::Error with to_string()
   (Display, outermost .context() only) instead of "{e:#}" (full chain).
   Fixed both: prefer an already-usable host over a link-local one
   instead of always taking the newest resolution, and preserve the full
   error chain to the GUI/bread event log.

2. The CASTV2 receive loop (run_io_loop) treated any error from
   device.receive() as connection-fatal and ended the whole loop --
   including a plain JSON deserialization failure on a single
   MEDIA_STATUS message (rust_cast's struct requires an `images` field
   this receiver didn't send). That killed the receiver-status/control
   channel for the rest of every session, on the very first status
   update, while the RTP stream itself kept flowing obliviously.
   rust_cast::Error already distinguishes Io/Tls/Dns (actually fatal)
   from Serialization/Parsing/etc (a bad message, not a dead socket) --
   only end the loop on the former now.

3. CastSession::stop() blocks on an unbounded reply_rx.recv() waiting for
   the io thread's device.receiver.stop_app() -- a network round trip
   rust_cast gives no way to put a read timeout on. If the receiver ever
   stops responding, that never returns, and since the daemon actor
   processes one command at a time, a single wedged stop_cast freezes
   every future IPC request too, recoverable only by killing the process
   -- which is exactly what was observed live. Bounded both that wait and
   join_pump's thread joins to 5s; past that, log a warning and tear down
   anyway rather than hang forever. The abandoned thread(s) may leak, but
   a leak beats an unrecoverable daemon.

4. VideoParams::default() advertised 1920x1080 in the OFFER while
   build_video_pipeline_for_streaming actually encodes 1280x720 --
   negotiated and actual resolution disagreeing is a real protocol
   violation, not just soft video, and a plausible cause of a receiver
   decoder corrupting or freezing outright rather than merely looking
   worse. Made the OFFER match what's actually sent.

breadcast-core/src/caststream.rs

@@ -48,8 +48,15 @@ pub struct VideoParams {
 impl Default for VideoParams {
     fn default() -> Self {
         Self {
-            width: 1920,
+            // Must match what `build_video_pipeline_for_streaming` actually
-            height: 1080,
+            // encodes (`breadcast-core/src/pipeline/mod.rs`), not just what
+            // we'd like to send -- this OFFER's resolution is what the
+            // receiver allocates its decoder/output surface for. Advertising
+            // 1920x1080 while actually sending 1280x720 frames is a real
+            // protocol mismatch that plausibly explains a receiver decoder
+            // corrupting/freezing rather than just looking soft.
+            width: 1280,
+            height: 720,
             max_bitrate_bps: 8_000_000,
             max_frame_rate_numerator: 30,
             max_frame_rate_denominator: 1,

breadcastd/src/cast_mirror.rs

@@ -204,8 +204,25 @@ impl CastMirrorSession {
         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() {
+        // `CastSession::stop` blocks on a round trip the receiver has to
-            tracing::warn!(error = ?e, "failed to cleanly stop the cast session");
+        // 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. This actor processes one command at a time, so an
+        // unbounded wait here doesn't just fail this 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 tear down anyway. The io thread may leak
+        // (still blocked in that same call), but a single leaked thread
+        // beats an unrecoverable daemon.
+        let session = self.session.clone();
+        let stop_result = tokio::time::timeout(std::time::Duration::from_secs(5), tokio::task::spawn_blocking(move || session.stop())).await;
+        match stop_result {
+            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(()))) => {}
         }
         if let Some(capture) = self.capture.take() {
             if let Err(e) = capture.close().await {
@@ -226,12 +243,18 @@ impl CastMirrorSession {
     }
 }
 
-/// `spawn_blocking` just keeps `.join()`'s wait off the async runtime's
+/// Bounded to 5s for the same reason [`CastMirrorSession::stop`]'s own wait
-/// worker threads.
+/// 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 };
-    if let Err(panic) = tokio::task::spawn_blocking(move || handle.join()).await {
+    match tokio::time::timeout(std::time::Duration::from_secs(5), tokio::task::spawn_blocking(move || handle.join())).await {
-        tracing::warn!(pump = what, error = ?panic, "mirror session pump thread join task panicked");
+        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(_)) => {}
     }
 }