diff --git a/crates/pstr-stream/src/reader.rs b/crates/pstr-stream/src/reader.rs index 70baa55..62e2b85 100644 --- a/crates/pstr-stream/src/reader.rs +++ b/crates/pstr-stream/src/reader.rs @@ -10,15 +10,23 @@ use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use async_trait::async_trait; use proton_drive_rs::ProtonDrivePublicLinkClient; +use proton_sdk::error::ProtonError; use proton_sdk::ids::NodeUid; use crate::block::{BlockMap, BlockSource, SharedBlocks}; use crate::error::{Error, Result}; use crate::source::RevisionOpener; +/// How many times one block is fetched before its failure reaches the reader. +const BLOCK_ATTEMPTS: u32 = 3; + +/// The pause before a block's second fetch, doubled before each one after. +const BLOCK_RETRY_DELAY: Duration = Duration::from_millis(250); + /// One open revision on a public link. pub struct RevisionBlocks { reader: proton_drive_rs::RevisionReader, @@ -50,7 +58,7 @@ impl BlockSource for RevisionBlocks { ))); }; - let block = self.reader.read_at(start, size).await?; + let block = fetch_retrying(BLOCK_RETRY_DELAY, || self.reader.read_at(start, size)).await?; // A short block here is not a tail — the map says how long it is. It // means the revision changed underneath us or the block table lied, and // serving it would silently shift every later byte. @@ -64,6 +72,38 @@ impl BlockSource for RevisionBlocks { } } +/// Fetch a block, again if the failure is one a replay could get past. +/// +/// The SDK retries a request that fails before its response starts, but not a +/// block body cut off part way through: that error reaches us as it happened, +/// and without this it stops playback or fails a download. Its keep-alive pings +/// turn a silently dead connection into exactly that error, seconds in rather +/// than at the storage timeout, and the next attempt goes out on a new +/// connection. [`ProtonError::is_retriable`] decides what counts, so a +/// permanent refusal still fails on the first attempt. +async fn fetch_retrying( + first_delay: Duration, + mut fetch: F, +) -> std::result::Result +where + F: FnMut() -> Fut, + Fut: Future>, +{ + let mut delay = first_delay; + let mut attempt = 1; + loop { + match fetch().await { + Err(error) if attempt < BLOCK_ATTEMPTS && error.is_retriable() => { + tracing::debug!(attempt, %error, "block fetch failed; fetching again"); + tokio::time::sleep(delay).await; + delay *= 2; + attempt += 1; + } + result => return result, + } + } +} + /// A fully-downloaded revision. It uses the same block seam as Proton so mpv /// can seek normally without a second playback path. pub struct FileBlocks { @@ -136,8 +176,60 @@ impl RevisionOpener for LibraryOpener { #[cfg(test)] mod tests { + use std::sync::atomic::{AtomicU32, Ordering}; + + use proton_sdk::api::ResponseCode; + use proton_sdk::error::ProtonApiError; + use super::*; + fn api_error(http_status: u16) -> ProtonError { + ProtonError::Api(ProtonApiError { + code: ResponseCode::Unknown, + http_status, + message: String::new(), + details: None, + }) + } + + /// Run [`fetch_retrying`] over a fetch that fails with `failures` in turn and + /// then succeeds, returning the result and how many fetches it took. + async fn fetch_through( + failures: Vec, + ) -> (std::result::Result, u32) { + let calls = AtomicU32::new(0); + let failures = std::sync::Mutex::new(failures.into_iter()); + let result = fetch_retrying(Duration::ZERO, || { + calls.fetch_add(1, Ordering::Relaxed); + let next = failures.lock().unwrap().next(); + async move { next.map_or(Ok(7), Err) } + }) + .await; + (result, calls.into_inner()) + } + + #[tokio::test] + async fn a_block_that_fails_retriably_is_fetched_again() { + let (result, calls) = fetch_through(vec![api_error(503)]).await; + assert_eq!(result.expect("second fetch succeeds"), 7); + assert_eq!(calls, 2); + } + + #[tokio::test] + async fn a_permanent_refusal_is_not_fetched_again() { + let (result, calls) = fetch_through(vec![api_error(403)]).await; + assert!(result.is_err()); + assert_eq!(calls, 1); + } + + #[tokio::test] + async fn a_block_that_keeps_failing_returns_its_last_error() { + let (result, calls) = + fetch_through((0..BLOCK_ATTEMPTS + 1).map(|_| api_error(503)).collect()).await; + assert!(result.is_err()); + assert_eq!(calls, BLOCK_ATTEMPTS); + } + /// The player hands its stream to mpv's demuxer thread, so everything from /// the opener down has to cross threads. A compile-time guard, because the /// failure mode is an unhelpful `FnOnce is not general enough` error deep in