From 1b39bf05d9cbaa1d4d8a82291a2d3603fd97a57e Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sat, 25 Jul 2026 01:50:04 -0500 Subject: [PATCH] add the relay conformance harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Runs the atmoq relay-conformance corpora through the real decode and validation path and writes one JSONL verdict per case, so the table rows are measured rather than inferred. The atmoq README already documented this entry point; it had never been committed here. The harness reports the relay-level verdict (would zlay drop this frame?) rather than the raw result of a validation function, mirroring the drop sequence in frame_worker.processFrame. Prior repo state for the stateful sync-1.1 sequences is held in memory so no database is needed. Measured against the published baseline: 43 of 49 rows unchanged, the five intended reject flips confirmed, and one regression surfaced — sync11/commit2-valid, a valid @atproto second commit, is dropped with PrevDataMismatch. See the report accompanying the zat investigation. --- build.zig | 22 +++ src/conformance_main.zig | 374 +++++++++++++++++++++++++++++++++++++ src/internal/validator.zig | 20 ++ 3 files changed, 416 insertions(+) create mode 100644 src/conformance_main.zig diff --git a/build.zig b/build.zig index e66d15a..a361303 100644 --- a/build.zig +++ b/build.zig @@ -64,6 +64,28 @@ pub fn build(b: *std.Build) void { const run_step = b.step("run", "run the relay"); run_step.dependOn(&run_relay.step); + // conformance harness — runs the atmoq relay-conformance corpora through + // the real decode + validation path (see src/conformance_main.zig) + const conformance_mod = b.createModule(.{ + .root_source_file = b.path("src/conformance_main.zig"), + .target = target, + .optimize = optimize, + .imports = imports, + }); + conformance_mod.addImport("build_options", build_options.createModule()); + conformance_mod.link_libc = true; + conformance_mod.link_libcpp = true; + const conformance = b.addExecutable(.{ + .name = "conformance", + .root_module = conformance_mod, + }); + b.installArtifact(conformance); + + const run_conformance = b.addRunArtifact(conformance); + if (b.args) |args| run_conformance.addArgs(args); + const conformance_step = b.step("conformance", "run the relay conformance harness"); + conformance_step.dependOn(&run_conformance.step); + // tests // tests — a single module rooted at src/ so every internal/ import resolves const test_step = b.step("test", "run unit tests"); diff --git a/src/conformance_main.zig b/src/conformance_main.zig new file mode 100644 index 0000000..55e5412 --- /dev/null +++ b/src/conformance_main.zig @@ -0,0 +1,374 @@ +//! conformance — runs the atmoq relay-conformance corpora through zlay's real +//! decode and validation path and writes one JSONL verdict per case. +//! +//! The harness reports the *relay-level* verdict (would zlay drop this frame?), +//! not the raw result of a validation function: `accept` = forwarded downstream, +//! `reject` = dropped as malformed/invalid, `skip` = dropped for a non-defect +//! reason (stale rev) or forwarded without validation. That distinction is the +//! whole point of the differential table — see the atmoq README. +//! +//! Config comes from the environment because the harness runs inside the +//! vendored Docker toolchain: +//! +//! MODE=account|commit|sync CORPUS=corpus.json OUT=results/zlay.jsonl +//! +//! account mode judges frame decode alone (the corpus rides its encoding +//! defects on signature-free #account events). commit and sync mode seed the +//! corpus signing key into the key cache so the commit path takes its cache-hit +//! branch, then mirror the drop decisions frame_worker.zig makes in production. + +const std = @import("std"); +const zat = @import("zat"); +const Io = std.Io; + +const validator_mod = @import("internal/validator.zig"); +const broadcaster = @import("internal/broadcaster.zig"); +const env = @import("internal/util/env.zig"); + +const Validator = validator_mod.Validator; + +const Outcome = enum { + accept, + reject, + skip, + + fn str(self: Outcome) []const u8 { + return @tagName(self); + } +}; + +const Verdict = struct { + outcome: Outcome, + detail: []const u8, +}; + +const Mode = enum { account, commit, sync }; + +/// a case from corpus.json / corpus-commit.json +const Case = struct { + id: []const u8, + layer: []const u8, + hex: []const u8, + repoDid: ?[]const u8 = null, + signingKey: ?[]const u8 = null, +}; + +/// a case from corpus-sync.json — a setup/test frame sequence +const SyncCase = struct { + id: []const u8, + layer: []const u8, + repoDid: ?[]const u8 = null, + signingKey: ?[]const u8 = null, + frames: []const Frame, + + const Frame = struct { + role: []const u8, + hex: []const u8, + }; +}; + +pub fn main() !void { + var backend = Io.Threaded.init(std.heap.c_allocator, .{}); + defer backend.deinit(); + const io = backend.io(); + + var arena = std.heap.ArenaAllocator.init(std.heap.c_allocator); + defer arena.deinit(); + const alloc = arena.allocator(); + + const mode_name = env.getenv("MODE") orelse "account"; + const mode = std.meta.stringToEnum(Mode, mode_name) orelse { + std.debug.print("unknown MODE={s} (want account|commit|sync)\n", .{mode_name}); + return error.InvalidMode; + }; + const corpus_path = env.getenv("CORPUS") orelse switch (mode) { + .account => "corpus.json", + .commit => "corpus-commit.json", + .sync => "corpus-sync.json", + }; + const out_path = env.getenv("OUT") orelse switch (mode) { + .account => "results/zlay.jsonl", + .commit => "results/zlay-commit.jsonl", + .sync => "results/zlay-sync.jsonl", + }; + + const source = try Io.Dir.cwd().readFileAlloc(io, corpus_path, alloc, .limited(64 * 1024 * 1024)); + + var out: std.ArrayListUnmanaged(u8) = .empty; + + switch (mode) { + .account => try runAccount(alloc, source, &out), + .commit => try runCommit(alloc, io, source, &out), + .sync => try runSync(alloc, io, source, &out), + } + + try Io.Dir.cwd().writeFile(io, .{ .sub_path = out_path, .data = out.items }); + std.debug.print("{s}: {s} -> {s}\n", .{ mode_name, corpus_path, out_path }); +} + +// --- modes --- + +fn runAccount(alloc: std.mem.Allocator, source: []const u8, out: *std.ArrayListUnmanaged(u8)) !void { + const parsed = try std.json.parseFromSlice([]const Case, alloc, source, .{ .ignore_unknown_fields = true }); + defer parsed.deinit(); + + for (parsed.value) |case| { + var case_arena = std.heap.ArenaAllocator.init(alloc); + defer case_arena.deinit(); + const a = case_arena.allocator(); + + const verdict = blk: { + const data = decodeHex(a, case.hex) catch break :blk Verdict{ + .outcome = .skip, + .detail = "corpus hex could not be decoded", + }; + const frame = decodeFrame(a, data) catch |err| break :blk frameRejection(err); + break :blk Verdict{ .outcome = .accept, .detail = acceptDetail(frame.frame_type) }; + }; + + try emit(alloc, out, case.id, verdict); + } +} + +fn runCommit(alloc: std.mem.Allocator, io: Io, source: []const u8, out: *std.ArrayListUnmanaged(u8)) !void { + const parsed = try std.json.parseFromSlice([]const Case, alloc, source, .{ .ignore_unknown_fields = true }); + defer parsed.deinit(); + + for (parsed.value) |case| { + var case_arena = std.heap.ArenaAllocator.init(alloc); + defer case_arena.deinit(); + const a = case_arena.allocator(); + + // one validator per case: no cross-case key or state carryover + var stats = broadcaster.Stats{}; + var v = Validator.init(alloc, &stats, io); + defer v.deinit(); + + const verdict = blk: { + const data = decodeHex(a, case.hex) catch break :blk Verdict{ + .outcome = .skip, + .detail = "corpus hex could not be decoded", + }; + const frame = decodeFrame(a, data) catch |err| break :blk frameRejection(err); + + seedCase(&v, a, case.repoDid, case.signingKey) catch |err| break :blk Verdict{ + .outcome = .skip, + .detail = std.fmt.allocPrint(a, "could not seed corpus signing key: {s}", .{@errorName(err)}) catch "seed failed", + }; + + break :blk judge(&v, frame, null).verdict; + }; + + try emit(alloc, out, case.id, verdict); + } +} + +fn runSync(alloc: std.mem.Allocator, io: Io, source: []const u8, out: *std.ArrayListUnmanaged(u8)) !void { + const parsed = try std.json.parseFromSlice([]const SyncCase, alloc, source, .{ .ignore_unknown_fields = true }); + defer parsed.deinit(); + + for (parsed.value) |case| { + var case_arena = std.heap.ArenaAllocator.init(alloc); + defer case_arena.deinit(); + const a = case_arena.allocator(); + + var stats = broadcaster.Stats{}; + var v = Validator.init(alloc, &stats, io); + defer v.deinit(); + + var prior: ?PriorState = null; + + const verdict = blk: { + seedCase(&v, a, case.repoDid, case.signingKey) catch |err| break :blk Verdict{ + .outcome = .skip, + .detail = std.fmt.allocPrint(a, "could not seed corpus signing key: {s}", .{@errorName(err)}) catch "seed failed", + }; + + for (case.frames) |raw| { + const is_test = std.mem.eql(u8, raw.role, "test"); + + const data = decodeHex(a, raw.hex) catch break :blk Verdict{ + .outcome = .skip, + .detail = "corpus hex could not be decoded", + }; + const frame = decodeFrame(a, data) catch |err| { + if (is_test) break :blk frameRejection(err); + break :blk Verdict{ + .outcome = .skip, + .detail = std.fmt.allocPrint(a, "setup frame did not decode: {s}", .{@errorName(err)}) catch "setup decode failed", + }; + }; + + const judged = judge(&v, frame, prior); + + if (is_test) break :blk judged.verdict; + + if (judged.verdict.outcome != .accept) break :blk Verdict{ + .outcome = .skip, + .detail = std.fmt.allocPrint(a, "setup frame was not accepted ({s})", .{judged.verdict.detail}) catch "setup rejected", + }; + + // the setup frame establishes the prior repo state that the + // sync-1.1 second-commit checks fire against + if (judged.commit_rev orelse frame.payload.getString("rev")) |rev| { + prior = .{ + .rev = try a.dupe(u8, rev), + .data_cid = if (judged.data_cid) |cid| try a.dupe(u8, cid) else null, + }; + } + } + + break :blk Verdict{ .outcome = .skip, .detail = "case had no test frame" }; + }; + + try emit(alloc, out, case.id, verdict); + } +} + +// --- relay-level judgement (mirrors frame_worker.processFrame) --- + +/// prior repo state for a DID — what DiskPersist.getAccountState supplies in +/// production. Held in memory here so the harness needs no database. +const PriorState = struct { + rev: []const u8, + data_cid: ?[]const u8, +}; + +const Judgement = struct { + verdict: Verdict, + /// verified MST root and rev, when validation reached the success path — + /// this is what becomes the next frame's prior repo state + data_cid: ?[]const u8 = null, + commit_rev: ?[]const u8 = null, +}; + +fn judge(v: *Validator, frame: DecodedFrame, prior: ?PriorState) Judgement { + const is_commit = std.mem.eql(u8, frame.frame_type, "#commit"); + const is_sync = std.mem.eql(u8, frame.frame_type, "#sync"); + + if (!is_commit and !is_sync) { + return .{ .verdict = .{ .outcome = .accept, .detail = acceptDetail(frame.frame_type) } }; + } + + if (is_commit) { + if (prior) |prev| { + // stale-rev drop: not a defect verdict, so it reports as skip + if (frame.payload.getString("rev")) |incoming_rev| { + if (std.mem.order(u8, incoming_rev, prev.rev) != .gt) { + return .{ .verdict = .{ .outcome = .skip, .detail = "dropped as stale rev (not greater than prior rev)" } }; + } + } + + // with prior state, a missing prevData cannot be repaired by MST + // inversion, so the commit is dropped before the expensive verify + if (!v.validatePrevDataPresence(frame.payload)) { + return .{ .verdict = .{ .outcome = .reject, .detail = "prior repo state held but prevData absent" } }; + } + } + + const result = v.validateCommit(frame.payload); + if (!result.valid) return .{ .verdict = .{ .outcome = .reject, .detail = "commit validation failed (signature, CAR integrity, or MST inversion)" } }; + if (result.skipped) return .{ .verdict = .{ .outcome = .accept, .detail = "forwarded unvalidated (signing key cache miss)" } }; + return .{ + .verdict = .{ .outcome = .accept, .detail = "commit validated" }, + .data_cid = result.data_cid, + .commit_rev = result.commit_rev, + }; + } + + const result = v.validateSync(frame.payload); + if (!result.valid) return .{ .verdict = .{ .outcome = .reject, .detail = "#sync validation failed" } }; + if (result.skipped) return .{ .verdict = .{ .outcome = .accept, .detail = "forwarded unvalidated (signing key cache miss)" } }; + return .{ + .verdict = .{ .outcome = .accept, .detail = "#sync validated" }, + .data_cid = result.data_cid, + .commit_rev = result.commit_rev, + }; +} + +// --- frame decode (mirrors subscriber.handleFrame) --- + +const DecodedFrame = struct { + frame_type: []const u8, + payload: zat.cbor.Value, +}; + +const FrameError = error{ + HeaderDecodeFailed, + MissingOp, + ErrorFrame, + MissingType, + PayloadDecodeFailed, +}; + +fn decodeFrame(a: std.mem.Allocator, data: []const u8) FrameError!DecodedFrame { + const header_result = zat.cbor.decode(a, data) catch return error.HeaderDecodeFailed; + const header = header_result.value; + const payload_data = data[header_result.consumed..]; + + const op = header.getInt("op") orelse return error.MissingOp; + if (op == -1) return error.ErrorFrame; + + const frame_type = header.getString("t") orelse return error.MissingType; + const payload = zat.cbor.decodeAll(a, payload_data) catch return error.PayloadDecodeFailed; + + return .{ .frame_type = frame_type, .payload = payload }; +} + +fn frameRejection(err: FrameError) Verdict { + return switch (err) { + error.HeaderDecodeFailed => .{ .outcome = .reject, .detail = "frame header failed DAG-CBOR decode" }, + error.MissingOp => .{ .outcome = .reject, .detail = "frame header has no op field" }, + error.ErrorFrame => .{ .outcome = .skip, .detail = "upstream error frame (op=-1)" }, + error.MissingType => .{ .outcome = .reject, .detail = "op=1 header has no t field" }, + error.PayloadDecodeFailed => .{ .outcome = .reject, .detail = "frame payload failed DAG-CBOR decode" }, + }; +} + +/// zlay filters unknown message types before the frame pool rather than +/// treating them as malformed, so they count as accepted at the decode axis. +fn acceptDetail(frame_type: []const u8) []const u8 { + const known = [_][]const u8{ "#commit", "#sync", "#account", "#identity" }; + for (known) |k| { + if (std.mem.eql(u8, frame_type, k)) return "frame decoded"; + } + return "frame decoded; unknown type filtered (forward-compat)"; +} + +// --- helpers --- + +fn seedCase(v: *Validator, a: std.mem.Allocator, repo_did: ?[]const u8, signing_key: ?[]const u8) !void { + const did = repo_did orelse return error.CaseHasNoRepoDid; + const key = signing_key orelse return error.CaseHasNoSigningKey; + + const prefix = "did:key:"; + if (!std.mem.startsWith(u8, key, prefix)) return error.NotADidKey; + + const key_bytes = try zat.multibase.decode(a, key[prefix.len..]); + const public_key = try zat.multicodec.parsePublicKey(key_bytes); + + // a far-future resolve_time keeps refreshSigningKey's min-interval gate + // permanently closed: the corpus key is authoritative and the harness has + // no business reaching the PLC directory + try v.seedSigningKey(did, public_key, std.math.maxInt(i64) >> 2); +} + +fn decodeHex(a: std.mem.Allocator, hex: []const u8) ![]u8 { + if (hex.len % 2 != 0) return error.OddLengthHex; + const buf = try a.alloc(u8, hex.len / 2); + return std.fmt.hexToBytes(buf, hex); +} + +/// `out` outlives the per-case arenas, so it is grown with the outer allocator. +fn emit( + gpa: std.mem.Allocator, + out: *std.ArrayListUnmanaged(u8), + id: []const u8, + verdict: Verdict, +) !void { + try out.print(gpa, "{{\"id\":\"{s}\",\"outcome\":\"{s}\",\"detail\":\"{s}\"}}\n", .{ + id, + verdict.outcome.str(), + verdict.detail, + }); +} diff --git a/src/internal/validator.zig b/src/internal/validator.zig index 4f17c35..cccf106 100644 --- a/src/internal/validator.zig +++ b/src/internal/validator.zig @@ -755,6 +755,26 @@ pub const Validator = struct { } } + /// seed a DID's signing key directly, bypassing DID-document resolution. + /// the conformance harness needs the corpus keypair in the cache so the + /// commit path takes the cache-hit branch instead of skipping on a miss. + pub fn seedSigningKey( + self: *Validator, + did: []const u8, + public_key: zat.multicodec.PublicKey, + resolve_time: i64, + ) !void { + if (public_key.raw.len > 33) return error.KeyTooLong; + var cached: CachedKey = .{ + .key_type = public_key.key_type, + .raw = undefined, + .len = @intCast(public_key.raw.len), + .resolve_time = resolve_time, + }; + @memcpy(cached.raw[0..public_key.raw.len], public_key.raw); + try self.cache.put(did, cached); + } + /// evict a DID's cached signing key (e.g. on #identity event). /// the next commit from this DID will trigger a fresh resolution. pub fn evictKey(self: *Validator, did: []const u8) void { -- 2.51.2