rustdesk/hbb_common
1
0
Fork 0
mirror of https://github.com/rustdesk/hbb_common.git synced 2026-08-29 05:21:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

webrtc: trim the comments to AGENTS.md length

Browse Source 
435 comment lines to 328, and the module header from 40 to 17. What went
is what the rules say does not belong in the source: past-bug narration,
rejected alternatives, measurements, and restatements of the adjacent
code. What stays is the why a reader cannot derive locally - the
single-reader invariant on the reassembly state, the send-permit handoff,
why the close must be detached and why that lock is held across an await,
and the webrtc-rs behaviours this module depends on.

Comments only; no code changed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ExUfAkYbq8UC9pQCiLy8TQ
This commit is contained in:
rustdesk
2026-08-09 22:38:15 +08:00
parent 2cb8d0c7d8
commit 5897012949
1 changed files with 67 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/webrtc.rs
+67 -197
View File
@@ -1,43 +1,20 @@
//! WebRTC transport for RustDesk streams. //! WebRTC transport for RustDesk streams.
//! //!
//! # webrtc crate upgrade checklist //! # Bumping the webrtc crate
//! //!
//! The webrtc crate version is MSRV-pinned in Cargo.toml (see the comment there). Beyond plain //! This module depends on webrtc-rs internals its public API does not promise. Re-verify each
//! API compatibility, this module relies on webrtc-rs *internals* that its public API does not //! against the new sources (last checked: webrtc 0.13, -data 0.11, -sctp 0.12), then run
//! guarantee. All of them were verified against webrtc 0.13 (webrtc-data 0.11, webrtc-sctp 0.12); //! `cargo test --features webrtc webrtc::tests`.
//! re-verify each against the new crate sources when bumping:
//! //!
//! - **Send backpressure is bounded**: `data::DataChannel::write` PARKS when webrtc-sctp's //! - `DataChannel::write` parks on a full PendingQueue instead of buffering without bound.
//! PendingQueue is full (byte-counting semaphore, `QUEUE_BYTES_LIMIT` = 128 KiB; permits return //! - SCTP max message size is 65536: `MAX_FRAGMENT_PAYLOAD` + 1 must stay below it.
//! as chunks drain) and inflight data is cwnd/rwnd-capped (peer default rwnd 1 MiB). //! - `read_sctp` does not await after dequeuing, so a dropped `next()` cannot lose a fragment.
//! `send_bytes` depends on this both for bounded memory on slow links and for its //! - `detach()` is an idempotent Arc clone that does not close on drop.
//! send_timeout-then-close semantics. If a new version buffers unboundedly instead, video can //! - `on_*` handlers live in the pc: one holding a strong `Arc<RTCPeerConnection>` leaks it.
//! OOM a slow session and the send timeout never fires. //! - `close()` latches `is_closed` before its first await, so a cancelled close is unretryable.
//! - **Max SCTP message size 65536**: `MAX_FRAGMENT_PAYLOAD` + 1 header byte must stay below it. //! - `Disconnected` is transient; only `Failed`/`Closed` are terminal.
//! - **The successful read path is cancel-safe**: `read_sctp` dequeues synchronously and returns //! - `RTCIceCandidatePair`'s `Display` layout — `is_relayed()` parses it because 0.13 keeps the
//! with no `.await` after it, and `read_data_channel` adds none on the user-data path. So //! candidates private. 0.17+ exposes them; switch and delete the parse.
//! `next_timeout`, which drops that future routinely, cannot lose a fragment — a version that
//! awaited after dequeuing would, undetectably, since the header carries no length or checksum.
//! Scope: the *read*. `read_data_channel` does await after dequeuing on its ErrShortBuffer and
//! DCEP branches, and `next()` itself awaits `pc.close()` on its error paths — and
//! `RTCPeerConnection::close` latches `is_closed` before its first await, so a cancelled close
//! silently makes every later close a no-op and leaves the pc in `SESSIONS`.
//! - **`detach()` is an idempotent Arc clone with no close-on-drop** (`detached_dc` caches it and
//! clones are shared across `WebRTCStream` clones).
//! - **`on_*` handlers are stored inside the pc**: a handler capturing a strong
//! `Arc<RTCPeerConnection>` forms an uncollectable cycle and leaks the pc permanently — see the
//! `Arc::downgrade` in `new()`; any newly added handler must follow it.
//! - **`Disconnected` peer-connection state is transient/recoverable** (ICE consent lapse);
//! only `Failed`/`Closed` are treated as terminal by the state handler.
//! - **`RTCIceCandidatePair`'s `Display` format**: its candidates are private in 0.13, so
//! `is_relayed()` reads the candidate type out of
//! `"(local) {protocol} {typ} {addr}:{port}{related} <-> (remote) ..."` by position. 0.17+
//! makes the fields `pub`; switch to them and delete the parse. (It cannot use the stats
//! report's `nominated` flag instead: webrtc-ice sets that per checklist entry and never
//! clears it, so several entries carry it after a pair switch.)
//!
//! Then re-run the loopback tests at the bottom of this file (`cargo test --features webrtc
//! webrtc::tests`).
use std::collections::HashMap; use std::collections::HashMap;
use std::io::{Error, ErrorKind}; use std::io::{Error, ErrorKind};
@@ -94,12 +71,9 @@ pub struct WebRTCStream {
// Serialize a complete logical message across clones. Each fragment is a separate SCTP // 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. // message, so serializing only individual writes would allow two large messages to interleave.
send_gate: Arc<Semaphore>, send_gate: Arc<Semaphore>,
// Receive-side reassembly state, behind the Arc rather than in the future so a cancelled // Behind the Arc, not in the future, so a cancelled `next()` keeps the partial message.
// `next()` does not lose the partial message. SINGLE READER ONLY: reassembly spans calls, and // SINGLE READER ONLY: reassembly spans calls and the fragment header carries no length or
// the fragment header carries no length or sequence number, so two readers splice unrelated // sequence number, so two readers splice unrelated fragments into one wrong-but-valid message.
// fragments into one `acc` and produce a well-formed, undetectably wrong message. Every
// consumer today reads from one task; keep it that way (this is why no API hands out a clone
// that outlives its `Stream`).
recv_state: Arc<Mutex<RecvState>>, recv_state: Arc<Mutex<RecvState>>,
// True once the controller has completed the RustDesk identity binding (DTLS fingerprint // 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 // matched to the signed peer id, via `set_key`). DTLS always encrypts; this flag mirrors TCP's
@@ -136,20 +110,9 @@ const SCRATCH_REFILL: usize = 4 * RECV_BUF_SIZE;
const FRAG_MORE: u8 = 1; const FRAG_MORE: u8 = 1;
/// Fragment header byte: final (or only) fragment of a logical message. /// Fragment header byte: final (or only) fragment of a logical message.
const FRAG_END: u8 = 0; const FRAG_END: u8 = 0;
/// Largest logical message this transport will send or reassemble. /// 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
/// Held at parity with TCP on purpose. A transport-specific ceiling would be a trap: the same /// moves between both paths, so a transport-specific ceiling would kill it on the other one.
/// session moves between WebRTC and the relay, so a message under TCP's limit but over this one
/// would work on one path and kill the connection on the other — and it would bite hardest on
/// exactly the large messages (a keyframe, a big clipboard image) that a direct path exists to
/// carry. Whatever the right maximum is, both transports need the same one.
///
/// That leaves a real exposure, shared with TCP and not created here: the cap is the memory an
/// unauthenticated peer can make a receiver hold, and the answerer runs before any password
/// check. TCP is no better off — `BytesCodec::new` leaves `max_packet_length` at `usize::MAX`,
/// so its only bound is the 30-bit length field in the frame header. Tightening it belongs in
/// one change across both paths, with a number measured against real traffic rather than
/// guessed; until then this at least holds the peak to the cap instead of twice it.
const MAX_RECV_MESSAGE: usize = crate::bytes_codec::MAX_FRAME_LENGTH; 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 // 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 // https://stackoverflow.com/questions/72805316/determine-nat-mapping-behaviour-using-two-stun-servers
@@ -257,15 +220,8 @@ impl WebRTCStream {
} }
/// Reject a data channel from inside `on_data_channel`, where closing it directly does not /// Reject a data channel from inside `on_data_channel`, where closing it directly does not
/// work. /// 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.
/// webrtc-rs runs this handler to completion BEFORE `handle_open` binds the SCTP stream, so
/// at this point `close()` only flips the ready state and returns — no stream reset — and
/// `handle_open` then sets it back to Open. With detached channels no read loop is spawned
/// either, so a channel refused here would stay open and undrained, and its queued bytes
/// count against the association-wide receive window: about a megabyte written into an
/// ignored channel stalls the one carrying the session. Close from the channel's own
/// `on_open` instead, which runs once the stream exists.
fn close_unbound_channel(dc: Arc<RTCDataChannel>) { fn close_unbound_channel(dc: Arc<RTCDataChannel>) {
let dc_for_close = dc.clone(); let dc_for_close = dc.clone();
dc.on_open(Box::new(move || { dc.on_open(Box::new(move || {
@@ -278,17 +234,9 @@ impl WebRTCStream {
})); }));
} }
/// Whether a `SESSIONS` entry may be handed to a caller asking for `force_relay`. /// 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
/// A hit returns a pc built for the FIRST caller, and two of its properties belong to this /// evicts late enough that dead entries linger. Both the lookup and the insert must use it.
/// caller instead: the ICE policy (`rendezvous_mediator` recomputes relay-only per PunchHole,
/// so a replayed offer wanting Relay-only must not get an All-policy pc free to pick a direct
/// pair), and liveness (the state handler latches Closed and closes the stream before it
/// reaches SESSIONS to evict, so the map briefly holds dead entries). `peer_verified` stays
/// shared on purpose: it is a fact about the DTLS certificate the entry is keyed by.
///
/// Both the lookup and the insert-time duplicate check must use this — they read the same key,
/// so a test applied at only one of them is not applied at all.
fn is_reusable_for(&self, force_relay: bool) -> bool { fn is_reusable_for(&self, force_relay: bool) -> bool {
self.relay_only == force_relay self.relay_only == force_relay
&& !matches!( && !matches!(
@@ -323,17 +271,9 @@ impl WebRTCStream {
if !matches!(u.scheme(), "turn" | "turns" | "stun" | "stuns") { if !matches!(u.scheme(), "turn" | "turns" | "stun" | "stuns") {
return None; return None;
} }
// Two spellings are accepted, and they parse very differently: // 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
// - `turn://user:pass@host:port` — non-standard, but the only form that can carry // cannot-be-a-base, so `host_str()` is None and the whole `host:port` lands in the path.
// credentials. `url` sees an authority and fills host/port/username/password.
// - `turn:host:port` — the RFC 7065 form users actually copy from TURN docs. These
// schemes are not special and there is no `//`, so `url` makes it cannot-be-a-base:
// `host_str()` is None and the whole `host:port` lands in the path. Reading it back
// out is what makes this form work at all; it previously produced a hostless
// `turn::3478` that no ICE agent can resolve — while still satisfying
// `has_turn_server()`, so a force_relay peer connection was built against it and
// could only ever time out.
let (host, port) = match u.host_str() { let (host, port) = match u.host_str() {
Some(host) => (host.to_owned(), u.port().unwrap_or(3478)), Some(host) => (host.to_owned(), u.port().unwrap_or(3478)),
None => Self::split_host_port(u.path())?, None => Self::split_host_port(u.path())?,
@@ -522,14 +462,9 @@ impl WebRTCStream {
let (ice_tx, ice_rx) = mpsc::unbounded_channel::<String>(); let (ice_tx, ice_rx) = mpsc::unbounded_channel::<String>();
// Create a new RTCPeerConnection // Create a new RTCPeerConnection
let pc = Arc::new(api.new_peer_connection(config).await?); let pc = Arc::new(api.new_peer_connection(config).await?);
// This closure is the only owner of the ICE-candidate sender, and `on_*` handlers live // `on_*` handlers are never cleared — not by `close()`, not by `SESSIONS` eviction — so
// inside the pc: nothing clears them — not `RTCPeerConnection::close`, not the ICE // this sender outlives the pc and would park a caller's ICE forwarder on `recv()` forever,
// gatherer's own close — so the sender outlives `close()` and every `SESSIONS` eviction. // holding the pc with it. The terminal-state handler below drops this closure to break it.
// The receiver therefore never sees the channel close, and the forwarder task this API
// asks callers to write (`while let Some(c) = rx.recv().await { peer.add_remote_ice(..) }`
// with a stream clone moved in) parks on `recv()` forever, holding the pc alive with it:
// one leaked task plus one leaked peer connection per connection. The terminal-state
// handler below drops this closure to break that; see the note there.
let local_ice_tx = ice_tx.clone(); let local_ice_tx = ice_tx.clone();
pc.on_ice_candidate(Box::new(move |candidate| { pc.on_ice_candidate(Box::new(move |candidate| {
let local_ice_tx = local_ice_tx.clone(); let local_ice_tx = local_ice_tx.clone();
@@ -589,14 +524,9 @@ impl WebRTCStream {
let stream_for_dc_clone = stream_for_dc.clone(); let stream_for_dc_clone = stream_for_dc.clone();
let dc_bound = dc_bound.clone(); let dc_bound = dc_bound.clone();
Box::pin(async move { Box::pin(async move {
// Reassembly spans data-channel messages, so it is sound only on an ordered, // Reassembly is sound only on an ordered, fully-reliable channel: the fragment
// fully-reliable channel: the 1-byte fragment header carries no sequence // header has no sequence number, so a reorder or a loss silently merges
// number, so a reorder splices fragments into a well-formed but wrong // messages. webrtc-rs takes these from the unauthenticated REMOTE's DCEP OPEN.
// message, and a dropped fragment merges two messages instead of erroring.
// webrtc-rs derives these parameters entirely from the REMOTE's DCEP OPEN
// (`channel_type` picks ordered / max_retransmits / max_packet_life_time),
// and this pc is built from an unauthenticated offer, so the peer would
// otherwise choose our reassembly's correctness conditions for us.
if !dc.ordered() if !dc.ordered()
|| dc.max_retransmits().is_some() || dc.max_retransmits().is_some()
|| dc.max_packet_lifetime().is_some() || dc.max_packet_lifetime().is_some()
@@ -660,13 +590,9 @@ impl WebRTCStream {
return; return;
}; };
// Drop the ICE-candidate handler, and with it the only sender for the // Nothing else ever drops this sender (`close()` clears no handler), so
// local-candidate channel. Nothing else ever will: `close()` clears no // replace the handler to close the channel: a caller's forwarder loop then
// handler, so the receiver would stay open forever and a caller's // ends instead of parking on `recv()` and holding this pc alive.
// forwarder task would park on `recv()` holding a stream clone — keeping
// this pc alive past its own teardown. Replacing the handler with a
// no-op closes the channel, the forwarder's loop ends, and its clone
// goes with it. Gathering is over by this state anyway.
pc_for_close2.on_ice_candidate(Box::new(|_| Box::pin(async {}))); pc_for_close2.on_ice_candidate(Box::new(|_| Box::pin(async {})));
let mut sessions_lock = SESSIONS.lock().await; let mut sessions_lock = SESSIONS.lock().await;
@@ -712,14 +638,9 @@ impl WebRTCStream {
}) })
})); }));
// process offer/answer // 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
// Trickle ICE: this block is local-only work (pc construction, DTLS keygen, SDP marshal), // state handler only fires on a terminal ICE state, so close before propagating.
// no gathering wait. The controlled side awaits answer creation inline on its punch-reply
// critical path, so adding a network wait here would delay every hole punch.
// A failure below leaves a live pc whose state handler only fires on a terminal ICE state,
// so a bare `?`-drop leaks it — remotely triggerable via a crafted `type:"answer"` offer
// that passes the pre-check but fails `set_remote_description`. Close before propagating.
let offer_answer: ResultType<String> = async { let offer_answer: ResultType<String> = async {
if start_local_offer { if start_local_offer {
let sdp = pc.create_offer(None).await?; let sdp = pc.create_offer(None).await?;
@@ -811,14 +732,9 @@ impl WebRTCStream {
let sdp = if self.relay_only { let sdp = if self.relay_only {
serde_json::to_string(&local_desc)? serde_json::to_string(&local_desc)?
} else { } else {
// Declare the ICE transport policy inside the envelope. The receiver of an // Rides in the envelope, not a proto field the rendezvous server would forward,
// offer that arrives with force_relay set must know whether it may answer // because it is the offer's own property: the receiver must know whether
// with full ICE (relay forced by the transport, e.g. WebSocket signaling) // force_relay was policy (stay Relay-only) or transport. Older peers ignore it.
// or must stay Relay-only + TURN-gated (relay by policy) — and the envelope
// is the offer's own property, so it rides here rather than in a proto
// field the rendezvous server would have to forward. An extra key is
// invisible to older peers: serde ignores unknown fields when parsing
// RTCSessionDescription, so absence — not an error — is the old semantics.
let mut v = serde_json::to_value(&local_desc)?; let mut v = serde_json::to_value(&local_desc)?;
v[Self::ICE_POLICY_KEY] = serde_json::Value::from(Self::ICE_POLICY_ALL); v[Self::ICE_POLICY_KEY] = serde_json::Value::from(Self::ICE_POLICY_ALL);
serde_json::to_string(&v)? serde_json::to_string(&v)?
@@ -883,17 +799,9 @@ impl WebRTCStream {
let dtls = self.pc.sctp().transport(); let dtls = self.pc.sctp().transport();
let pair = dtls.ice_transport().get_selected_candidate_pair().await?; let pair = dtls.ice_transport().get_selected_candidate_pair().await?;
// Answer from the selected pair, not from the stats report's `nominated` flag: in // Not the stats report's `nominated` flag: webrtc-ice never clears it per checklist entry,
// webrtc-ice that flag is sticky per checklist entry (set on nomination, never cleared — // so after a pair switch several entries carry it and map order decides the answer.
// a pair switch only clears the agent's own `nominated_pair` slot), and the report // 0.13 keeps the candidates private, so read the types out of `Display` by position.
// enumerates the whole checklist. After an ICE switch, or on a controlled agent that saw
// USE-CANDIDATE more than once, several entries carry it and `HashMap::values()` order
// decides the answer — which can differ between two calls in one session.
//
// 0.13 keeps the pair's candidates private, so read the types out of its `Display`:
// "(local) {protocol} {typ} {addr}:{port}{related} <-> (remote) ...". Positionally, not
// by substring — an address must not be able to pass for a candidate type. See the
// upgrade checklist: 0.17+ exposes the fields and this parse goes away.
let relay = RTCIceCandidateType::Relay.to_string(); let relay = RTCIceCandidateType::Relay.to_string();
Some( Some(
pair.to_string() pair.to_string()
@@ -944,22 +852,11 @@ impl WebRTCStream {
self.pc.close().await.ok(); self.pc.close().await.ok();
} }
/// Tear the pc down on the runtime instead of awaiting it here, keeping `keep` alive until /// Tear the pc down on the runtime, keeping `keep` alive until the teardown finishes.
/// the teardown finishes.
/// ///
/// Every caller polls `next()` — and often `send_bytes` — inside a `tokio::select!`, so an /// Callers poll this path inside a `select!`, and `close()` latches `is_closed` before its
/// `.await` on these paths is a cancellation point, and a competing arm (connection.rs and /// first await, so a cancelled close strands the pc: unretryable, never evicted from
/// io_loop.rs both run a 1s timer next to the read) routinely wins one. That is fatal to a /// `SESSIONS`, never notifying `Closed`. The send path passes its permit as `keep`.
/// close: `RTCPeerConnection::close` latches `is_closed` before its first await and fires
/// the state handler only at the very end, so a cancelled close leaves a pc no later
/// `close()` can retry (they early-return on `is_closed`), whose `SESSIONS` entry — evicted
/// only from that handler — is stranded for the life of the process, and whose
/// `state_notify` never reaches `Closed`, leaving `wait_for_connect_result` reporting a live
/// connection on a dead pc.
///
/// `keep` carries anything whose lifetime must span the teardown rather than the caller's:
/// the send path passes its logical-message permit, so no waiting clone can append to a
/// partially-written fragment sequence while the close is still in flight.
pub fn close_detached_with<T: Send + 'static>(&self, keep: T) { pub fn close_detached_with<T: Send + 'static>(&self, keep: T) {
let pc = self.pc.clone(); let pc = self.pc.clone();
// Take the runtime handle explicitly rather than calling `tokio::spawn`: this also runs // Take the runtime handle explicitly rather than calling `tokio::spawn`: this also runs
@@ -1165,13 +1062,9 @@ impl WebRTCStream {
framed.put_u8(if is_last { FRAG_END } else { FRAG_MORE }); framed.put_u8(if is_last { FRAG_END } else { FRAG_MORE });
framed.put_slice(chunk); framed.put_slice(chunk);
if let Err(err) = dc.write(&framed.freeze()).await { if let Err(err) = dc.write(&framed.freeze()).await {
// A sequence that stops with fragments already on the wire leaves the peer's // Fragments already on the wire leave the peer an unterminated prefix, and this
// accumulator holding a prefix that will never be terminated — and this framing // framing has no length for it to notice with, so the next message is appended
// carries no length or sequence number for it to notice with, so the next message // to it. Callers wrap sends in `allow_err!`, so the error alone is not enough.
// is appended to the orphan and parsed as one corrupt frame. Report it so the
// caller can tear the stream down while still holding the send permit; callers
// wrap sends in allow_err! and keep going, so returning the error alone would
// leave the corruption in place.
*desynced = wrote_any; *desynced = wrote_any;
return Err(err.into()); return Err(err.into());
} }
@@ -1223,24 +1116,17 @@ impl WebRTCStream {
} }
}; };
if n == 0 { if n == 0 {
// End of stream. Our own sender never produces this — every fragment carries at // Not exclusively a stream reset: webrtc-data maps the empty-message PPIDs to
// least its header byte — but it is NOT exclusively a reset: webrtc-data maps the // n == 0 as well and `read` drops the flag that would separate them. Both mean
// StringEmpty/BinaryEmpty PPIDs to n == 0 as well, and `read` discards the flag // nothing frameable follows, so treat them alike but not as a clean remote close.
// that would separate them, so a peer can also reach here by sending one empty
// data-channel message. Both mean the same thing to us (this peer will send us
// nothing more we can frame), so treat them alike, but do not report it as a
// clean remote close: it is equally a peer that just violated the framing.
log::debug!("WebRTC data channel ended (reset or empty message)"); log::debug!("WebRTC data channel ended (reset or empty message)");
*acc = BytesMut::new(); *acc = BytesMut::new();
self.close_detached(); self.close_detached();
return None; return None;
} }
// Two framing violations, both of which would otherwise be read as "more fragments": // Both would otherwise read as "more fragments". The payload-less FRAG_MORE is the
// an unrecognized header, and a FRAG_MORE carrying no payload. The latter is the // nastier one: it adds nothing to `acc`, so the cap below never trips and the loop
// nastier one — it adds nothing to `acc`, so the MAX_FRAME_LENGTH cap below never // spins for as long as the peer keeps sending.
// trips and the loop spins for as long as the peer keeps sending. `send_bytes_inner`
// emits FRAG_MORE only for a full MAX_FRAGMENT_PAYLOAD chunk, so neither is reachable
// from our own sender.
let header = scratch[0]; let header = scratch[0];
let bad = match header { let bad = match header {
FRAG_END => None, FRAG_END => None,
@@ -1258,12 +1144,9 @@ impl WebRTCStream {
format!("WebRTC {why}"), format!("WebRTC {why}"),
))); )));
} }
// Bound BEFORE growing, not after. This framing carries no length, so an oversize // Bound BEFORE growing: this framing carries no length, so an oversize message is only
// message can only be discovered by accumulating it — unlike the TCP codec, which // discovered by accumulating it, and checking after the append would let `BytesMut`'s
// rejects on the declared length before a single payload byte is buffered. Checking // reallocate-and-copy hold the old and the new buffer at once.
// after the append would make the peak the cap plus a fragment, and `BytesMut` grows
// by reallocate-and-copy, so the final doubling would hold the old and new buffers at
// once: ~2x the cap in RSS for a session that has produced no message at all.
if acc.len() + (n - 1) > MAX_RECV_MESSAGE { if acc.len() + (n - 1) > MAX_RECV_MESSAGE {
// Release the buffer, don't just truncate it: by definition it is near the cap // 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. // here, and `recv_state` outlives this call through the `SESSIONS` clone.
@@ -1663,15 +1546,9 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc
(offerer, answerer) (offerer, answerer)
} }
// next()'s teardown paths must not be cancellable: every consumer polls next() inside a // next()'s teardown must not be cancellable: an awaited close that loses the select! race
// select! against a timer, and an awaited close that loses that race strands the pc // against a consumer's timer strands the pc (is_closed already latched) and its SESSIONS
// (is_closed is already latched, so no later close retries) along with its SESSIONS entry, // entry. `close_detached` is non-async, so what needs proving is that the handoff completes.
// which only the state handler evicts.
//
// The cancellation itself is not what this test pins — `close_detached` is a non-async fn,
// so its callers have no await point to be cancelled at, and the compiler enforces that.
// What needs proving is the other half: that work handed to the runtime still runs to
// completion once the caller has walked away.
#[tokio::test] #[tokio::test]
async fn test_close_detached_completes_without_the_caller() { async fn test_close_detached_completes_without_the_caller() {
let (offerer, answerer) = connect_loopback().await; let (offerer, answerer) = connect_loopback().await;
@@ -1802,16 +1679,9 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc
.unwrap(); .unwrap();
} }
// Extra channels opened after the bootstrap one must not displace it: rebinding would leave // Extra channels must not displace the bound one: the newcomer's on_open would push Open onto
// teardown closing a channel send/recv no longer use, and the newcomer's on_open would push // the watch that gates send and recv, re-arming a session already latched Closed. Only the
// Open onto the watch that gates both, re-arming a session already latched Closed. // bind-once guard is exercised; the ordered+reliable check cannot decide anything here.
//
// Scope, honestly: only the bind-once guard is exercised. The ordered+reliable check sits
// ahead of it but cannot decide anything here — `WebRTCStream::new` always creates its
// bootstrap channel first, so whatever the offerer adds afterwards is refused for being
// second regardless of its parameters. Covering that check needs a hand-built peer whose
// FIRST channel is unordered, which is more scaffolding than the guard is worth; it is a
// three-accessor test against parameters webrtc-rs derives verbatim from the remote's DCEP.
#[tokio::test] #[tokio::test]
async fn test_webrtc_answerer_binds_only_the_first_data_channel() { async fn test_webrtc_answerer_binds_only_the_first_data_channel() {
use webrtc::data_channel::data_channel_init::RTCDataChannelInit; use webrtc::data_channel::data_channel_init::RTCDataChannelInit;