test(lambda): hold a daemon invocation while serving ends on listener failure

The wake test now waits for the start pass before storing code, so only
the code-stored wake can activate the second function.

Makefile

@@ -90,6 +90,7 @@ go_build:
 LAMBDA_TESTLINK_DIR := ./bin/lambda-testlink
 LAMBDA_TESTLINK_PKG := github.com/mulgadc/spinifex/spinifex/runtime/daemon
 LAMBDA_TESTLINK_PIN := $(LAMBDA_TESTLINK_PKG).lambdaHardKillSHA256
+LAMBDA_TESTLINK_GW := github.com/mulgadc/spinifex/spinifex/runtime/roles/awsgw
 LAMBDA_DIGEST_RE := ^[0-9a-f]{64}$$
 build-lambda-testlink: override export LAMBDA_IMAGE_DIGEST := $(value LAMBDA_IMAGE_DIGEST)
 build-lambda-testlink:
@@ -104,8 +105,8 @@ build-lambda-testlink:
 	@digest=$$(sha256sum $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so | cut -d' ' -f1); \
 	test $${#digest} -eq 64 || { echo "initializer digest missing"; exit 1; }; \
 	echo "initializer sha256 $$digest"; \
-	pin="-X $(LAMBDA_TESTLINK_PIN)=$$digest -X $(LAMBDA_TESTLINK_PKG).lambdaImageDigest=$$LAMBDA_IMAGE_DIGEST"; \
-	LAMBDA_TESTLINK_INITIALIZER=$(abspath $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so) go test -count=1 -tags lambdatestlink -ldflags "$$pin" -run '^TestLambda(HardKillPin|ImageDigestPin)$$' ./spinifex/runtime/daemon/ && \
+	pin="-X $(LAMBDA_TESTLINK_PIN)=$$digest -X $(LAMBDA_TESTLINK_PKG).lambdaImageDigest=$$LAMBDA_IMAGE_DIGEST -X $(LAMBDA_TESTLINK_GW).lambdaImageDigest=$$LAMBDA_IMAGE_DIGEST"; \
+	LAMBDA_TESTLINK_INITIALIZER=$(abspath $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so) go test -count=1 -tags lambdatestlink -ldflags "$$pin" -run '^TestLambda(HardKillPin|ImageDigestPin)$$' ./spinifex/runtime/daemon/ ./spinifex/runtime/roles/awsgw/ && \
 	go build $(GO_BUILD_MOD) -tags lambdatestlink -ldflags "$(subst -s -w ,,$(LDFLAGS)) $$pin" -o $(LAMBDA_TESTLINK_DIR).tmp/$(GO_PROJECT_NAME) cmd/spinifex/main.go
 	@mkdir -p $(LAMBDA_TESTLINK_DIR)
 	mv $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so $(LAMBDA_TESTLINK_DIR).tmp/$(GO_PROJECT_NAME) $(LAMBDA_TESTLINK_DIR)/

architecture/layering/allowlist_test.go

@@ -17,7 +17,6 @@ var allowlist = []allowance{
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/instance", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/routetable", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/snapshot", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
-	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/volume", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/vpc", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 
 	{From: "agents/ecs/agent", To: "domains/ecs", Debt: "agents/ecs/agent", Reason: "assignment/state wire types from the ECS implementation; needs a guest/controller contract"},

contracts/lambda/v1/README.md

@@ -22,7 +22,15 @@ Tenant code may be able to reach the listener, but it cannot open the agent's co
 For each generation the executor creates ephemeral self-signed host and guest certificates whose single URI SAN is `spinifex-lambda://<Region>/<account>/<function ID>/<environment ID>/<generation>/<role>`, with role `host` or `guest`.
 Each end validates the peer certificate against the one pinned peer certificate and separately requires that URI SAN with the opposite role; hostname verification is not relied on.
 ALPN carries the L-GUEST major version, `spinifex-lambda-guest/1`, and an unsupported version fails the handshake. Builds with the `lambdatestlink` tag offer only `spinifex-lambda-testlink/2` on both host and guest, whose first frame carries the execution-role credential, so a test-link end and a release end fail the handshake; release builds are unchanged.
+In the same builds, the gateway reaches the host daemon on the local abstract socket `spinifex-lambda-testlink-control/3`, and each request carries the runtime-image digest the gateway was built with; the daemon refuses a missing or different digest before starting the attempt, and a gateway and daemon on different socket names never connect.
+On that socket a request's raw payload is at most 6 MiB, checked by the gateway before encoding and by the daemon after decoding.
+Its encoded frame is at most the base64 length of 6 MiB plus 64 KiB of metadata headroom and 64 bytes of field slack, and a receiver checks the frame length before allocating; metadata has no separate bound within the frame.
+The credential travels as its own frame of at most 8 KiB, and the reply has the same payload and frame bounds as the request.
+The daemon holds at most two connections at once, from accept to close, and leaves the rest unaccepted in the listen backlog.
 Under TLS 1.3 the guest's handshake can complete before the host refuses its certificate, so the guest treats the connection as accepted only after the host's first message.
+In the same builds, the function owner moves a created function with stored code from `Pending` to `Active` once this node's daemon serves the pinned image and the cluster configuration has exactly one node; a function with environment variables or a code copy over the test-link bound stays `Pending`, and `StateReason` names the unmet condition.
+Release builds never leave `Pending`.
+The Mulga deploy role installs a test-link build with `deploy_lambda_testlink=true`; this profile does not meet Q-199 fail-closed revocation, and its guests take management addresses whose record can expire (mulga-0hsub).
 
 **Bootstrap material:** the executor passes these fw_cfg entries under `opt/spinifex/lambda/`, and the agent reads them as root and unloads `qemu_fw_cfg` before user code starts.
 

spinifex/agents/lambda/hostlink/test_stream.go

@@ -321,6 +321,19 @@ func (s *TestStream) HostSend(req Request) error {
 	return s.write(frame{Type: typeInvoke, Request: &req})
 }
 
+// CheckInvokeFits encodes req as HostSend would and returns ErrFrameTooLarge
+// when the frame exceeds the bound, so a caller can refuse before launch.
+func CheckInvokeFits(req Request) error {
+	b, err := json.Marshal(frame{Type: typeInvoke, Request: &req})
+	if err != nil {
+		return err
+	}
+	if len(b) > maxFrame {
+		return ErrFrameTooLarge
+	}
+	return nil
+}
+
 // HostShutdown writes a shutdown frame, as the host side of the carrier.
 func (s *TestStream) HostShutdown() error {
 	return s.write(frame{Type: typeShutdown})

spinifex/domains/acm/store.go

@@ -14,6 +14,7 @@ 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"
 )
@@ -183,7 +184,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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketACM, KVBucketACMVersion)
+	kv, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketBedrockRPMUsage, 1)
+	kv, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.Update(ctx, s.overrides, accountID, kvstore.CASConfig{
+	_, err := kvutil.Update(ctx, s.overrides, accountID, kvutil.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 kvstore.IsRevisionConflict(err):
+		case kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.GetOrCreateBucket(ctx, s.js, KVBucketAccountQuota, 1)
+	kv, err := kvutil.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 kvstore.IsRevisionConflict(writeErr):
+		case kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 kvstore.IsRevisionConflict(writeErr):
+		case kvutil.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/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"
@@ -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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccountSettings, 10)
+	settingsKV, err := kvutil.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, kvstore.ErrRead):
+	case errors.Is(err, kvutil.ErrRead):
 		slog.ErrorContext(ctx, "Failed to read account settings from KV", "accountID", accountID, "err", err)
-	case errors.Is(err, kvstore.ErrDecode):
+	case errors.Is(err, kvutil.ErrDecode):
 		slog.ErrorContext(ctx, "Corrupt account settings record in KV", "accountID", accountID, "err", err)
-	case errors.Is(err, kvstore.ErrEncode):
+	case errors.Is(err, kvutil.ErrEncode):
 		slog.ErrorContext(ctx, "Failed to marshal account settings record", "accountID", accountID, "err", err)
-	case errors.Is(err, kvstore.ErrWrite):
+	case errors.Is(err, kvutil.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 := kvstore.Update(ctx, s.settingsKV, settingsKey(accountID), kvstore.CASConfig{
+	_, err := kvutil.Update(ctx, s.settingsKV, settingsKey(accountID), kvutil.CASConfig{
 		CreateIfAbsent: true,
 		Exhausted:      func(string, int) error { return errSettingsContended },
 	}, func(rec *AccountSettingsRecord) (bool, error) {

spinifex/domains/ec2/accountcleanup/remainder.go

@@ -0,0 +1,155 @@
+package accountcleanup
+
+import (
+	"context"
+	"log/slog"
+	"slices"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+)
+
+// ReasonRecordRemains marks a resource whose record the owner still holds.
+const ReasonRecordRemains = "record-remains"
+
+// ComputeRemainder reads what compute the account still holds, changing nothing.
+type ComputeRemainder interface {
+	ComputeRemaining(ctx context.Context, accountID string) (*Outcome, error)
+}
+
+// StorageRemainder reads what volumes and tag residue the account still holds,
+// changing nothing.
+type StorageRemainder interface {
+	StorageRemaining(ctx context.Context, accountID string) (*Outcome, error)
+}
+
+var (
+	_ ComputeRemainder = (*ComputeCleanup)(nil)
+	_ StorageRemainder = (*StorageCleanup)(nil)
+)
+
+func validRemainderAccount(accountID string) error {
+	if accountID == "" {
+		return awserrors.Errorf(awserrors.ErrorMissingParameter, "The request must specify an account ID.")
+	}
+	if !awsidentifiers.IsAccountID(accountID) {
+		return awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "The account ID must be 12 digits.")
+	}
+	return nil
+}
+
+// ComputeRemaining lists every instance the account still holds and every spot
+// request that can still launch. A read that does not complete is an incomplete
+// reason, never a shorter list. The result is a snapshot, not evidence of absence.
+func (c *ComputeCleanup) ComputeRemaining(ctx context.Context, accountID string) (*Outcome, error) {
+	if err := validRemainderAccount(accountID); err != nil {
+		return nil, err
+	}
+	if c == nil || c.spot == nil || c.instances == nil || c.teardown == nil {
+		return nil, awserrors.Errorf(awserrors.ErrorServerInternal, "Compute cleanup is not fully wired.")
+	}
+	res := &Outcome{AccountID: accountID}
+	defer res.finish()
+
+	scan, err := c.spot.ScanAccount(ctx, accountID)
+	if err != nil {
+		slog.WarnContext(ctx, "ComputeRemaining: spot scan failed", "account_id", accountID, "err", err)
+		res.incomplete(ReasonSpotInventoryIncomplete)
+	} else {
+		for _, r := range slices.Concat(scan.Active, scan.Terminal) {
+			if !terminalSpotState(r.State) {
+				res.Items = append(res.Items, Item{ResourceID: r.RequestID, Kind: KindSpotRequest, State: r.State, Disposition: DispositionResidue, Reason: ReasonRecordRemains})
+			}
+		}
+	}
+
+	inventory, err := c.instances.AccountInstances(ctx, accountID)
+	if err != nil || !inventoryUsable(inventory) {
+		slog.WarnContext(ctx, "ComputeRemaining: instance inventory incomplete", "account_id", accountID, "err", err)
+		res.incomplete(ReasonInstanceInventoryIncomplete)
+	} else {
+		for _, reservation := range inventory.Reservations {
+			for _, in := range reservation.Instances {
+				state := ""
+				if in.State != nil {
+					state = aws.StringValue(in.State.Name)
+				}
+				res.Items = append(res.Items, Item{ResourceID: aws.StringValue(in.InstanceId), Kind: KindInstance, State: state, Disposition: DispositionResidue, Reason: ReasonRecordRemains})
+			}
+		}
+	}
+
+	outstanding, err := c.teardown.AccountTeardownOutstanding(accountID)
+	if err != nil {
+		slog.WarnContext(ctx, "ComputeRemaining: teardown unread", "account_id", accountID, "err", err)
+		res.incomplete(ReasonTeardownUnread)
+	} else {
+		for _, o := range outstanding {
+			if !res.hasInstance(o.InstanceID) {
+				res.Items = append(res.Items, Item{ResourceID: o.InstanceID, Kind: KindInstance, State: o.State, Disposition: DispositionResidue, Reason: ReasonTerminating})
+			}
+		}
+	}
+	return res, nil
+}
+
+// StorageRemaining lists every volume record the account still holds, deleting
+// ones included, and the tag entries of resources whose owner holds no record.
+func (c *StorageCleanup) StorageRemaining(ctx context.Context, accountID string) (*Outcome, error) {
+	if err := validRemainderAccount(accountID); err != nil {
+		return nil, err
+	}
+	if c == nil || c.volumes == nil || c.instances == nil || c.teardown == nil || c.spot == nil || c.tags == nil {
+		return nil, awserrors.Errorf(awserrors.ErrorServerInternal, "Storage cleanup is not fully wired.")
+	}
+	res := &Outcome{AccountID: accountID}
+	defer res.finish()
+
+	volumes, err := c.volumes.VolumeInventory(ctx, accountID)
+	if err != nil {
+		slog.WarnContext(ctx, "StorageRemaining: volume inventory incomplete", "account_id", accountID, "err", err)
+		res.incomplete(ReasonVolumeInventoryIncomplete)
+	}
+	for _, v := range volumes {
+		reason := ReasonRecordRemains
+		if v.State == ec2.VolumeStateDeleting {
+			reason = ReasonVolumeDeleting
+		}
+		res.Items = append(res.Items, Item{ResourceID: v.VolumeID, Kind: KindVolume, State: v.State, Disposition: DispositionResidue, Reason: reason})
+	}
+
+	live, err := c.instances.AccountInstances(ctx, accountID)
+	if err != nil || !inventoryUsable(live) {
+		slog.WarnContext(ctx, "StorageRemaining: no complete instance view", "account_id", accountID, "err", err)
+		res.incomplete(ReasonHolderProofIncomplete)
+		return res, nil
+	}
+	outstanding, err := c.teardown.AccountTeardownOutstanding(accountID)
+	if err != nil {
+		slog.WarnContext(ctx, "StorageRemaining: teardown unread", "account_id", accountID, "err", err)
+		res.incomplete(ReasonHolderProofIncomplete)
+		return res, nil
+	}
+
+	tagged, err := c.tags.ScanAccountResourceIDs(ctx, accountID, storageTagPrefixes...)
+	if err != nil {
+		slog.WarnContext(ctx, "StorageRemaining: tag index read failed", "account_id", accountID, "err", err)
+		res.incomplete(ReasonTagIndexUnread)
+		return res, nil
+	}
+	owned, resolved := c.ownedResourceIDs(ctx, accountID, live, outstanding)
+	for _, id := range tagged {
+		switch {
+		case !resolved[tagPrefix(id)]:
+			res.incomplete(ReasonTagOwnerUnresolved)
+			res.Items = append(res.Items, Item{ResourceID: id, Kind: KindTag, Disposition: DispositionResidue, Reason: ReasonOwnerRecordsUnresolved})
+		case owned[id]:
+			// The resource itself is reported; its tags are not residue.
+		default:
+			res.Items = append(res.Items, Item{ResourceID: id, Kind: KindTag, Disposition: DispositionResidue, Reason: ReasonRecordRemains})
+		}
+	}
+	return res, nil
+}

spinifex/domains/ec2/accountcleanup/remainder_test.go

@@ -0,0 +1,97 @@
+package accountcleanup
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountinventory"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	ec2spotinstance "github.com/mulgadc/spinifex/spinifex/domains/ec2/spotinstance"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestComputeRemainingReportsEveryHeldInstanceAndOpenSpot(t *testing.T) {
+	r := newComputeRig()
+	r.spot.scans = []*ec2spotinstance.AccountScan{{
+		Active:   []ec2spotinstance.AccountSpotRequest{{RequestID: "sir-0aaaaaaaaaaaaaaa1", State: ec2.SpotInstanceStateOpen}},
+		Terminal: []ec2spotinstance.AccountSpotRequest{{RequestID: "sir-0aaaaaaaaaaaaaaa2", State: ec2.SpotInstanceStateClosed}},
+	}}
+	r.teardown.outstanding = []ec2instance.OutstandingInstance{{InstanceID: "i-0aaaaaaaaaaaaaa99", State: ec2.InstanceStateNameTerminated}}
+
+	res, err := r.cleanup().ComputeRemaining(t.Context(), testAccount)
+	require.NoError(t, err)
+
+	assert.Equal(t, StatusIncomplete, res.Status)
+	assert.NotNil(t, itemFor(res, runningID))
+	assert.NotNil(t, itemFor(res, stoppedID))
+	assert.NotNil(t, itemFor(res, "i-0aaaaaaaaaaaaaa99"))
+	assert.NotNil(t, itemFor(res, "sir-0aaaaaaaaaaaaaaa1"))
+	assert.Nil(t, itemFor(res, "sir-0aaaaaaaaaaaaaaa2"), "a closed request launches nothing")
+	assert.Empty(t, r.terminator.terminated, "a re-read changes nothing")
+}
+
+func TestComputeRemainingCleanWhenNothingHeld(t *testing.T) {
+	r := newComputeRig()
+	r.inventory.out = listing(nil)
+	res, err := r.cleanup().ComputeRemaining(t.Context(), testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, StatusClean, res.Status)
+}
+
+func TestComputeRemainingUnreadStoreIsIncompleteNeverClean(t *testing.T) {
+	for name, mutate := range map[string]func(*computeRig){
+		"instances": func(r *computeRig) { r.inventory.out, r.inventory.err = nil, errors.New("node silent") },
+		"spot":      func(r *computeRig) { r.spot.scanErr = errors.New("spot unreadable") },
+		"teardown":  func(r *computeRig) { r.teardown.err = errors.New("no responder") },
+	} {
+		t.Run(name, func(t *testing.T) {
+			r := newComputeRig()
+			r.inventory.out = listing(nil)
+			mutate(r)
+			res, err := r.cleanup().ComputeRemaining(t.Context(), testAccount)
+			require.NoError(t, err)
+			assert.Equal(t, StatusIncomplete, res.Status)
+			assert.NotEmpty(t, res.IncompleteReasons)
+		})
+	}
+}
+
+func TestStorageRemainingReportsVolumeRecordsAndTagResidue(t *testing.T) {
+	r := newStorageRig()
+	r.volumes.vols = []accountinventory.VolumeItem{
+		volume(freeVol, ec2.VolumeStateAvailable, ""),
+		volume("vol-0aaaaaaaaaaaaaaa2", ec2.VolumeStateDeleting, ""),
+	}
+	r.tags.ids = []string{"i-0aaaaaaaaaaaaaaaf", freeVol}
+
+	res, err := r.cleanup().StorageRemaining(t.Context(), testAccount)
+	require.NoError(t, err)
+
+	assert.Equal(t, StatusIncomplete, res.Status)
+	assert.Equal(t, ReasonRecordRemains, itemFor(res, freeVol).Reason)
+	assert.Equal(t, ReasonVolumeDeleting, itemFor(res, "vol-0aaaaaaaaaaaaaaa2").Reason)
+	assert.Equal(t, KindTag, itemFor(res, "i-0aaaaaaaaaaaaaaaf").Kind, "tags of a resource with no record are residue")
+	assert.Empty(t, r.deleter.deleted, "a re-read deletes nothing")
+	assert.Empty(t, r.tags.deleted, "a re-read clears no tag")
+}
+
+func TestStorageRemainingUnreadStoreIsIncompleteNeverClean(t *testing.T) {
+	for name, mutate := range map[string]func(*storageRig){
+		"volumes":   func(r *storageRig) { r.volumes.failAt = -1 },
+		"instances": func(r *storageRig) { r.instances.out, r.instances.err = nil, errors.New("node silent") },
+		"teardown":  func(r *storageRig) { r.teardown.err = errors.New("no responder") },
+		"tags":      func(r *storageRig) { r.tags.scanErr = errors.New("tag index unreadable") },
+	} {
+		t.Run(name, func(t *testing.T) {
+			r := newStorageRig()
+			r.volumes.vols = nil
+			mutate(r)
+			res, err := r.cleanup().StorageRemaining(t.Context(), testAccount)
+			require.NoError(t, err)
+			assert.Equal(t, StatusIncomplete, res.Status)
+			assert.NotEmpty(t, res.IncompleteReasons)
+		})
+	}
+}

spinifex/domains/ec2/awsapi/instance/AccountTeardownOutstanding.go

@@ -0,0 +1,28 @@
+package instance
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+)
+
+// AccountTeardownOutstandingSubject serves the instances of an account whose
+// teardown is not complete, from the shared record store.
+const AccountTeardownOutstandingSubject = "ec2.AccountTeardownOutstanding"
+
+var accountTeardownOutstandingTimeout = 15 * time.Second
+
+// AccountTeardownOutstanding reads the account's unfinished instance teardowns.
+// No responder, a timeout or an error reply is an error, never an empty list.
+func AccountTeardownOutstanding(ctx context.Context, natsConn *nats.Conn, accountID string) ([]ec2instance.OutstandingInstance, error) {
+	out, err := natsmsg.NATSRequest[ec2instance.TeardownOutstandingOutput](ctx, natsConn, AccountTeardownOutstandingSubject,
+		&ec2instance.TeardownOutstandingInput{}, accountTeardownOutstandingTimeout, accountID)
+	if err != nil {
+		return nil, fmt.Errorf("read outstanding teardown: %w", err)
+	}
+	return out.Instances, nil
+}

spinifex/domains/ec2/awsapi/instance/AccountTeardownOutstanding_test.go

@@ -0,0 +1,88 @@
+package instance
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func serveTeardownOutstanding(t *testing.T, nc *nats.Conn, reply []byte) {
+	t.Helper()
+	_, err := nc.Subscribe(AccountTeardownOutstandingSubject, func(msg *nats.Msg) { _ = msg.Respond(reply) })
+	require.NoError(t, err)
+	require.NoError(t, nc.Flush())
+}
+
+func TestAccountTeardownOutstanding_ReturnsTheServedInstances(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	reply, err := json.Marshal(ec2instance.TeardownOutstandingOutput{Instances: []ec2instance.OutstandingInstance{{InstanceID: "i-1"}}})
+	require.NoError(t, err)
+	serveTeardownOutstanding(t, nc, reply)
+
+	got, err := AccountTeardownOutstanding(context.Background(), nc, "123456789012")
+
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.Equal(t, "i-1", got[0].InstanceID)
+}
+
+// A read that cannot be answered must never look like an empty list, because an
+// empty list lets the storage cleanup believe no QEMU still holds a volume.
+func TestAccountTeardownOutstanding_UnansweredReadIsAnError(t *testing.T) {
+	t.Parallel()
+
+	t.Run("no responder", func(t *testing.T) {
+		t.Parallel()
+		_, nc := startTestNATSServer(t)
+
+		got, err := AccountTeardownOutstanding(context.Background(), nc, "123456789012")
+
+		require.Error(t, err)
+		assert.Nil(t, got)
+		assert.ErrorIs(t, err, nats.ErrNoResponders)
+	})
+	t.Run("error reply", func(t *testing.T) {
+		t.Parallel()
+		_, nc := startTestNATSServer(t)
+		serveTeardownOutstanding(t, nc, awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal))
+
+		got, err := AccountTeardownOutstanding(context.Background(), nc, "123456789012")
+
+		require.Error(t, err)
+		assert.Nil(t, got)
+	})
+	t.Run("unreadable reply", func(t *testing.T) {
+		t.Parallel()
+		_, nc := startTestNATSServer(t)
+		serveTeardownOutstanding(t, nc, []byte("not json"))
+
+		got, err := AccountTeardownOutstanding(context.Background(), nc, "123456789012")
+
+		require.Error(t, err)
+		assert.Nil(t, got)
+	})
+}
+
+// Not parallel: it shortens the package timeout.
+func TestAccountTeardownOutstanding_TimeoutIsAnError(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	_, err := nc.Subscribe(AccountTeardownOutstandingSubject, func(*nats.Msg) {})
+	require.NoError(t, err)
+	require.NoError(t, nc.Flush())
+	previous := accountTeardownOutstandingTimeout
+	accountTeardownOutstandingTimeout = 200 * time.Millisecond
+	t.Cleanup(func() { accountTeardownOutstandingTimeout = previous })
+
+	got, err := AccountTeardownOutstanding(context.Background(), nc, "123456789012")
+
+	require.Error(t, err)
+	assert.Nil(t, got)
+}

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/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"
@@ -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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketEgressOnlyIGW, 10)
+	eigwKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
+	vpcKV, err := kvutil.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, kvstore.AccountKey(accountID, *input.VpcId)); err != nil {
+	if _, err := s.vpcKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, eigwID), data); err != nil {
+	if _, err := s.eigwKV.Put(ctx, kvutil.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 := kvstore.AccountKey(accountID, eigwID)
+	key := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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[kvstore.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
+		vpcEntries[kvutil.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{
-		kvstore.AccountKey(testAccountID, vpcID): []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`),
+		kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.AccountKey(accountID, idempotencyKeyHash(key))
+	kvKey := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 != kvstore.VersionKey && strings.HasSuffix(k, suffix) {
+		if k != kvutil.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 == kvstore.VersionKey || !strings.HasSuffix(k, suffix) {
+		if k == kvutil.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/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"
@@ -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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketEIPs, 10)
+	eipKV, err := kvutil.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 := kvstore.GetOrCreateBucketWithTTL(ctx, js, KVBucketEIPIdempotency, 1, eipIdempotencyTTL)
+	idemKV, err := kvutil.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, kvstore.AccountKey(accountID, allocID), data); err != nil {
+	if _, err := s.eipKV.Put(ctx, kvutil.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 := kvstore.AccountKey(accountID, allocID)
+	key := kvutil.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 := kvstore.AccountKey(accountID, allocID)
+	key := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvutil.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 == kvstore.VersionKey {
+		if k == kvutil.VersionKey {
 			continue
 		}
 		if !strings.HasPrefix(k, prefix) {
@@ -699,7 +699,7 @@ func (s *EIPServiceImpl) DescribeAddressesAttribute(ctx context.Context, input *
 			if id == nil {
 				continue
 			}
-			key := kvstore.AccountKey(accountID, *id)
+			key := kvutil.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 == kvstore.VersionKey {
+			if k == kvutil.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 == kvstore.VersionKey {
+		if k == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvutil.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/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"
@@ -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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketIGW, 10)
+	igwKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
+	vpcKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
+	eipKV, err := kvutil.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, kvstore.AccountKey(accountID, igwID), data); err != nil {
+	if _, err := s.igwKV.Put(ctx, kvutil.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 := kvstore.AccountKey(accountID, igwID)
+	key := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 := kvstore.AccountKey(accountID, igwID)
+	key := kvutil.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, kvstore.AccountKey(accountID, vpcID)); err != nil {
+	if _, err := s.vpcKV.Get(ctx, kvutil.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 := kvstore.AccountKey(accountID, igwID)
+	key := kvutil.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, kvstore.AccountKey(accountID, "*"))
+	lister, err := s.eipKV.ListKeysFiltered(ctx, kvutil.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, kvstore.AccountKey(accountID, vpcID)); err != nil {
+	if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID)); err != nil {
 		return false, awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", vpcID)
 	}
 
-	key := kvstore.AccountKey(accountID, igwID)
+	key := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 := kvstore.Update(ctx, kv, recordKey, kvstore.CASConfig{}, func(r *IGWRecord) (bool, error) {
+	record, err := kvutil.Update(ctx, kv, recordKey, kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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[kvstore.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
+		vpcEntries[kvutil.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, kvstore.AccountKey(testAccountID, igwID), vpcID))
+	require.NoError(t, MarkAttached(context.Background(), svc.igwKV, kvutil.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 := kvstore.AccountKey(testAccountID, igwID)
+	key := kvutil.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 := kvstore.AccountKey(testAccountID, igwID)
+	key := kvutil.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(), kvstore.AccountKey(testAccountID, rec.AllocationId), data)
+		_, err = svc.eipKV.Put(context.Background(), kvutil.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(), kvstore.AccountKey("999999999999", "eipalloc-foreign"), foreign)
+	_, err = svc.eipKV.Put(context.Background(), kvutil.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(), kvstore.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`))
+	_, err = vpcKV.Put(t.Context(), kvutil.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/teardown_outstanding.go

@@ -1,12 +1,33 @@
 package instance
 
+import "context"
+
 // OutstandingInstance is an instance of an account that is not yet fully torn
 // down, with the state its record carries.
 type OutstandingInstance struct {
-	InstanceID string
+	InstanceID string `json:"instance_id"`
 	// State is the record's lifecycle state, such as running or terminated.
-	State string
+	State string `json:"state"`
 	// Teardown is the per-dependent progress (done, pending or failed) the
 	// record carries. It is empty when no dependent has been recorded.
-	Teardown map[string]string
+	Teardown map[string]string `json:"teardown,omitempty"`
+}
+
+// TeardownOutstandingInput is the empty request of the outstanding-teardown
+// read; the account comes from the message header.
+type TeardownOutstandingInput struct{}
+
+// TeardownOutstandingOutput carries every instance not yet fully torn down.
+type TeardownOutstandingOutput struct {
+	Instances []OutstandingInstance `json:"instances"`
+}
+
+// AccountTeardownOutstandingRPC serves AccountTeardownOutstanding over the
+// message bus, so a process without the instance record store can read it.
+func (s *InstanceServiceImpl) AccountTeardownOutstandingRPC(_ context.Context, _ *TeardownOutstandingInput, accountID string) (*TeardownOutstandingOutput, error) {
+	outstanding, err := s.AccountTeardownOutstanding(accountID)
+	if err != nil {
+		return nil, err
+	}
+	return &TeardownOutstandingOutput{Instances: outstanding}, nil
 }

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

@@ -10,6 +10,7 @@ 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"
 )
@@ -71,7 +72,7 @@ func (s *Store) InstanceIDs(ctx context.Context) ([]string, error) {
 
 	var ids []string
 	for key := range lister.Keys() {
-		if key != kvstore.VersionKey {
+		if key != kvutil.VersionKey {
 			ids = append(ids, key)
 		}
 	}

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

@@ -11,6 +11,7 @@ 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"
@@ -59,7 +60,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 := kvstore.ReadVersion(t.Context(), kv)
+	v, err := kvutil.ReadVersion(t.Context(), kv)
 	require.NoError(t, err)
 	assert.Equal(t, uefistore.BucketVersion, v)
 }

spinifex/domains/ec2/launchtemplate/account_cleanup.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.Keys(ctx, s.kv)
+	keys, err := kvutil.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 := kvstore.Keys(ctx, s.kv)
+	keys, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.Keys(ctx, s.kv)
+	keys, err := kvutil.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/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"
@@ -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(), kvstore.AccountKey(testAccountID, deleted)))
+	require.NoError(t, inner.Delete(t.Context(), kvutil.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, kvstore.AccountKey(testAccountID, "lt-0badbadbadbadbad0"), "{not json")
-	putRaw(t, inner, kvstore.AccountKey(testAccountID, "lt-0badbadbadbadbad0.v1"), `{"version_number":1}`)
-	putRaw(t, inner, kvstore.AccountKey(testAccountID, "lt-0mismatch00000000"), `{"launch_template_id":"lt-0other"}`)
-	putRaw(t, inner, kvstore.AccountKey(testAccountID, "lt-0unreadable000000"), `{"launch_template_id":"lt-0unreadable000000"}`)
+	putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0badbadbadbadbad0"), "{not json")
+	putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0badbadbadbadbad0.v1"), `{"version_number":1}`)
+	putRaw(t, inner, kvutil.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, nameKey(testAccountID, "unreadable"), "lt-0unreadable000000")
-	putRaw(t, inner, kvstore.AccountKey(testAccountID, "lt-0bad.vx"), "{}")
+	putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0bad.vx"), "{}")
 
 	recorder := &readRecordingKV{KeyValue: readonlystore.NewKV(t, inner)}
 	svc.kv = &faultyKV{KeyValue: recorder, failGet: func(key string) bool {
-		return key == kvstore.AccountKey(testAccountID, "lt-0unreadable000000")
+		return key == kvutil.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/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"
@@ -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, kvstore.VersionKey), storedKeys(t, svc.kv))
+	assert.Equal(t, sortedKeys(header, header+".v1", header+".v2", name, kvutil.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), kvstore.VersionKey)...), storedKeys(t, inner))
+			assert.Equal(t, sortedKeys(append(tc.left(header, nameIdx), kvutil.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{kvstore.VersionKey}, storedKeys(t, inner))
+			assert.Equal(t, []string{kvutil.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{kvstore.VersionKey}, storedKeys(t, inner))
+	assert.Equal(t, []string{kvutil.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{kvstore.VersionKey}, storedKeys(t, inner))
+			assert.Equal(t, []string{kvutil.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 := kvstore.AccountKey(testAccountID, brokenID)
+	brokenHeader := kvutil.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 := kvstore.AccountKey(testAccountID, "lt-0broken000000000")
+	brokenHeader := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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/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"
@@ -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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketLaunchTemplates, 10)
+	kv, err := kvutil.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 kvstore.AccountKey(accountID, "name."+hex.EncodeToString([]byte(name)))
+	return kvutil.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 kvstore.AccountKey(accountID, ltID+".v"+strconv.FormatInt(n, 10))
+	return kvutil.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 kvstore.AccountKey(accountID, ltID+".v")
+	return kvutil.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, kvstore.AccountKey(accountID, ltID))
+	entry, err := s.kv.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, string(entry.Value())))
+	_, herr := s.kv.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, h.LaunchTemplateId), data); err != nil {
+	if _, err := s.kv.Put(ctx, kvutil.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 := kvstore.AccountKey(accountID, ltID)
+	key := kvutil.AccountKey(accountID, ltID)
 	for range maxVersionRetries {
 		_, err := s.kv.Update(ctx, key, entry.Value(), entry.Revision())
 		if err == nil {
 			return nil
 		}
-		if !kvstore.IsRevisionConflict(err) {
+		if !kvutil.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 := kvstore.AccountKey(accountID, header.LaunchTemplateId)
+	key := kvutil.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 !kvstore.IsRevisionConflict(err) {
+		if !kvutil.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 := kvstore.AccountKey(accountID, ltID)
+	headerKey := kvutil.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, kvstore.AccountKey(accountID, ltID)); errors.Is(gerr, jetstream.ErrKeyNotFound) {
+	if _, gerr := s.kv.Get(ctx, kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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(), kvstore.AccountKey(testAccountID, aws.StringValue(lt.LaunchTemplateId))))
+	require.NoError(t, svc.kv.Delete(t.Context(), kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.AccountKey(testAccountID, id)
+			header := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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 == kvstore.AccountKey(testAccountID, id) }}
+			svc.kv = &faultyKV{KeyValue: svc.kv, failGet: func(key string) bool { return key == kvutil.AccountKey(testAccountID, id) }}
 		},
 		"header undecodable": func(t *testing.T, svc *LaunchTemplateServiceImpl, _ string) {
-			_, err := svc.kv.Put(t.Context(), kvstore.AccountKey(testAccountID, "lt-0broken000000000"), []byte("{not json"))
+			_, err := svc.kv.Put(t.Context(), kvutil.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 := kvstore.AccountKey(testAccountID, id)
+			header := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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 := kvstore.AccountKey(testAccountID, id)
+	header := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.Keys(ctx, s.kv)
+	keys, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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{
-			kvstore.AccountKey(testAccountID, "rtb-routed"): fmt.Appendf(nil,
+			kvutil.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(), kvstore.AccountKey(testAccountID, allocID))
+	entry, err := svc.eipKV.Get(t.Context(), kvutil.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/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"
@@ -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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketNatGateways, 10)
+	natgwKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
+	eipKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get EIP KV bucket: %w", err)
 	}
 
-	subnetKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
+	subnetKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get subnet KV bucket: %w", err)
 	}
 
-	vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
+	vpcKV, err := kvutil.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 kvstore.GetOrCreateBucketWithOptions(ctx, js, kvstore.BucketOptions{
+	return kvutil.GetOrCreateBucketWithOptions(ctx, js, kvutil.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, kvstore.AccountKey(accountID, subnetID))
+	subnetEntry, err := s.subnetKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, allocID))
+	eipEntry, err := s.eipKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, natgwID), data); err != nil {
+	if _, err := s.natgwKV.Put(ctx, kvutil.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, kvstore.AccountKey(accountID, allocID), eipData, eipEntry.Revision()); err != nil {
+		if _, err := s.eipKV.Update(ctx, kvutil.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 := kvstore.AccountKey(accountID, natgwID)
+	key := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, kvBucketRouteTables, 10)
+	rtbKV, err := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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, kvstore.AccountKey(accountID, assoc.SubnetId))
+			subnetEntry, err := s.subnetKV.Get(ctx, kvutil.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 := kvstore.AccountKey(accountID, record.AllocationId)
+	key := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 
@@ -415,7 +415,7 @@ func (s *NatGatewayServiceImpl) DescribeNatGateways(ctx context.Context, input *
 		if foundIDs[id] {
 			continue
 		}
-		key := kvstore.AccountKey(accountID, id)
+		key := kvutil.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/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"
@@ -31,18 +31,18 @@ func setupTestServiceJS(t *testing.T) (*NatGatewayServiceImpl, jetstream.JetStre
 
 	// Seed VPC KV
 	testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
-		kvstore.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
+		kvutil.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{
-		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}`),
+		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}`),
 	})
 
 	// Seed EIP KV
 	testutil.SeedKV(t, js, ec2eip.KVBucketEIPs, map[string][]byte{
-		kvstore.AccountKey(testAccountID, "eipalloc-test1"): []byte(`{"allocation_id":"eipalloc-test1","public_ip":"203.0.113.50","state":"allocated"}`),
-		kvstore.AccountKey(testAccountID, "eipalloc-used"):  []byte(`{"allocation_id":"eipalloc-used","public_ip":"203.0.113.51","state":"associated","association_id":"eipassoc-xxx"}`),
+		kvutil.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"}`),
 	})
 
 	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(), kvstore.AccountKey(testAccountID, natgwID))
+	_, err = svc.natgwKV.Get(t.Context(), kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.Keys(ctx, s.kv)
+	keys, err := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.Keys(ctx, s.kv)
+	keys, err := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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/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"
@@ -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(), kvstore.AccountKey(testAccountID, "undecodable"), []byte("{not json"))
+	_, err = svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "undecodable"), []byte("{not json"))
 	require.NoError(t, err)
-	_, err = svc.kv.Put(t.Context(), kvstore.AccountKey(testAccountID, "anonymous"), []byte(`{"group_name":"anonymous"}`))
+	_, err = svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "anonymous"), []byte(`{"group_name":"anonymous"}`))
 	require.NoError(t, err)
 	createTestGroup(t, svc, "unreadable", ec2.PlacementStrategySpread)
-	require.Contains(t, bucketKeys(t, svc.kv), kvstore.VersionKey)
+	require.Contains(t, bucketKeys(t, svc.kv), kvutil.VersionKey)
 
-	svc.kv = &hookedKV{KeyValue: readonlystore.NewKV(t, svc.kv), failGet: kvstore.AccountKey(testAccountID, "unreadable")}
+	svc.kv = &hookedKV{KeyValue: readonlystore.NewKV(t, svc.kv), failGet: kvutil.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/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"
@@ -84,7 +84,7 @@ func TestCurrentBehaviour_AccountTermination_PlacementGroupStorageLayout(t *test
 	}, testAccountID)
 	require.NoError(t, err)
 
-	assert.Equal(t, []string{testAccountID + ".spread-a", kvstore.VersionKey}, bucketKeys(t, svc.kv))
+	assert.Equal(t, []string{testAccountID + ".spread-a", kvutil.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 := kvstore.AccountKey(testAccountID, "broken")
+	brokenKey := kvutil.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 := kvstore.AccountKey(testAccountID, "cluster-a")
+	key := kvutil.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(), kvstore.AccountKey(testAccountID, "broken"), []byte("{not json"))
+	_, err := svc.kv.Put(t.Context(), kvutil.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(), kvstore.AccountKey(testAccountID, "broken"))
+	_, err = svc.kv.Get(t.Context(), kvutil.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 := kvstore.AccountKey(testAccountID, "cluster-a")
+	key := kvutil.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/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"
@@ -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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketPlacementGroups, 10)
+	kv, err := kvutil.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 := kvstore.AccountKey(accountID, groupName)
+	key := kvutil.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, kvstore.AccountKey(accountID, groupName)); err != nil {
+	if _, err := s.deleteGroup(ctx, accountID, kvutil.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 := kvstore.DeleteIf(ctx, s.kv, key, casConfig(), func(record *PlacementGroupRecord) error {
+	record, err := kvutil.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 == kvstore.VersionKey {
+		if k == kvutil.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 kvstore outcomes that map to
+// errPGAbsent and errPGContended mark the two kvutil 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 kvstore failure into the AWS error the caller returns.
-// A non-kvstore error came from mutate and is already an AWS code.
+// mapCASError turns a kvutil failure into the AWS error the caller returns.
+// A non-kvutil 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, kvstore.ErrRead):
+	case errors.Is(err, kvutil.ErrRead):
 		slog.ErrorContext(ctx, "Failed to read placement group from KV",
 			"op", op, "groupName", groupName, "accountID", accountID, "err", err)
-	case errors.Is(err, kvstore.ErrDecode):
+	case errors.Is(err, kvutil.ErrDecode):
 		slog.ErrorContext(ctx, "Corrupt placement group record in KV",
 			"op", op, "groupName", groupName, "accountID", accountID, "err", err)
-	case errors.Is(err, kvstore.ErrEncode):
+	case errors.Is(err, kvutil.ErrEncode):
 		slog.ErrorContext(ctx, "Failed to marshal placement group record",
 			"op", op, "groupName", groupName, "accountID", accountID, "err", err)
-	case errors.Is(err, kvstore.ErrWrite):
+	case errors.Is(err, kvutil.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() kvstore.CASConfig {
-	return kvstore.CASConfig{
+func casConfig() kvutil.CASConfig {
+	return kvutil.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 := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvutil.Update(ctx, s.kv, kvutil.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 := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvutil.Update(ctx, s.kv, kvutil.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 := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvutil.Update(ctx, s.kv, kvutil.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 := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvutil.Update(ctx, s.kv, kvutil.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 := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvutil.Update(ctx, s.kv, kvutil.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 := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
+	_, err := kvutil.Update(ctx, s.kv, kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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(), kvstore.AccountKey(testAccountID, groupName))
+	entry, err := svc.kv.Get(t.Context(), kvutil.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 := kvstore.Update(t.Context(), svc.kv, kvstore.AccountKey(testAccountID, groupName), kvstore.CASConfig{},
+	_, err := kvutil.Update(t.Context(), svc.kv, kvutil.AccountKey(testAccountID, groupName), kvutil.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/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"
@@ -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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketRouteTables, 10)
+	rtbKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
+	vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
 	}
 
-	igwKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 10)
+	igwKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get IGW KV bucket: %w", err)
 	}
 
-	subnetKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
+	subnetKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get subnet KV bucket: %w", err)
 	}
 
-	natgwKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2natgw.KVBucketNatGateways, 10)
+	natgwKV, err := kvutil.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, kvstore.AccountKey(accountID, rtbID))
+	entry, err := s.rtbKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, record.RouteTableId), data); err != nil {
+	if _, err := s.rtbKV.Put(ctx, kvutil.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 kvstore outcomes that map to
+// errRTBAbsent and errRTBContended mark the two kvutil 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 := kvstore.Update(ctx, s.rtbKV, kvstore.AccountKey(accountID, rtbID), kvstore.CASConfig{
+	_, err := kvutil.Update(ctx, s.rtbKV, kvutil.AccountKey(accountID, rtbID), kvutil.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, kvstore.ErrRead):
+	case errors.Is(err, kvutil.ErrRead):
 		slog.ErrorContext(ctx, "Failed to read route table from KV", "routeTableId", rtbID, "err", err)
-	case errors.Is(err, kvstore.ErrDecode):
+	case errors.Is(err, kvutil.ErrDecode):
 		slog.ErrorContext(ctx, "Corrupt route table record in KV", "routeTableId", rtbID, "err", err)
-	case errors.Is(err, kvstore.ErrEncode):
+	case errors.Is(err, kvutil.ErrEncode):
 		slog.ErrorContext(ctx, "Failed to marshal route table record", "routeTableId", rtbID, "err", err)
-	case errors.Is(err, kvstore.ErrWrite):
+	case errors.Is(err, kvutil.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, kvstore.AccountKey(accountID, vpcID))
+	entry, err := s.vpcKV.Get(ctx, kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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, kvstore.AccountKey(accountID, rtbID)); err != nil {
+	if err := s.rtbKV.Delete(ctx, kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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, kvstore.AccountKey(accountID, igwID))
+		igwEntry, err := s.igwKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, natgwID))
+		natgwEntry, err := s.natgwKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, igwID))
+	igwEntry, err := s.igwKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, subnetID))
+	subnetEntry, err := s.subnetKV.Get(ctx, kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
 			continue
 		}
 
@@ -1168,7 +1168,7 @@ func (s *RouteTableServiceImpl) ReplaceRouteTableAssociation(ctx context.Context
 	}
 
 	for _, key := range keys {
-		if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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, kvstore.AccountKey(accountID, natgwID))
+	natgwEntry, err := s.natgwKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, r.NatGatewayId))
+		natgwEntry, err := s.natgwKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, subnetID))
+	subnetEntry, err := s.subnetKV.Get(ctx, kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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{
-		kvstore.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
+		kvutil.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{
-		kvstore.AccountKey(testAccountID, "igw-test1"): []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"available"}`),
+		kvutil.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{
-		kvstore.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-priv1"): []byte(`{"subnet_id":"subnet-priv1","vpc_id":"vpc-test1","cidr_block":"10.0.11.0/24","state":"available"}`),
+		kvutil.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"}`),
 	})
 
 	// Seed NAT Gateway KV
 	testutil.SeedKV(t, js, ec2natgw.KVBucketNatGateways, map[string][]byte{
-		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"}`),
+		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"}`),
 	})
 
 	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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 )
 
 // 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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 )
 
 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(), kvstore.AccountKey(testAccountID, "sir-bad"), []byte("{not json"))
+	_, err := svc.activeKV.Put(t.Context(), kvutil.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(), kvstore.AccountKey(testAccountID, "sir-empty"), []byte(`{"account_id":"x"}`))
+	_, err := svc.terminalKV.Put(t.Context(), kvutil.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: kvstore.AccountKey(testAccountID, "sir-bbb")}
+	svc.activeKV = failingGetKV{KeyValue: svc.activeKV, key: kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.Keys(ctx, s.activeKV)
+	keys, err := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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(), kvstore.AccountKey(testAccountID, "sir-broken01"), []byte("{not json"))
+	_, err := svc.activeKV.Put(t.Context(), kvutil.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/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"
@@ -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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketSpotRequests, 10)
+	activeKV, err := kvutil.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 kvstore.GetOrCreateBucketWithOptions(ctx, js, kvstore.BucketOptions{
+	return kvutil.GetOrCreateBucketWithOptions(ctx, js, kvutil.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, kvstore.AccountKey(accountID, sirID), data); err != nil {
+		if _, err := s.activeKV.Put(ctx, kvutil.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 := kvstore.AccountKey(accountID, sirID)
+		key := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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/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"
@@ -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 := kvstore.AccountKey(acct, id)
+	key := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.AccountKey(accountID, resourceID)
-	_, err := kvstore.Update(ctx, kv, key, kvstore.CASConfig{NotFound: errRecordAbsent}, func(rec *R) (bool, error) {
+	key := kvutil.AccountKey(accountID, resourceID)
+	_, err := kvutil.Update(ctx, kv, key, kvutil.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/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"
@@ -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{
-		kvstore.AccountKey("acct", "res-1"):   rec,
-		kvstore.AccountKey("acct", "res-bad"): []byte("not-json"),
+		kvutil.AccountKey("acct", "res-1"):   rec,
+		kvutil.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(), kvstore.AccountKey("acct", "res-1"))
+		entry, err := kv.Get(t.Context(), kvutil.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{
-		kvstore.AccountKey("acct", "vol-1"): withTags,
-		kvstore.AccountKey("acct", "vol-2"): nilTags,
+		kvutil.AccountKey("acct", "vol-1"): withTags,
+		kvutil.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(), kvstore.AccountKey("acct", id))
+		entry, err := kv.Get(t.Context(), kvutil.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(), kvstore.AccountKey("acct", "snap-other"), otherRec)
+	_, err = kv.Put(t.Context(), kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.Keys(ctx, s.kv)
+	keys, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 kvstore.GetOrCreateBucket(ctx, js, KVBucketTags, kvTagsHistory)
+	return kvutil.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 := kvstore.Update(ctx, s.kv, key, kvstore.CASConfig{
+	_, err := kvutil.Update(ctx, s.kv, key, kvutil.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 kvstore.Put(ctx, s.kv, tagsKVKey(accountID, resourceID), kvstore.CASConfig{
+	return kvutil.Put(ctx, s.kv, tagsKVKey(accountID, resourceID), kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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, kvstore.AccountKey(accountID, claim.VpcId)); err == nil {
+		if _, err := s.vpcKV.Get(ctx, kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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, kvstore.AccountKey(accountID, claim.VpcId), VPCRecord{
+	created, err := createRecord(ctx, s.vpcKV, kvutil.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, kvstore.AccountKey(accountID, claim.SubnetId), SubnetRecord{
+	created, err = createRecord(ctx, s.subnetKV, kvutil.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 := kvstore.Update(ctx, s.defaultsKV, accountID, kvstore.CASConfig{}, func(c *defaultVPCClaim) (bool, error) {
+	if _, err := kvutil.Update(ctx, s.defaultsKV, accountID, kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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, kvstore.AccountKey(accountID, id), ec2vpc.VPCRecord{
+		putRecord(t, js, ec2vpc.KVBucketVPCs, kvutil.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, kvstore.AccountKey(accountID, "subnet-00000000000older"), ec2vpc.SubnetRecord{
+	putRecord(t, js, ec2vpc.KVBucketSubnets, kvutil.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, kvstore.AccountKey(accountID, "vpc-00000000000000half"), ec2vpc.VPCRecord{
+	putRecord(t, js, ec2vpc.KVBucketVPCs, kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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, kvstore.AccountKey(accountID, subnetId))
+	subnetEntry, err := s.subnetKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, eniId), data); err != nil {
+	if _, err := s.eniKV.Put(ctx, kvutil.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 := kvstore.AccountKey(accountID, eniId)
+	key := kvutil.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 := kvstore.AccountKey(accountID, eniID)
+	key := kvutil.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 := kvstore.AccountKey(accountID, eniId)
+	key := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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, kvstore.AccountKey(accountID, eniId), eniId, func(record *ENIRecord) error {
+	err := s.casENI(ctx, kvutil.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 := kvstore.AccountKey(accountID, eniId)
+	key := kvutil.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 := kvstore.AccountKey(accountID, eniId)
+	key := kvutil.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, kvstore.AccountKey(accountID, eniId), eniId, func(r *ENIRecord) error {
+	return s.casENI(ctx, kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 := kvstore.AccountKey(accountID, eniId)
+	key := kvutil.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, kvstore.AccountKey(accountID, id))
+		entry, err := s.sgKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, vpcID))
+	entry, err := s.vpcKV.Get(ctx, kvutil.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, kvstore.AccountKey(accountID, sgId))
+		sgEntry, err := s.sgKV.Get(ctx, kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 == kvstore.VersionKey || !strings.HasSuffix(k, suffix) {
+		if k == kvutil.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/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"
@@ -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(), kvstore.AccountKey(testAccountID, eniID))
+		entry, err := inner.Get(context.Background(), kvutil.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(), kvstore.AccountKey(testAccountID, eniID), eniID, func(*ENIRecord) error {
+	err := svc.casENI(t.Context(), kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 != kvstore.VersionKey && strings.HasSuffix(k, suffix) {
+		if k != kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 == kvstore.VersionKey {
+		if k == kvutil.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, kvstore.AccountKey(accountID, groupId), data); err != nil {
+	if _, err := s.sgKV.Put(ctx, kvutil.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 := kvstore.AccountKey(accountID, groupId)
+	key := kvutil.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, kvstore.AccountKey(accountID, r.SourceSG))
+		entry, err := s.sgKV.Get(ctx, kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 := kvstore.AccountKey(accountID, groupId)
+	key := kvutil.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 := kvstore.AccountKey(accountID, groupId)
+	key := kvutil.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 := kvstore.AccountKey(accountID, groupId)
+	key := kvutil.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 := kvstore.AccountKey(accountID, groupId)
+	key := kvutil.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 := kvstore.AccountKey(accountID, req.groupId)
+	key := kvutil.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 := kvstore.AccountKey(accountID, groupId)
+	key := kvutil.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, kvstore.AccountKey(accountID, groupId), data); err != nil {
+	if _, err := s.sgKV.Create(ctx, kvutil.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 := kvstore.AccountKey(accountID, groupId)
+	key := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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(), kvstore.AccountKey(testAccountID, sgID))
+	entry, err := svc.sgKV.Get(t.Context(), kvutil.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 := kvstore.AccountKey(testAccountID, sgID)
+	key := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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(), kvstore.AccountKey(testAccountID, sgID), data)
+	_, err = svc.sgKV.Put(t.Context(), kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 == kvstore.VersionKey {
+		if k == kvutil.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(), kvstore.AccountKey(testAccountID, vpcID), data)
+	_, err = svc.vpcKV.Put(t.Context(), kvutil.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(), kvstore.AccountKey(testAccountID, sgID))
+	entry, err := svc.sgKV.Get(t.Context(), kvutil.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/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"
@@ -173,7 +173,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
 		return nil, fmt.Errorf("create JetStream client: %w", err)
 	}
 
-	vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketVPCs, 10)
+	vpcKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketSubnets, 10)
+	subnetKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketVNICounter, 10)
+	vniKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketENIs, 10)
+	eniKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketSecurityGroups, 10)
+	sgKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketDefaultVPCs, 10)
+	defaultsKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, "spinifex-vpc-route-tables", 10)
+	rtbKV, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, "spinifex-igw", 10)
+	igwKV, err := kvutil.GetOrCreateBucket(ctx, js, "spinifex-igw", 10)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket spinifex-igw: %w", err)
 	}
-	eigwKV, err := kvstore.GetOrCreateBucket(ctx, js, "spinifex-eigw", 10)
+	eigwKV, err := kvutil.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 := kvstore.Update(ctx, s.vniKV, vniCounterKey, kvstore.CASConfig{CreateIfAbsent: true, Attempts: vniCASAttempts},
+	_, err := kvutil.Update(ctx, s.vniKV, vniCounterKey, kvutil.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, kvstore.AccountKey(accountID, vpcID), data); err != nil {
+	if _, err := s.vpcKV.Put(ctx, kvutil.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, kvstore.AccountKey(accountID, vpcId)); err != nil {
+	if _, err := s.vpcKV.Get(ctx, kvutil.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 := kvstore.AccountKey(accountID, vpcID)
+	key := kvutil.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 == kvstore.VersionKey {
+		if k == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvutil.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, kvstore.AccountKey(accountID, rtbID)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+		if err := s.rtbKV.Delete(ctx, kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
+		if k == kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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, kvstore.AccountKey(accountID, vpcID))
+	vpcEntry, err := s.vpcKV.Get(ctx, kvutil.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 == kvstore.VersionKey {
+		if k == kvutil.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, kvstore.AccountKey(accountID, subnetID), data); err != nil {
+	if _, err := s.subnetKV.Put(ctx, kvutil.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 := kvstore.AccountKey(accountID, subnetID)
+	key := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 := kvstore.AccountKey(accountID, subnetID)
+	key := kvutil.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 := kvstore.AccountKey(accountID, vpcID)
+	key := kvutil.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 := kvstore.AccountKey(accountID, vpcID)
+	key := kvutil.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, kvstore.AccountKey(accountID, rtbID), data); err != nil {
+	if _, err := s.rtbKV.Create(ctx, kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
+		if key == kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 := kvstore.AccountKey(accountID, subnetId)
+	key := kvutil.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 == kvstore.VersionKey {
+		if k == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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, kvstore.VersionKey, k, "no subnet record may exist")
+		assert.Equal(t, kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.Keys(t.Context(), kv)
+	entries, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 kvstore.GetOrCreateBucket(ctx, js, SigningBucket, signingKeyHistory)
+	return kvutil.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 := kvstore.Keys(ctx, kv)
+	names, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
+	kv, err := kvutil.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/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"
@@ -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 := kvstore.BucketNames(ctx, js)
+	names, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, bucket, KVBucketECSAccountHistory)
+	kv, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 kvstore.Update(ctx, kv, Key(cluster, principalARN), kvstore.CASConfig{NotFound: ErrNotFound}, func(r *Record) (bool, error) {
+	return kvutil.Update(ctx, kv, Key(cluster, principalARN), kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 kvstore.Update(ctx, kv, Key(cluster, addon), kvstore.CASConfig{NotFound: ErrNotFound}, func(r *Record) (bool, error) {
+	return kvutil.Update(ctx, kv, Key(cluster, addon), kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.Update(ctx, kv, ClusterMetaKey(name), kvstore.CASConfig{
+	_, err := kvutil.Update(ctx, kv, ClusterMetaKey(name), kvutil.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 !kvstore.IsRevisionConflict(err) {
+		if !kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	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 = kvstore.DeleteIf(ctx, acctKV, ClusterMetaKey(name), kvstore.CASConfig{
+	_, err = kvutil.DeleteIf(ctx, acctKV, ClusterMetaKey(name), kvutil.CASConfig{
 		Attempts: maxClusterStateCASRetries,
 		NotFound: ErrClusterNotFound,
 	}, func(m *ClusterMeta) error {

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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 = kvstore.Update(ctx, acctKV, NodegroupKey(cluster, ng), kvstore.CASConfig{
+	_, err = kvutil.Update(ctx, acctKV, NodegroupKey(cluster, ng), kvutil.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,6 +9,7 @@ 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"
@@ -156,7 +157,7 @@ func accountBucketNames(ctx context.Context, nc *nats.Conn) ([]string, error) {
 	if err != nil {
 		return nil, fmt.Errorf("jetstream: %w", err)
 	}
-	all, err := kvstore.BucketNames(ctx, js)
+	all, err := kvutil.BucketNames(ctx, js)
 	if err != nil {
 		return nil, err
 	}
@@ -195,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 := kvstore.GetOrCreateBucket(ctx, js, bucket, KVBucketEKSAccountHistory)
+	kv, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 = ekscluster.GetOrCreateAccountBucket(t.Context(), js, "222222222222")
 	require.NoError(t, err)
-	_, err = kvstore.GetOrCreateBucket(t.Context(), js, "something-else", 1)
+	_, err = kvutil.GetOrCreateBucket(t.Context(), js, "something-else", 1)
 	require.NoError(t, err)
 
 	buckets, err := ekscluster.AccountWatchBuckets(t.Context(), nc)

spinifex/domains/elbv2/store.go

@@ -11,6 +11,7 @@ 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"
@@ -34,7 +35,7 @@ const (
 	maxLBNameClaimRetries = 5
 
 	// targetGroupCASAttempts bounds a target-group read-modify-write. It sits
-	// above the kvstore default because Terraform registers targets at its own
+	// above the kvutil default because Terraform registers targets at its own
 	// parallelism (10 by default), every one contending for the same record.
 	targetGroupCASAttempts = 12
 
@@ -89,7 +90,7 @@ func NewStore(ctx context.Context, nc *nats.Conn) (*Store, error) {
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}
 
-	kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketELBv2, 1)
+	kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketELBv2, 1)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketELBv2, err)
 	}
@@ -188,7 +189,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 kvstore.Update(ctx, s.kv, KeyPrefixLB+lbID, kvstore.CASConfig{
+	return kvutil.Update(ctx, s.kv, KeyPrefixLB+lbID, kvutil.CASConfig{
 		Attempts: targetGroupCASAttempts,
 		NotFound: ErrLoadBalancerNotFound,
 		Exhausted: func(_ string, attempts int) error {
@@ -293,7 +294,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 kvstore.Update(ctx, s.kv, KeyPrefixTG+tgID, kvstore.CASConfig{
+	return kvutil.Update(ctx, s.kv, KeyPrefixTG+tgID, kvutil.CASConfig{
 		Attempts: targetGroupCASAttempts,
 		NotFound: ErrTargetGroupNotFound,
 		Exhausted: func(_ string, attempts int) error {
@@ -651,7 +652,7 @@ func listByPrefixOpt[T any](ctx context.Context, kv jetstream.KeyValue, prefix s
 
 	var result []*T
 	for _, key := range keys {
-		if key == kvstore.VersionKey {
+		if key == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccountNames, 5)
+	kv, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 )
 
 // 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 := kvstore.Keys(ctx, s.groupsBucket)
+	keys, err := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 := kvstore.Update(ctx, s.groupsBucket, accountID+"."+groupName, iamRecordCASConfig("group", accountID, groupName), mutate)
+	_, err := kvutil.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 := kvstore.Keys(ctx, s.usersBucket)
+	keys, err := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 )
 
 // 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 := kvstore.Keys(ctx, s.instanceProfilesBucket)
+	keys, err := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 := kvstore.DeleteIf(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName),
+	_, err := kvutil.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 := kvstore.Update(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName), mutate)
+	_, err := kvutil.Update(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName), mutate)
 	return err
 }
 
-func (s *IAMServiceImpl) instanceProfileCASConfig(accountID, profileName string) kvstore.CASConfig {
-	return kvstore.CASConfig{
+func (s *IAMServiceImpl) instanceProfileCASConfig(accountID, profileName string) kvutil.CASConfig {
+	return kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 )
 
 // 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 kvstore.GetOrCreateBucket(ctx, js, IAMAccountBucketName(accountID), 1)
+	return kvutil.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 := kvstore.Keys(ctx, kv)
+	keys, err := kvutil.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 == kvstore.VersionKey || !strings.HasPrefix(key, oidcProvidersKeyPrefix) {
+		if key == kvutil.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 = kvstore.Update(ctx, kv, OIDCProviderKey(issuer), iamRecordCASConfig("oidc-provider", accountID, arn), mutate)
+	_, err = kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 )
 
 const (
@@ -94,7 +94,7 @@ func (s *IAMServiceImpl) updatePolicyCAS(ctx context.Context, accountID, policyA
 		return err
 	}
 	defer s.recordLocks.lock("policy/" + key)()
-	_, err = kvstore.Update(ctx, s.policiesBucket, key, cfg, func(p *Policy) (bool, error) {
+	_, err = kvutil.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, kvstore.CASConfig, error) {
+func policyCASTarget(accountID, policyARN string) (string, kvutil.CASConfig, error) {
 	_, policyName, err := iamarn.ParsePolicyARN(policyARN)
 	if err != nil {
 		slog.Debug("policyCASTarget: unparseable policy ARN",
 			"accountID", accountID, "policyArn", policyARN, "err", err)
-		return "", kvstore.CASConfig{}, policyNotFound(policyARN)
+		return "", kvutil.CASConfig{}, policyNotFound(policyARN)
 	}
-	return accountID + "." + policyName, kvstore.CASConfig{
+	return accountID + "." + policyName, kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 )
 
 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 := kvstore.Keys(ctx, s.rolesBucket)
+	keys, err := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 := kvstore.Update(ctx, s.rolesBucket, accountID+"."+roleName, kvstore.CASConfig{
+	_, err := kvutil.Update(ctx, s.rolesBucket, accountID+"."+roleName, kvutil.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 := kvstore.Keys(ctx, s.instanceProfilesBucket)
+	keys, err := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketUsers, 10)
+	usersBucket, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccessKeys, 5)
+	accessKeysBucket, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketPolicies, 10)
+	policiesBucket, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccounts, 5)
+	accountsBucket, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccountCounter, 5)
+	accountCounterBucket, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketRoles, 10)
+	rolesBucket, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketInstanceProfiles, 10)
+	instanceProfilesBucket, err := kvutil.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 := kvstore.GetOrCreateBucket(ctx, js, KVBucketGroups, 10)
+	groupsBucket, err := kvutil.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 := kvstore.Keys(ctx, s.usersBucket)
+	keys, err := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 := kvstore.Update(ctx, s.accessKeysBucket, keyID, cfg, func(ak *AccessKey) (bool, error) {
+		_, err := kvutil.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 := kvstore.Keys(ctx, s.usersBucket)
+	keys, err := kvutil.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 != kvstore.VersionKey {
+		if key != kvutil.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 := kvstore.Keys(ctx, s.accountsBucket)
+	keys, err := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 := kvstore.Update(context.Background(), s.accountsBucket, accountID,
-		kvstore.CASConfig{NotFound: errors.New(awserrors.ErrorIAMNoSuchEntity)},
+	account, err := kvutil.Update(context.Background(), s.accountsBucket, accountID,
+		kvutil.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, kvstore.ErrWrite) || errors.Is(err, kvstore.ErrRead) || errors.Is(err, kvstore.ErrDecode) {
+		if errors.Is(err, kvutil.ErrWrite) || errors.Is(err, kvutil.ErrRead) || errors.Is(err, kvutil.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 := kvstore.Update(context.Background(), s.accountsBucket, accountID,
-		kvstore.CASConfig{NotFound: errors.New(awserrors.ErrorIAMNoSuchEntity)},
+	account, err := kvutil.Update(context.Background(), s.accountsBucket, accountID,
+		kvutil.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, kvstore.ErrWrite) || errors.Is(err, kvstore.ErrRead) || errors.Is(err, kvstore.ErrDecode) {
+		if errors.Is(err, kvutil.ErrWrite) || errors.Is(err, kvutil.ErrRead) || errors.Is(err, kvutil.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 := kvstore.Keys(ctx, s.policiesBucket)
+	keys, err := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.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 := kvstore.DeleteIf(ctx, s.policiesBucket, key, cfg, func(p *Policy) error {
+	policy, err := kvutil.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 := kvstore.Keys(ctx, scan.bucket)
+		keys, err := kvutil.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 == kvstore.VersionKey {
+			if key == kvutil.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 := kvstore.Update(ctx, s.usersBucket, accountID+"."+userName, iamRecordCASConfig("user", accountID, userName), mutate)
+	_, err := kvutil.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) kvstore.CASConfig {
+func iamRecordCASConfig(kind, accountID, name string) kvutil.CASConfig {
 	notFound := noSuchEntity(kind, name)
 	if kind == "oidc-provider" {
 		notFound = oidcProviderNotFound(name)
 	}
-	return kvstore.CASConfig{
+	return kvutil.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 := kvstore.Keys(ctx, bucket)
+	keys, err := kvutil.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 == kvstore.VersionKey {
+		if key == kvutil.VersionKey {
 			continue
 		}
 		if strings.HasPrefix(key, prefix) {

spinifex/domains/lambda/execution/grant.go

@@ -13,7 +13,7 @@ import (
 	"regexp"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.GetOrCreateBucket(ctx, js, KVBucket, 1)
+	kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucket, 1)
 	if err != nil {
 		return nil, fmt.Errorf("execution: open bucket: %w", err)
 	}

spinifex/domains/lambda/function/activate.go

@@ -0,0 +1,155 @@
+package function
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Readiness reports why a function whose code is stored cannot run yet, or nil
+// when it can. Its error text becomes the function's StateReason.
+type Readiness func(ctx context.Context, rec Record) error
+
+// SetReadiness enables activation. Without it no function ever leaves Pending.
+func (s *MetaService) SetReadiness(r Readiness) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.ready = r
+	if s.nudge == nil {
+		s.nudge = make(chan struct{}, 1)
+	}
+}
+
+func (s *MetaService) readiness() (Readiness, chan struct{}) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.ready, s.nudge
+}
+
+// wake asks the activation loop for an early pass; it never blocks.
+func (s *MetaService) wake() {
+	if _, nudge := s.readiness(); nudge != nil {
+		select {
+		case nudge <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// RunActivation activates pending functions until ctx ends: once at start, on
+// every code-stored completion, and every interval, so a crash between storing
+// the code and activating it is converged by the next pass.
+func (s *MetaService) RunActivation(ctx context.Context, interval time.Duration) {
+	ready, nudge := s.readiness()
+	if ready == nil {
+		return
+	}
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	for {
+		if err := s.ActivatePending(ctx); err != nil && ctx.Err() == nil {
+			slog.Warn("Lambda: activation pass incomplete; retrying", "err", err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-nudge:
+		case <-t.C:
+		}
+	}
+}
+
+// ActivatePending makes one pass over every account's functions, moving each
+// Pending function with stored code to Active when readiness allows it.
+func (s *MetaService) ActivatePending(ctx context.Context) error {
+	ready, _ := s.readiness()
+	if ready == nil {
+		return nil
+	}
+	var accounts []string
+	names := s.js.KeyValueStoreNames(ctx)
+	for name := range names.Name() {
+		if id, ok := strings.CutPrefix(name, KVBucketAccountPrefix); ok {
+			accounts = append(accounts, id)
+		}
+	}
+	if err := names.Error(); err != nil {
+		return fmt.Errorf("lambda: list account buckets: %w", err)
+	}
+	var errs []error
+	for _, accountID := range accounts {
+		if err := s.activateAccount(ctx, accountID, ready); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+func (s *MetaService) activateAccount(ctx context.Context, accountID string, ready Readiness) error {
+	kv, err := s.bucket(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	keys, err := kv.ListKeysFiltered(ctx, kvFunctionKeyPrefix+">")
+	if errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("lambda: list functions of account %s: %w", accountID, err)
+	}
+	var errs []error
+	for key := range keys.Keys() {
+		if err := s.activateOne(ctx, kv, key, ready); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// activateOne writes the readiness outcome by compare-and-swap at the revision
+// it read, so a concurrent update or delete wins and is never overwritten.
+func (s *MetaService) activateOne(ctx context.Context, kv jetstream.KeyValue, key string, ready Readiness) error {
+	entry, err := kv.Get(ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) || errors.Is(err, jetstream.ErrKeyDeleted) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("lambda: read %s: %w", key, err)
+	}
+	rec, err := decode(entry.Value())
+	if err != nil {
+		return err
+	}
+	if rec.State != StatePending || rec.Code.State != CodeStored {
+		return nil
+	}
+	if notReady := ready(ctx, rec); notReady != nil {
+		if rec.StateReason == notReady.Error() {
+			return nil
+		}
+		rec.StateReasonCode, rec.StateReason = ReasonCodeCreating, notReady.Error()
+	} else {
+		rec.State, rec.StateReasonCode, rec.StateReason = StateActive, "", ""
+	}
+	data, err := json.Marshal(rec)
+	if err != nil {
+		return err
+	}
+	if _, err := kv.Update(ctx, key, data, entry.Revision()); err != nil {
+		if errors.Is(err, jetstream.ErrKeyExists) {
+			return nil
+		}
+		return fmt.Errorf("lambda: write readiness of %s: %w", key, err)
+	}
+	if rec.State == StateActive {
+		slog.Info("Lambda: function active", "account_id", rec.AccountID, "region", rec.Region,
+			"function_name", rec.FunctionName, "function_id", rec.FunctionID)
+	}
+	return nil
+}

spinifex/domains/lambda/function/activate_test.go

@@ -0,0 +1,120 @@
+package function_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	actAccount = "000000000001"
+	actRegion  = "ap-southeast-2"
+	actDigest  = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+)
+
+func newOwner(t *testing.T) *function.MetaService {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	return function.NewMetaService(testutil.NewJetStream(t, nc))
+}
+
+func create(t *testing.T, svc *function.MetaService, name string) uint64 {
+	t.Helper()
+	resp, err := svc.Create(t.Context(), &function.CreateRequest{Region: actRegion, Record: function.Record{
+		FunctionID: "id-" + name, FunctionName: name, Role: "arn:aws:iam::000000000001:role/r",
+		Runtime: "provided.al2023", Handler: "bootstrap", Timeout: 3, MemorySize: 128,
+		Code: function.Code{RevisionID: "rev-" + name},
+	}}, actAccount)
+	require.NoError(t, err)
+	return resp.Revision
+}
+
+func get(t *testing.T, svc *function.MetaService, name string) *function.GetResponse {
+	t.Helper()
+	resp, err := svc.Get(t.Context(), &function.GetRequest{Region: actRegion, FunctionName: name}, actAccount)
+	require.NoError(t, err)
+	return resp
+}
+
+func storeCode(t *testing.T, svc *function.MetaService, name string, rev uint64) {
+	t.Helper()
+	_, err := svc.CodeStored(t.Context(), &function.CodeStoredRequest{Region: actRegion, FunctionName: name, Revision: rev, SHA256: actDigest, Size: 9}, actAccount)
+	require.NoError(t, err)
+}
+
+func TestActivation_WithoutReadinessNothingLeavesPending(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	svc.RunActivation(ctx, time.Millisecond)
+
+	got := get(t, svc, "hello").Record
+	assert.Equal(t, function.StatePending, got.State)
+	assert.Equal(t, function.ReasonCreating, got.StateReason)
+}
+
+func TestActivation_UnmetConditionStaysPendingWithItsReason(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+	svc.SetReadiness(func(context.Context, function.Record) error { return errors.New("The node is not serving.") })
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	first := get(t, svc, "hello")
+	assert.Equal(t, function.StatePending, first.Record.State)
+	assert.Equal(t, function.ReasonCodeCreating, first.Record.StateReasonCode)
+	assert.Equal(t, "The node is not serving.", first.Record.StateReason)
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	assert.Equal(t, first.Revision, get(t, svc, "hello").Revision, "an unchanged reason is not rewritten")
+}
+
+func TestActivation_ReadyStoredFunctionBecomesActive(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+	storingRev := create(t, svc, "storing")
+	var seen []string
+	svc.SetReadiness(func(_ context.Context, rec function.Record) error {
+		seen = append(seen, rec.FunctionName)
+		return nil
+	})
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+
+	got := get(t, svc, "hello").Record
+	assert.Equal(t, function.StateActive, got.State)
+	assert.Empty(t, got.StateReason)
+	assert.Empty(t, got.StateReasonCode)
+	assert.Equal(t, function.CodeStored, got.Code.State)
+	storing := get(t, svc, "storing")
+	assert.Equal(t, function.StatePending, storing.Record.State, "a function still storing its code is never activated")
+	assert.Equal(t, storingRev, storing.Revision)
+	assert.Equal(t, []string{"hello"}, seen)
+}
+
+func TestActivation_CodeStoredWakesTheLoop(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "first", create(t, svc, "first"))
+	// Readiness is enabled after first is stored, so no wake is buffered and
+	// only the start pass can activate first.
+	svc.SetReadiness(func(context.Context, function.Record) error { return nil })
+	ctx, cancel := context.WithCancel(t.Context())
+	done := make(chan struct{})
+	go func() { defer close(done); svc.RunActivation(ctx, time.Hour) }()
+	t.Cleanup(func() { cancel(); <-done })
+	require.Eventually(t, func() bool { return get(t, svc, "first").Record.State == function.StateActive },
+		10*time.Second, 20*time.Millisecond, "the start pass activates what was already stored")
+
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+
+	require.Eventually(t, func() bool { return get(t, svc, "hello").Record.State == function.StateActive },
+		10*time.Second, 20*time.Millisecond, "after the start pass, only the code-stored wake can activate within the hour")
+}

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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -44,6 +44,8 @@ type MetaService struct {
 
 	mu      sync.Mutex
 	buckets map[string]jetstream.KeyValue
+	ready   Readiness
+	nudge   chan struct{}
 }
 
 // NewMetaService builds the service over a JetStream handle.
@@ -58,7 +60,7 @@ func (s *MetaService) bucket(ctx context.Context, accountID string) (jetstream.K
 		return kv, nil
 	}
 	name := KVAccountBucket(accountID)
-	kv, err := kvstore.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
+	kv, err := kvutil.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
 	if err != nil {
 		return nil, fmt.Errorf("lambda: open account bucket %s: %w", name, err)
 	}
@@ -306,6 +308,7 @@ func (s *MetaService) CodeStored(ctx context.Context, req *CodeStoredRequest, ac
 		return nil, fmt.Errorf("lambda: update function record: %w", err)
 	}
 	attach(ctx, kv, up, upRev, rev)
+	s.wake()
 	return &CodeStoredResponse{Record: rec, Revision: rev}, nil
 }
 

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

@@ -10,7 +10,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.GetOrCreateBucket(ctx, js, BucketName(az), 5)
+	kv, err := kvutil.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 == kvstore.VersionKey {
+		if k == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.GetOrCreateBucket(t.Context(), js, name, 1)
+	kv, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 1)
+	vpcs, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 1)
 	if err != nil {
 		t.Fatalf("create VPC bucket: %v", err)
 	}
-	igws, err := kvstore.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 1)
+	igws, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 == kvstore.VersionKey }
+func keyIsVersion(key string) bool { return key == kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 
 	"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 := kvstore.GetOrCreateBucket(t.Context(), js, bucket, 10)
+	kv, err := kvutil.GetOrCreateBucket(t.Context(), js, bucket, 10)
 	if err != nil {
 		t.Fatalf("GetOrCreateBucket %s: %v", bucket, err)
 	}

spinifex/domains/ochre/usage.go

@@ -10,6 +10,7 @@ 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"
 )
 
@@ -195,7 +196,7 @@ func (s *UsageStore) TokensThisPeriod(ctx context.Context, accountID string) (in
 	if err != nil {
 		return 0, err
 	}
-	keys, err := kvstore.Keys(ctx, kv)
+	keys, err := kvutil.Keys(ctx, kv)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
 			return 0, nil

spinifex/domains/rds/instance/store.go

@@ -13,6 +13,7 @@ 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"
@@ -243,7 +244,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 := kvstore.BucketNames(ctx, js)
+	all, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 = kvstore.GetOrCreateBucket(t.Context(), js, "something-else", 1)
+	_, err = kvutil.GetOrCreateBucket(t.Context(), js, "something-else", 1)
 	require.NoError(t, err)
 
 	buckets, err := rdsinstance.AccountWatchBuckets(t.Context(), js)

spinifex/domains/route53/store.go

@@ -7,6 +7,7 @@ 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"
 )
 
@@ -144,11 +145,11 @@ func NewStore(js jetstream.JetStream) *Store {
 	}
 }
 
-func zoneKey(account, id string) string   { return keyPrefixZone + kvstore.AccountKey(account, id) }
-func rrsetsKey(account, id string) string { return keyPrefixRRSets + kvstore.AccountKey(account, id) }
-func changeKey(account, id string) string { return keyPrefixChange + kvstore.AccountKey(account, id) }
+func zoneKey(account, id string) string   { return keyPrefixZone + kvutil.AccountKey(account, id) }
+func rrsetsKey(account, id string) string { return keyPrefixRRSets + kvutil.AccountKey(account, id) }
+func changeKey(account, id string) string { return keyPrefixChange + kvutil.AccountKey(account, id) }
 func refKey(account, ref string) string {
-	return keyPrefixCallerRf + kvstore.AccountKey(account, refDigest(ref))
+	return keyPrefixCallerRf + kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	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 := kvstore.GetOrCreateBucketWithOptions(ctx, js, kvstore.BucketOptions{
+	kv, err := kvutil.GetOrCreateBucketWithOptions(ctx, js, kvutil.BucketOptions{
 		Name: bucket, Description: description, History: 1,
 	})
 	if err != nil {
 		return nil, err
 	}
-	return kvstore.VerifyDurable(ctx, kv, bucket)
+	return kvutil.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 kvstore.BindDurable(ctx, js, bucket)
+	return kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 kvstore.LeaderMsgGet(ctx, js, stream, kvstore.MsgGetRequest{LastFor: subject})
+	return kvutil.LeaderMsgGet(ctx, js, stream, kvutil.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 := kvstore.MsgGetRequest{NextFor: subjectPrefix + keyPrefix + ">"}
+	req := kvutil.MsgGetRequest{NextFor: subjectPrefix + keyPrefix + ">"}
 	keys := map[string]struct{}{}
 	for after := uint64(0); after < bound; {
 		req.Seq = after + 1
-		m, err := kvstore.LeaderMsgGet(ctx, js, stream, req)
+		m, err := kvutil.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/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"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 := kvstore.GetOrCreateBucketWithOptions(ctx, js, kvstore.BucketOptions{
+		kv, err := kvutil.GetOrCreateBucketWithOptions(ctx, js, kvutil.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 := kvstore.MsgGetRequest{NextFor: filter}
+	req := kvutil.MsgGetRequest{NextFor: filter}
 	for after < bound {
 		req.Seq = after + 1
-		msg, err := kvstore.LeaderMsgGet(ctx, js, stream, req)
+		msg, err := kvutil.LeaderMsgGet(ctx, js, stream, req)
 		switch {
 		case errors.Is(err, jsapi.ErrMsgNotFound):
 			return bound, true, nil