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 8cc1e9a2c..d5a06d54f 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 @@ -37,7 +37,8 @@ Date: 2026-04-29 | 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-2 | Sidecar/KEK observability: `active_dek_id{purpose}`, `sidecar_raft_index`, `kek_unwrap_seconds` (§9.2) | shipped | — | | 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+ | Rotation budget/rewrap/retire/rewrite, `last_proposed_index_per_raft_dek` (needs the §5.4 raft-DEK Wrap path), remaining benchmarks (§5.2, §5.4, §6.5, §8.3) | open | — | +| 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 budget/rewrap/retire, the rewrite job's execution half, admission-control wiring, `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) | open | — | Stages 0–4 ship the entire byte-tag pipeline (storage envelope, raft envelope, FSM dispatch, halt-on-error) but leave it **production diff --git a/internal/encryption/rewrite_plan.go b/internal/encryption/rewrite_plan.go new file mode 100644 index 000000000..013c3dac8 --- /dev/null +++ b/internal/encryption/rewrite_plan.go @@ -0,0 +1,181 @@ +package encryption + +import ( + "math" + "time" +) + +// §5.4 rewrite-job decision layer. +// +// The rewrite job re-encrypts data under a new DEK so an old one can be +// retired. §5.4 is emphatic that it is NOT a single-pass conversion of +// "the live value of every key": Pebble holds MVCC history, and the +// snapshot and lease-read paths can read back retained versions, including +// the newest version at or below minRetainedTS. A rewrite that touched +// only the live version would +// leave older versions under the retiring DEK and quietly break +// snapshot reads the moment it was unloaded. +// +// Two pure decisions live here — which versions to rewrite, and when to +// yield to stay inside the write-rate budget — so both are testable +// without a Pebble instance. Driving the iterator and the batch is the +// execution half. + +// RewriteVerdict is what the job does with one MVCC version. +type RewriteVerdict int + +const ( + // RewriteSkip leaves the version untouched. + RewriteSkip RewriteVerdict = iota + // RewriteReencrypt re-encrypts the value under the active DEK, at + // the SAME internal key. §5.4: no new MVCC version, no OCC + // conflict, no visible change to readers. + RewriteReencrypt +) + +func (v RewriteVerdict) String() string { + if v == RewriteReencrypt { + return "reencrypt" + } + return "skip" +} + +// Reasons, a closed set so the job can meter them directly. +const ( + RewriteReasonRetiringDEK = "retiring_dek" + RewriteReasonCleartext = "cleartext_migration" + RewriteReasonTombstone = "tombstone" + RewriteReasonDifferentDEK = "different_dek" +) + +// RewriteDecision is a verdict plus its reason. +type RewriteDecision struct { + Verdict RewriteVerdict + Reason string +} + +// MVCCVersionRef is one retained version, as the iterator sees it. +// +// The iteration unit is (user_key, version_ts) rather than user_key — +// that distinction is the whole point of §5.4, so the type carries the +// timestamp rather than letting a caller forget it. +type MVCCVersionRef struct { + CommitTS uint64 + // KeyID is the DEK this version's envelope names, or zero when the + // version is stored cleartext. + KeyID uint32 + // Cleartext is the MVCC metadata bit saying this version predates + // the §7.1 cutover and holds no envelope. + Cleartext bool + // Tombstone versions carry no value bytes, so there is nothing to + // re-encrypt. + Tombstone bool + // ValueBytes sizes the write for the rate budget. + ValueBytes int64 +} + +// ClassifyMVCCRewrite decides one version's fate. +// +// retiringKeyID is the DEK being retired. migrateCleartext enables the +// §7.1 cleartext→encrypted sweep; it is separate from the retiring DEK +// because the two jobs run at different times and conflating them +// would have a routine rotation silently start encrypting data the +// operator had not opted in to encrypting. +// +// The iterator supplies physically retained versions. Its newest version +// at or below the retention floor can still serve a read at the floor, so +// the classifier must not skip an old version based on its timestamp alone. +func ClassifyMVCCRewrite( + version MVCCVersionRef, retiringKeyID uint32, migrateCleartext bool, +) RewriteDecision { + if version.Tombstone { + // No value bytes to re-encrypt. Checked first because a + // tombstone's KeyID is meaningless. + return RewriteDecision{Verdict: RewriteSkip, Reason: RewriteReasonTombstone} + } + if version.Cleartext { + if migrateCleartext { + return RewriteDecision{Verdict: RewriteReencrypt, Reason: RewriteReasonCleartext} + } + return RewriteDecision{Verdict: RewriteSkip, Reason: RewriteReasonCleartext} + } + if version.KeyID == retiringKeyID { + return RewriteDecision{Verdict: RewriteReencrypt, Reason: RewriteReasonRetiringDEK} + } + return RewriteDecision{Verdict: RewriteSkip, Reason: RewriteReasonDifferentDEK} +} + +// bytesPerMiB converts the operator-facing `--rate=N MiB/s` unit into +// the byte-denominated arithmetic the throttle runs on. +const bytesPerMiB = 1024 * 1024 + +// RewriteThrottle implements the §5.4 `--rate=N MiB/s` write-rate +// budget by telling the job how long to yield between batches. +// +// Rate-limiting the rewrite matters because it competes with live +// traffic for the same Pebble write path: an unthrottled sweep of MVCC +// history is a sustained write amplification spike against a database +// that is also serving requests. +type RewriteThrottle struct { + bytesPerSecond float64 + deadline time.Time + written int64 +} + +// NewRewriteThrottle returns a throttle for the given rate. A +// non-positive rate disables throttling, which is what --rate=0 means: +// run as fast as the store allows. +func NewRewriteThrottle(mibPerSecond float64, now time.Time) *RewriteThrottle { + return &RewriteThrottle{ + bytesPerSecond: mibPerSecond * bytesPerMiB, + deadline: now, + } +} + +// Record accounts for a committed batch and returns how long the job +// should yield before the next one. +// +// Each batch advances a virtual deadline from the later of the previous +// deadline and now. Idle time therefore never becomes credit for a burst. +func (t *RewriteThrottle) Record(batchBytes int64, now time.Time) time.Duration { + if t == nil { + return 0 + } + if batchBytes > 0 { + if batchBytes > math.MaxInt64-t.written { + t.written = math.MaxInt64 + } else { + t.written += batchBytes + } + } + if t.bytesPerSecond <= 0 || math.IsNaN(t.bytesPerSecond) { + return 0 + } + if now.After(t.deadline) { + t.deadline = now + } + if batchBytes > 0 { + t.deadline = t.deadline.Add(rewriteBatchDelay(batchBytes, t.bytesPerSecond)) + } + if !t.deadline.After(now) { + return 0 + } + return t.deadline.Sub(now) +} + +func rewriteBatchDelay(batchBytes int64, bytesPerSecond float64) time.Duration { + seconds := float64(batchBytes) / bytesPerSecond + delay := seconds * float64(time.Second) + if math.IsInf(delay, 1) || delay >= float64(math.MaxInt64) { + return time.Duration(math.MaxInt64) + } + return time.Duration(delay) +} + +// Written reports the cumulative bytes recorded. +func (t *RewriteThrottle) Written() int64 { + if t == nil { + return 0 + } + return t.written +} diff --git a/internal/encryption/rewrite_plan_test.go b/internal/encryption/rewrite_plan_test.go new file mode 100644 index 000000000..58e82a409 --- /dev/null +++ b/internal/encryption/rewrite_plan_test.go @@ -0,0 +1,234 @@ +package encryption_test + +import ( + "math" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/encryption" + "github.com/stretchr/testify/require" +) + +const ( + retiringDEK = uint32(7) + activeDEK = uint32(8) + retainFloor = uint64(100) +) + +func version(over encryption.MVCCVersionRef) encryption.MVCCVersionRef { + if over.CommitTS == 0 { + over.CommitTS = retainFloor + 50 + } + return over +} + +// TestClassifyMVCCRewriteCoversEveryVersionShape is the §5.4 iteration +// contract. The unit is (user_key, version_ts): a rewrite that touched +// only live values would leave older versions under the retiring DEK +// and break snapshot reads the moment it was unloaded. +func TestClassifyMVCCRewriteCoversEveryVersionShape(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + version encryption.MVCCVersionRef + migrate bool + wantVerb encryption.RewriteVerdict + wantReason string + }{ + { + name: "version under the retiring DEK", + version: version(encryption.MVCCVersionRef{KeyID: retiringDEK}), + wantVerb: encryption.RewriteReencrypt, + wantReason: encryption.RewriteReasonRetiringDEK, + }, + { + name: "version already under the active DEK", + version: version(encryption.MVCCVersionRef{KeyID: activeDEK}), + wantVerb: encryption.RewriteSkip, + wantReason: encryption.RewriteReasonDifferentDEK, + }, + { + name: "version under another old DEK", + version: version(encryption.MVCCVersionRef{KeyID: activeDEK + 1}), + wantVerb: encryption.RewriteSkip, + wantReason: encryption.RewriteReasonDifferentDEK, + }, + { + name: "tombstone carries no value bytes", + version: version(encryption.MVCCVersionRef{KeyID: retiringDEK, Tombstone: true}), + wantVerb: encryption.RewriteSkip, + wantReason: encryption.RewriteReasonTombstone, + }, + { + name: "cleartext during a migration sweep", + version: version(encryption.MVCCVersionRef{Cleartext: true}), + migrate: true, + wantVerb: encryption.RewriteReencrypt, + wantReason: encryption.RewriteReasonCleartext, + }, + { + name: "cleartext outside a migration sweep", + version: version(encryption.MVCCVersionRef{Cleartext: true}), + migrate: false, + wantVerb: encryption.RewriteSkip, + wantReason: encryption.RewriteReasonCleartext, + }, + { + name: "retained anchor below the retention floor", + version: encryption.MVCCVersionRef{CommitTS: retainFloor - 1, KeyID: retiringDEK}, + wantVerb: encryption.RewriteReencrypt, + wantReason: encryption.RewriteReasonRetiringDEK, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got := encryption.ClassifyMVCCRewrite(tc.version, retiringDEK, tc.migrate) + require.Equal(t, tc.wantVerb, got.Verdict) + require.Equal(t, tc.wantReason, got.Reason) + }) + } +} + +// TestClassifyMVCCRewriteDoesNotEncryptCleartextDuringAPlainRotation +// keeps the two jobs separate. A routine rotation must not silently +// start encrypting data the operator never opted in to encrypting — +// that is the §7.1 migration, a different decision with a different +// blast radius. +func TestClassifyMVCCRewriteDoesNotEncryptCleartextDuringAPlainRotation(t *testing.T) { + t.Parallel() + + v := version(encryption.MVCCVersionRef{Cleartext: true}) + require.Equal(t, encryption.RewriteSkip, + encryption.ClassifyMVCCRewrite(v, retiringDEK, false).Verdict) + require.Equal(t, encryption.RewriteReencrypt, + encryption.ClassifyMVCCRewrite(v, retiringDEK, true).Verdict) +} + +// TestClassifyMVCCRewriteRewritesHistoricalVersionsNotJustTheLatest is +// the property §5.4 opens with: every RETAINED version under the +// retiring DEK must be rewritten, however old, or unloading the DEK +// breaks snapshot reads. +func TestClassifyMVCCRewriteRewritesHistoricalVersionsNotJustTheLatest(t *testing.T) { + t.Parallel() + + for _, ts := range []uint64{retainFloor + 1, retainFloor + 10, retainFloor + 10_000} { + got := encryption.ClassifyMVCCRewrite( + encryption.MVCCVersionRef{CommitTS: ts, KeyID: retiringDEK}, + retiringDEK, false) + require.Equal(t, encryption.RewriteReencrypt, got.Verdict, + "retained version at ts=%d must be rewritten", ts) + } +} + +// TestClassifyMVCCRewriteKeepsTheRetentionFloorAnchor pins the +// version an exact-floor read can still reach. +func TestClassifyMVCCRewriteKeepsTheRetentionFloorAnchor(t *testing.T) { + t.Parallel() + + at := encryption.ClassifyMVCCRewrite( + encryption.MVCCVersionRef{CommitTS: retainFloor, KeyID: retiringDEK}, + retiringDEK, false) + require.Equal(t, encryption.RewriteReencrypt, at.Verdict) + + above := encryption.ClassifyMVCCRewrite( + encryption.MVCCVersionRef{CommitTS: retainFloor + 1, KeyID: retiringDEK}, + retiringDEK, false) + require.Equal(t, encryption.RewriteReencrypt, above.Verdict) +} + +// --------------------------------------------------------------------------- +// Rate budget +// --------------------------------------------------------------------------- + +func TestRewriteThrottleYieldsToHoldTheConfiguredRate(t *testing.T) { + t.Parallel() + + start := time.Unix(1_700_000_000, 0) + // 1 MiB/s. + th := encryption.NewRewriteThrottle(1, start) + + // 2 MiB written instantly needs 2s of elapsed time. + require.Equal(t, 2*time.Second, th.Record(2*1024*1024, start)) + + // After 2s have actually passed, no further yield is owed. + require.Zero(t, th.Record(0, start.Add(2*time.Second))) +} + +// TestRewriteThrottleCarriesDebtAcrossBackToBackBatches pins the +// anti-burst property for successive batches at the same instant. +func TestRewriteThrottleCarriesDebtAcrossBackToBackBatches(t *testing.T) { + t.Parallel() + + start := time.Unix(1_700_000_000, 0) + // 1 MiB/s. + th := encryption.NewRewriteThrottle(1, start) + + // First MiB at t=0: owes 1s under either rule. + require.Equal(t, time.Second, th.Record(1024*1024, start)) + + // Second MiB, still at t=0, extends the virtual deadline to t=2s. + require.Equal(t, 2*time.Second, th.Record(1024*1024, start), + "the debt must accumulate across batches") + + // Third MiB after 2s have passed: 3 MiB owes 3s, 2s elapsed. + require.Equal(t, time.Second, + th.Record(1024*1024, start.Add(2*time.Second))) +} + +func TestRewriteThrottleDoesNotBankIdleTime(t *testing.T) { + t.Parallel() + start := time.Unix(1_700_000_000, 0) + th := encryption.NewRewriteThrottle(1, start) + require.Equal(t, time.Second, th.Record(1024*1024, start)) + + afterIdle := start.Add(time.Minute) + require.Equal(t, time.Second, th.Record(1024*1024, afterIdle)) + require.Equal(t, 2*time.Second, th.Record(1024*1024, afterIdle)) +} + +func TestRewriteThrottleDisabledAtNonPositiveRate(t *testing.T) { + t.Parallel() + + start := time.Unix(1_700_000_000, 0) + for _, rate := range []float64{0, -1} { + th := encryption.NewRewriteThrottle(rate, start) + require.Zero(t, th.Record(1<<30, start), "--rate=%v means run unthrottled", rate) + require.Equal(t, int64(1<<30), th.Written()) + } +} + +func TestRewriteThrottleClampsUnrepresentableDelay(t *testing.T) { + t.Parallel() + start := time.Unix(1_700_000_000, 0) + th := encryption.NewRewriteThrottle(math.SmallestNonzeroFloat64, start) + require.Equal(t, time.Duration(math.MaxInt64), th.Record(1, start)) + require.Equal(t, int64(1), th.Written()) +} + +func TestRewriteThrottleTracksCumulativeBytes(t *testing.T) { + t.Parallel() + + start := time.Unix(1_700_000_000, 0) + th := encryption.NewRewriteThrottle(1, start) + th.Record(1024, start) + th.Record(2048, start) + require.Equal(t, int64(3072), th.Written()) +} + +func TestRewriteThrottleNilReceiverIsInert(t *testing.T) { + t.Parallel() + + var th *encryption.RewriteThrottle + require.Zero(t, th.Record(1<<20, time.Unix(0, 0))) + require.Zero(t, th.Written()) +} + +func TestRewriteVerdictStringsAreStable(t *testing.T) { + t.Parallel() + + require.Equal(t, "reencrypt", encryption.RewriteReencrypt.String()) + require.Equal(t, "skip", encryption.RewriteSkip.String()) +}