fix(stream): fetch a block again after a retriable failure

The SDK retries a request that fails before its response starts, but not a
block body cut off part way through. That error stopped playback or failed
a download, and the SDK's new keep-alive pings make it the way a dead
connection now ends: seconds in, rather than at the storage timeout.

A block is now fetched up to three times, 250 ms and then 500 ms apart,
while ProtonError::is_retriable says a replay could succeed. A permanent
refusal still fails on the first attempt.

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<T, F, Fut>(
+    first_delay: Duration,
+    mut fetch: F,
+) -> std::result::Result<T, ProtonError>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = std::result::Result<T, ProtonError>>,
+{
+    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<ProtonError>,
+    ) -> (std::result::Result<u8, ProtonError>, 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