Check if variant exists when serving existing and warn if it doesn't

src/lib.rs

@@ -842,6 +842,56 @@ async fn not_found_hash(repo: &ArcRepo) -> Result<Option<(Alias, Hash)>, Error>
     Ok(Some((alias, hash)))
 }
 
+async fn exists<S: Store>(store: &S, identifier: &Arc<str>) -> Result<bool, Error> {
+    if let Err(e) = store.len(identifier).await {
+        if e.is_not_found() {
+            return Ok(false);
+        }
+
+        return Err(e.into());
+    }
+
+    Ok(true)
+}
+
+async fn existing_variant_identifier<S: Store>(
+    state: &State<S>,
+    hash: Hash,
+    variant: String,
+) -> Result<Option<Arc<str>>, Error> {
+    let identifier_opt = state
+        .repo
+        .variant_identifier(hash.clone(), variant.clone())
+        .await?;
+
+    if let Some(identifier) = identifier_opt {
+        if !exists(&state.store, &identifier).await? {
+            let clean =
+                if let Some(original_identifier) = state.repo.identifier(hash.clone()).await? {
+                    !exists(&state.store, &original_identifier).await?
+                } else {
+                    true
+                };
+
+            if clean {
+                if state.config.server.read_only {
+                    tracing::warn!("Stored variant {variant} for hash {hash:?} doesn't exist");
+                    return Err(UploadError::ReadOnly.into());
+                }
+
+                tracing::warn!("Stored variant {variant} for hash {hash:?} doesn't exist, spawning cleanup job");
+                queue::cleanup_variants(&state.repo, hash, Some(variant)).await?;
+            }
+
+            Ok(None)
+        } else {
+            Ok(Some(identifier))
+        }
+    } else {
+        Ok(None)
+    }
+}
+
 /// Process files
 #[tracing::instrument(name = "Serving processed image", skip(state))]
 async fn process<S: Store + 'static>(
@@ -871,10 +921,7 @@ async fn process<S: Store + 'static>(
             .await?;
     }
 
-    let identifier_opt = state
-        .repo
-        .variant_identifier(hash.clone(), variant.clone())
-        .await?;
+    let identifier_opt = existing_variant_identifier(&state, hash.clone(), variant.clone()).await?;
 
     let (details, identifier) = if let Some(identifier) = identifier_opt {
         let details = ensure_details_identifier(&state, &identifier).await?;

src/queue.rs

@@ -97,7 +97,7 @@ pub(crate) async fn cleanup_identifier(repo: &ArcRepo, identifier: &Arc<str>) ->
     Ok(())
 }
 
-async fn cleanup_variants(
+pub(super) async fn cleanup_variants(
     repo: &ArcRepo,
     hash: Hash,
     variant: Option<String>,