feat(snapshot-skip B2): plumb metaAppliedIndex through raft-Apply + both snapshot persist sites (#915)

## Summary

Implements **Branch 2** of the cold-start snapshot-restore skip
optimisation designed in PR #910. After this lands the
`metaAppliedIndex` Pebble meta key is durably written on every
raft-Apply data mutation AND at every snapshot persist — but the skip
gate itself (Branch 3) is NOT yet wired, so behaviour is observationally
identical to `main` except for the new meta key in fsm.db. Branch 2 is
meant to soak in production for at least one release before Branch 3
enables the skip; this PR is intentionally a no-op-from-the-outside
change with comprehensive plumbing.

## Reading order (6 commits, designed to review one-at-a-time)

| # | commit | scope |
|---|---|---|
| 1 | `2339a6f2` | raftengine: opt-in interfaces (`AppliedIndexReader` /
`AppliedIndexWriter`) |
| 2 | `525fc152` | pebbleStore: `metaAppliedIndex` const +
`LastAppliedIndex` + `SetDurableAppliedIndex` (with `pebble.Sync`
UNCONDITIONALLY) |
| 3 | `aa9b8acc` | `MVCCStore` interface extension:
`ApplyMutationsRaftAt` / `DeletePrefixAtRaftAt` overloads, threading
appliedIndex through `applyMutationsWithOpts` + `deletePrefixAtWithOpts`
|
| 4 | `7cd72bda` | kvFSM seam wiring: `AppliedIndexReader()` /
`SetDurableAppliedIndex()` accessors + all 7 data-Apply leaves switched
to `*RaftAt` with `f.pendingApplyIdx` |
| 5 | `f1e8748c` | engine hooks at BOTH snapshot persist sites:
`persistCreatedSnapshot` + `e.persistLocalSnapshotPayload` call
`SetDurableAppliedIndex` BEFORE `persist.SaveSnap` |
| 6 | `2c42f7d6` | tests (10 new tests across store + engine) |

## Design constraints honoured

All from `docs/design/2026_06_02_idempotent_snapshot_restore.md`:

- **§2 "Why both leaves"**: meta key bundle in BOTH
`applyMutationsWithOpts` AND `deletePrefixAtWithOpts` so DEL_PREFIX
entries don't silently leave `LastAppliedIndex` behind. Tested by
`TestDeletePrefixAtRaftAt_BundlesMetaAppliedIndex`.
- **§3 `dbMu.RLock()`**: both `LastAppliedIndex` and
`SetDurableAppliedIndex` acquire the read-lock, matching the
lock-ordering discipline at `lsm_store.go:153 / :553 / :675`.
- **§4 fallback policy**: `AppliedIndexReader()` returns nil when the
store doesn't implement the seam; `LastAppliedIndex` returns `(0, false,
nil)` for missing OR truncated meta key. Branch 3 will then fall back to
full restore conservatively.
- **§6 `ELASTICKV_FSM_SYNC_MODE=nosync` mode**: `SetDurableAppliedIndex`
is **pinned to `pebble.Sync` unconditionally**. Rationale documented at
length in the method's doc-comment — once `persist.SaveSnap` returns,
WAL compaction discards every log entry ≤ `snap.Metadata.Index`, so
there's no source to replay the meta key bump from. +1 fsync per
snapshot persist (rare; default `SnapshotCount=10000`). Tested by
`TestSetDurableAppliedIndex_UsesPebbleSync`.
- **§6 "HLC lease entries — checkpoint at snapshot persist"**: BOTH
`persistCreatedSnapshot` (config snapshots) AND
`e.persistLocalSnapshotPayload` (steady-state `SnapshotCount`-triggered
hot path) call the hook. Both crash-ordering tested by
`TestPersistCreatedSnapshot_*`.
- **§8 compatibility**: `StateMachine.Apply`'s public signature is
unchanged. New interfaces are opt-in. Old call sites
(`ApplyMutationsRaft` without `*At`) still work, just pass
`appliedIndex=0` to opt out of the meta key bump.

## Test results

```
go vet ./...                                                    → 0 issues
go test ./store/ -short                                         → ok 29.4s
go test ./kv/    -short                                         → ok 10.4s
go test ./internal/raftengine/... -short                        → ok 32.8s
go test ./store/ -run 'TestLastAppliedIndex|TestSetDurable...|TestApply...|TestDelete...' → ok 1.6s
go test ./internal/raftengine/etcd/ -run 'TestRecording|TestPersistCreatedSnapshot_' → ok 0.03s
```

10 new tests added (see commit `2c42f7d6` for the full inventory).

## What this does NOT do

- **Does NOT enable the skip gate.** `restoreSnapshotState` still always
restores. Branch 3 wires the `fsmAlreadyAtIndex` check +
`applyHeaderStateOnSkip` + the two-phase `SnapshotHeaderApplier` seam.
- **Does NOT change `HEALTH_TIMEOUT_SECONDS=300`.** Branch 4 lowers it
once Branch 3 has soaked.
- **Does NOT touch the snapshot-install hot path**
(`Engine.applySnapshot`) per Non-Goals in the design.

## Soak plan

Branch 2 should run in production for at least one release before Branch
3 opens. Operators can verify the meta key is being written via:

```bash
# Inspect a pebble fsm.db (read-only)
ldb --db=/var/lib/elastickv/n3/fsm.db get '_meta_applied_index' --hex
# Expected: 8 little-endian bytes equal to the current applied index
```

## Refs

- PR #910 (design) — round 1..7 design history + retraction sections
explaining the design constraints this PR honours
- PR #909 — `HEALTH_TIMEOUT_SECONDS` band-aid that this series
eventually obviates


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Durable tracking of Raft-applied indexes to ensure consistent
snapshot/save ordering.

* **Bug Fixes**
* Improved snapshot persistence reliability by pinning durable applied
index before snapshot writes.
* Stronger durability for writes bundled with Raft entry indices,
reducing restore/recovery surprises.

* **Tests**
* Added comprehensive tests covering applied-index ordering, failure
handling, and persistence behavior.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: bootjp

docs/design/2026_06_02_idempotent_snapshot_restore.md

@@ -236,21 +236,37 @@ type AppliedIndexReader interface {
 }
 ```
 
-`kvFSM` exposes its store through a typed accessor so wal_store.go can
-reach it without importing the concrete pebble type:
+`kvFSM` directly satisfies `raftengine.AppliedIndexReader` by
+forwarding to its underlying store (PR #915 round-5 — round-1 had a
+factory `AppliedIndexReader() AppliedIndexReader` method intended to
+be called through a separate `AppliedIndexReporter` interface, but
+that pattern would require the skip gate to know about the reporter
+shim; the direct interface satisfaction is simpler and a compile-time
+guard catches future signature drift):
 
 ```go
-type AppliedIndexReporter interface {
-    AppliedIndexReader() AppliedIndexReader
-}
-func (f *kvFSM) AppliedIndexReader() AppliedIndexReader {
-    if r, ok := f.store.(AppliedIndexReader); ok {
-        return r
+// kv/fsm.go
+func (f *kvFSM) LastAppliedIndex() (uint64, bool, error) {
+    r, ok := f.store.(raftengine.AppliedIndexReader)
+    if !ok {
+        return 0, false, nil
     }
-    return nil
+    idx, present, err := r.LastAppliedIndex()
+    if err != nil {
+        return 0, false, errors.WithStack(err)
+    }
+    return idx, present, nil
 }
+
+// kv/fsm_applied_index_iface_check.go (compile-time guard)
+var _ raftengine.AppliedIndexReader = (*kvFSM)(nil)
+var _ raftengine.AppliedIndexWriter = (*kvFSM)(nil)
 ```
 
+The compile-time guard means any future rename or signature drift
+fails `go build` immediately — the soak investment is protected at
+the compiler level.
+
 ### 4. Conditional restore (with conservative error fallback)
 
 ```go
@@ -279,16 +295,20 @@ func restoreSnapshotState(fsm StateMachine, snapshot raftpb.Snapshot, fsmSnapDir
 
 // fsmAlreadyAtIndex returns true ONLY when we can prove the FSM is
 // already at or past `want`. Any uncertainty -- FSM doesn't expose
-// the reporter, store doesn't expose the reader, read error, or
-// missing meta key -- returns false so we fall back to the full
-// restore. A stale-but-incorrect skip is far worse than a wasteful
-// full restore; the fallback errs strictly toward restoring.
+// the reader interface, read error, or missing meta key -- returns
+// false so we fall back to the full restore. A stale-but-incorrect
+// skip is far worse than a wasteful full restore; the fallback errs
+// strictly toward restoring.
+//
+// Direct type-assert against raftengine.AppliedIndexReader (PR #915
+// round-5): kvFSM satisfies the interface directly via its
+// LastAppliedIndex method, so no separate AppliedIndexReporter shim
+// is needed. The compile-time guard in kv/fsm_applied_index_iface_check.go
+// keeps this stable.
 func fsmAlreadyAtIndex(fsm StateMachine, want uint64) bool {
-    reporter, ok := fsm.(AppliedIndexReporter)
+    r, ok := fsm.(raftengine.AppliedIndexReader)
     if !ok { return false }
-    reader := reporter.AppliedIndexReader()
-    if reader == nil { return false }
-    have, present, err := reader.LastAppliedIndex()
+    have, present, err := r.LastAppliedIndex()
     if err != nil || !present { return false }
     return have >= want
 }
@@ -814,8 +834,12 @@ present.
 - `ApplyIndexAware` is **already** in `main`; this design only adds
   consumers.
 - The new opt-in interfaces (`AppliedIndexReader`,
-  `AppliedIndexReporter`, `SnapshotHeaderApplier`) are additive.
+  `AppliedIndexWriter`, `SnapshotHeaderApplier`) are additive.
   FSMs that don't implement them fall back to the current behaviour.
+  (Round-1 / round-2 of this doc mentioned an `AppliedIndexReporter`
+  factory-method shim; PR #915 round-5 superseded it by having
+  `kvFSM` satisfy `AppliedIndexReader` directly via its
+  `LastAppliedIndex` method.)
 - `metaAppliedIndexBytes` is new.  Older fsm.db files don't have it.
   The `present=false` branch makes the first restart after upgrade
   fall back to full restore, populating the meta key from the next
@@ -863,7 +887,7 @@ restoreSnapshotState skipped (FSM at index %d, snapshot at %d, ceiling=%d, cutov
 | Branch | Content | Behaviour change |
 |---|---|---|
 | **B1** (this PR) | Design doc | None |
-| **B2** | `ApplyMutationsRaftAt` / `DeletePrefixAtRaftAt` overloads + meta-key bundling in both leaves + `pebbleStore.LastAppliedIndex()` (under `dbMu.RLock()`) + `pebbleStore.SetDurableAppliedIndex()` (under `dbMu.RLock()`, **`pebble.Sync` unconditionally**) + `kvFSM.AppliedIndexReader()` accessor + `kvFSM.SetDurableAppliedIndex` forwarding + thread `f.pendingApplyIdx` into the data-Apply leaves + BOTH `persistCreatedSnapshot` (`engine.go:2679`) AND `e.persistLocalSnapshotPayload` (`engine.go:4032`, the SnapshotCount-triggered hot path) call `SetDurableAppliedIndex` BEFORE the corresponding `persist.SaveSnap` | Meta key starts being written on every data Apply AND at every snapshot persist (both config-snapshot and steady-state local-snapshot paths). Skip is still disabled. Soak in production for one release. |
+| **B2** | `ApplyMutationsRaftAt` / `DeletePrefixAtRaftAt` overloads + meta-key bundling in both leaves + `pebbleStore.LastAppliedIndex()` (under `dbMu.RLock()`) + `pebbleStore.SetDurableAppliedIndex()` (under `dbMu.RLock()` + `applyMu.Lock()` RMW monotonic guard, **`pebble.Sync` unconditionally**) + `kvFSM.LastAppliedIndex()` directly satisfies `raftengine.AppliedIndexReader` (compile-time guard in `kv/fsm_applied_index_iface_check.go`) + `kvFSM.SetDurableAppliedIndex` forwarding + thread `f.pendingApplyIdx` into the data-Apply leaves + BOTH `persistCreatedSnapshot` (`engine.go:2679`) AND `e.persistLocalSnapshotPayload` (`engine.go:4032`, the SnapshotCount-triggered hot path) call `SetDurableAppliedIndex` BEFORE the corresponding `persist.SaveSnap` | Meta key starts being written on every data Apply AND at every snapshot persist (both config-snapshot and steady-state local-snapshot paths). Skip is still disabled. Soak in production for one release. |
 | **B3** | `restoreSnapshotState` skip gate + `applyHeaderStateOnSkip(snapPath, tok.CRC32C)` orchestrating size + footer-vs-tokenCRC + full-body-CRC verification using `internal/raftengine/etcd`'s existing helpers (matching `openAndRestoreFSMSnapshot`'s safety contract) + two-phase `SnapshotHeaderApplier` seam on `kvFSM` (`ParseSnapshotHeader(r io.Reader) (ceiling, cutover, err)` + pure `ApplySnapshotHeader(ceiling, cutover)`) + metrics + INFO log | **User-visible cold-start win.** |
 | **B4** | Lower `HEALTH_TIMEOUT_SECONDS` default once production data shows steady-state skip rate ≥ 90 % | Tighter ceiling; the env override remains honoured. |
 

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/aws/smithy-go v1.25.1
 	github.com/cockroachdb/errors v1.13.0
 	github.com/cockroachdb/pebble/v2 v2.1.5
+	github.com/coreos/go-semver v0.3.1
 	github.com/emirpasic/gods v1.18.1
 	github.com/getsentry/sentry-go v0.46.2
 	github.com/goccy/go-json v0.10.6
@@ -58,7 +59,6 @@ require (
 	github.com/cockroachdb/redact v1.1.5 // indirect
 	github.com/cockroachdb/swiss v0.0.0-20251224182025-b0f6560f979b // indirect
 	github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect
-	github.com/coreos/go-semver v0.3.1 // indirect
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect

internal/raftengine/etcd/engine.go

@@ -2026,6 +2026,15 @@ func (e *Engine) applyReadySnapshot(snapshot raftpb.Snapshot) error {
 		if err != nil {
 			return errors.Wrapf(err, "decode snapshot token index=%d", snapshot.Metadata.Index)
 		}
+		// B3/follow-up: also call SetDurableAppliedIndex(tok.Index) here
+		// after Restore so peer-after-InstallSnapshot populates the meta
+		// key. The local-snapshot persist path already bumps the live
+		// store (engine.persistLocalSnapshotPayload), but the receiving
+		// node's restored store inherits the pre-bump value embedded in
+		// the snapshot artifact. Design Non-Goals §
+		// docs/design/2026_06_02_idempotent_snapshot_restore.md:71-74
+		// scopes this out of Branch 2; see PR #915 round-4/5 codex P2 on
+		// engine.go:4077 for the rationale.
 		if err := openAndRestoreFSMSnapshot(e.fsm, fsmSnapPath(e.fsmSnapDir, tok.Index), tok.CRC32C); err != nil {
 			return errors.Wrapf(err, "restore fsm snapshot file index=%d crc=%08x", tok.Index, tok.CRC32C)
 		}
@@ -2704,10 +2713,38 @@ func (e *Engine) createConfigSnapshot(index uint64, confState raftpb.ConfState,
 	}
 }
 
+// bumpDurableAppliedIndexBeforeSave pins the FSM's durable applied
+// index to `index` BEFORE the engine calls persist.SaveSnap, so a
+// successful snapshot persist always implies LastAppliedIndex >=
+// snap.Metadata.Index — closes the HLC-lease-only / encryption-only
+// fallback (PR #910 design §6).
+//
+// FSMs that do not expose raftengine.AppliedIndexWriter silently
+// no-op; the skip optimisation falls back to full restore for them
+// (legacy test fakes, in-memory backends). pebble.Sync is forced on
+// the writer side regardless of ELASTICKV_FSM_SYNC_MODE — once
+// persist.SaveSnap returns, WAL compaction discards every log entry
+// at or before snap.Metadata.Index, so there is no source to replay
+// the meta key bump from.
+//
+// Used by BOTH snapshot persist sites: persistCreatedSnapshot (this
+// file) and e.persistLocalSnapshotPayload (the steady-state
+// SnapshotCount-triggered hot path).
+func (e *Engine) bumpDurableAppliedIndexBeforeSave(index uint64) error {
+	w, ok := e.fsm.(raftengine.AppliedIndexWriter)
+	if !ok {
+		return nil
+	}
+	return errors.WithStack(w.SetDurableAppliedIndex(index))
+}
+
 func (e *Engine) persistCreatedSnapshot(snap raftpb.Snapshot) error {
 	if etcdraft.IsEmptySnap(snap) || e.persist == nil {
 		return nil
 	}
+	if err := e.bumpDurableAppliedIndexBeforeSave(snap.Metadata.Index); err != nil {
+		return err
+	}
 	if err := e.persist.SaveSnap(snap); err != nil {
 		return errors.WithStack(err)
 	}
@@ -4072,29 +4109,49 @@ func (e *Engine) persistLocalSnapshotPayload(index uint64, payload []byte) error
 		return nil
 	}
 
+	if err := e.bumpDurableAppliedIndexBeforeSave(index); err != nil {
+		return err
+	}
+
 	_, err = persistLocalSnapshotPayload(e.storage, e.persist, index, payload)
+	return e.handleLocalSnapshotPersistResult(err)
+}
+
+// handleLocalSnapshotPersistResult collapses the post-SaveSnap error
+// switch into a single helper so persistLocalSnapshotPayload stays
+// under the cyclomatic-complexity budget. The three raft-side
+// 'snapshot already moved on' cases (ErrCompacted / ErrUnavailable /
+// ErrSnapOutOfDate) are all treated as no-ops; only the success path
+// runs the disk-side purge.
+func (e *Engine) handleLocalSnapshotPersistResult(err error) error {
 	switch {
 	case err == nil:
-		snapDir := filepath.Join(e.dataDir, snapDirName)
-		if purgeErr := purgeOldSnapshotFiles(snapDir, e.fsmSnapDir); purgeErr != nil {
-			slog.Warn("failed to purge old snap files", "error", purgeErr)
-		}
-		walDir := filepath.Join(e.dataDir, walDirName)
-		if purgeErr := purgeOldWALFiles(walDir, e.walRetention()); purgeErr != nil {
-			slog.Warn("failed to purge old wal files", "error", purgeErr)
-		}
-		return nil
-	case errors.Is(err, etcdraft.ErrCompacted):
-		return nil
-	case errors.Is(err, etcdraft.ErrUnavailable):
+		e.purgeAfterLocalSnapshot()
 		return nil
-	case errors.Is(err, etcdraft.ErrSnapOutOfDate):
+	case errors.Is(err, etcdraft.ErrCompacted),
+		errors.Is(err, etcdraft.ErrUnavailable),
+		errors.Is(err, etcdraft.ErrSnapOutOfDate):
 		return nil
 	default:
 		return err
 	}
 }
 
+// purgeAfterLocalSnapshot runs the disk-side cleanup that follows a
+// successful local-snapshot persist: trim old .snap/.fsm files and
+// rotate ageing WAL segments. Both calls log on error but do not
+// propagate — failing to purge is non-fatal.
+func (e *Engine) purgeAfterLocalSnapshot() {
+	snapDir := filepath.Join(e.dataDir, snapDirName)
+	if purgeErr := purgeOldSnapshotFiles(snapDir, e.fsmSnapDir); purgeErr != nil {
+		slog.Warn("failed to purge old snap files", "error", purgeErr)
+	}
+	walDir := filepath.Join(e.dataDir, walDirName)
+	if purgeErr := purgeOldWALFiles(walDir, e.walRetention()); purgeErr != nil {
+		slog.Warn("failed to purge old wal files", "error", purgeErr)
+	}
+}
+
 func encodeReadContext(id uint64) []byte {
 	out := make([]byte, envelopeHeaderSize)
 	out[0] = readContextVersion