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

liminal_server/services/
site.rs
