feat(minter): re-arm the automatic deposits of a dropped sweep

docs/src/development/design.md

@@ -641,7 +641,7 @@ At this rate, it would take 300B instructions / (20K instructions per second) =
 
 Regarding the stable memory, the total size of 500 GiB is reached after **49.6 years** at a rate of 1 event per second. This is more than enough time to introduce some pruning mechanism. In other words, the limit on the stable memory size will not be the limiting factor; the upgrade instruction limit will be reached much sooner.
 
-The automatic flow of [Section 3.1.1](#311-automatic-flow) records no event per registration: `watch_deposit_address` is free, so an event per call would let anyone grow the event log without limit. The watchlist is kept on the heap and written to the event log as a single snapshot by the `pre_upgrade` hook, following the ckETH minter. The snapshot carries the window, the capacity and every entry with its scan progress, so its size is bounded by the capacity of the watchlist, about 100 KB for 1,000 entries. It is recorded on every upgrade, even when the watchlist is empty, since the latest snapshot in the log is what replay restores, and skipping an empty one would let an older snapshot resurrect entries that have since expired and been evicted. Replaying a snapshot restores the watchlist verbatim, without evicting expired entries or enforcing the capacity of the running version, so that the state rebuilt from the log equals the state that produced it; expiry is evaluated against the current time when an entry is accessed. The `QueuedDeposit`, `FailedTransaction` and `ExpiredTransaction` events that follow a snapshot remove and re-insert entries as described in [Section 3.1.1](#311-automatic-flow). A trap in `pre_upgrade` fails the upgrade and leaves the running minter, and its watchlist, in place. A reinstallation erases the event log, and with it the watchlist like everything else (see [Section 3.2.3](#323-reinstallation)); users register again for free.
+The automatic flow of [Section 3.1.1](#311-automatic-flow) records no event per registration: `watch_deposit_address` is free, so an event per call would let anyone grow the event log without limit. The watchlist is kept on the heap and written to the event log as a single snapshot by the `pre_upgrade` hook, following the ckETH minter. The snapshot carries the window, the capacity and every entry with its scan progress, so its size is bounded by the capacity of the watchlist, about 100 KB for 1,000 entries. It is recorded on every upgrade, even when the watchlist is empty, since the latest snapshot in the log is what replay restores, and skipping an empty one would let an older snapshot resurrect entries that have since expired and been evicted. Replaying a snapshot restores the watchlist verbatim, without evicting expired entries or enforcing the capacity of the running version, so that the state rebuilt from the log equals the state that produced it; expiry is evaluated against the current time when an entry is accessed. The `QueuedDeposit` events that follow a snapshot remove entries as described in [Section 3.1.1](#311-automatic-flow); the re-arm of a dropped sweep, like a registration, records no event and reaches the next upgrade through the snapshot. A trap in `pre_upgrade` fails the upgrade and leaves the running minter, and its watchlist, in place. A reinstallation erases the event log, and with it the watchlist like everything else (see [Section 3.2.3](#323-reinstallation)); users register again for free.
 
 ### 3.6. API
 

libs/types/src/lib.rs

@@ -100,7 +100,10 @@ pub enum DepositSolStatus {
     },
     /// The sweep transaction failed, or expired without ever being seen on chain, so no
     /// ckSOL is owed. Calling `deposit_sol` again queues a new sweep of whatever balance
-    /// the deposit address still holds.
+    /// the deposit address still holds. A deposit detected by the automatic flow also puts
+    /// its deposit address back on watch for a new window, unless too many addresses are
+    /// watched, so that `watch_deposit_address` reports it as `Watching` and the balance
+    /// is swept again without a call.
     Dropped {
         /// The signature of the sweep transaction.
         signature: Signature,

minter/cksol_minter.did

@@ -107,7 +107,11 @@ type DepositSolStatus = variant {
     };
     // The sweep transaction failed, or expired without ever being seen on chain,
     // so no ckSOL is owed. Calling `deposit_sol` again queues a new sweep of
-    // whatever balance the deposit address still holds.
+    // whatever balance the deposit address still holds. A deposit detected by
+    // the automatic flow also puts its deposit address back on watch for a new
+    // window, unless too many addresses are watched, so that
+    // `watch_deposit_address` reports it as `Watching` and the balance is swept
+    // again without a call.
     Dropped : record {
         // The signature of the sweep transaction.
         signature: Signature;

minter/src/monitor/mod.rs

@@ -8,8 +8,10 @@ use crate::{
     rpc::{BlockHeight, get_recent_block, get_signature_statuses},
     runtime::CanisterRuntime,
     state::{
-        MinterTransaction, TaskType, audit::process_event, event::EventType, mutate_state,
-        read_state,
+        MinterTransaction, TaskType,
+        audit::{process_dropped_sweep, process_event},
+        event::EventType,
+        mutate_state, read_state,
     },
 };
 use canlog::log;
@@ -105,7 +107,7 @@ async fn check_sweep_transactions<R: CanisterRuntime>(runtime: &R) -> bool {
             "Transaction {signature} finalized with on-chain error: {error}"
         );
         mutate_state(|state| {
-            process_event(
+            process_dropped_sweep(
                 state,
                 EventType::FailedTransaction {
                     signature: *signature,
@@ -168,7 +170,7 @@ fn expire_transactions<R: CanisterRuntime>(
         }
         log!(Priority::Info, "Transaction {signature} expired");
         mutate_state(|state| {
-            process_event(
+            process_dropped_sweep(
                 state,
                 EventType::ExpiredTransaction {
                     signature: *signature,

minter/src/monitor/tests.rs

@@ -13,11 +13,11 @@ use crate::{
     },
     storage::{UndecidedWithdrawalReason, reset_events, undecided_withdrawal_transaction_count},
     test_fixtures::{
-        EventsAssert, GetTransactionResult, MINIMUM_WITHDRAWAL_AMOUNT, NONCE_ACCOUNT, account,
-        confirmed_block_at_height, devnet_sweep, durable_nonce, events, failed_withdrawal_response,
-        finalized_status, init_balance, init_schnorr_master_key, init_state, nonce_account_info,
-        runtime::TestCanisterRuntime, signature, succeeded_transaction_response,
-        succeeded_withdrawal_response,
+        AUTOMATIC_DEPOSIT_FEE, EventsAssert, GetTransactionResult, MINIMUM_WITHDRAWAL_AMOUNT,
+        NONCE_ACCOUNT, account, automatic_deposit_of, confirmed_block_at_height, devnet_sweep,
+        durable_nonce, events, failed_withdrawal_response, finalized_status, init_balance,
+        init_schnorr_master_key, init_state, nonce_account_info, runtime::TestCanisterRuntime,
+        signature, succeeded_transaction_response, succeeded_withdrawal_response,
     },
 };
 use sol_rpc_types::{
@@ -396,6 +396,49 @@ mod finalization {
         }
     }
 
+    #[tokio::test]
+    async fn should_rearm_the_automatic_deposit_of_a_failed_or_expired_sweep() {
+        let failed_status = TransactionStatus {
+            slot: SUBMISSION_SLOT,
+            status: Err(TransactionError::InsufficientFundsForFee),
+            err: Some(TransactionError::InsufficientFundsForFee),
+            confirmation_status: Some(TransactionConfirmationStatus::Finalized),
+        };
+        let cases = [
+            ("failed", CURRENT_BLOCK_HEIGHT, Some(failed_status)),
+            ("expired", EXPIRED_BLOCK_HEIGHT, None),
+        ];
+
+        for (outcome, block_height, status) in cases {
+            reset_state();
+            reset_events();
+            setup();
+            let deposit_id = read_state(|state| state.deposits().next_id());
+            let deposit = automatic_deposit_of(account(1), 1_000_000, AUTOMATIC_DEPOSIT_FEE);
+            events::queue(deposit_id, deposit);
+            events::submit_sweep_at_height(signature(1), vec![deposit_id], block_height);
+            let runtime = TestCanisterRuntime::new()
+                .with_increasing_time()
+                .add_stub_response(SlotResult::Consistent(Ok(CURRENT_SLOT)))
+                .add_stub_response(BlockResult::Consistent(Ok(current_block())))
+                .add_stub_response(SignatureStatusesResult::Consistent(Ok(vec![status])));
+
+            finalize_transactions(runtime).await;
+
+            read_state(|s| {
+                assert_eq!(s.deposits().dropped().len(), 1, "{outcome}");
+                assert_eq!(
+                    s.watchlist()
+                        .iter()
+                        .map(|watched| watched.account)
+                        .collect::<Vec<_>>(),
+                    vec![account(1)],
+                    "{outcome}"
+                );
+            });
+        }
+    }
+
     fn confirmed_status() -> TransactionStatus {
         TransactionStatus {
             slot: 0,

minter/src/state/audit.rs

@@ -3,16 +3,66 @@ use crate::{
     state::{
         QueuedDeposit, SchnorrPublicKey, State,
         event::{Event, EventType},
+        watchlist::RearmError,
     },
     storage,
+    utils::timed_sized_map::Timestamp,
 };
+use canlog::log;
+use cksol_types_internal::log::Priority;
 
 /// Records the given event payload in the event log and updates the state to reflect the change.
 pub fn process_event<R: CanisterRuntime>(state: &mut State, payload: EventType, runtime: &R) {
     apply_state_transition(state, &payload, runtime.time());
     storage::record_event(payload, runtime);
 }
 
+/// Records the given `FailedTransaction` or `ExpiredTransaction` event of a sweep, which
+/// drops its deposits, then puts the accounts of its automatic deposits back on the
+/// watchlist for a fresh window, so that the funds still on their deposit addresses are
+/// swept again without the user's intervention. Every sweep drop of the minter goes through
+/// here; test fixtures and benchmarks that record these events directly skip the re-arm.
+///
+/// The re-arm is not part of the state transition of the event: the watchlist is transient,
+/// so replaying it from the event log would not see the registrations it was made against.
+pub fn process_dropped_sweep<R: CanisterRuntime>(
+    state: &mut State,
+    payload: EventType,
+    runtime: &R,
+) {
+    let signature = match &payload {
+        EventType::FailedTransaction { signature }
+        | EventType::ExpiredTransaction { signature } => *signature,
+        _ => panic!("BUG: {payload:?} does not drop a transaction"),
+    };
+    let deposits: Vec<_> = state
+        .deposits()
+        .swept()
+        .get(&signature)
+        .unwrap_or_else(|| panic!("BUG: cannot drop sweep {signature} that is not swept"))
+        .deposits()
+        .iter()
+        .map(|(deposit_id, deposit)| (*deposit_id, *deposit))
+        .collect();
+    process_event(state, payload, runtime);
+    let now = Timestamp::from_nanos(runtime.time());
+    for (deposit_id, deposit) in deposits {
+        match state.watchlist_mut().rearm(now, &deposit) {
+            Ok(_) => log!(
+                Priority::Info,
+                "Watching account {} again after its deposit {deposit_id} was dropped with sweep {signature}",
+                deposit.account
+            ),
+            Err(RearmError::ManualDeposit) => {}
+            Err(RearmError::WatchlistFull) => log!(
+                Priority::Info,
+                "Failed to watch account {} again after its deposit {deposit_id} was dropped with sweep {signature}: the watchlist is full",
+                deposit.account
+            ),
+        }
+    }
+}
+
 /// Updates the state to reflect the given state transition.
 fn apply_state_transition(state: &mut State, payload: &EventType, timestamp: u64) {
     match payload {

minter/src/state/deposits/mod.rs

@@ -39,7 +39,8 @@ mod tests;
 ///   transaction is what the finalization timer tracks from here on. A sweep is never
 ///   resubmitted: when it fails or expires, its deposits are dropped and their accounts
 ///   released, so that `deposit_sol` can queue a new sweep of the balance still on the
-///   deposit address.
+///   deposit address. The accounts of the automatic deposits are also put back on the
+///   watchlist when it has room, so that a balance scan queues that sweep without a call.
 /// * `finalized`: `getSignatureStatuses` reported the sweep as finalized without error. The
 ///   amount received by the main account is still unknown, because Solana may charge a
 ///   different transaction fee than the sweep was built with.

minter/src/state/event.rs

@@ -113,6 +113,10 @@ pub enum EventType {
         signature: Signature,
     },
     /// A previously submitted Solana transaction has failed.
+    /// The deposits of a failed sweep transaction are dropped. For a sweep, the event is
+    /// recorded by [`process_dropped_sweep`], which also re-arms the automatic deposits.
+    ///
+    /// [`process_dropped_sweep`]: crate::state::audit::process_dropped_sweep
     #[n(6)]
     FailedTransaction {
         /// The signature of the failed Solana transaction.
@@ -121,7 +125,10 @@ pub enum EventType {
     },
     /// A previously submitted Solana transaction has an expired blockhash
     /// and a null on-chain status, meaning it will never be executed.
-    /// The deposits of the expired sweep transaction are dropped.
+    /// The deposits of the expired sweep transaction are dropped. The event is recorded by
+    /// [`process_dropped_sweep`], which also re-arms the automatic deposits.
+    ///
+    /// [`process_dropped_sweep`]: crate::state::audit::process_dropped_sweep
     #[n(7)]
     ExpiredTransaction {
         /// The signature of the expired Solana transaction.

minter/src/state/tests.rs

@@ -166,13 +166,17 @@ mod watched_deposit_addresses {
 mod swept_deposits {
     use super::*;
     use crate::{
-        state::{audit::replay_events, reset_state},
+        state::{
+            audit::{process_dropped_sweep, replay_events},
+            reset_state,
+            watchlist::{MAX_WATCHED_ADDRESSES, Watchlist},
+        },
         storage::{reset_events, with_event_iter},
         test_fixtures::{
-            DEFAULT_BLOCK_HEIGHT,
+            DEFAULT_BLOCK_HEIGHT, REGISTERED_AT, WATCH_WINDOW_NANOS, at,
             events::{credit_sweep, quarantine_sweep, queue_deposits, submit_sweep},
             flow::deposit::{CreditedSweepFlow, DepositFlow, SweepFlow},
-            replayed_recorded_log, sweep_message,
+            replayed_recorded_log, sweep_message, watch,
         },
     };
 
@@ -249,6 +253,94 @@ mod swept_deposits {
         }
     }
 
+    #[test]
+    fn should_rearm_the_automatic_deposits_of_a_dropped_sweep_without_replaying_the_rearm() {
+        const DROPPED_AT: u64 = REGISTERED_AT + WATCH_WINDOW_NANOS + 1;
+        let sweep_signature = signature(SWEEP_SIGNATURE_INDEX);
+        let drop_events = [
+            EventType::FailedTransaction {
+                signature: sweep_signature,
+            },
+            EventType::ExpiredTransaction {
+                signature: sweep_signature,
+            },
+        ];
+        for drop_event in drop_events {
+            reset_state();
+            reset_events();
+            init_state();
+            init_schnorr_master_key();
+            let manual = DepositFlow::queue(account(1), 10_000_000);
+            let automatic =
+                DepositFlow::queue_automatic(account(2), 20_000_000, AUTOMATIC_DEPOSIT_FEE);
+            SweepFlow::of([manual, automatic]).submit(sweep_signature);
+
+            mutate_state(|state| {
+                process_dropped_sweep(
+                    state,
+                    drop_event.clone(),
+                    &TestCanisterRuntime::new().add_times([DROPPED_AT; 3]),
+                )
+            });
+
+            let rearmed = WatchedAddress {
+                account: account(2),
+                address: automatic.address,
+                valid_until: at(DROPPED_AT + WATCH_WINDOW_NANOS),
+                scan_count: 0,
+                last_scanned_at: None,
+                next_scan_at: Some(at(DROPPED_AT)),
+            };
+            read_state(|s| {
+                assert_eq!(s.deposits().dropped().len(), 2, "{drop_event:?}");
+                assert_eq!(
+                    s.watchlist().iter().collect::<Vec<_>>(),
+                    vec![rearmed],
+                    "{drop_event:?}"
+                );
+                assert_eq!(
+                    s.watchlist().due(at(DROPPED_AT)),
+                    vec![rearmed],
+                    "{drop_event:?}"
+                );
+            });
+            let mut live_without_rearm = read_state(State::clone);
+            *live_without_rearm.watchlist_mut() = Watchlist::default();
+            assert_eq!(
+                replayed_recorded_log(),
+                live_without_rearm,
+                "{drop_event:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn should_not_rearm_a_dropped_automatic_deposit_when_the_watchlist_is_full() {
+        init_state();
+        init_schnorr_master_key();
+        let dropped = DepositFlow::queue_automatic(account(0), 20_000_000, AUTOMATIC_DEPOSIT_FEE)
+            .sweep(signature(SWEEP_SIGNATURE_INDEX));
+        for index in 1..=MAX_WATCHED_ADDRESSES.get() {
+            watch(at(REGISTERED_AT), account(index));
+        }
+
+        mutate_state(|state| {
+            process_dropped_sweep(
+                state,
+                EventType::FailedTransaction {
+                    signature: dropped.signature,
+                },
+                &TestCanisterRuntime::new().add_times([REGISTERED_AT; 3]),
+            )
+        });
+
+        read_state(|s| {
+            assert_eq!(s.watchlist().get(at(REGISTERED_AT), &account(0)), None);
+            assert_eq!(s.watchlist().len(), MAX_WATCHED_ADDRESSES.get());
+            assert_eq!(s.deposits().in_flight_id(&account(0)), None);
+        });
+    }
+
     #[test]
     fn should_credit_the_balance_with_the_amount_received_by_the_sweep() {
         init_state();

minter/src/state/watchlist/mod.rs

@@ -1,4 +1,7 @@
-use crate::utils::timed_sized_map::{Entry, InsertError, TimedSizedMap, Timestamp};
+use crate::{
+    state::QueuedDeposit,
+    utils::timed_sized_map::{Entry, InsertError, TimedSizedMap, Timestamp},
+};
 use icrc_ledger_types::icrc1::account::Account;
 use solana_address::Address;
 use std::{num::NonZeroUsize, time::Duration};
@@ -99,6 +102,14 @@ pub struct WatchedAddress {
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct WatchlistFull;
 
+/// Why a dropped deposit was not put back on the watchlist.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum RearmError {
+    /// The deposit was queued by `deposit_sol`, whose caller registered nothing.
+    ManualDeposit,
+    WatchlistFull,
+}
+
 impl Default for Watchlist {
     fn default() -> Self {
         Self {
@@ -127,6 +138,23 @@ impl Watchlist {
         }
     }
 
+    /// Puts the account of a dropped automatic deposit back on the watchlist for a fresh
+    /// window and schedule starting now, so that the funds still on its deposit address are
+    /// swept again without the user's intervention. The account is not watched at that point:
+    /// queueing the deposit removed it from the watchlist, and an account with a deposit in
+    /// flight is not registered.
+    pub fn rearm(
+        &mut self,
+        now: Timestamp,
+        dropped: &QueuedDeposit,
+    ) -> Result<WatchedAddress, RearmError> {
+        if dropped.deposit_fee.is_none() {
+            return Err(RearmError::ManualDeposit);
+        }
+        self.register(now, dropped.account, dropped.address)
+            .map_err(|WatchlistFull| RearmError::WatchlistFull)
+    }
+
     /// The live registration of the account, if any.
     pub fn get(&self, now: Timestamp, account: &Account) -> Option<WatchedAddress> {
         let entry = self.entries.get_entry(now, account)?;