haematite-cli 0.21.8

use std::collections::BTreeMap;
use std::num::NonZeroU32;
use std::time::{Duration, Instant};

use haematite_query::handle::{Error, Handle, StartupStage};
use haematite_query::{Limits, Ready, Refusal, Start};
use serde_json::value::RawValue;

use super::{Settings, probe};
use crate::serve::storage::Failure;

#[must_use]
pub fn request(
    settings: &Settings,
    frame_bytes: NonZeroU32,
    filter: String,
    alternate: Option<String>,
    parameters: BTreeMap<String, Box<RawValue>>,
) -> Start {
    Start {
        protocol: haematite_query::PROTOCOL,
        filter,
        alternate,
        limits: Limits {
            time_ms: settings.time_ms,
            memory_bytes: settings.memory_bytes.get(),
            outputs_per_input: settings.outputs_per_record.get(),
            frame_bytes,
        },
        parameters,
        binding_names: haematite_query::page::BINDING_NAMES
            .iter()
            .map(|name| (*name).to_owned())
            .collect(),
    }
}

pub(super) fn deadline(settings: &Settings) -> Result<Instant, Failure> {
    Instant::now()
        .checked_add(Duration::from_millis(settings.time_ms.get()))
        .ok_or_else(|| {
            Failure::new(
                "query_worker_start_timeout",
                "query_time_ms cannot be represented by the monotonic clock",
            )
        })
}

pub(super) fn check(settings: &Settings, deadline: Instant) -> Result<(), Failure> {
    if !settings.worker.is_absolute() {
        return Err(Failure::new(
            "query_worker_path_invalid",
            "query_worker must be an absolute path",
        ));
    }
    let metadata = std::fs::metadata(&settings.worker).map_err(|error| {
        let code = if error.kind() == std::io::ErrorKind::NotFound {
            "query_worker_missing"
        } else {
            "query_worker_path_unreadable"
        };
        Failure::new(
            code,
            format!("query_worker {}: {error}", settings.worker.display()),
        )
    })?;
    if !metadata.is_file() {
        return Err(Failure::new(
            "query_worker_not_file",
            format!(
                "query_worker {} is not a regular file",
                settings.worker.display()
            ),
        ));
    }
    if Instant::now() >= deadline {
        return Err(Failure::new(
            "query_worker_start_timeout",
            format!(
                "startup allowance query_time_ms={} ms expired before spawn",
                settings.time_ms
            ),
        ));
    }
    Ok(())
}

fn boundary_error(failure: &Failure) -> Error {
    let code = if failure.code == "query_worker_start_timeout" {
        Refusal::QueryTimeLimit
    } else {
        Refusal::QueryEngineUnavailable
    };
    Error::new(code, failure.to_string()).at_start(StartupStage::Spawn, None)
}

pub(super) fn connect(
    settings: &Settings,
    request: &Start,
    deadline: Instant,
) -> Result<(Handle, Ready), Error> {
    Handle::start(
        &settings.worker,
        request,
        env!("CARGO_PKG_VERSION"),
        deadline,
    )
}

/// Every call examines and starts the current worker, independently of the probe.
///
/// # Errors
/// Propagates path validation, startup deadline and worker handshake failures with their startup stage.
pub fn open(
    settings: &Settings,
    frame_bytes: NonZeroU32,
    filter: String,
    alternate: Option<String>,
    parameters: BTreeMap<String, Box<RawValue>>,
) -> Result<(Handle, Ready), Error> {
    let deadline = deadline(settings).map_err(|failure| boundary_error(&failure))?;
    check(settings, deadline).map_err(|failure| boundary_error(&failure))?;
    let request = request(settings, frame_bytes, filter, alternate, parameters);
    connect(settings, &request, deadline)
}

/// The wire name of a worker refusal.
#[must_use]
pub const fn code_name(code: Refusal) -> &'static str {
    match code {
        Refusal::QueryCompile => "query_compile",
        Refusal::QueryPattern => "query_pattern",
        Refusal::QueryFlag => "query_flag",
        Refusal::QueryRuntime => "query_runtime",
        Refusal::QueryOutputLimit => "query_output_limit",
        Refusal::QueryValueTooLarge => "query_value_too_large",
        Refusal::QueryFunctionRefused => "query_function_refused",
        Refusal::QueryProtocol => "query_protocol",
        Refusal::QueryEngineMismatch => "query_engine_mismatch",
        Refusal::QueryEngineUnavailable => "query_engine_unavailable",
        Refusal::QueryTimeLimit => "query_time_limit",
        Refusal::QueryMemoryLimit => "query_memory_limit",
        Refusal::QueryStopped => "query_stopped",
        Refusal::QueryScanLimit => "query_scan_limit",
        Refusal::QueryPercentileRange => "query_percentile_range",
        Refusal::QueryMomentGone => "query_moment_gone",
        Refusal::QueryReadFailed => "query_read_failed",
    }
}

/// Classify startup for the caller without changing the operator's probe names.
#[must_use]
pub fn caller_failure(error: &Error) -> Failure {
    let code = match error.refusal.code {
        Refusal::QueryTimeLimit | Refusal::QueryMemoryLimit => error.refusal.code,
        _ => match error.startup.map(|startup| startup.stage) {
            Some(StartupStage::Failed(code)) => code,
            Some(StartupStage::Identity) => Refusal::QueryEngineMismatch,
            Some(StartupStage::UnofferedReading) => Refusal::QueryProtocol,
            _ => Refusal::QueryEngineUnavailable,
        },
    };
    let mut message = if code == Refusal::QueryEngineUnavailable {
        format!("{}: {error}", probe::classified_name(error))
    } else {
        error.to_string()
    };
    if let Some(status) = &error.status {
        message = format!("{message}; worker status: {status}");
    }
    if let Some(cause) = &error.stderr_error {
        message = format!("{message}; stderr reader: {cause}");
    }
    Failure::new(code_name(code), message)
}

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