From ee4ee8ea08b1f145d5e7696cd5a6d96b025e5f80 Mon Sep 17 00:00:00 2001 From: sletmoe Date: Fri, 6 Nov 2026 04:35:43 +0000 Subject: [PATCH 1/7] feat(net): let a refusal carry a retry interval Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/request_stream.test.ts | 17 ++ js/net/src/ietf/request_stream.ts | 5 + rs/moq-net/src/error.rs | 53 ++++++ rs/moq-net/src/ietf/publisher.rs | 225 ++++++++++++++++++++++--- rs/moq-net/src/ietf/request_stream.rs | 43 ++++- rs/moq-net/src/model/origin.rs | 136 ++++++++++++--- rs/moq-net/src/model/track.rs | 33 +++- 7 files changed, 452 insertions(+), 60 deletions(-) diff --git a/js/net/src/ietf/request_stream.test.ts b/js/net/src/ietf/request_stream.test.ts index aab85a684cb..892eec7299a 100644 --- a/js/net/src/ietf/request_stream.test.ts +++ b/js/net/src/ietf/request_stream.test.ts @@ -167,6 +167,23 @@ test("a refusal carries its reason", async () => { } }); +test("a refusal carries its retry interval, rounded up, and 0 without one", async () => { + for (const version of [Version.DRAFT_18, Version.DRAFT_20, Version.DRAFT_22]) { + for (const [retryAfter, expected] of [ + [undefined, 0n], + [Milli(1500), 1500n], + [Milli(0.25), 1n], + ] as const) { + const { stream, wire } = recording(version); + await refuse(stream, version, "fetch", { ...refusal, retryAfter }); + const r = wire(); + expect(await r.u53()).toBe(RequestError.id); + const err = await RequestError.decode(r, version); + expect(err.retryInterval).toBe(expected); + } + } +}); + test("a request's deadline ends it through the request, so later ends do nothing", async () => { const clock = new Manual(); const installed = clock.install(); diff --git a/js/net/src/ietf/request_stream.ts b/js/net/src/ietf/request_stream.ts index fb27a97ee3a..58ec0520daa 100644 --- a/js/net/src/ietf/request_stream.ts +++ b/js/net/src/ietf/request_stream.ts @@ -1,3 +1,4 @@ +import type { Milli } from "@moq/clock"; import { error, ProtocolViolation, StreamCode, StreamError } from "../error.ts"; import type * as Path from "../path.ts"; import { Reader, type Stream, type Writer } from "../stream.ts"; @@ -30,12 +31,16 @@ function finCancels(version: IetfVersion): boolean { export interface Refusal { condition: RequestCondition; reasonPhrase: string; + /** How long the peer may wait before retrying; absent or 0 asks it not to. */ + retryAfter?: Milli; } async function writeRefusal(w: Writer, version: IetfVersion, kind: RequestKind, refusal: Refusal): Promise { const err = new RequestError({ errorCode: toRequestCode(refusal.condition, kind, version), reasonPhrase: refusal.reasonPhrase, + // Rounded up to whole milliseconds, since 0 tells the peer never to retry. + retryInterval: BigInt(Math.ceil(refusal.retryAfter ?? 0)), }); await Message.write(w, RequestError.id, err, version); } diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index e305e025b6c..58b9a90bdc2 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -596,6 +596,44 @@ impl Error { } } +/// A refusal: the error a request is refused with, and how long the peer may wait +/// before asking again. +/// +/// Every refusal API takes `impl Into`, so a bare [`Error`] is a refusal with +/// no retry hint. The error alone decides what the refusal means; `retry_after` only +/// reaches moq-transport's REQUEST_ERROR, as its Retry Interval. moq-lite has no such +/// field and ignores it. +#[derive(Clone, Debug)] +#[non_exhaustive] +pub struct Refusal { + /// Why the request is refused. + pub error: Error, + /// How long the peer may wait before retrying, or `None` if it should not. + pub retry_after: Option, +} + +impl Refusal { + /// Refuse with `error`, telling the peer it may retry once `after` has passed. + /// + /// A zero `after` is no hint: moq-transport reads a zero interval as "do not retry". + pub fn retry(error: Error, after: std::time::Duration) -> Self { + Self { + error, + retry_after: (!after.is_zero()).then_some(after), + } + } +} + +/// A refusal with no retry hint. +impl From for Refusal { + fn from(error: Error) -> Self { + Self { + error, + retry_after: None, + } + } +} + /// Preserve the session registry when carrying a protocol error. impl From for Error { fn from(err: SessionError) -> Self { @@ -763,6 +801,21 @@ pub type Result = std::result::Result; mod tests { use super::*; + /// A retry hint rides beside the error, which stays exactly what was refused, and a + /// zero hint is no hint. + #[test] + fn a_refusal_keeps_its_error() { + let after = std::time::Duration::from_secs(1); + let hinted = Refusal::retry(Error::GoingAway, after); + assert!(matches!(hinted.error, Error::GoingAway)); + assert_eq!(hinted.retry_after, Some(after)); + assert_eq!( + Refusal::retry(Error::NotFound, std::time::Duration::ZERO).retry_after, + None + ); + assert_eq!(Refusal::from(Error::NotFound).retry_after, None); + } + // Both registries are wire contracts now (draft-lcurley-moq-lite, Error Codes), so // every code we send must decode back to what we meant. #[test] diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index d5394ffbe45..65e78f45085 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -83,13 +83,20 @@ fn track_properties( /// The rendezvous cap is zero, but asking to wait still makes a missing publisher a timeout. /// Once the broadcast resolves, a missing track is the publisher's answer and stays DOES_NOT_EXIST. -fn rendezvous_error(error: Error, requested: bool, resolved: bool) -> Error { - let missing = match &error { +/// The timeout keeps the refusal's retry hint: it still says when to ask again. +fn rendezvous_error(refusal: crate::Refusal, requested: bool, resolved: bool) -> crate::Refusal { + let missing = match &refusal.error { Error::Unroutable | Error::Stream(crate::StreamError::Unroutable) => true, Error::NotFound | Error::Stream(crate::StreamError::NotFound) => !resolved, _ => false, }; - if requested && missing { Error::Timeout } else { error } + match requested && missing { + true => crate::Refusal { + error: Error::Timeout, + ..refusal + }, + false => refusal, + } } #[cfg_attr( @@ -782,8 +789,8 @@ struct SubscriptionUpdates { /// How an admitted subscription came out of serving. enum Subscribed { - /// It never started, refused with this reason and error before SUBSCRIBE_OK. - Refused(String, Error), + /// It never started, refused with this reason and refusal before SUBSCRIBE_OK. + Refused(String, crate::Refusal), /// The subscriber withdrew it, or the session closed, before SUBSCRIBE_OK. Withdrawn, /// It was served, and opened this many group streams. @@ -1244,8 +1251,13 @@ impl SubscribeRequest { /// Refuse the request with `err`, delivering its REQUEST_ERROR next. fn refuse(&mut self, err: &Error, reason: &str) -> Result { + self.refuse_with(&err.clone().into(), reason) + } + + /// [`Self::refuse`], carrying the refusal's retry hint. + fn refuse_with(&mut self, refusal: &crate::Refusal, reason: &str) -> Result { let stream = self.stream.as_mut().expect("open until PUBLISH_DONE"); - let deliver = self.publisher.reject_subscribe(&mut stream.writer, err, reason)?; + let deliver = self.publisher.reject_subscribe(&mut stream.writer, refusal, reason)?; Ok(SubscribeState::Refuse(deliver)) } @@ -1315,7 +1327,7 @@ impl SubscribeRequest { } } Ok(Ok(Subscribed::Served(served, streams))) => self.publish_done(served, streams, retain), - Ok(Ok(Subscribed::Refused(reason, err))) => self.refuse(&err, &reason)?, + Ok(Ok(Subscribed::Refused(reason, refusal))) => self.refuse_with(&refusal, &reason)?, // Nobody is left to answer. Ok(Ok(Subscribed::Withdrawn)) => SubscribeState::Release, Ok(Err(err)) => return Err(err), @@ -1452,8 +1464,11 @@ impl Serving { /// Stats (subscriptions, viewer refcount, groups/frames/bytes) are counted in the model, /// through the tagged `origin::Consumer` the broadcast resolves from. fn poll(&mut self, waiter: &kio::Waiter, request: &mut SubscribeRequest) -> Poll> { - let refused = |err: Error, rendezvous, resolved| { - Subscribed::Refused(err.to_string(), rendezvous_error(err, rendezvous, resolved)) + let refused = |refusal: crate::Refusal, rendezvous, resolved| { + Subscribed::Refused( + refusal.error.to_string(), + rendezvous_error(refusal, rendezvous, resolved), + ) }; let version = request.publisher.version; loop { @@ -1465,12 +1480,18 @@ impl Serving { // The reason is the origin's, not ours: a dynamic router refusing on // authorization or a drain says so, and only an unroutable path means the // broadcast is not here. - Err(err) => return Poll::Ready(Ok(refused(err, request.msg.rendezvous, false))), + Err(error) => { + let refusal = crate::Refusal { + error, + retry_after: resolve.retry_after, + }; + return Poll::Ready(Ok(refused(refusal, request.msg.rendezvous, false))); + } Ok(None) => return Poll::Ready(Ok(Subscribed::Withdrawn)), }; let track = match broadcast.track(&request.msg.track_name) { Ok(track) => track, - Err(err) => return Poll::Ready(Ok(refused(err, request.msg.rendezvous, true))), + Err(err) => return Poll::Ready(Ok(refused(err.into(), request.msg.rendezvous, true))), }; self.broadcast = Some(broadcast); self.step = ServingStep::Subscribe { @@ -1483,7 +1504,13 @@ impl Serving { ready!(subscribe.poll(waiter, &request.publisher, stream, track, &request.msg, &request.lapse)); let (serve, edge) = match subscribed { Some(Ok(subscribed)) => subscribed, - Some(Err(err)) => return Poll::Ready(Ok(refused(err, request.msg.rendezvous, true))), + Some(Err(error)) => { + let refusal = crate::Refusal { + error, + retry_after: subscribe.retry_after, + }; + return Poll::Ready(Ok(refused(refusal, request.msg.rendezvous, true))); + } None => return Poll::Ready(Ok(Subscribed::Withdrawn)), }; let ServingStep::Subscribe { track: cache, .. } = @@ -1505,7 +1532,7 @@ impl Serving { let properties = match track_properties(serve.track.info(), request.msg.properties_wanted, version, flate) { Ok(properties) => properties, - Err(err) => return Poll::Ready(Ok(Subscribed::Refused(err.to_string(), err))), + Err(err) => return Poll::Ready(Ok(Subscribed::Refused(err.to_string(), err.into()))), }; request.save_join(self.filter, edge.largest, &cache, &serve.ranking); @@ -1762,12 +1789,15 @@ impl SubscribeRequest { /// the request, since the producer may never answer. struct SubscribeTrack { state: SubscribeTrackState, + /// The retry hint the track's refusal carried, once it refused. + retry_after: Option, } #[kio::machine(self.state: SubscribeTrackState)] impl SubscribeTrack { fn new() -> Self { Self { + retry_after: None, state: SubscribeTrackState::Start, } } @@ -1825,7 +1855,10 @@ impl SubscribeTrack { let mut subscribed: track::Subscriber = match answer { // A peer reads the end of the subscription as the track's. Some(Ok(found)) => found.parked(), - Some(Err(err)) => return Poll::Ready(Some(Err(err))), + Some(Err(err)) => { + self.retry_after = subscribing.retry_after(); + return Poll::Ready(Some(Err(err))); + } None => return Poll::Ready(None), }; @@ -3236,10 +3269,10 @@ impl Publisher { fn reject_subscribe( &self, writer: &mut Writer, - err: &Error, + refusal: &crate::Refusal, reason: &str, ) -> Result { - self.buffer_subscribe_error(writer, err, reason)?; + super::request_stream::buffer_refusal(writer, request::Kind::Subscribe, refusal, reason, self.version)?; Ok(super::request_stream::Deliver::new()) } @@ -4213,6 +4246,8 @@ struct Resolve { session: Transport, fin_cancels: bool, path: crate::Path, + /// The retry hint the broadcast's refusal carried, once it refused. + retry_after: Option, state: ResolveState, } @@ -4223,6 +4258,7 @@ impl Resolve { session, path, fin_cancels, + retry_after: None, state: ResolveState::Start, } } @@ -4244,9 +4280,15 @@ impl Resolve { let requesting: kio::Pending = origin.lookup(self.path.clone(), None); let found = wait!(Lookup => match publisher.poll_withdrawn(&mut self.session, stream, self.fin_cancels, waiter) { Poll::Ready(()) => Poll::Ready(None), - Poll::Pending => requesting.poll_ok(waiter).map(Some), + Poll::Pending => requesting.poll_refused(waiter).map(Some), }); - Poll::Ready(found.transpose()) + Poll::Ready(match found.transpose() { + Ok(found) => Ok(found), + Err(refusal) => { + self.retry_after = refusal.retry_after; + Err(refusal.error) + } + }) } } @@ -4674,9 +4716,14 @@ impl FetchServe { /// Refuse the FETCH with `err`. See [`Publisher::reject_subscribe`] for why the close is /// not optional. fn reject(&mut self, err: &Error, reason: &str) -> Result<(), Error> { + self.reject_with(&err.clone().into(), reason) + } + + /// [`Self::reject`], carrying the refusal's retry hint. + fn reject_with(&mut self, refusal: &crate::Refusal, reason: &str) -> Result<(), Error> { let stream = self.stream.as_mut().expect("refused before answering"); let kind = request::Kind::Fetch; - super::request_stream::buffer_error(&mut stream.writer, kind, err, reason, self.publisher.version)?; + super::request_stream::buffer_refusal(&mut stream.writer, kind, refusal, reason, self.publisher.version)?; self.state = FetchState::Reject(super::request_stream::Deliver::new()); Ok(()) } @@ -4963,6 +5010,7 @@ impl FetchServe { Some(err) => Err(err), None => ready!(resolve.poll(waiter, &self.publisher, stream)), }; + let retry_after = resolve.retry_after; let FetchState::Resolve { gate, request, .. } = std::mem::replace(&mut self.state, FetchState::Done) else { @@ -4972,7 +5020,14 @@ impl FetchServe { Ok(Some(broadcast)) => broadcast, Ok(None) => return Poll::Ready(Ok(())), Err(err) => { - self.reject(&err, &err.to_string())?; + let reason = err.to_string(); + self.reject_with( + &crate::Refusal { + error: err, + retry_after, + }, + &reason, + )?; continue; } }; @@ -5329,9 +5384,14 @@ impl TrackStatusServe { /// Refuse the TRACK_STATUS with `err`, delivered on the following polls. fn reject(&mut self, err: &Error) -> Result<(), Error> { + self.reject_with(&err.clone().into()) + } + + /// [`Self::reject`], carrying the refusal's retry hint. + fn reject_with(&mut self, refusal: &crate::Refusal) -> Result<(), Error> { let stream = self.stream.as_mut().expect("refused before answering"); - let (kind, reason) = (request::Kind::TrackStatus, err.to_string()); - super::request_stream::buffer_error(&mut stream.writer, kind, err, &reason, self.publisher.version)?; + let (kind, reason) = (request::Kind::TrackStatus, refusal.error.to_string()); + super::request_stream::buffer_refusal(&mut stream.writer, kind, refusal, &reason, self.publisher.version)?; self.state = TrackStatusState::Reject(super::request_stream::Deliver::new()); Ok(()) } @@ -5378,11 +5438,12 @@ impl TrackStatusServe { Some(err) => Err(err), None => ready!(resolve.poll(waiter, &self.publisher, stream)), }; + let retry_after = resolve.retry_after; let broadcast = match resolved { Ok(Some(broadcast)) => broadcast, Ok(None) => return Poll::Ready(Ok(())), - Err(err) => { - self.reject(&err)?; + Err(error) => { + self.reject_with(&crate::Refusal { error, retry_after })?; continue; } }; @@ -15445,6 +15506,122 @@ mod serve_tests { drop(claim); } + /// What an `answer` keeps alive until its refusal is on the wire. + type Held = Box; + + /// A SUBSCRIBE `answer` refuses, as the REQUEST_ERROR it writes: (code, Retry Interval). + async fn refused_subscribe( + version: Version, + rendezvous: bool, + answer: impl FnOnce(&origin::Producer) -> std::pin::Pin + '_>>, + ) -> (u64, u64) { + let ms = std::time::Duration::from_millis; + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let peer_setup = peer::PeerSetup::default(); + peer_setup.set(peer::Peer::default()); + let publisher = Publisher::new( + crate::time::sim(), + crate::driver::Transport::new(ScriptedSession::new(Vec::new())), + origin.consume(), + Control::new(false), + None, + peer_setup, + version, + ); + let request = ScriptedSession::new(Vec::new()); + let stream = Stream::open(&crate::driver::Transport::new(request.clone()), version) + .await + .unwrap(); + let mut msg = subscribe(Filter::NextObject, None); + msg.rendezvous = rendezvous; + let mut serving = std::pin::pin!(publisher.clone().run_subscribe_stream(stream, msg)); + let answering = answer(&origin); + futures::pin_mut!(answering); + let _held = match futures::future::select(answering, serving.as_mut()).await { + futures::future::Either::Left((held, _)) => held, + futures::future::Either::Right((res, _)) => panic!("{version}: ended unasked: {res:?}"), + }; + kio::clock::timeout(std::time::Duration::from_secs(10), async { + while request.log.writes.lock().unwrap().is_empty() && futures::poll!(serving.as_mut()).is_pending() { + kio::clock::sleep(ms(1)).await; + } + }) + .await + .unwrap_or_else(|_| panic!("{version}: the refusal never went out")); + + let wire = request.log.writes.lock().unwrap().clone(); + let mut buf = Decoder::new(&wire, version.into()); + assert_eq!(buf.varint().unwrap(), ietf::RequestError::ID, "{version}"); + let error = ietf::RequestError::decode(&mut buf, version).unwrap(); + (error.error_code, error.retry_interval) + } + + /// A router's refusal of a broadcast sends its retry hint as REQUEST_ERROR's Retry + /// Interval, rounded up to a millisecond so it never reads as "do not retry". A bare + /// refusal sends 0, and a rendezvous that times out keeps the hint. + #[kio::test] + async fn a_broadcast_refusal_carries_its_retry_interval() { + let ms = std::time::Duration::from_millis; + for version in ALL_VERSIONS { + for (refusal, rendezvous, expected) in [ + (crate::Refusal::from(Error::Unroutable), false, (0x10, 0)), + (crate::Refusal::retry(Error::Unroutable, ms(1500)), false, (0x10, 1500)), + ( + crate::Refusal::retry(Error::Unroutable, std::time::Duration::from_micros(1)), + false, + (0x10, 1), + ), + ( + crate::Refusal::retry(Error::Unroutable, std::time::Duration::ZERO), + false, + (0x10, 0), + ), + (crate::Refusal::retry(Error::Unauthorized, ms(250)), false, (0x1, 250)), + (crate::Refusal::retry(Error::Unroutable, ms(1500)), true, (0x2, 1500)), + ] { + let case = format!("{version}: {refusal:?} rendezvous={rendezvous}"); + let wire = refused_subscribe(version, rendezvous, |origin| { + Box::pin(async move { + let route = origin + .forward("room", crate::origin::RouteInfo::default(), None) + .unwrap(); + route.requested().await.unwrap().reject(refusal); + // The route stays up: only its answer refuses. + Box::new(route) as Held + }) + }) + .await; + assert_eq!(wire, expected, "{case}"); + } + } + } + + /// A dynamic track's rejection carries its retry hint the same way. + #[kio::test] + async fn a_track_refusal_carries_its_retry_interval() { + for version in ALL_VERSIONS { + for (refusal, expected) in [ + (crate::Refusal::from(Error::NotFound), (0x10, 0)), + ( + crate::Refusal::retry(Error::NotFound, std::time::Duration::from_secs(2)), + (0x10, 2000), + ), + ] { + let case = format!("{version}: {refusal:?}"); + let wire = refused_subscribe(version, false, |origin| { + Box::pin(async move { + let broadcast = origin.route("room", crate::origin::RouteOptions::default()).unwrap(); + let mut dynamic = broadcast.broadcast().requests(); + dynamic.requested_track().await.unwrap().reject(refusal); + Box::new((broadcast, dynamic)) as Held + }) + }) + .await; + assert_eq!(wire, expected, "{case}"); + } + } + } + /// A resume subscribes again as a new SUBSCRIBE does, so a subscriber withdrawing the /// request while the resume waits on an idle copy's live edge ends it at once too. #[kio::test] diff --git a/rs/moq-net/src/ietf/request_stream.rs b/rs/moq-net/src/ietf/request_stream.rs index c6bc9c98950..c8f55d0ba2b 100644 --- a/rs/moq-net/src/ietf/request_stream.rs +++ b/rs/moq-net/src/ietf/request_stream.rs @@ -315,7 +315,7 @@ impl Deliver { } } -/// Buffer the REQUEST_ERROR refusing a `kind` request with `err`. +/// Buffer the REQUEST_ERROR refusing a `kind` request with `err`, asking the peer not to retry. pub(super) fn buffer_error( writer: &mut Writer, kind: super::error::request::Kind, @@ -323,14 +323,39 @@ pub(super) fn buffer_error( reason: &str, version: Version, ) -> Result<(), Error> { - writer.buffer_message( - super::RequestError::ID, - &super::RequestError { - error_code: super::error::request::to_code(err, kind, version), - reason_phrase: reason.into(), - retry_interval: 0, - }, - ) + writer.buffer_message(ID, &request_error(kind, err, None, reason, version)) +} + +/// Buffer the REQUEST_ERROR refusing a `kind` request with `refusal`, carrying its retry hint. +pub(super) fn buffer_refusal( + writer: &mut Writer, + kind: super::error::request::Kind, + refusal: &crate::Refusal, + reason: &str, + version: Version, +) -> Result<(), Error> { + let error = request_error(kind, &refusal.error, refusal.retry_after, reason, version); + writer.buffer_message(ID, &error) +} + +const ID: u64 = super::RequestError::ID; + +/// The REQUEST_ERROR for a refusal. The Retry Interval rounds up to whole milliseconds, since +/// 0 tells the peer never to retry. +fn request_error<'a>( + kind: super::error::request::Kind, + err: &Error, + retry_after: Option, + reason: &'a str, + version: Version, +) -> super::RequestError<'a> { + super::RequestError { + error_code: super::error::request::to_code(err, kind, version), + reason_phrase: reason.into(), + retry_interval: retry_after.map_or(0, |after| { + u64::try_from(after.as_nanos().div_ceil(1_000_000)).unwrap_or(u64::MAX) + }), + } } /// A framed subscription update, preserving omitted preferences. diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 7949c6ef0e0..f18193b910d 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -19,7 +19,7 @@ use super::{ front::{Action, Candidate, Event, Front, Refusal}, }; use crate::{ - Error, Instance, InvalidPattern, Path, Pattern, Patterns, + Error, Instance, InvalidPattern, Path, Pattern, Patterns, Refusal as Refused, coding::{BoundsExceeded, Decode, DecodeError, Decoder, Encode, EncodeError, Encoder, Form, varint}, path::{self, Segment}, time::{Deadlines, Instant}, @@ -3687,6 +3687,8 @@ struct TrackIo { held: Option>, /// Whether the track had a reader as of the last demand edge. wanted: bool, + /// The retry hint of the last copy that refused the track. + retry_after: Option, } impl TrackIo { @@ -3711,18 +3713,23 @@ impl TrackIo { self.staged = None; self.copy = None; self.held = None; + let retry_after = self.retry_after.take(); match self.request.take() { // Ended before any route served it: nothing to read, nor to fetch. Some(request) => { self.routes.end(result.clone()); - request.reject(result.err().unwrap_or(Error::NotFound)); + let error = result.err().unwrap_or(Error::NotFound); + request.reject(Refused { error, retry_after }); } None => { if let Some(accepted) = self.accepted.take() { let _ = match &result { Ok(()) => accepted.finish(), // A failed track refuses newcomers too, so they ask afresh. - Err(err) => accepted.abort(err.clone()), + Err(err) => accepted.abort(Refused { + error: err.clone(), + retry_after, + }), }; } self.routes.end(result); @@ -3929,7 +3936,8 @@ fn pick<'a>( /// What [`FrontServe::poll_step`] returns: one thing that happened. enum FrontStep { Assigned(track::Request), - Resolved(Result), + /// The upstream answer, and the retry hint a rejection carried. + Resolved(Result, Option), SourceClosed(u64), Info(Arc, u64, Result), Ended(Arc, u64, Result<(), Error>, bool), @@ -3991,6 +3999,8 @@ struct FrontServe { ready: kio::Queue>, tracks: HashMap, TrackIo>, upstream: Option, + /// The retry hint the upstream refusal queued in `events` carried. + retry_after: Option, next_source: u64, /// The instance each source's route served when it was asked, and the table's clock /// then: what its content is from. @@ -4042,6 +4052,7 @@ impl FrontServe { ready: kio::Queue::new(), tracks: HashMap::new(), upstream: None, + retry_after: None, next_source: 0, instances: HashMap::new(), sources: HashMap::new(), @@ -4175,15 +4186,15 @@ impl FrontServe { } if let Some(upstream) = &self.upstream && let Poll::Ready(result) = upstream.pending.poll(waiter, |p| match &p.resolved { - Some(result) => Poll::Ready(result.clone()), + Some(result) => Poll::Ready((result.clone(), p.retry_after)), None => Poll::Pending, }) { - return Poll::Ready(FrontStep::Resolved(match result { - Ok(resolved) => resolved, + return Poll::Ready(match result { + Ok((resolved, retry_after)) => FrontStep::Resolved(resolved, retry_after), // The queue died unresolved (its handler dropped): the route // could not serve. - Err(_closed) => Err(Error::Unroutable), - })); + Err(_closed) => FrontStep::Resolved(Err(Error::Unroutable), None), + }); } // The listing first: a source ends its listing before it closes, so this forwards // the tracks it finished last before its close detaches it. @@ -4323,6 +4334,11 @@ impl kio::Task for FrontServe { .. } ); + // Only the standing refusal that ends the front passes its hint on. + let retry_after = match &event { + Event::Resolved { .. } => self.retry_after.take().filter(|_| refused), + _ => None, + }; let answered = match &event { Event::TrackInfo { track, source, .. } => Some((track.clone(), *source)), _ => None, @@ -4603,6 +4619,9 @@ impl kio::Task for FrontServe { }); if let Ok(mut pending) = self.request.write() { pending.refused = refused; + if pending.resolved.is_none() { + pending.retry_after = retry_after; + } pending.resolved.get_or_insert(Err(err.clone())); } // Ending the broadcast only retracts it: no new requesters or @@ -4701,6 +4720,7 @@ impl kio::Task for FrontServe { copy: None, held: None, wanted: false, + retry_after: None, }, ); Event::TrackAssigned { @@ -4708,7 +4728,7 @@ impl kio::Task for FrontServe { now: self.timers.now(), } } - FrontStep::Resolved(result) => { + FrontStep::Resolved(result, retry_after) => { let Some(FrontUpstream { route, asked, @@ -4750,13 +4770,16 @@ impl kio::Task for FrontServe { self.instances.insert(id, (asked, asked_at)); Event::Resolved { route, result: Ok(id) } } - Err(err) => Event::Resolved { - route, - result: Err(Refusal { - err, - standing: self.standing(route), - }), - }, + Err(err) => { + self.retry_after = retry_after; + Event::Resolved { + route, + result: Err(Refusal { + err, + standing: self.standing(route), + }), + } + } } } FrontStep::SourceClosed(source) => Event::SourceClosed { source }, @@ -4774,6 +4797,8 @@ impl kio::Task for FrontServe { }, Err(err) => Err(err), }; + // What the refusing copy said about retrying, should it end the track. + io.retry_after = result.as_ref().err().and(asked.query.retry_after()); // Staged only while the track has a reader: without one the machine // will not splice, and a held copy would keep the source subscribed. if let Ok(info) = &result @@ -6214,6 +6239,8 @@ impl OriginState { #[derive(Default)] struct PendingBroadcast { resolved: Option>, + /// The retry hint a handler's rejection carried. + retry_after: Option, } /// A front's verdict, which its requesters wait on: the instance it resolved, or the @@ -6236,6 +6263,8 @@ struct PendingFront { selected: Instance, /// The selected serving handler refused, rather than the route disappearing. refused: bool, + /// The retry hint that refusal carried. + retry_after: Option, } impl PendingFront { @@ -6351,12 +6380,16 @@ impl ServeState { path: &Path, producer: &kio::Producer, result: Result, + retry_after: Option, ) -> bool { let mut state = shared.lock(); let Some(resolved) = state.answer(path, producer, result) else { return false; }; if let Ok(mut pending) = producer.write() { + if pending.resolved.is_none() { + pending.retry_after = retry_after; + } pending.resolved.get_or_insert(resolved); drop(state); } @@ -6491,13 +6524,16 @@ impl Request { /// use [`Self::route`] to announce its own instance. pub fn serve(self, broadcast: impl Consume) { let broadcast = broadcast.consume(); - let _ = ServeState::resolve(&self.home, &self.path, &self.producer, Ok(broadcast)); + let _ = ServeState::resolve(&self.home, &self.path, &self.producer, Ok(broadcast), None); // `self.producer` drops here, closing the channel; the value is still observable. } - /// Reject the request, resolving every awaiting requester with `err`. - pub fn reject(self, err: Error) { - let _ = ServeState::resolve(&self.home, &self.path, &self.producer, Err(err)); + /// Reject the request, resolving every awaiting requester with the refusal's error. + /// + /// A bare [`Error`] converts; see [`Refusal`](crate::Refusal) to add a retry hint. + pub fn reject(self, refusal: impl Into) { + let Refused { error, retry_after } = refusal.into(); + let _ = ServeState::resolve(&self.home, &self.path, &self.producer, Err(error), retry_after); } } @@ -6588,11 +6624,29 @@ impl Lookup { /// Poll for the requested broadcast without blocking. pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll> { + self.poll_refused(waiter).map_err(|refused| refused.error) + } + + /// [`Self::poll_ok`], keeping the retry hint a handler's refusal carried. + pub(crate) fn poll_refused(&self, waiter: &kio::Waiter) -> Poll> { + let mut retry_after = None; + let found = ready!(self.poll_found(waiter, &mut retry_after)); + Poll::Ready(found.map_err(|error| Refused { error, retry_after })) + } + + fn poll_found( + &self, + waiter: &kio::Waiter, + retry_after: &mut Option, + ) -> Poll> { match &self.inner { RequestState::Failed(error) => Poll::Ready(Err(error.clone())), RequestState::Pending { front, broadcast } => Poll::Ready( match ready!(front.poll(waiter, |state| match &state.resolved { - Some(result) => Poll::Ready(result.clone().map(|instance| (instance, state.first.clone())),), + Some(result) => { + *retry_after = state.retry_after; + Poll::Ready(result.clone().map(|instance| (instance, state.first.clone()))) + } None => Poll::Pending, })) { // Another instance than the one named, and not one continuing it: never hand @@ -7520,6 +7574,7 @@ impl Consumer { first: None, selected: best, refused: false, + retry_after: None, }); let consumer = request.consume(); // The front starts held by this request; see [`Front::new`]. @@ -8304,6 +8359,7 @@ mod tests { copy: None, held: None, wanted: true, + retry_after: None, }; let mut subscriber = consumer.subscribe(None).now_or_never().unwrap().unwrap(); io.end(Ok(())); @@ -10583,6 +10639,7 @@ mod tests { first: None, selected: instance(), refused: false, + retry_after: None, }), timers: consumer.timers.clone(), }; @@ -14142,6 +14199,41 @@ mod tests { (producer, server, upstream, old, subscription) } + /// A retry hint on a carried copy's abort changes nothing the front decides: a + /// `GoingAway` copy that delivered is asked for again from the same source, hinted or not. + #[kio::test] + #[cfg(feature = "async")] + async fn a_retry_hint_keeps_a_going_away_copy_retryable() { + for refusal in [ + crate::Refusal::from(Error::GoingAway), + crate::Refusal::retry(Error::GoingAway, Duration::from_secs(1)), + ] { + let hinted = refusal.retry_after.is_some(); + let (_producer, _server, upstream, old, mut subscription) = remote_front().await; + let mut handler = upstream.requests(); + old.abort(refusal).unwrap(); + + let request = kio::clock::timeout(Duration::from_secs(1), handler.requested_track()) + .await + .unwrap_or_else(|_| panic!("hinted={hinted}: the front never asked the source again")) + .expect("request"); + let track = request.accept(crate::track::Info::timed()); + // The copy resumes past the group the reader already has. + let mut group = track.create_group(crate::group::Info { sequence: 1 }).unwrap(); + group.write_frame(crate::Timestamp::ZERO, b"after".as_ref()).unwrap(); + group.finish().unwrap(); + let mut group = next_group(&mut subscription) + .await + .unwrap_or_else(|err| panic!("hinted={hinted}: the track ended: {err}")) + .expect("track open"); + assert_eq!( + &group.read_frame().await.unwrap().unwrap().payload[..], + b"after", + "hinted={hinted}" + ); + } + } + /// A local newcomer takes over while its answer for "video" is pending, so the track /// drains the incumbent. The reader leaves: the incumbent's copy is dropped at once, /// like any unread track's, not held until the linger. diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 1e869a15af6..9ef97ccc5bc 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -549,6 +549,8 @@ pub(crate) struct TrackState { // The error that caused the track to be aborted, if any. abort: Option, + // The retry hint the abort or rejection carried, for a request it refuses. + retry_after: Option, // Whether the declared end settled before a write that could reopen it or an abort, or // `closed()` reported it: every group below it was produced and finished then, so the @@ -3906,10 +3908,16 @@ impl Producer { /// /// [`finish`](Self::finish) is deliberately not terminal: it declares the final /// sequence, and lower-numbered groups may still be written afterwards. - pub fn abort(self, err: Error) -> Result<()> { + /// + /// A bare [`Error`] converts; see [`Refusal`](crate::Refusal) to tell a request it + /// refuses when to retry. + pub fn abort(self, refusal: impl Into) -> Result<()> { + let crate::Refusal { error, retry_after } = refusal.into(); // Unlisted first, so the track leaves the listing before its name is free again. self.alive.unlist(false); - commit_abort(self.modify()?, err).abort(); + let mut state = self.modify()?; + state.retry_after = retry_after; + commit_abort(state, error).abort(); Ok(()) } @@ -5523,6 +5531,12 @@ impl Subscribing { } } + /// The retry hint the track's abort or rejection carried. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn retry_after(&self) -> Option { + self.state.read().retry_after + } + /// Poll until the peer confirms the subscription, yielding the [`Subscriber`]. /// Errors if the track is aborted or not found. pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll> { @@ -5587,6 +5601,11 @@ pub struct Resolving { } impl Resolving { + /// The retry hint the track's abort or rejection carried. + pub(crate) fn retry_after(&self) -> Option { + self.state.read().retry_after + } + /// A handle on the track instance this resolves, whatever its broadcast serves later. pub fn track(&self) -> Consumer { Consumer { @@ -9458,10 +9477,14 @@ impl Request { } } - /// Reject the request, waking all waiting subscribers with `err`. - pub fn reject(self, err: Error) { + /// Reject the request, waking all waiting subscribers with the refusal's error. + /// + /// A bare [`Error`] converts; see [`Refusal`](crate::Refusal) to add a retry hint. + pub fn reject(self, refusal: impl Into) { + let crate::Refusal { error, retry_after } = refusal.into(); if let Ok(mut state) = self.state.write() { - state.abort = Some(err); + state.abort = Some(error); + state.retry_after = retry_after; } } From cb8e56bbdadfaeaf74fb4787a67ae4a3edb80fe0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 23:56:46 -0800 Subject: [PATCH 2/7] feat(net): derive REQUEST_ERROR's Retry Interval from the refusal Rework the retry hint into the IETF layer: drop the public Refusal type and its model state, and have the moq-transport encoder invite a retry only for an error asking again may pass. Error::retryable() is the one classification, shared with the resume path. Co-Authored-By: Claude Opus 5.5 --- js/net/src/auth_session.ts | 11 +- js/net/src/ietf/publisher.test.ts | 33 +++ js/net/src/ietf/publisher.ts | 17 +- js/net/src/ietf/request_stream.test.ts | 17 -- js/net/src/ietf/request_stream.ts | 10 +- js/net/src/resume.ts | 4 +- rs/moq-net/src/auth.rs | 5 +- rs/moq-net/src/error.rs | 78 +++---- rs/moq-net/src/ietf/error.rs | 32 +++ rs/moq-net/src/ietf/namespace.rs | 15 +- rs/moq-net/src/ietf/publisher.rs | 271 +++++++++---------------- rs/moq-net/src/ietf/request.rs | 17 ++ rs/moq-net/src/ietf/request_stream.rs | 37 +--- rs/moq-net/src/ietf/session.rs | 14 +- rs/moq-net/src/model/front.rs | 2 +- rs/moq-net/src/model/origin.rs | 138 +++---------- rs/moq-net/src/model/resume.rs | 63 ++---- rs/moq-net/src/model/track.rs | 33 +-- 18 files changed, 290 insertions(+), 507 deletions(-) diff --git a/js/net/src/auth_session.ts b/js/net/src/auth_session.ts index 5a1af021d2b..8fe9e9f664c 100644 --- a/js/net/src/auth_session.ts +++ b/js/net/src/auth_session.ts @@ -50,9 +50,14 @@ export class Duplicate extends Error { } } -// How long a token refused for a reason that may pass waits before its session presents it -// again, doubling on each refusal up to RETRY_MAX. -const RETRY_INITIAL = Milli(1000); +/** + * How long a token refused for a reason that may pass waits before its session presents it + * again, doubling on each refusal up to RETRY_MAX. A moq-transport refusal that may pass asks + * its peer to wait as long. + * + * @internal + */ +export const RETRY_INITIAL = Milli(1000); const RETRY_MAX = Milli(30_000); // Whether a refusal is the acceptor failing to decide rather than its verdict on the token, diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 256cee7b9f4..b3cc1f99d52 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -418,6 +418,39 @@ test("TRACK_STATUS for a missing broadcast or track is refused DOES_NOT_EXIST", } }); +// A refused track invites a retry when asking again may pass, as in Rust, and asks the peer never +// to retry a verdict. +test("a TRACK_STATUS refusal that may pass invites a retry", async () => { + for (const version of IETF_VERSIONS) { + for (const [err, retryInterval] of [ + [new StreamError(StreamCode.GoingAway, { message: "draining" }), 1000n], + [new StreamError(StreamCode.NotFound, { message: "no video" }), 0n], + ] as const) { + const pair = createMockTransportPair(ALPNS[version]); + const session = new NativeSession(pair.server, version, true); + const { pub, origin } = publisher(pair.server, { session }); + const requests = origin.routePrefix(Path.from("room")).requested(); + + const status = requestStream(version); + const answering = pub.runTrackStatusRequest(trackStatus("room"), status.stream); + const request = (await requests.next()).value; + if (!request) throw new Error("no request for room"); + const served = new BroadcastProducer(); + const trackRequest = wireOf(served).requested(); + request.serve(served); + (await trackRequest)?.reject(err); + await answering; + + const wire = status.wire(); + expect(await wire.u53()).toBe(RequestError.id); + expect((await RequestError.decode(wire, version)).retryInterval).toBe(retryInterval); + await requests.return?.(); + served.close(); + origin.close(); + } + } +}); + // The requester leaving is owed nothing, and gives back the origin demand or the track // lookup it held, whichever was still resolving. test("a cancelled TRACK_STATUS releases its demand on every draft", async () => { diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 7bb17df7053..b6c17c3e606 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -25,6 +25,7 @@ import type { Consumer as OriginConsumer } from "../origin.ts"; import * as Path from "../path.ts"; import { sendPayload } from "../payload.ts"; import { Priority, sendOrder } from "../priority.ts"; +import { routeFailed } from "../resume.ts"; import { type Stream, Writer } from "../stream.ts"; import { Timescale } from "../time.ts"; import type { Groups, Info, Range, Position as TrackPosition, Subscriber as TrackSubscriber } from "../track.ts"; @@ -2544,16 +2545,16 @@ export class Publisher { } } - /** What a request that failed to resolve is refused with. */ + /** + * What a request that failed to resolve is refused with. A missing publisher a rendezvous + * waited for is a timeout, which, like Rust's `rendezvous_error`, invites no retry. + */ #failed(err: unknown, wait = false): Refusal { const e = error(err); - const condition = - e instanceof StreamError && (e.code === StreamCode.NotFound || e.code === StreamCode.Unroutable) - ? wait - ? "timeout" - : "does_not_exist" - : "internal"; - return { condition, reasonPhrase: reason(e) }; + const reasonPhrase = reason(e); + const missing = e instanceof StreamError && (e.code === StreamCode.NotFound || e.code === StreamCode.Unroutable); + if (missing && wait) return { condition: "timeout", reasonPhrase }; + return { condition: missing ? "does_not_exist" : "internal", reasonPhrase, retryable: routeFailed(e) }; } /** diff --git a/js/net/src/ietf/request_stream.test.ts b/js/net/src/ietf/request_stream.test.ts index 892eec7299a..aab85a684cb 100644 --- a/js/net/src/ietf/request_stream.test.ts +++ b/js/net/src/ietf/request_stream.test.ts @@ -167,23 +167,6 @@ test("a refusal carries its reason", async () => { } }); -test("a refusal carries its retry interval, rounded up, and 0 without one", async () => { - for (const version of [Version.DRAFT_18, Version.DRAFT_20, Version.DRAFT_22]) { - for (const [retryAfter, expected] of [ - [undefined, 0n], - [Milli(1500), 1500n], - [Milli(0.25), 1n], - ] as const) { - const { stream, wire } = recording(version); - await refuse(stream, version, "fetch", { ...refusal, retryAfter }); - const r = wire(); - expect(await r.u53()).toBe(RequestError.id); - const err = await RequestError.decode(r, version); - expect(err.retryInterval).toBe(expected); - } - } -}); - test("a request's deadline ends it through the request, so later ends do nothing", async () => { const clock = new Manual(); const installed = clock.install(); diff --git a/js/net/src/ietf/request_stream.ts b/js/net/src/ietf/request_stream.ts index 58ec0520daa..9409341f799 100644 --- a/js/net/src/ietf/request_stream.ts +++ b/js/net/src/ietf/request_stream.ts @@ -1,4 +1,4 @@ -import type { Milli } from "@moq/clock"; +import { RETRY_INITIAL } from "../auth_session.ts"; import { error, ProtocolViolation, StreamCode, StreamError } from "../error.ts"; import type * as Path from "../path.ts"; import { Reader, type Stream, type Writer } from "../stream.ts"; @@ -31,16 +31,16 @@ function finCancels(version: IetfVersion): boolean { export interface Refusal { condition: RequestCondition; reasonPhrase: string; - /** How long the peer may wait before retrying; absent or 0 asks it not to. */ - retryAfter?: Milli; + /** Whether asking again may pass, which invites the peer to retry. Absent is a verdict. */ + retryable?: boolean; } async function writeRefusal(w: Writer, version: IetfVersion, kind: RequestKind, refusal: Refusal): Promise { const err = new RequestError({ errorCode: toRequestCode(refusal.condition, kind, version), reasonPhrase: refusal.reasonPhrase, - // Rounded up to whole milliseconds, since 0 tells the peer never to retry. - retryInterval: BigInt(Math.ceil(refusal.retryAfter ?? 0)), + // 0 asks the peer never to retry. + retryInterval: refusal.retryable ? BigInt(RETRY_INITIAL) : 0n, }); await Message.write(w, RequestError.id, err, version); } diff --git a/js/net/src/resume.ts b/js/net/src/resume.ts index a7d3cf908a8..561070ea4a9 100644 --- a/js/net/src/resume.ts +++ b/js/net/src/resume.ts @@ -74,7 +74,7 @@ import { clamped, type Resolved, type Window } from "./window.ts"; import { type Broadcast as BroadcastWire, registerWire, wireOf } from "./wire.ts"; // Stream codes that report the route failing to deliver rather than judging what it carried: -// another copy may still deliver what it failed. Mirrors the Rust `route_failed`. +// another copy may still deliver what it failed. Mirrors the Rust `Error::retryable`. const ROUTE_FAILURES: ReadonlySet = new Set([ StreamCode.Internal, StreamCode.Cancel, @@ -113,7 +113,7 @@ function closing(err: Error): boolean { // A source losing its route to the content: only another route continues it, since asking the // same one again could only replace the withdrawal's cause with a refusal. Mirrors the Rust -// `unroutable`. +// `Error::unroutable`. function unroutable(err: Error): boolean { return err instanceof StreamError && err.code === StreamCode.Unroutable; } diff --git a/rs/moq-net/src/auth.rs b/rs/moq-net/src/auth.rs index 35b6955bf14..9b36d440c09 100644 --- a/rs/moq-net/src/auth.rs +++ b/rs/moq-net/src/auth.rs @@ -150,8 +150,9 @@ fn earliest(a: Option, b: Option) -> Option { } /// How long a token refused for a reason that may pass waits before its session presents it -/// again, doubling on each refusal up to [`RETRY_MAX`]. -const RETRY_INITIAL: Duration = Duration::from_secs(1); +/// again, doubling on each refusal up to [`RETRY_MAX`]. A moq-transport refusal that may pass +/// asks its peer to wait as long. +pub(crate) const RETRY_INITIAL: Duration = Duration::from_secs(1); const RETRY_MAX: Duration = Duration::from_secs(30); /// Whether a refusal is the acceptor failing to decide rather than its verdict on the diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index 58b9a90bdc2..1fb7652a95b 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -594,43 +594,40 @@ impl Error { _ => Self::Transport(err.to_string()), } } -} -/// A refusal: the error a request is refused with, and how long the peer may wait -/// before asking again. -/// -/// Every refusal API takes `impl Into`, so a bare [`Error`] is a refusal with -/// no retry hint. The error alone decides what the refusal means; `retry_after` only -/// reaches moq-transport's REQUEST_ERROR, as its Retry Interval. moq-lite has no such -/// field and ignores it. -#[derive(Clone, Debug)] -#[non_exhaustive] -pub struct Refusal { - /// Why the request is refused. - pub error: Error, - /// How long the peer may wait before retrying, or `None` if it should not. - pub retry_after: Option, -} + /// Whether asking again may succeed: the route failed to deliver rather than judging what + /// it carried, which would repeat on every attempt. Mirrors `routeFailed` in + /// `js/net/src/resume.ts`. + pub(crate) fn retryable(&self) -> bool { + self.unroutable() || self.transient() + } -impl Refusal { - /// Refuse with `error`, telling the peer it may retry once `after` has passed. - /// - /// A zero `after` is no hint: moq-transport reads a zero interval as "do not retry". - pub fn retry(error: Error, after: std::time::Duration) -> Self { - Self { - error, - retry_after: (!after.is_zero()).then_some(after), - } + /// Whether a source lost its route to the content: only another route continues it. + pub(crate) fn unroutable(&self) -> bool { + matches!(self, Self::Unroutable | Self::Stream(StreamError::Unroutable)) } -} -/// A refusal with no retry hint. -impl From for Refusal { - fn from(error: Error) -> Self { - Self { - error, - retry_after: None, - } + /// Whether the same source may deliver a fresh copy. An unroutable source needs a new + /// route instead: asking it again can replace the withdrawal's cause with a refusal. + pub(crate) fn transient(&self) -> bool { + matches!( + self, + Self::Transport(_) + | Self::IdleTimeout + | Self::Session(_) + | Self::GoingAway + | Self::GoawayTimeout + | Self::Dropped + | Self::Closed + | Self::Cancel + | Self::Stream( + StreamError::Session(_) + | StreamError::Internal + | StreamError::Cancel + | StreamError::GoingAway + | StreamError::ControlTimeout + ) + ) } } @@ -801,21 +798,6 @@ pub type Result = std::result::Result; mod tests { use super::*; - /// A retry hint rides beside the error, which stays exactly what was refused, and a - /// zero hint is no hint. - #[test] - fn a_refusal_keeps_its_error() { - let after = std::time::Duration::from_secs(1); - let hinted = Refusal::retry(Error::GoingAway, after); - assert!(matches!(hinted.error, Error::GoingAway)); - assert_eq!(hinted.retry_after, Some(after)); - assert_eq!( - Refusal::retry(Error::NotFound, std::time::Duration::ZERO).retry_after, - None - ); - assert_eq!(Refusal::from(Error::NotFound).retry_after, None); - } - // Both registries are wire contracts now (draft-lcurley-moq-lite, Error Codes), so // every code we send must decode back to what we meant. #[test] diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index fbf3ecbd519..b78f36b7b4a 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -248,6 +248,17 @@ pub(crate) mod request { registered.unwrap_or(INTERNAL_ERROR) } + /// The Retry Interval, in milliseconds, for a refusal with `err` that reads `code` on the wire. + /// + /// A refusal asking again may pass invites the backoff a refused token starts from. 0 asks the + /// peer never to retry: a verdict would repeat, and UNINTERESTED tells an offer to stop. + pub(crate) fn retry_interval(err: &Error, code: u64) -> u64 { + match err.retryable() && code != UNINTERESTED { + true => crate::auth::RETRY_INITIAL.as_millis() as u64, + false => 0, + } + } + /// Read a rejection code received on the negotiated draft. /// /// A code the draft does not assign for this request, INTERNAL_ERROR included, stays @@ -304,6 +315,27 @@ pub(crate) mod request { Error::Remote(0x30), ]; + /// Only a refusal that asking again may pass invites a retry. An unroutable offer reads + /// UNINTERESTED, which tells the peer to stop offering, so it never does. + #[test] + fn only_a_refusal_that_may_pass_invites_a_retry() { + let retry = crate::auth::RETRY_INITIAL.as_millis() as u64; + for version in ALL { + for kind in KINDS { + for err in EVERY_ERROR { + let code = to_code(&err, kind, version); + let expected = match err { + // The session failing under a request is no verdict on it. + Error::GoingAway | Error::Session(_) => retry, + Error::Unroutable if matches!(kind, Kind::Subscribe | Kind::Fetch | Kind::TrackStatus) => retry, + _ => 0, + }; + assert_eq!(retry_interval(&err, code), expected, "{version:?} {kind:?} {err:?}"); + } + } + } + } + /// A relay bridging a moq-lite revocation onto a moq-transport request refuses it as /// UNAUTHORIZED, the same as a local refusal. #[test] diff --git a/rs/moq-net/src/ietf/namespace.rs b/rs/moq-net/src/ietf/namespace.rs index 2de3c63f3fc..3df1294e884 100644 --- a/rs/moq-net/src/ietf/namespace.rs +++ b/rs/moq-net/src/ietf/namespace.rs @@ -175,15 +175,12 @@ pub(super) fn refusal(namespace: &Path, version: super::Version) -> Option crate::Refusal { - let missing = match &refusal.error { +fn rendezvous_error(error: Error, requested: bool, resolved: bool) -> Error { + let missing = match &error { Error::Unroutable | Error::Stream(crate::StreamError::Unroutable) => true, Error::NotFound | Error::Stream(crate::StreamError::NotFound) => !resolved, _ => false, }; - match requested && missing { - true => crate::Refusal { - error: Error::Timeout, - ..refusal - }, - false => refusal, - } + if requested && missing { Error::Timeout } else { error } } #[cfg_attr( @@ -789,8 +782,8 @@ struct SubscriptionUpdates { /// How an admitted subscription came out of serving. enum Subscribed { - /// It never started, refused with this reason and refusal before SUBSCRIBE_OK. - Refused(String, crate::Refusal), + /// It never started, refused with this reason and error before SUBSCRIBE_OK. + Refused(String, Error), /// The subscriber withdrew it, or the session closed, before SUBSCRIBE_OK. Withdrawn, /// It was served, and opened this many group streams. @@ -1251,13 +1244,8 @@ impl SubscribeRequest { /// Refuse the request with `err`, delivering its REQUEST_ERROR next. fn refuse(&mut self, err: &Error, reason: &str) -> Result { - self.refuse_with(&err.clone().into(), reason) - } - - /// [`Self::refuse`], carrying the refusal's retry hint. - fn refuse_with(&mut self, refusal: &crate::Refusal, reason: &str) -> Result { let stream = self.stream.as_mut().expect("open until PUBLISH_DONE"); - let deliver = self.publisher.reject_subscribe(&mut stream.writer, refusal, reason)?; + let deliver = self.publisher.reject_subscribe(&mut stream.writer, err, reason)?; Ok(SubscribeState::Refuse(deliver)) } @@ -1327,7 +1315,7 @@ impl SubscribeRequest { } } Ok(Ok(Subscribed::Served(served, streams))) => self.publish_done(served, streams, retain), - Ok(Ok(Subscribed::Refused(reason, refusal))) => self.refuse_with(&refusal, &reason)?, + Ok(Ok(Subscribed::Refused(reason, err))) => self.refuse(&err, &reason)?, // Nobody is left to answer. Ok(Ok(Subscribed::Withdrawn)) => SubscribeState::Release, Ok(Err(err)) => return Err(err), @@ -1464,11 +1452,8 @@ impl Serving { /// Stats (subscriptions, viewer refcount, groups/frames/bytes) are counted in the model, /// through the tagged `origin::Consumer` the broadcast resolves from. fn poll(&mut self, waiter: &kio::Waiter, request: &mut SubscribeRequest) -> Poll> { - let refused = |refusal: crate::Refusal, rendezvous, resolved| { - Subscribed::Refused( - refusal.error.to_string(), - rendezvous_error(refusal, rendezvous, resolved), - ) + let refused = |err: Error, rendezvous, resolved| { + Subscribed::Refused(err.to_string(), rendezvous_error(err, rendezvous, resolved)) }; let version = request.publisher.version; loop { @@ -1480,18 +1465,12 @@ impl Serving { // The reason is the origin's, not ours: a dynamic router refusing on // authorization or a drain says so, and only an unroutable path means the // broadcast is not here. - Err(error) => { - let refusal = crate::Refusal { - error, - retry_after: resolve.retry_after, - }; - return Poll::Ready(Ok(refused(refusal, request.msg.rendezvous, false))); - } + Err(err) => return Poll::Ready(Ok(refused(err, request.msg.rendezvous, false))), Ok(None) => return Poll::Ready(Ok(Subscribed::Withdrawn)), }; let track = match broadcast.track(&request.msg.track_name) { Ok(track) => track, - Err(err) => return Poll::Ready(Ok(refused(err.into(), request.msg.rendezvous, true))), + Err(err) => return Poll::Ready(Ok(refused(err, request.msg.rendezvous, true))), }; self.broadcast = Some(broadcast); self.step = ServingStep::Subscribe { @@ -1504,13 +1483,7 @@ impl Serving { ready!(subscribe.poll(waiter, &request.publisher, stream, track, &request.msg, &request.lapse)); let (serve, edge) = match subscribed { Some(Ok(subscribed)) => subscribed, - Some(Err(error)) => { - let refusal = crate::Refusal { - error, - retry_after: subscribe.retry_after, - }; - return Poll::Ready(Ok(refused(refusal, request.msg.rendezvous, true))); - } + Some(Err(err)) => return Poll::Ready(Ok(refused(err, request.msg.rendezvous, true))), None => return Poll::Ready(Ok(Subscribed::Withdrawn)), }; let ServingStep::Subscribe { track: cache, .. } = @@ -1532,7 +1505,7 @@ impl Serving { let properties = match track_properties(serve.track.info(), request.msg.properties_wanted, version, flate) { Ok(properties) => properties, - Err(err) => return Poll::Ready(Ok(Subscribed::Refused(err.to_string(), err.into()))), + Err(err) => return Poll::Ready(Ok(Subscribed::Refused(err.to_string(), err))), }; request.save_join(self.filter, edge.largest, &cache, &serve.ranking); @@ -1789,15 +1762,12 @@ impl SubscribeRequest { /// the request, since the producer may never answer. struct SubscribeTrack { state: SubscribeTrackState, - /// The retry hint the track's refusal carried, once it refused. - retry_after: Option, } #[kio::machine(self.state: SubscribeTrackState)] impl SubscribeTrack { fn new() -> Self { Self { - retry_after: None, state: SubscribeTrackState::Start, } } @@ -1855,10 +1825,7 @@ impl SubscribeTrack { let mut subscribed: track::Subscriber = match answer { // A peer reads the end of the subscription as the track's. Some(Ok(found)) => found.parked(), - Some(Err(err)) => { - self.retry_after = subscribing.retry_after(); - return Poll::Ready(Some(Err(err))); - } + Some(Err(err)) => return Poll::Ready(Some(Err(err))), None => return Poll::Ready(None), }; @@ -3269,10 +3236,10 @@ impl Publisher { fn reject_subscribe( &self, writer: &mut Writer, - refusal: &crate::Refusal, + err: &Error, reason: &str, ) -> Result { - super::request_stream::buffer_refusal(writer, request::Kind::Subscribe, refusal, reason, self.version)?; + self.buffer_subscribe_error(writer, err, reason)?; Ok(super::request_stream::Deliver::new()) } @@ -4246,8 +4213,6 @@ struct Resolve { session: Transport, fin_cancels: bool, path: crate::Path, - /// The retry hint the broadcast's refusal carried, once it refused. - retry_after: Option, state: ResolveState, } @@ -4258,7 +4223,6 @@ impl Resolve { session, path, fin_cancels, - retry_after: None, state: ResolveState::Start, } } @@ -4280,15 +4244,9 @@ impl Resolve { let requesting: kio::Pending = origin.lookup(self.path.clone(), None); let found = wait!(Lookup => match publisher.poll_withdrawn(&mut self.session, stream, self.fin_cancels, waiter) { Poll::Ready(()) => Poll::Ready(None), - Poll::Pending => requesting.poll_refused(waiter).map(Some), + Poll::Pending => requesting.poll_ok(waiter).map(Some), }); - Poll::Ready(match found.transpose() { - Ok(found) => Ok(found), - Err(refusal) => { - self.retry_after = refusal.retry_after; - Err(refusal.error) - } - }) + Poll::Ready(found.transpose()) } } @@ -4716,14 +4674,9 @@ impl FetchServe { /// Refuse the FETCH with `err`. See [`Publisher::reject_subscribe`] for why the close is /// not optional. fn reject(&mut self, err: &Error, reason: &str) -> Result<(), Error> { - self.reject_with(&err.clone().into(), reason) - } - - /// [`Self::reject`], carrying the refusal's retry hint. - fn reject_with(&mut self, refusal: &crate::Refusal, reason: &str) -> Result<(), Error> { let stream = self.stream.as_mut().expect("refused before answering"); let kind = request::Kind::Fetch; - super::request_stream::buffer_refusal(&mut stream.writer, kind, refusal, reason, self.publisher.version)?; + super::request_stream::buffer_error(&mut stream.writer, kind, err, reason, self.publisher.version)?; self.state = FetchState::Reject(super::request_stream::Deliver::new()); Ok(()) } @@ -5010,7 +4963,6 @@ impl FetchServe { Some(err) => Err(err), None => ready!(resolve.poll(waiter, &self.publisher, stream)), }; - let retry_after = resolve.retry_after; let FetchState::Resolve { gate, request, .. } = std::mem::replace(&mut self.state, FetchState::Done) else { @@ -5020,14 +4972,7 @@ impl FetchServe { Ok(Some(broadcast)) => broadcast, Ok(None) => return Poll::Ready(Ok(())), Err(err) => { - let reason = err.to_string(); - self.reject_with( - &crate::Refusal { - error: err, - retry_after, - }, - &reason, - )?; + self.reject(&err, &err.to_string())?; continue; } }; @@ -5384,14 +5329,9 @@ impl TrackStatusServe { /// Refuse the TRACK_STATUS with `err`, delivered on the following polls. fn reject(&mut self, err: &Error) -> Result<(), Error> { - self.reject_with(&err.clone().into()) - } - - /// [`Self::reject`], carrying the refusal's retry hint. - fn reject_with(&mut self, refusal: &crate::Refusal) -> Result<(), Error> { let stream = self.stream.as_mut().expect("refused before answering"); - let (kind, reason) = (request::Kind::TrackStatus, refusal.error.to_string()); - super::request_stream::buffer_refusal(&mut stream.writer, kind, refusal, &reason, self.publisher.version)?; + let (kind, reason) = (request::Kind::TrackStatus, err.to_string()); + super::request_stream::buffer_error(&mut stream.writer, kind, err, &reason, self.publisher.version)?; self.state = TrackStatusState::Reject(super::request_stream::Deliver::new()); Ok(()) } @@ -5438,12 +5378,11 @@ impl TrackStatusServe { Some(err) => Err(err), None => ready!(resolve.poll(waiter, &self.publisher, stream)), }; - let retry_after = resolve.retry_after; let broadcast = match resolved { Ok(Some(broadcast)) => broadcast, Ok(None) => return Poll::Ready(Ok(())), - Err(error) => { - self.reject_with(&crate::Refusal { error, retry_after })?; + Err(err) => { + self.reject(&err)?; continue; } }; @@ -5912,10 +5851,9 @@ impl ForwardNamespaces { } let refusal = update .refused - .map(|error| ietf::RequestError { - error_code: request::to_code(&error, request::Kind::SubscribeNamespace, version), - reason_phrase: "namespace update refused".into(), - retry_interval: 0, + .map(|error| { + let kind = request::Kind::SubscribeNamespace; + ietf::RequestError::refuse(&error, kind, "namespace update refused", version) }) .or_else(|| { update @@ -15506,17 +15444,12 @@ mod serve_tests { drop(claim); } - /// What an `answer` keeps alive until its refusal is on the wire. - type Held = Box; - - /// A SUBSCRIBE `answer` refuses, as the REQUEST_ERROR it writes: (code, Retry Interval). - async fn refused_subscribe( - version: Version, - rendezvous: bool, - answer: impl FnOnce(&origin::Producer) -> std::pin::Pin + '_>>, - ) -> (u64, u64) { - let ms = std::time::Duration::from_millis; + /// A `kind` request whose track the publisher refuses with `err`, as the REQUEST_ERROR it + /// writes: (code, Retry Interval). + async fn refused_track(version: Version, kind: request::Kind, err: Error) -> (u64, u64) { let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let route = origin.route("room", crate::origin::RouteOptions::default()).unwrap(); + let mut requests = route.broadcast().requests(); let peer_setup = peer::PeerSetup::default(); peer_setup.set(peer::Peer::default()); let publisher = Publisher::new( @@ -15528,96 +15461,86 @@ mod serve_tests { peer_setup, version, ); - let request = ScriptedSession::new(Vec::new()); - let stream = Stream::open(&crate::driver::Transport::new(request.clone()), version) + let session = ScriptedSession::new(Vec::new()); + let stream = Stream::open(&crate::driver::Transport::new(session.clone()), version) .await .unwrap(); - let mut msg = subscribe(Filter::NextObject, None); - msg.rendezvous = rendezvous; - let mut serving = std::pin::pin!(publisher.clone().run_subscribe_stream(stream, msg)); - let answering = answer(&origin); - futures::pin_mut!(answering); - let _held = match futures::future::select(answering, serving.as_mut()).await { - futures::future::Either::Left((held, _)) => held, - futures::future::Either::Right((res, _)) => panic!("{version}: ended unasked: {res:?}"), + let mut serving: std::pin::Pin>>> = match kind { + request::Kind::Subscribe => Box::pin(publisher.run_subscribe_stream(stream, subscribe(Filter::NextObject, None))), + request::Kind::Fetch => { + let (namespace, track) = (crate::Path::from_static("room"), "video".into()); + let fetch_type = match version { + Version::Draft18 => FetchType::Standalone { + namespace, + track, + start: Location { group: 0, object: 0 }, + end: Location { group: 0, object: 1 }, + }, + _ => FetchType::Filtered { + namespace, + track, + filter: Filter::Unfiltered, + }, + }; + let msg = ietf::Fetch { + authorization_token: None, + request_id: FETCH_ID, + subscriber_priority: 128, + group_order: GroupOrder::Ascending, + fetch_type, + range_filters: false, + fill_timeout: false, + properties_wanted: true, + }; + Box::pin(publisher.run_fetch_stream(stream, msg)) + } + request::Kind::TrackStatus => Box::pin(publisher.run_track_status_stream( + stream, + ietf::TrackStatus { + authorization_token: None, + request_id: RequestId(REQUEST_ID), + track_namespace: crate::Path::from_static("room"), + track_name: "video".into(), + properties_wanted: true, + }, + )), + kind => unreachable!("{kind:?} asks for no track"), }; + let rejecting = async { requests.requested_track().await.unwrap().reject(err) }; + futures::pin_mut!(rejecting); + if let futures::future::Either::Right((res, _)) = futures::future::select(rejecting, serving.as_mut()).await { + panic!("{version} {kind:?}: ended unasked: {res:?}"); + } kio::clock::timeout(std::time::Duration::from_secs(10), async { - while request.log.writes.lock().unwrap().is_empty() && futures::poll!(serving.as_mut()).is_pending() { - kio::clock::sleep(ms(1)).await; + while session.log.writes.lock().unwrap().is_empty() && futures::poll!(serving.as_mut()).is_pending() { + kio::clock::sleep(std::time::Duration::from_millis(1)).await; } }) .await - .unwrap_or_else(|_| panic!("{version}: the refusal never went out")); + .unwrap_or_else(|_| panic!("{version} {kind:?}: the refusal never went out")); - let wire = request.log.writes.lock().unwrap().clone(); + let wire = session.log.writes.lock().unwrap().clone(); let mut buf = Decoder::new(&wire, version.into()); - assert_eq!(buf.varint().unwrap(), ietf::RequestError::ID, "{version}"); + assert_eq!(buf.varint().unwrap(), ietf::RequestError::ID, "{version} {kind:?}"); let error = ietf::RequestError::decode(&mut buf, version).unwrap(); (error.error_code, error.retry_interval) } - /// A router's refusal of a broadcast sends its retry hint as REQUEST_ERROR's Retry - /// Interval, rounded up to a millisecond so it never reads as "do not retry". A bare - /// refusal sends 0, and a rendezvous that times out keeps the hint. + /// A refused track invites a retry on every request for it when asking again may pass, + /// and asks the peer never to retry a verdict. #[kio::test] - async fn a_broadcast_refusal_carries_its_retry_interval() { - let ms = std::time::Duration::from_millis; + async fn a_track_refusal_that_may_pass_invites_a_retry() { + let retry = crate::auth::RETRY_INITIAL.as_millis() as u64; for version in ALL_VERSIONS { - for (refusal, rendezvous, expected) in [ - (crate::Refusal::from(Error::Unroutable), false, (0x10, 0)), - (crate::Refusal::retry(Error::Unroutable, ms(1500)), false, (0x10, 1500)), - ( - crate::Refusal::retry(Error::Unroutable, std::time::Duration::from_micros(1)), - false, - (0x10, 1), - ), - ( - crate::Refusal::retry(Error::Unroutable, std::time::Duration::ZERO), - false, - (0x10, 0), - ), - (crate::Refusal::retry(Error::Unauthorized, ms(250)), false, (0x1, 250)), - (crate::Refusal::retry(Error::Unroutable, ms(1500)), true, (0x2, 1500)), - ] { - let case = format!("{version}: {refusal:?} rendezvous={rendezvous}"); - let wire = refused_subscribe(version, rendezvous, |origin| { - Box::pin(async move { - let route = origin - .forward("room", crate::origin::RouteInfo::default(), None) - .unwrap(); - route.requested().await.unwrap().reject(refusal); - // The route stays up: only its answer refuses. - Box::new(route) as Held - }) - }) - .await; - assert_eq!(wire, expected, "{case}"); - } - } - } - - /// A dynamic track's rejection carries its retry hint the same way. - #[kio::test] - async fn a_track_refusal_carries_its_retry_interval() { - for version in ALL_VERSIONS { - for (refusal, expected) in [ - (crate::Refusal::from(Error::NotFound), (0x10, 0)), - ( - crate::Refusal::retry(Error::NotFound, std::time::Duration::from_secs(2)), - (0x10, 2000), - ), - ] { - let case = format!("{version}: {refusal:?}"); - let wire = refused_subscribe(version, false, |origin| { - Box::pin(async move { - let broadcast = origin.route("room", crate::origin::RouteOptions::default()).unwrap(); - let mut dynamic = broadcast.broadcast().requests(); - dynamic.requested_track().await.unwrap().reject(refusal); - Box::new((broadcast, dynamic)) as Held - }) - }) - .await; - assert_eq!(wire, expected, "{case}"); + for kind in [request::Kind::Subscribe, request::Kind::Fetch, request::Kind::TrackStatus] { + for (err, expected) in [ + (Error::GoingAway, (0x6, retry)), + (Error::NotFound, (0x10, 0)), + (Error::Unauthorized, (0x1, 0)), + ] { + let case = format!("{version} {kind:?} {err:?}"); + assert_eq!(refused_track(version, kind, err).await, expected, "{case}"); + } } } } diff --git a/rs/moq-net/src/ietf/request.rs b/rs/moq-net/src/ietf/request.rs index d31ea854b01..87768376419 100644 --- a/rs/moq-net/src/ietf/request.rs +++ b/rs/moq-net/src/ietf/request.rs @@ -86,6 +86,23 @@ pub struct RequestError<'a> { pub retry_interval: u64, } +impl<'a> RequestError<'a> { + /// The REQUEST_ERROR refusing a `kind` request with `err`. + pub(crate) fn refuse( + err: &crate::Error, + kind: super::error::request::Kind, + reason: impl Into>, + version: Version, + ) -> Self { + let error_code = super::error::request::to_code(err, kind, version); + Self { + error_code, + reason_phrase: reason.into(), + retry_interval: super::error::request::retry_interval(err, error_code), + } + } +} + impl<'a> Message<'a> for RequestError<'a> { const ID: u64 = 0x05; diff --git a/rs/moq-net/src/ietf/request_stream.rs b/rs/moq-net/src/ietf/request_stream.rs index c8f55d0ba2b..a9c97865835 100644 --- a/rs/moq-net/src/ietf/request_stream.rs +++ b/rs/moq-net/src/ietf/request_stream.rs @@ -315,7 +315,7 @@ impl Deliver { } } -/// Buffer the REQUEST_ERROR refusing a `kind` request with `err`, asking the peer not to retry. +/// Buffer the REQUEST_ERROR refusing a `kind` request with `err`. pub(super) fn buffer_error( writer: &mut Writer, kind: super::error::request::Kind, @@ -323,39 +323,8 @@ pub(super) fn buffer_error( reason: &str, version: Version, ) -> Result<(), Error> { - writer.buffer_message(ID, &request_error(kind, err, None, reason, version)) -} - -/// Buffer the REQUEST_ERROR refusing a `kind` request with `refusal`, carrying its retry hint. -pub(super) fn buffer_refusal( - writer: &mut Writer, - kind: super::error::request::Kind, - refusal: &crate::Refusal, - reason: &str, - version: Version, -) -> Result<(), Error> { - let error = request_error(kind, &refusal.error, refusal.retry_after, reason, version); - writer.buffer_message(ID, &error) -} - -const ID: u64 = super::RequestError::ID; - -/// The REQUEST_ERROR for a refusal. The Retry Interval rounds up to whole milliseconds, since -/// 0 tells the peer never to retry. -fn request_error<'a>( - kind: super::error::request::Kind, - err: &Error, - retry_after: Option, - reason: &'a str, - version: Version, -) -> super::RequestError<'a> { - super::RequestError { - error_code: super::error::request::to_code(err, kind, version), - reason_phrase: reason.into(), - retry_interval: retry_after.map_or(0, |after| { - u64::try_from(after.as_nanos().div_ceil(1_000_000)).unwrap_or(u64::MAX) - }), - } + let error = super::RequestError::refuse(err, kind, reason, version); + writer.buffer_message(super::RequestError::ID, &error) } /// A framed subscription update, preserving omitted preferences. diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 7ab881730c1..92ede1ab41a 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -1711,18 +1711,8 @@ fn refuse_going_away(stream: Stream, id: u64, version: Versi if id == auth::Auth::ID { return auth::going_away(stream); } - Reject::new( - stream, - ietf::RequestError { - error_code: super::error::request::to_code( - &Error::GoingAway, - super::error::request::Kind::Subscribe, - version, - ), - reason_phrase: "going away".into(), - retry_interval: 0, - }, - ) + let kind = super::error::request::Kind::Subscribe; + Reject::new(stream, ietf::RequestError::refuse(&Error::GoingAway, kind, "going away", version)) } /// A control stream cannot end independently of its session, so its own failure is the diff --git a/rs/moq-net/src/model/front.rs b/rs/moq-net/src/model/front.rs index 768e0118f00..3849442ad24 100644 --- a/rs/moq-net/src/model/front.rs +++ b/rs/moq-net/src/model/front.rs @@ -567,7 +567,7 @@ impl Front { // failover, re-splice from the serving source at the first missing group. Err(err) if delivered - && super::resume::retryable(&err) + && err.transient() && self.serving.is_some_and(|(_, candidate)| candidate.carried) => { track.state = TrackState::Idle; diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 2254723c554..98052543090 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -19,7 +19,7 @@ use super::{ front::{Action, Candidate, Event, Front, Refusal}, }; use crate::{ - Error, Instance, InvalidPattern, Path, Pattern, Patterns, Refusal as Refused, + Error, Instance, InvalidPattern, Path, Pattern, Patterns, coding::{BoundsExceeded, Decode, DecodeError, Decoder, Encode, EncodeError, Encoder, Form, varint}, path::{self, Segment}, time::{Deadlines, Instant}, @@ -3687,8 +3687,6 @@ struct TrackIo { held: Option>, /// Whether the track had a reader as of the last demand edge. wanted: bool, - /// The retry hint of the last copy that refused the track. - retry_after: Option, } impl TrackIo { @@ -3713,23 +3711,18 @@ impl TrackIo { self.staged = None; self.copy = None; self.held = None; - let retry_after = self.retry_after.take(); match self.request.take() { // Ended before any route served it: nothing to read, nor to fetch. Some(request) => { self.routes.end(result.clone()); - let error = result.err().unwrap_or(Error::NotFound); - request.reject(Refused { error, retry_after }); + request.reject(result.err().unwrap_or(Error::NotFound)); } None => { if let Some(accepted) = self.accepted.take() { let _ = match &result { Ok(()) => accepted.finish(), // A failed track refuses newcomers too, so they ask afresh. - Err(err) => accepted.abort(Refused { - error: err.clone(), - retry_after, - }), + Err(err) => accepted.abort(err.clone()), }; } self.routes.end(result); @@ -3890,7 +3883,7 @@ impl kio::Task for FrontServeRun { } Poll::Ready(Err(err)) => { // No metadata ever arrived: a lost route cannot answer the request. - let err = match super::resume::route_failed(&err) { + let err = match err.retryable() { true => io.end.clone(), false => err, }; @@ -3936,8 +3929,7 @@ fn pick<'a>( /// What [`FrontServe::poll_step`] returns: one thing that happened. enum FrontStep { Assigned(track::Request), - /// The upstream answer, and the retry hint a rejection carried. - Resolved(Result, Option), + Resolved(Result), SourceClosed(u64), Info(Arc, u64, Result), Ended(Arc, u64, Result<(), Error>, bool), @@ -3999,8 +3991,6 @@ struct FrontServe { ready: kio::Queue>, tracks: HashMap, TrackIo>, upstream: Option, - /// The retry hint the upstream refusal queued in `events` carried. - retry_after: Option, next_source: u64, /// The instance each source's route served when it was asked, and the table's clock /// then: what its content is from. @@ -4052,7 +4042,6 @@ impl FrontServe { ready: kio::Queue::new(), tracks: HashMap::new(), upstream: None, - retry_after: None, next_source: 0, instances: HashMap::new(), sources: HashMap::new(), @@ -4186,15 +4175,15 @@ impl FrontServe { } if let Some(upstream) = &self.upstream && let Poll::Ready(result) = upstream.pending.poll(waiter, |p| match &p.resolved { - Some(result) => Poll::Ready((result.clone(), p.retry_after)), + Some(result) => Poll::Ready(result.clone()), None => Poll::Pending, }) { - return Poll::Ready(match result { - Ok((resolved, retry_after)) => FrontStep::Resolved(resolved, retry_after), + return Poll::Ready(FrontStep::Resolved(match result { + Ok(resolved) => resolved, // The queue died unresolved (its handler dropped): the route // could not serve. - Err(_closed) => FrontStep::Resolved(Err(Error::Unroutable), None), - }); + Err(_closed) => Err(Error::Unroutable), + })); } // The listing first: a source ends its listing before it closes, so this forwards // the tracks it finished last before its close detaches it. @@ -4334,11 +4323,6 @@ impl kio::Task for FrontServe { .. } ); - // Only the standing refusal that ends the front passes its hint on. - let retry_after = match &event { - Event::Resolved { .. } => self.retry_after.take().filter(|_| refused), - _ => None, - }; let answered = match &event { Event::TrackInfo { track, source, .. } => Some((track.clone(), *source)), _ => None, @@ -4619,9 +4603,6 @@ impl kio::Task for FrontServe { }); if let Ok(mut pending) = self.request.write() { pending.refused = refused; - if pending.resolved.is_none() { - pending.retry_after = retry_after; - } pending.resolved.get_or_insert(Err(err.clone())); } // Ending the broadcast only retracts it: no new requesters or @@ -4720,7 +4701,6 @@ impl kio::Task for FrontServe { copy: None, held: None, wanted: false, - retry_after: None, }, ); Event::TrackAssigned { @@ -4728,7 +4708,7 @@ impl kio::Task for FrontServe { now: self.timers.now(), } } - FrontStep::Resolved(result, retry_after) => { + FrontStep::Resolved(result) => { let Some(FrontUpstream { route, asked, @@ -4770,16 +4750,13 @@ impl kio::Task for FrontServe { self.instances.insert(id, (asked, asked_at)); Event::Resolved { route, result: Ok(id) } } - Err(err) => { - self.retry_after = retry_after; - Event::Resolved { - route, - result: Err(Refusal { - err, - standing: self.standing(route), - }), - } - } + Err(err) => Event::Resolved { + route, + result: Err(Refusal { + err, + standing: self.standing(route), + }), + }, } } FrontStep::SourceClosed(source) => Event::SourceClosed { source }, @@ -4797,8 +4774,6 @@ impl kio::Task for FrontServe { }, Err(err) => Err(err), }; - // What the refusing copy said about retrying, should it end the track. - io.retry_after = result.as_ref().err().and(asked.query.retry_after()); // Staged only while the track has a reader: without one the machine // will not splice, and a held copy would keep the source subscribed. if let Ok(info) = &result @@ -6239,8 +6214,6 @@ impl OriginState { #[derive(Default)] struct PendingBroadcast { resolved: Option>, - /// The retry hint a handler's rejection carried. - retry_after: Option, } /// A front's verdict, which its requesters wait on: the instance it resolved, or the @@ -6263,8 +6236,6 @@ struct PendingFront { selected: Instance, /// The selected serving handler refused, rather than the route disappearing. refused: bool, - /// The retry hint that refusal carried. - retry_after: Option, } impl PendingFront { @@ -6380,16 +6351,12 @@ impl ServeState { path: &Path, producer: &kio::Producer, result: Result, - retry_after: Option, ) -> bool { let mut state = shared.lock(); let Some(resolved) = state.answer(path, producer, result) else { return false; }; if let Ok(mut pending) = producer.write() { - if pending.resolved.is_none() { - pending.retry_after = retry_after; - } pending.resolved.get_or_insert(resolved); drop(state); } @@ -6524,16 +6491,13 @@ impl Request { /// use [`Self::route`] to announce its own instance. pub fn serve(self, broadcast: impl Consume) { let broadcast = broadcast.consume(); - let _ = ServeState::resolve(&self.home, &self.path, &self.producer, Ok(broadcast), None); + let _ = ServeState::resolve(&self.home, &self.path, &self.producer, Ok(broadcast)); // `self.producer` drops here, closing the channel; the value is still observable. } - /// Reject the request, resolving every awaiting requester with the refusal's error. - /// - /// A bare [`Error`] converts; see [`Refusal`](crate::Refusal) to add a retry hint. - pub fn reject(self, refusal: impl Into) { - let Refused { error, retry_after } = refusal.into(); - let _ = ServeState::resolve(&self.home, &self.path, &self.producer, Err(error), retry_after); + /// Reject the request, resolving every awaiting requester with `err`. + pub fn reject(self, err: Error) { + let _ = ServeState::resolve(&self.home, &self.path, &self.producer, Err(err)); } } @@ -6624,29 +6588,11 @@ impl Lookup { /// Poll for the requested broadcast without blocking. pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll> { - self.poll_refused(waiter).map_err(|refused| refused.error) - } - - /// [`Self::poll_ok`], keeping the retry hint a handler's refusal carried. - pub(crate) fn poll_refused(&self, waiter: &kio::Waiter) -> Poll> { - let mut retry_after = None; - let found = ready!(self.poll_found(waiter, &mut retry_after)); - Poll::Ready(found.map_err(|error| Refused { error, retry_after })) - } - - fn poll_found( - &self, - waiter: &kio::Waiter, - retry_after: &mut Option, - ) -> Poll> { match &self.inner { RequestState::Failed(error) => Poll::Ready(Err(error.clone())), RequestState::Pending { front, broadcast } => Poll::Ready( match ready!(front.poll(waiter, |state| match &state.resolved { - Some(result) => { - *retry_after = state.retry_after; - Poll::Ready(result.clone().map(|instance| (instance, state.first.clone()))) - } + Some(result) => Poll::Ready(result.clone().map(|instance| (instance, state.first.clone())),), None => Poll::Pending, })) { // Another instance than the one named, and not one continuing it: never hand @@ -7574,7 +7520,6 @@ impl Consumer { first: None, selected: best, refused: false, - retry_after: None, }); let consumer = request.consume(); // The front starts held by this request; see [`Front::new`]. @@ -8359,7 +8304,6 @@ mod tests { copy: None, held: None, wanted: true, - retry_after: None, }; let mut subscriber = consumer.subscribe(None).now_or_never().unwrap().unwrap(); io.end(Ok(())); @@ -10639,7 +10583,6 @@ mod tests { first: None, selected: instance(), refused: false, - retry_after: None, }), timers: consumer.timers.clone(), }; @@ -14199,41 +14142,6 @@ mod tests { (producer, server, upstream, old, subscription) } - /// A retry hint on a carried copy's abort changes nothing the front decides: a - /// `GoingAway` copy that delivered is asked for again from the same source, hinted or not. - #[kio::test] - #[cfg(feature = "async")] - async fn a_retry_hint_keeps_a_going_away_copy_retryable() { - for refusal in [ - crate::Refusal::from(Error::GoingAway), - crate::Refusal::retry(Error::GoingAway, Duration::from_secs(1)), - ] { - let hinted = refusal.retry_after.is_some(); - let (_producer, _server, upstream, old, mut subscription) = remote_front().await; - let mut handler = upstream.requests(); - old.abort(refusal).unwrap(); - - let request = kio::clock::timeout(Duration::from_secs(1), handler.requested_track()) - .await - .unwrap_or_else(|_| panic!("hinted={hinted}: the front never asked the source again")) - .expect("request"); - let track = request.accept(crate::track::Info::timed()); - // The copy resumes past the group the reader already has. - let mut group = track.create_group(crate::group::Info { sequence: 1 }).unwrap(); - group.write_frame(crate::Timestamp::ZERO, b"after".as_ref()).unwrap(); - group.finish().unwrap(); - let mut group = next_group(&mut subscription) - .await - .unwrap_or_else(|err| panic!("hinted={hinted}: the track ended: {err}")) - .expect("track open"); - assert_eq!( - &group.read_frame().await.unwrap().unwrap().payload[..], - b"after", - "hinted={hinted}" - ); - } - } - /// A local newcomer takes over while its answer for "video" is pending, so the track /// drains the incumbent. The reader leaves: the incumbent's copy is dropped at once, /// like any unread track's, not held until the linger. diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index e3cf3c06def..f97c45212c0 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -466,45 +466,10 @@ impl Consumer { } } -/// Whether `err` is the route failing to deliver rather than judging what it carried, so a new -/// copy may fix it. Anything else would repeat on every copy. Mirrors `routeFailed` in -/// `js/net/src/resume.ts`. -pub(super) fn route_failed(err: &Error) -> bool { - unroutable(err) || retryable(err) -} - -/// Whether `err` is a source losing its route to the content: only another route continues it. -fn unroutable(err: &Error) -> bool { - matches!(err, Error::Unroutable | Error::Stream(StreamError::Unroutable)) -} - -/// Whether the same source may deliver a fresh copy. An unroutable source needs a new -/// route instead: asking it again can replace the withdrawal's cause with a refusal. -pub(super) fn retryable(err: &Error) -> bool { - matches!( - err, - Error::Transport(_) - | Error::IdleTimeout - | Error::Session(_) - | Error::GoingAway - | Error::GoawayTimeout - | Error::Dropped - | Error::Closed - | Error::Cancel - | Error::Stream( - StreamError::Session(_) - | StreamError::Internal - | StreamError::Cancel - | StreamError::GoingAway - | StreamError::ControlTimeout - ) - ) -} - /// Whether `copy` lost its route, so an error read from it is no verdict on what it carried. /// A terminal track verdict, unlike a delivery failure, still judges its content. fn copy_failed(copy: &track::Consumer) -> bool { - matches!(copy.poll_complete(&kio::Waiter::noop()), Poll::Ready(Err(err)) if route_failed(&err)) + matches!(copy.poll_complete(&kio::Waiter::noop()), Poll::Ready(Err(err)) if err.retryable()) } /// A route's copy, as one reader reads it. @@ -1667,7 +1632,7 @@ impl Reader { // failure is the front's to replace. Mirrors the JS reader's `#end`. match serving.and_then(&done) { Some(Ok(())) => return Some(Poll::Ready(Ok(None))), - Some(Err(err)) if !route_failed(&err) => return Some(Poll::Ready(Err(err))), + Some(Err(err)) if !err.retryable() => return Some(Poll::Ready(Err(err))), _ => {} } let end = self.route.end(); @@ -1728,12 +1693,12 @@ impl Reader { Poll::Ready(Ok(sub)) => match sub.poll_finished(waiter) { Poll::Ready(Ok(fin)) => return Poll::Ready(Ok(fin)), // Only a delivery failure waits for the front to replace its copy. - Poll::Ready(Err(err)) if !route_failed(&err) || concluded.is_some() => { + Poll::Ready(Err(err)) if !err.retryable() || concluded.is_some() => { return Poll::Ready(Err(err)); } _ => {} }, - Poll::Ready(Err(err)) if !route_failed(&err) || concluded.is_some() => { + Poll::Ready(Err(err)) if !err.retryable() || concluded.is_some() => { return Poll::Ready(Err(err)); } _ => {} @@ -1771,7 +1736,7 @@ impl Reader { } } for (run, err) in given_up { - if !route_failed(&Error::Stream(err.clone())) { + if !Error::Stream(err.clone()).retryable() { self.drops.insert(run, err); } } @@ -2106,7 +2071,7 @@ impl Recover { /// exists, so neither is a verdict on a fetch: one answered from a group still arriving /// over a subscription asks the route again for the rest when that delivery gives up on it. fn undelivered(&self, err: &Error) -> bool { - route_failed(err) + err.retryable() || copy_failed(&self.copy) || matches!(err, Error::Lagged) || (self.reader.is_none() && matches!(StreamError::from(err), StreamError::Old | StreamError::Evicted)) @@ -2171,7 +2136,7 @@ impl Recover { } // An unroutable copy waits for another route: asking its own again could only replace // the withdrawal's cause with a refusal. None comes once the front is done. - let withdrawn = failed.filter(|err| generation == self.generation && unroutable(err)); + let withdrawn = failed.filter(|err| generation == self.generation && err.unroutable()); if let Some(err) = withdrawn && terminal { @@ -2259,7 +2224,7 @@ impl Recover { // route's subscription, not whether the group exists, so it is no verdict on a // fetch; nor is one a copy that failed dropped its open groups with, or a failure // of the route it came through. - Poll::Ready(Err(err)) if self.reader.is_some() && !route_failed(&err) && !copy_failed(serving) => { + Poll::Ready(Err(err)) if self.reader.is_some() && !err.retryable() && !copy_failed(serving) => { return Poll::Ready(Err(err)); } _ => {} @@ -2272,7 +2237,7 @@ impl Recover { if self.reader.is_some() && serving.poll_closed(waiter).is_ready() && let Poll::Ready(Err(err)) = serving.poll_complete(&kio::Waiter::noop()) - && !route_failed(&err) + && !err.retryable() { verdict = Some(err); } @@ -2313,7 +2278,7 @@ impl Recover { }; let err = match fetch { // That route failed: the next one is asked. - Fetch::Refused(err) if route_failed(err) || copy_failed(serving) => return Poll::Pending, + Fetch::Refused(err) if err.retryable() || copy_failed(serving) => return Poll::Pending, Fetch::Refused(err) => err.clone(), Fetch::Pending(pending) => match ready!(kio::Task::poll(&mut **pending, waiter)) { Ok(group) => { @@ -2330,7 +2295,7 @@ impl Recover { Error::NotFound } // That route is failing too: wait for the next one. - Err(err) if route_failed(&err) || copy_failed(serving) => { + Err(err) if err.retryable() || copy_failed(serving) => { *fetch = Fetch::Refused(err); return Poll::Pending; } @@ -2576,7 +2541,7 @@ impl kio::Task for Fetching { } // The route failed: ask the next one, unless none will come. Kept as the answer, // since the same copy is never asked twice. - Err(err) if (route_failed(&err) || copy_failed(&serving)) && self.route.end().is_none() => { + Err(err) if (err.retryable() || copy_failed(&serving)) && self.route.end().is_none() => { if let Some(asked) = &mut self.pending { asked.answer = Some(Err(err)); } @@ -6550,9 +6515,9 @@ mod test { const DELIVERY: [u32; 6] = [0x0, 0x1, 0x3, 0x4, 0x31, 0x36]; for code in (0..0x40).chain([64, 64 + 5, 0xffff_ffff]) { let err = Error::Stream(StreamError::from_code(code)); - assert_eq!(route_failed(&err), DELIVERY.contains(&code), "{code:#x}: {err:?}"); + assert_eq!(err.retryable(), DELIVERY.contains(&code), "{code:#x}: {err:?}"); } - assert!(route_failed(&Error::Unroutable)); + assert!(Error::Unroutable.retryable()); } /// A failure judging the group repeats on every copy from its route, so it ends a read or diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 9ef97ccc5bc..1e869a15af6 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -549,8 +549,6 @@ pub(crate) struct TrackState { // The error that caused the track to be aborted, if any. abort: Option, - // The retry hint the abort or rejection carried, for a request it refuses. - retry_after: Option, // Whether the declared end settled before a write that could reopen it or an abort, or // `closed()` reported it: every group below it was produced and finished then, so the @@ -3908,16 +3906,10 @@ impl Producer { /// /// [`finish`](Self::finish) is deliberately not terminal: it declares the final /// sequence, and lower-numbered groups may still be written afterwards. - /// - /// A bare [`Error`] converts; see [`Refusal`](crate::Refusal) to tell a request it - /// refuses when to retry. - pub fn abort(self, refusal: impl Into) -> Result<()> { - let crate::Refusal { error, retry_after } = refusal.into(); + pub fn abort(self, err: Error) -> Result<()> { // Unlisted first, so the track leaves the listing before its name is free again. self.alive.unlist(false); - let mut state = self.modify()?; - state.retry_after = retry_after; - commit_abort(state, error).abort(); + commit_abort(self.modify()?, err).abort(); Ok(()) } @@ -5531,12 +5523,6 @@ impl Subscribing { } } - /// The retry hint the track's abort or rejection carried. - #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] - pub(crate) fn retry_after(&self) -> Option { - self.state.read().retry_after - } - /// Poll until the peer confirms the subscription, yielding the [`Subscriber`]. /// Errors if the track is aborted or not found. pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll> { @@ -5601,11 +5587,6 @@ pub struct Resolving { } impl Resolving { - /// The retry hint the track's abort or rejection carried. - pub(crate) fn retry_after(&self) -> Option { - self.state.read().retry_after - } - /// A handle on the track instance this resolves, whatever its broadcast serves later. pub fn track(&self) -> Consumer { Consumer { @@ -9477,14 +9458,10 @@ impl Request { } } - /// Reject the request, waking all waiting subscribers with the refusal's error. - /// - /// A bare [`Error`] converts; see [`Refusal`](crate::Refusal) to add a retry hint. - pub fn reject(self, refusal: impl Into) { - let crate::Refusal { error, retry_after } = refusal.into(); + /// Reject the request, waking all waiting subscribers with `err`. + pub fn reject(self, err: Error) { if let Ok(mut state) = self.state.write() { - state.abort = Some(error); - state.retry_after = retry_after; + state.abort = Some(err); } } From 3a794f23b171c839be472b331e202d1e65aaad91 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 00:01:01 -0800 Subject: [PATCH 3/7] style: format Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/publisher.ts | 3 ++- rs/moq-net/src/ietf/error.rs | 4 +++- rs/moq-net/src/ietf/publisher.rs | 10 ++++++++-- rs/moq-net/src/ietf/session.rs | 5 ++++- rs/moq-net/src/model/front.rs | 4 +--- 5 files changed, 18 insertions(+), 8 deletions(-) diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index b6c17c3e606..ff1c4740c6d 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -2552,7 +2552,8 @@ export class Publisher { #failed(err: unknown, wait = false): Refusal { const e = error(err); const reasonPhrase = reason(e); - const missing = e instanceof StreamError && (e.code === StreamCode.NotFound || e.code === StreamCode.Unroutable); + const missing = + e instanceof StreamError && (e.code === StreamCode.NotFound || e.code === StreamCode.Unroutable); if (missing && wait) return { condition: "timeout", reasonPhrase }; return { condition: missing ? "does_not_exist" : "internal", reasonPhrase, retryable: routeFailed(e) }; } diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index b78f36b7b4a..48614771019 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -327,7 +327,9 @@ pub(crate) mod request { let expected = match err { // The session failing under a request is no verdict on it. Error::GoingAway | Error::Session(_) => retry, - Error::Unroutable if matches!(kind, Kind::Subscribe | Kind::Fetch | Kind::TrackStatus) => retry, + Error::Unroutable if matches!(kind, Kind::Subscribe | Kind::Fetch | Kind::TrackStatus) => { + retry + } _ => 0, }; assert_eq!(retry_interval(&err, code), expected, "{version:?} {kind:?} {err:?}"); diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 8d467070f9d..8514fdedb12 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -15466,7 +15466,9 @@ mod serve_tests { .await .unwrap(); let mut serving: std::pin::Pin>>> = match kind { - request::Kind::Subscribe => Box::pin(publisher.run_subscribe_stream(stream, subscribe(Filter::NextObject, None))), + request::Kind::Subscribe => { + Box::pin(publisher.run_subscribe_stream(stream, subscribe(Filter::NextObject, None))) + } request::Kind::Fetch => { let (namespace, track) = (crate::Path::from_static("room"), "video".into()); let fetch_type = match version { @@ -15532,7 +15534,11 @@ mod serve_tests { async fn a_track_refusal_that_may_pass_invites_a_retry() { let retry = crate::auth::RETRY_INITIAL.as_millis() as u64; for version in ALL_VERSIONS { - for kind in [request::Kind::Subscribe, request::Kind::Fetch, request::Kind::TrackStatus] { + for kind in [ + request::Kind::Subscribe, + request::Kind::Fetch, + request::Kind::TrackStatus, + ] { for (err, expected) in [ (Error::GoingAway, (0x6, retry)), (Error::NotFound, (0x10, 0)), diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 92ede1ab41a..77a727186dd 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -1712,7 +1712,10 @@ fn refuse_going_away(stream: Stream, id: u64, version: Versi return auth::going_away(stream); } let kind = super::error::request::Kind::Subscribe; - Reject::new(stream, ietf::RequestError::refuse(&Error::GoingAway, kind, "going away", version)) + Reject::new( + stream, + ietf::RequestError::refuse(&Error::GoingAway, kind, "going away", version), + ) } /// A control stream cannot end independently of its session, so its own failure is the diff --git a/rs/moq-net/src/model/front.rs b/rs/moq-net/src/model/front.rs index 3849442ad24..c4c6933ca61 100644 --- a/rs/moq-net/src/model/front.rs +++ b/rs/moq-net/src/model/front.rs @@ -566,9 +566,7 @@ impl Front { // Died mid-serve after delivering from the same instance: normal // failover, re-splice from the serving source at the first missing group. Err(err) - if delivered - && err.transient() - && self.serving.is_some_and(|(_, candidate)| candidate.carried) => + if delivered && err.transient() && self.serving.is_some_and(|(_, candidate)| candidate.carried) => { track.state = TrackState::Idle; self.redispatch(name, actions); From 2dcc7afa97770927f49e2bb897342317a9a3d144 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 00:57:01 -0800 Subject: [PATCH 4/7] fix(net): invite a retry only where the code leaves the request open A failed session can carry a verdict, such as a renewal refused UNAUTHORIZED, which the resume classification still counts as no verdict on the content. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/error.rs | 12 +++++++----- rs/moq-net/src/ietf/publisher.rs | 5 +++-- rs/moq-net/src/ietf/session.rs | 5 +++-- 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index 48614771019..cb13f6c23ce 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -251,9 +251,11 @@ pub(crate) mod request { /// The Retry Interval, in milliseconds, for a refusal with `err` that reads `code` on the wire. /// /// A refusal asking again may pass invites the backoff a refused token starts from. 0 asks the - /// peer never to retry: a verdict would repeat, and UNINTERESTED tells an offer to stop. + /// peer never to retry a verdict, which would repeat. The code must leave the request open + /// too: a failed session can carry a verdict (UNAUTHORIZED, NOT_SUPPORTED), and UNINTERESTED + /// tells an offer to stop. pub(crate) fn retry_interval(err: &Error, code: u64) -> u64 { - match err.retryable() && code != UNINTERESTED { + match err.retryable() && matches!(code, INTERNAL_ERROR | DOES_NOT_EXIST | GOING_AWAY) { true => crate::auth::RETRY_INITIAL.as_millis() as u64, false => 0, } @@ -316,7 +318,8 @@ pub(crate) mod request { ]; /// Only a refusal that asking again may pass invites a retry. An unroutable offer reads - /// UNINTERESTED, which tells the peer to stop offering, so it never does. + /// UNINTERESTED, which tells the peer to stop offering, and a session failing with a + /// verdict (`Session(Unsupported)`) keeps it, so neither does. #[test] fn only_a_refusal_that_may_pass_invites_a_retry() { let retry = crate::auth::RETRY_INITIAL.as_millis() as u64; @@ -325,8 +328,7 @@ pub(crate) mod request { for err in EVERY_ERROR { let code = to_code(&err, kind, version); let expected = match err { - // The session failing under a request is no verdict on it. - Error::GoingAway | Error::Session(_) => retry, + Error::GoingAway => retry, Error::Unroutable if matches!(kind, Kind::Subscribe | Kind::Fetch | Kind::TrackStatus) => { retry } diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 8514fdedb12..702fa0f07e8 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -10852,7 +10852,8 @@ mod serve_tests { /// "does not exist" value, 0x10, not 0x4 (MALFORMED_AUTH_TOKEN): a peer told the wrong /// one re-authenticates instead of waiting for the announcement. The reply is encoded here rather than matched by code alone, so /// a value slipping outside the draft's table cannot pass, and the reason phrase is the - /// origin's own, which is what carries a refusal the registry has no value for. + /// origin's own, which is what carries a refusal the registry has no value for. The + /// publisher may still come up, so the refusal invites a retry. #[kio::test] async fn a_missing_broadcast_is_refused_with_the_draft_s_code() { for version in [Version::Draft18, Version::Draft20, Version::Draft22] { @@ -10881,7 +10882,7 @@ mod serve_tests { .encode(&ietf::RequestError { error_code, reason_phrase: Error::Unroutable.to_string().into(), - retry_interval: 0, + retry_interval: crate::auth::RETRY_INITIAL.as_millis() as u64, }) .await .unwrap(); diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 77a727186dd..ee7643765df 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -4088,7 +4088,7 @@ mod tests { /// A request past our GOAWAY cutoff is refused in the answer its kind reads: an AUTH /// with AUTH_ERROR carrying GOING_AWAY, since its presenter takes anything else as a - /// violation, and the rest with REQUEST_ERROR. + /// violation, and the rest with REQUEST_ERROR, inviting a retry on the next session. #[kio::test] async fn a_request_past_our_goaway_is_refused_in_its_own_answer() { const VERSION: Version = Version::Draft18; @@ -4098,7 +4098,8 @@ mod tests { (auth::Auth::ID, frame(VERSION, auth::AuthError::ID, &body)), ( ietf::Subscribe::ID, - frame(VERSION, ietf::RequestError::ID, &[&[0x6, 0][..], &body[1..]].concat()), + // A Retry Interval of 1000ms, as a draft-18 varint. + frame(VERSION, ietf::RequestError::ID, &[&[0x6, 0x83, 0xe8][..], &body[1..]].concat()), ), ] { let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()); From 3334629f96881da12be715e89c87021f195f31ca Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 01:59:40 -0800 Subject: [PATCH 5/7] style: format Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/session.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index ee7643765df..2cc00635487 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -4099,7 +4099,11 @@ mod tests { ( ietf::Subscribe::ID, // A Retry Interval of 1000ms, as a draft-18 varint. - frame(VERSION, ietf::RequestError::ID, &[&[0x6, 0x83, 0xe8][..], &body[1..]].concat()), + frame( + VERSION, + ietf::RequestError::ID, + &[&[0x6, 0x83, 0xe8][..], &body[1..]].concat(), + ), ), ] { let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()); From 39e9a3e83ebf02800dcfe77876944793dd7c5257 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 07:57:01 -0800 Subject: [PATCH 6/7] fix(net): keep a session verdict's refusal a verdict in js/net A TRACK_STATUS or FETCH that failed because its upstream session closed with Unauthorized, Expired, Timeout, Version or Unsupported was refused INTERNAL_ERROR with a 1s Retry Interval. Map those session codes to their request codes, as Rust's `to_code` does, and invite a retry only when the code leaves the request open, as Rust's `retry_interval` does. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/publisher.test.ts | 13 +++++++++---- js/net/src/ietf/publisher.ts | 23 +++++++++++++++++++++-- js/net/src/ietf/request_stream.ts | 9 ++++++--- 3 files changed, 36 insertions(+), 9 deletions(-) diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index b3cc1f99d52..a220071a2f3 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -422,9 +422,13 @@ test("TRACK_STATUS for a missing broadcast or track is refused DOES_NOT_EXIST", // to retry a verdict. test("a TRACK_STATUS refusal that may pass invites a retry", async () => { for (const version of IETF_VERSIONS) { - for (const [err, retryInterval] of [ - [new StreamError(StreamCode.GoingAway, { message: "draining" }), 1000n], - [new StreamError(StreamCode.NotFound, { message: "no video" }), 0n], + for (const [err, errorCode, retryInterval] of [ + [new StreamError(StreamCode.GoingAway, { message: "draining" }), 0x0, 1000n], + [new SessionError(SessionCode.Internal), 0x0, 1000n], + [new StreamError(StreamCode.NotFound, { message: "no video" }), 0x10, 0n], + [new SessionError(SessionCode.Unauthorized), 0x1, 0n], + [new SessionError(SessionCode.Expired), 0x5, 0n], + [new SessionError(SessionCode.Version), 0x3, 0n], ] as const) { const pair = createMockTransportPair(ALPNS[version]); const session = new NativeSession(pair.server, version, true); @@ -443,7 +447,8 @@ test("a TRACK_STATUS refusal that may pass invites a retry", async () => { const wire = status.wire(); expect(await wire.u53()).toBe(RequestError.id); - expect((await RequestError.decode(wire, version)).retryInterval).toBe(retryInterval); + const refused = await RequestError.decode(wire, version); + expect([refused.errorCode, refused.retryInterval]).toEqual([errorCode, retryInterval]); await requests.return?.(); served.close(); origin.close(); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index ff1c4740c6d..912adc22461 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -39,7 +39,7 @@ import { TimeoutError, withTimeout } from "../util/timeout.ts"; import * as Varint from "../varint.ts"; import { type Advertised, type Advertisements, sameInstance, wireOf } from "../wire.ts"; import * as Cluster from "./cluster.ts"; -import { requestReason } from "./error.ts"; +import { type RequestCondition, requestReason } from "./error.ts"; import { type Fetch, FetchHeader, FetchOk, type FetchType } from "./fetch.ts"; import * as Filter from "./filter.ts"; import { MAX_AGE_PARAMETER } from "./max_age.ts"; @@ -302,6 +302,24 @@ interface RunFill { streams: StreamCount; } +// The condition a request failed by its session closing is refused with: a session verdict stays +// one, as in Rust's `to_code`, so the peer is never invited to retry it. +function sessionVerdict(code: SessionCode): RequestCondition { + switch (code) { + case SessionCode.Unauthorized: + return "unauthorized"; + case SessionCode.Expired: + return "expired"; + case SessionCode.Timeout: + return "timeout"; + case SessionCode.Version: + case SessionCode.Unsupported: + return "not_supported"; + default: + return "internal"; + } +} + /** * Handles publishing broadcasts using moq-transport protocol. * Uses the stream-per-request pattern (real bidi streams for v17, virtual for v14-v16). @@ -2555,7 +2573,8 @@ export class Publisher { const missing = e instanceof StreamError && (e.code === StreamCode.NotFound || e.code === StreamCode.Unroutable); if (missing && wait) return { condition: "timeout", reasonPhrase }; - return { condition: missing ? "does_not_exist" : "internal", reasonPhrase, retryable: routeFailed(e) }; + const condition = missing ? "does_not_exist" : e instanceof SessionError ? sessionVerdict(e.code) : "internal"; + return { condition, reasonPhrase, retryable: routeFailed(e) }; } /** diff --git a/js/net/src/ietf/request_stream.ts b/js/net/src/ietf/request_stream.ts index 9409341f799..a2c12ca7116 100644 --- a/js/net/src/ietf/request_stream.ts +++ b/js/net/src/ietf/request_stream.ts @@ -36,11 +36,14 @@ export interface Refusal { } async function writeRefusal(w: Writer, version: IetfVersion, kind: RequestKind, refusal: Refusal): Promise { + const errorCode = toRequestCode(refusal.condition, kind, version); + // 0 asks the peer never to retry. As in Rust's `retry_interval`, the code must leave the + // request open too: UNINTERESTED tells an offer to stop, and the rest are verdicts. + const open = errorCode === 0x0 || errorCode === 0x6 || errorCode === 0x10; const err = new RequestError({ - errorCode: toRequestCode(refusal.condition, kind, version), + errorCode, reasonPhrase: refusal.reasonPhrase, - // 0 asks the peer never to retry. - retryInterval: refusal.retryable ? BigInt(RETRY_INITIAL) : 0n, + retryInterval: refusal.retryable && open ? BigInt(RETRY_INITIAL) : 0n, }); await Message.write(w, RequestError.id, err, version); } From 360b092342f0c50834ea57491a0f751a453e2097 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 09:20:04 -0800 Subject: [PATCH 7/7] fix(net): refuse a stream verdict with its own code in js/net A track rejected with a StreamError now refuses with the code Rust's to_code gives it: GoingAway is GOING_AWAY, Unauthorized, DeliveryTimeout and MalformedTrack their verdicts, instead of INTERNAL_ERROR. The TRACK_STATUS wire test covers every session and stream verdict. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/publisher.test.ts | 7 +++- js/net/src/ietf/publisher.ts | 55 +++++++++++++++++++------------ 2 files changed, 40 insertions(+), 22 deletions(-) diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 1f52cd92bcb..bdfbd9f57ae 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -425,12 +425,17 @@ test("TRACK_STATUS for a missing broadcast or track is refused DOES_NOT_EXIST", test("a TRACK_STATUS refusal that may pass invites a retry", async () => { for (const version of IETF_VERSIONS) { for (const [err, errorCode, retryInterval] of [ - [new StreamError(StreamCode.GoingAway, { message: "draining" }), 0x0, 1000n], + [new StreamError(StreamCode.GoingAway, { message: "draining" }), 0x6, 1000n], [new SessionError(SessionCode.Internal), 0x0, 1000n], [new StreamError(StreamCode.NotFound, { message: "no video" }), 0x10, 0n], + [new StreamError(StreamCode.Unauthorized), 0x1, 0n], + [new StreamError(StreamCode.DeliveryTimeout), 0x2, 0n], + [new StreamError(StreamCode.MalformedTrack), 0x12, 0n], [new SessionError(SessionCode.Unauthorized), 0x1, 0n], [new SessionError(SessionCode.Expired), 0x5, 0n], + [new SessionError(SessionCode.Timeout), 0x2, 0n], [new SessionError(SessionCode.Version), 0x3, 0n], + [new SessionError(SessionCode.Unsupported), 0x3, 0n], ] as const) { const pair = createMockTransportPair(ALPNS[version]); const session = new NativeSession(pair.server, version, true); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index b94f474b1ed..bf894ab6273 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -312,22 +312,37 @@ interface RunFill { streams: StreamCount; } -// The condition a request failed by its session closing is refused with: a session verdict stays -// one, as in Rust's `to_code`, so the peer is never invited to retry it. -function sessionVerdict(code: SessionCode): RequestCondition { - switch (code) { - case SessionCode.Unauthorized: - return "unauthorized"; - case SessionCode.Expired: - return "expired"; - case SessionCode.Timeout: - return "timeout"; - case SessionCode.Version: - case SessionCode.Unsupported: - return "not_supported"; - default: - return "internal"; +// The condition a request that failed for `err` is refused with, mirroring Rust's `to_code`: a +// verdict stays one, so the peer is never invited to retry it. +function refusedAs(err: Error): RequestCondition { + if (err instanceof SessionError) { + switch (err.code) { + case SessionCode.Unauthorized: + return "unauthorized"; + case SessionCode.Expired: + return "expired"; + case SessionCode.Timeout: + return "timeout"; + case SessionCode.Version: + case SessionCode.Unsupported: + return "not_supported"; + } + } else if (err instanceof StreamError) { + switch (err.code) { + case StreamCode.Unauthorized: + return "unauthorized"; + case StreamCode.DeliveryTimeout: + return "timeout"; + case StreamCode.MalformedTrack: + return "malformed_track"; + case StreamCode.GoingAway: + return "going_away"; + case StreamCode.NotFound: + case StreamCode.Unroutable: + return "does_not_exist"; + } } + return "internal"; } /** @@ -2735,12 +2750,10 @@ export class Publisher { const e = error(err); const reasonPhrase = reason(e); const missing = - e instanceof StreamError && (e.code === StreamCode.NotFound || e.code === StreamCode.Unroutable); - if (missing && wait && (!resolved || e.code === StreamCode.Unroutable)) { - return { condition: "timeout", reasonPhrase }; - } - const condition = missing ? "does_not_exist" : e instanceof SessionError ? sessionVerdict(e.code) : "internal"; - return { condition, reasonPhrase, retryable: routeFailed(e) }; + e instanceof StreamError && + (e.code === StreamCode.Unroutable || (e.code === StreamCode.NotFound && !resolved)); + if (wait && missing) return { condition: "timeout", reasonPhrase }; + return { condition: refusedAs(e), reasonPhrase, retryable: routeFailed(e) }; } /**