liminal_server/services/
site.rs

1//! Contracts-first construction, with immutable acts and atomic receipts.
2
3use std::os::unix::fs::PermissionsExt;
4use std::sync::Arc;
5
6use serde::{Deserialize, Serialize};
7
8use super::{Lifecycle, LifecycleError, ServiceRecord, ServiceState, Transition};
9use crate::config::templates::is_segment;
10
11#[path = "site/acts.rs"]
12mod acts;
13#[path = "site/effects.rs"]
14pub mod effects;
15#[path = "site/executor.rs"]
16pub mod executor;
17#[path = "site/outright.rs"]
18mod outright;
19#[path = "site/preparation.rs"]
20pub mod preparation;
21#[path = "site/records.rs"]
22mod records;
23#[path = "site/storage.rs"]
24pub mod storage;
25#[path = "site/topology.rs"]
26mod topology;
27pub use records::{
28    Approval, BuilderGrants, ConstructionConfig, Contract, SiteAct, SiteChange, SiteLimits,
29    SiteSpec,
30};
31
32/// All construction refusals contain boundary names, never rejected payloads.
33#[derive(Debug, thiserror::Error)]
34pub enum SiteError {
35    /// The boundary named here is missing or invalid.
36    #[error("site_invalid: {0}")]
37    Invalid(&'static str),
38    /// A configured count or byte bound was reached.
39    #[error("site_bound_exceeded: {0}")]
40    Bound(&'static str),
41    /// Safe business words generated from validated routing metadata.
42    #[error("{0}")]
43    Refused(String),
44    /// Admission is poisoned after an uncertain durable operation.
45    #[error("site_store_unavailable")]
46    Store,
47    /// The admitted pass's own refusal, without token or payload bytes.
48    #[error("site_pass_refused: {0}")]
49    Pass(String),
50    /// The existing lifecycle failed.
51    #[error("{0}")]
52    Lifecycle(#[from] LifecycleError),
53}
54
55/// The schema strings are carried exactly; equivalence compares parsed shapes.
56///
57/// # Errors
58/// Missing or invalid schema, reserved subject, or configured boundary refusal.
59pub fn validate_contracts(spec: &SiteSpec, limits: &SiteLimits) -> Result<(), SiteError> {
60    if !is_segment(&spec.name) || spec.name.len() > limits.text_bytes.get() {
61        return Err(SiteError::Invalid("name"));
62    }
63    if spec.purpose.trim().is_empty()
64        || spec.purpose.contains(['\n', '\r'])
65        || spec.purpose.len() > limits.text_bytes.get()
66    {
67        return Err(SiteError::Invalid("purpose: one sentence"));
68    }
69    if spec.subjects.is_empty() || spec.subjects.len() > limits.subjects.get() {
70        return Err(SiteError::Bound("subjects"));
71    }
72    let mut seen = std::collections::BTreeSet::new();
73    for subject in &spec.subjects {
74        if subject.len() > limits.text_bytes.get()
75            || !subject.split('.').all(is_segment)
76            || subject.starts_with("console.")
77            || subject.starts_with("agents.")
78            || subject.starts_with("services.")
79            || crate::server::connection::services::is_site_route(subject)
80            || !seen.insert(subject)
81        {
82            return Err(SiteError::Invalid("subject"));
83        }
84        let contract = spec
85            .contracts
86            .get(subject)
87            .ok_or_else(|| SiteError::Refused(format!("subject {subject} has no schema")))?;
88        if contract.input_schema_json.is_empty() || contract.output_schema_json.is_empty() {
89            return Err(SiteError::Refused(format!(
90                "subject {subject} has no schema"
91            )));
92        }
93        for (kind, bytes) in [
94            ("input", &contract.input_schema_json),
95            ("output", &contract.output_schema_json),
96        ] {
97            if bytes.len() > limits.schema_bytes.get() {
98                return Err(SiteError::Bound("schema_bytes"));
99            }
100            let value = serde_json::from_str(bytes).map_err(|_| {
101                SiteError::Refused(format!("subject {subject} has no valid {kind} schema"))
102            })?;
103            liminal::channel::Schema::new(value).map_err(|_| {
104                SiteError::Refused(format!("subject {subject} has no valid {kind} schema"))
105            })?;
106        }
107    }
108    if spec.contracts.len() != seen.len() {
109        return Err(SiteError::Invalid("contracts name undeclared subjects"));
110    }
111    Ok(())
112}
113
114impl Lifecycle {
115    /// Opens construction with explicit migration bounds before parsing any old
116    /// file. Ordinary configuration-only registries retain their existing path.
117    ///
118    /// # Errors
119    /// Invalid old snapshot, nonregular file, count/byte bound, or migration write.
120    pub fn open_bounded(
121        templates: &[crate::config::types::ServiceTemplateConfig],
122        dir: std::path::PathBuf,
123        network: String,
124        placer: Option<Arc<dyn super::Placer>>,
125        limits: &SiteLimits,
126    ) -> Result<Arc<Self>, LifecycleError> {
127        std::fs::create_dir_all(&dir).map_err(|error| super::store(&dir, &error))?;
128        if std::fs::metadata(&dir)
129            .map_err(|error| super::store(&dir, &error))?
130            .permissions()
131            .mode()
132            & 0o022
133            != 0
134        {
135            return Err(LifecycleError::Store(
136                "construction_registry_writable_by_others".into(),
137            ));
138        }
139        let snapshot = storage::load_bounded(&dir, limits)?;
140        Ok(Self::open_snapshot(
141            templates, dir, network, placer, snapshot,
142        ))
143    }
144
145    /// Installs operator bounds and the grant effect once, before admission.
146    /// The retained projection shares the existing scheduler and inbox bounds.
147    ///
148    /// # Errors
149    /// Repeated installation or invalid restored construction records.
150    pub fn configure_sites(
151        &self,
152        limits: SiteLimits,
153        store: Arc<dyn liminal::durability::DurableStore>,
154        templates: Option<super::templates::TemplatePolicy>,
155        effects: Arc<dyn effects::SiteEffects>,
156        retained: Arc<crate::agents::retained::Retained>,
157    ) -> Result<(), SiteError> {
158        if std::fs::metadata(&self.dir)
159            .map_err(|_| SiteError::Store)?
160            .permissions()
161            .mode()
162            & 0o022
163            != 0
164        {
165            return Err(SiteError::Invalid(
166                "construction registry must not be writable by group or others",
167            ));
168        }
169        self.site_config
170            .set(storage::SiteBindings {
171                store,
172                limits,
173                templates,
174                effects,
175                retained,
176            })
177            .map_err(|_| SiteError::Invalid("construction already configured"))?;
178        let mut registry = self.lock()?;
179        storage::restore(self, &mut registry)?;
180        drop(registry);
181        self.validate_site_storage()?;
182        let registry = self.lock()?;
183        for record in registry
184            .records
185            .values()
186            .filter(|record| record.purpose.is_some())
187        {
188            validate_contracts(
189                &SiteSpec {
190                    name: record.name.clone(),
191                    purpose: record.purpose.clone().ok_or(SiteError::Store)?,
192                    subjects: record.subjects.clone(),
193                    contracts: record.contracts.clone(),
194                },
195                &self.site_bindings()?.limits,
196            )?;
197            self.publish_plan(record)?;
198        }
199        drop(registry);
200        Ok(())
201    }
202
203    pub(super) fn site_bindings(&self) -> Result<&storage::SiteBindings, SiteError> {
204        self.site_config.get().ok_or(SiteError::Invalid(
205            "[auth.lys.construction] is not configured",
206        ))
207    }
208
209    pub(super) fn building(&self, id: &str) -> Result<ServiceRecord, SiteError> {
210        let record = self.record(id)?;
211        if record.state != ServiceState::Building {
212            return Err(SiteError::Refused(format!(
213                "service {} is not building",
214                record.name
215            )));
216        }
217        Ok(record)
218    }
219
220    pub(super) fn publish_plan(&self, record: &ServiceRecord) -> Result<(), SiteError> {
221        let retained = &self.site_bindings()?.retained;
222        let subject = format!("services.{}.plan", record.name);
223        let value = super::plan::current_value(&record.plan);
224        if retained
225            .value(&subject)
226            .map_err(|_| SiteError::Store)?
227            .as_ref()
228            == Some(&value)
229        {
230            return Ok(());
231        }
232        retained
233            .publish(&[(subject, value)])
234            .map_err(|_| SiteError::Store)
235    }
236
237    pub(super) fn site_transition(
238        record: &mut ServiceRecord,
239        holder: &str,
240        at_ms: u64,
241        words: &str,
242    ) {
243        record.transitions.push(Transition {
244            from: record.state,
245            to: record.state,
246            holder: holder.into(),
247            at_ms,
248            note: Some(words.into()),
249        });
250    }
251}
252
253/// A construction snapshot never exposes its free text through Debug.
254#[derive(Clone, Default, Serialize, Deserialize)]
255#[serde(deny_unknown_fields)]
256pub(super) struct Construction {
257    pub templates: std::collections::BTreeMap<String, super::templates::TemplateRecord>,
258    pub drafts: std::collections::BTreeMap<String, executor::Prepared>,
259    pub outcomes: std::collections::BTreeMap<String, executor::Outcome>,
260}
261
262#[cfg(test)]
263#[path = "site/test_support.rs"]
264pub(crate) mod test_support;
265#[cfg(test)]
266#[path = "site_tests.rs"]
267mod tests;
268
269impl Lifecycle {
270    /// Network placement used for the caller's exact permission binding.
271    #[must_use]
272    pub fn network_id(&self) -> &str {
273        &self.network
274    }
275    /// Returns the explicitly configured positive construction limits.
276    ///
277    /// # Errors
278    /// Construction is not configured.
279    pub fn construction_limits(&self) -> Result<SiteLimits, SiteError> {
280        Ok(self.site_bindings()?.limits.clone())
281    }
282    /// Subscribes to the current plan before any later changes, under its writer lock.
283    ///
284    /// # Errors
285    /// Missing subject, watcher bound, schema or inbox refusal.
286    pub fn subscribe_plan(
287        &self,
288        id: u64,
289        subject: &str,
290        schemas: &[liminal::protocol::SchemaId],
291        install: Option<liminal::channel::InboxInstall>,
292    ) -> Result<crate::server::connection::services::ConnectionSubscription, SiteError> {
293        self.site_bindings()?
294            .retained
295            .subscribe(id, subject, schemas, install)
296            .map_err(|error| SiteError::Refused(error.to_string()))
297    }
298}
299
300impl Lifecycle {
301    /// Installs the existing hub's event signal without another publisher or timer.
302    ///
303    /// # Errors
304    /// The signal was already bound.
305    pub fn bind_site_console(
306        &self,
307        console: &Arc<crate::console::ConsoleHub>,
308    ) -> Result<(), SiteError> {
309        self.site_console
310            .set(Arc::downgrade(console))
311            .map_err(|_| SiteError::Invalid("site console already bound"))
312    }
313    /// Repairs retained current values after a durable outcome or restart.
314    ///
315    /// # Errors
316    /// A target, retained stream or hub signal failed.
317    pub fn refresh_sites(&self) -> Result<(), SiteError> {
318        let registry = self.lock()?;
319        for record in registry
320            .records
321            .values()
322            .filter(|record| record.purpose.is_some())
323        {
324            self.publish_plan(record)?;
325        }
326        drop(registry);
327        self.notify_sites()
328    }
329    pub(super) fn notify_sites(&self) -> Result<(), SiteError> {
330        if let Some(console) = self.site_console.get().and_then(std::sync::Weak::upgrade) {
331            console
332                .agent_changed()
333                .map_err(|_| SiteError::Invalid("site topology signal failed"))?;
334        }
335        Ok(())
336    }
337}
liminal_server/services/site.rs

liminal_server/services/
site.rs
