fix(ec2): make EBS unmount seal + stopped-instance RMW idempotent (#491)

Close four pre-existing lifecycle races surfaced during env19 hardening.

Volume seal/reaper (42hbm, c7ce2): the ebs.unmount handler now seals the
volume before removing it from MountedVolumes, so a retry after a
timed-out-but-completed seal reliably reports NotFound. EBSUnMountResponse
carries a NotFound flag and unmountResponseError treats it as seal success,
so state reconciles to available/detached instead of leaking. The volume
leak reaper reconciles the attachment record (never the data) for volumes
whose owning instance is terminated, so GC can delete them.

Stopped-instance CAS (ozlbq, wlmea): add UpdateStoppedInstance, a
mutate-callback backed by casUpdate(createIfAbsent=false), and migrate
TagStoppedInstance and ModifyInstanceAttribute off the resurrect-prone
Load->mutate->WriteStoppedInstance path. A tag/attribute write racing a
winning start-claim now fails cleanly instead of re-creating the deleted
entry. StartStoppedInstance returns IncorrectInstanceState (not
InvalidInstanceID.NotFound) when the stopped entry is absent but the
instance is present locally.

spinifex/daemon/daemon_handlers_instance_tags_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -52,6 +53,17 @@ func (m *memStoppedStore) DeleteStoppedInstance(id string) error {
 	return nil
 }
 
+// UpdateStoppedInstance mimics the real CAS semantics: a missing record
+// returns nats.ErrKeyNotFound instead of resurrecting it.
+func (m *memStoppedStore) UpdateStoppedInstance(id string, mutate func(*vm.VM)) (*vm.VM, error) {
+	v, ok := m.instances[id]
+	if !ok {
+		return nil, nats.ErrKeyNotFound
+	}
+	mutate(v)
+	return v, nil
+}
+
 func (m *memStoppedStore) ClaimStoppedInstance(id string) (*vm.VM, error) {
 	v, ok := m.instances[id]
 	if !ok {

spinifex/daemon/daemon_handlers_instance_test.go

@@ -36,6 +36,7 @@ type fakeStateStore struct {
 	stopped          map[string]*vm.VM
 	loadStoppedErr   error
 	writeStoppedErr  error
+	updateStoppedErr error
 	listStoppedErr   error
 	deleteStoppedErr error
 
@@ -123,6 +124,24 @@ func (f *fakeStateStore) ClaimStoppedInstance(id string) (*vm.VM, error) {
 	return v, nil
 }
 
+// UpdateStoppedInstance mimics the real CAS semantics for tests: mutate runs
+// under the store lock against the stored value, and a missing record
+// returns nats.ErrKeyNotFound (matching JetStreamManager's createIfAbsent=false
+// behavior) rather than resurrecting it.
+func (f *fakeStateStore) UpdateStoppedInstance(id string, mutate func(*vm.VM)) (*vm.VM, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.updateStoppedErr != nil {
+		return nil, f.updateStoppedErr
+	}
+	v, ok := f.stopped[id]
+	if !ok {
+		return nil, nats.ErrKeyNotFound
+	}
+	mutate(v)
+	return v, nil
+}
+
 func (f *fakeStateStore) ListStoppedInstances() ([]*vm.VM, error) {
 	if f.listStoppedErr != nil {
 		return nil, f.listStoppedErr
@@ -477,7 +496,7 @@ func TestHandleEC2TerminateStoppedInstance_CrossTenantRejected(t *testing.T) {
 
 func TestHandleEC2ModifyInstanceAttribute_WriteFailureReturnsServerInternal(t *testing.T) {
 	store := newFakeStateStore()
-	store.writeStoppedErr = errors.New("kv write failed")
+	store.updateStoppedErr = errors.New("kv write failed")
 	v := stoppedVMFixture("i-mod-write-fail", testAccountID)
 	store.stopped[v.ID] = v
 	d := daemonWithFakeStateStore(t, store)

spinifex/daemon/daemon_test.go

@@ -3277,6 +3277,48 @@ func TestVolumeMounterAdapter_Unmount_SealFailureSkipsAvailable(t *testing.T) {
 		"a successful seal must still transition the volume to available")
 }
 
+// TestVolumeMounterAdapter_Unmount_NotFoundFlipsToAvailable verifies the
+// timeout-then-completed-seal retry path: an ebs.unmount response reporting
+// NotFound (the seal already completed on a prior, client-timed-out request)
+// must still flip a non-boot data volume to available, while the boot/EFI
+// gate stays intact.
+func TestVolumeMounterAdapter_Unmount_NotFoundFlipsToAvailable(t *testing.T) {
+	daemon := createTestDaemon(t, sharedNATSURL)
+	volState := &recordingVolState{}
+	adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, volState)
+
+	sub, err := daemon.natsConn.Subscribe(adapter.topic("unmount"), func(msg *nats.Msg) {
+		var req types.EBSRequest
+		json.Unmarshal(msg.Data, &req)
+		resp := types.EBSUnMountResponse{
+			Volume:   req.Name,
+			NotFound: true,
+			Error:    fmt.Sprintf("Volume %s not found", req.Name),
+		}
+		data, _ := json.Marshal(resp)
+		msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	inst := &vm.VM{
+		ID: "i-unmount-retry",
+		EBSRequests: types.EBSRequests{
+			Requests: []types.EBSRequest{
+				{Name: "vol-boot-retry", Boot: true, EFI: false},
+				{Name: "vol-data-retry", Boot: false, EFI: false},
+			},
+		},
+	}
+	require.NoError(t, adapter.Unmount(inst))
+
+	calls := volState.snapshot()
+	assert.Contains(t, calls, "vol-data-retry:available",
+		"a NotFound response (already-completed seal) must still flip a non-boot data volume to available")
+	assert.NotContains(t, calls, "vol-boot-retry:available",
+		"a boot volume must stay attached even when the seal retry reports NotFound")
+}
+
 // TestUnmountResponseError covers the shared seal-result parser used by both the
 // gated DetachVolume path (unmountOne) and the tolerated teardown path (Unmount).
 func TestUnmountResponseError(t *testing.T) {
@@ -3295,6 +3337,11 @@ func TestUnmountResponseError(t *testing.T) {
 	t.Run("malformed payload propagates", func(t *testing.T) {
 		require.Error(t, unmountResponseError([]byte("not json")))
 	})
+	t.Run("not found treated as idempotent success", func(t *testing.T) {
+		data, _ := json.Marshal(types.EBSUnMountResponse{Volume: "v", NotFound: true, Error: "Volume v not found"})
+		require.NoError(t, unmountResponseError(data),
+			"a NotFound response means the seal already completed on a prior request; a timeout-then-retry must not be treated as a failure")
+	})
 }
 
 // TestVolumeMounterAdapter_Mount_PartialFailureRollback verifies that when

spinifex/daemon/jetstream.go

@@ -927,6 +927,35 @@ func (m *JetStreamManager) ClaimStoppedInstance(instanceID string) (*vm.VM, erro
 	return instance, nil
 }
 
+// UpdateStoppedInstance atomically applies mutate to the current KV-stored
+// stopped record for instanceID and writes it back using optimistic
+// concurrency (CAS), retrying on a concurrent writer's revision conflict.
+// createIfAbsent is false: if a winning ClaimStoppedInstance deletes the
+// record mid-flight, the CAS write fails with nats.ErrKeyNotFound instead of
+// resurrecting it. Used by tag/attribute mutations that read-modify-write a
+// stopped instance so they cannot race a claim into recreating a stale
+// record. Returns nats.ErrKeyNotFound if no record exists.
+func (m *JetStreamManager) UpdateStoppedInstance(instanceID string, mutate func(*vm.VM)) (*vm.VM, error) {
+	if m.kv == nil {
+		return nil, errors.New("KV bucket not initialized")
+	}
+
+	key := StoppedInstancePrefix + instanceID
+	updated, err := casUpdate(m.kv, key, mutate, false)
+	if err != nil {
+		if isStreamUnavailable(err) {
+			slog.Warn("KV stream unavailable, attempting recovery", "operation", "UpdateStoppedInstance", "key", key, "err", err)
+			kv, recoverErr := m.recoverKVBucket()
+			if recoverErr != nil {
+				return nil, err
+			}
+			return casUpdate(kv, key, mutate, false)
+		}
+		return nil, err
+	}
+	return updated, nil
+}
+
 // ListStoppedInstances returns all stopped instances from the shared KV store.
 func (m *JetStreamManager) ListStoppedInstances() ([]*vm.VM, error) {
 	if m.kv == nil {

spinifex/daemon/jetstream_test.go

@@ -553,6 +553,106 @@ func TestJetStreamManager_ClaimStoppedInstance_ConcurrentClaim(t *testing.T) {
 	assert.Nil(t, loaded, "the winning claim must have removed the record")
 }
 
+// TestJetStreamManager_UpdateStoppedInstance_ConflictRetry mirrors
+// UpdateTerminatedInstance's CAS contract for the stopped-instance bucket: a
+// concurrent writer landing between Get and Update must be detected via a
+// revision conflict and retried, and both updates must land.
+func TestJetStreamManager_UpdateStoppedInstance_ConflictRetry(t *testing.T) {
+	nc, err := nats.Connect(sharedJSNATSURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	jsm, err := NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, jsm.InitKVBucket())
+
+	testVM := &vm.VM{ID: "i-stopped-cas-conflict", Status: vm.StateStopped, InstanceType: "t3.micro"}
+	require.NoError(t, jsm.WriteStoppedInstance(testVM.ID, testVM))
+	defer func() { _ = jsm.DeleteStoppedInstance(testVM.ID) }()
+
+	var attempts int
+	var injectOnce sync.Once
+	updated, err := jsm.UpdateStoppedInstance(testVM.ID, func(v *vm.VM) {
+		attempts++
+		v.InstanceType = "t3.small"
+
+		// Simulate a concurrent writer landing between our Get and Update on
+		// the first attempt only, forcing exactly one revision conflict.
+		injectOnce.Do(func() {
+			_, cerr := jsm.UpdateStoppedInstance(testVM.ID, func(cv *vm.VM) {
+				cv.LastNode = "node-concurrent"
+			})
+			require.NoError(t, cerr)
+		})
+	})
+	require.NoError(t, err, "the retried Update must eventually succeed against the fresh revision")
+	require.NotNil(t, updated)
+	assert.Equal(t, 2, attempts, "the injected concurrent write must force exactly one retry")
+	assert.Equal(t, "t3.small", updated.InstanceType)
+
+	loaded, err := jsm.LoadStoppedInstance(testVM.ID)
+	require.NoError(t, err)
+	require.NotNil(t, loaded)
+	assert.Equal(t, "node-concurrent", loaded.LastNode, "the concurrent writer's update must survive, not be clobbered")
+	assert.Equal(t, "t3.small", loaded.InstanceType, "the retried update must also land")
+}
+
+// TestJetStreamManager_UpdateStoppedInstance_NotFound verifies
+// UpdateStoppedInstance surfaces nats.ErrKeyNotFound rather than silently
+// creating a record when nothing exists to update.
+func TestJetStreamManager_UpdateStoppedInstance_NotFound(t *testing.T) {
+	nc, err := nats.Connect(sharedJSNATSURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	jsm, err := NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, jsm.InitKVBucket())
+
+	_, err = jsm.UpdateStoppedInstance("i-does-not-exist", func(v *vm.VM) {})
+	assert.ErrorIs(t, err, nats.ErrKeyNotFound)
+}
+
+// TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim is the
+// core TOCTOU regression test: a claim (delete) that lands between an
+// UpdateStoppedInstance caller's own Load and its CAS write must not be
+// undone. createIfAbsent=false means the CAS write observes the key gone and
+// fails with ErrKeyNotFound instead of recreating the stopped record after
+// ClaimStoppedInstance already handed it off to a winning start.
+func TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim(t *testing.T) {
+	nc, err := nats.Connect(sharedJSNATSURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	jsm, err := NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, jsm.InitKVBucket())
+
+	testVM := &vm.VM{ID: "i-stopped-claim-race", Status: vm.StateStopped, InstanceType: "t3.micro"}
+	require.NoError(t, jsm.WriteStoppedInstance(testVM.ID, testVM))
+
+	// A caller loads the record (as TagStoppedInstance / ModifyInstanceAttribute
+	// do) before a winning claim removes it.
+	loaded, err := jsm.LoadStoppedInstance(testVM.ID)
+	require.NoError(t, err)
+	require.NotNil(t, loaded)
+
+	claimed, err := jsm.ClaimStoppedInstance(testVM.ID)
+	require.NoError(t, err)
+	require.NotNil(t, claimed)
+
+	// The tag/attribute caller's CAS write now races the already-completed
+	// claim: it must fail cleanly, not resurrect the record.
+	_, err = jsm.UpdateStoppedInstance(testVM.ID, func(v *vm.VM) {
+		v.InstanceType = "should-not-land"
+	})
+	assert.ErrorIs(t, err, nats.ErrKeyNotFound, "a claim that deleted the record must not be resurrected by a losing racer's update")
+
+	stillGone, err := jsm.LoadStoppedInstance(testVM.ID)
+	require.NoError(t, err)
+	assert.Nil(t, stillGone, "the stopped record must stay deleted after the losing update")
+}
+
 // TestJetStreamManager_ListStoppedInstances tests listing multiple stopped instances
 func TestJetStreamManager_ListStoppedInstances(t *testing.T) {
 	nc, err := nats.Connect(sharedJSNATSURL)

spinifex/daemon/vm_adapters.go

@@ -52,6 +52,10 @@ func (a *stateStoreAdapter) DeleteStoppedInstance(id string) error {
 	return a.js.DeleteStoppedInstance(id)
 }
 
+func (a *stateStoreAdapter) UpdateStoppedInstance(id string, mutate func(*vm.VM)) (*vm.VM, error) {
+	return a.js.UpdateStoppedInstance(id, mutate)
+}
+
 func (a *stateStoreAdapter) ClaimStoppedInstance(id string) (*vm.VM, error) {
 	return a.js.ClaimStoppedInstance(id)
 }
@@ -256,12 +260,18 @@ func (a *volumeMounterAdapter) unmountOne(req types.EBSRequest) error {
 
 // unmountResponseError reports a seal/unmount failure from an ebs.unmount
 // response payload: a non-empty Error, or a volume still reported mounted.
-// Returns nil when the unmount and its block-map seal succeeded.
+// Returns nil when the unmount and its block-map seal succeeded, or when the
+// volume was already unmounted and sealed (NotFound) — viperblockd only
+// reports NotFound once the seal itself has completed, so a retry landing
+// here is an idempotent success, not a failure.
 func unmountResponseError(data []byte) error {
 	var resp types.EBSUnMountResponse
 	if err := json.Unmarshal(data, &resp); err != nil {
 		return fmt.Errorf("unmarshal unmount response: %w", err)
 	}
+	if resp.NotFound {
+		return nil
+	}
 	if resp.Error != "" {
 		return fmt.Errorf("ebs.unmount returned error: %s", resp.Error)
 	}

spinifex/handlers/ec2/instance/instance_tags_test.go

@@ -151,12 +151,13 @@ func TestTagStoppedInstance_WritesRecordAndCentral(t *testing.T) {
 	require.NoError(t, err)
 
 	want := map[string]string{"Name": "web", "env": "prod"}
-	written := store.wroteStopped["i-123"]
+	written := store.loadByID["i-123"]
 	require.NotNil(t, written)
 	assert.Equal(t, want, tagsAsMap(written.Instance.Tags))
 	assert.Equal(t, want, writer.tags)
 	assert.Equal(t, "i-123", writer.resourceID)
 	assert.Equal(t, "111122223333", writer.accountID)
+	assert.Equal(t, 1, store.updateStoppedCalls)
 }
 
 func TestTagStoppedInstance_RemoveKeys(t *testing.T) {
@@ -171,7 +172,7 @@ func TestTagStoppedInstance_RemoveKeys(t *testing.T) {
 	require.NoError(t, err)
 
 	want := map[string]string{"Name": "web"}
-	assert.Equal(t, want, tagsAsMap(store.wroteStopped["i-123"].Instance.Tags))
+	assert.Equal(t, want, tagsAsMap(store.loadByID["i-123"].Instance.Tags))
 	assert.Equal(t, want, writer.tags)
 }
 
@@ -204,6 +205,26 @@ func TestTagStoppedInstance_CrossAccountRejected(t *testing.T) {
 	assert.Empty(t, store.wroteStopped)
 }
 
+// TestTagStoppedInstance_ConcurrentClaimDoesNotResurrect covers a
+// TagStoppedInstance racing a winning ClaimStoppedInstance (e.g.
+// StartStoppedInstance) that deletes the record between this call's Load and
+// its CAS write: the KV update must fail cleanly instead of resurrecting a
+// stale stopped entry, and the caller must see a NotFound-shaped error.
+func TestTagStoppedInstance_ConcurrentClaimDoesNotResurrect(t *testing.T) {
+	stored := stoppedTagInstance("i-123", "111122223333", tagList("Name", "web"))
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{"i-123": stored}, claimAfterLoad: true}
+	writer := &fakeTagWriter{}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	err := svc.TagStoppedInstance(context.Background(), "i-123", &spxtypes.InstanceTagsData{
+		Tags: map[string]string{"env": "prod"},
+	}, false, writer, "111122223333")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.Empty(t, store.loadByID, "the claimed record must not be resurrected")
+	assert.Equal(t, []string{"i-123"}, store.claimedStopped)
+}
+
 func TestTagStoppedInstance_CentralWriteErrorSkipsRecordWrite(t *testing.T) {
 	stored := stoppedTagInstance("i-123", "111122223333", tagList("Name", "web"))
 	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{"i-123": stored}}

spinifex/handlers/ec2/instance/service.go

@@ -85,6 +85,11 @@ type StoppedInstanceStore interface {
 	ListTerminatedInstances() ([]*vm.VM, error)
 	WriteStoppedInstance(instanceID string, instance *vm.VM) error
 	DeleteStoppedInstance(instanceID string) error
+	// UpdateStoppedInstance atomically applies mutate to the current stopped
+	// record under optimistic concurrency (CAS) with createIfAbsent=false, so
+	// a caller racing a winning ClaimStoppedInstance gets a clean
+	// nats.ErrKeyNotFound instead of resurrecting a deleted record.
+	UpdateStoppedInstance(instanceID string, mutate func(*vm.VM)) (*vm.VM, error)
 	WriteTerminatedInstance(instanceID string, instance *vm.VM) error
 	// ClaimStoppedInstance atomically removes instanceID's record and
 	// returns the VM it held, so at most one caller can ever win a race to

spinifex/handlers/ec2/instance/service_impl.go

@@ -365,7 +365,21 @@ func (s *InstanceServiceImpl) TagStoppedInstance(ctx context.Context, instanceID
 		slog.ErrorContext(ctx, "TagStoppedInstance: central tag store write failed", "instanceId", instanceID, "err", err)
 		return errors.New(awserrors.ErrorServerInternal)
 	}
-	if err := s.stoppedStore.WriteStoppedInstance(instanceID, instance); err != nil {
+
+	// CAS the tag mutation into the shared KV record instead of a wholesale
+	// WriteStoppedInstance: createIfAbsent=false means a claim that deletes
+	// the record between the Load above and this write fails cleanly with
+	// nats.ErrKeyNotFound rather than resurrecting a stale stopped entry.
+	newTags := instance.Instance.Tags
+	if _, err := s.stoppedStore.UpdateStoppedInstance(instanceID, func(v *vm.VM) {
+		if v.Instance != nil {
+			v.Instance.Tags = newTags
+		}
+	}); err != nil {
+		if errors.Is(err, nats.ErrKeyNotFound) {
+			slog.WarnContext(ctx, "TagStoppedInstance: instance claimed concurrently, not resurrecting", "instanceId", instanceID)
+			return errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+		}
 		slog.ErrorContext(ctx, "TagStoppedInstance: failed to write stopped instance", "instanceId", instanceID, "err", err)
 		return errors.New(awserrors.ErrorServerInternal)
 	}
@@ -1987,32 +2001,46 @@ func (s *InstanceServiceImpl) ModifyInstanceAttribute(ctx context.Context, input
 		}
 		slog.InfoContext(ctx, "ModifyInstanceAttribute: changing instance type",
 			"instanceId", instanceID, "oldType", instance.InstanceType, "newType", newType)
-
-		instance.InstanceType = newType
-		instance.Config.InstanceType = newType
-		instance.Instance.InstanceType = aws.String(newType)
-		// Clear StateReason — resolves capacity-unavailable state from instance-type-missing bug.
-		instance.Instance.StateReason = nil
 	}
 
-	if input.UserData != nil && input.UserData.Value != nil {
-		slog.InfoContext(ctx, "ModifyInstanceAttribute: changing user data", "instanceId", instanceID)
-		if instance.RunInstancesInput == nil {
-			instance.RunInstancesInput = &ec2.RunInstancesInput{}
+	// CAS the mutations into the shared KV record instead of a wholesale
+	// WriteStoppedInstance: createIfAbsent=false means a claim that deletes
+	// the record between the Load above and this write fails cleanly with
+	// nats.ErrKeyNotFound rather than resurrecting a stale stopped entry.
+	_, err = s.stoppedStore.UpdateStoppedInstance(instanceID, func(v *vm.VM) {
+		if input.InstanceType != nil && input.InstanceType.Value != nil && v.Instance != nil {
+			newType := *input.InstanceType.Value
+			v.InstanceType = newType
+			v.Config.InstanceType = newType
+			v.Instance.InstanceType = aws.String(newType)
+			// Clear StateReason — resolves capacity-unavailable state from instance-type-missing bug.
+			v.Instance.StateReason = nil
 		}
-		instance.RunInstancesInput.UserData = aws.String(base64.StdEncoding.EncodeToString(input.UserData.Value))
-	}
 
-	if input.DisableApiTermination != nil && input.DisableApiTermination.Value != nil {
-		slog.InfoContext(ctx, "ModifyInstanceAttribute: changing DisableApiTermination on stopped instance",
-			"instanceId", instanceID, "value", *input.DisableApiTermination.Value)
-		if instance.RunInstancesInput == nil {
-			instance.RunInstancesInput = &ec2.RunInstancesInput{}
+		if input.UserData != nil && input.UserData.Value != nil {
+			slog.InfoContext(ctx, "ModifyInstanceAttribute: changing user data", "instanceId", instanceID)
+			if v.RunInstancesInput == nil {
+				v.RunInstancesInput = &ec2.RunInstancesInput{}
+			}
+			v.RunInstancesInput.UserData = aws.String(base64.StdEncoding.EncodeToString(input.UserData.Value))
 		}
-		instance.RunInstancesInput.DisableApiTermination = input.DisableApiTermination.Value
-	}
 
-	if err := s.stoppedStore.WriteStoppedInstance(instanceID, instance); err != nil {
+		if input.DisableApiTermination != nil && input.DisableApiTermination.Value != nil {
+			slog.InfoContext(ctx, "ModifyInstanceAttribute: changing DisableApiTermination on stopped instance",
+				"instanceId", instanceID, "value", *input.DisableApiTermination.Value)
+			if v.RunInstancesInput == nil {
+				v.RunInstancesInput = &ec2.RunInstancesInput{}
+			}
+			v.RunInstancesInput.DisableApiTermination = input.DisableApiTermination.Value
+		}
+	})
+	if err != nil {
+		if errors.Is(err, nats.ErrKeyNotFound) {
+			// The record existed moments ago (Load above) but a concurrent
+			// claim removed it: the instance is mid-transition, not gone.
+			slog.WarnContext(ctx, "ModifyInstanceAttribute: instance claimed concurrently, not resurrecting", "instanceId", instanceID)
+			return nil, errors.New(awserrors.ErrorIncorrectInstanceState)
+		}
 		slog.ErrorContext(ctx, "ModifyInstanceAttribute: failed to write modified instance to KV",
 			"instanceId", instanceID, "err", err)
 		return nil, errors.New(awserrors.ErrorServerInternal)
@@ -2148,6 +2176,16 @@ func (s *InstanceServiceImpl) StartStoppedInstance(ctx context.Context, input *S
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 	if instance == nil {
+		// Disambiguate a genuinely-absent record from a local race: if this
+		// node's vmMgr already shows the instance running, a winning claim
+		// started it here between another caller's checks and this Load, so
+		// the correct taxonomy is IncorrectInstanceState, not NotFound. A
+		// cross-node winner is invisible to this node's vmMgr and still
+		// falls through to NotFound.
+		if _, ok := s.vmMgr.Get(input.InstanceID); ok {
+			slog.WarnContext(ctx, "StartStoppedInstance: instance already running locally", "instanceId", input.InstanceID)
+			return nil, errors.New(awserrors.ErrorIncorrectInstanceState)
+		}
 		slog.WarnContext(ctx, "StartStoppedInstance: instance not found in shared KV", "instanceId", input.InstanceID)
 		return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
 	}

spinifex/handlers/ec2/instance/service_impl_test.go

@@ -613,6 +613,7 @@ type fakeStoppedStore struct {
 	listTermErr     error
 	loadErr         error
 	writeErr        error
+	updateErr       error
 	writeTermErr    error
 	deleteErr       error
 	deleteFailFirst bool
@@ -625,6 +626,16 @@ type fakeStoppedStore struct {
 	claimedStopped []string
 	claimErr       error
 	claimAttempts  int
+
+	// updateStoppedCalls counts UpdateStoppedInstance invocations for tests
+	// that assert on retry/race behavior.
+	updateStoppedCalls int
+
+	// claimAfterLoad simulates a winning ClaimStoppedInstance landing between
+	// a caller's Load and its later UpdateStoppedInstance: the record is
+	// removed right after LoadStoppedInstance returns it, so the subsequent
+	// UpdateStoppedInstance sees a missing entry exactly like the real CAS.
+	claimAfterLoad bool
 }
 
 func (f *fakeStoppedStore) LoadStoppedInstance(id string) (*vm.VM, error) {
@@ -633,10 +644,15 @@ func (f *fakeStoppedStore) LoadStoppedInstance(id string) (*vm.VM, error) {
 	}
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	if v, ok := f.loadByID[id]; ok {
-		return v, nil
+	v, ok := f.loadByID[id]
+	if !ok {
+		return nil, nil
 	}
-	return nil, nil
+	if f.claimAfterLoad {
+		delete(f.loadByID, id)
+		f.claimedStopped = append(f.claimedStopped, id)
+	}
+	return v, nil
 }
 func (f *fakeStoppedStore) ListStoppedInstances() ([]*vm.VM, error) {
 	if f.listErr != nil {
@@ -696,6 +712,25 @@ func (f *fakeStoppedStore) DeleteStoppedInstance(id string) error {
 }
 func (f *fakeStoppedStore) DeleteTerminatedInstance(string) error { return nil }
 
+// UpdateStoppedInstance mimics the real CAS semantics: mutate runs under the
+// store lock against loadByID's entry, and a missing record (e.g. removed by
+// a concurrent ClaimStoppedInstance) returns nats.ErrKeyNotFound instead of
+// resurrecting it — the same createIfAbsent=false contract as JetStreamManager.
+func (f *fakeStoppedStore) UpdateStoppedInstance(id string, mutate func(*vm.VM)) (*vm.VM, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.updateStoppedCalls++
+	if f.updateErr != nil {
+		return nil, f.updateErr
+	}
+	v, ok := f.loadByID[id]
+	if !ok {
+		return nil, nats.ErrKeyNotFound
+	}
+	mutate(v)
+	return v, nil
+}
+
 // ClaimStoppedInstance mimics the real atomic delete-as-claim: under the
 // store lock, remove and return loadByID[id], or vm.ErrStoppedInstanceClaimed
 // if it is already gone — mirroring the real KV's "revision conflict or
@@ -1317,7 +1352,7 @@ func TestModifyInstanceAttribute_ChangeInstanceType(t *testing.T) {
 	}, "acc")
 	require.NoError(t, err)
 
-	updated := store.wroteStopped[id]
+	updated := store.loadByID[id]
 	require.NotNil(t, updated)
 	assert.Equal(t, "t3.medium", updated.InstanceType)
 	assert.Equal(t, "t3.medium", updated.Config.InstanceType)
@@ -1376,7 +1411,7 @@ func TestModifyInstanceAttribute_ChangeUserData(t *testing.T) {
 	}, "acc")
 	require.NoError(t, err)
 
-	updated := store.wroteStopped[id]
+	updated := store.loadByID[id]
 	require.NotNil(t, updated)
 	assert.Equal(t, "IyEvYmluL2Jhc2g=", *updated.RunInstancesInput.UserData)
 }
@@ -1408,7 +1443,7 @@ func TestModifyInstanceAttribute_ClearsStateReason(t *testing.T) {
 	}, "acc")
 	require.NoError(t, err)
 
-	updated := store.wroteStopped[id]
+	updated := store.loadByID[id]
 	require.NotNil(t, updated)
 	assert.Nil(t, updated.Instance.StateReason)
 }
@@ -1482,7 +1517,7 @@ func TestModifyInstanceAttribute_DisableApiTermination_Stopped(t *testing.T) {
 	}, owner)
 	require.NoError(t, err)
 
-	updated := store.wroteStopped[id]
+	updated := store.loadByID[id]
 	require.NotNil(t, updated)
 	require.NotNil(t, updated.RunInstancesInput)
 	assert.True(t, *updated.RunInstancesInput.DisableApiTermination)
@@ -1504,7 +1539,7 @@ func TestModifyInstanceAttribute_WriteError(t *testing.T) {
 				},
 			},
 		},
-		writeErr: fmt.Errorf("kv write boom"),
+		updateErr: fmt.Errorf("kv write boom"),
 	}
 	svc := &InstanceServiceImpl{stoppedStore: store}
 
@@ -1516,6 +1551,41 @@ func TestModifyInstanceAttribute_WriteError(t *testing.T) {
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 }
 
+// TestModifyInstanceAttribute_ConcurrentClaimDoesNotResurrect covers a
+// ModifyInstanceAttribute racing a winning ClaimStoppedInstance (e.g.
+// StartStoppedInstance) that deletes the record between this call's Load and
+// its CAS write: the update must fail cleanly instead of resurrecting a
+// stale stopped entry.
+func TestModifyInstanceAttribute_ConcurrentClaimDoesNotResurrect(t *testing.T) {
+	id := "i-raced"
+	store := &fakeStoppedStore{
+		loadByID: map[string]*vm.VM{
+			id: {
+				ID:           id,
+				Status:       vm.StateStopped,
+				AccountID:    "acc",
+				InstanceType: "t3.micro",
+				Config:       vm.Config{InstanceType: "t3.micro"},
+				Instance: &ec2.Instance{
+					InstanceId:   aws.String(id),
+					InstanceType: aws.String("t3.micro"),
+				},
+			},
+		},
+		claimAfterLoad: true,
+	}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	_, err := svc.ModifyInstanceAttribute(context.Background(), &ec2.ModifyInstanceAttributeInput{
+		InstanceId:   aws.String(id),
+		InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
+	}, "acc")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error())
+	assert.Empty(t, store.loadByID, "the claimed record must not be resurrected")
+	assert.Equal(t, []string{id}, store.claimedStopped)
+}
+
 // --- TerminateStoppedInstance tests ---
 
 type fakeVolumeDeleter struct {
@@ -1947,6 +2017,28 @@ func TestStartStoppedInstance_NotFound(t *testing.T) {
 	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
 }
 
+// TestStartStoppedInstance_NotFoundButRunningLocally covers the local-race
+// taxonomy fix: when the shared KV has no stopped record for the id but this
+// node's vmMgr already shows it running (a winning claim landed here between
+// another caller's checks and this Load), the correct error is
+// IncorrectInstanceState, not InvalidInstanceID.NotFound — the instance is
+// mid-transition, not gone. A genuinely-absent instance (empty vmMgr) still
+// returns NotFound (TestStartStoppedInstance_NotFound).
+func TestStartStoppedInstance_NotFoundButRunningLocally(t *testing.T) {
+	id := "i-won-locally"
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{}}
+	mgr := vm.NewManager()
+	mgr.Insert(&vm.VM{ID: id, Status: vm.StateRunning, AccountID: "acc"})
+	svc := &InstanceServiceImpl{
+		stoppedStore: store,
+		resourceMgr:  &fakeResourceCapacityProvider{},
+		vmMgr:        mgr,
+	}
+	_, err := svc.StartStoppedInstance(context.Background(), &StartStoppedInstanceInput{InstanceID: id}, "acc")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error())
+}
+
 func TestStartStoppedInstance_NotStopped(t *testing.T) {
 	id := "i-running"
 	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{

spinifex/handlers/ec2/volume/volume_leak_reaper.go

@@ -87,6 +87,14 @@ func (r *VolumeLeakReaper) Sweep(ctx context.Context) (int, error) {
 		// reclamation is the explicit --purge / operator path (ADR-0005 §3).
 		slog.Warn("DATA-SAFETY ALARM: orphaned volume retained, not deleted",
 			"volumeId", id, "attachedInstance", md.AttachedInstance, "sizeGiB", md.SizeGiB)
+
+		// Reconcile the attachment record only — never the volume data. The
+		// owning instance is definitively terminated (this node's leaked
+		// set), so nothing still claims the volume; clearing the stale
+		// attachment lets an operator DeleteVolume it after seeing the alarm.
+		if err := r.svc.UpdateVolumeState(id, "available", "", ""); err != nil {
+			slog.Error("volume-leak: failed to reconcile orphaned volume attachment", "volumeId", id, "err", err)
+		}
 		marked++
 	}
 	return marked, nil

spinifex/handlers/ec2/volume/volume_leak_reaper_test.go

@@ -0,0 +1,69 @@
+package handlers_ec2_volume
+
+import (
+	"context"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/objectstore"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestVolumeLeakReaper_ReconcilesAttachmentNeverDeletesData verifies decision
+// 1(b): once the owning instance is definitively terminated (this node's
+// leaked set), the reaper reconciles ONLY the attachment record — via
+// UpdateVolumeState to available/detached — never the volume data. This
+// unblocks an explicit operator DeleteVolume without weakening ADR-0005 §3's
+// mark-and-alarm, never-delete contract (TestRLC5).
+func TestVolumeLeakReaper_ReconcilesAttachmentNeverDeletesData(t *testing.T) {
+	store := objectstore.NewMemoryObjectStore()
+	svc := newTestVolumeServiceWithStore("ap-southeast-2a", store)
+	seedVolume(t, svc, "vol-leaked-reconcile", "in-use", "i-gone0000000000")
+
+	reaper := svc.NewVolumeLeakReaper(func() (map[string]bool, error) {
+		return map[string]bool{"i-gone0000000000": true}, nil
+	})
+
+	marked, err := reaper.Sweep(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, 1, marked)
+
+	cfg, err := svc.GetVolumeConfig("vol-leaked-reconcile")
+	require.NoError(t, err, "the reaper must never delete the volume, only mark + reconcile it")
+	assert.NotEmpty(t, cfg.VolumeMetadata.Tags[orphanTagKey], "the volume must still be marked orphaned")
+	assert.Equal(t, "available", cfg.VolumeMetadata.State,
+		"the reaper must reconcile the attachment record to available so DeleteVolume can proceed")
+	assert.Empty(t, cfg.VolumeMetadata.AttachedInstance,
+		"the reaper must clear the stale attachment left by the terminated instance")
+
+	// The reconcile only clears control-plane attachment state — this is
+	// exactly the guard DeleteVolume checks (service_impl.go DeleteVolume):
+	// AttachedInstance=="" and State in {"available",""}. Confirm the reaper
+	// leaves the volume passing that guard, without exercising DeleteVolume's
+	// full path here (which needs a snapshotKV this unit fixture omits).
+	assert.True(t, cfg.VolumeMetadata.AttachedInstance == "" && (cfg.VolumeMetadata.State == "available" || cfg.VolumeMetadata.State == ""),
+		"the reconciled volume must pass DeleteVolume's in-use guard so an operator can delete it")
+}
+
+// TestVolumeLeakReaper_LiveOwnerAttachmentUntouched verifies the reconcile is
+// scoped to leaked (terminated, no-live-owner) volumes only: a volume
+// attached to an instance NOT in the leaked set must keep its attachment
+// state exactly as-is.
+func TestVolumeLeakReaper_LiveOwnerAttachmentUntouched(t *testing.T) {
+	store := objectstore.NewMemoryObjectStore()
+	svc := newTestVolumeServiceWithStore("ap-southeast-2a", store)
+	seedVolume(t, svc, "vol-live-reconcile", "in-use", "i-running00000000")
+
+	reaper := svc.NewVolumeLeakReaper(func() (map[string]bool, error) {
+		return map[string]bool{"i-gone0000000000": true}, nil
+	})
+
+	marked, err := reaper.Sweep(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, 0, marked)
+
+	cfg, err := svc.GetVolumeConfig("vol-live-reconcile")
+	require.NoError(t, err)
+	assert.Equal(t, "in-use", cfg.VolumeMetadata.State, "a live-owner volume's attachment must never be reconciled")
+	assert.Equal(t, "i-running00000000", cfg.VolumeMetadata.AttachedInstance)
+}

spinifex/services/viperblockd/viperblockd.go

@@ -475,8 +475,6 @@ func launchService(cfg *Config) (err error) {
 			if volume.Name == ebsRequest.Name {
 				matched = volume
 				matchIdx = i
-				// Remove from slice while we hold the lock
-				cfg.MountedVolumes = append(cfg.MountedVolumes[:i], cfg.MountedVolumes[i+1:]...)
 				break
 			}
 		}
@@ -532,12 +530,29 @@ func launchService(cfg *Config) (err error) {
 					slog.ErrorContext(ctx, "Failed to delete nbd socket", "err", err, "socket", matched.Socket)
 				}
 			}
+
+			// Only drop the volume from MountedVolumes once the seal actually
+			// succeeded (or none was needed): a failed seal must leave it
+			// mounted so a retry re-attempts sealVolumeVB, rather than a
+			// caller seeing "not found" and mistaking a failed seal for a
+			// completed one.
+			if ebsResponse.Error == "" {
+				cfg.mu.Lock()
+				for i, volume := range cfg.MountedVolumes {
+					if volume.Name == matched.Name {
+						cfg.MountedVolumes = append(cfg.MountedVolumes[:i], cfg.MountedVolumes[i+1:]...)
+						break
+					}
+				}
+				cfg.mu.Unlock()
+			}
 		}
 
 		if matchIdx < 0 {
 			ebsResponse = types.EBSUnMountResponse{
-				Volume: ebsRequest.Name,
-				Error:  fmt.Sprintf("Volume %s not found", ebsRequest.Name),
+				Volume:   ebsRequest.Name,
+				Error:    fmt.Sprintf("Volume %s not found", ebsRequest.Name),
+				NotFound: true,
 			}
 		}
 

spinifex/services/viperblockd/viperblockd_integration_test.go

@@ -16,6 +16,7 @@ import (
 	natstest "github.com/nats-io/nats-server/v2/test"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 // setupEmbeddedNATS starts an embedded NATS server for testing
@@ -293,6 +294,102 @@ func TestIntegration_EBSUnmountNonExistentVolume(t *testing.T) {
 	assert.Equal(t, "vol-does-not-exist", response.Volume)
 	assert.NotEmpty(t, response.Error)
 	assert.Contains(t, response.Error, "not found")
+	assert.True(t, response.NotFound, "a volume absent from MountedVolumes must set NotFound so the caller treats the retry as an idempotent success")
+}
+
+// TestIntegration_EBSUnmountRetryAfterCompletedSealReportsNotFound verifies the
+// idempotency contract a slow-but-completed seal relies on: once a volume has
+// been fully unmounted (removed from MountedVolumes), a second ebs.unmount for
+// the same name reports NotFound=true rather than a bare "not found" the
+// caller could mistake for an unrelated failure.
+func TestIntegration_EBSUnmountRetryAfterCompletedSealReportsNotFound(t *testing.T) {
+	t.Parallel()
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	ns, natsURL := setupEmbeddedNATS(t)
+	defer ns.Shutdown()
+
+	cfg := setupTestConfig(t, natsURL)
+	cfg.MountedVolumes = []MountedVolume{
+		{Name: "vol-retry-unmount", PID: 99999},
+	}
+
+	go func() { launchService(cfg) }()
+	time.Sleep(500 * time.Millisecond)
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	requestData, err := json.Marshal(types.EBSRequest{Name: "vol-retry-unmount"})
+	require.NoError(t, err)
+
+	// First call: volume is mounted, no local WAL to seal (not created via
+	// createMockVolumeState), so it completes and is dropped from MountedVolumes.
+	msg, err := nc.Request("ebs.test-node.unmount", requestData, 3*time.Second)
+	require.NoError(t, err)
+	var first types.EBSUnMountResponse
+	require.NoError(t, json.Unmarshal(msg.Data, &first))
+	assert.Empty(t, first.Error)
+	assert.False(t, first.NotFound)
+
+	// Retry: the volume is gone from MountedVolumes, so this must report
+	// NotFound=true — the signal the caller's adapter treats as a completed,
+	// idempotent seal rather than a hard failure.
+	msg, err = nc.Request("ebs.test-node.unmount", requestData, 3*time.Second)
+	require.NoError(t, err)
+	var retry types.EBSUnMountResponse
+	require.NoError(t, json.Unmarshal(msg.Data, &retry))
+	assert.True(t, retry.NotFound, "a retry after a completed unmount must report NotFound")
+}
+
+// TestIntegration_EBSUnmountSealFailureKeepsVolumeMounted verifies the
+// seal-before-remove reorder: when a volume needs sealing (a local WAL
+// directory exists) and the seal fails, the volume must remain in
+// MountedVolumes rather than being dropped — a caller must never see NotFound
+// for a seal that actually failed, since that would flip an unattached
+// volume to available with no durable checkpoint.
+func TestIntegration_EBSUnmountSealFailureKeepsVolumeMounted(t *testing.T) {
+	t.Parallel()
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	ns, natsURL := setupEmbeddedNATS(t)
+	defer ns.Shutdown()
+
+	cfg := setupTestConfig(t, natsURL)
+	// A local WAL directory makes volumeNeedsSeal true; the configured S3 host
+	// is an unreachable mock endpoint, so sealVolumeVB's LoadState fails.
+	createMockVolumeState(t, cfg.BaseDir, "vol-seal-fail")
+	cfg.MountedVolumes = []MountedVolume{
+		{Name: "vol-seal-fail", PID: 99999},
+	}
+
+	go func() { launchService(cfg) }()
+	time.Sleep(500 * time.Millisecond)
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	requestData, err := json.Marshal(types.EBSRequest{Name: "vol-seal-fail"})
+	require.NoError(t, err)
+
+	msg, err := nc.Request("ebs.test-node.unmount", requestData, 5*time.Second)
+	require.NoError(t, err)
+
+	var resp types.EBSUnMountResponse
+	require.NoError(t, json.Unmarshal(msg.Data, &resp))
+	assert.NotEmpty(t, resp.Error, "a seal failure must surface as an error, not a silent success")
+	assert.False(t, resp.NotFound, "a failed seal must not report NotFound")
+
+	cfg.mu.Lock()
+	defer cfg.mu.Unlock()
+	require.Len(t, cfg.MountedVolumes, 1, "a failed seal must leave the volume in MountedVolumes for retry")
+	assert.Equal(t, "vol-seal-fail", cfg.MountedVolumes[0].Name)
 }
 
 // TestIntegration_ConcurrentMountRequests tests multiple concurrent mount requests

spinifex/types/ebs.go

@@ -44,6 +44,10 @@ type EBSUnMountResponse struct {
 	Volume  string `json:"Volume"`
 	Mounted bool   `json:"Mounted"`
 	Error   string `json:"Error"`
+	// NotFound signals the volume was already unmounted and sealed (a retry
+	// after a request that timed out client-side but completed server-side).
+	// The caller treats this as a successful, idempotent seal.
+	NotFound bool `json:"NotFound"`
 }
 
 type EBSSyncRequest struct {