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    /// Flush writes retained by an overlay store.
92    ///
93    /// Ordinary stores write through and keep the default no-op. Buffered
94    /// stores override this so callers that build several independent index
95    /// projections can bound memory without exposing an incomplete root.
96    async fn flush_pending(&self) -> Result<usize, StoreError> {
97        Ok(0)
98    }
99
100    /// Retrieve data by hash
101    /// Returns data or None if not found
102    async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError>;
103
104    /// Retrieve a byte range from a blob by hash.
105    ///
106    /// `end_inclusive` is inclusive, matching HTTP range requests and the sync
107    /// filesystem/LMDB storage primitives. Implementations may override this to
108    /// avoid loading whole blobs.
109    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    /// Return the stored blob size in bytes.
122    ///
123    /// The default implementation reads the blob. Backends with metadata should
124    /// override this to answer without loading payload bytes.
125    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    /// Check if hash exists
130    async fn has(&self, hash: &Hash) -> Result<bool, StoreError>;
131
132    /// Delete by hash
133    /// Returns true if deleted, false if didn't exist
134    async fn delete(&self, hash: &Hash) -> Result<bool, StoreError>;
135
136    /// Delete multiple hashes and return the number that existed.
137    ///
138    /// Persistent stores should override this to use one bounded transaction
139    /// rather than forcing one durability barrier per hash.
140    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    // ========================================================================
151    // Optional: Storage limits and eviction (default no-op implementations)
152    // ========================================================================
153
154    /// Set maximum storage size in bytes. 0 = unlimited.
155    fn set_max_bytes(&self, _max: u64) {}
156
157    /// Get maximum storage size. None = unlimited.
158    fn max_bytes(&self) -> Option<u64> {
159        None
160    }
161
162    /// Get storage statistics
163    async fn stats(&self) -> StoreStats {
164        StoreStats::default()
165    }
166
167    /// Evict unpinned items if over storage limit.
168    /// Returns number of bytes freed.
169    async fn evict_if_needed(&self) -> Result<u64, StoreError> {
170        Ok(0)
171    }
172
173    // ========================================================================
174    // Optional: Pinning (default no-op implementations)
175    // ========================================================================
176
177    /// Pin a hash (increment ref count). Pinned items are not evicted.
178    async fn pin(&self, _hash: &Hash) -> Result<(), StoreError> {
179        Ok(())
180    }
181
182    /// Unpin a hash (decrement ref count). Item can be evicted when count reaches 0.
183    async fn unpin(&self, _hash: &Hash) -> Result<(), StoreError> {
184        Ok(())
185    }
186
187    /// Get pin count for a hash. 0 = not pinned.
188    fn pin_count(&self, _hash: &Hash) -> u32 {
189        0
190    }
191
192    /// Check if hash is pinned (pin count > 0)
193    fn is_pinned(&self, hash: &Hash) -> bool {
194        self.pin_count(hash) > 0
195    }
196}
197
198/// Store error type
199#[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/// Buffered overlay store that keeps writes in memory until flushed.
227#[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/// Entry in the memory store with metadata for LRU
416#[derive(Debug, Clone)]
417struct MemoryEntry {
418    data: Vec<u8>,
419    /// Insertion order for LRU (lower = older)
420    order: u64,
421}
422
423/// Internal state for MemoryStore
424#[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/// In-memory content-addressed store with LRU eviction and pinning
433#[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    /// Create a new store with a maximum size limit
446    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    /// Get number of stored items
456    pub fn size(&self) -> usize {
457        self.inner.read().unwrap().data.len()
458    }
459
460    /// Get total bytes stored
461    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    /// Clear all data (but not pins)
472    pub fn clear(&self) {
473        self.inner.write().unwrap().data.clear();
474    }
475
476    /// List all hashes
477    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    /// Evict oldest unpinned entries until under target bytes
496    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        // Collect unpinned entries sorted by order (oldest first)
505        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        // Also remove pin entry if exists
595        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), // No limit set
635        };
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        // Evict to 90% of max to avoid frequent evictions
651        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        // Initially not pinned
960        assert!(!store.is_pinned(&hash));
961        assert_eq!(store.pin_count(&hash), 0);
962
963        // Pin
964        store.pin(&hash).await.unwrap();
965        assert!(store.is_pinned(&hash));
966        assert_eq!(store.pin_count(&hash), 1);
967
968        // Unpin
969        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        // Pin multiple times
983        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        // Unpin once
989        store.unpin(&hash).await.unwrap();
990        assert_eq!(store.pin_count(&hash), 2);
991        assert!(store.is_pinned(&hash));
992
993        // Unpin remaining
994        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        // Extra unpin shouldn't go negative
1000        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]; // 3 bytes
1009        let data2 = vec![4u8, 5]; // 2 bytes
1010        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        // Pin one item
1017        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        // 0 means unlimited
1035        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        // Store with 10 byte limit
1051        let store = MemoryStore::with_max_bytes(10);
1052
1053        // Insert 3 items: 3 + 3 + 3 = 9 bytes
1054        let data1 = vec![1u8, 1, 1]; // oldest
1055        let data2 = vec![2u8, 2, 2];
1056        let data3 = vec![3u8, 3, 3]; // newest
1057        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        // Pin the oldest item
1066        store.pin(&hash1).await.unwrap();
1067
1068        // Add more data to exceed limit: 9 + 3 = 12 bytes > 10
1069        let data4 = vec![4u8, 4, 4];
1070        let hash4 = sha256(&data4);
1071        store.put(hash4, data4).await.unwrap();
1072
1073        // Evict - should remove hash2 (oldest unpinned)
1074        let freed = store.evict_if_needed().await.unwrap();
1075        assert!(freed > 0);
1076
1077        // hash1 should still exist (pinned)
1078        assert!(store.has(&hash1).await.unwrap());
1079        // hash2 should be gone (oldest unpinned)
1080        assert!(!store.has(&hash2).await.unwrap());
1081        // hash3 and hash4 should exist
1082        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        // Store with 15 byte limit
1089        let store = MemoryStore::with_max_bytes(15);
1090
1091        // Insert items in order (oldest first)
1092        let data1 = vec![1u8; 5]; // oldest
1093        let data2 = vec![2u8; 5];
1094        let data3 = vec![3u8; 5];
1095        let data4 = vec![4u8; 5]; // newest
1096        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        // Now at 20 bytes, limit is 15
1107        assert_eq!(store.total_bytes(), 20);
1108
1109        // Evict - should remove oldest items first
1110        let freed = store.evict_if_needed().await.unwrap();
1111        assert!(freed >= 5); // At least one item evicted
1112
1113        // Oldest should be gone
1114        assert!(!store.has(&hash1).await.unwrap());
1115        // Newest should still exist
1116        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        // Add lots of data
1137        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        // Pin should be gone after delete
1160        assert_eq!(store.pin_count(&hash), 0);
1161    }
1162}