liminal_server/auth_lys/
grant_stream_owner.rs

1//! One owner of Lys's consumer-opened grant log, stopped and joined before store release.
2
3#[path = "grant_stream_owner/binding.rs"]
4mod binding;
5#[path = "grant_stream_owner/config.rs"]
6mod config;
7#[path = "grant_stream_owner/configured.rs"]
8mod configured;
9#[path = "grant_stream_owner/engine.rs"]
10mod engine;
11#[path = "grant_stream_owner/http.rs"]
12mod http;
13#[path = "grant_stream_owner/state.rs"]
14mod state;
15
16pub use config::{GrantStreamConfig, GrantSubjectConfig};
17pub use state::{StreamDiagnostics, StreamHealth, StreamState};
18
19use crate::auth_lys::live::drafts::{DraftDecisionHook, PullReport};
20use crate::config::types::LysConfig;
21use lys_identity::grants::change_stream::{ChangesPage, ChangesRequest, LogName};
22use std::sync::Arc;
23use std::thread::JoinHandle;
24
25/// Only stable local names cross diagnostics, errors, Debug and tracing.
26#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
27#[error("{name}")]
28pub struct StreamError {
29    name: &'static str,
30    transport_ended: bool,
31}
32
33impl StreamError {
34    pub(super) const fn named(name: &'static str) -> Self {
35        Self {
36            name,
37            transport_ended: false,
38        }
39    }
40    const fn transport() -> Self {
41        Self {
42            name: "revocation_stream_unavailable",
43            transport_ended: true,
44        }
45    }
46    /// The named failure, containing no transport error or remote response text.
47    #[must_use]
48    pub const fn name(self) -> &'static str {
49        self.name
50    }
51}
52
53/// The guard must remain on the server start stack until its shutdown finalizer joins it.
54pub struct GrantStreamOwner {
55    state: Arc<StreamState>,
56    worker: Option<JoinHandle<Result<(), StreamError>>>,
57    draft_hook: Option<Arc<dyn DraftDecisionHook>>,
58}
59
60impl std::fmt::Debug for GrantStreamOwner {
61    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62        f.debug_struct("GrantStreamOwner")
63            .field("state", &self.state)
64            .field("worker", &self.worker.is_some())
65            .finish_non_exhaustive()
66    }
67}
68
69impl GrantStreamOwner {
70    /// Starts the configured owner on the standalone server's existing ask runtime.
71    ///
72    /// # Errors
73    /// A missing runtime or any refusal from [`Self::start_configured`].
74    pub fn start_for_server(
75        config: &crate::config::types::ServerConfig,
76        supervisor: &crate::server::ConnectionSupervisor,
77        health: &crate::health::HealthServerHandle,
78        runtime: Option<&tokio::runtime::Runtime>,
79        draft_hook: Option<Arc<dyn DraftDecisionHook>>,
80    ) -> Result<Option<Self>, crate::ServerError> {
81        let Some(lys) = config.auth.as_ref().and_then(|auth| auth.lys.as_ref()) else {
82            return Ok(None);
83        };
84        if lys.membership.is_none() && lys.grant_stream.is_none() {
85            return Ok(None);
86        }
87        let runtime = runtime.ok_or_else(|| config_error("grant_stream_runtime_missing"))?;
88        Self::start_configured(
89            lys,
90            supervisor,
91            health,
92            runtime.handle().clone(),
93            draft_hook,
94        )
95    }
96
97    /// Installs both authorities before any connection and starts exactly one read.
98    ///
99    /// The supplied handle is the server's already owned Lys runtime; shutdown
100    /// joins this guard before that runtime or the store can be released.
101    ///
102    /// # Errors
103    /// Refuses invalid bounds, unreadable credentials or keys, a foreign membership
104    /// log, a second owner or authority, and failure to spawn the owned worker.
105    pub fn start_configured(
106        lys: &LysConfig,
107        supervisor: &crate::server::ConnectionSupervisor,
108        health: &crate::health::HealthServerHandle,
109        handle: tokio::runtime::Handle,
110        draft_hook: Option<Arc<dyn DraftDecisionHook>>,
111    ) -> Result<Option<Self>, crate::ServerError> {
112        let Some(config) = lys.grant_stream.as_ref() else {
113            if lys.membership.is_some() {
114                return Err(config_error(
115                    "auth.lys.membership requires auth.lys.grant_stream",
116                ));
117            }
118            return Ok(None);
119        };
120        let configured::Installation {
121            state,
122            engine,
123            http,
124        } = configured::prepare(lys, config, supervisor, health, handle.clone())?;
125        let worker_state = Arc::clone(&state);
126        let worker = std::thread::Builder::new()
127            .name("liminal-grants".into())
128            .spawn(move || {
129                let result = handle.block_on(read(engine, http));
130                if let Err(error) = result {
131                    worker_state.update(|inner| {
132                        if !matches!(inner.diagnostics.readiness, "gap" | "reset") {
133                            inner.diagnostics.readiness = "terminal";
134                        }
135                        inner.diagnostics.last_error = Some(error.name());
136                        inner.diagnostics.stopped = true;
137                        Ok(())
138                    })?;
139                } else {
140                    worker_state.update(|inner| {
141                        inner.diagnostics.readiness = "unavailable";
142                        inner.diagnostics.last_error = Some("grant_stream_stopped");
143                        inner.diagnostics.stopped = true;
144                        Ok(())
145                    })?;
146                }
147                result
148            })
149            .map_err(|_| config_error("grant_stream_worker_unavailable"))?;
150        Ok(Some(Self {
151            state,
152            worker: Some(worker),
153            draft_hook,
154        }))
155    }
156
157    /// The health/progress authority installed at start.
158    #[must_use]
159    pub fn state(&self) -> Arc<StreamState> {
160        Arc::clone(&self.state)
161    }
162
163    /// Hook for an explicitly accepted published draft decision, never a grant change.
164    /// The current Lys frame contract has no such variant; unknown rows remain terminal.
165    ///
166    /// # Errors
167    /// A missing hook or a named approved-pull failure, retained without draft words.
168    pub async fn draft_decided(&self) -> Result<PullReport, StreamError> {
169        let hook = self
170            .draft_hook
171            .as_ref()
172            .ok_or(StreamError::named("grant_stream_draft_hook_missing"))?;
173        let result = hook
174            .draft_decided()
175            .await
176            .map_err(|_| StreamError::named("grant_stream_draft_pull_failed"));
177        if let Err(error) = result.as_ref() {
178            self.state.update(|inner| {
179                inner.diagnostics.last_error = Some(error.name());
180                Ok(())
181            })?;
182        }
183        result
184    }
185
186    /// Cancels every pending HTTP ask and joins the worker, including on startup unwind.
187    ///
188    /// # Errors
189    /// The terminal reader failure or an unexpectedly terminated worker.
190    pub fn finish(&mut self) -> Result<(), StreamError> {
191        self.draft_hook = None;
192        self.state.stop.send_replace(true);
193        match self.worker.take() {
194            Some(worker) => worker
195                .join()
196                .map_err(|_| StreamError::named("grant_stream_worker_panicked"))?,
197            None => Ok(()),
198        }
199    }
200}
201
202impl Drop for GrantStreamOwner {
203    fn drop(&mut self) {
204        if let Err(error) = self.finish() {
205            tracing::error!(cause = error.name(), "grant stream shutdown failed");
206        }
207    }
208}
209
210fn config_error(message: &str) -> crate::ServerError {
211    crate::ServerError::ConfigValidation {
212        message: message.into(),
213    }
214}
215fn as_config(error: StreamError) -> crate::ServerError {
216    config_error(error.name())
217}
218
219/// The shutdown finalizer joins the reader before flushing or releasing the store.
220///
221/// # Errors
222/// The owner's terminal failure or a failed worker join, by its stable name.
223pub fn finish_configured(owner: &mut Option<GrantStreamOwner>) -> Result<(), crate::ServerError> {
224    owner
225        .as_mut()
226        .map_or(Ok(()), |owner| owner.finish().map_err(as_config))
227}
228
229pub(crate) fn read_credential(
230    path: &std::path::Path,
231    bound: usize,
232) -> Result<zeroize::Zeroizing<String>, crate::ServerError> {
233    http::read_file(path, bound).map_err(as_config)
234}
235
236async fn read(mut engine: engine::Engine, http: http::Http) -> Result<(), StreamError> {
237    let result = read_pages(&mut engine, &http).await;
238    if let Err(error) = result.as_ref() {
239        engine.terminal(error.name())?;
240    }
241    result
242}
243
244async fn read_pages(engine: &mut engine::Engine, http: &http::Http) -> Result<(), StreamError> {
245    let mut reopens = 0_usize;
246    loop {
247        if *http.stop.borrow() {
248            engine.lost()?;
249            return Ok(());
250        }
251        let before = engine.state.snapshot()?;
252        let question = ChangesRequest {
253            log: Some(LogName {
254                identity: engine.state.log.identity.clone(),
255                epoch: engine.state.log.epoch,
256            }),
257            after: before.cursor,
258            limit: engine.config.page_positions,
259            wait: true,
260            placements: true,
261            placed: before.placed,
262        };
263        engine.state.update(|inner| {
264            inner.diagnostics.requests = inner
265                .diagnostics
266                .requests
267                .checked_add(1)
268                .ok_or(StreamError::named("grant_stream_counter_exhausted"))?;
269            Ok(())
270        })?;
271        let page = match http
272            .post::<_, ChangesPage>("grants/changes", &question)
273            .await
274        {
275            Ok(page) => page,
276            Err(error) if error.name() == "grant_stream_stopped" => {
277                engine.lost()?;
278                return Ok(());
279            }
280            Err(error) if error.transport_ended => {
281                engine.lost()?;
282                if reopens >= engine.config.max_transport_reopens {
283                    return Err(error);
284                }
285                reopens += 1;
286                engine.state.update(|inner| {
287                    inner.diagnostics.reopens = inner
288                        .diagnostics
289                        .reopens
290                        .checked_add(1)
291                        .ok_or(StreamError::named("grant_stream_counter_exhausted"))?;
292                    Ok(())
293                })?;
294                continue;
295            }
296            Err(error) => {
297                engine.lost()?;
298                return Err(error);
299            }
300        };
301        if let Err(error) = engine.page(page) {
302            if error.name() == "grant_stream_membership_failed" && *http.stop.borrow() {
303                engine.lost()?;
304                return Ok(());
305            }
306            return Err(error);
307        }
308        let after = engine.state.snapshot()?;
309        if (before.cursor, before.placed) == (after.cursor, after.placed) {
310            return Err(StreamError::named("grant_stream_no_progress"));
311        }
312        reopens = 0;
313    }
314}
315
316#[cfg(test)]
317#[path = "grant_stream_owner/tests.rs"]
318pub(crate) mod tests;
liminal_server/auth_lys/grant_stream_owner.rs

liminal_server/auth_lys/
grant_stream_owner.rs
