Skip to main content

fedimint_api_client/api/
mod.rs

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/// Set of api versions for each component (core + modules)
66///
67/// E.g. result of federated common api versions discovery.
68#[derive(Debug, Clone, Serialize, Deserialize, Encodable, Decodable)]
69pub struct ApiVersionSet {
70    pub core: ApiVersion,
71    pub modules: BTreeMap<ModuleInstanceId, ApiVersion>,
72}
73
74/// An API (module or global) that can query a federation
75#[apply(async_trait_maybe_send!)]
76pub trait IRawFederationApi: Debug + MaybeSend + MaybeSync {
77    /// List of all federation peers for the purpose of iterating each peer
78    /// in the federation.
79    ///
80    /// The underlying implementation is responsible for knowing how many
81    /// and `PeerId`s of each. The caller of this interface most probably
82    /// have some idea as well, but passing this set across every
83    /// API call to the federation would be inconvenient.
84    fn all_peers(&self) -> &BTreeSet<PeerId>;
85
86    /// `PeerId` of the Guardian node, if set
87    ///
88    /// This is for using Client in a "Admin" mode, making authenticated
89    /// calls to own `fedimintd` instance.
90    fn self_peer(&self) -> Option<PeerId>;
91
92    fn with_module(&self, id: ModuleInstanceId) -> DynModuleApi;
93
94    /// Make request to a specific federation peer by `peer_id`
95    async fn request_raw(
96        &self,
97        peer_id: PeerId,
98        method: &str,
99        params: &ApiRequestErased,
100    ) -> ServerResult<Value>;
101
102    /// Returns a stream of connection status for each peer
103    ///
104    /// The stream emits a new value whenever the connection status changes.
105    fn connection_status_stream(&self) -> BoxStream<'static, BTreeMap<PeerId, bool>>;
106    /// Wait for some connections being initialized
107    ///
108    /// This is useful to avoid initializing networking by
109    /// tasks that are not high priority.
110    async fn wait_for_initialized_connections(&self);
111}
112
113/// An extension trait allowing to making federation-wide API call on top
114/// [`IRawFederationApi`].
115#[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, &params)
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, &params)
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    /// Make an aggregate request to federation, using `strategy` to logically
153    /// merge the responses.
154    #[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        // NOTE: `FuturesUnorderded` is a footgun, but all we do here is polling
162        // completed results from it and we don't do any `await`s when
163        // processing them, it should be totally OK.
164        #[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 = &params;
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 = &params;
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        // NOTE: `FuturesUnorderded` is a footgun, but all we do here is polling
239        // completed results from it and we don't do any `await`s when
240        // processing them, it should be totally OK.
241        #[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 = &params;
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 = &params;
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
396/// Trait marker for the module (non-global) endpoints
397pub 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        // PeerIds are used only for informational purposes, but just in case, make a
443        // big number so it stands out
444        Self::new_admin(
445            connectors,
446            PeerId::from(1024),
447            url,
448            // Setup does not have api secrets yet
449            None,
450        )
451    }
452}
453
454/// The API for the global (non-module) endpoints
455#[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    /// Sets the password used to decrypt the configs and authenticate
488    ///
489    /// Must be called first before any other calls to the API
490    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    /// Reset the peer setup codes during the federation setup process
510    async fn reset_peer_setup_codes(&self, auth: ApiAuth) -> FederationResult<()>;
511
512    /// Returns the setup code if `set_local_params` was already called
513    async fn get_setup_code(&self, auth: ApiAuth) -> FederationResult<Option<String>>;
514
515    /// Runs DKG, can only be called once after configs have been generated in
516    /// `get_consensus_config_gen_params`.  If DKG fails this returns a 500
517    /// error and config gen must be restarted.
518    async fn start_dkg(&self, auth: ApiAuth) -> FederationResult<()>;
519
520    /// Returns the status of the server
521    async fn status(&self) -> FederationResult<StatusResponse>;
522
523    /// Show an audit across all modules
524    async fn audit(&self, auth: ApiAuth) -> FederationResult<AuditSummary>;
525
526    /// Download the guardian config to back it up
527    async fn guardian_config_backup(&self, auth: ApiAuth)
528    -> FederationResult<GuardianConfigBackup>;
529
530    /// Check auth credentials
531    async fn auth(&self, auth: ApiAuth) -> FederationResult<()>;
532
533    async fn restart_federation_setup(&self, auth: ApiAuth) -> FederationResult<()>;
534
535    /// Publish our signed API announcement to other guardians
536    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    /// Returns the fedimintd version a peer is running
556    async fn fedimintd_version(&self, peer_id: PeerId) -> ServerResult<String>;
557
558    /// Fetch the backup statistics from the federation (admin endpoint)
559    async fn backup_statistics(&self, auth: ApiAuth) -> FederationResult<BackupStatistics>;
560
561    /// Get the invite code for the federation guardian.
562    /// For instance, useful after DKG
563    async fn get_invite_code(&self, guardian: PeerId) -> ServerResult<InviteCode>;
564
565    /// Change the password used to encrypt the configs and for guardian
566    /// authentication
567    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/// Federation API client
592///
593/// The core underlying object used to make API requests to a federation.
594///
595/// It has an `connectors` handle to actually making outgoing connections
596/// to given URLs, and knows which peers there are and what URLs to connect to
597/// to reach them.
598// TODO: As it is currently it mixes a bit the role of connecting to "peers" with
599// general purpose outgoing connection. Not a big deal, but might need refactor
600// in the future.
601#[derive(Clone, Debug)]
602pub struct FederationApi {
603    /// Map of known URLs to use to connect to peers
604    peers: BTreeMap<PeerId, SafeUrl>,
605    /// List of peer ids, redundant to avoid collecting all the time
606    peers_keys: BTreeSet<PeerId>,
607    /// Our own [`PeerId`] to use when making admin apis
608    admin_id: Option<PeerId>,
609    /// Set when this API is used to communicate with a module
610    module_id: Option<ModuleInstanceId>,
611    /// Api secret of the federation
612    api_secret: Option<String>,
613    /// Connection pool
614    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    /// Get receiver for changes in the active connections
687    ///
688    /// This allows real-time monitoring of connection status.
689    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/// The status of a server, including how it views its peers
761#[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    /// This should always be 0 if everything is okay, so a monitoring tool
768    /// should generate an alert if this is not the case.
769    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    /// Indicates that this peer needs attention from the operator since
778    /// it has not contributed to the consensus in a long time
779    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;