diff --git a/src/broadcaster.zig b/src/broadcaster.zig index 9571619..f7907db 100644 --- a/src/broadcaster.zig +++ b/src/broadcaster.zig @@ -31,6 +31,7 @@ pub const Stats = struct { cache_hits: std.atomic.Value(u64) = .{ .raw = 0 }, cache_misses: std.atomic.Value(u64) = .{ .raw = 0 }, slow_consumers: std.atomic.Value(u64) = .{ .raw = 0 }, + connected_inbound: std.atomic.Value(u64) = .{ .raw = 0 }, start_time: i64 = 0, }; @@ -541,6 +542,9 @@ pub fn formatPrometheusMetrics(stats: *const Stats, buf: []u8) []const u8 { \\# TYPE relay_slow_consumers_total counter \\relay_slow_consumers_total {d} \\ + \\# TYPE relay_connected_inbound gauge + \\relay_connected_inbound {d} + \\ \\# TYPE relay_upstream_seq gauge \\relay_upstream_seq {d} \\ @@ -561,6 +565,7 @@ pub fn formatPrometheusMetrics(stats: *const Stats, buf: []u8) []const u8 { stats.cache_hits.load(.acquire), stats.cache_misses.load(.acquire), stats.slow_consumers.load(.acquire), + stats.connected_inbound.load(.acquire), stats.seq.load(.acquire), stats.relay_seq.load(.acquire), uptime, @@ -570,11 +575,12 @@ pub fn formatPrometheusMetrics(stats: *const Stats, buf: []u8) []const u8 { pub fn formatStatsResponse(stats: *const Stats, buf: []u8) []const u8 { var json_buf: [2048]u8 = undefined; const json = std.fmt.bufPrint(&json_buf, - \\{{"seq":{d},"relay_seq":{d},"consumers":{d},"frames_in":{d},"frames_out":{d},"validated":{d},"failed":{d},"skipped":{d},"decode_errors":{d},"cache_hits":{d},"cache_misses":{d},"slow_consumers":{d},"uptime_seconds":{d}}} + \\{{"seq":{d},"relay_seq":{d},"consumers":{d},"connected_inbound":{d},"frames_in":{d},"frames_out":{d},"validated":{d},"failed":{d},"skipped":{d},"decode_errors":{d},"cache_hits":{d},"cache_misses":{d},"slow_consumers":{d},"uptime_seconds":{d}}} , .{ stats.seq.load(.acquire), stats.relay_seq.load(.acquire), stats.consumer_count.load(.acquire), + stats.connected_inbound.load(.acquire), stats.frames_in.load(.acquire), stats.frames_out.load(.acquire), stats.validated.load(.acquire), @@ -689,6 +695,7 @@ test "formatPrometheusMetrics produces valid output" { try std.testing.expect(std.mem.indexOf(u8, output, "relay_consumers_active 3") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_validation_total{result=\"validated\"} 500") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_validation_total{result=\"failed\"} 2") != null); + try std.testing.expect(std.mem.indexOf(u8, output, "relay_connected_inbound 0") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_upstream_seq 99999") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_seq 12345") != null); try std.testing.expect(std.mem.indexOf(u8, output, "# TYPE relay_uptime_seconds gauge") != null); diff --git a/src/slurper.zig b/src/slurper.zig index 40ec0eb..0c61701 100644 --- a/src/slurper.zig +++ b/src/slurper.zig @@ -237,11 +237,24 @@ pub const Slurper = struct { }; } - /// start the slurper: load active hosts from DB, spawn workers, start crawl processor. - /// the seed host is always added if no hosts exist yet. + /// start the slurper: bootstrap hosts from seed relay, load from DB, spawn workers. + /// Go relay: pull-hosts bootstraps from bsky.network's listHosts, then crawls each PDS directly. pub fn start(self: *Slurper) !void { - // ensure seed host exists in DB - _ = try self.persist.getOrCreateHost(self.options.seed_host); + // bootstrap: if DB has no active hosts, pull from seed relay's listHosts API + const existing = try self.persist.listActiveHosts(self.allocator); + const need_bootstrap = existing.len == 0; + for (existing) |h| { + self.allocator.free(h.hostname); + self.allocator.free(h.status); + } + self.allocator.free(existing); + + if (need_bootstrap) { + log.info("no active hosts in DB, bootstrapping from {s}", .{self.options.seed_host}); + self.pullHosts() catch |err| { + log.warn("bootstrap from {s} failed: {s}", .{ self.options.seed_host, @errorName(err) }); + }; + } // load all active hosts and spawn workers const hosts = try self.persist.listActiveHosts(self.allocator); @@ -265,6 +278,100 @@ pub const Slurper = struct { self.crawl_thread = try std.Thread.spawn(.{}, processCrawlQueue, .{self}); } + /// pull PDS host list from the seed relay's com.atproto.sync.listHosts endpoint. + /// Go relay: cmd/relay/pull.go — one-time bootstrap, reads REST API, not firehose. + pub fn pullHosts(self: *Slurper) !void { + var cursor: ?[]const u8 = null; + var total: usize = 0; + const limit = 500; + + var client: http.Client = .{ .allocator = self.allocator }; + defer client.deinit(); + + while (true) { + if (self.shutdown.load(.acquire)) break; + + // build URL with pagination + var url_buf: [512]u8 = undefined; + const url = if (cursor) |c| + std.fmt.bufPrint(&url_buf, "https://{s}/xrpc/com.atproto.sync.listHosts?limit={d}&cursor={s}", .{ self.options.seed_host, limit, c }) catch break + else + std.fmt.bufPrint(&url_buf, "https://{s}/xrpc/com.atproto.sync.listHosts?limit={d}", .{ self.options.seed_host, limit }) catch break; + + var aw: std.Io.Writer.Allocating = .init(self.allocator); + defer aw.deinit(); + + const result = client.fetch(.{ + .location = .{ .url = url }, + .response_writer = &aw.writer, + .method = .GET, + }) catch |err| { + log.warn("pullHosts: fetch failed: {s}", .{@errorName(err)}); + break; + }; + + if (result.status != .ok) { + log.warn("pullHosts: got status {d}", .{@intFromEnum(result.status)}); + break; + } + + const body = aw.toArrayList().items; + + // parse JSON response: { "hosts": [{"hostname": "...", "status": "..."}, ...], "cursor": "..." } + const parsed = std.json.parseFromSlice(ListHostsResponse, self.allocator, body, .{ .ignore_unknown_fields = true }) catch |err| { + log.warn("pullHosts: JSON parse failed: {s}", .{@errorName(err)}); + break; + }; + defer parsed.deinit(); + + const hosts = parsed.value.hosts orelse break; + if (hosts.len == 0) break; + + var added: usize = 0; + for (hosts) |host| { + // skip non-active hosts + if (host.status) |s| { + if (!std.mem.eql(u8, s, "active")) continue; + } + // validate hostname format (rejects IPs, localhost, etc.) + const normalized = validateHostname(self.allocator, host.hostname) catch continue; + defer self.allocator.free(normalized); + + // skip banned domains + if (self.persist.isDomainBanned(normalized)) continue; + + // insert into DB (no describeServer check — the seed relay already vetted them) + _ = self.persist.getOrCreateHost(normalized) catch continue; + added += 1; + } + total += added; + log.info("pullHosts: page fetched, {d} hosts added ({d} total)", .{ added, total }); + + // advance cursor + if (parsed.value.cursor) |next_cursor| { + // free previous cursor if we allocated one + if (cursor) |prev| self.allocator.free(prev); + cursor = self.allocator.dupe(u8, next_cursor) catch break; + } else { + break; // no more pages + } + } + + // free final cursor + if (cursor) |c| self.allocator.free(c); + log.info("pullHosts: bootstrap complete, {d} hosts added from {s}", .{ total, self.options.seed_host }); + } + + const ListHostsResponse = struct { + hosts: ?[]const ListHostEntry = null, + cursor: ?[]const u8 = null, + }; + + const ListHostEntry = struct { + hostname: []const u8, + status: ?[]const u8 = null, + }; + /// add a crawl request (from requestCrawl endpoint) pub fn addCrawlRequest(self: *Slurper, hostname: []const u8) !void { const duped = try self.allocator.dupe(u8, hostname); @@ -372,6 +479,7 @@ pub const Slurper = struct { .thread = thread, .subscriber = sub, }); + _ = self.bc.stats.connected_inbound.fetchAdd(1, .monotonic); } /// worker thread wrapper — runs subscriber, cleans up on exit @@ -382,6 +490,7 @@ pub const Slurper = struct { self.workers_mutex.lock(); defer self.workers_mutex.unlock(); _ = self.workers.remove(host_id); + _ = self.bc.stats.connected_inbound.fetchSub(1, .monotonic); log.info("worker for host_id={d} ({s}) exited", .{ host_id, sub.options.hostname }); diff --git a/src/subscriber.zig b/src/subscriber.zig index 9338b07..30eb597 100644 --- a/src/subscriber.zig +++ b/src/subscriber.zig @@ -18,7 +18,6 @@ const log = std.log.scoped(.relay); const max_consecutive_failures = 15; const cursor_flush_interval_sec = 4; // flush cursor to DB every N seconds (Go relay: 4s) -const default_rate_limit: u64 = 100; // events per second per host (Go relay: 50/sec baseline) pub const Options = struct { hostname: []const u8 = "bsky.network", @@ -39,11 +38,6 @@ pub const Subscriber = struct { // per-host shutdown (e.g. FutureCursor — stops only this subscriber) host_shutdown: std.atomic.Value(bool) = .{ .raw = false }, - // per-host rate limiting (token bucket) - rate_tokens: u64 = default_rate_limit, - rate_last_refill: i64 = 0, - rate_dropped: u64 = 0, - pub fn init( allocator: Allocator, bc: *broadcaster.Broadcaster, @@ -236,25 +230,6 @@ const FrameHandler = struct { } } - // per-host rate limiting (token bucket, refills once per second) - // Go relay: sliding window limiters per host (50/sec baseline) - { - const now = std.time.timestamp(); - if (now > sub.rate_last_refill) { - sub.rate_tokens = default_rate_limit; - sub.rate_last_refill = now; - if (sub.rate_dropped > 0) { - log.warn("host {s}: rate limited, dropped {d} events in last window", .{ sub.options.hostname, sub.rate_dropped }); - sub.rate_dropped = 0; - } - } - if (sub.rate_tokens == 0) { - sub.rate_dropped += 1; - return; - } - sub.rate_tokens -= 1; - } - // route by frame type const is_commit = std.mem.eql(u8, frame_type, "#commit"); const is_account = std.mem.eql(u8, frame_type, "#account");