1mod error;
2pub mod global_api;
3
4use std::collections::{BTreeMap, BTreeSet, HashMap};
5use std::fmt::Debug;
6use std::future::pending;
7use std::pin::Pin;
8use std::result;
9use std::sync::Arc;
10
11use anyhow::{Context, anyhow};
12use bitcoin::secp256k1;
13pub use error::{FederationError, OutputOutcomeError};
14pub use fedimint_connectors::ServerResult;
15pub use fedimint_connectors::error::ServerError;
16use fedimint_connectors::{
17 ConnectionPool, ConnectorRegistry, DynGuaridianConnection, IGuardianConnection,
18};
19use fedimint_core::admin_client::{GuardianConfigBackup, ServerStatusLegacy, SetupStatus};
20use fedimint_core::backup::{BackupStatistics, ClientBackupSnapshot};
21use fedimint_core::core::backup::SignedBackupRequest;
22use fedimint_core::core::{Decoder, DynOutputOutcome, ModuleInstanceId, ModuleKind, OutputOutcome};
23use fedimint_core::encoding::{Decodable, Encodable};
24use fedimint_core::invite_code::InviteCode;
25use fedimint_core::module::audit::AuditSummary;
26use fedimint_core::module::registry::ModuleDecoderRegistry;
27use fedimint_core::module::{
28 ApiAuth, ApiMethod, ApiRequestErased, ApiVersion, SerdeModuleEncoding,
29};
30use fedimint_core::net::api_announcement::SignedApiAnnouncement;
31use fedimint_core::session_outcome::{SessionOutcome, SessionStatus};
32use fedimint_core::task::{MaybeSend, MaybeSync};
33use fedimint_core::transaction::{Transaction, TransactionSubmissionOutcome};
34use fedimint_core::util::backoff_util::api_networking_backoff;
35use fedimint_core::util::{FmtCompact as _, SafeUrl};
36use fedimint_core::{
37 NumPeersExt, PeerId, TransactionId, apply, async_trait_maybe_send, dyn_newtype_define, util,
38};
39use fedimint_logging::LOG_CLIENT_NET_API;
40use fedimint_metrics::HistogramExt as _;
41use futures::stream::{BoxStream, FuturesUnordered};
42use futures::{Future, StreamExt};
43use global_api::with_cache::GlobalFederationApiWithCache;
44use jsonrpsee_core::DeserializeOwned;
45use serde::{Deserialize, Serialize};
46use serde_json::Value;
47use tokio::sync::watch;
48use tokio_stream::wrappers::WatchStream;
49use tracing::{debug, instrument, trace, warn};
50
51use crate::metrics::{CLIENT_API_REQUEST_DURATION_SECONDS, CLIENT_API_REQUESTS_TOTAL};
52use crate::query::{QueryStep, QueryStrategy, ThresholdConsensus};
53
54pub const VERSION_THAT_INTRODUCED_GET_SESSION_STATUS_V2: ApiVersion = ApiVersion::new(0, 5);
55
56pub const VERSION_THAT_INTRODUCED_GET_SESSION_STATUS: ApiVersion =
57 ApiVersion { major: 0, minor: 1 };
58
59pub const VERSION_THAT_INTRODUCED_AWAIT_OUTPUTS_OUTCOMES: ApiVersion = ApiVersion::new(0, 8);
60pub type FederationResult<T> = Result<T, FederationError>;
61pub type SerdeOutputOutcome = SerdeModuleEncoding<DynOutputOutcome>;
62
63pub type OutputOutcomeResult<O> = result::Result<O, OutputOutcomeError>;
64
65#[derive(Debug, Clone, Serialize, Deserialize, Encodable, Decodable)]
69pub struct ApiVersionSet {
70 pub core: ApiVersion,
71 pub modules: BTreeMap<ModuleInstanceId, ApiVersion>,
72}
73
74#[apply(async_trait_maybe_send!)]
76pub trait IRawFederationApi: Debug + MaybeSend + MaybeSync {
77 fn all_peers(&self) -> &BTreeSet<PeerId>;
85
86 fn self_peer(&self) -> Option<PeerId>;
91
92 fn with_module(&self, id: ModuleInstanceId) -> DynModuleApi;
93
94 async fn request_raw(
96 &self,
97 peer_id: PeerId,
98 method: &str,
99 params: &ApiRequestErased,
100 ) -> ServerResult<Value>;
101
102 fn connection_status_stream(&self) -> BoxStream<'static, BTreeMap<PeerId, bool>>;
106 async fn wait_for_initialized_connections(&self);
111}
112
113#[apply(async_trait_maybe_send!)]
116pub trait FederationApiExt: IRawFederationApi {
117 async fn request_single_peer<Ret>(
118 &self,
119 method: String,
120 params: ApiRequestErased,
121 peer: PeerId,
122 ) -> ServerResult<Ret>
123 where
124 Ret: DeserializeOwned,
125 {
126 self.request_raw(peer, &method, ¶ms)
127 .await
128 .and_then(|v| {
129 serde_json::from_value(v)
130 .map_err(|e| ServerError::ResponseDeserialization(e.into()))
131 })
132 }
133
134 async fn request_single_peer_federation<FedRet>(
135 &self,
136 method: String,
137 params: ApiRequestErased,
138 peer_id: PeerId,
139 ) -> FederationResult<FedRet>
140 where
141 FedRet: serde::de::DeserializeOwned + Eq + Debug + Clone + MaybeSend,
142 {
143 self.request_raw(peer_id, &method, ¶ms)
144 .await
145 .and_then(|v| {
146 serde_json::from_value(v)
147 .map_err(|e| ServerError::ResponseDeserialization(e.into()))
148 })
149 .map_err(|e| error::FederationError::new_one_peer(peer_id, method, params, e))
150 }
151
152 #[instrument(target = LOG_CLIENT_NET_API, skip_all, fields(method=method))]
155 async fn request_with_strategy<PR: DeserializeOwned, FR: Debug>(
156 &self,
157 mut strategy: impl QueryStrategy<PR, FR> + MaybeSend,
158 method: String,
159 params: ApiRequestErased,
160 ) -> FederationResult<FR> {
161 #[cfg(not(target_family = "wasm"))]
165 let mut futures = FuturesUnordered::<Pin<Box<dyn Future<Output = _> + Send>>>::new();
166 #[cfg(target_family = "wasm")]
167 let mut futures = FuturesUnordered::<Pin<Box<dyn Future<Output = _>>>>::new();
168
169 for peer in self.all_peers() {
170 futures.push(Box::pin({
171 let method = &method;
172 let params = ¶ms;
173 async move {
174 let result = self
175 .request_single_peer(method.clone(), params.clone(), *peer)
176 .await;
177
178 (*peer, result)
179 }
180 }));
181 }
182
183 let mut peer_errors = BTreeMap::new();
184 let peer_error_threshold = self.all_peers().to_num_peers().one_honest();
185
186 loop {
187 let (peer, result) = futures
188 .next()
189 .await
190 .expect("Query strategy ran out of peers to query without returning a result");
191
192 match result {
193 Ok(response) => match strategy.process(peer, response) {
194 QueryStep::Retry(peers) => {
195 for peer in peers {
196 futures.push(Box::pin({
197 let method = &method;
198 let params = ¶ms;
199 async move {
200 let result = self
201 .request_single_peer(method.clone(), params.clone(), peer)
202 .await;
203
204 (peer, result)
205 }
206 }));
207 }
208 }
209 QueryStep::Success(response) => return Ok(response),
210 QueryStep::Failure(e) => {
211 peer_errors.insert(peer, e);
212 }
213 QueryStep::Continue => {}
214 },
215 Err(e) => {
216 e.report_if_unusual(peer, "RequestWithStrategy");
217 peer_errors.insert(peer, e);
218 }
219 }
220
221 if peer_errors.len() == peer_error_threshold {
222 return Err(FederationError::peer_errors(
223 method.clone(),
224 params.params.clone(),
225 peer_errors,
226 ));
227 }
228 }
229 }
230
231 #[instrument(target = LOG_CLIENT_NET_API, level = "debug", skip(self, strategy))]
232 async fn request_with_strategy_retry<PR: DeserializeOwned + MaybeSend, FR: Debug>(
233 &self,
234 mut strategy: impl QueryStrategy<PR, FR> + MaybeSend,
235 method: String,
236 params: ApiRequestErased,
237 ) -> FR {
238 #[cfg(not(target_family = "wasm"))]
242 let mut futures = FuturesUnordered::<Pin<Box<dyn Future<Output = _> + Send>>>::new();
243 #[cfg(target_family = "wasm")]
244 let mut futures = FuturesUnordered::<Pin<Box<dyn Future<Output = _>>>>::new();
245
246 for peer in self.all_peers() {
247 futures.push(Box::pin({
248 let method = &method;
249 let params = ¶ms;
250 async move {
251 let response = util::retry(
252 format!("api-request-{method}-{peer}"),
253 api_networking_backoff(),
254 || async {
255 self.request_single_peer(method.clone(), params.clone(), *peer)
256 .await
257 .inspect_err(|e| {
258 e.report_if_unusual(*peer, "QueryWithStrategyRetry");
259 })
260 .map_err(|e| anyhow!(e.to_string()))
261 },
262 )
263 .await
264 .expect("Number of retries has no limit");
265
266 (*peer, response)
267 }
268 }));
269 }
270
271 loop {
272 let (peer, response) = match futures.next().await {
273 Some(t) => t,
274 None => pending().await,
275 };
276
277 match strategy.process(peer, response) {
278 QueryStep::Retry(peers) => {
279 for peer in peers {
280 futures.push(Box::pin({
281 let method = &method;
282 let params = ¶ms;
283 async move {
284 let response = util::retry(
285 format!("api-request-{method}-{peer}"),
286 api_networking_backoff(),
287 || async {
288 self.request_single_peer(
289 method.clone(),
290 params.clone(),
291 peer,
292 )
293 .await
294 .inspect_err(|err| {
295 if err.is_unusual() {
296 debug!(target: LOG_CLIENT_NET_API, err = %err.fmt_compact(), "Unusual peer error");
297 }
298 })
299 .map_err(|e| anyhow!(e.to_string()))
300 },
301 )
302 .await
303 .expect("Number of retries has no limit");
304
305 (peer, response)
306 }
307 }));
308 }
309 }
310 QueryStep::Success(response) => return response,
311 QueryStep::Failure(e) => {
312 warn!(target: LOG_CLIENT_NET_API, "Query strategy returned non-retryable failure for peer {peer}: {e}");
313 }
314 QueryStep::Continue => {}
315 }
316 }
317 }
318
319 async fn request_current_consensus<Ret>(
320 &self,
321 method: String,
322 params: ApiRequestErased,
323 ) -> FederationResult<Ret>
324 where
325 Ret: DeserializeOwned + Eq + Debug + Clone + MaybeSend,
326 {
327 self.request_with_strategy(
328 ThresholdConsensus::new(self.all_peers().to_num_peers()),
329 method,
330 params,
331 )
332 .await
333 }
334
335 async fn request_current_consensus_retry<Ret>(
336 &self,
337 method: String,
338 params: ApiRequestErased,
339 ) -> Ret
340 where
341 Ret: DeserializeOwned + Eq + Debug + Clone + MaybeSend,
342 {
343 self.request_with_strategy_retry(
344 ThresholdConsensus::new(self.all_peers().to_num_peers()),
345 method,
346 params,
347 )
348 .await
349 }
350
351 async fn request_admin<Ret>(
352 &self,
353 method: &str,
354 params: ApiRequestErased,
355 auth: ApiAuth,
356 ) -> FederationResult<Ret>
357 where
358 Ret: DeserializeOwned + Eq + Debug + Clone + MaybeSend,
359 {
360 let Some(self_peer_id) = self.self_peer() else {
361 return Err(FederationError::general(
362 method,
363 params,
364 anyhow::format_err!("Admin peer_id not set"),
365 ));
366 };
367
368 self.request_single_peer_federation(method.into(), params.with_auth(auth), self_peer_id)
369 .await
370 }
371
372 async fn request_admin_no_auth<Ret>(
373 &self,
374 method: &str,
375 params: ApiRequestErased,
376 ) -> FederationResult<Ret>
377 where
378 Ret: DeserializeOwned + Eq + Debug + Clone + MaybeSend,
379 {
380 let Some(self_peer_id) = self.self_peer() else {
381 return Err(FederationError::general(
382 method,
383 params,
384 anyhow::format_err!("Admin peer_id not set"),
385 ));
386 };
387
388 self.request_single_peer_federation(method.into(), params, self_peer_id)
389 .await
390 }
391}
392
393#[apply(async_trait_maybe_send!)]
394impl<T: ?Sized> FederationApiExt for T where T: IRawFederationApi {}
395
396pub trait IModuleFederationApi: IRawFederationApi {}
398
399dyn_newtype_define! {
400 #[derive(Clone)]
401 pub DynModuleApi(Arc<IModuleFederationApi>)
402}
403
404dyn_newtype_define! {
405 #[derive(Clone)]
406 pub DynGlobalApi(Arc<IGlobalFederationApi>)
407}
408
409impl AsRef<dyn IGlobalFederationApi + 'static> for DynGlobalApi {
410 fn as_ref(&self) -> &(dyn IGlobalFederationApi + 'static) {
411 self.inner.as_ref()
412 }
413}
414
415impl DynGlobalApi {
416 pub fn new(
417 connectors: ConnectorRegistry,
418 peers: BTreeMap<PeerId, SafeUrl>,
419 api_secret: Option<&str>,
420 ) -> anyhow::Result<Self> {
421 Ok(GlobalFederationApiWithCache::new(FederationApi::new(
422 connectors, peers, None, api_secret,
423 ))
424 .into())
425 }
426 pub fn new_admin(
427 connectors: ConnectorRegistry,
428 peer: PeerId,
429 url: SafeUrl,
430 api_secret: Option<&str>,
431 ) -> anyhow::Result<DynGlobalApi> {
432 Ok(GlobalFederationApiWithCache::new(FederationApi::new(
433 connectors,
434 [(peer, url)].into(),
435 Some(peer),
436 api_secret,
437 ))
438 .into())
439 }
440
441 pub fn new_admin_setup(connectors: ConnectorRegistry, url: SafeUrl) -> anyhow::Result<Self> {
442 Self::new_admin(
445 connectors,
446 PeerId::from(1024),
447 url,
448 None,
450 )
451 }
452}
453
454#[apply(async_trait_maybe_send!)]
456pub trait IGlobalFederationApi: IRawFederationApi {
457 async fn submit_transaction(
458 &self,
459 tx: Transaction,
460 ) -> SerdeModuleEncoding<TransactionSubmissionOutcome>;
461
462 async fn await_block(
463 &self,
464 block_index: u64,
465 decoders: &ModuleDecoderRegistry,
466 ) -> anyhow::Result<SessionOutcome>;
467
468 async fn get_session_status(
469 &self,
470 block_index: u64,
471 decoders: &ModuleDecoderRegistry,
472 core_api_version: ApiVersion,
473 broadcast_public_keys: Option<&BTreeMap<PeerId, secp256k1::PublicKey>>,
474 ) -> anyhow::Result<SessionStatus>;
475
476 async fn session_count(&self) -> FederationResult<u64>;
477
478 async fn await_transaction(&self, txid: TransactionId) -> TransactionId;
479
480 async fn upload_backup(&self, request: &SignedBackupRequest) -> FederationResult<()>;
481
482 async fn download_backup(
483 &self,
484 id: &secp256k1::PublicKey,
485 ) -> FederationResult<BTreeMap<PeerId, Option<ClientBackupSnapshot>>>;
486
487 async fn set_password(&self, auth: ApiAuth) -> FederationResult<()>;
491
492 async fn setup_status(&self, auth: ApiAuth) -> FederationResult<SetupStatus>;
493
494 async fn set_local_params(
495 &self,
496 name: String,
497 federation_name: Option<String>,
498 disable_base_fees: Option<bool>,
499 enabled_modules: Option<BTreeSet<ModuleKind>>,
500 auth: ApiAuth,
501 ) -> FederationResult<String>;
502
503 async fn add_peer_connection_info(
504 &self,
505 info: String,
506 auth: ApiAuth,
507 ) -> FederationResult<String>;
508
509 async fn reset_peer_setup_codes(&self, auth: ApiAuth) -> FederationResult<()>;
511
512 async fn get_setup_code(&self, auth: ApiAuth) -> FederationResult<Option<String>>;
514
515 async fn start_dkg(&self, auth: ApiAuth) -> FederationResult<()>;
519
520 async fn status(&self) -> FederationResult<StatusResponse>;
522
523 async fn audit(&self, auth: ApiAuth) -> FederationResult<AuditSummary>;
525
526 async fn guardian_config_backup(&self, auth: ApiAuth)
528 -> FederationResult<GuardianConfigBackup>;
529
530 async fn auth(&self, auth: ApiAuth) -> FederationResult<()>;
532
533 async fn restart_federation_setup(&self, auth: ApiAuth) -> FederationResult<()>;
534
535 async fn submit_api_announcement(
537 &self,
538 peer_id: PeerId,
539 announcement: SignedApiAnnouncement,
540 ) -> FederationResult<()>;
541
542 async fn api_announcements(
543 &self,
544 guardian: PeerId,
545 ) -> ServerResult<BTreeMap<PeerId, SignedApiAnnouncement>>;
546
547 async fn sign_api_announcement(
548 &self,
549 api_url: SafeUrl,
550 auth: ApiAuth,
551 ) -> FederationResult<SignedApiAnnouncement>;
552
553 async fn shutdown(&self, session: Option<u64>, auth: ApiAuth) -> FederationResult<()>;
554
555 async fn fedimintd_version(&self, peer_id: PeerId) -> ServerResult<String>;
557
558 async fn backup_statistics(&self, auth: ApiAuth) -> FederationResult<BackupStatistics>;
560
561 async fn get_invite_code(&self, guardian: PeerId) -> ServerResult<InviteCode>;
564
565 async fn change_password(&self, auth: ApiAuth, new_password: &str) -> FederationResult<()>;
568}
569
570pub fn deserialize_outcome<R>(
571 outcome: &SerdeOutputOutcome,
572 module_decoder: &Decoder,
573) -> OutputOutcomeResult<R>
574where
575 R: OutputOutcome + MaybeSend,
576{
577 let dyn_outcome = outcome
578 .try_into_inner_known_module_kind(module_decoder)
579 .map_err(|e| OutputOutcomeError::ResponseDeserialization(e.into()))?;
580
581 let source_instance = dyn_outcome.module_instance_id();
582
583 dyn_outcome.as_any().downcast_ref().cloned().ok_or_else(|| {
584 let target_type = std::any::type_name::<R>();
585 OutputOutcomeError::ResponseDeserialization(anyhow!(
586 "Could not downcast output outcome with instance id {source_instance} to {target_type}"
587 ))
588 })
589}
590
591#[derive(Clone, Debug)]
602pub struct FederationApi {
603 peers: BTreeMap<PeerId, SafeUrl>,
605 peers_keys: BTreeSet<PeerId>,
607 admin_id: Option<PeerId>,
609 module_id: Option<ModuleInstanceId>,
611 api_secret: Option<String>,
613 connection_pool: ConnectionPool<dyn IGuardianConnection>,
615}
616
617impl FederationApi {
618 pub fn new(
619 connectors: ConnectorRegistry,
620 peers: BTreeMap<PeerId, SafeUrl>,
621 admin_peer_id: Option<PeerId>,
622 api_secret: Option<&str>,
623 ) -> Self {
624 Self {
625 peers_keys: peers.keys().copied().collect(),
626 peers,
627 admin_id: admin_peer_id,
628 module_id: None,
629 api_secret: api_secret.map(ToOwned::to_owned),
630 connection_pool: ConnectionPool::new(connectors),
631 }
632 }
633
634 async fn get_or_create_connection(
635 &self,
636 url: &SafeUrl,
637 api_secret: Option<&str>,
638 ) -> ServerResult<DynGuaridianConnection> {
639 self.connection_pool
640 .get_or_create_connection(url, api_secret, |url, api_secret, connectors| async move {
641 let conn = connectors
642 .connect_guardian(&url, api_secret.as_deref())
643 .await?;
644 Ok(conn)
645 })
646 .await
647 }
648
649 async fn request(
650 &self,
651 peer: PeerId,
652 method: ApiMethod,
653 request: ApiRequestErased,
654 ) -> ServerResult<Value> {
655 trace!(target: LOG_CLIENT_NET_API, %peer, %method, "Api request");
656 let url = self
657 .peers
658 .get(&peer)
659 .ok_or_else(|| ServerError::InvalidPeerId { peer_id: peer })?;
660 let conn = self
661 .get_or_create_connection(url, self.api_secret.as_deref())
662 .await
663 .context("Failed to connect to peer")
664 .map_err(ServerError::Connection)?;
665
666 let method_str = method.to_string();
667 let peer_str = peer.to_string();
668 let timer = CLIENT_API_REQUEST_DURATION_SECONDS
669 .with_label_values(&[&method_str, &peer_str])
670 .start_timer_ext();
671
672 let res = conn.request(method.clone(), request).await;
673
674 timer.observe_duration();
675
676 let result_label = if res.is_ok() { "success" } else { "error" }.to_string();
677 CLIENT_API_REQUESTS_TOTAL
678 .with_label_values(&[&method_str, &peer_str, &result_label])
679 .inc();
680
681 trace!(target: LOG_CLIENT_NET_API, ?method, res_ok = res.is_ok(), "Api response");
682
683 res
684 }
685
686 pub fn get_active_connection_receiver(&self) -> watch::Receiver<BTreeSet<SafeUrl>> {
690 self.connection_pool.get_active_connection_receiver()
691 }
692}
693
694impl IModuleFederationApi for FederationApi {}
695
696#[apply(async_trait_maybe_send!)]
697impl IRawFederationApi for FederationApi {
698 fn all_peers(&self) -> &BTreeSet<PeerId> {
699 &self.peers_keys
700 }
701
702 fn self_peer(&self) -> Option<PeerId> {
703 self.admin_id
704 }
705
706 fn with_module(&self, id: ModuleInstanceId) -> DynModuleApi {
707 FederationApi {
708 api_secret: self.api_secret.clone(),
709 peers: self.peers.clone(),
710 peers_keys: self.peers_keys.clone(),
711 admin_id: self.admin_id,
712 module_id: Some(id),
713 connection_pool: self.connection_pool.clone(),
714 }
715 .into()
716 }
717
718 #[instrument(
719 target = LOG_CLIENT_NET_API,
720 skip_all,
721 fields(
722 peer_id = %peer_id,
723 method = %method,
724 params = %params.params,
725 )
726 )]
727 async fn request_raw(
728 &self,
729 peer_id: PeerId,
730 method: &str,
731 params: &ApiRequestErased,
732 ) -> ServerResult<Value> {
733 let method = match self.module_id {
734 Some(module_id) => ApiMethod::Module(module_id, method.to_string()),
735 None => ApiMethod::Core(method.to_string()),
736 };
737
738 self.request(peer_id, method, params.clone()).await
739 }
740
741 fn connection_status_stream(&self) -> BoxStream<'static, BTreeMap<PeerId, bool>> {
742 let peers = self.peers.clone();
743
744 WatchStream::new(self.connection_pool.get_active_connection_receiver())
745 .map(move |active_urls| {
746 peers
747 .iter()
748 .map(|(peer_id, url)| (*peer_id, active_urls.contains(url)))
749 .collect()
750 })
751 .boxed()
752 }
753 async fn wait_for_initialized_connections(&self) {
754 self.connection_pool
755 .wait_for_initialized_connections()
756 .await;
757 }
758}
759
760#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
762pub struct LegacyFederationStatus {
763 pub session_count: u64,
764 pub status_by_peer: HashMap<PeerId, LegacyPeerStatus>,
765 pub peers_online: u64,
766 pub peers_offline: u64,
767 pub peers_flagged: u64,
770 pub scheduled_shutdown: Option<u64>,
771}
772
773#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
774pub struct LegacyPeerStatus {
775 pub last_contribution: Option<u64>,
776 pub connection_status: LegacyP2PConnectionStatus,
777 pub flagged: bool,
780}
781
782#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
783#[serde(rename_all = "snake_case")]
784pub enum LegacyP2PConnectionStatus {
785 #[default]
786 Disconnected,
787 Connected,
788}
789
790#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
791pub struct StatusResponse {
792 pub server: ServerStatusLegacy,
793 pub federation: Option<LegacyFederationStatus>,
794}
795
796#[cfg(test)]
797mod tests;