diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 73b8119f0..1bc8bf78d 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -575,7 +575,7 @@ func (d *Daemon) restoreInstances() { instance := d.Instances.VMS[i] - if instance.Status == "terminated" { + if instance.Status == vm.StateTerminated { slog.Info("Instance state is terminated, skipping", "instance", instance.ID) continue } @@ -813,7 +813,8 @@ func (d *Daemon) ClusterManager() error { return nil } -// WriteState writes the instance state to JetStream KV store (required) +// WriteState writes the instance state to JetStream KV store (required). +// It acquires d.Instances.Mu internally. func (d *Daemon) WriteState() error { if d.jsManager == nil { return fmt.Errorf("JetStream manager not initialized - cannot write state") @@ -825,6 +826,7 @@ func (d *Daemon) WriteState() error { return nil } + // LoadState loads the instance state from JetStream KV store (required) func (d *Daemon) LoadState() error { if d.jsManager == nil { @@ -867,36 +869,36 @@ func (d *Daemon) SendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceId if _, ok := msg["event"]; ok { slog.Info("QMP event", "event", msg["event"]) - var updatedStatus string + // Look up the instance to handle state transitions + d.Instances.Mu.Lock() + instance, found := d.Instances.VMS[instanceId] + d.Instances.Mu.Unlock() + if !found { + slog.Info("QMP event - instance not found", "id", instanceId) + continue + } + // Handle QMP events with state machine transitions. + // TransitionState validates the current state atomically under lock, + // so no pre-check is needed here. + // RESET indicates a reboot but the VM remains running — no state change. switch msg["event"] { case "STOP": - updatedStatus = "stopped" - case "RESUME": - updatedStatus = "resuming" - case "RESET": - updatedStatus = "restarting" - case "POWERDOWN": - updatedStatus = "powering_down" - } - - if updatedStatus != "" { - - // Update the instance status - d.Instances.Mu.Lock() - instance, ok := d.Instances.VMS[instanceId] - if !ok { - slog.Info("QMP Status - Instance not found", "id", instanceId) - continue + if err := d.TransitionState(instance, vm.StateStopped); err != nil { + slog.Error("QMP STOP event: transition failed", "instanceId", instanceId, "err", err) } - - instance.Status = updatedStatus - - d.Instances.VMS[instanceId] = instance - d.Instances.Mu.Unlock() + case "RESUME": + if err := d.TransitionState(instance, vm.StateRunning); err != nil { + slog.Error("QMP RESUME event: transition failed", "instanceId", instanceId, "err", err) + } + case "POWERDOWN": + if err := d.TransitionState(instance, vm.StateStopping); err != nil { + slog.Error("QMP POWERDOWN event: transition failed", "instanceId", instanceId, "err", err) + } + case "RESET": + slog.Info("QMP RESET event", "instanceId", instanceId) } - // Optional: handle events here continue } if errObj, ok := msg["error"].(map[string]any); ok { @@ -1099,7 +1101,7 @@ func (d *Daemon) CreateQMPClient(instance *vm.VM) (err error) { status := instance.Status d.Instances.Mu.Unlock() - if status == "stopping" || status == "stopped" || status == "shutting-down" || status == "terminated" || status == "error" { + if status == vm.StateStopping || status == vm.StateStopped || status == vm.StateShuttingDown || status == vm.StateTerminated || status == vm.StateError { slog.Info("QMP heartbeat exiting - instance not running", "instance", instance.ID, "status", status) // Close the QMP client connection if it exists @@ -1186,21 +1188,15 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) { } // Step 9: Update the instance metadata for running state and volume attached - // Marshal to a JSON file - // Update state d.Instances.Mu.Lock() - // Update to running state - instance.Status = "running" - - // Update EC2 Instance state for API compatibility - if instance.Instance != nil { - instance.Instance.State.SetCode(16) // 16 = running - instance.Instance.State.SetName("running") - } - d.Instances.VMS[instance.ID] = instance d.Instances.Mu.Unlock() + if err := d.TransitionState(instance, vm.StateRunning); err != nil { + slog.Error("Failed to transition instance to running", "instanceId", instance.ID, "err", err) + return err + } + // Step 10: Mark boot volumes as "in-use" now that instance is confirmed running instance.EBSRequests.Mu.Lock() for _, ebsReq := range instance.EBSRequests.Requests { @@ -1212,37 +1208,22 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) { } instance.EBSRequests.Mu.Unlock() - err = d.WriteState() - - if err != nil { - slog.Error("Failed to marshal launchVm", "err", err) - return err - } - return nil } // markInstanceFailed updates an instance status to indicate a failure during launch func (d *Daemon) markInstanceFailed(instance *vm.VM, reason string) { + // Set state reason before transition (requires lock) d.Instances.Mu.Lock() - defer d.Instances.Mu.Unlock() - - instance.Status = "shutting-down" - - // Update EC2 Instance state for API compatibility if instance.Instance != nil { - instance.Instance.State.SetCode(32) // 32 = shutting-down - instance.Instance.State.SetName("shutting-down") - // Add state reason instance.Instance.StateReason = &ec2.StateReason{} instance.Instance.StateReason.SetCode("Server.InternalError") instance.Instance.StateReason.SetMessage(reason) } + d.Instances.Mu.Unlock() - d.Instances.VMS[instance.ID] = instance - - if err := d.WriteState(); err != nil { - slog.Error("markInstanceFailed failed to write state", "err", err) + if err := d.TransitionState(instance, vm.StateShuttingDown); err != nil { + slog.Error("markInstanceFailed transition failed", "instanceId", instance.ID, "err", err) } slog.Info("Instance marked as failed", "instanceId", instance.ID, "reason", reason) diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index eab44154c..2d1ffd683 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -33,9 +33,8 @@ func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) { // Check if the instance is running on this node d.Instances.Mu.Lock() - defer d.Instances.Mu.Unlock() - instance, ok := d.Instances.VMS[ec2StartInstance.InstanceID] + d.Instances.Mu.Unlock() if !ok { slog.Error("EC2 Start Request - Instance not found", "instanceId", ec2StartInstanceResponse.InstanceID) @@ -57,7 +56,7 @@ func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) { } } - // Launch the instance + // Launch the instance (acquires d.Instances.Mu internally via TransitionState) err := d.LaunchInstance(instance) if err != nil { @@ -68,7 +67,9 @@ func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) { ec2StartInstanceResponse.Error = err.Error() } else { ec2StartInstanceResponse.InstanceID = instance.ID - ec2StartInstanceResponse.Status = instance.Status + d.Instances.Mu.Lock() + ec2StartInstanceResponse.Status = string(instance.Status) + d.Instances.Mu.Unlock() } ec2StartInstanceResponse.Respond(msg) @@ -132,23 +133,13 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { return } - // Update instance state + // Update instance attributes (LaunchInstance already transitions to StateRunning) d.Instances.Mu.Lock() - instance.Status = "running" instance.Attributes = command.Attributes - if instance.Instance != nil { - instance.Instance.State.SetCode(16) // 16 = running - instance.Instance.State.SetName("running") - } d.Instances.Mu.Unlock() slog.Info("Instance started", "instanceId", instance.ID) - // Write state to disk - if writeErr := d.WriteState(); writeErr != nil { - slog.Error("Failed to write state to disk", "err", writeErr) - } - msg.Respond(fmt.Appendf(nil, `{"status":"running","instanceId":"%s"}`, instance.ID)) return } @@ -157,31 +148,22 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { if command.Attributes.StopInstance || command.Attributes.TerminateInstance { isTerminate := command.Attributes.TerminateInstance action := "Stopping" - initialStatus := "stopping" - finalStatus := "stopped" - finalCode := int64(80) + initialState := vm.StateStopping + finalState := vm.StateStopped if isTerminate { action = "Terminating" - initialStatus = "shutting-down" - finalStatus = "terminated" - finalCode = 48 + initialState = vm.StateShuttingDown + finalState = vm.StateTerminated } slog.Info(action+" instance", "id", command.ID) - // Update status to transitional state - d.Instances.Mu.Lock() - instance.Status = initialStatus - if instance.Instance != nil { - if isTerminate { - instance.Instance.State.SetCode(32) // 32 = shutting-down - instance.Instance.State.SetName("shutting-down") - } else { - instance.Instance.State.SetCode(64) // 64 = stopping - instance.Instance.State.SetName("stopping") - } + // Transition to the initial transitional state + if err := d.TransitionState(instance, initialState); err != nil { + slog.Error("Failed to transition to "+string(initialState), "instanceId", instance.ID, "err", err) + respondWithError(awserrors.ErrorServerInternal) + return } - d.Instances.Mu.Unlock() // Respond immediately - operation will complete in background // stopInstance() handles the QMP shutdown command, so we don't send it here @@ -191,28 +173,20 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { go func(inst *vm.VM, attrs qmp.Attributes) { stopErr := d.stopInstance(map[string]*vm.VM{inst.ID: inst}, isTerminate) - d.Instances.Mu.Lock() if stopErr != nil { slog.Error("Failed to "+strings.ToLower(action)+" instance", "err", stopErr, "id", inst.ID) - // On error, revert to previous state or mark as error - inst.Status = "error" - if inst.Instance != nil { - inst.Instance.State.SetCode(0) - inst.Instance.State.SetName("error") + if err := d.TransitionState(inst, vm.StateError); err != nil { + slog.Error("Failed to transition to error state", "instanceId", inst.ID, "err", err) } } else { - inst.Status = finalStatus + d.Instances.Mu.Lock() inst.Attributes = attrs - if inst.Instance != nil { - inst.Instance.State.SetCode(finalCode) - inst.Instance.State.SetName(finalStatus) - } - slog.Info("Instance "+finalStatus, "id", inst.ID) - } - d.Instances.Mu.Unlock() + d.Instances.Mu.Unlock() - if writeErr := d.WriteState(); writeErr != nil { - slog.Error("Failed to write state to disk", "err", writeErr) + if err := d.TransitionState(inst, finalState); err != nil { + slog.Error("Failed to transition to final state", "instanceId", inst.ID, "err", err) + } + slog.Info("Instance "+string(finalState), "id", inst.ID) } }(instance, command.Attributes) @@ -915,27 +889,13 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { instanceCopy := *instance.Instance instanceCopy.State = &ec2.InstanceState{} - // Map internal status to EC2 state codes - switch instance.Status { - case "pending", "provisioning": - instanceCopy.State.SetCode(0) - instanceCopy.State.SetName("pending") - case "running": - instanceCopy.State.SetCode(16) - instanceCopy.State.SetName("running") - case "stopping": - instanceCopy.State.SetCode(64) - instanceCopy.State.SetName("stopping") - case "stopped": - instanceCopy.State.SetCode(80) - instanceCopy.State.SetName("stopped") - case "shutting-down": - instanceCopy.State.SetCode(32) - instanceCopy.State.SetName("shutting-down") - case "terminated": - instanceCopy.State.SetCode(48) - instanceCopy.State.SetName("terminated") - default: + // Map internal status to EC2 state codes using the centralized mapping + if info, ok := vm.EC2StateCodes[instance.Status]; ok { + instanceCopy.State.SetCode(info.Code) + instanceCopy.State.SetName(info.Name) + } else { + slog.Warn("Instance has unmapped status, reporting as pending", + "instanceId", instance.ID, "status", string(instance.Status)) instanceCopy.State.SetCode(0) instanceCopy.State.SetName("pending") } diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index cc5dff290..533e60ca3 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -941,19 +941,19 @@ func TestDaemon_BootAllocation(t *testing.T) { "i-running": { ID: "i-running", InstanceType: getTestInstanceType(), - Status: "running", + Status: vm.StateRunning, Attributes: qmp.Attributes{StopInstance: false}, }, "i-stopped": { ID: "i-stopped", InstanceType: getTestInstanceType(), - Status: "stopped", + Status: vm.StateStopped, Attributes: qmp.Attributes{StopInstance: true}, }, "i-terminated": { ID: "i-terminated", InstanceType: getTestInstanceType(), - Status: "terminated", + Status: vm.StateTerminated, Attributes: qmp.Attributes{StopInstance: false}, }, } @@ -988,7 +988,7 @@ func TestDaemon_BootAllocation(t *testing.T) { // Simulate the allocation loop in Start() for _, instance := range daemon.Instances.VMS { - if instance.Status != "terminated" && !instance.Attributes.StopInstance { + if instance.Status != vm.StateTerminated && !instance.Attributes.StopInstance { instanceType, ok := daemon.resourceMgr.instanceTypes[instance.InstanceType] if ok { err := daemon.resourceMgr.allocate(instanceType) @@ -1024,7 +1024,7 @@ func TestHandleEC2StartInstances_Allocation(t *testing.T) { daemon.Instances.VMS[instanceId] = &vm.VM{ ID: instanceId, InstanceType: instanceType, - Status: "stopped", + Status: vm.StateStopped, } // Subscribe to ec2.startinstances @@ -1070,7 +1070,7 @@ func TestStopInstance_Deallocation(t *testing.T) { daemon.Instances.VMS[instanceId] = &vm.VM{ ID: instanceId, InstanceType: instanceTypeStr, - Status: "running", + Status: vm.StateRunning, } err := daemon.resourceMgr.allocate(instanceType) @@ -1254,7 +1254,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { daemon.Instances.VMS[instanceID] = &vm.VM{ ID: instanceID, - Status: "running", + Status: vm.StateRunning, Reservation: reservation1, Instance: ec2Instance, } @@ -1273,7 +1273,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { daemon.Instances.VMS[instanceID] = &vm.VM{ ID: instanceID, - Status: "running", + Status: vm.StateRunning, Reservation: reservation2, Instance: ec2Instance, } @@ -1290,7 +1290,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { daemon.Instances.VMS["i-single-001"] = &vm.VM{ ID: "i-single-001", - Status: "stopped", + Status: vm.StateStopped, Reservation: reservation3, Instance: ec2Instance, } diff --git a/hive/daemon/jetstream.go b/hive/daemon/jetstream.go index c242d60e9..96a696619 100644 --- a/hive/daemon/jetstream.go +++ b/hive/daemon/jetstream.go @@ -65,15 +65,16 @@ func (m *JetStreamManager) InitKVBucket() error { return nil } -// WriteState writes the instance state to the KV store for the given node +// WriteState writes the instance state to the KV store for the given node. +// It acquires instances.Mu internally. func (m *JetStreamManager) WriteState(nodeID string, instances *vm.Instances) error { + instances.Mu.Lock() + defer instances.Mu.Unlock() + if m.kv == nil { return errors.New("KV bucket not initialized") } - instances.Mu.Lock() - defer instances.Mu.Unlock() - // Create a struct without the mutex to avoid copying the lock state := struct { VMS map[string]*vm.VM `json:"vms"` diff --git a/hive/daemon/jetstream_test.go b/hive/daemon/jetstream_test.go index e8c78da94..6aab5d842 100644 --- a/hive/daemon/jetstream_test.go +++ b/hive/daemon/jetstream_test.go @@ -73,12 +73,12 @@ func TestJetStreamManager_WriteAndLoadState(t *testing.T) { VMS: map[string]*vm.VM{ "i-test-001": { ID: "i-test-001", - Status: "running", + Status: vm.StateRunning, InstanceType: "t3.micro", }, "i-test-002": { ID: "i-test-002", - Status: "stopped", + Status: vm.StateStopped, InstanceType: "t3.small", }, }, @@ -97,8 +97,8 @@ func TestJetStreamManager_WriteAndLoadState(t *testing.T) { assert.Len(t, loadedInstances.VMS, 2, "Should have 2 instances") assert.NotNil(t, loadedInstances.VMS["i-test-001"], "Should have i-test-001") assert.NotNil(t, loadedInstances.VMS["i-test-002"], "Should have i-test-002") - assert.Equal(t, "running", loadedInstances.VMS["i-test-001"].Status) - assert.Equal(t, "stopped", loadedInstances.VMS["i-test-002"].Status) + assert.Equal(t, vm.StateRunning, loadedInstances.VMS["i-test-001"].Status) + assert.Equal(t, vm.StateStopped, loadedInstances.VMS["i-test-002"].Status) assert.Equal(t, "t3.micro", loadedInstances.VMS["i-test-001"].InstanceType) assert.Equal(t, "t3.small", loadedInstances.VMS["i-test-002"].InstanceType) } @@ -165,7 +165,7 @@ func TestJetStreamManager_BucketReconnection(t *testing.T) { VMS: map[string]*vm.VM{ "i-persist": { ID: "i-persist", - Status: "running", + Status: vm.StateRunning, }, }, } @@ -190,7 +190,7 @@ func TestJetStreamManager_BucketReconnection(t *testing.T) { require.NoError(t, err) assert.NotEmpty(t, loadedInstances.VMS, "Should have persisted instances") assert.NotNil(t, loadedInstances.VMS["i-persist"], "Should have i-persist") - assert.Equal(t, "running", loadedInstances.VMS["i-persist"].Status) + assert.Equal(t, vm.StateRunning, loadedInstances.VMS["i-persist"].Status) } // TestJetStreamManager_DeleteState tests deleting state from the KV store @@ -214,7 +214,7 @@ func TestJetStreamManager_DeleteState(t *testing.T) { VMS: map[string]*vm.VM{ "i-delete-me": { ID: "i-delete-me", - Status: "running", + Status: vm.StateRunning, }, }, } @@ -278,7 +278,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) { VMS: map[string]*vm.VM{ "i-initial": { ID: "i-initial", - Status: "running", + Status: vm.StateRunning, }, }, } @@ -290,11 +290,11 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) { VMS: map[string]*vm.VM{ "i-initial": { ID: "i-initial", - Status: "stopped", // Changed status + Status: vm.StateStopped, // Changed status }, "i-new": { // Added new instance ID: "i-new", - Status: "running", + Status: vm.StateRunning, }, }, } @@ -305,7 +305,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) { loadedInstances, err := jsm.LoadState(testNodeID) require.NoError(t, err) assert.Len(t, loadedInstances.VMS, 2, "Should have 2 instances") - assert.Equal(t, "stopped", loadedInstances.VMS["i-initial"].Status, "Status should be updated") + assert.Equal(t, vm.StateStopped, loadedInstances.VMS["i-initial"].Status, "Status should be updated") assert.NotNil(t, loadedInstances.VMS["i-new"], "Should have new instance") } @@ -327,7 +327,7 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) { // Write state for node-1 node1Instances := &vm.Instances{ VMS: map[string]*vm.VM{ - "i-node1-001": {ID: "i-node1-001", Status: "running"}, + "i-node1-001": {ID: "i-node1-001", Status: vm.StateRunning}, }, } err = jsm.WriteState("node-1", node1Instances) @@ -336,8 +336,8 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) { // Write state for node-2 node2Instances := &vm.Instances{ VMS: map[string]*vm.VM{ - "i-node2-001": {ID: "i-node2-001", Status: "stopped"}, - "i-node2-002": {ID: "i-node2-002", Status: "running"}, + "i-node2-001": {ID: "i-node2-001", Status: vm.StateStopped}, + "i-node2-002": {ID: "i-node2-002", Status: vm.StateRunning}, }, } err = jsm.WriteState("node-2", node2Instances) diff --git a/hive/daemon/state.go b/hive/daemon/state.go new file mode 100644 index 000000000..2f06bc777 --- /dev/null +++ b/hive/daemon/state.go @@ -0,0 +1,34 @@ +package daemon + +import ( + "fmt" + "log/slog" + + "github.com/mulgadc/hive/hive/vm" +) + +// TransitionState validates and applies a state transition on the given instance. +// It sets instance.Status, persists via WriteState, and logs. +// On validation failure, the instance status is unchanged and an error is returned. +// If WriteState fails, the in-memory status retains the new value (the VM has +// physically changed state regardless) and an error is returned. +// The caller must NOT hold d.Instances.Mu; this method acquires it internally. +func (d *Daemon) TransitionState(instance *vm.VM, target vm.InstanceState) error { + d.Instances.Mu.Lock() + current := instance.Status + if !vm.IsValidTransition(current, target) { + d.Instances.Mu.Unlock() + return fmt.Errorf("invalid state transition: %s -> %s for instance %s", current, target, instance.ID) + } + instance.Status = target + d.Instances.Mu.Unlock() + + slog.Info("Instance state transition", "instanceId", instance.ID, "from", string(current), "to", string(target)) + + if err := d.WriteState(); err != nil { + slog.Error("Failed to persist state after transition", "instanceId", instance.ID, + "from", string(current), "to", string(target), "err", err) + return fmt.Errorf("state transition applied but write failed: %w", err) + } + return nil +} diff --git a/hive/daemon/state_test.go b/hive/daemon/state_test.go new file mode 100644 index 000000000..62fd72d81 --- /dev/null +++ b/hive/daemon/state_test.go @@ -0,0 +1,170 @@ +package daemon + +import ( + "os" + "testing" + "time" + + "github.com/mulgadc/hive/hive/config" + "github.com/mulgadc/hive/hive/vm" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// createDaemonWithJetStream creates a daemon backed by an in-process NATS+JetStream server +// so that TransitionState (which calls WriteState) works end-to-end. +func createDaemonWithJetStream(t *testing.T) *Daemon { + t.Helper() + + jsTmpDir, err := os.MkdirTemp("", "nats-js-state-test-*") + require.NoError(t, err) + t.Cleanup(func() { os.RemoveAll(jsTmpDir) }) + + opts := &server.Options{ + Host: "127.0.0.1", + Port: -1, + JetStream: true, + StoreDir: jsTmpDir, + NoLog: true, + NoSigs: true, + } + ns, err := server.NewServer(opts) + require.NoError(t, err) + go ns.Start() + require.True(t, ns.ReadyForConnections(5*time.Second), "NATS server failed to start") + t.Cleanup(func() { ns.Shutdown() }) + + natsURL := ns.ClientURL() + + tmpDir, err := os.MkdirTemp("", "hive-state-test-*") + require.NoError(t, err) + t.Cleanup(func() { os.RemoveAll(tmpDir) }) + + clusterCfg := &config.ClusterConfig{ + Node: "node-1", + Nodes: map[string]config.Config{"node-1": {BaseDir: tmpDir}}, + } + daemon := NewDaemon(clusterCfg) + daemon.config = &config.Config{BaseDir: tmpDir} + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + t.Cleanup(func() { nc.Close() }) + + daemon.natsConn = nc + daemon.jsManager, err = NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, daemon.jsManager.InitKVBucket()) + + return daemon +} + +func TestTransitionState_ValidTransitions(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + tests := []struct { + name string + from vm.InstanceState + to vm.InstanceState + }{ + {"provisioning->running", vm.StateProvisioning, vm.StateRunning}, + {"provisioning->error", vm.StateProvisioning, vm.StateError}, + {"provisioning->shutting-down", vm.StateProvisioning, vm.StateShuttingDown}, + {"pending->running", vm.StatePending, vm.StateRunning}, + {"pending->error", vm.StatePending, vm.StateError}, + {"pending->shutting-down", vm.StatePending, vm.StateShuttingDown}, + {"running->stopping", vm.StateRunning, vm.StateStopping}, + {"running->shutting-down", vm.StateRunning, vm.StateShuttingDown}, + {"running->error", vm.StateRunning, vm.StateError}, + {"stopping->stopped", vm.StateStopping, vm.StateStopped}, + {"stopping->shutting-down", vm.StateStopping, vm.StateShuttingDown}, + {"stopping->error", vm.StateStopping, vm.StateError}, + {"stopped->running", vm.StateStopped, vm.StateRunning}, + {"stopped->shutting-down", vm.StateStopped, vm.StateShuttingDown}, + {"stopped->error", vm.StateStopped, vm.StateError}, + {"shutting-down->terminated", vm.StateShuttingDown, vm.StateTerminated}, + {"shutting-down->error", vm.StateShuttingDown, vm.StateError}, + {"error->running", vm.StateError, vm.StateRunning}, + {"error->shutting-down", vm.StateError, vm.StateShuttingDown}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + instance := &vm.VM{ + ID: "i-test-valid", + Status: tt.from, + } + + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(instance, tt.to) + require.NoError(t, err) + + assert.Equal(t, tt.to, instance.Status) + }) + } +} + +func TestTransitionState_InvalidTransitions(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + tests := []struct { + name string + from vm.InstanceState + to vm.InstanceState + }{ + {"running->running", vm.StateRunning, vm.StateRunning}, + {"running->pending", vm.StateRunning, vm.StatePending}, + {"running->stopped", vm.StateRunning, vm.StateStopped}, + {"running->terminated", vm.StateRunning, vm.StateTerminated}, + {"stopped->stopping", vm.StateStopped, vm.StateStopping}, + {"stopped->terminated", vm.StateStopped, vm.StateTerminated}, + {"terminated->running", vm.StateTerminated, vm.StateRunning}, + {"terminated->stopped", vm.StateTerminated, vm.StateStopped}, + {"stopping->running", vm.StateStopping, vm.StateRunning}, + {"shutting-down->running", vm.StateShuttingDown, vm.StateRunning}, + {"shutting-down->stopped", vm.StateShuttingDown, vm.StateStopped}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + instance := &vm.VM{ + ID: "i-test-invalid", + Status: tt.from, + } + + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(instance, tt.to) + assert.Error(t, err) + assert.Contains(t, err.Error(), "invalid state transition") + + // Status should remain unchanged + assert.Equal(t, tt.from, instance.Status) + }) + } +} + +func TestTransitionState_NilEC2Instance(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + instance := &vm.VM{ + ID: "i-test-nil-ec2", + Status: vm.StateProvisioning, + Instance: nil, // no EC2 instance metadata + } + + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(instance, vm.StateRunning) + require.NoError(t, err) + assert.Equal(t, vm.StateRunning, instance.Status) +} diff --git a/hive/handlers/ec2/instance/service_impl.go b/hive/handlers/ec2/instance/service_impl.go index 7c7ed8691..e52c60356 100644 --- a/hive/handlers/ec2/instance/service_impl.go +++ b/hive/handlers/ec2/instance/service_impl.go @@ -116,7 +116,7 @@ func (s *InstanceServiceImpl) RunInstance(input *ec2.RunInstancesInput) (*vm.VM, // Create new instance structure instance := &vm.VM{ ID: instanceId, - Status: "provisioning", + Status: vm.StateProvisioning, InstanceType: *input.InstanceType, } diff --git a/hive/vm/state.go b/hive/vm/state.go new file mode 100644 index 000000000..00c080191 --- /dev/null +++ b/hive/vm/state.go @@ -0,0 +1,61 @@ +package vm + +// InstanceState represents a typed VM lifecycle state. +type InstanceState string + +const ( + StateProvisioning InstanceState = "provisioning" + StatePending InstanceState = "pending" + StateRunning InstanceState = "running" + StateStopping InstanceState = "stopping" + StateStopped InstanceState = "stopped" + StateShuttingDown InstanceState = "shutting-down" + StateTerminated InstanceState = "terminated" + StateError InstanceState = "error" +) + +// EC2StateInfo holds the EC2 API code and name for a given instance state. +type EC2StateInfo struct { + Code int64 + Name string +} + +// EC2StateCodes maps each InstanceState to its EC2 API code and name. +// Note: StateError and StateProvisioning are Hive-specific states with no direct +// AWS EC2 equivalent. Their Code/Name values are best-effort mappings. +var EC2StateCodes = map[InstanceState]EC2StateInfo{ + StateProvisioning: {Code: 0, Name: "pending"}, + StatePending: {Code: 0, Name: "pending"}, + StateRunning: {Code: 16, Name: "running"}, + StateStopping: {Code: 64, Name: "stopping"}, + StateStopped: {Code: 80, Name: "stopped"}, + StateShuttingDown: {Code: 32, Name: "shutting-down"}, + StateTerminated: {Code: 48, Name: "terminated"}, + StateError: {Code: 0, Name: "error"}, +} + +// ValidTransitions defines the allowed state transitions for an instance. +// StateTerminated is intentionally absent — it is a terminal state with no valid transitions. +var ValidTransitions = map[InstanceState][]InstanceState{ + StateProvisioning: {StateRunning, StateError, StateShuttingDown}, + StatePending: {StateRunning, StateError, StateShuttingDown}, + StateRunning: {StateStopping, StateShuttingDown, StateError}, + StateStopping: {StateStopped, StateShuttingDown, StateError}, + StateStopped: {StateRunning, StateShuttingDown, StateError}, + StateShuttingDown: {StateTerminated, StateError}, + StateError: {StateRunning, StateShuttingDown}, +} + +// IsValidTransition checks whether moving from current to target is allowed. +func IsValidTransition(current, target InstanceState) bool { + allowed, ok := ValidTransitions[current] + if !ok { + return false + } + for _, s := range allowed { + if s == target { + return true + } + } + return false +} diff --git a/hive/vm/state_test.go b/hive/vm/state_test.go new file mode 100644 index 000000000..24f3e3bd1 --- /dev/null +++ b/hive/vm/state_test.go @@ -0,0 +1,59 @@ +package vm + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestIsValidTransition(t *testing.T) { + assert.True(t, IsValidTransition(StateRunning, StateStopping)) + assert.True(t, IsValidTransition(StateRunning, StateShuttingDown)) + assert.True(t, IsValidTransition(StateProvisioning, StateRunning)) + assert.True(t, IsValidTransition(StateStopped, StateRunning)) + assert.True(t, IsValidTransition(StateShuttingDown, StateTerminated)) + assert.True(t, IsValidTransition(StateError, StateRunning)) + + assert.False(t, IsValidTransition(StateRunning, StateTerminated)) + assert.False(t, IsValidTransition(StateTerminated, StateRunning)) + assert.False(t, IsValidTransition(StateRunning, StateRunning)) + assert.False(t, IsValidTransition(StateStopped, StateStopping)) +} + +func TestEC2StateCodes_AllStatesHaveMapping(t *testing.T) { + allStates := []InstanceState{ + StateProvisioning, + StatePending, + StateRunning, + StateStopping, + StateStopped, + StateShuttingDown, + StateTerminated, + StateError, + } + + for _, s := range allStates { + info, ok := EC2StateCodes[s] + assert.True(t, ok, "State %s should have an EC2 mapping", s) + assert.NotEmpty(t, info.Name, "State %s EC2 name should not be empty", s) + } +} + +func TestEC2StateCodes_CorrectValues(t *testing.T) { + expected := map[InstanceState]EC2StateInfo{ + StateProvisioning: {Code: 0, Name: "pending"}, + StatePending: {Code: 0, Name: "pending"}, + StateRunning: {Code: 16, Name: "running"}, + StateStopping: {Code: 64, Name: "stopping"}, + StateStopped: {Code: 80, Name: "stopped"}, + StateShuttingDown: {Code: 32, Name: "shutting-down"}, + StateTerminated: {Code: 48, Name: "terminated"}, + StateError: {Code: 0, Name: "error"}, + } + + for state, exp := range expected { + actual := EC2StateCodes[state] + assert.Equal(t, exp.Code, actual.Code, "State %s code mismatch", state) + assert.Equal(t, exp.Name, actual.Name, "State %s name mismatch", state) + } +} diff --git a/hive/vm/vm.go b/hive/vm/vm.go index 4d4b982a2..ff7ef2988 100644 --- a/hive/vm/vm.go +++ b/hive/vm/vm.go @@ -36,7 +36,7 @@ type VM struct { PID int `json:"pid"` PTS int `json:"pts"` Running bool `json:"running"` - Status string `json:"status"` + Status InstanceState `json:"status"` InstanceType string `json:"instance_type"` Config Config `json:"config"`