Transaction-Science/open-standards
Watch
1
0
Fork
You've already forked open-standards
 0
Code Issues Pull requests Projects Releases 15 Packages Wiki Activity Actions

joule-moq-relay: the loop-of-relays test starts again on new ports when one it chose was taken #652

Merged
dcharlot merged 1 commit from test/relay-loop-port-race into main 2026-11-05 15:12:22 -05:00
Conversation 0 Commits 1 Files changed 1 +27 -12
1 changed files with 27 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

39
joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs
View file

@ -439,20 +439,35 @@ async fn a_loop_of_relays() {
async fn relay_loop() {
// A dials B, B dials C, C dials A, each routing everything under live: a loop.
let ports = [free_port(), free_port(), free_port()];
let ws = |i: usize| format!("ws://127.0.0.1:{}/moq", ports[i]);
let mk = |i: usize| {
let next = (i + 1) % 3;
let url = ws(next);
move |c: &mut RelayConfig| {
quick(c);
c.websocket_listen = Some(format!("127.0.0.1:{}", ports[i]).parse().unwrap());
c.upstream = vec![upstream(url.clone(), UpstreamTransport::WebSocket, &[&["live"]])];
// Each relay dials the next, so the three ports are chosen before any of them listens, and
// another test can bind one in between: a start refused for a port in use is tried again
// on three new ports, at most five times. Anything else a start refuses fails the test.
let (ra, rb, rc) = {
let mut attempt = 0;
'start: loop {
attempt += 1;
let ports = [free_port(), free_port(), free_port()];
let mut started = Vec::new();
for i in 0..3 {
let mut c = RelayConfig::loopback();
quick(&mut c);
c.websocket_listen = Some(format!("127.0.0.1:{}", ports[i]).parse().unwrap());
c.upstream = vec![upstream(format!("ws://127.0.0.1:{}/moq", ports[(i + 1) % 3]), UpstreamTransport::WebSocket, &[&["live"]])];
match RelayServer::start(c).await {
Ok(r) => started.push(r),
Err(e) if attempt < 5 && e.contains("Address already in use") => {
for r in started {
r.shutdown().await;
}
continue 'start;
}
Err(e) => panic!("the relay starts: {e}"),
}
}
let mut it = started.into_iter();
break (it.next().unwrap(), it.next().unwrap(), it.next().unwrap());
}
};
let ra = relay(mk(0)).await;
let rb = relay(mk(1)).await;
let rc = relay(mk(2)).await;
for r in [&ra, &rb, &rc] {
linked(r, 1).await;
}