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/* === Enums === */
95
96/* === Traits === */
97
98/* === Implementations === */
99
100impl StorageSettings {
101    pub fn validate(&self) -> crate::Result<()> {
102        if self.filesystem.poll_interval.is_zero() || self.filesystem.event_capacity.get() > tokio::sync::Semaphore::MAX_PERMITS {
103            return Err(crate::Error::InvalidInput {
104                field: "storage filesystem",
105                reason: "poll interval must be positive and event capacity bounded",
106            });
107        }
108
109        if self.request.max_active_queries.get() > tokio::sync::Semaphore::MAX_PERMITS
110            || self.request.max_pending_queries.get() > tokio::sync::Semaphore::MAX_PERMITS
111        {
112            return Err(crate::Error::InvalidInput {
113                field: "storage query admission",
114                reason: "query counts exceed semaphore capacity",
115            });
116        }
117        Ok(())
118    }
119
120    // Largest Atomic Batch: The Private Ceiling Within The Pending Write Budget.
121    pub fn max_batch_bytes(&self) -> NonZeroUsize {
122        self.request.max_pending_write_bytes.min(NonZeroUsize::new(MAX_BATCH_BYTES).unwrap())
123    }
124
125    pub(super) fn target_batch_bytes(&self) -> usize {
126        TARGET_BATCH_BYTES.min(self.max_batch_bytes().get())
127    }
128
129    pub(super) fn scan_bytes(&self) -> NonZeroUsize {
130        NonZeroUsize::new(SCAN_BYTES).unwrap()
131    }
132}
133
134/* === Public Functions === */
135
136/* === Private Functions === */
137
138/* === Tests === */
139
140#[cfg(test)]
141mod tests {
142    use super::{StorageRequestSettings, StorageResponseSettings, StorageSettings};
143    use std::num::NonZeroUsize;
144
145    fn settings() -> StorageSettings {
146        StorageSettings {
147            filesystem: crate::config::StorageFilesystemSettings {
148                poll_interval: std::time::Duration::from_millis(100),
149                max_entries: std::num::NonZeroUsize::new(100_000).unwrap(),
150                event_capacity: std::num::NonZeroUsize::new(1024).unwrap(),
151            },
152
153            request: StorageRequestSettings {
154                max_pending_queries: NonZeroUsize::new(64).unwrap(),
155                max_active_queries: NonZeroUsize::new(4).unwrap(),
156                max_pending_write_batches: NonZeroUsize::new(256).unwrap(),
157                max_active_file_operations: NonZeroUsize::MIN,
158                max_pending_write_bytes: NonZeroUsize::new(16 * 1024 * 1024).unwrap(),
159            },
160            response: StorageResponseSettings {
161                batch_rows: NonZeroUsize::new(64).unwrap(),
162                batch_bytes: NonZeroUsize::new(64 * 1024).unwrap(),
163            },
164        }
165    }
166
167    #[test]
168    fn query_counts_bounded() {
169        let mut settings = settings();
170        settings.request.max_active_queries = NonZeroUsize::new(tokio::sync::Semaphore::MAX_PERMITS + 1).unwrap();
171        assert!(settings.validate().is_err());
172
173        settings.request.max_active_queries = NonZeroUsize::MIN;
174        settings.request.max_pending_queries = NonZeroUsize::MIN;
175        settings.validate().unwrap();
176    }
177}
harmos_storage/config.rs

harmos_storage/
config.rs
