fix(station): Make DR sync + logging more efficient

Cargo.toml

@@ -57,7 +57,7 @@ ic-cdk = "0.16.0"
 ic-cdk-macros = "0.16.0"
 ic-cdk-timers = "0.9.0"
 ic-ledger-types = "0.12.0"
-ic-stable-structures = "0.6.4"
+ic-stable-structures = "0.6.6"
 icrc-ledger-types = "0.1.6"
 ic-utils = "0.38"
 itertools = "0.13.0"

core/station/impl/results.yml

@@ -13,7 +13,7 @@ benches:
     scopes: {}
   find_500_external_canister_policies_from_50k_dataset:
     total:
-      instructions: 27465679
+      instructions: 28629145
       heap_increase: 0
       stable_memory_increase: 0
     scopes: {}
@@ -25,32 +25,32 @@ benches:
     scopes: {}
   list_1k_requests:
     total:
-      instructions: 142176376
+      instructions: 142128177
       heap_increase: 14
       stable_memory_increase: 0
     scopes: {}
   list_external_canisters_with_all_statuses:
     total:
-      instructions: 213172583
+      instructions: 213226507
       heap_increase: 0
       stable_memory_increase: 0
     scopes: {}
   repository_find_1k_requests_from_10k_dataset_default_filters:
     total:
-      instructions: 92231168
+      instructions: 92053986
       heap_increase: 17
       stable_memory_increase: 0
     scopes: {}
   service_filter_5k_requests_from_100k_dataset:
     total:
-      instructions: 680919877
+      instructions: 680738365
       heap_increase: 106
       stable_memory_increase: 16
     scopes: {}
   service_find_all_requests_from_2k_dataset:
     total:
-      instructions: 275772307
+      instructions: 275726235
       heap_increase: 44
       stable_memory_increase: 16
     scopes: {}
-version: 0.1.5
+version: 0.1.8

core/station/impl/src/mappers/account.rs

@@ -216,3 +216,23 @@ impl From<station_api::ChangeAssets> for ChangeAssets {
         }
     }
 }
+
+impl From<Account> for upgrader_api::MultiAssetAccount {
+    fn from(account: Account) -> Self {
+        Self {
+            id: Uuid::from_bytes(account.id).hyphenated().to_string(),
+            seed: account.seed,
+            assets: account
+                .assets
+                .iter()
+                .map(|account_asset| {
+                    Uuid::from_bytes(account_asset.asset_id)
+                        .hyphenated()
+                        .to_string()
+                })
+                .collect(),
+            name: account.name.clone(),
+            metadata: account.metadata.clone().into(),
+        }
+    }
+}

core/station/impl/src/mappers/asset.rs

@@ -26,3 +26,17 @@ impl From<AssetCallerPrivileges> for AssetCallerPrivilegesDTO {
         }
     }
 }
+
+impl From<Asset> for upgrader_api::Asset {
+    fn from(asset: Asset) -> Self {
+        upgrader_api::Asset {
+            id: Uuid::from_bytes(asset.id).hyphenated().to_string(),
+            blockchain: asset.blockchain.to_string(),
+            symbol: asset.symbol.clone(),
+            name: asset.name.clone(),
+            decimals: asset.decimals,
+            standards: asset.standards.iter().map(|s| s.to_string()).collect(),
+            metadata: asset.metadata.clone().into(),
+        }
+    }
+}

core/station/impl/src/models/request.rs

@@ -198,6 +198,7 @@ fn validate_request_operation_foreign_keys(
         RequestOperation::ManageSystemInfo(_) => (),
         RequestOperation::Transfer(op) => {
             EnsureAccount::id_exists(&op.input.from_account_id)?;
+            EnsureAsset::id_exists(&op.input.from_asset_id)?;
         }
         RequestOperation::AddAccount(op) => {
             op.input.read_permission.validate()?;

core/station/impl/src/repositories/account.rs

@@ -1,11 +1,14 @@
-use super::indexes::unique_index::UniqueIndexRepository;
+use super::{indexes::unique_index::UniqueIndexRepository, InsertEntryObserverArgs};
 use crate::{
     core::{
         metrics::ACCOUNT_METRICS, observer::Observer, utils::format_unique_string,
         with_memory_manager, Memory, ACCOUNT_MEMORY_ID,
     },
     models::{indexes::unique_index::UniqueIndexKey, Account, AccountId, AccountKey},
-    services::disaster_recovery_sync_accounts_and_assets_on_change,
+    services::{
+        disaster_recovery_sync_accounts_and_assets_on_insert,
+        disaster_recovery_sync_accounts_and_assets_on_remove,
+    },
 };
 use ic_stable_structures::{memory_manager::VirtualMemory, StableBTreeMap};
 use lazy_static::lazy_static;
@@ -30,16 +33,21 @@ lazy_static! {
 #[derive(Debug)]
 pub struct AccountRepository {
     unique_index: UniqueIndexRepository,
-    change_observer: Observer<()>,
+    insert_observer: Observer<InsertEntryObserverArgs<Account>>,
+    remove_observer: Observer<()>,
 }
 
 impl Default for AccountRepository {
     fn default() -> Self {
-        let mut change_observer = Observer::default();
-        disaster_recovery_sync_accounts_and_assets_on_change(&mut change_observer);
+        let mut remove_observer = Observer::default();
+        disaster_recovery_sync_accounts_and_assets_on_remove(&mut remove_observer);
+
+        let mut insert_observer = Observer::default();
+        disaster_recovery_sync_accounts_and_assets_on_insert(&mut insert_observer);
 
         Self {
-            change_observer,
+            insert_observer,
+            remove_observer,
             unique_index: UniqueIndexRepository::default(),
         }
     }
@@ -94,9 +102,14 @@ impl Repository<AccountKey, Account, VirtualMemory<Memory>> for AccountRepositor
 
             self.save_entry_indexes(&value, prev.as_ref());
 
-            self.change_observer.notify(&());
+            let args = InsertEntryObserverArgs {
+                current: value,
+                prev,
+            };
 
-            prev
+            self.insert_observer.notify(&args);
+
+            args.prev
         })
     }
 
@@ -117,7 +130,7 @@ impl Repository<AccountKey, Account, VirtualMemory<Memory>> for AccountRepositor
                 self.remove_entry_indexes(prev);
             }
 
-            self.change_observer.notify(&());
+            self.remove_observer.notify(&());
 
             prev
         })
@@ -153,13 +166,6 @@ impl AccountRepository {
         self.unique_index
             .get(&UniqueIndexKey::AccountName(format_unique_string(name)))
     }
-
-    pub fn with_empty_observers() -> Self {
-        Self {
-            change_observer: Observer::default(),
-            ..Default::default()
-        }
-    }
 }
 
 #[derive(Debug, Clone)]

core/station/impl/src/repositories/asset.rs

@@ -1,11 +1,14 @@
-use super::indexes::unique_index::UniqueIndexRepository;
+use super::{indexes::unique_index::UniqueIndexRepository, InsertEntryObserverArgs};
 use crate::{
     core::{
         cache::Cache, ic_cdk::api::print, metrics::ASSET_METRICS, observer::Observer,
         with_memory_manager, Memory, ASSET_MEMORY_ID,
     },
     models::{indexes::unique_index::UniqueIndexKey, Asset, AssetId},
-    services::disaster_recovery_sync_accounts_and_assets_on_change,
+    services::{
+        disaster_recovery_sync_accounts_and_assets_on_insert,
+        disaster_recovery_sync_accounts_and_assets_on_remove,
+    },
 };
 use ic_stable_structures::{memory_manager::VirtualMemory, StableBTreeMap};
 use lazy_static::lazy_static;
@@ -33,16 +36,21 @@ lazy_static! {
 #[derive(Debug)]
 pub struct AssetRepository {
     unique_index: UniqueIndexRepository,
-    change_observer: Observer<()>,
+    insert_observer: Observer<InsertEntryObserverArgs<Asset>>,
+    remove_observer: Observer<()>,
 }
 
 impl Default for AssetRepository {
     fn default() -> Self {
-        let mut change_observer = Observer::default();
-        disaster_recovery_sync_accounts_and_assets_on_change(&mut change_observer);
+        let mut remove_observer = Observer::default();
+        disaster_recovery_sync_accounts_and_assets_on_remove(&mut remove_observer);
+
+        let mut insert_observer = Observer::default();
+        disaster_recovery_sync_accounts_and_assets_on_insert(&mut insert_observer);
 
         Self {
-            change_observer,
+            insert_observer,
+            remove_observer,
             unique_index: UniqueIndexRepository::default(),
         }
     }
@@ -130,9 +138,14 @@ impl Repository<UUID, Asset, VirtualMemory<Memory>> for AssetRepository {
 
             self.save_entry_indexes(&value, prev.as_ref());
 
-            self.change_observer.notify(&());
+            let args = InsertEntryObserverArgs {
+                current: value,
+                prev,
+            };
 
-            prev
+            self.insert_observer.notify(&args);
+
+            args.prev
         })
     }
 
@@ -153,7 +166,7 @@ impl Repository<UUID, Asset, VirtualMemory<Memory>> for AssetRepository {
                 self.remove_entry_indexes(prev);
             }
 
-            self.change_observer.notify(&());
+            self.remove_observer.notify(&());
 
             prev
         })

core/station/impl/src/repositories/mod.rs

@@ -36,3 +36,8 @@ pub use asset::*;
 pub mod permission;
 
 pub mod indexes;
+
+pub struct InsertEntryObserverArgs<T> {
+    pub current: T,
+    pub prev: Option<T>,
+}