Skip to main content

fedimint_mintv2_client/
lib.rs

1#![deny(clippy::pedantic)]
2#![allow(clippy::cast_possible_truncation)]
3#![allow(clippy::missing_errors_doc)]
4#![allow(clippy::missing_panics_doc)]
5#![allow(clippy::module_name_repetitions)]
6#![allow(clippy::must_use_candidate)]
7#![allow(clippy::return_self_not_must_use)]
8#![allow(clippy::too_many_lines)]
9
10pub use fedimint_mintv2_common as common;
11
12mod api;
13#[cfg(feature = "cli")]
14mod cli;
15pub mod client_db;
16mod ecash;
17mod events;
18mod input;
19pub mod issuance;
20mod output;
21mod receive;
22#[cfg(test)]
23mod tests;
24
25use std::collections::{BTreeMap, BTreeSet};
26use std::convert::Infallible;
27use std::sync::Arc;
28use std::time::Duration;
29
30use bitcoin_hashes::sha256;
31use client_db::{RecoveryState, RecoveryStateKey, SpendableNoteAmountPrefix, SpendableNotePrefix};
32pub use events::*;
33use fedimint_api_client::api::DynModuleApi;
34use fedimint_client::module::ClientModule;
35use fedimint_client::transaction::{
36    ClientInput, ClientInputBundle, ClientInputSM, ClientOutput, ClientOutputBundle,
37    ClientOutputSM, FeeQuote, FeeQuoteRequest, TransactionBuilder,
38};
39use fedimint_client_module::db::ClientModuleMigrationFn;
40use fedimint_client_module::error::{
41    ClientModuleError, InsufficientBalanceError, OperationLookupError, TransactionSubmitError,
42};
43use fedimint_client_module::module::init::{
44    ClientModuleInit, ClientModuleInitArgs, ClientModuleRecoverArgs,
45    ClientModuleRecoveryPrepareArgs, RecoveryMode,
46};
47use fedimint_client_module::module::recovery::{NoModuleBackup, RecoveryProgress};
48use fedimint_client_module::module::{
49    ClientContext, OutPointRange, PrimaryModulePriority, PrimaryModuleSupport,
50};
51use fedimint_client_module::sm::{Context, DynState, ModuleNotifier, State, StateTransition};
52use fedimint_client_module::{DynGlobalClientContext, sm_enum_variant_translation};
53use fedimint_core::base32::{self, FEDIMINT_PREFIX};
54use fedimint_core::config::FederationId;
55use fedimint_core::core::{IntoDynInstance, ModuleInstanceId, ModuleKind, OperationId};
56use fedimint_core::db::{DatabaseTransaction, DatabaseVersion, IDatabaseTransactionOpsCoreTyped};
57use fedimint_core::encoding::{Decodable, DecodeError, Encodable};
58use fedimint_core::module::registry::ModuleDecoderRegistry;
59use fedimint_core::module::{
60    AmountUnit, Amounts, ApiVersion, CommonModuleInit, ModuleCommon, ModuleInit, MultiApiVersion,
61};
62use fedimint_core::secp256k1::rand::{Rng, thread_rng};
63use fedimint_core::secp256k1::{Keypair, PublicKey};
64use fedimint_core::util::backoff_util::custom_backoff;
65use fedimint_core::util::{BoxStream, NextOrPending};
66use fedimint_core::{Amount, OutPoint, PeerId, apply, async_trait_maybe_send};
67use fedimint_derive_secret::DerivableSecret;
68use fedimint_mintv2_common::config::{FeeConsensus, MintClientConfig, client_denominations};
69use fedimint_mintv2_common::{
70    Denomination, KIND, MintCommonInit, MintInput, MintModuleTypes, MintOutput, Note, RecoveryItem,
71};
72use futures::{StreamExt, pin_mut};
73use itertools::Itertools;
74use serde::{Deserialize, Serialize};
75use serde_json::Value;
76use tbs::AggregatePublicKey;
77use thiserror::Error;
78
79use crate::api::MintV2ModuleApi;
80use crate::client_db::SpendableNoteKey;
81pub use crate::ecash::ECash;
82use crate::input::{InputSMCommon, InputSMState, InputStateMachine};
83use crate::issuance::NoteIssuanceRequest;
84use crate::output::{MintOutputStateMachine, OutputSMCommon, OutputSMState};
85use crate::receive::{ReceiveSMState, ReceiveStateMachine};
86
87const TARGET_PER_DENOMINATION: usize = 3;
88const SLICE_SIZE: u64 = 10000;
89/// How long a guardian sits out after failing to answer a slice request.
90const PEER_READMISSION: Duration = Duration::from_secs(60);
91/// How long a single peer is given to answer a slice request.
92///
93/// A slice takes a second or two from a healthy guardian, so waiting half a
94/// minute only delays noticing that one is not going to answer. The timeout
95/// grows on every failed attempt up to [`MAX_SLICE_TIMEOUT`], so a client
96/// on a slow connection where every guardian exceeds the initial timeout
97/// still makes progress instead of retrying forever.
98const SLICE_TIMEOUT: Duration = Duration::from_secs(10);
99/// Upper bound for the per-retry growth of [`SLICE_TIMEOUT`]; the flat
100/// timeout used before the growth was introduced.
101const MAX_SLICE_TIMEOUT: Duration = Duration::from_secs(30);
102const PARALLEL_HASH_REQUESTS: usize = 10;
103const PARALLEL_SLICE_REQUESTS: usize = 10;
104
105#[derive(Debug, Clone, PartialEq, Eq, Hash, Encodable, Decodable)]
106pub struct SpendableNote {
107    pub denomination: Denomination,
108    pub keypair: Keypair,
109    pub signature: tbs::Signature,
110}
111
112impl SpendableNote {
113    pub fn amount(&self) -> Amount {
114        self.denomination.amount()
115    }
116}
117
118impl SpendableNote {
119    fn nonce(&self) -> PublicKey {
120        self.keypair.public_key()
121    }
122
123    fn note(&self) -> Note {
124        Note {
125            denomination: self.denomination,
126            nonce: self.nonce(),
127            signature: self.signature,
128        }
129    }
130}
131
132/// A [`SpendableNote`] whose signature is not decoded yet, with the same
133/// encoding.
134///
135/// Decoding a [`tbs::Signature`] decompresses and subgroup-checks a curve
136/// point, which is expensive, and every spend reads far more notes than it
137/// spends. Reading the note table as this type defers that cost to the notes
138/// actually selected. Since not every byte string is a valid signature it is
139/// only for reading the client's own database; [`Self::decode`] validates.
140#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Encodable, Decodable)]
141pub struct SpendableNoteUndecoded {
142    pub denomination: Denomination,
143    pub keypair: Keypair,
144    // Must stay in sync with the encoding of `tbs::Signature`, which a test
145    // verifies.
146    pub signature: [u8; 48],
147}
148
149impl SpendableNoteUndecoded {
150    pub fn amount(&self) -> Amount {
151        self.denomination.amount()
152    }
153
154    pub fn decode(self) -> Result<SpendableNote, DecodeError> {
155        Ok(SpendableNote {
156            denomination: self.denomination,
157            keypair: self.keypair,
158            signature: Decodable::consensus_decode_whole(
159                &self.signature,
160                &ModuleDecoderRegistry::default(),
161            )?,
162        })
163    }
164
165    fn decode_own(self) -> SpendableNote {
166        self.decode()
167            .expect("Notes in our own database were encoded from valid signatures")
168    }
169}
170
171impl From<SpendableNote> for SpendableNoteUndecoded {
172    fn from(note: SpendableNote) -> Self {
173        Self {
174            denomination: note.denomination,
175            keypair: note.keypair,
176            signature: note.signature.0.to_compressed(),
177        }
178    }
179}
180
181#[derive(Debug, Clone, Serialize, Deserialize)]
182pub enum MintOperationMeta {
183    Send {
184        ecash: String,
185        custom_meta: Value,
186    },
187    Reissue {
188        change_outpoint_range: OutPointRange,
189        amount: Amount,
190        custom_meta: Value,
191    },
192    Receive {
193        change_outpoint_range: OutPointRange,
194        ecash: String,
195        custom_meta: Value,
196    },
197}
198
199#[derive(Debug, Clone)]
200pub struct MintClientInit;
201
202impl ModuleInit for MintClientInit {
203    type Common = MintCommonInit;
204
205    async fn dump_database(
206        &self,
207        _dbtx: &mut DatabaseTransaction<'_>,
208        _prefix_names: Vec<String>,
209    ) -> Box<dyn Iterator<Item = (String, Box<dyn erased_serde::Serialize + Send>)> + '_> {
210        Box::new(BTreeMap::new().into_iter())
211    }
212}
213
214#[apply(async_trait_maybe_send!)]
215impl ClientModuleInit for MintClientInit {
216    type Module = MintClientModule;
217
218    fn supported_api_versions(&self) -> MultiApiVersion {
219        MultiApiVersion::try_from_iter([ApiVersion { major: 0, minor: 1 }])
220            .expect("no version conflicts")
221    }
222
223    fn recovery_mode(&self) -> RecoveryMode {
224        RecoveryMode::Usable
225    }
226
227    async fn prepare_recovery(
228        &self,
229        args: &ClientModuleRecoveryPrepareArgs,
230    ) -> Result<(), ClientModuleError> {
231        if args
232            .db()
233            .begin_transaction_nc()
234            .await
235            .get_value(&RecoveryStateKey)
236            .await
237            .is_some()
238        {
239            return Ok(());
240        }
241
242        // The total is the number of items the federation had processed when
243        // the recovery began, so every note this client issues from here on
244        // lands beyond it and is never rediscovered by the scan. Committing
245        // that bound before the module is initialized is what lets the module
246        // be used while its own recovery is still running: the two cannot
247        // arrive at the same note.
248        let state = RecoveryState {
249            next_index: 0,
250            total_items: args
251                .module_api()
252                .fetch_recovery_count()
253                .await
254                .map_err(ClientModuleError::other)?,
255            requests: BTreeMap::new(),
256            nonces: BTreeSet::new(),
257        };
258
259        let mut dbtx = args.db().begin_transaction().await;
260
261        dbtx.insert_entry(&RecoveryStateKey, &state).await;
262
263        dbtx.commit_tx().await;
264
265        Ok(())
266    }
267
268    async fn recover(
269        &self,
270        args: &ClientModuleRecoverArgs<Self>,
271        _snapshot: Option<&NoModuleBackup>,
272    ) -> Result<Option<Amount>, ClientModuleError> {
273        let mut state = args
274            .db()
275            .begin_transaction_nc()
276            .await
277            .get_value(&RecoveryStateKey)
278            .await
279            .expect("Prepare recovery commits the state before the recovery is started");
280
281        if state.next_index == state.total_items {
282            return Ok(None);
283        }
284
285        let peer_pool = PeerPool::new(args.api().all_peers());
286
287        let mut recovery_stream = futures::stream::iter(
288            (state.next_index..state.total_items).step_by(SLICE_SIZE as usize),
289        )
290        .map(|start| {
291            let api = args.module_api().clone();
292            let end = std::cmp::min(start + SLICE_SIZE, state.total_items);
293
294            async move { (start, end, api.fetch_recovery_slice_hash(start, end).await) }
295        })
296        .buffered(PARALLEL_HASH_REQUESTS)
297        .map(|(start, end, hash)| {
298            let module_api = args.module_api().clone();
299            let peer_pool = peer_pool.clone();
300
301            async move {
302                (
303                    start,
304                    download_slice(module_api, peer_pool, start, end, hash).await,
305                )
306            }
307        })
308        // Unordered, so a guardian that goes quiet holds up only its own
309        // slice. Delivering in order would let it halt everything: the other
310        // requests would finish, their results would fill the buffer, and no
311        // new request could start until the straggler returned. Downloading
312        // runs ahead of it instead, and the items are put back in order below.
313        .buffer_unordered(PARALLEL_SLICE_REQUESTS);
314
315        let tweak_filter = issuance::tweak_filter(args.module_root_secret());
316
317        // Slices that arrived before the ones in front of them. An input
318        // spends an output that has to have been seen already, so items are
319        // scanned in index order however they turn up. This holds whatever
320        // downloaded while a slice was outstanding, which the timeout bounds.
321        let mut pending: BTreeMap<u64, Vec<RecoveryItem>> = BTreeMap::new();
322
323        loop {
324            let items = loop {
325                if let Some(items) = pending.remove(&state.next_index) {
326                    break items;
327                }
328
329                let (start, items) = recovery_stream.next().await.ok_or_else(|| {
330                    ClientModuleError::other("Recovery stream finished before recovery is complete")
331                })?;
332
333                pending.insert(start, items);
334            };
335
336            for item in &items {
337                match item {
338                    RecoveryItem::Output {
339                        denomination,
340                        nonce_hash,
341                        tweak,
342                    } => {
343                        if !issuance::check_tweak(*tweak, tweak_filter) {
344                            continue;
345                        }
346                        let output_secret = issuance::output_secret(
347                            *denomination,
348                            *tweak,
349                            args.module_root_secret(),
350                        );
351
352                        if !issuance::check_nonce(&output_secret, *nonce_hash) {
353                            continue;
354                        }
355
356                        let computed_nonce_hash = issuance::nonce(&output_secret).consensus_hash();
357
358                        // Ignore possible duplicate nonces
359                        if !state.nonces.insert(computed_nonce_hash) {
360                            continue;
361                        }
362
363                        state.requests.insert(
364                            computed_nonce_hash,
365                            NoteIssuanceRequest::new(
366                                *denomination,
367                                *tweak,
368                                args.module_root_secret(),
369                            ),
370                        );
371                    }
372                    RecoveryItem::Input { nonce_hash } => {
373                        state.requests.remove(nonce_hash);
374                        state.nonces.remove(nonce_hash);
375                    }
376                }
377            }
378
379            state.next_index += items.len() as u64;
380
381            let mut dbtx = args.db().begin_transaction().await;
382
383            dbtx.insert_entry(&RecoveryStateKey, &state).await;
384
385            if state.next_index == state.total_items {
386                // Total value of the notes reconstructed during recovery
387                let recovered_amount = state
388                    .requests
389                    .values()
390                    .map(|request| request.denomination.amount())
391                    .sum::<Amount>();
392
393                let state_machines = args
394                    .context()
395                    .map_dyn(vec![MintClientStateMachines::Output(
396                        MintOutputStateMachine {
397                            common: OutputSMCommon {
398                                operation_id: OperationId::new_random(),
399                                range: None,
400                                issuance_requests: state.requests.into_values().collect(),
401                            },
402                            state: OutputSMState::Pending,
403                        },
404                    )])
405                    .collect();
406
407                args.context()
408                    .add_state_machines_dbtx(&mut dbtx.to_ref_nc(), state_machines)
409                    .await
410                    .expect("state machine is valid");
411
412                dbtx.commit_tx().await;
413
414                return Ok(Some(recovered_amount));
415            }
416
417            dbtx.commit_tx().await;
418
419            args.update_recovery_progress(RecoveryProgress {
420                complete: state.next_index.try_into().unwrap_or(u32::MAX),
421                total: state.total_items.try_into().unwrap_or(u32::MAX),
422            });
423        }
424    }
425
426    async fn init(
427        &self,
428        args: &ClientModuleInitArgs<Self>,
429    ) -> Result<Self::Module, ClientModuleError> {
430        let (tweak_sender, tweak_receiver) = async_channel::bounded(50);
431
432        let filter = issuance::tweak_filter(args.module_root_secret());
433
434        // Only ~1/65536 random tweaks pass the filter, so grinding is
435        // CPU-bound. Pre-grind in the background and feed tweaks through a
436        // channel to keep the cost off the spend path. `send` only suspends
437        // once the channel is full, so we yield explicitly on each found tweak
438        // to keep single-threaded (wasm) runtimes responsive while the channel
439        // still has space.
440        fedimint_core::task::spawn("mintv2-tweak-grinder", async move {
441            loop {
442                let tweak: [u8; 16] = thread_rng().r#gen();
443
444                if !issuance::check_tweak(tweak, filter) {
445                    continue;
446                }
447
448                if tweak_sender.send(tweak).await.is_err() {
449                    return;
450                }
451
452                fedimint_core::task::sleep(Duration::ZERO).await;
453            }
454        });
455
456        Ok(MintClientModule {
457            federation_id: *args.federation_id(),
458            cfg: args.cfg().clone(),
459            root_secret: args.module_root_secret().clone(),
460            notifier: args.notifier().clone(),
461            client_ctx: args.context(),
462            balance_update_sender: tokio::sync::watch::channel(()).0,
463            tweak_receiver,
464        })
465    }
466
467    fn get_database_migrations(&self) -> BTreeMap<DatabaseVersion, ClientModuleMigrationFn> {
468        BTreeMap::new()
469    }
470}
471
472#[derive(Debug)]
473pub struct MintClientModule {
474    federation_id: FederationId,
475    cfg: MintClientConfig,
476    root_secret: DerivableSecret,
477    notifier: ModuleNotifier<MintClientStateMachines>,
478    client_ctx: ClientContext<Self>,
479    balance_update_sender: tokio::sync::watch::Sender<()>,
480    tweak_receiver: async_channel::Receiver<[u8; 16]>,
481}
482
483#[derive(Debug, Clone)]
484pub struct MintClientContext {
485    client_ctx: ClientContext<MintClientModule>,
486    tbs_agg_pks: BTreeMap<Denomination, AggregatePublicKey>,
487    tbs_pks: BTreeMap<Denomination, BTreeMap<PeerId, tbs::PublicKeyShare>>,
488    pub balance_update_sender: tokio::sync::watch::Sender<()>,
489}
490
491impl Context for MintClientContext {
492    const KIND: Option<ModuleKind> = Some(KIND);
493}
494
495#[apply(async_trait_maybe_send!)]
496impl ClientModule for MintClientModule {
497    type Init = MintClientInit;
498    type Common = MintModuleTypes;
499    type Backup = NoModuleBackup;
500    type ModuleStateMachineContext = MintClientContext;
501    type States = MintClientStateMachines;
502
503    fn context(&self) -> Self::ModuleStateMachineContext {
504        MintClientContext {
505            client_ctx: self.client_ctx.clone(),
506            tbs_agg_pks: self.cfg.tbs_agg_pks.clone(),
507            tbs_pks: self.cfg.tbs_pks.clone(),
508            balance_update_sender: self.balance_update_sender.clone(),
509        }
510    }
511
512    fn input_fee(
513        &self,
514        amounts: &Amounts,
515        _input: &<Self::Common as ModuleCommon>::Input,
516    ) -> Option<Amounts> {
517        let unit = self.cfg.amount_unit;
518        let amount = amounts.get(&unit).copied().unwrap_or_default();
519        let fee = self.cfg.fee_consensus.fee(amount);
520
521        Some(Amounts::new_custom(unit, fee))
522    }
523
524    fn output_fee(
525        &self,
526        amounts: &Amounts,
527        _output: &<Self::Common as ModuleCommon>::Output,
528    ) -> Option<Amounts> {
529        let unit = self.cfg.amount_unit;
530        let amount = amounts.get(&unit).copied().unwrap_or_default();
531        let fee = self.cfg.fee_consensus.fee(amount);
532
533        Some(Amounts::new_custom(unit, fee))
534    }
535
536    #[cfg(feature = "cli")]
537    async fn handle_cli_command(
538        &self,
539        args: &[std::ffi::OsString],
540    ) -> Result<serde_json::Value, ClientModuleError> {
541        cli::handle_cli_command(self, args)
542            .await
543            .map_err(ClientModuleError::other)
544    }
545
546    fn supports_being_primary(&self) -> PrimaryModuleSupport {
547        PrimaryModuleSupport::selected(PrimaryModulePriority::HIGH, [self.cfg.amount_unit])
548    }
549
550    async fn create_final_inputs_and_outputs(
551        &self,
552        dbtx: &mut DatabaseTransaction<'_>,
553        operation_id: OperationId,
554        unit: AmountUnit,
555        mut input_amount: Amount,
556        mut output_amount: Amount,
557    ) -> Result<
558        (
559            ClientInputBundle<MintInput, MintClientStateMachines>,
560            ClientOutputBundle<MintOutput, MintClientStateMachines>,
561        ),
562        ClientModuleError,
563    > {
564        if unit != self.cfg.amount_unit {
565            return Err(ClientModuleError::other(
566                "Module can only handle its configured amount unit",
567            ));
568        }
569
570        let requested_amount = output_amount.saturating_sub(input_amount);
571        // `select_funding_input` only reads notes, so the balance it left behind is
572        // still accurate for reporting a total below.
573        let Some(funding_notes) = self.select_funding_input(dbtx, requested_amount).await else {
574            let total_amount = self.get_balance(dbtx, unit).await;
575            return Err(InsufficientBalanceError {
576                requested_amount,
577                total_amount,
578            }
579            .into());
580        };
581
582        for note in &funding_notes {
583            self.remove_spendable_note(dbtx, note).await;
584        }
585
586        input_amount += funding_notes.iter().map(SpendableNote::amount).sum();
587
588        output_amount += funding_notes
589            .iter()
590            .map(|input| self.cfg.fee_consensus.fee(input.amount()))
591            .sum();
592
593        assert!(output_amount <= input_amount);
594
595        let (input_notes, output_amounts) = self
596            .rebalance(dbtx, &self.cfg.fee_consensus, input_amount - output_amount)
597            .await;
598
599        for note in &input_notes {
600            self.remove_spendable_note(dbtx, note).await;
601        }
602
603        input_amount += input_notes.iter().map(SpendableNote::amount).sum();
604
605        output_amount += input_notes
606            .iter()
607            .map(|note| self.cfg.fee_consensus.fee(note.amount()))
608            .sum();
609
610        output_amount += output_amounts
611            .iter()
612            .map(|denomination| {
613                denomination.amount() + self.cfg.fee_consensus.fee(denomination.amount())
614            })
615            .sum();
616
617        assert!(output_amount <= input_amount);
618
619        let mut spendable_notes = funding_notes
620            .into_iter()
621            .chain(input_notes)
622            .collect::<Vec<SpendableNote>>();
623
624        // We sort the notes by denomination to minimize the leaked information.
625        spendable_notes.sort_by_key(|note| note.denomination);
626
627        let input_bundle =
628            Self::create_input_bundle(operation_id, spendable_notes, false, self.cfg.amount_unit);
629
630        let mut denominations = represent_amount_with_fees(
631            input_amount.saturating_sub(output_amount),
632            &self.cfg.fee_consensus,
633        )
634        .into_iter()
635        .chain(output_amounts)
636        .collect::<Vec<Denomination>>();
637
638        // We sort the amounts to minimize the leaked information.
639        denominations.sort();
640
641        let output_bundle = self.create_output_bundle(operation_id, denominations).await;
642
643        let sender = self.balance_update_sender.clone();
644        dbtx.on_commit(move || sender.send_replace(()));
645
646        Ok((input_bundle, output_bundle))
647    }
648
649    async fn await_primary_module_output(
650        &self,
651        operation_id: OperationId,
652        outpoint: OutPoint,
653    ) -> Result<(), ClientModuleError> {
654        self.await_output_sm_success(operation_id, outpoint)
655            .await
656            .map_err(ClientModuleError::other)
657    }
658
659    async fn get_balance(&self, dbtx: &mut DatabaseTransaction<'_>, unit: AmountUnit) -> Amount {
660        if unit != self.cfg.amount_unit {
661            return Amount::ZERO;
662        }
663
664        self.get_count_by_denomination_dbtx(dbtx)
665            .await
666            .into_iter()
667            .map(|(denomination, count)| denomination.amount().mul_u64(count))
668            .sum()
669    }
670
671    async fn subscribe_balance_changes(&self) -> BoxStream<'static, ()> {
672        Box::pin(tokio_stream::wrappers::WatchStream::new(
673            self.balance_update_sender.subscribe(),
674        ))
675    }
676}
677
678impl MintClientModule {
679    async fn select_funding_input(
680        &self,
681        dbtx: &mut DatabaseTransaction<'_>,
682        mut excess_output: Amount,
683    ) -> Option<Vec<SpendableNote>> {
684        let mut selected_notes = Vec::new();
685        let mut target_notes = Vec::new();
686        let mut excess_notes = Vec::new();
687
688        for amount in client_denominations().rev() {
689            let notes_amount = dbtx
690                .find_by_prefix(&SpendableNoteAmountPrefix(amount))
691                .await
692                .map(|entry| entry.0.0)
693                .collect::<Vec<SpendableNoteUndecoded>>()
694                .await;
695
696            target_notes.extend(notes_amount.iter().take(TARGET_PER_DENOMINATION).copied());
697
698            if notes_amount.len() > 2 * TARGET_PER_DENOMINATION {
699                for note in notes_amount.into_iter().skip(TARGET_PER_DENOMINATION) {
700                    let note_fee = self.cfg.fee_consensus.fee(note.amount());
701
702                    let note_value = note
703                        .amount()
704                        .checked_sub(note_fee)
705                        .expect("All our notes are economical");
706
707                    excess_output = excess_output.saturating_sub(note_value);
708
709                    selected_notes.push(note);
710                }
711            } else {
712                excess_notes.extend(notes_amount.into_iter().skip(TARGET_PER_DENOMINATION));
713            }
714        }
715
716        if excess_output == Amount::ZERO {
717            return Some(
718                selected_notes
719                    .into_iter()
720                    .map(SpendableNoteUndecoded::decode_own)
721                    .collect(),
722            );
723        }
724
725        for note in excess_notes.into_iter().chain(target_notes) {
726            let note_amount = note.amount();
727            let note_value = note_amount
728                .checked_sub(self.cfg.fee_consensus.fee(note_amount))
729                .expect("All our notes are economical");
730
731            excess_output = excess_output.saturating_sub(note_value);
732
733            selected_notes.push(note);
734
735            if excess_output == Amount::ZERO {
736                return Some(
737                    selected_notes
738                        .into_iter()
739                        .map(SpendableNoteUndecoded::decode_own)
740                        .collect(),
741                );
742            }
743        }
744
745        None
746    }
747
748    async fn rebalance(
749        &self,
750        dbtx: &mut DatabaseTransaction<'_>,
751        fee: &FeeConsensus,
752        mut excess_input: Amount,
753    ) -> (Vec<SpendableNote>, Vec<Denomination>) {
754        let n_denominations = self.get_count_by_denomination_dbtx(dbtx).await;
755
756        let mut notes = dbtx
757            .find_by_prefix_sorted_descending(&SpendableNotePrefix)
758            .await
759            .map(|entry| entry.0.0)
760            .fuse();
761
762        let mut input_notes = Vec::new();
763        let mut output_denominations = Vec::new();
764
765        for d in client_denominations() {
766            let n_denomination = n_denominations.get(&d).copied().unwrap_or(0);
767
768            let n_missing = TARGET_PER_DENOMINATION.saturating_sub(n_denomination as usize);
769
770            for _ in 0..n_missing {
771                match excess_input.checked_sub(d.amount() + fee.fee(d.amount())) {
772                    Some(remaining_excess) => excess_input = remaining_excess,
773                    None => match notes.next().await {
774                        Some(note) => {
775                            if note.amount() <= d.amount() + fee.fee(d.amount()) {
776                                break;
777                            }
778
779                            excess_input += note.amount() - (d.amount() + fee.fee(d.amount()));
780
781                            input_notes.push(note.decode_own());
782                        }
783                        None => break,
784                    },
785                }
786
787                output_denominations.push(d);
788            }
789        }
790
791        (input_notes, output_denominations)
792    }
793
794    fn create_input_bundle(
795        operation_id: OperationId,
796        notes: Vec<SpendableNote>,
797        include_receive_sm: bool,
798        amount_unit: AmountUnit,
799    ) -> ClientInputBundle<MintInput, MintClientStateMachines> {
800        let inputs = notes
801            .iter()
802            .map(|spendable_note| ClientInput {
803                input: MintInput::new_v0(spendable_note.note()),
804                keys: vec![spendable_note.keypair],
805                amounts: Amounts::new_custom(amount_unit, spendable_note.amount()),
806            })
807            .collect();
808
809        let input_sms = vec![ClientInputSM {
810            state_machines: Arc::new(move |range: OutPointRange| {
811                let mut sms = vec![MintClientStateMachines::Input(InputStateMachine {
812                    common: InputSMCommon {
813                        operation_id,
814                        txid: range.txid(),
815                        spendable_notes: notes.clone(),
816                    },
817                    state: InputSMState::Pending,
818                })];
819
820                if include_receive_sm {
821                    sms.push(MintClientStateMachines::Receive(ReceiveStateMachine {
822                        common: crate::receive::ReceiveSMCommon {
823                            operation_id,
824                            txid: range.txid(),
825                        },
826                        state: crate::receive::ReceiveSMState::Pending,
827                    }));
828                }
829
830                sms
831            }),
832        }];
833
834        ClientInputBundle::new(inputs, input_sms)
835    }
836
837    /// Build the blinded outputs and their output state machines for a
838    /// reissue transaction.
839    ///
840    /// # Cancel safety
841    ///
842    /// Cancel safe: this only reads pre-ground blinding tweaks from an
843    /// in-memory channel and constructs values, it performs no database
844    /// writes. Dropping the future merely discards the tweaks it had taken,
845    /// which are random (so nothing is lost) and replenished by the
846    /// background grinder task.
847    async fn create_output_bundle(
848        &self,
849        operation_id: OperationId,
850        requested_denominations: Vec<Denomination>,
851    ) -> ClientOutputBundle<MintOutput, MintClientStateMachines> {
852        let issuance_requests = futures::stream::iter(requested_denominations)
853            .zip(self.tweak_receiver.clone())
854            .map(|(d, tweak)| NoteIssuanceRequest::new(d, tweak, &self.root_secret))
855            .collect::<Vec<NoteIssuanceRequest>>()
856            .await;
857
858        let amount_unit = self.cfg.amount_unit;
859        let outputs = issuance_requests
860            .iter()
861            .map(|request| ClientOutput {
862                output: request.output(),
863                amounts: Amounts::new_custom(amount_unit, request.denomination.amount()),
864            })
865            .collect();
866
867        let output_sms = vec![ClientOutputSM {
868            state_machines: Arc::new(move |range: OutPointRange| {
869                vec![MintClientStateMachines::Output(MintOutputStateMachine {
870                    common: OutputSMCommon {
871                        operation_id,
872                        range: Some(range),
873                        issuance_requests: issuance_requests.clone(),
874                    },
875                    state: OutputSMState::Pending,
876                })]
877            }),
878        }];
879
880        ClientOutputBundle::new(outputs, output_sms)
881    }
882
883    /// Wait for the output state machine of the given outpoint to reach a
884    /// terminal state.
885    ///
886    /// # Cancel safety
887    ///
888    /// Cancel safe: this only subscribes to the operation's state
889    /// notifications and waits, it performs no database writes. Dropping the
890    /// future stops the wait; the underlying state machine keeps running in
891    /// the executor.
892    async fn await_output_sm_success(
893        &self,
894        operation_id: OperationId,
895        outpoint: OutPoint,
896    ) -> Result<(), AwaitOutputError> {
897        let stream = self
898            .notifier
899            .subscribe(operation_id)
900            .await
901            .filter_map(|state| async {
902                let MintClientStateMachines::Output(state) = state else {
903                    return None;
904                };
905
906                if !state.common.range?.into_iter().contains(&outpoint) {
907                    return None;
908                }
909
910                match state.state {
911                    OutputSMState::Pending => None,
912                    OutputSMState::Success => Some(Ok(())),
913                    OutputSMState::Aborted => Some(Err(AwaitOutputError::Rejected)),
914                    OutputSMState::Failure => Some(Err(AwaitOutputError::Failed)),
915                }
916            });
917
918        pin_mut!(stream);
919
920        stream.next_or_pending().await
921    }
922
923    /// Count the `ECash` notes in the client's database by denomination.
924    pub async fn get_count_by_denomination(&self) -> BTreeMap<Denomination, u64> {
925        self.get_count_by_denomination_dbtx(
926            &mut self.client_ctx.module_db().begin_transaction_nc().await,
927        )
928        .await
929    }
930
931    async fn get_count_by_denomination_dbtx(
932        &self,
933        dbtx: &mut DatabaseTransaction<'_>,
934    ) -> BTreeMap<Denomination, u64> {
935        dbtx.find_by_prefix(&SpendableNotePrefix)
936            .await
937            .fold(BTreeMap::new(), |mut acc, entry| async move {
938                acc.entry(entry.0.0.denomination)
939                    .and_modify(|count| *count += 1)
940                    .or_insert(1);
941
942                acc
943            })
944            .await
945    }
946
947    /// Send `ECash` for the given amount and return the send operation ID. The
948    /// amount will be rounded up to a multiple of 512 msats which is the
949    /// smallest denomination used throughout the client. If the rounded
950    /// amount cannot be covered with the ecash notes in the client's
951    /// database the client will create a transaction to reissue the
952    /// required denominations. To cancel a successful ecash send simply
953    /// receive it yourself.
954    ///
955    /// If `include_invite` is set, the federation's invite code is embedded in
956    /// the returned ecash so a recipient that has not joined the federation can
957    /// do so directly from the received ecash.
958    ///
959    /// # Cancel safety
960    ///
961    /// This method is cancel safe. Every database mutation it makes happens
962    /// inside a single
963    /// [`autocommit`](fedimint_core::db::Database::autocommit) transaction
964    /// (spending the notes for the fast path, and creating the change-making
965    /// reissue transaction otherwise), so dropping the future either commits
966    /// a unit of work in full or leaves the database untouched: notes are
967    /// never removed without a persisted operation that produces the ecash or
968    /// returns the change. Federation submission is driven by a state machine
969    /// in the executor rather than awaited here, so a cancelled call cannot
970    /// abort an in-flight reissue; if change-making had already been
971    /// submitted it completes on its own and the notes return to the balance.
972    pub async fn send(
973        &self,
974        amount: Amount,
975        custom_meta: Value,
976        include_invite: bool,
977    ) -> Result<(OperationId, ECash), SendECashError> {
978        let amount = round_to_multiple(amount, client_denominations().next().unwrap().amount());
979
980        if let Some((operation_id, ecash)) = self
981            .client_ctx
982            .module_db()
983            .autocommit(
984                |dbtx, _| {
985                    Box::pin(self.send_ecash_dbtx(
986                        dbtx,
987                        amount,
988                        custom_meta.clone(),
989                        include_invite,
990                    ))
991                },
992                Some(100),
993            )
994            .await
995            .expect("Failed to commit dbtx after 100 retries")
996        {
997            return Ok((operation_id, ecash));
998        }
999
1000        self.client_ctx
1001            .global_api()
1002            .session_count()
1003            .await
1004            .map_err(|_| SendECashError::Offline)?;
1005
1006        let operation_id = OperationId::new_random();
1007
1008        let output = self
1009            .create_output_bundle(operation_id, represent_amount(amount))
1010            .await;
1011        let output = self.client_ctx.make_client_outputs(output);
1012        let cm = custom_meta.clone();
1013
1014        let range = self
1015            .client_ctx
1016            .finalize_and_submit_transaction(
1017                operation_id,
1018                MintCommonInit::KIND.as_str(),
1019                move |change_outpoint_range| MintOperationMeta::Reissue {
1020                    change_outpoint_range,
1021                    amount,
1022                    custom_meta: cm.clone(),
1023                },
1024                TransactionBuilder::new().with_outputs(output),
1025            )
1026            .await
1027            .map_err(|error| match error {
1028                TransactionSubmitError::InsufficientFunds(_) => SendECashError::InsufficientBalance,
1029                other => SendECashError::Failed(other),
1030            })?;
1031
1032        for outpoint in range {
1033            self.await_output_sm_success(operation_id, outpoint)
1034                .await
1035                .map_err(|_| SendECashError::Failure)?;
1036        }
1037
1038        Box::pin(self.send(amount, custom_meta, include_invite)).await
1039    }
1040
1041    /// Fast path for [`Self::send`]: if exact change is already held, spend
1042    /// those notes and record the send operation. Returns `None` when exact
1043    /// change is not available, leaving the caller to make change.
1044    ///
1045    /// # Cancel safety
1046    ///
1047    /// All writes go to the passed `dbtx`, so this inherits the cancel safety
1048    /// of that transaction: it must be run inside an `autocommit` (as `send`
1049    /// does) for the note spend and the operation log entry to commit or roll
1050    /// back together.
1051    async fn send_ecash_dbtx(
1052        &self,
1053        dbtx: &mut DatabaseTransaction<'_>,
1054        remaining_amount: Amount,
1055        custom_meta: Value,
1056        include_invite: bool,
1057    ) -> Result<Option<(OperationId, ECash)>, Infallible> {
1058        let Some(notes) = Self::select_exact_change(&mut dbtx.to_ref_nc(), remaining_amount).await
1059        else {
1060            return Ok(None);
1061        };
1062
1063        for spendable_note in &notes {
1064            self.remove_spendable_note(dbtx, spendable_note).await;
1065        }
1066
1067        let ecash = if include_invite {
1068            let invite = self.client_ctx.get_invite_code().await;
1069            ECash::new_with_invite(notes, &invite)
1070        } else {
1071            ECash::new(self.federation_id, notes)
1072        }
1073        .with_unit(self.cfg.amount_unit);
1074        let amount = ecash.amount();
1075        let operation_id = OperationId::new_random();
1076
1077        self.client_ctx
1078            .add_operation_log_entry_dbtx(
1079                dbtx,
1080                operation_id,
1081                MintCommonInit::KIND.as_str(),
1082                MintOperationMeta::Send {
1083                    ecash: base32::encode_prefixed(FEDIMINT_PREFIX, &ecash),
1084                    custom_meta,
1085                },
1086            )
1087            .await;
1088
1089        self.client_ctx
1090            .log_event(
1091                dbtx,
1092                SendPaymentEvent {
1093                    operation_id,
1094                    amount,
1095                    ecash: base32::encode_prefixed(FEDIMINT_PREFIX, &ecash),
1096                },
1097            )
1098            .await;
1099
1100        let sender = self.balance_update_sender.clone();
1101        dbtx.on_commit(move || sender.send_replace(()));
1102
1103        Ok(Some((operation_id, ecash)))
1104    }
1105
1106    /// Receive the `ECash` by reissuing the notes and return the operation ID.
1107    pub async fn receive(
1108        &self,
1109        ecash: ECash,
1110        custom_meta: Value,
1111    ) -> Result<OperationId, ReceiveECashError> {
1112        let operation_id = OperationId::from_encodable(&ecash);
1113
1114        if ecash.mint() != Some(self.federation_id) {
1115            return Err(ReceiveECashError::WrongFederation);
1116        }
1117
1118        if ecash
1119            .notes()
1120            .iter()
1121            .any(|note| note.amount() <= self.cfg.fee_consensus.base_fee())
1122        {
1123            return Err(ReceiveECashError::UneconomicalDenomination);
1124        }
1125
1126        let input =
1127            Self::create_input_bundle(operation_id, ecash.notes(), true, self.cfg.amount_unit);
1128        let input = self.client_ctx.make_client_inputs(input);
1129        let ec = base32::encode_prefixed(FEDIMINT_PREFIX, &ecash);
1130
1131        self.client_ctx
1132            .finalize_and_submit_transaction(
1133                operation_id,
1134                MintCommonInit::KIND.as_str(),
1135                move |change_outpoint_range| MintOperationMeta::Receive {
1136                    change_outpoint_range,
1137                    ecash: ec.clone(),
1138                    custom_meta: custom_meta.clone(),
1139                },
1140                TransactionBuilder::new().with_inputs(input),
1141            )
1142            .await
1143            .map_err(|error| match error {
1144                TransactionSubmitError::OperationAlreadyExists(_) => {
1145                    ReceiveECashError::AlreadyReceived
1146                }
1147                TransactionSubmitError::InsufficientFunds(_) => {
1148                    ReceiveECashError::InsufficientFunds
1149                }
1150                other => ReceiveECashError::Failed(other),
1151            })?;
1152
1153        let mut dbtx = self.client_ctx.module_db().begin_transaction().await;
1154
1155        self.client_ctx
1156            .log_event(
1157                &mut dbtx,
1158                ReceivePaymentEvent {
1159                    operation_id,
1160                    amount: ecash.amount(),
1161                },
1162            )
1163            .await;
1164
1165        dbtx.commit_tx().await;
1166
1167        Ok(operation_id)
1168    }
1169
1170    /// Computes the exact fee a `receive(ecash)` would incur given the client's
1171    /// current note inventory, without submitting anything.
1172    ///
1173    /// This runs the same change generation the real receive does
1174    /// (`create_final_inputs_and_outputs`, including funding selection and
1175    /// rebalancing) against a non-committable transaction that is dropped
1176    /// rather than committed, so the client's notes are read but left
1177    /// untouched. The quote is point-in-time: it depends on the current
1178    /// inventory and can move as notes change.
1179    pub async fn receive_fee_quote(
1180        &self,
1181        ecash: &ECash,
1182    ) -> Result<FeeQuote, TransactionSubmitError> {
1183        // A receive submits the ecash notes as explicit inputs and no explicit
1184        // outputs; the shared, module-agnostic fee quote runs the primary-module
1185        // balancing (rebalancing + minting change) over the real inventory.
1186        let notes = ecash.notes();
1187        let input_amount: Amount = notes.iter().map(SpendableNote::amount).sum();
1188        let input_fee: Amount = notes
1189            .iter()
1190            .map(|note| self.cfg.fee_consensus.fee(note.amount()))
1191            .sum();
1192
1193        self.client_ctx
1194            .fee_quote(
1195                OperationId::new_random(),
1196                FeeQuoteRequest {
1197                    input_amount: Amounts::new_custom(self.cfg.amount_unit, input_amount),
1198                    output_amount: Amounts::ZERO,
1199                    input_fee: Amounts::new_custom(self.cfg.amount_unit, input_fee),
1200                    output_fee: Amounts::ZERO,
1201                },
1202            )
1203            .await
1204    }
1205
1206    /// Computes the fee a `send(amount)` would incur given the client's current
1207    /// note inventory, without sending anything.
1208    ///
1209    /// A send is free when the client's existing notes can cover the (rounded)
1210    /// amount exactly — it just hands those notes out. Otherwise the send first
1211    /// reissues itself the right denominations, and that self-reissue
1212    /// transaction is the only thing a send ever pays a fee for. This quote
1213    /// mirrors that: it returns [`FeeQuote::ZERO`] when exact change is
1214    /// available, and otherwise quotes the reissue the same way the real send
1215    /// submits it (explicit outputs `represent_amount(amount)`, no explicit
1216    /// inputs) via the shared, module-agnostic fee quote over the real
1217    /// inventory. The quote is point-in-time: it depends on the current
1218    /// inventory and can move as notes change.
1219    pub async fn send_fee_quote(&self, amount: Amount) -> Result<FeeQuote, TransactionSubmitError> {
1220        let amount = round_to_multiple(amount, client_denominations().next().unwrap().amount());
1221
1222        // Exact-change path: handing out existing notes never costs a fee.
1223        if self.can_make_exact_change(amount).await {
1224            return Ok(FeeQuote::ZERO);
1225        }
1226
1227        // Reissue path: the send mints itself `represent_amount(amount)` as
1228        // explicit outputs (no explicit inputs) and the primary module funds and
1229        // balances it. Quote that exact transaction.
1230        let denominations = represent_amount(amount);
1231        let output_amount: Amount = denominations.iter().map(|d| d.amount()).sum();
1232        let output_fee: Amount = denominations
1233            .iter()
1234            .map(|d| self.cfg.fee_consensus.fee(d.amount()))
1235            .sum();
1236
1237        self.client_ctx
1238            .fee_quote(
1239                OperationId::new_random(),
1240                FeeQuoteRequest {
1241                    input_amount: Amounts::ZERO,
1242                    output_amount: Amounts::new_custom(self.cfg.amount_unit, output_amount),
1243                    input_fee: Amounts::ZERO,
1244                    output_fee: Amounts::new_custom(self.cfg.amount_unit, output_fee),
1245                },
1246            )
1247            .await
1248    }
1249
1250    /// Returns whether the client's current notes can be handed out to cover
1251    /// `amount` exactly — the free path in [`Self::send`] — without modifying
1252    /// the inventory. Shares its greedy selection with
1253    /// [`Self::send_ecash_dbtx`] via [`Self::select_exact_change`].
1254    async fn can_make_exact_change(&self, remaining_amount: Amount) -> bool {
1255        let mut dbtx = self.client_ctx.module_db().begin_transaction_nc().await;
1256
1257        Self::select_exact_change(&mut dbtx, remaining_amount)
1258            .await
1259            .is_some()
1260    }
1261
1262    /// Greedily selects spendable notes (largest-first) summing to *exactly*
1263    /// `remaining_amount`, or `None` if the inventory can't make exact change.
1264    /// Reads the DB but does not modify it. Single source of truth for the free
1265    /// (hand-out-existing-notes) selection shared by [`Self::send_ecash_dbtx`]
1266    /// and [`Self::can_make_exact_change`].
1267    async fn select_exact_change(
1268        dbtx: &mut DatabaseTransaction<'_>,
1269        mut remaining_amount: Amount,
1270    ) -> Option<Vec<SpendableNote>> {
1271        let mut stream = dbtx
1272            .find_by_prefix_sorted_descending(&SpendableNotePrefix)
1273            .await
1274            .map(|entry| entry.0.0);
1275
1276        let mut notes = vec![];
1277
1278        while let Some(spendable_note) = stream.next().await {
1279            remaining_amount = match remaining_amount.checked_sub(spendable_note.amount()) {
1280                Some(amount) => amount,
1281                None => continue,
1282            };
1283
1284            notes.push(spendable_note.decode_own());
1285
1286            if remaining_amount == Amount::ZERO {
1287                break;
1288            }
1289        }
1290
1291        (remaining_amount == Amount::ZERO).then_some(notes)
1292    }
1293
1294    /// Await the final state of the receive operation.
1295    pub async fn await_final_receive_operation_state(
1296        &self,
1297        operation_id: OperationId,
1298    ) -> Result<FinalReceiveOperationState, OperationLookupError> {
1299        let operation = self.client_ctx.get_operation(operation_id).await?;
1300        let mut stream = self.notifier.subscribe(operation_id).await;
1301
1302        let mut stream = self
1303            .client_ctx
1304            .outcome_or_updates(&operation, operation_id, |_| true, move || {
1305                async_stream::stream! {
1306                    loop {
1307                        if let Some(MintClientStateMachines::Receive(state)) = stream.next().await {
1308                            match state.state {
1309                                ReceiveSMState::Pending => {}
1310                                ReceiveSMState::Success => {
1311                                    yield FinalReceiveOperationState::Success;
1312                                    return;
1313                                }
1314                                ReceiveSMState::Rejected(..) => {
1315                                    yield FinalReceiveOperationState::Rejected;
1316                                    return;
1317                                }
1318                            }
1319                        }
1320                    }
1321                }
1322            })
1323            .into_stream();
1324
1325        let mut final_state = None;
1326
1327        while let Some(state) = stream.next().await {
1328            final_state = Some(state);
1329        }
1330
1331        Ok(final_state.expect("Stream contains one final state"))
1332    }
1333
1334    async fn remove_spendable_note(
1335        &self,
1336        dbtx: &mut DatabaseTransaction<'_>,
1337        spendable_note: &SpendableNote,
1338    ) {
1339        dbtx.remove_entry(&SpendableNoteKey(spendable_note.clone().into()))
1340            .await
1341            .expect("Must delete existing spendable note");
1342    }
1343}
1344
1345/// Hands out guardians so that only one slice request is outstanding to each.
1346///
1347/// Whichever peer finishes first takes the next slice, so a slow guardian
1348/// receives less work without anyone having to measure how slow it is, and one
1349/// that is not answering ties up a single request rather than a share of all
1350/// of them.
1351#[derive(Clone)]
1352struct PeerPool {
1353    receiver: async_channel::Receiver<PeerId>,
1354    sender: async_channel::Sender<PeerId>,
1355}
1356
1357impl PeerPool {
1358    fn new(peers: &BTreeSet<PeerId>) -> Self {
1359        let (sender, receiver) = async_channel::bounded(peers.len().max(1));
1360
1361        for peer in peers {
1362            sender
1363                .try_send(*peer)
1364                .expect("Capacity was sized to hold every peer");
1365        }
1366
1367        Self { receiver, sender }
1368    }
1369
1370    /// Wait for a guardian with no request outstanding.
1371    async fn acquire(&self) -> PeerId {
1372        self.receiver
1373            .recv()
1374            .await
1375            .expect("The sender is held for as long as the receiver")
1376    }
1377
1378    /// Take a guardian out of rotation, putting it back once it has sat out
1379    /// [`PEER_READMISSION`].
1380    ///
1381    /// Dropping it for good would cost a guardian that timed out once the rest
1382    /// of the recovery, which on a federation of four is a quarter of the
1383    /// capacity thrown away for a single bad request.
1384    fn retire(&self, peer: PeerId) {
1385        let pool = self.clone();
1386
1387        fedimint_core::runtime::spawn("mintv2 recovery peer readmission", async move {
1388            fedimint_core::runtime::sleep(PEER_READMISSION).await;
1389
1390            pool.release(peer);
1391        });
1392    }
1393
1394    /// Put a guardian back for the next slice.
1395    fn release(&self, peer: PeerId) {
1396        self.sender
1397            .try_send(peer)
1398            .expect("Only peers taken from the pool are put back");
1399    }
1400}
1401
1402/// Download a slice, asking one guardian at a time and holding it for the
1403/// duration of the request.
1404async fn download_slice(
1405    module_api: DynModuleApi,
1406    peers: PeerPool,
1407    start: u64,
1408    end: u64,
1409    expected_hash: sha256::Hash,
1410) -> Vec<RecoveryItem> {
1411    let mut timeouts = custom_backoff(SLICE_TIMEOUT, MAX_SLICE_TIMEOUT, None);
1412
1413    loop {
1414        let peer = peers.acquire().await;
1415
1416        let timeout = timeouts.next().expect("The backoff never gives up");
1417
1418        let result = module_api
1419            .fetch_recovery_slice(peer, timeout, start, end)
1420            .await;
1421
1422        match result {
1423            Ok(data) if data.consensus_hash::<sha256::Hash>() == expected_hash => {
1424                peers.release(peer);
1425
1426                return data;
1427            }
1428            // Either served something the other guardians disagree with, or
1429            // did not answer at all. Either way it sits out for a while: a
1430            // request that reaches the timeout took many times longer than a
1431            // healthy guardian needs, so asking it again mostly buys another
1432            // timeout. The timeout grows in case it is the client's own
1433            // connection that is too slow for the initial one.
1434            Ok(_) | Err(_) => peers.retire(peer),
1435        }
1436    }
1437}
1438
1439/// A failure to send e-cash by preparing notes to hand to the recipient.
1440#[derive(Error, Debug)]
1441#[non_exhaustive]
1442pub enum SendECashError {
1443    /// The client needs to reissue notes to make change, but has no
1444    /// connection to the federation to do so.
1445    #[error("We need to reissue notes but the client is offline")]
1446    Offline,
1447    /// The client's balance cannot cover the amount requested.
1448    #[error("The clients balance is insufficient")]
1449    InsufficientBalance,
1450    /// The client failed to prepare the notes for a reason it cannot
1451    /// recover from.
1452    #[error("A non-recoverable error has occurred")]
1453    Failure,
1454    /// The change-making reissue transaction could not be submitted for a
1455    /// reason unrelated to funding.
1456    #[error("The reissue transaction could not be submitted")]
1457    Failed(#[source] TransactionSubmitError),
1458}
1459
1460/// A failure to receive e-cash by reissuing it.
1461#[derive(Error, Debug)]
1462#[non_exhaustive]
1463pub enum ReceiveECashError {
1464    /// The e-cash was issued by a different federation.
1465    #[error("The ECash is from a different federation")]
1466    WrongFederation,
1467
1468    /// One of the notes is worth no more than the fee to reissue it.
1469    #[error("ECash contains an uneconomical denomination")]
1470    UneconomicalDenomination,
1471
1472    /// The client cannot cover the fee the reissue costs.
1473    #[error("Receiving ecash requires additional funds")]
1474    InsufficientFunds,
1475
1476    /// An operation for this exact e-cash already exists, so it was already
1477    /// received.
1478    #[error("The ECash was already received")]
1479    AlreadyReceived,
1480
1481    /// The reissue transaction could not be submitted for a reason unrelated
1482    /// to funding.
1483    #[error("The reissue transaction could not be submitted")]
1484    Failed(#[source] TransactionSubmitError),
1485}
1486
1487/// Why an output's state machine did not issue its notes.
1488#[derive(Debug, Error)]
1489enum AwaitOutputError {
1490    /// The transaction that creates the output was rejected.
1491    #[error("Transaction was rejected")]
1492    Rejected,
1493
1494    /// The notes of the output could not be finalized.
1495    #[error("Failed to finalize notes")]
1496    Failed,
1497}
1498
1499#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
1500pub enum FinalReceiveOperationState {
1501    // The ecash notes have been reissued
1502    Success,
1503    // The ecash notes were already spent
1504    Rejected,
1505}
1506
1507#[derive(Debug, Clone, Eq, PartialEq, Hash, Decodable, Encodable)]
1508pub enum MintClientStateMachines {
1509    Input(InputStateMachine),
1510    Output(MintOutputStateMachine),
1511    Receive(ReceiveStateMachine),
1512}
1513
1514impl IntoDynInstance for MintClientStateMachines {
1515    type DynType = DynState;
1516
1517    fn into_dyn(self, instance_id: ModuleInstanceId) -> Self::DynType {
1518        DynState::from_typed(instance_id, self)
1519    }
1520}
1521
1522impl State for MintClientStateMachines {
1523    type ModuleContext = MintClientContext;
1524
1525    fn transitions(
1526        &self,
1527        context: &Self::ModuleContext,
1528        global_context: &DynGlobalClientContext,
1529    ) -> Vec<StateTransition<Self>> {
1530        match self {
1531            MintClientStateMachines::Input(redemption_state) => {
1532                sm_enum_variant_translation!(
1533                    redemption_state.transitions(context, global_context),
1534                    MintClientStateMachines::Input
1535                )
1536            }
1537            MintClientStateMachines::Output(issuance_state) => {
1538                sm_enum_variant_translation!(
1539                    issuance_state.transitions(context, global_context),
1540                    MintClientStateMachines::Output
1541                )
1542            }
1543            MintClientStateMachines::Receive(receive_state) => {
1544                sm_enum_variant_translation!(
1545                    receive_state.transitions(context, global_context),
1546                    MintClientStateMachines::Receive
1547                )
1548            }
1549        }
1550    }
1551
1552    fn operation_id(&self) -> OperationId {
1553        match self {
1554            MintClientStateMachines::Input(redemption_state) => redemption_state.operation_id(),
1555            MintClientStateMachines::Output(issuance_state) => issuance_state.operation_id(),
1556            MintClientStateMachines::Receive(receive_state) => receive_state.operation_id(),
1557        }
1558    }
1559}
1560
1561fn round_to_multiple(amount: Amount, min_denomiation: Amount) -> Amount {
1562    Amount::from_msats(amount.msats.next_multiple_of(min_denomiation.msats))
1563}
1564
1565fn represent_amount_with_fees(
1566    mut remaining_amount: Amount,
1567    fee_consensus: &FeeConsensus,
1568) -> Vec<Denomination> {
1569    let mut denominations = Vec::new();
1570
1571    // Add denominations with a greedy algorithm
1572    for denomination in client_denominations().rev() {
1573        let n_add =
1574            remaining_amount / (denomination.amount() + fee_consensus.fee(denomination.amount()));
1575
1576        denominations.extend(std::iter::repeat_n(denomination, n_add as usize));
1577
1578        remaining_amount -=
1579            n_add * (denomination.amount() + fee_consensus.fee(denomination.amount()));
1580    }
1581
1582    // We sort the notes by amount to minimize the leaked information.
1583    denominations.sort();
1584
1585    denominations
1586}
1587
1588fn represent_amount(mut remaining_amount: Amount) -> Vec<Denomination> {
1589    let mut denominations = Vec::new();
1590
1591    // Add denominations with a greedy algorithm
1592    for denomination in client_denominations().rev() {
1593        let n_add = remaining_amount / denomination.amount();
1594
1595        denominations.extend(std::iter::repeat_n(denomination, n_add as usize));
1596
1597        remaining_amount -= n_add * denomination.amount();
1598    }
1599
1600    denominations
1601}