chore(ckbtc): Remove empty ReceivedUtxos events from event log (dfinity#3434)

XC-239: ckBTC minter: Clean-up event logs

For historical reasons, ckBTC mainnet canister has a lot of
ReceivedUtxos with empty Utxos, which is removed by this PR.

Author: ninegua

rs/bitcoin/ckbtc/minter/BUILD.bazel

@@ -145,6 +145,7 @@ rust_test(
         "@crate_index//:candid",
         "@crate_index//:flate2",
         "@crate_index//:ic-agent",
+        "@crate_index//:ic-stable-structures",
         "@crate_index//:serde",
         "@crate_index//:tokio",
     ],

rs/bitcoin/ckbtc/minter/src/lifecycle/upgrade.rs

@@ -2,7 +2,7 @@ use crate::logs::P0;
 use crate::state::eventlog::{replay, EventType};
 use crate::state::invariants::CheckInvariantsImpl;
 use crate::state::{replace_state, Mode};
-use crate::storage::{count_events, events, record_event};
+use crate::storage::{count_events, events, migrate_old_events_if_not_empty, record_event};
 use crate::IC_CANISTER_RUNTIME;
 use candid::{CandidType, Deserialize};
 use ic_base_types::CanisterId;
@@ -58,6 +58,9 @@ pub fn post_upgrade(upgrade_args: Option<UpgradeArgs>) {
 
     let start = ic_cdk::api::instruction_counter();
 
+    if let Some(removed) = migrate_old_events_if_not_empty() {
+        log!(P0, "[upgrade]: {} empty events removed", removed)
+    }
     log!(P0, "[upgrade]: replaying {} events", count_events());
 
     let state = replay::<CheckInvariantsImpl>(events()).unwrap_or_else(|e| {

rs/bitcoin/ckbtc/minter/src/storage.rs

@@ -11,8 +11,11 @@ use ic_stable_structures::{
 use serde::Deserialize;
 use std::cell::RefCell;
 
-const LOG_INDEX_MEMORY_ID: MemoryId = MemoryId::new(0);
-const LOG_DATA_MEMORY_ID: MemoryId = MemoryId::new(1);
+const V0_LOG_INDEX_MEMORY_ID: MemoryId = MemoryId::new(0);
+const V0_LOG_DATA_MEMORY_ID: MemoryId = MemoryId::new(1);
+
+const V1_LOG_INDEX_MEMORY_ID: MemoryId = MemoryId::new(2);
+const V1_LOG_DATA_MEMORY_ID: MemoryId = MemoryId::new(3);
 
 type VMem = VirtualMemory<DefaultMemoryImpl>;
 type EventLog = StableLog<Vec<u8>, VMem, VMem>;
@@ -22,13 +25,24 @@ thread_local! {
         MemoryManager::init(DefaultMemoryImpl::default())
     );
 
-    /// The log of the ckBTC state modifications.
-    static EVENTS: RefCell<EventLog> = MEMORY_MANAGER
+    /// The v0 log of the ckBTC state modifications that should be migrated to v1 and then set to empty.
+    static V0_EVENTS: RefCell<EventLog> = MEMORY_MANAGER
+        .with(|m|
+              RefCell::new(
+                  StableLog::init(
+                      m.borrow().get(V0_LOG_INDEX_MEMORY_ID),
+                      m.borrow().get(V0_LOG_DATA_MEMORY_ID)
+                  ).expect("failed to initialize stable log")
+              )
+        );
+
+    /// The latest log of the ckBTC state modifications.
+    static V1_EVENTS: RefCell<EventLog> = MEMORY_MANAGER
         .with(|m|
               RefCell::new(
                   StableLog::init(
-                      m.borrow().get(LOG_INDEX_MEMORY_ID),
-                      m.borrow().get(LOG_DATA_MEMORY_ID)
+                      m.borrow().get(V1_LOG_INDEX_MEMORY_ID),
+                      m.borrow().get(V1_LOG_DATA_MEMORY_ID)
                   ).expect("failed to initialize stable log")
               )
         );
@@ -43,7 +57,7 @@ impl Iterator for EventIterator {
     type Item = Event;
 
     fn next(&mut self) -> Option<Event> {
-        EVENTS.with(|events| {
+        V1_EVENTS.with(|events| {
             let events = events.borrow();
 
             match events.read_entry(self.pos, &mut self.buf) {
@@ -63,7 +77,7 @@ impl Iterator for EventIterator {
 }
 
 /// Encodes an event into a byte array.
-fn encode_event(event: &Event) -> Vec<u8> {
+pub fn encode_event(event: &Event) -> Vec<u8> {
     let mut buf = Vec::new();
     ciborium::ser::into_writer(event, &mut buf).expect("failed to encode a minter event");
     buf
@@ -72,7 +86,7 @@ fn encode_event(event: &Event) -> Vec<u8> {
 /// # Panics
 ///
 /// This function panics if the event decoding fails.
-fn decode_event(buf: &[u8]) -> Event {
+pub fn decode_event(buf: &[u8]) -> Event {
     // For backwards compatibility, we have to handle two cases:
     //  1. Legacy events: raw instances of the event type enum
     //  2. New events: a struct containing a timestamp and an event type
@@ -101,9 +115,49 @@ pub fn events() -> impl Iterator<Item = Event> {
     }
 }
 
+pub fn migrate_old_events_if_not_empty() -> Option<u64> {
+    let mut num_events_removed = None;
+    V0_EVENTS.with(|old_events| {
+        let mut old = old_events.borrow_mut();
+        if old.len() > 0 {
+            V1_EVENTS.with(|new| {
+                num_events_removed = Some(migrate_events(&old, &new.borrow()));
+            });
+            *old = MEMORY_MANAGER.with(|m| {
+                StableLog::new(
+                    m.borrow().get(V0_LOG_INDEX_MEMORY_ID),
+                    m.borrow().get(V0_LOG_DATA_MEMORY_ID),
+                )
+            });
+        }
+    });
+    assert_eq!(
+        V0_EVENTS.with(|events| events.borrow().len()),
+        0,
+        "Old events is not emptied after data migration"
+    );
+    num_events_removed
+}
+
+pub fn migrate_events(old_events: &EventLog, new_events: &EventLog) -> u64 {
+    let mut removed = 0;
+    for bytes in old_events.iter() {
+        let event = decode_event(&bytes);
+        match event.payload {
+            EventType::ReceivedUtxos { utxos, .. } if utxos.is_empty() => removed += 1,
+            _ => {
+                new_events
+                    .append(&bytes)
+                    .expect("failed to append an entry to the new event log");
+            }
+        }
+    }
+    removed
+}
+
 /// Returns the current number of events in the log.
 pub fn count_events() -> u64 {
-    EVENTS.with(|events| events.borrow().len())
+    V1_EVENTS.with(|events| events.borrow().len())
 }
 
 /// Records a new minter event.
@@ -112,7 +166,7 @@ pub fn record_event<R: CanisterRuntime>(payload: EventType, runtime: &R) {
         timestamp: Some(runtime.time()),
         payload,
     });
-    EVENTS.with(|events| {
+    V1_EVENTS.with(|events| {
         events
             .borrow()
             .append(&bytes)

rs/bitcoin/ckbtc/minter/tests/replay_events.rs

@@ -5,6 +5,66 @@ use ic_ckbtc_minter::state::invariants::{CheckInvariants, CheckInvariantsImpl};
 use ic_ckbtc_minter::state::{CkBtcMinterState, Network};
 use std::path::PathBuf;
 
+fn assert_useless_events_is_empty(events: impl Iterator<Item = Event>) {
+    let mut count = 0;
+    for event in events {
+        match &event.payload {
+            EventType::ReceivedUtxos { utxos, .. } if utxos.is_empty() => {
+                count += 1;
+            }
+            _ => {}
+        }
+    }
+    assert_eq!(count, 0);
+}
+
+async fn should_migrate_events_for(file: GetEventsFile) -> CkBtcMinterState {
+    use ic_ckbtc_minter::storage::{decode_event, encode_event, migrate_events};
+    use ic_stable_structures::{
+        log::Log as StableLog,
+        memory_manager::{MemoryId, MemoryManager},
+        DefaultMemoryImpl,
+    };
+
+    file.retrieve_and_store_events_if_env().await;
+
+    let mgr = MemoryManager::init(DefaultMemoryImpl::default());
+    let old_events = StableLog::new(mgr.get(MemoryId::new(0)), mgr.get(MemoryId::new(1)));
+    let new_events = StableLog::new(mgr.get(MemoryId::new(2)), mgr.get(MemoryId::new(3)));
+    let events = file.deserialize().events;
+    events.iter().for_each(|event| {
+        old_events.append(&encode_event(event)).unwrap();
+    });
+    let removed = migrate_events(&old_events, &new_events);
+    assert!(removed > 0);
+    assert!(!new_events.is_empty());
+    assert_eq!(new_events.len() + removed, old_events.len());
+    assert_useless_events_is_empty(new_events.iter().map(|bytes| decode_event(&bytes)));
+
+    let state =
+        replay::<SkipCheckInvariantsImpl>(new_events.iter().map(|bytes| decode_event(&bytes)))
+            .expect("Failed to replay events");
+    state
+        .check_invariants()
+        .expect("Failed to check invariants");
+
+    state
+}
+
+#[tokio::test]
+async fn should_migrate_events_for_mainnet() {
+    let state = should_migrate_events_for(GetEventsFile::Mainnet).await;
+    assert_eq!(state.btc_network, Network::Mainnet);
+    assert_eq!(state.get_total_btc_managed(), 21_723_786_340);
+}
+
+#[tokio::test]
+async fn should_migrate_events_for_testnet() {
+    let state = should_migrate_events_for(GetEventsFile::Testnet).await;
+    assert_eq!(state.btc_network, Network::Testnet);
+    assert_eq!(state.get_total_btc_managed(), 16578205978);
+}
+
 #[tokio::test]
 async fn should_replay_events_for_mainnet() {
     GetEventsFile::Mainnet