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};
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(
635 connectors.clone(),
636 peer_urls,
637 Some(admin_creds.peer_id),
638 Some(admin_creds.auth.as_str()),
639 )
640 .with_client_ext(db.clone(), log_ordering_wakeup_tx.clone())
641 .with_request_hook(&request_hook)
642 .with_cache()
643 .into(),
644 None => FederationApi::new(connectors.clone(), peer_urls, None, api_secret.as_deref())
645 .with_client_ext(db.clone(), log_ordering_wakeup_tx.clone())
646 .with_request_hook(&request_hook)
647 .with_cache()
648 .into(),
649 };
650
651 let task_group = TaskGroup::new();
652 let client_span = Client::make_client_span(fed_id);
653
654 self.migrate_module_dbs(&db, &config).await?;
657
658 let init_state = Self::load_init_state(&db).await;
659
660 let notifier = Notifier::new();
661
662 if let Some(p) = preview_prefetch_api_announcements {
663 let announcements = p.get().await;
667
668 store_api_announcements_updates_from_peers(&db, announcements).await?
669 }
670
671 if let Some(preview_prefetch_api_version_set) = preview_prefetch_api_version_set {
672 match preview_prefetch_api_version_set.get_try().await {
673 Ok(peer_api_versions) => {
674 Client::store_prefetched_api_versions(
675 &db,
676 &config,
677 &self.module_inits,
678 peer_api_versions,
679 )
680 .await;
681 }
682 Err(err) => {
683 debug!(target: LOG_CLIENT, err = %err.fmt_compact(), "Prefetching api version negotiation failed");
684 }
685 }
686 }
687
688 let common_api_versions = Client::load_and_refresh_common_api_version_static(
689 &config,
690 &self.module_inits,
691 connectors.clone(),
692 &api,
693 &db,
694 &task_group,
695 &client_span,
696 )
697 .await
698 .inspect_err(|err| {
699 warn!(target: LOG_CLIENT, err = %err.fmt_compact_anyhow(), "Failed to discover API version to use.");
700 })
701 .unwrap_or(ApiVersionSet {
702 core: ApiVersion::new(0, 0),
703 modules: BTreeMap::new(),
705 });
706
707 client_span.in_scope(|| {
708 debug!(
709 target: LOG_CLIENT,
710 core = %common_api_versions.core,
711 "Negotiated core API version",
712 );
713 for (module_id, api_version) in &common_api_versions.modules {
714 let kind = config.modules.get(module_id).map(|m| m.kind());
715 let kind_str = kind
716 .as_ref()
717 .map(|k| k.to_string())
718 .unwrap_or_else(|| format!("unknown({module_id})"));
719 let supported = kind
720 .and_then(|k| self.module_inits.get(k))
721 .map(|m| m.supported_api_versions().to_string());
722 debug!(
723 target: LOG_CLIENT,
724 module = %kind_str,
725 api = %api_version,
726 supported = %supported.as_deref().unwrap_or("unknown"),
727 "Negotiated module API version",
728 );
729 }
730 });
731
732 Self::load_and_refresh_client_config_static(&config, &api, &db, &task_group, &client_span);
734
735 if let Some(prefetch_chain_id) = prefetch_chain_id {
739 match prefetch_chain_id.get_try().await {
740 Ok(chain_id) => {
741 debug!(target: LOG_CLIENT, %chain_id, "Caching prefetched chain ID");
742 let mut dbtx = db.begin_transaction().await;
743 dbtx.insert_entry(&ChainIdKey, chain_id).await;
744 dbtx.commit_tx().await;
745 }
746 Err(err) => {
747 debug!(target: LOG_CLIENT, err = %err.fmt_compact(), "Failed to prefetch chain ID, will retry on next start");
748 }
749 }
750 }
751
752 let user_bitcoind_rpc = if let Some(factory) = self.bitcoind_rpc_factory.take() {
754 let chain_id = db.begin_transaction_nc().await.get_value(&ChainIdKey).await;
756
757 if let Some(chain_id) = chain_id {
758 debug!(target: LOG_CLIENT, %chain_id, "Creating user-provided bitcoind RPC client");
759 factory(chain_id).await
760 } else {
761 debug!(target: LOG_CLIENT, "Chain ID not available, skipping user-provided bitcoind RPC creation");
762 None
763 }
764 } else {
765 None
766 };
767
768 let mut module_recoveries: BTreeMap<ModuleInstanceId, ModuleRecoveryFuture> =
769 BTreeMap::new();
770 let mut module_recovery_progress_receivers: BTreeMap<
771 ModuleInstanceId,
772 watch::Receiver<RecoveryProgress>,
773 > = BTreeMap::new();
774
775 let final_client = FinalClientIface::default();
776
777 let root_secret = Self::federation_root_secret(&pre_root_secret, &config);
778
779 let modules = {
780 let mut modules = ClientModuleRegistry::default();
781 for (module_instance_id, module_config) in config.modules.clone() {
782 let kind = module_config.kind().clone();
783 let Some(module_init) = self.module_inits.get(&kind).cloned() else {
784 client_span.in_scope(|| {
785 debug!(
786 target: LOG_CLIENT,
787 kind=%kind,
788 instance_id=%module_instance_id,
789 "Module kind of instance not found in module gens, skipping");
790 });
791 continue;
792 };
793
794 let Some(&api_version) = common_api_versions.modules.get(&module_instance_id)
795 else {
796 client_span.in_scope(|| {
797 warn!(
798 target: LOG_CLIENT,
799 kind=%kind,
800 instance_id=%module_instance_id,
801 "Module kind of instance has incompatible api version, skipping"
802 );
803 });
804 continue;
805 };
806
807 let start_module_recover_fn =
810 |snapshot: Option<ClientBackup>, progress: RecoveryProgress| {
811 let module_config = module_config.clone();
812 let num_peers = NumPeers::from(config.global.api_endpoints.len());
813 let db = db.clone();
814 let kind = kind.clone();
815 let notifier = notifier.clone();
816 let api = api.clone();
817 let root_secret = root_secret.clone();
818 let admin_auth = self.admin_creds.as_ref().map(|creds| creds.auth.clone());
819 let final_client = final_client.clone();
820 let (progress_tx, progress_rx) = tokio::sync::watch::channel(progress);
821 let task_group = task_group.clone();
822 let module_init = module_init.clone();
823 let user_bitcoind_rpc = user_bitcoind_rpc.clone();
824 let user_bitcoind_rpc_no_chain_id =
825 self.bitcoind_rpc_no_chain_id_factory.clone();
826 let client_span = client_span.clone();
827 (
828 Box::pin(async move {
829 module_init
830 .recover(
831 final_client.clone(),
832 fed_id,
833 num_peers,
834 module_config.clone(),
835 db.clone(),
836 module_instance_id,
837 common_api_versions.core,
838 api_version,
839 root_secret.derive_module_secret(module_instance_id),
840 notifier.clone(),
841 api.clone(),
842 admin_auth,
843 snapshot.as_ref().and_then(|s| s.modules.get(&module_instance_id)),
844 progress_tx,
845 task_group,
846 client_span,
847 user_bitcoind_rpc,
848 user_bitcoind_rpc_no_chain_id,
849 )
850 .await
851 .inspect_err(|err| {
852 warn!(
853 target: LOG_CLIENT,
854 module_id = module_instance_id, %kind, err = %err.fmt_compact_anyhow(), "Module failed to recover"
855 );
856 })
857 }),
858 progress_rx,
859 )
860 };
861
862 let recovery = match init_state.does_require_recovery() {
863 Some(snapshot) => {
864 match db
865 .begin_transaction_nc()
866 .await
867 .get_value(&ClientModuleRecovery { module_instance_id })
868 .await
869 {
870 Some(module_recovery_state) => {
871 if module_recovery_state.is_done() {
872 debug!(
873 id = %module_instance_id,
874 %kind, "Module recovery already complete"
875 );
876 None
877 } else {
878 debug!(
879 id = %module_instance_id,
880 %kind,
881 progress = %module_recovery_state.progress,
882 "Starting module recovery with an existing progress"
883 );
884 Some(start_module_recover_fn(
885 snapshot,
886 module_recovery_state.progress,
887 ))
888 }
889 }
890 _ => {
891 let progress = RecoveryProgress::none();
892 let mut dbtx = db.begin_transaction().await;
893 dbtx.log_event(
894 log_ordering_wakeup_tx.clone(),
895 None,
896 ModuleRecoveryStarted::new(module_instance_id),
897 )
898 .await;
899 dbtx.insert_entry(
900 &ClientModuleRecovery { module_instance_id },
901 &ClientModuleRecoveryState { progress },
902 )
903 .await;
904
905 dbtx.commit_tx().await;
906
907 debug!(
908 id = %module_instance_id,
909 %kind, "Starting new module recovery"
910 );
911 Some(start_module_recover_fn(snapshot, progress))
912 }
913 }
914 }
915 _ => None,
916 };
917
918 match recovery {
919 Some((recovery, recovery_progress_rx)) => {
920 module_recoveries.insert(module_instance_id, recovery);
921 module_recovery_progress_receivers
922 .insert(module_instance_id, recovery_progress_rx);
923 }
924 _ => {
925 let module = module_init
926 .init(
927 final_client.clone(),
928 fed_id,
929 config.global.api_endpoints.len(),
930 module_config,
931 db.clone(),
932 module_instance_id,
933 common_api_versions.core,
934 api_version,
935 root_secret.derive_module_secret(module_instance_id),
942 notifier.clone(),
943 api.clone(),
944 self.admin_creds.as_ref().map(|cred| cred.auth.clone()),
945 task_group.clone(),
946 client_span.clone(),
947 connectors.clone(),
948 user_bitcoind_rpc.clone(),
949 self.bitcoind_rpc_no_chain_id_factory.clone(),
950 )
951 .await?;
952
953 modules.register_module(module_instance_id, kind, module);
954 }
955 }
956 }
957 modules
958 };
959
960 if init_state.is_pending() && module_recoveries.is_empty() {
961 let mut dbtx = db.begin_transaction().await;
962 dbtx.insert_entry(&ClientInitStateKey, &init_state.into_complete())
963 .await;
964 dbtx.commit_tx().await;
965 }
966
967 let mut primary_modules: BTreeMap<PrimaryModulePriority, PrimaryModuleCandidates> =
968 BTreeMap::new();
969
970 for (module_id, _kind, module) in modules.iter_modules() {
971 match module.supports_being_primary() {
972 PrimaryModuleSupport::Any { priority } => {
973 primary_modules
974 .entry(priority)
975 .or_default()
976 .wildcard
977 .push(module_id);
978 }
979 PrimaryModuleSupport::Selected { priority, units } => {
980 for unit in units {
981 primary_modules
982 .entry(priority)
983 .or_default()
984 .specific
985 .entry(unit)
986 .or_default()
987 .push(module_id);
988 }
989 }
990 PrimaryModuleSupport::None => {}
991 }
992 }
993
994 let executor = client_span.in_scope(|| {
995 let mut executor_builder = Executor::builder();
996 executor_builder
997 .with_module(TRANSACTION_SUBMISSION_MODULE_INSTANCE, TxSubmissionContext);
998
999 for (module_instance_id, _, module) in modules.iter_modules() {
1000 executor_builder.with_module_dyn(module.context(module_instance_id));
1001 }
1002
1003 for module_instance_id in module_recoveries.keys() {
1004 executor_builder.with_valid_module_id(*module_instance_id);
1005 }
1006
1007 executor_builder.build(
1008 db.clone(),
1009 notifier,
1010 task_group.clone(),
1011 log_ordering_wakeup_tx.clone(),
1012 )
1013 });
1014
1015 let recovery_receiver_init_val = module_recovery_progress_receivers
1016 .iter()
1017 .map(|(module_instance_id, rx)| (*module_instance_id, *rx.borrow()))
1018 .collect::<BTreeMap<_, _>>();
1019 let (client_recovery_progress_sender, client_recovery_progress_receiver) =
1020 watch::channel(recovery_receiver_init_val);
1021
1022 let client_inner = Arc::new(Client {
1023 final_client: final_client.clone(),
1024 config: tokio::sync::RwLock::new(config.clone()),
1025 api_secret,
1026 decoders,
1027 db: db.clone(),
1028 connectors,
1029 federation_id: fed_id,
1030 federation_config_meta: config.global.meta,
1031 primary_modules,
1032 modules,
1033 module_inits: self.module_inits.clone(),
1034 log_ordering_wakeup_tx,
1035 log_event_added_rx,
1036 log_event_added_transient_tx: log_event_added_transient_tx.clone(),
1037 request_hook,
1038 executor,
1039 api,
1040 secp_ctx: Secp256k1::new(),
1041 root_secret,
1042 task_group,
1043 client_span,
1044 operation_log: OperationLog::new(db.clone()),
1045 client_recovery_progress_receiver,
1046 meta_service: self.meta_service,
1047 iroh_enable_dht: self.iroh_enable_dht,
1048 iroh_enable_next,
1049 user_bitcoind_rpc,
1050 user_bitcoind_rpc_no_chain_id: self.bitcoind_rpc_no_chain_id_factory,
1051 });
1052 client_inner.spawn_cancellable("MetaService::update_continuously", {
1053 let client_inner = client_inner.clone();
1054 async move {
1055 client_inner
1056 .meta_service
1057 .update_continuously(&client_inner)
1058 .await;
1059 }
1060 });
1061
1062 client_inner.spawn_cancellable("update-api-announcements", {
1063 let client_inner = client_inner.clone();
1064 async move {
1065 client_inner
1066 .connectors
1067 .wait_for_initialized_connections()
1068 .await;
1069 run_api_announcement_refresh_task(client_inner.clone()).await
1070 }
1071 });
1072
1073 client_inner.spawn_cancellable("guardian metadata refresh task", {
1074 let client_inner = client_inner.clone();
1075 async move {
1076 client_inner
1077 .connectors
1078 .wait_for_initialized_connections()
1079 .await;
1080 run_guardian_metadata_refresh_task(client_inner.clone()).await
1081 }
1082 });
1083
1084 client_inner.spawn_cancellable("event log ordering task", {
1085 let client_inner = client_inner.clone();
1086 async move {
1087 client_inner
1088 .connectors
1089 .wait_for_initialized_connections()
1090 .await;
1091
1092 run_event_log_ordering_task(
1093 db.clone(),
1094 log_ordering_wakeup_rx,
1095 log_event_added_tx,
1096 log_event_added_transient_tx,
1097 )
1098 .await
1099 }
1100 });
1101
1102 if client_inner
1106 .db
1107 .begin_transaction_nc()
1108 .await
1109 .get_value(&ChainIdKey)
1110 .await
1111 .is_none()
1112 {
1113 client_inner.spawn_cancellable("fetch-chain-id", {
1114 let client_inner = client_inner.clone();
1115 async move {
1116 client_inner.api.wait_for_initialized_connections().await;
1117 match client_inner.api.chain_id().await {
1118 Ok(chain_id) => {
1119 debug!(target: LOG_CLIENT, %chain_id, "Caching chain ID from background fetch");
1120 let mut dbtx = client_inner.db.begin_transaction().await;
1121 dbtx.insert_entry(&ChainIdKey, &chain_id).await;
1122 dbtx.commit_tx().await;
1123 }
1124 Err(err) => {
1125 debug!(target: LOG_CLIENT, err = %err.fmt_compact(), "Background chain ID fetch failed, will retry on next start");
1126 }
1127 }
1128 }
1129 });
1130 }
1131
1132 let client_iface = std::sync::Arc::<Client>::downgrade(&client_inner);
1133
1134 let client_arc = ClientHandle::new(client_inner);
1135
1136 for (_, _, module) in client_arc.modules.iter_modules() {
1137 module.start().await;
1138 }
1139
1140 final_client.set(client_iface.clone());
1141
1142 if !module_recoveries.is_empty() {
1143 let module_kinds = client_arc
1147 .config()
1148 .await
1149 .modules
1150 .iter()
1151 .map(|(id, module_config)| (*id, module_config.kind().clone()))
1152 .collect();
1153 client_arc.spawn_module_recoveries_task(
1154 client_recovery_progress_sender,
1155 module_recoveries,
1156 module_recovery_progress_receivers,
1157 module_kinds,
1158 );
1159 }
1160
1161 Ok(client_arc)
1162 }
1163
1164 async fn load_init_state(db: &Database) -> InitState {
1165 let mut dbtx = db.begin_transaction_nc().await;
1166 dbtx.get_value(&ClientInitStateKey)
1167 .await
1168 .unwrap_or_else(|| {
1169 warn!(
1172 target: LOG_CLIENT,
1173 "Client missing ClientRequiresRecovery: assuming complete"
1174 );
1175 db::InitState::Complete(db::InitModeComplete::Fresh)
1176 })
1177 }
1178
1179 fn decoders(&self, config: &ClientConfig) -> ModuleDecoderRegistry {
1180 let mut decoders = client_decoders(
1181 &self.module_inits,
1182 config
1183 .modules
1184 .iter()
1185 .map(|(module_instance, module_config)| (*module_instance, module_config.kind())),
1186 );
1187
1188 decoders.register_module(
1189 TRANSACTION_SUBMISSION_MODULE_INSTANCE,
1190 ModuleKind::from_static_str("tx_submission"),
1191 tx_submission_sm_decoder(),
1192 );
1193
1194 decoders
1195 }
1196
1197 fn config_decoded(
1198 config: &ClientConfig,
1199 decoders: &ModuleDecoderRegistry,
1200 ) -> Result<ClientConfig, fedimint_core::encoding::DecodeError> {
1201 config.clone().redecode_raw(decoders)
1202 }
1203
1204 fn federation_root_secret(
1208 pre_root_secret: &DerivableSecret,
1209 config: &ClientConfig,
1210 ) -> DerivableSecret {
1211 pre_root_secret.federation_key(&config.global.calculate_federation_id())
1212 }
1213
1214 pub fn get_event_log_transient_receiver(&self) -> broadcast::Receiver<EventLogEntry> {
1216 self.log_event_added_transient_tx.subscribe()
1217 }
1218
1219 async fn migrate_pending_config_if_present(db: &Database) {
1223 if let Some(pending_config) = Client::get_pending_config_from_db(db).await {
1224 debug!(target: LOG_CLIENT, "Found pending client config, migrating to current config");
1225
1226 let mut dbtx = db.begin_transaction().await;
1227 dbtx.insert_entry(&crate::db::ClientConfigKey, &pending_config)
1229 .await;
1230 dbtx.remove_entry(&PendingClientConfigKey).await;
1232 dbtx.commit_tx().await;
1233
1234 debug!(target: LOG_CLIENT, "Successfully migrated pending config to current config");
1235 }
1236 }
1237
1238 fn load_and_refresh_client_config_static(
1241 config: &ClientConfig,
1242 api: &DynGlobalApi,
1243 db: &Database,
1244 task_group: &TaskGroup,
1245 client_span: &Span,
1246 ) {
1247 let config = config.clone();
1248 let api = api.clone();
1249 let db = db.clone();
1250 let task_group = task_group.clone();
1251
1252 task_group.spawn_cancellable_with_span(
1254 client_span.clone(),
1255 "refresh_client_config_static",
1256 async move {
1257 api.wait_for_initialized_connections().await;
1258 Self::refresh_client_config_static(&config, &api, &db).await;
1259 },
1260 );
1261 }
1262
1263 async fn refresh_client_config_static(
1265 config: &ClientConfig,
1266 api: &DynGlobalApi,
1267 db: &Database,
1268 ) {
1269 if let Err(error) = Self::refresh_client_config_static_try(config, api, db).await {
1270 warn!(
1271 target: LOG_CLIENT,
1272 err = %error.fmt_compact_anyhow(), "Failed to refresh client config"
1273 );
1274 }
1275 }
1276
1277 fn validate_config_update(
1279 current_config: &ClientConfig,
1280 new_config: &ClientConfig,
1281 ) -> anyhow::Result<()> {
1282 if current_config.global != new_config.global {
1284 bail!("Global configuration changes are not allowed in config updates");
1285 }
1286
1287 for (module_id, current_module_config) in ¤t_config.modules {
1289 match new_config.modules.get(module_id) {
1290 Some(new_module_config) => {
1291 if current_module_config != new_module_config {
1292 bail!(
1293 "Module {} configuration changes are not allowed, only additions are permitted",
1294 module_id
1295 );
1296 }
1297 }
1298 None => {
1299 bail!(
1300 "Module {} was removed in new config, only additions are allowed",
1301 module_id
1302 );
1303 }
1304 }
1305 }
1306
1307 Ok(())
1308 }
1309
1310 async fn refresh_client_config_static_try(
1312 current_config: &ClientConfig,
1313 api: &DynGlobalApi,
1314 db: &Database,
1315 ) -> anyhow::Result<()> {
1316 debug!(target: LOG_CLIENT, "Refreshing client config");
1317
1318 let fetched_config = api
1320 .request_current_consensus::<ClientConfig>(
1321 CLIENT_CONFIG_ENDPOINT.to_owned(),
1322 ApiRequestErased::default(),
1323 )
1324 .await?;
1325
1326 Self::validate_config_update(current_config, &fetched_config)?;
1328
1329 if current_config != &fetched_config {
1331 debug!(target: LOG_CLIENT, "Detected federation config change, saving as pending config");
1332
1333 let mut dbtx = db.begin_transaction().await;
1334 dbtx.insert_entry(&PendingClientConfigKey, &fetched_config)
1335 .await;
1336 dbtx.commit_tx().await;
1337 } else {
1338 debug!(target: LOG_CLIENT, "No federation config changes detected");
1339 }
1340
1341 Ok(())
1342 }
1343}
1344
1345pub struct ClientPreview {
1350 inner: ClientBuilder,
1351 config: ClientConfig,
1352 connectors: ConnectorRegistry,
1353 api_secret: Option<String>,
1354 prefetch_api_announcements: Option<Jit<Vec<PeersSignedApiAnnouncements>>>,
1355 preview_prefetch_api_version_set:
1356 Option<JitTryAnyhow<BTreeMap<PeerId, SupportedApiVersionsSummary>>>,
1357 prefetch_chain_id: Option<JitTryAnyhow<ChainId>>,
1358}
1359
1360impl ClientPreview {
1361 pub fn config(&self) -> &ClientConfig {
1363 &self.config
1364 }
1365
1366 pub async fn join(
1445 self,
1446 db_no_decoders: Database,
1447 pre_root_secret: RootSecret,
1448 ) -> anyhow::Result<ClientHandle> {
1449 let pre_root_secret = pre_root_secret.to_inner(self.config.calculate_federation_id());
1450
1451 let client = self
1452 .inner
1453 .init(
1454 self.connectors,
1455 db_no_decoders,
1456 pre_root_secret,
1457 self.config,
1458 self.api_secret,
1459 InitMode::Fresh,
1460 self.prefetch_api_announcements,
1461 self.preview_prefetch_api_version_set,
1462 self.prefetch_chain_id,
1463 )
1464 .await?;
1465
1466 Ok(client)
1467 }
1468
1469 pub async fn recover(
1481 self,
1482 db_no_decoders: Database,
1483 pre_root_secret: RootSecret,
1484 backup: Option<ClientBackup>,
1485 ) -> anyhow::Result<ClientHandle> {
1486 let pre_root_secret = pre_root_secret.to_inner(self.config.calculate_federation_id());
1487
1488 let client = self
1489 .inner
1490 .init(
1491 self.connectors,
1492 db_no_decoders,
1493 pre_root_secret,
1494 self.config,
1495 self.api_secret,
1496 InitMode::Recover {
1497 snapshot: backup.clone(),
1498 },
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 #[deprecated(
1510 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."
1511 )]
1512 #[allow(deprecated)]
1513 pub async fn download_backup_from_federation(
1514 &self,
1515 pre_root_secret: RootSecret,
1516 ) -> anyhow::Result<Option<ClientBackup>> {
1517 let pre_root_secret = pre_root_secret.to_inner(self.config.calculate_federation_id());
1518 let api = DynGlobalApi::new(
1519 self.connectors.clone(),
1520 self.config
1522 .global
1523 .api_endpoints
1524 .iter()
1525 .map(|(peer_id, peer_url)| (*peer_id, peer_url.url.clone()))
1526 .collect(),
1527 self.api_secret.as_deref(),
1528 )?;
1529
1530 Client::download_backup_from_federation_static(
1531 &api,
1532 &ClientBuilder::federation_root_secret(&pre_root_secret, &self.config),
1533 &self.inner.decoders(&self.config),
1534 )
1535 .await
1536 }
1537}