fix: fail DRAIN when a volume seal failed (#966)

shutdownAndUnmount logged the volume unmount error and returned nothing, so
StopAll returned nil and the DRAIN phase ACKed success with vms_stopped set.
The coordinator then advanced to STORAGE and stopped viperblock while the seal
was still in flight, tearing the VBState envelope and leaving the guest unable
to mount its root volume on the next boot. The deploy reported success.

shutdownAndUnmount now returns ErrVolumeSealFailed when the unmount did not
seal; QMP, PID-file, fw_cfg and telemetry cleanup stay best-effort. stopCleanup
and stopOne carry it without aborting the rest of the teardown, since QEMU is
already down by then, and StopAll aggregates the per-VM failures so
handleShutdownDrain ACKs with Error set and never reports vms_stopped.

volumeMounterAdapter.Unmount was the second swallow site: it logged each failed
seal and returned nil unconditionally. It now attempts every volume and returns
the aggregate, which is what the VolumeMounter contract already described.

Terminate still tolerates a failed seal — the volumes are deleted next, so an
unsealed block map loses nothing a caller can ask for.

This does not fix the seal itself. It converts silent data loss into a refused
upgrade that an operator can act on.

spinifex/daemon/daemon_test.go

@@ -3420,7 +3420,8 @@ var _ vm.VolumeStateUpdater = (*recordingVolState)(nil)
 // teardown-path durability gate: a per-volume seal failure during the bulk
 // Unmount (stop/terminate) must NOT transition that volume to available — a
 // reattach on a WAL-less node would find no checkpoint (bad superblock) — while
-// other volumes still seal and the loop completes (terminate stays idempotent).
+// other volumes still seal and the loop completes. The failure is returned, not
+// swallowed: the caller decides whether losing that seal is survivable.
 func TestVolumeMounterAdapter_Unmount_SealFailureSkipsAvailable(t *testing.T) {
 	daemon := createTestDaemon(t, sharedNATSURL)
 	volState := &recordingVolState{}
@@ -3445,8 +3446,9 @@ func TestVolumeMounterAdapter_Unmount_SealFailureSkipsAvailable(t *testing.T) {
 			Requests: []types.EBSRequest{{Name: "vol-fail"}, {Name: "vol-ok"}},
 		},
 	}
-	require.NoError(t, adapter.Unmount(t.Context(), inst),
-		"bulk Unmount must tolerate a per-volume seal failure so terminate stays idempotent")
+	err = adapter.Unmount(t.Context(), inst)
+	require.Error(t, err, "bulk Unmount must report a per-volume seal failure")
+	assert.Contains(t, err.Error(), "vol-fail", "the error must name the volume that did not seal")
 
 	calls := volState.snapshot()
 	assert.NotContains(t, calls, "vol-fail:available",

spinifex/daemon/shutdown.go

@@ -142,7 +142,11 @@ func (d *Daemon) handleShutdownDrain(msg *nats.Msg) string {
 	// iteration from concurrent terminate handlers.
 	if total > 0 {
 		if err := d.vmMgr.StopAll(); err != nil {
-			slog.Error("Failed to stop instances during DRAIN", "error", err)
+			// A failed volume seal lands here. ACK with the error and do not
+			// report vms_stopped: the coordinator must refuse to advance to
+			// STORAGE rather than stop viperblock over an unsealed block map.
+			slog.Error("Failed to stop instances during DRAIN, refusing to advance",
+				"node", d.node, "error", err)
 			ack := ShutdownACK{
 				Node:  d.node,
 				Phase: "drain",

spinifex/daemon/shutdown_test.go

@@ -11,6 +11,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/mulgadc/spinifex/spinifex/vm"
 	"github.com/nats-io/nats.go"
@@ -665,3 +666,126 @@ func TestHandleShutdownDrain(t *testing.T) {
 		assert.Equal(t, 0, last.Remaining, "final progress should report zero remaining")
 	})
 }
+
+// stubEBSUnmount answers the daemon's ebs.<node>.unmount subject with a fixed
+// response, standing in for viperblockd. Returns the volumes it was asked to
+// unmount once the subscription has been drained.
+func stubEBSUnmount(t *testing.T, d *Daemon, resp types.EBSUnMountResponse) func() []string {
+	t.Helper()
+	var (
+		mu      sync.Mutex
+		volumes []string
+	)
+	sub, err := d.natsConn.Subscribe("ebs."+d.node+".unmount", func(msg *nats.Msg) {
+		var req types.EBSRequest
+		require.NoError(t, json.Unmarshal(msg.Data, &req))
+		mu.Lock()
+		volumes = append(volumes, req.Name)
+		mu.Unlock()
+		reply := resp
+		reply.Volume = req.Name
+		data, marshalErr := json.Marshal(reply)
+		require.NoError(t, marshalErr)
+		require.NoError(t, msg.Respond(data))
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, d.natsConn.Flush())
+
+	return func() []string {
+		mu.Lock()
+		defer mu.Unlock()
+		return append([]string(nil), volumes...)
+	}
+}
+
+// drainRunningInstance is a guest in the state DRAIN finds it in: running, with
+// a boot volume attached, and drain-stopped rather than operator-stopped.
+func drainRunningInstance(id, volume string) *vm.VM {
+	instance := &vm.VM{ID: id, Status: vm.StateRunning, AccountID: "111122223333"}
+	instance.EBSRequests.Requests = []types.EBSRequest{
+		{Name: volume, VolType: types.VolumeTypeGP3, Boot: true},
+	}
+	return instance
+}
+
+// TestHandleShutdownDrain_SealFailure drives DRAIN end to end over NATS against
+// a fault-injected block store whose seal never completes — the stalled blob
+// node that destroyed two guests on upgrade. The phase must fail rather than
+// ACK success, so the coordinator refuses to advance to STORAGE and stop
+// viperblock over an unsealed block map.
+func TestHandleShutdownDrain_SealFailure(t *testing.T) {
+	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
+
+	t.Run("stalled blob node fails the phase", func(t *testing.T) {
+		daemon := createFullTestDaemonWithJetStream(t, sharedJSNATSURL)
+		require.NoError(t, daemon.jsManager.InitClusterStateBucket())
+		t.Cleanup(func() { _ = daemon.jsManager.DeleteShutdownMarker(daemon.node) })
+
+		unmounted := stubEBSUnmount(t, daemon, types.EBSUnMountResponse{
+			Mounted: true,
+			Error:   "seal volume to predastore: transfer stalled: idle timeout",
+		})
+		daemon.vmMgr.Insert(drainRunningInstance("i-drain-seal-fail", "vol-drain-seal-fail"))
+
+		subject := "spinifex.cluster.shutdown.drain.sealfail"
+		sub, err := daemon.natsConn.Subscribe(subject, asMsgHandler(daemon.handleShutdownDrain))
+		require.NoError(t, err)
+		defer sub.Unsubscribe()
+		require.NoError(t, daemon.natsConn.Flush())
+
+		payload, err := json.Marshal(ShutdownRequest{Phase: "drain"})
+		require.NoError(t, err)
+
+		reply, err := daemon.natsConn.Request(subject, payload, 60*time.Second)
+		require.NoError(t, err)
+
+		var ack ShutdownACK
+		require.NoError(t, json.Unmarshal(reply.Data, &ack))
+		assert.Equal(t, "drain", ack.Phase)
+		require.NotEmpty(t, ack.Error,
+			"DRAIN must ACK with an error when a volume seal failed")
+		assert.Contains(t, ack.Error, "vol-drain-seal-fail",
+			"the ACK must name the volume the operator has to fix")
+		assert.Equal(t, []string{"vol-drain-seal-fail"}, unmounted(),
+			"the seal must have been attempted")
+
+		marker, err := daemon.jsManager.ReadShutdownMarker(daemon.node)
+		require.NoError(t, err)
+		assert.False(t, marker,
+			"a failed DRAIN must not record the node as cleanly shut down")
+	})
+
+	t.Run("healthy blob node completes the phase", func(t *testing.T) {
+		daemon := createFullTestDaemonWithJetStream(t, sharedJSNATSURL)
+		require.NoError(t, daemon.jsManager.InitClusterStateBucket())
+		t.Cleanup(func() { _ = daemon.jsManager.DeleteShutdownMarker(daemon.node) })
+
+		unmounted := stubEBSUnmount(t, daemon, types.EBSUnMountResponse{})
+		instance := drainRunningInstance("i-drain-seal-ok", "vol-drain-seal-ok")
+		daemon.vmMgr.Insert(instance)
+
+		subject := "spinifex.cluster.shutdown.drain.sealok"
+		sub, err := daemon.natsConn.Subscribe(subject, asMsgHandler(daemon.handleShutdownDrain))
+		require.NoError(t, err)
+		defer sub.Unsubscribe()
+		require.NoError(t, daemon.natsConn.Flush())
+
+		payload, err := json.Marshal(ShutdownRequest{Phase: "drain"})
+		require.NoError(t, err)
+
+		reply, err := daemon.natsConn.Request(subject, payload, 60*time.Second)
+		require.NoError(t, err)
+
+		var ack ShutdownACK
+		require.NoError(t, json.Unmarshal(reply.Data, &ack))
+		assert.Equal(t, "drain", ack.Phase)
+		assert.Empty(t, ack.Error, "a sealed volume must not fail DRAIN")
+		assert.Equal(t, []string{"vol-drain-seal-ok"}, unmounted())
+		assert.Equal(t, vm.StateStopped, daemon.vmMgr.Status(instance))
+
+		marker, err := daemon.jsManager.ReadShutdownMarker(daemon.node)
+		require.NoError(t, err)
+		assert.True(t, marker)
+	})
+}

spinifex/daemon/vm_adapters.go

@@ -223,27 +223,30 @@ func (a *volumeMounterAdapter) Unmount(ctx context.Context, instance *vm.VM) err
 	instance.EBSRequests.Mu.Lock()
 	defer instance.EBSRequests.Mu.Unlock()
 
+	var sealErrs []error
 	for _, ebsRequest := range instance.EBSRequests.Requests {
 		ebsUnMountRequest, err := json.Marshal(ebsRequest)
 		if err != nil {
 			slog.Error("Failed to marshal volume payload for unmount", "err", err)
+			sealErrs = append(sealErrs, fmt.Errorf("marshal unmount request for %s: %w", ebsRequest.Name, err))
 			continue
 		}
 
-		// ebs.unmount seals the block map to predastore. Teardown tolerates a
-		// failed seal (log + continue) so terminate stays idempotent, but the
-		// volume must NOT then go available: a reattach on a node without the
-		// local WAL would find no checkpoint (bad superblock). On terminate the
-		// volume is deleted regardless; on stop it stays attached/retryable.
+		// ebs.unmount seals the block map to predastore. Every volume is
+		// attempted, and the failures are aggregated for the caller: the
+		// volume must NOT go available, because a reattach on a node without
+		// the local WAL would find no checkpoint (bad superblock).
 		sealed := true
 		msg, err := ebsRequestWithTrace(ctx, a.nc, instance.AccountID, a.topic("unmount"), ebsUnMountRequest, unmountSealTimeout)
 		if err != nil {
 			slog.Error("Failed to unmount volume",
 				"name", ebsRequest.Name, "instance", instance.ID, "err", err)
+			sealErrs = append(sealErrs, fmt.Errorf("unmount %s: %w", ebsRequest.Name, err))
 			sealed = false
 		} else if sealErr := unmountResponseError(msg.Data); sealErr != nil {
 			slog.Error("Volume unmount seal failed, leaving volume non-available",
 				"instance", instance.ID, "volume", ebsRequest.Name, "err", sealErr)
+			sealErrs = append(sealErrs, fmt.Errorf("seal %s: %w", ebsRequest.Name, sealErr))
 			sealed = false
 		} else {
 			slog.Info("Unmounted volume",
@@ -262,7 +265,7 @@ func (a *volumeMounterAdapter) Unmount(ctx context.Context, instance *vm.VM) err
 		}
 	}
 
-	return nil
+	return errors.Join(sealErrs...)
 }
 
 // MountOne sends ebs.mount for a single request and writes the resolved

spinifex/vm/errors.go

@@ -11,6 +11,11 @@ var ErrInstanceNotFound = errors.New("instance not found")
 // transition (e.g. Stop on an already-stopped instance).
 var ErrInvalidTransition = errors.New("invalid state transition")
 
+// ErrVolumeSealFailed wraps a failed volume unmount during teardown: the
+// block map was not sealed to the object store. Fatal to the DRAIN phase,
+// because stopping viperblock next is what tears the volume's state.
+var ErrVolumeSealFailed = errors.New("volume seal failed")
+
 // 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/vm/manager_deps_test.go

@@ -146,6 +146,7 @@ type fakeVolumeMounter struct {
 	mountedOne, unmountedOne []string
 	mountErr                 error
 	mountOneErr              error
+	unmountErr               error
 	unmountOneErr            error
 	mountOneURI              string
 	// onMount fires synchronously inside Mount before the configured
@@ -169,8 +170,9 @@ func (f *fakeVolumeMounter) Mount(_ context.Context, v *VM) error {
 func (f *fakeVolumeMounter) Unmount(_ context.Context, v *VM) error {
 	f.mu.Lock()
 	f.unmounted = append(f.unmounted, v.ID)
+	err := f.unmountErr
 	f.mu.Unlock()
-	return nil
+	return err
 }
 
 func (f *fakeVolumeMounter) MountOne(_ context.Context, _ string, req *types.EBSRequest) error {

spinifex/vm/shutdown.go

@@ -21,42 +21,49 @@ const pidFileRemovalTimeout = 20 * time.Second
 
 // Stop transitions a running instance to stopped: graceful QMP shutdown, volume
 // unmount, tap teardown, resource deallocation. Migrates to the "stopped" KV
-// bucket and fires OnInstanceDown. Returns ErrInstanceNotFound or ErrInvalidTransition.
+// bucket and fires OnInstanceDown. Returns ErrInstanceNotFound,
+// ErrInvalidTransition, or ErrVolumeSealFailed once the stop has completed.
 func (m *Manager) Stop(id string) error {
 	instance, ok := m.Get(id)
 	if !ok {
 		return ErrInstanceNotFound
 	}
 
-	migrated, err := m.stopOne(instance)
-	if err != nil {
-		return err
+	migrated, stopErr := m.stopOne(instance)
+	// A failed seal still ran the whole sequence, so the ownership hand-off
+	// below must still happen; the error is reported after it.
+	if stopErr != nil && !errors.Is(stopErr, ErrVolumeSealFailed) {
+		return stopErr
 	}
 	if !migrated {
-		return nil
+		return stopErr
 	}
 
 	if err := m.writeRunningState(); err != nil {
 		slog.Error("Failed to persist state after stop, re-adding to local map for consistency",
 			"instanceId", instance.ID, "err", err)
 		m.InsertIfAbsent(instance)
-		return nil
+		return stopErr
 	}
 	slog.Info("Released instance ownership to KV",
 		"instanceId", instance.ID, "state", string(StateStopped), "lastNode", m.deps.NodeID)
-	return nil
+	return stopErr
 }
 
 // stopOne runs the stop sequence shared by Stop and StopAll:
 // Stopping → stopCleanup → Stopped → migrate to "stopped" KV → OnInstanceDown.
-// Returns (true, nil) when migration removed the instance (caller must persist).
-// Returns (false, nil) on KV failure or slot reclaim; (false, err) on precheck failure.
+// The bool reports whether migration removed the instance (caller must persist).
+//
+// A failed volume seal (ErrVolumeSealFailed) does not abort the sequence —
+// QEMU is already down, so the remaining steps must still run — but it is
+// returned so the caller can refuse to advance. A precheck failure returns
+// ErrInvalidTransition with nothing torn down.
 func (m *Manager) stopOne(instance *VM) (bool, error) {
 	if err := m.transitionWithPrecheck(instance, StateStopping); err != nil {
 		return false, err
 	}
 
-	m.stopCleanup(instance)
+	sealErr := m.stopCleanup(instance)
 
 	m.UpdateState(instance.ID, func(v *VM) { v.LastNode = m.deps.NodeID })
 
@@ -69,7 +76,7 @@ func (m *Manager) stopOne(instance *VM) (bool, error) {
 		// map at StateStopped so Restore relaunches it on the next boot. Do
 		// not migrate to the operator-stopped shared bucket or fire
 		// OnInstanceDown; QEMU is already down and resources released.
-		return false, nil
+		return false, sealErr
 	}
 
 	if !m.MigrateStoppedToSharedKV(instance) {
@@ -79,23 +86,30 @@ func (m *Manager) stopOne(instance *VM) (bool, error) {
 		// VM). Either way, do not fire OnInstanceDown — firing it would
 		// unsubscribe the per-id NATS subscriptions of the reclaimed
 		// instance.
-		return false, nil
+		return false, sealErr
 	}
 
 	if m.deps.Hooks.OnInstanceDown != nil {
 		m.deps.Hooks.OnInstanceDown(instance.ID)
 	}
-	return true, nil
+	return true, sealErr
 }
 
 // StopAll fans stopOne across every VM for the coordinated shutdown DRAIN phase.
 // Runs one goroutine per VM; per-VM errors are logged but do not abort the fan-out.
 // AWS resources (ENI, public IP, placement group) are not released on stop.
+//
+// Volume seal failures are aggregated and returned so DRAIN fails: the caller
+// must not let the storage layer stop underneath a block map that never sealed.
 func (m *Manager) StopAll() error {
 	snapshot := m.Snapshot()
 	if len(snapshot) == 0 {
 		return nil
 	}
+	var (
+		mu       sync.Mutex
+		sealErrs []error
+	)
 	var wg sync.WaitGroup
 	for _, instance := range snapshot {
 		wg.Add(1)
@@ -108,15 +122,20 @@ func (m *Manager) StopAll() error {
 					return
 				}
 				slog.Error("StopAll: stopOne failed", "instanceId", v.ID, "err", err)
+				if errors.Is(err, ErrVolumeSealFailed) {
+					mu.Lock()
+					sealErrs = append(sealErrs, err)
+					mu.Unlock()
+				}
 			}
 		}(instance)
 	}
 	wg.Wait()
 	if err := m.writeRunningState(); err != nil {
 		slog.Error("StopAll: failed to persist running state after fan-out", "err", err)
-		return err
+		sealErrs = append(sealErrs, err)
 	}
-	return nil
+	return errors.Join(sealErrs...)
 }
 
 // Terminate transitions an instance to terminated: graceful shutdown, volume +
@@ -224,7 +243,10 @@ func (m *Manager) MarkRecoveryFailed(instance *VM, reason string) {
 		"instanceId", instance.ID, "reason", reason)
 
 	m.goroutineWg.Go(func() {
-		m.stopCleanup(instance)
+		if err := m.stopCleanup(instance); err != nil {
+			slog.Error("Volume seal failed during recovery cleanup; volume left unsealed for operator action",
+				"instanceId", instance.ID, "err", err)
+		}
 		m.Inspect(instance, func(v *VM) { v.LastNode = m.deps.NodeID })
 		if err := m.writeRunningState(); err != nil {
 			slog.Error("Failed to persist state after recovery failure",
@@ -379,9 +401,9 @@ func (m *Manager) stampOutstandingTeardownFailed(instance *VM) {
 // stopCleanup performs the per-instance teardown shared by Stop and the
 // initial section of Terminate: graceful QMP shutdown, PID-file wait,
 // volume unmount, tap teardown (main + extra ENI + mgmt), resource
-// deallocation. Per-step errors are logged and tolerated.
-func (m *Manager) stopCleanup(instance *VM) {
-	m.shutdownAndUnmount(instance)
+// deallocation. Every step runs; only a failed volume seal is returned.
+func (m *Manager) stopCleanup(instance *VM) error {
+	sealErr := m.shutdownAndUnmount(instance)
 	m.cleanupTapDevices(instance)
 	if m.deps.InstanceCleaner != nil {
 		if err := m.deps.InstanceCleaner.ReleaseGPU(instance); err != nil {
@@ -397,13 +419,18 @@ func (m *Manager) stopCleanup(instance *VM) {
 	if instance.CapacityReservationId != "" {
 		m.UpdateState(instance.ID, func(v *VM) { v.CapacityReservationId = "" })
 	}
+
+	return sealErr
 }
 
 // terminateCleanup is stopCleanup plus the AWS-resource cleanup that
 // only applies on terminate: volume deletion, public IP release, ENI
 // deletion, placement-group removal.
 func (m *Manager) terminateCleanup(instance *VM) {
-	m.shutdownAndUnmount(instance)
+	// Terminate deletes the volumes next, so an unsealed block map loses
+	// nothing a caller can still ask for. Tolerated to keep terminate
+	// idempotent; only stop and DRAIN treat a failed seal as fatal.
+	_ = m.shutdownAndUnmount(instance)
 	m.markTeardown(instance, TeardownQEMU, TeardownDone)
 
 	if m.deps.InstanceCleaner != nil {
@@ -457,8 +484,12 @@ func (m *Manager) terminateCleanup(instance *VM) {
 
 // shutdownAndUnmount asks QEMU to power down via QMP, waits for the PID
 // file to disappear (force-killing on timeout), then unmounts every
-// attached volume. Each step tolerates failure of the previous one.
-func (m *Manager) shutdownAndUnmount(instance *VM) {
+// attached volume. Each step runs regardless of the one before it.
+//
+// Returns ErrVolumeSealFailed when the unmount did not seal the block map:
+// that is the one step whose failure means data loss. QMP, PID-file, fw_cfg
+// and telemetry cleanup stay best-effort and are logged only.
+func (m *Manager) shutdownAndUnmount(instance *VM) error {
 	if instance.QMPClient != nil {
 		if _, err := sendQMPCommand(context.Background(), instance.QMPClient, qmp.QMPCommand{Execute: "system_powerdown"}, instance.ID); err != nil {
 			slog.Warn("QMP system_powerdown failed (VM may already be stopped)",
@@ -484,9 +515,11 @@ func (m *Manager) shutdownAndUnmount(instance *VM) {
 		}
 	}
 
+	var sealErr error
 	if m.deps.VolumeMounter != nil {
 		if err := m.deps.VolumeMounter.Unmount(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)
 		}
 	}
 
@@ -497,6 +530,8 @@ func (m *Manager) shutdownAndUnmount(instance *VM) {
 	}
 
 	removeTelemetryArtifacts(instance)
+
+	return sealErr
 }
 
 // cleanupTapDevices removes the primary VPC tap, every extra ENI tap, and

spinifex/vm/shutdown_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"os"
+	"path/filepath"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -631,7 +632,7 @@ func TestStopCleanup_InvokesReleaseGPU(t *testing.T) {
 	m := NewManagerWithDeps(Deps{InstanceCleaner: cleaner})
 	instance := &VM{ID: "i-stop", GPUAttachments: []gpu.GPUAttachment{{PCIAddress: "0000:01:00.0"}}}
 
-	m.stopCleanup(instance)
+	require.NoError(t, m.stopCleanup(instance))
 
 	if got := cleaner.releaseGPU; len(got) != 1 || got[0] != "i-stop" {
 		t.Fatalf("ReleaseGPU on stopCleanup: got %v, want [i-stop]", got)
@@ -653,7 +654,7 @@ func TestStopCleanup_ReservationBoundReleasesAndClears(t *testing.T) {
 	instance := &VM{ID: "i-cr", InstanceType: "t3.micro", CapacityReservationId: "cr-123"}
 	m.Insert(instance)
 
-	m.stopCleanup(instance)
+	require.NoError(t, m.stopCleanup(instance))
 
 	if got := rc.releasedToCRID; len(got) != 1 || got[0] != "cr-123" {
 		t.Fatalf("stop must release the slot to the reservation: got %v, want [cr-123]", got)
@@ -691,7 +692,7 @@ func TestCleanup_NoGPU_StillInvokesReleaseGPU(t *testing.T) {
 	m := NewManagerWithDeps(Deps{InstanceCleaner: cleaner})
 	instance := &VM{ID: "i-cpu"}
 
-	m.stopCleanup(instance)
+	require.NoError(t, m.stopCleanup(instance))
 	m.terminateCleanup(instance)
 
 	if got := len(cleaner.releaseGPU); got != 2 {
@@ -1355,7 +1356,7 @@ func TestShutdownAndUnmount_NilQMPClient_ProceedsToUnmount(t *testing.T) {
 	instance := &VM{ID: "i-noqmp", QMPClient: nil}
 
 	require.NotPanics(t, func() {
-		m.shutdownAndUnmount(instance)
+		_ = m.shutdownAndUnmount(instance)
 	})
 
 	mounter.mu.Lock()
@@ -1383,7 +1384,7 @@ func TestShutdownAndUnmount_PowerdownSent_NoForceKill(t *testing.T) {
 
 	instance := &VM{ID: "i-powerdown", QMPClient: qmpClient}
 
-	m.shutdownAndUnmount(instance)
+	require.NoError(t, m.shutdownAndUnmount(instance))
 
 	assert.Equal(t, []string{"system_powerdown"}, recorder.executes(),
 		"shutdownAndUnmount must dispatch exactly one system_powerdown QMP command")
@@ -1413,7 +1414,7 @@ func TestShutdownAndUnmount_PIDFileRemovedBeforeTimeout_NoForceKill(t *testing.T
 	instance := &VM{ID: "i-pid-removed", QMPClient: nil}
 
 	start := time.Now()
-	m.shutdownAndUnmount(instance)
+	require.NoError(t, m.shutdownAndUnmount(instance))
 	elapsed := time.Since(start)
 
 	assert.Less(t, elapsed, pidFileRemovalTimeout,
@@ -1436,6 +1437,144 @@ func TestShutdownAndUnmount_NilVolumeMounter_SkipsUnmount(t *testing.T) {
 	instance := &VM{ID: "i-nomounter", QMPClient: nil}
 
 	require.NotPanics(t, func() {
-		m.shutdownAndUnmount(instance)
+		_ = m.shutdownAndUnmount(instance)
 	})
 }
+
+// errSealStalled is the shape a stalled predastore blob node produces: the
+// block-map seal never completes, so ebs.unmount reports a failure.
+var errSealStalled = errors.New("seal volume to predastore: transfer stalled: idle timeout")
+
+// TestShutdownAndUnmount_SealFailure_ReturnsErrorAndCompletesCleanup proves the
+// volume seal is the one step whose failure is reported: the error wraps
+// ErrVolumeSealFailed, and the best-effort fw_cfg cleanup after it still runs.
+func TestShutdownAndUnmount_SealFailure_ReturnsErrorAndCompletesCleanup(t *testing.T) {
+	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
+	mounter := &fakeVolumeMounter{unmountErr: errSealStalled}
+	m := NewManagerWithDeps(Deps{VolumeMounter: mounter})
+
+	fwPath := filepath.Join(t.TempDir(), "netcfg.json")
+	require.NoError(t, os.WriteFile(fwPath, []byte("{}"), 0o600))
+	instance := &VM{ID: "i-seal-fail"}
+	instance.Config.FwCfg = []FwCfgEntry{{Name: "opt/spinifex/netcfg", File: fwPath}}
+
+	err := m.shutdownAndUnmount(instance)
+
+	require.ErrorIs(t, err, ErrVolumeSealFailed,
+		"a failed volume unmount must be reported as a seal failure")
+	require.ErrorIs(t, err, errSealStalled, "the mounter's cause must be preserved")
+	assert.Contains(t, err.Error(), "i-seal-fail", "the error must name the instance")
+	assert.NoFileExists(t, fwPath,
+		"fw_cfg cleanup is best-effort and must still run after a failed seal")
+}
+
+// TestStopCleanup_SealFailure_StillRunsBestEffortSteps verifies the seal error
+// does not abort the rest of the teardown: tap/GPU/resource cleanup all run,
+// because QEMU is already down by the time the seal is attempted.
+func TestStopCleanup_SealFailure_StillRunsBestEffortSteps(t *testing.T) {
+	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
+	cleaner := &recordingInstanceCleaner{}
+	rc := &countingResourceController{}
+	m := NewManagerWithDeps(Deps{
+		VolumeMounter:   &fakeVolumeMounter{unmountErr: errSealStalled},
+		InstanceCleaner: cleaner,
+		Resources:       rc,
+	})
+	instance := &VM{ID: "i-seal-cleanup", InstanceType: "t3.micro"}
+	m.Insert(instance)
+
+	err := m.stopCleanup(instance)
+
+	require.ErrorIs(t, err, ErrVolumeSealFailed)
+	assert.Equal(t, []string{"i-seal-cleanup"}, cleaner.releaseGPU,
+		"GPU release must still run after a failed seal")
+	assert.Equal(t, []string{"i-seal-cleanup"}, cleaner.cleanupMgmt,
+		"mgmt network cleanup must still run after a failed seal")
+	assert.Equal(t, 1, rc.deallocations,
+		"resources must still be released after a failed seal")
+	assert.Empty(t, cleaner.deleteVolumes,
+		"a failed seal must never escalate stop into volume deletion")
+}
+
+// TestStopAll_SealFailure_AggregatesAndPropagates is the DRAIN invariant: every
+// VM is still stopped, and every per-VM seal failure comes back to the caller so
+// the coordinated shutdown refuses to advance to the STORAGE phase.
+func TestStopAll_SealFailure_AggregatesAndPropagates(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	m, _, mounter, _, _ := shutdownTestManager(t)
+	mounter.unmountErr = errSealStalled
+
+	ids := []string{"i-drain-seal-a", "i-drain-seal-b"}
+	for _, id := range ids {
+		m.Insert(&VM{ID: id, Status: StateRunning, DesiredState: DesiredStopped})
+	}
+
+	err := m.StopAll()
+
+	require.Error(t, err, "StopAll must not report success when a volume seal failed")
+	require.ErrorIs(t, err, ErrVolumeSealFailed)
+	for _, id := range ids {
+		assert.Contains(t, err.Error(), id,
+			"the aggregate must name every VM whose seal failed")
+	}
+
+	mounter.mu.Lock()
+	defer mounter.mu.Unlock()
+	assert.ElementsMatch(t, ids, mounter.unmounted,
+		"a failed seal on one VM must not abort the fan-out")
+}
+
+// TestStopAll_SealSucceeds_ReturnsNil is the control for the test above: an
+// identical fan-out whose seals all succeed must still report success.
+func TestStopAll_SealSucceeds_ReturnsNil(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	m, _, mounter, _, _ := shutdownTestManager(t)
+
+	m.Insert(&VM{ID: "i-drain-ok", Status: StateRunning, DesiredState: DesiredStopped})
+
+	require.NoError(t, m.StopAll())
+
+	mounter.mu.Lock()
+	defer mounter.mu.Unlock()
+	assert.Equal(t, []string{"i-drain-ok"}, mounter.unmounted)
+}
+
+// TestStop_SealFailure_CompletesStopAndReportsError verifies the seal error does
+// not short-circuit Stop: the instance still reaches Stopped and migrates to the
+// shared KV, and the caller still learns the volume was not sealed.
+func TestStop_SealFailure_CompletesStopAndReportsError(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	m, store, mounter, _, _ := shutdownTestManager(t)
+	mounter.unmountErr = errSealStalled
+
+	instance := &VM{ID: "i-stop-seal", Status: StateRunning, DesiredState: DesiredStopped}
+	m.Insert(instance)
+
+	err := m.Stop(instance.ID)
+
+	require.ErrorIs(t, err, ErrVolumeSealFailed)
+	assert.Equal(t, StateStopped, m.Status(instance),
+		"the guest is down regardless of the seal, so the stop must complete")
+	_, present := m.Get(instance.ID)
+	assert.False(t, present, "a completed stop must still migrate out of the local map")
+	stopped, listErr := store.ListStoppedInstances()
+	require.NoError(t, listErr)
+	assert.Len(t, stopped, 1, "the stopped-instance record must still be written")
+}
+
+// TestTerminate_SealFailure_Tolerated verifies terminate is unchanged by the
+// stricter stop path: the volumes are deleted next, so an unsealed block map
+// loses nothing and terminate must stay idempotent.
+func TestTerminate_SealFailure_Tolerated(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	m, _, mounter, cleaner, _ := shutdownTestManager(t)
+	mounter.unmountErr = errSealStalled
+
+	instance := &VM{ID: "i-term-seal", Status: StateRunning}
+	m.Insert(instance)
+
+	require.NoError(t, m.Terminate(instance.ID),
+		"terminate must tolerate a failed seal")
+	assert.Equal(t, []string{"i-term-seal"}, cleaner.deleteVolumes,
+		"terminate must still delete the volumes after a failed seal")
+}

spinifex/vm/volume_mounter.go

@@ -18,8 +18,9 @@ type VolumeMounter interface {
 	// Mount mounts every attached volume in v.EBSRequests.Requests, recording
 	// the resolved NBDURI back onto each request entry.
 	Mount(ctx context.Context, v *VM) error
-	// Unmount sends ebs.unmount for each attached volume. Errors are logged
-	// per volume and aggregated; partial failure is tolerated.
+	// Unmount sends ebs.unmount for each attached volume, which drives the
+	// synchronous block-map seal. Every volume is attempted; the seal
+	// failures are aggregated and returned, and mean the data is not durable.
 	Unmount(ctx context.Context, v *VM) error
 
 	// MountOne sends ebs.mount for a single request and writes the resolved