v0.31.7

src/cache/cold/demotion.rs

@@ -197,8 +197,9 @@ impl Cache {
                     _memory_lease: memory_lease,
                 }
             };
-            self.persist_cold_inner(draft).await;
-            moved += 1;
+            if self.persist_cold_inner(draft).await {
+                moved += 1;
+            }
         }
         release_allocator_memory();
         moved

src/cache/cold/worker.rs

@@ -62,7 +62,7 @@ impl Cache {
         // reference and glibc is explicitly asked to return releasable pages.
         release_allocator_memory();
     }
-    async fn persist_cold_inner(&self, draft: ColdDraft) {
+    async fn persist_cold_inner(&self, draft: ColdDraft) -> bool {
         let current = {
             let store = self.store.lock().unwrap_or_else(|e| e.into_inner());
             store.revision == draft.revision
@@ -84,11 +84,11 @@ impl Cache {
                     == draft.key_generation
         };
         if !current {
-            return;
+            return false;
         }
         let Some(headers) = cold_header_pairs(&draft.entry.headers) else {
             self.counters.cold_errors.fetch_add(1, Ordering::Relaxed);
-            return;
+            return false;
         };
         let settings = self
             .tier_settings
@@ -96,17 +96,17 @@ impl Cache {
             .unwrap_or_else(|e| e.into_inner())
             .clone();
         if !settings.cold_enabled || self.cold_root_for(&settings).as_ref() != Some(&draft.root) {
-            return;
+            return false;
         }
         let key = key_hex(&draft.key);
         let body_name = format!("{key}.body");
         let (body_path, meta_path, body_tmp, meta_tmp) = cold_paths(&draft.root, &draft.key);
         let Some(shard) = body_path.parent() else {
-            return;
+            return false;
         };
         if tokio::fs::create_dir_all(shard).await.is_err() {
             self.counters.cold_errors.fetch_add(1, Ordering::Relaxed);
-            return;
+            return false;
         }
         let age_at_store = draft
             .entry
@@ -167,7 +167,7 @@ impl Cache {
         };
         let Ok(metadata) = serde_json::to_vec_pretty(&file) else {
             self.counters.cold_errors.fetch_add(1, Ordering::Relaxed);
-            return;
+            return false;
         };
         let write_ok = tokio::fs::write(&body_tmp, &draft.entry.body).await.is_ok()
             && tokio::fs::write(&meta_tmp, &metadata).await.is_ok()
@@ -177,7 +177,7 @@ impl Cache {
             let _ = tokio::fs::remove_file(&body_tmp).await;
             let _ = tokio::fs::remove_file(&meta_tmp).await;
             self.counters.cold_errors.fetch_add(1, Ordering::Relaxed);
-            return;
+            return false;
         }
         let still_current = {
             let store = self.store.lock().unwrap_or_else(|e| e.into_inner());
@@ -202,10 +202,11 @@ impl Cache {
         if !still_current {
             let _ = tokio::fs::remove_file(&body_path).await;
             let _ = tokio::fs::remove_file(&meta_path).await;
-            return;
+            return false;
         }
         let disk_bytes = draft.entry.body.len() as u64 + metadata.len() as u64;
         let mut delete = Vec::new();
+        let mut persisted = false;
         {
             let mut cold = self.cold.lock().unwrap_or_else(|e| e.into_inner());
             for key in incompatible_cold_variants(&cold, &draft.entry.base_key, &draft.entry.vary) {
@@ -278,11 +279,13 @@ impl Cache {
                 cold.lru.insert(sequence, draft.key);
                 cold.index_variant(entry.base_key, draft.key);
                 cold.entries.insert(draft.key, entry);
+                persisted = true;
                 self.counters.hot_demotions.fetch_add(1, Ordering::Relaxed);
             }
         }
         for path in delete {
             let _ = tokio::fs::remove_file(path).await;
         }
+        persisted
     }
 }