fix: Block ordered cache

crates/store/src/state/apply_block.rs

@@ -181,7 +181,9 @@ impl State {
         })?;
 
         // Push to cache and notify replica subscribers.
-        self.block_cache.push(block_num, BlockNotification::new(block_num, cache_bytes));
+        self.block_cache
+            .push(block_num, BlockNotification::new(block_num, cache_bytes))
+            .expect("block cache receives sequential block numbers");
         let _ = self.committed_tip_tx.send(block_num);
 
         info!(%block_commitment, block_num = block_num.as_u32(), COMPONENT, "apply_block successful");

crates/store/src/state/apply_proof.rs

@@ -13,7 +13,9 @@ impl State {
         proof_bytes: Vec<u8>,
     ) -> anyhow::Result<()> {
         self.block_store.commit_proof(block_num, &proof_bytes).await?;
-        self.proof_cache.push(block_num, ProofNotification::new(block_num, proof_bytes));
+        self.proof_cache
+            .push(block_num, ProofNotification::new(block_num, proof_bytes))
+            .expect("proof cache receives sequential block numbers");
         self.proven_tip.advance(block_num);
         Ok(())
     }

crates/store/src/state/replica.rs

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use miden_node_utils::fifo_cache::FifoCache;
+use miden_node_utils::block_cache::BlockOrderedCache;
 use miden_protocol::block::BlockNumber;
 
 // BLOCK NOTIFICATION
@@ -61,7 +61,7 @@ struct Proof {
 // ================================================================================================
 
 /// FIFO cache of recent committed blocks for replica subscriptions.
-pub type BlockCache = FifoCache<BlockNumber, BlockNotification>;
+pub type BlockCache = BlockOrderedCache<BlockNotification>;
 
 /// FIFO cache of recent block proofs for replica subscriptions.
-pub type ProofCache = FifoCache<BlockNumber, ProofNotification>;
+pub type ProofCache = BlockOrderedCache<ProofNotification>;

crates/store/src/state/subscription.rs

@@ -177,7 +177,7 @@ async fn fetch_block(
     cache: &BlockCache,
     state: &State,
 ) -> Result<Vec<u8>, StateSubscriptionError> {
-    if let Some(entry) = cache.get(&block_num) {
+    if let Some(entry) = cache.get(block_num) {
         return Ok(entry.block_bytes().to_vec());
     }
     state
@@ -192,7 +192,7 @@ async fn fetch_proof(
     cache: &ProofCache,
     state: &State,
 ) -> Result<Vec<u8>, StateSubscriptionError> {
-    if let Some(entry) = cache.get(&block_num) {
+    if let Some(entry) = cache.get(block_num) {
         return Ok(entry.proof_bytes().to_vec());
     }
     state

crates/utils/src/block_cache.rs

@@ -0,0 +1,134 @@
+use std::collections::VecDeque;
+use std::num::NonZeroUsize;
+use std::sync::{Arc, RwLock};
+
+use miden_protocol::block::BlockNumber;
+
+/// A cheaply cloneable block-ordered cache.
+#[derive(Clone)]
+pub struct BlockOrderedCache<T> {
+    inner: Arc<RwLock<Inner<T>>>,
+}
+
+struct Inner<T> {
+    fifo: VecDeque<T>,
+    youngest: Option<BlockNumber>,
+    capacity: usize,
+}
+
+impl<T> BlockOrderedCache<T> {
+    /// Creates a new cache with the given capacity.
+    pub fn new(capacity: NonZeroUsize) -> Self {
+        Self {
+            inner: Arc::new(RwLock::new(Inner {
+                fifo: VecDeque::new(),
+                youngest: None,
+                capacity: capacity.get(),
+            })),
+        }
+    }
+
+    /// Pushes a new value into the cache and evicts the oldest value if the cache is full.
+    ///
+    /// # Error
+    ///
+    /// Returns the value if the provided block number is not the next in sequence.
+    pub fn push(&self, number: BlockNumber, value: T) -> Result<(), T> {
+        let mut inner = self.inner.write().expect("block cache lock poisoned");
+
+        if let Some(youngest) = inner.youngest {
+            if youngest.child() != number {
+                return Err(value);
+            }
+        }
+
+        if inner.fifo.len() >= inner.capacity {
+            inner.fifo.pop_front();
+        }
+
+        inner.fifo.push_back(value);
+        inner.youngest = Some(number);
+
+        Ok(())
+    }
+}
+
+impl<T: Clone> BlockOrderedCache<T> {
+    /// Retrieves the value associated with the given block number from the cache.
+    pub fn get(&self, number: BlockNumber) -> Option<T> {
+        let inner = self.inner.read().expect("block cache lock poisoned");
+        let youngest = inner.youngest?;
+        let distance_to_oldest = u32::try_from(inner.fifo.len().checked_sub(1)?).ok()?;
+        let oldest = youngest.checked_sub(distance_to_oldest)?;
+
+        let offset = number.checked_sub(oldest.as_u32())?;
+        inner.fifo.get(offset.as_usize()).cloned()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::num::NonZeroUsize;
+
+    use super::BlockOrderedCache;
+
+    fn cache(cap: usize) -> BlockOrderedCache<&'static str> {
+        BlockOrderedCache::new(NonZeroUsize::new(cap).unwrap())
+    }
+
+    #[test]
+    fn get_returns_none_on_empty_cache() {
+        let c = cache(4);
+        assert_eq!(c.get(1.into()), None);
+    }
+
+    #[test]
+    fn get_returns_inserted_value() {
+        let c = cache(4);
+        assert_eq!(c.push(1.into(), "a"), Ok(()));
+        assert_eq!(c.get(1.into()), Some("a"));
+    }
+
+    #[test]
+    fn evicts_oldest_entry_when_full() {
+        let c = cache(2);
+        assert_eq!(c.push(5.into(), "a"), Ok(()));
+        assert_eq!(c.push(6.into(), "b"), Ok(()));
+        assert_eq!(c.push(7.into(), "c"), Ok(())); // evicts 5
+        assert_eq!(c.get(5.into()), None);
+        assert_eq!(c.get(6.into()), Some("b"));
+        assert_eq!(c.get(7.into()), Some("c"));
+    }
+
+    #[test]
+    fn overwrite_key_returns_value() {
+        let c = cache(2);
+        assert_eq!(c.push(1.into(), "a"), Ok(()));
+        assert_eq!(c.push(1.into(), "b"), Err("b"));
+        assert_eq!(c.get(1.into()), Some("a"));
+    }
+
+    #[test]
+    fn parent_returns_value() {
+        let c = cache(2);
+        assert_eq!(c.push(3.into(), "a"), Ok(()));
+        assert_eq!(c.push(2.into(), "b"), Err("b"));
+        assert_eq!(c.get(3.into()), Some("a"));
+    }
+
+    #[test]
+    fn wrong_child_returns_value() {
+        let c = cache(2);
+        assert_eq!(c.push(1.into(), "a"), Ok(()));
+        assert_eq!(c.push(3.into(), "b"), Err("b"));
+        assert_eq!(c.get(1.into()), Some("a"));
+    }
+
+    #[test]
+    fn clone_shares_state() {
+        let c1 = cache(4);
+        let c2 = c1.clone();
+        assert_eq!(c1.push(1.into(), "a"), Ok(()));
+        assert_eq!(c2.get(1.into()), Some("a"));
+    }
+}

crates/utils/src/lib.rs

@@ -1,9 +1,9 @@
+pub mod block_cache;
 pub mod clap;
 pub mod cors;
 pub mod crypto;
 #[cfg(feature = "testing")]
 pub mod fee;
-pub mod fifo_cache;
 pub mod formatting;
 pub mod fs;
 pub mod genesis;