feat(quota): add unwired owner-side quota override cleanup (#1240)

Pin that account teardown removes the usage counter but leaves the
per-account quota override, so a reused account ID inherits raised or
Unlimited limits. Add an unwired, idempotent CleanupAccount on the quota
service: a present override key is removed by a revision-guarded delete
and reported deleted with its revision; a missing key is reported only as
observed-absent-at-scan.

spinifex/accountteardown/reapers_ec2_account_settings_current_behaviour_test.go

@@ -0,0 +1,50 @@
+package accountteardown
+
+//test:in-package — drives the real final-stage reapers against the real
+// EC2 account-settings service to pin that teardown leaves the bucket alone.
+
+import (
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	ec2account "github.com/mulgadc/spinifex/spinifex/domains/ec2/account"
+	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// No reaper in this package references the EC2 account-settings bucket or
+// calls a settings delete. Running the real engine and the real StageAccount
+// reapers against an account with settings enabled shows the consequence:
+// the settings record outlives the account record it belongs to.
+func TestCurrentBehaviour_AccountTermination_EC2AccountSettingsSurviveTeardown(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+	ctx := testCtx(t)
+
+	settings, err := ec2account.NewAccountSettingsServiceImplWithNATS(ctx, nil, nc)
+	require.NoError(t, err)
+	_, err = settings.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+	_, err = settings.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID)
+	require.NoError(t, err)
+
+	accounts := newFakeAccounts()
+	iam := &accountIAM{account: &handlers_iam.Account{AccountID: testAccountID, AccountName: testAccountName}}
+	engine := quietEngine(accounts, AccountReapers(iam, nil, nil)...)
+
+	result, err := engine.Teardown(ctx, request())
+	require.NoError(t, err)
+	assert.True(t, result.AccountDeleted)
+	assert.True(t, accounts.deleted)
+	assert.Zero(t, result.StuckCount())
+
+	bucket, err := js.KeyValue(ctx, ec2account.KVBucketAccountSettings)
+	require.NoError(t, err)
+	_, err = bucket.Get(ctx, testAccountID)
+	require.NoError(t, err, "the settings key must still be present after the account is torn down")
+
+	out, err := settings.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+	assert.True(t, *out.EbsEncryptionByDefault)
+}

spinifex/accountteardown/reapers_quota_override_current_behaviour_test.go

@@ -0,0 +1,61 @@
+package accountteardown
+
+//test:in-package — drives the real final-stage reapers against the real quota service to pin that teardown leaves the override bucket alone.
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/admission/quota"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// No reaper in this package references the quota override bucket or calls an
+// override delete. Running the real engine and the real StageAccount reapers
+// against an account with a raised vCPU override, with usage wired, shows the
+// consequence: the override outlives the account record it belongs to, while
+// the usage counter the reapers do own is removed.
+func TestCurrentBehaviour_AccountTermination_QuotaOverridesSurviveTeardown(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	ctx := testCtx(t)
+
+	usage, err := kvutil.GetOrCreateBucket(ctx, js, quota.KVBucketAccountUsage, 1)
+	require.NoError(t, err)
+	_, err = usage.Put(ctx, testAccountID, []byte(`{"vcpus":4}`))
+	require.NoError(t, err)
+
+	overridesKV, err := kvutil.GetOrCreateBucket(ctx, js, quota.KVBucketAccountQuota, 1)
+	require.NoError(t, err)
+	svc := quota.New(quota.Limits{Enabled: true, VCPUs: 16}, usage)
+	svc.SetOverrides(overridesKV)
+	unlimited := quota.Unlimited
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccountID, quota.Overrides{VCPUs: &unlimited}, "operator"))
+
+	accounts := newFakeAccounts()
+	iam := &accountIAM{account: &handlers_iam.Account{AccountID: testAccountID, AccountName: testAccountName}}
+	engine := quietEngine(accounts, AccountReapers(iam, nil, usage)...)
+
+	result, err := engine.Teardown(ctx, request())
+	require.NoError(t, err)
+	assert.True(t, result.AccountDeleted)
+	assert.True(t, accounts.deleted)
+	assert.Zero(t, result.StuckCount())
+
+	// The reaper this package does own removed the usage counter, which shows
+	// which store teardown actually clears.
+	_, err = usage.Get(ctx, testAccountID)
+	assert.ErrorIs(t, err, jetstream.ErrKeyNotFound, "teardown's quota-usage reaper must remove the vCPU counter")
+
+	_, err = overridesKV.Get(ctx, testAccountID)
+	require.NoError(t, err, "the override key must still be present after the account is torn down")
+
+	over, effective, err := svc.GetAccountQuota(ctx, testAccountID)
+	require.NoError(t, err)
+	require.NotNil(t, over.VCPUs)
+	assert.Equal(t, quota.Unlimited, *over.VCPUs)
+	assert.Equal(t, quota.Unlimited, effective.VCPUs)
+}

spinifex/domains/admission/quota/cleanup.go

@@ -0,0 +1,93 @@
+package quota
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// OverrideCleanupDisposition is CleanupAccount's outcome for an account's
+// quota override key.
+type OverrideCleanupDisposition string
+
+const (
+	OverrideCleanupDeleted OverrideCleanupDisposition = "deleted"
+	OverrideCleanupAbsent  OverrideCleanupDisposition = "absent"
+	OverrideCleanupFailed  OverrideCleanupDisposition = "failed"
+)
+
+// evidenceRevisionGuardedDelete and evidenceObservedAbsent are the fixed
+// evidence bases CleanupAccount reports; see kvstore/bucket.go:112-120 for why
+// a plain read cannot support a stronger absence claim.
+const (
+	evidenceRevisionGuardedDelete = "revision-guarded-delete"
+	evidenceObservedAbsent        = "observed-absent-at-scan"
+)
+
+// cleanupMaxAttempts bounds retries against a revision that keeps moving
+// under concurrent writes, so cleanup fails closed instead of spinning.
+const cleanupMaxAttempts = 5
+
+// OverrideCleanup is CleanupAccount's result: what happened to an account's
+// quota override key, and the fixed evidence basis for that claim.
+type OverrideCleanup struct {
+	Disposition OverrideCleanupDisposition
+	Evidence    string
+	Revision    uint64
+}
+
+// CleanupAccount removes accountID's quota override key so a reused account ID
+// does not inherit a deleted account's raised or lowered limits. It is
+// unwired: nothing calls it yet.
+func (s *Service) CleanupAccount(ctx context.Context, accountID string) (*OverrideCleanup, error) {
+	if accountID == "" {
+		return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required for quota override cleanup")
+	}
+	if !awsidentifiers.IsAccountID(accountID) {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "accountID %q is not a valid AWS account ID", accountID)
+	}
+	if accountID == awsidentifiers.GlobalAccountID {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "quota override cleanup must not target the global account")
+	}
+	if s == nil || s.overrides == nil {
+		return &OverrideCleanup{Disposition: OverrideCleanupFailed}, errors.New("quota: override bucket not configured")
+	}
+
+	// accountID is never resolved through any fallback here: cleanup must act
+	// on exactly accountID and never fall back to the global key.
+	for range cleanupMaxAttempts {
+		entry, err := s.overrides.Get(ctx, accountID)
+		if err != nil {
+			if errors.Is(err, jetstream.ErrKeyNotFound) {
+				return &OverrideCleanup{Disposition: OverrideCleanupAbsent, Evidence: evidenceObservedAbsent}, nil
+			}
+			return &OverrideCleanup{Disposition: OverrideCleanupFailed}, fmt.Errorf("get quota override for cleanup: %w", err)
+		}
+		rev := entry.Revision()
+
+		err = s.overrides.Delete(ctx, accountID, jetstream.LastRevision(rev))
+		switch {
+		case err == nil:
+			return &OverrideCleanup{
+				Disposition: OverrideCleanupDeleted,
+				Evidence:    evidenceRevisionGuardedDelete,
+				Revision:    rev,
+			}, nil
+		case errors.Is(err, jetstream.ErrKeyNotFound):
+			return &OverrideCleanup{Disposition: OverrideCleanupAbsent, Evidence: evidenceObservedAbsent}, nil
+		case errors.Is(err, jetstream.ErrKeyRevisionMismatch):
+			// A concurrent write landed between the read and the guarded
+			// delete; retry against whatever revision now holds.
+			continue
+		default:
+			return &OverrideCleanup{Disposition: OverrideCleanupFailed}, fmt.Errorf("delete quota override for cleanup: %w", err)
+		}
+	}
+
+	return &OverrideCleanup{Disposition: OverrideCleanupFailed},
+		fmt.Errorf("quota override cleanup for account: exhausted %d attempts against concurrent writes", cleanupMaxAttempts)
+}

spinifex/domains/admission/quota/cleanup_test.go

@@ -0,0 +1,184 @@
+//test:in-package — CleanupAccount deletes through the unexported overrides field, and the retry/race behaviour is only observable from inside the package.
+package quota
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// overrideCleanupFakeKV wraps a real KeyValue so a test can inject a race
+// between CleanupAccount's read and its guarded delete, or a store failure,
+// deterministically rather than through goroutine timing.
+type overrideCleanupFakeKV struct {
+	jetstream.KeyValue
+
+	deleteCalls  int
+	beforeDelete func(attempt int) error
+	getErr       error
+}
+
+func (f *overrideCleanupFakeKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
+	if f.getErr != nil {
+		return nil, f.getErr
+	}
+	return f.KeyValue.Get(ctx, key)
+}
+
+func (f *overrideCleanupFakeKV) Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error {
+	f.deleteCalls++
+	if f.beforeDelete != nil {
+		if err := f.beforeDelete(f.deleteCalls); err != nil {
+			return err
+		}
+	}
+	return f.KeyValue.Delete(ctx, key, opts...)
+}
+
+func TestCleanupAccount_DeletesTheOverrideKey(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	ctx := t.Context()
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator"))
+
+	result, err := svc.CleanupAccount(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupDeleted, result.Disposition)
+	assert.Equal(t, evidenceRevisionGuardedDelete, result.Evidence)
+	assert.NotZero(t, result.Revision)
+
+	_, err = svc.overrides.Get(ctx, testAccount)
+	assert.ErrorIs(t, err, jetstream.ErrKeyNotFound)
+}
+
+func TestCleanupAccount_RepeatCallIsIdempotent(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	ctx := t.Context()
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator"))
+
+	first, err := svc.CleanupAccount(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupDeleted, first.Disposition)
+
+	second, err := svc.CleanupAccount(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupAbsent, second.Disposition)
+	assert.Equal(t, evidenceObservedAbsent, second.Evidence)
+}
+
+func TestCleanupAccount_AbsentKeyIsIdempotent(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	result, err := svc.CleanupAccount(t.Context(), "333333333333")
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupAbsent, result.Disposition)
+	assert.Equal(t, evidenceObservedAbsent, result.Evidence)
+}
+
+func TestCleanupAccount_OtherAccountsUntouched(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	ctx := t.Context()
+	accountA, accountB := testAccount, "222222222222"
+	require.NoError(t, svc.PutAccountQuota(ctx, accountA, Overrides{VCPUs: ptr(32)}, "operator"))
+	require.NoError(t, svc.PutAccountQuota(ctx, accountB, Overrides{VCPUs: ptr(8)}, "operator"))
+
+	result, err := svc.CleanupAccount(ctx, accountA)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupDeleted, result.Disposition)
+
+	over, _, err := svc.GetAccountQuota(ctx, accountB)
+	require.NoError(t, err)
+	require.NotNil(t, over.VCPUs)
+	assert.Equal(t, 8, *over.VCPUs)
+}
+
+func TestCleanupAccount_RefusesGlobalAccountID(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	result, err := svc.CleanupAccount(t.Context(), awsidentifiers.GlobalAccountID)
+	assert.Error(t, err)
+	assert.Nil(t, result)
+}
+
+func TestCleanupAccount_RefusesEmptyAccountID(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	result, err := svc.CleanupAccount(t.Context(), "")
+	assert.Error(t, err)
+	assert.Nil(t, result)
+}
+
+func TestCleanupAccount_RefusesInvalidAccountID(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	result, err := svc.CleanupAccount(t.Context(), "not-an-account")
+	assert.Error(t, err)
+	assert.Nil(t, result)
+}
+
+func TestCleanupAccount_WithoutBucketFails(t *testing.T) {
+	svc := New(baseLimits(), nil)
+	result, err := svc.CleanupAccount(t.Context(), testAccount)
+	assert.Error(t, err)
+	require.NotNil(t, result)
+	assert.Equal(t, OverrideCleanupFailed, result.Disposition)
+}
+
+// A concurrent Put landing between CleanupAccount's read and its
+// revision-guarded delete must not be lost silently: the delete is rejected
+// and retried against the revision the write left behind.
+func TestCleanupAccount_ConcurrentPutDuringCleanupRetriesAndDeletes(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	ctx := t.Context()
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator"))
+
+	fake := &overrideCleanupFakeKV{KeyValue: svc.overrides}
+	fake.beforeDelete = func(attempt int) error {
+		if attempt != 1 {
+			return nil
+		}
+		// Simulate a racing override write landing after CleanupAccount's
+		// read but before its guarded delete, through the real KV so the
+		// delete's expected revision is now stale.
+		return svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")
+	}
+	svc.overrides = fake
+
+	result, err := svc.CleanupAccount(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideCleanupDeleted, result.Disposition)
+	assert.Equal(t, 2, fake.deleteCalls, "the first delete is rejected by the race and retried once")
+}
+
+// A revision that keeps moving under sustained concurrent writes must not
+// spin forever: cleanup gives up and reports failed, bounded by attempts.
+func TestCleanupAccount_PersistentConcurrentWritesFailsBounded(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	ctx := t.Context()
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator"))
+
+	fake := &overrideCleanupFakeKV{KeyValue: svc.overrides}
+	fake.beforeDelete = func(int) error {
+		return svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")
+	}
+	svc.overrides = fake
+
+	result, err := svc.CleanupAccount(ctx, testAccount)
+	assert.Error(t, err)
+	require.NotNil(t, result)
+	assert.Equal(t, OverrideCleanupFailed, result.Disposition)
+	assert.Equal(t, cleanupMaxAttempts, fake.deleteCalls)
+}
+
+// A store error must report failed, never absent: absence is a claim about
+// what the store holds, which an error gives no basis to make.
+func TestCleanupAccount_StoreErrorIsFailedNotAbsent(t *testing.T) {
+	svc, _ := newOverrideService(t)
+	fake := &overrideCleanupFakeKV{KeyValue: svc.overrides, getErr: errors.New("kv unavailable")}
+	svc.overrides = fake
+
+	result, err := svc.CleanupAccount(t.Context(), testAccount)
+	assert.Error(t, err)
+	require.NotNil(t, result)
+	assert.Equal(t, OverrideCleanupFailed, result.Disposition)
+}

spinifex/domains/ec2/account/cleanup.go

@@ -0,0 +1,90 @@
+package account
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// AccountSettingsCleanupDisposition is CleanupAccount's outcome for an
+// account's settings key.
+type AccountSettingsCleanupDisposition string
+
+const (
+	AccountSettingsCleanupDeleted AccountSettingsCleanupDisposition = "deleted"
+	AccountSettingsCleanupAbsent  AccountSettingsCleanupDisposition = "absent"
+	AccountSettingsCleanupFailed  AccountSettingsCleanupDisposition = "failed"
+)
+
+// evidenceRevisionGuardedDelete and evidenceObservedAbsent are the fixed
+// evidence bases CleanupAccount reports; see kvstore/bucket.go:112-120 for why
+// a plain read cannot support a stronger absence claim.
+const (
+	evidenceRevisionGuardedDelete = "revision-guarded-delete"
+	evidenceObservedAbsent        = "observed-absent-at-scan"
+)
+
+// cleanupMaxAttempts bounds retries against a revision that keeps moving
+// under concurrent writes, so cleanup fails closed instead of spinning.
+const cleanupMaxAttempts = 5
+
+// AccountSettingsCleanup is CleanupAccount's result: what happened to an
+// account's settings key, and the fixed evidence basis for that claim.
+type AccountSettingsCleanup struct {
+	Disposition AccountSettingsCleanupDisposition
+	Evidence    string
+	Revision    uint64
+}
+
+// CleanupAccount removes accountID's settings key so a reused account ID does
+// not inherit a deleted account's EBS-encryption-by-default or
+// serial-console-access settings. It is unwired: nothing calls it yet.
+func (s *AccountSettingsServiceImpl) CleanupAccount(ctx context.Context, accountID string) (*AccountSettingsCleanup, error) {
+	if accountID == "" {
+		return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required for settings cleanup")
+	}
+	if !awsidentifiers.IsAccountID(accountID) {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "accountID %q is not a valid AWS account ID", accountID)
+	}
+	if accountID == awsidentifiers.GlobalAccountID {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "settings cleanup must not target the global account")
+	}
+
+	// settingsKey is never used here: cleanup must act on exactly accountID
+	// and never fall back to the global key.
+	for range cleanupMaxAttempts {
+		entry, err := s.settingsKV.Get(ctx, accountID)
+		if err != nil {
+			if errors.Is(err, jetstream.ErrKeyNotFound) {
+				return &AccountSettingsCleanup{Disposition: AccountSettingsCleanupAbsent, Evidence: evidenceObservedAbsent}, nil
+			}
+			return &AccountSettingsCleanup{Disposition: AccountSettingsCleanupFailed}, fmt.Errorf("get account settings for cleanup: %w", err)
+		}
+		rev := entry.Revision()
+
+		err = s.settingsKV.Delete(ctx, accountID, jetstream.LastRevision(rev))
+		switch {
+		case err == nil:
+			return &AccountSettingsCleanup{
+				Disposition: AccountSettingsCleanupDeleted,
+				Evidence:    evidenceRevisionGuardedDelete,
+				Revision:    rev,
+			}, nil
+		case errors.Is(err, jetstream.ErrKeyNotFound):
+			return &AccountSettingsCleanup{Disposition: AccountSettingsCleanupAbsent, Evidence: evidenceObservedAbsent}, nil
+		case errors.Is(err, jetstream.ErrKeyRevisionMismatch):
+			// A concurrent write landed between the read and the guarded
+			// delete; retry against whatever revision now holds.
+			continue
+		default:
+			return &AccountSettingsCleanup{Disposition: AccountSettingsCleanupFailed}, fmt.Errorf("delete account settings for cleanup: %w", err)
+		}
+	}
+
+	return &AccountSettingsCleanup{Disposition: AccountSettingsCleanupFailed},
+		fmt.Errorf("settings cleanup for account: exhausted %d attempts against concurrent writes", cleanupMaxAttempts)
+}

spinifex/domains/ec2/account/cleanup_test.go

@@ -0,0 +1,181 @@
+package account
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// cleanupFakeKV wraps a real KeyValue so a test can inject a race between
+// CleanupAccount's read and its guarded delete, or a store failure,
+// deterministically rather than through goroutine timing.
+type cleanupFakeKV struct {
+	jetstream.KeyValue
+
+	deleteCalls  int
+	beforeDelete func(attempt int) error
+	getErr       error
+}
+
+func (f *cleanupFakeKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
+	if f.getErr != nil {
+		return nil, f.getErr
+	}
+	return f.KeyValue.Get(ctx, key)
+}
+
+func (f *cleanupFakeKV) Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error {
+	f.deleteCalls++
+	if f.beforeDelete != nil {
+		if err := f.beforeDelete(f.deleteCalls); err != nil {
+			return err
+		}
+	}
+	return f.KeyValue.Delete(ctx, key, opts...)
+}
+
+func TestCleanupAccount_DeletesTheSettingsKey(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	result, err := svc.CleanupAccount(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, AccountSettingsCleanupDeleted, result.Disposition)
+	assert.Equal(t, evidenceRevisionGuardedDelete, result.Evidence)
+	assert.NotZero(t, result.Revision)
+
+	_, err = svc.settingsKV.Get(ctx, testAccountID)
+	assert.ErrorIs(t, err, jetstream.ErrKeyNotFound)
+}
+
+func TestCleanupAccount_RepeatCallIsIdempotent(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	first, err := svc.CleanupAccount(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, AccountSettingsCleanupDeleted, first.Disposition)
+
+	second, err := svc.CleanupAccount(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, AccountSettingsCleanupAbsent, second.Disposition)
+	assert.Equal(t, evidenceObservedAbsent, second.Evidence)
+}
+
+func TestCleanupAccount_AbsentKeyIsIdempotent(t *testing.T) {
+	svc := setupTestAccountService(t)
+	result, err := svc.CleanupAccount(context.Background(), "333333333333")
+	require.NoError(t, err)
+	assert.Equal(t, AccountSettingsCleanupAbsent, result.Disposition)
+	assert.Equal(t, evidenceObservedAbsent, result.Evidence)
+}
+
+func TestCleanupAccount_OtherAccountsUntouched(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	accountA, accountB := testAccountID, "222222222222"
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, accountA)
+	require.NoError(t, err)
+	_, err = svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, accountB)
+	require.NoError(t, err)
+
+	result, err := svc.CleanupAccount(ctx, accountA)
+	require.NoError(t, err)
+	assert.Equal(t, AccountSettingsCleanupDeleted, result.Disposition)
+
+	out, err := svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, accountB)
+	require.NoError(t, err)
+	assert.True(t, *out.EbsEncryptionByDefault)
+}
+
+func TestCleanupAccount_RefusesGlobalAccountID(t *testing.T) {
+	svc := setupTestAccountService(t)
+	result, err := svc.CleanupAccount(context.Background(), awsidentifiers.GlobalAccountID)
+	assert.Error(t, err)
+	assert.Nil(t, result)
+}
+
+func TestCleanupAccount_RefusesEmptyAccountID(t *testing.T) {
+	svc := setupTestAccountService(t)
+	result, err := svc.CleanupAccount(context.Background(), "")
+	assert.Error(t, err)
+	assert.Nil(t, result)
+}
+
+func TestCleanupAccount_RefusesInvalidAccountID(t *testing.T) {
+	svc := setupTestAccountService(t)
+	result, err := svc.CleanupAccount(context.Background(), "not-an-account")
+	assert.Error(t, err)
+	assert.Nil(t, result)
+}
+
+// A concurrent Put landing between CleanupAccount's read and its
+// revision-guarded delete must not be lost silently: the delete is rejected
+// and retried against the revision the write left behind.
+func TestCleanupAccount_ConcurrentPutDuringCleanupRetriesAndDeletes(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	fake := &cleanupFakeKV{KeyValue: svc.settingsKV}
+	fake.beforeDelete = func(attempt int) error {
+		if attempt != 1 {
+			return nil
+		}
+		// Simulate a racing settings write landing after CleanupAccount's
+		// read but before its guarded delete, through the real KV so the
+		// delete's expected revision is now stale.
+		return svc.saveSettings(ctx, &AccountSettingsRecord{SerialConsoleAccess: true}, testAccountID)
+	}
+	svc.settingsKV = fake
+
+	result, err := svc.CleanupAccount(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, AccountSettingsCleanupDeleted, result.Disposition)
+	assert.Equal(t, 2, fake.deleteCalls, "the first delete is rejected by the race and retried once")
+}
+
+// A revision that keeps moving under sustained concurrent writes must not
+// spin forever: cleanup gives up and reports failed, bounded by attempts.
+func TestCleanupAccount_PersistentConcurrentWritesFailsBounded(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	fake := &cleanupFakeKV{KeyValue: svc.settingsKV}
+	fake.beforeDelete = func(int) error {
+		return svc.saveSettings(ctx, &AccountSettingsRecord{SerialConsoleAccess: true}, testAccountID)
+	}
+	svc.settingsKV = fake
+
+	result, err := svc.CleanupAccount(ctx, testAccountID)
+	assert.Error(t, err)
+	require.NotNil(t, result)
+	assert.Equal(t, AccountSettingsCleanupFailed, result.Disposition)
+	assert.Equal(t, cleanupMaxAttempts, fake.deleteCalls)
+}
+
+// A store error must report failed, never absent: absence is a claim about
+// what the store holds, which an error gives no basis to make.
+func TestCleanupAccount_StoreErrorIsFailedNotAbsent(t *testing.T) {
+	svc := setupTestAccountService(t)
+	fake := &cleanupFakeKV{KeyValue: svc.settingsKV, getErr: errors.New("kv unavailable")}
+	svc.settingsKV = fake
+
+	result, err := svc.CleanupAccount(context.Background(), testAccountID)
+	assert.Error(t, err)
+	require.NotNil(t, result)
+	assert.Equal(t, AccountSettingsCleanupFailed, result.Disposition)
+}