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 99f306e4c..89081e51a 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 @@ -35,7 +35,8 @@ Date: 2026-04-29 | 9A | Compress-then-encrypt, authenticated compression flag, encrypted-store Pebble compression policy, storage benchmark (§6.4, §8.3) | shipped | `2026_07_18_implemented_9a_encryption_compression.md` | | 9B | AWS KMS, GCP KMS, Vault Transit, and test/CI env KEK providers; mutually-exclusive source loader and loaded-provider mutator gate (§5.1, §6.1, §6.5) | shipped | `2026_07_18_implemented_9b_kek_providers.md` | | 9C-1 | Storage-envelope observability: `decrypt_failures_total`, `writes_per_dek`, `value_overhead_bytes`, wired from the storage envelope path through `monitoring.Registry` (§9.2) | shipped | — | -| 9C+ | Rotation budget/rewrap/retire/rewrite, the remaining §9.2 metrics (`active_dek_id`, `last_proposed_index_per_raft_dek`, `kek_unwrap_seconds`, `sidecar_raft_index`), remaining benchmarks and encrypted Jepsen (§5.2, §5.4, §6.5, §8) | open | — | +| 9C-5 | §5.4 DEK retirement eligibility: the storage criteria (rewrite cursor, values-per-DEK, minRetainedTS) and the WAL-driven raft criteria (log start index, last committed snapshot index) evaluated **per Raft group**, since log indexes are per-group index spaces while the raft DEK is installed on every group. Reports are bound to the key being retired and must show a different key active, so the live DEK cannot be unloaded. Cluster-wide, no override. | shipped | — | +| 9C+ | Rotation budget/rewrap/retire/rewrite, the remaining §9.2 metrics (`active_dek_id`, `last_proposed_index_per_raft_dek`, `kek_unwrap_seconds`, `sidecar_raft_index`), remaining benchmarks and encrypted Jepsen (§5.2, §5.4, §6.5, §8, §9.2) | open | — | Stages 0–4 ship the entire byte-tag pipeline (storage envelope, raft envelope, FSM dispatch, halt-on-error) but leave it **production @@ -1271,12 +1272,62 @@ 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 the rotation entry's index**. A node restored + from an older snapshot replays the entries between that + snapshot and the rotation, and those were proposed under + the retiring DEK. + + The signal is the snapshot's own Raft index + (`raftpb.SnapshotMetadata.Index`), which advances with + every snapshot install. + + **Corrected from an earlier revision of this section,** + which said the check reads the §4.4 FSM snapshot + header's `raft_envelope_cutover_index`. That field cannot + express this criterion, for two independent reasons: + + The criterion is evaluated **per Raft group**. 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`. Judging + the shared DEK from a single group's indexes leaves the + other groups unchecked; aggregating across groups with a + minimum is worse, because 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 at all. + + **Retirement also refuses the ACTIVE key.** Every + criterion above can legitimately pass for the key still + selected for writes — an empty cluster reports a complete + rewrite cursor, zero values and an advanced retention + floor for its active storage DEK — so each node's report + names the key it describes and the key it currently + writes under, and `retire-dek` refuses unless a different + successor is active everywhere. Unloading the active key + breaks the next write or proposal rather than an old + read, which is why this is its own refusal + (`ErrRetiringDEKStillActive`) and not a blocker. + + - It is the one-shot Phase-2 *enablement* index. + `applier.go` preserves the original value across every + later rotation (`if sc.RaftEnvelopeCutoverIndex != 0` + takes the already-active branch and advances only + `RaftAppliedIndex`), and `kv/fsm.go` copies the + unchanged value into each new snapshot header. It is + therefore frozen and can never be "past the rotation + entry" for any post-cutover rotation — so the criterion + would classify the retiring raft DEK ineligible forever + and no raft DEK could ever be retired. + - "Taken under the new raft DEK" is not a property a + snapshot has. Per §4.4 the FSM snapshot stream "is + ciphertext by construction" from the storage layer and + "no additional wrapping is required at the snapshot + layer" — a snapshot is not encrypted under any *raft* + DEK. What matters is only which entries a restore from + it would replay, which is what the snapshot's index + states directly. 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..0e2574c0f --- /dev/null +++ b/internal/encryption/dek_retirement.go @@ -0,0 +1,440 @@ +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 and snapshots 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 + // 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 +} + +// 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. See RaftRetirementReport.SnapshotIndex for why it is the + // snapshot's own index rather than raft_envelope_cutover_index. + SnapshotIndex uint64 +} + +// RaftGroupBoundary is the old-key high-water mark for one Raft group. +type RaftGroupBoundary struct { + // LargestProposedIndex is the largest log index ever proposed under the + // retiring DEK in this group. + LargestProposedIndex uint64 + // RotationIndex is the index of the rotation entry that installed the + // successor in this group. + RotationIndex 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 + // LogCompactIndex is this node's persisted Raft log start index — + // 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. + LogCompactIndex uint64 + // SnapshotIndex is the Raft index of this node's last committed + // snapshot (raftpb.SnapshotMetadata.Index). It must be at least the + // rotation index: restoring from a snapshot taken BEFORE the + // rotation replays the entries between the snapshot and the + // rotation, and those were proposed under the retiring DEK. + // + // NOT the sidecar's raft_envelope_cutover_index, which §5.4's + // parenthetical names. 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. It can therefore never be "past the + // rotation entry" for any post-cutover rotation, so a criterion + // built on it classifies the retiring raft DEK ineligible forever + // and no raft DEK could ever be retired. + // + // The cutover index also does not mean what that criterion needs. + // Per §4.4 the FSM snapshot stream "is ciphertext by construction" + // from the storage layer and "no additional wrapping is required at + // the snapshot layer" — a snapshot is not encrypted under any raft + // DEK, so "taken under the new raft DEK" can only be about which + // entries a restore would replay. That is exactly what the + // snapshot's own index expresses, and it advances with every + // snapshot install. + SnapshotIndex uint64 +} + +// 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. +// +// largestCommitTS is the largest commit_ts ever written under 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, largestCommitTS uint64, +) (RetirementDecision, error) { + if retiringKeyID == 0 { + return RetirementDecision{}, errors.Wrap(ErrIncompleteRetirementReport, + "retiring key id 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 := requireFullCoverage(members, func(n string) bool { + _, ok := byNode[n] + return ok + }); 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 + } + 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)) + } + // Strictly greater: a minRetainedTS EQUAL to the largest + // commit_ts still admits a read at exactly that version. + if r.MinRetainedTS <= largestCommitTS { + blockers = append(blockers, + fmt.Sprintf("%s: minRetainedTS %d has not advanced past commit_ts %d", + node, r.MinRetainedTS, largestCommitTS)) + } + } + sort.Strings(blockers) + return RetirementDecision{ + Purpose: RetirementPurposeStorage, + Eligible: len(blockers) == 0, + Blockers: blockers, + }, nil +} + +// ClassifyRaftDEKRetirement applies the §5.4 raft criteria. +// +// largestProposedIndex is the largest log index ever proposed under the +// retiring DEK; rotationIndex is the index of the rotation entry that +// installed its successor. +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 := requireFullCoverage(members, func(n string) bool { + _, ok := byNode[n] + return ok + }); 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) + 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. +func raftGroupBlockers( + node string, + groups []RaftGroupRetirementReport, + boundaries map[uint64]RaftGroupBoundary, +) ([]string, 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 + } + var missing []uint64 + for groupID := range boundaries { + if _, ok := byGroup[groupID]; !ok { + missing = append(missing, groupID) + } + } + if len(missing) > 0 { + slices.Sort(missing) + return nil, errors.Wrapf(ErrIncompleteRetirementReport, + "node %s did not report groups %v", node, missing) + } + + 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] + g := byGroup[groupID] + // 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.RotationIndex { + blockers = append(blockers, + fmt.Sprintf("%s group %d: last snapshot predates the rotation (snapshot %d < rotation %d)", + node, groupID, g.SnapshotIndex, boundary.RotationIndex)) + } + } + 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) +} + +// requireFullCoverage rejects a report set that does not cover every +// member, and a membership list that is empty. +// +// Duplicates are caught by the callers, not here. This function +// previously compared the number of unique reporting nodes against the +// membership size, which cannot detect a duplicate at all: if every +// member is covered, the unique count equals the membership size +// whether or not a node reported twice, and if a member is missing the +// `missing` check above already fires. +// +// An empty membership is refused rather than treated as trivially +// satisfied: "no members reported a problem" is not evidence when +// nobody was asked. +func requireFullCoverage(members []string, covered func(string) bool) error { + if len(members) == 0 { + return errors.Wrap(ErrIncompleteRetirementReport, "cluster membership is empty") + } + var missing []string + for _, node := range members { + if !covered(node) { + missing = append(missing, node) + } + } + if len(missing) > 0 { + sort.Strings(missing) + return errors.Wrapf(ErrIncompleteRetirementReport, "no report from %v", missing) + } + return nil +} diff --git a/internal/encryption/dek_retirement_test.go b/internal/encryption/dek_retirement_test.go new file mode 100644 index 000000000..090ec2723 --- /dev/null +++ b/internal/encryption/dek_retirement_test.go @@ -0,0 +1,612 @@ +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) +) + +// 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, + MinRetainedTS: minRetained, + } +} + +func boundaries(largestProposed, rotation uint64) map[uint64]encryption.RaftGroupBoundary { + return map[uint64]encryption.RaftGroupBoundary{ + retireGroup: {LargestProposedIndex: largestProposed, RotationIndex: rotation}, + } +} + +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 TestStorageRetirementEligibleWhenEveryCriterionHolds(t *testing.T) { + t.Parallel() + + d, err := encryption.ClassifyStorageDEKRetirement(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 := encryption.ClassifyStorageDEKRetirement(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 := encryption.ClassifyStorageDEKRetirement(retireMembers, + []encryption.StorageRetirementReport{ + readyStorage("n1", 200), readyStorage("n2", 200), partial, + }, retiringKeyID, 100) + require.NoError(t, err) + require.False(t, d.Eligible) +} + +// 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 := encryption.ClassifyStorageDEKRetirement(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 := encryption.ClassifyStorageDEKRetirement(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, 350)) + 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, 350)) + 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, 350)) + 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, 350)) + 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, 350)) + 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 := encryption.ClassifyStorageDEKRetirement(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, 350)) + 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 := encryption.ClassifyStorageDEKRetirement(nil, nil, retiringKeyID, 100) + require.True(t, errors.Is(err, encryption.ErrIncompleteRetirementReport)) + + _, err = encryption.ClassifyRaftDEKRetirement(nil, nil, retiringKeyID, boundaries(300, 350)) + 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 := encryption.ClassifyStorageDEKRetirement(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, rotationIndex)) + 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, rotationIndex)) + 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 := encryption.ClassifyStorageDEKRetirement(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, 900)) + 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 := encryption.ClassifyStorageDEKRetirement( + 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, 900)) + 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 := encryption.ClassifyStorageDEKRetirement( + 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 := encryption.ClassifyStorageDEKRetirement( + retireMembers, reports, retiringKeyID, 100) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) +} + +func TestRetirementRequiresTheRetiringKeyID(t *testing.T) { + t.Parallel() + + _, err := encryption.ClassifyStorageDEKRetirement(retireMembers, nil, 0, 100) + require.ErrorIs(t, err, encryption.ErrIncompleteRetirementReport) + + _, err = encryption.ClassifyRaftDEKRetirement( + retireMembers, nil, 0, boundaries(100, 200)) + 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: {LargestProposedIndex: 10_000, RotationIndex: 9_000}, + // ...and a quiet one whose indexes are far lower. + 2: {LargestProposedIndex: 40, RotationIndex: 30}, + } + + 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: 9_500}, + encryption.RaftGroupRetirementReport{GroupID: 2, LogCompactIndex: 41, SnapshotIndex: 35}, + )) + } + + 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: 9_500}, + // ...group 2's WAL still holds entries proposed under the key. + encryption.RaftGroupRetirementReport{GroupID: 2, LogCompactIndex: 20, SnapshotIndex: 35}, + )) + } + + 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: 9_500}, + encryption.RaftGroupRetirementReport{GroupID: 2, LogCompactIndex: 41, SnapshotIndex: 35}), + // n2 reports only group 1, so group 2 is unverified on that node. + multiGroupRaft("n2", + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 9_500}), + multiGroupRaft("n3", + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 9_500}, + encryption.RaftGroupRetirementReport{GroupID: 2, LogCompactIndex: 41, SnapshotIndex: 35}), + } + + _, 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: 9_500}), + multiGroupRaft("n2", + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 9_500}), + multiGroupRaft("n3", + encryption.RaftGroupRetirementReport{GroupID: 1, LogCompactIndex: 10_001, SnapshotIndex: 9_500}), + } + + _, err := encryption.ClassifyRaftDEKRetirement(retireMembers, reports, retiringKeyID, + map[uint64]encryption.RaftGroupBoundary{1: {LargestProposedIndex: 10_000, RotationIndex: 9_000}}) + 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 := encryption.ClassifyStorageDEKRetirement(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, 900)) + 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") +}