feat(quota): seal per-account quota overrides behind a revision-guarded owner action (#1245)

Store the override record with a state field kept separate from the admin
wire type. Every set and clear is revision-guarded and refuses a sealed
record with Blocked, and every read of a sealed record, including
admission, returns Blocked. The read-failure fallback to configured limits
is unchanged.

Add Service.SealAccountOverrides, accepted only when IAM already holds the
account as TERMINATING. It creates the override bucket if needed after that
check, creates or updates the record as sealed, and succeeds only on a
leader-accepted write, returning its revision. awsgw serves it on
admission.v1.SealAccountQuotaOverride. Nothing calls it yet.

Replace the unwired delete-based CleanupAccount.

contracts/admission/v1/README.md

@@ -0,0 +1,42 @@
+# Admission contract, version 1
+
+This package owns the cross-process contract for the admission domain's
+quota override. It is not an in-process service interface or the owner of
+quota state.
+
+## Seal account quota override
+
+`SealAccountQuotaOverrideSubject` (`admission.v1.SealAccountQuotaOverride`) is
+a request/reply route on which a gateway, in the `awsgw-quota-seal` queue
+group, irreversibly seals one account's quota override record.
+`domains/admission/quota` owns it: `awsgw` serves it.
+`accountteardown` is the consumer: it calls this subject to fence every
+override writer at the quota owner before the account record is removed.
+
+The request names the account in its body (`account_id`) and may carry an
+opaque `correlation_ref` that is echoed for logs.
+`awsgw` 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, or no termination checker
+wired, answers `InternalError`.
+Both refusals write nothing and never open the override bucket.
+
+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.
+Any current gateway may answer; the leader-accepted revision is the evidence
+whichever one does, and the override bucket is opened at seal time even on a
+gateway whose quotas were disabled at startup.
+
+### Compatibility boundary
+
+The subject name carries the contract major; `account_quota_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 gateway 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/admission/v1/account_quota_seal.go

@@ -0,0 +1,31 @@
+// Package admissionv1 holds the admission domain's cross-process contract:
+// the subjects, request/reply shapes and wire vocabulary the quota owner and
+// its consumers share.
+package admissionv1
+
+// SealAccountQuotaOverrideSubject is the request/reply subject on which a
+// gateway seals one account's quota override record. Sealing is irreversible
+// and refused unless IAM already holds the account as TERMINATING.
+const SealAccountQuotaOverrideSubject = "admission.v1.SealAccountQuotaOverride"
+
+// AccountQuotaOverrideDispositionSealed is the only disposition a successful
+// reply carries: a reply is sent only on success, so there is no other value.
+const AccountQuotaOverrideDispositionSealed = "sealed"
+
+// SealAccountQuotaOverrideRequest asks for one account's quota override 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 SealAccountQuotaOverrideRequest struct {
+	AccountID      string `json:"account_id"`
+	CorrelationRef string `json:"correlation_ref,omitempty"`
+}
+
+// SealAccountQuotaOverrideReply confirms accountID's quota override 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 SealAccountQuotaOverrideReply struct {
+	AccountID      string `json:"account_id"`
+	CorrelationRef string `json:"correlation_ref,omitempty"`
+	Disposition    string `json:"disposition"`
+	Revision       uint64 `json:"revision"`
+}

contracts/admission/v1/account_quota_seal_test.go

@@ -0,0 +1,43 @@
+package admissionv1
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSealAccountQuotaOverrideSubject(t *testing.T) {
+	require.Equal(t, "admission.v1.SealAccountQuotaOverride", SealAccountQuotaOverrideSubject)
+}
+
+func TestSealAccountQuotaOverrideRequestJSONContract(t *testing.T) {
+	data, err := json.Marshal(SealAccountQuotaOverrideRequest{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(SealAccountQuotaOverrideRequest{AccountID: "111122223333"})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"account_id":"111122223333"}`, string(data))
+}
+
+func TestSealAccountQuotaOverrideReplyJSONContract(t *testing.T) {
+	reply := SealAccountQuotaOverrideReply{
+		AccountID:      "111122223333",
+		CorrelationRef: "req-1",
+		Disposition:    AccountQuotaOverrideDispositionSealed,
+		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(SealAccountQuotaOverrideReply{AccountID: "111122223333", Disposition: AccountQuotaOverrideDispositionSealed})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"account_id":"111122223333","disposition":"sealed","revision":0}`, string(data))
+}

contracts/ec2/v1/README.md

@@ -65,7 +65,7 @@ An incompatible change to the subject, fields or vocabulary needs a new major.
 
 `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.
+`accountteardown`'s StageAccount seal reaper is the consumer: it calls this subject only after the seal-capability probe below reports every configured node compatible, and completes only once it holds a successful sealed reply.
 
 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.
@@ -82,3 +82,20 @@ The subject name carries the contract major; `account_settings_seal_test.go` pre
 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.
+
+## Seal-account-settings capability
+
+`AccountSettingsSealCapabilitySubject(node)` (`ec2.v1.AccountSettingsSealCapability.<node>`) is a per-node, read-only request/reply route on which `accountteardown` asks one configured daemon whether it supports the account-settings seal, immediately before every seal attempt.
+EC2 owns it: the daemon serves it with no queue group, so a request addressed to one node reaches only subscribers configured as that node, and two daemons sharing a node ID both answer.
+`accountteardown` is the consumer: its StageAccount seal reaper probes every configured node and sends no seal request unless every one answers compatible.
+
+`AccountSettingsSealCapabilityVersion` is a claim about the whole binary's writers, not about the responder: version 1 means every EC2 account-settings writer in the binary is revision-guarded and refuses a sealed record, and the binary serves the seal subject.
+A daemon binary without this responder — predating the capability, or carrying only the seal subject without it — has no subscriber, so the probe for that node reads as incomplete (`no-reply`), the same as a stopped or partitioned node.
+The reply proves nothing beyond the moment it was sent: a daemon downgraded or replaced after a successful probe is outside the guarantee.
+
+### Compatibility boundary
+
+The subject name carries the contract major; `account_settings_seal_capability_test.go` preserves the version-1 JSON shape and the subject builder.
+Producer and consumer are in the coordinated Spinifex release, so the window is N and N-1.
+A daemon that predates the capability has no subscription, so the probe gets no responder for that node.
+An incompatible change to the subject or fields needs a new major.

contracts/ec2/v1/account_settings_seal_capability.go

@@ -0,0 +1,39 @@
+package ec2v1
+
+import "strings"
+
+// AccountSettingsSealCapabilityVersion is the version this binary reports.
+// Version 1: every EC2 account-settings writer in the binary is
+// revision-guarded and refuses a sealed record, and the seal subject is served.
+const AccountSettingsSealCapabilityVersion = 1
+
+// AccountSettingsSealCapabilitySubject is one daemon's read-only
+// seal-capability route, addressed per node so a probe reaches exactly the
+// node it asks about.
+func AccountSettingsSealCapabilitySubject(node string) string {
+	return "ec2.v1.AccountSettingsSealCapability." + node
+}
+
+// ValidSealCapabilityNode reports whether node is exactly one NATS subject
+// token. Responder and probe both check it before building a subject, so
+// neither can address a wildcard or another node's route.
+func ValidSealCapabilityNode(node string) bool {
+	return node != "" && !strings.ContainsAny(node, ".*>\t\n\r ")
+}
+
+// AccountSettingsSealCapabilityRequest asks node whether it supports the
+// account-settings seal. CorrelationRef is an opaque caller reference echoed
+// for logs; it grants nothing and is never a fence or authority to write.
+type AccountSettingsSealCapabilityRequest struct {
+	Node           string `json:"node"`
+	CorrelationRef string `json:"correlation_ref,omitempty"`
+}
+
+// AccountSettingsSealCapabilityReply reports the responding node's capability
+// version. It carries no account data and proves nothing beyond the moment it
+// was sent.
+type AccountSettingsSealCapabilityReply struct {
+	Node              string `json:"node"`
+	CapabilityVersion int    `json:"capability_version"`
+	CorrelationRef    string `json:"correlation_ref,omitempty"`
+}

contracts/ec2/v1/account_settings_seal_capability_test.go

@@ -0,0 +1,43 @@
+package ec2v1
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAccountSettingsSealCapabilitySubject(t *testing.T) {
+	require.Equal(t, "ec2.v1.AccountSettingsSealCapability.node-1", AccountSettingsSealCapabilitySubject("node-1"))
+	require.Equal(t, "ec2.v1.AccountSettingsSealCapability.node-2", AccountSettingsSealCapabilitySubject("node-2"))
+}
+
+func TestAccountSettingsSealCapabilityRequestJSONContract(t *testing.T) {
+	data, err := json.Marshal(AccountSettingsSealCapabilityRequest{Node: "node-1", CorrelationRef: "req-1"})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"node":"node-1","correlation_ref":"req-1"}`, string(data))
+
+	data, err = json.Marshal(AccountSettingsSealCapabilityRequest{Node: "node-1"})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"node":"node-1"}`, string(data))
+}
+
+func TestAccountSettingsSealCapabilityReplyJSONContract(t *testing.T) {
+	reply := AccountSettingsSealCapabilityReply{Node: "node-1", CapabilityVersion: 1, CorrelationRef: "req-1"}
+	data, err := json.Marshal(reply)
+	require.NoError(t, err)
+	require.JSONEq(t, `{"node":"node-1","capability_version":1,"correlation_ref":"req-1"}`, string(data))
+
+	data, err = json.Marshal(AccountSettingsSealCapabilityReply{Node: "node-1", CapabilityVersion: 1})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"node":"node-1","capability_version":1}`, string(data))
+}
+
+func TestValidSealCapabilityNode(t *testing.T) {
+	for _, node := range []string{"node-1", "node_2", "n3"} {
+		require.True(t, ValidSealCapabilityNode(node), node)
+	}
+	for _, node := range []string{"", "a.b", "*", ">", "node-*", "node >", "node\t1", "node\n1"} {
+		require.False(t, ValidSealCapabilityNode(node), "%q", node)
+	}
+}

contracts/viperblockd/legacy/v1/volume.go

@@ -48,11 +48,6 @@ type EBSRequest struct {
 	NBDURI              string `json:"NBDURI"`
 	DeviceName          string `json:"DeviceName"`
 	HotplugPort         int    `json:"HotplugPort,omitempty"`
-	// BlockDevice and BlockSizeBytes describe a host block device export, the
-	// alternative to NBDURI; the guest sees BlockSizeBytes of it. Exactly one
-	// of NBDURI and BlockDevice may be set.
-	BlockDevice    string `json:"BlockDevice,omitempty"`
-	BlockSizeBytes uint64 `json:"BlockSizeBytes,omitempty"`
 }
 
 // EBSMountResponse is the response to an EBSRequest on MountSubject.

contracts/viperblockd/legacy/v1/volume_test.go

@@ -35,18 +35,6 @@ func TestEBSRequestJSONContract(t *testing.T) {
 		"NBDURI":"nbd:unix:/run/spinifex/vol-123.sock",
 		"DeviceName":"/dev/sdf", "HotplugPort":3
 	}`, string(data))
-
-	block, err := json.Marshal(EBSRequest{
-		Name:           "vol-456",
-		BlockDevice:    "/dev/sdx",
-		BlockSizeBytes: 10 << 30,
-	})
-	require.NoError(t, err)
-	require.JSONEq(t, `{
-		"Name":"vol-456", "VolType":"", "Boot":false, "EFI":false,
-		"DeleteOnTermination":false, "NBDURI":"", "DeviceName":"",
-		"BlockDevice":"/dev/sdx", "BlockSizeBytes":10737418240
-	}`, string(block))
 }
 
 func TestVolumeResponseJSONContracts(t *testing.T) {

docs/PACKAGE_BOUNDARY_MIGRATION.md

@@ -446,7 +446,7 @@ These residuals were recorded by individual slices and must be resolved once the
 - `providers/ebs/viperblock` → `runtime/compute/nbd` and `runtime/host/process` (dependency, moved unchanged from `services/viperblockd`): the adapter formats NBD endpoints and manages nbdkit processes through runtime packages. ADR-0001 does not list providers → runtime as forbidden and the layering test does not enforce it (same category as `providers/ebs/conformance` → `runtime/compute/nbd`), but a provider should not depend on node runtime. Clears when NBD endpoint naming has a provider-facing home (a candidate for `contracts/ebs/v1`) and process supervision is injected by the role. The adapter also imports `providers/objectstore`, a provider-to-provider edge that is allowed.
 - `gateway/ec2.go` (registration): the EC2 dispatch map stays a `GatewayConfig` method in `gateway` and imports every `domains/ec2/awsapi/*` package; `gateway/ec2_idempotency.go` also imports `domains/ec2/awsapi/idem` (`KVBucket`, `TTL`, `TokenAndParams`) for the dispatch-time idempotency wrapper. ADR-0001 makes the domain own its action registration; moving the map waits on the generic ingress registration seam and must not be done as a package move.
 - `gateway/requestaudit.go`: the outermost request-audit middleware records AWS fields and is updated from four places in the gateway's authentication flow through the gateway's own context key. Moving it would export about seven names used only by that flow, so it stays in `gateway` until the generic ingress registration seam defines how ingress middleware receives authentication results.
-- `accountteardown` (dependency): its EC2 reapers call `domains/ec2/awsapi/*` action adapters directly, as they did under `gateway/ec2`. The ADR's `accountteardown` current-to-target map row replaces this with domain-owned cleanup capabilities before the package moves. It also holds a read-only consumer of the EC2 platform-inventory contract (`contracts/ec2/v1`, `ec2.v1.PlatformInventory`), served by `domains/ec2/accountinventory`; only the explicit dry run (`Engine.DryRun`) calls it, to log a shadow comparison, and the key-pair, placement-group and launch-template reapers still use the `awsapi` adapters. A domain-owned cleanup capability for the proven-safe subset of those three kinds (key pairs and empty placement groups no instance, spot request or live launch-template version names, and launch-template residue; never live launch templates) exists as `domains/ec2/accountcleanup` (a Go interface with injected owners and reference readers, no NATS subject); nothing calls it yet.
+- `accountteardown` (dependency): its EC2 reapers call `domains/ec2/awsapi/*` action adapters directly, as they did under `gateway/ec2`. The ADR's `accountteardown` current-to-target map row replaces this with domain-owned cleanup capabilities before the package moves. It also holds a read-only consumer of the EC2 platform-inventory contract (`contracts/ec2/v1`, `ec2.v1.PlatformInventory`), served by `domains/ec2/accountinventory`; only the explicit dry run (`Engine.DryRun`) calls it, to log a shadow comparison, and the key-pair, placement-group and launch-template reapers still use the `awsapi` adapters. A domain-owned cleanup capability for the proven-safe subset of those three kinds (key pairs and empty placement groups no instance, spot request or live launch-template version names, and launch-template residue; never live launch templates) exists as `domains/ec2/accountcleanup` (a Go interface with injected owners and reference readers, no NATS subject); nothing calls it yet. It also holds the StageAccount settings-seal reaper, a consumer of `contracts/ec2/v1`'s `ec2.v1.SealAccountSettings` and per-node `ec2.v1.AccountSettingsSealCapability.<node>` subjects, served by `domains/ec2/account` through the daemon; the reaper reads no KV and imports no domain package, reaching both subjects only through NATS and the configured node set `NewClusterEngine` already receives.
 - `domains/admission/quota` (dependency, moved from `handlers/quota` unchanged): `live.go` imports `gateway/elbv2` and `handlers/rds` for its live ELBv2 and RDS counters; `records.go` imports `runtime/compute/vm` for the instance-record vCPU sweep; `live.go`, `vcpu.go` and `storage.go` call `domains/ec2/awsapi/{vpc,eip,volume}` describe adapters to count VPCs, subnets, addresses and volumes, as they did under `gateway/ec2`. The ADR's `handlers/quota` current-to-target map row replaces these with reader capabilities once their final locations exist.
 - `domains/ec2/awsapi/instance` (dependency): imports `handlers/iam` (`IAMService` in `DescribeInstances.go`, `RunInstances.go` and `IamInstanceProfileAssociation.go`; `InstanceProfile` in the latter two), `runtime/compute/cache` (`NodeState`, `NodeUnknown`, `NodeLive`) and `runtime/compute/vm` (`VM`), both in `status_synthesis.go`. The IAM edge needs an IAM capability; the runtime edges are domain-to-runtime and need the instance record and node liveness exposed through an EC2-owned projection.
 - `domains/ec2/awsapi/spotinstance` (dependency): imports `handlers/iam.IAMService` and `domains/admission/quota.Service` in `RequestSpotInstances.go`.

spinifex/accountteardown/cluster.go

@@ -70,6 +70,11 @@ func NewClusterEngine(
 	reapers = append(reapers, S3Reapers(buckets)...)
 	reapers = append(reapers, ACMReapers(nc)...)
 	reapers = append(reapers, IAMReapers(svc)...)
+
+	// Before the account counters and reservations: the seal must still be
+	// able to find the account IAM holds as TERMINATING, which the account
+	// record reaper does not remove.
+	reapers = append(reapers, EC2AccountSettingsReapers(nc, nodeIDs)...)
 	reapers = append(reapers, AccountReapers(svc, names, usage)...)
 
 	engine := NewEngine(NewIAMAccounts(svc), reapers...)

spinifex/accountteardown/cluster_test.go

@@ -41,6 +41,27 @@ func TestNewClusterEngineCoversEveryStage(t *testing.T) {
 	}
 }
 
+// The EC2 settings seal reaper must be registered in StageAccount on every
+// surface that builds a cluster engine, or teardown would finalise accounts
+// without ever sealing their settings.
+func TestNewClusterEngineRegistersTheEC2SettingsSealReaper(t *testing.T) {
+	_, nc, _ := 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)
+
+	engine, err := NewClusterEngine(t.Context(), nc, 1, []string{"node-1"}, svc, newFakeBuckets())
+	require.NoError(t, err)
+
+	var kinds []string
+	for _, reaper := range engine.reapersFor(StageAccount) {
+		kinds = append(kinds, reaper.Kind())
+	}
+	assert.Contains(t, kinds, "ec2-account-settings")
+}
+
 // A teardown must refuse the accounts no credential may delete before it reads
 // anything, and refuse a name confirmation that does not match the record.
 func TestClusterEnginePrechecks(t *testing.T) {

spinifex/accountteardown/ec2_account_settings_seal.go

@@ -0,0 +1,125 @@
+package accountteardown
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+)
+
+// ec2AccountSettingsSealTimeout bounds one seal request; an earlier caller
+// deadline wins.
+const ec2AccountSettingsSealTimeout = 30 * time.Second
+
+// ec2AccountSettingsSealer is EC2's seal of one account's settings record.
+type ec2AccountSettingsSealer interface {
+	SealAccountSettings(ctx context.Context, accountID string) (revision uint64, err error)
+}
+
+// ec2AccountSettingsSealClient reaches the EC2 settings-seal contract over
+// NATS. Only a reply for this account, disposition sealed and a non-zero
+// revision is a seal; anything else, including no responder, is an error.
+type ec2AccountSettingsSealClient struct {
+	nc      *nats.Conn
+	timeout time.Duration
+}
+
+var _ ec2AccountSettingsSealer = (*ec2AccountSettingsSealClient)(nil)
+
+func newEC2AccountSettingsSealClient(nc *nats.Conn) *ec2AccountSettingsSealClient {
+	return &ec2AccountSettingsSealClient{nc: nc, timeout: ec2AccountSettingsSealTimeout}
+}
+
+func (c *ec2AccountSettingsSealClient) SealAccountSettings(ctx context.Context, accountID string) (uint64, error) {
+	timeout := c.timeout
+	if deadline, ok := ctx.Deadline(); ok {
+		timeout = min(timeout, time.Until(deadline))
+	}
+	if ctx.Err() != nil || timeout <= 0 {
+		return 0, fmt.Errorf("seal account settings: %w", context.DeadlineExceeded)
+	}
+
+	reply, err := natsmsg.NATSRequest[ec2v1.SealAccountSettingsReply](ctx, c.nc, ec2v1.SealAccountSettingsSubject,
+		&ec2v1.SealAccountSettingsRequest{AccountID: accountID}, timeout, accountID)
+	if err != nil {
+		return 0, fmt.Errorf("seal account settings: %w", err)
+	}
+	if reply.AccountID != accountID {
+		return 0, fmt.Errorf("seal account settings: reply named account %q, not %q", reply.AccountID, accountID)
+	}
+	if reply.Disposition != ec2v1.AccountSettingsDispositionSealed {
+		return 0, fmt.Errorf("seal account settings: unexpected disposition %q", reply.Disposition)
+	}
+	if reply.Revision == 0 {
+		return 0, errors.New("seal account settings: reply carried no revision")
+	}
+	return reply.Revision, nil
+}
+
+// accountSettingsSealReaper seals EC2's settings record in StageAccount. List
+// does no I/O and stays pending until this instance has sealed, so DryRun never
+// seals; state is per engine, so every fresh teardown run seals again.
+type accountSettingsSealReaper struct {
+	sealer     ec2AccountSettingsSealer
+	capability sealCapabilityChecker
+
+	mu     sync.Mutex
+	sealed map[string]uint64 // account ID -> leader revision, this run only
+}
+
+func (r *accountSettingsSealReaper) Kind() string { return "ec2-account-settings" }
+func (r *accountSettingsSealReaper) Stage() Stage { return StageAccount }
+
+func (r *accountSettingsSealReaper) alreadySealed(accountID string) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	_, ok := r.sealed[accountID]
+	return ok
+}
+
+func (r *accountSettingsSealReaper) List(_ context.Context, accountID string) ([]Resource, error) {
+	if r.alreadySealed(accountID) {
+		return nil, nil
+	}
+	return []Resource{{Kind: r.Kind(), ID: accountID, Detail: "seal"}}, nil
+}
+
+// Delete probes every configured daemon immediately before each seal attempt,
+// never reusing an earlier probe. Incomplete capability sends no seal and is
+// retryable; force is ignored, since nothing may force past the probe.
+func (r *accountSettingsSealReaper) Delete(ctx context.Context, accountID string, _ Resource, _ bool) error {
+	report := r.capability.CheckSealCapability(ctx, accountID)
+	if !report.Complete {
+		return errSealCapabilityIncomplete(report)
+	}
+
+	revision, err := r.sealer.SealAccountSettings(ctx, accountID)
+	if err != nil {
+		return err
+	}
+
+	r.mu.Lock()
+	if r.sealed == nil {
+		r.sealed = map[string]uint64{}
+	}
+	r.sealed[accountID] = revision
+	r.mu.Unlock()
+	slog.InfoContext(ctx, "Account teardown sealed EC2 account settings", "accountID", accountID, "revision", revision)
+	return nil
+}
+
+// EC2AccountSettingsReapers returns the StageAccount reaper that seals EC2's
+// settings record. nodeIDs is the configured node set the capability probe
+// treats as "every relevant writer"; an empty set is unknown, never complete.
+func EC2AccountSettingsReapers(nc *nats.Conn, nodeIDs []string) []Reaper {
+	return []Reaper{&accountSettingsSealReaper{
+		sealer:     newEC2AccountSettingsSealClient(nc),
+		capability: newSealCapabilityClient(nc, nodeIDs),
+	}}
+}

spinifex/accountteardown/ec2_account_settings_seal_capability.go

@@ -0,0 +1,190 @@
+package accountteardown
+
+import (
+	"context"
+	"encoding/json"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+)
+
+// sealCapabilityWait bounds the whole probe; shortened by the caller's
+// deadline. Every configured node is probed in parallel against this one
+// wait, so N nodes cost one wait, not N.
+const sealCapabilityWait = 3 * time.Second
+
+// sealCapabilityGrace is how long a node's probe keeps listening after its
+// one expected responder has answered, long enough for a genuine second
+// subscriber on the same node to be caught as a duplicate.
+const sealCapabilityGrace = 150 * time.Millisecond
+
+// Seal capability reasons, one per failing node. Each names a category, never
+// wire error text.
+const (
+	sealCapabilityReasonNoReply       = "no-reply"
+	sealCapabilityReasonDuplicate     = "duplicate"
+	sealCapabilityReasonWrongIdentity = "wrong-identity"
+	sealCapabilityReasonErrorReply    = "error-reply"
+	sealCapabilityReasonMalformed     = "malformed"
+	sealCapabilityReasonOldVersion    = "old-version"
+	sealCapabilityReasonTransport     = "transport"
+	sealCapabilityReasonInvalidNodeID = "invalid-node-id"
+	sealCapabilityReasonSetUnknown    = "configured-node-set-unknown"
+)
+
+// sealCapabilityChecker reports whether every configured daemon supports the
+// seal.
+type sealCapabilityChecker interface {
+	CheckSealCapability(ctx context.Context, correlationRef string) sealCapabilityReport
+}
+
+// sealCapabilityReport is one probe's result, one entry per configured node.
+type sealCapabilityReport struct {
+	Complete bool
+	Nodes    []nodeSealCapability // one per configured node, sorted by node
+}
+
+// nodeSealCapability is one configured node's compatibility. Reason is ""
+// when the node is compatible.
+type nodeSealCapability struct {
+	Node    string
+	Version int    // 0 unless a valid reply arrived
+	Reason  string // "" when compatible
+}
+
+// sealCapabilityClient probes every configured node's seal-capability subject
+// in parallel. "Configured" is the same node set NewClusterEngine already
+// receives; there is no separate liveness or membership source.
+type sealCapabilityClient struct {
+	nc      *nats.Conn
+	nodeIDs []string
+	wait    time.Duration // 3 s, shortened by the caller's deadline
+	grace   time.Duration // 150 ms ResponderGrace
+}
+
+var _ sealCapabilityChecker = (*sealCapabilityClient)(nil)
+
+func newSealCapabilityClient(nc *nats.Conn, nodeIDs []string) *sealCapabilityClient {
+	return &sealCapabilityClient{nc: nc, nodeIDs: nodeIDs, wait: sealCapabilityWait, grace: sealCapabilityGrace}
+}
+
+// CheckSealCapability probes every configured node in parallel, bounded by
+// one shared deadline, run immediately before each seal attempt. A cached or
+// earlier result is never reused.
+func (c *sealCapabilityClient) CheckSealCapability(ctx context.Context, correlationRef string) sealCapabilityReport {
+	if len(c.nodeIDs) == 0 {
+		return sealCapabilityReport{Complete: false}
+	}
+
+	wait := c.wait
+	if deadline, ok := ctx.Deadline(); ok {
+		wait = min(wait, time.Until(deadline))
+	}
+
+	nodes := append([]string(nil), c.nodeIDs...)
+	sort.Strings(nodes)
+
+	results := make([]nodeSealCapability, len(nodes))
+	if ctx.Err() != nil || wait <= 0 {
+		for i, node := range nodes {
+			results[i] = nodeSealCapability{Node: node, Reason: sealCapabilityReasonNoReply}
+		}
+		return sealCapabilityReport{Complete: false, Nodes: results}
+	}
+
+	var wg sync.WaitGroup
+	for i, node := range nodes {
+		wg.Add(1)
+		go func(i int, node string) {
+			defer wg.Done()
+			results[i] = c.checkNode(ctx, node, correlationRef, wait)
+		}(i, node)
+	}
+	wg.Wait()
+
+	complete := true
+	for _, r := range results {
+		if r.Reason != "" {
+			complete = false
+			break
+		}
+	}
+	return sealCapabilityReport{Complete: complete, Nodes: results}
+}
+
+// checkNode probes one node and classifies the result. A node is compatible
+// only when exactly one identified, decodable, matching, version >= 1 frame
+// arrived, with no duplicate, conflict or cap hit.
+func (c *sealCapabilityClient) checkNode(ctx context.Context, node, correlationRef string, wait time.Duration) nodeSealCapability {
+	if !ec2v1.ValidSealCapabilityNode(node) {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonInvalidNodeID}
+	}
+
+	data, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityRequest{Node: node, CorrelationRef: correlationRef})
+	if err != nil {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonMalformed}
+	}
+
+	frames, sum, err := natsmsg.Gather(ctx, c.nc, ec2v1.AccountSettingsSealCapabilitySubject(node), data, natsmsg.GatherOpts{
+		Timeout:            wait,
+		ExpectedResponders: 1,
+		Mode:               natsmsg.CollectUntilDeadline,
+		ResponderGrace:     c.grace,
+	})
+	if err != nil {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonTransport}
+	}
+
+	switch {
+	case sum.CapHit, len(sum.ConflictNodes) > 0, sum.DuplicateFrames > 0:
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonDuplicate}
+	case sum.Unidentified > 0:
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonWrongIdentity}
+	case len(sum.ErrorResponders) > 0:
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonErrorReply}
+	case sum.Received == 0, len(frames) == 0:
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonNoReply}
+	}
+
+	frame := frames[0]
+	if frame.NodeID != node {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonWrongIdentity}
+	}
+
+	var reply ec2v1.AccountSettingsSealCapabilityReply
+	if json.Unmarshal(frame.Data, &reply) != nil {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonMalformed}
+	}
+	if reply.Node != node {
+		return nodeSealCapability{Node: node, Reason: sealCapabilityReasonWrongIdentity}
+	}
+	if reply.CapabilityVersion < 1 {
+		return nodeSealCapability{Node: node, Version: reply.CapabilityVersion, Reason: sealCapabilityReasonOldVersion}
+	}
+	return nodeSealCapability{Node: node, Version: reply.CapabilityVersion}
+}
+
+// errSealCapabilityIncomplete builds the retryable refusal from report. It
+// names node IDs and reason categories only, never wire error text.
+func errSealCapabilityIncomplete(report sealCapabilityReport) error {
+	detail := sealCapabilityReasonSetUnknown
+	var parts []string
+	for _, n := range report.Nodes {
+		if n.Reason != "" {
+			parts = append(parts, n.Node+"="+n.Reason)
+		}
+	}
+	if len(parts) > 0 {
+		detail = strings.Join(parts, ", ")
+	}
+	return awserrors.WithRetryAfter(
+		awserrors.Errorf(awserrors.ErrorServiceUnavailable, "EC2 account-settings seal capability incomplete: %s", detail),
+		5*time.Second,
+	)
+}

spinifex/accountteardown/ec2_account_settings_seal_capability_test.go

@@ -0,0 +1,311 @@
+package accountteardown
+
+//test:in-package — the probe client, its report types and reason constants
+// are unexported, and what they guarantee (incomplete never seals, complete
+// does) is only observable from inside the package.
+
+import (
+	"encoding/json"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// testCapabilityClient builds a probe client with test-sized timeouts so the
+// incomplete cases below do not each pay the production 3 s wait.
+func testCapabilityClient(nc *nats.Conn, nodeIDs []string) *sealCapabilityClient {
+	c := newSealCapabilityClient(nc, nodeIDs)
+	c.wait = 200 * time.Millisecond
+	c.grace = 30 * time.Millisecond
+	return c
+}
+
+// serveCapability subscribes reply on subject with no queue group, which is
+// how the real responder is addressed: per node, so two subscribers on the
+// same node's subject both answer.
+func serveCapability(t *testing.T, nc *nats.Conn, subject string, reply nats.MsgHandler) *nats.Subscription {
+	t.Helper()
+	sub, err := nc.Subscribe(subject, reply)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, nc.Flush())
+	return sub
+}
+
+// capabilityReply answers with a well-formed reply naming replyNode (which
+// may deliberately differ from the subject's node) and headerNode as the
+// X-Node-ID ("" omits the header).
+func capabilityReply(replyNode, headerNode string, version int) nats.MsgHandler {
+	return func(msg *nats.Msg) {
+		data, _ := json.Marshal(ec2v1.AccountSettingsSealCapabilityReply{Node: replyNode, CapabilityVersion: version})
+		reply := nats.NewMsg(msg.Reply)
+		reply.Data = data
+		if headerNode != "" {
+			reply.Header.Set(natsmsg.NodeIDHeader, headerNode)
+		}
+		_ = msg.RespondMsg(reply)
+	}
+}
+
+func nodeResult(t *testing.T, report sealCapabilityReport, node string) nodeSealCapability {
+	t.Helper()
+	for _, n := range report.Nodes {
+		if n.Node == node {
+			return n
+		}
+	}
+	t.Fatalf("no result for node %q in %+v", node, report)
+	return nodeSealCapability{}
+}
+
+// Each sub-case names exactly one way a node fails to prove capability
+// version 1, and that failure alone makes the whole probe incomplete.
+func TestSealCapabilityIncomplete(t *testing.T) {
+	t.Run("missing", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonNoReply, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("duplicate", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 1))
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 1))
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonDuplicate, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("unreachable timeout", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), func(msg *nats.Msg) {
+			time.Sleep(time.Second)
+			_ = msg.Respond(nil)
+		})
+		client := testCapabilityClient(nc, []string{"node-1"})
+		report := client.CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonNoReply, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("transport closed connection", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		nc.Close()
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonTransport, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("malformed body", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), func(msg *nats.Msg) {
+			reply := nats.NewMsg(msg.Reply)
+			reply.Data = []byte("not json")
+			reply.Header.Set(natsmsg.NodeIDHeader, "node-1")
+			_ = msg.RespondMsg(reply)
+		})
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonMalformed, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("error reply", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), func(msg *nats.Msg) {
+			reply := nats.NewMsg(msg.Reply)
+			reply.Data = awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal)
+			reply.Header.Set(natsmsg.NodeIDHeader, "node-1")
+			_ = msg.RespondMsg(reply)
+		})
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonErrorReply, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("wrong identity no header", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "", 1))
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonWrongIdentity, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("wrong identity another node header", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-9", 1))
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonWrongIdentity, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("wrong identity body node differs", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-9", "node-1", 1))
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonWrongIdentity, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("lower version", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 0))
+		report := testCapabilityClient(nc, []string{"node-1"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonOldVersion, nodeResult(t, report, "node-1").Reason)
+	})
+
+	t.Run("configured set empty", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		report := testCapabilityClient(nc, nil).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Empty(t, report.Nodes)
+	})
+
+	t.Run("invalid node id", func(t *testing.T) {
+		_, nc, _ := testutil.StartTestJetStream(t)
+		report := testCapabilityClient(nc, []string{"bad.node"}).CheckSealCapability(testCtx(t), "req")
+		assert.False(t, report.Complete)
+		assert.Equal(t, sealCapabilityReasonInvalidNodeID, nodeResult(t, report, "bad.node").Reason)
+	})
+}
+
+// Every configured node answering version 1 or later makes the probe
+// complete, and the reaper sends exactly one seal request.
+func TestSealCapabilityCompleteAllowsSeal(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 1))
+	serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-2"), capabilityReply("node-2", "node-2", 2))
+
+	client := testCapabilityClient(nc, []string{"node-1", "node-2"})
+	report := client.CheckSealCapability(testCtx(t), "req")
+	require.True(t, report.Complete)
+	assert.Equal(t, 1, nodeResult(t, report, "node-1").Version)
+	assert.Equal(t, 2, nodeResult(t, report, "node-2").Version)
+
+	sealer := &fakeSealer{}
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: client}
+	require.NoError(t, reaper.Delete(testCtx(t), testAccountID, Resource{Kind: reaper.Kind(), ID: testAccountID}, false))
+	assert.Equal(t, []string{testAccountID}, sealer.calls)
+}
+
+// An incomplete probe keeps the account TERMINATING through the full engine,
+// names the silent node and its reason, sends no seal request, and leaves the
+// settings record untouched (the fake sealer records no call).
+func TestSealCapabilityIncompleteKeepsAccountTerminating(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 1))
+	// node-2 has no subscriber.
+
+	sealer := &fakeSealer{}
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: testCapabilityClient(nc, []string{"node-1", "node-2"})}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, reaper)
+
+	result, err := engine.Teardown(testCtx(t), request())
+	require.ErrorIs(t, err, ErrResourcesStuck)
+	assert.False(t, result.AccountDeleted)
+	assert.Equal(t, AccountStatusTerminating, accounts.account.Status)
+	assert.Empty(t, sealer.calls)
+
+	var reason string
+	for _, stage := range result.Stages {
+		for _, stuck := range stage.Stuck {
+			if stuck.Resource.Kind == "ec2-account-settings" {
+				reason = stuck.Reason
+			}
+		}
+	}
+	assert.Contains(t, reason, "node-2="+sealCapabilityReasonNoReply)
+}
+
+// The silent node starting to answer during the drain makes the very next
+// Delete re-probe and succeed, which proves the result is retryable and the
+// probe is never cached across attempts.
+func TestSealCapabilityRetriesThenSeals(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), capabilityReply("node-1", "node-1", 1))
+
+	var node2Up atomic.Bool
+	sub, err := nc.Subscribe(ec2v1.AccountSettingsSealCapabilitySubject("node-2"), func(msg *nats.Msg) {
+		if !node2Up.Load() {
+			return
+		}
+		capabilityReply("node-2", "node-2", 1)(msg)
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, nc.Flush())
+
+	sealer := &fakeSealer{}
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: testCapabilityClient(nc, []string{"node-1", "node-2"})}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, reaper)
+	engine.Timeouts = Timeouts{StageDrain: time.Second, DrainPoll: 5 * time.Millisecond}
+
+	go func() {
+		time.Sleep(80 * time.Millisecond)
+		node2Up.Store(true)
+	}()
+
+	result, err := engine.Teardown(testCtx(t), request())
+	require.NoError(t, err)
+	assert.True(t, result.AccountDeleted)
+	assert.Equal(t, []string{testAccountID}, sealer.calls)
+}
+
+// Two Delete calls on one reaper probe twice: the probe runs in Delete
+// itself, never cached from a previous attempt.
+func TestSealCapabilityProbedBeforeEverySeal(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	var probes int
+	serveCapability(t, nc, ec2v1.AccountSettingsSealCapabilitySubject("node-1"), func(msg *nats.Msg) {
+		probes++
+		capabilityReply("node-1", "node-1", 1)(msg)
+	})
+
+	sealer := &fakeSealer{}
+	client := testCapabilityClient(nc, []string{"node-1"})
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: client}
+
+	require.NoError(t, reaper.Delete(testCtx(t), testAccountID, Resource{Kind: reaper.Kind(), ID: testAccountID}, false))
+	// alreadySealed now makes List empty, but Delete itself does not consult
+	// it, so a second direct call still probes and seals again.
+	require.NoError(t, reaper.Delete(testCtx(t), testAccountID, Resource{Kind: reaper.Kind(), ID: testAccountID}, false))
+	assert.Equal(t, 2, probes)
+	assert.Len(t, sealer.calls, 2)
+}
+
+// Five silent nodes still return within one deadline, not five: every node
+// is probed in parallel against the same wait.
+func TestSealCapabilityBoundedByOneDeadline(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	client := testCapabilityClient(nc, []string{"node-1", "node-2", "node-3", "node-4", "node-5"})
+
+	started := time.Now()
+	report := client.CheckSealCapability(testCtx(t), "req")
+	elapsed := time.Since(started)
+
+	assert.False(t, report.Complete)
+	assert.Less(t, elapsed, client.wait+time.Second, "five silent nodes must not cost five deadlines")
+}
+
+// The stuck reason is logged and stored as one line naming each failing node;
+// the retry hint and the ServiceUnavailable code travel with it.
+func TestSealCapabilityIncompleteErrorIsOneLine(t *testing.T) {
+	err := errSealCapabilityIncomplete(sealCapabilityReport{Nodes: []nodeSealCapability{{Node: "node-1", Reason: sealCapabilityReasonDuplicate}}})
+
+	assert.Equal(t, "EC2 account-settings seal capability incomplete: node-1=duplicate: ServiceUnavailable", err.Error())
+	d, ok := awserrors.ResolveRetryAfter(err)
+	assert.True(t, ok)
+	assert.Equal(t, 5*time.Second, d)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorServiceUnavailable))
+}

spinifex/accountteardown/ec2_account_settings_seal_test.go

@@ -0,0 +1,273 @@
+package accountteardown
+
+//test:in-package — the reaper, sealer and capability checker are unexported;
+// no seal without a complete probe, completion only after a sealed reply and
+// idempotent fresh runs are only observable from inside the package.
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	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"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// fakeSealer is a recording ec2AccountSettingsSealer. Each call beyond
+// failFor succeeds with a fresh, strictly increasing revision, unless err is
+// set, in which case every call after failFor fails with it.
+type fakeSealer struct {
+	mu       sync.Mutex
+	calls    []string
+	failFor  int
+	err      error
+	revision uint64
+}
+
+func (f *fakeSealer) SealAccountSettings(_ context.Context, accountID string) (uint64, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.calls = append(f.calls, accountID)
+	if len(f.calls) <= f.failFor {
+		return 0, errors.New("account settings owner temporarily unavailable")
+	}
+	if f.err != nil {
+		return 0, f.err
+	}
+	f.revision++
+	return f.revision, nil
+}
+
+// fakeSealCapability is a recording sealCapabilityChecker returning a fixed report.
+type fakeSealCapability struct {
+	calls  int
+	report sealCapabilityReport
+}
+
+func (f *fakeSealCapability) CheckSealCapability(context.Context, string) sealCapabilityReport {
+	f.calls++
+	return f.report
+}
+
+// completeReport builds a sealCapabilityReport where every named node answers
+// compatible at version 1.
+func completeReport(nodeIDs ...string) sealCapabilityReport {
+	var nodes []nodeSealCapability
+	for _, node := range nodeIDs {
+		nodes = append(nodes, nodeSealCapability{Node: node, Version: 1})
+	}
+	return sealCapabilityReport{Complete: true, Nodes: nodes}
+}
+
+// alwaysTerminatingChecker stands in for IAM in the end-to-end test: the
+// account under test is always TERMINATING.
+type alwaysTerminatingChecker struct{}
+
+func (alwaysTerminatingChecker) AccountTerminating(context.Context, string) (bool, error) {
+	return true, nil
+}
+
+// No subscriber on either subject makes the capability probe incomplete, so
+// no seal request is ever sent and the account stays TERMINATING with one
+// stuck ec2-account-settings item.
+func TestSealReaperNoResponderKeepsAccountTerminating(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+
+	capability := newSealCapabilityClient(nc, []string{"node-1"})
+	capability.wait = 50 * time.Millisecond
+	capability.grace = 10 * time.Millisecond
+	sealer := newEC2AccountSettingsSealClient(nc)
+	sealer.timeout = 50 * time.Millisecond
+
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: capability}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, reaper)
+
+	result, err := engine.Teardown(testCtx(t), request())
+	require.ErrorIs(t, err, ErrResourcesStuck)
+	assert.False(t, result.AccountDeleted)
+	assert.Equal(t, AccountStatusTerminating, accounts.account.Status)
+
+	var stuck []Stuck
+	for _, stage := range result.Stages {
+		stuck = append(stuck, stage.Stuck...)
+	}
+	require.Len(t, stuck, 1)
+	assert.Equal(t, "ec2-account-settings", stuck[0].Resource.Kind)
+}
+
+// Whatever the owner refuses with, the account stays TERMINATING: an owner
+// refusal, a timeout, and each reply shape the client rejects (wrong account,
+// wrong disposition, no revision).
+func TestSealReaperOwnerFailureKeepsAccountTerminating(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+	}{
+		{"incorrect-state", errors.New(awserrors.ErrorIncorrectState)},
+		{"internal-error", errors.New(awserrors.ErrorInternalError)},
+		{"timeout", context.DeadlineExceeded},
+		{"wrong-account", errors.New(`seal account settings: reply named account "999999999999", not "000000000042"`)},
+		{"wrong-disposition", errors.New(`seal account settings: unexpected disposition "pending"`)},
+		{"zero-revision", errors.New("seal account settings: reply carried no revision")},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			sealer := &fakeSealer{err: tc.err}
+			reaper := &accountSettingsSealReaper{sealer: sealer, capability: &fakeSealCapability{report: completeReport("node-1")}}
+			accounts := newFakeAccounts()
+			engine := quietEngine(accounts, reaper)
+
+			result, err := engine.Teardown(testCtx(t), request())
+			require.ErrorIs(t, err, ErrResourcesStuck)
+			assert.False(t, result.AccountDeleted)
+			assert.Equal(t, AccountStatusTerminating, accounts.account.Status)
+			assert.NotEmpty(t, sealer.calls)
+		})
+	}
+}
+
+// The reaper retries through the engine's own drain mechanics: it fails
+// twice, then succeeds, and only the success lets the account be deleted.
+// List is non-empty before the success and empty only after it.
+func TestSealReaperCompletesOnlyAfterSealed(t *testing.T) {
+	sealer := &fakeSealer{failFor: 2}
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: &fakeSealCapability{report: completeReport("node-1")}}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, reaper)
+
+	before, err := reaper.List(testCtx(t), testAccountID)
+	require.NoError(t, err)
+	require.Len(t, before, 1)
+
+	result, err := engine.Teardown(testCtx(t), request())
+	require.NoError(t, err)
+	assert.True(t, result.AccountDeleted)
+	assert.GreaterOrEqual(t, len(sealer.calls), 3)
+
+	after, err := reaper.List(testCtx(t), testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, after)
+}
+
+// Each teardown builds a fresh engine and a fresh reaper, so a retried or
+// taken-over run seals again: two engines against the same owner both
+// succeed, each with a higher revision than the last.
+func TestSealReaperFreshRunSealsAgain(t *testing.T) {
+	sealer := &fakeSealer{}
+	capability := &fakeSealCapability{report: completeReport("node-1")}
+
+	firstEngine := quietEngine(newFakeAccounts(), &accountSettingsSealReaper{sealer: sealer, capability: capability})
+	firstResult, err := firstEngine.Teardown(testCtx(t), request())
+	require.NoError(t, err)
+	assert.True(t, firstResult.AccountDeleted)
+
+	secondEngine := quietEngine(newFakeAccounts(), &accountSettingsSealReaper{sealer: sealer, capability: capability})
+	secondResult, err := secondEngine.Teardown(testCtx(t), request())
+	require.NoError(t, err)
+	assert.True(t, secondResult.AccountDeleted)
+
+	require.Len(t, sealer.calls, 2)
+	assert.Equal(t, uint64(2), sealer.revision)
+}
+
+// Inventory and the explicit dry run, and a Teardown request marked dry run,
+// all list the pending seal and send no seal request: only a real teardown
+// may seal.
+func TestSealReaperInventoryDoesNotSeal(t *testing.T) {
+	sealer := &fakeSealer{}
+	capability := &fakeSealCapability{report: completeReport("node-1")}
+	reaper := &accountSettingsSealReaper{sealer: sealer, capability: capability}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, reaper)
+
+	inventory, err := engine.Inventory(testCtx(t), testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, sealer.calls)
+	assert.Zero(t, capability.calls)
+	found := false
+	for _, stage := range inventory.Stages {
+		for _, resource := range stage.Deleted {
+			if resource.Kind == "ec2-account-settings" {
+				found = true
+			}
+		}
+	}
+	assert.True(t, found, "inventory must still list the pending seal")
+
+	dryRun := request()
+	dryRun.DryRun = true
+	_, err = engine.Teardown(testCtx(t), dryRun)
+	require.NoError(t, err)
+	assert.Empty(t, sealer.calls)
+	assert.Zero(t, capability.calls)
+}
+
+// A real settings service subscribed on the real seal subject, behind a real
+// engine: teardown seals the record, replacing the characterisation test that
+// showed it used to survive.
+func TestTeardownSealsAccountSettings(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+	ctx := testCtx(t)
+
+	settings, err := ec2account.NewAccountSettingsServiceImplWithNATS(ctx, nil, nc)
+	require.NoError(t, err)
+	settings.SetAccountTerminationChecker(alwaysTerminatingChecker{})
+	_, err = settings.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID)
+	require.NoError(t, err)
+
+	sealSub, err := nc.QueueSubscribe(ec2v1.SealAccountSettingsSubject, "spinifex-workers", func(msg *nats.Msg) {
+		natsmsg.ServeNATSRequestCtx(msg, func(ctx context.Context, req *ec2v1.SealAccountSettingsRequest) (*ec2v1.SealAccountSettingsReply, error) {
+			seal, err := settings.SealAccount(ctx, req.AccountID)
+			if err != nil {
+				return nil, err
+			}
+			return &ec2v1.SealAccountSettingsReply{
+				AccountID: req.AccountID, Disposition: ec2v1.AccountSettingsDispositionSealed, Revision: seal.Revision,
+			}, nil
+		})
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sealSub.Unsubscribe() })
+
+	capSub, err := nc.Subscribe(ec2v1.AccountSettingsSealCapabilitySubject("node-1"), func(msg *nats.Msg) {
+		reply, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityReply{Node: "node-1", CapabilityVersion: 1})
+		require.NoError(t, err)
+		out := nats.NewMsg("")
+		out.Data = reply
+		out.Header.Set(natsmsg.NodeIDHeader, "node-1")
+		require.NoError(t, msg.RespondMsg(out))
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = capSub.Unsubscribe() })
+	require.NoError(t, nc.Flush())
+
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, EC2AccountSettingsReapers(nc, []string{"node-1"})...)
+	// The real capability probe holds CollectUntilDeadline's full ResponderGrace
+	// (150 ms) even when the one expected node answers instantly, so the quiet
+	// engine's default 200 ms StageDrain leaves no room for the sweep after it.
+	engine.Timeouts = Timeouts{StageDrain: 2 * time.Second, DrainPoll: 10 * time.Millisecond}
+
+	result, err := engine.Teardown(ctx, request())
+	require.NoError(t, err)
+	assert.True(t, result.AccountDeleted)
+
+	bucket, err := js.KeyValue(ctx, ec2account.KVBucketAccountSettings)
+	require.NoError(t, err)
+	entry, err := bucket.Get(ctx, testAccountID)
+	require.NoError(t, err)
+	var rec ec2account.AccountSettingsRecord
+	require.NoError(t, json.Unmarshal(entry.Value(), &rec))
+	assert.Equal(t, ec2account.AccountSettingsStateSealed, rec.State)
+}

spinifex/accountteardown/reapers_ec2_account_settings_current_behaviour_test.go

@@ -1,50 +0,0 @@
-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/settings_kv_boundary_test.go

@@ -0,0 +1,89 @@
+package accountteardown
+
+//test:in-package — pins ADR-0003:S4 and ADR-0001 Context item 2 for the EC2
+// settings seal: the legacy engine must reach the owner only through the
+// versioned contract and NATS, never by reading the settings bucket itself.
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"io/fs"
+	"path/filepath"
+	"runtime"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+// forbiddenSettingsImports are imports that would let this package reach the
+// EC2 settings owner or the configured node set by a path other than the
+// versioned contract and NewClusterEngine's parameters.
+var forbiddenSettingsImports = []string{
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/account",
+	"github.com/mulgadc/spinifex/spinifex/bootstrap/config",
+}
+
+// forbiddenSettingsIdentifiers would let this package name the settings
+// bucket or its key directly instead of reaching it only through
+// contracts/ec2/v1 and NATS.
+var forbiddenSettingsIdentifiers = []string{"spinifex-ec2-account-settings", "KVBucketAccountSettings"}
+
+// Teardown reaches EC2 settings only over contracts/ec2/v1: no import of the
+// settings owner or bootstrap/config (nodes arrive via NewClusterEngine), and
+// no literal naming the settings bucket or its constant.
+func TestAccountTeardownHasNoSettingsKVAccess(t *testing.T) {
+	_, file, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("cannot locate test source")
+	}
+	root := filepath.Dir(file)
+
+	fset := token.NewFileSet()
+	scanned := 0
+	err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
+		if err != nil || d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
+			return err
+		}
+		f, perr := parser.ParseFile(fset, path, nil, parser.ParseComments)
+		if perr != nil {
+			return perr
+		}
+		scanned++
+
+		for _, imp := range f.Imports {
+			p, _ := strconv.Unquote(imp.Path.Value)
+			for _, bad := range forbiddenSettingsImports {
+				if p == bad {
+					t.Errorf("%s: imports %s; the legacy engine must reach the EC2 settings owner only through contracts/ec2/v1 and the configured node set NewClusterEngine already receives",
+						fset.Position(imp.Pos()), p)
+				}
+			}
+		}
+
+		ast.Inspect(f, func(n ast.Node) bool {
+			lit, ok := n.(*ast.BasicLit)
+			if !ok || lit.Kind != token.STRING {
+				return true
+			}
+			value, uerr := strconv.Unquote(lit.Value)
+			if uerr != nil {
+				return true
+			}
+			for _, bad := range forbiddenSettingsIdentifiers {
+				if value == bad {
+					t.Errorf("%s: names %q; the settings bucket and its key are the owner's own, never this package's",
+						fset.Position(lit.Pos()), value)
+				}
+			}
+			return true
+		})
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if scanned == 0 {
+		t.Fatalf("scanned no Go files under %s; the invariant would pass vacuously", root)
+	}
+}

spinifex/daemon/daemon.go

@@ -1289,6 +1289,15 @@ func (d *Daemon) subscribeAll() error {
 		natsSub{viperblocklegacyv1.VolumeFencedSubject(d.node), d.handleVolumeFenced, ""},
 	)
 
+	// Per-node, no queue group: a second daemon claiming this node ID also
+	// answers, so the teardown probe sees a duplicate and refuses to seal. An
+	// invalid node ID serves nothing, so this daemon reads as incompatible.
+	if ec2v1.ValidSealCapabilityNode(d.node) {
+		subs = append(subs, natsSub{ec2v1.AccountSettingsSealCapabilitySubject(d.node), handleNATSRequest(d.node, d.accountSettingsSealCapability), ""})
+	} else {
+		slog.Error("Not serving account-settings seal capability: node ID is not a single NATS subject token", "node", d.node)
+	}
+
 	return d.registerNatsSubs(subs)
 }
 

spinifex/daemon/daemon_handlers_account_settings_seal_capability.go

@@ -0,0 +1,23 @@
+package daemon
+
+import (
+	"context"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+)
+
+// accountSettingsSealCapability serves this daemon's read-only seal capability.
+// It reads no KV and takes no account; a request naming another node is a
+// caller mistake, since the subject is already addressed to this node.
+func (d *Daemon) accountSettingsSealCapability(_ context.Context, req *ec2v1.AccountSettingsSealCapabilityRequest, _ string) (*ec2v1.AccountSettingsSealCapabilityReply, error) {
+	if req.Node != d.node {
+		return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue,
+			"seal capability request for node %q answered by node %q", req.Node, d.node)
+	}
+	return &ec2v1.AccountSettingsSealCapabilityReply{
+		Node:              d.node,
+		CapabilityVersion: ec2v1.AccountSettingsSealCapabilityVersion,
+		CorrelationRef:    req.CorrelationRef,
+	}, nil
+}

spinifex/daemon/daemon_handlers_account_settings_seal_capability_test.go

@@ -0,0 +1,115 @@
+package daemon
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	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"
+)
+
+// newCapabilityTestDaemon builds a daemon serving its real subscription table
+// with node as its configured node ID.
+func newCapabilityTestDaemon(t *testing.T, node string) (*nats.Conn, *nats.Conn) {
+	t.Helper()
+	ns, nc, _ := testutil.StartTestJetStream(t)
+
+	d := newSubscribeTestDaemon(t, nc, "", "")
+	d.node = node
+	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 nc, caller
+}
+
+func capabilityRequest(t *testing.T, nc *nats.Conn, node string) *nats.Msg {
+	t.Helper()
+	data, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityRequest{Node: node, CorrelationRef: "req-1"})
+	require.NoError(t, err)
+	reply, err := nc.Request(ec2v1.AccountSettingsSealCapabilitySubject(node), data, 5*time.Second)
+	require.NoError(t, err)
+	return reply
+}
+
+// The responder answers its own node with version 1, the configured node and
+// no KV access (no settings service is wired into the test daemon at all).
+func TestAccountSettingsSealCapabilityAnswersOwnNode(t *testing.T) {
+	_, caller := newCapabilityTestDaemon(t, "node-1")
+
+	raw := capabilityRequest(t, caller, "node-1")
+	assert.Equal(t, "node-1", raw.Header.Get("X-Node-ID"))
+
+	var reply ec2v1.AccountSettingsSealCapabilityReply
+	require.NoError(t, json.Unmarshal(raw.Data, &reply))
+	assert.Equal(t, "node-1", reply.Node)
+	assert.Equal(t, ec2v1.AccountSettingsSealCapabilityVersion, reply.CapabilityVersion)
+	assert.Equal(t, "req-1", reply.CorrelationRef)
+}
+
+// A request naming a node other than this daemon's own is refused, not
+// answered as if it were this node.
+func TestAccountSettingsSealCapabilityRefusesOtherNode(t *testing.T) {
+	_, caller := newCapabilityTestDaemon(t, "node-1")
+
+	data, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityRequest{Node: "node-2"})
+	require.NoError(t, err)
+	raw, err := caller.Request(ec2v1.AccountSettingsSealCapabilitySubject("node-1"), data, 5*time.Second)
+	require.NoError(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidParameterValue, replyErrCode(t, raw.Data))
+}
+
+// The subject is per node with no queue group, so two daemons configured
+// with the same node ID both answer a request addressed to that node.
+func TestAccountSettingsSealCapabilityHasNoQueueGroup(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+
+	first := newSubscribeTestDaemon(t, nc, "", "")
+	first.node = "node-1"
+	require.NoError(t, first.subscribeAll())
+
+	second := newSubscribeTestDaemon(t, nc, "", "")
+	second.node = "node-1"
+	require.NoError(t, second.subscribeAll())
+	require.NoError(t, nc.Flush())
+
+	sub, err := nc.SubscribeSync(nats.NewInbox())
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	data, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityRequest{Node: "node-1"})
+	require.NoError(t, err)
+	msg := nats.NewMsg(ec2v1.AccountSettingsSealCapabilitySubject("node-1"))
+	msg.Data = data
+	msg.Reply = sub.Subject
+	require.NoError(t, nc.PublishMsg(msg))
+	require.NoError(t, nc.Flush())
+
+	seen := 0
+	for {
+		_, err := sub.NextMsg(300 * time.Millisecond)
+		if err != nil {
+			break
+		}
+		seen++
+	}
+	assert.Equal(t, 2, seen)
+}
+
+// A wildcard node ID must not become a wildcard subscription: built unchecked,
+// "node-*" would answer probes addressed to node-1 on node-1's behalf.
+func TestAccountSettingsSealCapabilityNotServedForInvalidNodeID(t *testing.T) {
+	_, caller := newCapabilityTestDaemon(t, "node-*")
+
+	data, err := json.Marshal(ec2v1.AccountSettingsSealCapabilityRequest{Node: "node-1"})
+	require.NoError(t, err)
+	_, err = caller.Request(ec2v1.AccountSettingsSealCapabilitySubject("node-1"), data, time.Second)
+	require.ErrorIs(t, err, nats.ErrNoResponders)
+}

spinifex/domains/admission/quota/cleanup.go

@@ -1,93 +0,0 @@
-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

@@ -1,184 +0,0 @@
-//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/admission/quota/live.go

@@ -39,7 +39,11 @@ func (s *Service) EnforceVPCs(ctx context.Context, natsConn *nats.Conn, accountI
 	if err != nil {
 		return err
 	}
-	return exceeds(len(out.Vpcs), want, s.limitsFor(ctx, accountID).VPCs)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(len(out.Vpcs), want, limits.VPCs)
 }
 
 // EnforceSubnets gates CreateSubnet on the account's live DescribeSubnets count.
@@ -52,7 +56,11 @@ func (s *Service) EnforceSubnets(ctx context.Context, natsConn *nats.Conn, accou
 	if err != nil {
 		return err
 	}
-	return exceeds(len(out.Subnets), want, s.limitsFor(ctx, accountID).Subnets)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(len(out.Subnets), want, limits.Subnets)
 }
 
 // EnforceEIPs gates AllocateAddress on the account's live DescribeAddresses count.
@@ -64,7 +72,11 @@ func (s *Service) EnforceEIPs(ctx context.Context, natsConn *nats.Conn, accountI
 	if err != nil {
 		return err
 	}
-	return exceeds(len(out.Addresses), want, s.limitsFor(ctx, accountID).EIPs)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(len(out.Addresses), want, limits.EIPs)
 }
 
 // EnforceLoadBalancers gates CreateLoadBalancer on the account's live count.
@@ -77,7 +89,11 @@ func (s *Service) EnforceLoadBalancers(ctx context.Context, natsConn *nats.Conn,
 	if err != nil {
 		return err
 	}
-	return exceeds(len(out.LoadBalancers), want, s.limitsFor(ctx, accountID).LoadBalancers)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(len(out.LoadBalancers), want, limits.LoadBalancers)
 }
 
 // EnforceRDSInstances gates CreateDBInstance on the account's live count. A DB
@@ -91,5 +107,9 @@ func (s *Service) EnforceRDSInstances(ctx context.Context, natsConn *nats.Conn,
 	if err != nil {
 		return err
 	}
-	return exceeds(len(out.DBInstances), want, s.limitsFor(ctx, accountID).RDSInstances)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(len(out.DBInstances), want, limits.RDSInstances)
 }

spinifex/domains/admission/quota/override.go

@@ -8,12 +8,16 @@ import (
 	"log/slog"
 	"time"
 
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
 // Overrides raises or lowers individual dimensions for one account. Every field
 // is a pointer so an explicit 0 (a limit that denies everything) is distinct
-// from an absent field, which inherits the configured value.
+// from an absent field, which inherits the configured value. This is also the
+// admin wire type: an admin request body can carry only these fields, so it can
+// never set or clear a stored record's state.
 type Overrides struct {
 	VCPUs         *int `json:"vcpus,omitempty"`
 	VPCs          *int `json:"vpcs,omitempty"`
@@ -29,11 +33,62 @@ type Overrides struct {
 	UpdatedBy string `json:"updated_by,omitempty"`
 }
 
-// Empty reports whether no dimension is overridden, in which case the record
-// carries nothing but provenance and the account resolves to the config.
-func (o Overrides) Empty() bool {
-	return o.VCPUs == nil && o.VPCs == nil && o.Subnets == nil && o.EIPs == nil &&
-		o.VolumesGiB == nil && o.Volumes == nil && o.RDSInstances == nil && o.LoadBalancers == nil
+// OverrideState is a stored override record's lifecycle state. "" is active;
+// a record written before the state field existed has none and stays active.
+type OverrideState string
+
+// OverrideStateSealed marks a record no writer may change or recreate.
+const OverrideStateSealed OverrideState = "sealed"
+
+// overrideRecord is the stored form: the admin-settable fields inline, plus a
+// state no admin request can read or write.
+type overrideRecord struct {
+	Overrides
+
+	State OverrideState `json:"state,omitempty"`
+}
+
+// errOverridesSealed marks a read or write that found a sealed record, so a
+// caller maps it to Blocked without logging it as a fault.
+var errOverridesSealed = errors.New(awserrors.ErrorBlocked)
+
+// errOverridesContended marks a write whose CAS retries were exhausted under
+// concurrent writers, bounding the retry loop instead of spinning.
+var errOverridesContended = errors.New("quota override: CAS retries exhausted")
+
+// overrideMaxAttempts bounds a write's retries against a revision that keeps
+// moving under concurrent writers, so it fails closed instead of spinning.
+const overrideMaxAttempts = 5
+
+// checkActive rejects a sealed or unknown-state record; "" is active and
+// passes through. A caller must never read or mutate a record that failed
+// this as if it were active.
+func checkActive(rec overrideRecord) error {
+	switch rec.State {
+	case "":
+		return nil
+	case OverrideStateSealed:
+		return errOverridesSealed
+	default:
+		return fmt.Errorf("quota override: unknown state %q", rec.State)
+	}
+}
+
+// mapOverrideError turns a recordFor, PutAccountQuota or clear failure into
+// the AWS error a caller returns: sealed is an expected, unlogged Blocked;
+// everything else is InternalError, logged with the account ID.
+func mapOverrideError(ctx context.Context, accountID string, err error) error {
+	switch {
+	case err == nil:
+		return nil
+	case errors.Is(err, errOverridesSealed):
+		return errors.New(awserrors.ErrorBlocked)
+	case errors.Is(err, errOverridesContended):
+		slog.ErrorContext(ctx, "Quota override CAS retries exhausted under contention", "accountID", accountID)
+	default:
+		slog.ErrorContext(ctx, "Quota override operation failed", "accountID", accountID, "err", err)
+	}
+	return errors.New(awserrors.ErrorInternalError)
 }
 
 // apply overlays the set fields onto base, leaving Enabled and the Bedrock
@@ -60,72 +115,128 @@ func (o Overrides) apply(base Limits) Limits {
 	return out
 }
 
-// limitsFor resolves the effective limits for accountID: its overrides where
-// set, the configured limits everywhere else. A read failure falls back to the
-// configured limits rather than to no limit, so a KV outage cannot lift a cap.
-func (s *Service) limitsFor(ctx context.Context, accountID string) Limits {
-	over, err := s.overridesFor(ctx, accountID)
-	if err != nil {
-		slog.WarnContext(ctx, "quota: override read failed, using configured limits",
-			"account", accountID, "err", err)
-		return s.limits
-	}
-	return over.apply(s.limits)
+// Empty reports whether no dimension is overridden, in which case the record
+// carries nothing but provenance and the account resolves to the config.
+func (o Overrides) Empty() bool {
+	return o.VCPUs == nil && o.VPCs == nil && o.Subnets == nil && o.EIPs == nil &&
+		o.VolumesGiB == nil && o.Volumes == nil && o.RDSInstances == nil && o.LoadBalancers == nil
 }
 
-// overridesFor reads accountID's override record. An absent key is the normal
-// case for nearly every account and yields the zero value, not an error; only a
-// genuine read or decode failure is returned.
-func (s *Service) overridesFor(ctx context.Context, accountID string) (Overrides, error) {
+// recordFor reads and decodes accountID's stored override record. An absent
+// key is the active zero value, not an error; only a genuine read or decode
+// failure is returned.
+func (s *Service) recordFor(ctx context.Context, accountID string) (overrideRecord, error) {
 	if s == nil || s.overrides == nil {
-		return Overrides{}, nil
+		return overrideRecord{}, nil
 	}
 	entry, err := s.overrides.Get(ctx, accountID)
 	if errors.Is(err, jetstream.ErrKeyNotFound) {
-		return Overrides{}, nil
+		return overrideRecord{}, nil
 	}
 	if err != nil {
-		return Overrides{}, fmt.Errorf("read quota override for %s: %w", accountID, err)
+		return overrideRecord{}, fmt.Errorf("read quota override for %s: %w", accountID, err)
 	}
-	var over Overrides
-	if err := json.Unmarshal(entry.Value(), &over); err != nil {
-		return Overrides{}, fmt.Errorf("decode quota override for %s: %w", accountID, err)
+	var rec overrideRecord
+	if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+		return overrideRecord{}, fmt.Errorf("decode quota override for %s: %w", accountID, err)
 	}
-	return over, nil
+	return rec, nil
 }
 
-// GetAccountQuota returns accountID's stored overrides and the limits they
-// resolve to. A read failure is surfaced here rather than masked, because an
-// operator inspecting a quota must not be shown the fallback as if it were real.
+// limitsFor resolves accountID's effective limits: overrides where set, the
+// config elsewhere. A read or decode failure falls back to the config so an
+// outage cannot lift a cap; a sealed or unknown-state record refuses instead.
+func (s *Service) limitsFor(ctx context.Context, accountID string) (Limits, error) {
+	rec, err := s.recordFor(ctx, accountID)
+	if err != nil {
+		slog.WarnContext(ctx, "quota: override read failed, using configured limits",
+			"account", accountID, "err", err)
+		return s.limits, nil
+	}
+	if err := checkActive(rec); err != nil {
+		return Limits{}, mapOverrideError(ctx, accountID, err)
+	}
+	return rec.apply(s.limits), nil
+}
+
+// GetAccountQuota returns accountID's stored overrides and resolved limits.
+// Read failures surface rather than show the fallback as real; a sealed or
+// unknown-state record returns Blocked or InternalError.
 func (s *Service) GetAccountQuota(ctx context.Context, accountID string) (Overrides, Limits, error) {
-	over, err := s.overridesFor(ctx, accountID)
+	rec, err := s.recordFor(ctx, accountID)
 	if err != nil {
 		return Overrides{}, Limits{}, err
 	}
-	return over, over.apply(s.limits), nil
+	if err := checkActive(rec); err != nil {
+		return Overrides{}, Limits{}, mapOverrideError(ctx, accountID, err)
+	}
+	return rec.Overrides, rec.apply(s.limits), nil
 }
 
-// PutAccountQuota stores accountID's overrides, stamping provenance. An empty
-// set deletes the record so the account resolves to the config, which keeps
-// "has this account been touched?" a single existence check.
+// PutAccountQuota stores or clears accountID's overrides, stamping provenance
+// on a store. Both branches are revision-guarded and refuse a sealed record,
+// so a seal landing between a read and a write fails the CAS.
 func (s *Service) PutAccountQuota(ctx context.Context, accountID string, over Overrides, updatedBy string) error {
 	if s == nil || s.overrides == nil {
 		return errors.New("quota: override bucket not configured")
 	}
 	if over.Empty() {
-		if err := s.overrides.Delete(ctx, accountID); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
-			return fmt.Errorf("clear quota override for %s: %w", accountID, err)
-		}
-		return nil
+		return s.clearAccountQuota(ctx, accountID)
 	}
+
 	over.UpdatedAt = time.Now().UTC().Format(time.RFC3339)
 	over.UpdatedBy = updatedBy
-	data, err := json.Marshal(over)
-	if err != nil {
-		return fmt.Errorf("encode quota override for %s: %w", accountID, err)
-	}
-	if _, err := s.overrides.Put(ctx, accountID, data); err != nil {
-		return fmt.Errorf("store quota override for %s: %w", accountID, err)
-	}
-	return nil
+
+	_, err := kvutil.Update(ctx, s.overrides, accountID, kvutil.CASConfig{
+		CreateIfAbsent: true,
+		Attempts:       overrideMaxAttempts,
+		Exhausted:      func(string, int) error { return errOverridesContended },
+	}, func(rec *overrideRecord) (bool, error) {
+		if err := checkActive(*rec); err != nil {
+			return false, err
+		}
+		rec.Overrides = over
+		// Every success is a leader-accepted write over an active value: a
+		// seal landing between the read and the write fails this CAS.
+		return true, nil
+	})
+	return mapOverrideError(ctx, accountID, err)
+}
+
+// clearAccountQuota deletes an active override at the revision it read. An
+// observed-absent key writes nothing, so a stale read cannot unseal; a sealed
+// record refuses rather than being erased and later recreated active.
+func (s *Service) clearAccountQuota(ctx context.Context, accountID string) error {
+	for range overrideMaxAttempts {
+		entry, err := s.overrides.Get(ctx, accountID)
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return nil
+		}
+		if err != nil {
+			return mapOverrideError(ctx, accountID, fmt.Errorf("read quota override for clear: %w", err))
+		}
+
+		var rec overrideRecord
+		if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+			return mapOverrideError(ctx, accountID, fmt.Errorf("decode quota override for clear: %w", err))
+		}
+		if err := checkActive(rec); err != nil {
+			return mapOverrideError(ctx, accountID, err)
+		}
+
+		err = s.overrides.Delete(ctx, accountID, jetstream.LastRevision(entry.Revision()))
+		switch {
+		case err == nil:
+			return nil
+		case errors.Is(err, jetstream.ErrKeyNotFound):
+			return nil
+		case isRevisionConflict(err):
+			// A concurrent write landed between the read and the guarded
+			// delete; retry against whatever revision now holds.
+			continue
+		default:
+			return mapOverrideError(ctx, accountID, fmt.Errorf("clear quota override: %w", err))
+		}
+	}
+	return mapOverrideError(ctx, accountID, errOverridesContended)
 }

spinifex/domains/admission/quota/override_kv_test.go

@@ -31,7 +31,8 @@ func newOverrideService(t *testing.T) (*Service, jetstream.KeyValue) {
 func TestLimitsForAbsentKeyInherits(t *testing.T) {
 	svc, _ := newOverrideService(t)
 
-	got := svc.limitsFor(t.Context(), "123456789012")
+	got, err := svc.limitsFor(t.Context(), "123456789012")
+	require.NoError(t, err)
 	require.Equal(t, baseLimits(), got)
 }
 
@@ -44,10 +45,14 @@ func TestLimitsForIsPerAccount(t *testing.T) {
 	require.NoError(t, svc.PutAccountQuota(t.Context(), upgraded,
 		Overrides{VCPUs: ptr(32), RDSInstances: ptr(8)}, "operator"))
 
-	require.Equal(t, 32, svc.limitsFor(t.Context(), upgraded).VCPUs)
-	require.Equal(t, 8, svc.limitsFor(t.Context(), upgraded).RDSInstances)
-	require.Equal(t, baseLimits().VCPUs, svc.limitsFor(t.Context(), sandbox).VCPUs)
-	require.Equal(t, baseLimits().RDSInstances, svc.limitsFor(t.Context(), sandbox).RDSInstances)
+	upgradedLimits, err := svc.limitsFor(t.Context(), upgraded)
+	require.NoError(t, err)
+	sandboxLimits, err := svc.limitsFor(t.Context(), sandbox)
+	require.NoError(t, err)
+	require.Equal(t, 32, upgradedLimits.VCPUs)
+	require.Equal(t, 8, upgradedLimits.RDSInstances)
+	require.Equal(t, baseLimits().VCPUs, sandboxLimits.VCPUs)
+	require.Equal(t, baseLimits().RDSInstances, sandboxLimits.RDSInstances)
 }
 
 // A stored override must round-trip through KV with its provenance, and an
@@ -79,7 +84,9 @@ func TestPutAccountQuotaEmptyClears(t *testing.T) {
 
 	_, err := kv.Get(t.Context(), account)
 	require.ErrorIs(t, err, jetstream.ErrKeyNotFound)
-	require.Equal(t, baseLimits(), svc.limitsFor(t.Context(), account))
+	limits, err := svc.limitsFor(t.Context(), account)
+	require.NoError(t, err)
+	require.Equal(t, baseLimits(), limits)
 }
 
 // Clearing an account that never had a record must succeed rather than fail on
@@ -97,8 +104,10 @@ func TestPutAccountQuotaExplicitZero(t *testing.T) {
 
 	require.NoError(t, svc.PutAccountQuota(t.Context(), account, Overrides{VPCs: ptr(0)}, "operator"))
 
-	require.Equal(t, 0, svc.limitsFor(t.Context(), account).VPCs)
-	require.Error(t, exceeds(0, 1, svc.limitsFor(t.Context(), account).VPCs))
+	limits, err := svc.limitsFor(t.Context(), account)
+	require.NoError(t, err)
+	require.Equal(t, 0, limits.VPCs)
+	require.Error(t, exceeds(0, 1, limits.VPCs))
 }
 
 // A corrupt record is a read failure, not an absent one: it must fall back to
@@ -110,7 +119,9 @@ func TestLimitsForCorruptRecordFailsClosed(t *testing.T) {
 	_, err := kv.Put(t.Context(), account, []byte("{not json"))
 	require.NoError(t, err)
 
-	require.Equal(t, baseLimits(), svc.limitsFor(t.Context(), account),
+	limits, err := svc.limitsFor(t.Context(), account)
+	require.NoError(t, err, "a decode failure must fall back, not error, from limitsFor")
+	require.Equal(t, baseLimits(), limits,
 		"a corrupt override must fall back to the configured limits")
 
 	_, _, err = svc.GetAccountQuota(t.Context(), account)

spinifex/domains/admission/quota/override_test.go

@@ -126,7 +126,11 @@ func TestOverridesEmpty(t *testing.T) {
 // fall through to the configured limits rather than panicking.
 func TestLimitsForWithoutBucket(t *testing.T) {
 	svc := New(baseLimits(), nil)
-	if got := svc.limitsFor(t.Context(), "123456789012"); got != baseLimits() {
+	got, err := svc.limitsFor(t.Context(), "123456789012")
+	if err != nil {
+		t.Fatalf("limitsFor error = %v, want nil", err)
+	}
+	if got != baseLimits() {
 		t.Fatalf("limitsFor = %+v, want the configured limits", got)
 	}
 }

spinifex/domains/admission/quota/override_writes_guarded_test.go

@@ -0,0 +1,75 @@
+package quota
+
+//test:in-package — pins what AccountQuotaOverrideSealCapabilityVersion 1 will
+// assert: nothing in this package writes the override bucket except through
+// PutAccountQuota's CAS loop, clearAccountQuota's guarded delete or
+// SealAccountOverrides' CAS loop.
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"path/filepath"
+	"runtime"
+	"testing"
+)
+
+// overrideWriteFiles are the only files in this package that touch the
+// override bucket. The package also holds the vCPU usage and Bedrock usage
+// buckets, which write through their own stores and are out of scope here;
+// scanning the whole package directory, as the EC2 settings equivalent does,
+// would false-flag those unrelated writes.
+var overrideWriteFiles = []string{"override.go", "seal.go"}
+
+func quotaPackageDir(t *testing.T) string {
+	t.Helper()
+	_, file, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("cannot locate test source")
+	}
+	return filepath.Dir(file)
+}
+
+// No file owning the override bucket may call Put, PutString, or Delete
+// without a revision option: every write must go through PutAccountQuota's
+// CAS loop, clearAccountQuota's guarded delete, or SealAccountOverrides' CAS
+// loop.
+func TestOverrideWritesAreRevisionGuarded(t *testing.T) {
+	root := quotaPackageDir(t)
+	fset := token.NewFileSet()
+	scanned := 0
+
+	for _, name := range overrideWriteFiles {
+		path := filepath.Join(root, name)
+		f, err := parser.ParseFile(fset, path, nil, 0)
+		if err != nil {
+			t.Fatalf("parse %s: %v", path, err)
+		}
+		scanned++
+
+		ast.Inspect(f, func(n ast.Node) bool {
+			call, ok := n.(*ast.CallExpr)
+			if !ok {
+				return true
+			}
+			sel, ok := call.Fun.(*ast.SelectorExpr)
+			if !ok {
+				return true
+			}
+			switch sel.Sel.Name {
+			case "Put", "PutString":
+				t.Errorf("%s: unconditional %s call bypasses the revision-guarded override write; use PutAccountQuota or SealAccountOverrides instead",
+					fset.Position(call.Pos()), sel.Sel.Name)
+			case "Delete":
+				if len(call.Args) < 3 {
+					t.Errorf("%s: Delete call with no revision option bypasses the guarded clear; pin it to the read revision",
+						fset.Position(call.Pos()))
+				}
+			}
+			return true
+		})
+	}
+	if scanned == 0 {
+		t.Fatalf("scanned no override-write files under %s; the invariant would pass vacuously", root)
+	}
+}

spinifex/domains/admission/quota/quota.go

@@ -90,6 +90,14 @@ type Service struct {
 	// overrides holds the per-account limit overrides. Nil leaves every account
 	// on the configured limits, which is also what an absent key means.
 	overrides jetstream.KeyValue
+
+	// js opens the override bucket for SealAccountOverrides when quotas were
+	// disabled at startup; nil makes such a seal refuse rather than create it.
+	js jetstream.KeyValueManager
+
+	// terminationChecker is the seal precondition; SealAccountOverrides
+	// refuses until it is wired.
+	terminationChecker AccountTerminationChecker
 }
 
 // New constructs a quota Service from the configured limits and the gateway-owned
@@ -125,6 +133,16 @@ func (s *Service) SetOverrides(kv jetstream.KeyValue) {
 	s.overrides = kv
 }
 
+// SetJetStream wires the client SealAccountOverrides uses to get-or-create
+// the override bucket when this gateway disabled quotas. Called once at
+// wiring time, regardless of whether quotas are enabled here.
+func (s *Service) SetJetStream(js jetstream.KeyValueManager) {
+	if s == nil {
+		return
+	}
+	s.js = js
+}
+
 // Exempt returns true for the global/system account and whenever quotas are
 // disabled, so those callers bypass every quota check. A nil Service (quotas
 // never built) is treated as disabled so callers need no nil guard.

spinifex/domains/admission/quota/seal.go

@@ -0,0 +1,163 @@
+package quota
+
+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/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// sealMaxAttempts bounds SealAccountOverrides' 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. No dimension or provenance
+// survives, so no tenant commercial setting outlives the seal.
+var sealedRecord = &overrideRecord{State: OverrideStateSealed}
+
+// OverrideSeal is SealAccountOverrides' result: the revision at which the
+// stream leader accepted the sealed value.
+type OverrideSeal 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 seal precondition; until it is set,
+// SealAccountOverrides refuses.
+func (s *Service) SetAccountTerminationChecker(c AccountTerminationChecker) {
+	if s == nil {
+		return
+	}
+	s.terminationChecker = c
+}
+
+// checkTerminating runs before any override-KV call, so a refusal never
+// writes or opens the bucket. Not TERMINATING (incl. missing) is
+// IncorrectState; a read failure or unwired checker is InternalError.
+func (s *Service) checkTerminating(ctx context.Context, accountID string) error {
+	if s.terminationChecker == nil {
+		slog.ErrorContext(ctx, "Quota override 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 to retry, including the bare
+// wrong-last-sequence APIError from Create's retry against a delete marker.
+// A private copy of domains/ec2/account's helper, which this package may not import.
+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
+}
+
+// sealBucket returns this gateway's override bucket, or get-or-creates it
+// when quotas are disabled here. Called only after checkTerminating accepts,
+// so a refused seal never creates the bucket.
+func (s *Service) sealBucket(ctx context.Context) (jetstream.KeyValue, error) {
+	if s.overrides != nil {
+		return s.overrides, nil
+	}
+	if s.js == nil {
+		slog.ErrorContext(ctx, "Quota override seal has no JetStream client wired to open the override bucket")
+		return nil, errors.New(awserrors.ErrorInternalError)
+	}
+	kv, err := kvutil.GetOrCreateBucket(ctx, s.js, KVBucketAccountQuota, 1)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to open quota override bucket for seal", "err", err)
+		return nil, errors.New(awserrors.ErrorInternalError)
+	}
+	return kv, nil
+}
+
+// SealAccountOverrides seals accountID's override record so no later writer
+// can recreate it. Success is only a leader-accepted write, evidenced by the
+// revision; a missing key is created sealed, never reported from absence.
+func (s *Service) SealAccountOverrides(ctx context.Context, accountID string) (*OverrideSeal, error) {
+	if accountID == "" {
+		return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required to seal quota overrides")
+	}
+	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 seal must not target the global account")
+	}
+
+	if err := s.checkTerminating(ctx, accountID); err != nil {
+		return nil, err
+	}
+
+	overrides, err := s.sealBucket(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	data, err := json.Marshal(sealedRecord)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to encode sealed quota override record", "accountID", accountID, "err", err)
+		return nil, errors.New(awserrors.ErrorInternalError)
+	}
+
+	// accountID is never resolved through any fallback here: a seal must act
+	// on exactly accountID and never fall back to the global key.
+	for range sealMaxAttempts {
+		var rev uint64
+		entry, getErr := overrides.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 quota override for seal", "accountID", accountID, "err", getErr)
+			return nil, errors.New(awserrors.ErrorInternalError)
+		}
+
+		var newRev uint64
+		var writeErr error
+		if rev == 0 {
+			newRev, writeErr = overrides.Create(ctx, accountID, data)
+		} else {
+			newRev, writeErr = overrides.Update(ctx, accountID, data, rev)
+		}
+
+		switch {
+		case writeErr == nil:
+			return &OverrideSeal{Revision: newRev}, nil
+		case isRevisionConflict(writeErr):
+			continue
+		default:
+			slog.ErrorContext(ctx, "Failed to write sealed quota override record", "accountID", accountID, "err", writeErr)
+			return nil, errors.New(awserrors.ErrorInternalError)
+		}
+	}
+
+	slog.ErrorContext(ctx, "Quota override seal exhausted attempts against concurrent writes", "accountID", accountID, "attempts", sealMaxAttempts)
+	return nil, errors.New(awserrors.ErrorInternalError)
+}

spinifex/domains/admission/quota/seal_test.go

@@ -0,0 +1,806 @@
+package quota
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sync"
+	"testing"
+
+	"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}}
+}
+
+// newSealService builds a Service over a real JetStream KV bucket with the
+// override bucket already attached, mirroring a quota-enabled gateway.
+func newSealService(t *testing.T) (*Service, jetstream.JetStream) {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	kv, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{
+		Bucket:  KVBucketAccountQuota,
+		History: 1,
+	})
+	require.NoError(t, err)
+	usage, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{
+		Bucket:  KVBucketAccountUsage,
+		History: 1,
+	})
+	require.NoError(t, err)
+
+	svc := New(baseLimits(), usage)
+	svc.SetOverrides(kv)
+	svc.SetJetStream(js)
+	return svc, js
+}
+
+// newDisabledSealService builds a Service with only the JetStream client
+// wired, mirroring a quota-disabled gateway: the override bucket does not
+// exist until a seal creates it.
+func newDisabledSealService(t *testing.T) (*Service, jetstream.JetStream) {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	svc := New(baseLimits(), nil)
+	svc.SetJetStream(js)
+	return svc, js
+}
+
+// lastOverrideSeq reports the override stream's last sequence, so a test can
+// assert a failed attempt wrote nothing.
+func lastOverrideSeq(t *testing.T, js jetstream.JetStream) uint64 {
+	t.Helper()
+	stream, err := js.Stream(context.Background(), "KV_"+KVBucketAccountQuota)
+	require.NoError(t, err)
+	info, err := stream.Info(context.Background())
+	require.NoError(t, err)
+	return info.State.LastSeq
+}
+
+// mutationCase is one of PutAccountQuota's two mutations, called uniformly so
+// a test can table over them.
+type mutationCase struct {
+	name string
+	call func(svc *Service, ctx context.Context) error
+}
+
+func mutationTable() []mutationCase {
+	return []mutationCase{
+		{"Set", func(svc *Service, ctx context.Context) error {
+			return svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator")
+		}},
+		{"Clear", func(svc *Service, ctx context.Context) error {
+			return svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator")
+		}},
+	}
+}
+
+// seedOverrideState prepares testAccount'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 seedOverrideState(t *testing.T, svc *Service, state string) {
+	t.Helper()
+	ctx := context.Background()
+	switch state {
+	case "present":
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+	case "tombstone":
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+		require.NoError(t, svc.overrides.Delete(ctx, testAccount))
+	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, Update or Delete),
+// 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)
+}
+
+func (f *sealRaceKV) Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error {
+	f.writeCalls++
+	if f.beforeWrite != nil {
+		if err := f.beforeWrite(f.writeCalls); err != nil {
+			return err
+		}
+	}
+	return f.KeyValue.Delete(ctx, key, opts...)
+}
+
+// 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 decodeOverride(t *testing.T, entry jetstream.KeyValueEntry) overrideRecord {
+	t.Helper()
+	var rec overrideRecord
+	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. Clear on an absent or tombstoned key never reaches a
+// write at all (see TestClearOnMissingKeyWritesNothing), so only the four
+// cells with a real write are exercised here.
+func TestSealRaceMutation_SealLandsBetweenReadAndWrite(t *testing.T) {
+	cases := []struct {
+		name  string
+		state string
+		call  func(svc *Service, ctx context.Context) error
+	}{
+		{"Set_present", "present", mutationTable()[0].call},
+		{"Set_absent", "absent", mutationTable()[0].call},
+		{"Set_tombstone", "tombstone", mutationTable()[0].call},
+		{"Clear_present", "present", mutationTable()[1].call},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			svc, _ := newSealService(t)
+			ctx := context.Background()
+			seedOverrideState(t, svc, tc.state)
+
+			realKV := svc.overrides
+			sealer := New(baseLimits(), nil)
+			sealer.SetOverrides(realKV)
+			sealer.SetAccountTerminationChecker(terminatingChecker(testAccount))
+
+			var sealed *OverrideSeal
+			race := &sealRaceKV{KeyValue: realKV}
+			race.beforeWrite = func(attempt int) error {
+				if attempt != 1 {
+					return nil
+				}
+				var sealErr error
+				sealed, sealErr = sealer.SealAccountOverrides(ctx, testAccount)
+				return sealErr
+			}
+			svc.overrides = race
+
+			err := tc.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, testAccount)
+			require.NoError(t, err)
+			assert.Equal(t, OverrideStateSealed, decodeOverride(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 no overridden dimension.
+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, _ := newSealService(t)
+				ctx := context.Background()
+				seedOverrideState(t, svc, state)
+
+				realKV := svc.overrides
+				race := &sealRaceKV{KeyValue: realKV}
+				race.beforeWrite = func(attempt int) error {
+					if attempt != 1 {
+						return nil
+					}
+					_ = m.call(svc, ctx)
+					return nil
+				}
+
+				sealer := New(baseLimits(), nil)
+				sealer.SetOverrides(race)
+				sealer.SetAccountTerminationChecker(terminatingChecker(testAccount))
+
+				seal, err := sealer.SealAccountOverrides(ctx, testAccount)
+				require.NoError(t, err)
+
+				entry, err := realKV.Get(ctx, testAccount)
+				require.NoError(t, err)
+				rec := decodeOverride(t, entry)
+				assert.Equal(t, OverrideStateSealed, rec.State)
+				assert.Nil(t, rec.VCPUs)
+				assert.Equal(t, seal.Revision, entry.Revision())
+			})
+		}
+	}
+}
+
+// Neither mutation 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, _ := newSealService(t)
+			ctx := context.Background()
+			svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+			_, err := svc.SealAccountOverrides(ctx, testAccount)
+			require.NoError(t, err)
+
+			before, err := svc.overrides.Get(ctx, testAccount)
+			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.overrides.Get(ctx, testAccount)
+			require.NoError(t, err)
+			assert.Equal(t, before.Revision(), after.Revision())
+		})
+	}
+}
+
+// Every override 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, _ := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+
+	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 *OverrideSeal
+	var sealErr error
+	wg.Go(func() {
+		sealResult, sealErr = svc.SealAccountOverrides(ctx, testAccount)
+	})
+	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.overrides.Get(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State)
+	assert.Equal(t, sealResult.Revision, entry.Revision())
+
+	_, _, err = svc.GetAccountQuota(ctx, testAccount)
+	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, js := newSealService(t)
+	ctx := context.Background()
+
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+	stale, err := svc.overrides.Get(ctx, testAccount)
+	require.NoError(t, err)
+
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(16)}, "operator"))
+
+	before := lastOverrideSeq(t, js)
+	svc.overrides = &staleGetKV{KeyValue: svc.overrides, entry: stale}
+
+	err = svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+// 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, js := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+
+	before := lastOverrideSeq(t, js)
+	svc.overrides = &staleNotFoundKV{KeyValue: svc.overrides}
+
+	err = svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+// 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, _ := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+
+	realKV := svc.overrides
+	race := &sealRaceKV{KeyValue: realKV}
+	race.beforeWrite = func(int) error {
+		data, merr := json.Marshal(&overrideRecord{Overrides: Overrides{VCPUs: ptr(16)}})
+		require.NoError(t, merr)
+		_, perr := realKV.Put(ctx, testAccount, data)
+		return perr
+	}
+	svc.overrides = race
+
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, sealMaxAttempts, race.writeCalls)
+
+	entry, err := realKV.Get(ctx, testAccount)
+	require.NoError(t, err)
+	assert.NotEqual(t, OverrideStateSealed, decodeOverride(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, js := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+
+	before := lastOverrideSeq(t, js)
+	svc.overrides = &errGetKV{KeyValue: svc.overrides, err: errors.New("kv unavailable")}
+
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+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))
+}
+
+// Clearing never writes when there is nothing active to clear: an absent key,
+// a tombstoned key, and a sealed key hidden behind a stale not-found Get all
+// leave the stream exactly where they found it.
+func TestClearOnMissingKeyWritesNothing(t *testing.T) {
+	t.Run("absent", func(t *testing.T) {
+		svc, js := newSealService(t)
+		ctx := context.Background()
+		before := lastOverrideSeq(t, js)
+
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"))
+		assert.Equal(t, before, lastOverrideSeq(t, js))
+	})
+
+	t.Run("tombstone", func(t *testing.T) {
+		svc, js := newSealService(t)
+		ctx := context.Background()
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+		require.NoError(t, svc.overrides.Delete(ctx, testAccount))
+		before := lastOverrideSeq(t, js)
+
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"))
+		assert.Equal(t, before, lastOverrideSeq(t, js))
+	})
+
+	t.Run("sealed_hidden_by_stale_not_found", func(t *testing.T) {
+		svc, js := newSealService(t)
+		ctx := context.Background()
+		svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+		sealed, err := svc.SealAccountOverrides(ctx, testAccount)
+		require.NoError(t, err)
+		before := lastOverrideSeq(t, js)
+
+		realKV := svc.overrides
+		svc.overrides = &staleNotFoundKV{KeyValue: realKV}
+
+		require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"))
+		assert.Equal(t, before, lastOverrideSeq(t, js))
+
+		entry, err := realKV.Get(ctx, testAccount)
+		require.NoError(t, err)
+		rec := decodeOverride(t, entry)
+		assert.Equal(t, OverrideStateSealed, rec.State)
+		assert.Equal(t, sealed.Revision, entry.Revision())
+	})
+}
+
+// A seal against an absent or a tombstoned key creates the sealed record.
+func TestSealOnMissingKeyCreatesSealedRecord(t *testing.T) {
+	for _, state := range []string{"absent", "tombstone"} {
+		t.Run(state, func(t *testing.T) {
+			svc, _ := newSealService(t)
+			ctx := context.Background()
+			svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+			if state == "tombstone" {
+				require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+				require.NoError(t, svc.overrides.Delete(ctx, testAccount))
+			}
+			_, err := svc.overrides.Get(ctx, testAccount)
+			require.ErrorIs(t, err, jetstream.ErrKeyNotFound)
+
+			seal, err := svc.SealAccountOverrides(ctx, testAccount)
+			require.NoError(t, err)
+
+			entry, err := svc.overrides.Get(ctx, testAccount)
+			require.NoError(t, err)
+			assert.Equal(t, seal.Revision, entry.Revision())
+			assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State)
+		})
+	}
+}
+
+// alwaysConflictThenErrorKV answers Get as absent and every Create attempt
+// with a conflict, then a store error on the final attempt: a recording
+// wrapper proving the seal tried a real write on every attempt before giving
+// up, rather than ever reporting success from an observed absence.
+type alwaysConflictThenErrorKV struct {
+	jetstream.KeyValue
+
+	calls int
+}
+
+func (f *alwaysConflictThenErrorKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) {
+	return nil, jetstream.ErrKeyNotFound
+}
+
+func (f *alwaysConflictThenErrorKV) Create(context.Context, string, []byte, ...jetstream.KVCreateOpt) (uint64, error) {
+	f.calls++
+	if f.calls < sealMaxAttempts {
+		return 0, jetstream.ErrKeyExists
+	}
+	return 0, errors.New("store unavailable")
+}
+
+func TestSealNeverSucceedsWithoutAWrite(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	fake := &alwaysConflictThenErrorKV{}
+	svc.overrides = fake
+
+	seal, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.Error(t, err)
+	assert.Nil(t, seal)
+	assert.Equal(t, sealMaxAttempts, fake.calls, "every attempt must have tried a real write before giving up")
+}
+
+// An account IAM does not hold as TERMINATING refuses the seal before any
+// override-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, js := newSealService(t)
+			ctx := context.Background()
+			if state == "present" {
+				require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"))
+			}
+			svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}})
+
+			before := lastOverrideSeq(t, js)
+			_, err := svc.SealAccountOverrides(ctx, testAccount)
+			assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err)
+			assert.Equal(t, before, lastOverrideSeq(t, js))
+		})
+	}
+}
+
+func TestSealRefusedForMissingIAMRecord(t *testing.T) {
+	svc, js := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}})
+
+	before := lastOverrideSeq(t, js)
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+func TestSealRefusedWhenIAMReadFails(t *testing.T) {
+	svc, js := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(&stubTerminationChecker{err: errors.New("iam unavailable")})
+
+	before := lastOverrideSeq(t, js)
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+func TestSealRefusedWithNoCheckerWired(t *testing.T) {
+	svc, js := newSealService(t)
+	ctx := context.Background()
+
+	before := lastOverrideSeq(t, js)
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Equal(t, before, lastOverrideSeq(t, js))
+}
+
+func TestSealAcceptedForTerminatingAccount(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+
+	seal, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+	assert.NotZero(t, seal.Revision)
+
+	entry, err := svc.overrides.Get(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, seal.Revision, entry.Revision())
+}
+
+// A refused seal, on a gateway that disabled quotas and so holds no override
+// bucket, never creates one.
+func TestSealRefusalDoesNotCreateBucket(t *testing.T) {
+	t.Run("not_terminating", func(t *testing.T) {
+		svc, js := newDisabledSealService(t)
+		ctx := context.Background()
+		svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}})
+
+		_, err := svc.SealAccountOverrides(ctx, testAccount)
+		assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err)
+		_, err = js.KeyValue(context.Background(), KVBucketAccountQuota)
+		assert.ErrorIs(t, err, jetstream.ErrBucketNotFound)
+	})
+
+	t.Run("iam_read_fails", func(t *testing.T) {
+		svc, js := newDisabledSealService(t)
+		ctx := context.Background()
+		svc.SetAccountTerminationChecker(&stubTerminationChecker{err: errors.New("iam unavailable")})
+
+		_, err := svc.SealAccountOverrides(ctx, testAccount)
+		assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+		_, err = js.KeyValue(context.Background(), KVBucketAccountQuota)
+		assert.ErrorIs(t, err, jetstream.ErrBucketNotFound)
+	})
+}
+
+// A seal accepted on a gateway whose quotas were disabled at startup creates
+// the override bucket at seal time, never at startup.
+func TestSealOnQuotaDisabledGatewayCreatesBucket(t *testing.T) {
+	svc, js := newDisabledSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+
+	seal, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+
+	kv, err := js.KeyValue(context.Background(), KVBucketAccountQuota)
+	require.NoError(t, err)
+	entry, err := kv.Get(context.Background(), testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, seal.Revision, entry.Revision())
+	assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State)
+}
+
+// A sealed record stays sealed after the service is rebuilt over the same
+// JetStream, every read and write refuses it, and a re-seal gets a higher
+// revision.
+func TestSealSurvivesServiceReconstruction(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	ctx := context.Background()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: KVBucketAccountQuota, History: 1})
+	require.NoError(t, err)
+
+	svc := New(baseLimits(), nil)
+	svc.SetOverrides(kv)
+	svc.SetJetStream(js)
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	first, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+
+	rebuiltKV, err := js.KeyValue(ctx, KVBucketAccountQuota)
+	require.NoError(t, err)
+	rebuilt := New(baseLimits(), nil)
+	rebuilt.SetOverrides(rebuiltKV)
+	rebuilt.SetJetStream(js)
+
+	entry, err := rebuilt.overrides.Get(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State)
+
+	_, _, err = rebuilt.GetAccountQuota(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	assert.True(t, awserrors.IsErrorCode(rebuilt.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"), awserrors.ErrorBlocked))
+	assert.True(t, awserrors.IsErrorCode(rebuilt.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"), awserrors.ErrorBlocked))
+
+	limits, err := rebuilt.limitsFor(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked))
+	assert.Zero(t, limits)
+
+	rebuilt.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	second, err := rebuilt.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+	assert.Greater(t, second.Revision, first.Revision)
+}
+
+func TestSealRefusesEmptyInvalidAndGlobalAccount(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+
+	_, err := svc.SealAccountOverrides(ctx, "")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorMissingParameter), "got %v", err)
+
+	_, err = svc.SealAccountOverrides(ctx, "not-an-account")
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err)
+
+	_, err = svc.SealAccountOverrides(ctx, awsidentifiers.GlobalAccountID)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err)
+}
+
+func TestSealLeavesOtherAccountsUntouched(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+	accountB := "222222222222"
+	require.NoError(t, svc.PutAccountQuota(ctx, accountB, Overrides{VCPUs: ptr(8)}, "operator"))
+
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+
+	over, _, err := svc.GetAccountQuota(ctx, accountB)
+	require.NoError(t, err)
+	require.NotNil(t, over.VCPUs)
+	assert.Equal(t, 8, *over.VCPUs)
+}
+
+// A record written before the state field existed has none and is treated as
+// active by both reads and writes; a write on it preserves the old shape.
+func TestRecordWithoutStateIsActive(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+
+	data, err := json.Marshal(Overrides{VCPUs: ptr(8)})
+	require.NoError(t, err)
+	_, err = svc.overrides.Put(ctx, testAccount, data)
+	require.NoError(t, err)
+
+	over, limits, err := svc.GetAccountQuota(ctx, testAccount)
+	require.NoError(t, err)
+	require.NotNil(t, over.VCPUs)
+	assert.Equal(t, 8, limits.VCPUs)
+
+	require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(16)}, "operator"))
+	entry, err := svc.overrides.Get(ctx, testAccount)
+	require.NoError(t, err)
+	assert.NotContains(t, string(entry.Value()), `"state"`)
+}
+
+// A record whose state is neither active nor sealed fails every read and
+// write closed, rather than being treated as active.
+func TestUnknownStateFailsClosed(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+
+	_, err := svc.overrides.Put(ctx, testAccount, []byte(`{"state":"frozen"}`))
+	require.NoError(t, err)
+
+	_, _, err = svc.GetAccountQuota(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+
+	limits, err := svc.limitsFor(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err)
+	assert.Zero(t, limits)
+
+	assert.True(t, awserrors.IsErrorCode(svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"), awserrors.ErrorInternalError))
+	assert.True(t, awserrors.IsErrorCode(svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"), awserrors.ErrorInternalError))
+}
+
+// A sealed record is Blocked for admission's own resolver, which every
+// caller (admin read, ELBv2 DescribeAccountLimits, every Enforce* method)
+// shares.
+func TestAdmissionOnSealedRecordIsBlocked(t *testing.T) {
+	svc, _ := newSealService(t)
+	ctx := context.Background()
+	svc.SetAccountTerminationChecker(terminatingChecker(testAccount))
+	_, err := svc.SealAccountOverrides(ctx, testAccount)
+	require.NoError(t, err)
+
+	limits, err := svc.limitsFor(ctx, testAccount)
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err)
+	assert.Zero(t, limits)
+
+	assert.Error(t, svc.CheckVCPU(ctx, testAccount, 1))
+	assert.True(t, awserrors.IsErrorCode(svc.CheckVCPU(ctx, testAccount, 1), awserrors.ErrorBlocked))
+}
+
+// The existing read-failure fallback in limitsFor (a decode or store error
+// falls back to the configured limits) is unchanged by the seal work: it is
+// distinct from a successfully read sealed record, which refuses instead.
+func TestAdmissionReadFailureFallbackUnchanged(t *testing.T) {
+	svc, kv := newOverrideService(t)
+
+	_, err := kv.Put(t.Context(), testAccount, []byte("{not json"))
+	require.NoError(t, err)
+
+	limits, err := svc.limitsFor(t.Context(), testAccount)
+	require.NoError(t, err, "a decode failure must still fall back, not error, from limitsFor")
+	require.Equal(t, baseLimits(), limits)
+}

spinifex/domains/admission/quota/storage.go

@@ -23,7 +23,10 @@ func (s *Service) EnforceVolumeCreate(ctx context.Context, natsConn *nats.Conn,
 	if err != nil {
 		return err
 	}
-	limits := s.limitsFor(ctx, accountID)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
 	if err := exceeds(len(out.Volumes), 1, limits.Volumes); err != nil {
 		return err
 	}
@@ -43,7 +46,11 @@ func (s *Service) EnforceVolumeModify(ctx context.Context, natsConn *nats.Conn,
 	if err != nil {
 		return err
 	}
-	return exceeds(total-oldGiB, newGiB, s.limitsFor(ctx, accountID).VolumesGiB)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(total-oldGiB, newGiB, limits.VolumesGiB)
 }
 
 // volumeUsage sums the account's volume sizes in GiB via a single DescribeVolumes

spinifex/domains/admission/quota/vcpu.go

@@ -37,7 +37,11 @@ func (s *Service) CheckVCPU(ctx context.Context, accountID string, want int) err
 	if err != nil {
 		return err
 	}
-	return exceeds(current, want, s.limitsFor(ctx, accountID).VCPUs)
+	limits, err := s.limitsFor(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	return exceeds(current, want, limits.VCPUs)
 }
 
 // casVCPU applies next to accountID's counter under bounded JetStream CAS so

spinifex/domains/ec2/account/settings_writes_guarded_test.go

@@ -0,0 +1,68 @@
+package account
+
+//test:in-package — pins what AccountSettingsSealCapabilityVersion 1 asserts:
+// nothing in this package writes the settings KV except through the
+// revision-guarded helpers.
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"io/fs"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"testing"
+)
+
+func accountPackageDir(t *testing.T) string {
+	t.Helper()
+	_, file, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("cannot locate test source")
+	}
+	return filepath.Dir(file)
+}
+
+// The capability daemons report as version 1 is a claim about every writer in
+// the binary, not just the responder: no non-test file in this package may
+// call Put or PutString, which bypass the revision-guarded mutate/seal paths.
+func TestAccountSettingsWritesAreRevisionGuarded(t *testing.T) {
+	root := accountPackageDir(t)
+	fset := token.NewFileSet()
+	scanned := 0
+
+	err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
+		if err != nil || d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
+			return err
+		}
+		f, perr := parser.ParseFile(fset, path, nil, 0)
+		if perr != nil {
+			return perr
+		}
+		scanned++
+
+		ast.Inspect(f, func(n ast.Node) bool {
+			call, ok := n.(*ast.CallExpr)
+			if !ok {
+				return true
+			}
+			sel, ok := call.Fun.(*ast.SelectorExpr)
+			if !ok {
+				return true
+			}
+			if sel.Sel.Name == "Put" || sel.Sel.Name == "PutString" {
+				t.Errorf("%s: unconditional %s call bypasses the revision-guarded settings write; use mutateSettings or SealAccount's CAS loop instead",
+					fset.Position(call.Pos()), sel.Sel.Name)
+			}
+			return true
+		})
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if scanned == 0 {
+		t.Fatalf("scanned no Go files under %s; the invariant would pass vacuously", root)
+	}
+}

spinifex/domains/ec2/awsapi/instance/DescribeInstanceTypes.go

@@ -3,7 +3,6 @@ package instance
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"log/slog"
@@ -101,10 +100,10 @@ func DescribeInstanceTypes(ctx context.Context, input *ec2.DescribeInstanceTypes
 
 		if missing := missingInstanceTypes(finalInstanceTypes, input.InstanceTypes); len(missing) > 0 {
 			if !proveAbsence || !everyNodeAnswered(nodeIDs, answered, sum) {
-				return nil, errors.Join(
+				return nil, awserrors.WithRetryAfter(
 					awserrors.Errorf(awserrors.ErrorServiceUnavailable,
 						"cannot confirm instance type(s) %s: describe fan-out did not complete", strings.Join(missing, ", ")),
-					awserrors.RetryAfter(awserrors.ErrorServiceUnavailable, describeIncompleteRetryAfter),
+					describeIncompleteRetryAfter,
 				)
 			}
 			return nil, awserrors.Errorf(awserrors.ErrorInvalidInstanceType,

spinifex/domains/ec2/awsapi/instance/DescribeInstances.go

@@ -126,10 +126,10 @@ func DescribeInstancesChecked(ctx context.Context, input *ec2.DescribeInstancesI
 			// An incomplete sweep cannot prove absence, so a missing id is
 			// unresolved, not gone. NotFound would be a false negative and an
 			// empty success would look identical to a real one, so return 503.
-			return nil, errors.Join(
+			return nil, awserrors.WithRetryAfter(
 				awserrors.Errorf(awserrors.ErrorServiceUnavailable,
 					"cannot confirm status of instance(s) %s: describe fan-out did not complete", strings.Join(missing, ", ")),
-				awserrors.RetryAfter(awserrors.ErrorServiceUnavailable, describeIncompleteRetryAfter),
+				describeIncompleteRetryAfter,
 			)
 		}
 	}
@@ -198,10 +198,10 @@ func DescribeAccountInstancesComplete(ctx context.Context, natsConn *nats.Conn,
 }
 
 func incompleteListingError() error {
-	return errors.Join(
+	return awserrors.WithRetryAfter(
 		awserrors.Errorf(awserrors.ErrorServiceUnavailable,
 			"cannot list instances: not every node and instance store answered"),
-		awserrors.RetryAfter(awserrors.ErrorServiceUnavailable, describeIncompleteRetryAfter),
+		describeIncompleteRetryAfter,
 	)
 }
 

spinifex/foundation/aws/errors/awserrors.go

@@ -658,13 +658,22 @@ func HasDependencies(kind, id string) error {
 // retryableError wraps a registered code with a suggested Retry-After
 // duration, for a 503 that a client can expect to clear on its own within a
 // bounded window (e.g. a warm-up race) rather than an open-ended outage.
-// Error() returns only the code, so code resolution is unaffected.
+// Error() returns only the code, or the wrapped error's text, so code
+// resolution is unaffected.
 type retryableError struct {
 	code       string
+	err        error
 	retryAfter time.Duration
 }
 
-func (e *retryableError) Error() string { return e.code }
+func (e *retryableError) Error() string {
+	if e.err != nil {
+		return e.err.Error()
+	}
+	return e.code
+}
+
+func (e *retryableError) Unwrap() error { return e.err }
 
 // RetryAfter wraps code (a registered ErrorLookup key) with a suggested
 // Retry-After duration for ErrorHandler to surface as a response header.
@@ -672,6 +681,13 @@ func RetryAfter(code string, d time.Duration) error {
 	return &retryableError{code: code, retryAfter: d}
 }
 
+// WithRetryAfter attaches a suggested Retry-After duration to err without
+// changing its text, unlike joining err with RetryAfter, which appends the
+// bare code on a second line.
+func WithRetryAfter(err error, d time.Duration) error {
+	return &retryableError{err: err, retryAfter: d}
+}
+
 // ResolveRetryAfter returns the Retry-After duration attached to err via
 // RetryAfter, if any, walking the same unwrap tree ResolveErrorDetail does.
 func ResolveRetryAfter(err error) (time.Duration, bool) {

spinifex/foundation/aws/errors/awserrors_test.go

@@ -176,6 +176,24 @@ func TestResolveRetryAfter_WrappedAndResolvable(t *testing.T) {
 	}
 }
 
+// WithRetryAfter keeps the coded error's text and message while still
+// carrying the hint: a joined RetryAfter would log the code twice.
+func TestWithRetryAfter_KeepsTextAndResolves(t *testing.T) {
+	inner := Errorf(ErrorServiceUnavailable, "probe incomplete")
+	err := WithRetryAfter(inner, 5*time.Second)
+
+	if err.Error() != inner.Error() {
+		t.Fatalf("Error() = %q, want %q", err.Error(), inner.Error())
+	}
+	if d, ok := ResolveRetryAfter(fmt.Errorf("wrapped: %w", err)); !ok || d != 5*time.Second {
+		t.Fatalf("ResolveRetryAfter() = (%v, %v), want (5s, true)", d, ok)
+	}
+	code, msg, ok := ResolveErrorDetail(err)
+	if !ok || code != ErrorServiceUnavailable || msg != "probe incomplete" {
+		t.Fatalf("ResolveErrorDetail() = (%q, %q, %v), want (%q, %q, true)", code, msg, ok, ErrorServiceUnavailable, "probe incomplete")
+	}
+}
+
 // TestResolveRetryAfter_PlainErrorNotFound is the compatibility case: an
 // ordinary errors.New(code) -- the overwhelming majority of call sites --
 // carries no Retry-After hint.

spinifex/gateway/admin_accountquota.go

@@ -43,6 +43,9 @@ func (gw *GatewayConfig) adminGetAccountQuota(ctx context.Context, body []byte)
 
 	over, limits, err := gw.Quota.GetAccountQuota(ctx, accountID)
 	if err != nil {
+		if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) {
+			return nil, err
+		}
 		slog.ErrorContext(ctx, "GetAccountQuota: failed to read override", "accountID", accountID, "err", err)
 		return nil, errors.New(awserrors.ErrorInternalError)
 	}
@@ -61,6 +64,9 @@ func (gw *GatewayConfig) adminPutAccountQuota(ctx context.Context, body []byte)
 	// The caller's own identity, so a quota change names who made it.
 	updatedBy, _ := ctx.Value(ctxIdentity).(string)
 	if err := gw.Quota.PutAccountQuota(ctx, accountID, over, updatedBy); err != nil {
+		if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) {
+			return nil, err
+		}
 		slog.ErrorContext(ctx, "PutAccountQuota: failed to store override", "accountID", accountID, "err", err)
 		return nil, errors.New(awserrors.ErrorInternalError)
 	}
@@ -69,6 +75,9 @@ func (gw *GatewayConfig) adminPutAccountQuota(ctx context.Context, body []byte)
 
 	stored, limits, err := gw.Quota.GetAccountQuota(ctx, accountID)
 	if err != nil {
+		if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) {
+			return nil, err
+		}
 		slog.ErrorContext(ctx, "PutAccountQuota: failed to read back override", "accountID", accountID, "err", err)
 		return nil, errors.New(awserrors.ErrorInternalError)
 	}

spinifex/gateway/admin_accountquota_test.go

@@ -205,3 +205,56 @@ func TestAccountQuotaMethodsAreRegistered(t *testing.T) {
 	require.Contains(t, AdminMethodNames(), "GetAccountQuota")
 	require.Contains(t, AdminMethodNames(), "PutAccountQuota")
 }
+
+// quotaStubChecker answers AccountTerminating from a fixed set, so these
+// tests can seal a record without a real IAM bucket.
+type quotaStubChecker struct {
+	terminating map[string]bool
+}
+
+func (c *quotaStubChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) {
+	return c.terminating[accountID], nil
+}
+
+// A request body cannot carry or alter the stored state: an attempted
+// "state":"sealed" in the overrides object is simply unknown to the wire
+// type, the stored record stays active, and the response never shows state.
+func TestAdminRequestCannotCarryState(t *testing.T) {
+	gw, account := quotaTestGateway(t)
+	ctx := context.WithValue(t.Context(), ctxIdentity, "operator")
+
+	body := []byte(`{"accountId":"` + account + `","overrides":{"vcpus":32,"state":"sealed"}}`)
+	output, err := gw.adminPutAccountQuota(ctx, body)
+	require.NoError(t, err)
+
+	resp, ok := output.(*AccountQuotaResponse)
+	require.True(t, ok)
+	require.Equal(t, 32, resp.Limits["vcpus"])
+
+	data, err := json.Marshal(resp.Overrides)
+	require.NoError(t, err)
+	require.NotContains(t, string(data), "state")
+
+	_, _, err = gw.Quota.GetAccountQuota(ctx, account)
+	require.NoError(t, err, "the record must still be active, not sealed")
+}
+
+// Admin get and the put read-back on a sealed record must return Blocked,
+// not the generic InternalError every other read failure maps to.
+func TestAdminQuotaReadsMapBlocked(t *testing.T) {
+	gw, account := quotaTestGateway(t)
+	gw.Quota.SetAccountTerminationChecker(&quotaStubChecker{terminating: map[string]bool{account: true}})
+	_, err := gw.Quota.SealAccountOverrides(t.Context(), account)
+	require.NoError(t, err)
+
+	_, err = gw.adminGetAccountQuota(t.Context(), quotaRequestBody(t, AccountQuotaRequest{AccountID: account}))
+	require.Error(t, err)
+	require.Equal(t, awserrors.ErrorBlocked, err.Error())
+
+	ctx := context.WithValue(t.Context(), ctxIdentity, "operator")
+	_, err = gw.adminPutAccountQuota(ctx, quotaRequestBody(t, AccountQuotaRequest{
+		AccountID: account, Overrides: admissionquota.Overrides{VCPUs: quotaPtr(8)},
+	}))
+	require.Error(t, err)
+	require.Equal(t, awserrors.ErrorBlocked, err.Error())
+}

spinifex/gateway/admin_deleteaccount_job_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/s3"
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/accountteardown"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -31,6 +32,10 @@ const deleteClientToken = "8b9c1d2e3f4a5b6c7d8e9f0a1b2c3d4e"
 // to keep step with the reapers NewClusterEngine wires.
 var teardownSubjects = []string{"ec2.>", "ecs.>", "eks.>", "rds.>", "elbv2.>", "acm.>", "bedrock.>"}
 
+// sealCapabilitySubjectPrefix matches every per-node seal-capability subject,
+// which the generic ec2.> stub below must not also answer.
+const sealCapabilitySubjectPrefix = "ec2.v1.AccountSettingsSealCapability."
+
 // newDeleteAccountGateway wires a gateway against a real IAM service and
 // embedded JetStream, with a stub cluster answering the subjects the reapers
 // drive. The tenant has no resources, so every stage drains empty and the test
@@ -46,8 +51,12 @@ func newDeleteAccountGateway(t *testing.T) *GatewayConfig {
 
 	for _, subject := range teardownSubjects {
 		// Replies carry a node ID: the instance inventory only counts a frame
-		// it can attribute to a configured node.
+		// it can attribute to a configured node. The seal subjects have their
+		// own responders below, which a second reply here would race.
 		sub, err := nc.Subscribe(subject, func(msg *nats.Msg) {
+			if msg.Subject == ec2v1.SealAccountSettingsSubject || strings.HasPrefix(msg.Subject, sealCapabilitySubjectPrefix) {
+				return
+			}
 			reply := nats.NewMsg(msg.Reply)
 			reply.Data = []byte(`{}`)
 			reply.Header.Set(natsmsg.NodeIDHeader, "node-1")
@@ -56,6 +65,38 @@ func newDeleteAccountGateway(t *testing.T) *GatewayConfig {
 		require.NoError(t, err)
 		t.Cleanup(func() { _ = sub.Unsubscribe() })
 	}
+
+	// The seal reaper needs a real sealed disposition and revision, and every
+	// configured node answering capability version 1, or it never finalises:
+	// the generic `{}` stub above is neither.
+	sealSub, err := nc.Subscribe(ec2v1.SealAccountSettingsSubject, func(msg *nats.Msg) {
+		var req ec2v1.SealAccountSettingsRequest
+		if json.Unmarshal(msg.Data, &req) != nil {
+			return
+		}
+		data, _ := json.Marshal(ec2v1.SealAccountSettingsReply{
+			AccountID: req.AccountID, CorrelationRef: req.CorrelationRef,
+			Disposition: ec2v1.AccountSettingsDispositionSealed, Revision: 1,
+		})
+		reply := nats.NewMsg(msg.Reply)
+		reply.Data = data
+		reply.Header.Set(natsmsg.NodeIDHeader, "node-1")
+		_ = msg.RespondMsg(reply)
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sealSub.Unsubscribe() })
+
+	for _, node := range []string{"node-1"} {
+		capSub, err := nc.Subscribe(ec2v1.AccountSettingsSealCapabilitySubject(node), func(msg *nats.Msg) {
+			data, _ := json.Marshal(ec2v1.AccountSettingsSealCapabilityReply{Node: node, CapabilityVersion: ec2v1.AccountSettingsSealCapabilityVersion})
+			reply := nats.NewMsg(msg.Reply)
+			reply.Data = data
+			reply.Header.Set(natsmsg.NodeIDHeader, node)
+			_ = msg.RespondMsg(reply)
+		})
+		require.NoError(t, err)
+		t.Cleanup(func() { _ = capSub.Unsubscribe() })
+	}
 	require.NoError(t, nc.Flush())
 
 	return &GatewayConfig{

spinifex/gateway/elbv2.go

@@ -187,6 +187,9 @@ func accountLoadBalancerLimit(ctx context.Context, gw *GatewayConfig, accountID
 	}
 	_, limits, err := gw.Quota.GetAccountQuota(ctx, accountID)
 	if err != nil {
+		if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) {
+			return 0, err
+		}
 		slog.ErrorContext(ctx, "DescribeAccountLimits: failed to read account quota", "accountID", accountID, "err", err)
 		return 0, errors.New(awserrors.ErrorInternalError)
 	}

spinifex/gateway/elbv2_accountlimits_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/service/elbv2"
 	"github.com/mulgadc/spinifex/spinifex/domains/admission/quota"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	gateway_elbv2 "github.com/mulgadc/spinifex/spinifex/gateway/elbv2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -121,6 +122,20 @@ func TestELBv2AccountLoadBalancerLimitDefaultsForExemptAccount(t *testing.T) {
 	assert.Equal(t, gateway_elbv2.DefaultLoadBalancerLimit, limit)
 }
 
+// A sealed override must refuse the load balancer cap with Blocked rather
+// than the generic InternalError every other read failure maps to.
+func TestELBv2AccountLoadBalancerLimitBlockedOnSealedRecord(t *testing.T) {
+	gw, account := quotaTestGateway(t)
+	gw.Quota.SetAccountTerminationChecker(&quotaStubChecker{terminating: map[string]bool{account: true}})
+	_, err := gw.Quota.SealAccountOverrides(t.Context(), account)
+	require.NoError(t, err)
+
+	limit, err := accountLoadBalancerLimit(t.Context(), gw, account)
+	require.Error(t, err)
+	require.Equal(t, awserrors.ErrorBlocked, err.Error())
+	assert.Zero(t, limit)
+}
+
 // The dispatch entry must accept the paging parameters the model defines and
 // still answer with the whole list.
 func TestELBv2DescribeAccountLimitsAcceptsPaging(t *testing.T) {

spinifex/handlers/iam/account_status_reader.go

@@ -10,8 +10,8 @@ import (
 )
 
 // 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.
+// writes. It is the read-only capability EC2's settings seal and the quota
+// override seal both use to check their TERMINATING precondition.
 type AccountStatusReader struct {
 	accounts *kvstore.Bucket // AttachOnly: never creates the bucket
 }

spinifex/operator/admin/image_import.go

@@ -71,12 +71,7 @@ func ImportImage(ctx context.Context, provider ebsprovider.EBSProvider, opts Imp
 		return fmt.Errorf("import: publish volume %s: %w", opts.VolumeID, err)
 	}
 
-	var writeErr error
-	if published.Export.Kind == ebsprovider.ExportKindNBD && published.Export.NBD != nil {
-		writeErr = writeImage(ctx, opts.SourcePath, published.Export.NBD.NBDURI, opts.Progress)
-	} else {
-		writeErr = fmt.Errorf("volume is published as a %q export; image import writes only through nbd", published.Export.Kind)
-	}
+	writeErr := writeImage(ctx, opts.SourcePath, published.NBDURI, opts.Progress)
 
 	// Unpublish regardless: a volume left published cannot be snapshotted or
 	// attached, and on the failure path it would strand the export too.

spinifex/operator/admin/image_import_test.go

@@ -22,10 +22,6 @@ type recordingProvider struct {
 	publishes  []ebsprovider.PublishVolumeRequest
 	snapshots  []ebsprovider.CreateSnapshotRequest
 	publishURI string
-
-	// publishExport, when its Kind is set, replaces the nbd export built
-	// from publishURI.
-	publishExport ebsprovider.VolumeExport
 }
 
 func newRecordingProvider() *recordingProvider {
@@ -43,11 +39,7 @@ func (p *recordingProvider) PublishVolume(ctx context.Context, req ebsprovider.P
 	if _, err := p.MemoryProvider.PublishVolume(ctx, req); err != nil {
 		return nil, err
 	}
-	export := ebsprovider.NewNBDExport(p.publishURI)
-	if p.publishExport.Kind != "" {
-		export = p.publishExport
-	}
-	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: export}, nil
+	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, NBDURI: p.publishURI}, nil
 }
 
 func (p *recordingProvider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) error {
@@ -125,32 +117,6 @@ func TestImportImage_WriteFailureStillUnpublishes(t *testing.T) {
 		"a failed copy must not be snapshotted, and must still be unpublished")
 }
 
-// TestImportImage_BlockExportIsRefused covers a provider that publishes a host
-// block device: the import writes only through NBD, so it must fail without
-// copying and still release the export.
-func TestImportImage_BlockExportIsRefused(t *testing.T) {
-	var wroteTo string
-	stubWriteImage(t, nil, &wroteTo)
-	provider := newRecordingProvider()
-	provider.publishExport = ebsprovider.VolumeExport{
-		Kind:  ebsprovider.ExportKindBlock,
-		Block: &ebsprovider.BlockExport{DevicePath: "/dev/sdx", SizeBytes: 1 << 30},
-	}
-
-	err := ImportImage(context.Background(), provider, ImportOpts{
-		VolumeID:   "ami-import005",
-		NodeID:     "node-a",
-		SizeBytes:  1 << 30,
-		SourcePath: "/dev/null",
-		Snapshot:   true,
-	})
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), `"block" export`)
-	assert.Empty(t, wroteTo, "a block export must not reach the NBD writer")
-	assert.Equal(t, []string{"create", "publish", "unpublish"}, provider.calls,
-		"a refused export must not be snapshotted, and must still be unpublished")
-}
-
 // TestImportImage_NoSnapshotWhenNotAsked covers the weights-style import that
 // wants the volume but not an AMI snapshot.
 func TestImportImage_NoSnapshotWhenNotAsked(t *testing.T) {

spinifex/providers/ebs/conformance/conformance.go

@@ -791,10 +791,7 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 			require.NotNil(t, pub)
 			assert.Equal(t, cfg.id("vol-pub-ok"), pub.VolumeID)
 			assert.Equal(t, cfg.node(), pub.NodeID)
-			require.NoError(t, pub.Export.Validate(), "published export is not attachable")
-			if pub.Export.Kind == ebsprovider.ExportKindNBD {
-				assertNBDURI(t, pub.Export.NBD.NBDURI)
-			}
+			assertNBDURI(t, pub.NBDURI)
 		})
 
 		t.Run("idempotent republish to the same node", func(t *testing.T) {

spinifex/providers/ebs/conformance/datapath.go

@@ -167,9 +167,9 @@ func RunDataPathSuite(t *testing.T, newProvider func(t *testing.T) ebsprovider.E
 	provider := newProvider(t)
 	volumeID := cfg.VolumePrefix + "datapath"
 
-	nbdURI := publishForNBD(t, provider, cfg, volumeID, false)
-	assertNBDURI(t, nbdURI)
-	uri := dialURI(t, nbdURI)
+	pub := publishForNBD(t, provider, cfg, volumeID, false)
+	assertNBDURI(t, pub.NBDURI)
+	uri := dialURI(t, pub.NBDURI)
 
 	export := nbdInfo(t, uri)
 	require.Equalf(t, cfg.VolumeBytes, export.Size, "export size does not match the volume that was created")

spinifex/providers/ebs/conformance/nbdclient.go

@@ -177,9 +177,9 @@ func nbdReachable(uri string) bool {
 	return exec.CommandContext(context.Background(), "nbdinfo", "--size", uri).Run() == nil
 }
 
-// publishForNBD creates and publishes a volume and returns its NBD URI,
-// registering cleanup so a run against a real cluster leaves nothing behind.
-func publishForNBD(t *testing.T, provider ebsprovider.EBSProvider, cfg NBDClientConfig, volumeID string, readOnly bool) string {
+// publishForNBD creates and publishes a volume, registering cleanup so a run
+// against a real cluster does not leave volumes behind.
+func publishForNBD(t *testing.T, provider ebsprovider.EBSProvider, cfg NBDClientConfig, volumeID string, readOnly bool) *ebsprovider.PublishedVolume {
 	t.Helper()
 	ctx := context.Background()
 	_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{
@@ -198,17 +198,7 @@ func publishForNBD(t *testing.T, provider ebsprovider.EBSProvider, cfg NBDClient
 	})
 	require.NoError(t, err)
 	require.NotNil(t, pub)
-	return publishedNBDURI(t, pub)
-}
-
-// publishedNBDURI fails the test outright when a provider publishes anything
-// but an nbd export: the NBD-dialling suites have nothing to dial otherwise.
-func publishedNBDURI(t *testing.T, pub *ebsprovider.PublishedVolume) string {
-	t.Helper()
-	require.NoErrorf(t, pub.Export.Validate(), "published export for %s is invalid", pub.VolumeID)
-	require.Equalf(t, ebsprovider.ExportKindNBD, pub.Export.Kind,
-		"published export for %s is %q; this suite drives only nbd exports", pub.VolumeID, pub.Export.Kind)
-	return pub.Export.NBD.NBDURI
+	return pub
 }
 
 // cleanupVolume unpublishes and deletes best-effort: cleanup runs after the
@@ -244,21 +234,21 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
 
 	t.Run("published URI is a standard NBD URI", func(t *testing.T) {
 		provider := newProvider(t)
-		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-uri", false)
-		assertNBDURI(t, nbdURI)
+		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-uri", false)
+		assertNBDURI(t, pub.NBDURI)
 	})
 
 	t.Run("export is connectable by a standard NBD client", func(t *testing.T) {
 		provider := newProvider(t)
-		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-connect", false)
-		export := nbdInfo(t, dialURI(t, nbdURI))
+		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-connect", false)
+		export := nbdInfo(t, dialURI(t, pub.NBDURI))
 		assert.Equal(t, cfg.VolumeBytes, export.Size)
 	})
 
 	t.Run("export advertises usable block size constraints", func(t *testing.T) {
 		provider := newProvider(t)
-		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-blocksize", false)
-		export := nbdInfo(t, dialURI(t, nbdURI))
+		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-blocksize", false)
+		export := nbdInfo(t, dialURI(t, pub.NBDURI))
 
 		require.Positivef(t, export.BlockSizeMinimum,
 			"export advertises no minimum block size, so a client cannot know how to align its requests")
@@ -271,8 +261,8 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
 	// export. Publication is what holds the volume, not any one connection.
 	t.Run("export accepts a client reconnecting after the first disconnects", func(t *testing.T) {
 		provider := newProvider(t)
-		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-reconnect", false)
-		uri := dialURI(t, nbdURI)
+		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-reconnect", false)
+		uri := dialURI(t, pub.NBDURI)
 
 		first := nbdInfo(t, uri)
 
@@ -285,7 +275,7 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
 
 	t.Run("data written by an external client reads back byte for byte", func(t *testing.T) {
 		provider := newProvider(t)
-		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-roundtrip", false)
+		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-roundtrip", false)
 
 		dir := t.TempDir()
 		in := filepath.Join(dir, "in.bin")
@@ -293,7 +283,7 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
 		want := patternBytes(int(cfg.VolumeBytes))
 		require.NoError(t, os.WriteFile(in, want, 0o600))
 
-		uri := dialURI(t, nbdURI)
+		uri := dialURI(t, pub.NBDURI)
 		runNBDCopy(t, in, uri)
 		runNBDCopy(t, uri, out)
 
@@ -311,9 +301,9 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
 	t.Run("sparse extent reporting matches the advertised capability", func(t *testing.T) {
 		provider := newProvider(t)
 		capabilities := capabilitiesOf(t, provider)
-		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-extents", false)
+		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-extents", false)
 
-		holes := slices.ContainsFunc(nbdMap(t, dialURI(t, nbdURI)), func(extent nbdExtent) bool {
+		holes := slices.ContainsFunc(nbdMap(t, dialURI(t, pub.NBDURI)), func(extent nbdExtent) bool {
 			return strings.Contains(extent.Description, "hole")
 		})
 		assert.Equalf(t, capabilities.SparseExtentReporting, holes,
@@ -344,16 +334,16 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
 			return
 		}
 
-		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-readonly", true)
-		export := nbdInfo(t, dialURI(t, nbdURI))
+		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-readonly", true)
+		export := nbdInfo(t, dialURI(t, pub.NBDURI))
 		assert.True(t, export.ReadOnly, "a volume published ReadOnly must set the NBD read-only transmission flag")
 	})
 
 	t.Run("unpublish tears the export down", func(t *testing.T) {
 		provider := newProvider(t)
 		volumeID := cfg.VolumePrefix + "-unpublish"
-		nbdURI := publishForNBD(t, provider, cfg, volumeID, false)
-		uri := dialURI(t, nbdURI)
+		pub := publishForNBD(t, provider, cfg, volumeID, false)
+		uri := dialURI(t, pub.NBDURI)
 		require.True(t, nbdReachable(uri), "export should be reachable before unpublish")
 
 		require.NoError(t, provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{

spinifex/providers/ebs/hostile/verbs.go

@@ -74,24 +74,13 @@ func corruptPublished(published *ebsprovider.PublishedVolume, flavour string) *e
 	corrupted := *published
 	switch flavour {
 	case "handle", "capacity":
-		corrupted.Export = corruptExport(published.Export)
+		corrupted.NBDURI = "nbd+unix:///?socket=/hostile/wrong.sock"
 	case "state":
 		corrupted.NodeID = "hostile-wrong-node"
 	}
 	return &corrupted
 }
 
-// corruptExport points a still-valid export at the wrong place, building new
-// variants so the inner provider's own export is never mutated.
-func corruptExport(export ebsprovider.VolumeExport) ebsprovider.VolumeExport {
-	if export.Kind == ebsprovider.ExportKindBlock && export.Block != nil {
-		block := *export.Block
-		block.DevicePath = "/dev/hostile-wrong"
-		return ebsprovider.VolumeExport{Kind: ebsprovider.ExportKindBlock, Block: &block}
-	}
-	return ebsprovider.NewNBDExport("nbd+unix:///?socket=/hostile/wrong.sock")
-}
-
 func (p *Provider) GetCapabilities(ctx context.Context, req ebsprovider.GetCapabilitiesRequest) (*ebsprovider.GetCapabilitiesResponse, error) {
 	injection := p.draw(verbCapabilities, "")
 	if stop, err := p.apply(ctx, injection); stop || err != nil {

spinifex/providers/ebs/memory.go

@@ -375,7 +375,7 @@ func (m *MemoryProvider) PublishVolume(_ context.Context, req PublishVolumeReque
 	volume.published = &PublishedVolume{
 		VolumeID: req.VolumeID,
 		NodeID:   req.NodeID,
-		Export:   NewNBDExport("nbd+unix:///?socket=/memory/" + req.VolumeID + ".sock"),
+		NBDURI:   "nbd+unix:///?socket=/memory/" + req.VolumeID + ".sock",
 	}
 	volume.volume.State = VolumeStateInUse
 	return clonePublished(volume.published), nil
@@ -421,13 +421,5 @@ func clonePublished(published *PublishedVolume) *PublishedVolume {
 		return nil
 	}
 	clone := *published
-	if published.Export.NBD != nil {
-		nbd := *published.Export.NBD
-		clone.Export.NBD = &nbd
-	}
-	if published.Export.Block != nil {
-		block := *published.Export.Block
-		clone.Export.Block = &block
-	}
 	return &clone
 }

spinifex/providers/ebs/memory_test.go

@@ -40,7 +40,7 @@ func TestMemoryProviderLifecycleAndIdempotency(t *testing.T) {
 		Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle, NodeID: "node-1",
 	})
 	require.NoError(t, err)
-	assert.Equal(t, NewNBDExport("nbd+unix:///?socket=/memory/vol-1.sock"), published.Export)
+	assert.Equal(t, "nbd+unix:///?socket=/memory/vol-1.sock", published.NBDURI)
 
 	repeatedPublish, err := provider.PublishVolume(ctx, PublishVolumeRequest{
 		Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle, NodeID: "node-1",

spinifex/providers/ebs/nats.go

@@ -452,9 +452,6 @@ func (p *NATSProvider) PublishVolume(ctx context.Context, req PublishVolumeReque
 	if response.Published == nil {
 		return nil, fmt.Errorf("%s returned no published volume", subject)
 	}
-	if err := response.Published.Export.Validate(); err != nil {
-		return nil, fmt.Errorf("%s returned an invalid export: %w", subject, err)
-	}
 	return response.Published, nil
 }
 

spinifex/providers/ebs/nats_test.go

@@ -432,50 +432,6 @@ func TestNATSProviderRejectsEmptySuccessResponse(t *testing.T) {
 	})
 }
 
-// TestNATSProviderRejectsInvalidExport covers the client-side export check: a
-// reply the server marks successful but whose export cannot be attached must
-// fail here, not later when QEMU is handed it.
-func TestNATSProviderRejectsInvalidExport(t *testing.T) {
-	_, conn := testutil.StartTestNATS(t)
-	provider := NewNATSProvider(conn, time.Second)
-
-	tests := []struct {
-		node    string
-		export  VolumeExport
-		wantErr string
-	}{
-		{node: "node-nokind", export: VolumeExport{NBD: &NBDExport{NBDURI: "nbd://10.0.0.1:10809"}}, wantErr: `unknown export kind ""`},
-		{node: "node-noblock", export: VolumeExport{Kind: ExportKindBlock}, wantErr: "block export has no block details"},
-		{node: "node-emptyuri", export: VolumeExport{Kind: ExportKindNBD, NBD: &NBDExport{}}, wantErr: "nbd export has an empty NBD URI"},
-	}
-	for _, tt := range tests {
-		t.Run(tt.node, func(t *testing.T) {
-			subject, err := PublishSubject(tt.node)
-			require.NoError(t, err)
-			respondWith(t, conn, subject, PublishVolumeResponse{
-				Versioned: NewVersioned(),
-				Published: &PublishedVolume{VolumeID: "vol-1", NodeID: tt.node, Export: tt.export},
-			})
-			published, err := provider.PublishVolume(t.Context(), PublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: tt.node})
-			require.EqualError(t, err, subject+" returned an invalid export: "+tt.wantErr)
-			assert.Nil(t, published)
-		})
-	}
-
-	t.Run("valid block export passes", func(t *testing.T) {
-		subject, err := PublishSubject("node-block")
-		require.NoError(t, err)
-		export := VolumeExport{Kind: ExportKindBlock, Block: &BlockExport{DevicePath: "/dev/sdb", SizeBytes: 10 << 30}}
-		respondWith(t, conn, subject, PublishVolumeResponse{
-			Versioned: NewVersioned(),
-			Published: &PublishedVolume{VolumeID: "vol-1", NodeID: "node-block", Export: export},
-		})
-		published, err := provider.PublishVolume(t.Context(), PublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node-block"})
-		require.NoError(t, err)
-		assert.Equal(t, export, published.Export)
-	})
-}
-
 // TestNATSProviderPropagatesProviderErrors covers responseError's populated
 // branch on the enumeration and unpublish verbs, whose errors otherwise reach
 // the caller as an empty page or a silent success.

spinifex/providers/ebs/natsserve/handlers_test.go

@@ -276,7 +276,7 @@ func TestServedProviderRoundTrip(t *testing.T) {
 		Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip", NodeID: "node-1",
 	})
 	require.NoError(t, err)
-	assert.Equal(t, ebsprovider.NewNBDExport("nbd+unix:///?socket=/memory/vol-roundtrip.sock"), published.Export)
+	assert.Equal(t, "nbd+unix:///?socket=/memory/vol-roundtrip.sock", published.NBDURI)
 
 	inUse, err := client.GetVolume(ctx, ebsprovider.GetVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip"})
 	require.NoError(t, err)
@@ -432,42 +432,6 @@ func (p *countingProvider) PublishVolume(ctx context.Context, req ebsprovider.Pu
 	return p.EBSProvider.PublishVolume(ctx, req)
 }
 
-// exportProvider publishes a fixed export, so the server's own export check
-// can be driven with a provider that returns one it must not relay.
-type exportProvider struct {
-	ebsprovider.EBSProvider
-
-	export ebsprovider.VolumeExport
-}
-
-func (p *exportProvider) PublishVolume(_ context.Context, req ebsprovider.PublishVolumeRequest) (*ebsprovider.PublishedVolume, error) {
-	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: p.export}, nil
-}
-
-// TestPublishVolumeRejectsInvalidExport covers the server-side export check:
-// an unattachable export is the provider's fault and goes out as internal.
-func TestPublishVolumeRejectsInvalidExport(t *testing.T) {
-	provider := &exportProvider{
-		EBSProvider: ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}),
-		export:      ebsprovider.VolumeExport{Kind: ebsprovider.ExportKindBlock, Block: &ebsprovider.BlockExport{DevicePath: "/dev/sdb"}},
-	}
-	conn := serveProvider(t, provider, natsserve.Options{NoQueueGroup: true})
-	subject, err := ebsprovider.PublishSubject("node-1")
-	require.NoError(t, err)
-
-	payload, err := json.Marshal(ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", NodeID: "node-1"})
-	require.NoError(t, err)
-	msg, err := conn.Request(subject, payload, requestTimeout)
-	require.NoError(t, err)
-
-	var resp ebsprovider.PublishVolumeResponse
-	require.NoError(t, json.Unmarshal(msg.Data, &resp))
-	assert.Nil(t, resp.Published, "an invalid export must not be relayed")
-	require.NotNil(t, resp.Error)
-	assert.Equal(t, ebsprovider.ErrorCodeInternal, resp.Error.Code)
-	assert.Equal(t, "provider returned an invalid export for vol-1: block export has a zero size", resp.Error.Message)
-}
-
 // handlerSettle is how long a handler count is watched after it reaches the
 // expected value, to catch a delivery that should not have happened arriving
 // late. Generous, because a false pass here is a silently broken queue group.
@@ -579,7 +543,7 @@ func TestServeRejectsUnroutableNodeID(t *testing.T) {
 	assert.Contains(t, err.Error(), "build publish subject")
 	assert.Nil(t, stop)
 
-	_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(fmt.Sprintf(`{"schema_version":%d}`, ebsprovider.SchemaVersion)), 200*time.Millisecond)
+	_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(`{"schema_version":1}`), 200*time.Millisecond)
 	require.ErrorIs(t, err, nats.ErrNoResponders, "a failed Serve must leave no subscription behind")
 }
 
@@ -607,6 +571,6 @@ func TestServeStopIsIdempotent(t *testing.T) {
 	stop()
 	require.NoError(t, conn.Flush())
 
-	_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(fmt.Sprintf(`{"schema_version":%d}`, ebsprovider.SchemaVersion)), 200*time.Millisecond)
+	_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(`{"schema_version":1}`), 200*time.Millisecond)
 	require.ErrorIs(t, err, nats.ErrNoResponders, "stop must unsubscribe every subject Serve registered")
 }

spinifex/providers/ebs/natsserve/server.go

@@ -501,16 +501,6 @@ func handlePublishVolume(provider ebsprovider.EBSProvider) msgHandler {
 			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
 			return
 		}
-		if pub == nil {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("provider returned no published volume for %s", req.VolumeID)})
-			return
-		}
-		// An unattachable export is the provider's fault, so it is reported as
-		// internal rather than relayed for the client to reject.
-		if err := pub.Export.Validate(); err != nil {
-			respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("provider returned an invalid export for %s: %v", req.VolumeID, err)})
-			return
-		}
 		respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Published: pub})
 	}
 }

spinifex/providers/ebs/provider.go

@@ -6,16 +6,14 @@ package ebsprovider
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
-	"path/filepath"
 	"time"
 )
 
 // SchemaVersion is the only wire-contract version understood by this build.
 // Every request and response carries it so version skew fails at the boundary
 // instead of silently decoding a changed payload.
-const SchemaVersion uint16 = 2
+const SchemaVersion uint16 = 1
 
 // Versioned is embedded in every request and response sent over NATS.
 type Versioned struct {
@@ -445,80 +443,12 @@ type PublishVolumeRequest struct {
 	ReadOnly bool   `json:"read_only,omitempty"`
 }
 
-// ExportKind names how a published volume is reached on its node, and so
-// which VolumeExport variant carries the details.
-type ExportKind string
-
-const (
-	ExportKindNBD   ExportKind = "nbd"
-	ExportKindBlock ExportKind = "block"
-)
-
-// VolumeExport is how a published volume is reached; exactly one of NBD and
-// Block is set, and it is the one Kind names.
-type VolumeExport struct {
-	Kind  ExportKind   `json:"kind"`
-	NBD   *NBDExport   `json:"nbd,omitempty"`
-	Block *BlockExport `json:"block,omitempty"`
-}
-
-// NBDExport is a volume served over NBD: the URI on the node that QEMU
-// connects to.
-type NBDExport struct {
-	NBDURI string `json:"nbd_uri"`
-}
-
-// BlockExport is a volume exposed as a host block device. SizeBytes is the
-// size the guest sees, which may be smaller than the device itself.
-type BlockExport struct {
-	DevicePath string `json:"device_path"`
-	SizeBytes  uint64 `json:"size_bytes"`
-}
-
-// Validate rejects an export a consumer could not attach: an unknown kind, a
-// missing or extra variant, or a variant with an unusable field.
-func (e VolumeExport) Validate() error {
-	switch e.Kind {
-	case ExportKindNBD:
-		if e.NBD == nil {
-			return errors.New("nbd export has no nbd details")
-		}
-		if e.Block != nil {
-			return errors.New("nbd export also sets block details")
-		}
-		if e.NBD.NBDURI == "" {
-			return errors.New("nbd export has an empty NBD URI")
-		}
-	case ExportKindBlock:
-		if e.Block == nil {
-			return errors.New("block export has no block details")
-		}
-		if e.NBD != nil {
-			return errors.New("block export also sets nbd details")
-		}
-		if !filepath.IsAbs(e.Block.DevicePath) {
-			return fmt.Errorf("block export device path %q is not absolute", e.Block.DevicePath)
-		}
-		if e.Block.SizeBytes == 0 {
-			return errors.New("block export has a zero size")
-		}
-	default:
-		return fmt.Errorf("unknown export kind %q", e.Kind)
-	}
-	return nil
-}
-
-// PublishedVolume is where a published volume is served: the export on NodeID
-// that QEMU attaches.
+// PublishedVolume is where a published volume is served: the NBD URI on NodeID
+// that QEMU connects to.
 type PublishedVolume struct {
-	VolumeID string       `json:"volume_id"`
-	NodeID   string       `json:"node_id"`
-	Export   VolumeExport `json:"export"`
-}
-
-// NewNBDExport returns an nbd export for uri.
-func NewNBDExport(uri string) VolumeExport {
-	return VolumeExport{Kind: ExportKindNBD, NBD: &NBDExport{NBDURI: uri}}
+	VolumeID string `json:"volume_id"`
+	NodeID   string `json:"node_id"`
+	NBDURI   string `json:"nbd_uri"`
 }
 
 // PublishVolumeResponse is the reply to PublishVolumeRequest: the mount, or

spinifex/providers/ebs/provider_test.go

@@ -1,63 +0,0 @@
-package ebsprovider
-
-import (
-	"encoding/json"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestVolumeExportValidate(t *testing.T) {
-	nbd := &NBDExport{NBDURI: "nbd+unix:///?socket=/run/vol-1.sock"}
-	block := &BlockExport{DevicePath: "/dev/sdb", SizeBytes: 10 << 30}
-
-	tests := []struct {
-		name    string
-		export  VolumeExport
-		wantErr string
-	}{
-		{name: "valid nbd", export: VolumeExport{Kind: ExportKindNBD, NBD: nbd}},
-		{name: "valid block", export: VolumeExport{Kind: ExportKindBlock, Block: block}},
-		{name: "empty kind", export: VolumeExport{NBD: nbd}, wantErr: `unknown export kind ""`},
-		{name: "unknown kind", export: VolumeExport{Kind: "iscsi", NBD: nbd}, wantErr: `unknown export kind "iscsi"`},
-		{name: "nbd kind without nbd variant", export: VolumeExport{Kind: ExportKindNBD}, wantErr: "nbd export has no nbd details"},
-		{name: "nbd kind carrying only block variant", export: VolumeExport{Kind: ExportKindNBD, Block: block}, wantErr: "nbd export has no nbd details"},
-		{name: "nbd kind with block variant", export: VolumeExport{Kind: ExportKindNBD, NBD: nbd, Block: block}, wantErr: "nbd export also sets block details"},
-		{name: "empty nbd uri", export: VolumeExport{Kind: ExportKindNBD, NBD: &NBDExport{}}, wantErr: "nbd export has an empty NBD URI"},
-		{name: "block kind without block variant", export: VolumeExport{Kind: ExportKindBlock}, wantErr: "block export has no block details"},
-		{name: "block kind with nbd variant", export: VolumeExport{Kind: ExportKindBlock, NBD: nbd, Block: block}, wantErr: "block export also sets nbd details"},
-		{name: "relative device path", export: VolumeExport{Kind: ExportKindBlock, Block: &BlockExport{DevicePath: "dev/sdb", SizeBytes: 1}}, wantErr: `block export device path "dev/sdb" is not absolute`},
-		{name: "empty device path", export: VolumeExport{Kind: ExportKindBlock, Block: &BlockExport{SizeBytes: 1}}, wantErr: `block export device path "" is not absolute`},
-		{name: "zero size", export: VolumeExport{Kind: ExportKindBlock, Block: &BlockExport{DevicePath: "/dev/sdb"}}, wantErr: "block export has a zero size"},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			err := tt.export.Validate()
-			if tt.wantErr == "" {
-				require.NoError(t, err)
-				return
-			}
-			require.EqualError(t, err, tt.wantErr)
-		})
-	}
-}
-
-// TestPublishedVolumeWireShape pins the JSON the contract carries: a renamed
-// tag would decode as an empty variant on a peer at the same schema version.
-func TestPublishedVolumeWireShape(t *testing.T) {
-	wires := map[string]string{
-		"nbd":   `{"volume_id":"vol-1","node_id":"node-1","export":{"kind":"nbd","nbd":{"nbd_uri":"nbd://10.0.0.1:10809"}}}`,
-		"block": `{"volume_id":"vol-1","node_id":"node-1","export":{"kind":"block","block":{"device_path":"/dev/sdb","size_bytes":10737418240}}}`,
-	}
-	for name, wire := range wires {
-		t.Run(name, func(t *testing.T) {
-			var published PublishedVolume
-			require.NoError(t, json.Unmarshal([]byte(wire), &published))
-			require.NoError(t, published.Export.Validate())
-			encoded, err := json.Marshal(published)
-			require.NoError(t, err)
-			assert.JSONEq(t, wire, string(encoded))
-		})
-	}
-}

spinifex/providers/ebs/viperblock/provider_handlers.go

@@ -1765,7 +1765,7 @@ func handlePublishVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *n
 			Published: &ebsprovider.PublishedVolume{
 				VolumeID: req.VolumeID,
 				NodeID:   cfg.NodeName,
-				Export:   ebsprovider.NewNBDExport(mv.NBDURI),
+				NBDURI:   mv.NBDURI,
 			},
 		})
 		return
@@ -1795,7 +1795,7 @@ func handlePublishVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *n
 		Published: &ebsprovider.PublishedVolume{
 			VolumeID: req.VolumeID,
 			NodeID:   cfg.NodeName,
-			Export:   ebsprovider.NewNBDExport(mountResp.URI),
+			NBDURI:   mountResp.URI,
 		},
 	})
 }

spinifex/providers/ebs/viperblock/provider_handlers_test.go

@@ -837,7 +837,7 @@ func TestProviderHandlers_PublishVolume_IdempotentRepublish(t *testing.T) {
 	require.NotNil(t, resp.Published)
 	assert.Equal(t, volumeName, resp.Published.VolumeID)
 	assert.Equal(t, cfg.NodeName, resp.Published.NodeID)
-	assert.Equal(t, ebsprovider.NewNBDExport(existingURI), resp.Published.Export)
+	assert.Equal(t, existingURI, resp.Published.NBDURI)
 
 	cfg.mu.Lock()
 	count := len(cfg.MountedVolumes)

spinifex/runtime/compute/vm/devices.go

@@ -355,35 +355,3 @@ func VolumeBlockdev(nodeName string, server NBDServerOpts) Blockdev {
 	opts = append(opts, "export=", fmt.Sprintf("reconnect-delay=%d", NBDReconnectDelaySeconds))
 	return Blockdev{Value: strings.Join(opts, ",")}
 }
-
-// HostDeviceOpts is a host block device export. QEMU opens Path through the
-// host_device protocol driver with O_DIRECT and native AIO, and a raw format
-// node over it exposes only the first SizeBytes to the guest.
-type HostDeviceOpts struct {
-	Path      string
-	SizeBytes uint64
-}
-
-// QMPArgs renders the blockdev-add arguments for the raw node and its nested
-// host_device file node.
-func (o HostDeviceOpts) QMPArgs(nodeName string) map[string]any {
-	return map[string]any{
-		"node-name": nodeName,
-		"driver":    "raw",
-		"size":      o.SizeBytes,
-		"file": map[string]any{
-			"driver":   "host_device",
-			"filename": o.Path,
-			"aio":      "native",
-			"cache":    map[string]any{"direct": true},
-		},
-	}
-}
-
-// VolumeHostDeviceBlockdev renders the -blockdev equivalent of QMPArgs for a
-// data volume backed by a host block device.
-func VolumeHostDeviceBlockdev(nodeName string, dev HostDeviceOpts) Blockdev {
-	return Blockdev{Value: fmt.Sprintf(
-		"driver=raw,node-name=%s,size=%d,file.driver=host_device,file.filename=%s,file.aio=native,file.cache.direct=on",
-		nodeName, dev.SizeBytes, dev.Path)}
-}

spinifex/runtime/compute/vm/ebs_requests.go

@@ -1,10 +1,6 @@
 package vm
 
 import (
-	"fmt"
-	"os"
-	"path/filepath"
-	"strings"
 	"sync"
 
 	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
@@ -24,45 +20,3 @@ type EBSRequests struct {
 // VM's persisted attachment list. The alias keeps the VM state tied to the
 // deployed wire shape without making the old catch-all types package an owner.
 type EBSRequest = viperblocklegacyv1.EBSRequest
-
-// volumeExportKind is how a volume's backing store reaches QEMU.
-type volumeExportKind int
-
-const (
-	volumeExportInvalid volumeExportKind = iota
-	volumeExportNBD
-	volumeExportBlock
-)
-
-// classifyVolumeExport reports which export a request carries. A block device
-// must be absolute, comma-free (QEMU key=value options would split it) and
-// sized; both exports set, neither, or a stray size is invalid.
-func classifyVolumeExport(req EBSRequest) (volumeExportKind, error) {
-	switch {
-	case req.NBDURI != "" && req.BlockDevice != "":
-		return volumeExportInvalid, fmt.Errorf("volume %s has both NBDURI and BlockDevice set", req.Name)
-	case req.BlockSizeBytes != 0 && req.BlockDevice == "":
-		return volumeExportInvalid, fmt.Errorf("volume %s has BlockSizeBytes set without BlockDevice", req.Name)
-	case req.NBDURI != "":
-		return volumeExportNBD, nil
-	case req.BlockDevice != "":
-		if !filepath.IsAbs(req.BlockDevice) {
-			return volumeExportInvalid, fmt.Errorf("volume %s BlockDevice %q is not an absolute path", req.Name, req.BlockDevice)
-		}
-		if strings.Contains(req.BlockDevice, ",") {
-			return volumeExportInvalid, fmt.Errorf("volume %s BlockDevice %q contains a comma", req.Name, req.BlockDevice)
-		}
-		if req.BlockSizeBytes == 0 {
-			return volumeExportInvalid, fmt.Errorf("volume %s BlockDevice %q has zero BlockSizeBytes", req.Name, req.BlockDevice)
-		}
-		return volumeExportBlock, nil
-	default:
-		return volumeExportInvalid, fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", req.Name)
-	}
-}
-
-// checkBlockDevice reports whether a block export's device node is present.
-func checkBlockDevice(path string) error {
-	_, err := os.Stat(path)
-	return err
-}

spinifex/runtime/compute/vm/lifecycle.go

@@ -607,18 +607,11 @@ func (m *Manager) startQEMU(instance *VM) error {
 		refreshTelemetryMeta(instance)
 	}
 
-	// An NBD export is waited on until it accepts connections; a block export
-	// is local to the host and only has to exist. An invalid request has no
-	// endpoint to probe (buildDrives rejects it on the command-line path).
 	instance.EBSRequests.Mu.Lock()
 	nbdEndpoints := make([]struct{ name, uri string }, 0, len(instance.EBSRequests.Requests))
-	var blockDevices []struct{ name, path string }
 	for _, req := range instance.EBSRequests.Requests {
-		switch kind, _ := classifyVolumeExport(req); kind {
-		case volumeExportNBD:
+		if req.NBDURI != "" {
 			nbdEndpoints = append(nbdEndpoints, struct{ name, uri string }{req.Name, req.NBDURI})
-		case volumeExportBlock:
-			blockDevices = append(blockDevices, struct{ name, path string }{req.Name, req.BlockDevice})
 		}
 	}
 	instance.EBSRequests.Mu.Unlock()
@@ -627,11 +620,6 @@ func (m *Manager) startQEMU(instance *VM) error {
 			return fmt.Errorf("nbd endpoint not ready for %s: %w", ep.name, err)
 		}
 	}
-	for _, dev := range blockDevices {
-		if err := checkBlockDevice(dev.path); err != nil {
-			return fmt.Errorf("block device not present for %s: %w", dev.name, err)
-		}
-	}
 
 	processChan := make(chan int, 1)
 	exitChan := make(chan int, 1)
@@ -912,11 +900,11 @@ func qmpGreetingTimeout(v *VM) time.Duration {
 
 // hasNetworkBackedDrive reports whether any of the guest's drives is served
 // over NBD, and so has to be opened across the object store before QEMU's
-// monitor answers. A host block device opens locally and does not count.
+// monitor answers. Every production guest boots from one; a VM without is a
+// unit test.
 func hasNetworkBackedDrive(v *VM) bool {
 	return slices.ContainsFunc(v.EBSRequests.Requests, func(req EBSRequest) bool {
-		kind, _ := classifyVolumeExport(req)
-		return kind == volumeExportNBD
+		return req.NBDURI != ""
 	})
 }
 
@@ -1457,9 +1445,8 @@ type driveConfig struct {
 }
 
 // buildDrives converts EBS requests into QEMU drive/iothread/device/blockdev
-// configs. Returns an error if any volume has no valid export (see
-// classifyVolumeExport), or if the EBS hot-plug port pool is exhausted while
-// allocating one for a data volume.
+// configs. Returns an error if any volume is missing its NBDURI, or if the
+// EBS hot-plug port pool is exhausted while allocating one for a data volume.
 //
 // Boot volumes keep the existing anonymous if=none -drive shape exactly as
 // before — they are not detachable (see ErrVolumeNotDetachable), so the
@@ -1484,17 +1471,11 @@ func buildDrives(requests []EBSRequest, cpuCount int, machineType string) (drive
 
 	for i := range requests {
 		v := &requests[i]
-		kind, err := classifyVolumeExport(*v)
-		if err != nil {
-			return driveConfig{}, err
+		if v.NBDURI == "" {
+			return driveConfig{}, fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", v.Name)
 		}
 
 		switch {
-		case kind == volumeExportBlock && (v.Boot || v.EFI):
-			// Boot and EFI keep the legacy -drive shape, which has no
-			// host_device form here; only data volumes take a block export.
-			return driveConfig{}, fmt.Errorf("volume %s: block device export is only supported for data volumes", v.Name)
-
 		case v.Boot:
 			drive := Drive{
 				File:   v.NBDURI,
@@ -1548,39 +1529,29 @@ func buildDrives(requests []EBSRequest, cpuCount int, machineType string) (drive
 				v.HotplugPort = port
 			}
 
+			serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(v.NBDURI)
+			if err != nil {
+				return driveConfig{}, fmt.Errorf("parse NBDURI for volume %s: %w", v.Name, err)
+			}
+
 			nodeName := VolumeNodeName(v.Name)
 			iothreadID := VolumeIOThreadID(v.Name)
 			bus := HotplugEBSBus(v.HotplugPort)
 
-			var blockdev Blockdev
-			if kind == volumeExportBlock {
-				blockdev = VolumeHostDeviceBlockdev(nodeName, HostDeviceOpts{Path: v.BlockDevice, SizeBytes: v.BlockSizeBytes})
-			} else {
-				serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(v.NBDURI)
-				if err != nil {
-					return driveConfig{}, fmt.Errorf("parse NBDURI for volume %s: %w", v.Name, err)
-				}
-				blockdev = VolumeBlockdev(nodeName, NBDServerOpts{
-					Type: serverType,
-					Path: socketPath,
-					Host: nbdHost,
-					Port: nbdPort,
-				})
-			}
-
 			cfg.IOThreads = append(cfg.IOThreads, IOThread{ID: iothreadID})
-			cfg.Blockdevs = append(cfg.Blockdevs, blockdev)
+			cfg.Blockdevs = append(cfg.Blockdevs, VolumeBlockdev(nodeName, NBDServerOpts{
+				Type: serverType,
+				Path: socketPath,
+				Host: nbdHost,
+				Port: nbdPort,
+			}))
 			cfg.Devices = append(cfg.Devices, VolumeBlkDevice(v.Name, nodeName, iothreadID, bus))
 			// Deliberately no entry in cfg.Drives: the bare anonymous -drive
 			// this case used to emit (no id=, no node-name=) is exactly what
 			// left the volume undetachable.
 		}
 
-		if kind == volumeExportBlock {
-			slog.Info("Using block device for drive", "volume", v.Name, "device", v.BlockDevice, "size_bytes", v.BlockSizeBytes)
-		} else {
-			slog.Info("Using NBD URI for drive", "volume", v.Name, "uri", v.NBDURI)
-		}
+		slog.Info("Using NBD URI for drive", "volume", v.Name, "uri", v.NBDURI)
 	}
 
 	return cfg, nil

spinifex/runtime/compute/vm/lifecycle_test.go

@@ -273,83 +273,6 @@ func TestBuildDrives(t *testing.T) {
 			},
 			wantHotplugPorts: []int{0, 0, 3},
 		},
-		{
-			name: "block data volume emits a raw node over host_device with the same device wiring",
-			requests: []EBSRequest{
-				{Name: "vol-blk", BlockDevice: "/dev/sdx", BlockSizeBytes: 10 << 30, HotplugPort: 2},
-			},
-			cpuCount:      2,
-			wantIOThreads: []IOThread{{ID: "ioth-vol-blk"}},
-			wantBlockdevs: []Blockdev{
-				{Value: "driver=raw,node-name=nbd-vol-blk,size=10737418240,file.driver=host_device,file.filename=/dev/sdx,file.aio=native,file.cache.direct=on"},
-			},
-			wantDevices: []Device{
-				{Value: "virtio-blk-pci,id=vdisk-vol-blk,drive=nbd-vol-blk,iothread=ioth-vol-blk,serial=volblk,bus=hotplug-ebs2,werror=stop,rerror=stop"},
-			},
-			wantHotplugPorts: []int{2},
-		},
-		{
-			name: "NBD boot volume alongside a block data volume",
-			requests: []EBSRequest{
-				{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
-				{Name: "vol-blk", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20},
-			},
-			cpuCount: 2,
-			wantDrives: []Drive{
-				{File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "stop", Rerror: "stop", ReconnectDelay: NBDReconnectDelaySeconds},
-			},
-			wantIOThreads: []IOThread{{ID: "ioth-os"}, {ID: "ioth-vol-blk"}},
-			wantBlockdevs: []Blockdev{
-				{Value: "driver=raw,node-name=nbd-vol-blk,size=1048576,file.driver=host_device,file.filename=/dev/sdx,file.aio=native,file.cache.direct=on"},
-			},
-			wantDevices: []Device{
-				{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1"},
-				{Value: "virtio-blk-pci,id=vdisk-vol-blk,drive=nbd-vol-blk,iothread=ioth-vol-blk,serial=volblk,bus=hotplug-ebs1,werror=stop,rerror=stop"},
-			},
-			wantHotplugPorts: []int{0, 1},
-		},
-		{
-			name:     "block boot volume is rejected",
-			requests: []EBSRequest{{Name: "vol-boot", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20, Boot: true}},
-			cpuCount: 2,
-			wantErr:  "volume vol-boot: block device export is only supported for data volumes",
-		},
-		{
-			name:     "block EFI volume is rejected",
-			requests: []EBSRequest{{Name: "vol-efi", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20, EFI: true}},
-			cpuCount: 2,
-			wantErr:  "volume vol-efi: block device export is only supported for data volumes",
-		},
-		{
-			name:     "both NBDURI and BlockDevice set is rejected",
-			requests: []EBSRequest{{Name: "vol-both", NBDURI: "nbd:unix:/tmp/x.sock", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20}},
-			cpuCount: 2,
-			wantErr:  "volume vol-both has both NBDURI and BlockDevice set",
-		},
-		{
-			name:     "NBDURI with a stray BlockSizeBytes is rejected",
-			requests: []EBSRequest{{Name: "vol-stray", NBDURI: "nbd:unix:/tmp/x.sock", BlockSizeBytes: 1 << 20}},
-			cpuCount: 2,
-			wantErr:  "volume vol-stray has BlockSizeBytes set without BlockDevice",
-		},
-		{
-			name:     "block device with zero size is rejected",
-			requests: []EBSRequest{{Name: "vol-nosize", BlockDevice: "/dev/sdx"}},
-			cpuCount: 2,
-			wantErr:  `volume vol-nosize BlockDevice "/dev/sdx" has zero BlockSizeBytes`,
-		},
-		{
-			name:     "relative block device path is rejected",
-			requests: []EBSRequest{{Name: "vol-rel", BlockDevice: "dev/sdx", BlockSizeBytes: 1 << 20}},
-			cpuCount: 2,
-			wantErr:  `volume vol-rel BlockDevice "dev/sdx" is not an absolute path`,
-		},
-		{
-			name:     "block device path with a comma is rejected",
-			requests: []EBSRequest{{Name: "vol-comma", BlockDevice: "/dev/sdx,file.locking=off", BlockSizeBytes: 1 << 20}},
-			cpuCount: 2,
-			wantErr:  `volume vol-comma BlockDevice "/dev/sdx,file.locking=off" contains a comma`,
-		},
 		{
 			name:     "empty requests",
 			requests: []EBSRequest{},
@@ -1085,28 +1008,6 @@ func TestStartQEMU_DirectBoot_InstanceTypeNotFound(t *testing.T) {
 	assert.Empty(t, plumber.setupCalls)
 }
 
-// TestStartQEMU_BlockDeviceMissing verifies a block volume's launch readiness
-// is the device's presence, not an NBD connect.
-func TestStartQEMU_BlockDeviceMissing(t *testing.T) {
-	m := directBootManager(t, &fakeNetworkPlumber{})
-	missing := filepath.Join(t.TempDir(), "missing-dev")
-
-	instance := &VM{
-		ID:           "i-blk-missing",
-		InstanceType: "t3.nano",
-		DirectBoot:   true,
-	}
-	instance.EBSRequests.Requests = []EBSRequest{
-		{Name: "vol-blk", BlockDevice: missing, BlockSizeBytes: 1 << 20},
-	}
-
-	err := m.startQEMU(instance)
-
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "block device not present for vol-blk")
-	assert.ErrorIs(t, err, os.ErrNotExist)
-}
-
 // scriptedNetworkPlumber returns errors from a pre-loaded slice, one per call.
 type scriptedNetworkPlumber struct {
 	calls []TapSpec
@@ -1370,12 +1271,6 @@ func TestQMPGreetingTimeout(t *testing.T) {
 	assert.Greater(t, qmpNBDGreetingTimeout, 35*time.Second,
 		"the deadline must exceed the 35s cold open measured with one node of three away")
 
-	block := &VM{EBSRequests: EBSRequests{Requests: []EBSRequest{
-		{Name: "vol-data", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20},
-	}}}
-	assert.Equal(t, qmp.DefaultGreetingTimeout, qmpGreetingTimeout(block),
-		"a host block device opens locally, so it keeps the plain default")
-
 	// A small GPU guest (<1 GiB scaling contribution) sits at the floor.
 	smallGPU := &VM{
 		GPUAttachments: []gpu.GPUAttachment{{PCIAddress: "0000:5e:00.0"}},

spinifex/runtime/compute/vm/manager_deps_test.go

@@ -142,10 +142,6 @@ type fakeVolumeMounter struct {
 	unmountErr               error
 	unmountOneErr            error
 	mountOneURI              string
-	// mountOneBlockDevice and mountOneBlockSize, when set, are written onto
-	// the request by MountOne as a block device export.
-	mountOneBlockDevice string
-	mountOneBlockSize   uint64
 	// onMount fires synchronously inside Mount before the configured
 	// mountErr is returned. Used by lifecycle tests to simulate a
 	// concurrent terminate flipping VM.Status while Mount is in flight.
@@ -185,7 +181,6 @@ func (f *fakeVolumeMounter) MountOne(_ context.Context, _ string, req *EBSReques
 	f.mountedOne = append(f.mountedOne, req.Name)
 	mountOneErr := f.mountOneErr
 	uri := f.mountOneURI
-	blockDevice, blockSize := f.mountOneBlockDevice, f.mountOneBlockSize
 	f.mu.Unlock()
 	if mountOneErr != nil {
 		return mountOneErr
@@ -193,10 +188,6 @@ func (f *fakeVolumeMounter) MountOne(_ context.Context, _ string, req *EBSReques
 	if uri != "" {
 		req.NBDURI = uri
 	}
-	if blockDevice != "" {
-		req.BlockDevice = blockDevice
-		req.BlockSizeBytes = blockSize
-	}
 	return nil
 }
 

spinifex/runtime/compute/vm/restore.go

@@ -523,22 +523,14 @@ func isInstanceProcessRunning(instance *VM) bool {
 }
 
 // AreVolumeSocketsValid dials each NBD Unix socket to confirm viperblock is
-// still listening, and checks each block export's device still exists. TCP,
-// unparseable and invalid exports are treated as valid.
+// still listening. A stat-only check is insufficient since viperblock may restart
+// leaving stale socket files. TCP and unparseable URIs are treated as valid.
 func AreVolumeSocketsValid(instance *VM) bool {
 	instance.EBSRequests.Mu.Lock()
 	defer instance.EBSRequests.Mu.Unlock()
 
 	for _, req := range instance.EBSRequests.Requests {
-		kind, _ := classifyVolumeExport(req)
-		if kind == volumeExportBlock {
-			if err := checkBlockDevice(req.BlockDevice); err != nil {
-				slog.Debug("Block device missing", "volume", req.Name, "device", req.BlockDevice, "err", err)
-				return false
-			}
-			continue
-		}
-		if kind != volumeExportNBD {
+		if req.NBDURI == "" {
 			continue
 		}
 		serverType, sockPath, _, _, err := nbd.ParseNBDURI(req.NBDURI)

spinifex/runtime/compute/vm/restore_test.go

@@ -82,22 +82,6 @@ func TestAreVolumeSocketsValid_OneOfManyMissing(t *testing.T) {
 		"a single dead socket is enough to invalidate the whole instance")
 }
 
-// TestAreVolumeSocketsValid_BlockDevice checks a block export is judged by its
-// device node's presence, with no NBD probe.
-func TestAreVolumeSocketsValid_BlockDevice(t *testing.T) {
-	present := filepath.Join(t.TempDir(), "dev-present")
-	require.NoError(t, os.WriteFile(present, nil, 0o600))
-	missing := filepath.Join(t.TempDir(), "dev-missing")
-
-	blockVM := func(path string) *VM {
-		v := &VM{ID: "i-blk", Status: StateRunning}
-		v.EBSRequests.Requests = []EBSRequest{{Name: "vol-blk", BlockDevice: path, BlockSizeBytes: 1 << 20}}
-		return v
-	}
-	assert.True(t, AreVolumeSocketsValid(blockVM(present)), "a present block device is valid")
-	assert.False(t, AreVolumeSocketsValid(blockVM(missing)), "a vanished block device is the orphan-QEMU signal")
-}
-
 // classifyTestManager wires a manager with an in-memory StateStore plus
 // the InstanceTypes resolver and Resources controller that
 // classifyRestoredInstances consults. Returns the manager and the

spinifex/runtime/compute/vm/volumes.go

@@ -119,36 +119,17 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string)
 		return "", fmt.Errorf("mount volume %s: %w", volumeID, err)
 	}
 
+	serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(ebsRequest.NBDURI)
+	if err != nil {
+		slog.ErrorContext(ctx, "AttachVolume: failed to parse NBDURI", "uri", ebsRequest.NBDURI, "err", err)
+		_ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest)
+		return "", fmt.Errorf("parse NBDURI: %w", err)
+	}
+	serverArg := NBDServerOpts{Type: serverType, Path: socketPath, Host: nbdHost, Port: nbdPort}.QMPArg()
+
 	// Shared with buildDrives' cold-boot path so a relaunched volume's block
 	// graph uses exactly the same names DetachVolume addresses.
 	nodeName := VolumeNodeName(volumeID)
-
-	kind, err := classifyVolumeExport(ebsRequest)
-	if err != nil {
-		slog.ErrorContext(ctx, "AttachVolume: invalid volume export", "volumeId", volumeID, "err", err)
-		_ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest)
-		return "", fmt.Errorf("volume export: %w", err)
-	}
-	var blockdevArgs map[string]any
-	if kind == volumeExportBlock {
-		blockdevArgs = HostDeviceOpts{Path: ebsRequest.BlockDevice, SizeBytes: ebsRequest.BlockSizeBytes}.QMPArgs(nodeName)
-	} else {
-		serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(ebsRequest.NBDURI)
-		if err != nil {
-			slog.ErrorContext(ctx, "AttachVolume: failed to parse NBDURI", "uri", ebsRequest.NBDURI, "err", err)
-			_ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest)
-			return "", fmt.Errorf("parse NBDURI: %w", err)
-		}
-		blockdevArgs = map[string]any{
-			"node-name":       nodeName,
-			"driver":          "nbd",
-			"server":          NBDServerOpts{Type: serverType, Path: socketPath, Host: nbdHost, Port: nbdPort}.QMPArg(),
-			"export":          "",
-			"read-only":       false,
-			"reconnect-delay": NBDReconnectDelaySeconds,
-		}
-	}
-
 	deviceID := VolumeDeviceID(volumeID)
 	iothreadID := VolumeIOThreadID(volumeID)
 
@@ -182,8 +163,15 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string)
 	}
 
 	if _, err := sendQMPCommand(ctx, instance.QMPClient, qmp.QMPCommand{
-		Execute:   "blockdev-add",
-		Arguments: blockdevArgs,
+		Execute: "blockdev-add",
+		Arguments: map[string]any{
+			"node-name":       nodeName,
+			"driver":          "nbd",
+			"server":          serverArg,
+			"export":          "",
+			"read-only":       false,
+			"reconnect-delay": NBDReconnectDelaySeconds,
+		},
 	}, instance.ID); err != nil {
 		slog.ErrorContext(ctx, "AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err)
 		m.delIothreadBestEffort(ctx, instance, iothreadID, volumeID)

spinifex/runtime/compute/vm/volumes_test.go

@@ -943,91 +943,6 @@ func TestAttachVolume_NBDURIParseFailure_TriggersUnmountOne(t *testing.T) {
 	assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne)
 }
 
-// TestAttachVolume_BlockDeviceBlockdevAdd pins the blockdev-add a block export
-// sends: a sized raw node over a host_device file node, under the same node
-// name the NBD path uses, followed by the usual device_add.
-func TestAttachVolume_BlockDeviceBlockdevAdd(t *testing.T) {
-	recorder := &qmpRecorder{}
-	qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
-		recorder.record(cmd)
-		return nil
-	})
-	defer cancel()
-
-	mounter := &fakeVolumeMounter{mountOneBlockDevice: "/dev/sdx", mountOneBlockSize: 10 << 30}
-	m, v := attachVolumeRunningInstance(t, qmpClient, mounter)
-
-	_, err := m.AttachVolume(t.Context(), "i-1", "vol-1", "/dev/sdf")
-	require.NoError(t, err)
-
-	recorder.mu.Lock()
-	var blockdevAdd *qmp.QMPCommand
-	for i := range recorder.commands {
-		if recorder.commands[i].Execute == "blockdev-add" {
-			blockdevAdd = &recorder.commands[i]
-		}
-	}
-	recorder.mu.Unlock()
-	require.NotNil(t, blockdevAdd)
-	got, err := json.Marshal(blockdevAdd.Arguments)
-	require.NoError(t, err)
-	assert.JSONEq(t, `{
-		"node-name":"nbd-vol-1", "driver":"raw", "size":10737418240,
-		"file":{"driver":"host_device", "filename":"/dev/sdx", "aio":"native", "cache":{"direct":true}}
-	}`, string(got))
-	assert.Equal(t, []string{"object-add", "blockdev-add", "device_add"}, recorder.executes()[:3])
-
-	v.EBSRequests.Mu.Lock()
-	defer v.EBSRequests.Mu.Unlock()
-	require.Len(t, v.EBSRequests.Requests, 1)
-	assert.Equal(t, "/dev/sdx", v.EBSRequests.Requests[0].BlockDevice)
-	assert.Equal(t, uint64(10<<30), v.EBSRequests.Requests[0].BlockSizeBytes)
-	assert.Empty(t, v.EBSRequests.Requests[0].NBDURI)
-}
-
-// TestAttachVolume_InvalidExport_TriggersUnmountOne covers a mount that
-// returns both exports: AttachVolume must unmount before any QMP traffic.
-func TestAttachVolume_InvalidExport_TriggersUnmountOne(t *testing.T) {
-	recorder := &qmpRecorder{}
-	qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
-		recorder.record(cmd)
-		return nil
-	})
-	defer cancel()
-
-	mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock", mountOneBlockDevice: "/dev/sdx", mountOneBlockSize: 1 << 20}
-	m, _ := attachVolumeRunningInstance(t, qmpClient, mounter)
-
-	_, err := m.AttachVolume(t.Context(), "i-1", "vol-1", "/dev/sdf")
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "both NBDURI and BlockDevice set")
-	assert.Empty(t, recorder.executes())
-	assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne)
-}
-
-// TestAttachVolume_BlockDeviceBlockdevAddFailure_TriggersUnmountOne checks a
-// block export unwinds a failed blockdev-add exactly as an NBD one does.
-func TestAttachVolume_BlockDeviceBlockdevAddFailure_TriggersUnmountOne(t *testing.T) {
-	recorder := &qmpRecorder{}
-	qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
-		recorder.record(cmd)
-		if cmd.Execute == "blockdev-add" {
-			return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "Could not open '/dev/sdx'"}}
-		}
-		return nil
-	})
-	defer cancel()
-
-	mounter := &fakeVolumeMounter{mountOneBlockDevice: "/dev/sdx", mountOneBlockSize: 1 << 20}
-	m, _ := attachVolumeRunningInstance(t, qmpClient, mounter)
-
-	_, err := m.AttachVolume(t.Context(), "i-1", "vol-1", "/dev/sdf")
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "blockdev-add")
-	assert.Equal(t, []string{"object-add", "blockdev-add", "object-del"}, recorder.executes())
-	assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne)
-}
-
 // TestAttachVolume_ObjectAddFailure_TriggersUnmountOne covers QMP iothread
 // object-add failure: AttachVolume must unmount and bail before issuing
 // blockdev-add (no further QMP traffic past the failed object-add).

spinifex/runtime/roles/awsgw/awsgw.go

@@ -60,6 +60,11 @@ var serviceName = "awsgw"
 // handlers/eks, because handlers/eks must not import handlers/sts.
 var _ handlers_sts.ClusterJWKSResolver = (*handlers_eks.OIDCJWKSReader)(nil)
 
+// AccountStatusReader also serves as the quota override seal's TERMINATING
+// precondition, wired here because handlers/iam must not import
+// domains/admission/quota.
+var _ admissionquota.AccountTerminationChecker = (*handlers_iam.AccountStatusReader)(nil)
+
 // Version and Commit are set by the cmd package before Start() to pass
 // build-time ldflags to the gateway without creating an import cycle.
 var (
@@ -561,6 +566,25 @@ func launchService(config *config.ClusterConfig) error {
 	gw.Quota = admissionquota.New(quotaCfg, usageBucket)
 	gw.Quota.SetOverrides(quotaOverrides)
 
+	// Seal precondition: the quota owner's only fact from IAM, read directly
+	// rather than through the gateway's cached IAMService, so a seal sees a
+	// TERMINATING flip the moment IAM commits it.
+	gw.Quota.SetAccountTerminationChecker(handlers_iam.NewAccountStatusReader(js))
+	// SealAccountOverrides opens the override bucket itself when this
+	// gateway's quotas are disabled, so the seal subject works regardless of
+	// quotaCfg.Enabled and never touches the bucket at startup.
+	gw.Quota.SetJetStream(js)
+
+	quotaSealSub, err := registerQuotaOverrideSeal(natsConn, gw.Quota)
+	if err != nil {
+		return fmt.Errorf("subscribe quota override seal: %w", err)
+	}
+	defer func() {
+		if err := quotaSealSub.Unsubscribe(); err != nil {
+			slog.Warn("Quota override seal: unsubscribe failed", "err", err)
+		}
+	}()
+
 	// Bedrock token quota reads the stream-fed usage counters built above,
 	// independent of whether the standing-infra dimensions are enabled.
 	gw.Quota.SetBedrockUsage(bedrockUsage)

spinifex/runtime/roles/awsgw/quota_override_seal.go

@@ -0,0 +1,39 @@
+package awsgw
+
+import (
+	"context"
+
+	admissionv1 "github.com/mulgadc/spinifex/contracts/admission/v1"
+	admissionquota "github.com/mulgadc/spinifex/spinifex/domains/admission/quota"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+)
+
+// quotaOverrideSealQueue is the NATS queue group for the quota override seal
+// responders so the request is load-balanced across awsgw nodes.
+const quotaOverrideSealQueue = "awsgw-quota-seal"
+
+// registerQuotaOverrideSeal serves SealAccountQuotaOverrideSubject on this
+// gateway. Any current gateway may answer; the leader-accepted revision is
+// the evidence.
+func registerQuotaOverrideSeal(nc *nats.Conn, quota *admissionquota.Service) (*nats.Subscription, error) {
+	return nc.QueueSubscribe(admissionv1.SealAccountQuotaOverrideSubject, quotaOverrideSealQueue,
+		func(msg *nats.Msg) {
+			natsmsg.ServeNATSRequestCtx(msg, handleSealAccountQuotaOverride(quota))
+		})
+}
+
+func handleSealAccountQuotaOverride(quota *admissionquota.Service) func(context.Context, *admissionv1.SealAccountQuotaOverrideRequest) (*admissionv1.SealAccountQuotaOverrideReply, error) {
+	return func(ctx context.Context, req *admissionv1.SealAccountQuotaOverrideRequest) (*admissionv1.SealAccountQuotaOverrideReply, error) {
+		seal, err := quota.SealAccountOverrides(ctx, req.AccountID)
+		if err != nil {
+			return nil, err
+		}
+		return &admissionv1.SealAccountQuotaOverrideReply{
+			AccountID:      req.AccountID,
+			CorrelationRef: req.CorrelationRef,
+			Disposition:    admissionv1.AccountQuotaOverrideDispositionSealed,
+			Revision:       seal.Revision,
+		}, nil
+	}
+}

spinifex/runtime/roles/awsgw/quota_override_seal_test.go

@@ -0,0 +1,84 @@
+package awsgw
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	admissionv1 "github.com/mulgadc/spinifex/contracts/admission/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	admissionquota "github.com/mulgadc/spinifex/spinifex/domains/admission/quota"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// stubChecker answers AccountTerminating from a fixed set, standing in for
+// handlers_iam.AccountStatusReader so this test does not need a real IAM
+// bucket.
+type stubChecker struct {
+	terminating map[string]bool
+}
+
+func (c *stubChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) {
+	return c.terminating[accountID], nil
+}
+
+func newQuotaSealService(t *testing.T) (*admissionquota.Service, jetstream.JetStream) {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	quota := admissionquota.New(admissionquota.Limits{}, nil)
+	quota.SetJetStream(js)
+	return quota, js
+}
+
+func TestQuotaOverrideSeal_SealsTerminatingAccount(t *testing.T) {
+	quota, js := newQuotaSealService(t)
+	_, nc := testutil.StartTestNATS(t)
+	quota.SetAccountTerminationChecker(&stubChecker{terminating: map[string]bool{"111122223333": true}})
+
+	sub, err := registerQuotaOverrideSeal(nc, quota)
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
+	req := admissionv1.SealAccountQuotaOverrideRequest{AccountID: "111122223333", CorrelationRef: "req-1"}
+	resp, err := natsmsg.NATSRequest[admissionv1.SealAccountQuotaOverrideReply](context.Background(),
+		nc, admissionv1.SealAccountQuotaOverrideSubject, req, 2*time.Second, "")
+	require.NoError(t, err)
+
+	assert.Equal(t, "111122223333", resp.AccountID)
+	assert.Equal(t, "req-1", resp.CorrelationRef)
+	assert.Equal(t, admissionv1.AccountQuotaOverrideDispositionSealed, resp.Disposition)
+	assert.NotZero(t, resp.Revision)
+
+	kv, err := js.KeyValue(context.Background(), admissionquota.KVBucketAccountQuota)
+	require.NoError(t, err)
+	entry, err := kv.Get(context.Background(), "111122223333")
+	require.NoError(t, err)
+	assert.Equal(t, resp.Revision, entry.Revision())
+}
+
+func TestQuotaOverrideSeal_RefusesActiveAccount(t *testing.T) {
+	quota, _ := newQuotaSealService(t)
+	_, nc := testutil.StartTestNATS(t)
+	quota.SetAccountTerminationChecker(&stubChecker{terminating: map[string]bool{}})
+
+	sub, err := registerQuotaOverrideSeal(nc, quota)
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
+	req := admissionv1.SealAccountQuotaOverrideRequest{AccountID: "111122223333"}
+	_, err = natsmsg.NATSRequest[admissionv1.SealAccountQuotaOverrideReply](context.Background(),
+		nc, admissionv1.SealAccountQuotaOverrideSubject, req, 2*time.Second, "")
+	require.Error(t, err, "a non-TERMINATING account must be refused")
+}
+
+func TestQuotaOverrideSeal_NoResponderWithoutGateway(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+
+	req := admissionv1.SealAccountQuotaOverrideRequest{AccountID: "111122223333"}
+	_, err := natsmsg.NATSRequest[admissionv1.SealAccountQuotaOverrideReply](context.Background(),
+		nc, admissionv1.SealAccountQuotaOverrideSubject, req, 500*time.Millisecond, "")
+	require.Error(t, err, "no subscriber must surface as a no-responders error")
+}

spinifex/runtime/roles/qemunbdd/live_lifecycle_test.go

@@ -104,8 +104,8 @@ func TestLive_QEMUNBDLifecycle(t *testing.T) {
 
 	pub, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID})
 	require.NoError(t, err, "publish volume")
-	t.Logf("published volume %s to node %s: %s", pub.VolumeID, pub.NodeID, liveNBDURI(t, pub))
-	sockPath := requireUnixSocketPath(t, liveNBDURI(t, pub))
+	t.Logf("published volume %s to node %s: %s", pub.VolumeID, pub.NodeID, pub.NBDURI)
+	sockPath := requireUnixSocketPath(t, pub.NBDURI)
 
 	writePattern(t, sockPath, pattern, liveVolumeBytes)
 	readAndVerifyPattern(t, sockPath, pattern, liveVolumeBytes)
@@ -129,8 +129,8 @@ func TestLive_QEMUNBDLifecycle(t *testing.T) {
 
 	restoredPub, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID})
 	require.NoError(t, err, "publish restored volume")
-	t.Logf("published restored volume %s to node %s: %s", restoredPub.VolumeID, restoredPub.NodeID, liveNBDURI(t, restoredPub))
-	restoredSockPath := requireUnixSocketPath(t, liveNBDURI(t, restoredPub))
+	t.Logf("published restored volume %s to node %s: %s", restoredPub.VolumeID, restoredPub.NodeID, restoredPub.NBDURI)
+	restoredSockPath := requireUnixSocketPath(t, restoredPub.NBDURI)
 
 	// The real proof: bytes written before the snapshot survive both the
 	// snapshot and the restore into an independent volume.
@@ -151,15 +151,6 @@ func cleanupBestEffort(t *testing.T, step string, fn func() error) {
 	}
 }
 
-// liveNBDURI requires pub to carry an nbd export, which is all qemunbdd
-// publishes, and returns its URI.
-func liveNBDURI(t *testing.T, pub *ebsprovider.PublishedVolume) string {
-	t.Helper()
-	require.Equal(t, ebsprovider.ExportKindNBD, pub.Export.Kind, "qemunbdd must publish an nbd export")
-	require.NotNil(t, pub.Export.NBD, "nbd export has no nbd details")
-	return pub.Export.NBD.NBDURI
-}
-
 // requireUnixSocketPath asserts nbdURI is a nbd:unix: URI and returns its
 // socket path, parsed with the same helper the daemon uses to format it
 // rather than by slicing the string.

spinifex/runtime/roles/qemunbdd/provider.go

@@ -688,7 +688,7 @@ func (p *Provider) PublishVolume(ctx context.Context, req ebsprovider.PublishVol
 		if pub.nodeID != req.NodeID {
 			return nil, fmt.Errorf("%w: volume %s is published to %s", ebsprovider.ErrVolumeInUse, req.VolumeID, pub.nodeID)
 		}
-		return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: pub.nodeID, Export: ebsprovider.NewNBDExport(nbd.FormatNBDSocketURI(pub.socketPath))}, nil
+		return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: pub.nodeID, NBDURI: nbd.FormatNBDSocketURI(pub.socketPath)}, nil
 	}
 
 	sockPath := p.socketPath(req.VolumeID)
@@ -718,7 +718,7 @@ func (p *Provider) PublishVolume(ctx context.Context, req ebsprovider.PublishVol
 
 	pid := readPID(pidPath)
 	p.published[req.VolumeID] = &publication{nodeID: req.NodeID, socketPath: sockPath, pid: pid}
-	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: ebsprovider.NewNBDExport(nbd.FormatNBDSocketURI(sockPath))}, nil
+	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, NBDURI: nbd.FormatNBDSocketURI(sockPath)}, nil
 }
 
 // readPID best-effort reads the PID qemu-nbd's --fork wrote to pidPath. A

spinifex/runtime/roles/qemunbdd/provider_test.go

@@ -544,7 +544,7 @@ func TestPublishVolume_ArgvReadWrite(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, "vol-pub", pub.VolumeID)
 	assert.Equal(t, "node-1", pub.NodeID)
-	assert.Equal(t, ebsprovider.NewNBDExport("nbd+unix:///?socket="+p.socketPath("vol-pub")), pub.Export)
+	assert.Equal(t, "nbd+unix:///?socket="+p.socketPath("vol-pub"), pub.NBDURI)
 
 	calls := fr.callsSnapshot()
 	require.Len(t, calls, 1)