Skip to content

fix(media): restore cached range responses #1559

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
eleboucher wants to merge 1 commit into
base: dev
Choose a base branch
from
Open

fix(media): restore cached range responses #1559

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
181 changes: 175 additions & 6 deletions src-tauri/src/network/media_protocol.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
use std::{
collections::HashMap,
fs,
io::Read,
io::{Read, Seek, SeekFrom},
path::{Path, PathBuf},
sync::{
atomic::{AtomicBool, AtomicU64, Ordering},
Expand Down Expand Up @@ -50,6 +50,7 @@ const MAX_CONCURRENT_DOWNLOAD_REQUESTS: usize = 6;
const SESSION_WAIT: Duration = Duration::from_secs(5);
const SESSION_REFRESH_WAIT: Duration = Duration::from_millis(2500);
const MAX_CACHE_BYTES: u64 = 512 * 1024 * 1024;
const MAX_RANGE_CHUNK: u64 = 2 * 1024 * 1024;
// Short: a 4xx stops being true once an unreachable remote server comes back.
const NEGATIVE_CACHE_TTL: Duration = Duration::from_secs(600);
const MAX_NEGATIVE_CACHE_ENTRIES: usize = 512;
Expand Down Expand Up @@ -475,9 +476,10 @@ pub fn respond<R: Runtime>(
.and_then(|value| value.to_str().ok())
.map(str::to_owned)
};
let range = header_value(header::RANGE);
let origin = header_value(header::ORIGIN);
tauri::async_runtime::spawn(async move {
let mut response = handle_request(&app, uri)
let mut response = handle_request(&app, uri, range)
.await
.unwrap_or_else(error_response);
apply_cors_headers(&mut response, origin.as_deref());
Expand All @@ -488,6 +490,7 @@ pub fn respond<R: Runtime>(
async fn handle_request<R: Runtime>(
app: &AppHandle<R>,
uri: Uri,
range: Option<String>,
) -> Result<Response<Vec<u8>>, StatusCode> {
let target = percent_encoding::percent_decode_str(uri.path().trim_start_matches('/'))
.decode_utf8()
Expand Down Expand Up @@ -529,12 +532,16 @@ async fn handle_request<R: Runtime>(
let (content_type, in_memory_body, disk_path) =
ensure_cached(&state, &session, &key, media_url, dir, temp_dir).await?;

match in_memory_body {
Some(body) => {
match (range, in_memory_body) {
(Some(range_header), Some(body)) => {
Ok(serve_range_memory(&body, &content_type, &range_header))
}
(Some(range_header), None) => serve_range(disk_path, content_type, range_header).await,
(None, Some(body)) => {
let vec_body = Arc::try_unwrap(body).unwrap_or_else(|b| (*b).clone());
Ok(ok_response(vec_body, &content_type))
}
None => Ok(ok_response(read_full(disk_path).await?, &content_type)),
(None, None) => Ok(ok_response(read_full(disk_path).await?, &content_type)),
}
}

Expand Down Expand Up @@ -1087,6 +1094,72 @@ async fn read_full(body_path: PathBuf) -> Result<Vec<u8>, StatusCode> {
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
}

async fn serve_range(
body_path: PathBuf,
content_type: String,
range_header: String,
) -> Result<Response<Vec<u8>>, StatusCode> {
tokio::task::spawn_blocking(move || {
let mut file = fs::File::open(&body_path).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let total = file
.metadata()
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.len();
let Some((start, end)) = parse_range(&range_header, total) else {
return Ok(range_not_satisfiable(total));
};
let end = end.min(start.saturating_add(MAX_RANGE_CHUNK - 1));
let length = end - start + 1;
let mut body = vec![0; length as usize];
file.seek(SeekFrom::Start(start))
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
file.read_exact(&mut body)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(partial_response(body, &content_type, start, end, total))
})
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
}

fn serve_range_memory(body: &[u8], content_type: &str, range_header: &str) -> Response<Vec<u8>> {
let total = body.len() as u64;
let Some((start, end)) = parse_range(range_header, total) else {
return range_not_satisfiable(total);
};
let end = end.min(start.saturating_add(MAX_RANGE_CHUNK - 1));
partial_response(
body[start as usize..=end as usize].to_vec(),
content_type,
start,
end,
total,
)
}

fn parse_range(range_header: &str, total: u64) -> Option<(u64, u64)> {
if total == 0 {
return None;
}
let spec = range_header.strip_prefix("bytes=")?;
let (start, end) = spec.split(',').next()?.trim().split_once('-')?;
let (start, end) = if start.is_empty() {
let suffix = end.parse::<u64>().ok()?;
if suffix == 0 {
return None;
}
(total.saturating_sub(suffix), total - 1)
} else {
let start = start.parse::<u64>().ok()?;
let end = if end.is_empty() {
total - 1
} else {
end.parse::<u64>().ok()?.min(total - 1)
};
(start, end)
};
(start <= end && start < total).then_some((start, end))
}

// Trim oldest-first when the cache exceeds its byte budget.
fn evict_directory_if_needed(dir: &Path, max_bytes: u64) {
let Ok(entries) = fs::read_dir(dir) else {
Expand Down Expand Up @@ -1123,7 +1196,7 @@ fn evict_directory_if_needed(dir: &Path, max_bytes: u64) {
}

// Shared response headers. Media is content-addressed and the URL is session-scoped, so
// it is safe to let the webview cache it as immutable.
// it is safe to let the webview cache it as immutable and advertise byte-range support.
fn media_response_builder(status: StatusCode, content_type: &str) -> ResponseBuilder {
let cache_control = if content_type == "application/octet-stream" {
"no-store"
Expand All @@ -1133,6 +1206,7 @@ fn media_response_builder(status: StatusCode, content_type: &str) -> ResponseBui
Response::builder()
.status(status)
.header(header::CONTENT_TYPE, content_type)
.header(header::ACCEPT_RANGES, "bytes")
.header(header::CACHE_CONTROL, cache_control)
.header(header::X_CONTENT_TYPE_OPTIONS, "nosniff")
.header(
Expand All @@ -1149,6 +1223,32 @@ fn ok_response(body: Vec<u8>, content_type: &str) -> Response<Vec<u8>> {
.expect("failed to build media response")
}

fn partial_response(
body: Vec<u8>,
content_type: &str,
start: u64,
end: u64,
total: u64,
) -> Response<Vec<u8>> {
let content_length = body.len();
media_response_builder(StatusCode::PARTIAL_CONTENT, content_type)
.header(header::CONTENT_LENGTH, content_length)
.header(
header::CONTENT_RANGE,
format!("bytes {start}-{end}/{total}"),
)
.body(body)
.expect("failed to build partial media response")
}

fn range_not_satisfiable(total: u64) -> Response<Vec<u8>> {
Response::builder()
.status(StatusCode::RANGE_NOT_SATISFIABLE)
.header(header::CONTENT_RANGE, format!("bytes */{total}"))
.body(Vec::new())
.expect("failed to build 416 response")
}

fn session_unavailable_response() -> Response<Vec<u8>> {
Response::builder()
.status(StatusCode::SERVICE_UNAVAILABLE)
Expand Down Expand Up @@ -1725,6 +1825,75 @@ mod tests {
response.headers().get(header::CACHE_CONTROL).unwrap(),
"private, max-age=31536000, immutable"
);
assert_eq!(
response.headers().get(header::ACCEPT_RANGES).unwrap(),
"bytes"
);
}

#[test]
fn parse_range_supports_video_metadata_and_seek_requests() {
assert_eq!(super::parse_range("bytes=0-499", 1000), Some((0, 499)));
assert_eq!(super::parse_range("bytes=500-", 1000), Some((500, 999)));
assert_eq!(super::parse_range("bytes=-200", 1000), Some((800, 999)));
assert_eq!(super::parse_range("bytes=990-5000", 1000), Some((990, 999)));
}

#[test]
fn malformed_or_unsatisfiable_range_is_rejected() {
for range in [
"bytes=1000-1001",
"bytes=500-400",
"bytes=abc-def",
"items=0-1",
] {
assert_eq!(super::parse_range(range, 1000), None, "{range}");
}
assert_eq!(super::parse_range("bytes=0-0", 0), None);
}

#[test]
fn in_memory_ranges_are_partial_and_capped() {
let body: Vec<u8> = (0..200u8).collect();
let response = super::serve_range_memory(&body, "video/mp4", "bytes=10-19");
assert_eq!(response.status(), StatusCode::PARTIAL_CONTENT);
assert_eq!(response.body(), &body[10..=19]);
assert_eq!(
response.headers().get(header::CONTENT_RANGE).unwrap(),
"bytes 10-19/200"
);

let body = vec![7_u8; (super::MAX_RANGE_CHUNK * 2) as usize];
let response = super::serve_range_memory(&body, "video/mp4", "bytes=0-");
assert_eq!(response.body().len() as u64, super::MAX_RANGE_CHUNK);
assert_eq!(
response
.headers()
.get(header::CONTENT_RANGE)
.unwrap()
.to_str()
.unwrap(),
format!("bytes 0-{}/{}", super::MAX_RANGE_CHUNK - 1, body.len())
);
}

#[tokio::test]
async fn disk_range_reads_only_the_requested_cached_bytes() {
let path = std::env::temp_dir().join(format!(
"sable-media-range-test-{}-{}",
std::process::id(),
TEST_CACHE_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
));
let body: Vec<u8> = (0..200u8).collect();
fs::write(&path, &body).unwrap();

let response = super::serve_range(path.clone(), "video/mp4".into(), "bytes=10-19".into())
.await
.unwrap();
fs::remove_file(path).ok();

assert_eq!(response.status(), StatusCode::PARTIAL_CONTENT);
assert_eq!(response.body(), &body[10..=19]);
}

#[test]
Expand Down
Loading