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    /// HAEMNET-001 R1: the network endpoint, served beside the Unix socket
134    /// only when named, and only through `lys` (R2).
135    #[serde(default)]
136    pub network: Option<crate::serve::transport::network::NetworkConfig>,
137}
138
139/// The Lys a served database decides every request through. Nothing here
140/// grants anything: every right comes from a pass or a live answer.
141#[derive(Debug, Deserialize)]
142#[serde(deny_unknown_fields)]
143pub struct LysConfig {
144    /// The issuer origin whose passes are trusted and whose live check
145    /// deliberate routes ask.
146    pub issuer: String,
147    /// This database's id as Lys registers it, a `haematite.database`.
148    pub database: String,
149    /// The absolute path of a file holding haematite's Lys app credential,
150    /// `lys-app.haematite.<secret>`. Given, every collection and branch is
151    /// placed in Lys under this database, at start and after the write that
152    /// makes each one commits, so a grant on the database reaches it
153    /// (serve/placement.rs). Absent, nothing is placed, and a new collection
154    /// or branch is reached only by a grant on itself until someone places it.
155    #[serde(default)]
156    pub placement_credential_file: Option<PathBuf>,
157}
158
159/// The Lys gate and placer a configuration asks for, both made before a
160/// store opens or a socket binds. Neither waits on Lys: the gate reads its
161/// keys on a thread of its own and refuses every decided act as
162/// `lys_keys_pending` until they are read, so the service listens at once
163/// and never serves undecided. A wrong credential file stops it by name.
164pub(super) fn lys_parts(
165    lys: Option<&LysConfig>,
166) -> Result<(Option<Gate>, Option<placement::Placer>), String> {
167    let gate = lys.map(lys_gate).transpose()?;
168    let placer = lys
169        .and_then(|lys| {
170            lys.placement_credential_file.as_deref().map(|file| {
171                placement::Lys::new(&lys.issuer, &lys.database, file)
172                    .and_then(placement::Placer::start)
173            })
174        })
175        .transpose()?;
176    Ok((gate, placer))
177}
178
179/// `service` deciding through `gate` and placing through `placer`, each
180/// when given.
181pub(super) fn served_through_lys(
182    mut service: Service,
183    gate: Option<Gate>,
184    placer: Option<placement::Placer>,
185) -> Result<Service, String> {
186    if let Some(gate) = gate {
187        service = service.with_lys(gate);
188    }
189    if let Some(placer) = placer {
190        service = service
191            .with_placer(placer)
192            .map_err(|error| error.to_string())?;
193    }
194    Ok(service)
195}
196
197/// The gate for `config`, its keys read from Lys in the background.
198///
199/// # Errors
200///
201/// Refuses an empty or placed database id, or an issuer that is not an HTTP
202/// origin, by name.
203pub(super) fn lys_gate(config: &LysConfig) -> Result<Gate, String> {
204    if config.database.is_empty()
205        || !config
206            .database
207            .bytes()
208            .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_' || byte == b'-')
209    {
210        return Err(format!(
211            "lys database id {:?} must be letters, digits, _ or - only",
212            config.database
213        ));
214    }
215    let mut live = LysLive::new(&config.issuer)
216        .map_err(|error| format!("{}: the lys issuer {}", error.name(), config.issuer))?;
217    if let Some(file) = config.placement_credential_file.as_deref() {
218        live = live.asking_as(super::super::lys_schema::read_credential(file)?);
219    }
220    let live = Arc::new(live);
221    let asking = Arc::clone(&live);
222    Gate::fetching(
223        move || asking.keys(),
224        &config.issuer,
225        &config.database,
226        live,
227    )
228    .map_err(|error| format!("the thread reading Lys's keys could not start: {error}"))
229}
230
231const fn kept_versions() -> usize {
232    KEPT_VERSIONS_DEFAULT
233}
234
235/// `kept_versions` as read, refused by name outside 1 to the most.
236fn kept_versions_within<'de, D>(deserializer: D) -> Result<usize, D::Error>
237where
238    D: serde::Deserializer<'de>,
239{
240    let value = u64::deserialize(deserializer)?;
241    usize::try_from(value)
242        .ok()
243        .filter(|kept| (1..=KEPT_VERSIONS_MOST).contains(kept))
244        .ok_or_else(|| {
245            serde::de::Error::custom(format!(
246                "kept_versions must be a whole number from 1 to {KEPT_VERSIONS_MOST}, not {value}"
247            ))
248        })
249}
250
251const fn append_prune_keep_ceiling() -> u32 {
252    4_096
253}
254
255const fn append_prune_removal_cap() -> u32 {
256    64
257}
258
259fn changes_scan_records() -> NonZeroU64 {
260    changes::Settings::default().scan_records
261}
262
263fn changes_node_budget() -> NonZeroUsize {
264    changes::Settings::default().node_budget
265}
266
267fn changes_max_page() -> NonZeroU32 {
268    changes::Settings::default().max_page
269}
270
271impl Config {
272    /// Resolves the endpoint when a caller needs it; deserialization remains pure.
273    ///
274    /// # Errors
275    /// Returns the runtime resolver's named input, ownership and OS-query refusals.
276    pub fn endpoint(
277        &self,
278        create: bool,
279    ) -> Result<super::super::runtime::Endpoint, super::super::runtime::Refusal> {
280        #[cfg(test)]
281        if let Some((root, uid)) = endpoint_tests::input() {
282            return super::super::runtime::resolve_input(
283                &self.data_dir,
284                self.name.as_deref(),
285                self.socket.as_deref(),
286                create,
287                &root,
288                uid,
289            );
290        }
291        super::super::runtime::resolve(
292            &self.data_dir,
293            self.name.as_deref(),
294            self.socket.as_deref(),
295            create,
296        )
297    }
298
299    /// The kept window every branch record keeps.
300    ///
301    /// # Errors
302    /// Only for a `kept_versions` of zero, which loading refuses first.
303    pub fn kept_limit(&self) -> Result<KeptLimit, String> {
304        KeptLimit::new(self.kept_versions).map_err(|error| error.to_string())
305    }
306
307    /// # Errors
308    /// Refuses either zero retention setting before opening a store.
309    pub fn append_prune_limits(&self) -> Result<super::super::storage::AppendPruneLimits, String> {
310        super::super::storage::AppendPruneLimits::new(
311            self.append_prune_keep_ceiling,
312            self.append_prune_removal_cap,
313        )
314        .map_err(|failure| failure.to_string())
315    }
316
317    /// `changes.list`'s limits; a page must be able to move past a change.
318    ///
319    /// # Errors
320    /// Refuses `changes_scan_records` below two, which cannot advance a page.
321    pub fn changes_settings(&self) -> Result<changes::Settings, String> {
322        if self.changes_scan_records.get() < 2 {
323            return Err("changes_scan_records must be at least 2".to_owned());
324        }
325        Ok(changes::Settings {
326            scan_records: self.changes_scan_records,
327            node_budget: self.changes_node_budget,
328            max_page: self.changes_max_page,
329        })
330    }
331
332    /// Resolve the optional query settings as one complete group.
333    ///
334    /// # Errors
335    /// Refuses missing, zero, out-of-range or incompatible settings and a worker path that is not absolute.
336    pub fn query_settings(&self) -> Result<Option<Settings>, String> {
337        Options {
338            query_time_ms: self.query_time_ms,
339            query_memory_bytes: self.query_memory_bytes,
340            query_outputs_per_record: self.query_outputs_per_record,
341            query_scan_records: self.query_scan_records,
342            query_max_running: self.query_max_running,
343            query_worker: self.query_worker.clone(),
344        }
345        .resolve()
346    }
347
348    /// The required embedding settings, plus the optional writer wait.
349    ///
350    /// # Errors
351    ///
352    /// As [`embed::Options::resolve`].
353    pub fn embed_settings(&self) -> Result<Option<embed::Settings>, String> {
354        embed::Options {
355            worker: self.embed_worker.clone(),
356            model_dir: self.embed_model_dir.clone(),
357            model_sha256: self.embed_model_sha256.clone(),
358            memory_bytes: self.embed_memory_bytes,
359            threads: self.embed_threads,
360            batch_items: self.embed_batch_items,
361            input_bytes: self.embed_input_bytes,
362            tokens: self.embed_tokens,
363            frame_bytes: self.embed_frame_bytes,
364            ready_ms: self.embed_ready_ms,
365            call_ms: self.embed_call_ms,
366            waiting: self.embed_waiting,
367            request_items: self.embed_request_items,
368            page_records: self.embed_page_records,
369            writer_wait_ms: self.embed_writer_wait_ms,
370            windows: self.embed_windows,
371            max_windows: self.embed_max_windows,
372            overlap_tokens: self.embed_window_overlap_tokens,
373        }
374        .resolve()
375    }
376}
377
378#[derive(Debug, Deserialize)]
379#[serde(deny_unknown_fields)]
380pub struct Creation {
381    pub shards: NonZeroUsize,
382    pub cache_bytes_total: NonZeroUsize,
383    pub executor_threads: NonZeroUsize,
384}
385
386/// The engine configuration of a store the service creates: the one place it
387/// creates one. The store reclaims what nothing keeps, under the default
388/// marking bound (STOR-003 H1a).
389pub(super) const fn creation_config(data_dir: PathBuf, creation: &Creation) -> DatabaseConfig {
390    DatabaseConfig {
391        data_dir,
392        shard_count: creation.shards.get(),
393        distributed: None,
394        executor_threads: Some(creation.executor_threads.get()),
395        node_cache_budget: Some(NodeCacheBudget::Bytes(creation.cache_bytes_total)),
396        retention: haematite::Retention::reclaim(0),
397    }
398}
haematite_cli/serve/cli_config.rs

haematite_cli/serve/
cli_config.rs
