atproto relay in zig zlay.waow.tech
relay zig atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375//! 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.jsonconst 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 sequenceconst 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, });}