fix: keep surviving flows on a failed IMDS repair and repair guests concurrently

Clear a tap's flows only once its endpoint is set up, so a repair that fails
after a kernel-module reload no longer takes down a guest that still had
connectivity. Repair up to 8 datapaths at once, since an ovs-vswitchd restart
drops every guest together. Test that teardown holds the ENI lock across
detach and tap cleanup, and count only the test guest's flows in the e2e.

Makefile

@@ -379,9 +379,9 @@ install-system:
 		ovn-central ovn-host openvswitch-switch dhcpcd-base
 
 install-go:
-	@echo -e "\n....Installing Go 1.27.3 for $(ARCH) ($(GO_ARCH))...."
+	@echo -e "\n....Installing Go 1.27.2 for $(ARCH) ($(GO_ARCH))...."
 	@if [ ! -d "/usr/local/go" ]; then \
-		curl -L https://go.dev/dl/go1.27.3.linux-$(GO_ARCH).tar.gz | tar -C /usr/local -xz; \
+		curl -L https://go.dev/dl/go1.27.2.linux-$(GO_ARCH).tar.gz | tar -C /usr/local -xz; \
 	else \
 		echo "Go already installed in /usr/local/go"; \
 	fi

build/Dockerfile.distro

@@ -1,6 +1,6 @@
 ARG VERSION=dev
 
-FROM golang:1.27.3-bookworm@sha256:5c3e9b43cfd3bb95c7937462e29d87ada57b481a96fa23ad0eb61778d7a00110 AS builder
+FROM golang:1.27.2-bookworm@sha256:5cf287a799e6b94384bad13d16b14904c531f51ba65792237e122ce42b392f61 AS builder
 
 ARG VERSION
 

contracts/ec2/v1/README.md

@@ -60,25 +60,3 @@ 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

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

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

contracts/viperblockd/legacy/v1/volume_test.go

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

docs/COMMANDS.md

@@ -109,7 +109,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t
 | Command | Flags | Description |
 |---------|-------|-------------|
 | `spx admin kv digest` | `--store-dir` (default `<spinifex-dir>/nats/jetstream`), `--work-dir` (holds the temporary copy, default the system temp dir), `--stream` (repeatable filter), `--seqs` (also print a hash per sequence), `--json` (print one JSON document, the input `spx admin kv compare` reads) | Read-only per-replica consistency check. Copies this node's `$G` streams tree (without consumer state) to a temporary directory, opens the copy in a private in-process NATS server with no listener, and prints one line per stream with first/last sequence, message count and a SHA-256 over every stored message (sequence, subject, timestamp, headers, data). Run it on every node with `sudo` and diff the output with the `#` lines removed: replicas of one stream must match. A direct get over NATS is not used because any replica may answer it. Writes landing during the copy can change a replica's tail, so gate API writes or re-run before calling a tail difference divergence. Prefer `--json --seqs` on every node followed by `spx admin kv compare`, which tells a diverged replica from one that is merely behind. |
-| `spx admin kv compare <digest.json>...` | — (`-` reads any number of digests from stdin) | Reads the `spx admin kv digest --json --seqs` output of every node and checks that each stream's replicas hold the same content. A replica that is merely behind is not divergence; these are: the same sequence holding different content on two replicas, a message missing from one replica that nothing on it explains (no later write to the same subject, no expiry under `max_age` or a message TTL, no stream limit dropping its head), a stream held by fewer or more nodes than its replica count, replicas disagreeing on that count, or no sequence range common to every replica. A stream with a `max_age` whose replicas were copied at different times and share no sequence range is reported as `skipped` (volatile) rather than diverged, since a short-lived bucket such as a leader lease turns over between copies; any sequence two of its replicas share must still match, and each message a replica lacks must be superseded, expired, or written after that replica was copied. Sequences past the slowest replica's last are not compared. Exits 0 when every stream is consistent, 1 on divergence, 2 on bad input. Digests are taken from live stores seconds apart, so re-run before trusting a failure that a write landing between two copies could explain. `install-node.sh` gates cluster formation on it. |
+| `spx admin kv compare <digest.json>...` | — (`-` reads any number of digests from stdin) | Reads the `spx admin kv digest --json --seqs` output of every node and checks that each stream's replicas hold the same content. A replica that is merely behind is not divergence; these are: the same sequence holding different content on two replicas, a message missing from one replica that nothing on it explains (no later write to the same subject, no expiry under `max_age` or a message TTL, no stream limit dropping its head), a stream held by fewer or more nodes than its replica count, replicas disagreeing on that count, or no sequence range common to every replica. Sequences past the slowest replica's last are not compared. Exits 0 when every stream is consistent, 1 on divergence, 2 on bad input. Digests are taken from live stores seconds apart, so re-run before trusting a failure that a write landing between two copies could explain. `install-node.sh` gates cluster formation on it. |
 | `spx admin cluster shutdown` | `--force` (shutdown even if nodes don't respond), `--timeout` (max wait per phase, default 120s), `--dry-run` (print phase plan without executing) | Performs coordinated, phased shutdown of entire cluster. Phases execute in order: GATE (stop API/UI) → DRAIN (stop VMs) → STORAGE (stop viperblock) → PERSIST (stop predastore) → INFRA (stop NATS/daemon). Each phase waits for all nodes to ACK before proceeding. Uses JetStream state tracking. |
 | `spx admin cluster drain-dhcp` | `--timeout` (reply-collection window, default 30s) | Asks each vpcd to DHCPRELEASE every external-pool DHCP lease it currently holds, returning them to the upstream DHCP server. Run on teardown before stopping services — an env reset otherwise strands held leases upstream until their TTL expires, eventually exhausting the upstream scope. Best-effort: warns and exits 0 if the cluster is already down. |
 | `spx admin node drain --local` | `--local` (drain the local node only, required), `--timeout` (max wait per phase, default 120s), `--unless-restarting` (drain on a real host shutdown or a plain target stop, and skip only when a restart of the stack is already queued; unset, it always drains) | Runs the GATE and DRAIN phases against the local node only: powers down its guests via QMP and unmounts their volumes (flushing the viperblock WAL) while every service is still up. STORAGE/PERSIST/INFRA are left to systemd's ordered teardown. Wired as the `ExecStop` of `spinifex-shutdown.service`, so `systemctl stop spinifex.target` and host reboot/poweroff drain guests before any storage service stops. |

docs/identity/imds/README.md

@@ -286,6 +286,7 @@ Useful background for host-side troubleshooting:
 - The IMDS HTTP server runs inside the `spinifex-vpcd` service on each host, listening on `169.254.169.254:80` with one listener per instance's primary interface — attribution is structural, with no source-IP trust.
 - A freshly launched guest's metadata service comes up within a few seconds of launch; cloud-init's built-in retries absorb this window.
 - Session tokens and cached role credentials are not persisted. A vpcd restart drops them; SDKs and cloud-init transparently reissue. Datapath state on the host survives service restarts.
+- An Open vSwitch restart, upgrade or crash drops every `br-imds` flow, which takes the guest's whole network down, not only IMDS. `spinifex-daemon` checks each running guest's datapath every 10 seconds and re-installs what was lost.
 
 ## Limits and Defaults
 
@@ -362,6 +363,15 @@ ovs-vsctl list-ports br-imds          # per-instance endpoint + patch ports
 ovs-vsctl list-ports br-int | grep imi-
 ```
 
+3. If the guest has lost all networking, not just IMDS, check whether `br-imds` still has its ports but no flows:
+
+```bash
+sudo ovs-ofctl dump-flows br-imds | grep -c cookie=   # 7 per running instance
+journalctl -u spinifex-daemon | grep 'IMDS: per-tap datapath'
+```
+
+That is the state an Open vSwitch restart leaves behind. The daemon repairs it within 10 seconds and logs `IMDS: per-tap datapath was lost; re-installed`. A count that stays at 0 with `IMDS: per-tap datapath repair failed` in the log names the step that failed.
+
 ### cloud-init Did Not Apply SSH Key or User Data
 
 cloud-init sources both from IMDS at first boot. Check `public-keys/0/openssh-key` and `user-data` respond from inside the guest (see [Using IMDS from a Guest](#using-imds-from-a-guest)), and review `/var/log/cloud-init.log` in the guest. A 404 on `public-keys/` means the instance was launched without `--key-name` or the key pair was since deleted.

docs/terraform-workbooks/demo-app/Dockerfile

@@ -4,7 +4,7 @@
 # distroless base (no shell, no package manager) for a small, production-shaped
 # image suitable for pushing to ECR and pulling onto EKS workers.
 
-FROM golang:1.27.3-alpine@sha256:f2c872bcee949b7be9c490efd02b48a8a2fbf947cdf533f1c58525ec9d8170fd AS build
+FROM golang:1.27.2-alpine@sha256:f92b6ef800e499660581efdabdf25d9d817a9d124eaf900924f0504e7e27e12d AS build
 WORKDIR /src
 COPY go.mod ./
 COPY main.go ./

docs/terraform-workbooks/demo-app/go.mod

@@ -1,3 +1,3 @@
 module github.com/mulgadc/spinifex-demo
 
-go 1.27.3
+go 1.27.2

docs/testing/e2e-coverage.md

@@ -19,7 +19,7 @@ The nightly permutation sets are deliberately narrower than the full ones: those
 
 ### `single` — the broad single-node EC2/EBS/VPC surface
 
-The largest suite, 25 top-level tests. Notable entry points:
+The largest suite, 27 top-level tests. Notable entry points:
 
 - **`TestVPCEgressPaths`** — **the authority on subnet egress semantics.** One VPC with a public and a private subnet, sharing two guests across four stages: `PublicSubnetEgress` (IGW egress works), `RouteBeforeSubnet` (a regression guard for an IGW route installed on the main route table before the subnet exists), the NAT Gateway rounds, and `EIPFlip` (associating an Elastic IP onto a running instance).
   - **`NATBaseline` hard-fails if a private-subnet guest reaches the internet with no NAT Gateway.** `NATGatewayUp` then proves egress appears and `NATGatewayDown` that it disappears. `SPINIFEX_VPCEGRESS_NAT_ROUNDS` repeats the up/down cycle for a soak run.
@@ -27,6 +27,7 @@ The largest suite, 25 top-level tests. Notable entry points:
 - `TestSGPolicyDatapath`, `TestSGReachabilityPolicy` — security groups as an actual datapath, not just API state.
 - `TestWANEgressSMTPBlock` — egress port blocking.
 - `TestInstanceMetadata` — IMDS from inside a guest.
+- **`TestIMDSDatapathSurvivesVswitchdRestart`** restarts `ovs-vswitchd` under a guest with an Elastic IP, once through systemd and once with SIGKILL. After each, the guest must answer over its EIP and fetch its instance ID through IMDSv2 within 60 seconds, and `br-imds` must hold all 7 of the guest's own flows. It is sequential because it takes the whole node's datapath down.
 - `TestStopStart`, `TestEarlyRebootLiveness`, `TestGuestChurnDurability`, `TestSnapshotLifecycle`, `TestSnapshotBackedLaunch`, `TestCreateImage`, `TestVolumeLifecycle`, `TestSpotInstanceLifecycle`, `TestLaunchTemplateBoot`.
 - `TestNegativeErrorPaths` — the `8a`–`8t` AWS error-shape cases (malformed AMI ID, invalid instance type, volume in use, detach-root forbidden, unknown or malformed instance IDs on the lifecycle verbs, and so on).
 - **`TestKVReplication`** — the single-node half of the KV replication policy, and it is not the multi-node assertion with a smaller number in it. On one node one replica per bucket is the correct answer rather than the defect, so what it proves is the opposite: that nothing declared a replica count this deployment cannot place, that `spx admin kv replicas` exits 0 and reports `WANT 1`, and that `--repair` run twice against a cluster with nothing to repair changes no bucket's replica count and leaves every bucket serving. The multi-node suites cannot cover any of that, because on three nodes the unplaceable count is the right one. It lives here rather than in its own suite so it runs in every single-node nightly cell, across all three host and network permutations.

go.mod

@@ -1,6 +1,6 @@
 module github.com/mulgadc/spinifex
 
-go 1.27.3
+go 1.27.2
 
 require (
 	github.com/aws/aws-sdk-go v1.55.8

spinifex/accountteardown/reapers_ec2_account_settings_current_behaviour_test.go

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

spinifex/accountteardown/reapers_quota_override_current_behaviour_test.go

@@ -1,61 +0,0 @@
-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,7 +1026,6 @@ 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, ""},
@@ -1993,12 +1992,6 @@ 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
@@ -2314,6 +2307,9 @@ func (d *Daemon) startCluster() error {
 	d.startRecordRepair()
 	d.startInstanceRecovery()
 	d.vmMgr.StartPendingWatchdog(d.ctx)
+	// ovs-vswitchd keeps br-imds flows only in memory and nothing else restores
+	// them, so a restart would leave every guest on this node without a network.
+	d.vmMgr.StartIMDSDatapathReconcile(d.ctx)
 
 	// Reality→desired GC backstop (ADR-0003 §3): finish teardown interrupted by
 	// a node-down mid-cascade and purge completed terminated records. The volume

spinifex/daemon/daemon_handlers_account_settings_seal.go

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

@@ -1,101 +0,0 @@
-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/daemon/daemon_handlers_platform_inventory_test.go

@@ -164,9 +164,6 @@ func newInventoryCluster(t *testing.T) *inventoryCluster {
 	d.placementGroupService = groups
 	d.launchTemplateService = templates
 	require.NoError(t, d.subscribeAll())
-	// The caller is a separate connection, so the server must hold the SUBs
-	// before it publishes or the request can see no responders.
-	require.NoError(t, nc.Flush())
 
 	c.caller, err = nats.Connect(ns.ClientURL())
 	require.NoError(t, err)

spinifex/daemon/daemon_mgmt_ip_test.go

@@ -639,6 +639,10 @@ func (p *recordingPlumber) CleanupTap(name string) error {
 func (p *recordingPlumber) EnsureIMDSDatapathBridge() error         { return nil }
 func (p *recordingPlumber) AttachIMDSDatapath(_, _, _ string) error { return nil }
 func (p *recordingPlumber) DetachIMDSDatapath(string) error         { return nil }
+func (p *recordingPlumber) IMDSDatapathStates([]string) (map[string]vm.IMDSDatapathState, error) {
+	return nil, nil
+}
+func (p *recordingPlumber) RepairIMDSDatapath(_, _, _ string) error { return nil }
 func (p *recordingPlumber) EnsureVPCHostPort(_, _, _ string) error  { return nil }
 func (p *recordingPlumber) RemoveVPCHostPort(string) error          { return nil }
 

spinifex/daemon/platform_disclosure_current_behaviour_test.go

@@ -61,9 +61,6 @@ func newDisclosureCluster(t *testing.T) *disclosureCluster {
 	require.NoError(t, err)
 	d.wirePlatformTagIndex()
 	require.NoError(t, d.subscribeAll())
-	// The caller is a separate connection, so the server must hold the SUBs
-	// before it publishes or the request can see no responders.
-	require.NoError(t, nc.Flush())
 
 	caller, err := nats.Connect(ns.ClientURL())
 	require.NoError(t, err)

spinifex/domains/admission/quota/cleanup.go

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

spinifex/domains/admission/quota/cleanup_test.go

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

spinifex/domains/ec2/account/seal.go

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

@@ -1,559 +0,0 @@
-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,13 +5,12 @@ 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"
@@ -21,19 +20,14 @@ 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"`
-	State                  AccountSettingsState `json:"state,omitempty"`
+	EbsEncryptionByDefault bool `json:"ebs_encryption_by_default"`
+	SerialConsoleAccess    bool `json:"serial_console_access"`
 }
 
 // AccountSettingsServiceImpl implements account settings operations with NATS JetStream persistence.
@@ -41,10 +35,6 @@ 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)
@@ -82,99 +72,50 @@ func settingsKey(accountID string) string {
 	return accountID
 }
 
-// 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.
+// getSettings retrieves current account settings.
 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{}, nil
+			return &AccountSettingsRecord{
+				EbsEncryptionByDefault: false,
+				SerialConsoleAccess:    false,
+			}, nil
 		}
-		slog.ErrorContext(ctx, "Failed to read account settings from KV", "accountID", accountID, "err", err)
-		return nil, errors.New(awserrors.ErrorInternalError)
+		return nil, fmt.Errorf("failed to get account settings: %w", err)
 	}
 
 	var settings AccountSettingsRecord
 	if err := json.Unmarshal(entry.Value(), &settings); err != nil {
-		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 nil, 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)
 
-	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.EbsEncryptionByDefault = true }); err != nil {
+	settings, err := s.getSettings(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	settings.EbsEncryptionByDefault = true
+
+	if err := s.saveSettings(ctx, settings, accountID); err != nil {
 		return nil, err
 	}
 
@@ -187,7 +128,14 @@ 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)
 
-	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.EbsEncryptionByDefault = false }); err != nil {
+	settings, err := s.getSettings(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	settings.EbsEncryptionByDefault = false
+
+	if err := s.saveSettings(ctx, settings, accountID); err != nil {
 		return nil, err
 	}
 
@@ -228,7 +176,14 @@ 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)
 
-	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.SerialConsoleAccess = true }); err != nil {
+	settings, err := s.getSettings(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	settings.SerialConsoleAccess = true
+
+	if err := s.saveSettings(ctx, settings, accountID); err != nil {
 		return nil, err
 	}
 
@@ -241,7 +196,14 @@ 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)
 
-	if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.SerialConsoleAccess = false }); err != nil {
+	settings, err := s.getSettings(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	settings.SerialConsoleAccess = false
+
+	if err := s.saveSettings(ctx, settings, accountID); err != nil {
 		return nil, err
 	}
 

spinifex/domains/ec2/account/service_impl_test.go

@@ -2,13 +2,11 @@ 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"
 )
@@ -145,44 +143,3 @@ 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/domains/network/host/imds_attach.go

@@ -23,6 +23,19 @@ func (p *OVSPlumber) DetachIMDSDatapath(eniID string) error {
 	return removeIMDSDatapath(context.Background(), NewExecRunner(), imdsDetachSpec(eniID))
 }
 
+// IMDSDatapathStates reports what survives of each primary ENI's datapath, so
+// the daemon can re-install one that an ovs-vswitchd restart or a kernel-module
+// reload took down.
+func (p *OVSPlumber) IMDSDatapathStates(eniIDs []string) (map[string]vm.IMDSDatapathState, error) {
+	return IMDSDatapathStates(context.Background(), NewExecRunner(), eniIDs)
+}
+
+// RepairIMDSDatapath re-installs a running guest's datapath. It returns
+// vm.ErrIMDSTapDetached, changing nothing, once the tap has left br-imds.
+func (p *OVSPlumber) RepairIMDSDatapath(eniID, mac, subnetID string) error {
+	return repairIMDSDatapath(context.Background(), NewExecRunner(), imdsDatapathSpec(eniID, mac, subnetID))
+}
+
 // EnsureIMDSDatapathBridge idempotently creates the dedicated IMDS bridge. The
 // daemon calls it before SetupTap places a primary tap on br-imds, since OVS
 // cannot add a port to a bridge that does not exist yet.
@@ -61,7 +74,7 @@ func imdsDetachSpec(eniID string) IMDSTapDatapath {
 // endpoint, patch, demux/egress flows, and reply routing. Stale flows are cleared
 // once here (all installers share the tap's cookie) before any flow is added.
 //
-// Bridge, cookie clear, endpoint, and patch are connectivity-critical (error bare):
+// Bridge, endpoint, cookie clear, and patch are connectivity-critical (error bare):
 // ListIMDSTaps advertises the tap off the patch and vpcd binds the endpoint, so the
 // endpoint must exist whenever the patch does, else vpcd binds a missing device every
 // reconcile forever. The flows and reply routing only serve, so theirs wrap
@@ -77,10 +90,12 @@ func installIMDSDatapath(ctx context.Context, r Runner, d IMDSTapDatapath) error
 	// The INPUT accept and the remap DNAT are wildcarded over every endpoint and
 	// installed once by vpcd at start. They are not attempted here: the daemon
 	// has no iptables grant and no CAP_NET_ADMIN, so trying aborts every launch.
-	if err := clearIMDSFlowsByCookie(ctx, r, imdsFlowCookie(d.Endpoint)); err != nil {
+	if err := ensureIMDSEndpoint(ctx, r, d); err != nil {
 		return err
 	}
-	if err := ensureIMDSEndpoint(ctx, r, d); err != nil {
+	// Cleared after the endpoint so a failed endpoint step leaves a repaired
+	// guest's surviving flows, and with them its connectivity, in place.
+	if err := clearIMDSFlowsByCookie(ctx, r, imdsFlowCookie(d.Endpoint)); err != nil {
 		return err
 	}
 	if err := installTapPatch(ctx, r, d); err != nil {

spinifex/domains/network/host/imds_bridge.go

@@ -7,8 +7,8 @@ import (
 )
 
 // IMDSBridge is the dedicated OVS bridge carrying the per-tap IMDS / VPC-DNS flows.
-// ovn-controller flushes foreign br-int flows wholesale on restart; a bridge absent
-// from ovn-bridge-mappings and not named br-int is never touched, so flows survive.
+// ovn-controller flushes foreign br-int flows on restart and never touches this
+// bridge. An ovs-vswitchd restart still drops its flows; the daemon re-installs them.
 const IMDSBridge = "br-imds"
 
 // imdsCookiePrefix tags every flow the IMDS datapath installs on IMDSBridge.

spinifex/domains/network/host/imds_drift.go

@@ -0,0 +1,98 @@
+package host
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"slices"
+	"strconv"
+	"strings"
+
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
+)
+
+// IMDSTapFlowCount is how many flows installIMDSDatapath installs under a tap's
+// cookie: an ARP responder and an ingress demux per captured address, the
+// egress flow, and the two tap<->patch forward flows.
+const IMDSTapFlowCount = 7
+
+// IMDSFlowCookie is the cookie every br-imds flow of eniID's datapath carries.
+func IMDSFlowCookie(eniID string) string {
+	return imdsFlowCookie(IMDSEndpointName(eniID))
+}
+
+// endpointSysctlDir is where the kernel exposes per-device IPv4 settings.
+var endpointSysctlDir = "/proc/sys/net/ipv4/conf"
+
+// IMDSDatapathStates reports what survives of each primary ENI's datapath. An
+// ovs-vswitchd restart drops every flow. A kernel-module reload re-creates the
+// endpoint, which resets its sysctls and drops its reply rule and route.
+func IMDSDatapathStates(ctx context.Context, r Runner, eniIDs []string) (map[string]vm.IMDSDatapathState, error) {
+	flows, err := r.Run(ctx, "ovs-ofctl", "--no-stats", "dump-flows", IMDSBridge)
+	if err != nil {
+		return nil, fmt.Errorf("dump %s flows: %w", IMDSBridge, err)
+	}
+	byCookie, err := countFlowsByCookie(flows)
+	if err != nil {
+		return nil, err
+	}
+	states := make(map[string]vm.IMDSDatapathState, len(eniIDs))
+	for _, eniID := range eniIDs {
+		endpoint := IMDSEndpointName(eniID)
+		cookie, err := strconv.ParseUint(imdsFlowCookie(endpoint), 0, 64)
+		if err != nil {
+			return nil, fmt.Errorf("parse IMDS cookie for %s: %w", eniID, err)
+		}
+		states[eniID] = vm.IMDSDatapathState{
+			FlowsFound:    byCookie[cookie],
+			FlowsExpected: IMDSTapFlowCount,
+			EndpointReady: endpointConfigured(endpoint),
+		}
+	}
+	return states, nil
+}
+
+// countFlowsByCookie tallies `ovs-ofctl dump-flows` lines by cookie. A flow
+// with a zero cookie is printed without the field and is not counted.
+func countFlowsByCookie(out []byte) (map[uint64]int, error) {
+	counts := map[uint64]int{}
+	sc := bufio.NewScanner(bytes.NewReader(out))
+	for sc.Scan() {
+		rest, ok := strings.CutPrefix(strings.TrimSpace(sc.Text()), "cookie=")
+		if !ok {
+			continue
+		}
+		val, _, _ := strings.Cut(rest, ",")
+		cookie, err := strconv.ParseUint(val, 0, 64)
+		if err != nil {
+			return nil, fmt.Errorf("parse flow cookie %q: %w", val, err)
+		}
+		counts[cookie]++
+	}
+	return counts, sc.Err()
+}
+
+// endpointConfigured reports whether endpoint carries the accept_local the
+// install sets. Read from /proc rather than through ip, which the daemon can
+// only run under sudo; the device's reply rule and route go with the setting.
+func endpointConfigured(endpoint string) bool {
+	val, err := os.ReadFile(filepath.Join(endpointSysctlDir, endpoint, "accept_local"))
+	return err == nil && strings.TrimSpace(string(val)) == "1"
+}
+
+// repairIMDSDatapath re-installs a tap's datapath after part of it was lost.
+// It refuses a tap that is no longer on br-imds: installing would re-create the
+// patch and endpoint of a guest being torn down, which vpcd and OVN then claim.
+func repairIMDSDatapath(ctx context.Context, r Runner, d IMDSTapDatapath) error {
+	out, err := r.Run(ctx, "ovs-vsctl", "list-ports", IMDSBridge)
+	if err != nil {
+		return fmt.Errorf("list %s ports: %w", IMDSBridge, err)
+	}
+	if !slices.Contains(strings.Fields(string(out)), d.Tap) {
+		return vm.ErrIMDSTapDetached
+	}
+	return installIMDSDatapath(ctx, r, d)
+}

spinifex/domains/network/host/imds_drift_test.go

@@ -0,0 +1,207 @@
+package host
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
+	hostcommand "github.com/mulgadc/spinifex/spinifex/runtime/host/command"
+)
+
+// dumpFlowLines renders n flows under eniID's cookie as `ovs-ofctl --no-stats
+// dump-flows` prints them.
+func dumpFlowLines(eniID string, n int) string {
+	var b strings.Builder
+	for range n {
+		b.WriteString(" cookie=" + imdsFlowCookie(IMDSEndpointName(eniID)) + ", priority=100,in_port=1 actions=output:3\n")
+	}
+	return b.String()
+}
+
+// withEndpointSysctls points endpointSysctlDir at a temp tree in which each
+// listed endpoint has accept_local set to the given value.
+func withEndpointSysctls(t *testing.T, acceptLocal map[string]string) {
+	t.Helper()
+	dir := t.TempDir()
+	for endpoint, val := range acceptLocal {
+		if err := os.MkdirAll(filepath.Join(dir, endpoint), 0o755); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(filepath.Join(dir, endpoint, "accept_local"), []byte(val+"\n"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+	orig := endpointSysctlDir
+	endpointSysctlDir = dir
+	t.Cleanup(func() { endpointSysctlDir = orig })
+}
+
+func TestIMDSDatapathStates(t *testing.T) {
+	const (
+		intact     = "eni-0intact"
+		partial    = "eni-0partial"
+		flushed    = "eni-0flushed"
+		reloaded   = "eni-0reloaded"
+		noEndpoint = "eni-0noendpoint"
+		departed   = "eni-0departed"
+	)
+	dump := "NXST_FLOW reply (xid=0x4):\n" +
+		dumpFlowLines(intact, IMDSTapFlowCount) +
+		dumpFlowLines(partial, 3) +
+		dumpFlowLines(reloaded, IMDSTapFlowCount) +
+		dumpFlowLines(noEndpoint, IMDSTapFlowCount) +
+		// A guest torn down on this node leaves no running ENI to claim these.
+		dumpFlowLines(departed, IMDSTapFlowCount) +
+		// A zero-cookie flow is printed without the field.
+		" priority=0 actions=drop\n"
+
+	s := newStubRunner()
+	s.expect("ovs-ofctl --no-stats dump-flows "+IMDSBridge, []byte(dump), nil)
+	withEndpointSysctls(t, map[string]string{
+		IMDSEndpointName(intact):   "1",
+		IMDSEndpointName(partial):  "1",
+		IMDSEndpointName(flushed):  "1",
+		IMDSEndpointName(reloaded): "0",
+	})
+
+	got, err := IMDSDatapathStates(context.Background(), s, []string{intact, partial, flushed, reloaded, noEndpoint})
+	if err != nil {
+		t.Fatalf("IMDSDatapathStates: %v", err)
+	}
+	want := map[string]vm.IMDSDatapathState{
+		intact:     {FlowsFound: IMDSTapFlowCount, FlowsExpected: IMDSTapFlowCount, EndpointReady: true},
+		partial:    {FlowsFound: 3, FlowsExpected: IMDSTapFlowCount, EndpointReady: true},
+		flushed:    {FlowsFound: 0, FlowsExpected: IMDSTapFlowCount, EndpointReady: true},
+		reloaded:   {FlowsFound: IMDSTapFlowCount, FlowsExpected: IMDSTapFlowCount, EndpointReady: false},
+		noEndpoint: {FlowsFound: IMDSTapFlowCount, FlowsExpected: IMDSTapFlowCount, EndpointReady: false},
+	}
+	if len(got) != len(want) {
+		t.Errorf("got %d states, want %d: %v", len(got), len(want), got)
+	}
+	for eni, w := range want {
+		if got[eni] != w {
+			t.Errorf("%s: got %+v, want %+v", eni, got[eni], w)
+		}
+	}
+	if !got[intact].Intact() {
+		t.Errorf("a complete datapath must read as intact: %+v", got[intact])
+	}
+	for _, eni := range []string{partial, flushed, reloaded, noEndpoint} {
+		if got[eni].Intact() {
+			t.Errorf("%s must not read as intact: %+v", eni, got[eni])
+		}
+	}
+}
+
+// A failed dump must surface, never read as "every ENI lost its flows" and
+// trigger a repair of every guest on the node.
+func TestIMDSDatapathStates_DumpFailure(t *testing.T) {
+	s := newStubRunner()
+	s.expect("ovs-ofctl", nil, errors.New("br-imds.mgmt: connection refused"))
+	if got, err := IMDSDatapathStates(context.Background(), s, []string{"eni-0abc"}); err == nil {
+		t.Fatalf("expected the dump failure to surface, got %v", got)
+	}
+}
+
+func TestCountFlowsByCookie_MalformedCookie(t *testing.T) {
+	if _, err := countFlowsByCookie([]byte(" cookie=0xzz, priority=1 actions=drop\n")); err == nil {
+		t.Fatal("expected a malformed cookie to be an error")
+	}
+}
+
+// IMDSTapFlowCount is what the drift check compares against, so it must match
+// what an install actually adds. A new flow without a bumped count would make
+// every healthy guest read as drifted and get re-installed every pass.
+func TestIMDSTapFlowCount_MatchesInstall(t *testing.T) {
+	s := newStubRunner()
+	s.expect("ovs-vsctl", nil, nil)
+	s.expect("ip", nil, nil)
+	s.expect(hostcommand.EndpointSysctlHelper, nil, nil)
+	s.expect("ovs-ofctl", nil, nil)
+
+	d := imdsDatapathSpec("eni-0abc1234", "02:00:00:00:01:05", "subnet-0fedcba9")
+	if err := installIMDSDatapath(context.Background(), s, d); err != nil {
+		t.Fatalf("installIMDSDatapath: %v", err)
+	}
+	cookie := imdsFlowCookie(d.Endpoint)
+	if n := countCalls(s.calls, "ovs-ofctl add-flow "+IMDSBridge+" cookie="+cookie+","); n != IMDSTapFlowCount {
+		t.Errorf("install added %d flows under the tap's cookie, IMDSTapFlowCount is %d", n, IMDSTapFlowCount)
+	}
+}
+
+func TestRepairIMDSDatapath(t *testing.T) {
+	const eniID = "eni-0abc1234"
+	d := imdsDatapathSpec(eniID, "02:00:00:00:01:05", "subnet-0fedcba9")
+	cookie := imdsFlowCookie(d.Endpoint)
+
+	t.Run("tap on br-imds is re-installed", func(t *testing.T) {
+		s := newStubRunner()
+		s.expect("ovs-vsctl list-ports "+IMDSBridge, []byte(d.Endpoint+"\n"+d.PatchIMDS+"\n"+d.Tap+"\n"), nil)
+		s.expect("ovs-vsctl --may-exist", nil, nil)
+		s.expect("ovs-vsctl set", nil, nil)
+		s.expect("ip", nil, nil)
+		s.expect(hostcommand.EndpointSysctlHelper, nil, nil)
+		s.expect("ovs-ofctl", nil, nil)
+
+		if err := repairIMDSDatapath(context.Background(), s, d); err != nil {
+			t.Fatalf("repairIMDSDatapath: %v", err)
+		}
+		if n := countCalls(s.calls, "ovs-ofctl add-flow "+IMDSBridge+" cookie="+cookie+","); n != IMDSTapFlowCount {
+			t.Errorf("repair added %d flows, want %d", n, IMDSTapFlowCount)
+		}
+		if !s.called("ip rule add oif " + d.Endpoint) {
+			t.Errorf("repair must restore the reply rule a kmod reload drops; calls: %v", s.calls)
+		}
+	})
+
+	// A tap gone from br-imds belongs to a guest being torn down. Re-creating
+	// its patch would leave an imi- port that vpcd binds and OVN claims.
+	t.Run("tap gone from br-imds is left alone", func(t *testing.T) {
+		s := newStubRunner()
+		s.expect("ovs-vsctl list-ports "+IMDSBridge, []byte(d.Endpoint+"\n"), nil)
+
+		err := repairIMDSDatapath(context.Background(), s, d)
+		if !errors.Is(err, vm.ErrIMDSTapDetached) {
+			t.Fatalf("want ErrIMDSTapDetached, got %v", err)
+		}
+		if len(s.calls) != 1 {
+			t.Errorf("a detached tap must not be touched; calls: %v", s.calls)
+		}
+	})
+
+	// After a kmod reload the flows survive and only the endpoint is lost; a
+	// repair that fails on the endpoint must not take the flows with it.
+	t.Run("endpoint failure keeps the surviving flows", func(t *testing.T) {
+		s := newStubRunner()
+		s.expect("ovs-vsctl list-ports "+IMDSBridge, []byte(d.Tap+"\n"), nil)
+		s.expect("ovs-vsctl --may-exist add-br", nil, nil)
+		s.expect("ovs-vsctl set", nil, nil)
+		s.expect("ip link set "+IMDSBridge, nil, nil)
+		s.expect("ovs-vsctl --may-exist add-port "+IMDSBridge+" "+d.Endpoint, nil, errors.New("ovsdb busy"))
+		s.expect("ovs-ofctl", nil, nil)
+
+		if err := repairIMDSDatapath(context.Background(), s, d); err == nil {
+			t.Fatal("want the endpoint failure to surface")
+		}
+		if s.called("ovs-ofctl del-flows") {
+			t.Errorf("flows cleared before the endpoint was ready; calls: %v", s.calls)
+		}
+	})
+
+	t.Run("port list failure installs nothing", func(t *testing.T) {
+		s := newStubRunner()
+		s.expect("ovs-vsctl list-ports", nil, errors.New("ovsdb unreachable"))
+
+		err := repairIMDSDatapath(context.Background(), s, d)
+		if err == nil || errors.Is(err, vm.ErrIMDSTapDetached) {
+			t.Fatalf("want a plain error, got %v", err)
+		}
+		if len(s.calls) != 1 {
+			t.Errorf("nothing may be installed without knowing the tap is attached; calls: %v", s.calls)
+		}
+	})
+}

spinifex/handlers/iam/account_status_reader.go

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

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

spinifex/operator/admin/image_import_test.go

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

spinifex/operator/cli/admin_kv_compare.go

@@ -32,12 +32,6 @@ A replica that is merely behind is not divergent, so only these fail:
   - replicas disagree on the replica count;
   - no sequence range is common to every replica, so nothing can be checked.
 
-A stream with a max_age whose replicas, copied at different times, share no
-sequence range is reported as skipped (volatile) rather than diverged: a short
-lived bucket such as a leader lease turns over between two copies. Sequences
-two of its replicas do share must still hold the same content, and each
-message a replica lacks must be superseded, expired, or written after its copy.
-
 Sequences past the slowest replica's last one are not compared. Digests are
 taken from live stores a few seconds apart, so re-run before trusting a
 failure that could be a write landing between two copies.
@@ -116,8 +110,6 @@ type streamVerdict struct {
 	Holders  []string
 	Problems []string
 	Notes    []string
-	// Volatile says why a stream that turned over between copies was not compared.
-	Volatile string
 }
 
 type kvCompareResult struct {
@@ -236,16 +228,6 @@ func compareStream(name string, hosts []string, views []*replicaView) streamVerd
 		}
 	}
 	if lo > hi {
-		maxAge, spread := views[0].d.MaxAge, copySpread(views)
-		if maxAge > 0 && spread > 0 {
-			mismatches, unexplained := turnoverFindings(views)
-			v.Problems = append(v.Problems, mismatches...)
-			if len(unexplained) == 0 {
-				v.Volatile = fmt.Sprintf("volatile: max_age %s and copies %s apart share no sequence range, so it turned over between them; not compared", maxAge, spread)
-				return v
-			}
-			v.Problems = append(v.Problems, unexplained...)
-		}
 		v.Problems = append(v.Problems, "no sequence range is common to every replica, so nothing could be compared")
 		return v
 	}
@@ -271,8 +253,18 @@ func compareStream(name string, hosts []string, views []*replicaView) streamVerd
 		if len(present) == 0 {
 			continue
 		}
-		if p := contentMismatch(seq, present); p != "" {
-			problems = append(problems, p)
+		byHash := map[string][]string{}
+		for _, r := range present {
+			h := r.bySeq[seq].Hash
+			byHash[h] = append(byHash[h], r.host)
+		}
+		if len(byHash) > 1 {
+			var parts []string
+			for h, hs := range byHash {
+				parts = append(parts, strings.Join(hs, ",")+"="+h)
+			}
+			sort.Strings(parts)
+			problems = append(problems, fmt.Sprintf("seq %d %s holds different content: %s", seq, present[0].bySeq[seq].Subject, strings.Join(parts, " ")))
 			continue
 		}
 		msg := present[0].bySeq[seq]
@@ -285,7 +277,10 @@ func compareStream(name string, hosts []string, views []*replicaView) streamVerd
 				seq, msg.Subject, hostsOf(present), r.host))
 		}
 	}
-	v.Problems = append(v.Problems, capFindings(problems)...)
+	if len(problems) > maxFindingsPerStream {
+		problems = append(problems[:maxFindingsPerStream], fmt.Sprintf("... and %d more", len(problems)-maxFindingsPerStream))
+	}
+	v.Problems = append(v.Problems, problems...)
 
 	note := fmt.Sprintf("seqs %d-%d compared", lo, hi)
 	if explained > 0 {
@@ -304,84 +299,6 @@ func compareStream(name string, hosts []string, views []*replicaView) streamVerd
 	return v
 }
 
-// contentMismatch describes seq holding different content on the replicas
-// present at it, or returns "" when they agree.
-func contentMismatch(seq uint64, present []*replicaView) string {
-	byHash := map[string][]string{}
-	for _, r := range present {
-		h := r.bySeq[seq].Hash
-		byHash[h] = append(byHash[h], r.host)
-	}
-	if len(byHash) < 2 {
-		return ""
-	}
-	var parts []string
-	for h, hs := range byHash {
-		parts = append(parts, strings.Join(hs, ",")+"="+h)
-	}
-	sort.Strings(parts)
-	return fmt.Sprintf("seq %d %s holds different content: %s", seq, present[0].bySeq[seq].Subject, strings.Join(parts, " "))
-}
-
-// turnoverFindings compares every sequence held by two or more replicas, since
-// no timing explains two values at one sequence, and lists each absence that
-// neither absenceExplained nor a write after that replica's copy accounts for.
-func turnoverFindings(views []*replicaView) (mismatches, unexplained []string) {
-	seqSet := map[uint64]struct{}{}
-	for _, r := range views {
-		for s := range r.bySeq {
-			seqSet[s] = struct{}{}
-		}
-	}
-	for _, seq := range slices.Sorted(maps.Keys(seqSet)) {
-		var present, absent []*replicaView
-		for _, r := range views {
-			if e, ok := r.bySeq[seq]; ok && e.Hash != deletedHash {
-				present = append(present, r)
-			} else {
-				absent = append(absent, r)
-			}
-		}
-		if len(present) == 0 {
-			continue
-		}
-		if p := contentMismatch(seq, present); p != "" {
-			mismatches = append(mismatches, p)
-			continue
-		}
-		msg := present[0].bySeq[seq]
-		for _, r := range absent {
-			if absenceExplained(r, msg) || (seq > r.d.LastSeq && msg.Time.After(r.copiedAt)) {
-				continue
-			}
-			unexplained = append(unexplained, fmt.Sprintf("seq %d %s is on %s but missing from %s, with no later write to that subject, expiry, limit or copy order to explain it",
-				seq, msg.Subject, hostsOf(present), r.host))
-		}
-	}
-	return capFindings(mismatches), capFindings(unexplained)
-}
-
-func capFindings(problems []string) []string {
-	if len(problems) > maxFindingsPerStream {
-		return append(problems[:maxFindingsPerStream], fmt.Sprintf("... and %d more", len(problems)-maxFindingsPerStream))
-	}
-	return problems
-}
-
-// copySpread is how far apart the earliest and latest replica copies were taken.
-func copySpread(views []*replicaView) time.Duration {
-	first, last := views[0].copiedAt, views[0].copiedAt
-	for _, r := range views[1:] {
-		if r.copiedAt.Before(first) {
-			first = r.copiedAt
-		}
-		if r.copiedAt.After(last) {
-			last = r.copiedAt
-		}
-	}
-	return last.Sub(first)
-}
-
 // absenceExplained reports whether a replica lacking msg removed it legitimately:
 // the subject was written again later, the message aged out before that replica
 // was copied, or a stream limit dropped it from a head the replica has moved past.
@@ -421,16 +338,12 @@ func writeCompareResult(w io.Writer, nodes []nodeDigest, res kvCompareResult) {
 		fmt.Fprintf(w, " %s@%s", n.Host, n.CopiedAt.Format(time.TimeOnly))
 	}
 	fmt.Fprintln(w)
-	bad, volatile := 0, 0
+	bad := 0
 	for _, v := range res.Streams {
 		status := "ok"
-		switch {
-		case len(v.Problems) > 0:
+		if len(v.Problems) > 0 {
 			status = "DIVERGED"
 			bad++
-		case v.Volatile != "":
-			status = "skipped"
-			volatile++
 		}
 		fmt.Fprintf(w, "%-9s %s  [%s]\n", status, v.Name, strings.Join(v.Holders, ","))
 		for _, p := range v.Problems {
@@ -439,13 +352,6 @@ func writeCompareResult(w io.Writer, nodes []nodeDigest, res kvCompareResult) {
 		for _, n := range v.Notes {
 			fmt.Fprintf(w, "      %s\n", n)
 		}
-		if v.Volatile != "" {
-			fmt.Fprintf(w, "      %s\n", v.Volatile)
-		}
 	}
-	fmt.Fprintf(w, "%d streams: %d consistent, %d diverged", len(res.Streams), len(res.Streams)-bad-volatile, bad)
-	if volatile > 0 {
-		fmt.Fprintf(w, ", %d skipped as volatile", volatile)
-	}
-	fmt.Fprintln(w)
+	fmt.Fprintf(w, "%d streams: %d consistent, %d diverged\n", len(res.Streams), len(res.Streams)-bad, bad)
 }

spinifex/operator/cli/admin_kv_compare_test.go

@@ -160,83 +160,6 @@ func TestKVCompareNoCommonRange(t *testing.T) {
 	assert.Contains(t, report, "no sequence range is common")
 }
 
-// A 60s lease bucket renewed every few seconds turns over between digests
-// taken a couple of seconds apart, so its replicas share no window.
-func TestKVCompareVolatileStream(t *testing.T) {
-	t1, t2, t3 := t0, t0.Add(2*time.Second), t0.Add(4*time.Second)
-	lease := func(maxAge time.Duration, msgs ...msg) stream {
-		return stream{name: "KV_bedrock-leader", replicas: 3, maxAge: maxAge, msgs: msgs}
-	}
-	early := kvs(msg{3984, "$KV.bedrock-leader.leader", "h3984", t1.Add(-time.Second)})
-	late := kvs(msg{3985, "$KV.bedrock-leader.leader", "h3985", t2.Add(-time.Second)}, msg{3986, "$KV.bedrock-leader.leader", "h3986", t2})
-
-	t.Run("adjacent windows of a max_age stream are skipped, not diverged", func(t *testing.T) {
-		divergent, report, err := cli.CompareKVDigests(
-			doc(t, "node1", t1, lease(time.Minute, early...)),
-			doc(t, "node2", t2, lease(time.Minute, late...)),
-			doc(t, "node3", t3, lease(time.Minute, late...)))
-		require.NoError(t, err)
-		assert.False(t, divergent, report)
-		assert.Contains(t, report, "skipped   KV_bedrock-leader")
-		assert.Contains(t, report, "volatile: max_age 1m0s")
-		assert.Contains(t, report, "1 streams: 0 consistent, 0 diverged, 1 skipped as volatile")
-	})
-
-	t.Run("a sequence two replicas share must still match", func(t *testing.T) {
-		mid := kvs(msg{3984, "$KV.bedrock-leader.leader", "other", t1.Add(-time.Second)}, late[0])
-		divergent, report, err := cli.CompareKVDigests(
-			doc(t, "node1", t1, lease(time.Minute, early...)),
-			doc(t, "node2", t2, lease(time.Minute, mid...)),
-			doc(t, "node3", t3, lease(time.Minute, late[1])))
-		require.NoError(t, err)
-		assert.True(t, divergent, report)
-		assert.Contains(t, report, "seq 3984 $KV.bedrock-leader.leader holds different content: node1=h3984 node2=other")
-	})
-
-	t.Run("a long-lived bucket with no common window is still diverged", func(t *testing.T) {
-		divergent, report, err := cli.CompareKVDigests(
-			doc(t, "node1", t1, lease(0, early...)),
-			doc(t, "node2", t2, lease(0, late...)),
-			doc(t, "node3", t3, lease(0, late...)))
-		require.NoError(t, err)
-		assert.True(t, divergent, report)
-		assert.Contains(t, report, "no sequence range is common")
-	})
-
-	t.Run("a long max_age stream whose missing messages nothing explains is still diverged", func(t *testing.T) {
-		old := kvs(msg{1, "a", "h1", t1.Add(-time.Minute)}, msg{2, "b", "h2", t1.Add(-time.Minute)})
-		newer := kvs(msg{10, "c", "h10", t1.Add(-time.Minute)}, msg{11, "d", "h11", t1.Add(-time.Minute)})
-		divergent, report, err := cli.CompareKVDigests(
-			doc(t, "node1", t1, lease(30*24*time.Hour, old...)),
-			doc(t, "node2", t2, lease(30*24*time.Hour, newer...)),
-			doc(t, "node3", t3, lease(30*24*time.Hour, newer...)))
-		require.NoError(t, err)
-		assert.True(t, divergent, report)
-		assert.Contains(t, report, "no sequence range is common")
-		assert.NotContains(t, report, "volatile")
-	})
-
-	t.Run("a replica copied last but holding the older window is still diverged", func(t *testing.T) {
-		divergent, report, err := cli.CompareKVDigests(
-			doc(t, "node1", t1, lease(time.Minute, late...)),
-			doc(t, "node2", t2, lease(time.Minute, late...)),
-			doc(t, "node3", t3, lease(time.Minute, early...)))
-		require.NoError(t, err)
-		assert.True(t, divergent, report)
-		assert.Contains(t, report, "seq 3985 $KV.bedrock-leader.leader is on node1,node2 but missing from node3")
-	})
-
-	t.Run("digests copied at the same instant are not excused", func(t *testing.T) {
-		divergent, report, err := cli.CompareKVDigests(
-			doc(t, "node1", t1, lease(time.Minute, early...)),
-			doc(t, "node2", t1, lease(time.Minute, late...)),
-			doc(t, "node3", t1, lease(time.Minute, late...)))
-		require.NoError(t, err)
-		assert.True(t, divergent, report)
-		assert.Contains(t, report, "no sequence range is common")
-	})
-}
-
 func TestKVCompareReplicaPlacement(t *testing.T) {
 	s := stream{name: "KV_a", replicas: 3, msgs: kvs(msg{1, "k", "h1", t0})}
 	other := stream{name: "KV_b", replicas: 1, msgs: kvs(msg{1, "k", "h1", t0})}

spinifex/providers/ebs/conformance/conformance.go

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

spinifex/providers/ebs/conformance/datapath.go

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

spinifex/providers/ebs/conformance/nbdclient.go

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

spinifex/providers/ebs/hostile/verbs.go

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

spinifex/providers/ebs/memory.go

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

spinifex/providers/ebs/memory_test.go

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

spinifex/providers/ebs/nats.go

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

spinifex/providers/ebs/nats_test.go

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

spinifex/providers/ebs/natsserve/handlers_test.go

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

spinifex/providers/ebs/natsserve/server.go

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

spinifex/providers/ebs/provider.go

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

spinifex/providers/ebs/provider_test.go

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

spinifex/providers/ebs/viperblock/provider_handlers.go

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

spinifex/providers/ebs/viperblock/provider_handlers_test.go

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

spinifex/runtime/compute/vm/devices.go

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

spinifex/runtime/compute/vm/ebs_requests.go

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

spinifex/runtime/compute/vm/imds_reconcile.go

@@ -0,0 +1,198 @@
+package vm
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+)
+
+// IMDSDatapathReconcileInterval is how often running guests' IMDS datapaths are
+// checked against what was installed. It bounds how long a guest stays dark
+// after ovs-vswitchd restarts. A var so tests can shorten it.
+var IMDSDatapathReconcileInterval = 10 * time.Second
+
+// imdsRepairConcurrency bounds how many datapaths one pass re-installs at once.
+// An ovs-vswitchd restart drops every guest's flows together, so serial repairs
+// would keep the last guest dark for the sum of everyone else's install.
+const imdsRepairConcurrency = 8
+
+// imdsDatapathTarget is a running guest whose primary ENI has an IMDS datapath,
+// copied out of the manager lock with everything a re-install needs.
+type imdsDatapathTarget struct {
+	instanceID string
+	eniID      string
+	mac        string
+	subnetID   string
+}
+
+// StartIMDSDatapathReconcile re-installs the IMDS datapath of any running guest
+// that has lost part of it. br-imds is fail-mode=secure, so a guest whose flows
+// are gone has no network at all. Runs once now, then every interval.
+func (m *Manager) StartIMDSDatapathReconcile(ctx context.Context) {
+	if m.deps.NetworkPlumber == nil {
+		return
+	}
+	go func() {
+		m.reconcileIMDSDatapaths()
+		ticker := time.NewTicker(IMDSDatapathReconcileInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				m.reconcileIMDSDatapaths()
+			}
+		}
+	}()
+}
+
+// reconcileIMDSDatapaths runs one pass: read every running guest's datapath in
+// one query, then repair each one that is not intact.
+func (m *Manager) reconcileIMDSDatapaths() {
+	targets := m.imdsDatapathTargets()
+	if len(targets) == 0 {
+		return
+	}
+	eniIDs := make([]string, len(targets))
+	for i, t := range targets {
+		eniIDs[i] = t.eniID
+	}
+	states, err := m.deps.NetworkPlumber.IMDSDatapathStates(eniIDs)
+	if err != nil {
+		slog.Warn("IMDS: datapath state read failed; skipping reconcile pass", "err", err)
+		return
+	}
+	sem := make(chan struct{}, imdsRepairConcurrency)
+	var wg sync.WaitGroup
+	for _, t := range imdsDatapathsToRepair(targets, states) {
+		sem <- struct{}{}
+		wg.Go(func() {
+			defer func() { <-sem }()
+			m.repairIMDSDatapath(t, states[t.eniID])
+		})
+	}
+	wg.Wait()
+}
+
+// imdsDatapathTargets returns the running guests attachPrimaryIMDSDatapath
+// gave a datapath: those with a primary ENI and a subnet.
+func (m *Manager) imdsDatapathTargets() []imdsDatapathTarget {
+	var targets []imdsDatapathTarget
+	m.View(func(vms map[string]*VM) {
+		for _, v := range vms {
+			if t, ok := imdsTargetOf(v); ok {
+				targets = append(targets, t)
+			}
+		}
+	})
+	return targets
+}
+
+// imdsTargetOf returns v as a repair target, or false if it should have no
+// datapath right now. The caller holds the manager lock.
+func imdsTargetOf(v *VM) (imdsDatapathTarget, bool) {
+	if v.Status != StateRunning || v.ENIId == "" || v.Instance == nil {
+		return imdsDatapathTarget{}, false
+	}
+	subnetID := aws.StringValue(v.Instance.SubnetId)
+	if subnetID == "" {
+		return imdsDatapathTarget{}, false
+	}
+	return imdsDatapathTarget{instanceID: v.ID, eniID: v.ENIId, mac: v.ENIMac, subnetID: subnetID}, true
+}
+
+// imdsDatapathsToRepair picks the targets whose datapath is not intact. An ENI
+// missing from states was not read, so it is left alone rather than repaired.
+func imdsDatapathsToRepair(targets []imdsDatapathTarget, states map[string]IMDSDatapathState) []imdsDatapathTarget {
+	var out []imdsDatapathTarget
+	for _, t := range targets {
+		if s, ok := states[t.eniID]; ok && !s.Intact() {
+			out = append(out, t)
+		}
+	}
+	return out
+}
+
+// repairIMDSDatapath re-installs one guest's datapath under its ENI lock, so it
+// cannot interleave with an attach or a teardown. The guest is re-checked under
+// the lock because a teardown may have finished since the pass read it.
+func (m *Manager) repairIMDSDatapath(t imdsDatapathTarget, s IMDSDatapathState) {
+	unlock := m.imdsLocks.lock(t.eniID)
+	defer unlock()
+
+	if !m.stillIMDSTarget(t) {
+		return
+	}
+	err := m.deps.NetworkPlumber.RepairIMDSDatapath(t.eniID, t.mac, t.subnetID)
+	switch {
+	case errors.Is(err, ErrIMDSTapDetached):
+		slog.Debug("IMDS: tap left br-imds before repair; skipping",
+			"instance", t.instanceID, "eni", t.eniID)
+	case err != nil:
+		slog.Error("IMDS: per-tap datapath repair failed; guest has no network until it succeeds",
+			"instance", t.instanceID, "eni", t.eniID, "flows_found", s.FlowsFound,
+			"flows_expected", s.FlowsExpected, "endpoint_ready", s.EndpointReady, "err", err)
+	default:
+		slog.Warn("IMDS: per-tap datapath was lost; re-installed",
+			"instance", t.instanceID, "eni", t.eniID, "flows_found", s.FlowsFound,
+			"flows_expected", s.FlowsExpected, "endpoint_ready", s.EndpointReady)
+	}
+}
+
+// stillIMDSTarget reports whether t is still the same running guest with the
+// same ENI, MAC and subnet the pass read.
+func (m *Manager) stillIMDSTarget(t imdsDatapathTarget) bool {
+	var ok bool
+	m.View(func(vms map[string]*VM) {
+		v, found := vms[t.instanceID]
+		if !found {
+			return
+		}
+		cur, isTarget := imdsTargetOf(v)
+		ok = isTarget && cur == t
+	})
+	return ok
+}
+
+// keyedMutex serialises work per key. Entries are dropped once unheld, so it
+// does not grow with every ENI the node has ever seen. The zero value is ready.
+type keyedMutex struct {
+	mu    sync.Mutex
+	locks map[string]*keyedLock
+}
+
+type keyedLock struct {
+	mu   sync.Mutex
+	refs int
+}
+
+// lock blocks until key is free and returns the function that releases it.
+func (k *keyedMutex) lock(key string) (unlock func()) {
+	k.mu.Lock()
+	if k.locks == nil {
+		k.locks = map[string]*keyedLock{}
+	}
+	l, ok := k.locks[key]
+	if !ok {
+		l = &keyedLock{}
+		k.locks[key] = l
+	}
+	l.refs++
+	k.mu.Unlock()
+
+	l.mu.Lock()
+	return func() {
+		l.mu.Unlock()
+		k.mu.Lock()
+		l.refs--
+		if l.refs == 0 {
+			delete(k.locks, key)
+		}
+		k.mu.Unlock()
+	}
+}

spinifex/runtime/compute/vm/imds_reconcile_test.go

@@ -0,0 +1,309 @@
+package vm
+
+import (
+	"errors"
+	"fmt"
+	"slices"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+)
+
+func imdsTestVM(id, eni string, status InstanceState) *VM {
+	return &VM{
+		ID:       id,
+		Status:   status,
+		ENIId:    eni,
+		ENIMac:   "02:00:00:00:00:01",
+		Instance: &ec2.Instance{SubnetId: aws.String("subnet-0abc")},
+	}
+}
+
+var (
+	imdsIntact  = IMDSDatapathState{FlowsFound: 7, FlowsExpected: 7, EndpointReady: true}
+	imdsFlushed = IMDSDatapathState{FlowsFound: 0, FlowsExpected: 7, EndpointReady: true}
+)
+
+func TestIMDSTargetOf(t *testing.T) {
+	noSubnet := imdsTestVM("i-nosubnet", "eni-1", StateRunning)
+	noSubnet.Instance.SubnetId = nil
+	noInstance := imdsTestVM("i-noinstance", "eni-2", StateRunning)
+	noInstance.Instance = nil
+
+	cases := []struct {
+		name string
+		v    *VM
+		want bool
+	}{
+		{"running with ENI and subnet", imdsTestVM("i-run", "eni-0", StateRunning), true},
+		{"stopping", imdsTestVM("i-stop", "eni-0", StateStopping), false},
+		{"error after a crash", imdsTestVM("i-err", "eni-0", StateError), false},
+		{"pending mid-launch", imdsTestVM("i-pend", "eni-0", StatePending), false},
+		{"no primary ENI", imdsTestVM("i-noeni", "", StateRunning), false},
+		{"no subnet", noSubnet, false},
+		{"no instance record", noInstance, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if _, got := imdsTargetOf(tc.v); got != tc.want {
+				t.Errorf("imdsTargetOf = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestIMDSDatapathsToRepair(t *testing.T) {
+	targets := []imdsDatapathTarget{
+		{instanceID: "i-intact", eniID: "eni-intact"},
+		{instanceID: "i-partial", eniID: "eni-partial"},
+		{instanceID: "i-flushed", eniID: "eni-flushed"},
+		{instanceID: "i-reloaded", eniID: "eni-reloaded"},
+		{instanceID: "i-unread", eniID: "eni-unread"},
+	}
+	states := map[string]IMDSDatapathState{
+		"eni-intact":   imdsIntact,
+		"eni-partial":  {FlowsFound: 3, FlowsExpected: 7, EndpointReady: true},
+		"eni-flushed":  imdsFlushed,
+		"eni-reloaded": {FlowsFound: 7, FlowsExpected: 7, EndpointReady: false},
+	}
+	got := imdsDatapathsToRepair(targets, states)
+	var ids []string
+	for _, t := range got {
+		ids = append(ids, t.eniID)
+	}
+	want := []string{"eni-partial", "eni-flushed", "eni-reloaded"}
+	if len(ids) != len(want) {
+		t.Fatalf("repair set = %v, want %v", ids, want)
+	}
+	for i := range want {
+		if ids[i] != want[i] {
+			t.Errorf("repair set = %v, want %v", ids, want)
+		}
+	}
+}
+
+// A re-install starts by clearing the tap's flows, so repairing an intact
+// datapath would cut a working guest's network every pass.
+func TestReconcileIMDSDatapaths_LeavesIntactGuestAlone(t *testing.T) {
+	plumber := &fakeNetworkPlumber{imdsStates: map[string]IMDSDatapathState{"eni-ok": imdsIntact}}
+	m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})
+	m.Insert(imdsTestVM("i-ok", "eni-ok", StateRunning))
+
+	m.reconcileIMDSDatapaths()
+
+	if len(plumber.imdsRepairCalls) != 0 {
+		t.Errorf("intact datapath was repaired: %+v", plumber.imdsRepairCalls)
+	}
+}
+
+func TestReconcileIMDSDatapaths_RepairsLostDatapath(t *testing.T) {
+	plumber := &fakeNetworkPlumber{imdsStates: map[string]IMDSDatapathState{
+		"eni-ok":   imdsIntact,
+		"eni-lost": imdsFlushed,
+		"eni-stop": imdsFlushed,
+	}}
+	m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})
+	m.Insert(imdsTestVM("i-ok", "eni-ok", StateRunning))
+	lost := imdsTestVM("i-lost", "eni-lost", StateRunning)
+	lost.ENIMac = "02:00:00:00:00:99"
+	lost.Instance.SubnetId = aws.String("subnet-0lost")
+	m.Insert(lost)
+	m.Insert(imdsTestVM("i-stop", "eni-stop", StateStopped))
+
+	m.reconcileIMDSDatapaths()
+
+	want := []imdsAttachCall{{eniID: "eni-lost", mac: "02:00:00:00:00:99", subnetID: "subnet-0lost"}}
+	if len(plumber.imdsRepairCalls) != 1 || plumber.imdsRepairCalls[0] != want[0] {
+		t.Errorf("repair calls = %+v, want %+v", plumber.imdsRepairCalls, want)
+	}
+}
+
+// A failed read says nothing about the datapaths, so it must not be taken as
+// every guest having lost its flows.
+func TestReconcileIMDSDatapaths_ReadFailureRepairsNothing(t *testing.T) {
+	plumber := &fakeNetworkPlumber{imdsStatesErr: errors.New("br-imds.mgmt: connection refused")}
+	m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})
+	m.Insert(imdsTestVM("i-a", "eni-a", StateRunning))
+
+	m.reconcileIMDSDatapaths()
+
+	if len(plumber.imdsRepairCalls) != 0 {
+		t.Errorf("read failure triggered repairs: %+v", plumber.imdsRepairCalls)
+	}
+}
+
+// The pass reads its targets before it repairs them; a guest stopped or gone
+// in between must not have its datapath re-created.
+func TestRepairIMDSDatapath_RechecksGuestUnderLock(t *testing.T) {
+	cases := []struct {
+		name   string
+		mutate func(m *Manager)
+	}{
+		{"stopped", func(m *Manager) { m.UpdateState("i-a", func(v *VM) { v.Status = StateStopping }) }},
+		{"removed", func(m *Manager) { m.Delete("i-a") }},
+		{"ENI replaced", func(m *Manager) { m.UpdateState("i-a", func(v *VM) { v.ENIId = "eni-b" }) }},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			plumber := &fakeNetworkPlumber{}
+			m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})
+			v := imdsTestVM("i-a", "eni-a", StateRunning)
+			m.Insert(v)
+			target, _ := imdsTargetOf(v)
+
+			tc.mutate(m)
+			m.repairIMDSDatapath(target, imdsFlushed)
+
+			if len(plumber.imdsRepairCalls) != 0 {
+				t.Errorf("repaired a guest no longer running with that ENI: %+v", plumber.imdsRepairCalls)
+			}
+		})
+	}
+}
+
+// Teardown holds the ENI lock across detach and tap cleanup; a repair must
+// wait for it rather than re-create what the teardown is removing.
+func TestRepairIMDSDatapath_WaitsForTeardownLock(t *testing.T) {
+	plumber := &fakeNetworkPlumber{}
+	m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})
+	v := imdsTestVM("i-a", "eni-a", StateRunning)
+	m.Insert(v)
+	target, _ := imdsTargetOf(v)
+
+	unlock := m.imdsLocks.lock("eni-a")
+	done := make(chan struct{})
+	go func() {
+		m.repairIMDSDatapath(target, imdsFlushed)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		t.Fatal("repair ran while the ENI lock was held")
+	case <-time.After(50 * time.Millisecond):
+	}
+	unlock()
+	<-done
+	if len(plumber.imdsRepairCalls) != 1 {
+		t.Errorf("repair calls after unlock = %d, want 1", len(plumber.imdsRepairCalls))
+	}
+}
+
+// A repair landing between detach and tap cleanup would find the tap still on
+// br-imds and re-create the endpoint and patch of a guest being torn down.
+func TestCleanupTapDevices_HoldsENILockAcrossDetachAndCleanup(t *testing.T) {
+	inDetach, releaseDetach := make(chan struct{}), make(chan struct{})
+	inCleanup, releaseCleanup := make(chan struct{}), make(chan struct{})
+	plumber := &fakeNetworkPlumber{
+		detachHook:  func() { close(inDetach); <-releaseDetach },
+		cleanupHook: func() { close(inCleanup); <-releaseCleanup },
+	}
+	m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})
+	v := imdsTestVM("i-a", "eni-a", StateRunning)
+	m.Insert(v)
+	target, _ := imdsTargetOf(v)
+
+	var wg sync.WaitGroup
+	wg.Go(func() { m.cleanupTapDevices(v) })
+	<-inDetach
+	wg.Go(func() { m.repairIMDSDatapath(target, imdsFlushed) })
+	for _, step := range []struct {
+		name    string
+		release chan struct{}
+		next    chan struct{}
+	}{{"detach", releaseDetach, inCleanup}, {"tap cleanup", releaseCleanup, nil}} {
+		time.Sleep(50 * time.Millisecond)
+		if n := plumber.repairCount(); n != 0 {
+			t.Fatalf("repair ran during %s: %d calls", step.name, n)
+		}
+		close(step.release)
+		if step.next != nil {
+			<-step.next
+		}
+	}
+	wg.Wait()
+
+	want := []string{"detach:eni-a", "cleanup:" + TapDeviceName("eni-a"), "repair:eni-a"}
+	if !slices.Equal(plumber.order, want) {
+		t.Errorf("call order = %v, want %v", plumber.order, want)
+	}
+}
+
+// An ovs-vswitchd restart drifts every guest at once; repairs must overlap so
+// the last guest is not dark for the sum of everyone else's install.
+func TestReconcileIMDSDatapaths_RepairsConcurrentlyUpToLimit(t *testing.T) {
+	const guests = imdsRepairConcurrency + 4
+	var inFlight atomic.Int32
+	var overLimit atomic.Bool
+	full, release := make(chan struct{}), make(chan struct{})
+	var fullOnce sync.Once
+	plumber := &fakeNetworkPlumber{
+		imdsStates: map[string]IMDSDatapathState{},
+		repairHook: func() {
+			n := inFlight.Add(1)
+			if n > imdsRepairConcurrency {
+				overLimit.Store(true)
+			}
+			if n == imdsRepairConcurrency {
+				fullOnce.Do(func() { close(full) })
+			}
+			<-release
+			inFlight.Add(-1)
+		},
+	}
+	m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})
+	for i := range guests {
+		eni := fmt.Sprintf("eni-%d", i)
+		plumber.imdsStates[eni] = imdsFlushed
+		m.Insert(imdsTestVM(fmt.Sprintf("i-%d", i), eni, StateRunning))
+	}
+
+	done := make(chan struct{})
+	go func() {
+		m.reconcileIMDSDatapaths()
+		close(done)
+	}()
+	select {
+	case <-full:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("repairs did not reach %d in flight: %d", imdsRepairConcurrency, inFlight.Load())
+	}
+	time.Sleep(50 * time.Millisecond)
+	if overLimit.Load() {
+		t.Errorf("more than %d repairs ran at once", imdsRepairConcurrency)
+	}
+	close(release)
+	<-done
+	if n := plumber.repairCount(); n != guests {
+		t.Errorf("repairs = %d, want %d", n, guests)
+	}
+}
+
+func TestRepairIMDSDatapath_DetachedTapIsNotAnError(t *testing.T) {
+	plumber := &fakeNetworkPlumber{imdsRepairErr: ErrIMDSTapDetached}
+	m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})
+	v := imdsTestVM("i-a", "eni-a", StateRunning)
+	m.Insert(v)
+	target, _ := imdsTargetOf(v)
+
+	m.repairIMDSDatapath(target, imdsFlushed)
+
+	if len(plumber.imdsRepairCalls) != 1 {
+		t.Errorf("repair calls = %d, want 1", len(plumber.imdsRepairCalls))
+	}
+}
+
+func TestKeyedMutex_DropsUnheldKeys(t *testing.T) {
+	var k keyedMutex
+	unlockA := k.lock("a")
+	unlockB := k.lock("b")
+	unlockA()
+	unlockB()
+	if len(k.locks) != 0 {
+		t.Errorf("unheld keys retained: %d", len(k.locks))
+	}
+}

spinifex/runtime/compute/vm/lifecycle.go

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

spinifex/runtime/compute/vm/lifecycle_test.go

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

spinifex/runtime/compute/vm/manager.go

@@ -96,6 +96,10 @@ type Manager struct {
 	vms         map[string]*VM
 	deps        Deps
 	goroutineWg sync.WaitGroup
+
+	// imdsLocks serialises each primary ENI's IMDS datapath attach, repair and
+	// teardown, which run outside mu.
+	imdsLocks keyedMutex
 }
 
 // NewManager returns a Manager with no collaborators wired. Tests that only

spinifex/runtime/compute/vm/manager_deps_test.go

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

spinifex/runtime/compute/vm/network_helpers.go

@@ -126,6 +126,8 @@ func (m *Manager) attachPrimaryIMDSDatapath(instance *VM) error {
 			"instance", instance.ID, "eni", instance.ENIId)
 		return nil
 	}
+	unlock := m.imdsLocks.lock(instance.ENIId)
+	defer unlock()
 	if err := m.deps.NetworkPlumber.AttachIMDSDatapath(instance.ENIId, instance.ENIMac, subnetID); err != nil {
 		slog.Error("IMDS: per-tap datapath install failed; aborting launch",
 			"instance", instance.ID, "eni", instance.ENIId, "err", err)

spinifex/runtime/compute/vm/network_helpers_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"reflect"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -13,10 +14,19 @@ import (
 
 // fakeNetworkPlumber records calls so tests can assert per-spec behaviour.
 type fakeNetworkPlumber struct {
+	mu                  sync.Mutex
+	order               []string // cleanup, detach and repair calls, in order
+	detachHook          func()
+	cleanupHook         func()
+	repairHook          func()
 	setupCalls          []TapSpec
 	cleanupCalls        []string
 	imdsAttachCalls     []imdsAttachCall
 	imdsDetachCalls     []string
+	imdsRepairCalls     []imdsAttachCall
+	imdsStates          map[string]IMDSDatapathState
+	imdsStatesErr       error
+	imdsRepairErr       error
 	hostPortCalls       []hostPortCall
 	hostPortRemoveCalls []string
 	ensureBridgeCalls   int
@@ -49,7 +59,13 @@ func (p *fakeNetworkPlumber) SetupTap(spec TapSpec) error {
 }
 
 func (p *fakeNetworkPlumber) CleanupTap(name string) error {
+	if p.cleanupHook != nil {
+		p.cleanupHook()
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	p.cleanupCalls = append(p.cleanupCalls, name)
+	p.order = append(p.order, "cleanup:"+name)
 	return p.cleanupErr
 }
 
@@ -59,10 +75,46 @@ func (p *fakeNetworkPlumber) AttachIMDSDatapath(eniID, mac, subnetID string) err
 }
 
 func (p *fakeNetworkPlumber) DetachIMDSDatapath(eniID string) error {
+	if p.detachHook != nil {
+		p.detachHook()
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	p.imdsDetachCalls = append(p.imdsDetachCalls, eniID)
+	p.order = append(p.order, "detach:"+eniID)
 	return p.imdsDetachErr
 }
 
+func (p *fakeNetworkPlumber) IMDSDatapathStates(eniIDs []string) (map[string]IMDSDatapathState, error) {
+	if p.imdsStatesErr != nil {
+		return nil, p.imdsStatesErr
+	}
+	out := make(map[string]IMDSDatapathState, len(eniIDs))
+	for _, id := range eniIDs {
+		if s, ok := p.imdsStates[id]; ok {
+			out[id] = s
+		}
+	}
+	return out, nil
+}
+
+func (p *fakeNetworkPlumber) RepairIMDSDatapath(eniID, mac, subnetID string) error {
+	if p.repairHook != nil {
+		p.repairHook()
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.imdsRepairCalls = append(p.imdsRepairCalls, imdsAttachCall{eniID: eniID, mac: mac, subnetID: subnetID})
+	p.order = append(p.order, "repair:"+eniID)
+	return p.imdsRepairErr
+}
+
+func (p *fakeNetworkPlumber) repairCount() int {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return len(p.imdsRepairCalls)
+}
+
 func (p *fakeNetworkPlumber) EnsureIMDSDatapathBridge() error {
 	p.ensureBridgeCalls++
 	return p.ensureBridgeErr

spinifex/runtime/compute/vm/network_plumber.go

@@ -11,6 +11,24 @@ import "errors"
 // was not set up and is equally fatal.
 var ErrIMDSServingDegraded = errors.New("imds: per-tap serving install failed (guest connectivity intact)")
 
+// ErrIMDSTapDetached is RepairIMDSDatapath declining a tap no longer on br-imds:
+// its guest is being torn down, and a repair would re-create ports it just lost.
+var ErrIMDSTapDetached = errors.New("imds: tap is no longer on the IMDS bridge")
+
+// IMDSDatapathState is what survives of a primary ENI's IMDS datapath: its
+// br-imds flows against the number installed, and whether its endpoint still
+// carries the settings the install gave it.
+type IMDSDatapathState struct {
+	FlowsFound    int
+	FlowsExpected int
+	EndpointReady bool
+}
+
+// Intact reports whether nothing the install put in place has been lost.
+func (s IMDSDatapathState) Intact() bool {
+	return s.FlowsFound >= s.FlowsExpected && s.EndpointReady
+}
+
 // NetworkPlumber handles tap device and OVS bridge operations. Defined in vm
 // so the manager avoids importing network/host. All methods must be idempotent.
 type NetworkPlumber interface {
@@ -32,6 +50,15 @@ type NetworkPlumber interface {
 	// so safe for an ENI that never had a datapath installed.
 	DetachIMDSDatapath(eniID string) error
 
+	// IMDSDatapathStates reports, per primary ENI, what survives of its IMDS
+	// datapath. One pass covers every ENI; an error means nothing was read.
+	IMDSDatapathStates(eniIDs []string) (map[string]IMDSDatapathState, error)
+
+	// RepairIMDSDatapath re-installs the IMDS datapath of a running guest whose
+	// state was lost. Returns ErrIMDSTapDetached, changing nothing, when the tap
+	// is no longer on br-imds.
+	RepairIMDSDatapath(eniID, mac, subnetID string) error
+
 	// EnsureVPCHostPort gives the host a routed presence in a managed VPC subnet
 	// by binding an ENI the daemon itself owns to an internal port on br-int.
 	// addr is CIDR notation carrying the ENI address at the SUBNET's prefix

spinifex/runtime/compute/vm/restore.go

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

spinifex/runtime/compute/vm/restore_test.go

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

spinifex/runtime/compute/vm/shutdown.go

@@ -730,11 +730,14 @@ func (m *Manager) shutdownQEMU(instance *VM) {
 func (m *Manager) cleanupTapDevices(instance *VM) {
 	if instance.ENIId != "" && m.deps.NetworkPlumber != nil {
 		// Detach the primary ENI's IMDS datapath before removing its tap, the
-		// inverse of the launch-time attach-after-SetupTap order.
+		// inverse of the launch-time attach-after-SetupTap order. Both run under
+		// the ENI's lock so a datapath repair cannot re-create what this removes.
+		unlock := m.imdsLocks.lock(instance.ENIId)
 		m.detachPrimaryIMDSDatapath(instance)
 		if err := m.deps.NetworkPlumber.CleanupTap(TapDeviceName(instance.ENIId)); err != nil {
 			slog.Warn("Failed to clean up tap device", "eni", instance.ENIId, "err", err)
 		}
+		unlock()
 		m.cleanupExtraENITaps(instance)
 	}
 

spinifex/runtime/compute/vm/volumes.go

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

spinifex/runtime/compute/vm/volumes_test.go

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

spinifex/runtime/roles/qemunbdd/live_lifecycle_test.go

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

spinifex/runtime/roles/qemunbdd/provider.go

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

spinifex/runtime/roles/qemunbdd/provider_test.go

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

tests/e2e/single/tests_test.go

@@ -109,6 +109,10 @@ func TestGuestChurnDurability(t *testing.T) {
 	runGuestChurnDurability(t, requireSingleNodeFixture(t))
 }
 
+func TestIMDSDatapathSurvivesVswitchdRestart(t *testing.T) {
+	runIMDSDatapathSurvivesVswitchdRestart(t, requireSingleNodeFixture(t))
+}
+
 func TestSnapshotLifecycle(t *testing.T) {
 	runSnapshotLifecycle(t, requireSingleNodeFixture(t))
 }

tests/e2e/single/vswitchdrestart_test.go

@@ -0,0 +1,151 @@
+//go:build e2e
+
+package single
+
+import (
+	"context"
+	"fmt"
+	"os/exec"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/domains/network/host"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/require"
+)
+
+// vswitchdRecoveryBudget bounds how long a guest may stay dark after
+// ovs-vswitchd restarts: one daemon reconcile interval, the re-install, and
+// ovn-controller re-pushing br-int.
+const vswitchdRecoveryBudget = 60 * time.Second
+
+// openvswitchUnits are the systemd units that run ovs-vswitchd, by distro
+// family: Debian/Ubuntu first, then RHEL/Oracle Linux.
+var openvswitchUnits = []string{"openvswitch-switch.service", "openvswitch.service"}
+
+// runIMDSDatapathSurvivesVswitchdRestart restarts ovs-vswitchd under a running
+// guest and requires the guest's EIP, its IMDS and its br-imds flows to come
+// back. ovs-vswitchd holds flows only in memory, and br-imds is fail-mode=secure,
+// so before the daemon re-installed them every guest on the node went dark.
+//
+// It takes the whole node's datapath down, so it must never run in parallel.
+func runIMDSDatapathSurvivesVswitchdRestart(t *testing.T, fix *Fixture) {
+	if !fix.PublicPool {
+		t.Skip("needs a pool of public addresses to allocate an Elastic IP from")
+	}
+	harness.SkipIfNoOVN(t)
+	harness.Phase(t, "Single — a guest keeps its network and IMDS across an ovs-vswitchd restart")
+
+	c := fix.AWS
+	unit := openvswitchUnit(t)
+	instType, _ := needInstanceTypeArch(t, fix)
+	keyName, keyPath := needKeyPair(t, fix)
+	amiID := needAMI(t, fix)
+	vpc := harness.EnsureDefaultVPC(t, fix.Harness)
+	require.NotEmpty(t, vpc.SGID, "default SG ID required")
+	harness.AuthorizeSSHIngress(t, c, vpc.SGID)
+
+	// Whatever happens below, the node must not be left without ovs-vswitchd.
+	t.Cleanup(func() {
+		_ = exec.Command("sudo", "-n", "systemctl", "start", unit).Run()
+	})
+
+	harness.Step(t, "launch guest")
+	instanceID := launchBaselineInstance(t, fix, amiID, instType, keyName, vpc.SubnetID, []string{vpc.SGID})
+
+	harness.Step(t, "allocate-address and associate it with %s", instanceID)
+	eipOut, err := c.EC2.AllocateAddress(&ec2.AllocateAddressInput{Domain: aws.String("vpc")}) // e2e:allow-create — the Elastic IP is the ingress path that must survive the restart.
+	require.NoError(t, err, "allocate-address")
+	allocID, eipIP := aws.StringValue(eipOut.AllocationId), aws.StringValue(eipOut.PublicIp)
+	require.NotEmpty(t, eipIP, "EIP PublicIp empty")
+	t.Cleanup(func() {
+		_, _ = c.EC2.ReleaseAddress(&ec2.ReleaseAddressInput{AllocationId: aws.String(allocID)})
+	})
+	assocOut, err := c.EC2.AssociateAddress(&ec2.AssociateAddressInput{
+		AllocationId: aws.String(allocID),
+		InstanceId:   aws.String(instanceID),
+	})
+	require.NoError(t, err, "associate-address")
+	t.Cleanup(func() {
+		_, _ = c.EC2.DisassociateAddress(&ec2.DisassociateAddressInput{AssociationId: assocOut.AssociationId})
+	})
+	if !sshHealth.TryReady(eipIP, 22, keyPath, harness.SSHReadyBudget) {
+		t.Fatalf("guest unreachable via EIP %s within %s", eipIP, harness.SSHReadyBudget)
+	}
+	tgt := harness.SSHTarget{User: "ubuntu", Host: eipIP, Port: 22, KeyPath: keyPath}
+
+	cookie := host.IMDSFlowCookie(primaryENI(t, harness.WaitForInstanceState(t, c, instanceID, "running")))
+	require.NoError(t, guestIMDSAnswers(t, tgt, instanceID), "IMDS before any restart")
+	require.Equal(t, host.IMDSTapFlowCount, guestIMDSFlowCount(t, cookie), "guest's br-imds flows before any restart")
+	harness.Detail(t, "eip", eipIP, "unit", unit, "cookie", cookie)
+
+	for _, disruption := range []struct {
+		name string
+		argv []string
+	}{
+		{"systemctl restart " + unit, []string{"systemctl", "restart", unit}},
+		// systemd restarts it; nothing runs the scripted flow save/restore.
+		{"SIGKILL ovs-vswitchd", []string{"pkill", "-9", "-x", "ovs-vswitchd"}},
+	} {
+		harness.Step(t, "%s", disruption.name)
+		start := time.Now()
+		out, err := exec.Command("sudo", append([]string{"-n"}, disruption.argv...)...).CombinedOutput()
+		require.NoErrorf(t, err, "%s: %s", disruption.name, strings.TrimSpace(string(out)))
+
+		// One condition, so the time it reports is when everything was back.
+		harness.EventuallyErr(t, func() error {
+			if err := guestIMDSAnswers(t, tgt, instanceID); err != nil {
+				return err
+			}
+			if n := guestIMDSFlowCount(t, cookie); n != host.IMDSTapFlowCount {
+				return fmt.Errorf("guest has %d br-imds flows, want %d", n, host.IMDSTapFlowCount)
+			}
+			return nil
+		}, vswitchdRecoveryBudget, 2*time.Second)
+		harness.Detail(t, "disruption", disruption.name, "recovered_ms", time.Since(start).Milliseconds())
+	}
+}
+
+// guestIMDSAnswers reaches the guest over its EIP and has it fetch its own
+// instance ID through IMDSv2, which crosses both halves of the br-imds datapath.
+func guestIMDSAnswers(t *testing.T, tgt harness.SSHTarget, instanceID string) error {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second)
+	defer cancel()
+	const cmd = `T=$(curl -sf -m3 -X PUT -H "X-aws-ec2-metadata-token-ttl-seconds: 60" http://169.254.169.254/latest/api/token) && ` +
+		`curl -sf -m3 -H "X-aws-ec2-metadata-token: $T" http://169.254.169.254/latest/meta-data/instance-id`
+	out, err := harness.RunGuestSSH(ctx, tgt, cmd)
+	if err != nil {
+		return err
+	}
+	if got := strings.TrimSpace(string(out)); got != instanceID {
+		return fmt.Errorf("IMDS instance-id = %q, want %q", got, instanceID)
+	}
+	return nil
+}
+
+// guestIMDSFlowCount returns how many br-imds flows carry the guest's cookie, or
+// -1 while ovs-vswitchd is not answering. Other guests' teardowns cannot skew it.
+func guestIMDSFlowCount(t *testing.T, cookie string) int {
+	t.Helper()
+	out, err := exec.Command("sudo", "-n", "ovs-ofctl", "--no-stats", "dump-flows", host.IMDSBridge, "cookie="+cookie+"/-1").Output()
+	if err != nil {
+		return -1
+	}
+	return strings.Count(string(out), "cookie=")
+}
+
+// openvswitchUnit returns the unit that runs ovs-vswitchd on this host.
+func openvswitchUnit(t *testing.T) string {
+	t.Helper()
+	for _, unit := range openvswitchUnits {
+		if exec.Command("systemctl", "cat", unit).Run() == nil {
+			return unit
+		}
+	}
+	t.Fatalf("none of %v is installed", openvswitchUnits)
+	return ""
+}