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

liminal_server/services/
lifecycle.rs
