feat(ebs): record an unpublish obligation on force detach and terminate

A force detach, or a terminate whose unmount did not complete, clears a
volume's attachment while the node that served it may still export it.
The request now records an UnpublishObligation naming that publish in the
new spinifex-ebs-unpublish-obligations bucket before the clear, and
returns without waiting on the node.

The node-local unpublishObligationReaper settles its own node's
obligations under the mounter's per-volume lock, through the existing
UnpublishVolume client: it seals a surviving volume, drops a deleting or
absent one unsealed, and closes without a provider call when a newer
attachment or a newer publish on the node owns the export. A cluster-wide
pass closes the obligations of nodes removed from the cluster, and
spx admin volumes obligations lists the open ones.

The force-complete terminate and a stopped-instance terminate after a
failed seal record the same obligation; a terminate whose unmount
completed records none. ForceDetachVolume now answers with the attachment
in state detaching, as AWS answers a detach.

spinifex/domains/ec2/ebs/obligation/obligation.go

@@ -0,0 +1,185 @@
+// Package ebsobligation records the volume exports a detach left in place.
+//
+// A force detach, or a terminate whose unmount did not complete, clears a
+// volume's attachment while the node that served it may still export it. The
+// request cannot reach that node, so it records an UnpublishObligation here and
+// returns; the named node settles it once it answers.
+package ebsobligation
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"regexp"
+	"strconv"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Bucket holds one obligation per publish a cleared attachment left behind.
+// It has no TTL: an obligation ends only when it is settled or closed.
+const Bucket = "spinifex-ebs-unpublish-obligations"
+
+// BucketVersion is the bucket schema version RunKV stamps.
+const BucketVersion = 1
+
+// SchemaVersion is the version of the UnpublishObligation record.
+const SchemaVersion uint16 = 1
+
+// UnpublishObligation names a publish a cleared attachment left on a node: the
+// node, volume and handle, and the attachment it served. The attachment is what
+// tells it apart from a later publish of the same volume on the same node.
+type UnpublishObligation struct {
+	SchemaVersion uint16    `json:"schema_version"`
+	VolumeID      string    `json:"volume_id"`
+	AccountID     string    `json:"account_id"`
+	Handle        string    `json:"handle,omitempty"` // ProviderHandle when the attachment was cleared
+	NodeID        string    `json:"node_id"`          // the node holding the publish
+	InstanceID    string    `json:"instance_id"`      // the attachment that was cleared
+	AttachedAt    time.Time `json:"attached_at"`
+	RecordedAt    time.Time `json:"recorded_at"`
+	Attempts      int       `json:"attempts"`
+	Error         string    `json:"error,omitempty"`
+}
+
+// keyToken is one NATS subject token a key segment may be: the key is three
+// tokens, so a segment holding a dot would split it and escape the filters.
+var keyToken = regexp.MustCompile(`^[-_=a-zA-Z0-9]+$`)
+
+// Key is the obligation's key, <node>.<volume-id>.<attached-at>, with the
+// attachment time in Unix nanoseconds and an unknown time as 0.
+func Key(o UnpublishObligation) (string, error) {
+	if !keyToken.MatchString(o.NodeID) {
+		return "", fmt.Errorf("unpublish obligation: node %q is not a valid key token", o.NodeID)
+	}
+	if !keyToken.MatchString(o.VolumeID) {
+		return "", fmt.Errorf("unpublish obligation: volume %q is not a valid key token", o.VolumeID)
+	}
+	var nanos int64
+	if !o.AttachedAt.IsZero() {
+		nanos = o.AttachedAt.UnixNano()
+	}
+	return o.NodeID + "." + o.VolumeID + "." + strconv.FormatInt(nanos, 10), nil
+}
+
+// allKeys matches every obligation key and not the bucket's version stamp,
+// which is a single token.
+const allKeys = "*.*.*"
+
+func bucketConfig() kvstore.Config {
+	return kvstore.Config{
+		Name:        Bucket,
+		Description: "EBS exports a cleared attachment left on a node, until that node unpublishes them",
+		History:     1,
+		OnOpen: func(ctx context.Context, kv jetstream.KeyValue) error {
+			return migrate.DefaultRegistry.RunKV(ctx, Bucket, kv, BucketVersion)
+		},
+		Missing: "EBS unpublish obligation bucket not initialized",
+	}
+}
+
+// Store reads and writes the obligation bucket.
+type Store struct {
+	s *kvstore.Store[UnpublishObligation]
+}
+
+// NewStore returns a Store over js. The bucket is created on first use.
+func NewStore(js jetstream.JetStream) *Store {
+	return &Store{s: kvstore.New[UnpublishObligation](js, bucketConfig())}
+}
+
+// Open creates or opens the bucket now, so a node that cannot reach it fails
+// at startup rather than on its first force detach.
+func (s *Store) Open(ctx context.Context) error {
+	_, err := s.s.KV(ctx)
+	return err
+}
+
+// Record writes o if no obligation with its key exists. An existing key is the
+// same obligation written by an earlier attempt, so it counts as written.
+func (s *Store) Record(ctx context.Context, o UnpublishObligation) error {
+	key, err := Key(o)
+	if err != nil {
+		return err
+	}
+	o.SchemaVersion = SchemaVersion
+	if _, err := s.s.Create(ctx, key, &o); err != nil && !errors.Is(err, kvstore.ErrExists) {
+		return fmt.Errorf("record unpublish obligation %s: %w", key, err)
+	}
+	return nil
+}
+
+// ListNode returns the open obligations naming node.
+func (s *Store) ListNode(ctx context.Context, node string) ([]UnpublishObligation, error) {
+	if !keyToken.MatchString(node) {
+		return nil, fmt.Errorf("unpublish obligation: node %q is not a valid key token", node)
+	}
+	return s.list(ctx, node+".*.*")
+}
+
+// List returns every open obligation.
+func (s *Store) List(ctx context.Context) ([]UnpublishObligation, error) {
+	return s.list(ctx, allKeys)
+}
+
+func (s *Store) list(ctx context.Context, filter string) ([]UnpublishObligation, error) {
+	items, _, err := s.s.Snapshot(ctx, filter)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]UnpublishObligation, 0, len(items))
+	for _, item := range items {
+		out = append(out, item.Value)
+	}
+	return out, nil
+}
+
+// Close removes o. Removing one already gone is not an error.
+func (s *Store) Close(ctx context.Context, o UnpublishObligation) error {
+	key, err := Key(o)
+	if err != nil {
+		return err
+	}
+	return s.s.Delete(ctx, key)
+}
+
+// RecordFailure counts a failed settlement on o and keeps its error, so an
+// operator listing the obligations sees why it is still open.
+func (s *Store) RecordFailure(ctx context.Context, o UnpublishObligation, cause error) error {
+	key, err := Key(o)
+	if err != nil {
+		return err
+	}
+	err = s.s.Mutate(ctx, key, func(cur *UnpublishObligation) (bool, error) {
+		cur.Attempts++
+		cur.Error = cause.Error()
+		return true, nil
+	})
+	if errors.Is(err, kvstore.ErrNotFound) {
+		return nil
+	}
+	return err
+}
+
+// ListOpen reads the open obligations for an operator tool without creating
+// the bucket: a cluster that has never force-detached a volume has none, and
+// that is not an error.
+func ListOpen(ctx context.Context, nc *nats.Conn) ([]UnpublishObligation, error) {
+	js, err := jetstream.New(nc)
+	if err != nil {
+		return nil, fmt.Errorf("jetstream: %w", err)
+	}
+	if _, err := js.KeyValue(ctx, Bucket); errors.Is(err, jetstream.ErrBucketNotFound) {
+		return nil, nil
+	} else if err != nil {
+		return nil, fmt.Errorf("unpublish obligation bucket: %w", err)
+	}
+	cfg := bucketConfig()
+	cfg.AttachOnly = true
+	cfg.OnOpen = nil
+	return (&Store{s: kvstore.New[UnpublishObligation](js, cfg)}).List(ctx)
+}

spinifex/domains/ec2/ebs/obligation/obligation_test.go

@@ -0,0 +1,141 @@
+package ebsobligation_test
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	ebsobligation "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/obligation"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newStore(t *testing.T) *ebsobligation.Store {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	store := ebsobligation.NewStore(js)
+	require.NoError(t, store.Open(t.Context()))
+	return store
+}
+
+func obligation(node, volumeID string, attachedAt time.Time) ebsobligation.UnpublishObligation {
+	return ebsobligation.UnpublishObligation{
+		VolumeID: volumeID, AccountID: "123456789012", NodeID: node,
+		InstanceID: "i-1", AttachedAt: attachedAt, RecordedAt: attachedAt.Add(time.Minute),
+	}
+}
+
+func TestKey_NamesNodeVolumeAndAttachment(t *testing.T) {
+	at := time.Unix(1_700_000_000, 5)
+	key, err := ebsobligation.Key(obligation("node-a", "vol-1", at))
+	require.NoError(t, err)
+	assert.Equal(t, "node-a.vol-1.1700000000000000005", key)
+
+	key, err = ebsobligation.Key(obligation("node-a", "vol-1", time.Time{}))
+	require.NoError(t, err)
+	assert.Equal(t, "node-a.vol-1.0", key, "an unknown attachment time is 0, not a huge negative")
+}
+
+// A segment holding a subject separator or wildcard would split the key or
+// let one node's listing match another's.
+func TestKey_RejectsSegmentsThatAreNotOneToken(t *testing.T) {
+	for _, o := range []ebsobligation.UnpublishObligation{
+		obligation("node.a", "vol-1", time.Time{}),
+		obligation("node-a", "vol.1", time.Time{}),
+		obligation("*", "vol-1", time.Time{}),
+		obligation("node-a", ">", time.Time{}),
+		obligation("", "vol-1", time.Time{}),
+	} {
+		_, err := ebsobligation.Key(o)
+		assert.Error(t, err, "%s/%s", o.NodeID, o.VolumeID)
+	}
+}
+
+func TestStore_RecordIsIdempotentPerAttachment(t *testing.T) {
+	store := newStore(t)
+	at := time.Unix(1_700_000_000, 0).UTC()
+	first := obligation("node-a", "vol-1", at)
+	first.Handle = "h1"
+	require.NoError(t, store.Record(t.Context(), first))
+	retry := first
+	retry.RecordedAt = at.Add(time.Hour)
+	require.NoError(t, store.Record(t.Context(), retry), "a retry of the same obligation is not an error")
+	require.NoError(t, store.Record(t.Context(), obligation("node-a", "vol-1", at.Add(time.Second))),
+		"a different attachment of the same volume is a different obligation")
+
+	open, err := store.List(t.Context())
+	require.NoError(t, err)
+	require.Len(t, open, 2)
+	for _, o := range open {
+		assert.Equal(t, ebsobligation.SchemaVersion, o.SchemaVersion)
+		if o.AttachedAt.Equal(at) {
+			assert.True(t, o.RecordedAt.Equal(at.Add(time.Minute)), "the first write is kept")
+		}
+	}
+}
+
+// The bucket's version stamp is a key in the same bucket; listing must not
+// return it as an obligation.
+func TestStore_ListSkipsTheVersionStampAndListNodeIsolatesNodes(t *testing.T) {
+	store := newStore(t)
+	at := time.Unix(1_700_000_000, 0).UTC()
+	require.NoError(t, store.Record(t.Context(), obligation("node-a", "vol-1", at)))
+	require.NoError(t, store.Record(t.Context(), obligation("node-a", "vol-2", at)))
+	require.NoError(t, store.Record(t.Context(), obligation("node-ab", "vol-3", at)))
+
+	all, err := store.List(t.Context())
+	require.NoError(t, err)
+	assert.Len(t, all, 3)
+
+	mine, err := store.ListNode(t.Context(), "node-a")
+	require.NoError(t, err)
+	volumes := []string{}
+	for _, o := range mine {
+		volumes = append(volumes, o.VolumeID)
+	}
+	assert.ElementsMatch(t, []string{"vol-1", "vol-2"}, volumes, "a node whose name prefixes another's sees only its own")
+
+	_, err = store.ListNode(t.Context(), "*")
+	assert.Error(t, err)
+}
+
+func TestStore_CloseAndRecordFailure(t *testing.T) {
+	store := newStore(t)
+	o := obligation("node-a", "vol-1", time.Unix(1_700_000_000, 0).UTC())
+	require.NoError(t, store.Record(t.Context(), o))
+
+	require.NoError(t, store.RecordFailure(t.Context(), o, errors.New("first")))
+	require.NoError(t, store.RecordFailure(t.Context(), o, errors.New("second")))
+	open, err := store.List(t.Context())
+	require.NoError(t, err)
+	require.Len(t, open, 1)
+	assert.Equal(t, 2, open[0].Attempts)
+	assert.Equal(t, "second", open[0].Error)
+
+	require.NoError(t, store.Close(t.Context(), o))
+	require.NoError(t, store.Close(t.Context(), o), "closing one already gone is not an error")
+	require.NoError(t, store.RecordFailure(t.Context(), o, errors.New("late")),
+		"a failure recorded after another sweep closed it does not resurrect it")
+	open, err = store.List(t.Context())
+	require.NoError(t, err)
+	assert.Empty(t, open)
+}
+
+func TestListOpen_MissingBucketIsEmptyAndNotCreated(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+
+	open, err := ebsobligation.ListOpen(t.Context(), nc)
+	require.NoError(t, err)
+	assert.Empty(t, open)
+	_, err = js.KeyValue(t.Context(), ebsobligation.Bucket)
+	assert.Error(t, err, "an operator listing must not create the bucket")
+
+	store := ebsobligation.NewStore(js)
+	require.NoError(t, store.Open(t.Context()))
+	require.NoError(t, store.Record(t.Context(), obligation("node-a", "vol-1", time.Unix(1_700_000_000, 0).UTC())))
+	open, err = ebsobligation.ListOpen(t.Context(), nc)
+	require.NoError(t, err)
+	require.Len(t, open, 1)
+	assert.Equal(t, "vol-1", open[0].VolumeID)
+}

spinifex/domains/ec2/instance/service.go

@@ -129,6 +129,17 @@ type InstanceTagWriter interface {
 	DeleteAllTags(ctx context.Context, accountID, resourceID string) error
 }
 
+// OwedUnpublish names the publish a terminate leaves in place when its unmount
+// did not complete: the node still exporting the volume and the instance it
+// served. The zero value means the unmount completed and nothing is owed.
+type OwedUnpublish struct {
+	NodeID     string
+	InstanceID string
+}
+
+// Owed reports whether a publish is left in place.
+func (o OwedUnpublish) Owed() bool { return o.NodeID != "" }
+
 // VolumeDeleter deletes EBS volumes. Implemented by domains/ec2/volume's
 // VolumeServiceImpl (used by the daemon).
 type VolumeDeleter interface {
@@ -137,14 +148,16 @@ type VolumeDeleter interface {
 	// DeleteVolumeOnTerminate deletes a DeleteOnTermination volume as part of
 	// an instance terminate: it clears any stale attachment (terminate
 	// implies detach) before deleting, so a still-attached boot volume left
-	// over from Stop is not rejected by DeleteVolume's in-use guard.
-	DeleteVolumeOnTerminate(ctx context.Context, volumeID, accountID string) error
+	// over from Stop is not rejected by DeleteVolume's in-use guard. When owed
+	// names a publish, an unpublish obligation is recorded before the clear.
+	DeleteVolumeOnTerminate(ctx context.Context, volumeID, accountID string, owed OwedUnpublish) error
 
 	// DetachVolumeOnTerminate clears a still-attached volume's attachment on
 	// terminate without deleting it, matching AWS semantics for a
 	// DeleteOnTermination=false volume: terminate still implies detach, it
 	// just leaves the volume behind as available rather than deleting it.
-	DetachVolumeOnTerminate(ctx context.Context, volumeID, accountID string) error
+	// When owed names a publish, an unpublish obligation is recorded first.
+	DetachVolumeOnTerminate(ctx context.Context, volumeID, accountID string, owed OwedUnpublish) error
 }
 
 // VolumeCreator creates the volumes a launch names beyond the root device, and

spinifex/domains/ec2/instance/service_impl.go

@@ -2966,6 +2966,15 @@ func (s *InstanceServiceImpl) deleteInstanceVolumes(ctx context.Context, instanc
 	// stopped-path delete failure.
 	var lastErr error
 
+	// A volume whose seal failed at stop is still exported on the node that
+	// stopped it, so clearing its attachment owes that node an unpublish.
+	owedFor := func(volumeID string) OwedUnpublish {
+		if instance.LastNode == "" || !slices.Contains(instance.UnpublishOwed, volumeID) {
+			return OwedUnpublish{}
+		}
+		return OwedUnpublish{NodeID: instance.LastNode, InstanceID: instance.ID}
+	}
+
 	for _, ebsRequest := range instance.EBSRequests.Requests {
 		// Internal volumes (EFI) are always cleaned up via ebs.delete.
 		if ebsRequest.EFI {
@@ -2996,7 +3005,7 @@ func (s *InstanceServiceImpl) deleteInstanceVolumes(ctx context.Context, instanc
 				continue
 			}
 			// A volume already gone is the detach's own goal, not a failure.
-			if err := s.volumeDeleter.DetachVolumeOnTerminate(ctx, ebsRequest.Name, instance.AccountID); err != nil &&
+			if err := s.volumeDeleter.DetachVolumeOnTerminate(ctx, ebsRequest.Name, instance.AccountID, owedFor(ebsRequest.Name)); err != nil &&
 				!awserrors.IsNotFound(err) && !objectstore.IsNoSuchKeyError(err) {
 				slog.ErrorContext(ctx, "TerminateStoppedInstance: failed to detach volume", "name", ebsRequest.Name, "err", err)
 				lastErr = err
@@ -3015,7 +3024,7 @@ func (s *InstanceServiceImpl) deleteInstanceVolumes(ctx context.Context, instanc
 		// clears a Boot volume, runtime/daemon/vm_adapters.go) before deleting, so
 		// DeleteVolume's in-use guard does not reject an otherwise-legitimate
 		// delete. The resulting error is surfaced into lastErr, not swallowed.
-		if err := s.volumeDeleter.DeleteVolumeOnTerminate(ctx, ebsRequest.Name, instance.AccountID); err != nil {
+		if err := s.volumeDeleter.DeleteVolumeOnTerminate(ctx, ebsRequest.Name, instance.AccountID, owedFor(ebsRequest.Name)); err != nil {
 			slog.ErrorContext(ctx, "TerminateStoppedInstance: failed to delete volume", "name", ebsRequest.Name, "err", err)
 			lastErr = err
 		}

spinifex/domains/ec2/instance/service_impl_test.go

@@ -1734,6 +1734,7 @@ type fakeVolumeDeleter struct {
 	calls    []string
 	deleted  []string
 	detached []string
+	owed     []OwedUnpublish
 	err      error
 }
 
@@ -1749,7 +1750,8 @@ func (f *fakeVolumeDeleter) DeleteVolume(_ context.Context, input *ec2.DeleteVol
 
 // DeleteVolumeOnTerminate mirrors DeleteVolume's call/error bookkeeping; the
 // stopped-instance terminate path calls this instead of DeleteVolume.
-func (f *fakeVolumeDeleter) DeleteVolumeOnTerminate(_ context.Context, volumeID, _ string) error {
+func (f *fakeVolumeDeleter) DeleteVolumeOnTerminate(_ context.Context, volumeID, _ string, owed OwedUnpublish) error {
+	f.owed = append(f.owed, owed)
 	f.calls = append(f.calls, volumeID)
 	if f.err != nil {
 		return f.err
@@ -1761,7 +1763,8 @@ func (f *fakeVolumeDeleter) DeleteVolumeOnTerminate(_ context.Context, volumeID,
 // DetachVolumeOnTerminate mirrors DeleteVolumeOnTerminate's call/error
 // bookkeeping but records into detached instead of deleted; the non-DoT
 // Boot-volume branch calls this instead of deleting.
-func (f *fakeVolumeDeleter) DetachVolumeOnTerminate(_ context.Context, volumeID, _ string) error {
+func (f *fakeVolumeDeleter) DetachVolumeOnTerminate(_ context.Context, volumeID, _ string, owed OwedUnpublish) error {
+	f.owed = append(f.owed, owed)
 	f.calls = append(f.calls, volumeID)
 	if f.err != nil {
 		return f.err
@@ -2009,6 +2012,45 @@ func TestTerminateStoppedInstance_UserVolumeDeleted(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, []string{"vol-user-001"}, vd.deleted, "only DeleteOnTermination=true volumes deleted")
 	assert.Equal(t, []string{"vol-keep-001"}, vd.detached, "terminate implies detach even for a volume that survives it")
+	assert.Equal(t, []OwedUnpublish{{}, {}}, vd.owed, "a stop whose unmount completed owes no unpublish")
+}
+
+// A stop whose seal failed left those volumes exported on the node that
+// stopped the instance, so terminate names that node for each of them alone.
+func TestTerminateStoppedInstance_FailedSealOwesTheStoppingNodeAnUnpublish(t *testing.T) {
+	id := "i-seal-failed"
+	v := &vm.VM{ID: id, Status: vm.StateStopped, AccountID: "acc", LastNode: "node-a",
+		UnpublishOwed: []string{"vol-root-001", "vol-keep-001"}}
+	v.EBSRequests.Requests = []viperblocklegacyv1.EBSRequest{
+		{Name: "vol-root-001", Boot: true, DeleteOnTermination: true},
+		{Name: "vol-keep-001", DeleteOnTermination: false},
+		{Name: "vol-sealed-001", DeleteOnTermination: false},
+	}
+	store := &vmmock.StateStore{Stopped: map[string]*vm.VM{id: v}}
+	vd := &fakeVolumeDeleter{}
+	svc := &InstanceServiceImpl{stoppedStore: store, volumeDeleter: vd}
+
+	_, err := svc.TerminateStoppedInstance(context.Background(), &TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err)
+
+	owed := OwedUnpublish{NodeID: "node-a", InstanceID: id}
+	assert.Equal(t, []string{"vol-root-001", "vol-keep-001", "vol-sealed-001"}, vd.calls)
+	assert.Equal(t, []OwedUnpublish{owed, owed, {}}, vd.owed)
+}
+
+// With no node recorded there is no node to name; the terminate owes nothing
+// rather than naming an empty node.
+func TestTerminateStoppedInstance_NoLastNodeOwesNothing(t *testing.T) {
+	id := "i-seal-failed-nonode"
+	v := &vm.VM{ID: id, Status: vm.StateStopped, AccountID: "acc", UnpublishOwed: []string{"vol-keep-001"}}
+	v.EBSRequests.Requests = []viperblocklegacyv1.EBSRequest{{Name: "vol-keep-001"}}
+	store := &vmmock.StateStore{Stopped: map[string]*vm.VM{id: v}}
+	vd := &fakeVolumeDeleter{}
+	svc := &InstanceServiceImpl{stoppedStore: store, volumeDeleter: vd}
+
+	_, err := svc.TerminateStoppedInstance(context.Background(), &TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err)
+	assert.Equal(t, []OwedUnpublish{{}}, vd.owed)
 }
 
 // TestTerminateStoppedInstance_StampsTeardownVolumesDone locks the

spinifex/domains/ec2/volume/account_termination_current_behaviour_test.go

@@ -9,6 +9,7 @@ import (
 	testutil "github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	ebsmetadata "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
 	ec2snapshot "github.com/mulgadc/spinifex/spinifex/domains/ec2/snapshot"
 	ec2volume "github.com/mulgadc/spinifex/spinifex/domains/ec2/volume"
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -59,7 +60,7 @@ func TestCurrentBehaviour_AccountTermination_SnapshottedDeleteOnTerminationVolum
 	require.NoError(t, err)
 	require.NoError(t, volumes.UpdateVolumeState(accountID, volumeID, "in-use", instanceID, "/dev/sda1"))
 
-	err = volumes.DeleteVolumeOnTerminate(ctx, volumeID, accountID)
+	err = volumes.DeleteVolumeOnTerminate(ctx, volumeID, accountID, ec2instance.OwedUnpublish{})
 	assert.EqualError(t, err, awserrors.ErrorVolumeInUse, "the snapshot index refuses the terminate delete")
 
 	survivor, err := metadata.GetVolume(ctx, accountID, volumeID)

spinifex/domains/ec2/volume/delete_volume_unreadable_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	ebsprovider "github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
@@ -194,7 +195,7 @@ func TestDeleteVolumeOnTerminate_DeleteReadFailureIsNotSuccess(t *testing.T) {
 	svc.store = failing
 	svc.metadata = ebsmetadata.NewStore(failing, "test-bucket")
 
-	err := svc.DeleteVolumeOnTerminate(context.Background(), volumeID, testVolAccountID)
+	err := svc.DeleteVolumeOnTerminate(context.Background(), volumeID, testVolAccountID, ec2instance.OwedUnpublish{})
 
 	require.Error(t, err, "a delete-read failure after a successful attachment clear must not be reported as success")
 	assert.False(t, awserrors.IsNotFound(err),

spinifex/domains/ec2/volume/force_detach_test.go

@@ -20,6 +20,7 @@ func newForceDetachService(t *testing.T) *VolumeServiceImpl {
 	t.Helper()
 	svc := newTestVolumeService("ap-southeast-2a")
 	svc.SetEBSProvider(ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}))
+	wireObligations(t, svc, &fakeNodeLocator{nodes: map[string]string{"i-stuck0000000000": "node-a"}}, "node-a", "node-b")
 	return svc
 }
 
@@ -47,7 +48,11 @@ func TestForceDetachVolumeClearsTheAttachment(t *testing.T) {
 	// The previous instance is reported back: it is the only record of what
 	// the volume was torn away from.
 	assert.Equal(t, "i-stuck0000000000", aws.StringValue(attachment.InstanceId))
-	assert.Equal(t, "detached", aws.StringValue(attachment.State))
+	// The export is still up on the node that served it, so the attachment is
+	// reported mid-teardown, as AWS reports a detach it has not finished.
+	assert.Equal(t, "detaching", aws.StringValue(attachment.State))
+	assert.Equal(t, "/dev/sdf", aws.StringValue(attachment.Device))
+	assert.False(t, aws.TimeValue(attachment.AttachTime).IsZero())
 
 	meta, err := svc.GetVolumeMetadata("000000000011", "vol-forcedetach1")
 	require.NoError(t, err)
@@ -112,4 +117,8 @@ func TestForceDetachVolumeOnAnUnattachedVolume(t *testing.T) {
 	require.NoError(t, err)
 
 	assert.Empty(t, aws.StringValue(attachment.InstanceId))
+	assert.Equal(t, "detached", aws.StringValue(attachment.State))
+	open, err := svc.obligations.List(context.Background())
+	require.NoError(t, err)
+	assert.Empty(t, open, "an unattached volume has no publish to undo")
 }

spinifex/domains/ec2/volume/service_impl.go

@@ -20,6 +20,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	ebsencryption "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/encryption"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ebsobligation "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/obligation"
 	ebspolicy "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/policy"
 	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -59,6 +60,12 @@ type VolumeServiceImpl struct {
 	snapshotKV jetstream.KeyValue
 	provider   ebsprovider.EBSProvider
 	metadata   *ebsmetadata.Store
+
+	// The unpublish obligations a cleared attachment records, and what names
+	// the node holding the publish. Unset, a detach that leaves one fails.
+	obligations   *ebsobligation.Store
+	instanceNodes InstanceNodeLocator
+	members       func() []string
 }
 
 // SetEBSProvider injects the provider boundary used by the control plane. It
@@ -1082,7 +1089,7 @@ func (s *VolumeServiceImpl) ModifyVolume(ctx context.Context, input *ec2.ModifyV
 // terminating instance's QEMU has already been asked to shut down — so this
 // is always a metadata-only clear, never a QMP call. Errors from either step
 // are returned, not swallowed.
-func (s *VolumeServiceImpl) DeleteVolumeOnTerminate(ctx context.Context, volumeID, accountID string) error {
+func (s *VolumeServiceImpl) DeleteVolumeOnTerminate(ctx context.Context, volumeID, accountID string, owed ec2instance.OwedUnpublish) error {
 	// Clearing the attachment would overwrite a recorded deletion, so a volume
 	// already owed is left alone: the reclaimer owns it from here, and resetting
 	// it to available would let a second teardown run against the same data.
@@ -1090,6 +1097,9 @@ func (s *VolumeServiceImpl) DeleteVolumeOnTerminate(ctx context.Context, volumeI
 	if err == nil && meta.State == volumeStateDeleting {
 		return nil
 	}
+	if err := s.recordTerminateObligation(ctx, accountID, volumeID, owed); err != nil {
+		return err
+	}
 
 	if err := s.UpdateVolumeState(accountID, volumeID, "available", "", ""); err != nil {
 		// No metadata doc means the volume is already gone, per GetVolume's own
@@ -1109,7 +1119,10 @@ func (s *VolumeServiceImpl) DeleteVolumeOnTerminate(ctx context.Context, volumeI
 // available rather than deleting it. Metadata-only, no QMP — same rationale
 // as DeleteVolumeOnTerminate: there is no live QEMU to hot-unplug on either
 // terminate path.
-func (s *VolumeServiceImpl) DetachVolumeOnTerminate(_ context.Context, volumeID, accountID string) error {
+func (s *VolumeServiceImpl) DetachVolumeOnTerminate(ctx context.Context, volumeID, accountID string, owed ec2instance.OwedUnpublish) error {
+	if err := s.recordTerminateObligation(ctx, accountID, volumeID, owed); err != nil {
+		return err
+	}
 	if err := s.UpdateVolumeState(accountID, volumeID, "available", "", ""); err != nil {
 		// No metadata doc means the volume is already gone, per GetVolume's own
 		// contract, so absence here is success, not a teardown failure.
@@ -1128,8 +1141,12 @@ func (s *VolumeServiceImpl) DetachVolumeOnTerminate(_ context.Context, volumeID,
 // The ordinary DetachVolume routes to ec2.cmd.{instanceID} for the QMP unplug,
 // so a volume attached to an instance whose host stopped answering can never be
 // detached and therefore never deleted. This exists for that deadlock and for
-// nothing else: it leaves a live guest holding a device the control plane no
-// longer believes in, which is only safe once that guest is being destroyed.
+// nothing else.
+//
+// That host still exports the volume, and the unpublish has to go to it, so
+// it cannot be made here. The request records an unpublish obligation naming
+// the publish before it clears the attachment, and returns; the node settles
+// it once it answers. A failed record clears nothing.
 func (s *VolumeServiceImpl) ForceDetachVolume(ctx context.Context, input *ec2.DetachVolumeInput, accountID string) (*ec2.VolumeAttachment, error) {
 	if input == nil || input.VolumeId == nil || *input.VolumeId == "" {
 		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
@@ -1153,18 +1170,48 @@ func (s *VolumeServiceImpl) ForceDetachVolume(ctx context.Context, input *ec2.De
 	}
 
 	previous := meta.AttachedInstance
+	if previous != "" {
+		nodes, err := s.publishingNodes(previous)
+		if err != nil {
+			slog.ErrorContext(ctx, "ForceDetachVolume: cannot name the node holding the publish, clearing nothing",
+				"volumeId", volumeID, "instanceId", previous, "err", err)
+			return nil, errors.New(awserrors.ErrorInternalError)
+		}
+		if err := s.recordObligations(ctx, meta, nodes); err != nil {
+			slog.ErrorContext(ctx, "ForceDetachVolume: failed to record the unpublish obligation, clearing nothing",
+				"volumeId", volumeID, "instanceId", previous, "err", err)
+			return nil, errors.New(awserrors.ErrorInternalError)
+		}
+	}
+
 	if err := s.UpdateVolumeState(accountID, volumeID, "available", "", ""); err != nil {
 		slog.ErrorContext(ctx, "ForceDetachVolume: failed to clear attachment", "volumeId", volumeID, "err", err)
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
-	slog.WarnContext(ctx, "ForceDetachVolume: attachment cleared in the control plane only",
+
+	if previous == "" {
+		return &ec2.VolumeAttachment{
+			VolumeId: aws.String(volumeID),
+			State:    aws.String("detached"),
+		}, nil
+	}
+	slog.WarnContext(ctx, "ForceDetachVolume: attachment cleared, unpublish left to the node holding it",
 		"volumeId", volumeID, "previousInstance", previous, "accountId", accountID)
 
-	return &ec2.VolumeAttachment{
+	// AWS answers a detach with the attachment it is tearing down, in state
+	// detaching; the export here goes when the node settles the obligation.
+	attachment := &ec2.VolumeAttachment{
 		VolumeId:   aws.String(volumeID),
 		InstanceId: aws.String(previous),
-		State:      aws.String("detached"),
-	}, nil
+		State:      aws.String("detaching"),
+	}
+	if meta.DeviceName != "" {
+		attachment.Device = aws.String(meta.DeviceName)
+	}
+	if !meta.AttachedAt.IsZero() {
+		attachment.AttachTime = aws.Time(meta.AttachedAt)
+	}
+	return attachment, nil
 }
 
 // DeleteVolume records a durable obligation to destroy an EBS volume and

spinifex/domains/ec2/volume/service_impl_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/mulgadc/spinifex/internal/testkit/ebsfake"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
 	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
@@ -1767,7 +1768,7 @@ func TestDeleteVolumeOnTerminate_ClearsAttachmentThenDeletes(t *testing.T) {
 		AttachedInstance: "i-stopped",
 	})
 
-	err := svc.DeleteVolumeOnTerminate(context.Background(), volumeID, testVolAccountID)
+	err := svc.DeleteVolumeOnTerminate(context.Background(), volumeID, testVolAccountID, ec2instance.OwedUnpublish{})
 	require.NoError(t, err, "terminate implies detach: a stale attachment must not block the terminate delete")
 
 	deleting, err := svc.GetVolumeMetadata(testVolAccountID, volumeID)
@@ -1794,7 +1795,7 @@ func TestDeleteVolumeOnTerminate_AlreadyGoneMetadataIsSuccess(t *testing.T) {
 	svc.snapshotKV = kv
 
 	// No metadata doc is ever written for this volume ID.
-	err := svc.DeleteVolumeOnTerminate(context.Background(), "vol-already-gone", testVolAccountID)
+	err := svc.DeleteVolumeOnTerminate(context.Background(), "vol-already-gone", testVolAccountID, ec2instance.OwedUnpublish{})
 	require.NoError(t, err, "a volume with no metadata document is already gone, not a teardown failure")
 }
 
@@ -1807,7 +1808,7 @@ func TestDetachVolumeOnTerminate_AlreadyGoneMetadataIsSuccess(t *testing.T) {
 	svc := newTestVolumeServiceWithStore("ap-southeast-2a", store)
 	svc.snapshotKV = kv
 
-	err := svc.DetachVolumeOnTerminate(context.Background(), "vol-already-gone", testVolAccountID)
+	err := svc.DetachVolumeOnTerminate(context.Background(), "vol-already-gone", testVolAccountID, ec2instance.OwedUnpublish{})
 	require.NoError(t, err, "a volume with no metadata document is already gone, not a teardown failure")
 }
 
@@ -1891,7 +1892,7 @@ func TestDeleteVolume_SecondDeleteIsAcceptedWithoutASecondTeardown(t *testing.T)
 
 	// DeleteOnTermination arriving for a volume already owed must not reset it
 	// to available, which would let a second teardown through.
-	require.NoError(t, svc.DeleteVolumeOnTerminate(context.Background(), volumeID, testVolAccountID))
+	require.NoError(t, svc.DeleteVolumeOnTerminate(context.Background(), volumeID, testVolAccountID, ec2instance.OwedUnpublish{}))
 	meta, err := svc.metadata.GetVolume(context.Background(), testVolAccountID, volumeID)
 	require.NoError(t, err)
 	assert.Equal(t, "deleting", meta.State)
@@ -1939,7 +1940,7 @@ func TestDeleteVolumeOnTerminate_SurfacesDeleteFailure(t *testing.T) {
 	_, err = kv.Put(t.Context(), volumeID, snapData)
 	require.NoError(t, err)
 
-	err = svc.DeleteVolumeOnTerminate(context.Background(), volumeID, testVolAccountID)
+	err = svc.DeleteVolumeOnTerminate(context.Background(), volumeID, testVolAccountID, ec2instance.OwedUnpublish{})
 	require.Error(t, err, "a DeleteVolume failure must be surfaced, not swallowed")
 	assert.Contains(t, err.Error(), awserrors.ErrorVolumeInUse)
 

spinifex/domains/ec2/volume/unpublish_obligation.go

@@ -0,0 +1,364 @@
+package volume
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+
+	ebsmetadata "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ebsobligation "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/obligation"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+// InstanceNodeLocator reports the node that last ran an instance. found is
+// false when no record of the instance exists.
+type InstanceNodeLocator interface {
+	InstanceLastNode(instanceID string) (node string, found bool, err error)
+}
+
+// SetUnpublishObligations wires the obligation store a cleared attachment
+// records into, the locator that names the node holding the publish, and the
+// cluster members to name when the instance has no record.
+func (s *VolumeServiceImpl) SetUnpublishObligations(store *ebsobligation.Store, locator InstanceNodeLocator, members func() []string) {
+	s.obligations = store
+	s.instanceNodes = locator
+	s.members = members
+}
+
+// publishingNodes names the node holding instanceID's publish. With no record
+// of the instance there is no node to name, so every member is named and each
+// settles its own.
+func (s *VolumeServiceImpl) publishingNodes(instanceID string) ([]string, error) {
+	if s.instanceNodes == nil {
+		return nil, errors.New("no instance node locator configured")
+	}
+	node, found, err := s.instanceNodes.InstanceLastNode(instanceID)
+	if err != nil {
+		return nil, fmt.Errorf("read instance %s: %w", instanceID, err)
+	}
+	if found && node != "" {
+		return []string{node}, nil
+	}
+	var members []string
+	if s.members != nil {
+		members = s.members()
+	}
+	if len(members) == 0 {
+		return nil, fmt.Errorf("instance %s has no record and no cluster member is known", instanceID)
+	}
+	return members, nil
+}
+
+// recordObligations writes one obligation per node for the attachment meta
+// still carries. Every write must land before the caller clears it.
+func (s *VolumeServiceImpl) recordObligations(ctx context.Context, meta ebsmetadata.Volume, nodes []string) error {
+	if s.obligations == nil {
+		return errors.New("no unpublish obligation store configured")
+	}
+	now := time.Now()
+	for _, node := range nodes {
+		if err := s.obligations.Record(ctx, ebsobligation.UnpublishObligation{
+			VolumeID:   meta.VolumeID,
+			AccountID:  meta.TenantID,
+			Handle:     meta.ProviderHandle,
+			NodeID:     node,
+			InstanceID: meta.AttachedInstance,
+			AttachedAt: meta.AttachedAt,
+			RecordedAt: now,
+		}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// recordTerminateObligation records the obligation for a terminate that clears
+// owed's attachment with its publish still in place. A volume no longer
+// attached to that instance has nothing of it left to clear.
+func (s *VolumeServiceImpl) recordTerminateObligation(ctx context.Context, accountID, volumeID string, owed ec2instance.OwedUnpublish) error {
+	if !owed.Owed() {
+		return nil
+	}
+	meta, err := s.metadata.GetVolume(ctx, accountID, volumeID)
+	if objectstore.IsNoSuchKeyError(err) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("read volume %s before recording its unpublish obligation: %w", volumeID, err)
+	}
+	if meta.AttachedInstance != owed.InstanceID {
+		return nil
+	}
+	if err := s.recordObligations(ctx, meta, []string{owed.NodeID}); err != nil {
+		return fmt.Errorf("record unpublish obligation for %s: %w", volumeID, err)
+	}
+	return nil
+}
+
+// ObligationUnpublisher is this node's mounter as the obligation reaper drives
+// it. Its lock is the one every publish and unpublish on this node takes.
+type ObligationUnpublisher interface {
+	// HoldVolume takes volumeID's lock without waiting. false means a publish
+	// or unpublish of it is in flight here, and the sweep comes back later.
+	HoldVolume(volumeID string) (VolumeHold, bool)
+}
+
+// VolumeHold is one volume's publish lock on this node.
+type VolumeHold interface {
+	// PublishedSince reports whether this process published the volume after
+	// t and has not unpublished it since.
+	PublishedSince(t time.Time) bool
+	// Unpublish tears down this node's export of the volume. A volume the
+	// provider holds no export of counts as done.
+	Unpublish(ctx context.Context, accountID, handle string, noSeal bool) error
+	Release()
+}
+
+// obligationAction is what settling an obligation does with the volume as it
+// is now.
+type obligationAction int
+
+const (
+	// obligationWait leaves the obligation open: the attachment it names has
+	// not been cleared yet, or a publish here is newer than it.
+	obligationWait obligationAction = iota
+	// obligationSuperseded closes it without a provider call: a newer
+	// attachment owns the export.
+	obligationSuperseded
+	// obligationSeal unpublishes with a seal: nobody else writes the volume,
+	// so the copy here is the last owner's and its writes are kept.
+	obligationSeal
+	// obligationNoSeal unpublishes without sealing a volume being deleted.
+	obligationNoSeal
+)
+
+func (a obligationAction) String() string {
+	switch a {
+	case obligationSuperseded:
+		return "superseded"
+	case obligationSeal:
+		return "seal"
+	case obligationNoSeal:
+		return "no-seal"
+	default:
+		return "wait"
+	}
+}
+
+// decideObligation is the settlement policy over the volume's current record;
+// meta is nil when the volume has none. It never unpublishes an attachment the
+// volume still carries.
+func decideObligation(o ebsobligation.UnpublishObligation, meta *ebsmetadata.Volume) obligationAction {
+	if meta == nil || meta.State == volumeStateDeleting {
+		return obligationNoSeal
+	}
+	if meta.AttachedInstance != "" {
+		if meta.AttachedInstance != o.InstanceID || meta.AttachedAt.After(o.AttachedAt) {
+			return obligationSuperseded
+		}
+		return obligationWait
+	}
+	return obligationSeal
+}
+
+// obligationSince is the time a publish on this node must follow to be newer
+// than the one an obligation undoes.
+func obligationSince(o ebsobligation.UnpublishObligation) time.Time {
+	if o.AttachedAt.IsZero() {
+		return o.RecordedAt
+	}
+	return o.AttachedAt
+}
+
+// unpublishObligationReaper settles the obligations naming its node through
+// that node's mounter, once the node answers.
+type unpublishObligationReaper struct {
+	svc         *VolumeServiceImpl
+	node        string
+	unpublisher ObligationUnpublisher
+}
+
+var _ vm.Reaper = (*unpublishObligationReaper)(nil)
+
+// NewUnpublishObligationReaper builds the reaper that settles node's
+// obligations through unpublisher, which must be the node's own mounter.
+func (s *VolumeServiceImpl) NewUnpublishObligationReaper(node string, unpublisher ObligationUnpublisher) vm.Reaper {
+	return &unpublishObligationReaper{svc: s, node: node, unpublisher: unpublisher}
+}
+
+func (r *unpublishObligationReaper) Class() string { return "ebs-unpublish-obligation" }
+
+// Scope is node-local: an obligation is settled only by the node it names,
+// because only that node can tear its own export down.
+func (r *unpublishObligationReaper) Scope() vm.ReaperScope { return vm.ScopeNodeLocal }
+
+// Sweep settles what it can and reports how many obligations it closed. A
+// failure is recorded on the obligation and retried on the next sweep.
+func (r *unpublishObligationReaper) Sweep(ctx context.Context) (int, error) {
+	if r.svc.obligations == nil || r.unpublisher == nil {
+		return 0, nil
+	}
+	open, err := r.svc.obligations.ListNode(ctx, r.node)
+	if err != nil {
+		return 0, err
+	}
+	closed := 0
+	for _, o := range open {
+		if err := ctx.Err(); err != nil {
+			return closed, err
+		}
+		done, err := r.settle(ctx, o)
+		if err != nil {
+			slog.WarnContext(ctx, "ebs-unpublish-obligation: settlement failed, retrying next sweep",
+				"volumeId", o.VolumeID, "node", o.NodeID, "instanceId", o.InstanceID, "attempts", o.Attempts+1, "err", err)
+			if recErr := r.svc.obligations.RecordFailure(ctx, o, err); recErr != nil {
+				slog.WarnContext(ctx, "ebs-unpublish-obligation: could not record the failure",
+					"volumeId", o.VolumeID, "node", o.NodeID, "err", recErr)
+			}
+			continue
+		}
+		if done {
+			closed++
+		}
+	}
+	return closed, nil
+}
+
+// settle acts on one obligation under the volume's lock, so no publish on this
+// node falls between reading the volume and unpublishing it.
+func (r *unpublishObligationReaper) settle(ctx context.Context, o ebsobligation.UnpublishObligation) (bool, error) {
+	hold, ok := r.unpublisher.HoldVolume(o.VolumeID)
+	if !ok {
+		return false, nil
+	}
+	defer hold.Release()
+
+	if hold.PublishedSince(obligationSince(o)) {
+		return false, nil
+	}
+
+	var current *ebsmetadata.Volume
+	meta, err := r.svc.metadata.GetVolume(ctx, o.AccountID, o.VolumeID)
+	switch {
+	case err == nil:
+		current = &meta
+	case objectstore.IsNoSuchKeyError(err):
+	default:
+		return false, fmt.Errorf("read volume %s: %w", o.VolumeID, err)
+	}
+
+	action := decideObligation(o, current)
+	switch action {
+	case obligationWait:
+		return false, nil
+	case obligationSuperseded:
+		slog.WarnContext(ctx, "ebs-unpublish-obligation: a newer attachment owns the volume, closing without unpublishing",
+			"volumeId", o.VolumeID, "node", o.NodeID, "instanceId", o.InstanceID, "attachedInstance", current.AttachedInstance)
+	case obligationSeal, obligationNoSeal:
+		if err := hold.Unpublish(ctx, o.AccountID, o.Handle, action == obligationNoSeal); err != nil {
+			return false, err
+		}
+		slog.InfoContext(ctx, "ebs-unpublish-obligation: unpublished the export a cleared attachment left",
+			"volumeId", o.VolumeID, "node", o.NodeID, "instanceId", o.InstanceID, "noSeal", action == obligationNoSeal)
+	}
+	if err := r.svc.obligations.Close(ctx, o); err != nil {
+		return false, fmt.Errorf("close obligation: %w", err)
+	}
+	recordObligationClosed(o.NodeID, action.String())
+	return true, nil
+}
+
+// NodeDeparted reports whether node has left the cluster, rather than being
+// down, stale or partitioned, which must not count.
+type NodeDeparted func(ctx context.Context, node string) (bool, error)
+
+// unpublishObligationCloser closes, without a provider call, the obligations of
+// nodes removed from the cluster: such a node holds nothing a member can reach.
+type unpublishObligationCloser struct {
+	svc      *VolumeServiceImpl
+	departed NodeDeparted
+}
+
+var _ vm.Reaper = (*unpublishObligationCloser)(nil)
+
+// NewUnpublishObligationCloser builds the cluster-wide pass that closes the
+// obligations of nodes departed reports as gone.
+func (s *VolumeServiceImpl) NewUnpublishObligationCloser(departed NodeDeparted) vm.Reaper {
+	return &unpublishObligationCloser{svc: s, departed: departed}
+}
+
+func (c *unpublishObligationCloser) Class() string { return "ebs-unpublish-obligation-close" }
+
+// Scope is cluster-wide: membership is the cluster's, and one pass is enough.
+func (c *unpublishObligationCloser) Scope() vm.ReaperScope { return vm.ScopeClusterWide }
+
+func (c *unpublishObligationCloser) Sweep(ctx context.Context) (int, error) {
+	if c.svc.obligations == nil || c.departed == nil {
+		return 0, nil
+	}
+	open, err := c.svc.obligations.List(ctx)
+	if err != nil {
+		return 0, err
+	}
+	gone := map[string]bool{}
+	closed := 0
+	for _, o := range open {
+		departed, seen := gone[o.NodeID]
+		if !seen {
+			departed, err = c.departed(ctx, o.NodeID)
+			if err != nil {
+				slog.WarnContext(ctx, "ebs-unpublish-obligation: cannot tell whether a node left the cluster, keeping its obligations",
+					"node", o.NodeID, "err", err)
+				departed = false
+			}
+			gone[o.NodeID] = departed
+		}
+		if !departed {
+			continue
+		}
+		if err := c.svc.obligations.Close(ctx, o); err != nil {
+			slog.WarnContext(ctx, "ebs-unpublish-obligation: failed to close a departed node's obligation",
+				"volumeId", o.VolumeID, "node", o.NodeID, "err", err)
+			continue
+		}
+		slog.WarnContext(ctx, "ebs-unpublish-obligation: closed for a node removed from the cluster",
+			"volumeId", o.VolumeID, "node", o.NodeID, "instanceId", o.InstanceID)
+		recordObligationClosed(o.NodeID, "departed")
+		closed++
+	}
+	return closed, nil
+}
+
+const obligationMeterName = "github.com/mulgadc/spinifex/spinifex/domains/ec2/volume/obligation"
+
+var (
+	obligationClosedOnce    sync.Once
+	obligationClosedCounter metric.Int64Counter
+)
+
+// recordObligationClosed counts one closed obligation by the node it named
+// and how it closed.
+func recordObligationClosed(node, outcome string) {
+	obligationClosedOnce.Do(func() {
+		var err error
+		obligationClosedCounter, err = otel.Meter(obligationMeterName).Int64Counter("mulga.ebs.unpublish_obligation.closed",
+			metric.WithDescription("Unpublish obligations closed, by the node they named and how they closed."),
+			metric.WithUnit("{obligation}"))
+		if err != nil {
+			otel.Handle(err)
+		}
+	})
+	if obligationClosedCounter == nil {
+		return
+	}
+	obligationClosedCounter.Add(context.Background(), 1, metric.WithAttributes(
+		attribute.String("node", node), attribute.String("outcome", outcome)))
+}

spinifex/domains/ec2/volume/unpublish_obligation_test.go

@@ -0,0 +1,593 @@
+package volume
+
+//test:in-package — the obligation reaper and its policy are unexported, and
+// seeding a volume's metadata document is how an attachment is set up.
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	ebsmetadata "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ebsobligation "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/obligation"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	ebsprovider "github.com/mulgadc/spinifex/spinifex/providers/ebs"
+	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type fakeNodeLocator struct {
+	nodes map[string]string
+	err   error
+}
+
+var _ InstanceNodeLocator = (*fakeNodeLocator)(nil)
+
+func (f *fakeNodeLocator) InstanceLastNode(instanceID string) (string, bool, error) {
+	if f.err != nil {
+		return "", false, f.err
+	}
+	node, found := f.nodes[instanceID]
+	return node, found, nil
+}
+
+// wireObligations gives svc an obligation bucket on a JetStream of its own and
+// returns the connection under it, so a test can cut it.
+func wireObligations(t *testing.T, svc *VolumeServiceImpl, locator InstanceNodeLocator, members ...string) *nats.Conn {
+	t.Helper()
+	_, nc, js := testutil.StartTestJetStream(t)
+	store := ebsobligation.NewStore(js)
+	require.NoError(t, store.Open(t.Context()))
+	svc.SetUnpublishObligations(store, locator, func() []string { return members })
+	return nc
+}
+
+type unpublishCall struct {
+	volumeID, accountID, handle string
+	noSeal                      bool
+}
+
+// fakeUnpublisher stands in for a node's mounter: busy volumes refuse the
+// hold, publishedAt is the last publish this process made of each volume.
+type fakeUnpublisher struct {
+	mu          sync.Mutex
+	busy        map[string]bool
+	publishedAt map[string]time.Time
+	err         error
+	calls       []unpublishCall
+	held        int
+}
+
+var _ ObligationUnpublisher = (*fakeUnpublisher)(nil)
+
+func (f *fakeUnpublisher) HoldVolume(volumeID string) (VolumeHold, bool) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.busy[volumeID] {
+		return nil, false
+	}
+	f.held++
+	return &fakeHold{f: f, volumeID: volumeID}, true
+}
+
+func (f *fakeUnpublisher) unpublishes() []unpublishCall {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]unpublishCall(nil), f.calls...)
+}
+
+type fakeHold struct {
+	f        *fakeUnpublisher
+	volumeID string
+}
+
+var _ VolumeHold = (*fakeHold)(nil)
+
+func (h *fakeHold) PublishedSince(t time.Time) bool {
+	h.f.mu.Lock()
+	defer h.f.mu.Unlock()
+	at, ok := h.f.publishedAt[h.volumeID]
+	return ok && at.After(t)
+}
+
+func (h *fakeHold) Unpublish(_ context.Context, accountID, handle string, noSeal bool) error {
+	h.f.mu.Lock()
+	defer h.f.mu.Unlock()
+	h.f.calls = append(h.f.calls, unpublishCall{volumeID: h.volumeID, accountID: accountID, handle: handle, noSeal: noSeal})
+	return h.f.err
+}
+
+func (h *fakeHold) Release() {
+	h.f.mu.Lock()
+	defer h.f.mu.Unlock()
+	h.f.held--
+}
+
+// putAttachment writes volumeID's document with an explicit attachment, so a
+// test controls AttachedAt rather than racing the clock.
+func putAttachment(t *testing.T, svc *VolumeServiceImpl, volumeID, state, instanceID string, attachedAt time.Time) ebsmetadata.Volume {
+	t.Helper()
+	meta := ebsmetadata.Volume{
+		VolumeID: volumeID, TenantID: testVolAccountID, CapacityGiB: 1,
+		State: state, AvailabilityZone: "ap-southeast-2a", VolumeType: "gp3",
+		AttachedInstance: instanceID, AttachedAt: attachedAt, ProviderHandle: "handle-" + volumeID,
+	}
+	if instanceID != "" {
+		meta.DeviceName = "/dev/sdf"
+	}
+	require.NoError(t, svc.metadata.PutVolume(context.Background(), meta))
+	return meta
+}
+
+func readVolume(t *testing.T, svc *VolumeServiceImpl, volumeID string) ebsmetadata.Volume {
+	t.Helper()
+	meta, err := svc.metadata.GetVolume(context.Background(), testVolAccountID, volumeID)
+	require.NoError(t, err)
+	return meta
+}
+
+func openObligations(t *testing.T, svc *VolumeServiceImpl) []ebsobligation.UnpublishObligation {
+	t.Helper()
+	open, err := svc.obligations.List(context.Background())
+	require.NoError(t, err)
+	return open
+}
+
+func forceDetach(t *testing.T, svc *VolumeServiceImpl, volumeID string) (*ec2.VolumeAttachment, error) {
+	t.Helper()
+	return svc.ForceDetachVolume(context.Background(), &ec2.DetachVolumeInput{
+		VolumeId: aws.String(volumeID), Force: aws.Bool(true),
+	}, testVolAccountID)
+}
+
+// --- ForceDetachVolume records the obligation ---
+
+func TestForceDetach_RecordsOneObligationNamingThePublish(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	wireObligations(t, svc, &fakeNodeLocator{nodes: map[string]string{"i-old": "node-a"}}, "node-a", "node-b")
+	attachedAt := time.Unix(1_700_000_000, 0).UTC()
+	putAttachment(t, svc, "vol-fd-once", "in-use", "i-old", attachedAt)
+
+	_, err := forceDetach(t, svc, "vol-fd-once")
+	require.NoError(t, err)
+
+	open := openObligations(t, svc)
+	require.Len(t, open, 1, "only the node that last ran the instance is named")
+	o := open[0]
+	assert.Equal(t, "vol-fd-once", o.VolumeID)
+	assert.Equal(t, testVolAccountID, o.AccountID)
+	assert.Equal(t, "handle-vol-fd-once", o.Handle)
+	assert.Equal(t, "node-a", o.NodeID)
+	assert.Equal(t, "i-old", o.InstanceID)
+	assert.True(t, attachedAt.Equal(o.AttachedAt), "the obligation names the attachment it undoes")
+	assert.Equal(t, ebsobligation.SchemaVersion, o.SchemaVersion)
+}
+
+// A caller retrying a force detach whose clear failed writes the same key, so
+// the retry leaves one obligation, not two.
+func TestForceDetach_RetryLeavesOneObligation(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	wireObligations(t, svc, &fakeNodeLocator{nodes: map[string]string{"i-old": "node-a"}}, "node-a")
+	meta := putAttachment(t, svc, "vol-fd-retry", "in-use", "i-old", time.Unix(1_700_000_000, 0).UTC())
+
+	require.NoError(t, svc.recordObligations(context.Background(), meta, []string{"node-a"}))
+	_, err := forceDetach(t, svc, "vol-fd-retry")
+	require.NoError(t, err)
+
+	assert.Len(t, openObligations(t, svc), 1)
+}
+
+// With no record of the instance there is no node to name, so every member is
+// named and each one settles its own.
+func TestForceDetach_UnknownInstanceNamesEveryMember(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	wireObligations(t, svc, &fakeNodeLocator{}, "node-a", "node-b", "node-c")
+	putAttachment(t, svc, "vol-fd-every", "in-use", "i-gone", time.Unix(1_700_000_000, 0).UTC())
+
+	_, err := forceDetach(t, svc, "vol-fd-every")
+	require.NoError(t, err)
+
+	nodes := []string{}
+	for _, o := range openObligations(t, svc) {
+		nodes = append(nodes, o.NodeID)
+	}
+	assert.ElementsMatch(t, []string{"node-a", "node-b", "node-c"}, nodes)
+}
+
+// The attachment is the only record of where the export is. If the obligation
+// cannot be written, clearing it would forget an export, so nothing is cleared.
+func TestForceDetach_ObligationWriteFailureClearsNothing(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	nc := wireObligations(t, svc, &fakeNodeLocator{nodes: map[string]string{"i-old": "node-a"}}, "node-a")
+	putAttachment(t, svc, "vol-fd-nowrite", "in-use", "i-old", time.Unix(1_700_000_000, 0).UTC())
+	nc.Close()
+
+	_, err := forceDetach(t, svc, "vol-fd-nowrite")
+
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInternalError, err.Error())
+	meta := readVolume(t, svc, "vol-fd-nowrite")
+	assert.Equal(t, "in-use", meta.State)
+	assert.Equal(t, "i-old", meta.AttachedInstance)
+}
+
+func TestForceDetach_UnnameableNodeClearsNothing(t *testing.T) {
+	for name, locator := range map[string]InstanceNodeLocator{
+		"locator fails":            &fakeNodeLocator{err: errors.New("instance bucket unreachable")},
+		"no record and no members": &fakeNodeLocator{},
+		"no locator configured":    nil,
+	} {
+		t.Run(name, func(t *testing.T) {
+			svc := newTestVolumeService("ap-southeast-2a")
+			wireObligations(t, svc, locator)
+			putAttachment(t, svc, "vol-fd-noname", "in-use", "i-old", time.Unix(1_700_000_000, 0).UTC())
+
+			_, err := forceDetach(t, svc, "vol-fd-noname")
+
+			require.Error(t, err)
+			assert.Equal(t, awserrors.ErrorInternalError, err.Error())
+			assert.Equal(t, "i-old", readVolume(t, svc, "vol-fd-noname").AttachedInstance)
+			assert.Empty(t, openObligations(t, svc))
+		})
+	}
+}
+
+// providerCallCounter fails the test's expectations if the request path makes
+// any provider call: an unpublish waits on a node that may never answer.
+type providerCallCounter struct {
+	ebsprovider.EBSProvider
+
+	mu        sync.Mutex
+	unpublish int
+	publish   int
+}
+
+func (p *providerCallCounter) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
+	p.mu.Lock()
+	p.unpublish++
+	p.mu.Unlock()
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func (p *providerCallCounter) PublishVolume(ctx context.Context, req ebsprovider.PublishVolumeRequest) (*ebsprovider.PublishedVolume, error) {
+	p.mu.Lock()
+	p.publish++
+	p.mu.Unlock()
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+// TestS8_ForceDetachRecordsTheObligationAndReturns pins that a force detach
+// never waits on the node holding the export: it decides its errors, records
+// the obligation and returns, and the unpublish is left to that node's reaper.
+func TestS8_ForceDetachRecordsTheObligationAndReturns(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	provider := &providerCallCounter{EBSProvider: ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})}
+	svc.SetEBSProvider(provider)
+	wireObligations(t, svc, &fakeNodeLocator{nodes: map[string]string{"i-wedged": "node-unreachable"}}, "node-unreachable")
+	putAttachment(t, svc, "vol-s8-forcedetach", "in-use", "i-wedged", time.Unix(1_700_000_000, 0).UTC())
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	attachment, err := svc.ForceDetachVolume(ctx, &ec2.DetachVolumeInput{
+		VolumeId: aws.String("vol-s8-forcedetach"), Force: aws.Bool(true),
+	}, testVolAccountID)
+	require.NoError(t, err)
+	require.NoError(t, ctx.Err(), "the request returned on its own, not on the deadline")
+
+	assert.Equal(t, "detaching", aws.StringValue(attachment.State))
+	provider.mu.Lock()
+	assert.Zero(t, provider.unpublish, "the request path makes no unpublish call")
+	assert.Zero(t, provider.publish)
+	provider.mu.Unlock()
+
+	open := openObligations(t, svc)
+	require.Len(t, open, 1, "the unpublish the request did not wait for is recorded")
+	assert.Equal(t, "node-unreachable", open[0].NodeID)
+	assert.Empty(t, readVolume(t, svc, "vol-s8-forcedetach").AttachedInstance)
+}
+
+// --- Terminate paths ---
+
+func TestTerminate_OwedUnpublishRecordsOneObligationBeforeTheClear(t *testing.T) {
+	for name, terminate := range map[string]func(*VolumeServiceImpl, string, ec2instance.OwedUnpublish) error{
+		"detach": func(svc *VolumeServiceImpl, id string, owed ec2instance.OwedUnpublish) error {
+			return svc.DetachVolumeOnTerminate(context.Background(), id, testVolAccountID, owed)
+		},
+		"delete": func(svc *VolumeServiceImpl, id string, owed ec2instance.OwedUnpublish) error {
+			return svc.DeleteVolumeOnTerminate(context.Background(), id, testVolAccountID, owed)
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			svc := newTestVolumeService("ap-southeast-2a")
+			svc.snapshotKV = setupTestVolumeKV(t)
+			wireObligations(t, svc, &fakeNodeLocator{}, "node-a")
+			seedTestProviderVolume(t, svc, "vol-term-owed", 1)
+			putAttachment(t, svc, "vol-term-owed", "in-use", "i-term", time.Unix(1_700_000_000, 0).UTC())
+
+			require.NoError(t, terminate(svc, "vol-term-owed", ec2instance.OwedUnpublish{NodeID: "node-a", InstanceID: "i-term"}))
+
+			open := openObligations(t, svc)
+			require.Len(t, open, 1)
+			assert.Equal(t, "node-a", open[0].NodeID)
+			assert.Equal(t, "i-term", open[0].InstanceID)
+		})
+	}
+}
+
+// A terminate whose unmount completed owes nothing, and an attachment that is
+// already some other instance's is not this terminate's to undo.
+func TestTerminate_NothingOwedRecordsNoObligation(t *testing.T) {
+	for name, tc := range map[string]struct {
+		attachedTo string
+		owed       ec2instance.OwedUnpublish
+	}{
+		"unmount completed":   {attachedTo: "i-term", owed: ec2instance.OwedUnpublish{}},
+		"attached to another": {attachedTo: "i-other", owed: ec2instance.OwedUnpublish{NodeID: "node-a", InstanceID: "i-term"}},
+	} {
+		t.Run(name, func(t *testing.T) {
+			svc := newTestVolumeService("ap-southeast-2a")
+			wireObligations(t, svc, &fakeNodeLocator{}, "node-a")
+			putAttachment(t, svc, "vol-term-none", "in-use", tc.attachedTo, time.Unix(1_700_000_000, 0).UTC())
+
+			require.NoError(t, svc.DetachVolumeOnTerminate(context.Background(), "vol-term-none", testVolAccountID, tc.owed))
+
+			assert.Empty(t, openObligations(t, svc))
+		})
+	}
+}
+
+func TestTerminate_ObligationWriteFailureClearsNothing(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	nc := wireObligations(t, svc, &fakeNodeLocator{}, "node-a")
+	putAttachment(t, svc, "vol-term-nowrite", "in-use", "i-term", time.Unix(1_700_000_000, 0).UTC())
+	nc.Close()
+
+	err := svc.DetachVolumeOnTerminate(context.Background(), "vol-term-nowrite", testVolAccountID,
+		ec2instance.OwedUnpublish{NodeID: "node-a", InstanceID: "i-term"})
+
+	require.Error(t, err)
+	assert.Equal(t, "i-term", readVolume(t, svc, "vol-term-nowrite").AttachedInstance)
+}
+
+// --- The reaper ---
+
+func recordObligation(t *testing.T, svc *VolumeServiceImpl, node, volumeID, instanceID string, attachedAt time.Time) ebsobligation.UnpublishObligation {
+	t.Helper()
+	o := ebsobligation.UnpublishObligation{
+		VolumeID: volumeID, AccountID: testVolAccountID, Handle: "handle-" + volumeID,
+		NodeID: node, InstanceID: instanceID, AttachedAt: attachedAt, RecordedAt: attachedAt.Add(time.Minute),
+	}
+	require.NoError(t, svc.obligations.Record(context.Background(), o))
+	return o
+}
+
+func sweepObligations(t *testing.T, svc *VolumeServiceImpl, node string, unpublisher ObligationUnpublisher) int {
+	t.Helper()
+	closed, err := svc.NewUnpublishObligationReaper(node, unpublisher).Sweep(context.Background())
+	require.NoError(t, err)
+	return closed
+}
+
+// TestS7_ObligationReaperNeverUnpublishesACurrentAttachment pins that the
+// reaper, which runs with no customer asking, never tears down an export a
+// live attachment uses and never removes or rewrites the volume.
+func TestS7_ObligationReaperNeverUnpublishesACurrentAttachment(t *testing.T) {
+	cleared := time.Unix(1_700_000_000, 0).UTC()
+	for name, tc := range map[string]struct {
+		state       string
+		attachedTo  string
+		attachedAt  time.Time
+		publishedAt time.Time // a publish this process made, zero if none
+		stillOpen   bool
+	}{
+		// The volume moved to another instance, possibly on this very node.
+		"attached to another instance": {state: "in-use", attachedTo: "i-new", attachedAt: cleared.Add(time.Hour)},
+		// The same instance was attached again after the clear.
+		"same instance attached again": {state: "in-use", attachedTo: "i-old", attachedAt: cleared.Add(time.Hour)},
+		// The clear has not landed: the obligation was written first.
+		"attachment not yet cleared": {state: "in-use", attachedTo: "i-old", attachedAt: cleared, stillOpen: true},
+		// A hot attach published here and has not yet written in-use.
+		"newer publish here, record not yet in-use": {state: "available", publishedAt: cleared.Add(time.Hour), stillOpen: true},
+	} {
+		t.Run(name, func(t *testing.T) {
+			svc := newTestVolumeService("ap-southeast-2a")
+			wireObligations(t, svc, &fakeNodeLocator{}, "node-a")
+			before := putAttachment(t, svc, "vol-s7-current", tc.state, tc.attachedTo, tc.attachedAt)
+			recordObligation(t, svc, "node-a", "vol-s7-current", "i-old", cleared)
+			unpublisher := &fakeUnpublisher{publishedAt: map[string]time.Time{}}
+			if !tc.publishedAt.IsZero() {
+				unpublisher.publishedAt["vol-s7-current"] = tc.publishedAt
+			}
+
+			sweepObligations(t, svc, "node-a", unpublisher)
+
+			assert.Empty(t, unpublisher.unpublishes(), "a current attachment's export must never be unpublished")
+			assert.Zero(t, unpublisher.held, "every hold is released")
+			after := readVolume(t, svc, "vol-s7-current")
+			assert.Equal(t, before.State, after.State, "the reaper never rewrites the volume")
+			assert.Equal(t, before.AttachedInstance, after.AttachedInstance)
+			if tc.stillOpen {
+				assert.Len(t, openObligations(t, svc), 1, "the obligation waits for its attachment to be gone")
+			} else {
+				assert.Empty(t, openObligations(t, svc), "a superseded obligation closes without a provider call")
+			}
+		})
+	}
+}
+
+// TestS7_ObligationReaperSealsLiveDataAndDropsOnlyDeletingVolumes pins that
+// the reaper keeps a surviving volume's writes and only drops them unsealed
+// for a volume that is being deleted or is gone.
+func TestS7_ObligationReaperSealsLiveDataAndDropsOnlyDeletingVolumes(t *testing.T) {
+	cleared := time.Unix(1_700_000_000, 0).UTC()
+	for name, tc := range map[string]struct {
+		state  string // empty for no document
+		noSeal bool
+	}{
+		"available keeps its writes": {state: "available", noSeal: false},
+		"deleting is dropped":        {state: volumeStateDeleting, noSeal: true},
+		"absent is dropped":          {state: "", noSeal: true},
+	} {
+		t.Run(name, func(t *testing.T) {
+			svc := newTestVolumeService("ap-southeast-2a")
+			wireObligations(t, svc, &fakeNodeLocator{}, "node-a")
+			if tc.state != "" {
+				putAttachment(t, svc, "vol-s7-settle", tc.state, "", time.Time{})
+			}
+			recordObligation(t, svc, "node-a", "vol-s7-settle", "i-old", cleared)
+			unpublisher := &fakeUnpublisher{}
+
+			assert.Equal(t, 1, sweepObligations(t, svc, "node-a", unpublisher))
+
+			assert.Equal(t, []unpublishCall{{volumeID: "vol-s7-settle", accountID: testVolAccountID, handle: "handle-vol-s7-settle", noSeal: tc.noSeal}},
+				unpublisher.unpublishes())
+			assert.Empty(t, openObligations(t, svc))
+			if tc.state != "" {
+				assert.Equal(t, tc.state, readVolume(t, svc, "vol-s7-settle").State, "the reaper never rewrites the volume")
+			}
+		})
+	}
+}
+
+// TestS7_ObligationReaperNeverTreatsAnUnreadableVolumeAsGone pins that a read
+// failing for a reason other than absence leaves the obligation open: taking
+// it for a deleted volume would drop live writes unsealed.
+func TestS7_ObligationReaperNeverTreatsAnUnreadableVolumeAsGone(t *testing.T) {
+	memory := objectstore.NewMemoryObjectStore()
+	svc := newTestVolumeServiceWithStore("ap-southeast-2a", memory)
+	wireObligations(t, svc, &fakeNodeLocator{}, "node-a")
+	putAttachment(t, svc, "vol-s7-unread", "available", "", time.Time{})
+	recordObligation(t, svc, "node-a", "vol-s7-unread", "i-old", time.Unix(1_700_000_000, 0).UTC())
+	failing := &getFailingStore{MemoryObjectStore: memory, failKey: metadataKey(t, "vol-s7-unread")}
+	svc.metadata = ebsmetadata.NewStore(failing, "test-bucket")
+	unpublisher := &fakeUnpublisher{}
+
+	assert.Zero(t, sweepObligations(t, svc, "node-a", unpublisher))
+
+	assert.Empty(t, unpublisher.unpublishes())
+	open := openObligations(t, svc)
+	require.Len(t, open, 1)
+	assert.Equal(t, 1, open[0].Attempts, "the failure is counted where an operator can see it")
+	assert.Contains(t, open[0].Error, "reconstruction failed")
+}
+
+// End to end through the service: a force detach followed by a new attachment
+// of the same volume on the same node must leave that new export alone.
+func TestS7_ForceDetachThenReattachOnTheSameNodeKeepsTheNewExport(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	wireObligations(t, svc, &fakeNodeLocator{nodes: map[string]string{"i-old": "node-a", "i-new": "node-a"}}, "node-a")
+	putAttachment(t, svc, "vol-s7-reattach", "in-use", "i-old", time.Now().Add(-time.Hour).UTC())
+
+	_, err := forceDetach(t, svc, "vol-s7-reattach")
+	require.NoError(t, err)
+	require.NoError(t, svc.UpdateVolumeState(testVolAccountID, "vol-s7-reattach", "in-use", "i-new", "/dev/sdg"))
+	unpublisher := &fakeUnpublisher{}
+
+	assert.Equal(t, 1, sweepObligations(t, svc, "node-a", unpublisher))
+
+	assert.Empty(t, unpublisher.unpublishes(), "the new attachment's export must survive the old obligation")
+	assert.Equal(t, "i-new", readVolume(t, svc, "vol-s7-reattach").AttachedInstance)
+	assert.Empty(t, openObligations(t, svc))
+}
+
+func TestObligationReaper_SettlesOnlyItsOwnNode(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	wireObligations(t, svc, &fakeNodeLocator{}, "node-a", "node-b")
+	putAttachment(t, svc, "vol-own", "available", "", time.Time{})
+	recordObligation(t, svc, "node-b", "vol-own", "i-old", time.Unix(1_700_000_000, 0).UTC())
+	unpublisher := &fakeUnpublisher{}
+
+	assert.Zero(t, sweepObligations(t, svc, "node-a", unpublisher))
+
+	assert.Empty(t, unpublisher.unpublishes(), "only the node holding the export can tear it down")
+	assert.Len(t, openObligations(t, svc), 1)
+}
+
+// A publish or unpublish of the volume in flight on this node holds its lock;
+// the sweep skips it rather than waiting behind a possibly wedged call.
+func TestObligationReaper_SkipsABusyVolume(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	wireObligations(t, svc, &fakeNodeLocator{}, "node-a")
+	putAttachment(t, svc, "vol-busy", "available", "", time.Time{})
+	recordObligation(t, svc, "node-a", "vol-busy", "i-old", time.Unix(1_700_000_000, 0).UTC())
+	unpublisher := &fakeUnpublisher{busy: map[string]bool{"vol-busy": true}}
+
+	assert.Zero(t, sweepObligations(t, svc, "node-a", unpublisher))
+
+	assert.Empty(t, unpublisher.unpublishes())
+	open := openObligations(t, svc)
+	require.Len(t, open, 1)
+	assert.Zero(t, open[0].Attempts, "a busy volume is not a failure")
+}
+
+func TestObligationReaper_FailedUnpublishStaysOpenAndCounts(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	wireObligations(t, svc, &fakeNodeLocator{}, "node-a")
+	putAttachment(t, svc, "vol-fail", "available", "", time.Time{})
+	recordObligation(t, svc, "node-a", "vol-fail", "i-old", time.Unix(1_700_000_000, 0).UTC())
+	unpublisher := &fakeUnpublisher{err: errors.New("seal refused")}
+
+	assert.Zero(t, sweepObligations(t, svc, "node-a", unpublisher))
+	assert.Zero(t, sweepObligations(t, svc, "node-a", unpublisher))
+
+	open := openObligations(t, svc)
+	require.Len(t, open, 1)
+	assert.Equal(t, 2, open[0].Attempts)
+	assert.Equal(t, "seal refused", open[0].Error)
+	assert.Zero(t, unpublisher.held, "a failed unpublish still releases the hold")
+}
+
+// An obligation with no attachment time falls back to when it was recorded, so
+// a publish here after the clear still counts as newer.
+func TestObligationSince_FallsBackToRecordedAt(t *testing.T) {
+	recorded := time.Unix(1_700_000_000, 0)
+	assert.Equal(t, recorded, obligationSince(ebsobligation.UnpublishObligation{RecordedAt: recorded}))
+	attached := recorded.Add(-time.Hour)
+	assert.Equal(t, attached, obligationSince(ebsobligation.UnpublishObligation{AttachedAt: attached, RecordedAt: recorded}))
+}
+
+// --- The cluster-wide close ---
+
+func TestObligationCloser_ClosesOnlyDepartedNodes(t *testing.T) {
+	svc := newTestVolumeService("ap-southeast-2a")
+	wireObligations(t, svc, &fakeNodeLocator{}, "node-a")
+	at := time.Unix(1_700_000_000, 0).UTC()
+	recordObligation(t, svc, "node-gone", "vol-c1", "i-1", at)
+	recordObligation(t, svc, "node-gone", "vol-c2", "i-2", at)
+	recordObligation(t, svc, "node-down", "vol-c3", "i-3", at)
+	recordObligation(t, svc, "node-unknown", "vol-c4", "i-4", at)
+	asked := map[string]int{}
+	departed := func(_ context.Context, node string) (bool, error) {
+		asked[node]++
+		switch node {
+		case "node-gone":
+			return true, nil
+		case "node-unknown":
+			return false, errors.New("heartbeat bucket unreachable")
+		default:
+			return false, nil
+		}
+	}
+
+	closed, err := svc.NewUnpublishObligationCloser(departed).Sweep(context.Background())
+	require.NoError(t, err)
+
+	assert.Equal(t, 2, closed)
+	remaining := []string{}
+	for _, o := range openObligations(t, svc) {
+		remaining = append(remaining, o.NodeID)
+	}
+	assert.ElementsMatch(t, []string{"node-down", "node-unknown"}, remaining,
+		"a node that is down, or whose membership cannot be read, keeps its obligations")
+	assert.Equal(t, 1, asked["node-gone"], "membership is asked once per node per sweep")
+}

spinifex/operator/cli/admin.go

@@ -31,6 +31,7 @@ import (
 	"github.com/mulgadc/bluebottle/pkg/safecast"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ebsobligation "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/obligation"
 	ec2image "github.com/mulgadc/spinifex/spinifex/domains/ec2/image"
 	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
 	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
@@ -247,6 +248,18 @@ the backend. This is the list of volumes where that would cost something.`,
 	Run: runVolumesUnsealedCmd,
 }
 
+var volumesObligationsCmd = &cobra.Command{
+	Use:   "obligations",
+	Short: "Report volume exports a cleared attachment left on a node",
+	Long: `List the open unpublish obligations: exports a force detach, or a terminate
+whose unmount did not complete, left on the node that held the attachment.
+
+The node named settles each one once it answers. An obligation stays open
+while its node is down; it closes without the node only once the node has
+been removed from the cluster. This command only reports.`,
+	Run: runVolumesObligationsCmd,
+}
+
 var accountCmd = &cobra.Command{
 	Use:   "account",
 	Short: "Manage Spinifex accounts",
@@ -351,6 +364,7 @@ func init() {
 	adminCmd.AddCommand(volumesCmd)
 	volumesCmd.AddCommand(volumesOrphansCmd)
 	volumesCmd.AddCommand(volumesUnsealedCmd)
+	volumesCmd.AddCommand(volumesObligationsCmd)
 
 	adminCmd.AddCommand(accountCmd)
 	accountCmd.AddCommand(accountCreateCmd)
@@ -4103,6 +4117,45 @@ func printChecksumError(w io.Writer, imageFile, imageName string, image imagecat
 	fmt.Fprintf(w, "  spx admin images import --name %s --force\n", imageName)
 }
 
+func runVolumesObligationsCmd(_ *cobra.Command, _ []string) {
+	_, nc, err := loadConfigAndConnect()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Could not connect to the cluster: %v\n", err)
+		os.Exit(1)
+	}
+	defer nc.Close()
+
+	open, err := ebsobligation.ListOpen(context.Background(), nc)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, "Could not read the unpublish obligations:", err)
+		os.Exit(1)
+	}
+	printUnpublishObligations(os.Stdout, open, time.Now())
+}
+
+// printUnpublishObligations writes the open obligations, oldest first, with
+// the node each waits on and its age.
+func printUnpublishObligations(w io.Writer, open []ebsobligation.UnpublishObligation, now time.Time) {
+	if len(open) == 0 {
+		fmt.Fprintln(w, "No open unpublish obligations: no node holds an export of a cleared attachment.")
+		return
+	}
+	sort.Slice(open, func(i, j int) bool { return open[i].RecordedAt.Before(open[j].RecordedAt) })
+	fmt.Fprintf(w, "%d open unpublish obligation(s):\n\n", len(open))
+	for _, o := range open {
+		fmt.Fprintf(w, "  %s\n", o.VolumeID)
+		fmt.Fprintf(w, "    node:     %s\n", o.NodeID)
+		fmt.Fprintf(w, "    instance: %s\n", o.InstanceID)
+		fmt.Fprintf(w, "    age:      %s\n", now.Sub(o.RecordedAt).Truncate(time.Second))
+		if o.Attempts > 0 {
+			fmt.Fprintf(w, "    attempts: %d\n", o.Attempts)
+		}
+		if o.Error != "" {
+			fmt.Fprintf(w, "    error:    %s\n", o.Error)
+		}
+	}
+}
+
 func runVolumesUnsealedCmd(_ *cobra.Command, _ []string) {
 	_, nc, err := loadConfigAndConnect()
 	if err != nil {

spinifex/operator/cli/admin_volumes_obligations_test.go

@@ -0,0 +1,37 @@
+package cli
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+	"time"
+
+	ebsobligation "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/obligation"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPrintUnpublishObligations_OldestFirstWithFailures(t *testing.T) {
+	now := time.Unix(1_700_010_000, 0)
+	var out bytes.Buffer
+	printUnpublishObligations(&out, []ebsobligation.UnpublishObligation{
+		{VolumeID: "vol-newer", NodeID: "node-b", InstanceID: "i-2", RecordedAt: now.Add(-90 * time.Second)},
+		{VolumeID: "vol-older", NodeID: "node-a", InstanceID: "i-1", RecordedAt: now.Add(-2 * time.Hour),
+			Attempts: 3, Error: "seal refused"},
+	}, now)
+
+	text := out.String()
+	assert.Contains(t, text, "2 open unpublish obligation(s)")
+	assert.Less(t, strings.Index(text, "vol-older"), strings.Index(text, "vol-newer"), "the longest-open obligation is listed first")
+	assert.Contains(t, text, "node:     node-a")
+	assert.Contains(t, text, "instance: i-1")
+	assert.Contains(t, text, "age:      2h0m0s")
+	assert.Contains(t, text, "attempts: 3")
+	assert.Contains(t, text, "error:    seal refused")
+	assert.Equal(t, 1, strings.Count(text, "attempts:"), "an obligation that never failed shows no attempts")
+}
+
+func TestPrintUnpublishObligations_None(t *testing.T) {
+	var out bytes.Buffer
+	printUnpublishObligations(&out, nil, time.Now())
+	assert.Contains(t, out.String(), "No open unpublish obligations")
+}

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-ebs-unpublish-obligations` | `1` | `ebsobligation.BucketVersion` |
 
 ## Registered migrations
 

spinifex/runtime/compute/vm/crash_recovery.go

@@ -152,7 +152,7 @@ func (m *Manager) HandleCrash(instance *VM, waitErr error) {
 	removeTelemetryArtifacts(instance)
 
 	if m.deps.VolumeMounter != nil {
-		if err := m.deps.VolumeMounter.Unmount(context.Background(), instance); err != nil {
+		if err := m.unmountVolumes(context.Background(), instance); err != nil {
 			slog.Error("Volume unmount failed during crash handling",
 				"instance", instance.ID, "err", err)
 		}

spinifex/runtime/compute/vm/errors.go

@@ -1,6 +1,9 @@
 package vm
 
-import "errors"
+import (
+	"errors"
+	"fmt"
+)
 
 // ErrInstanceNotFound is returned by manager methods that look up an
 // instance by id when no entry exists in the running map.
@@ -16,6 +19,44 @@ var ErrInvalidTransition = errors.New("invalid state transition")
 // because stopping viperblock next is what tears the volume's state.
 var ErrVolumeSealFailed = errors.New("volume seal failed")
 
+// VolumeUnpublishError is one volume an unmount left exported. A VolumeMounter
+// reports each such volume with it, so the manager knows which publishes a
+// later terminate leaves in place.
+type VolumeUnpublishError struct {
+	VolumeID string
+	Err      error
+}
+
+func (e *VolumeUnpublishError) Error() string {
+	return fmt.Sprintf("seal %s: %v", e.VolumeID, e.Err)
+}
+
+func (e *VolumeUnpublishError) Unwrap() error { return e.Err }
+
+// volumesLeftPublished returns the volumes an unmount error leaves exported:
+// one VolumeUnpublishError per volume, joined. An error that names none leaves
+// the outcome of every volume unknown, so all of them are returned.
+func volumesLeftPublished(err error, volumes []string) []string {
+	if err == nil {
+		return nil
+	}
+	parts := []error{err}
+	var joined interface{ Unwrap() []error }
+	if errors.As(err, &joined) {
+		parts = joined.Unwrap()
+	}
+	var named []string
+	for _, part := range parts {
+		if ve, ok := errors.AsType[*VolumeUnpublishError](part); ok {
+			named = append(named, ve.VolumeID)
+		}
+	}
+	if len(named) == 0 {
+		return append([]string(nil), volumes...)
+	}
+	return named
+}
+
 // ErrAttachmentLimitExceeded is returned by AttachVolume when the
 // instance has no free /dev/sd[f-p] slot to assign.
 var ErrAttachmentLimitExceeded = errors.New("attachment limit exceeded")

spinifex/runtime/compute/vm/fakes_test.go

@@ -217,6 +217,9 @@ type recordingInstanceCleaner struct {
 	releaseGPU          []string
 	releaseAutoAssigned []string
 
+	// owedAtDelete is each DeleteVolumes call's UnpublishOwed as passed in.
+	owedAtDelete [][]string
+
 	// releaseAutoAssignedOK is what ReleaseAutoAssignedPublicIP reports: false is
 	// an Elastic IP the cleaner refuses to free, true an auto-assigned address.
 	releaseAutoAssignedOK bool
@@ -235,6 +238,7 @@ func (c *recordingInstanceCleaner) DeleteVolumes(v *VM) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.deleteVolumes = append(c.deleteVolumes, v.ID)
+	c.owedAtDelete = append(c.owedAtDelete, append([]string(nil), v.UnpublishOwed...))
 	return c.deleteVolumesErr
 }
 

spinifex/runtime/compute/vm/lifecycle.go

@@ -166,7 +166,7 @@ func (m *Manager) launch(ctx context.Context, instance *VM) (err error) {
 	// fails, unmounting seals the untouched backend and keeps the launch from
 	// exposing a writer that snapshots would classify as available.
 	if stateErr := m.markAttachedVolumesInUse(instance); stateErr != nil {
-		if unmountErr := m.deps.VolumeMounter.Unmount(ctx, instance); unmountErr != nil {
+		if unmountErr := m.unmountVolumes(ctx, instance); unmountErr != nil {
 			return errors.Join(
 				fmt.Errorf("persist attached volume state: %w", stateErr),
 				fmt.Errorf("rollback mounted volumes: %w", unmountErr),

spinifex/runtime/compute/vm/record.go

@@ -96,6 +96,7 @@ type InstanceStatus struct {
 	IamInstanceProfileAssociationId string              `json:"iam_instance_profile_association_id,omitempty"`
 
 	Teardown       map[string]string `json:"teardown,omitempty"`
+	UnpublishOwed  []string          `json:"unpublish_owed,omitempty"`
 	TerminatedAt   time.Time         `json:"terminated_at,omitzero"`
 	ShuttingDownAt time.Time         `json:"shutting_down_at,omitzero"`
 	FirstLaunch    bool              `json:"first_launch,omitempty"`
@@ -160,6 +161,7 @@ func (v *VM) Record() *InstanceRecord {
 			GPUAttachments:                  v.GPUAttachments,
 			IamInstanceProfileAssociationId: v.IamInstanceProfileAssociationId,
 			Teardown:                        v.Teardown,
+			UnpublishOwed:                   v.UnpublishOwed,
 			TerminatedAt:                    v.TerminatedAt,
 			ShuttingDownAt:                  v.ShuttingDownAt,
 			FirstLaunch:                     v.FirstLaunch,
@@ -219,6 +221,7 @@ func VMFromRecord(r *InstanceRecord) *VM {
 		GPUAttachments:                  r.Status.GPUAttachments,
 		IamInstanceProfileAssociationId: r.Status.IamInstanceProfileAssociationId,
 		Teardown:                        r.Status.Teardown,
+		UnpublishOwed:                   r.Status.UnpublishOwed,
 		TerminatedAt:                    r.Status.TerminatedAt,
 		ShuttingDownAt:                  r.Status.ShuttingDownAt,
 		FirstLaunch:                     r.Status.FirstLaunch,

spinifex/runtime/compute/vm/record_test.go

@@ -230,6 +230,7 @@ func populatedVM() *vm.VM {
 		GPUAttachments:                  []gpu.GPUAttachment{{PCIAddress: "0000:81:00.0"}},
 		IamInstanceProfileAssociationId: "iip-assoc-1",
 		Teardown:                        map[string]string{"volume": "done"},
+		UnpublishOwed:                   []string{"vol-1"},
 		TerminatedAt:                    time.Date(2026, 8, 28, 2, 0, 0, 0, time.UTC),
 		ShuttingDownAt:                  time.Date(2026, 8, 28, 3, 0, 0, 0, time.UTC),
 		FirstLaunch:                     true,

spinifex/runtime/compute/vm/shutdown.go

@@ -432,6 +432,8 @@ func (m *Manager) finalizeTerminated(instance *VM) error {
 // finalizeTerminated is a no-op — the shutting-down → terminated transition is
 // already spent and terminated is terminal.
 func (m *Manager) reconcileVanishedQEMU(instance *VM) error {
+	// The wedged terminate may not have finished its unmount.
+	m.markAllVolumesOwed(instance)
 	m.markTeardown(instance, TeardownQEMU, TeardownDone)
 	m.stampOutstandingTeardownFailed(instance)
 	return m.finalizeTerminated(instance)
@@ -457,6 +459,9 @@ func (m *Manager) forceFinalizeStuckTerminate(instance *VM) error {
 	}
 	m.markTeardown(instance, TeardownQEMU, TeardownDone)
 
+	// No unmount ran here, so every attachment the cleaner clears leaves its
+	// publish in place and is owed an unpublish.
+	m.markAllVolumesOwed(instance)
 	if m.deps.InstanceCleaner != nil {
 		m.markTeardownResult(instance, TeardownVolumes, m.deps.InstanceCleaner.DeleteVolumes(instance))
 	}
@@ -618,7 +623,7 @@ func (m *Manager) shutdownAndUnmount(instance *VM) error {
 
 	var sealErr error
 	if m.deps.VolumeMounter != nil {
-		if err := m.deps.VolumeMounter.Unmount(context.Background(), instance); err != nil {
+		if err := m.unmountVolumes(context.Background(), instance); err != nil {
 			slog.Error("Volume unmount failed", "id", instance.ID, "err", err)
 			sealErr = fmt.Errorf("%w for instance %s: %w", ErrVolumeSealFailed, instance.ID, err)
 		}
@@ -635,6 +640,34 @@ func (m *Manager) shutdownAndUnmount(instance *VM) error {
 	return sealErr
 }
 
+// unmountVolumes unmounts the instance's volumes and records the ones the
+// unmount left exported, which a terminate clearing their attachments then
+// owes an unpublish. Every unmount replaces the record.
+func (m *Manager) unmountVolumes(ctx context.Context, instance *VM) error {
+	err := m.deps.VolumeMounter.Unmount(ctx, instance)
+	owed := volumesLeftPublished(err, instanceVolumeNames(instance))
+	m.Inspect(instance, func(v *VM) { v.UnpublishOwed = owed })
+	return err
+}
+
+// markAllVolumesOwed records every volume as still exported, for a terminate
+// finished without an unmount that is known to have completed.
+func (m *Manager) markAllVolumesOwed(instance *VM) {
+	owed := instanceVolumeNames(instance)
+	m.Inspect(instance, func(v *VM) { v.UnpublishOwed = owed })
+}
+
+// instanceVolumeNames lists the volumes the instance has requests for.
+func instanceVolumeNames(instance *VM) []string {
+	instance.EBSRequests.Mu.Lock()
+	defer instance.EBSRequests.Mu.Unlock()
+	names := make([]string, 0, len(instance.EBSRequests.Requests))
+	for _, req := range instance.EBSRequests.Requests {
+		names = append(names, req.Name)
+	}
+	return names
+}
+
 // ForgetSuperseded stops any local copy of an instance another node now owns
 // and removes it from this node's view.
 //

spinifex/runtime/compute/vm/unpublish_owed_test.go

@@ -0,0 +1,96 @@
+package vm
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
+	hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func withVolumes(v *VM, names ...string) *VM {
+	for _, name := range names {
+		v.EBSRequests.Requests = append(v.EBSRequests.Requests, viperblocklegacyv1.EBSRequest{Name: name})
+	}
+	return v
+}
+
+func TestVolumesLeftPublished(t *testing.T) {
+	volumes := []string{"vol-a", "vol-b", "vol-c"}
+
+	assert.Nil(t, volumesLeftPublished(nil, volumes), "a completed unmount leaves nothing exported")
+
+	named := fmt.Errorf("unmount i-1: %w", errors.Join(
+		&VolumeUnpublishError{VolumeID: "vol-b", Err: errors.New("seal stalled")},
+		fmt.Errorf("wrapped: %w", &VolumeUnpublishError{VolumeID: "vol-c", Err: errors.New("no responders")}),
+	))
+	assert.Equal(t, []string{"vol-b", "vol-c"}, volumesLeftPublished(named, volumes),
+		"only the volumes the mounter names are left exported")
+
+	assert.Equal(t, volumes, volumesLeftPublished(errors.New("mounter unreachable"), volumes),
+		"an error naming no volume leaves every volume's outcome unknown")
+}
+
+func TestVolumeUnpublishError_UnwrapsItsCause(t *testing.T) {
+	cause := errors.New("seal stalled")
+	err := &VolumeUnpublishError{VolumeID: "vol-a", Err: cause}
+	require.ErrorIs(t, err, cause)
+	assert.Equal(t, "seal vol-a: seal stalled", err.Error())
+}
+
+// Every unmount replaces the record, so a later unmount that completes clears
+// what an earlier failed one left.
+func TestShutdownAndUnmount_RecordsTheVolumesTheUnmountLeftExported(t *testing.T) {
+	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
+	mounter := &fakeVolumeMounter{unmountErr: errors.Join(
+		&VolumeUnpublishError{VolumeID: "vol-data", Err: errSealStalled})}
+	m := NewManagerWithDeps(Deps{VolumeMounter: mounter})
+	instance := withVolumes(&VM{ID: "i-owed"}, "vol-root", "vol-data")
+
+	require.ErrorIs(t, m.shutdownAndUnmount(instance), ErrVolumeSealFailed)
+	assert.Equal(t, []string{"vol-data"}, instance.UnpublishOwed)
+
+	mounter.unmountErr = nil
+	require.NoError(t, m.shutdownAndUnmount(instance))
+	assert.Empty(t, instance.UnpublishOwed)
+}
+
+// The force-complete path kills QEMU without an unmount, so the cleaner that
+// clears the attachments must see every volume as still exported.
+func TestForceFinalizeStuckTerminate_OwesAnUnpublishForEveryVolume(t *testing.T) {
+	hostprocess.ShortenProcessExitPollForTest(t, 5*time.Millisecond)
+	cleaner := &recordingInstanceCleaner{}
+	reaper, store := newStuckTerminateReaper(t, cleaner)
+	const id = "i-wedged-owed"
+	reaper.m.InsertIfAbsent(withVolumes(&VM{
+		ID:             id,
+		Status:         StateShuttingDown,
+		ShuttingDownAt: time.Now().Add(-(stuckTerminateTimeout + time.Minute)),
+	}, "vol-root", "vol-data"))
+
+	_, err := reaper.Sweep(context.Background())
+	require.NoError(t, err)
+
+	require.Len(t, cleaner.owedAtDelete, 1)
+	assert.Equal(t, []string{"vol-root", "vol-data"}, cleaner.owedAtDelete[0])
+	assert.Equal(t, []string{"vol-root", "vol-data"}, store.terminated[id].UnpublishOwed,
+		"the terminated record carries it, so a retried teardown still records the obligation")
+}
+
+func TestReconcileVanishedQEMU_OwesAnUnpublishForEveryVolume(t *testing.T) {
+	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
+	store := newFakeStateStore()
+	m := NewManager()
+	m.SetDeps(Deps{NodeID: "test-node", StateStore: store})
+	instance := withVolumes(&VM{ID: "i-vanished-owed", Status: StateShuttingDown}, "vol-root")
+	m.InsertIfAbsent(instance)
+
+	require.NoError(t, m.reconcileVanishedQEMU(instance))
+
+	assert.Equal(t, []string{"vol-root"}, store.terminated["i-vanished-owed"].UnpublishOwed)
+}

spinifex/runtime/compute/vm/vm.go

@@ -280,6 +280,11 @@ type VM struct {
 	// pending/failed and purges the record once all dependents are done.
 	Teardown map[string]string `json:"teardown,omitempty"`
 
+	// UnpublishOwed lists the volumes the last unmount on LastNode left
+	// exported, so a terminate that clears their attachments records an
+	// unpublish obligation for each. Every unmount replaces it.
+	UnpublishOwed []string `json:"unpublish_owed,omitempty"`
+
 	// TerminatedAt is when the instance entered StateTerminated. The GC backstop
 	// keeps a completed terminated record describable until this is older than
 	// the visibility window, after which it may reclaim it early; the bucket's

spinifex/runtime/daemon/daemon.go

@@ -160,7 +160,11 @@ type Daemon struct {
 	imageService      *ec2image.ImageServiceImpl
 	volumeService     *ec2volume.VolumeServiceImpl
 	// ebsProvider is the sole EBS backend, set once during startup.
-	ebsProvider           ebsprovider.EBSProvider
+	ebsProvider ebsprovider.EBSProvider
+	// mounter is this node's one volume mounter, shared by the VM manager and
+	// the unpublish obligation reaper so both take the same volume locks.
+	mounter               *volumeMounterAdapter
+	mounterOnce           sync.Once
 	accountService        *ec2account.AccountSettingsServiceImpl
 	snapshotService       *ec2snapshot.SnapshotServiceImpl
 	tagsService           *ec2tags.TagsServiceImpl
@@ -2042,6 +2046,9 @@ func (d *Daemon) startCluster() error {
 	if err := d.configureEBSProvider(); err != nil {
 		return fmt.Errorf("configure EBS provider: %w", err)
 	}
+	if err := d.configureUnpublishObligations(); err != nil {
+		return fmt.Errorf("configure EBS unpublish obligations: %w", err)
+	}
 	tagsKV, err := initServiceWithRetry("tags KV bucket", func() (jetstream.KeyValue, error) {
 		tagsJS, jsErr := jetstream.New(d.natsConn)
 		if jsErr != nil {
@@ -2584,6 +2591,8 @@ func (d *Daemon) startCluster() error {
 		if d.volumeService != nil {
 			reapers = append(reapers, d.volumeService.NewVolumeLeakReaper(d.leakedVolumeInstances))
 			reapers = append(reapers, d.volumeService.NewVolumeReclaimReaper())
+			reapers = append(reapers, d.volumeService.NewUnpublishObligationReaper(d.node, d.volumeMounter()))
+			reapers = append(reapers, d.volumeService.NewUnpublishObligationCloser(d.nodeDeparted))
 		}
 		if d.eksService != nil {
 			reapers = append(reapers, d.eksService.NewBillableReaper(d.nodeRunningVMs))

spinifex/runtime/daemon/daemon_handlers_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	ec2account "github.com/mulgadc/spinifex/spinifex/domains/ec2/account"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ebsobligation "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/obligation"
 	ec2eigw "github.com/mulgadc/spinifex/spinifex/domains/ec2/eigw"
 	ec2eip "github.com/mulgadc/spinifex/spinifex/domains/ec2/eip"
 	ec2igw "github.com/mulgadc/spinifex/spinifex/domains/ec2/igw"
@@ -109,6 +110,17 @@ func wireTestEBSProvider(daemon *Daemon, store objectstore.ObjectStore) {
 	daemon.volumeService.SetEBSProvider(provider)
 }
 
+// wireTestUnpublishObligations gives the daemon's volume service an obligation
+// bucket on a JetStream of its own, so no other test's obligations are seen.
+func wireTestUnpublishObligations(t *testing.T, daemon *Daemon) *ebsobligation.Store {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	store := ebsobligation.NewStore(js)
+	require.NoError(t, store.Open(t.Context()))
+	daemon.volumeService.SetUnpublishObligations(store, nil, daemon.clusterMembers)
+	return store
+}
+
 // seedVolumeDocument registers a volume the only way the control plane knows
 // one: as an ebsmetadata document. Tests that need a volume to exist seed it
 // here rather than writing storage state the control plane never reads.

spinifex/runtime/daemon/daemon_test.go

@@ -1845,6 +1845,7 @@ func TestTerminatedTeardownReaper_SelfHealsFailedVolumeDetach(t *testing.T) {
 func TestStuckTerminateReaper_DetachesNonDoTVolumeWithoutUnmount(t *testing.T) {
 	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
 	daemon, store := createFullTestDaemonWithStore(t, sharedNATSURL)
+	obligations := wireTestUnpublishObligations(t, daemon)
 
 	const instanceID = "i-wedged-terminate"
 	volumeID := "vol-data-wedged"
@@ -1884,6 +1885,15 @@ func TestStuckTerminateReaper_DetachesNonDoTVolumeWithoutUnmount(t *testing.T) {
 	assert.Equal(t, "available", meta.State,
 		"the force-complete path must still detach a non-Boot DeleteOnTermination=false volume")
 	assert.Empty(t, meta.AttachedInstance)
+
+	// The unmount never ran, so the export may still be up on this node: the
+	// cleared attachment leaves exactly one obligation naming it.
+	open, err := obligations.List(context.Background())
+	require.NoError(t, err)
+	require.Len(t, open, 1)
+	assert.Equal(t, volumeID, open[0].VolumeID)
+	assert.Equal(t, daemon.node, open[0].NodeID)
+	assert.Equal(t, instanceID, open[0].InstanceID)
 }
 
 // TestHandleEC2Events_AttachVolume tests the attach-volume handler in handleEC2Events.

spinifex/runtime/daemon/instance_records.go

@@ -215,3 +215,22 @@ func loadRecord[T any](store *kvstore.Store[T], key string) (*T, error) {
 	}
 	return record, nil
 }
+
+// InstanceLastNode names the node that last ran instanceID, from its live
+// record or, failing that, its terminated one. found is false when neither
+// exists. It is the volume domain's one view of instance state.
+func (m *JetStreamManager) InstanceLastNode(instanceID string) (node string, found bool, err error) {
+	record, err := m.LoadInstanceRecord(instanceID)
+	if err != nil {
+		return "", false, err
+	}
+	if record == nil {
+		if record, err = m.LoadTerminatedInstanceRecord(instanceID); err != nil {
+			return "", false, err
+		}
+	}
+	if record == nil {
+		return "", false, nil
+	}
+	return record.Status.LastNode, true, nil
+}

spinifex/runtime/daemon/unpublish_obligations.go

@@ -0,0 +1,81 @@
+package daemon
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+
+	ebsobligation "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/obligation"
+	ec2volume "github.com/mulgadc/spinifex/spinifex/domains/ec2/volume"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+var _ ec2volume.InstanceNodeLocator = (*JetStreamManager)(nil)
+
+// volumeMounter returns this node's one mounter. The VM manager and the
+// unpublish obligation reaper must share it, because its volume locks are what
+// keep the reaper's check and unpublish clear of a publish on this node.
+func (d *Daemon) volumeMounter() *volumeMounterAdapter {
+	d.mounterOnce.Do(func() {
+		d.mounter = newVolumeMounterAdapter(d.ebsProvider, d.node, d.volumeService)
+	})
+	return d.mounter
+}
+
+// configureUnpublishObligations opens the obligation bucket and gives the
+// volume service what a cleared attachment needs to record one. A node that
+// cannot open it fails startup rather than its first force detach.
+func (d *Daemon) configureUnpublishObligations() error {
+	store, err := initServiceWithRetry("EBS unpublish obligation bucket", func() (*ebsobligation.Store, error) {
+		js, err := jetstream.New(d.natsConn)
+		if err != nil {
+			return nil, fmt.Errorf("jetstream handle: %w", err)
+		}
+		store := ebsobligation.NewStore(js)
+		if err := store.Open(d.ctx); err != nil {
+			return nil, err
+		}
+		return store, nil
+	})
+	if err != nil {
+		return err
+	}
+	var locator ec2volume.InstanceNodeLocator
+	if d.jsManager != nil {
+		locator = d.jsManager
+	}
+	d.volumeService.SetUnpublishObligations(store, locator, d.clusterMembers)
+	return nil
+}
+
+// clusterMembers names every node in this node's cluster configuration.
+func (d *Daemon) clusterMembers() []string {
+	if d.clusterConfig == nil {
+		return nil
+	}
+	return slices.Sorted(maps.Keys(d.clusterConfig.Nodes))
+}
+
+// nodeDeparted reports whether node has been removed from the cluster. A node
+// that is only down, stale or partitioned is still a member and keeps its
+// obligations. One this node's configuration lacks counts as removed only if
+// its heartbeat is gone or older than this node's epoch: a heartbeat at this
+// epoch or later means this node's view may be the stale one.
+func (d *Daemon) nodeDeparted(_ context.Context, node string) (bool, error) {
+	if d.clusterConfig == nil || d.jsManager == nil {
+		return false, nil
+	}
+	if _, member := d.clusterConfig.Nodes[node]; member {
+		return false, nil
+	}
+	heartbeat, err := d.jsManager.ReadHeartbeat(node)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return true, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("read heartbeat of %s: %w", node, err)
+	}
+	return heartbeat.Epoch < d.clusterConfig.Epoch, nil
+}

spinifex/runtime/daemon/unpublish_obligations_test.go

@@ -0,0 +1,268 @@
+package daemon
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	ebsmetadata "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	ebsprovider "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"
+)
+
+// --- nodeDeparted ---
+
+func newDepartedDaemon(t *testing.T) (*Daemon, *failingKV) {
+	t.Helper()
+	m, f := newFaultInjectedJSM(t)
+	return &Daemon{
+		jsManager:     m,
+		clusterConfig: &config.ClusterConfig{Epoch: 5, Nodes: map[string]config.Config{"node-a": {}, "node-b": {}}},
+	}, f
+}
+
+func TestNodeDeparted_AMemberIsNeverDeparted(t *testing.T) {
+	d, f := newDepartedDaemon(t)
+	f.setFail("Get", true)
+
+	departed, err := d.nodeDeparted(t.Context(), "node-a")
+	require.NoError(t, err)
+	assert.False(t, departed, "a member that is down or unreachable keeps its obligations")
+}
+
+func TestNodeDeparted_ANonMemberDepartsOnlyWithNoCurrentHeartbeat(t *testing.T) {
+	for name, tc := range map[string]struct {
+		heartbeat *Heartbeat
+		departed  bool
+	}{
+		"no heartbeat":                {departed: true},
+		"heartbeat at an older epoch": {heartbeat: &Heartbeat{Node: "node-gone", Epoch: 4}, departed: true},
+		"heartbeat at this epoch":     {heartbeat: &Heartbeat{Node: "node-gone", Epoch: 5}},
+		"heartbeat at a later epoch":  {heartbeat: &Heartbeat{Node: "node-gone", Epoch: 6}},
+	} {
+		t.Run(name, func(t *testing.T) {
+			d, _ := newDepartedDaemon(t)
+			if tc.heartbeat != nil {
+				require.NoError(t, d.jsManager.WriteHeartbeat(tc.heartbeat))
+			}
+
+			departed, err := d.nodeDeparted(t.Context(), "node-gone")
+			require.NoError(t, err)
+			assert.Equal(t, tc.departed, departed)
+		})
+	}
+}
+
+func TestNodeDeparted_AnUnreadableHeartbeatIsAnError(t *testing.T) {
+	d, f := newDepartedDaemon(t)
+	f.setFail("Get", true)
+
+	departed, err := d.nodeDeparted(t.Context(), "node-gone")
+	require.Error(t, err)
+	assert.False(t, departed)
+}
+
+func TestNodeDeparted_NoClusterViewDepartsNothing(t *testing.T) {
+	departed, err := (&Daemon{}).nodeDeparted(t.Context(), "node-gone")
+	require.NoError(t, err)
+	assert.False(t, departed)
+}
+
+func TestClusterMembers(t *testing.T) {
+	d, _ := newDepartedDaemon(t)
+	assert.Equal(t, []string{"node-a", "node-b"}, d.clusterMembers())
+	assert.Nil(t, (&Daemon{}).clusterMembers())
+}
+
+// --- InstanceLastNode ---
+
+func newLastNodeManager(t *testing.T) *JetStreamManager {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	m := &JetStreamManager{js: js}
+	m.setInstanceStateBucket(kvstore.NewBucket(js, kvstore.Config{Name: "last-node-live", History: 1}))
+	m.termRecords = kvstore.On[vm.InstanceRecord](kvstore.NewBucket(js, kvstore.Config{Name: "last-node-term", History: 1}))
+	return m
+}
+
+func TestInstanceLastNode_ReadsTheLiveThenTheTerminatedRecord(t *testing.T) {
+	m := newLastNodeManager(t)
+	require.NoError(t, m.WriteInstanceRecord("i-live", (&vm.VM{ID: "i-live", LastNode: "node-a"}).Record()))
+	require.NoError(t, m.WriteTerminatedInstanceRecord("i-term", (&vm.VM{ID: "i-term", LastNode: "node-b"}).Record()))
+
+	node, found, err := m.InstanceLastNode("i-live")
+	require.NoError(t, err)
+	assert.True(t, found)
+	assert.Equal(t, "node-a", node)
+
+	node, found, err = m.InstanceLastNode("i-term")
+	require.NoError(t, err)
+	assert.True(t, found)
+	assert.Equal(t, "node-b", node)
+
+	_, found, err = m.InstanceLastNode("i-none")
+	require.NoError(t, err)
+	assert.False(t, found)
+}
+
+// A record that cannot be read is not a missing one: treating it as missing
+// would name every member rather than fail the request.
+func TestInstanceLastNode_AnUnreadableRecordIsAnError(t *testing.T) {
+	m, f := newFaultInjectedJSM(t)
+	f.setFail("Get", true)
+
+	_, found, err := m.InstanceLastNode("i-live")
+	require.Error(t, err)
+	assert.False(t, found)
+}
+
+// --- The mounter's volume hold ---
+
+// holdTestProvider records each call and can hold a publish open until told.
+type holdTestProvider struct {
+	mu           sync.Mutex
+	unpublishes  []ebsprovider.UnpublishVolumeRequest
+	unpublishErr error
+	publishGate  chan struct{}
+	publishing   chan struct{}
+}
+
+var _ mounterProvider = (*holdTestProvider)(nil)
+
+func (p *holdTestProvider) PublishVolume(_ context.Context, req ebsprovider.PublishVolumeRequest) (*ebsprovider.PublishedVolume, error) {
+	if p.publishGate != nil {
+		close(p.publishing)
+		<-p.publishGate
+	}
+	return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: ebsprovider.NewNBDExport("nbd://x")}, nil
+}
+
+func (p *holdTestProvider) UnpublishVolume(_ context.Context, req ebsprovider.UnpublishVolumeRequest) (*ebsprovider.UnpublishedVolume, error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.unpublishes = append(p.unpublishes, req)
+	if p.unpublishErr != nil {
+		return nil, p.unpublishErr
+	}
+	return &ebsprovider.UnpublishedVolume{TornDown: true}, nil
+}
+
+func (p *holdTestProvider) calls() []ebsprovider.UnpublishVolumeRequest {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return append([]ebsprovider.UnpublishVolumeRequest(nil), p.unpublishes...)
+}
+
+// A hot attach publishes before it writes in-use. The publish stamp is what
+// lets the reaper tell that publish apart from the one an obligation undoes.
+func TestVolumeMounterHold_PublishedSinceTracksThisProcessesPublishes(t *testing.T) {
+	provider := &holdTestProvider{}
+	adapter := newVolumeMounterAdapter(provider, "node-hold", nil)
+	before := time.Now().Add(-time.Second)
+
+	require.NoError(t, adapter.MountOne(t.Context(), testAccountID, &viperblocklegacyv1.EBSRequest{Name: "vol-hold"}))
+
+	hold, ok := adapter.HoldVolume("vol-hold")
+	require.True(t, ok)
+	assert.True(t, hold.PublishedSince(before), "a publish after the obligation's attachment is newer")
+	assert.False(t, hold.PublishedSince(time.Now().Add(time.Second)), "a publish before it is not")
+	hold.Release()
+
+	require.NoError(t, adapter.UnmountOne(t.Context(), testAccountID, viperblocklegacyv1.EBSRequest{Name: "vol-hold"}))
+	hold, ok = adapter.HoldVolume("vol-hold")
+	require.True(t, ok)
+	defer hold.Release()
+	assert.False(t, hold.PublishedSince(before), "an unpublish clears the stamp")
+}
+
+func TestVolumeMounterHold_RefusedWhileAPublishIsInFlight(t *testing.T) {
+	provider := &holdTestProvider{publishGate: make(chan struct{}), publishing: make(chan struct{})}
+	adapter := newVolumeMounterAdapter(provider, "node-hold", nil)
+	done := make(chan error, 1)
+	go func() {
+		done <- adapter.MountOne(context.Background(), testAccountID, &viperblocklegacyv1.EBSRequest{Name: "vol-busy"})
+	}()
+	<-provider.publishing
+
+	_, ok := adapter.HoldVolume("vol-busy")
+	assert.False(t, ok, "the reaper never waits behind a publish")
+
+	close(provider.publishGate)
+	require.NoError(t, <-done)
+	hold, ok := adapter.HoldVolume("vol-busy")
+	require.True(t, ok, "the lock is free once the publish returns")
+	hold.Release()
+	hold.Release()
+	hold, ok = adapter.HoldVolume("vol-busy")
+	require.True(t, ok, "a second release does not unlock someone else's hold")
+	hold.Release()
+}
+
+func TestVolumeMounterHold_UnpublishSealsOrDropsAsAsked(t *testing.T) {
+	provider := &holdTestProvider{}
+	adapter := newVolumeMounterAdapter(provider, "node-hold", nil)
+
+	hold, ok := adapter.HoldVolume("vol-settle")
+	require.True(t, ok)
+	require.NoError(t, hold.Unpublish(t.Context(), testAccountID, "handle-1", false))
+	require.NoError(t, hold.Unpublish(t.Context(), testAccountID, "handle-1", true))
+	hold.Release()
+
+	calls := provider.calls()
+	require.Len(t, calls, 2)
+	for _, call := range calls {
+		assert.Equal(t, "vol-settle", call.VolumeID)
+		assert.Equal(t, "handle-1", call.Handle, "the obligation's handle names the publish it undoes")
+		assert.Equal(t, "node-hold", call.NodeID)
+		assert.Equal(t, ebsprovider.NoSealSchemaVersion, call.SchemaVersion)
+	}
+	assert.False(t, calls[0].NoSeal, "a surviving volume is sealed so its writes are kept")
+	assert.True(t, calls[1].NoSeal)
+}
+
+func TestVolumeMounterHold_NotFoundCountsAsDone(t *testing.T) {
+	for _, noSeal := range []bool{false, true} {
+		provider := &holdTestProvider{unpublishErr: ebsprovider.ErrNotFound}
+		adapter := newVolumeMounterAdapter(provider, "node-hold", nil)
+		hold, ok := adapter.HoldVolume("vol-gone")
+		require.True(t, ok)
+		assert.NoError(t, hold.Unpublish(t.Context(), testAccountID, "", noSeal))
+		hold.Release()
+	}
+}
+
+// --- DeleteVolumes names this node for what the last unmount left ---
+
+func TestInstanceCleaner_DeleteVolumes_RecordsOnlyTheOwedVolumes(t *testing.T) {
+	daemon, store := createFullTestDaemonWithStore(t, sharedNATSURL)
+	obligations := wireTestUnpublishObligations(t, daemon)
+	const instanceID = "i-owed-cleaner"
+	for _, volumeID := range []string{"vol-owed-1", "vol-sealed-1"} {
+		seedVolumeConfig(t, daemon, store, ebsmetadata.Volume{
+			VolumeID: volumeID, TenantID: testAccountID, CapacityGiB: 1,
+			State: "in-use", AttachedInstance: instanceID,
+		})
+	}
+	instance := &vm.VM{ID: instanceID, AccountID: testAccountID, UnpublishOwed: []string{"vol-owed-1"}}
+	instance.EBSRequests.Requests = []viperblocklegacyv1.EBSRequest{{Name: "vol-owed-1"}, {Name: "vol-sealed-1"}}
+
+	require.NoError(t, newInstanceCleanerAdapter(daemon).DeleteVolumes(instance))
+
+	open, err := obligations.List(t.Context())
+	require.NoError(t, err)
+	require.Len(t, open, 1, "a volume whose unmount completed owes nothing")
+	assert.Equal(t, "vol-owed-1", open[0].VolumeID)
+	assert.Equal(t, daemon.node, open[0].NodeID)
+	for _, volumeID := range []string{"vol-owed-1", "vol-sealed-1"} {
+		meta, err := daemon.volumeService.GetVolumeMetadata(testAccountID, volumeID)
+		require.NoError(t, err)
+		assert.Empty(t, meta.AttachedInstance)
+	}
+}

spinifex/runtime/daemon/vm_adapters.go

@@ -8,11 +8,15 @@ import (
 	"fmt"
 	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"log/slog"
+	"slices"
+	"sync"
 	"time"
 
 	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
 	ec2placementgroup "github.com/mulgadc/spinifex/spinifex/domains/ec2/placementgroup"
+	ec2volume "github.com/mulgadc/spinifex/spinifex/domains/ec2/volume"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/external/dhcp"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
@@ -102,6 +106,77 @@ type volumeMounterAdapter struct {
 	provider mounterProvider
 	// abandonTimeout bounds each no-seal unpublish Abandon sends.
 	abandonTimeout time.Duration
+	// volumes serialises every publish and unpublish of a volume on this
+	// node with the unpublish obligation reaper.
+	volumes volumeLocks
+}
+
+// volumeLocks is a lock per volume, and the time this process last published
+// each volume it has not unpublished since.
+type volumeLocks struct {
+	mu    sync.Mutex
+	locks map[string]*volumeLock
+}
+
+type volumeLock struct {
+	mu          sync.Mutex
+	refs        int
+	publishedAt time.Time
+}
+
+// acquire returns volumeID's lock entry, locked; wait false gives up rather
+// than block on a publish or unpublish in flight.
+func (l *volumeLocks) acquire(volumeID string, wait bool) (*volumeLock, bool) {
+	l.mu.Lock()
+	if l.locks == nil {
+		l.locks = make(map[string]*volumeLock)
+	}
+	entry, ok := l.locks[volumeID]
+	if !ok {
+		entry = &volumeLock{}
+		l.locks[volumeID] = entry
+	}
+	entry.refs++
+	l.mu.Unlock()
+
+	if wait {
+		entry.mu.Lock()
+		return entry, true
+	}
+	if entry.mu.TryLock() {
+		return entry, true
+	}
+	l.release(volumeID, entry, false)
+	return nil, false
+}
+
+// release drops a reference, unlocking the entry when locked is true. An entry
+// nobody holds and that records no publish is forgotten.
+func (l *volumeLocks) release(volumeID string, entry *volumeLock, locked bool) {
+	if locked {
+		entry.mu.Unlock()
+	}
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	entry.refs--
+	if entry.refs == 0 && entry.publishedAt.IsZero() {
+		delete(l.locks, volumeID)
+	}
+}
+
+// withVolume runs fn holding volumeID's lock.
+func (a *volumeMounterAdapter) withVolume(volumeID string, fn func(*volumeLock) error) error {
+	entry, _ := a.volumes.acquire(volumeID, true)
+	defer a.volumes.release(volumeID, entry, true)
+	return fn(entry)
+}
+
+// recordPublished stamps a publish that may have exported the volume, and
+// clears the stamp once an unpublish has torn it down. The caller holds the lock.
+func (a *volumeMounterAdapter) recordPublished(entry *volumeLock, at time.Time) {
+	a.volumes.mu.Lock()
+	entry.publishedAt = at
+	a.volumes.mu.Unlock()
 }
 
 // mounterProvider is the part of the EBS contract the mounter drives.
@@ -127,11 +202,18 @@ func newVolumeMounterAdapter(provider mounterProvider, node string, volState vm.
 // the export on req. A failure worth retrying is vm.ErrMountRetryable; one
 // that leaves unknown whether the provider exported it is vm.ErrMountAmbiguous.
 func (a *volumeMounterAdapter) publish(ctx context.Context, accountID string, req *viperblocklegacyv1.EBSRequest) error {
+	return a.withVolume(req.Name, func(entry *volumeLock) error {
+		return a.publishLocked(ctx, accountID, req, entry)
+	})
+}
+
+func (a *volumeMounterAdapter) publishLocked(ctx context.Context, accountID string, req *viperblocklegacyv1.EBSRequest, entry *volumeLock) error {
 	// A caller already gone sends nothing, so there is no export to tear down;
 	// only a cancellation after the request went out leaves the outcome unknown.
 	if err := ctx.Err(); err != nil {
 		return fmt.Errorf("publish volume %s: %w", req.Name, err)
 	}
+	started := time.Now()
 	published, err := a.provider.PublishVolume(ebsprovider.WithAccountID(ctx, accountID), ebsprovider.PublishVolumeRequest{
 		Versioned: ebsprovider.NewLatestVersioned(),
 		VolumeID:  req.Name,
@@ -142,10 +224,12 @@ func (a *volumeMounterAdapter) publish(ctx context.Context, accountID string, re
 		case publishRetryable(err):
 			return fmt.Errorf("publish volume %s: %w: %w", req.Name, err, vm.ErrMountRetryable)
 		case publishOutcomeUnknown(err):
+			a.recordPublished(entry, started)
 			return fmt.Errorf("publish volume %s: %w: %w", req.Name, err, vm.ErrMountAmbiguous)
 		}
 		return fmt.Errorf("publish volume %s: %w", req.Name, err)
 	}
+	a.recordPublished(entry, started)
 	applyExport(req, published.Export)
 	slog.InfoContext(ctx, "Published volume", "volume", req.Name, "node", a.node, "export_kind", string(published.Export.Kind))
 	return nil
@@ -199,25 +283,77 @@ func applyExport(req *viperblocklegacyv1.EBSRequest, export ebsprovider.VolumeEx
 // UnpublishVolume, which seals first, and returns nil only once its block map
 // is sealed or the provider holds no export of it.
 func (a *volumeMounterAdapter) unpublish(ctx context.Context, accountID string, req viperblocklegacyv1.EBSRequest) error {
+	return a.withVolume(req.Name, func(entry *volumeLock) error {
+		return a.unpublishLocked(ctx, accountID, req.Name, "", entry)
+	})
+}
+
+func (a *volumeMounterAdapter) unpublishLocked(ctx context.Context, accountID, volumeID, handle string, entry *volumeLock) error {
 	// The seal takes the provider's data-movement budget, not what is left of
 	// the caller's; the caller's trace is kept.
 	sealCtx := ebsprovider.WithAccountID(context.WithoutCancel(ctx), accountID)
 	_, err := a.provider.UnpublishVolume(sealCtx, ebsprovider.UnpublishVolumeRequest{
 		Versioned: ebsprovider.NewLatestVersioned(),
-		VolumeID:  req.Name,
+		VolumeID:  volumeID,
+		Handle:    handle,
 		NodeID:    a.node,
 	})
 	switch {
 	case err == nil:
+		a.recordPublished(entry, time.Time{})
 		return nil
 	case errors.Is(err, ebsprovider.ErrNotFound):
-		slog.InfoContext(ctx, "Unpublish found no volume, counting it as done", "volume", req.Name, "node", a.node)
+		a.recordPublished(entry, time.Time{})
+		slog.InfoContext(ctx, "Unpublish found no volume, counting it as done", "volume", volumeID, "node", a.node)
 		return nil
 	default:
-		return fmt.Errorf("unpublish volume %s: %w", req.Name, err)
+		return fmt.Errorf("unpublish volume %s: %w", volumeID, err)
 	}
 }
 
+var _ ec2volume.ObligationUnpublisher = (*volumeMounterAdapter)(nil)
+
+// HoldVolume takes volumeID's lock for the unpublish obligation reaper without
+// waiting, so a sweep never queues behind a seal or a wedged publish.
+func (a *volumeMounterAdapter) HoldVolume(volumeID string) (ec2volume.VolumeHold, bool) {
+	entry, ok := a.volumes.acquire(volumeID, false)
+	if !ok {
+		return nil, false
+	}
+	return &mounterVolumeHold{a: a, volumeID: volumeID, entry: entry}, true
+}
+
+// mounterVolumeHold is one volume's lock, held by the obligation reaper.
+type mounterVolumeHold struct {
+	a        *volumeMounterAdapter
+	volumeID string
+	entry    *volumeLock
+	released bool
+}
+
+var _ ec2volume.VolumeHold = (*mounterVolumeHold)(nil)
+
+func (h *mounterVolumeHold) PublishedSince(t time.Time) bool {
+	h.a.volumes.mu.Lock()
+	defer h.a.volumes.mu.Unlock()
+	return !h.entry.publishedAt.IsZero() && h.entry.publishedAt.After(t)
+}
+
+func (h *mounterVolumeHold) Unpublish(ctx context.Context, accountID, handle string, noSeal bool) error {
+	if !noSeal {
+		return h.a.unpublishLocked(ctx, accountID, h.volumeID, handle, h.entry)
+	}
+	return h.a.abandonLocked(ctx, accountID, "", h.volumeID, handle, "an unpublish obligation for a volume being deleted", h.entry)
+}
+
+func (h *mounterVolumeHold) Release() {
+	if h.released {
+		return
+	}
+	h.released = true
+	h.a.volumes.release(h.volumeID, h.entry, true)
+}
+
 // ebsRequestWithTrace sends an ebs.* NATS request, opening a client span under
 // ctx and injecting it into the message headers so viperblockd's consumer span
 // (natsmsg.StartConsumerSpan) joins this trace instead of rooting a new one.
@@ -302,7 +438,7 @@ func (a *volumeMounterAdapter) Unmount(ctx context.Context, instance *vm.VM) err
 		if err := a.unpublish(ctx, instance.AccountID, ebsRequest); err != nil {
 			slog.Error("Volume seal failed, leaving volume non-available",
 				"instance", instance.ID, "volume", ebsRequest.Name, "err", err)
-			sealErrs = append(sealErrs, fmt.Errorf("seal %s: %w", ebsRequest.Name, err))
+			sealErrs = append(sealErrs, &vm.VolumeUnpublishError{VolumeID: ebsRequest.Name, Err: err})
 			continue
 		}
 		slog.Info("Unmounted volume", "instance", instance.ID, "volume", ebsRequest.Name)
@@ -355,23 +491,32 @@ func (a *volumeMounterAdapter) Abandon(ctx context.Context, instance *vm.VM, rea
 // abandonOne drops this node's export of volumeID unsealed. A volume the
 // provider does not know is one this node no longer holds.
 func (a *volumeMounterAdapter) abandonOne(ctx context.Context, instance *vm.VM, volumeID, reason string) error {
-	abandonCtx, cancel := context.WithTimeout(ebsprovider.WithAccountID(ctx, instance.AccountID), a.abandonTimeout)
+	return a.withVolume(volumeID, func(entry *volumeLock) error {
+		return a.abandonLocked(ctx, instance.AccountID, instance.ID, volumeID, "", reason, entry)
+	})
+}
+
+func (a *volumeMounterAdapter) abandonLocked(ctx context.Context, accountID, instanceID, volumeID, handle, reason string, entry *volumeLock) error {
+	abandonCtx, cancel := context.WithTimeout(ebsprovider.WithAccountID(ctx, accountID), a.abandonTimeout)
 	defer cancel()
 	unpublished, err := a.provider.UnpublishVolume(abandonCtx, ebsprovider.UnpublishVolumeRequest{
 		Versioned: ebsprovider.NewLatestVersioned(),
 		VolumeID:  volumeID,
+		Handle:    handle,
 		NodeID:    a.node,
 		NoSeal:    true,
 	})
 	switch {
 	case errors.Is(err, ebsprovider.ErrNotFound):
+		a.recordPublished(entry, time.Time{})
 		return nil
 	case err != nil:
 		return fmt.Errorf("abandon %s (%s): %w", volumeID, reason, err)
 	}
+	a.recordPublished(entry, time.Time{})
 	if unpublished.TornDown {
-		slog.WarnContext(ctx, "Gave up a volume export to the node that owns the instance now",
-			"instance", instance.ID, "volume", volumeID, "reason", reason)
+		slog.WarnContext(ctx, "Gave up a volume export unsealed",
+			"instance", instanceID, "volume", volumeID, "reason", reason, "node", a.node)
 	}
 	return nil
 }
@@ -677,7 +822,7 @@ func (d *Daemon) buildVMManagerDeps() vm.Deps {
 	return vm.Deps{
 		NodeID:             d.node,
 		StateStore:         d.stateStore,
-		VolumeMounter:      newVolumeMounterAdapter(d.ebsProvider, d.node, d.volumeService),
+		VolumeMounter:      d.volumeMounter(),
 		NetworkPlumber:     d.networkPlumber,
 		InstanceTypes:      newInstanceTypeResolverAdapter(d.resourceMgr),
 		Resources:          newResourceControllerAdapter(d.resourceMgr),
@@ -721,6 +866,21 @@ func newInstanceCleanerAdapter(d *Daemon) *instanceCleanerAdapter {
 // flagged DeleteOnTermination via the volume service.
 // Errors are logged per volume; partial failure is tolerated.
 func (a *instanceCleanerAdapter) DeleteVolumes(instance *vm.VM) error {
+	// Volumes the last unmount here left exported: clearing their attachments
+	// owes this node an unpublish, recorded before the clear.
+	var leftPublished []string
+	if a.d.vmMgr != nil {
+		a.d.vmMgr.Inspect(instance, func(v *vm.VM) { leftPublished = slices.Clone(v.UnpublishOwed) })
+	} else {
+		leftPublished = slices.Clone(instance.UnpublishOwed)
+	}
+	owedFor := func(volumeID string) ec2instance.OwedUnpublish {
+		if !slices.Contains(leftPublished, volumeID) {
+			return ec2instance.OwedUnpublish{}
+		}
+		return ec2instance.OwedUnpublish{NodeID: a.d.node, InstanceID: instance.ID}
+	}
+
 	instance.EBSRequests.Mu.Lock()
 	defer instance.EBSRequests.Mu.Unlock()
 
@@ -764,7 +924,7 @@ func (a *instanceCleanerAdapter) DeleteVolumes(instance *vm.VM) error {
 					"name", ebsRequest.Name, "id", instance.ID)
 				continue
 			}
-			if err := a.d.volumeService.DetachVolumeOnTerminate(context.Background(), ebsRequest.Name, instance.AccountID); err != nil && !awserrors.IsNotFound(err) && !objectstore.IsNoSuchKeyError(err) {
+			if err := a.d.volumeService.DetachVolumeOnTerminate(context.Background(), ebsRequest.Name, instance.AccountID, owedFor(ebsRequest.Name)); err != nil && !awserrors.IsNotFound(err) && !objectstore.IsNoSuchKeyError(err) {
 				slog.Error("Failed to detach volume on termination",
 					"name", ebsRequest.Name, "id", instance.ID, "err", err)
 				firstErr = cmp.Or(firstErr, err)
@@ -787,7 +947,7 @@ func (a *instanceCleanerAdapter) DeleteVolumes(instance *vm.VM) error {
 		// (above, via terminateCleanup) deliberately never clears a Boot
 		// volume's attachment, so without this the root volume still hits
 		// DeleteVolume's in-use guard here.
-		if err := a.d.volumeService.DeleteVolumeOnTerminate(context.Background(), ebsRequest.Name, instance.AccountID); err != nil && !awserrors.IsNotFound(err) && !objectstore.IsNoSuchKeyError(err) {
+		if err := a.d.volumeService.DeleteVolumeOnTerminate(context.Background(), ebsRequest.Name, instance.AccountID, owedFor(ebsRequest.Name)); err != nil && !awserrors.IsNotFound(err) && !objectstore.IsNoSuchKeyError(err) {
 			slog.Error("Failed to delete volume on termination",
 				"name", ebsRequest.Name, "id", instance.ID, "err", err)
 			firstErr = cmp.Or(firstErr, err)