atproto relay in zig zlay.waow.tech
relay zig atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355//! frame processing worker — heavy frame handling offloaded from reader threads//!//! the reader thread (subscriber) does lightweight header decode, cursor tracking,//! and rate limiting, then submits raw frame bytes here for heavy processing://! CBOR decode, DID resolution, signature validation, DB persist, broadcast.//!//! double decode (reader + worker both parse CBOR header) is intentional —//! CBOR decode is ~1-2μs, far cheaper than serializing decoded values across threads.
const std = @import("std");const Io = std.Io;const zat = @import("zat");const broadcaster = @import("broadcaster.zig");const validator_mod = @import("validator.zig");const event_log_mod = @import("event_log.zig");const collection_index_mod = @import("collection_index/index.zig");const resync_mod = @import("collection_index/resync.zig");const status_checker_mod = @import("status_checker.zig");const thread_pool = @import("util/thread_pool.zig");const util = @import("util/util.zig");
const Allocator = std.mem.Allocator;const log = std.log.scoped(.relay);const microTimestamp = util.microTimestamp;
pub const FrameWork = struct { data: []u8, // raw frame bytes (heap-duped by reader, freed by worker) host_id: u64, hostname: []const u8, // owned (heap-duped at submit, freed by worker) allocator: Allocator, io: Io, // shared references (all thread-safe, all outlive the work item) bc: *broadcaster.Broadcaster, validator: *validator_mod.Validator, persist: ?*event_log_mod.DiskPersist, collection_index: ?*collection_index_mod.CollectionIndex, resyncer: ?*resync_mod.Resyncer, status_checker: ?*status_checker_mod.StatusChecker = null,};
pub fn processFrame(work: *FrameWork) void { _ = work.bc.stats.pool_queued_bytes.fetchSub(work.data.len, .monotonic); defer work.allocator.free(work.data); defer work.allocator.free(work.hostname);
var arena = std.heap.ArenaAllocator.init(work.allocator); defer arena.deinit(); const alloc = arena.allocator();
const data = work.data;
// re-decode header (cheap — ~1-2μs) const header_result = zat.cbor.decode(alloc, data) catch |err| { log.debug("worker: frame header decode failed: {s} (len={d})", .{ @errorName(err), data.len }); _ = work.bc.stats.decode_errors.fetchAdd(1, .monotonic); return; }; const header = header_result.value; const payload_data = data[header_result.consumed..];
const op = header.getInt("op") orelse return; if (op != 1) return; // only process message frames (error frames handled by reader)
const frame_type = header.getString("t") orelse return; const payload = zat.cbor.decodeAll(alloc, payload_data) catch |err| { log.debug("worker: frame payload decode failed: {s} (type={s})", .{ @errorName(err), frame_type }); _ = work.bc.stats.decode_errors.fetchAdd(1, .monotonic); return; };
// route by frame type — unknown types already filtered by reader const is_commit = std.mem.eql(u8, frame_type, "#commit"); const is_sync = std.mem.eql(u8, frame_type, "#sync"); const is_account = std.mem.eql(u8, frame_type, "#account"); const is_identity = std.mem.eql(u8, frame_type, "#identity");
if (!is_commit and !is_sync and !is_account and !is_identity) return;
// extract DID: "repo" for commits, "did" for identity/account const did: ?[]const u8 = if (is_commit) payload.getString("repo") else payload.getString("did");
// on #identity event, evict cached signing key so next commit re-resolves if (is_identity) { if (did) |d| work.validator.evictKey(d); }
// resolve DID → numeric UID for event header (host-aware) const uid: u64 = if (work.persist) |dp| blk: { if (did) |d| { const result = dp.uidForDidFromHost(d, work.host_id) catch break :blk @as(u64, 0);
// host authority enforcement: verify DID doc confirms this host. // #identity events are exempt — any PDS can emit them (same as indigo). // covers both first-seen DIDs (is_new) and host migrations (host_changed). if ((result.is_new or result.host_changed) and !is_identity) { if (result.is_new) _ = work.bc.stats.host_authority_is_new.fetchAdd(1, .monotonic); if (result.host_changed) _ = work.bc.stats.host_authority_host_changed.fetchAdd(1, .monotonic); _ = work.bc.stats.host_authority_checks.fetchAdd(1, .monotonic); const ha_t0 = microTimestamp(work.io); defer { const elapsed: u64 = @intCast(@max(0, microTimestamp(work.io) - ha_t0)); _ = work.bc.stats.host_authority_time_us.fetchAdd(elapsed, .monotonic); } switch (work.validator.resolveHostAuthority(d, work.host_id, work.hostname, .{ .skip_reject_cache = is_account, })) { .migrate => { // DID doc confirms the host — update and continue dp.setAccountHostId(result.uid, work.host_id) catch {}; if (result.host_changed) { log.info("host {s}: account migrated uid={d} did={s}", .{ work.hostname, result.uid, d, }); } }, .reject => { // DID doc does not confirm — drop the event. Network-confirmed // first occurrences are sampled by Validator; cache hits stay quiet. log.debug("host {s}: dropping event, host authority failed uid={d} did={s}", .{ work.hostname, result.uid, d, }); _ = work.bc.stats.failed.fetchAdd(1, .monotonic); _ = work.bc.stats.failed_host_authority.fetchAdd(1, .monotonic); return; }, .accept => {}, } }
break :blk result.uid; } else break :blk @as(u64, 0); } else 0;
// process #account events: update upstream status if (is_account) { if (work.persist) |dp| { if (uid > 0) { const is_active = payload.getBool("active") orelse false; const status_str = payload.getString("status"); const new_status: []const u8 = if (is_active) "active" else (status_str orelse "inactive"); dp.updateAccountUpstreamStatus(uid, new_status) catch |err| { log.debug("upstream status update failed: {s}", .{@errorName(err)}); };
// on any non-active status, remove all collection index entries // (covers deleted, takendown, suspended, deactivated, and future unknown statuses) if (!is_active) { if (work.collection_index) |ci| { if (did) |d| { ci.removeAll(d) catch |err| { log.debug("collection removeAll failed: {s}", .{@errorName(err)}); }; } } } } } }
// for commits and syncs: check account is active, validate, extract state var commit_data_cid: ?[]const u8 = null; var commit_rev: ?[]const u8 = null; if (is_commit or is_sync) { // drop events for inactive accounts. if (work.persist) |dp| { if (uid > 0) { const activity = dp.accountActivity(uid) catch event_log_mod.DiskPersist.AccountActivity{ .local_ok = true, .upstream_ok = true }; if (!activity.active()) { // a repo event from the account's own confirmed host while // upstream_status is non-active is self-contradicting — the // PDS wouldn't emit it for an account inactive on it. usual // cause: a migration's #account activation was lost. queue a // getRepoStatus re-check (never for local takedowns). if (activity.local_ok and !activity.upstream_ok) { if (work.status_checker) |sc| { if (did) |d| sc.enqueue(uid, d, work.hostname); } } _ = work.bc.stats.skipped.fetchAdd(1, .monotonic); return; } } }
// rev checks: stale, future, chain continuity if (is_commit and uid > 0) { if (payload.getString("rev")) |incoming_rev| { // future-rev rejection: drop commits with timestamps too far ahead if (zat.Tid.parse(incoming_rev)) |tid| { const rev_us: i64 = @intCast(tid.timestamp()); const now_us = microTimestamp(work.io); const skew_us: i64 = work.validator.config.rev_clock_skew * 1_000_000; if (rev_us > now_us + skew_us) { log.info("host {s}: dropping future rev uid={d} rev={s}", .{ work.hostname, uid, incoming_rev, }); _ = work.bc.stats.failed.fetchAdd(1, .monotonic); _ = work.bc.stats.failed_future_rev.fetchAdd(1, .monotonic); return; } }
if (work.persist) |dp| { if (dp.getAccountState(uid, alloc) catch null) |prev| { // stale rev check if (std.mem.order(u8, incoming_rev, prev.rev) != .gt) { log.debug("host {s}: dropping stale commit uid={d} rev={s} <= {s}", .{ work.hostname, uid, incoming_rev, prev.rev, }); _ = work.bc.stats.skipped.fetchAdd(1, .monotonic); return; }
// chain continuity: since should match stored rev if (payload.getString("since")) |since| { if (!std.mem.eql(u8, since, prev.rev)) { log.info("host {s}: chain break uid={d} since={s} stored_rev={s}", .{ work.hostname, uid, since, prev.rev, }); _ = work.bc.stats.chain_breaks_since.fetchAdd(1, .monotonic); } }
// With prior repo state, prevData is required for an // inductive Sync 1.1 proof. Missing/null cannot be // repaired by MST inversion, so drop the commit. if (!work.validator.validatePrevDataPresence(payload)) { log.info("host {s}: dropping commit with missing prevData uid={d}", .{ work.hostname, uid, }); return; }
// 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); } } } } } } } }
if (is_commit) { const result = work.validator.validateCommit(payload); if (!result.valid) return; commit_data_cid = result.data_cid; commit_rev = result.commit_rev;
// track collections from commit ops (phase 1 live indexing) if (work.collection_index) |ci| { if (did) |d| { if (payload.get("ops")) |ops| { ci.trackCommitOps(d, ops); } } } } else { // #sync: signature verification only, no ops/MST const result = work.validator.validateSync(payload); if (!result.valid) return; commit_data_cid = result.data_cid; commit_rev = result.commit_rev;
// enqueue collection index resync — #sync means repo discontinuity if (work.resyncer) |r| { if (did) |d| r.enqueue(d, work.hostname); } } }
// determine event kind for persistence const kind: event_log_mod.EvtKind = if (is_commit) .commit else if (is_sync) .sync else if (is_account) .account else .identity;
// persist + resequence + enqueue under ordering lock to guarantee // broadcast_queue insertion order matches seq assignment order. if (work.persist) |dp| { var spins: u64 = 0; while (work.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { spins += 1; std.atomic.spinLoopHint(); } if (spins > 0) { _ = work.bc.stats.persist_order_spins.fetchAdd(spins, .monotonic); }
const relay_seq = dp.persist(kind, uid, data) catch |err| { work.bc.persist_order.store(0, .release); log.warn("persist failed: {s}", .{@errorName(err)}); return; }; work.bc.stats.relay_seq.store(relay_seq, .release);
const broadcast_data = broadcaster.resequenceFrame(alloc, data, relay_seq) orelse data; const owned = work.allocator.dupe(u8, broadcast_data) catch { work.bc.persist_order.store(0, .release); return; }; work.bc.broadcast_queue.push(relay_seq, owned, &work.bc.stats); work.bc.persist_order.store(0, .release);
// update per-DID state outside the ordering lock (Postgres round-trip) if ((is_commit or is_sync) and uid > 0) { if (commit_rev) |rev| { const cid_str: []const u8 = if (commit_data_cid) |cid_raw| zat.multibase.encode(alloc, .base32lower, cid_raw) catch "" else ""; if (dp.updateAccountState(uid, rev, cid_str)) |updated| { if (!updated) { log.debug("host {s}: stale state update lost race uid={d} rev={s}", .{ work.hostname, uid, rev, }); } } else |err| { log.debug("account state update failed: {s}", .{@errorName(err)}); } } } } else { const upstream_seq = payload.getUint("seq") orelse 0; const owned = work.allocator.dupe(u8, data) catch return; work.bc.broadcast_queue.push(upstream_seq, owned, &work.bc.stats); }}
pub const FramePool = thread_pool.ThreadPool(FrameWork, processFrame);