diff --git a/AGENTS.md b/AGENTS.md index 9ba3e493f8..9de2546792 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. 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); extend the list as new persisted projections land. +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. 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), `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); extend the list as new persisted projections land. 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/technical/architecture/audit-vs-technical-state.md b/docs/technical/architecture/audit-vs-technical-state.md index bd945725f2..e2bf9d874a 100644 --- a/docs/technical/architecture/audit-vs-technical-state.md +++ b/docs/technical/architecture/audit-vs-technical-state.md @@ -93,17 +93,40 @@ manual repair, or future tooling. The cursor is wrong in **both** directions: - A cursor that advances **beyond** the highest audited `MirrorIngest.v2LogId` makes the worker skip source logs without any audit entry showing the missing proposals — silent under-ingestion. -- A cursor that drops **below** the highest audited `v2LogId` is equally unsafe. - `Worker.processBatch` (`internal/application/mirror/worker.go`) reloads the - cursor and calls `FetchLogs(cursor, ...)`, which returns source logs strictly - *after* the cursor. `TranslateBatch` then emits fresh `MirrorIngest` orders for - logs that were already ingested, and `processMirrorCreatedTransaction` - (`internal/domain/processing/processor_mirror.go`) reapplies every posting with - `force=true`, overwriting the transaction state and re-adding the amounts to the - volume pair via `applyPosting`. There is **no deduplication on `v2LogId`** and - volume mutation is additive, so a lowered cursor replays already-audited - transactions and applies their balance effects a second time — silent - double-application. +- A cursor that drops **below** the highest audited `v2LogId` makes + `Worker.processBatch` (`internal/application/mirror/worker.go`) reload the + cursor and call `FetchLogs(cursor, ...)`, which returns source logs strictly + *after* the cursor. `TranslateBatch` then re-emits `MirrorIngest` orders for + logs that were already ingested. **The FSM now enforces a contiguous applied + prefix (EN-1550).** `processMirrorIngest` + (`internal/domain/processing/processor_mirror.go`) records the highest applied + source id in `LedgerBoundaries.last_mirror_v2_log_id` — a true contiguous + prefix, since the worker ingests contiguously including `FillGap` orders — and, + *before* applying any posting or mutating any state, first rejects a malformed + `v2LogId == 0` (source v2 log ids are 1-based, so 0 is tamper/corruption) with + `ErrMirrorV2LogIDInvalid` (`KindInternal`) — it is never applied, since a 0 is + never recorded as `last` and a silently-applied 0 would be re-appliable forever. + It then decides three ways against the next slot (`expected = last + 1`): + - `v2LogId <= last`: already applied. Idempotent **no-op** — return `(nil, nil)`, + which `ProcessOrders` treats as "no log" (no sequence id consumed, no + audit-visible `Log`, no sink absorb). Postings are not re-forced, so balances + cannot double (pre-EN-1550 this replay caused silent **double-application** via + `force=true` additive volume mutation with no dedup on `v2LogId`). + - `v2LogId == expected`: the next contiguous log. Apply and advance `last`. + - `v2LogId > expected`: a **gap** — the cursor/high-water mark is ahead of the + applied prefix. Impossible under contiguous ingestion, so it is + corruption/tampering. The FSM **fails loud** (`ErrMirrorV2LogIDGap`, + `KindInternal`) and mutates nothing — it never silently applies past the gap + (which would desync nodes) or skips it. The rejection is deterministic (same + input → same rejection on every node); the worker surfaces it as a repeating + apply error rather than corrupting state. + + The guard is a pure function of applied per-ledger state; v2 log ids are 1-based + and strictly increasing per source (`TranslateBatch`), so it is FSM-deterministic + and needs no new preload/coverage key (it lives inside the already-covered + boundaries). Only the **behind** direction is handled here; recovering a cursor + that is *ahead* of the true source head is worker-side source-head recovery and + remains out of scope. Because it is only correct at one exact value, treat `MirrorCursor` as a persisted per-ledger projection (`ZonePerLedger` / `SubPLMirrorCursor`, in the @@ -138,6 +161,40 @@ every node identically. Mirror recovery, repair tooling, or a checker pass that re-derives the expected cursor from the audited mirror-ingest logs must enforce the equality before the stored cursor can be trusted. +`LedgerBoundaries.last_mirror_v2_log_id` (the EN-1550 idempotency high-water +mark) **is** covered by the checker as a full invariant-#8 **equality** check. +`compareMirrorV2LogID` (`internal/application/check/checker.go`) verifies, per +ledger, that the stored `last_mirror_v2_log_id` **equals** `max(audited +MirrorIngest.v2_log_id)`, emitting `CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH` +on **any** divergence. Because the FSM enforces a contiguous applied prefix (the +gap reject above), the persisted high-water mark must be exactly the max audited +`v2_log_id` at rest — no more, no less: +- `stored > max`: claims a source v2 log no audit entry recorded (future ingests + wrongly skipped). +- `stored < max`: the projection lost applied ground (an already-applied ingest + would be re-applied). + +The audited max is derived from the live audit chain +(`recordMirrorIngestMutations`) layered over a baseline floor +(`foldBaselineBoundaries` seeds it from the archived +`LedgerBoundaries.last_mirror_v2_log_id`, which the compact baseline snapshot now +carries — `writeBaselineAttributes` includes `Boundary` rows), so a ledger whose +mirror ingests live only in an archived chapter is not undercounted. A regular +(never-mirrored) ledger has stored `0` and audited max `0` → equal → not flagged. +The comparison is driven from the **union** of stored boundary rows and +audited-mirror ledgers, so a mirror ledger with audited ingests but no stored row +(audited max > 0, row absent) is treated as stored `0` and flagged — an absent +`last_mirror_v2_log_id` for an audited *active* mirror ledger is caught, not +skipped. The absent-row check is suppressed only for a ledger audited as +**deleted** (a `DeleteLedger` log replayed in the verified range): `WriteSet.Absorb` +removes the boundary row on deletion, so its missing row is legitimate, not +corruption. Present-row equality still applies to every ledger. + +There is **no** legacy / no-backfill leniency: existing pre-field clusters are +**unsupported** — we do not ship a compat shim, and a store that predates +`last_mirror_v2_log_id` would fail this equality (and simply re-mirror), which is +acceptable pre-GA. + ## Readstore and Indexes The index **registry** is business-visible and is already verified by the diff --git a/docs/technical/architecture/subsystems/checker/checker.md b/docs/technical/architecture/subsystems/checker/checker.md index 656c3dadfb..ccab04e525 100644 --- a/docs/technical/architecture/subsystems/checker/checker.md +++ b/docs/technical/architecture/subsystems/checker/checker.md @@ -40,11 +40,13 @@ Each pass takes a persisted projection, re-derives the expected value by replayi | 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) | Notes: - **`compareIndexes` covers presence + identity** (`IndexID` match), **not `BuildStatus`** and **not `IndexVersionState`**. `BuildStatus` is informational and per-replica; `IndexVersionState` is by design local and may legitimately differ across nodes mid-rewrite (see [indexer / indexes.md](../indexer/indexes.md)). - **`compareIdempotencyOutcomes`** is the one pass that consumes a side-effect of `verifyAuditHashChain` (the expected-outcome map). The two are coupled by design — re-deriving the idempotency outcome means re-walking the chain anyway. +- **`compareMirrorV2LogID` is a full equality check.** The FSM enforces a contiguous applied prefix (`processMirrorIngest` rejects any `v2LogId` gap with `ErrMirrorV2LogIDGap`), so at rest the stored EN-1550 high-water mark must be exactly `max(audited v2_log_id)`. Any divergence is flagged: `stored > max` claims a source log the audit never recorded; `stored < max` means the projection lost applied ground. A never-mirrored ledger has stored `0` and max `0` → equal → clean. There is no legacy/no-backfill leniency — pre-field clusters are unsupported. The audited max is seeded from a baseline floor (the compact baseline snapshot now includes `Boundary` rows) so archived-only mirror ingests are not undercounted. The comparison is driven from the **union** of stored boundary rows and audited-mirror ledgers, so a mirror ledger whose `Boundary` row is absent while it still has audited ingests (audited max > 0, no row) is treated as stored `0` and flagged — not silently skipped. The absent-row branch is suppressed only for a ledger audited as **deleted** (a `DeleteLedger` log replayed in the verified range, `deletedInReplay`): `WriteSet.Absorb` legitimately removes the boundary row on deletion, so its missing row is expected. Present-row equality still applies to every ledger (including one in the pending-cleanup window whose row is still present). - **Order matters**: `verifyAuditHashChain` runs **first**. A broken chain stops the walk before any downstream pass — running them with a tampered chain would produce noise from already-detected corruption. ## Replay machinery @@ -73,7 +75,8 @@ flowchart TB B --> H[verifySealingHash] B --> I[compareExclusionProjections] B --> J[compareIndexes] - D & E & F & G & H & I & J --> K[stream errors as they happen] + B --> L[compareMirrorV2LogID] + D & E & F & G & H & I & J & L --> K[stream errors as they happen] C --> K ``` diff --git a/internal/application/check/checker.go b/internal/application/check/checker.go index 78cc749c78..a4ca6266f4 100644 --- a/internal/application/check/checker.go +++ b/internal/application/check/checker.go @@ -631,6 +631,8 @@ func (c *Checker) Check(ctx context.Context, callback func(*servicepb.CheckStore c.compareIndexes(snap, expectedIndexes, indexReplayActivity, deletedInReplay, hasArchivedChapters, pendingCleanupLedgers, callback) + c.compareMirrorV2LogID(snap, chainBound, deletedInReplay, callback) + return nil } @@ -897,6 +899,130 @@ func (c *Checker) compareIndexes( } } +// compareMirrorV2LogID verifies each ledger's stored +// LedgerBoundaries.last_mirror_v2_log_id EQUALS the maximum audited +// MirrorIngest.v2_log_id for that ledger (chainBound.maxMirrorV2LogID, seeded +// from the baseline floor and layered with the live audit chain). This is a full +// invariant-#8 equality check, not a one-sided bound. +// +// Because the FSM enforces a contiguous applied prefix (processMirrorIngest +// rejects any gap), at rest the persisted high-water mark must be exactly the max +// audited v2_log_id — no more, no less. Any divergence is corruption and emits +// CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH: +// - stored > max: claims a source v2 log no audit entry recorded (future +// ingests wrongly skipped). +// - stored < max: the projection lost applied ground (an already-applied ingest +// would be re-applied, doubling balances). +// +// A non-mirror ledger has stored == 0 and audited max == 0 → equal → not flagged. +// A mirror ledger corrupted to 0 has audited max > 0 → not equal → flagged. There +// is no legacy/no-backfill leniency: existing pre-field clusters are unsupported +// (no compat shim). +// +// The comparison is driven from the UNION of (a) ledgers with a stored boundary +// row and (b) ledgers with audited MirrorIngest orders (maxMirrorV2LogID > 0). +// Iterating only stored rows would silently miss a mirror ledger whose Boundary +// row was deleted/lost (audited max > 0 but no row): treated as stored 0, it is +// flagged (0 != max) instead of skipped — otherwise the disappearance of +// last_mirror_v2_log_id would go undetected and the idempotency guard would treat +// already-applied source logs as fresh. +// +// deletedInReplay names ledgers whose DeleteLedger log was replayed in the +// verified range. WriteSet.Absorb legitimately removes the Boundary row on +// ledger deletion, so the absent-row branch is suppressed for a deleted ledger: +// its missing boundary is expected, not corruption. The present-row equality +// checks (ahead/behind/corrupt-to-zero) still apply to every ledger, including +// one in the pending-cleanup window whose row is still present. +func (c *Checker) compareMirrorV2LogID(reader dal.PebbleReader, chainBound *chainBoundState, deletedInReplay map[string]struct{}, callback func(*servicepb.CheckStoreEvent)) { + // Collect stored last_mirror_v2_log_id per ledger from the live boundary rows. + stored := make(map[string]uint64) + + iter, err := c.attrs.Boundary.NewStreamingIter(reader, nil) + if err != nil { + callback(errorEvent(servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH, + fmt.Sprintf("failed to create live boundaries iterator: %v", err), 0, "", "", "")) + + return + } + + for iter.Next() { + entry := iter.Entry() + + var lk domain.LedgerKey + if err := lk.Unmarshal(entry.CanonicalKey); err != nil { + continue + } + + if lk.Name == "" || entry.Value == nil { + continue + } + + stored[lk.Name] = entry.Value.GetLastMirrorV2LogId() + } + + if err := iter.Close(); err != nil { + callback(errorEvent(servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH, + fmt.Sprintf("closing live boundaries iterator: %v", err), 0, "", "", "")) + + return + } + + if err := iter.Err(); err != nil { + callback(errorEvent(servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH, + fmt.Sprintf("live boundaries iterator error: %v", err), 0, "", "", "")) + + return + } + + // Compare over the union of stored rows and audited-mirror ledgers, in + // sorted order so events are emitted deterministically. A ledger present in + // only one set defaults to 0 on the other side (absent row → stored 0; + // non-mirror ledger → audited max 0). + ledgers := make(map[string]struct{}, len(stored)+len(chainBound.maxMirrorV2LogID)) + for name := range stored { + ledgers[name] = struct{}{} + } + + for name := range chainBound.maxMirrorV2LogID { + ledgers[name] = struct{}{} + } + + names := make([]string, 0, len(ledgers)) + for name := range ledgers { + names = append(names, name) + } + + slices.Sort(names) + + for _, name := range names { + storedV2 := stored[name] + auditedMax := chainBound.maxMirrorV2LogID[name] + + if storedV2 != auditedMax { + _, hasRow := stored[name] + + // Absent-row case for a ledger audited as DELETED: the Boundary row + // is legitimately gone (WriteSet.Absorb removes it on deletion), so + // this is not corruption — skip. Present-row divergences still flag. + if !hasRow { + if _, deleted := deletedInReplay[name]; deleted { + continue + } + } + + detail := "the idempotent-replay high-water mark diverges from the audit chain" + if !hasRow { + detail = "the ledger has audited MirrorIngest orders but no stored boundary row (last_mirror_v2_log_id lost)" + } + + callback(errorEvent(servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH, + fmt.Sprintf("stored last_mirror_v2_log_id %d does not equal max audited MirrorIngest v2_log_id %d for ledger %q: %s", + storedV2, auditedMax, name, detail), + 0, name, "", "")) + } + } +} + // excludedVolumesSet maps ledger name to a set of (account, asset) tuples // that legitimately diverge between the replay store and the live Pebble // store. The set is populated incrementally by the replay-time @@ -1879,6 +2005,17 @@ type chainBoundState struct { // fire for it: a missing claim/mutation on a live-created ledger is a // forged skip, not an archive limitation (finding 2adbf685cc). ledgerCreationSeenLive map[string]struct{} + // maxMirrorV2LogID is the audit-derived EXPECTED value of + // LedgerBoundaries.last_mirror_v2_log_id per ledger: the highest audited + // MirrorIngest.v2_log_id, from the live audit chain + // (recordMirrorIngestMutations) plus a baseline floor (foldBaselineBoundaries + // seeds it from the archived LedgerBoundaries.last_mirror_v2_log_id, so + // ledgers whose mirror ingests live in an archived chapter are not + // undercounted). compareMirrorV2LogID checks the stored last_mirror_v2_log_id + // for EQUALITY against this value and flags ANY divergence (both stored > and + // stored <): the FSM enforces a contiguous applied prefix, so at rest the two + // must be exactly equal. There is no at-or-below exemption (EN-1550). + maxMirrorV2LogID map[string]uint64 } // chainBoundMutation records one presence-flip observed on the audit @@ -1903,6 +2040,7 @@ func newChainBoundState() *chainBoundState { nextTxID: make(map[string]uint64), ledgerCreationSeen: make(map[string]struct{}), ledgerCreationSeenLive: make(map[string]struct{}), + maxMirrorV2LogID: make(map[string]uint64), } } @@ -2588,6 +2726,13 @@ func recordMirrorIngestMutations( return } + // Track the highest audited source v2_log_id per ledger (any ingest kind + // carries it on the wrapping MirrorLogEntry). compareMirrorV2LogID bounds + // the stored LedgerBoundaries.last_mirror_v2_log_id against this max. + if v2 := entry.GetV2LogId(); v2 > chainBound.maxMirrorV2LogID[ledger] { + chainBound.maxMirrorV2LogID[ledger] = v2 + } + if mct := entry.GetCreatedTransaction(); mct != nil { if ref := mct.GetReference(); ref != "" { rememberReferenceClaim(chainBound.references, ledger, ref, logSeq) @@ -3102,6 +3247,17 @@ func (c *Checker) foldBaselineBoundaries( continue } + // Baseline floor for the mirror v2LogId high-water mark: at the archive + // boundary, LedgerBoundaries.last_mirror_v2_log_id equals the max + // MirrorIngest.v2_log_id applied up to that point, so it is a valid lower + // bound for the archived audited max. Without this, a ledger whose mirror + // ingests all live in an archived chapter would have live-only max 0 and + // compareMirrorV2LogID would false-positive on its legitimate stored + // value. Fold as a max so live ingests after the boundary still win. + if v2 := entry.Value.GetLastMirrorV2LogId(); v2 > chainBound.maxMirrorV2LogID[lk.Name] { + chainBound.maxMirrorV2LogID[lk.Name] = v2 + } + next := entry.Value.GetNextTransactionId() if next == 0 { continue diff --git a/internal/application/check/checker_mirror_v2logid_test.go b/internal/application/check/checker_mirror_v2logid_test.go new file mode 100644 index 0000000000..bcd757ad43 --- /dev/null +++ b/internal/application/check/checker_mirror_v2logid_test.go @@ -0,0 +1,313 @@ +package check + +import ( + "maps" + "path/filepath" + "testing" + + "github.com/cockroachdb/pebble/v2" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/metric/noop" + + logging "github.com/formancehq/go-libs/v5/pkg/observe/log" + + "github.com/formancehq/ledger/v3/internal/domain" + "github.com/formancehq/ledger/v3/internal/infra/attributes" + "github.com/formancehq/ledger/v3/internal/proto/raftcmdpb" + "github.com/formancehq/ledger/v3/internal/proto/servicepb" + "github.com/formancehq/ledger/v3/internal/storage/dal" +) + +// writeBoundaries persists a LedgerBoundaries row for the given ledger through +// the attribute writer, mirroring how the FSM stores it. +func writeBoundaries(t *testing.T, store *dal.Store, attrs *attributes.Attributes, ledger string, b *raftcmdpb.LedgerBoundaries) { + t.Helper() + + batch := store.OpenWriteSession() + _, err := attrs.Boundary.Set(batch, domain.LedgerKey{Name: ledger}.Bytes(), b) + require.NoError(t, err) + require.NoError(t, batch.Commit()) +} + +// collectMirrorV2LogIDEvents runs compareMirrorV2LogID against the store's live +// boundaries with the given audited-max map and returns only the +// MIRROR_V2LOGID_MISMATCH errors. deletedLedgers names ledgers audited as +// deleted (their absent boundary row is legitimate). +func collectMirrorV2LogIDEvents(t *testing.T, store *dal.Store, attrs *attributes.Attributes, maxV2 map[string]uint64, deletedLedgers ...string) []*servicepb.CheckStoreError { + t.Helper() + + checker := NewChecker(store, attrs, "mirror-v2logid-cluster", nil, nil, logging.Testing()) + + handle, err := store.NewReadHandle() + require.NoError(t, err) + + defer func() { _ = handle.Close() }() + + chainBound := newChainBoundState() + maps.Copy(chainBound.maxMirrorV2LogID, maxV2) + + deletedInReplay := make(map[string]struct{}, len(deletedLedgers)) + for _, name := range deletedLedgers { + deletedInReplay[name] = struct{}{} + } + + var got []*servicepb.CheckStoreError + + checker.compareMirrorV2LogID(handle, chainBound, deletedInReplay, func(event *servicepb.CheckStoreEvent) { + if e, ok := event.GetType().(*servicepb.CheckStoreEvent_Error); ok && + e.Error.GetErrorType() == servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH { + got = append(got, e.Error) + } + }) + + return got +} + +// TestCompareMirrorV2LogID_AheadFlagged: stored high-water mark strictly above +// the audited max is corruption (claims a v2 log the audit never recorded) and +// must emit exactly one MIRROR_V2LOGID_MISMATCH event naming the ledger. +func TestCompareMirrorV2LogID_AheadFlagged(t *testing.T) { + t.Parallel() + + store := createTestStore(t) + attrs := attributes.New() + + writeBoundaries(t, store, attrs, "mirror-ledger", &raftcmdpb.LedgerBoundaries{ + NextTransactionId: 5, + NextLogId: 5, + LastMirrorV2LogId: 10, // stored claims v2LogId 10 applied + }) + + // Audit only recorded up to v2LogId 7. + got := collectMirrorV2LogIDEvents(t, store, attrs, map[string]uint64{"mirror-ledger": 7}) + + require.Len(t, got, 1) + require.Equal(t, "mirror-ledger", got[0].GetLedger()) + require.Equal(t, servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH, got[0].GetErrorType()) +} + +// TestCompareMirrorV2LogID_BehindFlagged: stored high-water mark BELOW the +// audited max means the persisted projection lost applied ground — under the +// equality check (no backfill leniency) this is now FLAGGED. +func TestCompareMirrorV2LogID_BehindFlagged(t *testing.T) { + t.Parallel() + + store := createTestStore(t) + attrs := attributes.New() + + writeBoundaries(t, store, attrs, "behind-ledger", &raftcmdpb.LedgerBoundaries{ + NextTransactionId: 1, NextLogId: 1, LastMirrorV2LogId: 3, + }) + + got := collectMirrorV2LogIDEvents(t, store, attrs, map[string]uint64{"behind-ledger": 7}) + + require.Len(t, got, 1) + require.Equal(t, "behind-ledger", got[0].GetLedger()) + require.Equal(t, servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH, got[0].GetErrorType()) +} + +// TestCompareMirrorV2LogID_EqualNotFlagged: stored == audited max is the correct +// at-rest state for a mirror ledger and is not flagged. +func TestCompareMirrorV2LogID_EqualNotFlagged(t *testing.T) { + t.Parallel() + + store := createTestStore(t) + attrs := attributes.New() + + writeBoundaries(t, store, attrs, "at-max", &raftcmdpb.LedgerBoundaries{ + NextTransactionId: 1, NextLogId: 1, LastMirrorV2LogId: 7, + }) + + got := collectMirrorV2LogIDEvents(t, store, attrs, map[string]uint64{"at-max": 7}) + + require.Empty(t, got) +} + +// TestCompareMirrorV2LogID_NonMirrorLedgerNotFlagged: a regular (never-mirrored) +// ledger has stored 0 and no audited MirrorIngest (max 0) → equal → not flagged. +func TestCompareMirrorV2LogID_NonMirrorLedgerNotFlagged(t *testing.T) { + t.Parallel() + + store := createTestStore(t) + attrs := attributes.New() + + writeBoundaries(t, store, attrs, "regular-ledger", &raftcmdpb.LedgerBoundaries{ + NextTransactionId: 42, NextLogId: 42, LastMirrorV2LogId: 0, + }) + + // No audited MirrorIngest for this ledger (max == 0). + got := collectMirrorV2LogIDEvents(t, store, attrs, map[string]uint64{}) + + require.Empty(t, got) +} + +// TestCompareMirrorV2LogID_CorruptToZeroFlagged: a mirror ledger with audited +// ingests whose stored high-water mark was wiped to 0 must be flagged (0 != max). +func TestCompareMirrorV2LogID_CorruptToZeroFlagged(t *testing.T) { + t.Parallel() + + store := createTestStore(t) + attrs := attributes.New() + + writeBoundaries(t, store, attrs, "wiped-ledger", &raftcmdpb.LedgerBoundaries{ + NextTransactionId: 5, NextLogId: 5, LastMirrorV2LogId: 0, + }) + + got := collectMirrorV2LogIDEvents(t, store, attrs, map[string]uint64{"wiped-ledger": 4}) + + require.Len(t, got, 1) + require.Equal(t, "wiped-ledger", got[0].GetLedger()) +} + +// TestBaselineBoundaries_SeedArchivedMirrorV2LogID pins Finding 3: the compact +// baseline snapshot (CreateBaselineSnapshot) now includes Boundary rows, so +// foldBaselineBoundaries can seed the archived floor for a ledger whose mirror +// ingests live entirely in an archived chapter — and compareMirrorV2LogID then +// sees the correct audited max and does NOT false-positive on the live stored +// value. Before the fix, writeBaselineAttributes omitted Boundary rows, the +// baseline floor was 0, and a healthy archived-mirror ledger was flagged. +func TestBaselineBoundaries_SeedArchivedMirrorV2LogID(t *testing.T) { + t.Parallel() + + logger := logging.Testing() + meter := noop.NewMeterProvider().Meter("test") + attrs := attributes.New() + + // Source store carrying a mirror ledger's boundaries with an applied + // high-water mark of 9 (as if all its mirror ingests were archived). + src, err := dal.NewStore(t.TempDir(), logger, meter, dal.DefaultConfig()) + require.NoError(t, err) + t.Cleanup(func() { _ = src.Close() }) + + batch := src.OpenWriteSession() + _, err = attrs.Boundary.Set(batch, domain.LedgerKey{Name: "archived-mirror"}.Bytes(), &raftcmdpb.LedgerBoundaries{ + NextTransactionId: 12, + NextLogId: 12, + LastMirrorV2LogId: 9, + }) + require.NoError(t, err) + require.NoError(t, batch.Commit()) + + // Build the compact baseline snapshot the way archival does. + handle, err := src.NewReadHandle() + require.NoError(t, err) + + baselinePath := filepath.Join(t.TempDir(), "baseline") + require.NoError(t, attributes.CreateBaselineSnapshot(handle, attrs, baselinePath)) + require.NoError(t, handle.Close()) + + baselineDB, err := pebble.Open(baselinePath, &pebble.Options{ReadOnly: true}) + require.NoError(t, err) + t.Cleanup(func() { _ = baselineDB.Close() }) + + // foldBaselineBoundaries must read the archived last_mirror_v2_log_id (the + // Finding 3 fix) — and still seed nextTxID as before (no regression). + checker := NewChecker(nil, attrs, "test-cluster", nil, nil, logging.Testing()) + chainBound := newChainBoundState() + require.NoError(t, checker.foldBaselineBoundaries(baselineDB, chainBound)) + + require.Equal(t, uint64(9), chainBound.maxMirrorV2LogID["archived-mirror"], + "baseline Boundary row must seed the archived mirror v2LogId floor") + require.Equal(t, uint64(12), chainBound.nextTxID["archived-mirror"], + "baseline NextTransactionId seeding must still work (no regression)") + + // End-to-end: with the archived floor seeded, a live ledger whose stored + // high-water mark equals the archived max is NOT flagged. + live := createTestStore(t) + writeBoundaries(t, live, attrs, "archived-mirror", &raftcmdpb.LedgerBoundaries{ + NextTransactionId: 12, NextLogId: 12, LastMirrorV2LogId: 9, + }) + + liveHandle, err := live.NewReadHandle() + require.NoError(t, err) + + defer func() { _ = liveHandle.Close() }() + + var got []*servicepb.CheckStoreError + checker.compareMirrorV2LogID(liveHandle, chainBound, nil, func(event *servicepb.CheckStoreEvent) { + if e, ok := event.GetType().(*servicepb.CheckStoreEvent_Error); ok { + got = append(got, e.Error) + } + }) + + require.Empty(t, got, "archived-mirror ledger with stored == archived max must not be flagged") +} + +// TestCompareMirrorV2LogID_AbsentRowFlagged pins the union-driven comparison +// (NumaryBot checker.go:930): a mirror ledger that has audited MirrorIngest +// orders (auditedMax > 0) but NO stored boundary row must be flagged — treated as +// stored 0 (0 != max). Iterating only existing rows would silently skip it, +// leaving the disappearance of last_mirror_v2_log_id undetected. +func TestCompareMirrorV2LogID_AbsentRowFlagged(t *testing.T) { + t.Parallel() + + store := createTestStore(t) + attrs := attributes.New() + + // No boundary row written for "gone-mirror", but the audit shows ingests up + // to v2LogId 5. + got := collectMirrorV2LogIDEvents(t, store, attrs, map[string]uint64{"gone-mirror": 5}) + + require.Len(t, got, 1) + require.Equal(t, "gone-mirror", got[0].GetLedger()) + require.Equal(t, servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH, got[0].GetErrorType()) +} + +// TestCompareMirrorV2LogID_DeletedMirrorLedgerNotFlagged pins that a LEGITIMATELY +// DELETED mirror ledger — audited ingests (max > 0), a DeleteLedger log in the +// verified range, and its boundary row removed by WriteSet.Absorb — is NOT +// flagged. Without the deleted-ledger exclusion the absent-row branch would +// false-positive on a healthy deletion (NumaryBot checker.go:979). +func TestCompareMirrorV2LogID_DeletedMirrorLedgerNotFlagged(t *testing.T) { + t.Parallel() + + store := createTestStore(t) + attrs := attributes.New() + + // Audited ingests up to v2LogId 5, no boundary row (removed on delete), and + // the ledger is in the audited deleted set. + got := collectMirrorV2LogIDEvents(t, store, attrs, map[string]uint64{"deleted-mirror": 5}, "deleted-mirror") + + require.Empty(t, got) +} + +// TestCompareMirrorV2LogID_AbsentRowNonMirrorNotFlagged: a ledger with neither a +// boundary row nor audited ingests (both 0) is not in scope and not flagged — +// the union defaults both sides to 0. +func TestCompareMirrorV2LogID_AbsentRowNonMirrorNotFlagged(t *testing.T) { + t.Parallel() + + store := createTestStore(t) + attrs := attributes.New() + + // A present, healthy mirror row alongside no entry at all for any absent + // non-mirror ledger — nothing spurious is emitted. + writeBoundaries(t, store, attrs, "healthy", &raftcmdpb.LedgerBoundaries{ + NextTransactionId: 1, NextLogId: 1, LastMirrorV2LogId: 4, + }) + + got := collectMirrorV2LogIDEvents(t, store, attrs, map[string]uint64{"healthy": 4}) + + require.Empty(t, got) +} + +// TestCompareMirrorV2LogID_MultiLedgerIsolation: only the diverging ledger is +// flagged; a healthy (equal) sibling in the same store is untouched. +func TestCompareMirrorV2LogID_MultiLedgerIsolation(t *testing.T) { + t.Parallel() + + store := createTestStore(t) + attrs := attributes.New() + + writeBoundaries(t, store, attrs, "healthy", &raftcmdpb.LedgerBoundaries{ + NextTransactionId: 1, NextLogId: 1, LastMirrorV2LogId: 4, + }) + writeBoundaries(t, store, attrs, "corrupt", &raftcmdpb.LedgerBoundaries{ + NextTransactionId: 1, NextLogId: 1, LastMirrorV2LogId: 99, + }) + + got := collectMirrorV2LogIDEvents(t, store, attrs, map[string]uint64{"healthy": 4, "corrupt": 5}) + + require.Len(t, got, 1) + require.Equal(t, "corrupt", got[0].GetLedger()) +} diff --git a/internal/domain/errors.go b/internal/domain/errors.go index eef67b6717..3fe72bb3da 100644 --- a/internal/domain/errors.go +++ b/internal/domain/errors.go @@ -221,6 +221,8 @@ const ( ErrReasonNumscriptRuntime = "NUMSCRIPT_RUNTIME" ErrReasonVolumeNotMaterialized = "VOLUME_NOT_MATERIALIZED" ErrReasonNonDeterministicScript = "NON_DETERMINISTIC_SCRIPT" + ErrReasonMirrorV2LogIDGap = "MIRROR_V2_LOG_ID_GAP" + ErrReasonMirrorV2LogIDInvalid = "MIRROR_V2_LOG_ID_INVALID" // ErrReasonWritesBlockedDiskFull signals that the write gate rejected the // request because disk usage is at or above the configured block threshold. @@ -756,6 +758,51 @@ func (e *ErrLedgerNotInMirrorMode) Metadata() map[string]string { return map[string]string{"name": e.Name} } +// ErrMirrorV2LogIDGap — a mirror ingest arrived with a v2LogId beyond the next +// contiguous slot (Expected) after the applied prefix (LastMirrorV2LogId is +// Expected-1). The worker ingests contiguously (including FillGap orders), so a +// gap means the persisted high-water mark or the source cursor is ahead of the +// applied prefix — corruption/tampering, impossible in normal operation. The FSM +// rejects the order without mutation rather than silently applying past the gap +// (which would desync nodes) or skipping it. KindInternal: server-side +// data-corruption invariant, not a client mistake (invariant #7 fail-loud on an +// impossible-by-design branch). +type ErrMirrorV2LogIDGap struct { + Name string + Got uint64 + Expected uint64 +} + +func (e *ErrMirrorV2LogIDGap) Error() string { + return fmt.Sprintf("invariant: mirror v2LogId gap on ledger %s: got %d, expected %d", e.Name, e.Got, e.Expected) +} +func (*ErrMirrorV2LogIDGap) Reason() string { return ErrReasonMirrorV2LogIDGap } +func (e *ErrMirrorV2LogIDGap) Metadata() map[string]string { + return map[string]string{ + "name": e.Name, + "got": strconv.FormatUint(e.Got, 10), + "expected": strconv.FormatUint(e.Expected, 10), + } +} + +// ErrMirrorV2LogIDInvalid — a mirror ingest carried a v2LogId of 0. Source v2 +// log ids are 1-based, so 0 is malformed/tampered. The FSM rejects it before any +// mutation rather than applying it: a 0 is never recorded as the high-water mark, +// so applying it would leave it re-appliable forever (no marker stops the +// replay). KindInternal: server-side data-corruption invariant, not a client +// mistake (invariant #7 fail-loud on an impossible-by-design branch). +type ErrMirrorV2LogIDInvalid struct { + Name string +} + +func (e *ErrMirrorV2LogIDInvalid) Error() string { + return fmt.Sprintf("invariant: mirror ingest on ledger %s carries invalid v2LogId 0 (source v2 log ids are 1-based)", e.Name) +} +func (*ErrMirrorV2LogIDInvalid) Reason() string { return ErrReasonMirrorV2LogIDInvalid } +func (e *ErrMirrorV2LogIDInvalid) Metadata() map[string]string { + return map[string]string{"name": e.Name} +} + // ErrPreparedQueryAlreadyExists — creating a prepared query that already exists. type ErrPreparedQueryAlreadyExists struct { Ledger string diff --git a/internal/domain/errors_test.go b/internal/domain/errors_test.go index cd07327ed1..82aabc7634 100644 --- a/internal/domain/errors_test.go +++ b/internal/domain/errors_test.go @@ -311,6 +311,8 @@ func TestEveryDomainErrorImplementsDescribable(t *testing.T) { "ErrInvalidCronExpression": &ErrInvalidCronExpression{}, "ErrLedgerInMirrorMode": &ErrLedgerInMirrorMode{}, "ErrLedgerNotInMirrorMode": &ErrLedgerNotInMirrorMode{}, + "ErrMirrorV2LogIDGap": &ErrMirrorV2LogIDGap{}, + "ErrMirrorV2LogIDInvalid": &ErrMirrorV2LogIDInvalid{}, "ErrPreparedQueryAlreadyExists": &ErrPreparedQueryAlreadyExists{}, "ErrPreparedQueryNotFound": &ErrPreparedQueryNotFound{}, "ErrIndexNotFound": &ErrIndexNotFound{}, diff --git a/internal/domain/processing/processor.go b/internal/domain/processing/processor.go index dced754b99..44e67fc5b7 100644 --- a/internal/domain/processing/processor.go +++ b/internal/domain/processing/processor.go @@ -253,6 +253,24 @@ func (p *RequestProcessor) ProcessOrders(orders []*raftcmdpb.Order, scopeFactory return nil, err } + // No-log outcome: a processor may deterministically decide the order + // produces no fresh ledger log while still succeeding (payload==nil, + // err==nil). The only such case today is an idempotent mirror replay + // (processMirrorIngest guards on LastMirrorV2LogId), which mutates + // nothing and must leave no audit-visible log. Skip the log slot + // entirely — do NOT consume a sequence id, absorb into the sink, or + // append a degenerate Log{Payload:nil}. The nil slot mirrors the + // idempotency-replay ReferenceSequence path and is skipped identically + // by WriteSet.Merge (GetCreatedLog()==nil) and checkCloseChapter. + // + // Any overlay staged by such an order is dropped without Commit(): a + // no-log outcome makes no persistent mutation, so there is nothing to + // flush. (Mirror ingest declares no skippable_reasons, so no overlay is + // created here in practice.) + if payload == nil { + continue + } + if overlay != nil { if err := overlay.Commit(); err != nil { // Coverage-miss surfaced by a staged Delete (invariant #6): diff --git a/internal/domain/processing/processor_mirror.go b/internal/domain/processing/processor_mirror.go index 125044a41f..2f817548c6 100644 --- a/internal/domain/processing/processor_mirror.go +++ b/internal/domain/processing/processor_mirror.go @@ -27,10 +27,6 @@ func processMirrorIngest(ledger string, order *raftcmdpb.MirrorIngestOrder, ctx if info.GetMode() != commonpb.LedgerMode_LEDGER_MODE_MIRROR { return nil, &domain.ErrLedgerNotInMirrorMode{Name: ledger} } - // Re-touch ledger info so it enters the Merge buffer and gets propagated - // back to Gen0 on commit. Without this, ledger info is evicted after two - // cache rotations because mirror proposals bypass the admission preloader. - s.Ledgers().Put(domain.LedgerKey{Name: ledger}, info) boundariesReader, loadErr := loadBoundaries(s, ledger) if loadErr != nil { @@ -39,15 +35,68 @@ func processMirrorIngest(ledger string, order *raftcmdpb.MirrorIngestOrder, ctx boundaries := boundariesReader.Mutate() - // Stage per-apply context fields for child handlers. - ctx.Boundaries = boundaries - ctx.LedgerInfo = info - entry := order.GetEntry() if entry == nil { return nil, &domain.ErrLedgerNotInMirrorMode{Name: ledger} } + // Contiguous-applied-prefix guard — evaluated BEFORE any mutation (no ledger + // re-touch, no cache write) so a rejected/replayed ingest leaves no side + // effect. LastMirrorV2LogId is the highest source v2LogId already applied to + // this ledger and, because the worker ingests contiguously (including FillGap + // orders for source gaps — see adapter/v2/translator.go: TranslateBatch), it + // is a TRUE contiguous prefix: every id in [1, LastMirrorV2LogId] has been + // applied. v2 log ids are 1-based and strictly increasing per source. + v2LogID := entry.GetV2LogId() + last := boundaries.GetLastMirrorV2LogId() + + // v2LogID == 0 is malformed: source v2 log ids are 1-based, so 0 never + // occurs in a well-formed ingest. It must be rejected loud BEFORE the + // contiguous-prefix switch: it is <= last and == expected only vacuously, and + // falling through would apply it WITHOUT advancing the high-water mark (0 is + // never recorded as last), leaving it re-appliable forever (flemzord, #1587). + // Per invariant #7 this is an impossible-by-design branch → deterministic + // KindInternal reject, no mutation. + if v2LogID == 0 { + return nil, &domain.ErrMirrorV2LogIDInvalid{Name: ledger} + } + + // Three cases against the next contiguous slot (expected = last + 1): + // + // - v2LogID <= last → already applied. EXPECTED replay (a tampered/rolled- + // back MirrorCursor makes the worker re-emit applied logs — flemzord, + // #1581). Idempotent no-op: return (nil, nil), which ProcessOrders treats + // as a no-log outcome (no sequence id, no audit log, no sink absorb). Soft + // skip is correct per invariant #7's expected-vs-impossible distinction. + // + // - v2LogID == expected → the next contiguous log. Apply and advance below. + // + // - v2LogID > expected → a GAP: the cursor/high-water mark is ahead of the + // applied prefix. Impossible in normal operation (contiguous ingest), so + // it is corruption/tampering. Per invariant #7, fail LOUD on an impossible- + // by-design branch rather than silently applying past the gap (which would + // desync nodes) or skipping it: reject the order with a deterministic + // KindInternal error and mutate nothing. Same input → same rejection on + // every node, so determinism holds. + // + // First-ingest case: LastMirrorV2LogId defaults to 0, so expected = 1 and the + // first real v2LogId (>= 1) applies. + switch { + case v2LogID <= last: + return nil, nil + case v2LogID > last+1: + return nil, &domain.ErrMirrorV2LogIDGap{Name: ledger, Got: v2LogID, Expected: last + 1} + } + + // Re-touch ledger info so it enters the Merge buffer and gets propagated + // back to Gen0 on commit. Without this, ledger info is evicted after two + // cache rotations because mirror proposals bypass the admission preloader. + s.Ledgers().Put(domain.LedgerKey{Name: ledger}, info) + + // Stage per-apply context fields for child handlers. + ctx.Boundaries = boundaries + ctx.LedgerInfo = info + var logPayload *commonpb.LedgerLogPayload switch data := entry.GetData().(type) { @@ -93,6 +142,14 @@ func processMirrorIngest(ledger string, order *raftcmdpb.MirrorIngestOrder, ctx // Assign per-ledger log ID and advance boundaries nextLogID := boundaries.GetNextLogId() boundaries.NextLogId = nextLogID + 1 + // Advance the idempotent-replay high-water mark. Set once here regardless of + // the inner ingest kind, because v2LogId lives on the wrapping MirrorLogEntry + // and applies to every kind (CreatedTransaction, SavedMetadata, + // DeletedMetadata, RevertedTransaction, FillGap). Guarded by v2LogID != 0 so + // an entry that somehow carries no source id never rewinds the mark. + if v2LogID != 0 { + boundaries.LastMirrorV2LogId = v2LogID + } s.Boundaries().Put(domain.LedgerKey{Name: ledger}, boundaries) return &commonpb.LogPayload{ diff --git a/internal/domain/processing/processor_mirror_test.go b/internal/domain/processing/processor_mirror_test.go index 3aaaafef75..395ae4f0e2 100644 --- a/internal/domain/processing/processor_mirror_test.go +++ b/internal/domain/processing/processor_mirror_test.go @@ -22,7 +22,8 @@ func TestMirrorIngest_FillGap(t *testing.T) { require.NoError(t, err) now := &commonpb.Timestamp{Data: 1234567890} - boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 1, NextLogId: 1} + // Contiguous prefix: v2LogId 5 requires the applied prefix to be at 4. + boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 1, NextLogId: 1, LastMirrorV2LogId: 4} ledgerInfo := &commonpb.LedgerInfo{ Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR, @@ -168,6 +169,261 @@ func TestMirrorIngest_CreatedTransaction(t *testing.T) { require.Equal(t, uint64(43), putBoundaries.GetNextTransactionId()) } +// mirrorCreatedTxOrder builds a MirrorIngest order for a created transaction +// carrying the given v2LogId. Shared by the idempotency tests so they exercise +// the exact wrapper shape (v2LogId on MirrorLogEntry) the guard reads. +func mirrorCreatedTxOrder(ledger string, v2LogID, txID uint64) *raftcmdpb.Order { + return &raftcmdpb.Order{ + Type: &raftcmdpb.Order_LedgerScoped{ + LedgerScoped: &raftcmdpb.LedgerScopedOrder{ + Ledger: ledger, + Payload: &raftcmdpb.LedgerScopedOrder_MirrorIngest{ + MirrorIngest: &raftcmdpb.MirrorIngestOrder{Entry: &raftcmdpb.MirrorLogEntry{ + V2LogId: v2LogID, + Data: &raftcmdpb.MirrorLogEntry_CreatedTransaction{ + CreatedTransaction: &raftcmdpb.MirrorCreatedTransaction{ + TransactionId: txID, + Postings: []*commonpb.Posting{{ + Source: "world", + Destination: "users:001", + Amount: commonpb.NewUint256FromUint64(500), + Asset: "USD/2", + }}, + }, + }, + }}, + }, + }, + }, + } +} + +// TestMirrorIngest_ReplayIsNoOp pins the EN-1550 idempotency guard: an ingest +// whose v2LogId equals the recorded high-water mark (LastMirrorV2LogId) has +// already been applied, so re-applying it is a deterministic side-effect-free +// no-op — no postings applied (no volume writes → balances not doubled), no +// ledger re-touch, no boundary advance, and (nil, nil) returned so ProcessOrders +// emits no log. +func TestMirrorIngest_ReplayIsNoOp(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockStore := NewMockScope(ctrl) + processor, err := NewRequestProcessor(nil, 0) + require.NoError(t, err) + + // v2LogId 7 was already applied: LastMirrorV2LogId == 7. + boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 43, NextLogId: 5, LastMirrorV2LogId: 7} + ledgerInfo := &commonpb.LedgerInfo{Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR} + + expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes() + + boundariesStub := setupBoundariesStub(mockStore) + boundariesStub.expectGet(domain.LedgerKey{Name: "mirror-ledger"}, boundaries.AsReader(), nil) + // The guard runs before any mutation: no boundary Put on replay. + boundariesStub.onPut(func(_ domain.LedgerKey, _ *raftcmdpb.LedgerBoundaries) { + t.Errorf("Boundaries().Put must not be called on an idempotent replay") + }) + // No ledger re-touch on replay (guarded before s.Ledgers().Put). + ledgersStub := setupLedgersStub(mockStore) + ledgersStub.onPut(func(_ domain.LedgerKey, _ *commonpb.LedgerInfo) { + t.Errorf("Ledgers().Put must not be called on an idempotent replay") + }) + // No posting applied → no volume writes → balances cannot double. + volumesStub := setupVolumesStub(mockStore) + volumesStub.onPut(func(_ domain.VolumeKey, _ *raftcmdpb.VolumePair) { + t.Errorf("Volumes().Put must not be called on an idempotent replay") + }) + + // Replay v2LogId 7 (already applied). + result, err := processor.ProcessOrder(mirrorCreatedTxOrder("mirror-ledger", 7, 42), mockStore) + require.NoError(t, err) + require.Nil(t, result, "replayed ingest must produce no log payload") +} + +// TestMirrorIngest_LowerV2LogIdSkipped pins that an OLDER v2LogId arriving after +// a higher one (e.g. a tampered/rolled-back MirrorCursor makes the worker +// re-emit past logs) is skipped as a no-op. +func TestMirrorIngest_LowerV2LogIdSkipped(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockStore := NewMockScope(ctrl) + processor, err := NewRequestProcessor(nil, 0) + require.NoError(t, err) + + boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 43, NextLogId: 5, LastMirrorV2LogId: 10} + ledgerInfo := &commonpb.LedgerInfo{Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR} + + expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes() + + boundariesStub := setupBoundariesStub(mockStore) + boundariesStub.expectGet(domain.LedgerKey{Name: "mirror-ledger"}, boundaries.AsReader(), nil) + boundariesStub.onPut(func(_ domain.LedgerKey, _ *raftcmdpb.LedgerBoundaries) { + t.Errorf("Boundaries().Put must not be called for a stale (lower) v2LogId") + }) + volumesStub := setupVolumesStub(mockStore) + volumesStub.onPut(func(_ domain.VolumeKey, _ *raftcmdpb.VolumePair) { + t.Errorf("Volumes().Put must not be called for a stale (lower) v2LogId") + }) + + // v2LogId 6 is older than the applied high-water mark 10. + result, err := processor.ProcessOrder(mirrorCreatedTxOrder("mirror-ledger", 6, 99), mockStore) + require.NoError(t, err) + require.Nil(t, result) +} + +// TestMirrorIngest_AdvancesLastMirrorV2LogId pins that a forward ingest applies +// and records its v2LogId as the new high-water mark on the boundaries it writes. +func TestMirrorIngest_AdvancesLastMirrorV2LogId(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockStore := NewMockScope(ctrl) + processor, err := NewRequestProcessor(nil, 0) + require.NoError(t, err) + + now := &commonpb.Timestamp{Data: 1234567890} + // Previously applied up to v2LogId 3; the new entry (4) is strictly greater. + boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 1, NextLogId: 1, LastMirrorV2LogId: 3} + ledgerInfo := &commonpb.LedgerInfo{Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR} + + var putBoundaries *raftcmdpb.LedgerBoundaries + + expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes() + expectPutLedger(t, mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo) + mockStore.EXPECT().GetDate().Return(now.AsReader()).AnyTimes() + mockStore.EXPECT().GetNextSequenceID().Return(uint64(100)) + mockStore.EXPECT().GetCurrentOpenChapter().Return(nil, false) + + boundariesStub := setupBoundariesStub(mockStore) + boundariesStub.expectGet(domain.LedgerKey{Name: "mirror-ledger"}, boundaries.AsReader(), nil) + boundariesStub.onPut(func(_ domain.LedgerKey, b *raftcmdpb.LedgerBoundaries) { putBoundaries = b }) + + zeroVol := &raftcmdpb.VolumePair{ + Input: commonpb.NewUint256FromUint64(0), + Output: commonpb.NewUint256FromUint64(0), + } + volumes := setupVolumesStub(mockStore) + volumes.expectGet(domain.NewVolumeKey("mirror-ledger", "world", "USD/2"), zeroVol.AsReader(), nil) + volumes.expectGet(domain.NewVolumeKey("mirror-ledger", "users:001", "USD/2"), zeroVol.AsReader(), nil) + expectPutTransactionState(t, mockStore, domain.TransactionKey{LedgerName: "mirror-ledger", ID: 42}, nil) + + result, err := processor.ProcessOrder(mirrorCreatedTxOrder("mirror-ledger", 4, 42), mockStore) + require.NoError(t, err) + require.NotNil(t, result) + + require.NotNil(t, putBoundaries) + require.Equal(t, uint64(4), putBoundaries.GetLastMirrorV2LogId(), + "applied ingest must advance the v2LogId high-water mark") +} + +// TestMirrorIngest_GapRejected pins the contiguous-prefix invariant: an ingest +// whose v2LogId is beyond the next contiguous slot (last+1) is a gap — impossible +// in normal contiguous ingestion, so the FSM fails LOUD (ErrMirrorV2LogIDGap, +// KindInternal) and mutates nothing rather than silently applying past it. +func TestMirrorIngest_GapRejected(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockStore := NewMockScope(ctrl) + processor, err := NewRequestProcessor(nil, 0) + require.NoError(t, err) + + // Applied prefix at 3, so the next contiguous slot is 4; v2LogId 6 is a gap. + boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 1, NextLogId: 1, LastMirrorV2LogId: 3} + ledgerInfo := &commonpb.LedgerInfo{Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR} + + expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes() + + boundariesStub := setupBoundariesStub(mockStore) + boundariesStub.expectGet(domain.LedgerKey{Name: "mirror-ledger"}, boundaries.AsReader(), nil) + // No mutation on a gap rejection: no boundary Put, no ledger re-touch, no volume Put. + boundariesStub.onPut(func(_ domain.LedgerKey, _ *raftcmdpb.LedgerBoundaries) { + t.Errorf("Boundaries().Put must not be called on a gap rejection") + }) + ledgersStub := setupLedgersStub(mockStore) + ledgersStub.onPut(func(_ domain.LedgerKey, _ *commonpb.LedgerInfo) { + t.Errorf("Ledgers().Put must not be called on a gap rejection") + }) + volumesStub := setupVolumesStub(mockStore) + volumesStub.onPut(func(_ domain.VolumeKey, _ *raftcmdpb.VolumePair) { + t.Errorf("Volumes().Put must not be called on a gap rejection") + }) + + result, err := processor.ProcessOrder(mirrorCreatedTxOrder("mirror-ledger", 6, 42), mockStore) + require.Error(t, err) + require.Nil(t, result) + + var gap *domain.ErrMirrorV2LogIDGap + require.ErrorAs(t, err, &gap) + require.Equal(t, "mirror-ledger", gap.Name) + require.Equal(t, uint64(6), gap.Got) + require.Equal(t, uint64(4), gap.Expected) +} + +// TestMirrorIngest_ZeroV2LogIdRejected pins that a malformed v2LogId == 0 ingest +// is rejected LOUD before any mutation (ErrMirrorV2LogIDInvalid, KindInternal) — +// it must NOT fall through to apply-without-advancing (0 is never recorded as the +// high-water mark, so a silently-applied 0 would be re-appliable forever, +// flemzord #1587). A second identical zero-id order is rejected identically +// (the replay path), proving the reject is stateless and deterministic. +func TestMirrorIngest_ZeroV2LogIdRejected(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockStore := NewMockScope(ctrl) + processor, err := NewRequestProcessor(nil, 0) + require.NoError(t, err) + + boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 1, NextLogId: 1, LastMirrorV2LogId: 5} + ledgerInfo := &commonpb.LedgerInfo{Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR} + + expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes() + + boundariesStub := setupBoundariesStub(mockStore) + boundariesStub.onGet(func(_ domain.LedgerKey) (raftcmdpb.LedgerBoundariesReader, error) { + return boundaries.AsReader(), nil + }) + // No mutation on a zero-id rejection. + boundariesStub.onPut(func(_ domain.LedgerKey, _ *raftcmdpb.LedgerBoundaries) { + t.Errorf("Boundaries().Put must not be called on a zero v2LogId rejection") + }) + ledgersStub := setupLedgersStub(mockStore) + ledgersStub.onPut(func(_ domain.LedgerKey, _ *commonpb.LedgerInfo) { + t.Errorf("Ledgers().Put must not be called on a zero v2LogId rejection") + }) + volumesStub := setupVolumesStub(mockStore) + volumesStub.onPut(func(_ domain.VolumeKey, _ *raftcmdpb.VolumePair) { + t.Errorf("Volumes().Put must not be called on a zero v2LogId rejection") + }) + + assertRejected := func() { + result, err := processor.ProcessOrder(mirrorCreatedTxOrder("mirror-ledger", 0, 42), mockStore) + require.Error(t, err) + require.Nil(t, result) + + var invalid *domain.ErrMirrorV2LogIDInvalid + require.ErrorAs(t, err, &invalid) + require.Equal(t, "mirror-ledger", invalid.Name) + } + + // Both the first and a replayed identical zero-id order are rejected the + // same way — the reject is a pure function of the order, mutating nothing. + assertRejected() + assertRejected() +} + func TestMirrorIngest_NotMirrorMode(t *testing.T) { t.Parallel() @@ -463,7 +719,8 @@ func TestMirrorIngest_RevertedTransaction_AbsentVolumes(t *testing.T) { Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR, } - boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 10, NextLogId: 1} + // Contiguous prefix: v2LogId 2 requires the applied prefix to be at 1. + boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 10, NextLogId: 1, LastMirrorV2LogId: 1} expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes() expectPutLedger(t, mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo) @@ -532,7 +789,8 @@ func TestMirrorIngest_RevertedTransaction_LinksOriginal(t *testing.T) { Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR, } - boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 10, NextLogId: 1} + // Contiguous prefix: v2LogId 2 requires the applied prefix to be at 1. + boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 10, NextLogId: 1, LastMirrorV2LogId: 1} expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes() expectPutLedger(t, mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo) diff --git a/internal/domain/processing/processor_test.go b/internal/domain/processing/processor_test.go index a193a66231..e3b3b3f541 100644 --- a/internal/domain/processing/processor_test.go +++ b/internal/domain/processing/processor_test.go @@ -440,3 +440,62 @@ func TestProcessOrders_OrdersResultEmpty(t *testing.T) { require.Equal(t, uint64(0), response.MinLogSequence) require.Equal(t, uint64(0), response.MaxLogSequence) } + +// TestProcessOrders_MirrorReplayEmitsNoLog pins the EN-1550 no-log outcome at +// the batch level: an idempotent mirror replay (v2LogId <= LastMirrorV2LogId) +// returns (nil, nil) from processMirrorIngest, so ProcessOrders must NOT consume +// a sequence id, NOT append a degenerate Log{Payload:nil}, and NOT record a +// CreatedLog — the per-order slot stays nil (skipped by WriteSet.Merge exactly +// like an idempotency-replay ReferenceSequence). Without the guard, replaying an +// already-applied source log would re-force its postings and double balances +// (#1581). The sink is asserted to receive zero Absorb calls. +func TestProcessOrders_MirrorReplayEmitsNoLog(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockStore := NewMockScope(ctrl) + processor, err := NewRequestProcessor(nil, 0) + require.NoError(t, err) + + // v2LogId 7 already applied. + boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 43, NextLogId: 5, LastMirrorV2LogId: 7} + ledgerInfo := &commonpb.LedgerInfo{Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR} + + expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes() + expectGetBoundaries(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, boundaries.AsReader(), nil).AnyTimes() + + // A no-op must consume no proposal-level sequence id: IncrementNextSequenceID + // is registered with Times(0) so any call fails the test. + mockStore.EXPECT().IncrementNextSequenceID().Times(0) + + // A replay of v2LogId 7 (already applied) followed by an even older one (5). + orders := []*raftcmdpb.Order{ + mirrorCreatedTxOrder("mirror-ledger", 7, 42), + mirrorCreatedTxOrder("mirror-ledger", 5, 41), + } + + sink := &countingSink{} + response, err := processor.ProcessOrders(orders, mockFactory(mockStore), sink) + require.NoError(t, err) + require.NotNil(t, response) + + // Two order slots exist (index alignment preserved) but both are nil — no + // log produced for either replay. + require.Len(t, response.Logs, 2) + require.Nil(t, response.Logs[0].GetCreatedLog()) + require.Nil(t, response.Logs[1].GetCreatedLog()) + + // No CreatedLogs, no sequence range, no sink absorb. + require.Empty(t, response.CreatedLogs) + require.Equal(t, uint64(0), response.MinLogSequence) + require.Equal(t, uint64(0), response.MaxLogSequence) + require.Equal(t, 0, sink.count) +} + +// countingSink records how many (order, log) pairs it absorbed so a test can +// assert a no-log order never reaches the sink. +type countingSink struct{ count int } + +func (s *countingSink) Absorb(_ *raftcmdpb.Order, _ *commonpb.Log) { s.count++ } diff --git a/internal/domain/reason.go b/internal/domain/reason.go index 53899d17dc..3a4ba77e06 100644 --- a/internal/domain/reason.go +++ b/internal/domain/reason.go @@ -123,6 +123,8 @@ func KindForReason(code commonpb.ErrorReason) ErrorKind { commonpb.ErrorReason_ERROR_REASON_STORAGE_OPERATION_FAILED, commonpb.ErrorReason_ERROR_REASON_TRANSACTION_STATE_INCONSISTENT, commonpb.ErrorReason_ERROR_REASON_NUMSCRIPT_RUNTIME, + commonpb.ErrorReason_ERROR_REASON_MIRROR_V2_LOG_ID_GAP, + commonpb.ErrorReason_ERROR_REASON_MIRROR_V2_LOG_ID_INVALID, commonpb.ErrorReason_ERROR_REASON_VOLUME_NOT_MATERIALIZED: return KindInternal default: diff --git a/internal/infra/attributes/baseline.go b/internal/infra/attributes/baseline.go index 2bdb68263d..970000269f 100644 --- a/internal/infra/attributes/baseline.go +++ b/internal/infra/attributes/baseline.go @@ -95,6 +95,16 @@ func writeBaselineAttributes(reader dal.PebbleReader, attrs *Attributes, db *peb return fmt.Errorf("writing baseline transactions: %w", err) } + // Boundary rows carry the per-ledger LedgerBoundaries — NextTransactionId + // (seeds the checker's tx-id counter for archived-history ledgers) and + // last_mirror_v2_log_id (the archived floor for the mirror-idempotency + // equality check, compareMirrorV2LogID). Without these, a ledger whose + // history sits entirely in an archived chapter is invisible to the + // baseline fold (foldBaselineBoundaries) and the checker false-positives. + if err := writeBaselineAttr(reader, attrs.Boundary, db); err != nil { + return fmt.Errorf("writing baseline boundaries: %w", err) + } + return nil } diff --git a/internal/proto/commonpb/common.pb.go b/internal/proto/commonpb/common.pb.go index 39f3cf31c3..544dfc3948 100644 --- a/internal/proto/commonpb/common.pb.go +++ b/internal/proto/commonpb/common.pb.go @@ -692,6 +692,8 @@ const ( ErrorReason_ERROR_REASON_WRITES_BLOCKED_DISK_FULL ErrorReason = 61 ErrorReason_ERROR_REASON_WRITES_BLOCKED_CLOCK_SKEW ErrorReason = 62 ErrorReason_ERROR_REASON_CHECKPOINT_NOT_READY ErrorReason = 63 + ErrorReason_ERROR_REASON_MIRROR_V2_LOG_ID_GAP ErrorReason = 64 + ErrorReason_ERROR_REASON_MIRROR_V2_LOG_ID_INVALID ErrorReason = 65 ) // Enum value maps for ErrorReason. @@ -761,6 +763,8 @@ var ( 61: "ERROR_REASON_WRITES_BLOCKED_DISK_FULL", 62: "ERROR_REASON_WRITES_BLOCKED_CLOCK_SKEW", 63: "ERROR_REASON_CHECKPOINT_NOT_READY", + 64: "ERROR_REASON_MIRROR_V2_LOG_ID_GAP", + 65: "ERROR_REASON_MIRROR_V2_LOG_ID_INVALID", } ErrorReason_value = map[string]int32{ "ERROR_REASON_UNSPECIFIED": 0, @@ -827,6 +831,8 @@ var ( "ERROR_REASON_WRITES_BLOCKED_DISK_FULL": 61, "ERROR_REASON_WRITES_BLOCKED_CLOCK_SKEW": 62, "ERROR_REASON_CHECKPOINT_NOT_READY": 63, + "ERROR_REASON_MIRROR_V2_LOG_ID_GAP": 64, + "ERROR_REASON_MIRROR_V2_LOG_ID_INVALID": 65, } ) @@ -13505,7 +13511,7 @@ const file_common_proto_rawDesc = "" + "\x12LEDGER_MODE_MIRROR\x10\x01*Q\n" + "\x0fMirrorSyncState\x12\x1d\n" + "\x19MIRROR_SYNC_STATE_SYNCING\x10\x00\x12\x1f\n" + - "\x1bMIRROR_SYNC_STATE_FOLLOWING\x10\x01*\xff\x13\n" + + "\x1bMIRROR_SYNC_STATE_FOLLOWING\x10\x01*\xd1\x14\n" + "\vErrorReason\x12\x1c\n" + "\x18ERROR_REASON_UNSPECIFIED\x10\x00\x12&\n" + "\"ERROR_REASON_LEDGER_ALREADY_EXISTS\x10\x01\x12!\n" + @@ -13571,7 +13577,9 @@ const file_common_proto_rawDesc = "" + "\x1eERROR_REASON_CLUSTER_UNHEALTHY\x10<\x12)\n" + "%ERROR_REASON_WRITES_BLOCKED_DISK_FULL\x10=\x12*\n" + "&ERROR_REASON_WRITES_BLOCKED_CLOCK_SKEW\x10>\x12%\n" + - "!ERROR_REASON_CHECKPOINT_NOT_READY\x10?*Q\n" + + "!ERROR_REASON_CHECKPOINT_NOT_READY\x10?\x12%\n" + + "!ERROR_REASON_MIRROR_V2_LOG_ID_GAP\x10@\x12)\n" + + "%ERROR_REASON_MIRROR_V2_LOG_ID_INVALID\x10A*Q\n" + "\x14ChartEnforcementMode\x12\x1c\n" + "\x18CHART_ENFORCEMENT_STRICT\x10\x00\x12\x1b\n" + "\x17CHART_ENFORCEMENT_AUDIT\x10\x01*i\n" + diff --git a/internal/proto/raftcmdpb/raft_cmd.pb.go b/internal/proto/raftcmdpb/raft_cmd.pb.go index e85e6a6e0b..2ffcead892 100644 --- a/internal/proto/raftcmdpb/raft_cmd.pb.go +++ b/internal/proto/raftcmdpb/raft_cmd.pb.go @@ -4925,8 +4925,12 @@ type LedgerBoundaries struct { TransientUsedCount uint64 `protobuf:"fixed64,8,opt,name=transient_used_count,json=transientUsedCount,proto3" json:"transient_used_count,omitempty"` RevertCount uint64 `protobuf:"fixed64,9,opt,name=revert_count,json=revertCount,proto3" json:"revert_count,omitempty"` NumscriptExecutionCount uint64 `protobuf:"fixed64,10,opt,name=numscript_execution_count,json=numscriptExecutionCount,proto3" json:"numscript_execution_count,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Highest source (v2) log ID already applied to this mirror ledger. Gates + // idempotent replay: a MirrorIngest whose v2LogId is <= this value is a + // deterministic no-op on the FSM apply path (see processMirrorIngest). + LastMirrorV2LogId uint64 `protobuf:"fixed64,11,opt,name=last_mirror_v2_log_id,json=lastMirrorV2LogId,proto3" json:"last_mirror_v2_log_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *LedgerBoundaries) Reset() { @@ -5029,6 +5033,13 @@ func (x *LedgerBoundaries) GetNumscriptExecutionCount() uint64 { return 0 } +func (x *LedgerBoundaries) GetLastMirrorV2LogId() uint64 { + if x != nil { + return x.LastMirrorV2LogId + } + return 0 +} + type VolumePair struct { state protoimpl.MessageState `protogen:"open.v1"` Input *commonpb.Uint256 `protobuf:"bytes,1,opt,name=input,proto3" json:"input,omitempty"` @@ -6455,7 +6466,7 @@ const file_raft_cmd_proto_rawDesc = "" + "\vcreated_log\x18\x01 \x01(\v2\v.common.LogH\x00R\n" + "createdLog\x12/\n" + "\x12reference_sequence\x18\x02 \x01(\x06H\x00R\x11referenceSequenceB\x06\n" + - "\x04type\"\xc3\x03\n" + + "\x04type\"\xf5\x03\n" + "\x10LedgerBoundaries\x12.\n" + "\x13next_transaction_id\x18\x01 \x01(\x06R\x11nextTransactionId\x12\x1e\n" + "\vnext_log_id\x18\x02 \x01(\x06R\tnextLogId\x12!\n" + @@ -6467,7 +6478,8 @@ const file_raft_cmd_proto_rawDesc = "" + "\x14transient_used_count\x18\b \x01(\x06R\x12transientUsedCount\x12!\n" + "\frevert_count\x18\t \x01(\x06R\vrevertCount\x12:\n" + "\x19numscript_execution_count\x18\n" + - " \x01(\x06R\x17numscriptExecutionCount\"\\\n" + + " \x01(\x06R\x17numscriptExecutionCount\x120\n" + + "\x15last_mirror_v2_log_id\x18\v \x01(\x06R\x11lastMirrorV2LogId\"\\\n" + "\n" + "VolumePair\x12%\n" + "\x05input\x18\x01 \x01(\v2\x0f.common.Uint256R\x05input\x12'\n" + diff --git a/internal/proto/raftcmdpb/raft_cmd_reader.pb.go b/internal/proto/raftcmdpb/raft_cmd_reader.pb.go index 84a4a03111..75383e2004 100644 --- a/internal/proto/raftcmdpb/raft_cmd_reader.pb.go +++ b/internal/proto/raftcmdpb/raft_cmd_reader.pb.go @@ -5262,6 +5262,7 @@ type LedgerBoundariesReader interface { GetTransientUsedCount() uint64 GetRevertCount() uint64 GetNumscriptExecutionCount() uint64 + GetLastMirrorV2LogId() uint64 Mutate() *LedgerBoundaries } @@ -5307,6 +5308,10 @@ func (r *ledgerBoundariesReadonly) GetNumscriptExecutionCount() uint64 { return (*LedgerBoundaries)(r).GetNumscriptExecutionCount() } +func (r *ledgerBoundariesReadonly) GetLastMirrorV2LogId() uint64 { + return (*LedgerBoundaries)(r).GetLastMirrorV2LogId() +} + func (r *ledgerBoundariesReadonly) Mutate() *LedgerBoundaries { return (*LedgerBoundaries)(r).CloneVT() } diff --git a/internal/proto/raftcmdpb/raft_cmd_vtproto.pb.go b/internal/proto/raftcmdpb/raft_cmd_vtproto.pb.go index ae702a345b..b8fc14cdbb 100644 --- a/internal/proto/raftcmdpb/raft_cmd_vtproto.pb.go +++ b/internal/proto/raftcmdpb/raft_cmd_vtproto.pb.go @@ -1973,6 +1973,7 @@ func (m *LedgerBoundaries) CloneVT() *LedgerBoundaries { r.TransientUsedCount = m.TransientUsedCount r.RevertCount = m.RevertCount r.NumscriptExecutionCount = m.NumscriptExecutionCount + r.LastMirrorV2LogId = m.LastMirrorV2LogId if len(m.unknownFields) > 0 { r.unknownFields = make([]byte, len(m.unknownFields)) copy(r.unknownFields, m.unknownFields) @@ -5847,6 +5848,9 @@ func (this *LedgerBoundaries) EqualVT(that *LedgerBoundaries) bool { if this.NumscriptExecutionCount != that.NumscriptExecutionCount { return false } + if this.LastMirrorV2LogId != that.LastMirrorV2LogId { + return false + } return string(this.unknownFields) == string(that.unknownFields) } @@ -11117,6 +11121,12 @@ func (m *LedgerBoundaries) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.LastMirrorV2LogId != 0 { + i -= 8 + binary.LittleEndian.PutUint64(dAtA[i:], uint64(m.LastMirrorV2LogId)) + i-- + dAtA[i] = 0x59 + } if m.NumscriptExecutionCount != 0 { i -= 8 binary.LittleEndian.PutUint64(dAtA[i:], uint64(m.NumscriptExecutionCount)) @@ -14461,6 +14471,9 @@ func (m *LedgerBoundaries) SizeVT() (n int) { if m.NumscriptExecutionCount != 0 { n += 9 } + if m.LastMirrorV2LogId != 0 { + n += 9 + } n += len(m.unknownFields) return n } @@ -25777,6 +25790,16 @@ func (m *LedgerBoundaries) UnmarshalVT(dAtA []byte) error { } m.NumscriptExecutionCount = uint64(binary.LittleEndian.Uint64(dAtA[iNdEx:])) iNdEx += 8 + case 11: + if wireType != 1 { + return fmt.Errorf("proto: wrong wireType = %d for field LastMirrorV2LogId", wireType) + } + m.LastMirrorV2LogId = 0 + if (iNdEx + 8) > l { + return io.ErrUnexpectedEOF + } + m.LastMirrorV2LogId = uint64(binary.LittleEndian.Uint64(dAtA[iNdEx:])) + iNdEx += 8 default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/internal/proto/servicepb/bucket.pb.go b/internal/proto/servicepb/bucket.pb.go index 363f5cc3e0..8e9f69759f 100644 --- a/internal/proto/servicepb/bucket.pb.go +++ b/internal/proto/servicepb/bucket.pb.go @@ -67,6 +67,15 @@ const ( // for an order that should have been rejected hard. The skippable_reasons // whitelist on the audit-bound Order is the source of truth. CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_INVALID_SKIP CheckStoreErrorType = 11 + // Emitted when a mirror ledger's stored LedgerBoundaries.last_mirror_v2_log_id + // does NOT equal the maximum audited MirrorIngest.v2_log_id for that ledger. + // last_mirror_v2_log_id is the idempotent-replay high-water mark (EN-1550): the + // FSM enforces a contiguous applied prefix, so at rest the stored value must + // equal exactly the max audited v2_log_id. A value ABOVE the audit claims to + // have consumed a source v2 log no audit entry recorded (future ingests wrongly + // skipped); a value BELOW means the persisted projection lost applied ground — + // both are corruption. This is a full invariant-#8 equality check, not a bound. + CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH CheckStoreErrorType = 12 ) // Enum value maps for CheckStoreErrorType. @@ -84,6 +93,7 @@ var ( 9: "CHECK_STORE_ERROR_TYPE_IDEMPOTENCY_MISMATCH", 10: "CHECK_STORE_ERROR_TYPE_INDEX_MISMATCH", 11: "CHECK_STORE_ERROR_TYPE_INVALID_SKIP", + 12: "CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH", } CheckStoreErrorType_value = map[string]int32{ "CHECK_STORE_ERROR_TYPE_UNSPECIFIED": 0, @@ -98,6 +108,7 @@ var ( "CHECK_STORE_ERROR_TYPE_IDEMPOTENCY_MISMATCH": 9, "CHECK_STORE_ERROR_TYPE_INDEX_MISMATCH": 10, "CHECK_STORE_ERROR_TYPE_INVALID_SKIP": 11, + "CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH": 12, } ) @@ -9452,7 +9463,7 @@ const file_bucket_proto_rawDesc = "" + "\x12entities_with_null\x18\x05 \x01(\x06R\x10entitiesWithNull\"\x10\n" + "\x0eBarrierRequest\"4\n" + "\x0fBarrierResponse\x12!\n" + - "\fcommit_index\x18\x01 \x01(\x06R\vcommitIndex*\xb6\x04\n" + + "\fcommit_index\x18\x01 \x01(\x06R\vcommitIndex*\xea\x04\n" + "\x13CheckStoreErrorType\x12&\n" + "\"CHECK_STORE_ERROR_TYPE_UNSPECIFIED\x10\x00\x12(\n" + "$CHECK_STORE_ERROR_TYPE_HASH_MISMATCH\x10\x01\x12'\n" + @@ -9466,7 +9477,8 @@ const file_bucket_proto_rawDesc = "" + "+CHECK_STORE_ERROR_TYPE_IDEMPOTENCY_MISMATCH\x10\t\x12)\n" + "%CHECK_STORE_ERROR_TYPE_INDEX_MISMATCH\x10\n" + "\x12'\n" + - "#CHECK_STORE_ERROR_TYPE_INVALID_SKIP\x10\v*W\n" + + "#CHECK_STORE_ERROR_TYPE_INVALID_SKIP\x10\v\x122\n" + + ".CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH\x10\f*W\n" + "\x12PatternSegmentType\x12\x1e\n" + "\x1aPATTERN_SEGMENT_TYPE_FIXED\x10\x00\x12!\n" + "\x1dPATTERN_SEGMENT_TYPE_VARIABLE\x10\x01*\x9c\x01\n" + diff --git a/misc/proto/bucket.proto b/misc/proto/bucket.proto index 41f60f9977..08df8cb521 100644 --- a/misc/proto/bucket.proto +++ b/misc/proto/bucket.proto @@ -799,6 +799,15 @@ enum CheckStoreErrorType { // for an order that should have been rejected hard. The skippable_reasons // whitelist on the audit-bound Order is the source of truth. CHECK_STORE_ERROR_TYPE_INVALID_SKIP = 11; + // Emitted when a mirror ledger's stored LedgerBoundaries.last_mirror_v2_log_id + // does NOT equal the maximum audited MirrorIngest.v2_log_id for that ledger. + // last_mirror_v2_log_id is the idempotent-replay high-water mark (EN-1550): the + // FSM enforces a contiguous applied prefix, so at rest the stored value must + // equal exactly the max audited v2_log_id. A value ABOVE the audit claims to + // have consumed a source v2 log no audit entry recorded (future ingests wrongly + // skipped); a value BELOW means the persisted projection lost applied ground — + // both are corruption. This is a full invariant-#8 equality check, not a bound. + CHECK_STORE_ERROR_TYPE_MIRROR_V2LOGID_MISMATCH = 12; } message CheckStoreError { diff --git a/misc/proto/common.proto b/misc/proto/common.proto index 1d4d6426c0..4c7334da89 100644 --- a/misc/proto/common.proto +++ b/misc/proto/common.proto @@ -1191,6 +1191,8 @@ enum ErrorReason { ERROR_REASON_WRITES_BLOCKED_DISK_FULL = 61; ERROR_REASON_WRITES_BLOCKED_CLOCK_SKEW = 62; ERROR_REASON_CHECKPOINT_NOT_READY = 63; + ERROR_REASON_MIRROR_V2_LOG_ID_GAP = 64; + ERROR_REASON_MIRROR_V2_LOG_ID_INVALID = 65; } // IdempotencyFailure captures a definitive business rejection so a retried key diff --git a/misc/proto/raft_cmd.proto b/misc/proto/raft_cmd.proto index 59b5d2fe03..364e2c4d8c 100644 --- a/misc/proto/raft_cmd.proto +++ b/misc/proto/raft_cmd.proto @@ -706,6 +706,10 @@ message LedgerBoundaries { fixed64 transient_used_count = 8; fixed64 revert_count = 9; fixed64 numscript_execution_count = 10; + // Highest source (v2) log ID already applied to this mirror ledger. Gates + // idempotent replay: a MirrorIngest whose v2LogId is <= this value is a + // deterministic no-op on the FSM apply path (see processMirrorIngest). + fixed64 last_mirror_v2_log_id = 11; } message VolumePair {