liminal-server 0.27.4

Standalone server for the liminal messaging bus
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

//! Placement of a service record in Lys (ACCESS-014): the one path `create`
//! and the catch-up after listening both take, with the same operation ids,
//! so a repeat at Lys is the same operation, never a second resource.

use super::{Lifecycle, LifecycleError, Placer, ServiceState, digest};

impl Lifecycle {
    /// Places `liminal.service:<id>` and then each `liminal.subject` under
    /// `liminal.network:<network>`, stopping at the first refusal.
    pub(super) fn place_children(
        &self,
        placer: &dyn Placer,
        id: &str,
        subjects: &[String],
    ) -> Result<(), LifecycleError> {
        let network = ("liminal.network", self.network.as_str());
        let children = std::iter::once(("liminal.service", id)).chain(
            subjects
                .iter()
                .map(|subject| ("liminal.subject", subject.as_str())),
        );
        for child in children {
            let operation = format!("op-{}", digest(&["liminal-place/v1", child.0, child.1]));
            placer
                .place(&operation, child, network)
                .map_err(LifecycleError::Placement)?;
            tracing::info!(
                service = id,
                kind = child.0,
                child = child.1,
                "SERVICE_PLACED"
            );
        }
        Ok(())
    }

    /// Places every record that is neither retired nor placed, marks each
    /// whose children all placed and commits the registry once; answers
    /// each record left unplaced with its refusal. With no placer it places
    /// nothing.
    #[must_use]
    pub fn catch_up(&self) -> Vec<(String, String)> {
        let Some(placer) = self.placer.as_ref() else {
            return Vec::new();
        };
        let pending: Vec<(String, Vec<String>)> = match self.lock() {
            Ok(registry) => registry
                .records
                .values()
                .filter(|record| !record.placed && record.state != ServiceState::Retired)
                .map(|record| (record.id.clone(), record.subjects.clone()))
                .collect(),
            Err(error) => return vec![(String::new(), error.to_string())],
        };
        let mut refused = Vec::new();
        let mut marked = Vec::new();
        for (id, subjects) in pending {
            match self.place_children(placer.as_ref(), &id, &subjects) {
                Ok(()) => marked.push(id),
                Err(error) => refused.push((id, error.to_string())),
            }
        }
        if marked.is_empty() {
            return refused;
        }
        if let Err(error) = self.mark_placed(&marked) {
            refused.extend(marked.into_iter().map(|id| (id, error.to_string())));
        }
        refused
    }

    fn mark_placed(&self, ids: &[String]) -> Result<(), LifecycleError> {
        let mut registry = self.lock()?;
        let mut snapshot = super::site::storage::Snapshot {
            records: registry.records.clone(),
            construction: registry.construction.clone(),
        };
        for id in ids {
            if let Some(record) = snapshot.records.get_mut(id) {
                record.placed = true;
            }
        }
        let committed = super::site::storage::commit(self, &mut registry, &snapshot);
        drop(registry);
        committed
    }
}