prevent read-blocking during contiguous fixed journal prune

Author: roberto-bayardo

Cargo.lock

@@ -705,6 +705,8 @@ stability_scope!(BETA {
         /// Remove a blob from a given partition.
         ///
         /// If no `name` is provided, the entire partition is removed.
+        /// If a `name` is provided, existing [Blob] handles for that blob must remain readable
+        /// until they are dropped, but the blob must be removed from future namespace lookups.
         ///
         /// An Ok result indicates the blob is durably removed.
         fn remove(

runtime/src/lib.rs

@@ -38,6 +38,7 @@ commonware-math.workspace = true
 commonware-runtime = { workspace = true, features = ["test-utils"] }
 commonware-storage = { path = ".", features = ["std"] }
 criterion.workspace = true
+governor.workspace = true
 rand.workspace = true
 rstest.workspace = true
 tracing-subscriber.workspace = true

storage/Cargo.toml

@@ -935,7 +935,7 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
     /// Note that this operation may NOT be atomic, however it's guaranteed not to leave gaps in the
     /// event of failure as items are always pruned in order from oldest to newest.
     pub async fn prune(&self, min_item_pos: u64) -> Result<bool, Error> {
-        let mut inner = self.inner.write().await;
+        let inner = self.inner.write().await;
 
         // Calculate the section that would contain min_item_pos
         let target_section = min_item_pos / self.items_per_blob;
@@ -946,7 +946,15 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
         // Cap to tail section. The tail section is guaranteed to exist by our invariant.
         let min_section = std::cmp::min(target_section, tail_section);
 
-        let pruned = inner.journal.prune(min_section).await?;
+        // Unlink old sections while allowing concurrent readers to use existing open handles.
+        let inner_ref = inner.downgrade_to_upgradable();
+        let pruned = inner_ref.journal.unlink_before(min_section).await?;
+        if !pruned {
+            return Ok(false);
+        }
+
+        let mut inner = inner_ref.upgrade().await;
+        let pruned = inner.journal.commit_prune(min_section);
 
         // After pruning, update pruning_boundary to the start of the oldest remaining section
         if pruned {
@@ -1120,14 +1128,19 @@ impl<E: Context, A: CodecFixedShared> crate::journal::authenticated::Inner<E> fo
 mod tests {
     use super::*;
     use commonware_cryptography::{sha256::Digest, Hasher as _, Sha256};
-    use commonware_macros::test_traced;
+    use commonware_macros::{select, test_traced};
     use commonware_runtime::{
         deterministic::{self, Context},
-        Blob, BufferPooler, Error as RuntimeError, Metrics as _, Runner, Storage, Supervisor as _,
+        Blob, BufferPooler, Clock as _, Error as RuntimeError, Metrics as _, Runner, Spawner as _,
+        Storage, Supervisor as _,
     };
-    use commonware_utils::{NZUsize, NZU16, NZU64};
+    use commonware_utils::{sync::Notify, NZUsize, NZU16, NZU64};
     use futures::{pin_mut, StreamExt};
-    use std::num::NonZeroU16;
+    use std::{
+        num::NonZeroU16,
+        sync::Arc,
+        time::{Duration, SystemTime},
+    };
 
     const PAGE_SIZE: NonZeroU16 = NZU16!(44);
     const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(3);
@@ -1158,6 +1171,141 @@ mod tests {
         }
     }
 
+    struct RemoveBlocker {
+        target: Vec<u8>,
+        removed: Notify,
+        release: Notify,
+    }
+
+    impl RemoveBlocker {
+        /// Create a blocker for removal of the given section.
+        fn new(section: u64) -> Self {
+            Self {
+                target: section.to_be_bytes().to_vec(),
+                removed: Notify::new(),
+                release: Notify::new(),
+            }
+        }
+    }
+
+    struct BlockingContext {
+        inner: Context,
+        blocker: Arc<RemoveBlocker>,
+    }
+
+    impl BlockingContext {
+        /// Wrap a deterministic context and pause removal of the blocker's target section.
+        fn new(inner: Context, blocker: Arc<RemoveBlocker>) -> Self {
+            Self { inner, blocker }
+        }
+    }
+
+    impl commonware_runtime::Supervisor for BlockingContext {
+        fn name(&self) -> commonware_runtime::Name {
+            self.inner.name()
+        }
+
+        fn child(&self, label: &'static str) -> Self {
+            Self {
+                inner: self.inner.child(label),
+                blocker: self.blocker.clone(),
+            }
+        }
+
+        fn with_attribute(self, key: &'static str, value: impl std::fmt::Display) -> Self {
+            Self {
+                inner: self.inner.with_attribute(key, value),
+                blocker: self.blocker,
+            }
+        }
+    }
+
+    impl commonware_runtime::Metrics for BlockingContext {
+        fn register<
+            N: Into<String>,
+            H: Into<String>,
+            M: commonware_runtime::telemetry::metrics::Metric,
+        >(
+            &self,
+            name: N,
+            help: H,
+            metric: M,
+        ) -> commonware_runtime::telemetry::metrics::Registered<M> {
+            self.inner.register(name, help, metric)
+        }
+
+        fn encode(&self) -> String {
+            self.inner.encode()
+        }
+    }
+
+    impl governor::clock::Clock for BlockingContext {
+        type Instant = SystemTime;
+
+        fn now(&self) -> Self::Instant {
+            self.inner.current()
+        }
+    }
+
+    impl governor::clock::ReasonablyRealtime for BlockingContext {}
+
+    impl commonware_runtime::Clock for BlockingContext {
+        fn current(&self) -> SystemTime {
+            self.inner.current()
+        }
+
+        fn sleep(
+            &self,
+            duration: Duration,
+        ) -> impl std::future::Future<Output = ()> + Send + 'static {
+            self.inner.sleep(duration)
+        }
+
+        fn sleep_until(
+            &self,
+            deadline: SystemTime,
+        ) -> impl std::future::Future<Output = ()> + Send + 'static {
+            self.inner.sleep_until(deadline)
+        }
+    }
+
+    impl BufferPooler for BlockingContext {
+        fn network_buffer_pool(&self) -> &commonware_runtime::BufferPool {
+            self.inner.network_buffer_pool()
+        }
+
+        fn storage_buffer_pool(&self) -> &commonware_runtime::BufferPool {
+            self.inner.storage_buffer_pool()
+        }
+    }
+
+    impl Storage for BlockingContext {
+        type Blob = <Context as Storage>::Blob;
+
+        async fn open_versioned(
+            &self,
+            partition: &str,
+            name: &[u8],
+            versions: std::ops::RangeInclusive<u16>,
+        ) -> Result<(Self::Blob, u64, u16), RuntimeError> {
+            self.inner.open_versioned(partition, name, versions).await
+        }
+
+        async fn remove(&self, partition: &str, name: Option<&[u8]>) -> Result<(), RuntimeError> {
+            let block = name.is_some_and(|name| name == self.blocker.target.as_slice());
+            let result = self.inner.remove(partition, name).await;
+            if block {
+                self.blocker.removed.notify_one();
+                self.blocker.release.notified().await;
+            }
+            result
+        }
+
+        async fn scan(&self, partition: &str) -> Result<Vec<Vec<u8>>, RuntimeError> {
+            self.inner.scan(partition).await
+        }
+    }
+
     #[test_traced]
     fn test_fixed_journal_init_conflicting_partitions() {
         let executor = deterministic::Runner::default();
@@ -1400,6 +1548,91 @@ mod tests {
         });
     }
 
+    #[test_traced]
+    fn test_fixed_journal_reads_during_prune_unlink() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let blocker = Arc::new(RemoveBlocker::new(0));
+            let journal_context = BlockingContext::new(context.child("journal"), blocker.clone());
+            let cfg = test_cfg(&journal_context, NZU64!(2));
+            let journal = Arc::new(
+                Journal::<_, Digest>::init(journal_context.child("inner"), cfg)
+                    .await
+                    .expect("failed to initialize journal"),
+            );
+
+            for i in 0..6 {
+                let pos = journal.append(&test_digest(i)).await.unwrap();
+                assert_eq!(pos, i);
+            }
+            journal.sync().await.unwrap();
+
+            let prune = context.child("prune").spawn({
+                let journal = journal.clone();
+                |_| async move { journal.prune(4).await }
+            });
+
+            select! {
+                _ = blocker.removed.notified() => {},
+                _ = context.sleep(Duration::from_secs(1)) => panic!("prune did not unlink section"),
+            }
+
+            let reader = journal.reader();
+            pin_mut!(reader);
+            let reader = select! {
+                reader = reader => reader,
+                _ = context.sleep(Duration::from_secs(1)) => {
+                    panic!("reader blocked while prune was unlinking")
+                },
+            };
+
+            assert_eq!(reader.read(0).await.unwrap(), test_digest(0));
+            assert_eq!(reader.read(4).await.unwrap(), test_digest(4));
+            drop(reader);
+
+            blocker.release.notify_one();
+            assert!(prune.await.unwrap().unwrap());
+
+            assert_eq!(journal.pruning_boundary().await, 4);
+            assert_eq!(journal.test_oldest_section().await, Some(2));
+            assert!(matches!(journal.read(0).await, Err(Error::ItemPruned(0))));
+            assert_eq!(journal.read(4).await.unwrap(), test_digest(4));
+        });
+    }
+
+    #[test_traced]
+    fn test_fixed_journal_prune_remove_failure_reopens_contiguous() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let cfg = test_cfg(&context, NZU64!(2));
+            let journal = Journal::<_, Digest>::init(context.child("first"), cfg.clone())
+                .await
+                .expect("failed to initialize journal");
+
+            for i in 0..6 {
+                let pos = journal.append(&test_digest(i)).await.unwrap();
+                assert_eq!(pos, i);
+            }
+            journal.sync().await.unwrap();
+
+            let fault_cfg = context.storage_fault_config();
+            *fault_cfg.write() = deterministic::FaultConfig::default().remove(1.0);
+            let err = journal.prune(4).await.expect_err("prune should fail");
+            assert!(matches!(err, Error::Runtime(RuntimeError::Io(_))));
+            drop(journal);
+
+            *fault_cfg.write() = deterministic::FaultConfig::default();
+            let journal = Journal::<_, Digest>::init(context.child("second"), cfg)
+                .await
+                .expect("failed to re-initialize journal");
+            assert_eq!(journal.bounds().await, 0..6);
+            assert_eq!(journal.test_oldest_section().await, Some(0));
+            for i in 0..6 {
+                assert_eq!(journal.read(i).await.unwrap(), test_digest(i));
+            }
+        });
+    }
+
     /// Append a lot of data to make sure we exercise page cache paging boundaries.
     #[test_traced]
     fn test_fixed_journal_append_a_lot_of_data() {

storage/src/journal/contiguous/fixed.rs

@@ -381,6 +381,21 @@ impl<E: Storage + Metrics, A: CodecFixedShared> Journal<E, A> {
         self.manager.prune(min).await
     }
 
+    /// Remove section blobs less than `min` from storage without dropping open handles.
+    ///
+    /// This allows fixed contiguous journals to perform physical removal while readers keep using
+    /// existing section handles.
+    pub(in crate::journal) async fn unlink_before(&self, min: u64) -> Result<bool, Error> {
+        self.manager.unlink_before(min).await
+    }
+
+    /// Commit a successful unlink by removing sections less than `min` from memory.
+    ///
+    /// This must only be called after [Self::unlink_before] succeeds for the same `min`.
+    pub(in crate::journal) fn commit_prune(&mut self, min: u64) -> bool {
+        self.manager.commit_prune(min)
+    }
+
     /// Returns the oldest section number, if any blobs exist.
     pub fn oldest_section(&self) -> Option<u64> {
         self.manager.oldest_section()