haematite_cli/serve/query/
start.rs

1use std::collections::BTreeMap;
2use std::num::NonZeroU32;
3use std::time::{Duration, Instant};
4
5use haematite_query::handle::{Error, Handle, StartupStage};
6use haematite_query::{Limits, Ready, Refusal, Start};
7use serde_json::value::RawValue;
8
9use super::{Settings, probe};
10use crate::serve::storage::Failure;
11
12#[must_use]
13pub fn request(
14    settings: &Settings,
15    frame_bytes: NonZeroU32,
16    filter: String,
17    alternate: Option<String>,
18    parameters: BTreeMap<String, Box<RawValue>>,
19) -> Start {
20    Start {
21        protocol: haematite_query::PROTOCOL,
22        filter,
23        alternate,
24        limits: Limits {
25            time_ms: settings.time_ms,
26            memory_bytes: settings.memory_bytes.get(),
27            outputs_per_input: settings.outputs_per_record.get(),
28            frame_bytes,
29        },
30        parameters,
31        binding_names: haematite_query::page::BINDING_NAMES
32            .iter()
33            .map(|name| (*name).to_owned())
34            .collect(),
35    }
36}
37
38pub(super) fn deadline(settings: &Settings) -> Result<Instant, Failure> {
39    Instant::now()
40        .checked_add(Duration::from_millis(settings.time_ms.get()))
41        .ok_or_else(|| {
42            Failure::new(
43                "query_worker_start_timeout",
44                "query_time_ms cannot be represented by the monotonic clock",
45            )
46        })
47}
48
49pub(super) fn check(settings: &Settings, deadline: Instant) -> Result<(), Failure> {
50    if !settings.worker.is_absolute() {
51        return Err(Failure::new(
52            "query_worker_path_invalid",
53            "query_worker must be an absolute path",
54        ));
55    }
56    let metadata = std::fs::metadata(&settings.worker).map_err(|error| {
57        let code = if error.kind() == std::io::ErrorKind::NotFound {
58            "query_worker_missing"
59        } else {
60            "query_worker_path_unreadable"
61        };
62        Failure::new(
63            code,
64            format!("query_worker {}: {error}", settings.worker.display()),
65        )
66    })?;
67    if !metadata.is_file() {
68        return Err(Failure::new(
69            "query_worker_not_file",
70            format!(
71                "query_worker {} is not a regular file",
72                settings.worker.display()
73            ),
74        ));
75    }
76    if Instant::now() >= deadline {
77        return Err(Failure::new(
78            "query_worker_start_timeout",
79            format!(
80                "startup allowance query_time_ms={} ms expired before spawn",
81                settings.time_ms
82            ),
83        ));
84    }
85    Ok(())
86}
87
88fn boundary_error(failure: &Failure) -> Error {
89    let code = if failure.code == "query_worker_start_timeout" {
90        Refusal::QueryTimeLimit
91    } else {
92        Refusal::QueryEngineUnavailable
93    };
94    Error::new(code, failure.to_string()).at_start(StartupStage::Spawn, None)
95}
96
97pub(super) fn connect(
98    settings: &Settings,
99    request: &Start,
100    deadline: Instant,
101) -> Result<(Handle, Ready), Error> {
102    Handle::start(
103        &settings.worker,
104        request,
105        env!("CARGO_PKG_VERSION"),
106        deadline,
107    )
108}
109
110/// Every call examines and starts the current worker, independently of the probe.
111///
112/// # Errors
113/// Propagates path validation, startup deadline and worker handshake failures with their startup stage.
114pub fn open(
115    settings: &Settings,
116    frame_bytes: NonZeroU32,
117    filter: String,
118    alternate: Option<String>,
119    parameters: BTreeMap<String, Box<RawValue>>,
120) -> Result<(Handle, Ready), Error> {
121    let deadline = deadline(settings).map_err(|failure| boundary_error(&failure))?;
122    check(settings, deadline).map_err(|failure| boundary_error(&failure))?;
123    let request = request(settings, frame_bytes, filter, alternate, parameters);
124    connect(settings, &request, deadline)
125}
126
127/// The wire name of a worker refusal.
128#[must_use]
129pub const fn code_name(code: Refusal) -> &'static str {
130    match code {
131        Refusal::QueryCompile => "query_compile",
132        Refusal::QueryPattern => "query_pattern",
133        Refusal::QueryFlag => "query_flag",
134        Refusal::QueryRuntime => "query_runtime",
135        Refusal::QueryOutputLimit => "query_output_limit",
136        Refusal::QueryValueTooLarge => "query_value_too_large",
137        Refusal::QueryFunctionRefused => "query_function_refused",
138        Refusal::QueryProtocol => "query_protocol",
139        Refusal::QueryEngineMismatch => "query_engine_mismatch",
140        Refusal::QueryEngineUnavailable => "query_engine_unavailable",
141        Refusal::QueryTimeLimit => "query_time_limit",
142        Refusal::QueryMemoryLimit => "query_memory_limit",
143        Refusal::QueryStopped => "query_stopped",
144        Refusal::QueryScanLimit => "query_scan_limit",
145        Refusal::QueryPercentileRange => "query_percentile_range",
146        Refusal::QueryMomentGone => "query_moment_gone",
147        Refusal::QueryReadFailed => "query_read_failed",
148    }
149}
150
151/// Classify startup for the caller without changing the operator's probe names.
152#[must_use]
153pub fn caller_failure(error: &Error) -> Failure {
154    let code = match error.refusal.code {
155        Refusal::QueryTimeLimit | Refusal::QueryMemoryLimit => error.refusal.code,
156        _ => match error.startup.map(|startup| startup.stage) {
157            Some(StartupStage::Failed(code)) => code,
158            Some(StartupStage::Identity) => Refusal::QueryEngineMismatch,
159            Some(StartupStage::UnofferedReading) => Refusal::QueryProtocol,
160            _ => Refusal::QueryEngineUnavailable,
161        },
162    };
163    let mut message = if code == Refusal::QueryEngineUnavailable {
164        format!("{}: {error}", probe::classified_name(error))
165    } else {
166        error.to_string()
167    };
168    if let Some(status) = &error.status {
169        message = format!("{message}; worker status: {status}");
170    }
171    if let Some(cause) = &error.stderr_error {
172        message = format!("{message}; stderr reader: {cause}");
173    }
174    Failure::new(code_name(code), message)
175}
176
177#[cfg(test)]
178#[path = "start_tests.rs"]
179mod tests;
haematite_cli/serve/query/start.rs

haematite_cli/serve/query/
start.rs
