Skip to main content

hashtree_core/
store.rs

1//! Content-addressed key-value store interfaces and implementations
2
3use async_trait::async_trait;
4use std::collections::HashMap;
5use std::sync::{Arc, RwLock};
6
7use crate::types::{to_hex, Hash};
8
9/// Return a byte range from an in-memory blob.
10///
11/// `end_inclusive` follows HTTP/storage range conventions. Out-of-bounds ranges
12/// return an empty or clamped slice instead of an error.
13pub 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/// Storage statistics
37#[derive(Debug, Clone, Default)]
38pub struct StoreStats {
39    /// Number of items in store
40    pub count: u64,
41    /// Total bytes stored
42    pub bytes: u64,
43    /// Number of pinned items
44    pub pinned_count: u64,
45    /// Bytes used by pinned items
46    pub pinned_bytes: u64,
47}
48
49/// Result of a batch insert.
50#[derive(Debug, Clone, Default, PartialEq, Eq)]
51pub struct PutManyReport {
52    /// Number of candidate items passed by the caller, including duplicates.
53    pub total: usize,
54    /// Number of blobs newly inserted into the store.
55    pub inserted: usize,
56    /// Logical bytes newly inserted into the store.
57    pub inserted_bytes: u64,
58    /// Hashes that were newly inserted, in insertion order.
59    pub inserted_hashes: Vec<Hash>,
60}
61
62/// Content-addressed key-value store interface
63#[async_trait]
64pub trait Store: Send + Sync {
65    /// Store data by its hash
66    /// Returns true if newly stored, false if already existed
67    async fn put(&self, hash: Hash, data: Vec<u8>) -> Result<bool, StoreError>;
68
69    /// Store multiple blobs.
70    /// Returns the number of newly stored items.
71    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    /// Store a locally generated content-addressed batch without requiring an
82    /// implementation to reread bytes for hashes it already tracks.
83    ///
84    /// Callers must provide bytes matching every hash. The default keeps the
85    /// ordinary verified batch behavior; stores with an authenticated local
86    /// location catalog may override this to skip known hashes cheaply.
87    async fn put_many_optimistic(&self, items: Vec<(Hash, Vec<u8>)>) -> Result<usize, StoreError> {
88        self.put_many(items).await
89    }
90
91    /// Retrieve data by hash
92    /// Returns data or None if not found
93    async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError>;
94
95    /// Retrieve a byte range from a blob by hash.
96    ///
97    /// `end_inclusive` is inclusive, matching HTTP range requests and the sync
98    /// filesystem/LMDB storage primitives. Implementations may override this to
99    /// avoid loading whole blobs.
100    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    /// Return the stored blob size in bytes.
113    ///
114    /// The default implementation reads the blob. Backends with metadata should
115    /// override this to answer without loading payload bytes.
116    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    /// Check if hash exists
121    async fn has(&self, hash: &Hash) -> Result<bool, StoreError>;
122
123    /// Delete by hash
124    /// Returns true if deleted, false if didn't exist
125    async fn delete(&self, hash: &Hash) -> Result<bool, StoreError>;
126
127    // ========================================================================
128    // Optional: Storage limits and eviction (default no-op implementations)
129    // ========================================================================
130
131    /// Set maximum storage size in bytes. 0 = unlimited.
132    fn set_max_bytes(&self, _max: u64) {}
133
134    /// Get maximum storage size. None = unlimited.
135    fn max_bytes(&self) -> Option<u64> {
136        None
137    }
138
139    /// Get storage statistics
140    async fn stats(&self) -> StoreStats {
141        StoreStats::default()
142    }
143
144    /// Evict unpinned items if over storage limit.
145    /// Returns number of bytes freed.
146    async fn evict_if_needed(&self) -> Result<u64, StoreError> {
147        Ok(0)
148    }
149
150    // ========================================================================
151    // Optional: Pinning (default no-op implementations)
152    // ========================================================================
153
154    /// Pin a hash (increment ref count). Pinned items are not evicted.
155    async fn pin(&self, _hash: &Hash) -> Result<(), StoreError> {
156        Ok(())
157    }
158
159    /// Unpin a hash (decrement ref count). Item can be evicted when count reaches 0.
160    async fn unpin(&self, _hash: &Hash) -> Result<(), StoreError> {
161        Ok(())
162    }
163
164    /// Get pin count for a hash. 0 = not pinned.
165    fn pin_count(&self, _hash: &Hash) -> u32 {
166        0
167    }
168
169    /// Check if hash is pinned (pin count > 0)
170    fn is_pinned(&self, hash: &Hash) -> bool {
171        self.pin_count(hash) > 0
172    }
173}
174
175/// Store error type
176#[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/// Buffered overlay store that keeps writes in memory until flushed.
204#[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/// Entry in the memory store with metadata for LRU
389#[derive(Debug, Clone)]
390struct MemoryEntry {
391    data: Vec<u8>,
392    /// Insertion order for LRU (lower = older)
393    order: u64,
394}
395
396/// Internal state for MemoryStore
397#[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/// In-memory content-addressed store with LRU eviction and pinning
406#[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    /// Create a new store with a maximum size limit
419    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    /// Get number of stored items
429    pub fn size(&self) -> usize {
430        self.inner.read().unwrap().data.len()
431    }
432
433    /// Get total bytes stored
434    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    /// Clear all data (but not pins)
445    pub fn clear(&self) {
446        self.inner.write().unwrap().data.clear();
447    }
448
449    /// List all hashes
450    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    /// Evict oldest unpinned entries until under target bytes
469    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        // Collect unpinned entries sorted by order (oldest first)
478        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        // Also remove pin entry if exists
568        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), // No limit set
608        };
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        // Evict to 90% of max to avoid frequent evictions
624        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        // Initially not pinned
933        assert!(!store.is_pinned(&hash));
934        assert_eq!(store.pin_count(&hash), 0);
935
936        // Pin
937        store.pin(&hash).await.unwrap();
938        assert!(store.is_pinned(&hash));
939        assert_eq!(store.pin_count(&hash), 1);
940
941        // Unpin
942        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        // Pin multiple times
956        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        // Unpin once
962        store.unpin(&hash).await.unwrap();
963        assert_eq!(store.pin_count(&hash), 2);
964        assert!(store.is_pinned(&hash));
965
966        // Unpin remaining
967        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        // Extra unpin shouldn't go negative
973        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]; // 3 bytes
982        let data2 = vec![4u8, 5]; // 2 bytes
983        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        // Pin one item
990        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        // 0 means unlimited
1008        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        // Store with 10 byte limit
1024        let store = MemoryStore::with_max_bytes(10);
1025
1026        // Insert 3 items: 3 + 3 + 3 = 9 bytes
1027        let data1 = vec![1u8, 1, 1]; // oldest
1028        let data2 = vec![2u8, 2, 2];
1029        let data3 = vec![3u8, 3, 3]; // newest
1030        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        // Pin the oldest item
1039        store.pin(&hash1).await.unwrap();
1040
1041        // Add more data to exceed limit: 9 + 3 = 12 bytes > 10
1042        let data4 = vec![4u8, 4, 4];
1043        let hash4 = sha256(&data4);
1044        store.put(hash4, data4).await.unwrap();
1045
1046        // Evict - should remove hash2 (oldest unpinned)
1047        let freed = store.evict_if_needed().await.unwrap();
1048        assert!(freed > 0);
1049
1050        // hash1 should still exist (pinned)
1051        assert!(store.has(&hash1).await.unwrap());
1052        // hash2 should be gone (oldest unpinned)
1053        assert!(!store.has(&hash2).await.unwrap());
1054        // hash3 and hash4 should exist
1055        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        // Store with 15 byte limit
1062        let store = MemoryStore::with_max_bytes(15);
1063
1064        // Insert items in order (oldest first)
1065        let data1 = vec![1u8; 5]; // oldest
1066        let data2 = vec![2u8; 5];
1067        let data3 = vec![3u8; 5];
1068        let data4 = vec![4u8; 5]; // newest
1069        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        // Now at 20 bytes, limit is 15
1080        assert_eq!(store.total_bytes(), 20);
1081
1082        // Evict - should remove oldest items first
1083        let freed = store.evict_if_needed().await.unwrap();
1084        assert!(freed >= 5); // At least one item evicted
1085
1086        // Oldest should be gone
1087        assert!(!store.has(&hash1).await.unwrap());
1088        // Newest should still exist
1089        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        // Add lots of data
1110        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        // Pin should be gone after delete
1133        assert_eq!(store.pin_count(&hash), 0);
1134    }
1135}