From 5e81128fa66d47464ecba55c75ecefda2ba56724 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Tue, 28 Jul 2026 16:18:29 -0500 Subject: [PATCH] main: lift serve configuration into internal/runtime/cli The 65-branch flag chain, its 44 option locals and the environment merge now live in cli.parseServe, returning a typed ServeConfig. main.zig 1263 -> 1018 lines; `pub fn main` 856 -> 611, and what is left is construction and lifecycle rather than argument dispatch. Two bugs the extraction introduced, both caught by the contract suites rather than by unit tests: - A blanket identifier rewrite renamed variables *inside string literals*, so "--cursor-lookback=" became "--cfg.cursor-lookback=" and five flags silently stopped parsing. The environment contract caught it; nothing in the unit tests would have. - Moving the parse out of main made env_serve_args die at function exit while the config still borrowed into it -- every JETSTREAM_*-supplied value dangled, surfacing as BadPathName. Ownership now lives in Parsed, which the caller frees. This is the same borrowed-data-outliving-its-owner shape that has bitten this codebase before. environment, listener, logging and status contracts all pass. Co-Authored-By: Claude Opus 5 (1M context) --- src/internal/runtime/cli.zig | 350 +++++++++++++++++++++++++- src/main.zig | 457 ++++++++-------------------------- tests/environment_contract.py | 2 +- 3 files changed, 455 insertions(+), 354 deletions(-) diff --git a/src/internal/runtime/cli.zig b/src/internal/runtime/cli.zig index 1d882ee..ba10161 100644 --- a/src/internal/runtime/cli.zig +++ b/src/internal/runtime/cli.zig @@ -139,7 +139,6 @@ test "the derived option set reproduces the hand-written environment map" { // 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 environment = @import("environment.zig"); const derived = comptime exposures(ServeOptions); try testing.expectEqual(environment.mappings.len, derived.len); @@ -370,4 +369,351 @@ test "boolean flags match Go strconv syntax" { try std.testing.expect(!(try parseBool("F"))); try std.testing.expect(!(try parseBool("0"))); try std.testing.expectError(error.InvalidBoolean, parseBool("yes")); -} \ No newline at end of file +} +// === 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, + 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 = "", + 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, "--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, "--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 }; +} diff --git a/src/main.zig b/src/main.zig index ad7e75b..2007eaa 100644 --- a/src/main.zig +++ b/src/main.zig @@ -449,298 +449,53 @@ pub fn main(init: std.process.Init.Minimal) !void { shutdown_requested.store(false, .release); installSignalHandlers(); - var upstream: []const u8 = "ws://localhost:7777"; - var canonical_relay: ?RelayUrls = null; - defer if (canonical_relay) |owned| owned.deinit(allocator); - // the simulator serves DID docs itself; production passes --plc-url. - var plc_url: []const u8 = "http://localhost:7777"; - var data_dir: []const u8 = "./data"; - var cursor: ?i64 = null; - var port: u16 = 6008; - var legacy_port_set = false; - var public_addr_raw: ?[]const u8 = if (explicit_serve) ":8080" else null; - var debug_addr_raw: ?[]const u8 = null; - var shutdown_timeout: Io.Duration = .fromSeconds(5); - var client_drain_timeout: Io.Duration = .fromSeconds(10); - var otel_service_name: []const u8 = environment.value(env_entries, "OTEL_SERVICE_NAME") orelse "jetstream"; - if (otel_service_name.len == 0) otel_service_name = "jetstream"; - var to_stdout = false; - var do_verify = true; - var max_segment_bytes: usize = archive_mod.default_max_segment_bytes; - var backfill_max_repos: usize = 0; - var backfill_repos_raw: ?[]const u8 = null; - var do_backfill = false; - var backfill_workers: usize = backfill_engine.default_workers; - var backfill_batch_size: usize = backfill_engine.default_batch_size; - var backfill_async_flush_workers: usize = lifecycle.default_async_flush_workers; - var skip_merge_discovery = false; - var relay_http: []const u8 = "http://localhost:7777"; - var compaction_interval_ns: u64 = 4 * std.time.ns_per_hour; - var compaction_tombstone_cap: usize = compact_pass.default_tombstone_cap; - var compaction_rewrite_workers: usize = 0; - var retry_interval_ns: u64 = retry_mod.default_interval_ns; - var retry_workers: usize = retry_mod.default_workers; - var retry_host_workers: usize = retry_mod.default_host_workers; - var retry_max_delay_ns: u64 = retry_mod.default_max_delay_ns; - var timestamp_import_token: []const u8 = ""; - var timestamp_import_dir_arg: ?[]const u8 = null; - var repo_action_rate_limits = true; - var store_fault_prefix: ?[]const u8 = null; - var store_fault_ordinal: u64 = 1; - var segment_fault_spec: ?[]const u8 = null; - var segment_cache_max_age_s: u64 = 0; - var cursor_lookback_ns: u64 = 36 * std.time.ns_per_hour; - var subscribe_read_log_retention_bytes: usize = default_subscribe_read_log_retention_bytes; - var subscribe_block_cache_bytes: usize = cold.default_block_cache_bytes; - var subscribe_slow_window_ns: u64 = server_mod.default_slow_window_ns; - var subscribe_slow_min_rate: f64 = server_mod.default_slow_min_rate; - var subscribe_read_batch: usize = server_mod.default_subscribe_read_batch; - var plan_config: xrpcapi.PlanConfig = .{}; - - var env_serve_args: std.ArrayList([]u8) = .empty; - defer { - 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=")) { - upstream = arg["--upstream=".len..]; - } else if (std.mem.startsWith(u8, arg, "--relay-url=")) { - const relay_url = arg["--relay-url=".len..]; - if (canonical_relay) |owned| { - owned.deinit(allocator); - canonical_relay = null; - } - canonical_relay = try normalizeRelayUrls(allocator, relay_url); - upstream = canonical_relay.?.websocket; - relay_http = canonical_relay.?.http; - } else if (std.mem.startsWith(u8, arg, "--cursor=")) { - cursor = try std.fmt.parseInt(i64, arg["--cursor=".len..], 10); - } else if (std.mem.startsWith(u8, arg, "--port=")) { - port = try std.fmt.parseInt(u16, arg["--port=".len..], 10); - legacy_port_set = true; - } else if (std.mem.startsWith(u8, arg, "--addr=")) { - public_addr_raw = arg["--addr=".len..]; - } else if (std.mem.startsWith(u8, arg, "--debug-addr=")) { - const configured = arg["--debug-addr=".len..]; - debug_addr_raw = if (configured.len == 0) null else configured; - if (public_addr_raw == null) public_addr_raw = ":8080"; - } else if (std.mem.startsWith(u8, arg, "--shutdown-timeout=")) { - shutdown_timeout = try cli.parseGoDuration(arg["--shutdown-timeout=".len..]); - } else if (std.mem.startsWith(u8, arg, "--client-drain-timeout=")) { - client_drain_timeout = try cli.parseGoDuration(arg["--client-drain-timeout=".len..]); - } else if (std.mem.startsWith(u8, arg, "--otel-service-name=")) { - 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; - 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=")) { - plc_url = arg["--plc=".len..]; - } else if (std.mem.startsWith(u8, arg, "--plc-url=")) { - plc_url = arg["--plc-url=".len..]; - } else if (std.mem.startsWith(u8, arg, "--data-dir=")) { - data_dir = arg["--data-dir=".len..]; - } else if (std.mem.startsWith(u8, arg, "--max-segment-bytes=")) { - max_segment_bytes = try std.fmt.parseInt(usize, arg["--max-segment-bytes=".len..], 10); - } else if (std.mem.startsWith(u8, arg, "--backfill-max-repos=")) { - 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); - 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=")) { - 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); - 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); - backfill_batch_size = if (configured == 0) backfill_engine.default_batch_size else configured; - } else if (std.mem.startsWith(u8, arg, "--backfill-async-flush-workers=")) { - 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=")) { - relay_http = arg["--relay-http=".len..]; - } else if (std.mem.startsWith(u8, arg, "--compaction-interval=")) { - compaction_interval_ns = try cli.parseDurationNanoseconds(arg["--compaction-interval=".len..]); - } else if (std.mem.startsWith(u8, arg, "--compaction-tombstone-cap=")) { - compaction_tombstone_cap = try std.fmt.parseInt(usize, arg["--compaction-tombstone-cap=".len..], 10); - } else if (std.mem.startsWith(u8, arg, "--compaction-rewrite-workers=")) { - 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); - 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=")) { - retry_interval_ns = try cli.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); - 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); - 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 cli.parseDurationNanoseconds(arg["--failed-repo-retry-max-delay=".len..]); - 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=")) { - segment_cache_max_age_s = try cli.parseDurationCeilSeconds(arg["--segment-cache-max-age=".len..]); - } else if (std.mem.startsWith(u8, arg, "--cursor-lookback=")) { - cursor_lookback_ns = try cli.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); - 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); - 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=")) { - subscribe_slow_window_ns = try cli.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..]); - 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); - 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 upstream. - _ = try std.fmt.parseInt(i64, arg["--cursor-block-index-cache-size=".len..], 10); - } else if (std.mem.startsWith(u8, arg, "--plan-max-dids=")) { - 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=")) { - 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=")) { - 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=")) { - plan_config.whole_segment_threshold = try std.fmt.parseFloat(f64, arg["--plan-whole-segment-threshold=".len..]); - } else if (std.mem.startsWith(u8, arg, "--timestamp-import-token=")) { - timestamp_import_token = arg["--timestamp-import-token=".len..]; - } else if (std.mem.startsWith(u8, arg, "--timestamp-import-dir=")) { - 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 - store_fault_prefix = arg["--store-fault-prefix=".len..]; - } else if (std.mem.startsWith(u8, arg, "--store-fault-ordinal=")) { - 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 - segment_fault_spec = arg["--segment-fault=".len..]; - } else if (std.mem.eql(u8, arg, "--backfill")) { - do_backfill = true; - } else if (std.mem.eql(u8, arg, "--skip-merge-discovery")) { - skip_merge_discovery = true; - } else if (std.mem.startsWith(u8, arg, "--skip-merge-discovery=")) { - skip_merge_discovery = try cli.parseBool(arg["--skip-merge-discovery=".len..]); - } else if (std.mem.eql(u8, arg, "--no-verify")) { - do_verify = false; - } else if (std.mem.eql(u8, arg, "--disable-repo-action-rate-limits")) { - repo_action_rate_limits = false; - } else if (std.mem.startsWith(u8, arg, "--disable-repo-action-rate-limits=")) { - repo_action_rate_limits = !(try cli.parseBool(arg["--disable-repo-action-rate-limits=".len..])); - } else if (std.mem.eql(u8, arg, "--stdout")) { - to_stdout = true; - } else if (std.mem.eql(u8, arg, "serve") and !explicit_serve) { - // Root flags are persistent upstream, so `--log-level=debug - // serve` is equivalent to `serve --log-level=debug`. - explicit_serve = true; - if (public_addr_raw == null) public_addr_raw = ":8080"; - } else { - log.err("unknown arg: {s}", .{arg}); - return error.BadArgs; - } - } + var parsed = try cli.parseServe(allocator, env_entries, first_arg, &arg_it, explicit_serve, log_level, log_format); + defer parsed.deinit(allocator); + const cfg = &parsed.config; + explicit_serve = parsed.explicit_serve; + log_level = parsed.log_level; + log_format = parsed.log_format; try logging.configure(log_level, log_format); log.debug("logging configured: level={s} format={s}", .{ log_level, log_format }); - if (legacy_port_set and public_addr_raw != null) { + if (cfg.legacy_port_set and cfg.public_addr_raw != null) { log.err("--port cannot be combined with upstream --addr/--debug-addr or explicit serve mode", .{}); return error.BadArgs; } - if (!(plan_config.whole_segment_threshold > 0 and plan_config.whole_segment_threshold <= 1)) { + if (!(cfg.plan_config.whole_segment_threshold > 0 and cfg.plan_config.whole_segment_threshold <= 1)) { log.err("--plan-whole-segment-threshold must be > 0 and <= 1", .{}); return error.BadArgs; } - if (store_fault_prefix) |prefix| { - if (!meta_store.armFault(prefix, store_fault_ordinal)) { + if (cfg.store_fault_prefix) |prefix| { + if (!meta_store.armFault(prefix, cfg.store_fault_ordinal)) { log.err("invalid store fault: prefix must be non-empty and ordinal must be positive", .{}); return error.BadArgs; } } - if (segment_fault_spec) |spec| { + if (cfg.segment_fault_spec) |spec| { if (!segment_io.armSpec(spec)) { log.err("invalid segment fault; expected op:positive-ordinal:kind", .{}); return error.BadArgs; } } - const backfill_repos: []const []const u8 = if (backfill_repos_raw) |raw| try parseBackfillRepos(allocator, raw) else &.{}; + const backfill_repos: []const []const u8 = if (cfg.backfill_repos_raw) |raw| try parseBackfillRepos(allocator, raw) else &.{}; defer if (backfill_repos.len > 0) allocator.free(backfill_repos); - if (backfill_repos.len > 0 and backfill_max_repos > 0) return error.ConflictingBackfillSelection; - // Pinned upstream automatically enables the debug short circuit for + if (backfill_repos.len > 0 and cfg.backfill_max_repos > 0) return error.ConflictingBackfillSelection; + // Pinned cfg.upstream automatically enables the debug short circuit for // either partial selection mode, in addition to accepting it explicitly. - skip_merge_discovery = effectiveSkipMergeDiscovery(skip_merge_discovery, backfill_max_repos, backfill_repos.len); + cfg.skip_merge_discovery = effectiveSkipMergeDiscovery(cfg.skip_merge_discovery, cfg.backfill_max_repos, backfill_repos.len); var tracing = observability.Runtime.init(allocator, .{ .io = io, .environment = env_entries, - .service_name = otel_service_name, + .service_name = cfg.otel_service_name, .service_version = build_options.version, - .shutdown_timeout_ms = shutdownTimeoutMillis(shutdown_timeout), + .shutdown_timeout_ms = shutdownTimeoutMillis(cfg.shutdown_timeout), }) catch |err| { log.err("setup tracing failed: {s}", .{@errorName(err)}); return @errorCast(err); }; defer tracing.deinit(); - if (tracing.enabled()) log.info("OTLP tracing enabled (service.name={s})", .{otel_service_name}); + if (tracing.enabled()) log.info("OTLP tracing enabled (service.name={s})", .{cfg.otel_service_name}); var orchestrator_trace = observability.Span.start( allocator, @@ -753,7 +508,7 @@ pub fn main(init: std.process.Init.Minimal) !void { defer orchestrator_trace.succeed(); errdefer |err| orchestrator_trace.fail(err); - var hot_tail = tail_mod.Tail.init(allocator, io, subscribe_read_log_retention_bytes); + var hot_tail = tail_mod.Tail.init(allocator, io, cfg.subscribe_read_log_retention_bytes); defer hot_tail.deinit(); var stats: metrics.Stats = .{ .start_time_s = @divTrunc(Io.Timestamp.now(io, .real).toMicroseconds(), std.time.us_per_s), @@ -763,43 +518,43 @@ pub fn main(init: std.process.Init.Minimal) !void { .io = io, .tail = &hot_tail, .stats = &stats, - .bootstrap_enabled = do_backfill or backfill_max_repos > 0 or backfill_repos.len > 0, - .compaction_enabled = compaction_interval_ns > 0, - .retry_enabled = retry_interval_ns > 0, - .plc_url = plc_url, - .data_dir = data_dir, - .repo_action_rate_limits = repo_action_rate_limits, - .cursor_lookback_ns = cursor_lookback_ns, - .subscribe_slow_window_ns = subscribe_slow_window_ns, - .subscribe_slow_min_rate = subscribe_slow_min_rate, - .subscribe_read_batch = subscribe_read_batch, + .bootstrap_enabled = cfg.do_backfill or cfg.backfill_max_repos > 0 or backfill_repos.len > 0, + .compaction_enabled = cfg.compaction_interval_ns > 0, + .retry_enabled = cfg.retry_interval_ns > 0, + .plc_url = cfg.plc_url, + .data_dir = cfg.data_dir, + .repo_action_rate_limits = cfg.repo_action_rate_limits, + .cursor_lookback_ns = cfg.cursor_lookback_ns, + .subscribe_slow_window_ns = cfg.subscribe_slow_window_ns, + .subscribe_slow_min_rate = cfg.subscribe_slow_min_rate, + .subscribe_read_batch = cfg.subscribe_read_batch, }; defer hub.deinit(); log.info("subscribe controls: read-log-retention={d} bytes block-cache={d} bytes read-batch={d} slow-window={d}ns slow-min-rate={d}", .{ - subscribe_read_log_retention_bytes, - subscribe_block_cache_bytes, - subscribe_read_batch, - subscribe_slow_window_ns, - subscribe_slow_min_rate, + cfg.subscribe_read_log_retention_bytes, + cfg.subscribe_block_cache_bytes, + cfg.subscribe_read_batch, + cfg.subscribe_slow_window_ns, + cfg.subscribe_slow_min_rate, }); - var archive = archive_mod.Archive.initWithStats(allocator, io, data_dir, &stats) catch |err| { - logPersistenceFailure(data_dir, err); + var archive = archive_mod.Archive.initWithStats(allocator, io, cfg.data_dir, &stats) catch |err| { + logPersistenceFailure(cfg.data_dir, err); return @errorCast(err); }; var archive_closed = false; defer { archive.deinit(); } - archive.max_segment_bytes = max_segment_bytes; - var rule_store = try timestamp_rules.Store.open(allocator, io, data_dir); + archive.max_segment_bytes = cfg.max_segment_bytes; + var rule_store = try timestamp_rules.Store.open(allocator, io, cfg.data_dir); defer rule_store.deinit(); archive.stamper_ctx = &rule_store; archive.stamper = timestamp_rules.Store.stampOpaque; hot_tail.enableDurabilityGate(archive.committed_seq.load(.acquire)); - log.info("archive at {s}/segments (next seq {d})", .{ data_dir, archive.next_seq }); + log.info("archive at {s}/segments (next seq {d})", .{ cfg.data_dir, archive.next_seq }); hub.archive = &archive; - var cold_reader = cold.ColdReader.init(allocator, io, subscribe_block_cache_bytes); + var cold_reader = cold.ColdReader.init(allocator, io, cfg.subscribe_block_cache_bytes); defer cold_reader.deinit(); archive.manifest.setInvalidator(&cold_reader, cold.ColdReader.invalidateOpaque); defer archive.manifest.setInvalidator(null, null); @@ -808,14 +563,14 @@ pub fn main(init: std.process.Init.Minimal) !void { .allocator = allocator, .io = io, .archive = &archive, - .segment_cache_max_age_s = segment_cache_max_age_s, - .plan = plan_config, + .segment_cache_max_age_s = cfg.segment_cache_max_age_s, + .plan = cfg.plan_config, }; hub.xrpc = &xrpc_api; - var meta = try meta_store.Store.openWithStats(allocator, io, data_dir, &stats); + var meta = try meta_store.Store.openWithStats(allocator, io, cfg.data_dir, &stats); defer meta.deinit(); - try meta.migrateLegacy(io, data_dir); + try meta.migrateLegacy(io, cfg.data_dir); var repo_diagnostics = try repo_store.Store.init(allocator, io, &meta); defer repo_diagnostics.deinit(); hub.repo_status = &repo_diagnostics; @@ -834,15 +589,15 @@ pub fn main(init: std.process.Init.Minimal) !void { .steady_state => .steady_state, }); const mid_lifecycle = if (stored_phase) |ph| ph != .steady_state else false; - const bootstrap_requested = do_backfill or backfill_max_repos > 0 or backfill_repos.len > 0; + const bootstrap_requested = cfg.do_backfill or cfg.backfill_max_repos > 0 or backfill_repos.len > 0; if (mid_lifecycle or bootstrap_requested) hub.serving.store(false, .release); var import_dir_buffer: [Io.Dir.max_path_bytes]u8 = undefined; - const import_dir = timestamp_import_dir_arg orelse - try std.fmt.bufPrint(&import_dir_buffer, "{s}/imports", .{data_dir}); + const import_dir = cfg.timestamp_import_dir_arg orelse + try std.fmt.bufPrint(&import_dir_buffer, "{s}/imports", .{cfg.data_dir}); try Io.Dir.cwd().createDirPath(io, import_dir); var import_scratch_buffer: [Io.Dir.max_path_bytes]u8 = undefined; - const import_scratch = try std.fmt.bufPrint(&import_scratch_buffer, "{s}/timestamp-import-jobs", .{data_dir}); + const import_scratch = try std.fmt.bufPrint(&import_scratch_buffer, "{s}/timestamp-import-jobs", .{cfg.data_dir}); var import_job_store = timestamp_jobs.Store.init(allocator, &meta); var import_ready: std.atomic.Value(bool) = .init(false); var import_manager = try timestamp_manager.Manager.init(.{ @@ -865,36 +620,36 @@ pub fn main(init: std.process.Init.Minimal) !void { var import_manager_err_owned = true; errdefer if (import_manager_err_owned) import_manager.deinit(); xrpc_api.import = &import_manager; - xrpc_api.import_token = timestamp_import_token; + xrpc_api.import_token = cfg.timestamp_import_token; // Killed getRepo downloads can leave incomplete scratch files. Bootstrap // owns backfill/, while steady repair has a separate root so a legitimate // live repair can never resurrect lifecycle state after cutover. const fetched_car = @import("internal/ingest/backfill/fetched_car.zig"); var bootstrap_scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; - const bootstrap_scratch = try std.fmt.bufPrint(&bootstrap_scratch_buf, "{s}/backfill/repo-scratch", .{data_dir}); + const bootstrap_scratch = try std.fmt.bufPrint(&bootstrap_scratch_buf, "{s}/backfill/repo-scratch", .{cfg.data_dir}); try fetched_car.resetScratch(io, bootstrap_scratch); var repair_scratch_root_buf: [Io.Dir.max_path_bytes]u8 = undefined; - const repair_scratch_root = try std.fmt.bufPrint(&repair_scratch_root_buf, "{s}/repair-scratch", .{data_dir}); + const repair_scratch_root = try std.fmt.bufPrint(&repair_scratch_root_buf, "{s}/repair-scratch", .{cfg.data_dir}); try fetched_car.resetScratch(io, repair_scratch_root); var live_repair_scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; const live_repair_scratch = try std.fmt.bufPrint(&live_repair_scratch_buf, "{s}/live", .{repair_scratch_root}); - const public_bind = if (public_addr_raw) |raw| + const public_bind = if (cfg.public_addr_raw) |raw| resolveBindAddress(io, raw) catch |err| { log.err("invalid public bind address {s}: {s}", .{ raw, @errorName(err) }); return @errorCast(err); } else - BindAddress{ .address = .{ .ip4 = .unspecified(port) }, .port = port }; - const debug_bind: ?BindAddress = if (debug_addr_raw) |raw| + BindAddress{ .address = .{ .ip4 = .unspecified(cfg.port) }, .port = cfg.port }; + const debug_bind: ?BindAddress = if (cfg.debug_addr_raw) |raw| resolveBindAddress(io, raw) catch |err| { log.err("invalid debug bind address {s}: {s}", .{ raw, @errorName(err) }); return @errorCast(err); } else null; - const canonical_listeners = public_addr_raw != null; + const canonical_listeners = cfg.public_addr_raw != null; var public_context: server_mod.ListenerContext = .{ .hub = &hub, .role = if (canonical_listeners) .public else .legacy, @@ -906,9 +661,9 @@ pub fn main(init: std.process.Init.Minimal) !void { .address = "0.0.0.0", .max_conn = 4096, .max_message_size = 10_000_000, - .websocket_shutdown_grace = client_drain_timeout, - .http_shutdown_grace = cli.effectiveHttpShutdownDuration(shutdown_timeout), - // Match upstream v1's coder/websocket defaults exactly. The handler + .websocket_shutdown_grace = cfg.client_drain_timeout, + .http_shutdown_grace = cli.effectiveHttpShutdownDuration(cfg.shutdown_timeout), + // Match cfg.upstream v1's coder/websocket defaults exactly. The handler // declines this per connection for v2 and custom-zstd subscribers. .compression = .{ .write_threshold = 128, @@ -923,7 +678,7 @@ pub fn main(init: std.process.Init.Minimal) !void { }); defer ws_server.deinit(); var listener = public_bind.address.listen(io, .{ .reuse_address = true }) catch |err| { - log.err("failed to bind public listener {s}: {s}", .{ public_addr_raw orelse "legacy --port", @errorName(err) }); + log.err("failed to bind public listener {s}: {s}", .{ cfg.public_addr_raw orelse "legacy --port", @errorName(err) }); return @errorCast(err); }; defer listener.deinit(io); @@ -935,8 +690,8 @@ pub fn main(init: std.process.Init.Minimal) !void { .address = "0.0.0.0", .max_conn = 256, .max_message_size = 1024, - .websocket_shutdown_grace = client_drain_timeout, - .http_shutdown_grace = cli.effectiveHttpShutdownDuration(shutdown_timeout), + .websocket_shutdown_grace = cfg.client_drain_timeout, + .http_shutdown_grace = cli.effectiveHttpShutdownDuration(cfg.shutdown_timeout), .compression = null, .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 }, }) @@ -945,7 +700,7 @@ pub fn main(init: std.process.Init.Minimal) !void { defer if (debug_ws_server) |*server| server.deinit(); var debug_listener: ?Io.net.Server = if (debug_bind) |bind| bind.address.listen(io, .{ .reuse_address = true }) catch |err| { - log.err("failed to bind debug listener {s}: {s}", .{ debug_addr_raw.?, @errorName(err) }); + log.err("failed to bind debug listener {s}: {s}", .{ cfg.debug_addr_raw.?, @errorName(err) }); return @errorCast(err); } else @@ -961,18 +716,18 @@ pub fn main(init: std.process.Init.Minimal) !void { hub.listeners_ready.store(true, .release); defer hub.listeners_ready.store(false, .release); if (canonical_listeners) { - log.info("public listener {s}; debug listener {s}", .{ public_addr_raw.?, debug_addr_raw orelse "disabled" }); + log.info("public listener {s}; debug listener {s}", .{ cfg.public_addr_raw.?, cfg.debug_addr_raw orelse "disabled" }); } else { - log.info("legacy combined listener on :{d}", .{port}); + log.info("legacy combined listener on :{d}", .{cfg.port}); } var cursor_store = cursor_mod.Store.init(&meta); defer cursor_store.deinit(io); - var verifier: ?verify.Verifier = if (do_verify) verify.Verifier.init(allocator, io, plc_url, &meta) else null; + var verifier: ?verify.Verifier = if (cfg.do_verify) verify.Verifier.init(allocator, io, cfg.plc_url, &meta) else null; if (verifier) |*v| v.stats = &stats; defer if (verifier) |*v| v.deinit(); - if (do_verify) log.info("signature verification on (plc: {s})", .{plc_url}); + if (cfg.do_verify) log.info("signature verification on (plc: {s})", .{cfg.plc_url}); // lifecycle: a stored non-steady phase means a crash mid-bootstrap or // mid-merge — always resume it. otherwise only --backfill (full crawl) @@ -983,17 +738,17 @@ pub fn main(init: std.process.Init.Minimal) !void { .allocator = allocator, .io = io, .options = .{ - .data_dir = data_dir, - .upstream = upstream, - .relay_http = relay_http, - .plc_url = plc_url, - .max_repos = backfill_max_repos, + .data_dir = cfg.data_dir, + .upstream = cfg.upstream, + .relay_http = cfg.relay_http, + .plc_url = cfg.plc_url, + .max_repos = cfg.backfill_max_repos, .selected_repos = backfill_repos, - .workers = backfill_workers, - .batch_size = backfill_batch_size, - .async_flush_workers = backfill_async_flush_workers, - .skip_merge_discovery = skip_merge_discovery, - .compaction_enabled = compaction_interval_ns != 0, + .workers = cfg.backfill_workers, + .batch_size = cfg.backfill_batch_size, + .async_flush_workers = cfg.backfill_async_flush_workers, + .skip_merge_discovery = cfg.skip_merge_discovery, + .compaction_enabled = cfg.compaction_interval_ns != 0, }, .archive = &archive, .meta = &meta, @@ -1023,19 +778,19 @@ pub fn main(init: std.process.Init.Minimal) !void { archive_closed = true; try listener_shutdowns.await(io); orchestrator_trace.succeed(); - try shutdownTracing(&tracing, shutdown_timeout); + try shutdownTracing(&tracing, cfg.shutdown_timeout); return; } lifecycle_future.await(io); if (lifecycle_run.result) |err| { - logPersistenceFailure(data_dir, err); + logPersistenceFailure(cfg.data_dir, err); return @errorCast(err); } hub.serving.store(true, .release); } - if (cursor == null) { - cursor = try cursor_store.load(io); - if (cursor) |c| log.info("resuming from persisted cursor {d}", .{c}); + if (cfg.cursor == null) { + cfg.cursor = try cursor_store.load(io); + if (cfg.cursor) |c| log.info("resuming from persisted cursor {d}", .{c}); } var steady_trace = observability.Span.start( @@ -1053,21 +808,21 @@ pub fn main(init: std.process.Init.Minimal) !void { // steady compactor: live tombstone set fed from the archive's append // hook, periodic + cap-triggered passes. interval 0 disables all of it // (run() returns immediately, no hook, no rebuild). - var compactor = steady_mod.Compactor.init(allocator, io, &archive, &meta, data_dir); + var compactor = steady_mod.Compactor.init(allocator, io, &archive, &meta, cfg.data_dir); defer compactor.deinit(); - compactor.interval_ns = compaction_interval_ns; - compactor.tombstone_cap = compaction_tombstone_cap; - compactor.rewrite_workers = compaction_rewrite_workers; + compactor.interval_ns = cfg.compaction_interval_ns; + compactor.tombstone_cap = cfg.compaction_tombstone_cap; + compactor.rewrite_workers = cfg.compaction_rewrite_workers; compactor.stats = &stats; compactor.trace_context = steady_trace.context(); - if (compaction_interval_ns != 0) { + if (cfg.compaction_interval_ns != 0) { try compactor.rebuild(); archive.on_append_ctx = &compactor; archive.on_append = steady_mod.Compactor.onAppend; log.info("steady compactor on (interval {d}ns, tombstone cap {d}, rewrite workers {d})", .{ - compaction_interval_ns, - compaction_tombstone_cap, - compaction_rewrite_workers, + cfg.compaction_interval_ns, + cfg.compaction_tombstone_cap, + cfg.compaction_rewrite_workers, }); } var compactor_future = try io.concurrent(steady_mod.Compactor.run, .{&compactor}); @@ -1079,20 +834,20 @@ pub fn main(init: std.process.Init.Minimal) !void { // failed-repo retry loop: periodic resync of repos whose backfill // download failed (or that discovery queued). appends ride the // archive writer mutex beside the live consumer. - var retry = retry_mod.Retry.init(allocator, io, &archive, &meta, data_dir, relay_http); + var retry = retry_mod.Retry.init(allocator, io, &archive, &meta, cfg.data_dir, cfg.relay_http); defer retry.deinit(); - retry.interval_ns = retry_interval_ns; - retry.workers = retry_workers; - retry.host_workers = retry_host_workers; - retry.max_delay_ns = retry_max_delay_ns; + retry.interval_ns = cfg.retry_interval_ns; + retry.workers = cfg.retry_workers; + retry.host_workers = cfg.retry_host_workers; + retry.max_delay_ns = cfg.retry_max_delay_ns; retry.stats = &stats; retry.trace_context = steady_trace.context(); retry.shutdown = &shutdown_requested; - if (retry_interval_ns != 0) log.info("failed-repo retry loop on (interval {d}ns, workers {d}, host workers {d}, max delay {d}ns)", .{ - retry_interval_ns, - retry_workers, - retry_host_workers, - retry_max_delay_ns, + if (cfg.retry_interval_ns != 0) log.info("failed-repo retry loop on (interval {d}ns, workers {d}, host workers {d}, max delay {d}ns)", .{ + cfg.retry_interval_ns, + cfg.retry_workers, + cfg.retry_host_workers, + cfg.retry_max_delay_ns, }); var retry_future = try io.concurrent(retry_mod.Retry.run, .{&retry}); defer { @@ -1104,9 +859,9 @@ pub fn main(init: std.process.Init.Minimal) !void { .allocator = allocator, .io = io, .tail = &hot_tail, - .upstream = upstream, - .cursor = cursor, - .to_stdout = to_stdout, + .upstream = cfg.upstream, + .cursor = cfg.cursor, + .to_stdout = cfg.to_stdout, .verifier = if (verifier) |*v| v else null, .cursor_store = &cursor_store, .archive = &archive, @@ -1132,14 +887,14 @@ pub fn main(init: std.process.Init.Minimal) !void { .emit = ingest.Consumer.emitVerified, .emit_repair = ingest.Consumer.emitRepair, .emit_cursor = ingest.Consumer.emitCursor, - .cursor_watermark = cursor orelse 0, + .cursor_watermark = cfg.cursor orelse 0, .trace_context = steady_trace.context(), }; var repair_coordinator: ?repair_mod.Coordinator = if (verifier) |*v| .{ .allocator = allocator, .io = io, .verifier = v, - .relay_http = relay_http, + .relay_http = cfg.relay_http, .scratch_dir = live_repair_scratch, .stats = &stats, .publish_ctx = &pipe, @@ -1201,7 +956,7 @@ pub fn main(init: std.process.Init.Minimal) !void { if (shutdown_err) |err| return @errorCast(err); steady_trace.succeed(); orchestrator_trace.succeed(); - try shutdownTracing(&tracing, shutdown_timeout); + try shutdownTracing(&tracing, cfg.shutdown_timeout); return; } try consumer_future.await(io); @@ -1210,7 +965,7 @@ pub fn main(init: std.process.Init.Minimal) !void { archive_closed = true; steady_trace.succeed(); orchestrator_trace.succeed(); - try shutdownTracing(&tracing, shutdown_timeout); + try shutdownTracing(&tracing, cfg.shutdown_timeout); } fn shutdownTracing(runtime: *observability.Runtime, configured: Io.Duration) !void { diff --git a/tests/environment_contract.py b/tests/environment_contract.py index 604f049..5ed147a 100644 --- a/tests/environment_contract.py +++ b/tests/environment_contract.py @@ -44,7 +44,7 @@ def exact_pinned_map_case() -> None: prior = upstream.setdefault(env.group(1), name.group(1)) assert prior == name.group(1) - stream_text = (ROOT / "src/internal/environment.zig").read_text() + stream_text = (ROOT / "src/internal/runtime/environment.zig").read_text() stream = { env: flag for env, flag in re.findall( -- 2.51.2