fix(streaming): support resume and seeking for server downloads (builds on #986)

.github/workflows/media-stream-tests.yml

@@ -0,0 +1,16 @@
+name: Media Stream Tests
+permissions:
+  contents: read
+on:
+  pull_request:
+    paths: ["rust-api/**", ".github/workflows/media-stream-tests.yml"]
+  workflow_dispatch:
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+      - uses: hecrj/setup-rust-action@v2
+        with:
+          rust-version: "1.99.0"
+      - run: cargo test --locked --manifest-path rust-api/Cargo.toml handlers::media_stream

rust-api/src/handlers/media_stream.rs

@@ -0,0 +1,257 @@
+use axum::{
+    body::Body,
+    http::{header, HeaderMap, Method, Request, StatusCode},
+    response::Response,
+};
+use tower::ServiceExt;
+use tower_http::services::ServeFile;
+
+use crate::error::AppError;
+
+// The caller must authorize access to this path before serving it.
+pub(super) async fn serve_episode_file(
+    path: &str,
+    method: Method,
+    headers: &HeaderMap,
+) -> Result<Response, AppError> {
+    let mut request = Request::builder().method(method.clone()).uri("/");
+    for name in [
+        header::RANGE,
+        header::IF_RANGE,
+        header::IF_MODIFIED_SINCE,
+        header::IF_NONE_MATCH,
+        header::IF_UNMODIFIED_SINCE,
+        header::IF_MATCH,
+    ] {
+        // RFC 9110 defines Range only for GET, including when HEAD probes a file.
+        if method != Method::GET && (name == header::RANGE || name == header::IF_RANGE) {
+            continue;
+        }
+        for value in headers.get_all(&name) {
+            request = request.header(name.clone(), value);
+        }
+    }
+    let request = request
+        .body(Body::empty())
+        .map_err(|e| AppError::external_error(format!("Failed to build request: {e}")))?;
+    let retry_headers = request.headers().clone();
+    let response = ServeFile::new(path)
+        .oneshot(request)
+        .await
+        .map_err(|e| AppError::external_error(format!("Failed to serve file: {e}")))?;
+    // ServeFile evaluates cache preconditions but does not implement If-Range.
+    // Only a matching strong validator permits combining partial responses.
+    if let Some(validator) = headers.get(header::IF_RANGE) {
+        let matching = !validator.as_bytes().starts_with(b"W/")
+            && (response.headers().get(header::ETAG) == Some(validator)
+                || response.headers().get(header::LAST_MODIFIED) == Some(validator));
+        if !matching
+            && matches!(
+                response.status(),
+                StatusCode::PARTIAL_CONTENT | StatusCode::RANGE_NOT_SATISFIABLE
+            )
+        {
+            let mut request = Request::builder()
+                .method(method)
+                .uri("/")
+                .body(Body::empty())
+                .map_err(|e| AppError::external_error(format!("Failed to build request: {e}")))?;
+            *request.headers_mut() = retry_headers;
+            request.headers_mut().remove(header::RANGE);
+            request.headers_mut().remove(header::IF_RANGE);
+            return ServeFile::new(path)
+                .oneshot(request)
+                .await
+                .map(|response| response.map(Body::new))
+                .map_err(|e| AppError::external_error(format!("Failed to serve file: {e}")));
+        }
+    }
+    Ok(response.map(Body::new))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::{body::to_bytes, http::StatusCode};
+    use std::path::PathBuf;
+
+    struct EpisodeFile(PathBuf);
+
+    impl EpisodeFile {
+        async fn new() -> Self {
+            let path = std::env::temp_dir().join(format!("pinepods-{}.mp3", uuid::Uuid::new_v4()));
+            tokio::fs::write(&path, b"0123456789").await.unwrap();
+            Self(path)
+        }
+
+        async fn request(&self, method: Method, pairs: &[(&str, &str)]) -> Response {
+            let mut headers = HeaderMap::new();
+            for &(name, value) in pairs {
+                headers.append(
+                    name.parse::<header::HeaderName>().unwrap(),
+                    value.parse().unwrap(),
+                );
+            }
+            serve_episode_file(self.0.to_str().unwrap(), method, &headers)
+                .await
+                .unwrap()
+        }
+    }
+
+    impl Drop for EpisodeFile {
+        fn drop(&mut self) {
+            let _ = std::fs::remove_file(&self.0);
+        }
+    }
+
+    #[tokio::test]
+    async fn full_file_and_seek_ranges_return_the_requested_bytes() {
+        let file = EpisodeFile::new().await;
+        let full = file.request(Method::GET, &[]).await;
+        assert_eq!(full.status(), StatusCode::OK);
+        assert_eq!(full.headers()[header::CONTENT_LENGTH], "10");
+        assert_eq!(to_bytes(full.into_body(), 10).await.unwrap(), "0123456789");
+
+        for (range, content_range, expected) in [
+            ("bytes=2-5", "bytes 2-5/10", "2345"),
+            ("bytes=7-", "bytes 7-9/10", "789"),
+            ("bytes=-2", "bytes 8-9/10", "89"),
+        ] {
+            let response = file.request(Method::GET, &[("range", range)]).await;
+            assert_eq!(response.status(), StatusCode::PARTIAL_CONTENT);
+            assert_eq!(response.headers()[header::CONTENT_RANGE], content_range);
+            assert_eq!(to_bytes(response.into_body(), 10).await.unwrap(), expected);
+        }
+    }
+
+    #[tokio::test]
+    async fn head_ignores_bounded_and_unsatisfiable_ranges() {
+        let file = EpisodeFile::new().await;
+        for range in ["bytes=2-5", "bytes=20-"] {
+            let response = file.request(Method::HEAD, &[("range", range)]).await;
+            assert_eq!(response.status(), StatusCode::OK);
+            assert_eq!(response.headers()[header::CONTENT_LENGTH], "10");
+            assert!(!response.headers().contains_key(header::CONTENT_RANGE));
+            assert!(to_bytes(response.into_body(), 10).await.unwrap().is_empty());
+        }
+    }
+
+    #[tokio::test]
+    async fn if_range_requires_a_matching_strong_etag() {
+        let file = EpisodeFile::new().await;
+        let full = file.request(Method::HEAD, &[]).await;
+        let etag = full.headers()[header::ETAG].to_str().unwrap();
+        let weak = format!("W/{etag}");
+        for (validator, expected) in [
+            (etag, StatusCode::PARTIAL_CONTENT),
+            (weak.as_str(), StatusCode::OK),
+            ("\"stale\"", StatusCode::OK),
+        ] {
+            let response = file
+                .request(
+                    Method::GET,
+                    &[("range", "bytes=2-5"), ("if-range", validator)],
+                )
+                .await;
+            assert_eq!(response.status(), expected);
+            let expected_body = if expected == StatusCode::PARTIAL_CONTENT {
+                "2345"
+            } else {
+                "0123456789"
+            };
+            assert_eq!(
+                to_bytes(response.into_body(), 10).await.unwrap(),
+                expected_body
+            );
+        }
+        for (validator, expected) in [
+            (etag, StatusCode::RANGE_NOT_SATISFIABLE),
+            ("\"stale\"", StatusCode::OK),
+        ] {
+            let response = file
+                .request(
+                    Method::GET,
+                    &[("range", "bytes=20-"), ("if-range", validator)],
+                )
+                .await;
+            assert_eq!(response.status(), expected);
+        }
+    }
+
+    #[tokio::test]
+    async fn cache_preconditions_take_precedence_over_ranges() {
+        let file = EpisodeFile::new().await;
+        let full = file.request(Method::HEAD, &[]).await;
+        let etag = full.headers()[header::ETAG].to_str().unwrap();
+        let cached = file
+            .request(
+                Method::GET,
+                &[("range", "bytes=2-5"), ("if-none-match", etag)],
+            )
+            .await;
+        assert_eq!(cached.status(), StatusCode::NOT_MODIFIED);
+        let failed = file
+            .request(
+                Method::GET,
+                &[("range", "bytes=2-5"), ("if-match", "\"stale\"")],
+            )
+            .await;
+        assert_eq!(failed.status(), StatusCode::PRECONDITION_FAILED);
+    }
+
+    #[tokio::test]
+    async fn out_of_bounds_range_is_rejected() {
+        let file = EpisodeFile::new().await;
+        let response = file.request(Method::GET, &[("range", "bytes=20-")]).await;
+        assert_eq!(response.status(), StatusCode::RANGE_NOT_SATISFIABLE);
+        assert_eq!(response.headers()[header::CONTENT_RANGE], "bytes */10");
+    }
+
+    #[tokio::test]
+    async fn stale_if_range_returns_the_full_resource() {
+        let file = EpisodeFile::new().await;
+        let response = file
+            .request(
+                Method::GET,
+                &[
+                    ("range", "bytes=2-5"),
+                    ("if-range", "Thu, 01 Jan 1970 00:00:00 GMT"),
+                ],
+            )
+            .await;
+        assert_eq!(response.status(), StatusCode::OK);
+        assert!(!response.headers().contains_key(header::CONTENT_RANGE));
+        assert_eq!(
+            to_bytes(response.into_body(), 10).await.unwrap(),
+            "0123456789"
+        );
+    }
+
+    #[tokio::test]
+    async fn matching_validator_and_head_preserve_file_semantics() {
+        let file = EpisodeFile::new().await;
+        let response = file.request(Method::HEAD, &[]).await;
+        assert_eq!(response.status(), StatusCode::OK);
+        assert_eq!(response.headers()[header::CONTENT_LENGTH], "10");
+        let modified = response.headers()[header::LAST_MODIFIED]
+            .to_str()
+            .unwrap()
+            .to_owned();
+        assert!(to_bytes(response.into_body(), 10).await.unwrap().is_empty());
+
+        let partial = file
+            .request(
+                Method::GET,
+                &[("range", "bytes=2-5"), ("if-range", &modified)],
+            )
+            .await;
+        assert_eq!(partial.status(), StatusCode::PARTIAL_CONTENT);
+        assert_eq!(to_bytes(partial.into_body(), 10).await.unwrap(), "2345");
+
+        let cached = file
+            .request(Method::GET, &[("if-modified-since", &modified)])
+            .await;
+        assert_eq!(cached.status(), StatusCode::NOT_MODIFIED);
+        assert!(to_bytes(cached.into_body(), 10).await.unwrap().is_empty());
+    }
+}

rust-api/src/handlers/mod.rs

@@ -1,3 +1,4 @@
+mod media_stream;
 pub mod auth;
 pub mod health;
 pub mod podcasts;

rust-api/src/handlers/podcasts.rs

@@ -3265,6 +3265,8 @@ pub async fn stream_episode(
     State(state): State<crate::AppState>,
     Path(episode_id): Path<i32>,
     Query(query): Query<StreamQuery>,
+    headers: HeaderMap,
+    method: axum::http::Method,
 ) -> Result<axum::response::Response, AppError> {
     let api_key = &query.api_key;
     info!("Stream request for episode {} with api_key {} and user_id {}", episode_id, api_key, query.user_id);
@@ -3350,26 +3352,7 @@ pub async fn stream_episode(
     if let Some(path) = file_path {
         debug!("Found file at: {}", path);
 
-        // Use tower_http's ServeFile for proper file serving with range support
-        use tower_http::services::ServeFile;
-        use tower::ServiceExt;
-
-        let service = ServeFile::new(&path);
-        let request = axum::http::Request::builder()
-            .method("GET")
-            .uri("/")
-            .body(axum::body::Body::empty())
-            .map_err(|e| AppError::external_error(&format!("Failed to build request: {}", e)))?;
-
-        let response = service.oneshot(request).await
-            .map_err(|e| AppError::external_error(&format!("Failed to serve file: {}", e)))?;
-
-        // Convert the response body to the expected type
-        let (parts, body) = response.into_parts();
-        let body = axum::body::Body::new(body);
-        let response = axum::response::Response::from_parts(parts, body);
-
-        Ok(response)
+        super::media_stream::serve_episode_file(&path, method, &headers).await
     } else {
         Err(AppError::not_found("Episode not found or not downloaded"))
     }