DedupCache: Reset head if delta too high

Author: RageKnify

crates/generic_log_worker/src/lib.rs

@@ -17,7 +17,7 @@ pub use log_ops::upload_issuers;
 pub use sequencer_do::*;
 
 use byteorder::{BigEndian, WriteBytesExt};
-use log::{info, Level};
+use log::{error, info, Level};
 use log_ops::UploadOptions;
 use metrics::{millis_diff_as_secs, AsF64, ObjectMetrics};
 use serde::{Deserialize, Serialize};
@@ -259,7 +259,7 @@ impl DedupCache {
     // Load batches of cache entries from DO storage into the in-memory cache.
     async fn load(&self, name: &str) -> Result<()> {
         info!("{name} DedupCache: Loading head");
-        let head = self
+        let mut head = self
             .storage
             .get::<usize>(Self::FIFO_HEAD_KEY)
             .await
@@ -271,6 +271,11 @@ impl DedupCache {
             .await
             .unwrap_or_default();
         info!("{name} DedupCache: head({head}) and tail({tail}) loaded");
+        if tail - head > Self::MAX_BATCHES {
+            head = tail - Self::MAX_BATCHES;
+            error!("{name} DedupCache: delta too high, setting head to tail - MAX_BATCHES ({head})");
+            self.storage.put(Self::FIFO_HEAD_KEY, head).await?;
+        }
         let keys = (0..(tail - head)).map(Self::fifo_key).collect::<Vec<_>>();
         info!("{name} DedupCache: Getting keys");
         let map = self.storage.get_multiple(keys).await?;