From 5f5a33deea355daa5a5870dfbaa2398ab8541d18 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Wed, 10 Jun 2026 11:26:20 -0500 Subject: [PATCH] fix chain-break metric: ~91% false positives from commit CID in data_cid MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit with verify_commit_diff off (prod default), sig-only verification returned the commit CID in the data_cid field — a different block than the MST root the prevData continuity check compares against, so nearly every commit with prevData counted as a chain break (80.9M breaks vs 75.1M frames received). - validator: sig-only paths (validateSync + verifyCommit fallback) now return null data_cid instead of the commit CID - frame_worker/subscriber: prevData comparison gated on verify_commit_diff (the only path that stores a real MST root) - split relay_chain_breaks_total by reason ("since" | "prev_data") so the real since-vs-stored-rev signal is separable; JSON stats keeps the sum - regression test: real signed commit CAR through both validate paths, asserting data_cid is null Co-Authored-By: Claude Fable 5 --- src/internal/broadcaster.zig | 18 ++++--- src/internal/frame_worker.zig | 28 ++++++----- src/internal/subscriber.zig | 28 ++++++----- src/internal/validator.zig | 88 +++++++++++++++++++++++++++++++++-- 4 files changed, 128 insertions(+), 34 deletions(-) diff --git a/src/internal/broadcaster.zig b/src/internal/broadcaster.zig index 209948a..47ff95f 100644 --- a/src/internal/broadcaster.zig +++ b/src/internal/broadcaster.zig @@ -51,7 +51,9 @@ pub const Stats = struct { slow_consumers: std.atomic.Value(u64) = .{ .raw = 0 }, connected_inbound: std.atomic.Value(u64) = .{ .raw = 0 }, cache_evictions: std.atomic.Value(u64) = .{ .raw = 0 }, - chain_breaks: std.atomic.Value(u64) = .{ .raw = 0 }, + // chain continuity failures by reason (since vs stored rev, prevData vs stored data CID) + chain_breaks_since: std.atomic.Value(u64) = .{ .raw = 0 }, + chain_breaks_prev_data: std.atomic.Value(u64) = .{ .raw = 0 }, pool_backpressure: std.atomic.Value(u64) = .{ .raw = 0 }, // host authority resolution metrics host_authority_checks: std.atomic.Value(u64) = .{ .raw = 0 }, @@ -1021,10 +1023,6 @@ pub fn formatPrometheusMetrics(stats: *const Stats, cache_entries: usize, attrib \\# HELP relay_resolve_queued_set_count DID resolve dedup set size \\relay_resolve_queued_set_count {d} \\ - \\# TYPE relay_chain_breaks_total counter - \\# HELP relay_chain_breaks_total since/prevData chain continuity failures - \\relay_chain_breaks_total {d} - \\ \\# TYPE relay_host_authority_checks_total counter \\# HELP relay_host_authority_checks_total resolveHostAuthority calls on frame workers \\relay_host_authority_checks_total {d} @@ -1069,7 +1067,6 @@ pub fn formatPrometheusMetrics(stats: *const Stats, cache_entries: usize, attrib attribution.did_cache_entries, attribution.resolve_queue_len, attribution.resolve_queued_set_count, - stats.chain_breaks.load(.acquire), stats.host_authority_checks.load(.acquire), stats.host_authority_is_new.load(.acquire), stats.host_authority_host_changed.load(.acquire), @@ -1079,6 +1076,11 @@ pub fn formatPrometheusMetrics(stats: *const Stats, cache_entries: usize, attrib // pipeline contention metrics (separate print to stay under 32-arg limit) w.print( + \\# TYPE relay_chain_breaks_total counter + \\# HELP relay_chain_breaks_total chain continuity failures by reason + \\relay_chain_breaks_total{{reason="since"}} {d} + \\relay_chain_breaks_total{{reason="prev_data"}} {d} + \\ \\# TYPE relay_persist_order_spins_total counter \\# HELP relay_persist_order_spins_total spin iterations waiting for persist ordering lock \\relay_persist_order_spins_total {d} @@ -1100,6 +1102,8 @@ pub fn formatPrometheusMetrics(stats: *const Stats, cache_entries: usize, attrib \\relay_broadcast_no_consumers_total {d} \\ , .{ + stats.chain_breaks_since.load(.acquire), + stats.chain_breaks_prev_data.load(.acquire), stats.persist_order_spins.load(.acquire), stats.broadcast_queue_push_lock_spins.load(.acquire), stats.broadcast_queue_full.load(.acquire), @@ -1476,7 +1480,7 @@ pub fn formatStatsResponse(stats: *const Stats, buf: []u8, io: Io) []const u8 { stats.cache_hits.load(.acquire), stats.cache_misses.load(.acquire), stats.slow_consumers.load(.acquire), - stats.chain_breaks.load(.acquire), + stats.chain_breaks_since.load(.acquire) + stats.chain_breaks_prev_data.load(.acquire), stats.pool_backpressure.load(.acquire), timestamp(io) - stats.start_time, }) catch ""; diff --git a/src/internal/frame_worker.zig b/src/internal/frame_worker.zig index 68c5628..1fc6b0f 100644 --- a/src/internal/frame_worker.zig +++ b/src/internal/frame_worker.zig @@ -211,21 +211,25 @@ pub fn processFrame(work: *FrameWork) void { log.info("host {s}: chain break uid={d} since={s} stored_rev={s}", .{ work.hostname, uid, since, prev.rev, }); - _ = work.bc.stats.chain_breaks.fetchAdd(1, .monotonic); + _ = work.bc.stats.chain_breaks_since.fetchAdd(1, .monotonic); } } - // chain continuity: prevData CID should match stored data_cid - if (payload.get("prevData")) |pd| { - if (pd == .cid) { - const prev_data_encoded = zat.multibase.encode(alloc, .base32lower, pd.cid.raw) catch ""; - if (prev_data_encoded.len > 0 and prev.data_cid.len > 0 and - !std.mem.eql(u8, prev_data_encoded, prev.data_cid)) - { - log.info("host {s}: chain break uid={d} prevData mismatch", .{ - work.hostname, uid, - }); - _ = work.bc.stats.chain_breaks.fetchAdd(1, .monotonic); + // chain continuity: prevData CID should match stored data_cid. + // only meaningful with verify_commit_diff — that's the only path + // that stores a real MST root in data_cid. + if (work.validator.config.verify_commit_diff) { + if (payload.get("prevData")) |pd| { + if (pd == .cid) { + const prev_data_encoded = zat.multibase.encode(alloc, .base32lower, pd.cid.raw) catch ""; + if (prev_data_encoded.len > 0 and prev.data_cid.len > 0 and + !std.mem.eql(u8, prev_data_encoded, prev.data_cid)) + { + log.info("host {s}: chain break uid={d} prevData mismatch", .{ + work.hostname, uid, + }); + _ = work.bc.stats.chain_breaks_prev_data.fetchAdd(1, .monotonic); + } } } } diff --git a/src/internal/subscriber.zig b/src/internal/subscriber.zig index eef3d16..86f975d 100644 --- a/src/internal/subscriber.zig +++ b/src/internal/subscriber.zig @@ -667,21 +667,25 @@ const FrameHandler = struct { log.info("host {s}: chain break uid={d} since={s} stored_rev={s}", .{ sub.options.hostname, uid, since, prev.rev, }); - _ = sub.bc.stats.chain_breaks.fetchAdd(1, .monotonic); + _ = sub.bc.stats.chain_breaks_since.fetchAdd(1, .monotonic); } } - // chain continuity: prevData CID should match stored data_cid - if (payload.get("prevData")) |pd| { - if (pd == .cid) { - const prev_data_encoded = zat.multibase.encode(alloc, .base32lower, pd.cid.raw) catch ""; - if (prev_data_encoded.len > 0 and prev.data_cid.len > 0 and - !std.mem.eql(u8, prev_data_encoded, prev.data_cid)) - { - log.info("host {s}: chain break uid={d} prevData mismatch", .{ - sub.options.hostname, uid, - }); - _ = sub.bc.stats.chain_breaks.fetchAdd(1, .monotonic); + // chain continuity: prevData CID should match stored data_cid. + // only meaningful with verify_commit_diff — that's the only path + // that stores a real MST root in data_cid. + if (sub.validator.config.verify_commit_diff) { + if (payload.get("prevData")) |pd| { + if (pd == .cid) { + const prev_data_encoded = zat.multibase.encode(alloc, .base32lower, pd.cid.raw) catch ""; + if (prev_data_encoded.len > 0 and prev.data_cid.len > 0 and + !std.mem.eql(u8, prev_data_encoded, prev.data_cid)) + { + log.info("host {s}: chain break uid={d} prevData mismatch", .{ + sub.options.hostname, uid, + }); + _ = sub.bc.stats.chain_breaks_prev_data.fetchAdd(1, .monotonic); + } } } } diff --git a/src/internal/validator.zig b/src/internal/validator.zig index 08a87a9..c1069d1 100644 --- a/src/internal/validator.zig +++ b/src/internal/validator.zig @@ -32,7 +32,11 @@ const CachedKey = struct { pub const ValidationResult = struct { valid: bool, skipped: bool, - data_cid: ?[]const u8 = null, // MST root CID from verified commit + // MST root CID from verified commit. only set by the verify_commit_diff + // path — sig-only verification can't recover the MST root (verifyCommitCar + // exposes the commit CID, which is a different block), so it returns null + // rather than poison the prevData chain-continuity check downstream. + data_cid: ?[]const u8 = null, commit_rev: ?[]const u8 = null, // rev from verified commit }; @@ -215,7 +219,6 @@ pub const Validator = struct { return .{ .valid = true, .skipped = false, - .data_cid = result.commit_cid, .commit_rev = result.commit_rev, }; } @@ -319,7 +322,6 @@ pub const Validator = struct { return .{ .valid = true, .skipped = false, - .data_cid = result.commit_cid, .commit_rev = result.commit_rev, }; } @@ -1022,3 +1024,83 @@ test "queueResolve deduplicates repeated DIDs" { try std.testing.expectEqual(@as(usize, 1), v.queue.items.len); try std.testing.expectEqual(@as(u32, 1), v.queued_set.count()); } + +// build a commit CAR with a real signature over the unsigned commit bytes, +// so verification reaches the success path (not a sig failure → skip) +fn buildSignedCommitCar(a: Allocator, kp: zat.Keypair, did_str: []const u8, rev: []const u8) ![]u8 { + const data_cid = try zat.cbor.Cid.forDagCbor(a, "mst-root-placeholder"); + const unsigned: zat.cbor.Value = .{ .map = &.{ + .{ .key = "did", .value = .{ .text = did_str } }, + .{ .key = "rev", .value = .{ .text = rev } }, + .{ .key = "data", .value = .{ .cid = data_cid } }, + .{ .key = "version", .value = .{ .unsigned = 3 } }, + } }; + const unsigned_bytes = try zat.cbor.encodeAlloc(a, unsigned); + const sig = try kp.sign(unsigned_bytes); + + const signed: zat.cbor.Value = .{ .map = &.{ + .{ .key = "did", .value = .{ .text = did_str } }, + .{ .key = "rev", .value = .{ .text = rev } }, + .{ .key = "data", .value = .{ .cid = data_cid } }, + .{ .key = "version", .value = .{ .unsigned = 3 } }, + .{ .key = "sig", .value = .{ .bytes = &sig.bytes } }, + } }; + const commit_bytes = try zat.cbor.encodeAlloc(a, signed); + const commit_cid = try zat.cbor.Cid.forDagCbor(a, commit_bytes); + + return zat.car.writeAlloc(a, .{ + .roots = &.{commit_cid}, + .blocks = &.{.{ .cid_raw = commit_cid.raw, .data = commit_bytes }}, + }); +} + +test "regression: sig-only verification returns null data_cid" { + // with verify_commit_diff off (prod default), verification can only recover + // the commit CID — not the MST root. returning it as data_cid poisoned the + // prevData chain-continuity check (~91% of relay_chain_breaks_total were + // commit-CID-vs-MST-root comparisons that can never be equal). + var stats = broadcaster.Stats{}; + var v = Validator.init(std.testing.allocator, &stats, std.testing.io); + defer v.deinit(); + + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + + const kp = try zat.Keypair.fromSecretKey(.p256, .{1} ** 32); + const did = "did:plc:test123"; + const rev = "3k2abcdefghij"; + const car_bytes = try buildSignedCommitCar(a, kp, did, rev); + + const pubkey = try kp.publicKey(); + try v.cache.put(did, .{ + .key_type = .p256, + .raw = pubkey, + .len = 33, + .resolve_time = 100, + }); + + // #commit path (verifyCommit fallback) + const commit_payload: zat.cbor.Value = .{ .map = &.{ + .{ .key = "repo", .value = .{ .text = did } }, + .{ .key = "rev", .value = .{ .text = rev } }, + .{ .key = "blocks", .value = .{ .bytes = car_bytes } }, + } }; + const commit_result = v.validateCommit(commit_payload); + try std.testing.expect(commit_result.valid); + try std.testing.expect(!commit_result.skipped); + try std.testing.expectEqualStrings(rev, commit_result.commit_rev.?); + try std.testing.expect(commit_result.data_cid == null); + + // #sync path (validateSync) + const sync_payload: zat.cbor.Value = .{ .map = &.{ + .{ .key = "did", .value = .{ .text = did } }, + .{ .key = "rev", .value = .{ .text = rev } }, + .{ .key = "blocks", .value = .{ .bytes = car_bytes } }, + } }; + const sync_result = v.validateSync(sync_payload); + try std.testing.expect(sync_result.valid); + try std.testing.expect(!sync_result.skipped); + try std.testing.expectEqualStrings(rev, sync_result.commit_rev.?); + try std.testing.expect(sync_result.data_cid == null); +} -- 2.51.2