diff --git a/AGENTS.md b/AGENTS.md index 42d6ae058e..e7d2de157c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,7 +16,7 @@ This document contains rules and conventions for AI agents working on this codeb 6. **Every FSM `Registry.X.Get(...)` must have a matching preload, declared by the component that proposes the command** — The FSM apply path reads from the in-memory cache; a cache miss turns the read into a silent no-op. Each component that emits a proposal (the metadata converter, the index builder, the cluster-config reconciler, the idempotency-eviction scheduler, the mirror worker, admission) is responsible for declaring its own `preload.Needs` covering every key its apply path will read. There is NO central proposal→needs registry — coupling the preload package to every proposal type creates a single point that easily falls behind. The component knows what it reads; the component declares it. The shared `proposeTechnical` helper takes a `*preload.Needs` parameter the caller fills in (pass nil or an empty `Needs` when the apply path has no cache-keyed reads, e.g. cluster config / idempotency eviction). The preload populates the cache via `MirrorPreload` with a fresh value read at propose time (Pebble fallback on cache miss), and `PredictedIndex` catches mutations between propose and apply. 7. **Never silently skip a "should not happen" branch** — A branch that is reachable only if an invariant is violated (nil where the contract says non-nil, a state we believe unreachable, a cache miss after a guaranteed preload, etc.) MUST surface a loud signal: `return fmt.Errorf("invariant: ...")` so it bubbles up, or `assert.Unreachable(...)` for SUT-level invariants exercised under antithesis. A silent `return nil` / `continue` on these branches hides real bugs — particularly catastrophic in the FSM apply path, where a no-op desyncs nodes from each other. Branches that represent genuine runtime conditions (cache miss as an expected outcome, stale proposal, deleted entity) keep their soft `return nil`. The distinction is whether the case is *expected* (soft skip OK) or *impossible by design* (must fail loudly). The comment must say *why* the case is impossible so a reader can decide whether to add a hard fail or relax the rule. -8. **The audit log is the only source of truth — every other persisted dataset is a projection and must be verified by the checker** — Only `AuditEntry` (zone `Cold`, sub `Audit`) is cryptographically bound, via the hash chain that `state.BuildHashedHeaderPayload` + `processing.HashGenerator` produce and `checker.verifyAuditHashChain` verifies on every Check() run. Everything else stored in Pebble — `Log`, `AuditItem`, `AppliedProposal`, `LedgerLog.PurgedVolumes`, attribute caches (`Volume`, `Metadata`, `Transaction`, `Reference`, `Boundary`, etc.), reversion bitsets, idempotency keys, mirror cursors, chapters, bloom filters, signing keys, the read-side index — is a *projection* of orders that already live in the audit chain. Projections are rebuildable from the audit on demand, so we deliberately do NOT extend the hash chain to cover them (refactor over hash binding — see `feedback_audit_is_source_of_truth`). In exchange, **`internal/application/check/checker.go` MUST verify every projection it persists**: re-derive the value the projection should hold by replaying the audit (`ReplayLedgerLog`, `SimulateEphemeralPurge`, `partitionVolumes`, etc.) and compare to what is stored, emitting the matching `CHECK_STORE_ERROR_TYPE_*` event on divergence. A projection that the checker does not verify is a tampering vector — adding a new persisted projection without a matching compare* / collect* pass in the checker is the violation. Two scope refinements: (a) this applies to the **primary FSM store** — the single store `Check()` opens and walks; **peer secondary stores** (the `readstore` inverted index today; the `usagestore` counters forthcoming with EN-1334) are out of *main-store checker* scope *by construction* since `Check()` never opens them. That scope carve-out is **not** a claim they are integrity-safe: the readstore serves READY indexes directly to business-visible queries with no scan fallback, and its automated detect/drop/rebuild is **not yet wired** (tracked under `EN-1323`), so a corrupted or tampered index is a **current open integrity gap on the peer-store side** — a per-replica rebuild-health concern of the index builder, not an invariant-#8 main-store concern. Out of the main-store checker's mandate, not out of every integrity concern. (b) A primary-store projection may be exempted from a dedicated pass only when it is either (i) deterministically rebuildable from still-retained verified state through a real, *wired* rebuild path — not rebuildable merely in principle — or (ii) purely informational and intentionally carried across restore (cf. `BuildStatus`). These are distinct bases: of the *Known projection gaps* below, `DefaultEnforcementMode` and `SubAttrLedgerMetadata` qualify under (i) via `RebuildDelta`, while `EphemeralEvictedCount` / `TransientUsedCount` qualify under (ii) — they are NOT rebuilt by `RebuildDelta`, so the wired-rebuild basis must not be claimed for them. The current passes are `compareVolumes`, `compareMetadata`, `compareTransactions`, `compareExclusionProjections` (AppliedProposal.TransientVolumes + LedgerLog.PurgedVolumes), `checkReversionInvariants`, `verifySealingHash`, `compareIdempotencyOutcomes` (frozen idempotency outcomes in SubIdempKeys vs the hash-chained AuditFailure/AuditSuccess that wrote them — the failure kind is re-derived from the chain-bound reason via `domain.KindForReason`, never stored), and `compareIndexes` (SubAttrIndex registry vs CreateIndex/DropIndex/RemovedMetadataFieldType/DeleteLedger logs — covers presence + identity; BuildStatus is intentionally excluded because it is purely informational on the cluster-wide registry entry — queries gate on the per-replica `IndexVersionState.CurrentVersion`, not on BuildStatus), and `compareMirrorV2LogID` (stored `LedgerBoundaries.last_mirror_v2_log_id` **==** `max(audited MirrorIngest.v2_log_id)` per ledger — a full equality check, `CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH` on ANY divergence; the FSM enforces a contiguous applied prefix so at rest the two must be exactly equal, and the baseline floor is seeded from archived `Boundary` rows now included in the compact baseline snapshot; pre-field clusters are unsupported, no backfill leniency), and `compareSchema` (per-ledger `LedgerInfo.MetadataSchema` vs the replayed CreateLedger.initial_schema + SetMetadataFieldType/RemovedMetadataFieldType logs; under archived chapters the boundary schema is seeded from the baseline checkpoint, which carries LedgerInfo for this purpose), and `compareAccountTypes` (per-ledger `LedgerInfo.AccountTypes` vs the replayed AddAccountType/RemoveAccountType logs, baseline-seeded under archiving like the schema), and `compareLedgerPresence` (the live ledger set in the store must match the audit-derived set both ways: every audit-live ledger — CreateLedger with no later DeleteLedger, or a non-deleted baseline ledger under archiving — must have a *live* stored `LedgerInfo` else `MISSING_LEDGER` (covers an entry deleted outright OR tampered to a soft-deleted tombstone, which the compareSchema/compareAccountTypes loops skip), and every live stored `LedgerInfo` must be audit-backed else `UNAUDITED_LEDGER` (an injected empty ledger row carries no schema/types for the projection passes to flag). The audit-derived set is NEVER seeded from the live store — under archiving it comes from the baseline checkpoint, keeping the check independent of the data it verifies), and `compareReferences` (the SubAttrReference reference→txID uniqueness index vs the references replayed from CreatedTransaction/RevertedTransaction logs, baseline-seeded under archiving; verified both ways — missing, unaudited, and retargeted rows are all flagged; rows for deleted or cleanup-pending ledgers are skipped since they legitimately linger until a covering purge runs deleteLedgerData), and `compareBoundaries` (per-ledger `LedgerBoundaries` vs the checker’s re-derivation: NextTransactionId/NextLogId/PostingCount/RevertCount from the replayed logs plus the chain-bound AuditItem order effects — mirror fill-gap advances and NumscriptExecutionCount live on the orders, not the ledger-log stream — baseline-seeded under archiving; VolumeCount/MetadataCount/ReferenceCount are compared against a recount of the stored attribute rows they summarize, which their own passes verify entry-by-entry, so the chain is rows↔audit, counter↔rows; EphemeralEvictedCount/TransientUsedCount are intentionally excluded — informational, carried across restore, cf. BuildStatus); extend the list as new persisted projections land. **Known projection gaps**: `LedgerBoundaries.EphemeralEvictedCount` / `TransientUsedCount` are excluded from `compareBoundaries` and carried across restore rather than re-derived — informational counters (cf. BuildStatus); `LedgerInfo.DefaultEnforcementMode` and the ledger-metadata attribute (`SubAttrLedgerMetadata`) have no compare pass yet (both are rebuilt on restore by RebuildDelta). +8. **The audit log is the only source of truth — every other persisted dataset is a projection and must be verified by the checker** — Only `AuditEntry` (zone `Cold`, sub `Audit`) is cryptographically bound, via the hash chain that `state.BuildHashedHeaderPayload` + `processing.HashGenerator` produce and `checker.verifyAuditHashChain` verifies on every Check() run. Everything else stored in Pebble — `Log`, `AuditItem`, `AppliedProposal`, `LedgerLog.PurgedVolumes`, attribute caches (`Volume`, `Metadata`, `Transaction`, `Reference`, `Boundary`, etc.), reversion bitsets, idempotency keys, mirror cursors, chapters, bloom filters, signing keys, the read-side index — is a *projection* of orders that already live in the audit chain. Projections are rebuildable from the audit on demand, so we deliberately do NOT extend the hash chain to cover them (refactor over hash binding — see `feedback_audit_is_source_of_truth`). In exchange, **`internal/application/check/checker.go` MUST verify every projection it persists**: re-derive the value the projection should hold by replaying the audit (`ReplayLedgerLog`, `SimulateEphemeralPurge`, `partitionVolumes`, etc.) and compare to what is stored, emitting the matching `CHECK_STORE_ERROR_TYPE_*` event on divergence. A projection that the checker does not verify is a tampering vector — adding a new persisted projection without a matching compare* / collect* pass in the checker is the violation. Two scope refinements: (a) this applies to the **primary FSM store** — the single store `Check()` opens and walks; **peer secondary stores** (the `readstore` inverted index today; the `usagestore` counters forthcoming with EN-1334) are out of *main-store checker* scope *by construction* since `Check()` never opens them. That scope carve-out is **not** a claim they are integrity-safe: the readstore serves READY indexes directly to business-visible queries with no scan fallback, and its automated detect/drop/rebuild is **not yet wired** (tracked under `EN-1323`), so a corrupted or tampered index is a **current open integrity gap on the peer-store side** — a per-replica rebuild-health concern of the index builder, not an invariant-#8 main-store concern. Out of the main-store checker's mandate, not out of every integrity concern. (b) A primary-store projection may be exempted from a dedicated pass only when it is either (i) deterministically rebuildable from still-retained verified state through a real, *wired* rebuild path — not rebuildable merely in principle — or (ii) purely informational and intentionally carried across restore (cf. `BuildStatus`). These are distinct bases: of the *Known projection gaps* below, `DefaultEnforcementMode` and `SubAttrLedgerMetadata` qualify under (i) via `RebuildDelta`, while `EphemeralEvictedCount` / `TransientUsedCount` qualify under (ii) — they are NOT rebuilt by `RebuildDelta`, so the wired-rebuild basis must not be claimed for them. The current passes are `compareVolumes`, `compareMetadata`, `compareTransactions`, `compareExclusionProjections` (AppliedProposal.TransientVolumes + LedgerLog.PurgedVolumes), `checkReversionInvariants`, `verifySealingHash`, `compareIdempotencyOutcomes` (frozen idempotency outcomes in SubIdempKeys vs the hash-chained AuditFailure/AuditSuccess that wrote them — the failure kind is re-derived from the chain-bound reason via `domain.KindForReason`, never stored), and `compareIndexes` (SubAttrIndex registry vs CreateIndex/DropIndex/RemovedMetadataFieldType/DeleteLedger logs — covers presence + identity; BuildStatus is intentionally excluded because it is purely informational on the cluster-wide registry entry — queries gate on the per-replica `IndexVersionState.CurrentVersion`, not on BuildStatus), and `compareMirrorV2LogID` (stored `LedgerBoundaries.last_mirror_v2_log_id` **==** `max(audited MirrorIngest.v2_log_id)` per ledger — a full equality check, `CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH` on ANY divergence; the FSM enforces a contiguous applied prefix so at rest the two must be exactly equal, and the baseline floor is seeded from archived `Boundary` rows now included in the compact baseline snapshot; pre-field clusters are unsupported, no backfill leniency), and `compareSchema` (per-ledger `LedgerInfo.MetadataSchema` vs the replayed CreateLedger.initial_schema + SetMetadataFieldType/RemovedMetadataFieldType logs; under archived chapters the boundary schema is seeded from the baseline checkpoint, which carries LedgerInfo for this purpose), and `compareAccountTypes` (per-ledger `LedgerInfo.AccountTypes` vs the replayed AddAccountType/RemoveAccountType logs, baseline-seeded under archiving like the schema), and `compareLedgerPresence` (the live ledger set in the store must match the audit-derived set both ways: every audit-live ledger — CreateLedger with no later DeleteLedger, or a non-deleted baseline ledger under archiving — must have a *live* stored `LedgerInfo` else `MISSING_LEDGER` (covers an entry deleted outright OR tampered to a soft-deleted tombstone, which the compareSchema/compareAccountTypes loops skip), and every live stored `LedgerInfo` must be audit-backed else `UNAUDITED_LEDGER` (an injected empty ledger row carries no schema/types for the projection passes to flag). The audit-derived set is NEVER seeded from the live store — under archiving it comes from the baseline checkpoint, keeping the check independent of the data it verifies), and `compareReferences` (the SubAttrReference reference→txID uniqueness index vs the references replayed from CreatedTransaction/RevertedTransaction logs, baseline-seeded under archiving; verified both ways — missing, unaudited, and retargeted rows are all flagged; rows for deleted or cleanup-pending ledgers are skipped since they legitimately linger until a covering purge runs deleteLedgerData), and `compareBoundaries` (per-ledger `LedgerBoundaries` vs the checker’s re-derivation: NextTransactionId/NextLogId/PostingCount/RevertCount from the replayed logs plus the chain-bound AuditItem order effects — mirror fill-gap advances and NumscriptExecutionCount live on the orders, not the ledger-log stream — baseline-seeded under archiving; VolumeCount/MetadataCount/ReferenceCount are compared against a recount of the stored attribute rows they summarize, which their own passes verify entry-by-entry, so the chain is rows↔audit, counter↔rows; EphemeralEvictedCount/TransientUsedCount are intentionally excluded — informational, carried across restore, cf. BuildStatus), and `compareReversions` (the per-ledger reversion bitsets — `ZonePerLedger`/`SubPLReversions`, the projection the FSM's already-reverted gate reads — vs the reverted set derived from baseline tx-row markers plus the replayed RevertedTransaction logs; exact equality both ways — a lost bit re-admits a double revert, an unaudited bit blocks a legitimate one; driven purely by audit-derived state — no persisted marker (pending-cleanup included) can exempt a live ledger, stored rows for non-live ledgers are flagged since DeleteLedger deletes them at apply on both the live path and the replay, and undecodable rows are reported); extend the list as new persisted projections land. **Known projection gaps**: `LedgerBoundaries.EphemeralEvictedCount` / `TransientUsedCount` are excluded from `compareBoundaries` and carried across restore rather than re-derived — informational counters (cf. BuildStatus); `LedgerInfo.DefaultEnforcementMode` and the ledger-metadata attribute (`SubAttrLedgerMetadata`) have no compare pass yet (both are rebuilt on restore by RebuildDelta). 9. **Never bypass the FSM coverage gate** — Every cache-attribute read on the FSM hot path MUST go through `Scope.GetX(...)` so the per-order `coverage_bits` admit it. Reading the underlying `Registry.X.KeyStore().M` (or any other parent-cache iterator) directly skips the gate and produces non-deterministic FSM behavior: the gate is what binds the order to the admission-declared preload set, and a direct read silently sees keys the proposer never declared. There is NO documented exception — paths that need to iterate (e.g. cascade-on-delete) MUST either declare the relevant `preload.Needs` upfront, defer the work to a lifecycle path (`batch.deleteLedgerData` + `MarkLedgerForCleanup`), or be rejected at design review. New helpers that scan the parent KeyStore from inside an order/TU handler are the violation, even when wrapped in a method on `WriteSet`. The coverage gate exists precisely so admission's declared key set is the FSM's only legitimate read horizon — under no circumstances should the apply path widen it on the fly. diff --git a/docs/ops/backup-restore.md b/docs/ops/backup-restore.md index dbd8528780..7d6e77e91f 100644 --- a/docs/ops/backup-restore.md +++ b/docs/ops/backup-restore.md @@ -137,9 +137,9 @@ The backup is a complete Pebble database that contains: | Zone | Prefix | Contents | |------|--------|----------| -| Attributes | `0x01` | Volumes, account metadata, ledger metadata, reversions, references, ledger info, boundaries | +| Attributes | `0x01` | Volumes, account metadata, ledger metadata, references, ledger info, boundaries | | Cache | `0x02` | Derived/cached state | -| Per-Ledger | `0x03` | Per-ledger data | +| Per-Ledger | `0x03` | Per-ledger data (reversion bitset words, pending cleanups, mirror state) | | Cold | `0x04` | Transaction logs (`{0x04, 0x01}`), audit entries (`{0x04, 0x02}`) | | Idempotency | `0x05` | Idempotency keys | | Global | `0x06` | Last applied index (reset to 0), last applied timestamp, signing keys, signing config, chapters, sink configs, sink cursors, sink statuses | @@ -320,7 +320,7 @@ The server-side job: 3. Reads the manifest from S3 and downloads the checkpoint files in parallel through an `errgroup` worker pool. The pool size is set by the server flag `--restore-download-parallelism` (default 16, clamped to `[1, 64]`). -4. Applies any incremental export segments on top of the checkpoint and rebuilds derived state (volumes, metadata, transactions) from the exported logs, starting at the checkpoint's last log sequence. This is the same `ApplyExports` + `RebuildDelta` path used by the offline `ledgerctl store bootstrap` command, so a manifest with incremental backups restores all data written after the last full checkpoint. +4. Applies any incremental export segments on top of the checkpoint and rebuilds derived state (volumes, metadata, transactions, reversion bitsets) from the exported logs, starting at the checkpoint's last log sequence. This is the same `ApplyExports` + `RebuildDelta` path used by the offline `ledgerctl store bootstrap` command, so a manifest with incremental backups restores all data written after the last full checkpoint. 5. On success, marks the staging as ready. If the job fails or is cancelled, the staging directory is wiped so the diff --git a/docs/technical/architecture/subsystems/attributes/attributes.md b/docs/technical/architecture/subsystems/attributes/attributes.md index bb9544c9f4..eaa07771e5 100644 --- a/docs/technical/architecture/subsystems/attributes/attributes.md +++ b/docs/technical/architecture/subsystems/attributes/attributes.md @@ -96,12 +96,12 @@ Value: "production" Track whether a transaction has been reverted using an **in-memory bitset** (`ReversionBitset`). -Unlike other attributes, reversions are **not** stored as Pebble attributes under zone `0x01`. Instead, each ledger maintains a `[]uint64` bitset where bit N indicates whether transaction N has been reverted. Reversion words are persisted to Pebble under zone `0x03` (Per-Ledger) with key format `[0x03][0x01][ledgerID BE 4B][wordIndex BE 8 bytes]` via the `SaveReversionWord` function in `internal/infra/state/batch.go`. +Unlike other attributes, reversions are **not** stored as Pebble attributes under zone `0x01`. Instead, each ledger maintains a `[]uint64` bitset where bit N indicates whether transaction N has been reverted. Reversion words are persisted to Pebble under zone `0x03` (Per-Ledger) with key format `[0x03][0x01][ledgerName padded 64B][wordIndex BE 8 bytes]` via the `SaveReversionWord` function in `internal/infra/state/batch.go`. | Property | Description | |----------|-------------| -| **In-memory** | `map[uint32]*bitset.Bitset` -- one bitset per ledger (from `internal/pkg/bitset/bitset.go`) | -| **Pebble persistence** | Per-word in zone `0x03` (`ZonePerLedger` + `SubPLReversions`), key: `[0x03][0x01][ledgerID BE 4B][wordIndex BE 8]`, value: `[uint64 LE 8]` | +| **In-memory** | `map[string]*bitset.Bitset` -- one bitset per ledger, keyed by ledger name (from `internal/pkg/bitset/bitset.go`) | +| **Pebble persistence** | Per-word in zone `0x03` (`ZonePerLedger` + `SubPLReversions`), key: `[0x03][0x01][ledgerName padded 64B][wordIndex BE 8]`, value: `[uint64 LE 8]` | | **Lookup** | O(1) -- `words[txID/64] & (1 << (txID%64))` | | **Memory** | 1 bit per transaction (vs ~82 bytes per entry with the old KeyStore approach) | | **Restore** | Reconstructed from Pebble via `ReadReversions` (`internal/query/reversions.go`) on startup or snapshot restore | diff --git a/docs/technical/architecture/subsystems/checker/checker.md b/docs/technical/architecture/subsystems/checker/checker.md index ccab04e525..05c6798ce0 100644 --- a/docs/technical/architecture/subsystems/checker/checker.md +++ b/docs/technical/architecture/subsystems/checker/checker.md @@ -35,12 +35,13 @@ Each pass takes a persisted projection, re-derives the expected value by replayi | 2 | `compareVolumes` | Per-`(ledger, account, asset)` volume rows in the attribute store | `ReplayLedgerLog` + `ApplyPostings` over the audit-chain-bound orders | `VOLUME_MISMATCH` | | 3 | `compareMetadata` | Account/transaction metadata attribute rows | Replay of `SavedMetadata` / `DeletedMetadata` orders | `METADATA_MISMATCH` | | 4 | `compareTransactions` | Per-transaction state (postings, timestamp, reverted flag, fabricated/system) | Replay of `CreatedTransaction` / `RevertedTransaction` orders | `TRANSACTION_UPDATE_MISMATCH` | -| 5 | `checkReversionInvariants` | Each transaction is reverted at most once; the reversion bitset agrees with the replayed reverted flag | Replay-derived revert flags | `REVERTED_MISMATCH` | +| 5 | `checkReversionInvariants` | Log-stream consistency: each transaction is reverted at most once, and reverts target transactions that exist | Replay-derived revert flags | `REVERTED_MISMATCH` | | 6 | `verifySealingHash` | Each closed chapter's sealing hash equals `BLAKE3(chapter_id ‖ close_seq ‖ last_audit_hash ‖ state_hash)` | Recompute from the audit-chain-bound chapter close payload | `HASH_MISMATCH` (chapter-scoped) | | 7 | `compareExclusionProjections` | `AppliedProposal.TransientVolumes` and `LedgerLog.PurgedVolumes` agree with what `SimulateEphemeralPurge` would have produced | Replay + `SimulateEphemeralPurge` | `EXCLUSION_RECORD_MISMATCH` | | 8 | `compareIdempotencyOutcomes` | Frozen idempotency outcomes in `SubIdempKeys` match the outcome of the chain-bound `AuditSuccess` / `AuditFailure` that wrote them | Outcome map built during `verifyAuditHashChain` | `IDEMPOTENCY_MISMATCH` | | 9 | `compareIndexes` | `SubAttrIndex` registry matches the set derived from `CreateIndex` / `DropIndex` / `RemovedMetadataFieldType` / `DeleteLedger` logs (presence + identity only) | Replay of the index-affecting log types | `INDEX_MISMATCH` | | 10 | `compareMirrorV2LogID` | Stored `LedgerBoundaries.last_mirror_v2_log_id` **equals** the max audited `MirrorIngest.v2_log_id` per ledger (full equality) | Live audit chain (`recordMirrorIngestMutations`) over a baseline floor (`foldBaselineBoundaries`) | `MIRROR_V2LOGID_MISMATCH` (any divergence) | +| 11 | `compareReversions` | Stored reversion bitsets (`ZonePerLedger`/`SubPLReversions`, the rows the already-reverted gate reads) equal the audit-derived reverted set, both ways; stored rows for non-live ledgers and undecodable rows are flagged | Baseline tx-row markers + replayed `RevertedTransaction` logs | `REVERTED_MISMATCH` | Notes: @@ -76,7 +77,8 @@ flowchart TB B --> I[compareExclusionProjections] B --> J[compareIndexes] B --> L[compareMirrorV2LogID] - D & E & F & G & H & I & J & L --> K[stream errors as they happen] + B --> M[compareReversions] + D & E & F & G & H & I & J & L & M --> K[stream errors as they happen] C --> K ``` diff --git a/internal/application/check/checker.go b/internal/application/check/checker.go index 41802130e5..d5c8d88fbc 100644 --- a/internal/application/check/checker.go +++ b/internal/application/check/checker.go @@ -10,6 +10,7 @@ import ( "io" "maps" "math/big" + "math/bits" "slices" "strconv" "time" @@ -664,6 +665,10 @@ func (c *Checker) Check(ctx context.Context, callback func(*servicepb.CheckStore return err } + if err := c.compareReversions(snap, ledgerRevertedTxIDs, knownLedgers, callback); err != nil { + return err + } + return nil } @@ -5053,6 +5058,83 @@ func errorEventWithTx(errorType servicepb.CheckStoreErrorType, message, ledger s } } +// compareReversions verifies the persisted reversion bitsets +// (ZonePerLedger/SubPLReversions) — the projection the FSM's already-reverted +// gate reads — against the reverted-transaction set derived from the audit: +// baseline tx-row markers (seedTxTrackingFromBaseline) plus the replayed +// RevertedTransaction logs. That derivation is complete across archived +// chapters, so the comparison is exact equality both ways: a bit the audit +// set but the store lost re-admits a double revert (double refund) past the +// gate, and a stored bit the audit never set silently blocks a legitimate +// revert. +// +// The comparison is driven purely by audit-derived state (knownLedgers, +// derived): no persisted marker — pending-cleanup included, since it is +// itself an unverified projection — may exempt an audit-live ledger from the +// check. Stored rows for ledgers the audit does not know as live are flagged +// too: DeleteLedger deletes the rows at apply time on both the live path and +// the replay, so nothing legitimately lingers. Rows that fail to decode are +// reported rather than silently narrowing the comparison. +func (c *Checker) compareReversions(reader dal.PebbleReader, derived map[string]*bitset.Bitset, knownLedgers map[string]struct{}, callback func(*servicepb.CheckStoreEvent)) error { + stored, malformed, err := query.ReadReversions(reader) + if err != nil { + return fmt.Errorf("reading stored reversion bitsets: %w", err) + } + + for _, m := range malformed { + callback(errorEventWithTx(servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_REVERTED_MISMATCH, + fmt.Sprintf("malformed reversion row at key %x: %s", m.Key, m.Reason), "", 0)) + } + + for name := range stored { + if _, live := knownLedgers[name]; !live { + callback(errorEventWithTx(servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_REVERTED_MISMATCH, + fmt.Sprintf("stored reversion rows for non-live ledger %q: DeleteLedger removes them at apply, so they are unaudited leftovers", name), + name, 0)) + } + } + + for name := range knownLedgers { + var derivedWords, storedWords []uint64 + if bs := derived[name]; bs != nil { + derivedWords = bs.Words() + } + + if bs := stored[name]; bs != nil { + storedWords = bs.Words() + } + + words := max(len(derivedWords), len(storedWords)) + for i := range words { + var d, s uint64 + if i < len(derivedWords) { + d = derivedWords[i] + } + + if i < len(storedWords) { + s = storedWords[i] + } + + for diff := d ^ s; diff != 0; diff &= diff - 1 { + bit := bits.TrailingZeros64(diff) + txID := uint64(i)*64 + uint64(bit) + + if d&(1< 0 { + _ = batch.Cancel() + + return fmt.Errorf("seeding reversion bitsets: %d malformed row(s), first at key %x: %s", + len(malformedReversions), malformedReversions[0].Key, malformedReversions[0].Reason) + } + + writer.reversions = seededReversions + logCursor, err := query.ReadLogsSince(ctx, readHandle, fromLogSeq) if err != nil { _ = batch.Cancel() @@ -429,6 +452,12 @@ func RebuildDelta( return fmt.Errorf("flushing rebuilt boundaries: %w", err) } + if err := writer.flushReversions(); err != nil { + _ = writer.batch.Cancel() + + return fmt.Errorf("flushing rebuilt reversion bitsets: %w", err) + } + if err := writer.batch.Commit(); err != nil { return fmt.Errorf("committing boundaries batch: %w", err) } @@ -695,6 +724,16 @@ type attributeReplayWriter struct { // commit (net counts are derived from the committed 0xF1 state). boundaries map[string]*raftcmdpb.LedgerBoundaries readHandle dal.PebbleReader + + // Reversion bitsets per ledger (ZonePerLedger/SubPLReversions). The FSM's + // already-reverted gate reads these — not the tx rows' + // RevertedByTransaction markers — so every replayed RevertedTransaction + // must fold into them or a restored node re-admits reverts of + // already-reverted transactions. Seeded from the checkpoint rows; ledgers + // touched by the replay are flushed by flushReversions alongside the + // boundaries. + reversions map[string]*bitset.Bitset + dirtyReversions map[string]struct{} } // applyAuditOrderEffects folds order-level boundary effects that the ledger-log @@ -800,6 +839,17 @@ func (w *attributeReplayWriter) deleteLedger(name string, deletedAt *commonpb.Ti delete(w.boundaries, name) + // Unlike the rest of the per-ledger data (deferred to the covering + // purge), the live path deletes the reversion rows at DeleteLedger apply + // (WriteSet.Merge) — mirror it so the restored store does not resurrect + // them into Registry.Reversions on boot. + delete(w.reversions, name) + delete(w.dirtyReversions, name) + + if err := state.DeleteReversionsByLedger(w.batch, name); err != nil { + return fmt.Errorf("deleting reversions for ledger %q: %w", name, err) + } + return state.SavePendingLedgerCleanup(w.batch, name, seq) } @@ -957,6 +1007,32 @@ func countAttributeKeys[V proto.Message](attr *attributes.Attribute[V], reader d return count, si.Err() } +// flushReversions writes the reversion bitset words of every ledger the +// replay reverted in, in the same [zone][sub][ledger][wordIndex] layout the +// FSM persists (SaveReversionWord). Untouched ledgers keep their +// checkpoint-time rows; zero words are skipped — a missing row reads as an +// all-zero word (query.ReadReversions). +func (w *attributeReplayWriter) flushReversions() error { + for name := range w.dirtyReversions { + bs := w.reversions[name] + if bs == nil { + continue + } + + for i, word := range bs.Words() { + if word == 0 { + continue + } + + if err := state.SaveReversionWord(w.batch, name, uint64(i), word); err != nil { + return fmt.Errorf("writing reversion word for ledger %q: %w", name, err) + } + } + } + + return nil +} + // flushBoundaries writes every touched ledger's boundaries to the SubAttrBoundary // attribute. Called after countNetAttributes, in its own post-commit batch. func (w *attributeReplayWriter) flushBoundaries() error { @@ -1135,6 +1211,22 @@ func (w *attributeReplayWriter) SetRevertedBy(canonicalKey []byte, revertTxID ui w.pendingTx[string(canonicalKey)] = existing + // Fold the reverted id into the ledger's reversion bitset — the structure + // the FSM's already-reverted gate actually reads. + var tk domain.TransactionKey + if err := tk.Unmarshal(canonicalKey); err != nil { + return fmt.Errorf("unmarshaling reverted transaction key: %w", err) + } + + bs := w.reversions[tk.LedgerName] + if bs == nil { + bs = &bitset.Bitset{} + w.reversions[tk.LedgerName] = bs + } + + bs.Set(tk.ID) + w.dirtyReversions[tk.LedgerName] = struct{}{} + return nil } diff --git a/internal/infra/backup/rebuild_test.go b/internal/infra/backup/rebuild_test.go index c3bc0da5c5..b08b86b31a 100644 --- a/internal/infra/backup/rebuild_test.go +++ b/internal/infra/backup/rebuild_test.go @@ -11,6 +11,8 @@ import ( "github.com/formancehq/ledger/v3/internal/domain" "github.com/formancehq/ledger/v3/internal/infra/attributes" + "github.com/formancehq/ledger/v3/internal/infra/state" + "github.com/formancehq/ledger/v3/internal/pkg/bitset" "github.com/formancehq/ledger/v3/internal/proto/auditpb" "github.com/formancehq/ledger/v3/internal/proto/commonpb" "github.com/formancehq/ledger/v3/internal/proto/raftcmdpb" @@ -717,19 +719,21 @@ func newAttributeReplayWriter(t *testing.T) (*attributeReplayWriter, *attributes t.Cleanup(func() { _ = readHandle.Close() }) writer := &attributeReplayWriter{ - store: store, - batch: store.OpenWriteSession(), - volume: attrs.Volume, - metadata: attrs.Metadata, - tx: attrs.Transaction, - ledger: attrs.Ledger, - references: attrs.References, - boundary: attrs.Boundary, - pendingVolumes: make(map[string]*raftcmdpb.VolumePair), - pendingTx: make(map[string]*commonpb.TransactionState), - ledgerInfos: make(map[string]*commonpb.LedgerInfo), - boundaries: make(map[string]*raftcmdpb.LedgerBoundaries), - readHandle: readHandle, + store: store, + batch: store.OpenWriteSession(), + volume: attrs.Volume, + metadata: attrs.Metadata, + tx: attrs.Transaction, + ledger: attrs.Ledger, + references: attrs.References, + boundary: attrs.Boundary, + pendingVolumes: make(map[string]*raftcmdpb.VolumePair), + pendingTx: make(map[string]*commonpb.TransactionState), + ledgerInfos: make(map[string]*commonpb.LedgerInfo), + boundaries: make(map[string]*raftcmdpb.LedgerBoundaries), + reversions: make(map[string]*bitset.Bitset), + dirtyReversions: make(map[string]struct{}), + readHandle: readHandle, } t.Cleanup(func() { _ = writer.batch.Cancel() }) @@ -928,3 +932,33 @@ func TestRebuildDelta_CountsDoNotBleedAcrossPrefixLedgers(t *testing.T) { require.Equal(t, uint64(2), boundary.GetVolumeCount(), `"pay" must count only its own rows (world + alice), not "payments" rows too`) } + +// TestAttributeReplayWriter_DeleteLedgerRemovesReversionRows: DeleteLedger +// replay must delete the ledger's persisted reversion words — the live path +// does so at apply time (not at the covering purge), so leaving them would +// resurrect a deleted ledger's bitset into Registry.Reversions on boot. +func TestAttributeReplayWriter_DeleteLedgerRemovesReversionRows(t *testing.T) { + t.Parallel() + + writer, _, store := newAttributeReplayWriter(t) + + // Checkpoint-time reversion rows for the ledger about to be deleted. + require.NoError(t, state.SaveReversionWord(writer.batch, "doomed", 0, 1<<3)) + require.NoError(t, state.SaveReversionWord(writer.batch, "doomed", 1, 1<<7)) + + writer.ledgerInfos["doomed"] = &commonpb.LedgerInfo{Name: "doomed"} + writer.reversions["doomed"] = &bitset.Bitset{} + + require.NoError(t, writer.deleteLedger("doomed", &commonpb.Timestamp{Data: 42}, 7)) + require.NoError(t, writer.batch.Commit()) + + handle, err := store.NewReadHandle() + require.NoError(t, err) + + defer func() { _ = handle.Close() }() + + bs, err := query.ReadReversionBitset(handle, "doomed") + require.NoError(t, err) + require.Empty(t, bs.Words(), "deleted ledger's reversion rows must not survive the replay") + require.Empty(t, writer.reversions["doomed"]) +} diff --git a/internal/infra/state/batch.go b/internal/infra/state/batch.go index 979f0ef257..fc898cc3dc 100644 --- a/internal/infra/state/batch.go +++ b/internal/infra/state/batch.go @@ -240,7 +240,7 @@ func batchDeleteQueryCheckpointSchedule(b *dal.WriteSession) error { // SaveReversionWord writes a single bitset word for a ledger. // Key: [0x03][0x01][ledgerName padded 64B][wordIndex BE 8 bytes] → [uint64 LE 8 bytes]. -func saveReversionWord(b *dal.WriteSession, ledgerName string, wordIndex uint64, value uint64) error { +func SaveReversionWord(b *dal.WriteSession, ledgerName string, wordIndex uint64, value uint64) error { b.KeyBuilder.PutZonePrefix(dal.ZonePerLedger, dal.SubPLReversions). PutLedgerNameFixed(ledgerName). PutUint64(wordIndex) @@ -253,7 +253,7 @@ func saveReversionWord(b *dal.WriteSession, ledgerName string, wordIndex uint64, } // DeleteReversionsByLedger removes all reversion words for a ledger. -func deleteReversionsByLedger(b *dal.WriteSession, ledgerName string) error { +func DeleteReversionsByLedger(b *dal.WriteSession, ledgerName string) error { start := buildLedgerScopedPrefix(dal.ZonePerLedger, dal.SubPLReversions, ledgerName) end := buildLedgerScopedPrefixSuccessor(dal.ZonePerLedger, dal.SubPLReversions, ledgerName) diff --git a/internal/infra/state/cache_snapshotter_test.go b/internal/infra/state/cache_snapshotter_test.go index f12f92ea1b..9d8825d5db 100644 --- a/internal/infra/state/cache_snapshotter_test.go +++ b/internal/infra/state/cache_snapshotter_test.go @@ -64,7 +64,7 @@ func persistToStore(s *CacheSnapshotter, sessions dal.WriteSessionFactory) error for ledger, bs := range s.registry.Reversions { for i := range bs.WordCount() { - if err := saveReversionWord(batch, ledger, i, bs.Word(i)); err != nil { + if err := SaveReversionWord(batch, ledger, i, bs.Word(i)); err != nil { return fmt.Errorf("saving reversion word for %q: %w", ledger, err) } } diff --git a/internal/infra/state/recovery.go b/internal/infra/state/recovery.go index b2f897dea8..ddfafb13d1 100644 --- a/internal/infra/state/recovery.go +++ b/internal/infra/state/recovery.go @@ -119,11 +119,18 @@ func (r *Recovery) RecoverState() error { r.apply.Chapters.SetSchedule(chapterSchedule) - reversions, err := query.ReadReversions(handle) + reversions, malformedReversions, err := query.ReadReversions(handle) if err != nil { return fmt.Errorf("recovering reversions: %w", err) } + // Only SaveReversionWord produces these rows, so a malformed one means + // corruption or tampering. Boot proceeds on the decodable words — the + // checker reports malformed rows as integrity events — but never silently. + for _, m := range malformedReversions { + r.apply.logger.Errorf("malformed reversion row at key %x skipped during recovery: %s", m.Key, m.Reason) + } + r.apply.Registry.Reversions = reversions if r.apply.keyStore != nil { diff --git a/internal/infra/state/write_set.go b/internal/infra/state/write_set.go index def61311b3..b8095cc133 100644 --- a/internal/infra/state/write_set.go +++ b/internal/infra/state/write_set.go @@ -541,7 +541,7 @@ func (b *WriteSet) Merge(batch *dal.WriteSession, logsOrRefs []*raftcmdpb.Create // SubPLReversions (0x01) for _, dw := range dirtyWords { word := b.fsm.Registry.Reversions[dw.ledgerName].Word(dw.wordIndex) - if err := saveReversionWord(batch, dw.ledgerName, dw.wordIndex, word); err != nil { + if err := SaveReversionWord(batch, dw.ledgerName, dw.wordIndex, word); err != nil { return fmt.Errorf("saving reversion word for %q: %w", dw.ledgerName, err) } } @@ -767,7 +767,7 @@ func (b *WriteSet) Merge(batch *dal.WriteSession, logsOrRefs []*raftcmdpb.Create // Clean in-memory reversion bitset and Pebble words — not needed after deletion. delete(b.fsm.Registry.Reversions, ledgerName) - if err := deleteReversionsByLedger(batch, ledgerName); err != nil { + if err := DeleteReversionsByLedger(batch, ledgerName); err != nil { return fmt.Errorf("deleting reversions for %q: %w", ledgerName, err) } } diff --git a/internal/query/reversions.go b/internal/query/reversions.go index d56b6df242..d169dab133 100644 --- a/internal/query/reversions.go +++ b/internal/query/reversions.go @@ -12,9 +12,19 @@ import ( "github.com/formancehq/ledger/v3/internal/storage/readstore" ) +// MalformedReversionRow describes a SubPLReversions row whose key or value +// does not decode as a reversion word. The runtime readers skip such rows; +// they are reported so the checker can surface them as integrity events +// instead of silently narrowing the comparison, and so recovery can log them. +type MalformedReversionRow struct { + Key []byte + Reason string +} + // ReadReversions loads all per-ledger reversion bitsets from Pebble. // Key format: [0x03][0x01][ledgerName padded 64B][wordIndex BE 8 bytes] → [uint64 LE 8 bytes]. -func ReadReversions(reader dal.PebbleReader) (map[string]*bitset.Bitset, error) { +// Rows that do not decode are skipped and returned in the malformed slice. +func ReadReversions(reader dal.PebbleReader) (map[string]*bitset.Bitset, []MalformedReversionRow, error) { lowerBound := []byte{dal.ZonePerLedger, dal.SubPLReversions} upperBound := []byte{dal.ZonePerLedger, dal.SubPLReversions + 1} @@ -23,17 +33,24 @@ func ReadReversions(reader dal.PebbleReader) (map[string]*bitset.Bitset, error) UpperBound: upperBound, }) if err != nil { - return nil, fmt.Errorf("creating reversions iterator: %w", err) + return nil, nil, fmt.Errorf("creating reversions iterator: %w", err) } defer func() { _ = iter.Close() }() result := make(map[string]*bitset.Bitset) + var malformed []MalformedReversionRow + for iter.First(); iter.Valid(); iter.Next() { key := iter.Key() // Key format: [0x03][0x01][ledgerName padded 64B][wordIndex BE 8 bytes]. if len(key) < 2+dal.LedgerNameFixedSize+8 { + malformed = append(malformed, MalformedReversionRow{ + Key: append([]byte(nil), key...), + Reason: "key shorter than the reversion word layout", + }) + continue } @@ -49,10 +66,15 @@ func ReadReversions(reader dal.PebbleReader) (map[string]*bitset.Bitset, error) val, err := iter.ValueAndErr() if err != nil { - return nil, fmt.Errorf("reading reversion word for ledger %q: %w", ledgerName, err) + return nil, nil, fmt.Errorf("reading reversion word for ledger %q: %w", ledgerName, err) } if len(val) < 8 { + malformed = append(malformed, MalformedReversionRow{ + Key: append([]byte(nil), key...), + Reason: "value shorter than a bitset word", + }) + continue } @@ -67,7 +89,11 @@ func ReadReversions(reader dal.PebbleReader) (map[string]*bitset.Bitset, error) bs.SetWord(wordIndex, word) } - return result, nil + if err := iter.Error(); err != nil { + return nil, nil, fmt.Errorf("iterating reversions: %w", err) + } + + return result, malformed, nil } // ReadReversionBitset loads a single ledger's reversion bitset from Pebble. @@ -112,5 +138,9 @@ func ReadReversionBitset(reader dal.PebbleReader, ledgerName string) (*bitset.Bi bs.SetWord(wordIndex, binary.LittleEndian.Uint64(val)) } + if err := iter.Error(); err != nil { + return nil, fmt.Errorf("iterating reversions for ledger %q: %w", ledgerName, err) + } + return bs, nil } diff --git a/tests/e2e/cluster/restore_reversion_test.go b/tests/e2e/cluster/restore_reversion_test.go new file mode 100644 index 0000000000..b72748936a --- /dev/null +++ b/tests/e2e/cluster/restore_reversion_test.go @@ -0,0 +1,338 @@ +//go:build e2e && s3 + +package cluster + +import ( + "context" + "math/big" + "os" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" + logging "github.com/formancehq/go-libs/v5/pkg/observe/log" + "github.com/formancehq/go-libs/v5/pkg/testing/testservice" + cmdserver "github.com/formancehq/ledger/v3/cmd/server" + "github.com/formancehq/ledger/v3/internal/proto/clusterpb" + "github.com/formancehq/ledger/v3/internal/proto/commonpb" + "github.com/formancehq/ledger/v3/internal/proto/restorepb" + "github.com/formancehq/ledger/v3/internal/proto/servicepb" + "github.com/formancehq/ledger/v3/pkg/actions" + "github.com/formancehq/ledger/v3/pkg/testserver" + "github.com/formancehq/ledger/v3/tests/e2e/testutil" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" + "google.golang.org/grpc" +) + +// This suite pins the reversion-state loss the model test surfaced (a revert +// the model knew was already applied got re-attempted after a restore). The +// FSM's already-reverted gate reads the per-ledger reversion bitset +// (ZonePerLedger/SubPLReversions), which only the FSM hot path persists; the +// restore rebuild replays RevertedTransaction logs into the per-transaction +// RevertedByTransaction markers but must also rebuild the bitset — otherwise +// every revert inside the exported delta is forgotten by the gate, and a +// second revert of the same transaction passes it. When the account still +// holds funds, that second revert SUCCEEDS and moves them again (a double +// refund); otherwise it surfaces as a wrong rejection (insufficient funds / +// account-type mismatch) instead of TRANSACTION_ALREADY_REVERTED. +var _ = Describe("Restore reversion bitset", Ordered, func() { + const ( + httpPort = testutil.TestSingleHTTPPort + grpcPort = testutil.TestSingleGRPCPort + raftPort = grpcPort - 1000 + ledgerName = "revbits-ledger" + s3Bucket = "restore-reversion-bitset" + clusterID = "revbits-cluster" + + account = "acc:1" + // revertedTxID is funded first (id 1) and reverted pre-backup; fundTxID + // (id 2) keeps the account solvent so a post-restore double revert + // would succeed rather than trip on the balance check. + revertedTxID = 1 + ) + + var ( + ctx context.Context + restoreWalDir string + restoreDataDir string + minioEndpoint string + ) + + // expectAlreadyReverted asserts a second revert of revertedTxID is + // rejected by the already-reverted gate. + expectAlreadyReverted := func(client servicepb.BucketServiceClient, phase string) { + _, err := client.Apply(ctx, servicepb.UnsignedApplyRequest("", + actions.RevertTransactionAction(ledgerName, revertedTxID, false, false, nil), + )) + Expect(err).To(HaveOccurred(), "%s: double revert of tx %d must be rejected", phase, revertedTxID) + Expect(err.Error()).To(ContainSubstring("already reverted"), "%s: rejection reason", phase) + } + + // expectAccountVolumes asserts the account's USD volumes: 600 in + // (100 + 500 funding), 100 out (the single legitimate revert). + expectAccountVolumes := func(client servicepb.BucketServiceClient, phase string) { + acct, err := actions.GetAccount(ctx, client, ledgerName, account) + Expect(err).To(Succeed()) + + vol := acct.GetVolumes()["USD"] + Expect(vol).ToNot(BeNil(), "%s: %s USD volumes missing", phase, account) + Expect(vol.GetInput()).To(Equal("600"), "%s: %s USD input", phase, account) + Expect(vol.GetOutput()).To(Equal("100"), "%s: %s USD output", phase, account) + } + + storage := func() *commonpb.BackupStorage { + return testutil.S3BackupStorage(&commonpb.S3StorageConfig{ + Bucket: s3Bucket, + Region: restoreS3Region, + Endpoint: minioEndpoint, + }) + } + + BeforeAll(func() { + ctx = logging.TestingContext() + + container, err := testcontainers.Run(context.Background(), "minio/minio:latest", + testcontainers.WithEnv(map[string]string{ + "MINIO_ROOT_USER": restoreMinioAccessKey, + "MINIO_ROOT_PASSWORD": restoreMinioSecretKey, + }), + testcontainers.WithCmd("server", "/data"), + testcontainers.WithExposedPorts("9000/tcp"), + testcontainers.WithWaitStrategy( + wait.ForHTTP("/minio/health/live").WithPort("9000/tcp").WithStartupTimeout(30*time.Second), + ), + ) + Expect(err).To(Succeed()) + DeferCleanup(func() { _ = container.Terminate(context.Background()) }) + + minioEndpoint, err = container.Endpoint(context.Background(), "http") + Expect(err).To(Succeed()) + + cfg, err := awsconfig.LoadDefaultConfig(context.Background(), + awsconfig.WithRegion(restoreS3Region), + awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider( + restoreMinioAccessKey, restoreMinioSecretKey, "", + )), + ) + Expect(err).To(Succeed()) + + s3Client := s3.NewFromConfig(cfg, func(o *s3.Options) { + o.BaseEndpoint = aws.String(minioEndpoint) + o.UsePathStyle = true + }) + _, err = s3Client.CreateBucket(context.Background(), &s3.CreateBucketInput{Bucket: aws.String(s3Bucket)}) + Expect(err).To(Succeed()) + + GinkgoT().Setenv("AWS_ACCESS_KEY_ID", restoreMinioAccessKey) + GinkgoT().Setenv("AWS_SECRET_ACCESS_KEY", restoreMinioSecretKey) + + restoreWalDir, err = os.MkdirTemp("", "revbits-restore-wal-*") + Expect(err).To(Succeed()) + restoreDataDir, err = os.MkdirTemp("", "revbits-restore-data-*") + Expect(err).To(Succeed()) + }) + + Describe("Phase 1: a reverted transaction in the exported delta", Ordered, func() { + var ( + sourceServer *testservice.Service + client servicepb.BucketServiceClient + clusterClient clusterpb.ClusterServiceClient + grpcConn *grpc.ClientConn + ) + + BeforeAll(func() { + instruments := testserver.DefaultTestInstruments(testserver.TestNodeConfig{ + NodeID: 1, + ClusterID: clusterID, + HTTPPort: httpPort, + RaftPort: raftPort, + GRPCPort: grpcPort, + WalDir: GinkgoT().TempDir(), + DataDir: GinkgoT().TempDir(), + Debug: testutil.Debug, + Output: GinkgoWriter, + }) + instruments = append(instruments, testserver.WithBootstrap()) + + sourceServer = testservice.New(cmdserver.NewRunCommand, testservice.WithInstruments(instruments...)) + Expect(sourceServer.Start(ctx)).To(Succeed()) + + var err error + client, clusterClient, grpcConn, err = testutil.NewGRPCClient(grpcPort) + Expect(err).To(Succeed()) + + Eventually(func(g Gomega) bool { + state, err := clusterClient.GetClusterState(ctx, &clusterpb.GetClusterStateRequest{}) + g.Expect(err).To(Succeed()) + return state.Leader != 0 + }).Within(10 * time.Second).ProbeEvery(100 * time.Millisecond).Should(BeTrue()) + + // Full checkpoint on the EMPTY store: everything after it lands in + // the incremental delta, so the restore reconstructs the reversion + // state purely by replaying the exported log instead of copying + // checkpoint files. + backupResp, err := clusterClient.Backup(ctx, &clusterpb.BackupRequest{Storage: storage()}) + Expect(err).To(Succeed()) + Expect(backupResp.GetTotalFiles()).To(BeNumerically(">", 0)) + }) + + AfterAll(func() { + _ = grpcConn.Close() + stopCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + Expect(sourceServer.Stop(stopCtx)).To(Succeed()) + }) + + It("creates, funds, and reverts a transaction", func() { + _, err := client.Apply(ctx, servicepb.UnsignedApplyRequest("", + actions.CreateLedgerAction(ledgerName, nil), + )) + Expect(err).To(Succeed()) + + // tx 1: the transaction that gets reverted. + _, err = client.Apply(ctx, servicepb.UnsignedApplyRequest("", + actions.CreateTransactionAction(ledgerName, []*commonpb.Posting{ + actions.NewPosting("world", account, big.NewInt(100), "USD"), + }, nil, nil), + )) + Expect(err).To(Succeed()) + + // tx 2: independent funding, so the account stays solvent after + // the revert and a double revert would not trip the balance check. + _, err = client.Apply(ctx, servicepb.UnsignedApplyRequest("", + actions.CreateTransactionAction(ledgerName, []*commonpb.Posting{ + actions.NewPosting("world", account, big.NewInt(500), "USD"), + }, nil, nil), + )) + Expect(err).To(Succeed()) + + // Revert tx 1 (creates tx 3, moving the 100 back to world). + _, err = client.Apply(ctx, servicepb.UnsignedApplyRequest("", + actions.RevertTransactionAction(ledgerName, revertedTxID, false, false, nil), + )) + Expect(err).To(Succeed()) + }) + + It("rejects a double revert on the live node", func() { + // Premise guard: if the live gate did not hold, the post-restore + // assertions would be vacuous. + expectAlreadyReverted(client, "live") + expectAccountVolumes(client, "live") + }) + + It("exports the delta", func() { + incResp, err := clusterClient.IncrementalBackup(ctx, &clusterpb.IncrementalBackupRequest{Storage: storage()}) + Expect(err).To(Succeed()) + Expect(incResp.GetLogEntriesExported()).To(BeNumerically(">", 0)) + }) + }) + + Describe("Phase 2: restore", Ordered, func() { + var ( + restoreClient restorepb.RestoreServiceClient + grpcConn *grpc.ClientConn + server *testservice.Service + ) + + BeforeAll(func() { + server = testservice.New(cmdserver.NewRunCommand, + testservice.WithInstruments( + testservice.DebugInstrumentation(testutil.Debug), + testservice.OutputInstrumentation(GinkgoWriter), + testserver.WithNodeID(1), + testserver.WithClusterID(clusterID), + testserver.WithHTTPPort(httpPort), + testserver.WithWalDir(restoreWalDir), + testserver.WithDataDir(restoreDataDir), + testserver.WithRaftPort(raftPort), + testserver.WithGRPCPort(grpcPort), + testserver.WithRestore(), + ), + ) + Expect(server.Start(ctx)).To(Succeed()) + + var err error + restoreClient, grpcConn, err = newRestoreGRPCClient(grpcPort) + Expect(err).To(Succeed()) + }) + + AfterAll(func() { + _ = grpcConn.Close() + stopCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + Expect(server.Stop(stopCtx)).To(Succeed()) + }) + + It("downloads and finalizes the backup", func() { + startResp, err := restoreClient.StartDownloadBackup(ctx, &restorepb.StartDownloadBackupRequest{Storage: storage()}) + Expect(err).To(Succeed()) + + Eventually(func() restorepb.DownloadState { + resp, statusErr := restoreClient.GetDownloadStatus(ctx, &restorepb.GetDownloadStatusRequest{JobId: startResp.GetJobId()}) + Expect(statusErr).To(Succeed()) + return resp.GetState() + }, 2*time.Minute, 500*time.Millisecond).Should(Equal(restorepb.DownloadState_DOWNLOAD_STATE_SUCCEEDED)) + + _, err = restoreClient.FinalizeRestore(ctx, &restorepb.FinalizeRestoreRequest{}) + Expect(err).To(Succeed()) + }) + }) + + Describe("Phase 3: verify the restored reversion state", Ordered, func() { + var ( + client servicepb.BucketServiceClient + grpcConn *grpc.ClientConn + server *testservice.Service + ) + + BeforeAll(func() { + instruments := testserver.DefaultTestInstruments(testserver.TestNodeConfig{ + NodeID: 1, + ClusterID: clusterID, + HTTPPort: httpPort, + RaftPort: raftPort, + GRPCPort: grpcPort, + WalDir: restoreWalDir, + DataDir: restoreDataDir, + Debug: testutil.Debug, + Output: GinkgoWriter, + }) + instruments = append(instruments, testserver.WithBootstrap()) + + server = testservice.New(cmdserver.NewRunCommand, testservice.WithInstruments(instruments...)) + Expect(server.Start(ctx)).To(Succeed()) + + var clusterClient clusterpb.ClusterServiceClient + var err error + client, clusterClient, grpcConn, err = testutil.NewGRPCClient(grpcPort) + Expect(err).To(Succeed()) + + Eventually(func(g Gomega) bool { + state, err := clusterClient.GetClusterState(ctx, &clusterpb.GetClusterStateRequest{}) + g.Expect(err).To(Succeed()) + return state.Leader != 0 + }).Within(10 * time.Second).ProbeEvery(100 * time.Millisecond).Should(BeTrue()) + }) + + AfterAll(func() { + _ = grpcConn.Close() + stopCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + Expect(server.Stop(stopCtx)).To(Succeed()) + _ = os.RemoveAll(restoreWalDir) + _ = os.RemoveAll(restoreDataDir) + }) + + It("still rejects a double revert after the rebuild", func() { + expectAlreadyReverted(client, "restored") + // A successful double revert would have moved another 100 out of + // the account; the volumes pin that no funds moved. + expectAccountVolumes(client, "restored") + }) + }) +})