feat: add kvstore typed store and convert the ochrevector job store (#888)

spinifex/handlers/ochrevector/ingest_test.go

@@ -6,7 +6,6 @@ package handlers_ochrevector
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"math"
 	"strings"
@@ -401,23 +400,15 @@ func TestRunJob_EmptyDocumentClearsExistingRows(t *testing.T) {
 	assert.Empty(t, backend.documentRows(ingestAccountA, "idx-one", key))
 }
 
-// forceJobUpdatedAt seeds jobID's UpdatedAt directly via the underlying KV
-// bucket, bypassing JobStore.Update (which always refreshes UpdatedAt to
-// now), so a test can simulate a record a crashed worker left stale --
-// mirroring appliance_test.go's raw-KV seeding for the same reason.
+// forceJobUpdatedAt seeds jobID's UpdatedAt directly, bypassing
+// JobStore.Update (which always refreshes UpdatedAt to now), so a test can
+// simulate a record a crashed worker left stale.
 func forceJobUpdatedAt(t *testing.T, store *JobStore, accountID, jobID string, when time.Time) {
 	t.Helper()
-	ctx := context.Background()
-	kv, err := store.bucket(ctx)
-	require.NoError(t, err)
-	entry, err := kv.Get(ctx, jobKey(accountID, jobID))
-	require.NoError(t, err)
-	var rec JobRecord
-	require.NoError(t, json.Unmarshal(entry.Value(), &rec))
-	rec.UpdatedAt = when
-	data, err := json.Marshal(rec)
-	require.NoError(t, err)
-	_, err = kv.Update(ctx, jobKey(accountID, jobID), data, entry.Revision())
+	err := store.store.Mutate(context.Background(), jobKey(accountID, jobID), func(rec *JobRecord) (bool, error) {
+		rec.UpdatedAt = when
+		return true, nil
+	})
 	require.NoError(t, err)
 }
 

spinifex/handlers/ochrevector/jobs.go

@@ -2,14 +2,11 @@ package handlers_ochrevector
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
-	"strings"
-	"sync"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -66,36 +63,18 @@ type JobRecord struct {
 }
 
 // JobStore persists JobRecords in the ochre-vector-jobs JetStream KV bucket,
-// mirroring Registry's lazy get-or-create bucket, key "accountID/id",
-// per-account prefix-scan list.
+// key "accountID/id", per-account prefix-scan list.
 type JobStore struct {
-	js jetstream.JetStream
-
-	mu sync.Mutex
-	kv jetstream.KeyValue
+	store *kvstore.Store[JobRecord]
 }
 
 // NewJobStore constructs a JobStore over js.
 func NewJobStore(js jetstream.JetStream) *JobStore {
-	return &JobStore{js: js}
-}
-
-// bucket lazily opens (or creates) the jobs KV bucket, caching the handle.
-func (s *JobStore) bucket(ctx context.Context) (jetstream.KeyValue, error) {
-	if s.js == nil {
-		return nil, errors.New("ochrevector: job store has no JetStream client configured")
-	}
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if s.kv != nil {
-		return s.kv, nil
-	}
-	kv, err := kvutil.GetOrCreateBucket(ctx, s.js, jobsBucket, jobsBucketHistory)
-	if err != nil {
-		return nil, err
-	}
-	s.kv = kv
-	return kv, nil
+	return &JobStore{store: kvstore.New[JobRecord](js, kvstore.Config{
+		Name:    jobsBucket,
+		History: jobsBucketHistory,
+		Missing: "ochrevector: job store has no JetStream client configured",
+	})}
 }
 
 // jobKey scopes every record to its owning account, so a foreign account's
@@ -109,17 +88,9 @@ func jobKey(accountID, jobID string) string {
 // starts of the same job id race safely and exactly one wins. rec.AccountID
 // is stamped from accountID, overriding whatever the caller set.
 func (s *JobStore) Reserve(ctx context.Context, accountID string, rec JobRecord) error {
-	kv, err := s.bucket(ctx)
-	if err != nil {
-		return err
-	}
 	rec.AccountID = accountID
-	data, err := json.Marshal(rec)
-	if err != nil {
-		return fmt.Errorf("ochrevector: encode job record %s: %w", rec.ID, err)
-	}
-	if _, err := kv.Create(ctx, jobKey(accountID, rec.ID), data); err != nil {
-		if errors.Is(err, jetstream.ErrKeyExists) {
+	if err := s.store.Create(ctx, jobKey(accountID, rec.ID), &rec); err != nil {
+		if errors.Is(err, kvstore.ErrExists) {
 			return ErrJobExists
 		}
 		return fmt.Errorf("ochrevector: reserve job %s: %w", rec.ID, err)
@@ -129,74 +100,23 @@ func (s *JobStore) Reserve(ctx context.Context, accountID string, rec JobRecord)
 
 // Get reads accountID's record for jobID, returning (nil, nil) when absent.
 func (s *JobStore) Get(ctx context.Context, accountID, jobID string) (*JobRecord, error) {
-	kv, err := s.bucket(ctx)
-	if err != nil {
-		return nil, err
+	rec, _, err := s.store.Get(ctx, jobKey(accountID, jobID))
+	if errors.Is(err, kvstore.ErrNotFound) {
+		return nil, nil
 	}
-	return getJobRecord(ctx, kv, jobKey(accountID, jobID))
-}
-
-// getJobRecord reads and decodes one record, returning (nil, nil) when absent.
-func getJobRecord(ctx context.Context, kv jetstream.KeyValue, key string) (*JobRecord, error) {
-	entry, err := kv.Get(ctx, key)
-	if err != nil {
-		if errors.Is(err, jetstream.ErrKeyNotFound) {
-			return nil, nil
-		}
-		return nil, fmt.Errorf("ochrevector: kv get %s: %w", key, err)
-	}
-	var rec JobRecord
-	if err := json.Unmarshal(entry.Value(), &rec); err != nil {
-		return nil, fmt.Errorf("ochrevector: decode %s: %w", key, err)
-	}
-	return &rec, nil
+	return rec, err
 }
 
 // List returns every job record owned by accountID, so one tenant's listing
 // never surfaces another's.
 func (s *JobStore) List(ctx context.Context, accountID string) ([]JobRecord, error) {
-	kv, err := s.bucket(ctx)
-	if err != nil {
-		return nil, err
-	}
-	return listJobRecords(ctx, kv, accountID+"/")
+	return s.store.List(ctx, accountID+"/")
 }
 
 // ListAll returns every job record across every account, for the ingestion
 // reconciler's crash-recovery sweep.
 func (s *JobStore) ListAll(ctx context.Context) ([]JobRecord, error) {
-	kv, err := s.bucket(ctx)
-	if err != nil {
-		return nil, err
-	}
-	return listJobRecords(ctx, kv, "")
-}
-
-// listJobRecords walks every key with the given prefix ("" matches
-// everything) and decodes each into a JobRecord, skipping any key that
-// disappears between the key listing and the read.
-func listJobRecords(ctx context.Context, kv jetstream.KeyValue, prefix string) ([]JobRecord, error) {
-	keys, err := kv.Keys(ctx)
-	if err != nil {
-		if errors.Is(err, jetstream.ErrNoKeysFound) {
-			return nil, nil
-		}
-		return nil, fmt.Errorf("ochrevector: list keys: %w", err)
-	}
-	var out []JobRecord
-	for _, key := range keys {
-		if !strings.HasPrefix(key, prefix) {
-			continue
-		}
-		rec, err := getJobRecord(ctx, kv, key)
-		if err != nil {
-			return nil, err
-		}
-		if rec != nil {
-			out = append(out, *rec)
-		}
-	}
-	return out, nil
+	return s.store.List(ctx, "")
 }
 
 // Update reads accountID's jobID record, applies fn to mutate it in place,
@@ -204,29 +124,15 @@ func listJobRecords(ctx context.Context, kv jetstream.KeyValue, prefix string) (
 // update, mirroring Registry.SetState's pattern for the general case of an
 // arbitrary field mutation (progress counters, failed-document list, error).
 func (s *JobStore) Update(ctx context.Context, accountID, jobID string, fn func(rec *JobRecord)) error {
-	kv, err := s.bucket(ctx)
+	err := s.store.Mutate(ctx, jobKey(accountID, jobID), func(rec *JobRecord) (bool, error) {
+		fn(rec)
+		rec.UpdatedAt = time.Now().UTC()
+		return true, nil
+	})
+	if errors.Is(err, kvstore.ErrNotFound) {
+		return ErrJobNotFound
+	}
 	if err != nil {
-		return err
-	}
-	key := jobKey(accountID, jobID)
-	entry, err := kv.Get(ctx, key)
-	if err != nil {
-		if errors.Is(err, jetstream.ErrKeyNotFound) {
-			return ErrJobNotFound
-		}
-		return fmt.Errorf("ochrevector: kv get %s: %w", key, err)
-	}
-	var rec JobRecord
-	if err := json.Unmarshal(entry.Value(), &rec); err != nil {
-		return fmt.Errorf("ochrevector: decode %s: %w", key, err)
-	}
-	fn(&rec)
-	rec.UpdatedAt = time.Now().UTC()
-	data, err := json.Marshal(rec)
-	if err != nil {
-		return fmt.Errorf("ochrevector: encode job record %s: %w", rec.ID, err)
-	}
-	if _, err := kv.Update(ctx, key, data, entry.Revision()); err != nil {
 		return fmt.Errorf("ochrevector: update job %s: %w", jobID, err)
 	}
 	return nil
@@ -244,35 +150,11 @@ func (s *JobStore) SetState(ctx context.Context, accountID, jobID, state string)
 // appliance's job history never references data that no longer exists.
 // Idempotent: an empty bucket is a no-op success.
 func (s *JobStore) PurgeAll(ctx context.Context) error {
-	kv, err := s.bucket(ctx)
-	if err != nil {
-		return err
-	}
-	keys, err := kv.Keys(ctx)
-	if err != nil {
-		if errors.Is(err, jetstream.ErrNoKeysFound) {
-			return nil
-		}
-		return fmt.Errorf("ochrevector: list keys: %w", err)
-	}
-	for _, key := range keys {
-		if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
-			return fmt.Errorf("ochrevector: purge job %s: %w", key, err)
-		}
-	}
-	return nil
+	return s.store.DeletePrefix(ctx, "")
 }
 
 // Delete removes accountID's jobID record. Idempotent: deleting an
 // already-absent record is a no-op success.
 func (s *JobStore) Delete(ctx context.Context, accountID, jobID string) error {
-	kv, err := s.bucket(ctx)
-	if err != nil {
-		return err
-	}
-	key := jobKey(accountID, jobID)
-	if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
-		return fmt.Errorf("ochrevector: delete job %s: %w", jobID, err)
-	}
-	return nil
+	return s.store.Delete(ctx, jobKey(accountID, jobID))
 }

spinifex/handlers/ochrevector/jobs_test.go

@@ -196,7 +196,7 @@ func TestJobStore_ListAll(t *testing.T) {
 }
 
 func TestJobStore_BucketRequiresJetStream(t *testing.T) {
-	store := &JobStore{}
-	_, err := store.bucket(context.Background())
-	assert.Error(t, err)
+	store := NewJobStore(nil)
+	_, err := store.Get(context.Background(), "acct-a", "job-1")
+	require.ErrorContains(t, err, "job store has no JetStream client configured")
 }

spinifex/kvstore/bucket.go

@@ -0,0 +1,68 @@
+// Package kvstore provides a typed store over a single JetStream KV bucket:
+// one memoised get-or-create accessor and a JSON codec for the record type,
+// replacing the accessor each caller used to hand-roll.
+package kvstore
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/kvutil"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// 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
+}
+
+// Bucket memoises a lazily created KV bucket. Construct it with NewBucket.
+type Bucket struct {
+	js  jetstream.JetStream
+	cfg Config
+
+	mu sync.Mutex
+	kv jetstream.KeyValue
+}
+
+// NewBucket returns a Bucket over js. A nil js is permitted: every call to KV then
+// fails with cfg.Missing rather than panicking.
+func NewBucket(js jetstream.JetStream, cfg Config) *Bucket {
+	return &Bucket{js: js, 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
+	}
+	kv, err := b.open(ctx)
+	if err != nil {
+		return nil, err
+	}
+	b.kv = kv
+	return kv, nil
+}
+
+// open picks the kvutil helper matching the configured TTL and replica count.
+func (b *Bucket) open(ctx context.Context) (jetstream.KeyValue, error) {
+	switch {
+	case b.cfg.TTL > 0:
+		return kvutil.GetOrCreateBucketWithTTL(ctx, b.js, b.cfg.Name, b.cfg.History, b.cfg.TTL)
+	case b.cfg.Replicas > 0:
+		return kvutil.GetOrCreateBucketWithReplicas(ctx, b.js, b.cfg.Name, b.cfg.History, b.cfg.Replicas)
+	default:
+		return kvutil.GetOrCreateBucket(ctx, b.js, b.cfg.Name, b.cfg.History)
+	}
+}

spinifex/kvstore/store.go

@@ -0,0 +1,159 @@
+package kvstore
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/mulgadc/spinifex/spinifex/kvutil"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Sentinels for the two conditions every caller distinguishes. Callers map
+// these onto their own package errors.
+var (
+	ErrNotFound = errors.New("kvstore: key not found")
+	ErrExists   = errors.New("kvstore: key already exists")
+)
+
+// Store is a Bucket plus a JSON codec for T.
+type Store[T any] struct {
+	*Bucket
+}
+
+// New returns a Store over js for records of type T.
+func New[T any](js jetstream.JetStream, cfg Config) *Store[T] {
+	return &Store[T]{Bucket: NewBucket(js, cfg)}
+}
+
+// Get reads and decodes one record, returning ErrNotFound when the key is
+// absent. The revision is returned for callers doing their own CAS.
+func (s *Store[T]) Get(ctx context.Context, key string) (*T, uint64, error) {
+	kv, err := s.KV(ctx)
+	if err != nil {
+		return nil, 0, err
+	}
+	return s.get(ctx, kv, key)
+}
+
+// Create writes a record only if the key is absent, returning ErrExists when
+// it is not. The create-only write is the single-writer claim.
+func (s *Store[T]) Create(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.Create(ctx, key, data); err != nil {
+		if errors.Is(err, jetstream.ErrKeyExists) {
+			return fmt.Errorf("%w: %s", ErrExists, key)
+		}
+		return fmt.Errorf("kvstore: create %s: %w", key, err)
+	}
+	return nil
+}
+
+// Mutate applies fn under a revision-guarded retry loop. fn reports whether it
+// changed anything; false commits no write. An absent key is ErrNotFound.
+func (s *Store[T]) Mutate(ctx context.Context, key string, fn func(*T) (bool, error)) error {
+	kv, err := s.KV(ctx)
+	if err != nil {
+		return err
+	}
+	_, err = kvutil.Update(ctx, kv, key, kvutil.CASConfig{
+		NotFound: fmt.Errorf("%w: %s", ErrNotFound, key),
+	}, fn)
+	return err
+}
+
+// Delete removes a key. Idempotent: an already-absent key is success.
+func (s *Store[T]) Delete(ctx context.Context, key string) error {
+	kv, err := s.KV(ctx)
+	if err != nil {
+		return err
+	}
+	if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+		return fmt.Errorf("kvstore: delete %s: %w", key, err)
+	}
+	return nil
+}
+
+// List decodes every record whose key carries the given prefix. An empty
+// prefix matches everything and an empty bucket is not an error. A key that
+// disappears between the listing and the read is skipped.
+func (s *Store[T]) List(ctx context.Context, prefix string) ([]T, error) {
+	kv, keys, err := s.keys(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var out []T
+	for _, key := range keys {
+		if !strings.HasPrefix(key, prefix) {
+			continue
+		}
+		v, _, err := s.get(ctx, kv, key)
+		if errors.Is(err, ErrNotFound) {
+			continue
+		}
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, *v)
+	}
+	return out, nil
+}
+
+// DeletePrefix removes every key carrying the given prefix. An empty prefix
+// purges the bucket. Idempotent, like Delete.
+func (s *Store[T]) DeletePrefix(ctx context.Context, prefix string) error {
+	kv, keys, err := s.keys(ctx)
+	if err != nil {
+		return err
+	}
+	for _, key := range keys {
+		if !strings.HasPrefix(key, prefix) {
+			continue
+		}
+		if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+			return fmt.Errorf("kvstore: delete %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)
+	if err != nil {
+		return nil, nil, err
+	}
+	keys, err := kv.Keys(ctx)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrNoKeysFound) {
+			return kv, nil, nil
+		}
+		return nil, nil, fmt.Errorf("kvstore: list keys: %w", err)
+	}
+	return kv, keys, nil
+}
+
+// get is Get against an already-opened bucket, for loops that opened it once.
+func (s *Store[T]) get(ctx context.Context, kv jetstream.KeyValue, key string) (*T, uint64, error) {
+	entry, err := kv.Get(ctx, key)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return nil, 0, fmt.Errorf("%w: %s", ErrNotFound, key)
+		}
+		return nil, 0, fmt.Errorf("kvstore: get %s: %w", key, err)
+	}
+	var v T
+	if err := json.Unmarshal(entry.Value(), &v); err != nil {
+		return nil, 0, fmt.Errorf("kvstore: decode %s: %w", key, err)
+	}
+	return &v, entry.Revision(), nil
+}

spinifex/kvstore/store_test.go

@@ -0,0 +1,186 @@
+package kvstore_test
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// record is a stand-in for any caller's JSON document.
+type record struct {
+	Name  string `json:"name"`
+	Count int    `json:"count"`
+}
+
+// newStore starts an embedded JetStream server and returns a Store over a
+// bucket on it. Each test gets its own server, so bucket names need not differ.
+func newStore(t *testing.T) *kvstore.Store[record] {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	return kvstore.New[record](testutil.NewJetStream(t, nc), kvstore.Config{
+		Name:    "kvstore-test",
+		History: 1,
+		Missing: "kvstore test: no JetStream client configured",
+	})
+}
+
+// seedRaw writes a record straight to the bucket, bypassing Store, so a test
+// can move the revision underneath an in-flight Mutate.
+func seedRaw(t *testing.T, store *kvstore.Store[record], key string, rec record) {
+	t.Helper()
+	kv, err := store.KV(t.Context())
+	require.NoError(t, err)
+	data, err := json.Marshal(rec)
+	require.NoError(t, err)
+	_, err = kv.Put(t.Context(), key, data)
+	require.NoError(t, err)
+}
+
+func TestStore_GetAbsentKeyIsNotFound(t *testing.T) {
+	store := newStore(t)
+
+	_, _, err := store.Get(t.Context(), "nobody")
+	require.ErrorIs(t, err, kvstore.ErrNotFound)
+	assert.ErrorContains(t, err, "nobody", "the sentinel should name the key it could not find")
+}
+
+func TestStore_CreateRoundTripsAndRejectsDuplicate(t *testing.T) {
+	store := newStore(t)
+
+	require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one", Count: 3}))
+
+	got, rev, err := store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+	assert.Equal(t, record{Name: "one", Count: 3}, *got)
+	assert.NotZero(t, rev, "a committed record has a revision")
+
+	err = store.Create(t.Context(), "acct-a/one", &record{Name: "impostor"})
+	require.ErrorIs(t, err, kvstore.ErrExists)
+
+	// The loser of the create must not have overwritten the winner.
+	got, _, err = store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+	assert.Equal(t, "one", got.Name)
+}
+
+func TestStore_DeleteIsIdempotent(t *testing.T) {
+	store := newStore(t)
+	require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one"}))
+
+	require.NoError(t, store.Delete(t.Context(), "acct-a/one"))
+	require.NoError(t, store.Delete(t.Context(), "acct-a/one"), "deleting an absent key is success")
+	require.NoError(t, store.Delete(t.Context(), "never-existed"))
+
+	_, _, err := store.Get(t.Context(), "acct-a/one")
+	require.ErrorIs(t, err, kvstore.ErrNotFound)
+}
+
+func TestStore_ListOnEmptyBucketIsNotAnError(t *testing.T) {
+	store := newStore(t)
+
+	out, err := store.List(t.Context(), "")
+	require.NoError(t, err, "an empty bucket is an empty listing, not a failure")
+	assert.Empty(t, out)
+}
+
+func TestStore_ListFiltersByPrefix(t *testing.T) {
+	store := newStore(t)
+	require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one"}))
+	require.NoError(t, store.Create(t.Context(), "acct-a/two", &record{Name: "two"}))
+	require.NoError(t, store.Create(t.Context(), "acct-b/three", &record{Name: "three"}))
+
+	mine, err := store.List(t.Context(), "acct-a/")
+	require.NoError(t, err)
+	assert.Len(t, mine, 2, "one tenant's listing must not surface another's")
+
+	all, err := store.List(t.Context(), "")
+	require.NoError(t, err)
+	assert.Len(t, all, 3, "an empty prefix matches everything")
+}
+
+// TestStore_MutateRetriesARevisionConflict forces the conflict deterministically
+// rather than racing two goroutines: the first mutate attempt writes a competing
+// value out of band, so the commit that follows it must lose and re-read.
+func TestStore_MutateRetriesARevisionConflict(t *testing.T) {
+	store := newStore(t)
+	require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one", Count: 1}))
+
+	attempts := 0
+	err := store.Mutate(t.Context(), "acct-a/one", func(rec *record) (bool, error) {
+		attempts++
+		if attempts == 1 {
+			seedRaw(t, store, "acct-a/one", record{Name: "one", Count: 99})
+		}
+		rec.Count++
+		return true, nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, 2, attempts, "the first commit loses the CAS and the loop re-runs mutate")
+
+	got, _, err := store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+	assert.Equal(t, 100, got.Count, "the retry must re-read the competing write, not reapply to a stale copy")
+}
+
+func TestStore_MutateReportingNoChangeCommitsNoWrite(t *testing.T) {
+	store := newStore(t)
+	require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one", Count: 1}))
+	_, before, err := store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+
+	err = store.Mutate(t.Context(), "acct-a/one", func(rec *record) (bool, error) {
+		rec.Count = 42
+		return false, nil
+	})
+	require.NoError(t, err)
+
+	got, after, err := store.Get(t.Context(), "acct-a/one")
+	require.NoError(t, err)
+	assert.Equal(t, 1, got.Count, "a mutate reporting no change must not commit its edits")
+	assert.Equal(t, before, after, "no write means no new revision")
+}
+
+func TestStore_MutateOnAbsentKeyIsNotFound(t *testing.T) {
+	store := newStore(t)
+
+	err := store.Mutate(t.Context(), "nobody", func(*record) (bool, error) {
+		t.Error("mutate must not run against an absent key")
+		return false, nil
+	})
+	require.ErrorIs(t, err, kvstore.ErrNotFound)
+}
+
+func TestStore_DeletePrefixLeavesOtherPrefixes(t *testing.T) {
+	store := newStore(t)
+	require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one"}))
+	require.NoError(t, store.Create(t.Context(), "acct-b/two", &record{Name: "two"}))
+
+	require.NoError(t, store.DeletePrefix(t.Context(), "acct-a/"))
+
+	remaining, err := store.List(t.Context(), "")
+	require.NoError(t, err)
+	require.Len(t, remaining, 1)
+	assert.Equal(t, "two", remaining[0].Name)
+
+	require.NoError(t, store.DeletePrefix(t.Context(), "acct-a/"), "purging an already-empty prefix is success")
+}
+
+func TestStore_NilJetStreamReportsTheConfiguredMessage(t *testing.T) {
+	store := kvstore.New[record](nil, kvstore.Config{
+		Name:    "kvstore-test",
+		Missing: "kvstore test: no JetStream client configured",
+	})
+
+	_, _, err := store.Get(t.Context(), "acct-a/one")
+	require.ErrorContains(t, err, "no JetStream client configured")
+
+	err = store.Create(t.Context(), "acct-a/one", &record{Name: "one"})
+	require.ErrorContains(t, err, "no JetStream client configured")
+
+	_, err = store.List(t.Context(), "")
+	require.ErrorContains(t, err, "no JetStream client configured")
+}