feat: [MR-354] Load & validate protos asynchronously (dfinity#2594)

As a part of the asynchronous checkpointing, we move reading protos to
the Tip thread.
Currently we have the ReplicatedState from execution, serialize it into
a checkpoint, load back the checkpoint in order to ensure it's loadable
and equivalent to the original one, use the loaded checkpoint to switch
the file descriptors in the original state to point to the new files.
The protos are only used for validation and at replica init.
With this change we do loading and validation away from the critical
path.

---------

Co-authored-by: Stefan Schneider <[email protected]>

Author: pakhomov-dfinity

rs/replicated_state/src/canister_state/queues.rs

@@ -152,6 +152,7 @@ pub struct CanisterQueues {
 
     /// Slot and memory reservation stats. Message count and size stats are
     /// maintained separately in the `MessagePool`.
+    #[validate_eq(CompareWithValidateEq)]
     queue_stats: QueueStats,
 
     /// Round-robin schedule for `pop_input()` across ingress, local subnet senders
@@ -1896,7 +1897,7 @@ impl TryFrom<(pb_queues::CanisterQueues, &dyn CheckpointLoadingMetrics)> for Can
 ///
 /// Stats for the enqueued messages themselves (counts and sizes by kind,
 /// context and class) are tracked separately in `message_pool::MessageStats`.
-#[derive(Clone, Eq, PartialEq, Debug, Default)]
+#[derive(Clone, Eq, PartialEq, Debug, Default, ValidateEq)]
 struct QueueStats {
     /// Count of guaranteed response memory reservations across input and output
     /// queues. This is equivalent to the number of outstanding (inbound or outbound)

rs/replicated_state/src/page_map/storage.rs

@@ -161,7 +161,8 @@ pub(crate) struct StorageImpl {
     overlays: Vec<OverlayFile>,
 }
 
-pub fn verify(storage_layout: &dyn StorageLayout) -> Result<(), PersistenceError> {
+/// Validate that the overlay files are loadable.
+pub fn validate(storage_layout: &dyn StorageLayout) -> Result<(), PersistenceError> {
     StorageImpl::load(storage_layout)?;
     Ok(())
 }

rs/replicated_state/src/page_map/storage/tests.rs

@@ -9,7 +9,7 @@ use std::{
 
 use crate::page_map::{
     storage::{
-        verify, Checkpoint, FileIndex, MergeCandidate, MergeDestination, OverlayFile,
+        validate, Checkpoint, FileIndex, MergeCandidate, MergeDestination, OverlayFile,
         PageIndexRange, Shard, Storage, StorageLayout, CURRENT_OVERLAY_VERSION,
         PAGE_INDEX_RANGE_NUM_BYTES, SIZE_NUM_BYTES, VERSION_NUM_BYTES,
     },
@@ -1388,7 +1388,7 @@ fn overlapping_shards_is_an_error() {
             tempdir.path().join("000000_010_vmemory_0.overlay"),
         ]
     );
-    assert!(verify(&ShardedTestStorageLayout {
+    assert!(validate(&ShardedTestStorageLayout {
         dir_path: tempdir.path().to_path_buf(),
         base: tempdir.path().join("vmemory_0.bin"),
         overlay_suffix: "vmemory_0.overlay".to_owned(),
@@ -1399,7 +1399,7 @@ fn overlapping_shards_is_an_error() {
         tempdir.path().join("000000_011_vmemory_0.overlay"),
     )
     .unwrap();
-    assert!(verify(&ShardedTestStorageLayout {
+    assert!(validate(&ShardedTestStorageLayout {
         dir_path: tempdir.path().to_path_buf(),
         base: tempdir.path().join("vmemory_0.bin"),
         overlay_suffix: "vmemory_0.overlay".to_owned(),
@@ -1425,7 +1425,7 @@ fn returns_an_error_if_file_size_is_not_a_multiple_of_page_size() {
         .write_all(&vec![1; PAGE_SIZE / 2])
         .unwrap();
 
-    match verify(&base_only_storage_layout(heap_file.to_path_buf())) {
+    match validate(&base_only_storage_layout(heap_file.to_path_buf())) {
         Err(err) => assert!(
             err.is_invalid_heap_file(),
             "Expected invalid heap file error, got {:?}",

rs/replicated_state/src/replicated_state.rs

@@ -447,6 +447,34 @@ impl ReplicatedState {
         }
     }
 
+    /// References into _all_ fields.
+    pub fn component_refs(
+        &self,
+    ) -> (
+        &BTreeMap<CanisterId, CanisterState>,
+        &SystemMetadata,
+        &CanisterQueues,
+        &Vec<ConsensusResponse>,
+        &RawQueryStats,
+        &CanisterSnapshots,
+    ) {
+        let ReplicatedState {
+            ref canister_states,
+            ref metadata,
+            ref subnet_queues,
+            ref consensus_queue,
+            ref epoch_query_stats,
+            ref canister_snapshots,
+        } = self;
+        (
+            canister_states,
+            metadata,
+            subnet_queues,
+            consensus_queue,
+            epoch_query_stats,
+            canister_snapshots,
+        )
+    }
     pub fn canister_state(&self, canister_id: &CanisterId) -> Option<&CanisterState> {
         self.canister_states.get(canister_id)
     }