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 WRITE_BATCH_ENTRIES: NonZeroUsize = NonZeroUsize::new(4096).unwrap();
34pub(super) const OBJECT_PRUNE_ENTRIES: NonZeroUsize = NonZeroUsize::new(256).unwrap();
35pub(super) const SNAPSHOT_BYTES: usize = 1024 * 1024;
36
37// Backend I/O And Transient Persistence Retry.
38pub(super) const IO_TIMEOUT: Duration = Duration::from_secs(30);
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// The Receipt Ledger Takes This Share Of The Read Budget That Checkpoints Copy It Into.
49const MUTATION_BUDGET_SHARE: usize = 8;
50
51// Internal Scan Pages; Response Sizing Belongs To StorageResponseSettings.
52pub(super) const SCAN_ROWS: NonZeroUsize = NonZeroUsize::new(1024).unwrap();
53pub(super) const SCAN_BYTES: usize = 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 budget: StorageBudget,
65    pub request: StorageRequestSettings,
66    pub response: StorageResponseSettings,
67}
68
69#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
70#[serde(deny_unknown_fields)]
71pub struct StorageBudget {
72    pub read_memory_bytes: NonZeroUsize,
73}
74
75#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
76#[serde(deny_unknown_fields)]
77pub struct StorageRequestSettings {
78    pub max_pending_queries: NonZeroUsize,
79    pub max_active_queries: NonZeroUsize,
80    pub max_pending_write_batches: NonZeroUsize,
81    pub max_pending_write_bytes: NonZeroUsize,
82}
83
84#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
85#[serde(deny_unknown_fields)]
86pub struct StorageResponseSettings {
87    pub batch_rows: NonZeroUsize,
88    pub batch_bytes: NonZeroUsize,
89}
90
91/* === Enums === */
92
93/* === Traits === */
94
95/* === Implementations === */
96
97impl StorageSettings {
98    pub fn validate(&self) -> crate::Result<()> {
99        if self.request.max_active_queries.get() > tokio::sync::Semaphore::MAX_PERMITS
100            || self.request.max_pending_queries.get() > tokio::sync::Semaphore::MAX_PERMITS
101        {
102            return Err(crate::Error::InvalidInput { field: "storage query admission", reason: "query counts exceed semaphore capacity" });
103        }
104        if self.response.batch_bytes.get() > self.budget.read_memory_bytes.get() / 2 {
105            return Err(crate::Error::InvalidInput {
106                field: "storage response",
107                reason: "response batches cannot exceed half the shared read budget",
108            });
109        }
110        Ok(())
111    }
112
113    // Largest Atomic Batch: The Private Ceiling Within The Pending Write Budget.
114    pub fn max_batch_bytes(&self) -> NonZeroUsize {
115        self.request.max_pending_write_bytes.min(NonZeroUsize::new(MAX_BATCH_BYTES).unwrap())
116    }
117
118    pub(super) fn target_batch_bytes(&self) -> usize {
119        TARGET_BATCH_BYTES.min(self.max_batch_bytes().get())
120    }
121
122    // Receipt Ledger Bound: A Fixed Share Of The Shared Read Budget.
123    pub(super) fn mutation_bytes(&self) -> NonZeroUsize {
124        NonZeroUsize::new(self.budget.read_memory_bytes.get() / MUTATION_BUDGET_SHARE).unwrap_or(NonZeroUsize::MIN)
125    }
126
127    // Scan Page Bytes, Small Enough That Several Active Queries Share The Budget.
128    pub(super) fn scan_bytes(&self) -> NonZeroUsize {
129        let shared = self.budget.read_memory_bytes.get() / self.request.max_active_queries.get().saturating_mul(4).max(1);
130        NonZeroUsize::new(SCAN_BYTES.min(shared).max(1)).unwrap()
131    }
132}
133
134/* === Public Functions === */
135
136/* === Private Functions === */
137
138/* === Tests === */
139
140#[cfg(test)]
141mod tests {
142    use super::{StorageBudget, StorageRequestSettings, StorageResponseSettings, StorageSettings};
143    use std::num::NonZeroUsize;
144
145    fn settings() -> StorageSettings {
146        StorageSettings {
147            budget: StorageBudget { read_memory_bytes: NonZeroUsize::new(16 * 1024 * 1024).unwrap() },
148            request: StorageRequestSettings {
149                max_pending_queries: NonZeroUsize::new(64).unwrap(),
150                max_active_queries: NonZeroUsize::new(4).unwrap(),
151                max_pending_write_batches: NonZeroUsize::new(256).unwrap(),
152                max_pending_write_bytes: NonZeroUsize::new(16 * 1024 * 1024).unwrap(),
153            },
154            response: StorageResponseSettings {
155                batch_rows: NonZeroUsize::new(64).unwrap(),
156                batch_bytes: NonZeroUsize::new(64 * 1024).unwrap(),
157            },
158        }
159    }
160
161    #[test]
162    fn response_fits_budget() {
163        let mut settings = settings();
164        settings.validate().unwrap();
165
166        settings.response.batch_bytes = NonZeroUsize::new(settings.budget.read_memory_bytes.get() + 1).unwrap();
167        assert!(settings.validate().is_err());
168        settings.response.batch_bytes = settings.budget.read_memory_bytes;
169        assert!(settings.validate().is_err());
170
171        settings.response.batch_bytes = NonZeroUsize::new(settings.budget.read_memory_bytes.get() / 2).unwrap();
172        settings.validate().unwrap();
173    }
174
175    #[test]
176    fn query_counts_bounded() {
177        let mut settings = settings();
178        settings.request.max_active_queries = NonZeroUsize::new(tokio::sync::Semaphore::MAX_PERMITS + 1).unwrap();
179        assert!(settings.validate().is_err());
180
181        settings.request.max_active_queries = NonZeroUsize::MIN;
182        settings.request.max_pending_queries = NonZeroUsize::MIN;
183        settings.validate().unwrap();
184    }
185}
harmos_storage/config.rs

harmos_storage/
config.rs
