diff --git a/quest/m1/README.md b/quest/m1/README.md index ffade19231..720ef863cb 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -94,7 +94,6 @@ lands before the cut. - [JS request owner](/quest/m1/js-request-owner.md) - every JS IETF request stream has one owning handle with an `ended` promise and one closer - [Container reads drop backfill](/quest/m1/read-backfill.md) - the JS container consumer drops groups below its read cursor like Rust - [Rust request credit](/quest/m1/rs-request-credit.md) - a moq-net request fails at once without stream credit instead of waiting -- [A stalled subscriber holds bounded state](/quest/m1/egress-held-state.md) - group machines and bytes a credit-withholding downstream makes a publisher hold plateau at what its credit and the latency budget allow, on lite and moq-transport - [qmux no-wait opens](/quest/m1/qmux-no-wait.md) - the in-tree @moq/qmux rejects an over-limit create when waitUntilAvailable is false, like Chrome - [Timed-out requests clean up](/quest/m1/legacy-request-timeout-cleanup.md) - on drafts 14-16, Rust and JS send PUBLISH_NAMESPACE_DONE or UNSUBSCRIBE_NAMESPACE for a timed-out request before its local reset - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide diff --git a/quest/m1/egress-held-state.md b/quest/m1/egress-held-state.md deleted file mode 100644 index d8ec6f6539..0000000000 --- a/quest/m1/egress-held-state.md +++ /dev/null @@ -1,119 +0,0 @@ -# [L] A stalled subscriber holds bounded publisher state - -## Goal - -What a downstream peer can make a publisher hold is bounded by the protocol's -own shape, not by how long it stalls. A subscriber that withholds stream -credit, or drains slower than the media, across as many tracks as its session -limits admit, makes a relay (or any moq-net publisher) hold group state and -bytes that plateau at what the peer's credit and the latency budget allow, -on both moq-lite and moq-transport. A healthy viewer behind a congested link -still gets the freshest groups. - -Non-goals: capping sessions or connections (see Plan), ingress memory -([Bound incomplete messages](/quest/m1/quic/receive-memory.md) owns it), and -new limit constants that only refuse input. - -## Plan - -Planned 2026-10-19 after the egress audit in #5958, under the rule that -network input is untrusted and its cost is bounded by design (#5961). Facts -as of 83805955a; re-verify, they move. - -Where it stands: - -- Each group a subscription reaches becomes its own serve machine at once, - and waits in `Open` for stream credit with no in-flight bound: lite - `TrackRun::serve` (`rs/moq-net/src/lite/publisher.rs`, around 3722), IETF - `TrackServe` ("one machine per group, with unlimited concurrency", - `rs/moq-net/src/ietf/publisher.rs`, around 4002). -- A parked machine leaves only when its group expires. An aborted group never - does: `expiry_pending` is false once aborted (`model/group.rs`, around 1924 - and 2174), and the `Open` arms check only expiry. The pool's idle window - (30 s by default) evicts the unread group's bytes, so the machine outlives - its content and parks forever. When credit finally arrives, it may be spent - on a group that can no longer be served. -- The serving budget is effectively unbounded. A peer whose wire carries no - max delay (lite-01/02, every moq-transport draft) is served at about 2^53 ms - (`MAX_SAFE_DELAY_MS` in both publishers), and a lite-03+ peer names any - value it likes. Locally only the origin's `cache_duration` (`Duration::MAX` - unless the relay sets `--cache-duration`) or the publisher's `max_age` - clamps it. -- Session limits cap subscriptions and fetches (`rs/moq-relay/src/limits.rs`: - 1000 subscriptions and 10000 fetches per client session), never groups in - flight per subscription. - -Abuse scenarios to cover: - -1. A subscriber at its subscription cap never grants uni-stream credit. Held - machines grow with every group published, forever, even after the pool - frees their bytes. -2. The same peer with an inert pool (no capacity, no idle window, as a library - user may configure). The pool keeps every group by policy, so judge only - what the subscriber adds: held state beyond a no-subscriber baseline. -3. A peer grants stream credit but withholds byte credit, so opened streams - pin their groups past the point they are useful. -4. A lite-03+ peer asks for an enormous max delay so nothing ever expires. -5. The benign twin: a real viewer on a bufferbloated link. The fix must not - cost it the live edge. - -Recommended directions (the implementer settles the shape): - -- Make a group's serve state follow credit rather than arrival. One option is - to open a stream only when the session can grant one and then pick the best - group still worth sending at that moment, so a parked backlog lives in the - cache (already billed and evicted) rather than in per-subscriber machines. - Whatever the shape, cross-subscription priority and per-broadcast fairness - ([Scope track priority](/quest/m1/track-priority-scope.md)) must still - decide who gets the next stream. -- A parked group that can no longer be served (aborted, evicted, expired) - leaves at once and is named to the subscriber where the version can, through - SUBSCRIBE_DROP. -- The serving budget never exceeds what this hop would retain anyway: clamp - the requested or defaulted max delay by the local retention (the - `cache_duration` ceiling, the track's `max_age`, and possibly the pool's idle - window). Consider whether the relay's default `cache_duration` should follow - the pool's window instead of `Duration::MAX`; that changes a user-facing - default, so update `doc/bin/relay/` with it. -- Prefer bounds the protocol already carries (stream credit, max delay, - `max_age`, the cache pool) over a new per-subscription constant. If a number - is unavoidable, it must come from one of those. - -Sessions: capping them is out of scope. A session costs a handshake, auth, and -the per-session limits above, and a count cap would be a constant that only -throws. Admission control (per-token or per-address session caps) belongs to -the deployment's auth server, which already refuses a connect with a 401 or -403 ([relay auth](/doc/bin/relay/auth.md)); this repo needs nothing more. - -Reproduce first. A mock-time sim test (or a `just` drill against a real relay) -with a subscriber that withholds stream credit, or drains below the media rate -behind a deep finite queue, while the publisher writes groups across many -tracks. Measure, over time, live serve machines (pending and opened, as a -gauge) and the bytes they retain outside the pool (batches, chunks, and -unsent transport data), beside pool-billed bytes compared with a -no-subscriber baseline. Only lite counts pending opens today; IETF -`TrackServe` counts streams only after they open, so add the gauge there. On -main they grow with time; done, they plateau. Per the fan-out benchmark rule, -sweep tracks by groups per track, and subscribers by tracks, timing each -stream grant too, so a cost that grows with the table (such as scanning every -subscription when one stream frees up) shows up as a slope. Cover lite and -moq-transport, and scenario 5 still receives the latest group. -Include a benign withholder: a peer bounded by -[Bound incomplete messages](/quest/m1/quic/receive-memory.md) withholds credit -by design under memory pressure, so two relays running both quests must still -plateau and keep the live edge. - -Done looks like: those measurements plateau on both protocols and every -scenario above, the regression tests fail on main and run in CI, the JS -publisher in `js/net` is checked for the same shape and fixed or its gap -named in the PR, and the PR reports public API and wire impact (expected: -none on the wire; possibly a relay default). - -## Related - -- [Bound incomplete messages](/quest/m1/quic/receive-memory.md) - the ingress counterpart: a receiver withholds QUIC credit to bound what it holds, and this quest bounds what the sender holds while it does -- [Rust request credit](/quest/m1/rs-request-credit.md) - requests fail fast without stream credit; publisher group streams are left to this quest -- [Scope track priority](/quest/m1/track-priority-scope.md) - who gets the next stream across subscriptions and tenants -- [Public serving](/quest/m1/public-model/serve.md) - makes group expiry while serving public; coordinate the publisher serve loops with it -- [Treat network input as untrusted](https://github.com/moq-dev/moq/pull/5961) - the rule this quest applies -- [Resumed group head](https://github.com/moq-dev/moq/pull/5958) - the audit that found this, and bounded held heads by the cache pool diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index c4fe599cdb..972772fd4c 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -16,6 +16,10 @@ //! groups have a `_cpu` suffix. Cache-clock advancement remains outside both clocks. //! `session_delivery_group_cost` sweeps publishers and viewers independently for SUBSCRIBE //! and FETCH; its constrained cases release held streams only after a real blocked write. +//! `session_delivery_credit` grants a viewer one uni stream at a time, swept over its +//! subscriptions and the backlog each holds, so every group waits for a stream grant. A +//! publisher that parks state per backlogged group pays for it on each grant, which shows +//! as a slope over the backlog. //! `SESSION_EXPIRY_ONLY=1` uses the standalone 30s-expiry pool and reports groups //! written plus charged cache bytes every 10s during long profiling runs. Each round //! advances the cache clock by a second, so the expiry window spans 30 rounds. @@ -111,6 +115,8 @@ struct Shape { history: usize, /// Bytes queued before a mock stream applies backpressure. window: usize, + /// Group streams a viewer's relay may have open toward it at once. + credit: Option, /// Read each group via a fresh FETCH rather than a live subscription. fetch: bool, } @@ -130,6 +136,7 @@ impl Shape { burst: 1, history: 1, window: 1024 * 1024, + credit: None, fetch: false, }; @@ -154,6 +161,11 @@ impl Shape { } fn id(&self, version: &str) -> String { + // The credit sweep names only what it varies: the full form passes Criterion's + // 100-character title limit, which truncates and renumbers it. + if let Some(credit) = self.credit { + return format!("{version}/watch={}/burst={}/credit={credit}", self.watch, self.burst); + } format!( "{version}/relays={}/publishers={}/broadcasts={}/viewers={}/watch={}/frame={}{}{}{}{}{}{}", self.relays, @@ -305,6 +317,7 @@ fn path(broadcast: usize) -> String { struct Cluster { version: Version, window: usize, + credit: Option, relays: Vec, /// One track per broadcast, indexed like [`path`]. tracks: Vec, @@ -321,6 +334,7 @@ impl Cluster { let mut this = Self { version: version.parse().unwrap(), window: shape.window, + credit: shape.credit, relays: Vec::new(), tracks: Vec::new(), _broadcasts: Vec::new(), @@ -393,6 +407,9 @@ impl Cluster { let (origin, pool) = self.origin(CLIENT_CACHE); let mut options = MockConnectOptions::new(self.version); options.transport.stream_buffer = self.window; + // Draft 22, the one IETF version here, keeps its SETUP stream open beside the groups. + options.transport.max_concurrent_uni_streams = + self.credit.map(|credit| credit + usize::from(self.version.is_ietf())); // Hidden included, as the dash's `.stats/` scope sees its subtree. options.server_publish = Some(self.relays[relay].consume().with_hidden(true)); options.client_subscribe = Some(origin.clone()); @@ -1249,6 +1266,22 @@ fn session(c: &mut Criterion) { ..base }), ); + // The same backlog behind one uni stream grant at a time. + delivery( + c, + "credit", + [1, 16, 64].into_iter().flat_map(|watch| { + [1, 16, 64].map(move |burst| Shape { + publishers: 1, + broadcasts: 64, + viewers: 1, + watch, + burst, + credit: Some(1), + ..base + }) + }), + ); delivery( c, "relays", diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 4e7d6e9ef5..3f762802b2 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -4117,7 +4117,9 @@ struct Terminal { sent: AtomicBool, } -/// Serves a track's groups, one machine per group, with unlimited concurrency. +/// Serves a track's groups, one machine per group. One at a time waits for stream credit; +/// the rest wait in the cache, so what a peer withholding credit makes this hold is bounded +/// by the credit it grants, not by how long it withholds it. struct TrackServe { session: S, track: track::Subscriber, @@ -4125,6 +4127,11 @@ struct TrackServe { version: Version, range: ServeRange, timescale: Option, + /// The one group waiting for stream credit. The cursor hands out no other meanwhile, so + /// the backlog stays in the cache, which bills, ages, and evicts it, and the cursor's + /// budget skips whatever went stale. + opening: Option>, + /// The groups with a stream, serving or awaiting the acknowledgement of their FIN. children: kio::Tasks>, /// The subscriber priority, which every group stream follows while it is open. priority: kio::Producer, @@ -4186,6 +4193,7 @@ impl TrackServe { version, range, timescale, + opening: None, children: kio::Tasks::new(), priority, ranking, @@ -4238,9 +4246,16 @@ impl TrackServe { /// Close every group stream still in flight. Dropping a group machine resets its /// stream, as a cancelled subscription's would; a drained track has none left. fn close_groups(&mut self) { + self.opening = None; self.children = kio::Tasks::new(); } + /// Group machines held: the one waiting for stream credit, and those with a stream. + #[cfg(test)] + fn machines(&self) -> usize { + self.children.len() + usize::from(self.opening.is_some()) + } + /// Where the track ends, when the subscription ran to that end rather than stopping at /// its own range first. fn end(&self) -> Option { @@ -4303,6 +4318,16 @@ impl TrackServe { if self.budget.poll_yield(waiter).is_pending() { break; } + if let Some(opening) = &mut self.opening { + let Poll::Ready(res) = opening.poll_open(waiter) else { + break; + }; + let opening = self.opening.take().expect("polled above"); + // A group that failed without a stream has nothing to reset. + if res.is_ok() { + self.children.push(opening); + } + } match self.track.poll_recv_group(waiter) { Poll::Ready(Ok(Some(group))) => { let sequence = group.sequence; @@ -4351,7 +4376,7 @@ impl TrackServe { group: sequence, object: slice.skip, }); - self.children.push( + self.opening = Some( GroupServe::new( self.session.clone(), msg, @@ -4817,6 +4842,43 @@ impl GroupServe { self } + /// Wait for stream credit, then write the group header, leaving the machine serving. + /// Gives up once the group can no longer be sent, so a parked group never spends the + /// credit it was waiting for on a reset. + fn poll_open(&mut self, waiter: &kio::Waiter) -> Poll> { + if !matches!(self.state, GroupState::Open) { + return Poll::Ready(Ok(())); + } + if let Some(err) = self.group.poll_abandoned(waiter, false) { + self.state = GroupState::Done; + return Poll::Ready(Err(err)); + } + // The current rank, so a move while parked applies to the stream as it opens. + let _ = self.rank.poll_moved(waiter); + let stream = match ready!(self.session.poll_open_uni(&mut waiter.context(), self.rank.options())) { + Ok(stream) => stream, + Err(err) => { + self.state = GroupState::Done; + return Poll::Ready(Err(Error::from_transport(err))); + } + }; + self.opened.fetch_add(1, Ordering::Relaxed); + + let mut writer = Writer::new(stream, self.version); + if let Err(err) = writer.buffer(&self.msg) { + self.state = GroupState::Done; + return Poll::Ready(Err(err)); + } + self.state = GroupState::Serve { + writer, + frame: None, + chunk: None, + batch: Box::new(arrayvec::ArrayVec::new()), + batch_pos: 0, + }; + Poll::Ready(Ok(())) + } + fn poll_serve(&mut self, waiter: &kio::Waiter) -> Poll> { let mut cx = waiter.context(); // A REQUEST_UPDATE can move the subscriber priority while this stream is open, and @@ -4827,33 +4889,9 @@ impl GroupServe { loop { match &mut self.state { GroupState::Open => { - if self.group.poll_expired(waiter) { - self.state = GroupState::Done; - return Poll::Ready(Err(Error::Old)); - } + ready!(self.poll_open(waiter))?; // The current rank, which already includes anything `moved` saw. - let stream = match ready!(self.session.poll_open_uni(&mut cx, self.rank.options())) { - Ok(stream) => stream, - Err(err) => { - self.state = GroupState::Done; - return Poll::Ready(Err(Error::from_transport(err))); - } - }; - self.opened.fetch_add(1, Ordering::Relaxed); moved = None; - - let mut writer = Writer::new(stream, self.version); - if let Err(err) = writer.buffer(&self.msg) { - self.state = GroupState::Done; - return Poll::Ready(Err(err)); - } - self.state = GroupState::Serve { - writer, - frame: None, - chunk: None, - batch: Box::new(arrayvec::ArrayVec::new()), - batch_pos: 0, - }; } GroupState::Serve { writer, @@ -4878,11 +4916,12 @@ impl GroupServe { Poll::Ready(Err(err)) => break 'serve Err(err), // Parking on the transport is the one stall the group cursor cannot // see, and the only place a served group applies the drift budget: - // flow control must not pin a stream that has gone stale. `true` - // because the transport still owns bytes the cursor has released. + // flow control must not pin a stream that has gone stale, or one + // whose unread rest the cache evicted. `true` because the transport + // still owns bytes the cursor has released. Poll::Pending => { - if self.group.poll_expired_while_pending(waiter, true) { - break 'serve Err(Error::Old); + if let Some(err) = self.group.poll_abandoned(waiter, true) { + break 'serve Err(err); } return Poll::Pending; } @@ -4897,11 +4936,12 @@ impl GroupServe { Poll::Ready(Err(err)) => break 'serve Err(err), // Parking on the transport is the one stall the group cursor cannot // see, and the only place a served group applies the drift budget: - // flow control must not pin a stream that has gone stale. `true` - // because the transport still owns bytes the cursor has released. + // flow control must not pin a stream that has gone stale, or one + // whose unread rest the cache evicted. `true` because the transport + // still owns bytes the cursor has released. Poll::Pending => { - if self.group.poll_expired_while_pending(waiter, true) { - break 'serve Err(Error::Old); + if let Some(err) = self.group.poll_abandoned(waiter, true) { + break 'serve Err(err); } return Poll::Pending; } @@ -5876,6 +5916,108 @@ mod group_priority_test { 60 } + /// Poll `serve` a few times, yielding between them, so its per-poll budget lets every + /// group that is ready move on. + async fn settle_serve(serve: &mut TrackServe) { + for _ in 0..8 { + let _ = serve.poll(&kio::Waiter::noop()); + moq_clock::sim::yield_now().await; + } + } + + /// A peer withholding stream credit makes each subscription hold one group machine, + /// however many tracks it subscribes to and groups they publish: the backlog waits in the + /// cache. Granting credit then serves every group still worth sending. + #[moq_clock::test] + async fn withheld_credit_holds_one_group_machine_per_subscription() { + for (tracks, groups) in [(1, 1), (1, 64), (16, 1), (16, 64)] { + let gate = kio::Producer::new(false); + let session = SinkSession::gated_open_uni(gate.consume()); + let mut serves: Vec<_> = (0..tracks) + .map(|id| { + let track = track::Producer::new( + std::sync::Arc::new(crate::broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + // Nothing goes stale, so only the credit bounds what is held. + let subscriber = + track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); + let serve = TrackServe::new( + session.clone(), + subscriber, + RequestId(id), + Version::Draft22, + ServeRange::default(), + Some(Timescale::MILLI), + ); + (track, serve) + }) + .collect(); + + for sequence in 0..groups { + for (track, serve) in &mut serves { + let mut group = track.append_group().unwrap(); + group + .write_frame(crate::Timestamp::from_millis(sequence).unwrap(), b"x".as_slice()) + .unwrap(); + group.finish().unwrap(); + settle_serve(serve).await; + } + let held = serves.iter().map(|(_, serve)| serve.machines() as u64).sum::(); + assert_eq!(held, tracks, "{tracks} tracks after {} groups each", sequence + 1); + } + + *gate.write().ok().unwrap() = true; + for (_, serve) in &mut serves { + settle_serve(serve).await; + assert_eq!(serve.opened(), groups); + assert_eq!(serve.machines(), 0, "every group was served"); + } + } + } + + /// A group waiting for stream credit that the cache evicts leaves at once, rather than + /// spending the credit on a reset once it arrives. The next group takes its place. + #[moq_clock::test] + async fn an_evicted_group_waiting_for_credit_leaves_at_once() { + let gate = kio::Producer::new(false); + let track = track::Producer::new( + std::sync::Arc::new(crate::broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + let subscriber = track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); + let mut serve = TrackServe::new( + SinkSession::gated_open_uni(gate.consume()), + subscriber, + RequestId(0), + Version::Draft22, + ServeRange::default(), + Some(Timescale::MILLI), + ); + + let mut group = track.append_group().unwrap(); + group.write_frame(crate::Timestamp::ZERO, b"x".as_slice()).unwrap(); + settle_serve(&mut serve).await; + assert_eq!(serve.machines(), 1, "group 0 waits for credit"); + + group.abort(Error::Evicted).unwrap(); + settle_serve(&mut serve).await; + assert_eq!(serve.machines(), 0, "the evicted group leaves without credit"); + + let mut next = track.append_group().unwrap(); + next.write_frame(crate::Timestamp::from_millis(1).unwrap(), b"x".as_slice()) + .unwrap(); + next.finish().unwrap(); + settle_serve(&mut serve).await; + assert_eq!(serve.machines(), 1, "group 1 takes its place"); + + *gate.write().ok().unwrap() = true; + settle_serve(&mut serve).await; + assert_eq!(serve.opened(), 1, "the credit goes to group 1 alone"); + } + /// A subgroup waiting for stream credit keeps its subscription expiry armed. #[moq_clock::test] async fn group_waiting_for_stream_credit_expires() { diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 6350458537..4217b63f39 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3425,19 +3425,27 @@ fn final_sequence(end: track::Position) -> u64 { /// permanently skip the older group even though it is cached and in demand. Staleness is /// governed by the max age window (cache expiry), not arrival raciness. /// +/// Without `groups` the cursor hands out none, leaving them in the cache, so it never reports +/// [`Recv::Finished`] either. +/// /// When `emit_boundary` is set, a declared-but-not-yet-reached final sequence surfaces as /// [`Recv::Boundary`] in an idle moment (after groups and datagrams), so the caller can send /// SUBSCRIBE_END as soon as the ending is known rather than waiting for the live edge to reach /// it. The caller clears `emit_boundary` after the first boundary so it fires once. fn poll_recv_next( track: &mut track::Subscriber, + groups: bool, datagrams: bool, emit_boundary: bool, waiter: &kio::Waiter, ) -> Poll> { { let mut groups_finished = false; - match track.poll_recv_group(waiter)? { + let next = match groups { + true => track.poll_recv_group(waiter)?, + false => Poll::Pending, + }; + match next { Poll::Ready(Some(group)) => return Poll::Ready(Ok(Recv::Group(group))), Poll::Ready(None) => groups_finished = true, Poll::Pending => {} @@ -3465,7 +3473,7 @@ fn poll_recv_next( /// The async form of [`poll_recv_next`], for callers with nothing else to poll. #[cfg(test)] async fn recv_next(track: &mut track::Subscriber, datagrams: bool, emit_boundary: bool) -> Result { - kio::wait(|waiter| poll_recv_next(track, datagrams, emit_boundary, waiter)).await + kio::wait(|waiter| poll_recv_next(track, true, datagrams, emit_boundary, waiter)).await } /// Bound `group` to what this subscription asked for, reporting whether it can be served @@ -3730,7 +3738,8 @@ impl Subscription { } /// A subscription's run loop: one subscriber cursor serving groups, datagrams, -/// and SUBSCRIBE_UPDATE messages, with an in-flight group machine per group. +/// and SUBSCRIBE_UPDATE messages, with a machine per group it serves, one at a time +/// waiting for stream credit. struct TrackRun { ctx: Subscription, track: track::Subscriber, @@ -3762,6 +3771,12 @@ struct TrackRun { // datagram-capable transport (qmux/WebSocket/TCP/UDS report size 0). No group // fallback: otherwise off. datagrams: bool, + /// The one group waiting for stream credit. The rest of the backlog waits in the cache, + /// which bills, ages, and evicts it, and the cursor's budget skips whatever went stale + /// meanwhile: a machine per parked group would hold state for as long as the peer + /// withholds credit. + opening: Option>, + /// The groups with a stream, serving or awaiting the acknowledgement of their FIN. children: kio::Tasks>, } @@ -3789,10 +3804,17 @@ impl TrackRun { end_sent: false, served, datagrams, + opening: None, children: kio::Tasks::new(), } } + /// Group machines held: the one waiting for stream credit, and those with a stream. + #[cfg(test)] + fn machines(&self) -> usize { + self.children.len() + usize::from(self.opening.is_some()) + } + /// Apply a SUBSCRIBE_UPDATE. /// /// `end_group` is a serving cap, not a subscription terminator: groups past the @@ -3888,6 +3910,17 @@ impl TrackRun { // Deliver the buffered range messages before selecting more work. ready!(writer.poll_flush(&mut cx))?; + // The group waiting for stream credit joins the others once it has one, or fails + // without one. + if let Some(opening) = &mut self.opening + && let Poll::Ready(res) = opening.poll_open(waiter) + { + let opening = self.opening.take().expect("polled above"); + match res { + Ok(()) => self.children.push(opening), + Err(_) => opening.settle_failed(), + } + } // Drive the in-flight group machines; completions just retire. let _ = self.children.poll(waiter); // A group whose header does not fit this peer's varint wrote nothing. The @@ -3955,7 +3988,8 @@ impl TrackRun { // first so a datagram burst can't starve them; datagrams flow whenever no // group is ready (including while groups are parked above the cap). let emit_boundary = self.emit_range && !self.end_sent; - if let Poll::Ready(res) = poll_recv_next(&mut self.track, self.datagrams, emit_boundary, waiter) { + let groups = self.opening.is_none(); + if let Poll::Ready(res) = poll_recv_next(&mut self.track, groups, self.datagrams, emit_boundary, waiter) { match res? { Recv::Group(mut group) => { // Never a second stream for a sequence this subscription already accounted @@ -4153,7 +4187,8 @@ impl TrackRun { Ok(()) } - /// Open a group machine for `group`. + /// Open a group machine for `group`, which waits for stream credit before the cursor + /// hands out another. fn serve(&mut self, group: group::Consumer) { let sequence = group.sequence; let frame_start = group.index(); @@ -4175,7 +4210,8 @@ impl TrackRun { .insert(Priority::new(steer.priority, self.ctx.id, sequence).with_order(steer.order)), ), }; - self.children.push(GroupServe::new( + debug_assert!(self.opening.is_none(), "one group waits for stream credit at a time"); + self.opening = Some(GroupServe::new( self.ctx.clone(), sequence, frame_start, @@ -4268,6 +4304,79 @@ impl GroupServe { } } + /// Wait for stream credit, then write the group header, leaving the machine serving. + /// Gives up once the group can no longer be sent, so a parked group never spends the + /// credit it was waiting for on a reset. + fn poll_open(&mut self, waiter: &kio::Waiter) -> Poll> { + if !matches!(self.state, GroupState::Open) { + return Poll::Ready(Ok(())); + } + if let Some(err) = self.group.poll_abandoned(waiter, false) { + self.settle_open(false); + self.state = GroupState::Done; + return Poll::Ready(Err(err)); + } + let mut cx = waiter.context(); + let options = match &mut self.rank { + Rank::Group(group) => SendOptions { + group: Some(&*group), + priority: crate::priority::in_group(self.sequence, self.frame_start, self.ctx.steer_seen.order), + }, + Rank::Queue(handle) => SendOptions { + group: None, + priority: handle.send_order().into(), + }, + }; + let stream = match ready!(self.ctx.session.poll_open_uni(&mut cx, options)) { + Ok(stream) => stream, + Err(err) => { + self.settle_open(false); + self.state = GroupState::Done; + return Poll::Ready(Err(crate::lite::error::from_transport(err))); + } + }; + self.settle_open(true); + let mut writer = Writer::new(stream, self.ctx.version); + + let msg = lite::Group { + subscribe: self.ctx.id, + sequence: self.sequence, + frame_start: self.frame_start, + }; + let header = writer.buffer(&lite::DataType::Group).and_then(|()| writer.buffer(&msg)); + let records = header.and_then(|()| { + lite::ietf::Records::start( + &mut writer, + self.ctx.ietf, + &self.group, + self.capped, + self.ctx.timescale, + self.ctx.version, + ) + }); + let records = match records { + Ok(records) => records, + Err(err) => { + // The header wrote nothing, so the group-stream reset has nothing to pin + // it to this subscription. The run loop reads the slot and resets the + // subscribe stream instead. + self.ctx.note_unencodable(&err); + self.state = GroupState::Done; + writer.abort(&err); + return Poll::Ready(Err(err)); + } + }; + self.state = GroupState::Serve { + writer, + frame: None, + chunk: None, + batch: Box::default(), + batch_pos: 0, + records, + }; + Poll::Ready(Ok(())) + } + /// Serve the group, aborting the stream with the real reason (Old, Lagged, /// Evicted, ...) on failure so the subscriber can tell a truncated group from /// a routine cancel. Without this the Writer's Drop fallback would report @@ -4306,75 +4415,7 @@ impl GroupServe { } match &mut self.state { - GroupState::Open => { - if self.group.poll_expired(waiter) { - self.settle_open(false); - self.state = GroupState::Done; - return Poll::Ready(Err(Error::Old)); - } - let mut cx = waiter.context(); - let options = match &mut self.rank { - Rank::Group(group) => SendOptions { - group: Some(&*group), - priority: crate::priority::in_group( - self.sequence, - self.frame_start, - self.ctx.steer_seen.order, - ), - }, - Rank::Queue(handle) => SendOptions { - group: None, - priority: handle.send_order().into(), - }, - }; - let stream = match ready!(self.ctx.session.poll_open_uni(&mut cx, options)) { - Ok(stream) => stream, - Err(err) => { - self.settle_open(false); - self.state = GroupState::Done; - return Poll::Ready(Err(crate::lite::error::from_transport(err))); - } - }; - self.settle_open(true); - let mut writer = Writer::new(stream, self.ctx.version); - - let msg = lite::Group { - subscribe: self.ctx.id, - sequence: self.sequence, - frame_start: self.frame_start, - }; - let header = writer.buffer(&lite::DataType::Group).and_then(|()| writer.buffer(&msg)); - let records = header.and_then(|()| { - lite::ietf::Records::start( - &mut writer, - self.ctx.ietf, - &self.group, - self.capped, - self.ctx.timescale, - self.ctx.version, - ) - }); - let records = match records { - Ok(records) => records, - Err(err) => { - // The header wrote nothing, so the group-stream reset has nothing to pin - // it to this subscription. The run loop reads the slot and resets the - // subscribe stream instead. - self.ctx.note_unencodable(&err); - self.state = GroupState::Done; - writer.abort(&err); - return Poll::Ready(Err(err)); - } - }; - self.state = GroupState::Serve { - writer, - frame: None, - chunk: None, - batch: Box::default(), - batch_pos: 0, - records, - }; - } + GroupState::Open => ready!(self.poll_open(waiter))?, GroupState::Serve { writer, frame, @@ -4397,11 +4438,12 @@ impl GroupServe { Poll::Ready(Err(err)) => break 'serve Err(err), // Parking on the transport is the one stall the group cursor cannot // see, and the only place a served group applies the drift budget: - // flow control must not pin a stream that has gone stale. `true` - // because the transport still owns bytes the cursor has released. + // flow control must not pin a stream that has gone stale, or one + // whose unread rest the cache evicted. `true` because the transport + // still owns bytes the cursor has released. Poll::Pending => { - if self.group.poll_expired_while_pending(waiter, true) { - break 'serve Err(Error::Old); + if let Some(err) = self.group.poll_abandoned(waiter, true) { + break 'serve Err(err); } return Poll::Pending; } @@ -4416,11 +4458,12 @@ impl GroupServe { Poll::Ready(Err(err)) => break 'serve Err(err), // Parking on the transport is the one stall the group cursor cannot // see, and the only place a served group applies the drift budget: - // flow control must not pin a stream that has gone stale. `true` - // because the transport still owns bytes the cursor has released. + // flow control must not pin a stream that has gone stale, or one + // whose unread rest the cache evicted. `true` because the transport + // still owns bytes the cursor has released. Poll::Pending => { - if self.group.poll_expired_while_pending(waiter, true) { - break 'serve Err(Error::Old); + if let Some(err) = self.group.poll_abandoned(waiter, true) { + break 'serve Err(err); } return Poll::Pending; } @@ -4569,15 +4612,24 @@ impl kio::Task for GroupServe { type Output = (); fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { - // A reset can lose its header, so the control stream also accounts for the group. - if ready!(self.poll_serve(waiter)).is_err() && self.ctx.version.has_track_stream() { + if ready!(self.poll_serve(waiter)).is_err() { + self.settle_failed(); + } + Poll::Ready(()) + } +} + +impl GroupServe { + /// Account for a group that failed: a reset can lose its header, and a group that never + /// got a stream has none, so the control stream names it. + fn settle_failed(&self) { + if self.ctx.version.has_track_stream() { self.ctx.outcomes.dropped.lock().unwrap().push(lite::SubscribeDrop { start: self.sequence, end: self.sequence, error: 0, }); } - Poll::Ready(()) } } @@ -4969,6 +5021,177 @@ mod serve_group_test { assert!(matches!(serve.await, Err(Error::Old))); } + /// Step `run` until it has nothing ready, yielding between steps so a long drain stays + /// under the simulator's per-poll limit. + async fn settle_run(run: &mut TrackRun, writer: &mut Writer) { + while run.poll_step(writer, &kio::Waiter::noop()).is_ready() { + moq_clock::sim::yield_now().await; + } + } + + /// A peer withholding stream credit makes each subscription hold one group machine, however + /// many tracks it subscribes to and groups they publish: the backlog waits in the cache. + /// Granting credit then serves every group still worth sending. + #[moq_clock::test] + async fn withheld_credit_holds_one_group_machine_per_subscription() { + for (tracks, groups) in [(1, 1), (1, 64), (16, 1), (16, 64)] { + let gate = kio::Producer::new(false); + let session = SinkSession::gated_open_uni(gate.consume()); + let mut runs: Vec<_> = (0..tracks) + .map(|_| { + let track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + // Nothing goes stale, so only the credit bounds what is held. + let subscriber = + track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); + let (run, writer, _) = lite_run(Version::Lite07, session.clone(), subscriber); + (track, run, writer) + }) + .collect(); + + for sequence in 0..groups { + for (track, run, writer) in &mut runs { + write_group(track, sequence, sequence); + settle_run(run, writer).await; + } + let held: usize = runs.iter().map(|(_, run, _)| run.machines()).sum(); + assert_eq!(held, tracks, "{tracks} tracks after {} groups each", sequence + 1); + } + + *gate.write().ok().unwrap() = true; + for (_, run, writer) in &mut runs { + settle_run(run, writer).await; + assert_eq!(run.ctx.outcomes.opened.load(Ordering::Relaxed), groups); + assert_eq!(run.machines(), 0, "every group was served"); + } + } + } + + /// A group waiting for stream credit that the cache evicts leaves at once, named with + /// SUBSCRIBE_DROP, rather than spending the credit on a reset once it arrives. The next + /// group takes its place. + #[moq_clock::test] + async fn an_evicted_group_waiting_for_credit_leaves_at_once() { + let version = Version::Lite07; + let gate = kio::Producer::new(false); + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + let subscriber = track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); + let (mut run, mut writer, log) = lite_run(version, SinkSession::gated_open_uni(gate.consume()), subscriber); + + let mut group = track.create_group(group::Info { sequence: 0 }).unwrap(); + group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap(); + while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} + assert_eq!(run.machines(), 1, "group 0 waits for credit"); + + group.abort(Error::Evicted).unwrap(); + while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} + assert_eq!(run.machines(), 0, "the evicted group leaves without credit"); + assert_eq!(drops(&mut writer, &log, version).await, [(0, 0)]); + + write_group(&mut track, 1, 1); + while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} + assert_eq!(run.machines(), 1, "group 1 takes its place"); + + *gate.write().ok().unwrap() = true; + while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} + assert_eq!( + run.ctx.outcomes.opened.load(Ordering::Relaxed), + 1, + "the credit goes to group 1 alone" + ); + } + + /// After a stall, the credit goes to the live edge: the group parked when credit ran out + /// expires, and the cursor skips the backlog the budget gave up on while it waited. + #[moq_clock::test] + async fn credit_after_a_stall_serves_the_live_edge() { + let version = Version::Lite07; + let gate = kio::Producer::new(false); + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + let subscriber = track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(1))); + let (mut run, mut writer, log) = lite_run(version, SinkSession::gated_open_uni(gate.consume()), subscriber); + for sequence in 0..10 { + write_group(&mut track, sequence, sequence * 1000); + while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} + assert!(run.machines() <= 1); + moq_clock::sim::advance(Duration::from_secs(1)).await; + } + + *gate.write().ok().unwrap() = true; + while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} + assert_eq!( + run.ctx.outcomes.opened.load(Ordering::Relaxed), + 1, + "one stream, for the newest group" + ); + let drops = drops(&mut writer, &log, version).await; + assert_eq!(drops.first(), Some(&(0, 0)), "the parked group expired"); + assert!( + drops.iter().all(|&(_, end)| end < 9), + "the newest group is not dropped: {drops:?}" + ); + } + + /// A stream parked on the transport resets once the cache evicts the part of its group + /// it has not read: the rest can never be sent, so the buffered bytes are released + /// rather than held until the peer grants flow control. + #[moq_clock::test] + async fn blocked_transport_write_resets_once_the_group_is_evicted() { + let gate = kio::Producer::new(false); + let session = SinkSession::gated_uni(gate.consume()); + let log = session.log.clone(); + let steer = kio::Producer::new(Steer::default()); + let subscription = Subscription { + session, + id: 0, + track_name: "test".into(), + priority: PriorityQueue::default(), + group: None, + steer: steer.consume(), + steer_seen: Steer::default(), + version: Version::Lite06, + timescale: Some(crate::Timescale::MILLI), + runtime: crate::time::sim(), + outcomes: Default::default(), + ietf: false, + }; + + let track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + let mut subscriber = track.subscribe(None); + let mut producer = track.append_group().unwrap(); + producer.write_frame(Timestamp::ZERO, b"head".as_slice()).unwrap(); + let group = subscriber.recv_group().await.unwrap().expect("group"); + + let handle = subscription.priority.insert(Priority::new(0, 0, 0)); + let mut serve = std::pin::pin!(subscription.serve_group(0, 0, handle, group)); + assert!( + futures::poll!(serve.as_mut()).is_pending(), + "transport write is blocked" + ); + + producer.abort(Error::Evicted).unwrap(); + assert!(matches!( + futures::poll!(serve.as_mut()), + Poll::Ready(Err(Error::Evicted)) + )); + assert_eq!(log.resets(), vec![crate::StreamError::Evicted.to_code()]); + } + /// Lite01/02 have no max delay field, so a SUBSCRIBE from one decodes as /// `Duration::ZERO`. Serving that as a real-time budget would hold every legacy /// peer to the live edge and discard backlog it never declined, so those versions diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index a6737f3904..81e49575e2 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -2207,6 +2207,44 @@ impl Consumer { self.expired } + /// Why a wire publisher waiting on the transport, for stream credit or to drain, should + /// give up on this group: the drift budget convicted it ([`Error::Old`]), or its copy + /// was aborted (evicted, cut, or failed upstream) before this cursor read it all and no + /// route will continue it. `pending` as in [`Self::poll_expired_while_pending`]. + /// + /// A machine parked on stream credit reads nothing, so without this an aborted group, + /// which never expires, would wait for credit it could only spend on a reset. + // pending: /quest/m1/public-model/serve.md + pub(crate) fn poll_abandoned(&mut self, waiter: &kio::Waiter, pending: bool) -> Option { + if self.poll_expired_while_pending(waiter, pending) { + return Some(Error::Old); + } + // Registered on the close alone: an abort closes the channel, and waking on every + // frame written to a parked group would cost a stalled subscriber per frame. + if self.cursor.state.poll_closed(waiter).is_pending() { + return None; + } + let aborted = { + let state = self.cursor.state.read(); + let unread = !self.cursor.capped() && state.fin.is_none_or(|fin| self.cursor.index < fin); + match &state.abort { + // The rest of the group is gone; what was read already is not. + Some(err) if unread => err.clone(), + // Read to its end, or finished and closed with its last producer: still servable. + _ => return None, + } + }; + // A front's reader may continue the group from another route's copy when it reads. + if self + .recover + .as_ref() + .is_some_and(|recover| recover.wants(Some(&aborted), waiter)) + { + return None; + } + Some(aborted) + } + /// Whether expiry can still discard content or unblock a group that may grow. fn expiry_pending(&self) -> bool { self.cursor.expiry_pending() diff --git a/rs/moq-net/src/transport/memory.rs b/rs/moq-net/src/transport/memory.rs index db944fd01f..66709f2a17 100644 --- a/rs/moq-net/src/transport/memory.rs +++ b/rs/moq-net/src/transport/memory.rs @@ -8,6 +8,9 @@ //! pends while it is full, so a slow reader throttles only its own stream. A write is //! never split: one accepted while the channel has room may overshoot the bound. //! +//! [`Config::max_concurrent_uni_streams`] bounds the uni streams each side has open toward +//! its peer, as QUIC's stream credit does: an open pends until the peer is done with one. +//! //! [`finish`](poll::SendStream::finish) closes the channel cleanly once its messages are //! read. A [`reset`](poll::SendStream::reset) or [`stop`](poll::RecvStream::stop) closes it //! with a code and drops what it queued. Closing the session keeps queued messages @@ -42,6 +45,10 @@ pub struct Config { pub stream_buffer: usize, /// The application protocol both sides report, standing in for a negotiated ALPN. pub protocol: Option, + /// Uni streams one side may have open toward its peer at once, as QUIC's stream credit. + /// A stream is open until its reader reaches the end, stops, or drops it, accepted or + /// not. `None` grants credit without limit. + pub max_concurrent_uni_streams: Option, } impl Default for Config { @@ -51,6 +58,7 @@ impl Default for Config { Self { stream_buffer: 1024 * 1024, protocol: None, + max_concurrent_uni_streams: None, } } } @@ -70,6 +78,7 @@ pub fn pair(config: Config) -> (Session, Session) { }), accept_uni: Default::default(), accept_bi: Default::default(), + open_uni: Default::default(), datagram: Default::default(), closed: Default::default(), }; @@ -154,7 +163,7 @@ impl Conn { } } - fn stream(&self) -> (SendStream, RecvStream) { + fn stream(&self, credit: Option>) -> (SendStream, RecvStream) { let channel = kio::Shared::::default(); let send = SendStream { channel: channel.clone(), @@ -167,6 +176,7 @@ impl Conn { channel, close: self.close.clone(), stopped: false, + credit, park: Default::default(), }; (send, recv) @@ -176,6 +186,8 @@ impl Conn { /// Streams and datagrams one side has yet to accept. struct Incoming { uni: kio::Queue, + /// The uni streams toward this side still open, against [`Config::max_concurrent_uni_streams`]. + uni_open: kio::Shared, bi: kio::Queue<(SendStream, RecvStream)>, datagrams: kio::Queue, } @@ -184,6 +196,7 @@ impl Default for Incoming { fn default() -> Self { Self { uni: kio::Queue::new(), + uni_open: Default::default(), bi: kio::Queue::new(), datagrams: kio::Queue::bounded(DATAGRAM_QUEUE), } @@ -236,6 +249,7 @@ pub struct Session { // polling two classes at once must not clobber its own registration. accept_uni: kio::Park, accept_bi: kio::Park, + open_uni: kio::Park, datagram: kio::Park, closed: kio::Park, } @@ -284,11 +298,28 @@ impl poll::Session for Session { fn poll_open_uni( &mut self, - _cx: &mut Context<'_>, + cx: &mut Context<'_>, _: poll::SendOptions<'_, poll::Ungrouped>, ) -> Poll> { - let (send, recv) = self.conn().stream(); - Poll::Ready(self.open(&self.side.outgoing().uni, recv).map(|()| send)) + let outgoing = self.side.outgoing(); + let credit = match self.conn().config.max_concurrent_uni_streams { + Some(max) => { + let waiter = self.open_uni.hold(cx); + // Check the close first, so a credit granted before it is still taken below. + let closed = poll_close(&self.side.conn.close, waiter); + let Poll::Ready(mut open) = outgoing.uni_open.poll(waiter, |open| match **open < max { + true => Poll::Ready(()), + false => Poll::Pending, + }) else { + return closed.map(Err); + }; + *open += 1; + Some(outgoing.uni_open.clone()) + } + None => None, + }; + let (send, recv) = self.conn().stream(credit); + Poll::Ready(self.open(&outgoing.uni, recv).map(|()| send)) } fn poll_open_bi( @@ -296,8 +327,8 @@ impl poll::Session for Session { _cx: &mut Context<'_>, _: poll::SendOptions<'_, poll::Ungrouped>, ) -> Poll> { - let (send, peer_recv) = self.conn().stream(); - let (peer_send, recv) = self.conn().stream(); + let (send, peer_recv) = self.conn().stream(None); + let (peer_send, recv) = self.conn().stream(None); let opened = self.open(&self.side.outgoing().bi, (peer_send, peer_recv)); Poll::Ready(opened.map(|()| (send, recv))) } @@ -504,9 +535,20 @@ pub struct RecvStream { close: Close, /// This side stopped the stream. stopped: bool, + /// The open-stream count this stream holds a credit in, until it is done. + credit: Option>, park: kio::Park, } +impl RecvStream { + /// Return this stream's credit to its writer's side, once. + fn release(&mut self) { + if let Some(credit) = self.credit.take() { + *credit.lock() -= 1; + } + } +} + impl poll::RecvStream for RecvStream { type Error = Error; @@ -552,7 +594,10 @@ impl poll::RecvStream for RecvStream { }; let Some(front) = channel.messages.front_mut() else { - return Poll::Ready(match channel.end.expect("checked above") { + let end = channel.end.expect("checked above"); + drop(channel); + self.release(); + return Poll::Ready(match end { End::Fin => Ok(None), End::Reset(code) => Err(Error::Reset(code)), End::Stop(_) => Err(Error::Closed), @@ -577,6 +622,8 @@ impl poll::RecvStream for RecvStream { if closed(&self.close).is_none() { channel.abort(End::Stop(code)); } + drop(channel); + self.release(); } /// Ready once the writer finished or reset the stream, or this side stopped it. @@ -735,6 +782,52 @@ mod tests { assert_eq!(ready(write(&mut send, &Bytes::from_static(b"two"))).unwrap(), 3); } + #[test] + fn uni_credit_pends_opens_until_a_stream_is_done() { + let (mut a, mut b) = pair(Config { + max_concurrent_uni_streams: Some(2), + ..Default::default() + }); + let open = |a: &mut Session, cx: &mut Context<'_>| a.poll_open_uni(cx, Default::default()); + + // Streams the peer has not accepted yet hold credit too. + let mut finished = ready(open(&mut a, &mut noop())).unwrap(); + let _reset = ready(open(&mut a, &mut noop())).unwrap(); + let count = Arc::new(Count::default()); + let waker = Waker::from(count.clone()); + assert!(open(&mut a, &mut Context::from_waker(&waker)).is_pending()); + + // Credit is per direction. + let _back = ready(open(&mut b, &mut noop())).unwrap(); + + // Reading to the end returns the credit, and wakes the parked open. + ready(write(&mut finished, &Bytes::from_static(b"x"))).unwrap(); + finished.finish().unwrap(); + let mut recv = ready(b.poll_accept_uni(&mut noop())).unwrap(); + assert_eq!(text(read(&mut recv, usize::MAX)), "x"); + assert!( + open(&mut a, &mut Context::from_waker(&waker)).is_pending(), + "an unread FIN keeps the credit" + ); + assert_eq!(ready(read(&mut recv, usize::MAX)).unwrap(), None); + assert!(count.0.load(Ordering::Relaxed) > 0); + let _third = ready(open(&mut a, &mut noop())).unwrap(); + assert!(open(&mut a, &mut noop()).is_pending()); + + // So does dropping a stream, accepted or not. + drop(recv); + assert!(open(&mut a, &mut noop()).is_pending(), "the credit was returned once"); + drop(ready(b.poll_accept_uni(&mut noop())).unwrap()); + let _fourth = ready(open(&mut a, &mut noop())).unwrap(); + + // A parked open fails once the session closes. + b.close(1, "bye"); + assert!(matches!( + ready(open(&mut a, &mut noop())), + Err(Error::Session { code: 1, .. }) + )); + } + #[test] fn reset_drops_queued_messages() { let (mut a, mut b) = pair(buffer(8)); diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index 5a838d8a60..dbd83270e0 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -48,6 +48,7 @@ mod session_close; mod session_panic; mod settled_end_survives_abort; mod source_errors; +mod stream_credit; mod subscribe_priority; mod subscribe_ranges; mod subscribe_tracks; diff --git a/rs/moq-net/tests/integration/stream_credit.rs b/rs/moq-net/tests/integration/stream_credit.rs new file mode 100644 index 0000000000..2fac6daaf9 --- /dev/null +++ b/rs/moq-net/tests/integration/stream_credit.rs @@ -0,0 +1,67 @@ +#![cfg(feature = "async")] + +//! A peer granting one uni stream at a time still receives every group, on each protocol. + +use std::time::Duration; + +use moq_net::{Hop, Timestamp, Version, origin, track}; + +use crate::support::harness::{MockConnectOptions, connect_mock, spawn}; + +fn origin(hop: u64) -> origin::Producer { + let (origin, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); + spawn(driver); + origin +} + +#[moq_clock::test] +async fn one_stream_credit_serves_every_backlogged_group() { + for version in ["moq-lite-06", "moq-lite-07-wip", "moq-transport-22"] { + let version: Version = version.parse().unwrap(); + let publisher = origin(1); + let viewer = origin(2); + let broadcast = publisher.create_broadcast("live").unwrap(); + let info = track::Info::default().with_timescale(moq_net::Timescale::MILLI); + let tracks: Vec<_> = (0..2) + .map(|i| broadcast.create_track(format!("{i}"), info.clone()).unwrap()) + .collect(); + broadcast.announce(Default::default()).unwrap(); + + let mut options = MockConnectOptions::new(version); + // Draft 22 keeps its SETUP stream open, so it needs one more for groups. + options.transport.max_concurrent_uni_streams = Some(1 + usize::from(version.is_ietf())); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(viewer.clone()); + let _pair = connect_mock(options).await; + + let remote = viewer.consume().routed_broadcast("live").await.unwrap(); + let prefs = track::Subscription::default().with_max_delay(Duration::from_secs(30)); + let mut subscribers = Vec::new(); + for i in 0..tracks.len() { + let track = remote.track(&format!("{i}")).unwrap(); + subscribers.push(track.subscribe(prefs.clone()).await.unwrap()); + } + + for _ in 0..4 { + for track in &tracks { + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap(); + group.finish().unwrap(); + } + } + for subscriber in &mut subscribers { + // The transport, not the publisher, picks which parked open gets the next grant. + let mut received = Vec::new(); + while received.len() < 4 { + let group = moq_clock::timeout(Duration::from_secs(5), subscriber.recv_group()) + .await + .unwrap_or_else(|_| panic!("{version}: only {received:?} arrived")) + .unwrap() + .unwrap(); + received.push(group.sequence); + } + received.sort(); + assert_eq!(received, [0, 1, 2, 3], "{version}"); + } + } +} diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index 69ac35ea3d..69defd4cec 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -1373,11 +1373,22 @@ async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq } } -/// A finite deep queue, below the media bitrate, must shed superseded groups while -/// preserving the newest catalog snapshot. Random loss is disabled in this acceptance test. #[tokio::test] async fn ietf_subscriber_max_age_skips_bufferbloat() { - let version: moq_net::Version = "moq-transport-22".parse().unwrap(); + subscriber_max_age_skips_bufferbloat("moq-transport-22").await +} + +#[tokio::test] +async fn lite_subscriber_max_age_skips_bufferbloat() { + subscriber_max_age_skips_bufferbloat("moq-lite-06").await +} + +/// A finite deep queue, below the media bitrate, must shed superseded groups while +/// preserving the newest catalog snapshot. Random loss is disabled in this acceptance test. +/// On both protocols: this is the benign twin of a subscriber that stalls on purpose, so +/// bounding what a stalled one makes the relay hold must not cost it the live edge. +async fn subscriber_max_age_skips_bufferbloat(version: &str) { + let version: moq_net::Version = version.parse().unwrap(); let mut config = relay_config(); config.listen.version = vec![version]; let relay = RelayHost::start(config).await; diff --git a/test/drill/README.md b/test/drill/README.md index b08c19e9b0..cfb2a12309 100644 --- a/test/drill/README.md +++ b/test/drill/README.md @@ -32,7 +32,7 @@ checks that what the fault owned was released. | `bursts_cross_a_cluster` | Bursts of one-frame groups cross from an origin relay to a clustered edge relay; the subscriber reads newest-only and FETCHes every group it misses | how many groups missed the live subscription, which live groups were reset, the completed ingress burst and overflow count, and the slowest group's latency | every group arrives, live or by FETCH, within 10s of being written; a lost group fails by name (FETCH failed, FETCH unanswered, live group stalled), and the newest group comes live | | `bursts_cross_a_flapping_peer` | The same bursts, with the peer link cut as the second burst is written and restored once the edge withdraws the route | how long the link was down before the withdrawal, and the reads and FETCHes the flap failed | the route comes back and every group still arrives within 10s of being written, or of the route's return when it arrives after it; only a failure the flap caused is retried, on the restored route, so a hang on the dead one past the relays' idle timeout or a withdrawal with the link up still fails | | `bursts_cross_a_transport_peer` | The same bursts, with the relays peering over `moq-transport-19`, so the edge recovers gaps with IETF FETCHes | as `bursts_cross_a_cluster` | as `bursts_cross_a_cluster`: a group the origin still holds is served, never refused | -| `ietf_subscriber_max_age_skips_bufferbloat` | moq-transport-22 media at 5.12Mb/s crosses a 64kb/s link behind a finite 500ms queue, with no random loss | the propagated 100ms budget and forced queue overflow | superseded groups are skipped and the newest catalog snapshot survives its budget | +| `ietf_subscriber_max_age_skips_bufferbloat`, `lite_subscriber_max_age_skips_bufferbloat` | moq-transport-22 or moq-lite-06 media at 5.12Mb/s crosses a 64kb/s link behind a finite 500ms queue, with no random loss | the propagated 100ms budget and forced queue overflow | superseded groups are skipped and the newest catalog snapshot survives its budget | | `no_publisher_never_delivers` | none: the negative control | - | with nothing publishing, nothing is announced and no broadcast resolves | A moq publisher is never blocked by a slow reader: `write_frame` is synchronous