feat(ec2): seal account settings with revision-guarded writes (#1241)

Replace the unwired delete-based CleanupAccount with an owner-local sealed
state in the existing account-settings KV value. All settings mutations now
use revision-guarded create/update; a sealed record rejects every read and
mutation with Blocked and cannot be recreated. Records without a state
field remain active.

SealAccount CAS-writes the sealed value whether the key is present, absent
or tombstoned, and succeeds only on a leader-accepted write, returning its
revision. It first checks, through a read-only IAM account-status reader,
that the account is already TERMINATING; otherwise it refuses with
IncorrectState and writes nothing.

The daemon serves the internal subject ec2.v1.SealAccountSettings. Nothing
calls it yet.

contracts/ec2/v1/README.md

@@ -60,3 +60,25 @@ The subject name carries the contract major; `platform_inventory_test.go` preser
 Producer and consumer are in the coordinated Spinifex release, so the window is N and N-1.
 A daemon that predates the subject has no subscription, so the request gets no responder; the consumer treats no responder, a timeout, an error reply, an undecodable reply, a reply for another account, and a missing, repeated or unknown-status kind as `scan-incomplete`, never clean.
 An incompatible change to the subject, fields or vocabulary needs a new major.
+
+## Seal account settings
+
+`SealAccountSettingsSubject` (`ec2.v1.SealAccountSettings`) is a request/reply route on which a daemon in the `spinifex-workers` queue group irreversibly seals one account's EC2 settings record.
+EC2 owns it: `domains/ec2/account` serves it through the daemon.
+`accountteardown` is the intended consumer from a later release; no teardown calls it yet.
+
+The request names the account in its body (`account_id`) and may carry an opaque `correlation_ref` that is echoed for logs.
+The daemon ignores `X-Account-ID` for this subject: the account sealed is the one named in the body.
+A seal is refused with `IncorrectState` unless IAM already holds the account as `TERMINATING`; a precondition read failure answers `InternalError`.
+Both refusals write nothing.
+
+A reply is sent only on success: `disposition` is always `sealed` and `revision` is the stream leader's accepted write for that seal.
+A failure is the standard awserrors error payload, not this reply shape.
+Sealing is idempotent: a seal on an already-sealed record re-writes the same value and returns a fresh, higher revision.
+
+### Compatibility boundary
+
+The subject name carries the contract major; `account_settings_seal_test.go` preserves the version-1 JSON shape.
+Producer and consumer are in the coordinated Spinifex release, so the window is N and N-1.
+A daemon that predates the subject has no subscription, so the request gets no responder.
+An incompatible change to the subject or fields needs a new major.

contracts/ec2/v1/account_settings_seal.go

@@ -0,0 +1,28 @@
+package ec2v1
+
+// SealAccountSettingsSubject is the request/reply subject on which a daemon
+// seals one account's EC2 settings record. Sealing is irreversible and refused
+// unless IAM already holds the account as TERMINATING.
+const SealAccountSettingsSubject = "ec2.v1.SealAccountSettings"
+
+// AccountSettingsDispositionSealed is the only disposition a successful reply
+// carries: a reply is sent only on success, so there is no other value.
+const AccountSettingsDispositionSealed = "sealed"
+
+// SealAccountSettingsRequest asks for one account's EC2 settings to be
+// sealed. CorrelationRef is an opaque caller reference echoed for logs; it
+// grants nothing and is never a fence or authority to write.
+type SealAccountSettingsRequest struct {
+	AccountID      string `json:"account_id"`
+	CorrelationRef string `json:"correlation_ref,omitempty"`
+}
+
+// SealAccountSettingsReply confirms accountID's settings record was sealed at
+// Revision, the stream leader's accepted write for that seal. A failure is
+// the standard awserrors error payload, not this type.
+type SealAccountSettingsReply struct {
+	AccountID      string `json:"account_id"`
+	CorrelationRef string `json:"correlation_ref,omitempty"`
+	Disposition    string `json:"disposition"`
+	Revision       uint64 `json:"revision"`
+}

contracts/ec2/v1/account_settings_seal_test.go

@@ -0,0 +1,43 @@
+package ec2v1
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSealAccountSettingsSubject(t *testing.T) {
+	require.Equal(t, "ec2.v1.SealAccountSettings", SealAccountSettingsSubject)
+}
+
+func TestSealAccountSettingsRequestJSONContract(t *testing.T) {
+	data, err := json.Marshal(SealAccountSettingsRequest{AccountID: "111122223333", CorrelationRef: "req-1"})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"account_id":"111122223333","correlation_ref":"req-1"}`, string(data))
+
+	data, err = json.Marshal(SealAccountSettingsRequest{AccountID: "111122223333"})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"account_id":"111122223333"}`, string(data))
+}
+
+func TestSealAccountSettingsReplyJSONContract(t *testing.T) {
+	reply := SealAccountSettingsReply{
+		AccountID:      "111122223333",
+		CorrelationRef: "req-1",
+		Disposition:    AccountSettingsDispositionSealed,
+		Revision:       42,
+	}
+	data, err := json.Marshal(reply)
+	require.NoError(t, err)
+	require.JSONEq(t, `{
+		"account_id": "111122223333",
+		"correlation_ref": "req-1",
+		"disposition": "sealed",
+		"revision": 42
+	}`, string(data))
+
+	data, err = json.Marshal(SealAccountSettingsReply{AccountID: "111122223333", Disposition: AccountSettingsDispositionSealed})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"account_id":"111122223333","disposition":"sealed","revision":0}`, string(data))
+}

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/daemon/daemon.go

@@ -1026,6 +1026,7 @@ func (d *Daemon) subscribeAll() error {
 		{"ec2.GetSerialConsoleAccessStatus", handleNATSRequest(d.node, d.accountService.GetSerialConsoleAccessStatus), "spinifex-workers"},
 		{"ec2.EnableSerialConsoleAccess", handleNATSRequest(d.node, d.accountService.EnableSerialConsoleAccess), "spinifex-workers"},
 		{"ec2.DisableSerialConsoleAccess", handleNATSRequest(d.node, d.accountService.DisableSerialConsoleAccess), "spinifex-workers"},
+		{ec2v1.SealAccountSettingsSubject, handleNATSRequest(d.node, d.sealAccountSettings), "spinifex-workers"},
 		{clusterv1.NodeHealthSubject(d.node), d.handleHealthCheck, ""},
 		{clusterv1.NodesDiscoverSubject, d.handleNodeDiscover, ""},
 		{clusterv1.NodeStatusSubject, d.handleNodeStatus, ""},
@@ -1992,6 +1993,12 @@ func (d *Daemon) startCluster() error {
 	if err != nil {
 		return fmt.Errorf("failed to initialize account settings service: %w", err)
 	}
+	accountStatusJS, err := jetstream.New(d.natsConn)
+	if err != nil {
+		return fmt.Errorf("jetstream handle for account status reader: %w", err)
+	}
+	var _ ec2account.AccountTerminationChecker = (*handlers_iam.AccountStatusReader)(nil)
+	d.accountService.SetAccountTerminationChecker(handlers_iam.NewAccountStatusReader(accountStatusJS))
 
 	// ELBv2 resolves listener certificate private keys through its own ACM
 	// Store over the same JetStream bucket the ACM service writes, so it needs

spinifex/daemon/daemon_handlers_account_settings_seal.go

@@ -0,0 +1,23 @@
+package daemon
+
+import (
+	"context"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+)
+
+// sealAccountSettings serves the irreversible EC2 settings seal. The account
+// comes from the request body, not X-Account-ID: the sealed account is part
+// of the versioned contract, so the header is ignored.
+func (d *Daemon) sealAccountSettings(ctx context.Context, req *ec2v1.SealAccountSettingsRequest, _ string) (*ec2v1.SealAccountSettingsReply, error) {
+	seal, err := d.accountService.SealAccount(ctx, req.AccountID)
+	if err != nil {
+		return nil, err
+	}
+	return &ec2v1.SealAccountSettingsReply{
+		AccountID:      req.AccountID,
+		CorrelationRef: req.CorrelationRef,
+		Disposition:    ec2v1.AccountSettingsDispositionSealed,
+		Revision:       seal.Revision,
+	}, nil
+}

spinifex/daemon/daemon_handlers_account_settings_seal_test.go

@@ -0,0 +1,101 @@
+package daemon
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	ec2account "github.com/mulgadc/spinifex/spinifex/domains/ec2/account"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// sealTestChecker answers AccountTerminating from a fixed set, standing in
+// for IAM in these daemon-level tests.
+type sealTestChecker struct {
+	terminating map[string]bool
+}
+
+func (c *sealTestChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) {
+	return c.terminating[accountID], nil
+}
+
+// newSealTestDaemon builds a daemon serving its real subscription table over
+// a real account settings service, wired with checker as its termination
+// precondition.
+func newSealTestDaemon(t *testing.T, checker ec2account.AccountTerminationChecker) (*Daemon, *nats.Conn) {
+	t.Helper()
+	ns, nc, _ := testutil.StartTestJetStream(t)
+
+	acctSvc, err := ec2account.NewAccountSettingsServiceImplWithNATS(t.Context(), nil, nc)
+	require.NoError(t, err)
+	acctSvc.SetAccountTerminationChecker(checker)
+
+	d := newSubscribeTestDaemon(t, nc, "", "")
+	d.accountService = acctSvc
+	require.NoError(t, d.subscribeAll())
+	require.NoError(t, nc.Flush())
+
+	caller, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	t.Cleanup(caller.Close)
+	return d, caller
+}
+
+const (
+	sealTestTerminating = "111122223333"
+	sealTestActive      = "444455556666"
+)
+
+func sealRequest(t *testing.T, nc *nats.Conn, body any, header string) []byte {
+	t.Helper()
+	data, err := json.Marshal(body)
+	require.NoError(t, err)
+	reply, err := natsRequestAs(nc, ec2v1.SealAccountSettingsSubject, data, header, 5*time.Second)
+	require.NoError(t, err)
+	return reply.Data
+}
+
+// The header is ignored: the body account is what gets sealed, even when
+// X-Account-ID names a different one.
+func TestSealAccountSettingsSubjectSealsTheBodyAccount(t *testing.T) {
+	_, nc := newSealTestDaemon(t, &sealTestChecker{terminating: map[string]bool{sealTestTerminating: true}})
+
+	raw := sealRequest(t, nc, ec2v1.SealAccountSettingsRequest{AccountID: sealTestTerminating, CorrelationRef: "req-1"}, sealTestActive)
+	require.Empty(t, replyErrCode(t, raw), string(raw))
+
+	var reply ec2v1.SealAccountSettingsReply
+	require.NoError(t, json.Unmarshal(raw, &reply))
+	assert.Equal(t, sealTestTerminating, reply.AccountID)
+	assert.Equal(t, "req-1", reply.CorrelationRef)
+	assert.Equal(t, ec2v1.AccountSettingsDispositionSealed, reply.Disposition)
+	assert.NotZero(t, reply.Revision)
+}
+
+// Only an account IAM holds as TERMINATING may be sealed; ACTIVE is refused
+// before any write.
+func TestSealAccountSettingsSubjectRefusesAnActiveAccount(t *testing.T) {
+	_, nc := newSealTestDaemon(t, &sealTestChecker{terminating: map[string]bool{sealTestTerminating: true}})
+
+	raw := sealRequest(t, nc, ec2v1.SealAccountSettingsRequest{AccountID: sealTestActive}, "")
+	assert.Equal(t, awserrors.ErrorIncorrectState, replyErrCode(t, raw))
+}
+
+// A request without a valid account is refused, not sealed.
+func TestSealAccountSettingsSubjectRefusesAMissingAccount(t *testing.T) {
+	_, nc := newSealTestDaemon(t, &sealTestChecker{terminating: map[string]bool{sealTestTerminating: true}})
+
+	assert.Equal(t, awserrors.ErrorMissingParameter, replyErrCode(t, sealRequest(t, nc, ec2v1.SealAccountSettingsRequest{}, "")))
+}
+
+// A daemon that does not serve the subject leaves the request with no responder.
+func TestSealAccountSettingsSubjectHasNoResponderWithoutTheDaemon(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	_, err := natsRequestAs(nc, ec2v1.SealAccountSettingsSubject, []byte(`{"account_id":"111122223333"}`), "", time.Second)
+	require.ErrorIs(t, err, nats.ErrNoResponders)
+}

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/seal.go

@@ -0,0 +1,135 @@
+package account
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+
+	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"
+)
+
+// sealMaxAttempts bounds SealAccount's retry loop against a revision that
+// keeps moving under concurrent writes, so it fails closed instead of
+// spinning.
+const sealMaxAttempts = 5
+
+// sealedRecord is the fixed value a seal writes. Clearing both booleans
+// retains no tenant setting; an older reader that ignores state sees the
+// AWS defaults.
+var sealedRecord = &AccountSettingsRecord{State: AccountSettingsStateSealed}
+
+// AccountSettingsSeal is SealAccount's result: the revision at which the
+// stream leader accepted the sealed value.
+type AccountSettingsSeal struct {
+	Revision uint64
+}
+
+// AccountTerminationChecker reports whether IAM holds accountID as TERMINATING.
+// A missing record is (false, nil); only a failed read is an error.
+type AccountTerminationChecker interface {
+	AccountTerminating(ctx context.Context, accountID string) (bool, error)
+}
+
+// SetAccountTerminationChecker wires the precondition; until it is set, SealAccount refuses.
+func (s *AccountSettingsServiceImpl) SetAccountTerminationChecker(c AccountTerminationChecker) {
+	s.terminationChecker = c
+}
+
+// checkTerminating runs before any settings-KV call, so a refusal never writes.
+// Not TERMINATING, including a missing IAM record, is IncorrectState; a read
+// failure or an unwired checker is InternalError.
+func (s *AccountSettingsServiceImpl) checkTerminating(ctx context.Context, accountID string) error {
+	if s.terminationChecker == nil {
+		slog.ErrorContext(ctx, "Account settings seal has no termination checker wired", "accountID", accountID)
+		return errors.New(awserrors.ErrorInternalError)
+	}
+
+	terminating, err := s.terminationChecker.AccountTerminating(ctx, accountID)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to read account status for seal precondition", "accountID", accountID, "err", err)
+		return errors.New(awserrors.ErrorInternalError)
+	}
+	if !terminating {
+		return awserrors.Errorf(awserrors.ErrorIncorrectState, "account %s is not TERMINATING", accountID)
+	}
+	return nil
+}
+
+// isRevisionConflict reports a lost CAS race SealAccount should retry. Besides
+// the two sentinels, Create's retry against a delete marker returns a bare
+// *jetstream.APIError carrying a wrong-last-sequence code.
+func isRevisionConflict(err error) bool {
+	if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
+		return true
+	}
+	if apiErr, ok := errors.AsType[*jetstream.APIError](err); ok {
+		return apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence ||
+			apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequenceConstant
+	}
+	return false
+}
+
+// SealAccount seals accountID's settings record so no later mutation can
+// recreate it. It never relies on a read for its claim: the only success is a
+// leader-accepted write, evidenced by the returned revision.
+func (s *AccountSettingsServiceImpl) SealAccount(ctx context.Context, accountID string) (*AccountSettingsSeal, error) {
+	if accountID == "" {
+		return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required to seal settings")
+	}
+	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 seal must not target the global account")
+	}
+
+	if err := s.checkTerminating(ctx, accountID); err != nil {
+		return nil, err
+	}
+
+	data, err := json.Marshal(sealedRecord)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to encode sealed account settings record", "accountID", accountID, "err", err)
+		return nil, errors.New(awserrors.ErrorInternalError)
+	}
+
+	// settingsKey is never used here: a seal must act on exactly accountID
+	// and never fall back to the global key.
+	for range sealMaxAttempts {
+		var rev uint64
+		entry, getErr := s.settingsKV.Get(ctx, accountID)
+		switch {
+		case getErr == nil:
+			rev = entry.Revision()
+		case errors.Is(getErr, jetstream.ErrKeyNotFound):
+			rev = 0
+		default:
+			slog.ErrorContext(ctx, "Failed to read account settings for seal", "accountID", accountID, "err", getErr)
+			return nil, errors.New(awserrors.ErrorInternalError)
+		}
+
+		var newRev uint64
+		var writeErr error
+		if rev == 0 {
+			newRev, writeErr = s.settingsKV.Create(ctx, accountID, data)
+		} else {
+			newRev, writeErr = s.settingsKV.Update(ctx, accountID, data, rev)
+		}
+
+		switch {
+		case writeErr == nil:
+			return &AccountSettingsSeal{Revision: newRev}, nil
+		case isRevisionConflict(writeErr):
+			continue
+		default:
+			slog.ErrorContext(ctx, "Failed to write sealed account settings record", "accountID", accountID, "err", writeErr)
+			return nil, errors.New(awserrors.ErrorInternalError)
+		}
+	}
+
+	slog.ErrorContext(ctx, "Account settings seal exhausted attempts against concurrent writes", "accountID", accountID, "attempts", sealMaxAttempts)
+	return nil, errors.New(awserrors.ErrorInternalError)
+}

spinifex/domains/ec2/account/seal_test.go

@@ -0,0 +1,559 @@
+package account
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	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"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// stubTerminationChecker answers AccountTerminating from a fixed set, for
+// tests that exercise the KV paths rather than IAM itself.
+type stubTerminationChecker struct {
+	terminating map[string]bool
+	err         error
+}
+
+func (c *stubTerminationChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) {
+	if c.err != nil {
+		return false, c.err
+	}
+	return c.terminating[accountID], nil
+}
+
+// terminatingChecker answers true for exactly accountID.
+func terminatingChecker(accountID string) *stubTerminationChecker {
+	return &stubTerminationChecker{terminating: map[string]bool{accountID: true}}
+}
+
+// lastSettingsSeq reports the account-settings stream's last sequence, so a
+// test can assert a failed attempt wrote nothing.
+func lastSettingsSeq(t *testing.T, svc *AccountSettingsServiceImpl) uint64 {
+	t.Helper()
+	stream, err := svc.js.Stream(context.Background(), "KV_"+KVBucketAccountSettings)
+	require.NoError(t, err)
+	info, err := stream.Info(context.Background())
+	require.NoError(t, err)
+	return info.State.LastSeq
+}
+
+// mutationCase is one of the four settings mutations, called uniformly so a
+// test can table over them.
+type mutationCase struct {
+	name string
+	call func(svc *AccountSettingsServiceImpl, ctx context.Context) error
+}
+
+func mutationTable() []mutationCase {
+	return []mutationCase{
+		{"EnableEbs", func(svc *AccountSettingsServiceImpl, ctx context.Context) error {
+			_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+			return err
+		}},
+		{"DisableEbs", func(svc *AccountSettingsServiceImpl, ctx context.Context) error {
+			_, err := svc.DisableEbsEncryptionByDefault(ctx, &ec2.DisableEbsEncryptionByDefaultInput{}, testAccountID)
+			return err
+		}},
+		{"EnableSerial", func(svc *AccountSettingsServiceImpl, ctx context.Context) error {
+			_, err := svc.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID)
+			return err
+		}},
+		{"DisableSerial", func(svc *AccountSettingsServiceImpl, ctx context.Context) error {
+			_, err := svc.DisableSerialConsoleAccess(ctx, &ec2.DisableSerialConsoleAccessInput{}, testAccountID)
+			return err
+		}},
+	}
+}
+
+// seedSettingsState prepares testAccountID's record for a race test: present
+// (an active record), absent (never written) or tombstone (written then
+// deleted, leaving a delete marker Get hides).
+func seedSettingsState(t *testing.T, svc *AccountSettingsServiceImpl, state string) {
+	t.Helper()
+	ctx := context.Background()
+	switch state {
+	case "present":
+		_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+		require.NoError(t, err)
+	case "tombstone":
+		_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+		require.NoError(t, err)
+		require.NoError(t, svc.settingsKV.Delete(ctx, testAccountID))
+	case "absent":
+	default:
+		t.Fatalf("unknown seed state %q", state)
+	}
+}
+
+// sealRaceKV wraps a real KeyValue so a test can inject a deterministic race
+// between a caller's read and its guarded write (Create or Update), instead
+// of relying on goroutine timing.
+type sealRaceKV struct {
+	jetstream.KeyValue
+
+	writeCalls  int
+	beforeWrite func(attempt int) error
+}
+
+func (f *sealRaceKV) Create(ctx context.Context, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error) {
+	f.writeCalls++
+	if f.beforeWrite != nil {
+		if err := f.beforeWrite(f.writeCalls); err != nil {
+			return 0, err
+		}
+	}
+	return f.KeyValue.Create(ctx, key, value, opts...)
+}
+
+func (f *sealRaceKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) {
+	f.writeCalls++
+	if f.beforeWrite != nil {
+		if err := f.beforeWrite(f.writeCalls); err != nil {
+			return 0, err
+		}
+	}
+	return f.KeyValue.Update(ctx, key, value, revision)
+}
+
+// staleGetKV always answers Get with a fixed snapshot taken at construction,
+// so a caller's CAS keeps retrying against a revision that can never win.
+type staleGetKV struct {
+	jetstream.KeyValue
+
+	entry jetstream.KeyValueEntry
+}
+
+func (f *staleGetKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) {
+	return f.entry, nil
+}
+
+// staleNotFoundKV always answers Get as absent, regardless of what the real
+// bucket holds.
+type staleNotFoundKV struct {
+	jetstream.KeyValue
+}
+
+func (f *staleNotFoundKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) {
+	return nil, jetstream.ErrKeyNotFound
+}
+
+// errGetKV always fails Get with a fixed store error.
+type errGetKV struct {
+	jetstream.KeyValue
+
+	err error
+}
+
+func (f *errGetKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) {
+	return nil, f.err
+}
+
+func decodeSettings(t *testing.T, entry jetstream.KeyValueEntry) AccountSettingsRecord {
+	t.Helper()
+	var rec AccountSettingsRecord
+	require.NoError(t, json.Unmarshal(entry.Value(), &rec))
+	return rec
+}
+
+// A seal landing between a mutation's read and its guarded write is not
+// outlived: the mutation's retry sees the seal and is rejected, and nothing
+// it wrote survives.
+func TestSealRaceMutation_SealLandsBetweenReadAndWrite(t *testing.T) {
+	for _, m := range mutationTable() {
+		for _, state := range []string{"present", "absent", "tombstone"} {
+			t.Run(m.name+"_"+state, func(t *testing.T) {
+				svc := setupTestAccountService(t)
+				ctx := context.Background()
+				seedSettingsState(t, svc, state)
+
+				realKV := svc.settingsKV
+				sealer := &AccountSettingsServiceImpl{js: svc.js, settingsKV: realKV}
+				sealer.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+
+				var sealed *AccountSettingsSeal
+				race := &sealRaceKV{KeyValue: realKV}
+				race.beforeWrite = func(attempt int) error {
+					if attempt != 1 {
+						return nil
+					}
+					var sealErr error
+					sealed, sealErr = sealer.SealAccount(ctx, testAccountID)
+					return sealErr
+				}
+				svc.settingsKV = race
+
+				err := m.call(svc, ctx)
+				require.Error(t, err)
+				assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err)
+				require.NotNil(t, sealed)
+
+				entry, err := realKV.Get(ctx, testAccountID)
+				require.NoError(t, err)
+				assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State)
+				assert.Equal(t, sealed.Revision, entry.Revision(), "no write landed after the seal")
+			})
+		}
+	}
+}
+
+// A mutation landing inside a seal's read-write window does not survive it:
+// the seal retries against the mutation's write and wins, leaving the
+// sealed value with both booleans false.
+func TestSealRaceMutation_MutationLandsBetweenSealReadAndWrite(t *testing.T) {
+	for _, m := range mutationTable() {
+		for _, state := range []string{"present", "absent", "tombstone"} {
+			t.Run(m.name+"_"+state, func(t *testing.T) {
+				svc := setupTestAccountService(t)
+				ctx := context.Background()
+				seedSettingsState(t, svc, state)
+
+				realKV := svc.settingsKV
+				race := &sealRaceKV{KeyValue: realKV}
+				race.beforeWrite = func(attempt int) error {
+					if attempt != 1 {
+						return nil
+					}
+					_ = m.call(svc, ctx)
+					return nil
+				}
+
+				sealer := &AccountSettingsServiceImpl{js: svc.js, settingsKV: race}
+				sealer.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+
+				seal, err := sealer.SealAccount(ctx, testAccountID)
+				require.NoError(t, err)
+
+				entry, err := realKV.Get(ctx, testAccountID)
+				require.NoError(t, err)
+				rec := decodeSettings(t, entry)
+				assert.Equal(t, AccountSettingsStateSealed, rec.State)
+				assert.False(t, rec.EbsEncryptionByDefault)
+				assert.False(t, rec.SerialConsoleAccess)
+				assert.Equal(t, seal.Revision, entry.Revision())
+			})
+		}
+	}
+}
+
+// None of the four mutations can act on an already-sealed record.
+func TestMutationOnSealedRecordIsRejected(t *testing.T) {
+	for _, m := range mutationTable() {
+		t.Run(m.name, func(t *testing.T) {
+			svc := setupTestAccountService(t)
+			ctx := context.Background()
+			svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+			_, err := svc.SealAccount(ctx, testAccountID)
+			require.NoError(t, err)
+
+			before, err := svc.settingsKV.Get(ctx, testAccountID)
+			require.NoError(t, err)
+
+			err = m.call(svc, ctx)
+			require.Error(t, err)
+			assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err)
+
+			after, err := svc.settingsKV.Get(ctx, testAccountID)
+			require.NoError(t, err)
+			assert.Equal(t, before.Revision(), after.Revision())
+		})
+	}
+}
+
+// Every settings mutation races one seal on a real bucket; afterwards the
+// record is sealed and stays sealed, so no success among them could have
+// landed after the seal.
+func TestSealConcurrentMutations(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	mutations := mutationTable()
+	errs := make([]error, len(mutations))
+	var wg sync.WaitGroup
+	for i, m := range mutations {
+		wg.Add(1)
+		go func(i int, m mutationCase) {
+			defer wg.Done()
+			errs[i] = m.call(svc, ctx)
+		}(i, m)
+	}
+
+	var sealResult *AccountSettingsSeal
+	var sealErr error
+	wg.Go(func() {
+		sealResult, sealErr = svc.SealAccount(ctx, testAccountID)
+	})
+	wg.Wait()
+
+	require.NoError(t, sealErr)
+	require.NotNil(t, sealResult)
+	for i, err := range errs {
+		if err != nil {
+			assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "mutation %s: got %v", mutations[i].name, err)
+		}
+	}
+
+	entry, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State)
+	assert.Equal(t, sealResult.Revision, entry.Revision())
+
+	_, err = svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+}
+
+// A Get pinned to a revision the record has already moved past exhausts the
+// mutation's retries without ever writing.
+func TestMutationStaleReadExhaustsWithoutWriting(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+	stale, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+
+	_, err = svc.DisableEbsEncryptionByDefault(ctx, &ec2.DisableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	before := lastSettingsSeq(t, svc)
+	svc.settingsKV = &staleGetKV{KeyValue: svc.settingsKV, entry: stale}
+
+	_, err = svc.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+// A Get that always reports absent, while a sealed record actually exists,
+// never gets to recreate it: every Create attempt conflicts.
+func TestMutationStaleAbsentDoesNotRecreate(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	_, err := svc.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+
+	before := lastSettingsSeq(t, svc)
+	svc.settingsKV = &staleNotFoundKV{KeyValue: svc.settingsKV}
+
+	_, err = svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+// A revision that keeps moving under sustained concurrent writes must not
+// let the seal spin forever: it gives up, bounded by its attempt budget,
+// and its own write never lands.
+func TestSealSustainedConflictFailsWithoutWriting(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	realKV := svc.settingsKV
+	race := &sealRaceKV{KeyValue: realKV}
+	race.beforeWrite = func(int) error {
+		data, merr := json.Marshal(&AccountSettingsRecord{SerialConsoleAccess: true})
+		require.NoError(t, merr)
+		_, perr := realKV.Put(ctx, testAccountID, data)
+		return perr
+	}
+	svc.settingsKV = race
+
+	_, err = svc.SealAccount(ctx, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, sealMaxAttempts, race.writeCalls)
+
+	entry, err := realKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.NotEqual(t, AccountSettingsStateSealed, decodeSettings(t, entry).State, "the seal's own write never landed")
+}
+
+// A store error on the seal's read reports failed, never written.
+func TestSealStoreErrorWritesNothing(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+
+	before := lastSettingsSeq(t, svc)
+	svc.settingsKV = &errGetKV{KeyValue: svc.settingsKV, err: errors.New("kv unavailable")}
+
+	_, err := svc.SealAccount(ctx, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+func TestIsRevisionConflictRecognisesTombstoneRetry(t *testing.T) {
+	assert.True(t, isRevisionConflict(jetstream.ErrKeyExists))
+	assert.True(t, isRevisionConflict(jetstream.ErrKeyRevisionMismatch))
+	assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence}))
+	assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequenceConstant}))
+	assert.True(t, isRevisionConflict(fmt.Errorf("wrapped: %w", &jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence})))
+	assert.False(t, isRevisionConflict(errors.New("unrelated failure")))
+	assert.False(t, isRevisionConflict(nil))
+}
+
+// An account IAM does not hold as TERMINATING refuses the seal before any
+// settings-KV write, whether the key is present or absent.
+func TestSealRefusedForActiveAccount(t *testing.T) {
+	for _, state := range []string{"present", "absent"} {
+		t.Run(state, func(t *testing.T) {
+			svc := setupTestAccountService(t)
+			ctx := context.Background()
+			if state == "present" {
+				_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+				require.NoError(t, err)
+			}
+			svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}})
+
+			before := lastSettingsSeq(t, svc)
+			_, err := svc.SealAccount(ctx, testAccountID)
+			assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err)
+			assert.Equal(t, before, lastSettingsSeq(t, svc))
+		})
+	}
+}
+
+func TestSealRefusedForMissingIAMRecord(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}})
+
+	before := lastSettingsSeq(t, svc)
+	_, err := svc.SealAccount(ctx, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+func TestSealRefusedWhenIAMReadFails(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(&stubTerminationChecker{err: errors.New("iam unavailable")})
+
+	before := lastSettingsSeq(t, svc)
+	_, err := svc.SealAccount(ctx, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+func TestSealRefusedWithNoCheckerWired(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+
+	before := lastSettingsSeq(t, svc)
+	_, err := svc.SealAccount(ctx, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastSettingsSeq(t, svc))
+}
+
+func TestSealAcceptedForTerminatingAccount(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+
+	seal, err := svc.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.NotZero(t, seal.Revision)
+
+	entry, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, seal.Revision, entry.Revision())
+}
+
+// A sealed record stays sealed after the service is rebuilt over the same
+// JetStream, and every one of the six settings actions still refuses it.
+func TestSealSurvivesServiceReconstruction(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	ctx := context.Background()
+
+	svc, err := NewAccountSettingsServiceImplWithNATS(ctx, nil, nc)
+	require.NoError(t, err)
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	first, err := svc.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+
+	rebuilt, err := NewAccountSettingsServiceImplWithNATS(ctx, nil, nc)
+	require.NoError(t, err)
+
+	entry, err := rebuilt.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State)
+
+	_, err = rebuilt.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	_, err = rebuilt.GetSerialConsoleAccessStatus(ctx, &ec2.GetSerialConsoleAccessStatusInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	_, err = rebuilt.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	_, err = rebuilt.DisableEbsEncryptionByDefault(ctx, &ec2.DisableEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	_, err = rebuilt.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	_, err = rebuilt.DisableSerialConsoleAccess(ctx, &ec2.DisableSerialConsoleAccessInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+
+	rebuilt.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	second, err := rebuilt.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Greater(t, second.Revision, first.Revision)
+}
+
+func TestSealAbsentKeyCreatesSealedRecord(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+
+	_, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.ErrorIs(t, err, jetstream.ErrKeyNotFound)
+
+	seal, err := svc.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+
+	entry, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, seal.Revision, entry.Revision())
+	assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State)
+}
+
+func TestSealRefusesEmptyInvalidAndGlobalAccount(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+
+	_, err := svc.SealAccount(ctx, "")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorMissingParameter), "got %v", err)
+
+	_, err = svc.SealAccount(ctx, "not-an-account")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err)
+
+	_, err = svc.SealAccount(ctx, awsidentifiers.GlobalAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err)
+}
+
+func TestSealLeavesOtherAccountsUntouched(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	accountB := "222222222222"
+	_, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, accountB)
+	require.NoError(t, err)
+
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccountID))
+	_, err = svc.SealAccount(ctx, testAccountID)
+	require.NoError(t, err)
+
+	out, err := svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, accountB)
+	require.NoError(t, err)
+	assert.True(t, *out.EbsEncryptionByDefault)
+}

spinifex/domains/ec2/account/service_impl.go

@@ -5,12 +5,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"log/slog"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/nats-io/nats.go"
@@ -20,14 +21,19 @@ import (
 const (
 	KVBucketAccountSettings        = "spinifex-ec2-account-settings"
 	KVBucketAccountSettingsVersion = 1
-	KeyEbsEncryptionDefault        = "ebs-encryption-default"
-	KeySerialConsoleAccess         = "serial-console-access"
 )
 
+// AccountSettingsState is a settings record's lifecycle state. "" is active;
+// a record written before the state field existed has none and stays active.
+type AccountSettingsState string
+
+const AccountSettingsStateSealed AccountSettingsState = "sealed"
+
 // AccountSettingsRecord represents stored account settings.
 type AccountSettingsRecord struct {
-	EbsEncryptionByDefault bool `json:"ebs_encryption_by_default"`
-	SerialConsoleAccess    bool `json:"serial_console_access"`
+	EbsEncryptionByDefault bool                 `json:"ebs_encryption_by_default"`
+	SerialConsoleAccess    bool                 `json:"serial_console_access"`
+	State                  AccountSettingsState `json:"state,omitempty"`
 }
 
 // AccountSettingsServiceImpl implements account settings operations with NATS JetStream persistence.
@@ -35,6 +41,10 @@ type AccountSettingsServiceImpl struct {
 	config     *config.Config
 	js         jetstream.JetStream
 	settingsKV jetstream.KeyValue
+
+	// terminationChecker is the seal precondition; SealAccount refuses until
+	// it is wired.
+	terminationChecker AccountTerminationChecker
 }
 
 var _ AccountSettingsService = (*AccountSettingsServiceImpl)(nil)
@@ -72,50 +82,99 @@ func settingsKey(accountID string) string {
 	return accountID
 }
 
-// getSettings retrieves current account settings.
+// errSettingsSealed marks a mutation or read that found a sealed record, so
+// callers map it to Blocked without logging it as a fault.
+var errSettingsSealed = errors.New(awserrors.ErrorBlocked)
+
+// errSettingsContended marks a mutation whose CAS retries were exhausted
+// under concurrent writers, bounding the retry loop instead of spinning.
+var errSettingsContended = errors.New("account settings: CAS retries exhausted")
+
+// checkActive rejects a sealed or unknown-state record; "" is active and
+// passes through. A mutation must never act on a record that failed this.
+func checkActive(rec *AccountSettingsRecord) error {
+	switch rec.State {
+	case "":
+		return nil
+	case AccountSettingsStateSealed:
+		return errSettingsSealed
+	default:
+		return fmt.Errorf("account settings: unknown state %q", rec.State)
+	}
+}
+
+// mapSettingsError turns a mutateSettings or getSettings failure into the AWS
+// error a caller returns: sealed is an expected, unlogged Blocked; everything
+// else is InternalError, logged with the account ID.
+func mapSettingsError(ctx context.Context, accountID string, err error) error {
+	switch {
+	case err == nil:
+		return nil
+	case errors.Is(err, errSettingsSealed):
+		return errors.New(awserrors.ErrorBlocked)
+	case errors.Is(err, errSettingsContended):
+		slog.ErrorContext(ctx, "Account settings CAS retries exhausted under contention", "accountID", accountID)
+	case errors.Is(err, kvutil.ErrRead):
+		slog.ErrorContext(ctx, "Failed to read account settings from KV", "accountID", accountID, "err", err)
+	case errors.Is(err, kvutil.ErrDecode):
+		slog.ErrorContext(ctx, "Corrupt account settings record in KV", "accountID", accountID, "err", err)
+	case errors.Is(err, kvutil.ErrEncode):
+		slog.ErrorContext(ctx, "Failed to marshal account settings record", "accountID", accountID, "err", err)
+	case errors.Is(err, kvutil.ErrWrite):
+		slog.ErrorContext(ctx, "Failed to write account settings to KV", "accountID", accountID, "err", err)
+	default:
+		slog.ErrorContext(ctx, "Account settings operation failed", "accountID", accountID, "err", err)
+	}
+	return errors.New(awserrors.ErrorInternalError)
+}
+
+// mutateSettings applies set to accountID's record at the revision it read.
+// changed is always true, so every success is a leader-accepted write over an
+// active value; a seal landing between the read and the write fails the CAS.
+func (s *AccountSettingsServiceImpl) mutateSettings(ctx context.Context, accountID string, set func(*AccountSettingsRecord)) error {
+	_, err := kvutil.Update(ctx, s.settingsKV, settingsKey(accountID), kvutil.CASConfig{
+		CreateIfAbsent: true,
+		Exhausted:      func(string, int) error { return errSettingsContended },
+	}, func(rec *AccountSettingsRecord) (bool, error) {
+		if err := checkActive(rec); err != nil {
+			return false, err
+		}
+		set(rec)
+		return true, nil
+	})
+	return mapSettingsError(ctx, accountID, err)
+}
+
+// getSettings retrieves current account settings, rejecting a sealed or
+// unknown-state record so a read never answers from behind a seal. Absent is
+// the AWS defaults; a read is a direct get and can be stale.
 func (s *AccountSettingsServiceImpl) getSettings(ctx context.Context, accountID string) (*AccountSettingsRecord, error) {
 	entry, err := s.settingsKV.Get(ctx, settingsKey(accountID))
 	if err != nil {
 		if errors.Is(err, jetstream.ErrKeyNotFound) {
-			return &AccountSettingsRecord{
-				EbsEncryptionByDefault: false,
-				SerialConsoleAccess:    false,
-			}, nil
+			return &AccountSettingsRecord{}, nil
 		}
-		return nil, fmt.Errorf("failed to get account settings: %w", err)
+		slog.ErrorContext(ctx, "Failed to read account settings from KV", "accountID", accountID, "err", err)
+		return nil, errors.New(awserrors.ErrorInternalError)
 	}
 
 	var settings AccountSettingsRecord
 	if err := json.Unmarshal(entry.Value(), &settings); err != nil {
-		return nil, err
+		slog.ErrorContext(ctx, "Corrupt account settings record in KV", "accountID", accountID, "err", err)
+		return nil, errors.New(awserrors.ErrorInternalError)
+	}
+	if err := checkActive(&settings); err != nil {
+		return nil, mapSettingsError(ctx, accountID, err)
 	}
 
 	return &settings, nil
 }
 
-// saveSettings saves current account settings.
-func (s *AccountSettingsServiceImpl) saveSettings(ctx context.Context, settings *AccountSettingsRecord, accountID string) error {
-	data, err := json.Marshal(settings)
-	if err != nil {
-		return err
-	}
-
-	_, err = s.settingsKV.Put(ctx, settingsKey(accountID), data)
-	return err
-}
-
 // EnableEbsEncryptionByDefault enables EBS encryption by default for the account.
 func (s *AccountSettingsServiceImpl) EnableEbsEncryptionByDefault(ctx context.Context, input *ec2.EnableEbsEncryptionByDefaultInput, accountID string) (*ec2.EnableEbsEncryptionByDefaultOutput, error) {
 	slog.InfoContext(ctx, "EnableEbsEncryptionByDefault called", "accountID", accountID)
 
-	settings, err := s.getSettings(ctx, accountID)
-	if err != nil {
-		return nil, err
-	}
-
-	settings.EbsEncryptionByDefault = true
-
-	if err := s.saveSettings(ctx, settings, accountID); err != nil {
+	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.EbsEncryptionByDefault = true }); err != nil {
 		return nil, err
 	}
 
@@ -128,14 +187,7 @@ func (s *AccountSettingsServiceImpl) EnableEbsEncryptionByDefault(ctx context.Co
 func (s *AccountSettingsServiceImpl) DisableEbsEncryptionByDefault(ctx context.Context, input *ec2.DisableEbsEncryptionByDefaultInput, accountID string) (*ec2.DisableEbsEncryptionByDefaultOutput, error) {
 	slog.InfoContext(ctx, "DisableEbsEncryptionByDefault called", "accountID", accountID)
 
-	settings, err := s.getSettings(ctx, accountID)
-	if err != nil {
-		return nil, err
-	}
-
-	settings.EbsEncryptionByDefault = false
-
-	if err := s.saveSettings(ctx, settings, accountID); err != nil {
+	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.EbsEncryptionByDefault = false }); err != nil {
 		return nil, err
 	}
 
@@ -176,14 +228,7 @@ func (s *AccountSettingsServiceImpl) GetSerialConsoleAccessStatus(ctx context.Co
 func (s *AccountSettingsServiceImpl) EnableSerialConsoleAccess(ctx context.Context, input *ec2.EnableSerialConsoleAccessInput, accountID string) (*ec2.EnableSerialConsoleAccessOutput, error) {
 	slog.InfoContext(ctx, "EnableSerialConsoleAccess called", "accountID", accountID)
 
-	settings, err := s.getSettings(ctx, accountID)
-	if err != nil {
-		return nil, err
-	}
-
-	settings.SerialConsoleAccess = true
-
-	if err := s.saveSettings(ctx, settings, accountID); err != nil {
+	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.SerialConsoleAccess = true }); err != nil {
 		return nil, err
 	}
 
@@ -196,14 +241,7 @@ func (s *AccountSettingsServiceImpl) EnableSerialConsoleAccess(ctx context.Conte
 func (s *AccountSettingsServiceImpl) DisableSerialConsoleAccess(ctx context.Context, input *ec2.DisableSerialConsoleAccessInput, accountID string) (*ec2.DisableSerialConsoleAccessOutput, error) {
 	slog.InfoContext(ctx, "DisableSerialConsoleAccess called", "accountID", accountID)
 
-	settings, err := s.getSettings(ctx, accountID)
-	if err != nil {
-		return nil, err
-	}
-
-	settings.SerialConsoleAccess = false
-
-	if err := s.saveSettings(ctx, settings, accountID); err != nil {
+	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.SerialConsoleAccess = false }); err != nil {
 		return nil, err
 	}
 

spinifex/domains/ec2/account/service_impl_test.go

@@ -2,11 +2,13 @@ package account
 
 import (
 	"context"
+	"encoding/json"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/internal/testkit"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -143,3 +145,44 @@ func TestSettingsKey_EmptyAccountIDFallback(t *testing.T) {
 	assert.Equal(t, awsidentifiers.GlobalAccountID, settingsKey(""))
 	assert.Equal(t, "123456789012", settingsKey("123456789012"))
 }
+
+// A raw pre-seal record, with no state field at all, is readable and
+// mutable, and a mutation writes it back without adding one.
+func TestRecordWithoutStateIsActive(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	_, err := svc.settingsKV.Put(ctx, testAccountID, []byte(`{"ebs_encryption_by_default":true,"serial_console_access":false}`))
+	require.NoError(t, err)
+
+	out, err := svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+	assert.True(t, *out.EbsEncryptionByDefault)
+
+	_, err = svc.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID)
+	require.NoError(t, err)
+
+	entry, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	assert.NotContains(t, string(entry.Value()), "state")
+}
+
+// A record with an unrecognised state fails closed rather than being treated
+// as active or sealed, for both a read and a mutation.
+func TestUnknownStateFailsClosed(t *testing.T) {
+	svc := setupTestAccountService(t)
+	ctx := context.Background()
+	_, err := svc.settingsKV.Put(ctx, testAccountID, []byte(`{"ebs_encryption_by_default":false,"serial_console_access":false,"state":"quarantined"}`))
+	require.NoError(t, err)
+
+	_, err = svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+
+	_, err = svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+
+	entry, err := svc.settingsKV.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	var rec AccountSettingsRecord
+	require.NoError(t, json.Unmarshal(entry.Value(), &rec))
+	assert.EqualValues(t, "quarantined", rec.State, "the unknown-state record was never overwritten")
+}

spinifex/handlers/iam/account_status_reader.go

@@ -0,0 +1,56 @@
+package handlers_iam
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// AccountStatusReader reads account status from spinifex-accounts and never
+// writes. It is the read-only capability EC2's settings seal uses to check
+// its TERMINATING precondition.
+type AccountStatusReader struct {
+	accounts *kvstore.Bucket // AttachOnly: never creates the bucket
+}
+
+// NewAccountStatusReader attaches to the existing spinifex-accounts bucket.
+// It needs no master key and runs no migration: a caller that only reads
+// account status must not bootstrap IAM's bucket out from under it.
+func NewAccountStatusReader(js jetstream.JetStream) *AccountStatusReader {
+	return &AccountStatusReader{
+		accounts: kvstore.NewBucket(js, kvstore.Config{
+			Name:       KVBucketAccounts,
+			Missing:    "account status reader: nil jetstream client",
+			AttachOnly: true,
+		}),
+	}
+}
+
+// AccountTerminating is a direct read, which suffices because TERMINATING is
+// one-way until the record is deleted. A missing record is (false, nil); only
+// a failed read or decode is an error.
+func (r *AccountStatusReader) AccountTerminating(ctx context.Context, accountID string) (bool, error) {
+	kv, err := r.accounts.KV(ctx)
+	if err != nil {
+		return false, err
+	}
+
+	entry, err := kv.Get(ctx, accountID)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return false, nil
+		}
+		return false, err
+	}
+
+	var record struct {
+		Status string `json:"status"`
+	}
+	if err := json.Unmarshal(entry.Value(), &record); err != nil {
+		return false, err
+	}
+	return record.Status == AccountStatusTerminating, nil
+}

spinifex/handlers/iam/account_status_reader_test.go

@@ -0,0 +1,102 @@
+package handlers_iam_test
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	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"
+)
+
+// newStatusReaderFixture starts a real IAM service (which creates and
+// migrates spinifex-accounts) and a reader attached to the same JetStream.
+func newStatusReaderFixture(t *testing.T) (handlers_iam.IAMService, *handlers_iam.AccountStatusReader, jetstream.JetStream) {
+	t.Helper()
+	_, nc, js := testutil.StartTestJetStream(t)
+
+	masterKey, err := handlers_iam.GenerateMasterKey()
+	require.NoError(t, err)
+	svc, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, masterKey)
+	require.NoError(t, err)
+
+	return svc, handlers_iam.NewAccountStatusReader(js), js
+}
+
+func TestAccountStatusReader_TerminatingIsTrue(t *testing.T) {
+	t.Parallel()
+	svc, reader, _ := newStatusReaderFixture(t)
+	account, err := svc.CreateAccount("tenant@example.com")
+	require.NoError(t, err)
+	_, err = svc.SetAccountStatus(account.AccountID, handlers_iam.AccountStatusTerminating)
+	require.NoError(t, err)
+
+	terminating, err := reader.AccountTerminating(t.Context(), account.AccountID)
+	require.NoError(t, err)
+	assert.True(t, terminating)
+}
+
+func TestAccountStatusReader_ActiveAndSuspendedAreFalse(t *testing.T) {
+	t.Parallel()
+	svc, reader, _ := newStatusReaderFixture(t)
+	active, err := svc.CreateAccount("active@example.com")
+	require.NoError(t, err)
+	terminating, err := reader.AccountTerminating(t.Context(), active.AccountID)
+	require.NoError(t, err)
+	assert.False(t, terminating)
+
+	suspended, err := svc.CreateAccount("suspended@example.com")
+	require.NoError(t, err)
+	_, err = svc.SetAccountStatus(suspended.AccountID, handlers_iam.AccountStatusSuspended)
+	require.NoError(t, err)
+	terminating, err = reader.AccountTerminating(t.Context(), suspended.AccountID)
+	require.NoError(t, err)
+	assert.False(t, terminating)
+}
+
+func TestAccountStatusReader_MissingRecordIsFalseNotError(t *testing.T) {
+	t.Parallel()
+	_, reader, _ := newStatusReaderFixture(t)
+	terminating, err := reader.AccountTerminating(t.Context(), "999999999999")
+	require.NoError(t, err)
+	assert.False(t, terminating)
+}
+
+// A missing bucket is an error, and the reader never creates one: a caller
+// that only reads account status must not bootstrap IAM's bucket.
+func TestAccountStatusReader_MissingBucketIsErrorAndNeverCreated(t *testing.T) {
+	t.Parallel()
+	_, nc, js := testutil.StartTestJetStream(t)
+	reader := handlers_iam.NewAccountStatusReader(js)
+
+	_, err := reader.AccountTerminating(t.Context(), "111111111111")
+	require.Error(t, err)
+
+	_, bucketErr := js.KeyValue(t.Context(), handlers_iam.KVBucketAccounts)
+	assert.ErrorIs(t, bucketErr, jetstream.ErrBucketNotFound)
+	_ = nc
+}
+
+func TestAccountStatusReader_UndecodableValueIsError(t *testing.T) {
+	t.Parallel()
+	_, nc, js := testutil.StartTestJetStream(t)
+	_, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, mustMasterKey(t))
+	require.NoError(t, err)
+
+	kv, err := js.KeyValue(t.Context(), handlers_iam.KVBucketAccounts)
+	require.NoError(t, err)
+	_, err = kv.Put(t.Context(), "222222222222", []byte(`{"status":`))
+	require.NoError(t, err)
+
+	reader := handlers_iam.NewAccountStatusReader(js)
+	_, err = reader.AccountTerminating(t.Context(), "222222222222")
+	require.Error(t, err)
+}
+
+func mustMasterKey(t *testing.T) []byte {
+	t.Helper()
+	key, err := handlers_iam.GenerateMasterKey()
+	require.NoError(t, err)
+	return key
+}