fix: make mmap capacity atomic and prevent truncated JSON responses

mmap.zig: capacity was a plain usize read by at()/slice() without
locks while grow() writes it from another thread — a formal data race.
Now uses std.atomic.Value(usize) with acquire/release ordering.
The pre-reserved VA range design means the base pointer is stable,
but capacity must be visible to readers after grow extends the mapping.

server.zig: scan and search handlers used a 256-byte headroom check
before writing each doc, but docs can exceed 256 bytes (code files),
causing partial JSON objects in the response buffer. Now saves the
stream position before each doc and rewinds on overflow, ensuring
only complete JSON objects appear in truncated responses.

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

Author: justrach

src/page.zig

@@ -78,7 +78,7 @@ pub const PageFile = struct {
         // Extend the file.
         const pno = self.next_alloc.fetchAdd(1, .seq_cst);
         const needed = (@as(usize, pno) + 1) * PAGE_SIZE;
-        if (needed > self.mm.capacity) try self.mm.grow(needed);
+        if (needed > self.mm.capacity.load(.acquire)) try self.mm.grow(needed);
 
         const ph = self.pageHeader(pno);
         ph.* = std.mem.zeroes(PageHeader);

src/server.zig

@@ -631,8 +631,7 @@ fn handleScan(srv: *Server, tenant_id: []const u8, col_name: []const u8, query_s
         .{ esc_tid, esc_col, result.docs.len }) catch {};
     var truncated = false;
     for (result.docs, 0..) |d, i| {
-        // Stop writing if buffer is nearly full to avoid truncated JSON.
-        if (fbs.pos + 256 >= MAX_BODY) { truncated = true; break; }
+        const saved_pos = fbs.pos;
         if (i > 0) w.writeByte(',') catch {};
         var esc_dk_buf: [1024]u8 = undefined;
         const esc_dk = jsonEscape(d.key, &esc_dk_buf);
@@ -649,6 +648,13 @@ fn handleScan(srv: *Server, tenant_id: []const u8, col_name: []const u8, query_s
             }
             w.writeAll("\"}") catch {};
         }
+        // If this doc pushed us near the buffer limit, rewind to discard the
+        // partial write and stop — this prevents broken JSON in the response.
+        if (fbs.pos + 64 >= MAX_BODY) {
+            fbs.pos = saved_pos;
+            truncated = true;
+            break;
+        }
     }
     w.writeAll("]}") catch {};
     if (truncated) {
@@ -692,7 +698,7 @@ fn handleSearch(srv: *Server, tenant_id: []const u8, col_name: []const u8, query
         .{ result.docs.len, result.candidate_paths.len, col.docCount(), result.total_files }) catch {};
     var truncated = false;
     for (result.docs, 0..) |d, i| {
-        if (fbs.pos + 256 >= MAX_BODY) { truncated = true; break; }
+        const saved_pos = fbs.pos;
         if (i > 0) w.writeByte(',') catch {};
         // Output value as valid JSON — objects/arrays as-is, strings quoted
         var esc_dk_buf: [1024]u8 = undefined;
@@ -711,6 +717,11 @@ fn handleSearch(srv: *Server, tenant_id: []const u8, col_name: []const u8, query
             }
             w.writeAll("\"}") catch {};
         }
+        if (fbs.pos + 64 >= MAX_BODY) {
+            fbs.pos = saved_pos;
+            truncated = true;
+            break;
+        }
     }
     w.writeAll("]}") catch {};
     if (truncated) {

src/storage/mmap.zig

@@ -34,8 +34,9 @@ pub const MmapFile = struct {
     /// concurrent readers (no lock needed thanks to the stable VA range).
     len:      std.atomic.Value(usize),
     /// How many bytes of the VA range are currently file-backed and mapped
-    /// PROT_READ|PROT_WRITE.  Protected by `grow_mu` (only grow mutates it).
-    capacity: usize,
+    /// PROT_READ|PROT_WRITE.  Atomic so concurrent at()/slice() callers
+    /// can read without holding grow_mu.
+    capacity: std.atomic.Value(usize),
     /// Serialises concurrent grow() calls.  NOT needed for readers.
     grow_mu:  std.Thread.Mutex,
 
@@ -80,7 +81,7 @@ pub const MmapFile = struct {
             .fd       = fd,
             .ptr      = base_ptr,
             .len      = std.atomic.Value(usize).init(existing),
-            .capacity = capacity,
+            .capacity = std.atomic.Value(usize).init(capacity),
             .grow_mu  = .{},
         };
     }
@@ -93,12 +94,14 @@ pub const MmapFile = struct {
 
     // ── Sync / Checkpoint ─────────────────────────────────────────────────────
 
-    pub fn syncAsync(self: MmapFile) void {
-        posix.msync(@alignCast(self.ptr[0..self.capacity]), posix.MSF.ASYNC) catch {};
+    pub fn syncAsync(self: *MmapFile) void {
+        const cap = self.capacity.load(.acquire);
+        posix.msync(@alignCast(self.ptr[0..cap]), posix.MSF.ASYNC) catch {};
     }
 
-    pub fn syncSync(self: MmapFile) !void {
-        try posix.msync(@alignCast(self.ptr[0..self.capacity]), posix.MSF.SYNC);
+    pub fn syncSync(self: *MmapFile) !void {
+        const cap = self.capacity.load(.acquire);
+        try posix.msync(@alignCast(self.ptr[0..cap]), posix.MSF.SYNC);
     }
 
     // ── Grow ──────────────────────────────────────────────────────────────────
@@ -107,7 +110,7 @@ pub const MmapFile = struct {
     /// extends the file and maps the new pages into the pre-reserved VA
     /// range with MAP_FIXED.  The base pointer never changes.
     pub fn grow(self: *MmapFile, needed_len: usize) !void {
-        if (needed_len <= self.capacity) {
+        if (needed_len <= self.capacity.load(.acquire)) {
             // Capacity is sufficient — just advance the logical length.
             self.len.store(needed_len, .release);
             return;
@@ -118,12 +121,13 @@ pub const MmapFile = struct {
         defer self.grow_mu.unlock();
 
         // Re-check after acquiring lock (another thread may have grown).
-        if (needed_len <= self.capacity) {
+        const cur_cap = self.capacity.load(.acquire);
+        if (needed_len <= cur_cap) {
             self.len.store(needed_len, .release);
             return;
         }
 
-        const old_cap = self.capacity;
+        const old_cap = cur_cap;
         const new_cap = alignUp(needed_len + GROW_CHUNK, OS_PAGE_ALIGN);
 
         std.debug.assert(new_cap <= MAX_VA_SIZE);
@@ -141,7 +145,7 @@ pub const MmapFile = struct {
             self.fd, @intCast(old_cap),
         );
 
-        self.capacity = new_cap;
+        self.capacity.store(new_cap, .release);
         self.len.store(needed_len, .release);
     }
 
@@ -150,14 +154,14 @@ pub const MmapFile = struct {
     /// Return a pointer to the record at byte offset `off`.
     /// Lock-free: the base pointer is stable for the lifetime of the mapping.
     pub fn at(self: *MmapFile, comptime T: type, off: usize) *T {
-        std.debug.assert(off + @sizeOf(T) <= self.capacity);
+        std.debug.assert(off + @sizeOf(T) <= self.capacity.load(.acquire));
         return @alignCast(@ptrCast(&self.ptr[off]));
     }
 
     /// Return a slice of T starting at byte offset `off`, `count` elements.
     /// Lock-free: the base pointer is stable for the lifetime of the mapping.
     pub fn slice(self: *MmapFile, comptime T: type, off: usize, count: usize) []T {
-        std.debug.assert(off + count * @sizeOf(T) <= self.capacity);
+        std.debug.assert(off + count * @sizeOf(T) <= self.capacity.load(.acquire));
         return @as([*]T, @alignCast(@ptrCast(&self.ptr[off])))[0..count];
     }