webrtc/tests: look for a session that outlasts the window, not an idle instant

`test_cancelled_new_does_not_leak_the_pc` demanded an instant at which
no key had appeared since its snapshot. That asks the whole suite to go
quiet, which `--test-threads=2` never grants: one lane is this test for
its entire 30s wait while the other keeps starting sessions, so the
difference is never empty and the test fails whatever the pc it is
actually watching did.

Intersect the difference across samples instead. A concurrent test's
session appears and is closed again, so it drops out; a leaked pc never
does. The cancelled attempt's own key cannot be named here — its
fingerprint is generated inside the task that was abandoned — so
outlasting the window is the property available to test, and it is the
one that means "leaked".

This sharpens what the failure says; it does not make `--test-threads=2`
pass. With the new assertion a single `offer:` key still survives all
30s there, and the test passes in 0.5s when run alone, so the entry
belongs to another test rather than to the cancelled `new()`. The two
that close only their answerer and leave the offerer to an indirect
path — `test_session_end_close_reaches_the_peer` (Stream::close_webrtc)
and `test_eof_close_then_drop_still_evicts` (Drop) — are where to look.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019UzcMTdYTEv2QbMHcTSUy3

src/webrtc.rs

@@ -2243,33 +2243,44 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc
     }
 
     // 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. Keys are
-    // compared as a set difference so concurrent tests' own sessions do not interfere; theirs
-    // clean up within the wait too.
+    // 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.
         let _ = timeout(Duration::ZERO, WebRTCStream::new("", false, 20000)).await;
-        // Let the detached setup task finish (and insert its session) before demanding the
-        // difference be empty, or an early check passes vacuously while the leak forms later.
+        // 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;
-        // 30s: concurrent tests' transient sessions land in the difference too and must be
-        // given time to finish and evict (every test closes what it creates).
-        const ATTEMPTS: usize = 600;
+        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 leftover: Vec<&String> = now.difference(&before).collect();
-            if leftover.is_empty() {
+            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 `{leftover:?}`: on edition 2018 a lone-literal `panic!`
+            // 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 (leftover: {:?})",
-                leftover
+                "cancelled new() left a pc cached in SESSIONS (persisted: {:?})",
+                persisted
             );
             tokio::time::sleep(Duration::from_millis(50)).await;
         }