feat(uefi): add the instance UEFI variable store pointer, store records and their bucket (#1499)

InstanceStatus and VM carry an optional UEFIVarStoreRef naming the store
object and the firmware it was written under. The new uefistore package
holds FirmwareIdentity, UEFIStoreRecord and UEFIWriteBack, and a Store over
the spinifex-uefi-stores bucket (history 1, no TTL, schema 1) whose writes
are all revision-guarded and validated. The daemon creates the bucket at
startup. Nothing reads or writes either yet.

contracts/ec2/v1/README.md

@@ -37,13 +37,6 @@ contract or an approved compatibility plan; it is not a package refactor.
 Only the subject is owned here today.
 The request (`account_id`) and reply (`vpc_id`, `error`) bodies are still declared as anonymous structs on each side; moving them into this package is a contract change that needs its own characterization first.
 
-## Volume held elsewhere
-
-A start refused because another node holds one of the instance's volumes is an `IncorrectState` error payload on the `ec2.start`, `ec2.start.<node>` and `ec2.cmd.<instance-id>` replies, and the AWS body cannot say why.
-The owning node adds the `X-Volume-Held-Elsewhere: true` header to that reply, the `ec2.start` forwarder relays it, and `StartInstances` wraps `ErrVolumeHeldElsewhere` around the error it rebuilds, so a caller matches it with `errors.Is` rather than reading the message.
-The owning node sets the header only from the EBS provider's `volume_held_elsewhere` code, which every provider returns for this refusal; no reader matches the message text.
-`VolumeHeldElsewhereText` is the wording viperblockd puts in that refusal's message, not a signal.
-
 ## Platform inventory
 
 `PlatformInventorySubject` (`ec2.v1.PlatformInventory`) is a request/reply route on which a daemon in the `spinifex-workers` queue group reports one account's key pairs, placement groups, launch templates and their `ec2-tags` entries from the EC2 owner stores.

contracts/ec2/v1/volume_held.go

@@ -1,13 +1,7 @@
 package ec2v1
 
-import "errors"
-
-// VolumeHeldElsewhereText is the message viperblockd gives a lease refusal. It
-// is wording for the operator; callers classify the refusal by its code.
+// VolumeHeldElsewhereText is the text of a start refused because another node
+// holds one of the instance's volumes. It crosses NATS as message text, so
+// readers match on it; the provider that refuses and every reader share this
+// string so they cannot drift.
 const VolumeHeldElsewhereText = "volume is leased by another owner"
-
-// ErrVolumeHeldElsewhere marks a start refused because another node still
-// holds one of the instance's volumes. The owning node sets it from the EBS
-// provider's held-elsewhere code, and a start reply carries it across NATS as
-// the X-Volume-Held-Elsewhere header rather than as message text.
-var ErrVolumeHeldElsewhere = errors.New("a volume is leased by another node")

contracts/viperblockd/legacy/v1/README.md

@@ -51,19 +51,6 @@ request/reply replies.
 removed its local state and export. These routes and JSON shapes are deployed
 legacy compatibility behaviour, not a generic EBS-provider interface.
 
-## Fence route
-
-`VolumeFencedSubject(node)` is `ebs.<node>.fenced`, or `ebs.fenced` with no
-node for a single-node daemon. The Viperblock adapter publishes
-`VolumeFencedEvent` on it after tearing down an export whose volume lease
-moved to another node; the local daemon subscribes to stop the guest that
-was using it. Node-addressed, like mount and unmount: the guest is local, and
-no other node's daemon has anything to stop.
-
-The event is sent as a request and repeated until the daemon replies with `VolumeFencedAck`, because the fence often runs while the node's NATS server is restarting and a single publish can reach no subscriber.
-The daemon replies only when the event carries a reply subject, so a plain publish is still accepted.
-Repeats are harmless, since fencing a guest that is already fenced changes nothing.
-
 ## Abandon route
 
 `VolumeAbandonSubject(node)` is `ebs.<node>.abandon`. The daemon uses

contracts/viperblockd/legacy/v1/abandon_test.go

@@ -9,9 +9,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// TestVolumeAbandonSubject_BothForms pins the abandon route the same way the
-// fenced route is pinned: node-addressed, with an empty-node fallback for a
-// single-node daemon.
+// TestVolumeAbandonSubject_BothForms pins the abandon route: node-addressed,
+// with an empty-node fallback for a single-node daemon.
 func TestVolumeAbandonSubject_BothForms(t *testing.T) {
 	assert.Equal(t, "ebs.node-a.abandon", viperblocklegacyv1.VolumeAbandonSubject("node-a"))
 	assert.Equal(t, "ebs.abandon", viperblocklegacyv1.VolumeAbandonSubject(""))

contracts/viperblockd/legacy/v1/fence.go

@@ -1,29 +0,0 @@
-package viperblocklegacyv1
-
-// VolumeFencedSubject is where a fenced volume is announced so the daemon on
-// this node can stop the guest that was using it. Node-addressed: the guest is
-// local, and every other node's daemon has nothing to do about it.
-func VolumeFencedSubject(node string) string {
-	if node == "" {
-		return "ebs.fenced"
-	}
-	return "ebs." + node + ".fenced"
-}
-
-// VolumeFencedEvent announces that a node tore down a volume's export because
-// the lease moved. Winner is who holds it now, for the operator reading the
-// guest's stop reason.
-type VolumeFencedEvent struct {
-	Volume string `json:"volume"`
-	Node   string `json:"node"`
-	Winner string `json:"winner"`
-	Reason string `json:"reason"`
-}
-
-// VolumeFencedAck is the daemon's answer to a VolumeFencedEvent. Any answer
-// ends the sender's retries; Instance names the guest stopped, if one was, and
-// Error says why the daemon could not act on the event.
-type VolumeFencedAck struct {
-	Instance string `json:"instance,omitempty"`
-	Error    string `json:"error,omitempty"`
-}

contracts/viperblockd/legacy/v1/fence_test.go

@@ -1,54 +0,0 @@
-package viperblocklegacyv1_test
-
-import (
-	"encoding/json"
-	"testing"
-
-	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-// TestVolumeFencedSubject_IsNodeAddressed pins the routing. A fenced guest is on
-// the node that lost the volume, so a subject without the node in it would reach
-// daemons with nothing to stop and, on a queue group, miss the one that has.
-func TestVolumeFencedSubject_IsNodeAddressed(t *testing.T) {
-	assert.Equal(t, "ebs.node-a.fenced", viperblocklegacyv1.VolumeFencedSubject("node-a"))
-	assert.Equal(t, "ebs.fenced", viperblocklegacyv1.VolumeFencedSubject(""),
-		"a single-node daemon has no node name, and still has to hear its own fences")
-}
-
-// TestVolumeFencedEvent_JSON pins the exact field spelling of the fenced
-// event, published by viperblockd and consumed by the daemon.
-func TestVolumeFencedEvent_JSON(t *testing.T) {
-	golden := `{"volume":"vol-1","node":"node-a","winner":"node-b","reason":"volume lease moved to node-b while this node had the volume open"}`
-	event := viperblocklegacyv1.VolumeFencedEvent{
-		Volume: "vol-1",
-		Node:   "node-a",
-		Winner: "node-b",
-		Reason: "volume lease moved to node-b while this node had the volume open",
-	}
-
-	marshaled, err := json.Marshal(event)
-	require.NoError(t, err)
-	assert.JSONEq(t, golden, string(marshaled))
-
-	var decoded viperblocklegacyv1.VolumeFencedEvent
-	require.NoError(t, json.Unmarshal([]byte(golden), &decoded))
-	assert.Equal(t, event, decoded)
-}
-
-// TestVolumeFencedAck_JSON pins the daemon's answer, which is what ends
-// viperblockd's retries of a fence notice.
-func TestVolumeFencedAck_JSON(t *testing.T) {
-	golden := `{"instance":"i-1","error":"bad event"}`
-	ack := viperblocklegacyv1.VolumeFencedAck{Instance: "i-1", Error: "bad event"}
-
-	marshaled, err := json.Marshal(ack)
-	require.NoError(t, err)
-	assert.JSONEq(t, golden, string(marshaled))
-
-	empty, err := json.Marshal(viperblocklegacyv1.VolumeFencedAck{})
-	require.NoError(t, err)
-	assert.JSONEq(t, `{}`, string(empty), "an ack for a volume with no local guest carries nothing")
-}

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

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"log/slog"
 	"time"
@@ -103,7 +102,7 @@ func startLiveInstance(ctx context.Context, natsConn *nats.Conn, instanceID, acc
 
 	if responseError, parseErr := awserrors.ValidateErrorPayload(msg.Data); parseErr != nil {
 		slog.ErrorContext(ctx, "StartInstances: owner returned error", "instance_id", instanceID, "code", *responseError.Code)
-		return nil, false, startRefusalError(responseError, msg.Header)
+		return nil, false, startRefusalError(responseError)
 	}
 
 	slog.InfoContext(ctx, "StartInstances: restarted via owner node", "instance_id", instanceID, "response", string(msg.Data))
@@ -136,25 +135,18 @@ func startStoppedInstance(ctx context.Context, natsConn *nats.Conn, instanceID,
 
 	if responseError, parseErr := awserrors.ValidateErrorPayload(msg.Data); parseErr != nil {
 		slog.ErrorContext(ctx, "StartInstances: Daemon returned error", "instance_id", instanceID, "code", *responseError.Code)
-		return nil, startRefusalError(responseError, msg.Header)
+		return nil, startRefusalError(responseError)
 	}
 
 	slog.InfoContext(ctx, "StartInstances: Command sent successfully", "instance_id", instanceID, "response", string(msg.Data))
 	return newStateChange(instanceID, 0, "pending", 80, "stopped"), nil
 }
 
-// startRefusalError rebuilds the daemon's error, keeping the message it sent,
-// which names the node holding the volume, and wrapping ErrVolumeHeldElsewhere
-// when the reply's header says that is why the start was refused.
-func startRefusalError(responseError ec2.ResponseError, header nats.Header) error {
-	err := startRefusalCodeError(responseError)
-	if header.Get(natsmsg.VolumeHeldElsewhereHeader) == "true" {
-		return fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere, err)
-	}
-	return err
-}
-
-func startRefusalCodeError(responseError ec2.ResponseError) error {
+// startRefusalError rebuilds the daemon's error, keeping the message it sent.
+// Reconstructing from the code alone renders the generic ErrorLookup text
+// instead, which replaces the name of the node holding the volume with
+// boilerplate about attaching volumes -- the one fact the operator needed.
+func startRefusalError(responseError ec2.ResponseError) error {
 	var code, message string
 	if responseError.Code != nil {
 		code = *responseError.Code

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

@@ -3,14 +3,11 @@ package instance
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -164,33 +161,3 @@ func TestStartInstances_MixedSuccessAndFailure(t *testing.T) {
 	require.Error(t, err)
 	assert.Equal(t, "InvalidInstanceID.NotFound", err.Error())
 }
-
-// The AWS body says only IncorrectState, so the header is the one thing that
-// tells a held volume apart from any other refusal in that code.
-func TestStartInstances_VolumeHeldElsewhereHeader(t *testing.T) {
-	t.Parallel()
-	for name, held := range map[string]bool{"with header": true, "without header": false} {
-		t.Run(name, func(t *testing.T) {
-			t.Parallel()
-			_, nc := startTestNATSServer(t)
-			const id = "i-0abc0000000000002"
-			_, err := nc.Subscribe(ec2v1.InstanceCommandSubject(id), func(msg *nats.Msg) {
-				reply := nats.NewMsg(msg.Reply)
-				reply.Data = awserrors.GenerateErrorPayloadWithMessage(awserrors.ErrorIncorrectState, "volume vol-1 is published to node-a")
-				if held {
-					reply.Header.Set(natsmsg.VolumeHeldElsewhereHeader, "true")
-				}
-				_ = msg.RespondMsg(reply)
-			})
-			require.NoError(t, err)
-
-			_, err = StartInstances(t.Context(), &ec2.StartInstancesInput{InstanceIds: []*string{aws.String(id)}}, nc, "123456789012")
-			require.Error(t, err)
-			assert.Equal(t, held, errors.Is(err, ec2v1.ErrVolumeHeldElsewhere))
-			code, message, ok := awserrors.ResolveErrorDetail(err)
-			require.True(t, ok)
-			assert.Equal(t, awserrors.ErrorIncorrectState, code)
-			assert.Equal(t, "volume vol-1 is published to node-a", message)
-		})
-	}
-}

spinifex/domains/ec2/instance/service_impl.go

@@ -1237,12 +1237,7 @@ func (s *InstanceServiceImpl) StartInstance(ctx context.Context, instance *vm.VM
 				"instanceId", command.ID, "err", err)
 			return errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
 		default:
-			// A volume held elsewhere is the caller's 400, not a fault on this node.
-			if refusal := volumeHeldElsewhereError(err); refusal != nil {
-				slog.WarnContext(ctx, "StartInstance: vmMgr.Start failed", "instanceId", command.ID, "err", err)
-				return refusal
-			}
-			slog.ErrorContext(ctx, "StartInstance: vmMgr.Start failed", "instanceId", command.ID, "err", err)
+			slog.ErrorContext(ctx, "StartInstance: vmMgr.Start failed", "err", err)
 			return errors.New(awserrors.ErrorServerInternal)
 		}
 	}
@@ -2843,15 +2838,40 @@ func (s *InstanceServiceImpl) RecoverInstance(ctx context.Context, instance *vm.
 	return s.launchClaimedInstance(ctx, instance, undo)
 }
 
-// volumeHeldElsewhereError converts a mount the EBS provider refused with its
-// held-elsewhere code into IncorrectState naming the holder, wrapping
-// ec2v1.ErrVolumeHeldElsewhere, and returns nil for anything else.
+// volumeExclusionMarkers are the refusals that mean another node is writing this
+// volume right now, or is still live and holds its unsealed writes. A previous
+// owner that has stopped heartbeating is opened over with a warning instead. The
+// text arrives having crossed NATS from viperblockd.
+var volumeExclusionMarkers = []string{
+	ec2v1.VolumeHeldElsewhereText,
+}
+
+// ErrVolumeHeldElsewhere reports a launch refused because another node still
+// holds one of the instance's volume leases.
+//
+// Recovery treats it differently from every other failure — it clears on the
+// lease's own clock rather than needing anything fixed — so it has to be
+// recognisable with errors.Is rather than inferred from text that has already
+// crossed NATS once.
+var ErrVolumeHeldElsewhere = errors.New("a volume is leased by another node")
+
+// volumeHeldElsewhereError converts a mount refused because another node holds
+// the volume into a client-facing error carrying that reason. Returns nil for
+// anything else. Without it the caller gets ServerInternal and is told to
+// retry, which cannot succeed while the holder is down, and the text naming the
+// holder never leaves the node's log.
 func volumeHeldElsewhereError(err error) error {
-	if !errors.Is(err, ebsprovider.ErrVolumeHeldElsewhere) {
+	if err == nil {
 		return nil
 	}
-	return fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere,
-		awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", err.Error()))
+	msg := err.Error()
+	for _, marker := range volumeExclusionMarkers {
+		if strings.Contains(msg, marker) {
+			return fmt.Errorf("%w: %w", ErrVolumeHeldElsewhere,
+				awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", msg))
+		}
+	}
+	return nil
 }
 
 // restoreClaimedStoppedInstance writes instance back to the shared

spinifex/domains/ec2/instance/uefistore/uefistore.go

@@ -0,0 +1,185 @@
+// Package uefistore holds the types that describe an instance's UEFI variable
+// store and the bucket of store records that carry its deletion and write-back
+// obligations past the life of the instance record.
+package uefistore
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+const (
+	// BucketName holds one store record per instance, keyed by instance ID.
+	BucketName = "spinifex-uefi-stores"
+	// BucketVersion is the bucket's schema stamp.
+	BucketVersion = 1
+	// RecordSchemaVersion is the version every record is written at.
+	RecordSchemaVersion uint16 = 1
+
+	// StateLive is a store whose objects belong to a running or stopped instance.
+	StateLive = "live"
+	// StateDeleting is a store whose objects are to be deleted, and then the record.
+	StateDeleting = "deleting"
+
+	objectRoot = "uefi/"
+)
+
+// ErrInvalidRecord is returned for a store record that breaks its invariants,
+// whether it is about to be written or was just read.
+var ErrInvalidRecord = errors.New("uefistore: invalid store record")
+
+// FirmwareIdentity names the firmware a variable store was written under. A
+// store only boots where Arch and VarsSize match the node's firmware, because
+// pflash needs a byte-exact region; CodeSHA256 is recorded, not enforced.
+type FirmwareIdentity struct {
+	Arch       string `json:"arch"`
+	VarsSize   int64  `json:"vars_size"`
+	CodeSHA256 string `json:"code_sha256"`
+}
+
+// UEFIStoreRecord is the durable owner of one instance's variable store
+// objects. It outlives the instance record, so an obligation recorded here is
+// not lost when a terminated record expires.
+type UEFIStoreRecord struct {
+	SchemaVersion uint16         `json:"schema_version"`
+	InstanceID    string         `json:"instance_id"`
+	AccountID     string         `json:"account_id"`
+	State         string         `json:"state"`
+	WriteBack     *UEFIWriteBack `json:"write_back,omitempty"`
+}
+
+// UEFIWriteBack is a write-back that failed and is still owed. Key is the
+// object every retry puts to, so retries leave no extra objects; SeededFrom is
+// the pointer Node seeded from, which a retry must still find on the record.
+type UEFIWriteBack struct {
+	Node       string    `json:"node"`
+	Key        string    `json:"key"`
+	SeededFrom string    `json:"seeded_from"`
+	Since      time.Time `json:"since"`
+}
+
+// KeyPrefix is the object-store prefix every store object of ownerID (an
+// instance or image ID) is written under.
+func KeyPrefix(ownerID string) string {
+	return objectRoot + ownerID + "/"
+}
+
+// ownsKey reports whether key names one object directly under ownerID's prefix.
+func ownsKey(ownerID, key string) bool {
+	name, ok := strings.CutPrefix(key, KeyPrefix(ownerID))
+	return ok && name != "" && !strings.Contains(name, "/")
+}
+
+// validate checks rec against key. A write-back key outside the instance's own
+// prefix is refused because a superseded obligation deletes that key.
+func (rec *UEFIStoreRecord) validate(key string) error {
+	switch {
+	case rec.SchemaVersion > RecordSchemaVersion:
+		return fmt.Errorf("%w: %s is at schema_version %d, newer than this node understands (%d)",
+			ErrInvalidRecord, key, rec.SchemaVersion, RecordSchemaVersion)
+	case rec.SchemaVersion != RecordSchemaVersion:
+		return fmt.Errorf("%w: %s has schema_version %d", ErrInvalidRecord, key, rec.SchemaVersion)
+	case !strings.HasPrefix(rec.InstanceID, "i-"):
+		return fmt.Errorf("%w: instance_id %q is not an instance ID", ErrInvalidRecord, rec.InstanceID)
+	case rec.InstanceID != key:
+		return fmt.Errorf("%w: instance_id %s stored at key %s", ErrInvalidRecord, rec.InstanceID, key)
+	case rec.State != StateLive && rec.State != StateDeleting:
+		return fmt.Errorf("%w: %s has state %q", ErrInvalidRecord, key, rec.State)
+	}
+	if wb := rec.WriteBack; wb != nil {
+		switch {
+		case wb.Node == "":
+			return fmt.Errorf("%w: %s write-back names no node", ErrInvalidRecord, key)
+		case wb.Since.IsZero():
+			return fmt.Errorf("%w: %s write-back has no since", ErrInvalidRecord, key)
+		case !ownsKey(rec.InstanceID, wb.Key):
+			return fmt.Errorf("%w: %s write-back key %q is not under %s", ErrInvalidRecord, key, wb.Key, KeyPrefix(rec.InstanceID))
+		case !ownsKey(rec.InstanceID, wb.SeededFrom):
+			return fmt.Errorf("%w: %s write-back seeded_from %q is not under %s", ErrInvalidRecord, key, wb.SeededFrom, KeyPrefix(rec.InstanceID))
+		}
+	}
+	return nil
+}
+
+func bucketConfig() kvstore.Config {
+	return kvstore.Config{
+		Name:        BucketName,
+		Description: "Spinifex UEFI variable store records (deletion and write-back obligations)",
+		History:     1,
+		// Left false: a lost bucket has lost obligations, and an empty
+		// recreated one would report every store as having none.
+		RecreateIfMissing: false,
+		OnOpen: func(ctx context.Context, kv jetstream.KeyValue) error {
+			return migrate.DefaultRegistry.RunKV(ctx, BucketName, kv, BucketVersion)
+		},
+		Missing: "UEFI store KV bucket not initialized",
+	}
+}
+
+// Store reads and writes store records. Every change is guarded by the
+// revision the caller read, so a lost race is reported rather than retried.
+type Store struct {
+	kv *kvstore.Store[UEFIStoreRecord]
+}
+
+// NewStore returns a Store over js. The bucket is created on Open or first use.
+func NewStore(js jetstream.JetStream) *Store {
+	return &Store{kv: kvstore.New[UEFIStoreRecord](js, bucketConfig())}
+}
+
+// Open creates the bucket if it is missing and stamps its schema version.
+func (s *Store) Open(ctx context.Context) error {
+	_, err := s.kv.KV(ctx)
+	return err
+}
+
+// Get returns instanceID's record and the revision it was read at. A missing
+// record is kvstore.ErrNotFound; one that breaks its invariants is
+// ErrInvalidRecord, so a newer or corrupt record is never acted on.
+func (s *Store) Get(ctx context.Context, instanceID string) (*UEFIStoreRecord, uint64, error) {
+	rec, rev, err := s.kv.Get(ctx, instanceID)
+	if err != nil {
+		return nil, 0, err
+	}
+	if err := rec.validate(instanceID); err != nil {
+		return nil, 0, err
+	}
+	return rec, rev, nil
+}
+
+// Create writes a new record, which must be live: a store is recorded before
+// its first object is written. An existing record is kvstore.ErrExists. Every
+// write stamps rec with RecordSchemaVersion.
+func (s *Store) Create(ctx context.Context, rec *UEFIStoreRecord) (uint64, error) {
+	rec.SchemaVersion = RecordSchemaVersion
+	if rec.State != StateLive {
+		return 0, fmt.Errorf("%w: %s created in state %q, not %q", ErrInvalidRecord, rec.InstanceID, rec.State, StateLive)
+	}
+	if err := rec.validate(rec.InstanceID); err != nil {
+		return 0, err
+	}
+	return s.kv.Create(ctx, rec.InstanceID, rec)
+}
+
+// CompareAndSet replaces the record only if it is still at rev, returning
+// kvstore.ErrConflict when it is not, and the revision written otherwise.
+func (s *Store) CompareAndSet(ctx context.Context, rec *UEFIStoreRecord, rev uint64) (uint64, error) {
+	rec.SchemaVersion = RecordSchemaVersion
+	if err := rec.validate(rec.InstanceID); err != nil {
+		return 0, err
+	}
+	return s.kv.CompareAndSet(ctx, rec.InstanceID, rec, rev)
+}
+
+// CompareAndDelete removes instanceID's record only if it is still at rev,
+// returning kvstore.ErrConflict when it is not.
+func (s *Store) CompareAndDelete(ctx context.Context, instanceID string, rev uint64) error {
+	return s.kv.CompareAndDelete(ctx, instanceID, rev)
+}

spinifex/domains/ec2/instance/uefistore/uefistore_test.go

@@ -0,0 +1,285 @@
+package uefistore_test
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/instance/uefistore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const instanceID = "i-0123456789abcdef0"
+
+func openStore(t *testing.T) (*uefistore.Store, jetstream.JetStream) {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	s := uefistore.NewStore(js)
+	require.NoError(t, s.Open(t.Context()))
+	return s, js
+}
+
+func liveRecord() *uefistore.UEFIStoreRecord {
+	return &uefistore.UEFIStoreRecord{InstanceID: instanceID, AccountID: "111122223333", State: uefistore.StateLive}
+}
+
+func writeBack() *uefistore.UEFIWriteBack {
+	return &uefistore.UEFIWriteBack{
+		Node:       "node-2",
+		Key:        uefistore.KeyPrefix(instanceID) + "b",
+		SeededFrom: uefistore.KeyPrefix(instanceID) + "a",
+		Since:      time.Date(2026, 11, 3, 1, 2, 3, 0, time.UTC),
+	}
+}
+
+func rawPut(t *testing.T, js jetstream.JetStream, key, body string) {
+	t.Helper()
+	kv, err := js.KeyValue(t.Context(), uefistore.BucketName)
+	require.NoError(t, err)
+	_, err = kv.Put(t.Context(), key, []byte(body))
+	require.NoError(t, err)
+}
+
+// The records carry obligations past the 1 h terminated-record TTL, so the
+// bucket must not expire them, and it is stamped so a newer schema is refused.
+func TestOpen_CreatesBucketWithNoTTLAndStampsVersion(t *testing.T) {
+	_, js := openStore(t)
+
+	kv, err := js.KeyValue(t.Context(), uefistore.BucketName)
+	require.NoError(t, err)
+	status, err := kv.Status(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, int64(1), status.History())
+	assert.Zero(t, status.TTL(), "a TTL would expire deletion obligations before they are settled")
+
+	v, err := kvutil.ReadVersion(t.Context(), kv)
+	require.NoError(t, err)
+	assert.Equal(t, uefistore.BucketVersion, v)
+}
+
+func TestCreate_RoundTripsAndStampsSchemaVersion(t *testing.T) {
+	s, _ := openStore(t)
+	rec := liveRecord()
+	rec.WriteBack = writeBack()
+
+	rev, err := s.Create(t.Context(), rec)
+	require.NoError(t, err)
+
+	got, gotRev, err := s.Get(t.Context(), instanceID)
+	require.NoError(t, err)
+	assert.Equal(t, rev, gotRev)
+	assert.Equal(t, uefistore.RecordSchemaVersion, got.SchemaVersion)
+	assert.Equal(t, rec, got)
+}
+
+// The wire names are what every release reads, so they are pinned.
+func TestRecord_WireShape(t *testing.T) {
+	s, js := openStore(t)
+	rec := liveRecord()
+	rec.WriteBack = writeBack()
+	_, err := s.Create(t.Context(), rec)
+	require.NoError(t, err)
+
+	kv, err := js.KeyValue(t.Context(), uefistore.BucketName)
+	require.NoError(t, err)
+	entry, err := kv.Get(t.Context(), instanceID)
+	require.NoError(t, err)
+	assert.JSONEq(t, `{
+		"schema_version": 1,
+		"instance_id": "i-0123456789abcdef0",
+		"account_id": "111122223333",
+		"state": "live",
+		"write_back": {
+			"node": "node-2",
+			"key": "uefi/i-0123456789abcdef0/b",
+			"seeded_from": "uefi/i-0123456789abcdef0/a",
+			"since": "2026-11-03T01:02:03Z"
+		}
+	}`, string(entry.Value()))
+}
+
+func TestGet_MissingIsNotFound(t *testing.T) {
+	s, _ := openStore(t)
+	_, _, err := s.Get(t.Context(), instanceID)
+	require.ErrorIs(t, err, kvstore.ErrNotFound)
+}
+
+func TestCreate_ExistingRecordIsErrExists(t *testing.T) {
+	s, _ := openStore(t)
+	_, err := s.Create(t.Context(), liveRecord())
+	require.NoError(t, err)
+
+	dup := liveRecord()
+	dup.AccountID = "999999999999"
+	_, err = s.Create(t.Context(), dup)
+	require.ErrorIs(t, err, kvstore.ErrExists)
+
+	got, _, err := s.Get(t.Context(), instanceID)
+	require.NoError(t, err)
+	assert.Equal(t, "111122223333", got.AccountID, "the second create must not overwrite the first")
+}
+
+// A store is recorded live before its first object, so a record that starts
+// out deleting has no object it could be describing.
+func TestCreate_RefusesDeleting(t *testing.T) {
+	s, _ := openStore(t)
+	rec := liveRecord()
+	rec.State = uefistore.StateDeleting
+
+	_, err := s.Create(t.Context(), rec)
+	require.ErrorIs(t, err, uefistore.ErrInvalidRecord)
+	_, _, err = s.Get(t.Context(), instanceID)
+	require.ErrorIs(t, err, kvstore.ErrNotFound)
+}
+
+func TestCompareAndSet_StaleRevisionConflictsAndWritesNothing(t *testing.T) {
+	s, _ := openStore(t)
+	first, err := s.Create(t.Context(), liveRecord())
+	require.NoError(t, err)
+
+	moved := liveRecord()
+	moved.WriteBack = writeBack()
+	second, err := s.CompareAndSet(t.Context(), moved, first)
+	require.NoError(t, err)
+	assert.Greater(t, second, first)
+
+	stale := liveRecord()
+	stale.State = uefistore.StateDeleting
+	_, err = s.CompareAndSet(t.Context(), stale, first)
+	require.ErrorIs(t, err, kvstore.ErrConflict)
+
+	got, rev, err := s.Get(t.Context(), instanceID)
+	require.NoError(t, err)
+	assert.Equal(t, second, rev)
+	assert.Equal(t, uefistore.StateLive, got.State)
+	assert.Equal(t, writeBack(), got.WriteBack)
+}
+
+// Two actors that read the same revision race to change it; exactly one may
+// win, so neither can overwrite an obligation the other recorded.
+func TestCompareAndSet_ConcurrentWritersFromOneReadHaveOneWinner(t *testing.T) {
+	s, _ := openStore(t)
+	_, err := s.Create(t.Context(), liveRecord())
+	require.NoError(t, err)
+	_, rev, err := s.Get(t.Context(), instanceID)
+	require.NoError(t, err)
+
+	const writers = 8
+	var (
+		wg   sync.WaitGroup
+		mu   sync.Mutex
+		wins []string
+	)
+	for i := range writers {
+		wg.Go(func() {
+			rec := liveRecord()
+			rec.WriteBack = writeBack()
+			rec.WriteBack.Node = fmt.Sprintf("node-%d", i)
+			if _, err := s.CompareAndSet(context.Background(), rec, rev); err == nil {
+				mu.Lock()
+				wins = append(wins, rec.WriteBack.Node)
+				mu.Unlock()
+			} else {
+				assert.ErrorIs(t, err, kvstore.ErrConflict)
+			}
+		})
+	}
+	wg.Wait()
+
+	require.Len(t, wins, 1)
+	got, _, err := s.Get(t.Context(), instanceID)
+	require.NoError(t, err)
+	assert.Equal(t, wins[0], got.WriteBack.Node)
+}
+
+func TestCompareAndDelete_StaleRevisionKeepsRecord(t *testing.T) {
+	s, _ := openStore(t)
+	first, err := s.Create(t.Context(), liveRecord())
+	require.NoError(t, err)
+	deleting := liveRecord()
+	deleting.State = uefistore.StateDeleting
+	second, err := s.CompareAndSet(t.Context(), deleting, first)
+	require.NoError(t, err)
+
+	require.ErrorIs(t, s.CompareAndDelete(t.Context(), instanceID, first), kvstore.ErrConflict)
+	_, _, err = s.Get(t.Context(), instanceID)
+	require.NoError(t, err, "a delete guarded on a stale revision must leave the record")
+
+	require.NoError(t, s.CompareAndDelete(t.Context(), instanceID, second))
+	_, _, err = s.Get(t.Context(), instanceID)
+	require.ErrorIs(t, err, kvstore.ErrNotFound)
+}
+
+// A superseded obligation deletes its recorded key, so a key outside the
+// instance's own prefix must never reach the bucket.
+func TestCompareAndSet_RefusesInvalidRecords(t *testing.T) {
+	cases := map[string]func(*uefistore.UEFIStoreRecord){
+		"unknown state":            func(r *uefistore.UEFIStoreRecord) { r.State = "gone" },
+		"not an instance ID":       func(r *uefistore.UEFIStoreRecord) { r.InstanceID = "ami-0123456789abcdef0" },
+		"write-back with no node":  func(r *uefistore.UEFIStoreRecord) { r.WriteBack.Node = "" },
+		"write-back with no since": func(r *uefistore.UEFIStoreRecord) { r.WriteBack.Since = time.Time{} },
+		"write-back key under another instance": func(r *uefistore.UEFIStoreRecord) {
+			r.WriteBack.Key = uefistore.KeyPrefix("i-0fedcba9876543210") + "b"
+		},
+		"write-back key under an image": func(r *uefistore.UEFIStoreRecord) {
+			r.WriteBack.Key = uefistore.KeyPrefix("ami-0123456789abcdef0") + "b"
+		},
+		"write-back key is the prefix itself": func(r *uefistore.UEFIStoreRecord) {
+			r.WriteBack.Key = uefistore.KeyPrefix(instanceID)
+		},
+		"write-back key nested below the prefix": func(r *uefistore.UEFIStoreRecord) {
+			r.WriteBack.Key = uefistore.KeyPrefix(instanceID) + "x/y"
+		},
+		"write-back seeded from another instance": func(r *uefistore.UEFIStoreRecord) {
+			r.WriteBack.SeededFrom = uefistore.KeyPrefix("i-0fedcba9876543210") + "a"
+		},
+	}
+	for name, mutate := range cases {
+		t.Run(name, func(t *testing.T) {
+			s, _ := openStore(t)
+			rev, err := s.Create(t.Context(), liveRecord())
+			require.NoError(t, err)
+
+			rec := liveRecord()
+			rec.WriteBack = writeBack()
+			mutate(rec)
+			_, err = s.CompareAndSet(t.Context(), rec, rev)
+			require.ErrorIs(t, err, uefistore.ErrInvalidRecord)
+
+			got, gotRev, err := s.Get(t.Context(), instanceID)
+			require.NoError(t, err)
+			assert.Equal(t, rev, gotRev, "an invalid record must not be written")
+			assert.Nil(t, got.WriteBack)
+		})
+	}
+}
+
+// A record a newer release wrote, or one that does not describe its own key,
+// is reported rather than decoded into something a reaper would act on.
+func TestGet_RefusesRecordsItCannotTrust(t *testing.T) {
+	cases := map[string]string{
+		"newer schema":       `{"schema_version":2,"instance_id":"i-0123456789abcdef0","state":"live"}`,
+		"no schema":          `{"instance_id":"i-0123456789abcdef0","state":"live"}`,
+		"another instance":   `{"schema_version":1,"instance_id":"i-0fedcba9876543210","state":"deleting"}`,
+		"unknown state":      `{"schema_version":1,"instance_id":"i-0123456789abcdef0","state":"purged"}`,
+		"foreign write-back": `{"schema_version":1,"instance_id":"i-0123456789abcdef0","state":"live","write_back":{"node":"n","key":"uefi/i-0fedcba9876543210/b","seeded_from":"uefi/i-0123456789abcdef0/a","since":"2026-11-03T00:00:00Z"}}`,
+	}
+	for name, body := range cases {
+		t.Run(name, func(t *testing.T) {
+			s, js := openStore(t)
+			require.True(t, json.Valid([]byte(body)))
+			rawPut(t, js, instanceID, body)
+
+			_, _, err := s.Get(t.Context(), instanceID)
+			require.ErrorIs(t, err, uefistore.ErrInvalidRecord)
+		})
+	}
+}

spinifex/domains/ec2/instance/volume_held_elsewhere_test.go

@@ -4,19 +4,50 @@ package instance
 //would widen the handler API for a test rather than for a caller.
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"testing"
 
-	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
-	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+// TestVolumeHeldElsewhere_NamesTheHolder covers both refusals that mean the
+// data is on another node. An operator who gets ServerInternal is told to retry
+// something that cannot succeed while that node is down, so the node name and
+// the reason have to survive the trip out of the daemon.
+func TestVolumeHeldElsewhere_NamesTheHolder(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+	}{
+		{
+			name: "live holder, refused by the lease",
+			err: errors.New("failed to mount volume: volume is leased by another owner: " +
+				"held by node-b since 2026-09-02T17:04:39Z"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := volumeHeldElsewhereError(tc.err)
+			require.Error(t, got, "a volume held elsewhere must not fall through to ServerInternal")
+
+			code := awserrors.ValidErrorCodeFromError(got)
+			assert.NotEqual(t, awserrors.ErrorServerInternal, code,
+				"ServerInternal reads as a transient fault and invites a retry that cannot succeed")
+			assert.Contains(t, got.Error(), "node-b", "the caller has to learn where the data is")
+
+			assert.ErrorIs(t, got, ErrVolumeHeldElsewhere,
+				"recovery retries this on a shorter clock than anything else, so it has to be matchable")
+			resolved, _, ok := awserrors.ResolveErrorDetail(got)
+			require.True(t, ok, "the sentinel must not hide the code from the layer that returns it")
+			assert.Equal(t, awserrors.ErrorIncorrectState, resolved)
+		})
+	}
+}
+
 // TestVolumeHeldElsewhere_IgnoresUnrelatedFailures guards the other direction:
 // a genuine internal fault must keep reporting as one rather than being
 // relabelled as somebody else's volume.
@@ -29,76 +60,4 @@ func TestVolumeHeldElsewhere_IgnoresUnrelatedFailures(t *testing.T) {
 	// than refused, so this text is not an exclusion and must not become one.
 	assert.NoError(t, volumeHeldElsewhereError(errors.New(
 		"took over from node-b, which held unsealed writes since 2026-09-02T17:04:39Z")))
-
-	// Only the provider's code classifies a refusal, so its wording alone does not.
-	assert.NoError(t, volumeHeldElsewhereError(errors.New(
-		"failed to mount volume: "+ec2v1.VolumeHeldElsewhereText+": held by node-b")))
-}
-
-// A provider is free to word its refusal however it likes; the held-elsewhere
-// code is what says another node has the volume.
-func TestVolumeHeldElsewhere_MatchesTheProviderCodeWhateverTheWording(t *testing.T) {
-	refusal := fmt.Errorf("publish volume vol-1: %w", &ebsprovider.ProviderError{
-		Code:    ebsprovider.ErrorCodeVolumeHeldElsewhere,
-		Message: "volume vol-1 is published to node-a",
-	})
-
-	got := volumeHeldElsewhereError(refusal)
-	require.Error(t, got, "a held-elsewhere code must be classified even without viperblockd's text")
-	assert.ErrorIs(t, got, ec2v1.ErrVolumeHeldElsewhere)
-	assert.Contains(t, got.Error(), "node-a", "the caller has to learn where the data is")
-	code, _, ok := awserrors.ResolveErrorDetail(got)
-	require.True(t, ok)
-	assert.Equal(t, awserrors.ErrorIncorrectState, code)
-}
-
-// volume_in_use is a read-only mismatch, not another node's claim, so it must
-// not be relabelled as a volume held elsewhere even though the sentinels nest.
-func TestVolumeHeldElsewhere_VolumeInUseAloneIsNotHeldElsewhere(t *testing.T) {
-	mismatch := fmt.Errorf("publish volume vol-1: %w", &ebsprovider.ProviderError{
-		Code:    ebsprovider.ErrorCodeVolumeInUse,
-		Message: "volume vol-1 is exported read-write, not read-only",
-	})
-	assert.NoError(t, volumeHeldElsewhereError(mismatch))
-}
-
-// heldElsewhereMounter refuses every mount the way the contract mounter
-// reports a provider's held-elsewhere code, in wording that is not viperblockd's.
-type heldElsewhereMounter struct{ raceVolumeMounter }
-
-func (heldElsewhereMounter) Mount(context.Context, *vm.VM) error {
-	return fmt.Errorf("publish volume vol-1: %w", &ebsprovider.ProviderError{
-		Code:    ebsprovider.ErrorCodeVolumeHeldElsewhere,
-		Message: "volume vol-1 is published to node-a",
-	})
-}
-
-// Both start paths a caller can reach must hand back the classified refusal:
-// the stopped-instance claim (ec2.start) and the live owner (ec2.cmd), which
-// the EKS reconciler tries first.
-func TestStartPaths_ReportAVolumeHeldElsewhereByCode(t *testing.T) {
-	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
-	types, _ := defaultPrepareInstanceTypes()
-
-	t.Run("claimed relaunch", func(t *testing.T) {
-		mgr := vm.NewManager()
-		mgr.SetDeps(vm.Deps{NodeID: "node-b", VolumeMounter: heldElsewhereMounter{}})
-		svc := &InstanceServiceImpl{resourceMgr: &fakeResourceCapacityProvider{instanceTypes: types}, vmMgr: mgr}
-
-		err := svc.launchClaimedInstance(t.Context(), &vm.VM{ID: "i-held-claim", AccountID: "acc", InstanceType: "t3.micro", Status: vm.StatePending}, func() {})
-		assert.ErrorIs(t, err, ec2v1.ErrVolumeHeldElsewhere)
-		assert.Equal(t, awserrors.ErrorIncorrectState, awserrors.ValidErrorCodeFromError(err))
-	})
-
-	t.Run("live owner", func(t *testing.T) {
-		const id = "i-held-live"
-		mgr := mgrWith(map[string]*vm.VM{id: {ID: id, Status: vm.StateStopped, InstanceType: "t3.micro"}})
-		mgr.SetDeps(vm.Deps{NodeID: "node-b", VolumeMounter: heldElsewhereMounter{}})
-		v, _ := mgr.Get(id)
-		svc := &InstanceServiceImpl{resourceMgr: &fakeResourceCapacityProvider{instanceTypes: types}, vmMgr: mgr}
-
-		err := svc.StartInstance(t.Context(), v, ec2v1.EC2InstanceCommand{ID: id})
-		assert.ErrorIs(t, err, ec2v1.ErrVolumeHeldElsewhere)
-		assert.Equal(t, awserrors.ErrorIncorrectState, awserrors.ValidErrorCodeFromError(err))
-	})
 }

spinifex/domains/eks/cluster/cluster_reconciler.go

@@ -795,7 +795,8 @@ func retryableStartRefusal(err error) bool {
 	if code, ok := awserrors.ResolveErrorCode(err); ok && code == awserrors.ErrorInsufficientInstanceCapacity {
 		return true
 	}
-	return errors.Is(err, ec2v1.ErrVolumeHeldElsewhere)
+	_, message, _ := awserrors.ResolveErrorDetail(err)
+	return strings.Contains(message, ec2v1.VolumeHeldElsewhereText) || strings.Contains(err.Error(), ec2v1.VolumeHeldElsewhereText)
 }
 
 // maybeReformEtcdQuorum escalates a wedged HA control plane that the in-place

spinifex/domains/eks/cluster/cluster_reconciler_start_refusal_test.go

@@ -3,11 +3,9 @@ package cluster
 import (
 	"context"
 	"errors"
-	"fmt"
 	"testing"
 	"time"
 
-	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -32,12 +30,12 @@ func (c *refusingCPControl) StartInstance(context.Context, string) error {
 
 func (c *refusingCPControl) StopInstance(context.Context, string) error { return nil }
 
-// What the start path returns after crossing NATS: StartInstances wraps the
-// sentinel back on when the owner's reply carries the held-elsewhere header.
+// What the start path returns after crossing NATS: the code and the message the
+// owning node sent, with no sentinel left to match on.
 var (
 	errStartCapacity = errors.New(awserrors.ErrorInsufficientInstanceCapacity)
-	errStartHeld     = fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere,
-		awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", "volume vol-1 is published to node-a"))
+	errStartHeld     = awserrors.Errorf(awserrors.ErrorIncorrectState, "%s",
+		"volume is leased by another owner: held by node-a since 2026-10-30T01:00:00Z")
 )
 
 // A refusal that clears by itself must not use up the restart budget, or the
@@ -74,30 +72,18 @@ func TestMaybeRecoverCP_RetryableStartRefusalIsRetriedNotReplaced(t *testing.T)
 	}
 }
 
-// The cap still bounds the faults nothing says will clear. IncorrectState
-// without the held-elsewhere sentinel is one, whatever its message says.
+// The cap still bounds the faults nothing says will clear.
 func TestMaybeRecoverCP_UnknownStartFailureStillCountsAgainstTheBudget(t *testing.T) {
 	t.Parallel()
-	for name, refusal := range map[string]error{
-		"server internal": errors.New(awserrors.ErrorServerInternal),
-		"unrelated incorrect state": awserrors.Errorf(awserrors.ErrorIncorrectState, "%s",
-			"instance i-cp is not in a state from which it can be started"),
-		"lease text without the sentinel": awserrors.Errorf(awserrors.ErrorIncorrectState, "%s",
-			ec2v1.VolumeHeldElsewhereText+": held by node-a since 2026-10-30T01:00:00Z"),
-	} {
-		t.Run(name, func(t *testing.T) {
-			t.Parallel()
-			cp := &refusingCPControl{err: refusal}
-			r := newRecoveryReconciler(t, cp)
-			r.degradedSince = time.Now().Add(-10 * time.Minute)
+	cp := &refusingCPControl{err: errors.New(awserrors.ErrorServerInternal)}
+	r := newRecoveryReconciler(t, cp)
+	r.degradedSince = time.Now().Add(-10 * time.Minute)
 
-			for range r.maxRestartAttempts + 2 {
-				r.lastRestartAt = time.Time{}
-				r.maybeRecoverControlPlane(t.Context(), metaWithCP("i-cp"), "stale")
-			}
-
-			require.Equal(t, r.maxRestartAttempts, cp.startCalls, "stops at the attempt cap")
-			assert.Equal(t, r.maxRestartAttempts, r.restartAttempts)
-		})
+	for range r.maxRestartAttempts + 2 {
+		r.lastRestartAt = time.Time{}
+		r.maybeRecoverControlPlane(t.Context(), metaWithCP("i-cp"), "stale")
 	}
+
+	require.Equal(t, r.maxRestartAttempts, cp.startCalls, "stops at the attempt cap")
+	assert.Equal(t, r.maxRestartAttempts, r.restartAttempts)
 }

spinifex/domains/iam/service_impl.go

@@ -597,8 +597,14 @@ func (s *IAMServiceImpl) DeleteUser(accountID string, input *iam.DeleteUserInput
 	if len(user.AttachedPolicies) > 0 {
 		return nil, errDetachPoliciesFirst
 	}
-	if len(user.Groups) > 0 {
-		return nil, errors.New(awserrors.ErrorIAMDeleteConflict)
+	for _, groupName := range user.Groups {
+		if _, err := s.getGroup(ctx, accountID, groupName); err == nil {
+			return nil, awserrors.Errorf(awserrors.ErrorIAMDeleteConflict, "Cannot delete entity, must remove users from group first.")
+		} else if code, _ := awserrors.ResolveErrorCode(err); code != awserrors.ErrorIAMNoSuchEntity {
+			return nil, fmt.Errorf("get group: %w", err)
+		}
+		slog.Warn("DeleteUser: ignoring group name with no group record",
+			"accountID", accountID, "userName", userName, "groupName", groupName)
 	}
 	if len(user.InlinePolicies) > 0 {
 		return nil, errDeletePoliciesFirst

spinifex/domains/iam/service_impl_test.go

@@ -3180,3 +3180,37 @@ func TestDeleteUser_RevocationFailureKeepsDelete(t *testing.T) {
 	_, err = svc.GetUser(testAccountID, &iam.GetUserInput{UserName: aws.String("leaver")})
 	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIAMNoSuchEntity))
 }
+
+func TestDeleteUser_DanglingGroupName(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+	createTestUser(t, svc, "danglinggroup")
+	createTestGroup(t, svc, "gone")
+	_, err := svc.AddUserToGroup(testAccountID, &iam.AddUserToGroupInput{
+		UserName: aws.String("danglinggroup"), GroupName: aws.String("gone"),
+	})
+	require.NoError(t, err)
+	require.NoError(t, svc.groupsBucket.Delete(t.Context(), testAccountID+".gone"))
+
+	_, err = svc.DeleteUser(testAccountID, &iam.DeleteUserInput{UserName: aws.String("danglinggroup")})
+	require.NoError(t, err)
+
+	_, err = svc.GetUser(testAccountID, &iam.GetUserInput{UserName: aws.String("danglinggroup")})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorIAMNoSuchEntity)
+}
+
+func TestDeleteUser_LiveGroupStillConflicts(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+	createTestUser(t, svc, "ingroup")
+	createTestGroup(t, svc, "live")
+	_, err := svc.AddUserToGroup(testAccountID, &iam.AddUserToGroupInput{
+		UserName: aws.String("ingroup"), GroupName: aws.String("live"),
+	})
+	require.NoError(t, err)
+
+	_, err = svc.DeleteUser(testAccountID, &iam.DeleteUserInput{UserName: aws.String("ingroup")})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorIAMDeleteConflict)
+}

spinifex/foundation/messaging/nats/nats.go

@@ -238,10 +238,6 @@ const AccountIDHeader = "X-Account-ID"
 // forwarder must relay it instead of starting the instance elsewhere.
 const UnsealedWritesHeader = "X-Unsealed-Writes"
 
-// VolumeHeldElsewhereHeader marks a start reply refused because another node
-// holds one of the instance's volumes, which the AWS error body cannot say.
-const VolumeHeldElsewhereHeader = "X-Volume-Held-Elsewhere"
-
 // PrincipalARNHeader carries the caller's resolved IAM principal ARN from gateway to daemon handlers.
 const PrincipalARNHeader = "X-Principal-ARN"
 

spinifex/operator/accountlifecycle/reapers_iam.go

@@ -162,6 +162,24 @@ func (r *iamUserReaper) Delete(_ context.Context, accountID string, resource Res
 		}
 	}
 
+	// AWS refuses DeleteUser while the user is in a group.
+	groups, err := r.svc.ListGroupsForUser(accountID, &iam.ListGroupsForUserInput{UserName: aws.String(resource.ID)})
+	if err != nil && !isAlreadyGone(err) {
+		return err
+	}
+	if groups != nil {
+		for _, group := range groups.Groups {
+			if group == nil || group.GroupName == nil {
+				continue
+			}
+			if _, err := r.svc.RemoveUserFromGroup(accountID, &iam.RemoveUserFromGroupInput{
+				UserName: aws.String(resource.ID), GroupName: group.GroupName,
+			}); err != nil && !isAlreadyGone(err) {
+				return err
+			}
+		}
+	}
+
 	_, err = r.svc.DeleteUser(accountID, &iam.DeleteUserInput{UserName: aws.String(resource.ID)})
 	return ignoreAlreadyGone(err)
 }
@@ -249,8 +267,46 @@ func (r *iamGroupReaper) List(_ context.Context, accountID string) ([]Resource,
 	return found, nil
 }
 
+// Delete strips the group's policies first: AWS refuses to delete a group that
+// still has any. Its members were removed with their users, earlier in the stage.
 func (r *iamGroupReaper) Delete(_ context.Context, accountID string, resource Resource, _ bool) error {
-	_, err := r.svc.DeleteGroup(accountID, &iam.DeleteGroupInput{GroupName: aws.String(resource.ID)})
+	attached, err := r.svc.ListAttachedGroupPolicies(accountID, &iam.ListAttachedGroupPoliciesInput{
+		GroupName: aws.String(resource.ID),
+	})
+	if err != nil && !isAlreadyGone(err) {
+		return err
+	}
+	if attached != nil {
+		for _, policy := range attached.AttachedPolicies {
+			if policy == nil || policy.PolicyArn == nil {
+				continue
+			}
+			if _, err := r.svc.DetachGroupPolicy(accountID, &iam.DetachGroupPolicyInput{
+				GroupName: aws.String(resource.ID), PolicyArn: policy.PolicyArn,
+			}); err != nil && !isAlreadyGone(err) {
+				return err
+			}
+		}
+	}
+
+	inline, err := r.svc.ListGroupPolicies(accountID, &iam.ListGroupPoliciesInput{GroupName: aws.String(resource.ID)})
+	if err != nil && !isAlreadyGone(err) {
+		return err
+	}
+	if inline != nil {
+		for _, name := range inline.PolicyNames {
+			if name == nil {
+				continue
+			}
+			if _, err := r.svc.DeleteGroupPolicy(accountID, &iam.DeleteGroupPolicyInput{
+				GroupName: aws.String(resource.ID), PolicyName: name,
+			}); err != nil && !isAlreadyGone(err) {
+				return err
+			}
+		}
+	}
+
+	_, err = r.svc.DeleteGroup(accountID, &iam.DeleteGroupInput{GroupName: aws.String(resource.ID)})
 	return ignoreAlreadyGone(err)
 }
 

spinifex/operator/accountlifecycle/reapers_iam_groups_test.go

@@ -0,0 +1,95 @@
+package accountlifecycle
+
+//test:in-package — drives the unexported identity reapers against the real IAM
+// service, so the DeleteConflict refusals are the real ones and not a fake's.
+
+import (
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/iam"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const groupTestAccount = "000000000042"
+
+const groupTestPolicyDoc = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:ListAllMyBuckets","Resource":"*"}]}`
+
+func newRealIAM(t *testing.T) *iamdomain.IAMServiceImpl {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	masterKey, err := iamdomain.GenerateMasterKey()
+	require.NoError(t, err)
+	svc, err := iamdomain.NewIAMServiceImpl(t.Context(), nc, masterKey)
+	require.NoError(t, err)
+	return svc
+}
+
+// drainIdentity runs the identity-stage reapers in registration order for a
+// few passes, as a stage does, and returns what each reaper still lists and
+// the delete errors of the last pass.
+func drainIdentity(t *testing.T, svc iamdomain.IAMService) (remaining []Resource, lastErrs []error) {
+	t.Helper()
+	for range 3 {
+		lastErrs = nil
+		for _, reaper := range IAMReapers(svc) {
+			if reaper.Stage() != StageIdentity {
+				continue
+			}
+			found, err := reaper.List(testCtx(t), groupTestAccount)
+			require.NoError(t, err)
+			for _, resource := range found {
+				if err := reaper.Delete(testCtx(t), groupTestAccount, resource, false); err != nil {
+					lastErrs = append(lastErrs, err)
+				}
+			}
+		}
+	}
+	for _, reaper := range IAMReapers(svc) {
+		if reaper.Stage() != StageIdentity {
+			continue
+		}
+		found, err := reaper.List(testCtx(t), groupTestAccount)
+		require.NoError(t, err)
+		remaining = append(remaining, found...)
+	}
+	return remaining, lastErrs
+}
+
+func seedGroupAccount(t *testing.T, svc *iamdomain.IAMServiceImpl) string {
+	t.Helper()
+	_, err := svc.CreateUser(groupTestAccount, &iam.CreateUserInput{UserName: aws.String("u")})
+	require.NoError(t, err)
+	_, err = svc.CreateGroup(groupTestAccount, &iam.CreateGroupInput{GroupName: aws.String("g")})
+	require.NoError(t, err)
+	_, err = svc.AddUserToGroup(groupTestAccount, &iam.AddUserToGroupInput{UserName: aws.String("u"), GroupName: aws.String("g")})
+	require.NoError(t, err)
+	policy, err := svc.CreatePolicy(groupTestAccount, &iam.CreatePolicyInput{
+		PolicyName: aws.String("p"), PolicyDocument: aws.String(groupTestPolicyDoc),
+	})
+	require.NoError(t, err)
+	_, err = svc.AttachGroupPolicy(groupTestAccount, &iam.AttachGroupPolicyInput{
+		GroupName: aws.String("g"), PolicyArn: policy.Policy.Arn,
+	})
+	require.NoError(t, err)
+	_, err = svc.PutGroupPolicy(groupTestAccount, &iam.PutGroupPolicyInput{
+		GroupName: aws.String("g"), PolicyName: aws.String("inline"), PolicyDocument: aws.String(groupTestPolicyDoc),
+	})
+	require.NoError(t, err)
+	return *policy.Policy.Arn
+}
+
+// A user in a group, a group carrying an attached and an inline policy, and the
+// customer policy attached to it all drain; none of them may strand the account.
+func TestIdentityStageDrainsGroupsMembersAndGroupPolicies(t *testing.T) {
+	svc := newRealIAM(t)
+	seedGroupAccount(t, svc)
+
+	remaining, lastErrs := drainIdentity(t, svc)
+
+	assert.Empty(t, remaining)
+	assert.Empty(t, lastErrs)
+}

spinifex/operator/accountlifecycle/reapers_iam_test.go

@@ -34,6 +34,9 @@ type fakeIAM struct {
 	profileRoles     map[string][]string
 	attachedUser     []string
 	inlineUser       []string
+	userGroups       []string
+	attachedGroup    []string
+	inlineGroup      []string
 	attachedRole     []string
 	inlineRole       []string
 
@@ -99,6 +102,48 @@ func (f *fakeIAM) DeleteAccessKey(_ string, _ *iam.DeleteAccessKeyInput) (*iam.D
 	return &iam.DeleteAccessKeyOutput{}, nil
 }
 
+func (f *fakeIAM) ListGroupsForUser(_ string, _ *iam.ListGroupsForUserInput) (*iam.ListGroupsForUserOutput, error) {
+	f.record("ListGroupsForUser")
+	out := &iam.ListGroupsForUserOutput{}
+	for _, name := range f.userGroups {
+		out.Groups = append(out.Groups, &iam.Group{GroupName: aws.String(name)})
+	}
+	return out, nil
+}
+
+func (f *fakeIAM) RemoveUserFromGroup(_ string, _ *iam.RemoveUserFromGroupInput) (*iam.RemoveUserFromGroupOutput, error) {
+	f.record("RemoveUserFromGroup")
+	return &iam.RemoveUserFromGroupOutput{}, nil
+}
+
+func (f *fakeIAM) ListAttachedGroupPolicies(_ string, _ *iam.ListAttachedGroupPoliciesInput) (*iam.ListAttachedGroupPoliciesOutput, error) {
+	f.record("ListAttachedGroupPolicies")
+	out := &iam.ListAttachedGroupPoliciesOutput{}
+	for _, arn := range f.attachedGroup {
+		out.AttachedPolicies = append(out.AttachedPolicies, &iam.AttachedPolicy{PolicyArn: aws.String(arn)})
+	}
+	return out, nil
+}
+
+func (f *fakeIAM) DetachGroupPolicy(_ string, _ *iam.DetachGroupPolicyInput) (*iam.DetachGroupPolicyOutput, error) {
+	f.record("DetachGroupPolicy")
+	return &iam.DetachGroupPolicyOutput{}, nil
+}
+
+func (f *fakeIAM) ListGroupPolicies(_ string, _ *iam.ListGroupPoliciesInput) (*iam.ListGroupPoliciesOutput, error) {
+	f.record("ListGroupPolicies")
+	out := &iam.ListGroupPoliciesOutput{}
+	for _, name := range f.inlineGroup {
+		out.PolicyNames = append(out.PolicyNames, aws.String(name))
+	}
+	return out, nil
+}
+
+func (f *fakeIAM) DeleteGroupPolicy(_ string, _ *iam.DeleteGroupPolicyInput) (*iam.DeleteGroupPolicyOutput, error) {
+	f.record("DeleteGroupPolicy")
+	return &iam.DeleteGroupPolicyOutput{}, nil
+}
+
 func (f *fakeIAM) ListAttachedUserPolicies(_ string, _ *iam.ListAttachedUserPoliciesInput) (*iam.ListAttachedUserPoliciesOutput, error) {
 	f.record("ListAttachedUserPolicies")
 	out := &iam.ListAttachedUserPoliciesOutput{}
@@ -337,10 +382,41 @@ func TestUserReaperStripsPoliciesBeforeDeleting(t *testing.T) {
 	assert.Equal(t, []string{
 		"ListAttachedUserPolicies", "DetachUserPolicy",
 		"ListUserPolicies", "DeleteUserPolicy",
+		"ListGroupsForUser", "DeleteUser",
+	}, svc.calls)
+}
+
+func TestUserReaperRemovesGroupMembershipsBeforeDeleting(t *testing.T) {
+	svc := newFakeIAM()
+	svc.userGroups = []string{"g1", "g2"}
+
+	err := (&iamUserReaper{svc: svc}).Delete(testCtx(t), "000000000042",
+		Resource{Kind: "iam-user", ID: "admin"}, false)
+	require.NoError(t, err)
+
+	assert.Equal(t, []string{
+		"ListAttachedUserPolicies", "ListUserPolicies",
+		"ListGroupsForUser", "RemoveUserFromGroup", "RemoveUserFromGroup",
 		"DeleteUser",
 	}, svc.calls)
 }
 
+func TestGroupReaperStripsPoliciesBeforeDeleting(t *testing.T) {
+	svc := newFakeIAM()
+	svc.attachedGroup = []string{"arn:aws:iam::000000000042:policy/p"}
+	svc.inlineGroup = []string{"inline-1"}
+
+	err := (&iamGroupReaper{svc: svc}).Delete(testCtx(t), "000000000042",
+		Resource{Kind: "iam-group", ID: "g"}, false)
+	require.NoError(t, err)
+
+	assert.Equal(t, []string{
+		"ListAttachedGroupPolicies", "DetachGroupPolicy",
+		"ListGroupPolicies", "DeleteGroupPolicy",
+		"DeleteGroup",
+	}, svc.calls)
+}
+
 func TestRoleReaperStripsPoliciesBeforeDeleting(t *testing.T) {
 	svc := newFakeIAM()
 	svc.attachedRole = []string{"arn:aws:iam::000000000042:policy/task"}

spinifex/operator/upgrade/VERSIONS.md

@@ -23,6 +23,7 @@ Single source of truth for the current schema version of every config file Spini
 | `spinifex-cluster-state` | `1` | `daemon.ClusterStateBucketVersion` |
 | `spinifex-cluster-config` | `1` | `daemon.ClusterConfigBucketVersion` |
 | `spinifex-terminated-instances` | `3` | `daemon.TerminatedInstanceBucketVersion` |
+| `spinifex-uefi-stores` | `1` | `uefistore.BucketVersion` |
 
 ## Registered migrations
 

spinifex/providers/ebs/conformance/conformance.go

@@ -150,6 +150,11 @@ type SuiteConfig struct {
 	// a node other than NodeID: what a request meets when the volume's owner
 	// does not answer. Nil skips the cases that need it.
 	QueueOnOtherNode func(t *testing.T) ebsprovider.EBSProvider
+
+	// RevokeExport makes the provider revoke a live export as it does when the
+	// volume is lost to another node, and returns the FenceNotice it sent; see
+	// CaptureFenceNotice. Nil skips the case, for a provider that never revokes.
+	RevokeExport func(t *testing.T, published *ebsprovider.PublishedVolume) ebsprovider.FenceNotice
 }
 
 func (c SuiteConfig) node() string {
@@ -988,6 +993,24 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
 			nbdInfoRetry(t, dialURI(t, publishedNBDURI(t, second)))
 		})
 
+		t.Run("revoking a live export sends a fence notice to its node", func(t *testing.T) {
+			if cfg.RevokeExport == nil {
+				t.Skip("this provider never revokes a live export")
+			}
+			provider := newProvider(t)
+			ctx := context.Background()
+			volumeID := cfg.id("vol-pub-revoked")
+			_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30}})
+			require.NoError(t, err)
+			published, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: cfg.node()})
+			require.NoError(t, err)
+
+			notice := cfg.RevokeExport(t, published)
+			assert.Equal(t, volumeID, notice.VolumeID, "the daemon stops the guest on the volume the notice names")
+			assert.Equal(t, published.NodeID, notice.NodeID, "the notice must name the node whose export was revoked")
+			assert.True(t, ebsprovider.SupportsSchemaVersion(notice.SchemaVersion), "a daemon refuses a notice in a version it cannot read")
+		})
+
 		t.Run("not_found on absent volume", func(t *testing.T) {
 			provider := newProvider(t)
 			_, err := provider.PublishVolume(context.Background(), ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: cfg.id("vol-never-existed"), NodeID: cfg.node()})

spinifex/providers/ebs/conformance/conformance_test.go

@@ -75,6 +75,28 @@ func TestMemoryProviderNamePrefixConformance(t *testing.T) {
 	}
 }
 
+// TestMemoryProviderFenceNoticeConformance runs the suite with the revocation
+// case enabled, so the contract's fence notice is held to the same check here
+// as on viperblockd.
+func TestMemoryProviderFenceNoticeConformance(t *testing.T) {
+	_, conn := testutil.StartTestNATS(t)
+	shared := ebsprovider.NewMemoryProvider(conformance.ReferenceCapabilities)
+	shared.NotifyFencesOn(conn)
+
+	conformance.RunSuiteWithConfig(t, func(t *testing.T) ebsprovider.EBSProvider {
+		t.Helper()
+		return shared
+	}, conformance.SuiteConfig{
+		RevokeExport: func(t *testing.T, published *ebsprovider.PublishedVolume) ebsprovider.FenceNotice {
+			t.Helper()
+			return conformance.CaptureFenceNotice(t, conn, published.NodeID, func() error {
+				_, err := shared.RevokePublication(t.Context(), published.VolumeID, "node-elsewhere")
+				return err
+			})
+		},
+	})
+}
+
 // TestNATSProviderConformance runs the suite against a NATSProvider backed
 // by natsserve.Serve, the production-shaped neutral server, rather than a
 // test-only twin: this exercises the transport this package's implementors actually run in production, not a hand-written stand-in for it.

spinifex/providers/ebs/conformance/fence_notice.go

@@ -0,0 +1,60 @@
+package conformance
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/require"
+)
+
+// fenceNoticeWait bounds how long CaptureFenceNotice waits for the notice.
+const fenceNoticeWait = 10 * time.Second
+
+// CaptureFenceNotice answers nodeID's fence notices on conn as a daemon with no
+// guest on the volume does, starts revoke, and returns the first notice heard.
+// revoke runs on its own goroutine, since a sender may retry until answered.
+func CaptureFenceNotice(t *testing.T, conn *nats.Conn, nodeID string, revoke func() error) ebsprovider.FenceNotice {
+	t.Helper()
+	subject, err := ebsprovider.FenceNoticeSubject(nodeID)
+	require.NoError(t, err)
+
+	heard := make(chan ebsprovider.FenceNotice, 1)
+	sub, err := conn.Subscribe(subject, func(msg *nats.Msg) {
+		var notice ebsprovider.FenceNotice
+		ack := ebsprovider.FenceNoticeAck{Versioned: ebsprovider.NewVersioned()}
+		if err := json.Unmarshal(msg.Data, &notice); err != nil {
+			ack.Error = &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeInvalidArgument, Message: err.Error()}
+		} else {
+			ack.Versioned = ebsprovider.ReplyVersioned(notice.SchemaVersion)
+			select {
+			case heard <- notice:
+			default:
+			}
+		}
+		if payload, err := json.Marshal(ack); err == nil {
+			_ = msg.Respond(payload)
+		}
+	})
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+	require.NoError(t, conn.Flush())
+
+	revoked := make(chan error, 1)
+	go func() { revoked <- revoke() }()
+
+	timeout := time.After(fenceNoticeWait)
+	for {
+		select {
+		case notice := <-heard:
+			return notice
+		case err := <-revoked:
+			require.NoError(t, err, "revoking the export")
+			revoked = nil
+		case <-timeout:
+			t.Fatalf("no fence notice on %s within %s of revoking the export; the daemon would leave its guest running on a revoked disk", subject, fenceNoticeWait)
+		}
+	}
+}

spinifex/providers/ebs/fence_notice.go

@@ -0,0 +1,53 @@
+package ebsprovider
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// NotifyFenced sends notice once and returns the instance the daemon stopped.
+// An ErrInvalidArgument error is terminal: the notice can never be sent. A
+// *ProviderError (internal for an undecodable ack) was heard; others were not.
+func NotifyFenced(ctx context.Context, conn *nats.Conn, notice FenceNotice, timeout time.Duration) (string, error) {
+	if notice.VolumeID == "" {
+		return "", fmt.Errorf("%w: fence notice names no volume", ErrInvalidArgument)
+	}
+	subject, err := FenceNoticeSubject(notice.NodeID)
+	if err != nil {
+		return "", err
+	}
+	if conn == nil {
+		return "", fmt.Errorf("%w: no connection to send the fence notice on", ErrInvalidArgument)
+	}
+
+	reply, err := requestReply(ctx, conn, subject, timeout, notice)
+	if err != nil {
+		return "", err
+	}
+	var ack FenceNoticeAck
+	if err := json.Unmarshal(reply, &ack); err != nil {
+		// The daemon answered, so repeating the notice would only repeat this.
+		return "", &ProviderError{Code: ErrorCodeInternal, Message: fmt.Sprintf("decode %s ack: %v", subject, err)}
+	}
+	if ack.Error != nil {
+		return ack.InstanceID, ack.Error
+	}
+	return ack.InstanceID, nil
+}
+
+// FenceNoticeRetryable reports whether a NotifyFenced error leaves the notice
+// to be sent again: it was not heard, or the daemon refused it as unavailable.
+func FenceNoticeRetryable(err error) bool {
+	if err == nil || errors.Is(err, ErrInvalidArgument) {
+		return false
+	}
+	if _, refused := errors.AsType[*ProviderError](err); refused {
+		return errors.Is(err, ErrUnavailable)
+	}
+	return true
+}

spinifex/providers/ebs/fence_notice_test.go

@@ -0,0 +1,205 @@
+package ebsprovider
+
+import (
+	"context"
+	"encoding/json"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// answerFenceNotices replies to node-a's fence notices with reply.
+func answerFenceNotices(t *testing.T, conn *nats.Conn, reply []byte) {
+	t.Helper()
+	subject, err := FenceNoticeSubject("node-a")
+	require.NoError(t, err)
+	sub, err := conn.Subscribe(subject, func(msg *nats.Msg) { _ = msg.Respond(reply) })
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, conn.Flush())
+}
+
+func fenceNoticeFor(volumeID string) FenceNotice {
+	return FenceNotice{Versioned: NewVersioned(), VolumeID: volumeID, NodeID: "node-a", Winner: "node-b"}
+}
+
+// TestNotifyFenced_SeparatesHeardFromUnheard pins what a sender's retry turns
+// on: an answer of any kind ends the notice, and only silence repeats it.
+func TestNotifyFenced_SeparatesHeardFromUnheard(t *testing.T) {
+	t.Run("acknowledged", func(t *testing.T) {
+		_, conn := testutil.StartTestNATS(t)
+		ack, err := json.Marshal(FenceNoticeAck{Versioned: NewVersioned(), InstanceID: "i-stopped"})
+		require.NoError(t, err)
+		answerFenceNotices(t, conn, ack)
+
+		instanceID, err := NotifyFenced(t.Context(), conn, fenceNoticeFor("vol-1"), time.Second)
+		require.NoError(t, err)
+		assert.Equal(t, "i-stopped", instanceID)
+	})
+
+	t.Run("refused", func(t *testing.T) {
+		_, conn := testutil.StartTestNATS(t)
+		ack, err := json.Marshal(FenceNoticeAck{Versioned: NewVersioned(), Error: &ProviderError{Code: ErrorCodeInvalidArgument, Message: "no"}})
+		require.NoError(t, err)
+		answerFenceNotices(t, conn, ack)
+
+		_, err = NotifyFenced(t.Context(), conn, fenceNoticeFor("vol-1"), time.Second)
+		var refused *ProviderError
+		require.ErrorAs(t, err, &refused, "a refusal was heard, and repeating it would only be refused again")
+		assert.ErrorIs(t, err, ErrInvalidArgument)
+		assert.False(t, FenceNoticeRetryable(err))
+	})
+
+	t.Run("refused as unavailable", func(t *testing.T) {
+		_, conn := testutil.StartTestNATS(t)
+		ack, err := json.Marshal(FenceNoticeAck{Versioned: NewVersioned(), Error: &ProviderError{Code: ErrorCodeUnavailable, Message: "later"}})
+		require.NoError(t, err)
+		answerFenceNotices(t, conn, ack)
+
+		_, err = NotifyFenced(t.Context(), conn, fenceNoticeFor("vol-1"), time.Second)
+		require.ErrorIs(t, err, ErrUnavailable)
+		assert.True(t, FenceNoticeRetryable(err), "a daemon that cannot act yet may act on the next notice")
+	})
+
+	t.Run("undecodable answer", func(t *testing.T) {
+		_, conn := testutil.StartTestNATS(t)
+		answerFenceNotices(t, conn, []byte("{"))
+
+		_, err := NotifyFenced(t.Context(), conn, fenceNoticeFor("vol-1"), time.Second)
+		var refused *ProviderError
+		require.ErrorAs(t, err, &refused, "the daemon answered, so the notice was heard")
+		assert.Equal(t, ErrorCodeInternal, refused.Code)
+		assert.False(t, FenceNoticeRetryable(err))
+	})
+
+	t.Run("nobody listening", func(t *testing.T) {
+		_, conn := testutil.StartTestNATS(t)
+		_, err := NotifyFenced(t.Context(), conn, fenceNoticeFor("vol-1"), time.Second)
+		require.Error(t, err)
+		var refused *ProviderError
+		assert.NotErrorAs(t, err, &refused, "an unheard notice must read as unheard, or the sender stops retrying")
+		assert.True(t, FenceNoticeRetryable(err))
+	})
+
+	t.Run("connection closed", func(t *testing.T) {
+		_, conn := testutil.StartTestNATS(t)
+		conn.Close()
+		_, err := NotifyFenced(t.Context(), conn, fenceNoticeFor("vol-1"), time.Second)
+		require.ErrorIs(t, err, nats.ErrConnectionClosed)
+		assert.True(t, FenceNoticeRetryable(err), "a connection that comes back can still carry the notice")
+	})
+}
+
+// TestNotifyFenced_AttributesTheNoticeToItsAccount holds the notice to the
+// contract's request path: the daemon's span names the account it acts for.
+func TestNotifyFenced_AttributesTheNoticeToItsAccount(t *testing.T) {
+	_, conn := testutil.StartTestNATS(t)
+	subject, err := FenceNoticeSubject("node-a")
+	require.NoError(t, err)
+	accounts := make(chan string, 1)
+	ack, err := json.Marshal(FenceNoticeAck{Versioned: NewVersioned()})
+	require.NoError(t, err)
+	sub, err := conn.Subscribe(subject, func(msg *nats.Msg) {
+		accounts <- msg.Header.Get(AccountIDHeader)
+		_ = msg.Respond(ack)
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, conn.Flush())
+
+	_, err = NotifyFenced(WithAccountID(t.Context(), "123456789012"), conn, fenceNoticeFor("vol-1"), time.Second)
+	require.NoError(t, err)
+	assert.Equal(t, "123456789012", <-accounts)
+}
+
+// TestNotifyFenced_RefusesAnUnaddressableNotice keeps a notice that can never
+// be sent from reading as unheard, which a retrying sender would repeat forever.
+func TestNotifyFenced_RefusesAnUnaddressableNotice(t *testing.T) {
+	_, conn := testutil.StartTestNATS(t)
+	cases := map[string]struct {
+		conn   *nats.Conn
+		notice FenceNotice
+	}{
+		"no volume": {conn, FenceNotice{Versioned: NewVersioned(), NodeID: "node-a"}},
+		"no node":   {conn, FenceNotice{Versioned: NewVersioned(), VolumeID: "vol-1"}},
+		"bad node":  {conn, FenceNotice{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node.a"}},
+		"no conn":   {nil, fenceNoticeFor("vol-1")},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			_, err := NotifyFenced(context.Background(), tc.conn, tc.notice, time.Second)
+			require.ErrorIs(t, err, ErrInvalidArgument)
+			assert.False(t, FenceNoticeRetryable(err), "a notice that can never be sent must not be retried")
+		})
+	}
+}
+
+// TestMemoryProviderRevokePublication_RetriesAnUnheardNotice keeps the memory
+// provider to the fence's delivery rule: a notice nobody answered is sent
+// again until one does, not dropped after the first try.
+func TestMemoryProviderRevokePublication_RetriesAnUnheardNotice(t *testing.T) {
+	_, conn := testutil.StartTestNATS(t)
+	ctx := t.Context()
+	provider := NewMemoryProvider(Capabilities{})
+	provider.fenceNoticeTimeout = 100 * time.Millisecond
+	provider.fenceNoticeEvery = 10 * time.Millisecond
+	provider.NotifyFencesOn(conn)
+	_, err := provider.CreateVolume(ctx, CreateVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", CapacityRange: CapacityRange{RequiredBytes: 1 << 30}})
+	require.NoError(t, err)
+	_, err = provider.PublishVolume(ctx, PublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node-a"})
+	require.NoError(t, err)
+
+	subject, err := FenceNoticeSubject("node-a")
+	require.NoError(t, err)
+	ack, err := json.Marshal(FenceNoticeAck{Versioned: NewVersioned(), InstanceID: "i-stopped"})
+	require.NoError(t, err)
+	var heard atomic.Int32
+	sub, err := conn.Subscribe(subject, func(msg *nats.Msg) {
+		if heard.Add(1) == 1 {
+			return // the first notice is lost, as on a daemon still reconnecting
+		}
+		_ = msg.Respond(ack)
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, conn.Flush())
+
+	stopped, err := provider.RevokePublication(ctx, "vol-1", "node-b")
+	require.NoError(t, err, "a lost first notice must be sent again")
+	assert.Equal(t, "i-stopped", stopped)
+	assert.Equal(t, int32(2), heard.Load())
+}
+
+func TestMemoryProviderRevokePublication(t *testing.T) {
+	_, conn := testutil.StartTestNATS(t)
+	ctx := context.Background()
+	provider := NewMemoryProvider(Capabilities{})
+	_, err := provider.CreateVolume(ctx, CreateVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", CapacityRange: CapacityRange{RequiredBytes: 1 << 30}})
+	require.NoError(t, err)
+	_, err = provider.PublishVolume(ctx, PublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node-a"})
+	require.NoError(t, err)
+
+	_, err = provider.RevokePublication(ctx, "vol-1", "node-b")
+	require.ErrorIs(t, err, ErrUnavailable, "with nowhere to announce it, a revocation would leave the guest unstopped")
+	stillPublished, err := provider.PublishVolume(ctx, PublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node-b"})
+	require.ErrorIs(t, err, ErrVolumeInUse, "a refused revocation must leave the publication in place")
+	assert.Nil(t, stillPublished)
+
+	ack, err := json.Marshal(FenceNoticeAck{Versioned: NewVersioned()})
+	require.NoError(t, err)
+	answerFenceNotices(t, conn, ack)
+	provider.NotifyFencesOn(conn)
+
+	_, err = provider.RevokePublication(ctx, "vol-1", "node-b")
+	require.NoError(t, err)
+	_, err = provider.PublishVolume(ctx, PublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node-b"})
+	require.NoError(t, err, "a revoked publication no longer holds the volume")
+
+	_, err = provider.RevokePublication(ctx, "vol-never", "node-b")
+	require.ErrorIs(t, err, ErrNotFound)
+}

spinifex/providers/ebs/memory.go

@@ -2,12 +2,15 @@ package ebsprovider
 
 import (
 	"bytes"
+	"cmp"
 	"context"
 	"fmt"
 	"maps"
 	"slices"
 	"sync"
 	"time"
+
+	"github.com/nats-io/nats.go"
 )
 
 // MemoryProvider is a deterministic, concurrency-safe Provider used by
@@ -19,6 +22,10 @@ type MemoryProvider struct {
 	volumes      map[string]*memoryVolume
 	snapshots    map[string]*Snapshot
 	now          func() time.Time
+	fenceConn    *nats.Conn
+	// Zero takes the defaults; tests shorten them.
+	fenceNoticeTimeout time.Duration
+	fenceNoticeEvery   time.Duration
 }
 
 type memoryVolume struct {
@@ -406,13 +413,73 @@ func (m *MemoryProvider) UnpublishVolume(_ context.Context, req UnpublishVolumeR
 	if volume == nil || volume.published == nil || volume.published.NodeID != req.NodeID {
 		return result, nil
 	}
-	volume.published = nil
-	volume.readOnly = false
-	volume.volume.State = VolumeStateAvailable
+	volume.unpublish()
 	result.TornDown = req.SchemaVersion >= NoSealSchemaVersion
 	return result, nil
 }
 
+// NotifyFencesOn sets the connection RevokePublication announces fences on.
+func (m *MemoryProvider) NotifyFencesOn(conn *nats.Conn) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.fenceConn = conn
+}
+
+// unpublish drops the volume's publication. The caller holds the provider lock.
+func (v *memoryVolume) unpublish() {
+	v.published = nil
+	v.readOnly = false
+	v.volume.State = VolumeStateAvailable
+}
+
+// Defaults for RevokePublication's notice: each attempt's wait for an answer,
+// and the pause before an unanswered notice is sent again.
+const (
+	memoryFenceNoticeTimeout = 2 * time.Second
+	memoryFenceNoticeEvery   = time.Second
+)
+
+// RevokePublication tears volumeID's publication down as a provider that lost
+// the volume to winner does, then sends a FenceNotice to the node it was
+// published on until heard or ctx ends. It returns the instance stopped there.
+func (m *MemoryProvider) RevokePublication(ctx context.Context, volumeID, winner string) (string, error) {
+	m.mu.Lock()
+	conn := m.fenceConn
+	volume := m.volumes[volumeID]
+	timeout := cmp.Or(m.fenceNoticeTimeout, memoryFenceNoticeTimeout)
+	every := cmp.Or(m.fenceNoticeEvery, memoryFenceNoticeEvery)
+	if conn == nil {
+		m.mu.Unlock()
+		return "", fmt.Errorf("%w: no connection to announce the fence on", ErrUnavailable)
+	}
+	if volume == nil || volume.published == nil {
+		m.mu.Unlock()
+		return "", fmt.Errorf("%w: volume %s is not published", ErrNotFound, volumeID)
+	}
+	nodeID := volume.published.NodeID
+	volume.unpublish()
+	m.mu.Unlock()
+
+	notice := FenceNotice{
+		Versioned: NewVersioned(),
+		VolumeID:  volumeID,
+		NodeID:    nodeID,
+		Winner:    winner,
+		Reason:    "volume revoked in favour of " + winner,
+	}
+	for {
+		instanceID, err := NotifyFenced(ctx, conn, notice, timeout)
+		if !FenceNoticeRetryable(err) || conn.IsClosed() {
+			return instanceID, err
+		}
+		select {
+		case <-ctx.Done():
+			return "", fmt.Errorf("fence notice for %s never acknowledged: %w: %w", volumeID, ctx.Err(), err)
+		case <-time.After(every):
+		}
+	}
+}
+
 // HoldsUnsealedWrites answers that none are held: every write lands in this
 // process's map, which is the backend, before it is acknowledged.
 func (m *MemoryProvider) HoldsUnsealedWrites(_ context.Context, nodeID string, volumeIDs []string) ([]string, error) {

spinifex/providers/ebs/nats.go

@@ -569,17 +569,33 @@ func (p *NATSProvider) HoldsUnsealedWrites(ctx context.Context, nodeID string, v
 // connected reports whether a request can be sent now. A reconnecting
 // connection would buffer it and send it on reconnect, after its caller may
 // have given up, so the provider would act with nobody waiting on the result.
-func (p *NATSProvider) connected() bool {
-	return p.conn != nil && p.conn.IsConnected()
+func connected(conn *nats.Conn) bool {
+	return conn != nil && conn.IsConnected()
 }
 
+func (p *NATSProvider) connected() bool { return connected(p.conn) }
+
 func (p *NATSProvider) request(ctx context.Context, subject string, input, output any) error {
-	if !p.connected() {
-		return nats.ErrConnectionClosed
+	reply, err := requestReply(ctx, p.conn, subject, p.timeoutFor(subject), input)
+	if err != nil {
+		return err
+	}
+	if err := json.Unmarshal(reply, output); err != nil {
+		return fmt.Errorf("decode %s response: %w: %w", subject, ErrUnreadableReply, err)
+	}
+	return nil
+}
+
+// requestReply sends input on subject as ctx's account, waits up to timeout
+// and returns the raw reply. Its client span records a transport error, an
+// unreadable reply, or the error the reply carries; decoding is the caller's.
+func requestReply(ctx context.Context, conn *nats.Conn, subject string, timeout time.Duration, input any) ([]byte, error) {
+	if !connected(conn) {
+		return nil, nats.ErrConnectionClosed
 	}
 	payload, err := json.Marshal(input)
 	if err != nil {
-		return fmt.Errorf("encode %s request: %w", subject, err)
+		return nil, fmt.Errorf("encode %s request: %w", subject, err)
 	}
 
 	request := &nats.Msg{Subject: subject, Data: payload, Header: nats.Header{}}
@@ -590,18 +606,24 @@ func (p *NATSProvider) request(ctx context.Context, subject string, input, outpu
 	ctx, span := startClientSpan(ctx, subject, accountID, request.Header)
 	defer span.End()
 
-	requestCtx, cancel := context.WithTimeout(ctx, p.timeoutFor(subject))
+	requestCtx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
-	msg, err := p.conn.RequestMsgWithContext(requestCtx, request)
+	msg, err := conn.RequestMsgWithContext(requestCtx, request)
 	if err != nil {
 		RecordSpanError(span, err)
-		return fmt.Errorf("request %s: %w", subject, err)
-	}
-	if err := json.Unmarshal(msg.Data, output); err != nil {
-		RecordSpanError(span, err)
-		return fmt.Errorf("decode %s response: %w: %w", subject, ErrUnreadableReply, err)
+		return nil, fmt.Errorf("request %s: %w", subject, err)
 	}
 	RecordResponseError(span, msg.Data)
+	return msg.Data, nil
+}
+
+// ValidateNodeID reports whether nodeID can address this node's contract
+// subjects. A node that fails it has no fence-notice subject, so a revoked
+// export there would leave its guest running.
+func ValidateNodeID(nodeID string) error {
+	if err := validateSubjectToken(nodeID); err != nil {
+		return fmt.Errorf("node ID cannot address EBS provider subjects: %w", err)
+	}
 	return nil
 }
 

spinifex/providers/ebs/telemetry.go

@@ -3,6 +3,7 @@ package ebsprovider
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"maps"
 	"slices"
 	"strings"
@@ -218,9 +219,9 @@ func RecordProviderError(span trace.Span, providerErr *ProviderError) {
 	span.SetStatus(codes.Error, providerErr.Message)
 }
 
-// RecordResponseError tags span with the error an encoded response carries.
-// The response types share no accessor for it, so this re-reads the one field
-// rather than making every verb implement one; it runs only when recording.
+// RecordResponseError tags span with the error an encoded response carries, or
+// fails it for a response that is not readable. The response types share no
+// accessor, so this re-reads the one field; it runs only when recording.
 func RecordResponseError(span trace.Span, data []byte) {
 	if !span.IsRecording() {
 		return
@@ -229,6 +230,7 @@ func RecordResponseError(span trace.Span, data []byte) {
 		Error *ProviderError `json:"error"`
 	}
 	if err := json.Unmarshal(data, &envelope); err != nil {
+		RecordSpanError(span, fmt.Errorf("%w: %w", ErrUnreadableReply, err))
 		return
 	}
 	RecordProviderError(span, envelope.Error)

spinifex/providers/ebs/viperblock/conformance_test.go

@@ -75,6 +75,7 @@ type providerHarness struct {
 	nodeID           string
 	otherNodeID      string
 	killExport       func(*testing.T, *ebsprovider.PublishedVolume)
+	revokeExport     func(*testing.T, *ebsprovider.PublishedVolume) ebsprovider.FenceNotice
 	queueOnOtherNode func(t *testing.T) ebsprovider.EBSProvider
 }
 
@@ -86,6 +87,8 @@ type providerHarness struct {
 // would go to a subject nobody serves and time out instead of being refused.
 // Both share the NATS the volume leases live in and the bucket the volumes
 // live in, so a second opener is refused by the lease, not by the harness.
+// The hooks kill a published export's nbdkit behind the handlers' back, and
+// revoke one as a lost lease does.
 func inProcessProvider(t *testing.T) providerHarness {
 	t.Helper()
 	pluginPath := requireExportTools(t)
@@ -123,6 +126,17 @@ func inProcessProvider(t *testing.T) providerHarness {
 	_, err = js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: clusterv1.ClusterStateBucket})
 	require.NoError(t, err)
 
+	holder := func(t *testing.T, published *ebsprovider.PublishedVolume) (*Config, MountedVolume) {
+		t.Helper()
+		for _, node := range []*Config{first, second} {
+			if mv, ok := findMountedVolume(node, published.VolumeID); ok && node.NodeName == published.NodeID {
+				return node, mv
+			}
+		}
+		t.Fatalf("no mounted entry for %s on %s", published.VolumeID, published.NodeID)
+		return nil, MountedVolume{}
+	}
+
 	return providerHarness{
 		newProvider: func(t *testing.T) ebsprovider.EBSProvider {
 			t.Helper()
@@ -130,16 +144,18 @@ func inProcessProvider(t *testing.T) providerHarness {
 		},
 		nodeID:      first.NodeName,
 		otherNodeID: second.NodeName,
-		// Kills a published export's nbdkit behind the handlers' back.
 		killExport: func(t *testing.T, published *ebsprovider.PublishedVolume) {
 			t.Helper()
-			for _, node := range []*Config{first, second} {
-				if mv, ok := findMountedVolume(node, published.VolumeID); ok && node.NodeName == published.NodeID {
-					require.NoError(t, hostprocess.ForceKillProcess(mv.PID, 5*time.Second))
-					return
-				}
-			}
-			t.Fatalf("no mounted entry for %s on %s", published.VolumeID, published.NodeID)
+			_, mv := holder(t, published)
+			require.NoError(t, hostprocess.ForceKillProcess(mv.PID, 5*time.Second))
+		},
+		revokeExport: func(t *testing.T, published *ebsprovider.PublishedVolume) ebsprovider.FenceNotice {
+			t.Helper()
+			node, _ := holder(t, published)
+			return conformance.CaptureFenceNotice(t, nc, published.NodeID, func() error {
+				node.onVolumeLeaseLost(t.Context(), published.VolumeID, leaseLostStalled)
+				return nil
+			})
 		},
 		queueOnOtherNode: func(t *testing.T) ebsprovider.EBSProvider {
 			t.Helper()
@@ -163,6 +179,7 @@ func TestViperblockdConformance(t *testing.T) {
 		NodeID:           h.nodeID,
 		OtherNodeID:      h.otherNodeID,
 		KillExport:       h.killExport,
+		RevokeExport:     h.revokeExport,
 		QueueOnOtherNode: h.queueOnOtherNode,
 	})
 }

spinifex/providers/ebs/viperblock/provider_handlers.go

@@ -53,12 +53,12 @@ var providerObjectStoreFactory = func(cfg *Config) objectstore.ObjectStore {
 // viperblock engine construction out of the EC2 control-plane handlers and
 // into the storage daemon that owns BaseDir and the mounted-volume registry.
 //
-// PublishVolume/UnpublishVolume are node-addressed (ebsprovider.PublishSubject
-// / UnpublishSubject already route to one node), so they are only registered
-// when cfg.NodeName is set; there is no queue-group fallback the way the
-// legacy ebs.mount/ebs.unmount subjects have.
+// It refuses a cfg.NodeName that cannot address the node-addressed subjects.
 // Exported so harnesses can put a real provider behind the control plane.
 func RegisterProviderSubjects(cfg *Config, nc *nats.Conn) error {
+	if err := cfg.validateNodeName(); err != nil {
+		return err
+	}
 	// The lease store has to exist before any subject is served: a handler
 	// that reaches an engine open without one refuses, and refusing every
 	// publish is a worse failure than not starting.
@@ -114,11 +114,6 @@ func RegisterProviderSubjects(cfg *Config, nc *nats.Conn) error {
 		return fmt.Errorf("subscribe to %s: %w", snapshotCreateWildcard, err)
 	}
 
-	if cfg.NodeName == "" {
-		slog.Warn("ebs.provider: NodeName is empty, this node cannot serve PublishVolume/UnpublishVolume attachments")
-		return nil
-	}
-
 	publishSubject, err := ebsprovider.PublishSubject(cfg.NodeName)
 	if err != nil {
 		return fmt.Errorf("build publish subject: %w", err)
@@ -1809,8 +1804,8 @@ func unmountVolume(ctx context.Context, cfg *Config, volumeName string) (viperbl
 
 // handlePublishVolume serves ebs.provider.v1.<node>.mount, the provider-neutral
 // front for the same nbdkit mount path ebs.mount uses. It is node-addressed
-// (RegisterProviderSubjects only subscribes it when cfg.NodeName is set), so
-// unlike the legacy handler there is no queue-group fallback to reason about.
+// (subscribed on cfg.NodeName's subject), so unlike the legacy handler there
+// is no queue-group fallback to reason about.
 func handlePublishVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *nats.Msg) {
 	var req ebsprovider.PublishVolumeRequest
 	err := json.Unmarshal(msg.Data, &req)

spinifex/providers/ebs/viperblock/provider_handlers_test.go

@@ -1027,29 +1027,29 @@ func TestProviderHandlers_UnpublishVolume_AbsentIsIdempotent(t *testing.T) {
 	assert.Nil(t, resp.Error, "unpublishing a volume this node never mounted must succeed")
 }
 
-// TestProviderHandlers_PublishUnpublish_RequireNodeName covers
-// RegisterProviderSubjects' NodeName gate: PublishVolume/UnpublishVolume must
-// not be registered when cfg.NodeName is empty (PublishSubject has no node to
-// address), while every other ebs.provider.v1.* subject still is.
-func TestProviderHandlers_PublishUnpublish_RequireNodeName(t *testing.T) {
+// TestProviderHandlers_RefuseAnUnaddressableNodeName keeps viperblockd from
+// serving where the node-addressed subjects, the fence notice among them, have
+// no subject: a fence there would leave its guest running.
+func TestProviderHandlers_RefuseAnUnaddressableNodeName(t *testing.T) {
 	_, natsURL := setupEmbeddedNATS(t)
-	cfgNoNode := setupTestConfig(t, natsURL)
-	cfgNoNode.NodeName = ""
-	ncNoNode := startProviderSubjects(t, cfgNoNode, natsURL)
+	for _, node := range []string{"", "node.a", "node-*"} {
+		t.Run(node, func(t *testing.T) {
+			nc, err := nats.Connect(natsURL)
+			require.NoError(t, err)
+			t.Cleanup(nc.Close)
 
-	_, natsURL2 := setupEmbeddedNATS(t)
-	cfgNamed := setupTestConfig(t, natsURL2)
-	ncNamed := startProviderSubjects(t, cfgNamed, natsURL2)
+			register := setupTestConfig(t, natsURL)
+			register.NodeName = node
+			err = RegisterProviderSubjects(register, nc)
+			require.ErrorIs(t, err, ebsprovider.ErrInvalidArgument)
+			assert.Contains(t, err.Error(), "viperblockd node name")
 
-	assert.Equal(t, ncNamed.NumSubscriptions()-3, ncNoNode.NumSubscriptions(),
-		"an empty NodeName must skip exactly the publish, unpublish and unsealed-writes subscriptions")
-
-	// The other provider subjects must still work without a node name.
-	body := marshalRequest(t, map[string]any{"schema_version": ebsprovider.SchemaVersion})
-	msg := requestProvider(t, ncNoNode, ebsprovider.CapabilitiesSubject, body)
-	var resp ebsprovider.GetCapabilitiesResponse
-	require.NoError(t, json.Unmarshal(msg.Data, &resp))
-	assert.Nil(t, resp.Error)
+			run := setupTestConfig(t, natsURL)
+			run.NodeName = node
+			require.ErrorIs(t, run.Run(nc), ebsprovider.ErrInvalidArgument)
+			assert.Zero(t, nc.NumSubscriptions(), "a refused start must serve nothing")
+		})
+	}
 }
 
 // TestProviderHandlers_CreateVolume_SeedValidation covers the guards that must

spinifex/providers/ebs/viperblock/viperblock.go

@@ -24,6 +24,7 @@ import (
 	"github.com/mulgadc/bluebottle/pkg/masterkey"
 	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
 	"github.com/mulgadc/viperblock/viperblock"
 	"github.com/mulgadc/viperblock/viperblock/backends/s3"
@@ -826,6 +827,9 @@ func (cfg *Config) RecoverMountedVolumes(ctx context.Context, nc *nats.Conn) {
 // readiness: without the round trip a caller can act on the signal and still
 // have its request arrive before the interest does.
 func (cfg *Config) Run(nc *nats.Conn) error {
+	if err := cfg.validateNodeName(); err != nil {
+		return err
+	}
 	if err := cfg.LoadEncryptionKey(); err != nil {
 		return err
 	}
@@ -839,11 +843,7 @@ func (cfg *Config) Run(nc *nats.Conn) error {
 
 	cfg.RecoverMountedVolumes(context.Background(), nc)
 
-	if cfg.NodeName != "" {
-		slog.Info("Waiting for EBS events", "node", cfg.NodeName)
-	} else {
-		slog.Info("Waiting for EBS events (single-node mode)")
-	}
+	slog.Info("Waiting for EBS events", "node", cfg.NodeName)
 
 	if err := RegisterVolumeLifecycleSubjects(cfg, nc); err != nil {
 		return err
@@ -861,6 +861,15 @@ func (cfg *Config) Run(nc *nats.Conn) error {
 	return nil
 }
 
+// validateNodeName refuses a node name the contract cannot address. Without
+// one the fence notice has no subject, and a fenced guest would keep running.
+func (cfg *Config) validateNodeName() error {
+	if err := ebsprovider.ValidateNodeID(cfg.NodeName); err != nil {
+		return fmt.Errorf("viperblockd node name %q (cluster config node): %w", cfg.NodeName, err)
+	}
+	return nil
+}
+
 // Shutdown flushes each mounted volume's WAL and reaps idle nbdkit processes.
 // It snapshots MountedVolumes and clears it while holding the lock, then
 // flushes/kills outside the lock since VB.Close does heavy I/O.

spinifex/providers/ebs/viperblock/viperblockd_integration_test.go

@@ -839,42 +839,6 @@ func TestIntegration_ServiceGracefulShutdown(t *testing.T) {
 	t.Log("Service is running and responsive")
 }
 
-// TestIntegration_GenericTopicRouting tests that NodeName="" uses generic ebs.unmount with queue group.
-func TestIntegration_GenericTopicRouting(t *testing.T) {
-	t.Parallel()
-	if testing.Short() {
-		t.Skip("Skipping integration test in short mode")
-	}
-
-	ns, natsURL := setupEmbeddedNATS(t)
-	defer ns.Shutdown()
-
-	cfg := setupTestConfig(t, natsURL)
-	cfg.NodeName = "" // Single-node / generic mode
-	cfg.MountedVolumes = []MountedVolume{
-		{Name: "vol-generic", PID: deadpid.PID},
-	}
-
-	startTestService(t, cfg)
-
-	nc, err := nats.Connect(natsURL)
-	assert.NoError(t, err)
-	defer nc.Close()
-
-	// Send to generic topic (not ebs.{node}.unmount)
-	request := viperblocklegacyv1.EBSRequest{Name: "vol-generic"}
-	requestData, _ := json.Marshal(request)
-
-	msg, err := nc.Request("ebs.unmount", requestData, 3*time.Second)
-	assert.NoError(t, err)
-
-	var response viperblocklegacyv1.EBSUnMountResponse
-	assert.NoError(t, json.Unmarshal(msg.Data, &response))
-	assert.Equal(t, "vol-generic", response.Volume)
-	assert.False(t, response.Mounted)
-	assert.Empty(t, response.Error)
-}
-
 // TestIntegration_MountAuxiliaryVolumeSuffix tests that volumes with -cloudinit suffix
 // reach the mount handler and trigger the auxiliary volume code path.
 func TestIntegration_MountAuxiliaryVolumeSuffix(t *testing.T) {

spinifex/providers/ebs/viperblock/volume_fence.go

@@ -9,8 +9,8 @@ import (
 	"os"
 	"time"
 
-	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -319,46 +319,40 @@ func (cfg *Config) fenceNoticeRetry() time.Duration {
 }
 
 // publishVolumeFenced tells the local daemon a guest is running against a disk
-// that is gone, until the daemon acknowledges. It gives up on a closed
-// connection, or once the volume is exported here again by a newer mount.
+// that is gone, until the daemon acknowledges. It gives up on a notice that can
+// never be acted on, a closed connection, or the volume being exported again.
 func (cfg *Config) publishVolumeFenced(ctx context.Context, volumeName, winner string) {
 	if cfg.nc == nil {
 		return
 	}
-	event := viperblocklegacyv1.VolumeFencedEvent{
-		Volume: volumeName,
-		Node:   cfg.leaseOwner(),
-		Winner: winner,
-		Reason: fmt.Sprintf("volume lease moved to %s while this node had the volume open", winner),
+	notice := ebsprovider.FenceNotice{
+		Versioned: ebsprovider.NewVersioned(),
+		VolumeID:  volumeName,
+		NodeID:    cfg.NodeName,
+		Winner:    winner,
+		Reason:    fmt.Sprintf("volume lease moved to %s while this node had the volume open", winner),
 	}
-	payload, err := json.Marshal(event)
-	if err != nil {
-		slog.ErrorContext(ctx, "fence: marshal fenced event", "volume", volumeName, "err", err)
-		return
-	}
-	subject := viperblocklegacyv1.VolumeFencedSubject(cfg.NodeName)
 	began := time.Now()
 
 	for attempt := 1; ; attempt++ {
-		reqCtx, cancel := context.WithTimeout(ctx, fenceNoticeTimeout)
-		reply, err := cfg.nc.RequestWithContext(reqCtx, subject, payload)
-		cancel()
+		instanceID, err := ebsprovider.NotifyFenced(ctx, cfg.nc, notice, fenceNoticeTimeout)
 		if err == nil {
-			var ack viperblocklegacyv1.VolumeFencedAck
-			if jerr := json.Unmarshal(reply.Data, &ack); jerr == nil && ack.Error != "" {
-				slog.ErrorContext(ctx, "fence: the daemon could not act on the fenced volume",
-					"volume", volumeName, "subject", subject, "err", ack.Error)
-			} else if attempt > 1 {
+			if attempt > 1 {
 				slog.WarnContext(ctx, "fence: the daemon acknowledged the fenced volume after retrying",
-					"volume", volumeName, "instance", ack.Instance, "attempts", attempt,
+					"volume", volumeName, "instance", instanceID, "attempts", attempt,
 					"unacknowledged_ms", otelsetup.Millis(time.Since(began)))
 			}
 			return
 		}
+		if !ebsprovider.FenceNoticeRetryable(err) {
+			slog.ErrorContext(ctx, "fence: the daemon cannot act on the fenced volume, the guest is left with dead I/O",
+				"volume", volumeName, "node", notice.NodeID, "attempts", attempt, "err", err)
+			return
+		}
 
 		if cfg.nc.IsClosed() {
 			slog.ErrorContext(ctx, "fence: connection closed before the daemon acknowledged the fenced volume, the guest may be left with dead I/O",
-				"volume", volumeName, "subject", subject, "attempts", attempt)
+				"volume", volumeName, "node", notice.NodeID, "attempts", attempt)
 			return
 		}
 		if _, again := cfg.lookupMountedVolume(volumeName); again {
@@ -368,7 +362,7 @@ func (cfg *Config) publishVolumeFenced(ctx context.Context, volumeName, winner s
 		}
 		if attempt == 1 || attempt%fenceNoticeLogEveryTries == 0 {
 			slog.ErrorContext(ctx, "fence: the daemon has not acknowledged the fenced volume, retrying; until it does the guest keeps running with dead I/O",
-				"volume", volumeName, "subject", subject, "attempts", attempt,
+				"volume", volumeName, "node", notice.NodeID, "attempts", attempt,
 				"unacknowledged_ms", otelsetup.Millis(time.Since(began)), "err", err)
 		}
 

spinifex/providers/ebs/viperblock/volume_fence_invariants_test.go

@@ -11,7 +11,7 @@ import (
 	"testing"
 	"time"
 
-	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -133,16 +133,16 @@ func TestS4_FenceHolderReadIsBounded(t *testing.T) {
 // fenceResponder answers fence notices as the daemon does and records them.
 type fenceResponder struct {
 	mu     sync.Mutex
-	events []viperblocklegacyv1.VolumeFencedEvent
+	events []ebsprovider.FenceNotice
 }
 
 func (r *fenceResponder) handle(msg *nats.Msg) {
-	var event viperblocklegacyv1.VolumeFencedEvent
-	_ = json.Unmarshal(msg.Data, &event)
+	var notice ebsprovider.FenceNotice
+	_ = json.Unmarshal(msg.Data, &notice)
 	r.mu.Lock()
-	r.events = append(r.events, event)
+	r.events = append(r.events, notice)
 	r.mu.Unlock()
-	payload, _ := json.Marshal(viperblocklegacyv1.VolumeFencedAck{Instance: "i-fenced"})
+	payload, _ := json.Marshal(ebsprovider.FenceNoticeAck{Versioned: ebsprovider.ReplyVersioned(notice.SchemaVersion), InstanceID: "i-fenced"})
 	_ = msg.Respond(payload)
 }
 
@@ -185,7 +185,9 @@ func TestS4_FenceNoticeSurvivesAMissedDelivery(t *testing.T) {
 	require.NoError(t, err)
 	t.Cleanup(daemonConn.Close)
 	responder := &fenceResponder{}
-	_, err = daemonConn.Subscribe(viperblocklegacyv1.VolumeFencedSubject("node-a"), responder.handle)
+	subject, err := ebsprovider.FenceNoticeSubject("node-a")
+	require.NoError(t, err)
+	_, err = daemonConn.Subscribe(subject, responder.handle)
 	require.NoError(t, err)
 	require.NoError(t, daemonConn.Flush())
 
@@ -197,7 +199,10 @@ func TestS4_FenceNoticeSurvivesAMissedDelivery(t *testing.T) {
 	require.GreaterOrEqualf(t, responder.count(), 1,
 		"ADR-0007:S4: a daemon that subscribed after the first notice never heard the fence, so its guest runs on with no disk. "+
 			"Fix: retry publishVolumeFenced (volume_fence.go) until the daemon acknowledges")
-	assert.Equal(t, volumeName, responder.events[0].Volume)
+	first := responder.events[0]
+	assert.Equal(t, volumeName, first.VolumeID)
+	assert.Equal(t, "node-a", first.NodeID)
+	assert.True(t, ebsprovider.SupportsSchemaVersion(first.SchemaVersion), "the daemon refuses a notice in a version it cannot read")
 
 	seen := responder.count()
 	time.Sleep(5 * cfg.fenceNoticeEvery)
@@ -252,3 +257,93 @@ func TestS4_FenceNoticeStopsWithTheConnection(t *testing.T) {
 		t.Fatal("a fence notice kept retrying on a closed connection")
 	}
 }
+
+// TestFenceNotice_UnavailableDaemonIsAskedAgain keeps a refusal that says
+// "not now" from ending delivery: the guest is still running on a dead disk.
+func TestFenceNotice_UnavailableDaemonIsAskedAgain(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+
+	const volumeName = "vol-fencenoticeunavailable"
+	cfg, nc := noticeConfig(t, natsURL, volumeName)
+
+	subject, err := ebsprovider.FenceNoticeSubject("node-a")
+	require.NoError(t, err)
+	unavailable, err := json.Marshal(ebsprovider.FenceNoticeAck{
+		Versioned: ebsprovider.NewVersioned(),
+		Error:     &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnavailable, Message: "not yet"},
+	})
+	require.NoError(t, err)
+	ack, err := json.Marshal(ebsprovider.FenceNoticeAck{Versioned: ebsprovider.NewVersioned(), InstanceID: "i-fenced"})
+	require.NoError(t, err)
+	var heard atomic.Int32
+	_, err = nc.Subscribe(subject, func(msg *nats.Msg) {
+		if heard.Add(1) == 1 {
+			_ = msg.Respond(unavailable)
+			return
+		}
+		_ = msg.Respond(ack)
+	})
+	require.NoError(t, err)
+	require.NoError(t, nc.Flush())
+
+	done := make(chan struct{})
+	go func() { defer close(done); cfg.onVolumeLeaseLost(context.Background(), volumeName, leaseLostStalled) }()
+	select {
+	case <-done:
+	case <-time.After(fenceNoticeTimeout + 2*time.Second):
+		t.Fatal("a fence notice was still being sent after the daemon acknowledged it")
+	}
+	assert.Equal(t, int32(2), heard.Load(), "ADR-0007:S4: a daemon that was unavailable was never asked again")
+}
+
+// TestFenceNotice_UnsendableNoticeIsNotRetried stops a notice that can never
+// be addressed from spinning until shutdown on a goroutine nobody watches.
+func TestFenceNotice_UnsendableNoticeIsNotRetried(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+
+	const volumeName = "vol-fencenoticeunsendable"
+	cfg, _ := noticeConfig(t, natsURL, volumeName)
+	cfg.NodeName = ""
+	cfg.MountedVolumes = nil // gone, as after the teardown, so only the error can end the retry
+
+	done := make(chan struct{})
+	go func() { defer close(done); cfg.publishVolumeFenced(t.Context(), volumeName, "node-b") }()
+	select {
+	case <-done:
+	case <-time.After(fenceNoticeTimeout + 2*time.Second):
+		t.Fatal("a fence notice with no subject to send it on kept retrying")
+	}
+}
+
+// TestFenceNotice_RefusedNoticeIsNotRepeated keeps a notice the daemon heard and
+// could not act on from being resent forever: a refusal is an answer.
+func TestFenceNotice_RefusedNoticeIsNotRepeated(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+
+	const volumeName = "vol-fencenoticerefused"
+	cfg, nc := noticeConfig(t, natsURL, volumeName)
+
+	subject, err := ebsprovider.FenceNoticeSubject("node-a")
+	require.NoError(t, err)
+	var heard atomic.Int32
+	refusal, err := json.Marshal(ebsprovider.FenceNoticeAck{
+		Versioned: ebsprovider.NewVersioned(),
+		Error:     &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnsupportedVersion, Message: "no"},
+	})
+	require.NoError(t, err)
+	_, err = nc.Subscribe(subject, func(msg *nats.Msg) {
+		heard.Add(1)
+		_ = msg.Respond(refusal)
+	})
+	require.NoError(t, err)
+	require.NoError(t, nc.Flush())
+
+	done := make(chan struct{})
+	go func() { defer close(done); cfg.onVolumeLeaseLost(context.Background(), volumeName, leaseLostStalled) }()
+	select {
+	case <-done:
+	case <-time.After(fenceNoticeTimeout + 2*time.Second):
+		t.Fatal("a fence notice kept retrying after the daemon refused it")
+	}
+	assert.Equal(t, int32(1), heard.Load())
+}

spinifex/providers/ebs/viperblock/volume_fence_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -171,15 +172,25 @@ func TestVolumeFence_CompletesOnceTheWriterFinallyExits(t *testing.T) {
 
 	const volumeName = "vol-fencekilllate"
 
-	cfg := fencedConfig(t, natsURL, "node-a", volumeName)
+	cfg, nc := noticeConfig(t, natsURL, volumeName)
 	cfg.MountedVolumes[0].PID = 0 // rejected by ForceKillProcess, so the kill cannot succeed
 	// Alive for the guard and one watch pass, gone on the next: the fence has to
 	// see the failure persist and then resolve.
 	var probes atomic.Int32
 	cfg.processAlive = func(int) bool { return probes.Add(1) <= 2 }
 	cfg.fenceWatchEvery = 10 * time.Millisecond
+	responder := &fenceResponder{}
+	subject, err := ebsprovider.FenceNoticeSubject("node-a")
+	require.NoError(t, err)
+	_, err = nc.Subscribe(subject, responder.handle)
+	require.NoError(t, err)
+	require.NoError(t, nc.Flush())
 
 	cfg.fenceVolume(t.Context(), volumeName, leaseLostToPeer)
+	require.Equalf(t, 1, responder.count(),
+		"ADR-0007:S4: a fence that completed late never told the daemon, so its guest runs on with no disk. "+
+			"Fix: publishVolumeFenced once awaitFencedWriterExit sees the writer gone (volume_fence.go)")
+	assert.Equal(t, volumeName, responder.events[0].VolumeID)
 
 	cfg.mu.Lock()
 	mounted := len(cfg.MountedVolumes)

spinifex/runtime/compute/vm/record.go

@@ -106,6 +106,10 @@ type InstanceStatus struct {
 	EBSRequests       []EBSRequest   `json:"ebs_requests,omitempty"`
 	ENIAvailableSlots []int          `json:"eni_available_slots,omitempty"`
 	ENIAttachedByID   map[string]int `json:"eni_attached_by_id,omitempty"`
+
+	// UEFIVarStore is optional, so records written without it still decode
+	// and the bucket schema does not change.
+	UEFIVarStore *UEFIVarStoreRef `json:"uefi_var_store,omitempty"`
 }
 
 // Record projects a VM onto the persisted record. QMPClient and the request
@@ -166,6 +170,7 @@ func (v *VM) Record() *InstanceRecord {
 			EBSRequests:                     v.EBSRequests.Requests,
 			ENIAvailableSlots:               v.ENIRequests.AvailableSlots,
 			ENIAttachedByID:                 v.ENIRequests.AttachedByENIID,
+			UEFIVarStore:                    v.UEFIVarStore,
 		},
 	}
 }
@@ -222,6 +227,7 @@ func VMFromRecord(r *InstanceRecord) *VM {
 		TerminatedAt:                    r.Status.TerminatedAt,
 		ShuttingDownAt:                  r.Status.ShuttingDownAt,
 		FirstLaunch:                     r.Status.FirstLaunch,
+		UEFIVarStore:                    r.Status.UEFIVarStore,
 	}
 	v.EBSRequests.Requests = r.Status.EBSRequests
 	v.ENIRequests.AvailableSlots = r.Status.ENIAvailableSlots

spinifex/runtime/compute/vm/record_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/instance/uefistore"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
@@ -114,6 +115,37 @@ func TestRecord_RoundTripThroughJSON(t *testing.T) {
 	assertVMFieldsEqual(t, original, vm.VMFromRecord(&decoded))
 }
 
+// The pointer is optional: a record written before it existed decodes with
+// none, and a record without one writes no key, so the bucket schema holds.
+func TestRecord_UEFIVarStoreIsOptional(t *testing.T) {
+	var decoded vm.InstanceRecord
+	require.NoError(t, json.Unmarshal([]byte(`{"metadata":{"name":"i-old"},"spec":{"config":{}},"status":{"status":"stopped"}}`), &decoded))
+	assert.Nil(t, decoded.Status.UEFIVarStore)
+
+	encoded, err := json.Marshal((&vm.VM{ID: "i-bios"}).Record())
+	require.NoError(t, err)
+	assert.NotContains(t, string(encoded), "uefi_var_store")
+}
+
+// The pointer's wire names are what the record carries across releases, so
+// they are pinned rather than left to follow the Go field names.
+func TestRecord_UEFIVarStoreWireShape(t *testing.T) {
+	encoded, err := json.Marshal(populatedVM().Record())
+	require.NoError(t, err)
+
+	var wire struct {
+		Status struct {
+			UEFIVarStore map[string]json.RawMessage `json:"uefi_var_store"`
+		} `json:"status"`
+	}
+	require.NoError(t, json.Unmarshal(encoded, &wire))
+	ref := wire.Status.UEFIVarStore
+	assert.JSONEq(t, `"uefi/i-round-trip/0d6f3c1e-5b7a-4f43-9a8e-2c1d4e5f6a7b"`, string(ref["key"]))
+	assert.JSONEq(t, `{"arch":"arm64","vars_size":67108864,"code_sha256":"ab12"}`, string(ref["firmware"]))
+	assert.JSONEq(t, `"node-3"`, string(ref["writer_node"]))
+	assert.JSONEq(t, `"2026-08-28T04:00:00Z"`, string(ref["written_at"]))
+}
+
 // The mutexes are dropped on the way out, so the rebuilt instance must arrive
 // with usable zero-value locks rather than ones somebody else left held.
 func TestVMFromRecord_RequestWrappersAreUsable(t *testing.T) {
@@ -206,6 +238,12 @@ func populatedVM() *vm.VM {
 		CapacityReservationId: "cr-1",
 		InstanceLifecycle:     "spot",
 		SpotInstanceRequestId: "sir-1",
+		UEFIVarStore: &vm.UEFIVarStoreRef{
+			Key:        "uefi/i-round-trip/0d6f3c1e-5b7a-4f43-9a8e-2c1d4e5f6a7b",
+			Firmware:   uefistore.FirmwareIdentity{Arch: "arm64", VarsSize: 67108864, CodeSHA256: "ab12"},
+			WriterNode: "node-3",
+			WrittenAt:  time.Date(2026, 8, 28, 4, 0, 0, 0, time.UTC),
+		},
 
 		Status:                          vm.StateRunning,
 		Instance:                        &ec2.Instance{InstanceId: aws.String("i-round-trip")},

spinifex/runtime/compute/vm/vm.go

@@ -19,6 +19,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/instance/uefistore"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
 	"github.com/mulgadc/spinifex/spinifex/runtime/host/hostos"
@@ -308,6 +309,20 @@ type VM struct {
 	// tolerated by migrate.
 	InstanceLifecycle     string `json:"instance_lifecycle,omitempty"`
 	SpotInstanceRequestId string `json:"spot_instance_request_id,omitempty"`
+
+	// UEFIVarStore points at the object holding this instance's UEFI variable
+	// store. Status, so written only by the node; nil for a BIOS instance.
+	UEFIVarStore *UEFIVarStoreRef `json:"uefi_var_store,omitempty"`
+}
+
+// UEFIVarStoreRef names the store object an instance launches from, with the
+// firmware it was written under, so a launch can check it without reading the
+// object's metadata.
+type UEFIVarStoreRef struct {
+	Key        string                     `json:"key"`
+	Firmware   uefistore.FirmwareIdentity `json:"firmware"`
+	WriterNode string                     `json:"writer_node"`
+	WrittenAt  time.Time                  `json:"written_at"`
 }
 
 // UnmarshalJSON folds the fields of legacy records that mixed a request with an

spinifex/runtime/daemon/daemon.go

@@ -41,7 +41,6 @@ import (
 	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1"
 	telemetryv1 "github.com/mulgadc/spinifex/contracts/telemetry/v1"
-	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/preflight"
 	acmdomain "github.com/mulgadc/spinifex/spinifex/domains/acm"
@@ -733,6 +732,11 @@ func (d *Daemon) SetConfigPath(path string) {
 
 // NewDaemon creates a new daemon instance.
 func NewDaemon(cfg *config.ClusterConfig) (*Daemon, error) {
+	// The fence notice is node-addressed: a daemon that cannot subscribe to it
+	// would leave a guest running on a volume whose export was revoked.
+	if err := ebsprovider.ValidateNodeID(cfg.Node); err != nil {
+		return nil, fmt.Errorf("cluster config node %q: %w", cfg.Node, err)
+	}
 	ctx, cancel := context.WithCancel(context.Background())
 
 	// If WalDir is not set, use BaseDir
@@ -1353,10 +1357,8 @@ func (d *Daemon) subscribeAll() error {
 		// here, so the reconcile request flows daemon-ward.
 		natsSub{dhcp.TopicLeaseChanged, d.handleDHCPLeaseChanged, "spinifex-workers"},
 		natsSub{dhcp.TopicOwnerCheck, d.handleDHCPOwnerCheck, "spinifex-workers"},
-		// Node-addressed with no queue group: the fenced guest is on this host,
-		// so a worker on another node would find nothing to stop.
-		natsSub{viperblocklegacyv1.VolumeFencedSubject(d.node), d.handleVolumeFenced, ""},
 	)
+	subs = append(subs, d.fenceNoticeSubs()...)
 
 	// Per-node, no queue group: a second daemon claiming this node ID also
 	// answers, so the teardown probe sees a duplicate and refuses to seal. An
@@ -2754,6 +2756,10 @@ func (d *Daemon) initJetStream() error {
 			err = d.jsManager.InitClusterConfigBucket()
 		}
 
+		if err == nil {
+			err = d.jsManager.InitUEFIStoreBucket()
+		}
+
 		if err == nil {
 			d.jsManager.SetSyncObserver(d)
 			// Replicas cannot be undeclared here: bucket creation refuses

spinifex/runtime/daemon/daemon_handlers.go

@@ -85,13 +85,9 @@ func respondWithServiceError(nodeID string, msg *nats.Msg, err error) {
 	payload := awserrors.GenerateErrorPayloadWithMessage(awserrors.ValidErrorCodeFromError(err), message)
 	// The AWS body stays as it is; the header is how ec2.start tells its
 	// forwarder this refusal must not be answered by starting elsewhere.
-	extra := nats.Header{}
+	var extra nats.Header
 	if _, ok := errors.AsType[*ec2instance.UnsealedWritesError](err); ok {
-		extra.Set(natsmsg.UnsealedWritesHeader, "true")
-	}
-	// A start caller cannot tell a held volume from any other IncorrectState.
-	if errors.Is(err, ec2v1.ErrVolumeHeldElsewhere) {
-		extra.Set(natsmsg.VolumeHeldElsewhereHeader, "true")
+		extra = nats.Header{natsmsg.UnsealedWritesHeader: []string{"true"}}
 	}
 	respondNATSMsgWithHeader(nodeID, msg, payload, extra)
 }

spinifex/runtime/daemon/daemon_handlers_fence.go

@@ -3,56 +3,77 @@ package daemon
 import (
 	"context"
 	"encoding/json"
-	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"fmt"
 	"log/slog"
 
-	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	"github.com/nats-io/nats.go"
 )
 
-// handleVolumeFenced stops the local guest whose volume viperblockd has just
-// torn the export out from under.
+// fenceNoticeSubs is this node's fence-notice subscription, whatever the EBS
+// provider. Node-addressed with no queue group: the fenced guest is on this
+// host, so a worker on another node would find nothing to stop.
+func (d *Daemon) fenceNoticeSubs() []natsSub {
+	subject, err := ebsprovider.FenceNoticeSubject(d.node)
+	if err != nil {
+		slog.Error("Not serving EBS fence notices: node ID is not a single NATS subject token", "node", d.node, "err", err)
+		return nil
+	}
+	return []natsSub{{subject, d.handleFenceNotice, ""}}
+}
+
+// handleFenceNotice stops the local guest whose volume the EBS provider has
+// just revoked the export of.
 //
 // By the time this arrives the disk is already gone, so the guest is doing no
 // useful work and its writes cannot reach the backend. What is left is a QEMU
 // process holding memory and taps, and an instance record claiming this node
 // runs something that runs somewhere else now.
-func (d *Daemon) handleVolumeFenced(msg *nats.Msg) string {
-	ctx, span := natsmsg.StartConsumerSpan(msg)
+func (d *Daemon) handleFenceNotice(msg *nats.Msg) string {
+	ctx, span := ebsprovider.StartServerSpan(context.Background(), msg)
 	defer span.End()
 
-	var event viperblocklegacyv1.VolumeFencedEvent
-	if err := json.Unmarshal(msg.Data, &event); err != nil {
-		slog.ErrorContext(ctx, "volume fenced: bad event", "err", err)
-		ackVolumeFenced(ctx, msg, viperblocklegacyv1.VolumeFencedAck{Error: "bad event: " + err.Error()})
+	var notice ebsprovider.FenceNotice
+	decodeErr := json.Unmarshal(msg.Data, &notice)
+	ack := ebsprovider.FenceNoticeAck{Versioned: ebsprovider.ReplyVersioned(notice.SchemaVersion)}
+	refuse := func(code ebsprovider.ErrorCode, message string) string {
+		slog.ErrorContext(ctx, "volume fenced: refused the notice", "volumeId", notice.VolumeID, "code", code, "err", message)
+		ack.Error = &ebsprovider.ProviderError{Code: code, Message: message}
+		ebsprovider.RecordProviderError(span, ack.Error)
+		ackFenceNotice(ctx, msg, ack)
 		return outcomeError
 	}
-	if event.Volume == "" {
-		slog.ErrorContext(ctx, "volume fenced: event named no volume")
-		ackVolumeFenced(ctx, msg, viperblocklegacyv1.VolumeFencedAck{Error: "event named no volume"})
-		return outcomeError
+	if decodeErr != nil {
+		return refuse(ebsprovider.ErrorCodeInvalidArgument, "bad notice: "+decodeErr.Error())
+	}
+	if !ebsprovider.SupportsSchemaVersion(notice.SchemaVersion) {
+		return refuse(ebsprovider.ErrorCodeUnsupportedVersion,
+			fmt.Sprintf("unsupported schema version %d; ", notice.SchemaVersion)+ebsprovider.SupportedVersionsText())
+	}
+	if notice.VolumeID == "" {
+		return refuse(ebsprovider.ErrorCodeInvalidArgument, "notice named no volume")
 	}
 
 	// FenceVolume moves the guest out of running before it returns, so the ack
 	// below is sent only once nothing here will resume it.
-	instanceID := d.vmMgr.FenceVolume(ctx, event.Volume, event.Reason)
-	ackVolumeFenced(ctx, msg, viperblocklegacyv1.VolumeFencedAck{Instance: instanceID})
-	if instanceID == "" {
+	ack.InstanceID = d.vmMgr.FenceVolume(ctx, notice.VolumeID, notice.Reason)
+	ackFenceNotice(ctx, msg, ack)
+	if ack.InstanceID == "" {
 		// The usual case: the volume outlived the guest that was using it, so
 		// there is nothing local left to stop.
 		slog.InfoContext(ctx, "volume fenced with no local guest using it",
-			"volumeId", event.Volume, "winner", event.Winner)
+			"volumeId", notice.VolumeID, "winner", notice.Winner)
 		return outcomeSuccess
 	}
 
 	slog.ErrorContext(ctx, "volume fenced, stopped the guest that was using it",
-		"volumeId", event.Volume, "instanceId", instanceID, "winner", event.Winner)
+		"volumeId", notice.VolumeID, "instanceId", ack.InstanceID, "winner", notice.Winner)
 	return outcomeSuccess
 }
 
-// ackVolumeFenced answers viperblockd, which repeats the notice until it hears
+// ackFenceNotice answers the provider, which repeats the notice until it hears
 // back. A notice with no reply subject was published once and needs no answer.
-func ackVolumeFenced(ctx context.Context, msg *nats.Msg, ack viperblocklegacyv1.VolumeFencedAck) {
+func ackFenceNotice(ctx context.Context, msg *nats.Msg, ack ebsprovider.FenceNoticeAck) {
 	if msg.Reply == "" {
 		return
 	}
@@ -62,6 +83,6 @@ func ackVolumeFenced(ctx context.Context, msg *nats.Msg, ack viperblocklegacyv1.
 		return
 	}
 	if err := msg.Respond(payload); err != nil {
-		slog.WarnContext(ctx, "volume fenced: could not acknowledge, viperblockd will repeat the notice", "err", err)
+		slog.WarnContext(ctx, "volume fenced: could not acknowledge, the provider will repeat the notice", "err", err)
 	}
 }

spinifex/runtime/daemon/daemon_handlers_fence_invariants_test.go

@@ -2,92 +2,313 @@ package daemon
 
 import (
 	"encoding/json"
+	"fmt"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
-	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
 	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/codes"
 )
 
-// fenceDaemon is a daemon running one guest on volumeID, answering fence
-// notices on a real NATS server as it does in production.
+// fenceDaemon is a daemon running one guest on volumeID, subscribed to fence
+// notices on a real NATS server through the subscription it makes in production.
 func fenceDaemon(t *testing.T, volumeID string) (*Daemon, *vm.VM, *nats.Conn) {
+	t.Helper()
+	return fenceDaemonWithManager(t, volumeID, vm.NewManagerWithDeps(vm.Deps{NodeID: "node-a"}))
+}
+
+func fenceDaemonWithManager(t *testing.T, volumeID string, mgr *vm.Manager) (*Daemon, *vm.VM, *nats.Conn) {
 	t.Helper()
 	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
 
 	_, nc := testutil.StartTestNATS(t)
-	d := &Daemon{node: "node-a", vmMgr: vm.NewManagerWithDeps(vm.Deps{NodeID: "node-a"})}
-	guest := &vm.VM{
+	d := &Daemon{
+		node:              "node-a",
+		vmMgr:             mgr,
+		natsConn:          nc,
+		natsSubscriptions: map[string]*nats.Subscription{},
+	}
+	guest := fencedGuest(volumeID)
+	d.vmMgr.Insert(guest)
+
+	require.NoError(t, d.registerNatsSubs(d.fenceNoticeSubs()))
+	require.NoError(t, nc.Flush())
+	return d, guest, nc
+}
+
+func fencedGuest(volumeID string) *vm.VM {
+	return &vm.VM{
 		ID:          "i-fencedguest",
 		Status:      vm.StateRunning,
 		Instance:    &ec2.Instance{},
 		EBSRequests: vm.EBSRequests{Requests: []vm.EBSRequest{{Name: volumeID}}},
 	}
-	d.vmMgr.Insert(guest)
+}
 
-	_, err := nc.Subscribe(viperblocklegacyv1.VolumeFencedSubject(d.node), func(msg *nats.Msg) { d.handleVolumeFenced(msg) })
+// requestFenceNotice sends raw to node-a's fence-notice subject and decodes the ack.
+func requestFenceNotice(t *testing.T, nc *nats.Conn, raw []byte) ebsprovider.FenceNoticeAck {
+	t.Helper()
+	subject, err := ebsprovider.FenceNoticeSubject("node-a")
 	require.NoError(t, err)
-	require.NoError(t, nc.Flush())
-	return d, guest, nc
+	reply, err := nc.Request(subject, raw, 2*time.Second)
+	require.NoErrorf(t, err,
+		"ADR-0007:S4: the daemon did not answer a fence notice, so the provider can never tell it was heard. "+
+			"Fix: subscribe fenceNoticeSubs and respond in handleFenceNotice (daemon_handlers_fence.go)")
+	var ack ebsprovider.FenceNoticeAck
+	require.NoError(t, json.Unmarshal(reply.Data, &ack))
+	return ack
+}
+
+func fenceNoticePayload(t *testing.T, notice ebsprovider.FenceNotice) []byte {
+	t.Helper()
+	payload, err := json.Marshal(notice)
+	require.NoError(t, err)
+	return payload
+}
+
+// memoryProviderPublishing is a non-viperblock provider holding volumeID
+// published to node-a, announcing fences on nc.
+func memoryProviderPublishing(t *testing.T, nc *nats.Conn, volumeID string) *ebsprovider.MemoryProvider {
+	t.Helper()
+	provider := ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})
+	provider.NotifyFencesOn(nc)
+	_, err := provider.CreateVolume(t.Context(), ebsprovider.CreateVolumeRequest{
+		Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 1 << 30},
+	})
+	require.NoError(t, err)
+	_, err = provider.PublishVolume(t.Context(), ebsprovider.PublishVolumeRequest{
+		Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: "node-a",
+	})
+	require.NoError(t, err)
+	return provider
 }
 
 // TestS4_FenceNoticeIsAcknowledgedOnlyOnceTheGuestIsStopped enforces ADR-0007:S4
-// on the daemon's half of the fence. viperblockd repeats the notice until it is
+// on the daemon's half of the fence. The provider repeats the notice until it is
 // answered, so the answer is what ends delivery, and it must not come before the
 // guest is out of running, or a lost notice is acknowledged for a guest left up.
 func TestS4_FenceNoticeIsAcknowledgedOnlyOnceTheGuestIsStopped(t *testing.T) {
 	d, guest, nc := fenceDaemon(t, "vol-fencedack")
 
-	payload, err := json.Marshal(viperblocklegacyv1.VolumeFencedEvent{
-		Volume: "vol-fencedack", Node: "node-a", Winner: "node-b", Reason: "volume lease moved to node-b",
-	})
-	require.NoError(t, err)
-
-	reply, err := nc.Request(viperblocklegacyv1.VolumeFencedSubject(d.node), payload, 2*time.Second)
-	require.NoErrorf(t, err,
-		"ADR-0007:S4: the daemon did not answer a fence notice, so viperblockd can never tell it was heard. "+
-			"Fix: respond in handleVolumeFenced (daemon_handlers_fence.go)")
-
-	var ack viperblocklegacyv1.VolumeFencedAck
-	require.NoError(t, json.Unmarshal(reply.Data, &ack))
-	assert.Equal(t, guest.ID, ack.Instance, "the ack names the guest the fence stopped")
-	assert.Empty(t, ack.Error)
+	ack := requestFenceNotice(t, nc, fenceNoticePayload(t, ebsprovider.FenceNotice{
+		Versioned: ebsprovider.NewVersioned(),
+		VolumeID:  "vol-fencedack", NodeID: "node-a", Winner: "node-b", Reason: "volume lease moved to node-b",
+	}))
+	assert.Equal(t, guest.ID, ack.InstanceID, "the ack names the guest the fence stopped")
+	assert.Nil(t, ack.Error)
 	assert.NotEqual(t, vm.StateRunning, d.vmMgr.Status(guest),
 		"ADR-0007:S4: the fence was acknowledged while the guest was still running")
 }
 
+// TestS4_FenceNoticeIsNotAcknowledgedWhileTheStopIsInProgress enforces
+// ADR-0007:S4 on the order of the stop and the ack: with the stop held open,
+// no answer may reach the provider until it completes.
+func TestS4_FenceNoticeIsNotAcknowledgedWhileTheStopIsInProgress(t *testing.T) {
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) })
+
+	var mgr *vm.Manager
+	mgr = vm.NewManagerWithDeps(vm.Deps{NodeID: "node-a", TransitionState: func(v *vm.VM, target vm.InstanceState) error {
+		close(entered)
+		<-release
+		mgr.Inspect(v, func(v *vm.VM) { v.Status = target })
+		return nil
+	}})
+	_, guest, nc := fenceDaemonWithManager(t, "vol-fenceheld", mgr)
+
+	subject, err := ebsprovider.FenceNoticeSubject("node-a")
+	require.NoError(t, err)
+	type answer struct {
+		msg *nats.Msg
+		err error
+	}
+	answers := make(chan answer, 1)
+	payload := fenceNoticePayload(t, ebsprovider.FenceNotice{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-fenceheld", NodeID: "node-a"})
+	go func() {
+		msg, err := nc.Request(subject, payload, 10*time.Second)
+		answers <- answer{msg, err}
+	}()
+
+	select {
+	case <-entered:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the fence never reached the guest's stop, so this test proved nothing about the ack order")
+	}
+	select {
+	case got := <-answers:
+		t.Fatalf("ADR-0007:S4: the fence was acknowledged (%v) before the guest's stop finished. "+
+			"Fix: ack only after FenceVolume returns in handleFenceNotice (daemon_handlers_fence.go)", got.err)
+	case <-time.After(300 * time.Millisecond):
+	}
+
+	releaseOnce.Do(func() { close(release) })
+	var got answer
+	select {
+	case got = <-answers:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the fence was never acknowledged once the stop finished")
+	}
+	require.NoError(t, got.err)
+	var ack ebsprovider.FenceNoticeAck
+	require.NoError(t, json.Unmarshal(got.msg.Data, &ack))
+	assert.Equal(t, guest.ID, ack.InstanceID)
+}
+
+// TestS4_FenceNoticeIsServedByTheProductionSubscriptionTable keeps the
+// subscription in subscribeAll: a handler nothing subscribes is a fence no
+// daemon hears.
+func TestS4_FenceNoticeIsServedByTheProductionSubscriptionTable(t *testing.T) {
+	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
+	ns, nc, _ := testutil.StartTestJetStream(t)
+
+	d := newSubscribeTestDaemon(t, nc, "", "")
+	d.node = "node-a"
+	d.vmMgr = vm.NewManagerWithDeps(vm.Deps{NodeID: "node-a"})
+	guest := fencedGuest("vol-fencedsubscribed")
+	d.vmMgr.Insert(guest)
+	require.NoError(t, d.subscribeAll())
+	require.NoError(t, nc.Flush())
+
+	caller, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	t.Cleanup(caller.Close)
+	subject, err := ebsprovider.FenceNoticeSubject("node-a")
+	require.NoError(t, err)
+	reply, err := caller.Request(subject, fenceNoticePayload(t, ebsprovider.FenceNotice{
+		Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-fencedsubscribed", NodeID: "node-a",
+	}), 2*time.Second)
+	require.NoErrorf(t, err,
+		"ADR-0007:S4: the running daemon does not serve fence notices. Fix: append fenceNoticeSubs in subscribeAll (daemon.go)")
+	var ack ebsprovider.FenceNoticeAck
+	require.NoError(t, json.Unmarshal(reply.Data, &ack))
+	assert.Nil(t, ack.Error)
+	assert.Equal(t, guest.ID, ack.InstanceID)
+}
+
+// TestNewDaemon_RefusesANodeTheFenceNoticeCannotAddress keeps a daemon from
+// starting where no fence notice can reach it, which would leave a guest
+// running on a revoked export with nothing but a log line to say so.
+func TestNewDaemon_RefusesANodeTheFenceNoticeCannotAddress(t *testing.T) {
+	for _, node := range []string{"", "node.a", "node-*", "node>"} {
+		t.Run(node, func(t *testing.T) {
+			d, err := NewDaemon(&config.ClusterConfig{Node: node, Nodes: map[string]config.Config{node: {}}})
+			require.ErrorIs(t, err, ebsprovider.ErrInvalidArgument)
+			assert.Contains(t, err.Error(), "cluster config node")
+			assert.Nil(t, d)
+		})
+	}
+}
+
+// TestFenceNotice_SpanAttributesTheAccountAndTheRefusal reads the daemon's
+// span for a refused notice: a server span for the fenced verb, carrying the
+// sender's account and the refusal's code rather than a successful round trip.
+func TestFenceNotice_SpanAttributesTheAccountAndTheRefusal(t *testing.T) {
+	recorder := testutil.RecordSpans(t)
+	_, _, nc := fenceDaemon(t, "vol-unused")
+
+	ctx := ebsprovider.WithAccountID(t.Context(), "123456789012")
+	_, err := ebsprovider.NotifyFenced(ctx, nc, ebsprovider.FenceNotice{
+		Versioned: ebsprovider.Versioned{SchemaVersion: ebsprovider.SchemaVersion + 1}, VolumeID: "vol-unused", NodeID: "node-a",
+	}, 2*time.Second)
+	require.ErrorIs(t, err, ebsprovider.ErrUnsupportedVersion)
+
+	_, server := testutil.SpansByKind(t, recorder)
+	assert.Equal(t, "ebs.volume.fenced", server.Name())
+	assert.Equal(t, codes.Error, server.Status().Code)
+	assert.Equal(t, string(ebsprovider.ErrorCodeUnsupportedVersion), testutil.SpanAttribute(t, server, "ebs.provider.error_code").AsString())
+	assert.Equal(t, "123456789012", testutil.SpanAttribute(t, server, "aws.account_id").AsString())
+}
+
+// TestFenceNotice_AckEchoesTheNoticeVersion keeps the answer readable by the
+// sender: an ack in a version other than the notice's is one it may refuse.
+func TestFenceNotice_AckEchoesTheNoticeVersion(t *testing.T) {
+	for _, version := range ebsprovider.SupportedSchemaVersions() {
+		_, _, nc := fenceDaemon(t, "vol-versioned")
+		ack := requestFenceNotice(t, nc, fenceNoticePayload(t, ebsprovider.FenceNotice{
+			Versioned: ebsprovider.Versioned{SchemaVersion: version}, VolumeID: "vol-versioned", NodeID: "node-a",
+		}))
+		assert.Nil(t, ack.Error)
+		assert.Equalf(t, version, ack.SchemaVersion, "a notice at version %d was answered at another", version)
+	}
+}
+
+// TestS4_FenceNoticeFromAnyProviderStopsTheGuest holds the daemon to the
+// contract rather than to viperblockd: a provider that is not viperblock
+// revoking the export a local guest runs on gets that guest stopped.
+func TestS4_FenceNoticeFromAnyProviderStopsTheGuest(t *testing.T) {
+	d, guest, nc := fenceDaemon(t, "vol-memoryfenced")
+	provider := memoryProviderPublishing(t, nc, "vol-memoryfenced")
+
+	stopped, err := provider.RevokePublication(t.Context(), "vol-memoryfenced", "node-b")
+	require.NoErrorf(t, err,
+		"ADR-0007:S4: the daemon did not take a fence notice from a non-viperblock provider. "+
+			"Fix: subscribe the contract fence-notice subject in fenceNoticeSubs (daemon_handlers_fence.go)")
+	assert.Equal(t, guest.ID, stopped)
+	assert.Equal(t, vm.StateError, d.vmMgr.Status(guest),
+		"ADR-0007:S4: the guest runs on against an export its provider revoked")
+}
+
+// TestFenceNotice_VolumeNoLocalGuestUsesChangesNothing is the ordinary case: a
+// volume fenced after its guest has gone leaves every local guest as it was.
+func TestFenceNotice_VolumeNoLocalGuestUsesChangesNothing(t *testing.T) {
+	d, guest, nc := fenceDaemon(t, "vol-inuse")
+	provider := memoryProviderPublishing(t, nc, "vol-noguest")
+
+	stopped, err := provider.RevokePublication(t.Context(), "vol-noguest", "node-b")
+	require.NoError(t, err, "a fence with nothing to stop is not an error")
+	assert.Empty(t, stopped)
+	assert.Equal(t, vm.StateRunning, d.vmMgr.Status(guest), "a guest on another volume must not be stopped")
+}
+
 // TestS4_RepeatedFenceNoticeIsHarmless covers the cost of the retry: the notice
 // can arrive more than once, and a second copy must leave the fenced guest as the
 // first did rather than re-driving it or failing.
 func TestS4_RepeatedFenceNoticeIsHarmless(t *testing.T) {
 	d, guest, nc := fenceDaemon(t, "vol-fencedtwice")
 
-	payload, err := json.Marshal(viperblocklegacyv1.VolumeFencedEvent{Volume: "vol-fencedtwice", Winner: "node-b"})
-	require.NoError(t, err)
-
+	payload := fenceNoticePayload(t, ebsprovider.FenceNotice{
+		Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-fencedtwice", NodeID: "node-a", Winner: "node-b",
+	})
 	for range 2 {
-		reply, err := nc.Request(viperblocklegacyv1.VolumeFencedSubject(d.node), payload, 2*time.Second)
-		require.NoError(t, err)
-		var ack viperblocklegacyv1.VolumeFencedAck
-		require.NoError(t, json.Unmarshal(reply.Data, &ack))
-		assert.Empty(t, ack.Error, "a repeated notice is not an error")
+		ack := requestFenceNotice(t, nc, payload)
+		assert.Nil(t, ack.Error, "a repeated notice is not an error")
 	}
 	assert.Equal(t, vm.StateError, d.vmMgr.Status(guest))
 }
 
-// TestVolumeFenced_BadEventIsAnsweredWithAnError stops a notice that can never be
-// acted on from being repeated forever.
-func TestVolumeFenced_BadEventIsAnsweredWithAnError(t *testing.T) {
-	d, _, nc := fenceDaemon(t, "vol-unused")
-
-	reply, err := nc.Request(viperblocklegacyv1.VolumeFencedSubject(d.node), []byte(`{"volume":""}`), 2*time.Second)
-	require.NoError(t, err)
-	var ack viperblocklegacyv1.VolumeFencedAck
-	require.NoError(t, json.Unmarshal(reply.Data, &ack))
-	assert.NotEmpty(t, ack.Error)
+// TestFenceNotice_UnactionableNoticeIsAnsweredWithAnError stops a notice that
+// can never be acted on from being repeated forever, and leaves the guest alone.
+func TestFenceNotice_UnactionableNoticeIsAnsweredWithAnError(t *testing.T) {
+	cases := map[string]struct {
+		raw  []byte
+		code ebsprovider.ErrorCode
+	}{
+		"undecodable":     {[]byte(`{`), ebsprovider.ErrorCodeInvalidArgument},
+		"no volume":       {[]byte(`{"schema_version":1,"volume_id":""}`), ebsprovider.ErrorCodeInvalidArgument},
+		"unknown version": {[]byte(`{"schema_version":0,"volume_id":"vol-unused"}`), ebsprovider.ErrorCodeUnsupportedVersion},
+		"future version":  {fmt.Appendf(nil, `{"schema_version":%d,"volume_id":"vol-unused"}`, ebsprovider.SchemaVersion+1), ebsprovider.ErrorCodeUnsupportedVersion},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			d, guest, nc := fenceDaemon(t, "vol-unused")
+			ack := requestFenceNotice(t, nc, tc.raw)
+			require.NotNil(t, ack.Error)
+			assert.Equal(t, tc.code, ack.Error.Code)
+			if tc.code == ebsprovider.ErrorCodeUnsupportedVersion {
+				assert.Contains(t, ack.Error.Message, "unsupported schema version ")
+				assert.Contains(t, ack.Error.Message, ebsprovider.SupportedVersionsText())
+			}
+			assert.Equal(t, vm.StateRunning, d.vmMgr.Status(guest))
+		})
+	}
 }

spinifex/runtime/daemon/daemon_handlers_instance.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"log/slog"
-	"maps"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -315,7 +314,7 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) string {
 				// checkpoint and lose what this node has not sealed.
 				slog.Warn("ec2.start: owner at capacity holds unsealed writes; not starting elsewhere",
 					"instanceId", peek.InstanceID, "lastNode", lastNode)
-				if relayErr := relayStartReply(msg, resp); relayErr != nil {
+				if relayErr := msg.Respond(resp.Data); relayErr != nil {
 					slog.Error("ec2.start: failed to relay response from original node",
 						"instanceId", peek.InstanceID, "lastNode", lastNode, "err", relayErr)
 					return outcomeError
@@ -324,7 +323,7 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) string {
 			}
 			if !isCapacity {
 				// Relay success or any non-capacity error as-is.
-				if relayErr := relayStartReply(msg, resp); relayErr != nil {
+				if relayErr := msg.Respond(resp.Data); relayErr != nil {
 					slog.Error("ec2.start: failed to relay response from original node",
 						"instanceId", peek.InstanceID, "lastNode", lastNode, "err", relayErr)
 					return outcomeError
@@ -351,12 +350,3 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) string {
 
 	return handleNATSRequest(d.node, d.instanceService.StartStoppedInstance)(msg)
 }
-
-// relayStartReply answers msg with the owner's reply, body and every header,
-// since the headers carry what the AWS body cannot, such as why it refused.
-func relayStartReply(msg, resp *nats.Msg) error {
-	reply := nats.NewMsg("")
-	reply.Data = resp.Data
-	maps.Copy(reply.Header, resp.Header)
-	return msg.RespondMsg(reply)
-}

spinifex/runtime/daemon/daemon_handlers_instance_test.go

@@ -1,20 +1,15 @@
 package daemon
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
-	"strings"
 	"testing"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/mulgadc/spinifex/internal/testkit"
-	ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
 	ec2image "github.com/mulgadc/spinifex/spinifex/domains/ec2/image"
 	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
 	ec2key "github.com/mulgadc/spinifex/spinifex/domains/ec2/key"
@@ -591,17 +586,10 @@ func TestHandleEC2StartStoppedInstance_Forwarding(t *testing.T) {
 
 	t.Run("owner success is relayed", func(t *testing.T) {
 		d, _, body := newFixture(t, "i-p1-fwd-ok", "node-p1-ok")
-		owner(t, d, "node-p1-ok", func(m *nats.Msg) {
-			resp := nats.NewMsg(m.Reply)
-			resp.Data = []byte(`{"started":true}`)
-			resp.Header.Set(natsmsg.NodeIDHeader, "node-p1-ok")
-			_ = m.RespondMsg(resp)
-		})
+		owner(t, d, "node-p1-ok", func(m *nats.Msg) { _ = m.Respond([]byte(`{"started":true}`)) })
 
 		reply := requestHandler(t, d.natsConn, "ec2.start.p1-ok", asMsgHandler(d.handleEC2StartStoppedInstance), testAccountID, body)
 		assert.JSONEq(t, `{"started":true}`, string(reply.Data))
-		assert.Equal(t, "node-p1-ok", reply.Header.Get(natsmsg.NodeIDHeader),
-			"the relayed verdict is the owner's, headers included")
 	})
 
 	t.Run("owner at capacity falls back locally", func(t *testing.T) {
@@ -628,15 +616,12 @@ func TestHandleEC2StartStoppedInstance_Forwarding(t *testing.T) {
 			resp := nats.NewMsg(m.Reply)
 			resp.Data = awserrors.GenerateErrorPayload(awserrors.ErrorInsufficientInstanceCapacity)
 			resp.Header.Set(natsmsg.UnsealedWritesHeader, "true")
-			resp.Header.Set(natsmsg.NodeIDHeader, "node-p1-dirty")
 			_ = m.RespondMsg(resp)
 		})
 
 		reply := requestHandler(t, d.natsConn, "ec2.start.p1-dirty", asMsgHandler(d.handleEC2StartStoppedInstance), testAccountID, body)
 		assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, decodeError(t, reply.Data)["Code"],
 			"an owner that still holds unsealed writes must not be bypassed by a local start")
-		assert.Equal(t, "true", reply.Header.Get(natsmsg.UnsealedWritesHeader))
-		assert.Equal(t, "node-p1-dirty", reply.Header.Get(natsmsg.NodeIDHeader))
 	})
 
 	t.Run("owner not subscribed falls back locally", func(t *testing.T) {
@@ -657,71 +642,6 @@ func TestHandleEC2StartStoppedInstance_Forwarding(t *testing.T) {
 	})
 }
 
-// A start refused because another node holds a volume crosses the owner's
-// reply, the ec2.start forwarder's relay and the StartInstances client, and
-// must still match the sentinel at the end, whatever words the provider used.
-func TestStartInstances_VolumeHeldElsewhereSurvivesTheForward(t *testing.T) {
-	ns, nc := testutil.StartTestNATS(t)
-	store := vmmock.New()
-	v := stoppedVMFixture("i-0abc0000000000001", testAccountID)
-	v.LastNode = "node-owner"
-	store.Stopped[v.ID] = v
-	d := createTestDaemon(t, ns.ClientURL())
-	d.stateStore = store
-	d.instanceService = ec2instance.NewInstanceServiceImpl(
-		d.config, d.resourceMgr.instanceTypes, d.natsConn,
-		objectstore.NewMemoryObjectStore(), d.vmMgr, d.resourceMgr, store,
-	)
-
-	held := fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere,
-		awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", "volume vol-1 is published to node-a"))
-	owner := func(context.Context, *ec2instance.StartStoppedInstanceInput, string) (*ec2instance.StartStoppedInstanceOutput, error) {
-		return nil, held
-	}
-	for subject, fn := range map[string]nats.MsgHandler{
-		"ec2.start.node-owner": asMsgHandler(handleNATSRequest("node-owner", owner)),
-		"ec2.start":            asMsgHandler(d.handleEC2StartStoppedInstance),
-	} {
-		sub, err := nc.Subscribe(subject, fn)
-		require.NoError(t, err)
-		t.Cleanup(func() { _ = sub.Unsubscribe() })
-	}
-
-	_, err := ec2instanceapi.StartInstances(t.Context(),
-		&ec2.StartInstancesInput{InstanceIds: []*string{aws.String(v.ID)}}, nc, testAccountID)
-	require.Error(t, err)
-	assert.ErrorIs(t, err, ec2v1.ErrVolumeHeldElsewhere, "the caller cannot read this refusal from its words")
-	code, message, ok := awserrors.ResolveErrorDetail(err)
-	require.True(t, ok)
-	assert.Equal(t, awserrors.ErrorIncorrectState, code)
-	assert.Contains(t, message, "node-a", "the operator still learns where the volume is")
-}
-
-// The header is the one signal that a refusal was a volume held elsewhere, so
-// the owner must not put it on any other start failure in the same AWS code.
-func TestStartOwner_HeldElsewhereHeaderOnlyOnAHeldVolume(t *testing.T) {
-	_, nc := testutil.StartTestNATS(t)
-	for name, tc := range map[string]struct {
-		err  error
-		want string
-	}{
-		"held elsewhere": {fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere,
-			awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", "volume vol-1 is published to node-a")), "true"},
-		"other incorrect state": {awserrors.Errorf(awserrors.ErrorIncorrectState, "%s",
-			"instance i-0abc0000000000003 is not stopped"), ""},
-		"internal fault": {errors.New(awserrors.ErrorServerInternal), ""},
-	} {
-		t.Run(name, func(t *testing.T) {
-			owner := func(context.Context, *ec2instance.StartStoppedInstanceInput, string) (*ec2instance.StartStoppedInstanceOutput, error) {
-				return nil, tc.err
-			}
-			reply := requestHandler(t, nc, "ec2.start.owner-"+strings.ReplaceAll(name, " ", "-"),
-				asMsgHandler(handleNATSRequest("node-owner", owner)), testAccountID, []byte(`{}`))
-			assert.Equal(t, tc.want, reply.Header.Get(natsmsg.VolumeHeldElsewhereHeader))
-		})
-	}
-}
-
 // --- handleSetInstanceTags / handleSetInstanceMonitoring ---
 
 func setTagsCommand(id string) ec2v1.EC2InstanceCommand {

spinifex/runtime/daemon/instance_recovery.go

@@ -10,7 +10,7 @@ import (
 	"slices"
 	"time"
 
-	"github.com/mulgadc/spinifex/contracts/ec2/v1"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/cache"
@@ -301,7 +301,7 @@ func (f recoveryFault) stateReasonCode() string {
 
 // classifyRecoveryFault reads what the launch refused with.
 func classifyRecoveryFault(err error) recoveryFault {
-	if errors.Is(err, ec2v1.ErrVolumeHeldElsewhere) {
+	if errors.Is(err, ec2instance.ErrVolumeHeldElsewhere) {
 		return recoveryFaultLeaseHeld
 	}
 	if code, ok := awserrors.ResolveErrorCode(err); ok && code == awserrors.ErrorInsufficientInstanceCapacity {

spinifex/runtime/daemon/instance_recovery_test.go

@@ -17,7 +17,6 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -404,7 +403,7 @@ func TestWhatTheLaunchRefusalMeansForTheNextAttempt(t *testing.T) {
 		{
 			name: "the old owner still holds the volume lease",
 			err: fmt.Errorf("mount refused: %w", fmt.Errorf("%w: %w",
-				ec2v1.ErrVolumeHeldElsewhere,
+				ec2instance.ErrVolumeHeldElsewhere,
 				awserrors.Errorf(awserrors.ErrorIncorrectState, "volume is leased by another owner: node-1"))),
 			want: recoveryFaultLeaseHeld,
 			code: "Server.HostRecoveryFailed",

spinifex/runtime/daemon/jetstream.go

@@ -11,6 +11,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/instance/uefistore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
@@ -88,6 +89,7 @@ type JetStreamManager struct {
 	termRecords   *kvstore.Store[vm.InstanceRecord] // i.<id>, terminated bucket
 	clusterKV     jetstream.KeyValue                // spinifex-cluster-state
 	clusterConfig *kvstore.Bucket                   // spinifex-cluster-config
+	uefiStores    *uefistore.Store                  // spinifex-uefi-stores
 	obs           KVSyncObserver
 	running       runningSetState
 }
@@ -232,6 +234,18 @@ func (m *JetStreamManager) InitClusterConfigBucket() error {
 	return err
 }
 
+// InitUEFIStoreBucket opens the UEFI store-record bucket, creating it if it
+// doesn't exist, so a bucket that cannot be created fails startup.
+func (m *JetStreamManager) InitUEFIStoreBucket() error {
+	m.uefiStores = uefistore.NewStore(m.js)
+	return m.uefiStores.Open(context.Background())
+}
+
+// UEFIStores returns the store-record bucket opened by InitUEFIStoreBucket.
+func (m *JetStreamManager) UEFIStores() *uefistore.Store {
+	return m.uefiStores
+}
+
 // InitTerminatedInstanceBucket initializes the terminated-instances KV bucket with a 1-hour TTL.
 // JetStream automatically purges keys after 1 hour, matching AWS behavior for terminated instances.
 func (m *JetStreamManager) InitTerminatedInstanceBucket() error {

spinifex/runtime/daemon/uefi_store_bucket_test.go

@@ -0,0 +1,36 @@
+package daemon_test
+
+import (
+	"testing"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/instance/uefistore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/runtime/daemon"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// The daemon creates the store-record bucket at startup and hands out the
+// store it opened, so later readers share one handle.
+func TestInitUEFIStoreBucket_CreatesBucketAndServesTheStore(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+	m, err := daemon.NewJetStreamManager(nc)
+	require.NoError(t, err)
+
+	require.NoError(t, m.InitUEFIStoreBucket())
+
+	_, err = js.KeyValue(t.Context(), uefistore.BucketName)
+	require.NoError(t, err, "init must create the bucket rather than leave it to first use")
+
+	rec := &uefistore.UEFIStoreRecord{InstanceID: "i-0123456789abcdef0", State: uefistore.StateLive}
+	rev, err := m.UEFIStores().Create(t.Context(), rec)
+	require.NoError(t, err)
+	got, gotRev, err := m.UEFIStores().Get(t.Context(), rec.InstanceID)
+	require.NoError(t, err)
+	assert.Equal(t, rev, gotRev)
+	assert.Equal(t, rec, got)
+
+	_, err = m.UEFIStores().Create(t.Context(), rec)
+	require.ErrorIs(t, err, kvstore.ErrExists)
+}