feat(lambda): provide test-link SDK gateway settings (#1655)

* fix(eks): alarm on a control-plane VM with no cluster record instead of terminating it

* chore(eks): order test imports

* feat(eks): add spx admin eks orphans and share the cluster-record judgement with the billable reaper

* fix(eks): list orphan control planes through an EC2-owned inventory capability

* test(eks): cover the orphan report ENI reader and the inventory fail-closed path

* fix(eks): report only running control planes and treat a missing ENI as gone

* fix(eks): open DNS watch buckets attach-only so a deleted account bucket is not recreated

* fix(eks): read control-plane ENIs from their record so a corrupt one is unreadable, and report the zone

cmd/lambda-agent/sdk_config_testlink.go

@@ -0,0 +1,65 @@
+//go:build lambdatestlink
+
+package main
+
+import (
+	"crypto/x509"
+	"errors"
+	"fmt"
+	"net"
+	"net/url"
+	"os"
+	"path/filepath"
+)
+
+const (
+	sdkEndpointEntry = "test-sdk-endpoint"
+	sdkCAEntry       = "test-sdk-ca"
+	sdkBundlePath    = "/run/lambda-sdk/ca-bundle.pem"
+	systemBundlePath = "/etc/ssl/certs/ca-certificates.crt"
+)
+
+type sdkConfig struct {
+	endpoint string
+	ca       []byte
+}
+
+// readSDKConfig reads optional public settings before the boot config is sealed.
+func readSDKConfig(dir string) (*sdkConfig, error) {
+	endpoint, endpointErr := os.ReadFile(filepath.Join(dir, sdkEndpointEntry, "raw"))
+	ca, caErr := os.ReadFile(filepath.Join(dir, sdkCAEntry, "raw"))
+	if errors.Is(endpointErr, os.ErrNotExist) && errors.Is(caErr, os.ErrNotExist) {
+		return nil, nil
+	}
+	if endpointErr != nil || caErr != nil {
+		return nil, errors.New("incomplete SDK gateway settings")
+	}
+	u, err := url.Parse(string(endpoint))
+	if err != nil || u.Scheme != "https" || u.User != nil || u.Path != "" || u.RawQuery != "" || u.Fragment != "" ||
+		net.ParseIP(u.Hostname()) == nil || u.Port() == "" {
+		return nil, errors.New("invalid SDK gateway endpoint")
+	}
+	if len(ca) > 64<<10 || !x509.NewCertPool().AppendCertsFromPEM(ca) {
+		return nil, errors.New("invalid SDK gateway CA")
+	}
+	return &sdkConfig{endpoint: string(endpoint), ca: ca}, nil
+}
+
+// writeBundle combines the system roots with the cluster's public root.
+func (c *sdkConfig) writeBundle(path, systemPath string) error {
+	system, err := os.ReadFile(systemPath)
+	if err != nil && !errors.Is(err, os.ErrNotExist) {
+		return fmt.Errorf("read system CA bundle: %w", err)
+	}
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		return fmt.Errorf("create SDK CA directory: %w", err)
+	}
+	if err := os.Chmod(filepath.Dir(path), 0o755); err != nil {
+		return fmt.Errorf("make SDK CA directory readable: %w", err)
+	}
+	bundle := append(append(system, '\n'), c.ca...)
+	if err := os.WriteFile(path, bundle, 0o644); err != nil {
+		return fmt.Errorf("write SDK CA bundle: %w", err)
+	}
+	return os.Chmod(path, 0o644)
+}

cmd/lambda-agent/sdk_config_testlink_test.go

@@ -0,0 +1,106 @@
+//go:build lambdatestlink
+
+//test:in-package exercises the unexported boot configuration reader
+package main
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"math/big"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSDKConfig_OptionalAndInvalid(t *testing.T) {
+	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }))
+	defer server.Close()
+	ca := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw})
+	for _, tc := range []struct {
+		name     string
+		endpoint []byte
+		ca       []byte
+		wantErr  bool
+		want     bool
+	}{
+		{name: "absent"},
+		{name: "valid", endpoint: []byte(server.URL), ca: ca, want: true},
+		{name: "missing ca", endpoint: []byte(server.URL), wantErr: true},
+		{name: "missing endpoint", ca: ca, wantErr: true},
+		{name: "invalid endpoint", endpoint: []byte("http://127.0.0.1:1234"), ca: ca, wantErr: true},
+		{name: "invalid ca", endpoint: []byte(server.URL), ca: []byte("not a certificate"), wantErr: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			dir := t.TempDir()
+			for name, value := range map[string][]byte{sdkEndpointEntry: tc.endpoint, sdkCAEntry: tc.ca} {
+				if value != nil {
+					require.NoError(t, os.MkdirAll(filepath.Join(dir, name), 0o700))
+					require.NoError(t, os.WriteFile(filepath.Join(dir, name, "raw"), value, 0o600))
+				}
+			}
+			cfg, err := readSDKConfig(dir)
+			if tc.wantErr {
+				require.Error(t, err)
+				require.Nil(t, cfg)
+				return
+			}
+			require.NoError(t, err)
+			if !tc.want {
+				require.Nil(t, cfg)
+				return
+			}
+			bundle := filepath.Join(t.TempDir(), "sdk", "ca.pem")
+			require.NoError(t, cfg.writeBundle(bundle, filepath.Join(t.TempDir(), "absent")))
+			data, err := os.ReadFile(bundle)
+			require.NoError(t, err)
+			roots := x509.NewCertPool()
+			require.True(t, roots.AppendCertsFromPEM(data))
+			client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: roots}}}
+			resp, err := client.Get(cfg.endpoint)
+			require.NoError(t, err)
+			defer resp.Body.Close()
+			assert.Equal(t, http.StatusOK, resp.StatusCode)
+		})
+	}
+}
+
+func TestSDKConfig_BundleKeepsSystemRoots(t *testing.T) {
+	cluster := httptest.NewTLSServer(nil)
+	defer cluster.Close()
+	clusterCA := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cluster.Certificate().Raw})
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	require.NoError(t, err)
+	template := &x509.Certificate{SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "test system root"},
+		NotBefore: time.Now().Add(-time.Minute), NotAfter: time.Now().Add(time.Hour),
+		IsCA: true, BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign}
+	der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key)
+	require.NoError(t, err)
+	systemCA := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
+	dir := t.TempDir()
+	systemPath, bundlePath := filepath.Join(dir, "system.pem"), filepath.Join(dir, "sdk", "ca.pem")
+	require.NoError(t, os.WriteFile(systemPath, systemCA, 0o600))
+	require.NoError(t, (&sdkConfig{ca: clusterCA}).writeBundle(bundlePath, systemPath))
+	bundle, err := os.ReadFile(bundlePath)
+	require.NoError(t, err)
+	roots := x509.NewCertPool()
+	require.True(t, roots.AppendCertsFromPEM(bundle))
+	client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: roots}}}
+	resp, err := client.Get(cluster.URL)
+	require.NoError(t, err)
+	require.NoError(t, resp.Body.Close())
+	systemCert, err := x509.ParseCertificate(der)
+	require.NoError(t, err)
+	_, err = systemCert.Verify(x509.VerifyOptions{Roots: roots})
+	require.NoError(t, err)
+}

cmd/lambda-agent/start_testlink.go

@@ -12,6 +12,7 @@ import (
 	"net/http"
 	"os"
 	"os/exec"
+	"strings"
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/agents/lambda/agent"
@@ -31,13 +32,17 @@ type testlink struct {
 	pkg            taskPackage
 	runtimeAddr    string
 	startBootstrap func(ctx context.Context, env []string) (*exec.Cmd, error)
+	caBundlePath   string
+	systemCAPath   string
 }
 
 func start(ctx context.Context) int {
 	t := testlink{
-		startup:     defaultStartup(),
-		pkg:         devTaskPackage(),
-		runtimeAddr: runtimeAddr,
+		startup:      defaultStartup(),
+		pkg:          devTaskPackage(),
+		runtimeAddr:  runtimeAddr,
+		caBundlePath: sdkBundlePath,
+		systemCAPath: systemBundlePath,
 		startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
 			return agent.StartBootstrap(ctx, taskRoot+"/bootstrap", env, os.Stdout, os.Stderr)
 		},
@@ -53,6 +58,7 @@ func (t testlink) run(parent context.Context) (int, error) {
 	// The package values are fw_cfg entries, so read them before connect seals it.
 	spec, specErr := readPackageSpec(t.fwCfgDir)
 	grant, grantErr := readGrant(t.fwCfgDir)
+	sdk, sdkErr := readSDKConfig(t.fwCfgDir)
 	conn, scope, code, err := t.connect(parent)
 	if err != nil {
 		return code, errors.Join(err, specErr, grantErr)
@@ -85,6 +91,15 @@ func (t testlink) run(parent context.Context) (int, error) {
 	}
 	cred := *first.Credential
 	first.Credential = nil
+	if sdkErr != nil {
+		slog.Warn("lambda-agent: SDK gateway settings unavailable", "err", sdkErr)
+	}
+	if sdk != nil {
+		if err := sdk.writeBundle(t.caBundlePath, t.systemCAPath); err != nil {
+			slog.Warn("lambda-agent: SDK gateway settings unavailable", "err", err)
+			sdk = nil
+		}
+	}
 	if err := t.pkg.install(spec); err != nil {
 		return exitRefused, fmt.Errorf("install package: %w", err)
 	}
@@ -100,7 +115,8 @@ func (t testlink) run(parent context.Context) (int, error) {
 
 	ctx, cancel := context.WithCancelCause(parent)
 	defer cancel(nil)
-	cmd, err := t.startBootstrap(ctx, bootstrapEnv(ln.Addr().String(), spec.handler, cred))
+	region, _, _ := strings.Cut(string(scope), "/")
+	cmd, err := t.startBootstrap(ctx, bootstrapEnv(ln.Addr().String(), spec.handler, cred, region, sdk, t.caBundlePath))
 	if err != nil {
 		return exitLink, fmt.Errorf("start bootstrap: %w", err)
 	}
@@ -124,10 +140,9 @@ func (t testlink) run(parent context.Context) (int, error) {
 	return exitClosed, nil
 }
 
-// bootstrapEnv is the bootstrap's whole environment: the runtime variables and
-// the execution role's three credential variables.
-func bootstrapEnv(runtimeAPI, handler string, cred hostlink.Credential) []string {
-	return []string{
+// bootstrapEnv supplies the runtime, execution-role and SDK settings.
+func bootstrapEnv(runtimeAPI, handler string, cred hostlink.Credential, region string, sdk *sdkConfig, caPath string) []string {
+	env := []string{
 		"AWS_LAMBDA_RUNTIME_API=" + runtimeAPI,
 		"_HANDLER=" + handler,
 		"LAMBDA_TASK_ROOT=" + taskRoot,
@@ -135,7 +150,13 @@ func bootstrapEnv(runtimeAPI, handler string, cred hostlink.Credential) []string
 		"AWS_ACCESS_KEY_ID=" + cred.AccessKeyID,
 		"AWS_SECRET_ACCESS_KEY=" + cred.SecretAccessKey,
 		"AWS_SESSION_TOKEN=" + cred.SessionToken,
+		"AWS_REGION=" + region,
+		"AWS_DEFAULT_REGION=" + region,
 	}
+	if sdk != nil {
+		env = append(env, "AWS_ENDPOINT_URL="+sdk.endpoint, "AWS_CA_BUNDLE="+caPath)
+	}
+	return env
 }
 
 // shutdownGrace is how long past the invocation deadline the agent waits for

cmd/lambda-agent/start_testlink_test.go

@@ -7,8 +7,10 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/pem"
 	"io"
 	"net/http"
+	"net/http/httptest"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -70,6 +72,7 @@ type bootstrapStarts struct {
 	sealed  bool
 	// handlerOut receives the _HANDLER the bootstrap process itself saw.
 	handlerOut string
+	caBundle   string
 	// env is the environment the agent gave the bootstrap.
 	env []string
 }
@@ -94,6 +97,7 @@ func (b *bootstrapStarts) testlink(t *testing.T, entries map[string][]byte) test
 	b.started = make(chan struct{}, 1)
 	b.task = t.TempDir()
 	b.handlerOut = filepath.Join(t.TempDir(), "handler")
+	b.caBundle = filepath.Join(t.TempDir(), "sdk", "ca-bundle.pem")
 	return testlink{
 		startup: g.startup(),
 		pkg: taskPackage{
@@ -102,7 +106,9 @@ func (b *bootstrapStarts) testlink(t *testing.T, entries map[string][]byte) test
 			mountDir:   func(string) error { return nil },
 			sealDir:    func(string) error { b.sealed = true; return nil },
 		},
-		runtimeAddr: "127.0.0.1:0",
+		runtimeAddr:  "127.0.0.1:0",
+		caBundlePath: b.caBundle,
+		systemCAPath: filepath.Join(t.TempDir(), "system-ca.pem"),
 		startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
 			b.n++
 			b.env = env
@@ -176,10 +182,45 @@ func TestTestlinkServesOneInvocationThenStops(t *testing.T) {
 		"AWS_ACCESS_KEY_ID=" + testCredential.AccessKeyID,
 		"AWS_SECRET_ACCESS_KEY=" + testCredential.SecretAccessKey,
 		"AWS_SESSION_TOKEN=" + testCredential.SessionToken,
+		"AWS_REGION=ap-southeast-2",
+		"AWS_DEFAULT_REGION=ap-southeast-2",
 	}, b.env, "the bootstrap gets exactly the runtime and credential variables")
 	assert.NotEmpty(t, envValue(b.env, "AWS_LAMBDA_RUNTIME_API"))
 }
 
+func TestTestlinkProvidesSDKSettingsToBootstrap(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }))
+	defer server.Close()
+	h := newHost(t)
+	h.accept(ctx)
+	entries := h.entries()
+	entries[sdkEndpointEntry] = []byte(server.URL)
+	entries[sdkCAEntry] = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw})
+	var b bootstrapStarts
+	done := runTestlinkAsync(ctx, b.testlink(t, entries))
+	conn := h.guest(t)
+	defer conn.Close()
+	hs := hostlink.NewTestStream(conn)
+	req := hostlink.Request{Credential: &testCredential, GrantID: "g-1", AttemptID: "a-1", RequestID: "r-1", DeadlineMs: time.Now().Add(5 * time.Second).UnixMilli(), Payload: []byte(`{}`)}
+	require.NoError(t, hs.HostSend(req))
+	_, err := hs.HostResult(ctx)
+	require.NoError(t, err)
+	res := wait(t, done, "agent to stop")
+	require.NoError(t, res.err)
+	assert.Equal(t, server.URL, envValue(b.env, "AWS_ENDPOINT_URL"))
+	assert.Equal(t, b.caBundle, envValue(b.env, "AWS_CA_BUNDLE"))
+	assert.Equal(t, "ap-southeast-2", envValue(b.env, "AWS_REGION"))
+	assert.Equal(t, "ap-southeast-2", envValue(b.env, "AWS_DEFAULT_REGION"))
+	ca, err := os.ReadFile(b.caBundle)
+	require.NoError(t, err)
+	assert.Contains(t, string(ca), string(entries[sdkCAEntry]))
+	info, err := os.Stat(b.caBundle)
+	require.NoError(t, err)
+	assert.Equal(t, os.FileMode(0o644), info.Mode().Perm())
+}
+
 var testCredential = hostlink.Credential{AccessKeyID: "ASIAGUESTKEY", SecretAccessKey: "guest-secret", SessionToken: "guest-token"}
 
 func envValue(env []string, name string) string {

spinifex/domains/ec2/inventory/inventory.go

@@ -0,0 +1,67 @@
+// Package inventory is the EC2-owned read capability for an account's complete
+// instance list, for callers outside the EC2 domain.
+package inventory
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"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/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Instances lists every non-terminated instance an account owns, or returns an
+// error. It never returns a shorter list than the truth.
+type Instances interface {
+	ListAccountInstances(ctx context.Context, accountID string) (*ec2.DescribeInstancesOutput, error)
+}
+
+// NATSInstances reads the complete list across nodeIDs over NATS.
+type NATSInstances struct {
+	nc      *nats.Conn
+	nodeIDs []string
+}
+
+var _ Instances = (*NATSInstances)(nil)
+
+// NewNATSInstances returns a reader over the configured node set.
+func NewNATSInstances(nc *nats.Conn, nodeIDs []string) *NATSInstances {
+	return &NATSInstances{nc: nc, nodeIDs: nodeIDs}
+}
+
+// ListAccountInstances passes through the strict listing unchanged: any node
+// that did not answer, or any record that could not be read, is an error.
+func (n *NATSInstances) ListAccountInstances(ctx context.Context, accountID string) (*ec2.DescribeInstancesOutput, error) {
+	return ec2instanceapi.DescribeAccountInstancesStrict(ctx, n.nc, n.nodeIDs, accountID)
+}
+
+// NewENITagReader returns a reader of one ENI's tags straight from its stored
+// record, so an ENI that is absent and one that cannot be read are told apart:
+// a missing key is found=false, a failed read or an undecodable record is an
+// error. accountID is the account the ENI is stored under.
+func NewENITagReader(js jetstream.JetStream, accountID string) func(ctx context.Context, eniID string) (map[string]string, bool, error) {
+	return func(ctx context.Context, eniID string) (map[string]string, bool, error) {
+		kv, err := js.KeyValue(ctx, ec2vpc.KVBucketENIs)
+		if err != nil {
+			return nil, false, fmt.Errorf("open ENI bucket: %w", err)
+		}
+		entry, err := kv.Get(ctx, kvutil.AccountKey(accountID, eniID))
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return nil, false, nil
+		}
+		if err != nil {
+			return nil, false, fmt.Errorf("read ENI record %s: %w", eniID, err)
+		}
+		var record ec2vpc.ENIRecord
+		if err := json.Unmarshal(entry.Value(), &record); err != nil {
+			return nil, false, fmt.Errorf("decode ENI record %s: %w", eniID, err)
+		}
+		return record.Tags, true, nil
+	}
+}

spinifex/domains/ec2/inventory/inventory_test.go

@@ -0,0 +1,49 @@
+package inventory
+
+import (
+	"encoding/json"
+	"testing"
+
+	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/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestListAccountInstancesRefusesAnUnknownNodeSet(t *testing.T) {
+	out, err := NewNATSInstances(nil, nil).ListAccountInstances(t.Context(), "000000000000")
+	require.Error(t, err)
+	assert.Nil(t, out, "an unknown node set must not yield a list")
+}
+
+func TestENITagReaderSeparatesGoneFromUnreadable(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	js := testutil.NewJetStream(t, nc)
+	read := NewENITagReader(js, "000000000000")
+
+	_, _, 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)
+	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)
+	require.NoError(t, err)
+	_, err = enis.Put(t.Context(), kvutil.AccountKey("000000000000", "eni-corrupt"), []byte("{not json"))
+	require.NoError(t, err)
+
+	tags, found, err := read(t.Context(), "eni-good")
+	require.NoError(t, err)
+	assert.True(t, found)
+	assert.Equal(t, map[string]string{"k": "v"}, tags)
+
+	_, found, err = read(t.Context(), "eni-absent")
+	require.NoError(t, err)
+	assert.False(t, found, "a missing key is the ENI gone")
+
+	_, found, err = read(t.Context(), "eni-corrupt")
+	require.Error(t, err, "an undecodable record is a read failure")
+	assert.False(t, found)
+}

spinifex/domains/eks/cluster/cp_orphans.go

@@ -0,0 +1,158 @@
+package cluster
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sort"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// ClusterRecordState is what a read of a cluster's record established.
+type ClusterRecordState int
+
+const (
+	ClusterRecordPresent ClusterRecordState = iota
+	ClusterRecordMissing
+	ClusterRecordUnreadable
+)
+
+// ClusterRecordVerdict is the outcome of looking a cluster record up without
+// creating anything. Reason names a missing or unreadable record; Err is set
+// only when the record could not be read.
+type ClusterRecordVerdict struct {
+	State  ClusterRecordState
+	Reason string
+	Err    error
+}
+
+// JudgeClusterRecord reads the cluster record read-only. A missing bucket or a
+// missing key is Missing, any other failure is Unreadable, and neither is ever
+// repaired. The billable reaper and the orphan report both use it, so they
+// judge a record the same way.
+func JudgeClusterRecord(ctx context.Context, js jetstream.JetStream, account, cluster string) ClusterRecordVerdict {
+	kv, err := OpenAccountBucket(ctx, js, account)
+	switch {
+	case errors.Is(err, jetstream.ErrBucketNotFound):
+		return ClusterRecordVerdict{State: ClusterRecordMissing, Reason: "account bucket missing"}
+	case err != nil:
+		return ClusterRecordVerdict{State: ClusterRecordUnreadable, Reason: "account bucket lookup failed", Err: err}
+	}
+	_, err = GetClusterMeta(ctx, kv, cluster)
+	switch {
+	case err == nil:
+		return ClusterRecordVerdict{State: ClusterRecordPresent}
+	case errors.Is(err, ErrClusterNotFound):
+		return ClusterRecordVerdict{State: ClusterRecordMissing, Reason: "cluster record missing"}
+	}
+	return ClusterRecordVerdict{State: ClusterRecordUnreadable, Reason: "cluster record read failed", Err: err}
+}
+
+// ENITagReader returns the tags on a control-plane ENI. found is false when
+// the ENI is gone.
+type ENITagReader func(ctx context.Context, eniID string) (tagSet map[string]string, found bool, err error)
+
+// ControlPlaneRow is one running EKS control-plane VM the report could not tie
+// to a live cluster record. Unreadable rows carry Err and are never omitted.
+type ControlPlaneRow struct {
+	InstanceID       string
+	Cluster          string
+	Account          string
+	Reason           string
+	Age              time.Duration
+	AvailabilityZone string
+	Unreadable       bool
+	Err              error
+}
+
+// FindControlPlaneOrphans classifies the instances in reservations. An instance
+// with no ENI, or whose ENI is gone or not tagged as an EKS control plane, is
+// skipped. A failed ENI read is an unreadable row, never a silent drop. A
+// control plane whose cluster record is present is omitted; one whose bucket or
+// record is missing is an orphan row; any other read failure is unreadable. It
+// only reads.
+func FindControlPlaneOrphans(ctx context.Context, js jetstream.JetStream, reservations []*ec2.Reservation, eni ENITagReader, now time.Time) []ControlPlaneRow {
+	var rows []ControlPlaneRow
+	for _, res := range reservations {
+		if res == nil {
+			continue
+		}
+		for _, inst := range res.Instances {
+			if inst == nil || !instanceRunning(inst) || len(inst.NetworkInterfaces) == 0 || inst.NetworkInterfaces[0] == nil {
+				continue
+			}
+			eniID := aws.StringValue(inst.NetworkInterfaces[0].NetworkInterfaceId)
+			if eniID == "" {
+				continue
+			}
+			row := ControlPlaneRow{InstanceID: aws.StringValue(inst.InstanceId)}
+			if inst.Placement != nil {
+				row.AvailabilityZone = aws.StringValue(inst.Placement.AvailabilityZone)
+			}
+			if inst.LaunchTime != nil {
+				row.Age = now.Sub(*inst.LaunchTime).Truncate(time.Second)
+			}
+
+			tagSet, found, err := eni(ctx, eniID)
+			if err != nil {
+				row.Unreadable, row.Err, row.Reason = true, err, "control-plane ENI read failed"
+				rows = append(rows, row)
+				continue
+			}
+			if !found || tagSet[tags.ManagedByKey] != tags.ManagedByEKS || tagSet[clusterEKSRoleTagKey] != clusterEKSRoleControlPlane {
+				continue
+			}
+			row.Cluster, row.Account = tagSet[clusterEKSClusterTagKey], tagSet[clusterEKSAccountTagKey]
+			if row.Cluster == "" || row.Account == "" {
+				continue
+			}
+
+			verdict := JudgeClusterRecord(ctx, js, row.Account, row.Cluster)
+			switch verdict.State {
+			case ClusterRecordPresent:
+				continue
+			case ClusterRecordUnreadable:
+				row.Unreadable, row.Err = true, verdict.Err
+			}
+			row.Reason = verdict.Reason
+			rows = append(rows, row)
+		}
+	}
+	sort.Slice(rows, func(i, j int) bool { return rows[i].InstanceID < rows[j].InstanceID })
+	return rows
+}
+
+// ReportControlPlaneOrphans writes the orphan report. list must return every
+// non-terminated system instance or an error: a failed or partial listing is
+// returned as an error and nothing is printed, so a short list is never taken
+// for a complete one. Unreadable rows are printed and counted separately.
+func ReportControlPlaneOrphans(ctx context.Context, w io.Writer, js jetstream.JetStream, list func(context.Context) (*ec2.DescribeInstancesOutput, error), eni ENITagReader, now time.Time) (orphans, unreadable int, err error) {
+	out, err := list(ctx)
+	if err != nil {
+		return 0, 0, fmt.Errorf("listing incomplete: %w", err)
+	}
+	if out == nil {
+		return 0, 0, errors.New("listing incomplete: no result")
+	}
+	rows := FindControlPlaneOrphans(ctx, js, out.Reservations, eni, now)
+	for _, r := range rows {
+		if r.Unreadable {
+			unreadable++
+			fmt.Fprintf(w, "UNREADABLE  %s  cluster=%s account=%s az=%s age=%s  %s: %v\n", r.InstanceID, r.Cluster, r.Account, r.AvailabilityZone, r.Age, r.Reason, r.Err)
+			continue
+		}
+		orphans++
+		fmt.Fprintf(w, "ORPHAN      %s  cluster=%s account=%s az=%s age=%s  %s\n", r.InstanceID, r.Cluster, r.Account, r.AvailabilityZone, r.Age, r.Reason)
+	}
+	return orphans, unreadable, nil
+}
+
+func instanceRunning(inst *ec2.Instance) bool {
+	return inst.State != nil && aws.StringValue(inst.State.Name) == ec2.InstanceStateNameRunning
+}

spinifex/domains/eks/cluster/cp_orphans_test.go

@@ -0,0 +1,198 @@
+package cluster
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	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/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func orphanInstance(id, eni string) *ec2.Instance {
+	return &ec2.Instance{
+		InstanceId:        aws.String(id),
+		State:             &ec2.InstanceState{Name: aws.String(ec2.InstanceStateNameRunning)},
+		LaunchTime:        aws.Time(time.Unix(0, 0)),
+		Placement:         &ec2.Placement{AvailabilityZone: aws.String("ap-southeast-2a")},
+		NetworkInterfaces: []*ec2.InstanceNetworkInterface{{NetworkInterfaceId: aws.String(eni)}},
+	}
+}
+
+func cpTags(cluster, account string) map[string]string {
+	return map[string]string{
+		tags.ManagedByKey:       tags.ManagedByEKS,
+		clusterEKSClusterTagKey: cluster,
+		clusterEKSAccountTagKey: account,
+		clusterEKSRoleTagKey:    clusterEKSRoleControlPlane,
+	}
+}
+
+func workerTags(cluster, account string) map[string]string {
+	m := cpTags(cluster, account)
+	m[clusterEKSRoleTagKey] = "worker"
+	return m
+}
+
+func TestFindControlPlaneOrphansUnreadableRecordIsNotAnOrphan(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	js := testutil.NewJetStream(t, f.svc.deps.NATSConn)
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	eni := func(context.Context, string) (map[string]string, bool, error) {
+		return cpTags("gone", testAccountID), true, nil
+	}
+	res := []*ec2.Reservation{{Instances: []*ec2.Instance{orphanInstance("i-1", "eni-1")}}}
+
+	rows := FindControlPlaneOrphans(ctx, js, res, eni, time.Unix(60, 0))
+
+	require.Len(t, rows, 1)
+	assert.True(t, rows[0].Unreadable, "a read that failed is not a missing record")
+	assert.Error(t, rows[0].Err)
+}
+
+func TestFindControlPlaneOrphans(t *testing.T) {
+	const lost = "999999999999"
+	f := newEKSServiceFixture(t)
+	js := testutil.NewJetStream(t, f.svc.deps.NATSConn)
+	require.NoError(t, PutClusterMeta(t.Context(), f.kv, sampleClusterMeta("alive")))
+
+	eniTags := map[string]map[string]string{
+		"eni-orphan":   cpTags("gone", testAccountID),
+		"eni-live":     cpTags("alive", testAccountID),
+		"eni-nobucket": cpTags("lost", lost),
+		"eni-worker":   workerTags("gone", testAccountID),
+	}
+	eni := func(_ context.Context, id string) (map[string]string, bool, error) {
+		if id == "eni-broken" {
+			return nil, false, errors.New("vpc service down")
+		}
+		m, ok := eniTags[id]
+		return m, ok, nil
+	}
+	res := []*ec2.Reservation{{Instances: []*ec2.Instance{
+		orphanInstance("i-orphan", "eni-orphan"),
+		orphanInstance("i-live", "eni-live"),
+		orphanInstance("i-nobucket", "eni-nobucket"),
+		orphanInstance("i-worker", "eni-worker"),
+		orphanInstance("i-ghost", "eni-ghost"),
+		orphanInstance("i-broken", "eni-broken"),
+		{InstanceId: aws.String("i-noeni")},
+	}}}
+
+	rows := FindControlPlaneOrphans(t.Context(), js, res, eni, time.Unix(60, 0))
+
+	got := map[string]ControlPlaneRow{}
+	for _, r := range rows {
+		got[r.InstanceID] = r
+	}
+	assert.Len(t, got, 3, "only orphan, nobucket and the unreadable ENI are reported")
+	assert.Equal(t, "cluster record missing", got["i-orphan"].Reason)
+	assert.False(t, got["i-orphan"].Unreadable)
+	assert.Equal(t, "account bucket missing", got["i-nobucket"].Reason)
+	assert.True(t, got["i-broken"].Unreadable, "a failed ENI read is never dropped")
+	assert.Error(t, got["i-broken"].Err)
+	assert.NotContains(t, got, "i-live", "a present record is omitted")
+	assert.NotContains(t, got, "i-worker")
+	assert.NotContains(t, got, "i-ghost")
+	assert.NotContains(t, got, "i-noeni")
+}
+
+func TestReportControlPlaneOrphansIncompleteListingPrintsNothing(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	js := testutil.NewJetStream(t, f.svc.deps.NATSConn)
+	var buf bytes.Buffer
+	list := func(context.Context) (*ec2.DescribeInstancesOutput, error) {
+		return nil, errors.New("ServiceUnavailable")
+	}
+	orphans, unreadable, err := ReportControlPlaneOrphans(t.Context(), &buf, js, list, nil, time.Now())
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "listing incomplete")
+	assert.Zero(t, orphans+unreadable)
+	assert.Empty(t, buf.String(), "no partial list on a failed listing")
+}
+
+func TestReportControlPlaneOrphansPrintsRows(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	js := testutil.NewJetStream(t, f.svc.deps.NATSConn)
+	var buf bytes.Buffer
+	list := func(context.Context) (*ec2.DescribeInstancesOutput, error) {
+		return &ec2.DescribeInstancesOutput{Reservations: []*ec2.Reservation{{Instances: []*ec2.Instance{
+			orphanInstance("i-orphan", "eni-orphan"), orphanInstance("i-broken", "eni-broken"),
+		}}}}, nil
+	}
+	eni := func(_ context.Context, id string) (map[string]string, bool, error) {
+		if id == "eni-broken" {
+			return nil, false, errors.New("boom")
+		}
+		return cpTags("gone", testAccountID), true, nil
+	}
+	orphans, unreadable, err := ReportControlPlaneOrphans(t.Context(), &buf, js, list, eni, time.Unix(60, 0))
+	require.NoError(t, err)
+	assert.Equal(t, 1, orphans)
+	assert.Equal(t, 1, unreadable)
+	assert.Contains(t, buf.String(), "ORPHAN      i-orphan")
+	assert.Contains(t, buf.String(), "UNREADABLE  i-broken")
+	assert.Equal(t, 2, strings.Count(buf.String(), "az=ap-southeast-2a"), "both row kinds print the zone")
+}
+
+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)
+	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)
+	require.NoError(t, err)
+	_, err = enis.Put(t.Context(), kvutil.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"),
+	}}}
+
+	rows := FindControlPlaneOrphans(t.Context(), js, res, ec2inventory.NewENITagReader(js, "000000000000"), time.Unix(60, 0))
+
+	require.Len(t, rows, 2, "an absent ENI is skipped, the other two are reported")
+	assert.Equal(t, "i-corrupt", rows[0].InstanceID)
+	assert.True(t, rows[0].Unreadable, "a corrupt ENI record is unreadable, never gone")
+	assert.Equal(t, "i-good", rows[1].InstanceID)
+	assert.False(t, rows[1].Unreadable)
+}
+
+func TestFindControlPlaneOrphansOnlyReportsRunningInstances(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	js := testutil.NewJetStream(t, f.svc.deps.NATSConn)
+	eni := func(context.Context, string) (map[string]string, bool, error) {
+		return cpTags("gone", testAccountID), true, nil
+	}
+	mk := func(id, state string) *ec2.Instance {
+		i := orphanInstance(id, "eni-"+id)
+		i.State = &ec2.InstanceState{Name: aws.String(state)}
+		return i
+	}
+	noState := orphanInstance("i-nostate", "eni-nostate")
+	noState.State = nil
+	res := []*ec2.Reservation{{Instances: []*ec2.Instance{
+		mk("i-running", ec2.InstanceStateNameRunning),
+		mk("i-stopped", ec2.InstanceStateNameStopped),
+		mk("i-pending", ec2.InstanceStateNamePending),
+		noState,
+	}}}
+
+	rows := FindControlPlaneOrphans(t.Context(), js, res, eni, time.Unix(60, 0))
+
+	require.Len(t, rows, 1)
+	assert.Equal(t, "i-running", rows[0].InstanceID)
+}

spinifex/domains/eks/cluster/eks_billable_reaper.go

@@ -2,7 +2,6 @@ package cluster
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"log/slog"
 
@@ -13,13 +12,10 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 )
 
-// EKSBillableReaper is the meta-independent billable cleanup from historic PROP-LIFECYCLE-006 §5 (behaviour per ADR-0003:S2), built on
-// the resource lifecycle GC backstop. A normal DeleteCluster can leave a
-// running control-plane VM behind (terminate completes async; a daemon restart
-// re-launches it from saved running-state) after the cluster meta is already
-// swept — a billable orphan with no owner to drive a retry. This reaper finds a
-// running EKS control-plane VM whose cluster meta is DEFINITIVELY GONE and
-// terminates it. It is node-local: each node reaps the orphans it runs.
+// EKSBillableReaper raises the data-safety alarm for a running EKS control-plane
+// VM whose cluster record cannot be found. It never terminates anything: a
+// missing record cannot tell a completed DeleteCluster from a lost bucket or an
+// out-of-band delete, so removal is an operator decision. It is node-local.
 type EKSBillableReaper struct {
 	svc *EKSServiceImpl
 	// running returns this node's running VMs. The reaper filters to the EKS
@@ -38,10 +34,11 @@ func (s *EKSServiceImpl) NewBillableReaper(running func() ([]*vm.VM, error)) *EK
 func (r *EKSBillableReaper) Class() string         { return "eks-billable" }
 func (r *EKSBillableReaper) Scope() vm.ReaperScope { return vm.ScopeNodeLocal }
 
-// Sweep terminates every running EKS control-plane VM whose cluster meta is
-// absent. A VM whose cluster still exists (CREATING or DELETING) is left to the
-// cluster's own teardown/retry; an ENI that is gone or untagged is skipped, so
-// the reaper never reaps on uncertainty.
+// Sweep alarms on every running EKS control-plane VM whose cluster record, or
+// whole account bucket, is missing, and removes nothing. A VM whose cluster
+// exists is left to the cluster's own teardown; an ENI that is gone or untagged
+// is skipped, and a read that fails is logged and skipped. It returns 0 because
+// it never removes a VM.
 func (r *EKSBillableReaper) Sweep(ctx context.Context) (int, error) {
 	vms, err := r.running()
 	if err != nil {
@@ -53,11 +50,10 @@ func (r *EKSBillableReaper) Sweep(ctx context.Context) (int, error) {
 		return 0, fmt.Errorf("jetstream: %w", err)
 	}
 
-	reaped := 0
 	for _, v := range vms {
 		select {
 		case <-ctx.Done():
-			return reaped, ctx.Err()
+			return 0, ctx.Err()
 		default:
 		}
 		if v == nil || v.ManagedBy != tags.ManagedByEKS || v.ENIId == "" {
@@ -66,47 +62,23 @@ func (r *EKSBillableReaper) Sweep(ctx context.Context) (int, error) {
 
 		clusterName, clusterAccount, ok := r.clusterRefFromENI(ctx, v)
 		if !ok {
-			continue // ENI gone or untagged: cannot confirm orphan-hood, never reap
+			continue // ENI gone or untagged: cannot tie the VM to a cluster
 		}
 
-		acctKV, err := GetOrCreateAccountBucket(ctx, js, clusterAccount)
-		if err != nil {
-			slog.WarnContext(ctx, "eks-billable: account bucket lookup failed", "account", clusterAccount, "err", err)
-			continue
-		}
-		if _, err := GetClusterMeta(ctx, acctKV, clusterName); !errors.Is(err, ErrClusterNotFound) {
-			// Meta present (live/CREATING/DELETING) or unreadable: not a definitive
-			// orphan — leave it to the cluster's own teardown/retry.
+		verdict := JudgeClusterRecord(ctx, js, clusterAccount, clusterName)
+		switch verdict.State {
+		case ClusterRecordPresent:
+			continue // cluster exists (live, CREATING or DELETING)
+		case ClusterRecordUnreadable:
+			slog.WarnContext(ctx, "eks-billable: "+verdict.Reason, "cluster", clusterName, "account", clusterAccount, "err", verdict.Err)
 			continue
 		}
+		reason := verdict.Reason
 
-		// ALARM + reap: the cluster is gone but its control-plane VM still runs.
-		slog.WarnContext(ctx, "DATA-SAFETY ALARM: orphaned EKS control-plane VM reaped (cluster meta gone)",
-			"instanceId", v.ID, "cluster", clusterName, "account", clusterAccount, "node", v.LastNode)
-		if err := TerminateK3sServerVM(ctx, r.svc.deps.VPCK3s, r.svc.deps.Instance, v.AccountID, v.ID, v.ENIId); err != nil {
-			slog.WarnContext(ctx, "eks-billable: failed to terminate orphan CP VM", "instanceId", v.ID, "err", err)
-			continue
-		}
-		reaped++
-		r.reclaimOrphanInfra(ctx, v.AccountID, clusterName)
-	}
-	return reaped, nil
-}
-
-// reclaimOrphanInfra tears down the billable infra a reaped orphan's swept meta
-// no longer anchors: the NLB front-end NAT and the managed CP VPC's NAT-GW EIP.
-// Both are keyed by cluster name and idempotent, so a clean prior delete is a
-// no-op. Best-effort — the costliest resource (the VM) is already gone, and
-// DeleteClusterCPVPC releases the NAT-GW EIP before the VPC delete, so the
-// billable address is reclaimed even if the now-empty VPC delete trails. account
-// is the infra account (system account for the managed CP VPC topology).
-func (r *EKSBillableReaper) reclaimOrphanInfra(ctx context.Context, account, clusterName string) {
-	if err := DeleteClusterNLB(ctx, r.svc.deps.NLB, account, clusterName); err != nil {
-		slog.WarnContext(ctx, "eks-billable: reclaim orphan NLB failed", "cluster", clusterName, "err", err)
-	}
-	if err := DeleteClusterCPVPC(ctx, r.svc.cpVPCDeps(), account, clusterName, nil); err != nil {
-		slog.WarnContext(ctx, "eks-billable: reclaim orphan CP VPC (NAT-GW EIP) failed", "cluster", clusterName, "err", err)
+		slog.WarnContext(ctx, "DATA-SAFETY ALARM: EKS control-plane VM has no cluster record; retained for an operator",
+			"reason", reason, "instanceId", v.ID, "cluster", clusterName, "account", clusterAccount, "node", v.LastNode)
 	}
+	return 0, nil
 }
 
 // clusterRefFromENI reads the cluster name + customer account from the VM's

spinifex/domains/eks/cluster/eks_billable_reaper_test.go

@@ -1,14 +1,19 @@
 package cluster
 
 import (
+	"bytes"
 	"context"
+	"log/slog"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"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/runtime/compute/vm"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -33,53 +38,119 @@ func cpVM(id, eniID string) *vm.VM {
 	return &vm.VM{ID: id, ManagedBy: tags.ManagedByEKS, ENIId: eniID, AccountID: reaperSysAccount, LastNode: "node-1"}
 }
 
-// TestRLC5_EKSBillableReaperTerminatesOrphanCPVM exercises the historic PROP-LIFECYCLE-006 §5 (behaviour per ADR-0003:S2)
-// meta-independent billable cleanup: a running EKS control-plane VM whose
-// cluster meta is DEFINITIVELY GONE is a billable orphan and must be terminated
-// by the GC backstop — the real fix for the orphan CP VM surviving a
-// daemon restart after DeleteCluster swept the meta.
-func TestRLC5_EKSBillableReaperTerminatesOrphanCPVM(t *testing.T) {
-	t.Parallel()
-	f := newEKSServiceFixture(t)
-	f.vpc.describeByENI = map[string]*ec2.NetworkInterface{
-		"eni-orphan": cpENI("eni-orphan", "gone-cluster", testAccountID),
+// ADR-0003:S7. A running control-plane VM whose cluster record cannot be found
+// is retained and alarmed, never terminated and never has its NLB or VPC torn
+// down: a missing record cannot tell a completed DeleteCluster from a lost
+// bucket or an out-of-band delete.
+func TestS7_EKSBillableReaperNeverTerminatesAControlPlane(t *testing.T) {
+	const lostAccount = "999999999999"
+
+	cases := []struct {
+		name      string
+		account   string
+		prepare   func(t *testing.T, f *eksServiceFixture)
+		wantAlarm string
+		wantLog   string
+	}{
+		{
+			name:      "account bucket missing",
+			account:   lostAccount,
+			prepare:   func(*testing.T, *eksServiceFixture) {},
+			wantAlarm: "account bucket missing",
+		},
+		{
+			name:      "bucket has no meta key",
+			account:   testAccountID,
+			prepare:   func(*testing.T, *eksServiceFixture) {},
+			wantAlarm: "cluster record missing",
+		},
+		{
+			name:    "meta removed by the teardown revision-checked delete",
+			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{
+					Attempts: maxClusterStateCASRetries,
+					NotFound: ErrClusterNotFound,
+				}, func(*ClusterMeta) error { return nil })
+				require.NoError(t, err)
+			},
+			wantAlarm: "cluster record missing",
+		},
+		{
+			name:    "meta removed out of band",
+			account: testAccountID,
+			prepare: func(t *testing.T, f *eksServiceFixture) {
+				require.NoError(t, PutClusterMeta(t.Context(), f.kv, sampleClusterMeta("orphan")))
+				require.NoError(t, f.kv.Delete(t.Context(), ClusterMetaKey("orphan")))
+			},
+			wantAlarm: "cluster record missing",
+		},
+		{
+			name:    "meta unreadable",
+			account: testAccountID,
+			prepare: func(t *testing.T, f *eksServiceFixture) {
+				_, err := f.kv.Put(t.Context(), ClusterMetaKey("orphan"), []byte("{not json"))
+				require.NoError(t, err)
+			},
+			wantLog: "cluster record read failed",
+		},
 	}
-	orphan := cpVM("i-orphan", "eni-orphan")
 
-	// Seed the orphan cluster's name/tag-driven billable infra the swept meta no
-	// longer anchors: an NLB front-end and a tagged NAT gateway holding a billable
-	// EIP. The reaper's post-terminate reclaim must sweep both.
-	nlbName := ClusterNLBName("gone-cluster")
-	f.nlb.lbByName[nlbName] = &elbv2.LoadBalancer{
-		LoadBalancerArn:  aws.String("arn:lb-orphan"),
-		LoadBalancerName: aws.String(nlbName),
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			// slog.SetDefault is process-wide, so this test is not parallel.
+			var buf bytes.Buffer
+			restore := slog.Default()
+			slog.SetDefault(slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})))
+			t.Cleanup(func() { slog.SetDefault(restore) })
+
+			f := newEKSServiceFixture(t)
+			f.vpc.describeByENI = map[string]*ec2.NetworkInterface{
+				"eni-cp": cpENI("eni-cp", "orphan", tc.account),
+			}
+			nlbName := ClusterNLBName("orphan")
+			f.nlb.lbByName[nlbName] = &elbv2.LoadBalancer{
+				LoadBalancerArn:  aws.String("arn:lb-orphan"),
+				LoadBalancerName: aws.String(nlbName),
+			}
+			f.ngw.gws = []*fakeCPNatGateway{{
+				id:    "nat-orphan",
+				tags:  map[string]string{clusterEKSClusterTagKey: "orphan", clusterEKSRoleTagKey: cpVPCRoles.NatGW},
+				state: "available",
+				addrs: []*ec2.NatGatewayAddress{{AllocationId: aws.String("eipalloc-orphan")}},
+			}}
+			tc.prepare(t, f)
+
+			reaper := f.svc.NewBillableReaper(func() ([]*vm.VM, error) { return []*vm.VM{cpVM("i-cp", "eni-cp")}, nil })
+			for range 2 {
+				removed, err := reaper.Sweep(context.Background())
+				require.NoError(t, err)
+				assert.Zero(t, removed, "ADR-0003:S7: the reaper removes nothing")
+			}
+
+			assert.Empty(t, f.inst.terminateCalls, "ADR-0003:S7: a reaper's judgement is never the decision to terminate")
+			assert.Empty(t, f.vpc.deleteCalls, "the control-plane ENI must stay")
+			assert.Contains(t, f.nlb.lbByName, nlbName, "the NLB must stay")
+			assert.Empty(t, f.eip.releaseCalls, "no EIP may be released")
+
+			js := testutil.NewJetStream(t, f.svc.deps.NATSConn)
+			if tc.account == lostAccount {
+				_, err := OpenAccountBucket(t.Context(), js, lostAccount)
+				assert.ErrorIs(t, err, jetstream.ErrBucketNotFound, "a sweep must not create the bucket it found missing")
+			}
+
+			out := buf.String()
+			if tc.wantAlarm != "" {
+				assert.Contains(t, out, "DATA-SAFETY ALARM")
+				assert.Contains(t, out, tc.wantAlarm)
+				assert.Contains(t, out, "i-cp")
+			} else {
+				assert.NotContains(t, out, "DATA-SAFETY ALARM", "an unreadable record is not a missing one")
+				assert.Contains(t, out, tc.wantLog)
+			}
+		})
 	}
-	f.ngw.gws = []*fakeCPNatGateway{{
-		id:    "nat-orphan",
-		tags:  map[string]string{clusterEKSClusterTagKey: "gone-cluster", clusterEKSRoleTagKey: cpVPCRoles.NatGW},
-		state: "available",
-		addrs: []*ec2.NatGatewayAddress{{AllocationId: aws.String("eipalloc-orphan")}},
-	}}
-
-	reaper := f.svc.NewBillableReaper(func() ([]*vm.VM, error) { return []*vm.VM{orphan}, nil })
-
-	reaped, err := reaper.Sweep(context.Background())
-	require.NoError(t, err)
-	assert.Equal(t, 1, reaped, "ADR-0003:S2 behaviour; historic PROP-LIFECYCLE-006 §5 mechanism: a CP VM whose cluster meta is gone must be reaped")
-	assert.Contains(t, f.inst.terminateCalls, "i-orphan", "the orphan CP VM must be terminated")
-	require.Len(t, f.vpc.deleteCalls, 1, "the orphan CP ENI must be deleted")
-	assert.Equal(t, "eni-orphan", aws.StringValue(f.vpc.deleteCalls[0].NetworkInterfaceId))
-
-	// The orphan's billable NLB + NAT-GW EIP must also be reclaimed.
-	assert.NotContains(t, f.nlb.lbByName, nlbName, "the orphan NLB front-end must be reclaimed")
-	require.Len(t, f.eip.releaseCalls, 1, "the orphan NAT-GW EIP must be released")
-	assert.Equal(t, "eipalloc-orphan", aws.StringValue(f.eip.releaseCalls[0].AllocationId))
-
-	// Idempotent: a repeat sweep over the same (still meta-absent) VM is safe —
-	// the terminate + reclaim paths tolerate already-gone infra (no second release).
-	_, err = reaper.Sweep(context.Background())
-	require.NoError(t, err)
-	assert.Len(t, f.eip.releaseCalls, 1, "reclaim must not double-release an already-freed EIP")
 }
 
 // TestRLC5_EKSBillableReaperSpareLiveAndUncertain enforces the reaper's no-false-

spinifex/domains/eks/cluster/store.go

@@ -185,7 +185,8 @@ func AccountWatchBuckets(ctx context.Context, nc *nats.Conn) ([]*kvstore.Bucket,
 	}
 	buckets := make([]*kvstore.Bucket, 0, len(names))
 	for _, name := range names {
-		buckets = append(buckets, kvstore.NewBucket(js, kvstore.Config{Name: name, History: 1}))
+		// Attach only: a watch on a deleted bucket must fail and retry, not recreate it.
+		buckets = append(buckets, kvstore.NewBucket(js, kvstore.Config{Name: name, History: 1, AttachOnly: true}))
 	}
 	return buckets, nil
 }
@@ -206,6 +207,13 @@ func GetOrCreateAccountBucket(ctx context.Context, js jetstream.JetStream, accou
 	return kv, nil
 }
 
+// OpenAccountBucket returns the per-account KV bucket for accountID without
+// creating it. A bucket that does not exist is jetstream.ErrBucketNotFound, so a
+// read-only caller never turns a lost bucket into an empty one.
+func OpenAccountBucket(ctx context.Context, js jetstream.JetStream, accountID string) (jetstream.KeyValue, error) {
+	return js.KeyValue(ctx, AccountBucketName(accountID))
+}
+
 // InitLeaderBucket creates (or attaches to) the shared spinifex-eks-leader
 // bucket used for per-cluster reconciler leader-lease CAS locks.
 func InitLeaderBucket(ctx context.Context, js jetstream.JetStream) (jetstream.KeyValue, error) {

spinifex/domains/eks/cluster/watch_source_test.go

@@ -7,6 +7,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/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -79,3 +80,26 @@ func TestAccountWatchBuckets_EnumerationErrorPropagates(t *testing.T) {
 	_, err := ekscluster.AccountWatchBuckets(ctx, nc)
 	require.Error(t, err)
 }
+
+// TestAccountWatchBuckets_WatchOnDeletedBucketDoesNotRecreateIt pins attach-only
+// handles: a DNS watch that outlives its bucket must fail and retry, because
+// recreating it hides a deleted account bucket behind an empty one.
+func TestAccountWatchBuckets_WatchOnDeletedBucketDoesNotRecreateIt(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	js := testutil.NewJetStream(t, nc)
+	_, err := ekscluster.GetOrCreateAccountBucket(t.Context(), js, "111111111111")
+	require.NoError(t, err)
+
+	buckets, err := ekscluster.AccountWatchBuckets(t.Context(), nc)
+	require.NoError(t, err)
+	require.Len(t, buckets, 1)
+
+	require.NoError(t, js.DeleteKeyValue(t.Context(), ekscluster.AccountBucketName("111111111111")))
+
+	_, err = buckets[0].Watch(t.Context(), ">")
+	require.Error(t, err, "a watch on a deleted bucket must fail")
+
+	_, err = ekscluster.OpenAccountBucket(t.Context(), js, "111111111111")
+	assert.ErrorIs(t, err, jetstream.ErrBucketNotFound, "the watch must not have recreated the bucket")
+}

spinifex/operator/cli/admin_eks.go

@@ -3,11 +3,16 @@ package cli
 import (
 	"context"
 	"fmt"
+	"maps"
 	"os"
+	"slices"
 	"time"
 
+	"github.com/aws/aws-sdk-go/service/ec2"
+	ec2inventory "github.com/mulgadc/spinifex/spinifex/domains/ec2/inventory"
 	ekscluster "github.com/mulgadc/spinifex/spinifex/domains/eks/cluster"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/spf13/cobra"
 )
 
@@ -75,3 +80,57 @@ func runAdminEksRestoreSnapshot(cmd *cobra.Command, _ []string) {
 	fmt.Printf("New control plane %s launched (snapshot %s)\n", out.NewInstanceID, out.Snapshot)
 	fmt.Printf("Status: %s\n", out.Status)
 }
+
+var adminEksOrphansCmd = &cobra.Command{
+	Use:   "orphans",
+	Short: "List running EKS control-plane VMs that have no cluster record",
+	Long: `orphans reports each running EKS control-plane VM whose cluster record, or whole
+account bucket, is missing (ORPHAN), and each it could not judge because a read
+failed (UNREADABLE). It only reads and never creates or removes anything.
+
+It lists the system account's instances through the strict inventory, so if any
+configured node does not answer it prints "listing incomplete" and exits non-zero
+without a partial list. It cannot say which node was silent. An instance with no
+ENI, or whose ENI is gone or not tagged as an EKS control plane, is skipped.
+
+Exit status is non-zero when the listing is incomplete or any row is ORPHAN or
+UNREADABLE.`,
+	Run: runAdminEksOrphans,
+}
+
+func init() {
+	adminEksCmd.AddCommand(adminEksOrphansCmd)
+}
+
+func runAdminEksOrphans(_ *cobra.Command, _ []string) {
+	cfg, nc, err := loadConfigAndConnect()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+	defer nc.Close()
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+	nodeIDs := slices.Sorted(maps.Keys(cfg.Nodes))
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	list := func(ctx context.Context) (*ec2.DescribeInstancesOutput, error) {
+		return ec2inventory.NewNATSInstances(nc, nodeIDs).ListAccountInstances(ctx, awsidentifiers.GlobalAccountID)
+	}
+	eni := ekscluster.ENITagReader(ec2inventory.NewENITagReader(js, awsidentifiers.GlobalAccountID))
+
+	orphans, unreadable, err := ekscluster.ReportControlPlaneOrphans(ctx, os.Stdout, js, list, eni, time.Now())
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+	fmt.Printf("%d orphan(s), %d unreadable\n", orphans, unreadable)
+	if orphans+unreadable > 0 {
+		os.Exit(1)
+	}
+}

spinifex/runtime/compute/lambda/testlink/daemon.go

@@ -83,6 +83,9 @@ type Daemon struct {
 	MTU     int
 	WorkDir string
 	HTTP    *http.Client
+	// SDKEndpoint and SDKCA are public gateway connection settings for the guest.
+	SDKEndpoint string
+	SDKCA       []byte
 	// Image is the prepared runtime image every launch boots; the daemon
 	// prepares it once before serving and keeps it open for its whole life.
 	Image *Image
@@ -582,6 +585,10 @@ func (d *Daemon) prepare(req Request, code []byte, dir, tap, mac, ip string) (*v
 	entries["test-package-sha256"] = []byte(req.CodeSHA256)
 	entries["test-grant"] = grantEntry(req.Grant)
 	entries["test-handler"] = []byte(req.Handler)
+	if d.SDKEndpoint != "" && len(d.SDKCA) != 0 {
+		entries["test-sdk-endpoint"] = []byte(d.SDKEndpoint)
+		entries["test-sdk-ca"] = d.SDKCA
+	}
 	netcfgBytes := fmt.Appendf(nil, "NIC0_MAC=%s\nNIC0_CIDR=%s/24\nNIC0_DEFAULT=0\n", mac, ip)
 	if d.Firecracker != "" {
 		archive := make(map[string][]byte, len(entries)+1)

spinifex/runtime/compute/lambda/testlink/testlink_test.go

@@ -368,6 +368,36 @@ func TestPrepare_ScopeAndGrantEntryComeFromTheGrant(t *testing.T) {
 	require.Equal(t, "ap-southeast-2/123456789012/fn1/env-x/3", strings.TrimSpace(string(expect)))
 }
 
+func TestPrepare_DeliversSDKSettingsForEachVMM(t *testing.T) {
+	for _, firecracker := range []bool{false, true} {
+		t.Run(fmt.Sprintf("firecracker=%t", firecracker), func(t *testing.T) {
+			req := baseRequest("https://example.invalid/code", []byte("zip"))
+			req.Grant = execution.Grant{GrantID: "g-7", AttemptID: req.AttemptID, EnvironmentID: "env-x", Generation: 3, Node: testNode}
+			d := &Daemon{HostIP: "127.0.0.1", SDKEndpoint: "https://10.15.8.1:9999", SDKCA: []byte("public-ca")}
+			if firecracker {
+				d.Firecracker = "/test/firecracker"
+			}
+			dir := filepath.Join(t.TempDir(), "attempt")
+			_, _, ln, err := d.prepare(req, []byte("zip"), dir, "tap0", "02:00:00:00:00:01", "10.0.0.2")
+			require.NoError(t, err)
+			defer ln.Close()
+			endpointPath, caPath := filepath.Join(dir, "fwcfg-test-sdk-endpoint"), filepath.Join(dir, "fwcfg-test-sdk-ca")
+			if firecracker {
+				root := filepath.Join(t.TempDir(), "config")
+				extract(t, filepath.Join(dir, configArchive), root)
+				entryDir := filepath.Join(root, "by_name", lambdav1.FwCfgPrefix)
+				endpointPath, caPath = filepath.Join(entryDir, "test-sdk-endpoint", "raw"), filepath.Join(entryDir, "test-sdk-ca", "raw")
+			}
+			endpoint, err := os.ReadFile(endpointPath)
+			require.NoError(t, err)
+			ca, err := os.ReadFile(caPath)
+			require.NoError(t, err)
+			require.Equal(t, d.SDKEndpoint, string(endpoint))
+			require.Equal(t, d.SDKCA, ca)
+		})
+	}
+}
+
 // elfHeader is a bare 64-bit little-endian shared-object header for machine.
 func elfHeader(machine elf.Machine) []byte {
 	h := make([]byte, 64)

spinifex/runtime/daemon/daemon_lambda_testlink.go

@@ -7,6 +7,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/hex"
+	"encoding/pem"
 	"errors"
 	"fmt"
 	"io"
@@ -82,15 +83,21 @@ func (d *Daemon) startLambdaTestLink() {
 		return
 	}
 	httpClient, trustErr := d.lambdaTestLinkHTTP()
+	sdkEndpoint, sdkCA, sdkErr := d.lambdaTestLinkSDKConfig()
+	if sdkErr != nil {
+		slog.Warn("Lambda test link: SDK gateway settings unavailable", "err", sdkErr)
+	}
 	imageDir := filepath.Join(microvmDir, "lambda-testlink")
 	tl := &testlink.Daemon{
-		Node:    d.node,
-		Grants:  grants,
-		Taps:    taps,
-		IPs:     d.mgmtIPAllocator,
-		Bridge:  d.mgmtBridgeName(),
-		HostIP:  d.mgmtBridgeIP,
-		WorkDir: workDir,
+		Node:        d.node,
+		Grants:      grants,
+		Taps:        taps,
+		IPs:         d.mgmtIPAllocator,
+		Bridge:      d.mgmtBridgeName(),
+		HostIP:      d.mgmtBridgeIP,
+		WorkDir:     workDir,
+		SDKEndpoint: sdkEndpoint,
+		SDKCA:       sdkCA,
 
 		Initializer:       filepath.Join(imageDir, "hardkill.so"),
 		InitializerSHA256: lambdaHardKillSHA256,
@@ -135,7 +142,6 @@ func (d *Daemon) startLambdaTestLink() {
 // missed, by a crash or an earlier refusal, waits for the next pass.
 const lambdaActivationInterval = 30 * time.Second
 
-
 // lambdaTestLinkReady runs the serving checks and, only once they pass,
 // prepares the pinned image into tl; a refusal leaves tl.Image nil.
 func lambdaTestLinkReady(tl *testlink.Daemon, imageDir, imagePin string, trustErr error) error {
@@ -201,4 +207,32 @@ func (d *Daemon) lambdaTestLinkHTTP() (*http.Client, error) {
 	}}, nil
 }
 
+// lambdaTestLinkSDKConfig returns public settings for the guest's AWS SDK.
+func (d *Daemon) lambdaTestLinkSDKConfig() (string, []byte, error) {
+	if d.mgmtBridgeIP == "" || d.config.NATS.CACert == "" {
+		return "", nil, errors.New("management address or cluster CA not configured")
+	}
+	ca, err := os.ReadFile(d.config.NATS.CACert)
+	if err != nil {
+		return "", nil, errors.New("cluster CA unreadable")
+	}
+	if len(ca) > 64<<10 || !x509.NewCertPool().AppendCertsFromPEM(ca) {
+		return "", nil, errors.New("cluster CA invalid or too large")
+	}
+	var public []byte
+	for rest := ca; len(rest) != 0; {
+		block, next := pem.Decode(rest)
+		if block == nil {
+			break
+		}
+		rest = next
+		if block.Type == "CERTIFICATE" {
+			if _, err := x509.ParseCertificate(block.Bytes); err == nil {
+				public = append(public, pem.EncodeToMemory(block)...)
+			}
+		}
+	}
+	return d.resolveSystemGatewayBaseURL(), public, nil
+}
+
 var _ testlink.Taps = (*host.OVSPlumber)(nil)

spinifex/runtime/daemon/lambda_sdk_config_test.go

@@ -0,0 +1,44 @@
+//go:build lambdatestlink
+
+//test:in-package exercises daemon-owned gateway configuration
+package daemon
+
+import (
+	"encoding/pem"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLambdaSDKConfigUsesManagementGatewayAndClusterCA(t *testing.T) {
+	server := httptest.NewTLSServer(nil)
+	defer server.Close()
+	ca := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw})
+	path := filepath.Join(t.TempDir(), "ca.pem")
+	require.NoError(t, os.WriteFile(path, ca, 0o600))
+	d := &Daemon{mgmtBridgeIP: "10.15.8.1", config: &config.Config{
+		NATS: config.NATSConfig{CACert: path}, AWSGW: config.AWSGWConfig{Host: "0.0.0.0:9999"},
+	}}
+	endpoint, gotCA, err := d.lambdaTestLinkSDKConfig()
+	require.NoError(t, err)
+	assert.Equal(t, "https://10.15.8.1:9999", endpoint)
+	assert.Equal(t, ca, gotCA)
+
+	withKey := append(append([]byte{}, ca...), pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: []byte("private")})...)
+	withKey = append(withKey, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: []byte("not a certificate")})...)
+	require.NoError(t, os.WriteFile(path, withKey, 0o600))
+	_, gotCA, err = d.lambdaTestLinkSDKConfig()
+	require.NoError(t, err)
+	assert.Equal(t, ca, gotCA, "only public certificates reach the guest")
+
+	require.NoError(t, os.WriteFile(path, []byte("bad CA"), 0o600))
+	_, _, err = d.lambdaTestLinkSDKConfig()
+	require.Error(t, err)
+	_, err = d.lambdaTestLinkHTTP()
+	require.Error(t, err, "configured invalid download trust must still refuse serving")
+}