1use anyhow::Result;
2use futures::executor::block_on as sync_block_on;
3use hashtree_core::store::Store;
4use hashtree_core::{to_hex, types::Hash, HashTree, HashTreeConfig, LinkType};
5use serde::de::{self, IgnoredAny, MapAccess, SeqAccess, Visitor};
6use serde::{Deserialize, Serialize};
7use std::collections::HashSet;
8use std::path::{Path, PathBuf};
9use std::time::{SystemTime, UNIX_EPOCH};
10
11use super::{BlobMetadata, HashtreeStore, PRIORITY_FOLLOWED, PRIORITY_OWN};
12
13#[derive(Debug, Clone, Serialize)]
15pub struct TreeMeta {
16 pub owner: String,
18 pub name: Option<String>,
20 pub synced_at: u64,
22 pub total_size: u64,
24 pub priority: u8,
26}
27
28impl<'de> Deserialize<'de> for TreeMeta {
29 fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
30 where
31 D: serde::Deserializer<'de>,
32 {
33 const FIELDS: &[&str] = &[
34 "owner",
35 "name",
36 "synced_at",
37 "last_accessed_at",
38 "total_size",
39 "priority",
40 ];
41
42 struct TreeMetaVisitor;
43
44 impl<'de> Visitor<'de> for TreeMetaVisitor {
45 type Value = TreeMeta;
46
47 fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
48 formatter.write_str("TreeMeta as current or legacy metadata")
49 }
50
51 fn visit_seq<A>(self, mut seq: A) -> std::result::Result<Self::Value, A::Error>
52 where
53 A: SeqAccess<'de>,
54 {
55 let has_accidental_access_field = matches!(seq.size_hint(), Some(6));
56 let owner = seq
57 .next_element()?
58 .ok_or_else(|| de::Error::invalid_length(0, &self))?;
59 let name = seq
60 .next_element()?
61 .ok_or_else(|| de::Error::invalid_length(1, &self))?;
62 let synced_at = seq
63 .next_element()?
64 .ok_or_else(|| de::Error::invalid_length(2, &self))?;
65
66 if has_accidental_access_field {
67 let _: IgnoredAny = seq
68 .next_element()?
69 .ok_or_else(|| de::Error::invalid_length(3, &self))?;
70 }
71
72 let total_size = seq
73 .next_element()?
74 .ok_or_else(|| de::Error::invalid_length(3, &self))?;
75 let priority = seq
76 .next_element()?
77 .ok_or_else(|| de::Error::invalid_length(4, &self))?;
78
79 Ok(TreeMeta {
80 owner,
81 name,
82 synced_at,
83 total_size,
84 priority,
85 })
86 }
87
88 fn visit_map<A>(self, mut map: A) -> std::result::Result<Self::Value, A::Error>
89 where
90 A: MapAccess<'de>,
91 {
92 let mut owner = None;
93 let mut name = None;
94 let mut synced_at = None;
95 let mut total_size = None;
96 let mut priority = None;
97
98 while let Some(key) = map.next_key::<String>()? {
99 match key.as_str() {
100 "owner" => owner = Some(map.next_value()?),
101 "name" => name = Some(map.next_value()?),
102 "synced_at" => synced_at = Some(map.next_value()?),
103 "last_accessed_at" => {
104 let _: IgnoredAny = map.next_value()?;
105 }
106 "total_size" => total_size = Some(map.next_value()?),
107 "priority" => priority = Some(map.next_value()?),
108 _ => {
109 let _: IgnoredAny = map.next_value()?;
110 }
111 }
112 }
113
114 Ok(TreeMeta {
115 owner: owner.ok_or_else(|| de::Error::missing_field("owner"))?,
116 name: name.unwrap_or(None),
117 synced_at: synced_at.ok_or_else(|| de::Error::missing_field("synced_at"))?,
118 total_size: total_size.ok_or_else(|| de::Error::missing_field("total_size"))?,
119 priority: priority.ok_or_else(|| de::Error::missing_field("priority"))?,
120 })
121 }
122 }
123
124 deserializer.deserialize_struct("TreeMeta", FIELDS, TreeMetaVisitor)
125 }
126}
127
128#[derive(Debug)]
129pub struct StorageStats {
130 pub total_dags: usize,
131 pub pinned_dags: usize,
132 pub total_bytes: u64,
133}
134
135#[derive(Debug, Clone)]
137pub struct StorageByPriority {
138 pub own: u64,
140 pub followed: u64,
142 pub other: u64,
144}
145
146#[derive(Debug, Clone)]
147pub struct PinnedItem {
148 pub cid: String,
149 pub name: String,
150 pub is_directory: bool,
151 pub size_bytes: u64,
152}
153
154#[derive(Debug, Clone)]
155pub struct OwnedBlobStats {
156 pub owner: [u8; 32],
157 pub count: usize,
158 pub total_bytes: u64,
159}
160
161fn pinned_item_name(hash: &Hash, meta: Option<&TreeMeta>) -> String {
162 let Some(meta) = meta else {
163 return to_hex(hash);
164 };
165
166 match (meta.owner.as_str(), meta.name.as_deref()) {
167 ("pinned", Some(name)) => name.to_string(),
168 ("", Some(name)) => name.to_string(),
169 (owner, Some(name)) if !owner.is_empty() => format!("{owner}/{name}"),
170 (owner, None) if !owner.is_empty() && owner != "pinned" => owner.to_string(),
171 _ => to_hex(hash),
172 }
173}
174
175fn unix_timestamp_now() -> u64 {
176 SystemTime::now()
177 .duration_since(UNIX_EPOCH)
178 .unwrap_or_default()
179 .as_secs()
180}
181
182impl HashtreeStore {
183 fn socialgraph_root_files(&self) -> [PathBuf; 4] {
184 let socialgraph = self.base_path().join("socialgraph");
185 [
186 socialgraph.join("events-root.msgpack"),
187 socialgraph.join("events-root-ambient.msgpack"),
188 socialgraph.join("profile-search-root.msgpack"),
189 socialgraph.join("profiles-by-pubkey-root.msgpack"),
190 ]
191 }
192
193 fn read_stored_cid(path: &Path) -> Result<Option<Hash>> {
194 #[derive(Deserialize)]
195 struct StoredCid {
196 hash: [u8; 32],
197 #[allow(dead_code)]
198 key: Option<[u8; 32]>,
199 }
200
201 let Ok(bytes) = std::fs::read(path) else {
202 return Ok(None);
203 };
204 let stored: StoredCid = rmp_serde::from_slice(&bytes)
205 .map_err(|e| anyhow::anyhow!("Failed to decode root file {}: {}", path.display(), e))?;
206 Ok(Some(stored.hash))
207 }
208
209 async fn collect_tree_hashes<S: Store>(
210 &self,
211 tree: &HashTree<S>,
212 root: &Hash,
213 ) -> Result<HashSet<Hash>> {
214 let mut hashes = HashSet::new();
215 let mut stack = vec![*root];
216
217 while let Some(hash) = stack.pop() {
218 if !hashes.insert(hash) {
219 continue;
220 }
221
222 let is_tree = tree
223 .is_tree(&hash)
224 .await
225 .map_err(|e| anyhow::anyhow!("Failed to check tree: {}", e))?;
226
227 if !is_tree {
228 continue;
229 }
230
231 if let Some(node) = tree
232 .get_tree_node(&hash)
233 .await
234 .map_err(|e| anyhow::anyhow!("Failed to get tree node: {}", e))?
235 {
236 for link in &node.links {
237 stack.push(link.hash);
238 }
239 }
240 }
241
242 Ok(hashes)
243 }
244
245 fn protected_hashes(&self) -> Result<HashSet<Hash>> {
246 let mut protected = HashSet::new();
247
248 let rtxn = self.env.read_txn()?;
249 for (key_bytes, _) in self.blob_trees.iter(&rtxn)?.flatten() {
250 if key_bytes.len() >= 32 {
251 let hash: Hash = key_bytes[..32].try_into().unwrap();
252 protected.insert(hash);
253 }
254 }
255 drop(rtxn);
256
257 let tree = HashTree::new(HashTreeConfig::new(self.store_arc()).public());
258 for path in self.socialgraph_root_files() {
259 let Some(root_hash) = Self::read_stored_cid(&path)? else {
260 continue;
261 };
262 protected.extend(sync_block_on(self.collect_tree_hashes(&tree, &root_hash))?);
263 }
264
265 Ok(protected)
266 }
267
268 fn evict_disposable_orphans_to_target(&self, target_bytes: u64) -> Result<u64> {
269 let stats = self
270 .router
271 .writable_stats()
272 .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
273 let mut current_size = stats.total_bytes;
274 if current_size <= target_bytes {
275 return Ok(0);
276 }
277
278 let rtxn = self.env.read_txn()?;
279 let pinned: HashSet<Hash> = self
280 .pins
281 .iter(&rtxn)?
282 .filter_map(|item| item.ok())
283 .filter_map(|(hash_bytes, _)| {
284 if hash_bytes.len() == 32 {
285 let mut hash = [0u8; 32];
286 hash.copy_from_slice(hash_bytes);
287 Some(hash)
288 } else {
289 None
290 }
291 })
292 .collect();
293 drop(rtxn);
294
295 let protected_hashes = self.protected_hashes()?;
296 let all_hashes = self
297 .router
298 .list_writable()
299 .map_err(|e| anyhow::anyhow!("Failed to list writable hashes: {}", e))?;
300
301 let mut freed = 0u64;
302 for hash in all_hashes {
303 if current_size <= target_bytes {
304 break;
305 }
306
307 if pinned.contains(&hash) || protected_hashes.contains(&hash) {
308 continue;
309 }
310
311 if self.blob_has_owners(&hash)? {
312 continue;
313 }
314
315 let Some(size) = self
316 .router
317 .blob_size_sync(&hash)
318 .map_err(|e| anyhow::anyhow!("Failed to get blob size: {}", e))?
319 else {
320 continue;
321 };
322
323 if self
324 .router
325 .delete_local_only(&hash)
326 .map_err(|e| anyhow::anyhow!("Failed to delete orphaned blob: {}", e))?
327 {
328 freed = freed.saturating_add(size);
329 current_size = current_size.saturating_sub(size);
330 tracing::debug!(
331 "Deleted disposable orphaned blob {} ({} bytes)",
332 &to_hex(&hash)[..8],
333 size
334 );
335 }
336 }
337
338 Ok(freed)
339 }
340
341 pub fn make_room_for_cached_blob(&self, incoming_bytes: u64) -> Result<u64> {
342 if self.max_size_bytes == 0 {
343 return Ok(0);
344 }
345
346 let stats = self
347 .router
348 .writable_stats()
349 .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
350 if stats.total_bytes.saturating_add(incoming_bytes) <= self.max_size_bytes {
351 return Ok(0);
352 }
353
354 let target = if incoming_bytes >= self.max_size_bytes {
355 0
356 } else {
357 (self.max_size_bytes.saturating_mul(9) / 10)
358 .min(self.max_size_bytes.saturating_sub(incoming_bytes))
359 };
360 self.evict_disposable_orphans_to_target(target)
361 }
362
363 pub fn enforce_cached_blob_budget_after_insert(&self, inserted_bytes: u64) -> Result<u64> {
364 if self.max_size_bytes == 0 || inserted_bytes == 0 {
365 return Ok(0);
366 }
367
368 let stats = self
369 .router
370 .writable_stats()
371 .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
372 if stats.total_bytes <= self.max_size_bytes {
373 return Ok(0);
374 }
375
376 let target = if inserted_bytes >= self.max_size_bytes {
377 inserted_bytes
378 } else {
379 (self.max_size_bytes.saturating_mul(9) / 10)
380 .saturating_add(inserted_bytes)
381 .min(self.max_size_bytes)
382 };
383 self.evict_disposable_orphans_to_target(target)
384 }
385
386 pub fn make_room_for_durable_blob(&self, incoming_bytes: u64) -> Result<u64> {
387 if self.max_size_bytes == 0 || incoming_bytes == 0 {
388 return Ok(0);
389 }
390
391 if incoming_bytes > self.max_size_bytes {
392 anyhow::bail!(
393 "storage limit exceeded: incoming blob is {} bytes but limit is {} bytes",
394 incoming_bytes,
395 self.max_size_bytes
396 );
397 }
398
399 let stats = self
400 .router
401 .writable_stats()
402 .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
403 if stats.total_bytes.saturating_add(incoming_bytes) <= self.max_size_bytes {
404 return Ok(0);
405 }
406
407 let target = (self.max_size_bytes.saturating_mul(9) / 10)
408 .min(self.max_size_bytes.saturating_sub(incoming_bytes));
409 let freed = self.evict_with_policy_to_target(stats.total_bytes, target)?;
410
411 let next_stats = self
412 .router
413 .writable_stats()
414 .map_err(|e| anyhow::anyhow!("Failed to get writable stats after eviction: {}", e))?;
415 if next_stats.total_bytes.saturating_add(incoming_bytes) > self.max_size_bytes {
416 anyhow::bail!(
417 "storage limit exceeded: {} bytes used, {} byte incoming blob, {} byte limit",
418 next_stats.total_bytes,
419 incoming_bytes,
420 self.max_size_bytes
421 );
422 }
423
424 Ok(freed)
425 }
426
427 pub fn enforce_durable_blob_budget_after_insert(&self, inserted_bytes: u64) -> Result<u64> {
428 if self.max_size_bytes == 0 || inserted_bytes == 0 {
429 return Ok(0);
430 }
431
432 if inserted_bytes > self.max_size_bytes {
433 anyhow::bail!(
434 "storage limit exceeded: inserted blobs are {} bytes but limit is {} bytes",
435 inserted_bytes,
436 self.max_size_bytes
437 );
438 }
439
440 let stats = self
441 .router
442 .writable_stats()
443 .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
444 if stats.total_bytes <= self.max_size_bytes {
445 return Ok(0);
446 }
447
448 let target = (self.max_size_bytes.saturating_mul(9) / 10)
449 .saturating_add(inserted_bytes)
450 .min(self.max_size_bytes);
451 let freed = self.evict_with_policy_to_target(stats.total_bytes, target)?;
452
453 let next_stats = self
454 .router
455 .writable_stats()
456 .map_err(|e| anyhow::anyhow!("Failed to get writable stats after eviction: {}", e))?;
457 if next_stats.total_bytes > self.max_size_bytes {
458 anyhow::bail!(
459 "storage limit exceeded: {} bytes used after inserting {} bytes, {} byte limit",
460 next_stats.total_bytes,
461 inserted_bytes,
462 self.max_size_bytes
463 );
464 }
465
466 Ok(freed)
467 }
468
469 pub fn relieve_cached_blob_write_pressure(&self, incoming_bytes: u64) -> Result<u64> {
470 let stats = self
471 .router
472 .writable_stats()
473 .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
474 if stats.total_bytes == 0 {
475 return Ok(0);
476 }
477
478 let headroom = incoming_bytes.max(stats.total_bytes / 10).max(1);
479 let target = stats.total_bytes.saturating_sub(headroom);
480 self.evict_disposable_orphans_to_target(target)
481 }
482
483 pub fn pin(&self, hash: &[u8; 32]) -> Result<()> {
485 let mut wtxn = self.env.write_txn()?;
486 self.pins.put(&mut wtxn, hash.as_slice(), &())?;
487 wtxn.commit()?;
488 Ok(())
489 }
490
491 pub fn unpin(&self, hash: &[u8; 32]) -> Result<()> {
493 let mut wtxn = self.env.write_txn()?;
494 self.pins.delete(&mut wtxn, hash.as_slice())?;
495 wtxn.commit()?;
496 Ok(())
497 }
498
499 pub fn is_pinned(&self, hash: &[u8; 32]) -> Result<bool> {
501 let rtxn = self.env.read_txn()?;
502 Ok(self.pins.get(&rtxn, hash.as_slice())?.is_some())
503 }
504
505 pub fn list_pins_raw(&self) -> Result<Vec<[u8; 32]>> {
507 let rtxn = self.env.read_txn()?;
508 let mut pins = Vec::new();
509
510 for item in self.pins.iter(&rtxn)? {
511 let (hash_bytes, _) = item?;
512 if hash_bytes.len() == 32 {
513 let mut hash = [0u8; 32];
514 hash.copy_from_slice(hash_bytes);
515 pins.push(hash);
516 }
517 }
518
519 Ok(pins)
520 }
521
522 pub fn list_pins_with_names(&self) -> Result<Vec<PinnedItem>> {
524 let rtxn = self.env.read_txn()?;
525 let store = self.store_arc();
526 let tree = HashTree::new(HashTreeConfig::new(store).public());
527 let mut pins = Vec::new();
528
529 for item in self.pins.iter(&rtxn)? {
530 let (hash_bytes, _) = item?;
531 if hash_bytes.len() != 32 {
532 continue;
533 }
534 let mut hash = [0u8; 32];
535 hash.copy_from_slice(hash_bytes);
536
537 let is_directory =
539 sync_block_on(async { tree.is_directory(&hash).await.unwrap_or(false) });
540
541 let meta = self
542 .tree_meta
543 .get(&rtxn, hash.as_slice())?
544 .map(|bytes| {
545 rmp_serde::from_slice::<TreeMeta>(bytes)
546 .map_err(|e| anyhow::anyhow!("Failed to deserialize TreeMeta: {}", e))
547 })
548 .transpose()?;
549 let size_bytes = if let Some(meta) = meta.as_ref() {
550 meta.total_size
551 } else {
552 self.router
553 .blob_size_sync(&hash)
554 .map_err(|e| anyhow::anyhow!("Failed to get pinned blob size: {}", e))?
555 .unwrap_or(0)
556 };
557
558 pins.push(PinnedItem {
559 cid: to_hex(&hash),
560 name: pinned_item_name(&hash, meta.as_ref()),
561 is_directory,
562 size_bytes,
563 });
564 }
565
566 Ok(pins)
567 }
568
569 pub fn owned_blob_stats(&self) -> Result<Vec<OwnedBlobStats>> {
570 let rtxn = self.env.read_txn()?;
571 let mut owners = Vec::new();
572
573 for item in self.pubkey_blobs.iter(&rtxn)? {
574 let (owner_bytes, blobs_bytes) = item?;
575 if owner_bytes.len() != 32 {
576 continue;
577 }
578
579 let blobs: Vec<BlobMetadata> = serde_json::from_slice(blobs_bytes)
580 .map_err(|e| anyhow::anyhow!("Failed to deserialize blob metadata: {}", e))?;
581 let mut owner = [0u8; 32];
582 owner.copy_from_slice(owner_bytes);
583 let total_bytes = blobs
584 .iter()
585 .fold(0u64, |total, blob| total.saturating_add(blob.size));
586 owners.push(OwnedBlobStats {
587 owner,
588 count: blobs.len(),
589 total_bytes,
590 });
591 }
592
593 owners.sort_by_key(|stats| stats.owner);
594 Ok(owners)
595 }
596
597 pub fn index_tree(
604 &self,
605 root_hash: &Hash,
606 owner: &str,
607 name: Option<&str>,
608 priority: u8,
609 ref_key: Option<&str>,
610 ) -> Result<()> {
611 let root_hex = to_hex(root_hash);
612
613 if let Some(key) = ref_key {
615 let rtxn = self.env.read_txn()?;
616 if let Some(old_hash_bytes) = self.tree_refs.get(&rtxn, key)? {
617 if old_hash_bytes != root_hash.as_slice() {
618 let old_hash: Hash = old_hash_bytes
619 .try_into()
620 .map_err(|_| anyhow::anyhow!("Invalid hash in tree_refs"))?;
621 drop(rtxn);
622 let _ = self.unpin(&old_hash);
623 let _ = self.unindex_tree(&old_hash);
625 tracing::debug!("Replaced old tree for ref {}", key);
626 }
627 }
628 }
629
630 let store = self.store_arc();
631 let tree = HashTree::new(HashTreeConfig::new(store).public());
632
633 let (tracked_hashes, total_size) =
634 sync_block_on(async { self.collect_tree_index(&tree, root_hash).await })?;
635
636 let mut wtxn = self.env.write_txn()?;
637
638 for tracked_hash in &tracked_hashes {
640 let mut key = [0u8; 64];
641 key[..32].copy_from_slice(tracked_hash);
642 key[32..].copy_from_slice(root_hash);
643 self.blob_trees.put(&mut wtxn, &key[..], &())?;
644 }
645
646 let now = unix_timestamp_now();
648 let meta = TreeMeta {
649 owner: owner.to_string(),
650 name: name.map(|s| s.to_string()),
651 synced_at: now,
652 total_size,
653 priority,
654 };
655 let meta_bytes = rmp_serde::to_vec(&meta)
656 .map_err(|e| anyhow::anyhow!("Failed to serialize TreeMeta: {}", e))?;
657 self.tree_meta
658 .put(&mut wtxn, root_hash.as_slice(), &meta_bytes)?;
659
660 if let Some(key) = ref_key {
662 self.tree_refs.put(&mut wtxn, key, root_hash.as_slice())?;
663 }
664
665 wtxn.commit()?;
666
667 tracing::debug!(
668 "Indexed tree {} ({} blobs, {} bytes, priority {})",
669 &root_hex[..8],
670 tracked_hashes.len(),
671 total_size,
672 priority
673 );
674
675 Ok(())
676 }
677
678 async fn collect_tree_index<S: Store>(
679 &self,
680 tree: &HashTree<S>,
681 root: &Hash,
682 ) -> Result<(HashSet<Hash>, u64)> {
683 let mut hashes = HashSet::new();
684 let mut total_size = 0u64;
685 let mut stack = vec![(*root, true)];
686
687 while let Some((hash, count_bytes)) = stack.pop() {
688 hashes.insert(hash);
689
690 let Some(node) = tree
691 .get_tree_node(&hash)
692 .await
693 .map_err(|e| anyhow::anyhow!("Failed to get tree node: {}", e))?
694 else {
695 if count_bytes {
696 if let Some(size) = self
697 .router
698 .blob_size_sync(&hash)
699 .map_err(|e| anyhow::anyhow!("Failed to get blob size: {}", e))?
700 {
701 total_size = total_size.saturating_add(size);
702 }
703 }
704 continue;
705 };
706
707 for link in &node.links {
708 hashes.insert(link.hash);
709 match link.link_type {
710 LinkType::Blob => {
711 if count_bytes {
712 total_size = total_size.saturating_add(link.size);
713 }
714 }
715 LinkType::File => {
716 if count_bytes {
717 total_size = total_size.saturating_add(link.size);
718 }
719 stack.push((link.hash, false));
720 }
721 LinkType::Dir | LinkType::Fanout => {
722 stack.push((link.hash, count_bytes));
723 }
724 }
725 }
726 }
727
728 Ok((hashes, total_size))
729 }
730
731 pub fn unindex_tree(&self, root_hash: &Hash) -> Result<u64> {
734 let root_hex = to_hex(root_hash);
735
736 let store = self.store_arc();
737 let tree = HashTree::new(HashTreeConfig::new(store).public());
738
739 let tracked_hashes = sync_block_on(self.collect_tree_hashes(&tree, root_hash))?;
741
742 let mut wtxn = self.env.write_txn()?;
743 let mut freed = 0u64;
744
745 for tracked_hash in &tracked_hashes {
747 let mut key = [0u8; 64];
749 key[..32].copy_from_slice(tracked_hash);
750 key[32..].copy_from_slice(root_hash);
751 self.blob_trees.delete(&mut wtxn, &key[..])?;
752
753 let mut has_other_tree = false;
755 for item in self.blob_trees.prefix_iter(&wtxn, &tracked_hash[..])? {
756 if item.is_ok() {
757 has_other_tree = true;
758 break;
759 }
760 }
761
762 if !has_other_tree {
764 if let Some(size) = self
765 .router
766 .blob_size_sync(tracked_hash)
767 .map_err(|e| anyhow::anyhow!("Failed to get blob size: {}", e))?
768 {
769 freed += size;
770 self.router
772 .delete_local_only(tracked_hash)
773 .map_err(|e| anyhow::anyhow!("Failed to delete blob: {}", e))?;
774 }
775 }
776 }
777
778 self.tree_meta.delete(&mut wtxn, root_hash.as_slice())?;
780
781 wtxn.commit()?;
782
783 tracing::debug!("Unindexed tree {} ({} bytes freed)", &root_hex[..8], freed);
784
785 Ok(freed)
786 }
787
788 pub fn get_tree_meta(&self, root_hash: &Hash) -> Result<Option<TreeMeta>> {
790 let rtxn = self.env.read_txn()?;
791 if let Some(bytes) = self.tree_meta.get(&rtxn, root_hash.as_slice())? {
792 let meta: TreeMeta = rmp_serde::from_slice(bytes)
793 .map_err(|e| anyhow::anyhow!("Failed to deserialize TreeMeta: {}", e))?;
794 Ok(Some(meta))
795 } else {
796 Ok(None)
797 }
798 }
799
800 pub fn get_tree_ref(&self, key: &str) -> Result<Option<Hash>> {
801 let rtxn = self.env.read_txn()?;
802 let Some(bytes) = self.tree_refs.get(&rtxn, key)? else {
803 return Ok(None);
804 };
805
806 let hash: Hash = bytes
807 .try_into()
808 .map_err(|_| anyhow::anyhow!("Invalid hash in tree_refs"))?;
809 Ok(Some(hash))
810 }
811
812 pub fn list_indexed_trees(&self) -> Result<Vec<(Hash, TreeMeta)>> {
814 let rtxn = self.env.read_txn()?;
815 let mut trees = Vec::new();
816
817 for item in self.tree_meta.iter(&rtxn)? {
818 let (hash_bytes, meta_bytes) = item?;
819 let hash: Hash = hash_bytes
820 .try_into()
821 .map_err(|_| anyhow::anyhow!("Invalid hash in tree_meta"))?;
822 let meta: TreeMeta = rmp_serde::from_slice(meta_bytes)
823 .map_err(|e| anyhow::anyhow!("Failed to deserialize TreeMeta: {}", e))?;
824 trees.push((hash, meta));
825 }
826
827 Ok(trees)
828 }
829
830 pub fn tracked_size(&self) -> Result<u64> {
832 let rtxn = self.env.read_txn()?;
833 let mut total = 0u64;
834
835 for item in self.tree_meta.iter(&rtxn)? {
836 let (_, bytes) = item?;
837 let meta: TreeMeta = rmp_serde::from_slice(bytes)
838 .map_err(|e| anyhow::anyhow!("Failed to deserialize TreeMeta: {}", e))?;
839 total += meta.total_size;
840 }
841
842 Ok(total)
843 }
844
845 fn get_evictable_trees(&self) -> Result<Vec<(Hash, TreeMeta)>> {
851 let mut trees = self.list_indexed_trees()?;
852
853 trees.sort_by(|a, b| match a.1.priority.cmp(&b.1.priority) {
855 std::cmp::Ordering::Equal => a.1.synced_at.cmp(&b.1.synced_at),
856 other => other,
857 });
858
859 Ok(trees)
860 }
861
862 pub fn evict_if_needed(&self) -> Result<u64> {
869 let stats = self
872 .router
873 .writable_stats()
874 .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
875 let current = stats.total_bytes;
876
877 if current <= self.max_size_bytes {
878 return Ok(0);
879 }
880
881 let target = self.max_size_bytes * 90 / 100;
883 self.evict_with_policy_to_target(current, target)
884 }
885
886 fn evict_with_policy_to_target(&self, current: u64, target: u64) -> Result<u64> {
887 let mut freed = 0u64;
888 let mut current_size = current;
889
890 if self.evict_orphans {
892 let orphan_freed = self.evict_disposable_orphans_to_target(target)?;
893 freed += orphan_freed;
894 current_size = current_size.saturating_sub(orphan_freed);
895
896 if orphan_freed > 0 {
897 tracing::info!("Evicted orphaned blobs: {} bytes freed", orphan_freed);
898 }
899 } else {
900 tracing::debug!("Skipping orphan blob eviction; storage.evict_orphans=false");
901 }
902
903 if current_size <= target {
905 if freed > 0 {
906 tracing::info!("Eviction complete: {} bytes freed", freed);
907 }
908 return Ok(freed);
909 }
910
911 let evictable = self.get_evictable_trees()?;
914
915 for (root_hash, meta) in evictable {
916 if current_size <= target {
917 break;
918 }
919
920 let root_hex = to_hex(&root_hash);
921
922 if self.is_pinned(&root_hash)? {
924 continue;
925 }
926
927 let tree_freed = self.unindex_tree(&root_hash)?;
928 freed += tree_freed;
929 current_size = current_size.saturating_sub(tree_freed);
930
931 tracing::info!(
932 "Evicted tree {} (owner={}, priority={}, {} bytes)",
933 &root_hex[..8],
934 &meta.owner[..8.min(meta.owner.len())],
935 meta.priority,
936 tree_freed
937 );
938 }
939
940 if freed > 0 {
941 tracing::info!("Eviction complete: {} bytes freed", freed);
942 }
943
944 Ok(freed)
945 }
946
947 pub fn max_size_bytes(&self) -> u64 {
949 self.max_size_bytes
950 }
951
952 pub fn storage_by_priority(&self) -> Result<StorageByPriority> {
954 let rtxn = self.env.read_txn()?;
955 let mut own = 0u64;
956 let mut followed = 0u64;
957 let mut other = 0u64;
958
959 for item in self.tree_meta.iter(&rtxn)? {
960 let (_, bytes) = item?;
961 let meta: TreeMeta = rmp_serde::from_slice(bytes)
962 .map_err(|e| anyhow::anyhow!("Failed to deserialize TreeMeta: {}", e))?;
963
964 if meta.priority == PRIORITY_OWN {
965 own += meta.total_size;
966 } else if meta.priority >= PRIORITY_FOLLOWED {
967 followed += meta.total_size;
968 } else {
969 other += meta.total_size;
970 }
971 }
972
973 Ok(StorageByPriority {
974 own,
975 followed,
976 other,
977 })
978 }
979
980 pub fn get_storage_stats(&self) -> Result<StorageStats> {
982 let rtxn = self.env.read_txn()?;
983 let total_pins = self.pins.len(&rtxn)? as usize;
984
985 let stats = self
986 .router
987 .stats()
988 .map_err(|e| anyhow::anyhow!("Failed to get stats: {}", e))?;
989
990 Ok(StorageStats {
991 total_dags: stats.count,
992 pinned_dags: total_pins,
993 total_bytes: stats.total_bytes,
994 })
995 }
996}
997
998#[cfg(test)]
999mod tests {
1000 use super::*;
1001 use hashtree_core::Cid;
1002 use hashtree_index::{BTree, BTreeOptions};
1003 use tempfile::TempDir;
1004
1005 use crate::storage::PRIORITY_OTHER;
1006
1007 fn write_root_file(path: &Path, cid: &Cid) {
1008 #[derive(Serialize)]
1009 struct StoredCid {
1010 hash: [u8; 32],
1011 key: Option<[u8; 32]>,
1012 }
1013
1014 std::fs::create_dir_all(path.parent().expect("root file parent")).expect("create dir");
1015 let bytes = rmp_serde::to_vec_named(&StoredCid {
1016 hash: cid.hash,
1017 key: cid.key,
1018 })
1019 .expect("encode cid");
1020 std::fs::write(path, bytes).expect("write root file");
1021 }
1022
1023 fn build_test_tree(store: &HashtreeStore) -> Cid {
1024 let index = BTree::new(store.store_arc(), BTreeOptions { order: Some(8) });
1025 sync_block_on(index.build(vec![
1026 ("alpha".to_string(), "one".to_string()),
1027 ("beta".to_string(), "two".to_string()),
1028 ("gamma".to_string(), "three".to_string()),
1029 ]))
1030 .expect("build btree")
1031 .expect("non-empty root")
1032 }
1033
1034 #[test]
1035 fn orphan_cleanup_keeps_indexed_tree_hashes() {
1036 let temp_dir = TempDir::new().expect("temp dir");
1037 let store = HashtreeStore::with_options(temp_dir.path(), None, 1024).expect("store");
1038 let cid = build_test_tree(&store);
1039
1040 store
1041 .index_tree(
1042 &cid.hash,
1043 "owner",
1044 Some("tree"),
1045 PRIORITY_OTHER,
1046 Some("owner/tree"),
1047 )
1048 .expect("index tree");
1049 let freed = store
1050 .evict_disposable_orphans_to_target(0)
1051 .expect("orphan cleanup");
1052
1053 assert!(freed < 1024);
1054 assert!(store.blob_exists(&cid.hash).expect("root exists"));
1055 }
1056
1057 #[test]
1058 fn list_pins_with_names_uses_indexed_tree_metadata() {
1059 let temp_dir = TempDir::new().expect("temp dir");
1060 let store = HashtreeStore::with_options(temp_dir.path(), None, 1024 * 1024).expect("store");
1061 let cid = build_test_tree(&store);
1062
1063 store.pin(&cid.hash).expect("pin tree");
1064 store
1065 .index_tree(
1066 &cid.hash,
1067 "npub1example",
1068 Some("playlist"),
1069 PRIORITY_OTHER,
1070 Some("npub1example/playlist"),
1071 )
1072 .expect("index tree");
1073
1074 let pins = store.list_pins_with_names().expect("list pins");
1075
1076 assert_eq!(pins.len(), 1);
1077 assert_eq!(pins[0].name, "npub1example/playlist");
1078 assert!(pins[0].size_bytes > 0);
1079 }
1080
1081 #[test]
1082 fn index_tree_records_multilevel_file_size_from_links() {
1083 let temp_dir = TempDir::new().expect("temp dir");
1084 let store = HashtreeStore::with_options(temp_dir.path(), None, 1024 * 1024).expect("store");
1085 let tree = HashTree::new(
1086 HashTreeConfig::new(store.store_arc())
1087 .public()
1088 .with_chunk_size(4)
1089 .with_max_links(2),
1090 );
1091 let data = (0u8..31).collect::<Vec<_>>();
1092 let (cid, size) = sync_block_on(tree.put(&data)).expect("put file");
1093
1094 store
1095 .index_tree(
1096 &cid.hash,
1097 "npub1example",
1098 Some("large-file"),
1099 PRIORITY_OTHER,
1100 Some("npub1example/large-file"),
1101 )
1102 .expect("index tree");
1103
1104 let meta = store
1105 .get_tree_meta(&cid.hash)
1106 .expect("tree meta")
1107 .expect("indexed meta");
1108 assert_eq!(size, data.len() as u64);
1109 assert_eq!(meta.total_size, data.len() as u64);
1110 }
1111
1112 #[test]
1113 fn get_tree_ref_returns_stored_root() {
1114 let temp_dir = TempDir::new().expect("temp dir");
1115 let store = HashtreeStore::with_options(temp_dir.path(), None, 1024 * 1024).expect("store");
1116 let cid = build_test_tree(&store);
1117
1118 store
1119 .index_tree(
1120 &cid.hash,
1121 "npub1example",
1122 Some("playlist"),
1123 PRIORITY_OTHER,
1124 Some("npub1example/playlist"),
1125 )
1126 .expect("index tree");
1127
1128 assert_eq!(
1129 store
1130 .get_tree_ref("npub1example/playlist")
1131 .expect("tree ref lookup"),
1132 Some(cid.hash)
1133 );
1134 }
1135
1136 #[test]
1137 fn tree_meta_deserializes_metadata_without_tree_access_field() {
1138 #[derive(Serialize)]
1139 struct LegacyTreeMeta {
1140 owner: String,
1141 name: Option<String>,
1142 synced_at: u64,
1143 total_size: u64,
1144 priority: u8,
1145 }
1146
1147 let bytes = rmp_serde::to_vec(&LegacyTreeMeta {
1148 owner: "owner".to_string(),
1149 name: Some("tree".to_string()),
1150 synced_at: 123,
1151 total_size: 456,
1152 priority: PRIORITY_OTHER,
1153 })
1154 .expect("serialize legacy metadata");
1155 let meta: TreeMeta = rmp_serde::from_slice(&bytes).expect("deserialize tree metadata");
1156
1157 assert_eq!(meta.owner, "owner");
1158 assert_eq!(meta.name.as_deref(), Some("tree"));
1159 assert_eq!(meta.synced_at, 123);
1160 assert_eq!(meta.total_size, 456);
1161 assert_eq!(meta.priority, PRIORITY_OTHER);
1162 }
1163
1164 #[test]
1165 fn tree_meta_deserializes_accidental_access_field_but_drops_it_on_write() {
1166 #[derive(Serialize)]
1167 struct AccidentalTreeMeta {
1168 owner: String,
1169 name: Option<String>,
1170 synced_at: u64,
1171 last_accessed_at: u64,
1172 total_size: u64,
1173 priority: u8,
1174 }
1175
1176 let bytes = rmp_serde::to_vec(&AccidentalTreeMeta {
1177 owner: "owner".to_string(),
1178 name: Some("tree".to_string()),
1179 synced_at: 123,
1180 last_accessed_at: 999,
1181 total_size: 456,
1182 priority: PRIORITY_OTHER,
1183 })
1184 .expect("serialize accidental metadata");
1185 let meta: TreeMeta = rmp_serde::from_slice(&bytes).expect("deserialize tree metadata");
1186 let encoded = rmp_serde::to_vec(&meta).expect("serialize current metadata");
1187 let reparsed: (String, Option<String>, u64, u64, u8) =
1188 rmp_serde::from_slice(&encoded).expect("parse current metadata shape");
1189
1190 assert_eq!(meta.owner, "owner");
1191 assert_eq!(meta.name.as_deref(), Some("tree"));
1192 assert_eq!(meta.synced_at, 123);
1193 assert_eq!(meta.total_size, 456);
1194 assert_eq!(meta.priority, PRIORITY_OTHER);
1195 assert_eq!(reparsed.0, "owner");
1196 assert_eq!(reparsed.3, 456);
1197 assert_eq!(reparsed.4, PRIORITY_OTHER);
1198 }
1199
1200 #[test]
1201 fn eviction_prefers_oldest_tree_within_priority() {
1202 let temp_dir = TempDir::new().expect("temp dir");
1203 let store = HashtreeStore::with_options(temp_dir.path(), None, 500).expect("store");
1204
1205 let hash1 = hashtree_core::sha256(&[1u8; 200]);
1206 let hash2 = hashtree_core::sha256(&[2u8; 200]);
1207 let hash3 = hashtree_core::sha256(&[3u8; 200]);
1208 store.put_blob(&[1u8; 200]).expect("put blob 1");
1209 store.put_blob(&[2u8; 200]).expect("put blob 2");
1210 store.put_blob(&[3u8; 200]).expect("put blob 3");
1211 store
1212 .index_tree(&hash1, "owner1", Some("tree1"), PRIORITY_OTHER, None)
1213 .expect("index tree 1");
1214 store
1215 .index_tree(&hash2, "owner2", Some("tree2"), PRIORITY_OTHER, None)
1216 .expect("index tree 2");
1217 store
1218 .index_tree(&hash3, "owner3", Some("tree3"), PRIORITY_OTHER, None)
1219 .expect("index tree 3");
1220
1221 let freed = store.evict_if_needed().expect("evict");
1222
1223 assert!(freed > 0);
1224 assert!(
1225 store.get_tree_meta(&hash3).expect("tree meta").is_some(),
1226 "newest tree should survive before older peers at the same priority"
1227 );
1228 }
1229
1230 #[test]
1231 fn orphan_cleanup_keeps_socialgraph_root_hashes() {
1232 let temp_dir = TempDir::new().expect("temp dir");
1233 let store = HashtreeStore::with_options(temp_dir.path(), None, 1024).expect("store");
1234 let cid = build_test_tree(&store);
1235 write_root_file(
1236 &temp_dir.path().join("socialgraph/events-root.msgpack"),
1237 &cid,
1238 );
1239
1240 let freed = store
1241 .evict_disposable_orphans_to_target(0)
1242 .expect("orphan cleanup");
1243
1244 assert!(freed < 1024);
1245 assert!(store.blob_exists(&cid.hash).expect("root exists"));
1246 }
1247}