diff --git a/docs/full-network-experiment-2026-07.md b/docs/full-network-experiment-2026-07.md index a88962a..fcd16d6 100644 --- a/docs/full-network-experiment-2026-07.md +++ b/docs/full-network-experiment-2026-07.md @@ -142,3 +142,26 @@ The live implementation gate is maintained in [`bootstrap-semantic-parity.md`](bootstrap-semantic-parity.md). It records both the size-limit removal work and the remaining deployment blockers; none of the preserved experiment measurements waive that gate. + +## 2026-07-21 live resync capacity receipt + +The immutable `76f7eb0` live-only candidate was briefly placed on the +colocated canary with a 4 GiB container limit after its repair wire-encoding +ownership fix passed offline. It was rejected and rolled back, not promoted. +At 112 seconds the process reported 4,203,716,608 bytes RSS, all 32 upstream- +equivalent resync workers active, the exact 64-job queue full, 112 completed +repairs, two failed fetches, and eight pending same-DID events. Pipeline tickets +were 783 submitted, 783 claimed, and 393 emitted; public live conformance had no +matching events during its two eight-second observation windows. The prior +public build recovered immediately after rollback. + +The corresponding offline receipt uses the pinned simulator, 96 signed repos, +real chain breaks, real HTTP getRepo CARs, real MST validation, RocksDB, and JSS. +It reproduces exactly 32 active plus 64 queued repairs, then releases and +completes all 96 with zero failures and exact simulator-ground-truth convergence. +With uniformly small 128-record repos it measured 54,886,400 bytes RSS at +saturation and a 72,695,808-byte peak. This distinguishes a scheduler leak from +the intended concurrency multiplier: retained bytes scale with the sizes of up +to 32 simultaneous complete repositories. A parity-default candidate therefore +requires the upstream-sized 32 GiB memory class Jim reported; the 4 GiB +colocated canary is not an admissible host for that build. diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 1312dbc..b9d47c3 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -17,7 +17,7 @@ this document or `bootstrap-semantic-parity.md` is open. | Archive XRPC | Official Go client plus Stream conformance replay over listSegments/getSegment/getBlock/planBackfill | **closed.** With the pinned Zig dependencies cached, `just archive-contract` seeds three real sealed JSS segments and runs without network access. The pinned Go client consumed 5,000 ordered whole-segment rows, the exact 2,500-row DID block plan, 50 required identity sentinels for an empty collection match, the 234-row `(1000,1234]` bound, and a duplicate-free archive-to-live cutover at cursor 5000. Direct conformance also proves checksums/ETags, byte-identical segments, raw block decode, error names, DID/collection blooms, plan modes, and HTTP ranges. Sealed reads use the pinned vendored libzstd so per-block content checksums are enforced independently of the header/footer xxh3; mutating only an upstream fixture's block checksum leaves metadata parsing valid and is rejected at decode, matching upstream. The planner uses refcounted manifest-resident headers, block envelopes, DID blooms, and collection summaries; the offline contract hides every JSS file after startup and still requires an exact plan, proving the request path performs no segment-file I/O. The harness rejects any upstream checkout not at the recorded pin. | | Subscribe v1/v2 | Wire/filter/cursor/compression unit and local e2e coverage | Cursor parsing/resolution precedes upgrade; both endpoints use the seq/time-us magnitude split, v1 clamps below-floor seqs, and v2 rejects them. Timestamp translation uses bounded positional reads of real sealed JSS headers, block indexes, and one selected block before upgrade; disk/decode/index faults are retryable generic 503s, logged internally and counted as `resolve_failed`. Hot/cold scan, skip, encode, options-update, oversize-parser, clean peer/server close, cursor-resolution, and sustained adversarial-rate boundaries feed canonical metrics, with deterministic offline tests derived from upstream. Subscriber transport sends a ping every 30 seconds, applies a kernel-enforced five-second deadline to every frame write, and sends close code 1001 before intentional server shutdown while keeping transport failure out of clean-disconnect accounting. A failed delivery, cold read, ping, or adversarial-rate check interrupts the server reader and removes the connection instead of leaving a ping-only zombie. V1 now negotiates genuine RFC 7692 deflate with the upstream 128-byte threshold and 32 KiB context takeover; v1 custom zstd and all v2 connections explicitly decline it. V2 dictionary negotiation, exact pinned dictionary download/validators, and v2 binary delivery remain independent of the frozen v1 dictionary contract; pinned official Go clients have decoded both compression families offline. | | Sync 1.1 live verification | Upstream requires durable per-DID chain/hosting state, MST inversion, op-CID consistency, rev replay/future guards, default acceptance of legacy-shaped commits, transparent whole-repo repair, and Atmos delivery scheduling | **closed at the implementation boundary.** Diff-CAR/MST inversion (including upstream's narrow default lenient carve-out), post-state op-CID proof, signed inner/outer consistency, exact decimal size/future-rev/replay gates, exact legacy-shape detection with default `LegacyAccept`, two-phase durable chain/hosting state, and account/identity replay ratchets are implemented. Recoverable decode/inversion/duplicate-path/op-CID/chain failures route to repair; signature failures and outer/inner producer-integrity mismatches bypass it, and no failed event is archived. `prepareRepo` retains and authenticates the canonical signed complete head without a second multi-gigabyte parse/walk. The live path now uses Atmos's 32 worker slots, one worker-held FIFO chain per DID, a 64-pending drop-oldest boundary, completion-order batches of 50/500 ms across DIDs, and `min(inflight)-1` cursor watermarks. Dropped and silent events leave the inflight set without inventing a delivery batch; cursor advancement remains coupled to durable archive boundaries. Offline receipts use real encoded frames, real RocksDB verifier state, and actual worker blocking rather than fixture verdicts or scheduler mocks. See `live-scheduler.md`. | -| Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | **closed at the implementation boundary.** The live coordinator has 32 real fetch workers, a 64-job queue, a five-minute fetch budget plus the 1,000 B/s-for-30-seconds slow-transfer guard, a bounded 16,384-entry 5/minute-per-DID limiter, 2,048 pending commits with drop-oldest overflow, fetch outside the DID lane, authenticated apply with older/equal-contradictory head rejection, ordered pending replay, and trigger-ticket-gated outbox delivery. Commit divergence schedules async repair; `#sync` divergence attempts inline repair and queues one retry only for transient failure, preserving the original relay cursor only on inline success. Matching Atmos's gate ownership, a divergent `#sync` repairs before envelope/inner validation; an invalid envelope rev then durably advances only the authenticated chain and drops the original tombstone plus every replacement row as one `live/invalid_rev` event. Valid repairs emit `sync` then bounded 1,024-row `create_resync` batches and stage fetched/pending chain checkpoints under the archive lock before any matching fsync. The shared hot tail is an amortized physical deque: mass replacement eviction cannot shift the full 256 MiB log once per row, and a 20,000-entry burst regression locks that boundary. Wire encoding uses a per-row scratch arena before the tail takes its durable copy, matching upstream's bounded append/publish lifetime instead of retaining v1 and v2 JSON for the entire fetched repository. A real 4,096-record MST repair now completes under a 1 MiB repair-scratch cap; the prior ownership model deterministically fails the same test with `ArchiveAppendFailed`. Loopback-only tests exercise a real HTTP getRepo mmap, real DID resolution/signature, buffering while the response is held open, slow-transfer cancellation, durable RocksDB promotion/cursor coupling, v2 tail publication, and sealed JSS row order. | +| Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | **closed at the implementation boundary.** The live coordinator has 32 real fetch workers, a 64-job queue, a five-minute fetch budget plus the 1,000 B/s-for-30-seconds slow-transfer guard, a bounded 16,384-entry 5/minute-per-DID limiter, 2,048 pending commits with drop-oldest overflow, fetch outside the DID lane, authenticated apply with older/equal-contradictory head rejection, ordered pending replay, and trigger-ticket-gated outbox delivery. Commit divergence schedules async repair; `#sync` divergence attempts inline repair and queues one retry only for transient failure, preserving the original relay cursor only on inline success. Matching Atmos's gate ownership, a divergent `#sync` repairs before envelope/inner validation; an invalid envelope rev then durably advances only the authenticated chain and drops the original tombstone plus every replacement row as one `live/invalid_rev` event. Valid repairs emit `sync` then bounded 1,024-row `create_resync` batches and stage fetched/pending chain checkpoints under the archive lock before any matching fsync. The shared hot tail is an amortized physical deque: mass replacement eviction cannot shift the full 256 MiB log once per row, and a 20,000-entry burst regression locks that boundary. Wire encoding uses a per-row scratch arena before the tail takes its durable copy, matching upstream's bounded append/publish lifetime instead of retaining v1 and v2 JSON for the entire fetched repository. A real 4,096-record MST repair now completes under a 1 MiB repair-scratch cap; the prior ownership model deterministically fails the same test with `ArchiveAppendFailed`. A separate pinned-simulator saturation receipt drives 96 real chain breaks through the ReleaseSafe process, observes the exact 32-active/64-queued Atmos envelope, completes all 96 real CAR/MST repairs, and converges the physical archive to independent ground truth. Loopback-only tests additionally exercise real DID resolution/signature, buffering while a response is held open, slow-transfer cancellation, durable RocksDB promotion/cursor coupling, v2 tail publication, and sealed JSS row order. | | Timestamp import | Canonical dashboard and API include timestamp import lifecycle/rewrite metrics and behavior | **Closed at the implementation boundary.** The format preserves distinct `witnessed_at` and sentinel-zero `indexed_at` columns. Subscriber v1/v2 encoding applies upstream's `indexed_at != 0 ? indexed_at : witnessed_at` display rule while ranges and timestamp cursors remain anchored to immutable witness envelopes. The strict seekable RFC 4180 parser matches upstream's header, row rejection, byte-offset, CID, RFC3339, 64 KiB, and bounded-sampling contracts in a forward-only allocation-free pass. A separate Bloom-filtered RocksDB rule store bulk-loads sorted external SST chunks with CSV-order last-write-wins, reconstructs its resident collection gate on open, and implements specific-CID precedence with all-version fallback. Its central archive-lock hook stamps live, repair, bootstrap, and merge materializations before sequence assignment, and feeds the mutated rows to immediate delivery so hot and cold encodings agree. The sealed-JSS patch primitive preserves block topology/envelopes, clean frames, and the opaque bloom/collection footer while atomically changing only `indexed_at`; it is exercised against an upstream-produced segment. A concrete sealed-segment bloom catalog, bounded DID/FD bucketer, fsynced packed offsets, positioned revalidation, and collision-safe per-segment patch plan implement Phase B/C with specific-CID precedence and idempotent disk-to-disk coverage. A dedicated archive rewrite mutex serializes the real delete compactor with timestamp patching while leaving live append available. The unified metadata DB synchronously persists the current-job pointer, complete job record, and per-segment done set. The real runner activates rules before force-seal, durably crosses the bucketed handoff, checkpoints only after patch fsync/rename/dir-sync, retains partial cancellation progress, and resumes after full archive/rule/metadata reopen. The background manager canonicalizes symlinks, confines regular files, enforces one durable nonterminal job, and adopts it on startup. Bearer-gated import/status XRPC matches the pinned lexicons, fixed-width digest comparison, error names/statuses, steady-state admission, and current/by-id status behavior; a real loopback HTTP run exercised it against the local simulator. Every canonical import metric is produced at its real parse, route, patch, or durable-terminal boundary; cancellation remains a resumable pause and is not counted as terminal failure. The strict Linux campaign cuts all four patch write/fsync/rename/dir-sync boundaries and proves exact target and untargeted timestamps after daemon recovery and public V2 replay; it also exposed and closed a Linux `O_PATH` directory-fsync defect. | | Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | **closed at the implementation boundary.** Durable v5 repo rows distinguish initial Backfill.Rev from latest Rev/UpdatedAt, retain resolved Handle/PDS and reserved record/byte fields, and atomically maintain `handle/`. Merge source cursor commits refresh latest revisions without mutating backfill watermarks. Same-batch `host/` totals, active/status counts, cumulative error classes, and five bounded recent samples survive restart; host moves and active flips are atomic with the source repo row. `/status?tab=hosts` reads those rows without a whole-network scan and matches upstream ordering and error presentation. Account queries preserve resolver-first handle semantics, the legacy `did`/`handle` aliases, and missing-identity hydration, then reconstruct records last-writer-wins from checksum-verified, DID-bloom-pruned sealed JSS plus rotation-safe active/pending and bootstrap-live snapshots. The resulting canonical MST is compared with the PDS commit root through real Sync 1.1 `getLatestCommit`/`getBlocks`; the page renders the same match fields and error state as upstream. `HEAD` performs lookup without verification; successful status responses are `no-store` and carry a generation timestamp. Expensive verification uses upstream's exact 4/source-IP/minute fixed-window limiter, 4,096-entry bound, stale pruning, oldest eviction, and explicit disable option. The offline receipt uses a real HTTP Stream route, real RocksDB row, real JSS archive, signed commit, and loopback PLC/PDS; four separate connections verify successfully, a fifth changing only its ephemeral port is blocked without another upstream request, HEAD remains side-effect-free, and disabling the limiter performs a new authoritative verification. | | Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | **closed.** The exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract. Canonical build, live-firehose, archive, verifier, subscribe, backfill, timestamp-import, and compaction families all have same-semantics producers. Compaction measures pass success/failure/duration, early triggers, live tombstone size, folded reasons, worker examination/rewrite/clean outcomes, dropped-row reasons, bytes, manifest heals, durable watermark, and witnessed-time lag at the corresponding real boundaries. Real JSS tests cover bloom avoidance, bounded concurrent rewrites, malformed account failure, reason accounting, manifest-before-watermark ordering, and durable eviction. The ReleaseSafe production receipt reports `required=104 present=104 missing=0`; the runtime-only dashboard row remains an intentional checksum-pinned Zig/process adaptation with no synthetic `go_*` aliases. See `grafana-dashboard.md`. | diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md index 3537f45..b3ed46e 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -31,6 +31,7 @@ driver boots real server against simulator, walks lifecycle gated on durable-app - invariants: fsync data before committing cursor; seq assigned at append under writer mutex, starts at 1 (0 = sentinel); per-DID order preserved; never crash on upstream data / crash loud on own corruption - readable log (hot tail): writer-owned deque, deep-copied entries at seq allocation, 256MiB budget, evicts only below durable watermark (pinned above), notify channel per append, encode-once wire memo per entry shared across fan-out; cursor below floor → cold reader over sealed segments. Stream's tail enforces the same pin: an undurable suffix is retained even when it overruns the byte budget, `publishDurable` releases only the newly durable byte suffix before eviction, and scrape-time readable/pinned/overrun gauges are O(1) accumulators rather than a scan under the hot lock. Its physical deque retires prefix entries in O(1) and compacts only after a large amortized prefix; the 20,000-row burst regression catches the former `orderedRemove(0)` implementation that made a real 8,192-row repair quadratic and stalled global ingest. - resync memory lifetime: upstream appends each converted replacement and immediately forwards it to the bounded readable log; it does not retain a second whole-repository wire representation. Stream batches physical archive writes in groups of 1,024 but releases the complete v1/v2 encoder arena after each row is copied into the tail. The offline 4,096-record MST receipt is capped at 1 MiB of repair scratch and mutation-checks the former repair-lifetime JSON ownership by reproducing its exact `ArchiveAppendFailed`. +- resync capacity envelope: a pinned-simulator ReleaseSafe receipt seeds 96 signed repositories, establishes real verifier chain state, creates 96 genuine chain breaks, and holds only the real getRepo requests at the handler boundary. It observes exactly 32 active workers plus 64 queued jobs, releases 96 complete CAR/MST repairs, and requires the physical JSS archive to converge to simulator ground truth. With 128-record repos the receipt measured 54,886,400 bytes RSS at saturation and a 72,695,808-byte peak; repository size, not hidden scheduler work, determines the production multiplier. - archive health: free space is collected from the filesystem containing the open archive directory on every scrape; segment-seal latency starts before the pending-block flush and is observed only after footer write/fsync plus finalized-header pwrite/fsync succeeds, using upstream's `0.01 × 2^n` bucket boundaries. - block integrity: the JSS xxh3 authenticates only header/footer metadata, while upstream's zstd decoder independently verifies each frame's content checksum. Stream uses its pinned vendored libzstd for the same check; flipping only the final checksum byte of an upstream-produced block is the mutation receipt. diff --git a/tests/differential_oracle.py b/tests/differential_oracle.py index b0e8912..ef52f4d 100644 --- a/tests/differential_oracle.py +++ b/tests/differential_oracle.py @@ -26,6 +26,7 @@ UPSTREAM = Path( SOURCE = ROOT / "tests/upstream_oracle/stream_differential_test.go" FAULT_SOURCE = ROOT / "tests/upstream_oracle/stream_fault_differential_test.go" RESTART_SOURCE = ROOT / "tests/upstream_oracle/stream_restart_predicate_test.go" +RESYNC_CAPACITY_SOURCE = ROOT / "tests/upstream_oracle/stream_resync_capacity_test.go" STREAM_BIN = (ROOT / "zig-out/bin/stream").resolve() @@ -70,7 +71,7 @@ def main() -> None: go = go_binary() run_pattern = os.environ.get( "STREAM_ORACLE_RUN", - "^TestStream(Differential|ReplayFault|FrameFault|Adversarial)Oracle$", + "^TestStream(Differential|ReplayFault|FrameFault|Adversarial|ResyncCapacity)Oracle$", ) # -overlay can add a test file to an existing Go package. Compiling it as @@ -79,6 +80,7 @@ def main() -> None: virtual = UPSTREAM / "internal/oracle/stream_differential_test.go" fault_virtual = UPSTREAM / "internal/oracle/stream_fault_differential_test.go" restart_virtual = UPSTREAM / "internal/oracle/stream_restart_predicate_test.go" + resync_capacity_virtual = UPSTREAM / "internal/oracle/stream_resync_capacity_test.go" with tempfile.TemporaryDirectory(prefix="stream-differential-oracle-") as tmp: overlay = Path(tmp) / "overlay.json" overlay.write_text( @@ -88,6 +90,7 @@ def main() -> None: str(virtual): str(SOURCE.resolve()), str(fault_virtual): str(FAULT_SOURCE.resolve()), str(restart_virtual): str(RESTART_SOURCE.resolve()), + str(resync_capacity_virtual): str(RESYNC_CAPACITY_SOURCE.resolve()), } } ), diff --git a/tests/upstream_oracle/stream_resync_capacity_test.go b/tests/upstream_oracle/stream_resync_capacity_test.go new file mode 100644 index 0000000..aace968 --- /dev/null +++ b/tests/upstream_oracle/stream_resync_capacity_test.go @@ -0,0 +1,219 @@ +package oracle + +// This file is compiled into the pinned upstream internal/oracle package by +// tests/differential_oracle.py. It drives Stream through the real upstream +// simulator and holds real getRepo requests at the handler boundary long +// enough to make Atmos's 32-worker / 64-queued-job envelope observable. + +import ( + "context" + "log/slog" + "math/rand/v2" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/bluesky-social/jetstream/internal/simulator/fanout" + simhttp "github.com/bluesky-social/jetstream/internal/simulator/http" + "github.com/bluesky-social/jetstream/internal/simulator/world" + "github.com/stretchr/testify/require" +) + +type resyncGetRepoGate struct { + enabled atomic.Bool + active atomic.Int64 + maximum atomic.Int64 + total atomic.Int64 + release chan struct{} + once sync.Once +} + +func newResyncGetRepoGate() *resyncGetRepoGate { + return &resyncGetRepoGate{release: make(chan struct{})} +} + +func (g *resyncGetRepoGate) open() { g.once.Do(func() { close(g.release) }) } + +func (g *resyncGetRepoGate) wrap(next http.Handler) http.Handler { + return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) { + if g.enabled.Load() && r.URL.Path == "/xrpc/com.atproto.sync.getRepo" { + active := g.active.Add(1) + g.total.Add(1) + for { + seen := g.maximum.Load() + if active <= seen || g.maximum.CompareAndSwap(seen, active) { + break + } + } + defer g.active.Add(-1) + select { + case <-g.release: + case <-r.Context().Done(): + return + } + } + next.ServeHTTP(rw, r) + }) +} + +func TestStreamResyncCapacityOracle(t *testing.T) { + require.Equal(t, streamDifferentialPin, os.Getenv("STREAM_ORACLE_EXPECTED_PIN")) + streamBin, err := filepath.Abs(os.Getenv("STREAM_ORACLE_BIN")) + require.NoError(t, err) + require.FileExists(t, streamBin) + + const accounts = 96 // 32 active workers + the exact 64-job queue. + simCfg := world.DefaultConfig() + simCfg.DataDir = filepath.Join(t.TempDir(), "simulator") + simCfg.Seed = 0x32_64_2048 + simCfg.Accounts = accounts + simCfg.InitialRecords = 128 + simCfg.CommitsPerSec = 1 + simCfg.FirehoseHistory = 4096 + + 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(8192) + require.NoError(t, w.AttachRuntime( + rand.New(rand.NewPCG(simCfg.Seed^0xfeedf00d, simCfg.Seed^0xc0ffee)), + fan, + )) + + gate := newResyncGetRepoGate() + simSrv := httptest.NewServer(nil) + simSrv.Config.Handler = gate.wrap(simhttp.NewHandler(w, simSrv.URL)) + defer func() { + // Release held handlers before Close waits for them. This also keeps a + // failed anti-vacuity assertion from hanging the test process. + gate.open() + simSrv.Close() + }() + + port := freeStreamOraclePort(t) + baseURL := "http://127.0.0.1:" + strconv.Itoa(port) + dataDir := filepath.Join(t.TempDir(), "stream") + logs := &synchronizedBuffer{} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + 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, + "--max-segment-bytes=4194304", + "--compaction-interval=0", + "--retry-interval=0", + ) + cmd.Stdout = logs + cmd.Stderr = logs + require.NoError(t, cmd.Start()) + stopped := false + defer func() { + if stopped { + return + } + _ = cmd.Process.Signal(syscall.SIGTERM) + waitDone := make(chan struct{}) + go func() { + _ = cmd.Wait() + close(waitDone) + }() + select { + case <-waitDone: + case <-time.After(10 * time.Second): + _ = cmd.Process.Kill() + <-waitDone + } + }() + + waitForStreamOracleServing(t, cmd, baseURL, logs) + _, err = w.GenerateIdentityForTest(t.Context(), 0, false) + require.NoError(t, err) + waitForStreamPipelineDrain(t, baseURL, w.CurrentSeq(), logs) + + // First sighting is accepted by both implementations and seeds durable + // chain state. The next commit silently mutates its predecessor, forcing + // one authentic whole-repo repair for every distinct DID. + for i := range accounts { + _, _, err = w.GenerateRecordOpForTest(t.Context(), i, "create", "app.bsky.feed.post", "capacity-baseline") + require.NoError(t, err) + } + waitForStreamPipelineDrain(t, baseURL, w.CurrentSeq(), logs) + + gate.enabled.Store(true) + for i := range accounts { + _, err = w.GenerateSilentMutationThenCommitForTest(t.Context(), i) + require.NoError(t, err) + } + + var saturated string + deadline := time.Now().Add(45 * time.Second) + for time.Now().Before(deadline) { + saturated = scrapeStreamOracleMetrics(t, baseURL) + active, activeOK := metricValue(saturated, "stream_resync_workers_active") + queued, queueOK := metricValue(saturated, "stream_resync_queue_depth") + if activeOK && queueOK && active == 32 && queued == 64 && gate.active.Load() == 32 { + break + } + time.Sleep(20 * time.Millisecond) + } + requireStreamMetric(t, saturated, "stream_resync_workers_active", 32) + requireStreamMetric(t, saturated, "stream_resync_queue_depth", 64) + require.Equal(t, int64(32), gate.active.Load(), "all real fetch workers must reach getRepo") + require.Equal(t, int64(32), gate.maximum.Load()) + rssAtSaturation, ok := metricValue(saturated, "process_resident_memory_bytes") + require.True(t, ok) + + gate.open() + var drained string + var peakRSS = rssAtSaturation + deadline = time.Now().Add(90 * time.Second) + for time.Now().Before(deadline) { + drained = scrapeStreamOracleMetrics(t, baseURL) + if rss, ok := metricValue(drained, "process_resident_memory_bytes"); ok && rss > peakRSS { + peakRSS = rss + } + completed, completedOK := metricValue(drained, `stream_resync_total{result="completed"}`) + active, activeOK := metricValue(drained, "stream_resync_workers_active") + queued, queueOK := metricValue(drained, "stream_resync_queue_depth") + if completedOK && activeOK && queueOK && completed == accounts && active == 0 && queued == 0 { + break + } + time.Sleep(20 * time.Millisecond) + } + requireStreamMetric(t, drained, `stream_resync_total{result="completed"}`, accounts) + requireStreamMetric(t, drained, `stream_resync_total{result="failed"}`, 0) + requireStreamMetric(t, drained, "stream_resync_workers_active", 0) + requireStreamMetric(t, drained, "stream_resync_queue_depth", 0) + require.Equal(t, int64(accounts), gate.total.Load(), "every repair must fetch a real CAR") + + require.NoError(t, cmd.Process.Signal(syscall.SIGTERM)) + require.NoErrorf(t, cmd.Wait(), "Stream did not shut down cleanly:\n%s", logs.String()) + stopped = true + + events, err := ObserveSegments(dataDir) + require.NoError(t, err) + require.NoError(t, CheckInvariants(events)) + ground, err := GroundTruthFromWorld(w) + require.NoError(t, err) + reconstructed, err := Reconstruct(EventsSortedBySeq(events)) + require.NoError(t, err) + require.NoError(t, Compare(ground, reconstructed), + "all 96 saturated repairs must converge to the simulator MST") + t.Logf("resync capacity receipt: workers=32 queue=64 real_getRepo=%d saturation_rss=%.0f peak_rss=%.0f", + gate.total.Load(), rssAtSaturation, peakRSS) +}