🐛 fix(pypi): propagate prefetch errors (#1858)

Resident lookups converted storage failures into zero-byte cache hits, while
artifact downloads reopened their materialized path and could panic. Reuse HEAD
metadata and verified transfer lengths so row failures and byte totals match
storage outcomes.

Author: gaborbernat

crates/peryx-ecosystem-pypi/src/cache/download.rs

@@ -32,20 +32,29 @@ pub async fn file_path(
     route: String,
     filename: String,
 ) -> Result<BlobLease, CacheError> {
+    Ok(file_path_with_size(state, digest, route, filename).await?.0)
+}
+
+pub async fn file_path_with_size(
+    state: Arc<ServingState>,
+    digest: Digest,
+    route: String,
+    filename: String,
+) -> Result<(BlobLease, u64), CacheError> {
     ensure_digest_clear(&state, &digest)?;
-    if state.blobs.head(&digest).await?.is_some() {
-        return Ok(state.blobs.materialize(&digest).await?);
+    if let Some(metadata) = state.blobs.head(&digest).await? {
+        return Ok((state.blobs.materialize(&digest).await?, metadata.bytes));
     }
     let mut handle = {
         let gate = flight_gate(&state, digest.as_str());
         let guard = gate.lock_owned().await;
-        if state.blobs.head(&digest).await?.is_some() {
+        if let Some(metadata) = state.blobs.head(&digest).await? {
             release_flight(&state, digest.as_str(), guard);
-            return Ok(state.blobs.materialize(&digest).await?);
+            return Ok((state.blobs.materialize(&digest).await?, metadata.bytes));
         }
-        if fill_remote(&state, &digest).await.is_some() {
+        if let Some(metadata) = fill_remote(&state, &digest).await {
             release_flight(&state, digest.as_str(), guard);
-            return Ok(state.blobs.materialize(&digest).await?);
+            return Ok((state.blobs.materialize(&digest).await?, metadata.bytes));
         }
         let handle = if let Some(running) = existing_download(&state, &digest) {
             running
@@ -55,8 +64,8 @@ pub async fn file_path(
         release_flight(&state, digest.as_str(), guard);
         handle
     };
-    wait_for_download(&mut handle).await?;
-    Ok(state.blobs.materialize(&digest).await?)
+    let bytes = wait_for_download(&mut handle).await?;
+    Ok((state.blobs.materialize(&digest).await?, bytes))
 }
 
 pub enum FileProbe {
@@ -202,11 +211,12 @@ fn existing_download(state: &ServingState, digest: &Digest) -> Option<DownloadHa
     state.downloads.get(digest.as_str())
 }
 
-async fn wait_for_download(handle: &mut DownloadHandle) -> Result<(), CacheError> {
+async fn wait_for_download(handle: &mut DownloadHandle) -> Result<u64, CacheError> {
     loop {
-        let done = handle.progress().borrow_and_update().done.clone();
+        let progress = handle.progress().borrow_and_update().clone();
+        let done = progress.done;
         match done {
-            Some(Ok(())) => return Ok(()),
+            Some(Ok(())) => return Ok(progress.flushed),
             Some(Err(message)) => return Err(CacheError::Stream(message)),
             None => {
                 handle

crates/peryx-ecosystem-pypi/src/cache/mod.rs

@@ -22,6 +22,7 @@ mod resolve;
 mod shadow;
 
 pub(crate) use download::download_dimensions;
+pub(crate) use download::file_path_with_size;
 pub use download::{FileOutcome, FileProbe, file_path, probe_file, stream_file};
 pub use fetch::{
     MAX_PROJECT_BYTES, MAX_PROJECT_FILES, ProjectSyncError, ProjectSyncOutcome, RefreshSummary, refresh_stale_pages,

crates/peryx-ecosystem-pypi/src/mirror/report.rs

@@ -1,9 +1,6 @@
 use std::io::Write;
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use peryx_driver::ServingState;
-use peryx_storage::blob::Digest;
-
 use super::{PrefetchFile, Row};
 
 pub(super) fn write_page_row(
@@ -97,16 +94,6 @@ pub(super) fn write_count(
     )
 }
 
-pub(super) async fn blob_size(state: &ServingState, digest: &Digest) -> u64 {
-    state
-        .blobs
-        .head(digest)
-        .await
-        .ok()
-        .flatten()
-        .map_or(0, |metadata| metadata.bytes)
-}
-
 pub(super) fn unix_now() -> u64 {
     SystemTime::now()
         .duration_since(UNIX_EPOCH)

crates/peryx-ecosystem-pypi/src/mirror/run.rs

@@ -8,9 +8,7 @@ use anyhow::{Context as _, bail};
 use peryx_driver::{AppState, ServingState};
 use peryx_storage::blob::Digest;
 
-use super::report::{
-    blob_size, unix_now, write_count, write_file_row, write_file_row_bytes, write_page_row, write_row,
-};
+use super::report::{unix_now, write_count, write_file_row, write_file_row_bytes, write_page_row, write_row};
 use super::selection::{candidates, content_type_is_json, selection, target};
 use super::{
     BlobCheck, FileCandidate, HEADER, PrefetchConfig, PrefetchFile, PrefetchOptions, Row, Selection, SelectionSource,
@@ -280,16 +278,12 @@ async fn sync_file(
     file: &PrefetchFile,
 ) -> Result<SyncOutcome, crate::cache::CacheError> {
     let digest = Digest::from_hex(&file.digest).ok_or(crate::cache::CacheError::FileNotFound)?;
-    if state.blobs.head(&digest).await?.is_some() {
-        return Ok(SyncOutcome::Cached(blob_size(&state, &digest).await));
+    if let Some(metadata) = state.blobs.head(&digest).await? {
+        return Ok(SyncOutcome::Cached(metadata.bytes));
     }
-    let path = crate::cache::file_path(state, digest.clone(), target.route.clone(), file.filename.clone()).await?;
-    Ok(SyncOutcome::Downloaded(
-        path.path()
-            .metadata()
-            .expect("a blob lease keeps its materialized path available")
-            .len(),
-    ))
+    let (_, bytes) =
+        crate::cache::file_path_with_size(state, digest, target.route.clone(), file.filename.clone()).await?;
+    Ok(SyncOutcome::Downloaded(bytes))
 }
 
 async fn sync_metadata(
@@ -301,8 +295,8 @@ async fn sync_metadata(
 ) -> Result<SyncOutcome, crate::cache::CacheError> {
     let artifact = Digest::from_hex(artifact_digest).ok_or(crate::cache::CacheError::FileNotFound)?;
     let metadata = Digest::from_hex(metadata_digest).ok_or(crate::cache::CacheError::FileNotFound)?;
-    if state.blobs.head(&metadata).await?.is_some() {
-        return Ok(SyncOutcome::Cached(blob_size(state, &metadata).await));
+    if let Some(metadata) = state.blobs.head(&metadata).await? {
+        return Ok(SyncOutcome::Cached(metadata.bytes));
     }
     Ok(SyncOutcome::Downloaded(
         crate::cache::metadata_bytes(state, &artifact, route, metadata_filename)

crates/peryx-ecosystem-pypi/tests/unit/mirror/report_tests.rs

@@ -1,7 +1,4 @@
-use peryx_storage::blob::Digest;
-
-use super::{blob_size, unix_now, write_count, write_file_row, write_file_row_bytes, write_page_row, write_row};
-use crate::mirror::test_support;
+use super::{unix_now, write_count, write_file_row, write_file_row_bytes, write_page_row, write_row};
 use crate::mirror::{PrefetchFile, PrefetchMetadata, Row};
 
 fn file() -> PrefetchFile {
@@ -48,23 +45,6 @@ fn report_writes_each_row_shape() {
     assert!(rows.contains("metadata\tpypi\tdemo\tdemo.metadata"));
 }
 
-#[tokio::test]
-async fn blob_size_reports_present_and_missing_blobs() {
-    let fixture = test_support::state(Vec::new());
-    let present = Digest::of(b"present");
-    fixture
-        .state
-        .serving
-        .blobs
-        .blocking()
-        .put_bytes_as(b"present", &present)
-        .unwrap();
-
-    assert_eq!(blob_size(&fixture.state.serving, &present).await, 7);
-    assert_eq!(blob_size(&fixture.state.serving, &Digest::of(b"missing")).await, 0);
-    assert!(fixture.dir.path().exists());
-}
-
 #[test]
 fn unix_now_is_after_the_epoch() {
     assert!(unix_now() > 0);

crates/peryx-ecosystem-pypi/tests/unit/mirror/run_tests.rs

@@ -284,9 +284,80 @@ async fn sync_downloads_metadata_and_artifacts() {
     .unwrap();
 
     let output = String::from_utf8(output).unwrap();
-    assert!(output.contains("downloaded"));
-    assert!(output.contains("skipped"));
+    assert_eq!(
+        reported_sizes(&output, "downloaded"),
+        [
+            ("metadata", "demo-1.0-py3-none-any.whl.metadata", "8"),
+            ("file", "demo-1.0-py3-none-any.whl", "8"),
+            ("file", "demo-1.0.zip", "5"),
+        ]
+    );
     assert!(output.contains("files_downloaded\t\t\t3"));
+    assert!(output.contains("bytes_downloaded\t\t\t21"));
+    server.verify().await;
+}
+
+#[tokio::test]
+async fn mirror_sync_reports_blob_head_errors_for_artifacts_and_metadata() {
+    let server = MockServer::start().await;
+    let mut detail = artifact_detail(&server.uri());
+    detail.files.truncate(1);
+    Mock::given(method("GET"))
+        .and(path("/simple/demo/"))
+        .respond_with(ResponseTemplate::new(200).set_body_raw(to_json(&detail), "application/vnd.pypi.simple.v1+json"))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let fixture = test_support::state(vec![cached_index(&format!("{}/simple/", server.uri()), false)]);
+    let store = fixture.state.serving.blobs.filesystem_store().unwrap();
+    for digest in [Digest::of(b"artifact"), Digest::of(b"metadata")] {
+        let path = store.path_for(&digest);
+        std::fs::create_dir_all(path.parent().unwrap().parent().unwrap()).unwrap();
+        std::fs::write(path.parent().unwrap(), b"not a directory").unwrap();
+    }
+    let configured = toml::Table::from_iter([
+        ("mode".to_owned(), toml::Value::String("selected".to_owned())),
+        (
+            "packages".to_owned(),
+            toml::Value::Array(vec![toml::Value::String("demo".to_owned())]),
+        ),
+    ]);
+    let mut output = Vec::new();
+
+    let error = crate::PypiServing
+        .mirror(
+            fixture.state,
+            MirrorRequest {
+                action: MirrorAction::Sync,
+                index: "pypi",
+                settings: &toml::Table::new(),
+                configured: &configured,
+                overrides: &toml::Table::new(),
+            },
+            &mut output,
+        )
+        .await
+        .unwrap_err();
+
+    assert_eq!(error, "prefetch sync found 2 failure(s)");
+    let output = String::from_utf8(output).unwrap();
+    assert_eq!(
+        output
+            .lines()
+            .filter_map(|line| {
+                let cells = line.split('\t').collect::<Vec<_>>();
+                (cells.get(7) == Some(&"failure"))
+                    .then(|| (cells[0], cells[3], cells[6], cells[8].starts_with("blob store error:")))
+            })
+            .collect::<Vec<_>>(),
+        [
+            ("metadata", "demo-1.0-py3-none-any.whl.metadata", "", true),
+            ("file", "demo-1.0-py3-none-any.whl", "8", true),
+        ]
+    );
+    assert!(output.contains("files_downloaded\t\t\t0"));
+    assert!(output.contains("bytes_downloaded\t\t\t0"));
+    assert!(output.contains("failures\t\t\t2"));
     server.verify().await;
 }
 
@@ -494,7 +565,15 @@ async fn sync_files_reports_cached_metadata_only_and_filtered_files() {
     )
     .await
     .unwrap();
-    assert!(String::from_utf8(output).unwrap().contains("cached"));
+    let output = String::from_utf8(output).unwrap();
+    assert_eq!(
+        reported_sizes(&output, "cached"),
+        [
+            ("metadata", "demo-1.0-py3-none-any.whl.metadata", "8"),
+            ("file", "demo-1.0-py3-none-any.whl", "8"),
+            ("file", "demo-1.0.zip", "5"),
+        ]
+    );
     assert_eq!(summary.skipped, 1);
 
     selection.filters.metadata_only = true;
@@ -793,3 +872,13 @@ async fn mirror_driver_reports_a_valid_empty_selection_override() {
     );
     assert!(output.is_empty());
 }
+
+fn reported_sizes<'output>(output: &'output str, status: &str) -> Vec<(&'output str, &'output str, &'output str)> {
+    output
+        .lines()
+        .filter_map(|line| {
+            let cells = line.split('\t').collect::<Vec<_>>();
+            (cells.get(7) == Some(&status)).then(|| (cells[0], cells[3], cells[6]))
+        })
+        .collect()
+}