fix(backup): rebuild the reversion bitsets on restore

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

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 |

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
 ```
 

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<<bit) != 0 {
+					callback(errorEventWithTx(servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_REVERTED_MISMATCH,
+						fmt.Sprintf("reversion bit missing for tx %d in ledger %q: the audit reverts it but the stored bitset does not — the already-reverted gate would re-admit a double revert", txID, name),
+						name, txID))
+				} else {
+					callback(errorEventWithTx(servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_REVERTED_MISMATCH,
+						fmt.Sprintf("unaudited reversion bit for tx %d in ledger %q: the stored bitset marks it reverted but no audit-backed revert exists", txID, name),
+						name, txID))
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
 // checkReversionInvariants tracks transaction IDs and validates reversion invariants
 // during log replay.
 func checkReversionInvariants(

internal/application/check/checker_reversions_test.go

@@ -0,0 +1,147 @@
+package check
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	logging "github.com/formancehq/go-libs/v5/pkg/observe/log"
+
+	"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/servicepb"
+	"github.com/formancehq/ledger/v3/internal/storage/dal"
+)
+
+// writeReversionWord persists one reversion bitset word the way the FSM does.
+func writeReversionWord(t *testing.T, store *dal.Store, ledger string, wordIndex, value uint64) {
+	t.Helper()
+
+	batch := store.OpenWriteSession()
+	require.NoError(t, state.SaveReversionWord(batch, ledger, wordIndex, value))
+	require.NoError(t, batch.Commit())
+}
+
+// collectReversionEvents runs compareReversions against the store's persisted
+// bitsets with the given audit-derived set and returns the REVERTED_MISMATCH
+// errors.
+func collectReversionEvents(t *testing.T, store *dal.Store, derived map[string]*bitset.Bitset, knownLedgers map[string]struct{}) []*servicepb.CheckStoreError {
+	t.Helper()
+
+	checker := NewChecker(store, attributes.New(), "reversions-cluster", nil, nil, logging.Testing())
+
+	handle, err := store.NewReadHandle()
+	require.NoError(t, err)
+
+	defer func() { _ = handle.Close() }()
+
+	var got []*servicepb.CheckStoreError
+
+	require.NoError(t, checker.compareReversions(handle, derived, knownLedgers, func(event *servicepb.CheckStoreEvent) {
+		if e, ok := event.GetType().(*servicepb.CheckStoreEvent_Error); ok &&
+			e.Error.GetErrorType() == servicepb.CheckStoreErrorType_CHECK_STORE_ERROR_TYPE_REVERTED_MISMATCH {
+			got = append(got, e.Error)
+		}
+	}))
+
+	return got
+}
+
+func reversionSet(ids ...uint64) map[string]*bitset.Bitset {
+	bs := &bitset.Bitset{}
+	for _, id := range ids {
+		bs.Set(id)
+	}
+
+	return map[string]*bitset.Bitset{"ledger-a": bs}
+}
+
+func TestCompareReversions_Match(t *testing.T) {
+	t.Parallel()
+
+	store := createTestStore(t)
+	writeReversionWord(t, store, "ledger-a", 0, 1<<3|1<<7)
+	writeReversionWord(t, store, "ledger-a", 1, 1<<6) // tx 70
+
+	got := collectReversionEvents(t, store, reversionSet(3, 7, 70),
+		map[string]struct{}{"ledger-a": {}})
+	require.Empty(t, got)
+}
+
+func TestCompareReversions_MissingBitFlagged(t *testing.T) {
+	t.Parallel()
+
+	store := createTestStore(t)
+
+	got := collectReversionEvents(t, store, reversionSet(3),
+		map[string]struct{}{"ledger-a": {}})
+	require.Len(t, got, 1)
+	require.Equal(t, "ledger-a", got[0].GetLedger())
+	require.Equal(t, uint64(3), got[0].GetTransactionId())
+	require.Contains(t, got[0].GetMessage(), "reversion bit missing")
+}
+
+func TestCompareReversions_UnauditedBitFlagged(t *testing.T) {
+	t.Parallel()
+
+	store := createTestStore(t)
+	writeReversionWord(t, store, "ledger-a", 1, 1<<6) // tx 70, no audit backing
+
+	got := collectReversionEvents(t, store, nil,
+		map[string]struct{}{"ledger-a": {}})
+	require.Len(t, got, 1)
+	require.Equal(t, uint64(70), got[0].GetTransactionId())
+	require.Contains(t, got[0].GetMessage(), "unaudited reversion bit")
+}
+
+// A persisted pending-cleanup marker is itself an unverified projection: it
+// must not exempt an audit-live ledger from the comparison, or a forged
+// marker hides bitset tampering.
+func TestCompareReversions_PendingCleanupMarkerDoesNotHideMismatch(t *testing.T) {
+	t.Parallel()
+
+	store := createTestStore(t)
+	writeReversionWord(t, store, "ledger-a", 0, 1<<5)
+
+	batch := store.OpenWriteSession()
+	require.NoError(t, state.SavePendingLedgerCleanup(batch, "ledger-a", 9))
+	require.NoError(t, batch.Commit())
+
+	got := collectReversionEvents(t, store, nil,
+		map[string]struct{}{"ledger-a": {}})
+	require.Len(t, got, 1)
+	require.Equal(t, uint64(5), got[0].GetTransactionId())
+	require.Contains(t, got[0].GetMessage(), "unaudited reversion bit")
+}
+
+// DeleteLedger removes the reversion rows at apply on both the live path and
+// the replay, so stored rows for a ledger the audit does not know as live are
+// never legitimate.
+func TestCompareReversions_NonLiveLedgerRowsFlagged(t *testing.T) {
+	t.Parallel()
+
+	store := createTestStore(t)
+	writeReversionWord(t, store, "ghost", 0, 1<<2)
+
+	got := collectReversionEvents(t, store, nil, map[string]struct{}{"ledger-a": {}})
+	require.Len(t, got, 1)
+	require.Equal(t, "ghost", got[0].GetLedger())
+	require.Contains(t, got[0].GetMessage(), "non-live ledger")
+}
+
+// Rows that fail to decode must surface as events, not silently narrow the
+// comparison.
+func TestCompareReversions_MalformedRowFlagged(t *testing.T) {
+	t.Parallel()
+
+	store := createTestStore(t)
+
+	batch := store.OpenWriteSession()
+	require.NoError(t, batch.SetBytes([]byte{0x03, 0x01, 'x'}, []byte{0x01}))
+	require.NoError(t, batch.Commit())
+
+	got := collectReversionEvents(t, store, nil, map[string]struct{}{})
+	require.Len(t, got, 1)
+	require.Contains(t, got[0].GetMessage(), "malformed reversion row")
+}

internal/application/check/checker_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/formancehq/ledger/v3/internal/infra/attributes"
 	"github.com/formancehq/ledger/v3/internal/infra/cache"
 	"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"
@@ -60,6 +61,7 @@ type testEngine struct {
 	idempotency            map[string]*commonpb.IdempotencyKeyValue
 	references             map[string]*commonpb.TransactionReferenceValue
 	transactionStates      map[string]*commonpb.TransactionState
+	reversions             map[string]*bitset.Bitset
 	currentOpenChapter     *commonpb.Chapter
 	closingChapters        []*commonpb.Chapter
 	nextLedgerID           uint32
@@ -99,6 +101,7 @@ func newTestEngine(t *testing.T) *testEngine {
 		idempotency:         make(map[string]*commonpb.IdempotencyKeyValue),
 		references:          make(map[string]*commonpb.TransactionReferenceValue),
 		transactionStates:   make(map[string]*commonpb.TransactionState),
+		reversions:          make(map[string]*bitset.Bitset),
 		nextChapterID:       1,
 		nextAuditSequenceID: 1,
 		raftIndex:           1,
@@ -183,6 +186,28 @@ func (e *testEngine) processAndCommit(orders ...*raftcmdpb.Order) []*commonpb.Lo
 		require.NoError(e.t, err)
 	}
 
+	// Fold this proposal's reversions into the engine bitsets and persist the
+	// touched words, like WriteSet's dirty-word flush.
+	for keyStr, isReverted := range store.reverted {
+		if !isReverted {
+			continue
+		}
+
+		var tk domain.TransactionKey
+		require.NoError(e.t, tk.Unmarshal([]byte(keyStr)))
+
+		bs := e.reversions[tk.LedgerName]
+		if bs == nil {
+			bs = &bitset.Bitset{}
+			e.reversions[tk.LedgerName] = bs
+		}
+
+		bs.Set(tk.ID)
+
+		wordIndex := tk.ID / 64
+		require.NoError(e.t, state.SaveReversionWord(batch, tk.LedgerName, wordIndex, bs.Word(wordIndex)))
+	}
+
 	// Write ledger info
 	for _, info := range e.ledgers {
 		err := state.SaveLedger(batch, info)
@@ -369,7 +394,8 @@ type scopeImpl struct {
 	modifiedVolumes  map[string]struct{}
 	modifiedMetadata map[string]struct{}
 	modifiedTxStates map[string]struct{}
-	// In-memory reverted status (bitset-like, not persisted to Pebble)
+	// Reversions marked by this proposal; folded into the engine's per-ledger
+	// bitsets and persisted as reversion words at commit, like WriteSet does.
 	reverted map[string]bool
 }
 
@@ -559,7 +585,15 @@ func (s *scopeImpl) Indexes() processing.Accessor[domain.IndexKey, *commonpb.Ind
 }
 
 func (s *scopeImpl) GetReverted(key domain.TransactionKey) (bool, error) {
-	return s.reverted[string(key.Bytes())], nil
+	if s.reverted[string(key.Bytes())] {
+		return true, nil
+	}
+
+	if bs := s.engine.reversions[key.LedgerName]; bs != nil {
+		return bs.Test(key.ID), nil
+	}
+
+	return false, nil
 }
 
 func (s *scopeImpl) PutReverted(key domain.TransactionKey, reverted bool) {