hypercall_admin/

lifecycle.rs

//! Standby promotion + drain lifecycle admin endpoints.
//!
//! These route handlers were carved out of `handlers::health`. The drain
//! marker / engine-quiesce helpers are kept local here so the admin surface
//! depends only on stable lower crates.

use axum::{extract::State, http::StatusCode, response::IntoResponse};
use serde_json::json;

use crate::state::AdminState;
use hypercall_runtime_api::drain::{
    drain_marker_path, keep_drain_after_accepted_quiesce_failure, persist_drain_marker,
};
use hypercall_runtime_api::sonic_json::SonicJson;
use hypercall_runtime_api::StandbyPromoteOutcome;

async fn send_engine_quiesce_action(
    state: &AdminState,
    action: hypercall_runtime_api::EngineQuiesceAction,
) -> Result<hypercall_runtime_api::EngineQuiesceReport, &'static str> {
    let (tx, rx) = tokio::sync::oneshot::channel();
    state
        .engine_quiesce_sender
        .send(hypercall_runtime_api::EngineQuiesceRequest {
            action,
            response_tx: tx,
        })
        .await
        .map_err(|_| "engine quiesce channel is closed")?;
    tokio::time::timeout(std::time::Duration::from_secs(10), rx)
        .await
        .map_err(|_| "engine quiesce timed out")?
        .map_err(|_| "engine quiesce response channel closed")
}

async fn resume_after_failed_drain_marker(
    state: &AdminState,
) -> Result<hypercall_runtime_api::EngineQuiesceReport, &'static str> {
    let report =
        send_engine_quiesce_action(state, hypercall_runtime_api::EngineQuiesceAction::Resume)
            .await?;
    state
        .is_draining
        .store(false, std::sync::atomic::Ordering::SeqCst);
    Ok(report)
}

/// Promote standby to active — stops the replay loop and starts the engine.
pub async fn admin_promote(State(app_state): State<AdminState>) -> impl IntoResponse {
    let promote = match &app_state.standby_promote {
        None => {
            return (
                StatusCode::NOT_FOUND,
                SonicJson(json!({"status": "not_standby_mode"})),
            )
        }
        Some(p) => p,
    };

    if let Some(controller) = &app_state.standby_controller {
        match controller.promote().await {
            StandbyPromoteOutcome::Promoted { queued_orders } => {
                tracing::info!(queued_orders, "Standby controller promoted");
            }
            StandbyPromoteOutcome::AlreadyActive => {
                tracing::info!("Standby controller was already promoted");
            }
        }
    }

    let mut guard = promote.lock().await;
    match guard.take() {
        Some(tx) => {
            tx.send(()).ok();
            (
                StatusCode::OK,
                SonicJson(json!({
                    "status": "promoted",
                    "message": "Engine promoted to active mode",
                })),
            )
        }
        None => (
            StatusCode::CONFLICT,
            SonicJson(json!({"status": "already_promoted"})),
        ),
    }
}

pub async fn admin_drain(State(app_state): State<AdminState>) -> impl IntoResponse {
    app_state
        .is_draining
        .store(true, std::sync::atomic::Ordering::SeqCst);
    app_state.drain_signal.notify_waiters();

    let marker_path = drain_marker_path(&app_state.runtime_config.wal_path);

    // Wait for in-flight orders to finish processing.
    // The order_sender channel capacity tells us whether there are pending requests:
    // when capacity == max_capacity, no sends are buffered and the engine has consumed
    // everything queued before the drain flag was set.
    // Poll up to 5 seconds (50 * 100ms) for the channel to empty.
    let max_cap = app_state.order_sender.max_capacity();
    for _ in 0..50 {
        if app_state.order_sender.capacity() == max_cap {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }

    let (tx, rx) = tokio::sync::oneshot::channel();
    if app_state
        .engine_quiesce_sender
        .send(hypercall_runtime_api::EngineQuiesceRequest {
            action: hypercall_runtime_api::EngineQuiesceAction::Quiesce,
            response_tx: tx,
        })
        .await
        .is_err()
    {
        app_state
            .is_draining
            .store(false, std::sync::atomic::Ordering::SeqCst);
        return (
            StatusCode::SERVICE_UNAVAILABLE,
            SonicJson(json!({
                "status": "error",
                "message": "engine quiesce channel is closed",
            })),
        )
            .into_response();
    }

    let report = match tokio::time::timeout(std::time::Duration::from_secs(10), rx).await {
        Ok(Ok(report)) => report,
        Ok(Err(_)) => {
            return keep_drain_after_accepted_quiesce_failure(
                &app_state.is_draining,
                &marker_path,
                StatusCode::SERVICE_UNAVAILABLE,
                "engine quiesce response channel closed after request was accepted",
            )
            .await;
        }
        Err(_) => {
            return keep_drain_after_accepted_quiesce_failure(
                &app_state.is_draining,
                &marker_path,
                StatusCode::GATEWAY_TIMEOUT,
                "engine quiesce timed out after request was accepted",
            )
            .await;
        }
    };

    if let Some(parent) = marker_path.parent() {
        if let Err(error) = tokio::fs::create_dir_all(parent).await {
            return match resume_after_failed_drain_marker(&app_state).await {
                Ok(resume_report) => (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    SonicJson(json!({
                        "status": "error",
                        "message": format!("failed to create drain marker directory: {error}"),
                        "rollback": "engine resumed and drain flag cleared",
                        "engine": resume_report,
                    })),
                )
                    .into_response(),
                Err(resume_error) => (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    SonicJson(json!({
                        "status": "error",
                        "message": format!("failed to create drain marker directory: {error}"),
                        "rollback_error": resume_error,
                        "draining": true,
                    })),
                )
                    .into_response(),
            };
        }
    }
    if let Err(error) = tokio::fs::write(&marker_path, b"drained\n").await {
        return match resume_after_failed_drain_marker(&app_state).await {
            Ok(resume_report) => (
                StatusCode::INTERNAL_SERVER_ERROR,
                SonicJson(json!({
                    "status": "error",
                    "message": format!("failed to persist drain marker: {error}"),
                    "rollback": "engine resumed and drain flag cleared",
                    "engine": resume_report,
                })),
            )
                .into_response(),
            Err(resume_error) => (
                StatusCode::INTERNAL_SERVER_ERROR,
                SonicJson(json!({
                    "status": "error",
                    "message": format!("failed to persist drain marker: {error}"),
                    "rollback_error": resume_error,
                    "draining": true,
                })),
            )
                .into_response(),
        };
    }

    (
        StatusCode::OK,
        SonicJson(json!({
            "status": "drained",
            "message": "Server drained. No new orders will be accepted. Safe to switch traffic.",
            "engine": report,
            "drain_marker": marker_path,
        })),
    )
        .into_response()
}

pub async fn admin_undrain(State(app_state): State<AdminState>) -> impl IntoResponse {
    let marker_path = drain_marker_path(&app_state.runtime_config.wal_path);
    let mut removed_marker = false;
    if let Err(error) = tokio::fs::remove_file(&marker_path).await {
        if error.kind() != std::io::ErrorKind::NotFound {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                SonicJson(json!({
                    "status": "error",
                    "message": format!("failed to remove drain marker: {error}"),
                })),
            )
                .into_response();
        }
    } else {
        removed_marker = true;
    }

    let (tx, rx) = tokio::sync::oneshot::channel();
    if app_state
        .engine_quiesce_sender
        .send(hypercall_runtime_api::EngineQuiesceRequest {
            action: hypercall_runtime_api::EngineQuiesceAction::Resume,
            response_tx: tx,
        })
        .await
        .is_err()
    {
        let marker_restore_error = if removed_marker {
            persist_drain_marker(&marker_path).await.err()
        } else {
            None
        };
        return (
            StatusCode::SERVICE_UNAVAILABLE,
            SonicJson(json!({
                "status": "error",
                "message": "engine quiesce channel is closed",
                "draining": true,
                "marker_restore_error": marker_restore_error,
            })),
        )
            .into_response();
    }

    let report = match tokio::time::timeout(std::time::Duration::from_secs(10), rx).await {
        Ok(Ok(report)) => report,
        Ok(Err(_)) => {
            let marker_restore_error = if removed_marker {
                persist_drain_marker(&marker_path).await.err()
            } else {
                None
            };
            return (
                StatusCode::SERVICE_UNAVAILABLE,
                SonicJson(json!({
                    "status": "error",
                    "message": "engine resume response channel closed",
                    "draining": true,
                    "marker_restore_error": marker_restore_error,
                })),
            )
                .into_response();
        }
        Err(_) => {
            let marker_restore_error = if removed_marker {
                persist_drain_marker(&marker_path).await.err()
            } else {
                None
            };
            return (
                StatusCode::GATEWAY_TIMEOUT,
                SonicJson(json!({
                    "status": "error",
                    "message": "engine resume timed out",
                    "draining": true,
                    "marker_restore_error": marker_restore_error,
                })),
            )
                .into_response();
        }
    };

    app_state
        .is_draining
        .store(false, std::sync::atomic::Ordering::SeqCst);

    (
        StatusCode::OK,
        SonicJson(json!({
            "status": "undrained",
            "message": "Server accepting orders again.",
            "engine": report,
        })),
    )
        .into_response()
}

pub async fn drain_status(State(app_state): State<AdminState>) -> impl IntoResponse {
    let draining = app_state
        .is_draining
        .load(std::sync::atomic::Ordering::Relaxed);
    (
        StatusCode::OK,
        SonicJson(json!({
            "draining": draining,
        })),
    )
}