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 = 128 * 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 MAX_RETRIES: usize = 3;
39pub(super) const RETRY_BASE: Duration = Duration::from_millis(10);
40pub(super) const RETRY_CAP: Duration = Duration::from_secs(1);
41
42// Producer Sessions And Idempotent Mutation Receipts.
43pub(super) const MAX_FILES: NonZeroUsize = NonZeroUsize::new(256).unwrap();
44pub(super) const FILE_SESSION_TIMEOUT: Duration = Duration::from_secs(60);
45pub(super) const MAX_MUTATIONS: NonZeroUsize = NonZeroUsize::new(1024).unwrap();
46pub(super) const MUTATION_RETENTION: Duration = Duration::from_secs(300);
47// The Receipt Ledger Takes This Share Of The Read Budget That Checkpoints Copy It Into.
48const MUTATION_BUDGET_SHARE: usize = 8;
49
50// Internal Scan Pages; Response Sizing Belongs To StorageResponseSettings.
51pub(super) const SCAN_ROWS: NonZeroUsize = NonZeroUsize::new(1024).unwrap();
52pub(super) const SCAN_BYTES: usize = 1024 * 1024;
53
54/* === Macros === */
55
56/* === Type Aliases === */
57
58/* === Structs === */
59
60#[derive(Clone, Debug, Serialize, Deserialize)]
61#[serde(deny_unknown_fields)]
62pub struct StorageSettings {
63    pub budget: StorageBudget,
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 StorageBudget {
80    pub read_memory_bytes: NonZeroUsize,
81}
82
83#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
84#[serde(deny_unknown_fields)]
85pub struct StorageRequestSettings {
86    pub max_pending_queries: NonZeroUsize,
87    pub max_active_queries: NonZeroUsize,
88    pub max_pending_write_batches: NonZeroUsize,
89    pub max_active_file_operations: NonZeroUsize,
90    pub max_pending_write_bytes: NonZeroUsize,
91}
92
93#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
94#[serde(deny_unknown_fields)]
95pub struct StorageResponseSettings {
96    pub batch_rows: NonZeroUsize,
97    pub batch_bytes: NonZeroUsize,
98}
99
100/* === Enums === */
101
102/* === Traits === */
103
104/* === Implementations === */
105
106impl StorageSettings {
107    pub fn validate(&self) -> crate::Result<()> {
108        if self.filesystem.poll_interval.is_zero() || self.filesystem.event_capacity.get() > tokio::sync::Semaphore::MAX_PERMITS {
109            return Err(crate::Error::InvalidInput {
110                field: "storage filesystem",
111                reason: "poll interval must be positive and event capacity bounded",
112            });
113        }
114
115        if self.request.max_active_queries.get() > tokio::sync::Semaphore::MAX_PERMITS
116            || self.request.max_pending_queries.get() > tokio::sync::Semaphore::MAX_PERMITS
117        {
118            return Err(crate::Error::InvalidInput { field: "storage query admission", reason: "query counts exceed semaphore capacity" });
119        }
120        if self.response.batch_bytes.get() > self.budget.read_memory_bytes.get() / 2 {
121            return Err(crate::Error::InvalidInput {
122                field: "storage response",
123                reason: "response batches cannot exceed half the shared read budget",
124            });
125        }
126        Ok(())
127    }
128
129    // Largest Atomic Batch: The Private Ceiling Within The Pending Write Budget.
130    pub fn max_batch_bytes(&self) -> NonZeroUsize {
131        self.request.max_pending_write_bytes.min(NonZeroUsize::new(MAX_BATCH_BYTES).unwrap())
132    }
133
134    pub(super) fn target_batch_bytes(&self) -> usize {
135        TARGET_BATCH_BYTES.min(self.max_batch_bytes().get())
136    }
137
138    // Receipt Ledger Bound: A Fixed Share Of The Shared Read Budget.
139    pub(super) fn mutation_bytes(&self) -> NonZeroUsize {
140        NonZeroUsize::new(self.budget.read_memory_bytes.get() / MUTATION_BUDGET_SHARE).unwrap_or(NonZeroUsize::MIN)
141    }
142
143    // Scan Page Bytes, Small Enough That Several Active Queries Share The Budget.
144    pub(super) fn scan_bytes(&self) -> NonZeroUsize {
145        let shared = self.budget.read_memory_bytes.get() / self.request.max_active_queries.get().saturating_mul(4).max(1);
146        NonZeroUsize::new(SCAN_BYTES.min(shared).max(1)).unwrap()
147    }
148}
149
150/* === Public Functions === */
151
152/* === Private Functions === */
153
154/* === Tests === */
155
156#[cfg(test)]
157mod tests {
158    use super::{StorageBudget, StorageRequestSettings, StorageResponseSettings, StorageSettings};
159    use std::num::NonZeroUsize;
160
161    fn settings() -> StorageSettings {
162        StorageSettings {
163            filesystem: crate::config::StorageFilesystemSettings {
164                poll_interval: std::time::Duration::from_millis(100),
165                max_entries: std::num::NonZeroUsize::new(100_000).unwrap(),
166                event_capacity: std::num::NonZeroUsize::new(1024).unwrap(),
167            },
168            budget: StorageBudget { read_memory_bytes: NonZeroUsize::new(16 * 1024 * 1024).unwrap() },
169            request: StorageRequestSettings {
170                max_pending_queries: NonZeroUsize::new(64).unwrap(),
171                max_active_queries: NonZeroUsize::new(4).unwrap(),
172                max_pending_write_batches: NonZeroUsize::new(256).unwrap(),
173                max_active_file_operations: NonZeroUsize::MIN,
174                max_pending_write_bytes: NonZeroUsize::new(16 * 1024 * 1024).unwrap(),
175            },
176            response: StorageResponseSettings {
177                batch_rows: NonZeroUsize::new(64).unwrap(),
178                batch_bytes: NonZeroUsize::new(64 * 1024).unwrap(),
179            },
180        }
181    }
182
183    #[test]
184    fn response_fits_budget() {
185        let mut settings = settings();
186        settings.validate().unwrap();
187
188        settings.response.batch_bytes = NonZeroUsize::new(settings.budget.read_memory_bytes.get() + 1).unwrap();
189        assert!(settings.validate().is_err());
190        settings.response.batch_bytes = settings.budget.read_memory_bytes;
191        assert!(settings.validate().is_err());
192
193        settings.response.batch_bytes = NonZeroUsize::new(settings.budget.read_memory_bytes.get() / 2).unwrap();
194        settings.validate().unwrap();
195    }
196
197    #[test]
198    fn query_counts_bounded() {
199        let mut settings = settings();
200        settings.request.max_active_queries = NonZeroUsize::new(tokio::sync::Semaphore::MAX_PERMITS + 1).unwrap();
201        assert!(settings.validate().is_err());
202
203        settings.request.max_active_queries = NonZeroUsize::MIN;
204        settings.request.max_pending_queries = NonZeroUsize::MIN;
205        settings.validate().unwrap();
206    }
207}
harmos_storage/config.rs

harmos_storage/
config.rs
