aion-server 0.30.0

//! Withdrawing what a document no longer asks for, and reclaiming what nothing
//! reads any more.
//!
//! Minting is only half of "auto-provision keys on the `harness` section". If
//! removing a section left the record standing, the rule would hold only at the
//! moment a document was first deployed — and a worker would stay alive
//! replaying a launch the document stopped declaring, which is the same
//! staleness a re-mint exists to prevent, wearing different clothes.

use std::collections::BTreeSet;
use std::path::{Path, PathBuf};

use aion_core::ClusterEvent;
use aion_store::{WorkerDeployment, WorkerDeploymentListing};

use crate::ServerState;
use crate::worker::supervisor::{Convergence, converge_and_report};

use super::documents;
use super::listener::worker_dial_address;
use super::outcome::{AutoWorkerDecision, AutoWorkerOutcome};
use super::provision::{OPERATION, log_outcome};
use super::queues::HarnessQueue;
use super::record;

/// Withdraw every `auto/…` record this workflow type minted for a queue its
/// document no longer declares.
///
/// Attribution runs through the staged document directory the record's own argv
/// names (`<type>@<digest>/…`), so a record is only ever withdrawn by the
/// workflow type that minted it: two documents declaring different queues never
/// retire each other's workers.
pub(super) async fn withdraw_undeclared(
    state: &ServerState,
    workflow_type: &str,
    declared: &[HarnessQueue],
    listing: &WorkerDeploymentListing,
) -> Vec<AutoWorkerOutcome> {
    let declared: BTreeSet<&str> = declared
        .iter()
        .map(|queue| queue.task_queue.as_str())
        .collect();
    let mut outcomes = Vec::new();
    for stale in listing.deployments.iter().filter(|deployment| {
        record::is_auto_name(&deployment.name)
            && record::staged_workflow_type(&deployment.artifact).as_deref() == Some(workflow_type)
            && !declared.contains(deployment.task_queue.as_str())
    }) {
        outcomes.push(withdraw(state, workflow_type, stale, WithdrawCause::UndeclaredQueue).await);
    }
    outcomes
}

/// Why an auto-provisioned worker deployment is being withdrawn — the one
/// sentence every outcome for it opens with.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum WithdrawCause {
    /// The document was redeployed without a `harness` section for this queue.
    UndeclaredQueue,
    /// No deployed package carries the workflow type the record was staged for.
    NoPackage,
}

impl WithdrawCause {
    fn sentence(self, workflow_type: &str, task_queue: &str) -> String {
        match self {
            Self::UndeclaredQueue => format!(
                "`{workflow_type}` no longer declares a `harness` section for task queue \
                 `{task_queue}`"
            ),
            Self::NoPackage => format!(
                "no deployed package carries workflow type `{workflow_type}`, which task queue \
                 `{task_queue}` was auto-provisioned to serve"
            ),
        }
    }
}

/// Stop a doomed auto worker, remove its record, and say what happened.
pub(super) async fn withdraw(
    state: &ServerState,
    workflow_type: &str,
    doomed: &WorkerDeployment,
    cause: WithdrawCause,
) -> AutoWorkerOutcome {
    let name = doomed.name.as_str();
    let task_queue = doomed.task_queue.as_str();
    let because = cause.sentence(workflow_type, task_queue);
    if let Err(error) = state.worker_supervisor().stop(name).await {
        return AutoWorkerOutcome::new(
            task_queue,
            workflow_type,
            AutoWorkerDecision::Failed,
            Some(name.to_owned()),
            format!(
                "{because}, but worker deployment `{name}` could not be stopped, so it was NOT \
                 withdrawn: {error}"
            ),
        );
    }
    match state
        .worker_deployment_store()
        .delete_worker_deployment(name)
        .await
    {
        Ok(_) => {
            let event_name = name.to_owned();
            drop(
                state
                    .cluster_publisher()
                    .emit(|meta| ClusterEvent::WorkerDeploymentDeleted {
                        meta,
                        name: event_name,
                    }),
            );
            if let Err(error) = state.worker_supervisor().forget(name).await {
                tracing::error!(
                    operation = OPERATION,
                    worker = name,
                    %error,
                    "a withdrawn auto worker deployment could not be dropped out of supervision"
                );
            }
            AutoWorkerOutcome::new(
                task_queue,
                workflow_type,
                AutoWorkerDecision::Retired,
                Some(name.to_owned()),
                format!(
                    "{because}, so worker deployment `{name}` was stopped and withdrawn. Nothing \
                     serves that queue now"
                ),
            )
        }
        Err(error) => AutoWorkerOutcome::new(
            task_queue,
            workflow_type,
            AutoWorkerDecision::Failed,
            Some(name.to_owned()),
            format!(
                "{because}; worker deployment `{name}` was stopped but its record could not be \
                 removed: {error}"
            ),
        ),
    }
}

/// Remove staged document trees no live record's argv names.
///
/// Failures are logged, never propagated: a snapshot that could not be
/// reclaimed costs disk, and turning that into a failed deploy would be a
/// worse trade than the leak. The set of paths to KEEP is read from the durable
/// listing, so a listing that cannot be read keeps everything.
pub(super) async fn prune_snapshots(state: &ServerState, root: &Path) {
    let listing = match state
        .worker_deployment_store()
        .list_worker_deployments()
        .await
    {
        Ok(listing) => listing,
        Err(error) => {
            tracing::warn!(
                operation = OPERATION,
                %error,
                "the worker deployments could not be listed, so superseded staged documents were \
                 kept rather than risk removing one a record still replays"
            );
            return;
        }
    };
    let Some(live) = live_documents(&listing) else {
        for poisoned in &listing.undecodable {
            tracing::warn!(
                operation = OPERATION,
                record = %poisoned.name,
                error = %poisoned.error,
                "a worker deployment record could not be decoded, so superseded staged documents \
                 were kept rather than risk removing the one that record still replays"
            );
        }
        return;
    };
    match documents::prune(root, &live) {
        Ok(removed) if removed.is_empty() => {}
        Ok(removed) => tracing::info!(
            operation = OPERATION,
            count = removed.len(),
            "superseded staged documents reclaimed"
        ),
        Err(error) => tracing::warn!(
            operation = OPERATION,
            %error,
            "superseded staged documents could not be reclaimed"
        ),
    }
}

/// The documents a listing proves are still read — or `None` when the listing
/// cannot prove it for every record.
///
/// # An undecodable record is not an absent record
///
/// The store answers `Ok` for a listing that holds rows it could not decode;
/// it names them in `undecodable` instead of failing (#211). A keep-set built
/// from `deployments` alone would treat each of those as *no record*, and the
/// prune that follows would remove the document a RUNNING worker minted from
/// that row still replays — which is precisely what `prune`'s own contract
/// forbids ("a path this cannot classify is KEPT"). So one poisoned row keeps
/// every snapshot, exactly as a listing that failed outright does; the caller
/// names the rows so an operator can repair them.
pub(super) fn live_documents(listing: &WorkerDeploymentListing) -> Option<BTreeSet<PathBuf>> {
    if !listing.undecodable.is_empty() {
        return None;
    }
    Some(
        listing
            .deployments
            .iter()
            .filter_map(|deployment| record::document_path(&deployment.artifact))
            .map(Path::to_path_buf)
            .collect(),
    )
}

/// Re-derive every `auto/…` record's dial address against THIS boot's outbox,
/// re-minting the ones whose argv no longer matches.
///
/// A record freezes the address it was minted with, and the supervisor replays
/// that argv verbatim. Change `[outbox] liminal_listen_address` and restart,
/// and every auto record dials the old port and crash-loops into `Failed` — a
/// configuration change silently breaking workers nobody edited. The document
/// and the queue are unchanged, so only the connection is rewritten.
pub async fn refresh_dial_addresses(state: &ServerState) -> 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 its \
                 built-in agent workers still dial an address it binds"
            );
            return Vec::new();
        }
    };
    let auto: Vec<&WorkerDeployment> = listing
        .deployments
        .iter()
        .filter(|deployment| record::is_auto_name(&deployment.name))
        .collect();
    if auto.is_empty() {
        return Vec::new();
    }
    let namespace = state.runtime_config().default_namespace.clone();
    let dial = match worker_dial_address(&state.runtime_config().outbox) {
        Ok(dial) => dial,
        Err(dark) => {
            let outcomes: Vec<AutoWorkerOutcome> = auto
                .iter()
                .map(|deployment| {
                    AutoWorkerOutcome::new(
                        deployment.task_queue.clone(),
                        record::staged_workflow_type(&deployment.artifact).unwrap_or_default(),
                        AutoWorkerDecision::DarkOutbox,
                        Some(deployment.name.clone()),
                        format!(
                            "worker deployment `{name}` exists to serve task queue `{queue}`, but \
                             {dark}. It will keep failing to connect until that is fixed",
                            name = deployment.name,
                            queue = deployment.task_queue
                        ),
                    )
                })
                .collect();
            for outcome in &outcomes {
                log_outcome(outcome);
            }
            state.worker_supervisor().record_auto_provision(&outcomes);
            return outcomes;
        }
    };
    let mut outcomes = Vec::new();
    for deployment in auto {
        if let Some(outcome) = refresh_one(state, deployment, &dial, &namespace).await {
            log_outcome(&outcome);
            outcomes.push(outcome);
        }
    }
    state.worker_supervisor().record_auto_provision(&outcomes);
    outcomes
}

/// Re-mint one auto record whose argv no longer matches this boot's dial
/// address. `None` when it already matches — the ordinary case, and silent.
async fn refresh_one(
    state: &ServerState,
    deployment: &WorkerDeployment,
    dial: &str,
    namespace: &str,
) -> Option<AutoWorkerOutcome> {
    let document = record::document_path(&deployment.artifact)?.to_path_buf();
    let wanted = record::verb(&deployment.task_queue, &document, dial, namespace);
    let aion_store::WorkerArtifactRef::Builtin { verb } = &deployment.artifact;
    if verb == &wanted {
        return None;
    }
    let binary = match crate::worker::capture_binary_identity() {
        Ok(binary) => binary,
        Err(error) => {
            return Some(AutoWorkerOutcome::new(
                deployment.task_queue.clone(),
                record::staged_workflow_type(&deployment.artifact).unwrap_or_default(),
                AutoWorkerDecision::Failed,
                Some(deployment.name.clone()),
                format!(
                    "worker deployment `{}` dials an address this boot does not bind, and could \
                     not be rewritten because this executable could not be identified: {error}",
                    deployment.name
                ),
            ));
        }
    };
    let mut requested =
        record::new_deployment(&deployment.task_queue, &document, dial, namespace, binary);
    // Every operator decision on the record survives: only the connection is
    // being corrected, and a refresh that also started a stopped worker would
    // be a configuration change reversing an operator's act.
    requested.desired = deployment.desired;
    let record = WorkerDeployment::new(requested, chrono::Utc::now()).ok()?;
    let put = state
        .worker_deployment_store()
        .put_worker_deployment(record)
        .await;
    let workflow_type = record::staged_workflow_type(&deployment.artifact).unwrap_or_default();
    match put {
        Ok(_) => {
            let converged = converge_and_report(
                state.worker_supervisor(),
                &deployment.name,
                Convergence::Replacing,
                OPERATION,
            )
            .await
            .is_ok();
            Some(AutoWorkerOutcome::new(
                deployment.task_queue.clone(),
                workflow_type,
                if converged {
                    AutoWorkerDecision::Reminted
                } else {
                    AutoWorkerDecision::RecordedNotRunning
                },
                Some(deployment.name.clone()),
                format!(
                    "worker deployment `{name}` dialled an address this boot does not bind, so \
                     its launch was rewritten onto `{dial}`",
                    name = deployment.name
                ),
            ))
        }
        Err(error) => Some(AutoWorkerOutcome::new(
            deployment.task_queue.clone(),
            workflow_type,
            AutoWorkerDecision::Failed,
            Some(deployment.name.clone()),
            format!(
                "worker deployment `{}` dials an address this boot does not bind, and the \
                 corrected record could not be persisted: {error}",
                deployment.name
            ),
        )),
    }
}

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