From 8b93529b2f5fd4358a6066c119d2256a8c28d740 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 17:22:22 -0700 Subject: [PATCH 1/5] quest: claim relay-restart-rebind From 1932b2ca995a761055849cd4a28e86d4007a4496 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 18:05:08 -0700 Subject: [PATCH 2/5] test(relay): restart the crash drill on a fresh port behind a stable address relay_killed_mid_group_aborts_then_resumes restarted the relay on its old ephemeral UDP port, which sat free for the ~2s idle-timeout window between the kill and the restart. Any process's bind(0) could take it there, failing the restart with "Address already in use". Instrumented runs showed the old relay never kept its socket past the kill; the port was always held by a competing ephemeral bind. Clients now dial a moq-shaper in both lanes (unimpaired on loopback), and the replacement relay binds a fresh port that `Shaper::retarget` points the shaper at. The drill holds the dead port itself, so a restart that needs it back fails every run. Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/README.md | 1 - .../m1/test-flakes-2/relay-restart-rebind.md | 49 --------- rs/moq-relay/tests/drills.rs | 58 +++++++--- rs/moq-shaper/README.md | 4 +- rs/moq-shaper/src/lib.rs | 100 +++++++++++++++--- test/drill/README.md | 6 ++ 6 files changed, 136 insertions(+), 82 deletions(-) delete mode 100644 quest/m1/test-flakes-2/relay-restart-rebind.md diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index b0fc70f158..26f1aa64f3 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -34,6 +34,5 @@ Public API: none. Wire: none. - [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - a late joiner shows video promptly and catches up to live, and the check asserts that under load - [Import catalog finish](/quest/m1/test-flakes-2/import-catalog-finish.md) - `moq-cli`'s subprocess EOF catalog-finish test holds up under load with event-based fixture coordination -- [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 cluster drills' clients never time out while connecting - [Media audio tone](/quest/m1/media-audio-tone.md) - the `just test media` audio-tone check passes under load, fixed at its cause 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..7ac7d86d16 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(); @@ -574,6 +597,11 @@ async fn relay_killed_mid_group_aborts_then_resumes(lane: Lane) { relay.kill().await; + // Take the dead relay's port the way another process's ephemeral bind can, + // so the restart below never depends on getting it back. A failed bind means + // someone else already holds it, which tests the same thing. + let _taken = std::net::UdpSocket::bind(relay.addr()).ok(); + // Terminal result: the unread half of the open group fails. `Ok(None)` here // would be the relay's death passing for the publisher finishing. let err = tokio::time::timeout(TIMEOUT, open_reader.read_frame()) @@ -586,8 +614,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 = RelayHost::start(relay_config()).await; + 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 +668,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 +955,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 +970,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 +1262,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..08e02efa71 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,29 @@ 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. 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 +624,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 +656,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 +670,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 +683,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 +720,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 +752,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 +1109,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..7229029108 100644 --- a/test/drill/README.md +++ b/test/drill/README.md @@ -60,6 +60,12 @@ 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 drill holds the dead port +itself so a restart that needed it back fails every run. + ## Two lanes Every drill runs twice, as `::loopback` and `::impaired`. The From 5da6c965af7a615636b57186482cf0f0e06d2e69 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 18:55:38 -0700 Subject: [PATCH 3/5] test(relay): fail the crash drill if the restart reclaims the dead port The port reservation can lose to another process that later releases the port, so also assert the replacement relay binds a different address. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/tests/drills.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index 7ac7d86d16..6871dd56b8 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -600,7 +600,8 @@ async fn relay_killed_mid_group_aborts_then_resumes(lane: Lane) { // Take the dead relay's port the way another process's ephemeral bind can, // so the restart below never depends on getting it back. A failed bind means // someone else already holds it, which tests the same thing. - let _taken = std::net::UdpSocket::bind(relay.addr()).ok(); + let old = relay.addr(); + let _taken = std::net::UdpSocket::bind(old).ok(); // Terminal result: the unread half of the open group fails. `Ok(None)` here // would be the relay's death passing for the publisher finishing. @@ -615,6 +616,9 @@ async fn relay_killed_mid_group_aborts_then_resumes(lane: Lane) { expect_status(&mut publish_loop, moq_tokio::Status::Disconnected, "publisher").await; let relay = RelayHost::start(relay_config()).await; + // Whoever held the old port may have released it by now, so check the + // address too: a restart that reclaimed it must not pass. + assert_ne!(relay.addr(), old, "the restarted relay reused the dead relay's port"); path.retarget(&relay); expect_status(&mut subscribe_loop, moq_tokio::Status::Connected, "subscriber").await; From c922f8c569936038edffa15a19b1b60ecd74c9b7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 19:42:44 -0700 Subject: [PATCH 4/5] test(relay): start the crash drill's replacement before the kill The replacement relay now binds while the original still holds its port, so the two can never share one. This replaces the port reservation and the address assertion, which could reject a legitimate ephemeral reuse. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/tests/drills.rs | 14 ++++---------- test/drill/README.md | 5 +++-- 2 files changed, 7 insertions(+), 12 deletions(-) diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index 6871dd56b8..a03e1e0e8b 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -595,14 +595,11 @@ 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; - // Take the dead relay's port the way another process's ephemeral bind can, - // so the restart below never depends on getting it back. A failed bind means - // someone else already holds it, which tests the same thing. - let old = relay.addr(); - let _taken = std::net::UdpSocket::bind(old).ok(); - // Terminal result: the unread half of the open group fails. `Ok(None)` here // would be the relay's death passing for the publisher finishing. let err = tokio::time::timeout(TIMEOUT, open_reader.read_frame()) @@ -615,10 +612,7 @@ 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()).await; - // Whoever held the old port may have released it by now, so check the - // address too: a restart that reclaimed it must not pass. - assert_ne!(relay.addr(), old, "the restarted relay reused the dead relay's port"); + let relay = replacement; path.retarget(&relay); expect_status(&mut subscribe_loop, moq_tokio::Status::Connected, "subscriber").await; diff --git a/test/drill/README.md b/test/drill/README.md index 7229029108..c76d94aab0 100644 --- a/test/drill/README.md +++ b/test/drill/README.md @@ -63,8 +63,9 @@ 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 drill holds the dead port -itself so a restart that needed it back fails every run. +`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 From be7b56db5db62e7578d8d135842808e41e1667e7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 05:29:18 -0700 Subject: [PATCH 5/5] docs(moq-shaper): note retarget drops the old target's late replies Co-Authored-By: Claude Opus 5.5 --- rs/moq-shaper/src/lib.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/rs/moq-shaper/src/lib.rs b/rs/moq-shaper/src/lib.rs index 08e02efa71..4f112af5c2 100644 --- a/rs/moq-shaper/src/lib.rs +++ b/rs/moq-shaper/src/lib.rs @@ -475,7 +475,8 @@ impl Shaper { /// /// 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. The new + /// 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<()> {