fedimint_mintv2_client/

lib.rs

#![deny(clippy::pedantic)]
#![allow(clippy::cast_possible_truncation)]
#![allow(clippy::missing_errors_doc)]
#![allow(clippy::missing_panics_doc)]
#![allow(clippy::module_name_repetitions)]
#![allow(clippy::must_use_candidate)]
#![allow(clippy::return_self_not_must_use)]
#![allow(clippy::too_many_lines)]

pub use fedimint_mintv2_common as common;

mod api;
#[cfg(feature = "cli")]
mod cli;
pub mod client_db;
mod ecash;
mod events;
mod input;
pub mod issuance;
mod output;
mod receive;

use std::collections::{BTreeMap, BTreeSet};
use std::convert::Infallible;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context as _, anyhow};
use bitcoin_hashes::sha256;
use client_db::{RecoveryState, RecoveryStateKey, SpendableNoteAmountPrefix, SpendableNotePrefix};
pub use events::*;
use fedimint_api_client::api::DynModuleApi;
use fedimint_client::module::ClientModule;
use fedimint_client::transaction::{
    ClientInput, ClientInputBundle, ClientInputSM, ClientOutput, ClientOutputBundle,
    ClientOutputSM, FeeQuote, FeeQuoteRequest, TransactionBuilder,
};
use fedimint_client_module::db::ClientModuleMigrationFn;
use fedimint_client_module::error::{
    InsufficientBalanceError, OperationLookupError, TransactionSubmitError,
};
use fedimint_client_module::module::init::{
    ClientModuleInit, ClientModuleInitArgs, ClientModuleRecoverArgs,
    ClientModuleRecoveryPrepareArgs, RecoveryMode,
};
use fedimint_client_module::module::recovery::{NoModuleBackup, RecoveryProgress};
use fedimint_client_module::module::{
    ClientContext, OutPointRange, PrimaryModulePriority, PrimaryModuleSupport,
};
use fedimint_client_module::sm::{Context, DynState, ModuleNotifier, State, StateTransition};
use fedimint_client_module::{DynGlobalClientContext, sm_enum_variant_translation};
use fedimint_core::base32::{self, FEDIMINT_PREFIX};
use fedimint_core::config::FederationId;
use fedimint_core::core::{IntoDynInstance, ModuleInstanceId, ModuleKind, OperationId};
use fedimint_core::db::{DatabaseTransaction, DatabaseVersion, IDatabaseTransactionOpsCoreTyped};
use fedimint_core::encoding::{Decodable, Encodable};
use fedimint_core::module::{
    AmountUnit, Amounts, ApiVersion, CommonModuleInit, ModuleCommon, ModuleInit, MultiApiVersion,
};
use fedimint_core::secp256k1::rand::{Rng, thread_rng};
use fedimint_core::secp256k1::{Keypair, PublicKey};
use fedimint_core::util::backoff_util::custom_backoff;
use fedimint_core::util::{BoxStream, NextOrPending};
use fedimint_core::{Amount, OutPoint, PeerId, apply, async_trait_maybe_send};
use fedimint_derive_secret::DerivableSecret;
use fedimint_mintv2_common::config::{FeeConsensus, MintClientConfig, client_denominations};
use fedimint_mintv2_common::{
    Denomination, KIND, MintCommonInit, MintInput, MintModuleTypes, MintOutput, Note, RecoveryItem,
};
use futures::{StreamExt, pin_mut};
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tbs::AggregatePublicKey;
use thiserror::Error;

use crate::api::MintV2ModuleApi;
use crate::client_db::SpendableNoteKey;
pub use crate::ecash::ECash;
use crate::input::{InputSMCommon, InputSMState, InputStateMachine};
use crate::issuance::NoteIssuanceRequest;
use crate::output::{MintOutputStateMachine, OutputSMCommon, OutputSMState};
use crate::receive::{ReceiveSMState, ReceiveStateMachine};

const TARGET_PER_DENOMINATION: usize = 3;
const SLICE_SIZE: u64 = 10000;
/// How long a guardian sits out after failing to answer a slice request.
const PEER_READMISSION: Duration = Duration::from_secs(60);
/// How long a single peer is given to answer a slice request.
///
/// A slice takes a second or two from a healthy guardian, so waiting half a
/// minute only delays noticing that one is not going to answer. The timeout
/// grows on every failed attempt up to [`MAX_SLICE_TIMEOUT`], so a client
/// on a slow connection where every guardian exceeds the initial timeout
/// still makes progress instead of retrying forever.
const SLICE_TIMEOUT: Duration = Duration::from_secs(10);
/// Upper bound for the per-retry growth of [`SLICE_TIMEOUT`]; the flat
/// timeout used before the growth was introduced.
const MAX_SLICE_TIMEOUT: Duration = Duration::from_secs(30);
const PARALLEL_HASH_REQUESTS: usize = 10;
const PARALLEL_SLICE_REQUESTS: usize = 10;

#[derive(Debug, Clone, PartialEq, Eq, Hash, Encodable, Decodable)]
pub struct SpendableNote {
    pub denomination: Denomination,
    pub keypair: Keypair,
    pub signature: tbs::Signature,
}

impl SpendableNote {
    pub fn amount(&self) -> Amount {
        self.denomination.amount()
    }
}

impl SpendableNote {
    fn nonce(&self) -> PublicKey {
        self.keypair.public_key()
    }

    fn note(&self) -> Note {
        Note {
            denomination: self.denomination,
            nonce: self.nonce(),
            signature: self.signature,
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum MintOperationMeta {
    Send {
        ecash: String,
        custom_meta: Value,
    },
    Reissue {
        change_outpoint_range: OutPointRange,
        amount: Amount,
        custom_meta: Value,
    },
    Receive {
        change_outpoint_range: OutPointRange,
        ecash: String,
        custom_meta: Value,
    },
}

#[derive(Debug, Clone)]
pub struct MintClientInit;

impl ModuleInit for MintClientInit {
    type Common = MintCommonInit;

    async fn dump_database(
        &self,
        _dbtx: &mut DatabaseTransaction<'_>,
        _prefix_names: Vec<String>,
    ) -> Box<dyn Iterator<Item = (String, Box<dyn erased_serde::Serialize + Send>)> + '_> {
        Box::new(BTreeMap::new().into_iter())
    }
}

#[apply(async_trait_maybe_send!)]
impl ClientModuleInit for MintClientInit {
    type Module = MintClientModule;

    fn supported_api_versions(&self) -> MultiApiVersion {
        MultiApiVersion::try_from_iter([ApiVersion { major: 0, minor: 1 }])
            .expect("no version conflicts")
    }

    fn recovery_mode(&self) -> RecoveryMode {
        RecoveryMode::Usable
    }

    async fn prepare_recovery(&self, args: &ClientModuleRecoveryPrepareArgs) -> anyhow::Result<()> {
        if args
            .db()
            .begin_transaction_nc()
            .await
            .get_value(&RecoveryStateKey)
            .await
            .is_some()
        {
            return Ok(());
        }

        // The total is the number of items the federation had processed when
        // the recovery began, so every note this client issues from here on
        // lands beyond it and is never rediscovered by the scan. Committing
        // that bound before the module is initialized is what lets the module
        // be used while its own recovery is still running: the two cannot
        // arrive at the same note.
        let state = RecoveryState {
            next_index: 0,
            total_items: args.module_api().fetch_recovery_count().await?,
            requests: BTreeMap::new(),
            nonces: BTreeSet::new(),
        };

        let mut dbtx = args.db().begin_transaction().await;

        dbtx.insert_entry(&RecoveryStateKey, &state).await;

        dbtx.commit_tx().await;

        Ok(())
    }

    async fn recover(
        &self,
        args: &ClientModuleRecoverArgs<Self>,
        _snapshot: Option<&NoModuleBackup>,
    ) -> anyhow::Result<Option<Amount>> {
        let mut state = args
            .db()
            .begin_transaction_nc()
            .await
            .get_value(&RecoveryStateKey)
            .await
            .expect("Prepare recovery commits the state before the recovery is started");

        if state.next_index == state.total_items {
            return Ok(None);
        }

        let peer_pool = PeerPool::new(args.api().all_peers());

        let mut recovery_stream = futures::stream::iter(
            (state.next_index..state.total_items).step_by(SLICE_SIZE as usize),
        )
        .map(|start| {
            let api = args.module_api().clone();
            let end = std::cmp::min(start + SLICE_SIZE, state.total_items);

            async move { (start, end, api.fetch_recovery_slice_hash(start, end).await) }
        })
        .buffered(PARALLEL_HASH_REQUESTS)
        .map(|(start, end, hash)| {
            let module_api = args.module_api().clone();
            let peer_pool = peer_pool.clone();

            async move {
                (
                    start,
                    download_slice(module_api, peer_pool, start, end, hash).await,
                )
            }
        })
        // Unordered, so a guardian that goes quiet holds up only its own
        // slice. Delivering in order would let it halt everything: the other
        // requests would finish, their results would fill the buffer, and no
        // new request could start until the straggler returned. Downloading
        // runs ahead of it instead, and the items are put back in order below.
        .buffer_unordered(PARALLEL_SLICE_REQUESTS);

        let tweak_filter = issuance::tweak_filter(args.module_root_secret());

        // Slices that arrived before the ones in front of them. An input
        // spends an output that has to have been seen already, so items are
        // scanned in index order however they turn up. This holds whatever
        // downloaded while a slice was outstanding, which the timeout bounds.
        let mut pending: BTreeMap<u64, Vec<RecoveryItem>> = BTreeMap::new();

        loop {
            let items = loop {
                if let Some(items) = pending.remove(&state.next_index) {
                    break items;
                }

                let (start, items) = recovery_stream
                    .next()
                    .await
                    .context("Recovery stream finished before recovery is complete")?;

                pending.insert(start, items);
            };

            for item in &items {
                match item {
                    RecoveryItem::Output {
                        denomination,
                        nonce_hash,
                        tweak,
                    } => {
                        if !issuance::check_tweak(*tweak, tweak_filter) {
                            continue;
                        }
                        let output_secret = issuance::output_secret(
                            *denomination,
                            *tweak,
                            args.module_root_secret(),
                        );

                        if !issuance::check_nonce(&output_secret, *nonce_hash) {
                            continue;
                        }

                        let computed_nonce_hash = issuance::nonce(&output_secret).consensus_hash();

                        // Ignore possible duplicate nonces
                        if !state.nonces.insert(computed_nonce_hash) {
                            continue;
                        }

                        state.requests.insert(
                            computed_nonce_hash,
                            NoteIssuanceRequest::new(
                                *denomination,
                                *tweak,
                                args.module_root_secret(),
                            ),
                        );
                    }
                    RecoveryItem::Input { nonce_hash } => {
                        state.requests.remove(nonce_hash);
                        state.nonces.remove(nonce_hash);
                    }
                }
            }

            state.next_index += items.len() as u64;

            let mut dbtx = args.db().begin_transaction().await;

            dbtx.insert_entry(&RecoveryStateKey, &state).await;

            if state.next_index == state.total_items {
                // Total value of the notes reconstructed during recovery
                let recovered_amount = state
                    .requests
                    .values()
                    .map(|request| request.denomination.amount())
                    .sum::<Amount>();

                let state_machines = args
                    .context()
                    .map_dyn(vec![MintClientStateMachines::Output(
                        MintOutputStateMachine {
                            common: OutputSMCommon {
                                operation_id: OperationId::new_random(),
                                range: None,
                                issuance_requests: state.requests.into_values().collect(),
                            },
                            state: OutputSMState::Pending,
                        },
                    )])
                    .collect();

                args.context()
                    .add_state_machines_dbtx(&mut dbtx.to_ref_nc(), state_machines)
                    .await
                    .expect("state machine is valid");

                dbtx.commit_tx().await;

                return Ok(Some(recovered_amount));
            }

            dbtx.commit_tx().await;

            args.update_recovery_progress(RecoveryProgress {
                complete: state.next_index.try_into().unwrap_or(u32::MAX),
                total: state.total_items.try_into().unwrap_or(u32::MAX),
            });
        }
    }

    async fn init(&self, args: &ClientModuleInitArgs<Self>) -> anyhow::Result<Self::Module> {
        let (tweak_sender, tweak_receiver) = async_channel::bounded(50);

        let filter = issuance::tweak_filter(args.module_root_secret());

        // Only ~1/65536 random tweaks pass the filter, so grinding is
        // CPU-bound. Pre-grind in the background and feed tweaks through a
        // channel to keep the cost off the spend path. `send` only suspends
        // once the channel is full, so we yield explicitly on each found tweak
        // to keep single-threaded (wasm) runtimes responsive while the channel
        // still has space.
        fedimint_core::task::spawn("mintv2-tweak-grinder", async move {
            loop {
                let tweak: [u8; 16] = thread_rng().r#gen();

                if !issuance::check_tweak(tweak, filter) {
                    continue;
                }

                if tweak_sender.send(tweak).await.is_err() {
                    return;
                }

                fedimint_core::task::sleep(Duration::ZERO).await;
            }
        });

        Ok(MintClientModule {
            federation_id: *args.federation_id(),
            cfg: args.cfg().clone(),
            root_secret: args.module_root_secret().clone(),
            notifier: args.notifier().clone(),
            client_ctx: args.context(),
            balance_update_sender: tokio::sync::watch::channel(()).0,
            tweak_receiver,
        })
    }

    fn get_database_migrations(&self) -> BTreeMap<DatabaseVersion, ClientModuleMigrationFn> {
        BTreeMap::new()
    }
}

#[derive(Debug)]
pub struct MintClientModule {
    federation_id: FederationId,
    cfg: MintClientConfig,
    root_secret: DerivableSecret,
    notifier: ModuleNotifier<MintClientStateMachines>,
    client_ctx: ClientContext<Self>,
    balance_update_sender: tokio::sync::watch::Sender<()>,
    tweak_receiver: async_channel::Receiver<[u8; 16]>,
}

#[derive(Debug, Clone)]
pub struct MintClientContext {
    client_ctx: ClientContext<MintClientModule>,
    tbs_agg_pks: BTreeMap<Denomination, AggregatePublicKey>,
    tbs_pks: BTreeMap<Denomination, BTreeMap<PeerId, tbs::PublicKeyShare>>,
    pub balance_update_sender: tokio::sync::watch::Sender<()>,
}

impl Context for MintClientContext {
    const KIND: Option<ModuleKind> = Some(KIND);
}

#[apply(async_trait_maybe_send!)]
impl ClientModule for MintClientModule {
    type Init = MintClientInit;
    type Common = MintModuleTypes;
    type Backup = NoModuleBackup;
    type ModuleStateMachineContext = MintClientContext;
    type States = MintClientStateMachines;

    fn context(&self) -> Self::ModuleStateMachineContext {
        MintClientContext {
            client_ctx: self.client_ctx.clone(),
            tbs_agg_pks: self.cfg.tbs_agg_pks.clone(),
            tbs_pks: self.cfg.tbs_pks.clone(),
            balance_update_sender: self.balance_update_sender.clone(),
        }
    }

    fn input_fee(
        &self,
        amounts: &Amounts,
        _input: &<Self::Common as ModuleCommon>::Input,
    ) -> Option<Amounts> {
        let unit = self.cfg.amount_unit;
        let amount = amounts.get(&unit).copied().unwrap_or_default();
        let fee = self.cfg.fee_consensus.fee(amount);

        Some(Amounts::new_custom(unit, fee))
    }

    fn output_fee(
        &self,
        amounts: &Amounts,
        _output: &<Self::Common as ModuleCommon>::Output,
    ) -> Option<Amounts> {
        let unit = self.cfg.amount_unit;
        let amount = amounts.get(&unit).copied().unwrap_or_default();
        let fee = self.cfg.fee_consensus.fee(amount);

        Some(Amounts::new_custom(unit, fee))
    }

    #[cfg(feature = "cli")]
    async fn handle_cli_command(
        &self,
        args: &[std::ffi::OsString],
    ) -> anyhow::Result<serde_json::Value> {
        cli::handle_cli_command(self, args).await
    }

    fn supports_being_primary(&self) -> PrimaryModuleSupport {
        PrimaryModuleSupport::selected(PrimaryModulePriority::HIGH, [self.cfg.amount_unit])
    }

    async fn create_final_inputs_and_outputs(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        operation_id: OperationId,
        unit: AmountUnit,
        mut input_amount: Amount,
        mut output_amount: Amount,
    ) -> anyhow::Result<(
        ClientInputBundle<MintInput, MintClientStateMachines>,
        ClientOutputBundle<MintOutput, MintClientStateMachines>,
    )> {
        if unit != self.cfg.amount_unit {
            anyhow::bail!("Module can only handle its configured amount unit");
        }

        let requested_amount = output_amount.saturating_sub(input_amount);
        // `select_funding_input` only reads notes, so the balance it left behind is
        // still accurate for reporting a total below.
        let Some(funding_notes) = self.select_funding_input(dbtx, requested_amount).await else {
            let total_amount = self.get_balance(dbtx, unit).await;
            return Err(InsufficientBalanceError {
                requested_amount,
                total_amount,
            }
            .into());
        };

        for note in &funding_notes {
            self.remove_spendable_note(dbtx, note).await;
        }

        input_amount += funding_notes.iter().map(SpendableNote::amount).sum();

        output_amount += funding_notes
            .iter()
            .map(|input| self.cfg.fee_consensus.fee(input.amount()))
            .sum();

        assert!(output_amount <= input_amount);

        let (input_notes, output_amounts) = self
            .rebalance(dbtx, &self.cfg.fee_consensus, input_amount - output_amount)
            .await;

        for note in &input_notes {
            self.remove_spendable_note(dbtx, note).await;
        }

        input_amount += input_notes.iter().map(SpendableNote::amount).sum();

        output_amount += input_notes
            .iter()
            .map(|note| self.cfg.fee_consensus.fee(note.amount()))
            .sum();

        output_amount += output_amounts
            .iter()
            .map(|denomination| {
                denomination.amount() + self.cfg.fee_consensus.fee(denomination.amount())
            })
            .sum();

        assert!(output_amount <= input_amount);

        let mut spendable_notes = funding_notes
            .into_iter()
            .chain(input_notes)
            .collect::<Vec<SpendableNote>>();

        // We sort the notes by denomination to minimize the leaked information.
        spendable_notes.sort_by_key(|note| note.denomination);

        let input_bundle =
            Self::create_input_bundle(operation_id, spendable_notes, false, self.cfg.amount_unit);

        let mut denominations = represent_amount_with_fees(
            input_amount.saturating_sub(output_amount),
            &self.cfg.fee_consensus,
        )
        .into_iter()
        .chain(output_amounts)
        .collect::<Vec<Denomination>>();

        // We sort the amounts to minimize the leaked information.
        denominations.sort();

        let output_bundle = self.create_output_bundle(operation_id, denominations).await;

        let sender = self.balance_update_sender.clone();
        dbtx.on_commit(move || sender.send_replace(()));

        Ok((input_bundle, output_bundle))
    }

    async fn await_primary_module_output(
        &self,
        operation_id: OperationId,
        outpoint: OutPoint,
    ) -> anyhow::Result<()> {
        self.await_output_sm_success(operation_id, outpoint).await
    }

    async fn get_balance(&self, dbtx: &mut DatabaseTransaction<'_>, unit: AmountUnit) -> Amount {
        if unit != self.cfg.amount_unit {
            return Amount::ZERO;
        }

        self.get_count_by_denomination_dbtx(dbtx)
            .await
            .into_iter()
            .map(|(denomination, count)| denomination.amount().mul_u64(count))
            .sum()
    }

    async fn subscribe_balance_changes(&self) -> BoxStream<'static, ()> {
        Box::pin(tokio_stream::wrappers::WatchStream::new(
            self.balance_update_sender.subscribe(),
        ))
    }
}

impl MintClientModule {
    async fn select_funding_input(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        mut excess_output: Amount,
    ) -> Option<Vec<SpendableNote>> {
        let mut selected_notes = Vec::new();
        let mut target_notes = Vec::new();
        let mut excess_notes = Vec::new();

        for amount in client_denominations().rev() {
            let notes_amount = dbtx
                .find_by_prefix(&SpendableNoteAmountPrefix(amount))
                .await
                .map(|entry| entry.0.0)
                .collect::<Vec<SpendableNote>>()
                .await;

            target_notes.extend(notes_amount.iter().take(TARGET_PER_DENOMINATION).cloned());

            if notes_amount.len() > 2 * TARGET_PER_DENOMINATION {
                for note in notes_amount.into_iter().skip(TARGET_PER_DENOMINATION) {
                    let note_fee = self.cfg.fee_consensus.fee(note.amount());

                    let note_value = note
                        .amount()
                        .checked_sub(note_fee)
                        .expect("All our notes are economical");

                    excess_output = excess_output.saturating_sub(note_value);

                    selected_notes.push(note);
                }
            } else {
                excess_notes.extend(notes_amount.into_iter().skip(TARGET_PER_DENOMINATION));
            }
        }

        if excess_output == Amount::ZERO {
            return Some(selected_notes);
        }

        for note in excess_notes.into_iter().chain(target_notes) {
            let note_amount = note.amount();
            let note_value = note_amount
                .checked_sub(self.cfg.fee_consensus.fee(note_amount))
                .expect("All our notes are economical");

            excess_output = excess_output.saturating_sub(note_value);

            selected_notes.push(note);

            if excess_output == Amount::ZERO {
                return Some(selected_notes);
            }
        }

        None
    }

    async fn rebalance(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        fee: &FeeConsensus,
        mut excess_input: Amount,
    ) -> (Vec<SpendableNote>, Vec<Denomination>) {
        let n_denominations = self.get_count_by_denomination_dbtx(dbtx).await;

        let mut notes = dbtx
            .find_by_prefix_sorted_descending(&SpendableNotePrefix)
            .await
            .map(|entry| entry.0.0)
            .fuse();

        let mut input_notes = Vec::new();
        let mut output_denominations = Vec::new();

        for d in client_denominations() {
            let n_denomination = n_denominations.get(&d).copied().unwrap_or(0);

            let n_missing = TARGET_PER_DENOMINATION.saturating_sub(n_denomination as usize);

            for _ in 0..n_missing {
                match excess_input.checked_sub(d.amount() + fee.fee(d.amount())) {
                    Some(remaining_excess) => excess_input = remaining_excess,
                    None => match notes.next().await {
                        Some(note) => {
                            if note.amount() <= d.amount() + fee.fee(d.amount()) {
                                break;
                            }

                            excess_input += note.amount() - (d.amount() + fee.fee(d.amount()));

                            input_notes.push(note);
                        }
                        None => break,
                    },
                }

                output_denominations.push(d);
            }
        }

        (input_notes, output_denominations)
    }

    fn create_input_bundle(
        operation_id: OperationId,
        notes: Vec<SpendableNote>,
        include_receive_sm: bool,
        amount_unit: AmountUnit,
    ) -> ClientInputBundle<MintInput, MintClientStateMachines> {
        let inputs = notes
            .iter()
            .map(|spendable_note| ClientInput {
                input: MintInput::new_v0(spendable_note.note()),
                keys: vec![spendable_note.keypair],
                amounts: Amounts::new_custom(amount_unit, spendable_note.amount()),
            })
            .collect();

        let input_sms = vec![ClientInputSM {
            state_machines: Arc::new(move |range: OutPointRange| {
                let mut sms = vec![MintClientStateMachines::Input(InputStateMachine {
                    common: InputSMCommon {
                        operation_id,
                        txid: range.txid(),
                        spendable_notes: notes.clone(),
                    },
                    state: InputSMState::Pending,
                })];

                if include_receive_sm {
                    sms.push(MintClientStateMachines::Receive(ReceiveStateMachine {
                        common: crate::receive::ReceiveSMCommon {
                            operation_id,
                            txid: range.txid(),
                        },
                        state: crate::receive::ReceiveSMState::Pending,
                    }));
                }

                sms
            }),
        }];

        ClientInputBundle::new(inputs, input_sms)
    }

    /// Build the blinded outputs and their output state machines for a
    /// reissue transaction.
    ///
    /// # Cancel safety
    ///
    /// Cancel safe: this only reads pre-ground blinding tweaks from an
    /// in-memory channel and constructs values, it performs no database
    /// writes. Dropping the future merely discards the tweaks it had taken,
    /// which are random (so nothing is lost) and replenished by the
    /// background grinder task.
    async fn create_output_bundle(
        &self,
        operation_id: OperationId,
        requested_denominations: Vec<Denomination>,
    ) -> ClientOutputBundle<MintOutput, MintClientStateMachines> {
        let issuance_requests = futures::stream::iter(requested_denominations)
            .zip(self.tweak_receiver.clone())
            .map(|(d, tweak)| NoteIssuanceRequest::new(d, tweak, &self.root_secret))
            .collect::<Vec<NoteIssuanceRequest>>()
            .await;

        let amount_unit = self.cfg.amount_unit;
        let outputs = issuance_requests
            .iter()
            .map(|request| ClientOutput {
                output: request.output(),
                amounts: Amounts::new_custom(amount_unit, request.denomination.amount()),
            })
            .collect();

        let output_sms = vec![ClientOutputSM {
            state_machines: Arc::new(move |range: OutPointRange| {
                vec![MintClientStateMachines::Output(MintOutputStateMachine {
                    common: OutputSMCommon {
                        operation_id,
                        range: Some(range),
                        issuance_requests: issuance_requests.clone(),
                    },
                    state: OutputSMState::Pending,
                })]
            }),
        }];

        ClientOutputBundle::new(outputs, output_sms)
    }

    /// Wait for the output state machine of the given outpoint to reach a
    /// terminal state.
    ///
    /// # Cancel safety
    ///
    /// Cancel safe: this only subscribes to the operation's state
    /// notifications and waits, it performs no database writes. Dropping the
    /// future stops the wait; the underlying state machine keeps running in
    /// the executor.
    async fn await_output_sm_success(
        &self,
        operation_id: OperationId,
        outpoint: OutPoint,
    ) -> anyhow::Result<()> {
        let stream = self
            .notifier
            .subscribe(operation_id)
            .await
            .filter_map(|state| async {
                let MintClientStateMachines::Output(state) = state else {
                    return None;
                };

                if !state.common.range?.into_iter().contains(&outpoint) {
                    return None;
                }

                match state.state {
                    OutputSMState::Pending => None,
                    OutputSMState::Success => Some(Ok(())),
                    OutputSMState::Aborted => Some(Err(anyhow!("Transaction was rejected"))),
                    OutputSMState::Failure => Some(Err(anyhow!("Failed to finalize notes",))),
                }
            });

        pin_mut!(stream);

        stream.next_or_pending().await
    }

    /// Count the `ECash` notes in the client's database by denomination.
    pub async fn get_count_by_denomination(&self) -> BTreeMap<Denomination, u64> {
        self.get_count_by_denomination_dbtx(
            &mut self.client_ctx.module_db().begin_transaction_nc().await,
        )
        .await
    }

    async fn get_count_by_denomination_dbtx(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
    ) -> BTreeMap<Denomination, u64> {
        dbtx.find_by_prefix(&SpendableNotePrefix)
            .await
            .fold(BTreeMap::new(), |mut acc, entry| async move {
                acc.entry(entry.0.0.denomination)
                    .and_modify(|count| *count += 1)
                    .or_insert(1);

                acc
            })
            .await
    }

    /// Send `ECash` for the given amount and return the send operation ID. The
    /// amount will be rounded up to a multiple of 512 msats which is the
    /// smallest denomination used throughout the client. If the rounded
    /// amount cannot be covered with the ecash notes in the client's
    /// database the client will create a transaction to reissue the
    /// required denominations. To cancel a successful ecash send simply
    /// receive it yourself.
    ///
    /// If `include_invite` is set, the federation's invite code is embedded in
    /// the returned ecash so a recipient that has not joined the federation can
    /// do so directly from the received ecash.
    ///
    /// # Cancel safety
    ///
    /// This method is cancel safe. Every database mutation it makes happens
    /// inside a single
    /// [`autocommit`](fedimint_core::db::Database::autocommit) transaction
    /// (spending the notes for the fast path, and creating the change-making
    /// reissue transaction otherwise), so dropping the future either commits
    /// a unit of work in full or leaves the database untouched: notes are
    /// never removed without a persisted operation that produces the ecash or
    /// returns the change. Federation submission is driven by a state machine
    /// in the executor rather than awaited here, so a cancelled call cannot
    /// abort an in-flight reissue; if change-making had already been
    /// submitted it completes on its own and the notes return to the balance.
    pub async fn send(
        &self,
        amount: Amount,
        custom_meta: Value,
        include_invite: bool,
    ) -> Result<(OperationId, ECash), SendECashError> {
        let amount = round_to_multiple(amount, client_denominations().next().unwrap().amount());

        if let Some((operation_id, ecash)) = self
            .client_ctx
            .module_db()
            .autocommit(
                |dbtx, _| {
                    Box::pin(self.send_ecash_dbtx(
                        dbtx,
                        amount,
                        custom_meta.clone(),
                        include_invite,
                    ))
                },
                Some(100),
            )
            .await
            .expect("Failed to commit dbtx after 100 retries")
        {
            return Ok((operation_id, ecash));
        }

        self.client_ctx
            .global_api()
            .session_count()
            .await
            .map_err(|_| SendECashError::Offline)?;

        let operation_id = OperationId::new_random();

        let output = self
            .create_output_bundle(operation_id, represent_amount(amount))
            .await;
        let output = self.client_ctx.make_client_outputs(output);
        let cm = custom_meta.clone();

        let range = self
            .client_ctx
            .finalize_and_submit_transaction(
                operation_id,
                MintCommonInit::KIND.as_str(),
                move |change_outpoint_range| MintOperationMeta::Reissue {
                    change_outpoint_range,
                    amount,
                    custom_meta: cm.clone(),
                },
                TransactionBuilder::new().with_outputs(output),
            )
            .await
            .map_err(|error| match error {
                TransactionSubmitError::InsufficientFunds(_) => SendECashError::InsufficientBalance,
                other => SendECashError::Failed(other),
            })?;

        for outpoint in range {
            self.await_output_sm_success(operation_id, outpoint)
                .await
                .map_err(|_| SendECashError::Failure)?;
        }

        Box::pin(self.send(amount, custom_meta, include_invite)).await
    }

    /// Fast path for [`Self::send`]: if exact change is already held, spend
    /// those notes and record the send operation. Returns `None` when exact
    /// change is not available, leaving the caller to make change.
    ///
    /// # Cancel safety
    ///
    /// All writes go to the passed `dbtx`, so this inherits the cancel safety
    /// of that transaction: it must be run inside an `autocommit` (as `send`
    /// does) for the note spend and the operation log entry to commit or roll
    /// back together.
    async fn send_ecash_dbtx(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        remaining_amount: Amount,
        custom_meta: Value,
        include_invite: bool,
    ) -> Result<Option<(OperationId, ECash)>, Infallible> {
        let Some(notes) = Self::select_exact_change(&mut dbtx.to_ref_nc(), remaining_amount).await
        else {
            return Ok(None);
        };

        for spendable_note in &notes {
            self.remove_spendable_note(dbtx, spendable_note).await;
        }

        let ecash = if include_invite {
            let invite = self.client_ctx.get_invite_code().await;
            ECash::new_with_invite(notes, &invite)
        } else {
            ECash::new(self.federation_id, notes)
        }
        .with_unit(self.cfg.amount_unit);
        let amount = ecash.amount();
        let operation_id = OperationId::new_random();

        self.client_ctx
            .add_operation_log_entry_dbtx(
                dbtx,
                operation_id,
                MintCommonInit::KIND.as_str(),
                MintOperationMeta::Send {
                    ecash: base32::encode_prefixed(FEDIMINT_PREFIX, &ecash),
                    custom_meta,
                },
            )
            .await;

        self.client_ctx
            .log_event(
                dbtx,
                SendPaymentEvent {
                    operation_id,
                    amount,
                    ecash: base32::encode_prefixed(FEDIMINT_PREFIX, &ecash),
                },
            )
            .await;

        let sender = self.balance_update_sender.clone();
        dbtx.on_commit(move || sender.send_replace(()));

        Ok(Some((operation_id, ecash)))
    }

    /// Receive the `ECash` by reissuing the notes and return the operation ID.
    pub async fn receive(
        &self,
        ecash: ECash,
        custom_meta: Value,
    ) -> Result<OperationId, ReceiveECashError> {
        let operation_id = OperationId::from_encodable(&ecash);

        if ecash.mint() != Some(self.federation_id) {
            return Err(ReceiveECashError::WrongFederation);
        }

        if ecash
            .notes()
            .iter()
            .any(|note| note.amount() <= self.cfg.fee_consensus.base_fee())
        {
            return Err(ReceiveECashError::UneconomicalDenomination);
        }

        let input =
            Self::create_input_bundle(operation_id, ecash.notes(), true, self.cfg.amount_unit);
        let input = self.client_ctx.make_client_inputs(input);
        let ec = base32::encode_prefixed(FEDIMINT_PREFIX, &ecash);

        self.client_ctx
            .finalize_and_submit_transaction(
                operation_id,
                MintCommonInit::KIND.as_str(),
                move |change_outpoint_range| MintOperationMeta::Receive {
                    change_outpoint_range,
                    ecash: ec.clone(),
                    custom_meta: custom_meta.clone(),
                },
                TransactionBuilder::new().with_inputs(input),
            )
            .await
            .map_err(|error| match error {
                TransactionSubmitError::OperationAlreadyExists(_) => {
                    ReceiveECashError::AlreadyReceived
                }
                TransactionSubmitError::InsufficientFunds(_) => {
                    ReceiveECashError::InsufficientFunds
                }
                other => ReceiveECashError::Failed(other),
            })?;

        let mut dbtx = self.client_ctx.module_db().begin_transaction().await;

        self.client_ctx
            .log_event(
                &mut dbtx,
                ReceivePaymentEvent {
                    operation_id,
                    amount: ecash.amount(),
                },
            )
            .await;

        dbtx.commit_tx().await;

        Ok(operation_id)
    }

    /// Computes the exact fee a `receive(ecash)` would incur given the client's
    /// current note inventory, without submitting anything.
    ///
    /// This runs the same change generation the real receive does
    /// (`create_final_inputs_and_outputs`, including funding selection and
    /// rebalancing) against a non-committable transaction that is dropped
    /// rather than committed, so the client's notes are read but left
    /// untouched. The quote is point-in-time: it depends on the current
    /// inventory and can move as notes change.
    pub async fn receive_fee_quote(
        &self,
        ecash: &ECash,
    ) -> Result<FeeQuote, TransactionSubmitError> {
        // A receive submits the ecash notes as explicit inputs and no explicit
        // outputs; the shared, module-agnostic fee quote runs the primary-module
        // balancing (rebalancing + minting change) over the real inventory.
        let notes = ecash.notes();
        let input_amount: Amount = notes.iter().map(SpendableNote::amount).sum();
        let input_fee: Amount = notes
            .iter()
            .map(|note| self.cfg.fee_consensus.fee(note.amount()))
            .sum();

        self.client_ctx
            .fee_quote(
                OperationId::new_random(),
                FeeQuoteRequest {
                    input_amount: Amounts::new_custom(self.cfg.amount_unit, input_amount),
                    output_amount: Amounts::ZERO,
                    input_fee: Amounts::new_custom(self.cfg.amount_unit, input_fee),
                    output_fee: Amounts::ZERO,
                },
            )
            .await
    }

    /// Computes the fee a `send(amount)` would incur given the client's current
    /// note inventory, without sending anything.
    ///
    /// A send is free when the client's existing notes can cover the (rounded)
    /// amount exactly — it just hands those notes out. Otherwise the send first
    /// reissues itself the right denominations, and that self-reissue
    /// transaction is the only thing a send ever pays a fee for. This quote
    /// mirrors that: it returns [`FeeQuote::ZERO`] when exact change is
    /// available, and otherwise quotes the reissue the same way the real send
    /// submits it (explicit outputs `represent_amount(amount)`, no explicit
    /// inputs) via the shared, module-agnostic fee quote over the real
    /// inventory. The quote is point-in-time: it depends on the current
    /// inventory and can move as notes change.
    pub async fn send_fee_quote(&self, amount: Amount) -> Result<FeeQuote, TransactionSubmitError> {
        let amount = round_to_multiple(amount, client_denominations().next().unwrap().amount());

        // Exact-change path: handing out existing notes never costs a fee.
        if self.can_make_exact_change(amount).await {
            return Ok(FeeQuote::ZERO);
        }

        // Reissue path: the send mints itself `represent_amount(amount)` as
        // explicit outputs (no explicit inputs) and the primary module funds and
        // balances it. Quote that exact transaction.
        let denominations = represent_amount(amount);
        let output_amount: Amount = denominations.iter().map(|d| d.amount()).sum();
        let output_fee: Amount = denominations
            .iter()
            .map(|d| self.cfg.fee_consensus.fee(d.amount()))
            .sum();

        self.client_ctx
            .fee_quote(
                OperationId::new_random(),
                FeeQuoteRequest {
                    input_amount: Amounts::ZERO,
                    output_amount: Amounts::new_custom(self.cfg.amount_unit, output_amount),
                    input_fee: Amounts::ZERO,
                    output_fee: Amounts::new_custom(self.cfg.amount_unit, output_fee),
                },
            )
            .await
    }

    /// Returns whether the client's current notes can be handed out to cover
    /// `amount` exactly — the free path in [`Self::send`] — without modifying
    /// the inventory. Shares its greedy selection with
    /// [`Self::send_ecash_dbtx`] via [`Self::select_exact_change`].
    async fn can_make_exact_change(&self, remaining_amount: Amount) -> bool {
        let mut dbtx = self.client_ctx.module_db().begin_transaction_nc().await;

        Self::select_exact_change(&mut dbtx, remaining_amount)
            .await
            .is_some()
    }

    /// Greedily selects spendable notes (largest-first) summing to *exactly*
    /// `remaining_amount`, or `None` if the inventory can't make exact change.
    /// Reads the DB but does not modify it. Single source of truth for the free
    /// (hand-out-existing-notes) selection shared by [`Self::send_ecash_dbtx`]
    /// and [`Self::can_make_exact_change`].
    async fn select_exact_change(
        dbtx: &mut DatabaseTransaction<'_>,
        mut remaining_amount: Amount,
    ) -> Option<Vec<SpendableNote>> {
        let mut stream = dbtx
            .find_by_prefix_sorted_descending(&SpendableNotePrefix)
            .await
            .map(|entry| entry.0.0);

        let mut notes = vec![];

        while let Some(spendable_note) = stream.next().await {
            remaining_amount = match remaining_amount.checked_sub(spendable_note.amount()) {
                Some(amount) => amount,
                None => continue,
            };

            notes.push(spendable_note);

            if remaining_amount == Amount::ZERO {
                break;
            }
        }

        (remaining_amount == Amount::ZERO).then_some(notes)
    }

    /// Await the final state of the receive operation.
    pub async fn await_final_receive_operation_state(
        &self,
        operation_id: OperationId,
    ) -> Result<FinalReceiveOperationState, OperationLookupError> {
        let operation = self.client_ctx.get_operation(operation_id).await?;
        let mut stream = self.notifier.subscribe(operation_id).await;

        let mut stream = self
            .client_ctx
            .outcome_or_updates(&operation, operation_id, |_| true, move || {
                async_stream::stream! {
                    loop {
                        if let Some(MintClientStateMachines::Receive(state)) = stream.next().await {
                            match state.state {
                                ReceiveSMState::Pending => {}
                                ReceiveSMState::Success => {
                                    yield FinalReceiveOperationState::Success;
                                    return;
                                }
                                ReceiveSMState::Rejected(..) => {
                                    yield FinalReceiveOperationState::Rejected;
                                    return;
                                }
                            }
                        }
                    }
                }
            })
            .into_stream();

        let mut final_state = None;

        while let Some(state) = stream.next().await {
            final_state = Some(state);
        }

        Ok(final_state.expect("Stream contains one final state"))
    }

    async fn remove_spendable_note(
        &self,
        dbtx: &mut DatabaseTransaction<'_>,
        spendable_note: &SpendableNote,
    ) {
        dbtx.remove_entry(&SpendableNoteKey(spendable_note.clone()))
            .await
            .expect("Must delete existing spendable note");
    }
}

/// Hands out guardians so that only one slice request is outstanding to each.
///
/// Whichever peer finishes first takes the next slice, so a slow guardian
/// receives less work without anyone having to measure how slow it is, and one
/// that is not answering ties up a single request rather than a share of all
/// of them.
#[derive(Clone)]
struct PeerPool {
    receiver: async_channel::Receiver<PeerId>,
    sender: async_channel::Sender<PeerId>,
}

impl PeerPool {
    fn new(peers: &BTreeSet<PeerId>) -> Self {
        let (sender, receiver) = async_channel::bounded(peers.len().max(1));

        for peer in peers {
            sender
                .try_send(*peer)
                .expect("Capacity was sized to hold every peer");
        }

        Self { receiver, sender }
    }

    /// Wait for a guardian with no request outstanding.
    async fn acquire(&self) -> PeerId {
        self.receiver
            .recv()
            .await
            .expect("The sender is held for as long as the receiver")
    }

    /// Take a guardian out of rotation, putting it back once it has sat out
    /// [`PEER_READMISSION`].
    ///
    /// Dropping it for good would cost a guardian that timed out once the rest
    /// of the recovery, which on a federation of four is a quarter of the
    /// capacity thrown away for a single bad request.
    fn retire(&self, peer: PeerId) {
        let pool = self.clone();

        fedimint_core::runtime::spawn("mintv2 recovery peer readmission", async move {
            fedimint_core::runtime::sleep(PEER_READMISSION).await;

            pool.release(peer);
        });
    }

    /// Put a guardian back for the next slice.
    fn release(&self, peer: PeerId) {
        self.sender
            .try_send(peer)
            .expect("Only peers taken from the pool are put back");
    }
}

/// Download a slice, asking one guardian at a time and holding it for the
/// duration of the request.
async fn download_slice(
    module_api: DynModuleApi,
    peers: PeerPool,
    start: u64,
    end: u64,
    expected_hash: sha256::Hash,
) -> Vec<RecoveryItem> {
    let mut timeouts = custom_backoff(SLICE_TIMEOUT, MAX_SLICE_TIMEOUT, None);

    loop {
        let peer = peers.acquire().await;

        let timeout = timeouts.next().expect("The backoff never gives up");

        let result = module_api
            .fetch_recovery_slice(peer, timeout, start, end)
            .await;

        match result {
            Ok(data) if data.consensus_hash::<sha256::Hash>() == expected_hash => {
                peers.release(peer);

                return data;
            }
            // Either served something the other guardians disagree with, or
            // did not answer at all. Either way it sits out for a while: a
            // request that reaches the timeout took many times longer than a
            // healthy guardian needs, so asking it again mostly buys another
            // timeout. The timeout grows in case it is the client's own
            // connection that is too slow for the initial one.
            Ok(_) | Err(_) => peers.retire(peer),
        }
    }
}

/// A failure to send e-cash by preparing notes to hand to the recipient.
#[derive(Error, Debug)]
#[non_exhaustive]
pub enum SendECashError {
    /// The client needs to reissue notes to make change, but has no
    /// connection to the federation to do so.
    #[error("We need to reissue notes but the client is offline")]
    Offline,
    /// The client's balance cannot cover the amount requested.
    #[error("The clients balance is insufficient")]
    InsufficientBalance,
    /// The client failed to prepare the notes for a reason it cannot
    /// recover from.
    #[error("A non-recoverable error has occurred")]
    Failure,
    /// The change-making reissue transaction could not be submitted for a
    /// reason unrelated to funding.
    #[error("The reissue transaction could not be submitted")]
    Failed(#[source] TransactionSubmitError),
}

/// A failure to receive e-cash by reissuing it.
#[derive(Error, Debug)]
#[non_exhaustive]
pub enum ReceiveECashError {
    /// The e-cash was issued by a different federation.
    #[error("The ECash is from a different federation")]
    WrongFederation,

    /// One of the notes is worth no more than the fee to reissue it.
    #[error("ECash contains an uneconomical denomination")]
    UneconomicalDenomination,

    /// The client cannot cover the fee the reissue costs.
    #[error("Receiving ecash requires additional funds")]
    InsufficientFunds,

    /// An operation for this exact e-cash already exists, so it was already
    /// received.
    #[error("The ECash was already received")]
    AlreadyReceived,

    /// The reissue transaction could not be submitted for a reason unrelated
    /// to funding.
    #[error("The reissue transaction could not be submitted")]
    Failed(#[source] TransactionSubmitError),
}

#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
pub enum FinalReceiveOperationState {
    // The ecash notes have been reissued
    Success,
    // The ecash notes were already spent
    Rejected,
}

#[derive(Debug, Clone, Eq, PartialEq, Hash, Decodable, Encodable)]
pub enum MintClientStateMachines {
    Input(InputStateMachine),
    Output(MintOutputStateMachine),
    Receive(ReceiveStateMachine),
}

impl IntoDynInstance for MintClientStateMachines {
    type DynType = DynState;

    fn into_dyn(self, instance_id: ModuleInstanceId) -> Self::DynType {
        DynState::from_typed(instance_id, self)
    }
}

impl State for MintClientStateMachines {
    type ModuleContext = MintClientContext;

    fn transitions(
        &self,
        context: &Self::ModuleContext,
        global_context: &DynGlobalClientContext,
    ) -> Vec<StateTransition<Self>> {
        match self {
            MintClientStateMachines::Input(redemption_state) => {
                sm_enum_variant_translation!(
                    redemption_state.transitions(context, global_context),
                    MintClientStateMachines::Input
                )
            }
            MintClientStateMachines::Output(issuance_state) => {
                sm_enum_variant_translation!(
                    issuance_state.transitions(context, global_context),
                    MintClientStateMachines::Output
                )
            }
            MintClientStateMachines::Receive(receive_state) => {
                sm_enum_variant_translation!(
                    receive_state.transitions(context, global_context),
                    MintClientStateMachines::Receive
                )
            }
        }
    }

    fn operation_id(&self) -> OperationId {
        match self {
            MintClientStateMachines::Input(redemption_state) => redemption_state.operation_id(),
            MintClientStateMachines::Output(issuance_state) => issuance_state.operation_id(),
            MintClientStateMachines::Receive(receive_state) => receive_state.operation_id(),
        }
    }
}

fn round_to_multiple(amount: Amount, min_denomiation: Amount) -> Amount {
    Amount::from_msats(amount.msats.next_multiple_of(min_denomiation.msats))
}

fn represent_amount_with_fees(
    mut remaining_amount: Amount,
    fee_consensus: &FeeConsensus,
) -> Vec<Denomination> {
    let mut denominations = Vec::new();

    // Add denominations with a greedy algorithm
    for denomination in client_denominations().rev() {
        let n_add =
            remaining_amount / (denomination.amount() + fee_consensus.fee(denomination.amount()));

        denominations.extend(std::iter::repeat_n(denomination, n_add as usize));

        remaining_amount -=
            n_add * (denomination.amount() + fee_consensus.fee(denomination.amount()));
    }

    // We sort the notes by amount to minimize the leaked information.
    denominations.sort();

    denominations
}

fn represent_amount(mut remaining_amount: Amount) -> Vec<Denomination> {
    let mut denominations = Vec::new();

    // Add denominations with a greedy algorithm
    for denomination in client_denominations().rev() {
        let n_add = remaining_amount / denomination.amount();

        denominations.extend(std::iter::repeat_n(denomination, n_add as usize));

        remaining_amount -= n_add * denomination.amount();
    }

    denominations
}