fix: verify encryption snapshot checkpoints before gap scans

docs/encryption_compacted_restart.md

@@ -0,0 +1,123 @@
+# Encrypted restart after Raft compaction
+
+## Failure
+
+The sidecar's `raft_applied_index` advances only on sidecar-mutating
+encryption entries. Ordinary writes and writer registrations do not advance
+it. A healthy encrypted cluster can therefore retain a sidecar at index 6
+while the FSM has applied index 877 and its Raft prefix has been compacted.
+The startup gap scanner cannot inspect `(6, 877]` and correctly refuses to
+guess whether the missing prefix contains another encryption change.
+
+## Snapshot Witness
+
+After a successful bootstrap or rotation apply, the FSM captures the last
+sidecar-mutating index and a SHA-256 digest of the replicated sidecar state.
+The digest covers wrapped keys, key purposes, active keys, both envelope
+cutovers, and the sidecar index. It excludes replica-local epochs, creation
+times, and the sidecar serialization version. It is a comparison witness,
+not a replacement for authenticated storage values or an epoch authority.
+
+The witness is an `EKVENCP1` extension of the existing length-delimited v2
+FSM header. Older v2 readers can skip the extension. A full physical restore
+preserves it; the cold-start skip path installs it only after full payload
+CRC and Raft-token validation. A newly captured snapshot reuses the witness
+restored from its predecessor, rather than trusting a new disk-sidecar read
+to rewrite the history represented by that snapshot.
+
+The same witness is synchronously persisted as reserved operational FSM
+metadata after each successful encryption apply, before the apply handler
+returns. It does not issue a user timestamp or nonce. When the cold-start
+skip gate retains a newer durable FSM, it can retain that FSM's newer
+witness only through its durable applied index. A checkpoint ahead of that
+index must be reconciled by normal WAL replay; new snapshot capture refuses
+to pair an unreconciled durable record with an older header. Full restore
+requires its body record to match the header. Native physical snapshots
+preserve the record and the offline decoder classifies it as operational
+metadata rather than an MVCC user version.
+
+The startup guard opens the engine's installed snapshot, not an arbitrary
+file in the snapshot directory. Token index, Raft metadata index, token CRC,
+payload CRC, checkpoint index, sidecar state digest, and envelope cutover
+must agree. Only then does that snapshot cover the compacted prefix. The
+snapshot is selected and opened under `snapshotMu`; full checksum verification
+and reads then use that same descriptor without holding up snapshot work.
+Removing the old path after it has been opened cannot invalidate that reader.
+The
+retained suffix remains subject to the original gap scan; encrypted entries
+are authenticated and decrypted before inspecting their opcode. Missing
+keys or invalid authentication tags are startup failures, never "no hit".
+
+A newer snapshot can compact the retained suffix between snapshot selection
+and its scan. Only `ErrCompacted` from that scan triggers a retry against the
+currently installed, fully reverified snapshot. The verified covered prefix
+must advance, and at most three scans are attempted. Absent or inconsistent
+proof, CRC/authentication errors, a nonadvancing snapshot, or continuous
+compaction past that bound still refuse startup. No compacted history is
+treated as harmless merely because a scan was overtaken.
+Each retry rereads the sidecar because the running Raft apply loop may have
+completed a legitimate rotation in the meantime. A regressing sidecar index
+or read failure refuses startup; refreshed metadata does not authorize an
+epoch reset or bypass the snapshot witness comparison.
+
+The membership/collision, pre-bump epoch rollback, post-replay epoch rollback,
+writer-registration, and physical snapshot recovery paths remain enabled.
+The witness deliberately cannot authorize a nonce epoch reset.
+
+## Compatibility and Scope
+
+Older binaries do not recognize the reserved operational key
+`_meta_encryption_snapshot_checkpoint`. The ability to skip the v2 header
+extension does not establish compatibility with this physical body record.
+Downgrading a store written by this version has not been validated as safe.
+Mixed-version native physical snapshot transfer to an older follower is not
+supported by this change: upgrade the receiving follower before allowing it
+to install a snapshot produced by this version. Do not delete the checkpoint
+record or bypass snapshot, sidecar, or epoch checks to enable a downgrade.
+The complete checkpoint may be followed by future v2 header extensions;
+readers consume those extra header bytes without treating them as body data.
+An incomplete known checkpoint still fails parsing.
+
+An old snapshot without a witness does **not** authorize skipping compacted
+history. Such a node can still refuse startup and needs an independently
+validated operational recovery path. This change does not manufacture
+proof for already-compacted legacy data, silently resync a sidecar, remove
+snapshot checks, or claim that the separate DynamoDB acknowledged-append
+consistency anomaly is fixed. No change from PR #1286 is included.
+
+## Verification
+
+- The compacted-gap regression first failed with the original `(6, 877]`
+  scan; the snapshot-witness guard makes the valid case pass.
+- Negative controls retain refusal for stale sidecar/index, changed key
+  material, future checkpoint, absent/truncated proof, retained rotation,
+  snapshot verification error, mismatched token index/CRC, corrupt physical
+  payload, invalid encrypted-tail tag, and a rolled-back local epoch.
+- Snapshot witness tests cover full restore and the verified header-only
+  skip path, including subsequent snapshot production.
+- When the sidecar index equals or exceeds the engine's applied index, the
+  gap guard does not reopen the snapshot. Engine restore still verifies its
+  full CRC; any nonempty gap retains snapshot verification and suffix scans.
+- `go test . ./kv ./internal/encryption ./internal/raftengine/etcd -count=1`
+  passed. The encryption/snapshot/skip-gate selection also passed with `-race`.
+- Scoped `golangci-lint` passed with Go 1.26.6 (0 issues).
+- Local three-process encrypted workload, six verified leader transfers,
+  snapshot interval 20, then stop/reopen of all three processes preserved
+  all 28 items. This guard-specific run used an explicit HTTP leader map;
+  it is not evidence for the separate no-map acceptance requirement.
+
+## Caller Audit
+
+`rg` audit covered all production readers of the FSM header (`kvFSM.Restore`,
+`ParseSnapshotHeader`, `tsoFSM.Restore`, and startup cutover preflight), the
+verified-header skip path, both encryption apply handlers, the startup gap
+guard and scanner factory, and the scanner's sole per-entry predicate call.
+The old three-result header API remains available to callers that do not
+consume the extension. The pure `GuardSidecarBehindRaftLog` API is unchanged.
+The no-gap fast path keeps the original sidecar index rather than advancing
+it. Its sole production caller still invokes the pure gap guard afterward;
+the engine's full-restore and header-only-skip CRC checks are unchanged.
+The snapshot-gap retry helper has one production caller, the startup guard.
+Each retry recomputes the gap from a freshly read sidecar and a freshly verified
+installed snapshot; only the typed suffix-compaction error is retried. The
+pure guard and all its other callers retain their original semantics.

internal/backup/snapshot_reader.go

@@ -272,6 +272,7 @@ var pebbleOperationalKeys = [][]byte{
 	[]byte("_meta_min_retained_ts"),
 	[]byte("_meta_pending_min_retained_ts"),
 	[]byte("_meta_applied_index"),
+	[]byte("_meta_encryption_snapshot_checkpoint"),
 	[]byte("_migack"),
 	[]byte("_mighlc"),
 	[]byte("_migpromote"),

internal/encryption/snapshot_checkpoint.go

@@ -0,0 +1,46 @@
+package encryption
+
+import (
+	"crypto/sha256"
+	"encoding/json"
+	"maps"
+
+	"github.com/cockroachdb/errors"
+)
+
+// SnapshotCheckpoint witnesses the last sidecar-mutating entry represented
+// by an FSM snapshot. It is meaningful only inside a verified Raft snapshot.
+type SnapshotCheckpoint struct {
+	Index uint64
+	State [sha256.Size]byte
+}
+
+// SidecarCheckpoint excludes node-local nonce epochs and creation times, which
+// differ between replicas. Epoch rollback is checked separately against the
+// writer registry, both before the startup bump and after replay.
+func SidecarCheckpoint(sc *Sidecar) (SnapshotCheckpoint, error) {
+	if sc == nil {
+		return SnapshotCheckpoint{}, errors.New("encryption snapshot checkpoint: missing sidecar")
+	}
+	state := *sc
+	state.Version = 0
+	state.Keys = maps.Clone(sc.Keys)
+	for id, key := range state.Keys {
+		key.LocalEpoch = 0
+		key.Created = ""
+		state.Keys[id] = key
+	}
+	payload, err := json.Marshal(state)
+	if err != nil {
+		return SnapshotCheckpoint{}, errors.Wrap(err, "encode encryption snapshot checkpoint")
+	}
+	return SnapshotCheckpoint{Index: sc.RaftAppliedIndex, State: sha256.Sum256(payload)}, nil
+}
+
+func (a *Applier) SnapshotCheckpoint() (SnapshotCheckpoint, error) {
+	sc, err := ReadSidecar(a.sidecarPath)
+	if err != nil {
+		return SnapshotCheckpoint{}, err
+	}
+	return SidecarCheckpoint(sc)
+}

internal/raftengine/etcd/encryption_checkpoint_test.go

@@ -0,0 +1,74 @@
+package etcd
+
+import (
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/bootjp/elastickv/internal/encryption/fsmwire"
+	"github.com/stretchr/testify/require"
+	etcdraft "go.etcd.io/raft/v3"
+	raftpb "go.etcd.io/raft/v3/raftpb"
+)
+
+func TestEncryptionScannerEncryptedRetainedTail(t *testing.T) {
+	for _, corrupt := range []bool{false, true} {
+		t.Run(map[bool]string{false: "rotation", true: "invalid-tag"}[corrupt], func(t *testing.T) {
+			cipher, id := raftCipherFixture(t)
+			entry := envelopeEntry(t, cipher, id, 2, []byte{fsmwire.OpRotation})
+			if corrupt {
+				entry.Data[len(entry.Data)-1] ^= 1
+			}
+			engine := newTestEngine(nil, cipher, func() uint64 { return 1 })
+			engine.storage = scannerWithEntries(t, newEntry(1, fsmwire.OpBootstrap), entry).storage
+			hit, err := engine.EncryptionScanner().HasEncryptionRelevantEntryInRange(1, 2)
+			if corrupt {
+				require.Error(t, err, "ciphertext that cannot be inspected must refuse startup")
+			} else {
+				require.NoError(t, err)
+				require.True(t, hit, "encrypted rotation must not be treated as an ordinary write")
+			}
+		})
+	}
+}
+
+func TestEncryptionSnapshotRequiresMatchingPhysicalEnvelope(t *testing.T) {
+	for _, variant := range []string{"valid", "stale-token", "mismatched-index", "corrupt-payload"} {
+		t.Run(variant, func(t *testing.T) {
+			dir := t.TempDir()
+			payload := []byte("verified FSM snapshot bytes")
+			crc, path := writeFSMFileForTest(t, dir, 100, payload)
+			index := uint64(100)
+			if variant == "stale-token" {
+				crc ^= 1
+			}
+			if variant == "mismatched-index" {
+				index = 99
+			}
+			if variant == "corrupt-payload" {
+				file, err := os.OpenFile(path, os.O_RDWR, 0)
+				require.NoError(t, err)
+				_, err = file.WriteAt([]byte("X"), 0)
+				require.NoError(t, err)
+				require.NoError(t, file.Close())
+			}
+			storage := etcdraft.NewMemoryStorage()
+			require.NoError(t, storage.ApplySnapshot(&raftpb.Snapshot{Data: encodeSnapshotToken(index, crc), Metadata: testSnapshotMetadata(100, 1, nil)}))
+			engine := &Engine{storage: storage, fsmSnapDir: filepath.Clean(dir)}
+			reader, gotIndex, err := engine.EncryptionSnapshot()
+			if variant != "valid" {
+				require.Error(t, err)
+				require.Nil(t, reader)
+				return
+			}
+			require.NoError(t, err)
+			defer reader.Close()
+			require.NoError(t, os.Remove(path), "an opened descriptor remains valid after snapshot purge")
+			got, err := io.ReadAll(reader)
+			require.NoError(t, err)
+			require.Equal(t, payload, got)
+			require.Equal(t, uint64(100), gotIndex)
+		})
+	}
+}

internal/raftengine/etcd/encryption_scanner.go

@@ -42,6 +42,8 @@ const scanMaxBytes uint64 = 64 << 20 // 64 MiB
 // operator's primary problem; the scanner just reports honestly.
 type encryptionScanner struct {
 	storage *etcdraft.MemoryStorage
+	cipher  *encryption.Cipher
+	cutover RaftCutoverIndex
 }
 
 // HasEncryptionRelevantEntryInRange satisfies the
@@ -101,7 +103,11 @@ func (s *encryptionScanner) HasEncryptionRelevantEntryInRange(startExclusive, en
 				cursor, hi))
 		}
 		for _, ent := range batch {
-			if isEncryptionRelevantEntry(ent) {
+			hit, err := s.isEncryptionRelevantEntry(ent)
+			if err != nil {
+				return false, err
+			}
+			if hit {
 				return true, nil
 			}
 		}
@@ -128,15 +134,25 @@ func (s *encryptionScanner) HasEncryptionRelevantEntryInRange(startExclusive, en
 // version 0x01 — NEVER the FSM opcode. A scanner that misreads
 // the layout would return false for every entry in a real Raft
 // log, silently defeating the §9.1 ErrSidecarBehindRaftLog guard.
-func isEncryptionRelevantEntry(ent *raftpb.Entry) bool {
+func (s *encryptionScanner) isEncryptionRelevantEntry(ent *raftpb.Entry) (bool, error) {
 	if ent.GetType() != raftpb.EntryNormal {
-		return false
+		return false, nil
 	}
 	_, payload, ok := decodeProposalEnvelope(ent.GetData())
 	if !ok || len(payload) == 0 {
-		return false
+		return false, nil
+	}
+	if s.cutover != nil && ent.GetIndex() > s.cutover() {
+		if s.cipher == nil {
+			return false, errors.Wrap(ErrRaftUnwrapFailed, "encryption scanner: cipher missing after cutover")
+		}
+		var err error
+		payload, err = unwrapRaftPayload(s.cipher, payload)
+		if err != nil {
+			return false, errors.Wrap(err, "encryption scanner: unwrap retained entry")
+		}
 	}
-	return encryption.IsEncryptionRelevantOpcode(payload[0])
+	return len(payload) > 0 && encryption.IsEncryptionRelevantOpcode(payload[0]), nil
 }
 
 // EncryptionScanner returns the engine's
@@ -161,5 +177,5 @@ func (e *Engine) EncryptionScanner() encryption.EncryptionRelevantScanner {
 	if e == nil {
 		return &encryptionScanner{}
 	}
-	return &encryptionScanner{storage: e.storage}
+	return &encryptionScanner{storage: e.storage, cipher: e.raftCipher, cutover: e.raftCutoverIndex}
 }