diff --git a/src/client.rs b/src/client.rs index 7637249ed..f6f1ff2d3 100644 --- a/src/client.rs +++ b/src/client.rs @@ -977,8 +977,14 @@ impl Client { let addr = AddrMangle::decode(&rr.socket_addr_v6); if addr.port() > 0 { if s.connect(addr).await.is_ok() { - connect_futures - .push(udp_nat_connect(s, "IPv6", CONNECT_TIMEOUT).boxed()); + connect_futures.push( + async move { + let (conn, kcp, typ) = + udp_nat_connect(s, "IPv6", CONNECT_TIMEOUT).await?; + Ok((conn, kcp, typ, true)) + } + .boxed(), + ); } } } @@ -1046,7 +1052,12 @@ impl Client { connect_futures.push( async move { let conn = fut.await?; - Ok((conn, None, if use_ws() { "WebSocket" } else { "Relay" })) + Ok(( + conn, + None, + if use_ws() { "WebSocket" } else { "Relay" }, + false, + )) } .boxed(), ); @@ -1060,7 +1071,13 @@ impl Client { let mut raced = webrtc; let webrtc_fut = async move { raced.wait_connected(CONNECT_TIMEOUT).await?; - Ok((Stream::WebRTC(raced), None, "WebRTC")) + // Resolve relayed-ness here, not from the label: WebRTC is only a + // P2P path when ICE nominated a non-TURN pair, and the race has to + // know which it got. Committing a TURN pair as if it were direct + // cancels a genuine direct attempt still in flight — the same + // inversion the preference window exists to prevent, one level up. + let relayed = raced.is_relayed().await.unwrap_or(true); + Ok((Stream::WebRTC(raced), None, "WebRTC", !relayed)) } .boxed(); if interface.is_policy_relay() { @@ -1079,7 +1096,7 @@ impl Client { webrtc_fut, connect_futures, Self::WEBRTC_PREFER_WINDOW_MS, - |result| is_direct_transport(result.2), + |result| result.3, ) .await } @@ -1092,7 +1109,7 @@ impl Client { } // The ? / secure_connection failures below return early; webrtc_guard stays // in scope and closes the offerer on any such exit (loss, error, cancellation). - let (mut conn, kcp, mut typ) = race_result?; + let (mut conn, kcp, mut typ, direct) = race_result?; feedback = rr.feedback; log::info!("{:?} used to establish {typ} connection", start.elapsed()); let pk = match Self::secure_connection( @@ -1146,18 +1163,8 @@ impl Client { } Err(e) => return Err(e), }; - // Compute the direct/relayed flag (an await) BEFORE disarming the guard, so - // a cancellation of this future during webrtc_relayed() still closes the pc - // via the guard's drop. Matches connect()'s ordering. - let direct = match typ { - "IPv6" => true, - // WebRTC through a TURN server is relayed traffic; report it as such. - // An unknown answer (no selected pair yet, or the pc closed under a - // concurrent teardown) must not be read as "direct": claiming a P2P - // path needs evidence of one. - "WebRTC" => !conn.webrtc_relayed().await.unwrap_or(true), - _ => false, - }; + // `direct` came from the winning future, which resolved it while the pc was + // definitely alive — the race needed it to pick a winner at all. // Secured and WebRTC won: disarm so the returned conn keeps the pc alive. if typ == "WebRTC" { if let Some(guard) = webrtc_guard.take() { @@ -2598,13 +2605,15 @@ pub struct LoginConfigHandler { // reconnect before the real reboot disconnect. restart_remote_device_at: Option, pub force_relay: bool, - // The policy component of force_relay: the user's relay choice (option or explicit - // request) plus proxy, WITHOUT the WebSocket-transport component. ws kills classic - // TCP/UDP punching (force_relay stays set for those paths) but says nothing about - // ICE, so WebRTC decisions - offer ICE policy, prefer-P2P racing - key off this - // instead: relay-by-policy must stay Relay-only ICE, relay-by-transport may go - // direct over full ICE. + // The policy component of force_relay: peer_relay plus proxy, WITHOUT the WebSocket + // transport. ws kills classic TCP/UDP punching (force_relay stays set for those paths) + // but says nothing about ICE, so the WebRTC decisions - offer ICE policy, prefer-P2P + // racing - key off this instead: relay-by-policy must stay Relay-only ICE, while + // relay-by-transport may still go direct over full ICE. pub policy_relay: bool, + // The peer-scoped component: this peer's saved force-always-relay option, or an explicit + // relay request for it. The only part that may be written back into the peer's config. + pub peer_relay: bool, pub direct: Option, pub received: bool, switch_uuid: Option, @@ -2717,10 +2726,16 @@ impl LoginConfigHandler { self.session_id = sid; self.supported_encoding = Default::default(); self.clear_restarting_remote_device(); - self.policy_relay = + // Three scopes, and only the first belongs in the peer's saved config: what was decided + // about THIS PEER (its saved option, or an explicit relay request such as an `/r` id or + // a retry-via-relay), versus what is true of this client right now (a proxy, WebSocket). + // Persisting either of the latter turns a transient local setup into a permanent property + // of the peer — and relay-by-policy means Relay-only ICE, so WebRTC never goes direct to + // it again. + self.peer_relay = config::option2bool("force-always-relay", &self.get_option("force-always-relay")) - || force_relay - || Config::is_proxy(); + || force_relay; + self.policy_relay = self.peer_relay || Config::is_proxy(); self.force_relay = self.policy_relay || use_ws(); if let Some((real_id, server, key)) = &self.other_server { let other_server_key = self.get_option("other-server-key"); @@ -3438,12 +3453,9 @@ impl LoginConfigHandler { .insert("other-server-key".to_owned(), c.clone()); } } - // policy_relay, not force_relay: this writes the user's relay CHOICE back into the peer's - // saved config, and force_relay also carries the WebSocket transport, which is a property - // of this client's current setup rather than of the peer. Persisting that turned one ws - // session into a permanent relay-by-policy peer — and since relay-by-policy means - // Relay-only ICE, WebRTC could never go direct to it again. - if self.policy_relay { + // peer_relay only — see `initialize`: neither the proxy nor the WebSocket transport is a + // fact about this peer, and writing one here makes it permanent. + if self.peer_relay { config .options .insert("force-always-relay".to_owned(), "Y".to_owned()); diff --git a/src/rendezvous_mediator.rs b/src/rendezvous_mediator.rs index bec3b735d..233fb485b 100644 --- a/src/rendezvous_mediator.rs +++ b/src/rendezvous_mediator.rs @@ -53,8 +53,12 @@ lazy_static::lazy_static! { static ref SOLVING_PK_MISMATCH: Mutex = Default::default(); static ref LAST_MSG: Mutex<(SocketAddr, Instant)> = Mutex::new((SocketAddr::new([0; 4].into(), 0), Instant::now())); static ref LAST_RELAY_MSG: Mutex<(SocketAddr, Instant)> = Mutex::new((SocketAddr::new([0; 4].into(), 0), Instant::now())); - static ref WEBRTC_ICE_TXS: Mutex>> = Default::default(); + static ref WEBRTC_ICE_TXS: Mutex>> = Default::default(); } +/// Remote ICE candidates buffered per session while the answerer applies them. Mirrors the +/// controller's own cap: gathering yields host, then srflx, then relay, so a real peer sends +/// well under this, and anything past it is someone deciding how much memory this process holds. +const MAX_PENDING_REMOTE_ICE: usize = 64; // The rendezvous ICE route is reachable without a prior punch and the peer decides how many // candidates it sends, so these sites would let someone else set how much this machine writes to // its log file. One line a minute each, carrying the suppressed count. @@ -63,6 +67,8 @@ static UNKNOWN_ICE_SESSION_LOG: hbb_common::log_throttle::LogThrottle = hbb_common::log_throttle::LogThrottle::new(ICE_LOG_INTERVAL); static REJECTED_REMOTE_ICE_LOG: hbb_common::log_throttle::LogThrottle = hbb_common::log_throttle::LogThrottle::new(ICE_LOG_INTERVAL); +static FULL_ICE_QUEUE_LOG: hbb_common::log_throttle::LogThrottle = + hbb_common::log_throttle::LogThrottle::new(ICE_LOG_INTERVAL); static SHOULD_EXIT: AtomicBool = AtomicBool::new(false); static MANUAL_RESTARTED: AtomicBool = AtomicBool::new(false); @@ -418,7 +424,11 @@ impl RendezvousMediator { Some(rendezvous_message::Union::IceCandidate(ice)) => { let tx = WEBRTC_ICE_TXS.lock().await.get(&ice.session_key).cloned(); if let Some(tx) = tx { - let _ = tx.send(ice.candidate); + if tx.try_send(ice.candidate).is_err() { + if let Some(n) = FULL_ICE_QUEUE_LOG.due() { + log::debug!("dropped {} ICE candidate(s): queue full or closed", n); + } + } } else if let Some(n) = UNKNOWN_ICE_SESSION_LOG.due() { log::debug!( "dropped {} ICE candidate(s) for unknown WebRTC session key, last: {}", @@ -717,7 +727,13 @@ impl RendezvousMediator { return Ok(answer); }; - let (remote_ice_tx, mut remote_ice_rx) = mpsc::unbounded_channel::(); + // Bounded, like the controller's own candidate buffer: how many candidates arrive is the + // sender's choice, while draining one costs a JSON parse and the ICE agent's lock, so an + // unbounded queue lets whoever can reach this session's route grow it without limit inside + // a long-lived service process. A full queue drops the newest candidate, which costs at + // most one path; ICE keeps whatever pairs it already has. + let (remote_ice_tx, mut remote_ice_rx) = mpsc::channel::(MAX_PENDING_REMOTE_ICE); + let own_ice_tx = remote_ice_tx.clone(); WEBRTC_ICE_TXS .lock() .await @@ -795,10 +811,19 @@ impl RendezvousMediator { let session_key_for_cleanup = session_key.clone(); tokio::spawn(async move { let result = stream.wait_connected(CONNECT_TIMEOUT).await; - WEBRTC_ICE_TXS - .lock() - .await - .remove(&session_key_for_cleanup); + // Only evict our own route. The key is the offer's DTLS fingerprint, identical across + // the controller's punch retries, so a retry that built a fresh answerer has already + // replaced this entry — removing it blindly would delete the live session's sender and + // leave it receiving no candidates at all. + { + let mut txs = WEBRTC_ICE_TXS.lock().await; + if txs + .get(&session_key_for_cleanup) + .is_some_and(|tx| tx.same_channel(&own_ice_tx)) + { + txs.remove(&session_key_for_cleanup); + } + } if let Err(err) = result { log::warn!("webrtc wait_connected failed: {}", err); // Release the pc now rather than waiting for the ICE agent to time out into a diff --git a/src/ui_session_interface.rs b/src/ui_session_interface.rs index cd83cc8e6..fc09ed5e0 100644 --- a/src/ui_session_interface.rs +++ b/src/ui_session_interface.rs @@ -1296,9 +1296,11 @@ impl Session { if true == force_relay { let mut lc = self.lc.write().unwrap(); lc.force_relay = true; - // An explicit retry-via-relay is user policy, not transport necessity: keep - // WebRTC on Relay-only ICE for this round like any force-always-relay session. + // An explicit retry-via-relay is a decision about this peer, not transport + // necessity: Relay-only ICE for this round like any force-always-relay session, + // and it is the one kind of relay that belongs in the peer's saved config. lc.policy_relay = true; + lc.peer_relay = true; } self.lc.write().unwrap().peer_info = None; self.reconnect_count.fetch_add(1, Ordering::SeqCst);