dev/MultiforaDB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: adopt TigerStyle across src/; add docs/TIGER_STYLE.md

Browse Source 
Wrap signatures and long expressions to the 100-column limit and make every
file zig fmt clean. Semantics-preserving throughout: ignoring whitespace and
the trailing commas that wrapping introduces, every file here is byte-identical
to its predecessor, and the one apparent exception is a warning string split
with `++`, which concatenates at comptime to the same bytes.

src/index.zig and src/commands.zig are reformatted in the commits that follow,
because their reformat is interleaved with in-flight changes to them and
separating the two would need the reformat re-derived rather than moved.
This commit is contained in:
Alexey Shakhmatov
2026-08-03 17:08:21 +03:00
parent d4c9b04f21
commit 86ae8fa8af
14 changed files with 1130 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
src/bson.zig
View File

@@ -122,7 +122,11 @@ pub const Document = struct {
}
/// Serialize the full document (length-prefixed) into `out`.
pub fn to_bytes(self: *const Document, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) !void {
pub fn to_bytes(
self: *const Document,
gpa: std.mem.Allocator,
out: *std.ArrayListUnmanaged(u8),
) !void {
try write_doc(self.pairs, gpa, out);
}
};
@@ -157,7 +161,12 @@ const Parser = struct {
const ParseError = error{ InvalidBson, OutOfMemory };
fn parse_doc_into(allocator: std.mem.Allocator, bytes: []const u8, idx: *usize, borrow: bool) ParseError![]const Pair {
fn parse_doc_into(
allocator: std.mem.Allocator,
bytes: []const u8,
idx: *usize,
borrow: bool,
) ParseError![]const Pair {
const p = Parser{ .allocator = allocator, .bytes = bytes, .borrow = borrow };
return parse_doc_inner(p, idx);
}
@@ -356,7 +365,10 @@ fn parse_array(p: Parser, idx: *usize) ParseError![]const Value {
/// A borrowed parse of `bytes` into an arena: keys, strings, binary and
/// regex payloads point into `bytes`; only the pair/value skeleton is
/// allocated. The result is valid while both `bytes` and `arena` live.
pub fn spine(allocator: std.mem.Allocator, bytes: []const u8) error{ InvalidBson, OutOfMemory }![]const Pair {
pub fn spine(
allocator: std.mem.Allocator,
bytes: []const u8,
) error{ InvalidBson, OutOfMemory }![]const Pair {
var idx: usize = 0;
return parse_doc_into(allocator, bytes, &idx, true);
}
@@ -414,7 +426,12 @@ pub fn skip_value(bytes: []const u8, idx: *usize, tag: u8) error{InvalidBson}!vo
/// Read one value of `tag` into a Value whose leaves borrow `bytes`; nested
/// documents and arrays materialize their spines into `arena`. Advances
/// `idx` past the value.
pub fn read_value(allocator: std.mem.Allocator, bytes: []const u8, idx: *usize, tag: u8) error{ InvalidBson, OutOfMemory }!Value {
pub fn read_value(
allocator: std.mem.Allocator,
bytes: []const u8,
idx: *usize,
tag: u8,
) error{ InvalidBson, OutOfMemory }!Value {
switch (tag) {
0x01 => {
try ensure_available(bytes, idx.*, 8);
@@ -542,7 +559,11 @@ pub fn read_value(allocator: std.mem.Allocator, bytes: []const u8, idx: *usize,
/// The value stored under `key` in a document's bytes, or null when absent.
/// Nested documents and arrays materialize their spines into `arena`.
pub fn get_at(arena: std.mem.Allocator, bytes: []const u8, key: []const u8) error{ InvalidBson, OutOfMemory }!?Value {
pub fn get_at(
arena: std.mem.Allocator,
bytes: []const u8,
key: []const u8,
) error{ InvalidBson, OutOfMemory }!?Value {
var idx: usize = 4;
while (idx + 1 < bytes.len and bytes[idx] != 0) {
const tag = bytes[idx];
@@ -565,7 +586,11 @@ pub const SerializeError = error{
OutOfMemory,
};
pub fn write_value(v: Value, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) SerializeError!void {
pub fn write_value(
v: Value,
gpa: std.mem.Allocator,
out: *std.ArrayListUnmanaged(u8),
) SerializeError!void {
switch (v) {
.double => |d| {
var buf: [8]u8 = undefined;
@@ -618,13 +643,21 @@ pub fn write_value(v: Value, gpa: std.mem.Allocator, out: *std.ArrayListUnmanage
}
}
pub fn write_cstring(s: []const u8, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) SerializeError!void {
pub fn write_cstring(
s: []const u8,
gpa: std.mem.Allocator,
out: *std.ArrayListUnmanaged(u8),
) SerializeError!void {
if (std.mem.indexOfScalar(u8, s, 0) != null) return error.BsonNulInKey;
try out.appendSlice(gpa, s);
try out.append(gpa, 0);
}
pub fn write_string(s: []const u8, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) SerializeError!void {
pub fn write_string(
s: []const u8,
gpa: std.mem.Allocator,
out: *std.ArrayListUnmanaged(u8),
) SerializeError!void {
if (s.len + 1 > std.math.maxInt(u32)) return error.BsonTooLarge;
var buf: [4]u8 = undefined;
std.mem.writeInt(u32, &buf, @intCast(s.len + 1), .little);
@@ -633,7 +666,11 @@ pub fn write_string(s: []const u8, gpa: std.mem.Allocator, out: *std.ArrayListUn
try out.append(gpa, 0);
}
pub fn write_element(pair: Pair, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) SerializeError!void {
pub fn write_element(
pair: Pair,
gpa: std.mem.Allocator,
out: *std.ArrayListUnmanaged(u8),
) SerializeError!void {
try out.append(gpa, pair.value.type_tag());
try write_cstring(pair.key, gpa, out);
try write_value(pair.value, gpa, out);
@@ -648,7 +685,11 @@ fn begin_frame(gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) Seriali
}
/// Terminate the frame opened at `len_pos` and patch in its total length.
fn end_frame(gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8), len_pos: usize) SerializeError!void {
fn end_frame(
gpa: std.mem.Allocator,
out: *std.ArrayListUnmanaged(u8),
len_pos: usize,
) SerializeError!void {
try out.append(gpa, 0);
const total = out.items.len - len_pos;
if (total > std.math.maxInt(u32)) return error.BsonTooLarge;
@@ -656,13 +697,21 @@ fn end_frame(gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8), len_pos:
}
/// Write a length-prefixed document. Length is patched in after the body.
pub fn write_doc(pairs: []const Pair, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) SerializeError!void {
pub fn write_doc(
pairs: []const Pair,
gpa: std.mem.Allocator,
out: *std.ArrayListUnmanaged(u8),
) SerializeError!void {
const len_pos = try begin_frame(gpa, out);
for (pairs) |p| try write_element(p, gpa, out);
try end_frame(gpa, out, len_pos);
}
fn write_array(items: []const Value, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) SerializeError!void {
fn write_array(
items: []const Value,
gpa: std.mem.Allocator,
out: *std.ArrayListUnmanaged(u8),
) SerializeError!void {
const len_pos = try begin_frame(gpa, out);
var buf: [16]u8 = undefined;
for (items, 0..) |item, i| {
@@ -686,7 +735,11 @@ pub fn serialize_value(gpa: std.mem.Allocator, v: Value) ![]u8 {
/// Append the serialized-key bytes of `v` (type tag + payload) to `out`. The
/// appending form of serialize_value, for callers reusing one scratch buffer
/// across many keys.
pub fn write_serialized_value(v: Value, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) !void {
pub fn write_serialized_value(
v: Value,
gpa: std.mem.Allocator,
out: *std.ArrayListUnmanaged(u8),
) !void {
try out.append(gpa, v.type_tag());
try write_value(v, gpa, out);
}
@@ -709,7 +762,10 @@ pub fn copy_value(arena: std.mem.Allocator, v: Value) std.mem.Allocator.Error!Va
};
}
pub fn copy_pairs(arena: std.mem.Allocator, pairs: []const Pair) std.mem.Allocator.Error![]const Pair {
pub fn copy_pairs(
arena: std.mem.Allocator,
pairs: []const Pair,
) std.mem.Allocator.Error![]const Pair {
const out = try arena.alloc(Pair, pairs.len);
for (pairs, 0..) |p, i| {
out[i] = .{ .key = try arena.dupe(u8, p.key), .value = try copy_value(arena, p.value) };
@@ -717,7 +773,10 @@ pub fn copy_pairs(arena: std.mem.Allocator, pairs: []const Pair) std.mem.Allocat
return out;
}
fn copy_values(arena: std.mem.Allocator, items: []const Value) std.mem.Allocator.Error![]const Value {
fn copy_values(
arena: std.mem.Allocator,
items: []const Value,
) std.mem.Allocator.Error![]const Value {
const out = try arena.alloc(Value, items.len);
for (items, 0..) |item, i| out[i] = try copy_value(arena, item);
return out;
@@ -879,7 +938,11 @@ pub fn encoded_leading_datetime(key: []const u8) ?i64 {
/// ambiguous. Escaping `00` as `00 FF` fixes both problems at once: a real
/// NUL encodes above the `00 00` terminator, and any byte >= 01 is above it
/// too, so "shorter is less" falls out to match `std.mem.order`.
fn encode_escaped(bytes: []const u8, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) !void {
fn encode_escaped(
bytes: []const u8,
gpa: std.mem.Allocator,
out: *std.ArrayListUnmanaged(u8),
) !void {
for (bytes) |b| {
try out.append(gpa, b);
if (b == 0x00) try out.append(gpa, 0xFF);
@@ -1128,23 +1191,47 @@ test "encode_key order matches bson.compare on every pair" {
.min_key,
.null,
// Numbers: cross-type equality, sign, zero, extremes, NaN.
.{ .int32 = -2147483648 }, .{ .int32 = -1 }, .{ .int32 = 0 },
.{ .int32 = 1 }, .{ .int32 = 2 }, .{ .int32 = 2147483647 },
.{ .int64 = -9223372036854775807 }, .{ .int64 = -1 }, .{ .int64 = 0 },
.{ .int64 = 1 }, .{ .int64 = 9223372036854775807 },
.{ .double = -std.math.inf(f64) }, .{ .double = -1.5 }, .{ .double = -0.0 },
.{ .double = 0.0 }, .{ .double = 0.5 }, .{ .double = 1.0 },
.{ .double = 1.5 }, .{ .double = std.math.inf(f64) },
.{ .int32 = -2147483648 },
.{ .int32 = -1 },
.{ .int32 = 0 },
.{ .int32 = 1 },
.{ .int32 = 2 },
.{ .int32 = 2147483647 },
.{ .int64 = -9223372036854775807 },
.{ .int64 = -1 },
.{ .int64 = 0 },
.{ .int64 = 1 },
.{ .int64 = 9223372036854775807 },
.{ .double = -std.math.inf(f64) },
.{ .double = -1.5 },
.{ .double = -0.0 },
.{ .double = 0.0 },
.{ .double = 0.5 },
.{ .double = 1.0 },
.{ .double = 1.5 },
.{ .double = std.math.inf(f64) },
.{ .double = std.math.nan(f64) },
// Strings, including embedded NUL and prefix relationships.
.{ .string = "" }, .{ .string = "\x00" }, .{ .string = "\x00b" },
.{ .string = "a" }, .{ .string = "a\x00" }, .{ .string = "a\x00b" },
.{ .string = "ab" }, .{ .string = "b" }, .{ .string = "\xff" },
.{ .string = "" },
.{ .string = "\x00" },
.{ .string = "\x00b" },
.{ .string = "a" },
.{ .string = "a\x00" },
.{ .string = "a\x00b" },
.{ .string = "ab" },
.{ .string = "b" },
.{ .string = "\xff" },
// Same rank as string, so these must interleave with them.
.{ .symbol = "a" }, .{ .code = "ab" },
.{ .doc = &.{} }, .{ .doc = &nested }, .{ .doc = &nested2 },
.{ .doc = &nested_l }, .{ .doc = &two_pairs },
.{ .array = &.{} }, .{ .array = &arr1 }, .{ .array = &arr2 },
.{ .symbol = "a" },
.{ .code = "ab" },
.{ .doc = &.{} },
.{ .doc = &nested },
.{ .doc = &nested2 },
.{ .doc = &nested_l },
.{ .doc = &two_pairs },
.{ .array = &.{} },
.{ .array = &arr1 },
.{ .array = &arr2 },
.{ .array = &arr_str },
// Binary orders by length first, then bytes, then subtype.
.{ .binary = .{ .subtype = 0, .data = "" } },
@@ -1155,11 +1242,15 @@ test "encode_key order matches bson.compare on every pair" {
.{ .object_id = [_]u8{0} ** 12 },
.{ .object_id = [_]u8{0} ** 11 ++ [_]u8{1} },
.{ .object_id = [_]u8{255} ** 12 },
.{ .bool = false }, .{ .bool = true },
.{ .datetime = std.math.minInt(i64) }, .{ .datetime = -1 },
.{ .datetime = 0 }, .{ .datetime = 1 },
.{ .bool = false },
.{ .bool = true },
.{ .datetime = std.math.minInt(i64) },
.{ .datetime = -1 },
.{ .datetime = 0 },
.{ .datetime = 1 },
.{ .datetime = std.math.maxInt(i64) },
.{ .timestamp = 0 }, .{ .timestamp = 1 },
.{ .timestamp = 0 },
.{ .timestamp = 1 },
.{ .timestamp = std.math.maxInt(u64) },
.{ .regex = .{ .pattern = "a", .options = "" } },
.{ .regex = .{ .pattern = "a", .options = "i" } },
@@ -1199,9 +1290,9 @@ test "encode_key concatenates into unambiguous compound keys" {
// against the component-wise order they are supposed to reproduce.
const gpa = testing.allocator;
const parts = [_]Value{
.{ .string = "" }, .{ .string = "a" }, .{ .string = "a\x00" },
.{ .string = "ab" }, .{ .int32 = 1 }, .{ .int32 = 2 },
.null, .{ .array = &.{} }, .{ .bool = true },
.{ .string = "" }, .{ .string = "a" }, .{ .string = "a\x00" },
.{ .string = "ab" }, .{ .int32 = 1 }, .{ .int32 = 2 },
.null, .{ .array = &.{} }, .{ .bool = true },
};
for (parts) |a1| {

151
src/db.zig
View File

@@ -320,7 +320,13 @@ pub const Engine = struct {
/// shared); the collection lock is acquired before the exclusive catalog
/// lock is dropped, so a concurrent drop can never free it underneath.
/// Returns null when the collection does not exist (and create is off).
pub fn lock_collection(self: *Engine, db_name: []const u8, coll_name: []const u8, write: bool, create: bool) !?*Collection {
pub fn lock_collection(
self: *Engine,
db_name: []const u8,
coll_name: []const u8,
write: bool,
create: bool,
) !?*Collection {
var coll = self.get_collection(db_name, coll_name);
if (coll == null and create) {
self.catalog_lock.unlockShared(self.io);
@@ -425,7 +431,13 @@ pub const Engine = struct {
/// Log an append (and its seq increment) under the log lock, marking
/// the append as in flight so a commit leader's seal covers it.
fn log_append(self: *Engine, comptime kind: LogKind, db: []const u8, coll: []const u8, doc: []const u8) !void {
fn log_append(
self: *Engine,
comptime kind: LogKind,
db: []const u8,
coll: []const u8,
doc: []const u8,
) !void {
_ = self.pending_appends.fetchAdd(1, .acq_rel);
defer {
// The increment above pairs with this decrement on every return
@@ -468,12 +480,24 @@ pub const Engine = struct {
/// Insert a document. Fails with error.DuplicateKey if the _id exists.
/// Generates an ObjectId _id when absent.
pub fn insert(self: *Engine, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document, oid_gen: *bson.ObjectIdGen) !void {
pub fn insert(
self: *Engine,
db_name: []const u8,
coll_name: []const u8,
doc: *const bson.Document,
oid_gen: *bson.ObjectIdGen,
) !void {
return self.upsert(db_name, coll_name, doc, oid_gen, .insert);
}
/// Insert or replace a document by _id (upsert without existence check).
pub fn replace(self: *Engine, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document, oid_gen: *bson.ObjectIdGen) !void {
pub fn replace(
self: *Engine,
db_name: []const u8,
coll_name: []const u8,
doc: *const bson.Document,
oid_gen: *bson.ObjectIdGen,
) !void {
return self.upsert(db_name, coll_name, doc, oid_gen, .replace);
}
@@ -574,7 +598,12 @@ pub const Engine = struct {
/// Remove a document by its `_id` value. Returns true if it existed.
/// The serialized-key encoding stays private to the engine.
pub fn remove_by_id(self: *Engine, db_name: []const u8, coll_name: []const u8, id: bson.Value) !bool {
pub fn remove_by_id(
self: *Engine,
db_name: []const u8,
coll_name: []const u8,
id: bson.Value,
) !bool {
const id_key = try bson.serialize_value(self.gpa, id);
defer self.gpa.free(id_key);
return self.remove(db_name, coll_name, id_key);
@@ -610,7 +639,12 @@ pub const Engine = struct {
return db.collections.get(coll_name);
}
pub fn get_doc(self: *Engine, db_name: []const u8, coll_name: []const u8, id_key: []const u8) ?[]const u8 {
pub fn get_doc(
self: *Engine,
db_name: []const u8,
coll_name: []const u8,
id_key: []const u8,
) ?[]const u8 {
const coll = self.get_collection(db_name, coll_name) orelse return null;
const off = coll.docs.get(id_key) orelse return null;
return coll.doc_bytes(off);
@@ -636,7 +670,12 @@ pub const Engine = struct {
/// after the index builds over the existing documents and passes
/// uniqueness, so a rejected create persists nothing. Returns the new
/// index (or the existing one when the spec matches — idempotent).
pub fn create_index(self: *Engine, db_name: []const u8, coll_name: []const u8, spec_doc: *const bson.Document) !*index.Index {
pub fn create_index(
self: *Engine,
db_name: []const u8,
coll_name: []const u8,
spec_doc: *const bson.Document,
) !*index.Index {
const coll = try self.get_or_create_collection(db_name, coll_name);
var ix = try index.parse_spec(self.gpa, spec_doc);
var committed = false;
@@ -675,7 +714,12 @@ pub const Engine = struct {
/// Remove a secondary index by name, persisting a drop record first.
/// Returns false when no such index exists.
pub fn drop_index(self: *Engine, db_name: []const u8, coll_name: []const u8, index_name: []const u8) !bool {
pub fn drop_index(
self: *Engine,
db_name: []const u8,
coll_name: []const u8,
index_name: []const u8,
) !bool {
const db = self.dbs.get(db_name) orelse return false;
const coll = db.collections.get(coll_name) orelse return false;
if (coll.find_index(index_name) == null) return false;
@@ -722,7 +766,13 @@ pub const Engine = struct {
/// Sweep one collection under its write lock; the lock is released on
/// every return path. Returns how many documents were removed.
fn ttl_sweep_coll(self: *Engine, coll: *Collection, now_ms: i64, db_name: []const u8, coll_name: []const u8) !usize {
fn ttl_sweep_coll(
self: *Engine,
coll: *Collection,
now_ms: i64,
db_name: []const u8,
coll_name: []const u8,
) !usize {
try coll.lock.lock(self.io);
defer coll.lock.unlock(self.io);
// Ids are duped rather than aliased: `remove` frees the docs-map key
@@ -778,7 +828,11 @@ pub const Engine = struct {
while (it.next()) |entry| try out.append(self.gpa, entry.key_ptr.*);
}
pub fn collection_names(self: *Engine, db_name: []const u8, out: *std.ArrayListUnmanaged([]const u8)) !void {
pub fn collection_names(
self: *Engine,
db_name: []const u8,
out: *std.ArrayListUnmanaged([]const u8),
) !void {
const db = self.dbs.get(db_name) orelse return;
var it = db.collections.iterator();
while (it.next()) |entry| try out.append(self.gpa, entry.key_ptr.*);
@@ -786,7 +840,11 @@ pub const Engine = struct {
// -- internals -----------------------------------------------------------
pub fn get_or_create_collection(self: *Engine, db_name: []const u8, coll_name: []const u8) !*Collection {
pub fn get_or_create_collection(
self: *Engine,
db_name: []const u8,
coll_name: []const u8,
) !*Collection {
const db = self.dbs.getPtr(db_name) orelse {
const db_key = try self.gpa.dupe(u8, db_name);
errdefer self.gpa.free(db_key);
@@ -808,7 +866,11 @@ pub const Engine = struct {
/// generated ObjectId `_id` when absent.
/// The canonical bytes of `doc`, with an ObjectId `_id` generated when
/// absent. The result is owned by the caller.
fn serialize_with_id(self: *Engine, doc: *const bson.Document, oid_gen: *bson.ObjectIdGen) ![]u8 {
fn serialize_with_id(
self: *Engine,
doc: *const bson.Document,
oid_gen: *bson.ObjectIdGen,
) ![]u8 {
if (doc.get("_id") != null) return serialize_doc(self.gpa, doc);
var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
defer pairs.deinit(self.gpa);
@@ -994,7 +1056,13 @@ pub const Engine = struct {
/// Re-emit one collection's index specs and documents into the compacted
/// log, under the collection's write lock (released on every return
/// path, including errors).
fn compact_snapshot_coll(self: *Engine, coll: *Collection, new_log: *storage.Log, db_name: []const u8, coll_name: []const u8) !void {
fn compact_snapshot_coll(
self: *Engine,
coll: *Collection,
new_log: *storage.Log,
db_name: []const u8,
coll_name: []const u8,
) !void {
try coll.lock.lock(self.io);
defer coll.lock.unlock(self.io);
// Re-emit the index definitions first: a compacted log that dropped
@@ -1045,7 +1113,13 @@ pub const Engine = struct {
while (doc_it.next()) |doc_entry| {
ix.append_doc_entries(self.gpa, coll.doc_bytes(doc_entry.value_ptr.*), doc_entry.key_ptr.*) catch |err| switch (err) {
error.ParallelArrays => {
std.debug.print("multiforadb: WARNING: index '{s}' cannot index an existing document; entry skipped\n", .{ix.name});
std.debug.print(
"multiforadb: WARNING: index '{s}' cannot index an existing " ++
"document; entry skipped\n",
.{
ix.name,
},
);
continue;
},
else => return err,
@@ -1053,13 +1127,23 @@ pub const Engine = struct {
}
// Tolerated, not enforced: the database must always open.
if (try ix.finish_bulk(self.gpa, false)) {
std.debug.print("multiforadb: WARNING: unique index '{s}' has duplicate keys in existing data; duplicates not enforced for existing documents\n", .{ix.name});
std.debug.print(
"multiforadb: WARNING: unique index '{s}' has duplicate keys in existing " ++
"data; duplicates not enforced for existing documents\n",
.{
ix.name,
},
);
}
}
/// Register an (empty) index from a persisted spec document. A repeated
/// create record for the same name is an idempotent no-op.
fn register_index_from_spec(self: *Engine, coll: *Collection, spec_doc: *const bson.Document) !void {
fn register_index_from_spec(
self: *Engine,
coll: *Collection,
spec_doc: *const bson.Document,
) !void {
var ix = try index.parse_spec(self.gpa, spec_doc);
var committed = false;
defer if (!committed) ix.deinit(self.gpa);
@@ -1103,7 +1187,9 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
switch (record.type) {
storage.record_type_index_create => {
self.register_index_from_spec(coll, doc) catch |err| {
std.debug.print("multiforadb: index create record failed to apply: {s}\n", .{@errorName(err)});
std.debug.print("multiforadb: index create record failed to apply: {s}\n", .{
@errorName(err),
});
return;
};
return;
@@ -1455,7 +1541,12 @@ test "concurrent readers and writers on a threaded Io" {
var remaining = std.atomic.Value(usize).init(@intCast(total));
const Worker = struct {
fn writer(e: *Engine, id_counter: *std.atomic.Value(i32), pending: *std.atomic.Value(usize), alloc: std.mem.Allocator) error{Canceled}!void {
fn writer(
e: *Engine,
id_counter: *std.atomic.Value(i32),
pending: *std.atomic.Value(usize),
alloc: std.mem.Allocator,
) error{Canceled}!void {
while (true) {
const id = id_counter.fetchAdd(1, .monotonic);
if (id > total) return;
@@ -1605,7 +1696,11 @@ test "concurrent writers compacting: the log survives a reopen" {
}
}
fn writer(e: *Engine, id_counter: *std.atomic.Value(i32), alloc: std.mem.Allocator) error{Canceled}!void {
fn writer(
e: *Engine,
id_counter: *std.atomic.Value(i32),
alloc: std.mem.Allocator,
) error{Canceled}!void {
while (true) {
const id = id_counter.fetchAdd(1, .monotonic);
if (id > total) return;
@@ -1645,7 +1740,14 @@ test "concurrent writers compacting: the log survives a reopen" {
/// A spec document for a single-path index, built by serializing and
/// re-parsing so the pairs are arena-owned.
fn index_spec(gpa: std.mem.Allocator, path: []const u8, name: []const u8, unique: bool, sparse: bool, ttl: ?i64) !bson.Document {
fn index_spec(
gpa: std.mem.Allocator,
path: []const u8,
name: []const u8,
unique: bool,
sparse: bool,
ttl: ?i64,
) !bson.Document {
var out: std.ArrayListUnmanaged(u8) = .empty;
defer out.deinit(gpa);
var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
@@ -1662,7 +1764,14 @@ fn index_spec(gpa: std.mem.Allocator, path: []const u8, name: []const u8, unique
}
/// Number of entries the named index has for a single-value equality key.
fn index_count(gpa: std.mem.Allocator, engine: *Engine, db_name: []const u8, coll_name: []const u8, name: []const u8, key_value: bson.Value) !usize {
fn index_count(
gpa: std.mem.Allocator,
engine: *Engine,
db_name: []const u8,
coll_name: []const u8,
name: []const u8,
key_value: bson.Value,
) !usize {
const coll = engine.get_collection(db_name, coll_name) orelse return 0;
for (coll.indexes.items) |*ix| {
if (std.mem.eql(u8, ix.name, name)) {

7
src/fuzz_split.zig
View File

@@ -113,7 +113,12 @@ fn run(seed: u64, ops: usize, max_len: usize) !void {
if (std.mem.eql(u8, got, d.id)) hit = true;
}
if (!hit) {
std.debug.print("seed {d} op {d}: id {s} (key len {d}) not found by descent\n", .{ seed, op, d.id, d.s.len });
std.debug.print("seed {d} op {d}: id {s} (key len {d}) not found by descent\n", .{
seed,
op,
d.id,
d.s.len,
});
return error.EntryUnreachable;
}
}

5
src/main.zig
View File

@@ -99,7 +99,10 @@ pub fn main(init: std.process.Init) !void {
var engine = try mongo.db.Engine.open(init.gpa, init.io, db_path);
defer engine.deinit();
engine.compact_threshold = compact_threshold;
std.debug.print("multiforadb: opened database '{s}' (compact threshold {d})\n", .{ db_path, compact_threshold });
std.debug.print("multiforadb: opened database '{s}' (compact threshold {d})\n", .{
db_path,
compact_threshold,
});
var server = mongo.server.Server{
.gpa = init.gpa,

184
src/query.zig
View File

@@ -14,7 +14,11 @@ const bson = @import("bson.zig");
/// input — but it must still be a possible error, not a panic.
pub const QueryError = error{ OutOfMemory, InvalidBson };
pub fn matches(gpa: std.mem.Allocator, filter: *const bson.Document, doc: *const bson.Document) QueryError!bool {
pub fn matches(
gpa: std.mem.Allocator,
filter: *const bson.Document,
doc: *const bson.Document,
) QueryError!bool {
for (filter.pairs) |p| {
if (p.key.len > 0 and p.key[0] == '$') {
if (!try match_top_level(gpa, p.key, p.value, doc)) return false;
@@ -25,7 +29,12 @@ pub fn matches(gpa: std.mem.Allocator, filter: *const bson.Document, doc: *const
return true;
}
fn match_top_level(gpa: std.mem.Allocator, op: []const u8, value: bson.Value, doc: *const bson.Document) QueryError!bool {
fn match_top_level(
gpa: std.mem.Allocator,
op: []const u8,
value: bson.Value,
doc: *const bson.Document,
) QueryError!bool {
if (std.mem.eql(u8, op, "$and") or std.mem.eql(u8, op, "$or")) {
const want_and = std.mem.eql(u8, op, "$and");
const filters = switch (value) {
@@ -89,7 +98,12 @@ fn is_operator_doc(value: bson.Value) ?[]const bson.Pair {
/// collection scan.
const inline_candidates = 8;
fn field_matches(gpa: std.mem.Allocator, path: []const u8, expected: bson.Value, doc: *const bson.Document) QueryError!bool {
fn field_matches(
gpa: std.mem.Allocator,
path: []const u8,
expected: bson.Value,
doc: *const bson.Document,
) QueryError!bool {
var stack_fallback = std.heap.stackFallback(inline_candidates * @sizeOf(bson.Value), gpa);
const alloc = stack_fallback.get();
@@ -103,7 +117,12 @@ fn field_matches(gpa: std.mem.Allocator, path: []const u8, expected: bson.Value,
/// The byte counterpart of field_matches: collects values by walking the
/// canonical BSON element stream of a stored document, skipping by length
/// any field the filter does not name.
fn field_matches_bytes(gpa: std.mem.Allocator, path: []const u8, expected: bson.Value, bytes: []const u8) QueryError!bool {
fn field_matches_bytes(
gpa: std.mem.Allocator,
path: []const u8,
expected: bson.Value,
bytes: []const u8,
) QueryError!bool {
// An arena, not a stack fallback: the byte walker materializes nested
// doc/array values (whole-array equality, embedded docs) into the
// allocator it is given, and those must be freed with it.
@@ -120,7 +139,10 @@ fn field_matches_bytes(gpa: std.mem.Allocator, path: []const u8, expected: bson.
/// MongoDB applies queries to array elements as well as the array itself.
/// Index the snapshot length, re-reading items each iteration: appending
/// may reallocate the buffer, which would invalidate a captured slice.
fn expand_arrays(alloc: std.mem.Allocator, candidates: *std.ArrayListUnmanaged(bson.Value)) QueryError!void {
fn expand_arrays(
alloc: std.mem.Allocator,
candidates: *std.ArrayListUnmanaged(bson.Value),
) QueryError!void {
const direct_count = candidates.items.len;
var i: usize = 0;
while (i < direct_count) : (i += 1) {
@@ -133,7 +155,11 @@ fn expand_arrays(alloc: std.mem.Allocator, candidates: *std.ArrayListUnmanaged(b
/// The operator/equality half of field matching, shared by the tree and
/// byte collectors.
fn apply_expected(gpa: std.mem.Allocator, expected: bson.Value, candidates: []const bson.Value) QueryError!bool {
fn apply_expected(
gpa: std.mem.Allocator,
expected: bson.Value,
candidates: []const bson.Value,
) QueryError!bool {
if (is_operator_doc(expected)) |pairs| {
// $options modifies $regex wherever it appears in the document, so
// it has to be known before any operator runs.
@@ -166,7 +192,11 @@ fn apply_expected(gpa: std.mem.Allocator, expected: bson.Value, candidates: []co
/// byte-matcher counterpart of `matches`, used by scans. Same semantics,
/// different collection: fields the filter does not name are skipped by
/// length instead of materialized.
pub fn matches_bytes(gpa: std.mem.Allocator, filter: []const bson.Pair, bytes: []const u8) QueryError!bool {
pub fn matches_bytes(
gpa: std.mem.Allocator,
filter: []const bson.Pair,
bytes: []const u8,
) QueryError!bool {
for (filter) |p| {
if (p.key.len > 0 and p.key[0] == '$') {
if (!try match_top_level_bytes(gpa, p.key, p.value, bytes)) return false;
@@ -177,7 +207,12 @@ pub fn matches_bytes(gpa: std.mem.Allocator, filter: []const bson.Pair, bytes: [
return true;
}
fn match_top_level_bytes(gpa: std.mem.Allocator, op: []const u8, value: bson.Value, bytes: []const u8) QueryError!bool {
fn match_top_level_bytes(
gpa: std.mem.Allocator,
op: []const u8,
value: bson.Value,
bytes: []const u8,
) QueryError!bool {
if (std.mem.eql(u8, op, "$and") or std.mem.eql(u8, op, "$or")) {
const want_and = std.mem.eql(u8, op, "$and");
const filters = switch (value) {
@@ -215,7 +250,13 @@ fn match_top_level_bytes(gpa: std.mem.Allocator, op: []const u8, value: bson.Val
/// Collect values reachable at `path` from a document's canonical bytes —
/// the byte counterpart of `collect_values`, with the same traversal, the
/// same order and the same multikey semantics. Appends into `out`.
pub fn collect_values_bytes(gpa: std.mem.Allocator, bytes: []const u8, path: []const u8, out: *std.ArrayListUnmanaged(bson.Value), depth: usize) QueryError!void {
pub fn collect_values_bytes(
gpa: std.mem.Allocator,
bytes: []const u8,
path: []const u8,
out: *std.ArrayListUnmanaged(bson.Value),
depth: usize,
) QueryError!void {
var it = std.mem.splitScalar(u8, path, '.');
const first = it.next() orelse return;
const rest = it.rest();
@@ -241,7 +282,15 @@ pub fn collect_values_bytes(gpa: std.mem.Allocator, bytes: []const u8, path: []c
}
}
fn collect_from_value_bytes(gpa: std.mem.Allocator, bytes: []const u8, idx: *usize, tag: u8, path: []const u8, out: *std.ArrayListUnmanaged(bson.Value), depth: usize) QueryError!void {
fn collect_from_value_bytes(
gpa: std.mem.Allocator,
bytes: []const u8,
idx: *usize,
tag: u8,
path: []const u8,
out: *std.ArrayListUnmanaged(bson.Value),
depth: usize,
) QueryError!void {
if (depth > 8) {
try bson.skip_value(bytes, idx, tag);
return;
@@ -345,7 +394,13 @@ fn parse_op(name: []const u8) Op {
return op_names.get(name) orelse .unknown;
}
fn match_operator(gpa: std.mem.Allocator, op: Op, value: bson.Value, actuals: []const bson.Value, regex_options: []const u8) QueryError!bool {
fn match_operator(
gpa: std.mem.Allocator,
op: Op,
value: bson.Value,
actuals: []const bson.Value,
regex_options: []const u8,
) QueryError!bool {
if (op == .eq) {
for (actuals) |a| if (bson.compare(a, value) == .eq) return true;
return false;
@@ -489,7 +544,13 @@ fn match_operator(gpa: std.mem.Allocator, op: Op, value: bson.Value, actuals: []
/// Appends into `out`; on OOM, collection stops early (the engine is
/// already failing at that point). Public because index entry generation
/// must mirror field_matches exactly (src/index.zig).
pub fn collect_values(gpa: std.mem.Allocator, pairs: []const bson.Pair, path: []const u8, out: *std.ArrayListUnmanaged(bson.Value), depth: usize) QueryError!void {
pub fn collect_values(
gpa: std.mem.Allocator,
pairs: []const bson.Pair,
path: []const u8,
out: *std.ArrayListUnmanaged(bson.Value),
depth: usize,
) QueryError!void {
var it = std.mem.splitScalar(u8, path, '.');
const first = it.next() orelse return;
@@ -506,7 +567,13 @@ pub fn collect_values(gpa: std.mem.Allocator, pairs: []const bson.Pair, path: []
}
}
fn collect_from_value(gpa: std.mem.Allocator, v: bson.Value, path: []const u8, out: *std.ArrayListUnmanaged(bson.Value), depth: usize) QueryError!void {
fn collect_from_value(
gpa: std.mem.Allocator,
v: bson.Value,
path: []const u8,
out: *std.ArrayListUnmanaged(bson.Value),
depth: usize,
) QueryError!void {
if (depth > 8) return;
switch (v) {
.doc => |pairs| try collect_values(gpa, pairs, path, out, depth),
@@ -581,7 +648,14 @@ pub fn regex_match(pattern: []const u8, options: []const u8, text: []const u8) b
/// Match `pattern[p..]` against `text[t..]`, returning the new text
/// position on success (null on failure). Backtracks via recursion.
fn match_here(pattern: []const u8, p: *usize, text: []const u8, t: usize, ci: bool, dot_all: bool) ?usize {
fn match_here(
pattern: []const u8,
p: *usize,
text: []const u8,
t: usize,
ci: bool,
dot_all: bool,
) ?usize {
var pos = t;
while (p.* < pattern.len) {
const c = pattern[p.*];
@@ -669,7 +743,9 @@ fn match_here(pattern: []const u8, p: *usize, text: []const u8, t: usize, ci: bo
var q_end = element_end;
var min: usize = 1;
var max: usize = 1;
if (element_end < pattern.len and (pattern[element_end] == '*' or pattern[element_end] == '+' or pattern[element_end] == '?')) {
if (element_end < pattern.len and (pattern[element_end] == '*' or pattern[element_end] == '+' or pattern[
element_end
] == '?')) {
switch (pattern[element_end]) {
'*' => {
min = 0;
@@ -847,7 +923,11 @@ const SortCtx = struct {
/// Pull each document's sort-key values into one flat allocation, so the
/// comparator is pure and cannot fail.
fn decorate(arena: std.mem.Allocator, docs: []*const bson.Document, keys: []const SortKey) QueryError![]SortedDoc {
fn decorate(
arena: std.mem.Allocator,
docs: []*const bson.Document,
keys: []const SortKey,
) QueryError![]SortedDoc {
const entries = try arena.alloc(SortedDoc, docs.len);
const flat = try arena.alloc(bson.Value, docs.len * keys.len);
for (docs, 0..) |d, i| {
@@ -861,7 +941,11 @@ fn decorate(arena: std.mem.Allocator, docs: []*const bson.Document, keys: []cons
}
/// Sort `docs` in place by `keys`.
pub fn sort_docs(arena: std.mem.Allocator, docs: []*const bson.Document, keys: []const SortKey) QueryError!void {
pub fn sort_docs(
arena: std.mem.Allocator,
docs: []*const bson.Document,
keys: []const SortKey,
) QueryError!void {
if (keys.len == 0 or docs.len < 2) return;
const entries = try decorate(arena, docs, keys);
std.mem.sort(SortedDoc, entries, SortCtx{ .keys = keys }, SortCtx.less);
@@ -876,7 +960,12 @@ pub fn sort_docs(arena: std.mem.Allocator, docs: []*const bson.Document, keys: [
/// n log n comparisons to discard almost all of the result. This keeps a
/// k-element max-heap instead: one comparison against the heap root per
/// document, and only the survivors are ever ordered.
pub fn sort_docs_top_k(arena: std.mem.Allocator, docs: []*const bson.Document, keys: []const SortKey, k: usize) QueryError!void {
pub fn sort_docs_top_k(
arena: std.mem.Allocator,
docs: []*const bson.Document,
keys: []const SortKey,
k: usize,
) QueryError!void {
if (keys.len == 0 or docs.len < 2) return;
if (k == 0) return;
if (k >= docs.len) return sort_docs(arena, docs, keys);
@@ -936,7 +1025,12 @@ pub const ProjectionError = std.mem.Allocator.Error;
/// Apply a projection document, writing resulting pairs into `out` (which
/// should use the caller's arena so strings are owned).
pub fn project(arena: std.mem.Allocator, doc: *const bson.Document, proj: *const bson.Document, out: *std.ArrayListUnmanaged(bson.Pair)) ProjectionError!void {
pub fn project(
arena: std.mem.Allocator,
doc: *const bson.Document,
proj: *const bson.Document,
out: *std.ArrayListUnmanaged(bson.Pair),
) ProjectionError!void {
var inclusion: ?bool = null;
var non_id_count: usize = 0;
for (proj.pairs) |p| {
@@ -984,7 +1078,12 @@ pub fn project(arena: std.mem.Allocator, doc: *const bson.Document, proj: *const
}
/// Recursively apply exclusions to a nested document given the parent path.
fn exclude_doc(arena: std.mem.Allocator, pairs: []const bson.Pair, proj: *const bson.Document, parent: []const u8) ProjectionError![]const bson.Pair {
fn exclude_doc(
arena: std.mem.Allocator,
pairs: []const bson.Pair,
proj: *const bson.Document,
parent: []const u8,
) ProjectionError![]const bson.Pair {
var out: std.ArrayListUnmanaged(bson.Pair) = .empty;
errdefer out.deinit(arena);
for (pairs) |p| {
@@ -1024,7 +1123,12 @@ pub fn truthy(v: bson.Value) bool {
}
/// Include a dotted path (e.g. "a.b.c"), creating nested documents as needed.
fn project_path(arena: std.mem.Allocator, pairs: []const bson.Pair, path: []const u8, out: *std.ArrayListUnmanaged(bson.Pair)) ProjectionError!void {
fn project_path(
arena: std.mem.Allocator,
pairs: []const bson.Pair,
path: []const u8,
out: *std.ArrayListUnmanaged(bson.Pair),
) ProjectionError!void {
var it = std.mem.splitScalar(u8, path, '.');
const first = it.next() orelse return;
const rest = it.rest();
@@ -1294,10 +1398,10 @@ test "first_value_at agrees with collect_values on its first element" {
});
const paths = [_][]const u8{
"n", "sub", "sub.x", "sub.y", "sub.missing",
"items", "items.v", "items.0", "items.1.v", "items.9",
"nums", "nums.0", "nums.1", "nums.5",
"dup", "missing", "n.deeper", "", "sub.x.y",
"n", "sub", "sub.x", "sub.y", "sub.missing",
"items", "items.v", "items.0", "items.1.v", "items.9",
"nums", "nums.0", "nums.1", "nums.5", "dup",
"missing", "n.deeper", "", "sub.x.y",
};
for (paths) |path| {
@@ -1308,12 +1412,17 @@ test "first_value_at agrees with collect_values on its first element" {
const first = first_value_at(d.pairs, path, 0);
if (list.items.len == 0) {
testing.expect(first == null) catch |e| {
std.debug.print("path '{s}': collect empty but first_value_at returned a value\n", .{path});
std.debug.print("path '{s}': collect empty but first_value_at returned a value\n", .{
path,
});
return e;
};
} else {
testing.expect(first != null) catch |e| {
std.debug.print("path '{s}': collect got {d} values but first_value_at returned null\n", .{ path, list.items.len });
std.debug.print("path '{s}': collect got {d} values but first_value_at returned null\n", .{
path,
list.items.len,
});
return e;
};
testing.expectEqual(std.math.Order.eq, bson.compare(list.items[0], first.?)) catch |e| {
@@ -1372,7 +1481,11 @@ test "top-k selection matches a full sort on the leading page" {
const want = first_value_at(full[i].pairs, sk.path, 0) orelse bson.Value.null;
const got = first_value_at(topk[i].pairs, sk.path, 0) orelse bson.Value.null;
testing.expectEqual(std.math.Order.eq, bson.compare(want, got)) catch |e| {
std.debug.print("k={d} pos={d} key='{s}' diverged from the full sort\n", .{ k, i, sk.path });
std.debug.print("k={d} pos={d} key='{s}' diverged from the full sort\n", .{
k,
i,
sk.path,
});
return e;
};
}
@@ -1482,7 +1595,7 @@ test "byte matcher agrees with the tree matcher on a corpus" {
np += 1;
}
if (rand.boolean()) {
pairs[np] = .{ .key = "d", .value = .{ .doc = &.{ .{ .key = "e", .value = .{ .int32 = a } } } } };
pairs[np] = .{ .key = "d", .value = .{ .doc = &.{.{ .key = "e", .value = .{ .int32 = a } }} } };
np += 1;
}
var out: std.ArrayListUnmanaged(u8) = .empty;
@@ -1541,7 +1654,11 @@ test "byte matcher agrees with the tree matcher on a corpus" {
const tree = try matches(gpa, &filter_doc, &doc);
const byt = try matches_bytes(gpa, f_pairs.items, bytes);
if (tree != byt) {
std.debug.print("case {d}: filter mismatch: tree={} bytes={}\n", .{ case, tree, byt });
std.debug.print("case {d}: filter mismatch: tree={} bytes={}\n", .{
case,
tree,
byt,
});
return error.ByteMatcherMismatch;
}
}
@@ -1549,7 +1666,12 @@ test "byte matcher agrees with the tree matcher on a corpus" {
}
/// Public single-value operator matcher, used by $pull and $elemMatch.
pub fn value_matches_operator(gpa: std.mem.Allocator, op: []const u8, value: bson.Value, actual: bson.Value) QueryError!bool {
pub fn value_matches_operator(
gpa: std.mem.Allocator,
op: []const u8,
value: bson.Value,
actual: bson.Value,
) QueryError!bool {
var single: [1]bson.Value = .{actual};
return match_operator(gpa, parse_op(op), value, single[0..], "");
}

38
src/server.zig
View File

@@ -120,7 +120,10 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve
reader.interface.readSliceAll(&len_bytes) catch return; // clean client disconnect (EOF or RST)
const total: u32 = std.mem.readInt(u32, &len_bytes, .little);
if (total < 16 or total > wire.max_message_size) {
std.debug.print("multiforadb: bad message length {d} on conn {d}\n", .{ total, connection_id });
std.debug.print("multiforadb: bad message length {d} on conn {d}\n", .{
total,
connection_id,
});
return;
}
@@ -129,14 +132,22 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve
msg_buf.items.len = total;
std.mem.writeInt(u32, msg_buf.items[0..4], total, .little);
reader.interface.readSliceAll(msg_buf.items[4..]) catch |err| {
std.debug.print("multiforadb: read error on conn {d}: {s} (body, len {d})\n", .{ connection_id, @errorName(err), total });
std.debug.print("multiforadb: read error on conn {d}: {s} (body, len {d})\n", .{
connection_id,
@errorName(err),
total,
});
return;
};
var msg = wire.Message.parse(server.gpa, msg_buf.items) catch |err| {
// Unparseable request: close the connection.
const op: i32 = if (msg_buf.items.len >= 16) std.mem.readInt(i32, msg_buf.items[12..16], .little) else 0;
std.debug.print("multiforadb: bad message on conn {d}: {s} (opCode {d})\n", .{ connection_id, @errorName(err), op });
std.debug.print("multiforadb: bad message on conn {d}: {s} (opCode {d})\n", .{
connection_id,
@errorName(err),
op,
});
return;
};
defer msg.deinit();
@@ -146,7 +157,11 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve
commands.dispatch(&ctx, &msg, &reply) catch |err| {
// Discard any partial reply (the client would read the first
// ok field, which may already say 1) and send a clean error.
std.debug.print("multiforadb: dispatch error on conn {d} cmd {s}: {s}\n", .{ connection_id, msg.command_name(), @errorName(err) });
std.debug.print("multiforadb: dispatch error on conn {d} cmd {s}: {s}\n", .{
connection_id,
msg.command_name(),
@errorName(err),
});
reply.pairs.clearRetainingCapacity();
reply.put_error(
@intFromEnum(commands.ErrorCode.internal_error),
@@ -161,16 +176,25 @@ fn handle_connection_inner(io: std.Io, stream: std.Io.net.Stream, server: *Serve
else
reply.build(server.gpa, reply_request_id, msg.request_id, &out_buf);
built catch |err| {
std.debug.print("multiforadb: reply build error on conn {d}: {s}\n", .{ connection_id, @errorName(err) });
std.debug.print("multiforadb: reply build error on conn {d}: {s}\n", .{
connection_id,
@errorName(err),
});
return;
};
reply_request_id +%= 1;
writer.interface.writeAll(out_buf.items) catch |err| {
std.debug.print("multiforadb: write error on conn {d}: {s}\n", .{ connection_id, @errorName(err) });
std.debug.print("multiforadb: write error on conn {d}: {s}\n", .{
connection_id,
@errorName(err),
});
return;
};
writer.interface.flush() catch |err| {
std.debug.print("multiforadb: flush error on conn {d}: {s}\n", .{ connection_id, @errorName(err) });
std.debug.print("multiforadb: flush error on conn {d}: {s}\n", .{
connection_id,
@errorName(err),
});
return;
};
}

7
src/spill.zig
View File

@@ -37,7 +37,12 @@ pub fn main() !void {
docs[i] = try doc_of(gpa, &pairs);
_ = try ix.add_doc(gpa, docs[i], &[_]u8{ 'i', 'd', @intCast(i + 1) }, true);
}
std.debug.print("count={d} leaves={d} depth={d} overflow={d}\n", .{ ix.count(), ix.leaf_count, ix.depth, ix.overflow.items.len });
std.debug.print("count={d} leaves={d} depth={d} overflow={d}\n", .{
ix.count(),
ix.leaf_count,
ix.depth,
ix.overflow.items.len,
});
if (ix.overflow.items.len < 100_000) return error.NoSpill;
// Every entry is found by exact key.

7
src/spill2.zig
View File

@@ -48,7 +48,12 @@ pub fn main() !void {
const d = try doc_of(gpa, &pairs);
_ = try ix.add_doc(gpa, d, id, false);
}
std.debug.print("count={d} leaves={d} depth={d} overflow={d}\n", .{ ix.count(), ix.leaf_count, ix.depth, ix.overflow.items.len });
std.debug.print("count={d} leaves={d} depth={d} overflow={d}\n", .{
ix.count(),
ix.leaf_count,
ix.depth,
ix.overflow.items.len,
});
if (ix.count() != N) return error.Bad;
// Spot-check exact lookups.

65
src/storage.zig
View File

@@ -227,7 +227,10 @@ pub const Log = struct {
while (true) {
var hdr: [block_header_len]u8 = undefined;
const n = self.file.readPositionalAll(self.io, &hdr, pos) catch |err| {
std.debug.print("multiforadb: log read error at {d}: {s}\n", .{ pos, @errorName(err) });
std.debug.print("multiforadb: log read error at {d}: {s}\n", .{
pos,
@errorName(err),
});
return error.InvalidLog;
};
if (n == 0) return; // clean end
@@ -262,7 +265,10 @@ pub const Log = struct {
codec_raw => try decomp.appendSlice(self.gpa, payload),
codec_lz4 => try lz4_decompress(self.gpa, payload, &decomp),
else => {
std.debug.print("multiforadb: unknown block codec {d} at {d}\n", .{ codec, pos });
std.debug.print("multiforadb: unknown block codec {d} at {d}\n", .{
codec,
pos,
});
return error.InvalidLog;
},
}
@@ -280,7 +286,13 @@ pub const Log = struct {
/// and deliver it. Returns the record's byte length. Any framing failure
/// here is interior corruption: a block was sealed only with complete
/// records, and its hash proved the stored bytes intact.
fn parse_record(self: *Log, bytes: []const u8, pos: u64, ctx: *anyopaque, callback: ReplayFn) !usize {
fn parse_record(
self: *Log,
bytes: []const u8,
pos: u64,
ctx: *anyopaque,
callback: ReplayFn,
) !usize {
if (bytes.len < 4) return error.InvalidLog;
const total: u32 = std.mem.readInt(u32, bytes[0..4], .little);
if (total < header_len) {
@@ -321,26 +333,57 @@ pub const Log = struct {
return total;
}
pub fn append_upsert(self: *Log, db: []const u8, coll: []const u8, doc: []const u8, seq: u64) !void {
pub fn append_upsert(
self: *Log,
db: []const u8,
coll: []const u8,
doc: []const u8,
seq: u64,
) !void {
try self.append(record_type_upsert, db, coll, doc, seq);
}
pub fn append_delete(self: *Log, db: []const u8, coll: []const u8, doc: []const u8, seq: u64) !void {
pub fn append_delete(
self: *Log,
db: []const u8,
coll: []const u8,
doc: []const u8,
seq: u64,
) !void {
try self.append(record_type_delete, db, coll, doc, seq);
}
/// The payload is the canonical index spec document ({v, key, name,
/// unique?, sparse?}); only apply_record interprets it.
pub fn append_index_create(self: *Log, db: []const u8, coll: []const u8, doc: []const u8, seq: u64) !void {
pub fn append_index_create(
self: *Log,
db: []const u8,
coll: []const u8,
doc: []const u8,
seq: u64,
) !void {
try self.append(record_type_index_create, db, coll, doc, seq);
}
/// The payload is {name: "..."}; only apply_record interprets it.
pub fn append_index_drop(self: *Log, db: []const u8, coll: []const u8, doc: []const u8, seq: u64) !void {
pub fn append_index_drop(
self: *Log,
db: []const u8,
coll: []const u8,
doc: []const u8,
seq: u64,
) !void {
try self.append(record_type_index_drop, db, coll, doc, seq);
}
fn append(self: *Log, rtype: u8, db: []const u8, coll: []const u8, doc: []const u8, seq: u64) !void {
fn append(
self: *Log,
rtype: u8,
db: []const u8,
coll: []const u8,
doc: []const u8,
seq: u64,
) !void {
if (std.mem.indexOfScalar(u8, db, 0) != null or std.mem.indexOfScalar(u8, coll, 0) != null) {
return error.NulInName;
}
@@ -547,7 +590,11 @@ fn emit_literals(dst: []u8, literals: []const u8) usize {
/// Decompress an LZ4 block into `out` (appended). The block hash has
/// already proven the input intact when this runs during replay, so
/// structural failures here mean a bug or a raw-codec mismatch.
fn lz4_decompress(gpa: std.mem.Allocator, src: []const u8, out: *std.ArrayListUnmanaged(u8)) error{ CorruptLz4, OutOfMemory }!void {
fn lz4_decompress(
gpa: std.mem.Allocator,
src: []const u8,
out: *std.ArrayListUnmanaged(u8),
) error{ CorruptLz4, OutOfMemory }!void {
var ip: usize = 0;
while (ip < src.len) {
const token = src[ip];

61
src/stress.zig
View File

@@ -16,7 +16,15 @@ fn doc_of(gpa: std.mem.Allocator, pairs: []const bson.Pair) ![]u8 {
const Fact = struct { a: i32, b: i32 };
fn check_range(gpa: std.mem.Allocator, ix: *const index.Index, prefix: bson.Value, lo: ?bson.Value, hi: ?bson.Value, facts: []const Fact, alive: []const bool) !void {
fn check_range(
gpa: std.mem.Allocator,
ix: *const index.Index,
prefix: bson.Value,
lo: ?bson.Value,
hi: ?bson.Value,
facts: []const Fact,
alive: []const bool,
) !void {
var out: std.ArrayListUnmanaged([]const u8) = .empty;
defer out.deinit(gpa);
try ix.lookup_range(gpa, &.{prefix}, lo, true, hi, false, &out);
@@ -29,7 +37,13 @@ fn check_range(gpa: std.mem.Allocator, ix: *const index.Index, prefix: bson.Valu
expected += 1;
}
if (out.items.len != expected) {
std.debug.print("MISMATCH: prefix={d} lo={?d} hi={?d}: got {d}, want {d}\n", .{ prefix.int32, if (lo) |l| l.int32 else null, if (hi) |h| h.int32 else null, out.items.len, expected });
std.debug.print("MISMATCH: prefix={d} lo={?d} hi={?d}: got {d}, want {d}\n", .{
prefix.int32,
if (lo) |l| l.int32 else null,
if (hi) |h| h.int32 else null,
out.items.len,
expected,
});
std.process.exit(1);
}
}
@@ -61,7 +75,12 @@ pub fn main() !void {
try ix.append_doc_entries(gpa, d, id);
}
_ = try ix.finish_bulk(gpa, false);
std.debug.print("bulk: count={d} leaves={d} depth={d} nodes={d}\n", .{ ix.count(), ix.leaf_count, ix.depth, ix.nodes.items.len });
std.debug.print("bulk: count={d} leaves={d} depth={d} nodes={d}\n", .{
ix.count(),
ix.leaf_count,
ix.depth,
ix.nodes.items.len,
});
if (ix.count() != N) return error.BadCount;
// Random range checks against brute force.
@@ -88,7 +107,12 @@ pub fn main() !void {
const d = try doc_of(gpa, &pairs);
_ = try ix.add_doc(gpa, d, id, false);
}
std.debug.print("after inserts: count={d} leaves={d} depth={d} nodes={d}\n", .{ ix.count(), ix.leaf_count, ix.depth, ix.nodes.items.len });
std.debug.print("after inserts: count={d} leaves={d} depth={d} nodes={d}\n", .{
ix.count(),
ix.leaf_count,
ix.depth,
ix.nodes.items.len,
});
if (ix.count() != N + M) return error.BadCount;
for (0..500) |_| {
const a = rand.intRangeAtMost(i32, 0, 99);
@@ -113,7 +137,12 @@ pub fn main() !void {
ix.remove_doc(gpa, d, ids.items[i]);
alive.items[i] = false;
}
std.debug.print("after deletes: count={d} leaves={d} depth={d} nodes={d}\n", .{ ix.count(), ix.leaf_count, ix.depth, ix.nodes.items.len });
std.debug.print("after deletes: count={d} leaves={d} depth={d} nodes={d}\n", .{
ix.count(),
ix.leaf_count,
ix.depth,
ix.nodes.items.len,
});
if (ix.count() != (N + M) - (N + M) / 3 - 1) return error.BadCount;
for (0..500) |_| {
const a = rand.intRangeAtMost(i32, 0, 99);
@@ -131,9 +160,15 @@ pub fn main() !void {
defer out.deinit(gpa);
try ix.lookup_eq(gpa, &.{.{ .int32 = a }}, &out);
var expected: usize = 0;
for (facts.items, 0..) |f, fi| { if (alive.items[fi] and f.a == a) expected += 1; }
for (facts.items, 0..) |f, fi| {
if (alive.items[fi] and f.a == a) expected += 1;
}
if (out.items.len != expected) {
std.debug.print("EQ MISMATCH a={d}: got {d} want {d}\n", .{ a, out.items.len, expected });
std.debug.print("EQ MISMATCH a={d}: got {d} want {d}\n", .{
a,
out.items.len,
expected,
});
return error.BadCount;
}
}
@@ -154,11 +189,19 @@ pub fn main() !void {
ix.remove_doc(gpa, d, ids.items[i]);
remaining -= 1;
if (ix.count() != remaining) {
std.debug.print("COUNT MISMATCH during drain: {d} != {d}\n", .{ ix.count(), remaining });
std.debug.print("COUNT MISMATCH during drain: {d} != {d}\n", .{
ix.count(),
remaining,
});
return error.BadCount;
}
}
std.debug.print("after full drain: count={d} leaves={d} depth={d} nodes={d}\n", .{ ix.count(), ix.leaf_count, ix.depth, ix.nodes.items.len });
std.debug.print("after full drain: count={d} leaves={d} depth={d} nodes={d}\n", .{
ix.count(),
ix.leaf_count,
ix.depth,
ix.nodes.items.len,
});
if (ix.count() != 0) return error.BadCount;
// The drained tree still accepts and finds entries.
pairs[0] = .{ .key = "a", .value = .{ .int32 = 7 } };

26
src/update.zig
View File

@@ -21,7 +21,12 @@ pub fn apply(doc: *bson.Document, update: *const bson.Document) UpdateError!void
doc.pairs = try pairs.toOwnedSlice(arena);
}
fn apply_operator(arena: std.mem.Allocator, pairs: *std.ArrayListUnmanaged(bson.Pair), op: []const u8, value: bson.Value) UpdateError!void {
fn apply_operator(
arena: std.mem.Allocator,
pairs: *std.ArrayListUnmanaged(bson.Pair),
op: []const u8,
value: bson.Value,
) UpdateError!void {
if (std.mem.eql(u8, op, "$set")) {
const ops = doc_pairs(value) orelse return error.InvalidUpdate;
for (ops) |p| {
@@ -146,7 +151,11 @@ fn parse_index(seg: []const u8) ?usize {
}
/// Shallow-copy a slice into a growable list backed by `arena`.
fn copy_to_list(comptime T: type, arena: std.mem.Allocator, items: []const T) UpdateError!std.ArrayListUnmanaged(T) {
fn copy_to_list(
comptime T: type,
arena: std.mem.Allocator,
items: []const T,
) UpdateError!std.ArrayListUnmanaged(T) {
var out: std.ArrayListUnmanaged(T) = .empty;
errdefer out.deinit(arena);
try out.appendSlice(arena, items);
@@ -173,7 +182,12 @@ fn get_value(pairs: []const bson.Pair, segs: []const []const u8) ?bson.Value {
};
}
fn set_path(arena: std.mem.Allocator, pairs: *std.ArrayListUnmanaged(bson.Pair), segs: []const []const u8, value: bson.Value) UpdateError!void {
fn set_path(
arena: std.mem.Allocator,
pairs: *std.ArrayListUnmanaged(bson.Pair),
segs: []const []const u8,
value: bson.Value,
) UpdateError!void {
if (segs.len == 1) {
if (find_pair(pairs.items, segs[0])) |idx| {
pairs.items[idx].value = value;
@@ -237,7 +251,11 @@ fn set_path(arena: std.mem.Allocator, pairs: *std.ArrayListUnmanaged(bson.Pair),
}
}
fn unset_path(arena: std.mem.Allocator, pairs: *std.ArrayListUnmanaged(bson.Pair), segs: []const []const u8) void {
fn unset_path(
arena: std.mem.Allocator,
pairs: *std.ArrayListUnmanaged(bson.Pair),
segs: []const []const u8,
) void {
if (segs.len == 1) {
if (find_pair(pairs.items, segs[0])) |idx| {
_ = pairs.orderedRemove(idx);

8
src/wire.zig
View File

@@ -256,7 +256,13 @@ pub const Reply = struct {
}
/// Serialize this reply as a full OP_MSG message.
pub fn build(self: *Reply, gpa: std.mem.Allocator, request_id: u32, response_to: u32, out: *std.ArrayListUnmanaged(u8)) !void {
pub fn build(
self: *Reply,
gpa: std.mem.Allocator,
request_id: u32,
response_to: u32,
out: *std.ArrayListUnmanaged(u8),
) !void {
try write_message(gpa, request_id, response_to, 0, self.pairs.items, out);
}
};