diff --git a/staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config.go b/staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config.go
index f0922518e6b..205798bb5cc 100644
--- a/staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config.go
+++ b/staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config.go
@@ -36,6 +36,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/apimachinery/pkg/util/wait"
 	apiserverconfig "k8s.io/apiserver/pkg/apis/config"
 	apiserverconfigv1 "k8s.io/apiserver/pkg/apis/config/v1"
@@ -60,11 +61,39 @@ const (
 	secretboxTransformerPrefixV1 = "k8s:enc:secretbox:v1:"
 	kmsTransformerPrefixV1       = "k8s:enc:kms:v1:"
 	kmsTransformerPrefixV2       = "k8s:enc:kms:v2:"
-	kmsPluginHealthzInterval     = 1 * time.Minute
-	kmsPluginHealthzNegativeTTL  = 3 * time.Second
-	kmsPluginHealthzPositiveTTL  = 20 * time.Second
-	kmsAPIVersionV1              = "v1"
-	kmsAPIVersionV2              = "v2"
+
+	// these constants relate to how the KMS v2 plugin status poll logic
+	// and the DEK generation logic behave.  In particular, the positive
+	// interval and max TTL are closely related as the difference between
+	// these values defines the worst case window in which the write DEK
+	// could expire due to the plugin going into an error state.  The
+	// worst case window divided by the negative interval defines the
+	// minimum amount of times the server will attempt to return to a
+	// healthy state before the DEK expires and writes begin to fail.
+	//
+	// For now, these values are kept small and hardcoded to support being
+	// able to perform a "passive" storage migration while tolerating some
+	// amount of plugin downtime.
+	//
+	// With the current approach, a user can update the key ID their plugin
+	// is using and then can simply schedule a migration for 3 + N + M minutes
+	// later where N is how long it takes their plugin to pick up new config
+	// and M is extra buffer to allow the API server to process the config.
+	// At that point, they are guaranteed to either migrate to the new key
+	// or get errors during the migration.
+	//
+	// If the API server coasted forever on the last DEK, they would need
+	// to actively check if it had observed the new key ID before starting
+	// a migration - otherwise it could keep using the old DEK and their
+	// storage migration would not do what they thought it did.
+	kmsv2PluginHealthzPositiveInterval = 1 * time.Minute
+	kmsv2PluginHealthzNegativeInterval = 10 * time.Second
+	kmsv2PluginWriteDEKMaxTTL          = 3 * time.Minute
+
+	kmsPluginHealthzNegativeTTL = 3 * time.Second
+	kmsPluginHealthzPositiveTTL = 20 * time.Second
+	kmsAPIVersionV1             = "v1"
+	kmsAPIVersionV2             = "v2"
 	// this name is used for two different healthz endpoints:
 	// - when one or more KMS v2 plugins are in use and no KMS v1 plugins are in use
 	//   in this case, all v2 plugins are probed via this single endpoint
@@ -88,7 +117,7 @@ type kmsPluginProbe struct {
 }
 
 type kmsv2PluginProbe struct {
-	keyID        atomic.Pointer[string]
+	state        atomic.Pointer[envelopekmsv2.State]
 	name         string
 	ttl          time.Duration
 	service      kmsservice.Service
@@ -281,7 +310,7 @@ func (h *kmsv2PluginProbe) check(ctx context.Context) error {
 	h.l.Lock()
 	defer h.l.Unlock()
 
-	if (time.Since(h.lastResponse.received)) < h.ttl {
+	if time.Since(h.lastResponse.received) < h.ttl {
 		return h.lastResponse.err
 	}
 
@@ -291,15 +320,8 @@ func (h *kmsv2PluginProbe) check(ctx context.Context) error {
 		h.ttl = kmsPluginHealthzNegativeTTL
 		return fmt.Errorf("failed to perform status section of the healthz check for KMS Provider %s, error: %w", h.name, err)
 	}
-	// we coast on the last valid key ID that we have observed
-	if errCode, err := envelopekmsv2.ValidateKeyID(p.KeyID); err == nil {
-		h.keyID.Store(&p.KeyID)
-		metrics.RecordKeyIDFromStatus(h.name, p.KeyID)
-	} else {
-		metrics.RecordInvalidKeyIDFromStatus(h.name, string(errCode))
-	}
 
-	if err := isKMSv2ProviderHealthy(h.name, p); err != nil {
+	if err := h.isKMSv2ProviderHealthyAndMaybeRotateDEK(ctx, p); err != nil {
 		h.lastResponse = &kmsPluginHealthzResponse{err: err, received: time.Now()}
 		h.ttl = kmsPluginHealthzNegativeTTL
 		return err
@@ -310,17 +332,88 @@ func (h *kmsv2PluginProbe) check(ctx context.Context) error {
 	return nil
 }
 
-// getCurrentKeyID returns the latest keyID from the last Status() call or err if keyID is empty
-func (h *kmsv2PluginProbe) getCurrentKeyID(ctx context.Context) (string, error) {
-	keyID := *h.keyID.Load()
-	if len(keyID) == 0 {
-		return "", fmt.Errorf("got unexpected empty keyID")
+// rotateDEKOnKeyIDChange tries to rotate to a new DEK if the key ID returned by Status does not match the
+// current state.  If a successful rotation is performed, the new DEK and keyID overwrite the existing state.
+// On any failure during rotation (including mismatch between status and encrypt calls), the current state is
+// preserved and will remain valid to use for encryption until its expiration (the system attempts to coast).
+// If the key ID returned by Status matches the current state, the expiration of the current state is extended
+// and no rotation is performed.
+func (h *kmsv2PluginProbe) rotateDEKOnKeyIDChange(ctx context.Context, statusKeyID, uid string) error {
+	// we do not check ValidateEncryptCapability here because it is fine to re-use an old key
+	// that was marked as expired during an unhealthy period.  As long as the key ID matches
+	// what we expect then there is no need to rotate here.
+	state, errState := h.getCurrentState()
+
+	// allow reads indefinitely in all cases
+	// allow writes indefinitely as long as there is no error
+	// allow writes for only up to kmsv2PluginWriteDEKMaxTTL from now when there are errors
+	// we start the timer before we make the network call because kmsv2PluginWriteDEKMaxTTL is meant to be the upper bound
+	expirationTimestamp := envelopekmsv2.NowFunc().Add(kmsv2PluginWriteDEKMaxTTL)
+
+	// state is valid and status keyID is unchanged from when we generated this DEK so there is no need to rotate it
+	// just move the expiration of the current state forward by the reuse interval
+	if errState == nil && state.KeyID == statusKeyID {
+		state.ExpirationTimestamp = expirationTimestamp
+		h.state.Store(&state)
+		return nil
 	}
-	return keyID, nil
+
+	transformer, resp, errGen := envelopekmsv2.GenerateTransformer(ctx, uid, h.service)
+
+	if resp == nil {
+		resp = &kmsservice.EncryptResponse{} // avoid nil panics
+	}
+
+	// happy path, should be the common case
+	// TODO maybe add success metrics?
+	if errGen == nil && resp.KeyID == statusKeyID {
+		h.state.Store(&envelopekmsv2.State{
+			Transformer:         transformer,
+			EncryptedDEK:        resp.Ciphertext,
+			KeyID:               resp.KeyID,
+			Annotations:         resp.Annotations,
+			UID:                 uid,
+			ExpirationTimestamp: expirationTimestamp,
+		})
+		klog.V(6).InfoS("successfully rotated DEK",
+			"uid", uid,
+			"newKeyID", resp.KeyID,
+			"oldKeyID", state.KeyID,
+			"expirationTimestamp", expirationTimestamp.Format(time.RFC3339),
+		)
+		return nil
+	}
+
+	return fmt.Errorf("failed to rotate DEK uid=%q, errState=%v, errGen=%v, statusKeyID=%q, encryptKeyID=%q, stateKeyID=%q, expirationTimestamp=%s",
+		uid, errState, errGen, statusKeyID, resp.KeyID, state.KeyID, state.ExpirationTimestamp.Format(time.RFC3339))
 }
 
-// isKMSv2ProviderHealthy checks if the KMSv2-Plugin is healthy.
-func isKMSv2ProviderHealthy(name string, response *kmsservice.StatusResponse) error {
+// getCurrentState returns the latest state from the last status and encrypt calls.
+// If the returned error is nil, the state is considered valid indefinitely for read requests.
+// For write requests, the caller must also check that state.ValidateEncryptCapability does not error.
+func (h *kmsv2PluginProbe) getCurrentState() (envelopekmsv2.State, error) {
+	state := *h.state.Load()
+
+	if state.Transformer == nil {
+		return envelopekmsv2.State{}, fmt.Errorf("got unexpected nil transformer")
+	}
+
+	if len(state.EncryptedDEK) == 0 {
+		return envelopekmsv2.State{}, fmt.Errorf("got unexpected empty EncryptedDEK")
+	}
+
+	if len(state.KeyID) == 0 {
+		return envelopekmsv2.State{}, fmt.Errorf("got unexpected empty keyID")
+	}
+
+	if state.ExpirationTimestamp.IsZero() {
+		return envelopekmsv2.State{}, fmt.Errorf("got unexpected zero expirationTimestamp")
+	}
+
+	return state, nil
+}
+
+func (h *kmsv2PluginProbe) isKMSv2ProviderHealthyAndMaybeRotateDEK(ctx context.Context, response *kmsservice.StatusResponse) error {
 	var errs []error
 	if response.Healthz != "ok" {
 		errs = append(errs, fmt.Errorf("got unexpected healthz status: %s", response.Healthz))
@@ -328,12 +421,18 @@ func isKMSv2ProviderHealthy(name string, response *kmsservice.StatusResponse) er
 	if response.Version != envelopekmsv2.KMSAPIVersion {
 		errs = append(errs, fmt.Errorf("expected KMSv2 API version %s, got %s", envelopekmsv2.KMSAPIVersion, response.Version))
 	}
-	if _, err := envelopekmsv2.ValidateKeyID(response.KeyID); err != nil {
-		errs = append(errs, fmt.Errorf("expected KMSv2 KeyID to be set, got %s", response.KeyID))
+
+	if errCode, err := envelopekmsv2.ValidateKeyID(response.KeyID); err != nil {
+		metrics.RecordInvalidKeyIDFromStatus(h.name, string(errCode))
+		errs = append(errs, fmt.Errorf("got invalid KMSv2 KeyID %q: %w", response.KeyID, err))
+	} else {
+		metrics.RecordKeyIDFromStatus(h.name, response.KeyID)
+		// unconditionally append as we filter out nil errors below
+		errs = append(errs, h.rotateDEKOnKeyIDChange(ctx, response.KeyID, string(uuid.NewUUID())))
 	}
 
 	if err := utilerrors.Reduce(utilerrors.NewAggregate(errs)); err != nil {
-		return fmt.Errorf("kmsv2 Provider %s is not healthy, error: %w", name, err)
+		return fmt.Errorf("kmsv2 Provider %s is not healthy, error: %w", h.name, err)
 	}
 	return nil
 }
@@ -393,7 +492,10 @@ func prefixTransformersAndProbes(ctx context.Context, config apiserverconfig.Res
 			transformer, transformerErr = aesPrefixTransformer(provider.AESGCM, aestransformer.NewGCMTransformer, aesGCMTransformerPrefixV1)
 
 		case provider.AESCBC != nil:
-			transformer, transformerErr = aesPrefixTransformer(provider.AESCBC, aestransformer.NewCBCTransformer, aesCBCTransformerPrefixV1)
+			cbcTransformer := func(block cipher.Block) (value.Transformer, error) {
+				return aestransformer.NewCBCTransformer(block), nil
+			}
+			transformer, transformerErr = aesPrefixTransformer(provider.AESCBC, cbcTransformer, aesCBCTransformerPrefixV1)
 
 		case provider.Secretbox != nil:
 			transformer, transformerErr = secretboxPrefixTransformer(provider.Secretbox)
@@ -425,7 +527,7 @@ func prefixTransformersAndProbes(ctx context.Context, config apiserverconfig.Res
 	return transformers, probes, &kmsUsed, nil
 }
 
-type blockTransformerFunc func(cipher.Block) value.Transformer
+type blockTransformerFunc func(cipher.Block) (value.Transformer, error)
 
 func aesPrefixTransformer(config *apiserverconfig.AESConfiguration, fn blockTransformerFunc, prefix string) (value.PrefixTransformer, error) {
 	var result value.PrefixTransformer
@@ -449,17 +551,21 @@ func aesPrefixTransformer(config *apiserverconfig.AESConfiguration, fn blockTran
 		keyData := keyData
 		key, err := base64.StdEncoding.DecodeString(keyData.Secret)
 		if err != nil {
-			return result, fmt.Errorf("could not obtain secret for named key %s: %s", keyData.Name, err)
+			return result, fmt.Errorf("could not obtain secret for named key %s: %w", keyData.Name, err)
 		}
 		block, err := aes.NewCipher(key)
 		if err != nil {
-			return result, fmt.Errorf("error while creating cipher for named key %s: %s", keyData.Name, err)
+			return result, fmt.Errorf("error while creating cipher for named key %s: %w", keyData.Name, err)
+		}
+		transformer, err := fn(block)
+		if err != nil {
+			return result, fmt.Errorf("error while creating transformer for named key %s: %w", keyData.Name, err)
 		}
 
 		// Create a new PrefixTransformer for this key
 		keyTransformers = append(keyTransformers,
 			value.PrefixTransformer{
-				Transformer: fn(block),
+				Transformer: transformer,
 				Prefix:      []byte(keyData.Name + ":"),
 			})
 	}
@@ -596,27 +702,49 @@ func kmsPrefixTransformer(ctx context.Context, config *apiserverconfig.KMSConfig
 			l:            &sync.Mutex{},
 			lastResponse: &kmsPluginHealthzResponse{},
 		}
-		// initialize keyID so that Load always works
-		keyID := ""
-		probe.keyID.Store(&keyID)
+		// initialize state so that Load always works
+		probe.state.Store(&envelopekmsv2.State{})
 
-		// prime keyID by running the check inline once (this prevents unit tests from flaking)
+		runProbeCheckAndLog := func(ctx context.Context) error {
+			if err := probe.check(ctx); err != nil {
+				klog.VDepth(1, 2).ErrorS(err, "kms plugin failed health check probe", "name", kmsName)
+				return err
+			}
+			return nil
+		}
+
+		// on the happy path where the plugin is healthy and available on server start,
+		// prime keyID and DEK by running the check inline once (this also prevents unit tests from flaking)
 		// ignore the error here since we want to support the plugin starting up async with the API server
-		_ = probe.check(ctx)
+		_ = runProbeCheckAndLog(ctx)
 		// make sure that the plugin's key ID is reasonably up-to-date
+		// also, make sure that our DEK is up-to-date to with said key ID (if it expires the server will fail all writes)
+		// if this background loop ever stops running, the server will become unfunctional after kmsv2PluginWriteDEKMaxTTL
 		go wait.PollUntilWithContext(
 			ctx,
-			kmsPluginHealthzInterval,
+			kmsv2PluginHealthzPositiveInterval,
 			func(ctx context.Context) (bool, error) {
-				if err := probe.check(ctx); err != nil {
-					klog.V(2).ErrorS(err, "kms plugin failed health check probe", "name", kmsName)
+				if err := runProbeCheckAndLog(ctx); err == nil {
+					return false, nil
 				}
+
+				// TODO add integration test for quicker error poll on failure
+				// if we fail, block the outer polling and start a new quicker poll inline
+				// this limits the chance that our DEK expires during a transient failure
+				_ = wait.PollUntilWithContext(
+					ctx,
+					kmsv2PluginHealthzNegativeInterval,
+					func(ctx context.Context) (bool, error) {
+						return runProbeCheckAndLog(ctx) == nil, nil
+					},
+				)
+
 				return false, nil
 			})
 
 		// using AES-GCM by default for encrypting data with KMSv2
 		transformer := value.PrefixTransformer{
-			Transformer: envelopekmsv2.NewEnvelopeTransformer(envelopeService, kmsName, probe.getCurrentKeyID, probe.check, aestransformer.NewGCMTransformer),
+			Transformer: envelopekmsv2.NewEnvelopeTransformer(envelopeService, kmsName, probe.getCurrentState),
 			Prefix:      []byte(kmsTransformerPrefixV2 + kmsName + ":"),
 		}
 
@@ -631,12 +759,17 @@ func kmsPrefixTransformer(ctx context.Context, config *apiserverconfig.KMSConfig
 }
 
 func envelopePrefixTransformer(config *apiserverconfig.KMSConfiguration, envelopeService envelope.Service, prefix string) value.PrefixTransformer {
-	baseTransformerFunc := func(block cipher.Block) value.Transformer {
+	baseTransformerFunc := func(block cipher.Block) (value.Transformer, error) {
+		gcm, err := aestransformer.NewGCMTransformer(block)
+		if err != nil {
+			return nil, err
+		}
+
 		// v1.24: write using AES-CBC only but support reads via AES-CBC and AES-GCM (so we can move to AES-GCM)
 		// v1.25: write using AES-GCM only but support reads via AES-GCM and fallback to AES-CBC for backwards compatibility
 		// TODO(aramase): Post v1.25: We cannot drop CBC read support until we automate storage migration.
 		// We could have a release note that hard requires users to perform storage migration.
-		return unionTransformers{aestransformer.NewGCMTransformer(block), aestransformer.NewCBCTransformer(block)}
+		return unionTransformers{gcm, aestransformer.NewCBCTransformer(block)}, nil
 	}
 
 	return value.PrefixTransformer{
diff --git a/staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config_test.go b/staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config_test.go
index ba752581a78..9768a58c111 100644
--- a/staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config_test.go
+++ b/staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig/config_test.go
@@ -21,6 +21,8 @@ import (
 	"context"
 	"encoding/base64"
 	"errors"
+	"fmt"
+	"io"
 	"reflect"
 	"strings"
 	"sync"
@@ -31,6 +33,7 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/sets"
 	apiserverconfig "k8s.io/apiserver/pkg/apis/config"
 	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/storage/value"
@@ -41,6 +44,7 @@ import (
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	"k8s.io/component-base/metrics/legacyregistry"
 	"k8s.io/component-base/metrics/testutil"
+	"k8s.io/klog/v2"
 	kmsservice "k8s.io/kms/pkg/service"
 	"k8s.io/utils/pointer"
 )
@@ -77,8 +81,9 @@ func (t *testEnvelopeService) Encrypt(data []byte) ([]byte, error) {
 // testKMSv2EnvelopeService is a mock kmsv2 envelope service which can be used to simulate remote Envelope v2 services
 // for testing of the envelope transformer with other transformers.
 type testKMSv2EnvelopeService struct {
-	err   error
-	keyID string
+	err          error
+	keyID        string
+	encryptCalls int
 }
 
 func (t *testKMSv2EnvelopeService) Decrypt(ctx context.Context, uid string, req *kmsservice.DecryptRequest) ([]byte, error) {
@@ -89,6 +94,7 @@ func (t *testKMSv2EnvelopeService) Decrypt(ctx context.Context, uid string, req
 }
 
 func (t *testKMSv2EnvelopeService) Encrypt(ctx context.Context, uid string, data []byte) (*kmsservice.EncryptResponse, error) {
+	t.encryptCalls++
 	if t.err != nil {
 		return nil, t.err
 	}
@@ -117,17 +123,17 @@ func newMockErrorEnvelopeService(endpoint string, timeout time.Duration) (envelo
 
 // The factory method to create mock envelope kmsv2 service.
 func newMockEnvelopeKMSv2Service(ctx context.Context, endpoint, providerName string, timeout time.Duration) (kmsservice.Service, error) {
-	return &testKMSv2EnvelopeService{nil, "1"}, nil
+	return &testKMSv2EnvelopeService{nil, "1", 0}, nil
 }
 
 // The factory method to create mock envelope kmsv2 service which always returns error.
 func newMockErrorEnvelopeKMSv2Service(endpoint string, timeout time.Duration) (kmsservice.Service, error) {
-	return &testKMSv2EnvelopeService{errors.New("test"), "1"}, nil
+	return &testKMSv2EnvelopeService{errors.New("test"), "1", 0}, nil
 }
 
 // The factory method to create mock envelope kmsv2 service that always returns invalid keyID.
 func newMockInvalidKeyIDEnvelopeKMSv2Service(ctx context.Context, endpoint string, timeout time.Duration, keyID string) (kmsservice.Service, error) {
-	return &testKMSv2EnvelopeService{nil, keyID}, nil
+	return &testKMSv2EnvelopeService{nil, keyID, 0}, nil
 }
 
 func TestLegacyConfig(t *testing.T) {
@@ -274,7 +280,7 @@ func TestEncryptionProviderConfigCorrect(t *testing.T) {
 	kmsFirstTransformer := kmsFirstEncryptionConfiguration.Transformers[schema.ParseGroupResource("secrets")]
 	kmsv2FirstTransformer := kmsv2FirstEncryptionConfiguration.Transformers[schema.ParseGroupResource("secrets")]
 
-	dataCtx := value.DefaultContext([]byte(sampleContextText))
+	dataCtx := value.DefaultContext(sampleContextText)
 	originalText := []byte(sampleText)
 
 	transformers := []struct {
@@ -566,7 +572,7 @@ func TestKMSPluginHealthz(t *testing.T) {
 		ttl:  3 * time.Second,
 	}
 	keyID := "1"
-	kmsv2Probe.keyID.Store(&keyID)
+	kmsv2Probe.state.Store(&envelopekmsv2.State{KeyID: keyID})
 
 	testCases := []struct {
 		desc    string
@@ -680,7 +686,7 @@ func TestKMSPluginHealthz(t *testing.T) {
 					p.service = nil
 					p.l = nil
 					p.lastResponse = nil
-					p.keyID.Store(kmsv2Probe.keyID.Load())
+					p.state.Store(kmsv2Probe.state.Load())
 				default:
 					t.Fatalf("unexpected probe type %T", p)
 				}
@@ -1367,6 +1373,7 @@ func TestKMSv2PluginHealthzTTL(t *testing.T) {
 
 	for _, tt := range testCases {
 		t.Run(tt.desc, func(t *testing.T) {
+			tt.probe.state.Store(&envelopekmsv2.State{})
 			_ = tt.probe.check(ctx)
 			if tt.probe.ttl != tt.wantTTL {
 				t.Fatalf("want ttl %v, got ttl %v", tt.wantTTL, tt.probe.ttl)
@@ -1445,6 +1452,7 @@ func TestKMSv2InvalidKeyID(t *testing.T) {
 	for _, tt := range testCases {
 		t.Run(tt.desc, func(t *testing.T) {
 			defer metrics.InvalidKeyIDFromStatusTotal.Reset()
+			tt.probe.state.Store(&envelopekmsv2.State{})
 			_ = tt.probe.check(ctx)
 			if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(tt.want), tt.metrics...); err != nil {
 				t.Fatal(err)
@@ -1477,7 +1485,7 @@ func testCBCKeyRotationWithProviders(t *testing.T, firstEncryptionConfig, firstP
 	p := getTransformerFromEncryptionConfig(t, firstEncryptionConfig)
 
 	ctx := testContext(t)
-	dataCtx := value.DefaultContext([]byte("authenticated_data"))
+	dataCtx := value.DefaultContext("authenticated_data")
 
 	out, err := p.TransformToStorage(ctx, []byte("firstvalue"), dataCtx)
 	if err != nil {
@@ -1495,7 +1503,7 @@ func testCBCKeyRotationWithProviders(t *testing.T, firstEncryptionConfig, firstP
 	}
 
 	// verify changing the context fails storage
-	_, _, err = p.TransformFromStorage(ctx, out, value.DefaultContext([]byte("incorrect_context")))
+	_, _, err = p.TransformFromStorage(ctx, out, value.DefaultContext("incorrect_context"))
 	if err != nil {
 		t.Fatalf("CBC mode does not support authentication: %v", err)
 	}
@@ -1544,9 +1552,12 @@ func getTransformerFromEncryptionConfig(t *testing.T, encryptionConfigPath strin
 }
 
 func TestIsKMSv2ProviderHealthyError(t *testing.T) {
+	probe := &kmsv2PluginProbe{name: "testplugin"}
+
 	testCases := []struct {
 		desc           string
 		expectedErr    string
+		wantMetrics    string
 		statusResponse *kmsservice.StatusResponse
 	}{
 		{
@@ -1554,14 +1565,24 @@ func TestIsKMSv2ProviderHealthyError(t *testing.T) {
 			statusResponse: &kmsservice.StatusResponse{
 				Healthz: "unhealthy",
 			},
-			expectedErr: "got unexpected healthz status: unhealthy, expected KMSv2 API version v2alpha1, got , expected KMSv2 KeyID to be set, got ",
+			expectedErr: "got unexpected healthz status: unhealthy, expected KMSv2 API version v2alpha1, got , got invalid KMSv2 KeyID ",
+			wantMetrics: `
+			# HELP apiserver_envelope_encryption_invalid_key_id_from_status_total [ALPHA] Number of times an invalid keyID is returned by the Status RPC call split by error.
+			# TYPE apiserver_envelope_encryption_invalid_key_id_from_status_total counter
+			apiserver_envelope_encryption_invalid_key_id_from_status_total{error="empty",provider_name="testplugin"} 1
+			`,
 		},
 		{
 			desc: "version is not v2alpha1",
 			statusResponse: &kmsservice.StatusResponse{
 				Version: "v1beta1",
 			},
-			expectedErr: "got unexpected healthz status: , expected KMSv2 API version v2alpha1, got v1beta1, expected KMSv2 KeyID to be set, got ",
+			expectedErr: "got unexpected healthz status: , expected KMSv2 API version v2alpha1, got v1beta1, got invalid KMSv2 KeyID ",
+			wantMetrics: `
+			# HELP apiserver_envelope_encryption_invalid_key_id_from_status_total [ALPHA] Number of times an invalid keyID is returned by the Status RPC call split by error.
+			# TYPE apiserver_envelope_encryption_invalid_key_id_from_status_total counter
+			apiserver_envelope_encryption_invalid_key_id_from_status_total{error="empty",provider_name="testplugin"} 1
+			`,
 		},
 		{
 			desc: "missing keyID",
@@ -1569,7 +1590,12 @@ func TestIsKMSv2ProviderHealthyError(t *testing.T) {
 				Healthz: "ok",
 				Version: "v2alpha1",
 			},
-			expectedErr: "expected KMSv2 KeyID to be set, got ",
+			expectedErr: "got invalid KMSv2 KeyID ",
+			wantMetrics: `
+			# HELP apiserver_envelope_encryption_invalid_key_id_from_status_total [ALPHA] Number of times an invalid keyID is returned by the Status RPC call split by error.
+			# TYPE apiserver_envelope_encryption_invalid_key_id_from_status_total counter
+			apiserver_envelope_encryption_invalid_key_id_from_status_total{error="empty",provider_name="testplugin"} 1
+			`,
 		},
 		{
 			desc: "invalid long keyID",
@@ -1578,16 +1604,27 @@ func TestIsKMSv2ProviderHealthyError(t *testing.T) {
 				Version: "v2alpha1",
 				KeyID:   sampleInvalidKeyID,
 			},
-			expectedErr: "expected KMSv2 KeyID to be set, got ",
+			expectedErr: "got invalid KMSv2 KeyID ",
+			wantMetrics: `
+			# HELP apiserver_envelope_encryption_invalid_key_id_from_status_total [ALPHA] Number of times an invalid keyID is returned by the Status RPC call split by error.
+			# TYPE apiserver_envelope_encryption_invalid_key_id_from_status_total counter
+			apiserver_envelope_encryption_invalid_key_id_from_status_total{error="too_long",provider_name="testplugin"} 1
+			`,
 		},
 	}
 
 	for _, tt := range testCases {
 		t.Run(tt.desc, func(t *testing.T) {
-			err := isKMSv2ProviderHealthy("testplugin", tt.statusResponse)
+			metrics.InvalidKeyIDFromStatusTotal.Reset()
+			err := probe.isKMSv2ProviderHealthyAndMaybeRotateDEK(testContext(t), tt.statusResponse)
 			if !strings.Contains(errString(err), tt.expectedErr) {
 				t.Errorf("expected err %q, got %q", tt.expectedErr, errString(err))
 			}
+			if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(tt.wantMetrics),
+				"apiserver_envelope_encryption_invalid_key_id_from_status_total",
+			); err != nil {
+				t.Fatal(err)
+			}
 		})
 	}
 }
@@ -1615,35 +1652,171 @@ func TestComputeEncryptionConfigHash(t *testing.T) {
 	}
 }
 
-func TestGetCurrentKeyID(t *testing.T) {
-	ctx := testContext(t)
-	kmsv2Probe := &kmsv2PluginProbe{
-		name: "foo",
-		ttl:  3 * time.Second,
+func Test_kmsv2PluginProbe_rotateDEKOnKeyIDChange(t *testing.T) {
+	origNowFunc := envelopekmsv2.NowFunc
+	now := origNowFunc() // freeze time
+	t.Cleanup(func() { envelopekmsv2.NowFunc = origNowFunc })
+	envelopekmsv2.NowFunc = func() time.Time { return now }
+
+	klog.LogToStderr(false)
+	var level klog.Level
+	if err := level.Set("6"); err != nil {
+		t.Fatal(err)
 	}
-	testCases := []struct {
-		desc        string
-		keyID       string
-		expectedErr string
+	t.Cleanup(func() {
+		klog.LogToStderr(true)
+		if err := level.Set("0"); err != nil {
+			t.Fatal(err)
+		}
+		klog.SetOutput(io.Discard)
+	})
+
+	tests := []struct {
+		name             string
+		service          *testKMSv2EnvelopeService
+		state            envelopekmsv2.State
+		statusKeyID      string
+		wantState        envelopekmsv2.State
+		wantEncryptCalls int
+		wantLogs         []string
+		wantErr          string
 	}{
 		{
-			desc:        "empty keyID",
-			keyID:       "",
-			expectedErr: "got unexpected empty keyID",
+			name:        "happy path, no previous state",
+			service:     &testKMSv2EnvelopeService{keyID: "1"},
+			state:       envelopekmsv2.State{},
+			statusKeyID: "1",
+			wantState: envelopekmsv2.State{
+				KeyID:               "1",
+				ExpirationTimestamp: now.Add(3 * time.Minute),
+			},
+			wantEncryptCalls: 1,
+			wantLogs: []string{
+				`"encrypting content using envelope service" uid="panda"`,
+				fmt.Sprintf(`"successfully rotated DEK" uid="panda" newKeyID="1" oldKeyID="" expirationTimestamp="%s"`,
+					now.Add(3*time.Minute).Format(time.RFC3339)),
+			},
+			wantErr: "",
 		},
 		{
-			desc:        "valid keyID",
-			keyID:       "1",
-			expectedErr: "",
+			name:        "happy path, with previous state",
+			service:     &testKMSv2EnvelopeService{err: fmt.Errorf("broken")}, // not called
+			state:       validState("2", now),
+			statusKeyID: "2",
+			wantState: envelopekmsv2.State{
+				KeyID:               "2",
+				ExpirationTimestamp: now.Add(3 * time.Minute),
+			},
+			wantEncryptCalls: 0,
+			wantLogs:         nil,
+			wantErr:          "",
+		},
+		{
+			name:        "previous state expired but key ID matches",
+			service:     &testKMSv2EnvelopeService{err: fmt.Errorf("broken")}, // not called
+			state:       validState("3", now.Add(-time.Hour)),
+			statusKeyID: "3",
+			wantState: envelopekmsv2.State{
+				KeyID:               "3",
+				ExpirationTimestamp: now.Add(3 * time.Minute),
+			},
+			wantEncryptCalls: 0,
+			wantLogs:         nil,
+			wantErr:          "",
+		},
+		{
+			name:        "previous state expired but key ID does not match",
+			service:     &testKMSv2EnvelopeService{keyID: "4"},
+			state:       validState("3", now.Add(-time.Hour)),
+			statusKeyID: "4",
+			wantState: envelopekmsv2.State{
+				KeyID:               "4",
+				ExpirationTimestamp: now.Add(3 * time.Minute),
+			},
+			wantEncryptCalls: 1,
+			wantLogs: []string{
+				`"encrypting content using envelope service" uid="panda"`,
+				fmt.Sprintf(`"successfully rotated DEK" uid="panda" newKeyID="4" oldKeyID="3" expirationTimestamp="%s"`,
+					now.Add(3*time.Minute).Format(time.RFC3339)),
+			},
+			wantErr: "",
+		},
+		{
+			name:        "service down but key ID does not match",
+			service:     &testKMSv2EnvelopeService{err: fmt.Errorf("broken")},
+			state:       validState("4", now.Add(7*time.Minute)),
+			statusKeyID: "5",
+			wantState: envelopekmsv2.State{
+				KeyID:               "4",
+				ExpirationTimestamp: now.Add(7 * time.Minute),
+			},
+			wantEncryptCalls: 1,
+			wantLogs: []string{
+				`"encrypting content using envelope service" uid="panda"`,
+			},
+			wantErr: `failed to rotate DEK uid="panda", ` +
+				`errState=<nil>, errGen=failed to encrypt DEK, error: broken, statusKeyID="5", ` +
+				`encryptKeyID="", stateKeyID="4", expirationTimestamp=` + now.Add(7*time.Minute).Format(time.RFC3339),
 		},
 	}
-	for _, tt := range testCases {
-		t.Run(tt.desc, func(t *testing.T) {
-			kmsv2Probe.keyID.Store(&tt.keyID)
-			_, err := kmsv2Probe.getCurrentKeyID(ctx)
-			if errString(err) != tt.expectedErr {
-				t.Errorf("expected err %q, got %q", tt.expectedErr, errString(err))
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var buf bytes.Buffer
+			klog.SetOutput(&buf)
+
+			ctx := testContext(t)
+
+			h := &kmsv2PluginProbe{
+				name:    "panda",
+				service: tt.service,
+			}
+			h.state.Store(&tt.state)
+
+			err := h.rotateDEKOnKeyIDChange(ctx, tt.statusKeyID, "panda")
+
+			klog.Flush()
+			klog.SetOutput(io.Discard) // prevent further writes into buf
+
+			if diff := cmp.Diff(tt.wantLogs, logLines(buf.String())); len(diff) > 0 {
+				t.Errorf("log mismatch (-want +got):\n%s", diff)
+			}
+
+			ignoredFields := sets.NewString("Transformer", "EncryptedDEK", "UID")
+
+			if diff := cmp.Diff(tt.wantState, *h.state.Load(),
+				cmp.FilterPath(func(path cmp.Path) bool { return ignoredFields.Has(path.String()) }, cmp.Ignore()),
+			); len(diff) > 0 {
+				t.Errorf("state mismatch (-want +got):\n%s", diff)
+			}
+
+			if tt.wantEncryptCalls != tt.service.encryptCalls {
+				t.Errorf("want %d encryptCalls, got %d", tt.wantEncryptCalls, tt.service.encryptCalls)
+			}
+
+			if errString(err) != tt.wantErr {
+				t.Errorf("rotateDEKOnKeyIDChange() error = %v, wantErr %v", err, tt.wantErr)
 			}
 		})
 	}
 }
+
+func validState(keyID string, exp time.Time) envelopekmsv2.State {
+	return envelopekmsv2.State{
+		Transformer:         &resourceTransformer{},
+		EncryptedDEK:        []byte{1},
+		KeyID:               keyID,
+		ExpirationTimestamp: exp,
+	}
+}
+
+func logLines(logs string) []string {
+	if len(logs) == 0 {
+		return nil
+	}
+
+	lines := strings.Split(strings.TrimSpace(logs), "\n")
+	for i, line := range lines {
+		lines[i] = strings.SplitN(line, "] ", 2)[1]
+	}
+	return lines
+}
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/aes/aes.go b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/aes/aes.go
index 69930c03908..b26c92e2d55 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/aes/aes.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/aes/aes.go
@@ -23,14 +23,24 @@ import (
 	"crypto/aes"
 	"crypto/cipher"
 	"crypto/rand"
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"io"
+	"sync/atomic"
+	"time"
 
 	"k8s.io/apiserver/pkg/storage/value"
+	"k8s.io/klog/v2"
 )
 
-// gcm implements AEAD encryption of the provided values given a cipher.Block algorithm.
+type gcm struct {
+	aead      cipher.AEAD
+	nonceFunc func([]byte) error
+}
+
+// NewGCMTransformer takes the given block cipher and performs encryption and decryption on the given data.
+// It implements AEAD encryption of the provided values given a cipher.Block algorithm.
 // The authenticated data provided as part of the value.Context method must match when the same
 // value is set to and loaded from storage. In order to ensure that values cannot be copied by
 // an attacker from a location under their control, use characteristics of the storage location
@@ -43,44 +53,148 @@ import (
 // therefore transformers using this implementation *must* ensure they allow for frequent key
 // rotation. Future work should include investigation of AES-GCM-SIV as an alternative to
 // random nonces.
-type gcm struct {
-	block cipher.Block
+func NewGCMTransformer(block cipher.Block) (value.Transformer, error) {
+	aead, err := newGCM(block)
+	if err != nil {
+		return nil, err
+	}
+
+	return &gcm{aead: aead, nonceFunc: randomNonce}, nil
 }
 
-// NewGCMTransformer takes the given block cipher and performs encryption and decryption on the given
-// data.
-func NewGCMTransformer(block cipher.Block) value.Transformer {
-	return &gcm{block: block}
+// NewGCMTransformerWithUniqueKeyUnsafe is the same as NewGCMTransformer but is unsafe for general
+// use because it makes assumptions about the key underlying the block cipher.  Specifically,
+// it uses a 96-bit nonce where the first 32 bits are random data and the remaining 64 bits are
+// a monotonically incrementing atomic counter.  This means that the key must be randomly generated
+// on process startup and must never be used for encryption outside the lifetime of the process.
+// Unlike NewGCMTransformer, this function is immune to the birthday attack and thus the key can
+// be used for 2^64-1 writes without rotation.  Furthermore, cryptographic wear out of AES-GCM with
+// a sequential nonce occurs after 2^64 encryptions, which is not a concern for our use cases.
+// Even if that occurs, the nonce counter would overflow and crash the process.  We have no concerns
+// around plaintext length because all stored items are small (less than 2 MB).  To prevent the
+// chance of the block cipher being accidentally re-used, it is not taken in as input.  Instead,
+// a new random key is generated and returned on every invocation of this function.  This key is
+// used as the input to the block cipher.  If the key is stored and retrieved at a later point,
+// it can be passed to NewGCMTransformer(aes.NewCipher(key)) to construct a transformer capable
+// of decrypting values encrypted by this transformer (that transformer must not be used for encryption).
+func NewGCMTransformerWithUniqueKeyUnsafe() (value.Transformer, []byte, error) {
+	key, err := generateKey(32)
+	if err != nil {
+		return nil, nil, err
+	}
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	nonceGen := &nonceGenerator{
+		// we start the nonce counter at one billion so that we are
+		// guaranteed to detect rollover across different go routines
+		zero:  1_000_000_000,
+		fatal: die,
+	}
+	nonceGen.nonce.Add(nonceGen.zero)
+
+	transformer, err := newGCMTransformerWithUniqueKeyUnsafe(block, nonceGen)
+	if err != nil {
+		return nil, nil, err
+	}
+	return transformer, key, nil
+}
+
+func newGCMTransformerWithUniqueKeyUnsafe(block cipher.Block, nonceGen *nonceGenerator) (value.Transformer, error) {
+	aead, err := newGCM(block)
+	if err != nil {
+		return nil, err
+	}
+
+	nonceFunc := func(b []byte) error {
+		// we only need 8 bytes to store our 64 bit incrementing nonce
+		// instead of leaving the unused bytes as zeros, set those to random bits
+		// this mostly protects us from weird edge cases like a VM restore that rewinds our atomic counter
+		randNonceSize := len(b) - 8
+
+		if err := randomNonce(b[:randNonceSize]); err != nil {
+			return err
+		}
+
+		nonceGen.next(b[randNonceSize:])
+
+		return nil
+	}
+
+	return &gcm{aead: aead, nonceFunc: nonceFunc}, nil
+}
+
+func newGCM(block cipher.Block) (cipher.AEAD, error) {
+	aead, err := cipher.NewGCM(block)
+	if err != nil {
+		return nil, err
+	}
+	if nonceSize := aead.NonceSize(); nonceSize != 12 { // all data in etcd will be broken if this ever changes
+		return nil, fmt.Errorf("crypto/cipher.NewGCM returned unexpected nonce size: %d", nonceSize)
+	}
+	return aead, nil
+}
+
+func randomNonce(b []byte) error {
+	_, err := rand.Read(b)
+	return err
+}
+
+type nonceGenerator struct {
+	// even at one million encryptions per second, this counter is enough for half a million years
+	// using this struct avoids alignment bugs: https://pkg.go.dev/sync/atomic#pkg-note-BUG
+	nonce atomic.Uint64
+	zero  uint64
+	fatal func(msg string)
+}
+
+func (n *nonceGenerator) next(b []byte) {
+	incrementingNonce := n.nonce.Add(1)
+	if incrementingNonce <= n.zero {
+		// this should never happen, and is unrecoverable if it does
+		n.fatal("aes-gcm detected nonce overflow - cryptographic wear out has occurred")
+	}
+	binary.LittleEndian.PutUint64(b, incrementingNonce)
+}
+
+func die(msg string) {
+	// nolint:logcheck // we want the stack traces, log flushing, and process exiting logic from FatalDepth
+	klog.FatalDepth(1, msg)
+}
+
+// generateKey generates a random key using system randomness.
+func generateKey(length int) (key []byte, err error) {
+	defer func(start time.Time) {
+		value.RecordDataKeyGeneration(start, err)
+	}(time.Now())
+	key = make([]byte, length)
+	if _, err = rand.Read(key); err != nil {
+		return nil, err
+	}
+
+	return key, nil
 }
 
 func (t *gcm) TransformFromStorage(ctx context.Context, data []byte, dataCtx value.Context) ([]byte, bool, error) {
-	aead, err := cipher.NewGCM(t.block)
-	if err != nil {
-		return nil, false, err
-	}
-	nonceSize := aead.NonceSize()
+	nonceSize := t.aead.NonceSize()
 	if len(data) < nonceSize {
-		return nil, false, fmt.Errorf("the stored data was shorter than the required size")
+		return nil, false, errors.New("the stored data was shorter than the required size")
 	}
-	result, err := aead.Open(nil, data[:nonceSize], data[nonceSize:], dataCtx.AuthenticatedData())
+	result, err := t.aead.Open(nil, data[:nonceSize], data[nonceSize:], dataCtx.AuthenticatedData())
 	return result, false, err
 }
 
 func (t *gcm) TransformToStorage(ctx context.Context, data []byte, dataCtx value.Context) ([]byte, error) {
-	aead, err := cipher.NewGCM(t.block)
-	if err != nil {
-		return nil, err
+	nonceSize := t.aead.NonceSize()
+	result := make([]byte, nonceSize+t.aead.Overhead()+len(data))
+
+	if err := t.nonceFunc(result[:nonceSize]); err != nil {
+		return nil, fmt.Errorf("failed to write nonce for AES-GCM: %w", err)
 	}
-	nonceSize := aead.NonceSize()
-	result := make([]byte, nonceSize+aead.Overhead()+len(data))
-	n, err := rand.Read(result[:nonceSize])
-	if err != nil {
-		return nil, err
-	}
-	if n != nonceSize {
-		return nil, fmt.Errorf("unable to read sufficient random bytes")
-	}
-	cipherText := aead.Seal(result[nonceSize:nonceSize], result[:nonceSize], data, dataCtx.AuthenticatedData())
+
+	cipherText := t.aead.Seal(result[nonceSize:nonceSize], result[:nonceSize], data, dataCtx.AuthenticatedData())
 	return result[:nonceSize+len(cipherText)], nil
 }
 
@@ -96,7 +210,7 @@ func NewCBCTransformer(block cipher.Block) value.Transformer {
 }
 
 var (
-	ErrInvalidBlockSize    = fmt.Errorf("the stored data is not a multiple of the block size")
+	errInvalidBlockSize    = errors.New("the stored data is not a multiple of the block size")
 	errInvalidPKCS7Data    = errors.New("invalid PKCS7 data (empty or not padded)")
 	errInvalidPKCS7Padding = errors.New("invalid padding on input")
 )
@@ -104,13 +218,13 @@ var (
 func (t *cbc) TransformFromStorage(ctx context.Context, data []byte, dataCtx value.Context) ([]byte, bool, error) {
 	blockSize := aes.BlockSize
 	if len(data) < blockSize {
-		return nil, false, fmt.Errorf("the stored data was shorter than the required size")
+		return nil, false, errors.New("the stored data was shorter than the required size")
 	}
 	iv := data[:blockSize]
 	data = data[blockSize:]
 
 	if len(data)%blockSize != 0 {
-		return nil, false, ErrInvalidBlockSize
+		return nil, false, errInvalidBlockSize
 	}
 
 	result := make([]byte, len(data))
@@ -140,7 +254,7 @@ func (t *cbc) TransformToStorage(ctx context.Context, data []byte, dataCtx value
 	result := make([]byte, blockSize+len(data)+paddingSize)
 	iv := result[:blockSize]
 	if _, err := io.ReadFull(rand.Reader, iv); err != nil {
-		return nil, fmt.Errorf("unable to read sufficient random bytes")
+		return nil, errors.New("unable to read sufficient random bytes")
 	}
 	copy(result[blockSize:], data)
 
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/aes/aes_test.go b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/aes/aes_test.go
index 37564c1db0d..87132397611 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/aes/aes_test.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/aes/aes_test.go
@@ -22,10 +22,14 @@ import (
 	"crypto/aes"
 	"crypto/cipher"
 	"crypto/rand"
+	"encoding/binary"
 	"encoding/hex"
 	"fmt"
 	"io"
+	"math"
 	"reflect"
+	"sync"
+	"sync/atomic"
 	"testing"
 
 	"k8s.io/apiserver/pkg/storage/value"
@@ -42,11 +46,289 @@ func TestGCMDataStable(t *testing.T) {
 	}
 	// IMPORTANT: If you must fix this test, then all previously encrypted data from previously compiled versions is broken unless you hardcode the nonce size to 12
 	if aead.NonceSize() != 12 {
-		t.Fatalf("The underlying Golang crypto size has changed, old version of AES on disk will not be readable unless the AES implementation is changed to hardcode nonce size.")
+		t.Errorf("The underlying Golang crypto size has changed, old version of AES on disk will not be readable unless the AES implementation is changed to hardcode nonce size.")
+	}
+
+	transformerCounterNonce, _, err := NewGCMTransformerWithUniqueKeyUnsafe()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if nonceSize := transformerCounterNonce.(*gcm).aead.NonceSize(); nonceSize != 12 {
+		t.Errorf("counter nonce: backwards incompatible change to nonce size detected: %d", nonceSize)
+	}
+
+	transformerRandomNonce, err := NewGCMTransformer(block)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if nonceSize := transformerRandomNonce.(*gcm).aead.NonceSize(); nonceSize != 12 {
+		t.Errorf("random nonce: backwards incompatible change to nonce size detected: %d", nonceSize)
+	}
+}
+
+func TestGCMUnsafeNonceOverflow(t *testing.T) {
+	var msgFatal string
+	var count int
+
+	nonceGen := &nonceGenerator{
+		fatal: func(msg string) {
+			msgFatal = msg
+			count++
+		},
+	}
+
+	block, err := aes.NewCipher([]byte("abcdefghijklmnop"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	transformer, err := newGCMTransformerWithUniqueKeyUnsafe(block, nonceGen)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assertNonce(t, &nonceGen.nonce, 0)
+
+	runEncrypt(t, transformer)
+
+	assertNonce(t, &nonceGen.nonce, 1)
+
+	runEncrypt(t, transformer)
+
+	assertNonce(t, &nonceGen.nonce, 2)
+
+	nonceGen.nonce.Store(math.MaxUint64 - 1) // pretend lots of encryptions occurred
+
+	runEncrypt(t, transformer)
+
+	assertNonce(t, &nonceGen.nonce, math.MaxUint64)
+
+	if count != 0 {
+		t.Errorf("fatal should not have been called yet")
+	}
+
+	runEncrypt(t, transformer)
+
+	assertNonce(t, &nonceGen.nonce, 0)
+
+	if count != 1 {
+		t.Errorf("fatal should have been once, got %d", count)
+	}
+
+	if msgFatal != "aes-gcm detected nonce overflow - cryptographic wear out has occurred" {
+		t.Errorf("unexpected message: %s", msgFatal)
+	}
+}
+
+func assertNonce(t *testing.T, nonce *atomic.Uint64, want uint64) {
+	t.Helper()
+
+	if got := nonce.Load(); want != got {
+		t.Errorf("nonce should equal %d, got %d", want, got)
+	}
+}
+
+func runEncrypt(t *testing.T, transformer value.Transformer) {
+	t.Helper()
+
+	ctx := context.Background()
+	dataCtx := value.DefaultContext("authenticated_data")
+
+	_, err := transformer.TransformToStorage(ctx, []byte("firstvalue"), dataCtx)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+// TestGCMUnsafeCompatibility asserts that encryptions performed via
+// NewGCMTransformerWithUniqueKeyUnsafe can be decrypted via NewGCMTransformer.
+func TestGCMUnsafeCompatibility(t *testing.T) {
+	transformerEncrypt, key, err := NewGCMTransformerWithUniqueKeyUnsafe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	transformerDecrypt := newGCMTransformer(t, block)
+
+	ctx := context.Background()
+	dataCtx := value.DefaultContext("authenticated_data")
+
+	plaintext := []byte("firstvalue")
+
+	ciphertext, err := transformerEncrypt.TransformToStorage(ctx, plaintext, dataCtx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if bytes.Equal(plaintext, ciphertext) {
+		t.Errorf("plaintext %q matches ciphertext %q", string(plaintext), string(ciphertext))
+	}
+
+	plaintextAgain, _, err := transformerDecrypt.TransformFromStorage(ctx, ciphertext, dataCtx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !bytes.Equal(plaintext, plaintextAgain) {
+		t.Errorf("expected original plaintext %q, got %q", string(plaintext), string(plaintextAgain))
+	}
+}
+
+func TestGCMLegacyDataCompatibility(t *testing.T) {
+	block, err := aes.NewCipher([]byte("snorlax_awesomes"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	transformerDecrypt := newGCMTransformer(t, block)
+
+	// recorded output from NewGCMTransformer at commit 3b1fc60d8010dd8b53e97ba80e4710dbb430beee
+	const legacyCiphertext = "\x9f'\xc8\xfc\xea\x8aX\xc4g\xd8\xe47\xdb\xf2\xd8YU\xf9\xb4\xbd\x91/N\xf9g\u05c8\xa0\xcb\ay}\xac\n?\n\bE`\\\xa8Z\xc8V+J\xe1"
+
+	ctx := context.Background()
+	dataCtx := value.DefaultContext("bamboo")
+
+	plaintext := []byte("pandas are the best")
+
+	plaintextAgain, _, err := transformerDecrypt.TransformFromStorage(ctx, []byte(legacyCiphertext), dataCtx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !bytes.Equal(plaintext, plaintextAgain) {
+		t.Errorf("expected original plaintext %q, got %q", string(plaintext), string(plaintextAgain))
+	}
+}
+
+func TestGCMUnsafeNonceGen(t *testing.T) {
+	block, err := aes.NewCipher([]byte("abcdefghijklmnop"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	transformer := newGCMTransformerWithUniqueKeyUnsafeTest(t, block)
+
+	ctx := context.Background()
+	dataCtx := value.DefaultContext("authenticated_data")
+
+	const count = 1_000
+
+	counters := make([]uint64, count)
+
+	// run a bunch of go routines to make sure we are go routine safe
+	// on both the nonce generation and the actual encryption/decryption
+	var wg sync.WaitGroup
+	for i := 0; i < count; i++ {
+		i := i
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			plaintext := bytes.Repeat([]byte{byte(i % 8)}, count)
+
+			out, err := transformer.TransformToStorage(ctx, plaintext, dataCtx)
+			if err != nil {
+				t.Error(err)
+				return
+			}
+
+			nonce := out[:12]
+			randomN := nonce[:4]
+
+			if bytes.Equal(randomN, make([]byte, len(randomN))) {
+				t.Error("got all zeros for random four byte nonce")
+			}
+
+			counter := nonce[4:]
+			counters[binary.LittleEndian.Uint64(counter)-1]++ // subtract one because the counter starts at 1, not 0
+
+			plaintextAgain, _, err := transformer.TransformFromStorage(ctx, out, dataCtx)
+			if err != nil {
+				t.Error(err)
+				return
+			}
+
+			if !bytes.Equal(plaintext, plaintextAgain) {
+				t.Errorf("expected original plaintext %q, got %q", string(plaintext), string(plaintextAgain))
+			}
+		}()
+	}
+	wg.Wait()
+
+	want := make([]uint64, count)
+	for i := range want {
+		want[i] = 1
+	}
+
+	if !reflect.DeepEqual(want, counters) {
+		t.Error("unexpected counter state")
+	}
+}
+
+func TestGCMNonce(t *testing.T) {
+	t.Run("gcm", func(t *testing.T) {
+		testGCMNonce(t, newGCMTransformer, func(_ int, nonce []byte) {
+			if bytes.Equal(nonce, make([]byte, len(nonce))) {
+				t.Error("got all zeros for nonce")
+			}
+		})
+	})
+
+	t.Run("gcm unsafe", func(t *testing.T) {
+		testGCMNonce(t, newGCMTransformerWithUniqueKeyUnsafeTest, func(i int, nonce []byte) {
+			counter := binary.LittleEndian.Uint64(nonce)
+			if uint64(i+1) != counter { // add one because the counter starts at 1, not 0
+				t.Errorf("counter nonce is invalid: want %d, got %d", i+1, counter)
+			}
+		})
+	})
+}
+
+func testGCMNonce(t *testing.T, f func(t testingT, block cipher.Block) value.Transformer, check func(int, []byte)) {
+	block, err := aes.NewCipher([]byte("abcdefghijklmnop"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	transformer := f(t, block)
+
+	ctx := context.Background()
+	dataCtx := value.DefaultContext("authenticated_data")
+
+	const count = 1_000
+
+	for i := 0; i < count; i++ {
+		i := i
+
+		out, err := transformer.TransformToStorage(ctx, bytes.Repeat([]byte{byte(i % 8)}, count), dataCtx)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		nonce := out[:12]
+		randomN := nonce[:4]
+
+		if bytes.Equal(randomN, make([]byte, len(randomN))) {
+			t.Error("got all zeros for first four bytes")
+		}
+
+		check(i, nonce[4:])
 	}
 }
 
 func TestGCMKeyRotation(t *testing.T) {
+	t.Run("gcm", func(t *testing.T) {
+		testGCMKeyRotation(t, newGCMTransformer)
+	})
+
+	t.Run("gcm unsafe", func(t *testing.T) {
+		testGCMKeyRotation(t, newGCMTransformerWithUniqueKeyUnsafeTest)
+	})
+}
+
+func testGCMKeyRotation(t *testing.T, f func(t testingT, block cipher.Block) value.Transformer) {
 	testErr := fmt.Errorf("test error")
 	block1, err := aes.NewCipher([]byte("abcdefghijklmnop"))
 	if err != nil {
@@ -58,11 +340,11 @@ func TestGCMKeyRotation(t *testing.T) {
 	}
 
 	ctx := context.Background()
-	dataCtx := value.DefaultContext([]byte("authenticated_data"))
+	dataCtx := value.DefaultContext("authenticated_data")
 
 	p := value.NewPrefixTransformers(testErr,
-		value.PrefixTransformer{Prefix: []byte("first:"), Transformer: NewGCMTransformer(block1)},
-		value.PrefixTransformer{Prefix: []byte("second:"), Transformer: NewGCMTransformer(block2)},
+		value.PrefixTransformer{Prefix: []byte("first:"), Transformer: f(t, block1)},
+		value.PrefixTransformer{Prefix: []byte("second:"), Transformer: f(t, block2)},
 	)
 	out, err := p.TransformToStorage(ctx, []byte("firstvalue"), dataCtx)
 	if err != nil {
@@ -80,15 +362,15 @@ func TestGCMKeyRotation(t *testing.T) {
 	}
 
 	// verify changing the context fails storage
-	_, _, err = p.TransformFromStorage(ctx, out, value.DefaultContext([]byte("incorrect_context")))
+	_, _, err = p.TransformFromStorage(ctx, out, value.DefaultContext("incorrect_context"))
 	if err == nil {
 		t.Fatalf("expected unauthenticated data")
 	}
 
 	// reverse the order, use the second key
 	p = value.NewPrefixTransformers(testErr,
-		value.PrefixTransformer{Prefix: []byte("second:"), Transformer: NewGCMTransformer(block2)},
-		value.PrefixTransformer{Prefix: []byte("first:"), Transformer: NewGCMTransformer(block1)},
+		value.PrefixTransformer{Prefix: []byte("second:"), Transformer: f(t, block2)},
+		value.PrefixTransformer{Prefix: []byte("first:"), Transformer: f(t, block1)},
 	)
 	from, stale, err = p.TransformFromStorage(ctx, out, dataCtx)
 	if err != nil {
@@ -111,7 +393,7 @@ func TestCBCKeyRotation(t *testing.T) {
 	}
 
 	ctx := context.Background()
-	dataCtx := value.DefaultContext([]byte("authenticated_data"))
+	dataCtx := value.DefaultContext("authenticated_data")
 
 	p := value.NewPrefixTransformers(testErr,
 		value.PrefixTransformer{Prefix: []byte("first:"), Transformer: NewCBCTransformer(block1)},
@@ -133,7 +415,7 @@ func TestCBCKeyRotation(t *testing.T) {
 	}
 
 	// verify changing the context fails storage
-	_, _, err = p.TransformFromStorage(ctx, out, value.DefaultContext([]byte("incorrect_context")))
+	_, _, err = p.TransformFromStorage(ctx, out, value.DefaultContext("incorrect_context"))
 	if err != nil {
 		t.Fatalf("CBC mode does not support authentication: %v", err)
 	}
@@ -198,12 +480,12 @@ func benchmarkGCMRead(b *testing.B, keyLength int, valueLength int, expectStale
 		b.Fatal(err)
 	}
 	p := value.NewPrefixTransformers(nil,
-		value.PrefixTransformer{Prefix: []byte("first:"), Transformer: NewGCMTransformer(block1)},
-		value.PrefixTransformer{Prefix: []byte("second:"), Transformer: NewGCMTransformer(block2)},
+		value.PrefixTransformer{Prefix: []byte("first:"), Transformer: newGCMTransformer(b, block1)},
+		value.PrefixTransformer{Prefix: []byte("second:"), Transformer: newGCMTransformer(b, block2)},
 	)
 
 	ctx := context.Background()
-	dataCtx := value.DefaultContext([]byte("authenticated_data"))
+	dataCtx := value.DefaultContext("authenticated_data")
 	v := bytes.Repeat([]byte("0123456789abcdef"), valueLength/16)
 
 	out, err := p.TransformToStorage(ctx, v, dataCtx)
@@ -213,8 +495,8 @@ func benchmarkGCMRead(b *testing.B, keyLength int, valueLength int, expectStale
 	// reverse the key order if expecting stale
 	if expectStale {
 		p = value.NewPrefixTransformers(nil,
-			value.PrefixTransformer{Prefix: []byte("second:"), Transformer: NewGCMTransformer(block2)},
-			value.PrefixTransformer{Prefix: []byte("first:"), Transformer: NewGCMTransformer(block1)},
+			value.PrefixTransformer{Prefix: []byte("second:"), Transformer: newGCMTransformer(b, block2)},
+			value.PrefixTransformer{Prefix: []byte("first:"), Transformer: newGCMTransformer(b, block1)},
 		)
 	}
 
@@ -241,12 +523,12 @@ func benchmarkGCMWrite(b *testing.B, keyLength int, valueLength int) {
 		b.Fatal(err)
 	}
 	p := value.NewPrefixTransformers(nil,
-		value.PrefixTransformer{Prefix: []byte("first:"), Transformer: NewGCMTransformer(block1)},
-		value.PrefixTransformer{Prefix: []byte("second:"), Transformer: NewGCMTransformer(block2)},
+		value.PrefixTransformer{Prefix: []byte("first:"), Transformer: newGCMTransformer(b, block1)},
+		value.PrefixTransformer{Prefix: []byte("second:"), Transformer: newGCMTransformer(b, block2)},
 	)
 
 	ctx := context.Background()
-	dataCtx := value.DefaultContext([]byte("authenticated_data"))
+	dataCtx := value.DefaultContext("authenticated_data")
 	v := bytes.Repeat([]byte("0123456789abcdef"), valueLength/16)
 
 	b.ResetTimer()
@@ -308,7 +590,7 @@ func benchmarkCBCRead(b *testing.B, keyLength int, valueLength int, expectStale
 	)
 
 	ctx := context.Background()
-	dataCtx := value.DefaultContext([]byte("authenticated_data"))
+	dataCtx := value.DefaultContext("authenticated_data")
 	v := bytes.Repeat([]byte("0123456789abcdef"), valueLength/16)
 
 	out, err := p.TransformToStorage(ctx, v, dataCtx)
@@ -351,7 +633,7 @@ func benchmarkCBCWrite(b *testing.B, keyLength int, valueLength int) {
 	)
 
 	ctx := context.Background()
-	dataCtx := value.DefaultContext([]byte("authenticated_data"))
+	dataCtx := value.DefaultContext("authenticated_data")
 	v := bytes.Repeat([]byte("0123456789abcdef"), valueLength/16)
 
 	b.ResetTimer()
@@ -367,15 +649,15 @@ func benchmarkCBCWrite(b *testing.B, keyLength int, valueLength int) {
 func TestRoundTrip(t *testing.T) {
 	lengths := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 128, 1024}
 
-	aes16block, err := aes.NewCipher([]byte(bytes.Repeat([]byte("a"), 16)))
+	aes16block, err := aes.NewCipher(bytes.Repeat([]byte("a"), 16))
 	if err != nil {
 		t.Fatal(err)
 	}
-	aes24block, err := aes.NewCipher([]byte(bytes.Repeat([]byte("b"), 24)))
+	aes24block, err := aes.NewCipher(bytes.Repeat([]byte("b"), 24))
 	if err != nil {
 		t.Fatal(err)
 	}
-	aes32block, err := aes.NewCipher([]byte(bytes.Repeat([]byte("c"), 32)))
+	aes32block, err := aes.NewCipher(bytes.Repeat([]byte("c"), 32))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -386,16 +668,19 @@ func TestRoundTrip(t *testing.T) {
 		dataCtx value.Context
 		t       value.Transformer
 	}{
-		{name: "GCM 16 byte key", t: NewGCMTransformer(aes16block)},
-		{name: "GCM 24 byte key", t: NewGCMTransformer(aes24block)},
-		{name: "GCM 32 byte key", t: NewGCMTransformer(aes32block)},
+		{name: "GCM 16 byte key", t: newGCMTransformer(t, aes16block)},
+		{name: "GCM 24 byte key", t: newGCMTransformer(t, aes24block)},
+		{name: "GCM 32 byte key", t: newGCMTransformer(t, aes32block)},
+		{name: "GCM 16 byte unsafe key", t: newGCMTransformerWithUniqueKeyUnsafeTest(t, aes16block)},
+		{name: "GCM 24 byte unsafe key", t: newGCMTransformerWithUniqueKeyUnsafeTest(t, aes24block)},
+		{name: "GCM 32 byte unsafe key", t: newGCMTransformerWithUniqueKeyUnsafeTest(t, aes32block)},
 		{name: "CBC 32 byte key", t: NewCBCTransformer(aes32block)},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			dataCtx := tt.dataCtx
 			if dataCtx == nil {
-				dataCtx = value.DefaultContext([]byte(""))
+				dataCtx = value.DefaultContext("")
 			}
 			for _, l := range lengths {
 				data := make([]byte, l)
@@ -432,3 +717,31 @@ func TestRoundTrip(t *testing.T) {
 		})
 	}
 }
+
+type testingT interface {
+	Helper()
+	Fatal(...any)
+}
+
+func newGCMTransformer(t testingT, block cipher.Block) value.Transformer {
+	t.Helper()
+
+	transformer, err := NewGCMTransformer(block)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return transformer
+}
+
+func newGCMTransformerWithUniqueKeyUnsafeTest(t testingT, block cipher.Block) value.Transformer {
+	t.Helper()
+
+	nonceGen := &nonceGenerator{fatal: die}
+	transformer, err := newGCMTransformerWithUniqueKeyUnsafe(block, nonceGen)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return transformer
+}
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/envelope.go b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/envelope.go
index 43d2e00a22f..4bb18ee8baf 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/envelope.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/envelope.go
@@ -53,7 +53,7 @@ type envelopeTransformer struct {
 	transformers *lru.Cache
 
 	// baseTransformerFunc creates a new transformer for encrypting the data with the DEK.
-	baseTransformerFunc func(cipher.Block) value.Transformer
+	baseTransformerFunc func(cipher.Block) (value.Transformer, error)
 
 	cacheSize    int
 	cacheEnabled bool
@@ -63,7 +63,7 @@ type envelopeTransformer struct {
 // It uses envelopeService to encrypt and decrypt DEKs. Respective DEKs (in encrypted form) are prepended to
 // the data items they encrypt. A cache (of size cacheSize) is maintained to store the most recently
 // used decrypted DEKs in memory.
-func NewEnvelopeTransformer(envelopeService Service, cacheSize int, baseTransformerFunc func(cipher.Block) value.Transformer) value.Transformer {
+func NewEnvelopeTransformer(envelopeService Service, cacheSize int, baseTransformerFunc func(cipher.Block) (value.Transformer, error)) value.Transformer {
 	var (
 		cache *lru.Cache
 	)
@@ -161,7 +161,11 @@ func (t *envelopeTransformer) addTransformer(encKey []byte, key []byte) (value.T
 	if err != nil {
 		return nil, err
 	}
-	transformer := t.baseTransformerFunc(block)
+	transformer, err := t.baseTransformerFunc(block)
+	if err != nil {
+		return nil, err
+	}
+
 	// Use base64 of encKey as the key into the cache because hashicorp/golang-lru
 	// cannot hash []uint8.
 	if t.cacheEnabled {
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/envelope_test.go b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/envelope_test.go
index 920545d2c8a..e0f6fcea809 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/envelope_test.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/envelope_test.go
@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/aes"
+	"crypto/cipher"
 	"encoding/base64"
 	"encoding/binary"
 	"fmt"
@@ -106,9 +107,12 @@ func TestEnvelopeCaching(t *testing.T) {
 	for _, tt := range testCases {
 		t.Run(tt.desc, func(t *testing.T) {
 			envelopeService := newTestEnvelopeService()
-			envelopeTransformer := NewEnvelopeTransformer(envelopeService, tt.cacheSize, aestransformer.NewCBCTransformer)
+			cbcTransformer := func(block cipher.Block) (value.Transformer, error) {
+				return aestransformer.NewCBCTransformer(block), nil
+			}
+			envelopeTransformer := NewEnvelopeTransformer(envelopeService, tt.cacheSize, cbcTransformer)
 			ctx := context.Background()
-			dataCtx := value.DefaultContext([]byte(testContextText))
+			dataCtx := value.DefaultContext(testContextText)
 			originalText := []byte(testText)
 
 			transformedData, err := envelopeTransformer.TransformToStorage(ctx, originalText, dataCtx)
@@ -146,9 +150,12 @@ func TestEnvelopeCaching(t *testing.T) {
 
 // Makes Envelope transformer hit cache limit, throws error if it misbehaves.
 func TestEnvelopeCacheLimit(t *testing.T) {
-	envelopeTransformer := NewEnvelopeTransformer(newTestEnvelopeService(), testEnvelopeCacheSize, aestransformer.NewCBCTransformer)
+	cbcTransformer := func(block cipher.Block) (value.Transformer, error) {
+		return aestransformer.NewCBCTransformer(block), nil
+	}
+	envelopeTransformer := NewEnvelopeTransformer(newTestEnvelopeService(), testEnvelopeCacheSize, cbcTransformer)
 	ctx := context.Background()
-	dataCtx := value.DefaultContext([]byte(testContextText))
+	dataCtx := value.DefaultContext(testContextText)
 
 	transformedOutputs := map[int][]byte{}
 
@@ -179,7 +186,10 @@ func TestEnvelopeCacheLimit(t *testing.T) {
 }
 
 func BenchmarkEnvelopeCBCRead(b *testing.B) {
-	envelopeTransformer := NewEnvelopeTransformer(newTestEnvelopeService(), testEnvelopeCacheSize, aestransformer.NewCBCTransformer)
+	cbcTransformer := func(block cipher.Block) (value.Transformer, error) {
+		return aestransformer.NewCBCTransformer(block), nil
+	}
+	envelopeTransformer := NewEnvelopeTransformer(newTestEnvelopeService(), testEnvelopeCacheSize, cbcTransformer)
 	benchmarkRead(b, envelopeTransformer, 1024)
 }
 
@@ -204,13 +214,17 @@ func BenchmarkAESGCMRead(b *testing.B) {
 		b.Fatal(err)
 	}
 
-	aesGCMTransformer := aestransformer.NewGCMTransformer(block)
+	aesGCMTransformer, err := aestransformer.NewGCMTransformer(block)
+	if err != nil {
+		b.Fatal(err)
+	}
+
 	benchmarkRead(b, aesGCMTransformer, 1024)
 }
 
 func benchmarkRead(b *testing.B, transformer value.Transformer, valueLength int) {
 	ctx := context.Background()
-	dataCtx := value.DefaultContext([]byte(testContextText))
+	dataCtx := value.DefaultContext(testContextText)
 	v := bytes.Repeat([]byte("0123456789abcdef"), valueLength/16)
 
 	out, err := transformer.TransformToStorage(ctx, v, dataCtx)
@@ -234,9 +248,12 @@ func benchmarkRead(b *testing.B, transformer value.Transformer, valueLength int)
 // remove after 1.13
 func TestBackwardsCompatibility(t *testing.T) {
 	envelopeService := newTestEnvelopeService()
-	envelopeTransformerInst := NewEnvelopeTransformer(envelopeService, testEnvelopeCacheSize, aestransformer.NewCBCTransformer)
+	cbcTransformer := func(block cipher.Block) (value.Transformer, error) {
+		return aestransformer.NewCBCTransformer(block), nil
+	}
+	envelopeTransformerInst := NewEnvelopeTransformer(envelopeService, testEnvelopeCacheSize, cbcTransformer)
 	ctx := context.Background()
-	dataCtx := value.DefaultContext([]byte(testContextText))
+	dataCtx := value.DefaultContext(testContextText)
 	originalText := []byte(testText)
 
 	transformedData, err := oldTransformToStorage(ctx, envelopeTransformerInst.(*envelopeTransformer), originalText, dataCtx)
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/cache.go b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/cache.go
index d2485c4e4b9..ae2e344e158 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/cache.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/cache.go
@@ -18,6 +18,7 @@ limitations under the License.
 package kmsv2
 
 import (
+	"context"
 	"crypto/sha256"
 	"hash"
 	"sync"
@@ -29,6 +30,17 @@ import (
 	"k8s.io/utils/clock"
 )
 
+// prevent decryptTransformer from drifting from value.Transformer
+var _ decryptTransformer = value.Transformer(nil)
+
+// decryptTransformer is the decryption subset of value.Transformer.
+// this exists purely to statically enforce that transformers placed in the cache are not used for encryption.
+// this is relevant in the context of nonce collision since transformers that are created
+// from encrypted DEKs retrieved from etcd cannot maintain their nonce counter state.
+type decryptTransformer interface {
+	TransformFromStorage(ctx context.Context, data []byte, dataCtx value.Context) (out []byte, stale bool, err error)
+}
+
 type simpleCache struct {
 	cache *utilcache.Expiring
 	ttl   time.Duration
@@ -50,16 +62,16 @@ func newSimpleCache(clock clock.Clock, ttl time.Duration) *simpleCache {
 }
 
 // given a key, return the transformer, or nil if it does not exist in the cache
-func (c *simpleCache) get(key []byte) value.Transformer {
+func (c *simpleCache) get(key []byte) decryptTransformer {
 	record, ok := c.cache.Get(c.keyFunc(key))
 	if !ok {
 		return nil
 	}
-	return record.(value.Transformer)
+	return record.(decryptTransformer)
 }
 
 // set caches the record for the key
-func (c *simpleCache) set(key []byte, transformer value.Transformer) {
+func (c *simpleCache) set(key []byte, transformer decryptTransformer) {
 	if len(key) == 0 {
 		panic("key must not be empty")
 	}
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/cache_test.go b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/cache_test.go
index f629fc68cab..5a2e54d3624 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/cache_test.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/cache_test.go
@@ -18,6 +18,7 @@ limitations under the License.
 package kmsv2
 
 import (
+	"crypto/rand"
 	"crypto/sha256"
 	"fmt"
 	"sync"
@@ -40,7 +41,7 @@ func TestSimpleCacheSetError(t *testing.T) {
 		{
 			name:        "empty key",
 			key:         []byte{},
-			transformer: nil,
+			transformer: &envelopeTransformer{},
 		},
 		{
 			name:        "nil transformer",
@@ -99,7 +100,7 @@ func TestKeyFunc(t *testing.T) {
 func TestSimpleCache(t *testing.T) {
 	fakeClock := testingclock.NewFakeClock(time.Now())
 	cache := newSimpleCache(fakeClock, 5*time.Second)
-	envelopeTransformer := &envelopeTransformer{}
+	transformer := &envelopeTransformer{}
 
 	wg := sync.WaitGroup{}
 	for i := 0; i < 10; i++ {
@@ -107,7 +108,7 @@ func TestSimpleCache(t *testing.T) {
 		wg.Add(1)
 		go func(key string) {
 			defer wg.Done()
-			cache.set([]byte(key), envelopeTransformer)
+			cache.set([]byte(key), transformer)
 		}(k)
 	}
 	wg.Wait()
@@ -118,7 +119,7 @@ func TestSimpleCache(t *testing.T) {
 
 	for i := 0; i < 10; i++ {
 		k := fmt.Sprintf("key-%d", i)
-		if cache.get([]byte(k)) != envelopeTransformer {
+		if cache.get([]byte(k)) != transformer {
 			t.Fatalf("Expected to get the transformer for key %v", k)
 		}
 	}
@@ -132,3 +133,11 @@ func TestSimpleCache(t *testing.T) {
 		}
 	}
 }
+
+func generateKey(length int) (key []byte, err error) {
+	key = make([]byte, length)
+	if _, err = rand.Read(key); err != nil {
+		return nil, err
+	}
+	return key, nil
+}
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/envelope.go b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/envelope.go
index bd2fce29f7d..f8b1d8e2d35 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/envelope.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/envelope.go
@@ -20,18 +20,18 @@ package kmsv2
 import (
 	"context"
 	"crypto/aes"
-	"crypto/cipher"
-	"crypto/rand"
 	"fmt"
 	"time"
 
 	"github.com/gogo/protobuf/proto"
+
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/storage/value"
+	aestransformer "k8s.io/apiserver/pkg/storage/value/encrypt/aes"
 	kmstypes "k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/v2alpha1"
 	"k8s.io/apiserver/pkg/storage/value/encrypt/envelope/metrics"
 	"k8s.io/klog/v2"
@@ -54,25 +54,54 @@ const (
 	// encryptedDEKMaxSize is the maximum size of the encrypted DEK.
 	encryptedDEKMaxSize = 1 * 1024 // 1 kB
 	// cacheTTL is the default time-to-live for the cache entry.
-	cacheTTL = 1 * time.Hour
+	// this allows the cache to grow to an infinite size for up to a day.
+	// this is meant as a temporary solution until the cache is re-written to not have a TTL.
+	// there is unlikely to be any meaningful memory impact on the server
+	// because the cache will likely never have more than a few thousand entries
+	// and each entry is roughly ~200 bytes in size.  with DEK reuse
+	// and no storage migration, the number of entries in this cache
+	// would be approximated by unique key IDs used by the KMS plugin
+	// combined with the number of server restarts.  If storage migration
+	// is performed after key ID changes, and the number of restarts
+	// is limited, this cache size may be as small as the number of API
+	// servers in use (once old entries expire out from the TTL).
+	cacheTTL = 24 * time.Hour
 	// error code
 	errKeyIDOKCode      ErrCodeKeyID = "ok"
 	errKeyIDEmptyCode   ErrCodeKeyID = "empty"
 	errKeyIDTooLongCode ErrCodeKeyID = "too_long"
 )
 
-type KeyIDGetterFunc func(context.Context) (keyID string, err error)
-type ProbeHealthzCheckFunc func(context.Context) (err error)
+// NowFunc is exported so tests can override it.
+var NowFunc = time.Now
+
+type StateFunc func() (State, error)
 type ErrCodeKeyID string
 
-type envelopeTransformer struct {
-	envelopeService   kmsservice.Service
-	providerName      string
-	keyIDGetter       KeyIDGetterFunc
-	probeHealthzCheck ProbeHealthzCheckFunc
+type State struct {
+	Transformer  value.Transformer
+	EncryptedDEK []byte
+	KeyID        string
+	Annotations  map[string][]byte
+
+	UID string
+
+	ExpirationTimestamp time.Time
+}
+
+func (s *State) ValidateEncryptCapability() error {
+	if now := NowFunc(); now.After(s.ExpirationTimestamp) {
+		return fmt.Errorf("EDEK with keyID %q expired at %s (current time is %s)",
+			s.KeyID, s.ExpirationTimestamp.Format(time.RFC3339), now.Format(time.RFC3339))
+	}
+	return nil
+}
+
+type envelopeTransformer struct {
+	envelopeService kmsservice.Service
+	providerName    string
+	stateFunc       StateFunc
 
-	// baseTransformerFunc creates a new transformer for encrypting the data with the DEK.
-	baseTransformerFunc func(cipher.Block) value.Transformer
 	// cache is a thread-safe expiring lru cache which caches decrypted DEKs indexed by their encrypted form.
 	cache *simpleCache
 }
@@ -80,18 +109,16 @@ type envelopeTransformer struct {
 // NewEnvelopeTransformer returns a transformer which implements a KEK-DEK based envelope encryption scheme.
 // It uses envelopeService to encrypt and decrypt DEKs. Respective DEKs (in encrypted form) are prepended to
 // the data items they encrypt.
-func NewEnvelopeTransformer(envelopeService kmsservice.Service, providerName string, keyIDGetter KeyIDGetterFunc, probeHealthzCheck ProbeHealthzCheckFunc, baseTransformerFunc func(cipher.Block) value.Transformer) value.Transformer {
-	return newEnvelopeTransformerWithClock(envelopeService, providerName, keyIDGetter, probeHealthzCheck, baseTransformerFunc, cacheTTL, clock.RealClock{})
+func NewEnvelopeTransformer(envelopeService kmsservice.Service, providerName string, stateFunc StateFunc) value.Transformer {
+	return newEnvelopeTransformerWithClock(envelopeService, providerName, stateFunc, cacheTTL, clock.RealClock{})
 }
 
-func newEnvelopeTransformerWithClock(envelopeService kmsservice.Service, providerName string, keyIDGetter KeyIDGetterFunc, probeHealthzCheck ProbeHealthzCheckFunc, baseTransformerFunc func(cipher.Block) value.Transformer, cacheTTL time.Duration, clock clock.Clock) value.Transformer {
+func newEnvelopeTransformerWithClock(envelopeService kmsservice.Service, providerName string, stateFunc StateFunc, cacheTTL time.Duration, clock clock.Clock) value.Transformer {
 	return &envelopeTransformer{
-		envelopeService:     envelopeService,
-		providerName:        providerName,
-		keyIDGetter:         keyIDGetter,
-		probeHealthzCheck:   probeHealthzCheck,
-		cache:               newSimpleCache(clock, cacheTTL),
-		baseTransformerFunc: baseTransformerFunc,
+		envelopeService: envelopeService,
+		providerName:    providerName,
+		stateFunc:       stateFunc,
+		cache:           newSimpleCache(clock, cacheTTL),
 	}
 }
 
@@ -103,8 +130,17 @@ func (t *envelopeTransformer) TransformFromStorage(ctx context.Context, data []b
 		return nil, false, err
 	}
 
-	// Look up the decrypted DEK from cache or Envelope.
+	// TODO: consider marking state.EncryptedDEK != encryptedObject.EncryptedDEK as a stale read to support DEK defragmentation
+	//  at a minimum we should have a metric that helps the user understand if DEK fragmentation is high
+	state, err := t.stateFunc() // no need to call state.ValidateEncryptCapability on reads
+	if err != nil {
+		return nil, false, err
+	}
+
+	// Look up the decrypted DEK from cache first
 	transformer := t.cache.get(encryptedObject.EncryptedDEK)
+
+	// fallback to the envelope service if we do not have the transformer locally
 	if transformer == nil {
 		value.RecordCacheMiss()
 
@@ -123,90 +159,74 @@ func (t *envelopeTransformer) TransformFromStorage(ctx context.Context, data []b
 			return nil, false, fmt.Errorf("failed to decrypt DEK, error: %w", err)
 		}
 
-		transformer, err = t.addTransformer(encryptedObject.EncryptedDEK, key)
+		transformer, err = t.addTransformerForDecryption(encryptedObject.EncryptedDEK, key)
 		if err != nil {
 			return nil, false, err
 		}
 	}
-	// It's possible to record empty keyID
 	metrics.RecordKeyID(metrics.FromStorageLabel, t.providerName, encryptedObject.KeyID)
 
 	out, stale, err := transformer.TransformFromStorage(ctx, encryptedObject.EncryptedData, dataCtx)
 	if err != nil {
 		return nil, false, err
 	}
-	if stale {
-		return out, stale, nil
-	}
 
-	// Check keyID freshness in addition to data staleness
-	keyID, err := t.keyIDGetter(ctx)
-	if err != nil {
-		return nil, false, err
-	}
-
-	return out, encryptedObject.KeyID != keyID, nil
+	// data is considered stale if the key ID does not match our current write transformer
+	return out, stale || encryptedObject.KeyID != state.KeyID, nil
 
 }
 
 // TransformToStorage encrypts data to be written to disk using envelope encryption.
 func (t *envelopeTransformer) TransformToStorage(ctx context.Context, data []byte, dataCtx value.Context) ([]byte, error) {
-	newKey, err := generateKey(32)
+	state, err := t.stateFunc()
 	if err != nil {
 		return nil, err
 	}
+	if err := state.ValidateEncryptCapability(); err != nil {
+		return nil, err
+	}
+
+	// this prevents a cache miss every time the DEK rotates
+	// this has the side benefit of causing the cache to perform a GC
+	// TODO see if we can do this inside the stateFunc control loop
+	// TODO(aramase): Add metrics for cache fill percentage with custom cache implementation.
+	t.cache.set(state.EncryptedDEK, state.Transformer)
 
 	requestInfo := getRequestInfoFromContext(ctx)
-	uid := string(uuid.NewUUID())
-	klog.V(6).InfoS("encrypting content using envelope service", "uid", uid, "key", string(dataCtx.AuthenticatedData()),
+	klog.V(6).InfoS("encrypting content using DEK", "uid", state.UID, "key", string(dataCtx.AuthenticatedData()),
 		"group", requestInfo.APIGroup, "version", requestInfo.APIVersion, "resource", requestInfo.Resource, "subresource", requestInfo.Subresource,
 		"verb", requestInfo.Verb, "namespace", requestInfo.Namespace, "name", requestInfo.Name)
-	resp, err := t.envelopeService.Encrypt(ctx, uid, newKey)
-	if err != nil {
-		return nil, fmt.Errorf("failed to encrypt DEK, error: %w", err)
-	}
 
-	transformer, err := t.addTransformer(resp.Ciphertext, newKey)
+	result, err := state.Transformer.TransformToStorage(ctx, data, dataCtx)
 	if err != nil {
 		return nil, err
 	}
 
-	result, err := transformer.TransformToStorage(ctx, data, dataCtx)
-	if err != nil {
-		return nil, err
-	}
-
-	metrics.RecordKeyID(metrics.ToStorageLabel, t.providerName, resp.KeyID)
+	metrics.RecordKeyID(metrics.ToStorageLabel, t.providerName, state.KeyID)
 
 	encObject := &kmstypes.EncryptedObject{
-		KeyID:         resp.KeyID,
-		EncryptedDEK:  resp.Ciphertext,
+		KeyID:         state.KeyID,
+		EncryptedDEK:  state.EncryptedDEK,
 		EncryptedData: result,
-		Annotations:   resp.Annotations,
-	}
-
-	// Check keyID freshness and write to log if key IDs are different
-	statusKeyID, err := t.keyIDGetter(ctx)
-	if err == nil && encObject.KeyID != statusKeyID {
-		klog.V(2).InfoS("observed different key IDs when encrypting content using kms v2 envelope service", "uid", uid, "objectKeyID", encObject.KeyID, "statusKeyID", statusKeyID, "providerName", t.providerName)
-
-		// trigger health probe check immediately to ensure keyID freshness
-		if err := t.probeHealthzCheck(ctx); err != nil {
-			klog.V(2).ErrorS(err, "kms plugin failed health check probe", "name", t.providerName)
-		}
+		Annotations:   state.Annotations,
 	}
 
 	// Serialize the EncryptedObject to a byte array.
 	return t.doEncode(encObject)
 }
 
-// addTransformer inserts a new transformer to the Envelope cache of DEKs for future reads.
-func (t *envelopeTransformer) addTransformer(encKey []byte, key []byte) (value.Transformer, error) {
+// addTransformerForDecryption inserts a new transformer to the Envelope cache of DEKs for future reads.
+func (t *envelopeTransformer) addTransformerForDecryption(encKey []byte, key []byte) (decryptTransformer, error) {
 	block, err := aes.NewCipher(key)
 	if err != nil {
 		return nil, err
 	}
-	transformer := t.baseTransformerFunc(block)
+	// this is compatible with NewGCMTransformerWithUniqueKeyUnsafe for decryption
+	// it would use random nonces for encryption but we never do that
+	transformer, err := aestransformer.NewGCMTransformer(block)
+	if err != nil {
+		return nil, err
+	}
 	// TODO(aramase): Add metrics for cache fill percentage with custom cache implementation.
 	t.cache.set(encKey, transformer)
 	return transformer, nil
@@ -234,17 +254,20 @@ func (t *envelopeTransformer) doDecode(originalData []byte) (*kmstypes.Encrypted
 	return o, nil
 }
 
-// generateKey generates a random key using system randomness.
-func generateKey(length int) (key []byte, err error) {
-	defer func(start time.Time) {
-		value.RecordDataKeyGeneration(start, err)
-	}(time.Now())
-	key = make([]byte, length)
-	if _, err = rand.Read(key); err != nil {
-		return nil, err
+func GenerateTransformer(ctx context.Context, uid string, envelopeService kmsservice.Service) (value.Transformer, *kmsservice.EncryptResponse, error) {
+	transformer, newKey, err := aestransformer.NewGCMTransformerWithUniqueKeyUnsafe()
+	if err != nil {
+		return nil, nil, err
 	}
 
-	return key, nil
+	klog.V(6).InfoS("encrypting content using envelope service", "uid", uid)
+
+	resp, err := envelopeService.Encrypt(ctx, uid, newKey)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to encrypt DEK, error: %w", err)
+	}
+
+	return transformer, resp, nil
 }
 
 func validateEncryptedObject(o *kmstypes.EncryptedObject) error {
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/envelope_test.go b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/envelope_test.go
index a0540d60f5a..c56279b0833 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/envelope_test.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/envelope_test.go
@@ -30,15 +30,18 @@ import (
 	"testing"
 	"time"
 
+	"github.com/gogo/protobuf/proto"
+
+	"k8s.io/apimachinery/pkg/util/uuid"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/storage/value"
-	aestransformer "k8s.io/apiserver/pkg/storage/value/encrypt/aes"
 	kmstypes "k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/v2alpha1"
 	"k8s.io/apiserver/pkg/storage/value/encrypt/envelope/metrics"
 	"k8s.io/component-base/metrics/legacyregistry"
 	"k8s.io/component-base/metrics/testutil"
 	"k8s.io/klog/v2"
 	kmsservice "k8s.io/kms/pkg/service"
+	"k8s.io/utils/clock"
 	testingclock "k8s.io/utils/clock/testing"
 )
 
@@ -47,11 +50,6 @@ const (
 	testContextText = "0123456789"
 	testKeyHash     = "sha256:6b86b273ff34fce19d6b804eff5a3f5747ada4eaa22f1d49c01e52ddb7875b4b"
 	testKeyVersion  = "1"
-	testCacheTTL    = 10 * time.Second
-)
-
-var (
-	errCode = "empty"
 )
 
 // testEnvelopeService is a mock Envelope service which can be used to simulate remote Envelope services
@@ -142,26 +140,28 @@ func TestEnvelopeCaching(t *testing.T) {
 
 	for _, tt := range testCases {
 		t.Run(tt.desc, func(t *testing.T) {
+			ctx := testContext(t)
+
 			envelopeService := newTestEnvelopeService()
 			fakeClock := testingclock.NewFakeClock(time.Now())
-			envelopeTransformer := newEnvelopeTransformerWithClock(envelopeService, testProviderName,
-				func(ctx context.Context) (string, error) {
-					return "", nil
-				},
-				func(ctx context.Context) error {
-					return nil
-				},
-				aestransformer.NewGCMTransformer, tt.cacheTTL, fakeClock)
 
-			ctx := testContext(t)
-			dataCtx := value.DefaultContext([]byte(testContextText))
+			state, err := testStateFunc(ctx, envelopeService, fakeClock)()
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			transformer := newEnvelopeTransformerWithClock(envelopeService, testProviderName,
+				func() (State, error) { return state, nil },
+				tt.cacheTTL, fakeClock)
+
+			dataCtx := value.DefaultContext(testContextText)
 			originalText := []byte(testText)
 
-			transformedData, err := envelopeTransformer.TransformToStorage(ctx, originalText, dataCtx)
+			transformedData, err := transformer.TransformToStorage(ctx, originalText, dataCtx)
 			if err != nil {
 				t.Fatalf("envelopeTransformer: error while transforming data to storage: %s", err)
 			}
-			untransformedData, _, err := envelopeTransformer.TransformFromStorage(ctx, transformedData, dataCtx)
+			untransformedData, _, err := transformer.TransformFromStorage(ctx, transformedData, dataCtx)
 			if err != nil {
 				t.Fatalf("could not decrypt Envelope transformer's encrypted data even once: %v", err)
 			}
@@ -169,10 +169,15 @@ func TestEnvelopeCaching(t *testing.T) {
 				t.Fatalf("envelopeTransformer transformed data incorrectly. Expected: %v, got %v", originalText, untransformedData)
 			}
 
-			envelopeService.SetDisabledStatus(tt.simulateKMSPluginFailure)
 			fakeClock.Step(2 * time.Minute)
+			state, err = testStateFunc(ctx, envelopeService, fakeClock)()
+			if err != nil {
+				t.Fatal(err)
+			}
+			envelopeService.SetDisabledStatus(tt.simulateKMSPluginFailure)
+
 			// Subsequent read for the same data should work fine due to caching.
-			untransformedData, _, err = envelopeTransformer.TransformFromStorage(ctx, transformedData, dataCtx)
+			untransformedData, _, err = transformer.TransformFromStorage(ctx, transformedData, dataCtx)
 			if tt.expectedError != "" {
 				if err == nil {
 					t.Fatalf("expected error: %v, got nil", tt.expectedError)
@@ -192,8 +197,25 @@ func TestEnvelopeCaching(t *testing.T) {
 	}
 }
 
-// Test keyIDGetter as part of envelopeTransformer, throws error if returned err or staleness is incorrect.
-func TestEnvelopeTransformerKeyIDGetter(t *testing.T) {
+func testStateFunc(ctx context.Context, envelopeService kmsservice.Service, clock clock.Clock) func() (State, error) {
+	return func() (State, error) {
+		transformer, resp, errGen := GenerateTransformer(ctx, string(uuid.NewUUID()), envelopeService)
+		if errGen != nil {
+			return State{}, errGen
+		}
+		return State{
+			Transformer:         transformer,
+			EncryptedDEK:        resp.Ciphertext,
+			KeyID:               resp.KeyID,
+			Annotations:         resp.Annotations,
+			UID:                 "panda",
+			ExpirationTimestamp: clock.Now().Add(time.Hour),
+		}, nil
+	}
+}
+
+// TestEnvelopeTransformerStaleness validates that staleness checks on read honor the data returned from the StateFunc.
+func TestEnvelopeTransformerStaleness(t *testing.T) {
 	t.Parallel()
 	testCases := []struct {
 		desc          string
@@ -202,19 +224,19 @@ func TestEnvelopeTransformerKeyIDGetter(t *testing.T) {
 		testKeyID     string
 	}{
 		{
-			desc:          "keyIDGetter returns err",
+			desc:          "stateFunc returns err",
 			expectedStale: false,
 			testErr:       fmt.Errorf("failed to perform status section of the healthz check for KMS Provider"),
 			testKeyID:     "",
 		},
 		{
-			desc:          "keyIDGetter returns same keyID",
+			desc:          "stateFunc returns same keyID",
 			expectedStale: false,
 			testErr:       nil,
 			testKeyID:     testKeyVersion,
 		},
 		{
-			desc:          "keyIDGetter returns different keyID",
+			desc:          "stateFunc returns different keyID",
 			expectedStale: true,
 			testErr:       nil,
 			testKeyID:     "2",
@@ -225,26 +247,33 @@ func TestEnvelopeTransformerKeyIDGetter(t *testing.T) {
 		tt := tt
 		t.Run(tt.desc, func(t *testing.T) {
 			t.Parallel()
-			envelopeService := newTestEnvelopeService()
-			envelopeTransformer := NewEnvelopeTransformer(envelopeService, testProviderName,
-				func(ctx context.Context) (string, error) {
-					return tt.testKeyID, tt.testErr
-				},
-				func(ctx context.Context) error {
-					return nil
-				},
-				aestransformer.NewGCMTransformer)
 
 			ctx := testContext(t)
-			dataCtx := value.DefaultContext([]byte(testContextText))
+
+			envelopeService := newTestEnvelopeService()
+			state, err := testStateFunc(ctx, envelopeService, clock.RealClock{})()
+			if err != nil {
+				t.Fatal(err)
+			}
+			var stateErr error
+
+			transformer := NewEnvelopeTransformer(envelopeService, testProviderName,
+				func() (State, error) { return state, stateErr },
+			)
+
+			dataCtx := value.DefaultContext(testContextText)
 			originalText := []byte(testText)
 
-			transformedData, err := envelopeTransformer.TransformToStorage(ctx, originalText, dataCtx)
+			transformedData, err := transformer.TransformToStorage(ctx, originalText, dataCtx)
 			if err != nil {
 				t.Fatalf("envelopeTransformer: error while transforming data (%v) to storage: %s", originalText, err)
 			}
 
-			_, stale, err := envelopeTransformer.TransformFromStorage(ctx, transformedData, dataCtx)
+			// inject test data before performing a read
+			state.KeyID = tt.testKeyID
+			stateErr = tt.testErr
+
+			_, stale, err := transformer.TransformFromStorage(ctx, transformedData, dataCtx)
 			if tt.testErr != nil {
 				if err == nil {
 					t.Fatalf("envelopeTransformer: expected error: %v, got nil", tt.testErr)
@@ -264,6 +293,112 @@ func TestEnvelopeTransformerKeyIDGetter(t *testing.T) {
 	}
 }
 
+func TestEnvelopeTransformerStateFunc(t *testing.T) {
+	t.Parallel()
+
+	ctx := testContext(t)
+
+	envelopeService := newTestEnvelopeService()
+	state, err := testStateFunc(ctx, envelopeService, clock.RealClock{})()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// start with a broken state
+	stateErr := fmt.Errorf("some state error")
+
+	transformer := NewEnvelopeTransformer(envelopeService, testProviderName,
+		func() (State, error) { return state, stateErr },
+	)
+
+	dataCtx := value.DefaultContext(testContextText)
+	originalText := []byte(testText)
+
+	t.Run("nothing works when the state is broken", func(t *testing.T) {
+		_, err := transformer.TransformToStorage(ctx, originalText, dataCtx)
+		if err != stateErr {
+			t.Fatalf("expected state error, got: %v", err)
+		}
+		data, err := proto.Marshal(&kmstypes.EncryptedObject{
+			EncryptedData: []byte{1},
+			KeyID:         "2",
+			EncryptedDEK:  []byte{3},
+			Annotations:   nil,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, _, err = transformer.TransformFromStorage(ctx, data, dataCtx)
+		if err != stateErr {
+			t.Fatalf("expected state error, got: %v", err)
+		}
+	})
+
+	// fix the state
+	stateErr = nil
+
+	var encryptedData []byte
+
+	t.Run("everything works when the state is fixed", func(t *testing.T) {
+		encryptedData, err = transformer.TransformToStorage(ctx, originalText, dataCtx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, _, err = transformer.TransformFromStorage(ctx, encryptedData, dataCtx)
+		if err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	// break the plugin
+	envelopeService.SetDisabledStatus(true)
+
+	t.Run("everything works even when the plugin is down but the state is valid", func(t *testing.T) {
+		data, err := transformer.TransformToStorage(ctx, originalText, dataCtx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, _, err = transformer.TransformFromStorage(ctx, data, dataCtx)
+		if err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	// make the state invalid
+	state.ExpirationTimestamp = time.Now().Add(-time.Hour)
+
+	t.Run("writes fail when the plugin is down and the state is invalid", func(t *testing.T) {
+		_, err := transformer.TransformToStorage(ctx, originalText, dataCtx)
+		if !strings.Contains(errString(err), `EDEK with keyID "1" expired at`) {
+			t.Fatalf("expected expiration error, got: %v", err)
+		}
+	})
+
+	t.Run("reads succeed when the plugin is down and the state is invalid", func(t *testing.T) {
+		_, _, err = transformer.TransformFromStorage(ctx, encryptedData, dataCtx)
+		if err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	t.Run("reads for a different DEK fail when the plugin is down and the state is invalid", func(t *testing.T) {
+		obj := &kmstypes.EncryptedObject{}
+		if err := proto.Unmarshal(encryptedData, obj); err != nil {
+			t.Fatal(err)
+		}
+		obj.EncryptedDEK = append(obj.EncryptedDEK, 1) // skip StateFunc transformer
+		data, err := proto.Marshal(obj)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		_, _, err = transformer.TransformFromStorage(ctx, data, dataCtx)
+		if errString(err) != "failed to decrypt DEK, error: Envelope service was disabled" {
+			t.Fatal(err)
+		}
+	})
+}
+
 func TestTransformToStorageError(t *testing.T) {
 	t.Parallel()
 	testCases := []struct {
@@ -295,20 +430,17 @@ func TestTransformToStorageError(t *testing.T) {
 		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
+
+			ctx := testContext(t)
+
 			envelopeService := newTestEnvelopeService()
 			envelopeService.SetAnnotations(tt.annotations)
-			envelopeTransformer := NewEnvelopeTransformer(envelopeService, testProviderName,
-				func(ctx context.Context) (string, error) {
-					return "", nil
-				},
-				func(ctx context.Context) error {
-					return nil
-				},
-				aestransformer.NewGCMTransformer)
-			ctx := testContext(t)
-			dataCtx := value.DefaultContext([]byte(testContextText))
+			transformer := NewEnvelopeTransformer(envelopeService, testProviderName,
+				testStateFunc(ctx, envelopeService, clock.RealClock{}),
+			)
+			dataCtx := value.DefaultContext(testContextText)
 
-			_, err := envelopeTransformer.TransformToStorage(ctx, []byte(testText), dataCtx)
+			_, err := transformer.TransformToStorage(ctx, []byte(testText), dataCtx)
 			if err == nil {
 				t.Fatalf("expected error, got nil")
 			}
@@ -320,7 +452,7 @@ func TestTransformToStorageError(t *testing.T) {
 }
 
 func TestEncodeDecode(t *testing.T) {
-	envelopeTransformer := &envelopeTransformer{}
+	transformer := &envelopeTransformer{}
 
 	obj := &kmstypes.EncryptedObject{
 		EncryptedData: []byte{0x01, 0x02, 0x03},
@@ -328,11 +460,11 @@ func TestEncodeDecode(t *testing.T) {
 		EncryptedDEK:  []byte{0x04, 0x05, 0x06},
 	}
 
-	data, err := envelopeTransformer.doEncode(obj)
+	data, err := transformer.doEncode(obj)
 	if err != nil {
 		t.Fatalf("envelopeTransformer: error while encoding data: %s", err)
 	}
-	got, err := envelopeTransformer.doDecode(data)
+	got, err := transformer.doDecode(data)
 	if err != nil {
 		t.Fatalf("envelopeTransformer: error while decoding data: %s", err)
 	}
@@ -590,20 +722,13 @@ func TestValidateEncryptedDEK(t *testing.T) {
 
 func TestEnvelopeMetrics(t *testing.T) {
 	envelopeService := newTestEnvelopeService()
-	envelopeTransformer := NewEnvelopeTransformer(envelopeService, testProviderName,
-		func(ctx context.Context) (string, error) {
-			return testKeyVersion, nil
-		},
-		// health probe check to ensure keyID freshness
-		func(ctx context.Context) error {
-			metrics.RecordInvalidKeyIDFromStatus(testProviderName, errCode)
-			return nil
-		},
-		aestransformer.NewGCMTransformer)
+	transformer := NewEnvelopeTransformer(envelopeService, testProviderName,
+		testStateFunc(testContext(t), envelopeService, clock.RealClock{}),
+	)
 
-	dataCtx := value.DefaultContext([]byte(testContextText))
+	dataCtx := value.DefaultContext(testContextText)
 
-	kmsv2Transformer := value.PrefixTransformer{Prefix: []byte("k8s:enc:kms:v2:"), Transformer: envelopeTransformer}
+	kmsv2Transformer := value.PrefixTransformer{Prefix: []byte("k8s:enc:kms:v2:"), Transformer: transformer}
 
 	testCases := []struct {
 		desc                  string
@@ -623,26 +748,9 @@ func TestEnvelopeMetrics(t *testing.T) {
 				# HELP apiserver_envelope_encryption_key_id_hash_total [ALPHA] Number of times a keyID is used split by transformation type and provider.
 				# TYPE apiserver_envelope_encryption_key_id_hash_total counter
 				apiserver_envelope_encryption_key_id_hash_total{key_id_hash="%s",provider_name="%s",transformation_type="%s"} 1
-        		apiserver_envelope_encryption_key_id_hash_total{key_id_hash="%s",provider_name="%s",transformation_type="%s"} 1
+				apiserver_envelope_encryption_key_id_hash_total{key_id_hash="%s",provider_name="%s",transformation_type="%s"} 1
 				`, testKeyHash, testProviderName, metrics.FromStorageLabel, testKeyHash, testProviderName, metrics.ToStorageLabel),
 		},
-		{
-			// keyVersionFromEncrypt is returned from kms v2 envelope service
-			// when it is different from the key ID returned from last status call
-			// it will trigger health probe check immediately to ensure keyID freshness
-			// during probe check above, it will call RecordInvalidKeyIDFromStatus
-			desc:                  "invalid KeyID From Status Total",
-			keyVersionFromEncrypt: "2",
-			prefix:                value.NewPrefixTransformers(nil, kmsv2Transformer),
-			metrics: []string{
-				"apiserver_envelope_encryption_invalid_key_id_from_status_total",
-			},
-			want: fmt.Sprintf(`
-			# HELP apiserver_envelope_encryption_invalid_key_id_from_status_total [ALPHA] Number of times an invalid keyID is returned by the Status RPC call split by error.
-			# TYPE apiserver_envelope_encryption_invalid_key_id_from_status_total counter
-			apiserver_envelope_encryption_invalid_key_id_from_status_total{error="%s",provider_name="%s"} 1
-			`, errCode, testProviderName),
-		},
 	}
 
 	metrics.KeyIDHashTotal.Reset()
@@ -658,7 +766,9 @@ func TestEnvelopeMetrics(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			tt.prefix.TransformFromStorage(ctx, transformedData, dataCtx)
+			if _, _, err := tt.prefix.TransformFromStorage(ctx, transformedData, dataCtx); err != nil {
+				t.Fatal(err)
+			}
 
 			if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(tt.want), tt.metrics...); err != nil {
 				t.Fatal(err)
@@ -681,7 +791,8 @@ func TestEnvelopeLogging(t *testing.T) {
 			desc: "no request info in context",
 			ctx:  testContext(t),
 			wantLogs: []string{
-				`"encrypting content using envelope service" uid="UID" key="0123456789" group="" version="" resource="" subresource="" verb="" namespace="" name=""`,
+				`"encrypting content using envelope service" uid="UID"`,
+				`"encrypting content using DEK" uid="UID" key="0123456789" group="" version="" resource="" subresource="" verb="" namespace="" name=""`,
 				`"decrypting content using envelope service" uid="UID" key="0123456789" group="" version="" resource="" subresource="" verb="" namespace="" name=""`,
 			},
 		},
@@ -697,7 +808,8 @@ func TestEnvelopeLogging(t *testing.T) {
 				Verb:        "update",
 			}),
 			wantLogs: []string{
-				`"encrypting content using envelope service" uid="UID" key="0123456789" group="awesome.bears.com" version="v1" resource="pandas" subresource="status" verb="update" namespace="kube-system" name="panda"`,
+				`"encrypting content using envelope service" uid="UID"`,
+				`"encrypting content using DEK" uid="UID" key="0123456789" group="awesome.bears.com" version="v1" resource="pandas" subresource="status" verb="update" namespace="kube-system" name="panda"`,
 				`"decrypting content using envelope service" uid="UID" key="0123456789" group="awesome.bears.com" version="v1" resource="pandas" subresource="status" verb="update" namespace="kube-system" name="panda"`,
 			},
 		},
@@ -713,19 +825,14 @@ func TestEnvelopeLogging(t *testing.T) {
 
 			envelopeService := newTestEnvelopeService()
 			fakeClock := testingclock.NewFakeClock(time.Now())
-			envelopeTransformer := newEnvelopeTransformerWithClock(envelopeService, testProviderName,
-				func(ctx context.Context) (string, error) {
-					return "1", nil
-				},
-				func(ctx context.Context) error {
-					return nil
-				},
-				aestransformer.NewGCMTransformer, 1*time.Second, fakeClock)
+			transformer := newEnvelopeTransformerWithClock(envelopeService, testProviderName,
+				testStateFunc(tc.ctx, envelopeService, clock.RealClock{}),
+				1*time.Second, fakeClock)
 
 			dataCtx := value.DefaultContext([]byte(testContextText))
 			originalText := []byte(testText)
 
-			transformedData, err := envelopeTransformer.TransformToStorage(tc.ctx, originalText, dataCtx)
+			transformedData, err := transformer.TransformToStorage(tc.ctx, originalText, dataCtx)
 			if err != nil {
 				t.Fatalf("envelopeTransformer: error while transforming data to storage: %v", err)
 			}
@@ -733,7 +840,7 @@ func TestEnvelopeLogging(t *testing.T) {
 			// advance the clock to trigger cache to expire, so we make a decrypt call that will log
 			fakeClock.Step(2 * time.Second)
 
-			_, _, err = envelopeTransformer.TransformFromStorage(tc.ctx, transformedData, dataCtx)
+			_, _, err = transformer.TransformFromStorage(tc.ctx, transformedData, dataCtx)
 			if err != nil {
 				t.Fatalf("could not decrypt Envelope transformer's encrypted data even once: %v", err)
 			}
@@ -753,3 +860,11 @@ func TestEnvelopeLogging(t *testing.T) {
 		})
 	}
 }
+
+func errString(err error) string {
+	if err == nil {
+		return ""
+	}
+
+	return err.Error()
+}
diff --git a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/grpc_service.go b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/grpc_service.go
index 5f905b3deed..ee2cc9c2004 100644
--- a/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/grpc_service.go
+++ b/staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/grpc_service.go
@@ -144,9 +144,9 @@ func (g *gRPCService) Status(ctx context.Context) (*kmsservice.StatusResponse, e
 
 func recordMetricsInterceptor(providerName string) grpc.UnaryClientInterceptor {
 	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
-		start := time.Now()
+		start := NowFunc()
 		respErr := invoker(ctx, method, req, reply, cc, opts...)
-		elapsed := time.Since(start)
+		elapsed := NowFunc().Sub(start)
 		metrics.RecordKMSOperationLatency(providerName, method, elapsed, respErr)
 		return respErr
 	}
diff --git a/test/integration/controlplane/transformation/kms_transformation_test.go b/test/integration/controlplane/transformation/kms_transformation_test.go
index 78d5b26d047..7fab07c4caf 100644
--- a/test/integration/controlplane/transformation/kms_transformation_test.go
+++ b/test/integration/controlplane/transformation/kms_transformation_test.go
@@ -95,7 +95,10 @@ func (r envelope) plainTextPayload(secretETCDPath string) ([]byte, error) {
 	// etcd path of the key is used as the authenticated context - need to pass it to decrypt
 	ctx := context.Background()
 	dataCtx := value.DefaultContext([]byte(secretETCDPath))
-	aesgcmTransformer := aestransformer.NewGCMTransformer(block)
+	aesgcmTransformer, err := aestransformer.NewGCMTransformer(block)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create transformer from block: %v", err)
+	}
 	plainSecret, _, err := aesgcmTransformer.TransformFromStorage(ctx, r.cipherTextPayload(), dataCtx)
 	if err != nil {
 		return nil, fmt.Errorf("failed to transform from storage via AESGCM, err: %w", err)
diff --git a/test/integration/controlplane/transformation/kmsv2_transformation_test.go b/test/integration/controlplane/transformation/kmsv2_transformation_test.go
index 9a2c3e44d2f..11e37dfcc83 100644
--- a/test/integration/controlplane/transformation/kmsv2_transformation_test.go
+++ b/test/integration/controlplane/transformation/kmsv2_transformation_test.go
@@ -23,31 +23,38 @@ import (
 	"bytes"
 	"context"
 	"crypto/aes"
+	"encoding/binary"
 	"fmt"
 	"strings"
 	"testing"
 	"time"
 
 	"github.com/gogo/protobuf/proto"
+	clientv3 "go.etcd.io/etcd/client/v3"
 
+	corev1 "k8s.io/api/core/v1"
 	apiextensionsclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/server/options/encryptionconfig"
+	"k8s.io/apiserver/pkg/storage/storagebackend"
 	"k8s.io/apiserver/pkg/storage/value"
 	aestransformer "k8s.io/apiserver/pkg/storage/value/encrypt/aes"
+	"k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2"
 	kmstypes "k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2/v2alpha1"
 	kmsv2mock "k8s.io/apiserver/pkg/storage/value/encrypt/envelope/testing/v2alpha1"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	kmsv2api "k8s.io/kms/apis/v2alpha1"
 	kmsv2svc "k8s.io/kms/pkg/service"
+	"k8s.io/kubernetes/test/integration"
 	"k8s.io/kubernetes/test/integration/etcd"
 )
 
@@ -91,8 +98,11 @@ func (r envelopekmsv2) plainTextPayload(secretETCDPath string) ([]byte, error) {
 		return nil, fmt.Errorf("failed to initialize AES Cipher: %v", err)
 	}
 	ctx := context.Background()
-	dataCtx := value.DefaultContext([]byte(secretETCDPath))
-	aesgcmTransformer := aestransformer.NewGCMTransformer(block)
+	dataCtx := value.DefaultContext(secretETCDPath)
+	aesgcmTransformer, err := aestransformer.NewGCMTransformer(block)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create transformer from block: %v", err)
+	}
 	data, err := r.cipherTextPayload()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get cipher text payload: %v", err)
@@ -166,7 +176,7 @@ resources:
 		plainTextDEK: plainTextDEK,
 	}
 
-	wantPrefix := string(envelopeData.prefix())
+	wantPrefix := envelopeData.prefix()
 	if !bytes.HasPrefix(rawEnvelope, []byte(wantPrefix)) {
 		t.Fatalf("expected secret to be prefixed with %s, but got %s", wantPrefix, rawEnvelope)
 	}
@@ -177,7 +187,7 @@ resources:
 	if err != nil {
 		t.Fatalf("failed to get ciphertext DEK from KMSv2 Plugin: %v", err)
 	}
-	decryptResponse, err := pluginMock.Decrypt(ctx, &kmsv2api.DecryptRequest{Uid: string(types.UID(uuid.NewUUID())), Ciphertext: ciphertext})
+	decryptResponse, err := pluginMock.Decrypt(ctx, &kmsv2api.DecryptRequest{Uid: string(uuid.NewUUID()), Ciphertext: ciphertext})
 	if err != nil {
 		t.Fatalf("failed to decrypt DEK, %v", err)
 	}
@@ -213,8 +223,10 @@ resources:
 // 1. When the key ID is unchanged, the resource version must not change
 // 2. When the key ID changes, the resource version changes (but only once)
 // 3. For all subsequent updates, the resource version must not change
-// 4. When kms-plugin is down, expect creation of new pod and encryption to fail
-// 5. when kms-plugin is down, no-op update for a pod should succeed and not result in RV change
+// 4. When kms-plugin is down, expect creation of new pod and encryption to succeed while the DEK is valid
+// 5. when kms-plugin is down, no-op update for a pod should succeed and not result in RV change while the DEK is valid
+// 6. When kms-plugin is down, expect creation of new pod and encryption to fail once the DEK is invalid
+// 7. when kms-plugin is down, no-op update for a pod should succeed and not result in RV change even once the DEK is valid
 func TestKMSv2ProviderKeyIDStaleness(t *testing.T) {
 	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.KMSv2, true)()
 
@@ -247,14 +259,16 @@ resources:
 	}
 	defer test.cleanUp()
 
-	testPod, err := test.createPod(testNamespace, dynamic.NewForConfigOrDie(test.kubeAPIServer.ClientConfig))
+	dynamicClient := dynamic.NewForConfigOrDie(test.kubeAPIServer.ClientConfig)
+
+	testPod, err := test.createPod(testNamespace, dynamicClient)
 	if err != nil {
 		t.Fatalf("Failed to create test pod, error: %v, ns: %s", err, testNamespace)
 	}
 	version1 := testPod.GetResourceVersion()
 
 	// 1. no-op update for the test pod should not result in any RV change
-	updatedPod, err := test.inplaceUpdatePod(testNamespace, testPod, dynamic.NewForConfigOrDie(test.kubeAPIServer.ClientConfig))
+	updatedPod, err := test.inplaceUpdatePod(testNamespace, testPod, dynamicClient)
 	if err != nil {
 		t.Fatalf("Failed to update test pod, error: %v, ns: %s", err, testNamespace)
 	}
@@ -262,6 +276,30 @@ resources:
 	if version1 != version2 {
 		t.Fatalf("Resource version should not have changed. old pod: %v, new pod: %v", testPod, updatedPod)
 	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	t.Cleanup(cancel)
+
+	var firstEncryptedDEK []byte
+	assertPodDEKs(ctx, t, test.kubeAPIServer.ServerOpts.Etcd.StorageConfig,
+		1, 1,
+		"k8s:enc:kms:v2:kms-provider:",
+		func(_ int, counter uint64, etcdKey string, obj kmstypes.EncryptedObject) {
+			firstEncryptedDEK = obj.EncryptedDEK
+
+			if obj.KeyID != "1" {
+				t.Errorf("key %s: want key ID %s, got %s", etcdKey, "1", obj.KeyID)
+			}
+
+			// with the first key we perform encryption during the following steps:
+			// - create
+			const want = 1_000_000_000 + 1 // zero value of counter is one billion
+			if want != counter {
+				t.Errorf("key %s: counter nonce is invalid: want %d, got %d", etcdKey, want, counter)
+			}
+		},
+	)
+
 	// 2. no-op update for the test pod with keyID update should result in RV change
 	pluginMock.UpdateKeyID()
 	if err := kmsv2mock.WaitForBase64PluginToBeUpdated(pluginMock); err != nil {
@@ -272,7 +310,8 @@ resources:
 	version3 := ""
 	err = wait.Poll(time.Second, time.Minute,
 		func() (bool, error) {
-			updatedPod, err = test.inplaceUpdatePod(testNamespace, updatedPod, dynamic.NewForConfigOrDie(test.kubeAPIServer.ClientConfig))
+			t.Log("polling for in-place update rv change")
+			updatedPod, err = test.inplaceUpdatePod(testNamespace, updatedPod, dynamicClient)
 			if err != nil {
 				return false, err
 			}
@@ -290,8 +329,29 @@ resources:
 		t.Fatalf("Resource version should have changed after keyID update. old pod: %v, new pod: %v", testPod, updatedPod)
 	}
 
+	var wantCount uint64 = 1_000_000_000 // zero value of counter is one billion
+	wantCount++                          // in place update with RV change
+
+	// with the second key we perform encryption during the following steps:
+	// - in place update with RV change
+	// - delete (which does an update to set deletion timestamp)
+	// - create
+	checkDEK := func(_ int, counter uint64, etcdKey string, obj kmstypes.EncryptedObject) {
+		if bytes.Equal(obj.EncryptedDEK, firstEncryptedDEK) {
+			t.Errorf("key %s: incorrectly has the same EDEK", etcdKey)
+		}
+
+		if obj.KeyID != "2" {
+			t.Errorf("key %s: want key ID %s, got %s", etcdKey, "2", obj.KeyID)
+		}
+
+		if wantCount != counter {
+			t.Errorf("key %s: counter nonce is invalid: want %d, got %d", etcdKey, wantCount, counter)
+		}
+	}
+
 	// 3. no-op update for the updated pod should not result in RV change
-	updatedPod, err = test.inplaceUpdatePod(testNamespace, updatedPod, dynamic.NewForConfigOrDie(test.kubeAPIServer.ClientConfig))
+	updatedPod, err = test.inplaceUpdatePod(testNamespace, updatedPod, dynamicClient)
 	if err != nil {
 		t.Fatalf("Failed to update test pod, error: %v, ns: %s", err, testNamespace)
 	}
@@ -300,25 +360,180 @@ resources:
 		t.Fatalf("Resource version should not have changed again after the initial version updated as a result of the keyID update. old pod: %v, new pod: %v", testPod, updatedPod)
 	}
 
-	// 4. when kms-plugin is down, expect creation of new pod and encryption to fail
+	// delete the pod so that it can be recreated
+	if err := test.deletePod(testNamespace, dynamicClient); err != nil {
+		t.Fatalf("failed to delete test pod: %v", err)
+	}
+	wantCount++ // we cannot assert against the counter being 2 since the pod gets deleted
+
+	// 4. when kms-plugin is down, expect creation of new pod and encryption to succeed because the DEK is still valid
 	pluginMock.EnterFailedState()
 	mustBeUnHealthy(t, "/kms-providers",
 		"internal server error: kms-provider-0: rpc error: code = FailedPrecondition desc = failed precondition - key disabled",
 		test.kubeAPIServer.ClientConfig)
 
-	_, err = test.createPod(testNamespace, dynamic.NewForConfigOrDie(test.kubeAPIServer.ClientConfig))
-	if err == nil || !strings.Contains(err.Error(), "failed to encrypt") {
-		t.Fatalf("Create test pod should have failed due to encryption, ns: %s", testNamespace)
+	newPod, err := test.createPod(testNamespace, dynamicClient)
+	if err != nil {
+		t.Fatalf("Create test pod should have succeeded due to valid DEK, ns: %s, got: %v", testNamespace, err)
 	}
+	wantCount++
+	version5 := newPod.GetResourceVersion()
 
-	// 5. when kms-plugin is down, no-op update for a pod should succeed and not result in RV change
-	updatedPod, err = test.inplaceUpdatePod(testNamespace, updatedPod, dynamic.NewForConfigOrDie(test.kubeAPIServer.ClientConfig))
+	// 5. when kms-plugin is down and DEK is valid, no-op update for a pod should succeed and not result in RV change
+	updatedPod, err = test.inplaceUpdatePod(testNamespace, newPod, dynamicClient)
 	if err != nil {
 		t.Fatalf("Failed to perform no-op update on pod when kms-plugin is down, error: %v, ns: %s", err, testNamespace)
 	}
-	version5 := updatedPod.GetResourceVersion()
-	if version3 != version5 {
-		t.Fatalf("Resource version should not have changed again after the initial version updated as a result of the keyID update. old pod: %v, new pod: %v", testPod, updatedPod)
+	version6 := updatedPod.GetResourceVersion()
+	if version5 != version6 {
+		t.Fatalf("Resource version should not have changed again after the initial version updated as a result of the keyID update. old pod: %v, new pod: %v", newPod, updatedPod)
+	}
+
+	// Invalidate the DEK by moving the current time forward
+	origNowFunc := kmsv2.NowFunc
+	t.Cleanup(func() { kmsv2.NowFunc = origNowFunc })
+	kmsv2.NowFunc = func() time.Time { return origNowFunc().Add(5 * time.Minute) }
+
+	// 6. when kms-plugin is down, expect creation of new pod and encryption to fail because the DEK is invalid
+	_, err = test.createPod(testNamespace, dynamicClient)
+	if err == nil || !strings.Contains(err.Error(), `EDEK with keyID "2" expired at 2`) {
+		t.Fatalf("Create test pod should have failed due to encryption, ns: %s, got: %v", testNamespace, err)
+	}
+
+	// 7. when kms-plugin is down and DEK is invalid, no-op update for a pod should succeed and not result in RV change
+	updatedNewPod, err := test.inplaceUpdatePod(testNamespace, newPod, dynamicClient)
+	if err != nil {
+		t.Fatalf("Failed to perform no-op update on pod when kms-plugin is down, error: %v, ns: %s", err, testNamespace)
+	}
+	version7 := updatedNewPod.GetResourceVersion()
+	if version5 != version7 {
+		t.Fatalf("Resource version should not have changed again after the initial version updated as a result of the keyID update. old pod: %v, new pod: %v", newPod, updatedNewPod)
+	}
+
+	assertPodDEKs(ctx, t, test.kubeAPIServer.ServerOpts.Etcd.StorageConfig,
+		1, 1, "k8s:enc:kms:v2:kms-provider:", checkDEK,
+	)
+}
+
+func TestKMSv2ProviderDEKReuse(t *testing.T) {
+	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.KMSv2, true)()
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	t.Cleanup(cancel)
+
+	encryptionConfig := `
+kind: EncryptionConfiguration
+apiVersion: apiserver.config.k8s.io/v1
+resources:
+  - resources:
+    - pods
+    providers:
+    - kms:
+       apiVersion: v2
+       name: kms-provider
+       endpoint: unix:///@kms-provider.sock
+`
+	pluginMock, err := kmsv2mock.NewBase64Plugin("@kms-provider.sock")
+	if err != nil {
+		t.Fatalf("failed to create mock of KMSv2 Plugin: %v", err)
+	}
+
+	go pluginMock.Start()
+	if err := kmsv2mock.WaitForBase64PluginToBeUp(pluginMock); err != nil {
+		t.Fatalf("Failed start plugin, err: %v", err)
+	}
+	defer pluginMock.CleanUp()
+
+	test, err := newTransformTest(t, encryptionConfig, false, "")
+	if err != nil {
+		t.Fatalf("failed to start KUBE API Server with encryptionConfig\n %s, error: %v", encryptionConfig, err)
+	}
+	t.Cleanup(test.cleanUp)
+
+	client := kubernetes.NewForConfigOrDie(test.kubeAPIServer.ClientConfig)
+
+	const podCount = 1_000
+
+	for i := 0; i < podCount; i++ {
+		if _, err := client.CoreV1().Pods(testNamespace).Create(ctx, &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: fmt.Sprintf("dek-reuse-%04d", i+1), // making creation order match returned list order / nonce counter
+			},
+			Spec: corev1.PodSpec{
+				Containers: []corev1.Container{
+					{
+						Name:  "busybox",
+						Image: "busybox",
+					},
+				},
+			},
+		}, metav1.CreateOptions{}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	assertPodDEKs(ctx, t, test.kubeAPIServer.ServerOpts.Etcd.StorageConfig,
+		podCount, 1, // key ID does not change during the test so we should only have a single DEK
+		"k8s:enc:kms:v2:kms-provider:",
+		func(i int, counter uint64, etcdKey string, obj kmstypes.EncryptedObject) {
+			if obj.KeyID != "1" {
+				t.Errorf("key %s: want key ID %s, got %s", etcdKey, "1", obj.KeyID)
+			}
+
+			// zero value of counter is one billion so the first value will be one billion plus one
+			// hence we add that to our zero based index to calculate the expected nonce
+			if uint64(i+1_000_000_000+1) != counter {
+				t.Errorf("key %s: counter nonce is invalid: want %d, got %d", etcdKey, i+1, counter)
+			}
+		},
+	)
+}
+
+func assertPodDEKs(ctx context.Context, t *testing.T, config storagebackend.Config, podCount, dekCount int, kmsPrefix string,
+	f func(i int, counter uint64, etcdKey string, obj kmstypes.EncryptedObject)) {
+	t.Helper()
+
+	rawClient, etcdClient, err := integration.GetEtcdClients(config.Transport)
+	if err != nil {
+		t.Fatalf("failed to create etcd client: %v", err)
+	}
+	t.Cleanup(func() { _ = rawClient.Close() })
+
+	response, err := etcdClient.Get(ctx, "/"+config.Prefix+"/pods/"+testNamespace+"/", clientv3.WithPrefix())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(response.Kvs) != podCount {
+		t.Fatalf("expected %d KVs, but got %d", podCount, len(response.Kvs))
+	}
+
+	out := make([]kmstypes.EncryptedObject, len(response.Kvs))
+	for i, kv := range response.Kvs {
+		v := bytes.TrimPrefix(kv.Value, []byte(kmsPrefix))
+		if err := proto.Unmarshal(v, &out[i]); err != nil {
+			t.Fatal(err)
+		}
+
+		nonce := out[i].EncryptedData[:12]
+		randN := nonce[:4]
+		count := nonce[4:]
+
+		if bytes.Equal(randN, make([]byte, len(randN))) {
+			t.Errorf("key %s: got all zeros for first four bytes", string(kv.Key))
+		}
+
+		counter := binary.LittleEndian.Uint64(count)
+		f(i, counter, string(kv.Key), out[i])
+	}
+
+	uniqueDEKs := sets.NewString()
+	for _, object := range out {
+		uniqueDEKs.Insert(string(object.EncryptedDEK))
+	}
+
+	if uniqueDEKs.Len() != dekCount {
+		t.Errorf("expected %d DEKs, got: %d", dekCount, uniqueDEKs.Len())
 	}
 }
 
diff --git a/test/integration/controlplane/transformation/secrets_transformation_test.go b/test/integration/controlplane/transformation/secrets_transformation_test.go
index 820836bc19e..9cb940dfae7 100644
--- a/test/integration/controlplane/transformation/secrets_transformation_test.go
+++ b/test/integration/controlplane/transformation/secrets_transformation_test.go
@@ -140,7 +140,10 @@ func unSealWithGCMTransformer(ctx context.Context, cipherText []byte, dataCtx va
 		return nil, fmt.Errorf("failed to create block cipher: %v", err)
 	}
 
-	gcmTransformer := aestransformer.NewGCMTransformer(block)
+	gcmTransformer, err := aestransformer.NewGCMTransformer(block)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create transformer from block: %v", err)
+	}
 
 	clearText, _, err := gcmTransformer.TransformFromStorage(ctx, cipherText, dataCtx)
 	if err != nil {
diff --git a/test/integration/controlplane/transformation/transformation_test.go b/test/integration/controlplane/transformation/transformation_test.go
index 407945576cf..6162ec5374f 100644
--- a/test/integration/controlplane/transformation/transformation_test.go
+++ b/test/integration/controlplane/transformation/transformation_test.go
@@ -454,6 +454,15 @@ func (e *transformTest) createPod(namespace string, dynamicInterface dynamic.Int
 	return pod, nil
 }
 
+func (e *transformTest) deletePod(namespace string, dynamicInterface dynamic.Interface) error {
+	podGVR := gvr("", "v1", "pods")
+	stubObj, err := getStubObj(podGVR)
+	if err != nil {
+		return err
+	}
+	return dynamicInterface.Resource(podGVR).Namespace(namespace).Delete(context.TODO(), stubObj.GetName(), metav1.DeleteOptions{})
+}
+
 func (e *transformTest) inplaceUpdatePod(namespace string, obj *unstructured.Unstructured, dynamicInterface dynamic.Interface) (*unstructured.Unstructured, error) {
 	podGVR := gvr("", "v1", "pods")
 	pod, err := inplaceUpdateResource(dynamicInterface, podGVR, namespace, obj)
@@ -519,7 +528,7 @@ func (e *transformTest) printMetrics() error {
 func mustBeHealthy(t kubeapiservertesting.Logger, checkName, wantBodyContains string, clientConfig *rest.Config, excludes ...string) {
 	t.Helper()
 	var restErr error
-	pollErr := wait.PollImmediate(2*time.Second, wait.ForeverTestTimeout, func() (bool, error) {
+	pollErr := wait.PollImmediate(2*time.Second, time.Minute, func() (bool, error) {
 		body, ok, err := getHealthz(checkName, clientConfig, excludes...)
 		restErr = err
 		if err != nil {
@@ -540,7 +549,7 @@ func mustBeHealthy(t kubeapiservertesting.Logger, checkName, wantBodyContains st
 func mustBeUnHealthy(t kubeapiservertesting.Logger, checkName, wantBodyContains string, clientConfig *rest.Config, excludes ...string) {
 	t.Helper()
 	var restErr error
-	pollErr := wait.PollImmediate(2*time.Second, wait.ForeverTestTimeout, func() (bool, error) {
+	pollErr := wait.PollImmediate(2*time.Second, time.Minute, func() (bool, error) {
 		body, ok, err := getHealthz(checkName, clientConfig, excludes...)
 		restErr = err
 		if err != nil {