harmos_storage/
config.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Explicit Storage Configuration
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::num::NonZeroUsize;
20use std::time::Duration;
21
22// Third Party Imports
23use serde::{Deserialize, Serialize};
24
25// Local Imports
26
27/* === Constants === */
28
29// Persistence Cadence And Batching Are Private Storage Algorithm Choices.
30pub(super) const FLUSH_INTERVAL: Duration = Duration::from_millis(100);
31pub(super) const MAX_BATCH_BYTES: usize = 16 * 1024 * 1024;
32pub(super) const TARGET_BATCH_BYTES: usize = 1024 * 1024;
33pub(super) const OBJECT_PRUNE_ENTRIES: NonZeroUsize = NonZeroUsize::new(256).unwrap();
34pub(super) const SNAPSHOT_BYTES: usize = 1024 * 1024;
35
36// Backend I/O And Transient Persistence Retry.
37pub(super) const IO_TIMEOUT: Duration = Duration::from_secs(30);
38pub(super) const RECOVERY_TIMEOUT: Duration = Duration::from_secs(300);
39pub(super) const MAX_RETRIES: usize = 3;
40pub(super) const RETRY_BASE: Duration = Duration::from_millis(10);
41pub(super) const RETRY_CAP: Duration = Duration::from_secs(1);
42
43// Producer Sessions And Idempotent Mutation Receipts.
44pub(super) const MAX_FILES: NonZeroUsize = NonZeroUsize::new(256).unwrap();
45pub(super) const FILE_SESSION_TIMEOUT: Duration = Duration::from_secs(60);
46pub(super) const MAX_MUTATIONS: NonZeroUsize = NonZeroUsize::new(1024).unwrap();
47pub(super) const MUTATION_RETENTION: Duration = Duration::from_secs(300);
48// Internal Scan Pages Amortize Backend Calls On Desktop Storage: 16,384 Rows / 8 MiB.
49// Response Sizing And Concurrent Query Counts Remain Explicit Consumer Settings.
50// Query Reductions Cache At Most 32 MiB Of Arrow Pages And 32 MiB Of Coordinates,
51// Then Spill To Disk Within The Existing 128 MiB Algorithm Workspace Target.
52pub(super) const SCAN_ROWS: NonZeroUsize = NonZeroUsize::new(16_384).unwrap();
53pub(super) const SCAN_BYTES: usize = 8 * 1024 * 1024;
54
55/* === Macros === */
56
57/* === Type Aliases === */
58
59/* === Structs === */
60
61#[derive(Clone, Debug, Serialize, Deserialize)]
62#[serde(deny_unknown_fields)]
63pub struct StorageSettings {
64    pub request: StorageRequestSettings,
65    pub response: StorageResponseSettings,
66    pub filesystem: StorageFilesystemSettings,
67}
68
69#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
70#[serde(deny_unknown_fields)]
71pub struct StorageFilesystemSettings {
72    pub poll_interval: Duration,
73    pub max_entries: NonZeroUsize,
74    pub event_capacity: NonZeroUsize,
75}
76
77#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
78#[serde(deny_unknown_fields)]
79pub struct StorageRequestSettings {
80    pub max_pending_queries: NonZeroUsize,
81    pub max_active_queries: NonZeroUsize,
82    pub max_pending_write_batches: NonZeroUsize,
83    pub max_active_file_operations: NonZeroUsize,
84    pub max_pending_write_bytes: NonZeroUsize,
85}
86
87#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
88#[serde(deny_unknown_fields)]
89pub struct StorageResponseSettings {
90    pub batch_rows: NonZeroUsize,
91    pub batch_bytes: NonZeroUsize,
92}
93
94#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
95#[serde(deny_unknown_fields)]
96pub struct StorageAccumulatorSettings {
97    // Maximum Retained Numeric Slot Bytes Per Accumulator; Zero Spills Immediately.
98    pub memory_bytes: usize,
99}
100
101/* === Enums === */
102
103/* === Traits === */
104
105/* === Implementations === */
106
107impl Default for StorageAccumulatorSettings {
108    fn default() -> Self {
109        Self { memory_bytes: 64 * 1024 }
110    }
111}
112
113impl StorageSettings {
114    pub fn validate(&self) -> crate::Result<()> {
115        if self.filesystem.poll_interval.is_zero() || self.filesystem.event_capacity.get() > tokio::sync::Semaphore::MAX_PERMITS {
116            return Err(crate::Error::InvalidInput {
117                field: "storage filesystem",
118                reason: "poll interval must be positive and event capacity bounded",
119            });
120        }
121
122        if self.request.max_active_queries.get() > tokio::sync::Semaphore::MAX_PERMITS
123            || self.request.max_pending_queries.get() > tokio::sync::Semaphore::MAX_PERMITS
124        {
125            return Err(crate::Error::InvalidInput {
126                field: "storage query admission",
127                reason: "query counts exceed semaphore capacity",
128            });
129        }
130        Ok(())
131    }
132
133    // Largest Atomic Batch: The Private Ceiling Within The Pending Write Budget.
134    pub fn max_batch_bytes(&self) -> NonZeroUsize {
135        self.request.max_pending_write_bytes.min(NonZeroUsize::new(MAX_BATCH_BYTES).unwrap())
136    }
137
138    pub(super) fn target_batch_bytes(&self) -> usize {
139        TARGET_BATCH_BYTES.min(self.max_batch_bytes().get())
140    }
141
142    pub(super) fn scan_bytes(&self) -> NonZeroUsize {
143        NonZeroUsize::new(SCAN_BYTES).unwrap()
144    }
145}
146
147/* === Public Functions === */
148
149/* === Private Functions === */
150
151/* === Tests === */
152
153#[cfg(test)]
154mod tests {
155    use super::{StorageRequestSettings, StorageResponseSettings, StorageSettings};
156    use std::num::NonZeroUsize;
157
158    fn settings() -> StorageSettings {
159        StorageSettings {
160            filesystem: crate::config::StorageFilesystemSettings {
161                poll_interval: std::time::Duration::from_millis(100),
162                max_entries: std::num::NonZeroUsize::new(100_000).unwrap(),
163                event_capacity: std::num::NonZeroUsize::new(1024).unwrap(),
164            },
165
166            request: StorageRequestSettings {
167                max_pending_queries: NonZeroUsize::new(64).unwrap(),
168                max_active_queries: NonZeroUsize::new(4).unwrap(),
169                max_pending_write_batches: NonZeroUsize::new(256).unwrap(),
170                max_active_file_operations: NonZeroUsize::MIN,
171                max_pending_write_bytes: NonZeroUsize::new(16 * 1024 * 1024).unwrap(),
172            },
173            response: StorageResponseSettings {
174                batch_rows: NonZeroUsize::new(64).unwrap(),
175                batch_bytes: NonZeroUsize::new(64 * 1024).unwrap(),
176            },
177        }
178    }
179
180    #[test]
181    fn query_counts_bounded() {
182        let mut settings = settings();
183        settings.request.max_active_queries = NonZeroUsize::new(tokio::sync::Semaphore::MAX_PERMITS + 1).unwrap();
184        assert!(settings.validate().is_err());
185
186        settings.request.max_active_queries = NonZeroUsize::MIN;
187        settings.request.max_pending_queries = NonZeroUsize::MIN;
188        settings.validate().unwrap();
189    }
190}
harmos_storage/config.rs

harmos_storage/
config.rs
