synd_api/serve/

mod.rs

use std::{path::Path, time::Duration};

use axum::{
    BoxError, Extension, Router,
    error_handling::HandleErrorLayer,
    http::{StatusCode, header::AUTHORIZATION},
    response::IntoResponse,
    routing::{get, post},
};
use tokio::net::TcpListener;
#[cfg(unix)]
use tokio::net::UnixListener;
use tokio_metrics::TaskMonitor;
use tower::{ServiceBuilder, limit::ConcurrencyLimitLayer, timeout::TimeoutLayer};
use tower_http::{
    cors::CorsLayer, limit::RequestBodyLimitLayer, sensitive_headers::SetSensitiveHeadersLayer,
};
use tracing::{debug, info};

use crate::{
    Result, config,
    dependency::Dependency,
    gql::{self, SyndSchema},
    serve::layer::{authenticate, request_metrics::RequestMetricsLayer, trace},
    session::{DaemonSessionConfig, DaemonSessionSweeper, SessionIdleShutdown},
    shutdown::Shutdown,
};

pub mod auth;
mod probe;
mod session;

pub mod layer;

pub async fn rustls_config_from_pem_files(
    certificate: impl AsRef<Path>,
    private_key: impl AsRef<Path>,
) -> Result<axum_server::tls_rustls::RustlsConfig> {
    axum_server::tls_rustls::RustlsConfig::from_pem_file(certificate, private_key)
        .await
        .map_err(|source| crate::Error::TlsConfig { source })
}

pub struct ServeOptions {
    pub timeout: Duration,
    pub body_limit_bytes: usize,
    pub concurrency_limit: usize,
    pub daemon_sessions: DaemonSessionConfig,
}

impl ServeOptions {
    #[must_use]
    pub fn with_daemon_sessions(mut self, daemon_sessions: DaemonSessionConfig) -> Self {
        self.daemon_sessions = daemon_sessions;
        self
    }
}

impl Default for ServeOptions {
    fn default() -> Self {
        Self {
            timeout: Duration::from_secs(30),
            body_limit_bytes: config::serve::DEFAULT_REQUEST_BODY_LIMIT_BYTES,
            concurrency_limit: config::serve::DEFAULT_REQUEST_CONCURRENCY_LIMIT,
            daemon_sessions: DaemonSessionConfig::default(),
        }
    }
}

#[derive(Clone)]
pub(crate) struct Context {
    pub gql_monitor: TaskMonitor,
    pub schema: SyndSchema,
}

/// Start api server
pub async fn serve(listener: TcpListener, dep: Dependency, shutdown: Shutdown) -> Result<()> {
    let ApiService { router, tls_config } = build_service(dep, &shutdown);

    info!("Serving...");

    let listener = listener.into_std()?;
    let handle = shutdown.into_handle();

    if let Some(tls_config) = tls_config {
        axum_server::from_tcp_rustls(listener, tls_config)?
            .handle(handle)
            .serve(router.into_make_service())
            .await?;
    } else {
        axum_server::from_tcp(listener)?
            .handle(handle)
            .serve(router.into_make_service())
            .await?;
    }

    info!("Shutdown complete");

    Ok(())
}

#[cfg(unix)]
pub async fn serve_unix(listener: UnixListener, dep: Dependency, shutdown: Shutdown) -> Result<()> {
    let daemon_sessions = dep.serve_options.daemon_sessions;
    let sessions = dep
        .sessions
        .with_lease_policy(daemon_sessions.lease_policy())
        .with_idle_shutdown(SessionIdleShutdown::new(
            daemon_sessions.idle_shutdown_grace(),
            shutdown.clone(),
        ));
    let ApiService { router, .. } = build_service(dep, &shutdown);
    let shutdown_requested = shutdown.cancellation_token();
    let _session_sweeper =
        DaemonSessionSweeper::new(sessions.clone(), shutdown_requested.clone()).spawn();
    let router = router
        .route("/session/open", post(session::open))
        .route("/session/renew", post(session::renew))
        .route("/session/close", post(session::close))
        .route(synd_protocol::daemon::STATUS_PATH, get(control::status))
        .route("/daemon/shutdown", post(control::shutdown))
        .layer(Extension(sessions))
        .layer(Extension(shutdown));

    debug!("Serving on Unix socket");

    axum::serve(listener, router)
        .with_graceful_shutdown(async move {
            shutdown_requested.cancelled().await;
        })
        .await?;

    debug!("Unix socket server shutdown complete");

    Ok(())
}

struct ApiService {
    router: Router,
    tls_config: Option<axum_server::tls_rustls::RustlsConfig>,
}

fn build_service(dep: Dependency, shutdown: &Shutdown) -> ApiService {
    let Dependency {
        authenticator,
        registry,
        tls_config,
        serve_options:
            ServeOptions {
                timeout: request_timeout,
                body_limit_bytes: request_body_limit_bytes,
                concurrency_limit,
                daemon_sessions: _,
            },
        monitors,
        sessions: _,
    } = dep;

    let cx = Context {
        gql_monitor: monitors.graphql_task_monitor(),
        schema: gql::schema_builder().data(registry).finish(),
    };

    tokio::spawn(monitors.emit_metrics(
        config::metrics::MONITOR_INTERVAL,
        shutdown.cancellation_token(),
    ));

    let router = Router::new()
        .route("/graphql", post(gql::handler::graphql))
        .route("/graphql/ws", get(gql::handler::graphql_ws))
        .layer(Extension(cx))
        .layer(authenticate::AuthenticateLayer::new(authenticator))
        .route("/graphql", get(gql::handler::graphiql))
        .layer(
            ServiceBuilder::new()
                .layer(SetSensitiveHeadersLayer::new(std::iter::once(
                    AUTHORIZATION,
                )))
                .layer(trace::layer())
                .layer(HandleErrorLayer::new(handle_middleware_error))
                .layer(TimeoutLayer::new(request_timeout))
                .layer(ConcurrencyLimitLayer::new(concurrency_limit))
                .layer(RequestBodyLimitLayer::new(request_body_limit_bytes))
                .layer(CorsLayer::new()),
        )
        .route(config::serve::HEALTH_CHECK_PATH, get(probe::healthcheck))
        .layer(RequestMetricsLayer::new())
        .fallback(not_found);

    ApiService { router, tls_config }
}

mod control {
    use axum::{Extension, Json, http::StatusCode};
    use synd_protocol::daemon::DaemonStatusResponse;

    use crate::session::DaemonSessions;
    use crate::shutdown::{Shutdown, ShutdownReason};

    pub(super) async fn status(
        Extension(sessions): Extension<DaemonSessions>,
    ) -> Json<DaemonStatusResponse> {
        Json(DaemonStatusResponse::new(sessions.status()))
    }

    pub(super) async fn shutdown(Extension(shutdown): Extension<Shutdown>) -> StatusCode {
        shutdown.shutdown_with_reason(ShutdownReason::ApiRequest);
        StatusCode::ACCEPTED
    }
}

async fn handle_middleware_error(err: BoxError) -> (StatusCode, String) {
    if err.is::<tower::timeout::error::Elapsed>() {
        (
            StatusCode::REQUEST_TIMEOUT,
            "Request took too long".to_string(),
        )
    } else {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("Unhandled internal error: {err}"),
        )
    }
}

async fn not_found() -> impl IntoResponse {
    StatusCode::NOT_FOUND
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn error_mapping() {
        assert_eq!(
            handle_middleware_error(Box::new(tower::timeout::error::Elapsed::new()))
                .await
                .0,
            StatusCode::REQUEST_TIMEOUT
        );
        assert_eq!(
            handle_middleware_error(Box::new(std::io::Error::from(
                std::io::ErrorKind::OutOfMemory
            )))
            .await
            .0,
            StatusCode::INTERNAL_SERVER_ERROR,
        );

        assert_eq!(
            not_found().await.into_response().status(),
            StatusCode::NOT_FOUND
        );
    }
}