liminal-server 0.25.4

//! Trial admission is an additional fence over the pass's own serve/call grants.

use super::site::SiteError;
use super::{Lifecycle, ServiceRecord, ServiceState};
use crate::auth_lys::session::LysSession;
use crate::auth_lys::{HotAction, SubjectBinding};
use liminal::protocol::WorkerRegistration;
use std::sync::Arc;

/// A reservation blocks completion while a registration is being installed.
///
/// Dropping an uncommitted reservation rolls back that admission. A successful
/// owner commits it only after storing the registration in its own directory.
pub struct TrialPermit {
    lifecycle: Arc<Lifecycle>,
    service: String,
    connection: u64,
    committed: bool,
}
impl std::fmt::Debug for TrialPermit {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        out.debug_struct("TrialPermit")
            .field("connection", &self.connection)
            .finish_non_exhaustive()
    }
}
impl TrialPermit {
    /// Makes the reservation live after the directory accepted the worker.
    ///
    /// # Errors
    /// The connection closed or its reserved subject changed before commit.
    pub fn commit(mut self) -> Result<(), SiteError> {
        let mut registry = self.lifecycle.lock()?;
        let trial = registry
            .trials
            .get_mut(&self.service)
            .and_then(|trials| trials.get_mut(&self.connection))
            .ok_or(SiteError::Invalid("trial reservation lost"))?;
        trial.live = true;
        self.committed = true;
        drop(registry);
        self.lifecycle.notify_sites()?;
        Ok(())
    }
}

pub(super) struct Registration {
    pub subject: String,
    pub live: bool,
}
impl Drop for TrialPermit {
    fn drop(&mut self) {
        if !self.committed {
            if let Ok(mut registry) = self.lifecycle.lock() {
                if let Some(trials) = registry.trials.get_mut(&self.service) {
                    trials.remove(&self.connection);
                }
                registry.trials.retain(|_, trials| !trials.is_empty());
            } else {
                tracing::error!("trial_reservation_release_failed");
            }
        }
    }
}

fn schema_equal(expected: &str, actual: &str) -> Result<bool, SiteError> {
    let expected: serde_json::Value =
        serde_json::from_str(expected).map_err(|_| SiteError::Invalid("site schema"))?;
    let actual: serde_json::Value =
        serde_json::from_str(actual).map_err(|_| SiteError::Invalid("registered schema"))?;
    Ok(expected == actual)
}

/// Compares every descriptor's schemas to the immutable promise for this subject.
/// A raw/untyped registration cannot satisfy a contracts-first site.
///
/// # Errors
/// Missing contract, missing descriptor, or a subject-named input/output mismatch.
pub fn check_contract(
    record: &ServiceRecord,
    registration: &WorkerRegistration,
) -> Result<(), SiteError> {
    let subject = &registration.task_queue;
    let contract = record.contracts.get(subject).ok_or(SiteError::Invalid(
        "registered subject has no site contract",
    ))?;
    if registration.activities.is_empty() {
        return Err(SiteError::Refused(format!(
            "subject {subject}: registered input schema differs from site"
        )));
    }
    for activity in &registration.activities {
        for (kind, expected, actual) in [
            (
                "input",
                &contract.input_schema_json,
                &activity.input_schema_json,
            ),
            (
                "output",
                &contract.output_schema_json,
                &activity.output_schema_json,
            ),
        ] {
            if !schema_equal(expected, actual).map_err(|_| {
                SiteError::Refused(format!(
                    "subject {subject}: registered {kind} schema differs from site"
                ))
            })? {
                return Err(SiteError::Refused(format!(
                    "subject {subject}: registered {kind} schema differs from site"
                )));
            }
        }
    }
    Ok(())
}

/// Lys passes a builder's serve grant on the network, so the exchange keeps a bound builder to
/// its own site's subjects: while bound it serves nothing else.
fn refuse_builder_elsewhere<'a>(
    records: impl IntoIterator<Item = &'a ServiceRecord>,
    session: Option<&dyn LysSession>,
) -> Result<(), SiteError> {
    let Some(holder) = session
        .filter(|session| session.holder_kind() == Some("agent"))
        .map(LysSession::holder)
    else {
        return Ok(());
    };
    records
        .into_iter()
        .find(|record| record.builder_bound && record.builder.as_deref() == Some(holder))
        .map_or(Ok(()), |site| {
            Err(SiteError::Refused(format!(
                "builder of {} serves only its site's subjects",
                site.name
            )))
        })
}

impl Lifecycle {
    /// Validates schemas and signed builder identity, then reserves completion's
    /// barrier before the directory stores anything. Both checks precede serving.
    ///
    /// # Errors
    /// Foreign holder, schema mismatch, missing serve grant or another trial.
    pub fn prepare_trial(
        self: &Arc<Self>,
        connection: u64,
        session: Option<&dyn LysSession>,
        registration: &WorkerRegistration,
        process: Option<&super::ProcessGrant>,
        at_ms: u64,
    ) -> Result<Option<TrialPermit>, SiteError> {
        if crate::server::connection::services::plan_subject(&registration.task_queue).is_some()
            || crate::server::connection::services::is_site_route(&registration.task_queue)
        {
            return Err(SiteError::Invalid("service metadata is server-owned"));
        }
        let mut registry = self.lock()?;
        let Some(record) = registry
            .records
            .values()
            .find(|record| {
                record.purpose.is_some() && record.subjects.contains(&registration.task_queue)
            })
            .cloned()
        else {
            refuse_builder_elsewhere(registry.records.values(), session)?;
            return Ok(None);
        };
        check_contract(&record, registration)?;
        if registration
            .namespaces
            .iter()
            .any(|namespace| !record.subjects.contains(namespace))
        {
            return Err(SiteError::Invalid("registered namespaces differ from site"));
        }
        if record.state == ServiceState::Retired {
            return Err(SiteError::Refused(format!(
                "service {} is retired",
                record.name
            )));
        }
        if record.state != ServiceState::Building {
            if process.is_some_and(|grant| {
                grant.service == record.id
                    && grant.serves(&registration.task_queue)
                    && registry.running.contains_key(&record.id)
            }) {
                return Ok(None);
            }
            return Err(SiteError::Refused(format!(
                "service {} requires its template process",
                record.name
            )));
        }
        let session = session.ok_or(SiteError::Invalid("trial needs an admitted Lys pass"))?;
        if session.holder_kind() != Some("agent") {
            return Err(SiteError::Invalid("trial holder is not an agent"));
        }
        self.check_plan_holder(&record, session.holder())?;
        // A serve right on the service is explicitly reached in the signed pass;
        // the ordinary subject decision still runs in the registration owner.
        let binding = SubjectBinding::new(
            self.network.clone(),
            registration.task_queue.clone(),
            registration.task_queue.clone(),
        )
        .map_err(|_| SiteError::Invalid("trial subject binding"))?;
        session
            .decide(&binding, HotAction::Serve, at_ms / 1000)
            .map_err(|error| SiteError::Pass(error.to_string()))?;
        if registry
            .trials
            .values()
            .any(|trials| trials.contains_key(&connection))
        {
            return Err(SiteError::Refused(format!(
                "trial still registered on connection {connection}"
            )));
        }
        let trials = registry.trials.entry(record.id.clone()).or_default();
        if let Some((registered, _)) = trials
            .iter()
            .find(|(_, trial)| trial.subject == registration.task_queue)
        {
            return Err(SiteError::Refused(format!(
                "trial still registered on connection {registered}"
            )));
        }
        if trials.len() >= self.site_bindings()?.limits.subjects.get() {
            return Err(SiteError::Bound("trial registrations"));
        }
        trials.insert(
            connection,
            Registration {
                subject: registration.task_queue.clone(),
                live: false,
            },
        );
        drop(registry);
        Ok(Some(TrialPermit {
            lifecycle: Arc::clone(self),
            service: record.id,
            connection,
            committed: false,
        }))
    }

    /// Called by the directory owner for close and replacement, under the same
    /// registration hold that removes the index entry. No timer infers closure.
    ///
    /// # Errors
    /// A poisoned registry prevents an uncertain completion.
    pub fn trial_closed(&self, connection: u64) -> Result<(), SiteError> {
        let mut registry = self.lock()?;
        for trials in registry.trials.values_mut() {
            trials.remove(&connection);
        }
        registry.trials.retain(|_, trials| !trials.is_empty());
        drop(registry);
        Ok(())
    }

    /// Fences every Building subject before routing and again before reply
    /// release. Identity alone permits no call; the caller still needs its grant.
    ///
    /// # Errors
    /// A third party, absent pass or the pass's own call refusal.
    pub fn admit_trial_call(
        &self,
        session: Option<&dyn LysSession>,
        subject: &str,
        at_ms: u64,
    ) -> Result<bool, SiteError> {
        let registry = self.lock()?;
        let Some(record) = registry
            .records
            .values()
            .find(|record| {
                record.state == ServiceState::Building
                    && record.subjects.iter().any(|own| own == subject)
            })
            .cloned()
        else {
            return Ok(false);
        };
        drop(registry);
        let session = session
            .ok_or_else(|| SiteError::Refused(format!("service {} is a trial", record.name)))?;
        let is_builder = record.builder_bound
            && record.builder.as_deref() == Some(session.holder())
            && session.holder_kind() == Some("agent");
        // The approvers of a drafted site, or the person answerable for a site Lys admitted outright.
        let is_approver = session.holder_kind() == Some("person")
            && (record
                .approved_by
                .iter()
                .any(|person| person == session.holder())
                || record.responsible.as_deref() == Some(session.holder()));
        if !is_builder && !is_approver {
            return Err(SiteError::Refused(format!(
                "service {} is a trial",
                record.name
            )));
        }
        let binding = SubjectBinding::new(self.network.clone(), subject.into(), subject.into())
            .map_err(|_| SiteError::Invalid("trial call binding"))?;
        session
            .decide(
                &binding,
                if is_builder {
                    HotAction::Serve
                } else {
                    HotAction::Call
                },
                at_ms / 1000,
            )
            .map_err(|error| SiteError::Pass(error.to_string()))?;
        Ok(true)
    }

    /// Current registered or reserved trial connection, used by topology.
    ///
    /// # Errors
    /// The registry lock is poisoned.
    pub fn trial_connection(&self, service: &str) -> Result<Option<u64>, SiteError> {
        Ok(self.lock()?.trials.get(service).and_then(|trials| {
            trials
                .iter()
                .find(|(_, trial)| trial.live)
                .map(|(connection, _)| *connection)
        }))
    }
}

#[cfg(test)]
#[path = "trial_tests.rs"]
mod tests;