refactor: convert the acm certificate store to kvstore (#890)

spinifex/handlers/acm/getcertificate_test.go

@@ -125,7 +125,7 @@ func putUndecryptableCert(t *testing.T, store *Store, arn, domain string) {
 		Status:         acm.CertificateStatusIssued,
 	})
 	require.NoError(t, err)
-	_, err = store.kv.Put(t.Context(), certKey(arn), data)
+	_, err = rawKV(t, store).Put(t.Context(), certKey(arn), data)
 	require.NoError(t, err)
 }
 

spinifex/handlers/acm/service_impl_test.go

@@ -278,7 +278,7 @@ func TestImportCertificate_PrivateKeyEncryptedAtRest(t *testing.T) {
 	}, testAccountID)
 	require.NoError(t, err)
 
-	entry, err := svc.store.kv.Get(context.Background(), certKey(aws.StringValue(out.CertificateArn)))
+	entry, err := rawKV(t, svc.store).Get(context.Background(), certKey(aws.StringValue(out.CertificateArn)))
 	require.NoError(t, err)
 
 	var raw CertRecord

spinifex/handlers/acm/store.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/arn"
 	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
+	"github.com/mulgadc/spinifex/spinifex/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/kvutil"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -154,7 +155,7 @@ type CertRecord struct {
 
 // Store provides CRUD for ACM certificate records backed by JetStream KV.
 type Store struct {
-	kv jetstream.KeyValue
+	store *kvstore.Store[CertRecord]
 	// masterKey encrypts/decrypts CertRecord.PrivateKey at rest (AES-256-GCM via
 	// handlers_iam.EncryptSecret/DecryptSecret). Every Store over the ACM bucket
 	// — the ACM service and ELBv2's independent read-only Store alike — must be
@@ -184,8 +185,16 @@ func NewStore(ctx context.Context, nc *nats.Conn, masterKey []byte) (*Store, err
 		return nil, err
 	}
 
-	slog.Info("ACM store initialized", "bucket", KVBucketACM)
-	return &Store{kv: kv, masterKey: masterKey}, nil
+	slog.Info("ACM store initialised", "bucket", KVBucketACM)
+	return &Store{
+		store: kvstore.Over[CertRecord](kv, kvstore.Config{
+			Attempts: maxInUseByCASAttempts,
+			Exhausted: func(key string, attempts int) error {
+				return fmt.Errorf("acm store: exceeded %d CAS attempts updating %s", attempts, key)
+			},
+		}),
+		masterKey: masterKey,
+	}, nil
 }
 
 // certKey derives the KV key from a certificate ARN, reading the id exactly as
@@ -215,16 +224,11 @@ func (s *Store) PutCert(ctx context.Context, rec *CertRecord) error {
 	// this call.
 	clone := *rec
 	clone.PrivateKey = ciphertext
-	data, err := json.Marshal(&clone)
-	if err != nil {
-		return fmt.Errorf("marshal cert: %w", err)
-	}
 	key := certKey(rec.CertificateArn)
 	if key == "" {
 		return fmt.Errorf("acm store: not a certificate ARN: %q", rec.CertificateArn)
 	}
-	_, err = s.kv.Put(ctx, key, data)
-	return err
+	return s.store.Set(ctx, key, &clone)
 }
 
 // GetCert retrieves a certificate by ARN with PrivateKey decrypted to
@@ -260,25 +264,21 @@ func (s *Store) getCert(ctx context.Context, certArn string, decrypt bool) (*Cer
 	if key == "" {
 		return nil, nil
 	}
-	entry, err := s.kv.Get(ctx, key)
-	if err != nil {
-		if errors.Is(err, jetstream.ErrKeyNotFound) {
-			return nil, nil
-		}
-		return nil, err
+	rec, _, err := s.store.Get(ctx, key)
+	if errors.Is(err, kvstore.ErrNotFound) {
+		return nil, nil
 	}
-	var rec CertRecord
-	if err := json.Unmarshal(entry.Value(), &rec); err != nil {
-		return nil, fmt.Errorf("unmarshal cert: %w", err)
+	if err != nil {
+		return nil, err
 	}
 	if !decrypt {
 		rec.PrivateKey = ""
-		return &rec, nil
+		return rec, nil
 	}
-	if err := s.decryptPrivateKey(&rec); err != nil {
+	if err := s.decryptPrivateKey(rec); err != nil {
 		return nil, err
 	}
-	return &rec, nil
+	return rec, nil
 }
 
 // legacyPrivateKeyPEMTypes are the PEM block types ImportCertificate accepts
@@ -333,13 +333,11 @@ func (s *Store) DeleteCert(ctx context.Context, certArn string) (bool, error) {
 	if key == "" {
 		return false, nil
 	}
-	if _, err := s.kv.Get(ctx, key); err != nil {
-		if errors.Is(err, jetstream.ErrKeyNotFound) {
-			return false, nil
-		}
+	present, err := s.store.Exists(ctx, key)
+	if err != nil || !present {
 		return false, err
 	}
-	if err := s.kv.Delete(ctx, key); err != nil {
+	if err := s.store.Delete(ctx, key); err != nil {
 		return false, err
 	}
 	return true, nil
@@ -386,7 +384,11 @@ func (s *Store) ListAllCertMetadata(ctx context.Context) ([]*CertRecord, error)
 // ListAllCertMetadata — and an undecryptable key is not a reason to skip the
 // record, since nothing here reads it.
 func (s *Store) listCerts(ctx context.Context, keep func(*CertRecord) bool, decrypt bool) ([]*CertRecord, error) {
-	keys, err := s.kv.Keys(ctx)
+	kv, err := s.store.KV(ctx)
+	if err != nil {
+		return nil, err
+	}
+	keys, err := kv.Keys(ctx)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return nil, nil
@@ -398,7 +400,7 @@ func (s *Store) listCerts(ctx context.Context, keep func(*CertRecord) bool, decr
 		if !strings.HasPrefix(key, KeyPrefixCert) {
 			continue
 		}
-		entry, err := s.kv.Get(ctx, key)
+		entry, err := kv.Get(ctx, key)
 		if err != nil {
 			continue
 		}
@@ -464,11 +466,6 @@ func (s *Store) RemoveInUseBy(ctx context.Context, certArn, resourceArn string)
 	})
 }
 
-// errCertAbsent lets updateInUseByCAS tell an absent certificate apart from a
-// real failure. Both callers treat a missing certificate as a no-op, so it
-// never escapes this file.
-var errCertAbsent = errors.New("acm store: certificate not found")
-
 // updateInUseByCAS applies mutate to certArn's current InUseBy set and writes
 // the result back under CAS. mutate returns nil to mean "no change needed".
 // A certificate that does not exist is a no-op, not an error.
@@ -477,13 +474,7 @@ func (s *Store) updateInUseByCAS(ctx context.Context, certArn string, mutate fun
 	if key == "" {
 		return nil
 	}
-	_, err := kvutil.Update(ctx, s.kv, key, kvutil.CASConfig{
-		Attempts: maxInUseByCASAttempts,
-		NotFound: errCertAbsent,
-		Exhausted: func(string, int) error {
-			return fmt.Errorf("acm store: exceeded %d CAS attempts updating InUseBy for %s", maxInUseByCASAttempts, certArn)
-		},
-	}, func(rec *CertRecord) (bool, error) {
+	err := s.store.Mutate(ctx, key, func(rec *CertRecord) (bool, error) {
 		next := mutate(rec.InUseBy)
 		if next == nil {
 			return false, nil
@@ -491,7 +482,7 @@ func (s *Store) updateInUseByCAS(ctx context.Context, certArn string, mutate fun
 		rec.InUseBy = next
 		return true, nil
 	})
-	if errors.Is(err, errCertAbsent) {
+	if errors.Is(err, kvstore.ErrNotFound) {
 		return nil
 	}
 	return err
@@ -515,28 +506,20 @@ func (s *Store) AcquireLease(ctx context.Context, certArn, holderID string, ttl
 	if key == "" {
 		return false, nil
 	}
-	entry, err := s.kv.Get(ctx, key)
-	if err != nil {
-		if errors.Is(err, jetstream.ErrKeyNotFound) {
-			return false, nil
-		}
-		return false, err
+	rec, rev, err := s.store.Get(ctx, key)
+	if errors.Is(err, kvstore.ErrNotFound) {
+		return false, nil
 	}
-	var rec CertRecord
-	if err := json.Unmarshal(entry.Value(), &rec); err != nil {
-		return false, fmt.Errorf("unmarshal cert: %w", err)
+	if err != nil {
+		return false, err
 	}
 	if rec.LeaseHolder != "" && rec.LeaseHolder != holderID && rec.LeaseExpiresAt.After(now) {
 		return false, nil
 	}
 	rec.LeaseHolder = holderID
 	rec.LeaseExpiresAt = now.Add(ttl)
-	data, err := json.Marshal(&rec)
-	if err != nil {
-		return false, fmt.Errorf("marshal cert: %w", err)
-	}
-	if _, err := s.kv.Update(ctx, key, data, entry.Revision()); err != nil {
-		if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
+	if err := s.store.CompareAndSet(ctx, key, rec, rev); err != nil {
+		if errors.Is(err, kvstore.ErrConflict) {
 			// Lost the race to a concurrent acquirer between Get and Update;
 			// the caller skips this tick rather than retrying immediately.
 			return false, nil
@@ -555,28 +538,20 @@ func (s *Store) ReleaseLease(ctx context.Context, certArn, holderID string) erro
 	if key == "" {
 		return nil
 	}
-	entry, err := s.kv.Get(ctx, key)
-	if err != nil {
-		if errors.Is(err, jetstream.ErrKeyNotFound) {
-			return nil
-		}
-		return err
+	rec, rev, err := s.store.Get(ctx, key)
+	if errors.Is(err, kvstore.ErrNotFound) {
+		return nil
 	}
-	var rec CertRecord
-	if err := json.Unmarshal(entry.Value(), &rec); err != nil {
-		return fmt.Errorf("unmarshal cert: %w", err)
+	if err != nil {
+		return err
 	}
 	if rec.LeaseHolder != holderID {
 		return nil // already released, expired, or taken over by another holder
 	}
 	rec.LeaseHolder = ""
 	rec.LeaseExpiresAt = time.Time{}
-	data, err := json.Marshal(&rec)
-	if err != nil {
-		return fmt.Errorf("marshal cert: %w", err)
-	}
-	if _, err := s.kv.Update(ctx, key, data, entry.Revision()); err != nil {
-		if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
+	if err := s.store.CompareAndSet(ctx, key, rec, rev); err != nil {
+		if errors.Is(err, kvstore.ErrConflict) {
 			return nil // record changed concurrently; nothing to clean up
 		}
 		return err

spinifex/handlers/acm/store_test.go

@@ -9,6 +9,7 @@ import (
 
 	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -26,6 +27,16 @@ func setupACMStore(t *testing.T) *Store {
 	return store
 }
 
+// rawKV returns the underlying bucket so a test can assert on the bytes as
+// stored. Reading through Store would decode them, which is the one thing
+// these tests must not do.
+func rawKV(t *testing.T, s *Store) jetstream.KeyValue {
+	t.Helper()
+	kv, err := s.store.KV(t.Context())
+	require.NoError(t, err)
+	return kv
+}
+
 func TestAddInUseBy_AddsAndDedupes(t *testing.T) {
 	store := setupACMStore(t)
 	rec := &CertRecord{CertificateArn: "arn:aws:acm:ap-southeast-2:000000000001:certificate/inuse-1", AccountID: testAccountID}
@@ -142,7 +153,7 @@ func TestStore_PutCert_EncryptsPrivateKeyAtRest(t *testing.T) {
 	require.NoError(t, store.PutCert(t.Context(), rec))
 
 	// The raw bytes landed in KV must not contain the plaintext key.
-	entry, err := store.kv.Get(t.Context(), certKey(arn))
+	entry, err := rawKV(t, store).Get(t.Context(), certKey(arn))
 	require.NoError(t, err)
 	assert.NotContains(t, string(entry.Value()), "BEGIN EC PRIVATE KEY")
 
@@ -166,7 +177,7 @@ func TestStore_GetCert_LegacyPlaintextPassthroughAndReencrypt(t *testing.T) {
 	legacy := &CertRecord{CertificateArn: arn, AccountID: "000000000001", PrivateKey: plaintext}
 	data, err := json.Marshal(legacy)
 	require.NoError(t, err)
-	_, err = store.kv.Put(t.Context(), certKey(arn), data)
+	_, err = rawKV(t, store).Put(t.Context(), certKey(arn), data)
 	require.NoError(t, err)
 
 	got, err := store.GetCert(t.Context(), arn)
@@ -175,7 +186,7 @@ func TestStore_GetCert_LegacyPlaintextPassthroughAndReencrypt(t *testing.T) {
 
 	// The next write re-encrypts it — no operator action required.
 	require.NoError(t, store.PutCert(t.Context(), got))
-	entry, err := store.kv.Get(t.Context(), certKey(arn))
+	entry, err := rawKV(t, store).Get(t.Context(), certKey(arn))
 	require.NoError(t, err)
 	assert.NotContains(t, string(entry.Value()), "BEGIN RSA PRIVATE KEY", "record must be re-encrypted after the next PutCert")
 }
@@ -191,7 +202,7 @@ func TestStore_GetCert_RejectsGarbageAsLegacyPlaintext(t *testing.T) {
 	garbage := &CertRecord{CertificateArn: arn, AccountID: "000000000001", PrivateKey: "not-pem-and-not-ciphertext"}
 	data, err := json.Marshal(garbage)
 	require.NoError(t, err)
-	_, err = store.kv.Put(t.Context(), certKey(arn), data)
+	_, err = rawKV(t, store).Put(t.Context(), certKey(arn), data)
 	require.NoError(t, err)
 
 	_, err = store.GetCert(t.Context(), arn)
@@ -309,7 +320,7 @@ func TestAcquireLease_LeavesPrivateKeyEncrypted(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, acquired)
 
-	entry, err := store.kv.Get(t.Context(), certKey(arn))
+	entry, err := rawKV(t, store).Get(t.Context(), certKey(arn))
 	require.NoError(t, err)
 	assert.NotContains(t, string(entry.Value()), "BEGIN EC PRIVATE KEY",
 		"AcquireLease must not write the private key back in plaintext")
@@ -335,7 +346,7 @@ func TestReleaseLease_LeavesPrivateKeyEncrypted(t *testing.T) {
 
 	require.NoError(t, store.ReleaseLease(t.Context(), arn, "node-a"))
 
-	entry, err := store.kv.Get(t.Context(), certKey(arn))
+	entry, err := rawKV(t, store).Get(t.Context(), certKey(arn))
 	require.NoError(t, err)
 	assert.NotContains(t, string(entry.Value()), "BEGIN EC PRIVATE KEY",
 		"ReleaseLease must not write the private key back in plaintext")

spinifex/kvstore/bucket.go

@@ -15,11 +15,13 @@ import (
 
 // Config describes the bucket that a Store or Bucket sits over.
 type Config struct {
-	Name     string
-	History  int
-	Replicas int
-	TTL      time.Duration
-	Missing  string
+	Name      string
+	History   int
+	Replicas  int
+	TTL       time.Duration
+	Missing   string
+	Attempts  int
+	Exhausted func(key string, attempts int) error
 }
 
 // Bucket memoises a lazily created KV bucket. Construct it with NewBucket.
@@ -37,16 +39,26 @@ func NewBucket(js jetstream.JetStream, cfg Config) *Bucket {
 	return &Bucket{js: js, cfg: cfg}
 }
 
+// NewOpenBucket returns a Bucket over an already-open handle, for callers that
+// resolve their bucket at construction so a bad connection fails at startup
+// rather than on first use.
+func NewOpenBucket(kv jetstream.KeyValue, cfg Config) *Bucket {
+	return &Bucket{kv: kv, cfg: cfg}
+}
+
 // KV opens the bucket on first use and returns the cached handle thereafter.
 func (b *Bucket) KV(ctx context.Context) (jetstream.KeyValue, error) {
-	if b.js == nil {
-		return nil, errors.New(b.cfg.Missing)
-	}
 	b.mu.Lock()
 	defer b.mu.Unlock()
 	if b.kv != nil {
 		return b.kv, nil
 	}
+	if b.js == nil {
+		if b.cfg.Missing == "" {
+			return nil, errors.New("kvstore: no JetStream client configured")
+		}
+		return nil, errors.New(b.cfg.Missing)
+	}
 	kv, err := b.open(ctx)
 	if err != nil {
 		return nil, err

spinifex/kvstore/store.go

@@ -16,6 +16,7 @@ import (
 var (
 	ErrNotFound = errors.New("kvstore: key not found")
 	ErrExists   = errors.New("kvstore: key already exists")
+	ErrConflict = errors.New("kvstore: revision conflict")
 )
 
 // Store is a Bucket plus a JSON codec for T.
@@ -66,7 +67,9 @@ func (s *Store[T]) Mutate(ctx context.Context, key string, fn func(*T) (bool, er
 		return err
 	}
 	_, err = kvutil.Update(ctx, kv, key, kvutil.CASConfig{
-		NotFound: fmt.Errorf("%w: %s", ErrNotFound, key),
+		Attempts:  s.cfg.Attempts,
+		NotFound:  fmt.Errorf("%w: %s", ErrNotFound, key),
+		Exhausted: s.cfg.Exhausted,
 	}, fn)
 	return err
 }
@@ -126,6 +129,65 @@ func (s *Store[T]) DeletePrefix(ctx context.Context, prefix string) error {
 	return nil
 }
 
+// Over returns a Store for records of type T over an already-open bucket.
+func Over[T any](kv jetstream.KeyValue, cfg Config) *Store[T] {
+	return &Store[T]{Bucket: NewOpenBucket(kv, cfg)}
+}
+
+// Set writes a record whether or not the key exists, for callers whose write
+// is a replacement rather than a claim or a read-modify-write.
+func (s *Store[T]) Set(ctx context.Context, key string, v *T) error {
+	kv, err := s.KV(ctx)
+	if err != nil {
+		return err
+	}
+	data, err := json.Marshal(v)
+	if err != nil {
+		return fmt.Errorf("kvstore: encode %s: %w", key, err)
+	}
+	if _, err := kv.Put(ctx, key, data); err != nil {
+		return fmt.Errorf("kvstore: put %s: %w", key, err)
+	}
+	return nil
+}
+
+// Exists reports whether a key is present without decoding its value, so a
+// record that cannot be unmarshalled is still reported as present.
+func (s *Store[T]) Exists(ctx context.Context, key string) (bool, error) {
+	kv, err := s.KV(ctx)
+	if err != nil {
+		return false, err
+	}
+	if _, err := kv.Get(ctx, key); err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return false, nil
+		}
+		return false, fmt.Errorf("kvstore: get %s: %w", key, err)
+	}
+	return true, nil
+}
+
+// CompareAndSet writes v only if key is still at rev, returning ErrConflict
+// when it is not. For callers whose lost race is a decision rather than a
+// failure; callers that should retry want Mutate.
+func (s *Store[T]) CompareAndSet(ctx context.Context, key string, v *T, rev uint64) error {
+	kv, err := s.KV(ctx)
+	if err != nil {
+		return err
+	}
+	data, err := json.Marshal(v)
+	if err != nil {
+		return fmt.Errorf("kvstore: encode %s: %w", key, err)
+	}
+	if _, err := kv.Update(ctx, key, data, rev); err != nil {
+		if errors.Is(err, jetstream.ErrKeyRevisionMismatch) || errors.Is(err, jetstream.ErrKeyExists) {
+			return fmt.Errorf("%w: %s", ErrConflict, key)
+		}
+		return fmt.Errorf("kvstore: update %s: %w", key, err)
+	}
+	return nil
+}
+
 // keys lists the bucket, treating an empty bucket as an empty listing.
 func (s *Store[T]) keys(ctx context.Context) (jetstream.KeyValue, []string, error) {
 	kv, err := s.KV(ctx)

spinifex/kvstore/store_test.go

@@ -2,10 +2,13 @@ package kvstore_test
 
 import (
 	"encoding/json"
+	"errors"
 	"testing"
 
 	"github.com/mulgadc/spinifex/spinifex/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/kvutil"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -184,3 +187,157 @@ func TestStore_NilJetStreamReportsTheConfiguredMessage(t *testing.T) {
 	_, err = store.List(t.Context(), "")
 	require.ErrorContains(t, err, "no JetStream client configured")
 }
+
+// TestStore_NilJetStreamWithNoMessageStillSaysSomething guards the fallback: a
+// Config that omits Missing must not produce an error with an empty string,
+// which reads as a passing call to anything logging err.Error().
+func TestStore_NilJetStreamWithNoMessageStillSaysSomething(t *testing.T) {
+	store := kvstore.New[record](nil, kvstore.Config{Name: "kvstore-test"})
+
+	_, _, err := store.Get(t.Context(), "acct-a/one")
+	require.Error(t, err)
+	assert.NotEmpty(t, err.Error(), "an error with no message is worse than a wrong one")
+	assert.ErrorContains(t, err, "no JetStream client configured")
+}
+
+// newOpenBucket starts an embedded JetStream server and returns an already-open
+// bucket handle, the shape an eager caller holds at construction.
+func newOpenBucket(t *testing.T) jetstream.KeyValue {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	kv, err := kvutil.GetOrCreateBucket(t.Context(), testutil.NewJetStream(t, nc), "kvstore-over-test", 1)
+	require.NoError(t, err)
+	return kv
+}
+
+// TestOver_ServesAPreOpenedBucket covers the eager-caller path end to end. It
+// also pins the order of the checks in Bucket.KV: a Store built by Over holds
+// no JetStream client, so a nil-client guard running before the memo check
+// would fail every one of these calls with an empty error string.
+func TestOver_ServesAPreOpenedBucket(t *testing.T) {
+	kv := newOpenBucket(t)
+
+	// A record written straight to the bucket, as a pre-existing one would be.
+	data, err := json.Marshal(record{Name: "one", Count: 7})
+	require.NoError(t, err)
+	_, err = kv.Put(t.Context(), "acct-a/one", data)
+	require.NoError(t, err)
+
+	store := kvstore.Over[record](kv, kvstore.Config{})
+
+	got, _, err := store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+	assert.Equal(t, record{Name: "one", Count: 7}, *got)
+
+	require.NoError(t, store.Set(t.Context(), "acct-a/two", &record{Name: "two"}))
+	all, err := store.List(t.Context(), "acct-a/")
+	require.NoError(t, err)
+	assert.Len(t, all, 2)
+}
+
+func TestStore_SetOverwritesWhereCreateWouldConflict(t *testing.T) {
+	store := newStore(t)
+	require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "first"}))
+	require.ErrorIs(t, store.Create(t.Context(), "acct-a/one", &record{Name: "second"}), kvstore.ErrExists)
+
+	require.NoError(t, store.Set(t.Context(), "acct-a/one", &record{Name: "second", Count: 9}))
+	got, _, err := store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+	assert.Equal(t, record{Name: "second", Count: 9}, *got, "Set replaces where Create refuses")
+
+	require.NoError(t, store.Set(t.Context(), "acct-a/new", &record{Name: "new"}))
+	got, _, err = store.Get(t.Context(), "acct-a/new")
+	require.NoError(t, err)
+	assert.Equal(t, "new", got.Name, "Set on an absent key creates it")
+}
+
+// TestStore_ExistsDoesNotDecode is the reason Exists is a method rather than a
+// Get in disguise: a record whose bytes will not unmarshal must still be
+// reported as present, or it becomes impossible to delete.
+func TestStore_ExistsDoesNotDecode(t *testing.T) {
+	store := newStore(t)
+	kv, err := store.KV(t.Context())
+	require.NoError(t, err)
+	_, err = kv.Put(t.Context(), "acct-a/corrupt", []byte("{not json"))
+	require.NoError(t, err)
+
+	_, _, err = store.Get(t.Context(), "acct-a/corrupt")
+	require.Error(t, err, "Get cannot read it")
+
+	present, err := store.Exists(t.Context(), "acct-a/corrupt")
+	require.NoError(t, err)
+	assert.True(t, present, "a corrupt record is still present, and must stay deletable")
+
+	require.NoError(t, store.Delete(t.Context(), "acct-a/corrupt"))
+	present, err = store.Exists(t.Context(), "acct-a/corrupt")
+	require.NoError(t, err)
+	assert.False(t, present)
+}
+
+func TestStore_ExistsIsFalseForAnAbsentKey(t *testing.T) {
+	store := newStore(t)
+
+	present, err := store.Exists(t.Context(), "nobody")
+	require.NoError(t, err, "an absent key is not an error, unlike Get")
+	assert.False(t, present)
+}
+
+func TestStore_CompareAndSetRejectsAStaleRevision(t *testing.T) {
+	store := newStore(t)
+	require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one", Count: 1}))
+	_, stale, err := store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+
+	// Another writer commits first, moving the revision on.
+	seedRaw(t, store, "acct-a/one", record{Name: "one", Count: 2})
+
+	err = store.CompareAndSet(t.Context(), "acct-a/one", &record{Name: "one", Count: 99}, stale)
+	require.ErrorIs(t, err, kvstore.ErrConflict)
+
+	got, fresh, err := store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+	assert.Equal(t, 2, got.Count, "a rejected write must leave the winner's value in place")
+
+	require.NoError(t, store.CompareAndSet(t.Context(), "acct-a/one", &record{Name: "one", Count: 3}, fresh))
+	got, _, err = store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+	assert.Equal(t, 3, got.Count, "the same call commits at the current revision")
+}
+
+// TestStore_ConfigAttemptsBoundsMutate pins both halves of the per-store budget:
+// the loop stops after Attempts tries, and the caller's Exhausted error is what
+// surfaces rather than kvutil's default message.
+func TestStore_ConfigAttemptsBoundsMutate(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	errBudgetSpent := errors.New("budget spent")
+	var gotKey string
+	var gotAttempts int
+
+	store := kvstore.New[record](testutil.NewJetStream(t, nc), kvstore.Config{
+		Name:     "kvstore-attempts-test",
+		History:  1,
+		Attempts: 2,
+		Exhausted: func(key string, attempts int) error {
+			gotKey, gotAttempts = key, attempts
+			return errBudgetSpent
+		},
+	})
+	require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one"}))
+
+	calls := 0
+	err := store.Mutate(t.Context(), "acct-a/one", func(rec *record) (bool, error) {
+		calls++
+		// Move the revision on every attempt, so no commit can ever win.
+		seedRaw(t, store, "acct-a/one", record{Name: "one", Count: calls})
+		rec.Count = 100
+		return true, nil
+	})
+	require.ErrorIs(t, err, errBudgetSpent, "the configured error must surface, not kvutil's default")
+	assert.Equal(t, 2, calls, "Attempts bounds the loop")
+	assert.Equal(t, "acct-a/one", gotKey)
+	assert.Equal(t, 2, gotAttempts, "Exhausted is told the budget it spent")
+
+	got, _, err := store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+	assert.Equal(t, 2, got.Count, "an exhausted Mutate commits nothing of its own")
+}