rustdesk/hbb_common
1
0
Fork 0
mirror of https://github.com/rustdesk/hbb_common.git synced 2026-10-11 19:23:59 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
3261057af7134fa08595899aa8aea7f955fb64ca
hbb_common/src/webrtc.rs
T
rustdeskandClaude Opus 5 3261057af7 webrtc: record why WebRTCStream has no Drop 
An absent impl is invisible in the source, and this one keeps being
proposed. Closing on each clone's drop would end a live session the
moment a losing race future is dropped; closing on the last clone is
circular, since the cache entry is itself a clone and is removed by the
state handler a close fires. Ownership is carried by `OffererGuard` and
`Stream::WebRTC` instead, so say so where someone adding another holder
will look.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019UzcMTdYTEv2QbMHcTSUy3
2026-09-04 23:34:08 +08:00

2594 lines
119 KiB
Rust
Raw Blame History

//! WebRTC transport for RustDesk streams.
//!
//! # Bumping the webrtc crate
//!
//! This module depends on webrtc-rs internals its public API does not promise. Re-verify each
//! against the new sources (last checked: webrtc 0.13, -data 0.11, -sctp 0.12), then run
//! `cargo test --features webrtc webrtc::tests`.
//!
//! - `DataChannel::write` parks on a full PendingQueue instead of buffering without bound.
//! - SCTP max message size is 65536: `MAX_FRAGMENT_PAYLOAD` + 1 must stay below it.
//! - `read_sctp` does not await after dequeuing, so a dropped `next()` cannot lose a fragment.
//! - `detach()` is an idempotent Arc clone that does not close on drop.
//! - `on_*` handlers live in the pc: one holding a strong `Arc<RTCPeerConnection>` leaks it.
//! - `close()` latches `is_closed` before its first await, so a cancelled close is unretryable.
//! - `Disconnected` is transient; only `Failed`/`Closed` are terminal.
//! - `RTCIceCandidatePair`'s `Display` layout — `is_relayed()` parses it because 0.13 keeps the
//! candidates private. 0.17+ exposes them; switch and delete the parse.
use std::collections::HashMap;
use std::io::{Error, ErrorKind};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;
use webrtc::api::setting_engine::SettingEngine;
use webrtc::api::APIBuilder;
use webrtc::data::data_channel::DataChannel as DetachedDataChannel;
use webrtc::data_channel::RTCDataChannel;
use webrtc::ice::mdns::MulticastDnsMode;
use webrtc::ice_transport::ice_candidate::RTCIceCandidateInit;
use webrtc::ice_transport::ice_candidate_type::RTCIceCandidateType;
use webrtc::ice_transport::ice_server::RTCIceServer;
use webrtc::peer_connection::configuration::RTCConfiguration;
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
use webrtc::peer_connection::policy::ice_transport_policy::RTCIceTransportPolicy;
use webrtc::peer_connection::sdp::sdp_type::RTCSdpType;
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
use webrtc::peer_connection::RTCPeerConnection;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use base64::Engine;
use bytes::{Buf, BufMut, Bytes, BytesMut};
use tokio::sync::{mpsc, watch, Mutex, Semaphore};
use tokio::time::{timeout, timeout_at, Instant};
use url::Url;
use crate::config;
use crate::protobuf::Message;
use crate::sodiumoxide::crypto::secretbox::Key;
use crate::ResultType;
#[derive(Clone, Debug, PartialEq, Eq)]
enum WebRTCConnectionState {
Pending,
Open,
Closed(String),
}
/// A shared handle, not an owner: every clone points at the same `pc`, and `SESSIONS` holds one
/// of those clones. Hence no `Drop` — closing per clone would kill a live session as soon as a
/// losing race future is dropped, and closing on the last clone would wait on the very close
/// that evicts the cache entry. Ownership is `OffererGuard` until a connection attempt adopts
/// the stream and `Stream::WebRTC` after; holding one outside those two means closing it by hand.
pub struct WebRTCStream {
pc: Arc<RTCPeerConnection>,
stream: Arc<Mutex<Arc<RTCDataChannel>>>,
state_notify: watch::Receiver<WebRTCConnectionState>,
local_ice_rx: Arc<StdMutex<Option<mpsc::UnboundedReceiver<String>>>>,
session_key: String,
send_timeout: u64,
// Built with Relay-only ICE policy (force_relay): every selected pair goes through TURN.
relay_only: bool,
// Detached data channel, cached after the first `detach()` so send/recv do not re-lock and
// re-fetch it per message. Shared across clones; `detach()` is idempotent.
detached: Arc<Mutex<Option<Arc<DetachedDataChannel>>>>,
// Serialize a complete logical message across clones. Each fragment is a separate SCTP
// message, so serializing only individual writes would allow two large messages to interleave.
send_gate: Arc<Semaphore>,
// Behind the Arc, not in the future, so a cancelled `next()` keeps the partial message.
// SINGLE READER ONLY: reassembly spans calls and the fragment header carries no length or
// sequence number, so two readers splice unrelated fragments into one wrong-but-valid message.
recv_state: Arc<Mutex<RecvState>>,
// True once the controller has completed the RustDesk identity binding (DTLS fingerprint
// matched to the signed peer id, via `set_key`). DTLS always encrypts; this flag mirrors TCP's
// "secured after key exchange" so key-less / unbound WebRTC is not shown as peer-authenticated.
peer_verified: Arc<AtomicBool>,
// Whether anyone still wants this pc — see `HandoffState`. Shared by every clone, the
// `SESSIONS` entry included, because that question is about the pc, not about one handle.
handoff: Arc<HandoffState>,
// Taken once at construction: `pc.local_description()` grows with every gathered candidate
// and this must not (why: `UDP_ENDPOINT_BUDGET`). Shared so cloning the struct stays cheap.
local_endpoint: Arc<String>,
}
/// Whether a pc `new_inner` built is still wanted, shared by every `NewStreamHandoff` handed out
/// for it. `new_inner` caches the pc before its `new()` returns, so the builder's handoff and any
/// number of cache hits are in flight at once; a pc is abandoned only when the last of them is
/// dropped with none adopted, and closing on any single Drop would kill a live caller's session.
#[derive(Default)]
struct HandoffState {
// Handoffs handed out and neither adopted nor dropped yet. Incremented under the `SESSIONS`
// lock so a hit that is still arriving cannot be overtaken by the abandon-close below.
pending: AtomicUsize,
// Latched by the first `into_inner`: some caller owns the stream, and since `WebRTCStream`
// has no `Drop` there is nothing that could un-own it. Abandoning any later hit is then a
// no-op, which is what keeps a cached pc alive across a cancelled caller.
adopted: AtomicBool,
// Set under the `SESSIONS` lock once the abandon-close commits, so a hit that locks the map
// afterwards (eviction only happens later, from the state handler) rejects the entry instead
// of adopting a pc that is already going away.
closing: AtomicBool,
}
#[derive(Default)]
struct RecvState {
// Accumulated payload of the logical message currently being reassembled. Only multi-fragment
// messages reach it; a message that arrives whole is split straight out of `scratch`.
acc: BytesMut,
// Read buffer, refilled in `SCRATCH_REFILL` chunks rather than per read. A whole-message
// fragment is handed to the caller with `split_to`, which shares this allocation instead of
// copying, so the buffer is consumed from the front and re-initialized only when what is left
// can no longer hold one fragment — amortizing that cost over many small messages.
scratch: BytesMut,
}
// The SCTP data channel's 65536-byte max message size is handled by
// splitting a logical message into fragments carrying a 1-byte header. Fragment payload is kept
// well under the limit so header+payload never reaches the exact-65536 boundary that the
// receiver's reassembly would truncate with data loss.
const MAX_FRAGMENT_PAYLOAD: usize = 60000;
/// Receive scratch size: must be >= 1 (fragment header) + `MAX_FRAGMENT_PAYLOAD` and fit the
/// negotiated SCTP max message size.
const RECV_BUF_SIZE: usize = 64 * 1024;
/// How much read buffer to initialize at a time. Whole messages are split out of it without
/// copying, so it is consumed rather than reused; refilling in chunks spreads the one cost that
/// remains — zeroing — across every small message that fits. Slices handed to the caller keep the
/// whole chunk alive, so this trades a bounded amount of retention for the copy.
const SCRATCH_REFILL: usize = 4 * RECV_BUF_SIZE;
/// Fragment header byte: more fragments follow for this logical message.
const FRAG_MORE: u8 = 1;
/// Fragment header byte: final (or only) fragment of a logical message.
const FRAG_END: u8 = 0;
/// Largest logical message this transport will send or reassemble, and the memory an
/// unauthenticated peer can make a receiver hold. Kept at parity with TCP on purpose: one session
/// moves between both paths, so a transport-specific ceiling would kill it on the other one.
const MAX_RECV_MESSAGE: usize = crate::bytes_codec::MAX_FRAME_LENGTH;
// use 3 public STUN servers to find out the NAT type, 2 must be the same address but different ports
// https://stackoverflow.com/questions/72805316/determine-nat-mapping-behaviour-using-two-stun-servers
// luckily nextcloud supports two ports for STUN
// unluckily webrtc-rs does not use the same port to do the STUN request
static DEFAULT_ICE_SERVERS: [&str; 3] = [
"stun:stun.cloudflare.com:3478",
"stun:stun.nextcloud.com:3478",
"stun:stun.nextcloud.com:443",
];
lazy_static::lazy_static! {
static ref SESSIONS: Arc::<Mutex<HashMap<String, WebRTCStream>>> = Default::default();
// The process-lifetime runtime that owns ALL WebRTC I/O. A pc's UDP sockets register with
// the reactor — and its ICE/DTLS/SCTP pump tasks spawn on the runtime — that is current
// while `new_inner` and the handlers it installs execute; if that were a session's
// `#[tokio::main]` runtime (dropped the moment io_loop returns), the sockets and pumps
// would die with it and even a close driven elsewhere would "succeed" without ever putting
// close_notify on the wire — the peer then waits out ICE decay. So `new()` runs its body
// here, and detached closes run here too: as independent tasks (a deadline-free `close()`
// must not head-of-line block other sessions' teardown) that are never cancelled (a close
// cancelled after `RTCPeerConnection::close` latches `is_closed` is unretryable, stranding
// the pc in SESSIONS). Data-plane futures (send/recv/close) may be polled from any caller
// runtime — cross-runtime polling is fine; only the OWNING driver must stay alive. Two
// workers so one busy session cannot starve the rest. This is the one deliberate exception
// to AGENTS.md's no-runtime-in-libraries rule: a single lazily-built runtime for I/O whose
// lifetime exceeds any caller's.
static ref WEBRTC_RT: Option<tokio::runtime::Runtime> = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.thread_name("webrtc-io")
.enable_all()
.build()
.map_err(|err| log::error!("failed to build the WebRTC I/O runtime: {}", err))
.ok();
}
/// Hands a stream from `new_inner` (running detached on `WEBRTC_RT`) back to its `new()`
/// caller. If that caller was cancelled while awaiting, the runtime drops the task's output —
/// this guard — and an unanswered offerer never reaches a terminal ICE state on its own, so
/// without the Drop-close its pc, sockets and pump tasks would sit in SESSIONS forever. Claims
/// are counted in the shared `HandoffState`, so only the last one dropped with the stream still
/// unadopted closes it; a hit another caller is already using is left alone.
struct NewStreamHandoff {
stream: Option<WebRTCStream>,
}
impl NewStreamHandoff {
/// Take a claim on `stream`. Call under the `SESSIONS` lock: a claim registered after the map
/// is unlocked can land behind the abandon-close of the claim that was last outstanding.
fn claim(stream: WebRTCStream) -> Self {
stream.handoff.pending.fetch_add(1, Ordering::AcqRel);
Self {
stream: Some(stream),
}
}
fn into_inner(mut self) -> Option<WebRTCStream> {
let stream = self.stream.take()?;
// Before releasing the claim, or a concurrent Drop can see the count reach zero without
// seeing that this caller took ownership.
stream.handoff.adopted.store(true, Ordering::Release);
stream.handoff.pending.fetch_sub(1, Ordering::AcqRel);
Some(stream)
}
}
impl Drop for NewStreamHandoff {
fn drop(&mut self) {
if let Some(stream) = self.stream.take() {
let last = stream.handoff.pending.fetch_sub(1, Ordering::AcqRel) == 1;
if last && !stream.handoff.adopted.load(Ordering::Acquire) {
stream.close_if_abandoned();
}
}
}
}
impl Clone for WebRTCStream {
fn clone(&self) -> Self {
WebRTCStream {
pc: self.pc.clone(),
stream: self.stream.clone(),
state_notify: self.state_notify.clone(),
local_ice_rx: self.local_ice_rx.clone(),
session_key: self.session_key.clone(),
send_timeout: self.send_timeout,
relay_only: self.relay_only,
detached: self.detached.clone(),
send_gate: self.send_gate.clone(),
recv_state: self.recv_state.clone(),
peer_verified: self.peer_verified.clone(),
handoff: self.handoff.clone(),
local_endpoint: self.local_endpoint.clone(),
}
}
}
impl WebRTCStream {
#[inline]
fn get_remote_offer(endpoint: &str) -> ResultType<String> {
// Ensure the endpoint starts with the "webrtc://" prefix
if !endpoint.starts_with("webrtc://") {
return Err(
Error::new(ErrorKind::InvalidInput, "Invalid WebRTC endpoint format").into(),
);
}
// Extract the Base64-encoded SDP part
let encoded_sdp = &endpoint["webrtc://".len()..];
// Decode the Base64 string
let decoded_bytes = BASE64_STANDARD
.decode(encoded_sdp)
.map_err(|_| Error::new(ErrorKind::InvalidInput, "Failed to decode Base64 SDP"))?;
Ok(String::from_utf8(decoded_bytes).map_err(|_| {
Error::new(
ErrorKind::InvalidInput,
"Failed to convert decoded bytes to UTF-8",
)
})?)
}
#[inline]
fn sdp_to_endpoint(sdp: &str) -> String {
let encoded_sdp = BASE64_STANDARD.encode(sdp);
format!("webrtc://{}", encoded_sdp)
}
// Envelope JSON key carrying the local description's ICE transport policy, alongside the
// RTCSessionDescription fields (see `local_endpoint`).
// `'static` spelled out: an elided lifetime here is a warn-by-default future hard error on
// the 1.75 toolchain CI pins (elided_lifetimes_in_associated_constant).
const ICE_POLICY_KEY: &'static str = "ice_policy";
const ICE_POLICY_ALL: &'static str = "all";
#[inline]
fn get_key_for_sdp(sdp: &RTCSessionDescription) -> ResultType<String> {
let binding = sdp.unmarshal()?;
let Some(fingerprint) = binding.attribute("fingerprint") else {
// find fingerprint attribute in media descriptions
for media in &binding.media_descriptions {
if media.media_name.media != "application" {
continue;
}
if let Some(fp) = media
.attributes
.iter()
.find(|x| x.key == "fingerprint")
.and_then(|x| x.value.clone())
{
return Ok(fp);
}
}
return Err(anyhow::anyhow!("SDP fingerprint attribute not found"));
};
Ok(fingerprint.to_string())
}
/// Process-local SESSIONS-map key: the DTLS fingerprint prefixed by role. An offerer and an
/// answerer that share a fingerprint (a single process connecting to its own id) would
/// otherwise collide, handing the offerer back as the answerer. The wire-level `session_key`
/// used for ICE-candidate routing stays the bare fingerprint so both peers still match.
#[inline]
fn cache_key(fingerprint: &str, is_offerer: bool) -> String {
format!(
"{}:{}",
if is_offerer { "offer" } else { "answer" },
fingerprint
)
}
/// Reject a data channel from inside `on_data_channel`, where closing it directly does not
/// work: webrtc-rs runs that handler to completion BEFORE `handle_open` binds the SCTP stream,
/// so `close()` only flips the ready state and `handle_open` then sets it back to Open.
fn close_unbound_channel(dc: Arc<RTCDataChannel>) {
let dc_for_close = dc.clone();
dc.on_open(Box::new(move || {
let dc = dc_for_close.clone();
Box::pin(async move {
if let Err(err) = dc.close().await {
log::debug!("failed to close rejected data channel: {}", err);
}
})
}));
}
/// Whether a cached `SESSIONS` entry may be handed to a caller asking for `force_relay`: it
/// was built for the first caller, so its ICE policy may not match, and the state handler
/// evicts late enough that dead entries linger. Both the lookup and the insert must use it.
fn is_reusable_for(&self, force_relay: bool) -> bool {
self.relay_only == force_relay
&& !self.handoff.closing.load(Ordering::Acquire)
&& !matches!(
*self.state_notify.borrow(),
WebRTCConnectionState::Closed(_)
)
}
#[inline]
fn get_key_for_sdp_json(sdp_json: &str) -> ResultType<String> {
if sdp_json.is_empty() {
return Ok("".to_string());
}
let sdp = serde_json::from_str::<RTCSessionDescription>(sdp_json)?;
Self::get_key_for_sdp(&sdp)
}
fn remove_sessions_for_peer(
sessions: &mut HashMap<String, WebRTCStream>,
pc: &Arc<RTCPeerConnection>,
) -> Vec<String> {
let keys: Vec<String> = sessions
.iter()
.filter_map(|(key, session)| {
Arc::ptr_eq(&session.pc, pc).then(|| key.clone())
})
.collect();
for key in &keys {
sessions.remove(key);
}
keys
}
async fn cache_or_reuse_session(key: String, webrtc_stream: Self) -> Self {
let mut sessions_lock = SESSIONS.lock().await;
if let Some(session) = sessions_lock.get(&key).cloned() {
drop(sessions_lock);
webrtc_stream.pc.close().await.ok();
return session;
}
sessions_lock.insert(key, webrtc_stream.clone());
webrtc_stream
}
#[inline]
fn get_ice_server_from_url(url: &str) -> Option<RTCIceServer> {
let u = Url::parse(url).ok()?;
if !matches!(u.scheme(), "turn" | "turns" | "stun" | "stuns") {
return None;
}
// Two accepted spellings parse differently: `turn://user:pass@host:port` has an authority
// and is the only form that can carry credentials, while the RFC 7065 `turn:host:port` is
// cannot-be-a-base, so `host_str()` is None and the whole `host:port` lands in the path.
let (host, port) = match u.host_str() {
Some(host) => (host.to_owned(), u.port().unwrap_or(3478)),
None => Self::split_host_port(u.path())?,
};
if host.is_empty() {
return None;
}
let username = u.username().to_string();
let credential = u.password().unwrap_or_default().to_string();
// A TURN server without credentials is not merely useless: webrtc-rs validates every
// configured server in `new_peer_connection`, so one credential-less entry makes EVERY
// peer connection fail — including plain non-relay ones that never wanted TURN. The
// RFC 7065 spelling has nowhere to put credentials, so drop such entries here rather
// than let them poison the whole configuration; `turn://user:pass@host:port` carries them.
if matches!(u.scheme(), "turn" | "turns") && (username.is_empty() || credential.is_empty())
{
log::warn!(
"Ignoring TURN server without credentials: {}:{}:{} (use {}://user:pass@host:port)",
u.scheme(),
host,
port,
u.scheme()
);
return None;
}
Some(RTCIceServer {
urls: vec![format!("{}:{}:{}", u.scheme(), host, port)],
username,
credential,
..Default::default()
})
}
/// Split `host[:port]` from an RFC 7065 URL path, defaulting the port. Bracketed IPv6
/// literals keep their brackets, which is the form webrtc-rs re-parses.
fn split_host_port(path: &str) -> Option<(String, u16)> {
let rest = path.split(['?', '#']).next().unwrap_or_default();
if rest.is_empty() {
return None;
}
if let Some(after_open) = rest.strip_prefix('[') {
let (host, tail) = after_open.split_once(']')?;
let port = tail
.strip_prefix(':')
.and_then(|p| p.parse().ok())
.unwrap_or(3478);
return Some((format!("[{host}]"), port));
}
// An unbracketed IPv6 literal has no unambiguous split point — `2001:db8::1` would be cut
// at its last colon into host `2001:db8:` port `1` — so require the bracketed form for
// those rather than emit a host webrtc-ice cannot resolve.
if rest.matches(':').count() > 1 {
log::warn!("Ignoring ICE server {rest}: bracket IPv6 literals as [addr]:port");
return None;
}
match rest.rsplit_once(':') {
Some((host, port)) if !host.is_empty() => match port.parse() {
Ok(port) => Some((host.to_owned(), port)),
// A port that is present but unusable is a typo, not a host: folding it back in
// would produce `host:99999:3478`, which webrtc-ice rejects outright — taking
// every peer connection down with it, not just this server.
Err(_) => {
log::warn!("Ignoring ICE server {rest}: invalid port");
None
}
},
_ => Some((rest.to_owned(), 3478)),
}
}
/// Whether the ICE configuration contains a usable TURN server. A Relay-policy peer
/// connection (force_relay) can only gather relay candidates, so without a TURN server it can
/// never connect — callers use this to skip building a guaranteed-dead pc.
pub fn has_turn_server() -> bool {
// `get_ice_server_from_url` is what makes a bare scheme test sufficient: it drops entries
// with no host and TURN entries with no credentials, i.e. exactly the ones that would
// answer `true` here while being unusable — which is worse than answering `false`, since
// the caller skips its "don't build a guaranteed-dead Relay-only pc" guard on our word.
Self::get_ice_servers().iter().any(|s| {
s.urls
.iter()
.any(|u| u.starts_with("turn:") || u.starts_with("turns:"))
})
}
#[inline]
fn get_ice_servers() -> Vec<RTCIceServer> {
Self::parse_ice_servers(&config::Config::get_option(
config::keys::OPTION_ICE_SERVERS,
))
}
/// Split out from `get_ice_servers` so parsing can be exercised without touching the
/// process-global, on-disk-persisted option — the tests run in parallel threads of one
/// process, so a test that rewrote it raced every peer connection another test was building.
fn parse_ice_servers(cfg: &str) -> Vec<RTCIceServer> {
let mut ice_servers = Vec::new();
let mut has_stun = false;
for url in cfg.split(',').map(str::trim) {
if let Some(ice_server) = Self::get_ice_server_from_url(url) {
// Detect STUN in user config
if ice_server
.urls
.iter()
.any(|u| u.starts_with("stun:") || u.starts_with("stuns:"))
{
has_stun = true;
}
ice_servers.push(ice_server);
}
}
// If there is no STUN (either TURN-only or empty config) → prepend defaults
if !has_stun {
ice_servers.insert(
0,
RTCIceServer {
urls: DEFAULT_ICE_SERVERS.iter().map(|s| s.to_string()).collect(),
..Default::default()
},
);
}
ice_servers
}
/// Built and driven on `WEBRTC_RT`: every socket and background task the pc creates must
/// belong to a runtime that outlives the session (see `WEBRTC_RT`). A caller that abandons
/// this future mid-await leaves the setup task to finish there; the abandoned result then
/// closes a freshly built pc (and leaves a shared cache hit alone) — see `NewStreamHandoff`.
pub async fn new(
remote_endpoint: &str,
force_relay: bool,
ms_timeout: u64,
) -> ResultType<Self> {
let Some(rt) = WEBRTC_RT.as_ref() else {
return Err(anyhow::anyhow!("WebRTC I/O runtime unavailable"));
};
let remote_endpoint = remote_endpoint.to_owned();
rt.spawn(Self::new_inner(remote_endpoint, force_relay, ms_timeout))
.await
.map_err(|err| anyhow::anyhow!("WebRTC setup task failed: {}", err))??
.into_inner()
.ok_or_else(|| anyhow::anyhow!("WebRTC setup handoff was empty"))
}
async fn new_inner(
remote_endpoint: String,
force_relay: bool,
ms_timeout: u64,
) -> ResultType<NewStreamHandoff> {
// The endpoint contains a Base64-encoded SDP with host addresses and live ICE
// credentials. Log only its size so debug logs cannot disclose that information.
log::debug!(
"New webrtc stream (remote endpoint: {} bytes)",
remote_endpoint.len()
);
let remote_offer = if remote_endpoint.is_empty() {
"".into()
} else {
Self::get_remote_offer(&remote_endpoint)?
};
let mut key = Self::get_key_for_sdp_json(&remote_offer)?;
let start_local_offer = remote_offer.is_empty();
if !key.is_empty() {
// Claim inside the lock: `close_if_abandoned` commits under it too, so this hit
// either registers in time to call that close off, or finds the entry already rejected.
let cached = {
let sessions_lock = SESSIONS.lock().await;
match sessions_lock.get(&Self::cache_key(&key, start_local_offer)) {
Some(session) if session.is_reusable_for(force_relay) => {
Some(NewStreamHandoff::claim(session.clone()))
}
Some(session) => {
log::debug!(
"Ignoring cached webrtc peer (relay_only {}, wanted {})",
session.relay_only,
force_relay
);
None
}
None => None,
}
};
if let Some(handoff) = cached {
log::debug!("Start webrtc with cached peer");
return Ok(handoff);
}
}
// Create a SettingEngine and enable Detach
let mut s = SettingEngine::default();
s.detach_data_channels();
s.set_ice_multicast_dns_mode(MulticastDnsMode::Disabled);
// Create the API object
let api = APIBuilder::new().with_setting_engine(s).build();
// Prepare the configuration, get ICE servers from config
let config = RTCConfiguration {
ice_servers: Self::get_ice_servers(),
ice_transport_policy: if force_relay {
RTCIceTransportPolicy::Relay
} else {
RTCIceTransportPolicy::All
},
..Default::default()
};
let (notify_tx, notify_rx) = watch::channel(WebRTCConnectionState::Pending);
let (ice_tx, ice_rx) = mpsc::unbounded_channel::<String>();
// Create a new RTCPeerConnection
let pc = Arc::new(api.new_peer_connection(config).await?);
// `on_*` handlers are never cleared — not by `close()`, not by `SESSIONS` eviction — so
// this sender outlives the pc and would park a caller's ICE forwarder on `recv()` forever,
// holding the pc with it. The terminal-state handler below drops this closure to break it.
let local_ice_tx = ice_tx.clone();
pc.on_ice_candidate(Box::new(move |candidate| {
let local_ice_tx = local_ice_tx.clone();
Box::pin(async move {
let Some(candidate) = candidate else {
return;
};
match candidate.to_json() {
Ok(candidate) => match serde_json::to_string(&candidate) {
Ok(candidate_json) => {
let _ = local_ice_tx.send(candidate_json);
}
Err(err) => {
log::warn!("failed to serialize local ICE candidate: {}", err);
}
},
Err(err) => {
log::warn!("failed to convert local ICE candidate to JSON: {}", err);
}
}
})
}));
let bootstrap_dc = if start_local_offer {
let dc_open_notify = notify_tx.clone();
// Create a data channel with label "bootstrap"
let dc = match pc.create_data_channel("bootstrap", None).await {
Ok(dc) => dc,
Err(e) => {
// Close before propagating: the pc is live and would otherwise leak.
pc.close().await.ok();
return Err(e.into());
}
};
dc.on_open(Box::new(move || {
log::debug!("Local data channel bootstrap open.");
let _ = dc_open_notify.send(WebRTCConnectionState::Open);
Box::pin(async {})
}));
dc
} else {
// Wait for the data channel to be created by the remote peer
// Here we create a dummy data channel to satisfy the type system
Arc::new(RTCDataChannel::default())
};
let stream = Arc::new(Mutex::new(bootstrap_dc));
if !start_local_offer {
// Register data channel creation handling
let dc_open_notify = notify_tx.clone();
let stream_for_dc = stream.clone();
// The remote may open any number of channels; only the first is ever bound.
let dc_bound = Arc::new(AtomicBool::new(false));
pc.on_data_channel(Box::new(move |dc: Arc<RTCDataChannel>| {
let d_label = dc.label().to_owned();
let dc_open_notify2 = dc_open_notify.clone();
let stream_for_dc_clone = stream_for_dc.clone();
let dc_bound = dc_bound.clone();
Box::pin(async move {
// Reassembly is sound only on an ordered, fully-reliable channel: the fragment
// header has no sequence number, so a reorder or a loss silently merges
// messages. webrtc-rs takes these from the unauthenticated REMOTE's DCEP OPEN.
if !dc.ordered()
|| dc.max_retransmits().is_some()
|| dc.max_packet_lifetime().is_some()
{
log::warn!(
"Rejecting WebRTC data channel {}: not ordered and fully reliable",
d_label
);
Self::close_unbound_channel(dc);
return;
}
// Bind the first channel only. `detached` caches the first detached handle
// for the life of the stream, so rebinding would leave teardown closing a
// channel that send/recv no longer use; worse, a second channel's `on_open`
// would push Open onto the same watch and re-arm a session already latched
// Closed — and that watch gates both `send_bytes_inner` and `next()`.
if dc_bound.swap(true, Ordering::SeqCst) {
log::warn!("Ignoring extra WebRTC data channel {}", d_label);
Self::close_unbound_channel(dc);
return;
}
log::debug!("Remote data channel {} ready", d_label);
let mut stream_lock = stream_for_dc_clone.lock().await;
*stream_lock = dc.clone();
drop(stream_lock);
dc.on_open(Box::new(move || {
let _ = dc_open_notify2.send(WebRTCConnectionState::Open);
Box::pin(async {})
}));
})
}));
}
// This will notify you when the peer has connected/disconnected
let stream_for_close = stream.clone();
// Weak, not strong: a handler stored inside the pc that captured a strong
// `Arc<RTCPeerConnection>` forms a pc -> internal -> handler -> pc cycle that `close()`
// never breaks (it only fires the handler) and no `Drop` clears, permanently leaking every
// pc and the ICE-candidate sender's forwarding task. Upgrade inside the handler; if the pc
// is already gone there is nothing left in SESSIONS to evict.
let pc_for_close = Arc::downgrade(&pc);
pc.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
let stream_for_close2 = stream_for_close.clone();
let on_connection_notify = notify_tx.clone();
let pc_for_close2 = pc_for_close.clone();
Box::pin(async move {
log::debug!("WebRTC session peer connection state: {}", s);
match s {
// `Disconnected` is a transient, recoverable ICE state (webrtc-ice fires it
// after ~5s without consent and returns to `Connected` when traffic resumes).
// Only tear down on the terminal states so a short network blip (Wi-Fi roam,
// sleep/wake, cell handover) does not permanently kill an established session.
RTCPeerConnectionState::Failed | RTCPeerConnectionState::Closed => {
let _ =
on_connection_notify.send(WebRTCConnectionState::Closed(s.to_string()));
log::debug!("WebRTC session closing due to {}", s);
let _ = stream_for_close2.lock().await.close().await;
log::debug!("WebRTC session stream closed");
let Some(pc_for_close2) = pc_for_close2.upgrade() else {
return;
};
// Nothing else ever drops this sender (`close()` clears no handler), so
// replace the handler to close the channel: a caller's forwarder loop then
// ends instead of parking on `recv()` and holding this pc alive.
pc_for_close2.on_ice_candidate(Box::new(|_| Box::pin(async {})));
let mut sessions_lock = SESSIONS.lock().await;
match Self::get_key_for_peer(&pc_for_close2, start_local_offer).await {
Ok(fingerprint) => {
let k = Self::cache_key(&fingerprint, start_local_offer);
// Only evict if the cached entry IS this pc: a duplicate offer
// resolves to the same key, and closing the discarded duplicate pc
// must not remove the live winner sharing that key.
if sessions_lock
.get(&k)
.is_some_and(|s| Arc::ptr_eq(&s.pc, &pc_for_close2))
{
sessions_lock.remove(&k);
log::debug!("WebRTC session removed key: {}", k);
}
}
Err(e) => {
log::error!(
"Failed to extract key for peer during session cleanup: {:?}",
e
);
// Fallback: try to remove any session associated with this peer connection
let keys_to_remove: Vec<String> = sessions_lock
.iter()
.filter_map(|(key, session)| {
if Arc::ptr_eq(&session.pc, &pc_for_close2) {
Some(key.clone())
} else {
None
}
})
.collect();
for k in keys_to_remove {
sessions_lock.remove(&k);
log::debug!("WebRTC session removed by fallback key: {}", k);
}
}
}
}
_ => {}
}
})
}));
// Trickle ICE: local-only work, no gathering wait — the controlled side awaits answer
// creation inline on its punch-reply critical path. A failure below leaves a live pc whose
// state handler only fires on a terminal ICE state, so close before propagating.
// Encode each endpoint before `set_local_description`, which is what starts gathering, so
// there is nothing for `trickle_endpoint` to strip in the first place.
let offer_answer: ResultType<(String, String)> = async {
if start_local_offer {
let sdp = pc.create_offer(None).await?;
let endpoint = Self::trickle_endpoint(&sdp, force_relay)?;
pc.set_local_description(sdp.clone()).await?;
// SDP carries host/srflx IPs and ICE ufrag/pwd; log only its size, not the body.
log::debug!(
"local offer SDP built ({} bytes, endpoint {} bytes)",
sdp.sdp.len(),
endpoint.len()
);
let k = Self::get_key_for_sdp(&sdp)?;
log::debug!("Start webrtc with local key: {}", k);
Ok((k, endpoint))
} else {
let sdp = serde_json::from_str::<RTCSessionDescription>(&remote_offer)?;
pc.set_remote_description(sdp.clone()).await?;
let answer = pc.create_answer(None).await?;
let endpoint = Self::trickle_endpoint(&answer, force_relay)?;
pc.set_local_description(answer).await?;
log::debug!(
"remote offer SDP received ({} bytes), local answer endpoint {} bytes",
sdp.sdp.len(),
endpoint.len()
);
let k = Self::get_key_for_sdp(&sdp)?;
log::debug!("Start webrtc with remote key: {}", k);
Ok((k, endpoint))
}
}
.await;
let (new_key, local_endpoint) = match offer_answer {
Ok(x) => x,
Err(e) => {
pc.close().await.ok();
return Err(e);
}
};
// Only the UDP leg has a ceiling, and which leg carries this is the rendezvous server's
// to decide, so this reports rather than refuses: over a TCP/WS route a larger endpoint is
// delivered fine, and failing here would cost WebRTC for no reason.
if local_endpoint.len() > Self::UDP_ENDPOINT_BUDGET {
log::warn!(
"WebRTC endpoint is {} bytes, over the {} byte UDP budget; a peer reached over UDP \
may never receive it",
local_endpoint.len(),
Self::UDP_ENDPOINT_BUDGET
);
}
key = new_key;
let webrtc_stream = Self {
pc,
stream,
state_notify: notify_rx,
local_ice_rx: Arc::new(StdMutex::new(Some(ice_rx))),
session_key: key.clone(),
send_timeout: ms_timeout,
relay_only: force_relay,
detached: Arc::new(Mutex::new(None)),
send_gate: Arc::new(Semaphore::new(1)),
recv_state: Arc::new(Mutex::new(RecvState::default())),
peer_verified: Arc::new(AtomicBool::new(false)),
handoff: Default::default(),
local_endpoint: Arc::new(local_endpoint),
};
// Insert into the session cache, but never `await pc.close()` while holding this lock:
// `close()` fires the peer-connection-state handler inline, which itself locks SESSIONS,
// self-deadlocking the whole process. Resolve any duplicate off-lock.
let cache_key = Self::cache_key(&key, start_local_offer);
// Claim inside the lock, the insert included: with the pc cached but unclaimed, a hit
// could take the only claim on it and release it again, closing it under this caller.
let mut duplicate = None;
let handoff = {
let mut final_lock = SESSIONS.lock().await;
// Same admissibility test as the lookup above, or that lookup is dead code: an entry
// rejected there is still in the map when we get here, so returning it unconditionally
// would discard the pc we just built precisely because the cached one was unusable.
match final_lock.get(&cache_key) {
Some(session) if session.is_reusable_for(force_relay) => {
duplicate = Some(webrtc_stream.clone());
NewStreamHandoff::claim(session.clone())
}
_ => {
final_lock.insert(cache_key, webrtc_stream.clone());
NewStreamHandoff::claim(webrtc_stream)
}
}
};
if let Some(loser) = duplicate {
// A concurrent `new()` already cached an equivalent stream; discard this pc's
// resources (on the closer thread — awaiting here would strand the pc if this
// `new()` were cancelled mid-close) and return the cached one.
loser.close_detached();
}
Ok(handoff)
}
/// One-shot endpoint: waits for ICE gathering so the SDP already carries the candidates.
/// The wait is deliberately unbounded — a deadline here would return a half-gathered SDP and
/// defeat the contract; callers needing one should wrap the call or use the trickle variant,
/// which both rustdesk signaling paths do. Remaining callers are the loopback tests and
/// `examples/webrtc.rs`, neither of which bounds it.
#[inline]
pub async fn get_local_endpoint(&self) -> ResultType<String> {
// Preserve the original one-shot endpoint contract: callers that only exchange this SDP
// do not have a separate path for `take_local_ice_rx`, so their endpoint must contain the
// gathered host/srflx/relay candidates.
let mut gather_complete = self.pc.gathering_complete_promise().await;
let _gathering_channel_closed = gather_complete.recv().await;
let Some(local_desc) = self.pc.local_description().await else {
return Err(anyhow::anyhow!("Local desc is not set"));
};
Self::encode_endpoint(local_desc.sdp_type, &local_desc.sdp, self.relay_only)
}
/// The offer/answer exactly as it was created, for callers that signal candidates via
/// `take_local_ice_rx`. Taken at construction, so this cannot fail and cannot block — unlike
/// `get_local_endpoint`, which reads the live description and waits for gathering.
///
/// Deliberately not `pc.local_description()`, which runs `populate_local_candidates` and so
/// returns the SDP plus every candidate gathered up to that moment — and callers read this a
/// network round trip after `new`, by which time gathering has filled it in. Why the size
/// matters: `UDP_ENDPOINT_BUDGET`.
#[inline]
pub fn local_endpoint(&self) -> &str {
self.local_endpoint.as_str()
}
/// The size a trickle endpoint must stay under for the UDP leg of its route to be safe. It is
/// the one signaling message that can grow: a peer registered over UDP receives it inside
/// `PunchHole` — with the mangled addresses and the permission blobs — as a single datagram,
/// and one that needs IP fragmentation is dropped without a trace on paths that discard
/// fragments. The candidates that follow are one small message each and never approach it.
///
/// Not an MTU calculation: `PunchHole`'s other fields are not bounded here, so a value near
/// this number could still fragment. It is a tripwire on a quantity `trickle_endpoint` pins at
/// ~700 bytes, sized to leave that headroom while still catching a regression that lets
/// candidates back in — as a log line rather than another silent drop. Nothing enforces it: a
/// TCP/WS leg has no packet ceiling.
const UDP_ENDPOINT_BUDGET: usize = 1024;
/// The endpoint a trickling peer signals: the session parameters needed to start ICE and DTLS
/// (ice-ufrag, ice-pwd, fingerprint, setup, the sctp m-line), without the candidates — those
/// ride `take_local_ice_rx` and arrive as individual `IceCandidate` messages.
///
/// Candidates are the only part of a local description that grows. Stripping them makes the
/// size a property of the value rather than of when it was taken, so it cannot drift with
/// gathering however this is later refactored.
fn trickle_endpoint(local_desc: &RTCSessionDescription, relay_only: bool) -> ResultType<String> {
let mut sdp = String::with_capacity(local_desc.sdp.len());
for line in local_desc.sdp.lines() {
// `end-of-candidates` would tell the remote agent to stop waiting for the trickle.
if line.starts_with("a=candidate:") || line.starts_with("a=end-of-candidates") {
continue;
}
sdp.push_str(line);
sdp.push_str("\r\n");
}
Self::encode_endpoint(local_desc.sdp_type, &sdp, relay_only)
}
/// Base64 envelope of a local description, carrying its ICE transport policy unless the pc is
/// Relay-only (see `ICE_POLICY_KEY`).
/// Built from the two fields a session description serializes rather than from the value, so
/// a caller that rewrote the SDP cannot leave a stale `parsed` tree riding along with it.
fn encode_endpoint(sdp_type: RTCSdpType, sdp: &str, relay_only: bool) -> ResultType<String> {
let mut envelope = serde_json::Map::new();
envelope.insert("type".to_owned(), serde_json::to_value(sdp_type)?);
envelope.insert("sdp".to_owned(), serde_json::Value::from(sdp));
if !relay_only {
// Rides in the envelope, not a proto field the rendezvous server would forward,
// because it is the offer's own property: the receiver must know whether
// force_relay was policy (stay Relay-only) or transport. Older peers ignore it.
envelope.insert(
Self::ICE_POLICY_KEY.to_owned(),
serde_json::Value::from(Self::ICE_POLICY_ALL),
);
}
Ok(Self::sdp_to_endpoint(&serde_json::to_string(&envelope)?))
}
/// Whether the peer's endpoint declares it was built with ICE transport policy `all`
/// (W3C RTCIceTransportPolicy), i.e. it gathers host/srflx/relay candidates and a direct
/// pair may form even though the request carries force_relay. Absent key, foreign format
/// or parse failure all mean "not declared" — the old Relay-only reading.
pub fn endpoint_declares_all_ice(endpoint: &str) -> bool {
let Ok(sdp_json) = Self::get_remote_offer(endpoint) else {
return false;
};
serde_json::from_str::<serde_json::Value>(&sdp_json)
.ok()
.and_then(|v| {
v.get(Self::ICE_POLICY_KEY)?
.as_str()
.map(|p| p == Self::ICE_POLICY_ALL)
})
.unwrap_or(false)
}
#[inline]
pub async fn set_remote_endpoint(&self, endpoint: &str) -> ResultType<()> {
let offer = Self::get_remote_offer(endpoint)?;
log::debug!("WebRTC set remote sdp ({} bytes)", offer.len());
let sdp = serde_json::from_str::<RTCSessionDescription>(&offer)?;
self.pc.set_remote_description(sdp).await?;
Ok(())
}
/// DTLS certificate fingerprint of the local description (this endpoint's own cert).
#[inline]
pub async fn local_dtls_fingerprint(&self) -> ResultType<String> {
Self::get_key_for_peer(&self.pc, true).await
}
/// DTLS certificate fingerprint of the remote description (the peer's cert). webrtc-rs
/// verifies the negotiated peer certificate against this fingerprint during the DTLS
/// handshake, so once the channel is open a matching fingerprint identifies the peer's cert.
#[inline]
pub async fn remote_dtls_fingerprint(&self) -> ResultType<String> {
Self::get_key_for_peer(&self.pc, false).await
}
/// Whether the connection runs through a TURN relay; feeds the UI's direct/relayed flag.
/// Under Relay policy the answer is known by construction, so that arm returns `Some(true)`
/// without consulting ICE — including before a pair is selected, unlike the `None` the other
/// arm returns then. Both callers ask post-connection, where the arms agree; making the relay
/// arm await a pair would only add a stats round trip.
pub async fn is_relayed(&self) -> Option<bool> {
if self.relay_only {
return Some(true);
}
let dtls = self.pc.sctp().transport();
let pair = dtls.ice_transport().get_selected_candidate_pair().await?;
// Not the stats report's `nominated` flag: webrtc-ice never clears it per checklist entry,
// so after a pair switch several entries carry it and map order decides the answer.
// 0.13 keeps the candidates private, so read the types out of `Display` by position.
let relay = RTCIceCandidateType::Relay.to_string();
Some(
pair.to_string()
.split(" <-> ")
.filter_map(|side| side.split_whitespace().nth(2))
.any(|typ| typ == relay),
)
}
#[inline]
pub fn take_local_ice_rx(&self) -> Option<mpsc::UnboundedReceiver<String>> {
self.local_ice_rx.lock().ok().and_then(|mut rx| rx.take())
}
#[inline]
pub async fn add_remote_ice_candidate(&self, candidate_json: &str) -> ResultType<()> {
if candidate_json.is_empty() {
return Ok(());
}
let candidate = serde_json::from_str::<RTCIceCandidateInit>(candidate_json)?;
self.pc.add_ice_candidate(candidate).await?;
Ok(())
}
#[inline]
pub fn session_key(&self) -> &str {
&self.session_key
}
pub async fn wait_connected(&mut self, ms: u64) -> ResultType<()> {
if ms > 0 {
match timeout(Duration::from_millis(ms), self.wait_for_connect_result()).await {
Ok(result) => result?,
Err(_) => return Err(anyhow::anyhow!("WebRTC wait_connected timeout")),
}
} else {
self.wait_for_connect_result().await?;
}
Ok(())
}
/// Explicitly tear down the peer connection. Dropping the handle is not enough: `SESSIONS`
/// holds a clone, so the pc and its ICE/DTLS resources survive until it happens to reach a
/// terminal ICE state. Closing fires the state handler, which evicts the `SESSIONS` entry —
/// callers that abandon a stream (e.g. an offerer that lost the transport race) must call it.
#[inline]
pub async fn close(&self) {
self.pc.close().await.ok();
}
/// Tear the pc down as an independent task on `WEBRTC_RT` — the runtime that owns the
/// pc's sockets and pump tasks, so the close can actually reach the wire no matter which
/// caller runtimes have died — keeping `keep` alive until the teardown finishes. Callable
/// from any context, `Drop` on a runtime-less thread included, and never cancelled, so
/// `close()`'s `is_closed` latch is only ever set by a close that finishes. The send path
/// passes its permit as `keep`.
pub fn close_detached_with<T: Send + 'static>(&self, keep: T) {
// If the runtime never built, no pc exists either (`new` fails first); nothing to close.
let Some(rt) = WEBRTC_RT.as_ref() else {
return;
};
let pc = self.pc.clone();
rt.spawn(async move {
if let Err(err) = pc.close().await {
log::debug!("WebRTC close failed: {}", err);
}
drop(keep);
});
}
#[inline]
pub fn close_detached(&self) {
self.close_detached_with(());
}
/// Close a pc no caller ever took. Spawned on `WEBRTC_RT` like `close_detached`, but the
/// decision is remade under the `SESSIONS` lock that a cache hit claims under: either the hit
/// registers first and this call stands down, or `closing` commits first and
/// `is_reusable_for` rejects the entry, which the state handler evicts only later.
fn close_if_abandoned(&self) {
let Some(rt) = WEBRTC_RT.as_ref() else {
return;
};
let stream = self.clone();
rt.spawn(async move {
{
let _sessions_lock = SESSIONS.lock().await;
if stream.handoff.pending.load(Ordering::Acquire) != 0
|| stream.handoff.adopted.load(Ordering::Acquire)
{
return;
}
stream.handoff.closing.store(true, Ordering::Release);
}
// Off-lock: `close()` fires the state handler inline and that handler locks SESSIONS.
if let Err(err) = stream.pc.close().await {
log::debug!("WebRTC close of an abandoned peer failed: {}", err);
}
});
}
#[inline]
pub fn set_raw(&mut self) {
// not-supported
}
#[inline]
pub fn local_addr(&self) -> SocketAddr {
SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0)
}
#[inline]
pub fn set_send_timeout(&mut self, ms: u64) {
self.send_timeout = ms;
}
#[inline]
pub fn set_key(&mut self, _key: Key) {
// WebRTC traffic is DTLS-encrypted regardless; the secretbox key is unused.
// Callers invoke set_key only after the controller has bound the DTLS fingerprint to the
// verified peer identity (or the controlled side has completed the matching handshake).
// Mark peer-verified so is_secured() matches TCP's post-key-exchange meaning.
self.peer_verified.store(true, Ordering::Release);
}
#[inline]
pub fn is_secured(&self) -> bool {
self.peer_verified.load(Ordering::Acquire)
}
#[inline]
pub async fn send(&mut self, msg: &impl Message) -> ResultType<()> {
self.send_raw(msg.write_to_bytes()?).await
}
#[inline]
pub async fn send_raw(&mut self, msg: Vec<u8>) -> ResultType<()> {
self.send_bytes(Bytes::from(msg)).await
}
#[inline]
async fn wait_for_connect_result(&mut self) -> ResultType<()> {
loop {
match self.state_notify.borrow().clone() {
WebRTCConnectionState::Open => return Ok(()),
WebRTCConnectionState::Closed(reason) => {
return Err(anyhow::anyhow!("WebRTC connection closed: {}", reason));
}
WebRTCConnectionState::Pending => {}
}
self.state_notify.changed().await?;
}
}
/// Fetch (and cache) the detached data channel. `detach()` is idempotent and returns a
/// clone of the same underlying channel, so caching it just avoids re-locking per message.
async fn detached_dc(&self) -> ResultType<Arc<DetachedDataChannel>> {
{
let cache = self.detached.lock().await;
if let Some(dc) = cache.as_ref() {
return Ok(dc.clone());
}
}
let raw = self.stream.lock().await.clone();
let dc = raw.detach().await?;
let mut cache = self.detached.lock().await;
// Another task may have cached it while we were detaching.
if let Some(existing) = cache.as_ref() {
return Ok(existing.clone());
}
*cache = Some(dc.clone());
Ok(dc)
}
/// NOT cancel-safe: dropping this future mid-message (e.g. wrapping it in `select!`/`timeout`)
/// can leave a partial fragment sequence on the wire, corrupting reassembly of every later
/// message on this stream. A caller that abandons a send must treat the stream as dead and
/// close it; the built-in `send_timeout` path below already does (it closes the pc).
pub async fn send_bytes(&mut self, bytes: Bytes) -> ResultType<()> {
let send_timeout = self.send_timeout;
let send_gate = self.send_gate.clone();
// Bound the WHOLE send (wait-for-open, queueing behind another clone, every write) by
// send_timeout: otherwise a write parks indefinitely on SCTP backpressure and
// connection.rs's timer never runs. That parking is also what bounds sender memory —
// PendingQueue admits 128 KiB and inflight is cwnd/rwnd-capped — so this timeout is the
// TCP-send-timeout equivalent. See the checklist entry on send backpressure.
if send_timeout > 0 {
// Waiting for the connection to open is not this message's progress. Charging it to
// the send clock meant that on the first send after signaling, a slow ICE/DTLS
// completion ate most of the budget and the write inherited the remainder — then
// timed out and closed a peer connection that was an RTT from working. Give it its
// own budget and start the send clock once the channel is actually usable.
timeout(
Duration::from_millis(send_timeout),
self.wait_for_connect_result(),
)
.await
.map_err(|_| Error::new(ErrorKind::TimedOut, "WebRTC connect timeout"))??;
let gate_deadline = Instant::now() + Duration::from_millis(send_timeout);
let send_permit = match timeout_at(gate_deadline, send_gate.acquire_owned()).await {
Ok(Ok(permit)) => permit,
Ok(Err(err)) => {
return Err(Error::new(
ErrorKind::BrokenPipe,
format!("WebRTC send gate closed: {}", err),
)
.into());
}
Err(_) => {
// Deliberately no teardown: this task never held the permit, so all that
// happened is that another clone is legitimately mid-message. Closing from
// here would abort a healthy sender's fragment sequence — precisely the
// corruption the permit exists to prevent.
return Err(Error::new(ErrorKind::TimedOut, "WebRTC send gate timeout").into());
}
};
// Fresh budget once the permit is held: queueing behind another clone's message is
// not this message's progress, so charging it here meant a caller that only just lost
// the gate race got a few milliseconds to write in — and then tore the whole peer
// connection down for missing them. The narrower the miss, the more certain the
// teardown, which is precisely backwards.
let write_deadline = Instant::now() + Duration::from_millis(send_timeout);
let mut desynced = false;
match timeout_at(write_deadline, self.send_bytes_inner(bytes, &mut desynced)).await {
// A write that failed part-way leaves the same orphaned prefix a timeout does, so
// it needs the same teardown — and the same handoff of the permit, or a waiting
// clone appends its message to that prefix while the close is still in flight.
Ok(res) => {
if desynced {
self.close_detached_with(send_permit);
}
res
}
Err(_) => {
// Hand the logical-message permit to the teardown so no waiting clone can
// append a new message after a partially-written fragment sequence. Holding
// it across an awaited close would drop both on cancellation.
self.close_detached_with(send_permit);
Err(Error::new(ErrorKind::TimedOut, "WebRTC send timeout").into())
}
}
} else {
let send_permit = send_gate.acquire_owned().await.map_err(|err| {
Error::new(
ErrorKind::BrokenPipe,
format!("WebRTC send gate closed: {}", err),
)
})?;
let mut desynced = false;
let res = self.send_bytes_inner(bytes, &mut desynced).await;
if desynced {
self.close_detached_with(send_permit);
}
res
}
}
/// `desynced` is set when the failure left a partial fragment sequence on the wire, i.e. when
/// the stream can no longer be made consistent and the caller must close it.
async fn send_bytes_inner(&mut self, bytes: Bytes, desynced: &mut bool) -> ResultType<()> {
// Same bound the receiver enforces, so we never emit a message a same-version peer
// would have to kill the connection over.
if bytes.len() > MAX_RECV_MESSAGE {
return Err(Error::new(ErrorKind::InvalidInput, "Overflow").into());
}
self.wait_for_connect_result().await?;
let dc = self.detached_dc().await?;
let data = bytes.as_ref();
let mut offset = 0;
// Always emit at least one fragment (a lone FRAG_END header for an empty message), so a
// zero-length data-channel message — which the receiver cannot distinguish from EOF — is
// never sent.
let mut wrote_any = false;
loop {
let end = (offset + MAX_FRAGMENT_PAYLOAD).min(data.len());
let is_last = end >= data.len();
let chunk = &data[offset..end];
let mut framed = BytesMut::with_capacity(1 + chunk.len());
framed.put_u8(if is_last { FRAG_END } else { FRAG_MORE });
framed.put_slice(chunk);
if let Err(err) = dc.write(&framed.freeze()).await {
// Fragments already on the wire leave the peer an unterminated prefix, and this
// framing has no length for it to notice with, so the next message is appended
// to it. Callers wrap sends in `allow_err!`, so the error alone is not enough.
*desynced = wrote_any;
return Err(err.into());
}
wrote_any = true;
offset = end;
if is_last {
break;
}
}
Ok(())
}
#[inline]
pub async fn next(&mut self) -> Option<Result<BytesMut, Error>> {
if let Err(err) = self.wait_for_connect_result().await {
self.close_detached();
return Some(Err(Error::new(ErrorKind::Other, err.to_string())));
}
let dc = match self.detached_dc().await {
Ok(dc) => dc,
Err(err) => {
self.close_detached();
return Some(Err(Error::new(ErrorKind::Other, err.to_string())));
}
};
// Held across `dc.read().await` on purpose, against the usual "no locks across await"
// rule: it is what makes the single reader this reassembly requires (see the field) an
// exclusion rather than a convention. Releasing it around the read would admit exactly
// the second reader that corrupts `acc`. The cost — a would-be second reader blocking
// until a packet arrives — is a state the design does not permit anyway.
let mut st = self.recv_state.lock().await;
loop {
let RecvState { acc, scratch } = &mut *st;
if scratch.len() < RECV_BUF_SIZE {
scratch.resize(SCRATCH_REFILL, 0);
}
let n = match dc.read(&mut scratch[..RECV_BUF_SIZE]).await {
Ok(n) => n,
Err(err) => {
// Release the partial message, as the framing-violation paths below do: the
// buffer can hold up to the cap and `recv_state` outlives this call through
// the `SESSIONS` clone.
*acc = BytesMut::new();
self.close_detached();
return Some(Err(Error::new(
ErrorKind::Other,
format!("data channel read error: {}", err),
)));
}
};
if n == 0 {
// Not exclusively a stream reset: webrtc-data maps the empty-message PPIDs to
// n == 0 as well and `read` drops the flag that would separate them. Both mean
// nothing frameable follows, so treat them alike but not as a clean remote close.
log::debug!("WebRTC data channel ended (reset or empty message)");
*acc = BytesMut::new();
self.close_detached();
return None;
}
// Both would otherwise read as "more fragments". The payload-less FRAG_MORE is the
// nastier one: it adds nothing to `acc`, so the cap below never trips and the loop
// spins for as long as the peer keeps sending.
let header = scratch[0];
let bad = match header {
FRAG_END => None,
FRAG_MORE if n > 1 => None,
FRAG_MORE => Some("FRAG_MORE fragment carries no payload".to_owned()),
other => Some(format!(
"fragment header {other} is neither FRAG_END nor FRAG_MORE"
)),
};
if let Some(why) = bad {
*acc = BytesMut::new();
self.close_detached();
return Some(Err(Error::new(
ErrorKind::InvalidData,
format!("WebRTC {why}"),
)));
}
// Bound BEFORE growing: this framing carries no length, so an oversize message is only
// discovered by accumulating it, and checking after the append would let `BytesMut`'s
// reallocate-and-copy hold the old and the new buffer at once.
if acc.len() + (n - 1) > MAX_RECV_MESSAGE {
// Release the buffer, don't just truncate it: by definition it is near the cap
// here, and `recv_state` outlives this call through the `SESSIONS` clone.
*acc = BytesMut::new();
self.close_detached();
return Some(Err(Error::new(
ErrorKind::InvalidData,
"WebRTC reassembled message exceeded maximum frame size",
)));
}
// A message that arrived whole is handed over as a slice of the read buffer: no
// allocation and no copy, which is every input event and most audio packets.
if header == FRAG_END && acc.is_empty() {
scratch.advance(1);
return Some(Ok(scratch.split_to(n - 1)));
}
acc.extend_from_slice(&scratch[1..n]);
if header == FRAG_END {
let msg = std::mem::take(acc);
return Some(Ok(msg));
}
}
}
#[inline]
pub async fn next_timeout(&mut self, ms: u64) -> Option<Result<BytesMut, Error>> {
match timeout(Duration::from_millis(ms), self.next()).await {
Ok(res) => res,
Err(_) => None,
}
}
}
pub fn is_webrtc_endpoint(endpoint: &str) -> bool {
// use sdp base64 json string as endpoint, or prefix webrtc:
endpoint.starts_with("webrtc://")
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use crate::webrtc::{NewStreamHandoff, WebRTCStream, DEFAULT_ICE_SERVERS, FRAG_MORE, SESSIONS, WEBRTC_RT};
use bytes::{BufMut, Bytes, BytesMut};
use tokio::sync::{watch, Barrier, Mutex};
use tokio::time::timeout;
use webrtc::api::APIBuilder;
use webrtc::data_channel::RTCDataChannel;
use webrtc::peer_connection::configuration::RTCConfiguration;
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
async fn new_test_stream() -> WebRTCStream {
let api = APIBuilder::new().build();
let pc = Arc::new(
api.new_peer_connection(RTCConfiguration::default())
.await
.unwrap(),
);
let (_, state_notify) = watch::channel(false);
WebRTCStream {
pc,
stream: Arc::new(Mutex::new(Arc::new(RTCDataChannel::default()))),
state_notify,
send_timeout: 0,
}
}
#[test]
fn test_webrtc_ice_url() {
let turn = WebRTCStream::get_ice_server_from_url("turn://123:321@example.com:3478")
.expect("credentialed turn is usable");
assert_eq!(turn.urls[0], "turn:example.com:3478");
assert_eq!(turn.username, "123");
assert_eq!(turn.credential, "321");
assert_eq!(
WebRTCStream::get_ice_server_from_url("turn://123:321@example.com")
.unwrap_or_default()
.urls[0],
"turn:example.com:3478"
);
// TURN without both halves of the credential is dropped rather than passed on:
// webrtc-rs validates every configured server when the peer connection is built, so
// one such entry fails EVERY connection, including those that never wanted TURN.
for missing in [
"turn://example.com:3478",
"turn://example.com",
"turn://123@example.com",
"turns://example.com:5349",
"turn:example.com:3478",
] {
assert_eq!(
WebRTCStream::get_ice_server_from_url(missing),
None,
"credential-less {missing} must not reach the configuration"
);
}
// STUN needs no credentials, so both spellings stay usable.
assert_eq!(
WebRTCStream::get_ice_server_from_url("stun://example.com:3478")
.unwrap_or_default()
.urls[0],
"stun:example.com:3478"
);
assert_eq!(
WebRTCStream::get_ice_server_from_url("http://123:123@example.com:3478"),
None
);
// RFC 7065 spelling (`scheme:host:port`, no authority) — what STUN/TURN docs hand out.
// `url` cannot-be-a-base's it, so host and port come out of the path; getting this wrong
// produced a hostless "stun::3478" no ICE agent can resolve.
for (input, expected) in [
("stun:example.com:19302", "stun:example.com:19302"),
("stun:example.com", "stun:example.com:3478"),
("stun:[2001:db8::1]:19302", "stun:[2001:db8::1]:19302"),
("stun:[2001:db8::1]", "stun:[2001:db8::1]:3478"),
(
"stun:example.com:19302?transport=udp",
"stun:example.com:19302",
),
] {
assert_eq!(
WebRTCStream::get_ice_server_from_url(input)
.unwrap_or_default()
.urls[0],
expected,
"parsing {input}"
);
}
// Malformed entries are dropped, never repaired into something webrtc-ice will choke
// on: a folded-in bad port ("host:99999:3478") or an unbracketed IPv6 split at its last
// colon fails peer-connection construction outright, taking every server down with it.
for bad in [
"stun:",
"stun:example.com:99999",
"stun:example.com:abc",
"stun:2001:db8::1",
] {
assert_eq!(
WebRTCStream::get_ice_server_from_url(bad),
None,
"malformed {bad} must be dropped"
);
}
}
// Parsing is exercised through `parse_ice_servers`, never by rewriting the global
// `ice-servers` option: `set_option` persists to the real config file and the tests share
// one process, so mutating it raced every peer connection the loopback tests were building.
#[test]
fn test_webrtc_ice_server_list() {
assert_eq!(
WebRTCStream::parse_ice_servers("")[0].urls[0],
DEFAULT_ICE_SERVERS[0].to_string()
);
// Unusable entries drop out of the list; the rest of the config still applies.
let parsed = WebRTCStream::parse_ice_servers(
",stun://example.com,turn://u:p@example.com,turn://nocreds.example.com,sdf",
);
assert_eq!(parsed[0].urls[0], "stun:example.com:3478");
assert_eq!(parsed[1].urls[0], "turn:example.com:3478");
assert_eq!(parsed.len(), 2);
// TURN-only config still gets the default STUN servers prepended.
let turn_only = WebRTCStream::parse_ice_servers("turn://u:p@example.com:3478");
assert_eq!(turn_only[0].urls[0], DEFAULT_ICE_SERVERS[0].to_string());
assert_eq!(turn_only[1].urls[0], "turn:example.com:3478");
}
#[test]
fn test_endpoint_ice_policy_declaration() {
// An envelope with the marker declares full ICE; everything else — no marker,
// wrong value, foreign scheme, garbage — reads as the old Relay-only semantics.
let marked =
WebRTCStream::sdp_to_endpoint(r#"{"type":"offer","sdp":"v=0","ice_policy":"all"}"#);
assert!(WebRTCStream::endpoint_declares_all_ice(&marked));
let unmarked = WebRTCStream::sdp_to_endpoint(r#"{"type":"offer","sdp":"v=0"}"#);
assert!(!WebRTCStream::endpoint_declares_all_ice(&unmarked));
let wrong =
WebRTCStream::sdp_to_endpoint(r#"{"type":"offer","sdp":"v=0","ice_policy":"relay"}"#);
assert!(!WebRTCStream::endpoint_declares_all_ice(&wrong));
assert!(!WebRTCStream::endpoint_declares_all_ice(""));
assert!(!WebRTCStream::endpoint_declares_all_ice(
"webrtc://not-base64!"
));
assert!(!WebRTCStream::endpoint_declares_all_ice(
"https://example.com"
));
// The marker must be invisible to the plain RTCSessionDescription parse old peers do.
let sdp_json = WebRTCStream::get_remote_offer(&marked).unwrap();
serde_json::from_str::<
webrtc::peer_connection::sdp::session_description::RTCSessionDescription,
>(&sdp_json)
.expect("extra envelope key must not break RTCSessionDescription parsing");
}
/// The live description once it has gathered something. Polls instead of awaiting
/// `gathering_complete_promise`: a stream sits in the global `SESSIONS` for the whole wait, and
/// waiting out gathering there is long enough to trip `test_cancelled_new_does_not_leak_the_pc`
/// when the suite runs its tests in parallel. One candidate is all these tests need.
/// Returns `None` on timeout rather than panicking: the caller must get to `close()` before
/// it asserts, or a stranded pc fails `test_cancelled_new_does_not_leak_the_pc` as well and
/// one root failure is reported as two.
async fn first_gathered_description(stream: &WebRTCStream) -> Option<RTCSessionDescription> {
timeout(Duration::from_secs(10), async {
loop {
if let Some(desc) = stream.pc.local_description().await {
if desc.sdp.contains("a=candidate:") {
return desc;
}
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await
.ok()
}
/// The signalled endpoint and the live description must diverge. `pc.local_description()`
/// appends every candidate gathered so far; the endpoint the peer is handed must not grow with
/// them, because the rendezvous server delivers it to a UDP-registered peer as one datagram
/// and a fragmented one is dropped without a trace on paths that discard fragments.
#[tokio::test]
async fn test_stored_endpoint_does_not_grow_with_the_live_description() {
let offerer = WebRTCStream::new("", false, 20000).await.unwrap();
let live = first_gathered_description(&offerer).await;
let endpoint = offerer.local_endpoint().to_owned();
offerer.close().await;
// Some(_) is the assertion that the live description grew: the helper only returns once it
// carries a candidate, so the endpoint below really had something to grow by.
live.expect("no ICE candidate gathered within 10s");
assert!(
!WebRTCStream::get_remote_offer(&endpoint)
.unwrap()
.contains("a=candidate"),
"the signalled endpoint grew candidates alongside the live description"
);
assert!(
endpoint.len() <= WebRTCStream::UDP_ENDPOINT_BUDGET,
"signalled endpoint is {} bytes, over the {} byte budget",
endpoint.len(),
WebRTCStream::UDP_ENDPOINT_BUDGET
);
}
/// The split itself: candidates out, everything ICE and DTLS need to start in. A stripped
/// endpoint the peer cannot consume would trade a silent drop for a silent failure.
#[tokio::test]
async fn test_trickle_endpoint_splits_candidates_off() {
let offerer = WebRTCStream::new("", false, 20000).await.unwrap();
let gathered = first_gathered_description(&offerer).await;
offerer.close().await;
let gathered = gathered.expect("no ICE candidate gathered within 10s");
let endpoint = WebRTCStream::trickle_endpoint(&gathered, false).unwrap();
let json = WebRTCStream::get_remote_offer(&endpoint).unwrap();
assert!(!json.contains("a=candidate:"), "candidate survived the split");
assert!(
!json.contains("a=end-of-candidates"),
"end-of-candidates would stop the remote agent waiting for the trickle"
);
for keep in [
"m=application",
"a=ice-ufrag:",
"a=ice-pwd:",
"a=fingerprint:",
"a=setup:",
"a=mid:",
"a=sctp-port:",
] {
assert!(json.contains(keep), "trickle endpoint dropped {}", keep);
}
assert!(
endpoint.len() <= WebRTCStream::UDP_ENDPOINT_BUDGET,
"stripped endpoint is {} bytes",
endpoint.len()
);
// `parsed` is #[serde(skip)], so deserializing alone would accept any string as the SDP
// body. Unmarshal it, which is also what the peer does, and require the fingerprint the
// session key is derived from to have survived.
let desc: RTCSessionDescription =
serde_json::from_str(&json).expect("a stripped endpoint must deserialize");
WebRTCStream::get_key_for_sdp(&desc)
.expect("a stripped endpoint must still parse as SDP and keep its fingerprint");
// `encode_endpoint`, which `get_local_endpoint` also goes through, holds the opposite
// contract: carry whatever it is handed, candidates included. (What `get_local_endpoint`
// adds on top — the gathering wait — is covered by test_webrtc_loopback_gathered_endpoints.)
let kept = WebRTCStream::encode_endpoint(gathered.sdp_type, &gathered.sdp, false).unwrap();
assert!(
WebRTCStream::get_remote_offer(&kept)
.unwrap()
.contains("a=candidate:"),
"encode_endpoint dropped the candidates it was handed"
);
}
#[test]
fn test_webrtc_session_key() {
let mut sdp_str = "".to_owned();
assert_eq!(
WebRTCStream::get_key_for_sdp(
&RTCSessionDescription::offer(sdp_str).unwrap_or_default()
)
.unwrap_or_default(),
""
);
sdp_str = "\
v=0
o=- 7400546379179479477 208696200 IN IP4 0.0.0.0
s=-
t=0 0
a=fingerprint:sha-256 97:52:D6:1F:1E:87:6C:DA:B8:21:95:64:A5:85:89:FA:02:71:C7:4D:B3:FD:25:92:40:FB:6B:65:24:3C:79:88
a=group:BUNDLE 0
a=extmap-allow-mixed
m=application 9 UDP/DTLS/SCTP webrtc-datachannel
c=IN IP4 0.0.0.0
a=setup:actpass
a=mid:0
a=sendrecv
a=sctp-port:5000
a=ice-ufrag:RMWjjpXfpXbDPdMz
a=ice-pwd:BtIqlWHfwhsJdFiBROeLuEbNmYfHxRfT".to_owned();
assert_eq!(
WebRTCStream::get_key_for_sdp(
&RTCSessionDescription::offer(sdp_str).unwrap_or_default()
).unwrap_or_default(),
"sha-256 97:52:D6:1F:1E:87:6C:DA:B8:21:95:64:A5:85:89:FA:02:71:C7:4D:B3:FD:25:92:40:FB:6B:65:24:3C:79:88"
);
sdp_str = "\
v=0
o=- 7400546379179479477 208696200 IN IP4 0.0.0.0
s=-
t=0 0
a=group:BUNDLE 0
a=extmap-allow-mixed
m=application 9 UDP/DTLS/SCTP webrtc-datachannel
c=IN IP4 0.0.0.0
a=fingerprint:sha-256 97:52:D6:1F:1E:87:6C:DA:B8:21:95:64:A5:85:89:FA:02:71:C7:4D:B3:FD:25:92:40:FB:6B:65:24:3C:79:88
a=setup:actpass
a=mid:0
a=sendrecv
a=sctp-port:5000
a=ice-ufrag:RMWjjpXfpXbDPdMz
a=ice-pwd:BtIqlWHfwhsJdFiBROeLuEbNmYfHxRfT".to_owned();
assert_eq!(
WebRTCStream::get_key_for_sdp(
&RTCSessionDescription::offer(sdp_str).unwrap_or_default()
).unwrap_or_default(),
"sha-256 97:52:D6:1F:1E:87:6C:DA:B8:21:95:64:A5:85:89:FA:02:71:C7:4D:B3:FD:25:92:40:FB:6B:65:24:3C:79:88"
);
sdp_str = "\
v=0
o=- 7400546379179479477 208696200 IN IP4 0.0.0.0
s=-
t=0 0
a=group:BUNDLE 0
a=extmap-allow-mixed
m=application 9 UDP/DTLS/SCTP webrtc-datachannel
c=IN IP4 0.0.0.0
a=setup:actpass
a=mid:0
a=sendrecv
a=sctp-port:5000
a=ice-ufrag:RMWjjpXfpXbDPdMz
a=ice-pwd:BtIqlWHfwhsJdFiBROeLuEbNmYfHxRfT"
.to_owned();
assert!(
WebRTCStream::get_key_for_sdp(
&RTCSessionDescription::offer(sdp_str).unwrap_or_default()
)
.is_err(),
"can not find fingerprint attribute"
);
sdp_str = "\
v=0
o=- 7400546379179479477 208696200 IN IP4 0.0.0.0
s=-
t=0 0
a=group:BUNDLE 0
a=extmap-allow-mixed
m=audio 9 UDP/DTLS/SCTP webrtc-datachannel
c=IN IP4 0.0.0.0
a=fingerprint:sha-256 97:52:D6:1F:1E:87:6C:DA:B8:21:95:64:A5:85:89:FA:02:71:C7:4D:B3:FD:25:92:40:FB:6B:65:24:3C:79:88
a=setup:actpass
a=mid:0
a=sendrecv
a=sctp-port:5000
a=ice-ufrag:RMWjjpXfpXbDPdMz
a=ice-pwd:BtIqlWHfwhsJdFiBROeLuEbNmYfHxRfT".to_owned();
assert!(
WebRTCStream::get_key_for_sdp(
&RTCSessionDescription::offer(sdp_str).unwrap_or_default()
)
.is_err(),
"can not find datachannel fingerprint attribute"
);
assert!(
WebRTCStream::get_key_for_sdp(
&RTCSessionDescription::offer("".to_owned()).unwrap_or_default()
)
.is_err(),
"invalid sdp should error"
);
assert!(
WebRTCStream::get_key_for_sdp_json("{}").is_err(),
"empty sdp json should error"
);
assert!(
WebRTCStream::get_key_for_sdp_json("{ss}").is_err(),
"invalid sdp json should error"
);
let endpoint = "webrtc://eyJ0eXBlIjoiYW5zd2VyIiwic2RwIjoidj0wXHJcbm89LSA0MTA1NDk3NTY2NDgyMTQzODEwIDYwMzk1NzQw\
MCBJTiBJUDQgMC4wLjAuMFxyXG5zPS1cclxudD0wIDBcclxuYT1maW5nZXJwcmludDpzaGEtMjU2IDYxOjYwOjc0OjQwOjI4OkNFOjBCOjBDOjc1OjRCOj\
EwOjlBOkVFOjc3OkY1OjQ0OjU3Ojg0OjUxOkRCOjA0OjkyOjRBOjEwOjFDOjRFOjVGOjdFOkYxOkIzOjcxOjIyXHJcbmE9Z3JvdXA6QlVORExFIDBcclxu\
YT1leHRtYXAtYWxsb3ctbWl4ZWRcclxubT1hcHBsaWNhdGlvbiA5IFVEUC9EVExTL1NDVFAgd2VicnRjLWRhdGFjaGFubmVsXHJcbmM9SU4gSVA0IDAuMC\
4wLjBcclxuYT1zZXR1cDphY3RpdmVcclxuYT1taWQ6MFxyXG5hPXNlbmRyZWN2XHJcbmE9c2N0cC1wb3J0OjUwMDBcclxuYT1pY2UtdWZyYWc6SHlnU1Rr\
V2RsRlpHRG1XWlxyXG5hPWljZS1wd2Q6SkJneFZWaGZveVhHdHZha1VWcnBQeHVOSVpMU3llS1pcclxuYT1jYW5kaWRhdGU6OTYzOTg4MzQ4IDEgdWRwID\
IxMzA3MDY0MzEgMTkyLjE2OC4xLjIgNjQwMDcgdHlwIGhvc3RcclxuYT1jYW5kaWRhdGU6OTYzOTg4MzQ4IDIgdWRwIDIxMzA3MDY0MzEgMTkyLjE2OC4x\
LjIgNjQwMDcgdHlwIGhvc3RcclxuYT1jYW5kaWRhdGU6MTg2MTA0NTE5MCAxIHVkcCAxNjk0NDk4ODE1IDE0LjIxMi42OC4xMiAyNzAwNCB0eXAgc3JmbH\
ggcmFkZHIgMC4wLjAuMCBycG9ydCA2NDAwOFxyXG5hPWNhbmRpZGF0ZToxODYxMDQ1MTkwIDIgdWRwIDE2OTQ0OTg4MTUgMTQuMjEyLjY4LjEyIDI3MDA0\
IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNcclxuIn0=".to_owned();
assert_eq!(
WebRTCStream::get_key_for_sdp_json(
&WebRTCStream::get_remote_offer(&endpoint).unwrap_or_default()
).unwrap_or_default(),
"sha-256 61:60:74:40:28:CE:0B:0C:75:4B:10:9A:EE:77:F5:44:57:84:51:DB:04:92:4A:10:1C:4E:5F:7E:F1:B3:71:22"
);
}
#[tokio::test]
async fn test_webrtc_new_stream() {
let mut endpoint = "webrtc://sdfsdf".to_owned();
assert!(
WebRTCStream::new(&endpoint, false, 10000).await.is_err(),
"invalid webrtc endpoint should error"
);
endpoint = "wss://sdfsdf".to_owned();
assert!(
WebRTCStream::new(&endpoint, false, 10000).await.is_err(),
"invalid webrtc endpoint should error"
);
let stream = WebRTCStream::new("", false, 10000)
.await
.expect("local webrtc endpoint should ok");
// A raw WebRTCStream has no Drop: close it, or its session stays cached in SESSIONS
// and pollutes cross-test assertions about the cache.
stream.close().await;
endpoint = "webrtc://eyJ0eXBlIjoiYW5zd2VyIiwic2RwIjoidj0wXHJcbm89LSA0MTA1NDk3NTY2NDgyMTQzODEwIDYwMzk1NzQw\
MCBJTiBJUDQgMC4wLjAuMFxyXG5zPS1cclxudD0wIDBcclxuYT1maW5nZXJwcmludDpzaGEtMjU2IDYxOjYwOjc0OjQwOjI4OkNFOjBCOjBDOjc1OjRCOj\
EwOjlBOkVFOjc3OkY1OjQ0OjU3Ojg0OjUxOkRCOjA0OjkyOjRBOjEwOjFDOjRFOjVGOjdFOkYxOkIzOjcxOjIyXHJcbmE9Z3JvdXA6QlVORExFIDBcclxu\
YT1leHRtYXAtYWxsb3ctbWl4ZWRcclxubT1hcHBsaWNhdGlvbiA5IFVEUC9EVExTL1NDVFAgd2VicnRjLWRhdGFjaGFubmVsXHJcbmM9SU4gSVA0IDAuMC\
4wLjBcclxuYT1zZXR1cDphY3RpdmVcclxuYT1taWQ6MFxyXG5hPXNlbmRyZWN2XHJcbmE9c2N0cC1wb3J0OjUwMDBcclxuYT1pY2UtdWZyYWc6SHlnU1Rr\
V2RsRlpHRG1XWlxyXG5hPWljZS1wd2Q6SkJneFZWaGZveVhHdHZha1VWcnBQeHVOSVpMU3llS1pcclxuYT1jYW5kaWRhdGU6OTYzOTg4MzQ4IDEgdWRwID\
IxMzA3MDY0MzEgMTkyLjE2OC4xLjIgNjQwMDcgdHlwIGhvc3RcclxuYT1jYW5kaWRhdGU6OTYzOTg4MzQ4IDIgdWRwIDIxMzA3MDY0MzEgMTkyLjE2OC4x\
LjIgNjQwMDcgdHlwIGhvc3RcclxuYT1jYW5kaWRhdGU6MTg2MTA0NTE5MCAxIHVkcCAxNjk0NDk4ODE1IDE0LjIxMi42OC4xMiAyNzAwNCB0eXAgc3JmbH\
ggcmFkZHIgMC4wLjAuMCBycG9ydCA2NDAwOFxyXG5hPWNhbmRpZGF0ZToxODYxMDQ1MTkwIDIgdWRwIDE2OTQ0OTg4MTUgMTQuMjEyLjY4LjEyIDI3MDA0\
IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNcclxuIn0=".to_owned();
assert!(
WebRTCStream::new(&endpoint, false, 10000).await.is_err(),
"connect to an 'answer' webrtc endpoint should error"
);
}
#[tokio::test]
async fn test_reusing_cached_session_does_not_deadlock_during_close() {
let key = "test-reusing-cached-session".to_owned();
let cached = new_test_stream().await;
let candidate = new_test_stream().await;
let close_callback_completed = Arc::new(AtomicBool::new(false));
let close_callback_completed2 = close_callback_completed.clone();
candidate
.pc
.on_peer_connection_state_change(Box::new(move |state| {
let close_callback_completed = close_callback_completed2.clone();
Box::pin(async move {
if state == RTCPeerConnectionState::Closed {
let _sessions_lock = SESSIONS.lock().await;
close_callback_completed.store(true, Ordering::SeqCst);
}
})
}));
SESSIONS.lock().await.insert(key.clone(), cached.clone());
let reused = timeout(
Duration::from_secs(5),
WebRTCStream::cache_or_reuse_session(key.clone(), candidate),
)
.await
.expect("closing a duplicate peer deadlocked on the session cache");
assert!(Arc::ptr_eq(&reused.pc, &cached.pc));
assert!(close_callback_completed.load(Ordering::SeqCst));
SESSIONS.lock().await.remove(&key);
cached.pc.close().await.ok();
}
#[tokio::test]
async fn test_session_cleanup_only_removes_matching_peer() {
let key = "test-session-cleanup".to_owned();
let old_key = "test-session-cleanup-old".to_owned();
let old = new_test_stream().await;
let replacement = new_test_stream().await;
let mut sessions = HashMap::new();
sessions.insert(key.clone(), replacement.clone());
assert!(WebRTCStream::remove_sessions_for_peer(&mut sessions, &old.pc).is_empty());
assert!(Arc::ptr_eq(
&sessions.get(&key).unwrap().pc,
&replacement.pc,
));
sessions.insert(old_key.clone(), old.clone());
assert_eq!(
WebRTCStream::remove_sessions_for_peer(&mut sessions, &old.pc),
vec![old_key.clone()],
);
assert!(!sessions.contains_key(&old_key));
assert!(Arc::ptr_eq(
&sessions.get(&key).unwrap().pc,
&replacement.pc,
));
assert_eq!(
WebRTCStream::remove_sessions_for_peer(&mut sessions, &replacement.pc),
vec![key.clone()],
);
assert!(!sessions.contains_key(&key));
old.pc.close().await.ok();
replacement.pc.close().await.ok();
}
#[tokio::test]
async fn test_webrtc_wait_connected_timeout() {
let mut stream = WebRTCStream::new("", false, 100).await.unwrap();
let err = stream.wait_connected(10).await.unwrap_err();
assert!(err.to_string().contains("timeout"));
stream.close().await; // see test_webrtc_new_stream: no Drop on a raw WebRTCStream
}
async fn connect_loopback() -> (WebRTCStream, WebRTCStream) {
let mut offerer = WebRTCStream::new("", false, 20000).await.unwrap();
let offer = offerer.local_endpoint().to_owned();
let answerer = WebRTCStream::new(&offer, false, 20000).await.unwrap();
let answer = answerer.local_endpoint().to_owned();
offerer.set_remote_endpoint(&answer).await.unwrap();
// Bridge trickle candidates directly between the two peers, both directions.
let mut off_ice = offerer.take_local_ice_rx().unwrap();
let mut ans_ice = answerer.take_local_ice_rx().unwrap();
let answerer_for_ice = answerer.clone();
let offerer_for_ice = offerer.clone();
tokio::spawn(async move {
while let Some(c) = off_ice.recv().await {
let _ = answerer_for_ice.add_remote_ice_candidate(&c).await;
}
});
tokio::spawn(async move {
while let Some(c) = ans_ice.recv().await {
let _ = offerer_for_ice.add_remote_ice_candidate(&c).await;
}
});
offerer.wait_connected(20000).await.unwrap();
let mut answerer = answerer;
answerer.wait_connected(20000).await.unwrap();
(offerer, answerer)
}
// next()'s teardown must not be cancellable: an awaited close that loses the select! race
// against a consumer's timer strands the pc (is_closed already latched) and its SESSIONS
// entry. `close_detached` is non-async, so what needs proving is that the handoff completes.
#[tokio::test]
async fn test_close_detached_completes_without_the_caller() {
let (offerer, answerer) = connect_loopback().await;
let key = format!("offer:{}", offerer.session_key());
assert!(
SESSIONS.lock().await.contains_key(&key),
"offerer should be cached while live"
);
offerer.close_detached();
drop(offerer);
for _ in 0..200 {
if !SESSIONS.lock().await.contains_key(&key) {
answerer.close().await;
return;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
panic!("detached close never evicted the peer connection from SESSIONS");
}
// Whole messages are split out of the read buffer instead of copied, so the buffer is
// consumed from the front and periodically re-initialized. Alternating whole and fragmented
// messages exercises that against the accumulator path, and running well past one refill
// exercises the refill itself — a boundary the fixed-size buffer never had.
#[tokio::test]
async fn test_webrtc_mixed_message_sizes_survive_scratch_refill() {
let (mut offerer, mut answerer) = connect_loopback().await;
// Sized either side of MAX_FRAGMENT_PAYLOAD, and enough rounds to consume several
// SCRATCH_REFILL chunks.
let sizes = [1usize, 64, 60_000, 60_001, 130_000, 7];
for round in 0..40u8 {
for &len in &sizes {
let payload = Bytes::from(vec![round; len]);
offerer.send_bytes(payload.clone()).await.unwrap();
let got = timeout(Duration::from_secs(10), answerer.next())
.await
.expect("receiver starved")
.expect("stream ended")
.expect("read failed");
assert_eq!(got.len(), len, "round {round}, len {len}");
assert!(
got.iter().all(|&b| b == round),
"round {}, len {}: content or boundary corrupted",
round,
len
);
}
}
offerer.close().await;
answerer.close().await;
}
// Owning the Stream owns the peer connection: dropping it must close the pc and evict the
// session, without the owner having to remember to. Every exit path used to carry that
// obligation, and the controlled side never honoured it.
#[tokio::test]
async fn dropping_the_stream_closes_the_peer_connection() {
let offerer = WebRTCStream::new("", false, 20000).await.unwrap();
let key = format!("offer:{}", offerer.session_key());
assert!(
SESSIONS.lock().await.contains_key(&key),
"offerer should be cached while live"
);
// No explicit close anywhere: the stream simply goes out of scope, as it does on the
// exits that forget.
drop(crate::Stream::WebRTC(offerer));
for _ in 0..200 {
if !SESSIONS.lock().await.contains_key(&key) {
return;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
panic!("dropping the stream left the peer connection in SESSIONS");
}
// A replayed offer asking for a different ICE policy must not be handed the cached peer
// connection. The lookup and the insert-time duplicate check read the same key, so the test
// fails if either one stops applying `is_reusable_for` — which is how the guard was dead
// code: the lookup rejected the entry, and the insert handed back the very same one.
#[tokio::test]
async fn test_cached_peer_is_not_reused_across_ice_policies() {
let offerer = WebRTCStream::new("", false, 20000).await.unwrap();
let offer = offerer.local_endpoint().to_owned();
let all_ice = WebRTCStream::new(&offer, false, 20000).await.unwrap();
assert!(!all_ice.relay_only);
let relay_only = WebRTCStream::new(&offer, true, 20000).await.unwrap();
assert!(
relay_only.relay_only,
"a Relay-only request was answered with the cached All-policy peer connection"
);
relay_only.close().await;
all_ice.close().await;
offerer.close().await;
}
// The local-candidate channel must close when the pc does. Its only sender lives inside the
// on_ice_candidate handler, which close() does not clear, so without the teardown the
// receiver stays open forever — and the forwarder loop this API asks callers to write holds
// a stream clone while parked on recv(), keeping the pc alive past its own close.
#[tokio::test]
async fn test_local_ice_channel_closes_with_the_peer_connection() {
let offerer = WebRTCStream::new("", false, 20000).await.unwrap();
let mut ice_rx = offerer.take_local_ice_rx().unwrap();
// Exactly the loop every consumer writes, holding a clone of the stream.
let forwarder_stream = offerer.clone();
let forwarder = tokio::spawn(async move {
while let Some(candidate) = ice_rx.recv().await {
let _ = forwarder_stream.add_remote_ice_candidate(&candidate).await;
}
});
offerer.close().await;
drop(offerer);
tokio::time::timeout(Duration::from_secs(20), forwarder)
.await
.expect("forwarder task outlived the peer connection")
.unwrap();
}
// Extra channels must not displace the bound one: the newcomer's on_open would push Open onto
// the watch that gates send and recv, re-arming a session already latched Closed. Only the
// bind-once guard is exercised; the ordered+reliable check cannot decide anything here.
#[tokio::test]
async fn test_webrtc_answerer_binds_only_the_first_data_channel() {
use webrtc::data_channel::data_channel_init::RTCDataChannelInit;
let mut offerer = WebRTCStream::new("", false, 20000).await.unwrap();
// Replace the bootstrap channel's siblings: an unordered one and a duplicate.
let unordered = offerer
.pc
.create_data_channel(
"unordered",
Some(RTCDataChannelInit {
ordered: Some(false),
..Default::default()
}),
)
.await
.unwrap();
let duplicate = offerer
.pc
.create_data_channel("duplicate", None)
.await
.unwrap();
let offer = offerer.local_endpoint().to_owned();
let answerer = WebRTCStream::new(&offer, false, 20000).await.unwrap();
let answer = answerer.local_endpoint().to_owned();
offerer.set_remote_endpoint(&answer).await.unwrap();
let mut off_ice = offerer.take_local_ice_rx().unwrap();
let mut ans_ice = answerer.take_local_ice_rx().unwrap();
let answerer_for_ice = answerer.clone();
let offerer_for_ice = offerer.clone();
tokio::spawn(async move {
while let Some(c) = off_ice.recv().await {
let _ = answerer_for_ice.add_remote_ice_candidate(&c).await;
}
});
tokio::spawn(async move {
while let Some(c) = ans_ice.recv().await {
let _ = offerer_for_ice.add_remote_ice_candidate(&c).await;
}
});
offerer.wait_connected(20000).await.unwrap();
let mut answerer = answerer;
answerer.wait_connected(20000).await.unwrap();
// The bootstrap channel still carries data: the rejected siblings did not displace it.
let payload = Bytes::from_static(b"bootstrap still bound");
offerer.send_bytes(payload.clone()).await.unwrap();
let got = tokio::time::timeout(Duration::from_secs(10), answerer.next())
.await
.expect("answerer starved")
.expect("stream ended")
.expect("read failed");
assert_eq!(&got[..], &payload[..]);
drop(unordered);
drop(duplicate);
offerer.close().await;
answerer.close().await;
}
// One-shot callers exchange only the endpoints and never consume `take_local_ice_rx`.
#[tokio::test]
async fn test_webrtc_loopback_gathered_endpoints() {
let connect = async {
let mut offerer = WebRTCStream::new("", false, 20000).await.unwrap();
let offer = offerer.get_local_endpoint().await.unwrap();
let mut answerer = WebRTCStream::new(&offer, false, 20000).await.unwrap();
let answer = answerer.get_local_endpoint().await.unwrap();
offerer.set_remote_endpoint(&answer).await.unwrap();
offerer.wait_connected(20000).await.unwrap();
answerer.wait_connected(20000).await.unwrap();
offerer.close().await;
answerer.close().await;
};
timeout(Duration::from_secs(40), connect)
.await
.expect("gathered-endpoint WebRTC loopback did not complete in time");
}
// In-process offerer<->answerer loopback exercising the send/next data plane that the framing,
// empty-message, and EOF fixes live in. Connects over host candidates (works offline; any
// configured/default STUN just fails in the background without blocking the host pair).
#[tokio::test]
async fn test_webrtc_loopback_roundtrip() {
let connect = async {
let (mut offerer, mut answerer) = connect_loopback().await;
// Host-candidate loopback is direct, never TURN-relayed.
assert_eq!(offerer.is_relayed().await, Some(false));
// Small message.
offerer.send_raw(b"hello".to_vec()).await.unwrap();
let got = answerer.next().await.unwrap().unwrap();
assert_eq!(&got[..], b"hello");
// Empty message: must round-trip as an empty frame, not be seen as EOF.
offerer.send_raw(Vec::new()).await.unwrap();
let got = answerer.next().await.unwrap().unwrap();
assert_eq!(got.len(), 0, "empty message must not be treated as EOF");
// Payload far above the 64KB single-message cap: must be fragmented and reassembled.
let big = vec![0xABu8; 200_000];
offerer.send_raw(big.clone()).await.unwrap();
let got = answerer.next().await.unwrap().unwrap();
assert_eq!(
got.len(),
big.len(),
"large message must survive fragmentation"
);
assert_eq!(&got[..], &big[..]);
// Reverse direction.
answerer.send_raw(b"world".to_vec()).await.unwrap();
let got = offerer.next().await.unwrap().unwrap();
assert_eq!(&got[..], b"world");
// Peer close: the other side observes a clean EOF (None) or a close error, never a hang.
offerer.close().await;
match timeout(Duration::from_secs(10), answerer.next()).await {
Ok(None) | Ok(Some(Err(_))) => {}
Ok(Some(Ok(b))) => panic!("expected EOF after peer close, got {} bytes", b.len()),
Err(_) => panic!("answerer.next() hung after peer close"),
}
answerer.close().await;
};
timeout(Duration::from_secs(40), connect)
.await
.expect("webrtc loopback did not complete in time");
}
// A client session runs on its own `#[tokio::main(flavor = "current_thread")]` runtime that
// is dropped the moment io_loop returns, so a close spawned or awaited there can be killed
// before (or worse, after) `close()` latches `is_closed`. Pin the production sequence over
// `Stream::WebRTC`: `close_webrtc()` from a thread with no runtime at all, then Drop; the
// peer must still see EOF — the closer thread, not any caller runtime, delivers it.
#[tokio::test(flavor = "multi_thread")]
async fn test_session_end_close_reaches_the_peer() {
let (offerer, mut answerer) = connect_loopback().await;
let stream = crate::Stream::WebRTC(offerer);
tokio::task::spawn_blocking(move || {
stream.close_webrtc();
drop(stream);
})
.await
.expect("session thread");
match timeout(Duration::from_secs(5), answerer.next()).await {
Ok(None) | Ok(Some(Err(_))) => {}
Ok(Some(Ok(b))) => panic!("expected EOF after session close, got {} bytes", b.len()),
Err(_) => panic!("peer never observed the session-end close"),
}
answerer.close().await;
}
// The shared count, not any one handoff, decides: with a second claim outstanding on a pc
// no one has adopted, dropping the first must leave it open for that caller, and dropping
// the last must close it — an unanswered offerer never reaches a terminal ICE state itself.
#[tokio::test(flavor = "multi_thread")]
async fn test_last_abandoned_claim_closes_an_unadopted_pc() {
let builder = WEBRTC_RT
.as_ref()
.expect("WebRTC I/O runtime")
.spawn(WebRTCStream::new_inner(String::new(), false, 20000))
.await
.expect("setup task")
.expect("offerer builds");
let stream = builder
.stream
.as_ref()
.expect("a fresh handoff carries its stream")
.clone();
let key = format!("offer:{}", stream.session_key());
let hit = NewStreamHandoff::claim(stream);
drop(builder);
tokio::time::sleep(Duration::from_millis(300)).await;
assert!(
SESSIONS.lock().await.contains_key(&key),
"a claim was still outstanding, so the pc had to stay open"
);
drop(hit);
for _ in 0..200 {
if !SESSIONS.lock().await.contains_key(&key) {
return;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
panic!("the last abandoned claim never closed the unadopted pc");
}
// `new_inner` caches the pc BEFORE its own `new()` returns, so a concurrent caller can hit
// that cache and adopt the stream while the builder's handoff is still outstanding. Closing
// on the builder's Drop then tears down a connection another caller is already using, so
// abandonment has to be a property of the pc, not of one handoff.
#[tokio::test(flavor = "multi_thread")]
async fn test_abandoned_builder_handoff_spares_an_adopted_pc() {
// Exactly what `new()` does, minus the adoption: the setup task runs on WEBRTC_RT and
// hands back a fresh, still-unadopted handoff.
let builder = WEBRTC_RT
.as_ref()
.expect("WebRTC I/O runtime")
.spawn(WebRTCStream::new_inner(String::new(), false, 20000))
.await
.expect("setup task")
.expect("offerer builds");
let stream = builder
.stream
.as_ref()
.expect("a fresh handoff carries its stream")
.clone();
let key = format!("offer:{}", stream.session_key());
assert!(
SESSIONS.lock().await.contains_key(&key),
"new_inner must cache the pc it built"
);
// The concurrent `new()`: a cache hit that takes the stream and goes on using it.
let adopted = NewStreamHandoff::claim(stream)
.into_inner()
.expect("a cache hit hands the stream back");
// ...and now the builder's `new()` is cancelled, so the runtime drops its handoff.
drop(builder);
tokio::time::sleep(Duration::from_millis(500)).await;
assert!(
SESSIONS.lock().await.contains_key(&key),
"abandoning the builder's handoff closed a pc another caller had already adopted"
);
adopted.close_detached();
}
// End-to-end: cancel `new()` itself after its first poll (the spawn is already in flight)
// and every session it transiently created must be closed and evicted again.
//
// The cancelled attempt's own key is unknowable here — its DTLS fingerprint is generated
// inside the task that was abandoned — so what is watched for is a key that OUTLASTS the
// window, not an instant with no new keys at all. Concurrent tests each close what they
// create, so their keys come and go and drop out of the running intersection; a leaked pc
// never does. Waiting for an empty difference instead made this test depend on the suite
// having an idle moment, which `--test-threads=2` never gives it: one lane is this test for
// the whole wait while the other keeps starting sessions.
#[tokio::test(flavor = "multi_thread")]
async fn test_cancelled_new_does_not_leak_the_pc() {
use std::collections::HashSet;
let before: HashSet<String> = SESSIONS.lock().await.keys().cloned().collect();
// Zero timeout: polls the future exactly once (spawning new_inner), then cancels it —
// usually. The setup task runs on its own runtime and can finish inside that single poll,
// and then `new()` hands back a live stream instead. `WebRTCStream` has no `Drop`, so
// discarding that one is itself a leak, and this test would go on to report it as the
// cancelled attempt's. Close what comes back and try for a real cancellation. Fewer test
// threads make the setup task likelier to win, which is why this surfaced under
// `--test-threads=2` and not at the default.
let mut cancelled = false;
for _ in 0..20 {
match timeout(Duration::ZERO, WebRTCStream::new("", false, 20000)).await {
Err(_) => {
cancelled = true;
break;
}
Ok(Ok(stream)) => stream.close().await,
Ok(Err(_)) => {}
}
}
assert!(cancelled, "new() never lost the race with its own cancellation");
// Let the detached setup task finish (and insert its session) before sampling, or the
// first sample is taken before the leak has formed and every later one intersects to
// nothing.
tokio::time::sleep(Duration::from_millis(500)).await;
const ATTEMPTS: usize = 600; // 30s
let mut persisted: Option<HashSet<String>> = None;
for attempt in 1..=ATTEMPTS {
let now: HashSet<String> = SESSIONS.lock().await.keys().cloned().collect();
let new_keys: HashSet<String> = now.difference(&before).cloned().collect();
persisted = Some(match persisted {
None => new_keys,
Some(prev) => prev.intersection(&new_keys).cloned().collect(),
});
let persisted_keys = persisted.as_ref().map_or(0, HashSet::len);
if persisted_keys == 0 {
return;
}
// Positional, not an inline `{persisted:?}`: on edition 2018 a lone-literal `panic!`
// does not go through format_args and would print the placeholder verbatim.
assert!(
attempt < ATTEMPTS,
"cancelled new() left a pc cached in SESSIONS (persisted: {:?})",
persisted
);
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
// The production shape of a controller session end: the offerer was created on a session's
// own current-thread runtime, the session queues its detached close and the runtime is
// destroyed immediately after (io_loop returning drops it). The pc's sockets and its
// ICE/DTLS/SCTP pump tasks must not die with that runtime, or the queued close completes
// without ever putting close_notify on the wire and the peer waits out ICE decay. Gathered
// (non-trickle) endpoints keep the cross-runtime signaling to two string handoffs.
#[tokio::test(flavor = "multi_thread")]
async fn test_close_survives_creator_runtime_destruction() {
// tokio oneshots for every handoff: their sends are synchronous, and the session side
// awaits them INSIDE block_on — a current_thread runtime only drives its tasks while
// being block_on-driven, and the offerer's SCTP/ICE pumps must stay live until the
// answerer has confirmed the channel is open.
let (offer_tx, offer_rx) = tokio::sync::oneshot::channel::<String>();
let (answer_tx, answer_rx) = tokio::sync::oneshot::channel::<String>();
let (go_tx, go_rx) = tokio::sync::oneshot::channel::<()>();
let session = std::thread::spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("session runtime");
let offerer = rt.block_on(async {
let mut offerer = WebRTCStream::new("", false, 20000).await.unwrap();
offer_tx
.send(offerer.get_local_endpoint().await.unwrap())
.unwrap();
let answer = answer_rx.await.unwrap();
offerer.set_remote_endpoint(&answer).await.unwrap();
offerer.wait_connected(20000).await.unwrap();
// Keep driving the runtime until the answerer confirms open, so the EOF
// assertion — not connection setup — is what discriminates.
go_rx.await.unwrap();
offerer
});
// The session ends: queue the close, then destroy the runtime the pc was created
// on — exactly what io_loop returning does.
drop(crate::Stream::WebRTC(offerer));
drop(rt);
});
let offer = offer_rx.await.expect("offer from session thread");
let mut answerer = WebRTCStream::new(&offer, false, 20000).await.unwrap();
answer_tx
.send(answerer.get_local_endpoint().await.unwrap())
.expect("answer to session thread");
answerer.wait_connected(20000).await.unwrap();
go_tx.send(()).expect("release session thread");
session.join().expect("session thread");
match timeout(Duration::from_secs(5), answerer.next()).await {
Ok(None) | Ok(Some(Err(_))) => {}
Ok(Some(Ok(b))) => panic!("expected EOF after session close, got {} bytes", b.len()),
Err(_) => panic!("peer never observed the close after the creator runtime died"),
}
answerer.close().await;
}
// The peer-initiated end: `next()` hits EOF and fires its own `close_detached` before the
// session's final close (here via `Stream`'s Drop) is issued. Both land on the serialized
// closer thread, so the first runs to completion and the second is a benign no-op — the pc
// must end up closed and evicted, never stranded by one closer cancelling the other.
#[tokio::test(flavor = "multi_thread")]
async fn test_eof_close_then_drop_still_evicts() {
let (mut offerer, answerer) = connect_loopback().await;
let key = format!("offer:{}", offerer.session_key());
answerer.close().await;
match timeout(Duration::from_secs(5), offerer.next()).await {
Ok(None) | Ok(Some(Err(_))) => {} // EOF path fired close_detached internally
Ok(Some(Ok(b))) => panic!("expected EOF after peer close, got {} bytes", b.len()),
Err(_) => panic!("offerer.next() hung after peer close"),
}
drop(crate::Stream::WebRTC(offerer)); // second close via Drop
for _ in 0..200 {
if !SESSIONS.lock().await.contains_key(&key) {
return;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
panic!("pc still cached after EOF close + Drop close");
}
// Both framing violations must end the stream. The empty-FRAG_MORE case is the one that
// cannot be caught downstream: it adds nothing to the accumulator, so the MAX_FRAME_LENGTH
// cap never trips and `next()` would otherwise spin for as long as the peer keeps writing.
// The bad frame is injected mid-message so the branch's `acc` reset is exercised too.
#[tokio::test]
async fn test_webrtc_rejects_bad_fragment_framing() {
// (raw frame, expected substring)
let cases: [(&[u8], &str); 2] = [
(&[0x2A, b'x'], "fragment header 42"),
(&[FRAG_MORE], "carries no payload"),
];
for (frame, want) in cases {
let connect = async {
let (offerer, mut answerer) = connect_loopback().await;
let dc = offerer.detached_dc().await.unwrap();
// Leave a partial message in the accumulator first, so the rejection has
// something to discard. `send_bytes` only ever emits valid headers, so the bad
// frame itself has to be written through the raw channel below it.
let mut lead = BytesMut::with_capacity(1 + 8);
lead.put_u8(FRAG_MORE);
lead.put_slice(b"leading!");
dc.write(&lead.freeze()).await.unwrap();
dc.write(&bytes::Bytes::copy_from_slice(frame))
.await
.unwrap();
let err = answerer
.next()
.await
.expect("bad framing must surface as an error, not EOF")
.expect_err("bad framing must be rejected");
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert!(err.to_string().contains(want), "unexpected error: {}", err);
offerer.close().await;
answerer.close().await;
};
timeout(Duration::from_secs(40), connect)
.await
.expect("webrtc loopback did not complete in time");
}
}
#[tokio::test]
async fn test_webrtc_concurrent_large_sends_preserve_boundaries() {
let connect = async {
let (offerer, mut answerer) = connect_loopback().await;
let mut sender_a = offerer.clone();
let mut sender_b = offerer.clone();
let expected_a = vec![0xAA; 200_000];
let expected_b = vec![0xBB; 200_000];
let payload_a = expected_a.clone();
let payload_b = expected_b.clone();
let barrier = Arc::new(Barrier::new(3));
let barrier_a = barrier.clone();
let send_a = tokio::spawn(async move {
barrier_a.wait().await;
sender_a.send_raw(payload_a).await
});
let barrier_b = barrier.clone();
let send_b = tokio::spawn(async move {
barrier_b.wait().await;
sender_b.send_raw(payload_b).await
});
barrier.wait().await;
let receive = async {
let first = answerer.next().await.unwrap().unwrap();
let second = answerer.next().await.unwrap().unwrap();
(first, second)
};
let (send_a, send_b, (first, second)) = tokio::join!(send_a, send_b, receive);
send_a.unwrap().unwrap();
send_b.unwrap().unwrap();
let boundaries_preserved = (first.as_ref() == expected_a.as_slice()
&& second.as_ref() == expected_b.as_slice())
|| (first.as_ref() == expected_b.as_slice()
&& second.as_ref() == expected_a.as_slice());
assert!(boundaries_preserved, "concurrent messages were interleaved");
offerer.close().await;
answerer.close().await;
};
timeout(Duration::from_secs(40), connect)
.await
.expect("concurrent WebRTC sends did not complete in time");
}
}
Reference in New Issue View Git Blame Copy Permalink