aion-server 0.27.1

//! Deploy-granted HTTP management for durable worker deployments.
//!
//! # A durable desired state is an instruction, not a note
//!
//! Every mutation here writes the durable record AND converges this node's own
//! supervisor before it answers. Writing the record alone left a fresh
//! `desired = running` deployment STOPPED until the next boot reconcile or an
//! explicit `aion worker start` — observed live on 2026-08-26, three records
//! minted at 03:44Z that each needed starting by hand. The record is the
//! operator's instruction; the node that accepted it acts on it now.
//!
//! Convergence never re-writes desired state and never re-publishes a
//! desired-state event: the write above already happened once and already
//! published once. [`crate::worker::supervisor::WorkerSupervisor::converge`]
//! exists for exactly that reason — [`WorkerSupervisor::start`]/`stop` are for
//! a caller asking the intent to CHANGE, which is not what these handlers are.
//!
//! [`WorkerSupervisor::start`]: crate::worker::supervisor::WorkerSupervisor::start

use std::collections::BTreeSet;

use aion_core::{ClusterEvent, PutOutcome};
use aion_proto::WireError;
use aion_store::{
    DesiredState, NewWorkerDeployment, WorkerArtifactRef, WorkerDeployment,
    WorkerDeploymentDeleteOutcome, WorkerDeploymentListing,
};
use axum::{
    Json,
    extract::{Path, State},
};
use serde::{Deserialize, Serialize};

use super::auth::HttpCaller;
use super::error::HttpWireError;
use crate::worker::supervisor::{Convergence, converge_and_report};
use crate::worker::{BinaryIdentityCaptureError, capture_binary_identity};
use crate::{ServerError, ServerState};

/// Operator-authored fields for a builtin worker deployment.
#[derive(Debug, Deserialize)]
pub(crate) struct PutWorkerDeploymentRequest {
    artifact: WorkerArtifactRef,
    namespaces: BTreeSet<String>,
    task_queue: String,
    node: Option<String>,
    desired: DesiredState,
}

/// Desired-state mutation body.
#[derive(Debug, Deserialize)]
pub(crate) struct SetDesiredStateRequest {
    desired: DesiredState,
}

/// Create-or-replace response carrying both the outcome and durable record.
#[derive(Debug, Deserialize, Serialize)]
pub(crate) struct PutWorkerDeploymentResponse {
    outcome: PutOutcome,
    deployment: WorkerDeployment,
}

pub(crate) async fn put_worker_deployment(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path(name): Path<String>,
    Json(request): Json<PutWorkerDeploymentRequest>,
) -> Result<Json<PutWorkerDeploymentResponse>, HttpWireError> {
    require_deploy_grant(&caller)?;
    reserve_auto_prefix(&name)?;
    let binary = capture_binary_identity().map_err(|error| capture_refusal(&error))?;
    let requested = WorkerDeployment::new(
        NewWorkerDeployment {
            name,
            artifact: request.artifact,
            binary,
            namespaces: request.namespaces,
            task_queue: request.task_queue,
            node: request.node,
            desired: request.desired,
        },
        chrono::Utc::now(),
    )
    .map_err(|error| HttpWireError(WireError::invalid_input(error.to_string())))?;
    let result = state
        .worker_deployment_store()
        .put_worker_deployment(requested)
        .await
        .map_err(store_error)?;
    let outcome = result.outcome;
    let record = result.deployment;
    let event_name = record.name.clone();
    let desired_state = record.desired;
    let binary_version = record.binary.version.clone();
    let binary_content_hash = record.binary.content_hash.clone();
    drop(
        state
            .cluster_publisher()
            .emit(|meta| ClusterEvent::WorkerDeploymentPut {
                meta,
                name: event_name,
                outcome,
                desired_state,
                binary_version,
                binary_content_hash,
            }),
    );
    // A REPLACE rewrote the argv a live instance is replaying verbatim, so
    // what is running is serving a launch this record no longer names. A
    // CREATE has nothing live to replace.
    let mode = match outcome {
        PutOutcome::Created => Convergence::Idempotent,
        PutOutcome::Replaced => Convergence::Replacing,
    };
    drop(
        converge_and_report(
            state.worker_supervisor(),
            &record.name,
            mode,
            "worker-deployment put",
        )
        .await,
    );
    Ok(Json(PutWorkerDeploymentResponse {
        outcome,
        deployment: record,
    }))
}

pub(crate) async fn list_worker_deployments(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
) -> Result<Json<WorkerDeploymentListing>, HttpWireError> {
    require_deploy_grant(&caller)?;
    state
        .worker_deployment_store()
        .list_worker_deployments()
        .await
        .map(Json)
        .map_err(store_error)
}

pub(crate) async fn get_worker_deployment(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path(name): Path<String>,
) -> Result<Json<WorkerDeployment>, HttpWireError> {
    require_deploy_grant(&caller)?;
    state
        .worker_deployment_store()
        .get_worker_deployment(&name)
        .await
        .map_err(store_error)?
        .map(Json)
        .ok_or_else(|| not_found(&name))
}

pub(crate) async fn set_worker_deployment_desired_state(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path(name): Path<String>,
    Json(request): Json<SetDesiredStateRequest>,
) -> Result<Json<WorkerDeployment>, HttpWireError> {
    require_deploy_grant(&caller)?;
    let record = state
        .worker_deployment_store()
        .set_desired_state(&name, request.desired)
        .await
        .map_err(store_error)?
        .ok_or_else(|| not_found(&name))?;
    let event_name = record.name.clone();
    let desired_state = record.desired;
    drop(state.cluster_publisher().emit(|meta| {
        ClusterEvent::WorkerDeploymentDesiredStateChanged {
            meta,
            name: event_name,
            desired_state,
        }
    }));
    // The argv is untouched by a desired-state flip, so a live instance is
    // still serving the launch this record names: converging idempotently
    // starts a stopped one and stops a running one without churning either.
    drop(
        converge_and_report(
            state.worker_supervisor(),
            &record.name,
            Convergence::Idempotent,
            "worker-deployment desired-state",
        )
        .await,
    );
    Ok(Json(record))
}

pub(crate) async fn delete_worker_deployment(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path(name): Path<String>,
) -> Result<Json<WorkerDeploymentDeleteOutcome>, HttpWireError> {
    require_deploy_grant(&caller)?;
    let outcome = state
        .worker_deployment_store()
        .delete_worker_deployment(&name)
        .await
        .map_err(store_error)?;
    if !outcome.existed {
        return Err(not_found(&name));
    }
    // A supervised process whose record is gone is an orphan: `report()` joins
    // the durable listing, so it would disappear from every surface while
    // still holding its process group. Drop it out of supervision, proving it
    // stopped, before the delete is answered.
    if let Err(error) = state.worker_supervisor().forget(&name).await {
        tracing::error!(
            operation = "worker-deployment delete",
            worker = name.as_str(),
            %error,
            "the worker deployment record was deleted but its supervised process could not be \
             proven stopped; find it with the process group in `GET /workers/managed` before the \
             delete removed the row"
        );
    }
    let event_name = name;
    drop(
        state
            .cluster_publisher()
            .emit(|meta| ClusterEvent::WorkerDeploymentDeleted {
                meta,
                name: event_name,
            }),
    );
    Ok(Json(outcome))
}

/// Refuse an operator-authored record inside the server's own name space.
///
/// `auto/…` is where auto-provision keeps its records, and it re-mints them
/// whenever a document declaring that queue is deployed. A record an operator
/// wrote there would be replaced and its worker restarted with no warning, and
/// the auto-provision log would report it as a routine `reminted` — an
/// operator's deliberate configuration erased by a mechanism whose whole rule
/// is that an operator's record wins. The reservation is enforced at the
/// boundary, where it can still be refused, rather than only documented.
fn reserve_auto_prefix(name: &str) -> Result<(), HttpWireError> {
    if crate::worker::auto_provision::is_auto_name(name) {
        return Err(HttpWireError(WireError::invalid_input(format!(
            "worker deployment name `{name}` is refused: the `{prefix}` prefix is reserved for \
             the records this server mints itself for documents that declare a `harness` \
             section, and a deploy of such a document would replace anything written there. \
             Choose a name without that prefix — an operator-authored record serving the same \
             task queue already takes precedence over auto-provision.",
            prefix = crate::worker::auto_provision::AUTO_PREFIX
        ))));
    }
    Ok(())
}

fn require_deploy_grant(caller: &crate::CallerIdentity) -> Result<(), HttpWireError> {
    if caller.deploy_granted() {
        Ok(())
    } else {
        Err(HttpWireError(
            ServerError::namespace_denied(
                "worker-deployment management requires the deployment-wide deploy grant",
            )
            .to_wire_error(),
        ))
    }
}

fn capture_refusal(error: &BinaryIdentityCaptureError) -> HttpWireError {
    HttpWireError(WireError::backend(format!(
        "worker deployment refused because binary identity capture failed: {error}"
    )))
}

fn store_error(error: aion_store::StoreError) -> HttpWireError {
    HttpWireError(ServerError::from(error).to_wire_error())
}

fn not_found(name: &str) -> HttpWireError {
    HttpWireError(WireError::not_found(format!(
        "worker deployment `{name}` was not found"
    )))
}

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