1use async_trait::async_trait;
4use std::collections::HashMap;
5use std::sync::{Arc, RwLock};
6
7use crate::types::{to_hex, Hash};
8
9pub fn slice_blob_range(
14 data: &[u8],
15 start: u64,
16 end_inclusive: u64,
17) -> Result<Vec<u8>, StoreError> {
18 if data.is_empty() || end_inclusive < start {
19 return Ok(Vec::new());
20 }
21
22 let len = data.len() as u64;
23 if start >= len {
24 return Ok(Vec::new());
25 }
26
27 let actual_end = end_inclusive.min(len - 1);
28 let start = usize::try_from(start)
29 .map_err(|_| StoreError::Other("blob range start is too large".to_string()))?;
30 let end_exclusive = usize::try_from(actual_end.saturating_add(1))
31 .map_err(|_| StoreError::Other("blob range end is too large".to_string()))?;
32
33 Ok(data[start..end_exclusive].to_vec())
34}
35
36#[derive(Debug, Clone, Default)]
38pub struct StoreStats {
39 pub count: u64,
41 pub bytes: u64,
43 pub pinned_count: u64,
45 pub pinned_bytes: u64,
47}
48
49#[derive(Debug, Clone, Default, PartialEq, Eq)]
51pub struct PutManyReport {
52 pub total: usize,
54 pub inserted: usize,
56 pub inserted_bytes: u64,
58 pub inserted_hashes: Vec<Hash>,
60}
61
62#[async_trait]
64pub trait Store: Send + Sync {
65 async fn put(&self, hash: Hash, data: Vec<u8>) -> Result<bool, StoreError>;
68
69 async fn put_many(&self, items: Vec<(Hash, Vec<u8>)>) -> Result<usize, StoreError> {
72 let mut inserted = 0usize;
73 for (hash, data) in items {
74 if self.put(hash, data).await? {
75 inserted += 1;
76 }
77 }
78 Ok(inserted)
79 }
80
81 async fn put_many_optimistic(&self, items: Vec<(Hash, Vec<u8>)>) -> Result<usize, StoreError> {
88 self.put_many(items).await
89 }
90
91 async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError>;
94
95 async fn get_range(
101 &self,
102 hash: &Hash,
103 start: u64,
104 end_inclusive: u64,
105 ) -> Result<Option<Vec<u8>>, StoreError> {
106 let Some(data) = self.get(hash).await? else {
107 return Ok(None);
108 };
109 Ok(Some(slice_blob_range(&data, start, end_inclusive)?))
110 }
111
112 async fn blob_size(&self, hash: &Hash) -> Result<Option<u64>, StoreError> {
117 Ok(self.get(hash).await?.map(|data| data.len() as u64))
118 }
119
120 async fn has(&self, hash: &Hash) -> Result<bool, StoreError>;
122
123 async fn delete(&self, hash: &Hash) -> Result<bool, StoreError>;
126
127 fn set_max_bytes(&self, _max: u64) {}
133
134 fn max_bytes(&self) -> Option<u64> {
136 None
137 }
138
139 async fn stats(&self) -> StoreStats {
141 StoreStats::default()
142 }
143
144 async fn evict_if_needed(&self) -> Result<u64, StoreError> {
147 Ok(0)
148 }
149
150 async fn pin(&self, _hash: &Hash) -> Result<(), StoreError> {
156 Ok(())
157 }
158
159 async fn unpin(&self, _hash: &Hash) -> Result<(), StoreError> {
161 Ok(())
162 }
163
164 fn pin_count(&self, _hash: &Hash) -> u32 {
166 0
167 }
168
169 fn is_pinned(&self, hash: &Hash) -> bool {
171 self.pin_count(hash) > 0
172 }
173}
174
175#[derive(Debug, thiserror::Error)]
177pub enum StoreError {
178 #[error("IO error: {0}")]
179 Io(#[from] std::io::Error),
180 #[error("Store error: {0}")]
181 Other(String),
182}
183
184#[derive(Debug, Default)]
185struct BufferedStoreInner {
186 pending: HashMap<Hash, Vec<u8>>,
187 order: Vec<Hash>,
188}
189
190#[derive(Debug, Clone, Copy)]
191struct BufferedStoreOptions {
192 check_base_on_put: bool,
193}
194
195impl Default for BufferedStoreOptions {
196 fn default() -> Self {
197 Self {
198 check_base_on_put: true,
199 }
200 }
201}
202
203#[derive(Debug, Clone)]
205pub struct BufferedStore<S: Store> {
206 base: Arc<S>,
207 inner: Arc<RwLock<BufferedStoreInner>>,
208 options: BufferedStoreOptions,
209}
210
211impl<S: Store> BufferedStore<S> {
212 pub fn new(base: Arc<S>) -> Self {
213 Self::with_options(base, BufferedStoreOptions::default())
214 }
215
216 pub fn new_optimistic(base: Arc<S>) -> Self {
217 Self::with_options(
218 base,
219 BufferedStoreOptions {
220 check_base_on_put: false,
221 },
222 )
223 }
224
225 fn with_options(base: Arc<S>, options: BufferedStoreOptions) -> Self {
226 Self {
227 base,
228 inner: Arc::new(RwLock::new(BufferedStoreInner::default())),
229 options,
230 }
231 }
232
233 pub async fn flush(&self) -> Result<usize, StoreError> {
234 let items = {
235 let mut inner = self.inner.write().unwrap();
236 if inner.order.is_empty() {
237 return Ok(0);
238 }
239
240 let order = std::mem::take(&mut inner.order);
241 let mut items = Vec::with_capacity(order.len());
242 for hash in order {
243 if let Some(data) = inner.pending.remove(&hash) {
244 items.push((hash, data));
245 }
246 }
247 items
248 };
249
250 if self.options.check_base_on_put {
251 self.base.put_many(items).await
252 } else {
253 self.base.put_many_optimistic(items).await
254 }
255 }
256}
257
258#[async_trait]
259impl<S: Store> Store for BufferedStore<S> {
260 async fn put(&self, hash: Hash, data: Vec<u8>) -> Result<bool, StoreError> {
261 {
262 let inner = self.inner.read().unwrap();
263 if inner.pending.contains_key(&hash) {
264 return Ok(false);
265 }
266 }
267
268 if self.options.check_base_on_put && self.base.has(&hash).await? {
269 return Ok(false);
270 }
271
272 let mut inner = self.inner.write().unwrap();
273 if inner.pending.contains_key(&hash) {
274 return Ok(false);
275 }
276 inner.order.push(hash);
277 inner.pending.insert(hash, data);
278 Ok(true)
279 }
280
281 async fn put_many(&self, items: Vec<(Hash, Vec<u8>)>) -> Result<usize, StoreError> {
282 let mut inserted = 0usize;
283 for (hash, data) in items {
284 if self.put(hash, data).await? {
285 inserted += 1;
286 }
287 }
288 Ok(inserted)
289 }
290
291 async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
292 if let Some(data) = self.inner.read().unwrap().pending.get(hash).cloned() {
293 return Ok(Some(data));
294 }
295 self.base.get(hash).await
296 }
297
298 async fn get_range(
299 &self,
300 hash: &Hash,
301 start: u64,
302 end_inclusive: u64,
303 ) -> Result<Option<Vec<u8>>, StoreError> {
304 let pending = {
305 let inner = self.inner.read().unwrap();
306 inner
307 .pending
308 .get(hash)
309 .map(|data| slice_blob_range(data, start, end_inclusive))
310 .transpose()?
311 };
312 if pending.is_some() {
313 return Ok(pending);
314 }
315 self.base.get_range(hash, start, end_inclusive).await
316 }
317
318 async fn blob_size(&self, hash: &Hash) -> Result<Option<u64>, StoreError> {
319 if let Some(size) = self
320 .inner
321 .read()
322 .unwrap()
323 .pending
324 .get(hash)
325 .map(|data| data.len() as u64)
326 {
327 return Ok(Some(size));
328 }
329 self.base.blob_size(hash).await
330 }
331
332 async fn has(&self, hash: &Hash) -> Result<bool, StoreError> {
333 if self.inner.read().unwrap().pending.contains_key(hash) {
334 return Ok(true);
335 }
336 self.base.has(hash).await
337 }
338
339 async fn delete(&self, hash: &Hash) -> Result<bool, StoreError> {
340 let removed = {
341 let mut inner = self.inner.write().unwrap();
342 let removed = inner.pending.remove(hash).is_some();
343 if removed {
344 inner.order.retain(|queued| queued != hash);
345 }
346 removed
347 };
348
349 if removed {
350 return Ok(true);
351 }
352
353 self.base.delete(hash).await
354 }
355
356 async fn stats(&self) -> StoreStats {
357 let mut stats = self.base.stats().await;
358 let pending_bytes = self
359 .inner
360 .read()
361 .unwrap()
362 .pending
363 .values()
364 .map(|data| data.len() as u64)
365 .sum::<u64>();
366 stats.count += self.inner.read().unwrap().pending.len() as u64;
367 stats.bytes += pending_bytes;
368 stats
369 }
370
371 async fn evict_if_needed(&self) -> Result<u64, StoreError> {
372 self.base.evict_if_needed().await
373 }
374
375 async fn pin(&self, hash: &Hash) -> Result<(), StoreError> {
376 self.base.pin(hash).await
377 }
378
379 async fn unpin(&self, hash: &Hash) -> Result<(), StoreError> {
380 self.base.unpin(hash).await
381 }
382
383 fn pin_count(&self, hash: &Hash) -> u32 {
384 self.base.pin_count(hash)
385 }
386}
387
388#[derive(Debug, Clone)]
390struct MemoryEntry {
391 data: Vec<u8>,
392 order: u64,
394}
395
396#[derive(Debug, Default)]
398struct MemoryStoreInner {
399 data: HashMap<String, MemoryEntry>,
400 pins: HashMap<String, u32>,
401 next_order: u64,
402 max_bytes: Option<u64>,
403}
404
405#[derive(Debug, Clone, Default)]
407pub struct MemoryStore {
408 inner: Arc<RwLock<MemoryStoreInner>>,
409}
410
411impl MemoryStore {
412 pub fn new() -> Self {
413 Self {
414 inner: Arc::new(RwLock::new(MemoryStoreInner::default())),
415 }
416 }
417
418 pub fn with_max_bytes(max_bytes: u64) -> Self {
420 Self {
421 inner: Arc::new(RwLock::new(MemoryStoreInner {
422 max_bytes: if max_bytes > 0 { Some(max_bytes) } else { None },
423 ..Default::default()
424 })),
425 }
426 }
427
428 pub fn size(&self) -> usize {
430 self.inner.read().unwrap().data.len()
431 }
432
433 pub fn total_bytes(&self) -> usize {
435 self.inner
436 .read()
437 .unwrap()
438 .data
439 .values()
440 .map(|e| e.data.len())
441 .sum()
442 }
443
444 pub fn clear(&self) {
446 self.inner.write().unwrap().data.clear();
447 }
448
449 pub fn keys(&self) -> Vec<Hash> {
451 self.inner
452 .read()
453 .unwrap()
454 .data
455 .keys()
456 .filter_map(|hex| {
457 let bytes = hex::decode(hex).ok()?;
458 if bytes.len() != 32 {
459 return None;
460 }
461 let mut hash = [0u8; 32];
462 hash.copy_from_slice(&bytes);
463 Some(hash)
464 })
465 .collect()
466 }
467
468 fn evict_to_target(&self, target_bytes: u64) -> u64 {
470 let mut inner = self.inner.write().unwrap();
471
472 let current_bytes: u64 = inner.data.values().map(|e| e.data.len() as u64).sum();
473 if current_bytes <= target_bytes {
474 return 0;
475 }
476
477 let mut unpinned: Vec<(String, u64, u64)> = inner
479 .data
480 .iter()
481 .filter(|(key, _)| inner.pins.get(*key).copied().unwrap_or(0) == 0)
482 .map(|(key, entry)| (key.clone(), entry.order, entry.data.len() as u64))
483 .collect();
484
485 unpinned.sort_by_key(|(_, order, _)| *order);
486
487 let mut freed = 0u64;
488 let to_free = current_bytes - target_bytes;
489
490 for (key, _, size) in unpinned {
491 if freed >= to_free {
492 break;
493 }
494 inner.data.remove(&key);
495 freed += size;
496 }
497
498 freed
499 }
500}
501
502#[async_trait]
503impl Store for MemoryStore {
504 async fn put(&self, hash: Hash, data: Vec<u8>) -> Result<bool, StoreError> {
505 let key = to_hex(&hash);
506 let mut inner = self.inner.write().unwrap();
507 if inner.data.contains_key(&key) {
508 return Ok(false);
509 }
510 let order = inner.next_order;
511 inner.next_order += 1;
512 inner.data.insert(key, MemoryEntry { data, order });
513 Ok(true)
514 }
515
516 async fn put_many(&self, items: Vec<(Hash, Vec<u8>)>) -> Result<usize, StoreError> {
517 let mut inserted = 0usize;
518 let mut inner = self.inner.write().unwrap();
519 for (hash, data) in items {
520 let key = to_hex(&hash);
521 if inner.data.contains_key(&key) {
522 continue;
523 }
524 let order = inner.next_order;
525 inner.next_order += 1;
526 inner.data.insert(key, MemoryEntry { data, order });
527 inserted += 1;
528 }
529 Ok(inserted)
530 }
531
532 async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
533 let key = to_hex(hash);
534 let inner = self.inner.read().unwrap();
535 Ok(inner.data.get(&key).map(|e| e.data.clone()))
536 }
537
538 async fn get_range(
539 &self,
540 hash: &Hash,
541 start: u64,
542 end_inclusive: u64,
543 ) -> Result<Option<Vec<u8>>, StoreError> {
544 let key = to_hex(hash);
545 let inner = self.inner.read().unwrap();
546 inner
547 .data
548 .get(&key)
549 .map(|entry| slice_blob_range(&entry.data, start, end_inclusive))
550 .transpose()
551 }
552
553 async fn blob_size(&self, hash: &Hash) -> Result<Option<u64>, StoreError> {
554 let key = to_hex(hash);
555 let inner = self.inner.read().unwrap();
556 Ok(inner.data.get(&key).map(|entry| entry.data.len() as u64))
557 }
558
559 async fn has(&self, hash: &Hash) -> Result<bool, StoreError> {
560 let key = to_hex(hash);
561 Ok(self.inner.read().unwrap().data.contains_key(&key))
562 }
563
564 async fn delete(&self, hash: &Hash) -> Result<bool, StoreError> {
565 let key = to_hex(hash);
566 let mut inner = self.inner.write().unwrap();
567 inner.pins.remove(&key);
569 Ok(inner.data.remove(&key).is_some())
570 }
571
572 fn set_max_bytes(&self, max: u64) {
573 self.inner.write().unwrap().max_bytes = if max > 0 { Some(max) } else { None };
574 }
575
576 fn max_bytes(&self) -> Option<u64> {
577 self.inner.read().unwrap().max_bytes
578 }
579
580 async fn stats(&self) -> StoreStats {
581 let inner = self.inner.read().unwrap();
582 let mut count = 0u64;
583 let mut bytes = 0u64;
584 let mut pinned_count = 0u64;
585 let mut pinned_bytes = 0u64;
586
587 for (key, entry) in &inner.data {
588 count += 1;
589 bytes += entry.data.len() as u64;
590 if inner.pins.get(key).copied().unwrap_or(0) > 0 {
591 pinned_count += 1;
592 pinned_bytes += entry.data.len() as u64;
593 }
594 }
595
596 StoreStats {
597 count,
598 bytes,
599 pinned_count,
600 pinned_bytes,
601 }
602 }
603
604 async fn evict_if_needed(&self) -> Result<u64, StoreError> {
605 let max = match self.inner.read().unwrap().max_bytes {
606 Some(m) => m,
607 None => return Ok(0), };
609
610 let current: u64 = self
611 .inner
612 .read()
613 .unwrap()
614 .data
615 .values()
616 .map(|e| e.data.len() as u64)
617 .sum();
618
619 if current <= max {
620 return Ok(0);
621 }
622
623 let target = max * 9 / 10;
625 Ok(self.evict_to_target(target))
626 }
627
628 async fn pin(&self, hash: &Hash) -> Result<(), StoreError> {
629 let key = to_hex(hash);
630 let mut inner = self.inner.write().unwrap();
631 *inner.pins.entry(key).or_insert(0) += 1;
632 Ok(())
633 }
634
635 async fn unpin(&self, hash: &Hash) -> Result<(), StoreError> {
636 let key = to_hex(hash);
637 let mut inner = self.inner.write().unwrap();
638 if let Some(count) = inner.pins.get_mut(&key) {
639 if *count > 0 {
640 *count -= 1;
641 }
642 if *count == 0 {
643 inner.pins.remove(&key);
644 }
645 }
646 Ok(())
647 }
648
649 fn pin_count(&self, hash: &Hash) -> u32 {
650 let key = to_hex(hash);
651 self.inner
652 .read()
653 .unwrap()
654 .pins
655 .get(&key)
656 .copied()
657 .unwrap_or(0)
658 }
659}
660
661#[cfg(test)]
662mod tests {
663 use super::*;
664 use crate::hash::sha256;
665 use std::sync::atomic::{AtomicUsize, Ordering};
666
667 #[derive(Default)]
668 struct OptimisticBatchStore {
669 inner: MemoryStore,
670 regular_batches: AtomicUsize,
671 optimistic_batches: AtomicUsize,
672 }
673
674 #[async_trait]
675 impl Store for OptimisticBatchStore {
676 async fn put(&self, hash: Hash, data: Vec<u8>) -> Result<bool, StoreError> {
677 self.inner.put(hash, data).await
678 }
679
680 async fn put_many(&self, items: Vec<(Hash, Vec<u8>)>) -> Result<usize, StoreError> {
681 self.regular_batches.fetch_add(1, Ordering::Relaxed);
682 self.inner.put_many(items).await
683 }
684
685 async fn put_many_optimistic(
686 &self,
687 items: Vec<(Hash, Vec<u8>)>,
688 ) -> Result<usize, StoreError> {
689 self.optimistic_batches.fetch_add(1, Ordering::Relaxed);
690 self.inner.put_many(items).await
691 }
692
693 async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
694 self.inner.get(hash).await
695 }
696
697 async fn has(&self, hash: &Hash) -> Result<bool, StoreError> {
698 self.inner.has(hash).await
699 }
700
701 async fn delete(&self, hash: &Hash) -> Result<bool, StoreError> {
702 self.inner.delete(hash).await
703 }
704 }
705
706 #[tokio::test]
707 async fn test_put_returns_true_for_new() {
708 let store = MemoryStore::new();
709 let data = vec![1u8, 2, 3];
710 let hash = sha256(&data);
711
712 let result = store.put(hash, data).await.unwrap();
713 assert!(result);
714 }
715
716 #[tokio::test]
717 async fn test_put_returns_false_for_duplicate() {
718 let store = MemoryStore::new();
719 let data = vec![1u8, 2, 3];
720 let hash = sha256(&data);
721
722 store.put(hash, data.clone()).await.unwrap();
723 let result = store.put(hash, data).await.unwrap();
724 assert!(!result);
725 }
726
727 #[tokio::test]
728 async fn test_put_many_counts_only_new_items() {
729 let store = MemoryStore::new();
730 let data1 = vec![1u8, 2, 3];
731 let data2 = vec![4u8, 5, 6];
732 let hash1 = sha256(&data1);
733 let hash2 = sha256(&data2);
734
735 store.put(hash1, data1.clone()).await.unwrap();
736 let inserted = store
737 .put_many(vec![(hash1, data1), (hash2, data2.clone())])
738 .await
739 .unwrap();
740
741 assert_eq!(inserted, 1);
742 assert_eq!(store.get(&hash2).await.unwrap(), Some(data2));
743 }
744
745 #[tokio::test]
746 async fn test_buffered_store_flushes_pending_writes() {
747 let base = std::sync::Arc::new(MemoryStore::new());
748 let buffered = BufferedStore::new(std::sync::Arc::clone(&base));
749 let data = vec![9u8, 8, 7];
750 let hash = sha256(&data);
751
752 assert!(buffered.put(hash, data.clone()).await.unwrap());
753 assert_eq!(buffered.get(&hash).await.unwrap(), Some(data.clone()));
754 assert_eq!(base.get(&hash).await.unwrap(), None);
755
756 let flushed = buffered.flush().await.unwrap();
757
758 assert_eq!(flushed, 1);
759 assert_eq!(base.get(&hash).await.unwrap(), Some(data));
760 }
761
762 #[tokio::test]
763 async fn test_optimistic_buffered_store_avoids_base_probe_but_preserves_contents() {
764 let base = std::sync::Arc::new(MemoryStore::new());
765 let buffered = BufferedStore::new_optimistic(std::sync::Arc::clone(&base));
766 let data = vec![4u8, 5, 6];
767 let hash = sha256(&data);
768
769 base.put(hash, data.clone()).await.unwrap();
770
771 assert!(buffered.put(hash, data.clone()).await.unwrap());
772 assert_eq!(buffered.get(&hash).await.unwrap(), Some(data.clone()));
773
774 let flushed = buffered.flush().await.unwrap();
775
776 assert_eq!(flushed, 0);
777 assert_eq!(base.get(&hash).await.unwrap(), Some(data));
778 }
779
780 #[tokio::test]
781 async fn test_optimistic_buffered_store_uses_idempotent_batch_flush() {
782 let base = std::sync::Arc::new(OptimisticBatchStore::default());
783 let buffered = BufferedStore::new_optimistic(std::sync::Arc::clone(&base));
784 let data = vec![7u8; 1024];
785 let hash = sha256(&data);
786
787 base.inner.put(hash, data.clone()).await.unwrap();
788 assert!(buffered.put(hash, data).await.unwrap());
789 assert_eq!(buffered.flush().await.unwrap(), 0);
790
791 assert_eq!(base.regular_batches.load(Ordering::Relaxed), 0);
792 assert_eq!(base.optimistic_batches.load(Ordering::Relaxed), 1);
793 }
794
795 #[tokio::test]
796 async fn test_get_returns_data() {
797 let store = MemoryStore::new();
798 let data = vec![1u8, 2, 3];
799 let hash = sha256(&data);
800
801 store.put(hash, data.clone()).await.unwrap();
802 let result = store.get(&hash).await.unwrap();
803
804 assert_eq!(result, Some(data));
805 }
806
807 #[tokio::test]
808 async fn test_get_returns_none_for_missing() {
809 let store = MemoryStore::new();
810 let hash = [0u8; 32];
811
812 let result = store.get(&hash).await.unwrap();
813 assert!(result.is_none());
814 }
815
816 #[tokio::test]
817 async fn test_has_returns_true() {
818 let store = MemoryStore::new();
819 let data = vec![1u8, 2, 3];
820 let hash = sha256(&data);
821
822 store.put(hash, data).await.unwrap();
823 assert!(store.has(&hash).await.unwrap());
824 }
825
826 #[tokio::test]
827 async fn test_has_returns_false() {
828 let store = MemoryStore::new();
829 let hash = [0u8; 32];
830
831 assert!(!store.has(&hash).await.unwrap());
832 }
833
834 #[tokio::test]
835 async fn test_delete_returns_true() {
836 let store = MemoryStore::new();
837 let data = vec![1u8, 2, 3];
838 let hash = sha256(&data);
839
840 store.put(hash, data).await.unwrap();
841 let result = store.delete(&hash).await.unwrap();
842
843 assert!(result);
844 assert!(!store.has(&hash).await.unwrap());
845 }
846
847 #[tokio::test]
848 async fn test_delete_returns_false() {
849 let store = MemoryStore::new();
850 let hash = [0u8; 32];
851
852 let result = store.delete(&hash).await.unwrap();
853 assert!(!result);
854 }
855
856 #[tokio::test]
857 async fn test_size() {
858 let store = MemoryStore::new();
859 assert_eq!(store.size(), 0);
860
861 let data1 = vec![1u8];
862 let data2 = vec![2u8];
863 let hash1 = sha256(&data1);
864 let hash2 = sha256(&data2);
865
866 store.put(hash1, data1).await.unwrap();
867 store.put(hash2, data2).await.unwrap();
868
869 assert_eq!(store.size(), 2);
870 }
871
872 #[tokio::test]
873 async fn test_total_bytes() {
874 let store = MemoryStore::new();
875 assert_eq!(store.total_bytes(), 0);
876
877 let data1 = vec![1u8, 2, 3];
878 let data2 = vec![4u8, 5];
879 let hash1 = sha256(&data1);
880 let hash2 = sha256(&data2);
881
882 store.put(hash1, data1).await.unwrap();
883 store.put(hash2, data2).await.unwrap();
884
885 assert_eq!(store.total_bytes(), 5);
886 }
887
888 #[tokio::test]
889 async fn test_clear() {
890 let store = MemoryStore::new();
891 let data = vec![1u8, 2, 3];
892 let hash = sha256(&data);
893
894 store.put(hash, data).await.unwrap();
895 store.clear();
896
897 assert_eq!(store.size(), 0);
898 assert!(!store.has(&hash).await.unwrap());
899 }
900
901 #[tokio::test]
902 async fn test_keys() {
903 let store = MemoryStore::new();
904 assert!(store.keys().is_empty());
905
906 let data1 = vec![1u8];
907 let data2 = vec![2u8];
908 let hash1 = sha256(&data1);
909 let hash2 = sha256(&data2);
910
911 store.put(hash1, data1).await.unwrap();
912 store.put(hash2, data2).await.unwrap();
913
914 let keys = store.keys();
915 assert_eq!(keys.len(), 2);
916
917 let mut hex_keys: Vec<_> = keys.iter().map(to_hex).collect();
918 hex_keys.sort();
919 let mut expected: Vec<_> = vec![to_hex(&hash1), to_hex(&hash2)];
920 expected.sort();
921 assert_eq!(hex_keys, expected);
922 }
923
924 #[tokio::test]
925 async fn test_pin_and_unpin() {
926 let store = MemoryStore::new();
927 let data = vec![1u8, 2, 3];
928 let hash = sha256(&data);
929
930 store.put(hash, data).await.unwrap();
931
932 assert!(!store.is_pinned(&hash));
934 assert_eq!(store.pin_count(&hash), 0);
935
936 store.pin(&hash).await.unwrap();
938 assert!(store.is_pinned(&hash));
939 assert_eq!(store.pin_count(&hash), 1);
940
941 store.unpin(&hash).await.unwrap();
943 assert!(!store.is_pinned(&hash));
944 assert_eq!(store.pin_count(&hash), 0);
945 }
946
947 #[tokio::test]
948 async fn test_pin_count_ref_counting() {
949 let store = MemoryStore::new();
950 let data = vec![1u8, 2, 3];
951 let hash = sha256(&data);
952
953 store.put(hash, data).await.unwrap();
954
955 store.pin(&hash).await.unwrap();
957 store.pin(&hash).await.unwrap();
958 store.pin(&hash).await.unwrap();
959 assert_eq!(store.pin_count(&hash), 3);
960
961 store.unpin(&hash).await.unwrap();
963 assert_eq!(store.pin_count(&hash), 2);
964 assert!(store.is_pinned(&hash));
965
966 store.unpin(&hash).await.unwrap();
968 store.unpin(&hash).await.unwrap();
969 assert_eq!(store.pin_count(&hash), 0);
970 assert!(!store.is_pinned(&hash));
971
972 store.unpin(&hash).await.unwrap();
974 assert_eq!(store.pin_count(&hash), 0);
975 }
976
977 #[tokio::test]
978 async fn test_stats() {
979 let store = MemoryStore::new();
980
981 let data1 = vec![1u8, 2, 3]; let data2 = vec![4u8, 5]; let hash1 = sha256(&data1);
984 let hash2 = sha256(&data2);
985
986 store.put(hash1, data1).await.unwrap();
987 store.put(hash2, data2).await.unwrap();
988
989 store.pin(&hash1).await.unwrap();
991
992 let stats = store.stats().await;
993 assert_eq!(stats.count, 2);
994 assert_eq!(stats.bytes, 5);
995 assert_eq!(stats.pinned_count, 1);
996 assert_eq!(stats.pinned_bytes, 3);
997 }
998
999 #[tokio::test]
1000 async fn test_max_bytes() {
1001 let store = MemoryStore::new();
1002 assert!(store.max_bytes().is_none());
1003
1004 store.set_max_bytes(1000);
1005 assert_eq!(store.max_bytes(), Some(1000));
1006
1007 store.set_max_bytes(0);
1009 assert!(store.max_bytes().is_none());
1010 }
1011
1012 #[tokio::test]
1013 async fn test_with_max_bytes() {
1014 let store = MemoryStore::with_max_bytes(500);
1015 assert_eq!(store.max_bytes(), Some(500));
1016
1017 let store_unlimited = MemoryStore::with_max_bytes(0);
1018 assert!(store_unlimited.max_bytes().is_none());
1019 }
1020
1021 #[tokio::test]
1022 async fn test_eviction_respects_pins() {
1023 let store = MemoryStore::with_max_bytes(10);
1025
1026 let data1 = vec![1u8, 1, 1]; let data2 = vec![2u8, 2, 2];
1029 let data3 = vec![3u8, 3, 3]; let hash1 = sha256(&data1);
1031 let hash2 = sha256(&data2);
1032 let hash3 = sha256(&data3);
1033
1034 store.put(hash1, data1).await.unwrap();
1035 store.put(hash2, data2).await.unwrap();
1036 store.put(hash3, data3).await.unwrap();
1037
1038 store.pin(&hash1).await.unwrap();
1040
1041 let data4 = vec![4u8, 4, 4];
1043 let hash4 = sha256(&data4);
1044 store.put(hash4, data4).await.unwrap();
1045
1046 let freed = store.evict_if_needed().await.unwrap();
1048 assert!(freed > 0);
1049
1050 assert!(store.has(&hash1).await.unwrap());
1052 assert!(!store.has(&hash2).await.unwrap());
1054 assert!(store.has(&hash3).await.unwrap());
1056 assert!(store.has(&hash4).await.unwrap());
1057 }
1058
1059 #[tokio::test]
1060 async fn test_eviction_lru_order() {
1061 let store = MemoryStore::with_max_bytes(15);
1063
1064 let data1 = vec![1u8; 5]; let data2 = vec![2u8; 5];
1067 let data3 = vec![3u8; 5];
1068 let data4 = vec![4u8; 5]; let hash1 = sha256(&data1);
1070 let hash2 = sha256(&data2);
1071 let hash3 = sha256(&data3);
1072 let hash4 = sha256(&data4);
1073
1074 store.put(hash1, data1).await.unwrap();
1075 store.put(hash2, data2).await.unwrap();
1076 store.put(hash3, data3).await.unwrap();
1077 store.put(hash4, data4).await.unwrap();
1078
1079 assert_eq!(store.total_bytes(), 20);
1081
1082 let freed = store.evict_if_needed().await.unwrap();
1084 assert!(freed >= 5); assert!(!store.has(&hash1).await.unwrap());
1088 assert!(store.has(&hash4).await.unwrap());
1090 }
1091
1092 #[tokio::test]
1093 async fn test_no_eviction_when_under_limit() {
1094 let store = MemoryStore::with_max_bytes(100);
1095
1096 let data = vec![1u8, 2, 3];
1097 let hash = sha256(&data);
1098 store.put(hash, data).await.unwrap();
1099
1100 let freed = store.evict_if_needed().await.unwrap();
1101 assert_eq!(freed, 0);
1102 assert!(store.has(&hash).await.unwrap());
1103 }
1104
1105 #[tokio::test]
1106 async fn test_no_eviction_without_limit() {
1107 let store = MemoryStore::new();
1108
1109 for i in 0..100u8 {
1111 let data = vec![i; 100];
1112 let hash = sha256(&data);
1113 store.put(hash, data).await.unwrap();
1114 }
1115
1116 let freed = store.evict_if_needed().await.unwrap();
1117 assert_eq!(freed, 0);
1118 assert_eq!(store.size(), 100);
1119 }
1120
1121 #[tokio::test]
1122 async fn test_delete_removes_pin() {
1123 let store = MemoryStore::new();
1124 let data = vec![1u8, 2, 3];
1125 let hash = sha256(&data);
1126
1127 store.put(hash, data).await.unwrap();
1128 store.pin(&hash).await.unwrap();
1129 assert!(store.is_pinned(&hash));
1130
1131 store.delete(&hash).await.unwrap();
1132 assert_eq!(store.pin_count(&hash), 0);
1134 }
1135}