Add unverified checkpoint marker

rs/replicated_state/src/page_map/storage.rs

@@ -181,7 +181,7 @@ impl Storage {
         let mut base_overlays = Vec::<OverlayFile>::new();
         let mut overlays = Vec::<OverlayFile>::new();
         for path in overlay_paths.iter() {
-            let overlay = OverlayFile::load(path).unwrap();
+            let overlay = OverlayFile::load(path)?;
             let start_page_index = overlay
                 .index_iter()
                 .next()

rs/state_layout/src/error.rs

@@ -14,7 +14,10 @@ pub enum LayoutError {
 
     /// The state root doesn't have the expected layout.  It's either wrong or
     /// was corrupted.
-    CorruptedLayout { path: PathBuf, message: String },
+    CorruptedLayout {
+        path: PathBuf,
+        message: String,
+    },
 
     /// Checkpoint at the specified height already exists.
     AlreadyExists(Height),
@@ -24,6 +27,8 @@ pub enum LayoutError {
 
     /// Trying to remove the latest checkpoint.
     LatestCheckpoint(Height),
+
+    CheckpointUnverified(Height),
 }
 
 impl fmt::Display for LayoutError {
@@ -57,6 +62,9 @@ impl fmt::Display for LayoutError {
                 "Trying to remove the latest checkpoint at height @{}",
                 height
             ),
+            Self::CheckpointUnverified(height) => {
+                write!(f, "Checkpoint @{} is not verified", height)
+            }
         }
     }
 }

rs/state_layout/src/state_layout.rs

@@ -71,6 +71,8 @@ pub struct RwPolicy<'a, Owner> {
     lifetime_tag: PhantomData<&'a Owner>,
 }
 
+pub enum Verification {}
+
 pub trait AccessPolicy {
     /// `check_dir` specifies what to do the first time we enter a
     /// directory while reading/writing a checkpoint.
@@ -87,11 +89,15 @@ pub trait ReadPolicy: AccessPolicy {}
 pub trait WritePolicy: AccessPolicy {}
 pub trait ReadWritePolicy: ReadPolicy + WritePolicy {}
 
+pub trait LoadingPolicy: ReadPolicy {}
+
 impl<T> ReadWritePolicy for T where T: ReadPolicy + WritePolicy {}
 
 impl AccessPolicy for ReadOnly {}
 impl ReadPolicy for ReadOnly {}
 
+impl LoadingPolicy for ReadOnly {}
+
 impl AccessPolicy for WriteOnly {
     /// For `WriteOnly` mode we want to ensure the directory exists
     /// when we visit it for the first time as we'll certainly want to
@@ -116,6 +122,10 @@ impl<'a, T> AccessPolicy for RwPolicy<'a, T> {
 impl<'a, T> ReadPolicy for RwPolicy<'a, T> {}
 impl<'a, T> WritePolicy for RwPolicy<'a, T> {}
 
+impl AccessPolicy for Verification {}
+impl ReadPolicy for Verification {}
+impl LoadingPolicy for Verification {}
+
 pub type CompleteCheckpointLayout = CheckpointLayout<ReadOnly>;
 
 /// This struct contains bits of the `ExecutionState` that are not already
@@ -371,6 +381,14 @@ impl TipHandler {
             if path.extension() == Some(OsStr::new("pbuf")) {
                 // Do not copy protobufs.
                 CopyInstruction::Skip
+            } else if path == cp.unverified_checkpoint_marker() {
+                // With LSMT disabled, the unverified checkpoint marker is still present in the checkpoint at this point.
+                // We should not copy it back to the tip because it will be created later when promoting the tip as the next checkpoint.
+                // When we go for asynchronous checkpointing in the future, we should revisit this as the marker file will have a different lifespan.
+
+                // With LSMT enabled, the unverified checkpoint marker should already be removed at this point.
+                debug_assert_eq!(lsmt_storage, FlagStatus::Disabled);
+                CopyInstruction::Skip
             } else if path.extension() == Some(OsStr::new("bin"))
                 && lsmt_storage == FlagStatus::Disabled
             {
@@ -495,7 +513,7 @@ impl StateLayout {
         thread_pool: &mut Option<scoped_threadpool::Pool>,
     ) -> Result<(), LayoutError> {
         for height in self.checkpoint_heights()? {
-            let path = self.checkpoint(height)?.raw_path().to_path_buf();
+            let path = self.checkpoint_verified(height)?.raw_path().to_path_buf();
             sync_and_mark_files_readonly(&self.log, &path, &self.metrics, thread_pool.as_mut())
                 .map_err(|err| LayoutError::IoError {
                     path,
@@ -616,7 +634,7 @@ impl StateLayout {
         layout: CheckpointLayout<RwPolicy<'_, T>>,
         height: Height,
         thread_pool: Option<&mut scoped_threadpool::Pool>,
-    ) -> Result<CheckpointLayout<ReadOnly>, LayoutError> {
+    ) -> Result<CheckpointLayout<Verification>, LayoutError> {
         debug_assert_eq!(height, layout.height);
         let scratchpad = layout.raw_path();
         let checkpoints_path = self.checkpoints();
@@ -647,7 +665,7 @@ impl StateLayout {
             message: "Could not sync checkpoints".to_string(),
             io_err: err,
         })?;
-        self.checkpoint(height)
+        self.checkpoint_in_verification(height)
     }
 
     pub fn clone_checkpoint(&self, from: Height, to: Height) -> Result<(), LayoutError> {
@@ -668,9 +686,13 @@ impl StateLayout {
         Ok(())
     }
 
+    // TODO: make this method private.
     /// Returns the layout of the checkpoint with the given height (if
     /// there is one).
-    pub fn checkpoint(&self, height: Height) -> Result<CheckpointLayout<ReadOnly>, LayoutError> {
+    pub fn checkpoint<P>(&self, height: Height) -> Result<CheckpointLayout<P>, LayoutError>
+    where
+        P: LoadingPolicy,
+    {
         let cp_name = Self::checkpoint_name(height);
         let path = self.checkpoints().join(cp_name);
         if !path.exists() {
@@ -699,20 +721,36 @@ impl StateLayout {
                 }
             }
         }
-        CheckpointLayout::new(path, height, self.clone())
+        CheckpointLayout::<P>::new(path, height, self.clone())
     }
 
-    /// Returns the untracked `CheckpointLayout` for the given height (if there is one).
-    pub fn checkpoint_untracked(
+    pub fn checkpoint_verified(
         &self,
         height: Height,
     ) -> Result<CheckpointLayout<ReadOnly>, LayoutError> {
+        let cp = self.checkpoint::<ReadOnly>(height)?;
+        // TODO: move this check to the checkpoint layout constructor.
+        if !cp.is_checkpoint_verified() {
+            return Err(LayoutError::CheckpointUnverified(height));
+        };
+        Ok(cp)
+    }
+
+    pub fn checkpoint_in_verification(
+        &self,
+        height: Height,
+    ) -> Result<CheckpointLayout<Verification>, LayoutError> {
+        self.checkpoint::<Verification>(height)
+    }
+
+    pub fn checkpoint_verification_status(&self, height: Height) -> Result<bool, LayoutError> {
         let cp_name = Self::checkpoint_name(height);
         let path = self.checkpoints().join(cp_name);
         if !path.exists() {
             return Err(LayoutError::NotFound(height));
         }
-        CheckpointLayout::new_untracked(path, height)
+        let cp = CheckpointLayout::<ReadOnly>::new_untracked(path, height)?;
+        Ok(cp.is_checkpoint_verified())
     }
 
     fn increment_checkpoint_ref_counter(&self, height: Height) {
@@ -760,8 +798,19 @@ impl StateLayout {
         }
     }
 
-    /// Returns a sorted list of `Height`s for which a checkpoint is available.
+    /// Returns a sorted list of `Height`s for which a checkpoint is available and verified.
     pub fn checkpoint_heights(&self) -> Result<Vec<Height>, LayoutError> {
+        let checkpoint_heights = self
+            .unfiltered_checkpoint_heights()?
+            .into_iter()
+            .filter(|h| self.checkpoint_verification_status(*h).unwrap_or(false))
+            .collect();
+
+        Ok(checkpoint_heights)
+    }
+
+    /// Returns a sorted list of `Height`s for which a checkpoint is available, regardless of verification status.
+    pub fn unfiltered_checkpoint_heights(&self) -> Result<Vec<Height>, LayoutError> {
         let names = dir_file_names(&self.checkpoints()).map_err(|err| LayoutError::IoError {
             path: self.checkpoints(),
             message: format!("Failed to get all checkpoints (err kind: {:?})", err.kind()),
@@ -883,7 +932,7 @@ impl StateLayout {
                 if heights.is_empty() {
                     error!(
                         self.log,
-                        "Trying to remove non-existing checkpoint {}. The CheckpoinLayout was invalid",
+                        "Trying to remove non-existing checkpoint {}. The CheckpointLayout was invalid",
                         height,
                     );
                     self.metrics
@@ -1416,6 +1465,60 @@ impl<Permissions: AccessPolicy> CheckpointLayout<Permissions> {
     pub fn raw_path(&self) -> &Path {
         &self.root
     }
+
+    /// Returns if the checkpoint is marked as unverified or not.
+    pub fn is_checkpoint_verified(&self) -> bool {
+        !self.unverified_checkpoint_marker().exists()
+    }
+}
+
+impl<P> CheckpointLayout<P>
+where
+    P: WritePolicy,
+{
+    /// Creates the unverified checkpoint marker.
+    /// If the marker already exists, this function does nothing and returns `Ok(())`.
+    pub fn create_unverified_checkpoint_marker(&self) -> Result<(), LayoutError> {
+        let marker = self.unverified_checkpoint_marker();
+        if marker.exists() {
+            return Ok(());
+        }
+        open_for_write(&marker)?;
+        sync_path(&self.root).map_err(|err| LayoutError::IoError {
+            path: self.root.clone(),
+            message: "Failed to sync checkpoint directory for the creation of the unverified checkpoint marker".to_string(),
+            io_err: err,
+        })
+    }
+}
+
+impl CheckpointLayout<Verification> {
+    /// Removes the unverified checkpoint marker.
+    /// If the marker does not exist, this function does nothing and returns `Ok(())`.
+    pub fn remove_unverified_checkpoint_marker(&self) -> Result<(), LayoutError> {
+        let marker = self.unverified_checkpoint_marker();
+        if !marker.exists() {
+            return Ok(());
+        }
+        match std::fs::remove_file(&marker) {
+            Err(err) if err.kind() != std::io::ErrorKind::NotFound => {
+                return Err(LayoutError::IoError {
+                    path: marker.to_path_buf(),
+                    message: "failed to remove file from disk".to_string(),
+                    io_err: err,
+                });
+            }
+            _ => {}
+        }
+
+        // Sync the directory to make sure the marker is removed from disk.
+        // This is strict prerequisite for the manifest computation.
+        sync_path(&self.root).map_err(|err| LayoutError::IoError {
+            path: self.root.clone(),
+            message: "Failed to sync checkpoint directory for the creation of the unverified checkpoint marker".to_string(),
+            io_err: err,
+        })
+    }
 }
 
 pub struct PageMapLayout<Permissions: AccessPolicy> {

rs/state_layout/src/state_layout/tests.rs

@@ -354,6 +354,54 @@ fn test_last_removal_panics_in_debug() {
     });
 }
 
+#[test]
+fn test_can_remove_unverified_marker_file_twice() {
+    with_test_replica_logger(|log| {
+        let tempdir = tmpdir("state_layout");
+        let root_path = tempdir.path().to_path_buf();
+        let metrics_registry = ic_metrics::MetricsRegistry::new();
+        let state_layout = StateLayout::try_new(log, root_path, &metrics_registry).unwrap();
+
+        let height = Height::new(1);
+        let state_sync_scratchpad = state_layout.state_sync_scratchpad(height).unwrap();
+        let scratchpad_layout =
+            CheckpointLayout::<RwPolicy<()>>::new_untracked(state_sync_scratchpad, height)
+                .expect("failed to create checkpoint layout");
+        // Create at least a file in the scratchpad layout. Otherwise, empty folders can be overridden without errors
+        // and calling "scratchpad_to_checkpoint" twice will not fail as expected.
+        File::create(scratchpad_layout.raw_path().join(SYSTEM_METADATA_FILE)).unwrap();
+
+        let tip_path = state_layout.tip_path();
+        let tip = CheckpointLayout::<RwPolicy<()>>::new_untracked(tip_path, height)
+            .expect("failed to create tip layout");
+        File::create(tip.raw_path().join(SYSTEM_METADATA_FILE)).unwrap();
+
+        // Create marker files in both the scratchpad and tip and try to promote them to a checkpoint.
+        scratchpad_layout
+            .create_unverified_checkpoint_marker()
+            .unwrap();
+        tip.create_unverified_checkpoint_marker().unwrap();
+
+        let checkpoint = state_layout
+            .scratchpad_to_checkpoint(scratchpad_layout, height, None)
+            .unwrap();
+        checkpoint.remove_unverified_checkpoint_marker().unwrap();
+
+        // The checkpoint already exists, therefore promoting the tip to checkpoint should fail.
+        // However, it can still access the checkpoint and try to remove the marker file again from its perspective.
+        let checkpoint_result = state_layout.scratchpad_to_checkpoint(tip, height, None);
+        assert!(checkpoint_result.is_err());
+
+        let res = state_layout
+            .checkpoint(height)
+            .unwrap()
+            .remove_unverified_checkpoint_marker();
+        assert!(res.is_ok());
+
+        drop(checkpoint);
+    });
+}
+
 #[test]
 fn test_canister_id_from_path() {
     assert_eq!(

rs/state_manager/src/checkpoint.rs

@@ -14,7 +14,9 @@ use ic_replicated_state::{
     ExecutionState, ReplicatedState, SchedulerState, SystemState,
 };
 use ic_replicated_state::{CheckpointLoadingMetrics, Memory};
-use ic_state_layout::{CanisterLayout, CanisterStateBits, CheckpointLayout, ReadOnly};
+use ic_state_layout::{
+    CanisterLayout, CanisterStateBits, CheckpointLayout, LoadingPolicy, ReadOnly,
+};
 use ic_types::batch::RawQueryStats;
 use ic_types::{CanisterTimer, Height, Time};
 use ic_utils::thread::parallel_map;
@@ -131,13 +133,15 @@ pub(crate) fn make_checkpoint(
         )?
     };
 
+    cp.remove_unverified_checkpoint_marker()?;
+
     Ok((cp, state, has_downgrade))
 }
 
 /// Calls [load_checkpoint] with a newly created thread pool.
 /// See [load_checkpoint] for further details.
-pub fn load_checkpoint_parallel(
-    checkpoint_layout: &CheckpointLayout<ReadOnly>,
+pub fn load_checkpoint_parallel<T: LoadingPolicy>(
+    checkpoint_layout: &CheckpointLayout<T>,
     own_subnet_type: SubnetType,
     metrics: &CheckpointMetrics,
     fd_factory: Arc<dyn PageAllocatorFileDescriptor>,
@@ -155,8 +159,8 @@ pub fn load_checkpoint_parallel(
 
 /// Loads the node state heighted with `height` using the specified
 /// directory layout.
-pub fn load_checkpoint(
-    checkpoint_layout: &CheckpointLayout<ReadOnly>,
+pub fn load_checkpoint<T: LoadingPolicy>(
+    checkpoint_layout: &CheckpointLayout<T>,
     own_subnet_type: SubnetType,
     metrics: &CheckpointMetrics,
     thread_pool: Option<&mut scoped_threadpool::Pool>,