haematite_cli/serve/
cli_config.rs

1use super::super::{
2    changes, embed,
3    live::LysLive,
4    pass::Gate,
5    placement,
6    query::{Options, Setting, Settings},
7    releaser,
8    storage::{KEPT_VERSIONS_DEFAULT, KEPT_VERSIONS_MOST},
9};
10#[cfg(test)]
11use super::endpoint_tests;
12use crate::serve::service::Service;
13use haematite::branch::KeptLimit;
14use haematite::db::DatabaseConfig;
15use haematite::store::NodeCacheBudget;
16use serde::Deserialize;
17use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
18use std::path::PathBuf;
19use std::sync::Arc;
20#[derive(Debug, Deserialize)]
21#[serde(deny_unknown_fields)]
22pub struct Config {
23    pub mcp_http: Option<crate::mcp::http::Config>,
24    pub data_dir: PathBuf,
25    pub socket: Option<PathBuf>,
26    pub name: Option<String>,
27    pub branch_cache_bytes_per_shard: NonZeroUsize,
28    pub max_frame: NonZeroU32,
29    pub max_clients: NonZeroUsize,
30    pub max_page: NonZeroU32,
31    #[serde(default = "append_prune_keep_ceiling")]
32    pub append_prune_keep_ceiling: u32,
33    #[serde(default = "append_prune_removal_cap")]
34    pub append_prune_removal_cap: u32,
35    pub timeout_ms: NonZeroU64,
36    pub create: Option<Creation>,
37    /// `changes.list`'s limits, each defaulted when absent.
38    #[serde(default = "changes_scan_records")]
39    pub changes_scan_records: NonZeroU64,
40    #[serde(default = "changes_node_budget")]
41    pub changes_node_budget: NonZeroUsize,
42    #[serde(default = "changes_max_page")]
43    pub changes_max_page: NonZeroU32,
44    #[serde(default, deserialize_with = "super::super::query::settings::time_ms")]
45    pub query_time_ms: Setting<u64>,
46    #[serde(
47        default,
48        deserialize_with = "super::super::query::settings::memory_bytes"
49    )]
50    pub query_memory_bytes: Setting<u64>,
51    #[serde(
52        default,
53        deserialize_with = "super::super::query::settings::outputs_per_record"
54    )]
55    pub query_outputs_per_record: Setting<u64>,
56    #[serde(
57        default,
58        deserialize_with = "super::super::query::settings::scan_records"
59    )]
60    pub query_scan_records: Setting<u64>,
61    #[serde(
62        default,
63        deserialize_with = "super::super::query::settings::max_running"
64    )]
65    pub query_max_running: Setting<u64>,
66    #[serde(default, deserialize_with = "super::super::query::settings::worker")]
67    pub query_worker: Setting<PathBuf>,
68    #[serde(default)]
69    pub embed_worker: Setting<PathBuf>,
70    #[serde(default)]
71    pub embed_model_dir: Setting<PathBuf>,
72    #[serde(default)]
73    pub embed_model_sha256: Setting<String>,
74    #[serde(default)]
75    pub embed_memory_bytes: Setting<u64>,
76    #[serde(default)]
77    pub embed_threads: Setting<u64>,
78    #[serde(default)]
79    pub embed_batch_items: Setting<u64>,
80    #[serde(default)]
81    pub embed_input_bytes: Setting<u64>,
82    #[serde(default)]
83    pub embed_tokens: Setting<u64>,
84    #[serde(default)]
85    pub embed_frame_bytes: Setting<u32>,
86    #[serde(default)]
87    pub embed_ready_ms: Setting<u64>,
88    #[serde(default)]
89    pub embed_call_ms: Setting<u64>,
90    #[serde(default)]
91    pub embed_waiting: Setting<u64>,
92    #[serde(default)]
93    pub embed_request_items: Setting<u64>,
94    #[serde(default)]
95    pub embed_page_records: Setting<u64>,
96    #[serde(default)]
97    pub embed_writer_wait_ms: Setting<u64>,
98    /// EMB-E2: embed a text longer than the model reads whole, as windows.
99    #[serde(default)]
100    pub embed_windows: Setting<bool>,
101    /// EMB-E2: the most windows one text is embedded as; 32 when absent.
102    #[serde(default)]
103    pub embed_max_windows: Setting<u64>,
104    /// EMB-E2: the tokens consecutive windows share; 64 when absent.
105    #[serde(default)]
106    pub embed_window_overlap_tokens: Setting<u64>,
107    /// EWR1: the wait before the second start of an embedding worker that
108    /// died, doubled for each later one; one second when absent.
109    #[serde(default)]
110    pub embed_restart_ms: Setting<u64>,
111    /// EWR1: the longest wait between two starts of the embedding worker;
112    /// five minutes when absent.
113    #[serde(default)]
114    pub embed_restart_most_ms: Setting<u64>,
115    /// SL1: whether text search asks its questions of a worker of its own,
116    /// started on the first search, so a search never waits behind the
117    /// embedding walk. On when absent. It holds its own copy of the model:
118    /// with it the two workers may use twice `embed_memory_bytes`.
119    #[serde(default)]
120    pub embed_question_worker: Setting<bool>,
121    /// Number of previous versions each branch record keeps,
122    /// a whole number from 1 to [`KEPT_VERSIONS_MOST`].
123    #[serde(default = "kept_versions", deserialize_with = "kept_versions_within")]
124    pub kept_versions: usize,
125    /// STOR-003 HR4: when the release thread runs and what it keeps.
126    #[serde(default)]
127    pub retention: releaser::Config,
128    /// ACCESS-001 R2: decide every request through this Lys. Absent, the
129    /// service is its owner's alone, reached only by its 0700 socket, and
130    /// decides nothing.
131    #[serde(default)]
132    pub lys: Option<LysConfig>,
133}
134
135/// The Lys a served database decides every request through. Nothing here
136/// grants anything: every right comes from a pass or a live answer.
137#[derive(Debug, Deserialize)]
138#[serde(deny_unknown_fields)]
139pub struct LysConfig {
140    /// The issuer origin whose passes are trusted and whose live check
141    /// deliberate routes ask.
142    pub issuer: String,
143    /// This database's id as Lys registers it, a `haematite.database`.
144    pub database: String,
145    /// The absolute path of a file holding haematite's Lys app credential,
146    /// `lys-app.haematite.<secret>`. Given, every collection and branch is
147    /// placed in Lys under this database, at start and after the write that
148    /// makes each one commits, so a grant on the database reaches it
149    /// (serve/placement.rs). Absent, nothing is placed, and a new collection
150    /// or branch is reached only by a grant on itself until someone places it.
151    #[serde(default)]
152    pub placement_credential_file: Option<PathBuf>,
153}
154
155/// The Lys gate and placer a configuration asks for, both made before a
156/// store opens or a socket binds. Neither waits on Lys: the gate reads its
157/// keys on a thread of its own and refuses every decided act as
158/// `lys_keys_pending` until they are read, so the service listens at once
159/// and never serves undecided. A wrong credential file stops it by name.
160pub(super) fn lys_parts(
161    lys: Option<&LysConfig>,
162) -> Result<(Option<Gate>, Option<placement::Placer>), String> {
163    let gate = lys.map(lys_gate).transpose()?;
164    let placer = lys
165        .and_then(|lys| {
166            lys.placement_credential_file.as_deref().map(|file| {
167                placement::Lys::new(&lys.issuer, &lys.database, file)
168                    .and_then(placement::Placer::start)
169            })
170        })
171        .transpose()?;
172    Ok((gate, placer))
173}
174
175/// `service` deciding through `gate` and placing through `placer`, each
176/// when given.
177pub(super) fn served_through_lys(
178    mut service: Service,
179    gate: Option<Gate>,
180    placer: Option<placement::Placer>,
181) -> Result<Service, String> {
182    if let Some(gate) = gate {
183        service = service.with_lys(gate);
184    }
185    if let Some(placer) = placer {
186        service = service
187            .with_placer(placer)
188            .map_err(|error| error.to_string())?;
189    }
190    Ok(service)
191}
192
193/// The gate for `config`, its keys read from Lys in the background.
194///
195/// # Errors
196///
197/// Refuses an empty or placed database id, or an issuer that is not an HTTP
198/// origin, by name.
199pub(super) fn lys_gate(config: &LysConfig) -> Result<Gate, String> {
200    if config.database.is_empty()
201        || !config
202            .database
203            .bytes()
204            .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_' || byte == b'-')
205    {
206        return Err(format!(
207            "lys database id {:?} must be letters, digits, _ or - only",
208            config.database
209        ));
210    }
211    let mut live = LysLive::new(&config.issuer)
212        .map_err(|error| format!("{}: the lys issuer {}", error.name(), config.issuer))?;
213    if let Some(file) = config.placement_credential_file.as_deref() {
214        live = live.asking_as(super::super::lys_schema::read_credential(file)?);
215    }
216    let live = Arc::new(live);
217    let asking = Arc::clone(&live);
218    Gate::fetching(
219        move || asking.keys(),
220        &config.issuer,
221        &config.database,
222        live,
223    )
224    .map_err(|error| format!("the thread reading Lys's keys could not start: {error}"))
225}
226
227const fn kept_versions() -> usize {
228    KEPT_VERSIONS_DEFAULT
229}
230
231/// `kept_versions` as read, refused by name outside 1 to the most.
232fn kept_versions_within<'de, D>(deserializer: D) -> Result<usize, D::Error>
233where
234    D: serde::Deserializer<'de>,
235{
236    let value = u64::deserialize(deserializer)?;
237    usize::try_from(value)
238        .ok()
239        .filter(|kept| (1..=KEPT_VERSIONS_MOST).contains(kept))
240        .ok_or_else(|| {
241            serde::de::Error::custom(format!(
242                "kept_versions must be a whole number from 1 to {KEPT_VERSIONS_MOST}, not {value}"
243            ))
244        })
245}
246
247const fn append_prune_keep_ceiling() -> u32 {
248    4_096
249}
250
251const fn append_prune_removal_cap() -> u32 {
252    64
253}
254
255fn changes_scan_records() -> NonZeroU64 {
256    changes::Settings::default().scan_records
257}
258
259fn changes_node_budget() -> NonZeroUsize {
260    changes::Settings::default().node_budget
261}
262
263fn changes_max_page() -> NonZeroU32 {
264    changes::Settings::default().max_page
265}
266
267impl Config {
268    /// Resolves the endpoint when a caller needs it; deserialization remains pure.
269    ///
270    /// # Errors
271    /// Returns the runtime resolver's named input, ownership and OS-query refusals.
272    pub fn endpoint(
273        &self,
274        create: bool,
275    ) -> Result<super::super::runtime::Endpoint, super::super::runtime::Refusal> {
276        #[cfg(test)]
277        if let Some((root, uid)) = endpoint_tests::input() {
278            return super::super::runtime::resolve_input(
279                &self.data_dir,
280                self.name.as_deref(),
281                self.socket.as_deref(),
282                create,
283                &root,
284                uid,
285            );
286        }
287        super::super::runtime::resolve(
288            &self.data_dir,
289            self.name.as_deref(),
290            self.socket.as_deref(),
291            create,
292        )
293    }
294
295    /// The kept window every branch record keeps.
296    ///
297    /// # Errors
298    /// Only for a `kept_versions` of zero, which loading refuses first.
299    pub fn kept_limit(&self) -> Result<KeptLimit, String> {
300        KeptLimit::new(self.kept_versions).map_err(|error| error.to_string())
301    }
302
303    /// # Errors
304    /// Refuses either zero retention setting before opening a store.
305    pub fn append_prune_limits(&self) -> Result<super::super::storage::AppendPruneLimits, String> {
306        super::super::storage::AppendPruneLimits::new(
307            self.append_prune_keep_ceiling,
308            self.append_prune_removal_cap,
309        )
310        .map_err(|failure| failure.to_string())
311    }
312
313    /// `changes.list`'s limits; a page must be able to move past a change.
314    ///
315    /// # Errors
316    /// Refuses `changes_scan_records` below two, which cannot advance a page.
317    pub fn changes_settings(&self) -> Result<changes::Settings, String> {
318        if self.changes_scan_records.get() < 2 {
319            return Err("changes_scan_records must be at least 2".to_owned());
320        }
321        Ok(changes::Settings {
322            scan_records: self.changes_scan_records,
323            node_budget: self.changes_node_budget,
324            max_page: self.changes_max_page,
325        })
326    }
327
328    /// Resolve the optional query settings as one complete group.
329    ///
330    /// # Errors
331    /// Refuses missing, zero, out-of-range or incompatible settings and a worker path that is not absolute.
332    pub fn query_settings(&self) -> Result<Option<Settings>, String> {
333        Options {
334            query_time_ms: self.query_time_ms,
335            query_memory_bytes: self.query_memory_bytes,
336            query_outputs_per_record: self.query_outputs_per_record,
337            query_scan_records: self.query_scan_records,
338            query_max_running: self.query_max_running,
339            query_worker: self.query_worker.clone(),
340        }
341        .resolve()
342    }
343
344    /// The required embedding settings, plus the optional writer wait.
345    ///
346    /// # Errors
347    ///
348    /// As [`embed::Options::resolve`].
349    pub fn embed_settings(&self) -> Result<Option<embed::Settings>, String> {
350        embed::Options {
351            worker: self.embed_worker.clone(),
352            model_dir: self.embed_model_dir.clone(),
353            model_sha256: self.embed_model_sha256.clone(),
354            memory_bytes: self.embed_memory_bytes,
355            threads: self.embed_threads,
356            batch_items: self.embed_batch_items,
357            input_bytes: self.embed_input_bytes,
358            tokens: self.embed_tokens,
359            frame_bytes: self.embed_frame_bytes,
360            ready_ms: self.embed_ready_ms,
361            call_ms: self.embed_call_ms,
362            waiting: self.embed_waiting,
363            request_items: self.embed_request_items,
364            page_records: self.embed_page_records,
365            writer_wait_ms: self.embed_writer_wait_ms,
366            windows: self.embed_windows,
367            max_windows: self.embed_max_windows,
368            overlap_tokens: self.embed_window_overlap_tokens,
369        }
370        .resolve()
371    }
372}
373
374#[derive(Debug, Deserialize)]
375#[serde(deny_unknown_fields)]
376pub struct Creation {
377    pub shards: NonZeroUsize,
378    pub cache_bytes_total: NonZeroUsize,
379    pub executor_threads: NonZeroUsize,
380}
381
382/// The engine configuration of a store the service creates: the one place it
383/// creates one. The store reclaims what nothing keeps, under the default
384/// marking bound (STOR-003 H1a).
385pub(super) const fn creation_config(data_dir: PathBuf, creation: &Creation) -> DatabaseConfig {
386    DatabaseConfig {
387        data_dir,
388        shard_count: creation.shards.get(),
389        distributed: None,
390        executor_threads: Some(creation.executor_threads.get()),
391        node_cache_budget: Some(NodeCacheBudget::Bytes(creation.cache_bytes_total)),
392        retention: haematite::Retention::reclaim(0),
393    }
394}
haematite_cli/serve/cli_config.rs

haematite_cli/serve/
cli_config.rs
