fix: resolve full-codebase audit findings across all subsystems

Full-codebase audit (/x-overhaul): 25 findings fixed across the write/read
path, compaction, iterator, SST, WAL, cache, manifest, and memtable. 2 HIGH
findings (CRC-protected block decode, defensive ValueType decode) recorded as
Won't Fix with rationale.

Crash safety:
- Fail-stop via bg_error on any post-freeze flush failure so a later flush
  cannot advance log_number past an unflushed memtable (stale reads + lost
  recovery on restart).
- Fsync the DB directory in log_and_apply before any MANIFEST record can
  reference a newly created SST (covers flush, both compaction paths, and the
  maybe_compact_manifest snapshot).
- Make WAL recovery crash-idempotent: install the recovered SST and advance
  log_number in one edit, and never reuse a file number already on disk.
- Tolerate any corrupt-tail WAL record on recovery instead of failing open.

Concurrency:
- Serialize snapshot acquisition with compaction's snapshot-list capture
  (snapshot_seq under the DB lock; capture after picking inputs) so a
  compaction cannot GC a version a concurrently-registered snapshot needs.

Compaction correctness:
- Cut output files only at user-key boundaries (never split a key's versions,
  avoiding overlapping L1+ ranges) in both sub-compaction and force_merge_level.
- Apply range-tombstone filtering to snapshot-retained older versions to
  prevent resurrection when a covering tombstone is dropped at the bottom level.
- Delete orphaned output SSTs when installing a compaction edit fails.

Iterator correctness:
- Fix prefix-successor truncation, prefix seek_for_prev clamping, lazy
  next_back re-yield, and LevelIterator::seek_to_last on filtered files.
- Apply block_property_filters to prefix iterators.

Data integrity / robustness:
- Reject empty/inverted range deletions (avoid mis-deleting the begin key).
- Reject oversized SST entries the reader would reject; cap bloom filter size;
  reject u32 varint overflow.
- Bound the block-cache reverse index via a moka eviction listener; honor
  block_cache_capacity == 0 as a disabled cache.
- Validate num_levels >= 2.

Style/docs: grouped re-exports, error imports in sst::format, fixed-width
footer docs, test SAFETY comments. Adds 4 regression tests.

Author: ktmlm

.claude/audit.md

@@ -10,20 +10,30 @@
 
 ## Won't Fix
 
+### [HIGH] sst: corrupt block entry could read restart-array bytes / iteration degrades to EOF
+- **Where**: src/sst/block.rs:73-97, 411-528; src/sst/table_reader.rs:379-392
+- **What**: Entry decoders bound key/value lengths against `data.len()` rather than the data-region end (`restart_offset`), and a mid-block decode failure surfaces as iterator EOF rather than an error.
+- **Reason**: Data blocks are CRC32-verified before they are ever decoded (`table_reader.rs:446-452` rejects any block whose checksum does not match), so a corrupt block never reaches the decoders. The decoders already guard against out-of-bounds with `checked_add` + `value_end > data.len()`. Reaching the reported state requires a CRC32 collision (~2^-32). The suggested remedy (thread `restart_offset` through `decode_entry_at`/`decode_entry_reuse` to ~6 call sites and convert block iteration to a `Result`-returning protocol) is a large, hot-path API change disproportionate to a CRC-collision-only scenario.
+
+### [HIGH] types: invalid ValueType byte decodes as Deletion instead of erroring
+- **Where**: src/types.rs:54-62, 127-130, 178-181
+- **What**: `unpack_sequence_and_type` maps an unknown type tag to `Deletion` (invisible) rather than surfacing corruption.
+- **Reason**: This is a documented, intentional defensive choice (src/types.rs:58-60): all persisted internal keys come from CRC32-protected storage (WAL records and SST blocks are checksummed before decode), so an invalid tag is unreachable in practice; when it is reached (memory corruption / future format drift) defaulting to the *invisible* Deletion is strictly safer than the *phantom-data* Value. `value_type()` is on the hot read path (decoded on every key comparison/lookup); making it fallible across all SST/iterator/compaction call sites is disproportionate to a CRC-collision-only scenario.
+
 ### [MEDIUM] db: `do_compaction` holds db_mutex during blocking I/O
-- **Where**: src/db.rs:2405-2446
+- **Where**: src/db.rs:2438-2493
 - **What**: `do_compaction()` holds the inner mutex across `execute_compaction_with_cache` and `force_merge_level`, blocking all writers for the entire compaction duration.
 - **Reason**: Refactoring to a 3-phase pattern (lock → I/O → lock) requires `do_compaction` to not take `&self` with the lock, which is a deep architectural change. Only triggered by explicit `compact()`/`flush()` API calls, not the background compaction path.
 
 ### [MEDIUM] db: `freeze_and_flush` holds db_mutex during SST write I/O
-- **Where**: src/db.rs:2288-2299
+- **Where**: src/db.rs:2295-2320
 - **What**: Writes an entire SST file while the caller holds the DB lock.
 - **Reason**: Only called from `close()` (shutdown path). Refactoring is disproportionate risk for a non-hot path. The background flush path correctly uses the 3-phase pattern.
 
 ### [MEDIUM] manifest: `log_and_apply` performs I/O via `maybe_compact_manifest` while caller holds DB mutex
-- **Where**: src/manifest/version_set.rs:316, 406-414
-- **What**: Every 1000th edit, `maybe_compact_manifest` fires from inside `log_and_apply`, performing file create, write, fsync, rename, directory fsync, and file removal — all while the caller holds the main DB mutex.
-- **Reason**: Fixing requires separating `VersionSet` from the DB mutex (new manifest-specific lock), which is a deep architectural refactor. The current behavior causes a periodic write-stall spike (~10-100ms every 1000 edits) but does not affect correctness.
+- **Where**: src/manifest/version_set.rs:317, 421-466
+- **What**: Every 1000th edit, `maybe_compact_manifest` fires from inside `log_and_apply`, performing file create, write, fsync, rename, directory fsync, and file removal — all while the caller holds the main DB mutex. (`log_and_apply` now also fsyncs the DB directory when the edit adds files, for crash-safety.)
+- **Reason**: Fixing requires separating `VersionSet` from the DB mutex (new manifest-specific lock), which is a deep architectural refactor. The current behavior causes a periodic write-stall spike but does not affect correctness.
 
 ### [LOW] db: `max_immutable_memtables` option declared but not enforced
 - **Where**: src/options.rs:18-19
@@ -36,6 +46,6 @@
 - **Reason**: Public API for external consumers; cannot determine if unused without checking downstream crates.
 
 ### [LOW] types: "Safety:" comment in non-unsafe context
-- **Where**: src/types.rs:49-52
+- **Where**: src/types.rs:54-62
 - **What**: Comment uses "Safety:" but no unsafe block is involved.
 - **Reason**: Comment is a design note, not a safety justification. Cosmetic only.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mmdb"
-version = "3.2.4"
+version = "3.2.5"
 edition = "2024"
 description = "The storage engine behind vsdb — a pure-Rust LSM-Tree key-value store"
 license = "MIT"

src/cache/block_cache.rs

@@ -3,6 +3,7 @@
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
+use moka::notification::RemovalCause;
 use parking_lot::Mutex;
 
 /// Cache key: (sst_file_number, block_offset).
@@ -11,33 +12,61 @@ type CacheKey = (u64, u64);
 /// Cached block data.
 type CacheValue = Arc<Vec<u8>>;
 
+/// Reverse index: which block offsets belong to each file, used for bulk
+/// invalidation. Shared with the moka eviction listener so LRU-evicted offsets
+/// are pruned automatically (otherwise it would grow without bound).
+type FileOffsets = Arc<Mutex<HashMap<u64, HashSet<u64>>>>;
+
 /// Thread-safe LRU cache for SST data blocks.
 /// Supports pinning entries that should never be evicted (e.g., L0 index/filter blocks).
 pub struct BlockCache {
     inner: moka::sync::Cache<CacheKey, CacheValue>,
     /// Pinned entries — never evicted by LRU. Protected by mutex.
     pinned: Mutex<HashMap<CacheKey, CacheValue>>,
     /// Track which offsets belong to each file for bulk invalidation.
-    file_offsets: Mutex<HashMap<u64, HashSet<u64>>>,
+    file_offsets: FileOffsets,
+    /// When true (capacity 0), caching is disabled: inserts are no-ops and
+    /// lookups always miss. Honors the documented "0 disables caching" option.
+    disabled: bool,
 }
 
 impl BlockCache {
     /// Create a new block cache with the given capacity in bytes.
+    /// A capacity of 0 disables caching entirely.
     pub fn new(capacity_bytes: u64) -> Self {
+        let file_offsets: FileOffsets = Arc::new(Mutex::new(HashMap::new()));
+        let listener_offsets = file_offsets.clone();
+        let inner = moka::sync::Cache::builder()
+            .max_capacity(capacity_bytes)
+            .weigher(|_key: &CacheKey, value: &CacheValue| -> u32 {
+                value.len().min(u32::MAX as usize) as u32
+            })
+            .eviction_listener(
+                move |key: Arc<CacheKey>, _value: CacheValue, _cause: RemovalCause| {
+                    let (file_number, block_offset) = *key;
+                    let mut map = listener_offsets.lock();
+                    if let Some(set) = map.get_mut(&file_number) {
+                        set.remove(&block_offset);
+                        if set.is_empty() {
+                            map.remove(&file_number);
+                        }
+                    }
+                },
+            )
+            .build();
         Self {
-            inner: moka::sync::Cache::builder()
-                .max_capacity(capacity_bytes)
-                .weigher(|_key: &CacheKey, value: &CacheValue| -> u32 {
-                    value.len().min(u32::MAX as usize) as u32
-                })
-                .build(),
+            inner,
             pinned: Mutex::new(HashMap::new()),
-            file_offsets: Mutex::new(HashMap::new()),
+            file_offsets,
+            disabled: capacity_bytes == 0,
         }
     }
 
     /// Look up a cached block. Pinned entries are checked first.
     pub fn get(&self, file_number: u64, block_offset: u64) -> Option<Arc<Vec<u8>>> {
+        if self.disabled {
+            return None;
+        }
         let key = (file_number, block_offset);
         if let Some(v) = self.pinned.lock().get(&key) {
             return Some(v.clone());
@@ -48,6 +77,9 @@ impl BlockCache {
     /// Insert a block into the cache.
     pub fn insert(&self, file_number: u64, block_offset: u64, data: Vec<u8>) -> Arc<Vec<u8>> {
         let arc = Arc::new(data);
+        if self.disabled {
+            return arc;
+        }
         self.inner.insert((file_number, block_offset), arc.clone());
         self.file_offsets
             .lock()
@@ -65,8 +97,11 @@ impl BlockCache {
         block_offset: u64,
         data: Vec<u8>,
     ) -> Arc<Vec<u8>> {
-        let key = (file_number, block_offset);
         let arc = Arc::new(data);
+        if self.disabled {
+            return arc;
+        }
+        let key = (file_number, block_offset);
         self.pinned.lock().insert(key, arc.clone());
         self.file_offsets
             .lock()
@@ -96,7 +131,9 @@ impl BlockCache {
 
     /// Invalidate a specific cached block.
     pub fn invalidate(&self, file_number: u64, block_offset: u64) {
-        self.inner.invalidate(&(file_number, block_offset));
+        let key = (file_number, block_offset);
+        self.pinned.lock().remove(&key);
+        self.inner.invalidate(&key);
         if let Some(offsets) = self.file_offsets.lock().get_mut(&file_number) {
             offsets.remove(&block_offset);
         }