1use std::pin::Pin;
7use std::sync::Arc;
8
9use futures::io::AsyncRead;
10use futures::stream::{self, Stream};
11use futures::AsyncReadExt;
12
13use crate::builder::{BuilderError, DEFAULT_CHUNK_SIZE, DEFAULT_MAX_LINKS};
14use crate::codec::{decode_tree_node, encode_and_hash, is_directory_node, is_tree_node};
15use crate::directory::{directory_fanout_meta, is_internal_directory_link, DirectoryFanoutSpan};
16use crate::hash::sha256;
17use crate::reader::{ReaderError, TreeEntry, WalkEntry};
18use crate::store::Store;
19use crate::types::{to_hex, Cid, DirEntry, Hash, Link, LinkType, TreeNode};
20
21use crate::encryption::{decrypt_object, ObjectKey};
22
23const DEFAULT_STREAM_PUT_BATCH_TARGET_BYTES: usize = 64 * 1024 * 1024;
24const STREAM_PUT_BATCH_TARGET_BYTES_ENV: &str = "HTREE_STREAM_PUT_BATCH_TARGET_BYTES";
25const STREAM_PUT_BATCH_MAX_ITEMS: usize = 128;
26const STREAM_PUT_INITIAL_CHUNK_CAPACITY: usize = 64 * 1024;
27const ASSEMBLE_MAX_RESERVE_BYTES: u64 = 256 * 1024 * 1024;
29
30#[path = "hashtree/stream.rs"]
31mod read_stream;
32mod walk;
33
34#[derive(Clone)]
36pub struct HashTreeConfig<S: Store> {
37 pub store: Arc<S>,
38 pub chunk_size: usize,
39 pub max_links: usize,
40 pub encrypted: bool,
42}
43
44impl<S: Store> HashTreeConfig<S> {
45 pub fn new(store: Arc<S>) -> Self {
46 Self {
47 store,
48 chunk_size: DEFAULT_CHUNK_SIZE,
49 max_links: DEFAULT_MAX_LINKS,
50 encrypted: true,
51 }
52 }
53
54 pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
55 self.chunk_size = chunk_size;
56 self
57 }
58
59 pub fn with_max_links(mut self, max_links: usize) -> Self {
60 self.max_links = max_links;
61 self
62 }
63
64 pub fn public(mut self) -> Self {
66 self.encrypted = false;
67 self
68 }
69}
70
71#[derive(Debug, thiserror::Error)]
73pub enum HashTreeError {
74 #[error("Store error: {0}")]
75 Store(String),
76 #[error("Codec error: {0}")]
77 Codec(#[from] crate::codec::CodecError),
78 #[error("Missing chunk: {0}")]
79 MissingChunk(String),
80 #[error("Path not found: {0}")]
81 PathNotFound(String),
82 #[error("Entry not found: {0}")]
83 EntryNotFound(String),
84 #[error("Encryption error: {0}")]
85 Encryption(String),
86 #[error("Decryption error: {0}")]
87 Decryption(String),
88 #[error("Content size {actual_size} exceeds max_size {max_size}")]
89 SizeLimitExceeded { max_size: u64, actual_size: u64 },
90}
91
92impl From<BuilderError> for HashTreeError {
93 fn from(e: BuilderError) -> Self {
94 match e {
95 BuilderError::Store(s) => HashTreeError::Store(s),
96 BuilderError::Codec(c) => HashTreeError::Codec(c),
97 BuilderError::Encryption(s) => HashTreeError::Encryption(s),
98 }
99 }
100}
101
102impl From<ReaderError> for HashTreeError {
103 fn from(e: ReaderError) -> Self {
104 match e {
105 ReaderError::Store(s) => HashTreeError::Store(s),
106 ReaderError::Codec(c) => HashTreeError::Codec(c),
107 ReaderError::MissingChunk(s) => HashTreeError::MissingChunk(s),
108 ReaderError::Decryption(s) => HashTreeError::Encryption(s),
109 ReaderError::MissingKey => {
110 HashTreeError::Encryption("missing decryption key".to_string())
111 }
112 }
113 }
114}
115
116fn decrypt_cid_root_bytes(cid: &Cid, data: Vec<u8>) -> Result<Vec<u8>, HashTreeError> {
117 let Some(key) = &cid.key else {
118 return Ok(data);
119 };
120
121 match decrypt_object(&data, key) {
122 Ok(decrypted) => Ok(decrypted),
125 Err(_)
126 if key.mode() == crate::encryption::EncryptionMode::LegacyChk
127 && is_tree_node(&data) =>
128 {
129 Ok(data)
130 }
131 Err(error) => Err(HashTreeError::Decryption(error.to_string())),
132 }
133}
134
135pub fn decode_tree_node_by_cid(
138 cid: &Cid,
139 data: Vec<u8>,
140) -> Result<Option<TreeNode>, HashTreeError> {
141 let decrypted = decrypt_cid_root_bytes(cid, data)?;
142 if !is_tree_node(&decrypted) {
143 return Ok(None);
144 }
145 Ok(Some(decode_tree_node(&decrypted)?))
146}
147
148pub struct HashTree<S: Store> {
150 store: Arc<S>,
151 chunk_size: usize,
152 max_links: usize,
153 encrypted: bool,
154 encryption_mode: crate::encryption::EncryptionMode,
155}
156
157impl<S: Store> HashTree<S> {
158 pub fn new(config: HashTreeConfig<S>) -> Self {
159 Self {
160 store: config.store,
161 chunk_size: config.chunk_size,
162 max_links: config.max_links,
163 encrypted: config.encrypted,
164 encryption_mode: crate::encryption::EncryptionMode::default(),
165 }
166 }
167
168 pub fn with_encryption_mode(mut self, mode: crate::encryption::EncryptionMode) -> Self {
170 self.encryption_mode = mode;
171 self
172 }
173
174 pub fn is_encrypted(&self) -> bool {
176 self.encrypted
177 }
178
179 pub async fn put(&self, data: &[u8]) -> Result<(Cid, u64), HashTreeError> {
184 let size = data.len() as u64;
185
186 if data.len() <= self.chunk_size {
188 let (hash, key) = self.put_chunk_internal(data).await?;
189 return Ok((Cid { hash, key }, size));
190 }
191
192 let mut links: Vec<Link> = Vec::new();
194 let mut offset = 0;
195
196 while offset < data.len() {
197 let end = (offset + self.chunk_size).min(data.len());
198 let chunk = &data[offset..end];
199 let chunk_size = chunk.len() as u64;
200 let (hash, key) = self.put_chunk_internal(chunk).await?;
201 links.push(Link {
202 hash,
203 name: None,
204 size: chunk_size,
205 key,
206 link_type: LinkType::Blob, meta: None,
208 });
209 offset = end;
210 }
211
212 let (root_hash, root_key, _) = self.build_tree_internal(links, Some(size)).await?;
214 Ok((
215 Cid {
216 hash: root_hash,
217 key: root_key,
218 },
219 size,
220 ))
221 }
222
223 pub async fn get(
228 &self,
229 cid: &Cid,
230 max_size: Option<u64>,
231 ) -> Result<Option<Vec<u8>>, HashTreeError> {
232 if let Some(key) = cid.key {
233 self.get_encrypted(&cid.hash, &key, max_size).await
234 } else {
235 self.read_file_with_limit(&cid.hash, max_size).await
236 }
237 }
238
239 pub async fn put_stream<R: AsyncRead + Unpin>(
245 &self,
246 reader: R,
247 ) -> Result<(Cid, u64), HashTreeError> {
248 self.put_stream_with_progress(reader, |_| {}).await
249 }
250
251 pub async fn put_stream_with_progress<R, F>(
253 &self,
254 mut reader: R,
255 mut progress: F,
256 ) -> Result<(Cid, u64), HashTreeError>
257 where
258 R: AsyncRead + Unpin,
259 F: FnMut(u64),
260 {
261 let mut links = Vec::new();
262 let mut total_size: u64 = 0;
263 let mut consistent_key: Option<ObjectKey> = None;
264 let mut pending_store_items = Vec::new();
265 let mut pending_store_bytes = 0usize;
266 let batch_target_bytes = stream_put_batch_target_bytes();
267 let mut chunk_capacity = self.chunk_size.min(STREAM_PUT_INITIAL_CHUNK_CAPACITY);
268
269 loop {
270 let mut chunk = Vec::with_capacity(chunk_capacity);
273 (&mut reader)
274 .take(self.chunk_size as u64)
275 .read_to_end(&mut chunk)
276 .await
277 .map_err(|e| HashTreeError::Store(format!("read error: {}", e)))?;
278
279 if chunk.is_empty() {
280 break; }
282 if chunk.len() == self.chunk_size {
283 chunk_capacity = self.chunk_size;
285 }
286
287 let chunk_len = chunk.len() as u64;
288 total_size += chunk_len;
289
290 let (hash, data, key) = self.prepare_chunk_for_store(&chunk)?;
291
292 if links.is_empty() {
294 consistent_key = key;
295 } else if consistent_key != key {
296 consistent_key = None;
297 }
298
299 links.push(Link {
300 hash,
301 name: None,
302 size: chunk_len,
303 key,
304 link_type: LinkType::Blob, meta: None,
306 });
307
308 pending_store_bytes = pending_store_bytes.saturating_add(data.len());
309 pending_store_items.push((hash, data, chunk_len));
310 if pending_store_bytes >= batch_target_bytes
311 || pending_store_items.len() >= STREAM_PUT_BATCH_MAX_ITEMS
312 {
313 Self::flush_stream_store_batch(
314 self.store.as_ref(),
315 &mut pending_store_items,
316 &mut pending_store_bytes,
317 &mut progress,
318 )
319 .await?;
320 }
321 }
322
323 if links.is_empty() {
324 let (hash, key) = self.put_chunk_internal(&[]).await?;
326 return Ok((Cid { hash, key }, 0));
327 }
328
329 Self::flush_stream_store_batch(
330 self.store.as_ref(),
331 &mut pending_store_items,
332 &mut pending_store_bytes,
333 &mut progress,
334 )
335 .await?;
336
337 let (root_hash, root_key, _) = self.build_tree_internal(links, Some(total_size)).await?;
339 Ok((
340 Cid {
341 hash: root_hash,
342 key: root_key,
343 },
344 total_size,
345 ))
346 }
347
348 async fn put_chunk_internal(
350 &self,
351 data: &[u8],
352 ) -> Result<(Hash, Option<ObjectKey>), HashTreeError> {
353 let (hash, stored_data, key) = self.prepare_chunk_for_store(data)?;
354 self.store
355 .put(hash, stored_data)
356 .await
357 .map_err(|e| HashTreeError::Store(e.to_string()))?;
358 Ok((hash, key))
359 }
360
361 fn prepare_chunk_for_store(
362 &self,
363 data: &[u8],
364 ) -> Result<(Hash, Vec<u8>, Option<ObjectKey>), HashTreeError> {
365 if self.encrypted {
366 let (encrypted, key) = crate::encryption::encrypt_object(data, self.encryption_mode)
367 .map_err(|e| HashTreeError::Encryption(e.to_string()))?;
368 let hash = sha256(&encrypted);
369 Ok((hash, encrypted, Some(key)))
370 } else {
371 let hash = sha256(data);
372 Ok((hash, data.to_vec(), None))
373 }
374 }
375
376 async fn flush_stream_store_batch<F>(
377 store: &S,
378 pending: &mut Vec<(Hash, Vec<u8>, u64)>,
379 pending_bytes: &mut usize,
380 progress: &mut F,
381 ) -> Result<(), HashTreeError>
382 where
383 F: FnMut(u64),
384 {
385 if pending.is_empty() {
386 return Ok(());
387 }
388
389 let mut items = Vec::with_capacity(pending.len());
390 let mut sizes = Vec::with_capacity(pending.len());
391 for (hash, data, size) in pending.drain(..) {
392 items.push((hash, data));
393 sizes.push(size);
394 }
395 *pending_bytes = 0;
396
397 store
398 .put_many(items)
399 .await
400 .map_err(|e| HashTreeError::Store(e.to_string()))?;
401 for size in sizes {
402 progress(size);
403 }
404 Ok(())
405 }
406
407 async fn build_tree_internal(
409 &self,
410 links: Vec<Link>,
411 total_size: Option<u64>,
412 ) -> Result<(Hash, Option<ObjectKey>, LinkType), HashTreeError> {
413 if links.len() == 1 {
415 if let Some(ts) = total_size {
416 if links[0].size == ts {
417 return Ok((links[0].hash, links[0].key, links[0].link_type));
418 }
419 }
420 }
421
422 if links.len() <= self.max_links {
423 let node = TreeNode {
424 node_type: LinkType::File,
425 links,
426 };
427 let (data, _) = encode_and_hash(&node)?;
428
429 if self.encrypted {
430 let (encrypted, key) =
431 crate::encryption::encrypt_object(&data, self.encryption_mode)
432 .map_err(|e| HashTreeError::Encryption(e.to_string()))?;
433 let hash = sha256(&encrypted);
434 self.store
435 .put(hash, encrypted)
436 .await
437 .map_err(|e| HashTreeError::Store(e.to_string()))?;
438 return Ok((hash, Some(key), LinkType::File));
439 }
440
441 let hash = sha256(&data);
443 self.store
444 .put(hash, data)
445 .await
446 .map_err(|e| HashTreeError::Store(e.to_string()))?;
447 return Ok((hash, None, LinkType::File));
448 }
449
450 let mut sub_links = Vec::new();
452 for batch in links.chunks(self.max_links) {
453 let batch_size: u64 = batch.iter().map(|l| l.size).sum();
454 let (hash, key, link_type) =
455 Box::pin(self.build_tree_internal(batch.to_vec(), Some(batch_size))).await?;
456 sub_links.push(Link {
457 hash,
458 name: None,
459 size: batch_size,
460 key,
461 link_type,
462 meta: None,
463 });
464 }
465
466 Box::pin(self.build_tree_internal(sub_links, total_size)).await
467 }
468
469 async fn get_encrypted(
471 &self,
472 hash: &Hash,
473 key: &ObjectKey,
474 max_size: Option<u64>,
475 ) -> Result<Option<Vec<u8>>, HashTreeError> {
476 let decrypted = match self.get_encrypted_root(hash, key).await? {
477 Some(data) => data,
478 None => return Ok(None),
479 };
480
481 if is_tree_node(&decrypted) {
483 let node = decode_tree_node(&decrypted)?;
484 let declared_size: u64 = node.links.iter().map(|l| l.size).sum();
485 Self::ensure_size_limit(max_size, declared_size)?;
486
487 let mut bytes_read = 0u64;
488 let assembled = self
489 .assemble_chunks_limited(&node, max_size, &mut bytes_read)
490 .await?;
491 return Ok(Some(assembled));
492 }
493
494 Self::ensure_size_limit(max_size, decrypted.len() as u64)?;
496 Ok(Some(decrypted))
497 }
498
499 async fn get_encrypted_root(
500 &self,
501 hash: &Hash,
502 key: &ObjectKey,
503 ) -> Result<Option<Vec<u8>>, HashTreeError> {
504 self.get_cid_root_bytes(&Cid {
505 hash: *hash,
506 key: Some(*key),
507 })
508 .await
509 .map_err(|err| match err {
510 HashTreeError::Decryption(message) => HashTreeError::Encryption(message),
511 other => other,
512 })
513 }
514
515 fn ensure_size_limit(max_size: Option<u64>, actual_size: u64) -> Result<(), HashTreeError> {
516 if let Some(max_size) = max_size {
517 if actual_size > max_size {
518 return Err(HashTreeError::SizeLimitExceeded {
519 max_size,
520 actual_size,
521 });
522 }
523 }
524 Ok(())
525 }
526
527 fn decode_linked_file_node(
528 link: &Link,
529 data: &[u8],
530 ) -> Result<Option<TreeNode>, HashTreeError> {
531 match link.link_type {
532 LinkType::File => match decode_tree_node(data) {
533 Ok(node) => Ok(Some(node)),
534 Err(_) if link.size == data.len() as u64 => Ok(None),
535 Err(err) => Err(HashTreeError::Codec(err)),
536 },
537 LinkType::Blob => {
538 if link.size == data.len() as u64 {
539 return Ok(None);
540 }
541
542 match decode_tree_node(data) {
543 Ok(node) if node.node_type == LinkType::File => Ok(Some(node)),
544 _ => Ok(None),
545 }
546 }
547 LinkType::Dir | LinkType::Fanout => Ok(None),
548 }
549 }
550
551 async fn read_link_bytes(&self, link: &Link) -> Result<Vec<u8>, HashTreeError> {
552 let stored = self
553 .get_blob(&link.hash)
554 .await?
555 .ok_or_else(|| HashTreeError::MissingChunk(to_hex(&link.hash)))?;
556 match &link.key {
557 Some(key) => crate::encryption::decrypt_object(&stored, key)
558 .map_err(|e| HashTreeError::Decryption(e.to_string())),
559 None => Ok(stored),
560 }
561 }
562
563 pub async fn put_blob(&self, data: &[u8]) -> Result<Hash, HashTreeError> {
568 let hash = sha256(data);
569 self.store
570 .put(hash, data.to_vec())
571 .await
572 .map_err(|e| HashTreeError::Store(e.to_string()))?;
573 Ok(hash)
574 }
575
576 pub async fn put_file(&self, data: &[u8]) -> Result<(Cid, u64), HashTreeError> {
579 self.put(data).await
580 }
581
582 pub async fn put_directory(&self, entries: Vec<DirEntry>) -> Result<Cid, HashTreeError> {
587 let mut sorted = entries;
589 sorted.sort_by(|a, b| a.name.cmp(&b.name));
590
591 let links: Vec<Link> = sorted
592 .into_iter()
593 .map(|e| Link {
594 hash: e.hash,
595 name: Some(e.name),
596 size: e.size,
597 key: e.key,
598 link_type: e.link_type,
599 meta: e.meta,
600 })
601 .collect();
602
603 if links.len() <= self.max_links {
604 return self.put_directory_node(links).await;
605 }
606
607 self.build_directory_by_chunks(links).await
608 }
609
610 async fn put_directory_node(&self, links: Vec<Link>) -> Result<Cid, HashTreeError> {
611 self.put_tree_node_with_type(LinkType::Dir, links).await
612 }
613
614 async fn put_tree_node_with_type(
615 &self,
616 node_type: LinkType,
617 links: Vec<Link>,
618 ) -> Result<Cid, HashTreeError> {
619 let node = TreeNode { node_type, links };
620 let (data, plain_hash) = encode_and_hash(&node)?;
621
622 if self.encrypted {
623 let (encrypted, key) = crate::encryption::encrypt_object(&data, self.encryption_mode)
624 .map_err(|e| HashTreeError::Encryption(e.to_string()))?;
625 let hash = sha256(&encrypted);
626 self.store
627 .put(hash, encrypted)
628 .await
629 .map_err(|e| HashTreeError::Store(e.to_string()))?;
630 return Ok(Cid {
631 hash,
632 key: Some(key),
633 });
634 }
635
636 self.store
637 .put(plain_hash, data)
638 .await
639 .map_err(|e| HashTreeError::Store(e.to_string()))?;
640 Ok(Cid {
641 hash: plain_hash,
642 key: None,
643 })
644 }
645
646 async fn build_directory_by_chunks(&self, links: Vec<Link>) -> Result<Cid, HashTreeError> {
647 let spans = links
648 .into_iter()
649 .map(|link| {
650 let name = link.name.clone().unwrap_or_else(|| to_hex(&link.hash));
651 DirectoryFanoutSpan {
652 link,
653 count: 1,
654 first: name.clone(),
655 last: name,
656 }
657 })
658 .collect();
659 self.build_directory_fanout_level(spans, LinkType::Dir)
660 .await
661 }
662
663 async fn build_directory_fanout_level(
664 &self,
665 spans: Vec<DirectoryFanoutSpan>,
666 child_node_type: LinkType,
667 ) -> Result<Cid, HashTreeError> {
668 let mut sub_trees: Vec<DirectoryFanoutSpan> = Vec::new();
669
670 for batch in spans.chunks(self.max_links) {
671 let Some(first_span) = batch.first() else {
672 continue;
673 };
674 let last_span = batch.last().expect("non-empty fanout batch");
675 let count: usize = batch.iter().map(|span| span.count).sum();
676 let batch_size: u64 = batch.iter().map(|span| span.link.size).sum();
677 let child_links = batch.iter().map(|span| span.link.clone()).collect();
678 let child_cid = self
679 .put_tree_node_with_type(child_node_type, child_links)
680 .await?;
681
682 sub_trees.push(DirectoryFanoutSpan {
683 link: Link {
684 hash: child_cid.hash,
685 name: None,
686 size: batch_size,
687 key: child_cid.key,
688 link_type: child_node_type,
689 meta: Some(directory_fanout_meta(
690 count,
691 &first_span.first,
692 &last_span.last,
693 )),
694 },
695 count,
696 first: first_span.first.clone(),
697 last: last_span.last.clone(),
698 });
699 }
700
701 if sub_trees.len() <= self.max_links {
702 return self
703 .put_tree_node_with_type(
704 LinkType::Fanout,
705 sub_trees.into_iter().map(|span| span.link).collect(),
706 )
707 .await;
708 }
709
710 Box::pin(self.build_directory_fanout_level(sub_trees, LinkType::Fanout)).await
711 }
712
713 pub async fn put_tree_node(&self, links: Vec<Link>) -> Result<Hash, HashTreeError> {
715 let node = TreeNode {
716 node_type: LinkType::Dir,
717 links,
718 };
719
720 let (data, hash) = encode_and_hash(&node)?;
721 self.store
722 .put(hash, data)
723 .await
724 .map_err(|e| HashTreeError::Store(e.to_string()))?;
725 Ok(hash)
726 }
727
728 pub async fn get_blob(&self, hash: &Hash) -> Result<Option<Vec<u8>>, HashTreeError> {
732 self.store
733 .get(hash)
734 .await
735 .map_err(|e| HashTreeError::Store(e.to_string()))
736 }
737
738 async fn has_stored_chunk(&self, hash: &Hash) -> Result<bool, HashTreeError> {
739 self.store
740 .get(hash)
741 .await
742 .map(|data| data.is_some())
743 .map_err(|e| HashTreeError::Store(e.to_string()))
744 }
745
746 pub async fn get_tree_node(&self, hash: &Hash) -> Result<Option<TreeNode>, HashTreeError> {
748 let data = match self
749 .store
750 .get(hash)
751 .await
752 .map_err(|e| HashTreeError::Store(e.to_string()))?
753 {
754 Some(d) => d,
755 None => return Ok(None),
756 };
757
758 if !is_tree_node(&data) {
759 return Ok(None);
760 }
761
762 let node = decode_tree_node(&data)?;
763 Ok(Some(node))
764 }
765
766 pub async fn get_tree_node_by_cid(&self, cid: &Cid) -> Result<Option<TreeNode>, HashTreeError> {
768 self.get_node(cid).await
769 }
770
771 async fn get_cid_root_bytes(&self, cid: &Cid) -> Result<Option<Vec<u8>>, HashTreeError> {
772 let data = match self
773 .store
774 .get(&cid.hash)
775 .await
776 .map_err(|e| HashTreeError::Store(e.to_string()))?
777 {
778 Some(d) => d,
779 None => return Ok(None),
780 };
781
782 decrypt_cid_root_bytes(cid, data).map(Some)
783 }
784
785 pub async fn get_node(&self, cid: &Cid) -> Result<Option<TreeNode>, HashTreeError> {
787 let data = match self.get_blob(&cid.hash).await? {
788 Some(d) => d,
789 None => return Ok(None),
790 };
791 decode_tree_node_by_cid(cid, data)
792 }
793
794 pub async fn get_directory_node(&self, cid: &Cid) -> Result<Option<TreeNode>, HashTreeError> {
797 let decrypted = match self.get_cid_root_bytes(cid).await? {
798 Some(d) => d,
799 None => return Ok(None),
800 };
801
802 if !is_tree_node(&decrypted) {
803 return Ok(None);
804 }
805
806 let node = decode_tree_node(&decrypted)?;
807
808 if node.node_type == LinkType::File {
810 let mut bytes_read = 0u64;
811 let assembled = self
812 .assemble_chunks_limited(&node, None, &mut bytes_read)
813 .await?;
814 if is_tree_node(&assembled) {
815 let inner_node = decode_tree_node(&assembled)?;
816 return Ok(Some(inner_node));
817 }
818 }
819
820 Ok(Some(node))
821 }
822
823 pub async fn is_tree(&self, hash: &Hash) -> Result<bool, HashTreeError> {
825 let data = match self
826 .store
827 .get(hash)
828 .await
829 .map_err(|e| HashTreeError::Store(e.to_string()))?
830 {
831 Some(d) => d,
832 None => return Ok(false),
833 };
834 Ok(is_tree_node(&data))
835 }
836
837 pub async fn is_dir(&self, cid: &Cid) -> Result<bool, HashTreeError> {
839 Ok(matches!(
840 self.get_directory_node(cid).await?,
841 Some(node) if node.node_type.is_directory_like()
842 ))
843 }
844
845 pub async fn is_directory(&self, hash: &Hash) -> Result<bool, HashTreeError> {
847 let data = match self
848 .store
849 .get(hash)
850 .await
851 .map_err(|e| HashTreeError::Store(e.to_string()))?
852 {
853 Some(d) => d,
854 None => return Ok(false),
855 };
856 Ok(is_directory_node(&data))
857 }
858
859 pub async fn read_file(&self, hash: &Hash) -> Result<Option<Vec<u8>>, HashTreeError> {
861 self.read_file_with_limit(hash, None).await
862 }
863
864 async fn read_file_with_limit(
866 &self,
867 hash: &Hash,
868 max_size: Option<u64>,
869 ) -> Result<Option<Vec<u8>>, HashTreeError> {
870 let data = match self
871 .store
872 .get(hash)
873 .await
874 .map_err(|e| HashTreeError::Store(e.to_string()))?
875 {
876 Some(d) => d,
877 None => return Ok(None),
878 };
879
880 if !is_tree_node(&data) {
882 Self::ensure_size_limit(max_size, data.len() as u64)?;
883 return Ok(Some(data));
884 }
885
886 let node = decode_tree_node(&data)?;
888 let declared_size: u64 = node.links.iter().map(|l| l.size).sum();
889 Self::ensure_size_limit(max_size, declared_size)?;
890
891 let mut bytes_read = 0u64;
892 let assembled = self
893 .assemble_chunks_limited(&node, max_size, &mut bytes_read)
894 .await?;
895 Ok(Some(assembled))
896 }
897
898 pub async fn read_file_range(
905 &self,
906 hash: &Hash,
907 start: u64,
908 end: Option<u64>,
909 ) -> Result<Option<Vec<u8>>, HashTreeError> {
910 self.read_file_range_cid(&Cid::public(*hash), start, end)
911 .await
912 }
913
914 pub async fn read_file_range_cid(
916 &self,
917 cid: &Cid,
918 start: u64,
919 end: Option<u64>,
920 ) -> Result<Option<Vec<u8>>, HashTreeError> {
921 let Some(data) = self.get_cid_root_bytes(cid).await? else {
922 return Ok(None);
923 };
924 if !is_tree_node(&data) {
925 let end = end.unwrap_or(data.len() as u64).min(data.len() as u64);
926 return Ok(Some(data[start.min(end) as usize..end as usize].to_vec()));
927 }
928 let node = decode_tree_node(&data)?;
929 let mut result = Vec::new();
930 self.append_range(&node, start, end.unwrap_or(u64::MAX), 0, &mut result)
931 .await?;
932 Ok(Some(result))
933 }
934
935 pub async fn read_file_chunk_at(
943 &self,
944 cid: &Cid,
945 offset: u64,
946 ) -> Result<Option<(u64, Vec<u8>)>, HashTreeError> {
947 let Some(data) = self.get_cid_root_bytes(cid).await? else {
948 return Ok(None);
949 };
950 if !is_tree_node(&data) {
951 return Ok(Some((0, data)));
952 }
953
954 let mut node = decode_tree_node(&data)?;
955 let mut base_offset = 0u64;
956 'descend: loop {
957 let mut child_start = base_offset;
958 for link in &node.links {
959 let child_end = child_start.saturating_add(link.size);
960 if offset >= child_end {
961 child_start = child_end;
962 continue;
963 }
964
965 let child_data = self.read_link_bytes(link).await?;
966 match Self::decode_linked_file_node(link, &child_data)? {
967 Some(child_node) => {
968 node = child_node;
969 base_offset = child_start;
970 continue 'descend;
971 }
972 None => return Ok(Some((child_start, child_data))),
973 }
974 }
975 return Ok(Some((child_start, Vec::new())));
977 }
978 }
979
980 async fn append_range(
981 &self,
982 node: &TreeNode,
983 start: u64,
984 end: u64,
985 base_offset: u64,
986 result: &mut Vec<u8>,
987 ) -> Result<(), HashTreeError> {
988 let mut current_offset = base_offset;
989
990 for link in &node.links {
991 let child_start = current_offset;
992 let child_end = child_start.saturating_add(link.size);
993 current_offset = child_end;
994
995 if child_end <= start {
996 continue;
997 }
998 if child_start >= end {
999 break;
1000 }
1001
1002 let from = start.saturating_sub(child_start);
1003 let to = end.min(child_end) - child_start;
1004 if from >= to {
1005 continue;
1006 }
1007
1008 if link.link_type == LinkType::Blob && link.key.is_none() {
1009 let bytes = self
1010 .store
1011 .get_range(&link.hash, from, to - 1)
1012 .await
1013 .map_err(|e| HashTreeError::Store(e.to_string()))?
1014 .ok_or_else(|| HashTreeError::MissingChunk(to_hex(&link.hash)))?;
1015 if bytes.len() as u64 != to - from {
1016 return Err(HashTreeError::Store(format!(
1017 "unexpected range length for {}",
1018 to_hex(&link.hash)
1019 )));
1020 }
1021 result.extend_from_slice(&bytes);
1022 continue;
1023 }
1024
1025 let decrypted_child = self.read_link_bytes(link).await?;
1026 if let Some(child_node) = Self::decode_linked_file_node(link, &decrypted_child)? {
1027 Box::pin(self.append_range(&child_node, start, end, child_start, result)).await?;
1028 continue;
1029 }
1030
1031 let bytes = decrypted_child
1032 .get(from as usize..to as usize)
1033 .ok_or_else(|| {
1034 HashTreeError::Store(format!("invalid chunk size for {}", to_hex(&link.hash)))
1035 })?;
1036 result.extend_from_slice(bytes);
1037 }
1038
1039 Ok(())
1040 }
1041
1042 async fn assemble_chunks_limited(
1044 &self,
1045 node: &TreeNode,
1046 max_size: Option<u64>,
1047 bytes_read: &mut u64,
1048 ) -> Result<Vec<u8>, HashTreeError> {
1049 let mut out = Vec::new();
1050 self.assemble_chunks_into(node, max_size, bytes_read, &mut out)
1051 .await?;
1052 Ok(out)
1053 }
1054
1055 async fn assemble_chunks_into(
1057 &self,
1058 node: &TreeNode,
1059 max_size: Option<u64>,
1060 bytes_read: &mut u64,
1061 out: &mut Vec<u8>,
1062 ) -> Result<(), HashTreeError> {
1063 for link in &node.links {
1064 let projected = (*bytes_read).saturating_add(link.size);
1065 Self::ensure_size_limit(max_size, projected)?;
1066
1067 let child_data = self.read_link_bytes(link).await?;
1068
1069 if let Some(child_node) = Self::decode_linked_file_node(link, &child_data)? {
1070 Box::pin(self.assemble_chunks_into(&child_node, max_size, bytes_read, out)).await?;
1071 } else {
1072 let projected = (*bytes_read).saturating_add(child_data.len() as u64);
1073 Self::ensure_size_limit(max_size, projected)?;
1074 *bytes_read = projected;
1075 if out.is_empty() {
1076 let declared: u64 = node.links.iter().map(|l| l.size).sum();
1079 let extra = declared
1080 .saturating_sub(child_data.len() as u64)
1081 .min(ASSEMBLE_MAX_RESERVE_BYTES);
1082 *out = child_data;
1083 out.reserve(extra as usize);
1084 } else {
1085 out.extend_from_slice(&child_data);
1086 }
1087 }
1088 }
1089 Ok(())
1090 }
1091
1092 pub async fn read_file_chunks(&self, hash: &Hash) -> Result<Vec<Vec<u8>>, HashTreeError> {
1094 let data = match self
1095 .store
1096 .get(hash)
1097 .await
1098 .map_err(|e| HashTreeError::Store(e.to_string()))?
1099 {
1100 Some(d) => d,
1101 None => return Ok(vec![]),
1102 };
1103
1104 if !is_tree_node(&data) {
1105 return Ok(vec![data]);
1106 }
1107
1108 let node = decode_tree_node(&data)?;
1109 self.collect_chunks(&node).await
1110 }
1111
1112 async fn collect_chunks(&self, node: &TreeNode) -> Result<Vec<Vec<u8>>, HashTreeError> {
1113 let mut chunks = Vec::new();
1114
1115 for link in &node.links {
1116 let child_data = self.read_link_bytes(link).await?;
1117
1118 if let Some(child_node) = Self::decode_linked_file_node(link, &child_data)? {
1119 chunks.extend(Box::pin(self.collect_chunks(&child_node)).await?);
1120 } else {
1121 chunks.push(child_data);
1122 }
1123 }
1124
1125 Ok(chunks)
1126 }
1127
1128 pub async fn list(&self, cid: &Cid) -> Result<Vec<TreeEntry>, HashTreeError> {
1130 let node = match self.get_node(cid).await? {
1131 Some(n) => n,
1132 None => return Ok(vec![]),
1133 };
1134
1135 let mut entries = Vec::new();
1136
1137 for link in &node.links {
1138 if is_internal_directory_link(&node, link) {
1140 let chunk_cid = Cid {
1141 hash: link.hash,
1142 key: link.key,
1143 };
1144 let sub_entries = Box::pin(self.list(&chunk_cid)).await?;
1145 entries.extend(sub_entries);
1146 continue;
1147 }
1148
1149 entries.push(TreeEntry {
1150 name: link.name.clone().unwrap_or_else(|| to_hex(&link.hash)),
1151 hash: link.hash,
1152 size: link.size,
1153 link_type: link.link_type,
1154 key: link.key,
1155 meta: link.meta.clone(),
1156 });
1157 }
1158
1159 Ok(entries)
1160 }
1161
1162 pub async fn list_directory(&self, cid: &Cid) -> Result<Vec<TreeEntry>, HashTreeError> {
1165 let node = match self.get_directory_node(cid).await? {
1167 Some(n) => n,
1168 None => return Ok(vec![]),
1169 };
1170
1171 let mut entries = Vec::new();
1172
1173 for link in &node.links {
1174 if is_internal_directory_link(&node, link) {
1176 let sub_cid = Cid {
1177 hash: link.hash,
1178 key: link.key,
1179 };
1180 let sub_entries = Box::pin(self.list_directory(&sub_cid)).await?;
1181 entries.extend(sub_entries);
1182 continue;
1183 }
1184
1185 entries.push(TreeEntry {
1186 name: link.name.clone().unwrap_or_else(|| to_hex(&link.hash)),
1187 hash: link.hash,
1188 size: link.size,
1189 link_type: link.link_type,
1190 key: link.key,
1191 meta: link.meta.clone(),
1192 });
1193 }
1194
1195 Ok(entries)
1196 }
1197
1198 pub async fn list_directory_required(
1201 &self,
1202 cid: &Cid,
1203 ) -> Result<Vec<TreeEntry>, HashTreeError> {
1204 let node = self
1205 .get_directory_node(cid)
1206 .await?
1207 .ok_or_else(|| HashTreeError::MissingChunk(to_hex(&cid.hash)))?;
1208
1209 let mut entries = Vec::new();
1210 for link in &node.links {
1211 if is_internal_directory_link(&node, link) {
1212 let sub_cid = Cid {
1213 hash: link.hash,
1214 key: link.key,
1215 };
1216 let sub_entries = Box::pin(self.list_directory_required(&sub_cid)).await?;
1217 entries.extend(sub_entries);
1218 continue;
1219 }
1220
1221 entries.push(TreeEntry {
1222 name: link.name.clone().unwrap_or_else(|| to_hex(&link.hash)),
1223 hash: link.hash,
1224 size: link.size,
1225 link_type: link.link_type,
1226 key: link.key,
1227 meta: link.meta.clone(),
1228 });
1229 }
1230
1231 Ok(entries)
1232 }
1233
1234 pub async fn resolve(&self, cid: &Cid, path: &str) -> Result<Option<Cid>, HashTreeError> {
1236 let parts: Vec<&str> = path.split('/').filter(|p| !p.is_empty()).collect();
1237 if parts.is_empty() {
1238 return Ok(Some(cid.clone()));
1239 }
1240
1241 let mut current_cid = cid.clone();
1242
1243 for part in parts {
1244 let node = match self.get_directory_node(¤t_cid).await? {
1246 Some(n) => n,
1247 None => {
1248 if !self.has_stored_chunk(¤t_cid.hash).await? {
1249 return Err(HashTreeError::MissingChunk(to_hex(¤t_cid.hash)));
1250 }
1251 return Ok(None);
1252 }
1253 };
1254
1255 if let Some(link) = self.find_link(&node, part) {
1256 current_cid = Cid {
1257 hash: link.hash,
1258 key: link.key,
1259 };
1260 } else {
1261 match self
1263 .find_link_in_subtrees_cid(&node, part, ¤t_cid)
1264 .await?
1265 {
1266 Some(link) => {
1267 current_cid = Cid {
1268 hash: link.hash,
1269 key: link.key,
1270 };
1271 }
1272 None => return Ok(None),
1273 }
1274 }
1275 }
1276
1277 Ok(Some(current_cid))
1278 }
1279
1280 pub async fn resolve_path(&self, cid: &Cid, path: &str) -> Result<Option<Cid>, HashTreeError> {
1282 self.resolve(cid, path).await
1283 }
1284
1285 fn find_link(&self, node: &TreeNode, name: &str) -> Option<Link> {
1286 node.links
1287 .iter()
1288 .find(|l| !is_internal_directory_link(node, l) && l.name.as_deref() == Some(name))
1289 .cloned()
1290 }
1291
1292 async fn find_link_in_subtrees_cid(
1294 &self,
1295 node: &TreeNode,
1296 name: &str,
1297 _parent_cid: &Cid,
1298 ) -> Result<Option<Link>, HashTreeError> {
1299 for link in &node.links {
1300 if !is_internal_directory_link(node, link) {
1301 continue;
1302 }
1303
1304 let sub_cid = Cid {
1306 hash: link.hash,
1307 key: link.key,
1308 };
1309
1310 let sub_node = match self.get_node(&sub_cid).await? {
1311 Some(n) => n,
1312 None => {
1313 if !self.has_stored_chunk(&sub_cid.hash).await? {
1314 return Err(HashTreeError::MissingChunk(to_hex(&sub_cid.hash)));
1315 }
1316 continue;
1317 }
1318 };
1319
1320 if let Some(found) = self.find_link(&sub_node, name) {
1321 return Ok(Some(found));
1322 }
1323
1324 if let Some(deep_found) =
1325 Box::pin(self.find_link_in_subtrees_cid(&sub_node, name, &sub_cid)).await?
1326 {
1327 return Ok(Some(deep_found));
1328 }
1329 }
1330
1331 Ok(None)
1332 }
1333
1334 pub async fn get_size(&self, hash: &Hash) -> Result<u64, HashTreeError> {
1336 let data = match self
1337 .store
1338 .get(hash)
1339 .await
1340 .map_err(|e| HashTreeError::Store(e.to_string()))?
1341 {
1342 Some(d) => d,
1343 None => return Ok(0),
1344 };
1345
1346 if !is_tree_node(&data) {
1347 return Ok(data.len() as u64);
1348 }
1349
1350 let node = decode_tree_node(&data)?;
1351 let mut total = 0u64;
1353 for link in &node.links {
1354 total += link.size;
1355 }
1356 Ok(total)
1357 }
1358
1359 pub async fn get_size_cid(&self, cid: &Cid) -> Result<u64, HashTreeError> {
1361 if let Some(key) = cid.key {
1362 let data = match self.get_encrypted_root(&cid.hash, &key).await? {
1363 Some(d) => d,
1364 None => return Ok(0),
1365 };
1366 if is_tree_node(&data) {
1367 let node = decode_tree_node(&data)?;
1368 return Ok(node.links.iter().map(|link| link.size).sum());
1369 }
1370 return Ok(data.len() as u64);
1371 }
1372
1373 self.get_size(&cid.hash).await
1374 }
1375
1376 async fn put_edited_directory(
1380 &self,
1381 original: &Cid,
1382 entries: Vec<DirEntry>,
1383 ) -> Result<Cid, HashTreeError> {
1384 if original
1385 .key
1386 .is_some_and(|key| key.mode() == crate::encryption::EncryptionMode::RndV1)
1387 {
1388 let writer = Self {
1389 store: self.store.clone(),
1390 chunk_size: self.chunk_size,
1391 max_links: self.max_links,
1392 encrypted: true,
1393 encryption_mode: crate::encryption::EncryptionMode::RndV1,
1394 };
1395 writer.put_directory(entries).await
1396 } else {
1397 self.put_directory(entries).await
1398 }
1399 }
1400
1401 pub async fn set_entry(
1404 &self,
1405 root: &Cid,
1406 path: &[&str],
1407 name: &str,
1408 entry_cid: &Cid,
1409 size: u64,
1410 link_type: LinkType,
1411 ) -> Result<Cid, HashTreeError> {
1412 self.set_entry_with_meta(root, path, name, entry_cid, size, link_type, None)
1413 .await
1414 }
1415
1416 #[allow(clippy::too_many_arguments)]
1419 pub async fn set_entry_with_meta(
1420 &self,
1421 root: &Cid,
1422 path: &[&str],
1423 name: &str,
1424 entry_cid: &Cid,
1425 size: u64,
1426 link_type: LinkType,
1427 meta: Option<std::collections::HashMap<String, serde_json::Value>>,
1428 ) -> Result<Cid, HashTreeError> {
1429 let dir_cid = self.resolve_path_array(root, path).await?;
1430 let dir_cid = dir_cid.ok_or_else(|| HashTreeError::PathNotFound(path.join("/")))?;
1431
1432 let entries = self.list_directory(&dir_cid).await?;
1433 let mut new_entries: Vec<DirEntry> = entries
1434 .into_iter()
1435 .filter(|e| e.name != name)
1436 .map(|e| DirEntry {
1437 name: e.name,
1438 hash: e.hash,
1439 size: e.size,
1440 key: e.key,
1441 link_type: e.link_type,
1442 meta: e.meta,
1443 })
1444 .collect();
1445
1446 new_entries.push(DirEntry {
1447 name: name.to_string(),
1448 hash: entry_cid.hash,
1449 size,
1450 key: entry_cid.key,
1451 link_type,
1452 meta,
1453 });
1454
1455 let new_dir_cid = self.put_edited_directory(&dir_cid, new_entries).await?;
1456 self.rebuild_path(root, path, new_dir_cid).await
1457 }
1458
1459 pub async fn remove_entry(
1462 &self,
1463 root: &Cid,
1464 path: &[&str],
1465 name: &str,
1466 ) -> Result<Cid, HashTreeError> {
1467 let dir_cid = self.resolve_path_array(root, path).await?;
1468 let dir_cid = dir_cid.ok_or_else(|| HashTreeError::PathNotFound(path.join("/")))?;
1469
1470 let entries = self.list_directory(&dir_cid).await?;
1471 let new_entries: Vec<DirEntry> = entries
1472 .into_iter()
1473 .filter(|e| e.name != name)
1474 .map(|e| DirEntry {
1475 name: e.name,
1476 hash: e.hash,
1477 size: e.size,
1478 key: e.key,
1479 link_type: e.link_type,
1480 meta: e.meta,
1481 })
1482 .collect();
1483
1484 let new_dir_cid = self.put_edited_directory(&dir_cid, new_entries).await?;
1485 self.rebuild_path(root, path, new_dir_cid).await
1486 }
1487
1488 pub async fn rename_entry(
1491 &self,
1492 root: &Cid,
1493 path: &[&str],
1494 old_name: &str,
1495 new_name: &str,
1496 ) -> Result<Cid, HashTreeError> {
1497 if old_name == new_name {
1498 return Ok(root.clone());
1499 }
1500
1501 let dir_cid = self.resolve_path_array(root, path).await?;
1502 let dir_cid = dir_cid.ok_or_else(|| HashTreeError::PathNotFound(path.join("/")))?;
1503
1504 let entries = self.list_directory(&dir_cid).await?;
1505 let entry = entries
1506 .iter()
1507 .find(|e| e.name == old_name)
1508 .ok_or_else(|| HashTreeError::EntryNotFound(old_name.to_string()))?;
1509
1510 let entry_hash = entry.hash;
1511 let entry_size = entry.size;
1512 let entry_key = entry.key;
1513 let entry_link_type = entry.link_type;
1514 let entry_meta = entry.meta.clone();
1515
1516 let new_entries: Vec<DirEntry> = entries
1517 .into_iter()
1518 .filter(|e| e.name != old_name)
1519 .map(|e| DirEntry {
1520 name: e.name,
1521 hash: e.hash,
1522 size: e.size,
1523 key: e.key,
1524 link_type: e.link_type,
1525 meta: e.meta,
1526 })
1527 .chain(std::iter::once(DirEntry {
1528 name: new_name.to_string(),
1529 hash: entry_hash,
1530 size: entry_size,
1531 key: entry_key,
1532 link_type: entry_link_type,
1533 meta: entry_meta,
1534 }))
1535 .collect();
1536
1537 let new_dir_cid = self.put_edited_directory(&dir_cid, new_entries).await?;
1538 self.rebuild_path(root, path, new_dir_cid).await
1539 }
1540
1541 pub async fn move_entry(
1544 &self,
1545 root: &Cid,
1546 source_path: &[&str],
1547 name: &str,
1548 target_path: &[&str],
1549 ) -> Result<Cid, HashTreeError> {
1550 let source_dir_cid = self.resolve_path_array(root, source_path).await?;
1551 let source_dir_cid =
1552 source_dir_cid.ok_or_else(|| HashTreeError::PathNotFound(source_path.join("/")))?;
1553
1554 let source_entries = self.list_directory(&source_dir_cid).await?;
1555 let entry = source_entries
1556 .iter()
1557 .find(|e| e.name == name)
1558 .ok_or_else(|| HashTreeError::EntryNotFound(name.to_string()))?;
1559
1560 let entry_cid = Cid {
1561 hash: entry.hash,
1562 key: entry.key,
1563 };
1564 let entry_size = entry.size;
1565 let entry_link_type = entry.link_type;
1566
1567 let new_root = self.remove_entry(root, source_path, name).await?;
1569
1570 self.set_entry(
1572 &new_root,
1573 target_path,
1574 name,
1575 &entry_cid,
1576 entry_size,
1577 entry_link_type,
1578 )
1579 .await
1580 }
1581
1582 async fn resolve_path_array(
1583 &self,
1584 root: &Cid,
1585 path: &[&str],
1586 ) -> Result<Option<Cid>, HashTreeError> {
1587 if path.is_empty() {
1588 return Ok(Some(root.clone()));
1589 }
1590 self.resolve_path(root, &path.join("/")).await
1591 }
1592
1593 async fn rebuild_path(
1594 &self,
1595 root: &Cid,
1596 path: &[&str],
1597 new_child: Cid,
1598 ) -> Result<Cid, HashTreeError> {
1599 if path.is_empty() {
1600 return Ok(new_child);
1601 }
1602
1603 let mut child_cid = new_child;
1604 let parts: Vec<&str> = path.to_vec();
1605
1606 for i in (0..parts.len()).rev() {
1607 let child_name = parts[i];
1608 let parent_path = &parts[..i];
1609
1610 let parent_cid = if parent_path.is_empty() {
1611 root.clone()
1612 } else {
1613 self.resolve_path_array(root, parent_path)
1614 .await?
1615 .ok_or_else(|| HashTreeError::PathNotFound(parent_path.join("/")))?
1616 };
1617
1618 let parent_entries = self.list_directory(&parent_cid).await?;
1619 let new_parent_entries: Vec<DirEntry> = parent_entries
1620 .into_iter()
1621 .map(|e| {
1622 if e.name == child_name {
1623 DirEntry {
1624 name: e.name,
1625 hash: child_cid.hash,
1626 size: 0, key: child_cid.key,
1628 link_type: e.link_type,
1629 meta: e.meta,
1630 }
1631 } else {
1632 DirEntry {
1633 name: e.name,
1634 hash: e.hash,
1635 size: e.size,
1636 key: e.key,
1637 link_type: e.link_type,
1638 meta: e.meta,
1639 }
1640 }
1641 })
1642 .collect();
1643
1644 child_cid = self
1645 .put_edited_directory(&parent_cid, new_parent_entries)
1646 .await?;
1647 }
1648
1649 Ok(child_cid)
1650 }
1651
1652 pub fn get_store(&self) -> Arc<S> {
1656 self.store.clone()
1657 }
1658
1659 pub fn chunk_size(&self) -> usize {
1661 self.chunk_size
1662 }
1663
1664 pub fn max_links(&self) -> usize {
1666 self.max_links
1667 }
1668}
1669
1670fn stream_put_batch_target_bytes() -> usize {
1671 std::env::var(STREAM_PUT_BATCH_TARGET_BYTES_ENV)
1672 .ok()
1673 .and_then(|value| value.parse::<usize>().ok())
1674 .filter(|value| *value > 0)
1675 .unwrap_or(DEFAULT_STREAM_PUT_BATCH_TARGET_BYTES)
1676}
1677
1678pub async fn verify_tree<S: Store>(
1680 store: Arc<S>,
1681 root_hash: &Hash,
1682) -> Result<crate::reader::VerifyResult, HashTreeError> {
1683 let mut missing = Vec::new();
1684 let mut visited = std::collections::HashSet::new();
1685
1686 verify_recursive(store, root_hash, &mut missing, &mut visited).await?;
1687
1688 Ok(crate::reader::VerifyResult {
1689 valid: missing.is_empty(),
1690 missing,
1691 })
1692}
1693
1694async fn verify_recursive<S: Store>(
1695 store: Arc<S>,
1696 hash: &Hash,
1697 missing: &mut Vec<Hash>,
1698 visited: &mut std::collections::HashSet<String>,
1699) -> Result<(), HashTreeError> {
1700 let hex = to_hex(hash);
1701 if visited.contains(&hex) {
1702 return Ok(());
1703 }
1704 visited.insert(hex);
1705
1706 let data = match store
1707 .get(hash)
1708 .await
1709 .map_err(|e| HashTreeError::Store(e.to_string()))?
1710 {
1711 Some(d) => d,
1712 None => {
1713 missing.push(*hash);
1714 return Ok(());
1715 }
1716 };
1717
1718 if is_tree_node(&data) {
1719 let node = decode_tree_node(&data)?;
1720 for link in &node.links {
1721 Box::pin(verify_recursive(
1722 store.clone(),
1723 &link.hash,
1724 missing,
1725 visited,
1726 ))
1727 .await?;
1728 }
1729 }
1730
1731 Ok(())
1732}
1733
1734#[cfg(test)]
1735mod tests;