feat: introduce UnflushedCheckpointOp::DeleteSnapshot and raise a critical error if snapshots are filtered from tip

rs/execution_environment/src/canister_logs.rs

@@ -6,6 +6,7 @@ use ic_management_canister_types_private::{
     FetchCanisterLogsRequest, FetchCanisterLogsResponse, LogVisibilityV2,
 };
 use ic_replicated_state::CanisterState;
+use ic_replicated_state::metadata_state::UnflushedCheckpointOps;
 use ic_types::{NumBytes, NumInstructions, PrincipalId};
 
 pub(crate) fn fetch_canister_logs(
@@ -24,7 +25,7 @@ pub(crate) fn fetch_canister_logs(
         canister_id,
         reply: Some(reply),
         heap_delta_increase: NumBytes::new(0),
-        unflushed_checkpoint_op: None,
+        unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
         deleted_call_context_responses: vec![],
         stop_call_id_to_remove: None,
         stop_contexts_to_reject: vec![],

rs/execution_environment/src/canister_manager.rs

@@ -46,7 +46,7 @@ use ic_replicated_state::canister_state::system_state::wasm_chunk_store::{
     self, CHUNK_SIZE, ChunkValidationResult, WasmChunkHash, WasmChunkStore,
 };
 use ic_replicated_state::metadata_state::{
-    UnflushedCheckpointOp, subnet_call_context_manager::InstallCodeCallId,
+    UnflushedCheckpointOps, subnet_call_context_manager::InstallCodeCallId,
 };
 use ic_replicated_state::page_map::{Buffer, PageAllocatorFileDescriptor};
 use ic_replicated_state::{
@@ -776,7 +776,7 @@ impl CanisterManager {
             canister_id: canister.canister_id(),
             reply: Some(EmptyBlob.encode()),
             heap_delta_increase,
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -1042,7 +1042,7 @@ impl CanisterManager {
             canister_id: canister.canister_id(),
             reply: Some(EmptyBlob.encode()),
             heap_delta_increase: NumBytes::new(0),
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
             deleted_call_context_responses: rejects,
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -1087,7 +1087,7 @@ impl CanisterManager {
             canister_id: canister.canister_id(),
             reply,
             heap_delta_increase: NumBytes::new(0),
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove,
             stop_contexts_to_reject: vec![],
@@ -1121,7 +1121,7 @@ impl CanisterManager {
             canister_id: canister.canister_id(),
             reply: Some(EmptyBlob.encode()),
             heap_delta_increase: NumBytes::new(0),
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject,
@@ -1662,7 +1662,7 @@ impl CanisterManager {
             canister_id: canister.canister_id(),
             reply: Some(EmptyBlob.encode()),
             heap_delta_increase: NumBytes::new(0),
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -1694,7 +1694,7 @@ impl CanisterManager {
             canister_id,
             reply: Some(EmptyBlob.encode()),
             heap_delta_increase: NumBytes::new(0),
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -1809,7 +1809,7 @@ impl CanisterManager {
                     canister_id: canister.canister_id(),
                     reply: Some(reply.encode()),
                     heap_delta_increase: NumBytes::new(0),
-                    unflushed_checkpoint_op: None,
+                    unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
                     deleted_call_context_responses: vec![],
                     stop_call_id_to_remove: None,
                     stop_contexts_to_reject: vec![],
@@ -1857,7 +1857,7 @@ impl CanisterManager {
             canister_id: canister.canister_id(),
             reply: Some(reply.encode()),
             heap_delta_increase: chunk_bytes,
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -1899,7 +1899,7 @@ impl CanisterManager {
             canister_id: canister.canister_id(),
             reply: Some(EmptyBlob.encode()),
             heap_delta_increase: NumBytes::new(0),
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -2117,9 +2117,13 @@ impl CanisterManager {
         let new_snapshot =
             CanisterSnapshot::from_canister(canister, time).map_err(CanisterManagerError::from)?;
 
-        // Delete old snapshot identified by `replace_snapshot`.
+        // Delete old snapshot identified by `replace_snapshot`, recording the deletion
+        // so that its directory is also deleted from the tip.
+        let mut unflushed_checkpoint_ops = UnflushedCheckpointOps::default();
         if let Some(replace_snapshot) = replace_snapshot {
-            canister.canister_snapshots.remove(replace_snapshot);
+            canister
+                .canister_snapshots
+                .remove(replace_snapshot, &mut unflushed_checkpoint_ops);
         }
 
         let heap_delta = new_snapshot.heap_delta();
@@ -2129,6 +2133,7 @@ impl CanisterManager {
         canister
             .canister_snapshots
             .push(snapshot_id, Arc::new(new_snapshot));
+        unflushed_checkpoint_ops.take_snapshot(canister_id, snapshot_id);
 
         // Optionally uninstall the canister's code atomically after taking the
         // snapshot, recording the corresponding `CanisterCodeUninstall` canister
@@ -2176,10 +2181,7 @@ impl CanisterManager {
             canister_id,
             reply: Some(reply.encode()),
             heap_delta_increase: heap_delta,
-            unflushed_checkpoint_op: Some(UnflushedCheckpointOp::TakeSnapshot(
-                canister_id,
-                snapshot_id,
-            )),
+            unflushed_checkpoint_ops,
             deleted_call_context_responses,
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -2553,14 +2555,17 @@ impl CanisterManager {
         let heap_delta = new_canister.heap_delta();
 
         *canister = new_canister;
+
+        // Record the load, so that the canister's files in the tip are replaced by the
+        // snapshot's.
+        let mut unflushed_checkpoint_ops = UnflushedCheckpointOps::default();
+        unflushed_checkpoint_ops.load_snapshot(canister_id, snapshot_id);
+
         Ok(CanisterManagerResponse {
             canister_id,
             reply: Some(EmptyBlob.encode()),
             heap_delta_increase: heap_delta,
-            unflushed_checkpoint_op: Some(UnflushedCheckpointOp::LoadSnapshot(
-                canister_id,
-                snapshot_id,
-            )),
+            unflushed_checkpoint_ops,
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -2627,13 +2632,18 @@ impl CanisterManager {
             resource_saturation,
         )?;
 
-        canister.canister_snapshots.remove(delete_snapshot_id);
+        // Delete the snapshot, recording the deletion so that its directory is also
+        // deleted from the tip.
+        let mut unflushed_checkpoint_ops = UnflushedCheckpointOps::default();
+        canister
+            .canister_snapshots
+            .remove(delete_snapshot_id, &mut unflushed_checkpoint_ops);
 
         Ok(CanisterManagerResponse {
             canister_id: canister.canister_id(),
             reply: Some(EmptyBlob.encode()),
             heap_delta_increase: NumBytes::new(0),
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops,
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -2763,7 +2773,7 @@ impl CanisterManager {
             canister_id,
             reply: Some(reply.encode()),
             heap_delta_increase: NumBytes::new(0),
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -2853,9 +2863,13 @@ impl CanisterManager {
         )?;
         round_limits.instructions -= as_round_instructions(instructions);
 
-        // Delete old snapshot identified by `replace_snapshot`.
+        // Delete old snapshot identified by `replace_snapshot`, recording the deletion
+        // so that its directory is also deleted from the tip.
+        let mut unflushed_checkpoint_ops = UnflushedCheckpointOps::default();
         if let Some(replace_snapshot) = args.replace_snapshot() {
-            canister.canister_snapshots.remove(replace_snapshot);
+            canister
+                .canister_snapshots
+                .remove(replace_snapshot, &mut unflushed_checkpoint_ops);
         }
 
         // Create new snapshot.
@@ -2879,7 +2893,7 @@ impl CanisterManager {
             canister_id,
             reply: Some(reply.encode()),
             heap_delta_increase: heap_delta,
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops,
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],
@@ -3001,7 +3015,7 @@ impl CanisterManager {
                             canister_id: canister.canister_id(),
                             reply: Some(EmptyBlob.encode()),
                             heap_delta_increase: NumBytes::new(0),
-                            unflushed_checkpoint_op: None,
+                            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
                             deleted_call_context_responses: vec![],
                             stop_call_id_to_remove: None,
                             stop_contexts_to_reject: vec![],
@@ -3044,7 +3058,7 @@ impl CanisterManager {
             canister_id: canister.canister_id(),
             reply: Some(EmptyBlob.encode()),
             heap_delta_increase: NumBytes::new(bytes_written),
-            unflushed_checkpoint_op: None,
+            unflushed_checkpoint_ops: UnflushedCheckpointOps::default(),
             deleted_call_context_responses: vec![],
             stop_call_id_to_remove: None,
             stop_contexts_to_reject: vec![],

rs/execution_environment/src/canister_manager/types.rs

@@ -13,7 +13,7 @@ use ic_replicated_state::{
     CanisterState,
     canister_state::canister_snapshots::CanisterSnapshotError,
     canister_state::system_state::wasm_chunk_store::{WasmChunkStore, chunk_size},
-    metadata_state::UnflushedCheckpointOp,
+    metadata_state::UnflushedCheckpointOps,
     metadata_state::subnet_call_context_manager::InstallCodeCallId,
 };
 use ic_types::{
@@ -319,9 +319,9 @@ pub(crate) struct CanisterManagerResponse {
     /// The heap delta increase produced by processing
     /// the current request.
     pub heap_delta_increase: NumBytes,
-    /// An unflushed checkpoint operation that must be handled
+    /// Unflushed checkpoint operations that must be handled
     /// before the next checkpoint.
-    pub unflushed_checkpoint_op: Option<UnflushedCheckpointOp>,
+    pub unflushed_checkpoint_ops: UnflushedCheckpointOps,
     /// (Reject) responses from call contexts that were marked as "deleted" while processing the current request.
     /// Note. A call context is marked as "deleted" when a canister is uninstalled.
     pub deleted_call_context_responses: Vec<Response>,

rs/execution_environment/src/execution_environment.rs

@@ -2147,12 +2147,10 @@ impl ExecutionEnvironment {
                         .heap_delta_debit
                         .saturating_add(&response.heap_delta_increase);
                 }
-                if let Some(unflushed_checkpoint_op) = response.unflushed_checkpoint_op {
-                    state
-                        .metadata
-                        .unflushed_checkpoint_ops
-                        .push(unflushed_checkpoint_op);
-                }
+                state
+                    .metadata
+                    .unflushed_checkpoint_ops
+                    .extend(response.unflushed_checkpoint_ops);
                 if let Some(snapshot_id) = response.snapshot_to_make_immutable
                     && let Some(canister) =
                         state.canister_state_make_mut(&snapshot_id.get_canister_id())

rs/execution_environment/src/execution_environment/tests/canister_snapshots.rs

@@ -1059,6 +1059,91 @@ fn delete_canister_snapshot_succeeds() {
     );
 }
 
+/// Tests that every way of deleting a snapshot records an
+/// `UnflushedCheckpointOp::DeleteSnapshot`, so that the snapshot's directory is deleted
+/// from the tip.
+#[test]
+fn snapshot_deletions_record_unflushed_checkpoint_ops() {
+    const CYCLES: Cycles = Cycles::new(1_000_000_000_000);
+
+    let mut test = ExecutionTestBuilder::new().build();
+    let canister_id = test
+        .canister_from_cycles_and_binary(CYCLES, UNIVERSAL_CANISTER_WASM.to_vec())
+        .unwrap();
+
+    // Installing a canister does not require any checkpoint ops.
+    assert!(test.state().metadata.unflushed_checkpoint_ops.is_empty());
+
+    // Taking a snapshot requires copying the canister's files to the snapshot's
+    // directory in the tip.
+    let (snapshot_id, _) = helper_take_snapshot(&mut test, canister_id);
+    assert_eq!(
+        test.state_mut().metadata.unflushed_checkpoint_ops.take(),
+        vec![UnflushedCheckpointOp::TakeSnapshot(
+            canister_id,
+            snapshot_id
+        )]
+    );
+
+    // Replacing the snapshot requires deleting the replaced snapshot's directory before
+    // creating the new one.
+    let args = TakeCanisterSnapshotArgs::new(canister_id, Some(snapshot_id), None, None);
+    let result = test.subnet_message("take_canister_snapshot", args.encode());
+    let response = CanisterSnapshotResponse::decode(&result.unwrap().bytes()).unwrap();
+    let new_snapshot_id = response.snapshot_id();
+    assert_eq!(
+        test.state_mut().metadata.unflushed_checkpoint_ops.take(),
+        vec![
+            UnflushedCheckpointOp::DeleteSnapshot(snapshot_id),
+            UnflushedCheckpointOp::TakeSnapshot(canister_id, new_snapshot_id),
+        ]
+    );
+
+    // And so does explicitly deleting a snapshot.
+    let args = DeleteCanisterSnapshotArgs::new(canister_id, new_snapshot_id);
+    test.subnet_message("delete_canister_snapshot", args.encode())
+        .unwrap();
+    assert_eq!(
+        test.state_mut().metadata.unflushed_checkpoint_ops.take(),
+        vec![UnflushedCheckpointOp::DeleteSnapshot(new_snapshot_id)]
+    );
+}
+
+/// Tests that deleting a canister also records an
+/// `UnflushedCheckpointOp::DeleteSnapshot` for each of its snapshots, so that their
+/// directories are deleted from the tip along with the canister's.
+#[test]
+fn delete_canister_records_unflushed_checkpoint_ops_for_its_snapshots() {
+    const CYCLES: Cycles = Cycles::new(1_000_000_000_000);
+
+    let mut test = ExecutionTestBuilder::new().build();
+    let canister_id = test
+        .canister_from_cycles_and_binary(CYCLES, UNIVERSAL_CANISTER_WASM.to_vec())
+        .unwrap();
+
+    let (snapshot_id, _) = helper_take_snapshot(&mut test, canister_id);
+    assert_eq!(
+        test.state_mut().metadata.unflushed_checkpoint_ops.take(),
+        vec![UnflushedCheckpointOp::TakeSnapshot(
+            canister_id,
+            snapshot_id
+        )]
+    );
+
+    let _ = test.stop_canister(canister_id);
+    test.process_stopping_canisters();
+    test.delete_canister(canister_id).unwrap();
+
+    // Both the snapshot's and the canister's directories must be deleted from the tip.
+    assert_eq!(
+        test.state_mut().metadata.unflushed_checkpoint_ops.take(),
+        vec![
+            UnflushedCheckpointOp::DeleteSnapshot(snapshot_id),
+            UnflushedCheckpointOp::DeleteCanister(canister_id),
+        ]
+    );
+}
+
 #[test]
 fn list_canister_snapshot_fails_canister_not_found() {
     let own_subnet = subnet_test_id(1);