diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f319dbea..7ae4ff6b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -138,6 +138,15 @@ jobs: - name: Watchdog Lua CM e2e run: just test-watchdog-e2e + - name: Upload E2E failure logs + if: ${{ failure() }} + uses: actions/upload-artifact@v6 + with: + name: rollups-e2e-logs-${{ github.run_attempt }} + path: tests/e2e/results/*.log + retention-days: 7 + if-no-files-found: ignore + watchdog-docker: name: Watchdog Docker image smoke runs-on: ubuntu-latest diff --git a/README.md b/README.md index 515989c9..7d077933 100644 --- a/README.md +++ b/README.md @@ -167,6 +167,12 @@ Most queue sizes, polling intervals, and safety limits are now internal runtime ## API +JSON `sender` fields in successful `POST /tx` responses and WebSocket messages +use EIP-55 checksum casing. Address fields in `/history` and `sender` fields in +`/historical-l1-inputs` use lowercase hex. Clients must compare decoded 20-byte +addresses and use one normalized encoding for account or projection keys across +these routes. + ### `POST /tx` Request shape: @@ -234,6 +240,8 @@ After each successfully applied input at offset `X`, persist the claim with `HISTORY_UNAVAILABLE`, or `AHEAD_OF_HEAD`. Rebootstrap on a history mismatch. - A claim exactly at the head waits for the next input. Replay uses bounded pages and queues, with no total catch-up limit. The subscriber cap is `64`. +- Before upgrade, capacity exhaustion returns `429 OVERLOADED`; shutdown or an + operational subscription failure returns `503 UNAVAILABLE`. - Messages are JSON text frames; binary fields are `0x`-prefixed hex. Direct-input `block_timestamp` values are Unix seconds. - Batch envelopes are absent. Offsets count executed application inputs, @@ -406,8 +414,23 @@ and `X-Executed-Input-Count`, selected atomically with the artifact lease. Streaming holds the lease until the response ends or the client disconnects. The accepted endpoints return `404` until a comparable checkpoint exists: genesis is comparable at block zero; a rebuilt baseline is restorable but only -a later accepted batch establishes a comparison point. Divergence blocks -publication of the accepted checkpoint. See [snapshot lifecycle](docs/snapshots/lifecycle.md). +a later accepted batch establishes a comparison point. Known divergence makes +all three finalized endpoints return `503 UNAVAILABLE`, including conditional +state requests. The check shares the checkpoint-selection transaction, before +any lease or archive is created. See [snapshot lifecycle](docs/snapshots/lifecycle.md). + +### Health probes (internal only) + +- `GET /livez` returns `200` whenever the handler responds, with an empty body. +- `GET /readyz` returns `200` while the inclusion-lane receiver is open and + shutdown has not been requested; otherwise `503`. Its body is empty. +- `GET /healthz` uses the same status as `/readyz` and returns JSON: + `{ "status": "ok", "inclusion_lane": "ok" }`. `status` becomes `"degraded"` + for either failure condition; `inclusion_lane` becomes `"stopped"` only when + its receiver is closed, so it can remain `"ok"` during shutdown. + +These probes cover process reachability, the lane channel, and shutdown state. +They do not certify L1 freshness, submitter balance, or canonical agreement. ## Storage Model diff --git a/docs/invariants.md b/docs/invariants.md index 5a9b6c6e..0a68a8d9 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -80,12 +80,12 @@ by writer and are write-once (`0001_schema.sql`). |---|---| | inclusion lane | `batches` (insert + `sealed_at_ms`), `frames`, `user_ops`, `application_inputs`, `dumps`/`snapshots` (batch close) | | input reader | `safe_inputs`, `l1_safe_head`, `safe_accepted_batches`, `canonical_divergence` (the divergence poison marker) | -| recovery (startup) | `batches.invalidated_at_ms`, Tip reopen, current `application_inputs` suffix deletion | -| history metadata (setup/recovery) | `history_state` — complete era/application-count/L1-block baseline; generation advance and immutable preserved-prefix cut in a non-empty standard-recovery cascade | +| recovery (startup) | `batches.invalidated_at_ms`, Tip reopen, current `application_inputs` suffix deletion and replacement direct-input rows | +| history metadata (setup/recovery) | `history_state` — complete era/application-count/L1-block baseline and generation; `history_generation_cuts` — immutable preserved-prefix cuts written with non-empty standard-recovery cascades | | batch submitter and mempool flusher | `wallet_nonce_watermark` — deliberately shared under one protocol: each raises it before its first broadcast (write-before-broadcast, I14) | | egress (HTTP) | `dumps.lease_count` (leases); `run`'s startup hygiene resets it to zero as the crash backstop | -| setup | `deployment_identity` (pinned once), `batch_tree_anchor` (the root nonce, frozen once setup completes), the initial `dumps` + `snapshots` rows (genesis or rebuild registration, atomic with the complete history baseline), the `setup_complete` fact (written once), `batch_policy.log_gas_price` + `log_gas_price_updated_at_ms` (first write; Fixed and Uniswap) | -| snapshot GC (the lane after reconciliation, `run`'s startup hygiene) | unreferenced `dumps` row deletion (`gc_unreferenced_dumps`) | +| setup | `deployment_identity` (pinned once), `batch_tree_anchor` (the root nonce, frozen once setup completes), the initial `dumps` + `snapshots` rows and rebuild root `batches`/`frames` (atomic with the complete history baseline), the `setup_complete` fact (written once), `batch_policy.log_gas_price` + `log_gas_price_updated_at_ms` (first write; Fixed and Uniswap) | +| snapshot GC (the lane after reconciliation, `run`'s startup hygiene) | obsolete `snapshots` and unreferenced `dumps` row deletion (`gc_unreferenced_dumps`), including a superseded baseline artifact | | command brackets (run, setup, flush) | `terminal_faults` (append-only, best-effort at settlement) | | admin | `batch_policy` alpha knobs (`log_alpha`, `log_one_plus_alpha`) | | fee oracle | `batch_policy.log_gas_price` + `log_gas_price_updated_at_ms` (Uniswap mode only; stamps on every successful refresh) | @@ -224,9 +224,11 @@ by writer and are write-once (`0001_schema.sql`). remedy is cockroach recovery. - **Completeness boundary:** the check completely enforces the accepted-batch identity predicate above; it is intentionally not a general canonical/application - divergence oracle. It trusts collapsed history below the anchor and the - checkpoint application state, shares `scheduler_accepts` (including its - documented self-trust omissions), and does not independently detect bugs in + divergence oracle. The entire L1 prefix through baseline block `C` is opaque, + including previously rejected future-nonce batches; the check trusts the + checkpoint state and continuation nonce instead of reinterpreting that prefix. + It shares `scheduler_accepts` (including its documented self-trust omissions), + and does not independently detect bugs in direct-input/user-op execution. A wrong-high cockroach checkpoint nonce is a known example that can escape it. Absence of the marker therefore does not prove global agreement. Conversely, a structurally malformed foreign landing @@ -253,9 +255,11 @@ by writer and are write-once (`0001_schema.sql`). boundary selects external directs by the setup-pinned submitter address; only those inputs and included user ops enter `application_inputs`. - **Enforced by:** classified direct reads and complete receipt validation at - append. Startup/recovery derive the initial direct rows before catch-up, - which must execute them successfully before admission. Replay and WS need - no envelope filter because every row executes. + append. Standard startup recovery attributes undrained directs to the new Tip; + lane catch-up executes them before processing queued user operations. Manual + rebuild represents the folded prefix through `C` in its baseline snapshot, + with no application-history rows for that prefix. Replay and WS need no + envelope filter because every row executes. - **Depended on by:** application replay and replicated state correctness. ### I12. Safe head advances only on real observation; `synced_at_ms` is genuine progress time @@ -341,9 +345,12 @@ by writer and are write-once (`0001_schema.sql`). conflicting batch-tree writes; the detector and next typed read stop the process. A chunk committed before either runtime observation may acknowledge and later roll back. -- **Watchdog boundary:** the freeze blocks accepted-checkpoint publication before the - offending landing becomes a comparable sequencer checkpoint. Because the - watchdog skips replay when the finalized inclusion block is unchanged, it +- **Watchdog boundary:** accepted-checkpoint selection checks for divergence + in the same transaction as selection and any download lease, refusing while + the marker is present. A matching batch + before a divergent acceptance in the same L1 block cannot represent that + block's final state. Because the watchdog skips replay when the finalized + inclusion block is unchanged, it does not subsume this wire-identity detector. Conversely, the check does not subsume the watchdog's broader independent application-state comparison. @@ -429,10 +436,14 @@ by writer and are write-once (`0001_schema.sql`). before replacement directs. The entire transition commits in the cascade transaction. Clean restart changes neither token. Every intervening cut is required to authorize reusing a checkpoint from an older generation. -- **Enforced by:** `complete_baseline_setup`, immutable history triggers, - exact-`+1` generation trigger, and `cascade_and_reopen`. -- **Depended on by:** mandatory snapshot-derived WS claims. Identity is validated - before the requested count, including for empty history. +- **Enforced by:** `complete_baseline_setup`, immutable baseline and + `history_generation_cuts` triggers, the exact-`+1` generation trigger requiring + its cut, and `cascade_and_reopen`. `preserved_input_count_in` asserts that + every intervening generation has a cut before computing compatibility. +- **Depended on by:** mandatory snapshot-derived WS claims and `/history` + checkpoint compatibility across standard recoveries. Identity is validated + before the requested count, including for empty history. Cuts remain available + for the era's lifetime; their absence must never authorize a partial minimum. - **Breaks:** a client silently resumes a replaced suffix or inaccessible prefix. - **Operational boundary:** rebuilding uses a fresh/wiped data directory. Checkpoint state, inclusion block, and next nonce are trusted operator inputs; diff --git a/docs/plans/2026-07-coordination-tracks.md b/docs/plans/2026-07-coordination-tracks.md index 75c8581b..7f591d78 100644 --- a/docs/plans/2026-07-coordination-tracks.md +++ b/docs/plans/2026-07-coordination-tracks.md @@ -76,7 +76,10 @@ Remaining checks need the actual consumer: for production readiness: the old native state is unavailable, the exported bundle restores correctly, and execution after rebuild matches the canonical machine. For the DEX, pin the designated state drive/memory region and derive - resume metadata from canonical execution. Add the integration check to the + resume metadata from canonical execution. The exporter must check + [pending-direct eligibility](../recovery/cockroach.md#checkpoint-eligibility) + and the drill must exercise refusal and earlier-checkpoint fallback, alongside + eligible pending-queue recovery. Add the integration check to the release validation once the actual artifacts are available; no generic trait or deployment gate currently enforces this requirement. - Repeat snapshot-to-live bootstrap and canonical comparison with the private diff --git a/docs/protocol/application-contract.md b/docs/protocol/application-contract.md index 3bc28a28..24a45461 100644 --- a/docs/protocol/application-contract.md +++ b/docs/protocol/application-contract.md @@ -209,6 +209,10 @@ layout, and extraction procedure for each supported image. Other applications may require a different mapping. The recovery bundle also needs the exact L1 boundary and next scheduler nonce, obtained from trusted canonical execution; these are separate from merely extracting application bytes. +The exporter must also verify the canonical pending-direct queue satisfies +[checkpoint eligibility](../recovery/cockroach.md#checkpoint-eligibility). +An accurate application clock and `A < B` do not prove that condition after +faulty sequencing. Each integration supplies a versioned export command and operator procedure, and demonstrates recovery from a non-genesis canonical checkpoint before diff --git a/docs/protocol/projection-replay.md b/docs/protocol/projection-replay.md index 33f582f6..7278c485 100644 --- a/docs/protocol/projection-replay.md +++ b/docs/protocol/projection-replay.md @@ -66,8 +66,11 @@ share a count, and generations can reuse replaced offsets. Such a backup contains core state and projection at count `X`, inclusion block `B`, next scheduler nonce `N`, and the application's own clock `A`. The -[manual recovery contract](../recovery/cockroach.md#replay-boundaries) requires -`A < B`, except known empty genesis, and `B <= C` for the target rebuild: +[manual recovery contract](../recovery/cockroach.md#checkpoint-eligibility) +requires no pending canonical direct at or below `A`, as well as `A < B` +(except known empty genesis) and `B <= C` for the target rebuild. Establish +queue eligibility against the canonical checkpoint; application-state equality +and the scalar bounds alone cannot prove it after faulty sequencing: 1. Independently establish trust in the backup, projection implementation, and checkpoint boundary under the [incident playbook](../recovery/cockroach.md#application-specific-reader-state). diff --git a/docs/recovery/cockroach.md b/docs/recovery/cockroach.md index 38d07d15..34be82c6 100644 --- a/docs/recovery/cockroach.md +++ b/docs/recovery/cockroach.md @@ -9,7 +9,7 @@ rebuilding.** The operator initiates recovery; the command automates the rebuild The procedure is **flush → fold → fill**: 1. **Flush** outstanding submitter transactions and choose a fixed safe L1 - stopping block. + stopping block at or after the trusted checkpoint's inclusion block. 2. **Fold** the input history through the canonical scheduler, starting from the trusted checkpoint. Every input receives its normal scheduler treatment: accepted batches execute, malformed or rejected batches are skipped, and @@ -45,7 +45,8 @@ The application runbook must name: - How to select and preserve a trusted CM checkpoint and its exact L1 boundary. - The command extracting the application state, count/clock, and next scheduler nonce into the bundle accepted by `setup --recovery`, including supported - checkpoint boundaries and the loader's `A < B` requirement below. + checkpoint boundaries and verification of the + [pending-direct eligibility condition](#checkpoint-eligibility) below. - Artifact locations, access/backup procedures, validation commands, and the commands to rebuild, restart, compare, and resume affected readers. - A rehearsed fallback to an earlier trusted checkpoint or genesis if the @@ -56,6 +57,9 @@ pinned deployment data, and L1 access, while the old native database and dumps are unavailable. Run the actual exporter and restore its output; check the native application bytes/progress and scheduler nonce against the canonical source. Exercise directs pending at the checkpoint and inputs arriving after it. +The exporter must refuse a checkpoint with a pending direct at or below its +application clock, even when `A < B`; rehearse the earlier-checkpoint/genesis +fallback. Also cover an eligible nonempty queue whose directs are all above `A`. Run `setup --recovery` in a fresh directory, resume sequencing, and compare against independent canonical execution after a new batch is accepted. The terminal-drained baseline itself need not equal canonical state at `C`. @@ -95,9 +99,11 @@ latest possible sound checkpoint is not a prerequisite for recovery. state without establishing that all earlier executions were correct. 4. **Prepare a restorable native bundle.** Validate the candidate's restored application state, embedded count/clock, and next batch nonce against the - canonical reference at block `B`. Keep the artifact and its boundary metadata - together and record how its trust was established. The receipt alone is not - evidence that a faulty sequencer executed correctly. + canonical reference at block `B`. Verify its pending-direct queue meets + [checkpoint eligibility](#checkpoint-eligibility); otherwise choose an + eligible earlier checkpoint or genesis. Keep the artifact and its boundary + metadata together and record how its trust was established. The receipt alone + is not evidence that a faulty sequencer executed correctly. 5. **Rebuild in a fresh directory.** Use the invocation below. Recovery chooses its post-flush stopping block `C`, replays from the trusted checkpoint, and publishes a new era. Preserve that baseline artifact and its metadata for @@ -169,7 +175,9 @@ cargo run -p wallet-sequencer -- setup --recovery \ Recovery signs L1 transactions, so the key must match the configured submitter. After success, start `run` with that same data directory. A completed rebuild refuses another `setup --recovery`; failures before completion publish no partial -baseline. +baseline. If the RPC node has not reached the checkpoint, recovery exits with +retryable code 20. Synchronize that node and retry with the same checkpoint and +incomplete data directory. ## Implementation contract @@ -189,12 +197,42 @@ The replay boundaries are: | `S'`, `N'` | Recovered state and next batch nonce. | | `K` | Application count in `S'`; the first later application input has offset `K`. | +### Checkpoint eligibility + +At the exact end-of-block boundary `B`, the canonical scheduler must have **no +pending direct with inclusion block `<= A`**. The checkpoint therefore already +accounts for every direct through `A`; all remaining directs are reconstructed +from `(A, B]`. The exporter must inspect the canonical queue to establish this +condition, including when validating a retained native artifact against the CM. +If it fails, refuse the export and select an eligible earlier checkpoint or +genesis. + +Honest live sequencing establishes this condition: a frame's safe block precedes +its L1 inclusion, so all directs it covers have already arrived. The overdue-direct +backstop preserves it too: by the time a block's directs are overdue, that whole +block has been observed. The FIFO drain executes all equally aged directs from +that block and all older ones, so advancing `A` leaves none of them pending. + +The canonical scheduler also accepts equality. After faulty sequencing, a batch at +block 10 can execute at clock 10 before another direct in that block arrives. +An empty batch at block 11 advances the nonce without draining that direct. +The truthful checkpoint has `A=10 < B=11`, yet seeding `(A,B]` would omit it. +Application-state equality and `A < B` alone cannot establish eligibility. + Loading checks the receipt's block against configured `B` and its nonce against `info.toml`. It requires `A < B`, except for known empty genesis (`B`, nonce, and application count all zero). At non-genesis `A = B`, a direct arriving after the accepted batch in block `B` could still be pending but disappear from the seed -range. Checkpoint state and nonce remain operator-trusted; the later -content-identity check does not verify this prefix. +range. The bundle contains no scheduler queue evidence, so the loader cannot +verify the pending-direct condition. Eligibility, checkpoint state, and nonce +remain operator-trusted; the later content-identity check does not verify this +prefix. Exporter enforcement belongs to the application's required recovery +integration, not the generic loader. + +The complete ordering is `A < B <= C`, with `A = B = 0` allowed for empty +genesis. Before sourcing or executing the fold, recovery requires `B <= C`. +Otherwise publishing the checkpoint state at an earlier baseline block could +make normal reconciliation execute already-accounted direct inputs again. ### Flush and stopping block @@ -208,13 +246,20 @@ After flushing, raw L1 ingestion must reach at least `C`. It may advance farther but the fold stops at `C`. Accepted-batch projection is deferred until the new baseline and batch tree exist. +A trusted checkpoint can be ahead of an honest replacement node that is still +synchronizing. A successful flush only settles the wallet slots known to that +node; it does not establish `C >= B`. If `C < B`, recovery refuses with retryable +exit 20, even when the later re-sync head has reached `B`: that newer observation +does not replace the fixed stopping block. It publishes no baseline and must be +retried after the node catches up. + ### Replay boundaries Seed the scheduler's pending-direct queue from `(A, B]`, excluding inputs sent by the batch submitter. Then replay **all raw inputs** in `(B, C]` in L1 order with expected nonce `N`. Drain the remaining directs through `C` to obtain `(S', N')`. -The disjoint ranges preserve pending directs without executing the checkpoint's -accepted batches again. +For an [eligible checkpoint](#checkpoint-eligibility), the disjoint ranges +preserve pending directs without executing its accepted batches again. On the first `run` sync, acceptance starts at nonce `N'` and scans only blocks **strictly after `C`**. Nonce filtering alone would let a previously rejected diff --git a/docs/review/2026-09-16-track3-validation.md b/docs/review/2026-09-16-track3-validation.md index 56aeaa0e..b5491328 100644 --- a/docs/review/2026-09-16-track3-validation.md +++ b/docs/review/2026-09-16-track3-validation.md @@ -5,6 +5,13 @@ Scope: validate application-history commit a complete cold replica, and a same-host latency comparison against `91e25780854bb641c63135751f951f9f7ee1e744`. +These are the measured pre-rebase revisions. Their stack counterparts are +`799a5d3` → `3e5b971` and `91e2578` → `71b3b35`; the validation/test commit +`4fc010c` became `7f3229f`. The latter stack includes upstream changes, so these +measurements must not be attributed to its rebased trees. The latency baseline +already contains the initial versioned-history foundation; it isolates the +subsequent application-history refactor, not the cost of the complete stack. + Retained for the [Track 3 integration gates](../plans/2026-07-track3-feed-replay-design.md): this is the wallet baseline against which native-engine and deployment results can be assessed. Replace or delete it when those decisions no longer use these diff --git a/docs/review/2026-09-18-stack-review-validation.md b/docs/review/2026-09-18-stack-review-validation.md new file mode 100644 index 00000000..b4845fd3 --- /dev/null +++ b/docs/review/2026-09-18-stack-review-validation.md @@ -0,0 +1,76 @@ +# Stack review closeout validation + +Evidence for reviewing and landing the closing change on +`codex/stack-review-fixes`, above reviewed tip +`62ec150f18a27697220158f1ca4d84ec4a6fee06`. The implementation and regression tests +are at `18e99e4fe2d03426954f72df5641196d12103e99`; the closing documentation commit +adds contract corrections, this record, and a storage rustdoc correction only. +Retire this record after stack landing when no ongoing review decision uses it. + +## Environment and results + +Run on macOS arm64 through the project's parent Nix/direnv shell. Cargo and +rustc were both 1.95.0, Anvil was 1.5.1, and the locally rebuilt canonical guest +used the repository-pinned Cartesi Machine 0.20.0. Local Anvil differs from CI's +1.4.3 pin; these are local results, not a new CI run. + +| Check | Result | +|---|---| +| `cargo check --locked --workspace --all-targets` | Passed | +| `cargo fmt --all -- --check` and `git diff --check` | Passed | +| `cargo clippy --locked --workspace --all-targets --all-features -- -D warnings` | Passed | +| `cargo test --locked --workspace --exclude canonical-test -- --test-threads=1` | 754 passed; one existing ignored doc test | +| `lua watchdog/tests/run.lua` | 62 passed | +| `just canonical build-machine-image`, then `cargo run --locked -p canonical-test` | Fresh image; 10 guest scheduler tests passed | +| `bash scripts/ci-c-application-smoke.sh` | External archive, generic host, and independent downstream consumer built and ran their CLI smoke checks | + +The process binaries were rebuilt after the final SDK change. Each of these +scenario filters passed for both the Rust wallet host and the C wallet host: + +- `cold_replica_snapshot_backlog_live_recovery_test` +- `restart_and_replay_test` +- `recovery_after_stale_batches_test` +- `setup_recovery_round_trip_test` + +These eight process runs include canonical watchdog comparison after recovery +and rebuild. Their prerequisites were the checksum-verified rollups-contracts +Anvil fixture, locally built test contracts, and watchdog Lua dependencies. +The first attempted process run lacked the fixture and stopped before startup; +setup supplied it before the successful runs. + +## Discriminating regressions and independent review + +- A matching local batch followed by a foreign accepted batch in the same block + returned HTTP 200 before the finalized-selection guard. The canonical scheduler + advances beyond that local snapshot. With the guard, all three finalized routes + return 503, including a conditional state request, without acquiring a lease. +- With checkpoint clock 900, checkpoint block 901, and stop 899, the unguarded + deterministic rebuild phase reached artifact creation. The new check refuses + with exit 20 before replay or publication. A later resync reaching the checkpoint + does not substitute for the fixed stop; equality and genesis still succeed. +- Without the SDK header timeout, the stalled-header regression exceeded its + outer two-second limit. With the configured header deadline restored, all + 13 SDK tests passed, including a body that outlives that deadline. + +Separate reviewers checked finalized selection and streamed I/O classification, +and the manual-recovery guard and exit classification. No blocking findings +remained. Persistent SQLite errors still reach terminal classification through +the new error wrapper; operational filesystem errors remain nonterminal. + +## Limits and landing work + +A repeated parallel host run reproduced the already registered +`dropped_runtime_scope_keeps_lock_until_detached_worker_stops` failure: final +reacquisition returned `Locked`. Its isolated rerun passed, and the complete +serial suite passed. This does not resolve the concurrency investigation; it +remains in the [review register](register.md#bounded-investigations-and-cleanup). + +The full 49-scenario rollups suite and Sepolia-image scenarios were not rerun. +Private-engine conformance/export and representative deployment capacity remain +in their existing integration plans. The two current recovery TLA+ models were +read but not changed or rerun; neither models these new boundary checks. + +No lower stack branch was rewritten. Reconcile the known C-host test ancestry +once at landing, retaining the evolved coverage at the reviewed tip and the +new independent post-recovery assertions. No push, PR creation, or merge was +performed during this implementation. diff --git a/docs/review/2026-09-19-review-followup-validation.md b/docs/review/2026-09-19-review-followup-validation.md new file mode 100644 index 00000000..8b1a0063 --- /dev/null +++ b/docs/review/2026-09-19-review-followup-validation.md @@ -0,0 +1,82 @@ +# Review follow-up validation + +Evidence for the ongoing stack review and landing, covering changes above +`f504c2e88e2b432718493e29e3c56f74fce3ad00` on `codex/stack-review-fixes`. +Local validation below includes the filesystem-identity fix at +`8b6da04374c7a4aef61fba8e7963e71d3cd55122`, CI diagnostics at +`93e6a49ad19c35fe78f4c46d639eb8fd556c58ba`, and the accompanying documentation +clarifications. Those clarifications change no executable behavior. +Retire this record after landing when no ongoing review decision uses it. + +## Changes and discriminating checks + +- Startup compares filesystem device/inode identities after inspecting every + retained reference. Regressions preserve relative/absolute, leading-dot, + symlink, and macOS firmlink aliases, including mixed literal/aliased references, + while removing genuine orphans. The firmlink test fails against the old + canonical-path comparison because it deletes the registered artifact, and + passes with identity matching. A missing reference stops the sweep before any + deletion. Dangling orphan links are removed without following their targets. + All eleven startup hygiene tests pass. The firmlink regression runs on macOS; + no Linux bind-mount scenario was run locally. +- A scheduler/fold regression establishes that `A < B` can coexist with a + pending direct at `A` after a same-block frame and a later empty batch. It + verifies recovery from an eligible earlier checkpoint preserves every direct, + application progress, and the scheduler nonce. The + [checkpoint contract](../recovery/cockroach.md#checkpoint-eligibility) requires + canonical exporters to inspect the pending queue and refuse such candidates. +- The accepted-prefix fixture now places the old future-nonce input both below + and exactly at `C`, before the nonce-0 batch in L1 order. Temporarily changing + the production query from `> C` to `>= C` fails specifically at the equality + case. Restoring `>` passes all four acceptance-projection tests. +- API documentation states the existing address-casing/normalization contract, + WS admission statuses, health-probe semantics, and internal deployment + boundary. Writer ownership and stale source references were corrected. The + submitter's own-sender decode error remains visible under self-trust. +- CI retains harness logs from failed rollups E2E jobs for seven days. This adds + evidence for the unclassified C-host recovery WebSocket reset in the + [register](register.md), without changing test or recovery behavior. + +Separate reviewers examined snapshot cleanup and checkpoint eligibility. Review +caught the dangling-orphan-link case before final validation; its regression is +included. No findings remain within the implemented scope. + +## Validation + +macOS arm64, parent Nix/direnv environment, Rust and Cargo 1.95.0: + +| Check | Result | +|---|---| +| `cargo test --locked --workspace --exclude canonical-test -- --test-threads=1` | 760 passed; zero failed; one existing ignored harness doc test | +| `cargo check --locked --workspace --all-targets` | Passed | +| `cargo clippy --locked --workspace --all-targets --all-features -- -D warnings` | Passed | +| `cargo fmt --all -- --check` and `git diff --check` | Passed | +| Relative Markdown links and code fences across repository Markdown files | 409 targets/anchors checked | +| CI workflow YAML parsing | Passed; artifact upload itself requires a failed GitHub job | + +The full host suite includes the new regressions. Guest execution, standalone +rollups E2Es, watchdog Lua tests, and TLA+ model runs were not repeated for this +follow-up. Both current recovery models were read; their modeled transitions +are unchanged. Earlier validation remains in the +[initial closeout record](2026-09-18-stack-review-validation.md). + +## Remaining integration and landing work + +The bundle carries no canonical queue evidence, and no canonical-to-native +exporter is implemented here. The eligibility change is a supported-checkpoint +precondition and executable scheduler regression, not loader enforcement or a +completed CM export drill. The application's exporter must enforce refusal and +demonstrate fallback and eligible pending-queue recovery; the +[integration plan](../plans/2026-07-coordination-tracks.md#track-6--dump--application-api-redesign) +tracks that work. Canonical acceptance semantics and artifact formats are unchanged. + +Archive concurrency limits remain deferred pending deployment workload needs. +The existing process-lock concurrency investigation remains open; this host run +used the serial suite. The C-host WebSocket reset is a separate unresolved +investigation; one successful run cannot classify its cause. + +The lower-stack ancestry is reconciled: PR #38 is at +`7f3229f2e42585e055f2fabb8e280c4d42dd5a81`, and GitHub reports PR #42 mergeable +at `35697691d7a5ba5a2c868f51b3a45c3dd5b6ee44` (checked 2026-09-19). +Consult CI for the pushed revision before landing; local host checks do not +replace guest, watchdog, or standalone rollups E2E execution. diff --git a/docs/review/register.md b/docs/review/register.md index 238f2527..49ce13d3 100644 --- a/docs/review/register.md +++ b/docs/review/register.md @@ -58,6 +58,20 @@ exposure in an actual deployment was established by this review. passes. The worker drops its scope before signalling completion, so a simple worker-completion race does not explain the failure. Identify any remaining descriptor/process ownership before changing the assertion or lock behavior. + Reproduced during the 2026-09-18 stack closeout; isolated and full serial + runs passed. See the [current validation record](2026-09-18-stack-review-validation.md). +- **Intermittent C-host recovery WebSocket reset.** At `dc4dd78` on + 2026-09-19, `c_host_recovery_after_stale_batches_test` failed an expected + message receive with `Connection reset without closing handshake` in the + [push run](https://github.com/cartesi/sequencer/actions/runs/35456024614/job/105931662252). + The [PR run](https://github.com/cartesi/sequencer/actions/runs/35456514750/job/105934071941) + passed all 49 scenarios on the same source tree. The failed run retained no + child-process log artifact, so the reset's cause is unclassified. CI now + uploads `tests/e2e/results/*.log` on failure. On recurrence, use those logs + to identify the server's last events and the failing receive before changing + timeouts, teardown, or retry behavior. Evidence: the + [recovery scenario](../../tests/e2e/src/test_cases.rs) and + [WS receive helper](../../tests/harness/src/ws.rs). - **Transient SQLite contention stops the submitter.** Read handles use a 50 ms busy timeout; a storage/open failure escapes the submitter loop. BUSY/LOCKED are nonterminal but project to unclassified exit 1, causing diff --git a/docs/snapshots/lifecycle.md b/docs/snapshots/lifecycle.md index 0e9cd3f7..1cf78d96 100644 --- a/docs/snapshots/lifecycle.md +++ b/docs/snapshots/lifecycle.md @@ -44,9 +44,10 @@ Setup similarly makes the baseline artifact durable before publishing the complete era baseline, recovery root when applicable, and setup-completion facts atomically. -Restart selects the newest snapshot on the valid batch branch, or the baseline -when no batch snapshot exists. The selected artifact and stored application -count come from one row. Catch-up checks the restored engine's count and replays +Restart requires the newest valid closed batch's snapshot, or the baseline +when no valid closed batch exists. A missing required snapshot fails loud. +The selected artifact and stored application count come from one row. +Catch-up checks the restored engine's count and replays application inputs from that count. Invalidated branches are excluded by the same valid-batch relation used elsewhere. @@ -57,6 +58,12 @@ must exist; storage refuses a missing required row instead of falling back to an older snapshot. Acceptance already includes scheduler validation and local content identity, so merely observing an own-sender L1 input is insufficient. +A persisted canonical-divergence marker refuses accepted-checkpoint selection, +including when a matching batch precedes a divergent acceptance in the same +block. Selection checks the marker in the same SQLite transaction as its read +and any download lease. Finalized endpoints return HTTP 503 before consulting +conditional cache headers; they do not fall back to an older artifact. + This selection is independent of the lane's L1 reconciliation cursor. A crash between reader ingestion and lane reconciliation cannot miss a promotion or repeat one: acceptance is already durable, and the query derives the result. @@ -128,8 +135,13 @@ of all checkpoint resources. Filesystem deletion failure leaves a harmless orphan for the startup sweep. The reverse ordering would leave a durable row pointing at missing state and is forbidden. Startup clears leases left by the dead process, checks the rollback checkpoint's `info.toml` and format version, collects obsolete -rows, and sweeps orphan directories before workers start. Application restoration -runs afterward in the launched inclusion lane, before processing new user ops; -the metadata check does not validate the application bytes. Missing or corrupt -referenced artifacts fail loud when read or restored; operational filesystem -errors retain their normal error classification. +rows, and sweeps orphan directories before workers start. On supported Linux and +macOS hosts, the sweep reads every retained artifact's filesystem object identity +(device and inode) before deleting any orphans. Comparing those identities +preserves alternate spellings, symlinks, and mount aliases of the same artifact. +Retained identity lookup failures stop the sweep before deletion: missing +references or structural I/O errors are terminal; operational I/O errors retain +their normal classification. Application restoration runs afterward in the launched inclusion +lane, before processing new user ops; the metadata check does not validate the +application bytes. Missing or corrupt referenced artifacts also fail loud when +read or restored. diff --git a/docs/threat-model/README.md b/docs/threat-model/README.md index 08df4af2..6db808c9 100644 --- a/docs/threat-model/README.md +++ b/docs/threat-model/README.md @@ -67,8 +67,10 @@ blocking production diagnostics would require revisiting that assumption batch/frame spine is re-inspected by the runtime danger detector within seconds of launch. The accepted residual is narrow: corrupt payload bytes in rows at/below the lane's resume checkpoint re-trip only when the WS - feed pages them (bounded by its catch-up window) or the submitter - re-encodes a pending batch, and a fault with no durable evidence (a panic + feed pages them or the submitter re-encodes a pending batch. The feed can + read the whole available history from era base `K` in bounded pages, with no + total catch-up limit; this detection path requires a subscriber to read the + affected rows. A fault with no durable evidence (a panic whose trigger does not recur) does not re-trip at all. The window is entered by restarting after a terminal exit (including supervisors that restart regardless of exit status), and @@ -76,7 +78,7 @@ blocking production diagnostics would require revisiting that assumption rollbackable soft confirmations, the watchdog byte-compare, and the I15 divergence freeze. - **Adversarial mempool:** reorder, delay, drop, selective inclusion by builders -- **Zombie transactions:** a submitted batch may sit in a private mempool indefinitely and land long after we believed it was gone. Two load-bearing defenses: the recovery flusher consumes every wallet-nonce slot this deployment ever used (anchored by the persisted watermark, I14) so zombies cannot claim them; and the content-identity check (I9/I15) compares every at/above-anchor *simulated-accepted* landing against the valid closed batch we sealed at that nonce. A foreign or byte-different landing records divergence when it becomes safe and is ingested, freezes the accepted frontier, and requires cockroach recovery. This is trust-boundary validation of external input (the mempool replaying our own stale transactions at times we don't control), not defense-in-depth against self-bugs or a general canonical-state oracle. In cockroach recovery the watermark does not survive the wipe, so that flush is best-effort by construction; the content-identity check is what keeps the residual zombie detected-and-frozen rather than silent (see [the flush boundary](../recovery/cockroach.md#flush-and-stopping-block)). +- **Zombie transactions:** a submitted batch may sit in a private mempool indefinitely and land long after we believed it was gone. Two load-bearing defenses: the recovery flusher consumes every wallet-nonce slot this deployment ever used (anchored by the persisted watermark, I14) so zombies cannot claim them; and the content-identity check (I9/I15) compares every *simulated-accepted* landing strictly after baseline block `C` against the valid closed batch we sealed at that nonce. The complete prefix through `C` is opaque and is never reinterpreted using the recovered nonce. A foreign or byte-different landing records divergence when it becomes safe and is ingested, freezes the accepted frontier, and requires cockroach recovery. This is trust-boundary validation of external input (the mempool replaying our own stale transactions at times we don't control), not defense-in-depth against self-bugs or a general canonical-state oracle. In cockroach recovery the watermark does not survive the wipe, so that flush is best-effort by construction; the content-identity check is what keeps the residual zombie detected-and-frozen rather than silent (see [the flush boundary](../recovery/cockroach.md#flush-and-stopping-block)). - L1 reorgs up to safe depth - Malicious `POST /tx` callers: malformed signatures, spoofed sender, replay across chains or apps, nonce manipulation - Malicious direct-input senders: arbitrary payload, any intent; sender authenticity is guaranteed by InputBox diff --git a/docs/watchdog/design-notes.md b/docs/watchdog/design-notes.md index 5c2e9ec8..c00b8ba0 100644 --- a/docs/watchdog/design-notes.md +++ b/docs/watchdog/design-notes.md @@ -44,16 +44,18 @@ accepted-batch wire-identity detector, and neither mechanism subsumes the other. The content-identity check runs inside the input reader's atomic -safe-input sync. For every -at/above-anchor landing the mirrored scheduler accepts, it requires a -byte-identical valid local sealed batch at that nonce. A foreign or mismatched +safe-input sync. For every landing strictly after baseline block `C` that +the mirrored scheduler accepts, it requires a byte-identical valid local sealed +batch at that nonce. A foreign or mismatched landing persists `canonical_divergence`, which freezes the accepted frontier -and the selection of newer accepted comparison checkpoints +and prevents accepted comparison checkpoint selection ([I15](../invariants.md#i15-divergence-marker-present--acceptance-frontier-frozen)). -The offending landing therefore normally never produces a newer -`/finalized_state/inclusion_block` for the watchdog to compare. Under the -unchanged-head optimization above, a watchdog tick legitimately exits idle. Distinct wire bytes can also be application-state -equivalent, which a byte comparison of resulting snapshots would not expose. +The prefix through `C` is opaque; historical landings are not reconsidered using +the recovered nonce. Once the marker is committed, finalized endpoints refuse +with HTTP 503, including when a matching batch preceded the divergent landing +in the same block. The watchdog cannot compare that block through these routes. +Distinct wire bytes can also be application-state equivalent, which a byte +comparison of resulting snapshots would not expose. Conversely, the content-identity check shares the sequencer's off-chain acceptance predicate and does not independently replay application execution. The watchdog can catch diff --git a/docs/watchdog/operator-deployment.md b/docs/watchdog/operator-deployment.md index e7c3c7d4..10b8082e 100644 --- a/docs/watchdog/operator-deployment.md +++ b/docs/watchdog/operator-deployment.md @@ -9,20 +9,20 @@ For **local development only** (Anvil + `sequencer-devnet`, CI smoke tests), use ## Two deployment tiers ```text - ┌──────────────────────────────────────────────┐ - Internet / users │ Public ingress (POST /tx, GET /fee, WS) │ ← benchmarks, wallets - └──────────────────────┬───────────────────────┘ - │ - ┌─────────────────▼───────────────────┐ - Operator network │ Sequencer process │ - │ + internal snapshot HTTP │ ← watchdog ONLY here - │ /finalized_state* │ - └─────────┬───────────────┬─────────┘ - │ │ - ┌─────────▼───┐ ┌───────▼────────┐ - │ L1 (Sepolia │ │ Watchdog host │ - │ or mainnet)│ │ (compare) │ - └─────────────┘ └────────────────┘ + ┌─────────────────────────────────────┐ + Internet / users │ Public ingress: POST /tx, GET /fee │ ← wallets + └──────────────────┬──────────────────┘ + │ + ┌──────────────────▼──────────────────┐ + Operator network │ Sequencer process │ + │ Internal WS, history, and health │ ← indexers and probes + │ Snapshots: /finalized_state* │ ← watchdog + └─────────┬─────────────────┬─────────┘ + │ │ + ┌─────────▼────────┐ ┌──────▼─────────┐ + │ L1 (Sepolia │ │ Watchdog host │ + │ or mainnet) │ │ (compare) │ + └──────────────────┘ └────────────────┘ ``` The watchdog independently replays L1 through the canonical CM and compares diff --git a/sdk/rust-client/src/errors.rs b/sdk/rust-client/src/errors.rs index cd7ed739..cabe5673 100644 --- a/sdk/rust-client/src/errors.rs +++ b/sdk/rust-client/src/errors.rs @@ -79,6 +79,8 @@ pub enum SubscribeError { #[derive(Debug, Error)] pub enum SnapshotError { + #[error("snapshot response headers timed out")] + HeadersTimeout, #[error("snapshot request failed: {0}")] Request(#[from] reqwest::Error), #[error("invalid snapshot metadata: {0}")] diff --git a/sdk/rust-client/src/lib.rs b/sdk/rust-client/src/lib.rs index 79302ecd..fefee4e7 100644 --- a/sdk/rust-client/src/lib.rs +++ b/sdk/rust-client/src/lib.rs @@ -143,16 +143,18 @@ impl SequencerClient { serde_json::from_str::(&body).map_err(|e| GetFeeError::Decode(e.to_string())) } - /// Streams without the short transaction deadline; callers own download cancellation. + /// Bounds response headers by the request timeout; callers own body cancellation. pub async fn latest_snapshot(&self) -> Result { - let response = self + let request = self .http_client .get(format!( "{}/latest_snapshot", self.endpoint.trim_end_matches('/') )) - .send() - .await? + .send(); + let response = tokio::time::timeout(self.request_timeout, request) + .await + .map_err(|_| SnapshotError::HeadersTimeout)?? .error_for_status()?; let header = |name| { response @@ -302,6 +304,28 @@ mod tests { )); } + #[tokio::test] + async fn snapshot_headers_keep_the_configured_request_timeout() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let client = SequencerClient::new(format!("http://{address}")) + .unwrap() + .with_request_timeout(Duration::from_millis(100)); + let stalled_server = async { + let (_stream, _) = listener.accept().await.unwrap(); + std::future::pending::<()>().await; + }; + let result = tokio::time::timeout(Duration::from_secs(2), async { + tokio::select! { + result = client.latest_snapshot() => result, + () = stalled_server => unreachable!(), + } + }) + .await + .expect("an accepted connection with no headers must reach the configured deadline"); + assert!(matches!(result, Err(SnapshotError::HeadersTimeout))); + } + #[tokio::test] async fn snapshot_body_outlives_the_transaction_request_timeout() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -317,12 +341,12 @@ mod tests { received += count; } stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nX-History-Era: 00112233-4455-4677-8899-aabbccddeeff\r\nX-Recovery-Generation: 0\r\nX-Executed-Input-Count: 7\r\n\r\n").await.unwrap(); - tokio::time::sleep(Duration::from_millis(100)).await; + tokio::time::sleep(Duration::from_millis(300)).await; stream.write_all(b"dump").await.unwrap(); }); let client = SequencerClient::new_with_timeout( format!("http://{address}"), - Duration::from_millis(30), + Duration::from_millis(100), ) .unwrap(); let snapshot = client.latest_snapshot().await.unwrap(); @@ -364,4 +388,39 @@ mod tests { SubscribeError::History(actual) if actual == policy) ); } + + #[tokio::test] + async fn subscription_below_nonzero_baseline_decodes_history_unavailable() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + let mut byte = [0]; + assert_eq!(stream.read(&mut byte).await.unwrap(), 1); + request.push(byte[0]); + assert!(request.len() <= 8192); + } + let request = String::from_utf8(request).unwrap(); + assert!(request.starts_with("GET /ws/subscribe?")); + assert!(request.contains("next_input=40 ")); + stream.write_all(b"HTTP/1.1 409 Conflict\r\nX-History-Error: {\"code\":\"HISTORY_UNAVAILABLE\",\"available_from\":41}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").await.unwrap(); + }); + let client = SequencerClient::new(format!("http://{address}")).unwrap(); + let result = client + .subscribe(HistoryClaim { + version: HistoryVersion { + era_id: "00112233-4455-4677-8899-aabbccddeeff".parse().unwrap(), + recovery_generation: RecoveryGeneration::new(0), + }, + next_input: ExecutedInputCount::new(40), + }) + .await; + assert!(matches!(result, + Err(SubscribeError::History(HistoryPolicyError::HistoryUnavailable { available_from })) + if available_from == ExecutedInputCount::new(41))); + server.await.unwrap(); + } } diff --git a/sequencer-core/src/history.rs b/sequencer-core/src/history.rs index dbd1b7fa..abdf938e 100644 --- a/sequencer-core/src/history.rs +++ b/sequencer-core/src/history.rs @@ -258,6 +258,48 @@ mod tests { assert!("00112233445546778899aabbccddeeff".parse::().is_err()); } + #[test] + fn history_policy_errors_preserve_literal_wire_codes_and_fields() { + let current = HistoryVersion { + era_id: CANONICAL.parse().unwrap(), + recovery_generation: RecoveryGeneration::new(7), + }; + for (error, json) in [ + ( + HistoryPolicyError::EraChanged { current }, + serde_json::json!({ + "code": "ERA_CHANGED", + "current": { "era_id": CANONICAL, "recovery_generation": 7 } + }), + ), + ( + HistoryPolicyError::StaleGeneration { current }, + serde_json::json!({ + "code": "STALE_GENERATION", + "current": { "era_id": CANONICAL, "recovery_generation": 7 } + }), + ), + ( + HistoryPolicyError::HistoryUnavailable { + available_from: ExecutedInputCount::new(41), + }, + serde_json::json!({ "code": "HISTORY_UNAVAILABLE", "available_from": 41 }), + ), + ( + HistoryPolicyError::AheadOfHead { + head: ExecutedInputCount::new(50), + }, + serde_json::json!({ "code": "AHEAD_OF_HEAD", "head": 50 }), + ), + ] { + assert_eq!(serde_json::to_value(error).unwrap(), json); + assert_eq!( + serde_json::from_value::(json).unwrap(), + error + ); + } + } + #[test] fn era_id_displays_canonical_lowercase_hyphenated_form() { let era = EraId::from_bytes(CANONICAL_BYTES).expect("canonical UUIDv4"); diff --git a/sequencer-core/src/scheduler/fold.rs b/sequencer-core/src/scheduler/fold.rs index de340efc..0448a502 100644 --- a/sequencer-core/src/scheduler/fold.rs +++ b/sequencer-core/src/scheduler/fold.rs @@ -44,8 +44,10 @@ pub struct FoldInput { /// `B` and its scheduler nonce `N` (metadata — the bare-metal app cannot /// recompute it, so the engine is *told* it via `resume_at`). /// - `seeds`: directs reconstructed from `(A, B]`, with sequencer-sourced -/// batches already dropped by the caller (their content is already in `S`, -/// their frames' safe blocks `≤ A`). Must arrive in ascending L1 order. +/// batches already dropped by the caller. The checkpoint must account for +/// every direct through its application clock `A`: the exporter verifies no +/// pending canonical direct has inclusion block `<= A`. `A < B` alone cannot +/// establish this after faulty sequencing. Seeds arrive in ascending L1 order. /// - `replay`: the full `(B, C]` stream (batches + directs) in L1 order. The /// scheduler classifies each input (batch iff `sender == sequencer_address`), /// force-executes overdue directs on arrival, applies accepted batches, @@ -517,6 +519,81 @@ mod tests { ); } + #[test] + fn earlier_checkpoint_recovers_a_direct_hidden_below_a_later_checkpoint_clock() { + let feed = |scheduler: &mut Scheduler, inputs: Vec| { + for input in inputs { + scheduler + .process_input(SchedulerInput { + sender: input.sender, + inclusion_block: input.inclusion_block, + domain: domain(), + payload: input.payload, + }) + .expect("canonical execution"); + } + }; + let mut canonical = Scheduler::new(FoldApp::default(), config()); + feed( + &mut canonical, + vec![ + direct(DIRECT_SENDER, 5, 1), + cover_batch(9, 0, 5), + direct(DIRECT_SENDER, 9, 2), + ], + ); + + let checkpoint = canonical.app.clone(); + let checkpoint_nonce = canonical.next_expected_batch_nonce(); + let a = checkpoint.last_executed_safe_block(); + assert_eq!(a, 5); + assert!( + canonical + .direct_q + .iter() + .all(|input| input.inclusion_block > a), + "the earlier checkpoint is eligible, with a nonempty seed queue" + ); + assert_eq!(canonical.queued_direct_len(), 1); + + // Equality is canonically valid even though honest live sequencing + // cannot submit a batch into its own already-safe block. + let replay = vec![ + direct(DIRECT_SENDER, 10, 3), + cover_batch(10, 1, 10), + direct(DIRECT_SENDER, 10, 4), + empty_batch(11, 2), + ]; + feed(&mut canonical, replay.clone()); + let later_a = canonical.app.last_executed_safe_block(); + assert_eq!(later_a, 10); + assert!(later_a < 11, "the later checkpoint passes the scalar bound"); + assert_eq!(canonical.next_expected_batch_nonce(), 3); + assert_eq!(canonical.app.executed_directs, vec![1, 2, 3]); + assert_eq!(canonical.queued_direct_len(), 1); + assert_eq!(canonical.direct_q[0].payload, vec![4]); + assert_eq!(canonical.direct_q[0].inclusion_block, later_a); + // Its pending direct is excluded by (A,B], so the exporter must refuse + // this later checkpoint. Recovery can use the eligible earlier one. + + canonical.drain_covered_at(11).expect("terminal drain"); + let (expected, expected_nonce) = canonical.finish(); + let (recovered, recovered_nonce) = fold_replay( + checkpoint, + checkpoint_nonce, + config(), + domain(), + vec![direct(DIRECT_SENDER, 9, 2)], + replay, + 11, + ) + .expect("recovery from eligible checkpoint"); + assert_eq!(recovered.executed_directs, vec![1, 2, 3, 4]); + assert_eq!(recovered.executed_directs, expected.executed_directs); + assert_eq!(recovered.progress(), expected.progress()); + assert_eq!(recovered_nonce, expected_nonce); + } + #[test] fn fold_replay_reconstructs_identical_state_to_a_live_run() { // Differential: splitting the (genesis, C] stream at an intermediate B diff --git a/sequencer/src/commands/error.rs b/sequencer/src/commands/error.rs index 37aaf820..28204902 100644 --- a/sequencer/src/commands/error.rs +++ b/sequencer/src/commands/error.rs @@ -386,8 +386,9 @@ pub enum BootstrapError { } /// Terminal failures of the `setup --recovery` procedure — the ones -/// an operator must resolve (the flush and the post-flush re-sync reuse the -/// transient [`RecoveryError`] paths instead). All map to [`EXIT_TERMINAL`]: +/// an operator must resolve (flush, post-flush re-sync, and a stopping block +/// behind the checkpoint use the transient [`RecoveryError`] paths instead). +/// All map to [`EXIT_TERMINAL`]: /// a plain restart re-runs the same bad inputs and re-fails identically. #[derive(Debug, Error)] pub enum SetupRecoveryError { @@ -410,8 +411,10 @@ pub enum SetupRecoveryError { /// recovery export (`info.toml` + `checkpoint.toml` + `state/`), not a watchdog CM checkpoint. #[error("failed to load checkpoint dump at {path}: {message}")] CheckpointLoad { path: String, message: String }, - /// Outside the known empty genesis checkpoint, A must precede B so the - /// recovery seed includes all potentially pending directs in block B. + /// Outside empty genesis, A must precede B: equality can exclude pending + /// directs in B from the seed range. A < B alone does not certify queue + /// eligibility; the exporter must also verify no pending direct is at or + /// below A (docs/recovery/cockroach.md). #[error( "checkpoint last-executed safe block {executed_safe_block} (A) must precede \ checkpoint block {checkpoint_block} (B), except for empty genesis; \ @@ -432,11 +435,10 @@ pub enum SetupRecoveryError { /// `setup`'s read-only detection gate: the reasons a /// fresh `setup` refuses because a *previous* instance left work past the -/// checkpoint. Because plain setup has already initialized a genesis baseline, -/// the remedy is to wipe that uncompleted data dir and run `setup --recovery` -/// which flushes/folds the outstanding batches; a plain `setup` restart -/// re-detects and re-refuses (hence [`EXIT_SETUP_NEEDS_RECOVERY`], not the -/// auto-recovery class 10). +/// checkpoint. The gate runs before baseline publication and leaves setup +/// incomplete. Rebuild in a fresh data directory with `setup --recovery` to +/// flush/fold the outstanding batches; a plain `setup` restart re-detects and +/// re-refuses (hence [`EXIT_SETUP_NEEDS_RECOVERY`], not auto-recovery class 10). /// /// Both variants carry diagnostic fields for the refusal log line. #[derive(Debug, Error)] @@ -839,6 +841,13 @@ mod tests { }), "a resync behind the flush's observed view", ), + ( + recovery_retry(RecoveryRetryReason::CheckpointAheadOfStop { + checkpoint_block: 901, + stop_block: 899, + }), + "a valid manual recovery checkpoint ahead of the RPC stopping block", + ), ( CommandError::Bootstrap(BootstrapError::Identity( IdentityError::FirstBootRequiresL1, diff --git a/sequencer/src/commands/run/startup_hygiene.rs b/sequencer/src/commands/run/startup_hygiene.rs index e6005f4b..bfbf7366 100644 --- a/sequencer/src/commands/run/startup_hygiene.rs +++ b/sequencer/src/commands/run/startup_hygiene.rs @@ -4,6 +4,8 @@ //! Startup clears stale leases, checks rollback checkpoint metadata, then collects //! obsolete snapshots and orphan directories before workers are admitted. +use std::os::unix::fs::MetadataExt; + use crate::commands::error::CommandError; use crate::ingress::inclusion_lane::dump_info::{self, delete_dump_dir}; @@ -61,24 +63,36 @@ fn snapshot_gc_at_startup(storage: &mut crate::storage::Storage) -> Result Result { - let known: std::collections::HashSet = storage + let known = storage .list_dump_rows()? .into_iter() - .map(|row| row.prefix) - .collect(); + .map(|row| { + dump_identity(&row.prefix).map_err(|source| CommandError::ReferencedSnapshotArtifact { + path: row.prefix, + source, + }) + }) + .collect::, _>>()?; let mut removed = 0; for entry in std::fs::read_dir(dumps_dir)? { let entry = entry?; let path = entry.path(); - if known.contains(&path) { + let retained = match dump_identity(&path) { + Ok(identity) => known.contains(&identity), + // GC or an earlier orphan deletion can leave an unregistered + // dangling symlink. Retained references already resolved above. + Err(err) if err.kind() == std::io::ErrorKind::NotFound => false, + Err(err) => return Err(err.into()), + }; + if retained { continue; } match delete_dump_dir(&path) { @@ -95,6 +109,12 @@ fn sweep_orphan_dumps( Ok(removed) } +fn dump_identity(path: &std::path::Path) -> std::io::Result<(u64, u64)> { + // Canonical paths can still differ across mount aliases and macOS firmlinks. + let metadata = std::fs::metadata(path)?; + Ok((metadata.dev(), metadata.ino())) +} + #[cfg(test)] mod tests { use super::*; @@ -183,6 +203,281 @@ mod tests { assert_eq!(removed, 0); } + #[test] + fn sweep_preserves_mixed_references_across_path_spellings() { + for spelling in [ + "relative-to-absolute", + "absolute-to-relative", + "stored-dot", + "sweep-dot", + ] { + let db = temp_db(spelling); + let mut storage = Storage::open(&db.path).unwrap(); + // Avoid changing the process-wide cwd while other tests are running. + let root = tempfile::tempdir_in(".").unwrap(); + let relative = std::path::PathBuf::from(root.path().file_name().unwrap()).join("dumps"); + std::fs::create_dir(&relative).unwrap(); + let absolute = relative.canonicalize().unwrap(); + let (stored_dir, sweep_dir) = match spelling { + "relative-to-absolute" => (relative.clone(), absolute.clone()), + "absolute-to-relative" => (absolute.clone(), relative), + "stored-dot" => (std::path::Path::new(".").join(&relative), relative), + "sweep-dot" => (relative.clone(), std::path::Path::new(".").join(relative)), + _ => unreachable!(), + }; + let tracked = absolute.join("tracked"); + create_structured_dump(&tracked); + storage + .insert_baseline_snapshot( + &stored_dir.join("tracked"), + crate::storage::ExecutedInputCount::ZERO, + ) + .unwrap(); + // A literal match must not hide the other row's aliased spelling. + let literal_match = sweep_dir.join("literal-match"); + create_structured_dump(&literal_match); + storage + .write(|tx| { + tx.execute( + "INSERT INTO dumps(prefix) VALUES (?1)", + [literal_match.to_str().unwrap()], + ) + }) + .unwrap(); + let orphan = absolute.join("orphan"); + create_structured_dump(&orphan); + + assert_eq!( + sweep_orphan_dumps(&mut storage, &sweep_dir).unwrap(), + 1, + "{spelling}" + ); + + assert!(tracked.join("info.toml").is_file(), "{spelling}"); + assert!(literal_match.join("info.toml").is_file(), "{spelling}"); + assert!(!orphan.exists(), "{spelling}"); + assert_eq!(storage.list_dump_rows().unwrap().len(), 2); + } + } + + #[cfg(unix)] + #[test] + fn sweep_preserves_symlinked_parent_and_dump_aliases() { + for stored_via_alias in [false, true] { + let db = temp_db("sweep-symlinks"); + let mut storage = Storage::open(&db.path).unwrap(); + let root = tempfile::tempdir().unwrap(); + let dumps = root.path().join("dumps"); + std::fs::create_dir(&dumps).unwrap(); + let parent_alias = root.path().join("parent-alias"); + std::os::unix::fs::symlink(&dumps, &parent_alias).unwrap(); + let tracked = dumps.join("tracked"); + create_structured_dump(&tracked); + let dump_alias = dumps.join("dump-alias"); + std::os::unix::fs::symlink(&tracked, &dump_alias).unwrap(); + let stored = if stored_via_alias { + parent_alias.join("dump-alias") + } else { + tracked.clone() + }; + storage + .insert_baseline_snapshot(&stored, crate::storage::ExecutedInputCount::ZERO) + .unwrap(); + let orphan = dumps.join("orphan"); + create_structured_dump(&orphan); + let sweep_dir = if stored_via_alias { + &dumps + } else { + &parent_alias + }; + + assert_eq!(sweep_orphan_dumps(&mut storage, sweep_dir).unwrap(), 1); + + assert!(tracked.join("info.toml").is_file()); + assert!(dump_alias.join("info.toml").is_file()); + assert!(stored.join("info.toml").is_file()); + assert!(!orphan.exists()); + } + } + + #[cfg(target_os = "macos")] + #[test] + fn sweep_preserves_mixed_firmlink_and_literal_references() { + let db = temp_db("sweep-firmlink-alias"); + let mut storage = Storage::open(&db.path).unwrap(); + let root = tempfile::tempdir_in("/private/tmp").unwrap(); + let dumps = root.path().join("dumps"); + std::fs::create_dir(&dumps).unwrap(); + let alias = + std::path::Path::new("/System/Volumes/Data").join(dumps.strip_prefix("/").unwrap()); + let tracked = dumps.join("tracked"); + create_structured_dump(&tracked); + storage + .insert_baseline_snapshot(&tracked, crate::storage::ExecutedInputCount::ZERO) + .unwrap(); + + // This alias survives realpath resolution, unlike an ordinary symlink. + let listed = alias.join("tracked"); + assert_ne!( + tracked.canonicalize().unwrap(), + listed.canonicalize().unwrap() + ); + let stored_metadata = std::fs::metadata(&tracked).unwrap(); + let listed_metadata = std::fs::metadata(&listed).unwrap(); + assert_eq!( + (stored_metadata.dev(), stored_metadata.ino()), + (listed_metadata.dev(), listed_metadata.ino()) + ); + + // The sets overlap literally, so a global disjoint-set guard is insufficient. + let literal_match = alias.join("literal-match"); + create_structured_dump(&literal_match); + storage + .write(|tx| { + tx.execute( + "INSERT INTO dumps(prefix) VALUES (?1)", + [literal_match.to_str().unwrap()], + ) + }) + .unwrap(); + let orphan = dumps.join("orphan"); + create_structured_dump(&orphan); + + let removed = sweep_orphan_dumps(&mut storage, &alias).unwrap(); + + assert!( + tracked.join("info.toml").is_file(), + "the referenced artifact must survive its mount alias" + ); + assert!(literal_match.join("info.toml").is_file()); + assert!(!orphan.exists()); + assert_eq!(removed, 1); + assert_eq!(storage.list_dump_rows().unwrap().len(), 2); + } + + #[test] + fn sweep_resolves_every_reference_before_deleting_any_artifact() { + let db = temp_db("sweep-unresolved-reference"); + let mut storage = Storage::open(&db.path).unwrap(); + let dumps = tempfile::tempdir().unwrap(); + let tracked = dumps.path().join("tracked"); + create_structured_dump(&tracked); + storage + .insert_baseline_snapshot( + &dumps.path().join(".").join("tracked"), + crate::storage::ExecutedInputCount::ZERO, + ) + .unwrap(); + let missing = dumps.path().join("missing"); + storage + .write(|tx| { + tx.execute( + "INSERT INTO dumps(prefix) VALUES (?1)", + [missing.to_str().unwrap()], + ) + }) + .unwrap(); + let orphan = dumps.path().join("orphan"); + create_structured_dump(&orphan); + + let error = sweep_orphan_dumps(&mut storage, dumps.path()).unwrap_err(); + + assert!(matches!( + &error, + CommandError::ReferencedSnapshotArtifact { path, source } + if path == &missing && source.kind() == std::io::ErrorKind::NotFound + )); + assert_eq!(error.exit_code(), crate::commands::error::EXIT_TERMINAL); + assert!(tracked.join("info.toml").is_file()); + assert!( + orphan.join("info.toml").is_file(), + "resolution precedes deletion" + ); + assert_eq!(storage.list_dump_rows().unwrap().len(), 2); + } + + #[cfg(unix)] + #[test] + fn sweep_removes_orphan_symlinks_without_following_them() { + let db = temp_db("sweep-orphan-symlinks"); + let mut storage = Storage::open(&db.path).unwrap(); + let root = tempfile::tempdir().unwrap(); + let dumps = root.path().join("dumps"); + std::fs::create_dir(&dumps).unwrap(); + let tracked = dumps.join("tracked"); + create_structured_dump(&tracked); + storage + .insert_baseline_snapshot(&tracked, crate::storage::ExecutedInputCount::ZERO) + .unwrap(); + let outside = root.path().join("outside"); + create_structured_dump(&outside); + let orphan = dumps.join("orphan"); + create_structured_dump(&orphan); + for (name, target) in [ + ("orphan-alias", orphan), + ("dangling", root.path().join("missing")), + ("outside-alias", outside.clone()), + ] { + std::os::unix::fs::symlink(target, dumps.join(name)).unwrap(); + } + + assert_eq!(sweep_orphan_dumps(&mut storage, &dumps).unwrap(), 4); + assert!(tracked.join("info.toml").is_file()); + assert!(outside.join("info.toml").is_file()); + let remaining: Vec<_> = std::fs::read_dir(&dumps) + .unwrap() + .map(|entry| entry.unwrap().file_name()) + .collect(); + assert_eq!(remaining, vec![std::ffi::OsString::from("tracked")]); + assert_eq!(storage.list_dump_rows().unwrap().len(), 1); + } + + #[test] + fn startup_resets_persisted_crash_leases_before_collecting_artifacts() { + let db = temp_db("startup-crash-leases"); + let mut storage = Storage::open(&db.path).unwrap(); + let dumps = tempfile::tempdir().unwrap(); + let baseline = dumps.path().join("baseline"); + create_structured_dump(&baseline); + let baseline_id = storage + .insert_baseline_snapshot(&baseline, crate::storage::ExecutedInputCount::ZERO) + .unwrap(); + let obsolete = dumps.path().join("obsolete"); + create_structured_dump(&obsolete); + let obsolete_id = storage + .write(|tx| { + tx.execute( + "UPDATE dumps SET lease_count = 1 WHERE id = ?1", + [baseline_id], + )?; + tx.execute( + "INSERT INTO dumps(prefix, lease_count) VALUES (?1, 1)", + [obsolete.to_str().unwrap()], + )?; + Ok(tx.last_insert_rowid()) + }) + .unwrap(); + drop(storage); + + let mut storage = Storage::open(&db.path).unwrap(); + assert_eq!(storage.dump_lease_count(obsolete_id).unwrap(), Some(1)); + assert!(storage.gc_unreferenced_dumps().unwrap().is_empty()); + assert!(obsolete.exists(), "the persisted lease blocks ordinary GC"); + + run_snapshot_hygiene(&mut storage, dumps.path()).unwrap(); + + assert_eq!(storage.dump_lease_count(baseline_id).unwrap(), Some(0)); + assert_eq!(storage.dump_lease_count(obsolete_id).unwrap(), None); + assert!( + baseline.exists(), + "the rollback baseline survives startup GC" + ); + assert!( + !obsolete.exists(), + "startup collects the abandoned leased artifact" + ); + } + #[test] fn snapshot_gc_at_startup_removes_unreferenced_rows() { let db = temp_db("gc-startup"); diff --git a/sequencer/src/commands/setup/checkpoint_tests.rs b/sequencer/src/commands/setup/checkpoint_tests.rs new file mode 100644 index 00000000..21f603e4 --- /dev/null +++ b/sequencer/src/commands/setup/checkpoint_tests.rs @@ -0,0 +1,226 @@ +// (c) Cartesi and individual authors (see AUTHORS) +// SPDX-License-Identifier: Apache-2.0 (see LICENSE) + +use std::path::{Path, PathBuf}; + +use alloy_primitives::{Address, U256}; +use app_core::application::{WalletApp, WalletConfig}; +use sequencer_core::application::{ + AppError, AppOutputs, Application, ApplicationProgress, ValidationOutcome, execute_direct_input, +}; +use sequencer_core::history::ExecutedInputCount; +use sequencer_core::l2_tx::{DirectInput, ValidUserOp}; +use sequencer_core::user_op::UserOp; + +use super::{Checkpoint, rebuild_from_checkpoint}; +use crate::commands::error::{BootstrapError, CommandError, EXIT_RESTART_TRANSIENT}; +use crate::ingress::inclusion_lane::dump_info; +use crate::recovery::{RecoveryError, RecoveryFailure, RecoveryRetryReason}; +use crate::storage::test_helpers::{ + SENDER_A, default_protocol_timing, pin_test_deployment_identity, temp_db, +}; +use crate::storage::{FrontierMode, LifecycleCommand, Storage, StoredSafeInput}; + +struct ReplayForbiddenApp(ApplicationProgress); + +impl Application for ReplayForbiddenApp { + fn max_method_payload_bytes() -> usize { + 0 + } + + fn validate_user_op( + &self, + _: Address, + _: &UserOp, + _: u16, + ) -> Result { + panic!("an uncovered checkpoint must refuse before application execution") + } + + fn apply_valid_user_op(&mut self, _: &ValidUserOp, _: u64) -> Result { + panic!("an uncovered checkpoint must refuse before application execution") + } + + fn apply_direct_input(&mut self, _: &DirectInput) -> Result { + panic!("an uncovered checkpoint must refuse before application execution") + } + + fn progress(&self) -> ApplicationProgress { + self.0 + } + + fn from_dump(_: &Path) -> Result { + unreachable!("the checkpoint is supplied by the fixture") + } + + fn create_dump(&mut self, _: &Path) -> Result<(), AppError> { + panic!("an uncovered checkpoint must refuse before artifact creation") + } + + fn state_file_in_dump(_: &Path) -> PathBuf { + unreachable!("the refused checkpoint has no new artifact") + } +} + +#[test] +fn recovery_refuses_stop_before_checkpoint_without_replay_or_publication() { + // The last case has H1 >= B: resync reaching the checkpoint cannot replace + // the fixed fold boundary C. The earlier cases model an honestly lagging node. + for (application_clock, checkpoint_block, stop_block, resynced_head) in [ + (900, 901, 899, 899), + (900, 964, 930, 930), + (900, 964, 930, 964), + ] { + let db = temp_db("recovery-uncovered-checkpoint"); + let mut storage = + Storage::initialize_for_command(&db.path, LifecycleCommand::Rebuild).unwrap(); + pin_test_deployment_identity(&mut storage, SENDER_A); + let identity = storage.deployment_identity().unwrap().unwrap(); + let inputs = if resynced_head > application_clock { + vec![StoredSafeInput { + sender: Address::repeat_byte(0x22), + payload: vec![1], + block_number: application_clock + 1, + }] + } else { + vec![] + }; + storage + .append_safe_inputs_with_timestamp( + resynced_head, + resynced_head, + &inputs, + SENDER_A, + &default_protocol_timing(), + FrontierMode::DeferUntilAnchorSet, + ) + .unwrap(); + let checkpoint = Checkpoint { + app: ReplayForbiddenApp( + ApplicationProgress::try_new(ExecutedInputCount::new(1), application_clock) + .unwrap(), + ), + executed_safe_block: application_clock, + checkpoint_nonce: 1, + checkpoint_block, + }; + let dumps = tempfile::tempdir().unwrap(); + + let error = rebuild_from_checkpoint( + checkpoint, + &identity, + stop_block, + &mut storage, + dumps.path(), + ) + .expect_err("the RPC stopping block must cover the trusted checkpoint"); + + assert_eq!(error.exit_code(), EXIT_RESTART_TRANSIENT); + assert!(matches!( + error, + CommandError::Bootstrap(BootstrapError::Recovery(RecoveryError::Retry(ref failure))) + if matches!(failure.as_ref(), RecoveryFailure::PolicyRetry( + RecoveryRetryReason::CheckpointAheadOfStop { + checkpoint_block: found_checkpoint, + stop_block: found_stop, + }) if *found_checkpoint == checkpoint_block && *found_stop == stop_block) + )); + assert!(!storage.is_setup_complete().unwrap()); + assert!(matches!( + storage.history_state(), + Err(rusqlite::Error::QueryReturnedNoRows) + )); + assert!(storage.open_state().unwrap().is_none()); + assert!(storage.latest_snapshot().unwrap().is_none()); + assert_eq!(std::fs::read_dir(dumps.path()).unwrap().count(), 0); + assert_eq!(storage.current_safe_block().unwrap(), Some(resynced_head)); + assert_eq!( + storage.safe_input_end_exclusive().unwrap(), + inputs.len() as u64 + ); + } +} + +#[test] +fn recovery_publishes_at_checkpoint_or_later_including_genesis() { + let owner = Address::repeat_byte(0x77); + for (checkpoint_block, stop_block, expected_count, expected_balance) in + [(10, 10, 2, 120_u64), (10, 15, 3, 150), (0, 0, 0, 0)] + { + let db = temp_db("recovery-covered-checkpoint"); + let mut storage = + Storage::initialize_for_command(&db.path, LifecycleCommand::Rebuild).unwrap(); + pin_test_deployment_identity(&mut storage, SENDER_A); + let identity = storage.deployment_identity().unwrap().unwrap(); + let config = WalletConfig::devnet(); + let deposit = |block, amount: u64| DirectInput { + sender: config.erc20_portal_address, + block_number: block, + payload: [ + config.supported_erc20_token.as_slice(), + owner.as_slice(), + U256::from(amount).to_be_bytes::<32>().as_slice(), + ] + .concat(), + }; + let mut app = WalletApp::new(config); + if checkpoint_block != 0 { + execute_direct_input(&mut app, &deposit(5, 100)).unwrap(); + } + let checkpoint = Checkpoint { + executed_safe_block: app.last_executed_safe_block(), + app, + checkpoint_nonce: u64::from(checkpoint_block != 0), + checkpoint_block, + }; + let resynced_head = stop_block + 5; + let inputs = [(5, 100), (7, 20), (12, 30), (16, 40)] + .into_iter() + .filter(|(block, _)| *block <= resynced_head) + .map(|(block, amount)| { + let input = deposit(block, amount); + StoredSafeInput { + sender: input.sender, + payload: input.payload, + block_number: block, + } + }) + .collect::>(); + storage + .append_safe_inputs_with_timestamp( + resynced_head, + resynced_head, + &inputs, + SENDER_A, + &default_protocol_timing(), + FrontierMode::DeferUntilAnchorSet, + ) + .unwrap(); + let dumps = tempfile::tempdir().unwrap(); + + rebuild_from_checkpoint( + checkpoint, + &identity, + stop_block, + &mut storage, + dumps.path(), + ) + .unwrap(); + + assert!(storage.is_setup_complete().unwrap()); + let history = storage.history_state().unwrap(); + assert_eq!(history.base_safe_block, stop_block); + assert_eq!(history.base_executed_input_count, expected_count); + assert_eq!( + storage.open_state().unwrap().unwrap().safe_block, + stop_block + ); + let snapshot = storage.latest_snapshot().unwrap().unwrap(); + let app = WalletApp::from_dump(&dump_info::app_prefix(&snapshot.dump.prefix)).unwrap(); + assert_eq!(app.executed_input_count().get(), expected_count); + assert_eq!( + app.current_user_balance(owner), + U256::from(expected_balance) + ); + } +} diff --git a/sequencer/src/commands/setup/mod.rs b/sequencer/src/commands/setup/mod.rs index a6725e70..69ecb0a0 100644 --- a/sequencer/src/commands/setup/mod.rs +++ b/sequencer/src/commands/setup/mod.rs @@ -28,6 +28,8 @@ use alloy_primitives::Address; use sequencer_core::application::{AppError, Application}; use sequencer_core::scheduler::{FoldInput, SchedulerConfig, fold_replay}; +#[cfg(test)] +mod checkpoint_tests; pub(crate) mod fill; use super::{ensure_deployment_identity, validate_rpc_chain_id}; @@ -37,7 +39,9 @@ use crate::commands::error::{ }; use crate::ingress::inclusion_lane::dump_info; use crate::l1::reader::{InputReader, InputReaderConfig, InputReaderError}; -use crate::recovery::{MempoolFlusher, assert_resync_caught_up}; +use crate::recovery::{ + MempoolFlusher, RecoveryError, RecoveryRetryReason, assert_resync_caught_up, +}; use crate::storage::{self, DeploymentIdentity, FeeOracleIdentity}; pub async fn setup(config: SetupConfig, genesis_app: F) -> Result<(), CommandError> @@ -419,6 +423,8 @@ impl Checkpoint { /// Load `S` from the dump dir, derive `A` and `N`, and enforce the load-time /// precondition `A < B`, except for the known empty genesis checkpoint. /// Equality otherwise hides pending directs later in the same block. + /// The exporter must separately verify no canonical direct at or below A + /// remains queued; this bundle carries no queue evidence to check here. fn load(dir: &std::path::Path, checkpoint_block: u64) -> Result { let load_err = |message: String| SetupRecoveryError::CheckpointLoad { path: dir.display().to_string(), @@ -495,11 +501,11 @@ fn source_fold_inputs( /// The `setup --recovery` procedure: rebuild a freshly-wiped DB from /// a trusted checkpoint instead of refusing. Runs after the shared prefix /// (identity pinned, initial sync done); replaces the detection gate + genesis -/// snapshot. Distinct, terminal error type ([`SetupRecoveryError`]) from -/// `run`'s recovery — operator-driven, one-shot. +/// snapshot. Invalid checkpoint/configuration failures are terminal; transient +/// L1 failures leave setup incomplete for a fresh attempt. /// /// The `flush → fold → fill` steps are enumerated authoritatively in -/// **[`docs/recovery/cockroach.md`](../../../docs/recovery/cockroach.md)** (spec, +/// **[`docs/recovery/cockroach.md`](../../../../docs/recovery/cockroach.md)** (spec, /// data dictionary `A`/`B`/`C`/`N`/`N'`, and code map) and anchored inline below /// (`// 1.`…`// 6.`). Read the doc before editing this function. /// @@ -548,6 +554,28 @@ where let resynced_safe_block = require_resynced_safe_block(storage.current_safe_block()?)?; assert_resync_caught_up(resynced_safe_block, stop_block)?; + rebuild_from_checkpoint(checkpoint, identity, stop_block, storage, dumps_dir) +} + +fn rebuild_from_checkpoint( + checkpoint: Checkpoint, + identity: &DeploymentIdentity, + stop_block: u64, + storage: &mut storage::Storage, + dumps_dir: &std::path::Path, +) -> Result<(), CommandError> { + // A valid checkpoint can outpace an honestly lagging RPC node. Labelling + // its state with an earlier C would let run execute its directs again. + if stop_block < checkpoint.checkpoint_block { + return Err( + RecoveryError::retry(RecoveryRetryReason::CheckpointAheadOfStop { + checkpoint_block: checkpoint.checkpoint_block, + stop_block, + }) + .into(), + ); + } + // 4. Source the (A, B] direct seeds + the (B, C] replay stream. let submitter = identity.batch_submitter_address; let (seeds, replay) = source_fold_inputs(storage, &checkpoint, stop_block, submitter)?; diff --git a/sequencer/src/egress/api/snapshot.rs b/sequencer/src/egress/api/snapshot.rs index 1c83041a..0d802b12 100644 --- a/sequencer/src/egress/api/snapshot.rs +++ b/sequencer/src/egress/api/snapshot.rs @@ -23,9 +23,11 @@ use tokio::fs::File; use tokio::io::{AsyncRead, ReadBuf}; use tokio_util::io::{ReaderStream, SyncIoBridge}; -use crate::http::{StorageTaskError, storage_task}; +use crate::http::{ApiError, StorageTaskError, storage_task}; use crate::runtime::shutdown::{RuntimeScope, abort_terminal}; -use crate::storage::{FinalizedLease, LeaseGuard, LeasedDump, ReleaseScheduler, Storage}; +use crate::storage::{ + FinalizedLease, FinalizedSelectionError, LeaseGuard, LeasedDump, ReleaseScheduler, Storage, +}; type BoxError = StorageTaskError; @@ -89,7 +91,7 @@ async fn finalized_inclusion_block(State(state): State>) - }) .into_response(), Ok(None) => StatusCode::NOT_FOUND.into_response(), - Err(err) => internal_error("read finalized inclusion block", err), + Err(err) => finalized_error("read finalized inclusion block", err), } } @@ -105,7 +107,7 @@ async fn finalized_state( } = match acquire_finalized(&state).await { Ok(Some(leased)) => leased, Ok(None) => return StatusCode::NOT_FOUND.into_response(), - Err(err) => return internal_error("acquire finalized lease", err), + Err(err) => return finalized_error("acquire finalized lease", err), }; let etag = format!("\"block-{inclusion_block}\""); @@ -119,7 +121,7 @@ async fn finalized_state( let history = leased.history_version; let LeasedDump { guard, .. } = leased; - match File::open(&path).await { + match comparison_io(File::open(&path).await, &path) { Ok(file) => Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/octet-stream") @@ -131,17 +133,10 @@ async fn finalized_state( "X-Recovery-Generation", history.recovery_generation.get().to_string(), ) - .body(stream_body(file, guard)) + .body(stream_body(file, guard, path)) .expect("snapshot response headers are well-formed"), // `guard` is a local here; on this error path it drops → lease released. - Err(err) => { - if err.kind() == std::io::ErrorKind::NotFound { - abort_terminal(format!( - "durable finalized snapshot artifact missing: {path:?}" - )); - } - internal_error("open finalized state file", err) - } + Err(err) => internal_error("open finalized state file", err), } } @@ -170,7 +165,7 @@ async fn finalized_snapshot(State(state): State>) -> Respo archive_response(&state, dump, Some(checkpoint)) } Ok(None) => StatusCode::NOT_FOUND.into_response(), - Err(err) => internal_error("acquire accepted snapshot lease", err), + Err(err) => finalized_error("acquire accepted snapshot lease", err), } } @@ -247,13 +242,43 @@ fn state_file_path(state: &SnapshotState, prefix: &Path) -> PathBuf { .unwrap_or_else(|_| abort_terminal("application snapshot path callback panicked")) } -fn stream_body(file: File, guard: LeaseGuard) -> Body { +fn stream_body(file: File, guard: LeaseGuard, path: PathBuf) -> Body { Body::from_stream(ReaderStream::new(GuardedReader { - file, + file: ComparisonReader { reader: file, path }, _guard: Arc::new(guard), })) } +fn comparison_io(result: std::io::Result, path: &Path) -> std::io::Result { + if let Err(error) = &result + && dump_info::referenced_artifact_io_is_terminal(error) + { + abort_terminal(format!( + "durable comparison artifact is unusable: {}: {error}", + path.display() + )); + } + result +} + +struct ComparisonReader { + reader: R, + path: PathBuf, +} + +impl AsyncRead for ComparisonReader { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.get_mut(); + Pin::new(&mut this.reader) + .poll_read(cx, buf) + .map(|result| comparison_io(result, &this.path)) + } +} + /// Do not turn a producer failure into a successful truncated archive response. struct ArchiveReader { reader: tokio::io::DuplexStream, @@ -368,11 +393,95 @@ fn internal_error(context: &str, err: impl std::fmt::Display) -> Response { StatusCode::INTERNAL_SERVER_ERROR.into_response() } +fn finalized_error(context: &str, err: StorageTaskError) -> Response { + if matches!( + err.downcast_ref::(), + Some(FinalizedSelectionError::CanonicalDivergence) + ) { + return ApiError::unavailable( + "canonical divergence prevents accepted checkpoint selection", + ) + .into_response(); + } + internal_error(context, err) +} + #[cfg(test)] mod tests { use super::*; use crate::storage::test_helpers::temp_db; + async fn read_corrupt_comparison(path: &Path) { + let db = temp_db("corrupt-comparison-path"); + let mut storage = Storage::open(&db.path).unwrap(); + storage + .insert_baseline_snapshot(path, crate::storage::ExecutedInputCount::ZERO) + .unwrap(); + let state = Arc::new(SnapshotApiState { + snapshot: SnapshotState { + db_path: db.path, + state_file_in_dump: Path::to_path_buf, + }, + shutdown: RuntimeScope::default(), + release_scheduler: Arc::new(|release| release()), + }); + let response = finalized_state(State(state), HeaderMap::new()).await; + // Unix can open a directory successfully: the structural error first + // appears when the response body reads it. + let _ = axum::body::to_bytes(response.into_body(), 1024).await; + panic!("corrupt comparison artifact did not abort"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[cfg(unix)] + async fn comparison_directory_aborts_when_streamed() { + if !crate::runtime::shutdown::abort_test_child( + "egress::api::snapshot::tests::comparison_directory_aborts_when_streamed", + ) { + return; + } + let root = tempfile::tempdir().unwrap(); + read_corrupt_comparison(root.path()).await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[cfg(unix)] + async fn comparison_file_parent_aborts_on_open() { + if !crate::runtime::shutdown::abort_test_child( + "egress::api::snapshot::tests::comparison_file_parent_aborts_on_open", + ) { + return; + } + let root = tempfile::tempdir().unwrap(); + let parent = root.path().join("file"); + std::fs::write(&parent, b"not a directory").unwrap(); + read_corrupt_comparison(&parent.join("comparison")).await; + } + + #[tokio::test] + async fn comparison_operational_read_error_remains_nonterminal() { + struct Unavailable; + impl AsyncRead for Unavailable { + fn poll_read( + self: Pin<&mut Self>, + _: &mut Context<'_>, + _: &mut ReadBuf<'_>, + ) -> Poll> { + Poll::Ready(Err(std::io::Error::from( + std::io::ErrorKind::PermissionDenied, + ))) + } + } + let mut reader = ComparisonReader { + reader: Unavailable, + path: PathBuf::from("temporarily-unavailable"), + }; + let error = tokio::io::AsyncReadExt::read(&mut reader, &mut [0; 1]) + .await + .unwrap_err(); + assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[cfg(unix)] async fn finalized_state_path_panic_aborts_process() { diff --git a/sequencer/src/ingress/inclusion_lane/dump_info.rs b/sequencer/src/ingress/inclusion_lane/dump_info.rs index 8bcc7c6c..98cf12dd 100644 --- a/sequencer/src/ingress/inclusion_lane/dump_info.rs +++ b/sequencer/src/ingress/inclusion_lane/dump_info.rs @@ -100,7 +100,7 @@ pub(crate) fn write_archive( let mut archive = tar::Builder::new(writer); archive.append_path_with_name(dump_dir.join(INFO_FILE), INFO_FILE)?; let state = app_prefix(dump_dir); - if state.is_dir() { + if std::fs::metadata(&state)?.is_dir() { archive.append_dir_all(APP_STATE_SUBDIR, state)?; } else { archive.append_path_with_name(state, APP_STATE_SUBDIR)?; @@ -441,6 +441,14 @@ mod tests { check::(); } + #[test] + fn archive_propagates_missing_state_metadata() { + let root = tempfile::tempdir().unwrap(); + write_info(root.path(), &DumpInfo::at_baseline(0)).unwrap(); + let error = write_archive(Vec::new(), root.path(), 0, None).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::NotFound); + } + fn sample() -> DumpInfo { DumpInfo { format_version: FORMAT_VERSION, diff --git a/sequencer/src/ingress/inclusion_lane/tests.rs b/sequencer/src/ingress/inclusion_lane/tests.rs index b62ef352..9f58bfa0 100644 --- a/sequencer/src/ingress/inclusion_lane/tests.rs +++ b/sequencer/src/ingress/inclusion_lane/tests.rs @@ -56,6 +56,7 @@ fn decode_progress(bytes: &[u8], app_name: &str) -> Result, progress: ApplicationProgress, + fail_dump: bool, /// Test-only scheduling seam used to keep a rejected queue saturated long /// enough to distinguish one bounded turn from an unbounded drain. reject_user_ops_after: Option, @@ -113,6 +114,9 @@ impl Application for TestApp { } fn create_dump(&mut self, prefix: &Path) -> Result<(), AppError> { + if self.fail_dump { + return Err(std::io::Error::other("injected application dump failure").into()); + } std::fs::create_dir(prefix)?; std::fs::write(Self::state_file_in_dump(prefix), b"")?; Ok(()) @@ -1955,6 +1959,44 @@ async fn restart_resumes_from_pending_checkpoint_without_skipping_txs() { ); } +#[test] +fn application_dump_failure_leaves_tip_and_latest_snapshot_unchanged() { + let db = temp_db("application-dump-failure"); + let mut storage = Storage::open(&db.path).unwrap(); + let mut head = storage + .initialize_open_state(0, SafeInputRange::empty_at(0)) + .unwrap(); + let dumps = tempfile::tempdir().unwrap(); + let mut app = TestApp::default(); + register_genesis_snapshot(&mut app, &mut storage, dumps.path()); + super::snapshot::close_batch_with_snapshot(&mut app, &mut storage, &mut head, 0, dumps.path()) + .unwrap(); + let snapshot = storage.latest_snapshot().unwrap().unwrap(); + let tip = head.batch_index; + let dump_count = storage.list_dump_rows().unwrap().len(); + + app.fail_dump = true; + let error = super::snapshot::close_batch_with_snapshot( + &mut app, + &mut storage, + &mut head, + 0, + dumps.path(), + ) + .expect_err("an application dump failure must precede the batch seal"); + assert!(matches!( + error, + super::snapshot::TakeDumpError::CreateDump(super::dump_info::CreateDumpDirError::App( + AppError::Io(ref source) + )) if source.to_string() == "injected application dump failure" + )); + assert_eq!(head.batch_index, tip); + assert_eq!(storage.open_state().unwrap().unwrap().batch_index, tip); + assert_eq!(storage.latest_batch_index().unwrap(), Some(tip)); + assert_eq!(storage.latest_snapshot().unwrap().unwrap(), snapshot); + assert_eq!(storage.list_dump_rows().unwrap().len(), dump_count); +} + #[test] fn empty_batch_snapshot_preserves_application_count() { let db = temp_db("empty-snapshot-count"); diff --git a/sequencer/src/integration_tests/historical_bootstrap.rs b/sequencer/src/integration_tests/historical_bootstrap.rs index bfffbd44..4cb1bbc1 100644 --- a/sequencer/src/integration_tests/historical_bootstrap.rs +++ b/sequencer/src/integration_tests/historical_bootstrap.rs @@ -294,6 +294,14 @@ async fn historical_bootstrap_restores_transfer_history_and_hands_off_at_baselin let era = metadata.history.version.era_id; assert_eq!(metadata.history.available_from.get(), 7); assert_eq!(metadata.history.head.get(), 7); + assert!(matches!( + client.subscribe(HistoryClaim { + version: metadata.history.version, + next_input: ExecutedInputCount::new(6), + }).await, + Err(sequencer_rust_client::SubscribeError::History(HistoryPolicyError::HistoryUnavailable { available_from })) + if available_from == metadata.history.available_from + )); assert_eq!(metadata.baseline.l1_stop_block, STOP); assert_eq!(metadata.baseline.l1_end_input_index, 8); assert_eq!(metadata.baseline.next_batch_nonce, reference_nonce); diff --git a/sequencer/src/integration_tests/snapshot_endpoints.rs b/sequencer/src/integration_tests/snapshot_endpoints.rs index 11b0eb78..05d3c659 100644 --- a/sequencer/src/integration_tests/snapshot_endpoints.rs +++ b/sequencer/src/integration_tests/snapshot_endpoints.rs @@ -180,6 +180,103 @@ fn archive_state(bytes: &[u8]) -> Vec { .unwrap() } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn same_block_divergence_refuses_every_finalized_endpoint() { + use crate::storage::test_helpers::{ + default_protocol_timing, local_batch_payload, pin_test_deployment_identity, + }; + use crate::storage::{ExecutedInputCount, SafeInputRange, StoredSafeInput}; + use sequencer_core::batch::{Batch, Frame}; + use sequencer_core::scheduler::{Scheduler, SchedulerConfig, SchedulerInput}; + use ssz::Encode; + + let db = temp_db("same-block-divergence-http"); + let root = tempfile::tempdir().unwrap(); + let submitter = alloy_primitives::Address::repeat_byte(1); + let mut storage = Storage::open(&db.path).unwrap(); + pin_test_deployment_identity(&mut storage, submitter); + let mut head = storage + .initialize_open_state(5, SafeInputRange::empty_at(0)) + .unwrap(); + let prefix = write_state(root.path(), "local", b"local checkpoint", 1); + storage + .close_frame_and_batch_with_snapshot(&mut head, 5, &prefix, 0, ExecutedInputCount::ZERO) + .unwrap(); + let snapshot = storage.latest_snapshot().unwrap().unwrap(); + let inputs = vec![ + StoredSafeInput { + sender: alloy_primitives::Address::repeat_byte(2), + payload: vec![42], + block_number: 6, + }, + StoredSafeInput { + sender: submitter, + payload: local_batch_payload(&mut storage, 0), + block_number: 10, + }, + StoredSafeInput { + sender: submitter, + payload: Batch { + nonce: 1, + frames: vec![Frame { + safe_block: 6, + fee_price: 0, + user_ops: vec![], + }], + } + .as_ssz_bytes(), + block_number: 10, + }, + ]; + storage + .append_safe_inputs(10, &inputs, submitter, &default_protocol_timing()) + .unwrap(); + assert!(storage.canonical_divergence().unwrap().is_some()); + + // The foreign batch consumes the next nonce and drains a direct that the + // matching local batch did not: its snapshot is not the end of block 10. + let mut canonical = Scheduler::new(WalletApp::default(), SchedulerConfig::new(submitter)); + for input in inputs { + canonical + .process_input(SchedulerInput { + sender: input.sender, + payload: input.payload, + inclusion_block: input.block_number, + domain: sequencer_core::build_input_domain(1, alloy_primitives::Address::ZERO), + }) + .unwrap(); + } + let (app, nonce) = canonical.finish(); + assert_eq!(nonce, 2); + assert_eq!(app.executed_input_count().get(), 1); + assert_eq!(snapshot.executed_input_count, ExecutedInputCount::ZERO); + + let server = start_server(&db.path).await.expect("test listener"); + let client = reqwest::Client::new(); + for (path, conditional) in [ + ("/finalized_state/inclusion_block", false), + ("/finalized_state", false), + ("/finalized_state", true), + ("/finalized_snapshot", false), + ] { + let mut request = client.get(server.url(path)); + if conditional { + request = request.header("If-None-Match", "\"block-10\""); + } + let response = request.send().await.unwrap(); + assert_eq!( + response.status(), + reqwest::StatusCode::SERVICE_UNAVAILABLE, + "{path}, conditional={conditional}" + ); + assert_eq!( + response.json::().await.unwrap()["code"], + "UNAVAILABLE" + ); + assert_eq!(storage.dump_lease_count(snapshot.dump.id).unwrap(), Some(0)); + } +} + /// Transient WAL lock contention vs. a real read failure. /// /// SQLite readers can see `SQLITE_BUSY` in WAL mode during last-connection diff --git a/sequencer/src/l1/submitter/poster.rs b/sequencer/src/l1/submitter/poster.rs index aa6dc904..6db3ef6f 100644 --- a/sequencer/src/l1/submitter/poster.rs +++ b/sequencer/src/l1/submitter/poster.rs @@ -532,6 +532,9 @@ impl BatchPoster for EthereumBatchPoster { if evm_advance.msgSender != self.config.batch_submitter_address { continue; } + // This dedicated key emits our own well-formed batches. A decode + // failure is evidence to investigate under self-trust, not an + // untrusted direct input to skip (docs/threat-model/README.md). let batch: Batch = ssz::Decode::from_ssz_bytes(evm_advance.payload.as_ref()) .map_err(|err| BatchPosterError::Provider(format!("{err:?}")))?; observed_nonces.push(batch.nonce); diff --git a/sequencer/src/recovery/mod.rs b/sequencer/src/recovery/mod.rs index f862ac79..3bf9b9f9 100644 --- a/sequencer/src/recovery/mod.rs +++ b/sequencer/src/recovery/mod.rs @@ -76,6 +76,14 @@ pub enum RecoveryRetryReason { resynced_safe_block: u64, flush_observed_safe_block: u64, }, + #[error( + "post-flush stopping block {stop_block} predates checkpoint block {checkpoint_block}; \ + synchronize the RPC node and retry recovery" + )] + CheckpointAheadOfStop { + checkpoint_block: u64, + stop_block: u64, + }, #[error("local recovery facts changed before phase execution: {status:?}")] StaleDecision { status: DangerStatus }, #[error("the Tip was already open when the EnsureOpenTip phase ran")] @@ -430,10 +438,10 @@ fn classify_signer_provider( /// build), refused because re-running the same boot re-fails /// identically. The discovery-time facts (wrong contract, pre-v3 /// InputBox) never reach this function: they arise in `InputReader::new`, -/// which only `setup` calls and projects as a worker exit (register -/// finding 32). In the live loop the same URL was already proven by this -/// boot's initial sync, so a live `Bootstrap` restarts unclassified -/// rather than poisoning the data directory. +/// which only `setup` calls and currently projects as a live-worker exit +/// rather than a deterministic configuration failure. In the live loop the +/// same URL was already proven by this boot's initial sync, so a live +/// `Bootstrap` restarts unclassified rather than poisoning the data directory. /// - `Join` (a non-panic loss of a storage task) is shutdown-path /// cancellation in the live loop. During startup the runtime that would /// cancel it is the one driving this boot, so an unexplained loss is diff --git a/sequencer/src/storage/egress/historical.rs b/sequencer/src/storage/egress/historical.rs index 730879a0..2b5ba5f6 100644 --- a/sequencer/src/storage/egress/historical.rs +++ b/sequencer/src/storage/egress/historical.rs @@ -21,8 +21,9 @@ use crate::storage::history::{ }; use crate::storage::l1_inputs::query_deployment_identity; use crate::storage::mutations::batch_tree_anchor_in; -use crate::storage::safe_accepted_batches::canonical_divergence_in; -use crate::storage::snapshot_dumps::{finalized_dump_in, has_rollback_safe_snapshot_in}; +use crate::storage::snapshot_dumps::{ + FinalizedSelectionError, finalized_dump_in, has_rollback_safe_snapshot_in, +}; #[derive(Debug, thiserror::Error)] pub(crate) enum HistoricalReadError { @@ -30,8 +31,8 @@ pub(crate) enum HistoricalReadError { Policy(#[from] HistoryPolicyError), #[error("{0}")] BadRequest(String), - #[error("canonical divergence prevents accepted checkpoint selection")] - CanonicalDivergence, + #[error(transparent)] + Checkpoint(#[from] FinalizedSelectionError), #[error("reading historical L1 inputs: {0}")] Storage(#[from] rusqlite::Error), } @@ -60,12 +61,12 @@ impl Storage { "from_generation exceeds the current recovery generation".to_owned(), ))); } - if canonical_divergence_in(tx)?.is_some() { - return Ok(Err(HistoricalReadError::CanonicalDivergence)); - } + let accepted = match finalized_dump_in(tx) { + Ok(accepted) => accepted, + Err(error) => return Ok(Err(error.into())), + }; let deployment = query_deployment_identity(tx)?.ok_or(rusqlite::Error::QueryReturnedNoRows)?; - let accepted = finalized_dump_in(tx)?; if accepted.is_none() { assert!( has_rollback_safe_snapshot_in(tx)?, diff --git a/sequencer/src/storage/egress/historical/tests.rs b/sequencer/src/storage/egress/historical/tests.rs index cf1587d7..af0b4506 100644 --- a/sequencer/src/storage/egress/historical/tests.rs +++ b/sequencer/src/storage/egress/historical/tests.rs @@ -366,8 +366,8 @@ fn accepted_checkpoint_uses_the_exact_latest_snapshot_and_preserves_baseline() { .unwrap(); assert!(matches!( storage.history_info(None, None), - Err(HistoricalReadError::Storage( - rusqlite::Error::QueryReturnedNoRows + Err(HistoricalReadError::Checkpoint( + FinalizedSelectionError::Storage(rusqlite::Error::QueryReturnedNoRows) )) )); } @@ -385,7 +385,9 @@ fn canonical_divergence_cannot_be_advertised_as_an_accepted_receipt() { .unwrap(); assert!(matches!( storage.history_info(None, None), - Err(HistoricalReadError::CanonicalDivergence) + Err(HistoricalReadError::Checkpoint( + FinalizedSelectionError::CanonicalDivergence + )) )); } diff --git a/sequencer/src/storage/history.rs b/sequencer/src/storage/history.rs index 0948d1db..975e70ba 100644 --- a/sequencer/src/storage/history.rs +++ b/sequencer/src/storage/history.rs @@ -3,8 +3,6 @@ //! Immutable era baseline and the preserved prefix at each recovery generation. -#[cfg(test)] -use rusqlite::OptionalExtension; use rusqlite::{Connection, Result, Transaction, params, types::Type}; use sequencer_core::history::{EraId, ExecutedInputCount, HistoryVersion, RecoveryGeneration}; @@ -65,21 +63,6 @@ pub(super) fn initialize_history_in( base: ExecutedInputCount, base_safe_block: u64, ) -> Result<()> { - #[cfg(test)] - if let Some(existing) = query_history_state(tx).optional()? { - // Test fixtures initialize genesis when opening their schema. Production - // creates this row only with the complete durable baseline. - assert_eq!( - existing.base_executed_input_count, - base.get(), - "history base differs" - ); - assert_eq!( - existing.base_safe_block, base_safe_block, - "L1 prefix differs" - ); - return Ok(()); - } let mut bytes: [u8; EraId::BYTE_LEN] = tx.query_row("SELECT randomblob(16)", [], |row| row.get(0))?; bytes[6] = (bytes[6] & 0x0f) | 0x40; @@ -182,13 +165,33 @@ mod tests { .write(|tx| initialize_history_in(tx, ExecutedInputCount::new(41), 70)) .unwrap(); let state = storage.history_state().unwrap(); - for sql in [ - "UPDATE history_state SET era_id = era_id", - "UPDATE history_state SET base_executed_input_count = 42", - "UPDATE history_state SET base_safe_block = 71", - "DELETE FROM history_state", + let duplicate = storage + .write(|tx| initialize_history_in(tx, ExecutedInputCount::new(41), 70)) + .expect_err("even identical baseline values cannot initialize another era"); + assert_eq!( + duplicate.to_string(), + "history state is inserted once per database" + ); + for (sql, expected) in [ + ( + "UPDATE history_state SET era_id = X'00000000000040008000000000000001'", + "history baseline is immutable", + ), + ( + "UPDATE history_state SET base_executed_input_count = 42", + "history baseline is immutable", + ), + ( + "UPDATE history_state SET base_safe_block = 71", + "history baseline is immutable", + ), + ( + "DELETE FROM history_state", + "history state is write-once per database", + ), ] { - assert!(storage.conn.execute(sql, []).is_err(), "{sql}"); + let error = storage.conn.execute(sql, []).expect_err(sql); + assert_eq!(error.to_string(), expected, "{sql}"); } drop(storage); let mut reopened = Storage::open(&db.path).unwrap(); diff --git a/sequencer/src/storage/mod.rs b/sequencer/src/storage/mod.rs index d7fa5332..97d6fac3 100644 --- a/sequencer/src/storage/mod.rs +++ b/sequencer/src/storage/mod.rs @@ -62,7 +62,7 @@ pub use recovery::DangerStatus; pub(crate) use recovery::{RecoveryInspection, RecoveryMutationError}; pub use sequencer_core::history::{EraId, ExecutedInputCount, HistoryVersion, RecoveryGeneration}; pub use snapshot_dumps::{ - DumpRow, FinalizedDump, FinalizedLease, LeaseGuard, LeasedDump, + DumpRow, FinalizedDump, FinalizedLease, FinalizedSelectionError, LeaseGuard, LeasedDump, PersistentReleaseFailureReporter, ReleaseScheduler, Snapshot, }; diff --git a/sequencer/src/storage/open.rs b/sequencer/src/storage/open.rs index 89c7d895..d9beba94 100644 --- a/sequencer/src/storage/open.rs +++ b/sequencer/src/storage/open.rs @@ -71,8 +71,8 @@ impl Storage { }) } - /// Create the schema and record its owning command in one migration - /// transaction. The complete history baseline is published later. On + /// Create the schema for setup or rebuild in one migration transaction. + /// The complete history baseline is published later. On /// an already-migrated database the hook does not run; callers must /// inspect the existing facts. pub(crate) fn initialize_for_command( diff --git a/sequencer/src/storage/safe_accepted_batches.rs b/sequencer/src/storage/safe_accepted_batches.rs index 7d37fffd..3294b13c 100644 --- a/sequencer/src/storage/safe_accepted_batches.rs +++ b/sequencer/src/storage/safe_accepted_batches.rs @@ -387,115 +387,124 @@ mod tests { use sequencer_core::{batch::Batch, history::ExecutedInputCount}; use ssz::Encode; - let db = temp_db("accepted-recovered-prefix"); - let mut storage = Storage::initialize_for_command(&db.path, LifecycleCommand::Rebuild) - .expect("initialize rebuild"); - let submitter = Address::repeat_byte(0x99); - let timing = default_protocol_timing(); - pin_test_deployment_identity(&mut storage, submitter); - let old_future = Batch { - nonce: 1, - frames: vec![], - } - .as_ssz_bytes(); - let old_accepted = Batch { - nonce: 0, - frames: vec![], - } - .as_ssz_bytes(); - assert!( - timing - .scheduler_accepts( - submitter, - SafeInputView { - safe_input_index: 0, - sender: submitter, - payload: &old_future, - inclusion_block: 20, - }, - 0 - ) - .is_none() - ); - assert!( - timing - .scheduler_accepts( + // At C itself, the future nonce must precede nonce 0 in L1 order: + // it was rejected there and must not become accepted after the rebuild. + for old_future_block in [20, 30] { + let db = temp_db("accepted-recovered-prefix"); + let mut storage = Storage::initialize_for_command(&db.path, LifecycleCommand::Rebuild) + .expect("initialize rebuild"); + let submitter = Address::repeat_byte(0x99); + let timing = default_protocol_timing(); + pin_test_deployment_identity(&mut storage, submitter); + let old_future = Batch { + nonce: 1, + frames: vec![], + } + .as_ssz_bytes(); + let old_accepted = Batch { + nonce: 0, + frames: vec![], + } + .as_ssz_bytes(); + assert!( + timing + .scheduler_accepts( + submitter, + SafeInputView { + safe_input_index: 0, + sender: submitter, + payload: &old_future, + inclusion_block: old_future_block, + }, + 0 + ) + .is_none() + ); + assert!( + timing + .scheduler_accepts( + submitter, + SafeInputView { + safe_input_index: 1, + sender: submitter, + payload: &old_accepted, + inclusion_block: 30, + }, + 0 + ) + .is_some() + ); + storage + .append_safe_inputs_with_timestamp( + 30, + 30, + &[ + StoredSafeInput { + sender: submitter, + payload: old_future, + block_number: old_future_block, + }, + StoredSafeInput { + sender: submitter, + payload: old_accepted, + block_number: 30, + }, + ], submitter, - SafeInputView { - safe_input_index: 1, - sender: submitter, - payload: &old_accepted, - inclusion_block: 30, - }, - 0 + &timing, + FrontierMode::DeferUntilAnchorSet, ) - .is_some() - ); - storage - .append_safe_inputs_with_timestamp( - 30, - 30, - &[ - StoredSafeInput { - sender: submitter, - payload: old_future, - block_number: 20, - }, - StoredSafeInput { - sender: submitter, - payload: old_accepted, - block_number: 30, - }, - ], - submitter, - &timing, - FrontierMode::DeferUntilAnchorSet, - ) - .expect("ingest opaque prefix"); - storage - .write(|tx| { - super::super::history::initialize_history_in(tx, ExecutedInputCount::new(41), 30)?; - super::super::mutations::set_batch_tree_anchor_in(tx, 1)?; - super::super::ingress::open_recovery_tip_in_tx(tx, 30) - }) - .expect("install recovered baseline"); - let mut head = storage.open_state().expect("read root").expect("root"); - storage - .close_frame_and_batch(&mut head, 30) - .expect("close resumed batch"); - let payload = local_batch_payload(&mut storage, 1); - storage - .append_safe_inputs( - 31, - &[StoredSafeInput { - sender: submitter, - payload, - block_number: 31, - }], - submitter, - &timing, - ) - .expect("accept post-baseline batch"); - assert!( + .expect("ingest opaque prefix"); storage - .canonical_divergence() - .expect("divergence") - .is_none() - ); - let accepted = query_latest_safe_accepted_batch(&storage.conn) - .expect("accepted frontier") - .expect("resumed acceptance"); - assert_eq!((accepted.safe_input_index, accepted.nonce), (2, 1)); - assert_eq!( + .write(|tx| { + super::super::history::initialize_history_in( + tx, + ExecutedInputCount::new(41), + 30, + )?; + super::super::mutations::set_batch_tree_anchor_in(tx, 1)?; + super::super::ingress::open_recovery_tip_in_tx(tx, 30) + }) + .expect("install recovered baseline"); + let mut head = storage.open_state().expect("read root").expect("root"); storage - .conn - .query_row("SELECT COUNT(*) FROM safe_accepted_batches", [], |row| row - .get::<_, i64>( - 0 - )) - .expect("accepted count"), - 1 - ); + .close_frame_and_batch(&mut head, 30) + .expect("close resumed batch"); + let payload = local_batch_payload(&mut storage, 1); + storage + .append_safe_inputs( + 31, + &[StoredSafeInput { + sender: submitter, + payload, + block_number: 31, + }], + submitter, + &timing, + ) + .expect("accept post-baseline batch"); + assert!( + storage + .canonical_divergence() + .expect("divergence") + .is_none(), + "reinterpreted future-nonce input at block {old_future_block}" + ); + let accepted = query_latest_safe_accepted_batch(&storage.conn) + .expect("accepted frontier") + .expect("resumed acceptance"); + assert_eq!((accepted.safe_input_index, accepted.nonce), (2, 1)); + assert_eq!( + storage + .conn + .query_row("SELECT COUNT(*) FROM safe_accepted_batches", [], |row| row + .get::<_, i64>( + 0 + )) + .expect("accepted count"), + 1 + ); + } } fn insert_safe_input_zero(storage: &Storage) { diff --git a/sequencer/src/storage/snapshot_dumps.rs b/sequencer/src/storage/snapshot_dumps.rs index 41a1b140..e2503dec 100644 --- a/sequencer/src/storage/snapshot_dumps.rs +++ b/sequencer/src/storage/snapshot_dumps.rs @@ -35,6 +35,14 @@ pub struct FinalizedDump { pub executed_input_count: ExecutedInputCount, } +#[derive(Debug, thiserror::Error)] +pub enum FinalizedSelectionError { + #[error("canonical divergence prevents accepted checkpoint selection")] + CanonicalDivergence, + #[error("selecting finalized checkpoint: {0}")] + Storage(#[from] rusqlite::Error), +} + pub type ReleaseScheduler = Arc) + Send + Sync + 'static>; pub type PersistentReleaseFailureReporter = Arc; @@ -153,8 +161,10 @@ impl Storage { /// The latest accepted batch must have its own snapshot. A missing artifact /// is corruption, never a request to use an older accepted snapshot. - pub fn finalized_dump(&mut self) -> Result> { - self.read(|tx| finalized_dump_in(tx)) + pub fn finalized_dump( + &mut self, + ) -> std::result::Result, FinalizedSelectionError> { + self.read(|tx| Ok(finalized_dump_in(tx)))? } pub fn latest_snapshot(&mut self) -> Result> { @@ -173,15 +183,17 @@ impl Storage { &mut self, schedule: ReleaseScheduler, report_persistent_failure: PersistentReleaseFailureReporter, - ) -> Result> { + ) -> std::result::Result, FinalizedSelectionError> { let acquired = self.write(|tx| { - let Some(snapshot) = finalized_dump_in(tx)? else { - return Ok(None); + let snapshot = match finalized_dump_in(tx) { + Ok(Some(snapshot)) => snapshot, + Ok(None) => return Ok(Ok(None)), + Err(error) => return Ok(Err(error)), }; let history_version = query_history_state(tx)?.version; acquire_dump_lease_in(tx, snapshot.dump.id)?; - Ok(Some((snapshot, history_version))) - })?; + Ok(Ok(Some((snapshot, history_version)))) + })??; Ok(acquired.map(|(snapshot, history_version)| FinalizedLease { inclusion_block: snapshot.inclusion_block, dump: LeasedDump { @@ -353,7 +365,14 @@ fn baseline_snapshot_in(conn: &Connection) -> Result> { .optional() } -pub(super) fn finalized_dump_in(conn: &Connection) -> Result> { +pub(super) fn finalized_dump_in( + conn: &Connection, +) -> std::result::Result, FinalizedSelectionError> { + // A matched batch can precede a divergent accepted batch in the same L1 + // block. Its snapshot cannot represent that complete block boundary. + if super::safe_accepted_batches::canonical_divergence_in(conn)?.is_some() { + return Err(FinalizedSelectionError::CanonicalDivergence); + } if let Some((batch_index, nonce, inclusion_block)) = latest_accepted_boundary_in(conn)? { let snapshot = snapshot_for_batch_in(conn, batch_index)?; return Ok(Some(FinalizedDump { @@ -548,7 +567,9 @@ mod tests { .unwrap(); assert!(matches!( storage.finalized_dump(), - Err(rusqlite::Error::QueryReturnedNoRows) + Err(FinalizedSelectionError::Storage( + rusqlite::Error::QueryReturnedNoRows + )) )); assert!(matches!( storage.latest_snapshot(), @@ -734,6 +755,21 @@ mod tests { assert_eq!(storage.dump_lease_count(id).unwrap(), Some(0)); } + #[test] + fn lease_release_cannot_underflow() { + let db = temp_db("lease-underflow"); + let mut storage = Storage::open(&db.path).unwrap(); + let id = storage + .insert_baseline_snapshot(&prefix(0), ExecutedInputCount::ZERO) + .unwrap(); + let error = storage.release_dump_lease(id).unwrap_err(); + assert_eq!( + error.sqlite_error_code(), + Some(rusqlite::ErrorCode::ConstraintViolation) + ); + assert_eq!(storage.dump_lease_count(id).unwrap(), Some(0)); + } + #[test] fn persistent_release_failure_reaches_reporter() { let db = temp_db("persistent-lease"); diff --git a/tests/e2e/src/cold_replica.rs b/tests/e2e/src/cold_replica.rs index 97d49591..6b820472 100644 --- a/tests/e2e/src/cold_replica.rs +++ b/tests/e2e/src/cold_replica.rs @@ -12,6 +12,7 @@ use rollups_harness::replay::apply_ws_message; use rollups_harness::{ManagedSequencer, ReplayWalletApp, TestSigner, WsClient}; use sequencer_core::api::WsTxMessage; use sequencer_core::application::Application; +use sequencer_core::fee::fee_to_linear; use sequencer_rust_client::{ HistoryClaim, HistoryPolicyError, SequencerClient, SnapshotResponse, SubscribeError, }; @@ -230,12 +231,30 @@ async fn run_scenario(runtime: &mut ManagedSequencer) -> Scenari assert!(recovered.executed_input_count().get() < replica.executed_input_count().get()); let mut alice_l2 = runtime.wallet_l2(TestSigner::from_default(1)?)?; - alice_l2.set_next_nonce(recovered_reference.current_user_nonce(alice_address)); - alice_l2.transfer(bob_address, U256::from(6_000)).await?; + let expected_nonce = recovered_reference.current_user_nonce(alice_address); + let balance_before = recovered_reference.current_user_balance(alice_address); + // The fixed oracle keeps this quote valid across frame rotations. Derive the + // expected debit before receiving the event so wrong feed fees cannot agree by replay. + let quote = client.get_fee().await?; + assert_eq!(quote.fee, quote.recommended_fee); + let amount = U256::from(6_000); + let expected_balance = balance_before - amount - fee_to_linear(quote.fee); + alice_l2.set_next_nonce(expected_nonce); + alice_l2.transfer(bob_address, amount).await?; let resumed = recovered_ws.expect_user_op_from(alice_address).await?; + assert!(matches!(resumed, WsTxMessage::UserOp { fee, nonce, .. } + if fee == quote.fee && nonce == expected_nonce)); apply_ws_message(&mut recovered, resumed.clone())?; recovered_reference.apply(resumed)?; assert_same_state(&mut recovered, &recovered_reference)?; + assert_eq!( + recovered_reference.current_user_nonce(alice_address), + expected_nonce + 1 + ); + assert_eq!( + recovered_reference.current_user_balance(alice_address), + expected_balance + ); assert_eq!( recovered_reference.current_user_balance(bob_address), U256::from(6_000)