diff --git a/spinifex/handlers/bedrock/service.go b/spinifex/handlers/bedrock/service.go index f16e6c01d..a772b70f9 100644 --- a/spinifex/handlers/bedrock/service.go +++ b/spinifex/handlers/bedrock/service.go @@ -331,7 +331,9 @@ func (s *Service) List(ctx context.Context, _ *ListEndpointsInput, _ string) (*L // Delete moves a READY endpoint to DRAINING, tears down its VM/ENI/weights // volume, then deletes the record. Runs synchronously: unlike Ensure this is // an explicit, infrequent operator action, so there is no client-latency -// reason to background it. An already-ABSENT endpoint is a no-op success. +// reason to background it. An already-ABSENT endpoint is a no-op success, and +// a DRAINING one resumes: a teardown that failed partway must be retryable, +// or the record is stranded in a state nothing else clears. func (s *Service) Delete(ctx context.Context, in *DeleteEndpointInput, _ string) (*DeleteEndpointOutput, error) { if in == nil || in.ModelID == "" { return nil, errors.New(awserrors.ErrorInvalidParameterValue) @@ -353,17 +355,19 @@ func (s *Service) Delete(ctx context.Context, in *DeleteEndpointInput, _ string) if !found { return &DeleteEndpointOutput{}, nil } - if rec.State != StateReady { + if rec.State != StateReady && rec.State != StateDraining { return nil, awserrors.Errorf(awserrors.ErrorModelNotReadyException, - "bedrock: endpoint for %s is not READY (state=%s)", in.ModelID, rec.State) + "bedrock: endpoint for %s cannot be deleted (state=%s)", in.ModelID, rec.State) } - if err := validateTransition(rec.State, StateDraining); err != nil { - return nil, err - } - rec.State = StateDraining - rec.Generation++ - if err := updateJSON(ctx, kv, key, rev, rec); err != nil { - return nil, fmt.Errorf("bedrock: mark endpoint %s draining: %w", in.ModelID, err) + if rec.State == StateReady { + if err := validateTransition(rec.State, StateDraining); err != nil { + return nil, err + } + rec.State = StateDraining + rec.Generation++ + if err := updateJSON(ctx, kv, key, rev, rec); err != nil { + return nil, fmt.Errorf("bedrock: mark endpoint %s draining: %w", in.ModelID, err) + } } if err := TerminateServingVM(ctx, s.deps.Launch, rec); err != nil { diff --git a/spinifex/handlers/bedrock/service_test.go b/spinifex/handlers/bedrock/service_test.go index 7bd7e4dae..dac89291f 100644 --- a/spinifex/handlers/bedrock/service_test.go +++ b/spinifex/handlers/bedrock/service_test.go @@ -220,6 +220,33 @@ func TestDelete_RefusesNonReadyState(t *testing.T) { assert.Equal(t, awserrors.ErrorModelNotReadyException, awserrors.ValidErrorCodeFromError(err)) } +// TestDelete_ResumesFromDraining covers a teardown that failed partway: the +// record is already DRAINING and its VM may be gone. Without a resume the +// record is stranded, since nothing else moves it and the resolver treats +// DRAINING as neither servable nor relaunchable. +func TestDelete_ResumesFromDraining(t *testing.T) { + h := newLaunchHarness() + s, nc := newTestService(t, h, http.StatusOK, sufficientGPU()) + + js := testutil.NewJetStream(t, nc) + kv, err := GetOrCreateEndpointsBucket(t.Context(), js, 1) + require.NoError(t, err) + key := EndpointKey(utils.GlobalAccountID, testModelID) + _, err = createJSONRevision(t.Context(), kv, key, EndpointRecord{ + AccountID: utils.GlobalAccountID, ModelID: testModelID, State: StateDraining, + InstanceID: "i-stranded", Generation: 2, + }) + require.NoError(t, err) + + _, err = s.Delete(t.Context(), &DeleteEndpointInput{ModelID: testModelID}, "") + require.NoError(t, err) + + desc, err := s.Describe(t.Context(), &DescribeEndpointInput{ModelID: testModelID}, "") + require.NoError(t, err) + assert.Equal(t, StateAbsent, desc.Endpoint.State) + assert.Contains(t, h.launcher.terminated, "i-stranded", "the resume must still tear the VM down") +} + func TestList_ReturnsAllEnsuredEndpoints(t *testing.T) { h := newLaunchHarness() s, _ := newTestService(t, h, http.StatusOK, sufficientGPU()) diff --git a/spinifex/vm/shutdown.go b/spinifex/vm/shutdown.go index e93745811..21123aa7e 100644 --- a/spinifex/vm/shutdown.go +++ b/spinifex/vm/shutdown.go @@ -242,9 +242,15 @@ func (m *Manager) finalizeTerminated(instance *VM) error { transitionErr := m.transitionWithPrecheck(instance, StateTerminated) if transitionErr != nil && errors.Is(transitionErr, ErrInvalidTransition) { - // A genuine invalid/raced transition: in-memory status never reached - // terminated, so there is nothing durable to record yet. - return fmt.Errorf("transition to terminated: %w", transitionErr) + if m.Status(instance) != StateTerminated { + // A genuine invalid/raced transition: in-memory status never reached + // terminated, so there is nothing durable to record yet. + return fmt.Errorf("transition to terminated: %w", transitionErr) + } + // Another handler (typically the VM GC) finalised this instance while we + // were tearing it down. Its destination is the one we wanted, so fall + // through: everything below is idempotent. + transitionErr = nil } if transitionErr != nil { // In-memory status reached terminated even though local persistence diff --git a/spinifex/vm/shutdown_gcrace_test.go b/spinifex/vm/shutdown_gcrace_test.go new file mode 100644 index 000000000..51c16b4c8 --- /dev/null +++ b/spinifex/vm/shutdown_gcrace_test.go @@ -0,0 +1,91 @@ +package vm + +import ( + "testing" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// gcRacingCleaner drives an instance to terminated from inside the cleanup +// chain, standing in for the VM GC reconciling the same instance while a +// terminate is in flight. +type gcRacingCleaner struct { + recordingInstanceCleaner + + onReleaseGPU func(v *VM) +} + +func (c *gcRacingCleaner) ReleaseGPU(v *VM) error { + if c.onReleaseGPU != nil { + c.onReleaseGPU(v) + } + return c.recordingInstanceCleaner.ReleaseGPU(v) +} + +// TestTerminate_GCFinalisedDuringCleanup_Converges covers the race observed on +// wattle: the GC finalises the instance while Terminate is inside +// terminateCleanup, so finalizeTerminated's precheck sees terminated -> +// terminated. Reaching the state we were driving to is convergence, not +// failure — returning an error there aborted the caller's own teardown +// (an Ochre endpoint delete) partway and stranded its record. +func TestTerminate_GCFinalisedDuringCleanup_Converges(t *testing.T) { + t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) + + store := newFakeStateStore() + rt := &recordedTransitions{} + cleaner := &gcRacingCleaner{} + m := NewManager() + rt.bind(m) + m.SetDeps(Deps{ + NodeID: "test-node", + StateStore: store, + VolumeMounter: &fakeVolumeMounter{}, + InstanceCleaner: cleaner, + TransitionState: rt.apply, + ShutdownSignal: func() bool { return false }, + }) + + v := &VM{ID: "i-gc-raced", Status: StateRunning, InstanceType: "t3.micro", Instance: &ec2.Instance{}} + m.Insert(v) + cleaner.onReleaseGPU = func(raced *VM) { + m.Inspect(raced, func(x *VM) { x.Status = StateTerminated }) + } + + require.NoError(t, m.Terminate(v.ID), "a concurrent finalisation must not fail the terminate") + + assert.Equal(t, StateTerminated, m.Status(v)) + require.NotNil(t, store.terminated[v.ID], "the durable record must still land") + _, stillInMap := m.Get(v.ID) + assert.False(t, stillInMap, "the instance must leave the local map") +} + +// TestFinalizeTerminated_GenuineInvalidTransitionStillFails keeps the guard +// narrow: only an instance that actually reached terminated converges. One +// that did not is still a real refusal. +func TestFinalizeTerminated_GenuineInvalidTransitionStillFails(t *testing.T) { + t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) + + store := newFakeStateStore() + rt := &recordedTransitions{} + m := NewManager() + rt.bind(m) + m.SetDeps(Deps{ + NodeID: "test-node", + StateStore: store, + VolumeMounter: &fakeVolumeMounter{}, + InstanceCleaner: &recordingInstanceCleaner{}, + TransitionState: rt.apply, + ShutdownSignal: func() bool { return false }, + }) + + v := &VM{ID: "i-not-terminated", Status: StatePending, Instance: &ec2.Instance{}} + m.Insert(v) + + err := m.finalizeTerminated(v) + + require.Error(t, err) + assert.ErrorIs(t, err, ErrInvalidTransition) + assert.Nil(t, store.terminated[v.ID], "nothing durable may be recorded for a refused transition") +}