//! 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 repeat = @import("repeat.zig").repeat;
const artifact = @import("artifact.zig");
const logstore = @import("logstore.zig");
const repostate = @import("repostate.zig");
const nameindex = @import("nameindex.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| try fold(gpa, a, store, m, &clustered, &missing_set, &missing);
}
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);
}
// ---------------------------------------------------------------------------
// The resolved sidecar. `resolve` inflates every held cluster and looks up
// every member: O(store) on every sync open, though the answer changes only
// when the log does. Kept beside the index, keyed to the log prefix and the
// shun table it was resolved over:
//
// magic(8) | version(2) | end(8) | frame_count(8) | last_frame_offset(8) |
// last_header_crc(4) | shun_fp(4) | bit_len(8) | n_missing(4) |
// n_clusters(4) | bits | n_missing x (len(1) name) | n_clusters x
// (len(1) name) | crc(4)
//
// The bits are store rids. They stay true over an appended tail because the
// log-backed adapter numbers frames in log order and a tail only adds rids
// past the prefix. What a tail can change is answered by `adoptResolved`.
// ---------------------------------------------------------------------------
pub const resolved_magic = "ZITCLRS\x00";
const resolved_version: u16 = 1;
const resolved_header_len = 8 + 2 + 8 + 8 + 8 + 4 + 4 + 8 + 4 + 4;
/// `r` as a sidecar blob keyed to `log`'s current prefix and shun table.
/// Null when the key cannot be read or a name does not fit its length byte.
pub fn serializeResolved(gpa: std.mem.Allocator, log: *const logstore.LogStore, r: *const Resolved) error{OutOfMemory}!?[]u8 {
const key = log.logKey() catch return null;
const bit_len = r.clustered.capacity();
const bit_bytes = (bit_len + 7) / 8;
var names_len: usize = 0;
for ([_][]const []const u8{ r.missing, r.cluster_names }) |list| {
for (list) |nm| {
if (nm.len > 255) return null;
names_len += 1 + nm.len;
}
}
const blob = try gpa.alloc(u8, resolved_header_len + bit_bytes + names_len + 4);
errdefer gpa.free(blob);
@memcpy(blob[0..8], resolved_magic);
std.mem.writeInt(u16, blob[8..10], resolved_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(u32, blob[38..42], log.shunFingerprint(), .little);
std.mem.writeInt(u64, blob[42..50], bit_len, .little);
std.mem.writeInt(u32, blob[50..54], @intCast(r.missing.len), .little);
std.mem.writeInt(u32, blob[54..58], @intCast(r.cluster_names.len), .little);
const bits = blob[resolved_header_len..][0..bit_bytes];
@memset(bits, 0);
var it = r.clustered.iterator(.{});
while (it.next()) |i| bits[i / 8] |= @as(u8, 1) << @intCast(i % 8);
var pos: usize = resolved_header_len + bit_bytes;
for ([_][]const []const u8{ r.missing, r.cluster_names }) |list| {
for (list) |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;
}
/// What `adoptResolved` answered from.
pub const Adopted = struct {
resolved: Resolved,
/// The sidecar described this very log: nothing was re-derived, and
/// there is nothing to store again.
exact: bool,
};
/// The `Resolved` for `store`, from a sidecar `blob` written over a prefix
/// of `log`, or null when the blob is foreign, damaged, keyed to a log this
/// one does not extend, or written under a different shun table (a shun
/// masks a held member, which no tail walk can see).
///
/// Over an appended tail, only what the tail can change is re-derived: a
/// member the sidecar recorded missing may have arrived, and a cluster in
/// `cluster_names` the sidecar did not resolve contributes its members.
/// Every other bit holds, since a held member stays held and keeps its rid.
pub fn adoptResolved(
gpa: std.mem.Allocator,
blob: []const u8,
log: *const logstore.LogStore,
store: repostate.ArtifactStore,
cluster_names: []const []const u8,
) error{OutOfMemory}!?Adopted {
if (blob.len < resolved_header_len + 4) return null;
if (!std.mem.eql(u8, blob[0..8], resolved_magic)) return null;
if (std.mem.readInt(u16, blob[8..10], .little) != resolved_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 (std.mem.readInt(u32, blob[38..42], .little) != log.shunFingerprint()) return null;
if (!logstore.LogStore.validatePrefix(log.file, key)) return null;
const exact = logstore.LogStore.validateTail(log.file, key);
const bit_len = std.mem.readInt(u64, blob[42..50], .little);
const n_missing = std.mem.readInt(u32, blob[50..54], .little);
const n_clusters = std.mem.readInt(u32, blob[54..58], .little);
const bit_bytes = std.math.cast(usize, (bit_len + 7) / 8) orelse return null;
if (bit_bytes > blob.len - resolved_header_len - 4) return null;
const bits = blob[resolved_header_len..][0..bit_bytes];
// A refusal returns null, not an error, so the unwinding is by flag.
var kept = false;
var arena = std.heap.ArenaAllocator.init(gpa);
defer if (!kept) arena.deinit();
const a = arena.allocator();
var pos: usize = resolved_header_len + bit_bytes;
const recorded = try a.alloc([]const u8, n_missing + n_clusters);
for (recorded) |*nm| {
if (pos >= blob.len - 4) return null;
const len: usize = blob[pos];
if (pos + 1 + len > blob.len - 4) return null;
nm.* = blob[pos + 1 ..][0..len];
pos += 1 + len;
}
if (pos != blob.len - 4) return null;
const n: usize = store.count();
var clustered = try std.DynamicBitSetUnmanaged.initEmpty(gpa, n + 1);
defer if (!kept) clustered.deinit(gpa);
for (bits, 0..) |byte, at| {
if (byte == 0) continue;
for (0..8) |bit| {
const rid = at * 8 + bit;
if (byte & (@as(u8, 1) << @intCast(bit)) == 0) continue;
if (rid >= clustered.capacity()) return null; // a rid this store does not have
clustered.set(rid);
}
}
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);
var resolved_set: std.StringHashMapUnmanaged(void) = .empty;
defer resolved_set.deinit(gpa);
for (recorded[0..n_missing]) |m| try fold(gpa, a, store, m, &clustered, &missing_set, &missing);
for (recorded[n_missing..]) |c| {
const owned = try a.dupe(u8, c);
try names.append(a, owned);
try resolved_set.put(gpa, owned, {});
}
if (!exact) {
for (cluster_names) |name| {
if (resolved_set.contains(name)) continue;
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| try fold(gpa, a, store, m, &clustered, &missing_set, &missing);
}
}
kept = true;
return .{ .exact = exact, .resolved = .{
.clustered = clustered,
.missing = missing.items,
.cluster_names = names.items,
.arena = arena,
.gpa = gpa,
} };
}
/// One cluster member into a `Resolved` under construction: held sets its
/// bit, otherwise it is a missing promise, once.
fn fold(
gpa: std.mem.Allocator,
a: std.mem.Allocator,
store: repostate.ArtifactStore,
member: []const u8,
clustered: *std.DynamicBitSetUnmanaged,
missing_set: *std.StringHashMapUnmanaged(void),
missing: *std.ArrayListUnmanaged([]const u8),
) error{OutOfMemory}!void {
if (store.lookup(member)) |mr| {
if (store.held(mr)) {
if (mr < clustered.capacity()) clustered.set(mr);
return;
}
}
const gop = try missing_set.getOrPut(gpa, member);
if (gop.found_existing) return;
const owned = try a.dupe(u8, member);
gop.key_ptr.* = owned;
try missing.append(a, owned);
}
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 = repeat("1", 64);
const b = repeat("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 = repeat("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{ repeat("a", 64), repeat("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);
}
fn expectSameResolved(store: repostate.ArtifactStore, want: *const Resolved, got: *const Resolved) !void {
var rid: Rid = 1;
while (rid <= store.count()) : (rid += 1) {
try std.testing.expectEqual(want.clustered.isSet(rid), got.clustered.isSet(rid));
}
try std.testing.expectEqual(want.missing.len, got.missing.len);
for (want.missing) |m| {
for (got.missing) |g| {
if (std.mem.eql(u8, m, g)) break;
} else return error.TestExpectedEqual;
}
try std.testing.expectEqual(want.cluster_names.len, got.cluster_names.len);
for (want.cluster_names) |c| {
for (got.cluster_names) |g| {
if (std.mem.eql(u8, c, g)) break;
} else return error.TestExpectedEqual;
}
}
test "resolved sidecar: adopted whole over its own log, re-derived over a tail, refused after a shun" {
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);
var frame_hash: [32]u8 = @splat(1);
_ = try log.put(&frame_hash, "first frame");
try log.commit();
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[0], "held one");
_ = try s.put(&names[1], "held two");
const ghost = repeat("e", 64);
const c1 = try std.fmt.allocPrint(gpa, "M {s}\nM {s}\nZ 0\n", .{ names[0], ghost });
defer gpa.free(c1);
var c1n: [64]u8 = undefined;
_ = artifact.artifactName(.sha3_256, c1, &c1n);
_ = try s.put(&c1n, c1);
var first = try resolve(gpa, s, &.{&c1n});
defer first.deinit();
const blob = (try serializeResolved(gpa, &log, &first)).?;
defer gpa.free(blob);
// The same log: adopted as it stands, nothing re-derived.
{
var same = (try adoptResolved(gpa, blob, &log, s, &.{&c1n})).?;
defer same.resolved.deinit();
try std.testing.expect(same.exact);
try expectSameResolved(s, &first, &same.resolved);
}
// A tail: the missing member arrives, and a new cluster names a member
// no cluster named before. The adopted answer is the fresh one.
_ = try s.put(ghost, "ghost");
_ = try s.put(&names[2], "held three");
const c2 = try std.fmt.allocPrint(gpa, "M {s}\nZ 0\n", .{names[2]});
defer gpa.free(c2);
var c2n: [64]u8 = undefined;
_ = artifact.artifactName(.sha3_256, c2, &c2n);
_ = try s.put(&c2n, c2);
frame_hash = @splat(2);
_ = try log.put(&frame_hash, "second frame");
try log.commit();
{
var fresh = try resolve(gpa, s, &.{ &c1n, &c2n });
defer fresh.deinit();
var grown = (try adoptResolved(gpa, blob, &log, s, &.{ &c1n, &c2n })).?;
defer grown.resolved.deinit();
try std.testing.expect(!grown.exact);
try expectSameResolved(s, &fresh, &grown.resolved);
try std.testing.expect(grown.resolved.clustered.isSet(s.lookup(ghost).?));
try std.testing.expect(grown.resolved.clustered.isSet(s.lookup(&names[2]).?));
}
// A shun can unhold a member the bits still name: refused.
try log.putShun(&frame_hash, "", 0);
try log.commit();
try std.testing.expect((try adoptResolved(gpa, blob, &log, s, &.{ &c1n, &c2n })) == null);
// A damaged blob is refused, never adopted.
const bad = try gpa.dupe(u8, blob);
defer gpa.free(bad);
bad[resolved_header_len] ^= 1;
try std.testing.expect((try adoptResolved(gpa, bad, &log, s, &.{&c1n})) == null);
}
test "resolved sidecar: a rewritten log and a bit past the store are refused" {
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);
var frame_hash: [32]u8 = @splat(1);
_ = try log.put(&frame_hash, "first frame");
try log.commit();
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const s = ms.store();
var names: [3][64]u8 = undefined;
inline for (.{ "one", "two", "three" }, 0..) |body, i| {
_ = artifact.artifactName(.sha3_256, body, &names[i]);
_ = try s.put(&names[i], body);
}
const c1 = try std.fmt.allocPrint(gpa, "M {s}\nZ 0\n", .{names[2]});
defer gpa.free(c1);
var c1n: [64]u8 = undefined;
_ = artifact.artifactName(.sha3_256, c1, &c1n);
_ = try s.put(&c1n, c1);
var r = try resolve(gpa, s, &.{&c1n});
defer r.deinit();
const blob = (try serializeResolved(gpa, &log, &r)).?;
defer gpa.free(blob);
// A store with fewer rids than the bits name: the bits are not its own.
{
var small = repostate.MemStore.init(gpa);
defer small.deinit();
_ = try small.store().put(&names[0], "one");
try std.testing.expect((try adoptResolved(gpa, blob, &log, small.store(), &.{&c1n})) == null);
}
// The same path rewritten, as pack and purge leave it: not a prefix.
var other_tmp = std.testing.tmpDir(.{});
defer other_tmp.cleanup();
const other_raw = try other_tmp.dir.createFile(io, "log", .{ .read = true });
var other_fs = @import("storage.zig").FileStorage.init(io, other_raw, 0);
var other = try logstore.LogStore.open(gpa, other_fs.storageFile(), 0);
defer other.deinit(gpa);
frame_hash = @splat(9);
_ = try other.put(&frame_hash, "a different first frame");
try other.commit();
try std.testing.expect((try adoptResolved(gpa, blob, &other, s, &.{&c1n})) == null);
}