aion-server 0.31.0

//! The boot-time answer to a record that outlived its workflow.
//!
//! An auto-provisioned worker deployment exists to serve a `harness` section
//! of one deployed workflow type. When that type is gone from the package
//! catalogue — a document withdrawn, an example removed by an upgrade, the
//! assistant becoming a session rather than a workflow — the record still
//! names a document nothing will ever start, and the supervisor would spend
//! its restart budget on a worker with nothing to serve. This pass reads the
//! catalogue once at boot and withdraws every auto record whose staged type no
//! deployed package carries, by name, with the same outcome the redeploy path
//! records for a queue a document stopped declaring.
//!
//! A catalogue that cannot be read withdraws nothing: an unreadable answer is
//! not "no packages", and the caller logs the refusal instead.

use std::collections::BTreeSet;

use crate::ServerState;

use super::documents::sanitise_workflow_type;
use super::outcome::AutoWorkerOutcome;
use super::provision::{OPERATION, log_outcome};
use super::record;
use super::retire::{WithdrawCause, withdraw};

/// Every workflow type a deployed package carries, spelled as staged
/// directories spell it, so a record's staged type compares directly.
///
/// # Errors
///
/// The reason the catalogue could not be read — no engine behind this state,
/// or the store refusing the listing. The caller must treat either as "unknown",
/// never as "empty".
pub async fn deployed_workflow_types(state: &ServerState) -> Result<BTreeSet<String>, String> {
    let engine = state
        .engine()
        .map_err(|error| format!("no engine to read the package catalogue through: {error}"))?;
    let packages = engine
        .store()
        .list_packages()
        .await
        .map_err(|error| format!("the package catalogue could not be listed: {error}"))?;
    Ok(packages
        .iter()
        .map(|package| sanitise_workflow_type(&package.workflow_type))
        .collect())
}

/// Withdraw every auto-provisioned record whose staged workflow type is not in
/// `deployed`, and record the outcomes on the managed-worker report.
///
/// Records an operator wrote are never touched — they name no staged document
/// and were never this module's to retire. A record that lists but does not
/// decode is left alone and named in the log: what it serves cannot be read,
/// so it cannot be judged.
pub async fn withdraw_orphaned(
    state: &ServerState,
    deployed: &BTreeSet<String>,
) -> Vec<AutoWorkerOutcome> {
    let listing = match state
        .worker_deployment_store()
        .list_worker_deployments()
        .await
    {
        Ok(listing) => listing,
        Err(error) => {
            tracing::error!(
                operation = OPERATION,
                %error,
                "the worker deployments could not be listed, so this boot cannot tell whether an \
                 auto-provisioned record outlived its workflow"
            );
            return Vec::new();
        }
    };
    for poisoned in &listing.undecodable {
        tracing::warn!(
            operation = OPERATION,
            record = %poisoned.name,
            error = %poisoned.error,
            "a worker deployment record could not be decoded, so whether it outlived its \
             workflow cannot be judged; it was left as it is"
        );
    }
    let mut outcomes = Vec::new();
    for deployment in &listing.deployments {
        if !record::is_auto_name(&deployment.name) {
            continue;
        }
        let Some(workflow_type) = record::staged_workflow_type(&deployment.artifact) else {
            continue;
        };
        if deployed.contains(&workflow_type) {
            continue;
        }
        let outcome = withdraw(state, &workflow_type, deployment, WithdrawCause::NoPackage).await;
        log_outcome(&outcome);
        outcomes.push(outcome);
    }
    if !outcomes.is_empty() {
        state.worker_supervisor().record_auto_provision(&outcomes);
    }
    outcomes
}

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