aria2_core/filesystem/disk_cache.rs
1use std::collections::BTreeMap;
2use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
3use tokio::sync::Mutex;
4use tracing::debug;
5
6use crate::error::Result;
7
8/// Default maximum cache size: 16 MB
9const DEFAULT_MAX_SIZE_BYTES: usize = 16 * 1024 * 1024;
10
11impl Default for WrDiskCache {
12 fn default() -> Self {
13 Self::with_max_size_bytes(DEFAULT_MAX_SIZE_BYTES)
14 }
15}
16
17/// Eviction target ratio: when over limit, evict down to this fraction of max size
18const EVICTION_TARGET_RATIO: f64 = 0.5;
19#[derive(Clone)]
20pub struct CacheEntry {
21 offset: u64,
22 data: bytes::Bytes, // Zero-copy immutable buffer
23 dirty: bool,
24 /// Monotonic insertion sequence number used for LRU eviction ordering.
25 /// Lower `seq` = older insertion = evicted first.
26 seq: u64,
27}
28
29impl CacheEntry {
30 pub fn offset(&self) -> u64 {
31 self.offset
32 }
33
34 pub fn data(&self) -> &[u8] {
35 &self.data
36 }
37
38 pub fn is_dirty(&self) -> bool {
39 self.dirty
40 }
41
42 /// Consume the entry and return the underlying `Bytes` buffer.
43 ///
44 /// This enables zero-copy transfer of cached data to the disk writer
45 /// without cloning the buffer. The entry's `Bytes` is an `Arc`-backed
46 /// buffer, so moving it out is O(1) — no data copy occurs.
47 pub fn into_data(self) -> bytes::Bytes {
48 self.data
49 }
50
51 /// Returns the memory size of this entry's data in bytes
52 fn size_bytes(&self) -> usize {
53 self.data.len()
54 }
55}
56
57/// Write-back disk cache with LRU eviction and bounded memory usage.
58///
59/// `WrDiskCache` buffers disk writes before flushing them to persistent storage.
60/// It uses an LRU (Least Recently Used) eviction policy that **never evicts dirty
61/// (unflushed) entries**, guaranteeing no data loss under memory pressure.
62///
63/// Entries are keyed by their start offset in a [`BTreeMap`], enabling O(log n)
64/// range lookups for `read()`. LRU ordering is preserved via a per-entry monotonic
65/// `seq` number — during eviction the clean entry with the smallest `seq` (oldest
66/// insertion) is removed first.
67pub struct WrDiskCache {
68 /// Cache entries keyed by start offset, enabling O(log n) range queries.
69 entries: Mutex<BTreeMap<u64, CacheEntry>>,
70 /// Maximum allowed cache size in bytes
71 max_size_bytes: usize,
72 /// Current total cached data size in bytes (atomic for lock-free reads)
73 total_cached_bytes: AtomicUsize,
74 /// Monotonic counter assigning insertion sequence numbers for LRU ordering.
75 next_seq: AtomicU64,
76}
77
78impl WrDiskCache {
79 /// Create a new `WrDiskCache` with a maximum size specified in megabytes.
80 ///
81 /// # Arguments
82 /// * `max_size_mb` - Maximum cache capacity in megabytes
83 ///
84 /// # Example
85 /// ```ignore
86 /// let cache = WrDiskCache::new(16); // 16 MB max
87 /// ```
88 pub fn new(max_size_mb: usize) -> Self {
89 let max_size_bytes = max_size_mb * 1024 * 1024;
90
91 debug!(
92 "Initializing write-back disk cache, max capacity: {} MB ({} bytes)",
93 max_size_mb, max_size_bytes
94 );
95
96 WrDiskCache {
97 entries: Mutex::new(BTreeMap::new()),
98 max_size_bytes,
99 total_cached_bytes: AtomicUsize::new(0),
100 next_seq: AtomicU64::new(0),
101 }
102 }
103
104 /// Create a new `WrDiskCache` with a maximum size specified in bytes.
105 ///
106 /// This provides finer-grained control than [`WrDiskCache::new`] which takes megabytes.
107 ///
108 /// # Arguments
109 /// * `max_size_bytes` - Maximum cache capacity in bytes
110 pub fn with_max_size_bytes(max_size_bytes: usize) -> Self {
111 debug!(
112 "Initializing write-back disk cache, max capacity: {} bytes",
113 max_size_bytes
114 );
115
116 WrDiskCache {
117 entries: Mutex::new(BTreeMap::new()),
118 max_size_bytes,
119 total_cached_bytes: AtomicUsize::new(0),
120 next_seq: AtomicU64::new(0),
121 }
122 }
123
124 /// Returns the maximum cache size in bytes.
125 pub fn max_size_bytes(&self) -> usize {
126 self.max_size_bytes
127 }
128
129 /// Returns the current approximate cache size in bytes (lock-free).
130 ///
131 /// Note: This is an atomic snapshot and may be slightly stale if a write
132 /// or eviction is concurrently in progress.
133 pub fn current_size_bytes(&self) -> usize {
134 self.total_cached_bytes.load(Ordering::Relaxed)
135 }
136
137 /// Write data at the given offset into the cache.
138 ///
139 /// If writing this entry would exceed `max_size_bytes`, LRU eviction is triggered.
140 /// Only clean (already-flushed) entries are eligible for eviction — dirty entries
141 /// are never evicted to prevent data loss. If insufficient clean entries exist
142 /// to make room, the cache may temporarily exceed its limit until a flush occurs.
143 ///
144 /// # Zero-Copy
145 /// This method accepts `bytes::Bytes` which enables zero-copy slicing.
146 /// The caller can pass a slice of a larger buffer without copying.
147 pub async fn write(&self, offset: u64, data: bytes::Bytes) -> Result<()> {
148 let entry_size = data.len();
149 let seq = self.next_seq.fetch_add(1, Ordering::Relaxed);
150
151 // Pre-check: if adding this entry would exceed the limit, try eviction first
152 let current = self.total_cached_bytes.load(Ordering::Relaxed);
153 if current.saturating_add(entry_size) > self.max_size_bytes {
154 self.evict_clean_entries(entry_size).await;
155 }
156
157 let mut entries = self.entries.lock().await;
158
159 // Re-check after acquiring lock (another task may have changed things)
160 let current_locked = self.total_cached_bytes.load(Ordering::Relaxed);
161 if current_locked.saturating_add(entry_size) > self.max_size_bytes {
162 // Try again with lock held for precise accounting
163 self.evict_clean_entries_locked(&mut entries, entry_size);
164 }
165
166 // Insert the new entry. If an entry already exists at this offset, it
167 // is replaced — the new write supersedes the old one. Subtract the old
168 // entry's size so total_cached_bytes stays accurate.
169 let old_size = entries
170 .insert(
171 offset,
172 CacheEntry {
173 offset,
174 data,
175 dirty: true,
176 seq,
177 },
178 )
179 .map(|old| old.size_bytes());
180 if let Some(old) = old_size {
181 self.total_cached_bytes.fetch_sub(old, Ordering::Relaxed);
182 }
183 self.total_cached_bytes
184 .fetch_add(entry_size, Ordering::Relaxed);
185
186 debug!(
187 "Wrote to cache, offset: {}, size: {}, cache usage: {}/{} bytes",
188 offset,
189 entry_size,
190 self.total_cached_bytes.load(Ordering::Relaxed),
191 self.max_size_bytes
192 );
193
194 Ok(())
195 }
196
197 /// Read cached data at the given offset and length.
198 ///
199 /// Returns `Some(data)` if the requested range is fully contained in a cached entry,
200 /// or `None` if the data is not in the cache.
201 ///
202 /// # Complexity
203 /// O(log n) — a single `BTreeMap::range` lookup finds the unique candidate
204 /// entry (the one with the largest start key `<= offset`). If that entry
205 /// does not fully cover `[offset, offset+length)`, no other entry can
206 /// (entries with smaller keys end before `offset`; entries with larger keys
207 /// start after `offset`).
208 ///
209 /// # Zero-Copy
210 /// Returns `bytes::Bytes` slice instead of `Vec<u8>`, avoiding memory allocation.
211 pub async fn read(&self, offset: u64, length: u64) -> Result<Option<bytes::Bytes>> {
212 let entries = self.entries.lock().await;
213
214 let end = offset + length;
215
216 // The only candidate is the entry with the largest key <= offset.
217 // `range(..=offset).next_back()` returns exactly that in O(log n).
218 if let Some((&entry_offset, entry)) = entries.range(..=offset).next_back() {
219 let entry_end = entry_offset + entry.data.len() as u64;
220 if entry_end >= end {
221 let start = (offset - entry_offset) as usize;
222 let slice_end = start + length as usize;
223 if slice_end <= entry.data.len() {
224 // Zero-copy slice
225 return Ok(Some(entry.data.slice(start..slice_end)));
226 }
227 }
228 }
229
230 Ok(None)
231 }
232
233 /// Flush all dirty entries, returning them for persistence.
234 ///
235 /// After flushing, entries remain in the cache but are marked as clean
236 /// (eligible for future LRU eviction). The caller is responsible for
237 /// writing the returned entries to durable storage.
238 ///
239 /// The returned `CacheEntry` clones are O(1): `bytes::Bytes` is an
240 /// `Arc`-backed buffer, so cloning only bumps a reference count — no
241 /// data copy occurs.
242 pub async fn flush(&self) -> Result<Vec<CacheEntry>> {
243 let mut entries = self.entries.lock().await;
244
245 let mut flushed = Vec::new();
246 for entry in entries.values_mut() {
247 if entry.dirty {
248 // Clone is O(1): bytes::Bytes is Arc-backed (refcount bump only).
249 flushed.push(entry.clone());
250 entry.dirty = false;
251 }
252 }
253
254 debug!("Flushed {} dirty cache entries", flushed.len());
255
256 Ok(flushed)
257 }
258
259 /// Clear all entries from the cache and reset size tracking.
260 pub async fn clear(&self) -> Result<()> {
261 let mut entries = self.entries.lock().await;
262
263 let cleared_bytes: usize = entries.values().map(|e| e.size_bytes()).sum();
264 entries.clear();
265 self.total_cached_bytes
266 .fetch_sub(cleared_bytes, Ordering::Relaxed);
267
268 debug!("Cleared cache ({} bytes)", cleared_bytes);
269 Ok(())
270 }
271
272 /// Returns the current total size of cached data in bytes.
273 pub async fn size(&self) -> usize {
274 self.total_cached_bytes.load(Ordering::Relaxed)
275 }
276
277 /// Returns true if the cache contains no entries.
278 pub async fn is_empty(&self) -> bool {
279 self.size().await == 0
280 }
281
282 /// Returns the number of entries in the cache.
283 pub async fn count(&self) -> usize {
284 self.entries.lock().await.len()
285 }
286
287 /// Returns the number of dirty (unflushed) entries.
288 pub async fn dirty_count(&self) -> usize {
289 self.entries
290 .lock()
291 .await
292 .values()
293 .filter(|e| e.dirty)
294 .count()
295 }
296
297 // -----------------------------------------------------------------------
298 // LRU Eviction methods
299 // -----------------------------------------------------------------------
300
301 /// Evict clean (non-dirty) entries to make room for `needed_size` additional bytes.
302 ///
303 /// This method acquires the entries lock internally. For use when already holding
304 /// the lock, see [`evict_clean_entries_locked`](Self::evict_clean_entries_locked).
305 ///
306 /// # Invariant
307 /// Dirty entries are NEVER evicted. If all remaining entries are dirty and we still
308 /// need space, the cache will temporarily exceed its limit rather than lose data.
309 async fn evict_clean_entries(&self, needed_size: usize) {
310 let mut entries = self.entries.lock().await;
311 self.evict_clean_entries_locked(&mut entries, needed_size);
312 }
313
314 /// Core eviction logic — must be called with `entries` lock held.
315 ///
316 /// Repeatedly removes the clean (non-dirty) entry with the smallest `seq`
317 /// (oldest insertion = LRU candidate) until either:
318 /// - We have freed enough space for `needed_size` new bytes, OR
319 /// - We have reached the eviction target (50% of max), OR
320 /// - No more clean entries remain
321 ///
322 /// Unlike the old `VecDeque` front-only eviction, the `BTreeMap` + `seq`
323 /// design can evict ANY clean entry regardless of its offset ordering —
324 /// a dirty entry no longer blocks eviction of newer clean entries behind it.
325 ///
326 /// # Complexity
327 /// Each iteration is O(n) (a full scan to find the min-`seq` clean entry),
328 /// but eviction is infrequent (only when over the memory limit), so this is
329 /// acceptable. The common `read`/`write` paths remain O(log n).
330 fn evict_clean_entries_locked(
331 &self,
332 entries: &mut BTreeMap<u64, CacheEntry>,
333 needed_size: usize,
334 ) {
335 let target = ((self.max_size_bytes as f64) * EVICTION_TARGET_RATIO) as usize;
336 let mut evicted_count = 0usize;
337 let mut evicted_bytes = 0usize;
338
339 // Keep evicting while current size + needed > target (we still need room)
340 while self
341 .total_cached_bytes
342 .load(Ordering::Relaxed)
343 .saturating_add(needed_size)
344 > target
345 {
346 // Find the clean entry with the smallest seq (true LRU order).
347 // Iterating all entries is O(n), but eviction is rare.
348 let evict_key = entries
349 .iter()
350 .filter(|(_, e)| !e.dirty)
351 .min_by_key(|(_, e)| e.seq)
352 .map(|(&k, _)| k);
353
354 match evict_key {
355 Some(key) => {
356 if let Some(entry) = entries.remove(&key) {
357 let entry_size = entry.size_bytes();
358 self.total_cached_bytes
359 .fetch_sub(entry_size, Ordering::Relaxed);
360 evicted_bytes += entry_size;
361 evicted_count += 1;
362
363 debug!(
364 "Evicted clean cache entry (seq {}), offset: {}, size: {} bytes",
365 entry.seq, entry.offset, entry_size
366 );
367 }
368 }
369 None => {
370 // No clean entries remain — cannot evict without losing data.
371 // The cache may temporarily overshoot, which is safe.
372 debug!(
373 "Eviction blocked: all {} remaining entries are dirty",
374 entries.len()
375 );
376 break;
377 }
378 }
379 }
380
381 if evicted_count > 0 {
382 debug!(
383 "LRU eviction complete: evicted {} entries ({} bytes), cache now ~{} bytes",
384 evicted_count,
385 evicted_bytes,
386 self.total_cached_bytes.load(Ordering::Relaxed)
387 );
388 }
389 }
390}
391
392// =========================================================================
393// Tests
394// =========================================================================
395
396#[cfg(test)]
397mod tests {
398 use super::*;
399
400 /// Helper: create a cache with a small max size (in bytes) for testing eviction behavior
401 fn make_small_cache(max_bytes: usize) -> WrDiskCache {
402 WrDiskCache::with_max_size_bytes(max_bytes)
403 }
404
405 // -----------------------------------------------------------------------
406 // Basic functionality tests
407 // -----------------------------------------------------------------------
408
409 #[tokio::test]
410 async fn test_new_constructor_mb() {
411 let cache = WrDiskCache::new(16); // 16 MB
412 assert_eq!(cache.max_size_bytes(), 16 * 1024 * 1024);
413 assert_eq!(cache.current_size_bytes(), 0);
414 assert!(cache.is_empty().await);
415 }
416
417 #[tokio::test]
418 async fn test_with_max_size_bytes_constructor() {
419 let cache = WrDiskCache::with_max_size_bytes(1024); // 1 KB
420 assert_eq!(cache.max_size_bytes(), 1024);
421 assert_eq!(cache.current_size_bytes(), 0);
422 }
423
424 #[tokio::test]
425 async fn test_write_and_read() {
426 let cache = make_small_cache(4096);
427
428 cache.write(0, bytes::Bytes::from("hello")).await.unwrap();
429 cache.write(100, bytes::Bytes::from("world")).await.unwrap();
430
431 assert_eq!(cache.size().await, 10);
432 assert_eq!(cache.count().await, 2);
433
434 // Read back by exact offset
435 let result = cache.read(0, 5).await.unwrap();
436 assert!(result.is_some());
437 assert_eq!(&result.unwrap()[..], b"hello");
438
439 let result = cache.read(100, 5).await.unwrap();
440 assert!(result.is_some());
441 assert_eq!(&result.unwrap()[..], b"world");
442 }
443
444 #[tokio::test]
445 async fn test_flush_returns_dirty_entries() {
446 let cache = make_small_cache(4096);
447
448 cache.write(0, bytes::Bytes::from("data1")).await.unwrap();
449 cache.write(10, bytes::Bytes::from("data2")).await.unwrap();
450
451 assert_eq!(cache.dirty_count().await, 2);
452
453 let flushed = cache.flush().await.unwrap();
454 assert_eq!(flushed.len(), 2);
455
456 // After flush, entries are no longer dirty
457 assert_eq!(cache.dirty_count().await, 0);
458 // But they're still in the cache
459 assert_eq!(cache.count().await, 2);
460 }
461
462 #[tokio::test]
463 async fn test_clear_resets_cache() {
464 let cache = make_small_cache(4096);
465
466 cache
467 .write(0, bytes::Bytes::from(vec![0x42; 100]))
468 .await
469 .unwrap();
470 assert_eq!(cache.size().await, 100);
471
472 cache.clear().await.unwrap();
473 assert_eq!(cache.size().await, 0);
474 assert!(cache.is_empty().await);
475 assert_eq!(cache.count().await, 0);
476 }
477
478 // -----------------------------------------------------------------------
479 // LRU Eviction: clean entries are evicted under memory pressure
480 // -----------------------------------------------------------------------
481
482 #[tokio::test]
483 async fn test_cache_eviction_under_memory_pressure() {
484 // Cache max = 500 bytes. Write 6 entries of 100 bytes each.
485 // After exceeding the limit, eviction should kick in for CLEAN entries.
486 let cache = make_small_cache(500);
487
488 // Phase 1: Write 3 dirty entries (300 bytes) — under limit
489 cache
490 .write(0, bytes::Bytes::from(vec![0u8; 100]))
491 .await
492 .unwrap(); // entry A: offset=0
493 cache
494 .write(100, bytes::Bytes::from(vec![1u8; 100]))
495 .await
496 .unwrap(); // entry B: offset=100
497 cache
498 .write(200, bytes::Bytes::from(vec![2u8; 100]))
499 .await
500 .unwrap(); // entry C: offset=200
501 assert_eq!(cache.count().await, 3);
502 assert_eq!(cache.size().await, 300);
503
504 // Flush to make them clean
505 cache.flush().await.unwrap();
506 assert_eq!(cache.dirty_count().await, 0);
507 assert_eq!(cache.count().await, 3); // Still present
508
509 // Phase 2: Write more entries that will trigger eviction
510 // Writing 300 more bytes would exceed 500 limit → should evict old clean ones
511 cache
512 .write(300, bytes::Bytes::from(vec![3u8; 100]))
513 .await
514 .unwrap(); // entry D: dirty
515 cache
516 .write(400, bytes::Bytes::from(vec![4u8; 100]))
517 .await
518 .unwrap(); // entry E: dirty
519
520 // At this point we have 5 entries (500 bytes). Adding one more triggers eviction.
521 cache
522 .write(500, bytes::Bytes::from(vec![5u8; 100]))
523 .await
524 .unwrap(); // entry F: dirty
525
526 // The oldest CLEAN entries (A, B, C) should have been evicted to make room.
527 // Only D, E, F should remain (or possibly some of A/B/C if not all were evicted).
528 // Key invariant: total size should be roughly bounded (may slightly exceed due to
529 // dirty-only entries blocking eviction).
530 let final_count = cache.count().await;
531 let final_size = cache.size().await;
532
533 // We wrote 600 bytes into a 500-byte cache. Since A,B,C became clean before
534 // D,E,F were written, at least some of them should have been evicted.
535 // The cache should NOT contain all 6 entries.
536 assert!(
537 final_count <= 5,
538 "Expected at most 5 entries after eviction, got {}",
539 final_count
540 );
541
542 // The remaining entries should be the newer ones (D, E, F and possibly one of A/B/C)
543 // Verify the oldest clean entry (A at offset 0) was likely evicted
544 let entry_0 = cache.read(0, 100).await.unwrap();
545 // Entry A (offset 0) was the oldest clean — it should be gone
546 assert!(
547 entry_0.is_none(),
548 "Oldest clean entry (offset 0) should have been evicted"
549 );
550
551 debug!(
552 "Eviction test: final count={}, final_size={}, expected ~<=500",
553 final_count, final_size
554 );
555 }
556
557 #[tokio::test]
558 async fn test_dirty_entries_are_never_evicted() {
559 // This is the critical safety property: dirty entries MUST survive eviction.
560 // Cache max = 400 bytes.
561 let cache = make_small_cache(400);
562
563 // Write 4 dirty entries (400 bytes = exactly at limit)
564 cache
565 .write(0, bytes::Bytes::from(vec![0xAA; 100]))
566 .await
567 .unwrap(); // dirty A
568 cache
569 .write(100, bytes::Bytes::from(vec![0xBB; 100]))
570 .await
571 .unwrap(); // dirty B
572 cache
573 .write(200, bytes::Bytes::from(vec![0xCC; 100]))
574 .await
575 .unwrap(); // dirty C
576 cache
577 .write(300, bytes::Bytes::from(vec![0xDD; 100]))
578 .await
579 .unwrap(); // dirty D
580
581 assert_eq!(cache.dirty_count().await, 4);
582 assert_eq!(cache.count().await, 4);
583
584 // Now try to write another entry that would exceed the limit.
585 // All existing entries are dirty, so NONE can be evicted.
586 // The write must still succeed (cache may temporarily overshoot).
587 cache
588 .write(400, bytes::Bytes::from(vec![0xEE; 100]))
589 .await
590 .unwrap(); // dirty E
591
592 // ALL 5 dirty entries must still be present — zero data loss allowed
593 assert_eq!(
594 cache.count().await,
595 5,
596 "All dirty entries must be preserved — none should be evicted"
597 );
598 assert_eq!(
599 cache.dirty_count().await,
600 5,
601 "All entries must still be dirty"
602 );
603
604 // Verify each entry's data is intact
605 for (offset, byte_val) in [
606 (0u64, 0xAAu8),
607 (100, 0xBB),
608 (200, 0xCC),
609 (300, 0xDD),
610 (400, 0xEE),
611 ] {
612 let result = cache.read(offset, 100).await.unwrap();
613 assert!(
614 result.is_some(),
615 "Dirty entry at offset {} must still exist",
616 offset
617 );
618 let data = result.unwrap();
619 assert!(
620 data.iter().all(|&b| b == byte_val),
621 "Data integrity check failed for offset {}: expected 0x{:02X}",
622 offset,
623 byte_val
624 );
625 }
626
627 // Flush should return all 5 entries
628 let flushed = cache.flush().await.unwrap();
629 assert_eq!(flushed.len(), 5, "Flush must return all 5 dirty entries");
630 }
631
632 #[tokio::test]
633 async fn test_mixed_dirty_and_clean_eviction() {
634 // Cache max = 500 bytes.
635 // Mix of dirty and clean entries: only clean ones should be evicted.
636 let cache = make_small_cache(500);
637
638 // Write and flush (make clean) some older entries
639 cache
640 .write(0, bytes::Bytes::from(vec![1u8; 100]))
641 .await
642 .unwrap(); // will become clean
643 cache
644 .write(100, bytes::Bytes::from(vec![2u8; 100]))
645 .await
646 .unwrap(); // will become clean
647 cache.flush().await.unwrap(); // Mark A, B as clean
648
649 // Write new dirty entries
650 cache
651 .write(200, bytes::Bytes::from(vec![3u8; 100]))
652 .await
653 .unwrap(); // dirty C
654 cache
655 .write(300, bytes::Bytes::from(vec![4u8; 100]))
656 .await
657 .unwrap(); // dirty D
658
659 // Now: A(clean), B(clean), C(dirty), D(dirty) = 400 bytes
660 // Write more to trigger eviction
661 cache
662 .write(400, bytes::Bytes::from(vec![5u8; 100]))
663 .await
664 .unwrap(); // dirty E — now 500 bytes
665 cache
666 .write(500, bytes::Bytes::from(vec![6u8; 100]))
667 .await
668 .unwrap(); // dirty F — exceeds 500, triggers eviction
669
670 // Clean entries A and/or B should be evicted; C,D,E,F (dirty) must remain
671 let dirty_cnt = cache.dirty_count().await;
672 let total_cnt = cache.count().await;
673
674 // All 4 dirty entries (C, D, E, F) must survive
675 assert!(
676 dirty_cnt >= 4,
677 "At least 4 dirty entries must survive, got {}",
678 dirty_cnt
679 );
680
681 // At least some clean entries should have been evicted
682 assert!(
683 total_cnt <= 6,
684 "Total entries should be bounded, got {}",
685 total_cnt
686 );
687
688 // Verify dirty entries' data is intact
689 for (offset, expected_byte) in [(200, 3u8), (300, 4), (400, 5), (500, 6)] {
690 let result = cache.read(offset, 100).await.unwrap();
691 assert!(
692 result.is_some(),
693 "Dirty entry at offset {} must survive eviction",
694 offset
695 );
696 assert_eq!(result.unwrap()[0], expected_byte);
697 }
698
699 debug!("Mixed eviction: total={}, dirty={}", total_cnt, dirty_cnt);
700 }
701
702 #[tokio::test]
703 async fn test_flush_then_evict_frees_space() {
704 // Verify the lifecycle: write → flush (clean) → write more → evicts old clean
705 let cache = make_small_cache(300);
706
707 // Fill with dirty entries, then flush them
708 cache
709 .write(0, bytes::Bytes::from(vec![0u8; 150]))
710 .await
711 .unwrap();
712 cache.flush().await.unwrap(); // Now clean
713 assert_eq!(cache.dirty_count().await, 0);
714
715 // Write new dirty entry — should trigger eviction of the clean one
716 cache
717 .write(200, bytes::Bytes::from(vec![1u8; 150]))
718 .await
719 .unwrap();
720
721 // The first entry (now clean) may or may not have been evicted depending on
722 // whether 300 + 150 > 300 triggered it. With our logic, 150 + 150 = 300 which
723 // is NOT > 300, so no eviction yet. One more write should trigger it.
724 cache
725 .write(400, bytes::Bytes::from(vec![2u8; 150]))
726 .await
727 .unwrap(); // 450 > 300 → evict!
728
729 // Old clean entry at offset 0 should be evicted; new dirty entries remain
730 let _old_entry = cache.read(0, 150).await.unwrap();
731 // It may or may not be evicted depending on exact timing, but dirty entries survive
732 let new_entry = cache.read(200, 150).await.unwrap();
733 assert!(new_entry.is_some(), "Newer dirty entry must survive");
734 }
735
736 #[tokio::test]
737 async fn test_eviction_to_target_ratio() {
738 // When over limit, eviction should bring us down to ~50% of max
739 let cache = make_small_cache(1000); // 1KB max, target = 500
740
741 // Write and flush many small clean entries
742 for i in 0..20 {
743 cache
744 .write((i * 50) as u64, bytes::Bytes::from(vec![i as u8; 50]))
745 .await
746 .unwrap();
747 }
748 // 20 * 50 = 1000 bytes = exactly at limit
749
750 cache.flush().await.unwrap(); // All clean now
751
752 // Write one more entry to push over limit and trigger eviction
753 cache
754 .write(2000, bytes::Bytes::from(vec![0xFF; 50]))
755 .await
756 .unwrap(); // 1050 > 1000 → evict
757
758 // Should have evicted down to target (~500 bytes / 50 per entry ≈ 10 entries)
759 let size = cache.size().await;
760 let count = cache.count().await;
761
762 // Size should be significantly reduced from original 1050
763 assert!(
764 size <= 550, // Allow some tolerance around 50% target + new entry
765 "After eviction, size ({}) should be near target (~500), max is 1000",
766 size
767 );
768
769 debug!(
770 "Eviction to target: size={} bytes, count={} entries",
771 size, count
772 );
773 }
774
775 #[tokio::test]
776 async fn test_current_size_bytes_lock_free() {
777 // Verify current_size_bytes() works without holding the async lock
778 let cache = make_small_cache(4096);
779
780 cache
781 .write(0, bytes::Bytes::from(vec![0u8; 256]))
782 .await
783 .unwrap();
784 cache
785 .write(256, bytes::Bytes::from(vec![1u8; 256]))
786 .await
787 .unwrap();
788
789 // Lock-free read should match locked read
790 let lock_free_size = cache.current_size_bytes();
791 let locked_size = cache.size().await;
792
793 assert_eq!(lock_free_size, locked_size);
794 assert_eq!(lock_free_size, 512);
795 }
796
797 #[tokio::test]
798 async fn test_read_miss_returns_none() {
799 let cache = make_small_cache(1024);
800
801 cache.write(0, bytes::Bytes::from("hello")).await.unwrap();
802
803 // Non-existent offset
804 assert!(cache.read(999, 5).await.unwrap().is_none());
805
806 // Offset exists but length too long
807 assert!(cache.read(0, 100).await.unwrap().is_none());
808 }
809
810 #[tokio::test]
811 async fn test_range_based_read() {
812 let cache = make_small_cache(4096);
813
814 // Write an entry covering offsets 0-99
815 cache
816 .write(0, bytes::Bytes::from(vec![42u8; 100]))
817 .await
818 .unwrap();
819
820 // Read a sub-range within the entry
821 let result = cache.read(10, 30).await.unwrap();
822 assert!(result.is_some());
823 let data = result.unwrap();
824 assert_eq!(data.len(), 30);
825 assert!(data.iter().all(|&b| b == 42));
826 }
827
828 #[tokio::test]
829 async fn test_multiple_flushes_only_return_dirty() {
830 let cache = make_small_cache(4096);
831
832 cache.write(0, bytes::Bytes::from("A")).await.unwrap();
833 cache.write(1, bytes::Bytes::from("B")).await.unwrap();
834
835 // First flush returns both
836 let f1 = cache.flush().await.unwrap();
837 assert_eq!(f1.len(), 2);
838
839 // Second flush returns nothing (already clean)
840 let f2 = cache.flush().await.unwrap();
841 assert_eq!(f2.len(), 0);
842
843 // Write a third entry
844 cache.write(2, bytes::Bytes::from("C")).await.unwrap();
845
846 // Third flush returns only the new dirty entry
847 let f3 = cache.flush().await.unwrap();
848 assert_eq!(f3.len(), 1);
849 assert_eq!(f3[0].offset(), 2);
850 }
851
852 #[tokio::test]
853 async fn test_empty_cache_operations() {
854 let cache = make_small_cache(1024);
855
856 assert!(cache.is_empty().await);
857 assert_eq!(cache.size().await, 0);
858 assert_eq!(cache.count().await, 0);
859 assert_eq!(cache.dirty_count().await, 0);
860
861 // Read on empty cache
862 assert!(cache.read(0, 10).await.unwrap().is_none());
863
864 // Flush on empty cache
865 let flushed = cache.flush().await.unwrap();
866 assert!(flushed.is_empty());
867
868 // Clear on empty cache (should not panic)
869 cache.clear().await.unwrap();
870 assert!(cache.is_empty().await);
871 }
872
873 #[tokio::test]
874 async fn test_large_write_exceeding_max_with_only_dirty_entries() {
875 // Edge case: single write larger than max_size, all entries dirty
876 let cache = make_small_cache(100); // Tiny 100-byte cache
877
878 // Write a single entry larger than max
879 cache
880 .write(0, bytes::Bytes::from(vec![0u8; 200]))
881 .await
882 .unwrap();
883
884 // Must succeed without losing data (no clean entries to evict anyway)
885 assert_eq!(cache.count().await, 1);
886 assert_eq!(cache.size().await, 200);
887 assert_eq!(cache.dirty_count().await, 1);
888
889 let result = cache.read(0, 200).await.unwrap();
890 assert!(result.is_some());
891 assert_eq!(result.unwrap().len(), 200);
892 }
893}