diff --git a/docs/design/2026_04_29_partial_data_at_rest_encryption.md b/docs/design/2026_04_29_partial_data_at_rest_encryption.md index e4d6ca906..0642e5a13 100644 --- a/docs/design/2026_04_29_partial_data_at_rest_encryption.md +++ b/docs/design/2026_04_29_partial_data_at_rest_encryption.md @@ -1,6 +1,6 @@ # Data-at-rest encryption for elastickv -Status: Partial — Stages 0–8, 9A–9B, 9C-1 through 9C-4, 9C-6, and 9D shipped (5E deferred); remaining Stage 9 work open +Status: Partial — Stages 0–8, 9A–9B, 9C-1 through 9C-6, and 9D shipped (5E deferred); remaining Stage 9 work open Author: bootjp Date: 2026-04-29 @@ -38,6 +38,7 @@ Date: 2026-04-29 | 9C-2 | Sidecar/KEK observability: `active_dek_id{purpose}`, `sidecar_raft_index`, `kek_unwrap_seconds` (§9.2) | shipped | — | | 9C-3 | Startup KEK-unwrap memoization: collapses the guard + hydration double-unwrap that Stage 9B's KMS providers turned into doubled startup round-trips | shipped | — | | 9C-4 | §5.2 writes-per-DEK rotation budget: the 2^32 per-(DEK, process-load) ceiling, the 90% refuse-and-rotate threshold, and per-DEK accounting. The §9.2 `writes_per_dek` counter ships separately in 9C-1, from the storage-envelope path; wiring this budget to the same accounting is part of the open row below. | shipped | this PR | +| 9C-5 | §5.4 DEK retirement eligibility classifiers: storage references and a cluster-wide write-fence generation, plus WAL-driven raft criteria evaluated per Raft group. Reports are bound to the retiring key and must show a different key active. Cluster-wide, no override. Classification only; report collection, fence lifecycle management, and retire-command execution remain open. | shipped | this PR | | 9D | Encrypted Jepsen acceptance gate (§8.4): `elastickv-admin encryption wrap-dek` (local KEK-wrap helper so an operator or a script can produce the `bootstrap` / `rotate-dek` material without hand-rolling AES-GCM), `scripts/jepsen-enable-encryption.sh` (bootstrap → `enable-storage-envelope` → `enable-raft-envelope` → wait for every member), `scripts/run-jepsen-local.sh --encrypted`, and an `encryption: [off, on]` matrix on the per-push `Jepsen Test` workflow so every Redis / DynamoDB / S3 / SQS / filesystem workload also runs against an encrypted 3-node cluster. | shipped | this PR | | 9C-6 | §5.4 rewrite-job decisions: `ClassifyMVCCRewrite` (which MVCC versions a rotation must re-encrypt, iterating `(user_key, version_ts)` rather than `user_key`, so history under the retiring DEK is not left behind) and `RewriteThrottle` (the `--rate=N MiB/s` inter-batch yield, using a virtual deadline that advances for every batch so a stalled job cannot bank idle time and burst). Decision layer only — no store, no clock, no Raft. | shipped | — | | 9C+ | Rotation rewrap/retire, the rewrite job's execution half, admission-control and auto-propose wiring for the budget above, `last_proposed_index_per_raft_dek` (needs the §5.4 raft-DEK Wrap path), and remaining benchmarks (§5.2, §5.4, §6.5, §8.3, §9.2) | open | — | @@ -1250,12 +1251,27 @@ that DEK is unloaded. The rewrite must therefore be MVCC-aware: — overriding is silently equivalent to "lose data on the next read or replay." - **Storage-purpose DEKs** must satisfy both of: + **Storage-purpose DEKs** must satisfy all of: + - A cluster-wide, non-zero retiring-key write-fence token is + installed and remains held through the `retire-dek` apply that + unloads the key. Every node reports under that same token, and + every write that captured the retiring key before the fence has + drained. This closes the window where a pre-rotation write can + commit old-key ciphertext after the eligibility scan. - The rewrite cursor for that DEK has reached the end of the keyspace AND `elastickv_encryption_values_per_dek{key_id} == 0` across every node (verified by `encryption status --verify`). + The count requires a completed value scan; a missing or partial + scan is incomplete evidence even when its numeric count is zero. + - No persisted rewrite cursor envelope references the retiring + DEK. Cursor records are counted separately from rewritten user + values because a stalled job may resume after the key is unloaded. + - No retained or otherwise restorable FSM snapshot contains a + storage envelope under the retiring DEK. The snapshot inventory + records per-DEK references; a post-rewrite snapshot alone is not + sufficient until every older restorable snapshot is purged. - `minRetainedTS` on every node is greater than the largest `commit_ts` ever written under the retiring DEK. Until this holds, a snapshot or lease read can still @@ -1265,8 +1281,9 @@ that DEK is unloaded. The rewrite must therefore be MVCC-aware: **Raft-purpose DEKs** must satisfy a different, WAL-driven criterion because retaining MVCC versions does not cover their use sites — the raft envelope ciphertext - lives in the etcd raft WAL and in raft snapshots, not in - Pebble values. Specifically: + lives in the etcd raft WAL, not in the FSM snapshots or Pebble + values. Snapshot indexes still matter because they bound the WAL + range replayed during restore. Specifically: - Every node's persisted Raft log start index (the lower bound of un-truncated entries; advances with snapshot installs and `wal.purge`) must be **strictly greater @@ -1276,12 +1293,26 @@ that DEK is unloaded. The rewrite must therefore be MVCC-aware: `elastickv_encryption_last_proposed_index_per_raft_dek{key_id}`, updated on every leader Wrap. The former is exposed as `etcd_raft_log_compact_index` per node. - - Every node's last committed Raft snapshot must have - been **taken under the new raft DEK** (i.e., its FSM - snapshot header §4.4 carries - `raft_envelope_cutover_index` past the rotation entry). - A node restored from an older snapshot would replay - entries that still need the retiring DEK to decode. + - Every node's last committed Raft snapshot must be **at or past + that group's largest index proposed under the retiring DEK**. A + node restored from an older snapshot would replay entries that + still need the retiring DEK to decode. Non-default groups have + no rotation entry in their own index space, so the per-group + old-key high-water mark is the boundary to check. The signal is + the snapshot's own Raft index (`raftpb.SnapshotMetadata.Index`), + which advances with every snapshot install. + + The raft criteria are evaluated per Raft group. Log and snapshot + indexes live in independent per-group index spaces while the raft + DEK is cluster-wide and installed on every attached `ShardGroup`. + Every report and high-water boundary names the retiring key, every + node and group must be covered exactly once, and a different + successor key must be active everywhere. The one-shot + `raft_envelope_cutover_index` is not a retirement boundary: it is + frozen at Phase-2 enablement and cannot describe later rotations. + Each group's high-water mark must be explicitly observed. An absent + or incomplete proposal history cannot use a default zero boundary; + an observed zero denotes a key that was never used in that group. Without the WAL guard, retiring a raft DEK while the WAL still contains entries encrypted under it would cause diff --git a/internal/encryption/dek_retirement.go b/internal/encryption/dek_retirement.go new file mode 100644 index 000000000..f94848674 --- /dev/null +++ b/internal/encryption/dek_retirement.go @@ -0,0 +1,605 @@ +package encryption + +import ( + "fmt" + "slices" + "sort" + + "github.com/cockroachdb/errors" +) + +// §5.4 DEK retirement eligibility. +// +// Retiring a DEK unloads it. If any ciphertext anywhere still needs it, +// that data becomes unreadable — so the design states plainly that +// there is **no override flag**, because overriding "is silently +// equivalent to 'lose data on the next read or replay.'" This package +// therefore exposes only a classifier: every criterion must be +// satisfied cluster-wide, and there is no way to ask it to say yes +// anyway. +// +// The two purposes have genuinely different criteria because their +// ciphertext lives in different places — Pebble values for storage, +// the etcd raft WAL for raft — so retaining MVCC +// versions does not cover the raft case at all. + +// ErrDEKStillReferenced reports that a storage DEK cannot be retired +// yet: either the rewrite has not finished, or a snapshot/lease read +// could still legitimately ask for a version written under it. +var ErrDEKStillReferenced = errors.New("encryption: storage DEK is still referenced; refusing to retire") + +// ErrRaftDEKWALStillReferences is the §5.4 sentinel for retiring a raft +// DEK early. +// +// Without the WAL guard, unloading a raft DEK while the WAL still holds +// entries encrypted under it causes unknown_key_id apply failures on +// the next restart (replay from disk) or on a lagging follower's +// catch-up — the exact failure mode the writer registry and capability +// gate exist to prevent. +var ErrRaftDEKWALStillReferences = errors.New("encryption: raft WAL still references this DEK; refusing to retire") + +// ErrIncompleteRetirementReport reports that eligibility could not be +// established for the whole cluster. +// +// Distinct from "not yet eligible": the criteria are cluster-wide, so a +// missing node is not a node that passed. Treating an absent report as +// satisfied is how a single unreachable replica ends up replaying +// against an unloaded DEK. +var ErrIncompleteRetirementReport = errors.New("encryption: retirement report does not cover every cluster member") + +// ErrRetiringDEKStillActive reports an attempt to retire the DEK that is +// still selected for writes or proposals. +// +// Its own sentinel because it is not "not yet eligible": every retention +// criterion can legitimately pass for the ACTIVE key -- an empty cluster +// reports a complete rewrite cursor, zero values and an advanced retention +// floor for the key it is still writing under -- and unloading it breaks the +// next write or proposal rather than an old read. +var ErrRetiringDEKStillActive = errors.New( + "encryption: refusing to retire the DEK that is still active") + +// StorageRetirementReport is one node's view of a storage DEK. +type StorageRetirementReport struct { + NodeID string + // ReportedKeyID is the DEK this report describes. Checked against the + // key being retired: a report gathered for a different key says nothing + // about this one, and accepting it silently would judge the wrong DEK. + ReportedKeyID uint32 + // ActiveKeyID is the storage DEK this node is currently writing under. + // A successor must be active and must differ from the retiring key. + ActiveKeyID uint32 + // RewriteCursorComplete is true when the rewrite job has reached + // the end of the keyspace on this node. + RewriteCursorComplete bool + // ValuesPerDEK is elastickv_encryption_values_per_dek{key_id} on + // this node; zero is required. + ValuesPerDEK uint64 + // ValuesPerDEKObserved is true only after a complete scan of retained + // values; an unavailable or partial scan cannot establish a zero count. + ValuesPerDEKObserved bool + // MinRetainedTS is this node's MVCC retention floor. It must have + // advanced past every commit_ts written under the retiring DEK, + // or a snapshot read can still legitimately ask for one of those + // versions. + MinRetainedTS uint64 + // WriteFenceToken identifies the cluster-wide retiring-key write fence + // under which this report was collected. The fence prevents new writes + // from capturing the retiring key and must remain held through the + // retire-dek apply that unloads it. + WriteFenceToken uint64 + // InFlightOldKeyWrites counts writes that captured the retiring key + // before the fence and have not finished committing. It must be zero. + InFlightOldKeyWrites uint64 + // InFlightWriteDrainObserved distinguishes a measured zero from an + // unpopulated counter. + InFlightWriteDrainObserved bool + // RewriteCursorReferences counts persisted rewrite cursors whose own + // envelope still names the retiring key. It must be zero so a later + // rewrite resume does not need an unloaded key. + RewriteCursorReferences uint64 + // RewriteCursorReferencesObserved distinguishes a complete cursor scan + // with zero references from missing evidence. + RewriteCursorReferencesObserved bool + // RetainedSnapshotReferences counts retained or otherwise restorable FSM + // snapshots whose storage envelopes still name the retiring key. It must + // be zero because restoring one would reintroduce unreadable values. + RetainedSnapshotReferences uint64 + // RetainedSnapshotReferencesObserved distinguishes a complete snapshot + // inventory with zero references from missing evidence. + RetainedSnapshotReferencesObserved bool +} + +// StorageRetirementBoundary is the retiring key's collected commit high-water +// mark. Observed distinguishes a genuinely unused key from a missing metric +// whose zero value would otherwise look safe. +type StorageRetirementBoundary struct { + KeyID uint32 + LargestCommitTS uint64 + Observed bool + // WriteFenceToken is the non-zero token assigned when the coordinator + // installed the retiring-key write fence on every member. Reports must + // echo this token, binding all sampled evidence to the same live fence. + WriteFenceToken uint64 +} + +// RaftGroupRetirementReport is one node's view of ONE Raft group. +// +// Per group, because Raft log and snapshot indexes live in independent +// per-group index spaces while the raft DEK is cluster-wide: the runtime +// installs one wrap on every attached ShardGroup +// (raftEnvelopeRuntime.installFromApply loops over r.groups). Judging the +// shared DEK from a single group's indexes leaves every other group +// unchecked, and aggregating across groups with a minimum is worse still -- +// a quiet low-index group stays permanently below a busy group's boundary +// even after all of its old-key entries are gone, so the DEK could never be +// retired. +type RaftGroupRetirementReport struct { + GroupID uint64 + // LogCompactIndex is this node's persisted Raft log start index for + // THIS group -- the lower bound of un-truncated entries, exposed as + // etcd_raft_log_compact_index. It must be STRICTLY greater than the + // largest index ever proposed under the retiring DEK in this group. + LogCompactIndex uint64 + // SnapshotIndex is the Raft index of this group's last committed + // snapshot. It is the snapshot's own index rather than the one-shot + // raft_envelope_cutover_index: a restore replays later WAL entries. + SnapshotIndex uint64 +} + +// RaftGroupBoundary is the old-key high-water mark for one Raft group. +type RaftGroupBoundary struct { + // Observed distinguishes a measured high-water mark (including an unused + // key's zero) from an unavailable or incomplete proposal history. + Observed bool + // KeyID is the DEK this boundary describes. Checked against the key + // being retired for the same reason the reports are: without it, a + // boundary collected for a different DEK -- key 8's lower + // LargestProposedIndex while key 7 is being retired -- would be accepted + // by reports that all correctly name key 7, and the classifier would + // return eligible with key 7 entries still replayable. Mixing per-key + // metric series has to fail closed. + KeyID uint32 + // LargestProposedIndex is the largest log index ever proposed under the + // retiring DEK in this group. + // + // There is deliberately no per-group rotation index. The rotation is + // proposed once, through the DEFAULT group's engine + // (main_encryption_admin.go), and its apply handler installs the new + // wrapper on every attached group in memory + // (raftEnvelopeRuntime.installRotatedRaftDEK). A non-default group has + // no rotation entry in its own index space at all, so requiring one + // would force the report collector to invent a number and make the + // verdict arbitrary. This index is the quantity that does exist per + // group, and it is the one the snapshot criterion actually needs: + // restoring from a snapshot below it replays entries proposed under the + // retiring DEK. + LargestProposedIndex uint64 +} + +// RaftRetirementReport is one node's view of a raft DEK. +type RaftRetirementReport struct { + NodeID string + // ReportedKeyID is the DEK this report describes, checked against the + // key being retired. + ReportedKeyID uint32 + // ActiveKeyID is the raft DEK this node currently proposes under. + ActiveKeyID uint32 + // Groups is this node's per-group view. Every group the node hosts must + // be present, because the DEK is shared across all of them. + Groups []RaftGroupRetirementReport +} + +// RetirementDecision is the classifier's answer. +type RetirementDecision struct { + // Purpose records which classifier produced this decision, so Err can + // pick the matching sentinel without being told. + Purpose string + // Eligible is true only when every criterion holds on every node. + Eligible bool + // Blockers names each node that is not yet ready and why, so an + // operator can see which replica to wait on rather than being told + // only that the cluster is not ready. + Blockers []string +} + +// Err returns the sentinel matching this decision, or nil when eligible. +// Callers surface this from `retire-dek`. +// +// The sentinel comes from the CLASSIFIER that produced the decision, recorded +// in Purpose, not from an argument. Selecting it from a caller-supplied string +// meant a misspelled or omitted purpose silently reported a blocked raft +// decision as ErrDEKStillReferenced -- sending the operator to the +// rewrite/MVCC remediation for a WAL blocker, and vice versa. +func (d RetirementDecision) Err() error { + if d.Eligible { + return nil + } + switch d.Purpose { + case RetirementPurposeStorage: + return errors.Wrapf(ErrDEKStillReferenced, "blockers: %v", d.Blockers) + case RetirementPurposeRaft: + return errors.Wrapf(ErrRaftDEKWALStillReferences, "blockers: %v", d.Blockers) + default: + // Unreachable through the classifiers, which always set Purpose. A + // zero-value decision reaching here is a wiring bug, and guessing a + // sentinel would point the operator at the wrong remediation. + return errors.Wrapf(ErrIncompleteRetirementReport, + "decision has no purpose; blockers: %v", d.Blockers) + } +} + +// Purposes accepted by the classifier. +const ( + RetirementPurposeStorage = "storage" + RetirementPurposeRaft = "raft" +) + +// ClassifyStorageDEKRetirement applies the §5.4 storage criteria. +// +// boundary is the collected commit high-water mark for the retiring DEK. +// members is the full cluster membership; a report is +// required from each, because "cluster-wide" cannot be established +// from a subset. +func ClassifyStorageDEKRetirement( + members []string, reports []StorageRetirementReport, retiringKeyID uint32, boundary StorageRetirementBoundary, +) (RetirementDecision, error) { + if retiringKeyID == 0 { + return RetirementDecision{}, errors.Wrap(ErrIncompleteRetirementReport, + "retiring key id is required") + } + if !boundary.Observed || boundary.KeyID != retiringKeyID { + return RetirementDecision{}, errors.Wrapf(ErrIncompleteRetirementReport, + "storage boundary missing or describes key %d, not retiring key %d", + boundary.KeyID, retiringKeyID) + } + if boundary.WriteFenceToken == 0 { + return RetirementDecision{}, errors.Wrap(ErrIncompleteRetirementReport, + "storage retirement write fence token is required") + } + byNode := make(map[string]StorageRetirementReport, len(reports)) + for _, r := range reports { + if _, dup := byNode[r.NodeID]; dup { + return RetirementDecision{}, duplicateReportErr(r.NodeID) + } + byNode[r.NodeID] = r + } + if err := requireExactNodeCoverage(members, byNode); err != nil { + return RetirementDecision{}, err + } + + blockers, err := storageRetirementBlockers(members, byNode, retiringKeyID, boundary) + if err != nil { + return RetirementDecision{}, err + } + sort.Strings(blockers) + return RetirementDecision{ + Purpose: RetirementPurposeStorage, + Eligible: len(blockers) == 0, + Blockers: blockers, + }, nil +} + +func storageRetirementBlockers( + members []string, byNode map[string]StorageRetirementReport, + retiringKeyID uint32, boundary StorageRetirementBoundary, +) ([]string, error) { + var blockers []string + for _, node := range members { + r := byNode[node] + if err := validateStorageRetirementEvidence(node, r, retiringKeyID, boundary); err != nil { + return nil, err + } + blockers = append(blockers, storageNodeBlockers(node, r, boundary.LargestCommitTS)...) + } + return blockers, nil +} + +func validateStorageRetirementEvidence( + node string, r StorageRetirementReport, retiringKeyID uint32, boundary StorageRetirementBoundary, +) error { + if err := checkReportKeyBinding(node, r.ReportedKeyID, r.ActiveKeyID, retiringKeyID); err != nil { + return err + } + if r.WriteFenceToken != boundary.WriteFenceToken { + return errors.Wrapf(ErrIncompleteRetirementReport, + "node %s reported write fence token %d, expected %d", + node, r.WriteFenceToken, boundary.WriteFenceToken) + } + if !r.ValuesPerDEKObserved { + return missingStorageEvidenceErr(node, "old-key value scan") + } + if !r.InFlightWriteDrainObserved { + return missingStorageEvidenceErr(node, "in-flight old-key write drain") + } + if !r.RewriteCursorReferencesObserved { + return missingStorageEvidenceErr(node, "rewrite cursor reference scan") + } + if !r.RetainedSnapshotReferencesObserved { + return missingStorageEvidenceErr(node, "retained snapshot reference inventory") + } + return nil +} + +func storageNodeBlockers(node string, r StorageRetirementReport, largestCommitTS uint64) []string { + var blockers []string + if !r.RewriteCursorComplete { + blockers = append(blockers, + fmt.Sprintf("%s: rewrite cursor has not reached the end of the keyspace", node)) + } + if r.ValuesPerDEK != 0 { + blockers = append(blockers, + fmt.Sprintf("%s: %d values still encrypted under this DEK", node, r.ValuesPerDEK)) + } + if r.InFlightOldKeyWrites != 0 { + blockers = append(blockers, + fmt.Sprintf("%s: %d retiring-key writes are still in flight", node, r.InFlightOldKeyWrites)) + } + if r.RewriteCursorReferences != 0 { + blockers = append(blockers, + fmt.Sprintf("%s: %d rewrite cursors still reference this DEK", node, r.RewriteCursorReferences)) + } + if r.RetainedSnapshotReferences != 0 { + blockers = append(blockers, + fmt.Sprintf("%s: %d retained snapshots still reference this DEK", node, r.RetainedSnapshotReferences)) + } + // Equality still admits a read at exactly the last old-key version. + if largestCommitTS != 0 && r.MinRetainedTS <= largestCommitTS { + blockers = append(blockers, + fmt.Sprintf("%s: minRetainedTS %d has not advanced past commit_ts %d", + node, r.MinRetainedTS, largestCommitTS)) + } + return blockers +} + +func missingStorageEvidenceErr(node, evidence string) error { + return errors.Wrapf(ErrIncompleteRetirementReport, + "node %s did not report a completed %s", node, evidence) +} + +// ClassifyRaftDEKRetirement applies the §5.4 raft criteria. +// +// boundaries contains an observed, key-bound high-water mark for every group. +func ClassifyRaftDEKRetirement( + members []string, reports []RaftRetirementReport, + retiringKeyID uint32, boundaries map[uint64]RaftGroupBoundary, +) (RetirementDecision, error) { + if retiringKeyID == 0 { + return RetirementDecision{}, errors.Wrap(ErrIncompleteRetirementReport, + "retiring key id is required") + } + if len(boundaries) == 0 { + // No boundaries means no group's old-key high-water mark is known, + // so nothing can be verified. "Nothing to check" is not "safe". + return RetirementDecision{}, errors.Wrap(ErrIncompleteRetirementReport, + "per-group raft boundaries are required") + } + byNode := make(map[string]RaftRetirementReport, len(reports)) + for _, r := range reports { + if _, dup := byNode[r.NodeID]; dup { + return RetirementDecision{}, duplicateReportErr(r.NodeID) + } + byNode[r.NodeID] = r + } + if err := requireExactNodeCoverage(members, byNode); err != nil { + return RetirementDecision{}, err + } + + var blockers []string + for _, node := range members { + r := byNode[node] + if err := checkReportKeyBinding(node, r.ReportedKeyID, r.ActiveKeyID, retiringKeyID); err != nil { + return RetirementDecision{}, err + } + nodeBlockers, err := raftGroupBlockers(node, r.Groups, boundaries, retiringKeyID) + if err != nil { + return RetirementDecision{}, err + } + blockers = append(blockers, nodeBlockers...) + } + sort.Strings(blockers) + return RetirementDecision{ + Purpose: RetirementPurposeRaft, + Eligible: len(blockers) == 0, + Blockers: blockers, + }, nil +} + +// checkReportKeyBinding rejects a report that does not describe the DEK being +// retired, or that shows the retiring DEK still active. +// +// Both are fatal rather than blockers: a report for another key is not +// evidence about this one, and "still active" is a different operator mistake +// from "not yet eligible" -- every retention criterion can pass for the active +// key, and unloading it breaks the next write rather than an old read. +func checkReportKeyBinding(node string, reported, active, retiring uint32) error { + if reported != retiring { + return errors.Wrapf(ErrIncompleteRetirementReport, + "node %s reported on key %d, not the key being retired (%d)", + node, reported, retiring) + } + if active == retiring { + return errors.Wrapf(ErrRetiringDEKStillActive, + "node %s still has key %d active", node, retiring) + } + if active == 0 { + return errors.Wrapf(ErrRetiringDEKStillActive, + "node %s reports no active successor key", node) + } + return nil +} + +// raftGroupBlockers applies the §5.4 raft criteria to every group the node +// hosts, against that group's own boundary. +// +// A node that omits a group with a known boundary is an incomplete report, not +// a passing one: the DEK is installed on every group, so an unreported group +// is an unverified one. +// indexReportedGroups turns a node's per-group reports into a map, rejecting a +// node that reported the same group twice: last-write-wins over a duplicate +// would let a group's blocking report be overwritten by a passing one. +func indexReportedGroups( + node string, + groups []RaftGroupRetirementReport, +) (map[uint64]RaftGroupRetirementReport, error) { + byGroup := make(map[uint64]RaftGroupRetirementReport, len(groups)) + for _, g := range groups { + if _, dup := byGroup[g.GroupID]; dup { + return nil, errors.Wrapf(ErrIncompleteRetirementReport, + "node %s reported group %d more than once", node, g.GroupID) + } + byGroup[g.GroupID] = g + } + return byGroup, nil +} + +// requireExactGroupCoverage checks BOTH directions between the reports and the +// boundaries. +// +// A boundary with no report is an unverified group: the DEK is installed on +// every group, so a group nobody reported is one nobody checked. A report with +// no boundary is the same hole seen from the other side -- checking only the +// first treats the boundary map as the complete group universe, so if +// collection omits a group while every node reports it with old-key entries, +// that group is judged by nobody and the classifier returns eligible. +func requireExactGroupCoverage( + node string, + byGroup map[uint64]RaftGroupRetirementReport, + boundaries map[uint64]RaftGroupBoundary, +) error { + var missing []uint64 + for groupID := range boundaries { + if _, ok := byGroup[groupID]; !ok { + missing = append(missing, groupID) + } + } + if len(missing) > 0 { + slices.Sort(missing) + return errors.Wrapf(ErrIncompleteRetirementReport, + "node %s did not report groups %v", node, missing) + } + + var unbounded []uint64 + for groupID := range byGroup { + if _, ok := boundaries[groupID]; !ok { + unbounded = append(unbounded, groupID) + } + } + if len(unbounded) > 0 { + slices.Sort(unbounded) + return errors.Wrapf(ErrIncompleteRetirementReport, + "node %s reported groups %v with no boundary", node, unbounded) + } + return nil +} + +// groupBlockers applies the §5.4 raft criteria to ONE group. +func groupBlockers( + node string, + groupID uint64, + g RaftGroupRetirementReport, + boundary RaftGroupBoundary, +) []string { + var blockers []string + // Strictly greater, per §5.4: an index EQUAL to the largest proposed one + // means that entry is still un-truncated and would be replayed. + if g.LogCompactIndex <= boundary.LargestProposedIndex { + blockers = append(blockers, + fmt.Sprintf("%s group %d: raft log start index %d has not passed proposed index %d", + node, groupID, g.LogCompactIndex, boundary.LargestProposedIndex)) + } + if g.SnapshotIndex < boundary.LargestProposedIndex { + blockers = append(blockers, + fmt.Sprintf("%s group %d: last snapshot predates the old-key high-water mark (snapshot %d < proposed %d)", + node, groupID, g.SnapshotIndex, boundary.LargestProposedIndex)) + } + return blockers +} + +// raftGroupBlockers applies the §5.4 raft criteria to every group the node +// hosts, against that group's own boundary. +func raftGroupBlockers( + node string, + groups []RaftGroupRetirementReport, + boundaries map[uint64]RaftGroupBoundary, + retiringKeyID uint32, +) ([]string, error) { + byGroup, err := indexReportedGroups(node, groups) + if err != nil { + return nil, err + } + if err := requireExactGroupCoverage(node, byGroup, boundaries); err != nil { + return nil, err + } + + groupIDs := make([]uint64, 0, len(boundaries)) + for groupID := range boundaries { + groupIDs = append(groupIDs, groupID) + } + slices.Sort(groupIDs) + + var blockers []string + for _, groupID := range groupIDs { + boundary := boundaries[groupID] + if !boundary.Observed { + return nil, errors.Wrapf(ErrIncompleteRetirementReport, + "group %d proposal high-water mark was not observed", groupID) + } + if boundary.KeyID != retiringKeyID { + return nil, errors.Wrapf(ErrIncompleteRetirementReport, + "group %d boundary describes key %d, not the retiring key %d", + groupID, boundary.KeyID, retiringKeyID) + } + blockers = append(blockers, groupBlockers(node, groupID, byGroup[groupID], boundary)...) + } + return blockers, nil +} + +// duplicateReportErr rejects a report set containing the same node +// twice. +// +// Collapsing duplicates into a map is last-write-wins, so a node that +// reported a blocker and then reported ready would be recorded as +// ready, and coverage would still be complete because the map holds one +// entry per unique node. Eligibility would then depend on the order the +// reports arrived. Since the whole point of this classifier is that +// unloading a DEK with any live reference loses data, an ambiguous +// report set fails closed instead. +func duplicateReportErr(nodeID string) error { + return errors.Wrapf(ErrIncompleteRetirementReport, + "node %s reported more than once", nodeID) +} + +// requireExactNodeCoverage compares the membership snapshot and reports in +// both directions. A report from a node omitted by membership is not safe to +// ignore: it may contain the only old-key blocker. +func requireExactNodeCoverage[T any](members []string, byNode map[string]T) error { + if len(members) == 0 { + return errors.Wrap(ErrIncompleteRetirementReport, "cluster membership is empty") + } + memberSet := make(map[string]struct{}, len(members)) + var missing []string + for _, node := range members { + if _, duplicate := memberSet[node]; duplicate { + return errors.Wrapf(ErrIncompleteRetirementReport, "member %s listed more than once", node) + } + memberSet[node] = struct{}{} + if _, ok := byNode[node]; !ok { + missing = append(missing, node) + } + } + if len(missing) > 0 { + sort.Strings(missing) + return errors.Wrapf(ErrIncompleteRetirementReport, "no report from %v", missing) + } + var extra []string + for node := range byNode { + if _, ok := memberSet[node]; !ok { + extra = append(extra, node) + } + } + if len(extra) > 0 { + sort.Strings(extra) + return errors.Wrapf(ErrIncompleteRetirementReport, "reports from nonmembers %v", extra) + } + return nil +} diff --git a/internal/encryption/dek_retirement_test.go b/internal/encryption/dek_retirement_test.go new file mode 100644 index 000000000..8bd7ae93e --- /dev/null +++ b/internal/encryption/dek_retirement_test.go @@ -0,0 +1,916 @@ +package encryption_test + +import ( + "testing" + + "github.com/bootjp/elastickv/internal/encryption" + "github.com/cockroachdb/errors" + "github.com/stretchr/testify/require" +) + +var retireMembers = []string{"n1", "n2", "n3"} + +// retiringKeyID / successorKeyID: the reports must name the key under test and +// show a different key active, or the classifier refuses outright. +const ( + retiringKeyID = uint32(7) + successorKeyID = uint32(8) + writeFenceToken = uint64(41) +) + +// retireGroup is the single Raft group most cases use; multi-group behaviour +// has its own tests. +const retireGroup = uint64(1) + +func readyStorage(node string, minRetained uint64) encryption.StorageRetirementReport { + return encryption.StorageRetirementReport{ + NodeID: node, + ReportedKeyID: retiringKeyID, + ActiveKeyID: successorKeyID, + RewriteCursorComplete: true, + ValuesPerDEK: 0, + ValuesPerDEKObserved: true, + MinRetainedTS: minRetained, + WriteFenceToken: writeFenceToken, + InFlightWriteDrainObserved: true, + RewriteCursorReferencesObserved: true, + RetainedSnapshotReferencesObserved: true, + } +} + +func classifyStorage(members []string, reports []encryption.StorageRetirementReport, keyID uint32, largestCommitTS uint64) (encryption.RetirementDecision, error) { + return encryption.ClassifyStorageDEKRetirement(members, reports, keyID, + encryption.StorageRetirementBoundary{ + KeyID: keyID, LargestCommitTS: largestCommitTS, Observed: true, + WriteFenceToken: writeFenceToken, + }) +} + +func TestStorageRetirementRejectsUnboundHighWaterMark(t *testing.T) { + t.Parallel() + reports := []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), readyStorage("n3", 200), + } + for _, boundary := range []encryption.StorageRetirementBoundary{ + {KeyID: retiringKeyID + 1, LargestCommitTS: 100, Observed: true, WriteFenceToken: writeFenceToken}, + {KeyID: retiringKeyID, LargestCommitTS: 0, Observed: false, WriteFenceToken: writeFenceToken}, + } { + _, err := encryption.ClassifyStorageDEKRetirement(retireMembers, reports, retiringKeyID, boundary) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + } + d, err := classifyStorage(retireMembers, reports, retiringKeyID, 0) + require.NoError(t, err) + require.True(t, d.Eligible, "an observed zero high-water mark means this key was unused") + d, err = classifyStorage(retireMembers, []encryption.StorageRetirementReport{ + readyStorage("n1", 0), readyStorage("n2", 0), readyStorage("n3", 0), + }, retiringKeyID, 0) + require.NoError(t, err) + require.True(t, d.Eligible, "an unused key has no version to retain even at the initial retention floor") +} + +func TestRetirementRejectsReportsOutsideMembership(t *testing.T) { + t.Parallel() + storage := []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), readyStorage("n3", 200), + } + _, err := classifyStorage([]string{"n1", "n2"}, storage, retiringKeyID, 100) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + require.ErrorContains(t, err, "nonmembers") + + raft := []encryption.RaftRetirementReport{ + readyRaft("n1", 500, 500), readyRaft("n2", 500, 500), readyRaft("n3", 1, 1), + } + _, err = encryption.ClassifyRaftDEKRetirement([]string{"n1", "n2"}, raft, retiringKeyID, boundaries(100)) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + require.ErrorContains(t, err, "nonmembers") +} + +func boundaries(largestProposed uint64) map[uint64]encryption.RaftGroupBoundary { + return map[uint64]encryption.RaftGroupBoundary{ + retireGroup: {KeyID: retiringKeyID, LargestProposedIndex: largestProposed, Observed: true}, + } +} + +func readyRaft(node string, compact, snapshot uint64) encryption.RaftRetirementReport { + return encryption.RaftRetirementReport{ + NodeID: node, + ReportedKeyID: retiringKeyID, + ActiveKeyID: successorKeyID, + Groups: []encryption.RaftGroupRetirementReport{{ + GroupID: retireGroup, + LogCompactIndex: compact, + SnapshotIndex: snapshot, + }}, + } +} + +// --------------------------------------------------------------------------- +// Storage purpose +// --------------------------------------------------------------------------- + +func TestStorageRetirementRejectsUnobservedValueCount(t *testing.T) { + t.Parallel() + for _, count := range []uint64{0, 7} { + reports := []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), readyStorage("n3", 200), + } + reports[1].ValuesPerDEK = count + reports[1].ValuesPerDEKObserved = false + d, err := classifyStorage(retireMembers, reports, retiringKeyID, 100) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + require.ErrorContains(t, err, "n2") + require.ErrorContains(t, err, "old-key value scan") + require.False(t, d.Eligible) + } +} + +func TestRaftRetirementRequiresObservedHighWaterMark(t *testing.T) { + t.Parallel() + reports := []encryption.RaftRetirementReport{ + readyRaft("n1", 500, 400), readyRaft("n2", 500, 400), readyRaft("n3", 500, 400), + } + for _, proposed := range []uint64{0, 300} { + bounds := boundaries(proposed) + boundary := bounds[retireGroup] + boundary.Observed = false + bounds[retireGroup] = boundary + d, err := encryption.ClassifyRaftDEKRetirement(retireMembers, reports, retiringKeyID, bounds) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + require.ErrorContains(t, err, "not observed") + require.False(t, d.Eligible) + } + d, err := encryption.ClassifyRaftDEKRetirement(retireMembers, reports, retiringKeyID, boundaries(0)) + require.NoError(t, err) + require.True(t, d.Eligible, "an observed unused key is safe when all remaining criteria pass") +} + +func TestStorageRetirementEligibleWhenEveryCriterionHolds(t *testing.T) { + t.Parallel() + + d, err := classifyStorage(retireMembers, + []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), readyStorage("n3", 200), + }, retiringKeyID, 100) + require.NoError(t, err) + require.True(t, d.Eligible) + require.Empty(t, d.Blockers) + require.NoError(t, d.Err()) +} + +// TestStorageRetirementRefusesWhileAnyNodeStillHoldsValues pins the +// per-node nature of the criterion: one lagging replica is enough, +// because unloading the DEK makes ITS data unreadable. +func TestStorageRetirementRefusesWhileAnyNodeStillHoldsValues(t *testing.T) { + t.Parallel() + + behind := readyStorage("n2", 200) + behind.ValuesPerDEK = 7 + + d, err := classifyStorage(retireMembers, + []encryption.StorageRetirementReport{ + readyStorage("n1", 200), behind, readyStorage("n3", 200), + }, retiringKeyID, 100) + require.NoError(t, err) + require.False(t, d.Eligible) + require.Len(t, d.Blockers, 1) + require.Contains(t, d.Blockers[0], "n2") + require.True(t, errors.Is(d.Err(), + encryption.ErrDEKStillReferenced)) +} + +func TestStorageRetirementRefusesWhileTheRewriteIsIncomplete(t *testing.T) { + t.Parallel() + + partial := readyStorage("n3", 200) + partial.RewriteCursorComplete = false + + d, err := classifyStorage(retireMembers, + []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), partial, + }, retiringKeyID, 100) + require.NoError(t, err) + require.False(t, d.Eligible) +} + +func TestStorageRetirementRequiresWriteFenceAndDrain(t *testing.T) { + t.Parallel() + + reports := []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), readyStorage("n3", 200), + } + + t.Run("missing fence token", func(t *testing.T) { + t.Parallel() + _, err := encryption.ClassifyStorageDEKRetirement( + retireMembers, reports, retiringKeyID, + encryption.StorageRetirementBoundary{ + KeyID: retiringKeyID, LargestCommitTS: 100, Observed: true, + }) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + require.ErrorContains(t, err, "write fence token") + }) + + t.Run("report from another fence generation", func(t *testing.T) { + t.Parallel() + mismatched := append([]encryption.StorageRetirementReport(nil), reports...) + mismatched[1].WriteFenceToken++ + _, err := classifyStorage(retireMembers, mismatched, retiringKeyID, 100) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + require.ErrorContains(t, err, "expected") + }) + + t.Run("old-key write still in flight", func(t *testing.T) { + t.Parallel() + blocked := append([]encryption.StorageRetirementReport(nil), reports...) + blocked[1].InFlightOldKeyWrites = 1 + d, err := classifyStorage(retireMembers, blocked, retiringKeyID, 100) + require.NoError(t, err) + require.False(t, d.Eligible) + require.Contains(t, d.Blockers[0], "in flight") + }) + + t.Run("drain was not observed", func(t *testing.T) { + t.Parallel() + incomplete := append([]encryption.StorageRetirementReport(nil), reports...) + incomplete[1].InFlightWriteDrainObserved = false + _, err := classifyStorage(retireMembers, incomplete, retiringKeyID, 100) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + require.ErrorContains(t, err, "write drain") + }) +} + +func TestStorageRetirementRequiresPersistentReferencesGone(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + mutate func(*encryption.StorageRetirementReport) + wantBlock string + }{ + { + name: "rewrite cursor envelope", + mutate: func(r *encryption.StorageRetirementReport) { + r.RewriteCursorReferences = 1 + }, + wantBlock: "rewrite cursors", + }, + { + name: "retained snapshot", + mutate: func(r *encryption.StorageRetirementReport) { + r.RetainedSnapshotReferences = 2 + }, + wantBlock: "retained snapshots", + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + blocked := readyStorage("n2", 200) + tc.mutate(&blocked) + d, err := classifyStorage(retireMembers, + []encryption.StorageRetirementReport{ + readyStorage("n1", 200), blocked, readyStorage("n3", 200), + }, retiringKeyID, 100) + require.NoError(t, err) + require.False(t, d.Eligible) + require.Len(t, d.Blockers, 1) + require.Contains(t, d.Blockers[0], tc.wantBlock) + }) + } +} + +func TestStorageRetirementRefusesUnobservedPersistentReferences(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + mutate func(*encryption.StorageRetirementReport) + want string + }{ + { + name: "rewrite cursor scan", + mutate: func(r *encryption.StorageRetirementReport) { + r.RewriteCursorReferencesObserved = false + }, + want: "cursor reference scan", + }, + { + name: "snapshot inventory", + mutate: func(r *encryption.StorageRetirementReport) { + r.RetainedSnapshotReferencesObserved = false + }, + want: "snapshot reference inventory", + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + reports := []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), readyStorage("n3", 200), + } + tc.mutate(&reports[1]) + _, err := classifyStorage(retireMembers, reports, retiringKeyID, 100) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + require.ErrorContains(t, err, tc.want) + }) + } +} + +// TestStorageRetirementRequiresMinRetainedTSStrictlyPastTheCommitTS is +// the boundary §5.4 states as "greater than". Equality still admits a +// snapshot read at exactly that version, so it must not be eligible. +func TestStorageRetirementRequiresMinRetainedTSStrictlyPastTheCommitTS(t *testing.T) { + t.Parallel() + + const largestCommitTS = uint64(100) + + equal, err := classifyStorage(retireMembers, + []encryption.StorageRetirementReport{ + readyStorage("n1", largestCommitTS), + readyStorage("n2", largestCommitTS), + readyStorage("n3", largestCommitTS), + }, retiringKeyID, largestCommitTS) + require.NoError(t, err) + require.False(t, equal.Eligible, + "minRetainedTS equal to the commit_ts still admits a read at that version") + + past, err := classifyStorage(retireMembers, + []encryption.StorageRetirementReport{ + readyStorage("n1", largestCommitTS+1), + readyStorage("n2", largestCommitTS+1), + readyStorage("n3", largestCommitTS+1), + }, retiringKeyID, largestCommitTS) + require.NoError(t, err) + require.True(t, past.Eligible) +} + +// --------------------------------------------------------------------------- +// Raft purpose +// --------------------------------------------------------------------------- + +func TestRaftRetirementEligibleWhenWALAndSnapshotsHavePassed(t *testing.T) { + t.Parallel() + + d, err := encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", 500, 400), readyRaft("n2", 500, 400), readyRaft("n3", 500, 400), + }, retiringKeyID, boundaries(300)) + require.NoError(t, err) + require.True(t, d.Eligible) + require.NoError(t, d.Err()) +} + +// TestRaftRetirementRefusesWhileTheWALStillHoldsEntries is the failure +// mode §5.4 names: unloading the DEK while the WAL still references it +// causes unknown_key_id apply failures on the next restart or on a +// lagging follower's catch-up. +func TestRaftRetirementRefusesWhileTheWALStillHoldsEntries(t *testing.T) { + t.Parallel() + + lagging := readyRaft("n2", 250, 400) // compact index below the proposed index + + d, err := encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", 500, 400), lagging, readyRaft("n3", 500, 400), + }, retiringKeyID, boundaries(300)) + require.NoError(t, err) + require.False(t, d.Eligible) + require.True(t, errors.Is(d.Err(), + encryption.ErrRaftDEKWALStillReferences), + "the raft path has its own sentinel so the runbook points at the WAL, not the rewrite") +} + +// TestRaftRetirementRequiresTheCompactIndexStrictlyPast pins §5.4's +// "strictly greater than": an index EQUAL to the largest proposed one +// means that entry is still un-truncated and would be replayed. +func TestRaftRetirementRequiresTheCompactIndexStrictlyPast(t *testing.T) { + t.Parallel() + + const proposed = uint64(300) + + equal, err := encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", proposed, 400), readyRaft("n2", proposed, 400), readyRaft("n3", proposed, 400), + }, retiringKeyID, boundaries(proposed)) + require.NoError(t, err) + require.False(t, equal.Eligible) + + past, err := encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", proposed+1, 400), readyRaft("n2", proposed+1, 400), readyRaft("n3", proposed+1, 400), + }, retiringKeyID, boundaries(proposed)) + require.NoError(t, err) + require.True(t, past.Eligible) +} + +// TestRaftRetirementRefusesASnapshotPredatingTheRotation covers the +// second raft criterion: a node restored from an older snapshot would +// replay entries that still need the retiring DEK. +func TestRaftRetirementRefusesASnapshotPredatingTheRotation(t *testing.T) { + t.Parallel() + + stale := readyRaft("n3", 500, 100) // snapshot taken before the rotation at 350 + + d, err := encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", 500, 400), readyRaft("n2", 500, 400), stale, + }, retiringKeyID, boundaries(300)) + require.NoError(t, err) + require.False(t, d.Eligible) + require.Contains(t, d.Blockers[0], "n3") +} + +// --------------------------------------------------------------------------- +// Cluster-wide coverage — a missing node is not a passing node +// --------------------------------------------------------------------------- + +func TestRetirementRefusesAPartialReport(t *testing.T) { + t.Parallel() + + // n3 never reported. Treating silence as success is how a single + // unreachable replica ends up replaying against an unloaded DEK. + _, err := classifyStorage(retireMembers, + []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), + }, retiringKeyID, 100) + require.Error(t, err) + require.True(t, errors.Is(err, encryption.ErrIncompleteRetirementReport)) + require.ErrorContains(t, err, "n3") + + _, err = encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", 500, 400), readyRaft("n2", 500, 400), + }, retiringKeyID, boundaries(300)) + require.Error(t, err) + require.True(t, errors.Is(err, encryption.ErrIncompleteRetirementReport)) +} + +// TestRetirementRefusesAnEmptyMembership guards the degenerate case: +// "no member reported a problem" is not evidence when nobody was asked. +func TestRetirementRefusesAnEmptyMembership(t *testing.T) { + t.Parallel() + + _, err := classifyStorage(nil, nil, retiringKeyID, 100) + require.True(t, errors.Is(err, encryption.ErrIncompleteRetirementReport)) + + _, err = encryption.ClassifyRaftDEKRetirement(nil, nil, retiringKeyID, boundaries(300)) + require.True(t, errors.Is(err, encryption.ErrIncompleteRetirementReport)) +} + +// TestRetirementBlockersNameEveryOffendingNode keeps the operator +// output actionable: being told only "not ready" leaves them polling a +// cluster with no idea which replica to wait on. +func TestRetirementBlockersNameEveryOffendingNode(t *testing.T) { + t.Parallel() + + bad1 := readyStorage("n1", 200) + bad1.ValuesPerDEK = 3 + bad3 := readyStorage("n3", 200) + bad3.RewriteCursorComplete = false + + d, err := classifyStorage(retireMembers, + []encryption.StorageRetirementReport{bad1, readyStorage("n2", 200), bad3}, retiringKeyID, 100) + require.NoError(t, err) + require.False(t, d.Eligible) + require.Len(t, d.Blockers, 2) + joined := d.Blockers[0] + d.Blockers[1] + require.Contains(t, joined, "n1") + require.Contains(t, joined, "n3") +} + +// TestRaftRetirementStaysReachableAcrossSuccessiveRotations is the +// regression test for the criterion's signal. +// +// §5.4's parenthetical says the snapshot check should read the FSM +// snapshot header's raft_envelope_cutover_index. That value is the +// one-shot Phase-2 enablement index: applier.go preserves the original +// across every later rotation and kv/fsm.go copies the unchanged value +// into each new snapshot header, so it stays frozen at the enablement +// index forever. Sourcing the criterion from it makes every +// post-cutover rotation permanently ineligible — no raft DEK could ever +// be retired, which defeats the whole classifier. +// +// SnapshotIndex is the snapshot's own Raft index, so it advances with +// every snapshot install and the criterion is reachable. +func TestRaftRetirementStaysReachableAcrossSuccessiveRotations(t *testing.T) { + t.Parallel() + + // Phase-2 was enabled at index 50 and never moves again. + const frozenCutoverIndex = uint64(50) + + // Two rotations have happened since. + for _, rotationIndex := range []uint64{400, 900} { + require.Less(t, frozenCutoverIndex, rotationIndex, + "the frozen cutover index cannot reach rotation %d, so a criterion "+ + "sourced from it is unsatisfiable by construction", rotationIndex) + + // A snapshot taken after the rotation satisfies the criterion. + snapshotIndex := rotationIndex + 10 + d, err := encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", rotationIndex+1, snapshotIndex), + readyRaft("n2", rotationIndex+1, snapshotIndex), + readyRaft("n3", rotationIndex+1, snapshotIndex), + }, retiringKeyID, boundaries(rotationIndex-1)) + require.NoError(t, err) + require.True(t, d.Eligible, + "rotation %d must become retirable once every node has snapshotted past it; blockers: %v", + rotationIndex, d.Blockers) + + // Boundary: a snapshot exactly at the rotation index is enough, + // because a restore from it replays only later entries. + d, err = encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", rotationIndex+1, rotationIndex), + readyRaft("n2", rotationIndex+1, rotationIndex), + readyRaft("n3", rotationIndex+1, rotationIndex), + }, retiringKeyID, boundaries(rotationIndex-1)) + require.NoError(t, err) + require.True(t, d.Eligible, "blockers: %v", d.Blockers) + } +} + +// TestRetirementRefusesDuplicateNodeReports pins the fail-closed +// handling of an ambiguous report set. +// +// Collapsing reports into a map is last-write-wins, so a node that +// reported a blocker and then reported ready would be recorded as ready +// while coverage still looked complete — eligibility would depend on +// the order the reports arrived. Both classifiers must refuse instead. +func TestRetirementRefusesDuplicateNodeReports(t *testing.T) { + t.Parallel() + + t.Run("storage: a blocker followed by a ready report", func(t *testing.T) { + t.Parallel() + + blocked := readyStorage("n2", 200) + blocked.ValuesPerDEK = 7 + + _, err := classifyStorage(retireMembers, + []encryption.StorageRetirementReport{ + readyStorage("n1", 200), + blocked, + readyStorage("n2", 200), // duplicate: would win and hide the blocker + readyStorage("n3", 200), + }, retiringKeyID, 100) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + }) + + t.Run("raft: a blocker followed by a ready report", func(t *testing.T) { + t.Parallel() + + blocked := readyRaft("n2", 1, 1000) + + _, err := encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", 500, 1000), + blocked, + readyRaft("n2", 500, 1000), // duplicate: would win and hide the blocker + readyRaft("n3", 500, 1000), + }, retiringKeyID, boundaries(100)) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + }) +} + +// --------------------------------------------------------------------------- +// The retiring key must not be the active one +// --------------------------------------------------------------------------- + +// TestRetirementRefusesTheStillActiveDEK pins the guard against unloading the +// key still selected for writes. +// +// Every retention criterion can legitimately pass for the ACTIVE key: an empty +// cluster reports a complete rewrite cursor, zero values and an advanced +// retention floor for the key it is still writing under. The classifiers saw +// neither the retiring key nor each node's active key, so they returned +// eligible and a retire command would unload the live key — breaking the next +// write or proposal rather than an old read. +func TestRetirementRefusesTheStillActiveDEK(t *testing.T) { + t.Parallel() + + t.Run("storage: the retiring key is still active", func(t *testing.T) { + t.Parallel() + + reports := []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), readyStorage("n3", 200), + } + // Every criterion passes; only the active key is wrong. + reports[1].ActiveKeyID = retiringKeyID + + _, err := classifyStorage( + retireMembers, reports, retiringKeyID, 100) + require.ErrorIs(t, err, encryption.ErrRetiringDEKStillActive) + }) + + t.Run("raft: the retiring key is still active", func(t *testing.T) { + t.Parallel() + + reports := []encryption.RaftRetirementReport{ + readyRaft("n1", 500, 1000), readyRaft("n2", 500, 1000), readyRaft("n3", 500, 1000), + } + reports[2].ActiveKeyID = retiringKeyID + + _, err := encryption.ClassifyRaftDEKRetirement( + retireMembers, reports, retiringKeyID, boundaries(100)) + require.ErrorIs(t, err, encryption.ErrRetiringDEKStillActive) + }) + + t.Run("a node with no successor active is refused", func(t *testing.T) { + t.Parallel() + + reports := []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), readyStorage("n3", 200), + } + // Zero means "no key active", which cannot be treated as a successor. + reports[0].ActiveKeyID = 0 + + _, err := classifyStorage( + retireMembers, reports, retiringKeyID, 100) + require.ErrorIs(t, err, encryption.ErrRetiringDEKStillActive) + }) +} + +// TestRetirementRefusesAReportForAnotherKey pins the binding between the report +// and the key being retired: a report gathered for a different DEK says nothing +// about this one, and accepting it silently judges the wrong key. +func TestRetirementRefusesAReportForAnotherKey(t *testing.T) { + t.Parallel() + + reports := []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), readyStorage("n3", 200), + } + reports[1].ReportedKeyID = retiringKeyID + 99 + + _, err := classifyStorage( + retireMembers, reports, retiringKeyID, 100) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) +} + +func TestRetirementRequiresTheRetiringKeyID(t *testing.T) { + t.Parallel() + + _, err := classifyStorage(retireMembers, nil, 0, 100) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + + _, err = encryption.ClassifyRaftDEKRetirement( + retireMembers, nil, 0, boundaries(100)) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) +} + +// --------------------------------------------------------------------------- +// Raft boundaries are per group +// --------------------------------------------------------------------------- + +func multiGroupRaft(node string, groups ...encryption.RaftGroupRetirementReport) encryption.RaftRetirementReport { + return encryption.RaftRetirementReport{ + NodeID: node, + ReportedKeyID: retiringKeyID, + ActiveKeyID: successorKeyID, + Groups: groups, + } +} + +// TestRaftRetirementChecksEveryGroupAgainstItsOwnBoundary is the multi-shard +// regression. +// +// The raft DEK is cluster-wide — raftEnvelopeRuntime.installFromApply sets the +// same wrap on every attached ShardGroup — but Raft log and snapshot indexes +// live in independent per-group index spaces. Judging the shared DEK from one +// group's indexes left every other group unchecked, so a group whose WAL still +// held old-key entries could not block retirement. +func TestRaftRetirementChecksEveryGroupAgainstItsOwnBoundary(t *testing.T) { + t.Parallel() + + bounds := map[uint64]encryption.RaftGroupBoundary{ + // A busy group with high indexes... + 1: {KeyID: retiringKeyID, LargestProposedIndex: 10_000, Observed: true}, + // ...and a quiet one whose indexes are far lower. + 2: {KeyID: retiringKeyID, LargestProposedIndex: 40, Observed: true}, + } + + t.Run("every group past its own boundary is eligible", func(t *testing.T) { + t.Parallel() + + reports := make([]encryption.RaftRetirementReport, 0, len(retireMembers)) + for _, node := range retireMembers { + reports = append(reports, multiGroupRaft(node, + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 10_000}, + encryption.RaftGroupRetirementReport{GroupID: 2, LogCompactIndex: 41, SnapshotIndex: 40}, + )) + } + + d, err := encryption.ClassifyRaftDEKRetirement(retireMembers, reports, retiringKeyID, bounds) + require.NoError(t, err) + require.True(t, d.Eligible, + "the quiet group must be judged against ITS OWN low boundary, not the busy "+ + "group's: a global minimum would keep it permanently behind; blockers: %v", + d.Blockers) + }) + + t.Run("a group still behind its boundary blocks retirement", func(t *testing.T) { + t.Parallel() + + reports := make([]encryption.RaftRetirementReport, 0, len(retireMembers)) + for _, node := range retireMembers { + reports = append(reports, multiGroupRaft(node, + // Group 1 is ready... + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 10_000}, + // ...group 2's WAL still holds entries proposed under the key. + // Its snapshot matches its log start, as an untruncated node's + // does: the blocker is the log, and exactly one per node. + encryption.RaftGroupRetirementReport{GroupID: 2, LogCompactIndex: 20, SnapshotIndex: 40}, + )) + } + + d, err := encryption.ClassifyRaftDEKRetirement(retireMembers, reports, retiringKeyID, bounds) + require.NoError(t, err) + require.False(t, d.Eligible, + "an unready group must block the shared DEK even when other groups are ready") + require.Len(t, d.Blockers, len(retireMembers)) + require.Contains(t, d.Blockers[0], "group 2") + }) + + t.Run("a node that omits a group is an incomplete report", func(t *testing.T) { + t.Parallel() + + reports := []encryption.RaftRetirementReport{ + multiGroupRaft("n1", + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 10_000}, + encryption.RaftGroupRetirementReport{GroupID: 2, LogCompactIndex: 41, SnapshotIndex: 40}), + // n2 reports only group 1, so group 2 is unverified on that node. + multiGroupRaft("n2", + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 10_000}), + multiGroupRaft("n3", + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 10_000}, + encryption.RaftGroupRetirementReport{GroupID: 2, LogCompactIndex: 41, SnapshotIndex: 40}), + } + + _, err := encryption.ClassifyRaftDEKRetirement(retireMembers, reports, retiringKeyID, bounds) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + }) + + t.Run("a duplicate group on one node is refused", func(t *testing.T) { + t.Parallel() + + reports := []encryption.RaftRetirementReport{ + multiGroupRaft("n1", + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 5, SnapshotIndex: 1}, + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 10_000}), + multiGroupRaft("n2", + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 10_000}), + multiGroupRaft("n3", + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 10_000}), + } + + _, err := encryption.ClassifyRaftDEKRetirement(retireMembers, reports, retiringKeyID, + map[uint64]encryption.RaftGroupBoundary{1: {KeyID: retiringKeyID, LargestProposedIndex: 10_000, Observed: true}}) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + }) +} + +func TestRaftRetirementRequiresPerGroupBoundaries(t *testing.T) { + t.Parallel() + + // No boundaries means no group's old-key high-water mark is known. + // "Nothing to check" must not read as "safe". + _, err := encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", 500, 1000), readyRaft("n2", 500, 1000), readyRaft("n3", 500, 1000), + }, retiringKeyID, nil) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) +} + +// --------------------------------------------------------------------------- +// The sentinel follows the classifier, not an argument +// --------------------------------------------------------------------------- + +// TestRetirementErrSentinelComesFromTheClassifier pins that each blocked +// decision reports its own sentinel. +// +// Err used to pick from a caller-supplied purpose string, defaulting to the +// storage sentinel for anything unrecognised — so a blocked RAFT decision +// passed a misspelled or omitted purpose sent the operator to the rewrite/MVCC +// remediation for what is actually a WAL blocker. +func TestRetirementErrSentinelComesFromTheClassifier(t *testing.T) { + t.Parallel() + + storage, err := classifyStorage(retireMembers, + []encryption.StorageRetirementReport{ + readyStorage("n1", 1), readyStorage("n2", 1), readyStorage("n3", 1), + }, retiringKeyID, 100) + require.NoError(t, err) + require.False(t, storage.Eligible) + require.ErrorIs(t, storage.Err(), encryption.ErrDEKStillReferenced) + require.NotErrorIs(t, storage.Err(), encryption.ErrRaftDEKWALStillReferences) + + raft, err := encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", 10, 1000), readyRaft("n2", 10, 1000), readyRaft("n3", 10, 1000), + }, retiringKeyID, boundaries(500)) + require.NoError(t, err) + require.False(t, raft.Eligible) + require.ErrorIs(t, raft.Err(), encryption.ErrRaftDEKWALStillReferences, + "a WAL blocker must not be reported as an MVCC one") + require.NotErrorIs(t, raft.Err(), encryption.ErrDEKStillReferenced) +} + +// A decision with no purpose is a wiring bug, and guessing a sentinel would +// point the operator at the wrong remediation. +func TestRetirementErrRefusesToGuessAPurpose(t *testing.T) { + t.Parallel() + + blocked := encryption.RetirementDecision{Blockers: []string{"n1: something"}} + require.ErrorIs(t, blocked.Err(), encryption.ErrIncompleteRetirementReport) + + eligible := encryption.RetirementDecision{Eligible: true} + require.NoError(t, eligible.Err(), "an eligible decision has no error whatever its purpose") +} + +// TestRaftRetirementRejectsReportedGroupsWithoutABoundary closes the other +// direction of coverage. +// +// Checking only that every BOUNDARY was reported treats the boundary map as +// the complete group universe. If collection omits group 2 while every node +// reports group 2 with old-key entries still in its log, group 2 is judged by +// nobody -- and the classifier returns eligible for a DEK whose entries are +// still replayable. An incomplete boundary map must not be able to authorize +// unloading the shared key. +func TestRaftRetirementRejectsReportedGroupsWithoutABoundary(t *testing.T) { + t.Parallel() + + reports := make([]encryption.RaftRetirementReport, 0, len(retireMembers)) + for _, node := range retireMembers { + reports = append(reports, multiGroupRaft(node, + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 10_000}, + // Reported, and far behind -- but absent from the boundary map. + encryption.RaftGroupRetirementReport{GroupID: 2, LogCompactIndex: 1, SnapshotIndex: 1}, + )) + } + + _, err := encryption.ClassifyRaftDEKRetirement(retireMembers, reports, retiringKeyID, + map[uint64]encryption.RaftGroupBoundary{ + 1: {KeyID: retiringKeyID, LargestProposedIndex: 10_000, Observed: true}, + }) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + require.ErrorContains(t, err, "no boundary") +} + +// TestRaftRetirementRejectsABoundaryForAnotherKey pins the boundary's key +// binding, which the reports already had and the boundaries did not. +// +// Every report can correctly name key 7 while the boundary handed in for a +// group describes key 8. Key 8's lower high-water mark is then trivially +// cleared, the classifier returns eligible, and key 7's entries are still +// replayable. Mixing per-key metric series has to fail closed. +func TestRaftRetirementRejectsABoundaryForAnotherKey(t *testing.T) { + t.Parallel() + + reports := []encryption.RaftRetirementReport{ + readyRaft("n1", 10_001, 10_000), + readyRaft("n2", 10_001, 10_000), + readyRaft("n3", 10_001, 10_000), + } + + _, err := encryption.ClassifyRaftDEKRetirement(retireMembers, reports, retiringKeyID, + map[uint64]encryption.RaftGroupBoundary{ + retireGroup: {KeyID: retiringKeyID + 1, LargestProposedIndex: 10, Observed: true}, + }) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + require.ErrorContains(t, err, "not the retiring key") +} + +// TestRaftRetirementBlocksASnapshotBelowTheGroupsHighWaterMark states the +// snapshot criterion in the only index space that exists per group. +// +// There is no per-group rotation index to compare against: the rotation is +// proposed once through the DEFAULT group's engine and its apply handler +// installs the new wrapper on every group in memory, so a non-default group +// has no rotation entry of its own. The quantity that does exist per group is +// the largest index proposed under the retiring key, and restoring from a +// snapshot below it replays exactly those entries. +func TestRaftRetirementBlocksASnapshotBelowTheGroupsHighWaterMark(t *testing.T) { + t.Parallel() + + d, err := encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + // Log start is past the mark, but the snapshot is not: a restore + // would begin before the last old-key entry. + readyRaft("n1", 10_001, 9_000), + readyRaft("n2", 10_001, 10_000), + readyRaft("n3", 10_001, 10_000), + }, retiringKeyID, boundaries(10_000)) + require.NoError(t, err) + require.False(t, d.Eligible) + require.Len(t, d.Blockers, 1) + require.Contains(t, d.Blockers[0], "n1") + require.Contains(t, d.Blockers[0], "high-water mark") + + // Exactly at the mark is enough: a restore from it replays only entries + // at or after the last old-key one, and that one is in the snapshot. + d, err = encryption.ClassifyRaftDEKRetirement(retireMembers, + []encryption.RaftRetirementReport{ + readyRaft("n1", 10_001, 10_000), + readyRaft("n2", 10_001, 10_000), + readyRaft("n3", 10_001, 10_000), + }, retiringKeyID, boundaries(10_000)) + require.NoError(t, err) + require.True(t, d.Eligible, "blockers: %v", d.Blockers) +}