diff --git a/README.md b/README.md index 73f1821..053337d 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,11 @@ The Zig build pins and statically compiles its native dependencies, including RocksDB, zstd, xxHash, and websocket.zig's zlib 1.3.2; it does not depend on a system zlib installation. +Production configuration uses Jetstream V2's canonical `--relay-url`, +`--plc-url`, and `--max-backfill-repos` names. `--relay-url` drives both the +live WebSocket and HTTP repository clients; Stream's older split flags remain +accepted for existing local harnesses. + The subscriber's writer-owned hot log retains 256 MiB by default and never evicts rows above the durable archive watermark. Operators can tune the real byte budget with `--subscribe-read-log-retention-bytes`. The upstream slow diff --git a/deploy/README.md b/deploy/README.md index 794d5a0..dbe0c32 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -18,9 +18,8 @@ just package # ReleaseSafe x86_64-linux-gnu binary in zig-out/bin/ ## flags for production ``` ---upstream=wss://relay.waow.tech ---relay-http=https://relay.waow.tech ---plc=https://plc.directory +--relay-url=https://relay.waow.tech +--plc-url=https://plc.directory --data-dir=/data --backfill --backfill-workers=100 diff --git a/docs/configuration-parity.md b/docs/configuration-parity.md index 23b3d45..0a7fe65 100644 --- a/docs/configuration-parity.md +++ b/docs/configuration-parity.md @@ -7,8 +7,8 @@ the same runtime mechanism and an offline receipt observes that behavior. | Upstream surface | Stream state | Evidence or remaining work | |---|---|---| -| relay, PLC, data directory | behavioral equivalent | `--upstream`, `--relay-http`, `--plc`, and `--data-dir`; public naming and `JETSTREAM_*` environment sources remain open | -| max/selected backfill repos | behavioral equivalent | lifecycle and differential receipts; public max-repo flag naming remains open | +| relay, PLC, data directory | behavioral equivalent | canonical `--relay-url` drives both the real WebSocket consumer and HTTP backfill/repair clients, `--plc-url` drives identity resolution, and `--data-dir` owns storage; the differential oracle now uses those names, while `JETSTREAM_*` environment sources remain open | +| max/selected backfill repos | closed | lifecycle and differential receipts now invoke canonical `--max-backfill-repos` and `--backfill-repos`; the older Stream spelling remains a compatibility alias | | backfill workers | closed | real held-request processes observe explicit 7 and both zero/omitted defaults at 100 simultaneous getRepo requests; all complete through physical archive convergence | | backfill batch size | closed | page-aligned accumulation counts all listRepos entries; configured, zero/default, request ordering, shuffle boundary, and atomic relay/discovery cursor cadence are covered offline | | backfill async flush workers | closed | omitted defaults to four, positive values create that many real bootstrap-only zstd workers, and zero keeps synchronous compression; detached blocks commit/fsync in preparation order before repo completion, with exact four-worker overlap and byte-identical sync/async JSS receipts | @@ -21,7 +21,7 @@ the same runtime mechanism and an offline receipt observes that behavior. | subscribe hot-log retention and slow window/rate | closed | the flags drive the real durability-pinned byte eviction and per-subscriber log-scan detector; unit receipts cover physical eviction plus configured fractional rate/window, while `just subscribe-config-contract` runs explicit and non-positive/default values through a ReleaseSafe process | | subscribe cold-block cache | closed | the configured byte-bounded LRU shares real decoded JSS blocks plus v1/v2 JSON and dictionary-zstd memos; concurrent replay proves exact-one decode and pointer-identical bodies, memo growth triggers eviction, and manifest refresh invalidates the generation | | subscribe read batch | closed | `--subscribe-read-batch` defaults non-positive values to 1024 and bounds raw scanned entries before filters/v1 skips in both hot and cold pulls; cold snapshots atomically pair sealed metadata with the active segment's fsynced byte boundary, and receipts cover that active region plus a rejecting 5,000-row production replay with exactly 295 pulls at batch 17 and a stable hot handoff | -| deprecated cursor block-index cache flag | open compatibility surface | upstream accepts it as an explicit no-op; Stream rejects it | +| deprecated cursor block-index cache flag | closed | Stream accepts positive, zero, and negative signed values as the same explicit no-op because sealed metadata is always manifest-resident; the ReleaseSafe config receipt proves the public parser surface | | compaction interval | closed | Go-duration parsing, zero disablement, a real 250 ms production pass, and immediate coalesced cap-trigger wake behavior are covered offline | | compaction tombstone cap and rewrite workers | closed | `just compaction-config-contract` observes chunk count, actual rewrite group size, physical survivors, and durable watermark; zero sentinels match upstream | | timestamp import token and directory | closed | real HTTP/import, restart, and strict power-loss receipts | diff --git a/justfile b/justfile index e950402..4da6f85 100644 --- a/justfile +++ b/justfile @@ -265,14 +265,17 @@ subscribe-config-contract: } run_case explicit 'read-log-retention=12345 bytes block-cache=23456 bytes read-batch=17 slow-window=250000000ns slow-min-rate=7.5' \ --subscribe-read-log-retention-bytes=12345 --subscribe-slow-window=250ms \ - --subscribe-slow-min-rate=7.5 --subscribe-block-cache-bytes=23456 --subscribe-read-batch=17 + --subscribe-slow-min-rate=7.5 --subscribe-block-cache-bytes=23456 --subscribe-read-batch=17 \ + --cursor-block-index-cache-size=99 run_case zero 'read-log-retention=268435456 bytes block-cache=67108864 bytes read-batch=1024 slow-window=60000000000ns slow-min-rate=5' \ --subscribe-read-log-retention-bytes=0 --subscribe-slow-window=0 \ - --subscribe-slow-min-rate=0 --subscribe-block-cache-bytes=0 --subscribe-read-batch=0 + --subscribe-slow-min-rate=0 --subscribe-block-cache-bytes=0 --subscribe-read-batch=0 \ + --cursor-block-index-cache-size=0 run_case negative 'read-log-retention=268435456 bytes block-cache=67108864 bytes read-batch=1024 slow-window=60000000000ns slow-min-rate=5' \ --subscribe-read-log-retention-bytes=-1 --subscribe-slow-window=-1s \ - --subscribe-slow-min-rate=-1 --subscribe-block-cache-bytes=-1 --subscribe-read-batch=-1 - echo 'subscribe config: PASS (retention/cache/read-batch/slow controls reach runtime)' + --subscribe-slow-min-rate=-1 --subscribe-block-cache-bytes=-1 --subscribe-read-batch=-1 \ + --cursor-block-index-cache-size=-1 + echo 'subscribe config: PASS (retention/cache/read-batch/slow controls + deprecated no-op reach runtime)' # Replay a real 5,000-row sealed archive through the production WebSocket. # A rejecting filter proves the batch boundary counts raw rows, not frames. diff --git a/src/main.zig b/src/main.zig index fd72194..2336903 100644 --- a/src/main.zig +++ b/src/main.zig @@ -83,6 +83,51 @@ fn effectiveSkipMergeDiscovery(requested: bool, max_repos: usize, selected_repos return requested or max_repos > 0 or selected_repos > 0; } +const RelayUrls = struct { + http: []u8, + websocket: []u8, + + fn deinit(self: RelayUrls, allocator: std.mem.Allocator) void { + allocator.free(self.http); + allocator.free(self.websocket); + } +}; + +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 }), + }; +} + +test "canonical relay URL normalizes HTTP and live websocket endpoints" { + const testing = std.testing; + const secure = try normalizeRelayUrls(testing.allocator, "https://relay.example/ignored?x=1#fragment"); + defer secure.deinit(testing.allocator); + try testing.expectEqualStrings("https://relay.example", secure.http); + try testing.expectEqualStrings("wss://relay.example", secure.websocket); + const local = try normalizeRelayUrls(testing.allocator, "http://127.0.0.1:7777"); + defer local.deinit(testing.allocator); + try testing.expectEqualStrings("http://127.0.0.1:7777", local.http); + try testing.expectEqualStrings("ws://127.0.0.1:7777", local.websocket); + try testing.expectError(error.InvalidRelayUrl, normalizeRelayUrls(testing.allocator, "ws://relay.example")); + try testing.expectError(error.InvalidRelayUrl, normalizeRelayUrls(testing.allocator, "https:///missing")); +} + test "explicit backfill repo parsing matches upstream selection rules" { const testing = std.testing; const parsed = try parseBackfillRepos(testing.allocator, "did:plc:aaa, did:web:example.com"); @@ -228,8 +273,9 @@ pub fn main(init: std.process.Init.Minimal) !void { const io = threaded.io(); var upstream: []const u8 = "ws://localhost:7777"; - // the simulator serves DID docs itself; production would pass - // --plc=https://plc.directory + 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; @@ -273,18 +319,32 @@ pub fn main(init: std.process.Init.Minimal) !void { while (arg_it.next()) |arg| { 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); } 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=")) { @@ -346,6 +406,11 @@ pub fn main(init: std.process.Init.Minimal) !void { 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=")) { @@ -588,7 +653,7 @@ pub fn main(init: std.process.Init.Minimal) !void { // lifecycle: a stored non-steady phase means a crash mid-bootstrap or // mid-merge — always resume it. otherwise only --backfill (full crawl) - // or --backfill-max-repos (debug) start one; serving is 503-gated + // or --max-backfill-repos (debug) start one; serving is 503-gated // until the lifecycle commits steady_state. if (mid_lifecycle or bootstrap_requested) { lifecycle.run(allocator, io, .{ diff --git a/tests/oracle.py b/tests/oracle.py index a69314a..29d2a1e 100644 --- a/tests/oracle.py +++ b/tests/oracle.py @@ -41,10 +41,9 @@ ARGS = [ BIN, f"--port={PORT}", f"--data-dir={DATA}", - "--upstream=ws://127.0.0.1:7777", - f"--relay-http={SIM}", - f"--plc={SIM}", - "--backfill-max-repos=25", + f"--relay-url={SIM}", + f"--plc-url={SIM}", + "--max-backfill-repos=25", "--compaction-interval=0", "--retry-interval=0", ] diff --git a/tests/powerloss_oracle.py b/tests/powerloss_oracle.py index 7f8d91b..f72f04c 100644 --- a/tests/powerloss_oracle.py +++ b/tests/powerloss_oracle.py @@ -311,10 +311,9 @@ def calibrate(state): def stream_args(point=None, ordinal=1): args = ( "/oracle/stream --port=6099 --data-dir=/mnt/strict/data " - f"--upstream=ws://{UPSTREAM_HOST}:7777 " - f"--relay-http=http://{UPSTREAM_HOST}:7777 " - f"--plc=http://{UPSTREAM_HOST}:7777 " - "--backfill-max-repos=25 --compaction-interval=0 --retry-interval=0 --no-verify" + f"--relay-url=http://{UPSTREAM_HOST}:7777 " + f"--plc-url=http://{UPSTREAM_HOST}:7777 " + "--max-backfill-repos=25 --compaction-interval=0 --retry-interval=0 --no-verify" ) if point: args += f" --power-cutpoint={point}:{ordinal}" @@ -330,8 +329,8 @@ def mutation_stream_args(kind, point=None): interval = "1s" if kind == "compaction" else "0" args = ( "/oracle/stream --port=6099 --data-dir=/mnt/strict/data " - "--upstream=ws://127.0.0.1:1 --relay-http=http://127.0.0.1:1 " - "--plc=http://127.0.0.1:1 --retry-interval=0 --no-verify " + "--relay-url=http://127.0.0.1:1 " + "--plc-url=http://127.0.0.1:1 --retry-interval=0 --no-verify " f"--compaction-interval={interval} " "--timestamp-import-dir=/mnt/strict/imports " "--timestamp-import-token=powerloss-test-token" diff --git a/tests/upstream_oracle/stream_differential_test.go b/tests/upstream_oracle/stream_differential_test.go index e90be45..5ced1f9 100644 --- a/tests/upstream_oracle/stream_differential_test.go +++ b/tests/upstream_oracle/stream_differential_test.go @@ -149,9 +149,8 @@ func TestStreamBackfillWorkerControlsOracle(t *testing.T) { args := []string{ "--port=" + strconv.Itoa(port), "--data-dir=" + filepath.Join(t.TempDir(), "stream"), - "--upstream=" + strings.Replace(server.URL, "http://", "ws://", 1), - "--relay-http=" + server.URL, - "--plc=" + server.URL, + "--relay-url=" + server.URL, + "--plc-url=" + server.URL, "--backfill", "--skip-merge-discovery", "--max-segment-bytes=1", @@ -267,9 +266,8 @@ func TestStreamBootstrapControlsOracle(t *testing.T) { args := []string{ "--port=" + strconv.Itoa(port), "--data-dir=" + filepath.Join(t.TempDir(), "stream"), - "--upstream=" + strings.Replace(server.URL, "http://", "ws://", 1), - "--relay-http=" + server.URL, - "--plc=" + server.URL, + "--relay-url=" + server.URL, + "--plc-url=" + server.URL, "--backfill-workers=2", "--max-segment-bytes=1", "--compaction-interval=0", @@ -310,7 +308,7 @@ func TestStreamBootstrapControlsOracle(t *testing.T) { "skip must leave only the two bootstrap pages") }) t.Run("max repo selection automatically skips discovery", func(t *testing.T) { - calls, _ := run(t, []string{"--backfill-max-repos=4"}, true) + calls, _ := run(t, []string{"--max-backfill-repos=4"}, true) require.Equal(t, int64(2), calls, "partial crawl must not add a discovery request") }) @@ -410,10 +408,9 @@ func TestStreamDifferentialOracle(t *testing.T) { cmd := exec.CommandContext(ctx, streamBin, "--port="+strconv.Itoa(port), "--data-dir="+dataDir, - "--upstream="+strings.Replace(simSrv.URL, "http://", "ws://", 1), - "--relay-http="+simSrv.URL, - "--plc="+simSrv.URL, - "--backfill-max-repos="+strconv.Itoa(simCfg.Accounts), + "--relay-url="+simSrv.URL, + "--plc-url="+simSrv.URL, + "--max-backfill-repos="+strconv.Itoa(simCfg.Accounts), "--backfill-workers=4", "--max-segment-bytes=1", "--compaction-interval=0", diff --git a/tests/upstream_oracle/stream_fault_differential_test.go b/tests/upstream_oracle/stream_fault_differential_test.go index 05cf81b..0e0f009 100644 --- a/tests/upstream_oracle/stream_fault_differential_test.go +++ b/tests/upstream_oracle/stream_fault_differential_test.go @@ -98,9 +98,8 @@ func (h *streamFaultHarness) startWithArgs(t *testing.T, extra ...string) { args := []string{ "--port=" + port, "--data-dir=" + h.dataDir, - "--upstream=" + strings.Replace(h.server.URL, "http://", "ws://", 1), - "--relay-http=" + h.server.URL, - "--plc=" + h.server.URL, + "--relay-url=" + h.server.URL, + "--plc-url=" + h.server.URL, "--max-segment-bytes=1", "--compaction-interval=0", "--retry-interval=0", @@ -600,7 +599,7 @@ func TestStreamAdversarialOracle(t *testing.T) { h := newStreamFaultHarnessWithShape(t, 0x206402, 2, 2) injectAdversarialBackfillLies(t, h.world, 0) h.startWithArgs(t, - "--backfill-max-repos=2", + "--max-backfill-repos=2", "--backfill-workers=2", ) metrics := scrapeStreamOracleMetrics(t, h.baseURL) diff --git a/tests/upstream_oracle/stream_restart_predicate_test.go b/tests/upstream_oracle/stream_restart_predicate_test.go index b43e8ef..5992995 100644 --- a/tests/upstream_oracle/stream_restart_predicate_test.go +++ b/tests/upstream_oracle/stream_restart_predicate_test.go @@ -154,10 +154,9 @@ func streamRestartArgs(baseURL, dataDir, relayURL string, accounts int) []string return []string{ "--port=" + port, "--data-dir=" + dataDir, - "--upstream=" + strings.Replace(relayURL, "http://", "ws://", 1), - "--relay-http=" + relayURL, - "--plc=" + relayURL, - "--backfill-max-repos=" + strconv.Itoa(accounts), + "--relay-url=" + relayURL, + "--plc-url=" + relayURL, + "--max-backfill-repos=" + strconv.Itoa(accounts), "--backfill-workers=4", "--max-segment-bytes=1", "--compaction-interval=1s", diff --git a/tests/upstream_oracle/stream_resync_capacity_test.go b/tests/upstream_oracle/stream_resync_capacity_test.go index aace968..15840a2 100644 --- a/tests/upstream_oracle/stream_resync_capacity_test.go +++ b/tests/upstream_oracle/stream_resync_capacity_test.go @@ -15,7 +15,6 @@ import ( "os/exec" "path/filepath" "strconv" - "strings" "sync" "sync/atomic" "syscall" @@ -111,9 +110,8 @@ func TestStreamResyncCapacityOracle(t *testing.T) { cmd := exec.CommandContext(ctx, streamBin, "--port="+strconv.Itoa(port), "--data-dir="+dataDir, - "--upstream="+strings.Replace(simSrv.URL, "http://", "ws://", 1), - "--relay-http="+simSrv.URL, - "--plc="+simSrv.URL, + "--relay-url="+simSrv.URL, + "--plc-url="+simSrv.URL, "--max-segment-bytes=4194304", "--compaction-interval=0", "--retry-interval=0",