refactor: use mint indices in ConsolidatedDeposits event

Cargo.toml

@@ -36,6 +36,7 @@ ic-test-utilities-load-wasm = { git = "https://github.com/dfinity/ic", tag = "re
 icrc-cbor = "0.1.0"
 icrc-ledger-client-cdk = "0.1.4"
 icrc-ledger-types = "0.1.12"
+itertools = "0.14.0"
 minicbor = "0.19.1"
 num-traits = "0.2.19"
 phantom_newtype = "0.2.0"

integration_tests/tests/tests.rs

@@ -941,13 +941,11 @@ mod consolidation_tests {
             "Expected SubmittedTransaction event. Events: {events_after:?}"
         );
 
-        // Verify the consolidated deposits match the deposit amount
         for event in &events_after {
-            if let EventType::ConsolidatedDeposits { deposits } = &event.payload {
-                let total: Lamport = deposits.iter().map(|(_, amount)| amount).sum();
-                assert_eq!(
-                    total, DEPOSIT_AMOUNT,
-                    "Consolidated amount should match the deposit amount"
+            if let EventType::ConsolidatedDeposits { mint_indices } = &event.payload {
+                assert!(
+                    !mint_indices.is_empty(),
+                    "ConsolidatedDeposits should contain at least one mint index"
                 );
             }
         }

libs/types-internal/src/event.rs

@@ -87,9 +87,8 @@ pub enum EventType {
     /// Deposited funds from user deposit accounts have been consolidated
     /// into the minter's main account.
     ConsolidatedDeposits {
-        /// The deposit accounts from which funds were consolidated
-        /// and the amount consolidated from each account.
-        deposits: Vec<(Account, Lamport)>,
+        /// The mint indices of the deposits that were consolidated.
+        mint_indices: Vec<u64>,
     },
     /// A withdrawal transaction was signed and is ready to be sent to the network.
     SentWithdrawalTransaction {

minter/Cargo.toml

@@ -31,6 +31,7 @@ ic-http-types = { workspace = true }
 ic-stable-structures = { workspace = true }
 icrc-cbor = { workspace = true }
 icrc-ledger-types = { workspace = true }
+itertools = { workspace = true }
 minicbor = { workspace = true, features = ["derive"] }
 num-traits = { workspace = true }
 phantom_newtype = { workspace = true }

minter/cksol-minter.did

@@ -291,9 +291,8 @@ type EventType = variant {
     // Deposited funds from user deposit accounts have been consolidated
     // into the minter's main account.
     ConsolidatedDeposits : record {
-        // The deposit accounts from which funds were consolidated
-        // and the amount consolidated from each account.
-        deposits: vec record { Account; Lamport };
+        // The mint indices of the deposits that were consolidated.
+        mint_indices: vec nat64;
     };
     // A withdrawal transaction was signed and is ready to be sent to the network.
     SentWithdrawalTransaction : record {

minter/src/consolidate/mod.rs

@@ -1,6 +1,7 @@
 use crate::{
     address::{derivation_path, derive_public_key, lazy_get_schnorr_master_key},
     guard::TimerGuard,
+    numeric::LedgerMintIndex,
     runtime::CanisterRuntime,
     sol_transfer::{CreateTransferError, MAX_SIGNATURES, create_signed_transfer_transaction},
     state::{TaskType, audit::process_event, event::EventType, mutate_state, read_state},
@@ -9,10 +10,12 @@ use crate::{
 use canlog::log;
 use cksol_types_internal::log::Priority;
 use icrc_ledger_types::icrc1::account::Account;
+use itertools::Itertools;
 use sol_rpc_types::{Lamport, Slot};
 use solana_address::Address;
 use solana_hash::Hash;
 use solana_signature::Signature;
+use std::collections::BTreeMap;
 use std::time::Duration;
 use thiserror::Error;
 
@@ -21,30 +24,26 @@ mod tests;
 
 pub const DEPOSIT_CONSOLIDATION_DELAY: Duration = Duration::from_mins(10);
 const MAX_CONCURRENT_TRANSACTIONS: usize = 10;
-const MAX_TRANSFERS_PER_CONSOLIDATION: usize = MAX_SIGNATURES as usize - 1;
+pub(crate) const MAX_TRANSFERS_PER_CONSOLIDATION: usize = MAX_SIGNATURES as usize - 1;
 
 pub async fn consolidate_deposits<R: CanisterRuntime>(runtime: R) {
     let _guard = match TimerGuard::new(TaskType::DepositConsolidation) {
         Ok(guard) => guard,
         Err(_) => return,
     };
 
-    if read_state(|state| state.funds_to_consolidate().is_empty()) {
-        return;
-    }
-
-    let funds_to_consolidate: Vec<_> = read_state(|state| {
-        state
-            .funds_to_consolidate()
-            .clone()
+    let consolidation_rounds: Vec<Vec<_>> =
+        read_state(|s| group_deposits_by_account(s.deposits_to_consolidate()))
             .into_iter()
-            .collect::<Vec<_>>()
             .chunks(MAX_TRANSFERS_PER_CONSOLIDATION)
-            .map(|c| c.to_vec())
-            .collect()
-    });
+            .into_iter()
+            .map(Iterator::collect)
+            .collect();
 
-    for round in funds_to_consolidate.chunks(MAX_CONCURRENT_TRANSACTIONS) {
+    for round in &consolidation_rounds
+        .into_iter()
+        .chunks(MAX_CONCURRENT_TRANSACTIONS)
+    {
         let recent_blockhash = match get_recent_blockhash(&runtime).await {
             Ok(blockhash) => blockhash,
             Err(e) => {
@@ -61,28 +60,27 @@ pub async fn consolidate_deposits<R: CanisterRuntime>(runtime: R) {
                 return;
             }
         };
-        let _ = futures::future::join_all(round.iter().cloned().map(|funds| {
-            try_submit_consolidation_transaction(runtime.clone(), funds, slot, recent_blockhash)
+
+        futures::future::join_all(round.map(async |funds| {
+            match submit_consolidation_transaction(&runtime, funds, slot, recent_blockhash).await {
+                Ok(sig) => log!(Priority::Info, "Submitted consolidation transaction {sig}"),
+                Err(e) => log!(Priority::Info, "Deposit consolidation failed: {e}"),
+            }
         }))
         .await;
     }
 }
 
-async fn try_submit_consolidation_transaction<R: CanisterRuntime>(
-    runtime: R,
-    funds_to_consolidate: Vec<(Account, Lamport)>,
-    slot: Slot,
-    recent_blockhash: Hash,
-) -> Option<Signature> {
-    match submit_consolidation_transaction(&runtime, funds_to_consolidate, slot, recent_blockhash)
-        .await
-    {
-        Ok(signature) => Some(signature),
-        Err(e) => {
-            log!(Priority::Info, "Deposit consolidation failed: {e}");
-            None
-        }
+fn group_deposits_by_account(
+    deposits: &BTreeMap<LedgerMintIndex, (Account, Lamport)>,
+) -> Vec<(Account, (Lamport, Vec<LedgerMintIndex>))> {
+    let mut by_account: BTreeMap<Account, (Lamport, Vec<LedgerMintIndex>)> = BTreeMap::new();
+    for (mint_index, (account, lamport)) in deposits {
+        let entry = by_account.entry(*account).or_default();
+        entry.0 += lamport;
+        entry.1.push(*mint_index);
     }
+    by_account.into_iter().collect()
 }
 
 #[derive(Debug, Error)]
@@ -95,7 +93,7 @@ enum ConsolidationError {
 
 async fn submit_consolidation_transaction<R: CanisterRuntime>(
     runtime: &R,
-    funds_to_consolidate: Vec<(Account, Lamport)>,
+    funds_to_consolidate: Vec<(Account, (Lamport, Vec<LedgerMintIndex>))>,
     slot: Slot,
     recent_blockhash: Hash,
 ) -> Result<Signature, ConsolidationError> {
@@ -108,9 +106,13 @@ async fn submit_consolidation_transaction<R: CanisterRuntime>(
         derive_public_key(&master_key, derivation_path(&minter_account)).serialize_raw(),
     );
 
+    let sources: Vec<(Account, Lamport)> = funds_to_consolidate
+        .iter()
+        .map(|(account, (lamport, _))| (*account, *lamport))
+        .collect();
     let (transaction, signers) = create_signed_transfer_transaction(
         minter_account,
-        &funds_to_consolidate,
+        &sources,
         minter_address,
         recent_blockhash,
         &runtime.signer(),
@@ -119,15 +121,17 @@ async fn submit_consolidation_transaction<R: CanisterRuntime>(
 
     let signature = transaction.signatures[0];
     let message = transaction.message.clone();
+    let mint_indices: Vec<LedgerMintIndex> = funds_to_consolidate
+        .into_iter()
+        .flat_map(|(_, (_, indices))| indices)
+        .collect();
 
     // Record events before trying to submit the transaction to ensure we don't
     // resubmit the same transaction twice in case of a failed submission.
     mutate_state(|state| {
         process_event(
             state,
-            EventType::ConsolidatedDeposits {
-                deposits: funds_to_consolidate,
-            },
+            EventType::ConsolidatedDeposits { mint_indices },
             runtime,
         )
     });