45822e6a37
RelayServer::shutdown aborted the WebSocket and QUIC accept tasks without awaiting them, left each session's task (spawned per connection, holding the connection, and through it the endpoint's socket) running, and the control surface's listener likewise; quinn closes an endpoint's UDP socket only when the endpoint's driver ends, after the last handle to the endpoint and its connections is gone. So it returned with its ports bound, and a relay started again at once on them failed "binding ...: Address already in use" (waitransport's World::next_run saw it once, after two seconds of retries). It also returned before every session's summary was signed: a summary is pushed as the session's task ends, which can follow its close. Now every session's task is held (joule_moq_relay::tasks::Tasks, a JoinSet the server waits for, ending what is left past five seconds), and the shutdown closes the control surface (Control::close, which awaits its tasks), awaits the WebSocket accept task, sends GOAWAY and closes the endpoints as before, awaits the QUIC accept tasks and the sessions' tasks, drops the endpoints, and waits (up to five seconds) until each endpoint's socket, held from the start by a weak reference, is gone. The gateway in joule-moq-ingest uses Tasks too. tests/shutdown_ports.rs starts the relay 20 times on the same ports from a task on a runtime of one worker, a session open over QUIC, WebTransport and WebSocket each time; when the shutdown returns its UDP, WebSocket and control ports bind at once, no session is open and all three summaries are in. Before this change it failed at round 0, the UDP port still bound (os error 48 on macOS, the M4 at load 15.6 on 16 cores; os error 98 on Linux, the Framework at load 32 on 32). The unit test of Tasks holds that a close returns once each task has ended and its state is dropped. Co-Authored-By: Claude Opus 5.5 <[email protected]>
61 lines
3 KiB
Rust
61 lines
3 KiB
Rust
//! A shutdown returns with every port the relay listened on closed, so a relay can start again
|
|
//! at once on the same ports, in the same process: its UDP endpoint (every socket of its group,
|
|
//! where it has one), its WebSocket listener and its control listener. By then every session
|
|
//! has ended and signed its summary. The test runs in a task on a runtime of one worker, as a
|
|
//! restart from a spawned task does: there a task that a shutdown only told to go
|
|
//! (`JoinHandle::abort`), or a QUIC endpoint's driver that has yet to see its last handle go,
|
|
//! cannot have run to its end, socket and all, until the shutdown's caller yields. The relay is
|
|
//! shut down and started again on its ports many times, with sessions open over QUIC,
|
|
//! WebTransport and WebSocket.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use joule_moq_relay::client::ClientOptions;
|
|
use joule_moq_relay::{RelayConfig, RelayServer};
|
|
use std::net::SocketAddr;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
const ROUNDS: usize = 20;
|
|
|
|
/// Bind each address at once, as a relay starting again on it does first.
|
|
fn free(tcp: &[SocketAddr], udp: &[SocketAddr], round: usize) {
|
|
for &a in tcp {
|
|
if let Err(e) = std::net::TcpListener::bind(a) {
|
|
panic!("round {round}: TCP {a} is still bound when the shutdown returns: {e}");
|
|
}
|
|
}
|
|
for &a in udp {
|
|
if let Err(e) = std::net::UdpSocket::bind(a) {
|
|
panic!("round {round}: UDP {a} is still bound when the shutdown returns: {e}");
|
|
}
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
|
|
async fn a_shutdown_from_a_task_frees_every_port_for_a_start_at_once() {
|
|
let t = tokio::spawn(async {
|
|
let mut cfg = RelayConfig::loopback();
|
|
cfg.control.listen = Some(loopback());
|
|
cfg.limits.goaway_timeout_ms = 200;
|
|
for round in 0..ROUNDS {
|
|
let server = RelayServer::start(cfg.clone()).await.unwrap_or_else(|e| panic!("round {round}: the relay starts again on the same ports: {e}"));
|
|
let (quic, ws, control) = (server.quic_addr.unwrap(), server.websocket_addr.unwrap(), server.control_addr.unwrap());
|
|
(cfg.quic_listen, cfg.websocket_listen, cfg.control.listen) = (Some(quic), Some(ws), Some(control));
|
|
// A session over each transport, open as the shutdown begins.
|
|
let mut clients = Vec::new();
|
|
for via in [Via::Quic, Via::WebTransport, Via::WebSocket] {
|
|
clients.push(connect(&server, via, ClientOptions::default()).await);
|
|
}
|
|
let relay = server.relay().clone();
|
|
assert!(until(STEP, || relay.stats().sessions_open == 3).await, "round {round}: three sessions open");
|
|
server.shutdown().await;
|
|
free(&[ws, control], &[quic], round);
|
|
assert_eq!(relay.stats().sessions_open, 0, "round {round}: every session has ended");
|
|
assert_eq!(relay.summaries().len(), 3, "round {round}: every session's summary is in");
|
|
drop(clients);
|
|
}
|
|
});
|
|
timeout(Duration::from_secs(300), t).await.expect("in time").unwrap();
|
|
}
|