Something went wrong. Try again.
GET /xrpc/tech.waow.typeahead.searchActors typeahead.waow.tech
Something went wrong. Try again.
Zig
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341//! Identity resolution for DIDs received from the relay's firehose `#identity` events.//!//! Per the sync spec, `#identity` is a prod to invalidate cached identity data://! it "does not … reliably indicate what the current state of the identity is",//! and `handle` is optional (the bsky relay omits it on every event sampled,//! 21/21 on 2026-09-25). Identity is not self-certifying, so the consumer does//! its own resolution. For every `#identity` DID this resolver://! 1. fetches the DID document from its authority (plc.directory, or the//! did:web host) — never from an identity cache//! 2. verifies the claimed handle bidirectionally (well-known / DNS TXT must//! point back at the DID)//! 3. POSTs {did, handle, pds} to the worker's /admin/ingest//!//! Until 2026-09-25 step 1 went through slingshot, a cache nobody had prodded://! it answered seconds after the PLC op and wrote the OLD handle for ~20–26% of//! handle changes (sampled), which nothing downstream ever corrected.//!//! A handle that does not verify is retried after 60 s and again after 15 min//! (a fresh domain can lag its DNS or well-known), then recorded as//! `handle.invalid` — the atproto convention the appview also serves — rather//! than keeping a previous handle that no longer resolves to this DID either.//! A failed POST to the worker is retried on the same schedule. A resolution//! failure never drops the DID: we still POST the bare DID so the row exists,//! because the handle-enrichment drain retries `WHERE handle = ''` and can only//! see rows. Dropping it made birdlover.selfhosted.social invisible (2026-08-18).
const std = @import("std");const Io = std.Io;const Allocator = std.mem.Allocator;const mem = std.mem;const json = std.json;const zat = @import("zat");const HttpTransport = zat.HttpTransport;const ingest = @import("ingest.zig");const poolio = @import("poolio");
const log = std.log.scoped(.identity);
/// Delay before each retry of an unverified handle (or a failed POST). Each/// stage has its own single-worker pool, so every queue has a uniform delay.pub const RETRY_DELAYS_SECS = [_]i64{ 60, 15 * 60 };
pub const INVALID_HANDLE = "handle.invalid";
/// Inline DID storage — no allocation, fits the longest realistic DID./// PLC DIDs are ~32 chars; web DIDs can be longer. 96 covers everything we'll see.pub const DID_MAX = 96;
pub const Did = struct { buf: [DID_MAX]u8, len: u8,
pub fn fromSlice(s: []const u8) ?Did { if (s.len == 0 or s.len > DID_MAX) return null; var d: Did = undefined; d.len = @intCast(s.len); @memcpy(d.buf[0..s.len], s); return d; }
pub fn slice(self: *const Did) []const u8 { return self.buf[0..self.len]; }};
pub const Job = struct { did: Did, /// retries already spent; indexes RETRY_DELAYS_SECS for the next one attempt: u8 = 0,};
/// Outcome counters, served on the ingester's /health.pub const Stats = struct { verified: std.atomic.Value(u64) = .init(0), unverified: std.atomic.Value(u64) = .init(0), no_handle: std.atomic.Value(u64) = .init(0), did_failed: std.atomic.Value(u64) = .init(0), retried: std.atomic.Value(u64) = .init(0), post_failed: std.atomic.Value(u64) = .init(0),};pub var stats: Stats = .{};
/// What we learned about a DID, before deciding what to write.pub const Lookup = struct { /// first alsoKnownAs entry with valid handle syntax, lowercased handle: ?[]const u8 = null, pds: ?[]const u8 = null, /// true iff `handle` resolved back to this DID verified: bool = false,};
pub const Decision = struct { event: ingest.ActorEvent, retry: bool,};
/// What to record for a DID, given its DID-doc lookup (null = the DID/// document could not be fetched) and how many retries it has spent.////// Split out from `process` so the rules are testable without HTTP:/// - a verified handle is written, replacing whatever the row had/// - an unverified handle is retried while stages remain, writing only DID/// and pds meanwhile; once they are spent it is recorded as handle.invalid/// - the DID (and any pds) is always recordedpub fn decide(did: []const u8, lookup: ?Lookup, attempt: u8) Decision { const l = lookup orelse return .{ .event = .{ .did = did }, .retry = false }; if (l.handle == null) return .{ .event = .{ .did = did, .pds = l.pds }, .retry = false }; if (l.verified) { return .{ .event = .{ .did = did, .handle = l.handle, .pds = l.pds }, .retry = false }; } if (attempt < RETRY_DELAYS_SECS.len) { return .{ .event = .{ .did = did, .pds = l.pds }, .retry = true }; } return .{ .event = .{ .did = did, .handle = INVALID_HANDLE, .pds = l.pds }, .retry = false };}
pub const Resolver = struct { allocator: Allocator, io: Io, worker_url: []const u8, worker_secret: []const u8, /// one pool per RETRY_DELAYS_SECS stage, wired up by main retry_pools: [RETRY_DELAYS_SECS.len]?*RetryPool = @splat(null),
pub fn init( allocator: Allocator, io: Io, worker_url: []const u8, worker_secret: []const u8, ) Resolver { return .{ .allocator = allocator, .io = io, .worker_url = worker_url, .worker_secret = worker_secret, }; }
pub fn deinit(_: *Resolver) void {}
/// poolio process function — called per dequeued Job. pub fn process(self: *Resolver, io: Io, job: Job) void { _ = io; const did_slice = job.did.slice();
var arena = std.heap.ArenaAllocator.init(self.allocator); defer arena.deinit();
const lookup = self.fetchLookup(arena.allocator(), did_slice) catch |err| blk: { log.warn("resolve {s}: {s}", .{ did_slice, @errorName(err) }); _ = stats.did_failed.fetchAdd(1, .monotonic); break :blk null; };
const d = decide(did_slice, lookup, job.attempt); if (lookup) |l| { if (l.handle == null) { _ = stats.no_handle.fetchAdd(1, .monotonic); } else if (l.verified) { _ = stats.verified.fetchAdd(1, .monotonic); } else if (!d.retry) { _ = stats.unverified.fetchAdd(1, .monotonic); log.info("handle unverified after retries {s}: {s}", .{ did_slice, l.handle.? }); } } var retry_scheduled = false; if (d.retry) retry_scheduled = self.scheduleRetry(job);
// Pool workers share Resolver as their context. std.http.Client is not // safe to use concurrently, so each worker invocation owns its client. var transport = HttpTransport.init(self.io, self.allocator); transport.keep_alive = false; defer transport.deinit(); self.postIdentity(&transport, d.event) catch |err| { _ = stats.post_failed.fetchAdd(1, .monotonic); log.warn("post identity {s}: {s}", .{ did_slice, @errorName(err) }); if (!retry_scheduled) _ = self.scheduleRetry(job); }; }
/// Queue `job` for its next retry stage; false once the stages are spent /// or the stage's queue is full. fn scheduleRetry(self: *Resolver, job: Job) bool { if (job.attempt >= RETRY_DELAYS_SECS.len) return false; const rp = self.retry_pools[job.attempt] orelse return false; const queued = rp.offer(self.io, .{ .job = .{ .did = job.did, .attempt = job.attempt + 1 }, .due = ingest.timestamp() + RETRY_DELAYS_SECS[job.attempt], }); if (queued) _ = stats.retried.fetchAdd(1, .monotonic); return queued; }
/// Resolve the DID document from its authority and verify its handle. /// Strings in the result are owned by `arena`. fn fetchLookup(self: *Resolver, arena: Allocator, did_slice: []const u8) !Lookup { const did = zat.Did.parse(did_slice) orelse return error.InvalidDid;
var did_resolver = zat.DidResolver.initWithOptions(self.io, arena, .{ .keep_alive = false }); defer did_resolver.deinit(); var doc = try did_resolver.resolve(did); defer doc.deinit();
var out: Lookup = .{}; if (doc.pdsEndpoint()) |pds| out.pds = try arena.dupe(u8, pds); const claimed = firstHandle(doc.handles) orelse return out; out.handle = try std.ascii.allocLowerString(arena, claimed.str());
var handle_resolver = zat.HandleResolver.init(self.io, arena); handle_resolver.transport.keep_alive = false; defer handle_resolver.deinit(); const resolved = handle_resolver.resolve(claimed) catch |err| { log.debug("verify {s} -> {s}: {s}", .{ out.handle.?, did_slice, @errorName(err) }); return out; }; out.verified = mem.eql(u8, resolved, did_slice); return out; }
fn postIdentity(self: *Resolver, transport: *HttpTransport, event: ingest.ActorEvent) !void { var output: std.Io.Writer.Allocating = .init(self.allocator); defer output.deinit();
var jw: json.Stringify = .{ .writer = &output.writer, .options = .{ .emit_null_optional_fields = false }, };
// single-event batch, no cursor (we don't want to clobber the relay cursor) try jw.write(.{ .events = &[_]ingest.ActorEvent{event} });
var url_buf: [512]u8 = undefined; const url = try std.fmt.bufPrint(&url_buf, "{s}/admin/ingest", .{self.worker_url});
var auth_buf: [256]u8 = undefined; const auth = try std.fmt.bufPrint(&auth_buf, "Bearer {s}", .{self.worker_secret});
const result = try transport.fetch(ingest.workerPost(url, output.written(), auth)); defer self.allocator.free(result.body);
if (result.status != .ok) { log.warn("post identity HTTP {d}: {s}", .{ @backingInt(result.status), result.body }); return error.HttpError; } }};
/// The handle a DID document claims: its first alsoKnownAs entry that is/// valid handle syntax. zat strips `at://` but keeps other URI schemes, so/// those are skipped here by failing handle syntax.fn firstHandle(handles: []const []const u8) ?zat.Handle { for (handles) |h| { if (zat.Handle.parse(h)) |parsed| return parsed; } return null;}
pub const Delayed = struct { job: Job, due: i64,};
/// Holds a job until its due time, then re-offers it to the resolver pool./// One pool per retry stage, each with one worker: every item in a pool has/// the same delay, so its FIFO is already in due order and sleeping on the/// head is exact.pub const Retrier = struct { pool: *Pool,
pub fn process(self: *Retrier, io: Io, item: Delayed) void { const wait = item.due - ingest.timestamp(); if (wait > 0) io.sleep(Io.Duration.fromSeconds(wait), .awake) catch return; _ = self.pool.offer(io, item.job); }};
/// Convenience aliases — the poolio instantiations used throughout the ingester.pub const Pool = poolio.Pool(Job, Resolver);pub const RetryPool = poolio.Pool(Delayed, Retrier);
const testing = std.testing;
test "decide: an unreachable DID document still records the DID" { // Regression: dropping here meant a DID whose identity could not be // resolved never became a row — and the handle drain, which retries // `WHERE handle = ''`, only ever sees rows. birdlover.selfhosted.social sat // in the relay with a healthy repo and was absent from the corpus entirely. const d = decide("did:plc:iu2klrrbxplyyqgltk7sg5l3", null, 0); try testing.expectEqualStrings("did:plc:iu2klrrbxplyyqgltk7sg5l3", d.event.did); try testing.expect(d.event.handle == null); try testing.expect(d.event.pds == null); try testing.expect(!d.retry);}
test "decide: a verified handle change is written immediately" { // Regression (2026-09-25): moltrust.bsky.social -> moltrust.ch was written // back as the OLD handle because resolution went through a cache. The // write must carry the handle the authority now claims and that verifies. const d = decide("did:plc:bjrgfiemnynipuvv4h6md4qi", .{ .handle = "moltrust.ch", .pds = "https://morel.us-east.host.bsky.network", .verified = true, }, 0); try testing.expectEqualStrings("moltrust.ch", d.event.handle.?); try testing.expectEqualStrings("https://morel.us-east.host.bsky.network", d.event.pds.?); try testing.expect(!d.retry);}
test "decide: an unverified handle is retried per stage, then recorded as handle.invalid" { // Regression (2026-09-26): keeping the previous handle on a failed // verification served a name that no longer resolves to the DID either; // the appview serves handle.invalid. streetnewzvlogger.sprk.so failed its // first check and verified minutes later, hence more than one stage. const l: Lookup = .{ .handle = "new.example.com", .pds = "https://selfhosted.social", .verified = false }; var attempt: u8 = 0; while (attempt < RETRY_DELAYS_SECS.len) : (attempt += 1) { const d = decide("did:plc:abc", l, attempt); try testing.expect(d.event.handle == null); try testing.expectEqualStrings("https://selfhosted.social", d.event.pds.?); try testing.expect(d.retry); } const last = decide("did:plc:abc", l, attempt); try testing.expectEqualStrings(INVALID_HANDLE, last.event.handle.?); try testing.expectEqualStrings("https://selfhosted.social", last.event.pds.?); try testing.expect(!last.retry);}
test "decide: a document without a handle records DID and pds, no retry" { const d = decide("did:plc:abc", .{ .pds = "https://pds.example" }, 0); try testing.expect(d.event.handle == null); try testing.expectEqualStrings("https://pds.example", d.event.pds.?); try testing.expect(!d.retry);}
test "firstHandle skips non-handle alsoKnownAs entries" { const handles = [_][]const u8{ "https://example.com/profile", "alice.example.com" }; try testing.expectEqualStrings("alice.example.com", firstHandle(&handles).?.str()); const none = [_][]const u8{"https://example.com/profile"}; try testing.expect(firstHandle(&none) == null);}