Batch scrape_wallet transactions to handle large wallets

applications/minotari_console_wallet/src/automation/commands.rs

@@ -2302,10 +2302,10 @@ pub async fn command_runner(
                         .await
                         .map_err(CommandError::TransactionServiceError)
                     {
-                        Ok(tx_id) => {
-                            debug!(target: LOG_TARGET, "send-minotari concluded with tx_id {tx_id}");
+                        Ok(tx_ids) => {
+                            debug!(target: LOG_TARGET, "scrape-wallet concluded with tx_ids {:?}", tx_ids);
                             let duration = config.command_send_wait_timeout;
-                            match timeout(duration, monitor_transactions(tms.clone(), vec![tx_id], wait_stage)).await {
+                            match timeout(duration, monitor_transactions(tms.clone(), tx_ids, wait_stage)).await {
                                 Ok(txs) => {
                                     debug!(
                                         target: LOG_TARGET,

base_layer/wallet/src/output_manager_service/handle.rs

@@ -137,7 +137,6 @@ pub enum OutputManagerRequest {
     PreviewCoinJoin((Vec<CompressedCommitment>, MicroMinotari)),
     PreviewCoinSplitEven((Vec<CompressedCommitment>, usize, MicroMinotari)),
     ScrapeWallet {
-        tx_id: TxId,
         fee_per_gram: MicroMinotari,
     },
     CreateCoinJoin {
@@ -199,8 +198,8 @@ impl fmt::Display for OutputManagerRequest {
                 v.metadata_signature.u_y().to_hex(),
                 v.metadata_signature.u_a().to_hex(),
             ),
-            ScrapeWallet { tx_id, fee_per_gram } => {
-                write!(f, "ScrapeWallet (tx_id: {tx_id}, fee_per_gram: {fee_per_gram})")
+            ScrapeWallet { fee_per_gram } => {
+                write!(f, "ScrapeWallet (fee_per_gram: {fee_per_gram})")
             },
             EncumberAggregateUtxo {
                 expected_commitment,
@@ -327,6 +326,7 @@ pub enum OutputManagerResponse<KM> {
     PendingTransactionConfirmed,
     PayToSelfTransaction((MicroMinotari, Transaction, TxId)),
     TransactionBuilderToSend(Box<TransactionBuilder<KM>>),
+    TransactionBuildersToSend(Vec<(TxId, Box<TransactionBuilder<KM>>)>),
     TransactionCancelled,
     SpentOutputs(Vec<DbWalletOutput>),
     UnspentOutputs(Vec<DbWalletOutput>),
@@ -645,16 +645,17 @@ where KM: LegacyTransactionKeyManagerInterface
 
     pub async fn scrape_wallet(
         &mut self,
-        tx_id: TxId,
         fee_per_gram: MicroMinotari,
-    ) -> Result<TransactionBuilder<KM>, OutputManagerError> {
+    ) -> Result<Vec<(TxId, TransactionBuilder<KM>)>, OutputManagerError> {
         match self
             .handle
-            .call(OutputManagerRequest::ScrapeWallet { tx_id, fee_per_gram })
+            .call(OutputManagerRequest::ScrapeWallet { fee_per_gram })
             .await
             .inspect_err(|e| warn!(target: LOG_TARGET, "OutputManagerRequest::ScrapeWallet({e})"))??
         {
-            OutputManagerResponse::TransactionBuilderToSend(tx_builder) => Ok(*tx_builder),
+            OutputManagerResponse::TransactionBuildersToSend(batches) => {
+                Ok(batches.into_iter().map(|(tx_id, builder)| (tx_id, *builder)).collect())
+            },
             _ => Err(OutputManagerError::UnexpectedApiResponse(
                 "OutputManagerRequest::ScrapeWallet".to_string(),
             )),

base_layer/wallet/src/output_manager_service/service.rs

@@ -448,9 +448,16 @@ where
                         .await?,
                 ))
             },
-            OutputManagerRequest::ScrapeWallet { tx_id, fee_per_gram } => self
-                .scrape_wallet(tx_id, fee_per_gram)
-                .map(|tx_builder| OutputManagerResponse::TransactionBuilderToSend(Box::new(tx_builder))),
+            OutputManagerRequest::ScrapeWallet { fee_per_gram } => self
+                .scrape_wallet(fee_per_gram)
+                .map(|batches| {
+                    OutputManagerResponse::TransactionBuildersToSend(
+                        batches
+                            .into_iter()
+                            .map(|(tx_id, builder)| (tx_id, Box::new(builder)))
+                            .collect(),
+                    )
+                }),
 
             OutputManagerRequest::PreviewCoinSplitEven((commitments, number_of_splits, fee_per_gram)) => {
                 Ok(OutputManagerResponse::CoinPreview(
@@ -2723,34 +2730,49 @@ where
 
     pub fn scrape_wallet(
         &mut self,
-        tx_id: TxId,
         fee_per_gram: MicroMinotari,
-    ) -> Result<TransactionBuilder<TKeyManagerInterface>, OutputManagerError> {
+    ) -> Result<Vec<(TxId, TransactionBuilder<TKeyManagerInterface>)>, OutputManagerError> {
         let src_outputs = self
             .resources
             .db
             .fetch_all_unspent_outputs(&self.resources.key_manager)?;
 
-        let mut builder = TransactionBuilder::new(
-            self.resources.consensus_constants.clone(),
-            self.resources.key_manager.clone(),
-            self.resources.network,
-        )?;
-        builder
-            .with_fee_per_gram(fee_per_gram)
-            .with_memo(
-                MemoField::new_open_from_string("scraping wallet", TxType::PaymentToOther)
-                    .map_err(OutputManagerError::InvalidPaymentIdFormat)?,
-            )
-            .with_prevent_fee_gt_amount(self.resources.config.prevent_fee_gt_amount);
-
-        for uo in &src_outputs {
-            builder.with_input(uo.wallet_output.clone())?;
+        if src_outputs.is_empty() {
+            debug!(target: LOG_TARGET, "scrape_wallet called but wallet has no unspent outputs");
+            return Ok(vec![]);
         }
 
-        // encumbering transaction
-        self.resources.db.encumber_outputs(tx_id, src_outputs.clone(), vec![])?;
-        Ok(builder)
+        let mut batches = Vec::new();
+        for batch in src_outputs.chunks(TRANSACTION_INPUTS_LIMIT as usize) {
+            let tx_id = TxId::new_random();
+            let mut builder = TransactionBuilder::new(
+                self.resources.consensus_constants.clone(),
+                self.resources.key_manager.clone(),
+                self.resources.network,
+            )?;
+            builder
+                .with_fee_per_gram(fee_per_gram)
+                .with_memo(
+                    MemoField::new_open_from_string("scraping wallet", TxType::PaymentToOther)
+                        .map_err(OutputManagerError::InvalidPaymentIdFormat)?,
+                )
+                .with_prevent_fee_gt_amount(self.resources.config.prevent_fee_gt_amount);
+
+            for uo in batch {
+                builder.with_input(uo.wallet_output.clone())?;
+            }
+
+            // encumber this batch of outputs
+            self.resources.db.encumber_outputs(tx_id, batch.to_vec(), vec![])?;
+            batches.push((tx_id, builder));
+        }
+        debug!(
+            target: LOG_TARGET,
+            "scrape_wallet: created {} transaction batch(es) from {} unspent outputs",
+            batches.len(),
+            src_outputs.len()
+        );
+        Ok(batches)
     }
 
     pub async fn fetch_unspent_outputs_from_node(

base_layer/wallet/src/transaction_service/handle.rs

@@ -834,7 +834,7 @@ impl TransactionServiceHandle {
         &mut self,
         destination: TariAddress,
         fee_per_gram: MicroMinotari,
-    ) -> Result<TxId, TransactionServiceError> {
+    ) -> Result<Vec<TxId>, TransactionServiceError> {
         match self
             .handle
             .call(TransactionServiceRequest::ScrapeWallet {
@@ -844,7 +844,7 @@ impl TransactionServiceHandle {
             .await
             .inspect_err(|e| warn!(target: LOG_TARGET, "TransactionServiceRequest::ScrapeWallet({e})"))??
         {
-            TransactionServiceResponse::TransactionSent(tx_id) => Ok(tx_id),
+            TransactionServiceResponse::TransactionsSent(tx_ids) => Ok(tx_ids),
             _ => Err(TransactionServiceError::UnexpectedApiResponse(
                 "TransactionServiceRequest::ScrapeWallet".to_string(),
             )),