fix(eks): backoff and terminal state on delete re-drive, CSI volume reclaim reporting, independent e2e subtests (#642)

* fix: bound EKS DELETING re-drive with backoff and a terminal give-up

reapCluster re-drove purgeClusterInfra on every GC tick with no attempt
tracking, so a permanently-failing teardown (e.g. an unretriable
DependencyViolation) was retried forever. Track attempt count and last
attempt time on ClusterMeta, back off exponentially between re-drives,
and give up after maxDeleteReapAttempts with a single error-level log.
Status stays DELETING on give-up so the cluster's infra remains tracked
and billable pending operator intervention, preserving the ADR-0006 §6
billable-before-sweep invariant.

* fix: create the EKS e2e node role independently in each subtest

runEBSCSIVolume string-built the worker node role ARN without ever
creating it, only passing because runIRSAPod's t.Cleanup discarded
DeleteRole's error while the role was still attached to its instance
profile, leaving it behind for EBSCSIVolume to reuse by naming
convention. Extract ensureNodeRole and call it independently from both
subtests, detaching the role from its instance profile before deleting
it and asserting that error instead of discarding it.

* fix: reclaim CSI-provisioned EBS volumes on EKS cluster delete

purgeClusterInfra tore down the control plane, nodegroups, and
networking but never looked at EBS volumes, so any volume the
in-cluster CSI driver had provisioned for a PVC was left behind
untracked and unbilled for. Add a report-only reclaim step that scans
for volumes tagged with the cluster's CSI owner tag and logs them for
operator follow-up.

Volumes are never auto-deleted: a PV's reclaim policy (Delete vs
Retain) is a Kubernetes-side property the CSI driver does not tag onto
the EC2 volume, so there is no safe way to tell from tags alone
whether deleting a given volume is intended. This mirrors the
volume-leak reaper's existing mark-and-alarm policy for the same
reason.

The cluster-owner tag requires the EBS CSI controller's
--k8s-tag-cluster-id flag, added here to the controller manifest;
without it a CSI volume carries no cluster-scoped tag and reclaim
would have to guess across every cluster sharing the account.

* fix: persist the last purge error on a wedged DELETING cluster

DeleteReapExhausted told an operator a cluster's teardown had given
up, but not why: nothing recorded the failing purge's error.
Persist it on every failed re-drive, not only once the reaper
exhausts its attempts, so it is visible on a still-retrying cluster
too. Status stays DELETING throughout, preserving the billing
invariant that untorn-down infra remains tracked.

* fix: make the EKS e2e suite confirm its CSI volume is actually deleted

The EBSCSIVolume subtest deleted its PVC on cleanup but never
confirmed the underlying volume was reclaimed, so a CSI driver that
silently failed to delete it would leak a volume with nobody
noticing. The suite provisions and owns this volume, so unlike
production teardown it can safely wait on its StorageClass's Delete
reclaim policy: capture the CSI volumeHandle while the PVC is bound,
then have cleanup poll EC2 for that exact volume to disappear before
the addon/nodegroup that run the CSI controller are torn down.

scripts/images/eks-node/addons/aws-ebs-csi-driver/1.40.1/10-controller.yaml

@@ -94,6 +94,7 @@ spec:
         - --batching=true
         - --logging-format=text
         - --user-agent-extra=kustomize
+        - --k8s-tag-cluster-id={{CLUSTER_NAME}}
         - --v=2
         env:
         - name: CSI_ENDPOINT

spinifex/daemon/eks_deps.go

@@ -114,6 +114,14 @@ func (d *Daemon) buildEKSServiceDeps() handlers_eks.EKSServiceDeps {
 	// daemon startup; absent (no master key) workers/CP fall back accordingly.
 	deps.IAMProvider = d.systemRoleEnsurer
 
+	// Guarded rather than assigned unconditionally in the struct literal: a nil
+	// *VolumeServiceImpl assigned into the csiVolumeReclaimer interface field
+	// would be a non-nil interface wrapping a nil pointer, defeating
+	// reclaimCSIVolumes' own nil check.
+	if d.volumeService != nil {
+		deps.Volume = d.volumeService
+	}
+
 	return deps
 }
 

spinifex/handlers/eks/cluster_state.go

@@ -98,6 +98,24 @@ type ClusterMeta struct {
 	// teardown, never one still in progress. No omitempty: encoding/json never
 	// treats a time.Time as empty.
 	DeletingSince time.Time `json:"deletingSince"`
+	// DeleteReapAttempts counts backstop-reaper re-drive attempts against this
+	// DELETING cluster since DeletingSince. Drives the reaper's exponential
+	// backoff and the terminal give-up after maxDeleteReapAttempts.
+	DeleteReapAttempts int `json:"deleteReapAttempts,omitempty"`
+	// LastDeleteReapAttempt stamps the most recent backstop-reaper re-drive, so
+	// backoff grows from the last attempt rather than from DeletingSince.
+	// No omitempty: encoding/json never treats a time.Time as empty.
+	LastDeleteReapAttempt time.Time `json:"lastDeleteReapAttempt"`
+	// DeleteReapExhausted marks a DELETING cluster whose backstop reaper gave up
+	// after maxDeleteReapAttempts. Status stays DELETING — its infra remains
+	// tracked and billable pending operator intervention — only the automatic
+	// re-drive stops.
+	DeleteReapExhausted bool `json:"deleteReapExhausted,omitempty"`
+	// LastDeleteReapError is the error string from the most recent failed
+	// backstop-reaper re-drive, so an operator sees why teardown is stuck
+	// instead of just that it is. Set on every failed attempt, not only on
+	// exhaustion, and cleared once a re-drive succeeds.
+	LastDeleteReapError string `json:"lastDeleteReapError,omitempty"`
 	// HealthIssue is the last health failure reason ("" = healthy).
 	// DescribeCluster surfaces it as a ClusterHealth issue.
 	HealthIssue string `json:"healthIssue,omitempty"`
@@ -261,6 +279,60 @@ func MarkClusterFailed(ctx context.Context, kv jetstream.KeyValue, name, reason
 	})
 }
 
+// maxDeleteReapAttempts caps automatic backstop re-drives of a wedged DELETING
+// cluster. A teardown that still fails after this many attempts is failing for
+// a reason retries will not fix (e.g. an unretriable DependencyViolation), so
+// further re-drives would only keep hammering AWS/OVN forever.
+const maxDeleteReapAttempts = 6
+
+// RecordDeleteReapAttempt increments the backstop reaper's attempt counter and
+// stamps LastDeleteReapAttempt, returning the updated count. Called immediately
+// before each re-drive so a crash mid-purge still counts the attempt on restart.
+func RecordDeleteReapAttempt(ctx context.Context, kv jetstream.KeyValue, name string) (int, error) {
+	if name == "" {
+		return 0, errors.New("eks: RecordDeleteReapAttempt empty name")
+	}
+	var attempts int
+	err := casUpdateMeta(ctx, kv, name, func(m *ClusterMeta) bool {
+		m.DeleteReapAttempts++
+		m.LastDeleteReapAttempt = time.Now().UTC()
+		attempts = m.DeleteReapAttempts
+		return true
+	})
+	return attempts, err
+}
+
+// RecordDeleteReapFailure persists the error from a failed backstop re-drive
+// onto ClusterMeta, so an operator inspecting a wedged DELETING cluster sees
+// why teardown keeps failing instead of just that it is. Called on every
+// failed attempt, not only on eventual exhaustion.
+func RecordDeleteReapFailure(ctx context.Context, kv jetstream.KeyValue, name string, purgeErr error) error {
+	if name == "" {
+		return errors.New("eks: RecordDeleteReapFailure empty name")
+	}
+	return casUpdateMeta(ctx, kv, name, func(m *ClusterMeta) bool {
+		m.LastDeleteReapError = purgeErr.Error()
+		return true
+	})
+}
+
+// MarkDeleteReapExhausted flags a DELETING cluster whose backstop reaper has
+// given up after maxDeleteReapAttempts. Status stays DELETING — a cluster with
+// un-torn-down infra must stay DELETING so its resources remain tracked and
+// billable — only the automatic re-drive stops; an operator must intervene.
+func MarkDeleteReapExhausted(ctx context.Context, kv jetstream.KeyValue, name string) error {
+	if name == "" {
+		return errors.New("eks: MarkDeleteReapExhausted empty name")
+	}
+	return casUpdateMeta(ctx, kv, name, func(m *ClusterMeta) bool {
+		if m.Status != ClusterStatusDeleting || m.DeleteReapExhausted {
+			return false
+		}
+		m.DeleteReapExhausted = true
+		return true
+	})
+}
+
 // SetClusterCertificateAuthority does a CAS update of meta.CertificateAuthorityB64.
 // Called by the bootstrap subscriber when the K3s VM publishes its CA.
 func SetClusterCertificateAuthority(ctx context.Context, kv jetstream.KeyValue, name, caB64 string) error {

spinifex/handlers/eks/eks_deleting_reaper.go

@@ -85,9 +85,21 @@ func (r *EKSDeletingReaper) Sweep(ctx context.Context) (int, error) {
 	return reaped, nil
 }
 
+// deleteReapBackoff returns how long the reaper must wait since the last
+// re-drive attempt before trying again: minAge on the first attempt, doubling
+// after each subsequent failure. priorAttempts is the re-drive count already
+// made (meta.DeleteReapAttempts), so the 2nd attempt waits 2×minAge, the 3rd
+// waits 4×minAge, and so on — a permanently-failing purge backs off instead of
+// re-driving on every GC tick.
+func deleteReapBackoff(minAge time.Duration, priorAttempts int) time.Duration {
+	return minAge * time.Duration(1<<uint(priorAttempts))
+}
+
 // reapCluster re-drives one cluster's teardown if it is wedged in DELETING past
-// minAge and its leader lease can be acquired. Returns 1 when it completed a
-// teardown, 0 otherwise.
+// minAge (and any subsequent backoff) and its leader lease can be acquired.
+// Returns 1 when it completed a teardown, 0 otherwise. After maxDeleteReapAttempts
+// consecutive failures it gives up permanently (DeleteReapExhausted) rather than
+// re-driving a purge that keeps failing the same way forever.
 func (r *EKSDeletingReaper) reapCluster(ctx context.Context, accountID string, acctKV jetstream.KeyValue, cluster string) (int, error) {
 	meta, err := GetClusterMeta(ctx, acctKV, cluster)
 	if err != nil {
@@ -96,9 +108,22 @@ func (r *EKSDeletingReaper) reapCluster(ctx context.Context, accountID string, a
 	if meta.Status != ClusterStatusDeleting {
 		return 0, nil
 	}
+	if meta.DeleteReapExhausted {
+		return 0, nil // backstop already gave up; needs operator intervention
+	}
 	if !meta.DeletingSince.IsZero() && time.Since(meta.DeletingSince) < r.minAge {
 		return 0, nil // still within the in-flight synchronous-delete window
 	}
+	// backoffRef is the last re-drive attempt once one has happened, so the
+	// wait grows attempt-over-attempt instead of always measuring from
+	// DeletingSince (which would let backoff collapse back to minAge forever).
+	backoffRef := meta.DeletingSince
+	if !meta.LastDeleteReapAttempt.IsZero() {
+		backoffRef = meta.LastDeleteReapAttempt
+	}
+	if time.Since(backoffRef) < deleteReapBackoff(r.minAge, meta.DeleteReapAttempts) {
+		return 0, nil // backing off after a recent failed re-drive
+	}
 
 	release, ok := r.svc.acquireTeardownLease(ctx, accountID, cluster)
 	if !ok {
@@ -106,10 +131,28 @@ func (r *EKSDeletingReaper) reapCluster(ctx context.Context, accountID string, a
 	}
 	defer release()
 
+	attempts, err := RecordDeleteReapAttempt(ctx, acctKV, cluster)
+	if err != nil {
+		return 0, fmt.Errorf("record delete reap attempt: %w", err)
+	}
+
 	slog.Warn("eks-deleting: re-driving wedged DELETING teardown",
-		"cluster", cluster, "account", accountID, "deletingSince", meta.DeletingSince)
-	if err := r.svc.purgeClusterInfra(context.Background(), accountID, cluster, meta, acctKV, true); err != nil {
-		return 0, err
+		"cluster", cluster, "account", accountID, "deletingSince", meta.DeletingSince, "attempt", attempts)
+	if purgeErr := r.svc.purgeClusterInfra(context.Background(), accountID, cluster, meta, acctKV, true); purgeErr != nil {
+		// Persisted on every failed attempt, not only once the reaper gives
+		// up, so an operator inspecting a still-retrying cluster already sees
+		// why it is stuck.
+		if recErr := RecordDeleteReapFailure(context.Background(), acctKV, cluster, purgeErr); recErr != nil {
+			return 0, fmt.Errorf("record delete reap failure: %w", recErr)
+		}
+		if attempts >= maxDeleteReapAttempts {
+			slog.Error("eks-deleting: giving up on wedged DELETING teardown after repeated failures",
+				"cluster", cluster, "account", accountID, "attempts", attempts, "err", purgeErr.Error())
+			if markErr := MarkDeleteReapExhausted(context.Background(), acctKV, cluster); markErr != nil {
+				return 0, fmt.Errorf("mark delete reap exhausted: %w", markErr)
+			}
+		}
+		return 0, purgeErr
 	}
 	slog.Info("eks-deleting: teardown completed, meta swept", "cluster", cluster, "account", accountID)
 	return 1, nil

spinifex/handlers/eks/eks_deleting_reaper_test.go

@@ -2,6 +2,7 @@ package handlers_eks
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -74,6 +75,134 @@ func TestDeletingReaper_EnumerationFailureIsReported(t *testing.T) {
 	assert.Zero(t, n)
 }
 
+// TestDeletingReaper_BacksOffAfterFailedAttempt guards against the re-drive
+// re-running on every 2-minute GC tick with no backoff: a permanently-failing
+// purge must not be retried again until the exponential backoff window (2×
+// minAge after 1 prior attempt) has elapsed, and the attempt is persisted so
+// the window survives a restart. No sleeping — the clock is advanced by
+// backdating LastDeleteReapAttempt directly.
+func TestDeletingReaper_BacksOffAfterFailedAttempt(t *testing.T) {
+	f := newDeleteClusterFixture(t, "alpha")
+	f.inst.terminateErr = errors.New("hypervisor unreachable")
+	markDeleting(t, f, "alpha", 10*time.Minute)
+
+	reaper := f.svc.NewDeletingReaper()
+
+	n, err := reaper.Sweep(context.Background())
+	require.NoError(t, err, "Sweep logs and continues past a single cluster's re-drive failure")
+	assert.Equal(t, 0, n)
+	assert.Len(t, f.inst.terminateCalls, 1, "the first re-drive attempt must run")
+
+	meta, getErr := GetClusterMeta(t.Context(), f.kv, "alpha")
+	require.NoError(t, getErr)
+	assert.Equal(t, 1, meta.DeleteReapAttempts, "the attempt must be counted")
+	assert.False(t, meta.LastDeleteReapAttempt.IsZero(), "the attempt time must be stamped")
+
+	// Immediately re-sweeping must be a no-op: still inside the backoff window.
+	n, err = reaper.Sweep(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, 0, n)
+	assert.Len(t, f.inst.terminateCalls, 1, "must not re-drive again inside the backoff window")
+
+	// Advance past the backoff window without sleeping, by backdating the
+	// last-attempt timestamp; the reaper must then re-drive again.
+	meta.LastDeleteReapAttempt = time.Now().UTC().Add(-2 * deleteReapBackoff(deletingReapMinAge, meta.DeleteReapAttempts))
+	require.NoError(t, PutClusterMeta(t.Context(), f.kv, meta))
+
+	n, err = reaper.Sweep(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, 0, n)
+	assert.Len(t, f.inst.terminateCalls, 2, "must re-drive again once the backoff window elapses")
+
+	meta, getErr = GetClusterMeta(t.Context(), f.kv, "alpha")
+	require.NoError(t, getErr)
+	assert.Equal(t, 2, meta.DeleteReapAttempts)
+}
+
+// TestDeletingReaper_ExhaustsAfterMaxAttempts guards the terminal give-up: a
+// purge that keeps failing the same way forever (e.g. an unretriable
+// DependencyViolation) must stop being re-driven after maxDeleteReapAttempts,
+// not loop forever on every GC tick. It also locks the ADR-0006 §6 billing
+// invariant: an exhausted cluster must stay DELETING — its infra stays tracked
+// and billable — not silently vanish or move to some other status.
+func TestDeletingReaper_ExhaustsAfterMaxAttempts(t *testing.T) {
+	f := newDeleteClusterFixture(t, "alpha")
+	f.inst.terminateErr = errors.New("permanent hypervisor failure")
+	markDeleting(t, f, "alpha", 10*time.Minute)
+
+	reaper := f.svc.NewDeletingReaper()
+
+	for i := 1; i <= maxDeleteReapAttempts; i++ {
+		n, err := reaper.Sweep(context.Background())
+		require.NoError(t, err, "Sweep logs and continues past a single cluster's re-drive failure")
+		assert.Equal(t, 0, n)
+
+		meta, getErr := GetClusterMeta(t.Context(), f.kv, "alpha")
+		require.NoError(t, getErr)
+		assert.Equal(t, i, meta.DeleteReapAttempts, "attempt count after sweep %d", i)
+
+		if i < maxDeleteReapAttempts {
+			assert.False(t, meta.DeleteReapExhausted, "must not exhaust before maxDeleteReapAttempts")
+			// Fast-forward past the next backoff window without sleeping.
+			meta.LastDeleteReapAttempt = time.Now().UTC().Add(-2 * deleteReapBackoff(deletingReapMinAge, meta.DeleteReapAttempts))
+			require.NoError(t, PutClusterMeta(t.Context(), f.kv, meta))
+		}
+	}
+
+	meta, getErr := GetClusterMeta(t.Context(), f.kv, "alpha")
+	require.NoError(t, getErr)
+	assert.True(t, meta.DeleteReapExhausted, "the backstop must give up after maxDeleteReapAttempts")
+	assert.Equal(t, ClusterStatusDeleting, meta.Status,
+		"ADR-0006 §6 billing invariant: an exhausted cluster must stay DELETING")
+
+	terminateCallsAtExhaustion := len(f.inst.terminateCalls)
+
+	// Even long after exhaustion, further sweeps must never re-drive it again.
+	meta.LastDeleteReapAttempt = time.Now().UTC().Add(-24 * time.Hour)
+	require.NoError(t, PutClusterMeta(t.Context(), f.kv, meta))
+	n, err := reaper.Sweep(context.Background())
+	require.NoError(t, err, "a skipped exhausted cluster must not surface a sweep error")
+	assert.Equal(t, 0, n)
+	assert.Len(t, f.inst.terminateCalls, terminateCallsAtExhaustion, "an exhausted cluster must never be re-driven again")
+}
+
+// TestDeletingReaper_PersistsLastErrorOnFailure guards mulga-djaep criterion 2:
+// a terminal give-up must carry the last error, not just a bare boolean flag.
+// The error must be visible on the very first failed attempt (so an operator
+// checking a still-retrying cluster already sees why), and the same error must
+// still be there once the reaper exhausts its attempts and gives up.
+func TestDeletingReaper_PersistsLastErrorOnFailure(t *testing.T) {
+	f := newDeleteClusterFixture(t, "alpha")
+	f.inst.terminateErr = errors.New("dependency violation: eni still attached")
+	markDeleting(t, f, "alpha", 10*time.Minute)
+
+	reaper := f.svc.NewDeletingReaper()
+
+	n, err := reaper.Sweep(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, 0, n)
+
+	meta, getErr := GetClusterMeta(t.Context(), f.kv, "alpha")
+	require.NoError(t, getErr)
+	assert.Contains(t, meta.LastDeleteReapError, "dependency violation",
+		"the failure reason must be persisted after the very first failed attempt")
+
+	for i := 2; i <= maxDeleteReapAttempts; i++ {
+		meta.LastDeleteReapAttempt = time.Now().UTC().Add(-2 * deleteReapBackoff(deletingReapMinAge, meta.DeleteReapAttempts))
+		require.NoError(t, PutClusterMeta(t.Context(), f.kv, meta))
+
+		_, err := reaper.Sweep(context.Background())
+		require.NoError(t, err)
+
+		meta, getErr = GetClusterMeta(t.Context(), f.kv, "alpha")
+		require.NoError(t, getErr)
+	}
+
+	assert.True(t, meta.DeleteReapExhausted, "the backstop must have given up by now")
+	assert.Contains(t, meta.LastDeleteReapError, "dependency violation",
+		"the last error must still be carried once the cluster reaches its terminal give-up")
+}
+
 // TestDeletingReaperSkipsNonDeleting: a CREATING/ACTIVE cluster is never touched.
 func TestDeletingReaperSkipsNonDeleting(t *testing.T) {
 	f := newDeleteClusterFixture(t, "gamma") // stays CREATING

spinifex/handlers/eks/service_impl.go

@@ -83,6 +83,11 @@ type EKSServiceDeps struct {
 	IGW       igwProvisioner
 	Worker    WorkerLauncher
 
+	// Volume lets purgeClusterInfra find EBS volumes the in-cluster CSI driver
+	// provisioned for this cluster's PVCs, which the CP/nodegroup teardown never
+	// touches. Nil disables the reclaim step.
+	Volume csiVolumeReclaimer
+
 	// IAM backs a nodegroup's node role with an instance profile so workers expose
 	// the role over IMDS for the ECR credential provider. Nil disables the wiring
 	// (workers launch without a profile and cannot pull from the internal ECR).
@@ -226,6 +231,14 @@ type eipProvisioner interface {
 	ReleaseAddress(ctx context.Context, input *ec2.ReleaseAddressInput, accountID string) (*ec2.ReleaseAddressOutput, error)
 }
 
+// csiVolumeReclaimer is the narrow EC2 volume surface purgeClusterInfra needs
+// to find EBS volumes the in-cluster CSI driver provisioned for this cluster's
+// PVCs. Nil disables the reclaim step (CSI-provisioned volumes go unreported,
+// matching today's behavior).
+type csiVolumeReclaimer interface {
+	DescribeVolumes(ctx context.Context, input *ec2.DescribeVolumesInput, accountID string) (*ec2.DescribeVolumesOutput, error)
+}
+
 // EKSServiceImpl is the daemon-side EKSService implementation.
 type EKSServiceImpl struct {
 	deps     EKSServiceDeps
@@ -1317,6 +1330,15 @@ func (s *EKSServiceImpl) purgeClusterInfra(ctx context.Context, accountID, name
 	// group strands nothing.
 	s.teardownSpreadGroup(ctx, meta)
 
+	// Surface (never delete) any CSI-provisioned EBS volume this teardown left
+	// behind. Read-only and non-blocking: a scan failure must not wedge an
+	// otherwise-complete deletion in DELETING.
+	if reclaimed, rerr := s.reclaimCSIVolumes(ctx, accountID, name); rerr != nil {
+		slog.WarnContext(ctx, "purgeClusterInfra: CSI volume reclaim scan failed", "cluster", name, "err", rerr)
+	} else if reclaimed > 0 {
+		slog.WarnContext(ctx, "purgeClusterInfra: CSI-provisioned volumes require operator reclaim", "cluster", name, "count", reclaimed)
+	}
+
 	// Any blocking-teardown failure (billable infra or a VPC-pinning SG) keeps the
 	// meta — and the IDs that own the stranded resources — alive for a delete retry
 	// or the DELETING backstop, rather than completing deletion with infra leaked.

spinifex/handlers/eks/volume_reclaim.go

@@ -0,0 +1,81 @@
+package handlers_eks
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+)
+
+// csiClusterOwnerTag is the tag key the aws-ebs-csi-driver controller stamps
+// on every volume it provisions, once its --k8s-tag-cluster-id flag is set to
+// the cluster name (scripts/images/eks-node/addons/aws-ebs-csi-driver's
+// 10-controller.yaml). Without that flag a CSI-provisioned volume carries no
+// cluster-scoped tag at all — only PVC/PV identity — so reclaim would have to
+// guess across every cluster sharing the account, which risks matching a
+// different live cluster's volume.
+func csiClusterOwnerTag(name string) string {
+	return "kubernetes.io/cluster/" + name
+}
+
+// reclaimCSIVolumes finds EBS volumes the in-cluster CSI driver provisioned
+// for this cluster's PVCs and reports (never deletes) whichever are still
+// around after the CP/nodegroup teardown above. Real EKS has the identical
+// gap: the CSI controller's own DeleteVolume call only runs while the cluster
+// (and its controller pod) is still alive, so a cluster deleted with PVCs
+// still bound leaks their volumes there too — the customer is expected to
+// delete workloads before deleting the cluster. This step exists purely to
+// make that leak visible here instead of silent, since nothing else looks for
+// it: VolumeLeakReaper only catches a volume still attached to a definitively
+// gone instance, and an EKS worker's CSI-provisioned volume is detached by
+// the time its node is torn down.
+//
+// It never deletes: a PV's reclaimPolicy (Delete vs Retain) is a
+// Kubernetes-side property the CSI driver does not tag onto the volume, so
+// there is no safe way to tell from EC2 tags alone whether deleting a given
+// volume is what the customer wanted. Deleting a Retain volume would be
+// unrecoverable data loss, so — mirroring VolumeLeakReaper's mark-and-alarm
+// policy for the identical reason (ADR-0005 §3) — reclamation stays an
+// explicit operator action.
+func (s *EKSServiceImpl) reclaimCSIVolumes(ctx context.Context, accountID, name string) (int, error) {
+	if s.deps.Volume == nil {
+		return 0, nil
+	}
+	out, err := s.deps.Volume.DescribeVolumes(ctx, &ec2.DescribeVolumesInput{}, accountID)
+	if err != nil {
+		return 0, fmt.Errorf("describe volumes: %w", err)
+	}
+
+	ownerTag := csiClusterOwnerTag(name)
+	reported := 0
+	for _, vol := range out.Volumes {
+		if vol == nil {
+			continue
+		}
+		var owned bool
+		var pvcNamespace, pvcName string
+		for _, tag := range vol.Tags {
+			if tag == nil || tag.Key == nil {
+				continue
+			}
+			switch aws.StringValue(tag.Key) {
+			case ownerTag:
+				owned = true
+			case "kubernetes.io/created-for/pvc/namespace":
+				pvcNamespace = aws.StringValue(tag.Value)
+			case "kubernetes.io/created-for/pvc/name":
+				pvcName = aws.StringValue(tag.Value)
+			}
+		}
+		if !owned {
+			continue
+		}
+		reported++
+		slog.WarnContext(ctx, "purgeClusterInfra: CSI-provisioned volume outlived cluster teardown, not deleted (reclaimPolicy unknown from tags alone — operator action required)",
+			"cluster", name, "account", accountID, "volumeId", aws.StringValue(vol.VolumeId),
+			"pvcNamespace", pvcNamespace, "pvcName", pvcName, "state", aws.StringValue(vol.State))
+	}
+	return reported, nil
+}

spinifex/handlers/eks/volume_reclaim_test.go

@@ -0,0 +1,82 @@
+package handlers_eks
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/stretchr/testify/require"
+)
+
+// fakeCSIVolumeReclaimer hands back a fixed DescribeVolumes result so
+// reclaimCSIVolumes is exercised without a live EC2 volume service.
+type fakeCSIVolumeReclaimer struct {
+	out *ec2.DescribeVolumesOutput
+	err error
+}
+
+var _ csiVolumeReclaimer = (*fakeCSIVolumeReclaimer)(nil)
+
+func (f *fakeCSIVolumeReclaimer) DescribeVolumes(_ context.Context, _ *ec2.DescribeVolumesInput, _ string) (*ec2.DescribeVolumesOutput, error) {
+	if f.err != nil {
+		return nil, f.err
+	}
+	return f.out, nil
+}
+
+func tagVol(id string, tags map[string]string) *ec2.Volume {
+	v := &ec2.Volume{VolumeId: aws.String(id), State: aws.String("available")}
+	for k, val := range tags {
+		v.Tags = append(v.Tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(val)})
+	}
+	return v
+}
+
+func TestReclaimCSIVolumes_ReportsOwnedVolumes(t *testing.T) {
+	fake := &fakeCSIVolumeReclaimer{out: &ec2.DescribeVolumesOutput{Volumes: []*ec2.Volume{
+		tagVol("vol-owned1", map[string]string{
+			"kubernetes.io/cluster/test-cluster":      "owned",
+			"kubernetes.io/created-for/pvc/namespace": "default",
+			"kubernetes.io/created-for/pvc/name":      "data-pvc",
+		}),
+	}}}
+	svc := &EKSServiceImpl{deps: EKSServiceDeps{Volume: fake}}
+
+	n, err := svc.reclaimCSIVolumes(context.Background(), "acct1", "test-cluster")
+	require.NoError(t, err)
+	require.Equal(t, 1, n, "the cluster-owner-tagged volume must be reported")
+}
+
+func TestReclaimCSIVolumes_IgnoresUnrelatedVolumes(t *testing.T) {
+	fake := &fakeCSIVolumeReclaimer{out: &ec2.DescribeVolumesOutput{Volumes: []*ec2.Volume{
+		tagVol("vol-other-cluster", map[string]string{
+			"kubernetes.io/cluster/some-other-cluster": "owned",
+		}),
+		tagVol("vol-no-csi-tag", map[string]string{
+			"Name": "unrelated-volume",
+		}),
+	}}}
+	svc := &EKSServiceImpl{deps: EKSServiceDeps{Volume: fake}}
+
+	n, err := svc.reclaimCSIVolumes(context.Background(), "acct1", "test-cluster")
+	require.NoError(t, err)
+	require.Equal(t, 0, n, "volumes owned by a different cluster or untagged must not be reported")
+}
+
+func TestReclaimCSIVolumes_NilDepsIsNoop(t *testing.T) {
+	svc := &EKSServiceImpl{deps: EKSServiceDeps{}}
+
+	n, err := svc.reclaimCSIVolumes(context.Background(), "acct1", "test-cluster")
+	require.NoError(t, err)
+	require.Equal(t, 0, n, "a nil Volume dep must disable the reclaim step, not panic or error")
+}
+
+func TestReclaimCSIVolumes_DescribeErrorSurfaces(t *testing.T) {
+	fake := &fakeCSIVolumeReclaimer{err: errors.New("describe volumes: boom")}
+	svc := &EKSServiceImpl{deps: EKSServiceDeps{Volume: fake}}
+
+	_, err := svc.reclaimCSIVolumes(context.Background(), "acct1", "test-cluster")
+	require.Error(t, err, "a DescribeVolumes failure must surface, not be swallowed as zero volumes")
+}

tests/e2e/eks/eks_test.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/eks"
 	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/aws/aws-sdk-go/service/sts"
@@ -363,6 +364,42 @@ func registerOIDCRole(t *testing.T, c *harness.AWSClient, fx *clusterFixture, ro
 	return providerArn, roleArn, roleName
 }
 
+// ensureNodeRole creates the cluster's worker-node IAM role (standard
+// EC2-service trust policy — real EKS has the same prerequisite: the customer
+// creates the node instance role before CreateNodegroup). Both runIRSAPod and
+// runEBSCSIVolume call this independently rather than one leaking the role for
+// the other to pick up by naming convention: CreateNodegroup's worker launch
+// attaches the role to a same-named instance profile (ensureNodeInstanceProfile
+// in nodegroup.go), so DeleteRole fails while that attachment stands. The
+// cleanup here detaches it first and asserts the delete instead of discarding
+// its error, so a real leak fails the test instead of leaving the role behind
+// for the next subtest to silently depend on.
+func ensureNodeRole(t *testing.T, c *harness.AWSClient, fx *clusterFixture) (roleArn, roleName string) {
+	t.Helper()
+	roleName = fx.ClusterName + "-node"
+	const ec2TrustPolicy = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"ec2.amazonaws.com"},"Action":"sts:AssumeRole"}]}`
+	out, err := c.IAM.CreateRole(&iam.CreateRoleInput{
+		RoleName:                 aws.String(roleName),
+		AssumeRolePolicyDocument: aws.String(ec2TrustPolicy),
+		Description:              aws.String("E2E node instance role"),
+	})
+	require.NoError(t, err, "create-node-role")
+	roleArn = aws.StringValue(out.Role.Arn)
+	t.Cleanup(func() {
+		_, err := c.IAM.RemoveRoleFromInstanceProfile(&iam.RemoveRoleFromInstanceProfileInput{
+			InstanceProfileName: aws.String(roleName),
+			RoleName:            aws.String(roleName),
+		})
+		if err != nil && !harness.ErrorCodeIs(err, iam.ErrCodeNoSuchEntityException) {
+			t.Errorf("remove-role-from-instance-profile %s: %v", roleName, err)
+		}
+		if _, err := c.IAM.DeleteRole(&iam.DeleteRoleInput{RoleName: aws.String(roleName)}); err != nil {
+			t.Errorf("delete-node-role %s: %v", roleName, err)
+		}
+	})
+	return roleArn, roleName
+}
+
 // runIRSAPod exercises the in-cluster IRSA path a real workload depends on:
 // spinifex ships no pod-identity webhook, so a pod must wire the projected SA
 // token + AWS_* env explicitly (mirroring
@@ -432,21 +469,10 @@ func runIRSAPod(t *testing.T, c *harness.AWSClient, env *harness.Env, artifacts
 	// declared NodeRole (ensureNodeInstanceProfile in nodegroup.go), which
 	// requires the role to actually exist in IAM — real EKS has the same
 	// prerequisite (the customer creates the node instance role before
-	// CreateNodegroup). Standard EC2-service trust policy, no permission
-	// policy needed: the pod pulls from public.ecr.aws directly, never
-	// touching the node's own instance-profile credentials.
-	nodeRoleName := fx.ClusterName + "-node"
-	const ec2TrustPolicy = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"ec2.amazonaws.com"},"Action":"sts:AssumeRole"}]}`
-	nodeRoleOut, err := c.IAM.CreateRole(&iam.CreateRoleInput{
-		RoleName:                 aws.String(nodeRoleName),
-		AssumeRolePolicyDocument: aws.String(ec2TrustPolicy),
-		Description:              aws.String("E2E IRSA pod nodegroup worker role"),
-	})
-	require.NoError(t, err, "create-node-role")
-	nodeRoleArn := aws.StringValue(nodeRoleOut.Role.Arn)
-	t.Cleanup(func() {
-		_, _ = c.IAM.DeleteRole(&iam.DeleteRoleInput{RoleName: aws.String(nodeRoleName)})
-	})
+	// CreateNodegroup). No permission policy needed: the pod pulls from
+	// public.ecr.aws directly, never touching the node's own instance-profile
+	// credentials.
+	nodeRoleArn, _ := ensureNodeRole(t, c, fx)
 
 	const nodegroup = "irsa-pod-e2e-ng"
 	harness.Phase(t, "Creating worker nodegroup %s", nodegroup)
@@ -691,7 +717,10 @@ func runEBSCSIVolume(t *testing.T, c *harness.AWSClient, env *harness.Env, artif
 	// workloads need a customer worker node. Create a 1-node nodegroup; its
 	// worker is a customer-space instance, so the customer-owned volume can
 	// attach to it. DescribeNodegroup reports ACTIVE only once the worker has
-	// registered Ready.
+	// registered Ready. The node role is created here rather than assumed from
+	// another subtest — CreateNodegroup requires it to already exist in IAM.
+	nodeRoleArn, _ := ensureNodeRole(t, c, fx)
+
 	const nodegroup = "ebs-csi-e2e-ng"
 	harness.Phase(t, "Creating worker nodegroup %s", nodegroup)
 	// e2e:allow-create — the worker nodegroup is the subject under test (customer-space node for cross-space attach).
@@ -699,7 +728,7 @@ func runEBSCSIVolume(t *testing.T, c *harness.AWSClient, env *harness.Env, artif
 		ClusterName:   aws.String(fx.ClusterName),
 		NodegroupName: aws.String(nodegroup),
 		Subnets:       aws.StringSlice([]string{fx.SubnetID}),
-		NodeRole:      aws.String(fmt.Sprintf("arn:aws:iam::%s:role/%s-node", fx.AccountID, fx.ClusterName)),
+		NodeRole:      aws.String(nodeRoleArn),
 		ScalingConfig: &eks.NodegroupScalingConfig{
 			MinSize:     aws.Int64(1),
 			MaxSize:     aws.Int64(1),
@@ -823,6 +852,15 @@ func runEBSCSIVolume(t *testing.T, c *harness.AWSClient, env *harness.Env, artif
 	}, 5*time.Minute, 5*time.Second)
 	t.Logf("PVC %s Bound", pvcName)
 
+	// Resolved once, while the PVC is still bound, so the final cleanup can
+	// confirm the CSI driver actually deleted this exact volume rather than
+	// just that the Kubernetes objects are gone. Unlike production teardown
+	// (which cannot infer a PV's reclaim policy from EC2 tags alone), this
+	// suite owns the volume it provisioned and can safely wait for its
+	// ebs-gp3 StorageClass's Delete policy to reclaim it.
+	volID := csiVolumeID(t, kc, pvcName)
+	t.Logf("CSI volume for PVC %s: %s", pvcName, volID)
+
 	waitPodReady(t, kc, "ebs-csi-e2e-writer")
 
 	// The marker must have landed on the mounted volume.
@@ -839,8 +877,26 @@ func runEBSCSIVolume(t *testing.T, c *harness.AWSClient, env *harness.Env, artif
 	readerPath := filepath.Join(artifacts, "ebs-csi-reader.yaml")
 	require.NoError(t, os.WriteFile(readerPath, []byte(csiPVCPodManifest(pvcName, "ebs-csi-e2e-reader",
 		"cat /data/marker && sleep 3600")), 0o600))
+	// This runs before the addon/nodegroup cleanups registered earlier in this
+	// subtest, so the CSI controller that performs the actual DeleteVolume is
+	// still running when it does. It must not just fire the delete and move
+	// on: the suite provisioned this volume, so it must confirm EC2 shows it
+	// gone before continuing, not merely that the Kubernetes objects are.
 	t.Cleanup(func() {
-		_, _ = kc.Run(60*time.Second, "delete", "-f", readerPath, "--ignore-not-found", "--wait=false")
+		out, err := kc.Run(60*time.Second, "delete", "-f", readerPath, "--ignore-not-found", "--wait=false")
+		if err != nil {
+			t.Errorf("delete reader manifest:\n%s", out)
+		}
+		harness.EventuallyErr(t, func() error {
+			_, derr := c.EC2.DescribeVolumes(&ec2.DescribeVolumesInput{VolumeIds: aws.StringSlice([]string{volID})})
+			if derr != nil {
+				if harness.ErrorCodeIs(derr, "InvalidVolume.NotFound") {
+					return nil
+				}
+				return derr
+			}
+			return fmt.Errorf("CSI-provisioned volume %s still present after PVC deletion", volID)
+		}, 3*time.Minute, 5*time.Second)
 	})
 	out, err = kc.Run(60*time.Second, "apply", "-f", readerPath)
 	require.NoErrorf(t, err, "apply reader:\n%s", out)
@@ -852,6 +908,23 @@ func runEBSCSIVolume(t *testing.T, c *harness.AWSClient, env *harness.Env, artif
 	t.Logf("marker survived reschedule: %s", strings.TrimSpace(got))
 }
 
+// csiVolumeID resolves the EC2 volume ID backing a Bound PVC by following its
+// PersistentVolume's CSI volumeHandle, so teardown can confirm against EC2
+// that the specific volume this test provisioned is actually gone.
+func csiVolumeID(t *testing.T, kc *harness.Kubectl, pvc string) string {
+	t.Helper()
+	pvName, err := kc.Run(30*time.Second, "get", "pvc", pvc, "-o", `jsonpath={.spec.volumeName}`)
+	require.NoErrorf(t, err, "get pvc volumeName:\n%s", pvName)
+	pvName = strings.TrimSpace(pvName)
+	require.NotEmpty(t, pvName, "a Bound PVC must reference a PersistentVolume")
+
+	volID, err := kc.Run(30*time.Second, "get", "pv", pvName, "-o", `jsonpath={.spec.csi.volumeHandle}`)
+	require.NoErrorf(t, err, "get pv volumeHandle:\n%s", volID)
+	volID = strings.TrimSpace(volID)
+	require.NotEmpty(t, volID, "a CSI-provisioned PV must carry a volumeHandle")
+	return volID
+}
+
 // csiPVCPodManifest renders a gp3 PVC plus a single pod that mounts it at /data
 // and runs cmd. Reusing the claim name lets the reader pod rebind the volume the
 // writer provisioned. No control-plane toleration: the pod must land on the