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    /// The network an open exchange's services are recorded under (ACCESS-012
172    /// R1): there is no Lys network, and the word says so in every record.
173    pub const OPEN_NETWORK: &str = "open";
174
175    /// Opens the registry in `dir`, reading every record once.
176    ///
177    /// `network` is the Lys network services are placed under; the address a
178    /// started process connects to is given by [`Self::listening`].
179    ///
180    /// # Errors
181    /// The directory cannot be made or read, or a record in it is not one.
182    pub fn open(
183        templates: &[ServiceTemplateConfig],
184        dir: PathBuf,
185        network: String,
186        placer: Option<Arc<dyn Placer>>,
187    ) -> Result<Arc<Self>, LifecycleError> {
188        site::create_private_dir(&dir).map_err(|error| store(&dir, &error))?;
189        let snapshot = site::storage::load(&dir)?;
190        Ok(Self::open_snapshot(
191            templates, dir, network, placer, snapshot,
192        ))
193    }
194
195    fn open_snapshot(
196        templates: &[ServiceTemplateConfig],
197        dir: PathBuf,
198        network: String,
199        placer: Option<Arc<dyn Placer>>,
200        snapshot: site::storage::Snapshot,
201    ) -> Arc<Self> {
202        Arc::new(Self {
203            templates: templates
204                .iter()
205                .map(|template| (template.name.clone(), template.clone()))
206                .collect(),
207            dir,
208            network,
209            server: std::sync::OnceLock::new(),
210            told: (Mutex::new(()), std::sync::Condvar::new()),
211            placer,
212            site_config: std::sync::OnceLock::new(),
213            changed: std::sync::Condvar::new(),
214            site_console: std::sync::OnceLock::new(),
215            registry: Mutex::new(Registry {
216                records: snapshot.records,
217                construction: snapshot.construction,
218                ..Registry::default()
219            }),
220        })
221    }
222
223    /// Whether a new service is placed in Lys: a gated exchange with an app
224    /// connector places, an open exchange (ACCESS-012) never does.
225    #[must_use]
226    pub fn places(&self) -> bool {
227        self.placer.is_some()
228    }
229
230    /// Records the address the server listens on; a process is started only
231    /// after it is known. A second call changes nothing.
232    pub fn listening(&self, address: String) {
233        let _first = self.server.set(address);
234        let (lock, told) = &self.told;
235        drop(
236            lock.lock()
237                .unwrap_or_else(std::sync::PoisonError::into_inner),
238        );
239        told.notify_all();
240    }
241
242    /// The address the server listens on, waiting up to [`ADDRESS_WAIT`] to be
243    /// told it: the runtime sets it straight after the bind, so the wait is
244    /// only ever the few instructions between the two; the bound covers a
245    /// listener that never binds.
246    pub(super) fn server_address(&self) -> Option<&str> {
247        if let Some(address) = self.server.get() {
248            return Some(address);
249        }
250        let (lock, told) = &self.told;
251        drop(
252            told.wait_timeout_while(
253                lock.lock()
254                    .unwrap_or_else(std::sync::PoisonError::into_inner),
255                ADDRESS_WAIT,
256                |()| self.server.get().is_none(),
257            )
258            .unwrap_or_else(std::sync::PoisonError::into_inner),
259        );
260        self.server.get().map(String::as_str)
261    }
262
263    fn lock(&self) -> Result<MutexGuard<'_, Registry>, LifecycleError> {
264        let registry = self
265            .registry
266            .lock()
267            .map_err(|_| LifecycleError::Store("registry_lock_poisoned".into()))?;
268        if registry.poisoned {
269            return Err(LifecycleError::Store(
270                "registry_commit_uncertain: reopen required".into(),
271            ));
272        }
273        Ok(registry)
274    }
275
276    /// The templates and every service `visible` keeps, with whether a
277    /// process is running for it, as the `services.list` answer.
278    ///
279    /// # Errors
280    /// The registry lock is poisoned.
281    pub fn list(
282        &self,
283        visible: impl Fn(&ServiceRecord) -> bool,
284    ) -> Result<serde_json::Value, LifecycleError> {
285        let registry = self.lock()?;
286        let services: Vec<serde_json::Value> = registry
287            .records
288            .values()
289            .filter(|record| visible(record))
290            .map(|record| {
291                let mut value = serde_json::to_value(record)
292                    .map_err(|_| LifecycleError::Store("service_encoding_failed".into()))?;
293                if let Some(object) = value.as_object_mut() {
294                    object.insert(
295                        "running".to_owned(),
296                        registry.running.contains_key(&record.id).into(),
297                    );
298                }
299                Ok(value)
300            })
301            .collect::<Result<_, LifecycleError>>()?;
302        let mut templates: Vec<serde_json::Value> = self
303            .templates
304            .values()
305            .map(|template| {
306                serde_json::json!({"name": template.name, "subjects": template.subjects})
307            })
308            .collect();
309        templates.extend(registry.construction.templates.values().filter(|record| !record.retired)
310            .map(|record| serde_json::json!({"name":record.proposal.name,"subjects":record.proposal.subjects})));
311        drop(registry);
312        Ok(serde_json::json!({"templates": templates, "services": services}))
313    }
314
315    /// The record of `id`.
316    ///
317    /// # Errors
318    /// No such service, or the lock is poisoned.
319    pub fn record(&self, id: &str) -> Result<ServiceRecord, LifecycleError> {
320        self.lock()?
321            .records
322            .get(id)
323            .cloned()
324            .ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))
325    }
326
327    /// Creates a service named `name` from `template` for `holder`, asleep,
328    /// after placing it and its subjects in Lys. The same `operation` from
329    /// the same holder answers the same service and places nothing twice.
330    ///
331    /// # Errors
332    /// By name: the template, the name, the placement or the store.
333    pub fn create(
334        &self,
335        template: &str,
336        name: &str,
337        holder: &str,
338        operation: &str,
339        now_ms: u64,
340    ) -> Result<ServiceRecord, LifecycleError> {
341        let found = self
342            .templates
343            .get(template)
344            .ok_or_else(|| LifecycleError::UnknownTemplate(template.to_owned()))?;
345        if !is_segment(name) {
346            return Err(LifecycleError::NameInvalid);
347        }
348        let id = format!("svc-{}", digest(&["liminal-service/v1", holder, operation]));
349        {
350            let registry = self.lock()?;
351            if let Some(record) = registry.records.get(&id) {
352                return Ok(record.clone());
353            }
354            if registry
355                .records
356                .values()
357                .any(|record| record.template.as_deref() == Some(template) && record.name == name)
358            {
359                return Err(LifecycleError::NameTaken(name.to_owned()));
360            }
361        }
362        let subjects: Vec<String> = found
363            .subjects
364            .iter()
365            .map(|root| format!("{root}.{name}"))
366            .collect();
367        // ACCESS-012 R2: an open exchange's registry places nothing in Lys;
368        // every other registry without a placer refuses by name.
369        if let Some(placer) = self.placer.as_ref() {
370            let network = ("liminal.network", self.network.as_str());
371            let mut children = vec![("liminal.service", id.as_str())];
372            children.extend(
373                subjects
374                    .iter()
375                    .map(|subject| ("liminal.subject", subject.as_str())),
376            );
377            for child in children {
378                let op = format!("op-{}", digest(&["liminal-place/v1", child.0, child.1]));
379                placer
380                    .place(&op, child, network)
381                    .map_err(LifecycleError::Placement)?;
382            }
383        } else if self.network != Self::OPEN_NETWORK {
384            return Err(LifecycleError::NoPlacer);
385        }
386        let record = ServiceRecord {
387            id: id.clone(),
388            name: name.to_owned(),
389            template: Some(template.to_owned()),
390            purpose: None,
391            contracts: BTreeMap::new(),
392            builder: None,
393            builder_grants: None,
394            builder_bound: false,
395            approved_by: Vec::new(),
396            responsible: None,
397            plan: plan::Plan::default(),
398            approvals: Vec::new(),
399            subjects,
400            state: ServiceState::Asleep,
401            created_by: holder.to_owned(),
402            transitions: vec![Transition {
403                from: ServiceState::Asleep,
404                to: ServiceState::Asleep,
405                holder: holder.to_owned(),
406                at_ms: now_ms,
407                note: Some("created".to_owned()),
408            }],
409        };
410        let mut registry = self.lock()?;
411        if let Some(raced) = registry.records.get(&id) {
412            return Ok(raced.clone());
413        }
414        self.keep(&mut registry, &record)?;
415        registry.records.insert(id, record.clone());
416        drop(registry);
417        Ok(record)
418    }
419
420    /// Whether a service serving `subject`, not retired, was placed in Lys at
421    /// or after `since_ms`: a pass issued before then cannot carry its
422    /// rights, so its offline refusal says nothing and Lys is asked live.
423    #[must_use]
424    pub fn placed_since(&self, subject: &str, since_ms: u64) -> bool {
425        self.lock().is_ok_and(|registry| {
426            registry.records.values().any(|record| {
427                record.state != ServiceState::Retired
428                    && record.subjects.iter().any(|served| served == subject)
429                    && record
430                        .transitions
431                        .first()
432                        .is_some_and(|created| created.at_ms >= since_ms)
433            })
434        })
435    }
436
437    fn keep(&self, registry: &mut Registry, record: &ServiceRecord) -> Result<(), LifecycleError> {
438        site::storage::keep(self, registry, record)
439    }
440
441    /// The service's log file.
442    fn log_path(&self, id: &str) -> PathBuf {
443        self.dir.join(format!("{id}.log"))
444    }
445}
446
447/// Takes a running process out of the registry and revokes its credential,
448/// marking it asked to stop.
449fn revoke(registry: &mut Registry, id: &str) -> Option<run::Running> {
450    let process = registry.running.remove(id)?;
451    registry.tokens.remove(&process.digest);
452    Some(process)
453}
454
455fn store(path: &Path, error: &std::io::Error) -> LifecycleError {
456    LifecycleError::Store(format!("{}: {}", path.display(), error.kind()))
457}
458
459/// 32 hex digits of a domain-separated digest of `parts`.
460fn digest(parts: &[&str]) -> String {
461    let mut hasher = blake3::Hasher::new();
462    for part in parts {
463        hasher.update(&u64::try_from(part.len()).unwrap_or(u64::MAX).to_be_bytes());
464        hasher.update(part.as_bytes());
465    }
466    hasher.finalize().to_hex()[..32].to_owned()
467}
468
469#[cfg(test)]
470#[path = "lifecycle_tests.rs"]
471mod tests;
472
473impl std::fmt::Debug for Lifecycle {
474    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
475        out.debug_struct("Lifecycle").finish_non_exhaustive()
476    }
477}
liminal_server/services/lifecycle.rs

liminal_server/services/
lifecycle.rs
