fedimint_gateway_server/
registration_health.rs

1use std::collections::BTreeMap;
2use std::sync::Arc;
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::time::{Duration, SystemTime, UNIX_EPOCH};
5
6use fedimint_core::config::FederationId;
7use fedimint_core::runtime::Instant;
8use fedimint_gateway_common::{
9    RegisteredProtocol, RegistrationAttempt, RegistrationAttemptResult, RegistrationEndpointStatus,
10};
11use tokio::sync::RwLock;
12
13/// Lightning module generation whose gateway registration was attempted.
14#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
15enum RegisteredLightningModule {
16    /// Legacy Lightning module with gateway-managed announcements.
17    Lnv1,
18}
19
20/// Key for one independently attempted registration.
21#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
22struct RegistrationKey {
23    /// Federation receiving the registration.
24    federation_id: FederationId,
25    /// Lightning module generation being registered.
26    module: RegisteredLightningModule,
27    /// Public transport being advertised.
28    protocol: RegisteredProtocol,
29}
30
31/// Internal retained observation for one registration key.
32#[derive(Debug, Clone, Default)]
33struct RegistrationObservation {
34    /// Begin-order sequence of the newest retained completed attempt.
35    last_attempt_sequence: u64,
36    /// Completed attempt with the newest begin-order.
37    last_attempt: Option<RegistrationAttempt>,
38    /// Begin-order sequence and monotonic completion time of the newest
39    /// success.
40    last_success: Option<(u64, Instant)>,
41}
42
43/// Mutable registration observations and leave/rejoin invalidation watermarks.
44#[derive(Debug, Default)]
45struct RegistrationTrackerState {
46    /// Latest observations keyed by federation, module, and transport.
47    observations: BTreeMap<RegistrationKey, RegistrationObservation>,
48    /// Attempt sequences invalidated when a gateway left a federation.
49    cleared_through: BTreeMap<FederationId, u64>,
50}
51
52/// Token that orders concurrent registration attempt completions.
53#[derive(Debug)]
54pub(crate) struct RegistrationAttemptToken {
55    /// Registration key being updated.
56    key: RegistrationKey,
57    /// Monotonically increasing attempt sequence.
58    sequence: u64,
59}
60
61/// Retains detail-free runtime registration results for public health queries.
62#[derive(Debug, Clone, Default)]
63pub(crate) struct RegistrationHealthTracker {
64    /// Next sequence used to order attempts when they begin.
65    next_sequence: Arc<AtomicU64>,
66    /// Retained observations and leave/rejoin invalidation watermarks.
67    state: Arc<RwLock<RegistrationTrackerState>>,
68}
69
70impl RegistrationHealthTracker {
71    /// Starts an LNv1 registration attempt and returns its ordering token.
72    pub(crate) fn begin_lnv1_attempt(
73        &self,
74        federation_id: FederationId,
75        protocol: RegisteredProtocol,
76    ) -> RegistrationAttemptToken {
77        RegistrationAttemptToken {
78            key: RegistrationKey {
79                federation_id,
80                module: RegisteredLightningModule::Lnv1,
81                protocol,
82            },
83            sequence: self.next_sequence.fetch_add(1, Ordering::Relaxed) + 1,
84        }
85    }
86
87    /// Retains a finite result unless an attempt begun later already completed.
88    ///
89    /// Begin order, rather than completion wall time, defines freshness so a
90    /// slow stale request cannot overwrite the result of a newer logical
91    /// attempt. Successful TTL is measured from monotonic completion time.
92    pub(crate) async fn complete_attempt(
93        &self,
94        token: RegistrationAttemptToken,
95        succeeded: bool,
96        completed_wall_time: SystemTime,
97        completed_monotonic: Instant,
98    ) {
99        let mut state = self.state.write().await;
100        if state
101            .cleared_through
102            .get(&token.key.federation_id)
103            .is_some_and(|cleared_through| token.sequence <= *cleared_through)
104        {
105            return;
106        }
107
108        let observation = state.observations.entry(token.key).or_default();
109        if succeeded
110            && observation
111                .last_success
112                .is_none_or(|(sequence, _)| sequence <= token.sequence)
113        {
114            observation.last_success = Some((token.sequence, completed_monotonic));
115        }
116
117        if observation.last_attempt_sequence <= token.sequence {
118            observation.last_attempt_sequence = token.sequence;
119            observation.last_attempt = Some(RegistrationAttempt {
120                completed_at_unix_secs: completed_wall_time
121                    .duration_since(UNIX_EPOCH)
122                    .unwrap_or_default()
123                    .as_secs(),
124                result: if succeeded {
125                    RegistrationAttemptResult::Succeeded
126                } else {
127                    RegistrationAttemptResult::Failed
128                },
129            });
130        }
131    }
132
133    /// Returns status for every configured LNv1 public transport.
134    pub(crate) async fn lnv1_status(
135        &self,
136        federation_id: FederationId,
137        protocols: impl IntoIterator<Item = RegisteredProtocol>,
138        ttl: Duration,
139        now: Instant,
140    ) -> BTreeMap<RegisteredProtocol, RegistrationEndpointStatus> {
141        let state = self.state.read().await;
142        protocols
143            .into_iter()
144            .map(|protocol| {
145                let observation = state.observations.get(&RegistrationKey {
146                    federation_id,
147                    module: RegisteredLightningModule::Lnv1,
148                    protocol: protocol.clone(),
149                });
150                let ttl_remaining = observation.and_then(|observation| {
151                    observation.last_success.map(|(_, last_success)| {
152                        ttl.saturating_sub(now.saturating_duration_since(last_success))
153                            .min(ttl)
154                            .as_secs()
155                    })
156                });
157                (
158                    protocol,
159                    RegistrationEndpointStatus {
160                        last_attempt: observation.and_then(|value| value.last_attempt.clone()),
161                        advertised_ttl_remaining_secs: ttl_remaining,
162                    },
163                )
164            })
165            .collect()
166    }
167
168    /// Removes observations after the gateway leaves a federation.
169    pub(crate) async fn clear_federation(&self, federation_id: FederationId) {
170        let mut state = self.state.write().await;
171        state
172            .observations
173            .retain(|key, _| key.federation_id != federation_id);
174        state
175            .cleared_through
176            .insert(federation_id, self.next_sequence.load(Ordering::Relaxed));
177    }
178}
179
180#[cfg(test)]
181mod tests;
fedimint_gateway_server/registration_health.rs

fedimint_gateway_server/
registration_health.rs
