Compare commits

..

1 Commits

Author SHA1 Message Date
Bryn Mailer 1e3e0d5cec refactor: fold kvutil into kvstore and rename Claim to Take (#1608)
* refactor: fold kvutil into kvstore and rename Claim to Take

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

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

* refactor: merge the kvstore test exports into one file

* refactor: move the kvutil callers added on dev to kvstore
2026-11-08 10:50:44 +10:00
175 changed files with 803 additions and 1933 deletions
+1 -2
View File
@@ -14,7 +14,6 @@ import (
iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
)
@@ -184,7 +183,7 @@ func NewStore(ctx context.Context, nc *nats.Conn, masterKey []byte) (*Store, err
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketACM, KVBucketACMVersion)
kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketACM, KVBucketACMVersion)
if err != nil {
return nil, err
}
+2 -2
View File
@@ -10,7 +10,7 @@ import (
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
"golang.org/x/time/rate"
)
@@ -171,7 +171,7 @@ func (s *Service) RunBedrockRPMSync(ctx context.Context, js jetstream.JetStream)
if s == nil || !s.limits.RequestsPerMinuteEnabled {
return
}
kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketBedrockRPMUsage, 1)
kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketBedrockRPMUsage, 1)
if err != nil {
slog.Warn("bedrock rpm sync: open usage bucket failed, sync disabled", "err", err)
return
+3 -3
View File
@@ -9,7 +9,7 @@ import (
"time"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -187,7 +187,7 @@ func (s *Service) PutAccountQuota(ctx context.Context, accountID string, over Ov
over.UpdatedAt = time.Now().UTC().Format(time.RFC3339)
over.UpdatedBy = updatedBy
_, err := kvutil.Update(ctx, s.overrides, accountID, kvutil.CASConfig{
_, err := kvstore.Update(ctx, s.overrides, accountID, kvstore.CASConfig{
CreateIfAbsent: true,
Attempts: overrideMaxAttempts,
Exhausted: func(string, int) error { return errOverridesContended },
@@ -230,7 +230,7 @@ func (s *Service) clearAccountQuota(ctx context.Context, accountID string) error
return nil
case errors.Is(err, jetstream.ErrKeyNotFound):
return nil
case kvutil.IsRevisionConflict(err):
case kvstore.IsRevisionConflict(err):
// A concurrent write landed between the read and the guarded
// delete; retry against whatever revision now holds.
continue
+3 -3
View File
@@ -8,7 +8,7 @@ import (
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -73,7 +73,7 @@ func (s *Service) sealBucket(ctx context.Context) (jetstream.KeyValue, error) {
slog.ErrorContext(ctx, "Quota override seal has no JetStream client wired to open the override bucket")
return nil, errors.New(awserrors.ErrorInternalError)
}
kv, err := kvutil.GetOrCreateBucket(ctx, s.js, KVBucketAccountQuota, 1)
kv, err := kvstore.GetOrCreateBucket(ctx, s.js, KVBucketAccountQuota, 1)
if err != nil {
slog.ErrorContext(ctx, "Failed to open quota override bucket for seal", "err", err)
return nil, errors.New(awserrors.ErrorInternalError)
@@ -136,7 +136,7 @@ func (s *Service) SealAccountOverrides(ctx context.Context, accountID string) (*
switch {
case writeErr == nil:
return &OverrideSeal{Revision: newRev}, nil
case kvutil.IsRevisionConflict(writeErr):
case kvstore.IsRevisionConflict(writeErr):
continue
default:
slog.ErrorContext(ctx, "Failed to write sealed quota override record", "accountID", accountID, "err", writeErr)
+2 -2
View File
@@ -8,7 +8,7 @@ import (
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -109,7 +109,7 @@ func (s *AccountSettingsServiceImpl) SealAccount(ctx context.Context, accountID
switch {
case writeErr == nil:
return &AccountSettingsSeal{Revision: newRev}, nil
case kvutil.IsRevisionConflict(writeErr):
case kvstore.IsRevisionConflict(writeErr):
continue
default:
slog.ErrorContext(ctx, "Failed to write sealed account settings record", "accountID", accountID, "err", writeErr)
+7 -7
View File
@@ -13,7 +13,7 @@ import (
ebsencryption "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/encryption"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -57,7 +57,7 @@ func NewAccountSettingsServiceImplWithNATS(ctx context.Context, cfg *config.Conf
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
settingsKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketAccountSettings, 10)
settingsKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccountSettings, 10)
if err != nil {
return nil, fmt.Errorf("failed to create account settings KV bucket: %w", err)
}
@@ -115,13 +115,13 @@ func mapSettingsError(ctx context.Context, accountID string, err error) error {
return errors.New(awserrors.ErrorBlocked)
case errors.Is(err, errSettingsContended):
slog.ErrorContext(ctx, "Account settings CAS retries exhausted under contention", "accountID", accountID)
case errors.Is(err, kvutil.ErrRead):
case errors.Is(err, kvstore.ErrRead):
slog.ErrorContext(ctx, "Failed to read account settings from KV", "accountID", accountID, "err", err)
case errors.Is(err, kvutil.ErrDecode):
case errors.Is(err, kvstore.ErrDecode):
slog.ErrorContext(ctx, "Corrupt account settings record in KV", "accountID", accountID, "err", err)
case errors.Is(err, kvutil.ErrEncode):
case errors.Is(err, kvstore.ErrEncode):
slog.ErrorContext(ctx, "Failed to marshal account settings record", "accountID", accountID, "err", err)
case errors.Is(err, kvutil.ErrWrite):
case errors.Is(err, kvstore.ErrWrite):
slog.ErrorContext(ctx, "Failed to write account settings to KV", "accountID", accountID, "err", err)
default:
slog.ErrorContext(ctx, "Account settings operation failed", "accountID", accountID, "err", err)
@@ -133,7 +133,7 @@ func mapSettingsError(ctx context.Context, accountID string, err error) error {
// changed is always true, so every success is a leader-accepted write over an
// active value; a seal landing between the read and the write fails the CAS.
func (s *AccountSettingsServiceImpl) mutateSettings(ctx context.Context, accountID string, set func(*AccountSettingsRecord)) error {
_, err := kvutil.Update(ctx, s.settingsKV, settingsKey(accountID), kvutil.CASConfig{
_, err := kvstore.Update(ctx, s.settingsKV, settingsKey(accountID), kvstore.CASConfig{
CreateIfAbsent: true,
Exhausted: func(string, int) error { return errSettingsContended },
}, func(rec *AccountSettingsRecord) (bool, error) {
+7 -7
View File
@@ -19,7 +19,7 @@ import (
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -56,7 +56,7 @@ func NewEgressOnlyIGWServiceImplWithNATS(ctx context.Context, cfg *config.Config
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
eigwKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketEgressOnlyIGW, 10)
eigwKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketEgressOnlyIGW, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEgressOnlyIGW, err)
}
@@ -65,7 +65,7 @@ func NewEgressOnlyIGWServiceImplWithNATS(ctx context.Context, cfg *config.Config
}
// Get or create VPC KV bucket for cross-resource ownership validation
vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
if err != nil {
return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
}
@@ -90,7 +90,7 @@ func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(ctx context.C
slog.ErrorContext(ctx, "VPC KV unavailable, cannot verify VPC ownership")
return nil, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, *input.VpcId)); err != nil {
if _, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, *input.VpcId)); err != nil {
slog.WarnContext(ctx, "CreateEgressOnlyInternetGateway: VPC not found for account", "vpcId", *input.VpcId, "accountID", accountID)
return nil, awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", *input.VpcId)
}
@@ -109,7 +109,7 @@ func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(ctx context.C
if err != nil {
return nil, fmt.Errorf("failed to marshal Egress-only IGW record: %w", err)
}
if _, err := s.eigwKV.Put(ctx, kvutil.AccountKey(accountID, eigwID), data); err != nil {
if _, err := s.eigwKV.Put(ctx, kvstore.AccountKey(accountID, eigwID), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -127,7 +127,7 @@ func (s *EgressOnlyIGWServiceImpl) DeleteEgressOnlyInternetGateway(ctx context.C
}
eigwID := *input.EgressOnlyInternetGatewayId
key := kvutil.AccountKey(accountID, eigwID)
key := kvstore.AccountKey(accountID, eigwID)
// Verify the EIGW exists before deleting
if _, err := s.eigwKV.Get(ctx, key); err != nil {
@@ -178,7 +178,7 @@ func (s *EgressOnlyIGWServiceImpl) DescribeEgressOnlyInternetGateways(ctx contex
}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, prefix) {
@@ -9,7 +9,7 @@ import (
"github.com/mulgadc/spinifex/internal/testkit"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -23,7 +23,7 @@ func setupTestEIGWService(t *testing.T) *EgressOnlyIGWServiceImpl {
// Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass
vpcEntries := map[string][]byte{}
for _, vpcID := range []string{"vpc-test123", "vpc-tagged", "vpc-tagged2"} {
vpcEntries[kvutil.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
vpcEntries[kvstore.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
}
testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, vpcEntries)
@@ -312,7 +312,7 @@ func TestCreateEgressOnlyInternetGateway_CrossAccountVPCRejected(t *testing.T) {
// Create VPC KV bucket with a VPC owned by testAccountID
vpcID := "vpc-alpha456"
testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
kvutil.AccountKey(testAccountID, vpcID): []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`),
kvstore.AccountKey(testAccountID, vpcID): []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`),
})
var err error
@@ -12,7 +12,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -36,7 +36,7 @@ func (s *EIPServiceImpl) allocateOnce(ctx context.Context, accountID string, all
if key == "" || s.idemKV == nil {
return alloc()
}
kvKey := kvutil.AccountKey(accountID, idempotencyKeyHash(key))
kvKey := kvstore.AccountKey(accountID, idempotencyKeyHash(key))
if out := s.cachedAllocation(ctx, kvKey); out != nil {
slog.WarnContext(ctx, "AllocateAddress: returning first result for a retried request",
+3 -3
View File
@@ -10,7 +10,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -80,7 +80,7 @@ func (s *EIPServiceImpl) AllocationExists(ctx context.Context, allocationID stri
}
suffix := "." + allocationID
for _, k := range keys {
if k != kvutil.VersionKey && strings.HasSuffix(k, suffix) {
if k != kvstore.VersionKey && strings.HasSuffix(k, suffix) {
return true, nil
}
}
@@ -98,7 +98,7 @@ func (s *EIPServiceImpl) findByAllocationIDAnyAccount(ctx context.Context, alloc
suffix := "." + allocationID
for _, k := range keys {
if k == kvutil.VersionKey || !strings.HasSuffix(k, suffix) {
if k == kvstore.VersionKey || !strings.HasSuffix(k, suffix) {
continue
}
entry, err := s.eipKV.Get(ctx, k)
+12 -12
View File
@@ -20,7 +20,7 @@ import (
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -66,7 +66,7 @@ func NewEIPServiceImpl(ctx context.Context, natsConn *nats.Conn, externalIPAM *e
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
eipKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketEIPs, 10)
eipKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketEIPs, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEIPs, err)
}
@@ -76,7 +76,7 @@ func NewEIPServiceImpl(ctx context.Context, natsConn *nats.Conn, externalIPAM *e
// TTL'd, so a failure to create it is not worth refusing to serve EIPs over:
// a nil bucket only costs post-completion dedupe.
idemKV, err := kvutil.GetOrCreateBucketWithTTL(ctx, js, KVBucketEIPIdempotency, 1, eipIdempotencyTTL)
idemKV, err := kvstore.GetOrCreateBucketWithTTL(ctx, js, KVBucketEIPIdempotency, 1, eipIdempotencyTTL)
if err != nil {
slog.Warn("EIP service: idempotency bucket unavailable; a retried AllocateAddress may allocate twice",
"bucket", KVBucketEIPIdempotency, "err", err)
@@ -146,7 +146,7 @@ func (s *EIPServiceImpl) allocateAddress(ctx context.Context, input *ec2.Allocat
if err != nil {
return nil, fmt.Errorf("failed to marshal EIP record: %w", err)
}
if _, err := s.eipKV.Put(ctx, kvutil.AccountKey(accountID, allocID), data); err != nil {
if _, err := s.eipKV.Put(ctx, kvstore.AccountKey(accountID, allocID), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -167,7 +167,7 @@ func (s *EIPServiceImpl) ReleaseAddress(ctx context.Context, input *ec2.ReleaseA
}
allocID := *input.AllocationId
key := kvutil.AccountKey(accountID, allocID)
key := kvstore.AccountKey(accountID, allocID)
entry, err := s.eipKV.Get(ctx, key)
if err != nil {
@@ -206,7 +206,7 @@ func (s *EIPServiceImpl) AssociateAddress(ctx context.Context, input *ec2.Associ
}
allocID := *input.AllocationId
key := kvutil.AccountKey(accountID, allocID)
key := kvstore.AccountKey(accountID, allocID)
entry, err := s.eipKV.Get(ctx, key)
if err != nil {
@@ -536,7 +536,7 @@ func (s *EIPServiceImpl) findByENI(ctx context.Context, accountID, eniID string)
if err := ctx.Err(); err != nil {
return nil, err
}
if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
continue
}
entry, err := s.eipKV.Get(ctx, k)
@@ -590,7 +590,7 @@ func (s *EIPServiceImpl) DescribeAddresses(ctx context.Context, input *ec2.Descr
var addresses []*ec2.Address
for _, k := range keys {
if k == kvutil.VersionKey {
if k == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(k, prefix) {
@@ -699,7 +699,7 @@ func (s *EIPServiceImpl) DescribeAddressesAttribute(ctx context.Context, input *
if id == nil {
continue
}
key := kvutil.AccountKey(accountID, *id)
key := kvstore.AccountKey(accountID, *id)
entry, err := s.eipKV.Get(ctx, key)
if err != nil {
continue // not found, skip
@@ -722,7 +722,7 @@ func (s *EIPServiceImpl) DescribeAddressesAttribute(ctx context.Context, input *
return nil, errors.New(awserrors.ErrorServerInternal)
}
for _, k := range keys {
if k == kvutil.VersionKey {
if k == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(k, prefix) {
@@ -819,7 +819,7 @@ func (s *EIPServiceImpl) findByAssociationID(ctx context.Context, accountID, ass
}
for _, k := range keys {
if k == kvutil.VersionKey {
if k == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(k, prefix) {
@@ -857,7 +857,7 @@ func (s *EIPServiceImpl) AssociatedPublicIPForInstance(ctx context.Context, acco
return "", false
}
for _, k := range keys {
if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
continue
}
entry, err := s.eipKV.Get(ctx, k)
+15 -15
View File
@@ -22,7 +22,7 @@ import (
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -93,7 +93,7 @@ func NewIGWServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
igwKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketIGW, 10)
igwKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketIGW, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketIGW, err)
}
@@ -102,13 +102,13 @@ func NewIGWServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
}
// Get or create VPC KV bucket for cross-resource ownership validation
vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
if err != nil {
return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
}
// Read EIP associations so detach can refuse while addresses are mapped.
eipKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
eipKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
if err != nil {
return nil, fmt.Errorf("failed to get EIP KV bucket: %w", err)
}
@@ -142,7 +142,7 @@ func (s *IGWServiceImpl) createIGW(ctx context.Context, input *ec2.CreateInterne
if err != nil {
return nil, fmt.Errorf("failed to marshal IGW record: %w", err)
}
if _, err := s.igwKV.Put(ctx, kvutil.AccountKey(accountID, igwID), data); err != nil {
if _, err := s.igwKV.Put(ctx, kvstore.AccountKey(accountID, igwID), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -160,7 +160,7 @@ func (s *IGWServiceImpl) DeleteInternetGateway(ctx context.Context, input *ec2.D
}
igwID := *input.InternetGatewayId
key := kvutil.AccountKey(accountID, igwID)
key := kvstore.AccountKey(accountID, igwID)
entry, err := s.igwKV.Get(ctx, key)
if err != nil {
@@ -236,7 +236,7 @@ func (s *IGWServiceImpl) DescribeInternetGateways(ctx context.Context, input *ec
foundIDs := make(map[string]bool)
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, prefix) {
@@ -326,7 +326,7 @@ func (s *IGWServiceImpl) AttachInternetGateway(ctx context.Context, input *ec2.A
igwID := *input.InternetGatewayId
vpcID := *input.VpcId
key := kvutil.AccountKey(accountID, igwID)
key := kvstore.AccountKey(accountID, igwID)
entry, err := s.igwKV.Get(ctx, key)
if err != nil {
@@ -347,7 +347,7 @@ func (s *IGWServiceImpl) AttachInternetGateway(ctx context.Context, input *ec2.A
slog.ErrorContext(ctx, "VPC KV unavailable, cannot verify VPC ownership")
return nil, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID)); err != nil {
if _, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcID)); err != nil {
slog.WarnContext(ctx, "AttachInternetGateway: VPC not found for account", "vpcId", vpcID, "accountID", accountID)
return nil, awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", vpcID)
}
@@ -388,7 +388,7 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D
igwID := *input.InternetGatewayId
vpcID := *input.VpcId
key := kvutil.AccountKey(accountID, igwID)
key := kvstore.AccountKey(accountID, igwID)
entry, err := s.igwKV.Get(ctx, key)
if err != nil {
@@ -449,7 +449,7 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D
// associated into vpcID, whether to an interface or a NAT gateway. The listing
// is filtered server-side to the account's keys.
func (s *IGWServiceImpl) vpcHasMappedAddress(ctx context.Context, accountID, vpcID string) (bool, error) {
lister, err := s.eipKV.ListKeysFiltered(ctx, kvutil.AccountKey(accountID, "*"))
lister, err := s.eipKV.ListKeysFiltered(ctx, kvstore.AccountKey(accountID, "*"))
if err != nil {
return false, err
}
@@ -493,11 +493,11 @@ func (s *IGWServiceImpl) CreateAttachedInternetGateway(ctx context.Context, acco
if s.vpcKV == nil {
return false, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID)); err != nil {
if _, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcID)); err != nil {
return false, awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", vpcID)
}
key := kvutil.AccountKey(accountID, igwID)
key := kvstore.AccountKey(accountID, igwID)
data, err := json.Marshal(IGWRecord{
InternetGatewayId: igwID,
VpcId: vpcID,
@@ -558,7 +558,7 @@ func (s *IGWServiceImpl) AttachmentIntent(ctx context.Context, accountID, vpcID
}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.igwKV.Get(ctx, key)
@@ -599,7 +599,7 @@ func (s *IGWServiceImpl) AttachmentIntent(ctx context.Context, accountID, vpcID
// Writing only on the pending transition keeps a pass from waking the drift loop.
func MarkAttached(ctx context.Context, kv jetstream.KeyValue, recordKey, vpcID string) error {
confirmed := false
record, err := kvutil.Update(ctx, kv, recordKey, kvutil.CASConfig{}, func(r *IGWRecord) (bool, error) {
record, err := kvstore.Update(ctx, kv, recordKey, kvstore.CASConfig{}, func(r *IGWRecord) (bool, error) {
// Reset per attempt: a retried CAS may see a record another writer has
// already moved out of pending.
confirmed = r.VpcId == vpcID && r.AttachState == AttachStatePending
@@ -13,7 +13,7 @@ import (
ec2eip "github.com/mulgadc/spinifex/spinifex/domains/ec2/eip"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -55,7 +55,7 @@ func setupTestIGWService(t *testing.T) (*IGWServiceImpl, *nats.Conn) {
// Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass
vpcEntries := map[string][]byte{}
for _, vpcID := range []string{"vpc-test123", "vpc-other", "vpc-lifecycle", "vpc-event-test"} {
vpcEntries[kvutil.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
vpcEntries[kvstore.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
}
testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, vpcEntries)
@@ -68,7 +68,7 @@ func setupTestIGWService(t *testing.T) (*IGWServiceImpl, *nats.Conn) {
// gateway come up. Without it an attach stays pending and reports no attachment.
func confirmAttach(t *testing.T, svc *IGWServiceImpl, igwID, vpcID string) {
t.Helper()
require.NoError(t, MarkAttached(context.Background(), svc.igwKV, kvutil.AccountKey(testAccountID, igwID), vpcID))
require.NoError(t, MarkAttached(context.Background(), svc.igwKV, kvstore.AccountKey(testAccountID, igwID), vpcID))
}
func createTestIGW(t *testing.T, svc *IGWServiceImpl) string {
@@ -331,7 +331,7 @@ func TestDescribeInternetGateways_LegacyRecordReportsAttachment(t *testing.T) {
svc, _ := setupTestIGWService(t)
igwID := createTestIGW(t, svc)
key := kvutil.AccountKey(testAccountID, igwID)
key := kvstore.AccountKey(testAccountID, igwID)
entry, err := svc.igwKV.Get(context.Background(), key)
require.NoError(t, err)
var record IGWRecord
@@ -398,7 +398,7 @@ func TestMarkAttached_NoRewriteWhenAlreadyAttached(t *testing.T) {
}, testAccountID)
require.NoError(t, err)
key := kvutil.AccountKey(testAccountID, igwID)
key := kvstore.AccountKey(testAccountID, igwID)
confirmAttach(t, svc, igwID, "vpc-test123")
first, err := svc.igwKV.Get(context.Background(), key)
require.NoError(t, err)
@@ -605,7 +605,7 @@ func TestDetachInternetGateway_MappedPublicAddress(t *testing.T) {
putEIP := func(rec ec2eip.EIPRecord) {
data, err := json.Marshal(rec)
require.NoError(t, err)
_, err = svc.eipKV.Put(context.Background(), kvutil.AccountKey(testAccountID, rec.AllocationId), data)
_, err = svc.eipKV.Put(context.Background(), kvstore.AccountKey(testAccountID, rec.AllocationId), data)
require.NoError(t, err)
}
putEIP(ec2eip.EIPRecord{AllocationId: "eipalloc-other", AssociationId: "eipassoc-other", VpcId: "vpc-other", State: "associated"})
@@ -614,7 +614,7 @@ func TestDetachInternetGateway_MappedPublicAddress(t *testing.T) {
// Another account's address naming the same VPC ID is not this account's dependency.
foreign, err := json.Marshal(ec2eip.EIPRecord{AllocationId: "eipalloc-foreign", AssociationId: "eipassoc-foreign", VpcId: "vpc-test123", State: "associated"})
require.NoError(t, err)
_, err = svc.eipKV.Put(context.Background(), kvutil.AccountKey("999999999999", "eipalloc-foreign"), foreign)
_, err = svc.eipKV.Put(context.Background(), kvstore.AccountKey("999999999999", "eipalloc-foreign"), foreign)
require.NoError(t, err)
detach := func() error {
@@ -828,7 +828,7 @@ func TestAttachInternetGateway_CrossAccountVPCRejected(t *testing.T) {
require.NoError(t, err)
vpcID := "vpc-alpha123"
_, err = vpcKV.Put(t.Context(), kvutil.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`))
_, err = vpcKV.Put(t.Context(), kvstore.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`))
require.NoError(t, err)
// Refresh service to pick up the VPC KV bucket
@@ -10,7 +10,6 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -72,7 +71,7 @@ func (s *Store) InstanceIDs(ctx context.Context) ([]string, error) {
var ids []string
for key := range lister.Keys() {
if key != kvutil.VersionKey {
if key != kvstore.VersionKey {
ids = append(ids, key)
}
}
@@ -11,7 +11,6 @@ import (
testutil "github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/instance/uefistore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -60,7 +59,7 @@ func TestOpen_CreatesBucketWithNoTTLAndStampsVersion(t *testing.T) {
assert.Equal(t, int64(1), status.History())
assert.Zero(t, status.TTL(), "a TTL would expire deletion obligations before they are settled")
v, err := kvutil.ReadVersion(t.Context(), kv)
v, err := kvstore.ReadVersion(t.Context(), kv)
require.NoError(t, err)
assert.Equal(t, uefistore.BucketVersion, v)
}
+2 -2
View File
@@ -11,7 +11,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
)
@@ -51,7 +51,7 @@ func NewENITagReader(js jetstream.JetStream, accountID string) func(ctx context.
if err != nil {
return nil, false, fmt.Errorf("open ENI bucket: %w", err)
}
entry, err := kv.Get(ctx, kvutil.AccountKey(accountID, eniID))
entry, err := kv.Get(ctx, kvstore.AccountKey(accountID, eniID))
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, false, nil
}
@@ -6,7 +6,7 @@ import (
testutil "github.com/mulgadc/spinifex/internal/testkit"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -25,13 +25,13 @@ func TestENITagReaderSeparatesGoneFromUnreadable(t *testing.T) {
_, _, err := read(t.Context(), "eni-1")
require.Error(t, err, "a missing ENI bucket is a read failure, not an absent ENI")
enis, err := kvutil.GetOrCreateBucket(t.Context(), js, ec2vpc.KVBucketENIs, 10)
enis, err := kvstore.GetOrCreateBucket(t.Context(), js, ec2vpc.KVBucketENIs, 10)
require.NoError(t, err)
good, err := json.Marshal(ec2vpc.ENIRecord{NetworkInterfaceId: "eni-good", Tags: map[string]string{"k": "v"}})
require.NoError(t, err)
_, err = enis.Put(t.Context(), kvutil.AccountKey("000000000000", "eni-good"), good)
_, err = enis.Put(t.Context(), kvstore.AccountKey("000000000000", "eni-good"), good)
require.NoError(t, err)
_, err = enis.Put(t.Context(), kvutil.AccountKey("000000000000", "eni-corrupt"), []byte("{not json"))
_, err = enis.Put(t.Context(), kvstore.AccountKey("000000000000", "eni-corrupt"), []byte("{not json"))
require.NoError(t, err)
tags, found, err := read(t.Context(), "eni-good")
@@ -9,7 +9,7 @@ import (
"strings"
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -33,7 +33,7 @@ type AccountCleanup struct {
// Unreadable headers, their bodies and unrecognised entries are left for the
// owner scan to report. A listing failure is returned and nothing is removed.
func (s *LaunchTemplateServiceImpl) CleanupAccountResidue(ctx context.Context, accountID string, settledBefore time.Time) (*AccountCleanup, error) {
keys, err := kvutil.Keys(ctx, s.kv)
keys, err := kvstore.Keys(ctx, s.kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list launch-template keys: %w", err)
}
@@ -145,7 +145,7 @@ type AccountTemplateDeletes struct {
// it cannot read or decode is left for the owner scan to report. A listing
// failure is returned and nothing is deleted.
func (s *LaunchTemplateServiceImpl) DeleteLiveTemplates(ctx context.Context, accountID string, referenced func(id, name string) string) (*AccountTemplateDeletes, error) {
keys, err := kvutil.Keys(ctx, s.kv)
keys, err := kvstore.Keys(ctx, s.kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list launch-template keys: %w", err)
}
@@ -9,7 +9,7 @@ import (
"strconv"
"strings"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -37,7 +37,7 @@ type AccountScan struct {
// Headers and name-index entries are read; version bodies are classified from
// their keys alone, so template data and user data are never loaded.
func (s *LaunchTemplateServiceImpl) ScanAccount(ctx context.Context, accountID string) (*AccountScan, error) {
keys, err := kvutil.Keys(ctx, s.kv)
keys, err := kvstore.Keys(ctx, s.kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list launch-template keys: %w", err)
}
@@ -11,7 +11,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/internal/testkit/readonlystore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -45,7 +45,7 @@ func TestScanAccountClassifiesLaunchTemplateRecords(t *testing.T) {
// A header removed alone, as a delete interrupted after it leaves the rest.
deleted := createWithUserData(t, svc, "deleted")
inner := svc.kv
require.NoError(t, inner.Delete(t.Context(), kvutil.AccountKey(testAccountID, deleted)))
require.NoError(t, inner.Delete(t.Context(), kvstore.AccountKey(testAccountID, deleted)))
// A create whose header write and clean-up both fail.
svc.kv = &faultyKV{KeyValue: inner,
@@ -59,16 +59,16 @@ func TestScanAccountClassifiesLaunchTemplateRecords(t *testing.T) {
require.Error(t, err)
svc.kv = inner
putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0badbadbadbadbad0"), "{not json")
putRaw(t, inner, 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, 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, nameKey(testAccountID, "unreadable"), "lt-0unreadable000000")
putRaw(t, inner, kvutil.AccountKey(testAccountID, "lt-0bad.vx"), "{}")
putRaw(t, inner, kvstore.AccountKey(testAccountID, "lt-0bad.vx"), "{}")
recorder := &readRecordingKV{KeyValue: readonlystore.NewKV(t, inner)}
svc.kv = &faultyKV{KeyValue: recorder, failGet: func(key string) bool {
return key == kvutil.AccountKey(testAccountID, "lt-0unreadable000000")
return key == kvstore.AccountKey(testAccountID, "lt-0unreadable000000")
}}
scan, err := svc.ScanAccount(context.Background(), testAccountID)
require.NoError(t, err)
@@ -16,7 +16,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -159,7 +159,7 @@ func TestCurrentBehaviour_AccountTermination_LaunchTemplateStorageLayout(t *test
header := testAccountID + "." + id
name := testAccountID + ".name.776562283129" // hex of "web(1)"
assert.Equal(t, sortedKeys(header, header+".v1", header+".v2", name, kvutil.VersionKey), storedKeys(t, svc.kv))
assert.Equal(t, sortedKeys(header, header+".v1", header+".v2", name, kvstore.VersionKey), storedKeys(t, svc.kv))
entry, err := svc.kv.Get(t.Context(), name)
require.NoError(t, err)
@@ -223,13 +223,13 @@ func TestAccountTermination_DeleteLaunchTemplateKeepsTheTemplateWhenAStepFails(t
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
svc.kv = inner
assert.Equal(t, sortedKeys(append(tc.left(header, nameIdx), kvutil.VersionKey)...), storedKeys(t, inner))
assert.Equal(t, sortedKeys(append(tc.left(header, nameIdx), kvstore.VersionKey)...), storedKeys(t, inner))
assert.Equal(t, []string{id}, describedTemplateIDs(t, svc, testAccountID))
out, err := svc.DeleteLaunchTemplate(context.Background(), &ec2.DeleteLaunchTemplateInput{LaunchTemplateId: aws.String(id)}, testAccountID)
require.NoError(t, err)
assert.Equal(t, id, aws.StringValue(out.LaunchTemplate.LaunchTemplateId))
assert.Equal(t, []string{kvutil.VersionKey}, storedKeys(t, inner))
assert.Equal(t, []string{kvstore.VersionKey}, storedKeys(t, inner))
assert.Empty(t, describedTemplateIDs(t, svc, testAccountID))
})
}
@@ -260,7 +260,7 @@ func TestAccountTermination_DeleteLaunchTemplateRestartsWhenTheHeaderChanges(t *
_, err = svc.DeleteLaunchTemplate(context.Background(), &ec2.DeleteLaunchTemplateInput{LaunchTemplateId: aws.String(id)}, testAccountID)
require.NoError(t, err)
assert.True(t, raced)
assert.Equal(t, []string{kvutil.VersionKey}, storedKeys(t, inner))
assert.Equal(t, []string{kvstore.VersionKey}, storedKeys(t, inner))
}
// A create that fails after claiming the name removes what it wrote, so no
@@ -287,7 +287,7 @@ func TestAccountTermination_FailedCreateRemovesWhatItWrote(t *testing.T) {
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
svc.kv = inner
assert.Equal(t, []string{kvutil.VersionKey}, storedKeys(t, inner))
assert.Equal(t, []string{kvstore.VersionKey}, storedKeys(t, inner))
retried := createTemplate(t, svc, "web", "t3.micro")
assert.Equal(t, []string{aws.StringValue(retried.LaunchTemplateId)}, describedTemplateIDs(t, svc, testAccountID))
})
@@ -299,7 +299,7 @@ func TestAccountTermination_FailedCreateRemovesWhatItWrote(t *testing.T) {
// failure is an internal error, never a missing template.
func TestCurrentBehaviour_AccountTermination_DescribeLaunchTemplatesHidesUnreadableHeaders(t *testing.T) {
const brokenID = "lt-0badbadbadbadbad0"
brokenHeader := kvutil.AccountKey(testAccountID, brokenID)
brokenHeader := kvstore.AccountKey(testAccountID, brokenID)
tests := []struct {
name string
seed func(t *testing.T, svc *LaunchTemplateServiceImpl)
@@ -350,7 +350,7 @@ func TestAccountCleanup_KeepsTemplatesAndRemovesSettledResidue(t *testing.T) {
require.NoError(t, err)
orphanBody := versionKey(testAccountID, "lt-0orphan000000000", 1)
orphanName := nameKey(testAccountID, "gone")
brokenHeader := kvutil.AccountKey(testAccountID, "lt-0broken000000000")
brokenHeader := kvstore.AccountKey(testAccountID, "lt-0broken000000000")
brokenBody := versionKey(testAccountID, "lt-0broken000000000", 1)
brokenName := nameKey(testAccountID, "broken")
for key, value := range map[string]string{
@@ -384,7 +384,7 @@ func TestAccountCleanup_KeepsTemplatesAndRemovesSettledResidue(t *testing.T) {
func TestAccountCleanup_InterruptedTemplateDeleteIsLeftToThePublicRetry(t *testing.T) {
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
kv := svc.kv
svc.kv = &faultyKV{KeyValue: kv, failDelete: func(key string) bool { return key == header }}
_, err := svc.DeleteLaunchTemplate(context.Background(), &ec2.DeleteLaunchTemplateInput{LaunchTemplateId: aws.String(id)}, testAccountID)
@@ -19,7 +19,7 @@ import (
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -64,7 +64,7 @@ func NewLaunchTemplateServiceImplWithNATS(ctx context.Context, cfg *config.Confi
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketLaunchTemplates, 10)
kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketLaunchTemplates, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketLaunchTemplates, err)
}
@@ -87,24 +87,24 @@ func NewLaunchTemplateServiceImplWithNATS(ctx context.Context, cfg *config.Confi
// Launch template names allow characters such as parentheses that are not valid
// in NATS KV keys, so the name must not be used in its raw form here.
func nameKey(accountID, name string) string {
return kvutil.AccountKey(accountID, "name."+hex.EncodeToString([]byte(name)))
return kvstore.AccountKey(accountID, "name."+hex.EncodeToString([]byte(name)))
}
// versionKey returns a version-body key: account.lt-<id>.v<n>.
func versionKey(accountID, ltID string, n int64) string {
return kvutil.AccountKey(accountID, ltID+".v"+strconv.FormatInt(n, 10))
return kvstore.AccountKey(accountID, ltID+".v"+strconv.FormatInt(n, 10))
}
// versionPrefix returns the scan prefix for a template's version bodies.
func versionPrefix(accountID, ltID string) string {
return kvutil.AccountKey(accountID, ltID+".v")
return kvstore.AccountKey(accountID, ltID+".v")
}
// --- record load helpers ---
// getHeaderByID reads a header and its KV entry (for CAS) by launch template id.
func (s *LaunchTemplateServiceImpl) getHeaderByID(ctx context.Context, accountID, ltID string) (*LaunchTemplateHeader, jetstream.KeyValueEntry, error) {
entry, err := s.kv.Get(ctx, kvutil.AccountKey(accountID, ltID))
entry, err := s.kv.Get(ctx, kvstore.AccountKey(accountID, ltID))
if err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, nil, errors.New(awserrors.ErrorInvalidLaunchTemplateIdNotFound)
@@ -386,7 +386,7 @@ func (s *LaunchTemplateServiceImpl) claimName(ctx context.Context, accountID, na
// header means the name is taken; any other read error (transient fault, or a
// concurrent in-flight create whose header is not yet written) fails closed so
// the name is never stolen from a possibly-live template.
_, herr := s.kv.Get(ctx, kvutil.AccountKey(accountID, string(entry.Value())))
_, herr := s.kv.Get(ctx, kvstore.AccountKey(accountID, string(entry.Value())))
switch {
case herr == nil:
return errNameInUse()
@@ -404,7 +404,7 @@ func (s *LaunchTemplateServiceImpl) putHeader(ctx context.Context, accountID str
if err != nil {
return errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.kv.Put(ctx, kvutil.AccountKey(accountID, h.LaunchTemplateId), data); err != nil {
if _, err := s.kv.Put(ctx, kvstore.AccountKey(accountID, h.LaunchTemplateId), data); err != nil {
return errors.New(awserrors.ErrorServerInternal)
}
return nil
@@ -510,13 +510,13 @@ func (s *LaunchTemplateServiceImpl) commitVersion(ctx context.Context, accountID
// re-reading on a conflict. A non-nil check vets each re-read header and
// stops the commit with its error.
func (s *LaunchTemplateServiceImpl) commitHeader(ctx context.Context, op, accountID, ltID string, entry jetstream.KeyValueEntry, check func(*LaunchTemplateHeader) error) error {
key := kvutil.AccountKey(accountID, ltID)
key := kvstore.AccountKey(accountID, ltID)
for range maxVersionRetries {
_, err := s.kv.Update(ctx, key, entry.Value(), entry.Revision())
if err == nil {
return nil
}
if !kvutil.IsRevisionConflict(err) {
if !kvstore.IsRevisionConflict(err) {
slog.ErrorContext(ctx, op+": header commit failed", "launchTemplateId", ltID, "err", err)
return err
}
@@ -590,7 +590,7 @@ func (s *LaunchTemplateServiceImpl) ModifyLaunchTemplate(ctx context.Context, in
// header deleted meanwhile is NotFound for the identifier the request used.
func (s *LaunchTemplateServiceImpl) setDefaultVersion(ctx context.Context, accountID string, input *ec2.ModifyLaunchTemplateInput, header *LaunchTemplateHeader, entry jetstream.KeyValueEntry, sel string) (*LaunchTemplateHeader, error) {
notFound := fmt.Sprintf("The launch template version %s is not found for the specified launch template.", sel)
key := kvutil.AccountKey(accountID, header.LaunchTemplateId)
key := kvstore.AccountKey(accountID, header.LaunchTemplateId)
for range maxVersionRetries {
n, err := s.resolveVersionNumber(ctx, accountID, header, sel)
if err != nil {
@@ -611,7 +611,7 @@ func (s *LaunchTemplateServiceImpl) setDefaultVersion(ctx context.Context, accou
if err == nil {
return header, nil
}
if !kvutil.IsRevisionConflict(err) {
if !kvstore.IsRevisionConflict(err) {
slog.ErrorContext(ctx, "ModifyLaunchTemplate: header update failed", "launchTemplateId", header.LaunchTemplateId, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -662,7 +662,7 @@ const maxHeaderDeleteAttempts = 3
// through the public delete; a header changed in between restarts the sweep.
func (s *LaunchTemplateServiceImpl) deleteTemplate(ctx context.Context, accountID string, header *LaunchTemplateHeader, entry jetstream.KeyValueEntry) error {
ltID := header.LaunchTemplateId
headerKey := kvutil.AccountKey(accountID, ltID)
headerKey := kvstore.AccountKey(accountID, ltID)
for attempt := 1; ; attempt++ {
if err := s.deleteVersionBodies(ctx, accountID, ltID); err != nil {
return err
@@ -824,7 +824,7 @@ func (s *LaunchTemplateServiceImpl) commitVersionDeletes(ctx context.Context, ac
}
// A template delete that removed the header before the restore has already
// swept its bodies, so the restored ones would be left headerless.
if _, gerr := s.kv.Get(ctx, kvutil.AccountKey(accountID, ltID)); errors.Is(gerr, jetstream.ErrKeyNotFound) {
if _, gerr := s.kv.Get(ctx, kvstore.AccountKey(accountID, ltID)); errors.Is(gerr, jetstream.ErrKeyNotFound) {
for n := range deleted {
if derr := s.kv.Delete(ctx, versionKey(accountID, ltID, n)); derr != nil {
slog.ErrorContext(ctx, "DeleteLaunchTemplateVersions: failed to remove a restored version", "launchTemplateId", ltID, "version", n, "err", derr)
@@ -9,7 +9,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -116,7 +116,7 @@ func TestCreateLaunchTemplate_OrphanNameReclaim(t *testing.T) {
lt := createTemplate(t, svc, "orphan", "t3.micro")
// Simulate a crash after the name claim but with the header lost.
require.NoError(t, svc.kv.Delete(t.Context(), kvutil.AccountKey(testAccountID, aws.StringValue(lt.LaunchTemplateId))))
require.NoError(t, svc.kv.Delete(t.Context(), kvstore.AccountKey(testAccountID, aws.StringValue(lt.LaunchTemplateId))))
// The name index still points at the now-orphaned id; create must reclaim it.
out, err := svc.CreateLaunchTemplate(context.Background(), &ec2.CreateLaunchTemplateInput{
@@ -10,7 +10,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -38,7 +38,7 @@ func TestAccountTermination_VersionCreateRacingADeleteLeavesNoBody(t *testing.T)
t.Run(pause, func(t *testing.T) {
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
raced := false
race := func(key string, at string) {
if raced || pause != at {
@@ -71,7 +71,7 @@ func TestAccountTermination_DeleteRacingACommittedVersionRemovesIt(t *testing.T)
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
inner := svc.kv
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
raced := false
svc.kv = &faultyKV{KeyValue: inner, beforeDelete: func(key string) {
if key != header || raced {
@@ -95,7 +95,7 @@ func TestAccountTermination_FailedVersionCommitRemovesTheBody(t *testing.T) {
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
before := accountKeys(t, svc.kv, testAccountID)
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
svc.kv = &faultyKV{KeyValue: svc.kv, failUpdate: func(key string) bool { return key == header }}
_, err := svc.CreateLaunchTemplateVersion(context.Background(), newVersionInput(id), testAccountID)
@@ -110,7 +110,7 @@ func TestAccountTermination_VersionCommitKeepsAConcurrentHeaderChange(t *testing
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
inner := svc.kv
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
raced := false
svc.kv = &faultyKV{KeyValue: inner, beforeUpdate: func(key string) {
if key != header || raced {
@@ -215,10 +215,10 @@ func TestAccountTemplateReferences_FailsClosed(t *testing.T) {
require.NoError(t, err)
},
"header unreadable": func(t *testing.T, svc *LaunchTemplateServiceImpl, id string) {
svc.kv = &faultyKV{KeyValue: svc.kv, failGet: func(key string) bool { return key == kvutil.AccountKey(testAccountID, id) }}
svc.kv = &faultyKV{KeyValue: svc.kv, failGet: func(key string) bool { return key == kvstore.AccountKey(testAccountID, id) }}
},
"header undecodable": func(t *testing.T, svc *LaunchTemplateServiceImpl, _ string) {
_, err := svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "lt-0broken000000000"), []byte("{not json"))
_, err := svc.kv.Put(t.Context(), kvstore.AccountKey(testAccountID, "lt-0broken000000000"), []byte("{not json"))
require.NoError(t, err)
},
}
@@ -241,7 +241,7 @@ func TestModifyLaunchTemplate_RacingADeleteIsNotFound(t *testing.T) {
t.Run(map[bool]string{false: "by-id", true: "by-name"}[byName], func(t *testing.T) {
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
raced := false
svc.kv = &faultyKV{KeyValue: svc.kv, beforeUpdate: func(key string) {
if key == header && !raced {
@@ -275,7 +275,7 @@ func TestModifyLaunchTemplate_RetriesAConcurrentHeaderChange(t *testing.T) {
_, err := svc.CreateLaunchTemplateVersion(context.Background(), newVersionInput(id), testAccountID) // v3
require.NoError(t, err)
inner := svc.kv
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
raced := false
svc.kv = &faultyKV{KeyValue: inner, beforeUpdate: func(key string) {
if key != header || raced {
@@ -306,7 +306,7 @@ func TestModifyLaunchTemplate_ConflictRetriesAreBounded(t *testing.T) {
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
inner := svc.kv
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
attempts := 0
svc.kv = &faultyKV{KeyValue: inner, beforeUpdate: func(key string) {
if key != header {
@@ -333,7 +333,7 @@ func TestModifyLaunchTemplate_ConflictReReadFailureIsInternal(t *testing.T) {
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
inner := svc.kv
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
conflicted := false
svc.kv = &faultyKV{
KeyValue: inner,
@@ -410,7 +410,7 @@ func TestDefaultChangeRacingAVersionDeleteFailsNotFound(t *testing.T) {
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
inner := svc.kv
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
raced := false
svc.kv = &faultyKV{KeyValue: inner, beforeUpdate: func(key string) {
if key != header || raced {
@@ -460,7 +460,7 @@ func TestFailedVersionDeleteCommitRestoresTheBodies(t *testing.T) {
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
before := accountKeys(t, svc.kv, testAccountID)
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
svc.kv = &faultyKV{KeyValue: svc.kv, failUpdate: func(key string) bool { return key == header }}
_, err := deleteVersions(svc, id, "2")
@@ -474,7 +474,7 @@ func TestFailedVersionDeleteCommitRestoresTheBodies(t *testing.T) {
func TestVersionDeleteRacingATemplateDeleteLeavesNothing(t *testing.T) {
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
raced := false
svc.kv = &faultyKV{KeyValue: svc.kv, beforeUpdate: func(key string) {
if key != header || raced {
@@ -531,7 +531,7 @@ func TestDefaultChangeRetriesAfterAConcurrentVersionCreate(t *testing.T) {
svc := setupTestService(t)
id := createWithUserData(t, svc, "web")
inner := svc.kv
header := kvutil.AccountKey(testAccountID, id)
header := kvstore.AccountKey(testAccountID, id)
raced := false
svc.kv = &faultyKV{KeyValue: inner, beforeUpdate: func(key string) {
if key != header || raced {
@@ -8,7 +8,7 @@ import (
"strings"
"github.com/aws/aws-sdk-go/aws"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -27,7 +27,7 @@ type TemplateReferences struct {
// header or body it cannot read or decode, is an error, because the unread
// record may name the resource. Headerless bodies resolve from no launch.
func (s *LaunchTemplateServiceImpl) AccountTemplateReferences(ctx context.Context, accountID string) (*TemplateReferences, error) {
keys, err := kvutil.Keys(ctx, s.kv)
keys, err := kvstore.Keys(ctx, s.kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list launch-template keys: %w", err)
}
@@ -10,7 +10,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
ec2eip "github.com/mulgadc/spinifex/spinifex/domains/ec2/eip"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -42,7 +42,7 @@ func TestRLC3_NatGatewayDeletesWhileRouted(t *testing.T) {
natgwID := createTestNatGateway(t, svc)
testutil.SeedKV(t, js, kvBucketRouteTables, map[string][]byte{
kvutil.AccountKey(testAccountID, "rtb-routed"): fmt.Appendf(nil,
kvstore.AccountKey(testAccountID, "rtb-routed"): fmt.Appendf(nil,
`{"route_table_id":"rtb-routed","vpc_id":"vpc-test1","routes":[{"destination_cidr_block":"0.0.0.0/0","nat_gateway_id":%q}]}`, natgwID),
})
@@ -93,7 +93,7 @@ func createTestNatGateway(t *testing.T, svc *NatGatewayServiceImpl) string {
func readEIP(t *testing.T, svc *NatGatewayServiceImpl, allocID string) ec2eip.EIPRecord {
t.Helper()
entry, err := svc.eipKV.Get(t.Context(), kvutil.AccountKey(testAccountID, allocID))
entry, err := svc.eipKV.Get(t.Context(), kvstore.AccountKey(testAccountID, allocID))
require.NoError(t, err)
var eip ec2eip.EIPRecord
require.NoError(t, json.Unmarshal(entry.Value(), &eip))
+17 -17
View File
@@ -20,7 +20,7 @@ import (
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -45,7 +45,7 @@ func NewNatGatewayServiceImplWithNATS(ctx context.Context, natsConn *nats.Conn)
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
natgwKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketNatGateways, 10)
natgwKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketNatGateways, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketNatGateways, err)
}
@@ -61,17 +61,17 @@ func NewNatGatewayServiceImplWithNATS(ctx context.Context, natsConn *nats.Conn)
return nil, fmt.Errorf("migrate %s: %w", KVBucketDeletedNatGateways, err)
}
eipKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
eipKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2eip.KVBucketEIPs, 10)
if err != nil {
return nil, fmt.Errorf("failed to get EIP KV bucket: %w", err)
}
subnetKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
subnetKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
if err != nil {
return nil, fmt.Errorf("failed to get subnet KV bucket: %w", err)
}
vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
if err != nil {
return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
}
@@ -92,7 +92,7 @@ func NewNatGatewayServiceImplWithNATS(ctx context.Context, natsConn *nats.Conn)
// for as long as a client may still ask about it.
func getOrCreateDeletedBucket(ctx context.Context, js jetstream.KeyValueManager) (jetstream.KeyValue, error) {
// Terraform polls DescribeNatGateways after delete and expects state=deleted.
return kvutil.GetOrCreateBucketWithOptions(ctx, js, kvutil.BucketOptions{
return kvstore.GetOrCreateBucketWithOptions(ctx, js, kvstore.BucketOptions{
Name: KVBucketDeletedNatGateways,
Description: "Deleted NAT Gateways (auto-expire after 1 hour)",
History: 1,
@@ -113,7 +113,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
allocID := *input.AllocationId
// Validate subnet exists and get its VPC
subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetID))
subnetEntry, err := s.subnetKV.Get(ctx, kvstore.AccountKey(accountID, subnetID))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
}
@@ -123,7 +123,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
}
// Validate EIP exists and is not already associated
eipEntry, err := s.eipKV.Get(ctx, kvutil.AccountKey(accountID, allocID))
eipEntry, err := s.eipKV.Get(ctx, kvstore.AccountKey(accountID, allocID))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidAllocationIDNotFound)
}
@@ -152,7 +152,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
if err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.natgwKV.Put(ctx, kvutil.AccountKey(accountID, natgwID), data); err != nil {
if _, err := s.natgwKV.Put(ctx, kvstore.AccountKey(accountID, natgwID), data); err != nil {
slog.ErrorContext(ctx, "Failed to store NAT Gateway", "natGatewayId", natgwID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -165,7 +165,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
eipRecord.VpcId = subnetRecord.VpcId
eipRecord.State = "associated"
if eipData, err := json.Marshal(eipRecord); err == nil {
if _, err := s.eipKV.Update(ctx, kvutil.AccountKey(accountID, allocID), eipData, eipEntry.Revision()); err != nil {
if _, err := s.eipKV.Update(ctx, kvstore.AccountKey(accountID, allocID), eipData, eipEntry.Revision()); err != nil {
slog.WarnContext(ctx, "CreateNatGateway: failed to mark EIP associated", "natGatewayId", natgwID, "allocationId", allocID, "err", err)
}
}
@@ -185,7 +185,7 @@ func (s *NatGatewayServiceImpl) DeleteNatGateway(ctx context.Context, input *ec2
}
natgwID := *input.NatGatewayId
key := kvutil.AccountKey(accountID, natgwID)
key := kvstore.AccountKey(accountID, natgwID)
entry, err := s.natgwKV.Get(ctx, key)
if err != nil {
@@ -255,7 +255,7 @@ func (s *NatGatewayServiceImpl) publishDeleteEventsForNatGateway(ctx context.Con
slog.WarnContext(ctx, "DeleteNatGateway: JetStream setup failed, SNAT teardown deferred to route delete", "natGatewayId", record.NatGatewayId, "err", err)
return
}
rtbKV, err := kvutil.GetOrCreateBucket(ctx, js, kvBucketRouteTables, 10)
rtbKV, err := kvstore.GetOrCreateBucket(ctx, js, kvBucketRouteTables, 10)
if err != nil {
slog.WarnContext(ctx, "DeleteNatGateway: route-table scan failed, SNAT teardown deferred to route delete", "natGatewayId", record.NatGatewayId, "err", err)
return
@@ -266,7 +266,7 @@ func (s *NatGatewayServiceImpl) publishDeleteEventsForNatGateway(ctx context.Con
}
prefix := accountID + "."
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := rtbKV.Get(ctx, key)
@@ -302,7 +302,7 @@ func (s *NatGatewayServiceImpl) publishDeleteEventsForNatGateway(ctx context.Con
if assoc.SubnetId == "" {
continue
}
subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, assoc.SubnetId))
subnetEntry, err := s.subnetKV.Get(ctx, kvstore.AccountKey(accountID, assoc.SubnetId))
if err != nil {
continue
}
@@ -322,7 +322,7 @@ func (s *NatGatewayServiceImpl) disassociateEIP(ctx context.Context, record *Nat
if record.AllocationId == "" {
return
}
key := kvutil.AccountKey(accountID, record.AllocationId)
key := kvstore.AccountKey(accountID, record.AllocationId)
entry, err := s.eipKV.Get(ctx, key)
if err != nil {
if !errors.Is(err, jetstream.ErrKeyNotFound) {
@@ -380,7 +380,7 @@ func (s *NatGatewayServiceImpl) DescribeNatGateways(ctx context.Context, input *
foundIDs := make(map[string]bool)
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
@@ -415,7 +415,7 @@ func (s *NatGatewayServiceImpl) DescribeNatGateways(ctx context.Context, input *
if foundIDs[id] {
continue
}
key := kvutil.AccountKey(accountID, id)
key := kvstore.AccountKey(accountID, id)
entry, err := s.deletedNatgwKV.Get(ctx, key)
if err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
@@ -11,7 +11,7 @@ import (
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -31,18 +31,18 @@ func setupTestServiceJS(t *testing.T) (*NatGatewayServiceImpl, jetstream.JetStre
// Seed VPC KV
testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
kvutil.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
kvstore.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
})
// Seed subnet KV (public subnet)
testutil.SeedKV(t, js, ec2vpc.KVBucketSubnets, map[string][]byte{
kvutil.AccountKey(testAccountID, "subnet-pub1"): []byte(`{"subnet_id":"subnet-pub1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available","map_public_ip_on_launch":true}`),
kvstore.AccountKey(testAccountID, "subnet-pub1"): []byte(`{"subnet_id":"subnet-pub1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available","map_public_ip_on_launch":true}`),
})
// Seed EIP KV
testutil.SeedKV(t, js, ec2eip.KVBucketEIPs, map[string][]byte{
kvutil.AccountKey(testAccountID, "eipalloc-test1"): []byte(`{"allocation_id":"eipalloc-test1","public_ip":"203.0.113.50","state":"allocated"}`),
kvutil.AccountKey(testAccountID, "eipalloc-used"): []byte(`{"allocation_id":"eipalloc-used","public_ip":"203.0.113.51","state":"associated","association_id":"eipassoc-xxx"}`),
kvstore.AccountKey(testAccountID, "eipalloc-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"}`),
})
svc, err := NewNatGatewayServiceImplWithNATS(t.Context(), nc)
@@ -170,7 +170,7 @@ func TestDeleteNatGateway(t *testing.T) {
assert.Equal(t, natgwID, *deleteOut.NatGatewayId)
// Should be gone
_, err = svc.natgwKV.Get(t.Context(), kvutil.AccountKey(testAccountID, natgwID))
_, err = svc.natgwKV.Get(t.Context(), kvstore.AccountKey(testAccountID, natgwID))
assert.ErrorIs(t, err, jetstream.ErrKeyNotFound)
}
@@ -8,7 +8,7 @@ import (
"log/slog"
"strings"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -39,7 +39,7 @@ type AccountCleanup struct {
// referenced returns a reason is kept; unreadable records are left for the owner scan to report. A
// listing failure is returned and nothing is deleted.
func (s *PlacementGroupServiceImpl) CleanupAccount(ctx context.Context, accountID string, referenced func(groupID, groupName string) string) (*AccountCleanup, error) {
keys, err := kvutil.Keys(ctx, s.kv)
keys, err := kvstore.Keys(ctx, s.kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list placement-group keys: %w", err)
}
@@ -47,7 +47,7 @@ func (s *PlacementGroupServiceImpl) CleanupAccount(ctx context.Context, accountI
prefix := accountID + "."
out := &AccountCleanup{Referenced: map[string]string{}}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.kv.Get(ctx, key)
@@ -8,7 +8,7 @@ import (
"log/slog"
"strings"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -34,7 +34,7 @@ type AccountScan struct {
// A listing failure is returned as an error; per-record read failures are
// counted. The bucket's schema-version key belongs to no account.
func (s *PlacementGroupServiceImpl) ScanAccount(ctx context.Context, accountID string) (*AccountScan, error) {
keys, err := kvutil.Keys(ctx, s.kv)
keys, err := kvstore.Keys(ctx, s.kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list placement-group keys: %w", err)
}
@@ -42,7 +42,7 @@ func (s *PlacementGroupServiceImpl) ScanAccount(ctx context.Context, accountID s
prefix := accountID + "."
scan := &AccountScan{}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.kv.Get(ctx, key)
@@ -11,7 +11,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/internal/testkit/readonlystore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -42,14 +42,14 @@ func TestScanAccountClassifiesPlacementGroupRecords(t *testing.T) {
GroupName: "held", NodeInstances: map[string][]string{"node-2": {"i-0bbb", "i-0aaa"}, "node-3": {"i-0aaa"}},
}, testAccountID)
require.NoError(t, err)
_, err = svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "undecodable"), []byte("{not json"))
_, err = svc.kv.Put(t.Context(), kvstore.AccountKey(testAccountID, "undecodable"), []byte("{not json"))
require.NoError(t, err)
_, err = svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "anonymous"), []byte(`{"group_name":"anonymous"}`))
_, err = svc.kv.Put(t.Context(), kvstore.AccountKey(testAccountID, "anonymous"), []byte(`{"group_name":"anonymous"}`))
require.NoError(t, err)
createTestGroup(t, svc, "unreadable", ec2.PlacementStrategySpread)
require.Contains(t, bucketKeys(t, svc.kv), kvutil.VersionKey)
require.Contains(t, bucketKeys(t, svc.kv), kvstore.VersionKey)
svc.kv = &hookedKV{KeyValue: readonlystore.NewKV(t, svc.kv), failGet: kvutil.AccountKey(testAccountID, "unreadable")}
svc.kv = &hookedKV{KeyValue: readonlystore.NewKV(t, svc.kv), failGet: kvstore.AccountKey(testAccountID, "unreadable")}
scan, err := svc.ScanAccount(context.Background(), testAccountID)
require.NoError(t, err)
@@ -14,7 +14,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -84,7 +84,7 @@ func TestCurrentBehaviour_AccountTermination_PlacementGroupStorageLayout(t *test
}, testAccountID)
require.NoError(t, err)
assert.Equal(t, []string{testAccountID + ".spread-a", kvutil.VersionKey}, bucketKeys(t, svc.kv))
assert.Equal(t, []string{testAccountID + ".spread-a", kvstore.VersionKey}, bucketKeys(t, svc.kv))
entry, err := svc.kv.Get(t.Context(), testAccountID+".spread-a")
require.NoError(t, err)
@@ -100,7 +100,7 @@ func TestCurrentBehaviour_AccountTermination_PlacementGroupStorageLayout(t *test
// a record cannot be removed through DeletePlacementGroup either, and a read
// failure is an internal error, never an unknown group.
func TestCurrentBehaviour_AccountTermination_DescribePlacementGroupsHidesUnreadableRecords(t *testing.T) {
brokenKey := kvutil.AccountKey(testAccountID, "broken")
brokenKey := kvstore.AccountKey(testAccountID, "broken")
tests := []struct {
name string
seed func(t *testing.T, svc *PlacementGroupServiceImpl)
@@ -139,7 +139,7 @@ func TestCurrentBehaviour_AccountTermination_DescribePlacementGroupsHidesUnreada
func TestAccountTermination_DeletePlacementGroupLosesToAConcurrentMember(t *testing.T) {
svc := setupTestService(t)
createTestGroup(t, svc, "cluster-a", ec2.PlacementStrategyCluster)
key := kvutil.AccountKey(testAccountID, "cluster-a")
key := kvstore.AccountKey(testAccountID, "cluster-a")
launcher := &PlacementGroupServiceImpl{kv: svc.kv, region: svc.region}
var reserveErr, finalizeErr error
@@ -222,7 +222,7 @@ func TestAccountCleanup_DeletesEmptyGroupsAndBlocksOnMembers(t *testing.T) {
empty := createTestGroup(t, svc, "empty", ec2.PlacementStrategySpread)
held := createTestGroup(t, svc, "held", ec2.PlacementStrategyCluster)
require.NoError(t, addMember(t, svc, "held", "i-0123456789abcdef0"))
_, err := svc.kv.Put(t.Context(), kvutil.AccountKey(testAccountID, "broken"), []byte("{not json"))
_, err := svc.kv.Put(t.Context(), kvstore.AccountKey(testAccountID, "broken"), []byte("{not json"))
require.NoError(t, err)
other := "999988887777"
_, err = svc.CreatePlacementGroup(context.Background(), &ec2.CreatePlacementGroupInput{
@@ -238,7 +238,7 @@ func TestAccountCleanup_DeletesEmptyGroupsAndBlocksOnMembers(t *testing.T) {
assert.Equal(t, []string{"held"}, describedGroupNames(t, svc, testAccountID))
assert.Equal(t, []string{"theirs"}, describedGroupNames(t, svc, other))
_, err = svc.kv.Get(t.Context(), kvutil.AccountKey(testAccountID, "broken"))
_, err = svc.kv.Get(t.Context(), kvstore.AccountKey(testAccountID, "broken"))
require.NoError(t, err, "an unreadable record is never deleted")
}
@@ -247,7 +247,7 @@ func TestAccountCleanup_DeletesEmptyGroupsAndBlocksOnMembers(t *testing.T) {
func TestAccountCleanup_ConcurrentMemberBlocksTheDelete(t *testing.T) {
svc := setupTestService(t)
group := createTestGroup(t, svc, "cluster-a", ec2.PlacementStrategyCluster)
key := kvutil.AccountKey(testAccountID, "cluster-a")
key := kvstore.AccountKey(testAccountID, "cluster-a")
launcher := &PlacementGroupServiceImpl{kv: svc.kv, region: svc.region}
var memberErr error
// Read 1 is the cleanup's listing read; read 2 is the delete's check.
@@ -17,7 +17,7 @@ import (
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -87,7 +87,7 @@ func NewPlacementGroupServiceImplWithNATS(ctx context.Context, cfg *config.Confi
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketPlacementGroups, 10)
kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketPlacementGroups, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketPlacementGroups, err)
}
@@ -137,7 +137,7 @@ func (s *PlacementGroupServiceImpl) CreatePlacementGroup(ctx context.Context, in
}
groupName := *input.GroupName
key := kvutil.AccountKey(accountID, groupName)
key := kvstore.AccountKey(accountID, groupName)
groupID := awsidentifiers.GenerateResourceID("pg")
record := PlacementGroupRecord{
@@ -175,7 +175,7 @@ func (s *PlacementGroupServiceImpl) DeletePlacementGroup(ctx context.Context, in
}
groupName := *input.GroupName
if _, err := s.deleteGroup(ctx, accountID, kvutil.AccountKey(accountID, groupName)); err != nil {
if _, err := s.deleteGroup(ctx, accountID, kvstore.AccountKey(accountID, groupName)); err != nil {
if errors.Is(err, errPGAbsent) {
// Lower case, unlike DescribePlacementGroups' message for the same code.
return nil, awserrors.Errorf(awserrors.ErrorInvalidPlacementGroupUnknown, "The placement group '%s' is unknown.", groupName)
@@ -194,7 +194,7 @@ func (s *PlacementGroupServiceImpl) DeletePlacementGroup(ctx context.Context, in
// The returned count is the members seen by the last check.
func (s *PlacementGroupServiceImpl) deleteGroup(ctx context.Context, accountID, key string) (int, error) {
members := 0
record, err := kvutil.DeleteIf(ctx, s.kv, key, casConfig(), func(record *PlacementGroupRecord) error {
record, err := kvstore.DeleteIf(ctx, s.kv, key, casConfig(), func(record *PlacementGroupRecord) error {
members = memberCount(record.NodeInstances)
if members > 0 {
return errPGInUse
@@ -238,7 +238,7 @@ func (s *PlacementGroupServiceImpl) DescribePlacementGroups(ctx context.Context,
var groups []*ec2.PlacementGroup
for _, k := range keys {
if k == kvutil.VersionKey {
if k == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(k, prefix) {
@@ -341,7 +341,7 @@ func pgMatchesFilters(record *PlacementGroupRecord, filters map[string][]string)
const maxCASRetries = 5
// errPGAbsent and errPGContended mark the two kvutil outcomes that map to
// errPGAbsent and errPGContended mark the two kvstore outcomes that map to
// specific AWS errors rather than a generic internal failure, and neither
// escapes this file. errPGInUse is the AWS error itself.
var (
@@ -350,8 +350,8 @@ var (
errPGInUse = errors.New(awserrors.ErrorInvalidPlacementGroupInUse)
)
// mapCASError turns a kvutil failure into the AWS error the caller returns.
// A non-kvutil error came from mutate and is already an AWS code.
// 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.
func mapCASError(ctx context.Context, op, groupName, accountID string, err error) error {
switch {
case errors.Is(err, errPGAbsent):
@@ -359,16 +359,16 @@ func mapCASError(ctx context.Context, op, groupName, accountID string, err error
case errors.Is(err, errPGContended):
slog.ErrorContext(ctx, "Placement group CAS retries exhausted under contention",
"op", op, "groupName", groupName, "accountID", accountID)
case errors.Is(err, kvutil.ErrRead):
case errors.Is(err, kvstore.ErrRead):
slog.ErrorContext(ctx, "Failed to read placement group from KV",
"op", op, "groupName", groupName, "accountID", accountID, "err", err)
case errors.Is(err, kvutil.ErrDecode):
case errors.Is(err, kvstore.ErrDecode):
slog.ErrorContext(ctx, "Corrupt placement group record in KV",
"op", op, "groupName", groupName, "accountID", accountID, "err", err)
case errors.Is(err, kvutil.ErrEncode):
case errors.Is(err, kvstore.ErrEncode):
slog.ErrorContext(ctx, "Failed to marshal placement group record",
"op", op, "groupName", groupName, "accountID", accountID, "err", err)
case errors.Is(err, kvutil.ErrWrite):
case errors.Is(err, kvstore.ErrWrite):
slog.ErrorContext(ctx, "Failed to write placement group to KV",
"op", op, "groupName", groupName, "accountID", accountID, "err", err)
default:
@@ -378,8 +378,8 @@ func mapCASError(ctx context.Context, op, groupName, accountID string, err error
}
// casConfig is the shared CAS policy for every placement group mutation.
func casConfig() kvutil.CASConfig {
return kvutil.CASConfig{
func casConfig() kvstore.CASConfig {
return kvstore.CASConfig{
Attempts: maxCASRetries,
NotFound: errPGAbsent,
Exhausted: func(string, int) error { return errPGContended },
@@ -394,7 +394,7 @@ func (s *PlacementGroupServiceImpl) ReserveSpreadNodes(ctx context.Context, inpu
}
var selected []string
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
if record.State != ec2.PlacementGroupStateAvailable {
return false, errors.New(awserrors.ErrorInvalidPlacementGroupUnknown)
@@ -437,7 +437,7 @@ func (s *PlacementGroupServiceImpl) FinalizeSpreadInstances(ctx context.Context,
return nil, errors.New(awserrors.ErrorMissingParameter)
}
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
maps.Copy(record.NodeInstances, input.NodeInstances)
return true, nil
@@ -462,7 +462,7 @@ func (s *PlacementGroupServiceImpl) ReleaseSpreadNodes(ctx context.Context, inpu
releaseSet[n] = true
}
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
changed := false
for node := range releaseSet {
@@ -490,7 +490,7 @@ func (s *PlacementGroupServiceImpl) RemoveInstance(ctx context.Context, input *R
return nil, errors.New(awserrors.ErrorMissingParameter)
}
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
instances, exists := record.NodeInstances[input.NodeName]
if !exists {
@@ -533,7 +533,7 @@ func (s *PlacementGroupServiceImpl) ReserveClusterNode(ctx context.Context, inpu
}
var targetNode string
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
if record.State != ec2.PlacementGroupStateAvailable {
return false, errors.New(awserrors.ErrorInvalidPlacementGroupUnknown)
@@ -570,7 +570,7 @@ func (s *PlacementGroupServiceImpl) FinalizeClusterInstances(ctx context.Context
return nil, errors.New(awserrors.ErrorMissingParameter)
}
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvstore.Update(ctx, s.kv, kvstore.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
// Append rather than replace: a cluster group can have several launches in flight at once.
for node, ids := range input.NodeInstances {
@@ -10,7 +10,7 @@ import (
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -39,7 +39,7 @@ func createTestGroup(t *testing.T, svc *PlacementGroupServiceImpl, name, strateg
// readRecord returns the stored record for assertions.
func readRecord(t *testing.T, svc *PlacementGroupServiceImpl, groupName string) *PlacementGroupRecord {
t.Helper()
entry, err := svc.kv.Get(t.Context(), kvutil.AccountKey(testAccountID, groupName))
entry, err := svc.kv.Get(t.Context(), kvstore.AccountKey(testAccountID, groupName))
require.NoError(t, err)
var record PlacementGroupRecord
require.NoError(t, json.Unmarshal(entry.Value(), &record))
@@ -50,7 +50,7 @@ func readRecord(t *testing.T, svc *PlacementGroupServiceImpl, groupName string)
// starting state the public API cannot produce.
func seedRecord(t *testing.T, svc *PlacementGroupServiceImpl, groupName string, mutate func(*PlacementGroupRecord)) {
t.Helper()
_, err := kvutil.Update(t.Context(), svc.kv, kvutil.AccountKey(testAccountID, groupName), kvutil.CASConfig{},
_, err := kvstore.Update(t.Context(), svc.kv, kvstore.AccountKey(testAccountID, groupName), kvstore.CASConfig{},
func(record *PlacementGroupRecord) (bool, error) {
mutate(record)
return true, nil
+29 -29
View File
@@ -25,7 +25,7 @@ import (
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -52,7 +52,7 @@ func NewRouteTableServiceImplWithNATS(ctx context.Context, cfg *config.Config, n
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
rtbKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketRouteTables, 10)
rtbKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketRouteTables, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketRouteTables, err)
}
@@ -60,22 +60,22 @@ func NewRouteTableServiceImplWithNATS(ctx context.Context, cfg *config.Config, n
return nil, fmt.Errorf("migrate %s: %w", KVBucketRouteTables, err)
}
vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 10)
if err != nil {
return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
}
igwKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 10)
igwKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 10)
if err != nil {
return nil, fmt.Errorf("failed to get IGW KV bucket: %w", err)
}
subnetKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
subnetKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketSubnets, 10)
if err != nil {
return nil, fmt.Errorf("failed to get subnet KV bucket: %w", err)
}
natgwKV, err := kvutil.GetOrCreateBucket(ctx, js, ec2natgw.KVBucketNatGateways, 10)
natgwKV, err := kvstore.GetOrCreateBucket(ctx, js, ec2natgw.KVBucketNatGateways, 10)
if err != nil {
return nil, fmt.Errorf("failed to get NAT Gateway KV bucket: %w", err)
}
@@ -95,7 +95,7 @@ func NewRouteTableServiceImplWithNATS(ctx context.Context, cfg *config.Config, n
// getRouteTable retrieves a route table record from KV.
func (s *RouteTableServiceImpl) getRouteTable(ctx context.Context, accountID, rtbID string) (*RouteTableRecord, error) {
entry, err := s.rtbKV.Get(ctx, kvutil.AccountKey(accountID, rtbID))
entry, err := s.rtbKV.Get(ctx, kvstore.AccountKey(accountID, rtbID))
if err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, rtbNotFoundError(rtbID)
@@ -118,7 +118,7 @@ func (s *RouteTableServiceImpl) putRouteTable(ctx context.Context, accountID str
slog.ErrorContext(ctx, "Failed to marshal route table record", "routeTableId", record.RouteTableId, "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.rtbKV.Put(ctx, kvutil.AccountKey(accountID, record.RouteTableId), data); err != nil {
if _, err := s.rtbKV.Put(ctx, kvstore.AccountKey(accountID, record.RouteTableId), data); err != nil {
slog.ErrorContext(ctx, "Failed to write route table to KV", "routeTableId", record.RouteTableId, "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
@@ -130,7 +130,7 @@ func (s *RouteTableServiceImpl) putRouteTable(ctx context.Context, accountID str
// parallel AssociateRouteTable calls Terraform fans out per route table.
const rtbCASMaxRetries = 16
// errRTBAbsent and errRTBContended mark the two kvutil outcomes that map to
// errRTBAbsent and errRTBContended mark the two kvstore outcomes that map to
// specific AWS errors rather than a generic internal failure. Neither
// escapes this file.
var (
@@ -143,7 +143,7 @@ var (
// several subnets with one route table at once. mutate reports whether
// it changed the record; a false return commits nothing.
func (s *RouteTableServiceImpl) mutateRouteTableCAS(ctx context.Context, accountID, rtbID string, mutate func(*RouteTableRecord) (bool, error)) error {
_, err := kvutil.Update(ctx, s.rtbKV, kvutil.AccountKey(accountID, rtbID), kvutil.CASConfig{
_, err := kvstore.Update(ctx, s.rtbKV, kvstore.AccountKey(accountID, rtbID), kvstore.CASConfig{
Attempts: rtbCASMaxRetries,
NotFound: errRTBAbsent,
Exhausted: func(string, int) error { return errRTBContended },
@@ -157,13 +157,13 @@ func (s *RouteTableServiceImpl) mutateRouteTableCAS(ctx context.Context, account
// Contended rather than broken.
slog.ErrorContext(ctx, "Route table CAS retries exhausted under contention",
"routeTableId", rtbID, "attempts", rtbCASMaxRetries)
case errors.Is(err, kvutil.ErrRead):
case errors.Is(err, kvstore.ErrRead):
slog.ErrorContext(ctx, "Failed to read route table from KV", "routeTableId", rtbID, "err", err)
case errors.Is(err, kvutil.ErrDecode):
case errors.Is(err, kvstore.ErrDecode):
slog.ErrorContext(ctx, "Corrupt route table record in KV", "routeTableId", rtbID, "err", err)
case errors.Is(err, kvutil.ErrEncode):
case errors.Is(err, kvstore.ErrEncode):
slog.ErrorContext(ctx, "Failed to marshal route table record", "routeTableId", rtbID, "err", err)
case errors.Is(err, kvutil.ErrWrite):
case errors.Is(err, kvstore.ErrWrite):
slog.ErrorContext(ctx, "Failed to write route table to KV", "routeTableId", rtbID, "err", err)
default:
return err
@@ -173,7 +173,7 @@ func (s *RouteTableServiceImpl) mutateRouteTableCAS(ctx context.Context, account
// getVPCCidr looks up a VPC's CIDR block from the VPC KV bucket.
func (s *RouteTableServiceImpl) getVPCCidr(ctx context.Context, accountID, vpcID string) (string, error) {
entry, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID))
entry, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcID))
if err != nil {
return "", awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", vpcID)
}
@@ -229,7 +229,7 @@ func (s *RouteTableServiceImpl) subnetsImplicitlyOnMainRT(ctx context.Context, a
}
var implicit []string
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.subnetKV.Get(ctx, key)
@@ -387,7 +387,7 @@ func (s *RouteTableServiceImpl) allSubnetsForVPC(ctx context.Context, accountID,
}
var subnets []string
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.subnetKV.Get(ctx, key)
@@ -455,7 +455,7 @@ func (s *RouteTableServiceImpl) allRouteTablesForVPC(ctx context.Context, accoun
var results []RouteTableRecord
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.rtbKV.Get(ctx, key)
@@ -577,7 +577,7 @@ func (s *RouteTableServiceImpl) DeleteRouteTable(ctx context.Context, input *ec2
}
}
if err := s.rtbKV.Delete(ctx, kvutil.AccountKey(accountID, rtbID)); err != nil {
if err := s.rtbKV.Delete(ctx, kvstore.AccountKey(accountID, rtbID)); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -633,7 +633,7 @@ func (s *RouteTableServiceImpl) DescribeRouteTables(ctx context.Context, input *
foundIDs := make(map[string]bool)
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
@@ -740,7 +740,7 @@ func (s *RouteTableServiceImpl) CreateRoute(ctx context.Context, input *ec2.Crea
case input.GatewayId != nil && *input.GatewayId != "":
igwID := *input.GatewayId
// Verify IGW exists and is attached to the same VPC
igwEntry, err := s.igwKV.Get(ctx, kvutil.AccountKey(accountID, igwID))
igwEntry, err := s.igwKV.Get(ctx, kvstore.AccountKey(accountID, igwID))
if err != nil {
return nil, gatewayNotFoundError(igwID)
}
@@ -770,7 +770,7 @@ func (s *RouteTableServiceImpl) CreateRoute(ctx context.Context, input *ec2.Crea
case input.NatGatewayId != nil && *input.NatGatewayId != "":
natgwID := *input.NatGatewayId
// Verify NAT GW exists and belongs to the same VPC
natgwEntry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, natgwID))
natgwEntry, err := s.natgwKV.Get(ctx, kvstore.AccountKey(accountID, natgwID))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound)
}
@@ -924,7 +924,7 @@ func (s *RouteTableServiceImpl) ReplaceRoute(ctx context.Context, input *ec2.Rep
igwID := *input.GatewayId
// AWS reports a missing gateway ahead of a missing route.
igwEntry, err := s.igwKV.Get(ctx, kvutil.AccountKey(accountID, igwID))
igwEntry, err := s.igwKV.Get(ctx, kvstore.AccountKey(accountID, igwID))
if err != nil {
return nil, gatewayNotFoundError(igwID)
}
@@ -983,7 +983,7 @@ func (s *RouteTableServiceImpl) AssociateRouteTable(ctx context.Context, input *
}
// Verify subnet exists and belongs to the same VPC
subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetID))
subnetEntry, err := s.subnetKV.Get(ctx, kvstore.AccountKey(accountID, subnetID))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
}
@@ -1077,7 +1077,7 @@ func (s *RouteTableServiceImpl) DisassociateRouteTable(ctx context.Context, inpu
}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
@@ -1168,7 +1168,7 @@ func (s *RouteTableServiceImpl) ReplaceRouteTableAssociation(ctx context.Context
}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
@@ -1369,7 +1369,7 @@ func (s *RouteTableServiceImpl) publishNatGatewayDeleteEvents(ctx context.Contex
if s.natsConn == nil {
return
}
natgwEntry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, natgwID))
natgwEntry, err := s.natgwKV.Get(ctx, kvstore.AccountKey(accountID, natgwID))
if err != nil {
slog.WarnContext(ctx, "NAT GW event: natgw lookup failed", "topic", networkv1.NATGatewayDeleteSubject, "natGatewayId", natgwID, "err", err)
return
@@ -1417,7 +1417,7 @@ func (s *RouteTableServiceImpl) publishNatGatewayEventsForAssociation(ctx contex
if r.NatGatewayId == "" {
continue
}
natgwEntry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, r.NatGatewayId))
natgwEntry, err := s.natgwKV.Get(ctx, kvstore.AccountKey(accountID, r.NatGatewayId))
if err != nil {
slog.WarnContext(ctx, "NAT GW event: natgw lookup failed", "topic", topic, "natGatewayId", r.NatGatewayId, "err", err)
continue
@@ -1442,7 +1442,7 @@ func (s *RouteTableServiceImpl) publishNatGatewayEventForSubnet(ctx context.Cont
if s.natsConn == nil {
return
}
subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetID))
subnetEntry, err := s.subnetKV.Get(ctx, kvstore.AccountKey(accountID, subnetID))
if err != nil {
slog.WarnContext(ctx, "NAT GW event: subnet lookup failed", "topic", topic, "subnetId", subnetID, "err", err)
return
@@ -13,7 +13,7 @@ import (
ec2natgw "github.com/mulgadc/spinifex/spinifex/domains/ec2/natgw"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -28,23 +28,23 @@ func setupTestService(t *testing.T) *RouteTableServiceImpl {
// Seed VPC KV
testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
kvutil.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
kvstore.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
})
// Seed IGW KV (attached to vpc-test1)
testutil.SeedKV(t, js, ec2igw.KVBucketIGW, map[string][]byte{
kvutil.AccountKey(testAccountID, "igw-test1"): []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"available"}`),
kvstore.AccountKey(testAccountID, "igw-test1"): []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"available"}`),
})
// Seed subnet KV
testutil.SeedKV(t, js, ec2vpc.KVBucketSubnets, map[string][]byte{
kvutil.AccountKey(testAccountID, "subnet-test1"): []byte(`{"subnet_id":"subnet-test1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available"}`),
kvutil.AccountKey(testAccountID, "subnet-priv1"): []byte(`{"subnet_id":"subnet-priv1","vpc_id":"vpc-test1","cidr_block":"10.0.11.0/24","state":"available"}`),
kvstore.AccountKey(testAccountID, "subnet-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"}`),
})
// Seed NAT Gateway KV
testutil.SeedKV(t, js, ec2natgw.KVBucketNatGateways, map[string][]byte{
kvutil.AccountKey(testAccountID, "nat-test1"): []byte(`{"nat_gateway_id":"nat-test1","vpc_id":"vpc-test1","subnet_id":"subnet-test1","public_ip":"192.168.1.100","state":"available"}`),
kvstore.AccountKey(testAccountID, "nat-test1"): []byte(`{"nat_gateway_id":"nat-test1","vpc_id":"vpc-test1","subnet_id":"subnet-test1","public_ip":"192.168.1.100","state":"available"}`),
})
svc, err := NewRouteTableServiceImplWithNATS(t.Context(), nil, nc)
@@ -10,7 +10,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/nats-io/nats.go/jetstream"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
// AccountSpotRequest is one persisted request in an account scan.
@@ -54,7 +54,7 @@ func scanBucket(ctx context.Context, kv jetstream.KeyValue, accountID string) ([
prefix := accountID + "."
var out []AccountSpotRequest
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := kv.Get(ctx, key)
@@ -11,7 +11,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
func TestScanAccountListsActiveAndTerminalForOneAccount(t *testing.T) {
@@ -35,7 +35,7 @@ func TestScanAccountListsActiveAndTerminalForOneAccount(t *testing.T) {
func TestScanAccountFailsOnAnUndecodableRecord(t *testing.T) {
svc := setupTestService(t)
putSIRs(t, svc, makeSIR("sir-aaa", "i-aaa"))
_, err := svc.activeKV.Put(t.Context(), kvutil.AccountKey(testAccountID, "sir-bad"), []byte("{not json"))
_, err := svc.activeKV.Put(t.Context(), kvstore.AccountKey(testAccountID, "sir-bad"), []byte("{not json"))
require.NoError(t, err)
scan, err := svc.ScanAccount(t.Context(), testAccountID)
@@ -45,7 +45,7 @@ func TestScanAccountFailsOnAnUndecodableRecord(t *testing.T) {
func TestScanAccountFailsOnARecordWithoutARequest(t *testing.T) {
svc := setupTestService(t)
_, err := svc.terminalKV.Put(t.Context(), kvutil.AccountKey(testAccountID, "sir-empty"), []byte(`{"account_id":"x"}`))
_, err := svc.terminalKV.Put(t.Context(), kvstore.AccountKey(testAccountID, "sir-empty"), []byte(`{"account_id":"x"}`))
require.NoError(t, err)
_, err = svc.ScanAccount(t.Context(), testAccountID)
@@ -76,7 +76,7 @@ func (k failingGetKV) Get(ctx context.Context, key string) (jetstream.KeyValueEn
func TestScanAccountFailsWhenARecordCannotBeRead(t *testing.T) {
svc := setupTestService(t)
putSIRs(t, svc, makeSIR("sir-aaa", "i-aaa"), makeSIR("sir-bbb", "i-bbb"))
svc.activeKV = failingGetKV{KeyValue: svc.activeKV, key: kvutil.AccountKey(testAccountID, "sir-bbb")}
svc.activeKV = failingGetKV{KeyValue: svc.activeKV, key: kvstore.AccountKey(testAccountID, "sir-bbb")}
scan, err := svc.ScanAccount(t.Context(), testAccountID)
require.Error(t, err)
@@ -8,7 +8,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -24,14 +24,14 @@ type LaunchReferences struct {
// account. It is complete or an error: an unlistable bucket or an unreadable
// request is an error, because the unread request may name the resource.
func (s *SpotInstanceServiceImpl) AccountLaunchReferences(ctx context.Context, accountID string) (*LaunchReferences, error) {
keys, err := kvutil.Keys(ctx, s.activeKV)
keys, err := kvstore.Keys(ctx, s.activeKV)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list spot requests: %w", err)
}
prefix := accountID + "."
out := &LaunchReferences{}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
record, err := s.readRecord(ctx, s.activeKV, key)
@@ -5,7 +5,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -41,7 +41,7 @@ func TestAccountLaunchReferences_ReadsOpenAndActiveRequests(t *testing.T) {
func TestAccountLaunchReferences_UnreadableRequestFailsTheCheck(t *testing.T) {
svc := setupTestService(t)
putSIRs(t, svc, makeSIR("sir-active01", "i-0000000000000001"))
_, err := svc.activeKV.Put(t.Context(), kvutil.AccountKey(testAccountID, "sir-broken01"), []byte("{not json"))
_, err := svc.activeKV.Put(t.Context(), kvstore.AccountKey(testAccountID, "sir-broken01"), []byte("{not json"))
require.NoError(t, err)
_, err = svc.AccountLaunchReferences(t.Context(), testAccountID)
@@ -15,7 +15,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -41,7 +41,7 @@ func NewSpotInstanceServiceImplWithNATS(ctx context.Context, cfg *config.Config,
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
activeKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketSpotRequests, 10)
activeKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketSpotRequests, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSpotRequests, err)
}
@@ -71,7 +71,7 @@ func NewSpotInstanceServiceImplWithNATS(ctx context.Context, cfg *config.Config,
// getOrCreateTerminalBucket opens the terminal bucket without changing its
// existing replica or placement configuration, creating it only when absent.
func getOrCreateTerminalBucket(ctx context.Context, js jetstream.KeyValueManager) (jetstream.KeyValue, error) {
return kvutil.GetOrCreateBucketWithOptions(ctx, js, kvutil.BucketOptions{
return kvstore.GetOrCreateBucketWithOptions(ctx, js, kvstore.BucketOptions{
Name: KVBucketSpotRequestsTerminal,
Description: "Terminal Spot Instance Requests (auto-expire after 1 hour)",
History: 1,
@@ -91,7 +91,7 @@ func (s *SpotInstanceServiceImpl) PutSpotInstanceRequests(ctx context.Context, i
if err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.activeKV.Put(ctx, kvutil.AccountKey(accountID, sirID), data); err != nil {
if _, err := s.activeKV.Put(ctx, kvstore.AccountKey(accountID, sirID), data); err != nil {
slog.ErrorContext(ctx, "PutSpotInstanceRequests: KV put failed", "sirId", sirID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -171,7 +171,7 @@ func (s *SpotInstanceServiceImpl) CancelSpotInstanceRequests(ctx context.Context
cancelled := make([]*ec2.CancelledSpotInstanceRequest, 0, len(input.SpotInstanceRequestIds))
for _, idPtr := range input.SpotInstanceRequestIds {
sirID := aws.StringValue(idPtr)
key := kvutil.AccountKey(accountID, sirID)
key := kvstore.AccountKey(accountID, sirID)
record, err := s.readRecord(ctx, s.activeKV, key)
if err == nil {
@@ -213,7 +213,7 @@ func (s *SpotInstanceServiceImpl) CloseForInstance(ctx context.Context, instance
prefix := accountID + "."
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
record, err := s.readRecord(ctx, s.activeKV, key)
@@ -251,7 +251,7 @@ func (s *SpotInstanceServiceImpl) listRequests(ctx context.Context, kv jetstream
prefix := accountID + "."
var requests []*ec2.SpotInstanceRequest
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
record, err := s.readRecord(ctx, kv, key)
@@ -6,7 +6,7 @@ import (
"sync"
"testing"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -39,7 +39,7 @@ func (k *interleavingKV) Get(ctx context.Context, key string) (jetstream.KeyValu
// its write both survive alongside its own tag.
func TestCurrentBehaviour_Tagging_RecordMirrorKeepsInterleavedWrites(t *testing.T) {
const acct, id = "123456789012", "sg-0123456789abcdef0"
key := kvutil.AccountKey(acct, id)
key := kvstore.AccountKey(acct, id)
seed, err := json.Marshal(tagMirrorRecord{Name: "original", Tags: map[string]string{}})
require.NoError(t, err)
kv := seedKV(t, "tagmirror-cb", map[string][]byte{key: seed})
+3 -3
View File
@@ -10,7 +10,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -83,8 +83,8 @@ var errRecordAbsent = errors.New("tagmirror: record absent")
// applying mut and committing at the read revision, retrying on conflict so a
// concurrent write is never reverted. An absent resource is skipped.
func UpdateKVRecordTags[R any](ctx context.Context, kv jetstream.KeyValue, accountID, resourceID string, mut func(*R)) error {
key := kvutil.AccountKey(accountID, resourceID)
_, err := kvutil.Update(ctx, kv, key, kvutil.CASConfig{NotFound: errRecordAbsent}, func(rec *R) (bool, error) {
key := kvstore.AccountKey(accountID, resourceID)
_, err := kvstore.Update(ctx, kv, key, kvstore.CASConfig{NotFound: errRecordAbsent}, func(rec *R) (bool, error) {
mut(rec)
return true, nil
})
@@ -8,7 +8,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -75,15 +75,15 @@ func TestUpdateKVRecordTags(t *testing.T) {
rec, err := json.Marshal(&tagMirrorRecord{Name: "r1", Tags: map[string]string{"a": "1"}})
require.NoError(t, err)
kv := seedKV(t, "tagmirror-test", map[string][]byte{
kvutil.AccountKey("acct", "res-1"): rec,
kvutil.AccountKey("acct", "res-bad"): []byte("not-json"),
kvstore.AccountKey("acct", "res-1"): rec,
kvstore.AccountKey("acct", "res-bad"): []byte("not-json"),
})
t.Run("mutates and persists", func(t *testing.T) {
require.NoError(t, UpdateKVRecordTags(t.Context(), kv, "acct", "res-1", func(r *tagMirrorRecord) {
r.Tags["b"] = "2"
}))
entry, err := kv.Get(t.Context(), kvutil.AccountKey("acct", "res-1"))
entry, err := kv.Get(t.Context(), kvstore.AccountKey("acct", "res-1"))
require.NoError(t, err)
var got tagMirrorRecord
require.NoError(t, json.Unmarshal(entry.Value(), &got))
@@ -110,14 +110,14 @@ func TestMirrorKVRecordTags(t *testing.T) {
nilTags, err := json.Marshal(&tagMirrorRecord{})
require.NoError(t, err)
kv := seedKV(t, "tagmirror-mirror-test", map[string][]byte{
kvutil.AccountKey("acct", "vol-1"): withTags,
kvutil.AccountKey("acct", "vol-2"): nilTags,
kvstore.AccountKey("acct", "vol-1"): withTags,
kvstore.AccountKey("acct", "vol-2"): nilTags,
})
tagsOf := func(r *tagMirrorRecord) *map[string]string { return &r.Tags }
readTags := func(t *testing.T, id string) map[string]string {
entry, err := kv.Get(t.Context(), kvutil.AccountKey("acct", id))
entry, err := kv.Get(t.Context(), kvstore.AccountKey("acct", id))
require.NoError(t, err)
var got tagMirrorRecord
require.NoError(t, json.Unmarshal(entry.Value(), &got))
@@ -150,7 +150,7 @@ func TestMirrorKVRecordTags(t *testing.T) {
// must never be touched even though it exists in this bucket.
otherRec, err := json.Marshal(&tagMirrorRecord{Tags: map[string]string{"x": "y"}})
require.NoError(t, err)
_, err = kv.Put(t.Context(), kvutil.AccountKey("acct", "snap-other"), otherRec)
_, err = kv.Put(t.Context(), kvstore.AccountKey("acct", "snap-other"), otherRec)
require.NoError(t, err)
require.NoError(t, MirrorKVRecordTags(t.Context(), kv, "acct", "vol-", resources, tagsOf,
MergeTagsMut(&ec2.CreateTagsInput{Tags: []*ec2.Tag{
+2 -2
View File
@@ -23,7 +23,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -274,7 +274,7 @@ func (s *TagsServiceImpl) CreateTags(ctx context.Context, input *ec2.CreateTagsI
func (s *TagsServiceImpl) taggedResourceIDs(ctx context.Context, accountID string) ([]string, error) {
seen := make(map[string]struct{})
keys, err := kvutil.Keys(ctx, s.kv)
keys, err := kvstore.Keys(ctx, s.kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list tag keys: %w", err)
}
+4 -4
View File
@@ -6,7 +6,7 @@ import (
"fmt"
"strings"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -46,7 +46,7 @@ func resourceIDFromKVKey(key, accountID string) (string, bool) {
// GetOrCreateTagsBucket opens the tag bucket, creating it at the cluster's
// replica count if this is the first node to reach it.
func GetOrCreateTagsBucket(ctx context.Context, js jetstream.KeyValueManager) (jetstream.KeyValue, error) {
return kvutil.GetOrCreateBucket(ctx, js, KVBucketTags, kvTagsHistory)
return kvstore.GetOrCreateBucket(ctx, js, KVBucketTags, kvTagsHistory)
}
// readKVTags returns a resource's tags from the bucket, and whether the key
@@ -76,7 +76,7 @@ func (s *TagsServiceImpl) readKVTags(ctx context.Context, accountID, resourceID
// before the store moved.
func (s *TagsServiceImpl) mutateTags(ctx context.Context, accountID, resourceID string, mutate func(map[string]string)) error {
key := tagsKVKey(accountID, resourceID)
_, err := kvutil.Update(ctx, s.kv, key, kvutil.CASConfig{
_, err := kvstore.Update(ctx, s.kv, key, kvstore.CASConfig{
Attempts: tagCASAttempts,
CreateIfAbsent: true,
Exhausted: func(key string, attempts int) error {
@@ -104,7 +104,7 @@ func (s *TagsServiceImpl) PutResourceTags(ctx context.Context, accountID, resour
if err != nil {
return err
}
return kvutil.Put(ctx, s.kv, tagsKVKey(accountID, resourceID), kvutil.CASConfig{
return kvstore.Put(ctx, s.kv, tagsKVKey(accountID, resourceID), kvstore.CASConfig{
Attempts: tagCASAttempts,
CreateIfAbsent: true,
}, data)
+6 -6
View File
@@ -12,7 +12,7 @@ import (
networkv1 "github.com/mulgadc/spinifex/contracts/network/v1"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -105,7 +105,7 @@ func (s *VPCServiceImpl) ensureDefaultVPC(ctx context.Context, accountID string,
return claim.info(), nil
}
if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, claim.VpcId)); err == nil {
if _, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, claim.VpcId)); err == nil {
return claim.info(), nil
} else if !errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, fmt.Errorf("read default VPC %s: %w", claim.VpcId, err)
@@ -233,7 +233,7 @@ func oldestAccountRecord[T any](ctx context.Context, kv jetstream.KeyValue, acco
var bestID string
var bestCreated time.Time
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := kv.Get(ctx, key)
@@ -262,7 +262,7 @@ func oldestAccountRecord[T any](ctx context.Context, kv jetstream.KeyValue, acco
// marks the claim complete. Records are written with Create, so only the caller
// that actually wrote one announces it to vpcd.
func (s *VPCServiceImpl) buildDefaultVPC(ctx context.Context, accountID string, claim *defaultVPCClaim) error {
created, err := createRecord(ctx, s.vpcKV, kvutil.AccountKey(accountID, claim.VpcId), VPCRecord{
created, err := createRecord(ctx, s.vpcKV, kvstore.AccountKey(accountID, claim.VpcId), VPCRecord{
VpcId: claim.VpcId,
CidrBlock: claim.Cidr,
State: "available",
@@ -284,7 +284,7 @@ func (s *VPCServiceImpl) buildDefaultVPC(ctx context.Context, accountID string,
if s.config != nil && s.config.AZ != "" {
az = s.config.AZ
}
created, err = createRecord(ctx, s.subnetKV, kvutil.AccountKey(accountID, claim.SubnetId), SubnetRecord{
created, err = createRecord(ctx, s.subnetKV, kvstore.AccountKey(accountID, claim.SubnetId), SubnetRecord{
SubnetId: claim.SubnetId,
VpcId: claim.VpcId,
CidrBlock: claim.SubnetCidr,
@@ -320,7 +320,7 @@ func (s *VPCServiceImpl) buildDefaultVPC(ctx context.Context, accountID string,
}
}
if _, err := kvutil.Update(ctx, s.defaultsKV, accountID, kvutil.CASConfig{}, func(c *defaultVPCClaim) (bool, error) {
if _, err := kvstore.Update(ctx, s.defaultsKV, accountID, kvstore.CASConfig{}, func(c *defaultVPCClaim) (bool, error) {
if c.Complete || c.VpcId != claim.VpcId {
return false, nil
}
+4 -4
View File
@@ -13,7 +13,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/internal/testkit"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
@@ -149,12 +149,12 @@ func TestEnsureDefaultVPC_AdoptsOldestExistingDefault(t *testing.T) {
const accountID = "000000000051"
created := time.Date(2026, 9, 11, 14, 13, 40, 0, time.UTC)
for i, id := range []string{"vpc-0000000000000newer", "vpc-0000000000000older"} {
putRecord(t, js, ec2vpc.KVBucketVPCs, kvutil.AccountKey(accountID, id), ec2vpc.VPCRecord{
putRecord(t, js, ec2vpc.KVBucketVPCs, kvstore.AccountKey(accountID, id), ec2vpc.VPCRecord{
VpcId: id, CidrBlock: ec2vpc.DefaultVPCCidr, IsDefault: true, VNI: int64(200 + i),
CreatedAt: created.Add(time.Duration(1-i) * time.Minute),
})
}
putRecord(t, js, ec2vpc.KVBucketSubnets, kvutil.AccountKey(accountID, "subnet-00000000000older"), ec2vpc.SubnetRecord{
putRecord(t, js, ec2vpc.KVBucketSubnets, kvstore.AccountKey(accountID, "subnet-00000000000older"), ec2vpc.SubnetRecord{
SubnetId: "subnet-00000000000older", VpcId: "vpc-0000000000000older", CidrBlock: ec2vpc.DefaultSubnetCidr,
IsDefault: true, CreatedAt: created,
})
@@ -183,7 +183,7 @@ func TestEnsureDefaultVPC_FinishesHalfBuiltClaim(t *testing.T) {
"route_table_id": "rtb-00000000000half", "internet_gateway_id": "igw-00000000000half",
"cidr": ec2vpc.DefaultVPCCidr, "subnet_cidr": ec2vpc.DefaultSubnetCidr, "vni": 4242,
})
putRecord(t, js, ec2vpc.KVBucketVPCs, kvutil.AccountKey(accountID, "vpc-00000000000000half"), ec2vpc.VPCRecord{
putRecord(t, js, ec2vpc.KVBucketVPCs, kvstore.AccountKey(accountID, "vpc-00000000000000half"), ec2vpc.VPCRecord{
VpcId: "vpc-00000000000000half", CidrBlock: ec2vpc.DefaultVPCCidr, IsDefault: true, VNI: 4242,
})
+18 -18
View File
@@ -23,7 +23,7 @@ import (
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -118,7 +118,7 @@ func (s *VPCServiceImpl) CreateNetworkInterface(ctx context.Context, input *ec2.
subnetId := *input.SubnetId
// Verify subnet exists and belongs to this account
subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetId))
subnetEntry, err := s.subnetKV.Get(ctx, kvstore.AccountKey(accountID, subnetId))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
}
@@ -205,7 +205,7 @@ func (s *VPCServiceImpl) CreateNetworkInterface(ctx context.Context, input *ec2.
if err != nil {
return nil, fmt.Errorf("failed to marshal ENI record: %w", err)
}
if _, err := s.eniKV.Put(ctx, kvutil.AccountKey(accountID, eniId), data); err != nil {
if _, err := s.eniKV.Put(ctx, kvstore.AccountKey(accountID, eniId), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -251,7 +251,7 @@ func (s *VPCServiceImpl) ForceDeleteInstanceENI(ctx context.Context, accountID,
}
func (s *VPCServiceImpl) deleteNetworkInterface(ctx context.Context, eniId, accountID string, force bool) (*ec2.DeleteNetworkInterfaceOutput, error) {
key := kvutil.AccountKey(accountID, eniId)
key := kvstore.AccountKey(accountID, eniId)
// Get the ENI record
entry, err := s.eniKV.Get(ctx, key)
@@ -346,7 +346,7 @@ func (s *VPCServiceImpl) releaseENISideEffects(ctx context.Context, eniId, accou
// deletion that did not happen.
func (s *VPCServiceImpl) DetachAndDeleteENI(ctx context.Context, accountID, eniID string, force bool) (deleted bool, err error) {
const maxAttempts = 3
key := kvutil.AccountKey(accountID, eniID)
key := kvstore.AccountKey(accountID, eniID)
var lastErr error
for range maxAttempts {
@@ -458,7 +458,7 @@ func (s *VPCServiceImpl) ModifyNetworkInterfaceAttribute(ctx context.Context, in
}
eniId := *input.NetworkInterfaceId
key := kvutil.AccountKey(accountID, eniId)
key := kvstore.AccountKey(accountID, eniId)
entry, err := s.eniKV.Get(ctx, key)
if err != nil {
@@ -569,7 +569,7 @@ func (s *VPCServiceImpl) DescribeNetworkInterfaces(ctx context.Context, input *e
}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, prefix) {
@@ -707,7 +707,7 @@ func (s *VPCServiceImpl) AttachENIWithOwner(accountID, eniId, instanceId string,
func (s *VPCServiceImpl) attachENI(ctx context.Context, accountID, eniId, instanceId string, deviceIndex int64, instanceOwnerID string) (string, error) {
attachmentId := awsidentifiers.GenerateResourceID("eni-attach")
err := s.casENI(ctx, kvutil.AccountKey(accountID, eniId), eniId, func(record *ENIRecord) error {
err := s.casENI(ctx, kvstore.AccountKey(accountID, eniId), eniId, func(record *ENIRecord) error {
if record.Status == "in-use" {
return errors.New(awserrors.ErrorInvalidNetworkInterfaceInUse)
}
@@ -737,7 +737,7 @@ func (s *VPCServiceImpl) attachENI(ctx context.Context, accountID, eniId, instan
// DetachENI marks an ENI as detached from an instance (internal use by TerminateInstances).
// accountID scopes the lookup to the correct KV key.
func (s *VPCServiceImpl) DetachENI(ctx context.Context, accountID, eniId string) error {
key := kvutil.AccountKey(accountID, eniId)
key := kvstore.AccountKey(accountID, eniId)
entry, err := s.eniKV.Get(ctx, key)
if err != nil {
return eniNotFoundError(eniId)
@@ -773,7 +773,7 @@ func (s *VPCServiceImpl) GetENIRecord(accountID, eniId string) (ENIRecord, error
}
func (s *VPCServiceImpl) getENIRecord(ctx context.Context, accountID, eniId string) (ENIRecord, error) {
key := kvutil.AccountKey(accountID, eniId)
key := kvstore.AccountKey(accountID, eniId)
entry, err := s.eniKV.Get(ctx, key)
if err != nil {
return ENIRecord{}, eniNotFoundError(eniId)
@@ -793,7 +793,7 @@ func (s *VPCServiceImpl) UpdateENI(accountID, eniId string, fn func(*ENIRecord))
}
func (s *VPCServiceImpl) updateENI(ctx context.Context, accountID, eniId string, fn func(*ENIRecord)) error {
return s.casENI(ctx, kvutil.AccountKey(accountID, eniId), eniId, func(r *ENIRecord) error {
return s.casENI(ctx, kvstore.AccountKey(accountID, eniId), eniId, func(r *ENIRecord) error {
fn(r)
return nil
})
@@ -876,7 +876,7 @@ func (s *VPCServiceImpl) listInstanceENIs(ctx context.Context, accountID, instan
prefix := accountID + "."
var out []ENIRecord
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.eniKV.Get(ctx, key)
@@ -938,7 +938,7 @@ func (s *VPCServiceImpl) scanENIs(ctx context.Context, match func(*ENIRecord) bo
var out []AccountENI
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
accountID, _, found := strings.Cut(key, ".")
@@ -1004,7 +1004,7 @@ func (s *VPCServiceImpl) findENIByAttachment(ctx context.Context, accountID, att
}
prefix := accountID + "."
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.eniKV.Get(ctx, key)
@@ -1028,7 +1028,7 @@ func (s *VPCServiceImpl) UpdateENIPublicIP(accountID, eniId, publicIP, poolName
}
func (s *VPCServiceImpl) updateENIPublicIP(ctx context.Context, accountID, eniId, publicIP, poolName string) error {
key := kvutil.AccountKey(accountID, eniId)
key := kvstore.AccountKey(accountID, eniId)
entry, err := s.eniKV.Get(ctx, key)
if err != nil {
return fmt.Errorf("ENI %s not found: %w", eniId, err)
@@ -1063,7 +1063,7 @@ func (s *VPCServiceImpl) fillSGNames(ctx context.Context, accountID string, ids
continue
}
names[id] = ""
entry, err := s.sgKV.Get(ctx, kvutil.AccountKey(accountID, id))
entry, err := s.sgKV.Get(ctx, kvstore.AccountKey(accountID, id))
if err != nil {
slog.WarnContext(ctx, "fillSGNames: SG read failed", "groupId", id, "err", err)
continue
@@ -1080,7 +1080,7 @@ func (s *VPCServiceImpl) fillSGNames(ctx context.Context, accountID string, ids
// vpcDNSHostnames reports whether the VPC has enableDnsHostnames on. An
// unreadable VPC reads as off, so its interfaces go unnamed.
func (s *VPCServiceImpl) vpcDNSHostnames(ctx context.Context, accountID, vpcID string) bool {
entry, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID))
entry, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcID))
if err != nil {
slog.WarnContext(ctx, "vpcDNSHostnames: VPC read failed", "vpcId", vpcID, "err", err)
return false
@@ -1238,7 +1238,7 @@ func (s *VPCServiceImpl) validateSGAttachment(ctx context.Context, accountID str
// Each SG must exist in the caller's account and belong to the same VPC.
for _, sgId := range sgIds {
sgEntry, err := s.sgKV.Get(ctx, kvutil.AccountKey(accountID, sgId))
sgEntry, err := s.sgKV.Get(ctx, kvstore.AccountKey(accountID, sgId))
if err != nil {
return sgNotFoundError(sgId)
}
+2 -2
View File
@@ -10,7 +10,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -69,7 +69,7 @@ func (s *VPCServiceImpl) findENIAnyAccount(ctx context.Context, eniID string) (*
suffix := "." + eniID
for _, k := range keys {
if k == kvutil.VersionKey || !strings.HasSuffix(k, suffix) {
if k == kvstore.VersionKey || !strings.HasSuffix(k, suffix) {
continue
}
entry, err := s.eniKV.Get(ctx, k)
@@ -7,7 +7,7 @@ import (
"testing"
testutil "github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -30,7 +30,7 @@ func newStaleReadENI(t *testing.T) (*VPCServiceImpl, string, func() ENIRecord) {
// Reads the stored record directly: the wrapped bucket would serve the
// check itself from a lagging replica.
stored := func() ENIRecord {
entry, err := inner.Get(context.Background(), kvutil.AccountKey(testAccountID, eniID))
entry, err := inner.Get(context.Background(), kvstore.AccountKey(testAccountID, eniID))
require.NoError(t, err)
var rec ENIRecord
require.NoError(t, json.Unmarshal(entry.Value(), &rec))
@@ -114,7 +114,7 @@ func TestCASENIGivesUpAfterBoundedConflicts(t *testing.T) {
t.Parallel()
svc, eniID, _ := newStaleReadENI(t)
conflicts := 0
err := svc.casENI(t.Context(), kvutil.AccountKey(testAccountID, eniID), eniID, func(*ENIRecord) error {
err := svc.casENI(t.Context(), kvstore.AccountKey(testAccountID, eniID), eniID, func(*ENIRecord) error {
conflicts++
// Move the record under the next compare-and-swap, as a competing writer does.
return svc.UpdateENI(testAccountID, eniID, func(r *ENIRecord) { r.Description = "other writer" })
+2 -2
View File
@@ -6,7 +6,7 @@ import (
"fmt"
"strings"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -42,7 +42,7 @@ func keyWithSuffixExists(ctx context.Context, kv jetstream.KeyValue, resourceID
}
suffix := "." + resourceID
for _, k := range keys {
if k != kvutil.VersionKey && strings.HasSuffix(k, suffix) {
if k != kvstore.VersionKey && strings.HasSuffix(k, suffix) {
return true, nil
}
}
+20 -20
View File
@@ -23,7 +23,7 @@ import (
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -245,7 +245,7 @@ func (s *VPCServiceImpl) CreateSecurityGroup(ctx context.Context, input *ec2.Cre
}
sgsInVPC := 0
for _, k := range sgKeys {
if k == kvutil.VersionKey {
if k == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(k, prefix) {
@@ -305,7 +305,7 @@ func (s *VPCServiceImpl) CreateSecurityGroup(ctx context.Context, input *ec2.Cre
if err != nil {
return nil, fmt.Errorf("failed to marshal security group record: %w", err)
}
if _, err := s.sgKV.Put(ctx, kvutil.AccountKey(accountID, groupId), data); err != nil {
if _, err := s.sgKV.Put(ctx, kvstore.AccountKey(accountID, groupId), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -331,7 +331,7 @@ func (s *VPCServiceImpl) DeleteSecurityGroup(ctx context.Context, input *ec2.Del
}
groupId := *input.GroupId
key := kvutil.AccountKey(accountID, groupId)
key := kvstore.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -380,7 +380,7 @@ func (s *VPCServiceImpl) validateSGRuleReferences(ctx context.Context, accountID
if r.SourceSG == "" {
continue
}
entry, err := s.sgKV.Get(ctx, kvutil.AccountKey(accountID, r.SourceSG))
entry, err := s.sgKV.Get(ctx, kvstore.AccountKey(accountID, r.SourceSG))
if err != nil {
return sgNotFoundError(r.SourceSG)
}
@@ -406,7 +406,7 @@ func (s *VPCServiceImpl) checkSGDependencies(ctx context.Context, accountID, gro
return errors.New(awserrors.ErrorServerInternal)
}
for _, k := range eniKeys {
if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
continue
}
entry, err := s.eniKV.Get(ctx, k)
@@ -435,7 +435,7 @@ func (s *VPCServiceImpl) checkSGDependencies(ctx context.Context, accountID, gro
return errors.New(awserrors.ErrorServerInternal)
}
for _, k := range sgKeys {
if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
continue
}
entry, err := s.sgKV.Get(ctx, k)
@@ -561,7 +561,7 @@ func (s *VPCServiceImpl) DescribeSecurityGroups(ctx context.Context, input *ec2.
}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, prefix) {
@@ -713,7 +713,7 @@ func (s *VPCServiceImpl) GetSecurityGroupsForVpc(ctx context.Context, input *ec2
groups := []*ec2.SecurityGroupForVpc{}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.sgKV.Get(ctx, key)
@@ -844,7 +844,7 @@ func (s *VPCServiceImpl) DescribeSecurityGroupRules(ctx context.Context, input *
emitted := make(map[string]bool)
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.sgKV.Get(ctx, key)
@@ -945,7 +945,7 @@ func (s *VPCServiceImpl) updateSGRuleTags(ctx context.Context, accountID, ruleID
}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.sgKV.Get(ctx, key)
@@ -1056,7 +1056,7 @@ func (s *VPCServiceImpl) AuthorizeSecurityGroupIngress(ctx context.Context, inpu
}
groupId := *input.GroupId
key := kvutil.AccountKey(accountID, groupId)
key := kvstore.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -1134,7 +1134,7 @@ func (s *VPCServiceImpl) AuthorizeSecurityGroupEgress(ctx context.Context, input
}
groupId := *input.GroupId
key := kvutil.AccountKey(accountID, groupId)
key := kvstore.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -1212,7 +1212,7 @@ func (s *VPCServiceImpl) RevokeSecurityGroupIngress(ctx context.Context, input *
}
groupId := *input.GroupId
key := kvutil.AccountKey(accountID, groupId)
key := kvstore.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -1277,7 +1277,7 @@ func (s *VPCServiceImpl) RevokeSecurityGroupEgress(ctx context.Context, input *e
}
groupId := *input.GroupId
key := kvutil.AccountKey(accountID, groupId)
key := kvstore.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -1397,7 +1397,7 @@ func (s *VPCServiceImpl) updateSGRuleDescriptions(ctx context.Context, accountID
return errors.New(awserrors.ErrorInvalidParameterCombination)
}
key := kvutil.AccountKey(accountID, req.groupId)
key := kvstore.AccountKey(accountID, req.groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
return sgNotFoundError(req.groupId)
@@ -1581,7 +1581,7 @@ func (s *VPCServiceImpl) ModifySecurityGroupRules(ctx context.Context, input *ec
}
groupId := *input.GroupId
key := kvutil.AccountKey(accountID, groupId)
key := kvstore.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -2158,7 +2158,7 @@ func (s *VPCServiceImpl) storeDefaultSecurityGroup(ctx context.Context, accountI
if err != nil {
return nil, false, fmt.Errorf("marshal default security group: %w", err)
}
if _, err := s.sgKV.Create(ctx, kvutil.AccountKey(accountID, groupId), data); err != nil {
if _, err := s.sgKV.Create(ctx, kvstore.AccountKey(accountID, groupId), data); err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
return record, false, nil
}
@@ -2182,7 +2182,7 @@ func (s *VPCServiceImpl) announceDefaultSecurityGroup(record *SecurityGroupRecor
// CannotDelete guard for default SGs. Used by DeleteVpc to cascade-delete the
// per-VPC default SG. Surfaces vpcd tear-down failures to the caller.
func (s *VPCServiceImpl) deleteSecurityGroupInternal(ctx context.Context, accountID, groupId string) error {
key := kvutil.AccountKey(accountID, groupId)
key := kvstore.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
return fmt.Errorf("read default security group: %w", err)
@@ -2219,7 +2219,7 @@ func (s *VPCServiceImpl) findDefaultSGForVPC(ctx context.Context, accountID, vpc
return "", err
}
for _, k := range keys {
if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
continue
}
entry, err := s.sgKV.Get(ctx, k)
@@ -15,7 +15,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -62,7 +62,7 @@ func tcpPeerRequest(port int64, peer string) *ec2.SecurityGroupRuleRequest {
// modify wrote nothing.
func storedSGRecord(t *testing.T, svc *VPCServiceImpl, sgID string) []byte {
t.Helper()
entry, err := svc.sgKV.Get(t.Context(), kvutil.AccountKey(testAccountID, sgID))
entry, err := svc.sgKV.Get(t.Context(), kvstore.AccountKey(testAccountID, sgID))
require.NoError(t, err)
return entry.Value()
}
@@ -689,7 +689,7 @@ func TestRevokeSecurityGroupIngress_InvertedPortRangeStillRevocable(t *testing.T
vpcID := createTestVPC(t, svc, "10.0.0.0/16")
sgID := createTestSG(t, svc, vpcID, "stale-inverted-ports")
key := kvutil.AccountKey(testAccountID, sgID)
key := kvstore.AccountKey(testAccountID, sgID)
var rec SecurityGroupRecord
require.NoError(t, json.Unmarshal(storedSGRecord(t, svc, sgID), &rec))
rec.IngressRules = append(rec.IngressRules, SGRule{
@@ -13,7 +13,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -35,7 +35,7 @@ func plantIngressRules(t *testing.T, svc *VPCServiceImpl, sgID string, rules ...
rec.IngressRules = append(rec.IngressRules, rules...)
data, err := json.Marshal(rec)
require.NoError(t, err)
_, err = svc.sgKV.Put(t.Context(), kvutil.AccountKey(testAccountID, sgID), data)
_, err = svc.sgKV.Put(t.Context(), kvstore.AccountKey(testAccountID, sgID), data)
require.NoError(t, err)
}
@@ -13,7 +13,7 @@ import (
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -1524,7 +1524,7 @@ func findDefaultSGInVPC(t *testing.T, svc *VPCServiceImpl, vpcID string) string
require.NoError(t, err)
var matches []string
for _, k := range keys {
if k == kvutil.VersionKey {
if k == kvstore.VersionKey {
continue
}
entry, err := svc.sgKV.Get(t.Context(), k)
@@ -2122,7 +2122,7 @@ func seedAvailableVPC(t *testing.T, svc *VPCServiceImpl, vpcID string) {
rec := VPCRecord{VpcId: vpcID, CidrBlock: "10.0.0.0/16", State: "available"}
data, err := json.Marshal(rec)
require.NoError(t, err)
_, err = svc.vpcKV.Put(t.Context(), kvutil.AccountKey(testAccountID, vpcID), data)
_, err = svc.vpcKV.Put(t.Context(), kvstore.AccountKey(testAccountID, vpcID), data)
require.NoError(t, err)
}
@@ -2359,7 +2359,7 @@ func TestAuthorizeSecurityGroupIngress_AssignsRuleIDs(t *testing.T) {
}, testAccountID)
require.NoError(t, err)
entry, err := svc.sgKV.Get(t.Context(), kvutil.AccountKey(testAccountID, sgID))
entry, err := svc.sgKV.Get(t.Context(), kvstore.AccountKey(testAccountID, sgID))
require.NoError(t, err)
var record SecurityGroupRecord
require.NoError(t, json.Unmarshal(entry.Value(), &record))
+35 -35
View File
@@ -26,7 +26,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
awstags "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/zoneid"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -173,7 +173,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
return nil, fmt.Errorf("create JetStream client: %w", err)
}
vpcKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketVPCs, 10)
vpcKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketVPCs, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVPCs, err)
}
@@ -181,7 +181,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
return nil, fmt.Errorf("migrate %s: %w", KVBucketVPCs, err)
}
subnetKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketSubnets, 10)
subnetKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketSubnets, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSubnets, err)
}
@@ -189,7 +189,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
return nil, fmt.Errorf("migrate %s: %w", KVBucketSubnets, err)
}
vniKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketVNICounter, 10)
vniKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketVNICounter, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVNICounter, err)
}
@@ -197,7 +197,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
return nil, fmt.Errorf("migrate %s: %w", KVBucketVNICounter, err)
}
eniKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketENIs, 10)
eniKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketENIs, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketENIs, err)
}
@@ -205,7 +205,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
return nil, fmt.Errorf("migrate %s: %w", KVBucketENIs, err)
}
sgKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketSecurityGroups, 10)
sgKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketSecurityGroups, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSecurityGroups, err)
}
@@ -213,7 +213,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
return nil, fmt.Errorf("migrate %s: %w", KVBucketSecurityGroups, err)
}
defaultsKV, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketDefaultVPCs, 10)
defaultsKV, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketDefaultVPCs, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketDefaultVPCs, err)
}
@@ -221,18 +221,18 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
return nil, fmt.Errorf("migrate %s: %w", KVBucketDefaultVPCs, err)
}
rtbKV, err := kvutil.GetOrCreateBucket(ctx, js, "spinifex-vpc-route-tables", 10)
rtbKV, err := kvstore.GetOrCreateBucket(ctx, js, "spinifex-vpc-route-tables", 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket spinifex-vpc-route-tables: %w", err)
}
// Named literally, like the route table bucket above: the IGW package
// imports this one, so its constant cannot be referenced here.
igwKV, err := kvutil.GetOrCreateBucket(ctx, js, "spinifex-igw", 10)
igwKV, err := kvstore.GetOrCreateBucket(ctx, js, "spinifex-igw", 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket spinifex-igw: %w", err)
}
eigwKV, err := kvutil.GetOrCreateBucket(ctx, js, "spinifex-eigw", 10)
eigwKV, err := kvstore.GetOrCreateBucket(ctx, js, "spinifex-eigw", 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket spinifex-eigw: %w", err)
}
@@ -268,7 +268,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn
// nextVNI allocates the next VNI using atomic increment on the NATS KV counter.
func (s *VPCServiceImpl) nextVNI(ctx context.Context) (int64, error) {
var allocated int64
_, err := kvutil.Update(ctx, s.vniKV, vniCounterKey, kvutil.CASConfig{CreateIfAbsent: true, Attempts: vniCASAttempts},
_, err := kvstore.Update(ctx, s.vniKV, vniCounterKey, kvstore.CASConfig{CreateIfAbsent: true, Attempts: vniCASAttempts},
func(next *int64) (bool, error) {
if *next == 0 {
*next = vniStart
@@ -331,7 +331,7 @@ func (s *VPCServiceImpl) CreateVpc(ctx context.Context, input *ec2.CreateVpcInpu
if err != nil {
return nil, fmt.Errorf("failed to marshal VPC record: %w", err)
}
if _, err := s.vpcKV.Put(ctx, kvutil.AccountKey(accountID, vpcID), data); err != nil {
if _, err := s.vpcKV.Put(ctx, kvstore.AccountKey(accountID, vpcID), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -366,7 +366,7 @@ func (s *VPCServiceImpl) CreateVpc(ctx context.Context, input *ec2.CreateVpcInpu
// requireVPCExists returns InvalidVpcID.NotFound if the VPC doesn't exist for
// this account.
func (s *VPCServiceImpl) requireVPCExists(ctx context.Context, accountID, vpcId string) error {
if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcId)); err != nil {
if _, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcId)); err != nil {
return vpcNotFoundError(vpcId)
}
return nil
@@ -379,7 +379,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
}
vpcID := *input.VpcId
key := kvutil.AccountKey(accountID, vpcID)
key := kvstore.AccountKey(accountID, vpcID)
if _, err := s.vpcKV.Get(ctx, key); err != nil {
// AWS-faithful: an absent VPC is NotFound (the tofu/SDK provider
@@ -398,7 +398,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
return nil, errors.New(awserrors.ErrorServerInternal)
}
for _, k := range subnetKeys {
if k == kvutil.VersionKey {
if k == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(k, prefix) {
@@ -435,7 +435,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
return nil, errors.New(awserrors.ErrorServerInternal)
}
for _, k := range sgKeys {
if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
continue
}
entry, err := s.sgKV.Get(ctx, k)
@@ -472,7 +472,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
return nil, errors.New(awserrors.ErrorServerInternal)
}
for _, k := range rtbKeys {
if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
continue
}
entry, err := s.rtbKV.Get(ctx, k)
@@ -533,7 +533,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
return nil, errors.New(awserrors.ErrorServerInternal)
}
if rtbID != "" {
if err := s.rtbKV.Delete(ctx, kvutil.AccountKey(accountID, rtbID)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
if err := s.rtbKV.Delete(ctx, kvstore.AccountKey(accountID, rtbID)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
slog.ErrorContext(ctx, "DeleteVpc: main route table reap failed", "vpcId", vpcID, "routeTableId", rtbID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -574,7 +574,7 @@ func rejectAttachedGateway(ctx context.Context, kv jetstream.KeyValue, accountID
}
for _, k := range keys {
if k == kvutil.VersionKey || !strings.HasPrefix(k, prefix) {
if k == kvstore.VersionKey || !strings.HasPrefix(k, prefix) {
continue
}
entry, err := kv.Get(ctx, k)
@@ -653,7 +653,7 @@ func (s *VPCServiceImpl) DescribeVpcs(ctx context.Context, input *ec2.DescribeVp
}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, prefix) {
@@ -723,7 +723,7 @@ func (s *VPCServiceImpl) CreateSubnet(ctx context.Context, input *ec2.CreateSubn
}
// Verify VPC exists and belongs to this account
vpcEntry, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID))
vpcEntry, err := s.vpcKV.Get(ctx, kvstore.AccountKey(accountID, vpcID))
if err != nil {
return nil, vpcNotFoundError(vpcID)
}
@@ -757,7 +757,7 @@ func (s *VPCServiceImpl) CreateSubnet(ctx context.Context, input *ec2.CreateSubn
}
for _, k := range subnetKeys {
if k == kvutil.VersionKey {
if k == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(k, prefix) {
@@ -822,7 +822,7 @@ func (s *VPCServiceImpl) CreateSubnet(ctx context.Context, input *ec2.CreateSubn
if err != nil {
return nil, fmt.Errorf("failed to marshal subnet record: %w", err)
}
if _, err := s.subnetKV.Put(ctx, kvutil.AccountKey(accountID, subnetID), data); err != nil {
if _, err := s.subnetKV.Put(ctx, kvstore.AccountKey(accountID, subnetID), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -843,7 +843,7 @@ func (s *VPCServiceImpl) DeleteSubnet(ctx context.Context, input *ec2.DeleteSubn
}
subnetID := *input.SubnetId
key := kvutil.AccountKey(accountID, subnetID)
key := kvstore.AccountKey(accountID, subnetID)
// Read subnet record before deletion (needed for vpcd event)
subnetEntry, err := s.subnetKV.Get(ctx, key)
@@ -902,7 +902,7 @@ func (s *VPCServiceImpl) checkSubnetResidents(ctx context.Context, accountID, su
}
prefix := accountID + "."
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.eniKV.Get(ctx, key)
@@ -950,7 +950,7 @@ func (s *VPCServiceImpl) clearRouteTableAssociationsForSubnet(ctx context.Contex
}
prefix := accountID + "."
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.rtbKV.Get(ctx, key)
@@ -1035,7 +1035,7 @@ func (s *VPCServiceImpl) DescribeSubnets(ctx context.Context, input *ec2.Describ
}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, prefix) {
@@ -1386,7 +1386,7 @@ func (s *VPCServiceImpl) ModifySubnetAttribute(ctx context.Context, input *ec2.M
}
subnetID := *input.SubnetId
key := kvutil.AccountKey(accountID, subnetID)
key := kvstore.AccountKey(accountID, subnetID)
entry, err := s.subnetKV.Get(ctx, key)
if err != nil {
@@ -1438,7 +1438,7 @@ func (s *VPCServiceImpl) ModifyVpcAttribute(ctx context.Context, input *ec2.Modi
}
vpcID := *input.VpcId
key := kvutil.AccountKey(accountID, vpcID)
key := kvstore.AccountKey(accountID, vpcID)
entry, err := s.vpcKV.Get(ctx, key)
if err != nil {
@@ -1486,7 +1486,7 @@ func (s *VPCServiceImpl) DescribeVpcAttribute(ctx context.Context, input *ec2.De
}
vpcID := *input.VpcId
key := kvutil.AccountKey(accountID, vpcID)
key := kvstore.AccountKey(accountID, vpcID)
entry, err := s.vpcKV.Get(ctx, key)
if err != nil {
@@ -1587,7 +1587,7 @@ func (s *VPCServiceImpl) writeMainRouteTable(ctx context.Context, accountID, vpc
if err != nil {
return false, fmt.Errorf("marshal main route table: %w", err)
}
if _, err := s.rtbKV.Create(ctx, kvutil.AccountKey(accountID, rtbID), data); err != nil {
if _, err := s.rtbKV.Create(ctx, kvstore.AccountKey(accountID, rtbID), data); err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
return false, nil
}
@@ -1611,7 +1611,7 @@ func (s *VPCServiceImpl) findMainRouteTableID(ctx context.Context, accountID, vp
return "", fmt.Errorf("list rtb keys: %w", err)
}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.rtbKV.Get(ctx, key)
@@ -1652,7 +1652,7 @@ func (s *VPCServiceImpl) getDefaultSubnet(ctx context.Context, accountID string)
}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, prefix) {
@@ -1680,7 +1680,7 @@ func (s *VPCServiceImpl) GetSubnet(accountID, subnetId string) (*SubnetRecord, e
}
func (s *VPCServiceImpl) getSubnet(ctx context.Context, accountID, subnetId string) (*SubnetRecord, error) {
key := kvutil.AccountKey(accountID, subnetId)
key := kvstore.AccountKey(accountID, subnetId)
entry, err := s.subnetKV.Get(ctx, key)
if err != nil {
return nil, fmt.Errorf("subnet %s not found: %w", subnetId, err)
@@ -1726,7 +1726,7 @@ func (s *VPCServiceImpl) SubnetAvailabilityZones(ctx context.Context) ([]string,
}
seen := map[string]struct{}{}
for _, k := range keys {
if k == kvutil.VersionKey {
if k == kvstore.VersionKey {
continue
}
entry, err := s.subnetKV.Get(ctx, k)
@@ -16,7 +16,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/zoneid"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
@@ -2689,7 +2689,7 @@ func TestCreateSubnet_ZoneStoreFailureWritesNothingAndPublishesNothing(t *testin
require.ErrorIs(t, err, jetstream.ErrNoKeysFound)
}
for _, k := range keys {
assert.Equal(t, kvutil.VersionKey, k, "no subnet record may exist")
assert.Equal(t, kvstore.VersionKey, k, "no subnet record may exist")
}
assert.Zero(t, events.Load(), "no topology event may be published")
}
+2 -2
View File
@@ -9,7 +9,7 @@ import (
"github.com/golang-jwt/jwt/v5"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -111,7 +111,7 @@ func TestLoadOrCreateSigningKey_FailsWhenNoKeyDecrypts(t *testing.T) {
require.Error(t, err)
assert.Contains(t, err.Error(), "master key does not match")
entries, err := kvutil.Keys(t.Context(), kv)
entries, err := kvstore.Keys(t.Context(), kv)
require.NoError(t, err)
assert.Len(t, entries, 1, "a failed load must not mint a key alongside the unreadable one")
}
+3 -3
View File
@@ -21,7 +21,7 @@ import (
"time"
iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -111,7 +111,7 @@ func openSigningBucket(ctx context.Context, js jetstream.JetStream, masterKey []
if len(masterKey) == 0 {
return nil, errors.New("ecrauth: empty master key")
}
return kvutil.GetOrCreateBucket(ctx, js, SigningBucket, signingKeyHistory)
return kvstore.GetOrCreateBucket(ctx, js, SigningBucket, signingKeyHistory)
}
// reloadKeys reads every stored signing key, returning the verify set, each
@@ -125,7 +125,7 @@ func openSigningBucket(ctx context.Context, js jetstream.JetStream, masterKey []
// load instead takes the gateway down on every node, permanently, over a key no
// live request needs.
func reloadKeys(ctx context.Context, kv jetstream.KeyValue, masterKey []byte) (active *SigningKey, verify map[string]*ecdsa.PublicKey, metas []keyMeta, undecodable int, err error) {
names, err := kvutil.Keys(ctx, kv)
names, err := kvstore.Keys(ctx, kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil, nil, 0, fmt.Errorf("list signing keys: %w", err)
}
+2 -2
View File
@@ -10,7 +10,7 @@ import (
"sync"
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -179,7 +179,7 @@ func (s *KVMetaStore) bucket(ctx context.Context, accountID string) (jetstream.K
return kv, nil
}
name := KVAccountBucket(accountID)
kv, err := kvutil.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
kv, err := kvstore.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
if err != nil {
return nil, fmt.Errorf("ecr: open account bucket %s: %w", name, err)
}
+3 -3
View File
@@ -7,7 +7,7 @@ import (
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvlease"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -158,7 +158,7 @@ func accountBuckets(ctx context.Context, nc *nats.Conn) ([]accountBucket, error)
if err != nil {
return nil, fmt.Errorf("jetstream: %w", err)
}
names, err := kvutil.BucketNames(ctx, js)
names, err := kvstore.BucketNames(ctx, js)
if err != nil {
return nil, err
}
@@ -176,7 +176,7 @@ func accountBuckets(ctx context.Context, nc *nats.Conn) ([]accountBucket, error)
// return the existing handle.
func GetOrCreateAccountBucket(ctx context.Context, js jetstream.JetStream, accountID string) (jetstream.KeyValue, error) {
bucket := AccountBucketName(accountID)
kv, err := kvutil.GetOrCreateBucket(ctx, js, bucket, KVBucketECSAccountHistory)
kv, err := kvstore.GetOrCreateBucket(ctx, js, bucket, KVBucketECSAccountHistory)
if err != nil {
return nil, fmt.Errorf("failed to create ECS per-account KV bucket %s: %w", bucket, err)
}
+2 -2
View File
@@ -10,7 +10,7 @@ import (
"slices"
"strings"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -128,7 +128,7 @@ func deleteRecord(ctx context.Context, kv jetstream.KeyValue, cluster, principal
// casUpdate does a revision-checked read-modify-write. mutate returns true
// when a field changed. Returns ErrNotFound if absent.
func casUpdate(ctx context.Context, kv jetstream.KeyValue, cluster, principalARN string, mutate func(*Record) bool) (*Record, error) {
return kvutil.Update(ctx, kv, Key(cluster, principalARN), kvutil.CASConfig{NotFound: ErrNotFound}, func(r *Record) (bool, error) {
return kvstore.Update(ctx, kv, Key(cluster, principalARN), kvstore.CASConfig{NotFound: ErrNotFound}, func(r *Record) (bool, error) {
return mutate(r), nil
})
}
+2 -2
View File
@@ -8,7 +8,7 @@ import (
"slices"
"strings"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -193,7 +193,7 @@ func deleteRecord(ctx context.Context, kv jetstream.KeyValue, cluster, addon str
// casUpdate does a revision-checked read-modify-write. mutate returns true
// when a field changed. Returns ErrNotFound if absent.
func casUpdate(ctx context.Context, kv jetstream.KeyValue, cluster, addon string, mutate func(*Record) bool) (*Record, error) {
return kvutil.Update(ctx, kv, Key(cluster, addon), kvutil.CASConfig{NotFound: ErrNotFound}, func(r *Record) (bool, error) {
return kvstore.Update(ctx, kv, Key(cluster, addon), kvstore.CASConfig{NotFound: ErrNotFound}, func(r *Record) (bool, error) {
return mutate(r), nil
})
}
@@ -9,7 +9,7 @@ import (
"strings"
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -280,7 +280,7 @@ func casUpdateMeta(ctx context.Context, kv jetstream.KeyValue, name string, muta
if name == "" {
return errors.New("eks: casUpdateMeta empty name")
}
_, err := kvutil.Update(ctx, kv, ClusterMetaKey(name), kvutil.CASConfig{
_, err := kvstore.Update(ctx, kv, ClusterMetaKey(name), kvstore.CASConfig{
Attempts: maxClusterStateCASRetries,
NotFound: ErrClusterNotFound,
Exhausted: func(string, int) error {
@@ -435,7 +435,7 @@ func putIfGeneration(ctx context.Context, kv jetstream.KeyValue, meta *ClusterMe
if err == nil {
return nil
}
if !kvutil.IsRevisionConflict(err) {
if !kvstore.IsRevisionConflict(err) {
return fmt.Errorf("kv write %s: %w", key, err)
}
}
@@ -8,7 +8,7 @@ import (
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvlease"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
telemetry "github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
"github.com/nats-io/nats.go/jetstream"
)
@@ -292,7 +292,7 @@ func (s *EKSServiceImpl) removeClusterRecord(ctx context.Context, acctKV jetstre
}
errRefsAdded := errors.New("refs added")
_, err = kvutil.DeleteIf(ctx, acctKV, ClusterMetaKey(name), kvutil.CASConfig{
_, err = kvstore.DeleteIf(ctx, acctKV, ClusterMetaKey(name), kvstore.CASConfig{
Attempts: maxClusterStateCASRetries,
NotFound: ErrClusterNotFound,
}, func(m *ClusterMeta) error {
@@ -15,7 +15,7 @@ import (
ec2inventory "github.com/mulgadc/spinifex/spinifex/domains/ec2/inventory"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -150,13 +150,13 @@ func TestReportControlPlaneOrphansPrintsRows(t *testing.T) {
func TestFindControlPlaneOrphansCorruptENIRecordIsUnreadable(t *testing.T) {
f := newEKSServiceFixture(t)
js := testutil.NewJetStream(t, f.svc.deps.NATSConn)
enis, err := kvutil.GetOrCreateBucket(t.Context(), js, ec2vpc.KVBucketENIs, 10)
enis, err := kvstore.GetOrCreateBucket(t.Context(), js, ec2vpc.KVBucketENIs, 10)
require.NoError(t, err)
good, err := json.Marshal(ec2vpc.ENIRecord{NetworkInterfaceId: "eni-good", Tags: cpTags("gone", testAccountID)})
require.NoError(t, err)
_, err = enis.Put(t.Context(), kvutil.AccountKey("000000000000", "eni-good"), good)
_, err = enis.Put(t.Context(), kvstore.AccountKey("000000000000", "eni-good"), good)
require.NoError(t, err)
_, err = enis.Put(t.Context(), kvutil.AccountKey("000000000000", "eni-corrupt"), []byte("{not json"))
_, err = enis.Put(t.Context(), kvstore.AccountKey("000000000000", "eni-corrupt"), []byte("{not json"))
require.NoError(t, err)
res := []*ec2.Reservation{{Instances: []*ec2.Instance{
orphanInstance("i-good", "eni-good"), orphanInstance("i-corrupt", "eni-corrupt"), orphanInstance("i-absent", "eni-absent"),
@@ -11,7 +11,7 @@ import (
"github.com/aws/aws-sdk-go/service/elbv2"
testutil "github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
@@ -69,7 +69,7 @@ func TestS7_EKSBillableReaperNeverTerminatesAControlPlane(t *testing.T) {
account: testAccountID,
prepare: func(t *testing.T, f *eksServiceFixture) {
require.NoError(t, PutClusterMeta(t.Context(), f.kv, sampleClusterMeta("orphan")))
_, err := kvutil.DeleteIf(t.Context(), f.kv, ClusterMetaKey("orphan"), kvutil.CASConfig{
_, err := kvstore.DeleteIf(t.Context(), f.kv, ClusterMetaKey("orphan"), kvstore.CASConfig{
Attempts: maxClusterStateCASRetries,
NotFound: ErrClusterNotFound,
}, func(*ClusterMeta) error { return nil })
+2 -2
View File
@@ -33,7 +33,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/paging"
"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -2138,7 +2138,7 @@ func (s *EKSServiceImpl) casUpdateNodegroupTags(ctx context.Context, accountID,
if err != nil {
return err
}
_, err = kvutil.Update(ctx, acctKV, NodegroupKey(cluster, ng), kvutil.CASConfig{
_, err = kvstore.Update(ctx, acctKV, NodegroupKey(cluster, ng), kvstore.CASConfig{
Attempts: ngCASMaxRetries,
NotFound: errors.New(awserrors.ErrorEKSResourceNotFound),
Exhausted: func(string, int) error { return errors.New(awserrors.ErrorServerInternal) },
+2 -3
View File
@@ -9,7 +9,6 @@ import (
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvlease"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -157,7 +156,7 @@ func accountBucketNames(ctx context.Context, nc *nats.Conn) ([]string, error) {
if err != nil {
return nil, fmt.Errorf("jetstream: %w", err)
}
all, err := kvutil.BucketNames(ctx, js)
all, err := kvstore.BucketNames(ctx, js)
if err != nil {
return nil, err
}
@@ -197,7 +196,7 @@ func AccountWatchBuckets(ctx context.Context, nc *nats.Conn) ([]*kvstore.Bucket,
// existing handle.
func GetOrCreateAccountBucket(ctx context.Context, js jetstream.JetStream, accountID string) (jetstream.KeyValue, error) {
bucket := AccountBucketName(accountID)
kv, err := kvutil.GetOrCreateBucket(ctx, js, bucket, KVBucketEKSAccountHistory)
kv, err := kvstore.GetOrCreateBucket(ctx, js, bucket, KVBucketEKSAccountHistory)
if err != nil {
return nil, fmt.Errorf("failed to create EKS per-account KV bucket %s: %w", bucket, err)
}
@@ -6,7 +6,7 @@ import (
"github.com/mulgadc/spinifex/internal/testkit"
ekscluster "github.com/mulgadc/spinifex/spinifex/domains/eks/cluster"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -24,7 +24,7 @@ func TestAccountWatchBuckets_ReturnsOnePerAccountBucket(t *testing.T) {
require.NoError(t, err)
_, err = ekscluster.GetOrCreateAccountBucket(t.Context(), js, "222222222222")
require.NoError(t, err)
_, err = kvutil.GetOrCreateBucket(t.Context(), js, "something-else", 1)
_, err = kvstore.GetOrCreateBucket(t.Context(), js, "something-else", 1)
require.NoError(t, err)
buckets, err := ekscluster.AccountWatchBuckets(t.Context(), nc)
+2 -2
View File
@@ -10,7 +10,7 @@ import (
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -94,7 +94,7 @@ func (s *Store) GetLBHealth(ctx context.Context, lbID string) (*LBDataPathRecord
// UpdateLBHealth applies mutate to the evidence record under optimistic
// concurrency, creating it when absent. mutate returns false to write nothing.
func (s *Store) UpdateLBHealth(ctx context.Context, lbID string, mutate func(*LBDataPathRecord) (bool, error)) (*LBDataPathRecord, error) {
return kvutil.Update(ctx, s.kv, KeyPrefixLBHealth+lbID, kvutil.CASConfig{
return kvstore.Update(ctx, s.kv, KeyPrefixLBHealth+lbID, kvstore.CASConfig{
Attempts: targetGroupCASAttempts,
CreateIfAbsent: true,
Exhausted: func(_ string, attempts int) error {
+5 -6
View File
@@ -11,7 +11,6 @@ import (
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -35,7 +34,7 @@ const (
maxLBNameClaimRetries = 5
// targetGroupCASAttempts bounds a target-group read-modify-write. It sits
// above the kvutil default because Terraform registers targets at its own
// above the kvstore default because Terraform registers targets at its own
// parallelism (10 by default), every one contending for the same record.
targetGroupCASAttempts = 12
@@ -90,7 +89,7 @@ func NewStore(ctx context.Context, nc *nats.Conn) (*Store, error) {
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketELBv2, 1)
kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketELBv2, 1)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketELBv2, err)
}
@@ -189,7 +188,7 @@ var ErrLoadBalancerNotFound = errors.New("elbv2: load balancer not found")
// concurrency: a writer that loses the race re-reads and retries instead of
// overwriting the winner. mutate returns false to commit nothing.
func (s *Store) UpdateLoadBalancer(ctx context.Context, lbID string, mutate func(*LoadBalancerRecord) (bool, error)) (*LoadBalancerRecord, error) {
return kvutil.Update(ctx, s.kv, KeyPrefixLB+lbID, kvutil.CASConfig{
return kvstore.Update(ctx, s.kv, KeyPrefixLB+lbID, kvstore.CASConfig{
Attempts: targetGroupCASAttempts,
NotFound: ErrLoadBalancerNotFound,
Exhausted: func(_ string, attempts int) error {
@@ -294,7 +293,7 @@ var ErrTargetGroupNotFound = errors.New("elbv2: target group not found")
// concurrency: a writer that loses the race re-reads and retries instead of
// overwriting the winner. mutate returns false to commit nothing.
func (s *Store) UpdateTargetGroup(ctx context.Context, tgID string, mutate func(*TargetGroupRecord) (bool, error)) (*TargetGroupRecord, error) {
return kvutil.Update(ctx, s.kv, KeyPrefixTG+tgID, kvutil.CASConfig{
return kvstore.Update(ctx, s.kv, KeyPrefixTG+tgID, kvstore.CASConfig{
Attempts: targetGroupCASAttempts,
NotFound: ErrTargetGroupNotFound,
Exhausted: func(_ string, attempts int) error {
@@ -652,7 +651,7 @@ func listByPrefixOpt[T any](ctx context.Context, kv jetstream.KeyValue, prefix s
var result []*T
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, prefix) {
+2 -2
View File
@@ -9,7 +9,7 @@ import (
"fmt"
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -41,7 +41,7 @@ type AccountNameIndex struct {
// NewAccountNameIndex opens (or creates) the reservation bucket at the
// cluster's replica count.
func NewAccountNameIndex(ctx context.Context, js jetstream.JetStream) (*AccountNameIndex, error) {
kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketAccountNames, 5)
kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccountNames, 5)
if err != nil {
return nil, fmt.Errorf("init account names bucket: %w", err)
}
+6 -6
View File
@@ -16,7 +16,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
// maxGroupsPerUser caps how many groups a single user may belong to. Mirrors the
@@ -106,7 +106,7 @@ func (s *IAMServiceImpl) GetGroup(accountID string, input *iam.GetGroupInput) (*
func (s *IAMServiceImpl) ListGroups(accountID string, input *iam.ListGroupsInput) (*iam.ListGroupsOutput, error) {
ctx := context.Background()
keys, err := kvutil.Keys(ctx, s.groupsBucket)
keys, err := kvstore.Keys(ctx, s.groupsBucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return &iam.ListGroupsOutput{
@@ -127,7 +127,7 @@ func (s *IAMServiceImpl) ListGroups(accountID string, input *iam.ListGroupsInput
// element at all rather than an empty one.
groups := []*iam.Group{}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, keyPrefix) {
@@ -456,7 +456,7 @@ func (s *IAMServiceImpl) getGroup(ctx context.Context, accountID, groupName stri
// updateUserCAS.
func (s *IAMServiceImpl) updateGroupCAS(ctx context.Context, accountID, groupName string, mutate func(*Group) (bool, error)) error {
defer s.recordLocks.lock("group/" + accountID + "." + groupName)()
_, err := kvutil.Update(ctx, s.groupsBucket, accountID+"."+groupName, iamRecordCASConfig("group", accountID, groupName), mutate)
_, err := kvstore.Update(ctx, s.groupsBucket, accountID+"."+groupName, iamRecordCASConfig("group", accountID, groupName), mutate)
return err
}
@@ -465,7 +465,7 @@ func (s *IAMServiceImpl) updateGroupCAS(ctx context.Context, accountID, groupNam
// unmarshal errors so DeleteGroup cannot succeed while a real-but-unreadable
// member exists. Mirrors findInstanceProfilesForRole.
func (s *IAMServiceImpl) findGroupMembers(ctx context.Context, accountID, groupName string) ([]*User, error) {
keys, err := kvutil.Keys(ctx, s.usersBucket)
keys, err := kvstore.Keys(ctx, s.usersBucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
@@ -476,7 +476,7 @@ func (s *IAMServiceImpl) findGroupMembers(ctx context.Context, accountID, groupN
keyPrefix := accountID + "."
var members []*User
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, keyPrefix) {
+7 -7
View File
@@ -17,7 +17,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
// Bound on optimistic-concurrency retries when a concurrent writer wins the
@@ -97,7 +97,7 @@ func (s *IAMServiceImpl) GetInstanceProfile(accountID string, input *iam.GetInst
func (s *IAMServiceImpl) ListInstanceProfiles(accountID string, input *iam.ListInstanceProfilesInput) (*iam.ListInstanceProfilesOutput, error) {
ctx := context.Background()
keys, err := kvutil.Keys(ctx, s.instanceProfilesBucket)
keys, err := kvstore.Keys(ctx, s.instanceProfilesBucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return &iam.ListInstanceProfilesOutput{
@@ -118,7 +118,7 @@ func (s *IAMServiceImpl) ListInstanceProfiles(accountID string, input *iam.ListI
// to no element at all rather than an empty one.
profiles := []*iam.InstanceProfile{}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, keyPrefix) {
@@ -169,7 +169,7 @@ func (s *IAMServiceImpl) DeleteInstanceProfile(accountID string, input *iam.Dele
profileName := *input.InstanceProfileName
// Revision-guarded, so a role attached after the check sends the delete back through it.
_, err := kvutil.DeleteIf(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName),
_, err := kvstore.DeleteIf(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName),
func(p *InstanceProfile) error {
if p.RoleName != "" {
return awserrors.Errorf(awserrors.ErrorIAMDeleteConflict, "Cannot delete entity, must remove roles from instance profile first.")
@@ -378,12 +378,12 @@ func (s *IAMServiceImpl) getInstanceProfile(ctx context.Context, accountID, prof
// mutate reports whether it changed the record; a false return commits nothing.
func (s *IAMServiceImpl) updateInstanceProfileCAS(ctx context.Context, accountID, profileName string, mutate func(*InstanceProfile) (bool, error)) error {
defer s.recordLocks.lock("instance-profile/" + accountID + "." + profileName)()
_, err := kvutil.Update(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName), mutate)
_, err := kvstore.Update(ctx, s.instanceProfilesBucket, accountID+"."+profileName, s.instanceProfileCASConfig(accountID, profileName), mutate)
return err
}
func (s *IAMServiceImpl) instanceProfileCASConfig(accountID, profileName string) kvutil.CASConfig {
return kvutil.CASConfig{
func (s *IAMServiceImpl) instanceProfileCASConfig(accountID, profileName string) kvstore.CASConfig {
return kvstore.CASConfig{
Attempts: instanceProfileCASMaxRetries,
NotFound: instanceProfileNotFound(profileName),
Exhausted: func(string, int) error {
+5 -5
View File
@@ -20,7 +20,7 @@ import (
resourcearn "github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
// Per-account IAM bucket holds account-scoped resources (currently the OIDC provider registry).
@@ -52,7 +52,7 @@ func OIDCProviderARN(accountID, issuerHostPath string) string {
// GetOrCreateIAMAccountBucket opens the per-account IAM bucket, creating it on first use.
func GetOrCreateIAMAccountBucket(ctx context.Context, js jetstream.JetStream, accountID string) (jetstream.KeyValue, error) {
return kvutil.GetOrCreateBucket(ctx, js, IAMAccountBucketName(accountID), 1)
return kvstore.GetOrCreateBucket(ctx, js, IAMAccountBucketName(accountID), 1)
}
// OIDCProviderRecord is the stored shape of a registered OIDC identity provider.
@@ -224,7 +224,7 @@ func (s *IAMServiceImpl) ListOpenIDConnectProviders(accountID string, _ *iam.Lis
return nil, fmt.Errorf("open IAM account bucket: %w", err)
}
keys, err := kvutil.Keys(ctx, kv)
keys, err := kvstore.Keys(ctx, kv)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return out, nil
@@ -233,7 +233,7 @@ func (s *IAMServiceImpl) ListOpenIDConnectProviders(accountID string, _ *iam.Lis
}
for _, key := range keys {
if key == kvutil.VersionKey || !strings.HasPrefix(key, oidcProvidersKeyPrefix) {
if key == kvstore.VersionKey || !strings.HasPrefix(key, oidcProvidersKeyPrefix) {
continue
}
entry, err := kv.Get(ctx, key)
@@ -356,7 +356,7 @@ func (s *IAMServiceImpl) updateOIDCProviderCAS(ctx context.Context, accountID, a
return fmt.Errorf("open IAM account bucket: %w", err)
}
defer s.recordLocks.lock("oidc-provider/" + accountID + "/" + OIDCProviderKey(issuer))()
_, err = kvutil.Update(ctx, kv, OIDCProviderKey(issuer), iamRecordCASConfig("oidc-provider", accountID, arn), mutate)
_, err = kvstore.Update(ctx, kv, OIDCProviderKey(issuer), iamRecordCASConfig("oidc-provider", accountID, arn), mutate)
return err
}
+5 -5
View File
@@ -16,7 +16,7 @@ import (
iamarn "github.com/mulgadc/bluebottle/pkg/auth"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
const (
@@ -94,7 +94,7 @@ func (s *IAMServiceImpl) updatePolicyCAS(ctx context.Context, accountID, policyA
return err
}
defer s.recordLocks.lock("policy/" + key)()
_, err = kvutil.Update(ctx, s.policiesBucket, key, cfg, func(p *Policy) (bool, error) {
_, err = kvstore.Update(ctx, s.policiesBucket, key, cfg, func(p *Policy) (bool, error) {
// The key omits the path, so an ARN with the wrong path names no policy.
if p.ARN != policyARN {
return false, errors.New(awserrors.ErrorIAMNoSuchEntity)
@@ -106,14 +106,14 @@ func (s *IAMServiceImpl) updatePolicyCAS(ctx context.Context, accountID, policyA
// policyCASTarget resolves policyARN to its KV key and the CAS settings every
// write to a policy record shares.
func policyCASTarget(accountID, policyARN string) (string, kvutil.CASConfig, error) {
func policyCASTarget(accountID, policyARN string) (string, kvstore.CASConfig, error) {
_, policyName, err := iamarn.ParsePolicyARN(policyARN)
if err != nil {
slog.Debug("policyCASTarget: unparseable policy ARN",
"accountID", accountID, "policyArn", policyARN, "err", err)
return "", kvutil.CASConfig{}, policyNotFound(policyARN)
return "", kvstore.CASConfig{}, policyNotFound(policyARN)
}
return accountID + "." + policyName, kvutil.CASConfig{
return accountID + "." + policyName, kvstore.CASConfig{
Attempts: policyCASMaxRetries,
NotFound: policyNotFound(policyARN),
Exhausted: func(string, int) error {
+6 -6
View File
@@ -18,7 +18,7 @@ import (
iamarn "github.com/mulgadc/bluebottle/pkg/auth"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
const (
@@ -128,7 +128,7 @@ func (s *IAMServiceImpl) GetRole(accountID string, input *iam.GetRoleInput) (*ia
func (s *IAMServiceImpl) ListRoles(accountID string, input *iam.ListRolesInput) (*iam.ListRolesOutput, error) {
ctx := context.Background()
keys, err := kvutil.Keys(ctx, s.rolesBucket)
keys, err := kvstore.Keys(ctx, s.rolesBucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return &iam.ListRolesOutput{
@@ -149,7 +149,7 @@ func (s *IAMServiceImpl) ListRoles(accountID string, input *iam.ListRolesInput)
// element at all rather than an empty one.
roles := []*iam.Role{}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, keyPrefix) {
@@ -574,7 +574,7 @@ func (s *IAMServiceImpl) getRole(ctx context.Context, accountID, roleName string
// whether it changed the record; a false return commits nothing.
func (s *IAMServiceImpl) updateRoleCAS(ctx context.Context, accountID, roleName string, mutate func(*Role) (bool, error)) error {
defer s.recordLocks.lock("role/" + accountID + "." + roleName)()
_, err := kvutil.Update(ctx, s.rolesBucket, accountID+"."+roleName, kvutil.CASConfig{
_, err := kvstore.Update(ctx, s.rolesBucket, accountID+"."+roleName, kvstore.CASConfig{
Attempts: roleCASMaxRetries,
NotFound: noSuchEntity("role", roleName),
Exhausted: func(string, int) error {
@@ -593,7 +593,7 @@ func (s *IAMServiceImpl) updateRoleCAS(ctx context.Context, accountID, roleName
// per-key Get or unmarshal errors so DeleteRole cannot succeed while a
// real-but-unreadable reference exists.
func (s *IAMServiceImpl) findInstanceProfilesForRole(ctx context.Context, accountID, roleName string) ([]*InstanceProfile, error) {
keys, err := kvutil.Keys(ctx, s.instanceProfilesBucket)
keys, err := kvstore.Keys(ctx, s.instanceProfilesBucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
@@ -604,7 +604,7 @@ func (s *IAMServiceImpl) findInstanceProfilesForRole(ctx context.Context, accoun
keyPrefix := accountID + "."
var profiles []*InstanceProfile
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, keyPrefix) {
+32 -32
View File
@@ -29,7 +29,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
awscredentials "github.com/mulgadc/spinifex/spinifex/foundation/aws/credentials"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
)
@@ -143,7 +143,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
return nil, fmt.Errorf("get JetStream context: %w", err)
}
usersBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketUsers, 10)
usersBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketUsers, 10)
if err != nil {
return nil, fmt.Errorf("init users bucket: %w", err)
}
@@ -151,7 +151,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
return nil, fmt.Errorf("migrate %s: %w", KVBucketUsers, err)
}
accessKeysBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketAccessKeys, 5)
accessKeysBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccessKeys, 5)
if err != nil {
return nil, fmt.Errorf("init access keys bucket: %w", err)
}
@@ -159,7 +159,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
return nil, fmt.Errorf("migrate %s: %w", KVBucketAccessKeys, err)
}
policiesBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketPolicies, 10)
policiesBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketPolicies, 10)
if err != nil {
return nil, fmt.Errorf("init policies bucket: %w", err)
}
@@ -167,7 +167,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
return nil, fmt.Errorf("migrate %s: %w", KVBucketPolicies, err)
}
accountsBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketAccounts, 5)
accountsBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccounts, 5)
if err != nil {
return nil, fmt.Errorf("init accounts bucket: %w", err)
}
@@ -175,7 +175,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
return nil, fmt.Errorf("migrate %s: %w", KVBucketAccounts, err)
}
accountCounterBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketAccountCounter, 5)
accountCounterBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketAccountCounter, 5)
if err != nil {
return nil, fmt.Errorf("init account counter bucket: %w", err)
}
@@ -183,7 +183,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
return nil, fmt.Errorf("migrate %s: %w", KVBucketAccountCounter, err)
}
rolesBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketRoles, 10)
rolesBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketRoles, 10)
if err != nil {
return nil, fmt.Errorf("init roles bucket: %w", err)
}
@@ -191,7 +191,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
return nil, fmt.Errorf("migrate %s: %w", KVBucketRoles, err)
}
instanceProfilesBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketInstanceProfiles, 10)
instanceProfilesBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketInstanceProfiles, 10)
if err != nil {
return nil, fmt.Errorf("init instance profiles bucket: %w", err)
}
@@ -199,7 +199,7 @@ func NewIAMServiceImpl(ctx context.Context, natsConn *nats.Conn, masterKey []byt
return nil, fmt.Errorf("migrate %s: %w", KVBucketInstanceProfiles, err)
}
groupsBucket, err := kvutil.GetOrCreateBucket(ctx, js, KVBucketGroups, 10)
groupsBucket, err := kvstore.GetOrCreateBucket(ctx, js, KVBucketGroups, 10)
if err != nil {
return nil, fmt.Errorf("init groups bucket: %w", err)
}
@@ -508,7 +508,7 @@ func (s *IAMServiceImpl) GetUser(accountID string, input *iam.GetUserInput) (*ia
func (s *IAMServiceImpl) ListUsers(accountID string, input *iam.ListUsersInput) (*iam.ListUsersOutput, error) {
ctx := context.Background()
keys, err := kvutil.Keys(ctx, s.usersBucket)
keys, err := kvstore.Keys(ctx, s.usersBucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return &iam.ListUsersOutput{
@@ -529,7 +529,7 @@ func (s *IAMServiceImpl) ListUsers(accountID string, input *iam.ListUsersInput)
// element at all rather than an empty one.
users := []*iam.User{}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, keyPrefix) {
@@ -732,7 +732,7 @@ func (s *IAMServiceImpl) setAccessKeysOwner(ctx context.Context, accountID strin
for _, keyID := range keyIDs {
cfg := iamRecordCASConfig("access key", accountID, keyID)
cfg.NotFound = nil
_, err := kvutil.Update(ctx, s.accessKeysBucket, keyID, cfg, func(ak *AccessKey) (bool, error) {
_, err := kvstore.Update(ctx, s.accessKeysBucket, keyID, cfg, func(ak *AccessKey) (bool, error) {
if ak.UserName == userName {
return false, nil
}
@@ -1189,7 +1189,7 @@ func (s *IAMServiceImpl) seedAdminAccount(ctx context.Context, admin *AdminBoots
// IsEmpty returns true if the users bucket has no entries.
func (s *IAMServiceImpl) IsEmpty() (bool, error) {
ctx := context.Background()
keys, err := kvutil.Keys(ctx, s.usersBucket)
keys, err := kvstore.Keys(ctx, s.usersBucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return true, nil
@@ -1197,7 +1197,7 @@ func (s *IAMServiceImpl) IsEmpty() (bool, error) {
return false, fmt.Errorf("check users bucket: %w", err)
}
for _, key := range keys {
if key != kvutil.VersionKey {
if key != kvstore.VersionKey {
return false, nil
}
}
@@ -1303,7 +1303,7 @@ func (s *IAMServiceImpl) GetAccount(accountID string) (*Account, error) {
// ListAccounts returns all accounts.
func (s *IAMServiceImpl) ListAccounts() ([]*Account, error) {
ctx := context.Background()
keys, err := kvutil.Keys(ctx, s.accountsBucket)
keys, err := kvstore.Keys(ctx, s.accountsBucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return []*Account{}, nil
@@ -1313,7 +1313,7 @@ func (s *IAMServiceImpl) ListAccounts() ([]*Account, error) {
accounts := make([]*Account, 0, len(keys))
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
entry, err := s.accountsBucket.Get(ctx, key)
@@ -1367,8 +1367,8 @@ func (s *IAMServiceImpl) QuiesceAccount(accountID, operationID string, fence uin
return 0, fmt.Errorf("%s (%s) cannot be %s: %w", accountID, reason, AccountStatusTerminating, ErrAccountUndeletable)
}
account, err := kvutil.Update(context.Background(), s.accountsBucket, accountID,
kvutil.CASConfig{NotFound: errors.New(awserrors.ErrorIAMNoSuchEntity)},
account, err := kvstore.Update(context.Background(), s.accountsBucket, accountID,
kvstore.CASConfig{NotFound: errors.New(awserrors.ErrorIAMNoSuchEntity)},
func(a *Account) (bool, error) {
if a.Status != AccountStatusTerminating {
a.Status = AccountStatusTerminating
@@ -1390,7 +1390,7 @@ func (s *IAMServiceImpl) QuiesceAccount(accountID, operationID string, fence uin
return false, nil
})
if err != nil {
if errors.Is(err, kvutil.ErrWrite) || errors.Is(err, kvutil.ErrRead) || errors.Is(err, kvutil.ErrDecode) {
if errors.Is(err, kvstore.ErrWrite) || errors.Is(err, kvstore.ErrRead) || errors.Is(err, kvstore.ErrDecode) {
return 0, fmt.Errorf("store account quiesce: %w", err)
}
return 0, err
@@ -1421,8 +1421,8 @@ func (s *IAMServiceImpl) SetAccountStatus(accountID, status string) (*Account, e
return nil, fmt.Errorf("account %s: %w", accountID, ErrUseQuiesceAccount)
}
account, err := kvutil.Update(context.Background(), s.accountsBucket, accountID,
kvutil.CASConfig{NotFound: errors.New(awserrors.ErrorIAMNoSuchEntity)},
account, err := kvstore.Update(context.Background(), s.accountsBucket, accountID,
kvstore.CASConfig{NotFound: errors.New(awserrors.ErrorIAMNoSuchEntity)},
func(a *Account) (bool, error) {
if a.Status == AccountStatusTerminating && status != AccountStatusTerminating {
return false, fmt.Errorf("account %s is terminating: %w", accountID, ErrAccountUndeletable)
@@ -1435,7 +1435,7 @@ func (s *IAMServiceImpl) SetAccountStatus(accountID, status string) (*Account, e
return true, nil
})
if err != nil {
if errors.Is(err, kvutil.ErrWrite) || errors.Is(err, kvutil.ErrRead) || errors.Is(err, kvutil.ErrDecode) {
if errors.Is(err, kvstore.ErrWrite) || errors.Is(err, kvstore.ErrRead) || errors.Is(err, kvstore.ErrDecode) {
return nil, fmt.Errorf("store account status: %w", err)
}
return nil, err
@@ -1606,7 +1606,7 @@ func (s *IAMServiceImpl) ListPolicies(accountID string, input *iam.ListPoliciesI
if err := checkEnum("policyUsageFilter", input.PolicyUsageFilter, policyUsageValues); err != nil {
return nil, err
}
keys, err := kvutil.Keys(ctx, s.policiesBucket)
keys, err := kvstore.Keys(ctx, s.policiesBucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return &iam.ListPoliciesOutput{
@@ -1628,7 +1628,7 @@ func (s *IAMServiceImpl) ListPolicies(accountID string, input *iam.ListPoliciesI
onlyAttached := aws.BoolValue(input.OnlyAttached)
policies := []*iam.Policy{}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, keyPrefix) {
@@ -1712,7 +1712,7 @@ func (s *IAMServiceImpl) DeletePolicy(accountID string, input *iam.DeletePolicyI
}
// Revision-guarded, so a version created after the checks cannot be deleted unseen.
policy, err := kvutil.DeleteIf(ctx, s.policiesBucket, key, cfg, func(p *Policy) error {
policy, err := kvstore.DeleteIf(ctx, s.policiesBucket, key, cfg, func(p *Policy) error {
if p.ARN != policyARN {
return policyNotFound(policyARN)
}
@@ -2267,7 +2267,7 @@ func (s *IAMServiceImpl) buildPolicyAttachments(ctx context.Context, accountID s
keyPrefix := accountID + "."
for _, scan := range scans {
keys, err := kvutil.Keys(ctx, scan.bucket)
keys, err := kvstore.Keys(ctx, scan.bucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
continue
@@ -2276,7 +2276,7 @@ func (s *IAMServiceImpl) buildPolicyAttachments(ctx context.Context, accountID s
}
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if !strings.HasPrefix(key, keyPrefix) {
@@ -2387,18 +2387,18 @@ func (s *IAMServiceImpl) getUser(ctx context.Context, accountID, userName string
// restore a policy or group membership that a concurrent writer removed.
func (s *IAMServiceImpl) updateUserCAS(ctx context.Context, accountID, userName string, mutate func(*User) (bool, error)) error {
defer s.recordLocks.lock("user/" + accountID + "." + userName)()
_, err := kvutil.Update(ctx, s.usersBucket, accountID+"."+userName, iamRecordCASConfig("user", accountID, userName), mutate)
_, err := kvstore.Update(ctx, s.usersBucket, accountID+"."+userName, iamRecordCASConfig("user", accountID, userName), mutate)
return err
}
// iamRecordCASConfig is the CAS policy for an IAM user, group or OIDC provider
// record: a missing record is NoSuchEntity and exhausted retries are logged.
func iamRecordCASConfig(kind, accountID, name string) kvutil.CASConfig {
func iamRecordCASConfig(kind, accountID, name string) kvstore.CASConfig {
notFound := noSuchEntity(kind, name)
if kind == "oidc-provider" {
notFound = oidcProviderNotFound(name)
}
return kvutil.CASConfig{
return kvstore.CASConfig{
Attempts: roleCASMaxRetries,
NotFound: notFound,
Exhausted: func(string, int) error {
@@ -2828,7 +2828,7 @@ var summaryQuotaDefaults = map[string]int64{
// Get/unmarshal so it stays cheap with many resources. The version key is
// skipped and an empty bucket counts as zero.
func countBucket(ctx context.Context, bucket jetstream.KeyValue, prefix string) (int64, error) {
keys, err := kvutil.Keys(ctx, bucket)
keys, err := kvstore.Keys(ctx, bucket)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return 0, nil
@@ -2838,7 +2838,7 @@ func countBucket(ctx context.Context, bucket jetstream.KeyValue, prefix string)
var count int64
for _, key := range keys {
if key == kvutil.VersionKey {
if key == kvstore.VersionKey {
continue
}
if strings.HasPrefix(key, prefix) {
+6 -54
View File
@@ -49,12 +49,6 @@ type localClient struct {
failCodeStored bool
passOwnerClock bool
rollbacks int
// beginErr fails BeginCodeUpdate before the owner sees it.
beginErr error
// loseUpdatedReply commits CodeUpdated, then reports a transport error.
loseUpdatedReply bool
// beforeCommit runs once, unlocked, before the next CodeUpdated reaches the owner.
beforeCommit func()
}
var _ function.Client = (*localClient)(nil)
@@ -110,37 +104,6 @@ func (c *localClient) RollbackCreate(ctx context.Context, account, region, name
return err
}
func (c *localClient) BeginCodeUpdate(ctx context.Context, account, region string, rec function.Record, rev uint64, codeRevisionID string) error {
c.mu.Lock()
fail := c.beginErr
c.mu.Unlock()
if fail != nil {
return fail
}
_, err := c.svc.BeginCodeUpdate(ctx, &function.CodeUpdateBeginRequest{Region: region, FunctionName: rec.FunctionName,
FunctionID: rec.FunctionID, Revision: rev, RevisionID: codeRevisionID}, account)
return err
}
func (c *localClient) CodeUpdated(ctx context.Context, account, region string, rec function.Record, rev uint64, codeRevisionID, sha string, size int64) (function.Record, uint64, error) {
c.mu.Lock()
hook, lose := c.beforeCommit, c.loseUpdatedReply
c.beforeCommit = nil
c.mu.Unlock()
if hook != nil {
hook()
}
r, err := c.svc.CodeUpdated(ctx, &function.CodeUpdatedRequest{Region: region, FunctionName: rec.FunctionName, FunctionID: rec.FunctionID,
Revision: rev, RevisionID: codeRevisionID, SHA256: sha, Size: size}, account)
if err != nil {
return function.Record{}, 0, err
}
if lose {
return function.Record{}, 0, errors.New("nats: timeout")
}
return r.Record, r.Revision, nil
}
// memStore is an in-memory code store; objects is keyed by Ref.
type memStore struct {
mu sync.Mutex
@@ -218,16 +181,6 @@ func newHarness(t *testing.T, statements ...iamdomain.Statement) (*gwtest.Harnes
// uploads returns every upload record in the account bucket, by function ID.
func (e *env) uploads(t *testing.T) map[string]function.Upload {
t.Helper()
out := map[string]function.Upload{}
for _, up := range e.uploadsByRevision(t) {
out[up.FunctionID] = up
}
return out
}
// uploadsByRevision returns every upload record in the account bucket, by code revision ID.
func (e *env) uploadsByRevision(t *testing.T) map[string]function.Upload {
t.Helper()
kv := e.kv(t)
keys, err := kv.ListKeys(t.Context())
@@ -242,7 +195,7 @@ func (e *env) uploadsByRevision(t *testing.T) map[string]function.Upload {
var up function.Upload
require.NoError(t, json.Unmarshal(entry.Value(), &up))
require.Equal(t, function.KVUploadKey(up.Region, up.FunctionID, up.RevisionID), k)
out[up.RevisionID] = up
out[up.FunctionID] = up
}
return out
}
@@ -250,13 +203,13 @@ func (e *env) uploadsByRevision(t *testing.T) map[string]function.Upload {
// requireTracked asserts every code copy in the store has an upload record.
func (e *env) requireTracked(t *testing.T) {
t.Helper()
ups := e.uploadsByRevision(t)
ups := e.uploads(t)
e.store.mu.Lock()
defer e.store.mu.Unlock()
for ref := range e.store.objects {
up, ok := ups[ref.RevisionID]
require.True(t, ok, "code copy %s/%s has no upload record", ref.FunctionID, ref.RevisionID)
require.Equal(t, ref.FunctionID, up.FunctionID)
up, ok := ups[ref.FunctionID]
require.True(t, ok, "code copy %s has no upload record", ref.FunctionID)
require.Equal(t, ref.RevisionID, up.RevisionID)
}
}
@@ -836,14 +789,13 @@ func TestAuthorization(t *testing.T) {
func TestRegistration_ResolvesActionsAndInventory(t *testing.T) {
reg := awsapi.NewRegistration(awsapi.Deps{})
assert.Equal(t, []string{"CreateEventSourceMapping", "CreateFunction", "DeleteEventSourceMapping", "GetEventSourceMapping", "GetFunction", "GetFunctionCodeSigningConfig", "GetFunctionConfiguration", "Invoke", "ListEventSourceMappings", "ListVersionsByFunction", "UpdateFunctionCode"}, reg.Inventory.Registered)
assert.Equal(t, []string{"CreateEventSourceMapping", "CreateFunction", "DeleteEventSourceMapping", "GetEventSourceMapping", "GetFunction", "GetFunctionCodeSigningConfig", "GetFunctionConfiguration", "Invoke", "ListEventSourceMappings", "ListVersionsByFunction"}, reg.Inventory.Registered)
assert.Equal(t, "ListVersionsByFunction", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2015-03-31/functions/x/versions", nil)))
assert.Equal(t, "DeleteEventSourceMapping", reg.ResolveAction(httptest.NewRequest(http.MethodDelete, "/2015-03-31/event-source-mappings/x", nil)))
assert.Equal(t, "GetFunctionCodeSigningConfig", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2020-06-30/functions/x/code-signing-config", nil)))
assert.Equal(t, "CreateFunction", reg.ResolveAction(httptest.NewRequest(http.MethodPost, "/2015-03-31/functions", nil)))
assert.Equal(t, "GetFunction", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2015-03-31/functions/x", nil)))
assert.Equal(t, "GetFunctionConfiguration", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2015-03-31/functions/x/configuration", nil)))
assert.Equal(t, "UpdateFunctionCode", reg.ResolveAction(httptest.NewRequest(http.MethodPut, "/2015-03-31/functions/x/code", nil)))
assert.Empty(t, reg.ResolveAction(httptest.NewRequest(http.MethodDelete, "/2015-03-31/functions/x", nil)))
}
@@ -138,7 +138,7 @@ func (d Deps) serve(w http.ResponseWriter, inv dispatch.Invocation) error {
if decodeErr == nil {
resource = functionARN(inv.Region, inv.AccountID, aws.StringValue(c.Create.FunctionName))
}
case "GetFunction", "GetFunctionConfiguration", "GetFunctionCodeSigningConfig", "ListVersionsByFunction", "UpdateFunctionCode":
case "GetFunction", "GetFunctionConfiguration", "GetFunctionCodeSigningConfig", "ListVersionsByFunction":
if name, _, err := resolveFunctionName(inv.Region, inv.AccountID, params[0]); err == nil {
resource = functionARN(inv.Region, inv.AccountID, name)
}
@@ -167,8 +167,8 @@ func (d Deps) serve(w http.ResponseWriter, inv dispatch.Invocation) error {
return err
}
}
// Only CreateFunction, GetFunction and UpdateFunctionCode touch the code store.
needsCodes := action == "CreateFunction" || action == "GetFunction" || action == "UpdateFunctionCode"
// Only CreateFunction and GetFunction touch the code store.
needsCodes := action == "CreateFunction" || action == "GetFunction"
if d.Meta == nil || (d.Codes == nil && needsCodes) {
return errors.New(awserrors.ErrorServiceException)
}
-4
View File
@@ -50,10 +50,6 @@ var routes = []route{
Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
return d.getFunction(ctx, c)
}},
{Method: "PUT", Pattern: "/2015-03-31/functions/{FunctionName}/code", Action: "UpdateFunctionCode",
Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
return d.updateFunctionCode(ctx, c)
}},
{Method: "GET", Pattern: "/2015-03-31/functions/{FunctionName}/configuration", Action: "GetFunctionConfiguration",
Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
return d.getFunctionConfiguration(ctx, c)
@@ -1,113 +0,0 @@
package awsapi
import (
"context"
"log/slog"
"net/http"
"strconv"
"github.com/aws/aws-sdk-go/aws"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
)
// ownerError passes an AWS-coded owner refusal through and turns anything
// else, such as an owner with no responder for the subject, into a retryable
// ServiceException, logging the stage that failed.
func ownerError(ctx context.Context, stage string, rec function.Record, codeRevisionID string, err error) error {
if _, coded := awserrors.ResolveErrorCode(err); coded {
return err
}
slog.ErrorContext(ctx, "Lambda: code update owner request failed", "stage", stage,
"function_id", rec.FunctionID, "revision_id", codeRevisionID, "err", err)
return serviceError("the function code could not be updated; retry the request")
}
// updateFunctionCode implements UpdateFunctionCode for an inline ZIP. The new
// copy is written under a fresh code revision while the record still names the
// old one, and the record moves to it only by compare-and-swap.
func (d Deps) updateFunctionCode(ctx context.Context, c *call) (reply, error) {
in, err := decodeUpdateCode(c.Body)
if err != nil {
return reply{}, err
}
name, q, err := resolveFunctionName(c.Region, c.AccountID, c.Params[0])
if err != nil {
return reply{}, err
}
if q != "" {
return reply{}, invalidFunctionName()
}
if err := validateUpdateCode(in); err != nil {
return reply{}, err
}
rec, revision, err := d.lookupTarget(ctx, c, name, "")
if err != nil {
return reply{}, err
}
if rec.State != function.StateActive || rec.Code.State != function.CodeStored {
return reply{}, awserrors.Errorf(awserrors.ErrorResourceConflictException,
"The operation cannot be performed at this time. The function is currently in the following state: %s", rec.State)
}
if want := aws.StringValue(in.RevisionId); in.RevisionId != nil && want != strconv.FormatUint(revision, 10) {
return reply{}, awserrors.Errorf(awserrors.ErrorPreconditionFailedException,
"The Revision Id provided does not match the latest Revision Id. Call the GetFunction/GetAlias API to retrieve the latest Revision Id")
}
// The environment is opened before anything is written, so a commit is
// never followed by a response that cannot be rendered.
var env map[string]string
if rec.EnvCipher != "" {
if d.AEAD == nil {
return reply{}, serviceError("environment decryption is unavailable")
}
if env, err = function.OpenEnv(d.AEAD, rec.AccountID, rec.FunctionID, rec.EnvCipher); err != nil {
slog.ErrorContext(ctx, "Lambda: open environment failed", "function_id", rec.FunctionID, "err", err)
return reply{}, serviceError("could not read the function environment")
}
}
if aws.BoolValue(in.DryRun) {
cfg, err := configuration(rec, revision, env)
if err != nil {
return reply{}, serviceError("could not build the function configuration")
}
return reply{Status: http.StatusOK, Body: cfg}, nil
}
codeRevisionID, err := randomID()
if err != nil {
return reply{}, serviceError("could not generate a revision identifier")
}
if err := d.Meta.BeginCodeUpdate(ctx, c.AccountID, c.Region, rec, revision, codeRevisionID); err != nil {
return reply{}, ownerError(ctx, "begin", rec, codeRevisionID, err)
}
ref := codestore.Ref{AccountID: rec.AccountID, FunctionID: rec.FunctionID, RevisionID: codeRevisionID}
store, err := d.Codes(ctx)
if err != nil {
owedCopy(ctx, ref, "code store unavailable", err)
return reply{}, serviceError("the function code could not be stored")
}
sha, err := store.Put(ctx, ref, in.ZipFile)
if err != nil {
owedCopy(ctx, ref, "code write failed", err)
return reply{}, serviceError("the function code could not be stored")
}
updated, updatedRev, err := d.Meta.CodeUpdated(ctx, c.AccountID, c.Region, rec, revision, codeRevisionID, sha, int64(len(in.ZipFile)))
if err != nil {
// The commit may have landed with its reply lost; nothing is undone.
owedCopy(ctx, ref, "commit not confirmed", err)
return reply{}, ownerError(ctx, "commit", rec, codeRevisionID, err)
}
cfg, err := configuration(updated, updatedRev, env)
if err != nil {
return reply{}, serviceError("could not build the function configuration")
}
return reply{Status: http.StatusOK, Body: cfg}, nil
}
// owedCopy logs a code update whose new copy may exist without the function
// naming it; its upload record stays owed for later cleanup.
func owedCopy(ctx context.Context, ref codestore.Ref, why string, cause error) {
slog.WarnContext(ctx, "Lambda: code update did not complete; code object cleanup may be owed", "reason", why,
"account_id", ref.AccountID, "function_id", ref.FunctionID, "revision_id", ref.RevisionID, "err", cause)
}
@@ -1,353 +0,0 @@
package awsapi_test
import (
"context"
"crypto/sha256"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"net/http"
"strconv"
"testing"
"github.com/aws/aws-sdk-go/service/lambda"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/internal/testkit/gwtest"
iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
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"
)
const updatePath = "/2015-03-31/functions/hello/code"
var updateZip = buildZip("hello, version two")
func codeSHA(code []byte) string {
sum := sha256.Sum256(code)
return base64.StdEncoding.EncodeToString(sum[:])
}
// activate moves every stored function to Active through the owner's own pass.
func (e *env) activate(t *testing.T) {
t.Helper()
e.meta.svc.SetReadiness(func(context.Context, function.Record) error { return nil })
require.NoError(t, e.meta.svc.ActivatePending(t.Context()))
}
// record reads the stored function record and its revision.
func (e *env) record(t *testing.T, name string) (function.Record, uint64) {
t.Helper()
entry, err := e.kv(t).Get(t.Context(), function.KVFunctionKey(gwtest.Region, name))
require.NoError(t, err)
var rec function.Record
require.NoError(t, json.Unmarshal(entry.Value(), &rec))
return rec, entry.Revision()
}
// newUpdateEnv creates an Active "hello" with an environment and non-default configuration.
func newUpdateEnv(t *testing.T, statements ...iamdomain.Statement) *env {
t.Helper()
e := newEnv(t, statements...)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
b["Environment"] = map[string]any{"Variables": map[string]string{"MODE": "prod"}}
b["Description"] = "demo"
b["Timeout"] = 30
b["MemorySize"] = 256
}))
require.NoError(t, err)
e.activate(t)
return e
}
func updateBody(mutate func(map[string]any)) map[string]any {
b := map[string]any{"ZipFile": updateZip}
if mutate != nil {
mutate(b)
}
return b
}
func decodeConfig(t *testing.T, raw []byte) lambda.FunctionConfiguration {
t.Helper()
var cfg lambda.FunctionConfiguration
require.NoError(t, json.Unmarshal(raw, &cfg))
return cfg
}
// requireUnchanged asserts the record still names the original copy at its original revision.
func requireUnchanged(t *testing.T, e *env, before function.Record, beforeRev uint64) {
t.Helper()
after, rev := e.record(t, "hello")
assert.Equal(t, beforeRev, rev, "the function record was written")
assert.Equal(t, before.Code, after.Code)
}
func TestUpdateFunctionCode_SwapsTheCodeAndKeepsEverythingElse(t *testing.T) {
e := newUpdateEnv(t)
before, beforeRev := e.record(t, "hello")
rec, err := e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) {
b["Architectures"] = []string{"x86_64"}
b["RevisionId"] = strconv.FormatUint(beforeRev, 10)
}))
require.NoError(t, err)
require.Equal(t, http.StatusOK, rec.Code)
cfg := decodeConfig(t, rec.Body.Bytes())
assert.Equal(t, codeSHA(updateZip), *cfg.CodeSha256)
assert.Equal(t, int64(len(updateZip)), *cfg.CodeSize)
assert.Equal(t, "Active", *cfg.State)
assert.Equal(t, "Successful", *cfg.LastUpdateStatus)
assert.Equal(t, "prod", *cfg.Environment.Variables["MODE"])
assert.Equal(t, "demo", *cfg.Description)
assert.Equal(t, int64(30), *cfg.Timeout)
assert.Equal(t, int64(256), *cfg.MemorySize)
assert.Equal(t, roleARN, *cfg.Role)
after, afterRev := e.record(t, "hello")
assert.Equal(t, strconv.FormatUint(afterRev, 10), *cfg.RevisionId)
assert.Greater(t, afterRev, beforeRev)
assert.Equal(t, before.FunctionID, after.FunctionID)
assert.Equal(t, before.EnvCipher, after.EnvCipher)
assert.Equal(t, before.CreatedAt, after.CreatedAt)
assert.NotEqual(t, before.Code.RevisionID, after.Code.RevisionID)
assert.Equal(t, function.CodeStored, after.Code.State)
assert.Equal(t, function.StateActive, after.State)
// Both copies stay, each under its own revision and tracked.
require.Equal(t, 2, e.store.count())
assert.Equal(t, testZip, e.store.objects[codestore.Ref{AccountID: gwtest.AccountID, FunctionID: before.FunctionID, RevisionID: before.Code.RevisionID}])
assert.Equal(t, updateZip, e.store.objects[codestore.Ref{AccountID: gwtest.AccountID, FunctionID: after.FunctionID, RevisionID: after.Code.RevisionID}])
e.requireTracked(t)
ups := e.uploadsByRevision(t)
assert.Equal(t, function.UploadAttached, ups[before.Code.RevisionID].State)
assert.Equal(t, function.UploadAttached, ups[after.Code.RevisionID].State)
assert.Equal(t, afterRev, ups[after.Code.RevisionID].AttachedRevision)
got, err := e.do(http.MethodGet, "/2015-03-31/functions/hello", nil)
require.NoError(t, err)
var out lambda.GetFunctionOutput
require.NoError(t, json.Unmarshal(got.Body.Bytes(), &out))
assert.Equal(t, codeSHA(updateZip), *out.Configuration.CodeSha256)
}
func TestUpdateFunctionCode_RefusesWhatThisBuildDoesNotImplement(t *testing.T) {
cases := map[string]func(map[string]any){
"S3 bucket": func(b map[string]any) { delete(b, "ZipFile"); b["S3Bucket"] = "code" },
"S3 key": func(b map[string]any) { b["S3Key"] = "fn.zip" },
"S3 object version": func(b map[string]any) { b["S3ObjectVersion"] = "1" },
"image": func(b map[string]any) { delete(b, "ZipFile"); b["ImageUri"] = "repo/fn:latest" },
"publish": func(b map[string]any) { b["Publish"] = true },
"arm64": func(b map[string]any) { b["Architectures"] = []string{"arm64"} },
"two architectures": func(b map[string]any) { b["Architectures"] = []string{"x86_64", "arm64"} },
"empty architectures": func(b map[string]any) { b["Architectures"] = []string{} },
"unknown arch": func(b map[string]any) { b["Architectures"] = []string{"sparc"} },
"no source": func(b map[string]any) { delete(b, "ZipFile") },
"not a zip": func(b map[string]any) { b["ZipFile"] = []byte("plain bytes") },
"unmodelled member": func(b map[string]any) { b["SourceKMSKeyArn"] = "arn:aws:kms:x" },
}
e := newUpdateEnv(t)
before, beforeRev := e.record(t, "hello")
for name, mutate := range cases {
t.Run(name, func(t *testing.T) {
_, err := e.do(http.MethodPut, updatePath, updateBody(mutate))
testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
})
}
t.Run("publish names versions", func(t *testing.T) {
_, err := e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) { b["Publish"] = true }))
testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
assert.Contains(t, errMessage(err), "Publish")
assert.Contains(t, errMessage(err), "versions are not yet supported")
})
t.Run("qualified name", func(t *testing.T) {
_, err := e.do(http.MethodPut, "/2015-03-31/functions/hello:1/code", updateBody(nil))
testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
})
requireUnchanged(t, e, before, beforeRev)
assert.Equal(t, 1, e.store.count())
assert.Len(t, e.uploadsByRevision(t), 1)
}
func TestUpdateFunctionCode_MissingOrInactiveFunction(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
testutil.RequireAWSCode(t, err, awserrors.ErrorResourceNotFoundException)
_, err = e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
require.NoError(t, err)
before, beforeRev := e.record(t, "hello")
_, err = e.do(http.MethodPut, updatePath, updateBody(nil))
testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
assert.Contains(t, errMessage(err), "Pending")
requireUnchanged(t, e, before, beforeRev)
assert.Equal(t, 1, e.store.count())
}
func TestUpdateFunctionCode_Authorization(t *testing.T) {
arn := "arn:aws:lambda:" + gwtest.Region + ":" + gwtest.AccountID + ":function:hello"
t.Run("denied on the function ARN", func(t *testing.T) {
e := newUpdateEnv(t, gwtest.Statement("Allow", "*", "*"), gwtest.Statement("Deny", "lambda:UpdateFunctionCode", arn))
before, beforeRev := e.record(t, "hello")
_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
gwtest.AssertDenied(t, err)
requireUnchanged(t, e, before, beforeRev)
assert.Equal(t, 1, e.store.count())
})
t.Run("another function's grant does not reach it", func(t *testing.T) {
e := newUpdateEnv(t, gwtest.Statement("Allow", "lambda:CreateFunction", "*"), gwtest.Statement("Allow", "iam:PassRole", "*"),
gwtest.Statement("Allow", "lambda:UpdateFunctionCode", arn+"-other"))
_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
gwtest.AssertDenied(t, err)
})
t.Run("allowed on the function ARN", func(t *testing.T) {
e := newUpdateEnv(t, gwtest.Statement("Allow", "lambda:CreateFunction", "*"), gwtest.Statement("Allow", "iam:PassRole", "*"),
gwtest.Statement("Allow", "lambda:UpdateFunctionCode", arn))
_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
require.NoError(t, err)
})
}
func TestUpdateFunctionCode_DryRunWritesNothing(t *testing.T) {
e := newUpdateEnv(t)
before, beforeRev := e.record(t, "hello")
rec, err := e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) { b["DryRun"] = true }))
require.NoError(t, err)
require.Equal(t, http.StatusOK, rec.Code)
cfg := decodeConfig(t, rec.Body.Bytes())
assert.Equal(t, codeSHA(testZip), *cfg.CodeSha256, "a dry run reports the code that still runs")
assert.Equal(t, strconv.FormatUint(beforeRev, 10), *cfg.RevisionId)
requireUnchanged(t, e, before, beforeRev)
assert.Equal(t, 1, e.store.count())
assert.Len(t, e.uploadsByRevision(t), 1)
_, err = e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) { b["DryRun"] = true; b["Publish"] = true }))
testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
}
func TestUpdateFunctionCode_StaleRevisionIDIsAPreconditionFailure(t *testing.T) {
e := newUpdateEnv(t)
before, beforeRev := e.record(t, "hello")
_, err := e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) {
b["RevisionId"] = strconv.FormatUint(beforeRev-1, 10)
}))
testutil.RequireAWSCode(t, err, awserrors.ErrorPreconditionFailedException)
requireUnchanged(t, e, before, beforeRev)
assert.Equal(t, 1, e.store.count())
assert.Len(t, e.uploadsByRevision(t), 1)
}
func TestUpdateFunctionCode_FailedCopyLeavesTheFunctionOnItsOldCode(t *testing.T) {
e := newUpdateEnv(t)
before, beforeRev := e.record(t, "hello")
e.store.putErr = errors.New("predastore: unavailable")
_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
testutil.RequireAWSCode(t, err, awserrors.ErrorServiceException)
requireUnchanged(t, e, before, beforeRev)
ups := e.uploadsByRevision(t)
require.Len(t, ups, 2, "the attempted upload stays recorded")
for rev, up := range ups {
if rev == before.Code.RevisionID {
assert.Equal(t, function.UploadAttached, up.State)
continue
}
assert.Equal(t, function.UploadOwed, up.State, "an upload whose copy may exist stays owed")
}
}
func TestUpdateFunctionCode_OwnerWithoutTheSubjectIsRetryableAndChangesNothing(t *testing.T) {
e := newUpdateEnv(t)
before, beforeRev := e.record(t, "hello")
e.meta.beginErr = fmt.Errorf("NATS request to %s: %w", function.SubjectCodeUpdateBegin, nats.ErrNoResponders)
_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
testutil.RequireAWSCode(t, err, awserrors.ErrorServiceException)
assert.False(t, awserrors.IsTerminal(err), "the caller is told to retry once the owner is upgraded")
requireUnchanged(t, e, before, beforeRev)
assert.Equal(t, 1, e.store.count())
assert.Len(t, e.uploadsByRevision(t), 1)
}
func TestUpdateFunctionCode_CompetingUpdatesCommitOnce(t *testing.T) {
e := newUpdateEnv(t)
other := buildZip("hello, competing version")
// The second update starts and completes while the first is writing its copy.
e.store.beforePut = func() {
_, err := e.do(http.MethodPut, updatePath, updateBody(func(b map[string]any) { b["ZipFile"] = other }))
assert.NoError(t, err)
}
_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
after, _ := e.record(t, "hello")
cfg, err := e.do(http.MethodGet, "/2015-03-31/functions/hello/configuration", nil)
require.NoError(t, err)
assert.Equal(t, codeSHA(other), *decodeConfig(t, cfg.Body.Bytes()).CodeSha256, "the winner's code stays")
assert.Equal(t, 3, e.store.count(), "the original, the winner's and the loser's copies")
e.requireTracked(t)
owed := 0
for rev, up := range e.uploadsByRevision(t) {
if up.State == function.UploadOwed {
owed++
assert.NotEqual(t, after.Code.RevisionID, rev)
}
}
assert.Equal(t, 1, owed, "only the losing upload is owed")
}
func TestUpdateFunctionCode_LostCommitReplyIsAnErrorButTheUpdateStands(t *testing.T) {
e := newUpdateEnv(t)
before, _ := e.record(t, "hello")
e.meta.loseUpdatedReply = true
_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
testutil.RequireAWSCode(t, err, awserrors.ErrorServiceException)
after, _ := e.record(t, "hello")
assert.NotEqual(t, before.Code.RevisionID, after.Code.RevisionID, "the committed update is not rolled back")
cfg, err := e.do(http.MethodGet, "/2015-03-31/functions/hello/configuration", nil)
require.NoError(t, err)
assert.Equal(t, codeSHA(updateZip), *decodeConfig(t, cfg.Body.Bytes()).CodeSha256)
require.Equal(t, 2, e.store.count(), "neither copy is deleted")
ups := e.uploadsByRevision(t)
require.Len(t, ups, 2)
assert.Equal(t, function.UploadAttached, ups[before.Code.RevisionID].State)
assert.Equal(t, function.UploadAttached, ups[after.Code.RevisionID].State)
}
func TestUpdateFunctionCode_AnInvocationInFlightKeepsItsOldCopy(t *testing.T) {
e := newUpdateEnv(t)
ex := &fakeExec{result: okResult, started: make(chan struct{}), release: make(chan struct{})}
e.deps.Exec = ex
before, _ := e.record(t, "hello")
done := make(chan error, 1)
go func() {
_, err := e.invoke(t.Context(), invokePath, nil, nil)
done <- err
}()
<-ex.started
_, err := e.do(http.MethodPut, updatePath, updateBody(nil))
require.NoError(t, err)
close(ex.release)
require.NoError(t, <-done)
ex.mu.Lock()
inflight := ex.calls[0].Function.Code
ex.started = nil
ex.mu.Unlock()
assert.Equal(t, before.Code, inflight, "the running attempt stays bound to the copy it read")
assert.Equal(t, testZip, e.store.objects[codestore.Ref{AccountID: gwtest.AccountID, FunctionID: before.FunctionID, RevisionID: inflight.RevisionID}],
"the old copy is still there for it")
_, err = e.invoke(t.Context(), invokePath, nil, nil)
require.NoError(t, err)
after, _ := e.record(t, "hello")
ex.mu.Lock()
defer ex.mu.Unlock()
assert.Equal(t, after.Code, ex.calls[1].Function.Code, "a later attempt reads the new copy")
assert.NotEqual(t, before.Code.RevisionID, ex.calls[1].Function.Code.RevisionID)
}
+7 -77
View File
@@ -53,15 +53,6 @@ func unsupported(field string) error {
return invalid("%s is not supported", field)
}
// refuseField names a refused member; Publish also says why, since it waits on
// function versions rather than being out of scope.
func refuseField(field string) error {
if field == "Publish" {
return invalid("Publish is not supported: function versions are not yet supported")
}
return unsupported(field)
}
// decodeCreate reads a CreateFunction body strictly: a member this build does
// not model is refused rather than silently dropped.
func decodeCreate(body []byte) (*lambda.CreateFunctionInput, error) {
@@ -219,12 +210,14 @@ func validateCreate(in *lambda.CreateFunctionInput) (*createParams, error) {
if pt := aws.StringValue(in.PackageType); pt != "" && pt != lambda.PackageTypeZip {
return nil, invalid("PackageType must be %s", lambda.PackageTypeZip)
}
var zipFile []byte
if in.Code != nil {
zipFile = in.Code.ZipFile
if in.Code == nil || len(in.Code.ZipFile) == 0 {
return nil, invalid("Code.ZipFile is required")
}
if err := checkZip("Code.ZipFile", zipFile); err != nil {
return nil, err
if len(in.Code.ZipFile) > codestore.MaxCodeBytes {
return nil, awserrors.Errorf(awserrors.ErrorRequestEntityTooLarge, "Code.ZipFile exceeds %d bytes", codestore.MaxCodeBytes)
}
if _, err := zip.NewReader(bytes.NewReader(in.Code.ZipFile), int64(len(in.Code.ZipFile))); err != nil {
return nil, invalid("Could not unzip uploaded file. Please check your file, then try to upload again.")
}
handler := aws.StringValue(in.Handler)
if handler == "" {
@@ -261,69 +254,6 @@ func validateCreate(in *lambda.CreateFunctionInput) (*createParams, error) {
}, nil
}
// checkZip applies the inline ZIP rules CreateFunction and UpdateFunctionCode
// share: present, within the code store bound, and readable as a ZIP.
func checkZip(field string, code []byte) error {
if len(code) == 0 {
return invalid("%s is required", field)
}
if len(code) > codestore.MaxCodeBytes {
return awserrors.Errorf(awserrors.ErrorRequestEntityTooLarge, "%s exceeds %d bytes", field, codestore.MaxCodeBytes)
}
if _, err := zip.NewReader(bytes.NewReader(code), int64(len(code))); err != nil {
return invalid("Could not unzip uploaded file. Please check your file, then try to upload again.")
}
return nil
}
// decodeUpdateCode reads an UpdateFunctionCode body as strictly as
// decodeCreate reads CreateFunction.
func decodeUpdateCode(body []byte) (*lambda.UpdateFunctionCodeInput, error) {
in := new(lambda.UpdateFunctionCodeInput)
dec := json.NewDecoder(bytes.NewReader(body))
dec.DisallowUnknownFields()
if err := dec.Decode(in); err != nil {
return nil, invalid("the request body is not a valid UpdateFunctionCode document: %v", err)
}
return in, nil
}
// validateUpdateCode refuses every code source but an inline ZIP, publishing
// and an architecture change, then checks the ZIP.
func validateUpdateCode(in *lambda.UpdateFunctionCodeInput) error {
checks := []struct {
field string
set bool
}{
{"S3Bucket", isSet(in.S3Bucket)},
{"S3Key", isSet(in.S3Key)},
{"S3ObjectVersion", isSet(in.S3ObjectVersion)},
{"ImageUri", isSet(in.ImageUri)},
{"Publish", aws.BoolValue(in.Publish)},
}
for _, c := range checks {
if c.set {
return refuseField(c.field)
}
}
if in.Architectures != nil {
if len(in.Architectures) == 0 {
return invalid("1 validation error detected: Value at 'architectures' failed to satisfy constraint: Member must have length greater than or equal to 1")
}
if len(in.Architectures) > 1 {
return invalid("1 validation error detected: Value at 'architectures' failed to satisfy constraint: Member must have length less than or equal to 1")
}
switch a := aws.StringValue(in.Architectures[0]); a {
case lambda.ArchitectureX8664:
case lambda.ArchitectureArm64:
return unsupported("Architectures " + a)
default:
return invalid("1 validation error detected: Value at 'architectures' failed to satisfy constraint: Member must satisfy enum value set: [x86_64, arm64]")
}
}
return checkZip("ZipFile", in.ZipFile)
}
func envVars(e *lambda.Environment) (map[string]string, error) {
if e == nil || len(e.Variables) == 0 {
return nil, nil
@@ -9,7 +9,7 @@ import (
"fmt"
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -57,7 +57,7 @@ type Store struct{ kv jetstream.KeyValue }
// Open returns the store, creating its bucket on first use.
func Open(ctx context.Context, js jetstream.KeyValueManager) (*Store, error) {
kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucket, 1)
kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucket, 1)
if err != nil {
return nil, fmt.Errorf("eventsource: open bucket: %w", err)
}
+2 -2
View File
@@ -13,7 +13,7 @@ import (
"regexp"
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -132,7 +132,7 @@ type Store struct {
// Open opens (or creates) the execution bucket.
func Open(ctx context.Context, js jetstream.KeyValueManager) (*Store, error) {
kv, err := kvutil.GetOrCreateBucket(ctx, js, KVBucket, 1)
kv, err := kvstore.GetOrCreateBucket(ctx, js, KVBucket, 1)
if err != nil {
return nil, fmt.Errorf("execution: open bucket: %w", err)
}
@@ -20,10 +20,6 @@ type Client interface {
CodeStored(ctx context.Context, accountID, region, name string, revision uint64, sha256Hex string, size int64) (Record, uint64, error)
// RollbackCreate frees the name while the record is still storing at revision.
RollbackCreate(ctx context.Context, accountID, region, name string, revision uint64) error
// BeginCodeUpdate records an owed upload of codeRevisionID for the function at revision.
BeginCodeUpdate(ctx context.Context, accountID, region string, rec Record, revision uint64, codeRevisionID string) error
// CodeUpdated commits the stored copy and returns the updated record and revision.
CodeUpdated(ctx context.Context, accountID, region string, rec Record, revision uint64, codeRevisionID, sha256Hex string, size int64) (Record, uint64, error)
}
// NATSClient is the Client over NATS request/reply to the daemon.
@@ -69,20 +65,3 @@ func (c *NATSClient) RollbackCreate(ctx context.Context, accountID, region, name
&RollbackCreateRequest{Region: region, FunctionName: name, Revision: revision}, metaRequestTimeout, accountID)
return err
}
func (c *NATSClient) BeginCodeUpdate(ctx context.Context, accountID, region string, rec Record, revision uint64, codeRevisionID string) error {
_, err := natsmsg.NATSRequest[CodeUpdateBeginResponse](ctx, c.conn, SubjectCodeUpdateBegin,
&CodeUpdateBeginRequest{Region: region, FunctionName: rec.FunctionName, FunctionID: rec.FunctionID, Revision: revision, RevisionID: codeRevisionID},
metaRequestTimeout, accountID)
return err
}
func (c *NATSClient) CodeUpdated(ctx context.Context, accountID, region string, rec Record, revision uint64, codeRevisionID, sha256Hex string, size int64) (Record, uint64, error) {
resp, err := natsmsg.NATSRequest[CodeUpdatedResponse](ctx, c.conn, SubjectCodeUpdated,
&CodeUpdatedRequest{Region: region, FunctionName: rec.FunctionName, FunctionID: rec.FunctionID, Revision: revision,
RevisionID: codeRevisionID, SHA256: sha256Hex, Size: size}, metaRequestTimeout, accountID)
if err != nil {
return Record{}, 0, err
}
return resp.Record, resp.Revision, nil
}
+2 -2
View File
@@ -11,7 +11,7 @@ import (
"time"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -60,7 +60,7 @@ func (s *MetaService) bucket(ctx context.Context, accountID string) (jetstream.K
return kv, nil
}
name := KVAccountBucket(accountID)
kv, err := kvutil.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
kv, err := kvstore.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
if err != nil {
return nil, fmt.Errorf("lambda: open account bucket %s: %w", name, err)
}
+4 -36
View File
@@ -103,12 +103,10 @@ func KVFunctionKey(region, name string) string { return kvFunctionKeyPrefix + re
// NATS subjects served by the daemon's metadata service.
const (
SubjectCreate = "lambda.function.create"
SubjectGet = "lambda.function.get"
SubjectCodeStored = "lambda.function.code-stored"
SubjectRollbackCreate = "lambda.function.rollback-create"
SubjectCodeUpdateBegin = "lambda.function.code-update-begin"
SubjectCodeUpdated = "lambda.function.code-updated"
SubjectCreate = "lambda.function.create"
SubjectGet = "lambda.function.get"
SubjectCodeStored = "lambda.function.code-stored"
SubjectRollbackCreate = "lambda.function.rollback-create"
)
// CreateRequest carries the record the gateway built; the daemon owns the
@@ -160,33 +158,3 @@ type RollbackCreateRequest struct {
// RollbackCreateResponse is empty: success means the name is free.
type RollbackCreateResponse struct{}
// CodeUpdateBeginRequest asks for an owed upload of new code for the function
// as it stood at Revision; RevisionID names the new code copy.
type CodeUpdateBeginRequest struct {
Region string `json:"region"`
FunctionName string `json:"functionName"`
FunctionID string `json:"functionId"`
Revision uint64 `json:"revision"`
RevisionID string `json:"revisionId"`
}
// CodeUpdateBeginResponse is empty: success means the upload is owed.
type CodeUpdateBeginResponse struct{}
// CodeUpdatedRequest names the stored new copy for the function at Revision.
type CodeUpdatedRequest struct {
Region string `json:"region"`
FunctionName string `json:"functionName"`
FunctionID string `json:"functionId"`
Revision uint64 `json:"revision"`
RevisionID string `json:"revisionId"`
SHA256 string `json:"sha256"`
Size int64 `json:"size"`
}
// CodeUpdatedResponse returns the committed record and its revision.
type CodeUpdatedResponse struct {
Record Record `json:"record"`
Revision uint64 `json:"revision"`
}
-121
View File
@@ -1,121 +0,0 @@
package function
import (
"context"
"encoding/json"
"errors"
"fmt"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/nats-io/nats.go/jetstream"
)
// readServing loads the record a code update names and checks it is still the
// one the caller read: same revision and function, Active, with stored code.
func readServing(ctx context.Context, kv jetstream.KeyValue, region, name, functionID string, revision uint64) (Record, error) {
entry, err := kv.Get(ctx, KVFunctionKey(region, name))
if errors.Is(err, jetstream.ErrKeyNotFound) {
return Record{}, awserrors.Errorf(awserrors.ErrorResourceNotFoundException, "Function not found: %s", name)
}
if err != nil {
return Record{}, fmt.Errorf("lambda: get function record: %w", err)
}
if entry.Revision() != revision {
return Record{}, conflict("Function was modified by another request: %s", name)
}
rec, err := decode(entry.Value())
if err != nil {
return Record{}, err
}
if rec.FunctionID != functionID {
return Record{}, conflict("Function was modified by another request: %s", name)
}
if rec.State != StateActive || rec.Code.State != CodeStored {
return Record{}, conflict("The operation cannot be performed at this time. The function is currently in the following state: %s", rec.State)
}
return rec, nil
}
func validUpdate(region, name, functionID, revisionID, accountID string) error {
if !regionRE.MatchString(region) {
return badRegion()
}
if !ValidName(name) || !idRE.MatchString(functionID) || !idRE.MatchString(revisionID) || accountID == "" {
return awserrors.Errorf(awserrors.ErrorInvalidParameterValueException, "code update request is incomplete")
}
return nil
}
// BeginCodeUpdate records an owed upload for a new code copy of the function
// at the caller's revision. The function record is left unchanged, so invokes
// keep running the current copy while the new one is written.
func (s *MetaService) BeginCodeUpdate(ctx context.Context, req *CodeUpdateBeginRequest, accountID string) (*CodeUpdateBeginResponse, error) {
if err := validUpdate(req.Region, req.FunctionName, req.FunctionID, req.RevisionID, accountID); err != nil {
return nil, err
}
kv, err := s.bucket(ctx, accountID)
if err != nil {
return nil, err
}
rec, err := readServing(ctx, kv, req.Region, req.FunctionName, req.FunctionID, req.Revision)
if err != nil {
return nil, err
}
if rec.Code.RevisionID == req.RevisionID {
return nil, conflict("Function code revision is already in use: %s", req.FunctionName)
}
data, err := json.Marshal(Upload{
FormatVersion: FormatVersion, AccountID: accountID, Region: req.Region, FunctionName: req.FunctionName,
FunctionID: req.FunctionID, RevisionID: req.RevisionID, State: UploadOwed, CreatedAt: s.now().UTC(),
})
if err != nil {
return nil, err
}
if _, err := kv.Create(ctx, KVUploadKey(req.Region, req.FunctionID, req.RevisionID), data); err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
return nil, conflict("Function code revision is already in use: %s", req.FunctionName)
}
return nil, fmt.Errorf("lambda: record code upload: %w", err)
}
return &CodeUpdateBeginResponse{}, nil
}
// CodeUpdated points the function at its new stored copy by compare-and-swap at
// the revision the update began from; only the code and last-modified time
// change. The previous copy and its upload record are kept.
func (s *MetaService) CodeUpdated(ctx context.Context, req *CodeUpdatedRequest, accountID string) (*CodeUpdatedResponse, error) {
if err := validUpdate(req.Region, req.FunctionName, req.FunctionID, req.RevisionID, accountID); err != nil {
return nil, err
}
if !digestRE.MatchString(req.SHA256) || req.Size <= 0 {
return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValueException, "code digest is invalid")
}
kv, err := s.bucket(ctx, accountID)
if err != nil {
return nil, err
}
rec, err := readServing(ctx, kv, req.Region, req.FunctionName, req.FunctionID, req.Revision)
if err != nil {
return nil, err
}
next := rec
next.Code = Code{RevisionID: req.RevisionID, SHA256: req.SHA256, Size: req.Size, State: CodeStored}
up, upRev, err := checkUpload(ctx, kv, next)
if err != nil {
return nil, err
}
next.LastModified = s.now().UTC()
data, err := json.Marshal(next)
if err != nil {
return nil, err
}
rev, err := kv.Update(ctx, KVFunctionKey(req.Region, req.FunctionName), data, req.Revision)
if err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
return nil, conflict("Function was modified by another request: %s", req.FunctionName)
}
return nil, fmt.Errorf("lambda: update function record: %w", err)
}
attach(ctx, kv, up, upRev, rev)
return &CodeUpdatedResponse{Record: next, Revision: rev}, nil
}
@@ -1,211 +0,0 @@
package function
import (
"context"
"encoding/json"
"errors"
"sync/atomic"
"testing"
"time"
"github.com/mulgadc/spinifex/internal/testkit"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const updateDigest = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
// mustServe creates "hello" with stored code and activates it.
func mustServe(t *testing.T, svc *MetaService) *GetResponse {
t.Helper()
rev := mustCreate(t, svc, "hello", "f1")
_, err := svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "hello", Revision: rev, SHA256: testDigest, Size: 9}, testAccount)
require.NoError(t, err)
svc.SetReadiness(func(context.Context, Record) error { return nil })
require.NoError(t, svc.ActivatePending(t.Context()))
got := mustGet(t, svc, "hello")
require.Equal(t, StateActive, got.Record.State)
return got
}
func beginReq(revision uint64, codeRevisionID string) *CodeUpdateBeginRequest {
return &CodeUpdateBeginRequest{Region: testRegion, FunctionName: "hello", FunctionID: "f1", Revision: revision, RevisionID: codeRevisionID}
}
func commitReq(revision uint64, codeRevisionID string) *CodeUpdatedRequest {
return &CodeUpdatedRequest{Region: testRegion, FunctionName: "hello", FunctionID: "f1", Revision: revision,
RevisionID: codeRevisionID, SHA256: updateDigest, Size: 12}
}
func uploadOf(t *testing.T, svc *MetaService, codeRevisionID string) (Upload, bool) {
t.Helper()
kv, err := svc.bucket(t.Context(), testAccount)
require.NoError(t, err)
entry, err := kv.Get(t.Context(), KVUploadKey(testRegion, "f1", codeRevisionID))
if errors.Is(err, jetstream.ErrKeyNotFound) {
return Upload{}, false
}
require.NoError(t, err)
var up Upload
require.NoError(t, json.Unmarshal(entry.Value(), &up))
return up, true
}
func TestCodeUpdate_CommitChangesOnlyTheCodeReference(t *testing.T) {
svc, c := newTestService(t)
before := mustServe(t, svc)
c.t = c.t.Add(time.Minute)
_, err := svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, "rev-2"), testAccount)
require.NoError(t, err)
assert.Equal(t, before.Revision, mustGet(t, svc, "hello").Revision, "beginning an update leaves the record as it was")
up, ok := uploadOf(t, svc, "rev-2")
require.True(t, ok)
assert.Equal(t, UploadOwed, up.State)
resp, err := svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-2"), testAccount)
require.NoError(t, err)
got := mustGet(t, svc, "hello")
assert.Equal(t, resp.Revision, got.Revision)
assert.Equal(t, Code{RevisionID: "rev-2", SHA256: updateDigest, Size: 12, State: CodeStored}, got.Record.Code)
assert.Equal(t, c.t, got.Record.LastModified)
want := before.Record
want.Code, want.LastModified = got.Record.Code, got.Record.LastModified
assert.Equal(t, want, got.Record, "identity, configuration and state are unchanged")
up, _ = uploadOf(t, svc, "rev-2")
assert.Equal(t, UploadAttached, up.State)
assert.Equal(t, resp.Revision, up.AttachedRevision)
old, ok := uploadOf(t, svc, "rev-f1")
require.True(t, ok, "the previous copy stays tracked")
assert.Equal(t, UploadAttached, old.State)
}
func TestCodeUpdate_OnlyOneOfTwoUpdatesFromTheSameRevisionCommits(t *testing.T) {
svc, _ := newTestService(t)
before := mustServe(t, svc)
for _, id := range []string{"rev-a", "rev-b"} {
_, err := svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, id), testAccount)
require.NoError(t, err)
}
_, err := svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-b"), testAccount)
require.NoError(t, err)
_, err = svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-a"), testAccount)
testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
assert.Equal(t, "rev-b", mustGet(t, svc, "hello").Record.Code.RevisionID)
loser, _ := uploadOf(t, svc, "rev-a")
assert.Equal(t, UploadOwed, loser.State, "the losing copy stays owed for cleanup")
_, err = svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, "rev-c"), testAccount)
testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
_, ok := uploadOf(t, svc, "rev-c")
assert.False(t, ok, "an update from a stale revision records nothing")
}
// Both commits read the same record before either writes: the loser is held
// between its checks and its conditional write while the winner commits.
func TestCodeUpdate_InterleavedCommitLosesTheConditionalWrite(t *testing.T) {
svc, c := newTestService(t)
before := mustServe(t, svc)
for _, id := range []string{"rev-a", "rev-b"} {
_, err := svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, id), testAccount)
require.NoError(t, err)
}
paused, release := make(chan struct{}), make(chan struct{})
var calls atomic.Int32
svc.now = func() time.Time {
if calls.Add(1) == 1 {
close(paused)
<-release
}
return c.now()
}
loserErr := make(chan error, 1)
go func() {
_, err := svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-a"), testAccount)
loserErr <- err
}()
<-paused
won, err := svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-b"), testAccount)
require.NoError(t, err)
close(release)
testutil.RequireAWSCode(t, <-loserErr, awserrors.ErrorResourceConflictException)
got := mustGet(t, svc, "hello")
assert.Equal(t, won.Revision, got.Revision)
assert.Equal(t, "rev-b", got.Record.Code.RevisionID)
winner, _ := uploadOf(t, svc, "rev-b")
assert.Equal(t, UploadAttached, winner.State)
assert.Equal(t, won.Revision, winner.AttachedRevision)
loser, ok := uploadOf(t, svc, "rev-a")
require.True(t, ok, "the losing copy stays tracked")
assert.Equal(t, UploadOwed, loser.State)
}
func TestCodeUpdate_CommitNeedsItsOwnOwedUpload(t *testing.T) {
svc, _ := newTestService(t)
before := mustServe(t, svc)
_, err := svc.CodeUpdated(t.Context(), commitReq(before.Revision, "rev-never-begun"), testAccount)
testutil.RequireAWSCode(t, err, awserrors.ErrorServiceException)
_, err = svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, "rev-f1"), testAccount)
testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
other := commitReq(before.Revision, "rev-2")
_, err = svc.BeginCodeUpdate(t.Context(), beginReq(before.Revision, "rev-2"), testAccount)
require.NoError(t, err)
other.FunctionID = "f-other"
_, err = svc.CodeUpdated(t.Context(), other, testAccount)
testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
assert.Equal(t, before.Revision, mustGet(t, svc, "hello").Revision)
}
func TestCodeUpdate_RefusesAFunctionThatIsNotServing(t *testing.T) {
svc, _ := newTestService(t)
rev := mustCreate(t, svc, "hello", "f1")
_, err := svc.BeginCodeUpdate(t.Context(), beginReq(rev, "rev-2"), testAccount)
testutil.RequireAWSCode(t, err, awserrors.ErrorResourceConflictException)
_, ok := uploadOf(t, svc, "rev-2")
assert.False(t, ok)
ghost := beginReq(1, "rev-2")
ghost.FunctionName = "ghost"
_, err = svc.BeginCodeUpdate(t.Context(), ghost, testAccount)
testutil.RequireAWSCode(t, err, awserrors.ErrorResourceNotFoundException)
}
func TestCodeUpdate_RefusesMalformedRequests(t *testing.T) {
svc, _ := newTestService(t)
before := mustServe(t, svc)
begins := map[string]func(*CodeUpdateBeginRequest){
"region": func(r *CodeUpdateBeginRequest) { r.Region = "Bad Region" },
"name": func(r *CodeUpdateBeginRequest) { r.FunctionName = "a/b" },
"function id": func(r *CodeUpdateBeginRequest) { r.FunctionID = "" },
"revision id": func(r *CodeUpdateBeginRequest) { r.RevisionID = "../x" },
}
for name, mutate := range begins {
req := beginReq(before.Revision, "rev-2")
mutate(req)
t.Run("begin "+name, func(t *testing.T) {
_, err := svc.BeginCodeUpdate(t.Context(), req, testAccount)
testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
})
}
commits := map[string]func(*CodeUpdatedRequest){
"digest": func(r *CodeUpdatedRequest) { r.SHA256 = "xyz" },
"size": func(r *CodeUpdatedRequest) { r.Size = 0 },
}
for name, mutate := range commits {
req := commitReq(before.Revision, "rev-2")
mutate(req)
t.Run("commit "+name, func(t *testing.T) {
_, err := svc.CodeUpdated(t.Context(), req, testAccount)
testutil.RequireAWSCode(t, err, awserrors.ErrorInvalidParameterValueException)
})
}
assert.Equal(t, before.Revision, mustGet(t, svc, "hello").Revision)
}
+3 -3
View File
@@ -10,7 +10,7 @@ import (
"strings"
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -49,7 +49,7 @@ type Store struct {
// NewStore creates/opens the per-AZ lease bucket and returns a Store. ctx bounds
// the bucket get-or-create only; each operation carries its own.
func NewStore(ctx context.Context, js jetstream.KeyValueManager, az string) (*Store, error) {
kv, err := kvutil.GetOrCreateBucket(ctx, js, BucketName(az), 5)
kv, err := kvstore.GetOrCreateBucket(ctx, js, BucketName(az), 5)
if err != nil {
return nil, fmt.Errorf("create dhcp lease KV bucket: %w", err)
}
@@ -152,7 +152,7 @@ func (s *Store) List(ctx context.Context) ([]Entry, error) {
}
var out []Entry
for _, k := range keys {
if k == kvutil.VersionKey {
if k == kvstore.VersionKey {
continue
}
e, err := s.Get(ctx, k)
@@ -14,7 +14,7 @@ import (
"github.com/mulgadc/spinifex/internal/testkit"
ec2igw "github.com/mulgadc/spinifex/spinifex/domains/ec2/igw"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -22,7 +22,7 @@ import (
// so a test can write to it and have the source find it.
func createIntentBucket(t *testing.T, js jetstream.JetStream, name string) jetstream.KeyValue {
t.Helper()
kv, err := kvutil.GetOrCreateBucket(t.Context(), js, name, 1)
kv, err := kvstore.GetOrCreateBucket(t.Context(), js, name, 1)
if err != nil {
t.Fatalf("create bucket %s: %v", name, err)
}
@@ -19,7 +19,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/domains/network/policy"
"github.com/mulgadc/spinifex/spinifex/domains/network/reconcile"
"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -360,11 +360,11 @@ func TestHostIngress_IntentReadFailureNeverPrunes(t *testing.T) {
func TestRunHostIngress_IntentWriteConvergesEveryNodeWithoutLeaseOrEvent(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
ctx := t.Context()
vpcs, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 1)
vpcs, err := kvstore.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 1)
if err != nil {
t.Fatalf("create VPC bucket: %v", err)
}
igws, err := kvutil.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 1)
igws, err := kvstore.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 1)
if err != nil {
t.Fatalf("create IGW bucket: %v", err)
}
+2 -2
View File
@@ -19,7 +19,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/domains/network/external"
"github.com/mulgadc/spinifex/spinifex/domains/network/policy"
"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/nats-io/nats.go/jetstream"
)
@@ -236,7 +236,7 @@ func matchesLocalAZ(vpcAZ, localAZ string) bool {
return vpcAZ == "" || vpcAZ == localAZ
}
func keyIsVersion(key string) bool { return key == kvutil.VersionKey }
func keyIsVersion(key string) bool { return key == kvstore.VersionKey }
func loadVPCs(ctx context.Context, js jetstream.JetStream, localAZ string, out map[string]topology.VPCSpec) (map[string]struct{}, error) {
localVPCs := make(map[string]struct{})
@@ -11,7 +11,7 @@ import (
"testing"
vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/nats-io/nats.go/jetstream"
)
@@ -20,7 +20,7 @@ import (
// creating the bucket if absent — mirroring how the VPC handlers persist state.
func seedKV(t *testing.T, js jetstream.JetStream, bucket, key string, rec any) {
t.Helper()
kv, err := kvutil.GetOrCreateBucket(t.Context(), js, bucket, 10)
kv, err := kvstore.GetOrCreateBucket(t.Context(), js, bucket, 10)
if err != nil {
t.Fatalf("GetOrCreateBucket %s: %v", bucket, err)
}
+1 -2
View File
@@ -10,7 +10,6 @@ import (
"time"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go/jetstream"
)
@@ -196,7 +195,7 @@ func (s *UsageStore) TokensThisPeriod(ctx context.Context, accountID string) (in
if err != nil {
return 0, err
}
keys, err := kvutil.Keys(ctx, kv)
keys, err := kvstore.Keys(ctx, kv)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return 0, nil
+1 -2
View File
@@ -13,7 +13,6 @@ import (
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvlease"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -244,7 +243,7 @@ func InitLeaderBucket(ctx context.Context, js jetstream.JetStream) (jetstream.Ke
// AccountBucketNames lists every RDS per-account bucket in the cluster. The reconciler and the DNS
// desired-set both need a cross-tenant view, which only this provides.
func AccountBucketNames(ctx context.Context, js jetstream.JetStream) ([]string, error) {
all, err := kvutil.BucketNames(ctx, js)
all, err := kvstore.BucketNames(ctx, js)
if err != nil {
return nil, err
}

Some files were not shown because too many files have changed in this diff Show More