liminal_server/auth_lys/grant_stream_owner/
state.rs

1//! The complete revoked baseline and the membership authority share one cursor.
2
3use crate::auth_membership::{AuthorityState, MembershipAuthority};
4use crate::auth_revocation::{
5    DecisionReply, GrantDecisionSource, GrantId, GrantLogId, Revision, RevocationRefusal,
6};
7use lys_pass::membership::GrantLog;
8use serde::Serialize;
9use std::collections::BTreeSet;
10use std::sync::{Arc, Mutex};
11use tokio::sync::watch;
12
13/// Diagnostics contain counts and stable local error names, never response words.
14#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
15pub struct StreamDiagnostics {
16    /// Readiness of both authorities at the applied fence.
17    pub readiness: &'static str,
18    /// Last successfully applied grant receipt revision.
19    pub cursor: Option<u64>,
20    /// Last successfully applied placement position.
21    pub placed: Option<u64>,
22    /// Stable cause; response bodies and remote refusal text are excluded.
23    pub last_error: Option<&'static str>,
24    /// Uncovered receipt positions, when a gap is known.
25    pub missing: Option<u64>,
26    /// Settled membership generation.
27    pub generation: u64,
28    /// Reads actually issued, including the sole waiting read at idle.
29    pub requests: u64,
30    /// Transport-ended resumptions.
31    pub reopens: u64,
32    /// True after the reader loop ended; the shutdown guard still joins its worker.
33    pub stopped: bool,
34}
35
36/// A failed snapshot names its failure and carries no invented cursor or counts.
37#[derive(Clone, Debug, Serialize)]
38#[serde(untagged)]
39pub enum StreamHealth {
40    /// Counts read under the authority's trusted hold.
41    Known(StreamDiagnostics),
42    /// The hold cannot be trusted; admission remains closed.
43    Unavailable {
44        /// The owner cannot establish its state.
45        readiness: &'static str,
46        /// No cursor is claimed without the hold.
47        cursor: Option<u64>,
48        /// No placement cursor is claimed without the hold.
49        placed: Option<u64>,
50        /// The stable failure, without any untrusted words.
51        last_error: &'static str,
52    },
53}
54
55impl StreamHealth {
56    /// Whether this snapshot establishes the complete readiness fence.
57    #[must_use]
58    pub fn is_ready(&self) -> bool {
59        matches!(self, Self::Known(state) if state.readiness == "ready")
60    }
61}
62
63pub(super) struct Inner {
64    pub diagnostics: StreamDiagnostics,
65    pub revoked: BTreeSet<GrantId>,
66}
67
68/// Locally readable authority; a poisoned lock closes admission by name.
69pub struct StreamState {
70    pub(super) reader_claim: std::sync::atomic::AtomicBool,
71    pub(super) log: GrantLog,
72    pub(super) inner: Mutex<Inner>,
73    pub(super) changed: watch::Sender<Result<StreamDiagnostics, super::StreamError>>,
74    pub(super) stop: watch::Sender<bool>,
75}
76
77impl std::fmt::Debug for StreamState {
78    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
79        f.debug_struct("StreamState")
80            .field("diagnostics", &self.snapshot())
81            .finish_non_exhaustive()
82    }
83}
84
85impl StreamState {
86    /// Installs a closed authority before any reader or protected connection exists.
87    #[must_use]
88    pub fn for_log(log: GrantLog) -> Arc<Self> {
89        let diagnostics = StreamDiagnostics {
90            readiness: "not_ready",
91            cursor: None,
92            placed: None,
93            last_error: None,
94            missing: None,
95            generation: 0,
96            requests: 0,
97            reopens: 0,
98            stopped: false,
99        };
100        Arc::new(Self {
101            reader_claim: std::sync::atomic::AtomicBool::new(false),
102            log,
103            inner: Mutex::new(Inner {
104                diagnostics: diagnostics.clone(),
105                revoked: BTreeSet::new(),
106            }),
107            changed: watch::channel(Ok(diagnostics)).0,
108            stop: watch::channel(false).0,
109        })
110    }
111
112    /// Counts-only snapshot; nothing is answered from a poisoned hold.
113    ///
114    /// # Errors
115    /// Names an untrusted authority hold rather than fabricating counters.
116    pub fn snapshot(&self) -> Result<StreamDiagnostics, super::StreamError> {
117        self.inner
118            .lock()
119            .map(|inner| inner.diagnostics.clone())
120            .map_err(|_| super::StreamError::named("grant_stream_state_poisoned"))
121    }
122
123    /// Health reports a failure explicitly while preserving the liveness route.
124    #[must_use]
125    pub fn health(&self) -> StreamHealth {
126        match self.snapshot() {
127            Ok(status) => StreamHealth::Known(status),
128            Err(error) => StreamHealth::Unavailable {
129                readiness: "terminal",
130                cursor: None,
131                placed: None,
132                last_error: error.name(),
133            },
134        }
135    }
136
137    /// Owned progress signal for consumers and tests; no timer or request is created.
138    #[must_use]
139    pub fn subscribe(&self) -> watch::Receiver<Result<StreamDiagnostics, super::StreamError>> {
140        self.changed.subscribe()
141    }
142
143    /// Cancellation also interrupts in-flight membership and binding requests.
144    #[must_use]
145    pub fn cancellation(&self) -> watch::Receiver<bool> {
146        self.stop.subscribe()
147    }
148
149    pub(super) fn update(
150        &self,
151        edit: impl FnOnce(&mut Inner) -> Result<(), super::StreamError>,
152    ) -> Result<(), super::StreamError> {
153        let result = (|| {
154            let mut inner = self
155                .inner
156                .lock()
157                .map_err(|_| super::StreamError::named("grant_stream_state_poisoned"))?;
158            edit(&mut inner)?;
159            Ok(inner.diagnostics.clone())
160        })();
161        drop(self.changed.send_replace(result.clone()));
162        result.map(|_| ())
163    }
164
165    pub(super) fn unavailable(&self, name: &'static str) -> Result<(), super::StreamError> {
166        self.update(|inner| {
167            inner.diagnostics.readiness = "unavailable";
168            inner.diagnostics.last_error = Some(name);
169            Ok(())
170        })
171    }
172}
173
174impl MembershipAuthority for StreamState {
175    fn state(&self) -> AuthorityState {
176        let status = match self.snapshot() {
177            Ok(status) => status,
178            Err(error) => {
179                return AuthorityState::Unavailable {
180                    cause: error.name().into(),
181                };
182            }
183        };
184        match status.readiness {
185            "ready" => {
186                status
187                    .cursor
188                    .map_or(AuthorityState::NotReady, |revision| AuthorityState::Ready {
189                        log: self.log.clone(),
190                        revision,
191                        generation: status.generation,
192                    })
193            }
194            "not_ready" => AuthorityState::NotReady,
195            "gap" => match (status.cursor, status.missing) {
196                (Some(after), Some(missing)) => AuthorityState::Gap { after, missing },
197                _ => AuthorityState::Unavailable {
198                    cause: "revocation_stream_gap".into(),
199                },
200            },
201            "reset" => AuthorityState::Reset,
202            _ => AuthorityState::Unavailable {
203                cause: status
204                    .last_error
205                    .unwrap_or("revocation_stream_unavailable")
206                    .into(),
207            },
208        }
209    }
210}
211
212impl GrantDecisionSource for StreamState {
213    fn decide(
214        &self,
215        log: &GrantLogId,
216        grants: &[GrantId],
217        at_least: Revision,
218    ) -> Result<DecisionReply, RevocationRefusal> {
219        let inner = self
220            .inner
221            .lock()
222            .map_err(|_| RevocationRefusal::StreamUnavailable)?;
223        if log.identity() != self.log.identity || log.epoch() != self.log.epoch {
224            return Err(RevocationRefusal::LogReset);
225        }
226        if inner.diagnostics.readiness != "ready" {
227            return Err(RevocationRefusal::StreamUnavailable);
228        }
229        let revision = inner
230            .diagnostics
231            .cursor
232            .ok_or(RevocationRefusal::RevisionUnavailable)?;
233        if revision < at_least.get() {
234            return Err(RevocationRefusal::RevisionUnavailable);
235        }
236        // The authenticated baseline lists every visible revoked ancestor. Absence
237        // means not revoked only after contiguous replay and its Ready barrier.
238        let answer = DecisionReply {
239            log: log.clone(),
240            revision: Revision::new(revision),
241            decisions: grants
242                .iter()
243                .map(|grant| (grant.clone(), inner.revoked.contains(grant)))
244                .collect(),
245        };
246        drop(inner);
247        Ok(answer)
248    }
249}
liminal_server/auth_lys/grant_stream_owner/state.rs

liminal_server/auth_lys/grant_stream_owner/
state.rs
