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 bce722051da..bdfbd9f57ae 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -420,6 +420,49 @@ 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, errorCode, retryInterval] of [ + [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); + 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); + const refused = await RequestError.decode(wire, version); + expect([refused.errorCode, refused.retryInterval]).toEqual([errorCode, 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 96aa319d289..bf894ab6273 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -27,6 +27,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 { @@ -48,7 +49,7 @@ import * as Varint from "../varint.ts"; import { type Advertised, type Advertisements, sameInstance, wireOf } from "../wire.ts"; import * as Cluster from "./cluster.ts"; import { encodeDatagram } from "./datagram.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"; @@ -311,6 +312,39 @@ interface RunFill { streams: StreamCount; } +// 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"; +} + /** * Handles publishing broadcasts using moq-transport protocol. * Uses the stream-per-request pattern (real bidi streams for v17, virtual for v14-v16). @@ -2708,16 +2742,18 @@ 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, resolved = false): Refusal { const e = error(err); - const condition = - e instanceof StreamError && (e.code === StreamCode.NotFound || e.code === StreamCode.Unroutable) - ? wait && (!resolved || e.code === StreamCode.Unroutable) - ? "timeout" - : "does_not_exist" - : "internal"; - return { condition, reasonPhrase: reason(e) }; + const reasonPhrase = reason(e); + const missing = + 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) }; } /** diff --git a/js/net/src/ietf/request_stream.ts b/js/net/src/ietf/request_stream.ts index fb27a97ee3a..a2c12ca7116 100644 --- a/js/net/src/ietf/request_stream.ts +++ b/js/net/src/ietf/request_stream.ts @@ -1,3 +1,4 @@ +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"; @@ -30,12 +31,19 @@ function finCancels(version: IetfVersion): boolean { export interface Refusal { condition: RequestCondition; reasonPhrase: string; + /** 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 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, + retryInterval: refusal.retryable && open ? 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 969b137f282..b9aa9f6824a 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 e305e025b6c..1fb7652a95b 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -594,6 +594,41 @@ impl Error { _ => Self::Transport(err.to_string()), } } + + /// 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() + } + + /// 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)) + } + + /// 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 + ) + ) + } } /// Preserve the session registry when carrying a protocol error. diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index fbf3ecbd519..cb13f6c23ce 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -248,6 +248,19 @@ 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, 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() && matches!(code, INTERNAL_ERROR | DOES_NOT_EXIST | GOING_AWAY) { + 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 +317,29 @@ 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, 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; + for version in ALL { + for kind in KINDS { + for err in EVERY_ERROR { + let code = to_code(&err, kind, version); + let expected = match err { + Error::GoingAway => 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 (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( + crate::time::sim(), + crate::driver::Transport::new(ScriptedSession::new(Vec::new())), + origin.consume(), + Control::new(false), + None, + peer_setup, + version, + ); + let session = ScriptedSession::new(Vec::new()); + let stream = Stream::open(&crate::driver::Transport::new(session.clone()), version) + .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::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 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} {kind:?}: the refusal never went out")); + + 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} {kind:?}"); + let error = ietf::RequestError::decode(&mut buf, version).unwrap(); + (error.error_code, error.retry_interval) + } + + /// 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_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 (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}"); + } + } + } + } + /// 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.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 c6bc9c98950..a9c97865835 100644 --- a/rs/moq-net/src/ietf/request_stream.rs +++ b/rs/moq-net/src/ietf/request_stream.rs @@ -323,14 +323,8 @@ 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, - }, - ) + 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 f0256e50404..f4026a20324 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -1749,17 +1749,10 @@ fn refuse_going_away(stream: Stream, id: u64, version: Versi if id == auth::Auth::ID { return auth::going_away(stream); } + let kind = super::error::request::Kind::Subscribe; 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, - }, + ietf::RequestError::refuse(&Error::GoingAway, kind, "going away", version), ) } @@ -4133,7 +4126,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; @@ -4143,7 +4136,12 @@ 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()); diff --git a/rs/moq-net/src/model/front.rs b/rs/moq-net/src/model/front.rs index 768e0118f00..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 - && super::resume::retryable(&err) - && 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); diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index f378ebc836f..de7f1b46528 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -3883,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, }; 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