08f1b4ce3c
Found by the interop run against pinned peers. A PUBLISH with FORWARD 0 was answered REQUEST_OK and never resumed (-22 3.1.1, 7.4, 7.6). The client dropped an unreadable request and left its stream unanswered (-22 9.20). Co-Authored-By: Claude Opus 5.5 <[email protected]>
36 lines
2.3 KiB
Rust
36 lines
2.3 KiB
Rust
//! A request this crate's client cannot read is not dropped in silence.
|
|
//!
|
|
//! `draft-ietf-moq-transport-22` §9.20: "An endpoint that receives an unknown Message
|
|
//! Parameter MUST close the session with PROTOCOL_VIOLATION". Found by a run against another
|
|
//! implementation's relay, whose SUBSCRIBE to this client's publisher carried a parameter
|
|
//! this codec did not read: the client dropped the request stream, and the relay could only
|
|
//! report a request that ended with no response.
|
|
|
|
use joule_moq::draft22::{ControlMessage, Parameters, SessionError, TrackRequest};
|
|
use joule_moq::{TrackName, TrackNamespace};
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::transport::mux::{memory_pair, MuxConfig};
|
|
use joule_moq_relay::transport::{SharedConnection, TransportError};
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
#[tokio::test]
|
|
async fn an_unknown_message_parameter_in_a_request_closes_the_session() {
|
|
let (a, b, _) = memory_pair(MuxConfig::default(), false);
|
|
let (a, b): (SharedConnection, SharedConnection) = (Arc::new(a), Arc::new(b));
|
|
let (ca, cb) = tokio::join!(Client::connect(a.clone(), ClientOptions::default()), Client::connect(b.clone(), ClientOptions::default()));
|
|
let (_ca, _cb) = (ca.expect("one end's SETUP"), cb.expect("the other's"));
|
|
// A SUBSCRIBE with no parameter, and then one of a type the draft does not define put
|
|
// where its empty list was: the count 1, the type 0x3e (even, so a varint follows), 0.
|
|
let m = ControlMessage::Subscribe(TrackRequest { request_id: 1, namespace: TrackNamespace::from_strs(["a"]), name: TrackName::from_str("t"), params: Parameters::new() });
|
|
let mut bytes = m.encode().expect("a SUBSCRIBE");
|
|
assert_eq!(bytes.pop(), Some(0), "the empty parameter list is the message's last byte");
|
|
bytes.extend_from_slice(&[1, 0x3e, 0]);
|
|
let len = u16::from_be_bytes([bytes[1], bytes[2]]) + 2;
|
|
bytes[1..3].copy_from_slice(&len.to_be_bytes());
|
|
let (mut tx, _rx) = b.open_bi().await.expect("a request stream");
|
|
tx.write_all(&bytes).await.expect("written");
|
|
let why = timeout(Duration::from_secs(5), b.closed()).await.expect("the session is closed, not left open");
|
|
assert!(matches!(why, TransportError::Closed { code, by_peer: true, .. } if code == SessionError::PROTOCOL_VIOLATION.0), "{why:?}");
|
|
}
|