Compare commits
No commits in common. "092b46af07fe823d8c383184eca800eb9407638b" and "99a04c800bfe0af47146f29b0f7a962c41d5c095" have entirely different histories.
092b46af07
...
99a04c800b
9 changed files with 12 additions and 292 deletions
3
.gitignore
vendored
3
.gitignore
vendored
|
|
@ -35,6 +35,3 @@ CLAUDE.md
|
|||
|
||||
# graphify knowledge-graph output (local tool cache, not for commit)
|
||||
graphify-out/
|
||||
|
||||
# .freebuff local tool state (not for commit)
|
||||
.freebuff/
|
||||
|
|
|
|||
1
Cargo.lock
generated
1
Cargo.lock
generated
|
|
@ -187,7 +187,6 @@ dependencies = [
|
|||
"gstreamer",
|
||||
"gstreamer-app",
|
||||
"gstreamer-video",
|
||||
"if-addrs 0.15.0",
|
||||
"mdns-sd",
|
||||
"rupnp",
|
||||
"rust_cast",
|
||||
|
|
|
|||
|
|
@ -465,9 +465,6 @@ int32_t breadcast_caststream_sender_needs_key_frame(CastStreamSender* sender) {
|
|||
}
|
||||
|
||||
int32_t breadcast_caststream_sender_estimated_bandwidth_bps(CastStreamSender* sender) {
|
||||
if (!sender) {
|
||||
return 0;
|
||||
}
|
||||
return sender->estimated_bandwidth_bps.load(std::memory_order_relaxed);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -21,10 +21,6 @@ gstreamer-video = "0.25"
|
|||
tiny_http = "0.12"
|
||||
rupnp = "3.0.0"
|
||||
futures-util = "0.3.33"
|
||||
# Fallback LAN-IP enumeration used only when the `ip` command is unavailable
|
||||
# (net.rs) -- already a transitive dep via mdns-sd, so this adds no new crate
|
||||
# to the build.
|
||||
if-addrs = "0.15"
|
||||
breadcast-caststream-sys = { path = "../breadcast-caststream-sys" }
|
||||
|
||||
[dev-dependencies]
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
use anyhow::{Context, Result};
|
||||
use rupnp::Service;
|
||||
use std::time::Duration;
|
||||
|
||||
use super::device::DlnaDevice;
|
||||
use super::AV_TRANSPORT;
|
||||
|
|
@ -82,22 +81,8 @@ impl DlnaSession {
|
|||
{
|
||||
Ok(_) => {}
|
||||
Err(e) => {
|
||||
// The renderer may already be auto-playing after SetURI (and
|
||||
// reject a redundant Play), or may still be TRANSITIONING and
|
||||
// only reach PLAYING a moment later -- either is success.
|
||||
// A single immediate GetTransportInfo isn't enough for the
|
||||
// latter, so give it a short bounded window before giving up
|
||||
// on the Play error.
|
||||
let mut attempts = 0u32;
|
||||
loop {
|
||||
if self.transport_state().await.ok().as_deref() == Some("PLAYING") {
|
||||
break;
|
||||
}
|
||||
attempts += 1;
|
||||
if attempts >= 5 {
|
||||
return Err(e).context("Play failed");
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
if self.transport_state().await.ok().as_deref() != Some("PLAYING") {
|
||||
return Err(e).context("Play failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -132,30 +132,12 @@ fn random_token() -> String {
|
|||
.is_ok();
|
||||
if !read_ok {
|
||||
// Unreachable in practice on Linux, but better than a zero-entropy
|
||||
// token if it ever happened. Mix several low-cost, not-directly-
|
||||
// foreseeable sources (two time reads a hair apart, pid, a process
|
||||
// counter) into both hashes and write all 16 bytes, so the fallback
|
||||
// never reduces to a single low-resolution clock tick or per-pid
|
||||
// reuse -- still far weaker than /dev/urandom, which is why it stays
|
||||
// a fallback, but not trivially guessable.
|
||||
let mix = |salt: u64| {
|
||||
use std::hash::{Hash, Hasher};
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
std::time::SystemTime::now().hash(&mut hasher);
|
||||
std::time::SystemTime::UNIX_EPOCH
|
||||
.elapsed()
|
||||
.ok()
|
||||
.map(|d| d.as_nanos())
|
||||
.unwrap_or_default()
|
||||
.hash(&mut hasher);
|
||||
std::process::id().hash(&mut hasher);
|
||||
salt.hash(&mut hasher);
|
||||
hasher.finish()
|
||||
};
|
||||
let a = mix(0x9e3779b97f4a7c15);
|
||||
let b = mix(0xdead_beef_dead_beef);
|
||||
bytes[..8].copy_from_slice(&a.to_le_bytes());
|
||||
bytes[8..].copy_from_slice(&b.to_le_bytes());
|
||||
// token if it ever happened.
|
||||
use std::hash::{Hash, Hasher};
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
std::time::SystemTime::now().hash(&mut hasher);
|
||||
std::process::id().hash(&mut hasher);
|
||||
bytes[..8].copy_from_slice(&hasher.finish().to_le_bytes());
|
||||
}
|
||||
bytes.iter().map(|b| format!("{b:02x}")).collect()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -55,26 +55,7 @@ fn pick_lan_ip(peer: Option<IpAddr>) -> Result<IpAddr> {
|
|||
.context("no LAN-reachable IPv4 address found (excluding loopback/VPN/virtual interfaces) — is this machine connected to a network?")
|
||||
}
|
||||
|
||||
/// Enumerates this machine's private, LAN-reachable IPv4 addresses as
|
||||
/// `(interface, address, prefix-len)` tuples. Primary path parses `ip -o
|
||||
/// addr show` -- the formatting it relies on is stable, and it bakes in the
|
||||
/// `scope global up` filter (excluding link-local 169.254/16 and down
|
||||
/// interfaces) for free -- falling back to a `getifaddrs`-based enumeration
|
||||
/// (`if-addrs`, already in the dependency tree via `mdns-sd`) when the `ip`
|
||||
/// command is unavailable, so the LAN-IP discovery doesn't silently depend on
|
||||
/// iproute2 being installed.
|
||||
fn lan_ipv4_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
|
||||
match ip_command_candidates() {
|
||||
Ok(candidates) if !candidates.is_empty() => return Ok(candidates),
|
||||
Ok(_) => {} // `ip` ran but found no private address -- genuine, fall through
|
||||
Err(e) => {
|
||||
tracing::debug!(error = ?e, "`ip addr show` unavailable, falling back to getifaddrs");
|
||||
}
|
||||
}
|
||||
getifaddrs_candidates()
|
||||
}
|
||||
|
||||
fn ip_command_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
|
||||
let output = std::process::Command::new("ip")
|
||||
.args(["-4", "-o", "addr", "show", "scope", "global", "up"])
|
||||
.output()
|
||||
|
|
@ -108,30 +89,6 @@ fn ip_command_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
|
|||
Ok(candidates)
|
||||
}
|
||||
|
||||
/// `getifaddrs`-based counterpart to [`ip_command_candidates`], mirroring the
|
||||
/// `ip` path's filters (up, non-loopback, private, not link-local, not an
|
||||
/// excluded interface name) so the two produce equivalent candidate sets.
|
||||
fn getifaddrs_candidates() -> Result<Vec<(String, Ipv4Addr, u8)>> {
|
||||
let ifaces = if_addrs::get_if_addrs().context("failed to enumerate network interfaces (getifaddrs fallback)")?;
|
||||
let mut candidates: Vec<(String, Ipv4Addr, u8)> = Vec::new();
|
||||
for iface in ifaces {
|
||||
if EXCLUDED_PREFIXES.iter().any(|p| iface.name.starts_with(p)) {
|
||||
continue;
|
||||
}
|
||||
if !iface.is_oper_up() {
|
||||
continue;
|
||||
}
|
||||
let if_addrs::IfAddr::V4(v4) = iface.addr else { continue };
|
||||
// Mirrors `type(-4 -o addr show scope global up)`: drop loopback,
|
||||
// link-local (169.254) and non-private networks.
|
||||
if v4.ip.is_loopback() || v4.ip.is_link_local() || !v4.ip.is_private() {
|
||||
continue;
|
||||
}
|
||||
candidates.push((iface.name.clone(), v4.ip, v4.prefixlen.min(32)));
|
||||
}
|
||||
Ok(candidates)
|
||||
}
|
||||
|
||||
fn same_subnet(a: Ipv4Addr, b: Ipv4Addr, prefix: u8) -> bool {
|
||||
if prefix == 0 {
|
||||
return true;
|
||||
|
|
|
|||
|
|
@ -97,16 +97,9 @@ impl ActiveSession {
|
|||
}
|
||||
}
|
||||
|
||||
/// `host` strings come straight from mDNS resolution, and mdns-sd renders a
|
||||
/// link-local IPv6 it resolved via `ScopedIp` with a `%zone` suffix (e.g.
|
||||
/// `fe80::…%wlan0`) so the address stays connectable. `IpAddr` can't parse
|
||||
/// that suffix (std rejects zone-ids in `FromStr`), so the zone has to be
|
||||
/// stripped before the network prefix can be examined -- otherwise a scoped
|
||||
/// `fe80::…%iface` host would fail this check and, in `CastDeviceFound`, be
|
||||
/// allowed to clobber a good routable host that `rust_cast` could actually
|
||||
/// connect to.
|
||||
/// `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 {
|
||||
let host = host.split('%').next().unwrap_or(host);
|
||||
matches!(host.parse::<std::net::IpAddr>(), Ok(std::net::IpAddr::V6(v6)) if (v6.segments()[0] & 0xffc0) == 0xfe80)
|
||||
}
|
||||
|
||||
|
|
@ -184,23 +177,6 @@ impl Daemon {
|
|||
self.state = StateInfo::Idle;
|
||||
self.broadcast_state();
|
||||
bread_events::emit_mirroring_stopped(&self.bread_client);
|
||||
return;
|
||||
}
|
||||
// No installed session, but a start is currently in flight
|
||||
// (portal picker / OFFER-ANSWER). The event pump only reports
|
||||
// SessionEnded once negotiation succeeded, so a matching
|
||||
// generation here means the in-flight session died between
|
||||
// `Negotiated` and `StartFinished` landing. If we ignore it,
|
||||
// the pending `StartFinished` would install the dead session
|
||||
// and leave the UI stuck on "Casting" until the frame pump
|
||||
// happens to notice. Invalidate the pending start exactly as
|
||||
// `StopCast` does -- bump the generation so
|
||||
// `on_start_finished` tears the session down instead of
|
||||
// installing it.
|
||||
if self.starting {
|
||||
tracing::debug!("in-flight mirror session ended before it was installed");
|
||||
self.starting = false;
|
||||
self.start_generation = self.start_generation.wrapping_add(1);
|
||||
}
|
||||
}
|
||||
DaemonCommand::StartFinished { generation, outcome, reply } => {
|
||||
|
|
@ -360,159 +336,3 @@ impl Daemon {
|
|||
let _ = self.events_tx.send(ServerMessage::Event { event: "device_list_changed".to_string(), data });
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::collections::HashMap;
|
||||
|
||||
use bread_utils::bread_client::BreadClient;
|
||||
use tokio::sync::{broadcast, mpsc, oneshot};
|
||||
|
||||
use super::*;
|
||||
|
||||
/// A daemon parked mid-`StartCast`: `starting` is set and a concrete
|
||||
/// generation is outstanding, but no `StartFinished` has landed yet.
|
||||
/// This is exactly the window both races below target -- the portal
|
||||
/// picker / OFFER/ANSWER negotiation is still running off the actor,
|
||||
/// and the session that `StartFinished` would install is not installed.
|
||||
///
|
||||
/// A real `MirrorSession` can't be built in a unit test (it needs the
|
||||
/// portal capture session, a GStreamer encode pipeline, and a live
|
||||
/// renderer on the LAN), so the `Stale StartFinished` probes below use
|
||||
/// an `Err(StartFailed)` outcome. That still exercises the whole point
|
||||
/// of the guard: `on_start_finished` checks `generation`/`starting`
|
||||
/// *before* branching on the outcome, so the stale event is rejected
|
||||
/// with "start cancelled", never installs a session, and the UI stays
|
||||
/// `Idle`. The `Ok` branch of that same guard calls
|
||||
/// `started.session.stop()` to tear the dead session down -- the literal
|
||||
/// teardown line -- which is the only part not separately asserted here.
|
||||
fn daemon_with_in_flight_start(generation: u64) -> (Daemon, broadcast::Receiver<ServerMessage>) {
|
||||
let (events_tx, events_rx) = broadcast::channel(16);
|
||||
let (self_tx, _self_rx) = mpsc::channel(16);
|
||||
let daemon = Daemon {
|
||||
cast_devices: HashMap::new(),
|
||||
dlna_devices: HashMap::new(),
|
||||
state: StateInfo::Idle,
|
||||
active_session: None,
|
||||
starting: true,
|
||||
start_generation: generation,
|
||||
events_tx,
|
||||
bread_client: BreadClient::connect("daemon-test"),
|
||||
self_tx,
|
||||
};
|
||||
(daemon, events_rx)
|
||||
}
|
||||
|
||||
/// A `StartFinished` carrying the *dead* session's generation -- the
|
||||
/// probe both race tests replay after invalidating the in-flight start,
|
||||
/// to prove it can't sneak a session back in.
|
||||
fn stale_start_finished(reply: oneshot::Sender<Result<(), String>>) -> DaemonCommand {
|
||||
DaemonCommand::StartFinished {
|
||||
generation: 7,
|
||||
outcome: Err(StartFailed {
|
||||
device_id: "living-room-tv".to_string(),
|
||||
error: "negotiation failed".to_string(),
|
||||
}),
|
||||
reply,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn link_local_v6_is_detected_with_and_without_a_zone() {
|
||||
// Scoped form: mdns-sd's ScopedIp Display renders link-local IPv6 as
|
||||
// `fe80::…%<iface>`, and IpAddr can't parse the `%` suffix -- this is
|
||||
// the exact form the daemon-clobber bug shipped with.
|
||||
assert!(is_link_local_v6("fe80::1234%wlan0"));
|
||||
assert!(is_link_local_v6("fe80::1234%3"));
|
||||
// Unscoped (bare)`fe80::…` also matches.
|
||||
assert!(is_link_local_v6("fe80::1234"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_link_local_addresses_are_not_marked_link_local() {
|
||||
assert!(!is_link_local_v6("fe80"));
|
||||
assert!(!is_link_local_v6("192.168.1.50"));
|
||||
assert!(!is_link_local_v6("fd00::1")); // ULA
|
||||
assert!(!is_link_local_v6("2606:4700::1111")); // public
|
||||
assert!(!is_link_local_v6("2606:4700::1111%wlan0")); // public + spurious zone
|
||||
assert!(!is_link_local_v6("not-an-ip"));
|
||||
}
|
||||
|
||||
/// The session's event pump reports death (`SessionEnded`) *between*
|
||||
/// negotiation succeeding and `StartFinished` landing -- the exact race
|
||||
/// that used to install the already-dead session and leave the UI stuck
|
||||
/// on "Casting" until the frame pump happened to notice.
|
||||
#[tokio::test]
|
||||
async fn session_ended_during_an_in_flight_start_invalidates_the_pending_start() {
|
||||
let (mut daemon, mut events_rx) = daemon_with_in_flight_start(7);
|
||||
|
||||
daemon.handle(DaemonCommand::SessionEnded { generation: 7 }).await;
|
||||
|
||||
// Invalidate the pending start exactly as StopCast does: clear
|
||||
// `starting` and bump the generation so on_start_finished tears the
|
||||
// dead session down instead of installing it.
|
||||
assert!(!daemon.starting, "in-flight start must be cancelled");
|
||||
assert_eq!(daemon.start_generation, 8);
|
||||
assert!(daemon.active_session.is_none());
|
||||
assert!(matches!(daemon.state, StateInfo::Idle));
|
||||
|
||||
// The stale StartFinished from the dead session must not install it,
|
||||
// and the caller must hear that the start was cancelled.
|
||||
let (reply, reply_rx) = oneshot::channel();
|
||||
daemon.handle(stale_start_finished(reply)).await;
|
||||
|
||||
assert_eq!(reply_rx.await.unwrap(), Err("start cancelled".to_string()));
|
||||
assert!(daemon.active_session.is_none(), "dead session must not be installed");
|
||||
assert!(matches!(daemon.state, StateInfo::Idle), "UI must return to / stay Idle");
|
||||
|
||||
// A cancelled in-flight start broadcasts nothing -- no bogus
|
||||
// state_changed to a UI that is already Idle.
|
||||
assert!(events_rx.try_recv().is_err());
|
||||
}
|
||||
|
||||
/// `StopCast` while the portal/negotiation is still running must cancel
|
||||
/// the in-flight start, so its eventual `StartFinished` is rejected
|
||||
/// rather than installing a session the user just explicitly stopped.
|
||||
#[tokio::test]
|
||||
async fn stop_cast_cancels_an_in_flight_start() {
|
||||
let (mut daemon, mut events_rx) = daemon_with_in_flight_start(7);
|
||||
|
||||
let (reply, reply_rx) = oneshot::channel();
|
||||
daemon.handle(DaemonCommand::StopCast { reply }).await;
|
||||
|
||||
// StopCast still resolves Ok -- there was nothing installed to stop,
|
||||
// only a start to cancel.
|
||||
assert!(reply_rx.await.unwrap().is_ok());
|
||||
assert!(!daemon.starting, "StopCast must cancel the in-flight start");
|
||||
assert_eq!(daemon.start_generation, 8);
|
||||
assert!(daemon.active_session.is_none());
|
||||
assert!(matches!(daemon.state, StateInfo::Idle));
|
||||
|
||||
// A late StartFinished from the cancelled start never installs.
|
||||
let (reply, reply_rx) = oneshot::channel();
|
||||
daemon.handle(stale_start_finished(reply)).await;
|
||||
assert_eq!(reply_rx.await.unwrap(), Err("start cancelled".to_string()));
|
||||
assert!(daemon.active_session.is_none(), "cancelled start must not install a session");
|
||||
assert!(matches!(daemon.state, StateInfo::Idle));
|
||||
|
||||
// StopCast itself broadcast the transition back to Idle.
|
||||
assert!(events_rx.try_recv().is_ok());
|
||||
}
|
||||
|
||||
/// Defensive: a `SessionEnded` whose generation no longer matches (a
|
||||
/// late notification from an already-stopped or already-cancelled
|
||||
/// session) must be ignored without touching the current generation, so
|
||||
/// it can't confuse a brand-new later start.
|
||||
#[tokio::test]
|
||||
async fn a_stale_session_ended_while_idle_is_ignored_without_bumping_generation() {
|
||||
let (mut daemon, _events_rx) = daemon_with_in_flight_start(0);
|
||||
daemon.starting = false; // not starting -- the stale event's own forerunner already finished
|
||||
|
||||
daemon.handle(DaemonCommand::SessionEnded { generation: 99 }).await;
|
||||
|
||||
assert!(!daemon.starting);
|
||||
assert_eq!(daemon.start_generation, 0, "stale SessionEnded must not bump a future start's generation");
|
||||
assert!(daemon.active_session.is_none());
|
||||
assert!(matches!(daemon.state, StateInfo::Idle));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -73,17 +73,8 @@ async fn main() -> anyhow::Result<()> {
|
|||
tracing::info!(?device, "Cast device found");
|
||||
bread_events::emit_device_found(&bread_client, &device.id, &device.name, &device.model, "cast");
|
||||
known_cast_devices.insert(device.id.clone(), device.clone());
|
||||
// Forward only on an actual change. The daemon
|
||||
// actor is single-threaded and `Found` fires on
|
||||
// every mDNS re-resolution (per address, per
|
||||
// re-browse), so forwarding identical devices too
|
||||
// would spam `device_list_changed` broadcasts and
|
||||
// delay `list_devices`/`start_cast` replies that
|
||||
// share the same actor queue. The daemon's own
|
||||
// `CastDeviceLost`→refind logic still works, since
|
||||
// a loss is a state change on its side too.
|
||||
let _ = daemon_tx.send(DaemonCommand::CastDeviceFound(device)).await;
|
||||
}
|
||||
let _ = daemon_tx.send(DaemonCommand::CastDeviceFound(device)).await;
|
||||
}
|
||||
Some(DiscoveryEvent::Lost { id }) => {
|
||||
tracing::info!(%id, "Cast device lost");
|
||||
|
|
@ -108,12 +99,8 @@ async fn main() -> anyhow::Result<()> {
|
|||
tracing::info!(?device, "DLNA device found");
|
||||
bread_events::emit_device_found(&bread_client, &device.url, &device.friendly_name, "DLNA renderer", "dlna");
|
||||
known_dlna_devices.insert(device.url.clone(), device.clone());
|
||||
// Same rationale as the Cast arm above: each SSDP
|
||||
// poll only reports devices it *this* poll
|
||||
// confirmed, but a device re-confirmed unchanged
|
||||
// needs no daemon flush.
|
||||
let _ = daemon_tx.send(DaemonCommand::DlnaDeviceFound(device)).await;
|
||||
}
|
||||
let _ = daemon_tx.send(DaemonCommand::DlnaDeviceFound(device)).await;
|
||||
}
|
||||
Some(DlnaDiscoveryEvent::Lost { url }) => {
|
||||
tracing::info!(%url, "DLNA device lost");
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue