Merge branch 'feat/ebs-export-descriptor-compute' into feat/ebs-export-descriptor

PublishedVolume now carries an Export of kind nbd or block, checked by VolumeExport.Validate.
The NATS client rejects an invalid export in a PublishVolume reply, and the server refuses to relay one, reporting it as internal.
SchemaVersion moves to 2.

viperblockd, qemunbdd and the memory provider publish nbd exports exactly as before.
Image import refuses a non-nbd export and still unpublishes.
Conformance requires a valid export, and the NBD-dialling suites fail explicitly on a non-nbd export.

contracts/ec2/v1/README.md

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

contracts/ec2/v1/account_settings_seal.go

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

contracts/ec2/v1/account_settings_seal_test.go

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

contracts/viperblockd/legacy/v1/volume.go

@@ -48,6 +48,11 @@ 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,6 +35,18 @@ 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) {

spinifex/accountteardown/reapers_ec2_account_settings_current_behaviour_test.go

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

spinifex/accountteardown/reapers_quota_override_current_behaviour_test.go

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

spinifex/daemon/daemon.go

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

spinifex/daemon/daemon_handlers_account_settings_seal.go

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

spinifex/daemon/daemon_handlers_account_settings_seal_test.go

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

spinifex/domains/admission/quota/cleanup.go

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

spinifex/domains/admission/quota/cleanup_test.go

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

spinifex/domains/ec2/account/seal.go

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

spinifex/domains/ec2/account/seal_test.go

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

spinifex/domains/ec2/account/service_impl.go

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

spinifex/domains/ec2/account/service_impl_test.go

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

spinifex/handlers/iam/account_status_reader.go

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

spinifex/handlers/iam/account_status_reader_test.go

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

spinifex/operator/admin/image_import.go

@@ -71,7 +71,12 @@ func ImportImage(ctx context.Context, provider ebsprovider.EBSProvider, opts Imp
 		return fmt.Errorf("import: publish volume %s: %w", opts.VolumeID, err)
 	}
 
-	writeErr := writeImage(ctx, opts.SourcePath, published.NBDURI, opts.Progress)
+	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)
+	}
 
 	// 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,6 +22,10 @@ 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 {
@@ -39,7 +43,11 @@ func (p *recordingProvider) PublishVolume(ctx context.Context, req ebsprovider.P
 	if _, err := p.MemoryProvider.PublishVolume(ctx, req); err != nil {
 		return nil, err
 	}
-	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, NBDURI: p.publishURI}, nil
+	export := ebsprovider.NewNBDExport(p.publishURI)
+	if p.publishExport.Kind != "" {
+		export = p.publishExport
+	}
+	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: export}, nil
 }
 
 func (p *recordingProvider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) error {
@@ -117,6 +125,32 @@ 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,7 +791,10 @@ 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)
-			assertNBDURI(t, pub.NBDURI)
+			require.NoError(t, pub.Export.Validate(), "published export is not attachable")
+			if pub.Export.Kind == ebsprovider.ExportKindNBD {
+				assertNBDURI(t, pub.Export.NBD.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"
 
-	pub := publishForNBD(t, provider, cfg, volumeID, false)
-	assertNBDURI(t, pub.NBDURI)
-	uri := dialURI(t, pub.NBDURI)
+	nbdURI := publishForNBD(t, provider, cfg, volumeID, false)
+	assertNBDURI(t, nbdURI)
+	uri := dialURI(t, 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, 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 {
+// 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 {
 	t.Helper()
 	ctx := context.Background()
 	_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{
@@ -198,7 +198,17 @@ func publishForNBD(t *testing.T, provider ebsprovider.EBSProvider, cfg NBDClient
 	})
 	require.NoError(t, err)
 	require.NotNil(t, pub)
-	return 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
 }
 
 // cleanupVolume unpublishes and deletes best-effort: cleanup runs after the
@@ -234,21 +244,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)
-		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-uri", false)
-		assertNBDURI(t, pub.NBDURI)
+		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-uri", false)
+		assertNBDURI(t, nbdURI)
 	})
 
 	t.Run("export is connectable by a standard NBD client", func(t *testing.T) {
 		provider := newProvider(t)
-		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-connect", false)
-		export := nbdInfo(t, dialURI(t, pub.NBDURI))
+		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-connect", false)
+		export := nbdInfo(t, dialURI(t, nbdURI))
 		assert.Equal(t, cfg.VolumeBytes, export.Size)
 	})
 
 	t.Run("export advertises usable block size constraints", func(t *testing.T) {
 		provider := newProvider(t)
-		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-blocksize", false)
-		export := nbdInfo(t, dialURI(t, pub.NBDURI))
+		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-blocksize", false)
+		export := nbdInfo(t, dialURI(t, nbdURI))
 
 		require.Positivef(t, export.BlockSizeMinimum,
 			"export advertises no minimum block size, so a client cannot know how to align its requests")
@@ -261,8 +271,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)
-		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-reconnect", false)
-		uri := dialURI(t, pub.NBDURI)
+		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-reconnect", false)
+		uri := dialURI(t, nbdURI)
 
 		first := nbdInfo(t, uri)
 
@@ -275,7 +285,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)
-		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-roundtrip", false)
+		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-roundtrip", false)
 
 		dir := t.TempDir()
 		in := filepath.Join(dir, "in.bin")
@@ -283,7 +293,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, pub.NBDURI)
+		uri := dialURI(t, nbdURI)
 		runNBDCopy(t, in, uri)
 		runNBDCopy(t, uri, out)
 
@@ -301,9 +311,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)
-		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-extents", false)
+		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-extents", false)
 
-		holes := slices.ContainsFunc(nbdMap(t, dialURI(t, pub.NBDURI)), func(extent nbdExtent) bool {
+		holes := slices.ContainsFunc(nbdMap(t, dialURI(t, nbdURI)), func(extent nbdExtent) bool {
 			return strings.Contains(extent.Description, "hole")
 		})
 		assert.Equalf(t, capabilities.SparseExtentReporting, holes,
@@ -334,16 +344,16 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
 			return
 		}
 
-		pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-readonly", true)
-		export := nbdInfo(t, dialURI(t, pub.NBDURI))
+		nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-readonly", true)
+		export := nbdInfo(t, dialURI(t, 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"
-		pub := publishForNBD(t, provider, cfg, volumeID, false)
-		uri := dialURI(t, pub.NBDURI)
+		nbdURI := publishForNBD(t, provider, cfg, volumeID, false)
+		uri := dialURI(t, 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,13 +74,24 @@ func corruptPublished(published *ebsprovider.PublishedVolume, flavour string) *e
 	corrupted := *published
 	switch flavour {
 	case "handle", "capacity":
-		corrupted.NBDURI = "nbd+unix:///?socket=/hostile/wrong.sock"
+		corrupted.Export = corruptExport(published.Export)
 	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,
-		NBDURI:   "nbd+unix:///?socket=/memory/" + req.VolumeID + ".sock",
+		Export:   NewNBDExport("nbd+unix:///?socket=/memory/" + req.VolumeID + ".sock"),
 	}
 	volume.volume.State = VolumeStateInUse
 	return clonePublished(volume.published), nil
@@ -421,5 +421,13 @@ 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, "nbd+unix:///?socket=/memory/vol-1.sock", published.NBDURI)
+	assert.Equal(t, NewNBDExport("nbd+unix:///?socket=/memory/vol-1.sock"), published.Export)
 
 	repeatedPublish, err := provider.PublishVolume(ctx, PublishVolumeRequest{
 		Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle, NodeID: "node-1",

spinifex/providers/ebs/nats.go

@@ -452,6 +452,9 @@ 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,6 +432,50 @@ 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, "nbd+unix:///?socket=/memory/vol-roundtrip.sock", published.NBDURI)
+	assert.Equal(t, ebsprovider.NewNBDExport("nbd+unix:///?socket=/memory/vol-roundtrip.sock"), published.Export)
 
 	inUse, err := client.GetVolume(ctx, ebsprovider.GetVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip"})
 	require.NoError(t, err)
@@ -432,6 +432,42 @@ 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.
@@ -543,7 +579,7 @@ func TestServeRejectsUnroutableNodeID(t *testing.T) {
 	assert.Contains(t, err.Error(), "build publish subject")
 	assert.Nil(t, stop)
 
-	_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(`{"schema_version":1}`), 200*time.Millisecond)
+	_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(fmt.Sprintf(`{"schema_version":%d}`, ebsprovider.SchemaVersion)), 200*time.Millisecond)
 	require.ErrorIs(t, err, nats.ErrNoResponders, "a failed Serve must leave no subscription behind")
 }
 
@@ -571,6 +607,6 @@ func TestServeStopIsIdempotent(t *testing.T) {
 	stop()
 	require.NoError(t, conn.Flush())
 
-	_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(`{"schema_version":1}`), 200*time.Millisecond)
+	_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(fmt.Sprintf(`{"schema_version":%d}`, ebsprovider.SchemaVersion)), 200*time.Millisecond)
 	require.ErrorIs(t, err, nats.ErrNoResponders, "stop must unsubscribe every subject Serve registered")
 }

spinifex/providers/ebs/natsserve/server.go

@@ -501,6 +501,16 @@ 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,14 +6,16 @@ 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 = 1
+const SchemaVersion uint16 = 2
 
 // Versioned is embedded in every request and response sent over NATS.
 type Versioned struct {
@@ -443,12 +445,80 @@ type PublishVolumeRequest struct {
 	ReadOnly bool   `json:"read_only,omitempty"`
 }
 
-// PublishedVolume is where a published volume is served: the NBD URI on NodeID
-// that QEMU connects to.
+// 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.
 type PublishedVolume struct {
-	VolumeID string `json:"volume_id"`
-	NodeID   string `json:"node_id"`
-	NBDURI   string `json:"nbd_uri"`
+	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}}
 }
 
 // PublishVolumeResponse is the reply to PublishVolumeRequest: the mount, or

spinifex/providers/ebs/provider_test.go

@@ -0,0 +1,63 @@
+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,
-				NBDURI:   mv.NBDURI,
+				Export:   ebsprovider.NewNBDExport(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,
-			NBDURI:   mountResp.URI,
+			Export:   ebsprovider.NewNBDExport(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, existingURI, resp.Published.NBDURI)
+	assert.Equal(t, ebsprovider.NewNBDExport(existingURI), resp.Published.Export)
 
 	cfg.mu.Lock()
 	count := len(cfg.MountedVolumes)

spinifex/runtime/compute/vm/devices.go

@@ -355,3 +355,35 @@ 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,6 +1,10 @@
 package vm
 
 import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
 	"sync"
 
 	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
@@ -20,3 +24,45 @@ 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,11 +607,18 @@ 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 {
-		if req.NBDURI != "" {
+		switch kind, _ := classifyVolumeExport(req); kind {
+		case volumeExportNBD:
 			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()
@@ -620,6 +627,11 @@ 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)
@@ -900,11 +912,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. Every production guest boots from one; a VM without is a
-// unit test.
+// monitor answers. A host block device opens locally and does not count.
 func hasNetworkBackedDrive(v *VM) bool {
 	return slices.ContainsFunc(v.EBSRequests.Requests, func(req EBSRequest) bool {
-		return req.NBDURI != ""
+		kind, _ := classifyVolumeExport(req)
+		return kind == volumeExportNBD
 	})
 }
 
@@ -1445,8 +1457,9 @@ type driveConfig struct {
 }
 
 // buildDrives converts EBS requests into QEMU drive/iothread/device/blockdev
-// 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.
+// 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.
 //
 // Boot volumes keep the existing anonymous if=none -drive shape exactly as
 // before — they are not detachable (see ErrVolumeNotDetachable), so the
@@ -1471,11 +1484,17 @@ func buildDrives(requests []EBSRequest, cpuCount int, machineType string) (drive
 
 	for i := range requests {
 		v := &requests[i]
-		if v.NBDURI == "" {
-			return driveConfig{}, fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", v.Name)
+		kind, err := classifyVolumeExport(*v)
+		if err != nil {
+			return driveConfig{}, err
 		}
 
 		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,
@@ -1529,29 +1548,39 @@ 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, VolumeBlockdev(nodeName, NBDServerOpts{
-				Type: serverType,
-				Path: socketPath,
-				Host: nbdHost,
-				Port: nbdPort,
-			}))
+			cfg.Blockdevs = append(cfg.Blockdevs, blockdev)
 			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.
 		}
 
-		slog.Info("Using NBD URI for drive", "volume", v.Name, "uri", v.NBDURI)
+		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)
+		}
 	}
 
 	return cfg, nil

spinifex/runtime/compute/vm/lifecycle_test.go

@@ -273,6 +273,83 @@ 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{},
@@ -1008,6 +1085,28 @@ 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
@@ -1271,6 +1370,12 @@ 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,6 +142,10 @@ 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.
@@ -181,6 +185,7 @@ 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
@@ -188,6 +193,10 @@ 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,14 +523,22 @@ func isInstanceProcessRunning(instance *VM) bool {
 }
 
 // AreVolumeSocketsValid dials each NBD Unix socket to confirm viperblock is
-// still listening. A stat-only check is insufficient since viperblock may restart
-// leaving stale socket files. TCP and unparseable URIs are treated as valid.
+// still listening, and checks each block export's device still exists. TCP,
+// unparseable and invalid exports are treated as valid.
 func AreVolumeSocketsValid(instance *VM) bool {
 	instance.EBSRequests.Mu.Lock()
 	defer instance.EBSRequests.Mu.Unlock()
 
 	for _, req := range instance.EBSRequests.Requests {
-		if req.NBDURI == "" {
+		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 {
 			continue
 		}
 		serverType, sockPath, _, _, err := nbd.ParseNBDURI(req.NBDURI)

spinifex/runtime/compute/vm/restore_test.go

@@ -82,6 +82,22 @@ 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,17 +119,36 @@ 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)
 
@@ -163,15 +182,8 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string)
 	}
 
 	if _, err := sendQMPCommand(ctx, instance.QMPClient, qmp.QMPCommand{
-		Execute: "blockdev-add",
-		Arguments: map[string]any{
-			"node-name":       nodeName,
-			"driver":          "nbd",
-			"server":          serverArg,
-			"export":          "",
-			"read-only":       false,
-			"reconnect-delay": NBDReconnectDelaySeconds,
-		},
+		Execute:   "blockdev-add",
+		Arguments: blockdevArgs,
 	}, 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,6 +943,91 @@ 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/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, pub.NBDURI)
-	sockPath := requireUnixSocketPath(t, pub.NBDURI)
+	t.Logf("published volume %s to node %s: %s", pub.VolumeID, pub.NodeID, liveNBDURI(t, pub))
+	sockPath := requireUnixSocketPath(t, liveNBDURI(t, pub))
 
 	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, restoredPub.NBDURI)
-	restoredSockPath := requireUnixSocketPath(t, restoredPub.NBDURI)
+	t.Logf("published restored volume %s to node %s: %s", restoredPub.VolumeID, restoredPub.NodeID, liveNBDURI(t, restoredPub))
+	restoredSockPath := requireUnixSocketPath(t, liveNBDURI(t, restoredPub))
 
 	// The real proof: bytes written before the snapshot survive both the
 	// snapshot and the restore into an independent volume.
@@ -151,6 +151,15 @@ 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, NBDURI: nbd.FormatNBDSocketURI(pub.socketPath)}, nil
+		return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: pub.nodeID, Export: ebsprovider.NewNBDExport(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, NBDURI: nbd.FormatNBDSocketURI(sockPath)}, nil
+	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: ebsprovider.NewNBDExport(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, "nbd+unix:///?socket="+p.socketPath("vol-pub"), pub.NBDURI)
+	assert.Equal(t, ebsprovider.NewNBDExport("nbd+unix:///?socket="+p.socketPath("vol-pub")), pub.Export)
 
 	calls := fr.callsSnapshot()
 	require.Len(t, calls, 1)