882b08a25b
- goaway.rs: every session gets GOAWAY with the New Session URI and Timeout; a new request after it is refused with GOING_AWAY; the sessions that leave close with NO_ERROR, and the one that stays is closed with GOAWAY_TIMEOUT. - auth_tokens.rs: a token registered in SETUP authorizes a publisher; one registered in a request is used by alias in the next; no token or a wrong one is UNAUTHORIZED; an alias never registered closes the session with UNKNOWN_AUTH_TOKEN_ALIAS. - priority_fan_out.rs: two publishers, four tracks, three subscribers with their own subscriber priorities and group orders, the subscribers' pipes held shut while the load arrives; each receives every object in section 5.1.2's order. Co-Authored-By: Claude Opus 5.5 <[email protected]>
77 lines
3.4 KiB
Rust
77 lines
3.4 KiB
Rust
//! GOAWAY on shutdown (`draft-ietf-moq-transport-22` §6.6.1, §9.2): every session is told,
|
|
//! with the New Session URI and the Timeout; new requests are refused with GOING_AWAY; a
|
|
//! client that leaves is let go, and one that stays is closed with GOAWAY_TIMEOUT.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use joule_moq::draft22::{Parameters, RequestErrorCode, SessionError};
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::transport::TransportError;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
const URI: &str = "moqt://127.0.0.1:4999/moq";
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn goaway_on_shutdown() {
|
|
timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn goaway_of(c: &Client) -> joule_moq::draft22::Goaway {
|
|
let mut w = c.goaway();
|
|
timeout(STEP, async {
|
|
loop {
|
|
if let Some(g) = w.borrow_and_update().clone() {
|
|
return g;
|
|
}
|
|
w.changed().await.expect("the session lives until its GOAWAY");
|
|
}
|
|
})
|
|
.await
|
|
.expect("a GOAWAY in time")
|
|
}
|
|
|
|
async fn run() {
|
|
let server = relay(|c| {
|
|
c.limits.goaway_timeout_ms = 1500;
|
|
c.goaway_uri = URI.into();
|
|
})
|
|
.await;
|
|
let t = Tracks::new(ns(&["live", "g"]), "cam", 10, 4, 128);
|
|
let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let _pub = publish_tracks(&p, &t, None).await;
|
|
let (a, _ea) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let (b, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _sa = a.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.unwrap();
|
|
let _sb = b.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.unwrap();
|
|
|
|
let relay = server.relay().clone();
|
|
let shutdown = tokio::spawn(async move { server.shutdown().await });
|
|
|
|
for c in [&p, &a, &b] {
|
|
let g = goaway_of(c).await;
|
|
assert_eq!(g.new_session_uri, URI.as_bytes(), "the New Session URI");
|
|
assert_eq!(g.timeout_ms, 1500, "the Timeout the relay will wait");
|
|
}
|
|
// A new request after GOAWAY is refused (§9.2: "MAY reject new requests").
|
|
let refused = b.subscribe(t.ns.clone(), t.receipts.clone(), Parameters::new()).await.expect_err("a new SUBSCRIBE after GOAWAY");
|
|
assert!(refused.contains(&RequestErrorCode::GOING_AWAY.to_string()), "{refused}");
|
|
|
|
// The publisher and one subscriber migrate and close with NO_ERROR; the other stays.
|
|
a.close(SessionError::NO_ERROR.0, "migrated");
|
|
p.close(SessionError::NO_ERROR.0, "migrated");
|
|
match timeout(STEP, b.conn.closed()).await.expect("closed in time") {
|
|
TransportError::Closed { code, by_peer, .. } => {
|
|
assert_eq!(code, SessionError::GOAWAY_TIMEOUT.0, "the relay closes a session that stayed with GOAWAY_TIMEOUT");
|
|
assert!(by_peer);
|
|
}
|
|
other => panic!("{other:?}"),
|
|
}
|
|
let closed = timeout(STEP, shutdown).await.expect("the shutdown ends").unwrap();
|
|
assert_eq!(closed, 1, "one session outlived the GOAWAY timeout");
|
|
assert!(until(STEP, || relay.summaries().len() == 3).await);
|
|
let codes: Vec<Option<u64>> = relay.summaries().iter().map(|s| s.summary.close_code).collect();
|
|
assert_eq!(codes.iter().filter(|c| **c == Some(SessionError::GOAWAY_TIMEOUT.0)).count(), 1, "{codes:?}");
|
|
assert_eq!(codes.iter().filter(|c| **c == Some(SessionError::NO_ERROR.0)).count(), 2, "{codes:?}");
|
|
}
|