diff --git a/receipts/66956ef.json b/receipts/66956ef.json new file mode 100644 index 0000000..e80aba2 --- /dev/null +++ b/receipts/66956ef.json @@ -0,0 +1,47 @@ +{ + "created_utc": "20260808T025739Z", + "image": { + "local_id": "sha256:3a894519be8bd272e6ac178334c9c6ad4da33c2bd92bb7530599f30c77ac8d33", + "registry_digest": null, + "tag": "atcr.io/zat.dev/stream:66956ef" + }, + "revisions": { + "dashboard_sha256": "ef6033c182d8c3f9d88af7d91e61126724efbe72e6b940984c9de12e9530cd91", + "dependencies": { + "otel": "git+https://github.com/zzstoatzz/otel-zig.git#158b32d87762be4f6b65a7d723d08b20d74d1294", + "rocksdb": "https://github.com/zzstoatzz/rocksdb-zig/archive/9d2ebd8.tar.gz", + "websocket": "https://github.com/zzstoatzz/websocket.zig/archive/73429df.tar.gz", + "xxhash": "https://github.com/Cyan4973/xxHash/archive/refs/tags/v0.8.3.tar.gz", + "zat": "https://tangled.org/zat.dev/zat/archive/409711c0ff5465211f3ff3eb8a5332332da09b3b.tar.gz", + "zstd": "https://github.com/facebook/zstd/releases/download/v1.5.7/zstd-1.5.7.tar.gz" + }, + "stream": "66956ef86729320bed14e8420749eb7e55e20faf", + "stream_describe": "66956ef", + "upstream_jetstream": "f29815c391fc2644f8a3dd36b899fb3697dd1ea6", + "upstream_jetstream_dirty": false, + "zig": "0.16.0" + }, + "schema": "stream-admission/1", + "suites": { + "archive-contract": "pass", + "compaction-config-contract": "pass", + "cursor-lookback-contract": "pass", + "dashboard-test": "pass", + "differential-oracle": "pass", + "environment-contract": "pass", + "http-metrics-contract": "pass", + "lifecycle-oracle": "pass", + "listener-contract": "pass", + "logging-contract": "pass", + "plan-config-contract": "pass", + "powerloss-oracle": "pass", + "process-metrics-contract": "pass", + "retry-config-contract": "pass", + "shutdown-contract": "pass", + "status-contract": "pass", + "subscribe-config-contract": "pass", + "subscribe-read-batch-contract": "pass", + "unit-debug": "pass", + "unit-releasesafe": "pass" + } +} diff --git a/src/internal/runtime/cli.zig b/src/internal/runtime/cli.zig index 91b7153..6e10cdb 100644 --- a/src/internal/runtime/cli.zig +++ b/src/internal/runtime/cli.zig @@ -519,208 +519,208 @@ pub fn parseServe( 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, "--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}); + 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, "--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 }; } diff --git a/src/main.zig b/src/main.zig index 63b70da..52f98ec 100644 --- a/src/main.zig +++ b/src/main.zig @@ -180,20 +180,11 @@ test "merge discovery skip is explicit or automatic for either debug selection" /// Parse the non-negative subset of Go's time.ParseDuration used by upstream's /// duration flags. Like Go, fractional values are truncated below 1 ns. - - - /// Full signed Go duration surface used by urfave/cli DurationFlag. The /// non-negative parser above remains the implementation shared by flags whose /// downstream types cannot represent negative durations. - /// net/http Server.New applies this default when its configured timeout is /// non-positive; the CLI's websocket client-drain duration does not. - - - - - fn writeVersion(io: Io, args: []const []const u8) !void { if (args.len != 0) return error.BadArgs; var buffer: [1024]u8 = undefined; @@ -283,9 +274,6 @@ fn parseBlocksMode(raw: []const u8) ?operational.BlocksMode { return null; } - - - const BindAddress = struct { address: Io.net.IpAddress, port: u16,