3e1748aac8
joule-moq-room's impairment_changes_the_measured_loss again gave 119 of 120 objects after a minute's wait. Captured (Participants::test now answers `undelivered`: where each missing object stands at the viewer, and QUIC's counts of streams the relay stopped or reset): in every short run the missing objects had no stream at the viewer, and nothing was lost between the relay and the viewer. Two causes, both between the publisher and the relay. 1. Refused at the door (11 of about 60 runs under load; refused_at_door equal to the objects missing). The publisher sent a group's receipts and its media at one transport priority; a receipts stream whose packet was dropped was sent again in its turn among the streams waiting, after the door's two-second hold. A participant now sends receipts first, and a development relay with a QUIC endpoint holds an object at its door as long as a lossy path is given to deliver. 2. Stopped before its header (one seed of the impaired path: 103 of 240 streams stopped, 82 objects delivered to nobody). QUIC opens every stream below the highest one a packet has named, so a stream whose first packet was lost stood open and empty at the relay, which gave it ten seconds for its header and stopped it. The relay now waits for a data stream's header as long as the session stands: such a stream holds no bytes and one of max_uni_streams. - joule-moq-relay tests/late_stream_header.rs fails with the timer. - The viewer reads its streams beside one another, each from its header. - Left: a request stream's ten seconds, a session's ten seconds for SETUP, and the two-second door `wai dev --impair` writes. Changelog: none Co-Authored-By: Claude Fable 5.2 <[email protected]>
66 lines
3.8 KiB
Rust
66 lines
3.8 KiB
Rust
//! A data stream whose header comes late is still read. QUIC opens every stream below the
|
|
//! highest one a packet has named (RFC 9000 §3.2): a stream whose first packet was lost is
|
|
//! open at the relay, with nothing in it, from the moment a later stream's packet arrives
|
|
//! until QUIC sends its own again. The relay gave such a stream ten seconds for its header
|
|
//! and then stopped it. On a path that drops packets a publisher's retransmission took
|
|
//! longer: its object reached the relay's host after the stream was stopped, the publisher
|
|
//! was told nothing it reads (its write had long returned), and no subscriber was sent the
|
|
//! object (seen through `joule-moq-room`'s lossy participants: 103 of a publisher's 240
|
|
//! streams stopped in one run, 82 of 120 objects delivered to nobody).
|
|
//!
|
|
//! Here the delay is the publisher's own: it opens two streams, writes the second whole, and
|
|
//! writes the first twelve seconds later.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use joule_moq::draft22::data::{SubgroupCodec, SubgroupHeader, SubgroupId, SubgroupObject};
|
|
use joule_moq::draft22::Parameters;
|
|
use joule_moq_relay::client::ClientOptions;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
/// Longer than the ten seconds the relay waited.
|
|
const LATE: Duration = Duration::from_secs(12);
|
|
|
|
fn stream_of(alias: u64, g: u64) -> Vec<u8> {
|
|
let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(64), properties: false, end_of_group: true, first_object: true };
|
|
let mut b = Vec::new();
|
|
h.encode(&mut b);
|
|
SubgroupCodec::new(h).encode_object(&SubgroupObject::normal(0, format!("group {g}").into_bytes()), &mut b).expect("an object");
|
|
b
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_stream_whose_header_comes_late_is_read() {
|
|
timeout(Duration::from_secs(120), async {
|
|
let server = relay(|_| {}).await;
|
|
let (space, track) = (ns(&["late"]), name("header"));
|
|
let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let _held = p.publish(space.clone(), track.clone(), 4, Parameters::new(), Default::default()).await.expect("PUBLISH");
|
|
let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
|
|
let (ok, _stx, _srx) = s.subscribe(space.clone(), track.clone(), Parameters::new()).await.expect("SUBSCRIBE");
|
|
// Two streams, in the order of their IDs. Nothing of the first is sent yet; the
|
|
// second's bytes open both at the relay.
|
|
let mut first = p.conn.open_uni().await.expect("a stream");
|
|
let mut second = p.conn.open_uni().await.expect("a stream");
|
|
second.write_all(&stream_of(4, 1)).await.expect("written");
|
|
second.finish().expect("finished");
|
|
let got = collect(&s, STEP, |g| !g.is_empty()).await;
|
|
assert_eq!(got.iter().map(|g| (g.header.track_alias, g.header.group_id)).collect::<Vec<_>>(), [(ok.track_alias, 1)], "the second stream's object");
|
|
tokio::time::sleep(LATE).await;
|
|
// The first stream's bytes, late: the relay has not stopped it, and its object arrives.
|
|
first.write_all(&stream_of(4, 0)).await.expect("the relay has not stopped the stream");
|
|
first.finish().expect("finished");
|
|
let got = collect(&s, STEP, |g| !g.is_empty()).await;
|
|
assert_eq!(got.len(), 1, "the late stream's object reaches the subscriber");
|
|
assert_eq!((got[0].header.group_id, got[0].reset, got[0].objects[0].payload.as_slice()), (0, None, b"group 0".as_slice()));
|
|
assert!(matches!(timeout(Duration::from_millis(200), first.delivered()).await, Ok(Ok(None)) | Err(_)), "the relay did not stop the publisher's stream");
|
|
for c in [&p, &s] {
|
|
c.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|