hashtree_lmdb/pool/
model.rs

1use hashtree_core::store::StoreError;
2use serde::{Deserialize, Serialize};
3use std::fmt;
4use std::path::PathBuf;
5use std::str::FromStr;
6use std::time::Duration;
7use uuid::Uuid;
8
9pub(super) const DEFAULT_CATALOG_MAP_SIZE_BYTES: u64 = 16 * 1024 * 1024 * 1024;
10pub(super) const MIN_MEMBER_MAP_SIZE_BYTES: u64 = 16 * 1024 * 1024;
11
12/// Stable identity for one storage member in a local blob pool.
13#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
14pub struct PoolMemberId(pub(super) [u8; 16]);
15
16impl PoolMemberId {
17    pub fn new() -> Self {
18        Self(*Uuid::new_v4().as_bytes())
19    }
20
21    pub fn as_bytes(&self) -> &[u8; 16] {
22        &self.0
23    }
24}
25
26impl Default for PoolMemberId {
27    fn default() -> Self {
28        Self::new()
29    }
30}
31
32impl fmt::Debug for PoolMemberId {
33    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
34        fmt::Display::fmt(self, formatter)
35    }
36}
37
38impl fmt::Display for PoolMemberId {
39    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
40        Uuid::from_bytes(self.0).fmt(formatter)
41    }
42}
43
44impl FromStr for PoolMemberId {
45    type Err = StoreError;
46
47    fn from_str(value: &str) -> Result<Self, Self::Err> {
48        let id = Uuid::parse_str(value)
49            .map_err(|error| StoreError::Other(format!("invalid pool member id: {error}")))?;
50        Ok(Self(*id.as_bytes()))
51    }
52}
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
55#[serde(rename_all = "kebab-case")]
56pub enum PoolMemberState {
57    Active,
58    Draining,
59}
60
61/// Persistent configuration for one opaque LMDB storage member.
62#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
63pub struct PoolMemberConfig {
64    pub path: PathBuf,
65    pub capacity_bytes: u64,
66    pub map_size_bytes: u64,
67    pub external_blob_dir: Option<PathBuf>,
68    pub external_blob_min_bytes: Option<u64>,
69    pub external_blob_sync: bool,
70    pub external_pack_target_bytes: Option<u64>,
71    pub max_read_concurrency: u32,
72    pub max_write_concurrency: u32,
73}
74
75impl PoolMemberConfig {
76    pub fn new(path: PathBuf, capacity_bytes: u64) -> Self {
77        Self {
78            path,
79            capacity_bytes,
80            map_size_bytes: capacity_bytes.max(MIN_MEMBER_MAP_SIZE_BYTES),
81            external_blob_dir: None,
82            external_blob_min_bytes: None,
83            external_blob_sync: true,
84            external_pack_target_bytes: None,
85            max_read_concurrency: 64,
86            max_write_concurrency: 16,
87        }
88    }
89
90    pub fn with_map_size_bytes(mut self, map_size_bytes: u64) -> Self {
91        self.map_size_bytes = map_size_bytes.max(MIN_MEMBER_MAP_SIZE_BYTES);
92        self
93    }
94
95    pub fn with_external_blobs(
96        mut self,
97        directory: PathBuf,
98        min_bytes: u64,
99        sync: bool,
100        pack_target_bytes: Option<u64>,
101    ) -> Self {
102        self.external_blob_dir = Some(directory);
103        self.external_blob_min_bytes = Some(min_bytes.max(1));
104        self.external_blob_sync = sync;
105        self.external_pack_target_bytes = pack_target_bytes.filter(|value| *value > 0);
106        self
107    }
108}
109
110#[derive(Debug, Clone)]
111pub struct PoolStoreConfig {
112    pub catalog_map_size_bytes: u64,
113    pub member_failure_cooldown: Duration,
114}
115
116impl Default for PoolStoreConfig {
117    fn default() -> Self {
118        Self {
119            catalog_map_size_bytes: DEFAULT_CATALOG_MAP_SIZE_BYTES,
120            member_failure_cooldown: Duration::from_secs(5),
121        }
122    }
123}
124
125#[derive(Debug, Clone, PartialEq, Eq)]
126pub struct PoolMemberStatus {
127    pub id: PoolMemberId,
128    pub state: PoolMemberState,
129    pub path: PathBuf,
130    pub capacity_bytes: u64,
131    pub map_size_bytes: u64,
132    pub external_blob_dir: Option<PathBuf>,
133    pub external_blob_min_bytes: Option<u64>,
134    pub external_blob_sync: bool,
135    pub external_pack_target_bytes: Option<u64>,
136    pub max_read_concurrency: u32,
137    pub max_write_concurrency: u32,
138    pub logical_bytes: u64,
139    pub located_blobs: u64,
140    pub available: bool,
141    pub last_error: Option<String>,
142}
143
144#[derive(Debug, Clone, Default, PartialEq, Eq)]
145pub struct PoolMaintenanceReport {
146    pub examined: usize,
147    pub moved: usize,
148    pub bytes_moved: u64,
149    pub failed: Vec<String>,
150}
151
152#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
153pub(super) struct MemberRecord {
154    pub(super) id: PoolMemberId,
155    pub(super) state: PoolMemberState,
156    pub(super) config: PoolMemberConfig,
157}
158
159#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
160pub(super) struct PoolManifest {
161    pub(super) version: u32,
162    pub(super) generation: u64,
163    pub(super) members: Vec<MemberRecord>,
164}
165
166impl Default for PoolManifest {
167    fn default() -> Self {
168        Self {
169            version: 1,
170            generation: 0,
171            members: Vec::new(),
172        }
173    }
174}
175
176#[derive(Debug, Clone, Copy, PartialEq, Eq)]
177pub(super) enum LocationRecord {
178    Pending {
179        member: PoolMemberId,
180        size: u64,
181    },
182    Stored {
183        member: PoolMemberId,
184        size: u64,
185    },
186    Moving {
187        source: PoolMemberId,
188        target: PoolMemberId,
189        size: u64,
190    },
191}
192
193impl LocationRecord {
194    pub(super) fn size(self) -> u64 {
195        match self {
196            Self::Pending { size, .. } | Self::Stored { size, .. } | Self::Moving { size, .. } => {
197                size
198            }
199        }
200    }
201
202    pub(super) fn preferred_member(self) -> PoolMemberId {
203        match self {
204            Self::Pending { member, .. } | Self::Stored { member, .. } => member,
205            Self::Moving { target, .. } => target,
206        }
207    }
208
209    pub(super) fn members(self) -> ([PoolMemberId; 2], usize) {
210        match self {
211            Self::Pending { member, .. } | Self::Stored { member, .. } => ([member, member], 1),
212            Self::Moving { source, target, .. } => ([source, target], 2),
213        }
214    }
215
216    pub(super) fn encode(self) -> Vec<u8> {
217        let mut bytes = Vec::with_capacity(41);
218        match self {
219            Self::Pending { member, size } => {
220                bytes.push(1);
221                bytes.extend_from_slice(member.as_bytes());
222                bytes.extend_from_slice(&size.to_be_bytes());
223            }
224            Self::Stored { member, size } => {
225                bytes.push(2);
226                bytes.extend_from_slice(member.as_bytes());
227                bytes.extend_from_slice(&size.to_be_bytes());
228            }
229            Self::Moving {
230                source,
231                target,
232                size,
233            } => {
234                bytes.push(3);
235                bytes.extend_from_slice(source.as_bytes());
236                bytes.extend_from_slice(target.as_bytes());
237                bytes.extend_from_slice(&size.to_be_bytes());
238            }
239        }
240        bytes
241    }
242
243    pub(super) fn decode(bytes: &[u8]) -> Result<Self, StoreError> {
244        let member = |slice: &[u8]| -> Result<PoolMemberId, StoreError> {
245            let bytes: [u8; 16] = slice
246                .try_into()
247                .map_err(|_| StoreError::Other("invalid pool member id bytes".into()))?;
248            Ok(PoolMemberId(bytes))
249        };
250        let size = |slice: &[u8]| -> Result<u64, StoreError> {
251            Ok(u64::from_be_bytes(slice.try_into().map_err(|_| {
252                StoreError::Other("invalid pool location size".into())
253            })?))
254        };
255        match bytes.first().copied() {
256            Some(1) if bytes.len() == 25 => Ok(Self::Pending {
257                member: member(&bytes[1..17])?,
258                size: size(&bytes[17..25])?,
259            }),
260            Some(2) if bytes.len() == 25 => Ok(Self::Stored {
261                member: member(&bytes[1..17])?,
262                size: size(&bytes[17..25])?,
263            }),
264            Some(3) if bytes.len() == 41 => Ok(Self::Moving {
265                source: member(&bytes[1..17])?,
266                target: member(&bytes[17..33])?,
267                size: size(&bytes[33..41])?,
268            }),
269            _ => Err(StoreError::Other("invalid pool location record".into())),
270        }
271    }
272}
hashtree_lmdb/pool/model.rs

hashtree_lmdb/pool/
model.rs
