use std::collections::{BTreeMap, HashMap};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};
use serde::{Deserialize, Serialize};
use super::lys_http::LysHttp;
use crate::config::templates::is_segment;
use crate::config::types::ServiceTemplateConfig;
#[path = "lifecycle_run.rs"]
mod run;
const APP: &str = "liminal";
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ServiceState {
Asleep,
Live,
Held,
Retired,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Transition {
pub from: ServiceState,
pub to: ServiceState,
pub holder: String,
pub at_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub note: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ServiceRecord {
pub id: String,
pub name: String,
pub template: String,
pub subjects: Vec<String>,
pub state: ServiceState,
pub created_by: String,
pub transitions: Vec<Transition>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum LifecycleError {
UnknownTemplate(String),
NameInvalid,
NameTaken(String),
UnknownService(String),
NotNow {
verb: &'static str,
state: ServiceState,
running: bool,
},
Placement(String),
NoPlacer,
Store(String),
Process(String),
}
impl std::fmt::Display for LifecycleError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::UnknownTemplate(name) => write!(formatter, "unknown_template: {name}"),
Self::NameInvalid => formatter
.write_str("name_invalid: a name is one segment of letters, digits, '-' or '_'"),
Self::NameTaken(name) => write!(formatter, "name_taken: {name}"),
Self::UnknownService(id) => write!(formatter, "unknown_service: {id}"),
Self::NotNow {
verb,
state,
running,
} => write!(
formatter,
"not_now: {verb} does not apply to a service {state:?} with a process {}",
if *running { "running" } else { "stopped" }
),
Self::Placement(why) => write!(formatter, "placement_refused: {why}"),
Self::NoPlacer => {
formatter.write_str("placement_unavailable: no app_credential_path is configured")
}
Self::Store(why) => write!(formatter, "registry_unavailable: {why}"),
Self::Process(why) => write!(formatter, "process_failed: {why}"),
}
}
}
impl std::error::Error for LifecycleError {}
pub trait Placer: Send + Sync + std::fmt::Debug {
fn place(
&self,
operation: &str,
child: (&str, &str),
parent: (&str, &str),
) -> Result<(), String>;
}
pub struct LysPlacer {
http: Arc<dyn LysHttp>,
credential: String,
}
impl LysPlacer {
#[must_use]
pub fn new(http: Arc<dyn LysHttp>, credential: String) -> Self {
Self { http, credential }
}
}
impl std::fmt::Debug for LysPlacer {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.debug_struct("LysPlacer").finish_non_exhaustive()
}
}
impl Placer for LysPlacer {
fn place(
&self,
operation: &str,
child: (&str, &str),
parent: (&str, &str),
) -> Result<(), String> {
let route = format!("apps/{APP}/placements");
let body = serde_json::json!({
"operation": operation,
"child": {"kind": child.0, "id": child.1},
"parent": {"kind": parent.0, "id": parent.1},
});
let answer = self
.http
.post(&self.credential, &route, &body)
.map_err(|error| error.0)?;
if answer.succeeded() {
Ok(())
} else {
Err(answer.refusal())
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ProcessGrant {
pub service: String,
pub subjects: Vec<String>,
}
impl ProcessGrant {
#[must_use]
pub fn serves(&self, subject: &str) -> bool {
self.subjects.iter().any(|own| own == subject)
}
}
const ADDRESS_WAIT: std::time::Duration = std::time::Duration::from_secs(5);
#[derive(Debug)]
pub struct Lifecycle {
templates: BTreeMap<String, ServiceTemplateConfig>,
dir: PathBuf,
network: String,
server: std::sync::OnceLock<String>,
told: (Mutex<()>, std::sync::Condvar),
placer: Option<Arc<dyn Placer>>,
registry: Mutex<Registry>,
}
#[derive(Debug, Default)]
struct Registry {
records: BTreeMap<String, ServiceRecord>,
running: HashMap<String, run::Running>,
tokens: HashMap<[u8; 32], ProcessGrant>,
stopping_all: bool,
}
impl Lifecycle {
pub fn open(
templates: &[ServiceTemplateConfig],
dir: PathBuf,
network: String,
placer: Option<Arc<dyn Placer>>,
) -> Result<Arc<Self>, LifecycleError> {
std::fs::create_dir_all(&dir).map_err(|error| store(&dir, &error))?;
let mut records = BTreeMap::new();
for entry in std::fs::read_dir(&dir).map_err(|error| store(&dir, &error))? {
let path = entry.map_err(|error| store(&dir, &error))?.path();
if path.extension().is_none_or(|ext| ext != "json") {
continue;
}
let bytes = std::fs::read(&path).map_err(|error| store(&path, &error))?;
let record: ServiceRecord = serde_json::from_slice(&bytes).map_err(|error| {
LifecycleError::Store(format!(
"{} is not a service record: {error}",
path.display()
))
})?;
records.insert(record.id.clone(), record);
}
Ok(Arc::new(Self {
templates: templates
.iter()
.map(|template| (template.name.clone(), template.clone()))
.collect(),
dir,
network,
server: std::sync::OnceLock::new(),
told: (Mutex::new(()), std::sync::Condvar::new()),
placer,
registry: Mutex::new(Registry {
records,
..Registry::default()
}),
}))
}
pub fn listening(&self, address: String) {
let _first = self.server.set(address);
let (lock, told) = &self.told;
drop(
lock.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner),
);
told.notify_all();
}
pub(super) fn server_address(&self) -> Option<&str> {
if let Some(address) = self.server.get() {
return Some(address);
}
let (lock, told) = &self.told;
drop(
told.wait_timeout_while(
lock.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner),
ADDRESS_WAIT,
|()| self.server.get().is_none(),
)
.unwrap_or_else(std::sync::PoisonError::into_inner),
);
self.server.get().map(String::as_str)
}
fn lock(&self) -> Result<MutexGuard<'_, Registry>, LifecycleError> {
self.registry
.lock()
.map_err(|_| LifecycleError::Store("the registry lock is poisoned".to_owned()))
}
pub fn list(
&self,
visible: impl Fn(&ServiceRecord) -> bool,
) -> Result<serde_json::Value, LifecycleError> {
let registry = self.lock()?;
let services: Vec<serde_json::Value> = registry
.records
.values()
.filter(|record| visible(record))
.map(|record| {
let mut value = serde_json::to_value(record).unwrap_or_default();
if let Some(object) = value.as_object_mut() {
object.insert(
"running".to_owned(),
registry.running.contains_key(&record.id).into(),
);
}
value
})
.collect();
drop(registry);
let templates: Vec<serde_json::Value> = self
.templates
.values()
.map(|template| {
serde_json::json!({"name": template.name, "subjects": template.subjects})
})
.collect();
Ok(serde_json::json!({"templates": templates, "services": services}))
}
pub fn record(&self, id: &str) -> Result<ServiceRecord, LifecycleError> {
self.lock()?
.records
.get(id)
.cloned()
.ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))
}
pub fn create(
&self,
template: &str,
name: &str,
holder: &str,
operation: &str,
now_ms: u64,
) -> Result<ServiceRecord, LifecycleError> {
let found = self
.templates
.get(template)
.ok_or_else(|| LifecycleError::UnknownTemplate(template.to_owned()))?;
if !is_segment(name) {
return Err(LifecycleError::NameInvalid);
}
let id = format!("svc-{}", digest(&["liminal-service/v1", holder, operation]));
{
let registry = self.lock()?;
if let Some(record) = registry.records.get(&id) {
return Ok(record.clone());
}
if registry
.records
.values()
.any(|record| record.template == template && record.name == name)
{
return Err(LifecycleError::NameTaken(name.to_owned()));
}
}
let subjects: Vec<String> = found
.subjects
.iter()
.map(|root| format!("{root}.{name}"))
.collect();
let placer = self.placer.as_ref().ok_or(LifecycleError::NoPlacer)?;
let network = ("liminal.network", self.network.as_str());
let mut children = vec![("liminal.service", id.as_str())];
children.extend(
subjects
.iter()
.map(|subject| ("liminal.subject", subject.as_str())),
);
for child in children {
let op = format!("op-{}", digest(&["liminal-place/v1", child.0, child.1]));
placer
.place(&op, child, network)
.map_err(LifecycleError::Placement)?;
}
let record = ServiceRecord {
id: id.clone(),
name: name.to_owned(),
template: template.to_owned(),
subjects,
state: ServiceState::Asleep,
created_by: holder.to_owned(),
transitions: vec![Transition {
from: ServiceState::Asleep,
to: ServiceState::Asleep,
holder: holder.to_owned(),
at_ms: now_ms,
note: Some("created".to_owned()),
}],
};
let mut registry = self.lock()?;
if let Some(raced) = registry.records.get(&id) {
return Ok(raced.clone());
}
self.keep(&record)?;
registry.records.insert(id, record.clone());
drop(registry);
Ok(record)
}
pub fn deploy(
self: &Arc<Self>,
id: &str,
holder: &str,
) -> Result<ServiceRecord, LifecycleError> {
self.start(id, holder, "deploy", ServiceState::Asleep)
}
pub fn wake(self: &Arc<Self>, id: &str, holder: &str) -> Result<ServiceRecord, LifecycleError> {
self.start(id, holder, "wake", ServiceState::Held)
}
fn start(
self: &Arc<Self>,
id: &str,
holder: &str,
verb: &'static str,
from: ServiceState,
) -> Result<ServiceRecord, LifecycleError> {
let mut registry = self.lock()?;
let record = registry
.records
.get(id)
.cloned()
.ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))?;
let running = registry.running.contains_key(id);
if record.state != from || running || registry.stopping_all {
return Err(LifecycleError::NotNow {
verb,
state: record.state,
running,
});
}
let template = self
.templates
.get(&record.template)
.ok_or_else(|| LifecycleError::UnknownTemplate(record.template.clone()))?;
let (process, digest) = run::start(self, &record, template, holder)?;
registry.tokens.insert(
digest,
ProcessGrant {
service: record.id.clone(),
subjects: record.subjects.clone(),
},
);
registry.running.insert(record.id.clone(), process);
drop(registry);
Ok(record)
}
pub fn hold(
&self,
id: &str,
holder: &str,
now_ms: u64,
) -> Result<ServiceRecord, LifecycleError> {
self.stop(id, holder, now_ms, "hold", ServiceState::Held)
}
pub fn retire(
&self,
id: &str,
holder: &str,
now_ms: u64,
) -> Result<ServiceRecord, LifecycleError> {
self.stop(id, holder, now_ms, "retire", ServiceState::Retired)
}
fn stop(
&self,
id: &str,
holder: &str,
now_ms: u64,
verb: &'static str,
to: ServiceState,
) -> Result<ServiceRecord, LifecycleError> {
let (state, stopping) = {
let mut registry = self.lock()?;
let state = registry
.records
.get(id)
.map(|record| record.state)
.ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))?;
let running = registry.running.contains_key(id);
let applies = match to {
ServiceState::Held => state == ServiceState::Live || running,
_ => state != ServiceState::Retired,
};
if !applies {
return Err(LifecycleError::NotNow {
verb,
state,
running,
});
}
let process = revoke(&mut registry, id);
drop(registry);
(state, process)
};
if let Some(process) = stopping {
run::stop(&process, self.timeout_of(id)).map_err(LifecycleError::Process)?;
}
self.transition(id, state, to, holder, now_ms, None)
}
fn timeout_of(&self, id: &str) -> u64 {
self.lock()
.ok()
.and_then(|registry| {
registry
.records
.get(id)
.map(|record| record.template.clone())
})
.and_then(|template| self.templates.get(&template).map(|t| t.stop_timeout_ms))
.unwrap_or(1)
}
fn transition(
&self,
id: &str,
from: ServiceState,
to: ServiceState,
holder: &str,
now_ms: u64,
note: Option<String>,
) -> Result<ServiceRecord, LifecycleError> {
let mut registry = self.lock()?;
let mut record = registry
.records
.get(id)
.cloned()
.ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))?;
record.state = to;
record.transitions.push(Transition {
from,
to,
holder: holder.to_owned(),
at_ms: now_ms,
note,
});
self.keep(&record)?;
registry.records.insert(id.to_owned(), record.clone());
drop(registry);
Ok(record)
}
#[must_use]
pub fn admit_process(&self, token: &[u8]) -> Option<ProcessGrant> {
let digest = *blake3::hash(token).as_bytes();
self.lock().ok()?.tokens.get(&digest).cloned()
}
pub fn mark_live(&self, id: &str, now_ms: u64) -> Result<(), LifecycleError> {
let (state, starter) = {
let registry = self.lock()?;
let state = registry
.records
.get(id)
.map(|record| record.state)
.ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))?;
let starter = registry
.running
.get(id)
.map_or_else(|| "process".to_owned(), |process| process.starter.clone());
drop(registry);
(state, starter)
};
if state == ServiceState::Live {
return Ok(());
}
self.transition(id, state, ServiceState::Live, &starter, now_ms, None)
.map(|_| ())
}
pub fn restart_live(self: &Arc<Self>, now_ms: u64) -> Vec<(String, LifecycleError)> {
let live: Vec<(String, String)> = self.lock().map_or_else(
|_| Vec::new(),
|registry| {
registry
.records
.values()
.filter(|record| record.state == ServiceState::Live)
.map(|record| {
let starter = record
.transitions
.last()
.map_or_else(|| record.created_by.clone(), |t| t.holder.clone());
(record.id.clone(), starter)
})
.collect()
},
);
let mut failed = Vec::new();
for (id, starter) in live {
if let Err(error) = self.start(&id, &starter, "restart", ServiceState::Live) {
let note = Some(format!("restart failed: {error}"));
let _kept = self.transition(
&id,
ServiceState::Live,
ServiceState::Asleep,
"process",
now_ms,
note,
);
failed.push((id, error));
}
}
failed
}
#[must_use]
pub fn placed_since(&self, subject: &str, since_ms: u64) -> bool {
self.lock().is_ok_and(|registry| {
registry.records.values().any(|record| {
record.state != ServiceState::Retired
&& record.subjects.iter().any(|served| served == subject)
&& record
.transitions
.first()
.is_some_and(|created| created.at_ms >= since_ms)
})
})
}
pub fn shutdown_begun(&self) {
if let Ok(mut registry) = self.lock() {
registry.stopping_all = true;
}
}
pub fn shutdown(&self) {
self.shutdown_begun();
let ids: Vec<String> = match self.lock() {
Ok(registry) => registry.running.keys().cloned().collect(),
Err(_) => return,
};
for id in ids {
let timeout = self.timeout_of(&id);
let process = match self.lock() {
Ok(mut registry) => revoke(&mut registry, &id),
Err(_) => return,
};
if let Some(process) = process
&& let Err(error) = run::stop(&process, timeout)
{
tracing::warn!(service = %id, error, "a service process did not stop at shutdown");
}
}
}
fn exited(&self, id: &str, status: &str, now_ms: u64) {
let (asked, stopping) = match self.lock() {
Ok(mut registry) => {
let asked = registry.running.remove(id).is_none_or(|process| {
registry.tokens.remove(&process.digest);
false
});
(asked, registry.stopping_all)
}
Err(_) => return,
};
if asked {
return;
}
if stopping {
tracing::info!(
service = id,
status,
"a service process exited during shutdown; its state stays"
);
return;
}
let Ok(record) = self.record(id) else {
return;
};
let note = Some(format!("exited unasked: {status}"));
if let Err(error) = self.transition(
id,
record.state,
ServiceState::Asleep,
"process",
now_ms,
note,
) {
tracing::warn!(service = id, %error, "a service exit could not be recorded");
}
}
fn keep(&self, record: &ServiceRecord) -> Result<(), LifecycleError> {
let path = self.dir.join(format!("{}.json", record.id));
let temporary = self.dir.join(format!("{}.json.new", record.id));
let bytes = serde_json::to_vec_pretty(record)
.map_err(|error| LifecycleError::Store(error.to_string()))?;
let mut file =
std::fs::File::create(&temporary).map_err(|error| store(&temporary, &error))?;
file.write_all(&bytes)
.and_then(|()| file.sync_all())
.map_err(|error| store(&temporary, &error))?;
std::fs::rename(&temporary, &path).map_err(|error| store(&path, &error))?;
std::fs::File::open(&self.dir)
.and_then(|dir| dir.sync_all())
.map_err(|error| store(&self.dir, &error))
}
fn log_path(&self, id: &str) -> PathBuf {
self.dir.join(format!("{id}.log"))
}
}
fn revoke(registry: &mut Registry, id: &str) -> Option<run::Running> {
let process = registry.running.remove(id)?;
registry.tokens.remove(&process.digest);
Some(process)
}
fn store(path: &Path, error: &std::io::Error) -> LifecycleError {
LifecycleError::Store(format!("{}: {}", path.display(), error.kind()))
}
fn digest(parts: &[&str]) -> String {
let mut hasher = blake3::Hasher::new();
for part in parts {
hasher.update(&u64::try_from(part.len()).unwrap_or(u64::MAX).to_be_bytes());
hasher.update(part.as_bytes());
}
hasher.finalize().to_hex()[..32].to_owned()
}
#[cfg(test)]
#[path = "lifecycle_tests.rs"]
mod tests;