From 76409d940b97dd0e291a64beb6d84102b420df94 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Fri, 23 Oct 2026 00:41:01 +0900
Subject: [PATCH 1/6] fix: verify encryption snapshot checkpoints before gap
 scans

---
 docs/encryption_compacted_restart.md          |  74 +++++++++++
 internal/encryption/snapshot_checkpoint.go    |  46 +++++++
 .../etcd/encryption_checkpoint_test.go        |  73 +++++++++++
 .../raftengine/etcd/encryption_scanner.go     |  28 ++++-
 .../raftengine/etcd/encryption_snapshot.go    |  42 +++++++
 internal/raftengine/etcd/wal_store.go         |  10 +-
 kv/fsm.go                                     |  18 +--
 kv/fsm_encryption.go                          |   6 +
 kv/snapshot.go                                |  55 +++++---
 kv/snapshot_encryption_checkpoint.go          |  64 ++++++++++
 kv/snapshot_encryption_checkpoint_test.go     |  79 ++++++++++++
 main_encryption_compacted_restart_test.go     | 118 ++++++++++++++++++
 main_encryption_snapshot_checkpoint.go        |  55 ++++++++
 main_encryption_startup_guard.go              |   6 +-
 14 files changed, 640 insertions(+), 34 deletions(-)
 create mode 100644 docs/encryption_compacted_restart.md
 create mode 100644 internal/encryption/snapshot_checkpoint.go
 create mode 100644 internal/raftengine/etcd/encryption_checkpoint_test.go
 create mode 100644 internal/raftengine/etcd/encryption_snapshot.go
 create mode 100644 kv/snapshot_encryption_checkpoint.go
 create mode 100644 kv/snapshot_encryption_checkpoint_test.go
 create mode 100644 main_encryption_compacted_restart_test.go
 create mode 100644 main_encryption_snapshot_checkpoint.go

diff --git a/docs/encryption_compacted_restart.md b/docs/encryption_compacted_restart.md
new file mode 100644
index 000000000..d8eec1d54
--- /dev/null
+++ b/docs/encryption_compacted_restart.md
@@ -0,0 +1,74 @@
+# 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 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
+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".
+
+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
+
+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.
+- `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.
diff --git a/internal/encryption/snapshot_checkpoint.go b/internal/encryption/snapshot_checkpoint.go
new file mode 100644
index 000000000..8c235addf
--- /dev/null
+++ b/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)
+}
diff --git a/internal/raftengine/etcd/encryption_checkpoint_test.go b/internal/raftengine/etcd/encryption_checkpoint_test.go
new file mode 100644
index 000000000..b9da7fa6e
--- /dev/null
+++ b/internal/raftengine/etcd/encryption_checkpoint_test.go
@@ -0,0 +1,73 @@
+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()
+			got, err := io.ReadAll(reader)
+			require.NoError(t, err)
+			require.Equal(t, payload, got)
+			require.Equal(t, uint64(100), gotIndex)
+		})
+	}
+}
diff --git a/internal/raftengine/etcd/encryption_scanner.go b/internal/raftengine/etcd/encryption_scanner.go
index 47c3e0070..4d4f90e3c 100644
--- a/internal/raftengine/etcd/encryption_scanner.go
+++ b/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}
 }
diff --git a/internal/raftengine/etcd/encryption_snapshot.go b/internal/raftengine/etcd/encryption_snapshot.go
new file mode 100644
index 000000000..8a1c3134e
--- /dev/null
+++ b/internal/raftengine/etcd/encryption_snapshot.go
@@ -0,0 +1,42 @@
+package etcd
+
+import (
+	"bytes"
+	"io"
+
+	"github.com/cockroachdb/errors"
+	etcdraft "go.etcd.io/raft/v3"
+)
+
+// EncryptionSnapshot returns only the snapshot currently installed in this
+// engine, never an unreferenced/newer file found by directory enumeration.
+// The caller may use its metadata only after the token and full payload agree.
+func (e *Engine) EncryptionSnapshot() (io.ReadCloser, uint64, error) {
+	if e == nil || e.storage == nil {
+		return nil, 0, errors.New("encryption snapshot: engine storage unavailable")
+	}
+	snapshot, err := e.storage.Snapshot()
+	if err != nil {
+		return nil, 0, errors.Wrap(err, "encryption snapshot: read installed snapshot")
+	}
+	if etcdraft.IsEmptySnap(snapshot) || len(snapshot.Data) == 0 {
+		return nil, 0, nil
+	}
+	index := snapshot.GetMetadata().GetIndex()
+	if !isSnapshotToken(snapshot.Data) {
+		return io.NopCloser(bytes.NewReader(snapshot.Data)), index, nil
+	}
+	tok, err := decodeSnapshotToken(snapshot.Data)
+	if err != nil {
+		return nil, 0, err
+	}
+	if tok.Index != index {
+		return nil, 0, errors.Wrap(ErrFSMSnapshotTokenInvalid, "encryption snapshot: token/metadata index mismatch")
+	}
+	path := fsmSnapPath(e.fsmSnapDir, index)
+	if err := verifyFSMSnapshotFileWithToken(path, tok.CRC32C, true); err != nil {
+		return nil, 0, err
+	}
+	payload, err := openFSMSnapshotPayloadReader(path)
+	return payload, index, err
+}
diff --git a/internal/raftengine/etcd/wal_store.go b/internal/raftengine/etcd/wal_store.go
index 32c7cc88f..876cc6a1e 100644
--- a/internal/raftengine/etcd/wal_store.go
+++ b/internal/raftengine/etcd/wal_store.go
@@ -542,7 +542,16 @@ func applyHeaderStateOnSkip(fsm StateMachine, snapPath string, tokenCRC uint32)
 		return errors.Wrapf(ErrFSMSnapshotFileCRC,
 			"path=%s footer=%08x computed=%08x", snapPath, footer, computed)
 	}
+	return applyVerifiedSnapshotHeader(fsm, file, info.Size()-fsmFooterSize)
+}
 
+func applyVerifiedSnapshotHeader(fsm StateMachine, file *os.File, payloadSize int64) error {
+	if setter, ok := fsm.(interface{ RestoreSnapshotHeader(io.Reader) error }); ok {
+		if _, err := file.Seek(0, io.SeekStart); err != nil {
+			return errors.WithStack(err)
+		}
+		return errors.Wrap(setter.RestoreSnapshotHeader(io.LimitReader(file, payloadSize)), "restore verified snapshot header")
+	}
 	setter, hasSetter := fsm.(raftengine.SnapshotHeaderApplier)
 	if !hasSetter {
 		return nil
@@ -551,7 +560,6 @@ func applyHeaderStateOnSkip(fsm StateMachine, snapPath string, tokenCRC uint32)
 	if _, err := file.Seek(0, io.SeekStart); err != nil {
 		return errors.WithStack(err)
 	}
-	payloadSize := info.Size() - fsmFooterSize
 	ceiling, cutover, err := setter.ParseSnapshotHeader(io.LimitReader(file, payloadSize))
 	if err != nil {
 		return errors.WithStack(err)
diff --git a/kv/fsm.go b/kv/fsm.go
index 15eeaffc2..f07c7144a 100644
--- a/kv/fsm.go
+++ b/kv/fsm.go
@@ -11,6 +11,7 @@ import (
 	"sync/atomic"
 
 	"github.com/bootjp/elastickv/distribution"
+	"github.com/bootjp/elastickv/internal/encryption"
 	"github.com/bootjp/elastickv/internal/encryption/fsmwire"
 	"github.com/bootjp/elastickv/internal/fskeys"
 	"github.com/bootjp/elastickv/internal/raftengine"
@@ -70,7 +71,8 @@ type kvFSM struct {
 	// cutoverSource provides the writer-side view of the Phase-2
 	// envelope cutover index for snapshot v1/v2 selection (Stage
 	// 8a §3.3). nil = always v1 output.
-	cutoverSource CutoverSource
+	cutoverSource        CutoverSource
+	encryptionCheckpoint encryption.SnapshotCheckpoint
 	// snapLatch enforces the §3.3 no-downgrade invariant: once a
 	// non-zero cutover is observed, all subsequent snapshots from
 	// this FSM emit v2 even if the sidecar later reads 0.
@@ -1025,13 +1027,14 @@ func (f *kvFSM) Snapshot() (raftengine.Snapshot, error) {
 		cutover = f.cutoverSource.RaftEnvelopeCutoverIndex()
 	}
 	emitted := f.snapLatch.observe(cutover)
-	useV2 := emitted != 0 || f.snapLatch.latched()
+	useV2 := emitted != 0 || f.snapLatch.latched() || f.encryptionCheckpoint.Index != 0
 
 	return &kvFSMSnapshot{
-		snapshot:  snapshot,
-		ceilingMs: hlcCeilingFromHLC(f.hlc),
-		cutover:   emitted,
-		useV2:     useV2,
+		snapshot:   snapshot,
+		ceilingMs:  hlcCeilingFromHLC(f.hlc),
+		cutover:    emitted,
+		useV2:      useV2,
+		checkpoint: f.encryptionCheckpoint,
 	}, nil
 }
 
@@ -1047,7 +1050,7 @@ func (f *kvFSM) Restore(r io.Reader) error {
 	if !ok {
 		br = bufio.NewReader(r)
 	}
-	ceilingU, cutover, err := ReadSnapshotHeader(br)
+	ceilingU, cutover, checkpoint, err := ReadSnapshotHeaderWithCheckpoint(br)
 	if err != nil {
 		return errors.WithStack(err)
 	}
@@ -1061,6 +1064,7 @@ func (f *kvFSM) Restore(r io.Reader) error {
 	if err := f.store.Restore(io.NopCloser(br)); err != nil {
 		return errors.WithStack(err)
 	}
+	f.encryptionCheckpoint = checkpoint
 	// Volatile pins do not travel in a snapshot, and the entries that would
 	// have released them are compacted away by the time this replica installs
 	// one. A pin this node applied before falling behind would otherwise keep
diff --git a/kv/fsm_encryption.go b/kv/fsm_encryption.go
index 7a3c4f5a5..b498cb9df 100644
--- a/kv/fsm_encryption.go
+++ b/kv/fsm_encryption.go
@@ -122,6 +122,9 @@ func (f *kvFSM) applyBootstrap(raftIdx uint64, data []byte) any {
 		return haltErr(errors.Wrap(errors.Mark(err, encryption.ErrEncryptionApply),
 			"kv/fsm: apply bootstrap"))
 	}
+	if err := f.captureEncryptionCheckpoint(raftIdx); err != nil {
+		return haltErr(errors.Mark(err, encryption.ErrEncryptionApply))
+	}
 	return nil
 }
 
@@ -137,6 +140,9 @@ func (f *kvFSM) applyRotation(raftIdx uint64, data []byte) any {
 		return haltErr(errors.Wrap(errors.Mark(err, encryption.ErrEncryptionApply),
 			"kv/fsm: apply rotation"))
 	}
+	if err := f.captureEncryptionCheckpoint(raftIdx); err != nil {
+		return haltErr(errors.Mark(err, encryption.ErrEncryptionApply))
+	}
 	return nil
 }
 
diff --git a/kv/snapshot.go b/kv/snapshot.go
index b88c6eb6a..2a539b1fc 100644
--- a/kv/snapshot.go
+++ b/kv/snapshot.go
@@ -8,6 +8,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/bootjp/elastickv/internal/encryption"
 	"github.com/bootjp/elastickv/internal/raftengine"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
@@ -109,9 +110,10 @@ type kvFSMSnapshot struct {
 	// useV2 records the v1/v2 selection made at construction time so a
 	// concurrent sidecar mutation after Snapshot() returns cannot flip
 	// the format mid-write.
-	useV2 bool
-	once  sync.Once
-	err   error
+	useV2      bool
+	checkpoint encryption.SnapshotCheckpoint
+	once       sync.Once
+	err        error
 }
 
 func (f *kvFSMSnapshot) WriteTo(w io.Writer) (int64, error) {
@@ -133,8 +135,8 @@ func (f *kvFSMSnapshot) WriteTo(w io.Writer) (int64, error) {
 // and returns the populated prefix length. v2 emits the minimum payload
 // (ceiling+cutover) — forward-compat trailing bytes are a reader-side
 // concern; the writer never emits unused trailing slots.
-func (f *kvFSMSnapshot) encodeHeader() ([hlcSnapshotV2HeaderLen]byte, int) {
-	var hdr [hlcSnapshotV2HeaderLen]byte
+func (f *kvFSMSnapshot) encodeHeader() ([snapshotCheckpointHeaderLen]byte, int) {
+	var hdr [snapshotCheckpointHeaderLen]byte
 	if !f.useV2 {
 		copy(hdr[:8], hlcSnapshotMagic[:])
 		binary.BigEndian.PutUint64(hdr[8:16], uint64(f.ceilingMs)) //nolint:gosec // ceiling is a Unix ms timestamp, always positive
@@ -144,6 +146,13 @@ func (f *kvFSMSnapshot) encodeHeader() ([hlcSnapshotV2HeaderLen]byte, int) {
 	binary.BigEndian.PutUint16(hdr[8:10], uint16(hlcSnapshotV2MinPayload))
 	binary.BigEndian.PutUint64(hdr[10:18], uint64(f.ceilingMs)) //nolint:gosec // same as v1
 	binary.BigEndian.PutUint64(hdr[18:26], f.cutover)
+	if f.checkpoint.Index != 0 {
+		binary.BigEndian.PutUint16(hdr[8:10], uint16(hlcSnapshotV2MinPayload+encryptionCheckpointBytes))
+		copy(hdr[26:34], encryptionCheckpointMagic)
+		binary.BigEndian.PutUint64(hdr[34:42], f.checkpoint.Index)
+		copy(hdr[42:74], f.checkpoint.State[:])
+		return hdr, snapshotCheckpointHeaderLen
+	}
 	return hdr, hlcSnapshotV2HeaderLen
 }
 
@@ -185,6 +194,11 @@ func (f *kvFSMSnapshot) closeSnapshot() error {
 //   - Anything else (including streams shorter than 8 bytes): (0, 0, nil)
 //     and ALL bytes left in the *bufio.Reader for the inner-store path.
 func ReadSnapshotHeader(r *bufio.Reader) (ceiling, cutover uint64, err error) {
+	ceiling, cutover, _, err = ReadSnapshotHeaderWithCheckpoint(r)
+	return ceiling, cutover, err
+}
+
+func ReadSnapshotHeaderWithCheckpoint(r *bufio.Reader) (ceiling, cutover uint64, checkpoint encryption.SnapshotCheckpoint, err error) {
 	// Step 0: peek up to 8 bytes. Short streams (< 8 bytes) fall through
 	// to the headerless-legacy branch with the partial bytes left in
 	// place — this preserves TestFSMSnapshotRestoreSmallLegacy and is
@@ -197,25 +211,26 @@ func ReadSnapshotHeader(r *bufio.Reader) (ceiling, cutover uint64, err error) {
 	peeked, perr := r.Peek(8) //nolint:mnd
 	if len(peeked) < 8 {      //nolint:mnd
 		if perr != nil && !errors.Is(perr, io.EOF) && !errors.Is(perr, io.ErrUnexpectedEOF) {
-			return 0, 0, errors.Wrap(perr, "snapshot header: peek")
+			return 0, 0, checkpoint, errors.Wrap(perr, "snapshot header: peek")
 		}
 		// io.EOF / io.ErrUnexpectedEOF on a short stream: treat as
 		// headerless legacy. Bytes that were peeked stay in r per
 		// bufio.Reader semantics.
-		return 0, 0, nil
+		return 0, 0, checkpoint, nil
 	}
 
 	switch {
 	case isV2Magic(peeked):
 		return readV2(r)
 	case isV1Magic(peeked):
-		return readV1(r)
+		ceiling, err = readV1(r)
+		return ceiling, 0, checkpoint, err
 	case isUnknownEKVTHLC(peeked):
-		return 0, 0, ErrSnapshotHeaderUnknownMagic
+		return 0, 0, checkpoint, ErrSnapshotHeaderUnknownMagic
 	default:
 		// Headerless legacy: leave the peeked bytes in r so the
 		// inner-store path sees the payload from byte 0.
-		return 0, 0, nil
+		return 0, 0, checkpoint, nil
 	}
 }
 
@@ -245,28 +260,29 @@ func isUnknownEKVTHLC(p []byte) bool {
 		p[6] == hlcSnapshotMagicPrefix[6]
 }
 
-func readV1(r *bufio.Reader) (uint64, uint64, error) {
+func readV1(r *bufio.Reader) (uint64, error) {
 	var rest [16]byte //nolint:mnd
 	if _, err := io.ReadFull(r, rest[:]); err != nil {
-		return 0, 0, errors.Wrap(err, "v1 header: read magic+ceiling")
+		return 0, errors.Wrap(err, "v1 header: read magic+ceiling")
 	}
 	ceiling := binary.BigEndian.Uint64(rest[8:16])
-	return ceiling, 0, nil
+	return ceiling, nil
 }
 
-func readV2(r *bufio.Reader) (uint64, uint64, error) {
+func readV2(r *bufio.Reader) (uint64, uint64, encryption.SnapshotCheckpoint, error) {
+	var checkpoint encryption.SnapshotCheckpoint
 	// Consume the 8 magic bytes.
 	if _, err := r.Discard(8); err != nil { //nolint:mnd
-		return 0, 0, errors.Wrap(err, "v2 header: discard magic")
+		return 0, 0, checkpoint, errors.Wrap(err, "v2 header: discard magic")
 	}
 	// Read the 2-byte length prefix.
 	var lenBuf [hlcSnapshotV2LenBytes]byte
 	if _, err := io.ReadFull(r, lenBuf[:]); err != nil {
-		return 0, 0, errors.Wrap(err, "v2 header: read length prefix")
+		return 0, 0, checkpoint, errors.Wrap(err, "v2 header: read length prefix")
 	}
 	plen := int(binary.BigEndian.Uint16(lenBuf[:]))
 	if plen < hlcSnapshotV2MinPayload || plen > maxSnapshotHeaderPayload {
-		return 0, 0, ErrSnapshotHeaderInvalidLength
+		return 0, 0, checkpoint, ErrSnapshotHeaderInvalidLength
 	}
 	// Read exactly plen payload bytes. Parse ceiling/cutover from the
 	// first 16; ignore the remainder (forward-compat extension area).
@@ -276,9 +292,10 @@ func readV2(r *bufio.Reader) (uint64, uint64, error) {
 	var buf [maxSnapshotHeaderPayload]byte
 	payload := buf[:plen]
 	if _, err := io.ReadFull(r, payload); err != nil {
-		return 0, 0, errors.Wrap(err, "v2 header: read payload")
+		return 0, 0, checkpoint, errors.Wrap(err, "v2 header: read payload")
 	}
 	ceiling := binary.BigEndian.Uint64(payload[0:8])
 	cutover := binary.BigEndian.Uint64(payload[8:16])
-	return ceiling, cutover, nil
+	checkpoint, err := decodeSnapshotCheckpoint(payload)
+	return ceiling, cutover, checkpoint, err
 }
diff --git a/kv/snapshot_encryption_checkpoint.go b/kv/snapshot_encryption_checkpoint.go
new file mode 100644
index 000000000..1e9f99adc
--- /dev/null
+++ b/kv/snapshot_encryption_checkpoint.go
@@ -0,0 +1,64 @@
+package kv
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"io"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/cockroachdb/errors"
+)
+
+var encryptionCheckpointMagic = []byte("EKVENCP1")
+
+const encryptionCheckpointBytes = 8 + 8 + 32
+const snapshotCheckpointHeaderLen = hlcSnapshotV2HeaderLen + encryptionCheckpointBytes
+
+func decodeSnapshotCheckpoint(payload []byte) (encryption.SnapshotCheckpoint, error) {
+	var checkpoint encryption.SnapshotCheckpoint
+	if len(payload) < hlcSnapshotV2MinPayload+len(encryptionCheckpointMagic) ||
+		!bytes.Equal(payload[hlcSnapshotV2MinPayload:hlcSnapshotV2MinPayload+len(encryptionCheckpointMagic)], encryptionCheckpointMagic) {
+		return checkpoint, nil
+	}
+	if len(payload) != hlcSnapshotV2MinPayload+encryptionCheckpointBytes {
+		return checkpoint, errors.New("snapshot encryption checkpoint: invalid length")
+	}
+	checkpoint.Index = binary.BigEndian.Uint64(payload[24:32])
+	copy(checkpoint.State[:], payload[32:64])
+	if checkpoint.Index == 0 {
+		return checkpoint, errors.New("snapshot encryption checkpoint: zero index")
+	}
+	return checkpoint, nil
+}
+
+func (f *kvFSM) captureEncryptionCheckpoint(index uint64) error {
+	source, ok := f.encryption.(interface {
+		SnapshotCheckpoint() (encryption.SnapshotCheckpoint, error)
+	})
+	if !ok || index == 0 {
+		return nil
+	}
+	checkpoint, err := source.SnapshotCheckpoint()
+	if err != nil {
+		return errors.Wrap(err, "capture encryption snapshot checkpoint")
+	}
+	// During WAL replay the sidecar can already represent a later entry.
+	// Do not claim that future state in a snapshot of an earlier prefix.
+	if checkpoint.Index == index {
+		f.encryptionCheckpoint = checkpoint
+	}
+	return nil
+}
+
+// RestoreSnapshotHeader is used only after the engine has verified the whole
+// physical snapshot against its Raft token. Parse errors leave state untouched.
+func (f *kvFSM) RestoreSnapshotHeader(r io.Reader) error {
+	ceiling, cutover, checkpoint, err := ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(r))
+	if err != nil {
+		return err
+	}
+	f.ApplySnapshotHeader(ceiling, cutover)
+	f.encryptionCheckpoint = checkpoint
+	return nil
+}
diff --git a/kv/snapshot_encryption_checkpoint_test.go b/kv/snapshot_encryption_checkpoint_test.go
new file mode 100644
index 000000000..00d11e3ff
--- /dev/null
+++ b/kv/snapshot_encryption_checkpoint_test.go
@@ -0,0 +1,79 @@
+package kv
+
+import (
+	"bufio"
+	"bytes"
+	"testing"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/bootjp/elastickv/store"
+	"github.com/stretchr/testify/require"
+)
+
+type checkpointApplier struct {
+	fakeApplier
+	checkpoint encryption.SnapshotCheckpoint
+}
+
+func (a *checkpointApplier) SnapshotCheckpoint() (encryption.SnapshotCheckpoint, error) {
+	return a.checkpoint, nil
+}
+
+func TestEncryptionSnapshotCheckpointRoundTrip(t *testing.T) {
+	cp := encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{1, 2, 3}}
+	for _, headerOnly := range []bool{false, true} {
+		t.Run(map[bool]string{false: "full-physical-restore", true: "skip-gate-header-restore"}[headerOnly], func(t *testing.T) {
+			source, ok := NewKvFSMWithHLC(store.NewMVCCStore(), NewHLC()).(*kvFSM)
+			require.True(t, ok)
+			source.encryption = &checkpointApplier{checkpoint: cp}
+			require.NoError(t, source.captureEncryptionCheckpoint(6))
+			snapshot, err := source.Snapshot()
+			require.NoError(t, err)
+			defer snapshot.Close()
+			var payload bytes.Buffer
+			_, err = snapshot.WriteTo(&payload)
+			require.NoError(t, err)
+			_, _, parsed, err := ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(bytes.NewReader(payload.Bytes())))
+			require.NoError(t, err)
+			require.Equal(t, cp, parsed)
+			target, ok := NewKvFSMWithHLC(store.NewMVCCStore(), NewHLC()).(*kvFSM)
+			require.True(t, ok)
+			if headerOnly {
+				err = target.RestoreSnapshotHeader(bytes.NewReader(payload.Bytes()))
+			} else {
+				err = target.Restore(bytes.NewReader(payload.Bytes()))
+			}
+			require.NoError(t, err)
+			require.Equal(t, cp, target.encryptionCheckpoint)
+			// A rollback on disk must not overwrite the restored witness merely
+			// because this process starts making newer snapshots.
+			target.encryption = &checkpointApplier{checkpoint: encryption.SnapshotCheckpoint{Index: 5}}
+			require.NoError(t, target.captureEncryptionCheckpoint(4))
+			require.Equal(t, cp, target.encryptionCheckpoint)
+			next, err := target.Snapshot()
+			require.NoError(t, err)
+			defer next.Close()
+			var nextPayload bytes.Buffer
+			_, err = next.WriteTo(&nextPayload)
+			require.NoError(t, err)
+			_, _, carried, err := ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(&nextPayload))
+			require.NoError(t, err)
+			require.Equal(t, cp, carried)
+		})
+	}
+}
+
+func TestEncryptionSnapshotCheckpointRejectsTruncationWithoutMutation(t *testing.T) {
+	fsm, ok := NewKvFSMWithHLC(store.NewMVCCStore(), NewHLC()).(*kvFSM)
+	require.True(t, ok)
+	fsm.encryptionCheckpoint = encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{1}}
+	snapshot, err := fsm.Snapshot()
+	require.NoError(t, err)
+	defer snapshot.Close()
+	var payload bytes.Buffer
+	_, err = snapshot.WriteTo(&payload)
+	require.NoError(t, err)
+	before := fsm.encryptionCheckpoint
+	require.Error(t, fsm.RestoreSnapshotHeader(bytes.NewReader(payload.Bytes()[:snapshotCheckpointHeaderLen-1])))
+	require.Equal(t, before, fsm.encryptionCheckpoint)
+}
diff --git a/main_encryption_compacted_restart_test.go b/main_encryption_compacted_restart_test.go
new file mode 100644
index 000000000..8fadfb5cf
--- /dev/null
+++ b/main_encryption_compacted_restart_test.go
@@ -0,0 +1,118 @@
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"io"
+	"path/filepath"
+	"testing"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/bootjp/elastickv/kv"
+	"github.com/bootjp/elastickv/store"
+	"github.com/stretchr/testify/require"
+	etcdraft "go.etcd.io/raft/v3"
+)
+
+type checkpointGapEngine struct {
+	stubGapEngine
+	payload       []byte
+	snapshotIndex uint64
+	snapshotErr   error
+}
+
+func (e *checkpointGapEngine) EncryptionSnapshot() (io.ReadCloser, uint64, error) {
+	return io.NopCloser(bytes.NewReader(e.payload)), e.snapshotIndex, e.snapshotErr
+}
+
+type checkpointScanner struct {
+	t          *testing.T
+	start, end uint64
+	hit        bool
+}
+
+func (s *checkpointScanner) HasEncryptionRelevantEntryInRange(start, end uint64) (bool, error) {
+	if start < s.start {
+		return false, etcdraft.ErrCompacted
+	}
+	require.Equal(s.t, s.start, start)
+	require.Equal(s.t, s.end, end)
+	return s.hit, nil
+}
+
+func checkpointHeader(t *testing.T, sc *encryption.Sidecar) []byte {
+	t.Helper()
+	cp, err := encryption.SidecarCheckpoint(sc)
+	require.NoError(t, err)
+	var out bytes.Buffer
+	out.WriteString("EKVTHLC2")
+	require.NoError(t, binary.Write(&out, binary.BigEndian, uint16(64)))
+	require.NoError(t, binary.Write(&out, binary.BigEndian, uint64(1234)))
+	require.NoError(t, binary.Write(&out, binary.BigEndian, sc.RaftEnvelopeCutoverIndex))
+	out.WriteString("EKVENCP1")
+	require.NoError(t, binary.Write(&out, binary.BigEndian, cp.Index))
+	out.Write(cp.State[:])
+	return out.Bytes()
+}
+
+func TestEncryptionCompactedRestartCheckpoint(t *testing.T) {
+	for _, variant := range []string{"valid", "stale-sidecar", "mismatched-keys", "future-checkpoint", "missing-proof", "retained-rotation", "snapshot-verification-error", "truncated-proof"} {
+		t.Run(variant, func(t *testing.T) {
+			sc := &encryption.Sidecar{Version: encryption.SidecarVersion, RaftAppliedIndex: 6,
+				StorageEnvelopeActive: true, StorageEnvelopeCutoverIndex: 5, RaftEnvelopeCutoverIndex: 6,
+				Active: encryption.ActiveKeys{Storage: 1, Raft: 2},
+				Keys:   map[string]encryption.SidecarKey{"1": {Purpose: "storage", Wrapped: []byte("wrapped-storage"), LocalEpoch: 8}, "2": {Purpose: "raft", Wrapped: []byte("wrapped-raft")}}}
+			payload := checkpointHeader(t, sc)
+			engine := &checkpointGapEngine{stubGapEngine: stubGapEngine{appliedIndex: 877,
+				scanner: &checkpointScanner{t: t, start: 860, end: 877}}, payload: payload, snapshotIndex: 860}
+			switch variant {
+			case "stale-sidecar":
+				sc.RaftAppliedIndex = 5
+			case "mismatched-keys":
+				key := sc.Keys["1"]
+				key.Wrapped = []byte("different-wrapped-key")
+				sc.Keys["1"] = key
+			case "future-checkpoint":
+				binary.BigEndian.PutUint64(engine.payload[34:42], 861)
+			case "missing-proof":
+				engine.payload = []byte("EKVTHLC1\x00\x00\x00\x00\x00\x00\x00\x01")
+			case "retained-rotation":
+				scanner, ok := engine.scanner.(*checkpointScanner)
+				require.True(t, ok)
+				scanner.hit = true
+			case "snapshot-verification-error":
+				engine.snapshotErr = io.ErrUnexpectedEOF
+			case "truncated-proof":
+				engine.payload = payload[:len(payload)-1]
+			}
+			path := filepath.Join(t.TempDir(), "keys.json")
+			require.NoError(t, encryption.WriteSidecar(path, sc))
+			err := runSidecarBehindRaftLogGuard(engine, path, 1)
+			if variant == "valid" {
+				require.NoError(t, err, "verified snapshot witnesses the compacted prefix; retained tail still scanned")
+			} else {
+				require.Error(t, err, "uncertain or inconsistent metadata must refuse startup")
+			}
+		})
+	}
+}
+
+func TestEncryptionCheckpointDoesNotAuthorizeEpochRollback(t *testing.T) {
+	sc := &encryption.Sidecar{RaftAppliedIndex: 6, Keys: map[string]encryption.SidecarKey{"1": {LocalEpoch: 8}}}
+	payload := checkpointHeader(t, sc)
+	_, _, before, err := kv.ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(bytes.NewReader(payload)))
+	require.NoError(t, err)
+	key := sc.Keys["1"]
+	key.LocalEpoch = 1
+	sc.Keys["1"] = key
+	after, err := encryption.SidecarCheckpoint(sc)
+	require.NoError(t, err)
+	require.Equal(t, before, after, "snapshot state is replica-independent, not an epoch authorization")
+	st := newRegistrationTestStore(t)
+	preSeedRegistryRow(t, st, 1, 1, 8)
+	reg, err := store.WriterRegistryFor(st)
+	require.NoError(t, err)
+	err = checkLocalEpochRollbackBeforeBump(reg, 1, 1, key.LocalEpoch, true)
+	require.ErrorIs(t, err, encryption.ErrLocalEpochRollback)
+}
diff --git a/main_encryption_snapshot_checkpoint.go b/main_encryption_snapshot_checkpoint.go
new file mode 100644
index 000000000..3ba93d603
--- /dev/null
+++ b/main_encryption_snapshot_checkpoint.go
@@ -0,0 +1,55 @@
+package main
+
+import (
+	"bufio"
+	"io"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/bootjp/elastickv/kv"
+	"github.com/cockroachdb/errors"
+)
+
+func sidecarGapStartFromSnapshot(engine encryptionGapEngine, sc *encryption.Sidecar) (uint64, error) {
+	start := sc.RaftAppliedIndex
+	source, ok := engine.(interface {
+		EncryptionSnapshot() (io.ReadCloser, uint64, error)
+	})
+	if !ok {
+		return start, nil
+	}
+	payload, index, err := source.EncryptionSnapshot()
+	if err != nil {
+		return 0, errors.Wrap(err, "encryption startup guard: verify snapshot checkpoint")
+	}
+	if payload == nil {
+		return start, nil
+	}
+	defer payload.Close()
+	if index <= start {
+		return start, nil
+	}
+	if index > engine.AppliedIndex() {
+		return 0, errors.Wrap(encryption.ErrSidecarBehindRaftLog, "snapshot checkpoint exceeds applied index")
+	}
+	return verifySidecarSnapshotCheckpoint(payload, index, sc)
+}
+
+func verifySidecarSnapshotCheckpoint(payload io.Reader, index uint64, sc *encryption.Sidecar) (uint64, error) {
+	_, cutover, checkpoint, err := kv.ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(payload))
+	if err != nil {
+		return 0, errors.Wrap(err, "encryption startup guard: read snapshot checkpoint")
+	}
+	if checkpoint.Index == 0 {
+		// Old snapshots carry no proof about encryption changes in the
+		// compacted prefix. Keep the original complete-scan requirement.
+		return sc.RaftAppliedIndex, nil
+	}
+	want, err := encryption.SidecarCheckpoint(sc)
+	if err != nil {
+		return 0, errors.WithStack(err)
+	}
+	if checkpoint.Index > index || checkpoint != want || cutover != sc.RaftEnvelopeCutoverIndex {
+		return 0, errors.Wrap(encryption.ErrSidecarBehindRaftLog, "snapshot checkpoint disagrees with sidecar")
+	}
+	return index, nil
+}
diff --git a/main_encryption_startup_guard.go b/main_encryption_startup_guard.go
index 13c62aeb3..c51aeb268 100644
--- a/main_encryption_startup_guard.go
+++ b/main_encryption_startup_guard.go
@@ -184,8 +184,12 @@ func runSidecarBehindRaftLogGuard(
 			slog.Uint64("engine_applied_index", gapEngine.AppliedIndex()))
 		return nil
 	}
+	start, err := sidecarGapStartFromSnapshot(gapEngine, sidecar)
+	if err != nil {
+		return err
+	}
 	if err := encryption.GuardSidecarBehindRaftLog(
-		sidecar.RaftAppliedIndex,
+		start,
 		gapEngine.AppliedIndex(),
 		gapEngine.EncryptionScanner(),
 	); err != nil {

From 753e121dab65f2fb0597314c1f5fbca6e05dd902 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Fri, 23 Oct 2026 00:59:38 +0900
Subject: [PATCH 2/6] fix: retain durable encryption checkpoints across
 cold-start skips

---
 docs/encryption_compacted_restart.md          | 15 ++++
 internal/backup/snapshot_reader.go            |  1 +
 .../etcd/encryption_checkpoint_test.go        |  1 +
 .../raftengine/etcd/encryption_snapshot.go    | 80 ++++++++++++++++---
 kv/fsm.go                                     |  7 ++
 kv/snapshot_encryption_checkpoint.go          | 60 ++++++++++++++
 kv/snapshot_encryption_checkpoint_test.go     | 58 ++++++++++++++
 store/encryption_checkpoint.go                | 54 +++++++++++++
 store/encryption_checkpoint_test.go           | 55 +++++++++++++
 store/lsm_store.go                            |  1 +
 10 files changed, 320 insertions(+), 12 deletions(-)
 create mode 100644 store/encryption_checkpoint.go
 create mode 100644 store/encryption_checkpoint_test.go

diff --git a/docs/encryption_compacted_restart.md b/docs/encryption_compacted_restart.md
index d8eec1d54..12147f8c1 100644
--- a/docs/encryption_compacted_restart.md
+++ b/docs/encryption_compacted_restart.md
@@ -25,10 +25,25 @@ 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".
diff --git a/internal/backup/snapshot_reader.go b/internal/backup/snapshot_reader.go
index 653d23bf9..719e0fbee 100644
--- a/internal/backup/snapshot_reader.go
+++ b/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"),
diff --git a/internal/raftengine/etcd/encryption_checkpoint_test.go b/internal/raftengine/etcd/encryption_checkpoint_test.go
index b9da7fa6e..15b125f53 100644
--- a/internal/raftengine/etcd/encryption_checkpoint_test.go
+++ b/internal/raftengine/etcd/encryption_checkpoint_test.go
@@ -64,6 +64,7 @@ func TestEncryptionSnapshotRequiresMatchingPhysicalEnvelope(t *testing.T) {
 			}
 			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)
diff --git a/internal/raftengine/etcd/encryption_snapshot.go b/internal/raftengine/etcd/encryption_snapshot.go
index 8a1c3134e..dc4658d35 100644
--- a/internal/raftengine/etcd/encryption_snapshot.go
+++ b/internal/raftengine/etcd/encryption_snapshot.go
@@ -3,6 +3,7 @@ package etcd
 import (
 	"bytes"
 	"io"
+	"os"
 
 	"github.com/cockroachdb/errors"
 	etcdraft "go.etcd.io/raft/v3"
@@ -15,28 +16,83 @@ func (e *Engine) EncryptionSnapshot() (io.ReadCloser, uint64, error) {
 	if e == nil || e.storage == nil {
 		return nil, 0, errors.New("encryption snapshot: engine storage unavailable")
 	}
+	source, err := e.openEncryptionSnapshotSource()
+	if err != nil {
+		return nil, 0, err
+	}
+	if source.file == nil {
+		if source.index == 0 {
+			return nil, 0, nil
+		}
+		return io.NopCloser(bytes.NewReader(source.legacy)), source.index, nil
+	}
+	file := source.file
+	if err := verifyEncryptionSnapshotDescriptor(file, file.Name(), source.crc); err != nil {
+		_ = file.Close()
+		return nil, 0, err
+	}
+	info, err := file.Stat()
+	if err != nil {
+		_ = file.Close()
+		return nil, 0, errors.WithStack(err)
+	}
+	return &limitedReadCloser{Reader: io.LimitReader(file, info.Size()-fsmFooterSize), Closer: file}, source.index, nil
+}
+
+type encryptionSnapshotSource struct {
+	file   *os.File
+	legacy []byte
+	index  uint64
+	crc    uint32
+}
+
+func (e *Engine) openEncryptionSnapshotSource() (encryptionSnapshotSource, error) {
+	e.snapshotMu.Lock()
+	defer e.snapshotMu.Unlock()
+	var source encryptionSnapshotSource
 	snapshot, err := e.storage.Snapshot()
 	if err != nil {
-		return nil, 0, errors.Wrap(err, "encryption snapshot: read installed snapshot")
+		return source, errors.Wrap(err, "encryption snapshot: read installed snapshot")
 	}
 	if etcdraft.IsEmptySnap(snapshot) || len(snapshot.Data) == 0 {
-		return nil, 0, nil
+		return source, nil
 	}
-	index := snapshot.GetMetadata().GetIndex()
+	source.index = snapshot.GetMetadata().GetIndex()
 	if !isSnapshotToken(snapshot.Data) {
-		return io.NopCloser(bytes.NewReader(snapshot.Data)), index, nil
+		source.legacy = snapshot.Data
+		return source, nil
 	}
 	tok, err := decodeSnapshotToken(snapshot.Data)
 	if err != nil {
-		return nil, 0, err
+		return source, err
 	}
-	if tok.Index != index {
-		return nil, 0, errors.Wrap(ErrFSMSnapshotTokenInvalid, "encryption snapshot: token/metadata index mismatch")
+	if tok.Index != source.index {
+		return source, errors.Wrap(ErrFSMSnapshotTokenInvalid, "encryption snapshot: token/metadata index mismatch")
 	}
-	path := fsmSnapPath(e.fsmSnapDir, index)
-	if err := verifyFSMSnapshotFileWithToken(path, tok.CRC32C, true); err != nil {
-		return nil, 0, err
+	source.crc = tok.CRC32C
+	source.file, err = os.Open(fsmSnapPath(e.fsmSnapDir, source.index))
+	if err != nil {
+		return source, statFSMFileError(err)
+	}
+	return source, nil
+}
+
+func verifyEncryptionSnapshotDescriptor(file *os.File, path string, tokenCRC uint32) error {
+	info, err := file.Stat()
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	footer, err := verifyFSMSnapshotPrefix(file, info.Size(), path, tokenCRC)
+	if err != nil {
+		return err
+	}
+	computed, err := computeFSMSnapshotPayloadCRC(file, info.Size())
+	if err != nil {
+		return err
+	}
+	if computed != footer {
+		return errors.Wrap(ErrFSMSnapshotFileCRC, "encryption snapshot payload checksum mismatch")
 	}
-	payload, err := openFSMSnapshotPayloadReader(path)
-	return payload, index, err
+	_, err = file.Seek(0, io.SeekStart)
+	return errors.WithStack(err)
 }
diff --git a/kv/fsm.go b/kv/fsm.go
index f07c7144a..ee293051d 100644
--- a/kv/fsm.go
+++ b/kv/fsm.go
@@ -1017,6 +1017,9 @@ func (f *kvFSM) Snapshot() (raftengine.Snapshot, error) {
 	if err := f.rejectSnapshotWithActiveBackupPin(); err != nil {
 		return nil, err
 	}
+	if err := f.checkSnapshotCheckpointConsistency(); err != nil {
+		return nil, err
+	}
 	snapshot, err := f.store.Snapshot()
 	if err != nil {
 		return nil, errors.WithStack(err)
@@ -1064,6 +1067,10 @@ func (f *kvFSM) Restore(r io.Reader) error {
 	if err := f.store.Restore(io.NopCloser(br)); err != nil {
 		return errors.WithStack(err)
 	}
+	checkpoint, err = f.checkpointForDurableFSM(checkpoint, false)
+	if err != nil {
+		return err
+	}
 	f.encryptionCheckpoint = checkpoint
 	// Volatile pins do not travel in a snapshot, and the entries that would
 	// have released them are compacted away by the time this replica installs
diff --git a/kv/snapshot_encryption_checkpoint.go b/kv/snapshot_encryption_checkpoint.go
index 1e9f99adc..9a43a87e7 100644
--- a/kv/snapshot_encryption_checkpoint.go
+++ b/kv/snapshot_encryption_checkpoint.go
@@ -46,6 +46,13 @@ func (f *kvFSM) captureEncryptionCheckpoint(index uint64) error {
 	// During WAL replay the sidecar can already represent a later entry.
 	// Do not claim that future state in a snapshot of an earlier prefix.
 	if checkpoint.Index == index {
+		if sink, ok := f.store.(interface {
+			SaveEncryptionSnapshotCheckpoint(encryption.SnapshotCheckpoint) error
+		}); ok {
+			if err := sink.SaveEncryptionSnapshotCheckpoint(checkpoint); err != nil {
+				return errors.Wrap(err, "persist applied encryption checkpoint")
+			}
+		}
 		f.encryptionCheckpoint = checkpoint
 	}
 	return nil
@@ -58,7 +65,60 @@ func (f *kvFSM) RestoreSnapshotHeader(r io.Reader) error {
 	if err != nil {
 		return err
 	}
+	checkpoint, err = f.checkpointForDurableFSM(checkpoint, true)
+	if err != nil {
+		return err
+	}
 	f.ApplySnapshotHeader(ceiling, cutover)
 	f.encryptionCheckpoint = checkpoint
 	return nil
 }
+
+func (f *kvFSM) checkpointForDurableFSM(snapshot encryption.SnapshotCheckpoint, skip bool) (encryption.SnapshotCheckpoint, error) {
+	source, ok := f.store.(interface {
+		LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error)
+	})
+	if !ok {
+		return snapshot, nil
+	}
+	stored, present, err := source.LoadEncryptionSnapshotCheckpoint()
+	if err != nil {
+		return snapshot, errors.Wrap(err, "load durable FSM encryption checkpoint")
+	}
+	if !present {
+		return snapshot, nil
+	}
+	if stored == snapshot {
+		return stored, nil
+	}
+	if !skip || stored.Index <= snapshot.Index {
+		return snapshot, errors.New("encryption checkpoint: physical snapshot header and FSM record disagree")
+	}
+	applied, present, err := f.LastAppliedIndex()
+	if err != nil {
+		return snapshot, err
+	}
+	if present && stored.Index <= applied {
+		return stored, nil
+	}
+	// The durable checkpoint may precede a crash before the applied-index
+	// marker. That tail is not skipped and must replay through the handler.
+	return snapshot, nil
+}
+
+func (f *kvFSM) checkSnapshotCheckpointConsistency() error {
+	source, ok := f.store.(interface {
+		LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error)
+	})
+	if !ok {
+		return nil
+	}
+	stored, present, err := source.LoadEncryptionSnapshotCheckpoint()
+	if err != nil {
+		return errors.Wrap(err, "verify snapshot checkpoint before capture")
+	}
+	if present && stored != f.encryptionCheckpoint {
+		return errors.New("encryption checkpoint: durable tail must reconcile before snapshot capture")
+	}
+	return nil
+}
diff --git a/kv/snapshot_encryption_checkpoint_test.go b/kv/snapshot_encryption_checkpoint_test.go
index 00d11e3ff..9dd518f8e 100644
--- a/kv/snapshot_encryption_checkpoint_test.go
+++ b/kv/snapshot_encryption_checkpoint_test.go
@@ -77,3 +77,61 @@ func TestEncryptionSnapshotCheckpointRejectsTruncationWithoutMutation(t *testing
 	require.Error(t, fsm.RestoreSnapshotHeader(bytes.NewReader(payload.Bytes()[:snapshotCheckpointHeaderLen-1])))
 	require.Equal(t, before, fsm.encryptionCheckpoint)
 }
+
+type durableCheckpointFixture struct {
+	store.MVCCStore
+	checkpoint encryption.SnapshotCheckpoint
+	applied    uint64
+}
+
+func (s *durableCheckpointFixture) LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error) {
+	return s.checkpoint, true, nil
+}
+
+func (s *durableCheckpointFixture) LastAppliedIndex() (uint64, bool, error) {
+	return s.applied, true, nil
+}
+
+func TestEncryptionCheckpointSkipPreservesPostSnapshotRotation(t *testing.T) {
+	old := encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{6}}
+	newer := encryption.SnapshotCheckpoint{Index: 890, State: [32]byte{8}}
+	header := &kvFSMSnapshot{useV2: true, checkpoint: old}
+	encoded, size := header.encodeHeader()
+	for _, applied := range []uint64{900, 889} {
+		t.Run(map[bool]string{true: "durable-tail-skipped", false: "unapplied-tail-must-replay"}[applied == 900], func(t *testing.T) {
+			st := &durableCheckpointFixture{MVCCStore: store.NewMVCCStore(), checkpoint: newer, applied: applied}
+			fsm, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM)
+			require.True(t, ok)
+			require.NoError(t, fsm.RestoreSnapshotHeader(bytes.NewReader(encoded[:size])))
+			want := old
+			if applied == 900 {
+				want = newer
+			}
+			require.Equal(t, want, fsm.encryptionCheckpoint)
+			next, err := fsm.Snapshot()
+			if applied == 889 {
+				require.Error(t, err, "the future durable record must replay before any new snapshot")
+				return
+			}
+			require.NoError(t, err)
+			defer next.Close()
+			var out bytes.Buffer
+			_, err = next.WriteTo(&out)
+			require.NoError(t, err)
+			_, _, cp, err := ReadSnapshotHeaderWithCheckpoint(bufio.NewReader(&out))
+			require.NoError(t, err)
+			require.Equal(t, want, cp)
+		})
+	}
+}
+
+func TestEncryptionCheckpointSkipRejectsChangedDurableRecord(t *testing.T) {
+	old := encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{6}}
+	header := &kvFSMSnapshot{useV2: true, checkpoint: old}
+	encoded, size := header.encodeHeader()
+	st := &durableCheckpointFixture{MVCCStore: store.NewMVCCStore(), checkpoint: encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{7}}, applied: 900}
+	fsm, ok := NewKvFSMWithHLC(st, NewHLC()).(*kvFSM)
+	require.True(t, ok)
+	require.Error(t, fsm.RestoreSnapshotHeader(bytes.NewReader(encoded[:size])))
+	require.Zero(t, fsm.encryptionCheckpoint)
+}
diff --git a/store/encryption_checkpoint.go b/store/encryption_checkpoint.go
new file mode 100644
index 000000000..242ae8b20
--- /dev/null
+++ b/store/encryption_checkpoint.go
@@ -0,0 +1,54 @@
+package store
+
+import (
+	"encoding/binary"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/v2"
+)
+
+var encryptionCheckpointKey = []byte("_meta_encryption_snapshot_checkpoint")
+
+const encryptionCheckpointSize = 8 + 32
+
+// SaveEncryptionSnapshotCheckpoint is an operational metadata write, like the
+// writer registry: no user timestamp, nonce, or applied watermark is issued.
+// It is durable before the encryption handler returns and later data entries
+// can advance the durable applied index past it safely.
+func (s *pebbleStore) SaveEncryptionSnapshotCheckpoint(cp encryption.SnapshotCheckpoint) error {
+	s.applyMu.Lock()
+	defer s.applyMu.Unlock()
+	s.dbMu.RLock()
+	defer s.dbMu.RUnlock()
+	if cp.Index == 0 {
+		return errors.New("encryption snapshot checkpoint: zero durable index")
+	}
+	var raw [encryptionCheckpointSize]byte
+	binary.BigEndian.PutUint64(raw[:8], cp.Index)
+	copy(raw[8:], cp.State[:])
+	return errors.Wrap(s.db.Set(encryptionCheckpointKey, raw[:], pebble.Sync), "persist encryption snapshot checkpoint")
+}
+
+func (s *pebbleStore) LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error) {
+	s.dbMu.RLock()
+	defer s.dbMu.RUnlock()
+	var cp encryption.SnapshotCheckpoint
+	raw, closer, err := s.db.Get(encryptionCheckpointKey)
+	if errors.Is(err, pebble.ErrNotFound) {
+		return cp, false, nil
+	}
+	if err != nil {
+		return cp, false, errors.Wrap(err, "read durable encryption snapshot checkpoint")
+	}
+	defer closer.Close()
+	if len(raw) != encryptionCheckpointSize {
+		return cp, false, errors.New("encryption snapshot checkpoint: corrupt durable record")
+	}
+	cp.Index = binary.BigEndian.Uint64(raw[:8])
+	copy(cp.State[:], raw[8:])
+	if cp.Index == 0 {
+		return cp, false, errors.New("encryption snapshot checkpoint: zero durable index")
+	}
+	return cp, true, nil
+}
diff --git a/store/encryption_checkpoint_test.go b/store/encryption_checkpoint_test.go
new file mode 100644
index 000000000..a481bbf58
--- /dev/null
+++ b/store/encryption_checkpoint_test.go
@@ -0,0 +1,55 @@
+package store
+
+import (
+	"bytes"
+	"io"
+	"path/filepath"
+	"testing"
+
+	"github.com/bootjp/elastickv/internal/encryption"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEncryptionCheckpointDurableMetadataSurvivesReopenAndSnapshot(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "source")
+	st, err := NewPebbleStore(path)
+	require.NoError(t, err)
+	checkpointStore, ok := st.(interface {
+		SaveEncryptionSnapshotCheckpoint(encryption.SnapshotCheckpoint) error
+	})
+	require.True(t, ok)
+	cp := encryption.SnapshotCheckpoint{Index: 890, State: [32]byte{8, 9}}
+	require.NoError(t, checkpointStore.SaveEncryptionSnapshotCheckpoint(cp))
+	require.Equal(t, uint64(0), st.LastCommitTS(), "operational metadata cannot issue a user timestamp")
+	require.NoError(t, st.Close())
+	st, err = NewPebbleStore(path)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, st.Close()) })
+	loader, ok := st.(interface {
+		LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error)
+	})
+	require.True(t, ok)
+	got, present, err := loader.LoadEncryptionSnapshotCheckpoint()
+	require.NoError(t, err)
+	require.True(t, present)
+	require.Equal(t, cp, got)
+	require.True(t, IsPebbleOperationalKey(encryptionCheckpointKey))
+	snap, err := st.Snapshot()
+	require.NoError(t, err)
+	defer snap.Close()
+	var payload bytes.Buffer
+	_, err = snap.WriteTo(&payload)
+	require.NoError(t, err)
+	dst, err := NewPebbleStore(filepath.Join(t.TempDir(), "restored"))
+	require.NoError(t, err)
+	defer dst.Close()
+	require.NoError(t, dst.Restore(io.NopCloser(&payload)))
+	reader, ok := dst.(interface {
+		LoadEncryptionSnapshotCheckpoint() (encryption.SnapshotCheckpoint, bool, error)
+	})
+	require.True(t, ok)
+	got, present, err = reader.LoadEncryptionSnapshotCheckpoint()
+	require.NoError(t, err)
+	require.True(t, present)
+	require.Equal(t, cp, got)
+}
diff --git a/store/lsm_store.go b/store/lsm_store.go
index 62acb45b2..be7742d44 100644
--- a/store/lsm_store.go
+++ b/store/lsm_store.go
@@ -727,6 +727,7 @@ func isPebbleMetaKey(rawKey []byte) bool {
 		bytes.Equal(rawKey, metaMinRetainedTSBytes) ||
 		bytes.Equal(rawKey, metaPendingMinRetainedTSBytes) ||
 		bytes.Equal(rawKey, metaAppliedIndexBytes) ||
+		bytes.Equal(rawKey, encryptionCheckpointKey) ||
 		isMigrationMetadataKey(rawKey) ||
 		isPebbleWriterRegistryKey(rawKey)
 }

From e9ec3b87e6ef4dbd68eb4b027b059ff87bbd2e5d Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Fri, 23 Oct 2026 03:16:59 +0900
Subject: [PATCH 3/6] Avoid redundant snapshot scans without a sidecar gap

---
 docs/encryption_compacted_restart.md      | 15 ++++++++++++++
 main_encryption_compacted_restart_test.go | 24 +++++++++++++++++++++++
 main_encryption_snapshot_checkpoint.go    |  5 +++++
 3 files changed, 44 insertions(+)

diff --git a/docs/encryption_compacted_restart.md b/docs/encryption_compacted_restart.md
index 12147f8c1..4b72cb533 100644
--- a/docs/encryption_compacted_restart.md
+++ b/docs/encryption_compacted_restart.md
@@ -54,6 +54,15 @@ 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.
+
 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
@@ -71,6 +80,9 @@ consistency anomaly is fixed. No change from PR #1286 is included.
   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).
@@ -87,3 +99,6 @@ 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.
diff --git a/main_encryption_compacted_restart_test.go b/main_encryption_compacted_restart_test.go
index 8fadfb5cf..b4ef93d86 100644
--- a/main_encryption_compacted_restart_test.go
+++ b/main_encryption_compacted_restart_test.go
@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"bytes"
 	"encoding/binary"
+	"fmt"
 	"io"
 	"path/filepath"
 	"testing"
@@ -20,12 +21,35 @@ type checkpointGapEngine struct {
 	payload       []byte
 	snapshotIndex uint64
 	snapshotErr   error
+	snapshotCalls int
 }
 
 func (e *checkpointGapEngine) EncryptionSnapshot() (io.ReadCloser, uint64, error) {
+	e.snapshotCalls++
 	return io.NopCloser(bytes.NewReader(e.payload)), e.snapshotIndex, e.snapshotErr
 }
 
+func TestEncryptionCheckpointReadsSnapshotOnlyForGap(t *testing.T) {
+	for _, applied := range []uint64{5, 6, 7} {
+		t.Run(fmt.Sprintf("applied-%d", applied), func(t *testing.T) {
+			engine := &checkpointGapEngine{
+				stubGapEngine: stubGapEngine{appliedIndex: applied},
+				snapshotErr:   io.ErrUnexpectedEOF,
+			}
+			sc := &encryption.Sidecar{RaftAppliedIndex: 6}
+			start, err := sidecarGapStartFromSnapshot(engine, sc)
+			if applied > sc.RaftAppliedIndex {
+				require.ErrorIs(t, err, io.ErrUnexpectedEOF)
+				require.Equal(t, 1, engine.snapshotCalls)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, sc.RaftAppliedIndex, start)
+			require.Zero(t, engine.snapshotCalls)
+		})
+	}
+}
+
 type checkpointScanner struct {
 	t          *testing.T
 	start, end uint64
diff --git a/main_encryption_snapshot_checkpoint.go b/main_encryption_snapshot_checkpoint.go
index 3ba93d603..19a27fa6e 100644
--- a/main_encryption_snapshot_checkpoint.go
+++ b/main_encryption_snapshot_checkpoint.go
@@ -11,6 +11,11 @@ import (
 
 func sidecarGapStartFromSnapshot(engine encryptionGapEngine, sc *encryption.Sidecar) (uint64, error) {
 	start := sc.RaftAppliedIndex
+	if start >= engine.AppliedIndex() {
+		// Engine restore already validated the installed snapshot. No
+		// compacted gap needs a witness, so do not repeat its full CRC scan.
+		return start, nil
+	}
 	source, ok := engine.(interface {
 		EncryptionSnapshot() (io.ReadCloser, uint64, error)
 	})

From 80bc0979faf2b585b499820dedd60bd4c84856cd Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Fri, 23 Oct 2026 03:36:48 +0900
Subject: [PATCH 4/6] test: keep flusher alignment observations live

---
 keyviz/flusher_test.go | 85 ++++++++++++++++++++++++++++++++++++------
 1 file changed, 74 insertions(+), 11 deletions(-)

diff --git a/keyviz/flusher_test.go b/keyviz/flusher_test.go
index df82a1de2..24abcf590 100644
--- a/keyviz/flusher_test.go
+++ b/keyviz/flusher_test.go
@@ -62,25 +62,88 @@ func TestRunFlusherAttachesAlignedHotKeysWindow(t *testing.T) {
 		RunFlusher(ctx, s, s.Step())
 		done <- struct{}{}
 	}()
+	t.Cleanup(func() {
+		cancel()
+		for range 2 {
+			select {
+			case <-done:
+			case <-time.After(time.Second):
+				t.Fatal("hot-key workers did not return after cancel")
+			}
+		}
+	})
 	for i := 0; i < 8; i++ {
 		s.Observe(1, []byte("hot"), OpWrite, 0, LabelLegacy)
 	}
 
 	require.Eventually(t, func() bool {
-		for _, col := range s.Snapshot(time.Time{}, time.Time{}) {
-			if len(col.HotKeys) == 0 {
+		// Keep producing while polling: the 16-column history retains only
+		// 80ms, so a scheduler pause can evict every initial observation.
+		s.Observe(1, []byte("hot"), OpWrite, 0, LabelLegacy)
+		return alignedHotKeysWindowObserved(s.Snapshot(time.Time{}, time.Time{}))
+	}, time.Second, 5*time.Millisecond)
+}
+
+func alignedHotKeysWindowObserved(columns []MatrixColumn) bool {
+	observed := false
+	for _, col := range columns {
+		for _, snapshot := range col.HotKeys {
+			if len(snapshot.Entries) == 0 {
 				continue
 			}
-			snapshot := col.HotKeys[0]
-			return snapshot.WindowStart.Equal(col.WindowStart) &&
-				snapshot.WindowEnd.Equal(col.At) &&
-				len(snapshot.Entries) > 0
+			if !snapshot.WindowStart.Equal(col.WindowStart) || !snapshot.WindowEnd.Equal(col.At) {
+				return false
+			}
+			observed = true
 		}
-		return false
-	}, time.Second, 5*time.Millisecond)
-	cancel()
-	<-done
-	<-done
+	}
+	return observed
+}
+
+func TestAlignedHotKeysWindowObservedSkipsEmptyWindows(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	end := start.Add(5 * time.Millisecond)
+	columns := []MatrixColumn{
+		{WindowStart: start, At: end, HotKeys: []KeyvizHotKeysSnapshot{{WindowStart: start, WindowEnd: end}}},
+		{WindowStart: end, At: end.Add(5 * time.Millisecond), HotKeys: []KeyvizHotKeysSnapshot{{
+			WindowStart: end, WindowEnd: end.Add(5 * time.Millisecond), Entries: []KeyvizHotKeyEntry{{Key: []byte("hot")}},
+		}}},
+	}
+	require.True(t, alignedHotKeysWindowObserved(columns))
+	columns[1].HotKeys[0].WindowStart = start
+	require.False(t, alignedHotKeysWindowObserved(columns), "a populated misaligned window must still fail")
+	require.False(t, alignedHotKeysWindowObserved(columns[:1]), "empty windows cannot satisfy the assertion")
+}
+
+func TestFlusherAlignmentProbeAfterHistoryEviction(t *testing.T) {
+	start := time.Unix(1_700_000_000, 0)
+	step := 5 * time.Millisecond
+	s := NewMemSampler(MemSamplerOptions{
+		Step: step, HistoryColumns: 2, HotKeysEnabled: true,
+		HotKeysPerRoute: 8, HotKeysSampleRate: 1, HotKeysQueueSize: 32,
+		HotKeysMaxKeyLen: 64, Now: func() time.Time { return start },
+	})
+	require.True(t, s.RegisterRoute(1, []byte("a"), []byte("z"), 1))
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		RunHotKeysAggregator(ctx, s)
+	}()
+	t.Cleanup(func() {
+		cancel()
+		<-done
+	})
+	s.Observe(1, []byte("hot"), OpWrite, 0, LabelLegacy)
+	require.True(t, s.flushWindow(ctx, start.Add(step)))
+	require.True(t, alignedHotKeysWindowObserved(s.Snapshot(time.Time{}, time.Time{})))
+	for i := 2; i <= 3; i++ {
+		require.True(t, s.flushWindow(ctx, start.Add(time.Duration(i)*step)))
+	}
+	require.False(t, alignedHotKeysWindowObserved(s.Snapshot(time.Time{}, time.Time{})), "the initial sample has been evicted")
+	s.Observe(1, []byte("hot"), OpWrite, 0, LabelLegacy)
+	require.True(t, s.flushWindow(ctx, start.Add(4*step)))
+	require.True(t, alignedHotKeysWindowObserved(s.Snapshot(time.Time{}, time.Time{})), "an earlier empty window must not hide the new sample")
 }
 
 func TestFlushWindowKeepsHotKeyAndCounterInSameColumn(t *testing.T) {

From 32704e4d4a0084ed74145703d396f2ee89648517 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Fri, 23 Oct 2026 03:48:24 +0900
Subject: [PATCH 5/6] Retry sidecar gap scans overtaken by snapshot compaction

---
 docs/encryption_compacted_restart.md      | 12 ++++
 main_encryption_compacted_restart_test.go | 74 +++++++++++++++++++++++
 main_encryption_snapshot_checkpoint.go    | 25 ++++++++
 main_encryption_startup_guard.go          | 10 +--
 4 files changed, 112 insertions(+), 9 deletions(-)

diff --git a/docs/encryption_compacted_restart.md b/docs/encryption_compacted_restart.md
index 4b72cb533..e451481c0 100644
--- a/docs/encryption_compacted_restart.md
+++ b/docs/encryption_compacted_restart.md
@@ -48,6 +48,14 @@ 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.
+
 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.
@@ -102,3 +110,7 @@ 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 the unchanged 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.
diff --git a/main_encryption_compacted_restart_test.go b/main_encryption_compacted_restart_test.go
index b4ef93d86..d8d6b5862 100644
--- a/main_encryption_compacted_restart_test.go
+++ b/main_encryption_compacted_restart_test.go
@@ -56,6 +56,80 @@ type checkpointScanner struct {
 	hit        bool
 }
 
+type advancingCheckpointScanner struct {
+	t       *testing.T
+	engine  *checkpointGapEngine
+	variant string
+	calls   int
+}
+
+func (s *advancingCheckpointScanner) HasEncryptionRelevantEntryInRange(start, end uint64) (bool, error) {
+	s.calls++
+	require.Equal(s.t, uint64(877), end)
+	if s.variant == "keeps-compacting" {
+		require.Equal(s.t, s.engine.snapshotIndex, start)
+		s.engine.snapshotIndex++
+		return false, etcdraft.ErrCompacted
+	}
+	if s.calls > 1 {
+		require.Equal(s.t, uint64(870), start)
+		return s.variant == "retained-rotation", nil
+	}
+	require.Equal(s.t, uint64(860), start)
+	if s.variant == "unavailable" {
+		return false, etcdraft.ErrUnavailable
+	}
+	if s.variant != "unchanged" {
+		s.engine.snapshotIndex = 870
+	}
+	switch s.variant {
+	case "missing-proof":
+		s.engine.payload = []byte("EKVTHLC1\x00\x00\x00\x00\x00\x00\x00\x01")
+	case "changed-proof":
+		s.engine.payload[len(s.engine.payload)-1] ^= 1
+	case "verification-error":
+		s.engine.snapshotErr = io.ErrUnexpectedEOF
+	}
+	return false, etcdraft.ErrCompacted
+}
+
+func TestEncryptionCheckpointRetriesCompactionOnlyWithNewProof(t *testing.T) {
+	for _, variant := range []string{"advance", "unchanged", "missing-proof", "changed-proof", "verification-error", "retained-rotation", "unavailable", "keeps-compacting"} {
+		t.Run(variant, func(t *testing.T) {
+			sc := &encryption.Sidecar{Version: encryption.SidecarVersion, RaftAppliedIndex: 6}
+			engine := &checkpointGapEngine{
+				stubGapEngine: stubGapEngine{appliedIndex: 877},
+				payload:       checkpointHeader(t, sc), snapshotIndex: 860,
+			}
+			scanner := &advancingCheckpointScanner{t: t, engine: engine, variant: variant}
+			engine.scanner = scanner
+			path := filepath.Join(t.TempDir(), "keys.json")
+			require.NoError(t, encryption.WriteSidecar(path, sc))
+			err := runSidecarBehindRaftLogGuard(engine, path, 1)
+			if variant == "advance" {
+				require.NoError(t, err)
+				require.Equal(t, 2, scanner.calls)
+			} else {
+				require.Error(t, err)
+			}
+			switch variant {
+			case "unchanged", "missing-proof", "keeps-compacting":
+				require.ErrorIs(t, err, etcdraft.ErrCompacted)
+			case "changed-proof", "retained-rotation":
+				require.ErrorIs(t, err, encryption.ErrSidecarBehindRaftLog)
+			case "verification-error":
+				require.ErrorIs(t, err, io.ErrUnexpectedEOF)
+			case "unavailable":
+				require.ErrorIs(t, err, etcdraft.ErrUnavailable)
+				require.Equal(t, 1, scanner.calls)
+			}
+			if variant == "keeps-compacting" {
+				require.Equal(t, 3, scanner.calls, "continuous compaction must stop at the retry bound")
+			}
+		})
+	}
+}
+
 func (s *checkpointScanner) HasEncryptionRelevantEntryInRange(start, end uint64) (bool, error) {
 	if start < s.start {
 		return false, etcdraft.ErrCompacted
diff --git a/main_encryption_snapshot_checkpoint.go b/main_encryption_snapshot_checkpoint.go
index 19a27fa6e..3b317d07e 100644
--- a/main_encryption_snapshot_checkpoint.go
+++ b/main_encryption_snapshot_checkpoint.go
@@ -7,8 +7,33 @@ import (
 	"github.com/bootjp/elastickv/internal/encryption"
 	"github.com/bootjp/elastickv/kv"
 	"github.com/cockroachdb/errors"
+	etcdraft "go.etcd.io/raft/v3"
 )
 
+const maxEncryptionSnapshotGapAttempts = 3
+
+func verifySidecarGapWithSnapshot(engine encryptionGapEngine, sc *encryption.Sidecar) error {
+	var previousStart uint64
+	var scanErr error
+	for range maxEncryptionSnapshotGapAttempts {
+		start, err := sidecarGapStartFromSnapshot(engine, sc)
+		if err != nil {
+			return err
+		}
+		if scanErr != nil && start <= previousStart {
+			return errors.Wrap(scanErr, "encryption startup guard: snapshot proof did not advance")
+		}
+		scanErr = encryption.GuardSidecarBehindRaftLog(start, engine.AppliedIndex(), engine.EncryptionScanner())
+		if !errors.Is(scanErr, etcdraft.ErrCompacted) {
+			return errors.Wrap(scanErr, "encryption startup guard: scan retained suffix")
+		}
+		// Snapshot work can compact the selected generation during the scan.
+		// Retry only if a freshly verified witness advances its covered prefix.
+		previousStart = start
+	}
+	return errors.Wrap(scanErr, "encryption startup guard: snapshot compaction retry limit")
+}
+
 func sidecarGapStartFromSnapshot(engine encryptionGapEngine, sc *encryption.Sidecar) (uint64, error) {
 	start := sc.RaftAppliedIndex
 	if start >= engine.AppliedIndex() {
diff --git a/main_encryption_startup_guard.go b/main_encryption_startup_guard.go
index c51aeb268..904d51959 100644
--- a/main_encryption_startup_guard.go
+++ b/main_encryption_startup_guard.go
@@ -184,15 +184,7 @@ func runSidecarBehindRaftLogGuard(
 			slog.Uint64("engine_applied_index", gapEngine.AppliedIndex()))
 		return nil
 	}
-	start, err := sidecarGapStartFromSnapshot(gapEngine, sidecar)
-	if err != nil {
-		return err
-	}
-	if err := encryption.GuardSidecarBehindRaftLog(
-		start,
-		gapEngine.AppliedIndex(),
-		gapEngine.EncryptionScanner(),
-	); err != nil {
+	if err := verifySidecarGapWithSnapshot(gapEngine, sidecar); err != nil {
 		return pkgerrors.Wrapf(err,
 			"encryption startup guard: sidecar=%q default_group=%d",
 			sidecarPath, defaultGroup)

From 8b99614644003bb5a0b12cf798aa500e2bec051f Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Fri, 23 Oct 2026 04:03:56 +0900
Subject: [PATCH 6/6] Refresh retry evidence and preserve snapshot extensions

---
 docs/encryption_compacted_restart.md      |  9 ++++-
 keyviz/sampler_test.go                    | 30 +++++++-------
 kv/snapshot_encryption_checkpoint.go      |  2 +-
 kv/snapshot_encryption_checkpoint_test.go | 19 +++++++++
 main_encryption_compacted_restart_test.go | 49 +++++++++++++++++------
 main_encryption_snapshot_checkpoint.go    | 14 ++++++-
 main_encryption_startup_guard.go          |  2 +-
 7 files changed, 94 insertions(+), 31 deletions(-)

diff --git a/docs/encryption_compacted_restart.md b/docs/encryption_compacted_restart.md
index e451481c0..0fb363c96 100644
--- a/docs/encryption_compacted_restart.md
+++ b/docs/encryption_compacted_restart.md
@@ -55,6 +55,10 @@ 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.
@@ -70,6 +74,9 @@ 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
@@ -111,6 +118,6 @@ 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 the unchanged sidecar and a freshly verified
+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.
diff --git a/keyviz/sampler_test.go b/keyviz/sampler_test.go
index 127d46635..2f4a804f4 100644
--- a/keyviz/sampler_test.go
+++ b/keyviz/sampler_test.go
@@ -478,28 +478,30 @@ func TestConcurrentRegisterAndObserveRace(t *testing.T) {
 	t.Parallel()
 	s, _ := newTestSampler(t, MemSamplerOptions{Step: time.Second, HistoryColumns: 8})
 	const routes = 64
+	const observers = 4
+	const rounds = 64
 	var observed atomic.Uint64
 
-	stop := make(chan struct{})
+	start := make(chan struct{})
+	var ready sync.WaitGroup
+	ready.Add(1 + observers)
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
+		ready.Done()
+		<-start
 		for i := uint64(1); i <= routes; i++ {
 			s.RegisterRoute(i, []byte{byte('a' + i%26)}, []byte{byte('a' + (i+1)%26)}, 0)
-			time.Sleep(time.Microsecond)
 		}
 	}()
-	wg.Add(4)
-	for w := 0; w < 4; w++ {
+	wg.Add(observers)
+	for w := 0; w < observers; w++ {
 		go func() {
 			defer wg.Done()
-			for {
-				select {
-				case <-stop:
-					return
-				default:
-				}
+			ready.Done()
+			<-start
+			for range rounds {
 				for i := uint64(1); i <= routes; i++ {
 					s.Observe(i, make([]byte, 0), OpRead, 0, LabelLegacy)
 					observed.Add(1)
@@ -508,12 +510,12 @@ func TestConcurrentRegisterAndObserveRace(t *testing.T) {
 		}()
 	}
 
-	time.Sleep(20 * time.Millisecond)
-	close(stop)
+	ready.Wait()
+	close(start)
 	wg.Wait()
 
-	if observed.Load() == 0 {
-		t.Fatal("observers ran zero iterations — schedule starvation?")
+	if got, want := observed.Load(), uint64(observers*rounds*routes); got != want {
+		t.Fatalf("observer iterations = %d, want %d", got, want)
 	}
 }
 
diff --git a/kv/snapshot_encryption_checkpoint.go b/kv/snapshot_encryption_checkpoint.go
index 9a43a87e7..c2e5fb96f 100644
--- a/kv/snapshot_encryption_checkpoint.go
+++ b/kv/snapshot_encryption_checkpoint.go
@@ -21,7 +21,7 @@ func decodeSnapshotCheckpoint(payload []byte) (encryption.SnapshotCheckpoint, er
 		!bytes.Equal(payload[hlcSnapshotV2MinPayload:hlcSnapshotV2MinPayload+len(encryptionCheckpointMagic)], encryptionCheckpointMagic) {
 		return checkpoint, nil
 	}
-	if len(payload) != hlcSnapshotV2MinPayload+encryptionCheckpointBytes {
+	if len(payload) < hlcSnapshotV2MinPayload+encryptionCheckpointBytes {
 		return checkpoint, errors.New("snapshot encryption checkpoint: invalid length")
 	}
 	checkpoint.Index = binary.BigEndian.Uint64(payload[24:32])
diff --git a/kv/snapshot_encryption_checkpoint_test.go b/kv/snapshot_encryption_checkpoint_test.go
index 9dd518f8e..c0fad08b3 100644
--- a/kv/snapshot_encryption_checkpoint_test.go
+++ b/kv/snapshot_encryption_checkpoint_test.go
@@ -3,6 +3,8 @@ package kv
 import (
 	"bufio"
 	"bytes"
+	"encoding/binary"
+	"io"
 	"testing"
 
 	"github.com/bootjp/elastickv/internal/encryption"
@@ -78,6 +80,23 @@ func TestEncryptionSnapshotCheckpointRejectsTruncationWithoutMutation(t *testing
 	require.Equal(t, before, fsm.encryptionCheckpoint)
 }
 
+func TestEncryptionCheckpointAllowsTrailingV2Extensions(t *testing.T) {
+	cp := encryption.SnapshotCheckpoint{Index: 6, State: [32]byte{1, 2, 3}}
+	header := &kvFSMSnapshot{useV2: true, checkpoint: cp}
+	encoded, size := header.encodeHeader()
+	const extension = "FUTURE01"
+	payload := append(bytes.Clone(encoded[:size]), extension...)
+	binary.BigEndian.PutUint16(payload[8:10], uint16(hlcSnapshotV2MinPayload+encryptionCheckpointBytes+len(extension)))
+	body := []byte("unchanged physical snapshot body")
+	reader := bufio.NewReader(bytes.NewReader(append(payload, body...)))
+	_, _, got, err := ReadSnapshotHeaderWithCheckpoint(reader)
+	require.NoError(t, err)
+	require.Equal(t, cp, got)
+	remaining, err := io.ReadAll(reader)
+	require.NoError(t, err)
+	require.Equal(t, body, remaining)
+}
+
 type durableCheckpointFixture struct {
 	store.MVCCStore
 	checkpoint encryption.SnapshotCheckpoint
diff --git a/main_encryption_compacted_restart_test.go b/main_encryption_compacted_restart_test.go
index d8d6b5862..01d06d4b6 100644
--- a/main_encryption_compacted_restart_test.go
+++ b/main_encryption_compacted_restart_test.go
@@ -6,6 +6,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
+	"os"
 	"path/filepath"
 	"testing"
 
@@ -61,6 +62,7 @@ type advancingCheckpointScanner struct {
 	engine  *checkpointGapEngine
 	variant string
 	calls   int
+	path    string
 }
 
 func (s *advancingCheckpointScanner) HasEncryptionRelevantEntryInRange(start, end uint64) (bool, error) {
@@ -82,6 +84,11 @@ func (s *advancingCheckpointScanner) HasEncryptionRelevantEntryInRange(start, en
 	if s.variant != "unchanged" {
 		s.engine.snapshotIndex = 870
 	}
+	s.prepareNextSnapshot()
+	return false, etcdraft.ErrCompacted
+}
+
+func (s *advancingCheckpointScanner) prepareNextSnapshot() {
 	switch s.variant {
 	case "missing-proof":
 		s.engine.payload = []byte("EKVTHLC1\x00\x00\x00\x00\x00\x00\x00\x01")
@@ -89,12 +96,20 @@ func (s *advancingCheckpointScanner) HasEncryptionRelevantEntryInRange(start, en
 		s.engine.payload[len(s.engine.payload)-1] ^= 1
 	case "verification-error":
 		s.engine.snapshotErr = io.ErrUnexpectedEOF
+	case "sidecar-advanced", "sidecar-rollback":
+		sc := &encryption.Sidecar{Version: encryption.SidecarVersion, RaftAppliedIndex: 7}
+		if s.variant == "sidecar-rollback" {
+			sc.RaftAppliedIndex = 5
+		}
+		require.NoError(s.t, encryption.WriteSidecar(s.path, sc))
+		s.engine.payload = checkpointHeader(s.t, sc)
+	case "sidecar-read-error":
+		require.NoError(s.t, os.Remove(s.path))
 	}
-	return false, etcdraft.ErrCompacted
 }
 
 func TestEncryptionCheckpointRetriesCompactionOnlyWithNewProof(t *testing.T) {
-	for _, variant := range []string{"advance", "unchanged", "missing-proof", "changed-proof", "verification-error", "retained-rotation", "unavailable", "keeps-compacting"} {
+	for _, variant := range []string{"advance", "unchanged", "missing-proof", "changed-proof", "verification-error", "retained-rotation", "unavailable", "keeps-compacting", "sidecar-advanced", "sidecar-rollback", "sidecar-read-error"} {
 		t.Run(variant, func(t *testing.T) {
 			sc := &encryption.Sidecar{Version: encryption.SidecarVersion, RaftAppliedIndex: 6}
 			engine := &checkpointGapEngine{
@@ -104,23 +119,17 @@ func TestEncryptionCheckpointRetriesCompactionOnlyWithNewProof(t *testing.T) {
 			scanner := &advancingCheckpointScanner{t: t, engine: engine, variant: variant}
 			engine.scanner = scanner
 			path := filepath.Join(t.TempDir(), "keys.json")
+			scanner.path = path
 			require.NoError(t, encryption.WriteSidecar(path, sc))
 			err := runSidecarBehindRaftLogGuard(engine, path, 1)
-			if variant == "advance" {
+			if variant == "advance" || variant == "sidecar-advanced" {
 				require.NoError(t, err)
 				require.Equal(t, 2, scanner.calls)
 			} else {
 				require.Error(t, err)
 			}
-			switch variant {
-			case "unchanged", "missing-proof", "keeps-compacting":
-				require.ErrorIs(t, err, etcdraft.ErrCompacted)
-			case "changed-proof", "retained-rotation":
-				require.ErrorIs(t, err, encryption.ErrSidecarBehindRaftLog)
-			case "verification-error":
-				require.ErrorIs(t, err, io.ErrUnexpectedEOF)
-			case "unavailable":
-				require.ErrorIs(t, err, etcdraft.ErrUnavailable)
+			requireCheckpointRetryError(t, variant, err)
+			if variant == "unavailable" {
 				require.Equal(t, 1, scanner.calls)
 			}
 			if variant == "keeps-compacting" {
@@ -130,6 +139,22 @@ func TestEncryptionCheckpointRetriesCompactionOnlyWithNewProof(t *testing.T) {
 	}
 }
 
+func requireCheckpointRetryError(t *testing.T, variant string, err error) {
+	t.Helper()
+	switch variant {
+	case "unchanged", "missing-proof", "keeps-compacting":
+		require.ErrorIs(t, err, etcdraft.ErrCompacted)
+	case "changed-proof", "retained-rotation", "sidecar-rollback":
+		require.ErrorIs(t, err, encryption.ErrSidecarBehindRaftLog)
+	case "verification-error":
+		require.ErrorIs(t, err, io.ErrUnexpectedEOF)
+	case "sidecar-read-error":
+		require.ErrorIs(t, err, os.ErrNotExist)
+	case "unavailable":
+		require.ErrorIs(t, err, etcdraft.ErrUnavailable)
+	}
+}
+
 func (s *checkpointScanner) HasEncryptionRelevantEntryInRange(start, end uint64) (bool, error) {
 	if start < s.start {
 		return false, etcdraft.ErrCompacted
diff --git a/main_encryption_snapshot_checkpoint.go b/main_encryption_snapshot_checkpoint.go
index 3b317d07e..f804c8c44 100644
--- a/main_encryption_snapshot_checkpoint.go
+++ b/main_encryption_snapshot_checkpoint.go
@@ -12,10 +12,20 @@ import (
 
 const maxEncryptionSnapshotGapAttempts = 3
 
-func verifySidecarGapWithSnapshot(engine encryptionGapEngine, sc *encryption.Sidecar) error {
+func verifySidecarGapWithSnapshot(engine encryptionGapEngine, sidecarPath string, sc *encryption.Sidecar) error {
 	var previousStart uint64
 	var scanErr error
-	for range maxEncryptionSnapshotGapAttempts {
+	for attempt := range maxEncryptionSnapshotGapAttempts {
+		if attempt > 0 {
+			fresh, err := encryption.ReadSidecar(sidecarPath)
+			if err != nil {
+				return errors.Wrap(err, "encryption startup guard: reload sidecar after compaction")
+			}
+			if fresh.RaftAppliedIndex < sc.RaftAppliedIndex {
+				return errors.Wrap(encryption.ErrSidecarBehindRaftLog, "sidecar index regressed during gap verification")
+			}
+			sc = fresh
+		}
 		start, err := sidecarGapStartFromSnapshot(engine, sc)
 		if err != nil {
 			return err
diff --git a/main_encryption_startup_guard.go b/main_encryption_startup_guard.go
index 904d51959..a8171a25e 100644
--- a/main_encryption_startup_guard.go
+++ b/main_encryption_startup_guard.go
@@ -184,7 +184,7 @@ func runSidecarBehindRaftLogGuard(
 			slog.Uint64("engine_applied_index", gapEngine.AppliedIndex()))
 		return nil
 	}
-	if err := verifySidecarGapWithSnapshot(gapEngine, sidecar); err != nil {
+	if err := verifySidecarGapWithSnapshot(gapEngine, sidecarPath, sidecar); err != nil {
 		return pkgerrors.Wrapf(err,
 			"encryption startup guard: sidecar=%q default_group=%d",
 			sidecarPath, defaultGroup)