rhiza_node/

admin.rs

use std::{
    collections::HashMap,
    fmt, fs,
    io::Write,
    path::{Path, PathBuf},
    sync::atomic::{AtomicBool, AtomicUsize, Ordering},
    sync::Arc,
};

use axum::{
    extract::{rejection::JsonRejection, Extension, Request, State},
    http::StatusCode,
    middleware::{self, Next},
    response::{IntoResponse, Response},
    routing::{get, post},
    Json, Router,
};
use rhiza_core::{ConfigChange, ExecutionProfile, LogAnchor, LogEntry, LogHash, StoredCommand};
use rhiza_log::{IndexRange, LogStore};
use rhiza_quepaxa::{Membership, RecorderFileStore};
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::{
    client_authenticated, install_successor_recorder, valid_auth_token, ConfigError, NodeError,
    NodeRuntime, NodeStatus, StopInformation,
};
use crate::{CheckpointCoordinator, DurabilityError};

pub const ADMIN_STATUS_PATH: &str = "/v1/admin/membership/status";
pub const ADMIN_STOP_PATH: &str = "/v1/admin/membership/stop";
pub const ADMIN_INSTALL_SUCCESSOR_PATH: &str = "/v1/admin/membership/install-successor";
pub const ADMIN_ACTIVATE_PATH: &str = "/v1/admin/membership/activate";
pub const ADMIN_COMPACT_PATH: &str = "/v1/admin/checkpoint/compact";

#[derive(Clone, Eq, PartialEq)]
pub struct AdminConfig {
    token: String,
}

impl AdminConfig {
    pub fn new(token: impl Into<String>) -> Result<Self, ConfigError> {
        let token = token.into();
        if !valid_auth_token(&token) {
            return Err(ConfigError::EmptyAdminToken);
        }
        Ok(Self { token })
    }
}

impl fmt::Debug for AdminConfig {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("AdminConfig")
            .field("token", &"[redacted]")
            .finish()
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct AdminStatusResponse {
    pub cluster_id: String,
    pub execution_profile: ExecutionProfile,
    pub epoch: u64,
    pub node: NodeStatus,
    pub members: Vec<String>,
    pub recovery_generation: u64,
    pub qlog_root: LogAnchor,
    pub checkpoint_root: Option<LogAnchor>,
    pub stopped_transition: Option<AdminStoppedTransition>,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AdminStopRequest {
    pub operation_id: String,
    pub expected_config_id: u64,
    pub successor: AdminSuccessorBundle,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct AdminStopResponse {
    pub operation_id: String,
    pub stop: StopInformation,
    pub successor: AdminSuccessorBundle,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct AdminStoppedTransition {
    pub stop: StopInformation,
    pub successor: AdminSuccessorBundle,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AdminSuccessorBundle {
    pub config_id: u64,
    pub members: Vec<String>,
    pub digest: LogHash,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AdminInstallSuccessorRequest {
    pub operation_id: String,
    pub expected_config_id: u64,
    pub expected_stopped_anchor: LogAnchor,
    pub old_members: Vec<String>,
    pub stop: StopInformation,
    pub successor: AdminSuccessorBundle,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct AdminInstallSuccessorResponse {
    pub operation_id: String,
    pub config_id: u64,
    pub digest: LogHash,
    pub activated: bool,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AdminActivateRequest {
    pub operation_id: String,
    pub expected_config_id: u64,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct AdminActivateResponse {
    pub operation_id: String,
    pub entry: LogEntry,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AdminCompactRequest {
    pub operation_id: String,
    pub expected_config_id: u64,
    pub expected_recovery_generation: u64,
    pub expected_root: LogAnchor,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct AdminCompactResponse {
    pub operation_id: String,
    pub anchor: rhiza_core::RecoveryAnchor,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AdminErrorCode {
    Unauthorized,
    InvalidRequest,
    OperationConflict,
    PreconditionFailed,
    Unavailable,
    Internal,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct AdminErrorResponse {
    pub code: AdminErrorCode,
}

#[derive(Clone)]
struct AdminGateState {
    token: String,
    admission: Arc<tokio::sync::Semaphore>,
    tasks: AdminTaskTracker,
}

#[derive(Clone)]
pub struct AdminTaskTracker {
    state: Arc<AdminTaskState>,
}

struct AdminTaskState {
    accepting: AtomicBool,
    active: AtomicUsize,
    changed: tokio::sync::watch::Sender<()>,
}

impl AdminTaskTracker {
    fn new() -> Self {
        let (changed, _) = tokio::sync::watch::channel(());
        Self {
            state: Arc::new(AdminTaskState {
                accepting: AtomicBool::new(true),
                active: AtomicUsize::new(0),
                changed,
            }),
        }
    }

    fn try_start(&self) -> Option<AdminTaskGuard> {
        if !self.state.accepting.load(Ordering::Acquire) {
            return None;
        }
        self.state.active.fetch_add(1, Ordering::AcqRel);
        if self.state.accepting.load(Ordering::Acquire) {
            Some(AdminTaskGuard {
                state: Arc::clone(&self.state),
            })
        } else {
            finish_admin_task(&self.state);
            None
        }
    }

    pub fn stop_admission(&self) {
        self.state.accepting.store(false, Ordering::Release);
    }

    pub async fn wait_for_idle(&self) {
        let mut changed = self.state.changed.subscribe();
        loop {
            if self.state.active.load(Ordering::Acquire) == 0 {
                return;
            }
            if changed.changed().await.is_err() {
                return;
            }
        }
    }
}

struct AdminTaskGuard {
    state: Arc<AdminTaskState>,
}

impl Drop for AdminTaskGuard {
    fn drop(&mut self) {
        finish_admin_task(&self.state);
    }
}

fn finish_admin_task(state: &AdminTaskState) {
    if state.active.fetch_sub(1, Ordering::AcqRel) == 1 {
        state.changed.send_replace(());
    }
}

struct AdminPermit {
    _admission: tokio::sync::OwnedSemaphorePermit,
    _task: AdminTaskGuard,
}

#[derive(Clone)]
struct AdminRouteState {
    runtime: Arc<NodeRuntime>,
    recorder: RecorderFileStore,
    coordinator: Option<Arc<CheckpointCoordinator>>,
    operations: Arc<tokio::sync::Mutex<Option<HashMap<String, OperationRecord>>>>,
    ledger_path: PathBuf,
}

#[derive(Clone, Deserialize, Serialize)]
struct OperationRecord {
    fingerprint: Vec<u8>,
    status: u16,
    body: Value,
}

#[derive(Default, Deserialize, Serialize)]
struct OperationLedger {
    version: u16,
    operations: HashMap<String, OperationRecord>,
}

pub fn node_router_with_admin(
    runtime: Arc<NodeRuntime>,
    recorder: RecorderFileStore,
    admin: AdminConfig,
) -> Result<Router, ConfigError> {
    node_router_with_admin_and_tasks(runtime, recorder, admin).map(|(router, _)| router)
}

pub fn node_router_with_admin_and_tasks(
    runtime: Arc<NodeRuntime>,
    recorder: RecorderFileStore,
    admin: AdminConfig,
) -> Result<(Router, AdminTaskTracker), ConfigError> {
    validate_admin_token(&runtime, &admin)?;
    let (admin_router, tasks) = admin_router(runtime.clone(), recorder.clone(), None, admin);
    Ok((
        crate::node_router(runtime, recorder).merge(admin_router),
        tasks,
    ))
}

pub fn node_router_with_checkpoint_and_admin(
    runtime: Arc<NodeRuntime>,
    recorder: RecorderFileStore,
    coordinator: Arc<CheckpointCoordinator>,
    admin: AdminConfig,
) -> Result<Router, ConfigError> {
    node_router_with_checkpoint_and_admin_tasks(runtime, recorder, coordinator, admin)
        .map(|(router, _)| router)
}

pub fn node_router_with_checkpoint_and_admin_tasks(
    runtime: Arc<NodeRuntime>,
    recorder: RecorderFileStore,
    coordinator: Arc<CheckpointCoordinator>,
    admin: AdminConfig,
) -> Result<(Router, AdminTaskTracker), ConfigError> {
    validate_admin_token(&runtime, &admin)?;
    let (admin_router, tasks) = admin_router(
        runtime.clone(),
        recorder.clone(),
        Some(coordinator.clone()),
        admin,
    );
    Ok((
        crate::node_router_with_checkpoint(runtime, recorder, coordinator).merge(admin_router),
        tasks,
    ))
}

fn validate_admin_token(runtime: &NodeRuntime, admin: &AdminConfig) -> Result<(), ConfigError> {
    if runtime.config().client_token() == admin.token
        || runtime
            .config()
            .peers()
            .iter()
            .any(|peer| peer.token() == admin.token)
    {
        return Err(ConfigError::AdminTokenConflictsWithRuntime);
    }
    Ok(())
}

fn admin_router(
    runtime: Arc<NodeRuntime>,
    recorder: RecorderFileStore,
    coordinator: Option<Arc<CheckpointCoordinator>>,
    admin: AdminConfig,
) -> (Router, AdminTaskTracker) {
    let ledger_path = runtime.config().data_dir().join("admin-operations-v1.json");
    let operations = load_operations(&ledger_path)
        .map(Some)
        .unwrap_or_else(|error| {
            eprintln!("admin operation ledger is unavailable: {error}");
            None
        });
    let state = AdminRouteState {
        runtime,
        recorder,
        coordinator,
        operations: Arc::new(tokio::sync::Mutex::new(operations)),
        ledger_path,
    };
    let tasks = AdminTaskTracker::new();
    let router = Router::new()
        .route(ADMIN_STATUS_PATH, get(handle_status))
        .route(ADMIN_STOP_PATH, post(handle_stop))
        .route(ADMIN_INSTALL_SUCCESSOR_PATH, post(handle_install_successor))
        .route(ADMIN_ACTIVATE_PATH, post(handle_activate))
        .route(ADMIN_COMPACT_PATH, post(handle_compact))
        .route_layer(middleware::from_fn_with_state(
            AdminGateState {
                token: admin.token,
                admission: Arc::new(tokio::sync::Semaphore::new(1)),
                tasks: tasks.clone(),
            },
            admin_gate,
        ))
        .with_state(state);
    (router, tasks)
}

async fn admin_gate(
    State(state): State<AdminGateState>,
    mut request: Request,
    next: Next,
) -> Response {
    if !client_authenticated(request.headers(), &state.token) {
        return admin_error(StatusCode::UNAUTHORIZED, AdminErrorCode::Unauthorized);
    }
    let task = match state.tasks.try_start() {
        Some(task) => task,
        None => return admin_error(StatusCode::SERVICE_UNAVAILABLE, AdminErrorCode::Unavailable),
    };
    let permit = match state.admission.try_acquire_owned() {
        Ok(admission) => Arc::new(AdminPermit {
            _admission: admission,
            _task: task,
        }),
        Err(_) => return admin_error(StatusCode::TOO_MANY_REQUESTS, AdminErrorCode::Unavailable),
    };
    request.extensions_mut().insert(permit);
    next.run(request).await
}

async fn handle_status(
    State(state): State<AdminRouteState>,
    Extension(_permit): Extension<Arc<AdminPermit>>,
) -> Response {
    let _operations = state.operations.lock().await;
    match status_response(&state).await {
        Ok(response) => Json(response).into_response(),
        Err(error) => node_admin_error(error),
    }
}

async fn handle_stop(
    State(state): State<AdminRouteState>,
    Extension(permit): Extension<Arc<AdminPermit>>,
    payload: Result<Json<AdminStopRequest>, JsonRejection>,
) -> Response {
    let request = match payload {
        Ok(Json(request)) => request,
        Err(_) => return admin_error(StatusCode::BAD_REQUEST, AdminErrorCode::InvalidRequest),
    };
    let runtime = state.runtime.clone();
    let owned = request.clone();
    run_async_operation(&state, permit, "stop", &request, async move {
        tokio::task::spawn_blocking(move || {
            let successor = validate_successor(
                &owned.successor,
                owned.expected_config_id,
                runtime.config().cluster_id(),
            )?;
            runtime
                .stop_current_configuration_for_successor(&successor)
                .map(|stop| AdminStopResponse {
                    operation_id: owned.operation_id,
                    stop,
                    successor: owned.successor,
                })
                .map_err(OperationError::Node)
        })
        .await
        .unwrap_or(Err(OperationError::Unavailable))
    })
    .await
}

async fn handle_install_successor(
    State(state): State<AdminRouteState>,
    Extension(permit): Extension<Arc<AdminPermit>>,
    payload: Result<Json<AdminInstallSuccessorRequest>, JsonRejection>,
) -> Response {
    let request = match payload {
        Ok(Json(request)) => request,
        Err(_) => return admin_error(StatusCode::BAD_REQUEST, AdminErrorCode::InvalidRequest),
    };
    let runtime = state.runtime.clone();
    let recorder = state.recorder.clone();
    let owned = request.clone();
    run_async_operation(&state, permit, "install_successor", &request, async move {
        tokio::task::spawn_blocking(move || {
            install_successor(&runtime, &recorder, &owned).map_err(OperationError::Node)
        })
        .await
        .unwrap_or(Err(OperationError::Unavailable))
    })
    .await
}

async fn handle_activate(
    State(state): State<AdminRouteState>,
    Extension(permit): Extension<Arc<AdminPermit>>,
    payload: Result<Json<AdminActivateRequest>, JsonRejection>,
) -> Response {
    let request = match payload {
        Ok(Json(request)) => request,
        Err(_) => return admin_error(StatusCode::BAD_REQUEST, AdminErrorCode::InvalidRequest),
    };
    let runtime = state.runtime.clone();
    let owned = request.clone();
    run_async_operation(&state, permit, "activate", &request, async move {
        tokio::task::spawn_blocking(move || {
            runtime
                .activate_successor_if(owned.expected_config_id)
                .map(|entry| AdminActivateResponse {
                    operation_id: owned.operation_id,
                    entry,
                })
                .map_err(OperationError::Node)
        })
        .await
        .unwrap_or(Err(OperationError::Unavailable))
    })
    .await
}

async fn handle_compact(
    State(state): State<AdminRouteState>,
    Extension(permit): Extension<Arc<AdminPermit>>,
    payload: Result<Json<AdminCompactRequest>, JsonRejection>,
) -> Response {
    let request = match payload {
        Ok(Json(request)) => request,
        Err(_) => return admin_error(StatusCode::BAD_REQUEST, AdminErrorCode::InvalidRequest),
    };
    let runtime = state.runtime.clone();
    let coordinator = state.coordinator.clone();
    let owned = request.clone();
    run_async_operation(&state, permit, "compact", &request, async move {
        match coordinator {
            Some(coordinator) => coordinator
                .checkpoint_compact_fenced(
                    &runtime,
                    owned.expected_config_id,
                    owned.expected_recovery_generation,
                    owned.expected_root,
                )
                .await
                .map(|anchor| AdminCompactResponse {
                    operation_id: owned.operation_id,
                    anchor,
                })
                .map_err(OperationError::Durability),
            None => Err(OperationError::Unavailable),
        }
    })
    .await
}

async fn run_async_operation<T, R, F>(
    state: &AdminRouteState,
    permit: Arc<AdminPermit>,
    kind: &str,
    request: &T,
    operation: F,
) -> Response
where
    T: Serialize,
    R: Serialize + Send + 'static,
    F: std::future::Future<Output = Result<R, OperationError>> + Send + 'static,
{
    let operation_id = match serde_json::to_value(request)
        .ok()
        .and_then(|value| value.get("operation_id")?.as_str().map(str::to_owned))
    {
        Some(operation_id) => operation_id,
        None => return admin_error(StatusCode::BAD_REQUEST, AdminErrorCode::InvalidRequest),
    };
    let fingerprint = match operation_fingerprint(kind, request) {
        Ok(fingerprint) => fingerprint,
        Err(()) => return admin_error(StatusCode::BAD_REQUEST, AdminErrorCode::InvalidRequest),
    };
    {
        let operations = state.operations.lock().await;
        let Some(operations) = operations.as_ref() else {
            return admin_error(StatusCode::SERVICE_UNAVAILABLE, AdminErrorCode::Unavailable);
        };
        if let Some(response) = replay(operations, &operation_id, &fingerprint) {
            return response;
        }
    }
    if let Some(response) = validate_operation_id(&operation_id) {
        return response;
    }
    let detached_state = state.clone();
    let detached_operation_id = operation_id.clone();
    let detached_fingerprint = fingerprint.clone();
    let (completed, mut completion) = tokio::sync::watch::channel(false);
    tokio::spawn(async move {
        let result = operation.await;
        let mut operations = detached_state.operations.lock().await;
        if let Some(records) = operations.as_mut() {
            let _ = store_result(records, detached_operation_id, detached_fingerprint, result);
            if let Err(error) = persist_operations(&detached_state.ledger_path, records) {
                eprintln!("admin operation ledger persistence failed: {error}");
                *operations = None;
            }
        }
        drop(operations);
        drop(detached_state);
        drop(permit);
        completed.send_replace(true);
    });
    let waited = tokio::time::timeout(std::time::Duration::from_secs(10), async {
        while !*completion.borrow() {
            if completion.changed().await.is_err() {
                break;
            }
        }
    })
    .await;
    if waited.is_err() {
        return admin_error(StatusCode::SERVICE_UNAVAILABLE, AdminErrorCode::Unavailable);
    }
    let operations = state.operations.lock().await;
    let Some(operations) = operations.as_ref() else {
        return admin_error(StatusCode::SERVICE_UNAVAILABLE, AdminErrorCode::Unavailable);
    };
    replay(operations, &operation_id, &fingerprint)
        .unwrap_or_else(|| admin_error(StatusCode::INTERNAL_SERVER_ERROR, AdminErrorCode::Internal))
}

fn operation_fingerprint(kind: &str, request: &impl Serialize) -> Result<Vec<u8>, ()> {
    serde_json::to_vec(&(kind, request)).map_err(|_| ())
}

fn replay(
    operations: &HashMap<String, OperationRecord>,
    operation_id: &str,
    fingerprint: &[u8],
) -> Option<Response> {
    operations.get(operation_id).map(|record| {
        if record.fingerprint == fingerprint {
            (
                StatusCode::from_u16(record.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
                Json(record.body.clone()),
            )
                .into_response()
        } else {
            admin_error(StatusCode::CONFLICT, AdminErrorCode::OperationConflict)
        }
    })
}

fn validate_operation_id(operation_id: &str) -> Option<Response> {
    (operation_id.trim().is_empty() || operation_id.len() > 256)
        .then(|| admin_error(StatusCode::BAD_REQUEST, AdminErrorCode::InvalidRequest))
}

fn store_result<R: Serialize>(
    operations: &mut HashMap<String, OperationRecord>,
    operation_id: String,
    fingerprint: Vec<u8>,
    result: Result<R, OperationError>,
) -> Response {
    let (status, body) = match result {
        Ok(response) => match serde_json::to_value(response) {
            Ok(body) => (StatusCode::OK, body),
            Err(_) => (
                StatusCode::INTERNAL_SERVER_ERROR,
                error_value(AdminErrorCode::Internal),
            ),
        },
        Err(error) => operation_error_value(error),
    };
    operations.insert(
        operation_id,
        OperationRecord {
            fingerprint,
            status: status.as_u16(),
            body: body.clone(),
        },
    );
    (status, Json(body)).into_response()
}

async fn status_response(state: &AdminRouteState) -> Result<AdminStatusResponse, NodeError> {
    let checkpoint_root = match state.coordinator.as_ref() {
        Some(coordinator) => {
            let tip = coordinator
                .refresh_durable_tip()
                .await
                .map_err(|error| NodeError::Unavailable(error.to_string()))?;
            Some(LogAnchor::new(tip.index(), tip.hash()))
        }
        None => None,
    };
    let _commit = state.runtime.lock_commit()?;
    let node = state.runtime.status()?;
    let qlog_root = state.runtime.log_root_unlocked()?;
    let stopped_transition = stopped_transition(&state.runtime)?;
    Ok(AdminStatusResponse {
        cluster_id: state.runtime.config.cluster_id().to_owned(),
        execution_profile: state.runtime.config.execution_profile(),
        epoch: state.runtime.config.epoch(),
        node,
        members: state.runtime.config.membership().members().to_vec(),
        recovery_generation: state.runtime.config.recovery_generation(),
        qlog_root,
        checkpoint_root,
        stopped_transition,
    })
}

fn stopped_transition(runtime: &NodeRuntime) -> Result<Option<AdminStoppedTransition>, NodeError> {
    let configuration = runtime.configuration_state()?;
    let Some(anchor) = configuration.stop().copied() else {
        return Ok(None);
    };
    if configuration.config_id() != runtime.consensus.config_id() {
        return Ok(None);
    }
    let entry = runtime.recover_stop_entry(anchor)?;
    let successor = successor_from_entry(&entry)?;
    let proof = runtime
        .consensus
        .inspect_decision_proof_at(entry.index)
        .map_err(|error| NodeError::Unavailable(error.to_string()))?
        .ok_or_else(|| NodeError::Unavailable("durable Stop proof is unavailable".into()))?;
    Ok(Some(AdminStoppedTransition {
        stop: StopInformation {
            version: 2,
            entry,
            proof,
        },
        successor,
    }))
}

fn successor_from_entry(entry: &LogEntry) -> Result<AdminSuccessorBundle, NodeError> {
    let command = StoredCommand::new(entry.entry_type, entry.payload.clone());
    let change = ConfigChange::recognize(&command)
        .map_err(|_| NodeError::PreconditionFailed("legacy unbound Stop is rejected".into()))?;
    let successor = change
        .successor()
        .ok_or_else(|| NodeError::PreconditionFailed("legacy unbound Stop is rejected".into()))?;
    Ok(AdminSuccessorBundle {
        config_id: successor.config_id(),
        members: successor.members().to_vec(),
        digest: successor.digest(),
    })
}

fn validate_successor(
    bundle: &AdminSuccessorBundle,
    predecessor_config_id: u64,
    cluster_id: &str,
) -> Result<Membership, OperationError> {
    let membership = Membership::from_voters(bundle.members.clone()).map_err(|_| {
        OperationError::Node(NodeError::InvalidRequest(
            "successor membership is invalid".into(),
        ))
    })?;
    if predecessor_config_id.checked_add(1) != Some(bundle.config_id)
        || membership.digest() != bundle.digest
        || cluster_id.is_empty()
    {
        return Err(OperationError::Node(NodeError::PreconditionFailed(
            "successor descriptor does not match the active configuration".into(),
        )));
    }
    Ok(membership)
}

fn load_operations(path: &Path) -> Result<HashMap<String, OperationRecord>, std::io::Error> {
    let bytes = match fs::read(path) {
        Ok(bytes) => bytes,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(HashMap::new()),
        Err(error) => return Err(error),
    };
    let ledger: OperationLedger = serde_json::from_slice(&bytes)
        .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
    if ledger.version != 1 {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "unsupported admin operation ledger version",
        ));
    }
    Ok(ledger.operations)
}

fn persist_operations(
    path: &Path,
    operations: &HashMap<String, OperationRecord>,
) -> Result<(), std::io::Error> {
    let parent = path.parent().ok_or_else(|| {
        std::io::Error::new(std::io::ErrorKind::InvalidInput, "ledger has no parent")
    })?;
    fs::create_dir_all(parent)?;
    let temporary = path.with_extension(format!("tmp-{}", std::process::id()));
    let bytes = serde_json::to_vec(&OperationLedger {
        version: 1,
        operations: operations.clone(),
    })
    .map_err(std::io::Error::other)?;
    let mut file = fs::File::create(&temporary)?;
    file.write_all(&bytes)?;
    file.sync_all()?;
    fs::rename(&temporary, path)?;
    fs::File::open(parent)?.sync_all()
}

fn install_successor(
    runtime: &NodeRuntime,
    recorder: &RecorderFileStore,
    request: &AdminInstallSuccessorRequest,
) -> Result<AdminInstallSuccessorResponse, NodeError> {
    let _commit = runtime.lock_commit()?;
    runtime.ensure_ready()?;
    let state = runtime.configuration_state()?;
    if state.is_active()
        || state.config_id() != request.expected_config_id
        || state.stop().copied() != Some(request.expected_stopped_anchor)
    {
        return Err(NodeError::PreconditionFailed(
            "stopped configuration anchor does not match".into(),
        ));
    }
    let old_membership = Membership::from_voters(request.old_members.clone())
        .map_err(|_| NodeError::InvalidRequest("old membership is invalid".into()))?;
    if old_membership.digest() != state.digest()
        || request.stop.entry.cluster_id != runtime.config.cluster_id()
        || request.stop.entry.epoch != runtime.config.epoch()
        || request.stop.entry.config_id != request.expected_config_id
        || request.stop.entry.index != request.expected_stopped_anchor.index()
        || request.stop.entry.hash != request.expected_stopped_anchor.hash()
    {
        return Err(NodeError::PreconditionFailed(
            "old decision material does not match the stopped runtime".into(),
        ));
    }
    let entries = runtime
        .log_store
        .read_range(
            IndexRange::new(request.stop.entry.index, request.stop.entry.index)
                .map_err(|error| NodeError::Storage(error.to_string()))?,
        )
        .map_err(|error| NodeError::Storage(error.to_string()))?;
    if entries.as_slice() != [request.stop.entry.clone()] {
        return Err(NodeError::PreconditionFailed(
            "old stop entry is not the exact local qlog entry".into(),
        ));
    }
    let successor = Membership::from_voters(request.successor.members.clone())
        .map_err(|_| NodeError::InvalidRequest("successor membership is invalid".into()))?;
    if successor.digest() != request.successor.digest
        || request.expected_config_id.checked_add(1) != Some(request.successor.config_id)
        || successor_from_entry(&request.stop.entry)? != request.successor
    {
        return Err(NodeError::PreconditionFailed(
            "successor bundle or digest does not match".into(),
        ));
    }
    let installed = install_successor_recorder(
        recorder,
        request.successor.config_id,
        successor,
        &request.stop,
    )?;
    Ok(AdminInstallSuccessorResponse {
        operation_id: request.operation_id.clone(),
        config_id: installed.config_id(),
        digest: installed.config_digest(),
        activated: installed.is_activated(),
    })
}

enum OperationError {
    Node(NodeError),
    Durability(DurabilityError),
    Unavailable,
}

fn operation_error_value(error: OperationError) -> (StatusCode, Value) {
    match error {
        OperationError::Node(error) => {
            eprintln!("admin operation failed: {error}");
            let (status, code) = node_admin_status(&error);
            (status, error_value(code))
        }
        OperationError::Durability(DurabilityError::PreconditionFailed) => (
            StatusCode::CONFLICT,
            error_value(AdminErrorCode::PreconditionFailed),
        ),
        OperationError::Durability(error) => {
            eprintln!("admin durability operation failed: {error}");
            (
                StatusCode::SERVICE_UNAVAILABLE,
                error_value(AdminErrorCode::Unavailable),
            )
        }
        OperationError::Unavailable => (
            StatusCode::SERVICE_UNAVAILABLE,
            error_value(AdminErrorCode::Unavailable),
        ),
    }
}

fn node_admin_error(error: NodeError) -> Response {
    let (status, code) = node_admin_status(&error);
    admin_error(status, code)
}

fn node_admin_status(error: &NodeError) -> (StatusCode, AdminErrorCode) {
    match error {
        NodeError::InvalidRequest(_) => (StatusCode::BAD_REQUEST, AdminErrorCode::InvalidRequest),
        #[cfg(feature = "sql")]
        NodeError::InvalidSqlStatement { .. } => {
            (StatusCode::BAD_REQUEST, AdminErrorCode::InvalidRequest)
        }
        NodeError::PreconditionFailed(_) | NodeError::ConfigurationTransition { .. } => {
            (StatusCode::CONFLICT, AdminErrorCode::PreconditionFailed)
        }
        #[cfg(feature = "sql")]
        NodeError::RequestConflict(_) => (StatusCode::CONFLICT, AdminErrorCode::PreconditionFailed),
        NodeError::Unavailable(_)
        | NodeError::ResourceExhausted(_)
        | NodeError::Contention(_)
        | NodeError::WinnerLimitExceeded => {
            (StatusCode::SERVICE_UNAVAILABLE, AdminErrorCode::Unavailable)
        }
        NodeError::UnsupportedAckMode(_)
        | NodeError::ExecutionProfileMismatch { .. }
        | NodeError::DataRootLocked(_)
        | NodeError::SnapshotRequired(_)
        | NodeError::Storage(_)
        | NodeError::Reconciliation(_)
        | NodeError::Invariant(_)
        | NodeError::Fatal(_) => (StatusCode::INTERNAL_SERVER_ERROR, AdminErrorCode::Internal),
    }
}

fn admin_error(status: StatusCode, code: AdminErrorCode) -> Response {
    (status, Json(AdminErrorResponse { code })).into_response()
}

fn error_value(code: AdminErrorCode) -> Value {
    serde_json::to_value(AdminErrorResponse { code })
        .unwrap_or_else(|_| serde_json::json!({"code": "internal"}))
}

#[cfg(test)]
mod tests {
    use std::sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    };

    use super::AdminTaskTracker;

    #[tokio::test]
    async fn shutdown_observes_a_late_admin_mutation_before_sampling_state() {
        let tasks = AdminTaskTracker::new();
        let guard = tasks.try_start().unwrap();
        let committed_tip = Arc::new(AtomicUsize::new(1));
        let late_tip = Arc::clone(&committed_tip);
        let (release, wait) = tokio::sync::oneshot::channel();
        let operation = tokio::spawn(async move {
            let _ = wait.await;
            late_tip.store(2, Ordering::Release);
            drop(guard);
        });

        tasks.stop_admission();
        assert!(tasks.try_start().is_none());
        let mut idle = Box::pin(tasks.wait_for_idle());
        assert!(
            tokio::time::timeout(std::time::Duration::from_millis(10), &mut idle)
                .await
                .is_err()
        );

        let _ = release.send(());
        idle.await;
        operation.await.unwrap();
        assert_eq!(committed_tip.load(Ordering::Acquire), 2);
    }

    #[tokio::test]
    async fn shutdown_waits_until_every_admitted_admin_task_finishes() {
        let tasks = AdminTaskTracker::new();
        let first = tasks.try_start().unwrap();
        let second = tasks.try_start().unwrap();
        tasks.stop_admission();

        let mut idle = Box::pin(tasks.wait_for_idle());
        drop(first);
        assert!(
            tokio::time::timeout(std::time::Duration::from_millis(10), &mut idle)
                .await
                .is_err()
        );

        drop(second);
        tokio::time::timeout(std::time::Duration::from_secs(1), idle)
            .await
            .expect("last task completion must wake the shutdown waiter");
    }
}