From 166bb3d2ac3a59eeaafab6ac0aac7b98c8d93778 Mon Sep 17 00:00:00 2001 From: rustdesk Date: Tue, 8 Sep 2026 16:52:28 +0800 Subject: [PATCH] webrtc: do not hand out a cached session ICE has lost the peer on `is_reusable_for` rejected a closed session but not one ICE currently reports Disconnected. Nothing else does either - that state is deliberately not part of the connection-state enum - so a caller could be given a session that already carries the hint, and start off suspecting a peer it never used. Also drops a race from the receive-progress test: it raced `next()` against a sleeping sibling in a select, so a transfer that landed inside one poll left the assertion looking false, and a slow enough one exhausted the loop and failed with a misleading message. It now reads the counter after a `next_timeout` that returned nothing, which is a point where the message is known to be incomplete. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_019aokqJuhjvB3kijXtAg5Ns --- src/webrtc.rs | 29 +++++++++++++++++++---------- 1 file changed, 19 insertions(+), 10 deletions(-) diff --git a/src/webrtc.rs b/src/webrtc.rs index 39cf6ae9a..52960df06 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -347,6 +347,10 @@ impl WebRTCStream { fn is_reusable_for(&self, force_relay: bool) -> bool { self.relay_only == force_relay && !self.handoff.closing.load(Ordering::Acquire) + // ICE has stopped hearing from the peer on this one. Not terminal, so nothing else + // rejects it, but handing it to a caller that is about to judge liveness starts them + // off already suspecting a pc they never used. + && !self.is_disconnected() && !matches!( *self.state_notify.borrow(), WebRTCConnectionState::Closed(_) @@ -2314,22 +2318,24 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc assert_eq!(&answerer.next().await.unwrap().unwrap()[..], b"warmup"); let baseline = answerer.rx_progress(); - let big = vec![0x5Au8; 512 * 1024]; + // Large enough that it cannot land inside one poll of the loop below, so a probe + // that only counted whole messages would leave the counter untouched throughout. + let big = vec![0x5Au8; 4 * 1024 * 1024]; let expected = big.len(); let sender = tokio::spawn(async move { offerer.send_raw(big).await.map(|_| offerer) }); - // The receive loop's shape: sampling happens in a sibling arm, so the probe has to be - // readable while `next()` is in flight and a cancelled `next()` has to keep the - // partial message. + // `next_timeout` returning None leaves the partial message in place, so the counter + // is read at a point where nothing has been delivered. No select and no sleeping + // sibling: the assertion cannot turn on which branch a poll happened to take. let mut moved_early = false; let mut got = None; - for _ in 0..2_000 { - tokio::select! { - msg = answerer.next() => { - got = msg; + for _ in 0..600 { + match answerer.next_timeout(20).await { + Some(msg) => { + got = Some(msg); break; } - _ = tokio::time::sleep(Duration::from_millis(1)) => { + None => { if answerer.rx_progress() > baseline { moved_early = true; } @@ -2337,11 +2343,14 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc } } - assert_eq!(got.expect("message").unwrap().len(), expected); assert!( moved_early, "rx_progress must advance while the message is still incomplete" ); + assert_eq!( + got.expect("message never completed").unwrap().len(), + expected + ); let offerer = sender.await.unwrap().expect("send"); // `WebRTCStream` has no `Drop`, so both pcs stay in `SESSIONS` unless closed here; // `test_cancelled_new_does_not_leak_the_pc` scans that map and would report them.