docs(vultr): set up the cluster from node 1

docs/vultr-integration/README.md

@@ -265,7 +265,12 @@ If this returns a list of available instance types, your installation is working
 
 Spinifex is running, but it holds nothing yet: no machine images, no networks, no instances.
 
-Continue to [Setting Up Your Cluster](/docs/setting-up-your-cluster) to import an AMI, create an SSH key pair, create a VPC with a public subnet, and launch your first instance.
+Log in to node 1, and follow [Setting Up Your Cluster](/docs/setting-up-your-cluster) there to import an AMI, create an SSH key pair, create a VPC with a public subnet, and launch your first instance:
+
+```bash
+./deploy-vultr.sh ssh
+```
+
 An instance you launch into a public subnet gets a Vultr reserved IP, which you can see in the Vultr portal under **Network → Reserved IPs**.
 
 ## 8. Test resiliency

spinifex/domains/ecr/awsapi/gateway_image_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/mulgadc/spinifex/internal/testkit"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -26,8 +25,7 @@ import (
 // object store + memory meta) and its composed ECR JSON image-action service.
 func newImageGateway(t *testing.T) *ecrGW {
 	t.Helper()
-	_, _, js := testutil.StartTestJetStream(t)
-	meta := ecrdomain.NewKVMetaStore(js)
+	meta := ecrdomain.NewMemoryMetaStore()
 	reg := ecrregistry.NewRegistry(objectstore.NewMemoryObjectStore(), meta, ecrTestAccount)
 	gw := newECRGW(nil, awsapi.Deps{Registry: awsapi.NewRegistryActionService(reg, reg, reg, reg), LifecyclePreview: awsapi.NewLifecyclePreviewActionService(meta, reg)})
 	imageRegistries.Store(gw, reg)

spinifex/domains/ecr/meta.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -591,6 +593,295 @@ func (s *KVMetaStore) DeleteUpload(ctx context.Context, accountID, uploadID stri
 	return nil
 }
 
+// MemoryMetaStore is an in-memory MetaStore used by tests and single-process
+// dev runs. It is safe for concurrent use.
+type MemoryMetaStore struct {
+	mu        sync.Mutex
+	repos     map[string]map[string]RepoMeta     // account -> repo -> meta
+	policies  map[string]map[string][]byte       // account -> repo -> policyText
+	lifecycle map[string]map[string][]byte       // account -> repo -> lifecyclePolicyText
+	tags      map[string]map[string]string       // account -> repo|tag -> digest
+	manifests map[string]map[string]ManifestMeta // account -> repo|digest -> meta
+	uploads   map[string]map[string]uploadRev    // account -> id -> state+rev
+}
+
+type uploadRev struct {
+	state UploadState
+	rev   uint64
+}
+
+var _ MetaStore = (*MemoryMetaStore)(nil)
+
+// NewMemoryMetaStore constructs an empty in-memory MetaStore.
+func NewMemoryMetaStore() *MemoryMetaStore {
+	return &MemoryMetaStore{
+		repos:     make(map[string]map[string]RepoMeta),
+		policies:  make(map[string]map[string][]byte),
+		lifecycle: make(map[string]map[string][]byte),
+		tags:      make(map[string]map[string]string),
+		manifests: make(map[string]map[string]ManifestMeta),
+		uploads:   make(map[string]map[string]uploadRev),
+	}
+}
+
+func (m *MemoryMetaStore) PutRepo(_ context.Context, accountID string, meta RepoMeta) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.repos[accountID] == nil {
+		m.repos[accountID] = make(map[string]RepoMeta)
+	}
+	m.repos[accountID][meta.Name] = meta
+	return nil
+}
+
+func (m *MemoryMetaStore) GetRepo(_ context.Context, accountID, repo string) (RepoMeta, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	r, ok := m.repos[accountID][repo]
+	if !ok {
+		return RepoMeta{}, ErrNotFound
+	}
+	return r, nil
+}
+
+func (m *MemoryMetaStore) ListRepos(_ context.Context, accountID string) ([]string, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return slices.Sorted(maps.Keys(m.repos[accountID])), nil
+}
+
+func (m *MemoryMetaStore) DeleteRepo(_ context.Context, accountID, repo string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if _, ok := m.repos[accountID][repo]; !ok {
+		return ErrNotFound
+	}
+	delete(m.repos[accountID], repo)
+	delete(m.policies[accountID], repo)
+	delete(m.lifecycle[accountID], repo)
+	for k := range m.tags[accountID] {
+		if r, _, ok := strings.Cut(k, "|"); ok && r == repo {
+			delete(m.tags[accountID], k)
+		}
+	}
+	for k := range m.manifests[accountID] {
+		if r, _, ok := strings.Cut(k, "|"); ok && r == repo {
+			delete(m.manifests[accountID], k)
+		}
+	}
+	return nil
+}
+
+func (m *MemoryMetaStore) ListManifests(_ context.Context, accountID, repo string) ([]string, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var out []string
+	for k := range m.manifests[accountID] {
+		if r, d, ok := strings.Cut(k, "|"); ok && r == repo {
+			out = append(out, d)
+		}
+	}
+	sort.Strings(out)
+	return out, nil
+}
+
+func (m *MemoryMetaStore) PutRepoPolicy(_ context.Context, accountID, repo string, policyText []byte) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.policies[accountID] == nil {
+		m.policies[accountID] = make(map[string][]byte)
+	}
+	m.policies[accountID][repo] = policyText
+	return nil
+}
+
+func (m *MemoryMetaStore) GetRepoPolicy(_ context.Context, accountID, repo string) ([]byte, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	p, ok := m.policies[accountID][repo]
+	if !ok {
+		return nil, ErrNotFound
+	}
+	return p, nil
+}
+
+func (m *MemoryMetaStore) DeleteRepoPolicy(_ context.Context, accountID, repo string) ([]byte, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	p, ok := m.policies[accountID][repo]
+	if !ok {
+		return nil, ErrNotFound
+	}
+	delete(m.policies[accountID], repo)
+	return p, nil
+}
+
+func (m *MemoryMetaStore) PutLifecyclePolicy(_ context.Context, accountID, repo string, policyText []byte) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.lifecycle[accountID] == nil {
+		m.lifecycle[accountID] = make(map[string][]byte)
+	}
+	m.lifecycle[accountID][repo] = policyText
+	return nil
+}
+
+func (m *MemoryMetaStore) GetLifecyclePolicy(_ context.Context, accountID, repo string) ([]byte, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	p, ok := m.lifecycle[accountID][repo]
+	if !ok {
+		return nil, ErrNotFound
+	}
+	return p, nil
+}
+
+func (m *MemoryMetaStore) DeleteLifecyclePolicy(_ context.Context, accountID, repo string) ([]byte, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	p, ok := m.lifecycle[accountID][repo]
+	if !ok {
+		return nil, ErrNotFound
+	}
+	delete(m.lifecycle[accountID], repo)
+	return p, nil
+}
+
+func (m *MemoryMetaStore) PutTag(_ context.Context, accountID, repo, tag, digest string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.tags[accountID] == nil {
+		m.tags[accountID] = make(map[string]string)
+	}
+	m.tags[accountID][repo+"|"+tag] = digest
+	return nil
+}
+
+func (m *MemoryMetaStore) GetTag(_ context.Context, accountID, repo, tag string) (string, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	d, ok := m.tags[accountID][repo+"|"+tag]
+	if !ok {
+		return "", ErrNotFound
+	}
+	return d, nil
+}
+
+func (m *MemoryMetaStore) DeleteTag(_ context.Context, accountID, repo, tag string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	key := repo + "|" + tag
+	if _, ok := m.tags[accountID][key]; !ok {
+		return ErrNotFound
+	}
+	delete(m.tags[accountID], key)
+	return nil
+}
+
+func (m *MemoryMetaStore) ListTags(_ context.Context, accountID, repo string) ([]string, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var out []string
+	for k := range m.tags[accountID] {
+		if r, t, ok := strings.Cut(k, "|"); ok && r == repo {
+			out = append(out, t)
+		}
+	}
+	sort.Strings(out)
+	return out, nil
+}
+
+func (m *MemoryMetaStore) PutManifestMeta(_ context.Context, accountID, repo string, meta ManifestMeta) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.manifests[accountID] == nil {
+		m.manifests[accountID] = make(map[string]ManifestMeta)
+	}
+	m.manifests[accountID][repo+"|"+meta.Digest] = meta
+	return nil
+}
+
+func (m *MemoryMetaStore) GetManifestMeta(_ context.Context, accountID, repo, digest string) (ManifestMeta, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	meta, ok := m.manifests[accountID][repo+"|"+digest]
+	if !ok {
+		return ManifestMeta{}, ErrNotFound
+	}
+	return meta, nil
+}
+
+func (m *MemoryMetaStore) DeleteManifestMeta(_ context.Context, accountID, repo, digest string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	key := repo + "|" + digest
+	if _, ok := m.manifests[accountID][key]; !ok {
+		return ErrNotFound
+	}
+	delete(m.manifests[accountID], key)
+	return nil
+}
+
+func (m *MemoryMetaStore) ListReferrers(_ context.Context, accountID, repo, subjectDigest string) ([]ReferrerDescriptor, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	out := []ReferrerDescriptor{}
+	for k, meta := range m.manifests[accountID] {
+		if r, _, ok := strings.Cut(k, "|"); ok && r == repo && meta.SubjectDigest == subjectDigest {
+			out = append(out, meta.referrerDescriptor())
+		}
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Digest < out[j].Digest })
+	return out, nil
+}
+
+func (m *MemoryMetaStore) PutUpload(_ context.Context, accountID, uploadID string, state UploadState) (uint64, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.uploads[accountID] == nil {
+		m.uploads[accountID] = make(map[string]uploadRev)
+	}
+	rev := m.uploads[accountID][uploadID].rev + 1
+	m.uploads[accountID][uploadID] = uploadRev{state: state, rev: rev}
+	return rev, nil
+}
+
+func (m *MemoryMetaStore) GetUpload(_ context.Context, accountID, uploadID string) (UploadState, uint64, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	u, ok := m.uploads[accountID][uploadID]
+	if !ok {
+		return UploadState{}, 0, ErrNotFound
+	}
+	return u.state, u.rev, nil
+}
+
+func (m *MemoryMetaStore) UpdateUpload(_ context.Context, accountID, uploadID string, state UploadState, rev uint64) (uint64, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	u, ok := m.uploads[accountID][uploadID]
+	if !ok {
+		return 0, ErrNotFound
+	}
+	if u.rev != rev {
+		return 0, ErrConflict
+	}
+	u.state = state
+	u.rev = rev + 1
+	m.uploads[accountID][uploadID] = u
+	return u.rev, nil
+}
+
+func (m *MemoryMetaStore) DeleteUpload(_ context.Context, accountID, uploadID string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if _, ok := m.uploads[accountID][uploadID]; !ok {
+		return ErrNotFound
+	}
+	delete(m.uploads[accountID], uploadID)
+	return nil
+}
+
 // mapKVErr normalizes JetStream KV errors to the MetaStore error vocabulary.
 func mapKVErr(err error) error {
 	if errors.Is(err, jetstream.ErrKeyNotFound) {

spinifex/domains/ecr/meta_test.go

@@ -8,7 +8,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -62,8 +61,8 @@ func TestKVKeyHelpers(t *testing.T) {
 	assert.Equal(t, "sha256-abc", DigestToken("sha256:abc"))
 }
 
-func TestKVMetaStore_RepoRoundtrip(t *testing.T) {
-	m := newTestKVMetaStore(t)
+func TestMemoryMetaStore_RepoRoundtrip(t *testing.T) {
+	m := NewMemoryMetaStore()
 	const acct = "000000000000"
 
 	_, err := m.GetRepo(context.Background(), acct, "team/app")
@@ -81,8 +80,8 @@ func TestKVMetaStore_RepoRoundtrip(t *testing.T) {
 	assert.Equal(t, []string{"team/app", "team/web"}, repos)
 }
 
-func TestKVMetaStore_TagsAndManifests(t *testing.T) {
-	m := newTestKVMetaStore(t)
+func TestMemoryMetaStore_TagsAndManifests(t *testing.T) {
+	m := NewMemoryMetaStore()
 	const acct = "000000000000"
 
 	require.NoError(t, m.PutTag(context.Background(), acct, "r", "v1", "sha256:aaa"))
@@ -113,8 +112,8 @@ func TestKVMetaStore_TagsAndManifests(t *testing.T) {
 	assert.ErrorIs(t, m.DeleteManifestMeta(context.Background(), acct, "r", "sha256:ccc"), ErrNotFound)
 }
 
-func TestKVMetaStore_UploadCAS(t *testing.T) {
-	m := newTestKVMetaStore(t)
+func TestMemoryMetaStore_UploadCAS(t *testing.T) {
+	m := NewMemoryMetaStore()
 	const acct = "000000000000"
 
 	rev, err := m.PutUpload(context.Background(), acct, "u1", UploadState{RepoName: "r"})
@@ -144,8 +143,8 @@ func TestTrimSuffixMeta(t *testing.T) {
 	assert.Equal(t, "team/app", trimSuffixMeta("repos/team/app/meta"))
 }
 
-func TestKVMetaStore_DeleteRepoAndListManifests(t *testing.T) {
-	m := newTestKVMetaStore(t)
+func TestMemoryMetaStore_DeleteRepoAndListManifests(t *testing.T) {
+	m := NewMemoryMetaStore()
 	const acct = "000000000000"
 
 	require.NoError(t, m.PutRepo(context.Background(), acct, RepoMeta{Name: "team/app", CreatedAt: time.Now()}))
@@ -157,7 +156,7 @@ func TestKVMetaStore_DeleteRepoAndListManifests(t *testing.T) {
 
 	digs, err := m.ListManifests(context.Background(), acct, "team/app")
 	require.NoError(t, err)
-	assert.Equal(t, []string{DigestToken("sha256:ccc")}, digs)
+	assert.Equal(t, []string{"sha256:ccc"}, digs)
 
 	require.NoError(t, m.DeleteRepo(context.Background(), acct, "team/app"))
 	_, err = m.GetRepo(context.Background(), acct, "team/app")
@@ -181,6 +180,50 @@ func TestKVMetaStore_DeleteRepoAndListManifests(t *testing.T) {
 	assert.ErrorIs(t, m.DeleteRepo(context.Background(), acct, "team/ghost"), ErrNotFound)
 }
 
+func TestMemoryMetaStore_RepoPolicy(t *testing.T) {
+	m := NewMemoryMetaStore()
+	const acct = "000000000000"
+	const policy = `{"Version":"2012-10-17"}`
+
+	_, err := m.GetRepoPolicy(context.Background(), acct, "team/app")
+	assert.ErrorIs(t, err, ErrNotFound)
+	_, err = m.DeleteRepoPolicy(context.Background(), acct, "team/app")
+	assert.ErrorIs(t, err, ErrNotFound)
+
+	require.NoError(t, m.PutRepoPolicy(context.Background(), acct, "team/app", []byte(policy)))
+	got, err := m.GetRepoPolicy(context.Background(), acct, "team/app")
+	require.NoError(t, err)
+	assert.Equal(t, policy, string(got))
+
+	deleted, err := m.DeleteRepoPolicy(context.Background(), acct, "team/app")
+	require.NoError(t, err)
+	assert.Equal(t, policy, string(deleted))
+	_, err = m.GetRepoPolicy(context.Background(), acct, "team/app")
+	assert.ErrorIs(t, err, ErrNotFound)
+}
+
+func TestMemoryMetaStore_LifecyclePolicy(t *testing.T) {
+	m := NewMemoryMetaStore()
+	const acct = "000000000000"
+	const policy = `{"rules":[]}`
+
+	_, err := m.GetLifecyclePolicy(context.Background(), acct, "team/app")
+	assert.ErrorIs(t, err, ErrNotFound)
+	_, err = m.DeleteLifecyclePolicy(context.Background(), acct, "team/app")
+	assert.ErrorIs(t, err, ErrNotFound)
+
+	require.NoError(t, m.PutLifecyclePolicy(context.Background(), acct, "team/app", []byte(policy)))
+	got, err := m.GetLifecyclePolicy(context.Background(), acct, "team/app")
+	require.NoError(t, err)
+	assert.Equal(t, policy, string(got))
+
+	deleted, err := m.DeleteLifecyclePolicy(context.Background(), acct, "team/app")
+	require.NoError(t, err)
+	assert.Equal(t, policy, string(deleted))
+	_, err = m.GetLifecyclePolicy(context.Background(), acct, "team/app")
+	assert.ErrorIs(t, err, ErrNotFound)
+}
+
 func TestManifestMeta_DecodesRecordWithoutReferrerFields(t *testing.T) {
 	var m ManifestMeta
 	require.NoError(t, json.Unmarshal([]byte(`{"digest":"sha256:ccc","mediaType":"application/vnd.oci.image.manifest.v1+json","size":10,"pushedAt":"2026-01-01T00:00:00Z","childDigests":["sha256:ddd"]}`), &m))
@@ -189,9 +232,3 @@ func TestManifestMeta_DecodesRecordWithoutReferrerFields(t *testing.T) {
 	assert.Empty(t, m.SubjectDigest)
 	assert.Empty(t, m.ArtifactType)
 }
-
-func newTestKVMetaStore(t *testing.T) *KVMetaStore {
-	t.Helper()
-	_, _, js := testutil.StartTestJetStream(t)
-	return NewKVMetaStore(js)
-}

spinifex/domains/ecr/referrers_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"testing"
 
+	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -29,7 +30,11 @@ func referrerMeta(digest, subject string) ManifestMeta {
 // its own referrers state.
 func referrerStores(t *testing.T) map[string]MetaStore {
 	t.Helper()
-	return map[string]MetaStore{"kv": newTestKVMetaStore(t)}
+	_, nc, _ := testutil.StartTestJetStream(t)
+	return map[string]MetaStore{
+		"memory": NewMemoryMetaStore(),
+		"kv":     NewKVMetaStore(testutil.NewJetStream(t, nc)),
+	}
 }
 
 func TestListReferrers(t *testing.T) {

spinifex/domains/ecr/registry/lifecycle_sweep_test.go

@@ -50,7 +50,7 @@ func putPolicy(t *testing.T, reg *Registry, account, repo, policy string) {
 const untaggedExpirePolicy = `{"rules":[{"rulePriority":1,"selection":{"tagStatus":"untagged","countType":"sinceImagePushed","countUnit":"days","countNumber":7},"action":{"type":"expire"}}]}`
 
 func TestSweepRepo_SinceImagePushed(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:olduntagged", now.AddDate(0, 0, -10))
@@ -65,7 +65,7 @@ func TestSweepRepo_SinceImagePushed(t *testing.T) {
 }
 
 func TestSweepRepo_ImageCountMoreThan(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:a", now.AddDate(0, 0, -1), "a")
@@ -81,7 +81,7 @@ func TestSweepRepo_ImageCountMoreThan(t *testing.T) {
 }
 
 func TestSweepRepo_NoPolicy(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:x", now.AddDate(0, 0, -100))
@@ -92,7 +92,7 @@ func TestSweepRepo_NoPolicy(t *testing.T) {
 }
 
 func TestSweepRepo_Idempotent(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:olduntagged", now.AddDate(0, 0, -10))
@@ -105,7 +105,7 @@ func TestSweepRepo_Idempotent(t *testing.T) {
 }
 
 func TestSweepOnce_MultiAccount(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	now := time.Now().UTC()
 	const acct2 = "000000000001"
 	policy := `{"rules":[{"rulePriority":1,"selection":{"tagStatus":"any","countType":"imageCountMoreThan","countNumber":1},"action":{"type":"expire"}}]}`
@@ -125,7 +125,7 @@ func TestSweepOnce_MultiAccount(t *testing.T) {
 }
 
 func TestSweepOnce_AccountsError(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	s := NewLifecycleSweeper(reg, func() ([]string, error) {
 		return nil, assert.AnError
 	}, time.Hour)
@@ -133,7 +133,7 @@ func TestSweepOnce_AccountsError(t *testing.T) {
 }
 
 func TestNewLifecycleSweeper_DefaultInterval(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	s := NewLifecycleSweeper(reg, func() ([]string, error) { return nil, nil }, 0)
 	assert.Equal(t, DefaultLifecycleSweepInterval, s.interval)
 
@@ -147,7 +147,7 @@ func TestNewLifecycleSweeper_DefaultInterval(t *testing.T) {
 // TestSweeperRun drives the ticker loop: with a tiny interval and an expiring
 // image, the first tick must delete it; cancelling the context stops Run.
 func TestSweeperRun(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:olduntagged", now.AddDate(0, 0, -10))
@@ -184,7 +184,7 @@ func TestSweeperRun(t *testing.T) {
 // TestSweepRepo_InvalidStoredPolicy exercises the evaluate-error branch: a
 // malformed stored policy is logged and skipped, deleting nothing.
 func TestSweepRepo_InvalidStoredPolicy(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	now := time.Now().UTC()
 	repo := "team/app"
 	seedTimedImage(t, reg, testAccount, repo, "sha256:x", now.AddDate(0, 0, -100))
@@ -198,7 +198,7 @@ func TestSweepRepo_InvalidStoredPolicy(t *testing.T) {
 // TestSweepRepo_PolicyWithoutRepo exercises the ListImages-not-found branch: a
 // policy stored for a repo with no manifests yields nothing to expire.
 func TestSweepRepo_PolicyWithoutRepo(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	now := time.Now().UTC()
 	putPolicy(t, reg, testAccount, "ghost", untaggedExpirePolicy)
 

spinifex/domains/ecr/registry/registry_ensurebucket_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"testing"
 
+	"github.com/mulgadc/spinifex/spinifex/domains/ecr"
 	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -31,7 +32,7 @@ func (c *countingStore) EnsureBucket(ctx context.Context, bucket string) error {
 
 func TestRegistry_EnsureBucketFailureReturns500(t *testing.T) {
 	store := &countingStore{MemoryObjectStore: objectstore.NewMemoryObjectStore(), ensureErr: errors.New("predastore down")}
-	reg := NewRegistry(store, newTestMetaStore(t), testAccount)
+	reg := NewRegistry(store, ecr.NewMemoryMetaStore(), testAccount)
 
 	w := do(reg, http.MethodGet, "/v2/myrepo/tags/list", nil, nil)
 	assert.Equal(t, http.StatusInternalServerError, w.Code)
@@ -39,7 +40,7 @@ func TestRegistry_EnsureBucketFailureReturns500(t *testing.T) {
 
 func TestRegistry_EnsureBucketCachedAfterSuccess(t *testing.T) {
 	store := &countingStore{MemoryObjectStore: objectstore.NewMemoryObjectStore()}
-	reg := NewRegistry(store, newTestMetaStore(t), testAccount)
+	reg := NewRegistry(store, ecr.NewMemoryMetaStore(), testAccount)
 
 	do(reg, http.MethodGet, "/v2/myrepo/tags/list", nil, nil)
 	do(reg, http.MethodGet, "/v2/other/tags/list", nil, nil)

spinifex/domains/ecr/registry/registry_image_test.go

@@ -44,7 +44,7 @@ func storeLayersManifest(t *testing.T, reg *Registry, repo, tag string, layers .
 }
 
 func TestDeleteImage_ReclaimsBlobsKeepsShared(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	shared := pushBlob(t, reg, repo, []byte("shared-layer-bytes"))
 	exclusive := pushBlob(t, reg, repo, []byte("exclusive-layer-bytes"))
@@ -85,7 +85,7 @@ func seedImage(t *testing.T, reg *Registry, repo, tag string) ([]byte, string) {
 }
 
 func TestStoreAndGetManifest_RoundTrip(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	manifest, digest := seedImage(t, reg, "team/app", "v1")
 
 	// Fetch by tag.
@@ -114,7 +114,7 @@ func TestStoreAndGetManifest_RoundTrip(t *testing.T) {
 }
 
 func TestListImages_TaggedAndUntagged(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	seedImage(t, reg, repo, "v1")
 	seedImage(t, reg, repo, "v2")
@@ -176,7 +176,7 @@ func TestListImages_KVDigestForm(t *testing.T) {
 }
 
 func TestDeleteImage_ByTagAndByDigest(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	_, digest := seedImage(t, reg, repo, "v1")
 	seedImage(t, reg, repo, "v2") // shares nothing; distinct digest
@@ -205,7 +205,7 @@ func TestDeleteImage_ByTagAndByDigest(t *testing.T) {
 }
 
 func TestDeleteImage_ByDigestClearsTags(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	_, digest := seedImage(t, reg, repo, "v1")
 	// Add a second tag pointing at the same digest.
@@ -221,7 +221,7 @@ func TestDeleteImage_ByDigestClearsTags(t *testing.T) {
 }
 
 func TestStoreManifest_ImmutableTag(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	manA := fmt.Appendf(nil, `{"schemaVersion":2,"mediaType":"%s","layers":[{"digest":"%s"}]}`,
 		mediaTypeDockerManifest, pushBlob(t, reg, repo, []byte("layer-a")))
@@ -254,7 +254,7 @@ func TestStoreManifest_ImmutableTag(t *testing.T) {
 }
 
 func TestStoreManifest_BadDigestReference(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	cfgDg := pushBlob(t, reg, repo, []byte("cfg"))
 	manifest := fmt.Appendf(nil,

spinifex/domains/ecr/registry/registry_referrer_test.go

@@ -38,7 +38,7 @@ func storedMeta(t *testing.T, reg *Registry, body string) (subject, artifactType
 }
 
 func TestPutManifest_SubjectRecordedAndAdvertised(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	cfg := pushBlob(t, reg, referrerRepo, []byte("cfg"))
 
 	tests := []struct {
@@ -65,7 +65,7 @@ func TestPutManifest_SubjectRecordedAndAdvertised(t *testing.T) {
 }
 
 func TestPutManifest_IndexSubject(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	seedRepo(t, reg, referrerRepo)
 
 	body := fmt.Sprintf(`{"schemaVersion":2,"mediaType":%q,"manifests":[],"subject":{"mediaType":%q,"digest":%q,"size":10}}`,
@@ -81,7 +81,7 @@ func TestPutManifest_IndexSubject(t *testing.T) {
 }
 
 func TestPutManifest_NoSubjectLeavesFieldsEmpty(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	cfg := pushBlob(t, reg, referrerRepo, []byte("cfg"))
 	body := fmt.Sprintf(`{"schemaVersion":2,"mediaType":%q,"artifactType":%q,"config":{"mediaType":%q,"digest":%q,"size":3},"layers":[]}`,
 		mediaTypeOCIManifest, sigArtifactType, sigConfigType, cfg)
@@ -97,7 +97,7 @@ func TestPutManifest_NoSubjectLeavesFieldsEmpty(t *testing.T) {
 }
 
 func TestPutManifest_MalformedSubjectDigestRejected(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	cfg := pushBlob(t, reg, referrerRepo, []byte("cfg"))
 	body := imageWithSubject(cfg, "sha256:nothex", "")
 
@@ -111,7 +111,7 @@ func TestPutManifest_MalformedSubjectDigestRejected(t *testing.T) {
 }
 
 func TestPutManifest_IndexWithContentTypeParameterGetsNoConfigArtifactType(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	seedRepo(t, reg, referrerRepo)
 
 	body := fmt.Sprintf(`{"schemaVersion":2,"mediaType":%q,"manifests":[],"subject":{"mediaType":%q,"digest":%q,"size":10}}`,
@@ -128,7 +128,7 @@ func TestPutManifest_IndexWithContentTypeParameterGetsNoConfigArtifactType(t *te
 }
 
 func TestDeleteReferrerManifest_RemovesItFromTheSubjectListing(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	cfg := pushBlob(t, reg, referrerRepo, []byte("cfg"))
 	body := imageWithSubject(cfg, absentSubjectDgt, sigArtifactType)
 	dg := digestOf([]byte(body))
@@ -183,7 +183,7 @@ func getReferrersBody(t *testing.T, reg *Registry, path string) (*httptest.Respo
 }
 
 func TestGetReferrers_ListsReferrersAsAnOCIIndex(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	dgA := pushReferrer(t, reg, referrerRepo, absentSubjectDgt, sigArtifactType, "cfgA")
 	dgB := pushReferrer(t, reg, referrerRepo, absentSubjectDgt, "application/vnd.example.sbom.v1", "cfgB")
 
@@ -202,7 +202,7 @@ func TestGetReferrers_ListsReferrersAsAnOCIIndex(t *testing.T) {
 }
 
 func TestGetReferrers_NoReferrersIsAnEmptyIndexNotAnError(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	seedRepo(t, reg, referrerRepo)
 
 	w := do(reg, http.MethodGet, "/v2/"+referrerRepo+"/referrers/"+absentSubjectDgt, nil, nil)
@@ -212,7 +212,7 @@ func TestGetReferrers_NoReferrersIsAnEmptyIndexNotAnError(t *testing.T) {
 }
 
 func TestGetReferrers_ArtifactTypeFilter(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	dgSig := pushReferrer(t, reg, referrerRepo, absentSubjectDgt, sigArtifactType, "cfgA")
 	pushReferrer(t, reg, referrerRepo, absentSubjectDgt, "application/vnd.example.sbom.v1", "cfgB")
 
@@ -230,7 +230,7 @@ func TestGetReferrers_ArtifactTypeFilter(t *testing.T) {
 }
 
 func TestGetReferrers_MalformedDigest(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	seedRepo(t, reg, referrerRepo)
 
 	w := do(reg, http.MethodGet, "/v2/"+referrerRepo+"/referrers/not-a-digest", nil, nil)
@@ -239,14 +239,14 @@ func TestGetReferrers_MalformedDigest(t *testing.T) {
 }
 
 func TestGetReferrers_UnknownRepository(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	w := do(reg, http.MethodGet, "/v2/team/ghost/referrers/"+absentSubjectDgt, nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 	assertCode(t, w, "NAME_UNKNOWN")
 }
 
 func TestGetReferrers_DoesNotCrossRepositories(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	pushReferrer(t, reg, referrerRepo, absentSubjectDgt, sigArtifactType, "cfgA")
 	seedRepo(t, reg, "team/other")
 
@@ -256,7 +256,7 @@ func TestGetReferrers_DoesNotCrossRepositories(t *testing.T) {
 }
 
 func TestReferrersRoute_OnlyGET(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	seedRepo(t, reg, referrerRepo)
 	for _, method := range []string{http.MethodPut, http.MethodDelete, http.MethodPost, http.MethodPatch} {
 		w := do(reg, method, "/v2/"+referrerRepo+"/referrers/"+absentSubjectDgt, []byte("{}"), nil)
@@ -272,7 +272,7 @@ func (failingReferrerStore) ListReferrers(context.Context, string, string, strin
 }
 
 func TestGetReferrers_MetadataFailureIsA5xxNotA404(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	seedRepo(t, reg, referrerRepo)
 	reg.Meta = failingReferrerStore{reg.Meta}
 
@@ -310,7 +310,7 @@ func referrerDigests(t *testing.T, reg *Registry, repo, subject string) []string
 }
 
 func TestDeleteManifest_SubjectDeleteRemovesItsReferrers(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	sig := pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "sig")
 	sbom := pushReferrer(t, reg, referrerRepo, subject, "application/vnd.example.sbom.v1", "sbom")
@@ -326,7 +326,7 @@ func TestDeleteManifest_SubjectDeleteRemovesItsReferrers(t *testing.T) {
 }
 
 func TestDeleteManifest_NestedReferrersAreRemoved(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	sig := pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "sig")
 	sigOfSig := pushReferrer(t, reg, referrerRepo, sig, sigArtifactType, "sig-of-sig")
@@ -341,7 +341,7 @@ func TestDeleteManifest_NestedReferrersAreRemoved(t *testing.T) {
 }
 
 func TestDeleteManifest_ReferrerDeleteLeavesTheSubject(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	sig := pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "sig")
 	other := pushReferrer(t, reg, referrerRepo, subject, "application/vnd.example.sbom.v1", "sbom")
@@ -355,7 +355,7 @@ func TestDeleteManifest_ReferrerDeleteLeavesTheSubject(t *testing.T) {
 }
 
 func TestDeleteManifest_TagDeleteDoesNotCascade(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	sig := pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "sig")
 	require.NoError(t, reg.Meta.PutTag(context.Background(), testAccount, referrerRepo, "v1", subject))
@@ -368,7 +368,7 @@ func TestDeleteManifest_TagDeleteDoesNotCascade(t *testing.T) {
 }
 
 func TestDeleteManifest_CascadeStaysInsideTheRepository(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "sig")
 	elsewhere := pushReferrer(t, reg, "team/other", subject, sigArtifactType, "sig-other")
@@ -394,7 +394,7 @@ func (s failingDeleteStore) DeleteManifestMeta(ctx context.Context, account, rep
 }
 
 func TestDeleteManifest_ReferrerFailureDoesNotFailTheSubjectDelete(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	subject := pushSubjectImage(t, reg, referrerRepo, "subject")
 	stuck := pushReferrer(t, reg, referrerRepo, subject, sigArtifactType, "stuck")
 	fine := pushReferrer(t, reg, referrerRepo, subject, "application/vnd.example.sbom.v1", "fine")

spinifex/domains/ecr/registry/registry_test.go

@@ -2,7 +2,6 @@ package ecrregistry
 
 import (
 	"context"
-	"github.com/mulgadc/spinifex/internal/testkit"
 
 	"crypto/sha256"
 	"encoding/hex"
@@ -22,15 +21,8 @@ import (
 
 const testAccount = "000000000000"
 
-func newTestRegistry(t *testing.T) *Registry {
-	t.Helper()
-	return NewRegistry(objectstore.NewMemoryObjectStore(), newTestMetaStore(t), testAccount)
-}
-
-func newTestMetaStore(t *testing.T) *ecr.KVMetaStore {
-	t.Helper()
-	_, _, js := testutil.StartTestJetStream(t)
-	return ecr.NewKVMetaStore(js)
+func newTestRegistry() *Registry {
+	return NewRegistry(objectstore.NewMemoryObjectStore(), ecr.NewMemoryMetaStore(), testAccount)
 }
 
 func digestOf(b []byte) string {
@@ -77,7 +69,7 @@ func pushBlob(t *testing.T, reg *Registry, repo string, data []byte) string {
 }
 
 func TestBlobUpload_Monolithic_And_Head_Get(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	data := []byte("hello layer bytes")
 	dg := pushBlob(t, reg, "team/app", data)
 
@@ -92,20 +84,20 @@ func TestBlobUpload_Monolithic_And_Head_Get(t *testing.T) {
 }
 
 func TestBlobHead_NotFound(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	w := do(reg, http.MethodHead, "/v2/team/app/blobs/"+digestOf([]byte("absent")), nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 }
 
 func TestBlobGet_MalformedDigest(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	w := do(reg, http.MethodGet, "/v2/team/app/blobs/sha256:zzz", nil, nil)
 	assert.Equal(t, http.StatusBadRequest, w.Code)
 	assertCode(t, w, "DIGEST_INVALID")
 }
 
 func TestBlobUpload_Chunked(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	seedRepo(t, reg, "r/repo")
 	c1 := []byte("first-chunk-")
 	c2 := []byte("second-chunk")
@@ -132,7 +124,7 @@ func TestBlobUpload_Chunked(t *testing.T) {
 }
 
 func TestBlobUpload_DigestMismatch(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	seedRepo(t, reg, "r/repo")
 	w := do(reg, http.MethodPost, "/v2/r/repo/blobs/uploads/", nil, nil)
 	loc := w.Header().Get("Location")
@@ -142,7 +134,7 @@ func TestBlobUpload_DigestMismatch(t *testing.T) {
 }
 
 func TestBlobUpload_Cancel(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	seedRepo(t, reg, "r/repo")
 	w := do(reg, http.MethodPost, "/v2/r/repo/blobs/uploads/", nil, nil)
 	loc := w.Header().Get("Location")
@@ -169,7 +161,7 @@ func putManifestReferencing(t *testing.T, reg *Registry, repo, digest string) {
 }
 
 func TestBlobMount_HitAndMiss(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	data := []byte("shared-layer")
 	dg := pushBlob(t, reg, "team/source", data)
 	putManifestReferencing(t, reg, "team/source", dg)
@@ -193,7 +185,7 @@ func TestBlobMount_HitAndMiss(t *testing.T) {
 // as a plain upload start, matching the authorization middleware's
 // classification of the same request.
 func TestBlobMount_MissingFromFallsThroughToUpload(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	data := []byte("shared-layer")
 	dg := pushBlob(t, reg, "team/source", data)
 	putManifestReferencing(t, reg, "team/source", dg)
@@ -208,7 +200,7 @@ func TestBlobMount_MissingFromFallsThroughToUpload(t *testing.T) {
 // named source repository is rejected as a mount — proving digest existence
 // alone is not enough to mount cross-repo.
 func TestBlobMount_UnreferencedDigestFallsThrough(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	unreferenced := pushBlob(t, reg, "team/other", []byte("not-in-source"))
 	seedRepo(t, reg, "team/source")
 	seedRepo(t, reg, "team/dest")
@@ -222,7 +214,7 @@ func TestBlobMount_UnreferencedDigestFallsThrough(t *testing.T) {
 // repository that does not exist never mounts, and never distinguishes
 // itself from an unreferenced-digest miss (both fall through identically).
 func TestBlobMount_NonexistentSourceFallsThrough(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	dg := pushBlob(t, reg, "team/elsewhere", []byte("some-bytes"))
 	seedRepo(t, reg, "team/dest")
 
@@ -232,7 +224,7 @@ func TestBlobMount_NonexistentSourceFallsThrough(t *testing.T) {
 }
 
 func TestManifest_PutGetHead_ImageManifest(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 
 	cfg := []byte(`{"config":true}`)
@@ -263,7 +255,7 @@ func TestManifest_PutGetHead_ImageManifest(t *testing.T) {
 }
 
 func TestManifest_Put_MissingBlob(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	seedRepo(t, reg, repo)
 	manifest := fmt.Appendf(nil,
@@ -276,7 +268,7 @@ func TestManifest_Put_MissingBlob(t *testing.T) {
 }
 
 func TestManifest_Get_AcceptMismatch(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	cfg := pushBlob(t, reg, repo, []byte("c"))
 	manifest := fmt.Appendf(nil,
@@ -293,7 +285,7 @@ func TestManifest_Get_AcceptMismatch(t *testing.T) {
 // acceptable manifest media type as its own Accept header line rather than one
 // comma-joined value. Negotiation must consider every line, not just the first.
 func TestManifest_Get_MultipleAcceptHeaders(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	cfg := pushBlob(t, reg, repo, []byte("c"))
 	manifest := fmt.Appendf(nil,
@@ -313,7 +305,7 @@ func TestManifest_Get_MultipleAcceptHeaders(t *testing.T) {
 }
 
 func TestManifest_Get_Unknown(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	w := do(reg, http.MethodGet, "/v2/team/app/manifests/missing", nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 	assertCode(t, w, "MANIFEST_UNKNOWN")
@@ -322,7 +314,7 @@ func TestManifest_Get_Unknown(t *testing.T) {
 // TestStartUpload_RepoNotCreated asserts a blob upload to a repository that was
 // never created is rejected up front — ECR does not auto-create on push.
 func TestStartUpload_RepoNotCreated(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	w := do(reg, http.MethodPost, "/v2/team/ghost/blobs/uploads/", nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 	assertCode(t, w, "NAME_UNKNOWN")
@@ -331,7 +323,7 @@ func TestStartUpload_RepoNotCreated(t *testing.T) {
 // TestPutManifest_RepoNotCreated asserts a manifest PUT to an uncreated
 // repository is rejected with NAME_UNKNOWN.
 func TestPutManifest_RepoNotCreated(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	manifest := fmt.Appendf(nil, `{"mediaType":"%s","layers":[]}`, mediaTypeDockerManifest)
 	w := do(reg, http.MethodPut, "/v2/team/ghost/manifests/v1", manifest,
 		map[string]string{"Content-Type": mediaTypeDockerManifest})
@@ -340,7 +332,7 @@ func TestPutManifest_RepoNotCreated(t *testing.T) {
 }
 
 func TestManifest_ImageIndex_ChildValidation(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 
 	// Push a child image manifest first.
@@ -380,7 +372,7 @@ func TestManifest_ImageIndex_ChildValidation(t *testing.T) {
 }
 
 func TestManifest_Delete_Tag(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	cfg := pushBlob(t, reg, repo, []byte("c"))
 	manifest := fmt.Appendf(nil,
@@ -398,7 +390,7 @@ func TestManifest_Delete_Tag(t *testing.T) {
 // TestManifest_Delete_ByDigest proves delete-by-digest is no longer a no-op: it
 // removes the manifest record + object, and a subsequent GET is a 404.
 func TestManifest_Delete_ByDigest(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	cfg := pushBlob(t, reg, repo, []byte("cfg-bytes"))
 	manifest := fmt.Appendf(nil,
@@ -420,7 +412,7 @@ func TestManifest_Delete_ByDigest(t *testing.T) {
 }
 
 func TestTagsList(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	repo := "team/app"
 	cfg := pushBlob(t, reg, repo, []byte("c"))
 	manifest := fmt.Appendf(nil,
@@ -442,13 +434,13 @@ func TestTagsList(t *testing.T) {
 }
 
 func TestTagsList_UnknownRepo(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	w := do(reg, http.MethodGet, "/v2/team/ghost/tags/list", nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 }
 
 func TestCatalog(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	pushBlob(t, reg, "team/a", []byte("a"))
 	pushBlob(t, reg, "team/b", []byte("b"))
 
@@ -462,21 +454,21 @@ func TestCatalog(t *testing.T) {
 }
 
 func TestDispatch_InvalidRepoName(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	w := do(reg, http.MethodGet, "/v2/Team/App/tags/list", nil, nil)
 	assert.Equal(t, http.StatusBadRequest, w.Code)
 	assertCode(t, w, "NAME_INVALID")
 }
 
 func TestDispatch_RepeatedSeparatorRepoName(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	dg := pushBlob(t, reg, "team/a__b--c", []byte("layer"))
 	w := do(reg, http.MethodHead, "/v2/team/a__b--c/blobs/"+dg, nil, nil)
 	assert.Equal(t, http.StatusOK, w.Code)
 }
 
 func TestDispatch_UnknownPath(t *testing.T) {
-	reg := newTestRegistry(t)
+	reg := newTestRegistry()
 	w := do(reg, http.MethodGet, "/v2/team/app/frobnicate", nil, nil)
 	assert.Equal(t, http.StatusNotFound, w.Code)
 }

spinifex/domains/ecr/registry/registry_upload_test.go

@@ -35,7 +35,7 @@ func (s *blobPutFailStore) PutObject(ctx context.Context, input *s3.PutObjectInp
 func TestFinishUpload_StoreFailure_CleansUp(t *testing.T) {
 	mem := objectstore.NewMemoryObjectStore()
 	store := &blobPutFailStore{MemoryObjectStore: mem}
-	meta := newTestMetaStore(t)
+	meta := ecr.NewMemoryMetaStore()
 	reg := NewRegistry(store, meta, testAccount)
 	seedRepo(t, reg, "team/app")
 
@@ -72,7 +72,7 @@ func awsString(s string) *string { return &s }
 // updates issued from the same observed revision cannot both commit. One wins,
 // the other gets ErrConflict — so a racing PATCH is rejected, never merged.
 func TestUploadCAS_SameRevisionSingleWinner(t *testing.T) {
-	meta := newTestMetaStore(t)
+	meta := ecr.NewMemoryMetaStore()
 	rev, err := meta.PutUpload(context.Background(), testAccount, "u-race", ecr.UploadState{RepoName: "r"})
 	require.NoError(t, err)
 
@@ -116,7 +116,7 @@ func TestUploadCAS_SameRevisionSingleWinner(t *testing.T) {
 // hash state and the recorded byte object never desynchronise.
 func TestPatchUpload_ConcurrentNoCorruption(t *testing.T) {
 	for iter := range 25 {
-		reg := newTestRegistry(t)
+		reg := newTestRegistry()
 		repo := "team/race"
 		seedRepo(t, reg, repo)
 

spinifex/domains/ecr/service_impl.go

@@ -16,7 +16,8 @@ type MetaServiceImpl struct {
 
 var _ MetaService = (*MetaServiceImpl)(nil)
 
-// NewMetaServiceImpl builds a MetaService over an explicit MetaStore.
+// NewMetaServiceImpl builds a MetaService over an explicit MetaStore (used by
+// daemon-side tests with a MemoryMetaStore).
 func NewMetaServiceImpl(store MetaStore) *MetaServiceImpl {
 	return &MetaServiceImpl{store: store}
 }

spinifex/domains/ecr/service_test.go

@@ -21,7 +21,7 @@ const svcTestAccount = "000000000000"
 // ErrNotFound/ErrConflict into the response flags rather than transport errors,
 // so they survive the AWS-error-code-only NATS reply.
 func TestMetaServiceImpl_FlagMapping(t *testing.T) {
-	svc := NewMetaServiceImpl(newTestKVMetaStore(t))
+	svc := NewMetaServiceImpl(NewMemoryMetaStore())
 
 	// Absent repo -> Found:false, nil error.
 	dr, err := svc.RepoDescribe(context.Background(), &RepoDescribeRequest{Repo: "ghost"}, svcTestAccount)

tests/e2e/harness/aws.go

@@ -4,6 +4,7 @@ package harness
 
 import (
 	"crypto/tls"
+	"crypto/x509"
 	"fmt"
 	"net/http"
 	"os"
@@ -150,3 +151,18 @@ func newAWSClient(t *testing.T, env *Env, accessKey, secretKey, sessionToken str
 		BedrockRuntime: bedrockruntime.New(sess),
 	}
 }
+
+// IgnoreCertErrors disables TLS verification on all bundled clients.
+// Use only for fault-injection scenarios.
+func (c *AWSClient) IgnoreCertErrors() {
+	tr := &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}
+	hc := &http.Client{Transport: tr}
+	c.EC2Conf.Config.HTTPClient = hc
+	c.ELBv2.Config.HTTPClient = hc
+	c.EKS.Config.HTTPClient = hc
+	c.IAM.Config.HTTPClient = hc
+	c.STS.Config.HTTPClient = hc
+	c.ACM.Config.HTTPClient = hc
+	c.ECR.Config.HTTPClient = hc
+	_ = (*x509.CertPool)(nil) // silence unused-import on hardened paths
+}

tests/e2e/harness/ec2helpers.go

@@ -78,6 +78,18 @@ func LsblkRootGiB(t *testing.T, tgt SSHTarget) int {
 	return int(bytesN / (1 << 30))
 }
 
+// GetSerialConsoleAccessEnabled returns the current account attribute value.
+// Wraps ec2.GetSerialConsoleAccessStatus so callers don't have to deal with
+// the pointer-bool field.
+func GetSerialConsoleAccessEnabled(t *testing.T, c *AWSClient) bool {
+	t.Helper()
+	out, err := c.EC2.GetSerialConsoleAccessStatus(&ec2.GetSerialConsoleAccessStatusInput{})
+	if err != nil {
+		t.Fatalf("GetSerialConsoleAccessStatus: %v", err)
+	}
+	return aws.BoolValue(out.SerialConsoleAccessEnabled)
+}
+
 // EnableSerialConsoleAccess flips the account attribute on and returns the
 // new value (should be true).
 func EnableSerialConsoleAccess(t *testing.T, c *AWSClient) bool {

tests/e2e/harness/iam.go

@@ -83,3 +83,58 @@ func IAMDeleteRoleAndProfilesBestEffort(c *AWSClient, roleName string, profileNa
 	}
 	_, _ = c.IAM.DeleteRole(&iam.DeleteRoleInput{RoleName: aws.String(roleName)})
 }
+
+// IAMDeleteGroupBestEffort tears down a group: removes any members, detaches all
+// attached policies, deletes any inline policies, then deletes the group. Each
+// step swallows errors so a missing fragment never cascades; usable both as a
+// pre-test sweep and as a fixture-teardown cleanup. Order matters: a group cannot
+// be deleted while it still has members, attached policies, or inline policies.
+func IAMDeleteGroupBestEffort(c *AWSClient, groupName string, memberUsers []string, policyARNs ...string) {
+	for _, u := range memberUsers {
+		_, _ = c.IAM.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
+			GroupName: aws.String(groupName),
+			UserName:  aws.String(u),
+		})
+	}
+	for _, arn := range policyARNs {
+		_, _ = c.IAM.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
+			GroupName: aws.String(groupName),
+			PolicyArn: aws.String(arn),
+		})
+	}
+	// Defensive: pull any other attached policies so the final DeleteGroup isn't
+	// blocked by a stray attach from a partial run.
+	if attached, err := c.IAM.ListAttachedGroupPolicies(&iam.ListAttachedGroupPoliciesInput{
+		GroupName: aws.String(groupName),
+	}); err == nil {
+		for _, p := range attached.AttachedPolicies {
+			_, _ = c.IAM.DetachGroupPolicy(&iam.DetachGroupPolicyInput{
+				GroupName: aws.String(groupName),
+				PolicyArn: p.PolicyArn,
+			})
+		}
+	}
+	// Defensive: delete any inline policies so the inline-policy guard can't block
+	// the final DeleteGroup after a partial run.
+	if inline, err := c.IAM.ListGroupPolicies(&iam.ListGroupPoliciesInput{
+		GroupName: aws.String(groupName),
+	}); err == nil {
+		for _, name := range inline.PolicyNames {
+			_, _ = c.IAM.DeleteGroupPolicy(&iam.DeleteGroupPolicyInput{
+				GroupName:  aws.String(groupName),
+				PolicyName: name,
+			})
+		}
+	}
+	// Defensive: drop any members the caller didn't pass in (partial run could
+	// have added one), so the non-empty guard can't block the delete.
+	if got, err := c.IAM.GetGroup(&iam.GetGroupInput{GroupName: aws.String(groupName)}); err == nil {
+		for _, u := range got.Users {
+			_, _ = c.IAM.RemoveUserFromGroup(&iam.RemoveUserFromGroupInput{
+				GroupName: aws.String(groupName),
+				UserName:  u.UserName,
+			})
+		}
+	}
+	_, _ = c.IAM.DeleteGroup(&iam.DeleteGroupInput{GroupName: aws.String(groupName)})
+}

tests/e2e/harness/kubectl.go

@@ -4,8 +4,11 @@ package harness
 
 import (
 	"bytes"
+	"encoding/json"
+	"fmt"
 	"os"
 	"os/exec"
+	"strings"
 	"testing"
 	"time"
 )
@@ -62,6 +65,84 @@ func (k *Kubectl) Run(timeout time.Duration, args ...string) (string, error) {
 	}
 }
 
+// AuthDiagnostics reproduces the credential exchange behind a failing call
+// without mutating anything. `Unauthorized` on its own says only that the
+// apiserver refused the request, and the two explanations — a token it rejected
+// versus one the exec plugin never produced — need different fixes.
+//
+// It probes an authenticated path rather than `/version`, which is anonymously
+// readable and so returns 200 under either explanation, and it invokes the
+// credential plugin directly so a plugin that failed is named rather than
+// inferred from a missing header.
+func (k *Kubectl) AuthDiagnostics(timeout time.Duration) string {
+	var b strings.Builder
+
+	b.WriteString("=== kubectl auth whoami ===\n")
+	out, err := k.Run(timeout, "auth", "whoami")
+	b.WriteString(out)
+	if err != nil {
+		b.WriteString("\nerror: " + err.Error() + "\n")
+	}
+
+	// kube-system is not anonymously readable, so a 200 here proves a credential
+	// was both produced and accepted.
+	b.WriteString("\n=== GET /api/v1/namespaces/kube-system ===\n")
+	out, err = k.Run(timeout, "--v=8", "get", "--raw", "/api/v1/namespaces/kube-system")
+	b.WriteString(out)
+	if err != nil {
+		b.WriteString("\nerror: " + err.Error() + "\n")
+	}
+
+	b.WriteString("\n=== credential plugin, invoked directly ===\n")
+	b.WriteString(k.execPluginProbe(timeout))
+	return b.String()
+}
+
+// execPluginProbe runs the kubeconfig's own exec credential plugin and reports
+// whether it produced a token. The token itself is never printed — CI logs are
+// retained for weeks and it is a live cluster credential.
+func (k *Kubectl) execPluginProbe(timeout time.Duration) string {
+	spec, err := k.Run(timeout, "config", "view", "--raw", "-o",
+		"jsonpath={.users[0].user.exec.command}{range .users[0].user.exec.args[*]} {@}{end}")
+	if err != nil {
+		return "could not read exec block from kubeconfig: " + err.Error() + "\n" + spec
+	}
+	argv := strings.Fields(strings.TrimSpace(spec))
+	if len(argv) == 0 {
+		return "kubeconfig user has no exec credential plugin\n"
+	}
+
+	bin, err := exec.LookPath(argv[0])
+	if err != nil {
+		return "exec plugin " + argv[0] + " not on PATH: " + err.Error() + "\n"
+	}
+	cmd := exec.Command(bin, argv[1:]...)
+	cmd.Env = append(os.Environ(), k.AWSEnv...)
+	var stdout, stderr bytes.Buffer
+	cmd.Stdout, cmd.Stderr = &stdout, &stderr
+
+	var b strings.Builder
+	fmt.Fprintf(&b, "argv: %s\n", strings.Join(argv, " "))
+	runErr := runWithTimeout(cmd, timeout)
+	fmt.Fprintf(&b, "exit: %v\n", runErr)
+	fmt.Fprintf(&b, "stderr: %s\n", strings.TrimSpace(stderr.String()))
+
+	var cred struct {
+		Status struct {
+			Token               string `json:"token"`
+			ExpirationTimestamp string `json:"expirationTimestamp"`
+		} `json:"status"`
+	}
+	if jsonErr := json.Unmarshal(stdout.Bytes(), &cred); jsonErr != nil {
+		fmt.Fprintf(&b, "stdout did not parse as an ExecCredential: %v (%d bytes)\n",
+			jsonErr, stdout.Len())
+		return b.String()
+	}
+	fmt.Fprintf(&b, "token produced: %t (%d chars), expires: %q\n",
+		cred.Status.Token != "", len(cred.Status.Token), cred.Status.ExpirationTimestamp)
+	return b.String()
+}
+
 type timeoutError struct{}
 
 func (*timeoutError) Error() string { return "kubectl timed out" }

tests/e2e/harness/poll.go

@@ -151,3 +151,34 @@ func WaitForSnapshotState(t *testing.T, c *AWSClient, id, target string, opts ..
 	t.Logf("snapshot %s reached state %s", id, target)
 	return last
 }
+
+// WaitForImageState polls DescribeImages until State == target. AMI creation
+// from a snapshot is the slowest path; default timeout 10min.
+func WaitForImageState(t *testing.T, c *AWSClient, id, target string, opts ...PollOpt) *ec2.Image {
+	t.Helper()
+	cfg := applyOpts(pollCfg{timeout: 10 * time.Minute, interval: 2 * time.Second}, opts...)
+	var last *ec2.Image
+	var lastState string
+	EventuallyErr(t, func() error {
+		out, err := c.EC2.DescribeImages(&ec2.DescribeImagesInput{
+			ImageIds: []*string{aws.String(id)},
+		})
+		if err != nil {
+			return fmt.Errorf("describe-image %s: %w", id, err)
+		}
+		if len(out.Images) == 0 {
+			return fmt.Errorf("%s not found", id)
+		}
+		last = out.Images[0]
+		lastState = aws.StringValue(last.State)
+		if lastState == target {
+			return nil
+		}
+		if lastState == "failed" {
+			return fmt.Errorf("%s entered failed state: %s", id, aws.StringValue(last.StateReason.Message))
+		}
+		return fmt.Errorf("%s state=%s want=%s", id, lastState, target)
+	}, cfg.timeout, cfg.interval)
+	t.Logf("image %s reached state %s", id, target)
+	return last
+}

tests/e2e/harness/rds.go

@@ -121,6 +121,27 @@ func WaitForDBInstanceGone(t *testing.T, c *AWSClient, id string, opts ...PollOp
 	t.Logf("DB instance %s is gone", id)
 }
 
+// WaitForDBSnapshotAvailable polls until the snapshot reports available, which
+// is the point the data exists and a restore will accept it.
+func WaitForDBSnapshotAvailable(t *testing.T, c *AWSClient, id string, opts ...PollOpt) *rds.DBSnapshot {
+	t.Helper()
+	cfg := applyOpts(pollCfg{timeout: 10 * time.Minute, interval: 5 * time.Second}, opts...)
+	var last *rds.DBSnapshot
+	EventuallyErr(t, func() error {
+		snapshot, err := DescribeDBSnapshot(c, id)
+		if err != nil {
+			return fmt.Errorf("describe-db-snapshots %s: %w", id, err)
+		}
+		last = snapshot
+		if status := aws.StringValue(snapshot.Status); status != DBSnapshotAvailable {
+			return fmt.Errorf("%s status=%s want=%s", id, status, DBSnapshotAvailable)
+		}
+		return nil
+	}, cfg.timeout, cfg.interval)
+	t.Logf("DB snapshot %s reached status %s", id, DBSnapshotAvailable)
+	return last
+}
+
 // DescribeDBInstance returns the single named DB instance. A named instance that
 // does not exist is an error from the API, not an empty list.
 func DescribeDBInstance(c *AWSClient, id string) (*rds.DBInstance, error) {

tests/e2e/harness/traffic.go

@@ -83,6 +83,38 @@ func TCPRoundRobin(host string, port int, n int, timeout time.Duration) TrafficR
 	return r
 }
 
+// UDPRoundRobin sends n UDP datagrams to host:port, reads the echoed line per
+// datagram, and counts responders. Matches the app-userdata UDP echo responder.
+// UDP is connectionless, so each probe writes a payload then reads with a
+// deadline; a dropped datagram counts as a failure rather than blocking.
+func UDPRoundRobin(host string, port int, n int, timeout time.Duration) TrafficResult {
+	r := TrafficResult{Distribution: map[string]int{}, Total: n}
+	for i := 0; i < n; i++ {
+		conn, err := net.DialTimeout("udp", fmt.Sprintf("%s:%d", host, port), timeout)
+		if err != nil {
+			continue
+		}
+		if _, err := conn.Write([]byte("ping")); err != nil {
+			conn.Close()
+			continue
+		}
+		conn.SetReadDeadline(time.Now().Add(timeout))
+		buf := make([]byte, 256)
+		nb, err := conn.Read(buf)
+		conn.Close()
+		if err != nil {
+			continue
+		}
+		resp := strings.TrimSpace(string(buf[:nb]))
+		if resp == "" {
+			continue
+		}
+		r.Distribution[resp]++
+		r.Successful++
+	}
+	return r
+}
+
 // AssertRoundRobin asserts that r has at least minUnique distinct responders
 // and at least minSuccess successful probes. Logs the distribution either way.
 func AssertRoundRobin(t *testing.T, r TrafficResult, minUnique, minSuccess int, label string) {

tests/integration/ecr_daemon_lite.go

@@ -7,16 +7,15 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/domains/ecr"
 	"github.com/nats-io/nats.go"
-	"github.com/nats-io/nats.go/jetstream"
-	"github.com/stretchr/testify/require"
 )
 
 // StartECRDaemonLite subscribes a real ecr.MetaServiceImpl (the same
 // production service the live daemon runs, see runtime/daemon/daemon.go)
 // to every ecr.* NATS subject the gateway's ECR control-plane handlers and
 // ECRRegistry.Meta dial for metadata — repos, policies, lifecycle policies,
-// tags, manifest records, and in-progress upload state. It is backed by
-// JetStream KV on the test's isolated NATS account, like the live daemon.
+// tags, manifest records, and in-progress upload state. It is memory-backed
+// (ecr.NewMemoryMetaStore) rather than JetStream-KV-backed, matching
+// DaemonLite's in-scope resources.
 //
 // Blob and manifest bytes never travel these subjects (ECRRegistry.Store
 // handles those directly against its own objectstore.ObjectStore, wired in
@@ -29,10 +28,8 @@ import (
 func StartECRDaemonLite(t *testing.T, gw *Gateway) *ecr.MetaServiceImpl {
 	t.Helper()
 
+	svc := ecr.NewMetaServiceImpl(ecr.NewMemoryMetaStore())
 	nc := gw.NATSConn
-	js, err := jetstream.New(nc)
-	require.NoError(t, err, "jetstream handle for the ECR meta store")
-	svc := ecr.NewKVMetaService(js)
 
 	sub(t, nc, ecr.SubjectRepoCreate, func(m *nats.Msg) { dispatch(m, svc.RepoCreate) })
 	sub(t, nc, ecr.SubjectRepoDescribe, func(m *nats.Msg) { dispatch(m, svc.RepoDescribe) })