feat(qconnect): add official LAN wire service

cx

crates/Cargo.lock

@@ -4732,6 +4732,24 @@ dependencies = [
  "thiserror 2.0.20",
 ]
 
+[[package]]
+name = "qconnect-lan"
+version = "2.0.2"
+dependencies = [
+ "if-addrs 0.15.0",
+ "mdns-sd",
+ "reqwest",
+ "rustls",
+ "serde",
+ "serde_json",
+ "socket2",
+ "thiserror 2.0.20",
+ "tiny_http",
+ "tokio",
+ "url",
+ "zeroize",
+]
+
 [[package]]
 name = "qconnect-protocol"
 version = "2.0.2"

crates/Cargo.toml

@@ -31,6 +31,7 @@ members = [
     "qconnect-protocol",
     "qconnect-transport-ws",
     "qconnect-app",
+    "qconnect-lan",
     # The HiFi Wizard's frontend-agnostic core (ADR-006), consumed by Qt.
     "qbz-dac-wizard-core",
     "qbz-text-utils",

crates/qconnect-lan/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "qconnect-lan"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+description = "Official Qobuz Connect LAN receiver wire and discovery"
+
+[dependencies]
+serde = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+tiny_http = "0.12"
+mdns-sd = "0.21"
+if-addrs = "0.15"
+socket2 = "0.6"
+url = "2"
+zeroize = "1"
+
+[dev-dependencies]
+reqwest = { workspace = true }
+rustls = { version = "0.23", features = ["aws_lc_rs"] }
+tokio = { workspace = true }

crates/qconnect-lan/src/admission.rs

@@ -0,0 +1,143 @@
+use std::sync::{Arc, Condvar, Mutex};
+use std::time::Duration;
+
+use crate::HandoffCandidate;
+
+#[derive(Default)]
+struct AdmissionSlot {
+    candidate: Option<HandoffCandidate>,
+    closed: bool,
+}
+
+struct AdmissionShared {
+    slot: Mutex<AdmissionSlot>,
+    changed: Condvar,
+}
+
+#[derive(Clone)]
+pub struct AdmissionSender {
+    shared: Arc<AdmissionShared>,
+}
+
+pub struct AdmissionInbox {
+    shared: Arc<AdmissionShared>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SubmitError {
+    Closed,
+}
+
+/// One pending candidate, latest-wins. Replacement drops and zeroizes the old
+/// candidate before returning; memory use never grows with POST bursts.
+pub fn admission_channel() -> (AdmissionSender, AdmissionInbox) {
+    let shared = Arc::new(AdmissionShared {
+        slot: Mutex::new(AdmissionSlot::default()),
+        changed: Condvar::new(),
+    });
+    (
+        AdmissionSender {
+            shared: Arc::clone(&shared),
+        },
+        AdmissionInbox { shared },
+    )
+}
+
+impl AdmissionSender {
+    pub fn submit(&self, candidate: HandoffCandidate) -> Result<(), SubmitError> {
+        let mut slot = self
+            .shared
+            .slot
+            .lock()
+            .expect("LAN admission lock poisoned");
+        if slot.closed {
+            return Err(SubmitError::Closed);
+        }
+        slot.candidate = Some(candidate);
+        self.shared.changed.notify_one();
+        Ok(())
+    }
+
+    pub fn close(&self) {
+        let mut slot = self
+            .shared
+            .slot
+            .lock()
+            .expect("LAN admission lock poisoned");
+        slot.closed = true;
+        slot.candidate = None;
+        self.shared.changed.notify_all();
+    }
+}
+
+impl AdmissionInbox {
+    pub fn try_take(&self) -> Option<HandoffCandidate> {
+        self.shared
+            .slot
+            .lock()
+            .expect("LAN admission lock poisoned")
+            .candidate
+            .take()
+    }
+
+    pub fn take_timeout(&self, timeout: Duration) -> Option<HandoffCandidate> {
+        let slot = self
+            .shared
+            .slot
+            .lock()
+            .expect("LAN admission lock poisoned");
+        let mut slot = self
+            .shared
+            .changed
+            .wait_timeout_while(slot, timeout, |slot| {
+                slot.candidate.is_none() && !slot.closed
+            })
+            .expect("LAN admission lock poisoned")
+            .0;
+        slot.candidate.take()
+    }
+
+    pub fn is_closed(&self) -> bool {
+        self.shared
+            .slot
+            .lock()
+            .expect("LAN admission lock poisoned")
+            .closed
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::LanJwtToken;
+
+    fn candidate(session: &str) -> HandoffCandidate {
+        HandoffCandidate::new(
+            session.to_string(),
+            LanJwtToken::new("https://api.example".into(), 999, "api".into()),
+            LanJwtToken::new("wss://qws.example".into(), 999, "qws".into()),
+            true,
+        )
+    }
+
+    #[test]
+    fn pending_slot_is_latest_wins_and_bounded() {
+        let (sender, inbox) = admission_channel();
+        sender.submit(candidate("first")).unwrap();
+        sender.submit(candidate("second")).unwrap();
+
+        assert_eq!(inbox.try_take().unwrap().session_id(), "second");
+        assert!(inbox.try_take().is_none());
+    }
+
+    #[test]
+    fn close_drops_pending_and_rejects_late_submit() {
+        let (sender, inbox) = admission_channel();
+        sender.submit(candidate("pending")).unwrap();
+        sender.close();
+
+        assert!(inbox.is_closed());
+        assert!(inbox.try_take().is_none());
+        assert_eq!(sender.submit(candidate("late")), Err(SubmitError::Closed));
+    }
+}

crates/qconnect-lan/src/lib.rs

@@ -0,0 +1,20 @@
+//! Official Qobuz Connect LAN receiver surface.
+//!
+//! This crate owns only local discovery, HTTP wire validation and bounded
+//! admission. It deliberately has no Qobuz client, player, Qt or daemon
+//! dependency; credential validation and activation belong to the coordinator.
+
+mod admission;
+mod mdns;
+mod model;
+mod projection;
+mod server;
+mod validation;
+
+pub use admission::{admission_channel, AdmissionInbox, AdmissionSender, SubmitError};
+pub use model::{
+    ConnectInfo, DeviceType, DisplayInfo, HandoffCandidate, LanJwtToken, MaxAudioQuality,
+};
+pub use projection::LanProjection;
+pub use server::{LanError, LanService, LanServiceConfig, SERVICE_TYPE};
+pub use validation::{EndpointPolicy, ValidationError, MAX_BODY_BYTES};

crates/qconnect-lan/src/mdns.rs

@@ -0,0 +1,122 @@
+use std::collections::HashMap;
+use std::net::IpAddr;
+use std::time::Duration;
+
+use mdns_sd::{ServiceDaemon, ServiceInfo};
+
+use crate::server::{LanError, SERVICE_TYPE};
+
+pub(crate) struct MdnsRegistration {
+    daemon: ServiceDaemon,
+    fullname: String,
+}
+
+impl MdnsRegistration {
+    pub(crate) fn register(
+        device_uuid: &str,
+        port: u16,
+        sdk_version: &str,
+        addresses: Option<Vec<IpAddr>>,
+    ) -> Result<Self, LanError> {
+        let mut addresses = match addresses {
+            Some(addresses) => addresses,
+            None => lan_ipv4_addresses()?,
+        };
+        addresses.retain(|ip| ip.is_ipv4() && !ip.is_loopback() && !ip.is_unspecified());
+        addresses.sort_unstable();
+        addresses.dedup();
+        if addresses.is_empty() {
+            return Err(LanError::NoLanAddresses);
+        }
+
+        let info = service_info(device_uuid, port, sdk_version, &addresses)?;
+        let fullname = info.get_fullname().to_string();
+        let daemon = ServiceDaemon::new()?;
+        daemon.register(info)?;
+
+        Ok(Self { daemon, fullname })
+    }
+
+    pub(crate) fn shutdown(self) {
+        if let Ok(receiver) = self.daemon.unregister(&self.fullname) {
+            let _ = receiver.recv_timeout(Duration::from_secs(2));
+        }
+        if let Ok(receiver) = self.daemon.shutdown() {
+            let _ = receiver.recv_timeout(Duration::from_secs(2));
+        }
+    }
+}
+
+fn service_info(
+    device_uuid: &str,
+    port: u16,
+    sdk_version: &str,
+    addresses: &[IpAddr],
+) -> Result<ServiceInfo, mdns_sd::Error> {
+    let short = short_id(device_uuid);
+    let instance = format!("QBZ-{short}");
+    let hostname = format!("qbz-{short}.local.");
+    let properties = HashMap::from([
+        ("device_uuid".to_string(), device_uuid.to_string()),
+        ("sdk_version".to_string(), sdk_version.to_string()),
+        ("path".to_string(), String::new()),
+    ]);
+    ServiceInfo::new(
+        SERVICE_TYPE,
+        &instance,
+        &hostname,
+        addresses,
+        port,
+        properties,
+    )
+    .map(ServiceInfo::enable_addr_auto)
+}
+
+fn lan_ipv4_addresses() -> Result<Vec<IpAddr>, LanError> {
+    let mut addresses = if_addrs::get_if_addrs()
+        .map_err(|_| LanError::AddressDiscovery)?
+        .into_iter()
+        .map(|interface| interface.ip())
+        .filter(|ip| ip.is_ipv4() && !ip.is_loopback() && !ip.is_unspecified())
+        .collect::<Vec<_>>();
+    addresses.sort_unstable();
+    addresses.dedup();
+    Ok(addresses)
+}
+
+fn short_id(device_uuid: &str) -> String {
+    let short = device_uuid
+        .chars()
+        .filter(|ch| ch.is_ascii_alphanumeric())
+        .take(8)
+        .collect::<String>()
+        .to_ascii_lowercase();
+    if short.is_empty() {
+        "renderer".to_string()
+    } else {
+        short
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn instance_short_id_is_dns_safe_and_stable() {
+        assert_eq!(short_id("550e8400-e29b-41d4-a716-446655440000"), "550e8400");
+        assert_eq!(short_id("---"), "renderer");
+    }
+
+    #[test]
+    fn service_info_has_official_type_and_required_txt_union() {
+        let uuid = "550e8400-e29b-41d4-a716-446655440000";
+        let info = service_info(uuid, 43210, "0.9.5", &["192.0.2.10".parse().unwrap()]).unwrap();
+
+        assert_eq!(info.get_type(), SERVICE_TYPE);
+        assert_eq!(info.get_port(), 43210);
+        assert_eq!(info.get_property_val_str("device_uuid"), Some(uuid));
+        assert_eq!(info.get_property_val_str("sdk_version"), Some("0.9.5"));
+        assert_eq!(info.get_property_val_str("path"), Some(""));
+    }
+}

crates/qconnect-lan/src/model.rs

@@ -0,0 +1,161 @@
+use serde::{Deserialize, Serialize};
+use zeroize::Zeroize;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+pub enum MaxAudioQuality {
+    MP3,
+    #[serde(rename = "UP_TO_CD")]
+    UpToCd,
+    #[serde(rename = "UP_TO_HIRES_96")]
+    UpToHires96,
+    #[serde(rename = "UP_TO_HIRES_192")]
+    UpToHires192,
+    #[serde(rename = "UP_TO_HIRES_384")]
+    UpToHires384,
+    #[serde(rename = "UNKNOWN")]
+    Unknown,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+pub enum DeviceType {
+    Phone,
+    Speaker,
+    Streamer,
+    TV,
+    Soundbar,
+    Computer,
+    Headset,
+    Tablet,
+    GoogleCast,
+    Unknown,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct DisplayInfo {
+    pub friendly_name: String,
+    pub serial_number: String,
+    pub brand_display_name: String,
+    pub model_display_name: String,
+    pub max_audio_quality: MaxAudioQuality,
+    #[serde(rename = "type")]
+    pub device_type: DeviceType,
+    pub software_version: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct ConnectInfo {
+    pub app_id: String,
+    pub current_session_id: Option<String>,
+}
+
+/// Raw official LAN token shape. Deliberately has no `Debug` implementation.
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct RawLanJwtToken {
+    pub endpoint: String,
+    pub exp: i64,
+    pub jwt: String,
+}
+
+impl Drop for RawLanJwtToken {
+    fn drop(&mut self) {
+        self.endpoint.zeroize();
+        self.jwt.zeroize();
+    }
+}
+
+/// Raw official handoff body. Deliberately has no `Debug` implementation.
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct RawHandoffRequest {
+    pub session_id: String,
+    pub jwt_api: RawLanJwtToken,
+    pub jwt_qconnect: RawLanJwtToken,
+    pub become_active: bool,
+}
+
+impl Drop for RawHandoffRequest {
+    fn drop(&mut self) {
+        self.session_id.zeroize();
+    }
+}
+
+/// Validated delegated token. Access is borrow-only so Drop always scrubs the
+/// original token and endpoint buffers.
+pub struct LanJwtToken {
+    endpoint: String,
+    exp: i64,
+    jwt: String,
+}
+
+impl LanJwtToken {
+    pub(crate) fn new(endpoint: String, exp: i64, jwt: String) -> Self {
+        Self { endpoint, exp, jwt }
+    }
+
+    pub fn endpoint(&self) -> &str {
+        &self.endpoint
+    }
+
+    pub const fn expires_at(&self) -> i64 {
+        self.exp
+    }
+
+    pub fn jwt(&self) -> &str {
+        &self.jwt
+    }
+}
+
+impl Drop for LanJwtToken {
+    fn drop(&mut self) {
+        self.endpoint.zeroize();
+        self.jwt.zeroize();
+    }
+}
+
+/// A structurally and locally validated handoff admitted for cloud validation.
+/// Deliberately has no `Debug`, `Serialize` or `Clone` implementation.
+pub struct HandoffCandidate {
+    session_id: String,
+    jwt_api: LanJwtToken,
+    jwt_qconnect: LanJwtToken,
+    become_active: bool,
+}
+
+impl HandoffCandidate {
+    pub(crate) fn new(
+        session_id: String,
+        jwt_api: LanJwtToken,
+        jwt_qconnect: LanJwtToken,
+        become_active: bool,
+    ) -> Self {
+        Self {
+            session_id,
+            jwt_api,
+            jwt_qconnect,
+            become_active,
+        }
+    }
+
+    pub fn session_id(&self) -> &str {
+        &self.session_id
+    }
+
+    pub const fn api_token(&self) -> &LanJwtToken {
+        &self.jwt_api
+    }
+
+    pub const fn qconnect_token(&self) -> &LanJwtToken {
+        &self.jwt_qconnect
+    }
+
+    pub const fn become_active(&self) -> bool {
+        self.become_active
+    }
+}
+
+impl Drop for HandoffCandidate {
+    fn drop(&mut self) {
+        self.session_id.zeroize();
+    }
+}

crates/qconnect-lan/src/projection.rs

@@ -0,0 +1,62 @@
+use std::sync::{Arc, RwLock};
+
+use crate::{ConnectInfo, DisplayInfo};
+
+#[derive(Debug, Clone)]
+struct ProjectionSnapshot {
+    display: DisplayInfo,
+    connect: ConnectInfo,
+}
+
+/// Short, synchronous projection read by the HTTP thread. Runtime adapters
+/// update it from their own event loops; handlers never block on async I/O.
+#[derive(Debug, Clone)]
+pub struct LanProjection {
+    inner: Arc<RwLock<ProjectionSnapshot>>,
+}
+
+impl LanProjection {
+    pub fn new(display: DisplayInfo, connect: ConnectInfo) -> Self {
+        Self {
+            inner: Arc::new(RwLock::new(ProjectionSnapshot { display, connect })),
+        }
+    }
+
+    pub fn display_info(&self) -> DisplayInfo {
+        self.inner
+            .read()
+            .expect("LAN projection lock poisoned")
+            .display
+            .clone()
+    }
+
+    pub fn connect_info(&self) -> ConnectInfo {
+        self.inner
+            .read()
+            .expect("LAN projection lock poisoned")
+            .connect
+            .clone()
+    }
+
+    pub fn update_display(&self, display: DisplayInfo) {
+        self.inner
+            .write()
+            .expect("LAN projection lock poisoned")
+            .display = display;
+    }
+
+    pub fn update_connect(&self, connect: ConnectInfo) {
+        self.inner
+            .write()
+            .expect("LAN projection lock poisoned")
+            .connect = connect;
+    }
+
+    pub fn set_current_session_id(&self, session_id: Option<String>) {
+        self.inner
+            .write()
+            .expect("LAN projection lock poisoned")
+            .connect
+            .current_session_id = session_id;
+    }
+}

crates/qconnect-lan/src/server.rs

@@ -0,0 +1,472 @@
+use std::collections::HashMap;
+use std::io::Read;
+use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::mpsc::{self, TrySendError};
+use std::sync::{Arc, Mutex};
+use std::thread::JoinHandle;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use socket2::{Domain, Protocol, Socket, Type};
+use tiny_http::{Header, Method, Request, Response, Server, StatusCode};
+
+use crate::admission::{admission_channel, AdmissionInbox, AdmissionSender, SubmitError};
+use crate::mdns::MdnsRegistration;
+use crate::validation::{parse_and_validate, EndpointPolicy, ValidationError, MAX_BODY_BYTES};
+use crate::LanProjection;
+
+pub const SERVICE_TYPE: &str = "_qobuz-connect._tcp.local.";
+pub const DEFAULT_SDK_VERSION: &str = "0.9.5";
+
+const POST_RATE_PER_MINUTE: f64 = 6.0;
+const POST_BURST: f64 = 2.0;
+const MAX_RATE_ORIGINS: usize = 256;
+const HTTP_WORKERS: usize = 2;
+pub const DEFAULT_LOCAL_READ_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[derive(Debug, thiserror::Error)]
+pub enum LanError {
+    #[error("qconnect-lan-bind")]
+    Bind,
+    #[error("qconnect-lan-address-discovery")]
+    AddressDiscovery,
+    #[error("qconnect-lan-no-addresses")]
+    NoLanAddresses,
+    #[error("qconnect-lan-mdns")]
+    Mdns,
+    #[error("qconnect-lan-config")]
+    InvalidConfig,
+}
+
+impl From<mdns_sd::Error> for LanError {
+    fn from(_: mdns_sd::Error) -> Self {
+        Self::Mdns
+    }
+}
+
+#[derive(Clone)]
+pub struct LanServiceConfig {
+    pub bind_addr: SocketAddr,
+    pub projection: LanProjection,
+    pub endpoint_policy: EndpointPolicy,
+    pub device_uuid: String,
+    pub sdk_version: String,
+    /// Bounds stalled local request reads. The production default is the
+    /// contract's five-second local read/parse budget.
+    pub read_timeout: Duration,
+    /// Tests and environments without multicast can exercise the exact HTTP
+    /// wire while keeping discovery disabled. Production adapters leave true.
+    pub advertise_mdns: bool,
+    /// Optional explicit address set, primarily for deterministic integration
+    /// tests. Production uses interface discovery and mdns-sd auto updates.
+    pub advertised_addresses: Option<Vec<IpAddr>>,
+}
+
+impl LanServiceConfig {
+    pub fn new(
+        projection: LanProjection,
+        endpoint_policy: EndpointPolicy,
+        device_uuid: impl Into<String>,
+    ) -> Self {
+        Self {
+            bind_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
+            projection,
+            endpoint_policy,
+            device_uuid: device_uuid.into(),
+            sdk_version: DEFAULT_SDK_VERSION.to_string(),
+            read_timeout: DEFAULT_LOCAL_READ_TIMEOUT,
+            advertise_mdns: true,
+            advertised_addresses: None,
+        }
+    }
+}
+
+pub struct LanService {
+    port: u16,
+    server: Arc<Server>,
+    admission: AdmissionSender,
+    accepting: Arc<AtomicBool>,
+    accept_thread: Option<JoinHandle<()>>,
+    worker_threads: Vec<JoinHandle<()>>,
+    mdns: Option<MdnsRegistration>,
+}
+
+impl LanService {
+    /// Bind first, spawn HTTP workers, then publish DNS-SD. On any discovery
+    /// failure the bound listener is torn down before the error returns.
+    pub fn start(config: LanServiceConfig) -> Result<(Self, AdmissionInbox), LanError> {
+        validate_config(&config)?;
+        let server = Arc::new(bind_http(config.bind_addr, config.read_timeout)?);
+        let port = server
+            .server_addr()
+            .to_ip()
+            .map(|address| address.port())
+            .ok_or(LanError::Bind)?;
+        let (admission, inbox) = admission_channel();
+        let accepting = Arc::new(AtomicBool::new(true));
+        let rate_limiter = Arc::new(Mutex::new(PostRateLimiter::default()));
+
+        // Count queued plus active work explicitly. This avoids a scheduling
+        // race where a worker exists but has not reached `recv` yet.
+        let (request_tx, request_rx) = mpsc::sync_channel::<Request>(HTTP_WORKERS);
+        let request_rx = Arc::new(Mutex::new(request_rx));
+        let in_flight = Arc::new(AtomicUsize::new(0));
+        let mut worker_threads = Vec::with_capacity(HTTP_WORKERS);
+        for _ in 0..HTTP_WORKERS {
+            let worker_rx = Arc::clone(&request_rx);
+            let worker_accepting = Arc::clone(&accepting);
+            let worker_admission = admission.clone();
+            let worker_projection = config.projection.clone();
+            let worker_policy = config.endpoint_policy.clone();
+            let worker_limiter = Arc::clone(&rate_limiter);
+            let worker_in_flight = Arc::clone(&in_flight);
+            worker_threads.push(std::thread::spawn(move || loop {
+                let request = worker_rx
+                    .lock()
+                    .expect("LAN request queue lock poisoned")
+                    .recv();
+                let request = match request {
+                    Ok(request) => request,
+                    Err(_) => break,
+                };
+                if worker_accepting.load(Ordering::Acquire) {
+                    handle_request(
+                        request,
+                        &worker_projection,
+                        &worker_policy,
+                        &worker_admission,
+                        &worker_limiter,
+                    );
+                } else {
+                    let _ = request.respond(error_response(StatusCode(503), "unavailable"));
+                }
+                worker_in_flight.fetch_sub(1, Ordering::AcqRel);
+            }));
+        }
+
+        let thread_server = Arc::clone(&server);
+        let thread_accepting = Arc::clone(&accepting);
+        let accept_in_flight = Arc::clone(&in_flight);
+        let accept_thread = std::thread::spawn(move || {
+            while thread_accepting.load(Ordering::Acquire) {
+                match thread_server.recv_timeout(Duration::from_millis(250)) {
+                    Ok(Some(request)) => {
+                        let admitted = accept_in_flight
+                            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
+                                (current < HTTP_WORKERS).then_some(current + 1)
+                            })
+                            .is_ok();
+                        if !admitted {
+                            let _ = request.respond(error_response(StatusCode(429), "busy"));
+                            continue;
+                        }
+                        match request_tx.try_send(request) {
+                            Ok(()) => {}
+                            Err(TrySendError::Full(request)) => {
+                                accept_in_flight.fetch_sub(1, Ordering::AcqRel);
+                                let _ = request.respond(error_response(StatusCode(429), "busy"));
+                            }
+                            Err(TrySendError::Disconnected(request)) => {
+                                accept_in_flight.fetch_sub(1, Ordering::AcqRel);
+                                let _ =
+                                    request.respond(error_response(StatusCode(503), "unavailable"));
+                                break;
+                            }
+                        }
+                    }
+                    Ok(None) => {}
+                    Err(_) if !thread_accepting.load(Ordering::Acquire) => break,
+                    // Malformed or timed-out client connections are surfaced
+                    // by tiny_http through this queue too; they must not kill
+                    // discovery for every other controller.
+                    Err(_) => {}
+                }
+            }
+        });
+
+        let mdns = if config.advertise_mdns {
+            match MdnsRegistration::register(
+                &config.device_uuid,
+                port,
+                &config.sdk_version,
+                config.advertised_addresses,
+            ) {
+                Ok(mdns) => Some(mdns),
+                Err(error) => {
+                    admission.close();
+                    accepting.store(false, Ordering::Release);
+                    server.unblock();
+                    let _ = accept_thread.join();
+                    for worker in worker_threads {
+                        let _ = worker.join();
+                    }
+                    return Err(error);
+                }
+            }
+        } else {
+            None
+        };
+
+        Ok((
+            Self {
+                port,
+                server,
+                admission,
+                accepting,
+                accept_thread: Some(accept_thread),
+                worker_threads,
+                mdns,
+            },
+            inbox,
+        ))
+    }
+
+    pub const fn port(&self) -> u16 {
+        self.port
+    }
+
+    pub fn shutdown(&mut self) {
+        self.admission.close();
+        if let Some(mdns) = self.mdns.take() {
+            mdns.shutdown();
+        }
+        self.accepting.store(false, Ordering::Release);
+        self.server.unblock();
+        if let Some(thread) = self.accept_thread.take() {
+            let _ = thread.join();
+        }
+        for thread in self.worker_threads.drain(..) {
+            let _ = thread.join();
+        }
+    }
+}
+
+impl Drop for LanService {
+    fn drop(&mut self) {
+        self.shutdown();
+    }
+}
+
+fn validate_config(config: &LanServiceConfig) -> Result<(), LanError> {
+    let display = config.projection.display_info();
+    let connect = config.projection.connect_info();
+    let required = [
+        display.friendly_name.as_str(),
+        display.serial_number.as_str(),
+        display.brand_display_name.as_str(),
+        display.model_display_name.as_str(),
+        display.software_version.as_str(),
+        connect.app_id.as_str(),
+        config.device_uuid.as_str(),
+        config.sdk_version.as_str(),
+    ];
+    if config.bind_addr.is_ipv6()
+        || config.read_timeout.is_zero()
+        || required.iter().any(|value| value.trim().is_empty())
+        || display.serial_number != config.device_uuid
+    {
+        return Err(LanError::InvalidConfig);
+    }
+    Ok(())
+}
+
+fn bind_http(bind_addr: SocketAddr, read_timeout: Duration) -> Result<Server, LanError> {
+    let domain = if bind_addr.is_ipv4() {
+        Domain::IPV4
+    } else {
+        Domain::IPV6
+    };
+    let socket =
+        Socket::new(domain, Type::STREAM, Some(Protocol::TCP)).map_err(|_| LanError::Bind)?;
+    socket.set_reuse_address(true).map_err(|_| LanError::Bind)?;
+    if bind_addr.is_ipv6() {
+        socket.set_only_v6(true).map_err(|_| LanError::Bind)?;
+    }
+    // Linux propagates SO_RCVTIMEO from a listening socket to accepted TCP
+    // sockets. That bounds stalled request reads inside tiny_http.
+    socket
+        .set_read_timeout(Some(read_timeout))
+        .map_err(|_| LanError::Bind)?;
+    socket.bind(&bind_addr.into()).map_err(|_| LanError::Bind)?;
+    socket.listen(128).map_err(|_| LanError::Bind)?;
+    let listener: TcpListener = socket.into();
+    Server::from_listener(listener, None).map_err(|_| LanError::Bind)
+}
+
+fn handle_request(
+    mut request: Request,
+    projection: &LanProjection,
+    policy: &EndpointPolicy,
+    admission: &AdmissionSender,
+    rate_limiter: &Arc<Mutex<PostRateLimiter>>,
+) {
+    let method = request.method().clone();
+    let path = request.url().to_string();
+    let response = match path.as_str() {
+        "/get-display-info" if method == Method::Get => {
+            json_response(StatusCode(200), &projection.display_info())
+        }
+        "/get-connect-info" if method == Method::Get => {
+            json_response(StatusCode(200), &projection.connect_info())
+        }
+        "/connect-to-qconnect" if method == Method::Post => {
+            handle_connect(&mut request, policy, admission, rate_limiter)
+        }
+        "/get-display-info" | "/get-connect-info" | "/connect-to-qconnect" => {
+            error_response(StatusCode(405), "method_not_allowed")
+        }
+        _ => error_response(StatusCode(404), "not_found"),
+    };
+    let _ = request.respond(response);
+}
+
+fn handle_connect(
+    request: &mut Request,
+    policy: &EndpointPolicy,
+    admission: &AdmissionSender,
+    rate_limiter: &Arc<Mutex<PostRateLimiter>>,
+) -> Response<std::io::Cursor<Vec<u8>>> {
+    if !has_json_content_type(request) {
+        return error_response(StatusCode(415), "json_required");
+    }
+    if declared_body_too_large(request) {
+        return error_response(StatusCode(413), "body_too_large");
+    }
+    let remote_ip = request.remote_addr().map(SocketAddr::ip);
+    if !rate_limiter
+        .lock()
+        .expect("LAN rate-limit lock poisoned")
+        .allow(remote_ip, Instant::now())
+    {
+        return error_response(StatusCode(429), "rate_limited");
+    }
+
+    let mut body = Vec::new();
+    if request
+        .as_reader()
+        .take((MAX_BODY_BYTES + 1) as u64)
+        .read_to_end(&mut body)
+        .is_err()
+    {
+        return error_response(StatusCode(400), "body_read_failed");
+    }
+    if body.len() > MAX_BODY_BYTES {
+        return error_response(StatusCode(413), "body_too_large");
+    }
+
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+        .min(i64::MAX as u64) as i64;
+    let candidate = match parse_and_validate(&body, policy, now) {
+        Ok(candidate) => candidate,
+        Err(ValidationError::BecomeActiveRequired) => {
+            return error_response(StatusCode(422), "active_required")
+        }
+        Err(_) => return error_response(StatusCode(400), "invalid_handoff"),
+    };
+
+    match admission.submit(candidate) {
+        Ok(()) => json_response(StatusCode(200), &serde_json::json!({})),
+        Err(SubmitError::Closed) => error_response(StatusCode(503), "unavailable"),
+    }
+}
+
+fn has_json_content_type(request: &Request) -> bool {
+    request.headers().iter().any(|header| {
+        header.field.equiv("Content-Type")
+            && header
+                .value
+                .as_str()
+                .split(';')
+                .next()
+                .is_some_and(|value| value.trim().eq_ignore_ascii_case("application/json"))
+    })
+}
+
+fn declared_body_too_large(request: &Request) -> bool {
+    request.headers().iter().any(|header| {
+        header.field.equiv("Content-Length")
+            && header
+                .value
+                .as_str()
+                .trim()
+                .parse::<usize>()
+                .is_ok_and(|length| length > MAX_BODY_BYTES)
+    })
+}
+
+fn json_response<T: serde::Serialize>(
+    status: StatusCode,
+    value: &T,
+) -> Response<std::io::Cursor<Vec<u8>>> {
+    let body = serde_json::to_vec(value).unwrap_or_else(|_| b"{}".to_vec());
+    Response::from_data(body)
+        .with_status_code(status)
+        .with_header(json_content_type())
+}
+
+fn error_response(status: StatusCode, code: &'static str) -> Response<std::io::Cursor<Vec<u8>>> {
+    json_response(status, &serde_json::json!({ "error": code }))
+}
+
+fn json_content_type() -> Header {
+    Header::from_bytes("Content-Type", "application/json").expect("static JSON content-type header")
+}
+
+#[derive(Debug, Clone, Copy)]
+struct RateBucket {
+    tokens: f64,
+    updated_at: Instant,
+}
+
+#[derive(Default)]
+struct PostRateLimiter {
+    buckets: HashMap<Option<IpAddr>, RateBucket>,
+}
+
+impl PostRateLimiter {
+    fn allow(&mut self, origin: Option<IpAddr>, now: Instant) -> bool {
+        if self.buckets.len() >= MAX_RATE_ORIGINS && !self.buckets.contains_key(&origin) {
+            if let Some(oldest) = self
+                .buckets
+                .iter()
+                .min_by_key(|(_, bucket)| bucket.updated_at)
+                .map(|(origin, _)| *origin)
+            {
+                self.buckets.remove(&oldest);
+            }
+        }
+        let bucket = self.buckets.entry(origin).or_insert(RateBucket {
+            tokens: POST_BURST,
+            updated_at: now,
+        });
+        let elapsed = now
+            .saturating_duration_since(bucket.updated_at)
+            .as_secs_f64();
+        bucket.tokens = (bucket.tokens + elapsed * POST_RATE_PER_MINUTE / 60.0).min(POST_BURST);
+        bucket.updated_at = now;
+        if bucket.tokens < 1.0 {
+            return false;
+        }
+        bucket.tokens -= 1.0;
+        true
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn post_rate_limit_allows_burst_then_refills() {
+        let mut limiter = PostRateLimiter::default();
+        let now = Instant::now();
+        let ip = Some(IpAddr::V4(Ipv4Addr::LOCALHOST));
+        assert!(limiter.allow(ip, now));
+        assert!(limiter.allow(ip, now));
+        assert!(!limiter.allow(ip, now));
+        assert!(limiter.allow(ip, now + Duration::from_secs(10)));
+    }
+}

crates/qconnect-lan/src/validation.rs

@@ -0,0 +1,254 @@
+use std::collections::HashSet;
+
+use url::Url;
+
+use crate::model::{HandoffCandidate, LanJwtToken, RawHandoffRequest, RawLanJwtToken};
+
+pub const MAX_BODY_BYTES: usize = 64 * 1024;
+pub const MAX_SESSION_ID_BYTES: usize = 256;
+pub const MAX_ENDPOINT_BYTES: usize = 2 * 1024;
+pub const MAX_JWT_BYTES: usize = 16 * 1024;
+pub const MIN_TOKEN_LIFETIME_SECS: i64 = 60;
+
+#[derive(Debug, Clone, Default)]
+pub struct EndpointPolicy {
+    api_hosts: HashSet<String>,
+    qconnect_hosts: HashSet<String>,
+}
+
+impl EndpointPolicy {
+    pub fn new(
+        api_hosts: impl IntoIterator<Item = impl Into<String>>,
+        qconnect_hosts: impl IntoIterator<Item = impl Into<String>>,
+    ) -> Self {
+        Self {
+            api_hosts: normalize_hosts(api_hosts),
+            qconnect_hosts: normalize_hosts(qconnect_hosts),
+        }
+    }
+
+    pub fn from_trusted_endpoints(
+        api_endpoint: &str,
+        qconnect_endpoint: &str,
+    ) -> Result<Self, ValidationError> {
+        let api = endpoint_host(api_endpoint, "https")?;
+        let qconnect = endpoint_host(qconnect_endpoint, "wss")?;
+        Ok(Self::new([api], [qconnect]))
+    }
+
+    fn allows_api(&self, host: &str) -> bool {
+        self.api_hosts.contains(&host.to_ascii_lowercase())
+    }
+
+    fn allows_qconnect(&self, host: &str) -> bool {
+        self.qconnect_hosts.contains(&host.to_ascii_lowercase())
+    }
+}
+
+fn normalize_hosts(hosts: impl IntoIterator<Item = impl Into<String>>) -> HashSet<String> {
+    hosts
+        .into_iter()
+        .map(Into::into)
+        .map(|host: String| host.trim().trim_end_matches('.').to_ascii_lowercase())
+        .filter(|host| !host.is_empty())
+        .collect()
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ValidationError {
+    InvalidShape,
+    BecomeActiveRequired,
+    SessionIdInvalid,
+    TokenInvalid,
+    TokenExpiring,
+    EndpointInvalid,
+    EndpointNotAllowed,
+}
+
+pub(crate) fn parse_and_validate(
+    body: &[u8],
+    policy: &EndpointPolicy,
+    now_unix_secs: i64,
+) -> Result<HandoffCandidate, ValidationError> {
+    let mut raw: RawHandoffRequest =
+        serde_json::from_slice(body).map_err(|_| ValidationError::InvalidShape)?;
+
+    if !raw.become_active {
+        return Err(ValidationError::BecomeActiveRequired);
+    }
+    if raw.session_id.is_empty()
+        || raw.session_id.len() > MAX_SESSION_ID_BYTES
+        || raw.session_id.trim() != raw.session_id
+    {
+        return Err(ValidationError::SessionIdInvalid);
+    }
+
+    let api = validate_token(&mut raw.jwt_api, "https", policy, now_unix_secs, true)?;
+    let qconnect = validate_token(&mut raw.jwt_qconnect, "wss", policy, now_unix_secs, false)?;
+    let session_id = std::mem::take(&mut raw.session_id);
+    Ok(HandoffCandidate::new(session_id, api, qconnect, true))
+}
+
+fn validate_token(
+    raw: &mut RawLanJwtToken,
+    required_scheme: &str,
+    policy: &EndpointPolicy,
+    now_unix_secs: i64,
+    api: bool,
+) -> Result<LanJwtToken, ValidationError> {
+    if raw.endpoint.is_empty()
+        || raw.endpoint.len() > MAX_ENDPOINT_BYTES
+        || raw.jwt.is_empty()
+        || raw.jwt.len() > MAX_JWT_BYTES
+    {
+        return Err(ValidationError::TokenInvalid);
+    }
+    if raw.exp < now_unix_secs.saturating_add(MIN_TOKEN_LIFETIME_SECS) {
+        return Err(ValidationError::TokenExpiring);
+    }
+
+    let url = Url::parse(&raw.endpoint).map_err(|_| ValidationError::EndpointInvalid)?;
+    if url.scheme() != required_scheme
+        || !url.username().is_empty()
+        || url.password().is_some()
+        || url.fragment().is_some()
+    {
+        return Err(ValidationError::EndpointInvalid);
+    }
+    let host = url
+        .host_str()
+        .ok_or(ValidationError::EndpointInvalid)?
+        .trim_end_matches('.')
+        .to_ascii_lowercase();
+    let allowed = if api {
+        policy.allows_api(&host)
+    } else {
+        policy.allows_qconnect(&host)
+    };
+    if !allowed {
+        return Err(ValidationError::EndpointNotAllowed);
+    }
+
+    Ok(LanJwtToken::new(
+        std::mem::take(&mut raw.endpoint),
+        raw.exp,
+        std::mem::take(&mut raw.jwt),
+    ))
+}
+
+fn endpoint_host(endpoint: &str, required_scheme: &str) -> Result<String, ValidationError> {
+    let url = Url::parse(endpoint).map_err(|_| ValidationError::EndpointInvalid)?;
+    if url.scheme() != required_scheme
+        || !url.username().is_empty()
+        || url.password().is_some()
+        || url.fragment().is_some()
+    {
+        return Err(ValidationError::EndpointInvalid);
+    }
+    url.host_str()
+        .map(|host| host.trim_end_matches('.').to_ascii_lowercase())
+        .filter(|host| !host.is_empty())
+        .ok_or(ValidationError::EndpointInvalid)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const NOW: i64 = 1_780_000_000;
+
+    fn policy() -> EndpointPolicy {
+        EndpointPolicy::new(["api.qobuz.test"], ["qws.qobuz.test"])
+    }
+
+    fn body(api_endpoint: &str, qws_endpoint: &str, exp: i64, active: bool) -> Vec<u8> {
+        serde_json::to_vec(&serde_json::json!({
+            "session_id": "controller-session",
+            "jwt_api": { "endpoint": api_endpoint, "exp": exp, "jwt": "api-secret" },
+            "jwt_qconnect": { "endpoint": qws_endpoint, "exp": exp, "jwt": "qws-secret" },
+            "become_active": active
+        }))
+        .unwrap()
+    }
+
+    fn validation_error(body: &[u8]) -> ValidationError {
+        match parse_and_validate(body, &policy(), NOW) {
+            Ok(_) => panic!("handoff unexpectedly validated"),
+            Err(error) => error,
+        }
+    }
+
+    #[test]
+    fn validates_exact_official_shape() {
+        let candidate = parse_and_validate(
+            &body(
+                "https://api.qobuz.test/v1",
+                "wss://qws.qobuz.test/ws",
+                NOW + 600,
+                true,
+            ),
+            &policy(),
+            NOW,
+        )
+        .unwrap();
+
+        assert_eq!(candidate.session_id(), "controller-session");
+        assert_eq!(candidate.api_token().jwt(), "api-secret");
+        assert_eq!(candidate.qconnect_token().jwt(), "qws-secret");
+    }
+
+    #[test]
+    fn rejects_aliases_expiry_and_untrusted_hosts() {
+        let alias = serde_json::json!({
+            "session_id": "controller-session",
+            "jwt_api": { "endpoint": "https://api.qobuz.test", "exp": NOW + 600, "jwt": "a" },
+            "jwt_qws": { "endpoint": "wss://qws.qobuz.test", "exp": NOW + 600, "jwt": "q" },
+            "become_active": true
+        });
+        assert_eq!(
+            validation_error(&serde_json::to_vec(&alias).unwrap()),
+            ValidationError::InvalidShape
+        );
+        assert_eq!(
+            validation_error(&body(
+                "https://api.qobuz.test",
+                "wss://qws.qobuz.test",
+                NOW + 59,
+                true,
+            )),
+            ValidationError::TokenExpiring
+        );
+        assert_eq!(
+            validation_error(&body(
+                "https://evil.test",
+                "wss://qws.qobuz.test",
+                NOW + 600,
+                true,
+            )),
+            ValidationError::EndpointNotAllowed
+        );
+    }
+
+    #[test]
+    fn rejects_wrong_schemes_userinfo_and_inactive_handoff() {
+        for api in [
+            "http://api.qobuz.test",
+            "https://user@api.qobuz.test",
+            "https://api.qobuz.test/#fragment",
+        ] {
+            assert_eq!(
+                validation_error(&body(api, "wss://qws.qobuz.test", NOW + 600, true)),
+                ValidationError::EndpointInvalid
+            );
+        }
+        assert_eq!(
+            validation_error(&body(
+                "https://api.qobuz.test",
+                "wss://qws.qobuz.test",
+                NOW + 600,
+                false,
+            )),
+            ValidationError::BecomeActiveRequired
+        );
+    }
+}

crates/qconnect-lan/tests/official_wire.rs

@@ -0,0 +1,333 @@
+use std::io::{Read, Write};
+use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use qconnect_lan::{
+    ConnectInfo, DeviceType, DisplayInfo, EndpointPolicy, LanError, LanProjection, LanService,
+    LanServiceConfig, MaxAudioQuality, MAX_BODY_BYTES,
+};
+use reqwest::StatusCode;
+
+const DEVICE_UUID: &str = "550e8400-e29b-41d4-a716-446655440000";
+
+fn start_service() -> (
+    LanService,
+    qconnect_lan::AdmissionInbox,
+    String,
+    LanProjection,
+) {
+    start_service_with_timeout(Duration::from_secs(5))
+}
+
+fn start_service_with_timeout(
+    read_timeout: Duration,
+) -> (
+    LanService,
+    qconnect_lan::AdmissionInbox,
+    String,
+    LanProjection,
+) {
+    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+    let projection = LanProjection::new(
+        DisplayInfo {
+            friendly_name: "Kitchen".to_string(),
+            serial_number: DEVICE_UUID.to_string(),
+            brand_display_name: "QBZ".to_string(),
+            model_display_name: "QBZ Daemon".to_string(),
+            max_audio_quality: MaxAudioQuality::UpToHires192,
+            device_type: DeviceType::Streamer,
+            software_version: "2.1.0".to_string(),
+        },
+        ConnectInfo {
+            app_id: "trusted-app-id".to_string(),
+            current_session_id: None,
+        },
+    );
+    let mut config = LanServiceConfig::new(
+        projection.clone(),
+        EndpointPolicy::new(["api.qobuz.test"], ["qws.qobuz.test"]),
+        DEVICE_UUID,
+    );
+    config.bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0);
+    config.read_timeout = read_timeout;
+    config.advertise_mdns = false;
+    let (service, inbox) = LanService::start(config).unwrap();
+    let base = format!("http://127.0.0.1:{}", service.port());
+    (service, inbox, base, projection)
+}
+
+fn test_config(bind_addr: SocketAddr) -> LanServiceConfig {
+    let projection = LanProjection::new(
+        DisplayInfo {
+            friendly_name: "Kitchen".to_string(),
+            serial_number: DEVICE_UUID.to_string(),
+            brand_display_name: "QBZ".to_string(),
+            model_display_name: "QBZ Daemon".to_string(),
+            max_audio_quality: MaxAudioQuality::UpToHires192,
+            device_type: DeviceType::Streamer,
+            software_version: "2.1.0".to_string(),
+        },
+        ConnectInfo {
+            app_id: "trusted-app-id".to_string(),
+            current_session_id: None,
+        },
+    );
+    let mut config = LanServiceConfig::new(
+        projection,
+        EndpointPolicy::new(["api.qobuz.test"], ["qws.qobuz.test"]),
+        DEVICE_UUID,
+    );
+    config.bind_addr = bind_addr;
+    config
+}
+
+fn stalled_post(port: u16) -> TcpStream {
+    let mut stream = TcpStream::connect((Ipv4Addr::LOCALHOST, port)).unwrap();
+    stream
+        .set_read_timeout(Some(Duration::from_secs(2)))
+        .unwrap();
+    write!(
+        stream,
+        "POST /connect-to-qconnect HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: 2048\r\nConnection: close\r\n\r\n{{"
+    )
+    .unwrap();
+    stream.flush().unwrap();
+    stream
+}
+
+fn handoff(session_id: &str, become_active: bool) -> serde_json::Value {
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_secs() as i64;
+    serde_json::json!({
+        "session_id": session_id,
+        "jwt_api": {
+            "endpoint": "https://api.qobuz.test/api.json/0.2",
+            "exp": now + 600,
+            "jwt": "api-secret"
+        },
+        "jwt_qconnect": {
+            "endpoint": "wss://qws.qobuz.test/ws",
+            "exp": now + 600,
+            "jwt": "qws-secret"
+        },
+        "become_active": become_active
+    })
+}
+
+#[tokio::test]
+async fn serves_official_get_union_and_admits_exact_post_shape() {
+    let (mut service, inbox, base, _) = start_service();
+    let client = reqwest::Client::new();
+
+    let display = client
+        .get(format!("{base}/get-display-info"))
+        .header("Connection", "close")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(display.status(), StatusCode::OK);
+    assert_eq!(
+        display.headers()["content-type"].to_str().unwrap(),
+        "application/json"
+    );
+    assert_eq!(
+        display.json::<serde_json::Value>().await.unwrap(),
+        serde_json::json!({
+            "friendly_name": "Kitchen",
+            "serial_number": DEVICE_UUID,
+            "brand_display_name": "QBZ",
+            "model_display_name": "QBZ Daemon",
+            "max_audio_quality": "UP_TO_HIRES_192",
+            "type": "Streamer",
+            "software_version": "2.1.0"
+        })
+    );
+
+    let connect = client
+        .get(format!("{base}/get-connect-info"))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(connect.status(), StatusCode::OK);
+    assert_eq!(
+        connect.json::<serde_json::Value>().await.unwrap(),
+        serde_json::json!({ "app_id": "trusted-app-id", "current_session_id": null })
+    );
+
+    let accepted = client
+        .post(format!("{base}/connect-to-qconnect"))
+        .json(&handoff("controller-session", true))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(accepted.status(), StatusCode::OK);
+    assert_eq!(
+        accepted.json::<serde_json::Value>().await.unwrap(),
+        serde_json::json!({})
+    );
+    let candidate = inbox.take_timeout(Duration::from_secs(1)).unwrap();
+    assert_eq!(candidate.session_id(), "controller-session");
+    assert_eq!(
+        candidate.api_token().endpoint(),
+        "https://api.qobuz.test/api.json/0.2"
+    );
+    assert_eq!(
+        candidate.qconnect_token().endpoint(),
+        "wss://qws.qobuz.test/ws"
+    );
+
+    service.shutdown();
+    service.shutdown();
+    assert!(inbox.is_closed());
+}
+
+#[tokio::test]
+async fn pending_post_is_latest_wins_and_rate_limit_is_bounded() {
+    let (mut service, inbox, base, _) = start_service();
+    let client = reqwest::Client::new();
+
+    for session in ["first", "second"] {
+        let response = client
+            .post(format!("{base}/connect-to-qconnect"))
+            .json(&handoff(session, true))
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::OK);
+    }
+    assert_eq!(inbox.try_take().unwrap().session_id(), "second");
+
+    let limited = client
+        .post(format!("{base}/connect-to-qconnect"))
+        .json(&handoff("third", true))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(limited.status(), StatusCode::TOO_MANY_REQUESTS);
+    assert!(!limited.text().await.unwrap().contains("api-secret"));
+    service.shutdown();
+}
+
+#[tokio::test]
+async fn rejects_noncanonical_http_and_oversized_or_inactive_handoffs() {
+    let (mut service, _inbox, base, _) = start_service();
+    let client = reqwest::Client::new();
+
+    assert_eq!(
+        client
+            .get(format!("{base}/get-display-info?alias=1"))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        StatusCode::NOT_FOUND
+    );
+    assert_eq!(
+        client
+            .post(format!("{base}/get-connect-info"))
+            .json(&serde_json::json!({}))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        StatusCode::METHOD_NOT_ALLOWED
+    );
+    assert_eq!(
+        client
+            .post(format!("{base}/connect-to-qconnect"))
+            .body("{}")
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        StatusCode::UNSUPPORTED_MEDIA_TYPE
+    );
+    assert_eq!(
+        client
+            .post(format!("{base}/connect-to-qconnect"))
+            .json(&handoff("inactive", false))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        StatusCode::UNPROCESSABLE_ENTITY
+    );
+
+    let (mut oversized_service, _, oversized_base, _) = start_service();
+    let oversized = client
+        .post(format!("{oversized_base}/connect-to-qconnect"))
+        .header("Content-Type", "application/json")
+        .body(vec![b'x'; MAX_BODY_BYTES + 1])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(oversized.status(), StatusCode::PAYLOAD_TOO_LARGE);
+
+    service.shutdown();
+    oversized_service.shutdown();
+}
+
+#[tokio::test]
+async fn connect_projection_updates_without_restarting_listener() {
+    let (mut service, _, base, projection) = start_service();
+    projection.set_current_session_id(Some("active-session".to_string()));
+
+    let value = reqwest::get(format!("{base}/get-connect-info"))
+        .await
+        .unwrap()
+        .json::<serde_json::Value>()
+        .await
+        .unwrap();
+    assert_eq!(value["current_session_id"], "active-session");
+    service.shutdown();
+}
+
+#[tokio::test]
+async fn stalled_bodies_time_out_and_concurrency_stays_bounded() {
+    let (mut service, _, base, _) = start_service_with_timeout(Duration::from_millis(500));
+    let mut first = stalled_post(service.port());
+    let second = stalled_post(service.port());
+    let deadline = std::time::Instant::now() + Duration::from_millis(250);
+    loop {
+        let response = reqwest::get(format!("{base}/get-display-info"))
+            .await
+            .unwrap();
+        if response.status() == StatusCode::TOO_MANY_REQUESTS {
+            break;
+        }
+        assert_eq!(response.status(), StatusCode::OK);
+        assert!(
+            std::time::Instant::now() < deadline,
+            "workers never saturated"
+        );
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+
+    let mut response = String::new();
+    first.read_to_string(&mut response).unwrap();
+    assert!(response.starts_with("HTTP/1.1 400"), "{response}");
+
+    drop(second);
+    service.shutdown();
+}
+
+#[test]
+fn mdns_failure_after_bind_releases_listener_and_never_advertises_loopback() {
+    let reservation = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
+    let bind_addr = reservation.local_addr().unwrap();
+    drop(reservation);
+
+    let mut config = test_config(bind_addr);
+    config.advertise_mdns = true;
+    config.advertised_addresses = Some(vec![IpAddr::V4(Ipv4Addr::LOCALHOST)]);
+    let error = match LanService::start(config) {
+        Ok(_) => panic!("loopback mDNS address unexpectedly accepted"),
+        Err(error) => error,
+    };
+    assert!(matches!(error, LanError::NoLanAddresses));
+
+    std::thread::sleep(Duration::from_millis(25));
+    assert!(TcpStream::connect_timeout(&bind_addr, Duration::from_millis(100)).is_err());
+}