Something went wrong. Try again.
in-memory redis
redis in-memory testing
Something went wrong. Try again.
30 kB · 789 lines
Zig
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790//! Persistence — save/load the whole Store to a single file.//!//! Custom byte-safe binary format. We don't need wire compatibility with//! upstream burner-redis (different language, different process), so we//! skip MessagePack and serialise to a compact format defined here. The//! benefit: no external dep, no need for a MessagePack zig library, and//! every []const u8 is preserved bit-for-bit (no UTF-8 detour).//!//! Format (all integers little-endian, lengths prefixed)://!//! header: "BRDB" (4 bytes magic) | u32 version//! then://! u64 n_entries//! for each entry://! u64 key_len, key_bytes//! u8 has_ttl (0/1); if 1 then i128 expires_at_ns//! u8 variant tag (see ValueTag below)//! (variant-specific payload)//! u64 n_scripts//! for each script://! u64 sha_len, sha_bytes//! u64 body_len, body_bytes//!//! Variant payloads://! string: u64 len, bytes//! hash: u64 n, (u64 fk, field, u64 vk, value)*//! set: u64 n, (u64 lk, bytes)*//! sorted_set: u64 n, (f64 score, u64 lk, bytes)* (sorted by (score, member))//! stream: u64 n_entries, (u64 ms, u64 seq, u64 n_fields, (u64 fk, field, u64 vk, value)*)*//! u64 last_ms, u64 last_seq//! u64 n_groups, (u64 lk, group_name, group_data)*//! group_data: u64 last_delivered_ms, u64 last_delivered_seq//! u64 n_consumers, (u64 lk, consumer_name, consumer_data)*//! consumer_data: u64 n_pending, (u64 ms, u64 seq, i128 delivered_at_ns, u64 delivery_count)*//! list: u64 n, (u64 lk, bytes)*//!//! Pub/sub state is in-memory only; subscribers and their channel/pattern//! routing aren't persisted (they're tied to live process state).//!//! Write strategy is crash-safe: write the whole blob to `<path>.tmp`,//! fsync, then rename onto `<path>`. The intermediate is removed if the//! write fails.
const std = @import("std");const Allocator = std.mem.Allocator;const store_mod = @import("store.zig");const Store = store_mod.Store;const StreamId = store_mod.StreamId;
const MAGIC: [4]u8 = .{ 'B', 'R', 'D', 'B' };const VERSION: u32 = 1;
pub const PersistenceError = error{ BadMagic, UnsupportedVersion, Truncated, InvalidVariant, /// The bytes parse, but not as a store this code could have saved. Corrupt,} || std.Io.File.OpenError || std.Io.File.Writer.Error || std.Io.File.SyncError || std.Io.Dir.RenameError || std.Io.File.StatError || Allocator.Error || std.Io.Cancelable;
const ValueTag = enum(u8) { string = 0, hash = 1, set = 2, sorted_set = 3, stream = 4, list = 5,};
// ----------------------------------------------------------------------------// Save// ----------------------------------------------------------------------------
/// Snapshot `store` to `path`. Writes to a temp file + atomic rename; safe/// to call concurrently with reads but assumes the store mutex is held/// (use `Store.save` which locks for you).pub fn save(store: *Store, path: []const u8) PersistenceError!void { var arena = std.heap.ArenaAllocator.init(store.allocator); defer arena.deinit(); const a = arena.allocator();
// Build the entire snapshot in memory, then write+rename. For docket // workloads the snapshot is tiny (kilobytes); this keeps the file // layer simple. var buf: std.ArrayList(u8) = .empty; defer buf.deinit(a);
try writeAll(&buf, a, store);
const tmp_path = try std.fmt.allocPrint(a, "{s}.tmp", .{path}); const cwd = std.Io.Dir.cwd(); errdefer cwd.deleteFile(store.io, tmp_path) catch {}; { const f = try cwd.createFile(store.io, tmp_path, .{ .truncate = true }); defer f.close(store.io); try f.writeStreamingAll(store.io, buf.items); try f.sync(store.io); } try cwd.rename(tmp_path, cwd, path, store.io);}
fn writeAll(buf: *std.ArrayList(u8), a: Allocator, store: *Store) PersistenceError!void { try buf.appendSlice(a, &MAGIC); try writeU32(buf, a, VERSION);
const now_ns = store.nowNs(); var n_live: u64 = 0; var it = store.data.iterator(); while (it.next()) |kv| { if (kv.value_ptr.isExpired(now_ns)) continue; n_live += 1; } try writeU64(buf, a, n_live);
it = store.data.iterator(); while (it.next()) |kv| { if (kv.value_ptr.isExpired(now_ns)) continue; try writeBytes(buf, a, kv.key_ptr.*); if (kv.value_ptr.expires_at_ns) |exp| { try buf.append(a, 1); try writeI128(buf, a, exp); } else { try buf.append(a, 0); } try writeValueData(buf, a, &kv.value_ptr.data); }
// SCRIPT LOAD registry — matches upstream's "scripts persisted alongside // data" behaviour so EVALSHA still finds known scripts after restore. try writeU64(buf, a, store.scripts.count()); var sit = store.scripts.iterator(); while (sit.next()) |kv| { try writeBytes(buf, a, kv.key_ptr.*); try writeBytes(buf, a, kv.value_ptr.*); }}
fn writeValueData(buf: *std.ArrayList(u8), a: Allocator, vd: *const store_mod.ValueData) PersistenceError!void { switch (vd.*) { .string => |s| { try buf.append(a, @backingInt(ValueTag.string)); try writeBytes(buf, a, s); }, .hash => |*h| { try buf.append(a, @backingInt(ValueTag.hash)); try writeU64(buf, a, h.count()); var hit = h.iterator(); while (hit.next()) |fv| { try writeBytes(buf, a, fv.key_ptr.*); try writeBytes(buf, a, fv.value_ptr.*); } }, .set => |*s| { try buf.append(a, @backingInt(ValueTag.set)); try writeU64(buf, a, s.count()); var sit = s.keyIterator(); while (sit.next()) |m| try writeBytes(buf, a, m.*); }, .sorted_set => |*z| { try buf.append(a, @backingInt(ValueTag.sorted_set)); try writeU64(buf, a, z.by_score.count()); var zit = z.by_score.iterator(); while (zit.next()) |sm| { try writeF64(buf, a, sm.score); try writeBytes(buf, a, sm.member); } }, .stream => |*s| { try buf.append(a, @backingInt(ValueTag.stream)); try writeU64(buf, a, s.entries.items.len); for (s.entries.items) |e| { try writeU64(buf, a, e.id.ms); try writeU64(buf, a, e.id.seq); try writeU64(buf, a, e.fields.count()); var fit = e.fields.iterator(); while (fit.next()) |fv| { try writeBytes(buf, a, fv.key_ptr.*); try writeBytes(buf, a, fv.value_ptr.*); } } try writeU64(buf, a, s.last_id.ms); try writeU64(buf, a, s.last_id.seq); try writeU64(buf, a, s.groups.count()); var git = s.groups.iterator(); while (git.next()) |gkv| { try writeBytes(buf, a, gkv.key_ptr.*); try writeU64(buf, a, gkv.value_ptr.last_delivered_id.ms); try writeU64(buf, a, gkv.value_ptr.last_delivered_id.seq); try writeU64(buf, a, gkv.value_ptr.consumers.count()); var cit = gkv.value_ptr.consumers.iterator(); while (cit.next()) |ckv| { try writeBytes(buf, a, ckv.key_ptr.*); try writeU64(buf, a, ckv.value_ptr.pending.count()); var pit = ckv.value_ptr.pending.iterator(); while (pit.next()) |pkv| { try writeU64(buf, a, pkv.key_ptr.ms); try writeU64(buf, a, pkv.key_ptr.seq); try writeI128(buf, a, pkv.value_ptr.delivered_at_ns); try writeU64(buf, a, pkv.value_ptr.delivery_count); } } } }, .list => |*l| { try buf.append(a, @backingInt(ValueTag.list)); const n = listLen(l); try writeU64(buf, a, n); var node = l.first; while (node) |nd| : (node = nd.next) { const ln: *store_mod.ListNode = @fieldParentPtr("node", nd); try writeBytes(buf, a, ln.data); } }, }}
fn listLen(l: *const std.DoublyLinkedList) usize { var n: usize = 0; var nd = l.first; while (nd) |x| : (nd = x.next) n += 1; return n;}
// ----------------------------------------------------------------------------// Load// ----------------------------------------------------------------------------
/// Restore a previously-saved snapshot from `path` into `store`. Returns/// false if `path` is missing (treated as "no snapshot yet"); true if/// successfully loaded. A snapshot that is refused leaves `store` as it was.pub fn load(store: *Store, path: []const u8) PersistenceError!bool { const cwd = std.Io.Dir.cwd(); const f = cwd.openFile(store.io, path, .{}) catch |err| switch (err) { error.FileNotFound => return false, else => return err, }; defer f.close(store.io);
const st = try f.stat(store.io); const bytes = try store.allocator.alloc(u8, @intCast(st.size)); defer store.allocator.free(bytes); var reader_buf: [4096]u8 = undefined; var reader = f.reader(store.io, &reader_buf); reader.interface.readSliceAll(bytes) catch |err| switch (err) { error.EndOfStream, error.ReadFailed => return PersistenceError.Truncated, };
try restore(store, bytes); return true;}
const Keyspace = store_mod.BytesHashMap(store_mod.ValueEntry);const Scripts = std.StringHashMap([]const u8);
fn freeKeyspace(store: *Store, data: *Keyspace) void { var it = data.iterator(); while (it.next()) |kv| { store.allocator.free(kv.key_ptr.*); store.freeValueData(&kv.value_ptr.data); } data.deinit();}
fn freeScripts(store: *Store, scripts: *Scripts) void { var it = scripts.iterator(); while (it.next()) |kv| { store.allocator.free(kv.key_ptr.*); store.allocator.free(kv.value_ptr.*); } scripts.deinit();}
/// Replace what `store` holds with the snapshot in `bytes`. The snapshot is/// read whole before the store is touched, and it is refused if it is cut/// short or does not describe a store `save` could have written: nothing/// else stands between a damaged file and the store's own assumptions.fn restore(store: *Store, bytes: []const u8) PersistenceError!void { var p = Parser{ .buf = bytes, .pos = 0 };
var magic: [4]u8 = undefined; try p.readInto(&magic); if (!std.mem.eql(u8, &magic, &MAGIC)) return PersistenceError.BadMagic; const ver = try p.readU32(); if (ver != VERSION) return PersistenceError.UnsupportedVersion;
var data = Keyspace.init(store.allocator); errdefer freeKeyspace(store, &data); const n_entries = try p.readU64(); var i: u64 = 0; while (i < n_entries) : (i += 1) { const key = try p.readBytesDuped(store.allocator); errdefer store.allocator.free(key);
const expires_at_ns: ?i128 = switch (try p.readU8()) { 0 => null, 1 => try p.readI128(), else => return PersistenceError.Corrupt, };
const tag_raw = try p.readU8(); if (tag_raw > @backingInt(ValueTag.list)) return PersistenceError.InvalidVariant; const tag: ValueTag = @fromBackingInt(@intCast(tag_raw)); var value = try readValueData(&p, store, tag); errdefer store.freeValueData(&value);
const slot = try data.getOrPut(key); if (slot.found_existing) return PersistenceError.Corrupt; slot.value_ptr.* = .{ .data = value, .expires_at_ns = expires_at_ns }; }
// If the file ends here (older format that didn't write scripts), tolerate it. var scripts = Scripts.init(store.allocator); errdefer freeScripts(store, &scripts); if (p.pos < p.buf.len) { const n_scripts = try p.readU64(); var j: u64 = 0; while (j < n_scripts) : (j += 1) { const sha = try p.readBytesDuped(store.allocator); errdefer store.allocator.free(sha); const body = try p.readBytesDuped(store.allocator); errdefer store.allocator.free(body); const slot = try scripts.getOrPut(sha); if (slot.found_existing) return PersistenceError.Corrupt; slot.value_ptr.* = body; } }
store.clearZsetCache(); freeKeyspace(store, &store.data); freeScripts(store, &store.scripts); store.data = data; store.scripts = scripts;}
const Parser = struct { buf: []const u8, pos: usize,
fn need(self: *Parser, n: u64) PersistenceError!void { // `pos + n` would wrap for a length a damaged file can hold if (n > self.buf.len - self.pos) return PersistenceError.Truncated; }
/// A count of items that each take at least `item_bytes` of what is left. fn readCount(self: *Parser, item_bytes: usize) PersistenceError!usize { const n = try self.readU64(); if (n > (self.buf.len - self.pos) / item_bytes) return PersistenceError.Truncated; return @intCast(n); }
fn readInto(self: *Parser, dst: []u8) PersistenceError!void { try self.need(dst.len); @memcpy(dst, self.buf[self.pos..][0..dst.len]); self.pos += dst.len; }
fn readU8(self: *Parser) PersistenceError!u8 { try self.need(1); const v = self.buf[self.pos]; self.pos += 1; return v; }
fn readU32(self: *Parser) PersistenceError!u32 { try self.need(4); const v = std.mem.readInt(u32, self.buf[self.pos..][0..4], .little); self.pos += 4; return v; }
fn readU64(self: *Parser) PersistenceError!u64 { try self.need(8); const v = std.mem.readInt(u64, self.buf[self.pos..][0..8], .little); self.pos += 8; return v; }
fn readI128(self: *Parser) PersistenceError!i128 { try self.need(16); const v = std.mem.readInt(i128, self.buf[self.pos..][0..16], .little); self.pos += 16; return v; }
fn readF64(self: *Parser) PersistenceError!f64 { const bits = try self.readU64(); return @bitCast(bits); }
fn readBytesDuped(self: *Parser, alloc: Allocator) PersistenceError![]u8 { const len64 = try self.readU64(); try self.need(len64); const len: usize = @intCast(len64); const out = try alloc.dupe(u8, self.buf[self.pos..][0..len]); self.pos += len; return out; }};
fn freeFields(alloc: Allocator, fields: *store_mod.StreamFields) void { var it = fields.iterator(); while (it.next()) |kv| { alloc.free(kv.key_ptr.*); alloc.free(kv.value_ptr.*); } fields.deinit();}
/// Field-value pairs, as a hash and a stream entry both hold them. No field twice.fn readFields(p: *Parser, alloc: Allocator) PersistenceError!store_mod.StreamFields { const n = try p.readCount(16); var fields = store_mod.StreamFields.init(alloc); errdefer freeFields(alloc, &fields); for (0..n) |_| { const f = try p.readBytesDuped(alloc); errdefer alloc.free(f); const v = try p.readBytesDuped(alloc); errdefer alloc.free(v); const slot = try fields.getOrPut(f); if (slot.found_existing) return PersistenceError.Corrupt; slot.value_ptr.* = v; } return fields;}
fn readStreamId(p: *Parser) PersistenceError!StreamId { const ms = try p.readU64(); return .{ .ms = ms, .seq = try p.readU64() };}
fn readValueData(p: *Parser, store: *Store, tag: ValueTag) PersistenceError!store_mod.ValueData { const alloc = store.allocator; return switch (tag) { .string => .{ .string = try p.readBytesDuped(alloc) }, .hash => blk: { var h = try readFields(p, alloc); errdefer freeFields(alloc, &h); // the store deletes a key when its last field goes if (h.count() == 0) return PersistenceError.Corrupt; break :blk .{ .hash = h }; }, .set => blk: { const n = try p.readCount(8); if (n == 0) return PersistenceError.Corrupt; var s = store_mod.BytesHashSet.init(alloc); errdefer { var it = s.keyIterator(); while (it.next()) |k| alloc.free(k.*); s.deinit(); } for (0..n) |_| { const m = try p.readBytesDuped(alloc); errdefer alloc.free(m); const slot = try s.getOrPut(m); if (slot.found_existing) return PersistenceError.Corrupt; } break :blk .{ .set = s }; }, .sorted_set => blk: { const n = try p.readCount(16); if (n == 0) return PersistenceError.Corrupt; var z = store_mod.SortedSet.init(alloc); errdefer z.deinit(alloc); try z.by_score.ensureTotalCapacity(alloc, n); var before: ?store_mod.ScoredMember = null; for (0..n) |_| { const score = try p.readF64(); const member = try p.readBytesDuped(z.memberAllocator()); const member_hash = store_mod.zset_index.hash(member); const entry: store_mod.ScoredMember = .{ .score = score, .member = member, .member_hash = member_hash }; // saved in (score, member) order, each member once. NaN fails the first test. if (before) |b| { if (!store_mod.ScoredMember.lessThan({}, b, entry)) return PersistenceError.Corrupt; } else if (std.math.isNan(score)) return PersistenceError.Corrupt; if (z.by_member.get(member) != null) return PersistenceError.Corrupt; try z.by_member.put(.{ .hash = member_hash, .bytes = member }, score); try z.by_score.appendAssumeCapacity(alloc, entry); before = entry; } break :blk .{ .sorted_set = z }; }, .stream => blk: { var s = store_mod.Stream.init(alloc); errdefer s.deinit(alloc);
const n_entries = try p.readCount(24); var before: ?StreamId = null; for (0..n_entries) |_| { const id = try readStreamId(p); if (before) |b| if (b.cmp(id) != .lt) return PersistenceError.Corrupt; before = id; var fields = try readFields(p, alloc); errdefer freeFields(alloc, &fields); try s.entries.append(alloc, .{ .id = id, .fields = fields }); } s.last_id = try readStreamId(p); if (before) |last| if (s.last_id.cmp(last) == .lt) return PersistenceError.Corrupt;
const n_groups = try p.readCount(32); for (0..n_groups) |_| { const gname = try p.readBytesDuped(alloc); errdefer alloc.free(gname); var cg = store_mod.ConsumerGroup.init(alloc); errdefer cg.deinit(alloc); cg.last_delivered_id = try readStreamId(p);
const n_cons = try p.readCount(16); for (0..n_cons) |_| { const cname = try p.readBytesDuped(alloc); errdefer alloc.free(cname); var cons = store_mod.Consumer.init(alloc); errdefer cons.deinit(); const n_pending = try p.readCount(40); for (0..n_pending) |_| { const id = try readStreamId(p); const at = try p.readI128(); const dc = try p.readU64(); const slot = try cons.pending.getOrPut(id); if (slot.found_existing) return PersistenceError.Corrupt; slot.value_ptr.* = .{ .delivered_at_ns = at, .delivery_count = dc }; } const slot = try cg.consumers.getOrPut(cname); if (slot.found_existing) return PersistenceError.Corrupt; slot.value_ptr.* = cons; } const slot = try s.groups.getOrPut(gname); if (slot.found_existing) return PersistenceError.Corrupt; slot.value_ptr.* = cg; } break :blk .{ .stream = s }; }, .list => blk: { const n = try p.readCount(8); if (n == 0) return PersistenceError.Corrupt; var list: std.DoublyLinkedList = .{}; errdefer { while (list.popFirst()) |nd| { const ln: *store_mod.ListNode = @fieldParentPtr("node", nd); alloc.free(ln.data); alloc.destroy(ln); } } for (0..n) |_| { const item = try p.readBytesDuped(alloc); errdefer alloc.free(item); const ln = try alloc.create(store_mod.ListNode); ln.* = .{ .data = item }; list.append(&ln.node); } break :blk .{ .list = list }; }, };}
// ----------------------------------------------------------------------------// Buffer write helpers// ----------------------------------------------------------------------------
fn writeU32(buf: *std.ArrayList(u8), a: Allocator, v: u32) PersistenceError!void { var b: [4]u8 = undefined; std.mem.writeInt(u32, &b, v, .little); try buf.appendSlice(a, &b);}
fn writeU64(buf: *std.ArrayList(u8), a: Allocator, v: u64) PersistenceError!void { var b: [8]u8 = undefined; std.mem.writeInt(u64, &b, v, .little); try buf.appendSlice(a, &b);}
fn writeI128(buf: *std.ArrayList(u8), a: Allocator, v: i128) PersistenceError!void { var b: [16]u8 = undefined; std.mem.writeInt(i128, &b, v, .little); try buf.appendSlice(a, &b);}
fn writeF64(buf: *std.ArrayList(u8), a: Allocator, v: f64) PersistenceError!void { try writeU64(buf, a, @bitCast(v));}
fn writeBytes(buf: *std.ArrayList(u8), a: Allocator, bytes: []const u8) PersistenceError!void { try writeU64(buf, a, bytes.len); try buf.appendSlice(a, bytes);}
// ----------------------------------------------------------------------------// Tests// ----------------------------------------------------------------------------
test "save + load roundtrips a mix of value variants" { var threaded = std.Io.Threaded.init(std.testing.allocator, .{}); defer threaded.deinit();
var path_buf: [128]u8 = undefined; const ns: i128 = std.Io.Timestamp.now(threaded.io(), .real).nanoseconds; const path = try std.fmt.bufPrint(&path_buf, "/tmp/burner-test-{x}.brdb", .{ns}); defer std.Io.Dir.cwd().deleteFile(threaded.io(), path) catch {};
// Original store: hand-crafted entries across every variant. { var s = Store.init(std.testing.allocator, threaded.io()); defer s.deinit();
_ = try s.set("a", "1", .{}); _ = try s.hset("h", &.{ "f1", "v1", "f2", "v2" }); _ = try s.sadd("st", &.{ "m1", "m2" }); _ = try s.zadd("z", &.{ .{ .score = 1, .member = "x" }, .{ .score = 2, .member = "y" }, }, .{}); _ = try s.rpush("l", &.{ "a", "b", "c" }); _ = try s.xadd("xs", .{ .ms = 1, .seq = 0 }, &.{ "k", "v" }); try s.xgroupCreate("xs", "g1", .{ .id = StreamId.zero() }); const got = try s.xreadgroupOne(std.testing.allocator, "xs", "g1", "c1", .new_messages, 10); defer store_mod.freeDeliveredEntries(std.testing.allocator, got);
try save(&s, path); }
// Restore into a fresh store; check that every variant came back. var s2 = Store.init(std.testing.allocator, threaded.io()); defer s2.deinit(); const loaded = try load(&s2, path); try std.testing.expect(loaded);
const a_v = try s2.get(std.testing.allocator, "a"); defer if (a_v) |x| std.testing.allocator.free(x); try std.testing.expectEqualStrings("1", a_v.?);
const hpairs = try s2.hgetall(std.testing.allocator, "h"); defer Store.freeHashPairs(std.testing.allocator, hpairs); try std.testing.expectEqual(@as(usize, 2), hpairs.len);
try std.testing.expectEqual(@as(i64, 2), try s2.scard("st")); try std.testing.expectEqual(@as(i64, 2), try s2.zcard("z")); try std.testing.expectEqual(@as(i64, 3), try s2.llen("l")); try std.testing.expectEqual(@as(i64, 1), try s2.xlen("xs"));
// The consumer group and its pending entry should have persisted too. const groups = try s2.xinfoGroups(std.testing.allocator, "xs"); defer Store.freeGroupInfos(std.testing.allocator, groups); try std.testing.expectEqual(@as(usize, 1), groups.len); try std.testing.expectEqual(@as(i64, 1), groups[0].pending);}
test "load returns false when the file doesn't exist" { var threaded = std.Io.Threaded.init(std.testing.allocator, .{}); defer threaded.deinit(); var s = Store.init(std.testing.allocator, threaded.io()); defer s.deinit();
const present = try load(&s, "/tmp/burner-does-not-exist.brdb"); try std.testing.expect(!present);}
/// a snapshot of one key, for the tests below to damageconst TestSnapshot = struct { buf: std.ArrayList(u8) = .empty,
const a = std.testing.allocator;
fn begin(key: []const u8, tag: ValueTag) !TestSnapshot { var snap: TestSnapshot = .{}; errdefer snap.buf.deinit(a); try snap.buf.appendSlice(a, &MAGIC); try writeU32(&snap.buf, a, VERSION); try writeU64(&snap.buf, a, 1); try writeBytes(&snap.buf, a, key); try snap.buf.append(a, 0); try snap.buf.append(a, @backingInt(tag)); return snap; }
fn int(self: *TestSnapshot, v: u64) !void { try writeU64(&self.buf, a, v); }
fn bytes(self: *TestSnapshot, v: []const u8) !void { try writeBytes(&self.buf, a, v); }
/// what restoring it into a store that already holds `kept` does fn restoreOver(self: *TestSnapshot) !PersistenceError!void { defer self.buf.deinit(a); var threaded = std.Io.Threaded.init(a, .{}); defer threaded.deinit(); var s = Store.init(a, threaded.io()); defer s.deinit(); _ = try s.set("kept", "v", .{}); const result = restore(&s, self.buf.items); // a snapshot that is refused leaves the store as it was if (result) |_| {} else |_| try std.testing.expectEqual(@as(i64, 1), s.exists(&.{"kept"})); return result; }};
test "a length too large to add to the position is a truncated snapshot, not a read past the end" { var snap = try TestSnapshot.begin("k", .string); try snap.int(std.math.maxInt(u64)); try std.testing.expectError(PersistenceError.Truncated, try snap.restoreOver());
var counted = try TestSnapshot.begin("k", .sorted_set); try counted.int(std.math.maxInt(u64)); try std.testing.expectError(PersistenceError.Truncated, try counted.restoreOver());}
test "a stream whose ids do not ascend, or whose last id is below an entry, is refused" { var unordered = try TestSnapshot.begin("k", .stream); try unordered.int(2); for ([_]u64{ 2, 1 }) |ms| { try unordered.int(ms); try unordered.int(0); try unordered.int(0); } try unordered.int(2); try unordered.int(0); try unordered.int(0); try std.testing.expectError(PersistenceError.Corrupt, try unordered.restoreOver());
var behind = try TestSnapshot.begin("k", .stream); try behind.int(1); try behind.int(5); try behind.int(0); try behind.int(0); try behind.int(4); try behind.int(0); try behind.int(0); try std.testing.expectError(PersistenceError.Corrupt, try behind.restoreOver());}
test "a sorted set that is empty, out of order, or holds a member twice is refused" { var empty = try TestSnapshot.begin("k", .sorted_set); try empty.int(0); try empty.int(0); try std.testing.expectError(PersistenceError.Corrupt, try empty.restoreOver());
for ([_][2]struct { f64, []const u8 }{ .{ .{ 2, "a" }, .{ 1, "b" } }, .{ .{ 1, "a" }, .{ 2, "a" } }, .{ .{ std.math.nan(f64), "a" }, .{ 2, "b" } }, }) |pair| { var snap = try TestSnapshot.begin("k", .sorted_set); try snap.int(2); for (pair) |entry| { try snap.int(@bitCast(entry[0])); try snap.bytes(entry[1]); } try snap.int(0); try std.testing.expectError(PersistenceError.Corrupt, try snap.restoreOver()); }}
test "a hash that is empty or holds a field twice is refused" { var empty = try TestSnapshot.begin("k", .hash); try empty.int(0); try empty.int(0); try std.testing.expectError(PersistenceError.Corrupt, try empty.restoreOver());
var twice = try TestSnapshot.begin("k", .hash); try twice.int(2); for (0..2) |_| { try twice.bytes("f"); try twice.bytes("v"); } try twice.int(0); try std.testing.expectError(PersistenceError.Corrupt, try twice.restoreOver());}
test "a save that fails leaves no temporary file" { var threaded = std.Io.Threaded.init(std.testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var s = Store.init(std.testing.allocator, io); defer s.deinit(); _ = try s.set("a", "1", .{});
// a directory where the snapshot should go: the rename onto it fails var path_buf: [128]u8 = undefined; const path = try std.fmt.bufPrint(&path_buf, "/tmp/burner-test-dir-{x}", .{std.Io.Timestamp.now(io, .real).nanoseconds}); const cwd = std.Io.Dir.cwd(); try cwd.createDir(io, path, .default_dir); defer cwd.deleteDir(io, path) catch {};
try std.testing.expect(std.meta.isError(save(&s, path))); var tmp_buf: [140]u8 = undefined; const tmp = try std.fmt.bufPrint(&tmp_buf, "{s}.tmp", .{path}); try std.testing.expectError(error.FileNotFound, cwd.access(io, tmp, .{}));}