Prevent FallbackRepository from download storms

Use a moka Cache to funnel multiple requests for the same missing
payload into a single task to sync the payload locally. Any failure to
sync the payload will be cached for 5 minutes, after which another
request for that digest will attempt to sync it again.

This was inspired by observing that syncing a single spfs tag would
report having "repaired" several different digests multiple times, often
over a dozen times. It suggests that Syncer may be unexpectedly trying to
process the same digest multiple times. Each of these digests were for
large payloads (500MB+). The sync was happening over a high-latency link.
Without this change, it can end up performing excessive duplicate copies
over a limited bandwidth link.

Signed-off-by: J Robert Ray <[email protected]>

Author: jrray

Cargo.lock

@@ -76,6 +76,7 @@ is_default_derive_macro = { path = "crates/is_default_derive_macro" }
 itertools = "0.14"
 libc = "0.2.172"
 miette = "7.0"
+moka = "0.12.11"
 nix = { version = "0.29", features = ["mount", "sched", "user"] }
 nom = "7.1"
 nom-supreme = "0.8"

Cargo.toml

@@ -58,6 +58,7 @@ libc = { workspace = true }
 linux-raw-sys = "0.8.0"
 linux-syscall = "1.0.0"
 miette = { workspace = true }
+moka = { workspace = true, features = ["future"] }
 nix = { workspace = true, features = ["fs"] }
 nonempty = "0.8.1"
 num_cpus = "1.13.1"

crates/spfs/Cargo.toml

@@ -8,6 +8,7 @@ use std::sync::Arc;
 
 use chrono::{DateTime, Utc};
 use futures::Stream;
+use moka::future::Cache;
 use relative_path::RelativePath;
 
 use crate::config::{ToAddress, default_fallback_repo_include_secondary_tags};
@@ -85,6 +86,7 @@ pub struct FallbackProxy {
     primary: Arc<OpenFsRepository>,
     secondary: Vec<crate::storage::RepositoryHandle>,
     include_secondary_tags: bool,
+    open_payload_cache: Cache<encoding::Digest, ()>,
 }
 
 impl FallbackProxy {
@@ -103,6 +105,12 @@ impl FallbackProxy {
             primary: primary.into(),
             secondary,
             include_secondary_tags,
+            open_payload_cache: Cache::builder()
+                // The TTL thought here is that failed attempts will only get
+                // cached for so long and then reattempted on a subsequent
+                // open_payload call, if any.
+                .time_to_live(std::time::Duration::from_secs(300))
+                .build(),
         }
     }
 }
@@ -162,11 +170,11 @@ impl storage::FromConfig for FallbackProxy {
             Ok(secondary)
         };
         let (primary, secondary) = tokio::try_join!(primary, secondary)?;
-        Ok(Self {
-            primary: primary.into(),
+        Ok(Self::new(
+            Arc::new(primary),
             secondary,
-            include_secondary_tags: config.include_secondary_tags,
-        })
+            config.include_secondary_tags,
+        ))
     }
 }
 
@@ -302,62 +310,65 @@ impl PayloadStorage for FallbackProxy {
         &self,
         digest: encoding::Digest,
     ) -> PayloadResult<(Pin<Box<dyn BlobRead>>, std::path::PathBuf)> {
-        let mut fallbacks = self.secondary.iter();
-
-        'retry_open: loop {
-            let missing_payload_error = match self.primary.open_payload(digest).await {
-                Ok(r) => return Ok(r),
-                Err(err @ PayloadError::UnknownPayload(_)) => err,
-                Err(err) => return Err(err),
-            };
+        if let ok @ Ok(_) = self.primary.open_payload(digest).await {
+            return ok;
+        }
 
-            let mut repair_failure = None;
-
-            let dest_repo = self.primary.clone().into();
-            for fallback in fallbacks.by_ref() {
-                let syncer = crate::Syncer::new(fallback, &dest_repo)
-                    .with_policy(crate::sync::SyncPolicy::ResyncEverything)
-                    .with_reporter(
-                        // There may already be a progress bar in use in this
-                        // context, so don't make another one here.
-                        SyncReporters::silent(),
-                    );
-                match syncer.sync_payload(digest).await {
-                    Ok(_) => {
-                        // Warn for non-sentry users; info for sentry users.
-                        #[cfg(not(feature = "sentry"))]
-                        {
-                            tracing::warn!("Repaired a missing payload! {digest}",);
+        // If a payload is not available in the primary, we want only one task
+        // attempting to download it from the secondary repositories. Other
+        // concurrent attempts to open the same payload should wait for
+        // that attempt to complete, and then try to open the payload again.
+        let mut fallbacks = self.secondary.iter();
+        let dest_repo = self.primary.clone().into();
+        self.open_payload_cache
+            .try_get_with(digest, async move {
+                let mut repair_failure = None;
+                for fallback in fallbacks.by_ref() {
+                    let syncer = crate::Syncer::new(fallback, &dest_repo)
+                        .with_policy(crate::sync::SyncPolicy::ResyncEverything)
+                        .with_reporter(
+                            // There may already be a progress bar in use in this
+                            // context, so don't make another one here.
+                            SyncReporters::silent(),
+                        );
+                    match syncer.sync_payload(digest).await {
+                        Ok(_) => {
+                            // Warn for non-sentry users; info for sentry users.
+                            #[cfg(not(feature = "sentry"))]
+                            {
+                                tracing::warn!("Repaired a missing payload! {digest}",);
+                            }
+                            #[cfg(feature = "sentry")]
+                            {
+                                tracing::info!("Repaired a missing payload! {digest}",);
+                                tracing::error!(target: "sentry", object = %digest, "Repaired a missing payload!");
+                            }
+                            return Ok(());
                         }
-                        #[cfg(feature = "sentry")]
-                        {
-                            tracing::info!("Repaired a missing payload! {digest}",);
-                            tracing::error!(target: "sentry", object = %digest, "Repaired a missing payload!");
+                        Err(err) => {
+                            #[cfg(feature = "sentry")]
+                            tracing::error!(
+                                target: "sentry",
+                                object = %digest,
+                                ?err,
+                                "Could not repair a missing payload"
+                            );
+
+                            repair_failure = Some(PayloadError::String(format!("failed to repair payload: {err}")));
                         }
-                        continue 'retry_open;
-                    }
-                    Err(err) => {
-                        #[cfg(feature = "sentry")]
-                        tracing::error!(
-                            target: "sentry",
-                            object = %digest,
-                            ?err,
-                            "Could not repair a missing payload"
-                        );
-
-                        repair_failure = Some(err);
                     }
                 }
-            }
-
-            if let Some(err) = repair_failure {
-                // The different fallbacks may fail for different reasons,
-                // we just show the most recent failure here.
-                tracing::warn!("Could not repair a missing payload: {err}");
-            }
-
-            return Err(missing_payload_error);
-        }
+                if let Some(err) = repair_failure {
+                    return Err(err);
+                }
+                // Probably can only get here if there were no secondary repos.
+                Err(PayloadError::String("no repositories could successfully read the payload".into()))
+            })
+            .await.map_err(|err| (*err).clone())?;
+
+        // Then each caller needs to try to open the payload again, to get their
+        // own handle to it.
+        self.primary.open_payload(digest).await
     }
 
     async fn remove_payload(&self, digest: encoding::Digest) -> PayloadResult<()> {

crates/spfs/src/storage/fallback/repository.rs

@@ -426,6 +426,7 @@
     "mksource",
     "mksrc",
     "modversions",
+    "moka",
     "mountpoint",
     "mpfr",
     "mrtm",