hashtree_lmdb/pool/
model.rs

1use hashtree_core::store::StoreError;
2use hashtree_core::types::Hash;
3use heed::PinnedLmdbIdentity;
4use serde::{Deserialize, Serialize};
5use std::fmt;
6use std::path::PathBuf;
7use std::str::FromStr;
8use std::time::Duration;
9use uuid::Uuid;
10
11pub(super) const DEFAULT_CATALOG_MAP_SIZE_BYTES: u64 =
12    super::map_size::default_catalog_map_size_bytes(cfg!(target_os = "ios"));
13pub(super) const MIN_MEMBER_MAP_SIZE_BYTES: u64 = 16 * 1024 * 1024;
14
15/// Stable identity for one storage member in a local blob pool.
16#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
17pub struct PoolMemberId(pub(super) [u8; 16]);
18
19impl PoolMemberId {
20    pub fn new() -> Self {
21        Self(*Uuid::new_v4().as_bytes())
22    }
23
24    pub fn as_bytes(&self) -> &[u8; 16] {
25        &self.0
26    }
27}
28
29impl Default for PoolMemberId {
30    fn default() -> Self {
31        Self::new()
32    }
33}
34
35impl fmt::Debug for PoolMemberId {
36    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
37        fmt::Display::fmt(self, formatter)
38    }
39}
40
41impl fmt::Display for PoolMemberId {
42    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
43        Uuid::from_bytes(self.0).fmt(formatter)
44    }
45}
46
47impl FromStr for PoolMemberId {
48    type Err = StoreError;
49
50    fn from_str(value: &str) -> Result<Self, Self::Err> {
51        let id = Uuid::parse_str(value)
52            .map_err(|error| StoreError::Other(format!("invalid pool member id: {error}")))?;
53        Ok(Self(*id.as_bytes()))
54    }
55}
56
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
58#[serde(rename_all = "kebab-case")]
59pub enum PoolMemberState {
60    Active,
61    Draining,
62}
63
64/// Persistent configuration for one opaque LMDB storage member.
65#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
66pub struct PoolMemberConfig {
67    pub path: PathBuf,
68    pub capacity_bytes: u64,
69    pub map_size_bytes: u64,
70    pub external_blob_dir: Option<PathBuf>,
71    pub external_blob_min_bytes: Option<u64>,
72    pub external_blob_sync: bool,
73    pub external_pack_target_bytes: Option<u64>,
74    pub max_read_concurrency: u32,
75    pub max_write_concurrency: u32,
76    #[serde(default = "default_temperature_low_watermark_percent")]
77    pub temperature_low_watermark_percent: u8,
78    #[serde(default = "default_temperature_high_watermark_percent")]
79    pub temperature_high_watermark_percent: u8,
80}
81
82impl PoolMemberConfig {
83    pub fn new(path: PathBuf, capacity_bytes: u64) -> Self {
84        Self {
85            path,
86            capacity_bytes,
87            map_size_bytes: capacity_bytes.max(MIN_MEMBER_MAP_SIZE_BYTES),
88            external_blob_dir: None,
89            external_blob_min_bytes: None,
90            external_blob_sync: true,
91            external_pack_target_bytes: None,
92            max_read_concurrency: 64,
93            max_write_concurrency: 16,
94            temperature_low_watermark_percent: default_temperature_low_watermark_percent(),
95            temperature_high_watermark_percent: default_temperature_high_watermark_percent(),
96        }
97    }
98
99    pub fn with_map_size_bytes(mut self, map_size_bytes: u64) -> Self {
100        self.map_size_bytes = map_size_bytes.max(MIN_MEMBER_MAP_SIZE_BYTES);
101        self
102    }
103
104    pub fn with_external_blobs(
105        mut self,
106        directory: PathBuf,
107        min_bytes: u64,
108        sync: bool,
109        pack_target_bytes: Option<u64>,
110    ) -> Self {
111        self.external_blob_dir = Some(directory);
112        self.external_blob_min_bytes = Some(min_bytes.max(1));
113        self.external_blob_sync = sync;
114        self.external_pack_target_bytes = pack_target_bytes.filter(|value| *value > 0);
115        self
116    }
117
118    pub fn with_temperature_watermarks(mut self, low_percent: u8, high_percent: u8) -> Self {
119        self.temperature_low_watermark_percent = low_percent;
120        self.temperature_high_watermark_percent = high_percent;
121        self
122    }
123}
124
125fn default_temperature_low_watermark_percent() -> u8 {
126    70
127}
128
129fn default_temperature_high_watermark_percent() -> u8 {
130    85
131}
132
133#[derive(Debug, Clone)]
134pub struct PoolTemperatureConfig {
135    pub enabled: bool,
136    pub interval: Duration,
137    pub read_sample_rate: u32,
138    pub heat_half_life: Duration,
139    pub access_flush_batch: usize,
140    pub scan_items_per_cycle: usize,
141    pub candidate_capacity: usize,
142    pub max_moves_per_cycle: usize,
143    pub max_bytes_per_cycle: u64,
144    pub max_concurrent_moves: usize,
145    pub minimum_residence: Duration,
146    pub promotion_hysteresis_percent: u8,
147    pub foreground_load_percent: u8,
148    pub lease_duration: Duration,
149    pub copy_chunk_bytes: usize,
150}
151
152impl Default for PoolTemperatureConfig {
153    fn default() -> Self {
154        Self {
155            enabled: true,
156            interval: Duration::from_secs(30),
157            read_sample_rate: 64,
158            heat_half_life: Duration::from_secs(6 * 60 * 60),
159            access_flush_batch: 512,
160            scan_items_per_cycle: 2_048,
161            candidate_capacity: 4_096,
162            max_moves_per_cycle: 8,
163            max_bytes_per_cycle: 1024 * 1024 * 1024,
164            max_concurrent_moves: 2,
165            minimum_residence: Duration::from_secs(6 * 60 * 60),
166            promotion_hysteresis_percent: 20,
167            foreground_load_percent: 25,
168            lease_duration: Duration::from_secs(120),
169            copy_chunk_bytes: 1024 * 1024,
170        }
171    }
172}
173
174impl PoolTemperatureConfig {
175    pub(super) fn validate(&self) -> Result<(), StoreError> {
176        if !self.enabled {
177            return Ok(());
178        }
179        if self.interval.is_zero()
180            || self.read_sample_rate == 0
181            || self.heat_half_life.is_zero()
182            || self.access_flush_batch == 0
183            || self.scan_items_per_cycle == 0
184            || self.candidate_capacity == 0
185            || self.max_moves_per_cycle == 0
186            || self.max_bytes_per_cycle == 0
187            || self.max_concurrent_moves == 0
188            || self.lease_duration.is_zero()
189            || self.copy_chunk_bytes == 0
190        {
191            return Err(StoreError::Other(
192                "enabled pool temperature budgets and intervals must be non-zero".into(),
193            ));
194        }
195        if self.foreground_load_percent == 0
196            || self.foreground_load_percent > 100
197            || self.promotion_hysteresis_percent > 100
198        {
199            return Err(StoreError::Other(
200                "pool temperature percentages must be within their documented 1..=100 bounds"
201                    .into(),
202            ));
203        }
204        Ok(())
205    }
206}
207
208#[derive(Debug, Clone)]
209pub struct PoolStoreConfig {
210    /// Runtime-only opt-in admission; never persisted in the pool manifest.
211    pub physical_space: Option<crate::PhysicalSpaceGuard>,
212    pub catalog_map_size_bytes: u64,
213    pub member_failure_cooldown: Duration,
214    pub temperature: PoolTemperatureConfig,
215    /// Exact configured-to-runtime member path bindings for a controlled
216    /// process. When non-empty, every manifest member must match one binding
217    /// and member opens use only the runtime paths.
218    pub member_runtime_paths: Vec<PoolMemberRuntimePaths>,
219    /// Exact target catalog data/lock identities captured before a controlled
220    /// process was acknowledged.
221    pub catalog_lmdb_identity: Option<PinnedLmdbIdentity>,
222    /// SHA-256 of the exact stored manifest bytes authorized for a controlled
223    /// process, including generation, ordering, state, and all member config.
224    pub expected_manifest_sha256: Option<Hash>,
225}
226
227/// One durable, non-expiring guard that keeps PoolStore writable while
228/// rejecting logical deletion. The exact catalog record identity is required
229/// to release it, preventing a stale operator command from removing a newer
230/// guard.
231#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
232pub struct PoolDeleteProtectionStatus {
233    pub lease_id: Hash,
234    pub reason: String,
235    pub acquired_at_unix_secs: u64,
236    pub record_sha256: Hash,
237}
238
239#[derive(Debug, Clone, PartialEq, Eq)]
240pub struct PoolDeleteProtectionChange {
241    pub changed: bool,
242    pub status: PoolDeleteProtectionStatus,
243}
244
245impl Default for PoolStoreConfig {
246    fn default() -> Self {
247        Self {
248            physical_space: None,
249            catalog_map_size_bytes: DEFAULT_CATALOG_MAP_SIZE_BYTES,
250            member_failure_cooldown: Duration::from_secs(5),
251            temperature: PoolTemperatureConfig::default(),
252            member_runtime_paths: Vec::new(),
253            catalog_lmdb_identity: None,
254            expected_manifest_sha256: None,
255        }
256    }
257}
258
259/// One exact catalog record authorized for stale-`Pending` crash recovery.
260///
261/// Cleanup callers must build this list from a strict read-only audit while
262/// every legacy Pool writer is fenced. The cleanup operation compares all
263/// three fields again in one catalog transaction and never treats a `Stored`
264/// or `Moving` record as equivalent.
265#[derive(Debug, Clone, Copy, PartialEq, Eq)]
266pub struct PoolStalePending {
267    pub hash: Hash,
268    pub member: PoolMemberId,
269    pub size: u64,
270}
271
272/// Result of one exact stale-`Pending` recovery transaction.
273#[derive(Debug, Clone, Copy, PartialEq, Eq)]
274pub struct PoolStalePendingCleanupReport {
275    pub requested: usize,
276    pub declared_bytes: u64,
277    /// True only when the exact batch had already committed as wholly absent.
278    pub already_cleaned: bool,
279}
280
281#[derive(Debug, Clone, PartialEq, Eq)]
282pub struct PoolMemberRuntimePaths {
283    pub id: PoolMemberId,
284    pub configured_path: PathBuf,
285    pub runtime_path: PathBuf,
286    pub configured_external_path: Option<PathBuf>,
287    pub runtime_external_path: Option<PathBuf>,
288    pub lmdb_identity: PinnedLmdbIdentity,
289}
290
291#[derive(Debug, Clone, PartialEq, Eq)]
292pub struct PoolMemberStatus {
293    pub id: PoolMemberId,
294    pub state: PoolMemberState,
295    pub path: PathBuf,
296    pub capacity_bytes: u64,
297    pub map_size_bytes: u64,
298    pub external_blob_dir: Option<PathBuf>,
299    pub external_blob_min_bytes: Option<u64>,
300    pub external_blob_sync: bool,
301    pub external_pack_target_bytes: Option<u64>,
302    pub max_read_concurrency: u32,
303    pub max_write_concurrency: u32,
304    pub temperature_low_watermark_percent: u8,
305    pub temperature_high_watermark_percent: u8,
306    pub logical_bytes: u64,
307    pub located_blobs: u64,
308    pub available: bool,
309    pub last_error: Option<String>,
310}
311
312#[derive(Debug, Clone, Default, PartialEq, Eq)]
313pub struct PoolTemperatureReport {
314    pub sampled_accesses_flushed: usize,
315    pub scanned: usize,
316    pub candidates: usize,
317    pub moved: usize,
318    pub attempted_moves: usize,
319    pub peak_concurrent_moves: usize,
320    pub bytes_moved: u64,
321    pub resumed: usize,
322    pub throttled: bool,
323    pub lease_acquired: bool,
324    pub failed: Vec<String>,
325}
326
327#[derive(Debug, Clone, Default, PartialEq, Eq)]
328pub struct PoolMaintenanceReport {
329    pub examined: usize,
330    pub moved: usize,
331    pub bytes_moved: u64,
332    pub failed: Vec<String>,
333}
334
335#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
336pub(super) struct MemberRecord {
337    pub(super) id: PoolMemberId,
338    pub(super) state: PoolMemberState,
339    pub(super) config: PoolMemberConfig,
340}
341
342#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
343pub(super) struct PoolManifest {
344    pub(super) version: u32,
345    pub(super) generation: u64,
346    pub(super) members: Vec<MemberRecord>,
347}
348
349impl Default for PoolManifest {
350    fn default() -> Self {
351        Self {
352            version: 1,
353            generation: 0,
354            members: Vec::new(),
355        }
356    }
357}
358
359#[derive(Debug, Clone, Copy, PartialEq, Eq)]
360pub(super) enum LocationRecord {
361    Pending {
362        member: PoolMemberId,
363        size: u64,
364    },
365    Stored {
366        member: PoolMemberId,
367        size: u64,
368    },
369    Moving {
370        source: PoolMemberId,
371        target: PoolMemberId,
372        size: u64,
373    },
374}
375
376impl LocationRecord {
377    pub(super) fn size(self) -> u64 {
378        match self {
379            Self::Pending { size, .. } | Self::Stored { size, .. } | Self::Moving { size, .. } => {
380                size
381            }
382        }
383    }
384
385    pub(super) fn preferred_member(self) -> PoolMemberId {
386        match self {
387            Self::Pending { member, .. } | Self::Stored { member, .. } => member,
388            Self::Moving { target, .. } => target,
389        }
390    }
391
392    pub(super) fn members(self) -> ([PoolMemberId; 2], usize) {
393        match self {
394            Self::Pending { member, .. } | Self::Stored { member, .. } => ([member, member], 1),
395            Self::Moving { source, target, .. } => ([source, target], 2),
396        }
397    }
398
399    pub(super) fn encode(self) -> Vec<u8> {
400        let mut bytes = Vec::with_capacity(41);
401        match self {
402            Self::Pending { member, size } => {
403                bytes.push(1);
404                bytes.extend_from_slice(member.as_bytes());
405                bytes.extend_from_slice(&size.to_be_bytes());
406            }
407            Self::Stored { member, size } => {
408                bytes.push(2);
409                bytes.extend_from_slice(member.as_bytes());
410                bytes.extend_from_slice(&size.to_be_bytes());
411            }
412            Self::Moving {
413                source,
414                target,
415                size,
416            } => {
417                bytes.push(3);
418                bytes.extend_from_slice(source.as_bytes());
419                bytes.extend_from_slice(target.as_bytes());
420                bytes.extend_from_slice(&size.to_be_bytes());
421            }
422        }
423        bytes
424    }
425
426    pub(super) fn decode(bytes: &[u8]) -> Result<Self, StoreError> {
427        let member = |slice: &[u8]| -> Result<PoolMemberId, StoreError> {
428            let bytes: [u8; 16] = slice
429                .try_into()
430                .map_err(|_| StoreError::Other("invalid pool member id bytes".into()))?;
431            Ok(PoolMemberId(bytes))
432        };
433        let size = |slice: &[u8]| -> Result<u64, StoreError> {
434            Ok(u64::from_be_bytes(slice.try_into().map_err(|_| {
435                StoreError::Other("invalid pool location size".into())
436            })?))
437        };
438        match bytes.first().copied() {
439            Some(1) if bytes.len() == 25 => Ok(Self::Pending {
440                member: member(&bytes[1..17])?,
441                size: size(&bytes[17..25])?,
442            }),
443            Some(2) if bytes.len() == 25 => Ok(Self::Stored {
444                member: member(&bytes[1..17])?,
445                size: size(&bytes[17..25])?,
446            }),
447            Some(3) if bytes.len() == 41 => Ok(Self::Moving {
448                source: member(&bytes[1..17])?,
449                target: member(&bytes[17..33])?,
450                size: size(&bytes[33..41])?,
451            }),
452            _ => Err(StoreError::Other("invalid pool location record".into())),
453        }
454    }
455}
hashtree_lmdb/pool/model.rs

hashtree_lmdb/pool/
model.rs
