feat: add pending-signatures queue for automated deposit flow

minter/src/deposit/automatic/mod.rs

@@ -14,7 +14,17 @@ use cksol_types::UpdateBalanceError;
 use cksol_types_internal::log::Priority;
 use icrc_ledger_types::icrc1::account::Account;
 use sol_rpc_types::{CommitmentLevel, GetSignaturesForAddressParams};
-use std::time::Duration;
+use solana_signature::Signature;
+use std::{
+    cell::RefCell,
+    collections::{BTreeMap, VecDeque},
+    time::Duration,
+};
+
+thread_local! {
+    static PENDING_SIGNATURES: RefCell<BTreeMap<Account, VecDeque<Signature>>> =
+        RefCell::default();
+}
 
 #[cfg(test)]
 mod tests;
@@ -123,8 +133,21 @@ async fn poll_account<R: CanisterRuntime>(
                 "Failed to get signatures for address {deposit_address}: {e}"
             );
         }
-        Ok(_signatures) => {
-            // TODO(DEFI-2780): Process discovered deposit signatures.
+        Ok(signatures) => {
+            let new_sigs: Vec<Signature> = signatures
+                .into_iter()
+                .filter(|s| s.err.is_none())
+                .map(|s| s.signature.into())
+                .collect();
+            if !new_sigs.is_empty() {
+                PENDING_SIGNATURES.with(|pending| {
+                    pending
+                        .borrow_mut()
+                        .entry(account)
+                        .or_default()
+                        .extend(new_sigs);
+                });
+            }
         }
     }
 
@@ -136,3 +159,18 @@ async fn poll_account<R: CanisterRuntime>(
         );
     });
 }
+
+#[cfg(any(test, feature = "canbench-rs"))]
+pub fn pending_signatures_for(account: &Account) -> Vec<Signature> {
+    PENDING_SIGNATURES.with(|p| {
+        p.borrow()
+            .get(account)
+            .map(|q| q.iter().copied().collect())
+            .unwrap_or_default()
+    })
+}
+
+#[cfg(any(test, feature = "canbench-rs"))]
+pub fn reset_pending_signatures() {
+    PENDING_SIGNATURES.with(|p| p.borrow_mut().clear());
+}

minter/src/deposit/automatic/tests.rs

@@ -4,13 +4,35 @@ use crate::{
     state::{event::EventType, read_state},
     test_fixtures::{
         EventsAssert, account, events::start_monitoring_account, init_schnorr_master_key,
-        init_state, runtime::TestCanisterRuntime,
+        init_state, runtime::TestCanisterRuntime, signature,
     },
 };
-use sol_rpc_types::{ConfirmedTransactionStatusWithSignature, MultiRpcResult};
+use sol_rpc_types::{ConfirmedTransactionStatusWithSignature, MultiRpcResult, TransactionError};
 
 type SignaturesResult = MultiRpcResult<Vec<ConfirmedTransactionStatusWithSignature>>;
 
+fn confirmed_tx(signature: Signature) -> ConfirmedTransactionStatusWithSignature {
+    ConfirmedTransactionStatusWithSignature {
+        signature: signature.into(),
+        slot: 12345,
+        err: None,
+        memo: None,
+        block_time: None,
+        confirmation_status: None,
+    }
+}
+
+fn failed_tx(signature: Signature) -> ConfirmedTransactionStatusWithSignature {
+    ConfirmedTransactionStatusWithSignature {
+        signature: signature.into(),
+        slot: 12345,
+        err: Some(TransactionError::AccountNotFound),
+        memo: None,
+        block_time: None,
+        confirmation_status: None,
+    }
+}
+
 fn monitored_accounts_count() -> usize {
     read_state(|s| s.monitored_accounts().len())
 }
@@ -131,6 +153,69 @@ mod poll_monitored_addresses {
         }
     }
 
+    #[tokio::test]
+    async fn should_queue_discovered_signatures() {
+        setup();
+        reset_pending_signatures();
+
+        let acc = account(1);
+        start_monitoring_account(acc);
+
+        let s1 = signature(1);
+        let s2 = signature(2);
+        let runtime = TestCanisterRuntime::new()
+            .with_increasing_time()
+            .add_stub_response(SignaturesResult::Consistent(Ok(vec![
+                confirmed_tx(s1),
+                confirmed_tx(s2),
+            ])));
+
+        poll_monitored_addresses(runtime).await;
+
+        assert_eq!(pending_signatures_for(&acc), vec![s1, s2]);
+    }
+
+    #[tokio::test]
+    async fn should_not_queue_failed_transactions() {
+        setup();
+        reset_pending_signatures();
+
+        let acc = account(1);
+        start_monitoring_account(acc);
+
+        let s_ok = signature(1);
+        let s_fail = signature(2);
+        let runtime = TestCanisterRuntime::new()
+            .with_increasing_time()
+            .add_stub_response(SignaturesResult::Consistent(Ok(vec![
+                confirmed_tx(s_ok),
+                failed_tx(s_fail),
+            ])));
+
+        poll_monitored_addresses(runtime).await;
+
+        assert_eq!(pending_signatures_for(&acc), vec![s_ok]);
+    }
+
+    #[tokio::test]
+    async fn should_not_queue_signatures_if_rpc_call_fails() {
+        setup();
+        reset_pending_signatures();
+
+        let acc = account(1);
+        start_monitoring_account(acc);
+
+        let runtime = TestCanisterRuntime::new()
+            .with_increasing_time()
+            .add_stub_response(SignaturesResult::Consistent(Err(
+                sol_rpc_types::RpcError::ValidationError("RPC error".to_string()),
+            )));
+
+        poll_monitored_addresses(runtime).await;
+
+        assert_eq!(pending_signatures_for(&acc), vec![]);
+    }
+
     fn setup() {
         init_state();
         init_schnorr_master_key();