feat(main): Stage 6C-2d — wire ErrSidecarBehindRaftLog startup guard into main.go (#784)

## Summary

Stage 6C-2d per the [PR #762
plan](#762) and the 6C
sub-decomposition landed in PR #781 / #782 / #783.

Ties the encryption-side guard primitive ([PR #782
`GuardSidecarBehindRaftLog`](#782))
to the raftengine-side scanner ([PR #783
`Engine.EncryptionScanner()`](#783))
in `main.go`'s startup phase. **Completes the §9.1
`ErrSidecarBehindRaftLog` work end-to-end.**

## What this PR ships

### `main_encryption_startup_guard.go` (new)

| Function | Role |
|---|---|
| `checkSidecarBehindRaftLog(runtimes, defaultGroup, sidecarPath,
encryptionEnabled)` | Locates the default-group runtime, type-asserts
its engine to local `encryptionGapEngine` interface, reads the §5.1
sidecar, delegates to the inner helper |
| `runSidecarBehindRaftLogGuard(gapEngine, sidecarPath, defaultGroup)` |
Per-engine inner half. Split out so tests use a stub instead of
constructing a full `raftengine.Engine` |
| `chainEncryptionStartupGuard(prevErr, ...)` | Composes the guard with
the existing `buildShardGroups` error path. Keeps `run()`'s cyclop
budget intact |

### Lifecycle phase ordering

```
1. loadKEKAndRunStartupGuards   ← 6B-2 RPC gate + 6C-1/6C-2 flag/sidecar/exhaustion
2. buildShardGroups             ← opens every shard's engine
3. chainEncryptionStartupGuard  ← 6C-2d gap-coverage guard (NEW)
4. gRPC servers / accept loop
```

Step 3 runs in a **different lifecycle phase** from steps 1 — the guard
requires the engine to be open and its applied index populated from
WAL+snapshot replay.

### Skip conditions (`checkSidecarBehindRaftLog` returns nil)

- `encryptionEnabled` is false — no encryption opt-in, no gap to refuse
on
- `sidecarPath` is empty — operator hasn't configured a sidecar location
- On-disk sidecar file absent — bootstrap hasn't committed yet
- Default-group runtime missing — no engine to query
- Engine not open — earlier startup-guard layer would have caught this

### Why scope is restricted to the default group

The §5.1 sidecar tracks a **single** `raft_applied_index`, advanced only
by `WriteSidecar` calls in `ApplyBootstrap` / `ApplyRotation` on the
default group's FSM. Other shards' Raft logs don't carry encryption FSM
entries — running the guard against them would always report caught-up
(gap == 0). `findDefaultGroupRuntime` silently skips non-matching
runtimes.

## Test coverage (11 new tests)

| Test | Path |
|---|---|
| `_DisabledNoop` | `encryptionEnabled=false` → skip |
| `_NoSidecarPathNoop` | empty path → skip |
| `_SidecarAbsentNoop` | on-disk file missing → skip |
| `_SidecarStatError` | I/O error (NUL path) → wrapped, NOT `not-exist`
|
| `_NoRuntimes` | no default-group runtime → skip |
| `_CaughtUp` | sidecar idx >= engine idx → pass, scanner never
consulted |
| `_GapNotCovered` | gap exists, scanner says no → pass |
| `_GapCovered` | gap covers relevant entry → `ErrSidecarBehindRaftLog`
|
| `_ScannerError` | scanner error → propagated wrapped, NOT
`ErrSidecarBehindRaftLog` |
| `_PropagatesPrevError` | chain with non-nil prev → returns prev
verbatim |
| `_NilPrevRunsGuard` | chain with nil prev → forwards to guard |

Tests use `stubGapEngine` + `stubScanner` to exercise the per-engine
inner function directly, avoiding the full `raftengine.Engine` interface
(which would require panic-stubs for ~30 unrelated methods).

## Posture matrix (before/after this PR)

| Scenario | pre-this-PR | post-this-PR |
|---|---|---|
| Sidecar caught up to engine | boot, encrypted | boot, encrypted |
| Sidecar behind, gap harmless (no §5.5 opcodes) | boot, encrypted |
boot, encrypted |
| Sidecar behind, gap covers `OpBootstrap` / `OpRotation` | **boot,
HaltApply on first post-cutover read with `unknown_key_id`** | refuse:
`ErrSidecarBehindRaftLog` |
| Sidecar behind, gap covers only `OpRegistration` | boot, registry
replays cleanly | boot (PR #782 narrowed predicate excludes 0x03) |

## Caller audit (per cron directive)

No public function signatures changed. The new functions in
`main_encryption_startup_guard.go` are package-internal. `main.go run()`
is the sole production caller; tests call the inner helpers directly.

The semantic effect of the wiring: nodes whose sidecar is behind by an
encryption-relevant interval now refuse at startup instead of silently
booting into a HaltApply later. This is a tighter failure boundary —
operators see the §9.1 typed refusal pointing at `encryption
resync-sidecar` rather than triaging downstream `unknown_key_id` errors.

## Stage 6B-2 / 6C-1 / 6C-2 / 6C-2b / 6C-2c invariants preserved

All earlier guards and their lifecycle phases unchanged. This PR
composes existing primitives — no new sentinel, no new opcode, no new
wire format.

## Five-lens self-review

1. **Data loss** — net-positive. Catches the partial-write-crash failure
mode between an encryption-relevant Raft commit and the sidecar update.
The guard primitive existed since PR #782 but was operator-inert until
this wiring.
2. **Concurrency / distributed failures** — runs at startup before any
goroutine fan-out / gRPC accept. No shared state introduced.
3. **Performance** — single sidecar read + single scanner call per
process start on the happy path.
4. **Data consistency** — the only §9.1 mechanism that catches "Raft
committed but sidecar missed it" partial writes. Without it the first
post-cutover read would HaltApply.
5. **Test coverage** — 11 new tests cover every branch. Full main
package test suite passes with no regressions.

## Test plan

- [x] `go test -race -timeout=180s .` — PASS (all main package tests +
11 new)
- [x] `go test -race -timeout=120s ./internal/raftengine/etcd/...
./internal/encryption/...` — PASS (no regressions)
- [x] `go build ./...` — PASS
- [x] `golangci-lint run .` — 0 issues (cyclop holds at 10 via the
composition helper)
- [ ] Full Jepsen suite — not run; the guard is purely a startup refusal
that runs before any Raft entry can be served

## Plan

**Stage 6C is now complete** (6C-1 ✅ + 6C-2 ✅ + 6C-2b ✅ + 6C-2c ✅ +
6C-2d this PR).

Next stages per PR #762:
- **6C-3** — `ErrNodeIDCollision` + `ErrLocalEpochRollback` (bundles
with Stage 6D's capability fan-out)
- **6D** — `enable-storage-envelope` admin RPC + §7.1 Phase-1 cutover
(**now UNBLOCKED** — all `ErrSidecarBehindRaftLog` work has shipped)
- **6E** — `enable-raft-envelope` admin RPC + §7.1 Phase-2 cutover
(bundles 6C-4)
- **6F** — `--encryption-rotate-on-startup` ergonomics


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

## Summary by CodeRabbit

* **New Features**
* Added encryption startup validation to ensure sidecar state is
synchronized with the raft log during system initialization, preventing
startup if consistency checks fail.

* **Tests**
* Comprehensive test coverage for encryption startup guard behavior,
including fast-skip scenarios, gap detection, and error propagation.

<!-- review_stack_entry_start -->

[![Review Change
Stack](https://storage.googleapis.com/coderabbit_public_assets/review-stack-in-coderabbit-ui.svg)](https://app.coderabbit.ai/change-stack/bootjp/elastickv/pull/784?utm_source=github_walkthrough&utm_medium=github&utm_campaign=change_stack)

<!-- review_stack_entry_end -->

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: bootjp

docs/design/2026_04_29_partial_data_at_rest_encryption.md

@@ -24,7 +24,7 @@ Date: 2026-04-29
 | 6C-2 | Process-local §9.1 guards that read only the encryption package's own state: `ErrLocalEpochExhausted` (any active DEK with `local_epoch == 0xFFFF` — refuses to start until rotated, preventing GCM nonce reuse on the next write), `ErrUnsupportedFilesystem` startup probe (`ProbeSidecarFilesystem` exercises the §5.1 write+rename+dir.Sync sequence on a sentinel file in the sidecar directory at startup so filesystem-capability failures surface BEFORE the first encryption-relevant Raft entry, not on the first WriteSidecar call hours into operation). Also includes the gemini-r3 medium parseErr-annotation fix deferred from PR #778 (`ErrKEKMismatch` now wraps both `parseErr` and the unwrap error when the defensive non-decimal-key_id branch fires). The asymmetric `ErrLocalEpochRollback` peer is split into 6C-3 because it requires the writer-registry record (Stage 7). The `ErrSidecarBehindRaftLog` guard moves to 6C-2b because it requires raftengine integration — outside the encryption package boundary. | open | — |
 | 6C-2b | `ErrSidecarBehindRaftLog` guard PRIMITIVE — encryption-side sentinel, `IsEncryptionRelevantOpcode` predicate (§5.5), `EncryptionRelevantScanner` cross-package interface, and `GuardSidecarBehindRaftLog` pure-function guard that takes both indices + a scanner callback. Tests use a fake scanner to exercise every branch (caught-up / gap-not-covered / gap-covered / scanner-error / nil-scanner-fails-closed). Wires nothing into main.go; the raftengine-side implementation lands in 6C-2c. | shipped-in-this-PR | this PR |
 | 6C-2c | Raftengine implementation of `EncryptionRelevantScanner` against `etcdraft.MemoryStorage.Entries`. `Engine.EncryptionScanner()` exposes the scanner as the encryption-side interface so callers (Stage 6C-2d, 6D, 6E) inherit it without coupling to the engine internals. Tests pin every branch: empty range, no-relevant entries (with OpRegistration explicitly), bootstrap hit, rotation hit, conf-change skipped, empty-data skipped, compacted-range error propagation, nil-storage error. | shipped-in-this-PR | this PR |
-| 6C-2d | `main.go` startup-guard phase that runs after `buildShardGroups` for each shard with encryption enabled, before any gRPC server starts serving. Calls `encryption.GuardSidecarBehindRaftLog` with the shard's engine `AppliedIndex()` + sidecar `RaftAppliedIndex` + the engine's `EncryptionScanner()` from 6C-2c. The guard's existing branches (caught-up / gap-not-covered / gap-covered / scanner-error / nil-scanner-fails-closed) determine refusal behavior. | open | — |
+| 6C-2d | `main.go` startup-guard phase that runs once on the default group after `buildShardGroups`, before any gRPC server starts serving. `checkSidecarBehindRaftLog` (`main_encryption_startup_guard.go`) reads the §5.1 sidecar, looks up the default-group runtime, type-asserts it to `encryptionGapEngine` (local interface satisfied by `internal/raftengine/etcd.Engine`), and calls `encryption.GuardSidecarBehindRaftLog` with the engine's `AppliedIndex()` + sidecar `RaftAppliedIndex` + engine's `EncryptionScanner()` from 6C-2c. Scope is restricted to the default group because only its FSM advances the sidecar's `raft_applied_index` (via `WriteSidecar` in `ApplyBootstrap` / `ApplyRotation`). The `chainEncryptionStartupGuard` helper keeps run()'s cyclop budget intact by composing the guard with the existing buildShardGroups error path. The companion change extends the `EncryptionApplier` interface so `ApplyBootstrap` / `ApplyRotation` receive the Raft entry index via a new `raftengine.ApplyIndexAware` seam, and `writeBootstrapSidecar` / `writeRotationSidecar` advance `sidecar.RaftAppliedIndex` inside the same crash-durable WriteSidecar fsync so the guard has a non-zero baseline to compare against on the next restart. A nil engine on a present default-group runtime is treated as a fatal init failure (fail-closed). | shipped-in-this-PR | this PR |
 | 6C-3 | Cluster-wide §9.1 guards that require the Voters ∪ Learners membership view: `ErrNodeIDCollision` (two members hash to the same 16-bit `node_id`), `ErrLocalEpochRollback` (sidecar `local_epoch` ≤ writer-registry record). Naturally bundles with the Stage 6D capability fan-out which already needs that membership view. | open | — |
 | 6C-4 | Phase-2-specific §9.1 guards: `ErrEnvelopeCutoverDivergence` (sidecar `raft_envelope_cutover_index` disagrees with snapshot header), `ErrEncryptionNotBootstrapped` (`enable-raft-envelope` before bootstrap), `ErrLocalEpochOutOfRange` on the wire side of `GetCapability` / `GetSidecarState`. Bundles with Stage 6E (`enable-raft-envelope` admin RPC + Phase-2 cutover). | open | — |
 | 6D | §6.6 `enable-storage-envelope` admin RPC + §7.1 Phase-1 storage cutover (§6.2 toggle ON) + Voters ∪ Learners capability gate (depends on 6B for mutator wiring AND 6C-1+6C-2 for §9.1 startup-refusal guards; bundles 6C-3 for the membership-view-dependent collision check — see rationale) | open | — |

internal/encryption/applier.go

@@ -332,7 +332,7 @@ func (a *Applier) ApplyRegistration(p fsmwire.RegistrationPayload) error {
 // pebble.Batch.Commit(pebble.Sync), but the current shape is
 // correct and matches the Stage 6A ApplyRegistration semantics
 // row-for-row.
-func (a *Applier) ApplyBootstrap(p fsmwire.BootstrapPayload) error {
+func (a *Applier) ApplyBootstrap(raftIdx uint64, p fsmwire.BootstrapPayload) error {
 	if err := a.validateBootstrap(p); err != nil {
 		return err
 	}
@@ -350,7 +350,7 @@ func (a *Applier) ApplyBootstrap(p fsmwire.BootstrapPayload) error {
 	if err := a.keystore.Set(p.RaftDEKID, raftDEK); err != nil {
 		return errors.Wrap(err, "applier: keystore set raft DEK")
 	}
-	if err := a.writeBootstrapSidecar(p); err != nil {
+	if err := a.writeBootstrapSidecar(raftIdx, p); err != nil {
 		return err
 	}
 	for i, reg := range p.BatchRegistry {
@@ -483,7 +483,7 @@ func (a *Applier) checkBootstrapWrappedMatches(sc *Sidecar, p fsmwire.BootstrapP
 // §4.1 case 1 first-seen invariant holds and the registry batch
 // inserts will record FirstSeen = LastSeen = 0 for the
 // proposing node.
-func (a *Applier) writeBootstrapSidecar(p fsmwire.BootstrapPayload) error {
+func (a *Applier) writeBootstrapSidecar(raftIdx uint64, p fsmwire.BootstrapPayload) error {
 	sc, err := ReadSidecar(a.sidecarPath)
 	if err != nil && !IsNotExist(err) {
 		return errors.Wrap(err, "applier: read sidecar for bootstrap")
@@ -496,6 +496,7 @@ func (a *Applier) writeBootstrapSidecar(p fsmwire.BootstrapPayload) error {
 	}
 	sc.Active.Storage = p.StorageDEKID
 	sc.Active.Raft = p.RaftDEKID
+	advanceRaftAppliedIndex(sc, raftIdx)
 	createdAt := a.now().UTC().Format(time.RFC3339)
 	sc.Keys[strconv.FormatUint(uint64(p.StorageDEKID), 10)] = SidecarKey{
 		Purpose:    SidecarPurposeStorage,
@@ -515,6 +516,38 @@ func (a *Applier) writeBootstrapSidecar(p fsmwire.BootstrapPayload) error {
 	return nil
 }
 
+// advanceRaftAppliedIndex sets sc.RaftAppliedIndex to raftIdx when
+// raftIdx is non-zero AND strictly greater than the current value.
+//
+// Why monotonic-and-skip-zero (instead of unconditional assignment):
+//
+//   - Zero means "caller did not supply a Raft entry index" (the
+//     raftengine.ApplyIndexAware seam was not wired, e.g. unit
+//     tests that drive ApplyBootstrap directly without setting
+//     pendingApplyIdx). Overwriting a previously-good index with
+//     zero would silently regress the sidecar's freshness marker
+//     and the §9.1 ErrSidecarBehindRaftLog guard would over-fire
+//     on the next restart. Skip-on-zero preserves Stage-6A
+//     behavior for those callers.
+//
+//   - Monotonic guards against a malformed replay (or buggy engine)
+//     handing a strictly-lower index after a successful Bootstrap.
+//     The sidecar must only ever record "the last Raft index we
+//     have observed an encryption-relevant entry for"; once
+//     advanced, regression is a divergence signal we choose to
+//     swallow rather than HaltApply (the apply itself is still
+//     correct; only the sidecar's freshness annotation would be
+//     wrong, and the guard treats sidecar < engine as the harmless
+//     direction — it just runs the scanner instead of fast-pass).
+func advanceRaftAppliedIndex(sc *Sidecar, raftIdx uint64) {
+	if raftIdx == 0 {
+		return
+	}
+	if raftIdx > sc.RaftAppliedIndex {
+		sc.RaftAppliedIndex = raftIdx
+	}
+}
+
 // ApplyRotation implements §5.2 / §5.4 rotation apply. Stage 6B-1
 // handles only the RotateSubRotateDEK sub-tag (rotate the storage
 // or raft DEK to a new key_id). Other sub-tags (rewrap-deks,
@@ -537,7 +570,7 @@ func (a *Applier) writeBootstrapSidecar(p fsmwire.BootstrapPayload) error {
 // Same WithKEK / WithKeystore / WithSidecarPath trio requirement
 // as ApplyBootstrap; partial wiring returns ErrKEKNotConfigured
 // at apply time.
-func (a *Applier) ApplyRotation(p fsmwire.RotationPayload) error {
+func (a *Applier) ApplyRotation(raftIdx uint64, p fsmwire.RotationPayload) error {
 	if !a.bootstrapAndRotationConfigured() {
 		return errors.Wrap(ErrKEKNotConfigured, "applier: rotation requires WithKEK + WithKeystore + WithSidecarPath")
 	}
@@ -564,7 +597,7 @@ func (a *Applier) ApplyRotation(p fsmwire.RotationPayload) error {
 	if err := a.keystore.Set(p.DEKID, dek); err != nil {
 		return errors.Wrap(err, "applier: keystore set rotation DEK")
 	}
-	if err := a.writeRotationSidecar(p); err != nil {
+	if err := a.writeRotationSidecar(raftIdx, p); err != nil {
 		return err
 	}
 	if err := a.ApplyRegistration(p.ProposerRegistration); err != nil {
@@ -578,7 +611,7 @@ func (a *Applier) ApplyRotation(p fsmwire.RotationPayload) error {
 // p.DEKID, then crash-durably writes. Existing keys[] entries are
 // preserved (rotation does not retire old DEKs — that is a
 // separate sub-tag in Stage 6E).
-func (a *Applier) writeRotationSidecar(p fsmwire.RotationPayload) error {
+func (a *Applier) writeRotationSidecar(raftIdx uint64, p fsmwire.RotationPayload) error {
 	sc, err := ReadSidecar(a.sidecarPath)
 	if err != nil {
 		return errors.Wrap(err, "applier: read sidecar for rotation")
@@ -596,6 +629,7 @@ func (a *Applier) writeRotationSidecar(p fsmwire.RotationPayload) error {
 	case fsmwire.PurposeRaft:
 		sc.Active.Raft = p.DEKID
 	}
+	advanceRaftAppliedIndex(sc, raftIdx)
 	sc.Keys[strconv.FormatUint(uint64(p.DEKID), 10)] = SidecarKey{
 		Purpose:    purpose,
 		Wrapped:    append([]byte(nil), p.Wrapped...),