aion-server 0.27.1

//! 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).await);
    }
    outcomes
}

/// Stop and delete one stale auto record, proving its process gone.
async fn withdraw(
    state: &ServerState,
    workflow_type: &str,
    doomed: &WorkerDeployment,
) -> AutoWorkerOutcome {
    let name = doomed.name.as_str();
    let task_queue = doomed.task_queue.as_str();
    // Stop FIRST, and only delete once the process group is proven empty: a
    // delete that outran the stop would leave a process nothing can report on,
    // which is the orphan the DELETE transport already refuses to create.
    if let Err(error) = state.worker_supervisor().stop(name).await {
        return AutoWorkerOutcome::new(
            task_queue,
            workflow_type,
            AutoWorkerDecision::Failed,
            Some(name.to_owned()),
            format!(
                "`{workflow_type}` no longer declares a `harness` section for task queue \
                 `{task_queue}`, 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!(
                    "`{workflow_type}` no longer declares a `harness` section for task queue \
                     `{task_queue}`, 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!(
                "`{workflow_type}` no longer declares a `harness` section for task queue \
                 `{task_queue}`; 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 live: BTreeSet<PathBuf> = listing
        .deployments
        .iter()
        .filter_map(|deployment| record::document_path(&deployment.artifact))
        .map(Path::to_path_buf)
        .collect();
    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"
        ),
    }
}

/// 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
            ),
        )),
    }
}