liminal_server/services/
lifecycle.rs

1//! The services registry and its lifecycle (SRV-028 R2 to R5).
2//!
3//! Read-only configuration templates and approved live records share the same
4//! process lifecycle. Construction begins with contracts and an empty template,
5//! then a held act records the program that completion leaves asleep.
6//!
7//! The durable store keeps one complete bounded snapshot for construction,
8//! including prepared acts, approvals, receipts and notification intents. An old
9//! file registry is imported once before admission; historical records remain.
10//! Caller authority is decided through the admitted pass before mutations.
11
12use std::collections::{BTreeMap, HashMap};
13use std::path::{Path, PathBuf};
14use std::sync::{Arc, Mutex, MutexGuard};
15
16use crate::config::templates::is_segment;
17use crate::config::types::ServiceTemplateConfig;
18
19#[path = "lifecycle_run.rs"]
20mod run;
21
22#[path = "plan.rs"]
23pub mod plan;
24#[path = "site/processes.rs"]
25mod processes;
26#[path = "site.rs"]
27pub mod site;
28#[path = "templates.rs"]
29pub mod templates;
30#[path = "trial.rs"]
31pub mod trial;
32
33/// The Lys app that owns the `liminal.*` kinds.
34const APP: &str = "liminal";
35
36#[path = "site/lifecycle_records.rs"]
37mod records;
38pub use records::{ServiceRecord, ServiceState, Transition};
39
40/// Why a lifecycle request did nothing, by name.
41#[derive(Clone, Debug, PartialEq, Eq)]
42pub enum LifecycleError {
43    /// No template of that name.
44    UnknownTemplate(String),
45    /// The name is not one segment.
46    NameInvalid,
47    /// A service of that template already has that name.
48    NameTaken(String),
49    /// No service of that id.
50    UnknownService(String),
51    /// The verb does not apply in the service's state.
52    NotNow {
53        /// The verb asked.
54        verb: &'static str,
55        /// The state it was asked in.
56        state: ServiceState,
57        /// Whether a process was running.
58        running: bool,
59    },
60    /// Lys refused or could not be asked to place the service.
61    Placement(String),
62    /// No app credential is configured, so nothing can be placed.
63    NoPlacer,
64    /// The registry could not be read or written.
65    Store(String),
66    /// The template's command could not be started or stopped.
67    Process(String),
68}
69
70impl std::fmt::Display for LifecycleError {
71    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
72        match self {
73            Self::UnknownTemplate(name) => write!(formatter, "unknown_template: {name}"),
74            Self::NameInvalid => formatter
75                .write_str("name_invalid: a name is one segment of letters, digits, '-' or '_'"),
76            Self::NameTaken(name) => write!(formatter, "name_taken: {name}"),
77            Self::UnknownService(id) => write!(formatter, "unknown_service: {id}"),
78            Self::NotNow {
79                verb,
80                state,
81                running,
82            } => write!(
83                formatter,
84                "not_now: {verb} does not apply to a service {state:?} with a process {}",
85                if *running { "running" } else { "stopped" }
86            ),
87            Self::Placement(why) => write!(formatter, "placement_refused: {why}"),
88            Self::NoPlacer => {
89                formatter.write_str("placement_unavailable: no app_credential_path is configured")
90            }
91            Self::Store(why) => write!(formatter, "registry_unavailable: {why}"),
92            Self::Process(why) => write!(formatter, "process_failed: {why}"),
93        }
94    }
95}
96
97impl std::error::Error for LifecycleError {}
98
99/// Places a resource under its parent in Lys.
100pub trait Placer: Send + Sync + std::fmt::Debug {
101    /// Places `child` (kind, id) under `parent` (kind, id) once, under
102    /// `operation`.
103    ///
104    /// # Errors
105    /// Lys's refusal by its name, or why it could not be asked.
106    fn place(
107        &self,
108        operation: &str,
109        child: (&str, &str),
110        parent: (&str, &str),
111    ) -> Result<(), String>;
112}
113
114#[path = "site/placement.rs"]
115mod placement;
116pub use placement::LysPlacer;
117
118/// What a service process's own credential admits: registering its
119/// subjects, and nothing else.
120#[derive(Clone, Debug, PartialEq, Eq)]
121pub struct ProcessGrant {
122    /// The service the process runs.
123    pub service: String,
124    /// The only subjects it may register.
125    pub subjects: Vec<String>,
126}
127
128impl ProcessGrant {
129    /// Whether the process may serve `subject`.
130    #[must_use]
131    pub fn serves(&self, subject: &str) -> bool {
132        self.subjects.iter().any(|own| own == subject)
133    }
134}
135
136/// How long a start admitted before the bound address is handed over waits
137/// to be told it.
138const ADDRESS_WAIT: std::time::Duration = std::time::Duration::from_secs(5);
139
140/// The registry, its templates and the processes it started.
141pub struct Lifecycle {
142    templates: BTreeMap<String, ServiceTemplateConfig>,
143    dir: PathBuf,
144    network: String,
145    /// The address a started process connects to, set once the listener binds.
146    server: std::sync::OnceLock<String>,
147    /// Woken when `server` is set: the accept worker starts at bind, before
148    /// the runtime hands the address over, so a start admitted in between
149    /// waits to be told it rather than being refused.
150    told: (Mutex<()>, std::sync::Condvar),
151    placer: Option<Arc<dyn Placer>>,
152    registry: Mutex<Registry>,
153    site_config: std::sync::OnceLock<site::storage::SiteBindings>,
154    changed: std::sync::Condvar,
155    site_console: std::sync::OnceLock<std::sync::Weak<crate::console::ConsoleHub>>,
156}
157
158#[derive(Default)]
159struct Registry {
160    records: BTreeMap<String, ServiceRecord>,
161    running: HashMap<String, run::Running>,
162    tokens: HashMap<[u8; 32], ProcessGrant>,
163    stopping_all: bool,
164    construction: site::Construction,
165    construction_hash: Option<haematite::tree::Hash>,
166    trials: BTreeMap<String, BTreeMap<u64, trial::Registration>>,
167    poisoned: bool,
168}
169
170impl Lifecycle {
171    /// Opens the registry in `dir`, reading every record once.
172    ///
173    /// `network` is the Lys network services are placed under; the address a
174    /// started process connects to is given by [`Self::listening`].
175    ///
176    /// # Errors
177    /// The directory cannot be made or read, or a record in it is not one.
178    pub fn open(
179        templates: &[ServiceTemplateConfig],
180        dir: PathBuf,
181        network: String,
182        placer: Option<Arc<dyn Placer>>,
183    ) -> Result<Arc<Self>, LifecycleError> {
184        site::create_private_dir(&dir).map_err(|error| store(&dir, &error))?;
185        let snapshot = site::storage::load(&dir)?;
186        Ok(Self::open_snapshot(
187            templates, dir, network, placer, snapshot,
188        ))
189    }
190
191    fn open_snapshot(
192        templates: &[ServiceTemplateConfig],
193        dir: PathBuf,
194        network: String,
195        placer: Option<Arc<dyn Placer>>,
196        snapshot: site::storage::Snapshot,
197    ) -> Arc<Self> {
198        Arc::new(Self {
199            templates: templates
200                .iter()
201                .map(|template| (template.name.clone(), template.clone()))
202                .collect(),
203            dir,
204            network,
205            server: std::sync::OnceLock::new(),
206            told: (Mutex::new(()), std::sync::Condvar::new()),
207            placer,
208            site_config: std::sync::OnceLock::new(),
209            changed: std::sync::Condvar::new(),
210            site_console: std::sync::OnceLock::new(),
211            registry: Mutex::new(Registry {
212                records: snapshot.records,
213                construction: snapshot.construction,
214                ..Registry::default()
215            }),
216        })
217    }
218
219    /// Records the address the server listens on; a process is started only
220    /// after it is known. A second call changes nothing.
221    pub fn listening(&self, address: String) {
222        let _first = self.server.set(address);
223        let (lock, told) = &self.told;
224        drop(
225            lock.lock()
226                .unwrap_or_else(std::sync::PoisonError::into_inner),
227        );
228        told.notify_all();
229    }
230
231    /// The address the server listens on, waiting up to [`ADDRESS_WAIT`] to be
232    /// told it: the runtime sets it straight after the bind, so the wait is
233    /// only ever the few instructions between the two; the bound covers a
234    /// listener that never binds.
235    pub(super) fn server_address(&self) -> Option<&str> {
236        if let Some(address) = self.server.get() {
237            return Some(address);
238        }
239        let (lock, told) = &self.told;
240        drop(
241            told.wait_timeout_while(
242                lock.lock()
243                    .unwrap_or_else(std::sync::PoisonError::into_inner),
244                ADDRESS_WAIT,
245                |()| self.server.get().is_none(),
246            )
247            .unwrap_or_else(std::sync::PoisonError::into_inner),
248        );
249        self.server.get().map(String::as_str)
250    }
251
252    fn lock(&self) -> Result<MutexGuard<'_, Registry>, LifecycleError> {
253        let registry = self
254            .registry
255            .lock()
256            .map_err(|_| LifecycleError::Store("registry_lock_poisoned".into()))?;
257        if registry.poisoned {
258            return Err(LifecycleError::Store(
259                "registry_commit_uncertain: reopen required".into(),
260            ));
261        }
262        Ok(registry)
263    }
264
265    /// The templates and every service `visible` keeps, with whether a
266    /// process is running for it, as the `services.list` answer.
267    ///
268    /// # Errors
269    /// The registry lock is poisoned.
270    pub fn list(
271        &self,
272        visible: impl Fn(&ServiceRecord) -> bool,
273    ) -> Result<serde_json::Value, LifecycleError> {
274        let registry = self.lock()?;
275        let services: Vec<serde_json::Value> = registry
276            .records
277            .values()
278            .filter(|record| visible(record))
279            .map(|record| {
280                let mut value = serde_json::to_value(record)
281                    .map_err(|_| LifecycleError::Store("service_encoding_failed".into()))?;
282                if let Some(object) = value.as_object_mut() {
283                    object.insert(
284                        "running".to_owned(),
285                        registry.running.contains_key(&record.id).into(),
286                    );
287                }
288                Ok(value)
289            })
290            .collect::<Result<_, LifecycleError>>()?;
291        let mut templates: Vec<serde_json::Value> = self
292            .templates
293            .values()
294            .map(|template| {
295                serde_json::json!({"name": template.name, "subjects": template.subjects})
296            })
297            .collect();
298        templates.extend(registry.construction.templates.values().filter(|record| !record.retired)
299            .map(|record| serde_json::json!({"name":record.proposal.name,"subjects":record.proposal.subjects})));
300        drop(registry);
301        Ok(serde_json::json!({"templates": templates, "services": services}))
302    }
303
304    /// The record of `id`.
305    ///
306    /// # Errors
307    /// No such service, or the lock is poisoned.
308    pub fn record(&self, id: &str) -> Result<ServiceRecord, LifecycleError> {
309        self.lock()?
310            .records
311            .get(id)
312            .cloned()
313            .ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))
314    }
315
316    /// Creates a service named `name` from `template` for `holder`, asleep,
317    /// after placing it and its subjects in Lys. The same `operation` from
318    /// the same holder answers the same service and places nothing twice.
319    ///
320    /// # Errors
321    /// By name: the template, the name, the placement or the store.
322    pub fn create(
323        &self,
324        template: &str,
325        name: &str,
326        holder: &str,
327        operation: &str,
328        now_ms: u64,
329    ) -> Result<ServiceRecord, LifecycleError> {
330        let found = self
331            .templates
332            .get(template)
333            .ok_or_else(|| LifecycleError::UnknownTemplate(template.to_owned()))?;
334        if !is_segment(name) {
335            return Err(LifecycleError::NameInvalid);
336        }
337        let id = format!("svc-{}", digest(&["liminal-service/v1", holder, operation]));
338        {
339            let registry = self.lock()?;
340            if let Some(record) = registry.records.get(&id) {
341                return Ok(record.clone());
342            }
343            if registry
344                .records
345                .values()
346                .any(|record| record.template.as_deref() == Some(template) && record.name == name)
347            {
348                return Err(LifecycleError::NameTaken(name.to_owned()));
349            }
350        }
351        let subjects: Vec<String> = found
352            .subjects
353            .iter()
354            .map(|root| format!("{root}.{name}"))
355            .collect();
356        let placer = self.placer.as_ref().ok_or(LifecycleError::NoPlacer)?;
357        let network = ("liminal.network", self.network.as_str());
358        let mut children = vec![("liminal.service", id.as_str())];
359        children.extend(
360            subjects
361                .iter()
362                .map(|subject| ("liminal.subject", subject.as_str())),
363        );
364        for child in children {
365            let op = format!("op-{}", digest(&["liminal-place/v1", child.0, child.1]));
366            placer
367                .place(&op, child, network)
368                .map_err(LifecycleError::Placement)?;
369        }
370        let record = ServiceRecord {
371            id: id.clone(),
372            name: name.to_owned(),
373            template: Some(template.to_owned()),
374            purpose: None,
375            contracts: BTreeMap::new(),
376            builder: None,
377            builder_grants: None,
378            builder_bound: false,
379            approved_by: Vec::new(),
380            responsible: None,
381            plan: plan::Plan::default(),
382            approvals: Vec::new(),
383            subjects,
384            state: ServiceState::Asleep,
385            created_by: holder.to_owned(),
386            transitions: vec![Transition {
387                from: ServiceState::Asleep,
388                to: ServiceState::Asleep,
389                holder: holder.to_owned(),
390                at_ms: now_ms,
391                note: Some("created".to_owned()),
392            }],
393        };
394        let mut registry = self.lock()?;
395        if let Some(raced) = registry.records.get(&id) {
396            return Ok(raced.clone());
397        }
398        self.keep(&mut registry, &record)?;
399        registry.records.insert(id, record.clone());
400        drop(registry);
401        Ok(record)
402    }
403
404    /// Whether a service serving `subject`, not retired, was placed in Lys at
405    /// or after `since_ms`: a pass issued before then cannot carry its
406    /// rights, so its offline refusal says nothing and Lys is asked live.
407    #[must_use]
408    pub fn placed_since(&self, subject: &str, since_ms: u64) -> bool {
409        self.lock().is_ok_and(|registry| {
410            registry.records.values().any(|record| {
411                record.state != ServiceState::Retired
412                    && record.subjects.iter().any(|served| served == subject)
413                    && record
414                        .transitions
415                        .first()
416                        .is_some_and(|created| created.at_ms >= since_ms)
417            })
418        })
419    }
420
421    fn keep(&self, registry: &mut Registry, record: &ServiceRecord) -> Result<(), LifecycleError> {
422        site::storage::keep(self, registry, record)
423    }
424
425    /// The service's log file.
426    fn log_path(&self, id: &str) -> PathBuf {
427        self.dir.join(format!("{id}.log"))
428    }
429}
430
431/// Takes a running process out of the registry and revokes its credential,
432/// marking it asked to stop.
433fn revoke(registry: &mut Registry, id: &str) -> Option<run::Running> {
434    let process = registry.running.remove(id)?;
435    registry.tokens.remove(&process.digest);
436    Some(process)
437}
438
439fn store(path: &Path, error: &std::io::Error) -> LifecycleError {
440    LifecycleError::Store(format!("{}: {}", path.display(), error.kind()))
441}
442
443/// 32 hex digits of a domain-separated digest of `parts`.
444fn digest(parts: &[&str]) -> String {
445    let mut hasher = blake3::Hasher::new();
446    for part in parts {
447        hasher.update(&u64::try_from(part.len()).unwrap_or(u64::MAX).to_be_bytes());
448        hasher.update(part.as_bytes());
449    }
450    hasher.finalize().to_hex()[..32].to_owned()
451}
452
453#[cfg(test)]
454#[path = "lifecycle_tests.rs"]
455mod tests;
456
457impl std::fmt::Debug for Lifecycle {
458    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
459        out.debug_struct("Lifecycle").finish_non_exhaustive()
460    }
461}
liminal_server/services/lifecycle.rs

liminal_server/services/
lifecycle.rs
