diff --git a/doc/concept/transport.md b/doc/concept/transport.md index f620e4e51f..a0ff1a7067 100644 --- a/doc/concept/transport.md +++ b/doc/concept/transport.md @@ -42,9 +42,8 @@ connection (`wss://`) and keep whichever wins. A small multiplexer, [qmux](/draft/qmux-websocket), carries MoQ streams over the socket. It works everywhere but cannot escape TCP head-of-line blocking, so priority and resets only help once bytes leave the TCP queue. The fallback is automatic in every -client; the relay enables it with `[web.https]`. A native Rust client that -lands on WebSocket keeps dialing QUIC and moves the session over once QUIC -succeeds. +client; the relay enables it with `[web.https]`. A session that lands on +WebSocket stays there until it ends; the next connect races again. ## Raw QUIC (native) diff --git a/quest/m0/README.md b/quest/m0/README.md index 1bf7210264..077c10a4b7 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -59,7 +59,7 @@ Interop on main (2026-10-13): the cross-client matrix failed with `unroutable` whenever WebSocket won the connect race: since #5294, #4180's upgrade retired epochless handles before its held `Restart` reached anyone. Handles now retire only once it is deliverable. First-party consumers following it -moved to m1 (maintainer, 2026-10-13), since the upgrade is being removed. +moved to m1 (maintainer, 2026-10-13), since the upgrade is removed. Crashes (2026-10-13, #5407): a publisher on a host whose wall clock reads before 2020 panics, and `panic = "abort"` turned any library panic into a @@ -84,5 +84,4 @@ have fixes on `main`: macOS tests (#5335), drill-sensitivity (#5338), loom - [web-transport releases the qmux fixes](/quest/m0/qmux-credit-upstream.md) - waiting on moq-dev/web-transport#412 and #413 to merge and ship, which qmux credit bumps to - [qmux credit](/quest/m0/qmux-credit.md) - qmux returns connection credit for dropped and stopped streams and delivers its close frame, on both lines - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - every first-party publisher that can restart mints a fresh route epoch, the newest wins a path, and only routes with the same epoch resume a subscription -- [Keep the winning transport](/quest/m0/drop-ws-upgrade.md) - a connection stays on the transport that won the race; the WebSocket-to-QUIC upgrade and its epochless `Restart` are removed - [Parked drain](/quest/m0/parked-drain.md) - draining parked groups after a cap rise, as a same-epoch relay handover causes, is near-linear in the backlog, not quadratic diff --git a/quest/m0/drop-ws-upgrade.md b/quest/m0/drop-ws-upgrade.md deleted file mode 100644 index 73cab4f403..0000000000 --- a/quest/m0/drop-ws-upgrade.md +++ /dev/null @@ -1,60 +0,0 @@ -# [M] A connection keeps the transport that won the race - -## Goal - -A `moq-tokio` connection whose `https://` race is won by WebSocket stays on -WebSocket for that session's lifetime; nothing migrates it to QUIC. The -pending QUIC dial is cancelled the moment WebSocket wins, as when QUIC wins. -An epochless viewer no longer sees a `Restart` shortly after connecting. - -Non-goals: changing the race itself, its head start, or the "WebSocket won" -memo's effect on later connects. - -## Plan - -Decided with the maintainer on 2026-10-13: remove the WebSocket-to-QUIC -upgrade from #4180. On an epochless route, which is the default for -lite-06, the browser, and the bindings, the upgrade is a `Restart`: players -re-resolve, re-fetch the catalog, and reset decoders right after connecting. -That is a brutal user experience, and nothing shows the upgrade helps outside -CI's debug builds. Rejected: upgrading only when epochs match (keeps the -machinery for a minority of sessions), keeping it as is, and measuring first. - -- Delete the upgrade in `rs/moq-tokio` (`connection.rs`'s `upgrade` state and - its handover, `client.rs`'s post-race `Setup` stage) and its tests - (`websocket_upgrades_to_quic`, `a_refused_upgrade_keeps_websocket`). -- Cancel the pending QUIC dial at once when WebSocket wins, deleting #5284's - closed fallback too: a WebSocket session that closes early goes through the - ordinary reconnect, which races again. -- Keep the "WebSocket won" memo (`WEBSOCKET_WON`). Both transports still race - without the head start, so QUIC can still win a later connect. Its only - eviction was a successful upgrade; check what clears it now and document it. -- Remove `moq_net::Session::setup()` unless the GOAWAY drain migration gates - on it; if it does, keep it and say why in its doc. Keep - `Connection::transport()`, which reports the live transport. -- Keep the origin fix from #5443 (`Release` in `model/origin.rs`): GOAWAY - drains and publisher replacement still hand an epochless route over the - same way. -- `js/net` keeps closing the loser as today. This plan's PR deleted the m1 - transport-upgrade epic and its JavaScript child, and folded the QUIC-wins - race flake in below. -- Add a regression: when WebSocket wins, the QUIC dial is dropped and no - second session ever starts, with mocked time. -- Absorbs the rest of the QUIC-wins race flake: `quic_winning_opens_one_session` - already holds the fallback handshake, but `fallback.accepted.try_recv()` - only proves the queue is empty at one instant. Observe the loser's - cancellation directly, by a drop or connection-close signal, in both - directions. -- Update docs and comments that describe the upgrade inline, and search the - bindings for the `WebTransport` transport report #4180 added; it stays. - -Public API: breaking in moq-net (`Session::setup()` removed, unless kept as -above). Behavior: no QUIC upgrade after a WebSocket win. Wire: none. Report -both in the PR. - -## Related - -- [#4180](https://github.com/moq-dev/moq/pull/4180) - the upgrade this removes -- [#5284](https://github.com/moq-dev/moq/pull/5284) - the closed fallback this removes -- [#5443](https://github.com/moq-dev/moq/pull/5443) - the origin fix that stays -- [Restart followers](/quest/m1/restart-followers.md) - consumers that follow the `Restart` GOAWAY drains and publisher replacement still send diff --git a/quest/m1/restart-followers.md b/quest/m1/restart-followers.md index f365d8b50e..924814da6f 100644 --- a/quest/m1/restart-followers.md +++ b/quest/m1/restart-followers.md @@ -11,13 +11,13 @@ then costs a long-lived consumer nothing but the restart. ## Plan Moved to m1 on 2026-10-13 (maintainer): the WebSocket-to-QUIC upgrade is -being removed (separate m0 quest), so an epochless Restart no longer follows +removed, so an epochless Restart no longer follows every connect; it remains for GOAWAY drains and publisher replacement. The origin side is done: a handle resolved through a peer keeps serving new requests while the peer's epochless replacement is held, and refuses them with `Unroutable` only once the `Restart` is deliverable (`Release` in `rs/moq-net/src/model/origin.rs`, -`rs/moq-net/tests/upgrade.rs`). So a consumer that never follows `Restart` +`rs/moq-net/tests/replacement.rs`). So a consumer that never follows `Restart` works through the hold (300 ms by default), and its open subscriptions stay sticky on the old session until it closes, where they end cleanly. A consumer that opens a request after the hold, or reads past the old @@ -45,9 +45,9 @@ Audit and fix: - The interop clients under `test/interop/clients`, where they resolve a broadcast themselves. -Regression: extend `rs/moq-net/tests/upgrade.rs` (or the consumer's own +Regression: extend `rs/moq-net/tests/replacement.rs` (or the consumer's own tests) so each follower reads past the old session's close after an -epochless upgrade. Then run `just test interop --all`. +epochless replacement. Then run `just test interop --all`. If a failing client still leaves its connection to idle out, fix that client's error path to close its session, as a separate commit. The C @@ -59,5 +59,4 @@ Wire: none. ## Related -- [Keep the winning transport](/quest/m0/drop-ws-upgrade.md) - removes the upgrade, so `Restart` no longer follows every WebSocket connect - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - the players that already follow `Restart` diff --git a/quest/m1/slow-source.md b/quest/m1/slow-source.md index 46c8ff1f70..a95251d4bc 100644 --- a/quest/m1/slow-source.md +++ b/quest/m1/slow-source.md @@ -6,7 +6,7 @@ An export whose source arrives further off real time than the output's clock may follow ends with an error naming the measured rate, once it has used half its delay. It no longer drops nearly every frame for hours with only per-frame "missed its deadline" warnings. A short stall, such as a -WebSocket-to-QUIC upgrade, still drops late frames and recovers. +reconnect, still drops late frames and recovers. ## Plan diff --git a/quest/m2/quic-sim.md b/quest/m2/quic-sim.md index fe6e7781e3..97ec5a4cea 100644 --- a/quest/m2/quic-sim.md +++ b/quest/m2/quic-sim.md @@ -4,7 +4,7 @@ The `moq-time` sim drives two `moq-quic` endpoints over in-memory datagrams with configurable latency, loss, reordering, and bandwidth, so QUIC behavior -(BBR, reliable reset, stream deadlines, the transport upgrade) and MoQ over +(BBR, reliable reset, stream deadlines, the transport race) and MoQ over real QUIC are tested deterministically with no sockets. ## Plan diff --git a/quest/m3/quic-careful-resume.md b/quest/m3/quic-careful-resume.md index ad42ce4e13..369dd02321 100644 --- a/quest/m3/quic-careful-resume.md +++ b/quest/m3/quic-careful-resume.md @@ -11,8 +11,6 @@ to slow start without hurting the path. ## Plan Decided in the 2026-10-10 audit: scope this to remembered QUIC connections. -Transferring TCP estimates into a WebSocket-to-QUIC upgrade is out of scope; -that upgrade starts without a seed. Follow the shape of draft-ietf-ccwg-careful-resume: the sender keeps a per-destination record of the last validated bandwidth estimate and minimum diff --git a/quest/m3/uring-tcp/README.md b/quest/m3/uring-tcp/README.md index 124c71b3bc..fb485682b1 100644 --- a/quest/m3/uring-tcp/README.md +++ b/quest/m3/uring-tcp/README.md @@ -34,9 +34,8 @@ number is what justifies the rest of the line. Decided in the 2026-09-30 audit: moved to m2, since no fleet demand asks for ring TCP. (Decided 2026-10-13: the WebSocket-to-QUIC upgrade that would -have shrunk the WebSocket hot path is being removed by -[Keep the winning transport](/quest/m0/drop-ws-upgrade.md), so a WebSocket -session will stay on TCP for its lifetime.) +have shrunk the WebSocket hot path is removed, so a WebSocket session stays +on TCP for its lifetime.) Decided 2026-10-08: moved to m3. No fleet demand asks for ring TCP; the ablation stays the gate for the rest of the line. diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 47241f4ce8..3655a3a00b 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -229,7 +229,6 @@ impl Client { crate::session::Handles { goaway: start.goaway, auth: start.auth, - setup: start.setup, estimate: Some(start.estimate), }, )) @@ -292,7 +291,7 @@ impl Client { // The peer's SETUP decides whether AUTH is negotiated. A handle that does not // offer it keeps the option out of our SETUP and presents no credential. let auth = crate::auth::Handle::new(self.extensions.auth); - let (protocol, goaway, setup) = ietf::start(ietf::Config { + let (protocol, goaway) = ietf::start(ietf::Config { runtime: runtime.clone(), limits: self.limits, session: session.clone(), @@ -323,7 +322,6 @@ impl Client { crate::session::Handles { goaway, auth, - setup, estimate: None, }, )); @@ -418,7 +416,7 @@ impl Client { .copied() .ok_or(Error::Version)?; - let (recv_bw, protocol, goaway, auth, setup, estimate) = match version { + let (recv_bw, protocol, goaway, auth, estimate) = match version { Version::Lite(v) => { let stream = stream.with_version(v); let start = lite::start(lite::Config { @@ -444,7 +442,6 @@ impl Client { crate::driver::Protocol::Lite(Box::new(start.driver)), start.goaway, start.auth, - start.setup, Some(start.estimate), ) } @@ -466,7 +463,7 @@ impl Client { // Draft 14-16 carry no AUTH, but the session is still limited through this handle. let auth = crate::auth::Handle::new(false); // Draft 14-16: the path rode in the bidi SETUP above, not the uni one. - let (protocol, goaway, setup) = ietf::start(ietf::Config { + let (protocol, goaway) = ietf::start(ietf::Config { runtime: runtime.clone(), limits: self.limits, session: session.clone(), @@ -486,7 +483,7 @@ impl Client { early_unis: Vec::new(), extensions: self.extensions, })?; - (None, crate::driver::Protocol::Ietf(protocol), goaway, auth, setup, None) + (None, crate::driver::Protocol::Ietf(protocol), goaway, auth, None) } }; @@ -496,12 +493,7 @@ impl Client { version, recv_bw, protocol, - crate::session::Handles { - goaway, - auth, - setup, - estimate, - }, + crate::session::Handles { goaway, auth, estimate }, )) } } diff --git a/rs/moq-net/src/ietf/peer.rs b/rs/moq-net/src/ietf/peer.rs index c4efdd09c4..c85f53e70b 100644 --- a/rs/moq-net/src/ietf/peer.rs +++ b/rs/moq-net/src/ietf/peer.rs @@ -52,16 +52,6 @@ impl PeerSetup { } } - /// Poll until the peer's SETUP arrives. - pub fn poll_seen(&self, waiter: &kio::Waiter) -> std::task::Poll<()> { - self.0 - .poll(waiter, |peer| match peer.is_some() { - true => std::task::Poll::Ready(()), - false => std::task::Poll::Pending, - }) - .map(|_| ()) - } - /// Await the peer's SETUP. /// /// The peer MUST send exactly one, so this resolves once that stream is read. Waits diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 7e5a5791dc..e7c5eb117a 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -109,7 +109,7 @@ impl std::future::Future for Driver { } } -pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle, crate::session::Setup), Error> +pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> where S: crate::transport::poll::Boxable, { @@ -178,9 +178,6 @@ where None => {} } - // Settled already on drafts 14-16, which read the peer's SETUP in the handshake. - let setup_seen = crate::session::Setup::Ietf(peer_setup.clone()); - let driver = async move { // Held for the life of the session. let _setup_token = setup_token; @@ -632,7 +629,6 @@ where future: driver, }, goaway_handle, - setup_seen, )) } @@ -1490,7 +1486,7 @@ mod tests { let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_bidis(incoming); let log = session.log.clone(); - let (driver, _, _) = start(Config { + let (driver, _) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -1545,7 +1541,7 @@ mod tests { script.extend(body); let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![script]); let log = session.log.clone(); - let (driver, _, _) = start(Config { + let (driver, _) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -1591,7 +1587,7 @@ mod tests { let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_bidis(vec![script]); let log = session.log.clone(); - let (driver, _, _) = start(Config { + let (driver, _) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -1685,7 +1681,7 @@ mod tests { let session = crate::lite::test_transport::ScriptedSession::new(namespace_without_hop_path(VERSION).await); let log = session.log.clone(); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -1748,7 +1744,7 @@ mod tests { let session = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()); let log = session.log.clone(); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -1821,7 +1817,7 @@ mod tests { .await .expect("open the control stream"); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: Some(setup), @@ -1909,7 +1905,7 @@ mod tests { let session = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()); let log = session.log.clone(); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -2012,7 +2008,7 @@ mod tests { let log = session.log.clone(); let handle = crate::auth::Handle::new(true); - let (driver, goaway, _) = start(Config { + let (driver, goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -2096,7 +2092,7 @@ mod tests { let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()); let log = session.log.clone(); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -2192,7 +2188,7 @@ mod tests { let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()) .with_incoming_bidis(vec![auth_request(VERSION).await]); let log = session.log.clone(); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -2281,7 +2277,7 @@ mod tests { let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); let log = session.log.clone(); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -2324,7 +2320,7 @@ mod tests { let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()); let log = session.log.clone(); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -2728,7 +2724,7 @@ mod tests { let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_bidis(vec![payload]); let log = session.log.clone(); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -2796,7 +2792,7 @@ mod tests { let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_bidis(vec![payload]); let log = session.log.clone(); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: None, @@ -2897,7 +2893,7 @@ mod tests { .await .expect("open the control stream"); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup: Some(setup), @@ -3014,7 +3010,7 @@ mod tests { _ => None, }; let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); - let (driver, _goaway, _) = start(Config { + let (driver, _goaway) = start(Config { runtime: crate::time::Clock::sim(), session, setup, diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index 47152ccc24..17dfd2059d 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -22,8 +22,6 @@ pub(crate) struct SessionStart { pub goaway: crate::goaway::Handle, /// The session's AUTH tokens and grants, stored on the public [`crate::Session`]. pub auth: crate::auth::Handle, - /// Whether the peer's SETUP arrived, read by [`crate::Session::setup`]. - pub setup: crate::session::Setup, } /// Server: read the peer's single SETUP message off its Setup Stream before starting @@ -276,16 +274,6 @@ where }; let peer_setup = peer_setup_slot; - // Lite-05+ records the peer's SETUP from its Setup Stream. Before that, only the - // legacy bidi handshake carries one, and it was read before the session started. - let setup = if version.has_setup_stream() { - crate::session::Setup::Lite(peer_setup.clone()) - } else if setup_stream.is_some() { - crate::session::Setup::Read - } else { - crate::session::Setup::Never - }; - // GOAWAY wiring: the public Session holds one half (send trigger, received // signal), the protocol tasks below hold the other. moq-lite lets either side // name a redirect URI, unlike moq-transport. @@ -362,7 +350,6 @@ where driver, goaway: goaway_handle, auth, - setup, }) } diff --git a/rs/moq-net/src/lite/setup.rs b/rs/moq-net/src/lite/setup.rs index 446b75b7ba..a9d566c27e 100644 --- a/rs/moq-net/src/lite/setup.rs +++ b/rs/moq-net/src/lite/setup.rs @@ -283,11 +283,6 @@ impl PeerSetup { Ok(()) } - /// Poll until the peer's SETUP arrives. - pub fn poll_seen(&self, waiter: &kio::Waiter) -> std::task::Poll<()> { - self.poll_get(waiter, |_| ()) - } - /// Poll for the peer's advertised probe level, waiting until its SETUP arrives. pub fn poll_probe_level(&self, waiter: &kio::Waiter) -> std::task::Poll { self.poll_get(waiter, |setup| setup.probe) diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 86d2c1efc1..33a24a64dc 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -196,7 +196,6 @@ impl Server { crate::session::Handles { goaway: start.goaway, auth: start.auth, - setup: start.setup, estimate: Some(start.estimate), }, )) @@ -617,7 +616,7 @@ where // The client's SETUP was read at the pause; hand the stream back // for GOAWAY. A server never advertises a path, hence `None`. - let (protocol, goaway, setup) = ietf::start(ietf::Config { + let (protocol, goaway) = ietf::start(ietf::Config { runtime: runtime.clone(), limits: server.limits, session: session.clone(), @@ -648,7 +647,6 @@ where crate::session::Handles { goaway, auth, - setup, estimate: None, }, )) @@ -723,7 +721,7 @@ where }; stream.writer.encode(&server_setup).await?; - let (recv_bw, protocol, goaway, auth, setup, estimate) = match version { + let (recv_bw, protocol, goaway, auth, estimate) = match version { Version::Lite(v) => { let stream = stream.with_version(v); // Pre-lite-05: no Setup Stream, so nothing to advertise or seed. @@ -746,14 +744,13 @@ where crate::driver::Protocol::Lite(Box::new(start.driver)), start.goaway, start.auth, - start.setup, Some(start.estimate), ) } Version::Ietf(v) => { let stream = stream.with_version(v); // Draft 14-16: path came in the bidi SETUP, no uni SETUP to hand back. - let (protocol, goaway, setup) = ietf::start(ietf::Config { + let (protocol, goaway) = ietf::start(ietf::Config { runtime: runtime.clone(), limits: server.limits, session: session.clone(), @@ -773,7 +770,7 @@ where early_unis: Vec::new(), extensions: server.extensions, })?; - (None, crate::driver::Protocol::Ietf(protocol), goaway, auth, setup, None) + (None, crate::driver::Protocol::Ietf(protocol), goaway, auth, None) } }; @@ -783,12 +780,7 @@ where version, recv_bw, protocol, - crate::session::Handles { - goaway, - auth, - setup, - estimate, - }, + crate::session::Handles { goaway, auth, estimate }, )) } .maybe_boxed() diff --git a/rs/moq-net/src/session.rs b/rs/moq-net/src/session.rs index 38702fe7d7..2c7e97dd01 100644 --- a/rs/moq-net/src/session.rs +++ b/rs/moq-net/src/session.rs @@ -35,31 +35,6 @@ struct StatsState { demanded: bool, } -/// How [`Session::setup`] learns that the peer's SETUP arrived. -#[derive(Clone)] -pub(crate) enum Setup { - /// The handshake read it before the session started. - Read, - /// The lite driver records it from the peer's Setup Stream. - Lite(crate::lite::PeerSetup), - /// The moq-transport driver records it from the peer's SETUP stream. - Ietf(crate::ietf::peer::PeerSetup), - /// The version carries no SETUP from the peer. - Never, -} - -impl Setup { - /// `Ready(true)` once the SETUP arrived, `Ready(false)` if it never will. - fn poll(&self, waiter: &kio::Waiter) -> Poll { - match self { - Self::Read => Poll::Ready(true), - Self::Lite(setup) => setup.poll_seen(waiter).map(|()| true), - Self::Ietf(setup) => setup.poll_seen(waiter).map(|()| true), - Self::Never => Poll::Ready(false), - } - } -} - /// A snapshot of connection statistics for a [`Session`]. /// /// Every field is optional: availability depends on the transport backend (native QUIC @@ -153,7 +128,6 @@ pub struct Session { recv_bandwidth: Option, goaway: Arc, auth: auth::Handle, - setup: Setup, } impl Session { @@ -269,35 +243,6 @@ impl Session { } } - /// Wait for the peer's SETUP, or return the session's close reason. - /// - /// Resolves when the peer's SETUP arrives. This crate's servers send it only once - /// they admit the client, but neither protocol requires that ordering, so another - /// server may send SETUP and still refuse the session afterward. - /// [`crate::Client::connect`] returns before the SETUP on moq-lite-05+ and - /// moq-transport draft 17+. Older versions read it during the handshake and resolve - /// at once, except moq-lite-03 and -04, which carry none and return - /// [`Error::Unsupported`]. A session that already closed returns its close reason. - pub async fn setup(&self) -> Result<(), Error> { - kio::wait(|waiter| { - match self.closed.poll(waiter, |state| match &**state { - Some(ended) => Poll::Ready(ended.err.clone()), - None => Poll::Pending, - }) { - Poll::Ready(Ok(err)) => return Poll::Ready(Err(err)), - // The driver was dropped before it could observe the close. - Poll::Ready(Err(_)) => return Poll::Ready(Err(Error::Cancel)), - Poll::Pending => {} - } - match self.setup.poll(waiter) { - Poll::Ready(true) => Poll::Ready(Ok(())), - Poll::Ready(false) => Poll::Ready(Err(Error::Unsupported)), - Poll::Pending => Poll::Pending, - } - }) - .await - } - /// Drain the peer gracefully: the handle for sending this session's single /// GOAWAY. /// @@ -343,7 +288,6 @@ impl Session { pub(super) struct Handles { pub goaway: goaway::Handle, pub auth: auth::Handle, - pub setup: Setup, /// Where the driver publishes [`Estimate`] changes, for a protocol that reports /// them to the peer (moq-lite PROBE). Sampled only while a consumer exists. pub estimate: Option>, @@ -361,12 +305,7 @@ impl Session { where S: crate::transport::poll::Session, { - let Handles { - goaway, - auth, - setup, - estimate, - } = handles; + let Handles { goaway, auth, estimate } = handles; let sample = snapshot(&session); if let Some(estimate) = &estimate && let Ok(mut current) = estimate.write() @@ -415,7 +354,6 @@ impl Session { recv_bandwidth, goaway: Arc::new(goaway), auth, - setup, }; let driver = crate::Driver::new( runtime.clone(), diff --git a/rs/moq-net/tests/upgrade.rs b/rs/moq-net/tests/replacement.rs similarity index 81% rename from rs/moq-net/tests/upgrade.rs rename to rs/moq-net/tests/replacement.rs index 8530127336..63fe772cfe 100644 --- a/rs/moq-net/tests/upgrade.rs +++ b/rs/moq-net/tests/replacement.rs @@ -1,10 +1,11 @@ -//! A transport upgrade, end to end over real sessions: a second session from the -//! same peer re-announces every route while the first one is still open. +//! A replacement session, end to end over real sessions: a second session from the +//! same peer re-announces every route while the first one is still open, as a GOAWAY +//! drain does. //! //! Without an epoch the re-announced routes are new instances, so announce consumers -//! get a `Restart`. A handle resolved before the upgrade keeps serving new requests +//! get a `Restart`. A handle resolved before the replacement keeps serving new requests //! until that `Restart` is deliverable, so its holder can follow the `Restart` instead -//! of failing `Unroutable`. With an epoch the upgrade continues the instance: no +//! of failing `Unroutable`. With an epoch the replacement continues the instance: no //! `Restart`, and the old handle keeps working. mod support; @@ -67,13 +68,13 @@ async fn subscribe(broadcast: &broadcast::Consumer, name: &str) -> Result) { +async fn replace(version: &str, epoch: Option) { let version: Version = version.parse().unwrap(); let publisher = produce_origin(1); let viewer = produce_origin(3); @@ -85,7 +86,7 @@ async fn upgrade(version: &str, epoch: Option) { let audio = cam.create_track("audio", timed()).unwrap(); let meta = cam.create_track("meta", timed()).unwrap(); - let _websocket = link(version, &publisher, &viewer).await; + let _first = link(version, &publisher, &viewer).await; let mut announced = viewer.consume().announced(); match next_event(&mut announced).await { announce::Event::Start(announce) => assert_eq!(announce.prefix.as_str(), "cam"), @@ -97,12 +98,12 @@ async fn upgrade(version: &str, epoch: Option) { write(&video, "video:0"); assert_eq!(read(&mut sticky).await, "video:0", "{version}"); - // The upgrade: the same peer re-announces everything over a second session, + // The replacement: the same peer re-announces everything over a second session, // while the first one is still open. - let _quic = link(version, &publisher, &viewer).await; + let _second = link(version, &publisher, &viewer).await; moq_net_sim::sleep(WITHIN_HOLD).await; - // A request through the old handle right after the upgrade is still served. + // A request through the old handle right after the replacement is still served. let mut late = subscribe(&old, "audio") .await .unwrap_or_else(|err| panic!("{version}: the old handle refused a request before any restart: {err}")); @@ -110,10 +111,10 @@ async fn upgrade(version: &str, epoch: Option) { assert_eq!(read(&mut late).await, "audio:0", "{version}"); if epoch.is_some() { - // The epoch continues across the upgrade: nothing to restart, and the old + // The epoch continues across the replacement: nothing to restart, and the old // handle keeps serving. if let Ok(event) = moq_net_sim::timeout(PAST_HOLD, announced.next()).await { - panic!("{version}: an epoch route restarted on upgrade: {event:?}"); + panic!("{version}: an epoch route restarted on replacement: {event:?}"); } let mut later = subscribe(&old, "meta").await.unwrap(); write(&meta, "meta:0"); @@ -121,7 +122,7 @@ async fn upgrade(version: &str, epoch: Option) { return; } - // Without an epoch the upgrade is a restart: once the hold releases it, the old + // Without an epoch the replacement is a restart: once the hold releases it, the old // handle refuses new requests, and the restart is already there to follow. moq_net_sim::sleep(PAST_HOLD).await; let Err(err) = subscribe(&old, "meta").await else { @@ -141,21 +142,21 @@ async fn upgrade(version: &str, epoch: Option) { assert_eq!(read(&mut followed).await, "meta:0", "{version}"); } -/// The epoch an upgraded route carries on both sessions. +/// The epoch a replaced route carries on both sessions. fn epoch() -> Epoch { "01900000-0000-7000-8000-000000000001".parse().unwrap() } #[moq_net_sim::test] async fn epochless_lite06() { - moq_net_sim::timeout(TEST_TIMEOUT, upgrade("moq-lite-06", None)) + moq_net_sim::timeout(TEST_TIMEOUT, replace("moq-lite-06", None)) .await .expect("timed out"); } #[moq_net_sim::test] async fn epochless_lite07() { - moq_net_sim::timeout(TEST_TIMEOUT, upgrade("moq-lite-07-wip", None)) + moq_net_sim::timeout(TEST_TIMEOUT, replace("moq-lite-07-wip", None)) .await .expect("timed out"); } @@ -163,7 +164,7 @@ async fn epochless_lite07() { /// Only lite-07 carries the epoch. #[moq_net_sim::test] async fn epoch_lite07() { - moq_net_sim::timeout(TEST_TIMEOUT, upgrade("moq-lite-07-wip", Some(epoch()))) + moq_net_sim::timeout(TEST_TIMEOUT, replace("moq-lite-07-wip", Some(epoch()))) .await .expect("timed out"); } diff --git a/rs/moq-net/tests/session_setup.rs b/rs/moq-net/tests/session_setup.rs deleted file mode 100644 index cae78297c3..0000000000 --- a/rs/moq-net/tests/session_setup.rs +++ /dev/null @@ -1,195 +0,0 @@ -//! `Session::setup` waits for the server's SETUP, and reports the session's close reason. -//! -//! Deterministic: paused time over the mock transport. - -mod support; - -use std::time::Duration; - -use moq_net::{Client, Error, Server, Session, SessionError, Version, server::Handshake}; -use support::{ - harness::{now, spawn}, - mock::{MockSession, create_mock_session_pair}, -}; - -const TIMEOUT: Duration = Duration::from_secs(10); - -/// Versions whose server sends its SETUP after -/// `Client::connect` has returned. -const LATE: [&str; 9] = [ - "moq-lite-05", - "moq-lite-06", - "moq-lite-07-wip", - "moq-transport-17", - "moq-transport-18", - "moq-transport-19", - "moq-transport-20", - "moq-transport-21", - "moq-transport-22", -]; - -/// Versions whose handshake reads the server's SETUP before `Client::connect` returns. -const EARLY: [&str; 5] = [ - "moq-lite-01", - "moq-lite-02", - "moq-transport-14", - "moq-transport-15", - "moq-transport-16", -]; - -/// Versions with no server SETUP at all. -const NONE: [&str; 2] = ["moq-lite-03", "moq-lite-04"]; - -fn version(name: &str) -> Version { - name.parse().unwrap() -} - -/// Dial a `LATE` version, returning the client's session and the server's request, -/// still unanswered. -async fn dial(version: Version) -> (Session, Handshake) { - let (client, server) = create_mock_session_pair(Some(version.alpn())); - let client = async { - let (session, driver) = Client::new() - .with_versions(version.into()) - .connect(now(), client) - .await - .expect("client handshake failed"); - spawn(driver); - session - }; - let server = async { - Server::new() - .with_versions(version.into()) - .accept_request(now(), server) - .await - .expect("server never saw the request") - }; - futures::join!(client, server) -} - -async fn setup(session: &Session) -> Option> { - moq_net_sim::timeout(TIMEOUT, session.setup()).await.ok() -} - -#[moq_net_sim::test] -async fn setup_waits_for_the_server_to_answer() { - for name in LATE { - let (session, request) = dial(version(name)).await; - assert!( - setup(&session).await.is_none(), - "{name}: SETUP arrived before the server answered" - ); - - let (_server, driver) = request.ok().await.expect("server accept failed"); - spawn(driver); - assert!(matches!(setup(&session).await, Some(Ok(()))), "{name}"); - } -} - -/// A session that closed after its SETUP arrived reports the close reason: a caller that -/// checks only now must not take over a dead session. -#[moq_net_sim::test] -async fn setup_reports_a_close_after_the_setup() { - for name in LATE { - let (session, request) = dial(version(name)).await; - let (server, driver) = request.ok().await.expect("server accept failed"); - spawn(driver); - assert!(matches!(setup(&session).await, Some(Ok(()))), "{name}"); - - server.abort(Error::Unauthorized); - moq_net_sim::timeout(TIMEOUT, session.closed()) - .await - .expect("the close never arrived"); - let res = session.setup().await; - assert!(res.is_err(), "{name}: a closed session reported SETUP success"); - } -} - -#[moq_net_sim::test] -async fn setup_reports_a_refusal() { - for name in LATE { - let (session, request) = dial(version(name)).await; - request.close(Error::Unauthorized); - let err = setup(&session).await.expect("the refusal never arrived").unwrap_err(); - assert!( - matches!(err, Error::Session(SessionError::Unauthorized)), - "{name}: {err:?}" - ); - } -} - -#[moq_net_sim::test] -async fn setup_resolves_at_once_when_the_handshake_read_the_setup() { - for name in EARLY { - let version = version(name); - let (client, server) = create_mock_session_pair(Some(version.alpn())); - let client = async { - let (session, driver) = Client::new() - .with_versions(version.into()) - .connect(now(), client) - .await - .expect("client handshake failed"); - spawn(driver); - session - }; - let server = async { - let (session, driver) = Server::new() - .with_versions(version.into()) - .accept(now(), server) - .await - .expect("server handshake failed"); - spawn(driver); - session - }; - let (session, _server) = futures::join!(client, server); - let res = session.setup().await; - assert!(res.is_ok(), "{name}: {res:?}"); - } -} - -#[moq_net_sim::test] -async fn setup_is_unsupported_without_a_server_setup() { - for name in NONE { - let version = version(name); - let (client, _server) = create_mock_session_pair(Some(version.alpn())); - let (session, driver) = Client::new() - .with_versions(version.into()) - .connect(now(), client) - .await - .expect("client handshake failed"); - spawn(driver); - let res = session.setup().await; - assert!(matches!(res, Err(Error::Unsupported)), "{name}: {res:?}"); - } -} - -/// Every version is covered by exactly one of the tests above. -#[test] -fn every_version_is_classified() { - let classified: Vec<&str> = LATE.iter().chain(&EARLY).chain(&NONE).copied().collect(); - let mut all: Vec<&str> = Version::names().collect(); - all.sort_unstable(); - let mut sorted = classified.clone(); - sorted.sort_unstable(); - assert_eq!(sorted, all); -} - -/// Admission refusals use the same negotiated session registry as active sessions. -#[moq_net_sim::test] -async fn unsupported_admission_uses_the_negotiated_session_registry() { - for name in LATE { - let version = version(name); - let (session, request) = dial(version).await; - request.close(Error::Unsupported); - let err = setup(&session).await.expect("the refusal never arrived").unwrap_err(); - let expected = if version.is_lite() { - SessionError::Unsupported - } else { - SessionError::Internal - }; - assert!( - matches!(&err, Error::Session(code) if *code == expected), - "{name}: {err:?}" - ); - } -} diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index 4bb2d0ab33..ee963b8742 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -6,10 +6,8 @@ use crate::connection::Goaway; use crate::{Addrs, Backoff, Connection, Error}; -use futures::future::BoxFuture; #[cfg(all(feature = "websocket", feature = "noq"))] use std::future::Future; -use std::task::{Poll, ready}; use url::Url; /// Everything a [`Client`] is built from. @@ -332,8 +330,7 @@ impl Client { /// Dial the given URL and complete the MoQ handshake. /// /// The scheme picks the transport, and `https://` races QUIC against the - /// WebSocket fallback so a blocked UDP path still connects. When the fallback - /// wins, the QUIC dial keeps going as [`Dialed::upgrade`]. The session's + /// WebSocket fallback so a blocked UDP path still connects. The session's /// protocol driver is spawned on the current tokio runtime; the session /// closes once the last returned handle drops. #[cfg(any( @@ -353,17 +350,10 @@ impl Client { // arm was still pending. let dialed = match self.timeout.is_zero() { true => attempt.await?, - false => { - let deadline = tokio::time::Instant::now() + self.timeout; - let mut dialed = match tokio::time::timeout_at(deadline, attempt).await { - Ok(res) => res?, - Err(_) => return Err(Error::ConnectTimeout(self.timeout)), - }; - // The QUIC arm that lost to WebSocket is still part of this attempt, so the - // same deadline bounds it. - dialed.upgrade = dialed.upgrade.map(|upgrade| upgrade.until(deadline, self.timeout)); - dialed - } + false => match tokio::time::timeout(self.timeout, attempt).await { + Ok(res) => res?, + Err(_) => return Err(Error::ConnectTimeout(self.timeout)), + }, }; tracing::info!(version = %dialed.session.version(), transport = %dialed.transport, "connected"); @@ -467,18 +457,14 @@ impl Client { } #[cfg(feature = "noq")] - if let Some(noq) = self.noq.clone() { - // Owned rather than borrowed from `self`: when WebSocket wins the race, this dial - // outlives the attempt as the pending upgrade. - let tls = self.tls.clone(); - let versions = self.versions.clone(); + if let Some(noq) = self.noq.as_ref() { let quic_addr = addr.clone(); - let quic_handle = Box::pin(async move { - noq.connect(&tls, quic_addr, &versions) + let quic_handle = async { + noq.connect(&self.tls, quic_addr, &self.versions) .await .map(crate::transport::Session::new) .map_err(Error::from) - }); + }; #[cfg(feature = "websocket")] { @@ -512,10 +498,7 @@ impl Client { } /// Race the QUIC dial against the WebSocket fallback, handshaking whichever wins. - /// - /// When WebSocket wins while QUIC is still dialing, the QUIC dial carries on as - /// [`Dialed::upgrade`] rather than being dropped, and the attempt falls back to it - /// if the MoQ handshake over WebSocket fails. + /// The loser is cancelled as soon as the winner's transport is up. /// /// `moq` is the QUIC-side builder, which carries the SETUP path for a raw QUIC dial. /// The WebSocket fallback uses the plain builder: qmux over WebSocket carries the @@ -531,7 +514,7 @@ impl Client { quic: Q, ) -> crate::Result where - Q: Future> + Unpin + Send + 'static, + Q: Future>, S: moq_net::transport::poll::Boxable, { let url = addr.url().clone(); @@ -546,45 +529,11 @@ impl Client { .map(|res| res.map_err(Error::from)) }; - match race_transport_connect(quic, websocket).await? { + match race_transport_connect(&url, quic, websocket).await? { TransportRace::Quic(quic) => Ok(Dialed::new(connect_session(moq, quic).await?, transport)), - TransportRace::WebSocket { session, quic } => { - let session = match connect_session(&self.moq, crate::transport::Session::new(session)).await { - Ok(session) => session, - Err(err) => { - // The fallback got through but its MoQ handshake did not. A QUIC dial still - // pending may yet connect, bounded by the same deadline as the race. - let err = Error::from(err); - let quic = match quic { - Ok(dial) => { - tracing::warn!(%err, "WebSocket handshake failed; waiting on QUIC"); - dial.await - } - Err(quic) => Err(quic), - }; - let quic = match quic { - Ok(quic) => quic, - Err(quic) => return Err(race_error(quic, err)), - }; - // UDP gets through after all, so the next dial gives QUIC its head start. - crate::websocket::forget(&url); - // Both handshakes failing is still a two-arm loss: a mixed auth pair stays retryable. - let session = connect_session(moq, quic) - .await - .map_err(|quic| race_error(quic.into(), err))?; - return Ok(Dialed::new(session, transport)); - } - }; - let mut dialed = Dialed::new(session, crate::Transport::WebSocket); - dialed.upgrade = quic.ok().map(|quic| { - let moq = moq.clone(); - let dial = Box::pin(async move { - let quic = quic.await?; - Ok(Box::pin(async move { Ok(connect_session(&moq, quic).await?) }) as Handshake) - }); - Upgrade::new(dial, transport) - }); - Ok(dialed) + TransportRace::WebSocket(session) => { + let session = connect_session(&self.moq, crate::transport::Session::new(session)).await?; + Ok(Dialed::new(session, crate::Transport::WebSocket)) } } } @@ -661,8 +610,6 @@ pub(crate) struct Dialed { pub session: moq_net::Session, /// What `session` runs on. pub transport: crate::Transport, - /// The QUIC dial still in flight when the WebSocket fallback won the race. - pub upgrade: Option, } impl Dialed { @@ -677,112 +624,7 @@ impl Dialed { allow(dead_code) )] fn new(session: moq_net::Session, transport: crate::Transport) -> Self { - Self { - session, - transport, - upgrade: None, - } - } -} - -/// The MoQ handshake starting on an upgrade's QUIC session, up to sending our SETUP. -pub(crate) type Handshake = BoxFuture<'static, crate::Result>; - -/// A QUIC dial that lost the race to the WebSocket fallback but is still going. -/// -/// Polled by [`Connection`] alongside the WebSocket session, which it replaces -/// once both stages complete. Dropping it cancels the dial. -pub(crate) struct Upgrade { - stage: Stage, - /// What the upgraded session runs on. - transport: crate::Transport, - /// The connect deadline of the attempt this dial belongs to, and its length for - /// the error. - deadline: Option<(std::pin::Pin>, std::time::Duration)>, -} - -#[cfg_attr(not(all(feature = "websocket", feature = "noq")), allow(dead_code))] -enum Stage { - /// Waiting on the QUIC transport, and the WebTransport CONNECT for `https://`. - Dialing(BoxFuture<'static, crate::Result>), - /// The QUIC transport is up and the MoQ handshake is running on it. - Handshaking(Handshake), - /// Our SETUP is sent, and `setup` waits on the peer's. - Setup { - session: moq_net::Session, - setup: BoxFuture<'static, crate::Result<()>>, - }, -} - -/// How far an [`Upgrade`] got on one poll. -pub(crate) enum Step { - /// The QUIC transport is up; the MoQ handshake has started on it. - Handshaking, - /// The MoQ session exists and waits on the peer's SETUP. It is the caller's to close - /// if it stops before [`Step::Done`], since the peer already holds it open. - Session(moq_net::Session), - /// The peer's SETUP arrived over QUIC, which is ready to take over on this transport. - Done(moq_net::Session, crate::Transport), -} - -#[cfg_attr(not(all(feature = "websocket", feature = "noq")), allow(dead_code))] -impl Upgrade { - pub(crate) fn new(dial: BoxFuture<'static, crate::Result>, transport: crate::Transport) -> Self { - Self { - stage: Stage::Dialing(dial), - transport, - deadline: None, - } - } - - /// Fail with [`Error::ConnectTimeout`] if still pending at `deadline`. - fn until(mut self, deadline: tokio::time::Instant, timeout: std::time::Duration) -> Self { - self.deadline = Some((Box::pin(tokio::time::sleep_until(deadline)), timeout)); - self - } - - /// Drive the dial: `Ready(Ok(Step::Handshaking))` once when the QUIC transport - /// comes up, `Ready(Ok(Step::Session))` once our SETUP is sent, then - /// `Ready(Ok(Step::Done))` once the peer's SETUP arrives. Not polled again after - /// `Done` or an error. - pub(crate) fn poll(&mut self, waiter: &moq_net::kio::Waiter) -> Poll> { - if let Some((sleep, timeout)) = &mut self.deadline - && waiter.poll_future(sleep.as_mut()).is_ready() - { - return Poll::Ready(Err(Error::ConnectTimeout(*timeout))); - } - - match &mut self.stage { - Stage::Dialing(dial) => { - let handshake = ready!(waiter.poll_future(dial.as_mut()))?; - self.stage = Stage::Handshaking(handshake); - Poll::Ready(Ok(Step::Handshaking)) - } - Stage::Handshaking(handshake) => { - let session = ready!(waiter.poll_future(handshake.as_mut()))?; - let peer = session.clone(); - let setup = Box::pin(async move { - // Wait for the peer's SETUP before draining WebSocket. This crate's servers - // send SETUP after admission; other servers may still refuse afterward. - // A refusal, or a version with no SETUP to wait on, keeps WebSocket. - peer.setup().await?; - // A session already told to leave would hand straight back out of QUIC. - if peer.draining().peek().is_some() { - return Err(Error::ConnectFailed); - } - Ok(()) - }); - self.stage = Stage::Setup { - session: session.clone(), - setup, - }; - Poll::Ready(Ok(Step::Session(session))) - } - Stage::Setup { session, setup } => { - ready!(waiter.poll_future(setup.as_mut()))?; - Poll::Ready(Ok(Step::Done(session.clone(), self.transport))) - } - } + Self { session, transport } } } @@ -810,23 +652,22 @@ fn setup_authority(url: &Url) -> Option { }) } +/// The arm that won a [`race_transport_connect`]; the other was dropped, cancelling it. #[cfg(all(feature = "websocket", feature = "noq"))] -enum TransportRace { - Quic(QT), - /// The fallback won. `quic` is the QUIC dial if it was still pending, or the error - /// it had already failed with. - WebSocket { - session: WT, - quic: crate::Result, - }, +enum TransportRace { + Quic(Q), + WebSocket(W), } +/// Race the QUIC dial to `url` against the WebSocket fallback, returning whichever +/// transport comes up first. #[cfg(all(feature = "websocket", feature = "noq"))] -async fn race_transport_connect(mut quic: Q, websocket: W) -> crate::Result> +async fn race_transport_connect(url: &Url, quic: Q, websocket: W) -> crate::Result> where - Q: Future> + Unpin, + Q: Future>, W: Future>>, { + tokio::pin!(quic); tokio::pin!(websocket); let mut quic_err = None; @@ -838,7 +679,10 @@ where tokio::select! { res = &mut quic, if !quic_done => { match res { - Ok(session) => return Ok(TransportRace::Quic(session)), + Ok(session) => { + crate::websocket::forget(url); + return Ok(TransportRace::Quic(session)); + } Err(err) => { tracing::warn!(%err, "QUIC connection failed"); quic_err = Some(err); @@ -848,10 +692,7 @@ where } res = &mut websocket, if !websocket_done => { match res { - Some(Ok(session)) => { - let quic = quic_err.take().map_or(Ok(quic), Err); - return Ok(TransportRace::WebSocket { session, quic }); - } + Some(Ok(session)) => return Ok(TransportRace::WebSocket(session)), Some(Err(err)) => { tracing::warn!(%err, "WebSocket connection failed"); websocket_err = Some(err); @@ -883,7 +724,7 @@ where /// answers the fallback with 403 while QUIC is still in flight, and reconnect /// treats is_auth() as terminal, so a mixed pair reports the retryable error. #[cfg(all(feature = "websocket", feature = "noq"))] -pub(crate) fn race_error(quic: Error, websocket: Error) -> Error { +fn race_error(quic: Error, websocket: Error) -> Error { match (quic.is_auth(), websocket.is_auth()) { (false, false) => Error::TransportRace { quic: std::sync::Arc::new(quic), @@ -922,26 +763,6 @@ mod tests { moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI) } - /// A QUIC session whose peer never sends SETUP gives up at the connect deadline, so the - /// WebSocket session keeps serving instead of waiting on the upgrade forever. - #[tokio::test(start_paused = true)] - async fn a_stalled_upgrade_gives_up_at_the_deadline() { - const TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); - let waiter = moq_net::kio::Waiter::noop(); - let handshake: Handshake = Box::pin(std::future::pending()); - let mut upgrade = Upgrade::new(Box::pin(async move { Ok(handshake) }), crate::Transport::WebTransport) - .until(tokio::time::Instant::now() + TIMEOUT, TIMEOUT); - - assert!(matches!(upgrade.poll(&waiter), Poll::Ready(Ok(Step::Handshaking)))); - assert!(upgrade.poll(&waiter).is_pending()); - - tokio::time::advance(TIMEOUT).await; - assert!(matches!( - upgrade.poll(&waiter), - Poll::Ready(Err(Error::ConnectTimeout(TIMEOUT))) - )); - } - #[cfg(feature = "noq")] #[tokio::test] async fn fixed_target_preserves_request_and_refuses_redirect() { @@ -1512,43 +1333,10 @@ mod tests { Some(Ok(1usize)) }; - let value = super::race_transport_connect(Box::pin(quic), websocket).await.unwrap(); - let super::TransportRace::WebSocket { - session: 1, - quic: Err(err), - } = value - else { - panic!("WebSocket won after QUIC failed, and must carry the QUIC error"); - }; - assert!(err.is_auth(), "unexpected error: {err}"); - } - - /// A QUIC failure that lands before WebSocket connects stays with the race, so a - /// later WebSocket handshake failure combines both arms as `race_error` does. - #[cfg(all(feature = "websocket", feature = "noq"))] - #[tokio::test] - async fn race_transport_connect_keeps_an_earlier_quic_error() { - let quic = async { Err::(Error::ConnectFailed) }; - let websocket = async { - tokio::task::yield_now().await; - Some(Ok(1usize)) - }; - - let value = super::race_transport_connect(Box::pin(quic), websocket).await.unwrap(); - let super::TransportRace::WebSocket { - session: 1, - quic: Err(err), - } = value - else { - panic!("WebSocket won after QUIC failed, and must carry the QUIC error"); - }; - assert!(matches!(err, Error::ConnectFailed), "unexpected error: {err}"); - - let combined = super::race_error(err, crate::ConnectError::Unauthorized.into()); - assert!( - !combined.is_auth(), - "mixed auth/non-auth must stay retryable: {combined}" - ); + let value = super::race_transport_connect(&race_url(), quic, websocket) + .await + .unwrap(); + assert!(matches!(value, super::TransportRace::WebSocket(1))); } #[cfg(all(feature = "websocket", feature = "noq"))] @@ -1560,7 +1348,9 @@ mod tests { }; let websocket = async { Some(Err::(crate::ConnectError::Forbidden.into())) }; - let value = super::race_transport_connect(Box::pin(quic), websocket).await.unwrap(); + let value = super::race_transport_connect(&race_url(), quic, websocket) + .await + .unwrap(); assert!(matches!(value, super::TransportRace::Quic(3))); } @@ -1762,7 +1552,7 @@ mod tests { let quic = async { Err::(crate::ConnectError::Unauthorized.into()) }; let websocket = async { Some(Err::(crate::ConnectError::Forbidden.into())) }; - let Err(err) = super::race_transport_connect(Box::pin(quic), websocket).await else { + let Err(err) = super::race_transport_connect(&race_url(), quic, websocket).await else { panic!("the race must fail"); }; assert!(err.is_auth(), "unexpected error: {err}"); @@ -1777,7 +1567,7 @@ mod tests { }; let websocket = async { Some(Err::(crate::ConnectError::Forbidden.into())) }; - let Err(err) = super::race_transport_connect(Box::pin(quic), websocket).await else { + let Err(err) = super::race_transport_connect(&race_url(), quic, websocket).await else { panic!("the race must fail"); }; assert!(matches!(err, Error::ConnectFailed), "unexpected error: {err}"); @@ -1790,34 +1580,87 @@ mod tests { let quic = async { Err::(Error::ConnectFailed) }; let websocket = async { Some(Ok(7usize)) }; - let value = super::race_transport_connect(Box::pin(quic), websocket).await.unwrap(); - // `select!` may poll either arm first, so the failed QUIC dial can come back unpolled. - assert!(matches!(value, super::TransportRace::WebSocket { session: 7, .. })); + let value = super::race_transport_connect(&race_url(), quic, websocket) + .await + .unwrap(); + assert!(matches!(value, super::TransportRace::WebSocket(7))); } - /// WebSocket winning hands back the QUIC dial it beat, still running, so the - /// session can move onto QUIC once it lands. + /// The URL a race unit test dials. Each test runs in its own process under nextest, + /// so the process-wide [`crate::websocket::won`] memo starts empty. #[cfg(all(feature = "websocket", feature = "noq"))] - #[tokio::test] - async fn race_transport_connect_keeps_a_pending_quic_dial() { - let (land, landed) = tokio::sync::oneshot::channel::<()>(); - let quic = async move { - landed.await.unwrap(); - Ok("quic") + fn race_url() -> Url { + Url::parse("https://race.invalid/").unwrap() + } + + /// QUIC winning a race clears WebSocket's earlier win, so the next dial gives QUIC + /// its head start again. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[tokio::test(start_paused = true)] + async fn quic_winning_restores_its_head_start() { + crate::websocket::remember(&race_url()); + let quic = async { Ok("quic") }; + let (websocket, _dropped) = loser::>>(); + + let won = super::race_transport_connect(&race_url(), quic, websocket) + .await + .unwrap(); + assert!(matches!(won, super::TransportRace::Quic("quic"))); + assert!( + !crate::websocket::won(&race_url()), + "QUIC won, yet the next dial still skips its head start" + ); + } + + /// A race arm that never finishes, and reports through `dropped` once it is dropped. + #[cfg(all(feature = "websocket", feature = "noq"))] + fn loser() -> ( + impl Future, + tokio::sync::oneshot::Receiver, + ) { + let (alive, dropped) = tokio::sync::oneshot::channel(); + let arm = async move { + let _alive = alive; + std::future::pending().await }; + (arm, dropped) + } + + /// WebSocket winning cancels the QUIC dial before the race returns, so no QUIC + /// session can come up behind the one WebSocket serves. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[tokio::test(start_paused = true)] + async fn websocket_winning_cancels_the_quic_dial() { + let (quic, mut dropped) = loser::>(); let websocket = async { Some(Ok("websocket")) }; - let race = super::race_transport_connect(Box::pin(quic), websocket).await; - let Ok(super::TransportRace::WebSocket { - session: "websocket", - quic: Ok(quic), - }) = race - else { - panic!("WebSocket won, and the QUIC dial it beat must come back with it"); - }; + let won = super::race_transport_connect(&race_url(), quic, websocket) + .await + .unwrap(); + assert!(matches!(won, super::TransportRace::WebSocket("websocket"))); + assert_eq!( + dropped.try_recv(), + Err(tokio::sync::oneshot::error::TryRecvError::Closed), + "the QUIC dial outlived WebSocket's win" + ); + } - land.send(()).unwrap(); - assert_eq!(quic.await.unwrap(), "quic"); + /// QUIC winning cancels the WebSocket dial before the race returns. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[tokio::test(start_paused = true)] + async fn quic_winning_cancels_the_websocket_dial() { + let quic = async { Ok("quic") }; + let (websocket, mut dropped) = loser::>>(); + + let won = super::race_transport_connect(&race_url(), quic, websocket) + .await + .unwrap(); + assert!(matches!(won, super::TransportRace::Quic("quic"))); + assert_eq!( + dropped.try_recv(), + Err(tokio::sync::oneshot::error::TryRecvError::Closed), + "the WebSocket dial outlived QUIC's win" + ); } #[cfg(all(feature = "websocket", feature = "noq"))] @@ -1828,7 +1671,7 @@ mod tests { let value = tokio::time::timeout( std::time::Duration::from_secs(1), - super::race_transport_connect(Box::pin(quic), websocket), + super::race_transport_connect(&race_url(), quic, websocket), ) .await .expect("race waited for WebSocket after QUIC transport connected") diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs index 1d25f25a2e..b51f8f0cbc 100644 --- a/rs/moq-tokio/src/connection.rs +++ b/rs/moq-tokio/src/connection.rs @@ -174,8 +174,8 @@ pub enum Status { Connected, /// An established session dropped; a reconnect attempt follows. Disconnected, - /// A replacement session is coming up while the old one keeps serving: the - /// peer sent a GOAWAY, or a QUIC dial landed after the WebSocket fallback won. + /// The peer sent a GOAWAY; a replacement session is coming up while the old one + /// keeps serving. Migrating, } @@ -470,10 +470,6 @@ struct State { /// The loop's [`Draining`] predecessor and its handover deadline, so /// [`Connection::close`] drains it too without overstaying that window. predecessor: Option<(moq_net::Session, tokio::time::Instant)>, - /// A QUIC upgrade's session, from our SETUP until it takes over or is dropped, so - /// [`Connection::close`] and [`Connection::abort`] end it too. The peer already holds - /// it open, and the loop's teardown would only drop it after they return. - upgrade: Option, } /// The producer side of everything a [`Connection`] handle can observe. @@ -512,39 +508,13 @@ impl Shared { state.version = Some(session.version()); state.transport = Some(transport); state.session = Some(session.clone()); - // Whatever upgrade was coming up has taken over or is superseded. - state.upgrade = None; - } - } - - /// A QUIC upgrade's session exists, or (`None`) no longer needs ending on its own. - fn upgrading(&self, session: Option<&moq_net::Session>) { - // Held across the publish, as in `connected`. - let closed = self.closed.lock().unwrap(); - if let (Some(err), Some(session)) = (closed.as_ref(), session) { - session.abort(err.clone()); - return; - } - if let Ok(mut state) = self.state.write() { - state.upgrade = session.cloned(); } } /// The session in [`State`] keeps serving while its replacement comes up, so /// only the status moves. fn migrating(&self) { - // A closed fallback stays disconnected while its pending dial finishes. - if let Ok(mut state) = self.state.write() - && state.status != Some(Status::Disconnected) - { - state.status = Some(Status::Migrating); - } - } - - /// The replacement did not come up after all: the session in [`State`] was - /// serving throughout, so only the status moves back. - fn stayed(&self) { - self.status(Status::Connected); + self.status(Status::Migrating); } fn status(&self, status: Status) { @@ -555,7 +525,7 @@ impl Shared { /// Nothing is live. Clears the session so a handle can't be handed a closed one, /// and drops the estimates that belonged to it. - fn clear_session(&self) { + fn disconnected(&self) { if let Ok(mut state) = self.state.write() { // Count one close per live session; a second call without an // intervening connect (e.g. Drop after the loop already reported @@ -572,12 +542,6 @@ impl Shared { let _ = self.recv_bw.set(None); } - /// No live session or pending upgrade remains. - fn disconnected(&self) { - self.clear_session(); - self.upgrading(None); - } - /// Record the error the loop gave up with, for [`Connection::closed`]. fn failed(&self, err: Error) { if let Ok(mut state) = self.state.write() { @@ -777,15 +741,11 @@ impl Connection { pub fn abort(&self, err: moq_net::Error) { // Record the close and take the session under one lock, so a redial landing // in the abort window is refused rather than parked: see [`CloseGuard`]. - let (session, upgrade) = { + let session = { let mut closed = self.task.closed.lock().unwrap(); *closed = Some(err.clone()); - let state = self.state.read(); - (state.session.clone(), state.upgrade.clone()) + self.state.read().session.clone() }; - if let Some(upgrade) = upgrade { - upgrade.abort(err.clone()); - } if let Some(session) = session { session.abort(err); } @@ -794,7 +754,6 @@ impl Connection { /// Stop the loop for every clone, closing the live session (and any predecessor /// still finishing after a GOAWAY) once the data it queued has been delivered. - /// A QUIC upgrade still in its handshake has served nothing, so it closes at once. /// /// See [`moq_net::Session::close`]: finished tracks deliver their last groups and /// FIN and announcements are withdrawn first, bounded by a one second deadline @@ -803,11 +762,11 @@ impl Connection { /// when nothing was live, and a session's error if it did not drain. pub async fn close(self) -> crate::Result<()> { // Refuse redials and take the sessions under one lock: see [`CloseGuard`]. - let (session, predecessor, upgrade) = { + let (session, predecessor) = { let mut closed = self.task.closed.lock().unwrap(); *closed = Some(moq_net::Error::Cancel); let state = self.state.read(); - (state.session.clone(), state.predecessor.clone(), state.upgrade.clone()) + (state.session.clone(), state.predecessor.clone()) }; self.task.handle.abort(); let session = async move { @@ -830,14 +789,7 @@ impl Connection { } } }; - let upgrade = async move { - if let Some(upgrade) = upgrade { - upgrade.abort(moq_net::Error::Cancel); - // Within the live session's one second, so a stuck transport can't hold `close`. - let _ = tokio::time::timeout(Duration::from_secs(1), upgrade.closed()).await; - } - }; - let (session, predecessor, ()) = tokio::join!(session, predecessor, upgrade); + let (session, predecessor) = tokio::join!(session, predecessor); Ok(session.and(predecessor)?) } @@ -865,67 +817,17 @@ impl Connection { let url = addr.url().clone(); tracing::info!(peer = %Endpoint(&url), transport = %dialed.transport, "connected"); shared.connected(&dialed.session, dialed.transport); - let mut session = dialed.session; - let mut upgrade = dialed.upgrade; + let session = dialed.session; - let mut connected = tokio::time::Instant::now(); + let connected = tokio::time::Instant::now(); // Wait for the session to end, forwarding its bandwidth estimates into the // persistent producers meanwhile so consumers track the live stats across the - // connection, and draining any predecessor left over from a migration. A QUIC - // dial that lands after WebSocket won takes over here without a redial. - let (ended, tenure) = loop { - let next = run_session(shared, &session, &mut draining, &mut upgrade).await; - let tenure = connected.elapsed(); - let transport = match next { - // A short-lived fallback has not won the race against the QUIC - // attempt we already started. Its existing connect deadline bounds - // this wait, including in one-shot mode. - #[cfg(all(feature = "websocket", feature = "noq"))] - Next::Ended(Ended::Closed(closed)) if tenure < initial && upgrade.is_some() => { - match finish_upgrade(shared, &mut upgrade, &mut draining, closed).await { - Ok((next, transport)) => { - session = next; - transport - } - Err(err) => break (Ended::Closed(Err(err)), tenure), - } - } - Next::Ended(ended) => break (ended, tenure), - Next::Upgraded(next, transport) => { - // The same handover as a peer's GOAWAY, initiated by us: the new session - // is live, and the old one serves until its routes splice over at a group - // boundary or the cap closes it. - let old = std::mem::replace(&mut session, next); - shared.connected(&session, transport); - // An empty URI is legal from either endpoint on every version; the old - // session's driver closes it at the deadline if the peer lingers. - let msg = moq_net::goaway::Goaway::new().with_timeout(goaway.handover); - if let Err(err) = old.drain().send(msg) { - tracing::debug!(%err, "failed to send GOAWAY on the WebSocket session"); - } - if let Some(mut old) = draining.take() { - old.retire(); - } - draining = Some(Draining::new(old, goaway.handover, &shared.state)); - transport - } - }; - tracing::info!(peer = %Endpoint(&url), %transport, "upgraded from WebSocket"); - // UDP gets through after all, so the next dial gives QUIC its head start. - #[cfg(feature = "websocket")] - crate::websocket::forget(&url); - // Score the WebSocket tenure like any session's, so a QUIC session that - // drops right away retries in a fresh window, not one that opened - // before WebSocket served. - retry.settle(tenure, initial); - connected = tokio::time::Instant::now(); - }; - // Any upgrade still coming up is dropped with the session it was replacing. - shared.upgrading(None); + // connection, and draining any predecessor left over from a migration. + let ended = run_session(shared, &session, &mut draining).await; // A session that stayed up past the initial backoff is healthy; one that // ended sooner counts as a failed attempt however it ended. - let healthy = retry.settle(tenure, initial); + let healthy = retry.settle(connected.elapsed(), initial); // The connected target owns the policy, including in one-shot mode. A // refused redirect is terminal: the peer is leaving and named somewhere we @@ -1246,8 +1148,7 @@ impl Connection { /// What the live session runs on, or `None` while disconnected. /// - /// Changes without a disconnect when a session that came up over the WebSocket - /// fallback moves onto QUIC. + /// Fixed for a session's lifetime; a reconnect may land on a different one. pub fn transport(&self) -> Option { self.state.read().transport } @@ -1339,8 +1240,8 @@ impl Retry { } } - /// Score captured live time, excluding any wait after the session closed. - /// A tenure past `initial` is healthy and starts a fresh retry window. + /// Score how long a session served: past `initial` is healthy and starts a fresh + /// retry window. fn settle(&mut self, tenure: Duration, initial: Duration) -> bool { let healthy = tenure >= initial; if healthy { @@ -1368,15 +1269,6 @@ fn retry_wait(delay: Duration, retry_start: tokio::time::Instant, timeout: Durat Some(wait.min(timeout.checked_sub(retry_start.elapsed())?)) } -/// What [`run_session`] returns on. -enum Next { - /// The session stopped being the live one. - Ended(Ended), - /// The pending QUIC dial finished its handshake; this session, on this - /// transport, replaces the WebSocket one. - Upgraded(moq_net::Session, crate::Transport), -} - /// Why a session stopped being the live one. enum Ended { /// The transport closed. @@ -1491,17 +1383,7 @@ async fn sleep_draining(delay: Duration, draining: &mut Option, shared /// One `poll_*` step drives it all: [`poll_forward`] mirrors each kio bandwidth estimate, the /// GOAWAY consumer is a kio channel, and the transport's close future (the one non-kio source) is /// polled through the waiter's own waker. -/// -/// `upgrade` is a QUIC dial still in flight after the WebSocket fallback won. It reports -/// [`Status::Migrating`] once the QUIC transport is up and returns [`Next::Upgraded`] once -/// the peer's SETUP arrives. A failure leaves this session serving. It is dropped with -/// a healthy session otherwise; an early close still finishes the pending dial. -async fn run_session( - shared: &Shared, - session: &moq_net::Session, - draining: &mut Option, - upgrade: &mut Option, -) -> Next { +async fn run_session(shared: &Shared, session: &moq_net::Session, draining: &mut Option) -> Ended { let mut send = session.send_bandwidth(); let mut recv = session.recv_bandwidth(); let goaway = session.draining(); @@ -1525,79 +1407,16 @@ async fn run_session( // sides end up waiting on each other forever. The caller ends the connection // instead: asked to leave, with no way to migrate, leaving is the answer. if let Poll::Ready(Ok(msg)) = goaway.poll(waiter) { - return Poll::Ready(Next::Ended(Ended::Goaway(msg))); + return Poll::Ready(Ended::Goaway(msg)); } - if let Poll::Ready(err) = waiter.poll_future(closed.as_mut()) { - return Poll::Ready(Next::Ended(Ended::Closed(Err(err.into())))); - } - - match poll_upgrade(shared, upgrade, waiter) { - Poll::Ready(Ok((session, transport))) => Poll::Ready(Next::Upgraded(session, transport)), - Poll::Ready(Err(err)) => { - tracing::debug!(%err, "QUIC upgrade failed; staying on WebSocket"); - shared.stayed(); - Poll::Pending - } - Poll::Pending => Poll::Pending, - } + waiter + .poll_future(closed.as_mut()) + .map(|err| Ended::Closed(Err(err.into()))) }) .await } -/// Finish a closed fallback's pending dial, still enforcing the predecessor's handover -/// cap, and install it as the live session. Nothing serves in between, so unlike an -/// upgrade there is no fallback to send GOAWAY or drain. A failed dial folds both -/// arms' errors as the original race does. -#[cfg(all(feature = "websocket", feature = "noq"))] -async fn finish_upgrade( - shared: &Shared, - upgrade: &mut Option, - draining: &mut Option, - closed: crate::Result<()>, -) -> crate::Result<(moq_net::Session, crate::Transport)> { - shared.clear_session(); - let finished = kio::wait(|waiter| { - poll_draining(draining, waiter); - poll_upgrade(shared, upgrade, waiter) - }) - .await; - match finished { - Ok((session, transport)) => { - shared.connected(&session, transport); - Ok((session, transport)) - } - Err(quic) => Err(crate::client::race_error( - quic, - closed.err().unwrap_or_else(|| moq_net::Error::Cancel.into()), - )), - } -} - -/// Drive a pending upgrade, `Ready` with the QUIC session once the peer's SETUP arrives. -fn poll_upgrade( - shared: &Shared, - upgrade: &mut Option, - waiter: &kio::Waiter, -) -> Poll> { - while let Some(pending) = upgrade.as_mut() { - match ready!(pending.poll(waiter)) { - Ok(crate::client::Step::Handshaking) => shared.migrating(), - Ok(crate::client::Step::Session(session)) => shared.upgrading(Some(&session)), - Ok(crate::client::Step::Done(session, transport)) => { - *upgrade = None; - return Poll::Ready(Ok((session, transport))); - } - Err(err) => { - shared.upgrading(None); - *upgrade = None; - return Poll::Ready(Err(err)); - } - } - } - Poll::Pending -} - /// Mirror `bw`'s live estimate into `out` for as long as it changes, dropping the source handle once /// the session's producer is gone so we don't keep polling a dead arm. A `poll_*` step: on return, /// `waiter` is registered for the next change (unless the source is gone). Seeding is implicit @@ -2087,44 +1906,6 @@ mod tests { assert_eq!(pacing.next(Duration::from_millis(400)), Duration::from_millis(400)); } - /// A WebSocket session that served for longer than the retry window before QUIC - /// took over leaves no stale window behind: a QUIC session that drops at once is - /// a failed attempt, but one that still gets retried. - #[tokio::test(start_paused = true)] - async fn a_healthy_tenure_before_an_upgrade_starts_a_fresh_window() { - const INITIAL: Duration = Duration::from_secs(1); - const TIMEOUT: Duration = Duration::from_secs(10); - let mut retry = Retry::new(INITIAL); - - let websocket = tokio::time::Instant::now(); - tokio::time::advance(TIMEOUT * 2).await; - assert!(retry.settle(websocket.elapsed(), INITIAL)); - - let quic = tokio::time::Instant::now(); - tokio::time::advance(INITIAL / 2).await; - assert!(!retry.settle(quic.elapsed(), INITIAL)); - assert!(retry_wait(retry.delay, retry.start, TIMEOUT).is_some()); - } - - /// Waiting after a short-lived session closes must not turn it into a healthy tenure. - #[tokio::test(start_paused = true)] - async fn pending_dial_time_does_not_reset_the_retry_window() { - const INITIAL: Duration = Duration::from_secs(1); - const TIMEOUT: Duration = Duration::from_secs(10); - let mut retry = Retry::new(INITIAL); - let connected = tokio::time::Instant::now(); - tokio::time::advance(INITIAL / 2).await; - let tenure = connected.elapsed(); - assert!(tenure < INITIAL); - // The fallback is already closed while the original dial is still pending. - tokio::time::advance(TIMEOUT * 2).await; - assert!( - !retry.settle(tenure, INITIAL), - "pending dial time made a closed session healthy" - ); - assert!(retry_wait(retry.delay, retry.start, TIMEOUT).is_none()); - } - #[test] fn test_backoff_default() { let backoff = Backoff::default(); @@ -2422,153 +2203,24 @@ mod tests { assert_eq!(attempt_timeout(2, 3), None, "the last of three"); } - /// A QUIC server with the WebSocket fallback on the same port number, the QUIC - /// half reached through a [`Forwarder`], publishing `origin`. - /// - /// Yields each accepted session with the transport it arrived on. - #[cfg(all(feature = "websocket", feature = "noq"))] - struct Fallback { - url: Url, - forwarder: Forwarder, - accepted: tokio::sync::mpsc::UnboundedReceiver<(crate::Transport, moq_net::Session)>, - /// QUIC requests held unanswered in [`Quic::Refused`] mode. - held: tokio::sync::mpsc::UnboundedReceiver, - /// Keep the fallback bound but unaccepted when the test requires QUIC to win. - _websocket: Option, - } - - /// When a [`Fallback`] lets QUIC through. - #[cfg(all(feature = "websocket", feature = "noq"))] - #[derive(Clone, Copy, PartialEq)] - enum Quic { - /// From the start. - Open, - /// Once the test calls [`Forwarder::open`]. - Held, - /// Once the WebSocket fallback has closed its first session mid-handshake. No - /// MoQ server sits behind the fallback in this mode. - AfterWebSocketFails, - /// Once the test calls [`Forwarder::open`], like `Held`, and the server leaves - /// each QUIC request for the test to answer, as a session limit would refuse it. - Refused, - } - - #[cfg(all(feature = "websocket", feature = "noq"))] - impl Fallback { - async fn start(origin: &moq_net::origin::Producer, quic: Quic) -> Self { - let mut listen = crate::listen::Config { - bind: Some("127.0.0.1:0".parse().unwrap()), - ..Default::default() - }; - listen.tls.generate = vec!["localhost".into()]; - - // The fallback dials the URL's port over TCP, so the forwarder takes the same - // number over UDP. Nothing reserves the pair, so retry on a collision. - let (websocket, forwarder) = 'bind: { - for _ in 0..20 { - let websocket = crate::websocket::Listener::bind("127.0.0.1:0".parse().unwrap()) - .await - .unwrap(); - let port = websocket.local_addr().unwrap().port(); - if let Ok(front) = tokio::net::UdpSocket::bind(("127.0.0.1", port)).await { - break 'bind (websocket, front); - } - } - panic!("could not bind a matching TCP and UDP port after 20 attempts"); - }; - - // With no MoQ server behind it, the fallback's session gets as far as the MoQ - // handshake and no further. - let (websocket, failing, held_websocket) = match quic { - Quic::Open => (None, None, Some(websocket)), - Quic::AfterWebSocketFails => (None, Some(websocket), None), - _ => (Some(websocket), None, None), - }; - - let config = crate::server::Config { - listen, - websocket, - publisher: Some(origin.consume()), - ..Default::default() - }; - let mut server = config.init().unwrap().listen().await.unwrap(); - let server_addr = server.local_addr().unwrap(); - let port = forwarder.local_addr().unwrap().port(); - let forwarder = Forwarder::start(forwarder, server_addr, quic == Quic::Open).await; - - if let Some(websocket) = failing { - let gate = forwarder.clone(); - tokio::spawn(async move { - use web_transport_trait::Session as _; - let session = websocket.accept().await.unwrap().unwrap(); - // The client's SETUP stream: WebSocket has won the race. - let _setup = session.accept_bi().await.unwrap(); - session.close(1, "no MoQ here"); - gate.open(); - session.closed().await; - }); - } - - let (tx, accepted) = tokio::sync::mpsc::unbounded_channel(); - let (hold, held) = tokio::sync::mpsc::unbounded_channel(); - tokio::spawn(async move { - while let Some(request) = server.accept().await { - let transport = request.transport(); - if quic == Quic::Refused && transport != crate::Transport::WebSocket { - let _ = hold.send(request); - continue; - } - if let Ok(session) = request.ok().await { - let _ = tx.send((transport, session)); - } - } - }); - - Self { - url: url(&format!("http://127.0.0.1:{port}/")), - forwarder, - accepted, - held, - _websocket: held_websocket, - } - } - - /// The next QUIC request held in [`Quic::Refused`] mode. - async fn held(&mut self) -> crate::server::Request { - tokio::time::timeout(UPGRADE_WAIT, self.held.recv()) - .await - .expect("the QUIC request never reached the server") - .expect("the server stopped") - } - - async fn accept(&mut self) -> (crate::Transport, moq_net::Session) { - tokio::time::timeout(UPGRADE_WAIT, self.accepted.recv()) - .await - .expect("the server never accepted a session") - .expect("the server stopped") - } - } - /// Relays QUIC datagrams to `server`, holding them all until [`Forwarder::open`]. /// - /// Holding is what makes WebSocket win the race without depending on timing: the + /// Holding is what decides the transport race without depending on timing: the /// QUIC dial cannot finish until the test says so. #[cfg(all(feature = "websocket", feature = "noq"))] - #[derive(Clone)] struct Forwarder { - gate: std::sync::Arc>, + gate: tokio::sync::watch::Sender, } #[cfg(all(feature = "websocket", feature = "noq"))] impl Forwarder { - /// `open` is whether to forward from the start rather than hold. - async fn start(front: tokio::net::UdpSocket, server: std::net::SocketAddr, open: bool) -> Self { + async fn start(front: tokio::net::UdpSocket, server: std::net::SocketAddr) -> Self { use std::sync::Arc; let back = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap(); back.connect(server).await.unwrap(); let (front, back) = (Arc::new(front), Arc::new(back)); - let (tx, rx) = tokio::sync::watch::channel(open); + let (tx, rx) = tokio::sync::watch::channel(false); let (client_tx, client_rx) = tokio::sync::watch::channel(None); // Each datagram waits for the gate on its own task. @@ -2608,7 +2260,7 @@ mod tests { } }); - Self { gate: tx.into() } + Self { gate: tx } } /// Start forwarding, releasing whatever was held. @@ -2617,541 +2269,81 @@ mod tests { } } + /// When QUIC wins the race the WebSocket dial is cancelled at once: its socket + /// closes without carrying a session, and the one session runs over QUIC. #[cfg(all(feature = "websocket", feature = "noq"))] - const UPGRADE_WAIT: Duration = Duration::from_secs(10); - - /// The sequence of the next group `sub` receives. - #[cfg(all(feature = "websocket", feature = "noq"))] - async fn next_group(sub: &mut moq_net::track::Subscriber) -> u64 { - let group = tokio::time::timeout(UPGRADE_WAIT, sub.recv_group()) - .await - .expect("no group arrived") - .unwrap() - .expect("the track ended"); - group.sequence - } - - /// A replacement's pending dial cannot extend an older session's handover cap. - #[cfg(all(feature = "tcp", feature = "websocket", feature = "noq"))] #[tokio::test] - async fn closed_fallback_keeps_the_predecessor_deadline() { - let (mut server, url) = live().await; - let client = client(); - let (connection, request) = tokio::join!(client.connect(url).established(), server.accept()); - let connection = connection.unwrap(); - let _peer = request.unwrap().ok().await.unwrap(); - let session = connection.state.read().session.clone().unwrap(); - // Real sockets establish the fixture; only the handover timer is under test. - tokio::time::pause(); - let shared = shared(); - let mut draining = Some(Draining::new(session, Duration::from_secs(1), &shared.state)); - let mut upgrade = Some(crate::client::Upgrade::new( - Box::pin(std::future::pending()), - crate::Transport::WebTransport, - )); - let deadline = draining.as_ref().unwrap().deadline.deadline(); - let waiter = kio::Waiter::noop(); - let mut finishing = std::pin::pin!(finish_upgrade(&shared, &mut upgrade, &mut draining, Ok(()))); - assert!(waiter.poll_future(finishing.as_mut()).is_pending()); - assert!(shared.state.consume().read().predecessor.is_some()); - // Await the paused clock's timer driver, not only its clock adjustment. - tokio::time::sleep_until(deadline).await; - assert!(waiter.poll_future(finishing.as_mut()).is_pending()); - assert!( - shared.state.consume().read().predecessor.is_none(), - "pending QUIC kept the predecessor past its cap" - ); - } - - /// A closed fallback's dial takes over with nothing to drain, so a close racing - /// the takeover cannot wait on the dead fallback and return its error. - #[cfg(all(feature = "tcp", feature = "websocket", feature = "noq"))] - #[tokio::test] - async fn closed_fallback_takes_over_without_a_predecessor() { - let (mut server, url) = live().await; - let client = client().with_reconnect(false); - let mut held = Vec::new(); - let mut sessions = Vec::new(); - for _ in 0..2 { - let (connection, request) = tokio::join!(client.connect(url.clone()).established(), server.accept()); - let connection = connection.unwrap(); - sessions.push(connection.state.read().session.clone().unwrap()); - held.push((connection, request.unwrap().ok().await.unwrap())); - } - let quic = sessions.pop().unwrap(); - let fallback = sessions.pop().unwrap(); - - let shared = shared(); - shared.connected(&fallback, crate::Transport::WebSocket); - fallback.abort(moq_net::Error::Unauthorized); - let handshake: crate::client::Handshake = Box::pin(async move { Ok(quic) }); - let mut upgrade = Some(crate::client::Upgrade::new( - Box::pin(async move { Ok(handshake) }), - crate::Transport::WebTransport, - )); - let mut draining = None; - let closed = Err(moq_net::Error::Unauthorized.into()); - let (_, transport) = finish_upgrade(&shared, &mut upgrade, &mut draining, closed) - .await - .unwrap(); - assert_eq!(transport, crate::Transport::WebTransport); - - // The state a close woken by `Connected` reads. - let state = shared.state.consume(); - let state = state.read(); - assert_eq!(state.status, Some(Status::Connected)); - assert_eq!(state.transport, Some(crate::Transport::WebTransport)); - assert!( - state.predecessor.is_none(), - "the closed fallback was published as a predecessor" - ); - assert!(draining.is_none()); - } - - /// The upgrade reports [`Status::Migrating`] while the MoQ handshake runs on the - /// QUIC session, and a failure there puts the status back: the WebSocket session - /// served throughout. - /// - /// Pinned here rather than end to end, since a moq-lite client completes its - /// handshake without waiting on the server, leaving no window to observe. - #[test] - fn a_failed_upgrade_stays_connected() { - let shared = shared(); - shared.status(Status::Connected); - let waiter = kio::Waiter::noop(); - - let (fail, failed) = tokio::sync::oneshot::channel::<()>(); - let handshake: crate::client::Handshake = Box::pin(async move { - let _ = failed.await; - Err(Error::ConnectFailed) - }); - let mut upgrade = Some(crate::client::Upgrade::new( - Box::pin(async move { Ok(handshake) }), - crate::Transport::WebTransport, - )); - - assert!(poll_upgrade(&shared, &mut upgrade, &waiter).is_pending()); - assert_eq!(shared.state.consume().read().status, Some(Status::Migrating)); - assert!(upgrade.is_some()); - - fail.send(()).unwrap(); - assert!(matches!( - poll_upgrade(&shared, &mut upgrade, &waiter), - Poll::Ready(Err(Error::ConnectFailed)) - )); - // The caller decides whether a live WebSocket stays or an ended arm loses - // the transport race. Polling the upgrade itself only reports the failure. - shared.stayed(); - assert_eq!(shared.state.consume().read().status, Some(Status::Connected)); - assert!(upgrade.is_none(), "a failed upgrade is not retried"); - } - - /// A session that came up over the WebSocket fallback moves onto QUIC once the - /// QUIC dial lands, without dropping a group, and forgets that WebSocket won. - #[cfg(all(feature = "websocket", feature = "noq"))] - #[tracing_test::traced_test] - #[tokio::test] - async fn websocket_upgrades_to_quic() { - const HANDOVER: Duration = Duration::from_millis(500); + async fn quic_winning_cancels_the_websocket_dial() { + use tokio::io::AsyncReadExt; + const WAIT: Duration = Duration::from_secs(10); - let origin = crate::origin::spawn(); - let broadcast = origin.create_broadcast("cam").unwrap(); - broadcast.announce(Default::default()).unwrap(); - let track = broadcast.create_track("video", timed()).unwrap(); - let write = |payload: &'static [u8]| { - let mut group = track.append_group().unwrap(); - group.write_frame(moq_net::Timestamp::ZERO, payload).unwrap(); - group.finish().unwrap(); + let mut listen = crate::listen::Config { + bind: Some("127.0.0.1:0".parse().unwrap()), + ..Default::default() }; - - let mut fallback = Fallback::start(&origin, Quic::Held).await; - - let subscriber = crate::origin::spawn(); - let mut config = crate::connect::Config::default(); - config.tls.insecure = Some(true); - config.goaway.handover = HANDOVER; - let client = config - .init(Default::default()) - .unwrap() - .with_subscriber(subscriber.clone()); - let connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) - .await - .expect("never connected") - .unwrap(); - - // QUIC is held, so the fallback won. - assert_eq!(connection.transport(), Some(crate::Transport::WebSocket)); - assert!( - crate::websocket::won(&fallback.url), - "the fallback's win was not remembered" - ); - let mut monitor = connection.monitor(); - let (transport, websocket) = fallback.accept().await; - assert_eq!(transport, crate::Transport::WebSocket); - - let cam = tokio::time::timeout(UPGRADE_WAIT, subscriber.consume().routed_broadcast("cam")) - .await - .unwrap() - .unwrap(); - let mut sub = cam.track("video").unwrap().subscribe(None).await.unwrap(); - write(b"g0"); - assert_eq!(next_group(&mut sub).await, 0); - - // Let QUIC through, and publish while the replacement comes up. - fallback.forwarder.open(); - write(b"g1"); - - // The swap is a new session on the same handle, with no disconnect between. - let presence = tokio::time::timeout(UPGRADE_WAIT, async { - loop { - let presence = monitor.presence_changed().await.unwrap(); - if presence.sessions_started == 2 { - return presence; - } - } - }) - .await - .expect("never upgraded"); - assert_eq!( - presence.sessions_ended, 1, - "the WebSocket session was not counted as ended" - ); - assert!(connection.connected()); - assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); - assert_eq!(connection.connects(), 2); - let (transport, _quic) = fallback.accept().await; - assert_eq!(transport, crate::Transport::WebTransport); - // QUIC works on this network, so the next dial gives it the head start again. - assert!( - !crate::websocket::won(&fallback.url), - "the upgrade kept WebSocket's win" - ); - - // Every group arrives exactly once across the swap: the one written while QUIC - // came up and the one after it. - write(b"g2"); - assert_eq!(next_group(&mut sub).await, 1); - assert_eq!(next_group(&mut sub).await, 2); - - // The WebSocket session is told to leave, and closes within the handover cap - // rather than lingering alongside QUIC. - let goaway = websocket.draining(); - let goaway = tokio::time::timeout(UPGRADE_WAIT, kio::wait(|waiter| goaway.poll(waiter))) - .await - .expect("the WebSocket session never received a GOAWAY") - .unwrap(); - assert_eq!(goaway.uri(), "", "a client may not redirect its server"); - tokio::time::timeout(UPGRADE_WAIT, websocket.closed()) - .await - .expect("the WebSocket session outlived the handover cap"); - assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); - // The GOAWAY got through, so its sender has nothing to warn about. - assert!(!logs_contain("failed to send goaway")); - } - - /// A QUIC session the server refuses after its transport came up, as a session - /// limit would while the WebSocket session holds the slot, leaves the WebSocket - /// session serving: no GOAWAY, no disconnect, and the next group arrives over it. - #[cfg(all(feature = "websocket", feature = "noq"))] - #[tokio::test] - async fn a_refused_upgrade_keeps_websocket() { - let origin = crate::origin::spawn(); - let broadcast = origin.create_broadcast("cam").unwrap(); - broadcast.announce(Default::default()).unwrap(); - let track = broadcast.create_track("video", timed()).unwrap(); - let write = |payload: &'static [u8]| { - let mut group = track.append_group().unwrap(); - group.write_frame(moq_net::Timestamp::ZERO, payload).unwrap(); - group.finish().unwrap(); + listen.tls.generate = vec!["localhost".into()]; + let config = crate::server::Config { + listen, + ..Default::default() }; - - let mut fallback = Fallback::start(&origin, Quic::Refused).await; - - let subscriber = crate::origin::spawn(); - let mut config = crate::connect::Config::default(); - config.tls.insecure = Some(true); - let client = config - .init(Default::default()) - .unwrap() - .with_subscriber(subscriber.clone()); - let mut connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) - .await - .expect("never connected") - .unwrap(); - assert_eq!(connection.transport(), Some(crate::Transport::WebSocket)); - assert_eq!(connection.status().await.unwrap(), Status::Connected); - let (_, websocket) = fallback.accept().await; - - let cam = tokio::time::timeout(UPGRADE_WAIT, subscriber.consume().routed_broadcast("cam")) - .await - .unwrap() - .unwrap(); - let mut sub = cam.track("video").unwrap().subscribe(None).await.unwrap(); - write(b"g0"); - assert_eq!(next_group(&mut sub).await, 0); - - // The QUIC transport comes up and its MoQ handshake reaches the server, which - // holds the request: the WebSocket session must not be drained yet. - fallback.forwarder.open(); - let request = fallback.held().await; - assert_eq!(connection.transport(), Some(crate::Transport::WebSocket)); - assert!(!connection.connected(), "swapped onto QUIC before it was admitted"); - let status = tokio::time::timeout(UPGRADE_WAIT, connection.status()).await.unwrap(); - assert_eq!(status.unwrap(), Status::Migrating); - assert!( - websocket.draining().peek().is_none(), - "drained before QUIC was admitted" - ); - - request.reject(crate::server::Reject::Forbidden).await.unwrap(); - let status = tokio::time::timeout(UPGRADE_WAIT, connection.status()).await.unwrap(); - assert_eq!(status.unwrap(), Status::Connected); - assert_eq!(connection.transport(), Some(crate::Transport::WebSocket)); - assert_eq!(connection.connects(), 1); - assert!( - websocket.draining().peek().is_none(), - "the refused upgrade drained WebSocket" - ); - - write(b"g1"); - assert_eq!(next_group(&mut sub).await, 1); - } - - /// Aborting while a QUIC upgrade is still in its MoQ handshake ends the upgrade's - /// session too, with the caller's error, before `abort` returns. - /// - /// Left to the loop's teardown, the session would only be dropped once the aborted - /// task unwound, with a bare cancel. A process that exits in between, as an embedder - /// does right after `abort` or `close`, leaves the server to time the connection out. - /// The error the server reads is what tells the two paths apart. - #[cfg(all(feature = "websocket", feature = "noq"))] - #[tokio::test] - async fn abort_ends_an_upgrade_in_flight() { - for close_fallback in [false, true] { - let origin = crate::origin::spawn(); - let mut fallback = Fallback::start(&origin, Quic::Refused).await; - - let mut config = crate::connect::Config::default(); - config.tls.insecure = Some(true); - let client = config - .init(Default::default()) - .unwrap() - .with_subscriber(crate::origin::spawn()); - let mut connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) - .await - .expect("never connected") - .unwrap(); - assert_eq!(connection.transport(), Some(crate::Transport::WebSocket)); - let (_, websocket) = fallback.accept().await; - - // The QUIC transport comes up and its handshake waits on the server's answer. - fallback.forwarder.open(); - let request = fallback.held().await; - let status = tokio::time::timeout(UPGRADE_WAIT, connection.status()).await.unwrap(); - assert_eq!(status.unwrap(), Status::Migrating); - - if close_fallback { - let mut monitor = connection.monitor(); - websocket.abort(moq_net::Error::Cancel); - loop { - let presence = tokio::time::timeout(UPGRADE_WAIT, monitor.presence_changed()) - .await - .expect("closed fallback remains live") - .unwrap(); - if presence.sessions_ended == 1 { - break; - } + let mut server = config.init().unwrap().listen().await.unwrap(); + let server_addr = server.local_addr().unwrap(); + let (accepted, mut sessions) = tokio::sync::mpsc::unbounded_channel(); + tokio::spawn(async move { + while let Some(request) = server.accept().await { + let transport = request.transport(); + if let Ok(session) = request.ok().await { + let _ = accepted.send((transport, session)); } - assert_eq!(connection.transport(), None); } - connection.abort(moq_net::Error::GoawayTimeout); - let closed = async { - match request.ok().await { - Ok(session) => session.closed().await, - Err(err) => panic!("the server could not answer the upgrade: {err}"), - } - }; - let err = tokio::time::timeout(UPGRADE_WAIT, closed) - .await - .expect("the upgrade's connection was left open"); - let code = |err: &moq_net::Error| moq_net::SessionError::from(err).to_code(); - assert_eq!( - code(&err), - code(&moq_net::Error::GoawayTimeout), - "the upgrade ended without the abort's error: {err}" - ); - } - } - - /// Closing while a QUIC upgrade is still in its MoQ handshake ends the upgrade too, and - /// returns promptly rather than waiting on it. - #[cfg(all(feature = "websocket", feature = "noq"))] - #[tokio::test] - async fn close_ends_an_upgrade_in_flight() { - let origin = crate::origin::spawn(); - let mut fallback = Fallback::start(&origin, Quic::Refused).await; - - let mut config = crate::connect::Config::default(); - config.tls.insecure = Some(true); - let client = config - .init(Default::default()) - .unwrap() - .with_subscriber(crate::origin::spawn()); - let mut connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) - .await - .expect("never connected") - .unwrap(); - let _websocket = fallback.accept().await; - - fallback.forwarder.open(); - let request = fallback.held().await; - let status = tokio::time::timeout(UPGRADE_WAIT, connection.status()).await.unwrap(); - assert_eq!(status.unwrap(), Status::Migrating); - - // Inside the one second the live session may take to drain, plus slack. - tokio::time::timeout(Duration::from_secs(2), connection.close()) - .await - .expect("close waited on the upgrade") - .unwrap(); - let closed = async { - if let Ok(session) = request.ok().await { - session.closed().await; - } - }; - tokio::time::timeout(UPGRADE_WAIT, closed) - .await - .expect("the upgrade's connection was left open"); - } - - /// When WebSocket wins the race but its MoQ handshake fails, the attempt falls back - /// to the QUIC dial still pending instead of failing. - /// - /// One-shot, so a redial cannot stand in for the fallback. Draft 16, since it is the - /// newest version whose client handshake waits on the server's SETUP. - #[cfg(all(feature = "websocket", feature = "noq"))] - #[tokio::test] - async fn failed_websocket_handshake_falls_back_to_quic() { - let origin = crate::origin::spawn(); - let mut fallback = Fallback::start(&origin, Quic::AfterWebSocketFails).await; - - let mut config = crate::connect::Config::default(); - config.tls.insecure = Some(true); - config.version = vec!["moq-transport-16".parse().unwrap()]; - let client = config.init(Default::default()).unwrap().with_reconnect(false); - let connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) - .await - .expect("never connected") - .expect("a failed WebSocket handshake must fall back to the pending QUIC dial"); - - assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); - assert_eq!(connection.connects(), 1); - let (transport, _session) = fallback.accept().await; - assert_eq!(transport, crate::Transport::WebTransport); - // QUIC works on this network, so the next dial gives it the head start again. - assert!( - !crate::websocket::won(&fallback.url), - "the fallback kept WebSocket's win" - ); - } + }); - /// One-shot proves this reuses the held dial rather than redialing QUIC. - #[cfg(all(feature = "websocket", feature = "noq"))] - #[tokio::test] - async fn closed_websocket_session_falls_back_to_pending_quic() { - for error in [moq_net::Error::Cancel, moq_net::Error::Unauthorized] { - let origin = crate::origin::spawn(); - let mut fallback = Fallback::start(&origin, Quic::Held).await; - let mut config = crate::connect::Config::default(); - config.tls.insecure = Some(true); - let client = config.init(Default::default()).unwrap().with_reconnect(false); - let mut connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) - .await - .expect("never connected") - .unwrap(); - assert_eq!(connection.transport(), Some(crate::Transport::WebSocket)); - let (_, websocket) = fallback.accept().await; - websocket.abort(error.clone()); - let _ = websocket.closed().await; - loop { - let status = tokio::time::timeout(UPGRADE_WAIT, connection.status()) - .await - .expect("closed fallback never reported its pending dial") - .expect("connection ended before pending QUIC dial"); - if status == Status::Disconnected { - break; + // The fallback dials the URL's port over TCP, and QUIC the same number over UDP, + // which the forwarder takes. Nothing reserves the pair, so retry on a collision. + let (fallback, front) = 'bind: { + for _ in 0..20 { + let fallback = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = fallback.local_addr().unwrap().port(); + if let Ok(front) = tokio::net::UdpSocket::bind(("127.0.0.1", port)).await { + break 'bind (fallback, front); } } - assert_eq!(connection.transport(), None, "the closed fallback is no longer live"); - assert!( - connection.monitor().stats().is_none(), - "closed fallback stats remain visible" - ); - fallback.forwarder.open(); - loop { - if connection.transport() == Some(crate::Transport::WebTransport) { - break; - } - tokio::time::timeout(UPGRADE_WAIT, connection.status()) - .await - .expect("pending QUIC dial never took over") - .expect("connection ended before pending QUIC dial"); - } - let (transport, _quic) = fallback.accept().await; - assert_eq!(transport, crate::Transport::WebTransport); - assert_eq!(connection.connects(), 2); - // The dead fallback's error does not leak into closing the QUIC session. - tokio::time::timeout(UPGRADE_WAIT, connection.close()) - .await - .expect("close never finished") - .expect("close returned the closed fallback's error"); - } - } - - /// When QUIC wins the race nothing changes: one session, over QUIC, and no - /// WebSocket session is ever opened. - #[cfg(all(feature = "websocket", feature = "noq"))] - #[tokio::test] - async fn quic_winning_opens_one_session() { - let origin = crate::origin::spawn(); - let broadcast = origin.create_broadcast("cam").unwrap(); - broadcast.announce(Default::default()).unwrap(); - let track = broadcast.create_track("video", timed()).unwrap(); - - let mut fallback = Fallback::start(&origin, Quic::Open).await; + panic!("could not bind a matching TCP and UDP port after 20 attempts"); + }; + let port = front.local_addr().unwrap().port(); + let forwarder = Forwarder::start(front, server_addr).await; + + // Take the fallback's socket but never answer its upgrade, and only then let QUIC + // through: QUIC wins with both dials under way. The socket closing is the client + // cancelling the dial that lost. + let cancelled = tokio::spawn(async move { + let (mut socket, _) = fallback.accept().await.unwrap(); + forwarder.open(); + let mut buf = [0; 1024]; + while matches!(socket.read(&mut buf).await, Ok(len) if len > 0) {} + }); - let subscriber = crate::origin::spawn(); let mut config = crate::connect::Config::default(); config.tls.insecure = Some(true); - // Both dials start now; the fixture holds the fallback handshake so load - // cannot decide which transport this test exercises. config.websocket.delay = Duration::ZERO; - let client = config - .init(Default::default()) - .unwrap() - .with_subscriber(subscriber.clone()); - let connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) + let client = config.init(Default::default()).unwrap(); + let url = url(&format!("http://127.0.0.1:{port}/")); + let connection = tokio::time::timeout(WAIT, client.connect(url).established()) .await .expect("never connected") .unwrap(); assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); - let (transport, _session) = fallback.accept().await; + let (transport, _session) = tokio::time::timeout(WAIT, sessions.recv()) + .await + .expect("the server never accepted a session") + .unwrap(); assert_eq!(transport, crate::Transport::WebTransport); - let cam = tokio::time::timeout(UPGRADE_WAIT, subscriber.consume().routed_broadcast("cam")) + tokio::time::timeout(WAIT, cancelled) .await - .unwrap() + .expect("the WebSocket dial outlived QUIC's win") .unwrap(); - let mut sub = cam.track("video").unwrap().subscribe(None).await.unwrap(); - let mut group = track.append_group().unwrap(); - group.write_frame(moq_net::Timestamp::ZERO, b"g0".as_ref()).unwrap(); - group.finish().unwrap(); - assert_eq!(next_group(&mut sub).await, 0); - - // QUIC's win dropped the fallback before it handshook, so there is nothing else to - // accept and nothing to upgrade from. - assert!(fallback.accepted.try_recv().is_err(), "a second session was opened"); assert_eq!(connection.connects(), 1); - assert!(!crate::websocket::won(&fallback.url)); } /// A stream-only server on a free loopback port, publishing `origin`. diff --git a/rs/moq-tokio/src/websocket.rs b/rs/moq-tokio/src/websocket.rs index 7bcc7efa71..33007983fb 100644 --- a/rs/moq-tokio/src/websocket.rs +++ b/rs/moq-tokio/src/websocket.rs @@ -3,8 +3,8 @@ //! Used when QUIC is unreachable: UDP blocked by a firewall, a proxy in the way, a //! network that only passes TCP/443. The client races this against QUIC and gives QUIC //! a small head start ([`Config::delay`]), so WebSocket only wins when QUIC can't get -//! through. A [`crate::Connection`] that lands on WebSocket keeps the QUIC dial going and -//! moves onto it if it completes. Servers accept it on a separate TCP port via [`Listener`]. +//! through. A session that lands on WebSocket stays on it for its lifetime. Servers +//! accept it on a separate TCP port via [`Listener`]. use qmux::ws::tokio_tungstenite; use qmux::ws::tokio_tungstenite::tungstenite::{self, client::IntoClientRequest, http}; @@ -108,7 +108,10 @@ impl Error { type Result = std::result::Result; -// Track servers (hostname:port) where WebSocket won the race, so we won't give QUIC a headstart next time +/// Servers (host, port) where WebSocket won the race, so later dials skip QUIC's head start. +/// +/// Both arms still race on those dials, and a QUIC win clears the entry through [`forget`], +/// since UDP gets through after all. static WEBSOCKET_WON: LazyLock>> = LazyLock::new(|| Mutex::new(HashSet::new())); /// The [`WEBSOCKET_WON`] key for a dial URL: the host and port the fallback dials. @@ -123,14 +126,19 @@ fn won_key(url: &Url) -> Result<(String, u16)> { } /// Forget that WebSocket won for `url`, giving QUIC its head start again. -/// -/// Called once a QUIC dial to `url` lands after all, which proves UDP gets through. +#[cfg(feature = "noq")] pub(crate) fn forget(url: &Url) { if let Ok(key) = won_key(url) { WEBSOCKET_WON.lock().unwrap().remove(&key); } } +/// Record that WebSocket won for `url`, as a fallback dial that lands does. +#[cfg(all(test, feature = "noq"))] +pub(crate) fn remember(url: &Url) { + WEBSOCKET_WON.lock().unwrap().insert(won_key(url).unwrap()); +} + /// Whether the next dial to `url` skips QUIC's head start. #[cfg(all(test, feature = "noq"))] pub(crate) fn won(url: &Url) -> bool { diff --git a/test/portrait/relay.toml b/test/portrait/relay.toml index aa127e3c42..98f519cf42 100644 --- a/test/portrait/relay.toml +++ b/test/portrait/relay.toml @@ -6,7 +6,7 @@ level = "info" bind = "127.0.0.1:4443" tls.generate = ["localhost", "127.0.0.1"] # The CLI peers negotiate lite-07 so their epochs carry an initial request across a -# WebSocket-to-QUIC upgrade; lite-06 refuses it as a new instance. Chromium offers lite-06. +# replacement session; lite-06 refuses it as a new instance. Chromium offers lite-06. version = ["moq-lite-07-wip", "moq-lite-06"] [web.http] diff --git a/test/ts/README.md b/test/ts/README.md index 48398d1ef2..997c065a9a 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -595,12 +595,10 @@ claiming the CBR PCR-accuracy profile. The round-trip rig explicitly negotiates `moq-lite-07-wip` on the relay, publisher and subscribers. Its publisher epochs allow acquisition to continue -when WebSocket upgrades to QUIC. Production defaults remain unchanged: Lite-06 +when a session is replaced. Production defaults remain unchanged: Lite-06 cannot prove publisher identity across session replacement and must refuse a pending acquisition. `rs/moq-net/tests/catalog_acquisition.rs` covers both outcomes deterministically in Rust CI, without retries or implicit `--stitch`. -Actual transport teardown coverage belongs to the transport-upgrade tests; a -clean TS run alone does not prove that an upgrade occurred. [`.github/workflows/ts.yml`](../../.github/workflows/ts.yml) is independent of browser interop. Relevant mux, Hang, CLI, transport, harness and toolchain PRs run