Fix Cast teardown leaks, keyframe latch, and DLNA session lifecycle

Dropped frames never requested a keyframe, SessionEnded skipped ordered
stop (portal/TV/FFI leak, next start could abort), and a late end could
kill the following cast. Failed starts left PlatformClientPosix alive.
DLNA leaked its HTTP server and ignored portal EOS.

Also: start no longer blocks the daemon actor, IPC accept/request loops
stay up, HLS Range is clamped, LAN IP follows the renderer subnet, and
the picker closes before the portal dialog and handles Escape.

breadcast-core/src/capture/portal.rs

@@ -117,6 +117,10 @@ impl CaptureSession {
                 .open(&token_path)
             {
                 use std::io::Write;
+                use std::os::unix::fs::PermissionsExt;
+                // `mode()` only applies on create. A pre-existing
+                // world-readable token would keep its mode otherwise.
+                let _ = file.set_permissions(std::fs::Permissions::from_mode(0o600));
                 let _ = file.write_all(token.as_bytes());
             }
         }

breadcast-core/src/caststream.rs

@@ -225,16 +225,18 @@ impl CastStreamSender {
         }
     }
 
-    /// Enqueues one encoded video access unit (Annex-B H.264) for sending.
-    /// `capture_time_us` only needs to be monotonically increasing and
-    /// proportional to real elapsed time between frames -- it does not need
-    /// to be wall-clock-accurate.
+    /// Posts one encoded video access unit (Annex-B H.264) onto the C++
+    /// TaskRunner. `capture_time_us` only needs to be monotonically
+    /// increasing and proportional to real elapsed time between frames --
+    /// it does not need to be wall-clock-accurate.
     ///
-    /// Returns an error if the session isn't negotiated yet or the frame
-    /// was rejected under backpressure; callers should treat the latter as
-    /// a dropped frame, not a fatal condition (see
-    /// [`Self::needs_key_frame`]/[`Self::estimated_bandwidth_bps`] for how
-    /// to react).
+    /// Returns an error only if the session isn't negotiated (or is
+    /// shutting down). A `Ok(())` means the frame was *posted*, not that
+    /// `Sender::EnqueueFrame` accepted it -- accept/reject is visible via
+    /// [`Self::enqueue_stats`], and a reject that breaks the H.264
+    /// reference chain latches [`Self::needs_key_frame`]. Treating this
+    /// return as accept/reject is how an earlier version hid a frozen
+    /// picture behind a healthy "30fps enqueued" log line.
     pub fn enqueue_frame(&self, data: &[u8], is_key_frame: bool, capture_time_us: i64) -> Result<()> {
         let result = unsafe {
             breadcast_caststream_sender_enqueue_frame(
@@ -246,7 +248,7 @@ impl CastStreamSender {
             )
         };
         if result != 0 {
-            bail!("frame not enqueued (session not negotiated yet)");
+            bail!("frame not posted (session not negotiated or shutting down)");
         }
         Ok(())
     }

breadcast-core/src/discovery.rs

@@ -29,7 +29,7 @@ const ADDRESS_STALE_AFTER: Duration = Duration::from_secs(300);
 /// to two IPv4 addresses (e.g. wired + wireless, or a DHCP lease change),
 /// an unordered choice can silently pick a stale/unreachable one, and pick
 /// a *different* one across otherwise-identical runs.
-fn pick_address(addresses: &[(ScopedIp, Instant)]) -> Option<IpAddr> {
+fn pick_address(addresses: &[(ScopedIp, Instant)]) -> Option<String> {
     let now = Instant::now();
     let mut candidates: Vec<&(ScopedIp, Instant)> = addresses
         .iter()
@@ -50,7 +50,10 @@ fn pick_address(addresses: &[(ScopedIp, Instant)]) -> Option<IpAddr> {
             })
         })
         .or_else(|| candidates.first())
-        .map(|(ip, _)| ip.to_ip_addr())
+        // Prefer ScopedIp's Display so a last-resort link-local IPv6
+        // keeps its `%iface` zone -- `IpAddr` drops it and
+        // `TcpStream::connect("fe80::…")` then fails with EINVAL.
+        .map(|(ip, _)| ip.to_string())
 }
 
 const SERVICE_TYPE: &str = "_googlecast._tcp.local.";
@@ -140,10 +143,21 @@ impl Discovery {
                             }
                         }
 
-                        let Some(host) = pick_address(addrs).map(|ip| ip.to_string()) else {
+                        let Some(host) = pick_address(addrs) else {
                             warn!(%id, "cast device resolved with no usable addresses, skipping");
                             continue;
                         };
+                        // First ServiceResolved is often IPv6-only (or a
+                        // zoneless fe80::). rust_cast / Cast Streaming
+                        // cannot connect to that. Wait for a later
+                        // resolve that carries IPv4 or a scoped address.
+                        if host.parse::<std::net::Ipv6Addr>().is_ok()
+                            && host.starts_with("fe80:")
+                            && !host.contains('%')
+                        {
+                            warn!(%id, %host, "cast device resolved to an unscoped link-local IPv6, waiting for a better address");
+                            continue;
+                        }
 
                         fullname_to_id.insert(info.get_fullname().to_string(), id.clone());
 

breadcast-core/src/dlna/discovery.rs

@@ -6,7 +6,7 @@ use tokio::sync::mpsc;
 use tracing::{debug, warn};
 
 use super::device::DlnaDevice;
-use super::AV_TRANSPORT;
+use super::{AV_TRANSPORT, MEDIA_RENDERER};
 
 /// How often to re-issue an SSDP search burst. Unlike mDNS (continuous
 /// multicast browsing via `mdns-sd` in [`crate::discovery`]), SSDP has no
@@ -69,7 +69,7 @@ impl DlnaDiscovery {
             let mut known: HashMap<String, (DlnaDevice, u32)> = HashMap::new();
 
             loop {
-                let search_target = SearchTarget::URN(AV_TRANSPORT);
+                let search_target = SearchTarget::URN(MEDIA_RENDERER);
                 match rupnp::discover(&search_target, SEARCH_TIMEOUT, None).await {
                     Ok(stream) => {
                         use futures_util::StreamExt;
@@ -86,6 +86,10 @@ impl DlnaDiscovery {
                             };
 
                             let url = device.url().to_string();
+                            if device.find_service(&AV_TRANSPORT).is_none() {
+                                debug!(%url, "UPnP device has no AVTransport, skipping");
+                                continue;
+                            }
                             confirmed.insert(url.clone());
 
                             if let Some(entry) = known.get_mut(&url) {

breadcast-core/src/dlna/mod.rs

@@ -21,3 +21,7 @@ pub use discovery::{DlnaDiscovery, DlnaDiscoveryEvent};
 pub use session::DlnaSession;
 
 const AV_TRANSPORT: URN = URN::service("schemas-upnp-org", "AVTransport", 1);
+/// Device-type search. Many TVs answer M-SEARCH for MediaRenderer but not
+/// for the AVTransport *service* URN (Windows "Cast to Device" searches
+/// this). After resolve we still require AVTransport.
+const MEDIA_RENDERER: URN = URN::device("schemas-upnp-org", "MediaRenderer", 1);

breadcast-core/src/dlna/session.rs

@@ -56,18 +56,36 @@ impl DlnaSession {
     /// renderer would reject with no useful diagnostic.
     pub async fn load(&self, content_url: &str) -> Result<()> {
         let escaped = xml_escape(content_url);
+        // Many Samsung/LG renderers reject an empty CurrentURIMetaData and
+        // need DIDL-Lite + protocolInfo before they will play a live HLS
+        // playlist. The DIDL itself is then XML-escaped for the SOAP body.
+        let didl = format!(
+            r#"<DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/" xmlns:dc="http://purl.org/dc/elements/1.1/" xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/"><item id="0" parentID="-1" restricted="1"><dc:title>breadcast</dc:title><upnp:class>object.item.videoItem</upnp:class><res protocolInfo="http-get:*:application/vnd.apple.mpegurl:*">{escaped}</res></item></DIDL-Lite>"#
+        );
+        let metadata = xml_escape(&didl);
         let set_uri_payload = format!(
-            "<InstanceID>0</InstanceID><CurrentURI>{escaped}</CurrentURI><CurrentURIMetaData></CurrentURIMetaData>"
+            "<InstanceID>0</InstanceID><CurrentURI>{escaped}</CurrentURI><CurrentURIMetaData>{metadata}</CurrentURIMetaData>"
         );
         self.service
             .action(&self.device_url, "SetAVTransportURI", &set_uri_payload)
             .await
             .context("SetAVTransportURI failed")?;
 
-        self.service
+        // Some renderers auto-play after SetURI and then reject Play;
+        // others stay TRANSITIONING for a moment. Either PLAYING state is
+        // success.
+        match self
+            .service
             .action(&self.device_url, "Play", "<InstanceID>0</InstanceID><Speed>1</Speed>")
             .await
-            .context("Play failed")?;
+        {
+            Ok(_) => {}
+            Err(e) => {
+                if self.transport_state().await.ok().as_deref() != Some("PLAYING") {
+                    return Err(e).context("Play failed");
+                }
+            }
+        }
 
         Ok(())
     }
@@ -106,7 +124,7 @@ impl DlnaSession {
     }
 }
 
-fn xml_escape(input: &str) -> String {
+pub(crate) fn xml_escape(input: &str) -> String {
     let mut escaped = String::with_capacity(input.len());
     for c in input.chars() {
         match c {
@@ -120,3 +138,14 @@ fn xml_escape(input: &str) -> String {
     }
     escaped
 }
+
+#[cfg(test)]
+mod tests {
+    use super::xml_escape;
+
+    #[test]
+    fn xml_escape_covers_the_five_markup_chars() {
+        assert_eq!(xml_escape(r#"a&b<c>d"e'f"#), "a&amp;b&lt;c&gt;d&quot;e&apos;f");
+        assert_eq!(xml_escape("http://10.0.0.1:1/t/playlist.m3u8"), "http://10.0.0.1:1/t/playlist.m3u8");
+    }
+}

breadcast-core/src/http_server.rs

@@ -16,8 +16,9 @@ const WORKER_THREADS: usize = 8;
 
 /// Serves `root` (the `hlssink3` output directory: `playlist.m3u8` +
 /// `segment*.ts`) over plain HTTP. Runs a small fixed pool of worker
-/// threads; dropping the handle does not stop the server (there is no clean
-/// shutdown yet — matches the smoke-testing scope of the rest of Phase 2).
+/// threads. [`HttpServer::shutdown`] (also invoked from `Drop`) unblocks
+/// those workers and closes the listener so a finished DLNA session does
+/// not keep the last screen-recording segments reachable on the LAN.
 ///
 /// Every servable path is namespaced under a random token
 /// (`/<token>/playlist.m3u8`, etc. — see [`HttpServer::token`]) rather than
@@ -37,6 +38,7 @@ const WORKER_THREADS: usize = 8;
 pub struct HttpServer {
     addr: SocketAddr,
     token: String,
+    server: Option<Arc<tiny_http::Server>>,
 }
 
 impl HttpServer {
@@ -74,7 +76,19 @@ impl HttpServer {
             });
         }
 
-        Ok(Self { addr, token })
+        Ok(Self { addr, token, server: Some(server) })
+    }
+
+    /// Unblocks every worker and drops the listener. After this returns the
+    /// bind address is free and the path token no longer serves anything.
+    /// Safe to call more than once.
+    pub fn shutdown(&mut self) {
+        let Some(server) = self.server.take() else { return };
+        // `unblock` wakes one `recv()` at a time; wake every worker so
+        // they all observe the error and drop their Arc.
+        for _ in 0..WORKER_THREADS {
+            server.unblock();
+        }
     }
 
     /// The bound address, e.g. `0.0.0.0:41823`. Combine with this
@@ -90,7 +104,17 @@ impl HttpServer {
     /// doc for why that can't just be `self.addr()`), including the
     /// unguessable path token every request must carry.
     pub fn url(&self, host: std::net::IpAddr, relative: &str) -> String {
-        format!("http://{host}:{}/{}/{relative}", self.addr.port(), self.token)
+        let port = self.addr.port();
+        match host {
+            std::net::IpAddr::V6(v6) => format!("http://[{v6}]:{port}/{}/{relative}", self.token),
+            std::net::IpAddr::V4(v4) => format!("http://{v4}:{port}/{}/{relative}", self.token),
+        }
+    }
+}
+
+impl Drop for HttpServer {
+    fn drop(&mut self) {
+        self.shutdown();
     }
 }
 
@@ -123,7 +147,7 @@ fn random_token() -> String {
 /// multi-range (multipart ranges aren't needed for HLS segment fetches, and
 /// falling back to a full 200 response for those is always a valid
 /// response under the HTTP spec).
-fn parse_range(value: &str, len: usize) -> Option<(usize, usize)> {
+pub(crate) fn parse_range(value: &str, len: usize) -> Option<(usize, usize)> {
     let spec = value.strip_prefix("bytes=")?;
     if spec.contains(',') || len == 0 {
         return None;
@@ -233,7 +257,11 @@ fn handle_request(request: tiny_http::Request, root: &Path, token: &str) -> Resu
     ];
 
     let (status, body) = match range {
-        Some((start, end)) if start <= end && end < data.len() => {
+        // RFC 7233: a last-byte-pos past the end is clamped, not 416.
+        // HLS clients often probe `bytes=0-1048575` against a ~400KB
+        // segment; answering 416 stalls playback with no encoder error.
+        Some((start, end)) if start < data.len() && start <= end => {
+            let end = end.min(data.len() - 1);
             headers.push(("Content-Range".to_string(), format!("bytes {start}-{end}/{}", data.len())));
             (206u16, data[start..=end].to_vec())
         }
@@ -264,3 +292,25 @@ fn respond_status(request: tiny_http::Request, status: u16) -> Result<()> {
         .respond(tiny_http::Response::empty(status))
         .context("failed to write HTTP error response")
 }
+
+#[cfg(test)]
+mod tests {
+    use super::parse_range;
+
+    #[test]
+    fn parse_range_accepts_the_usual_hls_shapes() {
+        assert_eq!(parse_range("bytes=0-99", 200), Some((0, 99)));
+        assert_eq!(parse_range("bytes=50-", 200), Some((50, 199)));
+        assert_eq!(parse_range("bytes=-20", 200), Some((180, 199)));
+        assert_eq!(parse_range("bytes=0-0", 200), Some((0, 0)));
+    }
+
+    #[test]
+    fn parse_range_rejects_malformed_or_multipart() {
+        assert_eq!(parse_range("bytes=", 200), None);
+        assert_eq!(parse_range("bytes=-", 200), None);
+        assert_eq!(parse_range("bytes=0-10,20-30", 200), None);
+        assert_eq!(parse_range("items=0-10", 200), None);
+        assert_eq!(parse_range("bytes=0-10", 0), None);
+    }
+}

breadcast-core/src/net.rs

@@ -2,6 +2,8 @@ use std::net::{IpAddr, Ipv4Addr};
 
 use anyhow::{Context, Result};
 
+const EXCLUDED_PREFIXES: &[&str] = &["tailscale", "wg", "docker", "veth", "br-", "virbr", "lo"];
+
 /// Finds this machine's LAN-reachable IPv4 address by enumerating network
 /// interfaces directly, rather than the more common "UDP-connect to a
 /// public address and read back the local endpoint" trick — that trick
@@ -18,8 +20,42 @@ use anyhow::{Context, Result};
 /// Shared by every casting protocol (Cast, DLNA, ...) — they all need to
 /// embed this machine's own address in a URL handed to a receiver device.
 pub fn local_lan_ip() -> Result<IpAddr> {
-    const EXCLUDED_PREFIXES: &[&str] = &["tailscale", "wg", "docker", "veth", "br-", "virbr", "lo"];
+    pick_lan_ip(None)
+}
 
+/// Like [`local_lan_ip`], but prefers the interface whose subnet contains
+/// `peer`. Dual-homed machines (ethernet + wifi, two VLANs) otherwise
+/// embed the wrong host in the HLS URL and the renderer cannot fetch it —
+/// the same silent `LOAD FAILED` class as the Tailscale case above.
+pub fn local_lan_ip_for(peer: IpAddr) -> Result<IpAddr> {
+    pick_lan_ip(Some(peer))
+}
+
+fn pick_lan_ip(peer: Option<IpAddr>) -> Result<IpAddr> {
+    let mut candidates = lan_ipv4_candidates()?;
+
+    if let Some(IpAddr::V4(peer_v4)) = peer {
+        if let Some((_, addr, _)) = candidates
+            .iter()
+            .filter(|(_, addr, prefix)| same_subnet(*addr, peer_v4, *prefix))
+            .max_by_key(|(_, _, prefix)| *prefix)
+        {
+            return Ok(IpAddr::V4(*addr));
+        }
+    }
+
+    // Prefer a conventionally-named physical/Wi-Fi interface when there's a
+    // choice, but any private, non-excluded address is acceptable.
+    candidates.sort_by_key(|(iface, _, _)| !(iface.starts_with("wl") || iface.starts_with("en") || iface.starts_with("eth")));
+
+    candidates
+        .into_iter()
+        .map(|(_, addr, _)| IpAddr::V4(addr))
+        .next()
+        .context("no LAN-reachable IPv4 address found (excluding loopback/VPN/virtual interfaces) — is this machine connected to a network?")
+}
+
+fn lan_ipv4_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
     let output = std::process::Command::new("ip")
         .args(["-4", "-o", "addr", "show", "scope", "global", "up"])
         .output()
@@ -29,7 +65,7 @@ pub fn local_lan_ip() -> Result<IpAddr> {
     }
     let text = String::from_utf8_lossy(&output.stdout);
 
-    let mut candidates: Vec<(String, Ipv4Addr)> = Vec::new();
+    let mut candidates: Vec<(String, Ipv4Addr, u8)> = Vec::new();
     for line in text.lines() {
         // Format: "3: wlan0    inet 10.179.161.89/23 brd ... scope global dynamic wlan0"
         let mut fields = line.split_whitespace();
@@ -42,20 +78,41 @@ pub fn local_lan_ip() -> Result<IpAddr> {
             continue;
         }
         let Some(cidr) = fields.next() else { continue };
-        let Some(addr) = cidr.split('/').next().and_then(|a| a.parse::<Ipv4Addr>().ok()) else { continue };
+        let mut parts = cidr.split('/');
+        let Some(addr) = parts.next().and_then(|a| a.parse::<Ipv4Addr>().ok()) else { continue };
+        let prefix: u8 = parts.next().and_then(|p| p.parse().ok()).unwrap_or(32);
         if !addr.is_private() {
             continue;
         }
-        candidates.push((iface.to_string(), addr));
+        candidates.push((iface.to_string(), addr, prefix.min(32)));
+    }
+    Ok(candidates)
+}
+
+fn same_subnet(a: Ipv4Addr, b: Ipv4Addr, prefix: u8) -> bool {
+    if prefix == 0 {
+        return true;
+    }
+    let mask = if prefix >= 32 {
+        u32::MAX
+    } else {
+        !((1u32 << (32 - prefix)) - 1)
+    };
+    (u32::from(a) & mask) == (u32::from(b) & mask)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn same_subnet_respects_prefix_length() {
+        let a: Ipv4Addr = "10.179.161.89".parse().unwrap();
+        let b: Ipv4Addr = "10.179.160.1".parse().unwrap();
+        let other: Ipv4Addr = "10.0.0.1".parse().unwrap();
+        assert!(same_subnet(a, b, 23));
+        assert!(!same_subnet(a, other, 23));
+        assert!(same_subnet(a, a, 32));
+        assert!(!same_subnet(a, b, 32));
     }
-
-    // Prefer a conventionally-named physical/Wi-Fi interface when there's a
-    // choice, but any private, non-excluded address is acceptable.
-    candidates.sort_by_key(|(iface, _)| !(iface.starts_with("wl") || iface.starts_with("en") || iface.starts_with("eth")));
-
-    candidates
-        .into_iter()
-        .map(|(_, addr)| IpAddr::V4(addr))
-        .next()
-        .context("no LAN-reachable IPv4 address found (excluding loopback/VPN/virtual interfaces) — is this machine connected to a network?")
 }

breadcast-core/src/pipeline/mod.rs

@@ -451,6 +451,7 @@ pub fn pull_encoded_frame(appsink: &gst_app::AppSink) -> Result<Option<(Vec<u8>,
     let poll = gst::ClockTime::from_mseconds(CAPTURE_STALL_POLL.as_millis() as u64);
     let mut stalled_for = Duration::ZERO;
     loop {
+        let pulled_at = std::time::Instant::now();
         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.
@@ -458,16 +459,18 @@ pub fn pull_encoded_frame(appsink: &gst_app::AppSink) -> Result<Option<(Vec<u8>,
                 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) {
+            // the pipeline to Null) flushes the sink *before*
+            // `current_state()` leaves Playing. A flushing sink returns
+            // `None` immediately; counting that as stall time used to
+            // raise a spurious "capture stalled" on every normal stop.
+            if !matches!(appsink.current_state(), gst::State::Playing | gst::State::Paused)
+                || matches!(appsink.pending_state(), gst::State::Null | gst::State::Ready)
+                || pulled_at.elapsed() < Duration::from_millis(20)
+            {
                 return Ok(None);
             }
 
-            stalled_for += CAPTURE_STALL_POLL;
+            stalled_for += pulled_at.elapsed();
             if stalled_for < CAPTURE_STALL_TIMEOUT {
                 continue;
             }
@@ -529,11 +532,56 @@ pub enum RunOutcome {
     Timeout,
 }
 
+/// Blocks the calling thread until the pipeline reports an error or EOS.
+/// Unlike [`run_until_error_or_timeout`] this does not give up after a
+/// fixed duration -- a live mirror session can last hours, and a 1-hour
+/// leftover from the smoke-test helper was leaving GStreamer errors
+/// unobserved for the rest of the cast. Also returns [`RunOutcome::Eos`]
+/// once the pipeline has been torn down from another thread (`Null`), so
+/// a daemon watcher does not sit forever after `stop()`.
+pub fn run_until_eos_or_error(pipeline: &gst::Pipeline) -> Result<RunOutcome> {
+    let bus = pipeline.bus().context("pipeline has no bus")?;
+    loop {
+        let Some(msg) = bus.timed_pop_filtered(
+            gst::ClockTime::from_mseconds(500),
+            &[gst::MessageType::Error, gst::MessageType::Eos, gst::MessageType::Warning],
+        ) else {
+            if !matches!(
+                pipeline.current_state(),
+                gst::State::Playing | gst::State::Paused | gst::State::Ready
+            ) {
+                return Ok(RunOutcome::Eos);
+            }
+            continue;
+        };
+
+        use gst::MessageView;
+        match msg.view() {
+            MessageView::Error(e) => {
+                bail!(
+                    "GStreamer pipeline error from {:?}: {} ({:?})",
+                    e.src().map(|s| s.path_string()),
+                    e.error(),
+                    e.debug()
+                );
+            }
+            MessageView::Warning(w) => {
+                tracing::warn!(
+                    src = ?w.src().map(|s| s.path_string()),
+                    error = %w.error(),
+                    "GStreamer pipeline warning"
+                );
+            }
+            MessageView::Eos(_) => return Ok(RunOutcome::Eos),
+            _ => {}
+        }
+    }
+}
+
 /// Blocks the calling thread until the pipeline reports an error or EOS, or
 /// `timeout` elapses (whichever first). Returns which of those happened, or
 /// `Err` on a real pipeline error. Meant for smoke-testing from a
-/// synchronous `main`/example; the real daemon will want an async/watch-based
-/// version instead of blocking a thread.
+/// synchronous `main`/example; the daemon uses [`run_until_eos_or_error`].
 pub fn run_until_error_or_timeout(pipeline: &gst::Pipeline, timeout: gst::ClockTime) -> Result<RunOutcome> {
     let bus = pipeline.bus().context("pipeline has no bus")?;
     let deadline = std::time::Instant::now() + std::time::Duration::from(timeout);