diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 5da471366b..fae2e659a5 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -32,7 +32,6 @@ Public API: none. Wire: none. ## Required -- [Relay restart rebind](/quest/m1/test-flakes-2/relay-restart-rebind.md) - the crash drill restarts on its original UDP address under concurrent load - [Impaired handshake](/quest/m1/test-flakes-2/impaired-handshake.md) - the impaired drills' clients never time out while connecting - [Interop close code](/quest/m1/test-flakes-2/interop-close-code.md) - the browser close-code test reliably sees `unauthorized` for a refused session - [TS duration fidelity](/quest/m1/test-flakes-2/ts-duration-fidelity.md) - TS compliance captures the whole round-tripped stream diff --git a/quest/m1/test-flakes-2/relay-restart-rebind.md b/quest/m1/test-flakes-2/relay-restart-rebind.md deleted file mode 100644 index 108cd7afbc..0000000000 --- a/quest/m1/test-flakes-2/relay-restart-rebind.md +++ /dev/null @@ -1,49 +0,0 @@ -# [S] Relay crash drill rebinds its address under load - -## Goal - -`relay_killed_mid_group_aborts_then_resumes` reliably restarts its relay on -its original UDP address in the concurrent real-QUIC suite. The intermittent -`Address already in use` failure is reproduced and fixed at its cause, -while the interrupted group's failure and both clients' reconnect assertions -remain meaningful. - -## Plan - -Found during #4928's loaded `just check` on 2026-10-06, in unchanged code: -the impaired lane failed to load the replacement relay after recording that -the interrupted group aborted. Seed `16515990072133439138` failed in the -suite but passed alone, and a later full check passed. A passing rerun is -not evidence that the cause is gone. - -- Measure first the likeliest cause: the relay's first port comes from - `relay_config(None)`, so it is ephemeral, and it sits unbound between - `RelayHost::kill` and the restart. The kill sends no CONNECTION_CLOSE, so - clients only notice the loss by timeout, and in that window another test's - `bind(0)` under load can take the port. -- Reproduce under concurrent checks and capture which socket owns the address - between `RelayHost::kill` and restart. Distinguish that competing ephemeral - bind from retained ownership after runtime shutdown; do not assume the seed - alone reproduces a scheduling race. -- Fix the ownership or allocation race at its source. Fixture-level fixes are - in scope, such as a stable client-facing address (a forwarder, for - example) so the restarted relay can take a fresh port. Preserve the crash - semantics and real reconnect coverage; retrying the bind, sleeping, - widening timeouts, or substituting graceful shutdown stay out. -- Follow this questline's event-based testing rules. Add a regression that - fails without the fix and run the existing drill in both lanes through CI. - Coordinate with the socket-close work if the cause is in the endpoint; - do not expand this quest into that API refactor without evidence. - -Decided during the 2026-10-06 quest-complete sweep: the maintainer selected -separate investigation, then authorized the recommended course unattended. -Keep #4928's documentation correction focused. This is an m1 child of the -existing load-flake questline because it shares that validation goal. - -Public API: none expected. Wire: none. Any discovered product impact belongs -in the implementation PR's Impact section. No new user-facing guide is needed -for this fixture investigation; update any comments the fix makes stale inline. - -## Related - -- [Socket close](/quest/m2/noq-socket-close.md) - replaces the closable socket wrapper, rather than assuming runtime shutdown released it diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index dbcb46c276..a03e1e0e8b 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -118,7 +118,7 @@ impl Path { async fn through(relay: &RelayHost, profile: moq_shaper::Profile, seed: u64) -> Self { let shaper = moq_shaper::Shaper::bind(moq_shaper::Config { bind: "127.0.0.1:0".parse().unwrap(), - target: format!("127.0.0.1:{}", relay.port).parse().unwrap(), + target: relay.addr(), seed, up: profile.clone(), down: profile, @@ -133,6 +133,16 @@ impl Path { } } + /// Forward to `relay` from now on, so clients reach it at the address they + /// already dial. + fn retarget(&self, relay: &RelayHost) { + self.shaper + .as_ref() + .expect("only a path through a shaper can move") + .retarget(relay.addr()) + .expect("retarget the shaper"); + } + /// Fail the drill unless the impairment it ran under actually acted. /// /// A profile that silently did nothing would turn this lane into a second @@ -239,18 +249,23 @@ impl RelayHost { .expect("relay shutdown panicked"); } + /// The relay's UDP address. + fn addr(&self) -> std::net::SocketAddr { + std::net::SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, self.port)) + } + /// The URL clients dial. `https` is WebTransport over QUIC. fn url(&self) -> url::Url { - format!("https://127.0.0.1:{}/drill", self.port) + format!("https://{}/drill", self.addr()) .parse() .expect("parse relay url") } } -/// A public relay on loopback with a generated certificate, on `port` or an ephemeral one. -fn relay_config(port: Option) -> Config { +/// A public relay on an ephemeral loopback port with a generated certificate. +fn relay_config() -> Config { let mut config = Config::default(); - config.listen.bind = Some(format!("127.0.0.1:{}", port.unwrap_or_default()).parse().unwrap()); + config.listen.bind = Some("127.0.0.1:0".parse().unwrap()); config.listen.tls.generate = vec!["localhost".into()]; config.auth.public = vec![moq_auth::Pattern::all()]; config @@ -412,7 +427,7 @@ async fn subscribe(origin: &moq_net::origin::Consumer, path: &str) -> Reader { /// keeps being served. The rejoin below is the half of that behavior worth /// grading here. async fn cancel_under_backpressure_releases_the_reader(lane: Lane) { - let relay = RelayHost::start(relay_config(None)).await; + let relay = RelayHost::start(relay_config()).await; let path = Path::start(lane, &relay).await; let url = path.url.clone(); @@ -533,10 +548,18 @@ lanes!(cancel_under_backpressure_releases_the_reader); /// into one is the worst possible failure mode. And the reconnect loop has to /// restore service: the dead session's broadcast closes with it, and a fresh /// subscribe through the same origin resumes once the relay returns. +/// +/// Clients dial a shaper in both lanes, unimpaired on loopback, so the relay +/// can come back on a fresh port behind the address they already dial. The +/// dead relay's port is free for any process's ephemeral bind to take, so a +/// restart that needed it back would fail whenever one did. async fn relay_killed_mid_group_aborts_then_resumes(lane: Lane) { - let mut relay = RelayHost::start(relay_config(None)).await; - let port = relay.port; - let path = Path::start(lane, &relay).await; + let mut relay = RelayHost::start(relay_config()).await; + let profile = match lane { + Lane::Loopback => Default::default(), + Lane::Impaired => impairment(), + }; + let path = Path::through(&relay, profile, seed(lane)).await; let url = path.url.clone(); let publisher = moq_tokio::origin::spawn(); @@ -572,6 +595,9 @@ async fn relay_killed_mid_group_aborts_then_resumes(lane: Lane) { .expect("open group aborted") .expect("open group finished"); + // Start the replacement while the original still holds its port, so the two + // can never share one. Clients keep dialing the dead relay until the retarget. + let replacement = RelayHost::start(relay_config()).await; relay.kill().await; // Terminal result: the unread half of the open group fails. `Ok(None)` here @@ -586,8 +612,8 @@ async fn relay_killed_mid_group_aborts_then_resumes(lane: Lane) { expect_status(&mut subscribe_loop, moq_tokio::Status::Disconnected, "subscriber").await; expect_status(&mut publish_loop, moq_tokio::Status::Disconnected, "publisher").await; - let relay = RelayHost::start(relay_config(Some(port))).await; - assert_eq!(relay.port, port, "relay bound the wrong port"); + let relay = replacement; + path.retarget(&relay); expect_status(&mut subscribe_loop, moq_tokio::Status::Connected, "subscriber").await; expect_status(&mut publish_loop, moq_tokio::Status::Connected, "publisher").await; @@ -640,7 +666,7 @@ async fn expect_status(reconnect: &mut moq_tokio::Connection, want: moq_tokio::S /// subscriber that re-consumes the same name after a republish gets what the /// new publisher is sending, never the previous one's cache. async fn interrupted_publisher_republishes_new_content(lane: Lane) { - let relay = RelayHost::start(relay_config(None)).await; + let relay = RelayHost::start(relay_config()).await; let path = Path::start(lane, &relay).await; let url = path.url.clone(); @@ -927,7 +953,7 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { // bound each relay's side of an impaired handshake, which can sit silent // that long between the client's backed-off retransmits. So a cut peer link // idles out only after `relay_idle`, which the flap's waits allow for. - let mut config = relay_config(None); + let mut config = relay_config(); config.quic = credit(lane, config.quic); let relay_idle = config.quic.idle_timeout; let origin = RelayHost::start(config).await; @@ -942,7 +968,7 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { // The edge relay dials the origin as a cluster peer, through its own path. let mut peer_url = peer.url.clone(); peer_url.set_path("/"); - let mut config = relay_config(None); + let mut config = relay_config(); config.quic = credit(lane, config.quic); config.connect = dial(); // Retry forever: a give-up after a slow impaired handshake would look exactly @@ -1234,7 +1260,7 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { /// harness with no publisher and requires that no announcement and no broadcast /// ever arrive, so "the frame showed up" cannot be a harness artifact. async fn no_publisher_never_delivers(lane: Lane) { - let relay = RelayHost::start(relay_config(None)).await; + let relay = RelayHost::start(relay_config()).await; let path = Path::start(lane, &relay).await; let url = path.url.clone(); diff --git a/rs/moq-shaper/README.md b/rs/moq-shaper/README.md index 9367721e4c..fd3cf585e3 100644 --- a/rs/moq-shaper/README.md +++ b/rs/moq-shaper/README.md @@ -47,7 +47,9 @@ default. `Shaper::verify` fails when the shaper stopped forwarding, when an impa configures never acted and the traffic makes that silence implausible, or when a direction carried traffic but no datagram saw one of its phases, before the first step or after one. `Shaper::cut` takes the path down both ways until the `Outage` it returns drops, so a test can sever a link mid-run and -restore it when it chooses rather than on a timer. The relay's drills +restore it when it chooses rather than on a timer. `Shaper::retarget` forwards to a new target from then +on, keeping each client's flow, so a restarted server can come back on a fresh port behind the address +clients already dial. The relay's drills (`rs/moq-relay/tests/drills.rs`, described in `test/drill/README.md`) run every scenario through it. ## What a seed does and does not fix diff --git a/rs/moq-shaper/src/lib.rs b/rs/moq-shaper/src/lib.rs index 77490623c2..4f112af5c2 100644 --- a/rs/moq-shaper/src/lib.rs +++ b/rs/moq-shaper/src/lib.rs @@ -413,6 +413,8 @@ pub struct Shaper { failed: Arc>, /// How many [`Outage`]s are holding the path down. cuts: Arc, + /// Where datagrams go now, which [`Shaper::retarget`] moves. + target: Arc>, task: tokio::task::JoinHandle<()>, } @@ -432,13 +434,15 @@ impl Shaper { let tally = Arc::new([Tally::default(), Tally::default()]); let failed = Arc::new(OnceLock::new()); let cuts = Arc::new(AtomicU64::new(0)); + let target = Arc::new(Mutex::new(config.target)); let task = tokio::spawn({ let setup = setup.clone(); let tally = tally.clone(); let failed = failed.clone(); let cuts = cuts.clone(); + let target = target.clone(); async move { - if let Err(err) = run(Arc::new(listen), setup, tally, cuts).await { + if let Err(err) = run(Arc::new(listen), setup, tally, cuts, target).await { let _ = failed.set(format!("{err:#}")); } } @@ -450,6 +454,7 @@ impl Shaper { tally, failed, cuts, + target, task, }) } @@ -466,12 +471,30 @@ impl Shaper { } } + /// Forward to `target` from now on, keeping the address clients send to. + /// + /// The way a server comes back somewhere new behind a stable address: each + /// client keeps its flow, so the new target sees the same source addresses, + /// and datagrams already on their way still land at the old target, while its + /// late replies are dropped. The new + /// target must be reachable the way the first one was, loopback or not and + /// in the same address family, since the flows' sockets are already bound. + pub fn retarget(&self, target: SocketAddr) -> anyhow::Result<()> { + let mut current = self.target.lock().expect("target poisoned"); + anyhow::ensure!( + local_ip(target) == local_ip(*current), + "cannot retarget from {current} to {target}: the flows' sockets cannot reach it" + ); + *current = target; + Ok(()) + } + /// The address clients send to. pub fn addr(&self) -> SocketAddr { self.addr } - /// The configuration this shaper runs, including its seed. + /// The configuration this shaper started with, including its seed. pub fn config(&self) -> &Config { &self.setup.config } @@ -602,8 +625,13 @@ fn bump(counter: &AtomicU64) { /// A flow is a socket of its own toward the target, so the target sees one /// address per client just as it would without the shaper in the way. Each /// flow takes a link of its own each way, unless the path is shared. -async fn run(listen: Arc, setup: Setup, tally: Arc<[Tally; 2]>, cuts: Arc) -> anyhow::Result<()> { - let config = &setup.config; +async fn run( + listen: Arc, + setup: Setup, + tally: Arc<[Tally; 2]>, + cuts: Arc, + target: Arc>, +) -> anyhow::Result<()> { let mut flows = HashMap::::new(); let mut tasks = JoinSet::new(); let mut buf = vec![0u8; u16::MAX as usize]; @@ -629,12 +657,13 @@ async fn run(listen: Arc, setup: Setup, tally: Arc<[Tally; 2]>, cuts: continue; } let now = Instant::now(); + let to = *target.lock().expect("target poisoned"); let number = flows.len() as u64; let flow = match flows.entry(from) { hash_map::Entry::Occupied(entry) => entry.into_mut(), hash_map::Entry::Vacant(entry) => { - let upstream = Arc::new(bind_toward(config.target).await?); + let upstream = Arc::new(bind_toward(to).await?); let [up, down] = match &shared { Some(links) => links.clone(), @@ -642,7 +671,7 @@ async fn run(listen: Arc, setup: Setup, tally: Arc<[Tally; 2]>, cuts: }; tasks.spawn(reply( upstream.clone(), - config.target, + target.clone(), down, listen.clone(), from, @@ -655,7 +684,7 @@ async fn run(listen: Arc, setup: Setup, tally: Arc<[Tally; 2]>, cuts: }; let datagram = buf[..size].to_vec(); let mut up = flow.up.lock().expect("link poisoned"); - up.push(now, datagram, &flow.upstream, config.target, &tally[UP]); + up.push(now, datagram, &flow.upstream, to, &tally[UP]); } } @@ -692,24 +721,28 @@ fn link( ))) } -/// A socket that can reach `target`: loopback for a loopback target, so the -/// shaper never opens a port beyond the host when it does not need to. +/// A socket that can reach `target`. async fn bind_toward(target: SocketAddr) -> anyhow::Result { - let ip = match (target.ip().is_loopback(), target.is_ipv4()) { + UdpSocket::bind(SocketAddr::new(local_ip(target), 0)) + .await + .with_context(|| format!("bind a socket toward {target}")) +} + +/// The address a socket toward `target` binds: loopback for a loopback target, +/// so the shaper never opens a port beyond the host when it does not need to. +fn local_ip(target: SocketAddr) -> IpAddr { + match (target.ip().is_loopback(), target.is_ipv4()) { (true, true) => IpAddr::V4(Ipv4Addr::LOCALHOST), (true, false) => IpAddr::V6(Ipv6Addr::LOCALHOST), (false, true) => IpAddr::V4(Ipv4Addr::UNSPECIFIED), (false, false) => IpAddr::V6(Ipv6Addr::UNSPECIFIED), - }; - UdpSocket::bind(SocketAddr::new(ip, 0)) - .await - .with_context(|| format!("bind a socket toward {target}")) + } } /// Feed what the target sends a flow into the link back toward its client. async fn reply( socket: Arc, - target: SocketAddr, + target: Arc>, link: Arc>, listen: Arc, client: SocketAddr, @@ -720,7 +753,7 @@ async fn reply( loop { let (size, from) = socket.recv_from(&mut buf).await.context("receive from the target")?; // Anything else reaching this ephemeral port is not part of the path. - if from != target || is_cut(&cuts) { + if from != *target.lock().expect("target poisoned") || is_cut(&cuts) { continue; } let datagram = buf[..size].to_vec(); @@ -1077,6 +1110,42 @@ mod tests { assert_eq!(got, (0..10).collect::>(), "the restored path lost datagrams"); } + #[tokio::test(start_paused = true)] + async fn a_retarget_moves_the_flow_and_ignores_the_old_target() { + let first = UdpSocket::bind(LOCALHOST).await.unwrap(); + let second = UdpSocket::bind(LOCALHOST).await.unwrap(); + let shaper = Shaper::bind(Config { + bind: LOCALHOST, + target: first.local_addr().unwrap(), + seed: 1, + up: Profile::default(), + down: Profile::default(), + }) + .await + .unwrap(); + let client = UdpSocket::bind(LOCALHOST).await.unwrap(); + client.connect(shaper.addr()).await.unwrap(); + let mut buf = [0u8; 8]; + + client.send(b"before").await.unwrap(); + let (_, flow) = first.recv_from(&mut buf).await.unwrap(); + + shaper.retarget(second.local_addr().unwrap()).unwrap(); + client.send(b"after").await.unwrap(); + let (size, from) = second.recv_from(&mut buf).await.unwrap(); + assert_eq!(&buf[..size], b"after"); + assert_eq!(from, flow, "the new target saw a different flow"); + + // The old target answering late is no longer part of the path. + first.send_to(b"stale", flow).await.unwrap(); + second.send_to(b"fresh", flow).await.unwrap(); + let size = client.recv(&mut buf).await.unwrap(); + assert_eq!(&buf[..size], b"fresh"); + + let ipv6 = SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 1); + assert!(shaper.retarget(ipv6).is_err(), "retargeted where no flow can reach"); + } + #[tokio::test(start_paused = true)] async fn an_undelayed_datagram_waits_for_no_timer() { let (_shaper, client) = setup(1, Profile::default(), Profile::default()).await; diff --git a/test/drill/README.md b/test/drill/README.md index 59432d96dd..c76d94aab0 100644 --- a/test/drill/README.md +++ b/test/drill/README.md @@ -60,6 +60,13 @@ the clients discover the loss through the QUIC idle timeout. The drills set that to two seconds (and the keep-alive well inside it), which is what keeps a crash bounded rather than fast. +The replacement relay binds a fresh ephemeral port, never the dead one's: in +those two seconds any process's ephemeral bind can take the freed port. So this +drill's clients dial a shaper in both lanes, unimpaired on loopback, and +`Shaper::retarget` points it at the replacement. The replacement starts before +the kill, while the original still holds its port, so a restart that needed +that port back fails every run. + ## Two lanes Every drill runs twice, as `::loopback` and `::impaired`. The