haematite_cli/serve/query/
settings.rs

1use std::num::{NonZeroU64, NonZeroUsize};
2use std::path::PathBuf;
3use std::time::{Duration, Instant};
4
5use serde::{Deserialize, Deserializer, de::Error};
6
7pub const NAMES: [&str; 6] = [
8    "query_time_ms",
9    "query_memory_bytes",
10    "query_outputs_per_record",
11    "query_scan_records",
12    "query_max_running",
13    "query_worker",
14];
15
16/// Omission disables a group; an explicit null is an invalid setting.
17#[derive(Clone, Copy, Debug, Default)]
18pub enum Setting<T> {
19    #[default]
20    Absent,
21    Present(T),
22}
23
24impl<'de, T: Deserialize<'de>> Deserialize<'de> for Setting<T> {
25    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
26        T::deserialize(deserializer).map(Self::Present)
27    }
28}
29
30impl<T> Setting<T> {
31    pub const fn is_set(&self) -> bool {
32        matches!(self, Self::Present(_))
33    }
34
35    fn into_value(self) -> Option<T> {
36        match self {
37            Self::Absent => None,
38            Self::Present(value) => Some(value),
39        }
40    }
41}
42
43fn named<'de, T: Deserialize<'de>, D: Deserializer<'de>>(
44    deserializer: D,
45    name: &str,
46) -> Result<Setting<T>, D::Error> {
47    Setting::deserialize(deserializer)
48        .map_err(|error| D::Error::custom(format!("query setting {name}: {error}")))
49}
50
51/// Deserialize the query time allowance.
52///
53/// # Errors
54/// Returns a deserialization error naming `query_time_ms` for null or an invalid value.
55pub fn time_ms<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Setting<u64>, D::Error> {
56    named(deserializer, NAMES[0])
57}
58/// Deserialize the query memory allowance.
59///
60/// # Errors
61/// Returns a deserialization error naming `query_memory_bytes` for null or an invalid value.
62pub fn memory_bytes<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Setting<u64>, D::Error> {
63    named(deserializer, NAMES[1])
64}
65/// Deserialize the per-record output allowance.
66///
67/// # Errors
68/// Returns a deserialization error naming `query_outputs_per_record` for null or an invalid value.
69pub fn outputs_per_record<'de, D: Deserializer<'de>>(
70    deserializer: D,
71) -> Result<Setting<u64>, D::Error> {
72    named(deserializer, NAMES[2])
73}
74/// Deserialize the record scan allowance.
75///
76/// # Errors
77/// Returns a deserialization error naming `query_scan_records` for null or an invalid value.
78pub fn scan_records<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Setting<u64>, D::Error> {
79    named(deserializer, NAMES[3])
80}
81/// Deserialize the concurrent-query limit.
82///
83/// # Errors
84/// Returns a deserialization error naming `query_max_running` for null or an invalid value.
85pub fn max_running<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Setting<u64>, D::Error> {
86    named(deserializer, NAMES[4])
87}
88/// Deserialize the query worker path.
89///
90/// # Errors
91/// Returns a deserialization error naming `query_worker` for null or an invalid value.
92pub fn worker<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Setting<PathBuf>, D::Error> {
93    named(deserializer, NAMES[5])
94}
95
96#[derive(Clone, Debug, Default, Deserialize)]
97#[serde(deny_unknown_fields)]
98pub struct Options {
99    #[serde(default, deserialize_with = "time_ms")]
100    pub query_time_ms: Setting<u64>,
101    #[serde(default, deserialize_with = "memory_bytes")]
102    pub query_memory_bytes: Setting<u64>,
103    #[serde(default, deserialize_with = "outputs_per_record")]
104    pub query_outputs_per_record: Setting<u64>,
105    #[serde(default, deserialize_with = "scan_records")]
106    pub query_scan_records: Setting<u64>,
107    #[serde(default, deserialize_with = "max_running")]
108    pub query_max_running: Setting<u64>,
109    #[serde(default, deserialize_with = "worker")]
110    pub query_worker: Setting<PathBuf>,
111}
112
113#[derive(Debug)]
114pub struct Settings {
115    pub time_ms: NonZeroU64,
116    pub memory_bytes: NonZeroU64,
117    pub outputs_per_record: NonZeroU64,
118    pub scan_records: NonZeroU64,
119    pub max_running: NonZeroUsize,
120    pub worker: PathBuf,
121}
122
123/// The fewest records a page may read: a later page reads its pending record
124/// again, and must read one more to move on.
125pub const MIN_SCAN_RECORDS: u64 = 2;
126
127fn required<T>(setting: Setting<T>, name: &str) -> Result<T, String> {
128    setting
129        .into_value()
130        .ok_or_else(|| format!("missing query setting {name}"))
131}
132
133fn positive(setting: Setting<u64>, name: &str) -> Result<NonZeroU64, String> {
134    NonZeroU64::new(required(setting, name)?)
135        .ok_or_else(|| format!("query setting {name} must be positive"))
136}
137
138impl Options {
139    /// Resolve the optional query settings as a complete validated group.
140    ///
141    /// # Errors
142    /// Refuses missing or zero settings, an unrepresentable time or process count, a scan limit below two, or a worker path that is not absolute.
143    pub fn resolve(self) -> Result<Option<Settings>, String> {
144        let present = [
145            self.query_time_ms.is_set(),
146            self.query_memory_bytes.is_set(),
147            self.query_outputs_per_record.is_set(),
148            self.query_scan_records.is_set(),
149            self.query_max_running.is_set(),
150            self.query_worker.is_set(),
151        ];
152        if present.iter().all(|set| !set) {
153            return Ok(None);
154        }
155        let missing: Vec<_> = NAMES
156            .into_iter()
157            .zip(present)
158            .filter_map(|(name, set)| (!set).then_some(name))
159            .collect();
160        if !missing.is_empty() {
161            return Err(format!("missing query settings: {}", missing.join(", ")));
162        }
163        let time_ms = positive(self.query_time_ms, NAMES[0])?;
164        if Instant::now()
165            .checked_add(Duration::from_millis(time_ms.get()))
166            .is_none()
167        {
168            return Err(format!(
169                "query setting {} exceeds the monotonic clock range",
170                NAMES[0]
171            ));
172        }
173        let memory_bytes = positive(self.query_memory_bytes, NAMES[1])?;
174        let outputs_per_record = positive(self.query_outputs_per_record, NAMES[2])?;
175        let scan_records = positive(self.query_scan_records, NAMES[3])?;
176        if scan_records.get() < MIN_SCAN_RECORDS {
177            return Err(format!(
178                "query setting {} must be at least {MIN_SCAN_RECORDS}, so that every page \
179                 moves on past the record it reads again",
180                NAMES[3]
181            ));
182        }
183        let count = positive(self.query_max_running, NAMES[4])?;
184        let max_running = NonZeroUsize::new(
185            usize::try_from(count.get())
186                .map_err(|_| format!("query setting {} does not fit this platform", NAMES[4]))?,
187        )
188        .ok_or_else(|| format!("query setting {} must be positive", NAMES[4]))?;
189        let worker = required(self.query_worker, NAMES[5])?;
190        if !worker.is_absolute() {
191            return Err(format!(
192                "query setting {} must be an absolute path",
193                NAMES[5]
194            ));
195        }
196        Ok(Some(Settings {
197            time_ms,
198            memory_bytes,
199            outputs_per_record,
200            scan_records,
201            max_running,
202            worker,
203        }))
204    }
205}
haematite_cli/serve/query/settings.rs

haematite_cli/serve/query/
settings.rs
