//! The daemon's wire protocol: framing, request parsing, reply encoding, and
//! the dispatch from a parsed request to whatever answers it.
//!
//! A request is one JSON header line, then the raw bytes of each `put` the
//! header declares, in the order the puts appear, with no separator and no
//! encoding. A reply is one JSON line, followed for `get` by the raw bytes of
//! each artifact it found. Requests are answered strictly in order, each
//! reply echoing its request's `id`, so a client may pipeline.
//!
//! Nothing here performs I/O of its own. Bytes arrive through a
//! `std.Io.Reader` and leave through a `std.Io.Writer`, and the store is
//! reached through a `Backend` the caller supplies, so the socket server, a
//! C ABI and the fuzzers all drive the same code. What makes it safe to fuzz
//! is the framing rule `readRequest` keeps: it consumes the header line and
//! exactly the trailer bytes that header declared, never more, and a header
//! whose declarations cannot be read reports that the stream is lost rather
//! than guessing where the next request starts.
const std = @import("std");
const repeat = @import("../repeat.zig").repeat;
const artifact = @import("../artifact.zig");
const checkin = @import("../core/checkin.zig");
const feed = @import("../core/feed.zig");
/// The protocol `hello` speaks.
pub const version: i64 = 1;
/// The longest header line, newline excluded.
pub const header_max: usize = 1 << 20;
/// The exit numbers replies carry, the CLI's own (`cli.Exit`).
pub const exit_failure: u8 = 1;
pub const exit_usage: u8 = 2;
pub const exit_cursor_invalid: u8 = 4;
pub const exit_absent: u8 = 8;
/// A refusal as the wire states it: a snake_case code, the exit number the
/// CLI would end with, and the human-readable detail.
pub const Fault = struct {
code: []const u8,
exit: u8,
message: []const u8,
};
pub const Put = struct {
name: []const u8,
size: u64,
/// The content's SHA3-256 as 64 lowercase hex digits, checked against
/// the trailer before anything is committed.
hash: ?[]const u8 = null,
};
pub const Rename = struct { old: []const u8, new: []const u8 };
pub const Edit = union(enum) {
delete: []const u8,
rename: Rename,
put: Put,
};
pub const Commit = struct {
/// The client's own name for the commit, echoed in the reply.
key: ?[]const u8 = null,
/// Null: the branch's newest open leaf.
parent: ?[]const u8 = null,
branch: ?[]const u8 = null,
user: []const u8,
comment: []const u8,
/// Used verbatim as the check-in time, which is what makes a resend of
/// the same commit find the check-in it already made.
time: checkin.Time,
allow_fork: bool = false,
no_checks: bool = false,
edits: []const Edit,
};
pub const Get = union(enum) {
one: []const u8,
many: []const []const u8,
};
pub const Op = union(enum) {
/// The protocol the client speaks.
hello: i64,
commit: Commit,
batch: []const Commit,
/// The cursor to read from.
feed: []const u8,
get: Get,
status,
};
pub const Request = struct {
id: ?i64 = null,
op: Op,
};
/// A header line as understood: the request, or why it was refused, and the
/// trailer sizes it declared. `trailers` is null when they cannot be known,
/// and then the stream cannot be trusted past this line.
pub const Header = struct {
id: ?i64,
body: union(enum) {
request: Request,
fault: Fault,
},
trailers: ?[]const u64,
};
// ---------------------------------------------------------------------------
// Parsing
/// Understand one header line (without its newline). Every allocation is on
/// `arena`, which the caller resets per request.
pub fn parseHeader(arena: std.mem.Allocator, line: []const u8) error{OutOfMemory}!Header {
const root = std.json.parseFromSliceLeaky(std.json.Value, arena, line, .{
.duplicate_field_behavior = .@"error",
.max_value_len = header_max,
}) catch |e| {
if (e == error.OutOfMemory) return error.OutOfMemory;
return .{
.id = null,
.body = .{ .fault = badRequest("the header is not a JSON value") },
.trailers = null,
};
};
const trailers = try declaredSizes(arena, root);
var p: Parser = .{ .arena = arena };
const id = idOf(root);
const request = p.request(root) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Invalid => return .{ .id = id, .body = .{ .fault = badRequest(p.message) }, .trailers = trailers },
};
return .{ .id = request.id, .body = .{ .request = request }, .trailers = trailers };
}
fn badRequest(message: []const u8) Fault {
return .{ .code = "bad_request", .exit = exit_usage, .message = message };
}
/// The request's id when it has a well-formed one, so even a refusal can be
/// matched to its request.
fn idOf(root: std.json.Value) ?i64 {
if (root != .object) return null;
const v = root.object.get("id") orelse return null;
return if (v == .integer) v.integer else null;
}
/// The trailer sizes the header declares: every `put.size` under `edits` and
/// under each of `commits[].edits`, in order, whether or not the rest of the
/// request is valid, so a refused request's trailers can still be skipped.
/// Null when a declaration is there but unreadable.
fn declaredSizes(arena: std.mem.Allocator, root: std.json.Value) error{OutOfMemory}!?[]const u64 {
var sizes: std.ArrayList(u64) = .empty;
if (root != .object) return sizes.items;
if (root.object.get("edits")) |edits| {
if (!try appendSizes(arena, &sizes, edits)) return null;
}
if (root.object.get("commits")) |commits| {
if (commits != .array) return null;
for (commits.array.items) |c| {
if (c != .object) continue;
const edits = c.object.get("edits") orelse continue;
if (!try appendSizes(arena, &sizes, edits)) return null;
}
}
return sizes.items;
}
fn appendSizes(arena: std.mem.Allocator, sizes: *std.ArrayList(u64), edits: std.json.Value) error{OutOfMemory}!bool {
if (edits != .array) return false;
for (edits.array.items) |e| {
if (e != .object) continue;
const put = e.object.get("put") orelse continue;
if (put != .object) return false;
const size = put.object.get("size") orelse return false;
if (size != .integer or size.integer < 0) return false;
try sizes.append(arena, @intCast(size.integer));
}
return true;
}
const Parser = struct {
arena: std.mem.Allocator,
message: []const u8 = "",
const Error = error{ OutOfMemory, Invalid };
fn fail(p: *Parser, comptime fmt: []const u8, args: anytype) Error {
p.message = try std.fmt.allocPrint(p.arena, fmt, args);
return error.Invalid;
}
fn request(p: *Parser, root: std.json.Value) Error!Request {
if (root != .object) return p.fail("the header is not a JSON object", .{});
const obj = &root.object;
var id: ?i64 = null;
if (obj.get("id")) |v| switch (v) {
.null => {},
.integer => |n| id = n,
else => return p.fail("'id' must be an integer or null", .{}),
};
const op_name = try p.string(obj, "op") orelse return p.fail("'op' is missing", .{});
const Name = enum { hello, commit, batch, feed, get, status };
const name = std.meta.stringToEnum(Name, op_name) orelse return p.fail("'op' names no operation: '{s}'", .{op_name});
const op: Op = switch (name) {
.hello => blk: {
try p.only(obj, &.{ "id", "op", "protocol" });
const v = obj.get("protocol") orelse return p.fail("'hello' needs 'protocol'", .{});
if (v != .integer) return p.fail("'protocol' must be an integer", .{});
break :blk .{ .hello = v.integer };
},
.commit => blk: {
try p.only(obj, &(.{ "id", "op" } ++ commit_fields));
break :blk .{ .commit = try p.commit(obj) };
},
.batch => blk: {
try p.only(obj, &.{ "id", "op", "commits" });
const v = obj.get("commits") orelse return p.fail("'batch' needs 'commits'", .{});
if (v != .array) return p.fail("'commits' must be an array", .{});
const commits = try p.arena.alloc(Commit, v.array.items.len);
for (v.array.items, commits, 0..) |item, *out, i| {
if (item != .object) return p.fail("commits[{d}] is not an object", .{i});
p.only(&item.object, &commit_fields) catch |e| return p.within(e, i);
out.* = p.commit(&item.object) catch |e| return p.within(e, i);
}
break :blk .{ .batch = commits };
},
.feed => blk: {
try p.only(obj, &.{ "id", "op", "cursor" });
break :blk .{ .feed = try p.string(obj, "cursor") orelse return p.fail("'feed' needs 'cursor'", .{}) };
},
.get => blk: {
try p.only(obj, &.{ "id", "op", "uuid", "uuids" });
const one = try p.string(obj, "uuid");
const many = obj.get("uuids");
if ((one == null) == (many == null)) return p.fail("'get' needs exactly one of 'uuid' and 'uuids'", .{});
if (one) |u| break :blk .{ .get = .{ .one = u } };
if (many.? != .array) return p.fail("'uuids' must be an array of strings", .{});
const names = try p.arena.alloc([]const u8, many.?.array.items.len);
for (many.?.array.items, names) |item, *out| {
if (item != .string) return p.fail("'uuids' must be an array of strings", .{});
out.* = item.string;
}
break :blk .{ .get = .{ .many = names } };
},
.status => blk: {
try p.only(obj, &.{ "id", "op" });
break :blk .status;
},
};
return .{ .id = id, .op = op };
}
/// Prefix a refusal inside `commits[i]` with where it is.
fn within(p: *Parser, e: Error, i: usize) Error {
if (e == error.OutOfMemory) return error.OutOfMemory;
return p.fail("commits[{d}]: {s}", .{ i, p.message });
}
const commit_fields = .{ "key", "parent", "branch", "user", "comment", "time", "allow_fork", "no_checks", "edits" };
fn commit(p: *Parser, obj: *const std.json.ObjectMap) Error!Commit {
const time_text = try p.string(obj, "time") orelse return p.fail("'time' is missing", .{});
const time = parseTime(time_text) orelse
return p.fail("'time' must be RFC 3339 UTC with milliseconds (YYYY-MM-DDTHH:MM:SS.mmmZ), not '{s}'", .{time_text});
const edits_v = obj.get("edits") orelse return p.fail("'edits' is missing", .{});
if (edits_v != .array) return p.fail("'edits' must be an array", .{});
const edits = try p.arena.alloc(Edit, edits_v.array.items.len);
for (edits_v.array.items, edits, 0..) |item, *out, i| {
out.* = p.edit(item) catch |e| {
if (e == error.OutOfMemory) return error.OutOfMemory;
return p.fail("edits[{d}]: {s}", .{ i, p.message });
};
}
return .{
.key = try p.string(obj, "key"),
.parent = try p.string(obj, "parent"),
.branch = try p.string(obj, "branch"),
.user = try p.string(obj, "user") orelse return p.fail("'user' is missing", .{}),
.comment = try p.string(obj, "comment") orelse return p.fail("'comment' is missing", .{}),
.time = time,
.allow_fork = try p.flag(obj, "allow_fork"),
.no_checks = try p.flag(obj, "no_checks"),
.edits = edits,
};
}
fn edit(p: *Parser, v: std.json.Value) Error!Edit {
if (v != .object or v.object.count() != 1) return p.fail("an edit is an object with one of 'delete', 'rename', 'put'", .{});
const obj = &v.object;
if (obj.get("delete")) |d| {
if (d != .string) return p.fail("'delete' must name a path", .{});
return .{ .delete = d.string };
}
if (obj.get("rename")) |r| {
if (r != .object) return p.fail("'rename' must be an object with 'old' and 'new'", .{});
try p.only(&r.object, &.{ "old", "new" });
return .{ .rename = .{
.old = try p.string(&r.object, "old") orelse return p.fail("'rename' needs 'old'", .{}),
.new = try p.string(&r.object, "new") orelse return p.fail("'rename' needs 'new'", .{}),
} };
}
if (obj.get("put")) |put| {
if (put != .object) return p.fail("'put' must be an object with 'name' and 'size'", .{});
try p.only(&put.object, &.{ "name", "size", "hash" });
const size = put.object.get("size") orelse return p.fail("'put' needs 'size'", .{});
if (size != .integer or size.integer < 0) return p.fail("'size' must be a non-negative integer", .{});
const hash = try p.string(&put.object, "hash");
if (hash) |h| if (h.len != 64 or !feed.isName(h)) return p.fail("'hash' must be 64 lowercase hex digits", .{});
return .{ .put = .{
.name = try p.string(&put.object, "name") orelse return p.fail("'put' needs 'name'", .{}),
.size = @intCast(size.integer),
.hash = hash,
} };
}
return p.fail("an edit is an object with one of 'delete', 'rename', 'put'", .{});
}
/// The string at `key`; null when absent or JSON null.
fn string(p: *Parser, obj: *const std.json.ObjectMap, key: []const u8) Error!?[]const u8 {
const v = obj.get(key) orelse return null;
return switch (v) {
.null => null,
.string => |s| s,
else => p.fail("'{s}' must be a string", .{key}),
};
}
fn flag(p: *Parser, obj: *const std.json.ObjectMap, key: []const u8) Error!bool {
const v = obj.get(key) orelse return false;
return switch (v) {
.null => false,
.bool => |b| b,
else => p.fail("'{s}' must be true or false", .{key}),
};
}
/// Refuse a key outside `allowed`: a misspelled field is a mistake the
/// client should hear about, not a default it silently gets.
fn only(p: *Parser, obj: *const std.json.ObjectMap, allowed: []const []const u8) Error!void {
var it = obj.iterator();
next: while (it.next()) |entry| {
for (allowed) |a| if (std.mem.eql(u8, a, entry.key_ptr.*)) continue :next;
return p.fail("unknown field '{s}'", .{entry.key_ptr.*});
}
}
};
/// `YYYY-MM-DDTHH:MM:SS.mmmZ` exactly, within the years a D card can carry.
/// One spelling only: the time is part of what a check-in hashes, so two
/// spellings of one instant must not be two requests.
pub fn parseTime(text: []const u8) ?checkin.Time {
if (text.len != 24) return null;
if (text[4] != '-' or text[7] != '-' or text[10] != 'T' or text[13] != ':' or text[16] != ':' or text[19] != '.' or text[23] != 'Z') return null;
const year = digits(text[0..4]) orelse return null;
const month = digits(text[5..7]) orelse return null;
const day = digits(text[8..10]) orelse return null;
const hour = digits(text[11..13]) orelse return null;
const minute = digits(text[14..16]) orelse return null;
const second = digits(text[17..19]) orelse return null;
const ms = digits(text[20..23]) orelse return null;
if (year < 1970 or month < 1 or month > 12 or day < 1 or day > daysIn(year, month)) return null;
if (hour > 23 or minute > 59 or second > 59) return null;
const days = daysFromCivil(year, month, day);
const epoch = days * 86400 + hour * 3600 + minute * 60 + second;
std.debug.assert(epoch >= 0 and epoch <= artifact.max_writer_epoch);
return .{ .epoch = epoch, .ms = @intCast(ms) };
}
fn digits(text: []const u8) ?i64 {
var n: i64 = 0;
for (text) |c| {
if (c < '0' or c > '9') return null;
n = n * 10 + (c - '0');
}
return n;
}
fn daysIn(year: i64, month: i64) i64 {
const leap = (@mod(year, 4) == 0 and @mod(year, 100) != 0) or @mod(year, 400) == 0;
return switch (month) {
2 => if (leap) 29 else 28,
4, 6, 9, 11 => 30,
else => 31,
};
}
/// Days from 1970-01-01 to a proleptic Gregorian date (Howard Hinnant's
/// `days_from_civil`).
fn daysFromCivil(y_in: i64, m: i64, d: i64) i64 {
const y = if (m <= 2) y_in - 1 else y_in;
const era = @divFloor(y, 400);
const yoe = y - era * 400;
const mp = if (m > 2) m - 3 else m + 9;
const doy = @divFloor(153 * mp + 2, 5) + d - 1;
const doe = yoe * 365 + @divFloor(yoe, 4) - @divFloor(yoe, 100) + doy;
return era * 146097 + doe - 719468;
}
/// The wire spelling of a commit time.
pub fn formatTime(time: checkin.Time, buf: *[24]u8) []const u8 {
std.debug.assert(time.ms <= 999);
var ts: [23]u8 = undefined;
_ = artifact.formatWriterTimestampMs(time.epoch, time.ms, &ts);
@memcpy(buf[0..23], &ts);
buf[23] = 'Z';
return buf;
}
/// A millisecond timestamp in the wire spelling.
pub fn formatMillis(ms: u64, buf: *[24]u8) []const u8 {
const epoch: i64 = @intCast(@min(ms / 1000, @as(u64, @intCast(artifact.max_writer_epoch))));
return formatTime(.{ .epoch = epoch, .ms = @intCast(ms % 1000) }, buf);
}
// ---------------------------------------------------------------------------
// Framing
/// A request read off the stream: its header and, per declared size, the
/// trailer bytes, or null when the size passed `max_file_bytes` and the bytes
/// were skipped unread.
pub const Incoming = struct {
header: Header,
trailers: []const ?[]const u8,
};
pub const ReadError = error{
/// The stream ended cleanly, between requests.
Closed,
/// No newline within `header_max` bytes.
HeaderTooLarge,
/// The stream ended inside a request.
Truncated,
ReadFailed,
OutOfMemory,
};
/// How far the request being read has got, for a watcher on another thread:
/// a request whose trailers stop arriving is a stalled client, while a
/// connection idle between requests is an ordinary one.
pub const Progress = struct {
/// Set from the header's arrival until its last trailer byte.
in_trailers: std.atomic.Value(bool) = .init(false),
/// Trailer bytes read, ever; a watcher compares two readings.
bytes: std.atomic.Value(u64) = .init(0),
};
/// Trailer bytes are read in pieces this size, so memory follows what has
/// arrived rather than what a header claims.
const trailer_chunk: usize = 1 << 20;
/// Read one request: the header line, then exactly the trailer bytes it
/// declares. When the header's declarations are unreadable
/// (`header.trailers == null`) nothing past the line is consumed, and the
/// caller must not read another request from this stream; a header whose
/// declared trailers add up past `limits.max_request_bytes` is answered
/// `too_large` the same way, unread. `reader`'s buffer must hold
/// `header_max + 1` bytes for a header that long to be read.
pub fn readRequest(reader: *std.Io.Reader, arena: std.mem.Allocator, limits: Limits, progress: ?*Progress) ReadError!Incoming {
const line = reader.takeDelimiterInclusive('\n') catch |e| switch (e) {
error.EndOfStream => return if (reader.bufferedLen() == 0) error.Closed else error.Truncated,
error.StreamTooLong => return error.HeaderTooLarge,
error.ReadFailed => return error.ReadFailed,
};
if (line.len - 1 > header_max) return error.HeaderTooLarge;
// Parsed before any trailer is read: reading moves the buffer the line
// lives in, and every string the parse keeps is copied onto `arena`.
var header = try parseHeader(arena, line[0 .. line.len - 1]);
const sizes = header.trailers orelse return .{ .header = header, .trailers = &.{} };
var total: u64 = 0;
for (sizes) |size| total +|= size;
if (total > limits.max_request_bytes) {
header.body = .{ .fault = .{
.code = "too_large",
.exit = exit_failure,
.message = try std.fmt.allocPrint(arena, "the request declares {d} bytes of content; one request carries at most {d}", .{ total, limits.max_request_bytes }),
} };
header.trailers = null;
return .{ .header = header, .trailers = &.{} };
}
if (progress) |p| p.in_trailers.store(sizes.len != 0, .release);
defer if (progress) |p| p.in_trailers.store(false, .release);
const trailers = try arena.alloc(?[]const u8, sizes.len);
for (sizes, trailers) |size, *out| {
if (size > limits.max_file_bytes) {
try skipTrailer(reader, size, progress);
out.* = null;
continue;
}
out.* = try readTrailer(reader, arena, @intCast(size), progress);
}
return .{ .header = header, .trailers = trailers };
}
/// One trailer, grown as its bytes arrive.
fn readTrailer(reader: *std.Io.Reader, arena: std.mem.Allocator, size: usize, progress: ?*Progress) ReadError![]const u8 {
var bytes: std.ArrayList(u8) = .empty;
while (bytes.items.len < size) {
const n = @min(size - bytes.items.len, trailer_chunk);
try bytes.ensureUnusedCapacity(arena, n);
const dest = bytes.unusedCapacitySlice()[0..n];
reader.readSliceAll(dest) catch |e| return truncated(e);
bytes.items.len += n;
if (progress) |p| _ = p.bytes.fetchAdd(n, .monotonic);
}
return bytes.items;
}
fn skipTrailer(reader: *std.Io.Reader, size: u64, progress: ?*Progress) ReadError!void {
var left = size;
while (left > 0) {
const n = @min(left, trailer_chunk);
reader.discardAll64(n) catch |e| return truncated(e);
left -= n;
if (progress) |p| _ = p.bytes.fetchAdd(n, .monotonic);
}
}
fn truncated(e: std.Io.Reader.Error) ReadError {
return switch (e) {
error.EndOfStream => error.Truncated,
error.ReadFailed => error.ReadFailed,
};
}
// ---------------------------------------------------------------------------
// Dispatch
pub const OutcomeWord = enum { recorded, unchanged, existing };
pub const Outcome = struct {
outcome: OutcomeWord,
hash: []const u8,
/// Null for the first check-in of a store.
parent: ?[]const u8,
branch: []const u8,
forked: bool,
};
pub const ItemResult = union(enum) {
ok: Outcome,
fault: Fault,
};
/// One commit ready for the core: its request, and its edits with each
/// put's bytes in place, ordered deletes, renames, puts.
pub const CommitWork = struct {
body: *const Commit,
edits: []const checkin.Edit,
};
pub fn Answer(comptime T: type) type {
return union(enum) { ok: T, fault: Fault };
}
pub const Hello = struct {
zit: []const u8,
store: []const u8,
tip: feed.Status,
};
/// What answers requests. Everything returned lives on the `arena` passed
/// in, which outlives the reply.
pub const Backend = struct {
ctx: *anyopaque,
vtable: *const VTable,
pub const VTable = struct {
hello: *const fn (ctx: *anyopaque, arena: std.mem.Allocator) Answer(Hello),
/// Record every commit in `works`, filling `results` one for one.
/// The commits are independent: one refused leaves the others as
/// they would be alone. Returns only once they are all durable.
commit: *const fn (ctx: *anyopaque, arena: std.mem.Allocator, works: []const CommitWork, results: []ItemResult) void,
feed: *const fn (ctx: *anyopaque, arena: std.mem.Allocator, cursor: []const u8) Answer(feed.Feed),
get: *const fn (ctx: *anyopaque, arena: std.mem.Allocator, uuid: []const u8) Answer([]const u8),
status: *const fn (ctx: *anyopaque, arena: std.mem.Allocator) Answer(feed.Status),
};
};
pub const Limits = struct {
/// The largest put a commit may carry: the core's ceiling, lowered by a
/// test that cannot afford to send a gibibyte.
max_file_bytes: u64 = checkin.max_file_bytes,
/// The most content one request may declare across all its puts, so a
/// header cannot make the daemon hold more than this for one answer.
max_request_bytes: u64 = checkin.max_file_bytes,
};
/// Answer one request onto `out` (not flushed).
pub fn dispatch(
arena: std.mem.Allocator,
backend: Backend,
incoming: *const Incoming,
limits: Limits,
out: *std.Io.Writer,
) error{ WriteFailed, OutOfMemory }!void {
const header = &incoming.header;
const request = switch (header.body) {
.request => |r| r,
.fault => |f| return writeFault(out, header.id, f),
};
const id = request.id;
switch (request.op) {
.hello => |spoken| {
if (spoken != version) {
const message = try std.fmt.allocPrint(arena, "this daemon speaks protocol {d}, not {d}", .{ version, spoken });
return writeFault(out, id, badRequest(message));
}
switch (backend.vtable.hello(backend.ctx, arena)) {
.ok => |h| return writeHello(out, id, h),
.fault => |f| return writeFault(out, id, f),
}
},
.commit => |*c| {
var results: [1]ItemResult = undefined;
try commitAll(arena, backend, (c)[0..1], incoming.trailers, limits, &results);
switch (results[0]) {
.ok => |o| return writeCommitOk(out, id, c.key, o),
.fault => |f| return writeFault(out, id, f),
}
},
.batch => |commits| {
const results = try arena.alloc(ItemResult, commits.len);
try commitAll(arena, backend, commits, incoming.trailers, limits, results);
return writeBatch(out, id, commits, results);
},
.feed => |cursor| switch (backend.vtable.feed(backend.ctx, arena, cursor)) {
.ok => |f| return writeFeed(out, id, f),
.fault => |f| return writeFault(out, id, f),
},
.get => |g| switch (g) {
.one => |uuid| switch (backend.vtable.get(backend.ctx, arena, uuid)) {
.ok => |bytes| return writeGetOne(out, id, uuid, bytes),
.fault => |f| return writeFault(out, id, f),
},
.many => |uuids| {
const answers = try arena.alloc(Answer([]const u8), uuids.len);
for (uuids, answers) |uuid, *a| a.* = backend.vtable.get(backend.ctx, arena, uuid);
return writeGetMany(out, id, uuids, answers);
},
},
.status => switch (backend.vtable.status(backend.ctx, arena)) {
.ok => |s| return writeStatus(out, id, s),
.fault => |f| return writeFault(out, id, f),
},
}
}
/// Check each commit's puts against their trailers, hand the ones that pass
/// to the backend in one call, and merge the answers back in request order.
fn commitAll(
arena: std.mem.Allocator,
backend: Backend,
commits: []const Commit,
trailers: []const ?[]const u8,
limits: Limits,
results: []ItemResult,
) error{OutOfMemory}!void {
std.debug.assert(commits.len == results.len);
var works: std.ArrayList(CommitWork) = .empty;
var slots: std.ArrayList(usize) = .empty;
var next_trailer: usize = 0;
for (commits, results, 0..) |*c, *result, i| {
const work = try prepare(arena, c, trailers, &next_trailer, limits);
switch (work) {
.ok => |w| {
try works.append(arena, w);
try slots.append(arena, i);
},
.fault => |f| result.* = .{ .fault = f },
}
}
std.debug.assert(next_trailer == trailers.len);
if (works.items.len == 0) return;
const answered = try arena.alloc(ItemResult, works.items.len);
backend.vtable.commit(backend.ctx, arena, works.items, answered);
for (slots.items, answered) |slot, a| results[slot] = a;
}
/// One commit's edits for the core, or why it cannot go to the core at all.
/// Consumes this commit's trailers from `next` either way.
fn prepare(
arena: std.mem.Allocator,
c: *const Commit,
trailers: []const ?[]const u8,
next: *usize,
limits: Limits,
) error{OutOfMemory}!Answer(CommitWork) {
var fault: ?Fault = null;
const edits = try arena.alloc(checkin.Edit, c.edits.len);
// Deletes first, then renames, then puts, as `commit -R` orders them:
// a rename may take the name a delete frees.
var at: usize = 0;
for (c.edits) |e| if (e == .delete) {
edits[at] = .{ .delete = e.delete };
at += 1;
};
for (c.edits) |e| if (e == .rename) {
edits[at] = .{ .rename = .{ .old = e.rename.old, .new = e.rename.new } };
at += 1;
};
for (c.edits) |e| if (e == .put) {
const put = e.put;
const trailer = trailers[next.*];
next.* += 1;
const bytes = trailer orelse {
if (fault == null) fault = .{
.code = "too_large",
.exit = exit_failure,
.message = try std.fmt.allocPrint(arena, "cannot commit '{s}': {d} bytes exceeds the {d}-byte limit on one file", .{ put.name, put.size, limits.max_file_bytes }),
};
continue;
};
std.debug.assert(bytes.len == put.size);
if (put.hash) |want| {
var digest: [32]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(bytes, &digest, .{});
const have = std.fmt.bytesToHex(digest, .lower);
if (!std.mem.eql(u8, &have, want) and fault == null) fault = .{
.code = "hash_mismatch",
.exit = exit_usage,
.message = try std.fmt.allocPrint(arena, "the bytes sent for '{s}' hash to {s}, not the {s} the request names", .{ put.name, &have, want }),
};
}
edits[at] = .{ .put = .{ .path = put.name, .bytes = bytes } };
at += 1;
};
if (fault) |f| return .{ .fault = f };
std.debug.assert(at == edits.len);
return .{ .ok = .{ .body = c, .edits = edits } };
}
// ---------------------------------------------------------------------------
// Faults from the layers below
/// A core refusal on the wire: the error's name in snake_case, and the exit
/// `commit -R` ends with for it. A core error whose code must differ from
/// its name gets an arm here; the core's refusal of control bytes in a user,
/// branch or path is expected as `error.BadName`, which the derivation
/// already spells `bad_name`, exit 1, as the protocol documents.
pub fn coreFault(e: checkin.Error, message: []const u8) Fault {
return .{
.code = switch (e) {
inline else => |tag| comptime snakeCase(@errorName(tag)),
},
.exit = exit_failure,
.message = message,
};
}
/// A read-side refusal on the wire.
pub fn feedFault(e: feed.Error, message: []const u8) Fault {
return switch (e) {
error.CursorInvalid => .{ .code = "cursor_invalid", .exit = exit_cursor_invalid, .message = message },
error.BadName => badRequest("not an artifact name: 40 or 64 lowercase hex digits"),
inline else => |tag| .{ .code = comptime snakeCase(@errorName(tag)), .exit = exit_failure, .message = message },
};
}
fn snakeCase(comptime name: []const u8) []const u8 {
comptime {
var out: []const u8 = "";
for (name, 0..) |c, i| {
if (std.ascii.isUpper(c)) {
if (i != 0) out = out ++ "_";
out = out ++ [_]u8{std.ascii.toLower(c)};
} else out = out ++ [_]u8{c};
}
return out;
}
}
// ---------------------------------------------------------------------------
// Encoding
const Json = std.json.Stringify;
fn begin(out: *std.Io.Writer, id: ?i64, ok: bool) Json.Error!Json {
var jw: Json = .{ .writer = out };
try jw.beginObject();
try jw.objectField("id");
try jw.write(id);
try jw.objectField("ok");
try jw.write(ok);
return jw;
}
fn finish(jw: *Json) Json.Error!void {
try jw.endObject();
try jw.writer.writeByte('\n');
}
/// A string from the store, which need not be UTF-8: invalid sequences
/// become U+FFFD so the reply stays one JSON string.
fn storeText(jw: *Json, s: []const u8) Json.Error!void {
if (std.unicode.utf8ValidateSlice(s)) return jw.write(s);
try jw.beginWriteRaw();
defer jw.endWriteRaw();
const w = jw.writer;
try w.writeByte('"');
var i: usize = 0;
while (i < s.len) {
const n = std.unicode.utf8ByteSequenceLength(s[i]) catch 0;
if (n == 0 or i + n > s.len or !std.unicode.utf8ValidateSlice(s[i..][0..n])) {
try w.writeAll("\\ufffd");
i += 1;
continue;
}
try Json.encodeJsonStringChars(s[i..][0..n], .{}, w);
i += n;
}
try w.writeByte('"');
}
fn optStoreText(jw: *Json, s: ?[]const u8) Json.Error!void {
if (s) |v| return storeText(jw, v);
return jw.write(null);
}
fn faultObject(jw: *Json, f: Fault) Json.Error!void {
try jw.beginObject();
try jw.objectField("code");
try jw.write(f.code);
try jw.objectField("exit");
try jw.write(f.exit);
try jw.objectField("message");
try storeText(jw, f.message);
try jw.endObject();
}
pub fn writeFault(out: *std.Io.Writer, id: ?i64, f: Fault) Json.Error!void {
var jw = try begin(out, id, false);
try jw.objectField("error");
try faultObject(&jw, f);
try finish(&jw);
}
fn statusFields(jw: *Json, s: feed.Status) Json.Error!void {
try jw.objectField("branches");
try jw.beginArray();
for (s.branches) |b| {
try jw.beginObject();
try jw.objectField("name");
try storeText(jw, b.name);
try jw.objectField("tip");
try jw.write(b.tip);
try jw.objectField("leaves");
try jw.write(b.leaves);
try jw.endObject();
}
try jw.endArray();
try jw.objectField("cursor");
try jw.write(s.cursor);
}
fn writeHello(out: *std.Io.Writer, id: ?i64, h: Hello) Json.Error!void {
var jw = try begin(out, id, true);
try jw.objectField("protocol");
try jw.write(version);
try jw.objectField("zit");
try jw.write(h.zit);
try jw.objectField("store");
try storeText(&jw, h.store);
try jw.objectField("tip");
try jw.beginObject();
try statusFields(&jw, h.tip);
try jw.endObject();
try finish(&jw);
}
fn outcomeFields(jw: *Json, key: ?[]const u8, o: Outcome) Json.Error!void {
try jw.objectField("key");
try jw.write(key);
try jw.objectField("outcome");
try jw.write(@tagName(o.outcome));
try jw.objectField("hash");
try jw.write(o.hash);
try jw.objectField("parent");
try jw.write(o.parent);
try jw.objectField("branch");
try storeText(jw, o.branch);
try jw.objectField("forked");
try jw.write(o.forked);
}
fn writeCommitOk(out: *std.Io.Writer, id: ?i64, key: ?[]const u8, o: Outcome) Json.Error!void {
var jw = try begin(out, id, true);
try outcomeFields(&jw, key, o);
try finish(&jw);
}
fn writeBatch(out: *std.Io.Writer, id: ?i64, commits: []const Commit, results: []const ItemResult) Json.Error!void {
var jw = try begin(out, id, true);
try jw.objectField("results");
try jw.beginArray();
for (commits, results) |c, r| {
try jw.beginObject();
switch (r) {
.ok => |o| {
try jw.objectField("ok");
try jw.write(true);
try outcomeFields(&jw, c.key, o);
},
.fault => |f| {
try jw.objectField("key");
try jw.write(c.key);
try jw.objectField("ok");
try jw.write(false);
try jw.objectField("error");
try faultObject(&jw, f);
},
}
try jw.endObject();
}
try jw.endArray();
try finish(&jw);
}
fn writeFeed(out: *std.Io.Writer, id: ?i64, f: feed.Feed) Json.Error!void {
var jw = try begin(out, id, true);
try jw.objectField("cursor");
try jw.write(f.cursor);
try jw.objectField("changes");
try jw.beginArray();
for (f.changes) |c| {
try jw.beginObject();
try jw.objectField("checkin");
try jw.write(c.checkin);
try jw.objectField("parent");
try jw.write(c.parent);
try jw.objectField("branch");
try storeText(&jw, c.branch);
try jw.objectField("user");
try optStoreText(&jw, c.user);
try jw.objectField("comment");
try optStoreText(&jw, c.comment);
try jw.objectField("time");
if (c.time_ms) |ms| {
var buf: [24]u8 = undefined;
try jw.write(formatMillis(ms, &buf));
} else try jw.write(null);
try jw.objectField("files");
if (c.files) |files| {
try jw.beginArray();
for (files) |row| {
try jw.beginObject();
try jw.objectField("name");
try storeText(&jw, row.name);
try jw.objectField("prior");
try optStoreText(&jw, row.prior);
try jw.objectField("uuid");
try jw.write(row.uuid);
try jw.objectField("state");
try jw.write(@tagName(row.state));
try jw.endObject();
}
try jw.endArray();
} else try jw.write(null);
try jw.endObject();
}
try jw.endArray();
try jw.objectField("forks");
try jw.beginArray();
for (f.forks) |k| {
try jw.beginObject();
try jw.objectField("branch");
try storeText(&jw, k.branch);
try jw.objectField("leaves");
try jw.write(k.leaves);
try jw.endObject();
}
try jw.endArray();
try finish(&jw);
}
fn writeGetOne(out: *std.Io.Writer, id: ?i64, uuid: []const u8, bytes: []const u8) Json.Error!void {
var jw = try begin(out, id, true);
try jw.objectField("uuid");
try storeText(&jw, uuid);
try jw.objectField("size");
try jw.write(bytes.len);
try finish(&jw);
try out.writeAll(bytes);
}
fn writeGetMany(out: *std.Io.Writer, id: ?i64, uuids: []const []const u8, answers: []const Answer([]const u8)) Json.Error!void {
var jw = try begin(out, id, true);
try jw.objectField("items");
try jw.beginArray();
for (uuids, answers) |uuid, a| {
try jw.beginObject();
try jw.objectField("uuid");
try storeText(&jw, uuid);
switch (a) {
.ok => |bytes| {
try jw.objectField("size");
try jw.write(bytes.len);
},
.fault => |f| {
try jw.objectField("error");
try faultObject(&jw, f);
},
}
try jw.endObject();
}
try jw.endArray();
try finish(&jw);
for (answers) |a| switch (a) {
.ok => |bytes| try out.writeAll(bytes),
.fault => {},
};
}
fn writeStatus(out: *std.Io.Writer, id: ?i64, s: feed.Status) Json.Error!void {
var jw = try begin(out, id, true);
try statusFields(&jw, s);
try finish(&jw);
}
/// A request's header line, as a client sends it (the trailers follow it,
/// in put order). `parseHeader` reads back exactly `request`.
pub fn writeRequest(out: *std.Io.Writer, request: Request) Json.Error!void {
var jw: Json = .{ .writer = out };
try jw.beginObject();
try jw.objectField("id");
try jw.write(request.id);
try jw.objectField("op");
try jw.write(@tagName(request.op));
switch (request.op) {
.hello => |v| {
try jw.objectField("protocol");
try jw.write(v);
},
.commit => |c| try commitFields(&jw, c),
.batch => |commits| {
try jw.objectField("commits");
try jw.beginArray();
for (commits) |c| {
try jw.beginObject();
try commitFields(&jw, c);
try jw.endObject();
}
try jw.endArray();
},
.feed => |cursor| {
try jw.objectField("cursor");
try jw.write(cursor);
},
.get => |g| switch (g) {
.one => |u| {
try jw.objectField("uuid");
try jw.write(u);
},
.many => |us| {
try jw.objectField("uuids");
try jw.write(us);
},
},
.status => {},
}
try jw.endObject();
try out.writeByte('\n');
}
fn commitFields(jw: *Json, c: Commit) Json.Error!void {
var tbuf: [24]u8 = undefined;
try jw.objectField("key");
try jw.write(c.key);
try jw.objectField("parent");
try jw.write(c.parent);
try jw.objectField("branch");
try jw.write(c.branch);
try jw.objectField("user");
try jw.write(c.user);
try jw.objectField("comment");
try jw.write(c.comment);
try jw.objectField("time");
try jw.write(formatTime(c.time, &tbuf));
try jw.objectField("allow_fork");
try jw.write(c.allow_fork);
try jw.objectField("no_checks");
try jw.write(c.no_checks);
try jw.objectField("edits");
try jw.beginArray();
for (c.edits) |e| {
try jw.beginObject();
switch (e) {
.delete => |p| {
try jw.objectField("delete");
try jw.write(p);
},
.rename => |r| {
try jw.objectField("rename");
try jw.beginObject();
try jw.objectField("old");
try jw.write(r.old);
try jw.objectField("new");
try jw.write(r.new);
try jw.endObject();
},
.put => |p| {
try jw.objectField("put");
try jw.beginObject();
try jw.objectField("name");
try jw.write(p.name);
try jw.objectField("size");
try jw.write(p.size);
if (p.hash) |h| {
try jw.objectField("hash");
try jw.write(h);
}
try jw.endObject();
},
}
try jw.endObject();
}
try jw.endArray();
}
/// Structural equality over requests, for the round-trip property.
pub fn eqlRequest(a: Request, b: Request) bool {
if (a.id != b.id) return false;
if (std.meta.activeTag(a.op) != std.meta.activeTag(b.op)) return false;
return switch (a.op) {
.hello => |v| v == b.op.hello,
.commit => |c| eqlCommit(c, b.op.commit),
.batch => |cs| blk: {
if (cs.len != b.op.batch.len) break :blk false;
for (cs, b.op.batch) |x, y| if (!eqlCommit(x, y)) break :blk false;
break :blk true;
},
.feed => |cur| std.mem.eql(u8, cur, b.op.feed),
.get => |g| switch (g) {
.one => |u| b.op.get == .one and std.mem.eql(u8, u, b.op.get.one),
.many => |us| b.op.get == .many and eqlStrings(us, b.op.get.many),
},
.status => true,
};
}
fn eqlOpt(a: ?[]const u8, b: ?[]const u8) bool {
if (a == null or b == null) return a == null and b == null;
return std.mem.eql(u8, a.?, b.?);
}
fn eqlStrings(a: []const []const u8, b: []const []const u8) bool {
if (a.len != b.len) return false;
for (a, b) |x, y| if (!std.mem.eql(u8, x, y)) return false;
return true;
}
fn eqlCommit(a: Commit, b: Commit) bool {
if (!eqlOpt(a.key, b.key) or !eqlOpt(a.parent, b.parent) or !eqlOpt(a.branch, b.branch)) return false;
if (!std.mem.eql(u8, a.user, b.user) or !std.mem.eql(u8, a.comment, b.comment)) return false;
if (a.time.epoch != b.time.epoch or a.time.ms != b.time.ms) return false;
if (a.allow_fork != b.allow_fork or a.no_checks != b.no_checks) return false;
if (a.edits.len != b.edits.len) return false;
for (a.edits, b.edits) |x, y| {
if (std.meta.activeTag(x) != std.meta.activeTag(y)) return false;
const same = switch (x) {
.delete => |p| std.mem.eql(u8, p, y.delete),
.rename => |r| std.mem.eql(u8, r.old, y.rename.old) and std.mem.eql(u8, r.new, y.rename.new),
.put => |p| std.mem.eql(u8, p.name, y.put.name) and p.size == y.put.size and eqlOpt(p.hash, y.put.hash),
};
if (!same) return false;
}
return true;
}
// ---------------------------------------------------------------------------
// Tests
const testing = std.testing;
fn parseOk(arena: std.mem.Allocator, line: []const u8) !Request {
const h = try parseHeader(arena, line);
return switch (h.body) {
.request => |r| r,
.fault => |f| {
std.debug.print("unexpected refusal: {s}\n", .{f.message});
return error.TestUnexpectedResult;
},
};
}
fn expectRefused(arena: std.mem.Allocator, line: []const u8, has_trailers: bool) !Fault {
const h = try parseHeader(arena, line);
try testing.expectEqual(has_trailers, h.trailers != null);
return switch (h.body) {
.request => error.TestUnexpectedResult,
.fault => |f| f,
};
}
test "parseHeader: the commit example reads into every field" {
var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
defer arena_state.deinit();
const a = arena_state.allocator();
const line =
\\{"id":2,"op":"commit","key":"k-1","parent":null,"branch":null,"user":"e","comment":"save","time":"2026-11-05T12:34:56.789Z","allow_fork":true,"no_checks":false,"edits":[{"delete":"p"},{"rename":{"old":"x","new":"y"}},{"put":{"name":"q","size":3,"hash":"
++ repeat("ab", 32) ++
\\"}}]}
;
const h = try parseHeader(a, line);
try testing.expectEqualSlices(u64, &.{3}, h.trailers.?);
const c = h.body.request.op.commit;
try testing.expectEqual(@as(?i64, 2), h.body.request.id);
try testing.expectEqualStrings("k-1", c.key.?);
try testing.expect(c.parent == null and c.branch == null);
try testing.expectEqualStrings("e", c.user);
try testing.expect(c.allow_fork and !c.no_checks);
try testing.expectEqual(@as(i64, 1793882096), c.time.epoch);
try testing.expectEqual(@as(u16, 789), c.time.ms);
try testing.expectEqualStrings("p", c.edits[0].delete);
try testing.expectEqualStrings("y", c.edits[1].rename.new);
try testing.expectEqual(@as(u64, 3), c.edits[2].put.size);
}
test "parseHeader: refusals keep the id and, when they can, the trailer sizes" {
var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
defer arena_state.deinit();
const a = arena_state.allocator();
// Not JSON: nothing is known, so the stream is lost.
const garbage = try expectRefused(a, "{\"id\":1,", false);
try testing.expectEqualStrings("bad_request", garbage.code);
try testing.expectEqual(exit_usage, garbage.exit);
// A bad field: refused, but the declared put can still be skipped.
const h = try parseHeader(a, "{\"id\":7,\"op\":\"commit\",\"user\":\"e\",\"comment\":\"c\",\"time\":\"yesterday\",\"edits\":[{\"put\":{\"name\":\"a\",\"size\":5}}]}");
try testing.expectEqual(@as(?i64, 7), h.id);
try testing.expectEqualSlices(u64, &.{5}, h.trailers.?);
try testing.expect(h.body == .fault);
// An unreadable size: the request cannot be skipped.
_ = try expectRefused(a, "{\"op\":\"commit\",\"edits\":[{\"put\":{\"name\":\"a\",\"size\":-1}}]}", false);
_ = try expectRefused(a, "{\"op\":\"commit\",\"edits\":[{\"put\":{\"name\":\"a\"}}]}", false);
// Unknown fields and operations are refused, not ignored.
_ = try expectRefused(a, "{\"op\":\"status\",\"verbose\":true}", true);
_ = try expectRefused(a, "{\"op\":\"rebase\"}", true);
_ = try expectRefused(a, "{\"op\":\"get\",\"uuid\":\"a\",\"uuids\":[]}", true);
_ = try expectRefused(a, "{\"op\":\"get\"}", true);
_ = try expectRefused(a, "{\"op\":\"feed\",\"cursor\":\"0\",\"cursor\":\"1\"}", false);
_ = try expectRefused(a, "[1,2]", true);
}
test "parseTime: one spelling, real dates, within the D card's years" {
const t = parseTime("1970-01-01T00:00:00.000Z").?;
try testing.expectEqual(@as(i64, 0), t.epoch);
try testing.expectEqual(@as(i64, 951782400), parseTime("2000-02-29T00:00:00.000Z").?.epoch);
try testing.expectEqual(artifact.max_writer_epoch, parseTime("9999-12-31T23:59:59.999Z").?.epoch);
for ([_][]const u8{
"2001-02-29T00:00:00.000Z",
"1969-12-31T23:59:59.999Z",
"2026-13-01T00:00:00.000Z",
"2026-11-05T24:00:00.000Z",
"2026-11-05T12:00:60.000Z",
"2026-11-05T12:00:00Z",
"2026-11-05 12:00:00.000Z",
"2026-11-05T12:00:00.000+00:00",
"2026-11-05T12:00:00.0000Z",
}) |bad| try testing.expect(parseTime(bad) == null);
var buf: [24]u8 = undefined;
try testing.expectEqualStrings("2026-11-05T12:34:56.789Z", formatTime(parseTime("2026-11-05T12:34:56.789Z").?, &buf));
}
test "readRequest: consumes the header and exactly the declared trailers, skipping an oversized one" {
var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
defer arena_state.deinit();
const a = arena_state.allocator();
const input = "{\"id\":1,\"op\":\"batch\",\"commits\":[{\"user\":\"e\",\"comment\":\"c\",\"time\":\"2026-01-01T00:00:00.000Z\",\"edits\":[{\"put\":{\"name\":\"a\",\"size\":3}},{\"put\":{\"name\":\"b\",\"size\":9}}]}]}\nabc123456789{\"op\":\"status\"}\n";
var r: std.Io.Reader = .fixed(input);
const first = try readRequest(&r, a, .{ .max_file_bytes = 8 }, null);
try testing.expectEqualStrings("abc", first.trailers[0].?);
try testing.expect(first.trailers[1] == null);
const second = try readRequest(&r, a, .{ .max_file_bytes = 8 }, null);
try testing.expect(second.header.body.request.op == .status);
try testing.expectError(error.Closed, readRequest(&r, a, .{ .max_file_bytes = 8 }, null));
var cut: std.Io.Reader = .fixed("{\"op\":\"commit\",\"edits\":[{\"put\":{\"name\":\"a\",\"size\":4}}]}\nab");
try testing.expectError(error.Truncated, readRequest(&cut, a, .{ .max_file_bytes = 8 }, null));
var half: std.Io.Reader = .fixed("{\"op\":");
try testing.expectError(error.Truncated, readRequest(&half, a, .{ .max_file_bytes = 8 }, null));
}
test "readRequest: a request declaring more content than one request may carry is refused unread" {
var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
defer arena_state.deinit();
const a = arena_state.allocator();
const input = "{\"id\":3,\"op\":\"batch\",\"commits\":[{\"user\":\"e\",\"comment\":\"c\",\"time\":\"2026-01-01T00:00:00.000Z\",\"edits\":[{\"put\":{\"name\":\"a\",\"size\":6}},{\"put\":{\"name\":\"b\",\"size\":6}}]}]}\nabcdefghijkl";
var r: std.Io.Reader = .fixed(input);
var progress: Progress = .{};
const got = try readRequest(&r, a, .{ .max_file_bytes = 8, .max_request_bytes = 10 }, &progress);
try testing.expect(got.header.trailers == null);
try testing.expectEqualStrings("too_large", got.header.body.fault.code);
try testing.expectEqual(@as(?i64, 3), got.header.id);
// Nothing past the header was read; the caller closes the stream.
try testing.expectEqualStrings("abcdefghijkl", r.buffered());
try testing.expectEqual(@as(u64, 0), progress.bytes.load(.monotonic));
var ok: std.Io.Reader = .fixed(input);
const fits = try readRequest(&ok, a, .{ .max_file_bytes = 8, .max_request_bytes = 12 }, &progress);
try testing.expectEqualStrings("ghijkl", fits.trailers[1].?);
try testing.expectEqual(@as(u64, 12), progress.bytes.load(.monotonic));
try testing.expect(!progress.in_trailers.load(.monotonic));
}
test "writeRequest and parseHeader: what a client writes reads back as the same request" {
var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
defer arena_state.deinit();
const a = arena_state.allocator();
const edits = [_]Edit{
.{ .put = .{ .name = "d/\"q\".txt", .size = 0, .hash = repeat("0", 64) } },
.{ .delete = "gone" },
.{ .rename = .{ .old = "a", .new = "b\u{e9}" } },
};
const commits = [_]Commit{
.{ .key = "k", .user = "u", .comment = "line\nnext", .time = .{ .epoch = 12, .ms = 3 }, .allow_fork = true, .edits = &edits },
.{ .parent = repeat("f", 64), .branch = "b", .user = "u", .comment = "c", .time = .{ .epoch = 0, .ms = 0 }, .edits = &.{} },
};
const requests = [_]Request{
.{ .id = 1, .op = .{ .hello = 1 } },
.{ .id = null, .op = .{ .commit = commits[0] } },
.{ .id = -4, .op = .{ .batch = &commits } },
.{ .id = 9, .op = .{ .feed = "0" } },
.{ .id = 10, .op = .{ .get = .{ .one = "ab" } } },
.{ .id = 11, .op = .{ .get = .{ .many = &.{ "a", "b" } } } },
.{ .id = 12, .op = .status },
};
for (requests) |req| {
var aw: std.Io.Writer.Allocating = .init(a);
try writeRequest(&aw.writer, req);
const line = aw.written();
try testing.expectEqual(@as(u8, '\n'), line[line.len - 1]);
try testing.expect(std.mem.indexOfScalar(u8, line[0 .. line.len - 1], '\n') == null);
const back = try parseOk(a, line[0 .. line.len - 1]);
try testing.expect(eqlRequest(req, back));
}
}
const FakeBackend = struct {
commits_seen: usize = 0,
fn backend(self: *FakeBackend) Backend {
return .{ .ctx = self, .vtable = &.{ .hello = hello, .commit = commitFn, .feed = feedFn, .get = getFn, .status = statusFn } };
}
fn hello(_: *anyopaque, _: std.mem.Allocator) Answer(Hello) {
return .{ .ok = .{ .zit = "0.0.0", .store = "/s", .tip = .{ .branches = &.{}, .cursor = "0" } } };
}
fn commitFn(ctx: *anyopaque, _: std.mem.Allocator, works: []const CommitWork, results: []ItemResult) void {
const self: *FakeBackend = @ptrCast(@alignCast(ctx));
self.commits_seen += works.len;
for (results) |*r| r.* = .{ .ok = .{ .outcome = .recorded, .hash = "h", .parent = null, .branch = "trunk", .forked = false } };
}
fn feedFn(_: *anyopaque, _: std.mem.Allocator, cursor: []const u8) Answer(feed.Feed) {
return .{ .ok = .{ .cursor = cursor, .changes = &.{}, .forks = &.{} } };
}
fn getFn(_: *anyopaque, _: std.mem.Allocator, uuid: []const u8) Answer([]const u8) {
if (std.mem.eql(u8, uuid, "here")) return .{ .ok = "BYTES" };
return .{ .fault = .{ .code = "content_absent", .exit = exit_absent, .message = "purged" } };
}
fn statusFn(_: *anyopaque, _: std.mem.Allocator) Answer(feed.Status) {
return .{ .fault = .{ .code = "storage", .exit = exit_failure, .message = "x" } };
}
};
fn dispatchLine(a: std.mem.Allocator, fake: *FakeBackend, input: []const u8, limits: Limits) ![]const u8 {
var r: std.Io.Reader = .fixed(input);
const incoming = try readRequest(&r, a, limits, null);
var aw: std.Io.Writer.Allocating = .init(a);
try dispatch(a, fake.backend(), &incoming, limits, &aw.writer);
return aw.written();
}
test "dispatch: replies carry the documented keys, in order" {
var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
defer arena_state.deinit();
const a = arena_state.allocator();
var fake: FakeBackend = .{};
try testing.expectEqualStrings(
"{\"id\":1,\"ok\":true,\"protocol\":1,\"zit\":\"0.0.0\",\"store\":\"/s\",\"tip\":{\"branches\":[],\"cursor\":\"0\"}}\n",
try dispatchLine(a, &fake, "{\"id\":1,\"op\":\"hello\",\"protocol\":1}\n", .{}),
);
try testing.expectEqualStrings(
"{\"id\":2,\"ok\":true,\"key\":\"k\",\"outcome\":\"recorded\",\"hash\":\"h\",\"parent\":null,\"branch\":\"trunk\",\"forked\":false}\n",
try dispatchLine(a, &fake, "{\"id\":2,\"op\":\"commit\",\"key\":\"k\",\"user\":\"e\",\"comment\":\"c\",\"time\":\"2026-01-01T00:00:00.000Z\",\"edits\":[{\"put\":{\"name\":\"a\",\"size\":1}}]}\nx", .{}),
);
try testing.expectEqualStrings(
"{\"id\":3,\"ok\":true,\"items\":[{\"uuid\":\"here\",\"size\":5},{\"uuid\":\"gone\",\"error\":{\"code\":\"content_absent\",\"exit\":8,\"message\":\"purged\"}}]}\nBYTES",
try dispatchLine(a, &fake, "{\"id\":3,\"op\":\"get\",\"uuids\":[\"here\",\"gone\"]}\n", .{}),
);
try testing.expectEqualStrings(
"{\"id\":4,\"ok\":false,\"error\":{\"code\":\"storage\",\"exit\":1,\"message\":\"x\"}}\n",
try dispatchLine(a, &fake, "{\"id\":4,\"op\":\"status\"}\n", .{}),
);
try testing.expectEqualStrings(
"{\"id\":5,\"ok\":false,\"error\":{\"code\":\"bad_request\",\"exit\":2,\"message\":\"this daemon speaks protocol 1, not 2\"}}\n",
try dispatchLine(a, &fake, "{\"id\":5,\"op\":\"hello\",\"protocol\":2}\n", .{}),
);
}
test "dispatch: a refused batch item never reaches the backend and never blocks the rest" {
var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
defer arena_state.deinit();
const a = arena_state.allocator();
var fake: FakeBackend = .{};
const hdr = "{\"id\":6,\"op\":\"batch\",\"commits\":[" ++
"{\"key\":\"big\",\"user\":\"e\",\"comment\":\"c\",\"time\":\"2026-01-01T00:00:00.000Z\",\"edits\":[{\"put\":{\"name\":\"a\",\"size\":5}}]}," ++
"{\"key\":\"bad\",\"user\":\"e\",\"comment\":\"c\",\"time\":\"2026-01-01T00:00:00.000Z\",\"edits\":[{\"put\":{\"name\":\"b\",\"size\":1,\"hash\":\"" ++ repeat("0", 64) ++ "\"}}]}," ++
"{\"key\":\"fine\",\"user\":\"e\",\"comment\":\"c\",\"time\":\"2026-01-01T00:00:00.000Z\",\"edits\":[{\"put\":{\"name\":\"c\",\"size\":1}}]}]}\n";
const reply = try dispatchLine(a, &fake, hdr ++ "12345yz", .{ .max_file_bytes = 4 });
try testing.expectEqual(@as(usize, 1), fake.commits_seen);
const v = try std.json.parseFromSliceLeaky(std.json.Value, a, reply, .{});
const results = v.object.get("results").?.array.items;
try testing.expectEqualStrings("too_large", results[0].object.get("error").?.object.get("code").?.string);
try testing.expectEqual(@as(i64, 1), results[0].object.get("error").?.object.get("exit").?.integer);
try testing.expectEqualStrings("hash_mismatch", results[1].object.get("error").?.object.get("code").?.string);
try testing.expectEqual(@as(i64, 2), results[1].object.get("error").?.object.get("exit").?.integer);
try testing.expectEqualStrings("fine", results[2].object.get("key").?.string);
try testing.expect(results[2].object.get("ok").?.bool);
}
test "coreFault and feedFault: error names become snake_case codes with the CLI's exits" {
try testing.expectEqualStrings("store_busy", coreFault(error.StoreBusy, "").code);
try testing.expectEqualStrings("not_a_repo", coreFault(error.NotARepo, "").code);
try testing.expectEqualStrings("too_large", coreFault(error.TooLarge, "").code);
try testing.expectEqual(exit_failure, coreFault(error.NotTip, "").exit);
try testing.expectEqual(exit_cursor_invalid, feedFault(error.CursorInvalid, "").exit);
try testing.expectEqualStrings("bad_request", feedFault(error.BadName, "").code);
try testing.expectEqualStrings("out_of_memory", feedFault(error.OutOfMemory, "").code);
}
test "text: store strings that are not UTF-8 still make one JSON string" {
var aw: std.Io.Writer.Allocating = .init(testing.allocator);
defer aw.deinit();
var jw: Json = .{ .writer = &aw.writer };
try storeText(&jw, "a\xffb\"\xe9");
try testing.expectEqualStrings("\"a\\ufffdb\\\"\\ufffd\"", aw.written());
}