From c03f2ba4386d7574bdc722e4dafd11088e65c071 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 00:15:05 -0700 Subject: [PATCH 01/11] quest: claim session-death-error Co-Authored-By: Claude Opus 5.5 From 3f3c281e6bbf04ee63d21f76e210aaaf338f0557 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jakub=20Migda=C5=82?= Date: Thu, 24 Sep 2026 17:58:51 +0200 Subject: [PATCH 02/11] test: repro a subscription ending clean with its final group missing A subscription cut off by its session closing concludes `Ok(None)`, as if the track had ended after the last group the subscriber pulled, when the draft says a subscription ends by the publisher's FIN only once every group is accounted for, and by a reset when the serving session ends. Two tests, each with a passing control that keeps the session alive: - moq-net, mock transport, deterministic: SUBSCRIBE_END and the final group's first frame reach the subscriber, the rest never leaves, the session closes, and the subscriber reads on to `Ok(None)` with the head group alone. The final group, first frame included, is skipped. - moq-tokio, real QUIC: the same over a CONNECTION_CLOSE with a 4 MB final group still behind the flow-control window: 10/20 frames and `Ok(None)`. Co-Authored-By: Claude Fable 5.1 --- .../tests/subscription_end_integrity.rs | 184 +++++++++++++++++ .../tests/subscription_end_integrity.rs | 188 ++++++++++++++++++ 2 files changed, 372 insertions(+) create mode 100644 rs/moq-net/tests/subscription_end_integrity.rs create mode 100644 rs/moq-tokio/tests/subscription_end_integrity.rs diff --git a/rs/moq-net/tests/subscription_end_integrity.rs b/rs/moq-net/tests/subscription_end_integrity.rs new file mode 100644 index 0000000000..243e0fe702 --- /dev/null +++ b/rs/moq-net/tests/subscription_end_integrity.rs @@ -0,0 +1,184 @@ +//! A subscription ends cleanly only when every group in its range has been accounted +//! for; a session closing with a group still in flight ends it with an error. +//! +//! moq-lite, Subscribe: "The publisher closes the stream (FIN) only once every group +//! from start to end has been accounted for, either via a Group Stream (completed or +//! reset) or a SUBSCRIBE_DROP message." And Routing: "when the serving session ends, +//! in-flight subscriptions end with it (a reset)". So `recv_group()` may return +//! `Ok(None)` only after that FIN; any other ending is an error. +//! +//! The publisher opens the final group and writes its first frame (the group stream +//! opens), finishes the track (SUBSCRIBE_END goes out), then writes the rest, finishes +//! the group and disconnects with no await in between, so the final group's tail and +//! its FIN never leave. The close lands before the subscriber reads on. The subscriber +//! must then either see every frame or end with an error; it must not end `Ok(None)` +//! short of the final group. +//! +//! Deterministic: paused time, single-threaded runtime, and every "let the session +//! send" step is an idle-advance rather than a race. + +mod support; + +use std::time::Duration; + +use moq_net::{Hop, Timestamp, Version}; +use support::harness::{MockConnectOptions, MockPair, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(10); +const HEAD: [&[u8]; 2] = [b"head-a", b"head-b"]; +const TAIL: [&[u8]; 2] = [b"tail-a", b"tail-b"]; + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +/// Let the drivers run until nothing is runnable: under paused time an idle runtime +/// auto-advances the clock, so this returns once every queued write has been sent. +async fn settle() { + tokio::time::sleep(Duration::from_millis(1)).await; +} + +fn expected() -> Vec> { + HEAD.iter().chain(TAIL.iter()).map(|p| p.to_vec()).collect() +} + +/// Publish a two-group track over one mock session, cut the publisher's session with the +/// final group half sent (or keep it up), and return what the subscriber read: the frame +/// payloads and the error it ended with, if any. +async fn round(drop_session: bool) -> (Vec>, Option) { + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let mut track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let subscriber = produce_origin(2); + let mut options = MockConnectOptions::new("moq-lite-05".parse::().unwrap()); + options.server_publish = Some(publisher.clone()); + options.client_subscribe = Some(subscriber.clone()); + let MockPair { client, server } = connect_mock(options).await; + + let consumer = subscriber.consume(); + tokio::time::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + let remote = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("bcast")) + .await + .expect("resolve timeout") + .expect("broadcast resolves"); + + // The publisher only learns of a subscription once the subscriber polls it, so the + // reader runs concurrently. It reports the head group drained, then waits to be told + // to read on, so the final group is still unread when the session goes away. + let (drained_tx, drained_rx) = tokio::sync::oneshot::channel(); + let (go_tx, go_rx) = tokio::sync::oneshot::channel::<()>(); + + let reader = tokio::spawn(async move { + let subscription = moq_net::track::Subscription::default().with_start(moq_net::track::Position::group(0)); + let mut sub = remote + .track("video") + .unwrap() + .subscribe(subscription) + .await + .expect("subscribe"); + let mut got = Vec::new(); + let mut drained = Some(drained_tx); + let mut go = Some(go_rx); + loop { + let mut group = match sub.recv_group().await { + Ok(Some(group)) => group, + Ok(None) => return (got, None), + Err(err) => return (got, Some(err)), + }; + loop { + match group.read_frame().await { + Ok(Some(frame)) => got.push(frame.payload.to_vec()), + Ok(None) => break, + Err(err) => return (got, Some(err)), + } + } + if let Some(tx) = drained.take() { + let _ = tx.send(()); + let _ = go.take().unwrap().await; + } + } + }); + + tokio::time::timeout(TIMEOUT, track.used()) + .await + .expect("no subscriber appeared") + .unwrap(); + + let mut head = track.append_group().unwrap(); + for payload in HEAD { + head.write_frame(Timestamp::ZERO, payload).unwrap(); + } + head.finish().unwrap(); + tokio::time::timeout(TIMEOUT, drained_rx) + .await + .expect("the head group never arrived") + .expect("reader gone"); + + // The final group opens and its first frame is sent, then the track's end is + // declared: SUBSCRIBE_END reaches the subscriber with the group still open, as it + // does whenever a finish races a group still in flight. + let mut tail = track.append_group().unwrap(); + tail.write_frame(Timestamp::ZERO, TAIL[0]).unwrap(); + track.finish().unwrap(); + settle().await; + + // The rest of the group, its clean end, and the disconnect happen with no await in + // between, so the session has sent none of it when it closes. + tail.write_frame(Timestamp::ZERO, TAIL[1]).unwrap(); + tail.finish().unwrap(); + drop(track); + let server = if drop_session { + drop(server); + None + } else { + Some(server) + }; + + // The close lands before the subscriber reads on, as it does for a subscriber that is + // merely slower than the network. + settle().await; + let _ = go_tx.send(()); + let result = tokio::time::timeout(TIMEOUT, reader) + .await + .expect("the subscription never ended") + .expect("reader panicked"); + drop((client, server, broadcast, publisher, subscriber)); + result +} + +/// The session closes with the final group in flight: the subscription ends with an +/// error, or delivers every frame. +#[tokio::test] +async fn a_subscription_cut_by_the_session_close_does_not_end_clean() { + tokio::time::pause(); + let (got, err) = round(true).await; + assert!( + err.is_some() || got == expected(), + "the subscription ended Ok(None) with {}/{} frames, got {:?} \ + (a subscription that did not deliver every group must end with an error, not as complete)", + got.len(), + expected().len(), + got.iter() + .map(|p| String::from_utf8_lossy(p).into_owned()) + .collect::>(), + ); +} + +/// With the session kept alive the same track arrives whole and ends `Ok(None)`. +#[tokio::test] +async fn a_finished_track_ends_clean_while_its_session_lives() { + tokio::time::pause(); + let (got, err) = round(false).await; + assert!( + err.is_none() && got == expected(), + "session kept alive: got {} frames, err={err:?}", + got.len() + ); +} diff --git a/rs/moq-tokio/tests/subscription_end_integrity.rs b/rs/moq-tokio/tests/subscription_end_integrity.rs new file mode 100644 index 0000000000..e4da46f1a1 --- /dev/null +++ b/rs/moq-tokio/tests/subscription_end_integrity.rs @@ -0,0 +1,188 @@ +//! Over real QUIC, a subscription ends cleanly only when every group in its range has +//! been accounted for; the publisher disconnecting with a group still in flight ends it +//! with an error. +//! +//! moq-lite, Subscribe: "The publisher closes the stream (FIN) only once every group +//! from start to end has been accounted for, either via a Group Stream (completed or +//! reset) or a SUBSCRIBE_DROP message." And Routing: "when the serving session ends, +//! in-flight subscriptions end with it (a reset)". So `recv_group()` may return +//! `Ok(None)` only after that FIN; any other ending is an error. +//! +//! The in-process twin of this test is `moq-net`'s `subscription_end_integrity`, which +//! pins the rule deterministically over the mock transport. This one repeats it over a +//! real QUIC connection, where the publisher's disconnect is a `CONNECTION_CLOSE` that +//! cuts the final group's stream mid-flight. +//! +//! The subscriber drains a head group first (so the subscription is established and +//! only the tail is at stake), then stops reading for [`STALL`] while the publisher +//! writes a 4 MB final group, finishes the track and disconnects, so the group is still +//! queued behind the connection's flow-control window when the session goes away. The +//! subscriber must then either see every frame or end with an error; it must not end +//! `Ok(None)` short of the final group. +//! +//! Whether the publisher should have been able to flush before disconnecting is a +//! separate question (it has nothing to await today); this test is only about the +//! signal the subscriber gets when it could not. + +use std::time::Duration; + +use moq_tokio::moq_net; + +const TIMEOUT: Duration = Duration::from_secs(10); +const FRAMES: usize = 10; +/// Frames are padded so the final group (4 MB of it) cannot fit in the connection's +/// flow-control window while the subscriber is not reading: without unsent bytes at +/// close, the publisher would only ever lose its FIN, and the test would not cover +/// a group cut mid-flight. +const PADDING: usize = 400_000; +/// How long the subscriber stops reading after the head group, so the final group is +/// still queued in the publisher when the session goes away. +const STALL: Duration = Duration::from_millis(300); + +/// One round: a head group the subscriber drains, then a final group, `finish()`, and +/// the publisher's session dropped with no await in between. Returns how many frames +/// the subscriber read and the error its subscription ended with, if any. +async fn round(drop_session: bool) -> (usize, Option) { + let pub_origin = moq_tokio::origin::spawn(); + let broadcast = pub_origin.create_broadcast("test").expect("create broadcast"); + broadcast.announce(Default::default()).expect("announce"); + let track = broadcast.create_track("video", None).expect("create track"); + + let mut server_config = moq_tokio::listen::Config::default(); + server_config.bind = Some("127.0.0.1:0".parse().unwrap()); + server_config.tls.generate = vec!["localhost".into()]; + let mut server = server_config + .init(Default::default()) + .expect("server init") + .listen() + .await + .expect("listen"); + let port = server.local_addr().expect("local addr").port(); + + // The publisher only learns of a subscription once the subscriber polls it, so the + // reader runs concurrently. It tells the publisher when it has drained the head + // group, so the final group is written against an established subscription. + let (drained_tx, drained_rx) = tokio::sync::oneshot::channel(); + + let reader = tokio::spawn(async move { + let mut client_config = moq_tokio::connect::Config::default(); + client_config.tls.insecure = Some(true); + let sub_origin = moq_tokio::origin::spawn(); + let consumer = sub_origin.consume(); + let client = client_config.init(Default::default()).expect("client init"); + let url: url::Url = format!("https://localhost:{port}").parse().expect("parse url"); + let _connection = client + .with_subscriber(sub_origin) + .with_reconnect(false) + .connect(url) + .established() + .await + .expect("client connect"); + + consumer.routed("test").await.expect("routed"); + let remote = consumer.request_broadcast("test").await.expect("broadcast resolves"); + let mut sub = remote.track("video").unwrap().subscribe(None).await.expect("subscribe"); + + let mut got = 0; + let mut drained = Some(drained_tx); + loop { + let mut group = match sub.recv_group().await { + Ok(Some(group)) => group, + Ok(None) => return (got, None), + Err(err) => return (got, Some(err)), + }; + loop { + match group.read_frame().await { + Ok(Some(_)) => got += 1, + Ok(None) => break, + Err(err) => return (got, Some(err)), + } + } + // The head group is in: from here on, only the tail can be lost. Stop reading + // for a moment, so the publisher's final group is still queued behind the + // connection's flow-control window when it disconnects. + if let Some(tx) = drained.take() { + let _ = tx.send(()); + tokio::time::sleep(STALL).await; + } + } + }); + + let request = tokio::time::timeout(TIMEOUT, server.accept()) + .await + .expect("accept timeout") + .expect("no incoming connection"); + let session = request + .with_publisher(&pub_origin) + .ok() + .await + .expect("publisher session"); + + tokio::time::timeout(TIMEOUT, track.used()) + .await + .expect("no subscriber appeared") + .expect("track closed"); + + write_group(&track); + tokio::time::timeout(TIMEOUT, drained_rx) + .await + .expect("the head group never arrived") + .expect("reader gone"); + + // The final group, a clean group end and a clean track end, innermost first and with + // no await in between: the session has sent none of it yet. + write_group(&track); + track.finish().expect("finish track"); + drop(track); + + // The publisher is done, so it disconnects. Nothing it could have awaited first. + let session = if drop_session { + drop(session); + None + } else { + Some(session) + }; + + let result = tokio::time::timeout(TIMEOUT, reader) + .await + .expect("the subscription never ended") + .expect("reader panicked"); + drop((session, broadcast, pub_origin, server)); + result +} + +/// Append one group of [`FRAMES`] padded frames. +fn write_group(track: &moq_net::track::Producer) { + let mut group = track.append_group().expect("append group"); + for _ in 0..FRAMES { + group + .write_frame(moq_net::Timestamp::ZERO, vec![0; PADDING]) + .expect("write frame"); + } + group.finish().expect("finish group"); +} + +/// The publisher disconnects with the final group in flight: the subscription ends +/// with an error, or delivers every frame. +#[tokio::test] +async fn a_subscription_cut_by_the_publisher_disconnecting_does_not_end_clean() { + let (got, err) = round(true).await; + assert!( + err.is_some() || got == 2 * FRAMES, + "the subscription ended Ok(None) with {got}/{} frames \ + (a subscription that did not deliver every group must end with an error, not as complete)", + 2 * FRAMES, + ); +} + +/// With the session kept alive until the subscriber is done, the same track arrives +/// whole and ends `Ok(None)`. +#[tokio::test] +async fn a_finished_track_ends_clean_while_its_session_lives() { + let (got, err) = round(false).await; + assert!( + err.is_none() && got == 2 * FRAMES, + "session kept alive: got {got}/{} frames, err={err:?}", + 2 * FRAMES, + ); +} From 47c4e7e07ed10fc0fa4b2263c1280ffdac94a7ef Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 00:35:51 -0700 Subject: [PATCH 03/11] fix(net): end a Rust track with its session's error Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/session.rs | 18 +- rs/moq-net/src/ietf/subscriber.rs | 29 +++- rs/moq-net/src/lite/session.rs | 4 + rs/moq-net/src/lite/subscriber.rs | 44 +++-- rs/moq-net/src/model/resume.rs | 154 ++++++++++-------- rs/moq-net/src/model/track.rs | 63 ++++++- .../tests/subscription_end_integrity.rs | 81 ++++++++- rs/moq-net/tests/support/mock.rs | 126 +++++++++----- 8 files changed, 375 insertions(+), 144 deletions(-) diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index f1c7258608..cb0fd4538a 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -239,7 +239,7 @@ where Ok(()) })); - kio::wait(|waiter| { + let res = kio::wait(|waiter| { use std::task::Poll; if let Poll::Ready(err) = waiter.poll_future(adapter_run.as_mut()) { return Poll::Ready(Err::<(), Error>(err)); @@ -261,7 +261,12 @@ where } Poll::Pending }) - .await + .await; + if let Err(err) = &res { + // Every track this session was receiving ends with its error. + subscriber.abort(err); + } + res } _ => { // Send SETUP and keep the stream alive: it is also our GOAWAY channel. @@ -366,7 +371,7 @@ where Ok(()) })); - kio::wait(|waiter| { + let res = kio::wait(|waiter| { use std::task::Poll; if let Poll::Ready(err) = waiter.poll_future(unis.as_mut()) { return Poll::Ready(Err::<(), Error>(err)); @@ -391,7 +396,12 @@ where } Poll::Pending }) - .await + .await; + if let Err(err) = &res { + // Every track this session was receiving ends with its error. + subscriber.abort(err); + } + res } }; diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 342cf6e054..1394371e74 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -152,22 +152,32 @@ struct State { broadcasts: HashMap, } -impl Drop for State { - fn drop(&mut self) { - // The session dispatcher owns this state and can be dropped at any await. - // Active receive tasks abort their own groups. Cancel any head waiting for - // its tail here, along with the track. Ordinary unsubscribe removes its entry. +impl State { + /// End every active subscription with the error that ended the session. + /// + /// Active receive tasks abort their own groups. Abort any head waiting for its + /// tail here, along with the track. Ordinary unsubscribe removes its entry. + fn abort(&mut self, err: &Error) { for (_, track) in self.subscribes.drain() { if let Fill::Ready { producer, .. } = &*track.fill.read() { - let _ = producer.clone().abort(Error::Cancel); + let _ = producer.clone().abort(err.clone()); } if let Some(producer) = track.producer { - let _ = producer.abort(Error::Cancel); + let _ = producer.abort(err.clone()); } } } } +impl Drop for State { + fn drop(&mut self) { + // The session dispatcher owns this state and can be dropped at any await. A + // session that ended with an error already aborted these with it; what + // remains was cancelled with the dispatcher. + self.abort(&Error::Cancel); + } +} + /// The head of a joined group, delivered on the subscription's fill fetch stream. /// /// Draft-20's current-group join (section 5.1.6) splits one group across two streams: the @@ -513,6 +523,11 @@ where } } + /// End every active subscription with the error that ended the session. + pub fn abort(&self, err: &Error) { + self.state.lock().abort(err); + } + /// Leave `alias` in the state a cancelled subscription leaves behind: bound to a /// subscription, then retired. /// diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index 0d00e5a29d..7417c05c2c 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -230,6 +230,10 @@ where { pub(crate) fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { let res = std::task::ready!(self.poll_protocol(waiter)); + if let Err(err) = &res { + // Every track this session was receiving ends with its error. + self.subscriber.abort(err); + } match &res { Err(Error::Transport(_)) => { tracing::info!("session terminated"); diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index ca7bff6ba4..fd4b070e3e 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -462,20 +462,28 @@ impl Subscriber { // poll returning Ready. struct SubscriptionCleanup(Lock>); -impl Drop for SubscriptionCleanup { - fn drop(&mut self) { - // Group machines own their cancellation cleanup independently. This records - // session cancellation as the track's terminal state. +impl SubscriptionCleanup { + /// End every active subscription with the session's error. Group machines own + /// their cleanup independently; this records the track's terminal state. + fn abort(&self, err: &Error) { for (_, entry) in self.0.lock().drain() { - let _ = entry.producer.abort(Error::Cancel); + let _ = entry.producer.abort(err.clone()); } } } +impl Drop for SubscriptionCleanup { + fn drop(&mut self) { + // A session that ended with an error already aborted these with it; what + // remains was cancelled with the driver. + self.abort(&Error::Cancel); + } +} + pub(super) struct SubscriberDriver { subscriber: Subscriber, - /// Aborts whatever is still subscribed when the driver is dropped. - _cleanup: SubscriptionCleanup, + /// Aborts whatever is still subscribed when the session ends. + cleanup: SubscriptionCleanup, /// One machine per permitted prefix. Only an error ends the session; a /// prefix finishing cleanly (publisher FIN) just retires. prefixes: Vec>, @@ -499,7 +507,7 @@ impl SubscriberDriver { Self { prefixes, - _cleanup: SubscriptionCleanup(subscriber.subscribes.clone()), + cleanup: SubscriptionCleanup(subscriber.subscribes.clone()), uni: UniAccept::new(subscriber.clone()), bandwidth: Some(RecvBandwidth::new(subscriber.clone())), datagrams: Some(DatagramRecv::new(subscriber.clone())), @@ -508,6 +516,11 @@ impl SubscriberDriver { } } + /// End every active subscription with the error that ended the session. + pub fn abort(&self, err: &Error) { + self.cleanup.abort(err); + } + pub fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { let mut i = 0; while i < self.prefixes.len() { @@ -2997,8 +3010,8 @@ impl Establish { return Poll::Ready(Ok(self.activate(stream))); } EstablishState::WaitOk { stream } => { - if self.closed.poll_closed(&mut cx).is_ready() { - return Poll::Ready(Err(Error::Dropped)); + if let Poll::Ready(err) = self.closed.poll_closed(&mut cx) { + return Poll::Ready(Err(Error::from_transport(err))); } let resp = ready!(stream.reader.poll_decode::(&mut cx))?; if !matches!(resp, lite::SubscribeResponse::Ok(_)) { @@ -3181,8 +3194,8 @@ impl TrackInfoFetch { self.state = TrackInfoState::Read { stream }; } TrackInfoState::Read { stream } => { - if self.closed.poll_closed(&mut cx).is_ready() { - return Poll::Ready(Err(Error::Dropped)); + if let Poll::Ready(err) = self.closed.poll_closed(&mut cx) { + return Poll::Ready(Err(Error::from_transport(err))); } let info = ready!(stream.reader.poll_decode::(&mut cx))?; // The publisher FINs after TRACK_INFO; FIN our side too and let the @@ -3474,9 +3487,10 @@ impl ServeLoop { } } - // (5) The session died: hand the track back for another route. - if self.closed.poll_closed(&mut cx).is_ready() { - return Poll::Ready(ServeEnd::GiveBack(Error::Dropped)); + // (5) The session died: hand the track back for another route, with + // the session's error in case none takes it. + if let Poll::Ready(err) = self.closed.poll_closed(&mut cx) { + return Poll::Ready(ServeEnd::GiveBack(Error::from_transport(err))); } return Poll::Pending; diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index ee081f1474..cbce717f3d 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -1405,8 +1405,8 @@ impl SegmentSub { } /// Move an active cursor into terminal retention and mark the segment done. - fn complete(&mut self, count: Option) { - let previous = std::mem::replace(&mut self.sub, SubState::Done(count)); + fn complete(&mut self, end: Result) { + let previous = std::mem::replace(&mut self.sub, SubState::Done(end)); if let SubState::Active(sub) = previous { self.terminal = Some(*sub); } @@ -1449,23 +1449,22 @@ enum SubState { /// Live cursor over the underlying track. Boxed: the subscriber dwarfs the other /// variants, and every segment holds this enum. Active(Box), - /// The underlying track ended: `Some` with the group count when it finished - /// cleanly, `None` when it aborted or was dropped. An abort is deliberately - /// not surfaced: a dead route stalls the logical track until the next switch - /// replaces it. - Done(Option), + /// The underlying track ended: `Ok` with the group count when it finished + /// cleanly, `Err` with the cause when it aborted or was dropped. An abort only + /// surfaces once no switch can follow: until then a dead route stalls the + /// logical track for the next switch to replace. + Done(Result), } /// A live subscription spliced across every segment of a logical track. /// /// Reads switch between the underlying [`track::Subscriber`]s at the segment /// boundaries. A segment's session failing does not error the subscription; it -/// stalls until [`Producer::switch`] provides a replacement, or ends cleanly once -/// the producer [`finish`](Producer::finish)es and the final segment completes. -/// The producer itself going away without a terminal state ends the track as its -/// final segment's track ended, once the remaining segments drain: cleanly if it -/// finished, [`Error::Dropped`] if it died. With nobody left to splice a -/// replacement, a stall would never end. +/// stalls until [`Producer::switch`] provides a replacement. Once the producer +/// [`finish`](Producer::finish)es, or goes away without a terminal state, no +/// replacement can follow, so the track ends as its final segment's track ended +/// once the remaining segments drain: cleanly if it finished, with its error if it +/// died. With nobody left to splice a replacement, a stall would never end. pub struct Subscriber { state: kio::Consumer, @@ -1535,7 +1534,7 @@ impl Subscriber { break; } if seg.pruned { - seg.sub = SubState::Done(None); + seg.sub = SubState::Done(Err(Error::Dropped)); seg.terminal = None; seg.parked.clear(); cut -= 1; @@ -1831,7 +1830,8 @@ impl Subscriber { // The copy was asked for the cache's newest group and started past it: its // source judged that group stale, so the older cache is no use to live reads. if start.is_some_and(|start| edge.is_some_and(|edge| start > edge)) { - seg.complete(None); + // Discarded, not finished: there is no group count, same as a dropped segment. + seg.complete(Err(Error::Dropped)); return Poll::Ready(()); } } @@ -1852,7 +1852,7 @@ impl Subscriber { seg.sub = SubState::Active(Box::new(sub)); } // The underlying track was rejected or closed: stall, not error. - Err(_) => seg.sub = SubState::Done(None), + Err(err) => seg.sub = SubState::Done(Err(err)), } } Poll::Ready(()) @@ -1881,18 +1881,17 @@ impl Subscriber { return Poll::Ready(Some(group)); } Ok(None) => { - let count = sub.poll_finished(waiter).map(|res| res.ok()); - let count = match count { - Poll::Ready(count) => count, - Poll::Pending => None, + let end = match sub.poll_finished(waiter) { + Poll::Ready(end) => end, + Poll::Pending => Err(Error::Dropped), }; - seg.complete(count); + seg.complete(end); return Poll::Ready(None); } // A dead segment stalls the logical track rather than erroring; // the next switch resumes it. - Err(_) => { - seg.complete(None); + Err(err) => { + seg.complete(Err(err)); return Poll::Ready(None); } }, @@ -1971,15 +1970,15 @@ impl Subscriber { } // The track ran out at or below the floor: the segment drained. Poll::Ready(Ok(None)) => { - let count = match sub.poll_finished(waiter) { - Poll::Ready(count) => count.ok(), - Poll::Pending => None, + let end = match sub.poll_finished(waiter) { + Poll::Ready(end) => end, + Poll::Pending => Err(Error::Dropped), }; - seg.complete(count); + seg.complete(end); } // A dead segment stalls the logical track rather than erroring; // the next switch resumes it. - Poll::Ready(Err(_)) => seg.complete(None), + Poll::Ready(Err(err)) => seg.complete(Err(err)), Poll::Pending => all_done = false, }, SubState::Pending(_) => all_done = false, @@ -2007,15 +2006,10 @@ impl Subscriber { if let Some(err) = &self.abort { return Poll::Ready(Err(err.clone())); } - if all_done { - if self.finished { - return Poll::Ready(Ok(None)); - } - // The producer is gone without finishing: no takeover can ever resume - // the drained segments, so the track ends as its final segment did. - if self.closed { - return Poll::Ready(self.orphan_end().map(|()| None)); - } + // Finished or gone, no switch can follow: the track ends as its final + // segment did. + if all_done && (self.finished || self.closed) { + return Poll::Ready(self.final_end().map(|()| None)); } return Poll::Pending; } @@ -2028,9 +2022,9 @@ impl Subscriber { /// out exactly once no matter how many routes contributed to it. /// /// Returns `Poll::Ready(Ok(None))` once every segment completed and the - /// producer finished, or was dropped with the final segment's track finished. - /// Returns `Poll::Ready(Err(_))` if the producer aborted, or was dropped and - /// the final segment's track died ([`Error::Dropped`]). + /// producer finished or was dropped, with the final segment's track finished. + /// Returns `Poll::Ready(Err(_))` if the producer aborted, or if it finished or + /// was dropped and the final segment's track died (with that track's error). pub fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll>> { self.poll_sync(waiter); @@ -2134,16 +2128,10 @@ impl Subscriber { if let Some(err) = &self.abort { return Poll::Ready(Err(err.clone())); } - if all_done { - if self.finished { - return Poll::Ready(Ok(None)); - } - // The producer is gone without finishing: no takeover can ever - // resume the drained segments, so the track ends as its final - // segment did rather than stalling forever. - if self.closed { - return Poll::Ready(self.orphan_end().map(|()| None)); - } + // Finished or gone, no switch can follow: the track ends as its final + // segment did rather than stalling forever. + if all_done && (self.finished || self.closed) { + return Poll::Ready(self.final_end().map(|()| None)); } Poll::Pending } @@ -2224,7 +2212,7 @@ impl Subscriber { // its newest segment does, and the way it did. if self.closed && !pending_activation { return match ready!(self.poll_final(waiter)) { - Some(_) => Poll::Ready(Ok(None)), + Some(end) => Poll::Ready(end.map(|_| None)), None => Poll::Ready(Err(Error::Dropped)), }; } @@ -2249,40 +2237,42 @@ impl Subscriber { if !self.finished && !self.closed { return Poll::Pending; } - let end = ready!(self.poll_final(waiter)); - match self.finished { - true => Poll::Ready(Ok(end.unwrap_or(0))), - // The producer is gone without finishing: the track ends as its final - // segment did. - false => Poll::Ready(end.ok_or(Error::Dropped)), + // The track ends as its final segment did, or with nothing spliced, cleanly + // only if it finished. + match ready!(self.poll_final(waiter)) { + Some(end) => Poll::Ready(end), + None if self.finished => Poll::Ready(Ok(0)), + None => Poll::Ready(Err(Error::Dropped)), } } /// Wait for the final segment's track to end: its group count when it - /// finished, `None` when it died or there is no segment. Earlier segments don't + /// finished, its error when it died, and `None` when there is no segment. Earlier segments don't /// decide the end. Only the subscription is resolved here: consuming groups, or /// completing the segment, would steal them from a `recv_group` caller on the /// same subscriber. - fn poll_final(&mut self, waiter: &kio::Waiter) -> Poll> { + fn poll_final(&mut self, waiter: &kio::Waiter) -> Poll>> { let Some(seg) = self.segments.last_mut() else { return Poll::Ready(None); }; ready!(Self::poll_activate(seg, &self.last_prefs, self.min_sequence, waiter)); match &mut seg.sub { - SubState::Done(count) => Poll::Ready(*count), + SubState::Done(end) => Poll::Ready(Some(end.clone())), // Observe only: the cursor may still hold groups, so the read path // completes the segment once it drains. - SubState::Active(sub) => Poll::Ready(ready!(sub.poll_finished(waiter)).ok()), + SubState::Active(sub) => Poll::Ready(Some(ready!(sub.poll_finished(waiter)))), SubState::Pending(_) => unreachable!("poll_activate resolved above"), } } - /// How a logical track whose producer went away without finishing ends, once - /// every segment drained: cleanly if its final segment's track finished, since - /// that is where the content ended, and [`Error::Dropped`] otherwise. - fn orphan_end(&self) -> Result<()> { + /// How a logical track ends once every segment drained and no switch can follow + /// (the producer finished or went away): as its final segment's track ended, since + /// that is where the content ended. A finished track with no segment ends cleanly; + /// an orphaned one with [`Error::Dropped`]. + fn final_end(&self) -> Result<()> { match self.segments.last().map(|seg| &seg.sub) { - Some(SubState::Done(Some(_))) => Ok(()), + Some(SubState::Done(end)) => end.clone().map(|_| ()), + None if self.finished => Ok(()), _ => Err(Error::Dropped), } } @@ -4895,12 +4885,34 @@ mod test { assert_eq!(recv(&mut sub), 0); // The segment dies, then the producer goes away without finish/abort: - // no takeover can ever come, so stalling would hang forever. - track_a.abort(Error::Cancel).unwrap(); + // no takeover can ever come, so stalling would hang forever. The track + // ends with the segment's own error. + track_a.abort(Error::Timeout).unwrap(); drop(producer); let result = sub.recv_group().now_or_never().expect("must not stall forever"); - assert!(matches!(result, Err(Error::Dropped))); + assert!(matches!(result, Err(Error::Timeout))); + } + + /// A finished logical track ends as its final segment did: a segment cut off + /// after the finish (its session died with a group still in flight) is an + /// error, not a clean end. + #[tokio::test] + async fn finished_producer_ends_with_a_dead_final_segment() { + let (mut track_a, consumer_a) = track_pair("a"); + + let mut producer = Producer::new(); + producer.switch(&consumer_a, None).unwrap(); + let mut sub = producer.consume().subscribe(None); + + write_group(&mut track_a, 0, "a0"); + assert_eq!(recv(&mut sub), 0); + + producer.finish().unwrap(); + track_a.abort(Error::Timeout).unwrap(); + + let result = sub.recv_group().now_or_never().expect("must not stall forever"); + assert!(matches!(result, Err(Error::Timeout))); } #[tokio::test] @@ -4943,12 +4955,12 @@ mod test { ); match clean { true => track_a.finish().unwrap(), - false => track_a.abort(Error::Cancel).unwrap(), + false => track_a.abort(Error::Timeout).unwrap(), } let result = sub.finished().now_or_never().expect("must not stall forever"); match clean { true => assert!(matches!(result, Ok(0))), - false => assert!(matches!(result, Err(Error::Dropped))), + false => assert!(matches!(result, Err(Error::Timeout))), } } } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 8ac10f45b9..d4f3216a6d 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -241,6 +241,10 @@ pub(crate) struct TrackState { // The error that caused the track to be aborted, if any. abort: Option, + // Whether the declared end still stands after an abort: every group below it was + // produced and finished before the abort landed. See [`Self::is_complete`]. + settled: bool, + // Active subscriptions, in their own [`kio::Shared`] so a read-only `Consumer` // registers under that lock instead of writing back into the track state. // Kept here (rather than threaded through every handle) so any holder reaches it. @@ -1077,9 +1081,25 @@ impl TrackState { /// until the remaining groups are produced, or until the last producer drops without /// them. Drives the end-of-stream signal from /// the read methods (`recv_group` / `next_group` / `read_frame` return `None`). + /// + /// An abort before the end settled wins over it: a group below the boundary was + /// still open, so the track was cut off rather than ended. fn is_complete(&self) -> bool { - self.final_sequence - .is_some_and(|fin| self.sealed || self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin) + // `sealed` is a clean end when the last producer drops. An abort still wins + // unless that end had already settled: a group below it was still open. + let reached = self + .final_sequence + .is_some_and(|fin| self.sealed || self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin); + reached && (self.abort.is_none() || self.settled) + } + + /// Whether the declared end is reached and every cached group below it finished, + /// so nothing the end promised is still in flight. + fn is_settled(&self) -> bool { + let Some(fin) = self.final_sequence else { + return false; + }; + self.is_complete() && self.lookup.range(..fin).all(|(_, slot)| slot.group.is_finished()) } /// Where a replacement route should pick this track up: one past the last frame @@ -1172,6 +1192,8 @@ fn commit_abort(mut state: kio::Mut<'_, TrackState>, err: Error) { // Snapshot the frame boundary before the cache it's derived from goes away: an // abort is exactly when a replacement route asks where to resume. state.resume = state.resume_position(); + // Decided before the cache goes: the groups below the end are the evidence. + state.settled = state.is_settled(); state.abort = Some(err); state.clear_cache(); state.datagrams.clear(); @@ -6894,6 +6916,43 @@ mod test { assert!(done.is_none(), "track completes once the boundary is reached"); } + /// An abort before the declared end settled wins over it: the boundary was reached, + /// but a group below it was still open, so the track was cut off rather than ended. + #[tokio::test] + async fn abort_before_the_end_settles_wins() { + let mut producer = track_producer("test", None); + let mut consumer = producer.subscribe(None); + + let head = producer.create_group(group::Info { sequence: 0 }).unwrap(); + head.finish().unwrap(); + let _tail = producer.create_group(group::Info { sequence: 1 }).unwrap(); + producer.finish_at(2).unwrap(); + assert_eq!(consumer.assert_group().sequence, 0); + + producer.abort(Error::Timeout).unwrap(); + let res = consumer.recv_group().now_or_never().expect("should not block"); + assert!(matches!(res, Err(Error::Timeout))); + } + + /// An abort after every group below the declared end finished leaves the end standing. + #[tokio::test] + async fn abort_after_the_end_settles_ends_clean() { + let mut producer = track_producer("test", None); + let mut consumer = producer.subscribe(None); + + for sequence in 0..2 { + let group = producer.create_group(group::Info { sequence }).unwrap(); + group.finish().unwrap(); + } + producer.finish_at(2).unwrap(); + assert_eq!(consumer.assert_group().sequence, 0); + assert_eq!(consumer.assert_group().sequence, 1); + + producer.abort(Error::Timeout).unwrap(); + let res = consumer.recv_group().now_or_never().expect("should not block"); + assert!(matches!(res, Ok(None))); + } + #[tokio::test] async fn recv_group_finishes_without_waiting_for_gaps() { let producer = track_producer("test", None); diff --git a/rs/moq-net/tests/subscription_end_integrity.rs b/rs/moq-net/tests/subscription_end_integrity.rs index 243e0fe702..4a026897a3 100644 --- a/rs/moq-net/tests/subscription_end_integrity.rs +++ b/rs/moq-net/tests/subscription_end_integrity.rs @@ -50,7 +50,7 @@ fn expected() -> Vec> { async fn round(drop_session: bool) -> (Vec>, Option) { let publisher = produce_origin(1); let broadcast = publisher.create_broadcast("bcast").unwrap(); - let mut track = broadcast.create_track("video", None).unwrap(); + let track = broadcast.create_track("video", None).unwrap(); broadcast.announce(Default::default()).unwrap(); let subscriber = produce_origin(2); @@ -182,3 +182,82 @@ async fn a_finished_track_ends_clean_while_its_session_lives() { got.len() ); } + +/// Every wire the session-death rule covers: lite before and after the track and +/// subscribe-response streams, and IETF on its shared and per-request control streams. +const DEATH_VERSIONS: &[&str] = &[ + "moq-lite-03", + "moq-lite-05", + "moq-lite-07", + "moq-transport-14", + "moq-transport-17", + "moq-transport-22", +]; + +/// The error the publisher's session is aborted with, carried in the close code. +const DEATH: moq_net::SessionError = moq_net::SessionError::App(7); + +/// Kill the publisher's session with a group still open, and return the error the +/// subscriber's `recv_group` ends with once it drains what arrived. +async fn killed(version: &str) -> Option { + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let subscriber = produce_origin(2); + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(publisher.clone()); + options.client_subscribe = Some(subscriber.clone()); + let MockPair { client, server } = connect_mock(options).await; + + let consumer = subscriber.consume(); + consumer.routed("bcast").await.expect("routed"); + let remote = consumer.request_broadcast("bcast").await.expect("broadcast resolves"); + // Subscribing resolves only once the publisher serves the track, which it does + // only after seeing the subscription, so the reader runs concurrently. + let reader = tokio::spawn(async move { + let subscription = moq_net::track::Subscription::default().with_start(moq_net::track::Position::group(0)); + let mut sub = remote + .track("video") + .unwrap() + .subscribe(subscription) + .await + .expect("subscribe"); + loop { + let mut group = match sub.recv_group().await { + Ok(Some(group)) => group, + Ok(None) => return None, + Err(err) => return Some(err), + }; + // The open group ends however it ends; only the track's end is at stake here. + while let Ok(Some(_)) = group.read_frame().await {} + } + }); + + track.used().await.expect("no subscriber appeared"); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, HEAD[0]).unwrap(); + settle().await; + + server.abort(moq_net::Error::Session(DEATH)); + let err = reader.await.expect("reader panicked"); + drop((client, server, group, track, broadcast, publisher, subscriber)); + err +} + +/// A session dying mid-track ends the subscriber's track with the session's own +/// error: not a clean end, and not a generic `Dropped` or `Cancel`. +#[tokio::test] +async fn a_session_death_ends_the_track_with_its_error() { + tokio::time::pause(); + for version in DEATH_VERSIONS { + let err = tokio::time::timeout(TIMEOUT, killed(version)) + .await + .unwrap_or_else(|_| panic!("{version}: the track never ended")); + assert!( + matches!(&err, Some(moq_net::Error::Session(code)) if *code == DEATH), + "{version}: the track ended with {err:?}, not the session's error" + ); + } +} diff --git a/rs/moq-net/tests/support/mock.rs b/rs/moq-net/tests/support/mock.rs index 81fbd49f1c..1b5e052126 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -8,7 +8,9 @@ //! //! The mock guarantees that data written and FIN'd on a stream before `close()` //! is readable by the peer, eliminating the Quinn CONNECTION_CLOSE race that -//! plagues real-transport tests. +//! plagues real-transport tests. Like a real CONNECTION_CLOSE, nothing written +//! after it is delivered, and a stream left unfinished fails with the close once +//! its earlier data is read. use std::{ sync::{Arc, Mutex}, @@ -20,24 +22,32 @@ use web_transport_trait::poll; // ── Error ─────────────────────────────────────────────────────────── -/// Error type for mock transport operations. +/// Error type for mock transport operations: a session close or a stream reset, +/// each reporting its code only through its own registry. #[derive(Debug, Clone)] pub struct MockError { - code: Option, + session: Option, + stream: Option, reason: String, } impl MockError { fn closed() -> Self { + Self::session(0, "session closed".into()) + } + + fn session(code: u32, reason: String) -> Self { Self { - code: Some(0), - reason: "session closed".into(), + session: Some(code), + stream: None, + reason, } } fn stream_reset(code: u32) -> Self { Self { - code: Some(code), + session: None, + stream: Some(code), reason: "stream reset".into(), } } @@ -53,11 +63,11 @@ impl std::error::Error for MockError {} impl web_transport_trait::Error for MockError { fn session_error(&self) -> Option<(u32, String)> { - self.code.map(|c| (c, self.reason.clone())) + self.session.map(|c| (c, self.reason.clone())) } fn stream_error(&self) -> Option { - self.code + self.stream } } @@ -68,6 +78,8 @@ enum StreamChunk { Data(Bytes), Fin, Reset(u32), + /// The connection closed before the stream ended; never queued. + Closed, } /// Shared closed-signal state between a paired SendStream and RecvStream. @@ -100,45 +112,60 @@ pub struct MockSendStream { /// Acknowledge the FIN as soon as it is sent, for a stream the peer's transport holds /// back from its application (see [`MockSession::hold_unis`]). ack_fin: bool, + conn: Arc, +} + +impl MockSendStream { + /// Queue a chunk for the peer, unless the connection closed: nothing sent after a + /// CONNECTION_CLOSE reaches the peer. + fn push(&mut self, chunk: StreamChunk) -> Result<(), MockError> { + if let Some(err) = self.conn.error() { + self.tx = None; + return Err(err); + } + let tx = self.tx.as_ref().ok_or_else(MockError::closed)?; + tx.try_push(chunk).map_err(|_| MockError::closed()) + } } impl poll::SendStream for MockSendStream { type Error = MockError; fn poll_write(&mut self, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { - let Some(tx) = self.tx.as_ref() else { - return Poll::Ready(Err(MockError::closed())); - }; - match tx.try_push(StreamChunk::Data(Bytes::copy_from_slice(buf))) { - Ok(()) => Poll::Ready(Ok(buf.len())), - Err(_) => Poll::Ready(Err(MockError::closed())), - } + Poll::Ready(self.push(StreamChunk::Data(Bytes::copy_from_slice(buf))).map(|()| buf.len())) } fn set_priority(&mut self, _order: u8) {} fn finish(&mut self) -> Result<(), Self::Error> { - if let Some(tx) = self.tx.take() { - let _ = tx.try_push(StreamChunk::Fin); + if self.tx.is_some() { + let _ = self.push(StreamChunk::Fin); if self.ack_fin { self.closed.set(Ok(())); } + self.tx = None; } Ok(()) } fn reset(&mut self, code: u32) { - if let Some(tx) = self.tx.take() { - let _ = tx.try_push(StreamChunk::Reset(code)); + if self.tx.is_some() { + let _ = self.push(StreamChunk::Reset(code)); + self.tx = None; } } fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { // Register before checking so a signal racing this poll still wakes it. - self.closed.waiters.register(self.park.hold(cx)); + let waiter = self.park.hold(cx); + self.closed.waiters.register(waiter); + self.conn.waiters.register(waiter); match self.closed.result.lock().unwrap().clone() { Some(result) => Poll::Ready(result), - None => Poll::Pending, + None => match self.conn.error() { + Some(err) => Poll::Ready(Err(err)), + None => Poll::Pending, + }, } } } @@ -147,8 +174,8 @@ impl Drop for MockSendStream { fn drop(&mut self) { // Dropped without an explicit FIN: deliver an implicit one, matching a // sender that went away cleanly. - if let Some(tx) = self.tx.take() { - let _ = tx.try_push(StreamChunk::Fin); + if self.tx.is_some() { + let _ = self.push(StreamChunk::Fin); } } } @@ -165,16 +192,22 @@ pub struct MockRecvStream { /// Shared signal to notify the peer's send-side `poll_closed`. closed: Arc, park: kio::Park, + conn: Arc, } impl MockRecvStream { - /// Pop the next chunk, mapping queue closure to an implicit FIN. + /// Pop the next chunk, mapping queue closure to an implicit FIN. Once everything + /// sent before a CONNECTION_CLOSE is read, an unfinished stream fails with it. fn poll_chunk(&mut self, cx: &mut Context<'_>) -> Poll> { let waiter = self.park.hold(cx); + self.conn.waiters.register(waiter); match self.rx.poll_pop(waiter) { Poll::Ready(Ok(chunk)) => Poll::Ready(Some(chunk)), Poll::Ready(Err(_)) => Poll::Ready(None), - Poll::Pending => Poll::Pending, + Poll::Pending => match self.conn.error() { + Some(_) => Poll::Ready(Some(StreamChunk::Closed)), + None => Poll::Pending, + }, } } } @@ -212,6 +245,10 @@ impl poll::RecvStream for MockRecvStream { self.done = true; Poll::Ready(Err(MockError::stream_reset(code))) } + Some(StreamChunk::Closed) => { + self.done = true; + Poll::Ready(Err(self.conn.error().unwrap_or_else(MockError::closed))) + } } } @@ -236,6 +273,10 @@ impl poll::RecvStream for MockRecvStream { self.done = true; return Poll::Ready(Err(MockError::stream_reset(code))); } + Some(StreamChunk::Closed) => { + self.done = true; + return Poll::Ready(Err(self.conn.error().unwrap_or_else(MockError::closed))); + } } } } @@ -253,7 +294,7 @@ impl Drop for MockRecvStream { // ── Stream pair constructor ───────────────────────────────────────── /// Create a linked (send, recv) stream pair. -fn new_stream_pair() -> (MockSendStream, MockRecvStream) { +fn new_stream_pair(conn: &Arc) -> (MockSendStream, MockRecvStream) { let queue = kio::Queue::new(); let closed = Arc::new(ClosedSignal::default()); @@ -262,6 +303,7 @@ fn new_stream_pair() -> (MockSendStream, MockRecvStream) { closed: closed.clone(), park: kio::Park::default(), ack_fin: false, + conn: conn.clone(), }; let recv = MockRecvStream { rx: queue, @@ -269,6 +311,7 @@ fn new_stream_pair() -> (MockSendStream, MockRecvStream) { done: false, closed, park: kio::Park::default(), + conn: conn.clone(), }; (send, recv) } @@ -287,6 +330,14 @@ struct ConnectionState { waiters: kio::Fan, } +impl ConnectionState { + /// The close every stream and accept fails with, once the connection closed. + fn error(&self) -> Option { + let state = self.close_state.lock().unwrap(); + state.as_ref().map(|(code, reason)| MockError::session(*code, reason.clone())) + } +} + /// Per-side state: stream queues and a reference to the shared connection. struct SessionSide { /// Bidi streams opened by the peer, popped by accept_bi. @@ -370,8 +421,8 @@ impl poll::Session for MockSession { _cx: &mut Context<'_>, ) -> Poll> { // Create two stream pairs: one for each direction. - let (our_send, peer_recv) = new_stream_pair(); - let (peer_send, our_recv) = new_stream_pair(); + let (our_send, peer_recv) = new_stream_pair(&self.side.conn); + let (peer_send, our_recv) = new_stream_pair(&self.side.conn); // Deliver (peer_send, peer_recv) to the peer's accept_bi. match self.side.peer_bidi.try_push((peer_send, peer_recv)) { @@ -381,7 +432,7 @@ impl poll::Session for MockSession { } fn poll_open_uni(&mut self, _cx: &mut Context<'_>) -> Poll> { - let (mut our_send, peer_recv) = new_stream_pair(); + let (mut our_send, peer_recv) = new_stream_pair(&self.side.conn); if let Some(held) = self.side.held.lock().unwrap().as_mut() { our_send.ack_fin = true; @@ -442,10 +493,7 @@ impl poll::Session for MockSession { self.side.conn.waiters.register(self.closed.hold(cx)); let state = self.side.conn.close_state.lock().unwrap().clone(); match state { - Some((code, reason)) => Poll::Ready(MockError { - code: Some(code), - reason, - }), + Some((code, reason)) => Poll::Ready(MockError::session(code, reason)), None => Poll::Pending, } } @@ -485,17 +533,7 @@ impl MockSession { impl MockSession { fn close_error(&self) -> MockError { - self.side - .conn - .close_state - .lock() - .unwrap() - .as_ref() - .map(|(code, reason)| MockError { - code: Some(*code), - reason: reason.clone(), - }) - .unwrap_or_else(MockError::closed) + self.side.conn.error().unwrap_or_else(MockError::closed) } } From 3d5ef1dbe11c0f21c9f8e95e1872c22dc06bae34 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 00:43:07 -0700 Subject: [PATCH 04/11] fix(net): end a JS track with its session's error Co-Authored-By: Claude Opus 5.5 --- js/net/src/error.ts | 25 ++++++++ js/net/src/ietf/adapter.ts | 23 +++++-- js/net/src/ietf/connection.ts | 2 +- js/net/src/ietf/publisher.test.ts | 3 +- js/net/src/ietf/subscriber.ts | 12 +++- js/net/src/integration.test.ts | 99 ++++++++++++++++++++++++++++++- js/net/src/lite/connection.ts | 7 ++- js/net/src/lite/subscriber.ts | 29 ++++----- js/net/src/mock.ts | 26 ++++++-- quest/m1/README.md | 1 - quest/m1/session-close.md | 1 - quest/m1/session-death-error.md | 55 ----------------- quest/m1/track-tail-interop.md | 5 -- rs/moq-net/tests/support/mock.rs | 9 ++- 14 files changed, 204 insertions(+), 93 deletions(-) delete mode 100644 quest/m1/session-death-error.md diff --git a/js/net/src/error.ts b/js/net/src/error.ts index 912fae64cd..04ad2b84c2 100644 --- a/js/net/src/error.ts +++ b/js/net/src/error.ts @@ -424,6 +424,31 @@ export function fromClose(info: WebTransportCloseInfo): Session | null { return new Session(code, { reason: info.reason }); } +/** + * The session's close as the error it ends everything with, carrying the peer's code. A clean + * close code is still an error here: whatever the close cut off did not end. + * + * @internal + */ +export function closeError(quic: WebTransport): Promise { + return quic.closed.then( + (info) => fromClose(info) ?? new Session(SessionCode.Cancel, { reason: info.reason }), + (err: unknown) => error(err), + ); +} + +/** + * Report a failure the session's close caused as the session's own error, which carries the + * peer's close code; any other failure passes through. + * + * @internal + */ +export async function sessionCause(quic: WebTransport | undefined, err: unknown): Promise { + const source = typeof err === "object" && err !== null ? (err as { source?: unknown }).source : undefined; + if (quic && source === "session") return closeError(quic); + return error(err); +} + /** * Coerce an unknown thrown value into an `Error`. * diff --git a/js/net/src/ietf/adapter.ts b/js/net/src/ietf/adapter.ts index fd16a5349d..b51e48d8da 100644 --- a/js/net/src/ietf/adapter.ts +++ b/js/net/src/ietf/adapter.ts @@ -1,4 +1,5 @@ import { Mutex } from "async-mutex"; +import { error, ProtocolViolation } from "../error.ts"; import { Reader, Stream, type Writer } from "../stream.ts"; import * as Varint from "../varint.ts"; import * as Namespace from "./namespace.ts"; @@ -246,6 +247,8 @@ export class ControlStreamAdapter implements Session { * Must be called after construction. Runs until the control stream closes. */ async run(): Promise { + // Why the virtual streams end: undefined only for a GOAWAY, which is not a failure. + let cause: Error | undefined; try { // v16: also accept real bidi streams (for SubscribeNamespace) if (this.version === Version.DRAFT_16) { @@ -254,7 +257,10 @@ export class ControlStreamAdapter implements Session { for (;;) { const done = await this.#reader.done(); - if (done) break; + if (done) { + cause = new ProtocolViolation("control stream closed"); + break; + } const typeId = await this.#reader.u53(); const size = await this.#reader.u16(); @@ -293,8 +299,11 @@ export class ControlStreamAdapter implements Session { break; } } + } catch (err: unknown) { + cause = error(err); + throw err; } finally { - this.close(); + this.close(cause); } } @@ -719,15 +728,19 @@ export class ControlStreamAdapter implements Session { } } - close() { + /** + * Ends every virtual stream: cleanly for a deliberate close, or with `err` when the + * control stream died under them, since every request riding it was cut off. + */ + close(err?: Error) { if (this.#closed) return; this.#closed = true; console.debug("adapter: close() called"); - // Close all virtual streams for (const entry of this.#streams.values()) { try { - entry.controller.close(); + if (err) entry.controller.error(err); + else entry.controller.close(); } catch { // Already closed } diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index e1d22f9bf8..bcbcf01abc 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -141,7 +141,7 @@ export class Connection implements Established { }); this.#solicit = solicit; this.#cluster = cluster; - this.#subscriber = new Subscriber({ session: this.#session, cluster, hidden }); + this.#subscriber = new Subscriber({ session: this.#session, quic, cluster, hidden }); registerWire(this, { consume: (path) => this.#subscriber.consume(path) }); void this.#run(); diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index a16ce4ccbc..76fca819a2 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -639,8 +639,9 @@ test("closing the session ends the unsolicited announce loop", async () => { // The session ends. The origin is untouched: it is shared, and other sessions keep using it. pair.server.close(); + // Ending with the session's error is ending too: the close fails its open streams. await Promise.race([ - loop, + loop.catch(() => undefined), new Promise((_resolve, reject) => setTimeout(() => reject(new Error("the announce loop outlived its session")), STREAM_WAIT), ), diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index b107b0ca23..138b04e8e2 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -2,7 +2,7 @@ import { race, Signal } from "@moq/signals"; import * as announce from "../announced.ts"; import * as broadcast from "../broadcast.ts"; import { BroadcastCache } from "../consume.ts"; -import { controlTimeout, error, ProtocolViolation, reason } from "../error.ts"; +import { controlTimeout, error, ProtocolViolation, reason, sessionCause } from "../error.ts"; import * as netGroup from "../group.ts"; import { Cost, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts"; import { hiddenBelow, hooks, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; @@ -97,6 +97,10 @@ function sees(filter: Filter, path: Path.Valid): boolean { export class Subscriber { #session: Session; + // The transport, so a request cut off by the session's close ends with the session's + // error. Optional for tests that drive a bare session. + #quic?: WebTransport; + // The Hop IDs this session declared; see {@link Cluster}. What the peer declared is what // says whether an advertisement carries a hop path, and ours is what a path looping back // to us contains. @@ -140,17 +144,21 @@ export class Subscriber { */ constructor({ session, + quic, cluster, hidden = false, }: { /** The session abstraction for bidi streams and request IDs. */ session: Session; + /** The transport the session runs on. */ + quic?: WebTransport; /** The Hop IDs the SETUP exchange settled (MoQ Cluster). */ cluster?: Cluster.Hops; /** Whether the peer understands the HIDDEN parameter (MoQ Hidden). */ hidden?: boolean; }) { this.#session = session; + this.#quic = quic; this.#cluster = cluster; this.#hidden = hidden; } @@ -617,7 +625,7 @@ export class Subscriber { stream.close(); console.debug(`subscribe close: id=${requestId} broadcast=${broadcast} track=${request.name}`); } catch (err) { - const e = error(err); + const e = await sessionCause(this.#quic, err); producer.close(e); stream.abort(e); console.warn( diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index 3107cb2b14..aaed493344 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -9,7 +9,7 @@ import { connect as connectSession, type Established, } from "./connection/index.ts"; -import { StreamCode, StreamError, TooFarBehind } from "./error.ts"; +import { SessionCode, SessionError, StreamCode, StreamError, TooFarBehind } from "./error.ts"; import * as Ietf from "./ietf/index.ts"; import * as Lite from "./lite/index.ts"; import { createMockTransportPair } from "./mock.ts"; @@ -2218,3 +2218,100 @@ test("a handle serves a request under live/** over the wire", async () => { server.close(); origin.close(); }); + +// The peer's close code a killed session carries. +const DEATH = SessionCode(71); + +/** + * Serve one track with a group left open, kill the publisher's session once the subscriber has + * read into it, and return how the subscriber's track ended. + */ +async function runSessionDeath(protocol: string, version?: number): Promise { + const pair = createMockTransportPair(protocol); + const origin = new OriginProducer(); + + const [client, server] = await Promise.all([ + connect(url, { transport: pair.client }), + accept(pair.server, url, { version, publish: origin.consume() }), + ]); + + const broadcast = publish(origin, Path.from("test")); + const serving = (async () => { + for (;;) { + const req = await wireOf(broadcast).requested(); + if (!req) break; + req.accept().appendGroup().writeString("head"); + } + })(); + + const remote = wireOf(client).consume(Path.from("test")); + const track = remote.track("video").subscribe(); + const group = await track.recvGroup(); + expect(await group?.readString()).toBe("head"); + + pair.server.close({ closeCode: DEATH, reason: "killed" }); + const closed = await withTimeout(Promise.resolve(track.closed), 2000, "the track never ended"); + + broadcast.close(); + await serving; + remote.close(); + client.close(); + server.close(); + origin.close(); + return closed; +} + +for (const [name, protocol, version] of [ + ["lite draft-03", Lite.ALPN_03, undefined], + ["lite draft-05", Lite.ALPN_05, undefined], + ["ietf draft-14", "", Ietf.Version.DRAFT_14], + ["ietf draft-17", Ietf.ALPN.DRAFT_17, undefined], +] as const) { + test(`integration: ${name} ends a track with its session's error`, async () => { + const closed = await runSessionDeath(protocol, version); + expect(closed).toBeInstanceOf(SessionError); + expect((closed as SessionError).code).toBe(DEATH); + }); +} + +// On lite-05+ the subscribe stream carries responses until its FIN, so a reset of it is how the +// publisher ends a subscription with an error, and the track ends with that error. +test("integration: lite draft-05 ends a track with the publisher's reset", async () => { + const pair = createMockTransportPair(Lite.ALPN_05); + const origin = new OriginProducer(); + + const [client, server] = await Promise.all([ + connect(url, { transport: pair.client }), + accept(pair.server, url, { publish: origin.consume() }), + ]); + + const broadcast = publish(origin, Path.from("test")); + const served: TrackProducer[] = []; + const serving = (async () => { + for (;;) { + const req = await wireOf(broadcast).requested(); + if (!req) break; + const producer = req.accept(); + producer.appendGroup().writeString("head"); + served.push(producer); + } + })(); + + const remote = wireOf(client).consume(Path.from("test")); + const track = remote.track("video").subscribe(); + const group = await track.recvGroup(); + expect(await group?.readString()).toBe("head"); + + const reset = StreamCode(70); + for (const producer of served) producer.close(new StreamError(reset)); + const closed = await withTimeout(Promise.resolve(track.closed), 2000, "the track never ended"); + expect(closed).toBeInstanceOf(StreamError); + expect((closed as StreamError).code).toBe(reset); + + broadcast.close(); + await serving; + remote.close(); + client.close(); + server.close(); + origin.close(); +}); diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index 75e681d6f8..d5e665fcae 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -3,7 +3,7 @@ import type * as announce from "../announced.ts"; import type { Established } from "../connection/established.ts"; import { type Probe, type Stats, transportStats } from "../connection/stats.ts"; import { type Transport, transportOf } from "../connection/transport.ts"; -import { error, fromClose, StreamCode, StreamError } from "../error.ts"; +import { closeError, error, fromClose, StreamCode, StreamError } from "../error.ts"; import { type Hop, randomHop } from "../hop.ts"; import type { Consumer as OriginConsumer } from "../origin.ts"; import type * as Path from "../path.ts"; @@ -161,11 +161,16 @@ export class Connection implements Established { tasks.push(this.#subscriber.runDatagrams()); } + let fatal: Error | undefined; try { await Promise.all(tasks); } catch (err) { console.error("fatal error running connection", err); + fatal = error(err); } finally { + // The session died under every track it was receiving, so they end with its + // error. A deliberate close() already ended them cleanly, which makes this a no-op. + this.#subscriber.close(fatal ?? (await closeError(this.#quic))); this.close(); } } diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 6b6f97f18b..671809e486 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -3,7 +3,7 @@ import * as announce from "../announced.ts"; import * as broadcast from "../broadcast.ts"; import type { Probe as ProbeStats } from "../connection/stats.ts"; import { BroadcastCache } from "../consume.ts"; -import { controlTimeout, error, ProtocolViolation, reason, StreamCode, StreamError } from "../error.ts"; +import { controlTimeout, error, ProtocolViolation, reason, StreamCode, StreamError, sessionCause } from "../error.ts"; import * as netGroup from "../group.ts"; import { Cost, type Hop, MAX_HOPS, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts"; import { groupBounds, hiddenBelow, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; @@ -573,11 +573,14 @@ export class Subscriber { // // On lite-05+ the publisher sends SUBSCRIBE_START/END/DROP on this stream until // its FIN; older drafts just close it. Either way group streams can still be in - // flight, so the track ends only once the tail is accounted for. + // flight, so the track ends only once the tail is accounted for. A reset rejects + // instead, so the track ends with that error rather than a clean tail. const responses = supportsTrackStream(this.version) ? this.#runResponses(stream, entry) : stream.reader.closed; const closed = responses.then(() => this.#settleTail(entry)); + // A reset that lands after the race below settled is moot; the race observes one before. + closed.catch(() => {}); const subscriptionUpdates = this.version === Version.DRAFT_01 || this.version === Version.DRAFT_02 ? undefined @@ -605,7 +608,7 @@ export class Subscriber { stream.close(); console.debug(`subscribe close: id=${id} broadcast=${broadcast} track=${request.name}`); } catch (err) { - const e = error(err); + const e = await sessionCause(this.#quic, err); producer.close(e); console.warn(`subscribe error: id=${id} broadcast=${broadcast} track=${request.name} error=${reason(e)}`); stream.abort(e); @@ -809,17 +812,11 @@ export class Subscriber { // Reads SUBSCRIBE_START/END/DROP on the subscribe stream until FIN (lite-05+), recording // the range the tail is accounted against. SUBSCRIBE_END declares the track's end right - // away, so a consumer learns it before the last groups arrive. Resolves on FIN or on the - // stream being reset out from under it; rejects only on a response that breaks the range. + // away, so a consumer learns it before the last groups arrive. Resolves on FIN and rejects + // when the stream is reset, so the track ends with the publisher's error rather than cleanly. async #runResponses(stream: Stream, entry: SubscribeEntry): Promise { for (;;) { - let resp: Awaited>; - try { - resp = await decodeSubscribeResponseMaybe(stream.reader, this.version); - } catch { - // Stream closed or reset; nothing more to read. - return; - } + const resp = await decodeSubscribeResponseMaybe(stream.reader, this.version); if (!resp) return; if ("start" in resp) { @@ -1140,11 +1137,15 @@ export class Subscriber { } } - close() { + /** + * Ends every subscribed track: cleanly for a deliberate close, or with `err` when the + * session died, since those tracks were cut off rather than ended. + */ + close(err?: Error) { this.#closed.abort(); for (const { track } of this.#subscribes.values()) { - track.close(); + track.close(err); } this.#subscribes.clear(); diff --git a/js/net/src/mock.ts b/js/net/src/mock.ts index 7bfdc9035f..78fdd6d191 100644 --- a/js/net/src/mock.ts +++ b/js/net/src/mock.ts @@ -12,9 +12,15 @@ import type { SendStream } from "./stream.ts"; const WRITABLE_STRATEGY: QueuingStrategy = { highWaterMark: 256 }; const READABLE_STRATEGY: QueuingStrategy = { highWaterMark: 256 }; -function newStream(): TransformStream { +/** Every stream of a transport pair, so a session close can fail the ones still open. */ +type StreamSet = Set>; + +function newStream(streams: StreamSet): TransformStream { return new TransformStream( { + start(controller) { + streams.add(controller); + }, // Copy each chunk to simulate real WebTransport's kernel-boundary copy. // Without this, Writer's scratch buffer reuse corrupts queued data. transform(chunk, controller) { @@ -57,6 +63,10 @@ export class MockTransport implements WebTransport { // Reference to the peer so we can enqueue streams to them #peer?: MockTransport; + // Shared with the peer: a session close fails every stream either side opened, the way + // a real WebTransport session does. + #streams: StreamSet = new Set(); + constructor( protocol: string, datagramsEnabled = true, @@ -149,6 +159,7 @@ export class MockTransport implements WebTransport { setPeer(peer: MockTransport) { this.#peer = peer; + this.#streams = peer.#streams; } async createBidirectionalStream(options?: WebTransportSendStreamOptions): Promise { @@ -156,8 +167,8 @@ export class MockTransport implements WebTransport { if (!peer) throw new Error("no peer"); // Create two TransformStreams for the two directions - const c2s = newStream(); - const s2c = newStream(); + const c2s = newStream(this.#streams); + const s2c = newStream(this.#streams); // Local side: writes to c2s, reads from s2c const local = { @@ -184,7 +195,7 @@ export class MockTransport implements WebTransport { const peer = this.#peer; if (!peer) throw new Error("no peer"); - const c2s = newStream(); + const c2s = newStream(this.#streams); // Record before handing the peer its end, so a peer that waits for the stream to arrive // always finds it here. @@ -216,6 +227,13 @@ export class MockTransport implements WebTransport { const info = _closeInfo ?? { closeCode: 0, reason: "" }; this.#closeResolve(info); + // Shaped like the WebTransportError a real session close fails its streams with. + const closed = Object.assign(new Error("session closed"), { source: "session", streamErrorCode: null }); + for (const stream of this.#streams) { + stream.error(closed); + } + this.#streams.clear(); + try { this.#bidiController.close(); } catch { diff --git a/quest/m1/README.md b/quest/m1/README.md index 6c95748db6..bb5a6caa28 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -21,7 +21,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [libmoq hidden opt-in](/quest/m1/libmoq-hidden.md) - `moq_origin_announced` takes a `hidden` flag so C callers can list `.`-named broadcasts - [Announce compression](/quest/m1/announce-compression.md) - a lite-07 announce reuses the path head and hop-chain tail of a live announcement on its stream instead of resending them - [lite-07 count settle](/quest/m1/lite-count-settle.md) - moq-lite-07 subscribers stop waiting for a subscription's tail once SUBSCRIBE_END's stream count is reached -- [Session death error](/quest/m1/session-death-error.md) - a dying session ends its tracks with its own error in Rust and JS, never a clean end, `Dropped`, or `Cancel` - [Dropped sources](/quest/m1/dropped-sources.md) - consumers see the producer's real error on every end path, never `Dropped` - [JS bare FIN](/quest/m1/js-bare-fin.md) - a `@moq/net` subscriber aborts a track whose subscribe stream FINs before its declared end, like Rust - [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop` diff --git a/quest/m1/session-close.md b/quest/m1/session-close.md index a846d5e977..3a297810e6 100644 --- a/quest/m1/session-close.md +++ b/quest/m1/session-close.md @@ -25,5 +25,4 @@ abort does not. No new page. ## Related -- [Session death error](/quest/m1/session-death-error.md) - a session that dies still ends its tracks with its own error - [Broadcast close](/quest/m1/broadcast-close/README.md) - `close()` on a broadcast, which is not this session end diff --git a/quest/m1/session-death-error.md b/quest/m1/session-death-error.md deleted file mode 100644 index b4112d5c24..0000000000 --- a/quest/m1/session-death-error.md +++ /dev/null @@ -1,55 +0,0 @@ -# [M] Session death error - -## Goal - -When a session dies, every track it was receiving ends with the session's own -error, in Rust and JS, over moq-lite and IETF. A reader never sees a clean end -for a stream that was cut off, and never a generic `Dropped` or `Cancel` -standing in for the real cause. A publisher's reset of a subscription reaches -the reader as that reset's error. - -## Plan - -Today the cause is lost in several places. Rust: - -- The lite subscriber's serve loop gives back `Error::Dropped` when the - session dies, and `SubscriptionCleanup` aborts the remaining tracks with - `Error::Cancel` (`rs/moq-net/src/lite/subscriber.rs`). `Dropped` here is a - bug. -- The IETF subscriber's `State::drop` aborts every subscription with - `Error::Cancel` (`rs/moq-net/src/ietf/subscriber.rs`). -- Worse, the reader can see a clean end. Once SUBSCRIBE_END has declared - the boundary, a session that dies with a group below it still in flight - leaves `recv_group()` returning `Ok(None)`, skipping that group (#4061). - The abort does reach the track, but the clean-end check wins whenever the - reader polls after the close lands, so the outcome depends on who reads - first. A declared end is only clean once every group below it is - accounted for; an abort before then wins. - -JS: - -- The lite `Subscriber.close()` runs from `Connection.close()` even after a - fatal error and closes every track cleanly (`js/net/src/lite/subscriber.ts`, - `js/net/src/lite/connection.ts`). -- On lite-05 and later, `#drainResponses` swallows a reset of the subscribe - stream and reports it as a clean end. On lite-01 to -04 the error does reach - the track. -- On IETF drafts 14-16, where one control stream carries every request, the - adapter closes every per-request stream cleanly when the session or control - stream dies (`js/net/src/ietf/adapter.ts`), so the tracks end cleanly. Drafts - 17 and later reject with the transport error. - -Thread the session's close error to wherever tracks are torn down, and abort -with it. A deliberate local close of the session is still a close, not an -error. - -Test by killing a session mid-track in both languages and asserting the reader -sees the session's error, and by resetting a subscribe stream on lite-05 and -later. For #4061, pin the deterministic mock repro and the real-QUIC one from -`kidq330/bug/subscription_ends_clean_with_missing_group` -(`subscription_end_integrity` in `moq-net` and `moq-tokio`), with their -controls: a finished track still ends clean while its session lives. - -## Closes - -- [#4061](https://github.com/moq-dev/moq/issues/4061) - close this issue when the quest finishes diff --git a/quest/m1/track-tail-interop.md b/quest/m1/track-tail-interop.md index 0078cdae17..7489cbdfb3 100644 --- a/quest/m1/track-tail-interop.md +++ b/quest/m1/track-tail-interop.md @@ -27,8 +27,3 @@ What stood in the way when the Rust half landed: QUIC on localhost rarely reorders, so this is a smoke check that the end is delivered and clean. The ordering race itself stays in the unit tests. - -## Related - -- [Session death error](/quest/m1/session-death-error.md) - the other way a - track ends wrong diff --git a/rs/moq-net/tests/support/mock.rs b/rs/moq-net/tests/support/mock.rs index 1b5e052126..7767f827bd 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -132,7 +132,10 @@ impl poll::SendStream for MockSendStream { type Error = MockError; fn poll_write(&mut self, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { - Poll::Ready(self.push(StreamChunk::Data(Bytes::copy_from_slice(buf))).map(|()| buf.len())) + Poll::Ready( + self.push(StreamChunk::Data(Bytes::copy_from_slice(buf))) + .map(|()| buf.len()), + ) } fn set_priority(&mut self, _order: u8) {} @@ -334,7 +337,9 @@ impl ConnectionState { /// The close every stream and accept fails with, once the connection closed. fn error(&self) -> Option { let state = self.close_state.lock().unwrap(); - state.as_ref().map(|(code, reason)| MockError::session(*code, reason.clone())) + state + .as_ref() + .map(|(code, reason)| MockError::session(*code, reason.clone())) } } From bab53cab37a033eac3be9f0c6027370ceff6b295 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 14:29:36 -0700 Subject: [PATCH 05/11] test(net): cover session death on moq-lite-07-wip #4148 renamed that ALPN and took it off the default set. The session-death cases still run on that draft. Co-Authored-By: Grok 4.7 --- rs/moq-net/tests/subscription_end_integrity.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-net/tests/subscription_end_integrity.rs b/rs/moq-net/tests/subscription_end_integrity.rs index 4a026897a3..fa3ca2424d 100644 --- a/rs/moq-net/tests/subscription_end_integrity.rs +++ b/rs/moq-net/tests/subscription_end_integrity.rs @@ -188,7 +188,7 @@ async fn a_finished_track_ends_clean_while_its_session_lives() { const DEATH_VERSIONS: &[&str] = &[ "moq-lite-03", "moq-lite-05", - "moq-lite-07", + "moq-lite-07-wip", "moq-transport-14", "moq-transport-17", "moq-transport-22", From a3dc504bb3636d5669458fef851724afc0623ece Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 14:38:29 -0700 Subject: [PATCH 06/11] fix(net): report a session close during setup as the session error A fatal lite task and a subscribe that dies before it is accepted were still the raw transport error. sessionCause already turns a session-sourced failure into the peer's close. Co-Authored-By: Grok 4.7 --- js/net/src/ietf/subscriber.ts | 2 +- js/net/src/lite/connection.ts | 5 +++-- js/net/src/lite/subscriber.ts | 2 +- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 138b04e8e2..ba24feeb59 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -541,7 +541,7 @@ export class Subscriber { console.debug(`subscribe ok: id=${requestId} broadcast=${broadcast} track=${request.name}`); } catch (err) { // A control request that timed out is not late content, so it carries its own code. - const e = err instanceof TimeoutError ? controlTimeout(err) : error(err); + const e = err instanceof TimeoutError ? controlTimeout(err) : await sessionCause(this.#quic, err); request.reject(e); console.warn( `subscribe error: id=${requestId} broadcast=${broadcast} track=${request.name} error=${reason(e)}`, diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index d5e665fcae..4aec63c3a3 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -3,7 +3,7 @@ import type * as announce from "../announced.ts"; import type { Established } from "../connection/established.ts"; import { type Probe, type Stats, transportStats } from "../connection/stats.ts"; import { type Transport, transportOf } from "../connection/transport.ts"; -import { closeError, error, fromClose, StreamCode, StreamError } from "../error.ts"; +import { closeError, error, fromClose, StreamCode, StreamError, sessionCause } from "../error.ts"; import { type Hop, randomHop } from "../hop.ts"; import type { Consumer as OriginConsumer } from "../origin.ts"; import type * as Path from "../path.ts"; @@ -166,7 +166,8 @@ export class Connection implements Established { await Promise.all(tasks); } catch (err) { console.error("fatal error running connection", err); - fatal = error(err); + // A session-sourced failure is the peer's close, not the raw transport error. + fatal = await sessionCause(this.#quic, err); } finally { // The session died under every track it was receiving, so they end with its // error. A deliberate close() already ended them cleanly, which makes this a no-op. diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 671809e486..bf38094d0c 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -550,7 +550,7 @@ export class Subscriber { } catch (err) { // The setup outlived its deadline waiting for the first response: a control // timeout, not content that arrived late. - const e = err instanceof TimeoutError ? controlTimeout(err) : error(err); + const e = err instanceof TimeoutError ? controlTimeout(err) : await sessionCause(this.#quic, err); request.reject(e); this.#subscribes.delete(id); console.warn(`subscribe error: id=${id} broadcast=${broadcast} track=${request.name} error=${reason(e)}`); From cfd3c8f4ff583acfc246bbde02997f4751388262 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 14:47:10 -0700 Subject: [PATCH 07/11] fix(net): report a closed IETF session when a request id never arrives nextRequestId resolves undefined once the adapter is gone, before sessionCause runs. If the transport has already closed, that close is the rejection. A still-open transport is not waited out. Co-Authored-By: Grok 4.7 --- js/net/src/ietf/subscriber.ts | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index ba24feeb59..0294b35e3d 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -2,7 +2,7 @@ import { race, Signal } from "@moq/signals"; import * as announce from "../announced.ts"; import * as broadcast from "../broadcast.ts"; import { BroadcastCache } from "../consume.ts"; -import { controlTimeout, error, ProtocolViolation, reason, sessionCause } from "../error.ts"; +import { closeError, controlTimeout, error, ProtocolViolation, reason, sessionCause } from "../error.ts"; import * as netGroup from "../group.ts"; import { Cost, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts"; import { hiddenBelow, hooks, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; @@ -487,10 +487,22 @@ export class Subscriber { return consumer; } + // The adapter is gone. If the transport has already closed, that close is the + // error. A still-open transport, such as a GOAWAY drain, has no peer code yet, + // so this does not wait for it. + async #closedSession(): Promise { + const quic = this.#quic; + if (!quic) return new Error("session closed"); + return Promise.race([ + closeError(quic), + new Promise((resolve) => queueMicrotask(() => resolve(new Error("session closed")))), + ]); + } + async #runSubscribe(broadcast: Path.Valid, request: track.Request) { const requestId = await this.#session.nextRequestId(); if (requestId === undefined) { - request.reject(new Error("session closed")); + request.reject(await this.#closedSession()); return; } From 305af76a213ede8ed59135f676ffc9fbc6ff5370 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 14:50:25 -0700 Subject: [PATCH 08/11] test(net): ignore the mock pair's held transports #4116 added those fields. The session-death tests only drive the sessions. Co-Authored-By: Grok 4.7 --- rs/moq-net/tests/subscription_end_integrity.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/tests/subscription_end_integrity.rs b/rs/moq-net/tests/subscription_end_integrity.rs index fa3ca2424d..281863f6ba 100644 --- a/rs/moq-net/tests/subscription_end_integrity.rs +++ b/rs/moq-net/tests/subscription_end_integrity.rs @@ -57,7 +57,7 @@ async fn round(drop_session: bool) -> (Vec>, Option) { let mut options = MockConnectOptions::new("moq-lite-05".parse::().unwrap()); options.server_publish = Some(publisher.clone()); options.client_subscribe = Some(subscriber.clone()); - let MockPair { client, server } = connect_mock(options).await; + let MockPair { client, server, .. } = connect_mock(options).await; let consumer = subscriber.consume(); tokio::time::timeout(TIMEOUT, consumer.routed("bcast")) @@ -209,7 +209,7 @@ async fn killed(version: &str) -> Option { let mut options = MockConnectOptions::new(version.parse::().unwrap()); options.server_publish = Some(publisher.clone()); options.client_subscribe = Some(subscriber.clone()); - let MockPair { client, server } = connect_mock(options).await; + let MockPair { client, server, .. } = connect_mock(options).await; let consumer = subscriber.consume(); consumer.routed("bcast").await.expect("routed"); From d26bb24d8083190e71615057c2f5fe3f7a7a8c54 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 16:05:45 -0700 Subject: [PATCH 09/11] fix(net): surface a dead final segment on datagram reads A finished resume producer ended datagram reads cleanly because that poll drops Ready(Err). Ask the final segment, the same way groups do. Acknowledge a held mock FIN only when it was queued, so poll_closed still reports the connection error. Co-Authored-By: Grok 4.7 --- rs/moq-net/src/model/resume.rs | 39 +++++++++++++++++++++++++++++++- rs/moq-net/tests/support/mock.rs | 7 ++++-- 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index cbce717f3d..0d9b8ba038 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -2205,8 +2205,18 @@ impl Subscriber { if let Some(err) = &self.abort { return Poll::Ready(Err(err.clone())); } + // The producer finished, so this channel ends now. A final segment that + // already died is that death: the datagram read above drops a Ready(Err) + // on the floor, and a clean Ok(None) here would hide it. A segment that + // has not activated yet has no end to report. if self.finished { - return Poll::Ready(Ok(None)); + if pending_activation { + return Poll::Ready(Ok(None)); + } + return match ready!(self.poll_final(waiter)) { + Some(end) => Poll::Ready(end.map(|_| None)), + None => Poll::Ready(Ok(None)), + }; } // The producer is gone: no takeover is coming, so the track ends when // its newest segment does, and the way it did. @@ -4915,6 +4925,33 @@ mod test { assert!(matches!(result, Err(Error::Timeout))); } + /// A datagram-only reader of a finished logical track ends with the final + /// segment's error. The datagram poll drops a segment error that is not + /// `Ok(Some)`, so the finished path has to ask the segment itself. + #[tokio::test] + async fn finished_producer_ends_datagrams_with_a_dead_final_segment() { + let (track_a, consumer_a) = track_pair("a"); + + let mut producer = Producer::new(); + producer.switch(&consumer_a, None).unwrap(); + let mut sub = producer.consume().subscribe(None); + + assert!( + kio::wait(|waiter| sub.poll_recv_datagram(waiter)) + .now_or_never() + .is_none(), + "no datagram yet" + ); + + producer.finish().unwrap(); + track_a.abort(Error::Timeout).unwrap(); + + let result = kio::wait(|waiter| sub.poll_recv_datagram(waiter)) + .now_or_never() + .expect("must not stall forever"); + assert!(matches!(result, Err(Error::Timeout))); + } + #[tokio::test] async fn dropped_producer_keeps_a_live_segment_serving() { let (mut track_a, consumer_a) = track_pair("a"); diff --git a/rs/moq-net/tests/support/mock.rs b/rs/moq-net/tests/support/mock.rs index 7767f827bd..7043f020d2 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -142,11 +142,14 @@ impl poll::SendStream for MockSendStream { fn finish(&mut self) -> Result<(), Self::Error> { if self.tx.is_some() { - let _ = self.push(StreamChunk::Fin); - if self.ack_fin { + // A FIN that never left must not look acknowledged: poll_closed + // trusts this signal ahead of the connection error. + let pushed = self.push(StreamChunk::Fin); + if pushed.is_ok() && self.ack_fin { self.closed.set(Ok(())); } self.tx = None; + pushed?; } Ok(()) } From df2c4f459561e31cdddf6d681572bf6704eb8fd4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 17:50:17 -0700 Subject: [PATCH 10/11] fix(net): reject a subscribe still setting up with the session error A lite track waiting on TRACK_INFO is not in the subscribe map, and an IETF subscribe has no producer until SUBSCRIBE_OK. Dropping either was Dropped. Record the session error and reject the parked request with it. Co-Authored-By: Grok 4.7 --- rs/moq-net/src/ietf/subscriber.rs | 113 +++++++++++++++++++++++++++--- rs/moq-net/src/lite/subscriber.rs | 89 +++++++++++++++++++++++ 2 files changed, 193 insertions(+), 9 deletions(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 1394371e74..e01227f073 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -158,7 +158,10 @@ impl State { /// Active receive tasks abort their own groups. Abort any head waiting for its /// tail here, along with the track. Ordinary unsubscribe removes its entry. fn abort(&mut self, err: &Error) { - for (_, track) in self.subscribes.drain() { + for (_, mut track) in self.subscribes.drain() { + if let Some(request) = track.pending.take() { + request.reject(err.clone()); + } if let Fill::Ready { producer, .. } = &*track.fill.read() { let _ = producer.clone().abort(err.clone()); } @@ -308,6 +311,9 @@ struct Accepted { struct TrackState { producer: Option, + /// The origin request, until SUBSCRIBE_OK accepts it. Abort rejects this: the + /// producer does not exist yet, and dropping the setup task would be `Dropped`. + pending: Option, name: String, alias: Option, @@ -354,6 +360,7 @@ impl TrackState { fn pending(name: String, broadcast: PathOwned, fill: kio::Producer, joining: Option) -> Self { Self { producer: None, + pending: None, name, alias: None, broadcast, @@ -656,6 +663,11 @@ where held.broadcast == new.broadcast && held.name == new.name } + /// Take the origin request back out of a subscription that is still setting up. + fn take_pending(&self, request_id: RequestId) -> Option { + self.state.lock().subscribes.get_mut(&request_id)?.pending.take() + } + fn remove_subscribe(&self, request_id: RequestId) -> Option { let mut state = self.state.lock(); let track = state.subscribes.remove(&request_id)?; @@ -1657,6 +1669,19 @@ where tracing::info!(broadcast = %self.origin.absolute(&broadcast_path), track = %request.name(), "subscribe started"); + // Park the origin request where a session abort can reject it. The producer + // does not exist until SUBSCRIBE_OK, and dropping this task would otherwise + // end the track as `Dropped`. + let track_name = request.name().to_owned(); + { + let mut state = self.state.lock(); + let Some(held) = state.subscribes.get_mut(&request_id) else { + request.reject(Error::Cancel); + return; + }; + held.pending = Some(request); + } + // A publisher can be serving before its SUBSCRIBE_OK reaches us, since the data // streams are independent of the request stream. Waiting for the response alone would // miss the local side going away in that window and leave the publisher serving a @@ -1666,9 +1691,10 @@ where enum Setup { Response(Result, Error>), Unused, + /// `abort` already rejected the parked request. + Gone, } - let track_name = request.name().to_owned(); let setup = { let mut response = std::pin::pin!(self.read_subscribe_response(&mut stream)); loop { @@ -1680,7 +1706,15 @@ where if let Poll::Ready(res) = waiter.poll_future(response.as_mut()) { return Poll::Ready(Setup::Response(res)); } - if request.poll_unused(waiter).is_ready() { + let mut state = self.state.lock(); + let Some(pending) = state + .subscribes + .get_mut(&request_id) + .and_then(|held| held.pending.as_mut()) + else { + return Poll::Ready(Setup::Gone); + }; + if pending.poll_unused(waiter).is_ready() { return Poll::Ready(Setup::Unused); } Poll::Pending @@ -1688,23 +1722,36 @@ where .await; match setup { - Setup::Response(res) => break Some((res, request)), + Setup::Response(res) => break Some(res), + Setup::Gone => break None, Setup::Unused => { - if request.reject_unused(Error::Cancel) { + let mut state = self.state.lock(); + let Some(pending) = state + .subscribes + .get_mut(&request_id) + .and_then(|held| held.pending.take()) + else { + break None; + }; + if pending.reject_unused(Error::Cancel) { break None; } + if let Some(held) = state.subscribes.get_mut(&request_id) { + held.pending = Some(pending); + } } } } }; - let Some((response, request)) = setup else { + let Some(response) = setup else { tracing::info!( broadcast = %self.origin.absolute(&broadcast_path), track = %track_name, "subscribe abandoned before it was accepted" ); - // The publisher may already be serving before it answers. + // The publisher may already be serving before it answers. A session abort + // already rejected the parked request; dropping what remains is not a second one. self.remove_subscribe(request_id); self.cancel_subscribe(stream, request_id).await; return; @@ -1715,14 +1762,18 @@ where let accepted = match response { Ok(Some(accepted)) => accepted, Ok(None) => { + if let Some(pending) = self.take_pending(request_id) { + pending.reject(Error::UnexpectedMessage); + } self.remove_subscribe(request_id); - request.reject(Error::UnexpectedMessage); return; } Err(err) => { tracing::debug!(%err, "subscribe response error"); + if let Some(pending) = self.take_pending(request_id) { + pending.reject(err); + } self.remove_subscribe(request_id); - request.reject(err); return; } }; @@ -1738,6 +1789,11 @@ where .with_priority(super::priority::from_wire(priority.unwrap_or(128))); // Declared before the track is released to readers, so a warm cache waiting on // this copy judges itself against where the live feed actually starts. + let Some(request) = self.take_pending(request_id) else { + // Aborted while the answer was in hand. The parked request is already rejected. + self.remove_subscribe(request_id); + return; + }; let request = match live { true => request.resolving_start(), false => request, @@ -3006,6 +3062,45 @@ mod tests { assert_eq!(pending.await.unwrap(), RequestId(11)); } + /// SUBSCRIBE_OK has not accepted the track, so the map holds no producer. + /// Abort still has to reject the parked origin request with the session error. + #[tokio::test] + async fn session_death_rejects_a_subscribe_still_setting_up() { + let broadcast = crate::broadcast::Info::new().produce(); + let mut dynamic = broadcast.dynamic(); + let consumer = broadcast.consume(); + let mut waiting = std::pin::pin!(consumer.track("video").unwrap().subscribe(None)); + assert!(poll!(&mut waiting).is_pending()); + + let mut requested = std::pin::pin!(dynamic.requested_track()); + let std::task::Poll::Ready(Ok(request)) = poll!(&mut requested) else { + panic!("the subscribe did not request a track"); + }; + + let mut state = State::default(); + state.subscribes.insert( + RequestId(1), + TrackState { + pending: Some(request), + ..TrackState::pending( + "video".to_string(), + crate::Path::new("bcast").to_owned(), + kio::Producer::new(Fill::Done), + None, + ) + }, + ); + state.abort(&Error::Session(crate::SessionError::App(7))); + + assert!( + matches!( + poll!(&mut waiting), + std::task::Poll::Ready(Err(Error::Session(crate::SessionError::App(7)))) + ), + "setup was not rejected with the session error" + ); + } + #[tokio::test(start_paused = true)] async fn unknown_track_alias_times_out() { let aliases = TrackAliases::default(); diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index fd4b070e3e..bb5731ab6e 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -70,6 +70,10 @@ pub(super) struct Subscriber { // assigned id stays local and is never written into a hop chain. session_origin: crate::Hop, subscribes: Lock>, + /// Why this session ended, once it has. A track still waiting on TRACK_INFO is + /// not in [`Self::subscribes`], so dropping its request reads this instead of + /// becoming [`Error::Dropped`]. + ended: Lock>, next_id: Arc, version: Version, /// The peer's advertised SETUP (lite-05+), set when its Setup stream is read. @@ -107,6 +111,7 @@ impl Subscriber { self_origin, session_origin: config.peer_hop.unwrap_or(crate::Hop::UNKNOWN), subscribes: Default::default(), + ended: Default::default(), next_id: Default::default(), version: config.version, peer_setup: config.peer_setup, @@ -116,6 +121,20 @@ impl Subscriber { } } + /// Record the error that ended the session. The first one wins: a later cancel + /// from dropping the driver must not replace it. + fn note_end(&self, err: &Error) { + let mut slot = self.ended.lock(); + if slot.is_none() { + *slot = Some(err.clone()); + } + } + + /// The recorded session end, or [`Error::Dropped`] when nothing recorded one. + fn end_reason(&self) -> Error { + self.ended.lock().clone().unwrap_or(Error::Dropped) + } + /// Reject a new request once the peer has sent a GOAWAY: it told us to stop /// opening streams on this session (existing subscriptions keep flowing). fn check_going_away(&self) -> Result<(), Error> { @@ -518,6 +537,9 @@ impl SubscriberDriver { /// End every active subscription with the error that ended the session. pub fn abort(&self, err: &Error) { + // Before the subscribe map, so a TRACK_INFO request dropped with this driver + // rejects with `err` rather than `Dropped`. + self.subscriber.note_end(err); self.cleanup.abort(err); } @@ -562,6 +584,14 @@ impl SubscriberDriver { } } +impl Drop for SubscriberDriver { + fn drop(&mut self) { + // `abort` already recorded a session error when the driver failed. A driver + // dropped without that still has to name an end before its setup machines drop. + self.subscriber.note_end(&Error::Cancel); + } +} + /// Accepts incoming uni streams (GROUP data plus the peer's SETUP) and drives /// each as a child machine. Resolves only on a transport error. struct UniAccept { @@ -1429,6 +1459,51 @@ mod tests { ); } + /// A lite-05 subscribe still waiting on TRACK_INFO is not in the subscribe map. + /// Dropping that machine must reject the origin request with the session's error. + #[tokio::test] + async fn session_death_rejects_a_track_waiting_for_info() { + let subscriber = Subscriber::new(SubscriberConfig { + runtime: crate::time::Clock::tokio(), + session: SinkSession::default(), + origin: origin::Config::new(crate::Hop::new(1).unwrap()).produce(), + recv_bandwidth: None, + version: VERSION, + peer_setup: Default::default(), + peer_hop: None, + cost: None, + going_away: Default::default(), + }); + + let broadcast = crate::broadcast::Info::new().produce(); + let mut dynamic = broadcast.dynamic(); + let consumer = broadcast.consume(); + let mut waiting = std::pin::pin!(consumer.track("video").unwrap().subscribe(None)); + assert!( + futures::poll!(waiting.as_mut()).is_pending(), + "waiting on the publisher" + ); + let request = dynamic.requested_track().now_or_never().unwrap().expect("request"); + + let run = TrackServeRun::new( + TrackServe { + subscriber: subscriber.clone(), + path: Path::new("room").to_owned(), + name: "video".to_string(), + }, + request, + ); + let death = Error::Session(crate::SessionError::App(7)); + subscriber.note_end(&death); + drop(run); + + let ended = waiting.now_or_never().expect("subscribe must end"); + assert!( + matches!(ended, Err(Error::Session(crate::SessionError::App(7)))), + "waiting track did not end with the session error" + ); + } + /// `establish` puts exactly one SUBSCRIBE on the wire, and the id is registered /// before any of it reaches the transport. /// @@ -3146,6 +3221,20 @@ impl kio::Task for TrackServeRun { } } +impl Drop for TrackServeRun { + fn drop(&mut self) { + // Still waiting on TRACK_INFO: the request never reached the subscribe map, + // so the driver's abort cannot see it. Reject with the session's error. + let TrackRunState::Info { request, .. } = &mut self.state else { + return; + }; + let Some(request) = request.take() else { + return; + }; + request.reject(self.serve.subscriber.end_reason()); + } +} + /// Opens a TRACK stream, reads the single TRACK_INFO, and maps it to the /// model's [`track::Info`]. Lite05+ only. Bails if the session dies meanwhile. struct TrackInfoFetch { From c7857113ef21db46a3da3fcc2c12d2bd7daf48cc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 17:53:01 -0700 Subject: [PATCH 11/11] test(net): publish the session-death harness through a consumer Main now hands the mock session a origin consumer, not the producer. Co-Authored-By: Grok 4.7 --- rs/moq-net/tests/subscription_end_integrity.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/tests/subscription_end_integrity.rs b/rs/moq-net/tests/subscription_end_integrity.rs index 281863f6ba..11ade6b507 100644 --- a/rs/moq-net/tests/subscription_end_integrity.rs +++ b/rs/moq-net/tests/subscription_end_integrity.rs @@ -55,7 +55,7 @@ async fn round(drop_session: bool) -> (Vec>, Option) { let subscriber = produce_origin(2); let mut options = MockConnectOptions::new("moq-lite-05".parse::().unwrap()); - options.server_publish = Some(publisher.clone()); + options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(subscriber.clone()); let MockPair { client, server, .. } = connect_mock(options).await; @@ -207,7 +207,7 @@ async fn killed(version: &str) -> Option { let subscriber = produce_origin(2); let mut options = MockConnectOptions::new(version.parse::().unwrap()); - options.server_publish = Some(publisher.clone()); + options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(subscriber.clone()); let MockPair { client, server, .. } = connect_mock(options).await;