fix: complete audit — fix all remaining 35 medium and low severity bugs

Storage layer (4 fixes):
- parallel_wal: use writeAll instead of write to handle short writes
- parallel_wal: add completed counter to prevent torn writes (flusher
  only flushes fully-written entries)
- parallel_wal: remove fetchSub rollback on segment-full (eliminates race)
- page: leafRead returns null instead of silently truncating

Indexing subsystem (10 fixes):
- codeindex: indexBatch caps at 64 entries (prevents stack overflow)
- codeindex: indexBatch errdefer frees trigram slices on error
- codeindex: SparseNgramIndex now dupes path strings (was use-after-free)
- codeindex: SparseNgramIndex now has mutex for thread safety
- codeindex: removeFileById cleans up path_to_id and id_to_path
- fast_index: add mutex to FastTrigramIndex
- trigram: change tombstone from 0 to maxInt(u64) (0 is valid doc_id)
- disk_index: cap extractSparseNgrams loop at weights[1024] (OOB fix)
- disk_index: validate mmap size in DiskIndex.open (corrupt file safety)
- disk_index: add errdefer for mmap/fd cleanup on partial open failure

Core features (6 fixes):
- mvcc: GC rolls back on OOM instead of losing version entries
- mvcc: beginRead uses cmpxchgWeak loop for atomic min_active_epoch update
- auth: enabled flag is now atomic (eliminates auth bypass race window)
- auth: addKey returns null when MAX_KEYS reached (was silent failure)
- branch: use StringHashMap with actual key (was FNV-1a hash — collisions
  caused data loss)
- collection: round-robin index queue assignment (prevents double-processing)

CDC/query/errors (6 fixes):
- server: scan/search adds "truncated":true when results exceed buffer
- cdc: add events_dropped atomic counter (was silent event loss)
- cdc: increase event buffers + add truncated flag
- cdc: escape JSON in webhook payloads (was injection vector)
- query: increase field name buffer from 256 to 1024 bytes
- errors: escape detail string in JSON error responses

Advanced data structures (6 fixes):
- lsm: only drop tombstones at max level (prevents data resurrection)
- lsm: add RwLock for flush/get synchronization (prevents torn reads)
- compression: LZ4 hash table u16→u32 (fixes corruption >64KB)
- art: hold write lock through prefix mismatch (closes TOCTOU window)
- bwtree: epoch-based reclamation with 2-epoch grace period
- turboquant: dequantize handles FWHT mode (was crash on empty slice)

Replication (7 fixes):
- integration: replace globals with per-instance context
- integration: heap-allocate executor (was dangling ptr to optional field)
- integration: bounds check in packTxnData (was stack overflow)
- integration: free allocations on submit failure (was memory leak)
- shard: registerPartition also updates hash ring
- router: parseU16 uses checked arithmetic (was wrapping to wrong partition)
- sequencer: owns_transactions flag prevents freeing stack memory

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: justrach

src/art.zig

@@ -490,9 +490,13 @@ pub const ART = struct {
             }
 
             if (mismatch < prefix_len) {
-                // Prefix mismatch: split the node
-                hdr.writeUnlock();
-                const new_node = try Node4.create(self.alloc);
+                // Prefix mismatch: split the node.
+                // Keep write lock held through the entire operation to prevent
+                // TOCTOU races (another thread modifying prefix between unlock/relock).
+                const new_node = Node4.create(self.alloc) catch |err| {
+                    hdr.writeUnlock();
+                    return err;
+                };
 
                 // New node's prefix is the common part
                 new_node.header.prefix_len = @intCast(mismatch);
@@ -508,7 +512,6 @@ pub const ART = struct {
                 }
 
                 // Adjust existing node's prefix (remove consumed part)
-                hdr.writeLock();
                 const remaining = prefix_len - mismatch - 1;
                 if (remaining > 0 and mismatch + 1 < MAX_PREFIX_LEN) {
                     var dst: [MAX_PREFIX_LEN]u8 = [_]u8{0} ** MAX_PREFIX_LEN;

src/auth.zig

@@ -44,34 +44,36 @@ pub const AuthContext = struct {
 pub const AuthStore = struct {
     keys: [MAX_KEYS]KeyEntry = undefined,
     count: u32 = 0,
-    enabled: bool = false,
+    enabled: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
     lock: std.Thread.RwLock = .{},
 
-    /// Add an API key. Returns the BLAKE3 hash for storage.
-    pub fn addKey(self: *AuthStore, raw_key: []const u8, name: []const u8, perm: Permission) [32]u8 {
+    /// Add an API key. Returns the BLAKE3 hash if added, null if at MAX_KEYS.
+    pub fn addKey(self: *AuthStore, raw_key: []const u8, name: []const u8, perm: Permission) ?[32]u8 {
         return self.addKeyForTenant(raw_key, name, "default", perm);
     }
 
-    pub fn addKeyForTenant(self: *AuthStore, raw_key: []const u8, name: []const u8, tenant_id: []const u8, perm: Permission) [32]u8 {
+    /// Add an API key for a specific tenant. Returns the BLAKE3 hash if added,
+    /// null if the key store is full (MAX_KEYS reached).
+    pub fn addKeyForTenant(self: *AuthStore, raw_key: []const u8, name: []const u8, tenant_id: []const u8, perm: Permission) ?[32]u8 {
         self.lock.lock();
         defer self.lock.unlock();
 
+        if (self.count >= MAX_KEYS) return null;
+
         const hash = crypto.blake3(raw_key);
-        if (self.count < MAX_KEYS) {
-            var entry = KeyEntry{
-                .hash = hash,
-                .name = undefined,
-                .name_len = @intCast(@min(name.len, 64)),
-                .tenant_id = undefined,
-                .tenant_id_len = @intCast(@min(tenant_id.len, 64)),
-                .perm = perm,
-            };
-            @memcpy(entry.name[0..entry.name_len], name[0..entry.name_len]);
-            @memcpy(entry.tenant_id[0..entry.tenant_id_len], tenant_id[0..entry.tenant_id_len]);
-            self.keys[self.count] = entry;
-            self.count += 1;
-            self.enabled = true;
-        }
+        var entry = KeyEntry{
+            .hash = hash,
+            .name = undefined,
+            .name_len = @intCast(@min(name.len, 64)),
+            .tenant_id = undefined,
+            .tenant_id_len = @intCast(@min(tenant_id.len, 64)),
+            .perm = perm,
+        };
+        @memcpy(entry.name[0..entry.name_len], name[0..entry.name_len]);
+        @memcpy(entry.tenant_id[0..entry.tenant_id_len], tenant_id[0..entry.tenant_id_len]);
+        self.keys[self.count] = entry;
+        self.count += 1;
+        self.enabled.store(true, .release);
         return hash;
     }
 
@@ -82,7 +84,9 @@ pub const AuthStore = struct {
     }
 
     pub fn resolve(self: *AuthStore, raw_key: []const u8) ?AuthContext {
-        if (!self.enabled) {
+        // Fast-path: if no keys have been added, skip hashing and locking.
+        // Uses acquire to see the latest store from addKeyForTenant.
+        if (!self.enabled.load(.acquire)) {
             return AuthContext{
                 .perm = .admin,
                 .tenant_id = [_]u8{0} ** 64,
@@ -108,7 +112,7 @@ pub const AuthStore = struct {
 
     /// Check if auth is enabled.
     pub fn isEnabled(self: *AuthStore) bool {
-        return self.enabled;
+        return self.enabled.load(.acquire);
     }
 
     /// Extract API key from HTTP headers.

src/branch.zig

@@ -22,13 +22,12 @@ pub const Branch = struct {
     agent_id: [64]u8, // which agent owns this branch
     agent_id_len: u8,
 
-    // Branch-local storage: key_hash -> value (only modified keys)
+    // Branch-local storage: actual key string -> value (only modified keys)
     // This is the CoW layer — unmodified keys fall through to main
-    writes: std.AutoHashMap(u64, BranchWrite),
+    writes: std.StringHashMap(BranchWrite),
     allocator: Allocator,
 
     pub const BranchWrite = struct {
-        key: []const u8, // owned copy
         value: []const u8, // owned copy
         deleted: bool, // true = tombstone (deleted on branch)
         epoch: u64, // when this write happened
@@ -45,28 +44,30 @@ pub const Branch = struct {
     pub fn deinit(self: *Branch) void {
         var it = self.writes.iterator();
         while (it.next()) |entry| {
-            if (entry.value_ptr.key.len > 0) self.allocator.free(entry.value_ptr.key);
             if (entry.value_ptr.value.len > 0) self.allocator.free(entry.value_ptr.value);
+            self.allocator.free(@constCast(entry.key_ptr.*));
         }
         self.writes.deinit();
     }
 
     /// Write a key-value pair on this branch (CoW — only stores the delta)
     pub fn write(self: *Branch, key: []const u8, value: []const u8, epoch: u64) !void {
-        const key_hash = fnv1a(key);
         // Allocate new copies BEFORE freeing old ones — if alloc fails,
         // the existing entry stays valid.
         const owned_key = try self.allocator.dupe(u8, key);
         errdefer self.allocator.free(owned_key);
         const owned_val = try self.allocator.dupe(u8, value);
         errdefer self.allocator.free(owned_val);
         // Free old write if exists (safe — new copies already allocated)
-        if (self.writes.getPtr(key_hash)) |old| {
-            if (old.key.len > 0) self.allocator.free(old.key);
+        if (self.writes.getPtr(key)) |old| {
             if (old.value.len > 0) self.allocator.free(old.value);
+            // Free the old key that was used as the map key.
+            const old_map_key = self.writes.getKey(key).?;
+            self.allocator.free(@constCast(old_map_key));
+            // Remove old entry so we can insert with new owned key.
+            _ = self.writes.remove(key);
         }
-        try self.writes.put(key_hash, .{
-            .key = owned_key,
+        try self.writes.put(owned_key, .{
             .value = owned_val,
             .deleted = false,
             .epoch = epoch,
@@ -75,14 +76,14 @@ pub const Branch = struct {
 
     /// Mark a key as deleted on this branch
     pub fn delete(self: *Branch, key: []const u8, epoch: u64) !void {
-        const key_hash = fnv1a(key);
-        if (self.writes.getPtr(key_hash)) |old| {
-            if (old.key.len > 0) self.allocator.free(old.key);
+        if (self.writes.getPtr(key)) |old| {
             if (old.value.len > 0) self.allocator.free(old.value);
+            const old_map_key = self.writes.getKey(key).?;
+            self.allocator.free(@constCast(old_map_key));
+            _ = self.writes.remove(key);
         }
         const owned_key = try self.allocator.dupe(u8, key);
-        try self.writes.put(key_hash, .{
-            .key = owned_key,
+        try self.writes.put(owned_key, .{
             .value = &.{},
             .deleted = true,
             .epoch = epoch,
@@ -91,8 +92,7 @@ pub const Branch = struct {
 
     /// Read a key on this branch. Returns branch-local value or null (fall through to main).
     pub fn read(self: *const Branch, key: []const u8) ?BranchRead {
-        const key_hash = fnv1a(key);
-        if (self.writes.get(key_hash)) |w| {
+        if (self.writes.get(key)) |w| {
             if (w.deleted) return .{ .deleted = true, .value = null };
             return .{ .deleted = false, .value = w.value };
         }
@@ -111,7 +111,7 @@ pub const Branch = struct {
         while (it.next()) |entry| {
             const w = entry.value_ptr.*;
             try entries.append(alloc, .{
-                .key = w.key,
+                .key = entry.key_ptr.*,
                 .value = w.value,
                 .deleted = w.deleted,
                 .epoch = w.epoch,
@@ -168,7 +168,7 @@ pub fn compareBranches(branches: []*const Branch, alloc: Allocator) !CompareResu
     for (branches) |br| {
         var it = br.writes.iterator();
         while (it.next()) |entry| {
-            try all_keys.put(entry.value_ptr.key, {});
+            try all_keys.put(entry.key_ptr.*, {});
         }
     }
 
@@ -186,7 +186,7 @@ pub fn compareBranches(branches: []*const Branch, alloc: Allocator) !CompareResu
                     .agent_id = br.getAgentId(),
                     .value = r.value,
                     .deleted = r.deleted,
-                    .epoch = if (br.writes.get(fnv1a(key))) |w| w.epoch else 0,
+                    .epoch = if (br.writes.get(key)) |w| w.epoch else 0,
                 });
             }
         }
@@ -435,7 +435,7 @@ pub const BranchManager = struct {
             .status = .active,
             .agent_id = undefined,
             .agent_id_len = aid_len,
-            .writes = std.AutoHashMap(u64, Branch.BranchWrite).init(self.allocator),
+            .writes = std.StringHashMap(Branch.BranchWrite).init(self.allocator),
             .allocator = self.allocator,
         };
         @memcpy(branch.name[0..name_len], name_arg[0..name_len]);

src/bwtree.zig

@@ -49,17 +49,21 @@ pub const BwTree = struct {
     next_page_id: std.atomic.Value(usize),
     allocator: std.mem.Allocator,
     // Deferred reclamation: old chains retired after consolidation are parked here
-    // and freed on the next consolidation or deinit (simple two-phase approach).
+    // and freed once they are at least 2 epochs old (epoch-based safe reclamation).
     retired: std.ArrayList(*Page),
+    retired_epochs: std.ArrayList(u64),
     retired_mu: std.Thread.Mutex,
+    epoch: u64,
 
     pub fn init(allocator: std.mem.Allocator) !BwTree {
         var tree: BwTree = undefined;
         tree.allocator = allocator;
         tree.root_pid = 0;
         tree.next_page_id = std.atomic.Value(usize).init(1);
         tree.retired = std.ArrayList(*Page).init(allocator);
+        tree.retired_epochs = std.ArrayList(u64).init(allocator);
         tree.retired_mu = .{};
+        tree.epoch = 0;
 
         // Zero out all mapping slots
         var i: usize = 0;
@@ -78,9 +82,10 @@ pub const BwTree = struct {
     }
 
     pub fn deinit(self: *BwTree) void {
-        // Free all retired chains first
-        self.drainRetired();
+        // Free all retired chains unconditionally on shutdown
+        self.drainAllRetired();
         self.retired.deinit();
+        self.retired_epochs.deinit();
         var i: usize = 0;
         while (i < MAX_PAGES) : (i += 1) {
             const ptr_val = self.mapping[i].load(.acquire);
@@ -107,12 +112,32 @@ pub const BwTree = struct {
 
     /// Drain the retired list — frees chains that were parked on previous consolidations.
     fn drainRetired(self: *BwTree) void {
+        self.retired_mu.lock();
+        defer self.retired_mu.unlock();
+        const current_epoch = self.epoch;
+        // Free entries that are at least 2 epochs old -- any concurrent reader
+        // that started before the retirement has completed by now.
+        var i: usize = 0;
+        while (i < self.retired.items.len) {
+            if (current_epoch -| self.retired_epochs.items[i] >= 2) {
+                self.freeChain(self.retired.items[i]);
+                _ = self.retired.swapRemove(i);
+                _ = self.retired_epochs.swapRemove(i);
+            } else {
+                i += 1;
+            }
+        }
+    }
+
+    /// Drain ALL retired entries unconditionally (used by deinit).
+    fn drainAllRetired(self: *BwTree) void {
         self.retired_mu.lock();
         defer self.retired_mu.unlock();
         for (self.retired.items) |page| {
             self.freeChain(page);
         }
         self.retired.clearRetainingCapacity();
+        self.retired_epochs.clearRetainingCapacity();
     }
 
     // ─── allocPage ───────────────────────────────────────────────────────
@@ -245,8 +270,10 @@ pub const BwTree = struct {
 
     /// When delta chain exceeds MAX_DELTA_CHAIN, merge into a new base page.
     pub fn consolidate(self: *BwTree, page_id: usize) void {
-        // Drain previously retired chains — they've survived at least one full
-        // consolidation cycle, so readers from the previous epoch are done.
+        // Advance global epoch so drainRetired can age out old entries.
+        self.epoch +%= 1;
+
+        // Drain previously retired chains that are old enough.
         self.drainRetired();
 
         const old = self.mapping[page_id].load(.acquire);
@@ -290,11 +317,16 @@ pub const BwTree = struct {
         ) == null) {
             // Success — park old chain head for deferred reclamation.
             // Concurrent readers may still be traversing it; it will be freed
-            // on the next consolidation cycle (two-phase epoch approach).
+            // once the entry is at least 2 epochs old.
             self.retired_mu.lock();
             defer self.retired_mu.unlock();
             self.retired.append(page) catch {
-                // If we can't track it, leak it — better than use-after-free.
+                // If we can't track it, leak it -- better than use-after-free.
+                return;
+            };
+            self.retired_epochs.append(self.epoch) catch {
+                // Keep arrays in sync: undo the page append on epoch failure.
+                _ = self.retired.pop();
             };
         } else {
             // Another thread consolidated first; discard our work