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