//! The read side of a resident store: what changed since a cursor, which
//! branches have which open leaves, and an artifact's bytes by name. The
//! daemon answers `feed`, `status` and `get` from here; a C ABI can too,
//! since nothing below reaches the verb layer.
//!
//! A `Reader` holds one repository view and the store file's identity as of
//! that view. Every call first asks whether the store moved (another writer's
//! append, this process's own commit, a `pack` that installed a new file);
//! only then is the view reopened. Asking costs one `stat`, so a client that
//! polls an idle store pays a syscall, not a derive.
//!
//! The cursor is `zit frames`' own: `COUNT TOKEN`, where COUNT is how many
//! stored artifacts in persisted order the reader has consumed and TOKEN is
//! the log key of that prefix. A rewrite relocates frames and fails the
//! token's check, which is how a consumer learns to start again from "0".
const std = @import("std");
const repeat = @import("../repeat.zig").repeat;
const artifact = @import("../artifact.zig");
const logstore = @import("../logstore.zig");
const repostate = @import("../repostate.zig");
const repository = @import("../repository.zig");
const Repository = repository.Repository;
const repoview = @import("../repoview.zig");
const thinned = @import("../thinned.zig");
const checkin = @import("checkin.zig");
pub const Diag = checkin.Diag;
/// `COUNT ` plus the four dash-joined log-key fields, each at most 20 digits.
pub const cursor_max: usize = 20 + 1 + 4 * 21;
pub const Error = error{
OutOfMemory,
/// The store could not be read.
Storage,
/// The store does not decode.
Corrupt,
/// The store is not one this build can open (a pending import).
CannotOpen,
NotARepo,
/// The cursor is not one this store issued, or the store was rewritten
/// since it was.
CursorInvalid,
/// An artifact name that is not 40 or 64 lowercase hex digits.
BadName,
};
/// What a check-in did to one path, in `log PATH -F json`'s words: a rename
/// is a `modified` row on the new name carrying `prior`, beside a `removed`
/// row on the old one.
pub const State = enum { added, modified, removed };
pub const FileRow = struct {
name: []const u8,
/// The name this check-in renamed the file from; null on every other row.
prior: ?[]const u8,
/// The file's content in the check-in; null for a removal.
uuid: ?[]const u8,
state: State,
};
pub const Change = struct {
checkin: []const u8,
/// The primary parent; null for a root check-in.
parent: ?[]const u8,
branch: []const u8,
user: ?[]const u8,
comment: ?[]const u8,
/// Milliseconds since the epoch; null when the view has no event for it.
time_ms: ?repostate.Mtime,
/// Null when the change set cannot be computed from what the store holds
/// (an absent parent, an unresolvable baseline).
files: ?[]const FileRow,
};
pub const Fork = struct {
branch: []const u8,
/// Newest first.
leaves: []const []const u8,
};
pub const Feed = struct {
cursor: []const u8,
changes: []const Change,
forks: []const Fork,
};
pub const Branch = struct {
name: []const u8,
/// The newest open leaf.
tip: []const u8,
/// Every open leaf, newest first.
leaves: []const []const u8,
};
pub const Status = struct {
branches: []const Branch,
cursor: []const u8,
};
/// Why an artifact's bytes are not here.
pub const Absence = enum {
/// `purge artifacts` thinned it.
purged,
shunned,
/// The store's cone or horizon leaves it out.
elided,
/// No record of it at all: never stored, or lost.
missing,
};
pub const Got = union(enum) {
/// Owned by the allocator `get` was given.
bytes: []u8,
absent: Absence,
};
/// Whether the store's cone or horizon elides an artifact. The classifier
/// walks the whole store and lives with the sync machinery, so a caller that
/// has it supplies it; without one, an elided artifact reads as missing.
pub const ElidedHook = struct {
ctx: ?*anyopaque = null,
elided: *const fn (ctx: ?*anyopaque, gpa: std.mem.Allocator, io: std.Io, store_dir: std.Io.Dir, name: []const u8) bool,
};
/// Pinned once opened: the repository view points into it.
pub const Reader = struct {
gpa: std.mem.Allocator,
io: std.Io,
root_dir: std.Io.Dir,
repo: Repository,
repo_open: bool,
/// The store file the view was opened on: a different inode or size at
/// the store's path means the view is behind.
seen_inode: std.Io.File.INode,
seen_size: u64,
/// The view's stored artifacts in persisted order, phantoms excluded:
/// the clock a cursor counts on.
persisted: []logstore.Rid,
cursor_buf: [cursor_max]u8,
cursor_len: usize,
elided_hook: ?ElidedHook,
/// Why an artifact is absent, answered once per view: the classes are
/// store-wide facts (the thinned list, the shun table, the cone), and the
/// elision answer classifies the whole store to give it.
absence_memo: std.StringHashMapUnmanaged(Absence),
thinned_set: ?thinned.Set,
/// Open the store under `root_dir`, which is borrowed and must outlive
/// the reader.
pub fn open(self: *Reader, gpa: std.mem.Allocator, io: std.Io, root_dir: std.Io.Dir, elided_hook: ?ElidedHook, diag: *Diag) Error!void {
diag.clear();
self.* = .{
.gpa = gpa,
.io = io,
.root_dir = root_dir,
.repo = undefined,
.repo_open = false,
.seen_inode = undefined,
.seen_size = 0,
.persisted = &.{},
.cursor_buf = undefined,
.cursor_len = 0,
.elided_hook = elided_hook,
.absence_memo = .empty,
.thinned_set = null,
};
try self.openView(diag);
}
pub fn deinit(self: *Reader) void {
self.closeView();
self.* = undefined;
}
/// The cursor naming everything the current view holds.
pub fn cursor(self: *const Reader) []const u8 {
return self.cursor_buf[0..self.cursor_len];
}
fn openView(self: *Reader, diag: *Diag) Error!void {
std.debug.assert(!self.repo_open);
self.repo.open(self.gpa, self.io, self.root_dir, .{ .mode = .full }) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.NotARepo => return error.NotARepo,
error.CannotRead => {
diag.set("cannot read repository: input/output error", .{});
return error.Storage;
},
error.Corrupt => {
diag.set("cannot read repository: repository is corrupt", .{});
return error.Corrupt;
},
error.ImportPending => {
diag.set("an incremental import is pending; run 'zit import-git --recover DIR'", .{});
return error.CannotOpen;
},
};
self.repo_open = true;
errdefer self.closeView();
const st = self.repo.snapshot_file.stat(self.io) catch return storageFault(diag);
self.seen_inode = st.inode;
self.seen_size = self.repo.snapshot_size;
const log = try self.logOf(diag);
self.persisted = try log.ridsInPersistedOrder(self.gpa);
const key = cursorKey(log, self.persisted) catch return storageFault(diag);
const text = formatCursor(&self.cursor_buf, self.persisted.len, key);
self.cursor_len = text.len;
}
/// The log key of the persisted prefix. When its last artifact is the
/// log's last frame, as after an ordinary commit, that is the key of the
/// whole log, read from the tail; otherwise the frames are walked to it.
fn cursorKey(log: *logstore.LogStore, persisted: []const logstore.Rid) logstore.LogStore.KeyError!logstore.LogStore.LogKey {
if (persisted.len == 0) return log.logKey();
const last = persisted[persisted.len - 1];
if (log.offsetOf(last) + log.frameSpanOf(last) == log.end) return log.logKey();
return log.logKeyAt(last);
}
fn closeView(self: *Reader) void {
var names = self.absence_memo.keyIterator();
while (names.next()) |n| self.gpa.free(n.*);
self.absence_memo.clearAndFree(self.gpa);
if (self.thinned_set) |*t| t.deinit(self.gpa);
self.thinned_set = null;
if (!self.repo_open) return;
self.gpa.free(self.persisted);
self.persisted = &.{};
self.repo.deinit(self.gpa);
self.repo_open = false;
}
fn logOf(self: *Reader, diag: *Diag) Error!*logstore.LogStore {
return self.repo.logStore(self.gpa) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Corrupt => {
diag.set("cannot read repository: repository is corrupt", .{});
return error.Corrupt;
},
else => return storageFault(diag),
};
}
/// True when the store file at the store's path is still the one the
/// view was opened on, at the size it had: appends only grow a store, and
/// a rewrite installs a new file. One `stat`, no allocation.
pub fn current(self: *Reader) bool {
if (!self.repo_open) return false;
const st = self.repo.root_dir.statFile(self.io, repository.repo_filename, .{}) catch return false;
return st.inode == self.seen_inode and st.size == self.seen_size;
}
/// Bring the view up to the store as it is now.
pub fn refresh(self: *Reader, diag: *Diag) Error!void {
diag.clear();
if (self.current()) return;
self.closeView();
try self.openView(diag);
}
/// The nothing-new answer, when it applies: `since` names everything the
/// store holds and the store has not moved. No allocation, so a client
/// polling an idle store costs one `stat`.
pub fn unchangedSince(self: *Reader, since: []const u8) bool {
return std.mem.eql(u8, since, self.cursor()) and self.current();
}
/// The check-ins stored after `since` (`"0"` for all of them), in store
/// order, each with what it did to each path, and every branch with more
/// than one open leaf. Everything returned lives on `arena`.
pub fn feed(self: *Reader, arena: std.mem.Allocator, since: []const u8, diag: *Diag) Error!Feed {
if (self.unchangedSince(since)) return .{ .cursor = since, .changes = &.{}, .forks = &.{} };
try self.refresh(diag);
const log = try self.logOf(diag);
const count = try self.consumed(log, since, diag);
var collect: Collect = .{ .arena = arena, .reader = self, .log = log };
repoview.eachChangeSetOf(self.gpa, log, self.persisted[count..], collect.sink()) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
};
const forks = try self.forksOf(arena);
return .{
.cursor = try arena.dupe(u8, self.cursor()),
.changes = collect.changes.items,
.forks = forks,
};
}
/// How many persisted artifacts `since` has consumed, once its token
/// proves this store issued it over the frames it still holds.
fn consumed(self: *Reader, log: *logstore.LogStore, since: []const u8, diag: *Diag) Error!usize {
const parsed = parseCursor(since) orelse {
diag.set("cursor invalid: the cursor is not one this store issued; re-read from \"0\"", .{});
return error.CursorInvalid;
};
if (parsed.count > self.persisted.len) {
diag.set("cursor invalid: the cursor names artifacts this store does not hold; re-read from \"0\"", .{});
return error.CursorInvalid;
}
if (parsed.key) |key| if (!logstore.LogStore.validatePrefix(log.file, key)) {
diag.set("cursor invalid: the store was rewritten since the cursor was issued; re-read from \"0\"", .{});
return error.CursorInvalid;
};
return @intCast(parsed.count);
}
/// Every open branch's leaves, and the cursor they are true as of.
pub fn status(self: *Reader, arena: std.mem.Allocator, diag: *Diag) Error!Status {
try self.refresh(diag);
return .{ .branches = try self.branchesOf(arena), .cursor = try arena.dupe(u8, self.cursor()) };
}
fn forksOf(self: *Reader, arena: std.mem.Allocator) error{OutOfMemory}![]const Fork {
const branches = try self.branchesOf(arena);
var forks: std.ArrayList(Fork) = .empty;
for (branches) |b| {
if (b.leaves.len > 1) try forks.append(arena, .{ .branch = b.name, .leaves = b.leaves });
}
return forks.items;
}
/// Open leaves grouped by branch, branches by name, leaves newest first.
fn branchesOf(self: *Reader, arena: std.mem.Allocator) error{OutOfMemory}![]const Branch {
const state = self.repo.state();
const Leaf = struct { branch: []const u8, rid: repostate.Rid, mtime: repostate.Mtime };
var open_leaves: std.ArrayList(Leaf) = .empty;
var it = state.leaves();
while (it.next()) |rid| {
if (!state.isOpenLeaf(rid)) continue;
const branch = state.activeTagValue(rid, "branch") orelse continue;
try open_leaves.append(arena, .{ .branch = branch, .rid = rid, .mtime = state.mtimeOf(rid) orelse 0 });
}
const Order = struct {
state: *const repostate.RepoState,
fn lessThan(ctx: @This(), a: Leaf, b: Leaf) bool {
switch (std.mem.order(u8, a.branch, b.branch)) {
.lt => return true,
.gt => return false,
.eq => {},
}
if (a.mtime != b.mtime) return a.mtime > b.mtime;
return std.mem.order(u8, ctx.state.hashOf(a.rid), ctx.state.hashOf(b.rid)) == .lt;
}
};
std.mem.sort(Leaf, open_leaves.items, Order{ .state = state }, Order.lessThan);
var branches: std.ArrayList(Branch) = .empty;
var i: usize = 0;
while (i < open_leaves.items.len) {
var j = i;
while (j < open_leaves.items.len and std.mem.eql(u8, open_leaves.items[j].branch, open_leaves.items[i].branch)) j += 1;
const leaves = try arena.alloc([]const u8, j - i);
for (open_leaves.items[i..j], leaves) |l, *out| out.* = try arena.dupe(u8, state.hashOf(l.rid));
try branches.append(arena, .{ .name = try arena.dupe(u8, open_leaves.items[i].branch), .tip = leaves[0], .leaves = leaves });
i = j;
}
return branches.items;
}
/// The artifact named `name`, verified against its name, or why it is
/// not here.
pub fn get(self: *Reader, gpa: std.mem.Allocator, name: []const u8, diag: *Diag) Error!Got {
if (!isName(name)) return error.BadName;
try self.refresh(diag);
if (self.repo.blob(gpa, name)) |found| {
if (found) |bytes| return .{ .bytes = bytes };
} else |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Corrupt => {
diag.set("stored artifact {s} does not hash to its name", .{name});
return error.Corrupt;
},
else => return storageFault(diag),
}
return .{ .absent = try self.absenceOf(name, diag) };
}
fn absenceOf(self: *Reader, name: []const u8, diag: *Diag) Error!Absence {
if (self.absence_memo.get(name)) |known| return known;
const why = try self.classifyAbsence(name, diag);
const owned = try self.gpa.dupe(u8, name);
errdefer self.gpa.free(owned);
try self.absence_memo.put(self.gpa, owned, why);
return why;
}
fn classifyAbsence(self: *Reader, name: []const u8, diag: *Diag) Error!Absence {
if (self.thinned_set == null) self.thinned_set = thinned.load(self.gpa, self.io, self.repo.root_dir) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => .{},
};
if (self.thinned_set.?.contains(name)) return .purged;
var digest_buf: [32]u8 = undefined;
const digest = digest_buf[0 .. name.len / 2];
_ = std.fmt.hexToBytes(digest, name) catch unreachable; // `isName` checked it
const log = try self.logOf(diag);
if (log.isShunned(digest)) return .shunned;
if (self.elided_hook) |hook| if (hook.elided(hook.ctx, self.gpa, self.io, self.repo.root_dir, name)) return .elided;
return .missing;
}
/// The derived-view rid of a log rid: the two spaces differ (the view
/// interns phantoms by first sight), and the name is what both agree on.
fn viewRid(self: *Reader, log: *logstore.LogStore, log_rid: logstore.Rid, hex_buf: *[64]u8) ?repostate.Rid {
const digest = log.hashOf(log_rid);
if (digest.len != 32 and digest.len != 20) return null;
const hex = std.fmt.bufPrint(hex_buf, "{x}", .{digest}) catch unreachable;
return self.repo.lookup(hex);
}
};
/// The change-set sink `feed` hands the walk: one `Change` per check-in.
const Collect = struct {
arena: std.mem.Allocator,
reader: *Reader,
log: *logstore.LogStore,
changes: std.ArrayList(Change) = .empty,
fn sink(self: *Collect) repoview.ChangeSetSink {
return .{ .ctx = self, .emit = emit };
}
fn emit(ctx: *anyopaque, rid: logstore.Rid, files: ?[]const repoview.ChangedFile) error{OutOfMemory}!void {
const self: *Collect = @ptrCast(@alignCast(ctx));
const a = self.arena;
var hex_buf: [64]u8 = undefined;
const view_rid = self.reader.viewRid(self.log, rid, &hex_buf);
const hash = try a.dupe(u8, std.fmt.bufPrint(&hex_buf, "{x}", .{self.log.hashOf(rid)}) catch unreachable);
var change: Change = .{
.checkin = hash,
.parent = null,
.branch = "",
.user = null,
.comment = null,
.time_ms = null,
.files = null,
};
if (view_rid) |vr| {
const repo = &self.reader.repo;
const state = repo.state();
if (state.primaryParentOf(vr)) |p| change.parent = try a.dupe(u8, repo.hashOf(p));
change.branch = try a.dupe(u8, repo.branchOf(vr));
if (state.userOf(vr)) |u| change.user = try a.dupe(u8, u);
if (state.commentOf(vr)) |c| change.comment = try a.dupe(u8, c);
change.time_ms = state.mtimeOf(vr);
}
// The view leaves a comment unread when it spilled it; the manifest
// has it.
if (change.comment == null or change.user == null) self.fromManifest(rid, &change) catch |e| {
if (e == error.OutOfMemory) return error.OutOfMemory;
};
if (files) |fs| {
const rows = try a.alloc(FileRow, fs.len);
for (fs, rows) |f, *row| row.* = .{
.name = try a.dupe(u8, f.path),
.prior = if (f.prior) |p| try a.dupe(u8, p) else null,
.uuid = if (f.hash) |h| try a.dupe(u8, h) else null,
.state = switch (f.disp) {
.added => if (f.prior != null) .modified else .added,
.modified => .modified,
.deleted => .removed,
},
};
change.files = rows;
}
try self.changes.append(a, change);
}
fn fromManifest(self: *Collect, rid: logstore.Rid, change: *Change) error{ OutOfMemory, Unreadable }!void {
const gpa = self.reader.gpa;
const bytes = (self.log.content(gpa, rid) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Unreadable,
}) orelse return error.Unreadable;
defer gpa.free(bytes);
var loaded = artifact.parseForImport(gpa, bytes) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Unreadable,
};
defer loaded.deinit(gpa);
const m = &loaded.manifest;
if (change.comment == null) if (m.c) |c| {
change.comment = try self.arena.dupe(u8, c.comment);
};
if (change.user == null) if (m.u) |u| {
change.user = try self.arena.dupe(u8, u.login);
};
}
};
fn storageFault(diag: *Diag) Error {
diag.set("cannot read repository: input/output error", .{});
return error.Storage;
}
/// A full artifact name: 40 (SHA-1) or 64 (SHA3-256) lowercase hex digits.
pub fn isName(name: []const u8) bool {
if (name.len != 40 and name.len != 64) return false;
for (name) |c| switch (c) {
'0'...'9', 'a'...'f' => {},
else => return false,
};
return true;
}
pub const ParsedCursor = struct {
count: u64,
/// Null for the bootstrap cursor `"0"`, which needs no proof.
key: ?logstore.LogStore.LogKey,
};
/// `"0"`, or `COUNT TOKEN` as `formatCursor` writes it. Null on any other
/// shape, so a cursor is never half-understood.
pub fn parseCursor(text: []const u8) ?ParsedCursor {
if (std.mem.eql(u8, text, "0")) return .{ .count = 0, .key = null };
const space = std.mem.indexOfScalar(u8, text, ' ') orelse return null;
const count = parseDecimal(u64, text[0..space]) orelse return null;
var it = std.mem.splitScalar(u8, text[space + 1 ..], '-');
const end = parseDecimal(u64, it.next() orelse return null) orelse return null;
const frame_count = parseDecimal(u64, it.next() orelse return null) orelse return null;
const last_frame_offset = parseDecimal(u64, it.next() orelse return null) orelse return null;
const last_header_crc = parseDecimal(u32, it.next() orelse return null) orelse return null;
if (it.next() != null) return null;
return .{ .count = count, .key = .{
.end = end,
.frame_count = frame_count,
.last_frame_offset = last_frame_offset,
.last_header_crc = last_header_crc,
} };
}
/// Plain decimal digits only: `parseInt` would also take a sign or an
/// underscore, and a cursor has one spelling.
fn parseDecimal(comptime T: type, text: []const u8) ?T {
if (text.len == 0) return null;
for (text) |c| if (c < '0' or c > '9') return null;
return std.fmt.parseInt(T, text, 10) catch null;
}
/// `COUNT END-FRAMES-OFFSET-CRC`: the trailer `zit frames` prints after
/// `cursor`, so the same position can be handed to either.
pub fn formatCursor(buf: *[cursor_max]u8, count: usize, key: logstore.LogStore.LogKey) []const u8 {
return std.fmt.bufPrint(buf, "{d} {d}-{d}-{d}-{d}", .{
count, key.end, key.frame_count, key.last_frame_offset, key.last_header_crc,
}) catch unreachable;
}
test "cursor: what formatCursor writes parses back, and nothing else half-parses" {
var buf: [cursor_max]u8 = undefined;
const key: logstore.LogStore.LogKey = .{
.end = std.math.maxInt(u64),
.frame_count = std.math.maxInt(u64),
.last_frame_offset = std.math.maxInt(u64),
.last_header_crc = std.math.maxInt(u32),
};
const text = formatCursor(&buf, std.math.maxInt(u64), key);
const back = parseCursor(text).?;
try std.testing.expectEqual(@as(u64, std.math.maxInt(u64)), back.count);
try std.testing.expectEqual(key, back.key.?);
try std.testing.expectEqual(@as(u64, 0), parseCursor("0").?.count);
try std.testing.expect(parseCursor("0").?.key == null);
for ([_][]const u8{ "", "00", "1", "1 2-3-4", "1 2-3-4-5-6", "+1 2-3-4-5", "1 2-3-4-x", "1 2-3-4-5", "1 2-3-4-99999999999" }) |bad| {
try std.testing.expect(parseCursor(bad) == null);
}
}
test "isName: full lowercase hex of either width" {
try std.testing.expect(isName(repeat("a", 64)));
try std.testing.expect(isName(repeat("0", 40)));
try std.testing.expect(!isName(repeat("A", 64)));
try std.testing.expect(!isName(repeat("a", 63)));
try std.testing.expect(!isName(repeat("g", 40)));
}