Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819//! Comptime-derived flag and environment plumbing.//!//! Every option is a field on an options struct. Its flag spelling and its//! `JETSTREAM_*` name are *derived* from the field name rather than listed//! separately, so the three things that used to drift independently — the//! parser, the environment map, and the usage text — cannot disagree: there is//! only one list, and it is the struct.//!//! That drift was a real, documented gap. `docs/configuration-parity.md` marks//! the flag inventory **unverified** because nothing proved the hand-written//! `environment.zig` map covered every flag the parser accepts. Derivation//! turns that from a promise into a property.//!//! Naming rule: `backfill_workers` -> `--backfill-workers` and//! `JETSTREAM_BACKFILL_WORKERS`. Options whose spelling predates the rule, or//! which upstream declares root-only, carry an entry in the struct's optional//! `overrides` declaration; there is no other way to deviate, so deviations//! stay countable.
const std = @import("std");
pub const prefix = "JETSTREAM_";
/// How one option is exposed. `serve` mirrors upstream's command tree: a/// root-only value has no synthesized `serve` flag.pub const Exposure = struct { flag: []const u8, env: []const u8, serve: bool = true,};
/// `some_field` -> `some-field`pub fn flagName(comptime field: []const u8) []const u8 { return comptime blk: { var out: [field.len]u8 = undefined; for (field, 0..) |c, i| out[i] = if (c == '_') '-' else c; const frozen = out; break :blk &frozen; };}
/// `some_field` -> `JETSTREAM_SOME_FIELD`pub fn envName(comptime field: []const u8) []const u8 { return comptime blk: { var out: [prefix.len + field.len]u8 = undefined; @memcpy(out[0..prefix.len], prefix); for (field, 0..) |c, i| out[prefix.len + i] = std.ascii.toUpper(c); const frozen = out; break :blk &frozen; };}
fn overrideFor(comptime T: type, comptime field: []const u8) ?Exposure { @setEvalBranchQuota(10_000); if (!@hasDecl(T, "overrides")) return null; const ov = T.overrides; if (!@hasField(@TypeOf(ov), field)) return null; const entry = @field(ov, field); const E = @TypeOf(entry); return Exposure{ .flag = if (@hasField(E, "flag")) entry.flag else flagName(field), .env = if (@hasField(E, "env")) entry.env else envName(field), .serve = if (@hasField(E, "serve")) entry.serve else true, };}
/// The exposure of every field on `T`, derived unless overridden.pub fn exposures(comptime T: type) []const Exposure { return comptime blk: { @setEvalBranchQuota(20_000); const names = @typeInfo(T).@"struct".field_names; var out: [names.len]Exposure = undefined; for (names, 0..) |name, i| { out[i] = overrideFor(T, name) orelse .{ .flag = flagName(name), .env = envName(name), }; } const frozen = out; break :blk &frozen; };}
/// Is `key` an environment name this option set declares?pub fn declaresEnv(comptime T: type, key: []const u8) bool { inline for (comptime exposures(T)) |e| { if (std.mem.eql(u8, e.env, key)) return true; } return false;}
/// Is `flag` (without leading dashes) one this option set declares?pub fn declaresFlag(comptime T: type, flag: []const u8) bool { inline for (comptime exposures(T)) |e| { if (std.mem.eql(u8, e.flag, flag)) return true; } return false;}
// === tests ===
const testing = std.testing;
test "field names derive their flag and environment spellings" { try testing.expectEqualStrings("backfill-workers", flagName("backfill_workers")); try testing.expectEqualStrings("JETSTREAM_BACKFILL_WORKERS", envName("backfill_workers")); try testing.expectEqualStrings("addr", flagName("addr")); try testing.expectEqualStrings("JETSTREAM_ADDR", envName("addr"));}
test "overrides are the only way to deviate, and they compose with derivation" { const Opts = struct { addr: []const u8 = "", log_level: []const u8 = "", legacy: []const u8 = "",
pub const overrides = .{ // upstream declares this root-only: no synthesized serve flag .log_level = .{ .serve = false }, // a spelling that predates the naming rule .legacy = .{ .flag = "port", .env = "JETSTREAM_PORT" }, }; }; const e = comptime exposures(Opts); try testing.expectEqualStrings("addr", e[0].flag); try testing.expect(e[0].serve); try testing.expectEqualStrings("log-level", e[1].flag); try testing.expect(!e[1].serve); try testing.expectEqualStrings("port", e[2].flag); try testing.expectEqualStrings("JETSTREAM_PORT", e[2].env); try testing.expect(declaresEnv(Opts, "JETSTREAM_ADDR")); try testing.expect(!declaresEnv(Opts, "JETSTREAM_NOPE")); try testing.expect(declaresFlag(Opts, "port"));}
test "the derived option set reproduces the hand-written environment map" { // This is the point of the module. environment.zig's map is maintained by // hand, and docs/configuration-parity.md marks the flag inventory // unverified for exactly that reason. Deriving the same set from a struct // and asserting equality here makes the two impossible to drift apart -- // and when main.zig moves onto this struct, the map stops existing. const derived = comptime exposures(ServeOptions);
try testing.expectEqual(environment.mappings.len, derived.len); for (environment.mappings) |want| { var found = false; for (derived) |got| { if (!std.mem.eql(u8, want.flag, got.flag)) continue; found = true; try testing.expectEqualStrings(want.env, got.env); try testing.expectEqual(want.serve, got.serve); break; } if (!found) { std.debug.print("no derived option for flag --{s}\n", .{want.flag}); return error.MissingDerivedOption; } }}
/// The pinned Jetstream V2 flag surface, as a struct. Field order matches/// upstream's declaration order so the two read side by side.pub const ServeOptions = struct { log_level: []const u8 = "", log_format: []const u8 = "", addr: []const u8 = "", debug_addr: []const u8 = "", shutdown_timeout: []const u8 = "", client_drain_timeout: []const u8 = "", relay_url: []const u8 = "", plc_url: []const u8 = "", data_dir: []const u8 = "", max_backfill_repos: []const u8 = "", backfill_workers: []const u8 = "", backfill_batch_size: []const u8 = "", backfill_async_flush_workers: []const u8 = "", backfill_repos: []const u8 = "", skip_merge_discovery: []const u8 = "", failed_repo_retry_interval: []const u8 = "", failed_repo_retry_workers: []const u8 = "", failed_repo_retry_host_workers: []const u8 = "", failed_repo_retry_max_delay: []const u8 = "", disable_repo_action_rate_limits: []const u8 = "", cursor_lookback: []const u8 = "", segment_cache_max_age: []const u8 = "", plan_max_dids: []const u8 = "", plan_max_collections: []const u8 = "", plan_max_entries: []const u8 = "", plan_whole_segment_threshold: []const u8 = "", subscribe_read_log_retention_bytes: []const u8 = "", subscribe_block_cache_bytes: []const u8 = "", subscribe_read_batch: []const u8 = "", subscribe_slow_window: []const u8 = "", subscribe_slow_min_rate: []const u8 = "", cursor_block_index_cache_size: []const u8 = "", compaction_interval: []const u8 = "", rebloom_sweep: []const u8 = "", compaction_tombstone_cap: []const u8 = "", compaction_rewrite_workers: []const u8 = "", timestamp_import_token: []const u8 = "", /// bearer credential gating the archive endpoints (planSnapshot, /// listSegments, getSegment, getBlock). Empty (default) leaves the /// archive OPEN — unlike the import token, the archive is an existing /// public surface and gating is the operator's opt-in. Mirrors the /// hosted Bluesky instances' edge behavior (Jetstream v2, 2026-08-13: /// live tail and dictionary open, archive behind a token). archive_api_key: []const u8 = "", archive_api_keys_file: []const u8 = "", /// stream extensions: in-process admission caps (0/absent = unbounded). /// upstream keeps these at its CDN/proxy layer (docs/README.md §7). max_subscribers: []const u8 = "", max_cold_readers: []const u8 = "", /// stream extension: reconnect when the relay delivers fewer frames per /// second than this over a minute (0/absent = never, as upstream). upstream_slow_min_rate: []const u8 = "", timestamp_import_dir: []const u8 = "",
pub const overrides = .{ // Upstream declares these on the root command, so there is no // synthesized `serve` flag for them. .log_level = .{ .serve = false }, .log_format = .{ .serve = false }, };};
// === value parsers ===//// Go-duration, boolean and size syntax, lifted out of main.zig. These are the// shapes upstream's flag library accepts, and they are pure functions of their// input, so they belong beside the flag derivation rather than inline in a// 1,400-line entry point.
const Io = std.Io;const server_mod = @import("../serve/server.zig");
pub fn parseDurationNanoseconds(raw: []const u8) !u64 { return parseDurationMagnitude(raw, std.math.maxInt(i64));}pub fn parseDurationMagnitude(raw: []const u8, limit: u128) !u64 { if (raw.len == 0) return error.InvalidDuration; var i: usize = 0; if (raw[i] == '+') { i += 1; if (i == raw.len) return error.InvalidDuration; } else if (raw[i] == '-') return error.InvalidDuration; if (std.mem.eql(u8, raw[i..], "0")) return 0;
const Unit = struct { name: []const u8, nanoseconds: u64 }; const units = [_]Unit{ .{ .name = "ns", .nanoseconds = 1 }, .{ .name = "us", .nanoseconds = std.time.ns_per_us }, .{ .name = "µs", .nanoseconds = std.time.ns_per_us }, .{ .name = "μs", .nanoseconds = std.time.ns_per_us }, .{ .name = "ms", .nanoseconds = std.time.ns_per_ms }, .{ .name = "s", .nanoseconds = std.time.ns_per_s }, .{ .name = "m", .nanoseconds = std.time.ns_per_min }, .{ .name = "h", .nanoseconds = std.time.ns_per_hour }, }; var total: u128 = 0; var tokens: usize = 0; while (i < raw.len) : (tokens += 1) { var whole: u128 = 0; var digits: usize = 0; while (i < raw.len and std.ascii.isDigit(raw[i])) : (i += 1) { const digit = raw[i] - '0'; if (whole > (limit - digit) / 10) return error.DurationOverflow; whole = whole * 10 + digit; digits += 1; }
var fraction: u128 = 0; var fraction_scale: u128 = 1; if (i < raw.len and raw[i] == '.') { i += 1; while (i < raw.len and std.ascii.isDigit(raw[i])) : (i += 1) { digits += 1; // Nanosecond resolution never needs more than 18 decimal // places for the largest supported unit; Go truncates below // one nanosecond as well. if (fraction_scale < 1_000_000_000_000_000_000) { fraction = fraction * 10 + (raw[i] - '0'); fraction_scale *= 10; } } } if (digits == 0) return error.InvalidDuration;
var unit: ?Unit = null; for (units) |candidate| { if (std.mem.startsWith(u8, raw[i..], candidate.name)) { unit = candidate; i += candidate.name.len; break; } } const selected = unit orelse return error.InvalidDuration; if (whole > limit / selected.nanoseconds) return error.DurationOverflow; var component = whole * selected.nanoseconds; component += (fraction * selected.nanoseconds) / fraction_scale; if (component > limit - total) return error.DurationOverflow; total += component; } if (tokens == 0) return error.InvalidDuration; return @intCast(total);}pub fn parseDurationCeilSeconds(raw: []const u8) !u64 { const nanoseconds = try parseDurationNanoseconds(raw); return (nanoseconds + std.time.ns_per_s - 1) / std.time.ns_per_s;}pub fn parseGoDuration(raw: []const u8) !Io.Duration { if (raw.len > 0 and raw[0] == '-') { if (raw.len == 1 or raw[1] == '+' or raw[1] == '-') return error.InvalidDuration; const negative_limit: u128 = @as(u128, std.math.maxInt(i64)) + 1; const magnitude = try parseDurationMagnitude(raw[1..], negative_limit); return .fromNanoseconds(-@as(i96, magnitude)); } return .fromNanoseconds(@intCast(try parseDurationNanoseconds(raw)));}pub fn effectiveHttpShutdownDuration(configured: Io.Duration) Io.Duration { return if (configured.nanoseconds <= 0) .fromSeconds(30) else configured;}pub fn effectiveSubscribeSlowWindowNanoseconds(raw: []const u8) !u64 { const default = server_mod.default_slow_window_ns; if (raw.len > 0 and raw[0] == '-') { // Go's DurationFlag accepts a negative duration, then subscribe's // defaults replace every non-positive value. Parse the magnitude so // malformed negative values still fail instead of becoming defaults. _ = try parseDurationNanoseconds(raw[1..]); return default; } const configured = try parseDurationNanoseconds(raw); return if (configured == 0) default else configured;}/// stream-extension admission caps (--max-subscribers/--max-cold-readers):/// non-positive means unbounded, matching the pinned surface's sentinel/// convention for optional integer controls.fn parseAdmissionCap(raw: []const u8) !u32 { const configured = try std.fmt.parseInt(i64, raw, 10); if (configured <= 0) return 0; return std.math.cast(u32, configured) orelse error.Overflow;}
fn parseUpstreamSlowMinRate(raw: []const u8) !f64 { const configured = try std.fmt.parseFloat(f64, raw); if (std.math.isNan(configured)) return error.InvalidCharacter; return if (configured <= 0) 0 else configured;}
test "upstream slow min rate uses the non-positive disabled sentinel" { try std.testing.expectEqual(@as(f64, 0), try parseUpstreamSlowMinRate("0")); try std.testing.expectEqual(@as(f64, 0), try parseUpstreamSlowMinRate("-5")); try std.testing.expectEqual(@as(f64, 50), try parseUpstreamSlowMinRate("50")); try std.testing.expectEqual(@as(f64, 7.5), try parseUpstreamSlowMinRate("7.5")); try std.testing.expectError(error.InvalidCharacter, parseUpstreamSlowMinRate("fast")); try std.testing.expectError(error.InvalidCharacter, parseUpstreamSlowMinRate("nan"));}
test "admission caps use the non-positive unbounded sentinel" { try std.testing.expectEqual(@as(u32, 0), try parseAdmissionCap("0")); try std.testing.expectEqual(@as(u32, 0), try parseAdmissionCap("-1")); try std.testing.expectEqual(@as(u32, 500), try parseAdmissionCap("500")); try std.testing.expectError(error.Overflow, parseAdmissionCap("4294967296")); try std.testing.expectError(error.InvalidCharacter, parseAdmissionCap("many"));}
test "subscribe slow window uses the upstream non-positive sentinel" { const default = server_mod.default_slow_window_ns; try std.testing.expectEqual(default, try effectiveSubscribeSlowWindowNanoseconds("0")); try std.testing.expectEqual(default, try effectiveSubscribeSlowWindowNanoseconds("-250ms")); try std.testing.expectEqual(@as(u64, 250 * std.time.ns_per_ms), try effectiveSubscribeSlowWindowNanoseconds("250ms")); try std.testing.expectError(error.InvalidDuration, effectiveSubscribeSlowWindowNanoseconds("-garbage"));}test "shutdown durations preserve Go's signed CLI and HTTP default semantics" { try std.testing.expectEqual(@as(i96, 5 * std.time.ns_per_s), (try parseGoDuration("5s")).nanoseconds); try std.testing.expectEqual(@as(i96, -250 * std.time.ns_per_ms), (try parseGoDuration("-250ms")).nanoseconds); try std.testing.expectEqual(@as(i96, 0), (try parseGoDuration("-0")).nanoseconds); try std.testing.expectEqual(@as(i96, std.math.maxInt(i64)), (try parseGoDuration("9223372036854775807ns")).nanoseconds); try std.testing.expectEqual(@as(i96, std.math.minInt(i64)), (try parseGoDuration("-9223372036854775808ns")).nanoseconds); try std.testing.expectEqual(@as(i96, 30 * std.time.ns_per_s), effectiveHttpShutdownDuration(.zero).nanoseconds); try std.testing.expectEqual(@as(i96, 30 * std.time.ns_per_s), effectiveHttpShutdownDuration(.fromSeconds(-1)).nanoseconds); try std.testing.expectEqual(@as(i96, 3 * std.time.ns_per_s), effectiveHttpShutdownDuration(.fromSeconds(3)).nanoseconds); try std.testing.expectError(error.InvalidDuration, parseGoDuration("-garbage")); try std.testing.expectError(error.InvalidDuration, parseGoDuration("-+1s")); try std.testing.expectError(error.DurationOverflow, parseGoDuration("9223372036854775808ns")); try std.testing.expectError(error.DurationOverflow, parseGoDuration("-9223372036854775809ns"));}test "segment cache duration matches Go duration and ceil policy" { try testing.expectEqual(@as(u64, 0), try parseDurationCeilSeconds("0")); try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds("500ms")); try testing.expectEqual(@as(u64, 2), try parseDurationCeilSeconds("1500ms")); try testing.expectEqual(@as(u64, 3600), try parseDurationCeilSeconds("1h")); try testing.expectEqual(@as(u64, 5400), try parseDurationCeilSeconds("1h30m")); try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds(".1s")); try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds("1us")); try testing.expectEqual(@as(u64, 1), try parseDurationCeilSeconds("1µs")); try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("-1s")); try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("1")); try testing.expectError(error.InvalidDuration, parseDurationCeilSeconds("wat")); try testing.expectEqual(@as(u64, 1_500_000_000), try parseDurationNanoseconds("1.5s"));}pub fn parseNonNegativeUsize(raw: []const u8) !usize { if (raw.len > 0 and raw[0] == '-') return error.BadArgs; return std.fmt.parseInt(usize, raw, 10);}pub fn parseBool(raw: []const u8) !bool { if (std.mem.eql(u8, raw, "1") or std.mem.eql(u8, raw, "t") or std.mem.eql(u8, raw, "T") or std.mem.eql(u8, raw, "true") or std.mem.eql(u8, raw, "TRUE") or std.mem.eql(u8, raw, "True")) return true; if (std.mem.eql(u8, raw, "0") or std.mem.eql(u8, raw, "f") or std.mem.eql(u8, raw, "F") or std.mem.eql(u8, raw, "false") or std.mem.eql(u8, raw, "FALSE") or std.mem.eql(u8, raw, "False")) return false; return error.InvalidBoolean;}test "boolean flags match Go strconv syntax" { try std.testing.expect(try parseBool("true")); try std.testing.expect(try parseBool("T")); try std.testing.expect(try parseBool("1")); try std.testing.expect(!(try parseBool("false"))); try std.testing.expect(!(try parseBool("F"))); try std.testing.expect(!(try parseBool("0"))); try std.testing.expectError(error.InvalidBoolean, parseBool("yes"));}// === serve configuration ===//// The 65-branch flag chain that used to live inside `pub fn main`. Lifting it// here is what lets the environment map disappear: both the parser and the// JETSTREAM_* names now come from one place.
const archive_mod = @import("../storage/archive.zig");const backfill_engine = @import("../ingest/backfill/engine.zig");const lifecycle = @import("../ingest/backfill/lifecycle.zig");const compact_pass = @import("../compact/pass.zig");const retry_mod = @import("../ingest/backfill/retry.zig");const cold = @import("../storage/cold.zig");const xrpcapi = @import("../serve/xrpcapi.zig");const environment = @import("environment.zig");const crashpoint = @import("crashpoint.zig");const log = std.log.scoped(.stream);
pub const default_subscribe_read_log_retention_bytes: usize = 256 * 1024 * 1024;
const RelayUrls = struct { http: []u8, websocket: []u8,
fn deinit(self: RelayUrls, allocator: std.mem.Allocator) void { allocator.free(self.http); allocator.free(self.websocket); }};
pub fn normalizeRelayUrls(allocator: std.mem.Allocator, relay_url: []const u8) !RelayUrls { const secure = std.mem.startsWith(u8, relay_url, "https://"); const prefix_len: usize = if (secure) "https://".len else if (std.mem.startsWith(u8, relay_url, "http://")) "http://".len else return error.InvalidRelayUrl; const rest = relay_url[prefix_len..]; const authority_len = std.mem.indexOfAny(u8, rest, "/?#") orelse rest.len; if (authority_len == 0 or std.mem.indexOfAny(u8, rest[0..authority_len], " \t\r\n") != null) return error.InvalidRelayUrl; const authority = rest[0..authority_len]; const http = try std.fmt.allocPrint(allocator, "{s}://{s}", .{ if (secure) "https" else "http", authority }); errdefer allocator.free(http); return .{ .http = http, .websocket = try std.fmt.allocPrint(allocator, "{s}://{s}", .{ if (secure) "wss" else "ws", authority }), };}
pub const ServeConfig = struct { upstream: []const u8 = "ws://localhost:7777", canonical_relay: ?RelayUrls = null, // the simulator serves DID docs itself; production passes --plc-url. plc_url: []const u8 = "http://localhost:7777", data_dir: []const u8 = "./data", cursor: ?i64 = null, port: u16 = 6008, legacy_port_set: bool = false, public_addr_raw: ?[]const u8 = null, debug_addr_raw: ?[]const u8 = null, shutdown_timeout: Io.Duration = .fromSeconds(5), client_drain_timeout: Io.Duration = .fromSeconds(10), otel_service_name: []const u8 = "jetstream", to_stdout: bool = false, do_verify: bool = true, max_segment_bytes: usize = archive_mod.default_max_segment_bytes, backfill_max_repos: usize = 0, backfill_repos_raw: ?[]const u8 = null, do_backfill: bool = false, backfill_workers: usize = backfill_engine.default_workers, backfill_batch_size: usize = backfill_engine.default_batch_size, backfill_async_flush_workers: usize = lifecycle.default_async_flush_workers, skip_merge_discovery: bool = false, rebloom_sweep: bool = false, relay_http: []const u8 = "http://localhost:7777", compaction_interval_ns: u64 = 4 * std.time.ns_per_hour, compaction_tombstone_cap: usize = compact_pass.default_tombstone_cap, compaction_rewrite_workers: usize = 0, retry_interval_ns: u64 = retry_mod.default_interval_ns, retry_workers: usize = retry_mod.default_workers, retry_host_workers: usize = retry_mod.default_host_workers, retry_max_delay_ns: u64 = retry_mod.default_max_delay_ns, timestamp_import_token: []const u8 = "", /// bearer credential gating the archive endpoints (planSnapshot, /// listSegments, getSegment, getBlock). Empty (default) leaves the /// archive OPEN — unlike the import token, the archive is an existing /// public surface and gating is the operator's opt-in. Mirrors the /// hosted Bluesky instances' edge behavior (Jetstream v2, 2026-08-13: /// live tail and dictionary open, archive behind a token). archive_api_key: []const u8 = "", archive_api_keys_file: []const u8 = "", max_subscribers: u32 = 0, max_cold_readers: u32 = 0, upstream_slow_min_rate: f64 = 0, timestamp_import_dir_arg: ?[]const u8 = null, repo_action_rate_limits: bool = true, store_fault_prefix: ?[]const u8 = null, store_fault_ordinal: u64 = 1, segment_fault_spec: ?[]const u8 = null, segment_cache_max_age_s: u64 = 0, cursor_lookback_ns: u64 = 36 * std.time.ns_per_hour, subscribe_read_log_retention_bytes: usize = default_subscribe_read_log_retention_bytes, subscribe_block_cache_bytes: usize = cold.default_block_cache_bytes, subscribe_slow_window_ns: u64 = server_mod.default_slow_window_ns, subscribe_slow_min_rate: f64 = server_mod.default_slow_min_rate, subscribe_read_batch: usize = server_mod.default_subscribe_read_batch, plan_config: xrpcapi.PlanConfig = .{},
pub fn deinit(self: *ServeConfig, allocator: std.mem.Allocator) void { if (self.canonical_relay) |owned| owned.deinit(allocator); self.canonical_relay = null; }};
pub const Parsed = struct { config: ServeConfig, explicit_serve: bool, log_level: []const u8, log_format: []const u8, /// Flags synthesized from JETSTREAM_* live here. The parsed config borrows /// into them, so they must outlive it: freeing them inside parseServe left /// every environment-supplied value dangling, which surfaced as BadPathName /// on the first path built from one. env_args: std.ArrayList([]u8),
pub fn deinit(self: *Parsed, allocator: std.mem.Allocator) void { self.config.deinit(allocator); for (self.env_args.items) |arg| allocator.free(arg); self.env_args.deinit(allocator); }};
/// Merge environment defaults under the command line, then apply every flag./// Precedence is upstream's: an explicit CLI value wins without the environment/// value even being parsed.pub fn parseServe( allocator: std.mem.Allocator, env_entries: []const []const u8, first_arg: ?[]const u8, arg_it: *std.process.Args.Iterator, explicit_serve_in: bool, log_level_in: []const u8, log_format_in: []const u8,) !Parsed { var explicit_serve = explicit_serve_in; var log_level = log_level_in; var log_format = log_format_in; var cfg: ServeConfig = .{}; errdefer cfg.deinit(allocator); if (explicit_serve) cfg.public_addr_raw = ":8080"; cfg.otel_service_name = environment.value(env_entries, "OTEL_SERVICE_NAME") orelse "jetstream"; if (cfg.otel_service_name.len == 0) cfg.otel_service_name = "jetstream";
var env_serve_args: std.ArrayList([]u8) = .empty; errdefer { for (env_serve_args.items) |arg| allocator.free(arg); env_serve_args.deinit(allocator); } var cli_serve_args: std.ArrayList([]const u8) = .empty; defer cli_serve_args.deinit(allocator); if (first_arg) |arg| try cli_serve_args.append(allocator, arg); while (arg_it.next()) |arg| try cli_serve_args.append(allocator, arg); try environment.appendServeArguments(allocator, env_entries, cli_serve_args.items, &env_serve_args); var serve_args: std.ArrayList([]const u8) = .empty; defer serve_args.deinit(allocator); for (env_serve_args.items) |arg| try serve_args.append(allocator, arg); for (cli_serve_args.items) |arg| try serve_args.append(allocator, arg);
var arg_index: usize = 0; while (arg_index < serve_args.items.len) : (arg_index += 1) { const arg = serve_args.items[arg_index]; if (std.mem.startsWith(u8, arg, "--upstream=")) { cfg.upstream = arg["--upstream=".len..]; } else if (std.mem.startsWith(u8, arg, "--relay-url=")) { const relay_url = arg["--relay-url=".len..]; if (cfg.canonical_relay) |owned| { owned.deinit(allocator); cfg.canonical_relay = null; } cfg.canonical_relay = try normalizeRelayUrls(allocator, relay_url); cfg.upstream = cfg.canonical_relay.?.websocket; cfg.relay_http = cfg.canonical_relay.?.http; } else if (std.mem.startsWith(u8, arg, "--cursor=")) { cfg.cursor = try std.fmt.parseInt(i64, arg["--cursor=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--port=")) { cfg.port = try std.fmt.parseInt(u16, arg["--port=".len..], 10); cfg.legacy_port_set = true; } else if (std.mem.startsWith(u8, arg, "--addr=")) { cfg.public_addr_raw = arg["--addr=".len..]; } else if (std.mem.startsWith(u8, arg, "--debug-addr=")) { const configured = arg["--debug-addr=".len..]; cfg.debug_addr_raw = if (configured.len == 0) null else configured; if (cfg.public_addr_raw == null) cfg.public_addr_raw = ":8080"; } else if (std.mem.startsWith(u8, arg, "--shutdown-timeout=")) { cfg.shutdown_timeout = try parseGoDuration(arg["--shutdown-timeout=".len..]); } else if (std.mem.startsWith(u8, arg, "--client-drain-timeout=")) { cfg.client_drain_timeout = try parseGoDuration(arg["--client-drain-timeout=".len..]); } else if (std.mem.startsWith(u8, arg, "--otel-service-name=")) { cfg.otel_service_name = arg["--otel-service-name=".len..]; } else if (std.mem.eql(u8, arg, "--otel-service-name")) { arg_index += 1; if (arg_index >= serve_args.items.len) return error.BadArgs; cfg.otel_service_name = serve_args.items[arg_index]; } else if (std.mem.startsWith(u8, arg, "--log-level=")) { log_level = arg["--log-level=".len..]; } else if (std.mem.eql(u8, arg, "--log-level")) { arg_index += 1; if (arg_index >= serve_args.items.len) return error.BadArgs; log_level = serve_args.items[arg_index]; } else if (std.mem.startsWith(u8, arg, "--log-format=")) { log_format = arg["--log-format=".len..]; } else if (std.mem.eql(u8, arg, "--log-format")) { arg_index += 1; if (arg_index >= serve_args.items.len) return error.BadArgs; log_format = serve_args.items[arg_index]; } else if (std.mem.startsWith(u8, arg, "--plc=")) { cfg.plc_url = arg["--plc=".len..]; } else if (std.mem.startsWith(u8, arg, "--plc-url=")) { cfg.plc_url = arg["--plc-url=".len..]; } else if (std.mem.startsWith(u8, arg, "--data-dir=")) { cfg.data_dir = arg["--data-dir=".len..]; } else if (std.mem.startsWith(u8, arg, "--max-segment-bytes=")) { cfg.max_segment_bytes = try std.fmt.parseInt(usize, arg["--max-segment-bytes=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--backfill-max-repos=")) { cfg.backfill_max_repos = try std.fmt.parseInt(usize, arg["--backfill-max-repos=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--max-backfill-repos=")) { const configured = try std.fmt.parseInt(i64, arg["--max-backfill-repos=".len..], 10); cfg.backfill_max_repos = if (configured <= 0) 0 else std.math.cast(usize, configured) orelse return error.Overflow; } else if (std.mem.startsWith(u8, arg, "--backfill-repos=")) { cfg.backfill_repos_raw = arg["--backfill-repos=".len..]; } else if (std.mem.startsWith(u8, arg, "--backfill-workers=")) { const configured = try std.fmt.parseInt(usize, arg["--backfill-workers=".len..], 10); cfg.backfill_workers = if (configured == 0) backfill_engine.default_workers else configured; } else if (std.mem.startsWith(u8, arg, "--backfill-batch-size=")) { const configured = try std.fmt.parseInt(usize, arg["--backfill-batch-size=".len..], 10); cfg.backfill_batch_size = if (configured == 0) backfill_engine.default_batch_size else configured; } else if (std.mem.startsWith(u8, arg, "--backfill-async-flush-workers=")) { cfg.backfill_async_flush_workers = try std.fmt.parseInt(usize, arg["--backfill-async-flush-workers=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--relay-http=")) { cfg.relay_http = arg["--relay-http=".len..]; } else if (std.mem.startsWith(u8, arg, "--compaction-interval=")) { cfg.compaction_interval_ns = try parseDurationNanoseconds(arg["--compaction-interval=".len..]); } else if (std.mem.startsWith(u8, arg, "--compaction-tombstone-cap=")) { cfg.compaction_tombstone_cap = try std.fmt.parseInt(usize, arg["--compaction-tombstone-cap=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--compaction-rewrite-workers=")) { cfg.compaction_rewrite_workers = try std.fmt.parseInt(usize, arg["--compaction-rewrite-workers=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--retry-interval=")) { const seconds = try std.fmt.parseInt(u64, arg["--retry-interval=".len..], 10); cfg.retry_interval_ns = try std.math.mul(u64, seconds, std.time.ns_per_s); } else if (std.mem.startsWith(u8, arg, "--failed-repo-retry-interval=")) { cfg.retry_interval_ns = try parseDurationNanoseconds(arg["--failed-repo-retry-interval=".len..]); } else if (std.mem.startsWith(u8, arg, "--failed-repo-retry-workers=")) { const configured = try std.fmt.parseInt(usize, arg["--failed-repo-retry-workers=".len..], 10); cfg.retry_workers = if (configured == 0) retry_mod.default_workers else configured; } else if (std.mem.startsWith(u8, arg, "--failed-repo-retry-host-workers=")) { const configured = try std.fmt.parseInt(usize, arg["--failed-repo-retry-host-workers=".len..], 10); cfg.retry_host_workers = if (configured == 0) retry_mod.default_host_workers else configured; } else if (std.mem.startsWith(u8, arg, "--failed-repo-retry-max-delay=")) { const configured = try parseDurationNanoseconds(arg["--failed-repo-retry-max-delay=".len..]); cfg.retry_max_delay_ns = if (configured == 0) retry_mod.default_max_delay_ns else configured; } else if (std.mem.startsWith(u8, arg, "--segment-cache-max-age=")) { cfg.segment_cache_max_age_s = try parseDurationCeilSeconds(arg["--segment-cache-max-age=".len..]); } else if (std.mem.startsWith(u8, arg, "--cursor-lookback=")) { cfg.cursor_lookback_ns = try parseDurationNanoseconds(arg["--cursor-lookback=".len..]); } else if (std.mem.startsWith(u8, arg, "--subscribe-read-log-retention-bytes=")) { const configured = try std.fmt.parseInt(i64, arg["--subscribe-read-log-retention-bytes=".len..], 10); cfg.subscribe_read_log_retention_bytes = if (configured <= 0) default_subscribe_read_log_retention_bytes else std.math.cast(usize, configured) orelse return error.Overflow; } else if (std.mem.startsWith(u8, arg, "--subscribe-block-cache-bytes=")) { const configured = try std.fmt.parseInt(i64, arg["--subscribe-block-cache-bytes=".len..], 10); cfg.subscribe_block_cache_bytes = if (configured <= 0) cold.default_block_cache_bytes else std.math.cast(usize, configured) orelse return error.Overflow; } else if (std.mem.startsWith(u8, arg, "--subscribe-slow-window=")) { cfg.subscribe_slow_window_ns = try effectiveSubscribeSlowWindowNanoseconds(arg["--subscribe-slow-window=".len..]); } else if (std.mem.startsWith(u8, arg, "--subscribe-slow-min-rate=")) { const configured = try std.fmt.parseFloat(f64, arg["--subscribe-slow-min-rate=".len..]); cfg.subscribe_slow_min_rate = if (configured <= 0) server_mod.default_slow_min_rate else configured; } else if (std.mem.startsWith(u8, arg, "--subscribe-read-batch=")) { const configured = try std.fmt.parseInt(i64, arg["--subscribe-read-batch=".len..], 10); cfg.subscribe_read_batch = if (configured <= 0) server_mod.default_subscribe_read_batch else std.math.cast(usize, configured) orelse return error.Overflow; } else if (std.mem.startsWith(u8, arg, "--cursor-block-index-cache-size=")) { // Upstream compatibility no-op. Sealed block indexes are always // resident in the refcounted manifest, so there is no cache to // size; accept the full signed CLI surface exactly as cfg.upstream. _ = try std.fmt.parseInt(i64, arg["--cursor-block-index-cache-size=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--plan-max-dids=")) { cfg.plan_config.max_dids = try std.fmt.parseInt(usize, arg["--plan-max-dids=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--plan-max-collections=")) { cfg.plan_config.max_collections = try std.fmt.parseInt(usize, arg["--plan-max-collections=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--plan-max-entries=")) { cfg.plan_config.max_entries = try std.fmt.parseInt(usize, arg["--plan-max-entries=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--plan-whole-segment-threshold=")) { cfg.plan_config.whole_segment_threshold = try std.fmt.parseFloat(f64, arg["--plan-whole-segment-threshold=".len..]); } else if (std.mem.startsWith(u8, arg, "--archive-api-key=")) { cfg.archive_api_key = arg["--archive-api-key=".len..]; } else if (std.mem.startsWith(u8, arg, "--max-subscribers=")) { cfg.max_subscribers = try parseAdmissionCap(arg["--max-subscribers=".len..]); } else if (std.mem.startsWith(u8, arg, "--max-cold-readers=")) { cfg.max_cold_readers = try parseAdmissionCap(arg["--max-cold-readers=".len..]); } else if (std.mem.startsWith(u8, arg, "--upstream-slow-min-rate=")) { cfg.upstream_slow_min_rate = try parseUpstreamSlowMinRate(arg["--upstream-slow-min-rate=".len..]); } else if (std.mem.startsWith(u8, arg, "--archive-api-keys-file=")) { cfg.archive_api_keys_file = arg["--archive-api-keys-file=".len..]; } else if (std.mem.startsWith(u8, arg, "--timestamp-import-token=")) { cfg.timestamp_import_token = arg["--timestamp-import-token=".len..]; } else if (std.mem.startsWith(u8, arg, "--timestamp-import-dir=")) { cfg.timestamp_import_dir_arg = arg["--timestamp-import-dir=".len..]; } else if (std.mem.startsWith(u8, arg, "--crashpoint=")) { // debug/oracle only: NAME or NAME:ORDINAL const spec = arg["--crashpoint=".len..]; const split = std.mem.lastIndexOfScalar(u8, spec, ':'); const name = if (split) |i| spec[0..i] else spec; const ordinal = if (split) |i| try std.fmt.parseInt(u64, spec[i + 1 ..], 10) else 1; if (!crashpoint.armAt(name, ordinal)) { log.err("invalid crashpoint: {s}", .{spec}); return error.BadArgs; } } else if (std.mem.startsWith(u8, arg, "--power-cutpoint=")) { // Linux strict-storage oracle only: NAME or NAME:ORDINAL. const spec = arg["--power-cutpoint=".len..]; const split = std.mem.lastIndexOfScalar(u8, spec, ':'); const name = if (split) |i| spec[0..i] else spec; const ordinal = if (split) |i| try std.fmt.parseInt(u64, spec[i + 1 ..], 10) else 1; if (!crashpoint.armPowerCut(name, ordinal)) { log.err("invalid power cutpoint: {s}", .{spec}); return error.BadArgs; } } else if (std.mem.startsWith(u8, arg, "--store-fault-prefix=")) { // debug/oracle only: fail one matching durable RocksDB mutation cfg.store_fault_prefix = arg["--store-fault-prefix=".len..]; } else if (std.mem.startsWith(u8, arg, "--store-fault-ordinal=")) { cfg.store_fault_ordinal = try std.fmt.parseInt(u64, arg["--store-fault-ordinal=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--segment-fault=")) { // debug/oracle only: fail one real segment operation cfg.segment_fault_spec = arg["--segment-fault=".len..]; } else if (std.mem.eql(u8, arg, "--backfill")) { cfg.do_backfill = true; } else if (std.mem.eql(u8, arg, "--rebloom-sweep")) { cfg.rebloom_sweep = true; } else if (std.mem.startsWith(u8, arg, "--rebloom-sweep=")) { cfg.rebloom_sweep = try parseBool(arg["--rebloom-sweep=".len..]); } else if (std.mem.eql(u8, arg, "--skip-merge-discovery")) { cfg.skip_merge_discovery = true; } else if (std.mem.startsWith(u8, arg, "--skip-merge-discovery=")) { cfg.skip_merge_discovery = try parseBool(arg["--skip-merge-discovery=".len..]); } else if (std.mem.eql(u8, arg, "--no-verify")) { cfg.do_verify = false; } else if (std.mem.eql(u8, arg, "--disable-repo-action-rate-limits")) { cfg.repo_action_rate_limits = false; } else if (std.mem.startsWith(u8, arg, "--disable-repo-action-rate-limits=")) { cfg.repo_action_rate_limits = !(try parseBool(arg["--disable-repo-action-rate-limits=".len..])); } else if (std.mem.eql(u8, arg, "--stdout")) { cfg.to_stdout = true; } else if (std.mem.eql(u8, arg, "serve") and !explicit_serve) { // Root flags are persistent cfg.upstream, so `--log-level=debug // serve` is equivalent to `serve --log-level=debug`. explicit_serve = true; if (cfg.public_addr_raw == null) cfg.public_addr_raw = ":8080"; } else { log.err("unknown arg: {s}", .{arg}); return error.BadArgs; } } return .{ .config = cfg, .explicit_serve = explicit_serve, .log_level = log_level, .log_format = log_format, .env_args = env_serve_args };}
// === client identity ===
const build_options = @import("build_options");
/// What Stream tells the servers it calls.////// A whole-network crawl touches thousands of independently-operated PDSes. An/// operator seeing unexplained load should be able to tell what is calling and/// where to ask about it; std.http's default (`zig/<version> (std.http)`) says/// neither, and Stream showed up in a real operator's logs as exactly that./// Every other client on the network identifies itself.////// Derived from build_options so it cannot go stale the way a literal would.pub fn userAgent() []const u8 { return "stream/" ++ build_options.version ++ " (+https://tangled.org/zat.dev/stream)";}
test "the user agent names the software and a way to reach its operator" { const ua = userAgent(); try testing.expect(std.mem.startsWith(u8, ua, "stream/")); try testing.expect(std.mem.indexOf(u8, ua, "tangled.org/zat.dev/stream") != null); // the std default is what we are replacing; never ship it try testing.expect(std.mem.indexOf(u8, ua, "std.http") == null);}