Skip to content
zit Artifact [27ca96c385]
Login

Artifact [27ca96c385]

Check-ins Using Download Hex

Artifact 27ca96c3853e4dddd6e25db4253fa0be62c0295696a473b7c48913683b6d4e2f:

//! The cluster index.
//!
//! A CLUSTER artifact (artifact format §4.5.2/§4.7.11: M cards, closed by Z)
//! is `igot` in bulk. Once a peer folds an artifact into a cluster it stops
//! announcing that artifact individually; the cluster's M cards ARE the
//! announcement, and the cluster — itself an artifact — is announced by a
//! later cluster or, at the chain's newest end, individually. Bounded
//! announcement (sync §3.3; fossil's `send_unclustered`) therefore needs two
//! derived facts the log does not carry:
//!
//!   - which held artifacts are clusters (the index proper), and
//!   - from their M cards: which held artifacts some cluster already names
//!     (`clustered` — never announced individually), and which named
//!     artifacts are NOT held (`missing` — a promise a held cluster made
//!     that this repository never collected. Fossil recovers these from
//!     durable phantom rows; zit writes no phantom to disk, so it re-derives
//!     them here, at every sync open).
//!
//! The index is an advisory sidecar (`.zit-cache/clusters`), keyed to the log
//! prefix exactly like the dispositions map: adopted when the key validates,
//! extended over any appended tail, rebuilt by one full walk on a miss. The
//! persist path records clusters as it writes them, so a store that syncs
//! keeps its own index warm. Every consumer degrades to UNBOUNDED
//! announcement when the index is absent: the fallback costs bytes on the
//! wire, never correctness.

const std = @import("std");
const artifact = @import("artifact.zig");
const logstore = @import("logstore.zig");
const repostate = @import("repostate.zig");
const crc32 = @import("crc32.zig");

const Rid = repostate.Rid;

/// Evidence seam: how many held rids a full or tail walk had to read to find
/// clusters. Zero on an adopted sidecar — the pin asserts it.
pub var walk_rids: u64 = 0;

/// The two-byte reject. Cards are emitted in letter order and M is the
/// smallest letter a cluster may carry, so a cluster — and only a cluster —
/// opens with `M `. Full validation is `collectClusterMembers`.
pub fn isCluster(content: []const u8) bool {
    return content.len >= 2 and content[0] == 'M' and content[1] == ' ';
}

/// Validate-then-apply twin of `repostate.collectSubManifestRefs`: append
/// every M-card member of `content` to `out`, or append NOTHING and return
/// false when the card run is malformed (§4.5.2 — a cluster admits M and Z
/// only), so a malformed artifact that merely opens with an M card mints no
/// phantom. Members borrow `content`. A non-cluster is valid and contributes
/// nothing.
pub fn collectClusterMembers(
    gpa: std.mem.Allocator,
    content: []const u8,
    out: *std.ArrayListUnmanaged([]const u8),
) error{OutOfMemory}!bool {
    if (!isCluster(content)) return true;
    var cursor: usize = 0;
    while (cursor < content.len) {
        const end = std.mem.indexOfScalarPos(u8, content, cursor, '\n') orelse return false;
        const card = artifact.tokenizeCardLine(content[cursor .. end + 1]) orelse return false;
        switch (card.letter) {
            'M' => if (artifact.parseMCard(card) == null) return false,
            'Z' => {},
            else => return false,
        }
        cursor = end + 1;
    }
    std.debug.assert(cursor == content.len);
    cursor = 0;
    while (cursor < content.len) {
        const end = std.mem.indexOfScalarPos(u8, content, cursor, '\n').?;
        const card = artifact.tokenizeCardLine(content[cursor .. end + 1]).?;
        cursor = end + 1;
        if (card.letter != 'M') continue;
        try out.append(gpa, artifact.parseMCard(card).?.hash);
    }
    return true;
}

/// What a sync needs, derived from the index against one store. `clustered`
/// is indexed by the store's own rids (the index carries NAMES, so a store
/// whose rid space differs from the log's — an adapter, a rebuilt log — still
/// resolves correctly). `missing` is deduplicated; both it and
/// `cluster_names` are arena-owned copies, so nothing here borrows the store.
pub const Resolved = struct {
    clustered: std.DynamicBitSetUnmanaged,
    missing: []const []const u8,
    cluster_names: []const []const u8,
    arena: std.heap.ArenaAllocator,
    gpa: std.mem.Allocator,

    pub fn deinit(self: *Resolved) void {
        self.clustered.deinit(self.gpa);
        self.arena.deinit();
        self.* = undefined;
    }
};

/// Read every named cluster's M cards out of `store` and fold them into a
/// `Resolved`. A name the store cannot read (a phantom cluster, or one whose
/// content is unavailable) names nothing — the announcement simply stays
/// unbounded for what it would have covered. A held member sets its bit; an
/// unheld member is a `missing` promise.
pub fn resolve(
    gpa: std.mem.Allocator,
    store: repostate.ArtifactStore,
    cluster_names: []const []const u8,
) error{OutOfMemory}!Resolved {
    var arena = std.heap.ArenaAllocator.init(gpa);
    errdefer arena.deinit();
    const a = arena.allocator();
    var clustered = try std.DynamicBitSetUnmanaged.initEmpty(gpa, store.count() + 1);
    errdefer clustered.deinit(gpa);

    var missing_set: std.StringHashMapUnmanaged(void) = .empty;
    defer missing_set.deinit(gpa);
    var missing: std.ArrayListUnmanaged([]const u8) = .empty;
    var names: std.ArrayListUnmanaged([]const u8) = .empty;
    var members: std.ArrayListUnmanaged([]const u8) = .empty;
    defer members.deinit(gpa);

    for (cluster_names) |name| {
        const rid = store.lookup(name) orelse continue;
        if (!store.held(rid)) continue;
        const content = store.content(rid) orelse continue;
        members.clearRetainingCapacity();
        if (!try collectClusterMembers(gpa, content, &members)) continue;
        try names.append(a, try a.dupe(u8, name));
        for (members.items) |m| {
            if (store.lookup(m)) |mr| {
                if (store.held(mr)) {
                    if (mr < clustered.capacity()) clustered.set(mr);
                    continue;
                }
            }
            const gop = try missing_set.getOrPut(gpa, m);
            if (gop.found_existing) continue;
            const owned = try a.dupe(u8, m);
            gop.key_ptr.* = owned;
            try missing.append(a, owned);
        }
    }
    return .{
        .clustered = clustered,
        .missing = missing.items,
        .cluster_names = names.items,
        .arena = arena,
        .gpa = gpa,
    };
}

/// The walk: read every content-bearing log rid in (`after`, entryCount] and
/// append the wire names of those that are clusters. This is the one place
/// the index costs O(store) content reads — once on a miss, and over just the
/// appended tail thereafter. It reads through the log with a bounded
/// plaintext cache rather than through a store adapter, because the sync
/// adapters memoise every read for the round and a whole-store walk through
/// them would leave the whole store resident (a known OOM shape). Names are
/// duped into `arena`; a frame that fails to read is skipped (its cluster, if
/// it was one, simply stays unindexed — unbounded, never wrong).
pub fn walkClusters(
    gpa: std.mem.Allocator,
    arena: std.mem.Allocator,
    log: *logstore.LogStore,
    after: Rid,
    out: *std.ArrayListUnmanaged([]const u8),
) error{OutOfMemory}!void {
    var cache = logstore.PlaintextCache.init();
    defer cache.deinit(gpa);
    const n: Rid = @intCast(log.entryCount());
    var rid: Rid = after + 1;
    while (rid <= n) : (rid += 1) {
        if (log.isPhantom(rid)) continue;
        walk_rids += 1;
        const plain = (log.contentCached(gpa, rid, &cache) catch |e| switch (e) {
            error.OutOfMemory => return error.OutOfMemory,
            error.Storage, error.Corrupt => continue,
        }) orelse continue;
        defer gpa.free(plain);
        if (!isCluster(plain)) continue;
        const digest = log.hashOf(rid);
        const hex: []const u8 = if (digest.len == 32)
            &std.fmt.bytesToHex(digest[0..32].*, .lower)
        else
            &std.fmt.bytesToHex(digest[0..20].*, .lower);
        try out.append(arena, try arena.dupe(u8, hex));
    }
}

// ---------------------------------------------------------------------------
// Minting (fossil `xfer.c create_cluster`, run at its pull-reply point).
// ---------------------------------------------------------------------------

/// Mint only once this many held artifacts are unclustered; below it they
/// are announced individually, which is cheaper than a cluster would be.
pub const mint_threshold: usize = 100;
/// M cards per cluster. Fossil closes a cluster at this size only when more
/// than `mint_threshold` remain, so the last cluster of a batch absorbs a
/// short tail rather than leaving it to be announced individually.
pub const cluster_capacity: usize = 800;

fn nameLessThan(_: void, a: []const u8, b: []const u8) bool {
    return std.mem.lessThan(u8, a, b);
}

/// Fold the held-and-unclustered artifacts of `store` into cluster bodies:
/// wire names in strictly ascending order (§4.5.2), `cluster_capacity` per
/// cluster, closed by the Z card (md5 of the body, §4.7.19). Appends nothing
/// below `mint_threshold`. Bodies are gpa-owned; the caller stores them as
/// artifacts. The newest clusters are themselves unclustered, so the next
/// mint names them — that is how the chain grows.
pub fn mintClusters(
    gpa: std.mem.Allocator,
    store: repostate.ArtifactStore,
    clustered: *const std.DynamicBitSetUnmanaged,
    out: *std.ArrayListUnmanaged([]u8),
) error{OutOfMemory}!void {
    var names: std.ArrayListUnmanaged([]const u8) = .empty;
    defer names.deinit(gpa);
    const n: Rid = @intCast(store.count());
    var rid: Rid = 1;
    while (rid <= n) : (rid += 1) {
        if (!store.held(rid)) continue;
        if (rid < clustered.capacity() and clustered.isSet(rid)) continue;
        try names.append(gpa, store.hashOf(rid));
    }
    if (names.items.len < mint_threshold) return;
    std.mem.sort([]const u8, names.items, {}, nameLessThan);

    var i: usize = 0;
    while (i < names.items.len) {
        const remaining = names.items.len - i;
        const take = if (remaining > cluster_capacity + mint_threshold) cluster_capacity else remaining;
        var body: std.ArrayListUnmanaged(u8) = .empty;
        errdefer body.deinit(gpa);
        for (names.items[i .. i + take]) |nm| {
            try body.appendSlice(gpa, "M ");
            try body.appendSlice(gpa, nm);
            try body.append(gpa, '\n');
        }
        var digest: [16]u8 = undefined;
        std.crypto.hash.Md5.hash(body.items, &digest, .{});
        try body.appendSlice(gpa, "Z ");
        try body.appendSlice(gpa, &std.fmt.bytesToHex(digest, .lower));
        try body.append(gpa, '\n');
        try out.append(gpa, try body.toOwnedSlice(gpa));
        i += take;
    }
}

// ---------------------------------------------------------------------------
// Sidecar codec. `magic | version | LogKey | n_rids | count | (len name)* | crc`.
// `n_rids` is the store entry count the names cover, so an adopter knows
// where its tail walk begins; the LogKey is what `validatePrefix` checks.
// ---------------------------------------------------------------------------

const magic = "ZITCLIX\x00";
const version: u16 = 1;
const header_len = 8 + 2 + 8 + 8 + 8 + 4 + 8 + 4;

pub fn serialize(
    gpa: std.mem.Allocator,
    log: *const logstore.LogStore,
    n_rids: u64,
    cluster_names: []const []const u8,
) error{OutOfMemory}!?[]u8 {
    const key = log.logKey() catch return null;
    var body: usize = 0;
    for (cluster_names) |nm| {
        if (nm.len > 255) return null;
        body += 1 + nm.len;
    }
    const blob = try gpa.alloc(u8, header_len + body + 4);
    errdefer gpa.free(blob);
    @memcpy(blob[0..8], magic);
    std.mem.writeInt(u16, blob[8..10], version, .little);
    std.mem.writeInt(u64, blob[10..18], key.end, .little);
    std.mem.writeInt(u64, blob[18..26], key.frame_count, .little);
    std.mem.writeInt(u64, blob[26..34], key.last_frame_offset, .little);
    std.mem.writeInt(u32, blob[34..38], key.last_header_crc, .little);
    std.mem.writeInt(u64, blob[38..46], n_rids, .little);
    std.mem.writeInt(u32, blob[46..50], @intCast(cluster_names.len), .little);
    var pos: usize = header_len;
    for (cluster_names) |nm| {
        blob[pos] = @intCast(nm.len);
        @memcpy(blob[pos + 1 ..][0..nm.len], nm);
        pos += 1 + nm.len;
    }
    std.debug.assert(pos == blob.len - 4);
    var crc = crc32.Crc32.init();
    crc.update(blob[0 .. blob.len - 4]);
    std.mem.writeInt(u32, blob[blob.len - 4 ..][0..4], crc.final(), .little);
    return blob;
}

pub const Parsed = struct {
    /// Store entries the names cover; a tail walk begins after this rid.
    n_rids: u64,
    /// Borrowed from the blob.
    names: []const []const u8,
};

/// Adopt a sidecar blob against `log`: magic, version, CRC, and the log
/// prefix it was keyed to must all hold, or the answer is null and the
/// caller rebuilds. `names` are allocated from `arena`.
pub fn parse(arena: std.mem.Allocator, blob: []const u8, log: *const logstore.LogStore) error{OutOfMemory}!?Parsed {
    if (blob.len < header_len + 4) return null;
    if (!std.mem.eql(u8, blob[0..8], magic)) return null;
    if (std.mem.readInt(u16, blob[8..10], .little) != version) return null;
    var crc = crc32.Crc32.init();
    crc.update(blob[0 .. blob.len - 4]);
    if (std.mem.readInt(u32, blob[blob.len - 4 ..][0..4], .little) != crc.final()) return null;
    const key = logstore.LogStore.LogKey{
        .end = std.mem.readInt(u64, blob[10..18], .little),
        .frame_count = std.mem.readInt(u64, blob[18..26], .little),
        .last_frame_offset = std.mem.readInt(u64, blob[26..34], .little),
        .last_header_crc = std.mem.readInt(u32, blob[34..38], .little),
    };
    if (!logstore.LogStore.validatePrefix(log.file, key)) return null;
    const n_rids = std.mem.readInt(u64, blob[38..46], .little);
    const count = std.mem.readInt(u32, blob[46..50], .little);
    var names = try arena.alloc([]const u8, count);
    var pos: usize = header_len;
    var i: usize = 0;
    while (i < count) : (i += 1) {
        if (pos >= blob.len - 4) return null;
        const len: usize = blob[pos];
        if (pos + 1 + len > blob.len - 4) return null;
        names[i] = blob[pos + 1 ..][0..len];
        pos += 1 + len;
    }
    if (pos != blob.len - 4) return null;
    return .{ .n_rids = n_rids, .names = names };
}

test "collectClusterMembers: a cluster yields its M members; a malformed one yields nothing" {
    const gpa = std.testing.allocator;
    var out: std.ArrayListUnmanaged([]const u8) = .empty;
    defer out.deinit(gpa);
    const a = "1" ** 64;
    const b = "2" ** 64;
    const good = "M " ++ a ++ "\nM " ++ b ++ "\nZ 5d41402abc4b2a76b9719d911017c592\n";
    try std.testing.expect(try collectClusterMembers(gpa, good, &out));
    try std.testing.expectEqual(@as(usize, 2), out.items.len);
    try std.testing.expectEqualStrings(a, out.items[0]);
    out.clearRetainingCapacity();
    // A D card inside a cluster is out of grammar: nothing is applied.
    const bad = "M " ++ a ++ "\nD 2026-01-01T00:00:00\nZ 5d41402abc4b2a76b9719d911017c592\n";
    try std.testing.expect(!try collectClusterMembers(gpa, bad, &out));
    try std.testing.expectEqual(@as(usize, 0), out.items.len);
    // Not a cluster: valid, contributes nothing.
    try std.testing.expect(try collectClusterMembers(gpa, "D 2026-01-01T00:00:00\nZ 0\n", &out));
    try std.testing.expectEqual(@as(usize, 0), out.items.len);
}

test "resolve: held members are clustered, unheld members are missing, deduplicated across clusters" {
    const gpa = std.testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const s = ms.store();
    var names: [3][64]u8 = undefined;
    inline for (.{ "held one", "held two", "held three" }, 0..) |body, i| {
        _ = artifact.artifactName(.sha3_256, body, &names[i]);
        _ = try s.put(&names[i], body);
    }
    const ghost = "e" ** 64;
    const c1 = try std.fmt.allocPrint(gpa, "M {s}\nM {s}\nM {s}\nZ 0\n", .{ names[0], names[1], ghost });
    defer gpa.free(c1);
    const c2 = try std.fmt.allocPrint(gpa, "M {s}\nM {s}\nZ 0\n", .{ names[2], ghost });
    defer gpa.free(c2);
    var c1n: [64]u8 = undefined;
    var c2n: [64]u8 = undefined;
    _ = artifact.artifactName(.sha3_256, c1, &c1n);
    _ = artifact.artifactName(.sha3_256, c2, &c2n);
    _ = try s.put(&c1n, c1);
    _ = try s.put(&c2n, c2);

    var r = try resolve(gpa, s, &.{ &c1n, &c2n });
    defer r.deinit();
    for (&names) |*nm| try std.testing.expect(r.clustered.isSet(s.lookup(nm).?));
    // The clusters themselves are named by no cluster: unclustered.
    try std.testing.expect(!r.clustered.isSet(s.lookup(&c1n).?));
    try std.testing.expect(!r.clustered.isSet(s.lookup(&c2n).?));
    try std.testing.expectEqual(@as(usize, 1), r.missing.len); // ghost, once
    try std.testing.expectEqualStrings(ghost, r.missing[0]);
    try std.testing.expectEqual(@as(usize, 2), r.cluster_names.len);
}

test "mintClusters: below the threshold nothing is minted; above it, ascending M cards batched fossil's way" {
    const gpa = std.testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const s = ms.store();
    var body_buf: [24]u8 = undefined;
    var i: usize = 0;
    while (i < 950) : (i += 1) {
        const body = std.fmt.bufPrint(&body_buf, "loose {d}", .{i}) catch unreachable;
        var nm: [64]u8 = undefined;
        _ = artifact.artifactName(.sha3_256, body, &nm);
        _ = try s.put(&nm, body);
    }
    var none = try std.DynamicBitSetUnmanaged.initEmpty(gpa, 951);
    defer none.deinit(gpa);
    var out: std.ArrayListUnmanaged([]u8) = .empty;
    defer {
        for (out.items) |b| gpa.free(b);
        out.deinit(gpa);
    }
    // 950 unclustered: 800, then the 150 tail absorbed into one cluster.
    try mintClusters(gpa, s, &none, &out);
    try std.testing.expectEqual(@as(usize, 2), out.items.len);
    try std.testing.expectEqual(@as(usize, 800), std.mem.count(u8, out.items[0], "\nM ") + 1);
    try std.testing.expectEqual(@as(usize, 150), std.mem.count(u8, out.items[1], "\nM ") + 1);
    // Each body is a valid cluster whose members are strictly ascending.
    var members: std.ArrayListUnmanaged([]const u8) = .empty;
    defer members.deinit(gpa);
    try std.testing.expect(try collectClusterMembers(gpa, out.items[0], &members));
    var k: usize = 1;
    while (k < members.items.len) : (k += 1) try std.testing.expect(std.mem.lessThan(u8, members.items[k - 1], members.items[k]));
    // Z is the md5 of everything above it.
    const z = std.mem.lastIndexOf(u8, out.items[0], "Z ").?;
    var d: [16]u8 = undefined;
    std.crypto.hash.Md5.hash(out.items[0][0..z], &d, .{});
    try std.testing.expectEqualStrings(&std.fmt.bytesToHex(d, .lower), out.items[0][z + 2 .. z + 34]);

    // Once those are clustered, only the two clusters are loose: below threshold.
    for (out.items) |b| {
        var nm: [64]u8 = undefined;
        _ = artifact.artifactName(.sha3_256, b, &nm);
        _ = try s.put(&nm, b);
    }
    var cn: [2][64]u8 = undefined;
    _ = artifact.artifactName(.sha3_256, out.items[0], &cn[0]);
    _ = artifact.artifactName(.sha3_256, out.items[1], &cn[1]);
    var r = try resolve(gpa, s, &.{ &cn[0], &cn[1] });
    defer r.deinit();
    var again: std.ArrayListUnmanaged([]u8) = .empty;
    defer again.deinit(gpa);
    try mintClusters(gpa, s, &r.clustered, &again);
    try std.testing.expectEqual(@as(usize, 0), again.items.len);
}

test "sidecar: serialize/parse round-trips names and n_rids against the keyed log" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const raw = try tmp.dir.createFile(io, "log", .{ .read = true });
    var fs = @import("storage.zig").FileStorage.init(io, raw, 0);
    var log = try logstore.LogStore.open(gpa, fs.storageFile(), 0);
    defer log.deinit(gpa);
    const names = [_][]const u8{ "a" ** 64, "b" ** 40 };
    const blob = (try serialize(gpa, &log, 7, &names)).?;
    defer gpa.free(blob);
    var arena = std.heap.ArenaAllocator.init(gpa);
    defer arena.deinit();
    const p = (try parse(arena.allocator(), blob, &log)).?;
    try std.testing.expectEqual(@as(u64, 7), p.n_rids);
    try std.testing.expectEqual(@as(usize, 2), p.names.len);
    try std.testing.expectEqualStrings(names[1], p.names[1]);
    // A flipped byte is refused, never adopted.
    const bad = try gpa.dupe(u8, blob);
    defer gpa.free(bad);
    bad[header_len + 3] ^= 1;
    try std.testing.expect((try parse(arena.allocator(), bad, &log)) == null);
}