From 466dbf1652c5a9c8fc3b3f46f3610b7bfe3e2e88 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Fri, 10 Jul 2026 07:57:05 +1000 Subject: [PATCH] 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. --- .../daemon_handlers_instance_tags_test.go | 12 ++ .../daemon/daemon_handlers_instance_test.go | 21 +++- spinifex/daemon/daemon_test.go | 47 ++++++++ spinifex/daemon/jetstream.go | 29 +++++ spinifex/daemon/jetstream_test.go | 100 ++++++++++++++++ spinifex/daemon/vm_adapters.go | 12 +- .../ec2/instance/instance_tags_test.go | 25 +++- spinifex/handlers/ec2/instance/service.go | 5 + .../handlers/ec2/instance/service_impl.go | 80 +++++++++---- .../ec2/instance/service_impl_test.go | 108 ++++++++++++++++-- .../handlers/ec2/volume/volume_leak_reaper.go | 8 ++ .../ec2/volume/volume_leak_reaper_test.go | 69 +++++++++++ spinifex/services/viperblockd/viperblockd.go | 23 +++- .../viperblockd_integration_test.go | 97 ++++++++++++++++ spinifex/types/ebs.go | 4 + 15 files changed, 603 insertions(+), 37 deletions(-) create mode 100644 spinifex/handlers/ec2/volume/volume_leak_reaper_test.go diff --git a/spinifex/daemon/daemon_handlers_instance_tags_test.go b/spinifex/daemon/daemon_handlers_instance_tags_test.go index 234edc290..22d8a698d 100644 --- a/spinifex/daemon/daemon_handlers_instance_tags_test.go +++ b/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 { diff --git a/spinifex/daemon/daemon_handlers_instance_test.go b/spinifex/daemon/daemon_handlers_instance_test.go index e3bf3376a..f273cb326 100644 --- a/spinifex/daemon/daemon_handlers_instance_test.go +++ b/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) diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index 6a440f9f4..5807cb0b6 100644 --- a/spinifex/daemon/daemon_test.go +++ b/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 diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index c249ca109..abdb1ecf2 100644 --- a/spinifex/daemon/jetstream.go +++ b/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 { diff --git a/spinifex/daemon/jetstream_test.go b/spinifex/daemon/jetstream_test.go index c7f64e52b..e5a6f8280 100644 --- a/spinifex/daemon/jetstream_test.go +++ b/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) diff --git a/spinifex/daemon/vm_adapters.go b/spinifex/daemon/vm_adapters.go index fbbea414f..cf28e6a72 100644 --- a/spinifex/daemon/vm_adapters.go +++ b/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) } diff --git a/spinifex/handlers/ec2/instance/instance_tags_test.go b/spinifex/handlers/ec2/instance/instance_tags_test.go index 19345bc71..a3c5c09e2 100644 --- a/spinifex/handlers/ec2/instance/instance_tags_test.go +++ b/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}} diff --git a/spinifex/handlers/ec2/instance/service.go b/spinifex/handlers/ec2/instance/service.go index f69ccbcd3..888c7ad0b 100644 --- a/spinifex/handlers/ec2/instance/service.go +++ b/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 diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index aee44496e..dcad2204c 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/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) } diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index 7c99c5f20..1e306672f 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/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{ diff --git a/spinifex/handlers/ec2/volume/volume_leak_reaper.go b/spinifex/handlers/ec2/volume/volume_leak_reaper.go index fdffbc013..20fc64db2 100644 --- a/spinifex/handlers/ec2/volume/volume_leak_reaper.go +++ b/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 diff --git a/spinifex/handlers/ec2/volume/volume_leak_reaper_test.go b/spinifex/handlers/ec2/volume/volume_leak_reaper_test.go new file mode 100644 index 000000000..ef5db3cfd --- /dev/null +++ b/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) +} diff --git a/spinifex/services/viperblockd/viperblockd.go b/spinifex/services/viperblockd/viperblockd.go index 26efb666c..bac233e9b 100644 --- a/spinifex/services/viperblockd/viperblockd.go +++ b/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, } } diff --git a/spinifex/services/viperblockd/viperblockd_integration_test.go b/spinifex/services/viperblockd/viperblockd_integration_test.go index 362421927..7d20ee2d7 100644 --- a/spinifex/services/viperblockd/viperblockd_integration_test.go +++ b/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 diff --git a/spinifex/types/ebs.go b/spinifex/types/ebs.go index 9886311cd..da8e9aa82 100644 --- a/spinifex/types/ebs.go +++ b/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 {