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