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