1use std::collections::BTreeMap;
2use std::future::Future;
3use std::sync::Arc;
4use std::time::Duration;
5
6use bitcoin::key::Secp256k1;
7use fedimint_api_client::api::global_api::with_cache::GlobalFederationApiWithCacheExt as _;
8use fedimint_api_client::api::global_api::with_request_hook::{
9 ApiRequestHook, RawFederationApiWithRequestHookExt as _,
10};
11use fedimint_api_client::api::{
12 ApiVersionSet, ClientConfigDownloadError, DynGlobalApi, FederationApi, FederationApiExt as _,
13 FederationError,
14};
15use fedimint_api_client::download_from_invite_code;
16use fedimint_bitcoind::DynBitcoindRpc;
17use fedimint_client_module::api::ClientRawFederationApiExt as _;
18use fedimint_client_module::meta::LegacyMetaSource;
19use fedimint_client_module::module::init::{
20 BitcoindRpcFactory, BitcoindRpcNoChainIdFactory, ClientModuleInit, RecoveryMode,
21};
22use fedimint_client_module::module::recovery::RecoveryProgress;
23use fedimint_client_module::module::{
24 ClientModuleRegistry, FinalClientIface, PrimaryModulePriority, PrimaryModuleSupport,
25};
26use fedimint_client_module::secret::{DeriveableSecretClientExt as _, get_default_client_secret};
27use fedimint_client_module::transaction::{
28 TRANSACTION_SUBMISSION_MODULE_INSTANCE, TxSubmissionContext, tx_submission_sm_decoder,
29};
30use fedimint_client_module::{AdminCreds, ModuleRecoveryStarted};
31use fedimint_connectors::ConnectorRegistry;
32use fedimint_core::config::{ClientConfig, FederationId, ModuleInitRegistry};
33use fedimint_core::core::{ModuleInstanceId, ModuleKind};
34use fedimint_core::db::{
35 Database, IDatabaseTransactionOpsCoreTyped as _, verify_module_db_integrity_dbtx,
36};
37use fedimint_core::endpoint_constants::CLIENT_CONFIG_ENDPOINT;
38use fedimint_core::envs::is_running_in_test_env;
39use fedimint_core::invite_code::InviteCode;
40use fedimint_core::module::registry::ModuleDecoderRegistry;
41use fedimint_core::module::{ApiRequestErased, ApiVersion, SupportedApiVersionsSummary};
42use fedimint_core::task::TaskGroup;
43use fedimint_core::task::jit::{Jit, JitTry};
44use fedimint_core::util::{FmtCompact as _, SafeUrl};
45use fedimint_core::{ChainId, NumPeers, PeerId, fedimint_build_code_version_env};
46use fedimint_derive_secret::DerivableSecret;
47use fedimint_eventlog::{
48 DBTransactionEventLogExt as _, EventLogEntry, run_event_log_ordering_task,
49};
50use fedimint_logging::LOG_CLIENT;
51use tokio::sync::{broadcast, watch};
52use tracing::{Span, debug, trace, warn};
53
54use super::handle::ClientHandle;
55use super::{Client, client_decoders};
56use crate::api_announcements::{
57 PeersSignedApiAnnouncements, fetch_api_announcements_from_at_least_num_of_peers, get_api_urls,
58 run_api_announcement_refresh_task, store_api_announcements_updates_from_peers,
59};
60use crate::backup::{ClientBackup, Metadata};
61use crate::client::{ModuleRecoveryFuture, PrimaryModuleCandidates, RecoveryStatus};
62use crate::db::{
63 self, ApiSecretKey, ChainIdKey, ClientInitStateKey, ClientMetadataKey, ClientModuleRecovery,
64 ClientModuleRecoveryState, ClientPreRootSecretHashKey, InitMode, InitState,
65 PendingClientConfigKey, apply_migrations_client_module_dbtx,
66};
67use crate::error::ClientBuildError;
68use crate::guardian_metadata::run_guardian_metadata_refresh_task;
69use crate::meta::MetaService;
70use crate::module_init::ClientModuleInitRegistry;
71use crate::oplog::OperationLog;
72use crate::sm::executor::Executor;
73use crate::sm::notifier::Notifier;
74
75#[derive(Clone)]
97pub enum RootSecret {
98 StandardDoubleDerive(DerivableSecret),
103 Custom(DerivableSecret),
108}
109
110impl RootSecret {
111 fn to_inner(&self, federation_id: FederationId) -> DerivableSecret {
112 match self {
113 RootSecret::StandardDoubleDerive(derivable_secret) => {
114 get_default_client_secret(derivable_secret, &federation_id)
115 }
116 RootSecret::Custom(derivable_secret) => derivable_secret.clone(),
117 }
118 }
119}
120
121pub struct ClientBuilder {
123 module_inits: ClientModuleInitRegistry,
124 admin_creds: Option<AdminCreds>,
125 meta_service: Arc<crate::meta::MetaService>,
126 stopped: bool,
127 log_event_added_transient_tx: broadcast::Sender<EventLogEntry>,
128 request_hook: ApiRequestHook,
129 iroh_enable_dht: bool,
130 iroh_enable_next: bool,
131 bitcoind_rpc_factory: Option<BitcoindRpcFactory>,
132 bitcoind_rpc_no_chain_id_factory: Option<BitcoindRpcNoChainIdFactory>,
133}
134
135impl ClientBuilder {
136 pub(crate) fn new() -> Self {
137 trace!(
138 target: LOG_CLIENT,
139 version = %fedimint_build_code_version_env!(),
140 "Initializing fedimint client",
141 );
142 let meta_service = MetaService::new(LegacyMetaSource::default());
143 let (log_event_added_transient_tx, _log_event_added_transient_rx) =
144 broadcast::channel(1024);
145
146 ClientBuilder {
147 module_inits: ModuleInitRegistry::new(),
148 admin_creds: None,
149 stopped: false,
150 meta_service,
151 log_event_added_transient_tx,
152 request_hook: Arc::new(|api| api),
153 iroh_enable_dht: true,
154 iroh_enable_next: true,
155 bitcoind_rpc_factory: None,
156 bitcoind_rpc_no_chain_id_factory: None,
157 }
158 }
159
160 pub(crate) fn from_existing(client: &Client) -> Self {
161 ClientBuilder {
162 module_inits: client.module_inits.clone(),
163 admin_creds: None,
164 stopped: false,
165 meta_service: client.meta_service.clone(),
167 log_event_added_transient_tx: client.log_event_added_transient_tx.clone(),
168 request_hook: client.request_hook.clone(),
169 iroh_enable_dht: client.iroh_enable_dht,
170 iroh_enable_next: client.iroh_enable_next,
171 bitcoind_rpc_factory: None,
174 bitcoind_rpc_no_chain_id_factory: client.user_bitcoind_rpc_no_chain_id.clone(),
176 }
177 }
178
179 pub fn with_module_inits(&mut self, module_inits: ClientModuleInitRegistry) {
181 self.module_inits = module_inits;
182 }
183
184 pub fn with_module<M: ClientModuleInit>(&mut self, module_init: M) {
186 self.module_inits.attach(module_init);
187 }
188
189 pub fn stopped(&mut self) {
190 self.stopped = true;
191 }
192 pub fn with_api_request_hook(mut self, hook: ApiRequestHook) -> Self {
201 self.request_hook = hook;
202 self
203 }
204
205 pub fn with_meta_service(&mut self, meta_service: Arc<MetaService>) {
206 self.meta_service = meta_service;
207 }
208
209 pub fn with_iroh_enable_dht(mut self, iroh_enable_dht: bool) -> Self {
212 self.iroh_enable_dht = iroh_enable_dht;
213 self
214 }
215
216 pub fn with_bitcoind_rpc<F, Fut>(mut self, factory: F) -> Self
238 where
239 F: FnOnce(ChainId) -> Fut + Send + Sync + 'static,
240 Fut: Future<Output = Option<DynBitcoindRpc>> + Send + 'static,
241 {
242 self.bitcoind_rpc_factory = Some(Box::new(move |chain_id| Box::pin(factory(chain_id))));
243 self
244 }
245
246 pub fn with_bitcoind_rpc_no_chain_id<F, Fut>(mut self, factory: F) -> Self
269 where
270 F: Fn(SafeUrl) -> Fut + Send + Sync + 'static,
271 Fut: Future<Output = Option<DynBitcoindRpc>> + Send + 'static,
272 {
273 self.bitcoind_rpc_no_chain_id_factory = Some(Arc::new(move |url| Box::pin(factory(url))));
274 self
275 }
276
277 async fn migrate_module_dbs(
284 &self,
285 db: &Database,
286 client_config: &ClientConfig,
287 ) -> Result<(), ClientBuildError> {
288 for (module_id, module_cfg) in &client_config.modules {
289 let kind = module_cfg.kind.clone();
290 let Some(init) = self.module_inits.get(&kind) else {
291 continue;
293 };
294
295 let mut dbtx = db.begin_transaction().await;
296 apply_migrations_client_module_dbtx(
297 &mut dbtx.to_ref_nc(),
298 kind.to_string(),
299 init.get_database_migrations(),
300 *module_id,
301 )
302 .await?;
303 if let Some(used_db_prefixes) = init.used_db_prefixes()
304 && is_running_in_test_env()
305 {
306 verify_module_db_integrity_dbtx(
307 &mut dbtx.to_ref_nc(),
308 *module_id,
309 kind,
310 &used_db_prefixes,
311 )
312 .await;
313 }
314 dbtx.commit_tx_result().await?;
315 }
316
317 Ok(())
318 }
319
320 pub async fn load_existing_config(
321 &self,
322 db: &Database,
323 ) -> Result<ClientConfig, ClientBuildError> {
324 let Some(config) = Client::get_config_from_db(db).await else {
325 return Err(ClientBuildError::DatabaseNotInitialized);
326 };
327
328 Ok(config)
329 }
330
331 pub fn set_admin_creds(&mut self, creds: AdminCreds) {
332 self.admin_creds = Some(creds);
333 }
334
335 #[allow(clippy::too_many_arguments)]
336 async fn init(
337 self,
338 connectors: ConnectorRegistry,
339 db_no_decoders: Database,
340 pre_root_secret: DerivableSecret,
341 config: ClientConfig,
342 api_secret: Option<String>,
343 init_mode: InitMode,
344 preview_prefetch_api_announcements: Option<Jit<Vec<PeersSignedApiAnnouncements>>>,
345 preview_prefetch_api_version_set: Option<
346 Jit<BTreeMap<PeerId, SupportedApiVersionsSummary>>,
347 >,
348 prefetch_chain_id: Option<JitTry<ChainId, FederationError>>,
349 ) -> Result<ClientHandle, ClientBuildError> {
350 if Client::is_initialized(&db_no_decoders).await {
351 return Err(ClientBuildError::DatabaseAlreadyInitialized);
352 }
353
354 Client::run_core_migrations(&db_no_decoders).await?;
355
356 {
359 debug!(target: LOG_CLIENT, "Initializing client database");
360 let mut dbtx = db_no_decoders.begin_transaction().await;
361 dbtx.insert_new_entry(&crate::db::ClientConfigKey, &config)
363 .await;
364 dbtx.insert_entry(
365 &ClientPreRootSecretHashKey,
366 &pre_root_secret.derive_pre_root_secret_hash(),
367 )
368 .await;
369
370 if let Some(api_secret) = api_secret.as_ref() {
371 dbtx.insert_new_entry(&ApiSecretKey, api_secret).await;
372 }
373
374 let init_state = InitState::Pending(init_mode);
375 dbtx.insert_entry(&ClientInitStateKey, &init_state).await;
376
377 let metadata = init_state
378 .does_require_recovery()
379 .flatten()
380 .map_or(Metadata::empty(), |s| s.metadata);
381
382 dbtx.insert_new_entry(&ClientMetadataKey, &metadata).await;
383
384 dbtx.commit_tx_result().await?;
385 }
386
387 let stopped = self.stopped;
388 self.build(
389 connectors,
390 db_no_decoders,
391 pre_root_secret,
392 config,
393 api_secret,
394 stopped,
395 preview_prefetch_api_announcements,
396 preview_prefetch_api_version_set,
397 prefetch_chain_id,
398 )
399 .await
400 }
401
402 pub async fn preview(
403 self,
404 connectors: ConnectorRegistry,
405 invite_code: &InviteCode,
406 ) -> Result<ClientPreview, ClientConfigDownloadError> {
407 let (config, api) = download_from_invite_code(&connectors, invite_code).await?;
408
409 let prefetch_api_announcements =
410 config
411 .global
412 .broadcast_public_keys
413 .clone()
414 .map(|guardian_pub_keys| {
415 Jit::new({
416 let api = api.clone();
417 move || async move {
418 fetch_api_announcements_from_at_least_num_of_peers(
422 1,
423 &api,
424 &guardian_pub_keys,
425 Duration::from_millis(20),
428 )
429 .await
430 }
431 })
432 });
433
434 Ok(self
435 .preview_inner(
436 connectors,
437 config,
438 invite_code.api_secret(),
439 Some(api),
440 prefetch_api_announcements,
441 )
442 .await)
443 }
444
445 pub async fn preview_with_existing_config(
450 self,
451 connectors: ConnectorRegistry,
452 config: ClientConfig,
453 api_secret: Option<String>,
454 ) -> ClientPreview {
455 self.preview_inner(connectors, config, api_secret, None, None)
456 .await
457 }
458
459 async fn preview_inner(
460 self,
461 connectors: ConnectorRegistry,
462 config: ClientConfig,
463 api_secret: Option<String>,
464 prefetch_api: Option<DynGlobalApi>,
465 prefetch_api_announcements: Option<Jit<Vec<PeersSignedApiAnnouncements>>>,
466 ) -> ClientPreview {
467 let preview_prefetch_api_version_set = prefetch_api.as_ref().map(|api| {
468 Jit::new({
469 let config = config.clone();
470 let api = api.clone();
471 || async move { Client::fetch_common_api_versions(&config, &api).await }
472 })
473 });
474
475 let prefetch_chain_id =
476 prefetch_api.map(|api| JitTry::new_try(|| async move { api.chain_id().await }));
477
478 ClientPreview {
479 connectors,
480 inner: self,
481 config,
482 api_secret,
483 prefetch_api_announcements,
484 preview_prefetch_api_version_set,
485 prefetch_chain_id,
486 }
487 }
488
489 pub async fn open(
490 self,
491 connectors: ConnectorRegistry,
492 db_no_decoders: Database,
493 pre_root_secret: RootSecret,
494 ) -> Result<ClientHandle, ClientBuildError> {
495 Client::run_core_migrations(&db_no_decoders).await?;
496
497 Self::migrate_pending_config_if_present(&db_no_decoders).await;
499
500 let Some(config) = Client::get_config_from_db(&db_no_decoders).await else {
501 return Err(ClientBuildError::DatabaseNotInitialized);
502 };
503
504 let pre_root_secret = pre_root_secret.to_inner(config.calculate_federation_id());
505
506 match db_no_decoders
507 .begin_transaction_nc()
508 .await
509 .get_value(&ClientPreRootSecretHashKey)
510 .await
511 {
512 Some(secret_hash) => {
513 if pre_root_secret.derive_pre_root_secret_hash() != secret_hash {
514 return Err(ClientBuildError::SecretMismatch);
515 }
516 }
517 _ => {
518 debug!(target: LOG_CLIENT, "Backfilling secret hash");
519 let mut dbtx = db_no_decoders.begin_transaction().await;
521 dbtx.insert_entry(
522 &ClientPreRootSecretHashKey,
523 &pre_root_secret.derive_pre_root_secret_hash(),
524 )
525 .await;
526 dbtx.commit_tx().await;
527 }
528 }
529
530 let api_secret = Client::get_api_secret_from_db(&db_no_decoders).await;
531 let stopped = self.stopped;
532 let request_hook = self.request_hook.clone();
533
534 let log_event_added_transient_tx = self.log_event_added_transient_tx.clone();
535 let client = self
536 .build_stopped(
537 connectors,
538 db_no_decoders,
539 pre_root_secret,
540 &config,
541 api_secret,
542 log_event_added_transient_tx,
543 request_hook,
544 None,
545 None,
546 None, )
548 .await?;
549 if !stopped {
550 client.as_inner().start_executor();
551 }
552 Ok(client)
553 }
554
555 #[allow(clippy::too_many_arguments)]
557 pub(crate) async fn build(
558 self,
559 connectors: ConnectorRegistry,
560 db_no_decoders: Database,
561 pre_root_secret: DerivableSecret,
562 config: ClientConfig,
563 api_secret: Option<String>,
564 stopped: bool,
565 preview_prefetch_api_announcements: Option<Jit<Vec<PeersSignedApiAnnouncements>>>,
566 preview_prefetch_api_version_set: Option<
567 Jit<BTreeMap<PeerId, SupportedApiVersionsSummary>>,
568 >,
569 prefetch_chain_id: Option<JitTry<ChainId, FederationError>>,
570 ) -> Result<ClientHandle, ClientBuildError> {
571 let log_event_added_transient_tx = self.log_event_added_transient_tx.clone();
572 let request_hook = self.request_hook.clone();
573 let client = self
574 .build_stopped(
575 connectors,
576 db_no_decoders,
577 pre_root_secret,
578 &config,
579 api_secret,
580 log_event_added_transient_tx,
581 request_hook,
582 preview_prefetch_api_announcements,
583 preview_prefetch_api_version_set,
584 prefetch_chain_id,
585 )
586 .await?;
587 if !stopped {
588 client.as_inner().start_executor();
589 }
590
591 Ok(client)
592 }
593
594 fn should_enable_iroh_next(&self, connectors: &ConnectorRegistry) -> bool {
595 self.iroh_enable_next && connectors.iroh_next_enabled()
596 }
597
598 #[allow(clippy::too_many_arguments)]
601 async fn build_stopped(
602 mut self,
603 connectors: ConnectorRegistry,
604 db_no_decoders: Database,
605 pre_root_secret: DerivableSecret,
606 config: &ClientConfig,
607 api_secret: Option<String>,
608 log_event_added_transient_tx: broadcast::Sender<EventLogEntry>,
609 request_hook: ApiRequestHook,
610 preview_prefetch_api_announcements: Option<Jit<Vec<PeersSignedApiAnnouncements>>>,
611 preview_prefetch_api_version_set: Option<
612 Jit<BTreeMap<PeerId, SupportedApiVersionsSummary>>,
613 >,
614 prefetch_chain_id: Option<JitTry<ChainId, FederationError>>,
615 ) -> Result<ClientHandle, ClientBuildError> {
616 debug!(
617 target: LOG_CLIENT,
618 version = %fedimint_build_code_version_env!(),
619 "Building fedimint client",
620 );
621 for (kind, module) in self.module_inits.iter() {
622 debug!(
623 target: LOG_CLIENT,
624 module = %kind,
625 supported_api = %module.supported_api_versions(),
626 "Supported module api versions",
627 );
628 }
629 let (log_event_added_tx, log_event_added_rx) = watch::channel(());
630 let (log_ordering_wakeup_tx, log_ordering_wakeup_rx) = watch::channel(());
631
632 let decoders = self.decoders(config);
633 let config = Self::config_decoded(config, &decoders)?;
634 let fed_id = config.calculate_federation_id();
635 let db = db_no_decoders.with_decoders(decoders.clone());
636 let iroh_enable_next = self.should_enable_iroh_next(&connectors);
637 let peer_urls = get_api_urls(&db, &config, iroh_enable_next).await;
638 let api = match self.admin_creds.as_ref() {
639 Some(admin_creds) => FederationApi::new(
645 connectors.clone(),
646 peer_urls,
647 Some(admin_creds.peer_id),
648 api_secret.as_deref(),
649 )
650 .with_client_ext(db.clone(), log_ordering_wakeup_tx.clone())
651 .with_request_hook(&request_hook)
652 .with_cache()
653 .into(),
654 None => FederationApi::new(connectors.clone(), peer_urls, None, api_secret.as_deref())
655 .with_client_ext(db.clone(), log_ordering_wakeup_tx.clone())
656 .with_request_hook(&request_hook)
657 .with_cache()
658 .into(),
659 };
660
661 let task_group = TaskGroup::new();
662 let client_span = Client::make_client_span(fed_id);
663
664 self.migrate_module_dbs(&db, &config).await?;
667
668 let init_state = Self::load_init_state(&db).await;
669
670 let notifier = Notifier::new();
671
672 if let Some(p) = preview_prefetch_api_announcements {
673 let announcements = p.get().await;
676
677 store_api_announcements_updates_from_peers(&db, announcements).await;
678 }
679
680 if let Some(preview_prefetch_api_version_set) = preview_prefetch_api_version_set {
681 Client::store_prefetched_api_versions(
682 &db,
683 &config,
684 &self.module_inits,
685 preview_prefetch_api_version_set.get().await,
686 )
687 .await;
688 }
689
690 let common_api_versions = Client::load_and_refresh_common_api_version_static(
691 &config,
692 &self.module_inits,
693 connectors.clone(),
694 &api,
695 &db,
696 &task_group,
697 &client_span,
698 )
699 .await
700 .inspect_err(|err| {
701 warn!(target: LOG_CLIENT, err = %err.fmt_compact(), "Failed to discover API version to use.");
702 })
703 .unwrap_or(ApiVersionSet {
704 core: ApiVersion::new(0, 0),
705 modules: BTreeMap::new(),
707 });
708
709 client_span.in_scope(|| {
710 debug!(
711 target: LOG_CLIENT,
712 core = %common_api_versions.core,
713 "Negotiated core API version",
714 );
715 for (module_id, api_version) in &common_api_versions.modules {
716 let kind = config.modules.get(module_id).map(|m| m.kind());
717 let kind_str = kind
718 .as_ref()
719 .map(|k| k.to_string())
720 .unwrap_or_else(|| format!("unknown({module_id})"));
721 let supported = kind
722 .and_then(|k| self.module_inits.get(k))
723 .map(|m| m.supported_api_versions().to_string());
724 debug!(
725 target: LOG_CLIENT,
726 module = %kind_str,
727 api = %api_version,
728 supported = %supported.as_deref().unwrap_or("unknown"),
729 "Negotiated module API version",
730 );
731 }
732 });
733
734 Self::load_and_refresh_client_config_static(&config, &api, &db, &task_group, &client_span);
736
737 if let Some(prefetch_chain_id) = prefetch_chain_id {
741 match prefetch_chain_id.get_try().await {
742 Ok(chain_id) => {
743 debug!(target: LOG_CLIENT, %chain_id, "Caching prefetched chain ID");
744 let mut dbtx = db.begin_transaction().await;
745 dbtx.insert_entry(&ChainIdKey, chain_id).await;
746 dbtx.commit_tx().await;
747 }
748 Err(err) => {
749 debug!(target: LOG_CLIENT, err = %err.fmt_compact(), "Failed to prefetch chain ID, will retry on next start");
750 }
751 }
752 }
753
754 let user_bitcoind_rpc = if let Some(factory) = self.bitcoind_rpc_factory.take() {
756 let chain_id = db.begin_transaction_nc().await.get_value(&ChainIdKey).await;
758
759 if let Some(chain_id) = chain_id {
760 debug!(target: LOG_CLIENT, %chain_id, "Creating user-provided bitcoind RPC client");
761 factory(chain_id).await
762 } else {
763 debug!(target: LOG_CLIENT, "Chain ID not available, skipping user-provided bitcoind RPC creation");
764 None
765 }
766 } else {
767 None
768 };
769
770 let mut module_recoveries: BTreeMap<ModuleInstanceId, ModuleRecoveryFuture> =
771 BTreeMap::new();
772 let mut module_recovery_progress_receivers: BTreeMap<
773 ModuleInstanceId,
774 watch::Receiver<RecoveryProgress>,
775 > = BTreeMap::new();
776
777 let final_client = FinalClientIface::default();
778
779 let root_secret = Self::federation_root_secret(&pre_root_secret, &config);
780
781 let modules = {
782 let mut modules = ClientModuleRegistry::default();
783 for (module_instance_id, module_config) in config.modules.clone() {
784 let kind = module_config.kind().clone();
785 let Some(module_init) = self.module_inits.get(&kind).cloned() else {
786 client_span.in_scope(|| {
787 debug!(
788 target: LOG_CLIENT,
789 kind=%kind,
790 instance_id=%module_instance_id,
791 "Module kind of instance not found in module gens, skipping");
792 });
793 continue;
794 };
795
796 let Some(&api_version) = common_api_versions.modules.get(&module_instance_id)
797 else {
798 client_span.in_scope(|| {
799 warn!(
800 target: LOG_CLIENT,
801 kind=%kind,
802 instance_id=%module_instance_id,
803 "Module kind of instance has incompatible api version, skipping"
804 );
805 });
806 continue;
807 };
808
809 let start_module_recover_fn =
812 |snapshot: Option<ClientBackup>, progress: RecoveryProgress| {
813 let module_config = module_config.clone();
814 let num_peers = NumPeers::from(config.global.api_endpoints.len());
815 let db = db.clone();
816 let kind = kind.clone();
817 let notifier = notifier.clone();
818 let api = api.clone();
819 let root_secret = root_secret.clone();
820 let admin_auth = self.admin_creds.as_ref().map(|creds| creds.auth.clone());
821 let final_client = final_client.clone();
822 let (progress_tx, progress_rx) = tokio::sync::watch::channel(progress);
823 let task_group = task_group.clone();
824 let module_init = module_init.clone();
825 let user_bitcoind_rpc = user_bitcoind_rpc.clone();
826 let user_bitcoind_rpc_no_chain_id =
827 self.bitcoind_rpc_no_chain_id_factory.clone();
828 let client_span = client_span.clone();
829 (
830 Box::pin(async move {
831 module_init
832 .recover(
833 final_client.clone(),
834 fed_id,
835 num_peers,
836 module_config.clone(),
837 db.clone(),
838 module_instance_id,
839 common_api_versions.core,
840 api_version,
841 root_secret.derive_module_secret(module_instance_id),
842 notifier.clone(),
843 api.clone(),
844 admin_auth,
845 snapshot
846 .as_ref()
847 .and_then(|s| s.modules.get(&module_instance_id)),
848 progress_tx,
849 task_group,
850 client_span,
851 user_bitcoind_rpc,
852 user_bitcoind_rpc_no_chain_id,
853 )
854 .await
855 .inspect_err(|err| {
856 warn!(
857 target: LOG_CLIENT,
858 module_id = module_instance_id,
859 %kind,
860 err = %err.fmt_compact(),
861 "Module failed to recover"
862 );
863 })
864 }),
865 progress_rx,
866 )
867 };
868
869 let recovery_mode = module_init.recovery_mode();
874
875 let requires_recovery = init_state
876 .does_require_recovery()
877 .filter(|_| recovery_mode != RecoveryMode::None);
878
879 if requires_recovery.is_some() && recovery_mode == RecoveryMode::Usable {
884 module_init
885 .prepare_recovery(db.clone(), module_instance_id, api.clone())
886 .await
887 .map_err(|source| ClientBuildError::ModuleRecoveryPrepare {
888 kind: kind.clone(),
889 instance_id: module_instance_id,
890 source,
891 })?;
892 }
893
894 let recovery = match requires_recovery {
895 Some(snapshot) => {
896 match db
897 .begin_transaction_nc()
898 .await
899 .get_value(&ClientModuleRecovery { module_instance_id })
900 .await
901 {
902 Some(module_recovery_state) => {
903 if module_recovery_state.is_done() {
904 debug!(
905 id = %module_instance_id,
906 %kind, "Module recovery already complete"
907 );
908 None
909 } else {
910 debug!(
911 id = %module_instance_id,
912 %kind,
913 progress = %module_recovery_state.progress,
914 "Starting module recovery with an existing progress"
915 );
916 Some(start_module_recover_fn(
917 snapshot,
918 module_recovery_state.progress,
919 ))
920 }
921 }
922 _ => {
923 let progress = RecoveryProgress::none();
924 let mut dbtx = db.begin_transaction().await;
925 dbtx.log_event(
926 log_ordering_wakeup_tx.clone(),
927 None,
928 ModuleRecoveryStarted::new(module_instance_id),
929 )
930 .await;
931 dbtx.insert_entry(
932 &ClientModuleRecovery { module_instance_id },
933 &ClientModuleRecoveryState { progress },
934 )
935 .await;
936
937 dbtx.commit_tx().await;
938
939 debug!(
940 id = %module_instance_id,
941 %kind, "Starting new module recovery"
942 );
943 Some(start_module_recover_fn(snapshot, progress))
944 }
945 }
946 }
947 _ => None,
948 };
949
950 let initialize_module = recovery.is_none() || recovery_mode == RecoveryMode::Usable;
956
957 if let Some((recovery, recovery_progress_rx)) = recovery {
958 module_recoveries.insert(module_instance_id, recovery);
959 module_recovery_progress_receivers
960 .insert(module_instance_id, recovery_progress_rx);
961 }
962
963 if initialize_module {
964 let module = module_init
965 .init(
966 final_client.clone(),
967 fed_id,
968 config.global.api_endpoints.len(),
969 module_config,
970 db.clone(),
971 module_instance_id,
972 common_api_versions.core,
973 api_version,
974 root_secret.derive_module_secret(module_instance_id),
981 notifier.clone(),
982 api.clone(),
983 self.admin_creds.as_ref().map(|cred| cred.auth.clone()),
984 task_group.clone(),
985 client_span.clone(),
986 connectors.clone(),
987 user_bitcoind_rpc.clone(),
988 self.bitcoind_rpc_no_chain_id_factory.clone(),
989 )
990 .await
991 .map_err(|source| ClientBuildError::ModuleInit {
992 kind: kind.clone(),
993 instance_id: module_instance_id,
994 source,
995 })?;
996
997 modules.register_module(module_instance_id, kind, module);
998 }
999 }
1000 modules
1001 };
1002
1003 if init_state.is_pending() && module_recoveries.is_empty() {
1004 let mut dbtx = db.begin_transaction().await;
1005 dbtx.insert_entry(&ClientInitStateKey, &init_state.into_complete())
1006 .await;
1007 dbtx.commit_tx().await;
1008 }
1009
1010 let mut primary_modules: BTreeMap<PrimaryModulePriority, PrimaryModuleCandidates> =
1011 BTreeMap::new();
1012
1013 for (module_id, _kind, module) in modules.iter_modules() {
1014 match module.supports_being_primary() {
1015 PrimaryModuleSupport::Any { priority } => {
1016 primary_modules
1017 .entry(priority)
1018 .or_default()
1019 .wildcard
1020 .push(module_id);
1021 }
1022 PrimaryModuleSupport::Selected { priority, units } => {
1023 for unit in units {
1024 primary_modules
1025 .entry(priority)
1026 .or_default()
1027 .specific
1028 .entry(unit)
1029 .or_default()
1030 .push(module_id);
1031 }
1032 }
1033 PrimaryModuleSupport::None => {}
1034 }
1035 }
1036
1037 let executor = client_span.in_scope(|| {
1038 let mut executor_builder = Executor::builder();
1039 executor_builder
1040 .with_module(TRANSACTION_SUBMISSION_MODULE_INSTANCE, TxSubmissionContext);
1041
1042 for (module_instance_id, _, module) in modules.iter_modules() {
1043 executor_builder.with_module_dyn(module.context(module_instance_id));
1044 }
1045
1046 for module_instance_id in module_recoveries.keys() {
1047 executor_builder.with_valid_module_id(*module_instance_id);
1048 }
1049
1050 executor_builder.build(
1051 db.clone(),
1052 notifier,
1053 task_group.clone(),
1054 log_ordering_wakeup_tx.clone(),
1055 )
1056 });
1057
1058 let recovery_receiver_init_val = module_recovery_progress_receivers
1059 .iter()
1060 .map(|(module_instance_id, rx)| {
1061 (
1062 *module_instance_id,
1063 RecoveryStatus::InProgress(*rx.borrow()),
1064 )
1065 })
1066 .collect::<BTreeMap<_, _>>();
1067 let (client_recovery_status_sender, client_recovery_status_receiver) =
1068 watch::channel(recovery_receiver_init_val);
1069
1070 let client_inner = Arc::new(Client {
1071 final_client: final_client.clone(),
1072 config: tokio::sync::RwLock::new(config.clone()),
1073 api_secret,
1074 decoders,
1075 db: db.clone(),
1076 connectors,
1077 federation_id: fed_id,
1078 federation_config_meta: config.global.meta,
1079 primary_modules,
1080 modules,
1081 module_inits: self.module_inits.clone(),
1082 log_ordering_wakeup_tx,
1083 log_event_added_rx,
1084 log_event_added_transient_tx: log_event_added_transient_tx.clone(),
1085 request_hook,
1086 executor,
1087 api,
1088 secp_ctx: Secp256k1::new(),
1089 root_secret,
1090 task_group,
1091 client_span,
1092 operation_log: OperationLog::new(db.clone()),
1093 client_recovery_status_receiver,
1094 meta_service: self.meta_service,
1095 iroh_enable_dht: self.iroh_enable_dht,
1096 iroh_enable_next,
1097 user_bitcoind_rpc,
1098 user_bitcoind_rpc_no_chain_id: self.bitcoind_rpc_no_chain_id_factory,
1099 });
1100 client_inner.spawn_cancellable("MetaService::update_continuously", {
1101 let client_inner = client_inner.clone();
1102 async move {
1103 client_inner
1104 .meta_service
1105 .update_continuously(&client_inner)
1106 .await;
1107 }
1108 });
1109
1110 client_inner.spawn_cancellable("update-api-announcements", {
1111 let client_inner = client_inner.clone();
1112 async move {
1113 client_inner
1114 .connectors
1115 .wait_for_initialized_connections()
1116 .await;
1117 run_api_announcement_refresh_task(client_inner.clone()).await
1118 }
1119 });
1120
1121 client_inner.spawn_cancellable("guardian metadata refresh task", {
1122 let client_inner = client_inner.clone();
1123 async move {
1124 client_inner
1125 .connectors
1126 .wait_for_initialized_connections()
1127 .await;
1128 run_guardian_metadata_refresh_task(client_inner.clone()).await
1129 }
1130 });
1131
1132 client_inner.spawn_cancellable("event log ordering task", {
1133 let client_inner = client_inner.clone();
1134 async move {
1135 client_inner
1136 .connectors
1137 .wait_for_initialized_connections()
1138 .await;
1139
1140 run_event_log_ordering_task(
1141 db.clone(),
1142 log_ordering_wakeup_rx,
1143 log_event_added_tx,
1144 log_event_added_transient_tx,
1145 )
1146 .await
1147 }
1148 });
1149
1150 if client_inner
1154 .db
1155 .begin_transaction_nc()
1156 .await
1157 .get_value(&ChainIdKey)
1158 .await
1159 .is_none()
1160 {
1161 client_inner.spawn_cancellable("fetch-chain-id", {
1162 let client_inner = client_inner.clone();
1163 async move {
1164 client_inner.api.wait_for_initialized_connections().await;
1165 match client_inner.api.chain_id().await {
1166 Ok(chain_id) => {
1167 debug!(target: LOG_CLIENT, %chain_id, "Caching chain ID from background fetch");
1168 let mut dbtx = client_inner.db.begin_transaction().await;
1169 dbtx.insert_entry(&ChainIdKey, &chain_id).await;
1170 dbtx.commit_tx().await;
1171 }
1172 Err(err) => {
1173 debug!(target: LOG_CLIENT, err = %err.fmt_compact(), "Background chain ID fetch failed, will retry on next start");
1174 }
1175 }
1176 }
1177 });
1178 }
1179
1180 let client_iface = std::sync::Arc::<Client>::downgrade(&client_inner);
1181
1182 let client_arc = ClientHandle::new(client_inner);
1183
1184 for (_, _, module) in client_arc.modules.iter_modules() {
1185 module.start().await;
1186 }
1187
1188 final_client.set(client_iface.clone());
1189
1190 if !module_recoveries.is_empty() {
1191 let module_kinds = client_arc
1195 .config()
1196 .await
1197 .modules
1198 .iter()
1199 .map(|(id, module_config)| (*id, module_config.kind().clone()))
1200 .collect();
1201 client_arc.spawn_module_recoveries_task(
1202 client_recovery_status_sender,
1203 module_recoveries,
1204 module_recovery_progress_receivers,
1205 module_kinds,
1206 );
1207 }
1208
1209 Ok(client_arc)
1210 }
1211
1212 async fn load_init_state(db: &Database) -> InitState {
1213 let mut dbtx = db.begin_transaction_nc().await;
1214 dbtx.get_value(&ClientInitStateKey)
1215 .await
1216 .unwrap_or_else(|| {
1217 warn!(
1220 target: LOG_CLIENT,
1221 "Client missing ClientRequiresRecovery: assuming complete"
1222 );
1223 db::InitState::Complete(db::InitModeComplete::Fresh)
1224 })
1225 }
1226
1227 fn decoders(&self, config: &ClientConfig) -> ModuleDecoderRegistry {
1228 let mut decoders = client_decoders(
1229 &self.module_inits,
1230 config
1231 .modules
1232 .iter()
1233 .map(|(module_instance, module_config)| (*module_instance, module_config.kind())),
1234 );
1235
1236 decoders.register_module(
1237 TRANSACTION_SUBMISSION_MODULE_INSTANCE,
1238 ModuleKind::from_static_str("tx_submission"),
1239 tx_submission_sm_decoder(),
1240 );
1241
1242 decoders
1243 }
1244
1245 fn config_decoded(
1246 config: &ClientConfig,
1247 decoders: &ModuleDecoderRegistry,
1248 ) -> Result<ClientConfig, fedimint_core::encoding::DecodeError> {
1249 config.clone().redecode_raw(decoders)
1250 }
1251
1252 fn federation_root_secret(
1256 pre_root_secret: &DerivableSecret,
1257 config: &ClientConfig,
1258 ) -> DerivableSecret {
1259 pre_root_secret.federation_key(&config.global.calculate_federation_id())
1260 }
1261
1262 pub fn get_event_log_transient_receiver(&self) -> broadcast::Receiver<EventLogEntry> {
1264 self.log_event_added_transient_tx.subscribe()
1265 }
1266
1267 async fn migrate_pending_config_if_present(db: &Database) {
1271 if let Some(pending_config) = Client::get_pending_config_from_db(db).await {
1272 debug!(target: LOG_CLIENT, "Found pending client config, migrating to current config");
1273
1274 let mut dbtx = db.begin_transaction().await;
1275 dbtx.insert_entry(&crate::db::ClientConfigKey, &pending_config)
1277 .await;
1278 dbtx.remove_entry(&PendingClientConfigKey).await;
1280 dbtx.commit_tx().await;
1281
1282 debug!(target: LOG_CLIENT, "Successfully migrated pending config to current config");
1283 }
1284 }
1285
1286 fn load_and_refresh_client_config_static(
1289 config: &ClientConfig,
1290 api: &DynGlobalApi,
1291 db: &Database,
1292 task_group: &TaskGroup,
1293 client_span: &Span,
1294 ) {
1295 let config = config.clone();
1296 let api = api.clone();
1297 let db = db.clone();
1298 let task_group = task_group.clone();
1299
1300 task_group.spawn_cancellable_with_span(
1302 client_span.clone(),
1303 "refresh_client_config_static",
1304 async move {
1305 api.wait_for_initialized_connections().await;
1306 Self::refresh_client_config_static(&config, &api, &db).await;
1307 },
1308 );
1309 }
1310
1311 async fn refresh_client_config_static(
1313 config: &ClientConfig,
1314 api: &DynGlobalApi,
1315 db: &Database,
1316 ) {
1317 if let Err(error) = Self::refresh_client_config_static_try(config, api, db).await {
1318 warn!(
1319 target: LOG_CLIENT,
1320 err = %error.fmt_compact(), "Failed to refresh client config"
1321 );
1322 }
1323 }
1324
1325 fn validate_config_update(
1327 current_config: &ClientConfig,
1328 new_config: &ClientConfig,
1329 ) -> Result<(), ConfigUpdateError> {
1330 if current_config.global != new_config.global {
1332 return Err(ConfigUpdateError::GlobalChanged);
1333 }
1334
1335 for (module_id, current_module_config) in ¤t_config.modules {
1337 match new_config.modules.get(module_id) {
1338 Some(new_module_config) => {
1339 if current_module_config != new_module_config {
1340 return Err(ConfigUpdateError::ModuleChanged {
1341 module_id: *module_id,
1342 });
1343 }
1344 }
1345 None => {
1346 return Err(ConfigUpdateError::ModuleRemoved {
1347 module_id: *module_id,
1348 });
1349 }
1350 }
1351 }
1352
1353 Ok(())
1354 }
1355
1356 async fn refresh_client_config_static_try(
1358 current_config: &ClientConfig,
1359 api: &DynGlobalApi,
1360 db: &Database,
1361 ) -> Result<(), RefreshClientConfigError> {
1362 debug!(target: LOG_CLIENT, "Refreshing client config");
1363
1364 let fetched_config = api
1366 .request_current_consensus::<ClientConfig>(
1367 CLIENT_CONFIG_ENDPOINT.to_owned(),
1368 ApiRequestErased::default(),
1369 )
1370 .await?;
1371
1372 Self::validate_config_update(current_config, &fetched_config)?;
1374
1375 if current_config != &fetched_config {
1377 debug!(target: LOG_CLIENT, "Detected federation config change, saving as pending config");
1378
1379 let mut dbtx = db.begin_transaction().await;
1380 dbtx.insert_entry(&PendingClientConfigKey, &fetched_config)
1381 .await;
1382 dbtx.commit_tx().await;
1383 } else {
1384 debug!(target: LOG_CLIENT, "No federation config changes detected");
1385 }
1386
1387 Ok(())
1388 }
1389}
1390
1391pub struct ClientPreview {
1396 inner: ClientBuilder,
1397 config: ClientConfig,
1398 connectors: ConnectorRegistry,
1399 api_secret: Option<String>,
1400 prefetch_api_announcements: Option<Jit<Vec<PeersSignedApiAnnouncements>>>,
1401 preview_prefetch_api_version_set: Option<Jit<BTreeMap<PeerId, SupportedApiVersionsSummary>>>,
1402 prefetch_chain_id: Option<JitTry<ChainId, FederationError>>,
1403}
1404
1405impl ClientPreview {
1406 pub fn config(&self) -> &ClientConfig {
1408 &self.config
1409 }
1410
1411 pub async fn join(
1489 self,
1490 db_no_decoders: Database,
1491 pre_root_secret: RootSecret,
1492 ) -> Result<ClientHandle, ClientBuildError> {
1493 let pre_root_secret = pre_root_secret.to_inner(self.config.calculate_federation_id());
1494
1495 let client = self
1496 .inner
1497 .init(
1498 self.connectors,
1499 db_no_decoders,
1500 pre_root_secret,
1501 self.config,
1502 self.api_secret,
1503 InitMode::Fresh,
1504 self.prefetch_api_announcements,
1505 self.preview_prefetch_api_version_set,
1506 self.prefetch_chain_id,
1507 )
1508 .await?;
1509
1510 Ok(client)
1511 }
1512
1513 pub async fn recover(
1525 self,
1526 db_no_decoders: Database,
1527 pre_root_secret: RootSecret,
1528 backup: Option<ClientBackup>,
1529 ) -> Result<ClientHandle, ClientBuildError> {
1530 let pre_root_secret = pre_root_secret.to_inner(self.config.calculate_federation_id());
1531
1532 let client = self
1533 .inner
1534 .init(
1535 self.connectors,
1536 db_no_decoders,
1537 pre_root_secret,
1538 self.config,
1539 self.api_secret,
1540 InitMode::Recover {
1541 snapshot: backup.clone(),
1542 },
1543 self.prefetch_api_announcements,
1544 self.preview_prefetch_api_version_set,
1545 self.prefetch_chain_id,
1546 )
1547 .await?;
1548
1549 Ok(client)
1550 }
1551
1552 #[deprecated(
1554 note = "Recovery is now efficient enough that backups are no longer necessary. Backups will be removed in v0.13.0 due to backups being inherently complicated and brittle."
1555 )]
1556 #[allow(deprecated)]
1557 pub async fn download_backup_from_federation(
1558 &self,
1559 pre_root_secret: RootSecret,
1560 ) -> Result<Option<ClientBackup>, FederationError> {
1561 let pre_root_secret = pre_root_secret.to_inner(self.config.calculate_federation_id());
1562 let api = DynGlobalApi::new(
1563 self.connectors.clone(),
1564 self.config
1566 .global
1567 .api_endpoints
1568 .iter()
1569 .map(|(peer_id, peer_url)| (*peer_id, peer_url.url.clone()))
1570 .collect(),
1571 self.api_secret.as_deref(),
1572 );
1573
1574 Client::download_backup_from_federation_static(
1575 &api,
1576 &ClientBuilder::federation_root_secret(&pre_root_secret, &self.config),
1577 &self.inner.decoders(&self.config),
1578 )
1579 .await
1580 }
1581}
1582
1583#[derive(Debug, thiserror::Error)]
1585enum RefreshClientConfigError {
1586 #[error(transparent)]
1588 Federation(#[from] FederationError),
1589
1590 #[error(transparent)]
1592 InvalidUpdate(#[from] ConfigUpdateError),
1593}
1594
1595#[derive(Debug, thiserror::Error)]
1597enum ConfigUpdateError {
1598 #[error("Global configuration changes are not allowed in config updates")]
1600 GlobalChanged,
1601
1602 #[error(
1604 "Module {module_id} configuration changes are not allowed, only additions are permitted"
1605 )]
1606 ModuleChanged { module_id: ModuleInstanceId },
1607
1608 #[error("Module {module_id} was removed in new config, only additions are allowed")]
1610 ModuleRemoved { module_id: ModuleInstanceId },
1611}