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#[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#[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 #[serde(default = "default_temperature_low_watermark_percent")]
74 pub temperature_low_watermark_percent: u8,
75 #[serde(default = "default_temperature_high_watermark_percent")]
76 pub temperature_high_watermark_percent: u8,
77}
78
79impl PoolMemberConfig {
80 pub fn new(path: PathBuf, capacity_bytes: u64) -> Self {
81 Self {
82 path,
83 capacity_bytes,
84 map_size_bytes: capacity_bytes.max(MIN_MEMBER_MAP_SIZE_BYTES),
85 external_blob_dir: None,
86 external_blob_min_bytes: None,
87 external_blob_sync: true,
88 external_pack_target_bytes: None,
89 max_read_concurrency: 64,
90 max_write_concurrency: 16,
91 temperature_low_watermark_percent: default_temperature_low_watermark_percent(),
92 temperature_high_watermark_percent: default_temperature_high_watermark_percent(),
93 }
94 }
95
96 pub fn with_map_size_bytes(mut self, map_size_bytes: u64) -> Self {
97 self.map_size_bytes = map_size_bytes.max(MIN_MEMBER_MAP_SIZE_BYTES);
98 self
99 }
100
101 pub fn with_external_blobs(
102 mut self,
103 directory: PathBuf,
104 min_bytes: u64,
105 sync: bool,
106 pack_target_bytes: Option<u64>,
107 ) -> Self {
108 self.external_blob_dir = Some(directory);
109 self.external_blob_min_bytes = Some(min_bytes.max(1));
110 self.external_blob_sync = sync;
111 self.external_pack_target_bytes = pack_target_bytes.filter(|value| *value > 0);
112 self
113 }
114
115 pub fn with_temperature_watermarks(mut self, low_percent: u8, high_percent: u8) -> Self {
116 self.temperature_low_watermark_percent = low_percent;
117 self.temperature_high_watermark_percent = high_percent;
118 self
119 }
120}
121
122fn default_temperature_low_watermark_percent() -> u8 {
123 70
124}
125
126fn default_temperature_high_watermark_percent() -> u8 {
127 85
128}
129
130#[derive(Debug, Clone)]
131pub struct PoolTemperatureConfig {
132 pub enabled: bool,
133 pub interval: Duration,
134 pub read_sample_rate: u32,
135 pub heat_half_life: Duration,
136 pub access_flush_batch: usize,
137 pub scan_items_per_cycle: usize,
138 pub candidate_capacity: usize,
139 pub max_moves_per_cycle: usize,
140 pub max_bytes_per_cycle: u64,
141 pub max_concurrent_moves: usize,
142 pub minimum_residence: Duration,
143 pub promotion_hysteresis_percent: u8,
144 pub foreground_load_percent: u8,
145 pub lease_duration: Duration,
146 pub copy_chunk_bytes: usize,
147}
148
149impl Default for PoolTemperatureConfig {
150 fn default() -> Self {
151 Self {
152 enabled: true,
153 interval: Duration::from_secs(30),
154 read_sample_rate: 64,
155 heat_half_life: Duration::from_secs(6 * 60 * 60),
156 access_flush_batch: 512,
157 scan_items_per_cycle: 2_048,
158 candidate_capacity: 4_096,
159 max_moves_per_cycle: 8,
160 max_bytes_per_cycle: 1024 * 1024 * 1024,
161 max_concurrent_moves: 2,
162 minimum_residence: Duration::from_secs(6 * 60 * 60),
163 promotion_hysteresis_percent: 20,
164 foreground_load_percent: 25,
165 lease_duration: Duration::from_secs(120),
166 copy_chunk_bytes: 1024 * 1024,
167 }
168 }
169}
170
171impl PoolTemperatureConfig {
172 pub(super) fn validate(&self) -> Result<(), StoreError> {
173 if !self.enabled {
174 return Ok(());
175 }
176 if self.interval.is_zero()
177 || self.read_sample_rate == 0
178 || self.heat_half_life.is_zero()
179 || self.access_flush_batch == 0
180 || self.scan_items_per_cycle == 0
181 || self.candidate_capacity == 0
182 || self.max_moves_per_cycle == 0
183 || self.max_bytes_per_cycle == 0
184 || self.max_concurrent_moves == 0
185 || self.lease_duration.is_zero()
186 || self.copy_chunk_bytes == 0
187 {
188 return Err(StoreError::Other(
189 "enabled pool temperature budgets and intervals must be non-zero".into(),
190 ));
191 }
192 if self.foreground_load_percent == 0
193 || self.foreground_load_percent > 100
194 || self.promotion_hysteresis_percent > 100
195 {
196 return Err(StoreError::Other(
197 "pool temperature percentages must be within their documented 1..=100 bounds"
198 .into(),
199 ));
200 }
201 Ok(())
202 }
203}
204
205#[derive(Debug, Clone)]
206pub struct PoolStoreConfig {
207 pub catalog_map_size_bytes: u64,
208 pub member_failure_cooldown: Duration,
209 pub temperature: PoolTemperatureConfig,
210}
211
212impl Default for PoolStoreConfig {
213 fn default() -> Self {
214 Self {
215 catalog_map_size_bytes: DEFAULT_CATALOG_MAP_SIZE_BYTES,
216 member_failure_cooldown: Duration::from_secs(5),
217 temperature: PoolTemperatureConfig::default(),
218 }
219 }
220}
221
222#[derive(Debug, Clone, PartialEq, Eq)]
223pub struct PoolMemberStatus {
224 pub id: PoolMemberId,
225 pub state: PoolMemberState,
226 pub path: PathBuf,
227 pub capacity_bytes: u64,
228 pub map_size_bytes: u64,
229 pub external_blob_dir: Option<PathBuf>,
230 pub external_blob_min_bytes: Option<u64>,
231 pub external_blob_sync: bool,
232 pub external_pack_target_bytes: Option<u64>,
233 pub max_read_concurrency: u32,
234 pub max_write_concurrency: u32,
235 pub temperature_low_watermark_percent: u8,
236 pub temperature_high_watermark_percent: u8,
237 pub logical_bytes: u64,
238 pub located_blobs: u64,
239 pub available: bool,
240 pub last_error: Option<String>,
241}
242
243#[derive(Debug, Clone, Default, PartialEq, Eq)]
244pub struct PoolTemperatureReport {
245 pub sampled_accesses_flushed: usize,
246 pub scanned: usize,
247 pub candidates: usize,
248 pub moved: usize,
249 pub attempted_moves: usize,
250 pub peak_concurrent_moves: usize,
251 pub bytes_moved: u64,
252 pub resumed: usize,
253 pub throttled: bool,
254 pub lease_acquired: bool,
255 pub failed: Vec<String>,
256}
257
258#[derive(Debug, Clone, Default, PartialEq, Eq)]
259pub struct PoolMaintenanceReport {
260 pub examined: usize,
261 pub moved: usize,
262 pub bytes_moved: u64,
263 pub failed: Vec<String>,
264}
265
266#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
267pub(super) struct MemberRecord {
268 pub(super) id: PoolMemberId,
269 pub(super) state: PoolMemberState,
270 pub(super) config: PoolMemberConfig,
271}
272
273#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
274pub(super) struct PoolManifest {
275 pub(super) version: u32,
276 pub(super) generation: u64,
277 pub(super) members: Vec<MemberRecord>,
278}
279
280impl Default for PoolManifest {
281 fn default() -> Self {
282 Self {
283 version: 1,
284 generation: 0,
285 members: Vec::new(),
286 }
287 }
288}
289
290#[derive(Debug, Clone, Copy, PartialEq, Eq)]
291pub(super) enum LocationRecord {
292 Pending {
293 member: PoolMemberId,
294 size: u64,
295 },
296 Stored {
297 member: PoolMemberId,
298 size: u64,
299 },
300 Moving {
301 source: PoolMemberId,
302 target: PoolMemberId,
303 size: u64,
304 },
305}
306
307impl LocationRecord {
308 pub(super) fn size(self) -> u64 {
309 match self {
310 Self::Pending { size, .. } | Self::Stored { size, .. } | Self::Moving { size, .. } => {
311 size
312 }
313 }
314 }
315
316 pub(super) fn preferred_member(self) -> PoolMemberId {
317 match self {
318 Self::Pending { member, .. } | Self::Stored { member, .. } => member,
319 Self::Moving { target, .. } => target,
320 }
321 }
322
323 pub(super) fn members(self) -> ([PoolMemberId; 2], usize) {
324 match self {
325 Self::Pending { member, .. } | Self::Stored { member, .. } => ([member, member], 1),
326 Self::Moving { source, target, .. } => ([source, target], 2),
327 }
328 }
329
330 pub(super) fn encode(self) -> Vec<u8> {
331 let mut bytes = Vec::with_capacity(41);
332 match self {
333 Self::Pending { member, size } => {
334 bytes.push(1);
335 bytes.extend_from_slice(member.as_bytes());
336 bytes.extend_from_slice(&size.to_be_bytes());
337 }
338 Self::Stored { member, size } => {
339 bytes.push(2);
340 bytes.extend_from_slice(member.as_bytes());
341 bytes.extend_from_slice(&size.to_be_bytes());
342 }
343 Self::Moving {
344 source,
345 target,
346 size,
347 } => {
348 bytes.push(3);
349 bytes.extend_from_slice(source.as_bytes());
350 bytes.extend_from_slice(target.as_bytes());
351 bytes.extend_from_slice(&size.to_be_bytes());
352 }
353 }
354 bytes
355 }
356
357 pub(super) fn decode(bytes: &[u8]) -> Result<Self, StoreError> {
358 let member = |slice: &[u8]| -> Result<PoolMemberId, StoreError> {
359 let bytes: [u8; 16] = slice
360 .try_into()
361 .map_err(|_| StoreError::Other("invalid pool member id bytes".into()))?;
362 Ok(PoolMemberId(bytes))
363 };
364 let size = |slice: &[u8]| -> Result<u64, StoreError> {
365 Ok(u64::from_be_bytes(slice.try_into().map_err(|_| {
366 StoreError::Other("invalid pool location size".into())
367 })?))
368 };
369 match bytes.first().copied() {
370 Some(1) if bytes.len() == 25 => Ok(Self::Pending {
371 member: member(&bytes[1..17])?,
372 size: size(&bytes[17..25])?,
373 }),
374 Some(2) if bytes.len() == 25 => Ok(Self::Stored {
375 member: member(&bytes[1..17])?,
376 size: size(&bytes[17..25])?,
377 }),
378 Some(3) if bytes.len() == 41 => Ok(Self::Moving {
379 source: member(&bytes[1..17])?,
380 target: member(&bytes[17..33])?,
381 size: size(&bytes[33..41])?,
382 }),
383 _ => Err(StoreError::Other("invalid pool location record".into())),
384 }
385 }
386}