Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323//! named, revocable archive API keys with per-key byte-rate budgets.//!//! extension beyond upstream: the OSS server has NO archive auth at all//! (its only bearer surface is the single timestamp-import token,//! xrpcapi/auth.go); Bluesky gates and meters hosted archives in a//! proprietary gateway in front — 2 MB/s per key by default//! (alex.bsky.team, 2026-08-17), with the gateway existing precisely to//! adjust per-key limits. this ring approximates that gateway in-process://! named keys, per-key byte budgets, and revocation without rotating the//! fleet credential. recorded in docs/semantic-parity.md.//!//! the key file is one entry per line://!//! name:key[:mbps]//!//! `#` comments and blank lines are ignored; a missing or 0 mbps means//! unmetered. revocation is deleting the line — the file's mtime is//! re-checked (at most once per second) on the request path, so no restart//! or signal is needed. the legacy single fleet key (--archive-api-key)//! keeps working beside the ring, unmetered.//!//! metering is a token bucket in bytes, charged AFTER each response (the//! bucket may go negative; the next request is refused with 429 until the//! deficit refills). burst capacity is 4 seconds of rate. presented keys//! are compared as sha256 digests in constant time, same as authorized().
const std = @import("std");
const Io = std.Io;const Allocator = std.mem.Allocator;const log = std.log.scoped(.api_keys);
/// bucket capacity in seconds of rate. wide on purpose: with post-hoc/// charging, a small burst makes a saturating drain oscillate serve/429/// on a seconds cadence (observed as client-side 429 log spam within/// minutes of the first metered consumer, 2026-08-17). 30s of burst keeps/// the same average rate with coarse, quiet cycles.pub const burst_seconds = 30;/// stat the key file at most this oftenconst reload_check_interval_us: i64 = std.time.us_per_s;
pub const Entry = struct { name: []const u8, key_hash: [32]u8, /// bytes per second; 0 = unmetered rate_bps: u64, /// token bucket level; negative = in deficit bucket_bytes: i64, last_refill_us: i64, // per-key observability (rendered into /metrics; carried across // reloads with the bucket) served_bytes_total: u64 = 0, requests_total: u64 = 0, throttled_total: u64 = 0,};
pub const KeyRing = struct { allocator: Allocator, path: []const u8, mutex: Io.Mutex = .init, entries: std.ArrayList(Entry) = .empty, arena: std.heap.ArenaAllocator, loaded_mtime_us: i64 = std.math.minInt(i64), last_check_us: i64 = std.math.minInt(i64),
pub fn init(allocator: Allocator, path: []const u8) KeyRing { return .{ .allocator = allocator, .path = path, .arena = std.heap.ArenaAllocator.init(allocator), }; }
pub fn deinit(self: *KeyRing) void { self.entries.deinit(self.allocator); self.arena.deinit(); self.* = undefined; }
/// resolve a bearer header to a live key entry. reloads the file when /// its mtime changed (checked at most once per second), so adding or /// deleting a line takes effect without a restart. returns null for /// missing/unknown/revoked keys. pub fn authorize(self: *KeyRing, io: Io, authorization: ?[]const u8, now_us: i64) ?*Entry { const raw = authorization orelse return null; const prefix = "Bearer "; if (raw.len <= prefix.len or !std.ascii.eqlIgnoreCase(raw[0..prefix.len], prefix)) return null; var presented_hash: [32]u8 = undefined; std.crypto.hash.sha2.Sha256.hash(raw[prefix.len..], &presented_hash, .{});
self.mutex.lockUncancelable(io); defer self.mutex.unlock(io); self.maybeReload(io, now_us); for (self.entries.items) |*entry| { if (std.crypto.timing_safe.eql([32]u8, entry.key_hash, presented_hash)) return entry; } return null; }
/// deficit check before serving. returns seconds to wait when the /// key's bucket is negative (serve a 429), null when clear to serve. pub fn precheck(self: *KeyRing, io: Io, entry: *Entry, now_us: i64) ?u64 { self.mutex.lockUncancelable(io); defer self.mutex.unlock(io); if (entry.rate_bps == 0) return null; refill(entry, now_us); if (entry.bucket_bytes >= 0) return null; entry.throttled_total += 1; const deficit: u64 = @intCast(-entry.bucket_bytes); return @max(1, deficit / entry.rate_bps); }
/// charge served bytes to the key after the response. the bucket may /// go negative — one oversized response is always allowed, and the /// deficit gates the next one. pub fn charge(self: *KeyRing, io: Io, entry: *Entry, bytes: u64, now_us: i64) void { self.mutex.lockUncancelable(io); defer self.mutex.unlock(io); entry.requests_total += 1; entry.served_bytes_total +|= bytes; if (entry.rate_bps == 0) return; refill(entry, now_us); entry.bucket_bytes -= @intCast(@min(bytes, std.math.maxInt(i64))); }
/// prometheus lines for every key: consumption, throttling, and live /// bucket headroom. key names are operator-chosen and bounded, so the /// label cardinality is too. pub fn renderMetrics(self: *KeyRing, io: Io, w: anytype) void { self.mutex.lockUncancelable(io); defer self.mutex.unlock(io); if (self.entries.items.len == 0) return; w.print("# TYPE stream_archive_key_served_bytes_total counter\n", .{}) catch return; for (self.entries.items) |e| w.print("stream_archive_key_served_bytes_total{{key=\"{s}\"}} {d}\n", .{ e.name, e.served_bytes_total }) catch return; w.print("# TYPE stream_archive_key_requests_total counter\n", .{}) catch return; for (self.entries.items) |e| w.print("stream_archive_key_requests_total{{key=\"{s}\"}} {d}\n", .{ e.name, e.requests_total }) catch return; w.print("# TYPE stream_archive_key_throttled_total counter\n", .{}) catch return; for (self.entries.items) |e| w.print("stream_archive_key_throttled_total{{key=\"{s}\"}} {d}\n", .{ e.name, e.throttled_total }) catch return; w.print("# TYPE stream_archive_key_bucket_bytes gauge\n", .{}) catch return; for (self.entries.items) |e| w.print("stream_archive_key_bucket_bytes{{key=\"{s}\"}} {d}\n", .{ e.name, e.bucket_bytes }) catch return; w.print("# TYPE stream_archive_key_rate_bps gauge\n", .{}) catch return; for (self.entries.items) |e| w.print("stream_archive_key_rate_bps{{key=\"{s}\"}} {d}\n", .{ e.name, e.rate_bps }) catch return; }
fn refill(entry: *Entry, now_us: i64) void { const elapsed_us = now_us - entry.last_refill_us; entry.last_refill_us = now_us; if (elapsed_us <= 0) return; const cap: i64 = @intCast(entry.rate_bps * burst_seconds); const gained: i64 = @intCast(@min( (@as(u128, @intCast(elapsed_us)) * entry.rate_bps) / std.time.us_per_s, @as(u128, @intCast(std.math.maxInt(i64))), )); entry.bucket_bytes = @min(entry.bucket_bytes +| gained, cap); }
fn maybeReload(self: *KeyRing, io: Io, now_us: i64) void { if (now_us -| self.last_check_us < reload_check_interval_us) return; self.last_check_us = now_us; const cwd = Io.Dir.cwd(); var file = cwd.openFile(io, self.path, .{}) catch |err| { // a missing/unreadable file revokes everything — fail closed, // loudly if (self.entries.items.len > 0) { log.warn("key file {s} unreadable ({s}); all ring keys revoked", .{ self.path, @errorName(err) }); self.entries.clearRetainingCapacity(); self.loaded_mtime_us = std.math.minInt(i64); } return; }; defer file.close(io); const stat = file.stat(io) catch return; const mtime_us: i64 = stat.mtime.toMicroseconds(); if (mtime_us == self.loaded_mtime_us) return;
const bytes = cwd.readFileAlloc(io, self.path, self.allocator, .limited(1 << 20)) catch |err| { log.warn("key file {s} read failed: {s}", .{ self.path, @errorName(err) }); return; }; defer self.allocator.free(bytes); self.loadFromSlice(bytes) catch |err| { log.warn("key file {s} parse failed: {s}; keeping previous ring", .{ self.path, @errorName(err) }); return; }; self.loaded_mtime_us = mtime_us; log.info("archive key ring loaded: {d} keys from {s}", .{ self.entries.items.len, self.path }); }
/// parse `name:key[:mbps]` lines, replacing the ring. buckets of keys /// that survive the reload keep their level (revoke-and-re-add resets). pub fn loadFromSlice(self: *KeyRing, bytes: []const u8) !void { var fresh: std.ArrayList(Entry) = .empty; errdefer fresh.deinit(self.allocator); var next_arena = std.heap.ArenaAllocator.init(self.allocator); errdefer next_arena.deinit();
var lines = std.mem.tokenizeScalar(u8, bytes, '\n'); while (lines.next()) |raw_line| { const line = std.mem.trim(u8, raw_line, " \t\r"); if (line.len == 0 or line[0] == '#') continue; var parts = std.mem.splitScalar(u8, line, ':'); const name = parts.next() orelse return error.MalformedKeyLine; const key = parts.next() orelse return error.MalformedKeyLine; if (name.len == 0 or key.len == 0) return error.MalformedKeyLine; const rate_bps: u64 = if (parts.next()) |mbps_str| (std.fmt.parseInt(u64, mbps_str, 10) catch return error.MalformedKeyLine) * 1000 * 1000 else 0; var hash: [32]u8 = undefined; std.crypto.hash.sha2.Sha256.hash(key, &hash, .{}); var entry: Entry = .{ .name = try next_arena.allocator().dupe(u8, name), .key_hash = hash, .rate_bps = rate_bps, .bucket_bytes = @intCast(rate_bps * burst_seconds), .last_refill_us = 0, }; for (self.entries.items) |old| { if (std.crypto.timing_safe.eql([32]u8, old.key_hash, hash)) { entry.bucket_bytes = old.bucket_bytes; entry.last_refill_us = old.last_refill_us; entry.served_bytes_total = old.served_bytes_total; entry.requests_total = old.requests_total; entry.throttled_total = old.throttled_total; break; } } try fresh.append(self.allocator, entry); } self.entries.deinit(self.allocator); self.entries = fresh; self.arena.deinit(); self.arena = next_arena; }};
// === tests ===
const testing = std.testing;
test "loadFromSlice: names, rates, comments, revocation, bucket carry-over" { var ring = KeyRing.init(testing.allocator, "/nonexistent"); defer ring.deinit();
try ring.loadFromSlice( \\# fleet consumers \\evelyn:sekrit-key-1:8 \\ci:other-key ); try testing.expectEqual(@as(usize, 2), ring.entries.items.len); try testing.expectEqualStrings("evelyn", ring.entries.items[0].name); try testing.expectEqual(@as(u64, 8_000_000), ring.entries.items[0].rate_bps); try testing.expectEqual(@as(u64, 0), ring.entries.items[1].rate_bps);
// drain some budget, then reload with the key surviving: level carries ring.entries.items[0].bucket_bytes = -5_000_000; try ring.loadFromSlice("evelyn:sekrit-key-1:8\n"); try testing.expectEqual(@as(usize, 1), ring.entries.items.len); try testing.expectEqual(@as(i64, -5_000_000), ring.entries.items[0].bucket_bytes);
// revocation: the line is gone, the key is gone try ring.loadFromSlice("ci:other-key\n"); try testing.expectEqual(@as(usize, 1), ring.entries.items.len); try testing.expectEqualStrings("ci", ring.entries.items[0].name);
try testing.expectError(error.MalformedKeyLine, ring.loadFromSlice("no-key-here\n"));}
test "bucket: burst allowed, deficit 429s, refill clears it" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var ring = KeyRing.init(testing.allocator, "/nonexistent"); defer ring.deinit(); try ring.loadFromSlice("e:k:2\n"); // 2 MB/s, 60 MB burst const entry = &ring.entries.items[0]; entry.last_refill_us = 0;
// fresh bucket: clear to serve, then a 300 MB drain lands us in deficit try testing.expectEqual(@as(?u64, null), ring.precheck(io, entry, 1_000_000)); ring.charge(io, entry, 300_000_000, 1_000_000); const wait = ring.precheck(io, entry, 1_000_001) orelse return error.ExpectedDeficit; // ~292 MB deficit at 2 MB/s ≈ 146s try testing.expect(wait > 100 and wait < 200);
// after enough wall clock the bucket refills to burst and serves again try testing.expectEqual(@as(?u64, null), ring.precheck(io, entry, 1_000_000 + 200 * std.time.us_per_s));
// unmetered key never blocks try ring.loadFromSlice("open:k2\n"); const open = &ring.entries.items[0]; ring.charge(io, open, 1 << 40, 5); try testing.expectEqual(@as(?u64, null), ring.precheck(io, open, 6));}
test "renderMetrics: one line per key per series, counters carry across reload" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var ring = KeyRing.init(testing.allocator, "/nonexistent"); defer ring.deinit(); try ring.loadFromSlice("evelyn:k1:16\nci:k2\n"); ring.charge(io, &ring.entries.items[0], 1000, 10); ring.charge(io, &ring.entries.items[0], 500, 20); _ = ring.precheck(io, &ring.entries.items[0], 30); // no throttle: bucket positive
var buf: [4096]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); ring.renderMetrics(io, &w); const out = w.buffered(); try testing.expect(std.mem.indexOf(u8, out, "stream_archive_key_served_bytes_total{key=\"evelyn\"} 1500") != null); try testing.expect(std.mem.indexOf(u8, out, "stream_archive_key_requests_total{key=\"evelyn\"} 2") != null); try testing.expect(std.mem.indexOf(u8, out, "stream_archive_key_rate_bps{key=\"evelyn\"} 16000000") != null); try testing.expect(std.mem.indexOf(u8, out, "stream_archive_key_served_bytes_total{key=\"ci\"} 0") != null);
// reload keeps consumption history for surviving keys try ring.loadFromSlice("evelyn:k1:8\n"); try testing.expectEqual(@as(u64, 1500), ring.entries.items[0].served_bytes_total); try testing.expectEqual(@as(u64, 2), ring.entries.items[0].requests_total);}