fix: align relay semantics with indigo — stale rev drop, new account binding, sig refresh

+34

docs/design.md

··· 187 187 - eliminates thread overhead entirely, scales to 100K+ hosts per process 188 188 - requires rewriting subscriber, resolver, and consumer write paths 189 189 - pg.zig and websocket.zig would need async-compatible forks 190 + 191 + ## deliberate divergences from indigo 192 + 193 + documented policy choices where zlay intentionally differs from the Go relay 194 + (bluesky-social/indigo). these are not bugs — each reflects a tradeoff 195 + appropriate for zlay's architecture. 196 + 197 + ### per-PDS concurrency model 198 + 199 + indigo uses goroutines (M:N scheduling on a small thread pool). zlay uses one 200 + OS thread per host — simple, no async runtime, no event loop. each thread 201 + spends most time blocked in `recv()` with minimal per-frame CPU work. 202 + 203 + observability: prometheus metrics expose thread count, RSS, per-host memory. 204 + the 0.16 `Io` migration (io_uring/kqueue) is the planned optimization path, 205 + replacing OS threads with coroutines. 206 + 207 + ### skip-on-miss validation 208 + 209 + when the validator has no cached signing key for a DID (cache miss or pending 210 + new-account verification), zlay broadcasts the frame while the key resolves 211 + in the background. indigo blocks on DID resolution before forwarding. 212 + 213 + zlay trades a brief trust window for throughput. the window is bounded: 214 + - new accounts trigger async DID doc verification; on mismatch → rejected 215 + - signature failures trigger key eviction + re-resolution (sync spec guidance) 216 + - next commit from the same DID hits the refreshed cache 217 + 218 + ### consumer buffer sizing 219 + 220 + zlay uses an 8K-entry per-consumer ring buffer (vs indigo's 16K-entry channel). 221 + can be tuned independently based on observed `ConsumerTooSlow` disconnect rate. 222 + the ring buffer is lock-free (atomic read/write indices), so the bottleneck is 223 + consumer write throughput, not buffer contention.

+3 -1

src/event_log.zig

··· 253 253 pub const UidResult = struct { 254 254 uid: u64, 255 255 host_changed: bool = false, 256 + is_new: bool = false, 256 257 }; 257 258 258 259 /// resolve a DID to a numeric UID, associating with a host. ··· 264 265 if (host_id > 0) { 265 266 const current_host = self.getAccountHostId(uid) catch 0; 266 267 if (current_host == 0) { 267 - // first encounter: set host_id 268 + // first encounter: set host_id, queue verification 268 269 self.setAccountHostId(uid, host_id) catch {}; 270 + return .{ .uid = uid, .is_new = true }; 269 271 } else if (current_host != host_id) { 270 272 // host mismatch: don't update yet — caller should validate via DID resolution 271 273 log.info("account {s} (uid={d}) host mismatch: current={d} new={d}, queuing migration check", .{ did, uid, current_host, host_id });

+93 -1

src/subscriber.zig

··· 349 349 const uid: u64 = if (sub.persist) |dp| blk: { 350 350 if (did) |d| { 351 351 const result = dp.uidForDidFromHost(d, sub.options.host_id) catch break :blk @as(u64, 0); 352 - if (result.host_changed) { 352 + if (result.host_changed or result.is_new) { 353 353 sub.validator.queueMigrationCheck(d, sub.options.host_id); 354 354 } 355 355 break :blk result.uid; ··· 396 396 if (!active) { 397 397 _ = sub.bc.stats.skipped.fetchAdd(1, .monotonic); 398 398 return; 399 + } 400 + } 401 + } 402 + 403 + // stale rev check (indigo ingest.go:114, rsky utils.rs:77): 404 + // drop commits where rev <= stored rev to prevent duplicates/rollbacks 405 + if (is_commit and uid > 0) { 406 + if (payload.getString("rev")) |incoming_rev| { 407 + if (sub.persist) |dp| { 408 + if (dp.getAccountState(uid, alloc) catch null) |prev| { 409 + if (std.mem.order(u8, incoming_rev, prev.rev) != .gt) { 410 + log.debug("host {s}: dropping stale commit uid={d} rev={s} <= {s}", .{ 411 + sub.options.hostname, uid, incoming_rev, prev.rev, 412 + }); 413 + _ = sub.bc.stats.skipped.fetchAdd(1, .monotonic); 414 + return; 415 + } 416 + } 399 417 } 400 418 } 401 419 } ··· 611 629 612 630 try std.testing.expectEqual(@as(i64, -1), h.getInt("op").?); 613 631 } 632 + 633 + // --- spec conformance tests --- 634 + 635 + test "spec: unknown frame type (op=1, t=#unknown) is ignored" { 636 + // event stream spec: unknown t values must be ignored for forward-compat 637 + const cbor = zat.cbor; 638 + 639 + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); 640 + defer arena.deinit(); 641 + const alloc = arena.allocator(); 642 + 643 + const header: cbor.Value = .{ .map = &.{ 644 + .{ .key = "op", .value = .{ .unsigned = 1 } }, 645 + .{ .key = "t", .value = .{ .text = "#unknown" } }, 646 + } }; 647 + const payload: cbor.Value = .{ .map = &.{ 648 + .{ .key = "did", .value = .{ .text = "did:plc:test123" } }, 649 + .{ .key = "seq", .value = .{ .unsigned = 1 } }, 650 + } }; 651 + 652 + const header_bytes = try cbor.encodeAlloc(alloc, header); 653 + const payload_bytes = try cbor.encodeAlloc(alloc, payload); 654 + 655 + // decode header — verify it's a valid message with unknown type 656 + const h_result = try cbor.decode(alloc, header_bytes); 657 + const h = h_result.value; 658 + 659 + try std.testing.expectEqual(@as(i64, 1), h.getInt("op").?); 660 + const frame_type = h.getString("t").?; 661 + try std.testing.expectEqualStrings("#unknown", frame_type); 662 + 663 + // verify unknown type is NOT one of the known types (this is the filter logic) 664 + const is_commit = std.mem.eql(u8, frame_type, "#commit"); 665 + const is_sync = std.mem.eql(u8, frame_type, "#sync"); 666 + const is_account = std.mem.eql(u8, frame_type, "#account"); 667 + const is_identity = std.mem.eql(u8, frame_type, "#identity"); 668 + try std.testing.expect(!is_commit and !is_sync and !is_account and !is_identity); 669 + 670 + // verify payload still decodes (frame is valid, just ignored) 671 + const p = try cbor.decodeAll(alloc, payload_bytes); 672 + try std.testing.expectEqualStrings("did:plc:test123", p.getString("did").?); 673 + } 674 + 675 + test "spec: error frame (op=-1) is handled, not persisted" { 676 + // event stream spec: op=-1 frames are error notifications from upstream 677 + const cbor = zat.cbor; 678 + 679 + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); 680 + defer arena.deinit(); 681 + const alloc = arena.allocator(); 682 + 683 + const header: cbor.Value = .{ .map = &.{ 684 + .{ .key = "op", .value = .{ .negative = -1 } }, 685 + .{ .key = "t", .value = .{ .text = "#error" } }, 686 + } }; 687 + const err_payload: cbor.Value = .{ .map = &.{ 688 + .{ .key = "error", .value = .{ .text = "FutureCursor" } }, 689 + .{ .key = "message", .value = .{ .text = "cursor is ahead of server" } }, 690 + } }; 691 + 692 + const header_bytes = try cbor.encodeAlloc(alloc, header); 693 + const h_result = try cbor.decode(alloc, header_bytes); 694 + const h = h_result.value; 695 + 696 + // verify op=-1 is detected 697 + const op = h.getInt("op").?; 698 + try std.testing.expectEqual(@as(i64, -1), op); 699 + 700 + // verify error payload decodes correctly 701 + const payload_bytes = try cbor.encodeAlloc(alloc, err_payload); 702 + const p = try cbor.decodeAll(alloc, payload_bytes); 703 + try std.testing.expectEqualStrings("FutureCursor", p.getString("error").?); 704 + try std.testing.expectEqualStrings("cursor is ahead of server", p.getString("message").?); 705 + }

+128 -7

src/validator.zig

··· 190 190 .max_car_size = 10 * 1024, 191 191 }) catch |err| { 192 192 log.debug("sync verification failed for {s}: {s}", .{ did, @errorName(err) }); 193 - _ = self.stats.failed.fetchAdd(1, .monotonic); 194 - return .{ .valid = false, .skipped = false }; 193 + // sync spec: on signature failure, key may have rotated. 194 + // evict cached key and queue re-resolution. skip this frame. 195 + self.evictKey(did); 196 + self.queueResolve(did); 197 + _ = self.stats.skipped.fetchAdd(1, .monotonic); 198 + return .{ .valid = true, .skipped = true }; 195 199 }; 196 200 197 201 _ = self.stats.validated.fetchAdd(1, .monotonic); ··· 235 239 return vr; 236 240 } else |err| { 237 241 log.debug("commit verification failed for {s}: {s}", .{ did, @errorName(err) }); 238 - _ = self.stats.failed.fetchAdd(1, .monotonic); 239 - return .{ .valid = false, .skipped = false }; 242 + // sync spec: on signature failure, key may have rotated. 243 + // evict cached key and queue re-resolution. skip this frame 244 + // (treat as cache miss). next commit will use the refreshed key. 245 + self.evictKey(did); 246 + self.queueResolve(did); 247 + _ = self.stats.skipped.fetchAdd(1, .monotonic); 248 + return .{ .valid = true, .skipped = true }; 240 249 } 241 250 } 242 251 ··· 537 546 persist.setAccountHostId(uid, mc.new_host_id) catch return; 538 547 log.info("migration validated: {s} → host {d} (confirmed by DID doc)", .{ mc.did, mc.new_host_id }); 539 548 } else { 540 - log.warn("migration rejected: {s} claims host {d}, but DID doc says {s} (host {d})", .{ 541 - mc.did, mc.new_host_id, pds_host, resolved_host_id, 542 - }); 549 + // mismatch — reject new accounts, warn on migrations 550 + const uid = persist.uidForDid(mc.did) catch return; 551 + const current_host = persist.getAccountHostId(uid) catch return; 552 + if (current_host == mc.new_host_id) { 553 + // new account: host not confirmed by DID doc → reject 554 + persist.updateAccountUpstreamStatus(uid, "rejected") catch return; 555 + log.warn("new account rejected: {s} on host {d}, DID doc says {s} (host {d})", .{ 556 + mc.did, mc.new_host_id, pds_host, resolved_host_id, 557 + }); 558 + } else { 559 + // migration: host not confirmed (existing warning behavior) 560 + log.warn("migration rejected: {s} claims host {d}, but DID doc says {s} (host {d})", .{ 561 + mc.did, mc.new_host_id, pds_host, resolved_host_id, 562 + }); 563 + } 543 564 } 544 565 } 545 566 ··· 829 850 830 851 try std.testing.expectError(error.InvalidFrame, v.checkCommitStructure(payload)); 831 852 } 853 + 854 + // --- spec conformance tests --- 855 + 856 + test "spec: #commit blocks > 2,000,000 bytes rejected" { 857 + // lexicon maxLength for #commit blocks: 2,000,000 858 + var stats = broadcaster.Stats{}; 859 + var v = Validator.init(std.testing.allocator, &stats); 860 + defer v.deinit(); 861 + 862 + // insert a fake cached key so we reach the blocks size check 863 + const did = "did:plc:test123"; 864 + const did_duped = try std.testing.allocator.dupe(u8, did); 865 + try v.cache.put(std.testing.allocator, did_duped, .{ 866 + .key_type = .p256, 867 + .raw = .{0} ** 33, 868 + .len = 33, 869 + .resolve_time = 100, 870 + }); 871 + 872 + // blocks with 2,000,001 bytes (1 byte over limit) 873 + const oversized_blocks = try std.testing.allocator.alloc(u8, 2_000_001); 874 + defer std.testing.allocator.free(oversized_blocks); 875 + @memset(oversized_blocks, 0); 876 + 877 + const payload: zat.cbor.Value = .{ .map = &.{ 878 + .{ .key = "repo", .value = .{ .text = did } }, 879 + .{ .key = "rev", .value = .{ .text = "3k2abcdefghij" } }, 880 + .{ .key = "blocks", .value = .{ .bytes = oversized_blocks } }, 881 + } }; 882 + 883 + const result = v.validateCommit(payload); 884 + try std.testing.expect(!result.valid or result.skipped); 885 + } 886 + 887 + test "spec: #commit blocks = 2,000,000 bytes accepted (boundary)" { 888 + // lexicon maxLength for #commit blocks: 2,000,000 — exactly at limit should pass size check 889 + var stats = broadcaster.Stats{}; 890 + var v = Validator.init(std.testing.allocator, &stats); 891 + defer v.deinit(); 892 + 893 + const did = "did:plc:test123"; 894 + const did_duped = try std.testing.allocator.dupe(u8, did); 895 + try v.cache.put(std.testing.allocator, did_duped, .{ 896 + .key_type = .p256, 897 + .raw = .{0} ** 33, 898 + .len = 33, 899 + .resolve_time = 100, 900 + }); 901 + 902 + // exactly 2,000,000 bytes — should pass size check (may fail signature verify, that's ok) 903 + const exact_blocks = try std.testing.allocator.alloc(u8, 2_000_000); 904 + defer std.testing.allocator.free(exact_blocks); 905 + @memset(exact_blocks, 0); 906 + 907 + const payload: zat.cbor.Value = .{ .map = &.{ 908 + .{ .key = "repo", .value = .{ .text = did } }, 909 + .{ .key = "rev", .value = .{ .text = "3k2abcdefghij" } }, 910 + .{ .key = "blocks", .value = .{ .bytes = exact_blocks } }, 911 + } }; 912 + 913 + const result = v.validateCommit(payload); 914 + // should not be rejected for size — may fail signature verification (that's fine, 915 + // it means we passed the size check). with P1.1c, sig failure → skipped=true. 916 + try std.testing.expect(result.valid or result.skipped); 917 + } 918 + 919 + test "spec: #sync blocks > 10,000 bytes rejected" { 920 + // lexicon maxLength for #sync blocks: 10,000 921 + var stats = broadcaster.Stats{}; 922 + var v = Validator.init(std.testing.allocator, &stats); 923 + defer v.deinit(); 924 + 925 + const payload: zat.cbor.Value = .{ .map = &.{ 926 + .{ .key = "did", .value = .{ .text = "did:plc:test123" } }, 927 + .{ .key = "rev", .value = .{ .text = "3k2abcdefghij" } }, 928 + .{ .key = "blocks", .value = .{ .bytes = &([_]u8{0} ** 10_001) } }, 929 + } }; 930 + 931 + const result = v.validateSync(payload); 932 + try std.testing.expect(!result.valid); 933 + try std.testing.expect(!result.skipped); 934 + } 935 + 936 + test "spec: #sync blocks = 10,000 bytes accepted (boundary)" { 937 + // lexicon maxLength for #sync blocks: 10,000 — exactly at limit should pass size check 938 + var stats = broadcaster.Stats{}; 939 + var v = Validator.init(std.testing.allocator, &stats); 940 + defer v.deinit(); 941 + 942 + const payload: zat.cbor.Value = .{ .map = &.{ 943 + .{ .key = "did", .value = .{ .text = "did:plc:test123" } }, 944 + .{ .key = "rev", .value = .{ .text = "3k2abcdefghij" } }, 945 + .{ .key = "blocks", .value = .{ .bytes = &([_]u8{0} ** 10_000) } }, 946 + } }; 947 + 948 + const result = v.validateSync(payload); 949 + // should pass size check — will be a cache miss → skipped (no cached key) 950 + try std.testing.expect(result.valid); 951 + try std.testing.expect(result.skipped); 952 + }

Configure Feed

Configure Feed