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]>
41 lines
1.8 KiB
Rust
41 lines
1.8 KiB
Rust
//! A PUBLISH that begins paused is resumed by the relay.
|
|
//!
|
|
//! `draft-ietf-moq-transport-22` §3.1.1: "The initiator of the subscription sets the initial
|
|
//! state by including the FORWARD parameter in PUBLISH or SUBSCRIBE. The subscriber can ...
|
|
//! resume it by sending REQUEST_UPDATE with FORWARD set to 1"; §7.6 has a relay resume a
|
|
//! paused PUBLISH once a downstream subscription is not paused. The answer to a PUBLISH
|
|
//! carries no forward state at -22 or -23, so a relay that answers REQUEST_OK and no more
|
|
//! leaves such a publisher silent. Found by another implementation's test client, whose
|
|
//! publisher sends FORWARD 0 and waited ten seconds for nothing.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use joule_moq::draft::Draft;
|
|
use joule_moq::draft22::kvp::Properties;
|
|
use joule_moq::draft22::{ControlMessage, Parameter, Parameters};
|
|
use joule_moq_relay::client::ClientOptions;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
async fn resumed_at(draft: Draft) {
|
|
let server = relay(|_| {}).await;
|
|
let (publ, _ep) = connect(&server, Via::QuicDraft(draft), ClientOptions::default()).await;
|
|
let paused = Parameters::new().with(Parameter::Forward(false));
|
|
let (_tx, mut rx) = publ.publish(ns(&["paused"]), name("t"), 7, paused, Properties::new()).await.expect("PUBLISH is accepted");
|
|
let next = timeout(Duration::from_secs(5), rx.next()).await.expect("the relay says something after its REQUEST_OK").expect("the request stream reads");
|
|
match next {
|
|
Some(ControlMessage::RequestUpdate(u)) => assert_eq!(u.params.forward(), Some(true), "{draft}: {u:?}"),
|
|
other => panic!("{draft}: not a REQUEST_UPDATE: {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_publish_that_begins_paused_is_resumed_at_22() {
|
|
resumed_at(Draft::D22).await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_publish_that_begins_paused_is_resumed_at_23() {
|
|
resumed_at(Draft::D23).await;
|
|
}
|