From e0dcb2705af585a2b985fb2cdac0a8f3c7d8a96a Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Fri, 24 Jul 2026 01:59:47 -0500 Subject: [PATCH] fix bootstrap completion durability --- README.md | 98 +++-- deploy/README.md | 37 +- docs/bootstrap-semantic-parity.md | 141 +++++-- docs/configuration-parity.md | 117 ++++-- docs/grafana-dashboard.md | 50 +-- docs/semantic-parity.md | 191 ++++++---- docs/upstream-bootstrap-spec.md | 70 ++-- src/internal/bootstrap/engine.zig | 597 +++++++++++++++++++++++++----- 8 files changed, 932 insertions(+), 369 deletions(-) diff --git a/README.md b/README.md index ca6993f..7922340 100644 --- a/README.md +++ b/README.md @@ -2,46 +2,55 @@ a zig implementation of [jetstream](https://github.com/bluesky-social/jetstream) — the full-network archive and streaming service for atproto. built on [zat](https://tangled.org/zat.dev/zat). -The former live-only `stream.waow.tech` canary has been retired. No public -Stream deployment is currently promoted; the next deployment is a timeboxed, -separate-infrastructure full-network experiment built from the admitted -`4d163e8` candidate. +The former live-only `stream.waow.tech` canary and the July 2026 isolated +experiment have been retired. No public Stream deployment is currently +promoted, and no current artifact is admitted for another whole-network +experiment. ## capabilities - live Jetstream feeds at `/subscribe` and `/subscribe-v2`, with collection and DID filters, cursor replay, runtime option updates, RFC 7692 permessage-deflate on v1, frozen v1 zstd, and fetchable - dictionary-negotiated v2 zstd + dictionary-negotiated v2 zstd; cold replay continuity and remote-data + encoding isolation are not yet admitted - a disk-backed historical archive exposed through `listSegments`, `getSegment`, `getBlock`, and `planBackfill`; Stream's sealed `jss` v1 segments are byte-compatible with upstream Jetstream. Sealed headers, block envelopes, DID blooms, and collection summaries are resident and refresh atomically after seal, compaction, and timestamp rewrite; planning opens no segment - files, while cold replay reads only selected compressed blocks -- full-network bootstrap with concurrent live capture, deterministic merge, - and crash-safe cutover into steady-state ingestion + files, while cold replay reads only selected compressed blocks. Internal + planner errors and unreadable replay segments still have fail-open paths +- a whole-network bootstrap implementation with concurrent live capture and + deterministic merge. Repository completion is coupled to ordinary archive + writer durability independently of the 100,000-entry dispatch/checkpoint + batch; several lifecycle, discovery, and recovery paths remain blocked by + the deployment-gate audit - Sync 1.1 commit verification, including PLC key rotation, MST inversion, op-CID checks, replay protection, durable repository-chain state, and transparent authenticated whole-repo repair with ordered pending replay; - the live scheduler matches Atmos's 32-worker per-DID FIFO, bounded - drop-oldest, completion-order batching, and safe cursor watermark behavior -- durable metadata and archive publication through RocksDB transactions, - coupled to segment fsync boundaries + the live scheduler implements Atmos-shaped 32-worker per-DID FIFO, bounded + drop-oldest, and completion-order batching. Reconnect and cursor-recovery + behavior remain deployment blockers +- RocksDB-backed metadata and archive publication coupled to segment fsync + boundaries; several correctness-metadata reads still fail open and are + called out in the deployment gate - tombstone-aware compaction and continuing repair of repositories that were - unavailable or changed during bootstrap + unavailable or changed during bootstrap; compaction watermark corruption + and retry-supervisor failure handling are not yet equivalent - crash-resumable timestamp imports with a durable RocksDB rule map, topology-preserving sealed-segment repair, and bearer-gated operator XRPC - browser and terminal operations views at `/status`, including durable PDS diagnostics at `/status?tab=hosts` and per-repository reconstruction and authoritative root verification at `/status?tab=accounts`, plus Prometheus - metrics at `/metrics` on the optional debug listener - -The implementation and configuration parity matrix is closed against pinned -Jetstream V2 commit `f29815c`; that is not a claim that a full-network -deployment has completed. The immutable admission receipt and remaining -operational experiment boundary live in -[`docs/semantic-parity.md`](docs/semantic-parity.md) and + metrics at `/metrics` on the optional debug listener. The upstream + collections/segments status views and several lifecycle fields are absent + +The implementation is **not** at semantic parity with pinned Jetstream V2 +commit `f29815c`. The current evidence, blockers, and definition of a valid +admission are maintained in +[`docs/semantic-parity.md`](docs/semantic-parity.md). The July experiment is +historical evidence, not an admission receipt; see [`docs/full-network-experiment-2026-07.md`](docs/full-network-experiment-2026-07.md). ## develop @@ -103,10 +112,12 @@ process before command execution, catching misspelled production settings; Kubernetes's `JETSTREAM_APP_*` service variables and upstream's `JETSTREAM_ORACLE_*` and `JETSTREAM_SIM_*` test namespaces remain intentionally exempt. `inspect-all` inherits `JETSTREAM_DATA_DIR`, and root logging variables remain persistent across every -command. `just environment-contract` proves the complete source map, sorted -unknown rejection, typed validation, CLI precedence, inspection, listener, -storage, logging, and timestamp-import authentication through real ReleaseSafe -processes without external network access. +command. `just environment-contract` proves the hand-maintained Stream map, +sorted unknown rejection, typed validation, CLI precedence, inspection, +listener, storage, logging, and timestamp-import authentication through real +ReleaseSafe processes without external network access. It does not prove that +the map contains every flag in the pinned Go source; the deployment gate now +requires a generated inventory comparison. The canonical server command also matches Jetstream V2's listener topology: @@ -209,14 +220,16 @@ Zero disables non-empty DID or collection filters for the corresponding filter cap. For `--plan-max-entries`, zero instead means one unbounded page; the threshold must be greater than zero and at most one. -Subscriber replay matches upstream's 36-hour default lookback. A v1 sequence +Subscriber replay implements upstream's 36-hour default lookback controls. A v1 sequence cursor older than the retained window is conservatively clamped to the first eligible sealed segment; v2 rejects the same cursor before WebSocket upgrade with a `cursor too old` response. Timestamp cursors are clamped for both protocols. Operators may change the window with a Go-style duration such as `--cursor-lookback=72h`; zero makes both endpoints pure-live. The offline `just cursor-lookback-contract` receipt exercises all three modes against real -sealed JSS through production Stream processes. +sealed JSS through production Stream processes. Cold replay completeness is a +separate blocked row because an unreadable manifest-listed segment is currently +skipped. Delete compaction exposes upstream's resource controls: @@ -232,7 +245,8 @@ upstream's unlimited sentinel. Rewrite workers bound concurrent sealed-segment rewrites; zero selects `min(CPU count, 8)`. Cap triggers wake the compactor immediately, subject to upstream's 30-second between-pass floor. `just compaction-config-contract` proves the controls reach the physical -compactor rather than stopping at argument parsing, including a 250 ms pass. +compactor rather than stopping at argument parsing, including a 250 ms pass. It +does not waive the fail-open compaction-watermark read. Failed-repository healing exposes upstream's complete control set: @@ -249,7 +263,8 @@ interval disables the background loop. Zero worker values select upstream's Transient failures use upstream's capped exponential delay plus uniform `[0, delay/2)` jitter. `just retry-config-contract` holds real `getRepo` requests to measure both admission limits and then reopens RocksDB to verify -the durable retry schedule. +the durable retry schedule. The retry supervisor and candidate-scan design +remain blocked independently of those controls. The strict power-loss tier is deliberately separate from ordinary unit and process tests. It cross-builds the production ReleaseSafe Linux binary, runs @@ -280,8 +295,10 @@ Production bootstrap controls: Worker count controls concurrent network downloads. Batch size counts every `listRepos` entry, including inactive and already-complete repositories; Stream accumulates whole 1,000-entry protocol pages until it reaches the target, -then shuffles eligible repositories across PDS hosts and dispatches them. Zero -selects the 100,000-entry production default. The byte flag bounds the owned +then shuffles eligible repositories before dispatch. Requests still go through +the relay and follow its redirect, so this does not claim precise pre-request +PDS scheduling. Zero selects the 100,000-entry production default. The byte +flag bounds the owned complete CAR plus preparation/index state. Preparation is serialized, copies the downloaded CAR into that budget, and releases the scratch mmap before emission. Already prepared repositories emit concurrently from the owned CAR @@ -299,8 +316,15 @@ owns `/backfill/repo-scratch`; steady live and failed-repo healing use lifecycle state after cutover. Resource exhaustion is lifecycle-fatal, never persisted as a fake "large repo skipped" success/failure. Backfill progress, active workers, queue depth, current -transient bytes, and peak transient bytes are exported on `/metrics`. The -failed-repo loop uses 16 global workers and four workers per final +transient bytes, and peak transient bytes are exported on `/metrics`. +Successful CARs record their final emitted archive sequence; ordinary writer +durability commits every covered repository independently. The dispatch batch +still controls listRepos cursor-checkpoint cadence, but a slow sibling no +longer withholds already-durable repository completion. The authoritative +lifecycle analysis is +[`docs/bootstrap-semantic-parity.md`](docs/bootstrap-semantic-parity.md). + +The failed-repo loop uses 16 global workers and four workers per final post-redirect PDS host; attempts, error class/message, final host, retry count, and next attempt are durable in RocksDB. Every repo transition also updates a persistent `host/` diagnostic aggregate in the same synced batch: @@ -328,6 +352,10 @@ record/byte counters. Matching upstream, the debug-only explicit requested DIDs serially, resolves their real DID documents, and removes stale full-network merge-discovery state. Ordinary whole-network `listRepos` bootstrap does not add millions of extra identity requests. +Normal post-bootstrap discovery is not yet admitted: inactive unknown +repositories are skipped and an oversized cursor can silently terminate the +scan. + `handle/` is maintained atomically with every repo transition, and each merge-source commit advances its cursor together with latest-revision refreshes. Account lookup resolves handles through the configured identity @@ -370,5 +398,7 @@ fail-closed metric contract. The dashboard and process receipts are fully offline: run `just dashboard-test`, `just process-metrics-contract`, `just http-metrics-contract`, then `just dashboard-contract http://127.0.0.1:6008/metrics` against a local -candidate. `just process-metrics-contract` also runs that complete dashboard -family check against the ReleaseSafe binary it launches. +candidate. `just process-metrics-contract` also runs that dashboard +family-presence check against the ReleaseSafe binary it launches. Family +presence does not establish producer semantics, restart stability, or parity; +the dashboard audit in `docs/semantic-parity.md` is the deployment gate. diff --git a/deploy/README.md b/deploy/README.md index 78a3a9a..25ba8ca 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -1,20 +1,17 @@ # deploying stream -The retired live-only canary no longer runs on the Indigo relay node. The next -target is a timeboxed full-network experiment on an isolated 16-dedicated-core, -64 GB host with a separate 2.5 TB XFS volume. Its cost gate, destruction -watchdog, Compose bundle, Prometheus, and Grafana wiring live in the sibling -`zzstoatzz.io/relay` operations repository under `stream/experiment/`. +The retired live-only canary no longer runs on the Indigo relay node. The July +2026 isolated experiment has also been deleted. No replacement experiment +should be provisioned until the blockers in `../docs/semantic-parity.md` are +closed and an exact artifact passes the admission procedure defined there. The long-term production target remains a separate host with redundant local storage; passing the Cloud-Volume experiment is evidence for sizing and throughput, not approval to colocate Stream with Indigo. -The current experiment admits application commit `e1926f3`. Its clean-clone -linux/amd64 image is published with OCI index digest -`atcr.io/zat.dev/stream@sha256:adc276257fd8bc0cd8c0c94341afe220b50690cbef79539fe1ce3f7e6bcbd9fe`. -The timeboxed deployment transfers the commit-tagged image directly and -verifies that exact index identity after loading it on the workload host. +The previously deployed `e1926f3` image is historical evidence only. It was +not the artifact covered by the earlier admission receipt and it failed during +the experiment. Do not reuse its digest as a promotion candidate. ```sh just package # ReleaseSafe x86_64-linux-gnu binary in zig-out/bin/ @@ -70,14 +67,13 @@ limit equal to the backfill budget. ## promotion gate -- all eight lifecycle crashpoints pass with the parallel crawler; -- bootstrap, compaction, retry, and live capture run together; -- a consistent off-machine backup has restored into a private deployment; -- the official Go V2 client and the public conformance harness replay that - restore without a gap; -- disk, RAID, backup-age, backfill-stall, and memory-budget alerts are live. - -Absolute throughput still needs the dedicated-node soak before public cutover. +The complete gate is `../docs/semantic-parity.md`. In particular, a promotion +must test individual repository durability inside a large dispatch batch, +metadata read corruption, unreadable archive files, persistent live encoding +failures, and transient firehose disconnects. The receipt must bind those +results to the exact deployed image digest. Throughput, backup restore, client +interoperability, and alerts remain necessary, but they cannot waive a blocked +semantic row. ## dashboard gate @@ -90,5 +86,6 @@ steady state; every upstream protocol/lifecycle panel remains unchanged. Run `just dashboard-test`, then `just process-metrics-contract`, then `just dashboard-contract `. These checks are -network-independent when given a saved scrape or the loopback harness and -refuse promotion while any queried family is absent. +network-independent when given a saved scrape or the loopback harness and fail +when a queried family is absent. They verify dashboard structure and metric +presence only; they are not a promotion decision. diff --git a/docs/bootstrap-semantic-parity.md b/docs/bootstrap-semantic-parity.md index 6c086bb..10254ff 100644 --- a/docs/bootstrap-semantic-parity.md +++ b/docs/bootstrap-semantic-parity.md @@ -1,34 +1,113 @@ -# Jetstream V2 bootstrap semantic parity gate +# Jetstream V2 bootstrap and lifecycle audit -No full-network deployment is allowed while any row in this gate is open. -The reference is `bluesky-social/jetstream` commit -`f29815c391fc2644f8a3dd36b899fb3697dd1ea6` plus Atmos v0.2.14, not the -shape of Stream's API or a small simulator happy path. Recheck the pin before -every experiment. +Reference: -| Behavior | Upstream contract | Stream evidence | Gate | +- upstream Jetstream `f29815c391fc2644f8a3dd36b899fb3697dd1ea6` +- Atmos `v0.2.14` +- Stream `b709fe32c72e9eee4725f92e7999eb6533bb3c8f` + +The status words have the meanings defined in +[semantic-parity.md](semantic-parity.md). A configuration knob being wired +correctly does not verify the behavior it controls. + +## Current decision + +**Bootstrap, merge, and retry remain blocked.** The dispatch-batch completion +defect observed in the experiment has been replaced with the upstream +writer-durability-gated per-repository model and focused restart proof. The +source audit also found independent fail-open lifecycle, cursor, discovery, +and retry behavior; those still prevent another deployment. + +## Initial enumeration and download + +| Invariant | Status | Audited evidence | Gap or required proof | +|---|---|---|---| +| listRepos page size and accumulation | **verified** | Stream requests protocol-sized pages, counts every entry toward a page-aligned configurable batch, and defaults the batch target to 100,000. The producer ledger tests cover page ordering and the zero/default value. | This says nothing about completion durability inside the batch. | +| Eligible-repository shuffle | **verified** | The accumulated eligible jobs are shuffled immediately before dispatch. Deterministic helper tests establish that the production helper changes order. | Statistical PDS distribution is an operational measurement, not part of this narrow row. | +| Initial worker admission | **verified** | Positive worker counts become physical concurrent getRepo requests; zero/omitted select 100. Held-request production tests observed 7 and 100 active requests. | Higher concurrency is not automatically faster or safe; the 200-worker experiment regressed. | +| Request path through relay redirects | **verified** | Stream requests getRepo from the configured relay and follows redirects, retaining the final host for diagnostics. This matches pinned Atmos. | Neither implementation knows the redirected PDS before the request, so redirected requests receive reactive rather than proactive per-host rate limiting. This is a shared limitation, not a Stream parity gap. | +| Repository-size exclusion | **verified** | No hard CAR-size rejection remains. The response streams to data-volume scratch and a repository may consume the configured preparation budget. | Capacity exhaustion remains fatal by design and must be sized operationally. | +| Complete-CAR validation | **verified** | Preparation parses the CAR, verifies block CIDs, loads the MST, and walks referenced nodes/records before row emission. Truncation fixtures fail before rows are emitted. | The claim is limited to the tested truncation/missing-block shapes. | +| Authoritative DID | **verified** | Emitted rows use the listRepos/selected DID even when the embedded commit DID differs. A hostile fixture exercises this. | None known for this invariant. | +| Attempt deadline boundary | **verified** | One cancellable operation covers response streaming and CAR preparation; handler emission and durable writes are outside the five-minute network deadline. A deadline failure is terminal for that run. | None known for this invariant. | +| Ordinary retry budget | **verified** | The initial attempt plus three ordinary transient retries, exponential base, jitter, and 30-second cap have mixed 503/429 fixtures. | The production firehose reconnect defect is separate from getRepo retry. | +| Rate-limit retry budget | **verified** | 429 uses a separate twenty-retry budget and honors parsed server reset information with the bootstrap ceiling. Tests show 429 does not consume the ordinary budget. | Pre-request per-PDS scheduling is unavailable after relay indirection, as it is upstream. | +| Terminal repository errors | **verified** | Structured RepoNotFound becomes complete with no rows; deactivated/suspended/takendown become unavailable. Focused tests inspect the durable status. | None known for the classified responses. | +| Inactive repository during initial discovery | **verified** | A new inactive entry is stored as not_started/active=false and is not dispatched; later active flips preserve the row. | Post-bootstrap discovery does not preserve the same rule; see below. | +| Explicit selected backfill | **partial** | The selected DID list bypasses listRepos, preserves caller order, conflicts with max-repos, clears stale discovery state, resolves identity metadata, and maintains the handle index in focused tests. | The exact current artifact has not rerun the full selected-repo receipt, and selected completion still uses the blocked batch completion machinery. | +| Max-repos debug mode | **partial** | The flag truncates selected work and avoids committing a resumable whole-network cursor. | Its recovery behavior remains coupled to blocked phase/cursor reads. It is not evidence for whole-network behavior. | + +## Repository emission and durable completion + +| Invariant | Status | Audited evidence | Gap or required proof | +|---|---|---|---| +| Record path validation and drops | **verified** | Bad NSID/rkey/field-width records are dropped individually; other valid records from the repository survive. Counters and physical rows are asserted. | None known for the tested shapes. | +| Repository revision watermark | **verified** | Backfill rows receive the repository head revision and successful completion stores the immutable backfill revision plus latest revision. | Durability timing of that completion is blocked below. | +| 1,024-row handler emission | **verified** | Prepared repositories emit bounded row batches and tests exercise multi-batch repositories. | This is an emission/memory boundary, not a repository completion boundary. | +| Async compression ordering | **partial** | Four workers by default prepare real blocks concurrently; commit order is serialized; sync and async fixtures produce byte-equivalent JSS. | Cancellation/OOM behavior at every queued/compressed/commit state has not been compared against upstream in this audit. | +| Archive before repository completion | **verified** | Each successful CAR records its final emitted archive sequence. The archive durability hook stages only completions whose final sequence is covered; a forced drain fails rather than advancing past an uncovered completion. Focused tests inspect the physical boundary and terminal state. | The claim is limited to bootstrap repository completion; other metadata boundaries have their own rows. | +| Per-repository completion durability | **verified** | Stream now ports upstream's watermark queue: data-bearing and genuinely empty successful CARs queue independently, duplicate DIDs replace in place, RepoNotFound follows upstream's immediate terminal OnFail path, and captured completion time survives deferred durability. | Re-run the process-level oracle before artifact admission; the production mechanism and focused boundary are established here. | +| Slow or hung straggler isolation | **verified** | A production-boundary test leaves one repository without a worker result, crosses a writer durability boundary, proves its completed sibling durable while the listRepos cursor is absent, fully reopens RocksDB, and proves only the interrupted repository moves to pending. | The admission oracle must scale this invariant beyond a two-repository fixture and inject an actual process kill. | +| Completion metrics | **verified** | Queue count/depth change at queue/replacement; durable batch/repo counters change only after the synced repo-state batch; queue wait uses the captured completion time; RepoNotFound bypasses those queue metrics as upstream does. | Prometheus counters are process-local by definition. Restart-stable progress comes from durable repository states, not counter continuity. | +| Resource exhaustion classification | **partial** | Preparation allocation failure is treated as lifecycle-fatal rather than a durable “repo too large” result. | Fault coverage does not establish cleanup and restart behavior at each preparation/emission ownership transfer. Archive active-tail recovery also mishandles iterator allocation failure. | + +## Cursors and lifecycle transitions + +| Invariant | Status | Audited evidence | Gap or required proof | |---|---|---|---| -| Repository size | No CAR-size exclusion; stream the response | `fetched_car.zig` streams to data-volume scratch; serialized preparation copies the complete CAR into the configurable transient budget, proves completeness, and releases the mmap before concurrent row emission. The former 512 MiB branch and `RepoTooLarge` are gone; resource exhaustion remains lifecycle-fatal. | closed | -| Complete repository | Reject missing MST nodes and referenced records before handler emission | `prepareRepo` performs a full preflight walk; missing nodes/records are retryable; truncation test proves zero emitted rows | closed | -| Attempt deadline | Five minutes across response streaming and CAR completeness; a deadline failure is not retried inside the same bootstrap run | `prepared_fetch.zig` owns one cancellable future containing fetch + `prepareRepo`; row emission/fsync/RocksDB remain outside | closed | -| Ordinary retry | Three retries/four attempts, 1s exponential base, 30s ceiling | independent transient counter in `downloadOne`; redirected HTTP fixture proves the exact attempt count across 503 → 429 → 503 → terminal success | closed | -| Rate-limit retry | Twenty additional attempts; honor reset/Retry-After independently | independent counter; `RateLimit-Reset` precedence; 330s bootstrap clamp; the mixed redirected fixture proves a 429 does not consume the ordinary transient budget; steady retry persists/clamps the server deadline | closed | -| Terminal errors | RepoNotFound → complete/no rows; deactivated/suspended/takendown → unavailable | exact structured-error classification tests and durable statuses | closed | -| DID attribution | listRepos/handler DID is authoritative, never embedded commit DID | a complete, valid, signed CAR whose embedded DID differs from the listRepos DID emits every row under the listRepos DID | closed | -| Inactive discovery | persist not_started + active=false; do not dispatch; preserve status on active flips | RocksDB row encoding v5, backward v2/v3/v4 decode, restart test | closed | -| Explicit selected backfill and account identity metadata | `--backfill-repos` processes the caller's order serially, bypasses listRepos, cannot combine with max-repos, deletes stale merge-discovery state, resolves each selected DID even when its repo row is already complete, retains declared handle/PDS, and atomically maintains the normalized local handle index. Ordinary whole-network backfill does not perform this extra resolution. | Stream parses and deduplicates the explicit DID list, synthesizes one ordered selected page without touching listRepos, requires the configured real `DidResolver`, deletes the stale bootstrap discovery watermark, and never advances listRepos state. `repo/` v5 persists Handle/PDS with same-batch `handle/` transitions. A real loopback PLC + getRepo run fails on any listRepos request and proves terminal status plus index persistence; handle moves/collisions, rebuild, and restart reads are also exercised offline. | closed | -| Merge latest revision | For every kept rev-bearing row, refresh top-level Rev/UpdatedAt while preserving Backfill.Rev; commit those rows atomically with the per-source merge cursor | merge collects the last kept rev per DID and `commitMergeSource` stages cursor + repo refreshes in one synced RocksDB batch; missing repo rows are skipped rather than synthesized malformed | closed | -| Backfill dispatch batching | Accumulate whole 1,000-entry listRepos pages until at least 100,000 entries by default, counting inactive and complete rows; shuffle eligible repos; dispatch and drain before checkpoint | configurable and zero/default batch sizes drive the real pinned-simulator request ledger; a deterministic shuffle receipt exercises the production helper; full differential and predicate-kill restart tiers converge | closed | -| Backfill worker admission | Positive worker counts are physical network concurrency; zero and omitted values select Jetstream's 100-worker production default | held real simulator getRepo requests observe exact peaks of 7/100/100 across three ReleaseSafe processes, then all 104 repos complete; disk-backed fetch remains concurrent, preparation is cancelably serialized under the byte bound, and prepared repositories emit concurrently from owned CAR bytes after the source mmap is released | closed | -| Backfill async flush | Initial handlers append in 1,024-row batches; four bootstrap-only compression workers by default; zero disables; detach blocks, compress concurrently, commit and fsync in preparation order before handler completion | initial emission uses the upstream batch boundary; real 8 MiB columnar blocks reach exactly four simultaneous zstd workers; an independent archive receipt proves async bytes equal synchronous JSS bytes, all eight blocks remain sequence-contiguous, and the committed watermark reaches the final block | closed | -| Cursor durability | advance after every repo in the accumulated batch is durable; resume cursor drains empty; discovery cursor retains the last non-empty page cursor | the distinct relay/discovery keys commit atomically at the batch boundary; cursor unit receipt plus real after-repo-complete predicate kills converge after same-disk restart | closed | -| Resource exhaustion | capacity failure is fatal, never a durable per-repo “too large” skip | OOM is lifecycle-fatal and tested | closed | -| Resync ordering | validate replacement first, then append sync + replacement, then durable completion | `prepareRepo` precedes emission; sync plus replacement rows append in bounded 1,024-row batches, matching upstream. Memory does not require a second repository-sized event array; archive durability precedes RocksDB completion | closed | -| Steady retry host semantics | persist actual post-redirect PDS host; 16 global / 4 per-host workers; 429 parks only that host | redirected `Request.uri` is normalized and persisted; worker gates and durable host-specific parks match upstream limits; a real redirected 429 fixture proves the final host replaces the predicted host and only the final host is parked | closed | -| Retry diagnostics | persist attempts, last error/class, host, retry count, next attempt, and durable per-host aggregates | RocksDB row v5 persists every repo field and backward-decodes v2/v3/v4; retry, terminal resync, active-flip, and crash-recovery RMW paths preserve identity/latest fields, while success clears stale repo failure state. The same synced transition batch maintains normalized `host/` current counts plus cumulative error-class counts and five bounded recent samples; host moves remove the repo from the old bucket, successful retry retains diagnostic history, and a one-time migration rebuilds current aggregates without inventing history. The public host view reopens those durable rows through the production binary and exercises both upstream sort modes plus restart, text/HTML, filtering, and escaping offline. | closed | -| Deadline coverage | getRepo streaming read + CAR load/completeness check, excluding handler/durability | same `prepared_fetch.zig` future covers both and returns a prepared repo | closed | -| Oracle/adversity suite | truncation, hanging server, terminal errors, crash/restart, active flips, and power loss at canonical lifecycle boundaries | focused offline fixtures cover truncation, hanging fetches, terminal errors, retry/429 interaction, final-host parking, and hostile embedded DIDs. The shutdown receipt now drives the production ReleaseSafe process into bootstrap with connected live capture and a hung `listRepos`, sends SIGTERM, requires prompt exit without SIGKILL, and reopens the same data directory to repeat the proof. The ordinary crash matrix survived all 8 local lifecycle seams and rejects shutdown OOM/fatal-handler evidence. The strict Linux oracle runs production RocksDB/JSS on ext4 over calibrated NBD storage, kills storage before the process, reconstructs only FLUSH/FUA-acknowledged bytes, and recovered all 8 canonical lifecycle schedules plus the two-crash restart-after-cleanup guard through public archive and V2 replay. | closed; rerun both oracles against the exact candidate before deployment | - -The deleted July experiment is not evidence for closing these rows. A fresh -experiment may be provisioned only after every blocker is closed and the gate -is reviewed against the then-current upstream commit. +| Relay listRepos cursor checkpoint | **verified** | Stream atomically writes the relay cursor and last non-empty discovery cursor only after the dispatch batch drains. Final empty cursor handling has unit coverage, while the straggler/reopen test proves individual repository completion can become durable with the page cursor still absent. | The 100,000-entry unit is checkpoint/re-enumeration cadence, not repository completion durability. | +| Relay firehose cursor read | **blocked** | Durable cursor writes are coupled to archive metadata batches on the normal path. | Read error, malformed width, or a negative value is accepted as “no cursor”/a value. Upstream rejects corruption and negative encoding to avoid skipping the interval to current relay tip. | +| listRepos cursor read | **verified** | `loadRelayListReposCursor` propagates RocksDB errors and distinguishes a missing key. | Valid cursor length remains artificially bounded by fixed buffers during enumeration; initial enumeration fails loud, while merge discovery silently stops. | +| Lifecycle phase read | **blocked** | Phase writes are synced. | RocksDB read failure, malformed bytes, and an unknown phase all return null and are treated as a fresh data directory. Upstream propagates read/decode errors. | +| Bootstrap → merging commit point | **blocked** | Stream eventually persists merging before entering the merge driver. | Upstream writes merging immediately after backfill drains and before stopping bootstrap-live capture. Stream stops/joins/flushed/seals capture inside `runBootstrap`, returns, and only then writes merging. Port the ordering and test both sides of the write. | +| Merging → steady_state commit point | **partial** | Stream drains, compacts, reconciles the manifest, discovers, removes the source tree, syncs the data directory, deletes cursors, publishes seq metadata, then writes steady_state. Named write-side crash tests cover several seams. | Phase read corruption and the fail-open source-existence guard can enter this sequence from the wrong state. | +| Exact persisted phase timing | **blocked** | Metrics record process-local state durations. | Upstream persists/serves phase entry and backfill timing. Stream has no equivalent entered-at state, so status and restart timing cannot be equivalent. | + +## Bootstrap-live capture and merge + +| Invariant | Status | Audited evidence | Gap or required proof | +|---|---|---|---| +| Concurrent bootstrap-live attachment | **partial** | The production lifecycle starts a separate capture archive and crash/oracle fixtures send live events during bootstrap. | The phase commit ordering differs, and the firehose dependency can terminate during reconnect sleep. | +| Bootstrap-live failure propagation | **partial** | The lifecycle now observes an unexpectedly completed capture future and treats it as fatal. | `HttpConnectionClosing` can still escape Zat's reconnect loop and terminate the process; the exact dependency behavior needs a regression test on Linux. | +| Merge source existence guard | **blocked** | A genuinely absent source directory is intended to mean cleanup completed before the previous phase write. | Stream converts every `openDir` error to absence. Upstream distinguishes not-found from permission/I/O failures. Inject EACCES/EIO and require startup failure without cursor deletion or phase advancement. | +| Merge source cursor read | **blocked** | Successful source completion commits cursor and latest-revision updates together. | Read errors and malformed values become source zero, permitting duplicate re-drain. | +| Merge row filtering | **verified** | Rows are dropped only when the repository is complete and the source revision is at/below its backfill watermark; missing lookups are cached. Kept/dropped fixtures exercise this. | The surrounding cursor recovery remains blocked. | +| Latest revision refresh | **partial** | Kept rev-bearing rows update latest revision in the same batch as the source cursor while preserving the backfill revision. | Missing repo rows are skipped; this matches the intended defensive behavior, but corrupt cursor state can replay updates and rows. | +| Pending-repository pass | **blocked** | Pending repositories are repaired after captured live rows are merged, preserving intended row order. | Stream materializes pending DIDs and processes them through a separate path rather than upstream's bounded retry runner with the same global/per-host gates and retry semantics. | +| Merge-tail compaction and manifest reconcile | **partial** | Destination sealing precedes delete/update compaction; manifest reconciliation happens before serving is enabled. Physical rewrite tests cover named write failures. | Compaction watermark read corruption fails open. | +| Post-bootstrap discovery | **blocked** | It resumes from the last non-empty bootstrap cursor and creates failed rows for unknown active repositories on the normal path. | Unknown inactive repositories are skipped instead of recorded. A next cursor longer than 512 bytes silently exits the loop and cleanup proceeds, omitting the remaining network. | +| Cleanup durability | **verified** | On the successful path, the backfill tree is removed, the data directory is synced, and merge/discovery cursors are deleted in a synced RocksDB batch afterward. Restart-after-cleanup repeats the directory sync. | This row assumes the source was correctly classified as absent; that guard is separately blocked. | + +## Steady failed-repository retry + +| Invariant | Status | Audited evidence | Gap or required proof | +|---|---|---|---| +| Global and per-host concurrency | **verified** | The implementation has real global workers and host gates; focused held-request tests measure configured/default 16 and 4 limits. | The candidate set is still materialized eagerly. | +| Candidate scan memory | **blocked** | None at network scale. | `collectByStatus` allocates every failed DID, then builds every host gate before workers start. Upstream streams an iterator into bounded work. | +| Final-host attribution | **verified** | The final post-redirect authority replaces prior attribution after a response; redirected tests inspect the durable host. | Requests that fail before a response may remain unattributed, as upstream permits. | +| 429 host parking | **partial** | A 429 delays only work assigned to that final host in focused tests. | Stream persists host parking whereas upstream keeps it process-local. Allocation, RocksDB, and malformed-value errors all mean “not parked,” which can hammer a limited PDS. Decide whether persistence is an intentional divergence; in either case, fail loud on corrupted state. | +| Retry state and diagnostics | **verified** | Attempts, class, last error, next attempt, host aggregates, bounded recent samples, active/status counts, and success clearing are committed with repository transitions in focused restart tests. | This does not prove the retry supervisor remains alive. | +| Retry subsystem failure | **blocked** | Pass errors are logged. | RocksDB/archive/infrastructure errors are swallowed and retried at the next interval while service health stays green. Upstream lets the retry goroutine error terminate the service. | + +## Why the previous adversity gate passed + +The previous gate described the oracle as covering “after-repo-complete” +crashes. In Stream, that crashpoint is executed once per result only after the +full `putMany` succeeds. In the four-account restart world, every numbered +crash therefore occurred after all four completions were already durable. It +could not exercise the upstream state “repo A complete and durable while repo B +from the same listRepos dispatch is still running.” + +The replacement adversity matrix must include: + +1. N−1 durable completions plus one indefinitely held repository; +2. a crash after each individual writer durability hook, before listRepos + cursor advancement; +3. phase, relay cursor, merge cursor, compaction watermark, and host-park read + errors plus malformed bytes; +4. unreadable versus genuinely absent bootstrap-live directories; +5. inactive unknown repositories and a cursor longer than every local fixed + buffer; +6. retry-store failure with health/readiness assertions; and +7. a process restart proving completed repositories are skipped without + relying only on final-state convergence. diff --git a/docs/configuration-parity.md b/docs/configuration-parity.md index 6225cfa..0e3686c 100644 --- a/docs/configuration-parity.md +++ b/docs/configuration-parity.md @@ -1,40 +1,77 @@ -# Jetstream V2 configuration parity - -Authority: pinned upstream commit -`f29815c391fc2644f8a3dd36b899fb3697dd1ea6`, `cmd/jetstream/main.go` and -`internal/jetstreamd/options.go`. A row is closed only when the value changes -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 | 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 uses those names and the matching environment sources are covered by the dedicated environment row | -| 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 | -| 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 | -| segment cache max age | closed | real HTTP receipt covers Go duration, ceil-to-seconds, and cache headers | -| planBackfill caps, pagination, density threshold | closed | `just plan-config-contract` covers non-default and zero sentinel 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 | 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 | -| public/debug bind addresses | closed | `serve --addr` binds the protocol-only public listener, an empty `--debug-addr` binds nothing, and a configured debug listener exclusively serves health/readiness/metrics; `just listener-contract` proves route isolation, public WebSocket upgrade, rejected debug upgrade, disabled-debug refusal, listener readiness, and the historical combined-port compatibility mode through real offline processes | -| shutdown/client drain timeouts | closed | canonical signed Go-duration flags, 5s/10s defaults, net/http's non-positive 30s sentinel, the client drain's already-expired non-positive sentinel (immediate teardown with a best-effort close frame), listener-admission gating, concurrent public/debug/pipeline shutdown, close code 1001 for positive budgets, cooperative early exit, and one shared silent-client deadline are covered by unit and real ReleaseSafe process receipts. Bootstrap, merge, and discovery run behind a cancellable lifecycle boundary: a real process establishes live capture, hangs `listRepos`, exits promptly on SIGTERM, then reopens the same durable directory and proves the contract again. Per-client close hooks fan out concurrently with borrowed-handler lifetime protection; a real-socket transport regression saturates the first peer's send buffer, proves a later peer receives 1001 before the deadline, and proves the blocked writer is interrupted at the shared deadline. | -| log level/format | closed | `--log-level` retains debug calls in ReleaseSafe and applies the upstream case-insensitive aliases plus info default to every `std.log` producer; `--log-format` defaults to newline-delimited JSON and selects canonical text fields. `just logging-contract` proves default JSON, warn filtering, text/debug output, persistent root-flag placement, and fail-closed invalid values through real offline processes. | -| OpenTelemetry | closed | provider bootstrap is real and empirically covered: endpoint activation/precedence, HTTP/protobuf path rules, headers, gzip, insecure mode, request timeout, batch queue/delay/size/export timeout, upstream retry/backoff/throttle behavior, the six SDK samplers, process plus service resource attributes, global provider installation, W3C TraceContext+Baggage propagation (including no-exporter mode), bounded shutdown export, custom CA trust, and client certificate authentication all run through pinned `otel-zig`. Every pinned production span matches upstream names, scopes, parentage, attributes, and success/error finalization, including `xrpcapi/getBlock`; real offline OTLP collectors exercise the production sync/async archive, success/failure XRPC boundaries, native custom-CA HTTPS, rejection without a client certificate, and a successful verified mTLS request. The production image carries the libcurl runtime selected only by a complete client certificate/key pair. See `opentelemetry.md`. | -| `JETSTREAM_*` environment sources and unknown-variable rejection | closed | the exact pinned flag-source map applies root, serve, and `inspect-all` values with CLI-over-environment precedence; overridden invalid values are never parsed. Unknown names are sorted/deduplicated and rejected before command execution while upstream's Kubernetes `JETSTREAM_APP_*` and its `JETSTREAM_ORACLE_*`/`JETSTREAM_SIM_*` test namespaces are exempt. `just environment-contract` proves real ReleaseSafe root/inspect/serve processes, representative string behavior, every typed source, timestamp-import authentication, and precedence offline; the multiseed oracle proves its inherited upstream seed reaches the crash campaign without leaking into Stream configuration. | -| serve/version/inspect command surface | closed | `serve` is accepted explicitly while retaining Stream's historical flag-only invocation. `version` emits the upstream build-information shape. `inspect-segment` matches the pinned renderer byte-for-byte on an upstream-produced sealed fixture while also walking active files, surfacing partial tails, and labelling readable checksum corruption. `inspect-all` folds the real steady and bootstrap trees with upstream's missing-root, racing-tail, active-skip, aggregation, sorting, truncation, and text-rendering semantics; a copied upstream golden report is byte-exact. Listener and root logging controls remain tracked by their separate rows. | - -This table tracks configuration semantics only. It does not override the -behavioral gates in `semantic-parity.md` or `bootstrap-semantic-parity.md`. -The Go runtime's `/debug/pprof/*` handlers are intentionally not imitated: -Stream cannot emit truthful Go runtime profiles, and a route-shaped substitute -would violate the no-stub parity rule. Zig/process resource telemetry remains -real and fail-closed through the checksum-pinned dashboard contract. +# Jetstream V2 configuration audit + +Authority: + +- upstream `cmd/jetstream/main.go` and `internal/jetstreamd/options.go` at + `f29815c391fc2644f8a3dd36b899fb3697dd1ea6` +- Stream `b709fe32c72e9eee4725f92e7999eb6533bb3c8f` + +This document audits public configuration and command wiring. A **verified** +configuration row proves only that the input is parsed/defaulted and reaches +the named runtime mechanism. It does not make a blocked runtime mechanism +semantically correct. Status vocabulary is defined in +[semantic-parity.md](semantic-parity.md). + +## Inventory + +| Surface | Status | What source/tests establish | Limitation or remaining proof | +|---|---|---|---| +| Complete upstream flag inventory | **unverified** | The environment contract contains a hand-maintained pinned source map and rejects unknown `JETSTREAM_*` names. | There is no generated comparison proving every current upstream flag is represented here. Before another admission, generate the inventory from the pinned Go command/options definitions and fail on an unclassified addition/removal. | +| Relay URL | **partial** | `--relay-url` reaches the WebSocket consumer and relay-fronted HTTP listRepos/getRepo/repair clients; CLI and environment precedence are tested. | Correct wiring still feeds a blocked Zat reconnect loop. | +| PLC URL | **verified** | `--plc-url` configures the real DID resolver used by live verification and selected-repo metadata; loopback resolution tests observe requests. | This row does not prove every verification/repair outcome. | +| Data directory | **partial** | `--data-dir` owns JSS, scratch, metadata, bootstrap-live, import, and temporary trees in production-process tests. | Startup recovery and several metadata reads are blocked; path ownership alone is verified. | +| `--max-backfill-repos` | **partial** | Canonical spelling is parsed, limits dispatch, avoids committing a whole-network resume cursor, and automatically skips merge discovery. | The lifecycle and completion paths it invokes are blocked. It is a debug mode, not evidence for whole-network parity. | +| `--backfill-repos` | **partial** | Canonical spelling and compatibility alias parse; it is mutually exclusive with max-repos, preserves selected order, bypasses listRepos, and enables selected identity resolution. | The exact current artifact still needs the complete selected-repository admission receipt; configuration wiring itself is established. | +| Backfill workers | **verified** | Explicit positive values set physical getRepo concurrency; zero/omitted default to 100. Held real requests observe 7/100/100. | The 200-worker production experiment showed that accepting a value does not mean it is efficient. | +| Backfill batch size | **verified** | Explicit and zero/default values control only page-aligned listRepos accumulation and cursor-checkpoint cadence; zero/omitted select 100,000. Repository completion now follows independent writer durability boundaries. | This control does not define whole-network size or completion percentage. | +| Backfill async-flush workers | **partial** | Omitted defaults to four, positive values start that many bootstrap compression workers, and zero selects synchronous compression. Focused tests observe overlap and byte-equivalent output. | Full cancellation/OOM state coverage was not re-established in this audit. | +| Skip merge discovery | **partial** | Explicit configuration and automatic max/selected enablement prevent the discovery request in focused tests. | Normal discovery is blocked by inactive-entry and long-cursor behavior. | +| Failed-repo retry interval | **partial** | Duration parsing, zero disablement, and timer-driven pass execution have real-process tests. | Pass failures are hidden from health and candidate scans are unbounded in RAM. | +| Failed-repo global/per-host workers | **partial** | Explicit and zero/default values reach real 16-global/4-per-host gates in held-request tests. | Pending-merge retry uses a different path, and the steady candidate set is materialized eagerly. | +| Failed-repo maximum delay | **partial** | Parsed values affect persisted exponential-jitter deadlines in focused RocksDB tests. | Host-park read corruption fails open; whole retry supervision is blocked. | +| Expensive account verification limiter | **verified** | Enabled/default and disabled modes cross the real `/status?tab=accounts` HTTP path. Four requests per source IP succeed, the fifth is rejected, and HEAD does not consume a slot. | This verifies the control, not the completeness of the status surface. | +| Cursor lookback | **partial** | Default 36h, explicit duration, and zero/pure-live change v1/v2 cursor admission in production-process tests. | Accepted cold cursors still depend on a replay walker that can skip unreadable segments. | +| Segment cache max age | **verified** | Go-duration parsing and ceil-to-seconds response headers are observed on real getSegment/getBlock responses; zero/default no-cache behavior is covered. | None known for this control. | +| planBackfill DID/collection caps | **partial** | Positive and zero sentinel values affect request validation in real HTTP tests. | A valid request can still receive a false-negative plan when `matchBlocks` encounters an internal error. | +| planBackfill entry limit | **partial** | Positive values paginate by work unit; zero disables the cap; continuation fixtures match normal upstream responses. | Internal planner errors are silently converted into non-matches. | +| planBackfill density threshold | **partial** | Configured thresholds change whole-segment versus block mode at tested boundaries. | Same planner error blocker. | +| Subscribe hot-log retention | **verified** | The byte value drives physical eviction in the production tail; tests cover eviction while respecting the durable floor. | Does not verify cold recovery after eviction. | +| Subscribe slow-consumer window/rate | **verified** | Parsed duration and fractional rate reach the per-subscriber scan detector; explicit and default/non-positive behavior have focused tests. | None known for the detector control. | +| Subscribe cold-block cache | **partial** | The byte bound controls the shared decoded-block/wire memo LRU; concurrency, eviction, and manifest-generation invalidation have tests. | The enclosing cold walker skips an unreadable segment, so cache correctness cannot establish replay completeness. | +| Subscribe read batch | **partial** | Explicit positive values and non-positive default 1024 bound raw scanned entries before filters in hot and normal cold fixtures. | The 5,000-row receipt tests batching, not missing-file continuity. | +| Deprecated cursor block-index cache flag | **intentional divergence** | Stream accepts signed values as a compatibility no-op because its sealed indexes are always manifest-resident. Parser behavior is tested. | This is not the same runtime mechanism as an implementation that uses the cache. Keep it documented as compatibility, not parity. | +| Compaction interval | **partial** | Go-duration parsing, zero disablement, periodic execution, and cap-trigger wakeup are exercised. | A malformed/unreadable durable watermark is treated as missing and reinitialized. | +| Compaction tombstone cap | **partial** | Explicit and zero/default values affect real chunk count and physical survivors. | Correctness still depends on the blocked watermark read path. | +| Compaction rewrite workers | **partial** | The configured value controls real concurrent rewrite groups in focused physical-file tests. | Full pass admission remains blocked by watermark corruption behavior. | +| Timestamp import token | **verified** | Missing/wrong/correct bearer behavior crosses the production import/status XRPC handlers, including constant-width comparison. | None known for the token control. | +| Timestamp import directory | **partial** | Configured roots are canonicalized/confined and real jobs survive restart in focused tests. | Corrupt import metadata reads were not re-audited; public results depend on blocked cold replay. | +| Public bind address | **verified** | `serve --addr` binds the protocol listener; public route and WebSocket behavior are exercised by real processes. | Route availability does not imply archive readiness/completeness. | +| Debug bind address | **verified** | Empty disables binding; configured debug address exclusively serves health/readiness/metrics and rejects protocol upgrade. Historical combined mode is separately tested. | None known for listener wiring. | +| Shutdown timeout | **partial** | Signed duration parsing/defaults reach listener and lifecycle shutdown; real tests cover positive, non-positive, hung listRepos, and cooperative exit. | Exact-artifact admission was violated, and reconnect failures remain independently process-fatal. | +| Client drain timeout | **verified** | Configured/default/non-positive budgets reach concurrent client close hooks; real blocked-socket tests observe 1001 and shared-deadline interruption. | Limited to client-drain behavior. | +| Log level | **verified** | Case-insensitive aliases/default and invalid values affect all `std.log` producers in real ReleaseSafe processes. | Not byte-identical Go logger internals. | +| Log format | **verified** | JSON default and text selection produce the documented canonical fields in real processes. | Not byte-identical Go formatting beyond the asserted fields. | +| OpenTelemetry endpoint/protocol | **verified** | Activation, precedence, HTTP/protobuf path construction, insecure mode, headers, gzip, and timeout reach real collectors. | Limited to tested HTTP/protobuf transport; no claim about an unimplemented protocol. | +| OpenTelemetry batching/retry/sampling | **partial** | Queue/delay/batch/export timeout, upstream retry/backoff responses, six samplers, propagation, and bounded shutdown have focused tests. | This audit did not independently enumerate every upstream span/attribute producer. | +| OpenTelemetry TLS/mTLS | **verified** | Custom CA, missing client certificate rejection, complete certificate/key selection, and verified mTLS requests use the production transport/image dependencies. | None known for the tested TLS controls. | +| `JETSTREAM_*` environment mapping | **partial** | The hand-maintained map covers all currently listed Stream root/serve/inspect/import controls; CLI precedence and typed parsing run through real processes. | Map completeness must be generated from upstream rather than asserted by the map itself. | +| Unknown `JETSTREAM_*` rejection | **verified** | Unknown names are sorted/deduplicated and rejected before command execution; documented Kubernetes/test namespaces are exempt. | The exemption list must be rechecked when the upstream inventory changes. | +| `serve` command | **verified** | Explicit `serve` and the historical flag-only compatibility invocation both enter the same runtime; persistent root flags are exercised. | Historical invocation is a Stream extension, not an upstream requirement. | +| `version` command | **verified** | It emits the upstream build-information field shape in a focused process test. | Values necessarily identify the Stream artifact. | +| `inspect-segment` | **partial** | A sealed upstream fixture matches the pinned renderer for asserted output; active/partial/checksum cases have Stream tests. | The complete current golden set was not rerun during this audit. | +| `inspect-all` | **partial** | Representative steady/bootstrap trees, missing roots, active skip, aggregation, sorting, and truncation have fixtures, including a copied upstream report. | It does not replace missing online collections/segments status tabs; complete golden coverage was not independently re-audited. | +| Go pprof routes | **intentional divergence** | Stream does not expose fake `/debug/pprof/*` handlers. Zig/process metrics are used instead. | This is the correct explicit divergence; a route-shaped stub would be dishonest. | + +## Required configuration admission + +Before a deployment can cite this checklist: + +1. generate and diff the public flag/environment inventory from the exact + pinned upstream source; +2. require every inventory difference to have one row above; +3. run each **verified** row's production-boundary receipt against the exact + artifact digest; +4. forbid a configuration row from overriding a **blocked** behavioral row; +5. record intentional divergences explicitly in README/operator docs; and +6. invalidate the receipt after any source, dependency, dashboard, container, + or default-value change. diff --git a/docs/grafana-dashboard.md b/docs/grafana-dashboard.md index 08d59b6..f9c1359 100644 --- a/docs/grafana-dashboard.md +++ b/docs/grafana-dashboard.md @@ -15,31 +15,29 @@ network or a moving branch. Stream provisions deterministically generated by `tools/adapt_dashboard.py` from the pinned source. The generator refuses a changed source checksum, and the tests prove every upstream non-runtime panel remains structurally identical. Stream adds -two explicit final rows. The first pairs a continuously moving processed-repo -bar with upstream's durable page-commit ratio and labels both as progress -through the current `listRepos` page. It does not present that page ratio as -whole-network completion: the final network denominator is unknown until -discovery finishes. The second covers the scheduler, which has no upstream Go -equivalent: ticket boundaries, exact unsettled stage gaps, current fatal state, -and first-failure counters partitioned by scheduler/emit/repair/cursor stage. -These producers are active during historical bootstrap as well as steady -state; the dashboard never infers scheduler health from archive throughput. +two explicit final rows. The first displays Stream's current dispatch-batch +counters. Those counters are not whole-network progress, and the 2026-07-24 +audit found that their durability boundary is not equivalent to upstream's +per-repository completion batcher. The second covers Stream's scheduler, which +has no upstream Go equivalent: ticket boundaries, unsettled stage gaps, current +fatal state, and first-failure counters partitioned by stage. `just dashboard-test` verifies both checksums, regeneration, and the contract parser. `just dashboard-contract ` extracts every metric family referenced by the Stream dashboard's PromQL and compares it with the candidate's exposition, including histogram child series. It exits -non-zero and prints every missing family. This is deliberate: declaring -placeholder zero-valued metrics would make a dashboard appear healthy while -measuring nothing. +non-zero and prints every missing family. This is a schema-presence check, not +evidence that a producer has the upstream semantic boundary or remains correct +across restart. ## Reuse boundary Protocol and lifecycle rows are reused verbatim: overview, live firehose, ingest, backfill/retry, serving, storage, integrity, cutover/merge, and -timestamp import. Their queries are part of the parity contract, so Stream -emits the same `jetstream_*` family only after it owns the same measurement -boundary. +timestamp import. Structural reuse does not establish semantic parity. The +current producer audit is in +[semantic-parity.md](semantic-parity.md), and any family whose underlying +behavior is blocked must be treated as diagnostic rather than equivalent. The Go runtime row is the sole adaptation. Stream does not synthesize `go_*` series. It retains upstream's same-semantics standard process measurements for @@ -201,16 +199,18 @@ denominator. Upstream requests are pages of 1,000 entries; the production default accumulates roughly 100,000 entries into a shuffled dispatch/checkpoint batch before advancing durable progress. -The backfill families are measured at Stream's actual dispatch-batch checkpoint: worker -results enter the completion queue after download/handling, archive bytes are -flushed first, and only the succeeding RocksDB transition batch removes them -from queue depth and reports completion. This is the semantic counterpart of -upstream's writer-hook completion batcher without manufacturing a second queue. -Offline loopback receipts cover a terminal RepoNotFound, a generated complete -CAR, redirected rate limiting, and a restart-safe parked-host skip. +The backfill completion families are measured at the writer durability +boundary, independently of the dispatch/checkpoint batch. A successful CAR +records its final archive sequence; ordinary writer fsyncs commit every covered +repository state, and the terminal dispatch drain refuses to advance the +listRepos cursor if any queued completion is not covered. Queue and durable +batch/repository metrics therefore describe the same boundaries as upstream. +The Prometheus counters themselves are process-local; restart-stable progress +must be reconstructed from durable repository-state counts rather than by +pretending a counter cannot reset. Compaction metrics are emitted at the actual fold, worker rewrite, manifest refresh, watermark commit, live-set, and timer boundaries. The production -offline receipt now reports `required=104 present=104 missing=0`; presence is -paired with real-JSS unit receipts so registered zero series cannot substitute -for an implementation. +offline receipt reports `required=104 present=104 missing=0`. That result means +only that every queried family is exposed; it does not establish producer +semantics or close the deployment gate. diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index be31242..7d50f67 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -1,77 +1,114 @@ -# Jetstream V2 semantic parity gate - -Semantic parity means the same durable state transitions and externally -observable behavior as `bluesky-social/jetstream`, not merely compatible -routes or JSON. The current review receipt is upstream commit -`f29815c391fc2644f8a3dd36b899fb3697dd1ea6`, Atmos v0.2.14, and the canonical -dashboard at `contrib/grafana/jetstream.json` (SHA-256 -`ef6033c182d8c3f9d88af7d91e61126724efbe72e6b940984c9de12e9530cd91`). -The 2026-07-22 refresh from the prior `f6d220f` receipt to this pin changed -only the upstream Docker/Nix toolchains and Go module versions: no Go source, -lexicon, fixture, or dashboard file changed. The oracle pin and offline Go -module cache were still advanced so the candidate is exercised with the exact -current dependency graph rather than relying on that source-level comparison. - -No full-network machine may be provisioned while any deployment blocker in -this document or `bootstrap-semantic-parity.md` is open. -The public runtime-control audit is maintained in `configuration-parity.md`; -its open rows are also deployment blockers unless explicitly classified as a -deliberate, documented operational divergence. - -| Surface | Evidence | Gate | -|---|---|---| -| Bootstrap, merge, retry | Detailed row-by-row contract and adversity receipt | **closed at the current implementation boundary.** Every row in `bootstrap-semantic-parity.md` is closed; the offline candidate rerun survived all eight lifecycle crash seams. The experiment gate still requires rerunning that oracle against the exact eventual build. | -| JSS v1 storage | Upstream-produced fixture, reciprocal segment parsing, checksum/index/gloom vectors | closed for sealed format; rerun reciprocal corpus before experiment | -| Archive XRPC | Official Go client plus Stream conformance replay over listSegments/getSegment/getBlock/planBackfill | **closed for wire behavior, cache policy, and planner controls.** 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 the pinned `http.ServeContent` behavior for both download endpoints. Single, suffix, multipart, mixed-overlap, aggregate-size fallback, and the distinct malformed/non-overlap 416 responses cross the real production server. The same receipt proves descriptor- or manifest-derived Last-Modified values, Go's If-Match/If-Unmodified-Since/If-None-Match/If-Modified-Since precedence, strong ETag and exact-date If-Range, and 304/412 representation-header behavior. The date parser accepts all three forms supported by Go's `http.ParseTime`. A separate production-process receipt passes `--segment-cache-max-age=1500ms` and requires upstream's ceil-to-seconds `public, max-age=2` response; the default remains `public, no-cache`, and 416 clears the representation cache header. Three more production processes consume a physically sealed 3/4-density corpus and prove the distinct DID/collection caps, zero-disabled filters, one-entry pagination, zero-entry-limit unbounded paging, and exact 1.0/0.75 whole-segment threshold boundary. Real HTTP receipts also prove the middleware boundary: unknown NSIDs are 501 `MethodNotImplemented`, storage faults are 500 `InternalServerError`, and pre-steady-state XRPC requests receive a JSON 503 `ServiceUnavailable` while operational routes remain reachable. 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`. 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`. | -| Oracle | Event-log equivalence, final-state convergence, crash/power-loss, verifier repair, hostile input, and anti-vacuity receipts | **closed at the implementation boundary.** `just differential-oracle` reuses the exact pinned upstream simulator, physical JSS reader, event-log normalizer/comparator, independent MST ground truth, invariant checker, fault scheduler, and public Go client through a fail-closed Go overlay. A real ReleaseSafe Stream child now proves bootstrap convergence; exact normalized durable equivalence for controlled create/update/delete/identity/account/sync/create_resync traffic; duplicate and regressed relay windows with account lifecycle events; the complete upstream frame-fault campaign (malformed/error/oversized/unknown/swallowed/partial-CAR); exact reconnect, sequence-gap, decode, unknown, missing-block, and repair metrics; all-kind anti-vacuity; mutation detection; clean SIGTERM drain; same-disk restart; and public plan/getSegment/getBlock/subscribe-v2 reconstruction. The complete upstream adversarial ingest matrix is also closed: live sibling-preserving invalid collection/rkey/field-width drops, divergent invalid-rev sync repair-then-whole-event-drop, signed non-TID verifier rejection plus follow-up repair, DID-scoped PLC failure without sibling wedge, and backfill CAR invalid UTF-8/path/width drops all run through real sockets, RocksDB, signed repos, and physical JSS with source/reason counter floors and filtered MST convergence. The upstream merge-cursor store-fault contract runs through a real process and RocksDB: the selected durable batch fails loudly, proves it fired, and a fault-free reopen converges from the unadvanced cursor. Segment-file fault parity is also closed: every production JSS write/fsync/commit-rename crosses the same operation taxonomy, the real process proves ENOSPC/EIO/short-write failure and recovery, Patch and Rewrite exhaustively sweep every local ordinal against real files, and a compaction rename failure preserves both the source bytes and unadvanced watermark before a successful retry. The strict lifecycle power-loss tier runs the production Linux binary, RocksDB, and JSS on ext4 over a calibrated NBD volatile-write-cache model. It kills storage before the process and reconstructs solely from FLUSH/FUA-acknowledged bytes; all eight canonical lifecycle schedules plus upstream's two-crash restart-after-cleanup guard recover through real XRPC and public V2 replay. Six compaction schedules cover rewrite write/fsync/rename/dir-sync and both sides of the durable watermark; four timestamp-import schedules cover patch write/fsync/rename/dir-sync. All ten converge to exact row/timestamp results without stale temporary files. The campaign closed directory-fsync, cleanup-cursor ordering, Linux subscriber-stack, pre-bootstrap serving, cold-cursor, and Linux directory-handle defects. The five-world seeded predicate-kill tier reuses upstream's exact chain generator, selector, event-log coverage, compaction filter, final-state, visibility, and red-first assertions against real Stream children. It closed the bootstrap live-attachment and close-before-seal partial-block durability gaps. | - -## Experiment admission - -A new full-network experiment requires all of the following in one immutable -build receipt: - -1. Zig 0.16 unit/build success and a clean differential/conformance run. -2. Every row above and in the bootstrap gate closed against the then-current - upstream commit. -3. The checksum-pinned dashboard contract check passing against the candidate - binary, with no zero-value placeholder families. -4. A local simulator adversity run proving each injected fault actually fired - and the archive converged after restart/resync. -5. A capacity/cost stop policy recorded before the host is created. - -The detailed dashboard reuse and metric-family policy is in -`grafana-dashboard.md`. - -### Immutable candidate receipt: `4d163e8` (2026-07-22) - -All five admission conditions are satisfied for Stream commit `4d163e8` -against upstream `f29815c`: - -- the Zig 0.16 unit suite and executable build passed at the exact commit; -- the offline differential oracle passed bootstrap controls, event-log and - final-state equivalence, replay/frame/adversarial faults, the 96-repository - resync saturation boundary, and all five seeded process-kill worlds; -- the official Go archive client passed every whole-segment, filtered-block, - bounded-range, live-cutover, and resident-manifest assertion using the exact - cached Go 1.26.5 dependency graph; -- the candidate scrape supplied all 104 metric families required by the - checksum-pinned dashboard, with zero missing producers; -- the strict Linux ext4/NBD campaign recovered all eight lifecycle cuts, the - two-crash cleanup guard, six compaction cuts, and four timestamp-import cuts - using only FLUSH/FUA-acknowledged bytes; -- native custom-CA HTTPS and mutual-TLS OTLP collectors accepted the complete - client certificate/key configuration and rejected incomplete or untrusted - configurations; and -- sibling operations commit `086995f` records the 24-hour/$30 hard spend gate, - empty-project isolation, independent pre-armed destruction watchdog, and - evidence-capture path without provisioning a host. - -The exact tested linux/amd64 candidate is published at -`atcr.io/zat.dev/stream:4d163e8@sha256:d2a0fbd99e0546236b110396c828be99f2149fa02e736f80e2367ef68626eddb`. -No full-network host was created by this receipt. +# Jetstream V2 semantic parity audit + +This is a deployment gate, not a record of work performed. A row describes +what the current implementation can prove today. It must not be marked more +strongly because an earlier design was worse, because a route has the right +shape, or because a test written around Stream's implementation passes. + +## Audit basis + +- Stream: `b709fe32c72e9eee4725f92e7999eb6533bb3c8f` +- upstream Jetstream: `f29815c391fc2644f8a3dd36b899fb3697dd1ea6` +- Atmos: `v0.2.14` +- Zat: `649dae356c576a205ec97f2b23671af365756b55` +- audit date: 2026-07-24 + +The upstream checkout is exactly at the recorded pin. The untracked +`simulator` path in that checkout is not treated as upstream source. + +## Status vocabulary + +- **verified**: source comparison establishes the same narrow invariant and a + test exercises the production boundary, including the relevant failure mode. +- **partial**: some named invariants are verified, but the surface has an + unproved or non-equivalent boundary. +- **blocked**: a known semantic mismatch, data-loss risk, false-negative API + result, or process-fatal remote-input path exists. +- **intentional divergence**: the behavior deliberately differs and must not be + called parity. +- **unverified**: no known mismatch was found, but the available evidence is + insufficient. + +Only **verified** rows may contribute to a parity claim. Passing a test after +the audited commit does not update this document automatically. The source, +test, and this audit must be reviewed together. + +## Current decision + +**Stream is not at semantic parity and is not admitted for another +whole-network experiment.** + +The bootstrap completion model, correctness-metadata reads, merge restart +guard, post-bootstrap discovery, cold replay, ordinary live encoding, and +firehose reconnect loop are deployment blockers. The detailed bootstrap audit +is in [bootstrap-semantic-parity.md](bootstrap-semantic-parity.md). + +## Audited checklist + +| Surface | Status | What is actually established | Blocking or missing evidence | +|---|---|---|---| +| Lifecycle phase machine | **blocked** | The nominal phases are bootstrap → merging → steady_state and writes are synced to RocksDB. Cleanup ordering includes data-directory sync before cursor deletion. | `phase.Store.read` converts RocksDB errors and unknown values to “no phase.” Stream also commits `merging` after bootstrap-live teardown, while upstream commits it immediately after backfill drains and before teardown. Add read-corruption tests and port the upstream transition boundary. | +| Whole-network bootstrap | **blocked** | listRepos pages are accumulated to a configurable page-aligned dispatch batch; eligible repos are shuffled; download concurrency and CAR preparation are real. | Completed repositories are not recorded durably until every job in the dispatch batch finishes. See the detailed audit. | +| Merge and post-bootstrap discovery | **blocked** | Source rows are filtered by the backfill revision; the source cursor and latest-revision updates share a synced batch; successful cleanup is directory-synced. | Merge cursor read errors become zero. Any error opening the bootstrap-live directory is treated as “already removed.” Discovery skips inactive unknown repos and silently stops on cursors longer than 512 bytes. | +| Failed-repository healing | **blocked** | A real global/per-host worker gate exists, final redirect hosts are recorded, and retry state is stored. | Pass-level RocksDB/archive failures are logged and hidden from service health. Every failed DID and host gate is materialized in RAM before work. Persisted host parking is not upstream behavior, and park-read errors mean “not parked.” | +| JSS sealed format interoperability | **verified** | Upstream-produced sealed fixtures are parsed; Stream-produced sealed files are consumed by the pinned Go reader; header/footer, block index, blooms, collections, compression, and checksums have reciprocal fixtures. | This verifies sealed-format compatibility only. It does not verify startup recovery or replay completeness. | +| Archive startup recovery | **partial** | Torn active tails with ordinary decode termination are truncated to the last complete block and sealed. A truncated sealed segment is subsequently rejected by manifest loading in the current composition. | `Archive.recover` swallows file-read/header errors and converts any active-iterator error, including allocation failure, into end-of-valid-data before rewriting the file. Upstream fails loud on these errors and checksum-verifies a sealed high-water mark. Add fault-specific recovery tests before claiming parity. | +| `listSegments`, `getSegment`, `getBlock` | **verified** | The pinned Go client and direct HTTP tests cover normal responses, byte identity, checksums/ETags, conditional requests, ranges, cache headers, and storage-error responses through the production server. | The claim is limited to these three methods and the tested file states. | +| `planBackfill` | **blocked** | Normal DID/collection/sequence filtering, whole-segment versus block mode, pagination, and configured limits match the pinned client fixtures. | `matchBlocks` errors are handled with `catch continue`, creating a successful plan with a false negative. Upstream's planner returns the error. Add an injected allocation/internal-error contract. | +| Subscribe v1/v2 handshake and wire formats | **partial** | Cursor parsing, v1/v2 filter shapes, WebSocket framing, v1 deflate, v2 dictionary zstd, dictionary validators, size caps, pings, write deadlines, and shutdown close frames have production-path tests. | These tests do not establish gap-free cold replay or remote-data encoding isolation. | +| Cold archive replay | **blocked** | Normal sealed and active-prefix replay reaches the hot tail in order in existing fixtures. Timestamp cursor corruption is surfaced for the selected block. | Both cold walkers skip a manifest-listed segment when `openFile` fails. They may then advance beyond durable events never delivered. Upstream returns an error and enforces progress/hole invariants below the readable-log floor. | +| Live Sync 1.1 verification | **partial** | DID/signature checks, MST inversion, op-CID checks, replay/future-rev gates, durable chain/hosting updates, and several repair-trigger cases use real RocksDB, CARs, and sockets. | The surface cannot be called closed while an accepted live row can persist and then kill the process during eager wire encoding. Metadata cursor corruption also fails open. | +| Live scheduling and cursor durability | **partial** | Per-DID ordering, bounded pending work, worker admission, completion-order emission, and archive-before-cursor write ordering are represented in production code and focused tests. | The Zat reconnect sleep can propagate `HttpConnectionClosing`; production observed two process exits. Relay cursor load errors/malformed values become absence, which can omit the interval between the durable cursor and current relay tip. | +| Whole-repository repair/resync | **partial** | Fetched repositories are authenticated before replacement emission; valid repair emits sync then bounded replacement batches; repair-tail encoding errors are isolated per row. Capacity tests exercise 32 active plus 64 queued repairs. | The isolation applies only to repair-tail publication. Ordinary live publication still eagerly encodes and can terminate the process. Retry-subsystem terminal failures are not tied to health. | +| Delete/update compaction | **partial** | Rewrite selection, survivor correctness, manifest-before-watermark ordering, fsync/rename cutpoints, bounded workers, and several reason counters have physical-file tests. | Compaction watermark read/corruption is treated as missing and reinitialized. A read-error fault contract against the production metadata store is required. | +| Timestamp import | **partial** | CSV parsing, rule precedence, patch topology preservation, job persistence, restart, HTTP authentication, and write/fsync/rename power-loss cuts have concrete tests. | Public visibility after import still depends on the blocked cold-replay path. No audit has yet proved behavior when timestamp metadata reads are corrupt rather than absent. | +| Status: hosts and accounts | **partial** | Durable host aggregates and account lookup/verification have focused production HTTP tests, including rate limiting and restart. | This is not the complete upstream status surface. | +| Status: summary, collections, segments | **blocked** | Stream has a custom summary. | Upstream exposes phase entry time, backfill timing/cursor/counts, live cursors, storage/manifest data, imports, and collections/segments views. Stream implements no collections or segments tabs and lacks several underlying persisted fields. | +| Prometheus metric exposition | **partial** | Metrics are emitted in Prometheus syntax, and focused tests show that a subset increments at their intended boundaries. | The dashboard contract checks metric-family presence, not producer semantics, label equivalence, monotonicity, or non-placeholder behavior. Several metrics described as durable progress were process-local or tied to the wrong batch boundary. | +| Grafana dashboard reuse | **partial** | The upstream dashboard is checksum-pinned; deliberate runtime-only panels are separated from upstream panels; the public experiment dashboard can be provisioned read-only. | A query finding a named family does not prove semantic equivalence. Progress panels must be derived only from genuinely durable, restart-stable counters. Dashboard admission remains blocked by producer semantics. | +| Public/debug listeners | **verified** | Production-process tests establish route isolation, disabled debug binding, public WebSocket admission, readiness, and the historical combined listener mode. | This row does not establish lifecycle readiness or data completeness. | +| Graceful shutdown | **partial** | Listener admission stops, clients receive 1001 within configured budgets, blocked peers are interrupted, and a hung listRepos bootstrap exits cooperatively in the focused receipt. | The earlier production shutdown hang was addressed, but the exact deployed image was not rerun through the complete immutable admission suite. Dependency reconnect failures remain independently process-fatal. | +| Logging | **verified** | Real processes exercise JSON/text selection, level filtering, default values, and invalid configuration. | Semantic parity is limited to the public controls and rendered fields, not byte-identical Go logging internals. | +| OpenTelemetry | **partial** | Provider configuration, propagation, sampling, batching, TLS/mTLS, and representative production spans have real collector tests. | “Every pinned production span” has not been independently re-audited in this pass. Treat the enumerated tested spans as evidence, not a blanket closure. | +| Inspect/version command surfaces | **partial** | `version`, sealed `inspect-segment`, active inspection, and representative `inspect-all` reports have fixtures and golden comparisons. | Inspection does not compensate for missing online status tabs, and the current audit did not rerun every golden against the current commit. | +| Differential oracle | **blocked as an admission proof** | It provides useful event-log, final-state, restart, repair, and public-client coverage against a pinned upstream simulator. Focused production-boundary tests now prove per-repository writer-gated completion and same-disk reopen with a sibling held incomplete. | The oracle still needs an invariant-based process-kill world that scales this beyond a two-repository fixture; its old four-account `after_repo_complete` schedule could not detect a dispatch-scale replay defect. | +| Strict power-loss oracle | **partial** | It exercises acknowledged-write reconstruction at named write/fsync/rename boundaries, and the focused reopen test proves completed-versus-interrupted repository classification around an unchanged listRepos cursor. | It does not inject correctness-metadata read failures, missing manifest files during replay, long listRepos cursors, or storage loss specifically between archive fsync and the completion metadata commit. | +| Exact artifact admission | **blocked** | Commit `4d163e8` had a recorded suite and immutable OCI digest. | The experiment deployed `e1926f3`, which included source and dependency changes after that receipt. No mechanism prevented deployment of a non-admitted digest. | + +## Evidence that must not be overread + +- `required=104 present=104 missing=0` means only that the scrape exposed every + family named by the dashboard. +- A reciprocal JSS fixture proves sealed-format compatibility, not crash + recovery or replay continuity. +- Final-state convergence does not prove bounded replay work or the timing of + individual durable completions. +- A successful restart after a named crashpoint proves only that the + crashpoint was placed at a safe boundary. It does not prove the boundary + matches upstream. +- A four- or 104-account simulator cannot by itself establish behavior at a + 100,000-entry dispatch/checkpoint boundary. Per-repository completion needs + an explicit held-sibling invariant, not merely a smaller happy-path batch. + +## Admission requirements + +A new whole-network experiment is forbidden until all **blocked** rows are +resolved and all **partial** rows that can affect durability, completeness, +availability, or operator truth are either **verified** or explicitly accepted +as divergences. + +The admission command must: + +1. require a clean worktree and record the exact Stream, upstream, Atmos, Zat, + and dashboard revisions; +2. build the exact linux/amd64 artifact that will be deployed; +3. run invariant-based negative tests, including metadata read corruption, + per-repository completion crashes, replay holes, persistent encode errors, + and transient firehose disconnects; +4. run the positive interoperability and configuration receipts; +5. publish by digest and emit one signed/immutable receipt binding every result + to that digest; and +6. make deployment reject any other digest. + +The previous `4d163e8` receipt remains historical evidence for that artifact. +It is not a current admission and must not be copied forward. diff --git a/docs/upstream-bootstrap-spec.md b/docs/upstream-bootstrap-spec.md index fc4d3e3..679dcd7 100644 --- a/docs/upstream-bootstrap-spec.md +++ b/docs/upstream-bootstrap-spec.md @@ -72,52 +72,24 @@ segment). /subscribe + /subscribe-v2 + XRPC return 503 "bootstrap in progress" until steady_state. failed-repo retry loop: 4h interval, 16 workers, 4/host, 7d max backoff, 429 parks the host. -## port notes -- Stream keeps the upstream 1000-repo durability page and requests up to 16 - long-lived download workers on the 16-core target. There is no CAR-size - exclusion. HTTP bodies stream to per-attempt scratch files on the data - volume. Serialized preparation copies the complete CAR into the configurable - global allocation budget (8 GiB default), builds the complete index, and - releases the scratch mmap before emission. An individual repository may use - the entire budget; prepared repositories emit from owned CAR bytes and the - loaded MST through upstream's fixed 1,024-row batches and feed four ordered - async zstd workers by default. - Resource exhaustion is a - lifecycle-fatal capacity error, never a durable per-repo skip. Scratch is - cleared before bootstrap or steady retry starts and each attempt unlinks its - own file. Downloads remain disk-backed, while CAR and index ownership are - explicitly bounded in RSS. -- Before emission, Stream parses the full CAR, verifies block hashes, indexes - every block, loads the MST, and walks it once to prove every MST node and - referenced record block exists. Boundary and mid-block truncations are - retryable. The relay/listRepos DID is authoritative for emitted rows; the - commit's embedded DID cannot redirect attribution. -- getRepo does not pool HTTP connections across repositories. The relay - redirects downloads to independently operated PDSes, some of which close - their response connection; Zig 0.16's pool can otherwise reuse that closing - socket and falsely fail the next DID with HttpConnectionClosing. -- Stream uses one RocksDB metadata database at `meta.rocksdb`, preserving the - upstream Pebble keyspaces and synced atomic-batch contract. Phase, relay and - listRepos cursors, repo status/counts/retry deadlines, host park state, - merge cursor, sequence watermarks, and compaction watermark belong there. -- the segment durable hook fsyncs event bytes before committing `seq/next`, - `seq/durable`, and the covered relay cursor; the durable watermark is - published only after that metadata batch succeeds. -- canonical orchestrator phase, transition, state-duration, and merge counters - are updated only at these durable lifecycle boundaries. The offline crash - matrix checks their restart semantics, while a deterministic real-JSS and - real-RocksDB merge receipt proves kept/dropped/source/lookup/revision counts. -- Stream's page result list is the real completion queue: a successful worker - queues exactly once, the page archive checkpoint makes every appended row - durable, and the following atomic RocksDB batch makes the queued repo states - durable. Queue depth, wait time, durable batch/repo totals, completed totals, - and current-run progress advance only after that batch succeeds. A failed - completion batch remains queued and increments the stage-error signal. - RepoNotFound follows the same completion path with zero appended rows. -- Discovery and active-flip counters advance only after their RocksDB writes; - successful HandleRepo timing covers validated CAR traversal and archive - append but excludes download and serialization wait. Failed-repo retry pass, - due-candidate, attempt, outcome, and parked-host counters use the same scan - and post-gate boundaries as upstream. Failure-persistence errors are counted - per affected repo rather than being disguised as remote getRepo failures. -- reproduce crashpoint seams at every commit boundary for future oracle tests. +## Stream audit note + +This file condenses the pinned upstream behavior. It is not a checklist of +features that Stream has implemented, and a matching noun, keyspace, metric, or +test name is not evidence of parity. + +The 2026-07-24 source audit found that Stream's dispatch-result list was **not** +the upstream completion queue: it waited for the full dispatch batch and could +replay nearly 100,000 otherwise successful repositories. The replacement now +records a per-repository final-sequence watermark and stages covered +completions from the ordinary archive durability hook. It also preserves +upstream's empty-CAR, RepoNotFound, duplicate-DID, captured-timestamp, and +forced-drain rules. Focused tests hold one sibling incomplete, cross the writer +boundary, and fully reopen the store without replaying the completed sibling. + +The same audit found fail-open phase/cursor/watermark reads, a fail-open merge +source guard, incomplete post-bootstrap discovery, an unbounded retry candidate +scan, and retry-supervisor failures hidden from health. Current Stream status +and the evidence needed to close each gap are maintained in +[bootstrap-semantic-parity.md](bootstrap-semantic-parity.md) and +[semantic-parity.md](semantic-parity.md). diff --git a/src/internal/bootstrap/engine.zig b/src/internal/bootstrap/engine.zig index 962be0d..75e5f6e 100644 --- a/src/internal/bootstrap/engine.zig +++ b/src/internal/bootstrap/engine.zig @@ -3,14 +3,16 @@ //! listRepos remains a serial page walk. Whole pages accumulate until the //! configured entry target, then eligible repositories are shuffled and fan //! out across long-lived workers. Every worker allocates through one shared -//! byte budget, so worker count controls parallelism while `max_inflight_bytes` controls -//! actual transient heap ownership. A dispatch batch commits in this order: +//! byte budget, so worker count controls parallelism while `max_inflight_bytes` +//! controls actual transient heap ownership. Repository completion follows the +//! archive writer's durability boundary independently of the dispatch batch: //! //! repo/not_started -> concurrent archive appends -> archive fsync -//! -> repo completion/failure batch -> listRepos cursor +//! -> every covered repo completion -> listRepos cursor after full drain //! -//! A crash can therefore only turn dispatched repos into `pending`; it can -//! never publish completion metadata ahead of the bytes it describes. +//! A straggler can hold the listRepos cursor but cannot hold already durable +//! repository completions. A crash re-fetches only repositories whose final +//! rows had not crossed a writer durability boundary. const std = @import("std"); const zat = @import("zat"); @@ -80,9 +82,19 @@ const Result = struct { error_class: repo_store.ErrorClass = .unknown, attempts: u32 = 0, started_us: i64 = 0, - /// Monotonic time when a successful completion entered the real worker - /// result queue. It remains queued until the page's archive checkpoint - /// and metadata batch are both durable. + /// Last archive sequence emitted by this repository. Meaningful only when + /// rows > 0. + last_seq: u64 = 0, + /// True when this result came through the repository handler and therefore + /// has an archive watermark, including a successful CAR with zero rows. + /// RepoNotFound is a complete terminal result without that watermark and + /// follows upstream's immediate OnFail persistence path instead. + completion_requires_durability: bool = false, + /// Wall-clock completion time captured when the handler finished. Deferred + /// metadata durability must not rewrite this as the later flush time. + completion_real_us: i64 = 0, + /// Monotonic time when a successful completion entered the writer + /// completion queue. completion_queued_awake_us: i64 = 0, rows: u64 = 0, drops: repos.DropCounts = .{}, @@ -382,20 +394,29 @@ const Work = struct { } if (maybe_result) |raw_result| { var result = raw_result; - if (result.status == .complete) + if (result.status == .complete) { + result.completion_real_us = Io.Timestamp.now(self.io, .real).toMicroseconds(); result.completion_queued_awake_us = Io.Timestamp.now(self.io, .awake).toMicroseconds(); + } self.results.append(self.allocator, result) catch { freeResult(self.allocator, result); if (self.fatal_error == null) self.fatal_error = error.OutOfMemory; return; }; - if (result.status == .complete) if (self.config.metrics) |m| { - _ = m.backfill_completion_queued_total.fetchAdd(1, .monotonic); - _ = m.backfill_completion_queue_depth.fetchAdd(1, .monotonic); - }; } } + fn takeReadyThrough(self: *Work, target: usize) ?std.ArrayList(Result) { + self.result_mu.lockUncancelable(self.io); + defer self.result_mu.unlock(self.io); + while (self.results.items.len == 0 and self.finished < target and self.fatal_error == null) + self.result_ready.waitUncancelable(self.io, &self.result_mu); + if (self.results.items.len == 0) return null; + const out = self.results; + self.results = .empty; + return out; + } + fn publishGauges(self: *Work) void { const m = self.config.metrics orelse return; const gauges = self.budget.gauges(); @@ -422,78 +443,162 @@ fn shuffleJobs(jobs: []Job, seed: u64) void { prng.random().shuffle(Job, jobs); } -fn finishWorkBatch( +/// Stream equivalent of upstream's completionBatcher. Successful repository +/// results are retained until the archive durable hook reports a sequence +/// boundary covering their final row. Empty repositories are eligible at the +/// next hook or forced drain. The listRepos dispatch batch is deliberately not +/// a completion boundary. +const CompletionBatcher = struct { allocator: Allocator, io: Io, config: Config, - archive: *archive_mod.Archive, store: *repo_store.Store, - work: *Work, - jobs: []Job, - stats: *Stats, - shuffle: bool, -) !void { - if (jobs.len > 0 and crashpoint.wouldHit(.mid_backfill_download)) - crashpoint.hit(.mid_backfill_download); + queued: std.ArrayList(Result) = .empty, + accumulated: Stats = .{}, + mu: Io.Mutex = .init, + previous_ctx: ?*anyopaque = null, + previous_hook: ?*const fn (*anyopaque, u64, u64) anyerror!void = null, - if (shuffle) { - shuffleJobs(jobs, @bitCast(Io.Timestamp.now(io, .awake).toMicroseconds())); + fn init(allocator: Allocator, io: Io, config: Config, store: *repo_store.Store) CompletionBatcher { + return .{ .allocator = allocator, .io = io, .config = config, .store = store }; } - for (jobs) |job| try work.submit(job); - const target = work.submittedCount(); - work.waitThrough(target); - if (work.getFatal()) |err| return err; - var results = work.takeResults(); - defer { - for (results.items) |result| freeResult(allocator, result); - results.deinit(allocator); + fn deinit(self: *CompletionBatcher) void { + for (self.queued.items) |result| freeResult(self.allocator, result); + self.queued.deinit(self.allocator); } - // Archive first, metadata second. All worker appends have drained. - if (config.metrics) |m| _ = m.backfill_forced_checkpoint_flushes_total.fetchAdd(1, .monotonic); - try archive.flushBlock(); - var scratch = std.heap.ArenaAllocator.init(allocator); - defer scratch.deinit(); - const alloc = scratch.allocator(); - const transitions = try alloc.alloc(repo_store.Transition, results.items.len); - for (results.items, transitions) |result, *transition| { - const existing = try store.get(alloc, result.did); - const terminal_us = Io.Timestamp.now(io, .real).toMicroseconds(); - transition.* = .{ .did = result.did, .state = terminalState(result, existing, terminal_us) }; + fn install(self: *CompletionBatcher, archive: *archive_mod.Archive) void { + self.previous_ctx = archive.on_durable_ctx; + self.previous_hook = archive.on_durable; + archive.on_durable_ctx = self; + archive.on_durable = onArchiveDurable; } - store.putMany(transitions) catch |err| { - var has_completion = false; - var failed_persists: u64 = 0; - for (results.items) |result| switch (result.status) { - .complete => has_completion = true, - .failed => failed_persists += 1, - else => {}, + + fn restore(self: *CompletionBatcher, archive: *archive_mod.Archive) void { + archive.on_durable_ctx = self.previous_ctx; + archive.on_durable = self.previous_hook; + } + + fn queue(self: *CompletionBatcher, result: Result) !void { + std.debug.assert(result.status == .complete); + std.debug.assert(result.completion_requires_durability); + errdefer freeResult(self.allocator, result); + self.mu.lockUncancelable(self.io); + defer self.mu.unlock(self.io); + for (self.queued.items, 0..) |queued, index| { + if (!std.mem.eql(u8, queued.did, result.did)) continue; + freeResult(self.allocator, queued); + self.queued.items[index] = result; + if (self.config.metrics) |m| { + _ = m.backfill_completion_queued_total.fetchAdd(1, .monotonic); + m.backfill_completion_queue_depth.store(self.queued.items.len, .monotonic); + } + return; + } + try self.queued.append(self.allocator, result); + if (self.config.metrics) |m| { + _ = m.backfill_completion_queued_total.fetchAdd(1, .monotonic); + m.backfill_completion_queue_depth.store(self.queued.items.len, .monotonic); + } + } + + fn persistImmediate(self: *CompletionBatcher, result: Result) !void { + std.debug.assert(!result.completion_requires_durability); + defer freeResult(self.allocator, result); + var scratch = std.heap.ArenaAllocator.init(self.allocator); + defer scratch.deinit(); + const previous = try self.store.get(scratch.allocator(), result.did); + const terminal_us = if (result.status == .complete and result.completion_real_us > 0) + result.completion_real_us + else + Io.Timestamp.now(self.io, .real).toMicroseconds(); + self.store.put(result.did, terminalState(result, previous, terminal_us)) catch |err| { + if (result.status == .failed) { + if (self.config.metrics) |m| + _ = m.backfill_on_fail_store_errors_total.fetchAdd(1, .monotonic); + } + return err; }; - if (config.metrics) |m| { - if (has_completion) _ = m.backfill_completion_stage_errors_total.fetchAdd(1, .monotonic); - if (failed_persists > 0) _ = m.backfill_on_fail_store_errors_total.fetchAdd(failed_persists, .monotonic); + self.mu.lockUncancelable(self.io); + defer self.mu.unlock(self.io); + switch (result.status) { + .complete => { + self.accumulated.completed += 1; + if (self.config.metrics) |m| { + _ = m.backfill_completed_total.fetchAdd(1, .monotonic); + m.backfill_progress_completed.store(@intCast(self.accumulated.completed), .monotonic); + } + }, + .failed => { + self.accumulated.failed += 1; + if (self.config.metrics) |m| _ = m.backfill_failed_total.fetchAdd(1, .monotonic); + }, + else => {}, } - return err; - }; + } + + fn onArchiveDurable(ctx: *anyopaque, durable_seq: u64, next_seq: u64) anyerror!void { + const self: *CompletionBatcher = @ptrCast(@alignCast(ctx)); + if (self.previous_hook) |hook| try hook(self.previous_ctx.?, durable_seq, next_seq); + try self.stageDurable(durable_seq, false); + } + + fn drain(self: *CompletionBatcher, archive: *archive_mod.Archive) !void { + if (self.config.metrics) |m| _ = m.backfill_forced_checkpoint_flushes_total.fetchAdd(1, .monotonic); + try archive.flushBlock(); + try self.stageDurable(archive.committed_seq.load(.acquire), true); + } + + fn stageDurable(self: *CompletionBatcher, durable_seq: u64, force: bool) !void { + self.mu.lockUncancelable(self.io); + defer self.mu.unlock(self.io); + + var eligible: usize = 0; + for (self.queued.items) |result| { + if (result.rows == 0 or result.last_seq <= durable_seq) { + eligible += 1; + } else if (force) { + if (self.config.metrics) |m| _ = m.backfill_completion_stage_errors_total.fetchAdd(1, .monotonic); + return error.BackfillCompletionNotDurable; + } + } + if (eligible == 0) return; + + var scratch = std.heap.ArenaAllocator.init(self.allocator); + defer scratch.deinit(); + const alloc = scratch.allocator(); + const transitions = try alloc.alloc(repo_store.Transition, eligible); + var transition_index: usize = 0; + for (self.queued.items) |result| { + if (result.rows != 0 and result.last_seq > durable_seq) continue; + const previous = try self.store.get(alloc, result.did); + transitions[transition_index] = .{ + .did = result.did, + .state = terminalState(result, previous, result.completion_real_us), + }; + transition_index += 1; + } + self.store.putMany(transitions) catch |err| { + if (self.config.metrics) |m| _ = m.backfill_completion_stage_errors_total.fetchAdd(1, .monotonic); + return err; + }; - // Canonical upstream boundary: archive rows were flushed before the - // synced completion batch above, so a restart may skip this repo - // without losing its segment rows. - for (results.items) |result| - if (result.status == .complete) crashpoint.hit(.after_repo_complete); - - var durable_completions: usize = 0; - const durable_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(); - for (results.items) |result| switch (result.status) { - .complete => { - stats.completed += 1; - stats.rows += result.rows; - stats.drops.invalid_collection += result.drops.invalid_collection; - stats.drops.invalid_rkey += result.drops.invalid_rkey; - stats.drops.field_too_long += result.drops.field_too_long; - durable_completions += 1; - if (config.metrics) |m| { + const durable_awake_us = Io.Timestamp.now(self.io, .awake).toMicroseconds(); + var retained: usize = 0; + for (self.queued.items) |result| { + if (result.rows != 0 and result.last_seq > durable_seq) { + self.queued.items[retained] = result; + retained += 1; + continue; + } + crashpoint.hit(.after_repo_complete); + self.accumulated.completed += 1; + self.accumulated.rows += result.rows; + self.accumulated.drops.invalid_collection += result.drops.invalid_collection; + self.accumulated.drops.invalid_rkey += result.drops.invalid_rkey; + self.accumulated.drops.field_too_long += result.drops.field_too_long; + if (self.config.metrics) |m| { _ = m.backfill_completed_total.fetchAdd(1, .monotonic); _ = m.backfill_drops.getPtr(.invalid_collection).fetchAdd(result.drops.invalid_collection, .monotonic); _ = m.backfill_drops.getPtr(.invalid_rkey).fetchAdd(result.drops.invalid_rkey, .monotonic); @@ -501,21 +606,51 @@ fn finishWorkBatch( if (result.completion_queued_awake_us > 0 and durable_awake_us >= result.completion_queued_awake_us) m.observeBackfillCompletionWait(@intCast(durable_awake_us - result.completion_queued_awake_us)); } - }, - .failed => { - stats.failed += 1; - if (config.metrics) |m| _ = m.backfill_failed_total.fetchAdd(1, .monotonic); - }, - else => {}, - }; - if (config.metrics) |m| { - m.backfill_progress_completed.store(@intCast(stats.completed), .monotonic); - if (durable_completions > 0) { - _ = m.backfill_completion_queue_depth.fetchSub(durable_completions, .monotonic); + freeResult(self.allocator, result); + } + self.queued.items.len = retained; + if (self.config.metrics) |m| { + m.backfill_progress_completed.store(@intCast(self.accumulated.completed), .monotonic); + m.backfill_completion_queue_depth.store(retained, .monotonic); _ = m.backfill_completion_durable_batches_total.fetchAdd(1, .monotonic); - _ = m.backfill_completion_durable_repos_total.fetchAdd(durable_completions, .monotonic); + _ = m.backfill_completion_durable_repos_total.fetchAdd(eligible, .monotonic); + } + } + + fn snapshot(self: *CompletionBatcher) Stats { + self.mu.lockUncancelable(self.io); + defer self.mu.unlock(self.io); + return self.accumulated; + } +}; + +fn finishWorkBatch( + work: *Work, + completions: *CompletionBatcher, + jobs: []Job, + shuffle: bool, +) !void { + if (jobs.len > 0 and crashpoint.wouldHit(.mid_backfill_download)) + crashpoint.hit(.mid_backfill_download); + + if (shuffle) { + shuffleJobs(jobs, @bitCast(Io.Timestamp.now(work.io, .awake).toMicroseconds())); + } + for (jobs) |job| try work.submit(job); + const target = work.submittedCount(); + while (true) { + if (work.getFatal()) |err| return err; + var results = work.takeReadyThrough(target) orelse break; + defer results.deinit(work.allocator); + for (results.items) |result| { + if (result.status == .complete and result.completion_requires_durability) + try completions.queue(result) + else + try completions.persistImmediate(result); } } + if (work.getFatal()) |err| return err; + try completions.drain(work.archive); } pub fn run( @@ -560,7 +695,11 @@ pub fn runTraced( var work_completed = false; defer work.shutdown(work_completed); - var stats: Stats = .{}; + var completions = CompletionBatcher.init(allocator, io, config, store); + defer completions.deinit(); + completions.install(archive); + defer completions.restore(archive); + if (config.metrics) |m| m.backfill_progress_completed.store(0, .monotonic); var selected: usize = 0; // An explicit selection is a self-contained debug run. It must not leave @@ -676,7 +815,7 @@ pub fn runTraced( const batch_due = selected_page or max_reached or final_page or (full_network and batch_entries >= batch_size); if (!batch_due) continue; - try finishWorkBatch(allocator, io, config, archive, store, &work, jobs.items, &stats, !selected_page); + try finishWorkBatch(&work, &completions, jobs.items, !selected_page); if (full_network) try saveListReposCheckpoint( store.meta, if (page.cursor) |next| next else "", @@ -700,7 +839,7 @@ pub fn runTraced( } work.publishGauges(); work_completed = true; - return stats; + return completions.snapshot(); } test "bootstrap identity metadata comes from a real DID resolution response" { @@ -807,11 +946,11 @@ test "explicit selected backfill resolves identity, bypasses listRepos, and pers // there is no CAR to handle, but the DID must not be retried forever. try testing.expectEqual(@as(u64, 1), stats.completed); try testing.expectEqual(@as(u64, 1), instrumentation.backfill_discovered_total.load(.monotonic)); - try testing.expectEqual(@as(u64, 1), instrumentation.backfill_completion_queued_total.load(.monotonic)); + try testing.expectEqual(@as(u64, 0), instrumentation.backfill_completion_queued_total.load(.monotonic)); try testing.expectEqual(@as(usize, 0), instrumentation.backfill_completion_queue_depth.load(.monotonic)); - try testing.expectEqual(@as(u64, 1), instrumentation.backfill_completion_durable_batches_total.load(.monotonic)); - try testing.expectEqual(@as(u64, 1), instrumentation.backfill_completion_durable_repos_total.load(.monotonic)); - try testing.expectEqual(@as(u64, 1), instrumentation.backfill_completion_queue_wait_count.load(.monotonic)); + try testing.expectEqual(@as(u64, 0), instrumentation.backfill_completion_durable_batches_total.load(.monotonic)); + try testing.expectEqual(@as(u64, 0), instrumentation.backfill_completion_durable_repos_total.load(.monotonic)); + try testing.expectEqual(@as(u64, 0), instrumentation.backfill_completion_queue_wait_count.load(.monotonic)); try testing.expectEqual(@as(u64, 1), instrumentation.backfill_completed_total.load(.monotonic)); try testing.expectEqual(@as(usize, 1), instrumentation.backfill_progress_completed.load(.monotonic)); try testing.expectEqual(@as(u64, 1), instrumentation.backfill_forced_checkpoint_flushes_total.load(.monotonic)); @@ -1091,15 +1230,17 @@ fn downloadOne( batch: [1024]segment.Event = undefined, batch_len: usize = 0, rows: u64 = 0, + last_seq: u64 = 0, trace_context: observability.Context, fn flush(self: *@This()) anyerror!void { if (self.batch_len == 0) return; - _ = try self.archive.appendBatchTraced( + const first_seq = (try self.archive.appendBatchTraced( self.batch[0..self.batch_len], self.now_us, self.trace_context, - ); + )).?; + self.last_seq = first_seq + self.batch_len - 1; self.batch_len = 0; } @@ -1148,6 +1289,8 @@ fn downloadOne( .attempts = attempts, .started_us = started_us, .rows = sink.rows, + .last_seq = sink.last_seq, + .completion_requires_durability = true, .drops = local.drops, }; }, @@ -1236,6 +1379,274 @@ test "backfill dispatch batch is shuffled before worker admission" { try testing.expect(!unchanged); } +test "writer durability commits completed repositories independently of a dispatch straggler" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf); + const data_dir = data_dir_buf[0..data_dir_len]; + + var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer archive.deinit(); + archive.max_events_per_block = 1; + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, io, &meta); + defer store.deinit(); + const did_a = "did:plc:completion-a"; + const did_b = "did:plc:completion-b"; + const did_empty = "did:plc:completion-empty"; + try store.putMany(&.{ + .{ .did = did_a, .state = .{ .status = .not_started } }, + .{ .did = did_b, .state = .{ .status = .not_started } }, + .{ .did = did_empty, .state = .{ .status = .not_started } }, + }); + + var instrumentation: metrics_mod.Stats = .{}; + var completions = CompletionBatcher.init(testing.allocator, io, .{ + .relay_http = "", + .data_dir = data_dir, + .metrics = &instrumentation, + }, &store); + defer completions.deinit(); + completions.install(&archive); + defer completions.restore(&archive); + + const Event = struct { + fn make(did: []const u8, rkey: []const u8) segment.Event { + return .{ + .seq = 0, + .witnessed_at = 1_700_000_000_000_000, + .indexed_at = 0, + .kind = .create, + .did = did, + .collection = "app.bsky.feed.post", + .rkey = rkey, + .rev = "3l3qo2vutsw2b", + .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69", + }; + } + }; + var row_a = [_]segment.Event{Event.make(did_a, "3l3qo2vuowo2a")}; + const seq_a = (try archive.appendBatch(&row_a, 1_700_000_000_000_000)).?; + { + const rev_a = try testing.allocator.dupe(u8, "3l3qo2vutsw2b"); + errdefer testing.allocator.free(rev_a); + const latest_a = try testing.allocator.dupe(u8, "3l3qo2vutsw2b"); + errdefer testing.allocator.free(latest_a); + try completions.queue(.{ + .did = did_a, + .status = .complete, + .rev = rev_a, + .latest_rev = latest_a, + .rows = 1, + .last_seq = seq_a, + .completion_requires_durability = true, + .completion_real_us = 1_700_000_000_123_456, + .completion_queued_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(), + }); + } + + // The second repository is the dispatch straggler at this point. Its + // first durable block invokes the archive hook and must commit A without + // waiting for B's worker result or the listRepos cursor checkpoint. + var row_b = [_]segment.Event{Event.make(did_b, "3l3qo2vuowo2b")}; + const seq_b = (try archive.appendBatch(&row_b, 1_700_000_000_000_001)).?; + try archive.flushBlock(); + const state_a = (try store.get(testing.allocator, did_a)).?; + defer store.freeState(testing.allocator, state_a); + try testing.expectEqual(repo_store.Status.complete, state_a.status); + try testing.expectEqual(@as(i64, 1_700_000_000_123_456), state_a.completed_us); + const state_b_before = (try store.get(testing.allocator, did_b)).?; + defer store.freeState(testing.allocator, state_b_before); + try testing.expectEqual(repo_store.Status.not_started, state_b_before.status); + try testing.expect(try loadRelayListReposCursor(&meta, testing.allocator) == null); + + { + const rev_b = try testing.allocator.dupe(u8, "3l3qo2vutsw2b"); + errdefer testing.allocator.free(rev_b); + const latest_b = try testing.allocator.dupe(u8, "3l3qo2vutsw2b"); + errdefer testing.allocator.free(latest_b); + try completions.queue(.{ + .did = did_b, + .status = .complete, + .rev = rev_b, + .latest_rev = latest_b, + .rows = 1, + .last_seq = seq_b, + .completion_requires_durability = true, + .completion_real_us = Io.Timestamp.now(io, .real).toMicroseconds(), + .completion_queued_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(), + }); + } + // RepoNotFound arrives through upstream's OnFail special case rather than + // the CAR handler, so it is a durable immediate terminal transition. + try completions.persistImmediate(.{ + .did = did_empty, + .status = .complete, + .completion_real_us = Io.Timestamp.now(io, .real).toMicroseconds(), + .completion_queued_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(), + }); + try completions.drain(&archive); + + const state_b = (try store.get(testing.allocator, did_b)).?; + defer store.freeState(testing.allocator, state_b); + const state_empty = (try store.get(testing.allocator, did_empty)).?; + defer store.freeState(testing.allocator, state_empty); + try testing.expectEqual(repo_store.Status.complete, state_b.status); + try testing.expectEqual(repo_store.Status.complete, state_empty.status); + try testing.expectEqual(@as(u64, 2), instrumentation.backfill_completion_queued_total.load(.monotonic)); + try testing.expectEqual(@as(u64, 2), instrumentation.backfill_completion_durable_repos_total.load(.monotonic)); + try testing.expectEqual(@as(usize, 0), instrumentation.backfill_completion_queue_depth.load(.monotonic)); + try testing.expectEqual(@as(u64, 3), completions.snapshot().completed); +} + +test "completion queue replaces a duplicate DID without growing its durable set" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf); + const data_dir = data_dir_buf[0..data_dir_len]; + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, io, &meta); + defer store.deinit(); + var instrumentation: metrics_mod.Stats = .{}; + var completions = CompletionBatcher.init(testing.allocator, io, .{ + .relay_http = "", + .data_dir = data_dir, + .metrics = &instrumentation, + }, &store); + defer completions.deinit(); + + for ([_][]const u8{ "rev-old", "rev-new" }) |rev| { + const owned_rev = try testing.allocator.dupe(u8, rev); + errdefer testing.allocator.free(owned_rev); + const owned_latest = try testing.allocator.dupe(u8, rev); + errdefer testing.allocator.free(owned_latest); + try completions.queue(.{ + .did = "did:plc:completion-replaced", + .status = .complete, + .rev = owned_rev, + .latest_rev = owned_latest, + .rows = 1, + .last_seq = 42, + .completion_requires_durability = true, + .completion_real_us = Io.Timestamp.now(io, .real).toMicroseconds(), + .completion_queued_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(), + }); + } + try testing.expectEqual(@as(usize, 1), completions.queued.items.len); + try testing.expectEqualStrings("rev-new", completions.queued.items[0].rev); + try testing.expectEqual(@as(u64, 2), instrumentation.backfill_completion_queued_total.load(.monotonic)); + try testing.expectEqual(@as(usize, 1), instrumentation.backfill_completion_queue_depth.load(.monotonic)); +} + +test "durable repository completion survives reopen before listRepos cursor advance" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf); + const data_dir = data_dir_buf[0..data_dir_len]; + const ready_did = "did:plc:restart-ready"; + const held_did = "did:plc:restart-held"; + + { + var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer archive.deinit(); + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, io, &meta); + defer store.deinit(); + try store.putMany(&.{ + .{ .did = ready_did, .state = .{ .status = .not_started } }, + .{ .did = held_did, .state = .{ .status = .not_started } }, + }); + var completions = CompletionBatcher.init(testing.allocator, io, .{ + .relay_http = "", + .data_dir = data_dir, + }, &store); + defer completions.deinit(); + completions.install(&archive); + defer completions.restore(&archive); + + var ready_event = [_]segment.Event{.{ + .seq = 0, + .witnessed_at = 1_700_000_000_000_000, + .indexed_at = 0, + .kind = .create, + .did = ready_did, + .collection = "app.bsky.feed.post", + .rkey = "3l3qo2vuowo2a", + .rev = "3l3qo2vutsw2b", + .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69", + }}; + const ready_seq = (try archive.appendBatch(&ready_event, 1_700_000_000_000_000)).?; + { + const rev = try testing.allocator.dupe(u8, "3l3qo2vutsw2b"); + errdefer testing.allocator.free(rev); + const latest = try testing.allocator.dupe(u8, "3l3qo2vutsw2b"); + errdefer testing.allocator.free(latest); + try completions.queue(.{ + .did = ready_did, + .status = .complete, + .rev = rev, + .latest_rev = latest, + .rows = 1, + .last_seq = ready_seq, + .completion_requires_durability = true, + .completion_real_us = Io.Timestamp.now(io, .real).toMicroseconds(), + .completion_queued_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(), + }); + } + // held_did emits one row and then deliberately never returns a worker + // result. Its writer flush must still make ready_did durable without + // inventing a completion for held_did or advancing the page cursor. + var held_event = [_]segment.Event{.{ + .seq = 0, + .witnessed_at = 1_700_000_000_000_001, + .indexed_at = 0, + .kind = .create, + .did = held_did, + .collection = "app.bsky.feed.post", + .rkey = "3l3qo2vuowo2b", + .rev = "3l3qo2vutsw2b", + .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69", + }}; + _ = try archive.appendBatch(&held_event, 1_700_000_000_000_001); + try archive.flushBlock(); + const before_restart = (try store.get(testing.allocator, ready_did)).?; + defer store.freeState(testing.allocator, before_restart); + try testing.expectEqual(repo_store.Status.complete, before_restart.status); + try testing.expect(try loadRelayListReposCursor(&meta, testing.allocator) == null); + } + + { + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, io, &meta); + defer store.deinit(); + const ready = (try store.get(testing.allocator, ready_did)).?; + defer store.freeState(testing.allocator, ready); + const held = (try store.get(testing.allocator, held_did)).?; + defer store.freeState(testing.allocator, held); + try testing.expectEqual(repo_store.Status.complete, ready.status); + // Startup recovery moves only the actually interrupted repository to + // the merge-time pending pass; it must not replay ready_did. + try testing.expectEqual(repo_store.Status.pending, held.status); + try testing.expect(try loadRelayListReposCursor(&meta, testing.allocator) == null); + } +} + test "bootstrap redirects preserve independent transient and rate-limit retry budgets" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); -- 2.54.0