dfb24a1a2b
Some checks failed
joule-code conformance gate / Build + round-trip C1-C4 vectors (pull_request) Has been cancelled
joule-code conformance gate / joule-moq-relay, every target (pull_request) Has been cancelled
joule-code conformance gate / joule-moq-ingest, every target (pull_request) Has been cancelled
joule-moq-room room service / joule-moq-room · does this change touch it (pull_request) Has been cancelled
joule-moq-room room service / joule-moq-room, every target (pull_request) Has been cancelled
joule-moq-web browser client / joule-moq-web · does this change touch it (pull_request) Has been cancelled
joule-moq-web browser client / joule-moq-web · browser end-to-end (macos) (pull_request) Has been cancelled
repository checks / lockfiles, tests in CI, workflow hygiene, spec and section references, public text (pull_request) Has been cancelled
standards workspaces build + test / which workspaces this change touches (pull_request) Has been cancelled
standards workspaces build + test / macOS energy backends (pull_request) Has been cancelled
- transport::mux: a stream that arrived after the session closed (its peer not yet told) found nobody accepting, and its receiving half was dropped after the state was locked again: its Drop takes that lock, so the worker deadlocked, and every other worker after it. Any session the relay closes while its peer still opens streams (the session and ingress limits, a refused proof) could hang the relay. The half is now dropped before the lock is taken again; the regression test hangs without the fix and passes with it. - A FETCH the cache does not yet cover, of a track whose publisher is still sending, waits (within its fill budget, at most a second) for the groups still arriving before the relay asks upstream: a page's audio is a group per 20 ms, each on its own stream, and a late joiner's fetch from group 0 through an HTTP front met one in flight and was passed to the publisher, which takes no requests (seen on the deployed room). - room_authority: the ingress limit read from the relay's record when the client loses the race to see the close. joule-moq-web: a late joiner's refused FETCH is tried again (a refusal comes before any object). Co-Authored-By: Claude Opus 5.5 <[email protected]>
64 lines
3.2 KiB
Rust
64 lines
3.2 KiB
Rust
//! A FETCH that arrives while an earlier group's stream is still arriving: the cache does
|
|
//! not yet cover the range (the group is not whole until its FIN), and the publisher, which
|
|
//! published by PUBLISH, takes no FETCH. The relay waits a moment for the group to finish
|
|
//! (within the fetch's fill budget) and serves the whole range from its cache, rather than
|
|
//! asking the publisher.
|
|
//!
|
|
//! A track of one group per object, each on its own stream, as a page's audio is (one
|
|
//! group every 20 ms), always has a stream or two in flight; through an HTTP front the
|
|
//! window is wider, and a late joiner's fetch from group 0 met it (seen on the deployed
|
|
//! room, 2026-10-19).
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use joule_moq::draft22::data::{FetchEntry, SubgroupCodec, SubgroupHeader, SubgroupId, SubgroupObject};
|
|
use joule_moq::draft22::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_fetch_waits_for_a_group_still_arriving() {
|
|
timeout(Duration::from_secs(60), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
fn header(alias: u64, g: u64) -> SubgroupHeader {
|
|
SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(64), properties: false, end_of_group: true, first_object: true }
|
|
}
|
|
|
|
async fn run() {
|
|
let server = relay(|_| {}).await;
|
|
let t = Tracks::new(ns(&["room", "p", "audio"]), "0", 10, 5, 64);
|
|
let (publisher, _ep) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _held = publish_tracks(&publisher, &t, None).await;
|
|
for g in 0..4 {
|
|
publisher.send_subgroup(header(t.media_alias, g), &[SubgroupObject::normal(0, payload("a", g, 200))]).await.unwrap();
|
|
}
|
|
// Group 4's stream opens and carries its object, and its FIN is late; group 5 is whole.
|
|
let mut b = Vec::new();
|
|
let h4 = header(t.media_alias, 4);
|
|
h4.encode(&mut b);
|
|
SubgroupCodec::new(h4).encode_object(&SubgroupObject::normal(0, payload("a", 4, 200)), &mut b).unwrap();
|
|
let mut open = publisher.conn.open_uni().await.unwrap();
|
|
open.write_all(&b).await.unwrap();
|
|
publisher.send_subgroup(header(t.media_alias, 5), &[SubgroupObject::normal(0, payload("a", 5, 200))]).await.unwrap();
|
|
tokio::time::sleep(Duration::from_millis(200)).await;
|
|
|
|
let (late, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let fin = tokio::spawn(async move {
|
|
tokio::time::sleep(Duration::from_millis(300)).await;
|
|
open.finish().unwrap();
|
|
open
|
|
});
|
|
let (ok, entries) = late.fetch(t.ns.clone(), t.media.clone(), Parameters::new()).await.expect("the fetch is served");
|
|
let _open = fin.await.unwrap();
|
|
let groups: Vec<u64> = entries.iter().filter_map(|e| if let FetchEntry::Object(o) = e { Some(o.group_id) } else { None }).collect();
|
|
assert_eq!(groups, (0..6).collect::<Vec<_>>(), "every group, from the cache");
|
|
assert_eq!(ok.end_location.group, 5);
|
|
let s = server.relay().stats();
|
|
assert_eq!((s.fetch_served_from_cache, s.fetch_served_upstream), (1, 0), "{s:?}");
|
|
late.close(0, "done");
|
|
publisher.close(0, "done");
|
|
server.shutdown().await;
|
|
}
|