From 5fc4eb0289a0218144bebd51b2dec0f496785555 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Wed, 26 Aug 2026 22:07:46 +1000 Subject: [PATCH] refactor: convert the acm certificate store to kvstore (#890) --- spinifex/handlers/acm/getcertificate_test.go | 2 +- spinifex/handlers/acm/service_impl_test.go | 2 +- spinifex/handlers/acm/store.go | 117 ++++++-------- spinifex/handlers/acm/store_test.go | 23 ++- spinifex/kvstore/bucket.go | 28 +++- spinifex/kvstore/store.go | 64 +++++++- spinifex/kvstore/store_test.go | 157 +++++++++++++++++++ 7 files changed, 305 insertions(+), 88 deletions(-) diff --git a/spinifex/handlers/acm/getcertificate_test.go b/spinifex/handlers/acm/getcertificate_test.go index 24f9ef92e..7c2c757da 100644 --- a/spinifex/handlers/acm/getcertificate_test.go +++ b/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) } diff --git a/spinifex/handlers/acm/service_impl_test.go b/spinifex/handlers/acm/service_impl_test.go index 4da86d842..5c7df5b0d 100644 --- a/spinifex/handlers/acm/service_impl_test.go +++ b/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 diff --git a/spinifex/handlers/acm/store.go b/spinifex/handlers/acm/store.go index 065c6575e..1e267d753 100644 --- a/spinifex/handlers/acm/store.go +++ b/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 diff --git a/spinifex/handlers/acm/store_test.go b/spinifex/handlers/acm/store_test.go index 1722fb243..767e7d798 100644 --- a/spinifex/handlers/acm/store_test.go +++ b/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") diff --git a/spinifex/kvstore/bucket.go b/spinifex/kvstore/bucket.go index b9c053e29..1983e3a4e 100644 --- a/spinifex/kvstore/bucket.go +++ b/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 diff --git a/spinifex/kvstore/store.go b/spinifex/kvstore/store.go index 803d79941..1531eca9b 100644 --- a/spinifex/kvstore/store.go +++ b/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) diff --git a/spinifex/kvstore/store_test.go b/spinifex/kvstore/store_test.go index 4020c9d20..03c6ad4b8 100644 --- a/spinifex/kvstore/store_test.go +++ b/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") +}