1use std::collections::BTreeMap;
2use std::future::Future;
3use std::sync::Arc;
4use std::time::Duration;
5
6use anyhow::{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,
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 = match init_state.does_require_recovery() {
868 Some(snapshot) => {
869 match db
870 .begin_transaction_nc()
871 .await
872 .get_value(&ClientModuleRecovery { module_instance_id })
873 .await
874 {
875 Some(module_recovery_state) => {
876 if module_recovery_state.is_done() {
877 debug!(
878 id = %module_instance_id,
879 %kind, "Module recovery already complete"
880 );
881 None
882 } else {
883 debug!(
884 id = %module_instance_id,
885 %kind,
886 progress = %module_recovery_state.progress,
887 "Starting module recovery with an existing progress"
888 );
889 Some(start_module_recover_fn(
890 snapshot,
891 module_recovery_state.progress,
892 ))
893 }
894 }
895 _ => {
896 let progress = RecoveryProgress::none();
897 let mut dbtx = db.begin_transaction().await;
898 dbtx.log_event(
899 log_ordering_wakeup_tx.clone(),
900 None,
901 ModuleRecoveryStarted::new(module_instance_id),
902 )
903 .await;
904 dbtx.insert_entry(
905 &ClientModuleRecovery { module_instance_id },
906 &ClientModuleRecoveryState { progress },
907 )
908 .await;
909
910 dbtx.commit_tx().await;
911
912 debug!(
913 id = %module_instance_id,
914 %kind, "Starting new module recovery"
915 );
916 Some(start_module_recover_fn(snapshot, progress))
917 }
918 }
919 }
920 _ => None,
921 };
922
923 match recovery {
924 Some((recovery, recovery_progress_rx)) => {
925 module_recoveries.insert(module_instance_id, recovery);
926 module_recovery_progress_receivers
927 .insert(module_instance_id, recovery_progress_rx);
928 }
929 _ => {
930 let module = module_init
931 .init(
932 final_client.clone(),
933 fed_id,
934 config.global.api_endpoints.len(),
935 module_config,
936 db.clone(),
937 module_instance_id,
938 common_api_versions.core,
939 api_version,
940 root_secret.derive_module_secret(module_instance_id),
947 notifier.clone(),
948 api.clone(),
949 self.admin_creds.as_ref().map(|cred| cred.auth.clone()),
950 task_group.clone(),
951 client_span.clone(),
952 connectors.clone(),
953 user_bitcoind_rpc.clone(),
954 self.bitcoind_rpc_no_chain_id_factory.clone(),
955 )
956 .await?;
957
958 modules.register_module(module_instance_id, kind, module);
959 }
960 }
961 }
962 modules
963 };
964
965 if init_state.is_pending() && module_recoveries.is_empty() {
966 let mut dbtx = db.begin_transaction().await;
967 dbtx.insert_entry(&ClientInitStateKey, &init_state.into_complete())
968 .await;
969 dbtx.commit_tx().await;
970 }
971
972 let mut primary_modules: BTreeMap<PrimaryModulePriority, PrimaryModuleCandidates> =
973 BTreeMap::new();
974
975 for (module_id, _kind, module) in modules.iter_modules() {
976 match module.supports_being_primary() {
977 PrimaryModuleSupport::Any { priority } => {
978 primary_modules
979 .entry(priority)
980 .or_default()
981 .wildcard
982 .push(module_id);
983 }
984 PrimaryModuleSupport::Selected { priority, units } => {
985 for unit in units {
986 primary_modules
987 .entry(priority)
988 .or_default()
989 .specific
990 .entry(unit)
991 .or_default()
992 .push(module_id);
993 }
994 }
995 PrimaryModuleSupport::None => {}
996 }
997 }
998
999 let executor = client_span.in_scope(|| {
1000 let mut executor_builder = Executor::builder();
1001 executor_builder
1002 .with_module(TRANSACTION_SUBMISSION_MODULE_INSTANCE, TxSubmissionContext);
1003
1004 for (module_instance_id, _, module) in modules.iter_modules() {
1005 executor_builder.with_module_dyn(module.context(module_instance_id));
1006 }
1007
1008 for module_instance_id in module_recoveries.keys() {
1009 executor_builder.with_valid_module_id(*module_instance_id);
1010 }
1011
1012 executor_builder.build(
1013 db.clone(),
1014 notifier,
1015 task_group.clone(),
1016 log_ordering_wakeup_tx.clone(),
1017 )
1018 });
1019
1020 let recovery_receiver_init_val = module_recovery_progress_receivers
1021 .iter()
1022 .map(|(module_instance_id, rx)| {
1023 (
1024 *module_instance_id,
1025 RecoveryStatus::InProgress(*rx.borrow()),
1026 )
1027 })
1028 .collect::<BTreeMap<_, _>>();
1029 let (client_recovery_status_sender, client_recovery_status_receiver) =
1030 watch::channel(recovery_receiver_init_val);
1031
1032 let client_inner = Arc::new(Client {
1033 final_client: final_client.clone(),
1034 config: tokio::sync::RwLock::new(config.clone()),
1035 api_secret,
1036 decoders,
1037 db: db.clone(),
1038 connectors,
1039 federation_id: fed_id,
1040 federation_config_meta: config.global.meta,
1041 primary_modules,
1042 modules,
1043 module_inits: self.module_inits.clone(),
1044 log_ordering_wakeup_tx,
1045 log_event_added_rx,
1046 log_event_added_transient_tx: log_event_added_transient_tx.clone(),
1047 request_hook,
1048 executor,
1049 api,
1050 secp_ctx: Secp256k1::new(),
1051 root_secret,
1052 task_group,
1053 client_span,
1054 operation_log: OperationLog::new(db.clone()),
1055 client_recovery_status_receiver,
1056 meta_service: self.meta_service,
1057 iroh_enable_dht: self.iroh_enable_dht,
1058 iroh_enable_next,
1059 user_bitcoind_rpc,
1060 user_bitcoind_rpc_no_chain_id: self.bitcoind_rpc_no_chain_id_factory,
1061 });
1062 client_inner.spawn_cancellable("MetaService::update_continuously", {
1063 let client_inner = client_inner.clone();
1064 async move {
1065 client_inner
1066 .meta_service
1067 .update_continuously(&client_inner)
1068 .await;
1069 }
1070 });
1071
1072 client_inner.spawn_cancellable("update-api-announcements", {
1073 let client_inner = client_inner.clone();
1074 async move {
1075 client_inner
1076 .connectors
1077 .wait_for_initialized_connections()
1078 .await;
1079 run_api_announcement_refresh_task(client_inner.clone()).await
1080 }
1081 });
1082
1083 client_inner.spawn_cancellable("guardian metadata refresh task", {
1084 let client_inner = client_inner.clone();
1085 async move {
1086 client_inner
1087 .connectors
1088 .wait_for_initialized_connections()
1089 .await;
1090 run_guardian_metadata_refresh_task(client_inner.clone()).await
1091 }
1092 });
1093
1094 client_inner.spawn_cancellable("event log ordering task", {
1095 let client_inner = client_inner.clone();
1096 async move {
1097 client_inner
1098 .connectors
1099 .wait_for_initialized_connections()
1100 .await;
1101
1102 run_event_log_ordering_task(
1103 db.clone(),
1104 log_ordering_wakeup_rx,
1105 log_event_added_tx,
1106 log_event_added_transient_tx,
1107 )
1108 .await
1109 }
1110 });
1111
1112 if client_inner
1116 .db
1117 .begin_transaction_nc()
1118 .await
1119 .get_value(&ChainIdKey)
1120 .await
1121 .is_none()
1122 {
1123 client_inner.spawn_cancellable("fetch-chain-id", {
1124 let client_inner = client_inner.clone();
1125 async move {
1126 client_inner.api.wait_for_initialized_connections().await;
1127 match client_inner.api.chain_id().await {
1128 Ok(chain_id) => {
1129 debug!(target: LOG_CLIENT, %chain_id, "Caching chain ID from background fetch");
1130 let mut dbtx = client_inner.db.begin_transaction().await;
1131 dbtx.insert_entry(&ChainIdKey, &chain_id).await;
1132 dbtx.commit_tx().await;
1133 }
1134 Err(err) => {
1135 debug!(target: LOG_CLIENT, err = %err.fmt_compact(), "Background chain ID fetch failed, will retry on next start");
1136 }
1137 }
1138 }
1139 });
1140 }
1141
1142 let client_iface = std::sync::Arc::<Client>::downgrade(&client_inner);
1143
1144 let client_arc = ClientHandle::new(client_inner);
1145
1146 for (_, _, module) in client_arc.modules.iter_modules() {
1147 module.start().await;
1148 }
1149
1150 final_client.set(client_iface.clone());
1151
1152 if !module_recoveries.is_empty() {
1153 let module_kinds = client_arc
1157 .config()
1158 .await
1159 .modules
1160 .iter()
1161 .map(|(id, module_config)| (*id, module_config.kind().clone()))
1162 .collect();
1163 client_arc.spawn_module_recoveries_task(
1164 client_recovery_status_sender,
1165 module_recoveries,
1166 module_recovery_progress_receivers,
1167 module_kinds,
1168 );
1169 }
1170
1171 Ok(client_arc)
1172 }
1173
1174 async fn load_init_state(db: &Database) -> InitState {
1175 let mut dbtx = db.begin_transaction_nc().await;
1176 dbtx.get_value(&ClientInitStateKey)
1177 .await
1178 .unwrap_or_else(|| {
1179 warn!(
1182 target: LOG_CLIENT,
1183 "Client missing ClientRequiresRecovery: assuming complete"
1184 );
1185 db::InitState::Complete(db::InitModeComplete::Fresh)
1186 })
1187 }
1188
1189 fn decoders(&self, config: &ClientConfig) -> ModuleDecoderRegistry {
1190 let mut decoders = client_decoders(
1191 &self.module_inits,
1192 config
1193 .modules
1194 .iter()
1195 .map(|(module_instance, module_config)| (*module_instance, module_config.kind())),
1196 );
1197
1198 decoders.register_module(
1199 TRANSACTION_SUBMISSION_MODULE_INSTANCE,
1200 ModuleKind::from_static_str("tx_submission"),
1201 tx_submission_sm_decoder(),
1202 );
1203
1204 decoders
1205 }
1206
1207 fn config_decoded(
1208 config: &ClientConfig,
1209 decoders: &ModuleDecoderRegistry,
1210 ) -> Result<ClientConfig, fedimint_core::encoding::DecodeError> {
1211 config.clone().redecode_raw(decoders)
1212 }
1213
1214 fn federation_root_secret(
1218 pre_root_secret: &DerivableSecret,
1219 config: &ClientConfig,
1220 ) -> DerivableSecret {
1221 pre_root_secret.federation_key(&config.global.calculate_federation_id())
1222 }
1223
1224 pub fn get_event_log_transient_receiver(&self) -> broadcast::Receiver<EventLogEntry> {
1226 self.log_event_added_transient_tx.subscribe()
1227 }
1228
1229 async fn migrate_pending_config_if_present(db: &Database) {
1233 if let Some(pending_config) = Client::get_pending_config_from_db(db).await {
1234 debug!(target: LOG_CLIENT, "Found pending client config, migrating to current config");
1235
1236 let mut dbtx = db.begin_transaction().await;
1237 dbtx.insert_entry(&crate::db::ClientConfigKey, &pending_config)
1239 .await;
1240 dbtx.remove_entry(&PendingClientConfigKey).await;
1242 dbtx.commit_tx().await;
1243
1244 debug!(target: LOG_CLIENT, "Successfully migrated pending config to current config");
1245 }
1246 }
1247
1248 fn load_and_refresh_client_config_static(
1251 config: &ClientConfig,
1252 api: &DynGlobalApi,
1253 db: &Database,
1254 task_group: &TaskGroup,
1255 client_span: &Span,
1256 ) {
1257 let config = config.clone();
1258 let api = api.clone();
1259 let db = db.clone();
1260 let task_group = task_group.clone();
1261
1262 task_group.spawn_cancellable_with_span(
1264 client_span.clone(),
1265 "refresh_client_config_static",
1266 async move {
1267 api.wait_for_initialized_connections().await;
1268 Self::refresh_client_config_static(&config, &api, &db).await;
1269 },
1270 );
1271 }
1272
1273 async fn refresh_client_config_static(
1275 config: &ClientConfig,
1276 api: &DynGlobalApi,
1277 db: &Database,
1278 ) {
1279 if let Err(error) = Self::refresh_client_config_static_try(config, api, db).await {
1280 warn!(
1281 target: LOG_CLIENT,
1282 err = %error.fmt_compact_anyhow(), "Failed to refresh client config"
1283 );
1284 }
1285 }
1286
1287 fn validate_config_update(
1289 current_config: &ClientConfig,
1290 new_config: &ClientConfig,
1291 ) -> anyhow::Result<()> {
1292 if current_config.global != new_config.global {
1294 bail!("Global configuration changes are not allowed in config updates");
1295 }
1296
1297 for (module_id, current_module_config) in ¤t_config.modules {
1299 match new_config.modules.get(module_id) {
1300 Some(new_module_config) => {
1301 if current_module_config != new_module_config {
1302 bail!(
1303 "Module {} configuration changes are not allowed, only additions are permitted",
1304 module_id
1305 );
1306 }
1307 }
1308 None => {
1309 bail!(
1310 "Module {} was removed in new config, only additions are allowed",
1311 module_id
1312 );
1313 }
1314 }
1315 }
1316
1317 Ok(())
1318 }
1319
1320 async fn refresh_client_config_static_try(
1322 current_config: &ClientConfig,
1323 api: &DynGlobalApi,
1324 db: &Database,
1325 ) -> anyhow::Result<()> {
1326 debug!(target: LOG_CLIENT, "Refreshing client config");
1327
1328 let fetched_config = api
1330 .request_current_consensus::<ClientConfig>(
1331 CLIENT_CONFIG_ENDPOINT.to_owned(),
1332 ApiRequestErased::default(),
1333 )
1334 .await?;
1335
1336 Self::validate_config_update(current_config, &fetched_config)?;
1338
1339 if current_config != &fetched_config {
1341 debug!(target: LOG_CLIENT, "Detected federation config change, saving as pending config");
1342
1343 let mut dbtx = db.begin_transaction().await;
1344 dbtx.insert_entry(&PendingClientConfigKey, &fetched_config)
1345 .await;
1346 dbtx.commit_tx().await;
1347 } else {
1348 debug!(target: LOG_CLIENT, "No federation config changes detected");
1349 }
1350
1351 Ok(())
1352 }
1353}
1354
1355pub struct ClientPreview {
1360 inner: ClientBuilder,
1361 config: ClientConfig,
1362 connectors: ConnectorRegistry,
1363 api_secret: Option<String>,
1364 prefetch_api_announcements: Option<Jit<Vec<PeersSignedApiAnnouncements>>>,
1365 preview_prefetch_api_version_set:
1366 Option<JitTryAnyhow<BTreeMap<PeerId, SupportedApiVersionsSummary>>>,
1367 prefetch_chain_id: Option<JitTryAnyhow<ChainId>>,
1368}
1369
1370impl ClientPreview {
1371 pub fn config(&self) -> &ClientConfig {
1373 &self.config
1374 }
1375
1376 pub async fn join(
1455 self,
1456 db_no_decoders: Database,
1457 pre_root_secret: RootSecret,
1458 ) -> anyhow::Result<ClientHandle> {
1459 let pre_root_secret = pre_root_secret.to_inner(self.config.calculate_federation_id());
1460
1461 let client = self
1462 .inner
1463 .init(
1464 self.connectors,
1465 db_no_decoders,
1466 pre_root_secret,
1467 self.config,
1468 self.api_secret,
1469 InitMode::Fresh,
1470 self.prefetch_api_announcements,
1471 self.preview_prefetch_api_version_set,
1472 self.prefetch_chain_id,
1473 )
1474 .await?;
1475
1476 Ok(client)
1477 }
1478
1479 pub async fn recover(
1491 self,
1492 db_no_decoders: Database,
1493 pre_root_secret: RootSecret,
1494 backup: Option<ClientBackup>,
1495 ) -> anyhow::Result<ClientHandle> {
1496 let pre_root_secret = pre_root_secret.to_inner(self.config.calculate_federation_id());
1497
1498 let client = self
1499 .inner
1500 .init(
1501 self.connectors,
1502 db_no_decoders,
1503 pre_root_secret,
1504 self.config,
1505 self.api_secret,
1506 InitMode::Recover {
1507 snapshot: backup.clone(),
1508 },
1509 self.prefetch_api_announcements,
1510 self.preview_prefetch_api_version_set,
1511 self.prefetch_chain_id,
1512 )
1513 .await?;
1514
1515 Ok(client)
1516 }
1517
1518 #[deprecated(
1520 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."
1521 )]
1522 #[allow(deprecated)]
1523 pub async fn download_backup_from_federation(
1524 &self,
1525 pre_root_secret: RootSecret,
1526 ) -> anyhow::Result<Option<ClientBackup>> {
1527 let pre_root_secret = pre_root_secret.to_inner(self.config.calculate_federation_id());
1528 let api = DynGlobalApi::new(
1529 self.connectors.clone(),
1530 self.config
1532 .global
1533 .api_endpoints
1534 .iter()
1535 .map(|(peer_id, peer_url)| (*peer_id, peer_url.url.clone()))
1536 .collect(),
1537 self.api_secret.as_deref(),
1538 )?;
1539
1540 Client::download_backup_from_federation_static(
1541 &api,
1542 &ClientBuilder::federation_root_secret(&pre_root_secret, &self.config),
1543 &self.inner.decoders(&self.config),
1544 )
1545 .await
1546 }
1547}