From e806abb7c2a06a761f474b9db4293f21454b876c Mon Sep 17 00:00:00 2001 From: vincent Date: Wed, 3 Jun 2026 14:40:13 -0400 Subject: [PATCH 1/5] Defer media body downloads during refresh --- src/groups.rs | 52 +++++------------------ src/media.rs | 112 +++++++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 118 insertions(+), 46 deletions(-) diff --git a/src/groups.rs b/src/groups.rs index bce8683..31d7aa7 100644 --- a/src/groups.rs +++ b/src/groups.rs @@ -172,7 +172,6 @@ async fn refresh_group(group_id: web::Path) -> AppResult "refreshed_files": json!(Vec::::new()), // Initialize empty "all_files": json!(Vec::::new()) // Initialize empty }); - let mut refreshed_files_vec = Vec::new(); let mut all_files_vec: Vec = Vec::new(); if repo.can_write() { @@ -292,46 +291,17 @@ async fn refresh_group(group_id: web::Path) -> AppResult }; repo_info["all_files"] = json!(all_files_vec.clone()); - // For each file, check if it needs to be refreshed - for file_name in &all_files_vec { - match repo.get_file_hash(file_name).await { - Ok(file_hash) => { - if !group.has_hash(&file_hash).await? { - log_debug!( - TAG, - "File {} hash {} not found locally. Downloading...", - file_name, - file_hash - ); - match group.download_hash_from_peers(&file_hash).await { - Ok(_) => { - log_debug!( - TAG, - "Successfully downloaded file hash {} for {}", - file_hash, - file_name - ); - refreshed_files_vec.push(file_name.clone()); - } - Err(e) => { - log_debug!( - TAG, - "Error downloading file {} hash {}: {}", - file_name, - file_hash, - e - ); - // Optionally add to a list of files that failed to download - } - } - } - } - Err(e) => { - log_debug!(TAG, "Error getting hash for file {}: {}", file_name, e); - } - } - } - repo_info["refreshed_files"] = json!(refreshed_files_vec); + // Keep refresh metadata-only. Downloading every missing file body here + // can block later file discovery behind one slow or failing transfer. + // `refreshed_files` is retained for API compatibility; file bodies are + // now refreshed only by the explicit media endpoints. + log_debug!( + TAG, + "Repo {} refresh discovered {} files; body downloads are deferred to media endpoint.", + repo.id(), + all_files_vec.len() + ); + repo_info["refreshed_files"] = json!(Vec::::new()); } Ok(Err(e)) => { log_debug!(TAG, "Error getting repo hash for {}: {}", repo.id(), e); diff --git a/src/media.rs b/src/media.rs index 3407fc1..c310caf 100644 --- a/src/media.rs +++ b/src/media.rs @@ -12,6 +12,108 @@ use futures::Stream; use futures::StreamExt; use serde_json::json; use std::io; +use std::time::{Duration, Instant}; + +const MEDIA_DOWNLOAD_MAX_ATTEMPTS: u32 = 3; +const MEDIA_DOWNLOAD_PER_PEER_TIMEOUT: Duration = Duration::from_secs(18); +const MEDIA_DOWNLOAD_OVERALL_TIMEOUT: Duration = Duration::from_secs(55); +const MEDIA_DOWNLOAD_INITIAL_BACKOFF: Duration = Duration::from_millis(500); + +// TODO: fold this media-specific overall budget into save-dweb-backend's +// Group::download_hash_from_peers once the backend crate exposes that knob. +async fn download_hash_for_media( + group: &save_dweb_backend::group::Group, + hash: &iroh_blobs::Hash, +) -> AppResult<()> { + let mut last_error = None; + let started = Instant::now(); + + for attempt in 1..=MEDIA_DOWNLOAD_MAX_ATTEMPTS { + let mut peer_repos = group.list_peer_repos().await; + if peer_repos.is_empty() { + return Err(anyhow::anyhow!("Cannot download hash. No other peers found").into()); + } + + let peer_count = peer_repos.len(); + peer_repos.rotate_left((attempt as usize - 1) % peer_count); + + for peer_repo in peer_repos { + let peer_id = peer_repo.id().to_string(); + + let Some(remaining) = MEDIA_DOWNLOAD_OVERALL_TIMEOUT.checked_sub(started.elapsed()) + else { + let detail = format!( + "Timed out downloading hash {hash} after overall {}s media budget", + MEDIA_DOWNLOAD_OVERALL_TIMEOUT.as_secs() + ); + log_info!(TAG, "{}", detail); + let last_detail = match last_error { + Some(error) => format!("{detail}; last peer error: {error}"), + None => detail, + }; + return Err(anyhow::anyhow!( + "Unable to download hash {} from any peer after {} media attempts; last error: {}", + hash, + attempt.saturating_sub(1), + last_detail + ) + .into()); + }; + + log_info!( + TAG, + "Media download attempt {}/{} for hash {} from peer {}", + attempt, + MEDIA_DOWNLOAD_MAX_ATTEMPTS, + hash, + peer_id + ); + + let timeout_budget = std::cmp::min(MEDIA_DOWNLOAD_PER_PEER_TIMEOUT, remaining); + let download_result = tokio::time::timeout(timeout_budget, async { + let route_id_blob = peer_repo.get_route_id_blob().await?; + group + .iroh_blobs + .download_file_from(route_id_blob, hash) + .await + }) + .await; + + match download_result { + Ok(Ok(())) => return Ok(()), + Ok(Err(e)) => { + let detail = format!("Unable to download hash {hash} from peer {peer_id}: {e}"); + log_info!(TAG, "{}", detail); + last_error = Some(detail); + } + Err(_) => { + let detail = format!( + "Timed out downloading hash {hash} from peer {peer_id} after {}ms", + timeout_budget.as_millis() + ); + log_info!(TAG, "{}", detail); + last_error = Some(detail); + } + } + } + + if attempt < MEDIA_DOWNLOAD_MAX_ATTEMPTS { + let backoff = MEDIA_DOWNLOAD_INITIAL_BACKOFF * attempt; + if let Some(remaining) = MEDIA_DOWNLOAD_OVERALL_TIMEOUT.checked_sub(started.elapsed()) { + tokio::time::sleep(std::cmp::min(backoff, remaining)).await; + } + } + } + + let detail = last_error.unwrap_or_else(|| "no peer attempts completed".to_string()); + Err(anyhow::anyhow!( + "Unable to download hash {} from any peer after {} media attempts; last error: {}", + hash, + MEDIA_DOWNLOAD_MAX_ATTEMPTS, + detail + ) + .into()) +} pub fn scope() -> Scope { web::scope("/media") @@ -59,7 +161,7 @@ async fn list_files(path: web::Path) -> AppResult let hash = repo.get_hash_from_dht().await?; if !group.has_hash(&hash).await? { - group.download_hash_from_peers(&hash).await?; + download_hash_for_media(group.as_ref(), &hash).await?; } // List files and check if they are downloaded @@ -71,7 +173,7 @@ async fn list_files(path: web::Path) -> AppResult Ok(hash) => hash, Err(_) => continue, // Handle the error or skip the file if there's an issue }; - let is_downloaded = repo.has_hash(&file_hash).await.unwrap_or(false); // Check if the file is downloaded + let is_downloaded = group.has_hash(&file_hash).await.unwrap_or(false); // Check if the file is local files_with_status.push(json!({ "name": file_name, "hash": file_hash, @@ -100,15 +202,15 @@ async fn download_file(path: web::Path) -> AppResult Date: Wed, 3 Jun 2026 14:50:10 -0400 Subject: [PATCH 2/5] Add refresh metadata regression coverage --- src/lib.rs | 53 +++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 49 insertions(+), 4 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 8d57264..74b06e0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1355,10 +1355,8 @@ mod tests { .as_array() .expect("refreshed_files should be an array"); assert!( - refreshed_files.is_empty() - || (refreshed_files.len() == 1 - && refreshed_files[0].as_str() == Some(file_name)), - "First refresh should report either no-op or one refreshed expected file, got {refreshed_files:?}" + refreshed_files.is_empty(), + "Refresh should discover metadata without downloading media bodies, got {refreshed_files:?}" ); let all_files = repo_data["all_files"] @@ -1371,6 +1369,29 @@ mod tests { "all_files should contain the uploaded file" ); + let list_files_req = test::TestRequest::get() + .uri(&format!( + "/api/groups/{}/repos/{}/media", + group.id(), + refreshed_repo_id + )) + .to_request(); + let list_files_resp = test::call_service(&app, list_files_req).await; + assert!( + list_files_resp.status().is_success(), + "File list should be accessible after metadata-only refresh" + ); + let list_files_data: FilesResponse = test::read_body_json(list_files_resp).await; + let listed_file = list_files_data + .files + .iter() + .find(|file| file.name == file_name) + .expect("File list should include metadata for the uploaded file"); + assert!( + !listed_file.is_downloaded, + "Refresh should not mark the file body as downloaded before explicit media GET" + ); + // Verify file is accessible after refresh let get_file_req = test::TestRequest::get() .uri(&format!( @@ -1392,6 +1413,30 @@ mod tests { "File content should match after refresh" ); + let list_files_after_get_req = test::TestRequest::get() + .uri(&format!( + "/api/groups/{}/repos/{}/media", + group.id(), + refreshed_repo_id + )) + .to_request(); + let list_files_after_get_resp = test::call_service(&app, list_files_after_get_req).await; + assert!( + list_files_after_get_resp.status().is_success(), + "File list should still be accessible after explicit media GET" + ); + let list_files_after_get_data: FilesResponse = + test::read_body_json(list_files_after_get_resp).await; + let listed_file_after_get = list_files_after_get_data + .files + .iter() + .find(|file| file.name == file_name) + .expect("File list should still include the uploaded file"); + assert!( + listed_file_after_get.is_downloaded, + "Explicit media GET should mark the file body as downloaded locally" + ); + // Test second refresh - should be no-op since all files are present let refresh_req2 = test::TestRequest::post() .uri(&format!("/api/groups/{}/refresh", group.id())) From 2a96c38e711530b5306a76f9ebd860cad9884641 Mon Sep 17 00:00:00 2001 From: vincent Date: Wed, 3 Jun 2026 15:10:20 -0400 Subject: [PATCH 3/5] Strengthen metadata refresh regression --- src/lib.rs | 101 +++++++++++++++++++++++++++++++++-------------------- 1 file changed, 63 insertions(+), 38 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 74b06e0..5990b92 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1392,6 +1392,69 @@ mod tests { "Refresh should not mark the file body as downloaded before explicit media GET" ); + // Test second refresh before downloading the file body. It should remain metadata-only. + let refresh_req2 = test::TestRequest::post() + .uri(&format!("/api/groups/{}/refresh", group.id())) + .to_request(); + let refresh_resp2 = test::call_service(&app, refresh_req2).await; + assert!( + refresh_resp2.status().is_success(), + "Second refresh should succeed before explicit media GET" + ); + + let refresh_data2: serde_json::Value = test::read_body_json(refresh_resp2).await; + assert_eq!( + refresh_data2["status"], "success", + "Second refresh status should be success" + ); + + let repos2 = refresh_data2["repos"] + .as_array() + .expect("repos should be an array"); + assert!(!repos2.is_empty(), "Should still have repos"); + + let repo_data2 = repos2 + .iter() + .find(|r| { + r["all_files"] + .as_array() + .map(|files| files.iter().any(|f| f.as_str() == Some(file_name))) + .unwrap_or(false) + }) + .expect("Should still find the repo containing the uploaded file on second refresh"); + let refreshed_files2 = repo_data2["refreshed_files"] + .as_array() + .expect("refreshed_files should be an array"); + assert!( + refreshed_files2.is_empty(), + "Repeated refresh should still not download media bodies, got {refreshed_files2:?}" + ); + + let list_files_after_second_refresh_req = test::TestRequest::get() + .uri(&format!( + "/api/groups/{}/repos/{}/media", + group.id(), + refreshed_repo_id + )) + .to_request(); + let list_files_after_second_refresh_resp = + test::call_service(&app, list_files_after_second_refresh_req).await; + assert!( + list_files_after_second_refresh_resp.status().is_success(), + "File list should still be accessible after repeated metadata-only refresh" + ); + let list_files_after_second_refresh_data: FilesResponse = + test::read_body_json(list_files_after_second_refresh_resp).await; + let listed_file_after_second_refresh = list_files_after_second_refresh_data + .files + .iter() + .find(|file| file.name == file_name) + .expect("File list should still include metadata for the uploaded file"); + assert!( + !listed_file_after_second_refresh.is_downloaded, + "Repeated refresh should not mark the file body as downloaded before explicit media GET" + ); + // Verify file is accessible after refresh let get_file_req = test::TestRequest::get() .uri(&format!( @@ -1437,44 +1500,6 @@ mod tests { "Explicit media GET should mark the file body as downloaded locally" ); - // Test second refresh - should be no-op since all files are present - let refresh_req2 = test::TestRequest::post() - .uri(&format!("/api/groups/{}/refresh", group.id())) - .to_request(); - let refresh_resp2 = test::call_service(&app, refresh_req2).await; - assert!( - refresh_resp2.status().is_success(), - "Second refresh should succeed" - ); - - let refresh_data2: serde_json::Value = test::read_body_json(refresh_resp2).await; - assert_eq!( - refresh_data2["status"], "success", - "Second refresh status should be success" - ); - - let repos2 = refresh_data2["repos"] - .as_array() - .expect("repos should be an array"); - assert!(!repos2.is_empty(), "Should still have repos"); - - let repo_data2 = repos2 - .iter() - .find(|r| { - r["all_files"] - .as_array() - .map(|files| files.iter().any(|f| f.as_str() == Some(file_name))) - .unwrap_or(false) - }) - .expect("Should still find the repo containing the uploaded file on second refresh"); - let refreshed_files2 = repo_data2["refreshed_files"] - .as_array() - .expect("refreshed_files should be an array"); - assert!( - refreshed_files2.is_empty(), - "No files should be refreshed on second call since all are present" - ); - // Clean up both backends - secondary first, then main backend2.stop().await?; tokio::time::sleep(Duration::from_millis(500)).await; From ce1d7c6035a6ec42be7ba39eb2465c0138e2ef9d Mon Sep 17 00:00:00 2001 From: vincent Date: Tue, 9 Jun 2026 11:07:07 -0400 Subject: [PATCH 4/5] Fix empty repo media listing --- CHANGELOG.md | 7 +++++++ Cargo.lock | 35 ++++++++++++++++++----------------- Cargo.toml | 4 ++-- src/lib.rs | 25 +++++++++++++++++++++---- src/media.rs | 20 +++++++++++++++++--- 5 files changed, 65 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d3364d..00e379c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## 2026-06-09 + +- Bump `save` crate version to `0.2.6`. +- Bump `save-dweb-backend` dependency to `v0.3.8`. +- Defer media body downloads during refresh so file metadata remains available when large media transfers are slow or fail. +- Return an empty media list for empty repositories instead of surfacing a DHT root-hash error. + ## 2026-05-31 - Bump `save` crate version to `0.2.5`. diff --git a/Cargo.lock b/Cargo.lock index f45f3b8..91d659f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1636,7 +1636,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -1936,7 +1936,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2854,7 +2854,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.5.10", + "socket2 0.6.2", "tokio", "tower-service", "tracing", @@ -3525,7 +3525,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4230,7 +4230,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -4470,7 +4470,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7d8fae84b431384b68627d0f9b3b1245fcf9f46f6c0e3dc902e9dce64edd1967" dependencies = [ "libc", - "windows-sys 0.45.0", + "windows-sys 0.61.2", ] [[package]] @@ -5069,7 +5069,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls 0.23.36", - "socket2 0.5.10", + "socket2 0.6.2", "thiserror 2.0.18", "tokio", "tracing", @@ -5106,9 +5106,9 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.5.10", + "socket2 0.6.2", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.60.2", ] [[package]] @@ -5615,7 +5615,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -5795,7 +5795,7 @@ dependencies = [ [[package]] name = "save" -version = "0.2.5" +version = "0.2.6" dependencies = [ "actix-web", "anyhow", @@ -5827,8 +5827,8 @@ dependencies = [ [[package]] name = "save-dweb-backend" -version = "0.3.7" -source = "git+https://github.com/OpenArchive/save-dweb-backend?tag=v0.3.7#5c043085b09f1b0fd6da2c5f1b4e9195918353e3" +version = "0.3.8" +source = "git+https://github.com/OpenArchive/save-dweb-backend?tag=v0.3.8#1bb6559fe0df202806d40b6a1b0f43dd7673c89f" dependencies = [ "anyhow", "async-stream", @@ -6791,7 +6791,7 @@ dependencies = [ "getrandom 0.4.1", "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -7515,14 +7515,15 @@ dependencies = [ [[package]] name = "veilid-iroh-blobs" -version = "0.3.4" -source = "git+https://github.com/RangerMauve/veilid-iroh-blobs?tag=v0.3.4#62c64886942f78ef0855fa97c529c240e809eca2" +version = "0.3.5" +source = "git+https://github.com/RangerMauve/veilid-iroh-blobs?tag=v0.3.5#79c00a342691f12d01af426c271b93bfa9f1cf64" dependencies = [ "anyhow", "bytes", "futures-lite", "futures-util", "hex", + "hickory-resolver", "iroh-blobs", "iroh-io", "serde", @@ -7859,7 +7860,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 4d39d83..d12aa95 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "save" -version = "0.2.5" +version = "0.2.6" description = "Decentralized Web for Save" edition = "2021" publish = false @@ -27,7 +27,7 @@ ios = [] # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -save-dweb-backend = { git = "https://github.com/OpenArchive/save-dweb-backend", tag = "v0.3.7" } +save-dweb-backend = { git = "https://github.com/OpenArchive/save-dweb-backend", tag = "v0.3.8" } tokio = { version = "^1.43", default-features = false, features = ["rt", "rt-multi-thread", "sync", "time", "macros"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" diff --git a/src/lib.rs b/src/lib.rs index 5990b92..834f2fe 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -440,7 +440,24 @@ mod tests { .as_str() .expect("No repo key returned"); - // Step 3: Upload a file to the repository + // Step 3: List files before upload; empty repos should not fail with a DHT error. + let empty_list_files_req = test::TestRequest::get() + .uri(&format!("/api/groups/{group_id}/repos/{repo_id}/media")) + .to_request(); + let empty_list_files_resp = test::call_service(&app, empty_list_files_req).await; + assert!( + empty_list_files_resp.status().is_success(), + "Empty repo media list should not fail: {}", + empty_list_files_resp.status() + ); + let empty_list_files_data: FilesResponse = + test::read_body_json(empty_list_files_resp).await; + assert!( + empty_list_files_data.files.is_empty(), + "Empty repo media list should return no files" + ); + + // Step 4: Upload a file to the repository let file_name = "example.txt"; let file_content = b"Test content for file upload"; @@ -455,7 +472,7 @@ mod tests { // Check upload success assert!(upload_resp.status().is_success(), "File upload failed"); - // Step 4: List files in the repository + // Step 5: List files in the repository let list_files_req = test::TestRequest::get() .uri(&format!("/api/groups/{group_id}/repos/{repo_id}/media")) .to_request(); @@ -488,7 +505,7 @@ mod tests { "Downloaded back file content" ); - // Step 5: Delete the file from the repository + // Step 6: Delete the file from the repository let delete_file_req = test::TestRequest::delete() .uri(&format!( "/api/groups/{}/repos/{}/media/{}", @@ -499,7 +516,7 @@ mod tests { assert!(delete_resp.status().is_success(), "File deletion failed"); - // Step 6: Verify the file is no longer listed + // Step 7: Verify the file is no longer listed let list_files_after_deletion_req = test::TestRequest::get() .uri(&format!("/api/groups/{group_id}/repos/{repo_id}/media")) .to_request(); diff --git a/src/media.rs b/src/media.rs index c310caf..2c6bbee 100644 --- a/src/media.rs +++ b/src/media.rs @@ -159,9 +159,23 @@ async fn list_files(path: web::Path) -> AppResult let repo_crypto_key = create_veilid_cryptokey_from_base64(repo_id)?; let repo = group.get_repo(&repo_crypto_key).await?; - let hash = repo.get_hash_from_dht().await?; - if !group.has_hash(&hash).await? { - download_hash_for_media(group.as_ref(), &hash).await?; + if !repo.can_write() { + match repo.get_hash_from_dht().await { + Ok(hash) => { + if !group.has_hash(&hash).await? { + download_hash_for_media(group.as_ref(), &hash).await?; + } + } + Err(err) => { + log_info!( + TAG, + "Repo {} has no published collection hash while listing media; returning empty list: {}", + repo_id, + err + ); + return Ok(HttpResponse::Ok().json(json!({ "files": [] }))); + } + } } // List files and check if they are downloaded From ffb9eb175b1c7d23ca6a3aeee639274b42608850 Mon Sep 17 00:00:00 2001 From: vincent Date: Tue, 9 Jun 2026 12:06:55 -0400 Subject: [PATCH 5/5] Use backend media download budget --- CHANGELOG.md | 2 +- Cargo.lock | 4 +- Cargo.toml | 2 +- src/media.rs | 119 +++++++-------------------------------------------- 4 files changed, 20 insertions(+), 107 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 00e379c..f193cb3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,7 @@ ## 2026-06-09 - Bump `save` crate version to `0.2.6`. -- Bump `save-dweb-backend` dependency to `v0.3.8`. +- Bump `save-dweb-backend` dependency to `v0.3.9`. - Defer media body downloads during refresh so file metadata remains available when large media transfers are slow or fail. - Return an empty media list for empty repositories instead of surfacing a DHT root-hash error. diff --git a/Cargo.lock b/Cargo.lock index 91d659f..1896b83 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5827,8 +5827,8 @@ dependencies = [ [[package]] name = "save-dweb-backend" -version = "0.3.8" -source = "git+https://github.com/OpenArchive/save-dweb-backend?tag=v0.3.8#1bb6559fe0df202806d40b6a1b0f43dd7673c89f" +version = "0.3.9" +source = "git+https://github.com/OpenArchive/save-dweb-backend?tag=v0.3.9#e8966a6356af933a0905ded62dab6f95f9451198" dependencies = [ "anyhow", "async-stream", diff --git a/Cargo.toml b/Cargo.toml index d12aa95..53f3052 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,7 @@ ios = [] # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -save-dweb-backend = { git = "https://github.com/OpenArchive/save-dweb-backend", tag = "v0.3.8" } +save-dweb-backend = { git = "https://github.com/OpenArchive/save-dweb-backend", tag = "v0.3.9" } tokio = { version = "^1.43", default-features = false, features = ["rt", "rt-multi-thread", "sync", "time", "macros"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" diff --git a/src/media.rs b/src/media.rs index 2c6bbee..a079296 100644 --- a/src/media.rs +++ b/src/media.rs @@ -12,108 +12,9 @@ use futures::Stream; use futures::StreamExt; use serde_json::json; use std::io; -use std::time::{Duration, Instant}; +use std::time::Duration; -const MEDIA_DOWNLOAD_MAX_ATTEMPTS: u32 = 3; -const MEDIA_DOWNLOAD_PER_PEER_TIMEOUT: Duration = Duration::from_secs(18); const MEDIA_DOWNLOAD_OVERALL_TIMEOUT: Duration = Duration::from_secs(55); -const MEDIA_DOWNLOAD_INITIAL_BACKOFF: Duration = Duration::from_millis(500); - -// TODO: fold this media-specific overall budget into save-dweb-backend's -// Group::download_hash_from_peers once the backend crate exposes that knob. -async fn download_hash_for_media( - group: &save_dweb_backend::group::Group, - hash: &iroh_blobs::Hash, -) -> AppResult<()> { - let mut last_error = None; - let started = Instant::now(); - - for attempt in 1..=MEDIA_DOWNLOAD_MAX_ATTEMPTS { - let mut peer_repos = group.list_peer_repos().await; - if peer_repos.is_empty() { - return Err(anyhow::anyhow!("Cannot download hash. No other peers found").into()); - } - - let peer_count = peer_repos.len(); - peer_repos.rotate_left((attempt as usize - 1) % peer_count); - - for peer_repo in peer_repos { - let peer_id = peer_repo.id().to_string(); - - let Some(remaining) = MEDIA_DOWNLOAD_OVERALL_TIMEOUT.checked_sub(started.elapsed()) - else { - let detail = format!( - "Timed out downloading hash {hash} after overall {}s media budget", - MEDIA_DOWNLOAD_OVERALL_TIMEOUT.as_secs() - ); - log_info!(TAG, "{}", detail); - let last_detail = match last_error { - Some(error) => format!("{detail}; last peer error: {error}"), - None => detail, - }; - return Err(anyhow::anyhow!( - "Unable to download hash {} from any peer after {} media attempts; last error: {}", - hash, - attempt.saturating_sub(1), - last_detail - ) - .into()); - }; - - log_info!( - TAG, - "Media download attempt {}/{} for hash {} from peer {}", - attempt, - MEDIA_DOWNLOAD_MAX_ATTEMPTS, - hash, - peer_id - ); - - let timeout_budget = std::cmp::min(MEDIA_DOWNLOAD_PER_PEER_TIMEOUT, remaining); - let download_result = tokio::time::timeout(timeout_budget, async { - let route_id_blob = peer_repo.get_route_id_blob().await?; - group - .iroh_blobs - .download_file_from(route_id_blob, hash) - .await - }) - .await; - - match download_result { - Ok(Ok(())) => return Ok(()), - Ok(Err(e)) => { - let detail = format!("Unable to download hash {hash} from peer {peer_id}: {e}"); - log_info!(TAG, "{}", detail); - last_error = Some(detail); - } - Err(_) => { - let detail = format!( - "Timed out downloading hash {hash} from peer {peer_id} after {}ms", - timeout_budget.as_millis() - ); - log_info!(TAG, "{}", detail); - last_error = Some(detail); - } - } - } - - if attempt < MEDIA_DOWNLOAD_MAX_ATTEMPTS { - let backoff = MEDIA_DOWNLOAD_INITIAL_BACKOFF * attempt; - if let Some(remaining) = MEDIA_DOWNLOAD_OVERALL_TIMEOUT.checked_sub(started.elapsed()) { - tokio::time::sleep(std::cmp::min(backoff, remaining)).await; - } - } - } - - let detail = last_error.unwrap_or_else(|| "no peer attempts completed".to_string()); - Err(anyhow::anyhow!( - "Unable to download hash {} from any peer after {} media attempts; last error: {}", - hash, - MEDIA_DOWNLOAD_MAX_ATTEMPTS, - detail - ) - .into()) -} pub fn scope() -> Scope { web::scope("/media") @@ -163,7 +64,12 @@ async fn list_files(path: web::Path) -> AppResult match repo.get_hash_from_dht().await { Ok(hash) => { if !group.has_hash(&hash).await? { - download_hash_for_media(group.as_ref(), &hash).await?; + group + .download_hash_from_peers_with_timeout( + &hash, + Some(MEDIA_DOWNLOAD_OVERALL_TIMEOUT), + ) + .await?; } } Err(err) => { @@ -216,7 +122,12 @@ async fn download_file(path: web::Path) -> AppResult) -> AppResult