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;
89const PEER_READMISSION: Duration = Duration::from_secs(60);
91const SLICE_TIMEOUT: Duration = Duration::from_secs(10);
99const 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#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Encodable, Decodable)]
141pub struct SpendableNoteUndecoded {
142 pub denomination: Denomination,
143 pub keypair: Keypair,
144 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 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 .buffer_unordered(PARALLEL_SLICE_REQUESTS);
314
315 let tweak_filter = issuance::tweak_filter(args.module_root_secret());
316
317 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 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 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 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 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 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 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 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 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 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 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 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 ¬es {
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 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 pub async fn receive_fee_quote(
1180 &self,
1181 ecash: &ECash,
1182 ) -> Result<FeeQuote, TransactionSubmitError> {
1183 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 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 if self.can_make_exact_change(amount).await {
1224 return Ok(FeeQuote::ZERO);
1225 }
1226
1227 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 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 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 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#[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 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 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 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
1402async 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 Ok(_) | Err(_) => peers.retire(peer),
1435 }
1436 }
1437}
1438
1439#[derive(Error, Debug)]
1441#[non_exhaustive]
1442pub enum SendECashError {
1443 #[error("We need to reissue notes but the client is offline")]
1446 Offline,
1447 #[error("The clients balance is insufficient")]
1449 InsufficientBalance,
1450 #[error("A non-recoverable error has occurred")]
1453 Failure,
1454 #[error("The reissue transaction could not be submitted")]
1457 Failed(#[source] TransactionSubmitError),
1458}
1459
1460#[derive(Error, Debug)]
1462#[non_exhaustive]
1463pub enum ReceiveECashError {
1464 #[error("The ECash is from a different federation")]
1466 WrongFederation,
1467
1468 #[error("ECash contains an uneconomical denomination")]
1470 UneconomicalDenomination,
1471
1472 #[error("Receiving ecash requires additional funds")]
1474 InsufficientFunds,
1475
1476 #[error("The ECash was already received")]
1479 AlreadyReceived,
1480
1481 #[error("The reissue transaction could not be submitted")]
1484 Failed(#[source] TransactionSubmitError),
1485}
1486
1487#[derive(Debug, Error)]
1489enum AwaitOutputError {
1490 #[error("Transaction was rejected")]
1492 Rejected,
1493
1494 #[error("Failed to finalize notes")]
1496 Failed,
1497}
1498
1499#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
1500pub enum FinalReceiveOperationState {
1501 Success,
1503 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 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 denominations.sort();
1584
1585 denominations
1586}
1587
1588fn represent_amount(mut remaining_amount: Amount) -> Vec<Denomination> {
1589 let mut denominations = Vec::new();
1590
1591 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}