liminal_server/services/
lifecycle_place.rs

1//! Placement of a service record in Lys (ACCESS-014): the one path `create`
2//! and the catch-up after listening both take, with the same operation ids,
3//! so a repeat at Lys is the same operation, never a second resource.
4
5use super::{Lifecycle, LifecycleError, Placer, ServiceState, digest};
6
7impl Lifecycle {
8    /// Places `liminal.service:<id>` and then each `liminal.subject` under
9    /// `liminal.network:<network>`, stopping at the first refusal.
10    pub(super) fn place_children(
11        &self,
12        placer: &dyn Placer,
13        id: &str,
14        subjects: &[String],
15    ) -> Result<(), LifecycleError> {
16        let network = ("liminal.network", self.network.as_str());
17        let children = std::iter::once(("liminal.service", id)).chain(
18            subjects
19                .iter()
20                .map(|subject| ("liminal.subject", subject.as_str())),
21        );
22        for child in children {
23            let operation = format!("op-{}", digest(&["liminal-place/v1", child.0, child.1]));
24            placer
25                .place(&operation, child, network)
26                .map_err(LifecycleError::Placement)?;
27            tracing::info!(
28                service = id,
29                kind = child.0,
30                child = child.1,
31                "SERVICE_PLACED"
32            );
33        }
34        Ok(())
35    }
36
37    /// Places every record that is neither retired nor placed, marks each
38    /// whose children all placed and commits the registry once; answers
39    /// each record left unplaced with its refusal. With no placer it places
40    /// nothing.
41    #[must_use]
42    pub fn catch_up(&self) -> Vec<(String, String)> {
43        let Some(placer) = self.placer.as_ref() else {
44            return Vec::new();
45        };
46        let pending: Vec<(String, Vec<String>)> = match self.lock() {
47            Ok(registry) => registry
48                .records
49                .values()
50                .filter(|record| !record.placed && record.state != ServiceState::Retired)
51                .map(|record| (record.id.clone(), record.subjects.clone()))
52                .collect(),
53            Err(error) => return vec![(String::new(), error.to_string())],
54        };
55        let mut refused = Vec::new();
56        let mut marked = Vec::new();
57        for (id, subjects) in pending {
58            match self.place_children(placer.as_ref(), &id, &subjects) {
59                Ok(()) => marked.push(id),
60                Err(error) => refused.push((id, error.to_string())),
61            }
62        }
63        if marked.is_empty() {
64            return refused;
65        }
66        if let Err(error) = self.mark_placed(&marked) {
67            refused.extend(marked.into_iter().map(|id| (id, error.to_string())));
68        }
69        refused
70    }
71
72    fn mark_placed(&self, ids: &[String]) -> Result<(), LifecycleError> {
73        let mut registry = self.lock()?;
74        let mut snapshot = super::site::storage::Snapshot {
75            records: registry.records.clone(),
76            construction: registry.construction.clone(),
77        };
78        for id in ids {
79            if let Some(record) = snapshot.records.get_mut(id) {
80                record.placed = true;
81            }
82        }
83        let committed = super::site::storage::commit(self, &mut registry, &snapshot);
84        drop(registry);
85        committed
86    }
87}
liminal_server/services/lifecycle_place.rs

liminal_server/services/
lifecycle_place.rs
