Skip to content
Open
Show file tree
Hide file tree
Changes from 33 commits
Commits
Show all changes
37 commits
Select commit Hold shift + click to select a range
9b3a1e2
fix: prevent OOB crash in jsonValue on unterminated strings and lower…
justrach Apr 12, 2026
82dc5cd
fix: read full body for small bulk requests to prevent silent data loss
justrach Apr 12, 2026
ac9f366
fix: extractContentLength fully case-insensitive for Content-Length h…
justrach Apr 12, 2026
95d54b3
fix: collection drop now deletes the backing page file on disk
justrach Apr 12, 2026
ab8d669
fix: prevent OOB crash in jsonValue on unterminated strings and lower…
justrach Apr 12, 2026
54addd2
fix: harden against segfaults and OOM across core subsystems
justrach Apr 12, 2026
2f16de2
fix: bound index memory growth and truncate WAL after checkpoint
justrach Apr 12, 2026
5190777
fix: harden against segfaults and OOM across core subsystems
justrach Apr 12, 2026
806fd36
feat: WebSocket transport for persistent streaming connections
justrach Apr 12, 2026
cd77279
fix: prevent OOB crash in jsonValue on unterminated strings and lower…
justrach Apr 12, 2026
e2bdf26
fix: add mutex to WordIndex to prevent crash from concurrent indexer …
justrach Apr 12, 2026
d9b3f7f
fix: comprehensive audit — fix 14 critical and 8 high-severity bugs
justrach Apr 13, 2026
3fdffe2
fix: complete audit — fix all remaining 35 medium and low severity bugs
justrach Apr 13, 2026
9d7675a
fix: resolve final 5 issues + 7 regressions from audit fixes
justrach Apr 13, 2026
491e7b0
fix: protect BTree from concurrent access — likely crash cause
justrach Apr 13, 2026
5b09556
fix: readEntry always returned null — broke all key-based GET lookups
justrach Apr 13, 2026
ea3e173
fix: update() must sync BTree index alongside hash_idx
justrach Apr 13, 2026
f3b26de
fix: sync all indexes on update/delete and rebuild key_epochs on reopen
justrach Apr 13, 2026
547e3de
fix: add input validation and fix jsonEscape safety
justrach Apr 13, 2026
9ccd975
fix: make mmap capacity atomic and prevent truncated JSON responses
justrach Apr 13, 2026
98a84f2
fix: WAL re-queues buffer on write failure instead of dropping entries
justrach Apr 13, 2026
9d5efc7
fix: searchHybrid returns excess docs + delete leaks trigram/word ent…
justrach Apr 13, 2026
42f5366
fix: increase connection thread stack to 2 MiB
justrach Apr 13, 2026
09a0f24
perf: reduce per-collection memory footprint by ~56%
justrach Apr 13, 2026
811538d
perf: reduce mmap VA reservation from 4 GiB to 1 GiB
justrach Apr 13, 2026
bc0ba24
fix: CDC webhook payload corruption when value exceeds 8 KiB buffer
justrach Apr 13, 2026
6277cfa
perf: sharded page locks, LUT search, direct page writes + fix concur…
justrach Apr 13, 2026
ddb6727
fix: MVCC GC on all mutation paths, CDC batch drain, JSON escape corr…
justrach Apr 14, 2026
c894d48
fix: CAS backoff, WAL checkpoint lock scope, CDC/trigram memory leaks
justrach Apr 14, 2026
22c366d
fix: WAL txn_id race, scan thread safety, HashMap pre-alloc, queue ba…
justrach Apr 14, 2026
fcd8047
perf: batch insert path — single lock for all HashMap ops in bulk
justrach Apr 14, 2026
1b35cf7
perf: single-encode, WAL batch write, leaf cache in insertBatch
justrach Apr 15, 2026
cdb6663
perf: SIMD newline scanner for NDJSON bulk insert parsing
justrach Apr 15, 2026
eeaf44e
feat: WebSocket streaming bulk insert — fire-and-forget NDJSON protocol
justrach Apr 15, 2026
81dcab9
docs: add bulk ingestion and WS streaming sections to README
justrach Apr 15, 2026
4b75ce0
fix: reduce parseAndInsertNdjson stack arrays from 4096 to 1024
justrach Apr 15, 2026
f628ede
fix: HashMap data race in flushBatch + chunked lock for write concurr…
justrach Apr 15, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions src/art.zig
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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;
Expand Down
46 changes: 25 additions & 21 deletions src/auth.zig
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

Expand All @@ -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,
Expand All @@ -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.
Expand Down
49 changes: 26 additions & 23 deletions src/branch.zig
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -45,25 +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);
// Free old write if exists
if (self.writes.getPtr(key_hash)) |old| {
if (old.key.len > 0) self.allocator.free(old.key);
if (old.value.len > 0) self.allocator.free(old.value);
}
// 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);
try self.writes.put(key_hash, .{
.key = owned_key,
errdefer self.allocator.free(owned_val);
// Free old write if exists (safe — new copies already allocated)
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);
Comment on lines +65 to +68

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Remove map entry before freeing its key storage

In Branch.write, the old key buffer is freed before calling self.writes.remove(key). std.StringHashMap.remove still needs to probe and compare stored keys, so this turns the map lookup into a use-after-free read when rewriting an existing key; that can corrupt branch state or crash under allocator/debug builds. The same free-before-remove pattern appears in delete as well.

Useful? React with 👍 / 👎.

Comment on lines +65 to +68

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Remove branch entry before freeing its map key

When updating an existing branch key, this code frees old_map_key and then calls self.writes.remove(key). StringHashMap.remove may still read stored keys while probing/comparing entries, so removing after freeing can read freed memory and trigger allocator corruption or crashes during repeated writes/deletes to the same branch key. Remove/fetch the map entry first, then free the returned owned key/value buffers.

Useful? React with 👍 / 👎.

}
try self.writes.put(owned_key, .{
.value = owned_val,
.deleted = false,
.epoch = epoch,
Expand All @@ -72,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,
Expand All @@ -88,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 };
}
Expand All @@ -108,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,
Expand Down Expand Up @@ -165,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.*, {});
}
}

Expand All @@ -183,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,
});
}
}
Expand Down Expand Up @@ -432,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]);
Expand Down
48 changes: 41 additions & 7 deletions src/bwtree.zig
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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);
Expand All @@ -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 ───────────────────────────────────────────────────────
Expand Down Expand Up @@ -206,6 +231,7 @@ pub const BwTree = struct {
} else {
// CAS failed — another thread won; free our delta and retry
self.allocator.destroy(new_delta);
std.atomic.spinLoopHint();
}
}
}
Expand Down Expand Up @@ -237,6 +263,7 @@ pub const BwTree = struct {
return;
} else {
self.allocator.destroy(new_delta);
std.atomic.spinLoopHint();
}
}
}
Expand All @@ -245,8 +272,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;

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Synchronize BwTree epoch updates across concurrent consolidations

The new reclamation epoch is a plain u64, but consolidate() mutates it while other consolidation paths can read it for aging decisions, and those accesses are not synchronized through atomics or a shared lock. Because consolidation is reached from concurrent delta application, this introduces a data race on epoch, making the >= 2 retirement-age check unreliable under contention and risking incorrect reclamation timing.

Useful? React with 👍 / 👎.


// Drain previously retired chains that are old enough.
self.drainRetired();

const old = self.mapping[page_id].load(.acquire);
Expand Down Expand Up @@ -290,11 +319,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
Expand Down
Loading
Loading