refactor: fold kvutil into kvstore and rename Claim to Take (#1608)

* refactor: fold kvutil into kvstore and rename Claim to Take

kvutil was the layer under kvstore, so its bucket, replica, version, key
and CAS helpers now live in kvstore and every importer changes only its
import path and qualifier. kvutil.Claim becomes kvstore.Take so that it
does not read as a lock claim, and the kvutil package is deleted.

The bucket-creation invariant now exempts kvstore/buckets.go, the one
file holding the chokepoint, rather than the whole package.

* refactor: merge the kvstore test exports into one file

* refactor: move the kvutil callers added on dev to kvstore

spinifex/domains/acm/store.go

@@ -14,7 +14,6 @@ import (
 	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -184,7 +183,7 @@ func NewStore(ctx context.Context, nc *nats.Conn, masterKey []byte) (*Store, err
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketACM, KVBucketACMVersion)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketACM, KVBucketACMVersion)
 	if err != nil {
 		return nil, err
 	}

spinifex/domains/admission/quota/bedrock.go

@@ -10,7 +10,7 @@ import (
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 	"golang.org/x/time/rate"
 )
@@ -171,7 +171,7 @@ func (s *Service) RunBedrockRPMSync(ctx context.Context, js jetstream.JetStream)
 	if s == nil || !s.limits.RequestsPerMinuteEnabled {
 		return
 	}
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketBedrockRPMUsage, 1)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketBedrockRPMUsage, 1)
 	if err != nil {
 		slog.Warn("bedrock rpm sync: open usage bucket failed, sync disabled", "err", err)
 		return

spinifex/domains/admission/quota/override.go

@@ -9,7 +9,7 @@ import (
 	"time"
 
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -187,7 +187,7 @@ func (s *Service) PutAccountQuota(ctx context.Context, accountID string, over Ov
 	over.UpdatedAt = time.Now().UTC().Format(time.RFC3339)
 	over.UpdatedBy = updatedBy
 
-	_, err := kvutil.Update(ctx, s.overrides, accountID, kvutil.CASConfig{
+	_, err := kvstore.Update(ctx, s.overrides, accountID, kvstore.CASConfig{
 		CreateIfAbsent: true,
 		Attempts:       overrideMaxAttempts,
 		Exhausted:      func(string, int) error { return errOverridesContended },
@@ -230,7 +230,7 @@ func (s *Service) clearAccountQuota(ctx context.Context, accountID string) error
 			return nil
 		case errors.Is(err, jetstream.ErrKeyNotFound):
 			return nil
-		case kvutil.IsRevisionConflict(err):
+		case kvstore.IsRevisionConflict(err):
 			// A concurrent write landed between the read and the guarded
 			// delete; retry against whatever revision now holds.
 			continue

spinifex/domains/admission/quota/seal.go

@@ -8,7 +8,7 @@ import (
 
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -73,7 +73,7 @@ func (s *Service) sealBucket(ctx context.Context) (jetstream.KeyValue, error) {
 		slog.ErrorContext(ctx, "Quota override seal has no JetStream client wired to open the override bucket")
 		return nil, errors.New(awserrors.ErrorInternalError)
 	}
-	kv, err := kvutil.GetOrCreateBucket(ctx, s.js, KVBucketAccountQuota, 1)
+	kv, err := kvstore.GetOrCreateBucket(ctx, s.js, KVBucketAccountQuota, 1)
 	if err != nil {
 		slog.ErrorContext(ctx, "Failed to open quota override bucket for seal", "err", err)
 		return nil, errors.New(awserrors.ErrorInternalError)
@@ -136,7 +136,7 @@ func (s *Service) SealAccountOverrides(ctx context.Context, accountID string) (*
 		switch {
 		case writeErr == nil:
 			return &OverrideSeal{Revision: newRev}, nil
-		case kvutil.IsRevisionConflict(writeErr):
+		case kvstore.IsRevisionConflict(writeErr):
 			continue
 		default:
 			slog.ErrorContext(ctx, "Failed to write sealed quota override record", "accountID", accountID, "err", writeErr)

spinifex/domains/ec2/account/seal.go

@@ -8,7 +8,7 @@ import (
 
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -109,7 +109,7 @@ func (s *AccountSettingsServiceImpl) SealAccount(ctx context.Context, accountID
 		switch {
 		case writeErr == nil:
 			return &AccountSettingsSeal{Revision: newRev}, nil
-		case kvutil.IsRevisionConflict(writeErr):
+		case kvstore.IsRevisionConflict(writeErr):
 			continue
 		default:
 			slog.ErrorContext(ctx, "Failed to write sealed account settings record", "accountID", accountID, "err", writeErr)

spinifex/domains/ec2/account/service_impl.go

@@ -13,7 +13,7 @@ import (
 	ebsencryption "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/encryption"
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -57,7 +57,7 @@ func NewAccountSettingsServiceImplWithNATS(ctx context.Context, cfg *config.Conf
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	settingsKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketAccountSettings, 10)
+	settingsKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccountSettings, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create account settings KV bucket: %w", err)
 	}
@@ -115,13 +115,13 @@ func mapSettingsError(ctx context.Context, accountID string, err error) error {
 		return errors.New(awserrors.ErrorBlocked)
 	case errors.Is(err, errSettingsContended):
 		slog.ErrorContext(ctx, "Account settings CAS retries exhausted under contention", "accountID", accountID)
-	case errors.Is(err, kvutil.ErrRead):
+	case errors.Is(err, kvstore.ErrRead):
 		slog.ErrorContext(ctx, "Failed to read account settings from KV", "accountID", accountID, "err", err)
-	case errors.Is(err, kvutil.ErrDecode):
+	case errors.Is(err, kvstore.ErrDecode):
 		slog.ErrorContext(ctx, "Corrupt account settings record in KV", "accountID", accountID, "err", err)
-	case errors.Is(err, kvutil.ErrEncode):
+	case errors.Is(err, kvstore.ErrEncode):
 		slog.ErrorContext(ctx, "Failed to marshal account settings record", "accountID", accountID, "err", err)
-	case errors.Is(err, kvutil.ErrWrite):
+	case errors.Is(err, kvstore.ErrWrite):
 		slog.ErrorContext(ctx, "Failed to write account settings to KV", "accountID", accountID, "err", err)
 	default:
 		slog.ErrorContext(ctx, "Account settings operation failed", "accountID", accountID, "err", err)
@@ -133,7 +133,7 @@ func mapSettingsError(ctx context.Context, accountID string, err error) error {
 // changed is always true, so every success is a leader-accepted write over an
 // active value; a seal landing between the read and the write fails the CAS.
 func (s *AccountSettingsServiceImpl) mutateSettings(ctx context.Context, accountID string, set func(*AccountSettingsRecord)) error {
-	_, err := kvutil.Update(ctx, s.settingsKV, settingsKey(accountID), kvutil.CASConfig{
+	_, err := kvstore.Update(ctx, s.settingsKV, settingsKey(accountID), kvstore.CASConfig{
 		CreateIfAbsent: true,
 		Exhausted:      func(string, int) error { return errSettingsContended },
 	}, func(rec *AccountSettingsRecord) (bool, error) {

spinifex/domains/ec2/eigw/service_impl.go

@@ -19,7 +19,7 @@ import (
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -56,7 +56,7 @@ func NewEgressOnlyIGWServiceImplWithNATS(ctx context.Context, cfg *config.Config
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	eigwKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketEgressOnlyIGW, 10)
+	eigwKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketEgressOnlyIGW, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEgressOnlyIGW, err)
 	}
@@ -65,7 +65,7 @@ func NewEgressOnlyIGWServiceImplWithNATS(ctx context.Context, cfg *config.Config
 	}
 
 	// Get or create VPC KV bucket for cross-resource ownership validation
-	vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
+	vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
 	}
@@ -90,7 +90,7 @@ func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(ctx context.C
 		slog.ErrorContext(ctx, "VPC KV unavailable, cannot verify VPC ownership")
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
-	if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, *input.VpcId)); err != nil {
+	if _, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, *input.VpcId)); err != nil {
 		slog.WarnContext(ctx, "CreateEgressOnlyInternetGateway: VPC not found for account", "vpcId", *input.VpcId, "accountID", accountID)
 		return nil, awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", *input.VpcId)
 	}
@@ -109,7 +109,7 @@ func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(ctx context.C
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal Egress-only IGW record: %w", err)
 	}
-	if _, err := s.eigwKV.Put(ctx, kvutil.AccountKey(accountID, eigwID), data); err != nil {
+	if _, err := s.eigwKV.Put(ctx, kvstore.AccountKey(accountID, eigwID), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -127,7 +127,7 @@ func (s *EgressOnlyIGWServiceImpl) DeleteEgressOnlyInternetGateway(ctx context.C
 	}
 
 	eigwID := *input.EgressOnlyInternetGatewayId
-	key := kvutil.AccountKey(accountID, eigwID)
+	key := kvstore.AccountKey(accountID, eigwID)
 
 	// Verify the EIGW exists before deleting
 	if _, err := s.eigwKV.Get(ctx, key); err != nil {
@@ -178,7 +178,7 @@ func (s *EgressOnlyIGWServiceImpl) DescribeEgressOnlyInternetGateways(ctx contex
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, prefix) {

spinifex/domains/ec2/eigw/service_impl_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/mulgadc/spinifex/internal/testkit"
 	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -23,7 +23,7 @@ func setupTestEIGWService(t *testing.T) *EgressOnlyIGWServiceImpl {
 	// Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass
 	vpcEntries := map[string][]byte{}
 	for _, vpcID := range []string{"vpc-test123", "vpc-tagged", "vpc-tagged2"} {
-		vpcEntries[kvutil.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
+		vpcEntries[kvstore.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
 	}
 	testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, vpcEntries)
 
@@ -312,7 +312,7 @@ func TestCreateEgressOnlyInternetGateway_CrossAccountVPCRejected(t *testing.T) {
 	// Create VPC KV bucket with a VPC owned by testAccountID
 	vpcID := "vpc-alpha456"
 	testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
-		kvutil.AccountKey(testAccountID, vpcID): []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`),
+		kvstore.AccountKey(testAccountID, vpcID): []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`),
 	})
 
 	var err error

spinifex/domains/ec2/eip/allocate_idempotency.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -36,7 +36,7 @@ func (s *EIPServiceImpl) allocateOnce(ctx context.Context, accountID string, all
 	if key == "" || s.idemKV == nil {
 		return alloc()
 	}
-	kvKey := kvutil.AccountKey(accountID, idempotencyKeyHash(key))
+	kvKey := kvstore.AccountKey(accountID, idempotencyKeyHash(key))
 
 	if out := s.cachedAllocation(ctx, kvKey); out != nil {
 		slog.WarnContext(ctx, "AllocateAddress: returning first result for a retried request",

spinifex/domains/ec2/eip/lease_rebind.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -80,7 +80,7 @@ func (s *EIPServiceImpl) AllocationExists(ctx context.Context, allocationID stri
 	}
 	suffix := "." + allocationID
 	for _, k := range keys {
-		if k != kvutil.VersionKey && strings.HasSuffix(k, suffix) {
+		if k != kvstore.VersionKey && strings.HasSuffix(k, suffix) {
 			return true, nil
 		}
 	}
@@ -98,7 +98,7 @@ func (s *EIPServiceImpl) findByAllocationIDAnyAccount(ctx context.Context, alloc
 
 	suffix := "." + allocationID
 	for _, k := range keys {
-		if k == kvutil.VersionKey || !strings.HasSuffix(k, suffix) {
+		if k == kvstore.VersionKey || !strings.HasSuffix(k, suffix) {
 			continue
 		}
 		entry, err := s.eipKV.Get(ctx, k)

spinifex/domains/ec2/eip/service_impl.go

@@ -20,7 +20,7 @@ import (
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -66,7 +66,7 @@ func NewEIPServiceImpl(ctx context.Context, natsConn *nats.Conn, externalIPAM *e
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	eipKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketEIPs, 10)
+	eipKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketEIPs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEIPs, err)
 	}
@@ -76,7 +76,7 @@ func NewEIPServiceImpl(ctx context.Context, natsConn *nats.Conn, externalIPAM *e
 
 	// TTL'd, so a failure to create it is not worth refusing to serve EIPs over:
 	// a nil bucket only costs post-completion dedupe.
-	idemKV, err := kvutil.GetOrCreateBucketWithTTL(ctx, js, KVBucketEIPIdempotency, 1, eipIdempotencyTTL)
+	idemKV, err := kvstore.GetOrCreateBucketWithTTL(ctx, js, KVBucketEIPIdempotency, 1, eipIdempotencyTTL)
 	if err != nil {
 		slog.Warn("EIP service: idempotency bucket unavailable; a retried AllocateAddress may allocate twice",
 			"bucket", KVBucketEIPIdempotency, "err", err)
@@ -146,7 +146,7 @@ func (s *EIPServiceImpl) allocateAddress(ctx context.Context, input *ec2.Allocat
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal EIP record: %w", err)
 	}
-	if _, err := s.eipKV.Put(ctx, kvutil.AccountKey(accountID, allocID), data); err != nil {
+	if _, err := s.eipKV.Put(ctx, kvstore.AccountKey(accountID, allocID), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -167,7 +167,7 @@ func (s *EIPServiceImpl) ReleaseAddress(ctx context.Context, input *ec2.ReleaseA
 	}
 
 	allocID := *input.AllocationId
-	key := kvutil.AccountKey(accountID, allocID)
+	key := kvstore.AccountKey(accountID, allocID)
 
 	entry, err := s.eipKV.Get(ctx, key)
 	if err != nil {
@@ -206,7 +206,7 @@ func (s *EIPServiceImpl) AssociateAddress(ctx context.Context, input *ec2.Associ
 	}
 
 	allocID := *input.AllocationId
-	key := kvutil.AccountKey(accountID, allocID)
+	key := kvstore.AccountKey(accountID, allocID)
 
 	entry, err := s.eipKV.Get(ctx, key)
 	if err != nil {
@@ -536,7 +536,7 @@ func (s *EIPServiceImpl) findByENI(ctx context.Context, accountID, eniID string)
 		if err := ctx.Err(); err != nil {
 			return nil, err
 		}
-		if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
 			continue
 		}
 		entry, err := s.eipKV.Get(ctx, k)
@@ -590,7 +590,7 @@ func (s *EIPServiceImpl) DescribeAddresses(ctx context.Context, input *ec2.Descr
 
 	var addresses []*ec2.Address
 	for _, k := range keys {
-		if k == kvutil.VersionKey {
+		if k == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(k, prefix) {
@@ -699,7 +699,7 @@ func (s *EIPServiceImpl) DescribeAddressesAttribute(ctx context.Context, input *
 			if id == nil {
 				continue
 			}
-			key := kvutil.AccountKey(accountID, *id)
+			key := kvstore.AccountKey(accountID, *id)
 			entry, err := s.eipKV.Get(ctx, key)
 			if err != nil {
 				continue // not found, skip
@@ -722,7 +722,7 @@ func (s *EIPServiceImpl) DescribeAddressesAttribute(ctx context.Context, input *
 			return nil, errors.New(awserrors.ErrorServerInternal)
 		}
 		for _, k := range keys {
-			if k == kvutil.VersionKey {
+			if k == kvstore.VersionKey {
 				continue
 			}
 			if !strings.HasPrefix(k, prefix) {
@@ -819,7 +819,7 @@ func (s *EIPServiceImpl) findByAssociationID(ctx context.Context, accountID, ass
 	}
 
 	for _, k := range keys {
-		if k == kvutil.VersionKey {
+		if k == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(k, prefix) {
@@ -857,7 +857,7 @@ func (s *EIPServiceImpl) AssociatedPublicIPForInstance(ctx context.Context, acco
 		return "", false
 	}
 	for _, k := range keys {
-		if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
 			continue
 		}
 		entry, err := s.eipKV.Get(ctx, k)

spinifex/domains/ec2/igw/service_impl.go

@@ -22,7 +22,7 @@ import (
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
 	awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -93,7 +93,7 @@ func NewIGWServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	igwKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketIGW, 10)
+	igwKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketIGW, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketIGW, err)
 	}
@@ -102,13 +102,13 @@ func NewIGWServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 	}
 
 	// Get or create VPC KV bucket for cross-resource ownership validation
-	vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
+	vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
 	}
 
 	// Read EIP associations so detach can refuse while addresses are mapped.
-	eipKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
+	eipKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get EIP KV bucket: %w", err)
 	}
@@ -142,7 +142,7 @@ func (s *IGWServiceImpl) createIGW(ctx context.Context, input *ec2.CreateInterne
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal IGW record: %w", err)
 	}
-	if _, err := s.igwKV.Put(ctx, kvutil.AccountKey(accountID, igwID), data); err != nil {
+	if _, err := s.igwKV.Put(ctx, kvstore.AccountKey(accountID, igwID), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -160,7 +160,7 @@ func (s *IGWServiceImpl) DeleteInternetGateway(ctx context.Context, input *ec2.D
 	}
 
 	igwID := *input.InternetGatewayId
-	key := kvutil.AccountKey(accountID, igwID)
+	key := kvstore.AccountKey(accountID, igwID)
 
 	entry, err := s.igwKV.Get(ctx, key)
 	if err != nil {
@@ -236,7 +236,7 @@ func (s *IGWServiceImpl) DescribeInternetGateways(ctx context.Context, input *ec
 	foundIDs := make(map[string]bool)
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, prefix) {
@@ -326,7 +326,7 @@ func (s *IGWServiceImpl) AttachInternetGateway(ctx context.Context, input *ec2.A
 
 	igwID := *input.InternetGatewayId
 	vpcID := *input.VpcId
-	key := kvutil.AccountKey(accountID, igwID)
+	key := kvstore.AccountKey(accountID, igwID)
 
 	entry, err := s.igwKV.Get(ctx, key)
 	if err != nil {
@@ -347,7 +347,7 @@ func (s *IGWServiceImpl) AttachInternetGateway(ctx context.Context, input *ec2.A
 		slog.ErrorContext(ctx, "VPC KV unavailable, cannot verify VPC ownership")
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
-	if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID)); err != nil {
+	if _, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcID)); err != nil {
 		slog.WarnContext(ctx, "AttachInternetGateway: VPC not found for account", "vpcId", vpcID, "accountID", accountID)
 		return nil, awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", vpcID)
 	}
@@ -388,7 +388,7 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D
 
 	igwID := *input.InternetGatewayId
 	vpcID := *input.VpcId
-	key := kvutil.AccountKey(accountID, igwID)
+	key := kvstore.AccountKey(accountID, igwID)
 
 	entry, err := s.igwKV.Get(ctx, key)
 	if err != nil {
@@ -449,7 +449,7 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D
 // associated into vpcID, whether to an interface or a NAT gateway. The listing
 // is filtered server-side to the account's keys.
 func (s *IGWServiceImpl) vpcHasMappedAddress(ctx context.Context, accountID, vpcID string) (bool, error) {
-	lister, err := s.eipKV.ListKeysFiltered(ctx, kvutil.AccountKey(accountID, "*"))
+	lister, err := s.eipKV.ListKeysFiltered(ctx, kvstore.AccountKey(accountID, "*"))
 	if err != nil {
 		return false, err
 	}
@@ -493,11 +493,11 @@ func (s *IGWServiceImpl) CreateAttachedInternetGateway(ctx context.Context, acco
 	if s.vpcKV == nil {
 		return false, errors.New(awserrors.ErrorServerInternal)
 	}
-	if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID)); err != nil {
+	if _, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcID)); err != nil {
 		return false, awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", vpcID)
 	}
 
-	key := kvutil.AccountKey(accountID, igwID)
+	key := kvstore.AccountKey(accountID, igwID)
 	data, err := json.Marshal(IGWRecord{
 		InternetGatewayId: igwID,
 		VpcId:             vpcID,
@@ -558,7 +558,7 @@ func (s *IGWServiceImpl) AttachmentIntent(ctx context.Context, accountID, vpcID
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.igwKV.Get(ctx, key)
@@ -599,7 +599,7 @@ func (s *IGWServiceImpl) AttachmentIntent(ctx context.Context, accountID, vpcID
 // Writing only on the pending transition keeps a pass from waking the drift loop.
 func MarkAttached(ctx context.Context, kv jetstream.KeyValue, recordKey, vpcID string) error {
 	confirmed := false
-	record, err := kvutil.Update(ctx, kv, recordKey, kvutil.CASConfig{}, func(r *IGWRecord) (bool, error) {
+	record, err := kvstore.Update(ctx, kv, recordKey, kvstore.CASConfig{}, func(r *IGWRecord) (bool, error) {
 		// Reset per attempt: a retried CAS may see a record another writer has
 		// already moved out of pending.
 		confirmed = r.VpcId == vpcID && r.AttachState == AttachStatePending

spinifex/domains/ec2/igw/service_impl_test.go

@@ -13,7 +13,7 @@ import (
 	ec2eip "github.com/mulgadc/spinifex/spinifex/domains/ec2/eip"
 	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -55,7 +55,7 @@ func setupTestIGWService(t *testing.T) (*IGWServiceImpl, *nats.Conn) {
 	// Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass
 	vpcEntries := map[string][]byte{}
 	for _, vpcID := range []string{"vpc-test123", "vpc-other", "vpc-lifecycle", "vpc-event-test"} {
-		vpcEntries[kvutil.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
+		vpcEntries[kvstore.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
 	}
 	testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, vpcEntries)
 
@@ -68,7 +68,7 @@ func setupTestIGWService(t *testing.T) (*IGWServiceImpl, *nats.Conn) {
 // gateway come up. Without it an attach stays pending and reports no attachment.
 func confirmAttach(t *testing.T, svc *IGWServiceImpl, igwID, vpcID string) {
 	t.Helper()
-	require.NoError(t, MarkAttached(context.Background(), svc.igwKV, kvutil.AccountKey(testAccountID, igwID), vpcID))
+	require.NoError(t, MarkAttached(context.Background(), svc.igwKV, kvstore.AccountKey(testAccountID, igwID), vpcID))
 }
 
 func createTestIGW(t *testing.T, svc *IGWServiceImpl) string {
@@ -331,7 +331,7 @@ func TestDescribeInternetGateways_LegacyRecordReportsAttachment(t *testing.T) {
 	svc, _ := setupTestIGWService(t)
 	igwID := createTestIGW(t, svc)
 
-	key := kvutil.AccountKey(testAccountID, igwID)
+	key := kvstore.AccountKey(testAccountID, igwID)
 	entry, err := svc.igwKV.Get(context.Background(), key)
 	require.NoError(t, err)
 	var record IGWRecord
@@ -398,7 +398,7 @@ func TestMarkAttached_NoRewriteWhenAlreadyAttached(t *testing.T) {
 	}, testAccountID)
 	require.NoError(t, err)
 
-	key := kvutil.AccountKey(testAccountID, igwID)
+	key := kvstore.AccountKey(testAccountID, igwID)
 	confirmAttach(t, svc, igwID, "vpc-test123")
 	first, err := svc.igwKV.Get(context.Background(), key)
 	require.NoError(t, err)
@@ -605,7 +605,7 @@ func TestDetachInternetGateway_MappedPublicAddress(t *testing.T) {
 	putEIP := func(rec ec2eip.EIPRecord) {
 		data, err := json.Marshal(rec)
 		require.NoError(t, err)
-		_, err = svc.eipKV.Put(context.Background(), kvutil.AccountKey(testAccountID, rec.AllocationId), data)
+		_, err = svc.eipKV.Put(context.Background(), kvstore.AccountKey(testAccountID, rec.AllocationId), data)
 		require.NoError(t, err)
 	}
 	putEIP(ec2eip.EIPRecord{AllocationId: "eipalloc-other", AssociationId: "eipassoc-other", VpcId: "vpc-other", State: "associated"})
@@ -614,7 +614,7 @@ func TestDetachInternetGateway_MappedPublicAddress(t *testing.T) {
 	// Another account's address naming the same VPC ID is not this account's dependency.
 	foreign, err := json.Marshal(ec2eip.EIPRecord{AllocationId: "eipalloc-foreign", AssociationId: "eipassoc-foreign", VpcId: "vpc-test123", State: "associated"})
 	require.NoError(t, err)
-	_, err = svc.eipKV.Put(context.Background(), kvutil.AccountKey("999999999999", "eipalloc-foreign"), foreign)
+	_, err = svc.eipKV.Put(context.Background(), kvstore.AccountKey("999999999999", "eipalloc-foreign"), foreign)
 	require.NoError(t, err)
 
 	detach := func() error {
@@ -828,7 +828,7 @@ func TestAttachInternetGateway_CrossAccountVPCRejected(t *testing.T) {
 	require.NoError(t, err)
 
 	vpcID := "vpc-alpha123"
-	_, err = vpcKV.Put(t.Context(), kvutil.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`))
+	_, err = vpcKV.Put(t.Context(), kvstore.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`))
 	require.NoError(t, err)
 
 	// Refresh service to pick up the VPC KV bucket

spinifex/domains/ec2/instance/uefistore/cleanup.go

@@ -10,7 +10,6 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -72,7 +71,7 @@ func (s *Store) InstanceIDs(ctx context.Context) ([]string, error) {
 
 	var ids []string
 	for key := range lister.Keys() {
-		if key != kvutil.VersionKey {
+		if key != kvstore.VersionKey {
 			ids = append(ids, key)
 		}
 	}

spinifex/domains/ec2/instance/uefistore/uefistore_test.go

@@ -11,7 +11,6 @@ import (
 	testutil "github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/instance/uefistore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -60,7 +59,7 @@ func TestOpen_CreatesBucketWithNoTTLAndStampsVersion(t *testing.T) {
 	assert.Equal(t, int64(1), status.History())
 	assert.Zero(t, status.TTL(), "a TTL would expire deletion obligations before they are settled")
 
-	v, err := kvutil.ReadVersion(t.Context(), kv)
+	v, err := kvstore.ReadVersion(t.Context(), kv)
 	require.NoError(t, err)
 	assert.Equal(t, uefistore.BucketVersion, v)
 }

spinifex/domains/ec2/inventory/inventory.go

@@ -11,7 +11,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
 	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -51,7 +51,7 @@ func NewENITagReader(js jetstream.JetStream, accountID string) func(ctx context.
 		if err != nil {
 			return nil, false, fmt.Errorf("open ENI bucket: %w", err)
 		}
-		entry, err := kv.Get(ctx, kvutil.AccountKey(accountID, eniID))
+		entry, err := kv.Get(ctx, kvstore.AccountKey(accountID, eniID))
 		if errors.Is(err, jetstream.ErrKeyNotFound) {
 			return nil, false, nil
 		}

spinifex/domains/ec2/inventory/inventory_test.go

@@ -6,7 +6,7 @@ import (
 
 	testutil "github.com/mulgadc/spinifex/internal/testkit"
 	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -25,13 +25,13 @@ func TestENITagReaderSeparatesGoneFromUnreadable(t *testing.T) {
 	_, _, err := read(t.Context(), "eni-1")
 	require.Error(t, err, "a missing ENI bucket is a read failure, not an absent ENI")
 
-	enis, err := kvutil.GetOrCreateBucket(t.Context(), js, ec2vpc.KVBucketENIs, 10)
+	enis, err := kvstore.GetOrCreateBucket(t.Context(), js, ec2vpc.KVBucketENIs, 10)
 	require.NoError(t, err)
 	good, err := json.Marshal(ec2vpc.ENIRecord{NetworkInterfaceId: "eni-good", Tags: map[string]string{"k": "v"}})
 	require.NoError(t, err)
-	_, err = enis.Put(t.Context(), kvutil.AccountKey("000000000000", "eni-good"), good)
+	_, err = enis.Put(t.Context(), kvstore.AccountKey("000000000000", "eni-good"), good)
 	require.NoError(t, err)
-	_, err = enis.Put(t.Context(), kvutil.AccountKey("000000000000", "eni-corrupt"), []byte("{not json"))
+	_, err = enis.Put(t.Context(), kvstore.AccountKey("000000000000", "eni-corrupt"), []byte("{not json"))
 	require.NoError(t, err)
 
 	tags, found, err := read(t.Context(), "eni-good")

spinifex/domains/ec2/launchtemplate/account_cleanup.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -33,7 +33,7 @@ type AccountCleanup struct {
 // Unreadable headers, their bodies and unrecognised entries are left for the
 // owner scan to report. A listing failure is returned and nothing is removed.
 func (s *LaunchTemplateServiceImpl) CleanupAccountResidue(ctx context.Context, accountID string, settledBefore time.Time) (*AccountCleanup, error) {
-	keys, err := kvutil.Keys(ctx, s.kv)
+	keys, err := kvstore.Keys(ctx, s.kv)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
 		return nil, fmt.Errorf("list launch-template keys: %w", err)
 	}
@@ -145,7 +145,7 @@ type AccountTemplateDeletes struct {
 // it cannot read or decode is left for the owner scan to report. A listing
 // failure is returned and nothing is deleted.
 func (s *LaunchTemplateServiceImpl) DeleteLiveTemplates(ctx context.Context, accountID string, referenced func(id, name string) string) (*AccountTemplateDeletes, error) {
-	keys, err := kvutil.Keys(ctx, s.kv)
+	keys, err := kvstore.Keys(ctx, s.kv)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
 		return nil, fmt.Errorf("list launch-template keys: %w", err)
 	}

spinifex/domains/ec2/launchtemplate/account_scan.go

@@ -9,7 +9,7 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -37,7 +37,7 @@ type AccountScan struct {
 // Headers and name-index entries are read; version bodies are classified from
 // their keys alone, so template data and user data are never loaded.
 func (s *LaunchTemplateServiceImpl) ScanAccount(ctx context.Context, accountID string) (*AccountScan, error) {
-	keys, err := kvutil.Keys(ctx, s.kv)
+	keys, err := kvstore.Keys(ctx, s.kv)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
 		return nil, fmt.Errorf("list launch-template keys: %w", err)
 	}

spinifex/domains/ec2/launchtemplate/account_scan_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/internal/testkit/readonlystore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -45,7 +45,7 @@ func TestScanAccountClassifiesLaunchTemplateRecords(t *testing.T) {
 	// A header removed alone, as a delete interrupted after it leaves the rest.
 	deleted := createWithUserData(t, svc, "deleted")
 	inner := svc.kv
-	require.NoError(t, inner.Delete(t.Context(), kvutil.AccountKey(testAccountID, deleted)))
+	require.NoError(t, inner.Delete(t.Context(), kvstore.AccountKey(testAccountID, deleted)))
 
 	// A create whose header write and clean-up both fail.
 	svc.kv = &faultyKV{KeyValue: inner,
@@ -59,16 +59,16 @@ func TestScanAccountClassifiesLaunchTemplateRecords(t *testing.T) {
 	require.Error(t, err)
 	svc.kv = inner
 
-	putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0badbadbadbadbad0"), "{not json")
+	putRaw(t, inner, kvstore.AccountKey(testAccountID, "lt-0badbadbadbadbad0"), "{not json")
-	putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0badbadbadbadbad0.v1"), `{"version_number":1}`)
+	putRaw(t, inner, kvstore.AccountKey(testAccountID, "lt-0badbadbadbadbad0.v1"), `{"version_number":1}`)
-	putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0mismatch00000000"), `{"launch_template_id":"lt-0other"}`)
+	putRaw(t, inner, kvstore.AccountKey(testAccountID, "lt-0mismatch00000000"), `{"launch_template_id":"lt-0other"}`)
-	putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0unreadable000000"), `{"launch_template_id":"lt-0unreadable000000"}`)
+	putRaw(t, inner, kvstore.AccountKey(testAccountID, "lt-0unreadable000000"), `{"launch_template_id":"lt-0unreadable000000"}`)
 	putRaw(t, inner, nameKey(testAccountID, "unreadable"), "lt-0unreadable000000")
-	putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0bad.vx"), "{}")
+	putRaw(t, inner, kvstore.AccountKey(testAccountID, "lt-0bad.vx"), "{}")
 
 	recorder := &readRecordingKV{KeyValue: readonlystore.NewKV(t, inner)}
 	svc.kv = &faultyKV{KeyValue: recorder, failGet: func(key string) bool {
-		return key == kvutil.AccountKey(testAccountID, "lt-0unreadable000000")
+		return key == kvstore.AccountKey(testAccountID, "lt-0unreadable000000")
 	}}
 	scan, err := svc.ScanAccount(context.Background(), testAccountID)
 	require.NoError(t, err)

spinifex/domains/ec2/launchtemplate/account_termination_current_behaviour_test.go

@@ -16,7 +16,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -159,7 +159,7 @@ func TestCurrentBehaviour_AccountTermination_LaunchTemplateStorageLayout(t *test
 
 	header := testAccountID + "." + id
 	name := testAccountID + ".name.776562283129" // hex of "web(1)"
-	assert.Equal(t, sortedKeys(header, header+".v1", header+".v2", name, kvutil.VersionKey), storedKeys(t, svc.kv))
+	assert.Equal(t, sortedKeys(header, header+".v1", header+".v2", name, kvstore.VersionKey), storedKeys(t, svc.kv))
 
 	entry, err := svc.kv.Get(t.Context(), name)
 	require.NoError(t, err)
@@ -223,13 +223,13 @@ func TestAccountTermination_DeleteLaunchTemplateKeepsTheTemplateWhenAStepFails(t
 			assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 
 			svc.kv = inner
-			assert.Equal(t, sortedKeys(append(tc.left(header, nameIdx), kvutil.VersionKey)...), storedKeys(t, inner))
+			assert.Equal(t, sortedKeys(append(tc.left(header, nameIdx), kvstore.VersionKey)...), storedKeys(t, inner))
 			assert.Equal(t, []string{id}, describedTemplateIDs(t, svc, testAccountID))
 
 			out, err := svc.DeleteLaunchTemplate(context.Background(), &ec2.DeleteLaunchTemplateInput{LaunchTemplateId: aws.String(id)}, testAccountID)
 			require.NoError(t, err)
 			assert.Equal(t, id, aws.StringValue(out.LaunchTemplate.LaunchTemplateId))
-			assert.Equal(t, []string{kvutil.VersionKey}, storedKeys(t, inner))
+			assert.Equal(t, []string{kvstore.VersionKey}, storedKeys(t, inner))
 			assert.Empty(t, describedTemplateIDs(t, svc, testAccountID))
 		})
 	}
@@ -260,7 +260,7 @@ func TestAccountTermination_DeleteLaunchTemplateRestartsWhenTheHeaderChanges(t *
 	_, err = svc.DeleteLaunchTemplate(context.Background(), &ec2.DeleteLaunchTemplateInput{LaunchTemplateId: aws.String(id)}, testAccountID)
 	require.NoError(t, err)
 	assert.True(t, raced)
-	assert.Equal(t, []string{kvutil.VersionKey}, storedKeys(t, inner))
+	assert.Equal(t, []string{kvstore.VersionKey}, storedKeys(t, inner))
 }
 
 // A create that fails after claiming the name removes what it wrote, so no
@@ -287,7 +287,7 @@ func TestAccountTermination_FailedCreateRemovesWhatItWrote(t *testing.T) {
 			assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 
 			svc.kv = inner
-			assert.Equal(t, []string{kvutil.VersionKey}, storedKeys(t, inner))
+			assert.Equal(t, []string{kvstore.VersionKey}, storedKeys(t, inner))
 			retried := createTemplate(t, svc, "web", "t3.micro")
 			assert.Equal(t, []string{aws.StringValue(retried.LaunchTemplateId)}, describedTemplateIDs(t, svc, testAccountID))
 		})
@@ -299,7 +299,7 @@ func TestAccountTermination_FailedCreateRemovesWhatItWrote(t *testing.T) {
 // failure is an internal error, never a missing template.
 func TestCurrentBehaviour_AccountTermination_DescribeLaunchTemplatesHidesUnreadableHeaders(t *testing.T) {
 	const brokenID = "lt-0badbadbadbadbad0"
-	brokenHeader := kvutil.AccountKey(testAccountID, brokenID)
+	brokenHeader := kvstore.AccountKey(testAccountID, brokenID)
 	tests := []struct {
 		name       string
 		seed       func(t *testing.T, svc *LaunchTemplateServiceImpl)
@@ -350,7 +350,7 @@ func TestAccountCleanup_KeepsTemplatesAndRemovesSettledResidue(t *testing.T) {
 	require.NoError(t, err)
 	orphanBody := versionKey(testAccountID, "lt-0orphan000000000", 1)
 	orphanName := nameKey(testAccountID, "gone")
-	brokenHeader := kvutil.AccountKey(testAccountID, "lt-0broken000000000")
+	brokenHeader := kvstore.AccountKey(testAccountID, "lt-0broken000000000")
 	brokenBody := versionKey(testAccountID, "lt-0broken000000000", 1)
 	brokenName := nameKey(testAccountID, "broken")
 	for key, value := range map[string]string{
@@ -384,7 +384,7 @@ func TestAccountCleanup_KeepsTemplatesAndRemovesSettledResidue(t *testing.T) {
 func TestAccountCleanup_InterruptedTemplateDeleteIsLeftToThePublicRetry(t *testing.T) {
 	svc := setupTestService(t)
 	id := createWithUserData(t, svc, "web")
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	kv := svc.kv
 	svc.kv = &faultyKV{KeyValue: kv, failDelete: func(key string) bool { return key == header }}
 	_, err := svc.DeleteLaunchTemplate(context.Background(), &ec2.DeleteLaunchTemplateInput{LaunchTemplateId: aws.String(id)}, testAccountID)

spinifex/domains/ec2/launchtemplate/service_impl.go

@@ -19,7 +19,7 @@ import (
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -64,7 +64,7 @@ func NewLaunchTemplateServiceImplWithNATS(ctx context.Context, cfg *config.Confi
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketLaunchTemplates, 10)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketLaunchTemplates, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketLaunchTemplates, err)
 	}
@@ -87,24 +87,24 @@ func NewLaunchTemplateServiceImplWithNATS(ctx context.Context, cfg *config.Confi
 // Launch template names allow characters such as parentheses that are not valid
 // in NATS KV keys, so the name must not be used in its raw form here.
 func nameKey(accountID, name string) string {
-	return kvutil.AccountKey(accountID, "name."+hex.EncodeToString([]byte(name)))
+	return kvstore.AccountKey(accountID, "name."+hex.EncodeToString([]byte(name)))
 }
 
 // versionKey returns a version-body key: account.lt-<id>.v<n>.
 func versionKey(accountID, ltID string, n int64) string {
-	return kvutil.AccountKey(accountID, ltID+".v"+strconv.FormatInt(n, 10))
+	return kvstore.AccountKey(accountID, ltID+".v"+strconv.FormatInt(n, 10))
 }
 
 // versionPrefix returns the scan prefix for a template's version bodies.
 func versionPrefix(accountID, ltID string) string {
-	return kvutil.AccountKey(accountID, ltID+".v")
+	return kvstore.AccountKey(accountID, ltID+".v")
 }
 
 // --- record load helpers ---
 
 // getHeaderByID reads a header and its KV entry (for CAS) by launch template id.
 func (s *LaunchTemplateServiceImpl) getHeaderByID(ctx context.Context, accountID, ltID string) (*LaunchTemplateHeader, jetstream.KeyValueEntry, error) {
-	entry, err := s.kv.Get(ctx, kvutil.AccountKey(accountID, ltID))
+	entry, err := s.kv.Get(ctx, kvstore.AccountKey(accountID, ltID))
 	if err != nil {
 		if errors.Is(err, jetstream.ErrKeyNotFound) {
 			return nil, nil, errors.New(awserrors.ErrorInvalidLaunchTemplateIdNotFound)
@@ -386,7 +386,7 @@ func (s *LaunchTemplateServiceImpl) claimName(ctx context.Context, accountID, na
 	// header means the name is taken; any other read error (transient fault, or a
 	// concurrent in-flight create whose header is not yet written) fails closed so
 	// the name is never stolen from a possibly-live template.
-	_, herr := s.kv.Get(ctx, kvutil.AccountKey(accountID, string(entry.Value())))
+	_, herr := s.kv.Get(ctx, kvstore.AccountKey(accountID, string(entry.Value())))
 	switch {
 	case herr == nil:
 		return errNameInUse()
@@ -404,7 +404,7 @@ func (s *LaunchTemplateServiceImpl) putHeader(ctx context.Context, accountID str
 	if err != nil {
 		return errors.New(awserrors.ErrorServerInternal)
 	}
-	if _, err := s.kv.Put(ctx, kvutil.AccountKey(accountID, h.LaunchTemplateId), data); err != nil {
+	if _, err := s.kv.Put(ctx, kvstore.AccountKey(accountID, h.LaunchTemplateId), data); err != nil {
 		return errors.New(awserrors.ErrorServerInternal)
 	}
 	return nil
@@ -510,13 +510,13 @@ func (s *LaunchTemplateServiceImpl) commitVersion(ctx context.Context, accountID
 // re-reading on a conflict. A non-nil check vets each re-read header and
 // stops the commit with its error.
 func (s *LaunchTemplateServiceImpl) commitHeader(ctx context.Context, op, accountID, ltID string, entry jetstream.KeyValueEntry, check func(*LaunchTemplateHeader) error) error {
-	key := kvutil.AccountKey(accountID, ltID)
+	key := kvstore.AccountKey(accountID, ltID)
 	for range maxVersionRetries {
 		_, err := s.kv.Update(ctx, key, entry.Value(), entry.Revision())
 		if err == nil {
 			return nil
 		}
-		if !kvutil.IsRevisionConflict(err) {
+		if !kvstore.IsRevisionConflict(err) {
 			slog.ErrorContext(ctx, op+": header commit failed", "launchTemplateId", ltID, "err", err)
 			return err
 		}
@@ -590,7 +590,7 @@ func (s *LaunchTemplateServiceImpl) ModifyLaunchTemplate(ctx context.Context, in
 // header deleted meanwhile is NotFound for the identifier the request used.
 func (s *LaunchTemplateServiceImpl) setDefaultVersion(ctx context.Context, accountID string, input *ec2.ModifyLaunchTemplateInput, header *LaunchTemplateHeader, entry jetstream.KeyValueEntry, sel string) (*LaunchTemplateHeader, error) {
 	notFound := fmt.Sprintf("The launch template version %s is not found for the specified launch template.", sel)
-	key := kvutil.AccountKey(accountID, header.LaunchTemplateId)
+	key := kvstore.AccountKey(accountID, header.LaunchTemplateId)
 	for range maxVersionRetries {
 		n, err := s.resolveVersionNumber(ctx, accountID, header, sel)
 		if err != nil {
@@ -611,7 +611,7 @@ func (s *LaunchTemplateServiceImpl) setDefaultVersion(ctx context.Context, accou
 		if err == nil {
 			return header, nil
 		}
-		if !kvutil.IsRevisionConflict(err) {
+		if !kvstore.IsRevisionConflict(err) {
 			slog.ErrorContext(ctx, "ModifyLaunchTemplate: header update failed", "launchTemplateId", header.LaunchTemplateId, "err", err)
 			return nil, errors.New(awserrors.ErrorServerInternal)
 		}
@@ -662,7 +662,7 @@ const maxHeaderDeleteAttempts = 3
 // through the public delete; a header changed in between restarts the sweep.
 func (s *LaunchTemplateServiceImpl) deleteTemplate(ctx context.Context, accountID string, header *LaunchTemplateHeader, entry jetstream.KeyValueEntry) error {
 	ltID := header.LaunchTemplateId
-	headerKey := kvutil.AccountKey(accountID, ltID)
+	headerKey := kvstore.AccountKey(accountID, ltID)
 	for attempt := 1; ; attempt++ {
 		if err := s.deleteVersionBodies(ctx, accountID, ltID); err != nil {
 			return err
@@ -824,7 +824,7 @@ func (s *LaunchTemplateServiceImpl) commitVersionDeletes(ctx context.Context, ac
 	}
 	// A template delete that removed the header before the restore has already
 	// swept its bodies, so the restored ones would be left headerless.
-	if _, gerr := s.kv.Get(ctx, kvutil.AccountKey(accountID, ltID)); errors.Is(gerr, jetstream.ErrKeyNotFound) {
+	if _, gerr := s.kv.Get(ctx, kvstore.AccountKey(accountID, ltID)); errors.Is(gerr, jetstream.ErrKeyNotFound) {
 		for n := range deleted {
 			if derr := s.kv.Delete(ctx, versionKey(accountID, ltID, n)); derr != nil {
 				slog.ErrorContext(ctx, "DeleteLaunchTemplateVersions: failed to remove a restored version", "launchTemplateId", ltID, "version", n, "err", derr)

spinifex/domains/ec2/launchtemplate/service_impl_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -116,7 +116,7 @@ func TestCreateLaunchTemplate_OrphanNameReclaim(t *testing.T) {
 	lt := createTemplate(t, svc, "orphan", "t3.micro")
 
 	// Simulate a crash after the name claim but with the header lost.
-	require.NoError(t, svc.kv.Delete(t.Context(), kvutil.AccountKey(testAccountID, aws.StringValue(lt.LaunchTemplateId))))
+	require.NoError(t, svc.kv.Delete(t.Context(), kvstore.AccountKey(testAccountID, aws.StringValue(lt.LaunchTemplateId))))
 
 	// The name index still points at the now-orphaned id; create must reclaim it.
 	out, err := svc.CreateLaunchTemplate(context.Background(), &ec2.CreateLaunchTemplateInput{

spinifex/domains/ec2/launchtemplate/template_lifecycle_test.go

@@ -10,7 +10,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -38,7 +38,7 @@ func TestAccountTermination_VersionCreateRacingADeleteLeavesNoBody(t *testing.T)
 		t.Run(pause, func(t *testing.T) {
 			svc := setupTestService(t)
 			id := createWithUserData(t, svc, "web")
-			header := kvutil.AccountKey(testAccountID, id)
+			header := kvstore.AccountKey(testAccountID, id)
 			raced := false
 			race := func(key string, at string) {
 				if raced || pause != at {
@@ -71,7 +71,7 @@ func TestAccountTermination_DeleteRacingACommittedVersionRemovesIt(t *testing.T)
 	svc := setupTestService(t)
 	id := createWithUserData(t, svc, "web")
 	inner := svc.kv
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	raced := false
 	svc.kv = &faultyKV{KeyValue: inner, beforeDelete: func(key string) {
 		if key != header || raced {
@@ -95,7 +95,7 @@ func TestAccountTermination_FailedVersionCommitRemovesTheBody(t *testing.T) {
 	svc := setupTestService(t)
 	id := createWithUserData(t, svc, "web")
 	before := accountKeys(t, svc.kv, testAccountID)
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	svc.kv = &faultyKV{KeyValue: svc.kv, failUpdate: func(key string) bool { return key == header }}
 
 	_, err := svc.CreateLaunchTemplateVersion(context.Background(), newVersionInput(id), testAccountID)
@@ -110,7 +110,7 @@ func TestAccountTermination_VersionCommitKeepsAConcurrentHeaderChange(t *testing
 	svc := setupTestService(t)
 	id := createWithUserData(t, svc, "web")
 	inner := svc.kv
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	raced := false
 	svc.kv = &faultyKV{KeyValue: inner, beforeUpdate: func(key string) {
 		if key != header || raced {
@@ -215,10 +215,10 @@ func TestAccountTemplateReferences_FailsClosed(t *testing.T) {
 			require.NoError(t, err)
 		},
 		"header unreadable": func(t *testing.T, svc *LaunchTemplateServiceImpl, id string) {
-			svc.kv = &faultyKV{KeyValue: svc.kv, failGet: func(key string) bool { return key == kvutil.AccountKey(testAccountID, id) }}
+			svc.kv = &faultyKV{KeyValue: svc.kv, failGet: func(key string) bool { return key == kvstore.AccountKey(testAccountID, id) }}
 		},
 		"header undecodable": func(t *testing.T, svc *LaunchTemplateServiceImpl, _ string) {
-			_, err := svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "lt-0broken000000000"), []byte("{not json"))
+			_, err := svc.kv.Put(t.Context(), kvstore.AccountKey(testAccountID, "lt-0broken000000000"), []byte("{not json"))
 			require.NoError(t, err)
 		},
 	}
@@ -241,7 +241,7 @@ func TestModifyLaunchTemplate_RacingADeleteIsNotFound(t *testing.T) {
 		t.Run(map[bool]string{false: "by-id", true: "by-name"}[byName], func(t *testing.T) {
 			svc := setupTestService(t)
 			id := createWithUserData(t, svc, "web")
-			header := kvutil.AccountKey(testAccountID, id)
+			header := kvstore.AccountKey(testAccountID, id)
 			raced := false
 			svc.kv = &faultyKV{KeyValue: svc.kv, beforeUpdate: func(key string) {
 				if key == header && !raced {
@@ -275,7 +275,7 @@ func TestModifyLaunchTemplate_RetriesAConcurrentHeaderChange(t *testing.T) {
 	_, err := svc.CreateLaunchTemplateVersion(context.Background(), newVersionInput(id), testAccountID) // v3
 	require.NoError(t, err)
 	inner := svc.kv
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	raced := false
 	svc.kv = &faultyKV{KeyValue: inner, beforeUpdate: func(key string) {
 		if key != header || raced {
@@ -306,7 +306,7 @@ func TestModifyLaunchTemplate_ConflictRetriesAreBounded(t *testing.T) {
 	svc := setupTestService(t)
 	id := createWithUserData(t, svc, "web")
 	inner := svc.kv
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	attempts := 0
 	svc.kv = &faultyKV{KeyValue: inner, beforeUpdate: func(key string) {
 		if key != header {
@@ -333,7 +333,7 @@ func TestModifyLaunchTemplate_ConflictReReadFailureIsInternal(t *testing.T) {
 	svc := setupTestService(t)
 	id := createWithUserData(t, svc, "web")
 	inner := svc.kv
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	conflicted := false
 	svc.kv = &faultyKV{
 		KeyValue: inner,
@@ -410,7 +410,7 @@ func TestDefaultChangeRacingAVersionDeleteFailsNotFound(t *testing.T) {
 	svc := setupTestService(t)
 	id := createWithUserData(t, svc, "web")
 	inner := svc.kv
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	raced := false
 	svc.kv = &faultyKV{KeyValue: inner, beforeUpdate: func(key string) {
 		if key != header || raced {
@@ -460,7 +460,7 @@ func TestFailedVersionDeleteCommitRestoresTheBodies(t *testing.T) {
 	svc := setupTestService(t)
 	id := createWithUserData(t, svc, "web")
 	before := accountKeys(t, svc.kv, testAccountID)
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	svc.kv = &faultyKV{KeyValue: svc.kv, failUpdate: func(key string) bool { return key == header }}
 
 	_, err := deleteVersions(svc, id, "2")
@@ -474,7 +474,7 @@ func TestFailedVersionDeleteCommitRestoresTheBodies(t *testing.T) {
 func TestVersionDeleteRacingATemplateDeleteLeavesNothing(t *testing.T) {
 	svc := setupTestService(t)
 	id := createWithUserData(t, svc, "web")
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	raced := false
 	svc.kv = &faultyKV{KeyValue: svc.kv, beforeUpdate: func(key string) {
 		if key != header || raced {
@@ -531,7 +531,7 @@ func TestDefaultChangeRetriesAfterAConcurrentVersionCreate(t *testing.T) {
 	svc := setupTestService(t)
 	id := createWithUserData(t, svc, "web")
 	inner := svc.kv
-	header := kvutil.AccountKey(testAccountID, id)
+	header := kvstore.AccountKey(testAccountID, id)
 	raced := false
 	svc.kv = &faultyKV{KeyValue: inner, beforeUpdate: func(key string) {
 		if key != header || raced {

spinifex/domains/ec2/launchtemplate/template_references.go

@@ -8,7 +8,7 @@ import (
 	"strings"
 
 	"github.com/aws/aws-sdk-go/aws"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -27,7 +27,7 @@ type TemplateReferences struct {
 // header or body it cannot read or decode, is an error, because the unread
 // record may name the resource. Headerless bodies resolve from no launch.
 func (s *LaunchTemplateServiceImpl) AccountTemplateReferences(ctx context.Context, accountID string) (*TemplateReferences, error) {
-	keys, err := kvutil.Keys(ctx, s.kv)
+	keys, err := kvstore.Keys(ctx, s.kv)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
 		return nil, fmt.Errorf("list launch-template keys: %w", err)
 	}

spinifex/domains/ec2/natgw/lifecycle_invariants_test.go

@@ -10,7 +10,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	ec2eip "github.com/mulgadc/spinifex/spinifex/domains/ec2/eip"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -42,7 +42,7 @@ func TestRLC3_NatGatewayDeletesWhileRouted(t *testing.T) {
 		natgwID := createTestNatGateway(t, svc)
 
 		testutil.SeedKV(t, js, kvBucketRouteTables, map[string][]byte{
-			kvutil.AccountKey(testAccountID, "rtb-routed"): fmt.Appendf(nil,
+			kvstore.AccountKey(testAccountID, "rtb-routed"): fmt.Appendf(nil,
 				`{"route_table_id":"rtb-routed","vpc_id":"vpc-test1","routes":[{"destination_cidr_block":"0.0.0.0/0","nat_gateway_id":%q}]}`, natgwID),
 		})
 
@@ -93,7 +93,7 @@ func createTestNatGateway(t *testing.T, svc *NatGatewayServiceImpl) string {
 
 func readEIP(t *testing.T, svc *NatGatewayServiceImpl, allocID string) ec2eip.EIPRecord {
 	t.Helper()
-	entry, err := svc.eipKV.Get(t.Context(), kvutil.AccountKey(testAccountID, allocID))
+	entry, err := svc.eipKV.Get(t.Context(), kvstore.AccountKey(testAccountID, allocID))
 	require.NoError(t, err)
 	var eip ec2eip.EIPRecord
 	require.NoError(t, json.Unmarshal(entry.Value(), &eip))

spinifex/domains/ec2/natgw/service_impl.go

@@ -20,7 +20,7 @@ import (
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -45,7 +45,7 @@ func NewNatGatewayServiceImplWithNATS(ctx context.Context, natsConn *nats.Conn)
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	natgwKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketNatGateways, 10)
+	natgwKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketNatGateways, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketNatGateways, err)
 	}
@@ -61,17 +61,17 @@ func NewNatGatewayServiceImplWithNATS(ctx context.Context, natsConn *nats.Conn)
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketDeletedNatGateways, err)
 	}
 
-	eipKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
+	eipKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get EIP KV bucket: %w", err)
 	}
 
-	subnetKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
+	subnetKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get subnet KV bucket: %w", err)
 	}
 
-	vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
+	vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
 	}
@@ -92,7 +92,7 @@ func NewNatGatewayServiceImplWithNATS(ctx context.Context, natsConn *nats.Conn)
 // for as long as a client may still ask about it.
 func getOrCreateDeletedBucket(ctx context.Context, js jetstream.KeyValueManager) (jetstream.KeyValue, error) {
 	// Terraform polls DescribeNatGateways after delete and expects state=deleted.
-	return kvutil.GetOrCreateBucketWithOptions(ctx, js, kvutil.BucketOptions{
+	return kvstore.GetOrCreateBucketWithOptions(ctx, js, kvstore.BucketOptions{
 		Name:        KVBucketDeletedNatGateways,
 		Description: "Deleted NAT Gateways (auto-expire after 1 hour)",
 		History:     1,
@@ -113,7 +113,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
 	allocID := *input.AllocationId
 
 	// Validate subnet exists and get its VPC
-	subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetID))
+	subnetEntry, err := s.subnetKV.Get(ctx, kvstore.AccountKey(accountID, subnetID))
 	if err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
 	}
@@ -123,7 +123,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
 	}
 
 	// Validate EIP exists and is not already associated
-	eipEntry, err := s.eipKV.Get(ctx, kvutil.AccountKey(accountID, allocID))
+	eipEntry, err := s.eipKV.Get(ctx, kvstore.AccountKey(accountID, allocID))
 	if err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidAllocationIDNotFound)
 	}
@@ -152,7 +152,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
 	if err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
-	if _, err := s.natgwKV.Put(ctx, kvutil.AccountKey(accountID, natgwID), data); err != nil {
+	if _, err := s.natgwKV.Put(ctx, kvstore.AccountKey(accountID, natgwID), data); err != nil {
 		slog.ErrorContext(ctx, "Failed to store NAT Gateway", "natGatewayId", natgwID, "err", err)
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
@@ -165,7 +165,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
 	eipRecord.VpcId = subnetRecord.VpcId
 	eipRecord.State = "associated"
 	if eipData, err := json.Marshal(eipRecord); err == nil {
-		if _, err := s.eipKV.Update(ctx, kvutil.AccountKey(accountID, allocID), eipData, eipEntry.Revision()); err != nil {
+		if _, err := s.eipKV.Update(ctx, kvstore.AccountKey(accountID, allocID), eipData, eipEntry.Revision()); err != nil {
 			slog.WarnContext(ctx, "CreateNatGateway: failed to mark EIP associated", "natGatewayId", natgwID, "allocationId", allocID, "err", err)
 		}
 	}
@@ -185,7 +185,7 @@ func (s *NatGatewayServiceImpl) DeleteNatGateway(ctx context.Context, input *ec2
 	}
 
 	natgwID := *input.NatGatewayId
-	key := kvutil.AccountKey(accountID, natgwID)
+	key := kvstore.AccountKey(accountID, natgwID)
 
 	entry, err := s.natgwKV.Get(ctx, key)
 	if err != nil {
@@ -255,7 +255,7 @@ func (s *NatGatewayServiceImpl) publishDeleteEventsForNatGateway(ctx context.Con
 		slog.WarnContext(ctx, "DeleteNatGateway: JetStream setup failed, SNAT teardown deferred to route delete", "natGatewayId", record.NatGatewayId, "err", err)
 		return
 	}
-	rtbKV, err := kvutil.GetOrCreateBucket(ctx, js, kvBucketRouteTables, 10)
+	rtbKV, err := kvstore.GetOrCreateBucket(ctx, js, kvBucketRouteTables, 10)
 	if err != nil {
 		slog.WarnContext(ctx, "DeleteNatGateway: route-table scan failed, SNAT teardown deferred to route delete", "natGatewayId", record.NatGatewayId, "err", err)
 		return
@@ -266,7 +266,7 @@ func (s *NatGatewayServiceImpl) publishDeleteEventsForNatGateway(ctx context.Con
 	}
 	prefix := accountID + "."
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := rtbKV.Get(ctx, key)
@@ -302,7 +302,7 @@ func (s *NatGatewayServiceImpl) publishDeleteEventsForNatGateway(ctx context.Con
 			if assoc.SubnetId == "" {
 				continue
 			}
-			subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, assoc.SubnetId))
+			subnetEntry, err := s.subnetKV.Get(ctx, kvstore.AccountKey(accountID, assoc.SubnetId))
 			if err != nil {
 				continue
 			}
@@ -322,7 +322,7 @@ func (s *NatGatewayServiceImpl) disassociateEIP(ctx context.Context, record *Nat
 	if record.AllocationId == "" {
 		return
 	}
-	key := kvutil.AccountKey(accountID, record.AllocationId)
+	key := kvstore.AccountKey(accountID, record.AllocationId)
 	entry, err := s.eipKV.Get(ctx, key)
 	if err != nil {
 		if !errors.Is(err, jetstream.ErrKeyNotFound) {
@@ -380,7 +380,7 @@ func (s *NatGatewayServiceImpl) DescribeNatGateways(ctx context.Context, input *
 	foundIDs := make(map[string]bool)
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 
@@ -415,7 +415,7 @@ func (s *NatGatewayServiceImpl) DescribeNatGateways(ctx context.Context, input *
 		if foundIDs[id] {
 			continue
 		}
-		key := kvutil.AccountKey(accountID, id)
+		key := kvstore.AccountKey(accountID, id)
 		entry, err := s.deletedNatgwKV.Get(ctx, key)
 		if err != nil {
 			if errors.Is(err, jetstream.ErrKeyNotFound) {

spinifex/domains/ec2/natgw/service_impl_test.go

@@ -11,7 +11,7 @@ import (
 	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -31,18 +31,18 @@ func setupTestServiceJS(t *testing.T) (*NatGatewayServiceImpl, jetstream.JetStre
 
 	// Seed VPC KV
 	testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
-		kvutil.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
+		kvstore.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
 	})
 
 	// Seed subnet KV (public subnet)
 	testutil.SeedKV(t, js, ec2vpc.KVBucketSubnets, map[string][]byte{
-		kvutil.AccountKey(testAccountID, "subnet-pub1"): []byte(`{"subnet_id":"subnet-pub1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available","map_public_ip_on_launch":true}`),
+		kvstore.AccountKey(testAccountID, "subnet-pub1"): []byte(`{"subnet_id":"subnet-pub1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available","map_public_ip_on_launch":true}`),
 	})
 
 	// Seed EIP KV
 	testutil.SeedKV(t, js, ec2eip.KVBucketEIPs, map[string][]byte{
-		kvutil.AccountKey(testAccountID, "eipalloc-test1"): []byte(`{"allocation_id":"eipalloc-test1","public_ip":"203.0.113.50","state":"allocated"}`),
+		kvstore.AccountKey(testAccountID, "eipalloc-test1"): []byte(`{"allocation_id":"eipalloc-test1","public_ip":"203.0.113.50","state":"allocated"}`),
-		kvutil.AccountKey(testAccountID, "eipalloc-used"):  []byte(`{"allocation_id":"eipalloc-used","public_ip":"203.0.113.51","state":"associated","association_id":"eipassoc-xxx"}`),
+		kvstore.AccountKey(testAccountID, "eipalloc-used"):  []byte(`{"allocation_id":"eipalloc-used","public_ip":"203.0.113.51","state":"associated","association_id":"eipassoc-xxx"}`),
 	})
 
 	svc, err := NewNatGatewayServiceImplWithNATS(t.Context(), nc)
@@ -170,7 +170,7 @@ func TestDeleteNatGateway(t *testing.T) {
 	assert.Equal(t, natgwID, *deleteOut.NatGatewayId)
 
 	// Should be gone
-	_, err = svc.natgwKV.Get(t.Context(), kvutil.AccountKey(testAccountID, natgwID))
+	_, err = svc.natgwKV.Get(t.Context(), kvstore.AccountKey(testAccountID, natgwID))
 	assert.ErrorIs(t, err, jetstream.ErrKeyNotFound)
 }
 

spinifex/domains/ec2/placementgroup/account_cleanup.go

@@ -8,7 +8,7 @@ import (
 	"log/slog"
 	"strings"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -39,7 +39,7 @@ type AccountCleanup struct {
 // referenced returns a reason is kept; unreadable records are left for the owner scan to report. A
 // listing failure is returned and nothing is deleted.
 func (s *PlacementGroupServiceImpl) CleanupAccount(ctx context.Context, accountID string, referenced func(groupID, groupName string) string) (*AccountCleanup, error) {
-	keys, err := kvutil.Keys(ctx, s.kv)
+	keys, err := kvstore.Keys(ctx, s.kv)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
 		return nil, fmt.Errorf("list placement-group keys: %w", err)
 	}
@@ -47,7 +47,7 @@ func (s *PlacementGroupServiceImpl) CleanupAccount(ctx context.Context, accountI
 	prefix := accountID + "."
 	out := &AccountCleanup{Referenced: map[string]string{}}
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.kv.Get(ctx, key)

spinifex/domains/ec2/placementgroup/account_scan.go

@@ -8,7 +8,7 @@ import (
 	"log/slog"
 	"strings"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -34,7 +34,7 @@ type AccountScan struct {
 // A listing failure is returned as an error; per-record read failures are
 // counted. The bucket's schema-version key belongs to no account.
 func (s *PlacementGroupServiceImpl) ScanAccount(ctx context.Context, accountID string) (*AccountScan, error) {
-	keys, err := kvutil.Keys(ctx, s.kv)
+	keys, err := kvstore.Keys(ctx, s.kv)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
 		return nil, fmt.Errorf("list placement-group keys: %w", err)
 	}
@@ -42,7 +42,7 @@ func (s *PlacementGroupServiceImpl) ScanAccount(ctx context.Context, accountID s
 	prefix := accountID + "."
 	scan := &AccountScan{}
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.kv.Get(ctx, key)

spinifex/domains/ec2/placementgroup/account_scan_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/internal/testkit/readonlystore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -42,14 +42,14 @@ func TestScanAccountClassifiesPlacementGroupRecords(t *testing.T) {
 		GroupName: "held", NodeInstances: map[string][]string{"node-2": {"i-0bbb", "i-0aaa"}, "node-3": {"i-0aaa"}},
 	}, testAccountID)
 	require.NoError(t, err)
-	_, err = svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "undecodable"), []byte("{not json"))
+	_, err = svc.kv.Put(t.Context(), kvstore.AccountKey(testAccountID, "undecodable"), []byte("{not json"))
 	require.NoError(t, err)
-	_, err = svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "anonymous"), []byte(`{"group_name":"anonymous"}`))
+	_, err = svc.kv.Put(t.Context(), kvstore.AccountKey(testAccountID, "anonymous"), []byte(`{"group_name":"anonymous"}`))
 	require.NoError(t, err)
 	createTestGroup(t, svc, "unreadable", ec2.PlacementStrategySpread)
-	require.Contains(t, bucketKeys(t, svc.kv), kvutil.VersionKey)
+	require.Contains(t, bucketKeys(t, svc.kv), kvstore.VersionKey)
 
-	svc.kv = &hookedKV{KeyValue: readonlystore.NewKV(t, svc.kv), failGet: kvutil.AccountKey(testAccountID, "unreadable")}
+	svc.kv = &hookedKV{KeyValue: readonlystore.NewKV(t, svc.kv), failGet: kvstore.AccountKey(testAccountID, "unreadable")}
 	scan, err := svc.ScanAccount(context.Background(), testAccountID)
 	require.NoError(t, err)
 

spinifex/domains/ec2/placementgroup/account_termination_current_behaviour_test.go

@@ -14,7 +14,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -84,7 +84,7 @@ func TestCurrentBehaviour_AccountTermination_PlacementGroupStorageLayout(t *test
 	}, testAccountID)
 	require.NoError(t, err)
 
-	assert.Equal(t, []string{testAccountID + ".spread-a", kvutil.VersionKey}, bucketKeys(t, svc.kv))
+	assert.Equal(t, []string{testAccountID + ".spread-a", kvstore.VersionKey}, bucketKeys(t, svc.kv))
 
 	entry, err := svc.kv.Get(t.Context(), testAccountID+".spread-a")
 	require.NoError(t, err)
@@ -100,7 +100,7 @@ func TestCurrentBehaviour_AccountTermination_PlacementGroupStorageLayout(t *test
 // a record cannot be removed through DeletePlacementGroup either, and a read
 // failure is an internal error, never an unknown group.
 func TestCurrentBehaviour_AccountTermination_DescribePlacementGroupsHidesUnreadableRecords(t *testing.T) {
-	brokenKey := kvutil.AccountKey(testAccountID, "broken")
+	brokenKey := kvstore.AccountKey(testAccountID, "broken")
 	tests := []struct {
 		name       string
 		seed       func(t *testing.T, svc *PlacementGroupServiceImpl)
@@ -139,7 +139,7 @@ func TestCurrentBehaviour_AccountTermination_DescribePlacementGroupsHidesUnreada
 func TestAccountTermination_DeletePlacementGroupLosesToAConcurrentMember(t *testing.T) {
 	svc := setupTestService(t)
 	createTestGroup(t, svc, "cluster-a", ec2.PlacementStrategyCluster)
-	key := kvutil.AccountKey(testAccountID, "cluster-a")
+	key := kvstore.AccountKey(testAccountID, "cluster-a")
 
 	launcher := &PlacementGroupServiceImpl{kv: svc.kv, region: svc.region}
 	var reserveErr, finalizeErr error
@@ -222,7 +222,7 @@ func TestAccountCleanup_DeletesEmptyGroupsAndBlocksOnMembers(t *testing.T) {
 	empty := createTestGroup(t, svc, "empty", ec2.PlacementStrategySpread)
 	held := createTestGroup(t, svc, "held", ec2.PlacementStrategyCluster)
 	require.NoError(t, addMember(t, svc, "held", "i-0123456789abcdef0"))
-	_, err := svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "broken"), []byte("{not json"))
+	_, err := svc.kv.Put(t.Context(), kvstore.AccountKey(testAccountID, "broken"), []byte("{not json"))
 	require.NoError(t, err)
 	other := "999988887777"
 	_, err = svc.CreatePlacementGroup(context.Background(), &ec2.CreatePlacementGroupInput{
@@ -238,7 +238,7 @@ func TestAccountCleanup_DeletesEmptyGroupsAndBlocksOnMembers(t *testing.T) {
 
 	assert.Equal(t, []string{"held"}, describedGroupNames(t, svc, testAccountID))
 	assert.Equal(t, []string{"theirs"}, describedGroupNames(t, svc, other))
-	_, err = svc.kv.Get(t.Context(), kvutil.AccountKey(testAccountID, "broken"))
+	_, err = svc.kv.Get(t.Context(), kvstore.AccountKey(testAccountID, "broken"))
 	require.NoError(t, err, "an unreadable record is never deleted")
 }
 
@@ -247,7 +247,7 @@ func TestAccountCleanup_DeletesEmptyGroupsAndBlocksOnMembers(t *testing.T) {
 func TestAccountCleanup_ConcurrentMemberBlocksTheDelete(t *testing.T) {
 	svc := setupTestService(t)
 	group := createTestGroup(t, svc, "cluster-a", ec2.PlacementStrategyCluster)
-	key := kvutil.AccountKey(testAccountID, "cluster-a")
+	key := kvstore.AccountKey(testAccountID, "cluster-a")
 	launcher := &PlacementGroupServiceImpl{kv: svc.kv, region: svc.region}
 	var memberErr error
 	// Read 1 is the cleanup's listing read; read 2 is the delete's check.

spinifex/domains/ec2/placementgroup/service_impl.go

@@ -17,7 +17,7 @@ import (
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -87,7 +87,7 @@ func NewPlacementGroupServiceImplWithNATS(ctx context.Context, cfg *config.Confi
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketPlacementGroups, 10)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketPlacementGroups, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketPlacementGroups, err)
 	}
@@ -137,7 +137,7 @@ func (s *PlacementGroupServiceImpl) CreatePlacementGroup(ctx context.Context, in
 	}
 
 	groupName := *input.GroupName
-	key := kvutil.AccountKey(accountID, groupName)
+	key := kvstore.AccountKey(accountID, groupName)
 	groupID := awsidentifiers.GenerateResourceID("pg")
 
 	record := PlacementGroupRecord{
@@ -175,7 +175,7 @@ func (s *PlacementGroupServiceImpl) DeletePlacementGroup(ctx context.Context, in
 	}
 
 	groupName := *input.GroupName
-	if _, err := s.deleteGroup(ctx, accountID, kvutil.AccountKey(accountID, groupName)); err != nil {
+	if _, err := s.deleteGroup(ctx, accountID, kvstore.AccountKey(accountID, groupName)); err != nil {
 		if errors.Is(err, errPGAbsent) {
 			// Lower case, unlike DescribePlacementGroups' message for the same code.
 			return nil, awserrors.Errorf(awserrors.ErrorInvalidPlacementGroupUnknown, "The placement group '%s' is unknown.", groupName)
@@ -194,7 +194,7 @@ func (s *PlacementGroupServiceImpl) DeletePlacementGroup(ctx context.Context, in
 // The returned count is the members seen by the last check.
 func (s *PlacementGroupServiceImpl) deleteGroup(ctx context.Context, accountID, key string) (int, error) {
 	members := 0
-	record, err := kvutil.DeleteIf(ctx, s.kv, key, casConfig(), func(record *PlacementGroupRecord) error {
+	record, err := kvstore.DeleteIf(ctx, s.kv, key, casConfig(), func(record *PlacementGroupRecord) error {
 		members = memberCount(record.NodeInstances)
 		if members > 0 {
 			return errPGInUse
@@ -238,7 +238,7 @@ func (s *PlacementGroupServiceImpl) DescribePlacementGroups(ctx context.Context,
 
 	var groups []*ec2.PlacementGroup
 	for _, k := range keys {
-		if k == kvutil.VersionKey {
+		if k == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(k, prefix) {
@@ -341,7 +341,7 @@ func pgMatchesFilters(record *PlacementGroupRecord, filters map[string][]string)
 
 const maxCASRetries = 5
 
-// errPGAbsent and errPGContended mark the two kvutil outcomes that map to
+// errPGAbsent and errPGContended mark the two kvstore outcomes that map to
 // specific AWS errors rather than a generic internal failure, and neither
 // escapes this file. errPGInUse is the AWS error itself.
 var (
@@ -350,8 +350,8 @@ var (
 	errPGInUse     = errors.New(awserrors.ErrorInvalidPlacementGroupInUse)
 )
 
-// mapCASError turns a kvutil failure into the AWS error the caller returns.
+// mapCASError turns a kvstore failure into the AWS error the caller returns.
-// A non-kvutil error came from mutate and is already an AWS code.
+// A non-kvstore error came from mutate and is already an AWS code.
 func mapCASError(ctx context.Context, op, groupName, accountID string, err error) error {
 	switch {
 	case errors.Is(err, errPGAbsent):
@@ -359,16 +359,16 @@ func mapCASError(ctx context.Context, op, groupName, accountID string, err error
 	case errors.Is(err, errPGContended):
 		slog.ErrorContext(ctx, "Placement group CAS retries exhausted under contention",
 			"op", op, "groupName", groupName, "accountID", accountID)
-	case errors.Is(err, kvutil.ErrRead):
+	case errors.Is(err, kvstore.ErrRead):
 		slog.ErrorContext(ctx, "Failed to read placement group from KV",
 			"op", op, "groupName", groupName, "accountID", accountID, "err", err)
-	case errors.Is(err, kvutil.ErrDecode):
+	case errors.Is(err, kvstore.ErrDecode):
 		slog.ErrorContext(ctx, "Corrupt placement group record in KV",
 			"op", op, "groupName", groupName, "accountID", accountID, "err", err)
-	case errors.Is(err, kvutil.ErrEncode):
+	case errors.Is(err, kvstore.ErrEncode):
 		slog.ErrorContext(ctx, "Failed to marshal placement group record",
 			"op", op, "groupName", groupName, "accountID", accountID, "err", err)
-	case errors.Is(err, kvutil.ErrWrite):
+	case errors.Is(err, kvstore.ErrWrite):
 		slog.ErrorContext(ctx, "Failed to write placement group to KV",
 			"op", op, "groupName", groupName, "accountID", accountID, "err", err)
 	default:
@@ -378,8 +378,8 @@ func mapCASError(ctx context.Context, op, groupName, accountID string, err error
 }
 
 // casConfig is the shared CAS policy for every placement group mutation.
-func casConfig() kvutil.CASConfig {
+func casConfig() kvstore.CASConfig {
-	return kvutil.CASConfig{
+	return kvstore.CASConfig{
 		Attempts:  maxCASRetries,
 		NotFound:  errPGAbsent,
 		Exhausted: func(string, int) error { return errPGContended },
@@ -394,7 +394,7 @@ func (s *PlacementGroupServiceImpl) ReserveSpreadNodes(ctx context.Context, inpu
 	}
 
 	var selected []string
-	_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
 		func(record *PlacementGroupRecord) (bool, error) {
 			if record.State != ec2.PlacementGroupStateAvailable {
 				return false, errors.New(awserrors.ErrorInvalidPlacementGroupUnknown)
@@ -437,7 +437,7 @@ func (s *PlacementGroupServiceImpl) FinalizeSpreadInstances(ctx context.Context,
 		return nil, errors.New(awserrors.ErrorMissingParameter)
 	}
 
-	_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
 		func(record *PlacementGroupRecord) (bool, error) {
 			maps.Copy(record.NodeInstances, input.NodeInstances)
 			return true, nil
@@ -462,7 +462,7 @@ func (s *PlacementGroupServiceImpl) ReleaseSpreadNodes(ctx context.Context, inpu
 		releaseSet[n] = true
 	}
 
-	_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
 		func(record *PlacementGroupRecord) (bool, error) {
 			changed := false
 			for node := range releaseSet {
@@ -490,7 +490,7 @@ func (s *PlacementGroupServiceImpl) RemoveInstance(ctx context.Context, input *R
 		return nil, errors.New(awserrors.ErrorMissingParameter)
 	}
 
-	_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
 		func(record *PlacementGroupRecord) (bool, error) {
 			instances, exists := record.NodeInstances[input.NodeName]
 			if !exists {
@@ -533,7 +533,7 @@ func (s *PlacementGroupServiceImpl) ReserveClusterNode(ctx context.Context, inpu
 	}
 
 	var targetNode string
-	_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
 		func(record *PlacementGroupRecord) (bool, error) {
 			if record.State != ec2.PlacementGroupStateAvailable {
 				return false, errors.New(awserrors.ErrorInvalidPlacementGroupUnknown)
@@ -570,7 +570,7 @@ func (s *PlacementGroupServiceImpl) FinalizeClusterInstances(ctx context.Context
 		return nil, errors.New(awserrors.ErrorMissingParameter)
 	}
 
-	_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
 		func(record *PlacementGroupRecord) (bool, error) {
 			// Append rather than replace: a cluster group can have several launches in flight at once.
 			for node, ids := range input.NodeInstances {

spinifex/domains/ec2/placementgroup/service_impl_test.go

@@ -10,7 +10,7 @@ import (
 	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -39,7 +39,7 @@ func createTestGroup(t *testing.T, svc *PlacementGroupServiceImpl, name, strateg
 // readRecord returns the stored record for assertions.
 func readRecord(t *testing.T, svc *PlacementGroupServiceImpl, groupName string) *PlacementGroupRecord {
 	t.Helper()
-	entry, err := svc.kv.Get(t.Context(), kvutil.AccountKey(testAccountID, groupName))
+	entry, err := svc.kv.Get(t.Context(), kvstore.AccountKey(testAccountID, groupName))
 	require.NoError(t, err)
 	var record PlacementGroupRecord
 	require.NoError(t, json.Unmarshal(entry.Value(), &record))
@@ -50,7 +50,7 @@ func readRecord(t *testing.T, svc *PlacementGroupServiceImpl, groupName string)
 // starting state the public API cannot produce.
 func seedRecord(t *testing.T, svc *PlacementGroupServiceImpl, groupName string, mutate func(*PlacementGroupRecord)) {
 	t.Helper()
-	_, err := kvutil.Update(t.Context(), svc.kv, kvutil.AccountKey(testAccountID, groupName), kvutil.CASConfig{},
+	_, err := kvstore.Update(t.Context(), svc.kv, kvstore.AccountKey(testAccountID, groupName), kvstore.CASConfig{},
 		func(record *PlacementGroupRecord) (bool, error) {
 			mutate(record)
 			return true, nil

spinifex/domains/ec2/routetable/service_impl.go

@@ -25,7 +25,7 @@ import (
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
 	awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -52,7 +52,7 @@ func NewRouteTableServiceImplWithNATS(ctx context.Context, cfg *config.Config, n
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	rtbKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketRouteTables, 10)
+	rtbKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketRouteTables, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketRouteTables, err)
 	}
@@ -60,22 +60,22 @@ func NewRouteTableServiceImplWithNATS(ctx context.Context, cfg *config.Config, n
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketRouteTables, err)
 	}
 
-	vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
+	vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
 	}
 
-	igwKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 10)
+	igwKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get IGW KV bucket: %w", err)
 	}
 
-	subnetKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
+	subnetKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get subnet KV bucket: %w", err)
 	}
 
-	natgwKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2natgw.KVBucketNatGateways, 10)
+	natgwKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2natgw.KVBucketNatGateways, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get NAT Gateway KV bucket: %w", err)
 	}
@@ -95,7 +95,7 @@ func NewRouteTableServiceImplWithNATS(ctx context.Context, cfg *config.Config, n
 
 // getRouteTable retrieves a route table record from KV.
 func (s *RouteTableServiceImpl) getRouteTable(ctx context.Context, accountID, rtbID string) (*RouteTableRecord, error) {
-	entry, err := s.rtbKV.Get(ctx, kvutil.AccountKey(accountID, rtbID))
+	entry, err := s.rtbKV.Get(ctx, kvstore.AccountKey(accountID, rtbID))
 	if err != nil {
 		if errors.Is(err, jetstream.ErrKeyNotFound) {
 			return nil, rtbNotFoundError(rtbID)
@@ -118,7 +118,7 @@ func (s *RouteTableServiceImpl) putRouteTable(ctx context.Context, accountID str
 		slog.ErrorContext(ctx, "Failed to marshal route table record", "routeTableId", record.RouteTableId, "err", err)
 		return errors.New(awserrors.ErrorServerInternal)
 	}
-	if _, err := s.rtbKV.Put(ctx, kvutil.AccountKey(accountID, record.RouteTableId), data); err != nil {
+	if _, err := s.rtbKV.Put(ctx, kvstore.AccountKey(accountID, record.RouteTableId), data); err != nil {
 		slog.ErrorContext(ctx, "Failed to write route table to KV", "routeTableId", record.RouteTableId, "err", err)
 		return errors.New(awserrors.ErrorServerInternal)
 	}
@@ -130,7 +130,7 @@ func (s *RouteTableServiceImpl) putRouteTable(ctx context.Context, accountID str
 // parallel AssociateRouteTable calls Terraform fans out per route table.
 const rtbCASMaxRetries = 16
 
-// errRTBAbsent and errRTBContended mark the two kvutil outcomes that map to
+// errRTBAbsent and errRTBContended mark the two kvstore outcomes that map to
 // specific AWS errors rather than a generic internal failure. Neither
 // escapes this file.
 var (
@@ -143,7 +143,7 @@ var (
 // several subnets with one route table at once. mutate reports whether
 // it changed the record; a false return commits nothing.
 func (s *RouteTableServiceImpl) mutateRouteTableCAS(ctx context.Context, accountID, rtbID string, mutate func(*RouteTableRecord) (bool, error)) error {
-	_, err := kvutil.Update(ctx, s.rtbKV, kvutil.AccountKey(accountID, rtbID), kvutil.CASConfig{
+	_, err := kvstore.Update(ctx, s.rtbKV, kvstore.AccountKey(accountID, rtbID), kvstore.CASConfig{
 		Attempts:  rtbCASMaxRetries,
 		NotFound:  errRTBAbsent,
 		Exhausted: func(string, int) error { return errRTBContended },
@@ -157,13 +157,13 @@ func (s *RouteTableServiceImpl) mutateRouteTableCAS(ctx context.Context, account
 		// Contended rather than broken.
 		slog.ErrorContext(ctx, "Route table CAS retries exhausted under contention",
 			"routeTableId", rtbID, "attempts", rtbCASMaxRetries)
-	case errors.Is(err, kvutil.ErrRead):
+	case errors.Is(err, kvstore.ErrRead):
 		slog.ErrorContext(ctx, "Failed to read route table from KV", "routeTableId", rtbID, "err", err)
-	case errors.Is(err, kvutil.ErrDecode):
+	case errors.Is(err, kvstore.ErrDecode):
 		slog.ErrorContext(ctx, "Corrupt route table record in KV", "routeTableId", rtbID, "err", err)
-	case errors.Is(err, kvutil.ErrEncode):
+	case errors.Is(err, kvstore.ErrEncode):
 		slog.ErrorContext(ctx, "Failed to marshal route table record", "routeTableId", rtbID, "err", err)
-	case errors.Is(err, kvutil.ErrWrite):
+	case errors.Is(err, kvstore.ErrWrite):
 		slog.ErrorContext(ctx, "Failed to write route table to KV", "routeTableId", rtbID, "err", err)
 	default:
 		return err
@@ -173,7 +173,7 @@ func (s *RouteTableServiceImpl) mutateRouteTableCAS(ctx context.Context, account
 
 // getVPCCidr looks up a VPC's CIDR block from the VPC KV bucket.
 func (s *RouteTableServiceImpl) getVPCCidr(ctx context.Context, accountID, vpcID string) (string, error) {
-	entry, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID))
+	entry, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcID))
 	if err != nil {
 		return "", awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", vpcID)
 	}
@@ -229,7 +229,7 @@ func (s *RouteTableServiceImpl) subnetsImplicitlyOnMainRT(ctx context.Context, a
 	}
 	var implicit []string
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.subnetKV.Get(ctx, key)
@@ -387,7 +387,7 @@ func (s *RouteTableServiceImpl) allSubnetsForVPC(ctx context.Context, accountID,
 	}
 	var subnets []string
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.subnetKV.Get(ctx, key)
@@ -455,7 +455,7 @@ func (s *RouteTableServiceImpl) allRouteTablesForVPC(ctx context.Context, accoun
 
 	var results []RouteTableRecord
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.rtbKV.Get(ctx, key)
@@ -577,7 +577,7 @@ func (s *RouteTableServiceImpl) DeleteRouteTable(ctx context.Context, input *ec2
 		}
 	}
 
-	if err := s.rtbKV.Delete(ctx, kvutil.AccountKey(accountID, rtbID)); err != nil {
+	if err := s.rtbKV.Delete(ctx, kvstore.AccountKey(accountID, rtbID)); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -633,7 +633,7 @@ func (s *RouteTableServiceImpl) DescribeRouteTables(ctx context.Context, input *
 	foundIDs := make(map[string]bool)
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 
@@ -740,7 +740,7 @@ func (s *RouteTableServiceImpl) CreateRoute(ctx context.Context, input *ec2.Crea
 	case input.GatewayId != nil && *input.GatewayId != "":
 		igwID := *input.GatewayId
 		// Verify IGW exists and is attached to the same VPC
-		igwEntry, err := s.igwKV.Get(ctx, kvutil.AccountKey(accountID, igwID))
+		igwEntry, err := s.igwKV.Get(ctx, kvstore.AccountKey(accountID, igwID))
 		if err != nil {
 			return nil, gatewayNotFoundError(igwID)
 		}
@@ -770,7 +770,7 @@ func (s *RouteTableServiceImpl) CreateRoute(ctx context.Context, input *ec2.Crea
 	case input.NatGatewayId != nil && *input.NatGatewayId != "":
 		natgwID := *input.NatGatewayId
 		// Verify NAT GW exists and belongs to the same VPC
-		natgwEntry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, natgwID))
+		natgwEntry, err := s.natgwKV.Get(ctx, kvstore.AccountKey(accountID, natgwID))
 		if err != nil {
 			return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound)
 		}
@@ -924,7 +924,7 @@ func (s *RouteTableServiceImpl) ReplaceRoute(ctx context.Context, input *ec2.Rep
 	igwID := *input.GatewayId
 
 	// AWS reports a missing gateway ahead of a missing route.
-	igwEntry, err := s.igwKV.Get(ctx, kvutil.AccountKey(accountID, igwID))
+	igwEntry, err := s.igwKV.Get(ctx, kvstore.AccountKey(accountID, igwID))
 	if err != nil {
 		return nil, gatewayNotFoundError(igwID)
 	}
@@ -983,7 +983,7 @@ func (s *RouteTableServiceImpl) AssociateRouteTable(ctx context.Context, input *
 	}
 
 	// Verify subnet exists and belongs to the same VPC
-	subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetID))
+	subnetEntry, err := s.subnetKV.Get(ctx, kvstore.AccountKey(accountID, subnetID))
 	if err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
 	}
@@ -1077,7 +1077,7 @@ func (s *RouteTableServiceImpl) DisassociateRouteTable(ctx context.Context, inpu
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 
@@ -1168,7 +1168,7 @@ func (s *RouteTableServiceImpl) ReplaceRouteTableAssociation(ctx context.Context
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 
@@ -1369,7 +1369,7 @@ func (s *RouteTableServiceImpl) publishNatGatewayDeleteEvents(ctx context.Contex
 	if s.natsConn == nil {
 		return
 	}
-	natgwEntry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, natgwID))
+	natgwEntry, err := s.natgwKV.Get(ctx, kvstore.AccountKey(accountID, natgwID))
 	if err != nil {
 		slog.WarnContext(ctx, "NAT GW event: natgw lookup failed", "topic", networkv1.NATGatewayDeleteSubject, "natGatewayId", natgwID, "err", err)
 		return
@@ -1417,7 +1417,7 @@ func (s *RouteTableServiceImpl) publishNatGatewayEventsForAssociation(ctx contex
 		if r.NatGatewayId == "" {
 			continue
 		}
-		natgwEntry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, r.NatGatewayId))
+		natgwEntry, err := s.natgwKV.Get(ctx, kvstore.AccountKey(accountID, r.NatGatewayId))
 		if err != nil {
 			slog.WarnContext(ctx, "NAT GW event: natgw lookup failed", "topic", topic, "natGatewayId", r.NatGatewayId, "err", err)
 			continue
@@ -1442,7 +1442,7 @@ func (s *RouteTableServiceImpl) publishNatGatewayEventForSubnet(ctx context.Cont
 	if s.natsConn == nil {
 		return
 	}
-	subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetID))
+	subnetEntry, err := s.subnetKV.Get(ctx, kvstore.AccountKey(accountID, subnetID))
 	if err != nil {
 		slog.WarnContext(ctx, "NAT GW event: subnet lookup failed", "topic", topic, "subnetId", subnetID, "err", err)
 		return

spinifex/domains/ec2/routetable/service_impl_test.go

@@ -13,7 +13,7 @@ import (
 	ec2natgw "github.com/mulgadc/spinifex/spinifex/domains/ec2/natgw"
 	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -28,23 +28,23 @@ func setupTestService(t *testing.T) *RouteTableServiceImpl {
 
 	// Seed VPC KV
 	testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
-		kvutil.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
+		kvstore.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
 	})
 
 	// Seed IGW KV (attached to vpc-test1)
 	testutil.SeedKV(t, js, ec2igw.KVBucketIGW, map[string][]byte{
-		kvutil.AccountKey(testAccountID, "igw-test1"): []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"available"}`),
+		kvstore.AccountKey(testAccountID, "igw-test1"): []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"available"}`),
 	})
 
 	// Seed subnet KV
 	testutil.SeedKV(t, js, ec2vpc.KVBucketSubnets, map[string][]byte{
-		kvutil.AccountKey(testAccountID, "subnet-test1"): []byte(`{"subnet_id":"subnet-test1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available"}`),
+		kvstore.AccountKey(testAccountID, "subnet-test1"): []byte(`{"subnet_id":"subnet-test1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available"}`),
-		kvutil.AccountKey(testAccountID, "subnet-priv1"): []byte(`{"subnet_id":"subnet-priv1","vpc_id":"vpc-test1","cidr_block":"10.0.11.0/24","state":"available"}`),
+		kvstore.AccountKey(testAccountID, "subnet-priv1"): []byte(`{"subnet_id":"subnet-priv1","vpc_id":"vpc-test1","cidr_block":"10.0.11.0/24","state":"available"}`),
 	})
 
 	// Seed NAT Gateway KV
 	testutil.SeedKV(t, js, ec2natgw.KVBucketNatGateways, map[string][]byte{
-		kvutil.AccountKey(testAccountID, "nat-test1"): []byte(`{"nat_gateway_id":"nat-test1","vpc_id":"vpc-test1","subnet_id":"subnet-test1","public_ip":"192.168.1.100","state":"available"}`),
+		kvstore.AccountKey(testAccountID, "nat-test1"): []byte(`{"nat_gateway_id":"nat-test1","vpc_id":"vpc-test1","subnet_id":"subnet-test1","public_ip":"192.168.1.100","state":"available"}`),
 	})
 
 	svc, err := NewRouteTableServiceImplWithNATS(t.Context(), nil, nc)

spinifex/domains/ec2/spotinstance/account_scan.go

@@ -10,7 +10,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/nats-io/nats.go/jetstream"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 )
 
 // AccountSpotRequest is one persisted request in an account scan.
@@ -54,7 +54,7 @@ func scanBucket(ctx context.Context, kv jetstream.KeyValue, accountID string) ([
 	prefix := accountID + "."
 	var out []AccountSpotRequest
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := kv.Get(ctx, key)

spinifex/domains/ec2/spotinstance/account_scan_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 )
 
 func TestScanAccountListsActiveAndTerminalForOneAccount(t *testing.T) {
@@ -35,7 +35,7 @@ func TestScanAccountListsActiveAndTerminalForOneAccount(t *testing.T) {
 func TestScanAccountFailsOnAnUndecodableRecord(t *testing.T) {
 	svc := setupTestService(t)
 	putSIRs(t, svc, makeSIR("sir-aaa", "i-aaa"))
-	_, err := svc.activeKV.Put(t.Context(), kvutil.AccountKey(testAccountID, "sir-bad"), []byte("{not json"))
+	_, err := svc.activeKV.Put(t.Context(), kvstore.AccountKey(testAccountID, "sir-bad"), []byte("{not json"))
 	require.NoError(t, err)
 
 	scan, err := svc.ScanAccount(t.Context(), testAccountID)
@@ -45,7 +45,7 @@ func TestScanAccountFailsOnAnUndecodableRecord(t *testing.T) {
 
 func TestScanAccountFailsOnARecordWithoutARequest(t *testing.T) {
 	svc := setupTestService(t)
-	_, err := svc.terminalKV.Put(t.Context(), kvutil.AccountKey(testAccountID, "sir-empty"), []byte(`{"account_id":"x"}`))
+	_, err := svc.terminalKV.Put(t.Context(), kvstore.AccountKey(testAccountID, "sir-empty"), []byte(`{"account_id":"x"}`))
 	require.NoError(t, err)
 
 	_, err = svc.ScanAccount(t.Context(), testAccountID)
@@ -76,7 +76,7 @@ func (k failingGetKV) Get(ctx context.Context, key string) (jetstream.KeyValueEn
 func TestScanAccountFailsWhenARecordCannotBeRead(t *testing.T) {
 	svc := setupTestService(t)
 	putSIRs(t, svc, makeSIR("sir-aaa", "i-aaa"), makeSIR("sir-bbb", "i-bbb"))
-	svc.activeKV = failingGetKV{KeyValue: svc.activeKV, key: kvutil.AccountKey(testAccountID, "sir-bbb")}
+	svc.activeKV = failingGetKV{KeyValue: svc.activeKV, key: kvstore.AccountKey(testAccountID, "sir-bbb")}
 
 	scan, err := svc.ScanAccount(t.Context(), testAccountID)
 	require.Error(t, err)

spinifex/domains/ec2/spotinstance/launch_references.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -24,14 +24,14 @@ type LaunchReferences struct {
 // account. It is complete or an error: an unlistable bucket or an unreadable
 // request is an error, because the unread request may name the resource.
 func (s *SpotInstanceServiceImpl) AccountLaunchReferences(ctx context.Context, accountID string) (*LaunchReferences, error) {
-	keys, err := kvutil.Keys(ctx, s.activeKV)
+	keys, err := kvstore.Keys(ctx, s.activeKV)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
 		return nil, fmt.Errorf("list spot requests: %w", err)
 	}
 	prefix := accountID + "."
 	out := &LaunchReferences{}
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		record, err := s.readRecord(ctx, s.activeKV, key)

spinifex/domains/ec2/spotinstance/launch_references_test.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -41,7 +41,7 @@ func TestAccountLaunchReferences_ReadsOpenAndActiveRequests(t *testing.T) {
 func TestAccountLaunchReferences_UnreadableRequestFailsTheCheck(t *testing.T) {
 	svc := setupTestService(t)
 	putSIRs(t, svc, makeSIR("sir-active01", "i-0000000000000001"))
-	_, err := svc.activeKV.Put(t.Context(), kvutil.AccountKey(testAccountID, "sir-broken01"), []byte("{not json"))
+	_, err := svc.activeKV.Put(t.Context(), kvstore.AccountKey(testAccountID, "sir-broken01"), []byte("{not json"))
 	require.NoError(t, err)
 
 	_, err = svc.AccountLaunchReferences(t.Context(), testAccountID)

spinifex/domains/ec2/spotinstance/service_impl.go

@@ -15,7 +15,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -41,7 +41,7 @@ func NewSpotInstanceServiceImplWithNATS(ctx context.Context, cfg *config.Config,
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	activeKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketSpotRequests, 10)
+	activeKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketSpotRequests, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSpotRequests, err)
 	}
@@ -71,7 +71,7 @@ func NewSpotInstanceServiceImplWithNATS(ctx context.Context, cfg *config.Config,
 // getOrCreateTerminalBucket opens the terminal bucket without changing its
 // existing replica or placement configuration, creating it only when absent.
 func getOrCreateTerminalBucket(ctx context.Context, js jetstream.KeyValueManager) (jetstream.KeyValue, error) {
-	return kvutil.GetOrCreateBucketWithOptions(ctx, js, kvutil.BucketOptions{
+	return kvstore.GetOrCreateBucketWithOptions(ctx, js, kvstore.BucketOptions{
 		Name:        KVBucketSpotRequestsTerminal,
 		Description: "Terminal Spot Instance Requests (auto-expire after 1 hour)",
 		History:     1,
@@ -91,7 +91,7 @@ func (s *SpotInstanceServiceImpl) PutSpotInstanceRequests(ctx context.Context, i
 		if err != nil {
 			return nil, errors.New(awserrors.ErrorServerInternal)
 		}
-		if _, err := s.activeKV.Put(ctx, kvutil.AccountKey(accountID, sirID), data); err != nil {
+		if _, err := s.activeKV.Put(ctx, kvstore.AccountKey(accountID, sirID), data); err != nil {
 			slog.ErrorContext(ctx, "PutSpotInstanceRequests: KV put failed", "sirId", sirID, "err", err)
 			return nil, errors.New(awserrors.ErrorServerInternal)
 		}
@@ -171,7 +171,7 @@ func (s *SpotInstanceServiceImpl) CancelSpotInstanceRequests(ctx context.Context
 	cancelled := make([]*ec2.CancelledSpotInstanceRequest, 0, len(input.SpotInstanceRequestIds))
 	for _, idPtr := range input.SpotInstanceRequestIds {
 		sirID := aws.StringValue(idPtr)
-		key := kvutil.AccountKey(accountID, sirID)
+		key := kvstore.AccountKey(accountID, sirID)
 
 		record, err := s.readRecord(ctx, s.activeKV, key)
 		if err == nil {
@@ -213,7 +213,7 @@ func (s *SpotInstanceServiceImpl) CloseForInstance(ctx context.Context, instance
 
 	prefix := accountID + "."
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		record, err := s.readRecord(ctx, s.activeKV, key)
@@ -251,7 +251,7 @@ func (s *SpotInstanceServiceImpl) listRequests(ctx context.Context, kv jetstream
 	prefix := accountID + "."
 	var requests []*ec2.SpotInstanceRequest
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		record, err := s.readRecord(ctx, kv, key)

spinifex/domains/ec2/tagmirror/tagging_current_behaviour_test.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 	"testing"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -39,7 +39,7 @@ func (k *interleavingKV) Get(ctx context.Context, key string) (jetstream.KeyValu
 // its write both survive alongside its own tag.
 func TestCurrentBehaviour_Tagging_RecordMirrorKeepsInterleavedWrites(t *testing.T) {
 	const acct, id = "123456789012", "sg-0123456789abcdef0"
-	key := kvutil.AccountKey(acct, id)
+	key := kvstore.AccountKey(acct, id)
 	seed, err := json.Marshal(tagMirrorRecord{Name: "original", Tags: map[string]string{}})
 	require.NoError(t, err)
 	kv := seedKV(t, "tagmirror-cb", map[string][]byte{key: seed})

spinifex/domains/ec2/tagmirror/tagmirror.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -83,8 +83,8 @@ var errRecordAbsent = errors.New("tagmirror: record absent")
 // applying mut and committing at the read revision, retrying on conflict so a
 // concurrent write is never reverted. An absent resource is skipped.
 func UpdateKVRecordTags[R any](ctx context.Context, kv jetstream.KeyValue, accountID, resourceID string, mut func(*R)) error {
-	key := kvutil.AccountKey(accountID, resourceID)
+	key := kvstore.AccountKey(accountID, resourceID)
-	_, err := kvutil.Update(ctx, kv, key, kvutil.CASConfig{NotFound: errRecordAbsent}, func(rec *R) (bool, error) {
+	_, err := kvstore.Update(ctx, kv, key, kvstore.CASConfig{NotFound: errRecordAbsent}, func(rec *R) (bool, error) {
 		mut(rec)
 		return true, nil
 	})

spinifex/domains/ec2/tagmirror/tagmirror_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -75,15 +75,15 @@ func TestUpdateKVRecordTags(t *testing.T) {
 	rec, err := json.Marshal(&tagMirrorRecord{Name: "r1", Tags: map[string]string{"a": "1"}})
 	require.NoError(t, err)
 	kv := seedKV(t, "tagmirror-test", map[string][]byte{
-		kvutil.AccountKey("acct", "res-1"):   rec,
+		kvstore.AccountKey("acct", "res-1"):   rec,
-		kvutil.AccountKey("acct", "res-bad"): []byte("not-json"),
+		kvstore.AccountKey("acct", "res-bad"): []byte("not-json"),
 	})
 
 	t.Run("mutates and persists", func(t *testing.T) {
 		require.NoError(t, UpdateKVRecordTags(t.Context(), kv, "acct", "res-1", func(r *tagMirrorRecord) {
 			r.Tags["b"] = "2"
 		}))
-		entry, err := kv.Get(t.Context(), kvutil.AccountKey("acct", "res-1"))
+		entry, err := kv.Get(t.Context(), kvstore.AccountKey("acct", "res-1"))
 		require.NoError(t, err)
 		var got tagMirrorRecord
 		require.NoError(t, json.Unmarshal(entry.Value(), &got))
@@ -110,14 +110,14 @@ func TestMirrorKVRecordTags(t *testing.T) {
 	nilTags, err := json.Marshal(&tagMirrorRecord{})
 	require.NoError(t, err)
 	kv := seedKV(t, "tagmirror-mirror-test", map[string][]byte{
-		kvutil.AccountKey("acct", "vol-1"): withTags,
+		kvstore.AccountKey("acct", "vol-1"): withTags,
-		kvutil.AccountKey("acct", "vol-2"): nilTags,
+		kvstore.AccountKey("acct", "vol-2"): nilTags,
 	})
 
 	tagsOf := func(r *tagMirrorRecord) *map[string]string { return &r.Tags }
 
 	readTags := func(t *testing.T, id string) map[string]string {
-		entry, err := kv.Get(t.Context(), kvutil.AccountKey("acct", id))
+		entry, err := kv.Get(t.Context(), kvstore.AccountKey("acct", id))
 		require.NoError(t, err)
 		var got tagMirrorRecord
 		require.NoError(t, json.Unmarshal(entry.Value(), &got))
@@ -150,7 +150,7 @@ func TestMirrorKVRecordTags(t *testing.T) {
 	// must never be touched even though it exists in this bucket.
 	otherRec, err := json.Marshal(&tagMirrorRecord{Tags: map[string]string{"x": "y"}})
 	require.NoError(t, err)
-	_, err = kv.Put(t.Context(), kvutil.AccountKey("acct", "snap-other"), otherRec)
+	_, err = kv.Put(t.Context(), kvstore.AccountKey("acct", "snap-other"), otherRec)
 	require.NoError(t, err)
 	require.NoError(t, MirrorKVRecordTags(t.Context(), kv, "acct", "vol-", resources, tagsOf,
 		MergeTagsMut(&ec2.CreateTagsInput{Tags: []*ec2.Tag{

spinifex/domains/ec2/tags/service_impl.go

@@ -23,7 +23,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -274,7 +274,7 @@ func (s *TagsServiceImpl) CreateTags(ctx context.Context, input *ec2.CreateTagsI
 func (s *TagsServiceImpl) taggedResourceIDs(ctx context.Context, accountID string) ([]string, error) {
 	seen := make(map[string]struct{})
 
-	keys, err := kvutil.Keys(ctx, s.kv)
+	keys, err := kvstore.Keys(ctx, s.kv)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
 		return nil, fmt.Errorf("list tag keys: %w", err)
 	}

spinifex/domains/ec2/tags/store.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -46,7 +46,7 @@ func resourceIDFromKVKey(key, accountID string) (string, bool) {
 // GetOrCreateTagsBucket opens the tag bucket, creating it at the cluster's
 // replica count if this is the first node to reach it.
 func GetOrCreateTagsBucket(ctx context.Context, js jetstream.KeyValueManager) (jetstream.KeyValue, error) {
-	return kvutil.GetOrCreateBucket(ctx, js, KVBucketTags, kvTagsHistory)
+	return kvstore.GetOrCreateBucket(ctx, js, KVBucketTags, kvTagsHistory)
 }
 
 // readKVTags returns a resource's tags from the bucket, and whether the key
@@ -76,7 +76,7 @@ func (s *TagsServiceImpl) readKVTags(ctx context.Context, accountID, resourceID
 // before the store moved.
 func (s *TagsServiceImpl) mutateTags(ctx context.Context, accountID, resourceID string, mutate func(map[string]string)) error {
 	key := tagsKVKey(accountID, resourceID)
-	_, err := kvutil.Update(ctx, s.kv, key, kvutil.CASConfig{
+	_, err := kvstore.Update(ctx, s.kv, key, kvstore.CASConfig{
 		Attempts:       tagCASAttempts,
 		CreateIfAbsent: true,
 		Exhausted: func(key string, attempts int) error {
@@ -104,7 +104,7 @@ func (s *TagsServiceImpl) PutResourceTags(ctx context.Context, accountID, resour
 	if err != nil {
 		return err
 	}
-	return kvutil.Put(ctx, s.kv, tagsKVKey(accountID, resourceID), kvutil.CASConfig{
+	return kvstore.Put(ctx, s.kv, tagsKVKey(accountID, resourceID), kvstore.CASConfig{
 		Attempts:       tagCASAttempts,
 		CreateIfAbsent: true,
 	}, data)

spinifex/domains/ec2/vpc/default_vpc.go

@@ -12,7 +12,7 @@ import (
 
 	networkv1 "github.com/mulgadc/spinifex/contracts/network/v1"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -105,7 +105,7 @@ func (s *VPCServiceImpl) ensureDefaultVPC(ctx context.Context, accountID string,
 			return claim.info(), nil
 		}
 
-		if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, claim.VpcId)); err == nil {
+		if _, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, claim.VpcId)); err == nil {
 			return claim.info(), nil
 		} else if !errors.Is(err, jetstream.ErrKeyNotFound) {
 			return nil, fmt.Errorf("read default VPC %s: %w", claim.VpcId, err)
@@ -233,7 +233,7 @@ func oldestAccountRecord[T any](ctx context.Context, kv jetstream.KeyValue, acco
 	var bestID string
 	var bestCreated time.Time
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := kv.Get(ctx, key)
@@ -262,7 +262,7 @@ func oldestAccountRecord[T any](ctx context.Context, kv jetstream.KeyValue, acco
 // marks the claim complete. Records are written with Create, so only the caller
 // that actually wrote one announces it to vpcd.
 func (s *VPCServiceImpl) buildDefaultVPC(ctx context.Context, accountID string, claim *defaultVPCClaim) error {
-	created, err := createRecord(ctx, s.vpcKV, kvutil.AccountKey(accountID, claim.VpcId), VPCRecord{
+	created, err := createRecord(ctx, s.vpcKV, kvstore.AccountKey(accountID, claim.VpcId), VPCRecord{
 		VpcId:              claim.VpcId,
 		CidrBlock:          claim.Cidr,
 		State:              "available",
@@ -284,7 +284,7 @@ func (s *VPCServiceImpl) buildDefaultVPC(ctx context.Context, accountID string,
 	if s.config != nil && s.config.AZ != "" {
 		az = s.config.AZ
 	}
-	created, err = createRecord(ctx, s.subnetKV, kvutil.AccountKey(accountID, claim.SubnetId), SubnetRecord{
+	created, err = createRecord(ctx, s.subnetKV, kvstore.AccountKey(accountID, claim.SubnetId), SubnetRecord{
 		SubnetId:            claim.SubnetId,
 		VpcId:               claim.VpcId,
 		CidrBlock:           claim.SubnetCidr,
@@ -320,7 +320,7 @@ func (s *VPCServiceImpl) buildDefaultVPC(ctx context.Context, accountID string,
 		}
 	}
 
-	if _, err := kvutil.Update(ctx, s.defaultsKV, accountID, kvutil.CASConfig{}, func(c *defaultVPCClaim) (bool, error) {
+	if _, err := kvstore.Update(ctx, s.defaultsKV, accountID, kvstore.CASConfig{}, func(c *defaultVPCClaim) (bool, error) {
 		if c.Complete || c.VpcId != claim.VpcId {
 			return false, nil
 		}

spinifex/domains/ec2/vpc/default_vpc_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/internal/testkit"
 	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
@@ -149,12 +149,12 @@ func TestEnsureDefaultVPC_AdoptsOldestExistingDefault(t *testing.T) {
 	const accountID = "000000000051"
 	created := time.Date(2026, 9, 11, 14, 13, 40, 0, time.UTC)
 	for i, id := range []string{"vpc-0000000000000newer", "vpc-0000000000000older"} {
-		putRecord(t, js, ec2vpc.KVBucketVPCs, kvutil.AccountKey(accountID, id), ec2vpc.VPCRecord{
+		putRecord(t, js, ec2vpc.KVBucketVPCs, kvstore.AccountKey(accountID, id), ec2vpc.VPCRecord{
 			VpcId: id, CidrBlock: ec2vpc.DefaultVPCCidr, IsDefault: true, VNI: int64(200 + i),
 			CreatedAt: created.Add(time.Duration(1-i) * time.Minute),
 		})
 	}
-	putRecord(t, js, ec2vpc.KVBucketSubnets, kvutil.AccountKey(accountID, "subnet-00000000000older"), ec2vpc.SubnetRecord{
+	putRecord(t, js, ec2vpc.KVBucketSubnets, kvstore.AccountKey(accountID, "subnet-00000000000older"), ec2vpc.SubnetRecord{
 		SubnetId: "subnet-00000000000older", VpcId: "vpc-0000000000000older", CidrBlock: ec2vpc.DefaultSubnetCidr,
 		IsDefault: true, CreatedAt: created,
 	})
@@ -183,7 +183,7 @@ func TestEnsureDefaultVPC_FinishesHalfBuiltClaim(t *testing.T) {
 		"route_table_id": "rtb-00000000000half", "internet_gateway_id": "igw-00000000000half",
 		"cidr": ec2vpc.DefaultVPCCidr, "subnet_cidr": ec2vpc.DefaultSubnetCidr, "vni": 4242,
 	})
-	putRecord(t, js, ec2vpc.KVBucketVPCs, kvutil.AccountKey(accountID, "vpc-00000000000000half"), ec2vpc.VPCRecord{
+	putRecord(t, js, ec2vpc.KVBucketVPCs, kvstore.AccountKey(accountID, "vpc-00000000000000half"), ec2vpc.VPCRecord{
 		VpcId: "vpc-00000000000000half", CidrBlock: ec2vpc.DefaultVPCCidr, IsDefault: true, VNI: 4242,
 	})
 

spinifex/domains/ec2/vpc/eni.go

@@ -23,7 +23,7 @@ import (
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -118,7 +118,7 @@ func (s *VPCServiceImpl) CreateNetworkInterface(ctx context.Context, input *ec2.
 	subnetId := *input.SubnetId
 
 	// Verify subnet exists and belongs to this account
-	subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetId))
+	subnetEntry, err := s.subnetKV.Get(ctx, kvstore.AccountKey(accountID, subnetId))
 	if err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
 	}
@@ -205,7 +205,7 @@ func (s *VPCServiceImpl) CreateNetworkInterface(ctx context.Context, input *ec2.
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal ENI record: %w", err)
 	}
-	if _, err := s.eniKV.Put(ctx, kvutil.AccountKey(accountID, eniId), data); err != nil {
+	if _, err := s.eniKV.Put(ctx, kvstore.AccountKey(accountID, eniId), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -251,7 +251,7 @@ func (s *VPCServiceImpl) ForceDeleteInstanceENI(ctx context.Context, accountID,
 }
 
 func (s *VPCServiceImpl) deleteNetworkInterface(ctx context.Context, eniId, accountID string, force bool) (*ec2.DeleteNetworkInterfaceOutput, error) {
-	key := kvutil.AccountKey(accountID, eniId)
+	key := kvstore.AccountKey(accountID, eniId)
 
 	// Get the ENI record
 	entry, err := s.eniKV.Get(ctx, key)
@@ -346,7 +346,7 @@ func (s *VPCServiceImpl) releaseENISideEffects(ctx context.Context, eniId, accou
 // deletion that did not happen.
 func (s *VPCServiceImpl) DetachAndDeleteENI(ctx context.Context, accountID, eniID string, force bool) (deleted bool, err error) {
 	const maxAttempts = 3
-	key := kvutil.AccountKey(accountID, eniID)
+	key := kvstore.AccountKey(accountID, eniID)
 	var lastErr error
 
 	for range maxAttempts {
@@ -458,7 +458,7 @@ func (s *VPCServiceImpl) ModifyNetworkInterfaceAttribute(ctx context.Context, in
 	}
 
 	eniId := *input.NetworkInterfaceId
-	key := kvutil.AccountKey(accountID, eniId)
+	key := kvstore.AccountKey(accountID, eniId)
 
 	entry, err := s.eniKV.Get(ctx, key)
 	if err != nil {
@@ -569,7 +569,7 @@ func (s *VPCServiceImpl) DescribeNetworkInterfaces(ctx context.Context, input *e
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, prefix) {
@@ -707,7 +707,7 @@ func (s *VPCServiceImpl) AttachENIWithOwner(accountID, eniId, instanceId string,
 
 func (s *VPCServiceImpl) attachENI(ctx context.Context, accountID, eniId, instanceId string, deviceIndex int64, instanceOwnerID string) (string, error) {
 	attachmentId := awsidentifiers.GenerateResourceID("eni-attach")
-	err := s.casENI(ctx, kvutil.AccountKey(accountID, eniId), eniId, func(record *ENIRecord) error {
+	err := s.casENI(ctx, kvstore.AccountKey(accountID, eniId), eniId, func(record *ENIRecord) error {
 		if record.Status == "in-use" {
 			return errors.New(awserrors.ErrorInvalidNetworkInterfaceInUse)
 		}
@@ -737,7 +737,7 @@ func (s *VPCServiceImpl) attachENI(ctx context.Context, accountID, eniId, instan
 // DetachENI marks an ENI as detached from an instance (internal use by TerminateInstances).
 // accountID scopes the lookup to the correct KV key.
 func (s *VPCServiceImpl) DetachENI(ctx context.Context, accountID, eniId string) error {
-	key := kvutil.AccountKey(accountID, eniId)
+	key := kvstore.AccountKey(accountID, eniId)
 	entry, err := s.eniKV.Get(ctx, key)
 	if err != nil {
 		return eniNotFoundError(eniId)
@@ -773,7 +773,7 @@ func (s *VPCServiceImpl) GetENIRecord(accountID, eniId string) (ENIRecord, error
 }
 
 func (s *VPCServiceImpl) getENIRecord(ctx context.Context, accountID, eniId string) (ENIRecord, error) {
-	key := kvutil.AccountKey(accountID, eniId)
+	key := kvstore.AccountKey(accountID, eniId)
 	entry, err := s.eniKV.Get(ctx, key)
 	if err != nil {
 		return ENIRecord{}, eniNotFoundError(eniId)
@@ -793,7 +793,7 @@ func (s *VPCServiceImpl) UpdateENI(accountID, eniId string, fn func(*ENIRecord))
 }
 
 func (s *VPCServiceImpl) updateENI(ctx context.Context, accountID, eniId string, fn func(*ENIRecord)) error {
-	return s.casENI(ctx, kvutil.AccountKey(accountID, eniId), eniId, func(r *ENIRecord) error {
+	return s.casENI(ctx, kvstore.AccountKey(accountID, eniId), eniId, func(r *ENIRecord) error {
 		fn(r)
 		return nil
 	})
@@ -876,7 +876,7 @@ func (s *VPCServiceImpl) listInstanceENIs(ctx context.Context, accountID, instan
 	prefix := accountID + "."
 	var out []ENIRecord
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.eniKV.Get(ctx, key)
@@ -938,7 +938,7 @@ func (s *VPCServiceImpl) scanENIs(ctx context.Context, match func(*ENIRecord) bo
 
 	var out []AccountENI
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		accountID, _, found := strings.Cut(key, ".")
@@ -1004,7 +1004,7 @@ func (s *VPCServiceImpl) findENIByAttachment(ctx context.Context, accountID, att
 	}
 	prefix := accountID + "."
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.eniKV.Get(ctx, key)
@@ -1028,7 +1028,7 @@ func (s *VPCServiceImpl) UpdateENIPublicIP(accountID, eniId, publicIP, poolName
 }
 
 func (s *VPCServiceImpl) updateENIPublicIP(ctx context.Context, accountID, eniId, publicIP, poolName string) error {
-	key := kvutil.AccountKey(accountID, eniId)
+	key := kvstore.AccountKey(accountID, eniId)
 	entry, err := s.eniKV.Get(ctx, key)
 	if err != nil {
 		return fmt.Errorf("ENI %s not found: %w", eniId, err)
@@ -1063,7 +1063,7 @@ func (s *VPCServiceImpl) fillSGNames(ctx context.Context, accountID string, ids
 			continue
 		}
 		names[id] = ""
-		entry, err := s.sgKV.Get(ctx, kvutil.AccountKey(accountID, id))
+		entry, err := s.sgKV.Get(ctx, kvstore.AccountKey(accountID, id))
 		if err != nil {
 			slog.WarnContext(ctx, "fillSGNames: SG read failed", "groupId", id, "err", err)
 			continue
@@ -1080,7 +1080,7 @@ func (s *VPCServiceImpl) fillSGNames(ctx context.Context, accountID string, ids
 // vpcDNSHostnames reports whether the VPC has enableDnsHostnames on. An
 // unreadable VPC reads as off, so its interfaces go unnamed.
 func (s *VPCServiceImpl) vpcDNSHostnames(ctx context.Context, accountID, vpcID string) bool {
-	entry, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID))
+	entry, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcID))
 	if err != nil {
 		slog.WarnContext(ctx, "vpcDNSHostnames: VPC read failed", "vpcId", vpcID, "err", err)
 		return false
@@ -1238,7 +1238,7 @@ func (s *VPCServiceImpl) validateSGAttachment(ctx context.Context, accountID str
 
 	// Each SG must exist in the caller's account and belong to the same VPC.
 	for _, sgId := range sgIds {
-		sgEntry, err := s.sgKV.Get(ctx, kvutil.AccountKey(accountID, sgId))
+		sgEntry, err := s.sgKV.Get(ctx, kvstore.AccountKey(accountID, sgId))
 		if err != nil {
 			return sgNotFoundError(sgId)
 		}

spinifex/domains/ec2/vpc/eni_lease_rebind.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -69,7 +69,7 @@ func (s *VPCServiceImpl) findENIAnyAccount(ctx context.Context, eniID string) (*
 
 	suffix := "." + eniID
 	for _, k := range keys {
-		if k == kvutil.VersionKey || !strings.HasSuffix(k, suffix) {
+		if k == kvstore.VersionKey || !strings.HasSuffix(k, suffix) {
 			continue
 		}
 		entry, err := s.eniKV.Get(ctx, k)

spinifex/domains/ec2/vpc/eni_stale_read_test.go

@@ -7,7 +7,7 @@ import (
 	"testing"
 
 	testutil "github.com/mulgadc/spinifex/internal/testkit"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -30,7 +30,7 @@ func newStaleReadENI(t *testing.T) (*VPCServiceImpl, string, func() ENIRecord) {
 	// Reads the stored record directly: the wrapped bucket would serve the
 	// check itself from a lagging replica.
 	stored := func() ENIRecord {
-		entry, err := inner.Get(context.Background(), kvutil.AccountKey(testAccountID, eniID))
+		entry, err := inner.Get(context.Background(), kvstore.AccountKey(testAccountID, eniID))
 		require.NoError(t, err)
 		var rec ENIRecord
 		require.NoError(t, json.Unmarshal(entry.Value(), &rec))
@@ -114,7 +114,7 @@ func TestCASENIGivesUpAfterBoundedConflicts(t *testing.T) {
 	t.Parallel()
 	svc, eniID, _ := newStaleReadENI(t)
 	conflicts := 0
-	err := svc.casENI(t.Context(), kvutil.AccountKey(testAccountID, eniID), eniID, func(*ENIRecord) error {
+	err := svc.casENI(t.Context(), kvstore.AccountKey(testAccountID, eniID), eniID, func(*ENIRecord) error {
 		conflicts++
 		// Move the record under the next compare-and-swap, as a competing writer does.
 		return svc.UpdateENI(testAccountID, eniID, func(r *ENIRecord) { r.Description = "other writer" })

spinifex/domains/ec2/vpc/lease_owner.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -42,7 +42,7 @@ func keyWithSuffixExists(ctx context.Context, kv jetstream.KeyValue, resourceID
 	}
 	suffix := "." + resourceID
 	for _, k := range keys {
-		if k != kvutil.VersionKey && strings.HasSuffix(k, suffix) {
+		if k != kvstore.VersionKey && strings.HasSuffix(k, suffix) {
 			return true, nil
 		}
 	}

spinifex/domains/ec2/vpc/security_group.go

@@ -23,7 +23,7 @@ import (
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
 	awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -245,7 +245,7 @@ func (s *VPCServiceImpl) CreateSecurityGroup(ctx context.Context, input *ec2.Cre
 	}
 	sgsInVPC := 0
 	for _, k := range sgKeys {
-		if k == kvutil.VersionKey {
+		if k == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(k, prefix) {
@@ -305,7 +305,7 @@ func (s *VPCServiceImpl) CreateSecurityGroup(ctx context.Context, input *ec2.Cre
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal security group record: %w", err)
 	}
-	if _, err := s.sgKV.Put(ctx, kvutil.AccountKey(accountID, groupId), data); err != nil {
+	if _, err := s.sgKV.Put(ctx, kvstore.AccountKey(accountID, groupId), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -331,7 +331,7 @@ func (s *VPCServiceImpl) DeleteSecurityGroup(ctx context.Context, input *ec2.Del
 	}
 
 	groupId := *input.GroupId
-	key := kvutil.AccountKey(accountID, groupId)
+	key := kvstore.AccountKey(accountID, groupId)
 
 	entry, err := s.sgKV.Get(ctx, key)
 	if err != nil {
@@ -380,7 +380,7 @@ func (s *VPCServiceImpl) validateSGRuleReferences(ctx context.Context, accountID
 		if r.SourceSG == "" {
 			continue
 		}
-		entry, err := s.sgKV.Get(ctx, kvutil.AccountKey(accountID, r.SourceSG))
+		entry, err := s.sgKV.Get(ctx, kvstore.AccountKey(accountID, r.SourceSG))
 		if err != nil {
 			return sgNotFoundError(r.SourceSG)
 		}
@@ -406,7 +406,7 @@ func (s *VPCServiceImpl) checkSGDependencies(ctx context.Context, accountID, gro
 		return errors.New(awserrors.ErrorServerInternal)
 	}
 	for _, k := range eniKeys {
-		if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
 			continue
 		}
 		entry, err := s.eniKV.Get(ctx, k)
@@ -435,7 +435,7 @@ func (s *VPCServiceImpl) checkSGDependencies(ctx context.Context, accountID, gro
 		return errors.New(awserrors.ErrorServerInternal)
 	}
 	for _, k := range sgKeys {
-		if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
 			continue
 		}
 		entry, err := s.sgKV.Get(ctx, k)
@@ -561,7 +561,7 @@ func (s *VPCServiceImpl) DescribeSecurityGroups(ctx context.Context, input *ec2.
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, prefix) {
@@ -713,7 +713,7 @@ func (s *VPCServiceImpl) GetSecurityGroupsForVpc(ctx context.Context, input *ec2
 
 	groups := []*ec2.SecurityGroupForVpc{}
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.sgKV.Get(ctx, key)
@@ -844,7 +844,7 @@ func (s *VPCServiceImpl) DescribeSecurityGroupRules(ctx context.Context, input *
 	emitted := make(map[string]bool)
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.sgKV.Get(ctx, key)
@@ -945,7 +945,7 @@ func (s *VPCServiceImpl) updateSGRuleTags(ctx context.Context, accountID, ruleID
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.sgKV.Get(ctx, key)
@@ -1056,7 +1056,7 @@ func (s *VPCServiceImpl) AuthorizeSecurityGroupIngress(ctx context.Context, inpu
 	}
 
 	groupId := *input.GroupId
-	key := kvutil.AccountKey(accountID, groupId)
+	key := kvstore.AccountKey(accountID, groupId)
 
 	entry, err := s.sgKV.Get(ctx, key)
 	if err != nil {
@@ -1134,7 +1134,7 @@ func (s *VPCServiceImpl) AuthorizeSecurityGroupEgress(ctx context.Context, input
 	}
 
 	groupId := *input.GroupId
-	key := kvutil.AccountKey(accountID, groupId)
+	key := kvstore.AccountKey(accountID, groupId)
 
 	entry, err := s.sgKV.Get(ctx, key)
 	if err != nil {
@@ -1212,7 +1212,7 @@ func (s *VPCServiceImpl) RevokeSecurityGroupIngress(ctx context.Context, input *
 	}
 
 	groupId := *input.GroupId
-	key := kvutil.AccountKey(accountID, groupId)
+	key := kvstore.AccountKey(accountID, groupId)
 
 	entry, err := s.sgKV.Get(ctx, key)
 	if err != nil {
@@ -1277,7 +1277,7 @@ func (s *VPCServiceImpl) RevokeSecurityGroupEgress(ctx context.Context, input *e
 	}
 
 	groupId := *input.GroupId
-	key := kvutil.AccountKey(accountID, groupId)
+	key := kvstore.AccountKey(accountID, groupId)
 
 	entry, err := s.sgKV.Get(ctx, key)
 	if err != nil {
@@ -1397,7 +1397,7 @@ func (s *VPCServiceImpl) updateSGRuleDescriptions(ctx context.Context, accountID
 		return errors.New(awserrors.ErrorInvalidParameterCombination)
 	}
 
-	key := kvutil.AccountKey(accountID, req.groupId)
+	key := kvstore.AccountKey(accountID, req.groupId)
 	entry, err := s.sgKV.Get(ctx, key)
 	if err != nil {
 		return sgNotFoundError(req.groupId)
@@ -1581,7 +1581,7 @@ func (s *VPCServiceImpl) ModifySecurityGroupRules(ctx context.Context, input *ec
 	}
 
 	groupId := *input.GroupId
-	key := kvutil.AccountKey(accountID, groupId)
+	key := kvstore.AccountKey(accountID, groupId)
 
 	entry, err := s.sgKV.Get(ctx, key)
 	if err != nil {
@@ -2158,7 +2158,7 @@ func (s *VPCServiceImpl) storeDefaultSecurityGroup(ctx context.Context, accountI
 	if err != nil {
 		return nil, false, fmt.Errorf("marshal default security group: %w", err)
 	}
-	if _, err := s.sgKV.Create(ctx, kvutil.AccountKey(accountID, groupId), data); err != nil {
+	if _, err := s.sgKV.Create(ctx, kvstore.AccountKey(accountID, groupId), data); err != nil {
 		if errors.Is(err, jetstream.ErrKeyExists) {
 			return record, false, nil
 		}
@@ -2182,7 +2182,7 @@ func (s *VPCServiceImpl) announceDefaultSecurityGroup(record *SecurityGroupRecor
 // CannotDelete guard for default SGs. Used by DeleteVpc to cascade-delete the
 // per-VPC default SG. Surfaces vpcd tear-down failures to the caller.
 func (s *VPCServiceImpl) deleteSecurityGroupInternal(ctx context.Context, accountID, groupId string) error {
-	key := kvutil.AccountKey(accountID, groupId)
+	key := kvstore.AccountKey(accountID, groupId)
 	entry, err := s.sgKV.Get(ctx, key)
 	if err != nil {
 		return fmt.Errorf("read default security group: %w", err)
@@ -2219,7 +2219,7 @@ func (s *VPCServiceImpl) findDefaultSGForVPC(ctx context.Context, accountID, vpc
 		return "", err
 	}
 	for _, k := range keys {
-		if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
 			continue
 		}
 		entry, err := s.sgKV.Get(ctx, k)

spinifex/domains/ec2/vpc/security_group_modify_test.go

@@ -15,7 +15,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -62,7 +62,7 @@ func tcpPeerRequest(port int64, peer string) *ec2.SecurityGroupRuleRequest {
 // modify wrote nothing.
 func storedSGRecord(t *testing.T, svc *VPCServiceImpl, sgID string) []byte {
 	t.Helper()
-	entry, err := svc.sgKV.Get(t.Context(), kvutil.AccountKey(testAccountID, sgID))
+	entry, err := svc.sgKV.Get(t.Context(), kvstore.AccountKey(testAccountID, sgID))
 	require.NoError(t, err)
 	return entry.Value()
 }
@@ -689,7 +689,7 @@ func TestRevokeSecurityGroupIngress_InvertedPortRangeStillRevocable(t *testing.T
 	vpcID := createTestVPC(t, svc, "10.0.0.0/16")
 	sgID := createTestSG(t, svc, vpcID, "stale-inverted-ports")
 
-	key := kvutil.AccountKey(testAccountID, sgID)
+	key := kvstore.AccountKey(testAccountID, sgID)
 	var rec SecurityGroupRecord
 	require.NoError(t, json.Unmarshal(storedSGRecord(t, svc, sgID), &rec))
 	rec.IngressRules = append(rec.IngressRules, SGRule{

spinifex/domains/ec2/vpc/security_group_rule_validation_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -35,7 +35,7 @@ func plantIngressRules(t *testing.T, svc *VPCServiceImpl, sgID string, rules ...
 	rec.IngressRules = append(rec.IngressRules, rules...)
 	data, err := json.Marshal(rec)
 	require.NoError(t, err)
-	_, err = svc.sgKV.Put(t.Context(), kvutil.AccountKey(testAccountID, sgID), data)
+	_, err = svc.sgKV.Put(t.Context(), kvstore.AccountKey(testAccountID, sgID), data)
 	require.NoError(t, err)
 }
 

spinifex/domains/ec2/vpc/security_group_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -1524,7 +1524,7 @@ func findDefaultSGInVPC(t *testing.T, svc *VPCServiceImpl, vpcID string) string
 	require.NoError(t, err)
 	var matches []string
 	for _, k := range keys {
-		if k == kvutil.VersionKey {
+		if k == kvstore.VersionKey {
 			continue
 		}
 		entry, err := svc.sgKV.Get(t.Context(), k)
@@ -2122,7 +2122,7 @@ func seedAvailableVPC(t *testing.T, svc *VPCServiceImpl, vpcID string) {
 	rec := VPCRecord{VpcId: vpcID, CidrBlock: "10.0.0.0/16", State: "available"}
 	data, err := json.Marshal(rec)
 	require.NoError(t, err)
-	_, err = svc.vpcKV.Put(t.Context(), kvutil.AccountKey(testAccountID, vpcID), data)
+	_, err = svc.vpcKV.Put(t.Context(), kvstore.AccountKey(testAccountID, vpcID), data)
 	require.NoError(t, err)
 }
 
@@ -2359,7 +2359,7 @@ func TestAuthorizeSecurityGroupIngress_AssignsRuleIDs(t *testing.T) {
 	}, testAccountID)
 	require.NoError(t, err)
 
-	entry, err := svc.sgKV.Get(t.Context(), kvutil.AccountKey(testAccountID, sgID))
+	entry, err := svc.sgKV.Get(t.Context(), kvstore.AccountKey(testAccountID, sgID))
 	require.NoError(t, err)
 	var record SecurityGroupRecord
 	require.NoError(t, json.Unmarshal(entry.Value(), &record))

spinifex/domains/ec2/vpc/service_impl.go

@@ -26,7 +26,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
 	awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/zoneid"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -173,7 +173,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 		return nil, fmt.Errorf("create JetStream client: %w", err)
 	}
 
-	vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketVPCs, 10)
+	vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketVPCs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVPCs, err)
 	}
@@ -181,7 +181,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketVPCs, err)
 	}
 
-	subnetKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketSubnets, 10)
+	subnetKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketSubnets, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSubnets, err)
 	}
@@ -189,7 +189,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketSubnets, err)
 	}
 
-	vniKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketVNICounter, 10)
+	vniKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketVNICounter, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVNICounter, err)
 	}
@@ -197,7 +197,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketVNICounter, err)
 	}
 
-	eniKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketENIs, 10)
+	eniKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketENIs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketENIs, err)
 	}
@@ -205,7 +205,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketENIs, err)
 	}
 
-	sgKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketSecurityGroups, 10)
+	sgKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketSecurityGroups, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSecurityGroups, err)
 	}
@@ -213,7 +213,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketSecurityGroups, err)
 	}
 
-	defaultsKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketDefaultVPCs, 10)
+	defaultsKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketDefaultVPCs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketDefaultVPCs, err)
 	}
@@ -221,18 +221,18 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketDefaultVPCs, err)
 	}
 
-	rtbKV, err := kvutil.GetOrCreateBucket(ctx, js, "spinifex-vpc-route-tables", 10)
+	rtbKV, err := kvstore.GetOrCreateBucket(ctx, js, "spinifex-vpc-route-tables", 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket spinifex-vpc-route-tables: %w", err)
 	}
 
 	// Named literally, like the route table bucket above: the IGW package
 	// imports this one, so its constant cannot be referenced here.
-	igwKV, err := kvutil.GetOrCreateBucket(ctx, js, "spinifex-igw", 10)
+	igwKV, err := kvstore.GetOrCreateBucket(ctx, js, "spinifex-igw", 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket spinifex-igw: %w", err)
 	}
-	eigwKV, err := kvutil.GetOrCreateBucket(ctx, js, "spinifex-eigw", 10)
+	eigwKV, err := kvstore.GetOrCreateBucket(ctx, js, "spinifex-eigw", 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket spinifex-eigw: %w", err)
 	}
@@ -268,7 +268,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 // nextVNI allocates the next VNI using atomic increment on the NATS KV counter.
 func (s *VPCServiceImpl) nextVNI(ctx context.Context) (int64, error) {
 	var allocated int64
-	_, err := kvutil.Update(ctx, s.vniKV, vniCounterKey, kvutil.CASConfig{CreateIfAbsent: true, Attempts: vniCASAttempts},
+	_, err := kvstore.Update(ctx, s.vniKV, vniCounterKey, kvstore.CASConfig{CreateIfAbsent: true, Attempts: vniCASAttempts},
 		func(next *int64) (bool, error) {
 			if *next == 0 {
 				*next = vniStart
@@ -331,7 +331,7 @@ func (s *VPCServiceImpl) CreateVpc(ctx context.Context, input *ec2.CreateVpcInpu
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal VPC record: %w", err)
 	}
-	if _, err := s.vpcKV.Put(ctx, kvutil.AccountKey(accountID, vpcID), data); err != nil {
+	if _, err := s.vpcKV.Put(ctx, kvstore.AccountKey(accountID, vpcID), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -366,7 +366,7 @@ func (s *VPCServiceImpl) CreateVpc(ctx context.Context, input *ec2.CreateVpcInpu
 // requireVPCExists returns InvalidVpcID.NotFound if the VPC doesn't exist for
 // this account.
 func (s *VPCServiceImpl) requireVPCExists(ctx context.Context, accountID, vpcId string) error {
-	if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcId)); err != nil {
+	if _, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcId)); err != nil {
 		return vpcNotFoundError(vpcId)
 	}
 	return nil
@@ -379,7 +379,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
 	}
 
 	vpcID := *input.VpcId
-	key := kvutil.AccountKey(accountID, vpcID)
+	key := kvstore.AccountKey(accountID, vpcID)
 
 	if _, err := s.vpcKV.Get(ctx, key); err != nil {
 		// AWS-faithful: an absent VPC is NotFound (the tofu/SDK provider
@@ -398,7 +398,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 	for _, k := range subnetKeys {
-		if k == kvutil.VersionKey {
+		if k == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(k, prefix) {
@@ -435,7 +435,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 	for _, k := range sgKeys {
-		if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
 			continue
 		}
 		entry, err := s.sgKV.Get(ctx, k)
@@ -472,7 +472,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 	for _, k := range rtbKeys {
-		if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
 			continue
 		}
 		entry, err := s.rtbKV.Get(ctx, k)
@@ -533,7 +533,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 	if rtbID != "" {
-		if err := s.rtbKV.Delete(ctx, kvutil.AccountKey(accountID, rtbID)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+		if err := s.rtbKV.Delete(ctx, kvstore.AccountKey(accountID, rtbID)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
 			slog.ErrorContext(ctx, "DeleteVpc: main route table reap failed", "vpcId", vpcID, "routeTableId", rtbID, "err", err)
 			return nil, errors.New(awserrors.ErrorServerInternal)
 		}
@@ -574,7 +574,7 @@ func rejectAttachedGateway(ctx context.Context, kv jetstream.KeyValue, accountID
 	}
 
 	for _, k := range keys {
-		if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
 			continue
 		}
 		entry, err := kv.Get(ctx, k)
@@ -653,7 +653,7 @@ func (s *VPCServiceImpl) DescribeVpcs(ctx context.Context, input *ec2.DescribeVp
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, prefix) {
@@ -723,7 +723,7 @@ func (s *VPCServiceImpl) CreateSubnet(ctx context.Context, input *ec2.CreateSubn
 	}
 
 	// Verify VPC exists and belongs to this account
-	vpcEntry, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID))
+	vpcEntry, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcID))
 	if err != nil {
 		return nil, vpcNotFoundError(vpcID)
 	}
@@ -757,7 +757,7 @@ func (s *VPCServiceImpl) CreateSubnet(ctx context.Context, input *ec2.CreateSubn
 	}
 
 	for _, k := range subnetKeys {
-		if k == kvutil.VersionKey {
+		if k == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(k, prefix) {
@@ -822,7 +822,7 @@ func (s *VPCServiceImpl) CreateSubnet(ctx context.Context, input *ec2.CreateSubn
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal subnet record: %w", err)
 	}
-	if _, err := s.subnetKV.Put(ctx, kvutil.AccountKey(accountID, subnetID), data); err != nil {
+	if _, err := s.subnetKV.Put(ctx, kvstore.AccountKey(accountID, subnetID), data); err != nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
@@ -843,7 +843,7 @@ func (s *VPCServiceImpl) DeleteSubnet(ctx context.Context, input *ec2.DeleteSubn
 	}
 
 	subnetID := *input.SubnetId
-	key := kvutil.AccountKey(accountID, subnetID)
+	key := kvstore.AccountKey(accountID, subnetID)
 
 	// Read subnet record before deletion (needed for vpcd event)
 	subnetEntry, err := s.subnetKV.Get(ctx, key)
@@ -902,7 +902,7 @@ func (s *VPCServiceImpl) checkSubnetResidents(ctx context.Context, accountID, su
 	}
 	prefix := accountID + "."
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.eniKV.Get(ctx, key)
@@ -950,7 +950,7 @@ func (s *VPCServiceImpl) clearRouteTableAssociationsForSubnet(ctx context.Contex
 	}
 	prefix := accountID + "."
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.rtbKV.Get(ctx, key)
@@ -1035,7 +1035,7 @@ func (s *VPCServiceImpl) DescribeSubnets(ctx context.Context, input *ec2.Describ
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, prefix) {
@@ -1386,7 +1386,7 @@ func (s *VPCServiceImpl) ModifySubnetAttribute(ctx context.Context, input *ec2.M
 	}
 
 	subnetID := *input.SubnetId
-	key := kvutil.AccountKey(accountID, subnetID)
+	key := kvstore.AccountKey(accountID, subnetID)
 
 	entry, err := s.subnetKV.Get(ctx, key)
 	if err != nil {
@@ -1438,7 +1438,7 @@ func (s *VPCServiceImpl) ModifyVpcAttribute(ctx context.Context, input *ec2.Modi
 	}
 
 	vpcID := *input.VpcId
-	key := kvutil.AccountKey(accountID, vpcID)
+	key := kvstore.AccountKey(accountID, vpcID)
 
 	entry, err := s.vpcKV.Get(ctx, key)
 	if err != nil {
@@ -1486,7 +1486,7 @@ func (s *VPCServiceImpl) DescribeVpcAttribute(ctx context.Context, input *ec2.De
 	}
 
 	vpcID := *input.VpcId
-	key := kvutil.AccountKey(accountID, vpcID)
+	key := kvstore.AccountKey(accountID, vpcID)
 
 	entry, err := s.vpcKV.Get(ctx, key)
 	if err != nil {
@@ -1587,7 +1587,7 @@ func (s *VPCServiceImpl) writeMainRouteTable(ctx context.Context, accountID, vpc
 	if err != nil {
 		return false, fmt.Errorf("marshal main route table: %w", err)
 	}
-	if _, err := s.rtbKV.Create(ctx, kvutil.AccountKey(accountID, rtbID), data); err != nil {
+	if _, err := s.rtbKV.Create(ctx, kvstore.AccountKey(accountID, rtbID), data); err != nil {
 		if errors.Is(err, jetstream.ErrKeyExists) {
 			return false, nil
 		}
@@ -1611,7 +1611,7 @@ func (s *VPCServiceImpl) findMainRouteTableID(ctx context.Context, accountID, vp
 		return "", fmt.Errorf("list rtb keys: %w", err)
 	}
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 		entry, err := s.rtbKV.Get(ctx, key)
@@ -1652,7 +1652,7 @@ func (s *VPCServiceImpl) getDefaultSubnet(ctx context.Context, accountID string)
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, prefix) {
@@ -1680,7 +1680,7 @@ func (s *VPCServiceImpl) GetSubnet(accountID, subnetId string) (*SubnetRecord, e
 }
 
 func (s *VPCServiceImpl) getSubnet(ctx context.Context, accountID, subnetId string) (*SubnetRecord, error) {
-	key := kvutil.AccountKey(accountID, subnetId)
+	key := kvstore.AccountKey(accountID, subnetId)
 	entry, err := s.subnetKV.Get(ctx, key)
 	if err != nil {
 		return nil, fmt.Errorf("subnet %s not found: %w", subnetId, err)
@@ -1726,7 +1726,7 @@ func (s *VPCServiceImpl) SubnetAvailabilityZones(ctx context.Context) ([]string,
 	}
 	seen := map[string]struct{}{}
 	for _, k := range keys {
-		if k == kvutil.VersionKey {
+		if k == kvstore.VersionKey {
 			continue
 		}
 		entry, err := s.subnetKV.Get(ctx, k)

spinifex/domains/ec2/vpc/service_impl_test.go

@@ -16,7 +16,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/zoneid"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
@@ -2689,7 +2689,7 @@ func TestCreateSubnet_ZoneStoreFailureWritesNothingAndPublishesNothing(t *testin
 		require.ErrorIs(t, err, jetstream.ErrNoKeysFound)
 	}
 	for _, k := range keys {
-		assert.Equal(t, kvutil.VersionKey, k, "no subnet record may exist")
+		assert.Equal(t, kvstore.VersionKey, k, "no subnet record may exist")
 	}
 	assert.Zero(t, events.Load(), "no topology event may be published")
 }

spinifex/domains/ecr/auth/ecrauth_test.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/golang-jwt/jwt/v5"
 	"github.com/mulgadc/spinifex/internal/testkit"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -111,7 +111,7 @@ func TestLoadOrCreateSigningKey_FailsWhenNoKeyDecrypts(t *testing.T) {
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "master key does not match")
 
-	entries, err := kvutil.Keys(t.Context(), kv)
+	entries, err := kvstore.Keys(t.Context(), kv)
 	require.NoError(t, err)
 	assert.Len(t, entries, 1, "a failed load must not mint a key alongside the unreadable one")
 }

spinifex/domains/ecr/auth/signingkey.go

@@ -21,7 +21,7 @@ import (
 	"time"
 
 	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -111,7 +111,7 @@ func openSigningBucket(ctx context.Context, js jetstream.JetStream, masterKey []
 	if len(masterKey) == 0 {
 		return nil, errors.New("ecrauth: empty master key")
 	}
-	return kvutil.GetOrCreateBucket(ctx, js, SigningBucket, signingKeyHistory)
+	return kvstore.GetOrCreateBucket(ctx, js, SigningBucket, signingKeyHistory)
 }
 
 // reloadKeys reads every stored signing key, returning the verify set, each
@@ -125,7 +125,7 @@ func openSigningBucket(ctx context.Context, js jetstream.JetStream, masterKey []
 // load instead takes the gateway down on every node, permanently, over a key no
 // live request needs.
 func reloadKeys(ctx context.Context, kv jetstream.KeyValue, masterKey []byte) (active *SigningKey, verify map[string]*ecdsa.PublicKey, metas []keyMeta, undecodable int, err error) {
-	names, err := kvutil.Keys(ctx, kv)
+	names, err := kvstore.Keys(ctx, kv)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
 		return nil, nil, nil, 0, fmt.Errorf("list signing keys: %w", err)
 	}

spinifex/domains/ecr/meta.go

@@ -10,7 +10,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -179,7 +179,7 @@ func (s *KVMetaStore) bucket(ctx context.Context, accountID string) (jetstream.K
 		return kv, nil
 	}
 	name := KVAccountBucket(accountID)
-	kv, err := kvutil.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
+	kv, err := kvstore.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
 	if err != nil {
 		return nil, fmt.Errorf("ecr: open account bucket %s: %w", name, err)
 	}

spinifex/domains/ecs/store.go

@@ -7,7 +7,7 @@ import (
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvlease"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -158,7 +158,7 @@ func accountBuckets(ctx context.Context, nc *nats.Conn) ([]accountBucket, error)
 	if err != nil {
 		return nil, fmt.Errorf("jetstream: %w", err)
 	}
-	names, err := kvutil.BucketNames(ctx, js)
+	names, err := kvstore.BucketNames(ctx, js)
 	if err != nil {
 		return nil, err
 	}
@@ -176,7 +176,7 @@ func accountBuckets(ctx context.Context, nc *nats.Conn) ([]accountBucket, error)
 // return the existing handle.
 func GetOrCreateAccountBucket(ctx context.Context, js jetstream.JetStream, accountID string) (jetstream.KeyValue, error) {
 	bucket := AccountBucketName(accountID)
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, bucket, KVBucketECSAccountHistory)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, bucket, KVBucketECSAccountHistory)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create ECS per-account KV bucket %s: %w", bucket, err)
 	}

spinifex/domains/eks/access/store.go

@@ -10,7 +10,7 @@ import (
 	"slices"
 	"strings"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -128,7 +128,7 @@ func deleteRecord(ctx context.Context, kv jetstream.KeyValue, cluster, principal
 // casUpdate does a revision-checked read-modify-write. mutate returns true
 // when a field changed. Returns ErrNotFound if absent.
 func casUpdate(ctx context.Context, kv jetstream.KeyValue, cluster, principalARN string, mutate func(*Record) bool) (*Record, error) {
-	return kvutil.Update(ctx, kv, Key(cluster, principalARN), kvutil.CASConfig{NotFound: ErrNotFound}, func(r *Record) (bool, error) {
+	return kvstore.Update(ctx, kv, Key(cluster, principalARN), kvstore.CASConfig{NotFound: ErrNotFound}, func(r *Record) (bool, error) {
 		return mutate(r), nil
 	})
 }

spinifex/domains/eks/addon/store.go

@@ -8,7 +8,7 @@ import (
 	"slices"
 	"strings"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -193,7 +193,7 @@ func deleteRecord(ctx context.Context, kv jetstream.KeyValue, cluster, addon str
 // casUpdate does a revision-checked read-modify-write. mutate returns true
 // when a field changed. Returns ErrNotFound if absent.
 func casUpdate(ctx context.Context, kv jetstream.KeyValue, cluster, addon string, mutate func(*Record) bool) (*Record, error) {
-	return kvutil.Update(ctx, kv, Key(cluster, addon), kvutil.CASConfig{NotFound: ErrNotFound}, func(r *Record) (bool, error) {
+	return kvstore.Update(ctx, kv, Key(cluster, addon), kvstore.CASConfig{NotFound: ErrNotFound}, func(r *Record) (bool, error) {
 		return mutate(r), nil
 	})
 }

spinifex/domains/eks/cluster/cluster_state.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -280,7 +280,7 @@ func casUpdateMeta(ctx context.Context, kv jetstream.KeyValue, name string, muta
 	if name == "" {
 		return errors.New("eks: casUpdateMeta empty name")
 	}
-	_, err := kvutil.Update(ctx, kv, ClusterMetaKey(name), kvutil.CASConfig{
+	_, err := kvstore.Update(ctx, kv, ClusterMetaKey(name), kvstore.CASConfig{
 		Attempts: maxClusterStateCASRetries,
 		NotFound: ErrClusterNotFound,
 		Exhausted: func(string, int) error {
@@ -435,7 +435,7 @@ func putIfGeneration(ctx context.Context, kv jetstream.KeyValue, meta *ClusterMe
 		if err == nil {
 			return nil
 		}
-		if !kvutil.IsRevisionConflict(err) {
+		if !kvstore.IsRevisionConflict(err) {
 			return fmt.Errorf("kv write %s: %w", key, err)
 		}
 	}

spinifex/domains/eks/cluster/cluster_teardown.go

@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvlease"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	telemetry "github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -292,7 +292,7 @@ func (s *EKSServiceImpl) removeClusterRecord(ctx context.Context, acctKV jetstre
 	}
 
 	errRefsAdded := errors.New("refs added")
-	_, err = kvutil.DeleteIf(ctx, acctKV, ClusterMetaKey(name), kvutil.CASConfig{
+	_, err = kvstore.DeleteIf(ctx, acctKV, ClusterMetaKey(name), kvstore.CASConfig{
 		Attempts: maxClusterStateCASRetries,
 		NotFound: ErrClusterNotFound,
 	}, func(m *ClusterMeta) error {

spinifex/domains/eks/cluster/cp_orphans_test.go

@@ -15,7 +15,7 @@ import (
 	ec2inventory "github.com/mulgadc/spinifex/spinifex/domains/ec2/inventory"
 	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -150,13 +150,13 @@ func TestReportControlPlaneOrphansPrintsRows(t *testing.T) {
 func TestFindControlPlaneOrphansCorruptENIRecordIsUnreadable(t *testing.T) {
 	f := newEKSServiceFixture(t)
 	js := testutil.NewJetStream(t, f.svc.deps.NATSConn)
-	enis, err := kvutil.GetOrCreateBucket(t.Context(), js, ec2vpc.KVBucketENIs, 10)
+	enis, err := kvstore.GetOrCreateBucket(t.Context(), js, ec2vpc.KVBucketENIs, 10)
 	require.NoError(t, err)
 	good, err := json.Marshal(ec2vpc.ENIRecord{NetworkInterfaceId: "eni-good", Tags: cpTags("gone", testAccountID)})
 	require.NoError(t, err)
-	_, err = enis.Put(t.Context(), kvutil.AccountKey("000000000000", "eni-good"), good)
+	_, err = enis.Put(t.Context(), kvstore.AccountKey("000000000000", "eni-good"), good)
 	require.NoError(t, err)
-	_, err = enis.Put(t.Context(), kvutil.AccountKey("000000000000", "eni-corrupt"), []byte("{not json"))
+	_, err = enis.Put(t.Context(), kvstore.AccountKey("000000000000", "eni-corrupt"), []byte("{not json"))
 	require.NoError(t, err)
 	res := []*ec2.Reservation{{Instances: []*ec2.Instance{
 		orphanInstance("i-good", "eni-good"), orphanInstance("i-corrupt", "eni-corrupt"), orphanInstance("i-absent", "eni-absent"),

spinifex/domains/eks/cluster/eks_billable_reaper_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/elbv2"
 	testutil "github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
@@ -69,7 +69,7 @@ func TestS7_EKSBillableReaperNeverTerminatesAControlPlane(t *testing.T) {
 			account: testAccountID,
 			prepare: func(t *testing.T, f *eksServiceFixture) {
 				require.NoError(t, PutClusterMeta(t.Context(), f.kv, sampleClusterMeta("orphan")))
-				_, err := kvutil.DeleteIf(t.Context(), f.kv, ClusterMetaKey("orphan"), kvutil.CASConfig{
+				_, err := kvstore.DeleteIf(t.Context(), f.kv, ClusterMetaKey("orphan"), kvstore.CASConfig{
 					Attempts: maxClusterStateCASRetries,
 					NotFound: ErrClusterNotFound,
 				}, func(*ClusterMeta) error { return nil })

spinifex/domains/eks/cluster/service_impl.go

@@ -33,7 +33,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
 	"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -2138,7 +2138,7 @@ func (s *EKSServiceImpl) casUpdateNodegroupTags(ctx context.Context, accountID,
 	if err != nil {
 		return err
 	}
-	_, err = kvutil.Update(ctx, acctKV, NodegroupKey(cluster, ng), kvutil.CASConfig{
+	_, err = kvstore.Update(ctx, acctKV, NodegroupKey(cluster, ng), kvstore.CASConfig{
 		Attempts:  ngCASMaxRetries,
 		NotFound:  errors.New(awserrors.ErrorEKSResourceNotFound),
 		Exhausted: func(string, int) error { return errors.New(awserrors.ErrorServerInternal) },

spinifex/domains/eks/cluster/store.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvlease"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -157,7 +156,7 @@ func accountBucketNames(ctx context.Context, nc *nats.Conn) ([]string, error) {
 	if err != nil {
 		return nil, fmt.Errorf("jetstream: %w", err)
 	}
-	all, err := kvutil.BucketNames(ctx, js)
+	all, err := kvstore.BucketNames(ctx, js)
 	if err != nil {
 		return nil, err
 	}
@@ -197,7 +196,7 @@ func AccountWatchBuckets(ctx context.Context, nc *nats.Conn) ([]*kvstore.Bucket,
 // existing handle.
 func GetOrCreateAccountBucket(ctx context.Context, js jetstream.JetStream, accountID string) (jetstream.KeyValue, error) {
 	bucket := AccountBucketName(accountID)
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, bucket, KVBucketEKSAccountHistory)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, bucket, KVBucketEKSAccountHistory)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create EKS per-account KV bucket %s: %w", bucket, err)
 	}

spinifex/domains/eks/cluster/watch_source_test.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/mulgadc/spinifex/internal/testkit"
 	ekscluster "github.com/mulgadc/spinifex/spinifex/domains/eks/cluster"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -24,7 +24,7 @@ func TestAccountWatchBuckets_ReturnsOnePerAccountBucket(t *testing.T) {
 	require.NoError(t, err)
 	_, err = ekscluster.GetOrCreateAccountBucket(t.Context(), js, "222222222222")
 	require.NoError(t, err)
-	_, err = kvutil.GetOrCreateBucket(t.Context(), js, "something-else", 1)
+	_, err = kvstore.GetOrCreateBucket(t.Context(), js, "something-else", 1)
 	require.NoError(t, err)
 
 	buckets, err := ekscluster.AccountWatchBuckets(t.Context(), nc)

spinifex/domains/elbv2/lb_health.go

@@ -10,7 +10,7 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -94,7 +94,7 @@ func (s *Store) GetLBHealth(ctx context.Context, lbID string) (*LBDataPathRecord
 // UpdateLBHealth applies mutate to the evidence record under optimistic
 // concurrency, creating it when absent. mutate returns false to write nothing.
 func (s *Store) UpdateLBHealth(ctx context.Context, lbID string, mutate func(*LBDataPathRecord) (bool, error)) (*LBDataPathRecord, error) {
-	return kvutil.Update(ctx, s.kv, KeyPrefixLBHealth+lbID, kvutil.CASConfig{
+	return kvstore.Update(ctx, s.kv, KeyPrefixLBHealth+lbID, kvstore.CASConfig{
 		Attempts:       targetGroupCASAttempts,
 		CreateIfAbsent: true,
 		Exhausted: func(_ string, attempts int) error {

spinifex/domains/elbv2/store.go

@@ -11,7 +11,6 @@ import (
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -35,7 +34,7 @@ const (
 	maxLBNameClaimRetries = 5
 
 	// targetGroupCASAttempts bounds a target-group read-modify-write. It sits
-	// above the kvutil default because Terraform registers targets at its own
+	// above the kvstore default because Terraform registers targets at its own
 	// parallelism (10 by default), every one contending for the same record.
 	targetGroupCASAttempts = 12
 
@@ -90,7 +89,7 @@ func NewStore(ctx context.Context, nc *nats.Conn) (*Store, error) {
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketELBv2, 1)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketELBv2, 1)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketELBv2, err)
 	}
@@ -189,7 +188,7 @@ var ErrLoadBalancerNotFound = errors.New("elbv2: load balancer not found")
 // concurrency: a writer that loses the race re-reads and retries instead of
 // overwriting the winner. mutate returns false to commit nothing.
 func (s *Store) UpdateLoadBalancer(ctx context.Context, lbID string, mutate func(*LoadBalancerRecord) (bool, error)) (*LoadBalancerRecord, error) {
-	return kvutil.Update(ctx, s.kv, KeyPrefixLB+lbID, kvutil.CASConfig{
+	return kvstore.Update(ctx, s.kv, KeyPrefixLB+lbID, kvstore.CASConfig{
 		Attempts: targetGroupCASAttempts,
 		NotFound: ErrLoadBalancerNotFound,
 		Exhausted: func(_ string, attempts int) error {
@@ -294,7 +293,7 @@ var ErrTargetGroupNotFound = errors.New("elbv2: target group not found")
 // concurrency: a writer that loses the race re-reads and retries instead of
 // overwriting the winner. mutate returns false to commit nothing.
 func (s *Store) UpdateTargetGroup(ctx context.Context, tgID string, mutate func(*TargetGroupRecord) (bool, error)) (*TargetGroupRecord, error) {
-	return kvutil.Update(ctx, s.kv, KeyPrefixTG+tgID, kvutil.CASConfig{
+	return kvstore.Update(ctx, s.kv, KeyPrefixTG+tgID, kvstore.CASConfig{
 		Attempts: targetGroupCASAttempts,
 		NotFound: ErrTargetGroupNotFound,
 		Exhausted: func(_ string, attempts int) error {
@@ -652,7 +651,7 @@ func listByPrefixOpt[T any](ctx context.Context, kv jetstream.KeyValue, prefix s
 
 	var result []*T
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, prefix) {

spinifex/domains/iam/account_names.go

@@ -9,7 +9,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -41,7 +41,7 @@ type AccountNameIndex struct {
 // NewAccountNameIndex opens (or creates) the reservation bucket at the
 // cluster's replica count.
 func NewAccountNameIndex(ctx context.Context, js jetstream.JetStream) (*AccountNameIndex, error) {
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketAccountNames, 5)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccountNames, 5)
 	if err != nil {
 		return nil, fmt.Errorf("init account names bucket: %w", err)
 	}

spinifex/domains/iam/groups.go

@@ -16,7 +16,7 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 )
 
 // maxGroupsPerUser caps how many groups a single user may belong to. Mirrors the
@@ -106,7 +106,7 @@ func (s *IAMServiceImpl) GetGroup(accountID string, input *iam.GetGroupInput) (*
 
 func (s *IAMServiceImpl) ListGroups(accountID string, input *iam.ListGroupsInput) (*iam.ListGroupsOutput, error) {
 	ctx := context.Background()
-	keys, err := kvutil.Keys(ctx, s.groupsBucket)
+	keys, err := kvstore.Keys(ctx, s.groupsBucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return &iam.ListGroupsOutput{
@@ -127,7 +127,7 @@ func (s *IAMServiceImpl) ListGroups(accountID string, input *iam.ListGroupsInput
 	// element at all rather than an empty one.
 	groups := []*iam.Group{}
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, keyPrefix) {
@@ -456,7 +456,7 @@ func (s *IAMServiceImpl) getGroup(ctx context.Context, accountID, groupName stri
 // updateUserCAS.
 func (s *IAMServiceImpl) updateGroupCAS(ctx context.Context, accountID, groupName string, mutate func(*Group) (bool, error)) error {
 	defer s.recordLocks.lock("group/" + accountID + "." + groupName)()
-	_, err := kvutil.Update(ctx, s.groupsBucket, accountID+"."+groupName, iamRecordCASConfig("group", accountID, groupName), mutate)
+	_, err := kvstore.Update(ctx, s.groupsBucket, accountID+"."+groupName, iamRecordCASConfig("group", accountID, groupName), mutate)
 	return err
 }
 
@@ -465,7 +465,7 @@ func (s *IAMServiceImpl) updateGroupCAS(ctx context.Context, accountID, groupNam
 // unmarshal errors so DeleteGroup cannot succeed while a real-but-unreadable
 // member exists. Mirrors findInstanceProfilesForRole.
 func (s *IAMServiceImpl) findGroupMembers(ctx context.Context, accountID, groupName string) ([]*User, error) {
-	keys, err := kvutil.Keys(ctx, s.usersBucket)
+	keys, err := kvstore.Keys(ctx, s.usersBucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return nil, nil
@@ -476,7 +476,7 @@ func (s *IAMServiceImpl) findGroupMembers(ctx context.Context, accountID, groupN
 	keyPrefix := accountID + "."
 	var members []*User
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, keyPrefix) {

spinifex/domains/iam/instance_profiles.go

@@ -17,7 +17,7 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 )
 
 // Bound on optimistic-concurrency retries when a concurrent writer wins the
@@ -97,7 +97,7 @@ func (s *IAMServiceImpl) GetInstanceProfile(accountID string, input *iam.GetInst
 
 func (s *IAMServiceImpl) ListInstanceProfiles(accountID string, input *iam.ListInstanceProfilesInput) (*iam.ListInstanceProfilesOutput, error) {
 	ctx := context.Background()
-	keys, err := kvutil.Keys(ctx, s.instanceProfilesBucket)
+	keys, err := kvstore.Keys(ctx, s.instanceProfilesBucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return &iam.ListInstanceProfilesOutput{
@@ -118,7 +118,7 @@ func (s *IAMServiceImpl) ListInstanceProfiles(accountID string, input *iam.ListI
 	// to no element at all rather than an empty one.
 	profiles := []*iam.InstanceProfile{}
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, keyPrefix) {
@@ -169,7 +169,7 @@ func (s *IAMServiceImpl) DeleteInstanceProfile(accountID string, input *iam.Dele
 	profileName := *input.InstanceProfileName
 
 	// Revision-guarded, so a role attached after the check sends the delete back through it.
-	_, err := kvutil.DeleteIf(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName),
+	_, err := kvstore.DeleteIf(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName),
 		func(p *InstanceProfile) error {
 			if p.RoleName != "" {
 				return awserrors.Errorf(awserrors.ErrorIAMDeleteConflict, "Cannot delete entity, must remove roles from instance profile first.")
@@ -378,12 +378,12 @@ func (s *IAMServiceImpl) getInstanceProfile(ctx context.Context, accountID, prof
 // mutate reports whether it changed the record; a false return commits nothing.
 func (s *IAMServiceImpl) updateInstanceProfileCAS(ctx context.Context, accountID, profileName string, mutate func(*InstanceProfile) (bool, error)) error {
 	defer s.recordLocks.lock("instance-profile/" + accountID + "." + profileName)()
-	_, err := kvutil.Update(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName), mutate)
+	_, err := kvstore.Update(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName), mutate)
 	return err
 }
 
-func (s *IAMServiceImpl) instanceProfileCASConfig(accountID, profileName string) kvutil.CASConfig {
+func (s *IAMServiceImpl) instanceProfileCASConfig(accountID, profileName string) kvstore.CASConfig {
-	return kvutil.CASConfig{
+	return kvstore.CASConfig{
 		Attempts: instanceProfileCASMaxRetries,
 		NotFound: instanceProfileNotFound(profileName),
 		Exhausted: func(string, int) error {

spinifex/domains/iam/oidc_providers.go

@@ -20,7 +20,7 @@ import (
 
 	resourcearn "github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 )
 
 // Per-account IAM bucket holds account-scoped resources (currently the OIDC provider registry).
@@ -52,7 +52,7 @@ func OIDCProviderARN(accountID, issuerHostPath string) string {
 
 // GetOrCreateIAMAccountBucket opens the per-account IAM bucket, creating it on first use.
 func GetOrCreateIAMAccountBucket(ctx context.Context, js jetstream.JetStream, accountID string) (jetstream.KeyValue, error) {
-	return kvutil.GetOrCreateBucket(ctx, js, IAMAccountBucketName(accountID), 1)
+	return kvstore.GetOrCreateBucket(ctx, js, IAMAccountBucketName(accountID), 1)
 }
 
 // OIDCProviderRecord is the stored shape of a registered OIDC identity provider.
@@ -224,7 +224,7 @@ func (s *IAMServiceImpl) ListOpenIDConnectProviders(accountID string, _ *iam.Lis
 		return nil, fmt.Errorf("open IAM account bucket: %w", err)
 	}
 
-	keys, err := kvutil.Keys(ctx, kv)
+	keys, err := kvstore.Keys(ctx, kv)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return out, nil
@@ -233,7 +233,7 @@ func (s *IAMServiceImpl) ListOpenIDConnectProviders(accountID string, _ *iam.Lis
 	}
 
 	for _, key := range keys {
-		if key == kvutil.VersionKey || !strings.HasPrefix(key, oidcProvidersKeyPrefix) {
+		if key == kvstore.VersionKey || !strings.HasPrefix(key, oidcProvidersKeyPrefix) {
 			continue
 		}
 		entry, err := kv.Get(ctx, key)
@@ -356,7 +356,7 @@ func (s *IAMServiceImpl) updateOIDCProviderCAS(ctx context.Context, accountID, a
 		return fmt.Errorf("open IAM account bucket: %w", err)
 	}
 	defer s.recordLocks.lock("oidc-provider/" + accountID + "/" + OIDCProviderKey(issuer))()
-	_, err = kvutil.Update(ctx, kv, OIDCProviderKey(issuer), iamRecordCASConfig("oidc-provider", accountID, arn), mutate)
+	_, err = kvstore.Update(ctx, kv, OIDCProviderKey(issuer), iamRecordCASConfig("oidc-provider", accountID, arn), mutate)
 	return err
 }
 

spinifex/domains/iam/policy_versions.go

@@ -16,7 +16,7 @@ import (
 
 	iamarn "github.com/mulgadc/bluebottle/pkg/auth"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 )
 
 const (
@@ -94,7 +94,7 @@ func (s *IAMServiceImpl) updatePolicyCAS(ctx context.Context, accountID, policyA
 		return err
 	}
 	defer s.recordLocks.lock("policy/" + key)()
-	_, err = kvutil.Update(ctx, s.policiesBucket, key, cfg, func(p *Policy) (bool, error) {
+	_, err = kvstore.Update(ctx, s.policiesBucket, key, cfg, func(p *Policy) (bool, error) {
 		// The key omits the path, so an ARN with the wrong path names no policy.
 		if p.ARN != policyARN {
 			return false, errors.New(awserrors.ErrorIAMNoSuchEntity)
@@ -106,14 +106,14 @@ func (s *IAMServiceImpl) updatePolicyCAS(ctx context.Context, accountID, policyA
 
 // policyCASTarget resolves policyARN to its KV key and the CAS settings every
 // write to a policy record shares.
-func policyCASTarget(accountID, policyARN string) (string, kvutil.CASConfig, error) {
+func policyCASTarget(accountID, policyARN string) (string, kvstore.CASConfig, error) {
 	_, policyName, err := iamarn.ParsePolicyARN(policyARN)
 	if err != nil {
 		slog.Debug("policyCASTarget: unparseable policy ARN",
 			"accountID", accountID, "policyArn", policyARN, "err", err)
-		return "", kvutil.CASConfig{}, policyNotFound(policyARN)
+		return "", kvstore.CASConfig{}, policyNotFound(policyARN)
 	}
-	return accountID + "." + policyName, kvutil.CASConfig{
+	return accountID + "." + policyName, kvstore.CASConfig{
 		Attempts: policyCASMaxRetries,
 		NotFound: policyNotFound(policyARN),
 		Exhausted: func(string, int) error {

spinifex/domains/iam/roles.go

@@ -18,7 +18,7 @@ import (
 	iamarn "github.com/mulgadc/bluebottle/pkg/auth"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 )
 
 const (
@@ -128,7 +128,7 @@ func (s *IAMServiceImpl) GetRole(accountID string, input *iam.GetRoleInput) (*ia
 
 func (s *IAMServiceImpl) ListRoles(accountID string, input *iam.ListRolesInput) (*iam.ListRolesOutput, error) {
 	ctx := context.Background()
-	keys, err := kvutil.Keys(ctx, s.rolesBucket)
+	keys, err := kvstore.Keys(ctx, s.rolesBucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return &iam.ListRolesOutput{
@@ -149,7 +149,7 @@ func (s *IAMServiceImpl) ListRoles(accountID string, input *iam.ListRolesInput)
 	// element at all rather than an empty one.
 	roles := []*iam.Role{}
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, keyPrefix) {
@@ -574,7 +574,7 @@ func (s *IAMServiceImpl) getRole(ctx context.Context, accountID, roleName string
 // whether it changed the record; a false return commits nothing.
 func (s *IAMServiceImpl) updateRoleCAS(ctx context.Context, accountID, roleName string, mutate func(*Role) (bool, error)) error {
 	defer s.recordLocks.lock("role/" + accountID + "." + roleName)()
-	_, err := kvutil.Update(ctx, s.rolesBucket, accountID+"."+roleName, kvutil.CASConfig{
+	_, err := kvstore.Update(ctx, s.rolesBucket, accountID+"."+roleName, kvstore.CASConfig{
 		Attempts: roleCASMaxRetries,
 		NotFound: noSuchEntity("role", roleName),
 		Exhausted: func(string, int) error {
@@ -593,7 +593,7 @@ func (s *IAMServiceImpl) updateRoleCAS(ctx context.Context, accountID, roleName
 // per-key Get or unmarshal errors so DeleteRole cannot succeed while a
 // real-but-unreadable reference exists.
 func (s *IAMServiceImpl) findInstanceProfilesForRole(ctx context.Context, accountID, roleName string) ([]*InstanceProfile, error) {
-	keys, err := kvutil.Keys(ctx, s.instanceProfilesBucket)
+	keys, err := kvstore.Keys(ctx, s.instanceProfilesBucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return nil, nil
@@ -604,7 +604,7 @@ func (s *IAMServiceImpl) findInstanceProfilesForRole(ctx context.Context, accoun
 	keyPrefix := accountID + "."
 	var profiles []*InstanceProfile
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, keyPrefix) {

spinifex/domains/iam/service_impl.go

@@ -29,7 +29,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	awscredentials "github.com/mulgadc/spinifex/spinifex/foundation/aws/credentials"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 )
 
@@ -143,7 +143,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
 		return nil, fmt.Errorf("get JetStream context: %w", err)
 	}
 
-	usersBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketUsers, 10)
+	usersBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketUsers, 10)
 	if err != nil {
 		return nil, fmt.Errorf("init users bucket: %w", err)
 	}
@@ -151,7 +151,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketUsers, err)
 	}
 
-	accessKeysBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketAccessKeys, 5)
+	accessKeysBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccessKeys, 5)
 	if err != nil {
 		return nil, fmt.Errorf("init access keys bucket: %w", err)
 	}
@@ -159,7 +159,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketAccessKeys, err)
 	}
 
-	policiesBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketPolicies, 10)
+	policiesBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketPolicies, 10)
 	if err != nil {
 		return nil, fmt.Errorf("init policies bucket: %w", err)
 	}
@@ -167,7 +167,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketPolicies, err)
 	}
 
-	accountsBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketAccounts, 5)
+	accountsBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccounts, 5)
 	if err != nil {
 		return nil, fmt.Errorf("init accounts bucket: %w", err)
 	}
@@ -175,7 +175,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketAccounts, err)
 	}
 
-	accountCounterBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketAccountCounter, 5)
+	accountCounterBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccountCounter, 5)
 	if err != nil {
 		return nil, fmt.Errorf("init account counter bucket: %w", err)
 	}
@@ -183,7 +183,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketAccountCounter, err)
 	}
 
-	rolesBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketRoles, 10)
+	rolesBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketRoles, 10)
 	if err != nil {
 		return nil, fmt.Errorf("init roles bucket: %w", err)
 	}
@@ -191,7 +191,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketRoles, err)
 	}
 
-	instanceProfilesBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketInstanceProfiles, 10)
+	instanceProfilesBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketInstanceProfiles, 10)
 	if err != nil {
 		return nil, fmt.Errorf("init instance profiles bucket: %w", err)
 	}
@@ -199,7 +199,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketInstanceProfiles, err)
 	}
 
-	groupsBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketGroups, 10)
+	groupsBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketGroups, 10)
 	if err != nil {
 		return nil, fmt.Errorf("init groups bucket: %w", err)
 	}
@@ -508,7 +508,7 @@ func (s *IAMServiceImpl) GetUser(accountID string, input *iam.GetUserInput) (*ia
 
 func (s *IAMServiceImpl) ListUsers(accountID string, input *iam.ListUsersInput) (*iam.ListUsersOutput, error) {
 	ctx := context.Background()
-	keys, err := kvutil.Keys(ctx, s.usersBucket)
+	keys, err := kvstore.Keys(ctx, s.usersBucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return &iam.ListUsersOutput{
@@ -529,7 +529,7 @@ func (s *IAMServiceImpl) ListUsers(accountID string, input *iam.ListUsersInput)
 	// element at all rather than an empty one.
 	users := []*iam.User{}
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, keyPrefix) {
@@ -732,7 +732,7 @@ func (s *IAMServiceImpl) setAccessKeysOwner(ctx context.Context, accountID strin
 	for _, keyID := range keyIDs {
 		cfg := iamRecordCASConfig("access key", accountID, keyID)
 		cfg.NotFound = nil
-		_, err := kvutil.Update(ctx, s.accessKeysBucket, keyID, cfg, func(ak *AccessKey) (bool, error) {
+		_, err := kvstore.Update(ctx, s.accessKeysBucket, keyID, cfg, func(ak *AccessKey) (bool, error) {
 			if ak.UserName == userName {
 				return false, nil
 			}
@@ -1189,7 +1189,7 @@ func (s *IAMServiceImpl) seedAdminAccount(ctx context.Context, admin *AdminBoots
 // IsEmpty returns true if the users bucket has no entries.
 func (s *IAMServiceImpl) IsEmpty() (bool, error) {
 	ctx := context.Background()
-	keys, err := kvutil.Keys(ctx, s.usersBucket)
+	keys, err := kvstore.Keys(ctx, s.usersBucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return true, nil
@@ -1197,7 +1197,7 @@ func (s *IAMServiceImpl) IsEmpty() (bool, error) {
 		return false, fmt.Errorf("check users bucket: %w", err)
 	}
 	for _, key := range keys {
-		if key != kvutil.VersionKey {
+		if key != kvstore.VersionKey {
 			return false, nil
 		}
 	}
@@ -1303,7 +1303,7 @@ func (s *IAMServiceImpl) GetAccount(accountID string) (*Account, error) {
 // ListAccounts returns all accounts.
 func (s *IAMServiceImpl) ListAccounts() ([]*Account, error) {
 	ctx := context.Background()
-	keys, err := kvutil.Keys(ctx, s.accountsBucket)
+	keys, err := kvstore.Keys(ctx, s.accountsBucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return []*Account{}, nil
@@ -1313,7 +1313,7 @@ func (s *IAMServiceImpl) ListAccounts() ([]*Account, error) {
 
 	accounts := make([]*Account, 0, len(keys))
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		entry, err := s.accountsBucket.Get(ctx, key)
@@ -1367,8 +1367,8 @@ func (s *IAMServiceImpl) QuiesceAccount(accountID, operationID string, fence uin
 		return 0, fmt.Errorf("%s (%s) cannot be %s: %w", accountID, reason, AccountStatusTerminating, ErrAccountUndeletable)
 	}
 
-	account, err := kvutil.Update(context.Background(), s.accountsBucket, accountID,
+	account, err := kvstore.Update(context.Background(), s.accountsBucket, accountID,
-		kvutil.CASConfig{NotFound: errors.New(awserrors.ErrorIAMNoSuchEntity)},
+		kvstore.CASConfig{NotFound: errors.New(awserrors.ErrorIAMNoSuchEntity)},
 		func(a *Account) (bool, error) {
 			if a.Status != AccountStatusTerminating {
 				a.Status = AccountStatusTerminating
@@ -1390,7 +1390,7 @@ func (s *IAMServiceImpl) QuiesceAccount(accountID, operationID string, fence uin
 			return false, nil
 		})
 	if err != nil {
-		if errors.Is(err, kvutil.ErrWrite) || errors.Is(err, kvutil.ErrRead) || errors.Is(err, kvutil.ErrDecode) {
+		if errors.Is(err, kvstore.ErrWrite) || errors.Is(err, kvstore.ErrRead) || errors.Is(err, kvstore.ErrDecode) {
 			return 0, fmt.Errorf("store account quiesce: %w", err)
 		}
 		return 0, err
@@ -1421,8 +1421,8 @@ func (s *IAMServiceImpl) SetAccountStatus(accountID, status string) (*Account, e
 		return nil, fmt.Errorf("account %s: %w", accountID, ErrUseQuiesceAccount)
 	}
 
-	account, err := kvutil.Update(context.Background(), s.accountsBucket, accountID,
+	account, err := kvstore.Update(context.Background(), s.accountsBucket, accountID,
-		kvutil.CASConfig{NotFound: errors.New(awserrors.ErrorIAMNoSuchEntity)},
+		kvstore.CASConfig{NotFound: errors.New(awserrors.ErrorIAMNoSuchEntity)},
 		func(a *Account) (bool, error) {
 			if a.Status == AccountStatusTerminating && status != AccountStatusTerminating {
 				return false, fmt.Errorf("account %s is terminating: %w", accountID, ErrAccountUndeletable)
@@ -1435,7 +1435,7 @@ func (s *IAMServiceImpl) SetAccountStatus(accountID, status string) (*Account, e
 			return true, nil
 		})
 	if err != nil {
-		if errors.Is(err, kvutil.ErrWrite) || errors.Is(err, kvutil.ErrRead) || errors.Is(err, kvutil.ErrDecode) {
+		if errors.Is(err, kvstore.ErrWrite) || errors.Is(err, kvstore.ErrRead) || errors.Is(err, kvstore.ErrDecode) {
 			return nil, fmt.Errorf("store account status: %w", err)
 		}
 		return nil, err
@@ -1606,7 +1606,7 @@ func (s *IAMServiceImpl) ListPolicies(accountID string, input *iam.ListPoliciesI
 	if err := checkEnum("policyUsageFilter", input.PolicyUsageFilter, policyUsageValues); err != nil {
 		return nil, err
 	}
-	keys, err := kvutil.Keys(ctx, s.policiesBucket)
+	keys, err := kvstore.Keys(ctx, s.policiesBucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return &iam.ListPoliciesOutput{
@@ -1628,7 +1628,7 @@ func (s *IAMServiceImpl) ListPolicies(accountID string, input *iam.ListPoliciesI
 	onlyAttached := aws.BoolValue(input.OnlyAttached)
 	policies := []*iam.Policy{}
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(key, keyPrefix) {
@@ -1712,7 +1712,7 @@ func (s *IAMServiceImpl) DeletePolicy(accountID string, input *iam.DeletePolicyI
 	}
 
 	// Revision-guarded, so a version created after the checks cannot be deleted unseen.
-	policy, err := kvutil.DeleteIf(ctx, s.policiesBucket, key, cfg, func(p *Policy) error {
+	policy, err := kvstore.DeleteIf(ctx, s.policiesBucket, key, cfg, func(p *Policy) error {
 		if p.ARN != policyARN {
 			return policyNotFound(policyARN)
 		}
@@ -2267,7 +2267,7 @@ func (s *IAMServiceImpl) buildPolicyAttachments(ctx context.Context, accountID s
 
 	keyPrefix := accountID + "."
 	for _, scan := range scans {
-		keys, err := kvutil.Keys(ctx, scan.bucket)
+		keys, err := kvstore.Keys(ctx, scan.bucket)
 		if err != nil {
 			if errors.Is(err, jetstream.ErrNoKeysFound) {
 				continue
@@ -2276,7 +2276,7 @@ func (s *IAMServiceImpl) buildPolicyAttachments(ctx context.Context, accountID s
 		}
 
 		for _, key := range keys {
-			if key == kvutil.VersionKey {
+			if key == kvstore.VersionKey {
 				continue
 			}
 			if !strings.HasPrefix(key, keyPrefix) {
@@ -2387,18 +2387,18 @@ func (s *IAMServiceImpl) getUser(ctx context.Context, accountID, userName string
 // restore a policy or group membership that a concurrent writer removed.
 func (s *IAMServiceImpl) updateUserCAS(ctx context.Context, accountID, userName string, mutate func(*User) (bool, error)) error {
 	defer s.recordLocks.lock("user/" + accountID + "." + userName)()
-	_, err := kvutil.Update(ctx, s.usersBucket, accountID+"."+userName, iamRecordCASConfig("user", accountID, userName), mutate)
+	_, err := kvstore.Update(ctx, s.usersBucket, accountID+"."+userName, iamRecordCASConfig("user", accountID, userName), mutate)
 	return err
 }
 
 // iamRecordCASConfig is the CAS policy for an IAM user, group or OIDC provider
 // record: a missing record is NoSuchEntity and exhausted retries are logged.
-func iamRecordCASConfig(kind, accountID, name string) kvutil.CASConfig {
+func iamRecordCASConfig(kind, accountID, name string) kvstore.CASConfig {
 	notFound := noSuchEntity(kind, name)
 	if kind == "oidc-provider" {
 		notFound = oidcProviderNotFound(name)
 	}
-	return kvutil.CASConfig{
+	return kvstore.CASConfig{
 		Attempts: roleCASMaxRetries,
 		NotFound: notFound,
 		Exhausted: func(string, int) error {
@@ -2828,7 +2828,7 @@ var summaryQuotaDefaults = map[string]int64{
 // Get/unmarshal so it stays cheap with many resources. The version key is
 // skipped and an empty bucket counts as zero.
 func countBucket(ctx context.Context, bucket jetstream.KeyValue, prefix string) (int64, error) {
-	keys, err := kvutil.Keys(ctx, bucket)
+	keys, err := kvstore.Keys(ctx, bucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return 0, nil
@@ -2838,7 +2838,7 @@ func countBucket(ctx context.Context, bucket jetstream.KeyValue, prefix string)
 
 	var count int64
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		if strings.HasPrefix(key, prefix) {

spinifex/domains/lambda/eventsource/mapping.go

@@ -9,7 +9,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -57,7 +57,7 @@ type Store struct{ kv jetstream.KeyValue }
 
 // Open returns the store, creating its bucket on first use.
 func Open(ctx context.Context, js jetstream.KeyValueManager) (*Store, error) {
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucket, 1)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucket, 1)
 	if err != nil {
 		return nil, fmt.Errorf("eventsource: open bucket: %w", err)
 	}

spinifex/domains/lambda/execution/grant.go

@@ -13,7 +13,7 @@ import (
 	"regexp"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -132,7 +132,7 @@ type Store struct {
 
 // Open opens (or creates) the execution bucket.
 func Open(ctx context.Context, js jetstream.KeyValueManager) (*Store, error) {
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucket, 1)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucket, 1)
 	if err != nil {
 		return nil, fmt.Errorf("execution: open bucket: %w", err)
 	}

spinifex/domains/lambda/function/meta.go

@@ -11,7 +11,7 @@ import (
 	"time"
 
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -60,7 +60,7 @@ func (s *MetaService) bucket(ctx context.Context, accountID string) (jetstream.K
 		return kv, nil
 	}
 	name := KVAccountBucket(accountID)
-	kv, err := kvutil.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
+	kv, err := kvstore.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
 	if err != nil {
 		return nil, fmt.Errorf("lambda: open account bucket %s: %w", name, err)
 	}

spinifex/domains/network/external/dhcp/kv.go

@@ -10,7 +10,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -49,7 +49,7 @@ type Store struct {
 // NewStore creates/opens the per-AZ lease bucket and returns a Store. ctx bounds
 // the bucket get-or-create only; each operation carries its own.
 func NewStore(ctx context.Context, js jetstream.KeyValueManager, az string) (*Store, error) {
-	kv, err := kvutil.GetOrCreateBucket(ctx, js, BucketName(az), 5)
+	kv, err := kvstore.GetOrCreateBucket(ctx, js, BucketName(az), 5)
 	if err != nil {
 		return nil, fmt.Errorf("create dhcp lease KV bucket: %w", err)
 	}
@@ -152,7 +152,7 @@ func (s *Store) List(ctx context.Context) ([]Entry, error) {
 	}
 	var out []Entry
 	for _, k := range keys {
-		if k == kvutil.VersionKey {
+		if k == kvstore.VersionKey {
 			continue
 		}
 		e, err := s.Get(ctx, k)

spinifex/domains/network/reconcile/drift_watch_test.go

@@ -14,7 +14,7 @@ import (
 	"github.com/mulgadc/spinifex/internal/testkit"
 	ec2igw "github.com/mulgadc/spinifex/spinifex/domains/ec2/igw"
 	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -22,7 +22,7 @@ import (
 // so a test can write to it and have the source find it.
 func createIntentBucket(t *testing.T, js jetstream.JetStream, name string) jetstream.KeyValue {
 	t.Helper()
-	kv, err := kvutil.GetOrCreateBucket(t.Context(), js, name, 1)
+	kv, err := kvstore.GetOrCreateBucket(t.Context(), js, name, 1)
 	if err != nil {
 		t.Fatalf("create bucket %s: %v", name, err)
 	}

spinifex/domains/network/reconcile/host_ingress_test.go

@@ -19,7 +19,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/domains/network/policy"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/reconcile"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -360,11 +360,11 @@ func TestHostIngress_IntentReadFailureNeverPrunes(t *testing.T) {
 func TestRunHostIngress_IntentWriteConvergesEveryNodeWithoutLeaseOrEvent(t *testing.T) {
 	_, nc, js := testutil.StartTestJetStream(t)
 	ctx := t.Context()
-	vpcs, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 1)
+	vpcs, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 1)
 	if err != nil {
 		t.Fatalf("create VPC bucket: %v", err)
 	}
-	igws, err := kvutil.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 1)
+	igws, err := kvstore.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 1)
 	if err != nil {
 		t.Fatalf("create IGW bucket: %v", err)
 	}

spinifex/domains/network/reconcile/intent.go

@@ -19,7 +19,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/domains/network/external"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/policy"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -236,7 +236,7 @@ func matchesLocalAZ(vpcAZ, localAZ string) bool {
 	return vpcAZ == "" || vpcAZ == localAZ
 }
 
-func keyIsVersion(key string) bool { return key == kvutil.VersionKey }
+func keyIsVersion(key string) bool { return key == kvstore.VersionKey }
 
 func loadVPCs(ctx context.Context, js jetstream.JetStream, localAZ string, out map[string]topology.VPCSpec) (map[string]struct{}, error) {
 	localVPCs := make(map[string]struct{})

spinifex/domains/network/reconcile/scenario_kv_live_test.go

@@ -11,7 +11,7 @@ import (
 	"testing"
 
 	vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -20,7 +20,7 @@ import (
 // creating the bucket if absent — mirroring how the VPC handlers persist state.
 func seedKV(t *testing.T, js jetstream.JetStream, bucket, key string, rec any) {
 	t.Helper()
-	kv, err := kvutil.GetOrCreateBucket(t.Context(), js, bucket, 10)
+	kv, err := kvstore.GetOrCreateBucket(t.Context(), js, bucket, 10)
 	if err != nil {
 		t.Fatalf("GetOrCreateBucket %s: %v", bucket, err)
 	}

spinifex/domains/ochre/usage.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -196,7 +195,7 @@ func (s *UsageStore) TokensThisPeriod(ctx context.Context, accountID string) (in
 	if err != nil {
 		return 0, err
 	}
-	keys, err := kvutil.Keys(ctx, kv)
+	keys, err := kvstore.Keys(ctx, kv)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return 0, nil

spinifex/domains/rds/instance/store.go

@@ -13,7 +13,6 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvlease"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -244,7 +243,7 @@ func InitLeaderBucket(ctx context.Context, js jetstream.JetStream) (jetstream.Ke
 // AccountBucketNames lists every RDS per-account bucket in the cluster. The reconciler and the DNS
 // desired-set both need a cross-tenant view, which only this provides.
 func AccountBucketNames(ctx context.Context, js jetstream.JetStream) ([]string, error) {
-	all, err := kvutil.BucketNames(ctx, js)
+	all, err := kvstore.BucketNames(ctx, js)
 	if err != nil {
 		return nil, err
 	}

spinifex/domains/rds/instance/watch_source_test.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/mulgadc/spinifex/internal/testkit"
 	rdsinstance "github.com/mulgadc/spinifex/spinifex/domains/rds/instance"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -23,7 +23,7 @@ func TestAccountWatchBuckets_ReturnsOnePerAccountBucket(t *testing.T) {
 	require.NoError(t, err)
 	_, err = rdsinstance.GetOrCreateAccountBucket(t.Context(), js, "222222222222")
 	require.NoError(t, err)
-	_, err = kvutil.GetOrCreateBucket(t.Context(), js, "something-else", 1)
+	_, err = kvstore.GetOrCreateBucket(t.Context(), js, "something-else", 1)
 	require.NoError(t, err)
 
 	buckets, err := rdsinstance.AccountWatchBuckets(t.Context(), js)

spinifex/domains/route53/store.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -145,11 +144,11 @@ func NewStore(js jetstream.JetStream) *Store {
 	}
 }
 
-func zoneKey(account, id string) string   { return keyPrefixZone + kvutil.AccountKey(account, id) }
+func zoneKey(account, id string) string   { return keyPrefixZone + kvstore.AccountKey(account, id) }
-func rrsetsKey(account, id string) string { return keyPrefixRRSets + kvutil.AccountKey(account, id) }
+func rrsetsKey(account, id string) string { return keyPrefixRRSets + kvstore.AccountKey(account, id) }
-func changeKey(account, id string) string { return keyPrefixChange + kvutil.AccountKey(account, id) }
+func changeKey(account, id string) string { return keyPrefixChange + kvstore.AccountKey(account, id) }
 func refKey(account, ref string) string {
-	return keyPrefixCallerRf + kvutil.AccountKey(account, refDigest(ref))
+	return keyPrefixCallerRf + kvstore.AccountKey(account, refDigest(ref))
 }
 
 // maxIssuedIDLength bounds the IDs a caller can look up. An overlong key

spinifex/domains/sqs/message/jetstream/durablekv.go

@@ -3,7 +3,7 @@ package jetstream
 import (
 	"context"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	jsapi "github.com/nats-io/nats.go/jetstream"
 )
 
@@ -12,18 +12,18 @@ import (
 // whose stored configuration could expire, cap or evict records, or holds
 // fewer replicas than the cluster needs.
 func OpenDurableKV(ctx context.Context, js jsapi.JetStream, bucket, description string) (jsapi.KeyValue, error) {
-	kv, err := kvutil.GetOrCreateBucketWithOptions(ctx, js, kvutil.BucketOptions{
+	kv, err := kvstore.GetOrCreateBucketWithOptions(ctx, js, kvstore.BucketOptions{
 		Name: bucket, Description: description, History: 1,
 	})
 	if err != nil {
 		return nil, err
 	}
-	return kvutil.VerifyDurable(ctx, kv, bucket)
+	return kvstore.VerifyDurable(ctx, kv, bucket)
 }
 
 // BindDurableKV binds an existing SQS authority bucket without ever creating
 // one, so a reader that must not provision state fails closed when it is absent,
 // and refuses it on the same durability terms as OpenDurableKV.
 func BindDurableKV(ctx context.Context, js jsapi.JetStream, bucket string) (jsapi.KeyValue, error) {
-	return kvutil.BindDurable(ctx, js, bucket)
+	return kvstore.BindDurable(ctx, js, bucket)
 }

spinifex/domains/sqs/message/jetstream/leaderget.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go"
 	jsapi "github.com/nats-io/nats.go/jetstream"
 )
@@ -15,7 +15,7 @@ import (
 // message get, which only the stream leader answers. nats.go switches to a
 // direct get, which followers also answer, whenever the stream allows it.
 func leaderLastMsg(ctx context.Context, js jsapi.JetStream, stream, subject string) (*jsapi.RawStreamMsg, error) {
-	return kvutil.LeaderMsgGet(ctx, js, stream, kvutil.MsgGetRequest{LastFor: subject})
+	return kvstore.LeaderMsgGet(ctx, js, stream, kvstore.MsgGetRequest{LastFor: subject})
 }
 
 // LeaderKVGet reads key from an SQS KV bucket through the stream leader and
@@ -38,11 +38,11 @@ func LeaderKVGet(ctx context.Context, js jsapi.JetStream, bucket, key string) ([
 // to sequence bound, with one leader message get per record.
 func leaderStateKeys(ctx context.Context, js jsapi.JetStream, bucket, keyPrefix string, bound uint64) (map[string]struct{}, error) {
 	stream, subjectPrefix := "KV_"+bucket, "$KV."+bucket+"."
-	req := kvutil.MsgGetRequest{NextFor: subjectPrefix + keyPrefix + ">"}
+	req := kvstore.MsgGetRequest{NextFor: subjectPrefix + keyPrefix + ">"}
 	keys := map[string]struct{}{}
 	for after := uint64(0); after < bound; {
 		req.Seq = after + 1
-		m, err := kvutil.LeaderMsgGet(ctx, js, stream, req)
+		m, err := kvstore.LeaderMsgGet(ctx, js, stream, req)
 		switch {
 		case errors.Is(err, jsapi.ErrMsgNotFound):
 			return keys, nil

spinifex/domains/sqs/message/jetstream/store.go

@@ -16,7 +16,7 @@ import (
 	"github.com/mulgadc/bluebottle/pkg/safecast"
 	"github.com/mulgadc/spinifex/spinifex/domains/sqs/message"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/clustersize"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/nats-io/nats.go"
 	jsapi "github.com/nats-io/nats.go/jetstream"
 )
@@ -101,7 +101,7 @@ func New(ctx context.Context, js jsapi.JetStream, cfg Config) (*Store, error) {
 	}
 	for i := range cfg.StateBuckets {
 		shard := message.Shard(safecast.IntToUint32(i))
-		kv, err := kvutil.GetOrCreateBucketWithOptions(ctx, js, kvutil.BucketOptions{
+		kv, err := kvstore.GetOrCreateBucketWithOptions(ctx, js, kvstore.BucketOptions{
 			Name:        BucketName(shard),
 			Description: "SQS per-message state, shared by every queue allocated this state shard",
 			History:     1,
@@ -960,10 +960,10 @@ func shardFilter(shard message.Shard) string {
 // (after, bound] until fn returns false. Each match is a leader message get,
 // which returns the whole body: a follower's consumer may lag applied purges.
 func readRange(ctx context.Context, js jsapi.JetStream, stream, filter string, after, bound uint64, fn func(*jsapi.RawStreamMsg) (bool, error)) (uint64, bool, error) {
-	req := kvutil.MsgGetRequest{NextFor: filter}
+	req := kvstore.MsgGetRequest{NextFor: filter}
 	for after < bound {
 		req.Seq = after + 1
-		msg, err := kvutil.LeaderMsgGet(ctx, js, stream, req)
+		msg, err := kvstore.LeaderMsgGet(ctx, js, stream, req)
 		switch {
 		case errors.Is(err, jsapi.ErrMsgNotFound):
 			return bound, true, nil

spinifex/domains/sts/assume_role_with_web_identity_test.go

@@ -17,7 +17,7 @@ import (
 	"github.com/golang-jwt/jwt/v5"
 	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -82,7 +82,7 @@ func newWebIdentityFixture(t *testing.T, svc *STSServiceImpl, accountID string)
 	// Register the OIDC provider on the role-account's IAM bucket. Bucket is
 	// created on-the-fly here — the IAM CreateOpenIDConnectProvider API lands
 	// in Sprint 6e.
-	iamKV, err := kvutil.GetOrCreateBucket(t.Context(), svc.js,
+	iamKV, err := kvstore.GetOrCreateBucket(t.Context(), svc.js,
 		iamdomain.IAMAccountBucketName(accountID), iamdomain.KVBucketIAMAccountVersion)
 	require.NoError(t, err)
 	_, err = iamKV.Put(t.Context(), iamdomain.OIDCProviderKey(issuer), []byte(`{"registered":true}`))

spinifex/domains/sts/oidc_jwks_current_behaviour_test.go

@@ -19,7 +19,7 @@ import (
 
 	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 )
 
 // The TestCurrentBehaviour_ tests in this file pin today's AssumeRoleWithWebIdentity
@@ -43,7 +43,7 @@ func assumeWebIdentity(svc *STSServiceImpl, role *iam.Role, token string) (*sts.
 // registerIssuer records issuer as an OIDC provider in the role account's IAM bucket.
 func registerIssuer(t *testing.T, svc *STSServiceImpl, roleAccountID, issuer string) {
 	t.Helper()
-	iamKV, err := kvutil.GetOrCreateBucket(t.Context(), svc.js,
+	iamKV, err := kvstore.GetOrCreateBucket(t.Context(), svc.js,
 		iamdomain.IAMAccountBucketName(roleAccountID), iamdomain.KVBucketIAMAccountVersion)
 	require.NoError(t, err)
 	_, err = iamKV.Put(t.Context(), iamdomain.OIDCProviderKey(issuer), []byte(`{"registered":true}`))

spinifex/domains/sts/session_credentials.go

@@ -12,7 +12,6 @@ import (
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
 	"github.com/nats-io/nats.go/jetstream"
@@ -236,7 +235,7 @@ func (s *STSServiceImpl) deleteSessions(ctx context.Context, match func(*Session
 		return 0, fmt.Errorf("open session credential bucket: %w", err)
 	}
 
-	keys, err := kvutil.Keys(ctx, bucket)
+	keys, err := kvstore.Keys(ctx, bucket)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return 0, nil
@@ -247,7 +246,7 @@ func (s *STSServiceImpl) deleteSessions(ctx context.Context, match func(*Session
 	var deleted int
 	var errs []error
 	for _, key := range keys {
-		if key == kvutil.VersionKey {
+		if key == kvstore.VersionKey {
 			continue
 		}
 		entry, err := bucket.Get(ctx, key)

spinifex/domains/sts/session_credentials_test.go

@@ -11,7 +11,6 @@ import (
 	"github.com/mulgadc/spinifex/internal/testkit"
 	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -62,7 +61,7 @@ func TestInitSessionCredentialsBucket_StampsVersion(t *testing.T) {
 	kv, err := store.KV(t.Context())
 	require.NoError(t, err)
 
-	version, err := kvutil.ReadVersion(t.Context(), kv)
+	version, err := kvstore.ReadVersion(t.Context(), kv)
 	require.NoError(t, err)
 	assert.Equal(t, KVBucketSessionCredentialsVersion, version)
 }
@@ -131,7 +130,7 @@ func TestPutSessionCredential_RejectsNonASIAPrefix(t *testing.T) {
 	if !errors.Is(err, jetstream.ErrNoKeysFound) {
 		require.NoError(t, err)
 		for _, k := range keys {
-			if k == kvutil.VersionKey {
+			if k == kvstore.VersionKey {
 				continue
 			}
 			t.Fatalf("unexpected key written to session bucket: %q", k)
@@ -280,7 +279,7 @@ func TestSweepExpired_IgnoresVersionKey(t *testing.T) {
 	svc, _ := newTestSetup(t)
 	assert.Equal(t, 0, svc.sweep(t.Context(), time.Now().UTC()))
 
-	_, err := sessionsKV(t, svc).Get(t.Context(), kvutil.VersionKey)
+	_, err := sessionsKV(t, svc).Get(t.Context(), kvstore.VersionKey)
 	require.NoError(t, err, "version key must survive the sweep")
 }
 

spinifex/domains/sts/web_identity_trust_policy_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/sts"
 	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -186,7 +186,7 @@ func TestAssumeRoleWithWebIdentity_TrailingSlashIssuerProviderStillGrants(t *tes
 	hostPath := strings.TrimPrefix(f.issuer, "https://")
 	f.federatedARN = iamdomain.OIDCProviderARN(testCallerAccountID, hostPath)
 
-	iamKV, err := kvutil.GetOrCreateBucket(t.Context(), svc.js,
+	iamKV, err := kvstore.GetOrCreateBucket(t.Context(), svc.js,
 		iamdomain.IAMAccountBucketName(testCallerAccountID), iamdomain.KVBucketIAMAccountVersion)
 	require.NoError(t, err)
 	_, err = iamKV.Put(t.Context(), iamdomain.OIDCProviderKey(f.issuer), []byte(`{"registered":true}`))