diff --git a/README.md b/README.md index fc558eb..dd222f6 100644 --- a/README.md +++ b/README.md @@ -148,6 +148,7 @@ Production bootstrap controls: --backfill --backfill-workers=16 --backfill-max-inflight-bytes=8589934592 +--skip-merge-discovery ``` Worker count controls concurrent network downloads. The byte flag bounds CAR @@ -174,6 +175,13 @@ recent-error content in the browser. This implementation is not admission to another full-network run by itself: `docs/semantic-parity.md` remains authoritative. +The debug-only discovery flag omits the post-merge `listRepos` rescan. Matching +upstream, Stream enables it automatically for `--backfill-max-repos` and +`--backfill-repos`; a normal full crawl still replays the last page so accounts +created during bootstrap enter durable retry state. The pinned-simulator +process receipt observes three real `listRepos` requests for a normal two-page +crawl and exactly two when discovery is skipped. + The same durable account row retains the distinct initial-backfill and latest known revisions, update time, declared handle, PDS endpoint, and reserved record/byte counters. Matching upstream, the debug-only explicit diff --git a/docs/configuration-parity.md b/docs/configuration-parity.md index c418f46..3ecda2d 100644 --- a/docs/configuration-parity.md +++ b/docs/configuration-parity.md @@ -9,7 +9,8 @@ the same runtime mechanism and an offline receipt observes that behavior. |---|---|---| | 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, backfill workers | behavioral equivalent | lifecycle and differential receipts; public max-repo flag naming remains open | -| backfill batch size, async flush workers, skip merge discovery | open | Stream does not yet expose or implement these upstream controls | +| backfill batch size and async flush workers | open | Stream does not yet expose or implement these upstream controls | +| skip merge discovery | closed | explicit flag and automatic max/selected-repo enablement match upstream; the pinned simulator observes the normal discovery request and its absence when skipped | | failed-repo retry interval, global workers, per-host workers, max delay | closed | `just retry-config-contract` measures explicit and zero/default concurrency through held real HTTP requests, then reopens RocksDB to verify one durable failure and the configured jitter/cap window per row | | expensive repo-action rate limiter | closed | real HTTP/status receipt covers enabled and disabled modes | | cursor lookback | closed | `just cursor-lookback-contract` covers default 36h, widened, v1/v2 old-cursor behavior, and zero/pure-live | diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md index 3a8fe0e..7f1a406 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -196,6 +196,14 @@ one additional attempt plus a next-attempt delay inside the configured failure persistence, Go-duration parsing, zero defaults, and max-delay clamping without network access. +The pinned upstream simulator drives four complete bootstrap/merge lifecycles. +The normal two-page crawl makes three real `listRepos` requests: two bootstrap +pages and the final-page merge-discovery replay. With +`--skip-merge-discovery`, it makes exactly the two bootstrap requests. A +max-repo selection proves the same automatic two-request short circuit, while +an explicit-DID selection bypasses `listRepos` entirely. Both automatic cases +also traverse the configured-skip lifecycle branch, matching upstream. + ## pinned differential oracle (2026-07-20) `just differential-oracle` refuses to run unless the upstream checkout is at diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig index 6bc0d34..fb29a31 100644 --- a/src/internal/bootstrap/lifecycle.zig +++ b/src/internal/bootstrap/lifecycle.zig @@ -54,6 +54,9 @@ pub const Options = struct { max_repos: usize = 0, workers: usize = engine.default_workers, max_inflight_bytes: usize = engine.default_max_inflight_bytes, + /// Debug-only upstream control. Main also enables it automatically for + /// max-repo and explicit-DID bootstrap selections. + skip_merge_discovery: bool = false, /// Matches upstream: a zero interval disables both merge-tail and /// steady delete compaction. The destination is still sealed to serve. compaction_enabled: bool = true, @@ -185,7 +188,11 @@ pub fn run( // become failed rows the steady retry loop downloads. a relay // outage here fails the merge (upstream couples them the same // way) — re-entry is idempotent. - try runDiscovery(allocator, io, opts.relay_http, &repo_st, stats); + if (!opts.skip_merge_discovery) { + try runDiscovery(allocator, io, opts.relay_http, &repo_st, stats); + } else { + log.info("merge discovery: skipped by configuration", .{}); + } crashpoint.hit(.after_merge_discovery_before_cleanup); crashpoint.hit(.after_discovery_before_cleanup); diff --git a/src/main.zig b/src/main.zig index 1d98177..84ffa47 100644 --- a/src/main.zig +++ b/src/main.zig @@ -78,6 +78,10 @@ fn parseBackfillRepos(allocator: std.mem.Allocator, raw: []const u8) ![]const [] return out.toOwnedSlice(allocator); } +fn effectiveSkipMergeDiscovery(requested: bool, max_repos: usize, selected_repos: usize) bool { + return requested or max_repos > 0 or selected_repos > 0; +} + 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"); @@ -91,6 +95,14 @@ test "explicit backfill repo parsing matches upstream selection rules" { try testing.expectError(error.InvalidBackfillRepo, parseBackfillRepos(testing.allocator, "not-a-did")); } +test "merge discovery skip is explicit or automatic for either debug selection" { + const testing = std.testing; + try testing.expect(!effectiveSkipMergeDiscovery(false, 0, 0)); + try testing.expect(effectiveSkipMergeDiscovery(true, 0, 0)); + try testing.expect(effectiveSkipMergeDiscovery(false, 1, 0)); + try testing.expect(effectiveSkipMergeDiscovery(false, 0, 1)); +} + /// 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. fn parseDurationNanoseconds(raw: []const u8) !u64 { @@ -208,6 +220,7 @@ pub fn main(init: std.process.Init.Minimal) !void { var do_backfill = false; var backfill_workers: usize = backfill_engine.default_workers; var backfill_max_inflight_bytes: usize = backfill_engine.default_max_inflight_bytes; + 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; @@ -317,6 +330,8 @@ pub fn main(init: std.process.Init.Minimal) !void { 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.eql(u8, arg, "--no-verify")) { do_verify = false; } else if (std.mem.eql(u8, arg, "--disable-repo-action-rate-limits")) { @@ -347,6 +362,9 @@ pub fn main(init: std.process.Init.Minimal) !void { const backfill_repos: []const []const u8 = if (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 + // either partial selection mode, in addition to accepting it explicitly. + skip_merge_discovery = effectiveSkipMergeDiscovery(skip_merge_discovery, backfill_max_repos, backfill_repos.len); var hot_tail = tail_mod.Tail.init(allocator, io, default_tail_bytes); defer hot_tail.deinit(); @@ -509,6 +527,7 @@ pub fn main(init: std.process.Init.Minimal) !void { .selected_repos = backfill_repos, .workers = backfill_workers, .max_inflight_bytes = backfill_max_inflight_bytes, + .skip_merge_discovery = skip_merge_discovery, .compaction_enabled = compaction_interval_ns != 0, }, &archive, &meta, &cursor_store, if (verifier) |*v| v else null, &stats) catch |err| { logPersistenceFailure(data_dir, err); diff --git a/tests/differential_oracle.py b/tests/differential_oracle.py index ef52f4d..bde52c3 100644 --- a/tests/differential_oracle.py +++ b/tests/differential_oracle.py @@ -71,7 +71,7 @@ def main() -> None: go = go_binary() run_pattern = os.environ.get( "STREAM_ORACLE_RUN", - "^TestStream(Differential|ReplayFault|FrameFault|Adversarial|ResyncCapacity)Oracle$", + "^TestStream(Differential|ReplayFault|FrameFault|Adversarial|ResyncCapacity|SkipMergeDiscovery)Oracle$", ) # -overlay can add a test file to an existing Go package. Compiling it as diff --git a/tests/upstream_oracle/stream_differential_test.go b/tests/upstream_oracle/stream_differential_test.go index 8df9369..7b8f6c2 100644 --- a/tests/upstream_oracle/stream_differential_test.go +++ b/tests/upstream_oracle/stream_differential_test.go @@ -18,6 +18,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "syscall" "testing" "time" @@ -36,6 +37,104 @@ type synchronizedBuffer struct { b bytes.Buffer } +type listReposCountingHandler struct { + next http.Handler + calls atomic.Int64 +} + +func (h *listReposCountingHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/xrpc/com.atproto.sync.listRepos" { + h.calls.Add(1) + } + h.next.ServeHTTP(rw, r) +} + +func TestStreamSkipMergeDiscoveryOracle(t *testing.T) { + require.Equal(t, streamDifferentialPin, os.Getenv("STREAM_ORACLE_EXPECTED_PIN"), + "runner must attest the exact pinned upstream checkout") + + simCfg := world.DefaultConfig() + simCfg.DataDir = filepath.Join(t.TempDir(), "simulator") + simCfg.Seed = 0x51ced15c0 + simCfg.Accounts = 4 + simCfg.InitialRecords = 1 + simCfg.CommitsPerSec = 1 + simCfg.FirehoseHistory = 128 + w, err := world.New(t.Context(), simCfg) + require.NoError(t, err) + defer func() { require.NoError(t, w.Close()) }() + _, err = w.EnsureSeed() + require.NoError(t, err) + require.NoError(t, w.Bootstrap(t.Context(), slog.Default())) + fan := fanout.New(128) + require.NoError(t, w.AttachRuntime( + rand.New(rand.NewPCG(simCfg.Seed^0xfeedf00d, simCfg.Seed^0xc0ffee)), + fan, + )) + + server := httptest.NewUnstartedServer(nil) + counter := &listReposCountingHandler{next: simhttp.NewHandlerWithOptions(w, server.URL, simhttp.HandlerOptions{ + ListReposPageLimit: 2, + })} + server.Config.Handler = counter + server.Start() + defer server.Close() + selected, _, err := w.ListReposPage(0, 1) + require.NoError(t, err) + require.Len(t, selected, 1) + + run := func(t *testing.T, selectionArgs []string, expectSkip bool) int64 { + t.Helper() + counter.calls.Store(0) + port := freeStreamOraclePort(t) + baseURL := "http://127.0.0.1:" + strconv.Itoa(port) + logs := &synchronizedBuffer{} + 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, + "--backfill-workers=2", + "--max-segment-bytes=1", + "--compaction-interval=0", + "--retry-interval=0", + } + args = append(args, selectionArgs...) + cmd := exec.Command(streamOracleBinary(t), args...) + cmd.Stdout = logs + cmd.Stderr = logs + require.NoError(t, cmd.Start()) + waitForStreamOracleServing(t, cmd, baseURL, logs) + require.NoError(t, cmd.Process.Signal(syscall.SIGTERM)) + require.NoErrorf(t, waitStreamCommand(cmd, 15*time.Second), + "Stream did not stop cleanly:\n%s", logs.String()) + if expectSkip { + require.Contains(t, logs.String(), "merge discovery: skipped by configuration") + } else { + require.Contains(t, logs.String(), "merge discovery: 0 post-bootstrap DIDs queued for retry") + } + return counter.calls.Load() + } + + t.Run("normal full crawl performs discovery", func(t *testing.T) { + require.Equal(t, int64(3), run(t, []string{"--backfill"}, false), + "two bootstrap pages plus one final-page discovery replay") + }) + t.Run("explicit skip omits discovery", func(t *testing.T) { + require.Equal(t, int64(2), run(t, []string{"--backfill", "--skip-merge-discovery"}, true), + "skip must leave only the two bootstrap pages") + }) + t.Run("max repo selection automatically skips discovery", func(t *testing.T) { + require.Equal(t, int64(2), run(t, []string{"--backfill-max-repos=4"}, true), + "partial crawl must not add a discovery request") + }) + t.Run("explicit DID selection automatically skips discovery", func(t *testing.T) { + require.Equal(t, int64(0), run(t, []string{"--backfill-repos=" + string(selected[0].DID)}, true), + "explicit DID mode must bypass both listRepos and discovery") + }) +} + func (b *synchronizedBuffer) Write(p []byte) (int, error) { b.mu.Lock() defer b.mu.Unlock()