Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192const std = @import("std");
pub const jetstream_prefix = "JETSTREAM_";
pub const Mapping = struct { env: []const u8, flag: []const u8, serve: bool = true,};
// This is the exact flag-source set declared by the pinned Jetstream V2// command tree. Root-only values have no synthesized serve flag.pub const mappings = [_]Mapping{ .{ .env = "JETSTREAM_LOG_LEVEL", .flag = "log-level", .serve = false }, .{ .env = "JETSTREAM_LOG_FORMAT", .flag = "log-format", .serve = false }, .{ .env = "JETSTREAM_ADDR", .flag = "addr" }, .{ .env = "JETSTREAM_DEBUG_ADDR", .flag = "debug-addr" }, .{ .env = "JETSTREAM_SHUTDOWN_TIMEOUT", .flag = "shutdown-timeout" }, .{ .env = "JETSTREAM_CLIENT_DRAIN_TIMEOUT", .flag = "client-drain-timeout" }, .{ .env = "JETSTREAM_RELAY_URL", .flag = "relay-url" }, .{ .env = "JETSTREAM_PLC_URL", .flag = "plc-url" }, .{ .env = "JETSTREAM_DATA_DIR", .flag = "data-dir" }, .{ .env = "JETSTREAM_MAX_BACKFILL_REPOS", .flag = "max-backfill-repos" }, .{ .env = "JETSTREAM_BACKFILL_WORKERS", .flag = "backfill-workers" }, .{ .env = "JETSTREAM_BACKFILL_BATCH_SIZE", .flag = "backfill-batch-size" }, .{ .env = "JETSTREAM_BACKFILL_ASYNC_FLUSH_WORKERS", .flag = "backfill-async-flush-workers" }, .{ .env = "JETSTREAM_BACKFILL_REPOS", .flag = "backfill-repos" }, .{ .env = "JETSTREAM_SKIP_MERGE_DISCOVERY", .flag = "skip-merge-discovery" }, .{ .env = "JETSTREAM_REBLOOM_SWEEP", .flag = "rebloom-sweep" }, .{ .env = "JETSTREAM_FAILED_REPO_RETRY_INTERVAL", .flag = "failed-repo-retry-interval" }, .{ .env = "JETSTREAM_FAILED_REPO_RETRY_WORKERS", .flag = "failed-repo-retry-workers" }, .{ .env = "JETSTREAM_FAILED_REPO_RETRY_HOST_WORKERS", .flag = "failed-repo-retry-host-workers" }, .{ .env = "JETSTREAM_FAILED_REPO_RETRY_MAX_DELAY", .flag = "failed-repo-retry-max-delay" }, .{ .env = "JETSTREAM_DISABLE_REPO_ACTION_RATE_LIMITS", .flag = "disable-repo-action-rate-limits" }, .{ .env = "JETSTREAM_CURSOR_LOOKBACK", .flag = "cursor-lookback" }, .{ .env = "JETSTREAM_SEGMENT_CACHE_MAX_AGE", .flag = "segment-cache-max-age" }, .{ .env = "JETSTREAM_PLAN_MAX_DIDS", .flag = "plan-max-dids" }, .{ .env = "JETSTREAM_PLAN_MAX_COLLECTIONS", .flag = "plan-max-collections" }, .{ .env = "JETSTREAM_PLAN_MAX_ENTRIES", .flag = "plan-max-entries" }, .{ .env = "JETSTREAM_PLAN_WHOLE_SEGMENT_THRESHOLD", .flag = "plan-whole-segment-threshold" }, .{ .env = "JETSTREAM_SUBSCRIBE_READ_LOG_RETENTION_BYTES", .flag = "subscribe-read-log-retention-bytes" }, .{ .env = "JETSTREAM_SUBSCRIBE_BLOCK_CACHE_BYTES", .flag = "subscribe-block-cache-bytes" }, .{ .env = "JETSTREAM_SUBSCRIBE_READ_BATCH", .flag = "subscribe-read-batch" }, .{ .env = "JETSTREAM_SUBSCRIBE_SLOW_WINDOW", .flag = "subscribe-slow-window" }, .{ .env = "JETSTREAM_SUBSCRIBE_SLOW_MIN_RATE", .flag = "subscribe-slow-min-rate" }, .{ .env = "JETSTREAM_CURSOR_BLOCK_INDEX_CACHE_SIZE", .flag = "cursor-block-index-cache-size" }, .{ .env = "JETSTREAM_COMPACTION_INTERVAL", .flag = "compaction-interval" }, .{ .env = "JETSTREAM_COMPACTION_TOMBSTONE_CAP", .flag = "compaction-tombstone-cap" }, .{ .env = "JETSTREAM_COMPACTION_REWRITE_WORKERS", .flag = "compaction-rewrite-workers" }, .{ .env = "JETSTREAM_TIMESTAMP_IMPORT_TOKEN", .flag = "timestamp-import-token" }, .{ .env = "JETSTREAM_TIMESTAMP_IMPORT_DIR", .flag = "timestamp-import-dir" }, .{ .env = "JETSTREAM_ARCHIVE_API_KEY", .flag = "archive-api-key" }, .{ .env = "JETSTREAM_ARCHIVE_API_KEYS_FILE", .flag = "archive-api-keys-file" }, .{ .env = "JETSTREAM_MAX_SUBSCRIBERS", .flag = "max-subscribers" }, .{ .env = "JETSTREAM_MAX_COLD_READERS", .flag = "max-cold-readers" }, .{ .env = "JETSTREAM_UPSTREAM_SLOW_MIN_RATE", .flag = "upstream-slow-min-rate" },};
pub fn processEntries(allocator: std.mem.Allocator) ![]const []const u8 { var out: std.ArrayList([]const u8) = .empty; errdefer out.deinit(allocator); var i: usize = 0; while (std.c.environ[i]) |entry| : (i += 1) { try out.append(allocator, std.mem.span(entry)); } return out.toOwnedSlice(allocator);}
pub fn value(entries: []const []const u8, key: []const u8) ?[]const u8 { for (entries) |entry| { const split = std.mem.indexOfScalar(u8, entry, '=') orelse entry.len; if (std.mem.eql(u8, entry[0..split], key)) { return if (split < entry.len) entry[split + 1 ..] else ""; } } return null;}
pub fn unknown(allocator: std.mem.Allocator, entries: []const []const u8) ![]const []const u8 { var out: std.ArrayList([]const u8) = .empty; errdefer out.deinit(allocator); for (entries) |entry| { const split = std.mem.indexOfScalar(u8, entry, '=') orelse entry.len; const key = entry[0..split]; if (!std.mem.startsWith(u8, key, jetstream_prefix) or std.mem.startsWith(u8, key, "JETSTREAM_APP_") or std.mem.startsWith(u8, key, "JETSTREAM_ORACLE_") or std.mem.startsWith(u8, key, "JETSTREAM_SIM_") or isKnown(key)) continue; var seen = false; for (out.items) |prior| { if (std.mem.eql(u8, prior, key)) { seen = true; break; } } if (seen) continue; try out.append(allocator, key); } std.mem.sort([]const u8, out.items, {}, struct { fn lessThan(_: void, left: []const u8, right: []const u8) bool { return std.mem.order(u8, left, right) == .lt; } }.lessThan); return out.toOwnedSlice(allocator);}
pub fn appendServeArguments( allocator: std.mem.Allocator, entries: []const []const u8, cli_args: []const []const u8, out: *std.ArrayList([]u8),) !void { for (mappings) |mapping| { if (!mapping.serve) continue; if (hasCliFlag(cli_args, mapping.flag)) continue; const configured = value(entries, mapping.env) orelse continue; const rendered = try std.fmt.allocPrint(allocator, "--{s}={s}", .{ mapping.flag, configured }); errdefer allocator.free(rendered); try out.append(allocator, rendered); }}
fn hasCliFlag(args: []const []const u8, flag: []const u8) bool { for (args) |arg| { if (arg.len < 2 or arg[0] != '-' or arg[1] != '-') continue; const body = arg[2..]; if (std.mem.eql(u8, body, flag)) return true; if (body.len > flag.len and body[flag.len] == '=' and std.mem.eql(u8, body[0..flag.len], flag)) return true; } return false;}
fn isKnown(key: []const u8) bool { for (mappings) |mapping| if (std.mem.eql(u8, mapping.env, key)) return true; return false;}
test "unknown Jetstream variables are sorted, deduplicated, and scoped" { try std.testing.expectEqual(@as(usize, 43), mappings.len); const entries = [_][]const u8{ "JETSTREAM_ZZZ=1", "JETSTREAM_ADDR=127.0.0.1:0", "JETSTREAM_APP_INSTANCE=jetstream-0", "JETSTREAM_ORACLE_SEED=7001", "JETSTREAM_SIM_DATA_DIR=./sim", "OTEL_SERVICE_NAME=test", "JETSTREAM_AAA=value=with=equals", "JETSTREAM_ZZZ=again", }; const got = try unknown(std.testing.allocator, &entries); defer std.testing.allocator.free(got); try std.testing.expectEqual(@as(usize, 2), got.len); try std.testing.expectEqualStrings("JETSTREAM_AAA", got[0]); try std.testing.expectEqualStrings("JETSTREAM_ZZZ", got[1]);}
test "serve flag sources preserve empty and equals-containing values" { const entries = [_][]const u8{ "JETSTREAM_ADDR=", "JETSTREAM_TIMESTAMP_IMPORT_TOKEN=secret=part", "JETSTREAM_LOG_LEVEL=debug", }; var args: std.ArrayList([]u8) = .empty; defer { for (args.items) |arg| std.testing.allocator.free(arg); args.deinit(std.testing.allocator); } try appendServeArguments(std.testing.allocator, &entries, &.{}, &args); try std.testing.expectEqualStrings("--addr=", args.items[0]); try std.testing.expectEqualStrings("--timestamp-import-token=secret=part", args.items[1]); try std.testing.expectEqual(@as(usize, 2), args.items.len);}
test "explicit CLI sources suppress lower-priority environment values" { const entries = [_][]const u8{ "JETSTREAM_ADDR=bad-env", "JETSTREAM_SUBSCRIBE_READ_BATCH=not-an-int", }; const cli = [_][]const u8{ "--addr=127.0.0.1:8080", "--subscribe-read-batch=7", }; var args: std.ArrayList([]u8) = .empty; defer { for (args.items) |arg| std.testing.allocator.free(arg); args.deinit(std.testing.allocator); } try appendServeArguments(std.testing.allocator, &entries, &cli, &args); try std.testing.expectEqual(@as(usize, 0), args.items.len);}