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

storage: block-framed LZ4-compressed log (roadmap item 3)

Browse Source 
The log is now a 16-byte file header (magic, version, codec, block
target) plus a sequence of blocks. Each block keeps the pre-existing
record framing unchanged, so Engine.apply_record does not change; records
never straddle blocks (appends accumulate in memory and the block seals
at ~256 KiB). The block header's integrity hash covers the stored payload
bytes exactly as they sit on disk, so the decompressor only ever sees
input already proven intact. Torn tails stay distinguishable from
interior corruption exactly as before: a short read, an impossible
length, or a hash mismatch in the final block truncates cleanly (later
appends overwrite the garbage); a hash mismatch anywhere else is
error.InvalidLog.

The codec is a hand-rolled LZ4 block compressor/decompressor (~1.7 GB/s
measured) with a per-block codec byte falling back to raw when
compression does not help; the header keeps raw legal so zstd can be
swapped in later. Zig 0.16 ships zstd decompression only, and deflate
would cap writes below the insert rate.

Engine.compact goes through the same Log API (deferred sync, one commit)
and compresses for free; sync() seals the pending block before fsyncing,
so the acknowledged-write durability semantics are unchanged (an
unsealed block holds only unacknowledged batch records).

Measured (tests/e2e/results/phase4.txt): db on disk 1025 -> 97 MB, now
smaller than MongoDB's own compressed files; bulk insert 816 -> 722 MB/s
(the accepted compression cost); reopen unchanged at 0.8 s.

Verified: unit suite in all three optimize modes (new LZ4 round-trip,
corrupt-block, and torn-tail truncation tests), the crash pair, e2e6
(kill -9 mid-write), and two full benchmark runs.
This commit is contained in:
Alexey Shakhmatov
2026-08-02 21:35:22 +03:00
parent 58914a69c3
commit b4585106f1
4 changed files with 629 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

589
src/storage.zig
View File

@@ -1,13 +1,41 @@
//! Append-only record log. Each record is:
//! Append-only record log, block-framed and compressed (roadmap item 3).
//!
//! File layout:
//! [0..4) u32 magic "MLOG"
//! [4] u8 version (1)
//! [5] u8 codec (0 raw, 1 lz4) — the preferred codec for new blocks
//! [6..8) reserved
//! [8..16) u64 block_target — soft fill target per block (256 KiB)
//! [16..) blocks...
//!
//! Block layout:
//! [0..4) u32 total — block bytes including this header
//! [4..12) u64 hash over bytes [12..total) — the stored (compressed)
//! payload exactly as it sits on disk, so the decompressor only
//! ever sees input already proven intact
//! [12] u8 codec (0 raw, 1 lz4); a block falls back to raw when
//! compression does not help
//! [13..16) reserved
//! [16..) payload — the block's records, raw or LZ4-compressed
//!
//! Each block holds the pre-existing record framing unchanged:
//! [0..4) u32 len — total record bytes
//! [4..12) u64 hash over bytes [12..len)
//! [12..20) u64 seq
//! [20] u8 type
//! [21..24) reserved
//! [24..) db\0 coll\0 bson doc
//! All integers little-endian. Reads and writes are positional, so the fd
//! offset never matters. A torn tail record (crash mid-append) is detected
//! during replay and skipped.
//! All integers little-endian. Records never straddle blocks: appends
//! accumulate into an in-memory block that is sealed (compressed, written,
//! and, when not group-committed, synced) once the next record would push
//! it past the target. Engine.apply_record is untouched — it still sees the
//! same records.
//!
//! Torn tail versus interior corruption stay distinguishable, exactly as
//! before: a short read, an impossible length, or a hash mismatch in the
//! final block means a crash mid-append — replay stops there and later
//! appends overwrite the garbage. A hash mismatch in any earlier block is
//! error.InvalidLog.
//!
//! The integrity hash is XxHash3, not CRC32. Both are checked for the same
//! thing — did these bytes survive intact — but std.hash.Crc32 is the
@@ -23,9 +51,22 @@ pub const record_type_delete: u8 = 2;
pub const record_type_index_create: u8 = 3;
pub const record_type_index_drop: u8 = 4;
pub const header_len: usize = 24; // len + hash + seq + type + reserved
pub const header_len: usize = 24; // per-record: len + hash + seq + type + reserved
pub const block_header_len: usize = 16; // block: total + hash + codec + reserved
pub const file_header_len: usize = 16; // magic + version + codec + block_target
/// Integrity hash over a record's bytes after the length and hash fields.
const file_magic: u32 = 0x4D4C4F47; // "MLOG"
const file_version: u8 = 1;
pub const codec_raw: u8 = 0;
pub const codec_lz4: u8 = 1;
/// Soft fill target for a block's record region. Sealed when the next
/// record would push the block past this; a single oversized record gets
/// its own block.
pub const block_target: usize = 256 * 1024;
/// Integrity hash over a record's bytes after the length and hash fields,
/// or over a block's payload bytes after its header fields.
fn record_hash(bytes: []const u8) u64 {
return std.hash.XxHash3.hash(0, bytes);
}
@@ -34,6 +75,10 @@ fn record_hash(bytes: []const u8) u64 {
/// announced maxBsonObjectSize with room for names and header.
pub const max_record_payload: usize = 16 * 1024 * 1024 + 64 * 1024;
/// Largest stored block payload: block_target plus one oversized record,
/// plus LZ4's worst-case expansion of it.
const max_block_payload: usize = max_record_payload + 1024 * 1024;
pub const Record = struct {
seq: u64,
type: u8,
@@ -42,7 +87,8 @@ pub const Record = struct {
};
pub const Error = error{
InvalidLog, // corrupt interior record (bad hash or impossible length)
InvalidLog, // corrupt interior block or record (bad hash or impossible length)
CorruptLz4, // the decoder saw structurally invalid input
};
/// Callback receives transient slices and a heap-allocated, freshly parsed
@@ -56,10 +102,17 @@ pub const Log = struct {
path: []const u8,
end_pos: u64,
log_bytes: u64, // bytes written since the log was last rewritten
codec: u8,
// Reused record-framing buffer. Appends are single-writer (the engine's
// exclusive lock), so one buffer avoids a realloc cycle per record.
scratch: std.ArrayListUnmanaged(u8),
/// Group commit: while set, appends skip the per-record fsync and the
/// The current block's records, uncompressed, until the block is sealed.
block: std.ArrayListUnmanaged(u8),
/// LZ4 output buffer; also the sealed block's payload when raw.
compressed: std.ArrayListUnmanaged(u8),
/// LZ4 hash table (positions of recent 4-byte sequences).
lz4_table: []u32,
/// Group commit: while set, appends skip the per-block fsync and the
/// caller issues one sync for the whole batch (see Engine.begin_batch /
/// end_batch). Every acknowledged write is still fsynced before the
/// reply, so the crash guarantees are unchanged.
@@ -81,19 +134,56 @@ pub const Log = struct {
error.FileNotFound => try dir.createFile(io, abs_path, .{ .read = true }),
else => return err,
};
return .{
var self: Log = .{
.gpa = gpa,
.io = io,
.file = file,
.path = abs_path,
.end_pos = 0,
.end_pos = file_header_len,
.log_bytes = 0,
.codec = codec_lz4,
.scratch = .empty,
.block = .empty,
.compressed = .empty,
.lz4_table = undefined,
.defer_sync = false,
};
errdefer {
self.scratch.deinit(gpa);
self.block.deinit(gpa);
self.compressed.deinit(gpa);
file.close(io);
}
self.lz4_table = try gpa.alloc(u32, lz4_table_size);
errdefer gpa.free(self.lz4_table);
// Read or write the file header. A fresh file gets one; an existing
// file must carry a valid one (a torn header on a never-opened db is
// unrecoverable and reported as invalid).
var hdr: [file_header_len]u8 = undefined;
const n = file.readPositionalAll(io, &hdr, 0) catch return error.InvalidLog;
if (n == 0) {
write_file_header(&hdr, codec_lz4);
try file.writePositionalAll(io, &hdr, 0);
} else {
if (n < file_header_len) return error.InvalidLog;
if (std.mem.readInt(u32, hdr[0..4], .little) != file_magic) return error.InvalidLog;
if (hdr[4] != file_version) return error.InvalidLog;
self.codec = hdr[5];
if (self.codec != codec_raw and self.codec != codec_lz4) return error.InvalidLog;
}
return self;
}
pub fn close(self: *Log) void {
// Best-effort: flush an unsealed block so nothing acknowledged is
// left only in memory. The caller syncs before close on the paths
// that care about durability.
self.seal_block() catch {};
self.block.deinit(self.gpa);
self.compressed.deinit(self.gpa);
self.gpa.free(self.lz4_table);
self.scratch.deinit(self.gpa);
self.file.close(self.io);
self.gpa.free(self.path);
@@ -101,72 +191,108 @@ pub const Log = struct {
/// Replay all valid records from the beginning of the file.
pub fn replay(self: *Log, ctx: *anyopaque, callback: ReplayFn) !void {
var chunk: [64 * 1024]u8 = undefined;
var heap_buf: []u8 = &.{};
defer if (heap_buf.len > 0) self.gpa.free(heap_buf);
var pos: u64 = 0;
var decomp: std.ArrayListUnmanaged(u8) = .empty;
defer decomp.deinit(self.gpa);
var pos: u64 = file_header_len;
while (true) {
const len_read = self.file.readPositionalAll(self.io, chunk[0..4], pos) catch |err| {
var hdr: [block_header_len]u8 = undefined;
const n = self.file.readPositionalAll(self.io, &hdr, pos) catch |err| {
std.debug.print("mongo-lite: log read error at {d}: {s}\n", .{ pos, @errorName(err) });
return error.InvalidLog;
};
if (len_read == 0) return; // clean end
if (len_read < 4) return; // torn tail
const total: u32 = std.mem.readInt(u32, chunk[0..4], .little);
if (total < header_len) {
std.debug.print("mongo-lite: corrupt record length {d} at {d}\n", .{ total, pos });
if (n == 0) return; // clean end
if (n < block_header_len) return; // torn tail: partial block header
const total: u32 = std.mem.readInt(u32, hdr[0..4], .little);
if (total < block_header_len or total - block_header_len > max_block_payload) {
std.debug.print("mongo-lite: corrupt block length {d} at {d}\n", .{ total, pos });
return; // torn tail: impossible length, nothing to validate
}
const payload_len: usize = total - block_header_len;
const payload = try self.gpa.alloc(u8, payload_len);
defer self.gpa.free(payload);
const payload_read = self.file.readPositionalAll(self.io, payload, pos + block_header_len) catch return error.InvalidLog;
if (payload_read < payload_len) return; // torn tail: crash mid-append
const hash_stored: u64 = std.mem.readInt(u64, hdr[4..12], .little);
const hash_actual = record_hash(payload);
if (hash_stored != hash_actual) {
// A mismatch in the final block is a torn append (the block
// was partially flushed); a mismatch before more bytes
// follow is interior corruption. The block header hash
// covers the stored bytes, so the decompressor below only
// runs on input already proven intact.
const file_len = self.file.length(self.io) catch return error.InvalidLog;
if (pos + total >= file_len) return;
std.debug.print("mongo-lite: corrupt block hash at {d}\n", .{pos});
return error.InvalidLog;
}
const payload_len: usize = total - 4;
// Documents up to maxBsonObjectSize are legal; anything larger is
// corruption. Covers a hostile length prefix from a truncated file.
if (payload_len > max_record_payload) {
std.debug.print("mongo-lite: record too large at {d}\n", .{pos});
return error.InvalidLog;
const codec = hdr[12];
decomp.clearRetainingCapacity();
switch (codec) {
codec_raw => try decomp.appendSlice(self.gpa, payload),
codec_lz4 => try lz4_decompress(self.gpa, payload, &decomp),
else => {
std.debug.print("mongo-lite: unknown block codec {d} at {d}\n", .{ codec, pos });
return error.InvalidLog;
},
}
const payload = if (payload_len <= chunk.len) chunk[0..payload_len] else blk: {
if (heap_buf.len < payload_len) {
if (heap_buf.len > 0) self.gpa.free(heap_buf);
heap_buf = try self.gpa.alloc(u8, payload_len);
}
break :blk heap_buf[0..payload_len];
};
const payload_read = self.file.readPositionalAll(self.io, payload, pos + 4) catch return error.InvalidLog;
if (payload_read < payload_len) return; // torn tail — crash during append
// Offsets here are relative to `payload`, which starts at the
// record's byte 4 — so payload[0..8) is the record's hash field,
// and header_len - 4 is where db\0coll\0 begins.
const hash_stored: u64 = std.mem.readInt(u64, payload[0..8], .little);
const hash_actual = record_hash(payload[8..payload_len]);
if (hash_stored != hash_actual) return error.InvalidLog;
var idx: usize = header_len - 4;
const seq: u64 = std.mem.readInt(u64, payload[8..16], .little);
const rtype = payload[16];
const db = read_cstring(payload, &idx) orelse return error.InvalidLog;
const coll = read_cstring(payload, &idx) orelse return error.InvalidLog;
if (idx > payload_len) return error.InvalidLog;
const doc_bytes = payload[idx..payload_len];
const doc = try self.gpa.create(bson.Document);
doc.* = bson.Document.parse(self.gpa, doc_bytes) catch {
self.gpa.destroy(doc);
std.debug.print("mongo-lite: unparseable doc in log at {d}\n", .{pos});
return error.InvalidLog;
};
try callback(ctx, .{
.seq = seq,
.type = rtype,
.db = db,
.coll = coll,
}, doc);
var idx: usize = 0;
while (idx < decomp.items.len) {
idx += try self.parse_record(decomp.items[idx..], pos + idx, ctx, callback);
}
pos += total;
self.end_pos = pos;
}
}
/// Parse one record from `bytes` (a block's decompressed record region)
/// 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 {
if (bytes.len < 4) return error.InvalidLog;
const total: u32 = std.mem.readInt(u32, bytes[0..4], .little);
if (total < header_len) {
std.debug.print("mongo-lite: corrupt record length {d} at {d}\n", .{ total, pos });
return error.InvalidLog;
}
const payload_len: usize = total - 4;
if (payload_len > max_record_payload) {
std.debug.print("mongo-lite: record too large at {d}\n", .{pos});
return error.InvalidLog;
}
if (bytes.len < total) return error.InvalidLog; // record crosses the block end
const payload = bytes[4..total];
const hash_stored: u64 = std.mem.readInt(u64, payload[0..8], .little);
const hash_actual = record_hash(payload[8..]);
if (hash_stored != hash_actual) return error.InvalidLog;
var idx: usize = header_len - 4;
const seq: u64 = std.mem.readInt(u64, payload[8..16], .little);
const rtype = payload[16];
const db = read_cstring(payload, &idx) orelse return error.InvalidLog;
const coll = read_cstring(payload, &idx) orelse return error.InvalidLog;
if (idx > payload_len) return error.InvalidLog;
const doc_bytes = payload[idx..payload_len];
const doc = try self.gpa.create(bson.Document);
doc.* = bson.Document.parse(self.gpa, doc_bytes) catch {
self.gpa.destroy(doc);
std.debug.print("mongo-lite: unparseable doc in log at {d}\n", .{pos});
return error.InvalidLog;
};
try callback(ctx, .{
.seq = seq,
.type = rtype,
.db = db,
.coll = coll,
}, doc);
return total;
}
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);
}
@@ -204,15 +330,57 @@ pub const Log = struct {
const total: u32 = @intCast(buf.items.len);
std.mem.writeInt(u32, buf.items[0..4], total, .little);
std.mem.writeInt(u64, buf.items[4..12], record_hash(buf.items[12..]), .little);
try self.file.writePositionalAll(self.io, buf.items, self.end_pos);
self.end_pos += buf.items.len;
self.log_bytes += buf.items.len;
if (!self.defer_sync) try self.file.sync(self.io);
// Seal the current block when the next record would push it past the
// target; a single oversized record keeps its own block.
if (self.block.items.len > 0 and self.block.items.len + buf.items.len > block_target) {
try self.seal_block();
}
try self.block.appendSlice(self.gpa, buf.items);
if (!self.defer_sync) {
// Durable before the reply: seal this block and fsync.
try self.seal_block();
try self.file.sync(self.io);
}
}
/// One fsync for the whole deferred batch. Callers must have set
/// defer_sync, appended, and cleared defer_sync again before the reply.
/// Compress and write the current block, then reset it. No-op when it is
/// empty. Blocks fall back to raw when compression does not help.
fn seal_block(self: *Log) !void {
if (self.block.items.len == 0) return;
defer self.block.clearRetainingCapacity();
var codec: u8 = codec_raw;
var payload: []const u8 = self.block.items;
if (self.codec == codec_lz4) {
const worst = self.block.items.len + self.block.items.len / 255 + 16;
try self.compressed.ensureTotalCapacity(self.gpa, worst);
const out = self.compressed.allocatedSlice()[0..worst];
const n = lz4_compress(self.block.items, out, self.lz4_table);
if (n < self.block.items.len) {
payload = out[0..n];
codec = codec_lz4;
}
}
var hdr: [block_header_len]u8 = undefined;
const total: u32 = @intCast(block_header_len + payload.len);
std.mem.writeInt(u32, hdr[0..4], total, .little);
std.mem.writeInt(u64, hdr[4..12], record_hash(payload), .little);
hdr[12] = codec;
hdr[13..block_header_len].* = [_]u8{ 0, 0, 0 };
try self.file.writePositionalAll(self.io, &hdr, self.end_pos);
try self.file.writePositionalAll(self.io, payload, self.end_pos + block_header_len);
self.end_pos += total;
self.log_bytes += total;
}
/// One fsync for the whole deferred batch, sealing the current block
/// first. Callers must have set defer_sync, appended, and cleared
/// defer_sync again before the reply.
pub fn sync(self: *Log) !void {
try self.seal_block();
try self.file.sync(self.io);
}
@@ -225,6 +393,182 @@ pub const Log = struct {
}
};
fn write_file_header(hdr: *[file_header_len]u8, codec: u8) void {
std.mem.writeInt(u32, hdr[0..4], file_magic, .little);
hdr[4] = file_version;
hdr[5] = codec;
hdr[6] = 0;
hdr[7] = 0;
std.mem.writeInt(u64, hdr[8..16], block_target, .little);
}
// ---------------------------------------------------------------------------
// LZ4 block codec
// ---------------------------------------------------------------------------
//
// The standard LZ4 block format: a sequence of (token, literals, offset,
// match-length) items. The token's high nibble is the literal length
// (15 = extended by bytes), the low nibble is the match length minus 4
// (15 = extended). Matches reference earlier output by a 16-bit offset and
// may overlap it. Self-contained: only this decoder reads this log.
const lz4_table_size: usize = 4096;
fn lz4_hash(v: u32) u32 {
return (v *% 2654435761) >> 20;
}
/// Compress `src` into `dst`, which must hold at least
/// src.len + src.len/255 + 16 bytes. Returns the compressed length.
pub fn lz4_compress(src: []const u8, dst: []u8, table: []u32) usize {
if (src.len == 0) return 0;
@memset(table, 0);
var out: usize = 0;
var anchor: usize = 0; // start of the pending literal run
var ip: usize = 0;
// Matches may not start in the final 5 bytes (they end there), keeping
// the stream standard-compliant: the last 5 bytes are always literals.
const limit = if (src.len >= 5) src.len - 5 else 0;
while (ip < limit) {
// Scan for a 4-byte match via the hash table.
var match_pos: ?usize = null;
while (ip + 4 <= src.len) : (ip += 1) {
const h = lz4_hash(std.mem.readInt(u32, src[ip..][0..4], .little));
const cand = table[h];
table[h] = @intCast(ip + 1); // 0 means empty; positions are stored +1
if (cand != 0) {
const cp = cand - 1;
const back = ip - cp;
if (back >= lz4_min_match and back <= 65535 and
std.mem.readInt(u32, src[cp..][0..4], .little) == std.mem.readInt(u32, src[ip..][0..4], .little))
{
match_pos = cp;
break;
}
}
}
const mp = match_pos orelse break;
// Extend the match, stopping at the end or the 5-byte literal tail.
var match_len: usize = lz4_min_match;
const match_limit = src.len - 5;
while (ip + match_len < match_limit and src[ip + match_len] == src[mp + match_len]) : (match_len += 1) {}
out += emit_sequence(dst[out..], src[anchor..ip], @intCast(ip - mp), match_len);
ip += match_len;
anchor = ip;
}
out += emit_literals(dst[out..], src[anchor..]);
return out;
}
const lz4_min_match: usize = 4;
/// Emit one (token, literals, offset, match length) item.
fn emit_sequence(dst: []u8, literals: []const u8, offset: u16, match_len: usize) usize {
var out: usize = 0;
const lit_len = literals.len;
var token: u8 = if (lit_len >= 15) 0xF0 else @intCast(lit_len << 4);
const mlen = match_len - lz4_min_match;
token |= if (mlen >= 15) 0x0F else @intCast(mlen);
dst[out] = token;
out += 1;
if (lit_len >= 15) {
var rem = lit_len - 15;
while (rem >= 255) : (rem -= 255) {
dst[out] = 255;
out += 1;
}
dst[out] = @intCast(rem);
out += 1;
}
@memcpy(dst[out .. out + lit_len], literals);
out += lit_len;
std.mem.writeInt(u16, dst[out..][0..2], offset, .little);
out += 2;
if (mlen >= 15) {
var rem = mlen - 15;
while (rem >= 255) : (rem -= 255) {
dst[out] = 255;
out += 1;
}
dst[out] = @intCast(rem);
out += 1;
}
return out;
}
/// Emit a trailing literal run (a final item with no match).
fn emit_literals(dst: []u8, literals: []const u8) usize {
if (literals.len == 0) return 0;
var out: usize = 0;
dst[out] = if (literals.len >= 15) 0xF0 else @intCast(literals.len << 4);
out += 1;
if (literals.len >= 15) {
var rem = literals.len - 15;
while (rem >= 255) : (rem -= 255) {
dst[out] = 255;
out += 1;
}
dst[out] = @intCast(rem);
out += 1;
}
@memcpy(dst[out .. out + literals.len], literals);
out += literals.len;
return out;
}
/// 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 {
var ip: usize = 0;
while (ip < src.len) {
const token = src[ip];
ip += 1;
var lit_len: usize = token >> 4;
if (lit_len == 15) {
while (true) {
if (ip >= src.len) return error.CorruptLz4;
const b = src[ip];
ip += 1;
lit_len += b;
if (b != 255) break;
}
}
if (ip + lit_len > src.len) return error.CorruptLz4;
try out.appendSlice(gpa, src[ip .. ip + lit_len]);
ip += lit_len;
if (ip >= src.len) break; // trailing literals only
if (ip + 2 > src.len) return error.CorruptLz4;
const offset = std.mem.readInt(u16, src[ip..][0..2], .little);
ip += 2;
if (offset == 0 or offset > out.items.len) return error.CorruptLz4;
var match_len: usize = token & 0x0F;
if (match_len == 15) {
while (true) {
if (ip >= src.len) return error.CorruptLz4;
const b = src[ip];
ip += 1;
match_len += b;
if (b != 255) break;
}
}
match_len += lz4_min_match;
const base = out.items.len - offset;
// Grow once, then copy in place: the source cannot alias the
// destination's old buffer across a realloc. The non-overlapping
// prefix copies in one shot; the overlapping tail (offset < match
// length) copies itself forward byte by byte.
const region = try out.addManyAsSlice(gpa, match_len);
const non_overlap = @min(match_len, offset);
@memcpy(region[0..non_overlap], out.items[base .. base + non_overlap]);
var k = non_overlap;
while (k < match_len) : (k += 1) {
region[k] = out.items[base + k];
}
}
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
@@ -249,6 +593,44 @@ pub const TmpLog = struct {
}
};
test "lz4 round-trips, including long runs and incompressible input" {
const gpa = testing.allocator;
var prng = std.Random.DefaultPrng.init(0x1A2B3C4D);
const table = try gpa.alloc(u32, lz4_table_size);
defer gpa.free(table);
var src: std.ArrayListUnmanaged(u8) = .empty;
defer src.deinit(gpa);
// Highly compressible: a repeating pattern with some noise.
for (0..100_000) |i| {
try src.append(gpa, if (i % 37 == 0) @as(u8, @intCast(prng.next() & 0xFF)) else 'a' + @as(u8, @intCast(i % 26)));
}
const worst = src.items.len + src.items.len / 255 + 16;
const dst = try gpa.alloc(u8, worst);
defer gpa.free(dst);
const n = lz4_compress(src.items, dst, table);
try testing.expect(n < src.items.len / 3); // actually compressed
var out: std.ArrayListUnmanaged(u8) = .empty;
defer out.deinit(gpa);
try lz4_decompress(gpa, dst[0..n], &out);
try testing.expectEqualSlices(u8, src.items, out.items);
// Incompressible input still round-trips (and expands).
out.clearRetainingCapacity();
src.clearRetainingCapacity();
for (0..10_000) |_| try src.append(gpa, @intCast(prng.next() & 0xFF));
const n2 = lz4_compress(src.items, dst, table);
try testing.expect(n2 >= src.items.len);
try lz4_decompress(gpa, dst[0..n2], &out);
try testing.expectEqualSlices(u8, src.items, out.items);
// Truncated input is rejected, not read out of bounds: a token
// promising four literals with only two bytes left, and a match
// sequence with no offset.
try testing.expectError(error.CorruptLz4, lz4_decompress(gpa, &[_]u8{ 0x40, 1, 2 }, &out));
try testing.expectError(error.CorruptLz4, lz4_decompress(gpa, &[_]u8{ 0x00, 0x00 }, &out));
}
test "append, replay, torn tail" {
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();
@@ -285,7 +667,6 @@ test "append, replay, torn tail" {
try log.replay(@ptrCast(&ctx), Ctx.apply);
try testing.expectEqualSlices(u8, &[_]u8{ record_type_upsert, 1, record_type_delete, 2 }, seen.items);
try testing.expectEqual(@as(u64, log.end_pos), log.end_pos);
}
test "record larger than the read chunk replays" {
@@ -299,7 +680,8 @@ test "record larger than the read chunk replays" {
var log = try Log.open(gpa, io, tmp.path);
defer log.close();
// Build a doc whose payload pushes the record past the 64 KiB stack chunk.
// Build a doc whose payload pushes the record past the block target,
// so it gets its own oversized block.
const big = try gpa.alloc(u8, 80 * 1024);
defer gpa.free(big);
@memset(big, 'x');
@@ -329,7 +711,7 @@ test "record larger than the read chunk replays" {
try testing.expectEqual(@as(usize, 1), count);
}
test "reject corrupt interior record" {
test "reject corrupt interior block" {
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();
const io = threaded.io();
@@ -342,19 +724,20 @@ test "reject corrupt interior record" {
var log = try Log.open(gpa, io, path);
const doc_bytes = [_]u8{ 0x0E, 0, 0, 0, 0x10, '_', 'i', 'd', 0, 42, 0, 0, 0, 0 };
try log.append_upsert("db", "c", &doc_bytes, 1);
try log.append_upsert("db", "c", &doc_bytes, 2); // a second block
log.close();
// Corrupt the file: flip a byte in the middle of the record. The record
// is header_len ++ "db\0" ++ "c\0" ++ doc, so this lands on the first
// doc byte — inside the hashed range, and past every framing field.
const corrupt_at = header_len + 5;
// Corrupt the FIRST block's first record: flip a byte in the hashed
// region. The first block's payload starts at the file header plus the
// block header; the record framing puts db\0coll\0 before the doc.
const corrupt_at = file_header_len + block_header_len + header_len + 5;
const dir = std.Io.Dir.cwd();
var f = try dir.openFile(io, path, .{ .mode = .read_write });
var buf: [64]u8 = undefined;
var buf: [128]u8 = undefined;
const n = try f.readPositionalAll(io, &buf, 0);
_ = n;
buf[corrupt_at] ^= 0xFF;
try f.writePositionalAll(io, buf[0..64], 0);
try f.writePositionalAll(io, buf[0..128], 0);
f.close(io);
var log2 = try Log.open(gpa, io, path);
@@ -374,3 +757,55 @@ test "reject corrupt interior record" {
var ctx = Ctx{ .count = &count, .gpa = gpa };
try testing.expectError(error.InvalidLog, log2.replay(@ptrCast(&ctx), Ctx.apply));
}
test "torn tail truncates cleanly and appends overwrite it" {
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();
const io = threaded.io();
const gpa = testing.allocator;
var tmp = try TmpLog.init(gpa);
defer tmp.deinit(gpa);
const path = tmp.path;
var log = try Log.open(gpa, io, path);
const doc_bytes = [_]u8{ 0x0E, 0, 0, 0, 0x10, '_', 'i', 'd', 0, 42, 0, 0, 0, 0 };
try log.append_upsert("db", "c", &doc_bytes, 1);
try log.append_upsert("db", "c", &doc_bytes, 2);
log.close();
// Cut the file in the middle of the second block: a crash mid-append.
const dir = std.Io.Dir.cwd();
var f = try dir.openFile(io, path, .{ .mode = .read_write });
const full_len = try f.length(io);
try f.setLength(io, full_len - 10);
f.close(io);
var log2 = try Log.open(gpa, io, path);
defer log2.close();
defer dir.deleteFile(io, path) catch {};
var seen: std.ArrayListUnmanaged(u8) = .empty;
defer seen.deinit(gpa);
const Ctx = struct {
seen: *std.ArrayListUnmanaged(u8),
gpa: std.mem.Allocator,
fn apply(ctx: *anyopaque, record: Record, doc: *bson.Document) anyerror!void {
const self: *@This() = @ptrCast(@alignCast(ctx));
try self.seen.append(self.gpa, @intCast(record.seq));
doc.deinit();
self.gpa.destroy(doc);
}
};
var ctx = Ctx{ .seen = &seen, .gpa = gpa };
try log2.replay(@ptrCast(&ctx), Ctx.apply);
try testing.expectEqualSlices(u8, &[_]u8{1}, seen.items);
// A new append overwrites from the replay end and replays cleanly.
try log2.append_upsert("db", "c", &doc_bytes, 3);
seen.clearRetainingCapacity();
var log3 = try Log.open(gpa, io, path);
defer log3.close();
ctx.seen = &seen;
try log3.replay(@ptrCast(&ctx), Ctx.apply);
try testing.expectEqualSlices(u8, &[_]u8{ 1, 3 }, seen.items);
}