//! 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 nameindex = @import("nameindex.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.
// ---------------------------------------------------------------------------
/// The sidecar is a `nameindex` list under this magic; its bytes are
/// unchanged from when the encoding lived here.
pub const magic = "ZITCLIX\x00";
pub fn serialize(
gpa: std.mem.Allocator,
log: *const logstore.LogStore,
n_rids: u64,
cluster_names: []const []const u8,
) error{OutOfMemory}!?[]u8 {
return nameindex.serialize(gpa, magic, log, n_rids, cluster_names);
}
pub const Parsed = nameindex.Parsed;
/// 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 {
return nameindex.parse(arena, magic, blob, log);
}
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[nameindex.header_len + 3] ^= 1;
try std.testing.expect((try parse(arena.allocator(), bad, &log)) == null);
}