//! `zit daemon` over a real socket: the binary is spawned, a client connects
//! through `std.Io.net.UnixAddress`, and every assertion is on the bytes the
//! daemon wrote back or on the store as the CLI then reads it.
//!
//! Sockets are named relative to the temporary directory (the daemon's cwd)
//! and, from this process, relative to its own cwd: the absolute path of a
//! test directory can be longer than `sockaddr_un` holds.
//!
//! Imported by the e2e root, whose step builds the `zit` binary this spawns.
const std = @import("std");
const repeat = @import("zit").repeat;
const zit = @import("zit");
const build_options = @import("build_options");
const protocol = zit.daemon_protocol;
const net = std.Io.net;
const zit_exe: []const u8 = build_options.zit_exe;
const epoch0: i64 = 1_800_000_000;
const Fx = struct {
gpa: std.mem.Allocator,
io: std.Io,
tmp: std.testing.TmpDir,
arena_state: std.heap.ArenaAllocator,
fn init(fx: *Fx) void {
fx.* = .{
.gpa = std.testing.allocator,
.io = std.testing.io,
.tmp = std.testing.tmpDir(.{}),
.arena_state = .init(std.testing.allocator),
};
}
fn deinit(fx: *Fx) void {
fx.arena_state.deinit();
fx.tmp.cleanup();
}
fn a(fx: *Fx) std.mem.Allocator {
return fx.arena_state.allocator();
}
/// `name` (relative to the temporary directory) as this process names it.
fn here(fx: *Fx, name: []const u8) []const u8 {
return std.fmt.allocPrint(fx.a(), ".zig-cache/tmp/{s}/{s}", .{ fx.tmp.sub_path, name }) catch @panic("oom");
}
fn makeStore(fx: *Fx, name: []const u8) !void {
try fx.tmp.dir.createDir(fx.io, name, .default_dir);
var dir = try fx.tmp.dir.openDir(fx.io, name, .{});
defer dir.close(fx.io);
const res = try runZit(fx, dir, &.{ "init", "-q" }, null);
try expectExit(res, 0);
}
fn storeBytes(fx: *Fx, store: []const u8) ![]u8 {
const path = try std.fmt.allocPrint(fx.a(), "{s}/.zit", .{store});
return fx.tmp.dir.readFileAlloc(fx.io, path, fx.a(), .limited(1 << 30));
}
fn storeSize(fx: *Fx, store: []const u8) !u64 {
const path = try std.fmt.allocPrint(fx.a(), "{s}/.zit", .{store});
return (try fx.tmp.dir.statFile(fx.io, path, .{})).size;
}
/// Copy only the store file, so the copy shares no cache with the source.
fn copyStore(fx: *Fx, from: []const u8, to: []const u8) !void {
try fx.tmp.dir.createDir(fx.io, to, .default_dir);
const bytes = try fx.storeBytes(from);
const path = try std.fmt.allocPrint(fx.a(), "{s}/.zit", .{to});
try fx.tmp.dir.writeFile(fx.io, .{ .sub_path = path, .data = bytes });
}
/// `zit log -R STORE trunk`: the check-ins on trunk, newest first.
fn trunkLog(fx: *Fx, store: []const u8) ![]const []const u8 {
const res = try runZit(fx, fx.tmp.dir, &.{ "log", "-R", store, "trunk" }, null);
try expectExit(res, 0);
var out: std.ArrayList([]const u8) = .empty;
var lines = std.mem.tokenizeScalar(u8, res.stdout, '\n');
while (lines.next()) |line| if (std.mem.startsWith(u8, line, "checkin ")) try out.append(fx.a(), line["checkin ".len..]);
return out.items;
}
};
fn runZit(fx: *Fx, dir: std.Io.Dir, tail: []const []const u8, stdin_file: ?[]const u8) !std.process.RunResult {
var env = std.process.Environ.Map.init(fx.gpa);
defer env.deinit();
try env.put("TZ", "UTC");
try env.put("HOME", "/zit-e2e-isolated-home-does-not-exist");
try env.put("PATH", "/usr/bin:/bin");
const exe = try absExe(fx);
var cmd: std.ArrayList(u8) = .empty;
try cmd.print(fx.a(), "exec '{s}'", .{exe});
for (tail) |t| try cmd.print(fx.a(), " '{s}'", .{t});
if (stdin_file) |f| try cmd.print(fx.a(), " < '{s}'", .{f}) else try cmd.appendSlice(fx.a(), " < /dev/null");
const res = try std.process.run(fx.gpa, fx.io, .{
.argv = &.{ "/bin/sh", "-c", cmd.items },
.cwd = .{ .dir = dir },
.environ_map = &env,
});
// Kept on the arena so a caller need not free them.
const stdout = try fx.a().dupe(u8, res.stdout);
const stderr = try fx.a().dupe(u8, res.stderr);
fx.gpa.free(res.stdout);
fx.gpa.free(res.stderr);
return .{ .term = res.term, .stdout = stdout, .stderr = stderr };
}
fn absExe(fx: *Fx) ![]const u8 {
const cwd_abs = try std.process.currentPathAlloc(fx.io, fx.a());
return std.fs.path.resolve(fx.a(), &.{ cwd_abs, zit_exe });
}
fn expectExit(res: std.process.RunResult, want: u8) !void {
switch (res.term) {
.exited => |code| if (code == want) return,
else => {},
}
std.debug.print("\nzit: expected exit {d}, got {any}:\n{s}{s}\n", .{ want, res.term, res.stdout, res.stderr });
return error.TestUnexpectedResult;
}
const StartOptions = struct {
extra: []const []const u8 = &.{},
env: []const [2][]const u8 = &.{},
stdin_pipe: bool = false,
};
/// A running `zit daemon`, ready once its socket path came back on stdout.
const Daemon = struct {
child: std.process.Child,
fn start(fx: *Fx, store: []const u8, socket: []const u8, opts: StartOptions) !Daemon {
var env = std.process.Environ.Map.init(fx.gpa);
defer env.deinit();
try env.put("TZ", "UTC");
try env.put("HOME", "/zit-e2e-isolated-home-does-not-exist");
try env.put("PATH", "/usr/bin:/bin");
for (opts.env) |pair| try env.put(pair[0], pair[1]);
var argv: std.ArrayList([]const u8) = .empty;
try argv.appendSlice(fx.a(), &.{ try absExe(fx), "daemon", "-R", store, "--socket", socket });
try argv.appendSlice(fx.a(), opts.extra);
// Kept beside the store, so a failure can show what the daemon said.
const err_file = try fx.tmp.dir.createFile(fx.io, "daemon.err", .{ .truncate = false });
defer err_file.close(fx.io);
var child = try std.process.spawn(fx.io, .{
.argv = argv.items,
.cwd = .{ .dir = fx.tmp.dir },
.environ_map = &env,
.stdin = if (opts.stdin_pipe) .pipe else .ignore,
.stdout = .pipe,
.stderr = .{ .file = err_file },
});
errdefer child.kill(fx.io);
var buf: [512]u8 = undefined;
var r = child.stdout.?.readerStreaming(fx.io, &buf);
const line = r.interface.takeDelimiterInclusive('\n') catch return error.DaemonNeverListened;
try std.testing.expectEqualStrings(socket, line[0 .. line.len - 1]);
return .{ .child = child };
}
/// SIGTERM, then the exit status.
fn stop(d: *Daemon, io: std.Io) !std.process.Child.Term {
try std.posix.kill(d.child.id.?, .TERM);
return d.child.wait(io);
}
fn expectExited(fx: *Fx, term: std.process.Child.Term, code: u8) !void {
switch (term) {
.exited => |c| if (c == code) return,
else => {},
}
const said = fx.tmp.dir.readFileAlloc(fx.io, "daemon.err", fx.a(), .limited(1 << 20)) catch "";
std.debug.print("\ndaemon: expected exit {d}, got {any}; it said:\n{s}\n", .{ code, term, said });
return error.TestUnexpectedResult;
}
};
const Client = struct {
stream: net.Stream,
read_buf: []u8,
reader: net.Stream.Reader,
write_buf: [1 << 16]u8,
writer: net.Stream.Writer,
closed: bool,
fn open(fx: *Fx, socket: []const u8) !*Client {
const addr = try net.UnixAddress.init(fx.here(socket));
const c = try fx.a().create(Client);
c.stream = try addr.connect(fx.io);
c.read_buf = try fx.a().alloc(u8, 4 << 20);
c.reader = c.stream.reader(fx.io, c.read_buf);
c.writer = c.stream.writer(fx.io, &c.write_buf);
c.closed = false;
return c;
}
fn close(c: *Client, io: std.Io) void {
if (c.closed) return;
c.stream.close(io);
c.closed = true;
}
fn send(c: *Client, req: protocol.Request, trailers: []const []const u8) !void {
try c.queue(req, trailers);
try c.writer.interface.flush();
}
/// Written but not flushed, for pipelining several requests at once.
fn queue(c: *Client, req: protocol.Request, trailers: []const []const u8) !void {
try protocol.writeRequest(&c.writer.interface, req);
for (trailers) |t| try c.writer.interface.writeAll(t);
}
fn reply(c: *Client, arena: std.mem.Allocator) !std.json.Value {
const line = try c.reader.interface.takeDelimiterInclusive('\n');
return std.json.parseFromSliceLeaky(std.json.Value, arena, try arena.dupe(u8, line), .{});
}
fn take(c: *Client, arena: std.mem.Allocator, n: usize) ![]u8 {
const bytes = try arena.alloc(u8, n);
try c.reader.interface.readSliceAll(bytes);
return bytes;
}
fn call(c: *Client, arena: std.mem.Allocator, req: protocol.Request, trailers: []const []const u8) !std.json.Value {
try c.send(req, trailers);
const v = try c.reply(arena);
try std.testing.expectEqual(req.id.?, v.object.get("id").?.integer);
return v;
}
};
fn get(v: std.json.Value, key: []const u8) std.json.Value {
return v.object.get(key) orelse std.debug.panic("reply has no '{s}'", .{key});
}
fn str(v: std.json.Value, key: []const u8) []const u8 {
return get(v, key).string;
}
fn expectOk(v: std.json.Value) !void {
if (get(v, "ok").bool) return;
std.debug.print("\nrefused: {s}: {s}\n", .{ str(get(v, "error"), "code"), str(get(v, "error"), "message") });
return error.TestUnexpectedResult;
}
fn expectError(v: std.json.Value, code: []const u8, exit: i64) !void {
try std.testing.expect(!get(v, "ok").bool);
try std.testing.expectEqualStrings(code, str(get(v, "error"), "code"));
try std.testing.expectEqual(exit, get(get(v, "error"), "exit").integer);
}
fn put(name: []const u8, bytes: []const u8) protocol.Edit {
return .{ .put = .{ .name = name, .size = bytes.len } };
}
fn commit(key: []const u8, parent: ?[]const u8, comment: []const u8, sec: i64, edits: []const protocol.Edit) protocol.Commit {
return .{ .key = key, .parent = parent, .user = "e", .comment = comment, .time = .{ .epoch = epoch0 + sec, .ms = 7 }, .edits = edits };
}
fn sha3Hex(bytes: []const u8) [64]u8 {
var d: [32]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(bytes, &d, .{});
return std.fmt.bytesToHex(d, .lower);
}
test "daemon: hello, commit, a resend, pipelining, refusals, get, and the socket's guards" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
try fx.makeStore("store");
var d = try Daemon.start(&fx, "store", "s/d.sock", .{});
var running = true;
defer if (running) d.child.kill(fx.io);
// Socket 0600 in a 0700 directory.
const sock_st = try fx.tmp.dir.statFile(fx.io, "s/d.sock", .{});
try std.testing.expectEqual(@as(std.posix.mode_t, 0o600), sock_st.permissions.toMode() & 0o777);
const dir_st = try fx.tmp.dir.statFile(fx.io, "s", .{});
try std.testing.expectEqual(@as(std.posix.mode_t, 0o700), dir_st.permissions.toMode() & 0o777);
// A second daemon on the same store is refused, naming the first.
const second = try runZit(&fx, fx.tmp.dir, &.{ "daemon", "-R", "store", "--socket", "s/other.sock" }, null);
try expectExit(second, 1);
const holder = try std.fmt.allocPrint(a, "another daemon (process {d})", .{d.child.id.?});
try std.testing.expect(std.mem.indexOf(u8, second.stderr, holder) != null);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const hello = try c.call(a, .{ .id = 1, .op = .{ .hello = protocol.version } }, &.{});
try expectOk(hello);
try std.testing.expectEqual(protocol.version, get(hello, "protocol").integer);
try std.testing.expect(std.mem.endsWith(u8, str(hello, "store"), "store"));
try std.testing.expectEqual(@as(usize, 0), get(get(hello, "tip"), "branches").array.items.len);
// A commit, then the same commit again on the same parent.
const seed = try c.call(a, .{ .id = 2, .op = .{ .commit = commit("seed", null, "seed", 0, &.{put("a.txt", "alpha\n")}) } }, &.{"alpha\n"});
try expectOk(seed);
try std.testing.expectEqualStrings("recorded", str(seed, "outcome"));
try std.testing.expectEqualStrings("seed", str(seed, "key"));
try std.testing.expect(get(seed, "parent") == .null);
try std.testing.expectEqualStrings("trunk", str(seed, "branch"));
const p0 = str(seed, "hash");
const edits1 = [_]protocol.Edit{put("b.txt", "beta\n")};
const first = try c.call(a, .{ .id = 3, .op = .{ .commit = commit("k1", p0, "one", 1, &edits1) } }, &.{"beta\n"});
try expectOk(first);
try std.testing.expectEqualStrings("recorded", str(first, "outcome"));
try std.testing.expectEqualStrings(p0, str(first, "parent"));
try std.testing.expect(!get(first, "forked").bool);
const size_before = try fx.storeSize("store");
const again = try c.call(a, .{ .id = 4, .op = .{ .commit = commit("k1", p0, "one", 1, &edits1) } }, &.{"beta\n"});
try expectOk(again);
try std.testing.expectEqualStrings("existing", str(again, "outcome"));
try std.testing.expectEqualStrings(str(first, "hash"), str(again, "hash"));
try std.testing.expect(!get(again, "forked").bool);
try std.testing.expectEqual(size_before, try fx.storeSize("store"));
// Pipelined: three commits written before any reply is read, answered
// in order, each on the last.
var tip = try a.dupe(u8, str(first, "hash"));
{
const names = [_][]const u8{ "p1.txt", "p2.txt", "p3.txt" };
for (names, 0..) |n, i| try c.queue(.{ .id = @intCast(10 + i), .op = .{ .commit = commit(n, null, n, @intCast(10 + i), &.{put(n, n)}) } }, &.{n});
try c.writer.interface.flush();
for (0..names.len) |i| {
const v = try c.reply(a);
try std.testing.expectEqual(@as(i64, @intCast(10 + i)), get(v, "id").integer);
try expectOk(v);
try std.testing.expectEqualStrings(names[i], str(v, "key"));
try std.testing.expectEqualStrings(tip, str(v, "parent"));
tip = try a.dupe(u8, str(v, "hash"));
}
}
// The puts' hash is checked before anything is committed.
{
const bad = protocol.Edit{ .put = .{ .name = "h.txt", .size = 4, .hash = repeat("0", 64) } };
const v = try c.call(a, .{ .id = 20, .op = .{ .commit = commit("h", null, "h", 20, &.{bad}) } }, &.{"oops"});
try expectError(v, "hash_mismatch", 2);
const good_hash = sha3Hex("good");
const good = protocol.Edit{ .put = .{ .name = "h.txt", .size = 4, .hash = &good_hash } };
const ok = try c.call(a, .{ .id = 21, .op = .{ .commit = commit("h", null, "h", 21, &.{good}) } }, &.{"good"});
try expectOk(ok);
tip = try a.dupe(u8, str(ok, "hash"));
}
// The core's refusals come back by name.
{
const v = try c.call(a, .{ .id = 22, .op = .{ .commit = commit("x", null, "x", 22, &.{.{ .delete = "never.txt" }}) } }, &.{});
try expectError(v, "path_conflict", 1);
const stale = try c.call(a, .{ .id = 23, .op = .{ .commit = commit("x", p0, "x", 23, &.{put("z.txt", "z")}) } }, &.{"z"});
try expectError(stale, "not_tip", 1);
}
// get, one and several.
{
const alpha = sha3Hex("alpha\n");
const one = try c.call(a, .{ .id = 30, .op = .{ .get = .{ .one = &alpha } } }, &.{});
try expectOk(one);
try std.testing.expectEqual(@as(i64, 6), get(one, "size").integer);
try std.testing.expectEqualStrings("alpha\n", try c.take(a, 6));
const beta = sha3Hex("beta\n");
const nowhere = repeat("e", 64);
const many = try c.call(a, .{ .id = 31, .op = .{ .get = .{ .many = &.{ &beta, nowhere, &alpha } } } }, &.{});
try expectOk(many);
const items = get(many, "items").array.items;
try std.testing.expectEqual(@as(i64, 5), get(items[0], "size").integer);
try std.testing.expectEqualStrings("not_found", str(get(items[1], "error"), "code"));
try std.testing.expectEqual(@as(i64, 6), get(items[2], "size").integer);
try std.testing.expectEqualStrings("beta\nalpha\n", try c.take(a, 11));
const malformed = try c.call(a, .{ .id = 32, .op = .{ .get = .{ .one = "ABC" } } }, &.{});
try expectError(malformed, "bad_request", 2);
}
// A refused request leaves the connection usable; status still answers.
{
try c.writer.interface.writeAll("{\"id\":40,\"op\":\"rebase\"}\n");
try c.writer.interface.flush();
try expectError(try c.reply(a), "bad_request", 2);
const st = try c.call(a, .{ .id = 41, .op = .status }, &.{});
try expectOk(st);
const branches = get(st, "branches").array.items;
try std.testing.expectEqual(@as(usize, 1), branches.len);
try std.testing.expectEqualStrings("trunk", str(branches[0], "name"));
try std.testing.expectEqualStrings(tip, str(branches[0], "tip"));
}
// Unreadable framing closes the connection after the refusal.
{
try c.writer.interface.writeAll("not json\n");
try c.writer.interface.flush();
const v = try c.reply(a);
try expectError(v, "bad_request", 2);
try std.testing.expect(get(v, "id") == .null);
try std.testing.expectError(error.EndOfStream, c.reader.interface.takeByte());
}
running = false;
try Daemon.expectExited(&fx, try d.stop(fx.io), 0);
// Stopped: the socket is gone and the lock released.
try std.testing.expectError(error.FileNotFound, fx.tmp.dir.statFile(fx.io, "s/d.sock", .{}));
try std.testing.expectError(error.FileNotFound, fx.tmp.dir.statFile(fx.io, "store/.zit.daemon", .{}));
}
test "daemon: a put past the file ceiling is refused too_large and its bytes skipped" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
try fx.makeStore("store");
var d = try Daemon.start(&fx, "store", "s/d.sock", .{ .env = &.{.{ "ZIT_DAEMON_MAX_FILE_BYTES", "16" }} });
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const big = repeat("x", 17);
const v = try c.call(a, .{ .id = 1, .op = .{ .commit = commit("big", null, "big", 0, &.{put("big.bin", big)}) } }, &.{big});
try expectError(v, "too_large", 1);
try std.testing.expect(std.mem.indexOf(u8, str(get(v, "error"), "message"), "16-byte limit") != null);
// The skipped bytes did not run into the next request.
const ok = try c.call(a, .{ .id = 2, .op = .{ .commit = commit("small", null, "small", 1, &.{put("s.txt", repeat("x", 16))}) } }, &.{repeat("x", 16)});
try expectOk(ok);
}
test "daemon: a batch records exactly what the same commits sent one at a time do" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
try fx.makeStore("one");
{
var d = try Daemon.start(&fx, "one", "s/one.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/one.sock");
defer c.close(fx.io);
try expectOk(try c.call(a, .{ .id = 1, .op = .{ .commit = commit("seed", null, "seed", 0, &.{ put("a.txt", "a\n"), put("b.txt", "b\n"), put("d/c.txt", "c\n") }) } }, &.{ "a\n", "b\n", "c\n" }));
c.close(fx.io);
try Daemon.expectExited(&fx, try d.stop(fx.io), 0);
}
try fx.copyStore("one", "many");
// A put, a delete, a rename, an edit that changes nothing, a refusal,
// and a commit after the refusal.
const commits = [_]protocol.Commit{
commit("c1", null, "edit a", 1, &.{put("a.txt", "a2\n")}),
commit("c2", null, "drop b", 2, &.{.{ .delete = "b.txt" }}),
commit("c3", null, "move c", 3, &.{.{ .rename = .{ .old = "d/c.txt", .new = "e/c.txt" } }}),
commit("c4", null, "same", 4, &.{put("a.txt", "a2\n")}),
commit("c5", null, "refused", 5, &.{.{ .delete = "b.txt" }}),
commit("c6", null, "add f", 6, &.{ put("f.txt", "f\n"), .{ .delete = "a.txt" } }),
};
const trailers = [_][]const []const u8{ &.{"a2\n"}, &.{}, &.{}, &.{"a2\n"}, &.{}, &.{"f\n"} };
var singles: std.ArrayList(std.json.Value) = .empty;
{
var d = try Daemon.start(&fx, "one", "s/one.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/one.sock");
defer c.close(fx.io);
for (commits, trailers, 0..) |cm, t, i| try singles.append(a, try c.call(a, .{ .id = @intCast(i), .op = .{ .commit = cm } }, t));
c.close(fx.io);
try Daemon.expectExited(&fx, try d.stop(fx.io), 0);
}
var batch: std.json.Value = undefined;
{
var d = try Daemon.start(&fx, "many", "s/many.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/many.sock");
defer c.close(fx.io);
var all: std.ArrayList([]const u8) = .empty;
for (trailers) |t| try all.appendSlice(a, t);
batch = try c.call(a, .{ .id = 99, .op = .{ .batch = &commits } }, all.items);
c.close(fx.io);
try Daemon.expectExited(&fx, try d.stop(fx.io), 0);
}
try expectOk(batch);
const results = get(batch, "results").array.items;
try std.testing.expectEqual(commits.len, results.len);
for (singles.items, results, commits) |single, item, cm| {
try std.testing.expectEqualStrings(cm.key.?, str(item, "key"));
try std.testing.expectEqual(get(single, "ok").bool, get(item, "ok").bool);
if (get(single, "ok").bool) {
for ([_][]const u8{ "outcome", "hash", "parent", "branch" }) |k| try std.testing.expectEqualStrings(str(single, k), str(item, k));
} else {
try std.testing.expectEqualStrings(str(get(single, "error"), "code"), str(get(item, "error"), "code"));
}
}
try std.testing.expectEqualStrings("unchanged", str(results[3], "outcome"));
try std.testing.expectEqualStrings("path_conflict", str(get(results[4], "error"), "code"));
try std.testing.expect(get(results[5], "ok").bool);
try std.testing.expectEqualSlices(u8, try fx.storeBytes("one"), try fx.storeBytes("many"));
}
test "daemon: a rename and a put of its new name are one rename with content, as commit -R records it" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
try fx.makeStore("store");
const old_bytes = "alpha\nbeta\n";
const new_bytes = "alpha\nbeta\ngamma\n";
const seed_hash = blk: {
var d = try Daemon.start(&fx, "store", "s/d.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const v = try c.call(a, .{ .id = 1, .op = .{ .commit = commit("seed", null, "seed", 0, &.{ put("a.md", old_bytes), put("c.md", "c\n"), put("d.md", "d\n") }) } }, &.{ old_bytes, "c\n", "d\n" });
try expectOk(v);
c.close(fx.io);
try Daemon.expectExited(&fx, try d.stop(fx.io), 0);
break :blk try a.dupe(u8, str(v, "hash"));
};
try fx.copyStore("store", "cli");
try fx.copyStore("store", "batch");
const move = commit("mv", null, "move", 1, &.{ .{ .rename = .{ .old = "a.md", .new = "b.md" } }, put("b.md", new_bytes) });
var d = try Daemon.start(&fx, "store", "s/d.sock", .{});
var running = true;
defer if (running) d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const moved = try c.call(a, .{ .id = 2, .op = .{ .commit = move } }, &.{new_bytes});
try expectOk(moved);
try std.testing.expectEqualStrings("recorded", str(moved, "outcome"));
const hash = str(moved, "hash");
// The CLI's rename-with-content, at the same time, is the same check-in.
{
var time_buf: [24]u8 = undefined;
try fx.tmp.dir.writeFile(fx.io, .{ .sub_path = "stdin.in", .data = new_bytes });
const res = try runZit(&fx, fx.tmp.dir, &.{ "commit", "-R", "cli", "-m", "move", "--user", "e", "--rename", "a.md", "b.md", "--stdin", "b.md", "--date-override", protocol.formatTime(move.time, &time_buf) }, "stdin.in");
try expectExit(res, 0);
try std.testing.expectEqualStrings(hash, std.mem.trimEnd(u8, res.stdout, "\n"));
}
// And so is the same commit in a batch.
{
var bd = try Daemon.start(&fx, "batch", "s/b.sock", .{});
defer bd.child.kill(fx.io);
var bc = try Client.open(&fx, "s/b.sock");
defer bc.close(fx.io);
const v = try bc.call(a, .{ .id = 3, .op = .{ .batch = &.{move} } }, &.{new_bytes});
try expectOk(v);
const item = get(v, "results").array.items[0];
try expectOk(item);
try std.testing.expectEqualStrings(hash, str(item, "hash"));
bc.close(fx.io);
try Daemon.expectExited(&fx, try bd.stop(fx.io), 0);
}
// log and annotate follow the rename back to the seed.
{
const res = try runZit(&fx, fx.tmp.dir, &.{ "log", hash, "b.md", "-R", "store" }, null);
try expectExit(res, 0);
try std.testing.expect(std.mem.indexOf(u8, res.stdout, seed_hash[0..10]) != null);
const ann = try runZit(&fx, fx.tmp.dir, &.{ "annotate", "b.md", hash, "-R", "store" }, null);
try expectExit(ann, 0);
try std.testing.expect(std.mem.indexOf(u8, ann.stdout, seed_hash[0..12]) != null);
try std.testing.expect(std.mem.indexOf(u8, ann.stdout, hash[0..12]) != null);
}
// Every other repeat of a path is refused, and the store is unchanged.
const refusals = [_]struct { edits: []const protocol.Edit, trailers: []const []const u8 }{
.{ .edits = &.{ put("c.md", "1\n"), put("c.md", "2\n") }, .trailers = &.{ "1\n", "2\n" } },
.{ .edits = &.{ .{ .rename = .{ .old = "c.md", .new = "x.md" } }, put("c.md", "1\n") }, .trailers = &.{"1\n"} },
.{ .edits = &.{ .{ .rename = .{ .old = "c.md", .new = "x.md" } }, .{ .rename = .{ .old = "d.md", .new = "x.md" } } }, .trailers = &.{} },
.{ .edits = &.{ .{ .rename = .{ .old = "c.md", .new = "x.md" } }, .{ .rename = .{ .old = "d.md", .new = "x.md" } }, put("x.md", "1\n") }, .trailers = &.{"1\n"} },
.{ .edits = &.{ .{ .rename = .{ .old = "c.md", .new = "x.md" } }, put("x.md", "1\n"), put("x.md", "2\n") }, .trailers = &.{ "1\n", "2\n" } },
};
const before = try fx.storeSize("store");
for (refusals, 0..) |r, i| {
const v = try c.call(a, .{ .id = @intCast(10 + i), .op = .{ .commit = commit("dup", null, "dup", @intCast(10 + i), r.edits) } }, r.trailers);
try expectError(v, "invalid_request", 1);
try std.testing.expectEqual(before, try fx.storeSize("store"));
}
c.close(fx.io);
running = false;
try Daemon.expectExited(&fx, try d.stop(fx.io), 0);
try std.testing.expectEqual(before, try fx.storeSize("store"));
}
test "daemon: feed from zero, nothing new, a CLI commit beside it, forks, purged content, and a pack" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
try fx.makeStore("store");
var d = try Daemon.start(&fx, "store", "s/d.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const seed = try c.call(a, .{ .id = 1, .op = .{ .commit = commit("seed", null, "seed", 0, &.{ put("a.txt", "a1\n"), put("b.txt", "b\n") }) } }, &.{ "a1\n", "b\n" });
try expectOk(seed);
const p0 = try a.dupe(u8, str(seed, "hash"));
const second = try c.call(a, .{ .id = 2, .op = .{ .commit = commit("two", p0, "two", 1, &.{ put("a.txt", "a2\n"), .{ .rename = .{ .old = "b.txt", .new = "c.txt" } } }) } }, &.{"a2\n"});
try expectOk(second);
const p1 = try a.dupe(u8, str(second, "hash"));
const all = try c.call(a, .{ .id = 3, .op = .{ .feed = "0" } }, &.{});
try expectOk(all);
const changes = get(all, "changes").array.items;
try std.testing.expectEqual(@as(usize, 2), changes.len);
try std.testing.expectEqualStrings(p0, str(changes[0], "checkin"));
try std.testing.expect(get(changes[0], "parent") == .null);
try std.testing.expectEqualStrings("trunk", str(changes[0], "branch"));
try std.testing.expectEqualStrings("e", str(changes[0], "user"));
try std.testing.expectEqualStrings("seed", str(changes[0], "comment"));
try std.testing.expectEqualStrings("2027-01-15T08:00:00.007Z", str(changes[0], "time"));
try std.testing.expectEqual(@as(usize, 2), get(changes[0], "files").array.items.len);
try std.testing.expectEqualStrings("added", str(get(changes[0], "files").array.items[0], "state"));
try std.testing.expectEqualStrings(p1, str(changes[1], "checkin"));
try std.testing.expectEqualStrings(p0, str(changes[1], "parent"));
{
var seen: usize = 0;
for (get(changes[1], "files").array.items) |row| {
const name = str(row, "name");
if (std.mem.eql(u8, name, "a.txt")) {
try std.testing.expectEqualStrings("modified", str(row, "state"));
try std.testing.expectEqualStrings(&sha3Hex("a2\n"), str(row, "uuid"));
} else if (std.mem.eql(u8, name, "b.txt")) {
try std.testing.expectEqualStrings("removed", str(row, "state"));
try std.testing.expect(get(row, "uuid") == .null);
} else {
try std.testing.expectEqualStrings("c.txt", name);
try std.testing.expectEqualStrings("modified", str(row, "state"));
try std.testing.expectEqualStrings("b.txt", str(row, "prior"));
}
seen += 1;
}
try std.testing.expectEqual(@as(usize, 3), seen);
}
try std.testing.expectEqual(@as(usize, 0), get(all, "forks").array.items.len);
const cursor1 = try a.dupe(u8, str(all, "cursor"));
// Nothing new: the same cursor and empty lists.
const idle = try c.call(a, .{ .id = 4, .op = .{ .feed = cursor1 } }, &.{});
try expectOk(idle);
try std.testing.expectEqualStrings(cursor1, str(idle, "cursor"));
try std.testing.expectEqual(@as(usize, 0), get(idle, "changes").array.items.len);
// A CLI commit lands while the daemon runs; the next feed has it, and the
// daemon's own next commit builds on it.
try fx.tmp.dir.writeFile(fx.io, .{ .sub_path = "cli.txt", .data = "from the cli\n" });
const cli = try runZit(&fx, fx.tmp.dir, &.{ "commit", "-R", "store", "--stdin", "cli.txt", "-m", "cli", "--user", "u" }, "cli.txt");
try expectExit(cli, 0);
const cli_hash = std.mem.trimEnd(u8, cli.stdout, "\n");
const after = try c.call(a, .{ .id = 5, .op = .{ .feed = cursor1 } }, &.{});
try expectOk(after);
try std.testing.expectEqual(@as(usize, 1), get(after, "changes").array.items.len);
try std.testing.expectEqualStrings(cli_hash, str(get(after, "changes").array.items[0], "checkin"));
try std.testing.expectEqualStrings("u", str(get(after, "changes").array.items[0], "user"));
const cursor2 = try a.dupe(u8, str(after, "cursor"));
const on_cli = try c.call(a, .{ .id = 6, .op = .{ .commit = commit("on", null, "on cli", 3, &.{put("d.txt", "d\n")}) } }, &.{"d\n"});
try expectOk(on_cli);
try std.testing.expectEqualStrings(cli_hash, str(on_cli, "parent"));
// A deliberate fork: the feed names the branch's two leaves.
const fork = try c.call(a, .{ .id = 7, .op = .{ .commit = blk: {
var cm = commit("fork", p1, "fork", 4, &.{put("f.txt", "f\n")});
cm.allow_fork = true;
break :blk cm;
} } }, &.{"f\n"});
try expectOk(fork);
try std.testing.expect(get(fork, "forked").bool);
// Its resend is the same check-in, still reported as the fork it made.
{
const size = try fx.storeSize("store");
var cm = commit("fork", p1, "fork", 4, &.{put("f.txt", "f\n")});
cm.allow_fork = true;
const again = try c.call(a, .{ .id = 70, .op = .{ .commit = cm } }, &.{"f\n"});
try expectOk(again);
try std.testing.expectEqualStrings("existing", str(again, "outcome"));
try std.testing.expectEqualStrings(str(fork, "hash"), str(again, "hash"));
try std.testing.expectEqualStrings(p1, str(again, "parent"));
try std.testing.expect(get(again, "forked").bool);
try std.testing.expectEqual(size, try fx.storeSize("store"));
}
const forked = try c.call(a, .{ .id = 8, .op = .{ .feed = cursor2 } }, &.{});
try expectOk(forked);
try std.testing.expectEqual(@as(usize, 2), get(forked, "changes").array.items.len);
const forks = get(forked, "forks").array.items;
try std.testing.expectEqual(@as(usize, 1), forks.len);
try std.testing.expectEqualStrings("trunk", str(forks[0], "branch"));
try std.testing.expectEqual(@as(usize, 2), get(forks[0], "leaves").array.items.len);
const cursor3 = try a.dupe(u8, str(forked, "cursor"));
// The first a.txt is held by no open leaf: purged, it reads as absent
// by policy, in either form of get.
const a1 = sha3Hex("a1\n");
const purged = try runZit(&fx, fx.tmp.dir, &.{ "purge", "artifacts", "-R", "store", "--force", &a1 }, null);
try expectExit(purged, 0);
const gone = try c.call(a, .{ .id = 9, .op = .{ .get = .{ .one = &a1 } } }, &.{});
try expectError(gone, "content_absent", 8);
const a2 = sha3Hex("a2\n");
const mixed = try c.call(a, .{ .id = 10, .op = .{ .get = .{ .many = &.{ &a1, &a2 } } } }, &.{});
try expectOk(mixed);
try std.testing.expectEqualStrings("content_absent", str(get(get(mixed, "items").array.items[0], "error"), "code"));
try std.testing.expectEqual(@as(i64, 8), get(get(get(mixed, "items").array.items[0], "error"), "exit").integer);
try std.testing.expectEqualStrings("a2\n", try c.take(a, 3));
// A pack that reclaims a deferred shun relocates frames: the old cursor
// is refused, and the feed starts again from zero.
const shunned = sha3Hex("f\n");
const shun = try runZit(&fx, fx.tmp.dir, &.{ "shun", "-R", "store", "--defer", "--force", &shunned }, null);
try expectExit(shun, 0);
const pack = try runZit(&fx, fx.tmp.dir, &.{ "pack", "-R", "store" }, null);
try expectExit(pack, 0);
const dead = try c.call(a, .{ .id = 11, .op = .{ .feed = cursor3 } }, &.{});
try expectError(dead, "cursor_invalid", 4);
const restart = try c.call(a, .{ .id = 12, .op = .{ .feed = "0" } }, &.{});
try expectOk(restart);
try std.testing.expectEqual(@as(usize, 5), get(restart, "changes").array.items.len);
// The daemon commits on the rewritten store as on any other.
const after_pack = try c.call(a, .{ .id = 13, .op = .{ .commit = commit("post", null, "post", 5, &.{put("g.txt", "g\n")}) } }, &.{"g\n"});
try expectOk(after_pack);
try std.testing.expectEqualStrings("recorded", str(after_pack, "outcome"));
}
test "daemon: --exit-on-stdin-close stops when the parent's pipe closes" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
try fx.makeStore("store");
var d = try Daemon.start(&fx, "store", "s/d.sock", .{ .extra = &.{"--exit-on-stdin-close"}, .stdin_pipe = true });
defer if (d.child.id != null) d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
try expectOk(try c.call(fx.a(), .{ .id = 1, .op = .status }, &.{}));
c.close(fx.io);
d.child.stdin.?.close(fx.io);
d.child.stdin = null;
try Daemon.expectExited(&fx, try d.child.wait(fx.io), 0);
try std.testing.expectError(error.FileNotFound, fx.tmp.dir.statFile(fx.io, "s/d.sock", .{}));
try std.testing.expectError(error.FileNotFound, fx.tmp.dir.statFile(fx.io, "store/.zit.daemon", .{}));
}
test "daemon: a crash on either side of the fsync leaves one check-in for the resend" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
for ([_][]const u8{ "after_append", "after_sync" }, 0..) |point, i| {
const store = try std.fmt.allocPrint(a, "store{d}", .{i});
try fx.makeStore(store);
var p0: []const u8 = undefined;
{
var d = try Daemon.start(&fx, store, "s/d.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const seed = try c.call(a, .{ .id = 1, .op = .{ .commit = commit("seed", null, "seed", 0, &.{put("a.txt", "a\n")}) } }, &.{"a\n"});
try expectOk(seed);
p0 = try a.dupe(u8, str(seed, "hash"));
c.close(fx.io);
try Daemon.expectExited(&fx, try d.stop(fx.io), 0);
}
const edits = [_]protocol.Edit{put("b.txt", "b\n")};
{
var d = try Daemon.start(&fx, store, "s/d.sock", .{ .env = &.{.{ "ZIT_DAEMON_FAULT", point }} });
defer if (d.child.id != null) d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
try c.send(.{ .id = 2, .op = .{ .commit = commit("b", p0, "b", 1, &edits) } }, &.{"b\n"});
// The process stops before it answers.
try std.testing.expectError(error.EndOfStream, c.reader.interface.takeByte());
try Daemon.expectExited(&fx, try d.child.wait(fx.io), 70);
}
{
// The dead daemon's socket and lock are reclaimed by the next.
var d = try Daemon.start(&fx, store, "s/d.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const resent = try c.call(a, .{ .id = 3, .op = .{ .commit = commit("b", p0, "b", 1, &edits) } }, &.{"b\n"});
try expectOk(resent);
const outcome = str(resent, "outcome");
try std.testing.expect(std.mem.eql(u8, outcome, "recorded") or std.mem.eql(u8, outcome, "existing"));
c.close(fx.io);
try Daemon.expectExited(&fx, try d.stop(fx.io), 0);
const log = try fx.trunkLog(store);
try std.testing.expectEqual(@as(usize, 2), log.len);
try std.testing.expectEqualStrings(str(resent, "hash"), log[0]);
try std.testing.expectEqualStrings(p0, log[1]);
}
}
}
/// What a resend case does between the crash and the resend.
const Between = enum { nothing, unrelated_cli_commit, cli_commit_same_path };
test "daemon: every resend after a crash is answered by the check-in the first send made" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
try fx.makeStore("store");
var p_tip: []const u8 = undefined;
{
var d = try Daemon.start(&fx, "store", "s/d.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const seed = try c.call(a, .{ .id = 1, .op = .{ .commit = commit("seed", null, "seed", 0, &.{
put("a.txt", "a\n"), put("b.txt", "b\n"), put("c.txt", "c\n"), put("d.txt", "d\n"),
put("e.txt", "e\n"), put("f.txt", "f\n"), put("g.txt", "g\n"), put("h.txt", "h\n"),
}) } }, &.{ "a\n", "b\n", "c\n", "d\n", "e\n", "f\n", "g\n", "h\n" });
try expectOk(seed);
p_tip = try a.dupe(u8, str(seed, "hash"));
c.close(fx.io);
try Daemon.expectExited(&fx, try d.stop(fx.io), 0);
}
const Case = struct { edits: []const protocol.Edit, trailers: []const []const u8, explicit: bool, between: Between = .nothing };
const cases = [_]Case{
.{ .edits = &.{put("a.txt", "a2\n")}, .trailers = &.{"a2\n"}, .explicit = false },
.{ .edits = &.{put("a.txt", "a3\n")}, .trailers = &.{"a3\n"}, .explicit = true },
.{ .edits = &.{.{ .delete = "b.txt" }}, .trailers = &.{}, .explicit = false },
.{ .edits = &.{.{ .delete = "c.txt" }}, .trailers = &.{}, .explicit = true },
.{ .edits = &.{.{ .rename = .{ .old = "d.txt", .new = "d2.txt" } }}, .trailers = &.{}, .explicit = false },
.{ .edits = &.{.{ .rename = .{ .old = "e.txt", .new = "e2.txt" } }}, .trailers = &.{}, .explicit = true },
.{ .edits = &.{ .{ .delete = "f.txt" }, .{ .rename = .{ .old = "g.txt", .new = "g2.txt" } }, put("n1.txt", "n1\n") }, .trailers = &.{"n1\n"}, .explicit = false },
.{ .edits = &.{ .{ .delete = "g2.txt" }, .{ .rename = .{ .old = "h.txt", .new = "h2.txt" } }, put("n2.txt", "n2\n") }, .trailers = &.{"n2\n"}, .explicit = true },
.{ .edits = &.{put("n3.txt", "n3\n")}, .trailers = &.{"n3\n"}, .explicit = false, .between = .unrelated_cli_commit },
.{ .edits = &.{put("n4.txt", "n4\n")}, .trailers = &.{"n4\n"}, .explicit = false, .between = .cli_commit_same_path },
};
for (cases, 0..) |case, i| {
const key = try std.fmt.allocPrint(a, "case{d}", .{i});
const cm = commit(key, if (case.explicit) p_tip else null, key, @intCast(10 + i), case.edits);
{
var d = try Daemon.start(&fx, "store", "s/d.sock", .{ .env = &.{.{ "ZIT_DAEMON_FAULT", "after_sync" }} });
defer if (d.child.id != null) d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
try c.send(.{ .id = 2, .op = .{ .commit = cm } }, case.trailers);
try std.testing.expectError(error.EndOfStream, c.reader.interface.takeByte());
try Daemon.expectExited(&fx, try d.child.wait(fx.io), 70);
}
// The crash came after the fsync: the check-in is the store's tip.
const made = (try fx.trunkLog("store"))[0];
try std.testing.expect(!std.mem.eql(u8, made, p_tip));
switch (case.between) {
.nothing => {},
.unrelated_cli_commit, .cli_commit_same_path => {
const path = if (case.between == .unrelated_cli_commit) "other.txt" else "n4.txt";
const body = try std.fmt.allocPrint(a, "cli {d}\n", .{i});
try fx.tmp.dir.writeFile(fx.io, .{ .sub_path = "cli.txt", .data = body });
const cli = try runZit(&fx, fx.tmp.dir, &.{ "commit", "-R", "store", "--stdin", path, "-m", "cli", "--user", "u" }, "cli.txt");
try expectExit(cli, 0);
},
}
const size = try fx.storeSize("store");
var d = try Daemon.start(&fx, "store", "s/d.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const resent = try c.call(a, .{ .id = 3, .op = .{ .commit = cm } }, case.trailers);
try expectOk(resent);
try std.testing.expectEqualStrings("existing", str(resent, "outcome"));
try std.testing.expectEqualStrings(made, str(resent, "hash"));
try std.testing.expectEqualStrings(p_tip, str(resent, "parent"));
try std.testing.expect(!get(resent, "forked").bool);
try std.testing.expectEqual(size, try fx.storeSize("store"));
// The tip by ancestry, not by time: a CLI commit carries today's
// clock, older than these requests' times.
const st = try c.call(a, .{ .id = 4, .op = .status }, &.{});
p_tip = try a.dupe(u8, str(get(st, "branches").array.items[0], "tip"));
c.close(fx.io);
try Daemon.expectExited(&fx, try d.stop(fx.io), 0);
}
}
test "daemon: the socket path is taken over only from a dead daemon" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
try fx.makeStore("one");
try fx.makeStore("two");
// A regular file at the path is the user's, not a stale socket.
try fx.tmp.dir.createDir(fx.io, "s", .fromMode(0o700));
try fx.tmp.dir.writeFile(fx.io, .{ .sub_path = "s/d.sock", .data = "mine\n" });
const refused = try runZit(&fx, fx.tmp.dir, &.{ "daemon", "-R", "one", "--socket", "s/d.sock" }, null);
try expectExit(refused, 1);
try std.testing.expect(std.mem.indexOf(u8, refused.stderr, "is not a socket") != null);
try std.testing.expectEqualStrings("mine\n", try fx.tmp.dir.readFileAlloc(fx.io, "s/d.sock", fx.a(), .limited(64)));
try fx.tmp.dir.deleteFile(fx.io, "s/d.sock");
// A live daemon of another store keeps its socket.
var d = try Daemon.start(&fx, "one", "s/d.sock", .{});
defer d.child.kill(fx.io);
const second = try runZit(&fx, fx.tmp.dir, &.{ "daemon", "-R", "two", "--socket", "s/d.sock" }, null);
try expectExit(second, 1);
try std.testing.expect(std.mem.indexOf(u8, second.stderr, "already answers") != null);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const hello = try c.call(fx.a(), .{ .id = 1, .op = .{ .hello = protocol.version } }, &.{});
try std.testing.expect(std.mem.endsWith(u8, str(hello, "store"), "one"));
}
test "daemon: a failed fsync is answered as a failure and stops the daemon" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
try fx.makeStore("store");
var d = try Daemon.start(&fx, "store", "s/d.sock", .{ .env = &.{.{ "ZIT_DAEMON_FAULT", "sync_fails" }} });
defer if (d.child.id != null) d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const v = try c.call(fx.a(), .{ .id = 1, .op = .{ .commit = commit("k", null, "k", 0, &.{put("a.txt", "a\n")}) } }, &.{"a\n"});
try expectError(v, "storage", 1);
try std.testing.expectError(error.EndOfStream, c.reader.interface.takeByte());
try Daemon.expectExited(&fx, try d.child.wait(fx.io), 1);
const said = try fx.tmp.dir.readFileAlloc(fx.io, "daemon.err", fx.a(), .limited(1 << 20));
try std.testing.expect(std.mem.indexOf(u8, said, "fsync failed") != null);
}
test "daemon: oversized requests, stalled trailers and overlong headers end their connection, not the daemon" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
try fx.makeStore("store");
var d = try Daemon.start(&fx, "store", "s/d.sock", .{ .env = &.{
.{ "ZIT_DAEMON_MAX_REQUEST_BYTES", "10" },
.{ "ZIT_DAEMON_STALL_MS", "200" },
} });
defer d.child.kill(fx.io);
{
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const v = try c.call(a, .{ .id = 1, .op = .{ .commit = commit("big", null, "big", 0, &.{ put("a", "123456"), put("b", "123456") }) } }, &.{ "123456", "123456" });
try expectError(v, "too_large", 1);
try std.testing.expectError(error.EndOfStream, c.reader.interface.takeByte());
}
{
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
try c.send(.{ .id = 2, .op = .{ .commit = commit("slow", null, "slow", 1, &.{put("a", "12345678")}) } }, &.{"123"});
// The rest never comes: the daemon closes the connection.
try std.testing.expectError(error.EndOfStream, c.reader.interface.takeByte());
}
{
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const chunk = try a.alloc(u8, 64 * 1024);
@memset(chunk, ' ');
// Past 1 MiB with no newline; the daemon may close before it all goes.
for (0..17) |_| c.writer.interface.writeAll(chunk) catch break;
c.writer.interface.flush() catch {};
const v = try c.reply(a);
try expectError(v, "header_too_large", 2);
try std.testing.expect(get(v, "id") == .null);
}
// An idle connection is never a stalled one, and the daemon still serves.
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
try expectOk(try c.call(a, .{ .id = 3, .op = .status }, &.{}));
try fx.io.sleep(.fromMilliseconds(600), .awake);
try expectOk(try c.call(a, .{ .id = 4, .op = .status }, &.{}));
const said = try fx.tmp.dir.readFileAlloc(fx.io, "daemon.err", a, .limited(1 << 20));
try std.testing.expect(std.mem.indexOf(u8, said, "closed a connection whose request sent nothing") != null);
}
test "daemon: a second client is served once the first leaves, and --idle-exit stops an unused daemon" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
try fx.makeStore("store");
var d = try Daemon.start(&fx, "store", "s/d.sock", .{ .extra = &.{ "--idle-exit", "1" } });
defer if (d.child.id != null) d.child.kill(fx.io);
var first = try Client.open(&fx, "s/d.sock");
defer first.close(fx.io);
try expectOk(try first.call(a, .{ .id = 1, .op = .status }, &.{}));
var second = try Client.open(&fx, "s/d.sock");
defer second.close(fx.io);
try second.send(.{ .id = 2, .op = .status }, &.{});
// The first keeps its connection past the idle deadline: an open
// connection is not idle, and the first is still answered.
try fx.io.sleep(.fromMilliseconds(1500), .awake);
try expectOk(try first.call(a, .{ .id = 3, .op = .status }, &.{}));
first.close(fx.io);
const queued = try second.reply(a);
try std.testing.expectEqual(@as(i64, 2), get(queued, "id").integer);
try expectOk(queued);
second.close(fx.io);
// Nobody connected: the daemon goes on its own, cleanly.
try Daemon.expectExited(&fx, try d.child.wait(fx.io), 0);
try std.testing.expectError(error.FileNotFound, fx.tmp.dir.statFile(fx.io, "s/d.sock", .{}));
}
fn lookup(time: protocol.Lookup) protocol.Request {
return .{ .id = 0, .op = .{ .lookup = .{ .one = time } } };
}
fn ident(comment: []const u8, sec: i64, ms: u16) protocol.Lookup {
return .{ .user = "e", .comment = comment, .time = .{ .epoch = epoch0 + sec, .ms = ms } };
}
/// A lookup's answer is the commit resend's, field for field.
fn expectSameAsResend(found: std.json.Value, resend: std.json.Value) !void {
try expectOk(found);
try expectOk(resend);
try std.testing.expectEqualStrings("existing", str(resend, "outcome"));
try std.testing.expect(get(found, "found").bool);
try std.testing.expectEqualStrings(str(resend, "hash"), str(found, "hash"));
try std.testing.expectEqualStrings(str(resend, "branch"), str(found, "branch"));
try std.testing.expectEqual(get(resend, "forked").bool, get(found, "forked").bool);
switch (get(resend, "parent")) {
.null => try std.testing.expect(get(found, "parent") == .null),
else => try std.testing.expectEqualStrings(str(resend, "parent"), str(found, "parent")),
}
}
fn expectNotFound(v: std.json.Value) !void {
try expectOk(v);
try std.testing.expect(!get(v, "found").bool);
try std.testing.expectEqual(@as(usize, 3), v.object.count());
}
test "daemon: lookup answers what a resend would, for daemon and CLI commits, and writes nothing" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
try fx.makeStore("store");
var d = try Daemon.start(&fx, "store", "s/d.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
// Nothing stored yet: not found, and no index to build.
try expectNotFound(try c.call(a, lookup(ident("seed", 0, 7)), &.{}));
const seed = try c.call(a, .{ .id = 1, .op = .{ .commit = commit("seed", null, "seed", 0, &.{put("a.txt", "a1\n")}) } }, &.{"a1\n"});
try expectOk(seed);
const p0 = try a.dupe(u8, str(seed, "hash"));
// Found right after the daemon's own commit.
{
const v = try c.call(a, lookup(ident("seed", 0, 7)), &.{});
try expectOk(v);
try std.testing.expect(get(v, "found").bool);
try std.testing.expectEqualStrings(p0, str(v, "hash"));
try std.testing.expect(get(v, "parent") == .null);
try std.testing.expectEqualStrings("trunk", str(v, "branch"));
try std.testing.expect(!get(v, "forked").bool);
}
const two = try c.call(a, .{ .id = 2, .op = .{ .commit = commit("two", p0, "two", 1, &.{put("a.txt", "a2\n")}) } }, &.{"a2\n"});
try expectOk(two);
const p1 = try a.dupe(u8, str(two, "hash"));
// A fork on p0: found as forked, as its resend reports it.
const fork = try c.call(a, .{ .id = 3, .op = .{ .commit = blk: {
var cm = commit("fork", p0, "fork", 2, &.{put("f.txt", "f\n")});
cm.allow_fork = true;
break :blk cm;
} } }, &.{"f\n"});
try expectOk(fork);
try std.testing.expect(get(fork, "forked").bool);
// A CLI commit with a pinned time lands beside the daemon.
try fx.tmp.dir.writeFile(fx.io, .{ .sub_path = "cli.txt", .data = "from the cli\n" });
const cli = try runZit(&fx, fx.tmp.dir, &.{ "commit", "-R", "store", "--stdin", "cli.txt", "-m", "from cli", "--user", "u", "--date-override", "2027-01-15T08:00:10.123Z" }, "cli.txt");
try expectExit(cli, 0);
const cli_hash = try a.dupe(u8, std.mem.trimEnd(u8, cli.stdout, "\n"));
const cli_ident: protocol.Lookup = .{ .user = "u", .comment = "from cli", .time = .{ .epoch = epoch0 + 10, .ms = 123 } };
// From here on nothing may write: every lookup leaves the bytes alone.
const before = try fx.storeBytes("store");
{
const v = try c.call(a, lookup(cli_ident), &.{});
try expectOk(v);
try std.testing.expect(get(v, "found").bool);
try std.testing.expectEqualStrings(cli_hash, str(v, "hash"));
try std.testing.expectEqualStrings("trunk", str(v, "branch"));
}
// Each lookup agrees with the resend of the same commit.
try expectSameAsResend(
try c.call(a, .{ .id = 4, .op = .{ .lookup = .{ .one = .{ .user = "e", .comment = "two", .time = .{ .epoch = epoch0 + 1, .ms = 7 }, .parent = p0 } } } }, &.{}),
try c.call(a, .{ .id = 4, .op = .{ .commit = commit("two", p0, "two", 1, &.{put("a.txt", "a2\n")}) } }, &.{"a2\n"}),
);
try expectSameAsResend(
try c.call(a, lookup(ident("fork", 2, 7)), &.{}),
try c.call(a, .{ .id = 0, .op = .{ .commit = blk: {
var cm = commit("fork", null, "fork", 2, &.{put("f.txt", "f\n")});
cm.allow_fork = true;
break :blk cm;
} } }, &.{"f\n"}),
);
try expectSameAsResend(
try c.call(a, lookup(ident("seed", 0, 7)), &.{}),
try c.call(a, .{ .id = 0, .op = .{ .commit = commit("seed", null, "seed", 0, &.{put("a.txt", "a1\n")}) } }, &.{"a1\n"}),
);
// A different millisecond, user or comment is another check-in.
try expectNotFound(try c.call(a, lookup(ident("seed", 0, 8)), &.{}));
try expectNotFound(try c.call(a, lookup(.{ .user = "f", .comment = "seed", .time = .{ .epoch = epoch0, .ms = 7 } }), &.{}));
try expectNotFound(try c.call(a, lookup(ident("seed ", 0, 7)), &.{}));
// A named parent must be the one it was built on; a named branch, its own.
{
var q = ident("two", 1, 7);
q.parent = p0;
const v = try c.call(a, lookup(q), &.{});
try expectOk(v);
try std.testing.expectEqualStrings(p1, str(v, "hash"));
try std.testing.expectEqualStrings(p0, str(v, "parent"));
q.parent = p1;
try expectNotFound(try c.call(a, lookup(q), &.{}));
q.parent = repeat("0", 64);
try expectNotFound(try c.call(a, lookup(q), &.{}));
q.parent = null;
q.branch = "trunk";
try std.testing.expect(get(try c.call(a, lookup(q), &.{}), "found").bool);
q.branch = "other";
try expectNotFound(try c.call(a, lookup(q), &.{}));
}
// The list form answers each in order.
{
const items = [_]protocol.Lookup{ cli_ident, ident("seed", 0, 8), ident("seed", 0, 7) };
const v = try c.call(a, .{ .id = 5, .op = .{ .lookup = .{ .many = &items } } }, &.{});
try expectOk(v);
const results = get(v, "results").array.items;
try std.testing.expectEqual(@as(usize, 3), results.len);
try std.testing.expectEqualStrings(cli_hash, str(results[0], "hash"));
try std.testing.expect(!get(results[1], "found").bool);
try std.testing.expectEqual(@as(usize, 1), results[1].object.count());
try std.testing.expectEqualStrings(p0, str(results[2], "hash"));
}
// Malformed requests are refused as any other request is.
try c.send(.{ .id = 6, .op = .status }, &.{});
_ = try c.reply(a);
try c.writer.interface.writeAll("{\"id\":7,\"op\":\"lookup\",\"time\":\"2027-01-15T08:00:00Z\",\"user\":\"e\",\"comment\":\"seed\"}\n");
try c.writer.interface.flush();
try expectError(try c.reply(a), "bad_request", 2);
try c.writer.interface.writeAll("{\"id\":8,\"op\":\"lookup\",\"items\":[{\"user\":\"e\"}]}\n");
try c.writer.interface.flush();
try expectError(try c.reply(a), "bad_request", 2);
const after = try fx.storeBytes("store");
try std.testing.expectEqual(before.len, after.len);
try std.testing.expectEqualSlices(u8, before, after);
}
test "daemon: lookup answers while another process holds the store's writer lock" {
var fx: Fx = undefined;
fx.init();
defer fx.deinit();
const a = fx.a();
try fx.makeStore("store");
var d = try Daemon.start(&fx, "store", "s/d.sock", .{});
defer d.child.kill(fx.io);
var c = try Client.open(&fx, "s/d.sock");
defer c.close(fx.io);
const seed = try c.call(a, .{ .id = 1, .op = .{ .commit = commit("seed", null, "seed", 0, &.{put("a.txt", "a\n")}) } }, &.{"a\n"});
try expectOk(seed);
// A CLI commit lands, then this process takes the writer lock and keeps
// it: anything that waited for the lock would wait its full minute.
try fx.tmp.dir.writeFile(fx.io, .{ .sub_path = "cli.txt", .data = "cli\n" });
const cli = try runZit(&fx, fx.tmp.dir, &.{ "commit", "-R", "store", "--stdin", "cli.txt", "-m", "cli", "--user", "u", "--date-override", "2027-01-15T08:00:05.001Z" }, "cli.txt");
try expectExit(cli, 0);
var store_dir = try fx.tmp.dir.openDir(fx.io, "store", .{});
defer store_dir.close(fx.io);
var discard: std.Io.Writer.Discarding = .init(&.{});
var held = try zit.storelock.acquire(fx.io, store_dir, zit.storelock.writer_lock_name, 0, &discard.writer);
defer held.release(fx.io);
const started = std.Io.Timestamp.now(fx.io, .awake);
const items = [_]protocol.Lookup{ ident("seed", 0, 7), .{ .user = "u", .comment = "cli", .time = .{ .epoch = epoch0 + 5, .ms = 1 } } };
const v = try c.call(a, .{ .id = 2, .op = .{ .lookup = .{ .many = &items } } }, &.{});
const waited_ms = started.durationTo(std.Io.Timestamp.now(fx.io, .awake)).toMilliseconds();
try expectOk(v);
const results = get(v, "results").array.items;
try std.testing.expectEqualStrings(str(seed, "hash"), str(results[0], "hash"));
try std.testing.expectEqualStrings(std.mem.trimEnd(u8, cli.stdout, "\n"), str(results[1], "hash"));
// A writer would wait sixty seconds for this lock; ten is slack for a
// loaded machine, and still no wait.
try std.testing.expect(waited_ms < 10_000);
}