811164ff32
The interop run of 11 November (D30) saw a PUBLISH_NAMESPACE withdrawn by STOP_SENDING after a FIN go unnoticed. -22 §6.4.2.3: "An endpoint that has already sent a FIN on its sending direction and subsequently wishes to cancel sends STOP_SENDING on the receiving direction"; a FIN alone is no withdrawal (§4.1). Once the publisher's FIN came the relay waited for the session's end alone. ReqTx::stopped waits for the peer's STOP_SENDING on this end's direction (asking the stream a quarter second at a time, so no send on it waits longer); publish_namespace, after the publisher's FIN, withdraws the namespace when it comes, and its subscribers are told NAMESPACE_DONE. tests/namespace_withdrawn.rs, over native QUIC and the WebSocket binding: the namespace stands after the publisher's FIN, and its STOP_SENDING then withdraws it. Falsified: with the direction not watched, no NAMESPACE_DONE comes and the test fails. Co-Authored-By: Claude Opus 5.5 <[email protected]>
44 lines
2.4 KiB
Rust
44 lines
2.4 KiB
Rust
//! A PUBLISH_NAMESPACE withdrawn after its FIN (`draft-ietf-moq-transport-22` §6.4.2.3: "An
|
|
//! endpoint that has already sent a FIN on its sending direction and subsequently wishes to
|
|
//! cancel sends STOP_SENDING on the receiving direction"). Over native QUIC and the WebSocket
|
|
//! binding, a publisher's namespace stands after the publisher's FIN, as a FIN is no
|
|
//! withdrawal (§4.1); its STOP_SENDING then withdraws it, and a namespace subscriber is told
|
|
//! NAMESPACE_DONE. The relay used to wait for the session's end alone once the FIN came.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use joule_moq::draft22::{ControlMessage, Parameters};
|
|
use joule_moq_relay::client::ClientOptions;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_namespace_is_withdrawn_by_stop_sending_after_its_fin() {
|
|
timeout(Duration::from_secs(90), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn run() {
|
|
let server = relay(|_| {}).await;
|
|
for (i, via) in [Via::Quic, Via::WebSocket].into_iter().enumerate() {
|
|
let top = format!("withdrawn{i}");
|
|
let prefix = ns(&[top.as_str(), "x"]);
|
|
let (p, _pe) = connect(&server, via, ClientOptions::default()).await;
|
|
let (s, _se) = connect(&server, via, ClientOptions::default()).await;
|
|
let (ptx, mut prx) = p.publish_namespace(prefix.clone(), Parameters::new()).await.expect("PUBLISH_NAMESPACE");
|
|
let (_stx, mut srx) = s.subscribe_namespace(ns(&[top.as_str()])).await.expect("SUBSCRIBE_NAMESPACE");
|
|
let seen = timeout(STEP, srx.next()).await.expect("NAMESPACE in time").expect("it reads");
|
|
assert!(matches!(seen, Some(ControlMessage::Namespace(_))), "{via:?}: {seen:?}");
|
|
// The publisher's FIN: no withdrawal.
|
|
ptx.fin().await;
|
|
let quiet = timeout(Duration::from_millis(500), srx.next()).await;
|
|
assert!(quiet.is_err(), "{via:?}: a FIN withdrew the namespace: {quiet:?}");
|
|
// Its STOP_SENDING on the direction it receives: a withdrawal.
|
|
prx.reader.stop(0x1);
|
|
let done = timeout(STEP, srx.next()).await.unwrap_or_else(|_| panic!("{via:?}: no NAMESPACE_DONE after the publisher's STOP_SENDING")).expect("it reads");
|
|
assert!(matches!(done, Some(ControlMessage::NamespaceDone(_))), "{via:?}: {done:?}");
|
|
p.close(0, "done");
|
|
s.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
}
|