Perform GO_ERROR after failed STOP_ACTIVITY and RESET during env destruction

Even if we destroy an environment, it is good to recognize that it was not a healthy shutdown, e.g. for bookkeeping reasons.
Also, we might miss out on some necessary calls to integrated services by not going to ERROR.

Fixes OCTRL-1063.

Author: knopers8

core/environment/environment.go

@@ -1233,7 +1233,7 @@ func (env *Environment) subscribeToWfState(taskman *task.Manager) {
 								)
 								err := env.TryTransition(NewGoErrorTransition(taskman))
 								if err != nil {
-									handleFailedGoError(err, env)
+									HandleFailedGoError(err, env)
 								}
 							})
 							break WORKFLOW_STATE_LOOP
@@ -1461,7 +1461,7 @@ func (env *Environment) scheduleAutoStopTransition() (scheduled bool, expected t
 
 						err = env.TryTransition(NewGoErrorTransition(ManagerInstance().taskman))
 						if err != nil {
-							handleFailedGoError(err, env)
+							HandleFailedGoError(err, env)
 						}
 						return
 					}

core/environment/manager.go

@@ -496,7 +496,7 @@ func (envs *Manager) CreateEnvironment(workflowPath string, userVars map[string]
 						envs.taskman),
 					)
 					if err != nil {
-						handleFailedGoError(err, env)
+						HandleFailedGoError(err, env)
 					}
 
 					envTasks := env.Workflow().GetTasks()
@@ -600,7 +600,7 @@ func (envs *Manager) CreateEnvironment(workflowPath string, userVars map[string]
 		envs.taskman),
 	)
 	if errTxErr != nil {
-		handleFailedGoError(errTxErr, env)
+		HandleFailedGoError(errTxErr, env)
 	}
 	envTasks := env.Workflow().GetTasks()
 	// TeardownEnvironment manages the envs.mu internally
@@ -1058,7 +1058,7 @@ func (envs *Manager) handleIntegratedServiceEvent(evt event.IntegratedServiceEve
 							)
 							err = env.TryTransition(NewGoErrorTransition(envs.taskman))
 							if err != nil {
-								handleFailedGoError(err, env)
+								HandleFailedGoError(err, env)
 							}
 						}
 					}()
@@ -1114,7 +1114,7 @@ func (envs *Manager) handleLhcEvents(evt event.IntegratedServiceEvent) {
 					if env.CurrentState() != "ERROR" {
 						err = env.TryTransition(NewGoErrorTransition(envs.taskman))
 						if err != nil {
-							handleFailedGoError(err, env)
+							HandleFailedGoError(err, env)
 						}
 					}
 				}
@@ -1468,7 +1468,7 @@ func (envs *Manager) CreateAutoEnvironment(workflowPath string, userVars map[str
 			envs.taskman),
 		)
 		if err != nil {
-			handleFailedGoError(err, env)
+			HandleFailedGoError(err, env)
 			env.sendEnvironmentEvent(&event.EnvironmentEvent{Message: "transition ERROR failed, forcing", EnvironmentID: env.Id().String(), Error: err})
 		}
 

core/environment/utils.go

@@ -143,7 +143,7 @@ func newCriticalTasksErrorMessage(env *Environment) string {
 	}
 }
 
-func handleFailedGoError(err error, env *Environment) {
+func HandleFailedGoError(err error, env *Environment) {
 	var invalidEventErr fsm.InvalidEventError
 	if errors.As(err, &invalidEventErr) {
 		// this case can occur if the environment is in either:

core/environment/utils_test.go

@@ -30,21 +30,21 @@ import (
 	. "github.com/onsi/gomega"
 )
 
-var _ = Describe("handleFailedGoError", func() {
+var _ = Describe("HandleFailedGoError", func() {
 	It("does not overwrite state for InvalidEventError", func() {
 		env := &Environment{}
 		env.Sm = fsm.NewFSM("DONE", fsm.Events{}, fsm.Callbacks{})
 		Expect(env.Sm.Current()).To(Equal("DONE"))
 
-		handleFailedGoError(fsm.InvalidEventError{Event: "GO_ERROR", State: "DONE"}, env)
+		HandleFailedGoError(fsm.InvalidEventError{Event: "GO_ERROR", State: "DONE"}, env)
 		Expect(env.Sm.Current()).To(Equal("DONE"))
 	})
 
 	It("overwrites state to ERROR for other errors", func() {
 		env := &Environment{}
 		env.Sm = fsm.NewFSM("CONFIGURED", fsm.Events{}, fsm.Callbacks{})
 
-		handleFailedGoError(fsm.UnknownEventError{Event: "BOOM"}, env)
+		HandleFailedGoError(fsm.UnknownEventError{Event: "BOOM"}, env)
 		Expect(env.Sm.Current()).To(Equal("ERROR"))
 	})
 })

core/server.go

@@ -718,10 +718,18 @@ func (m *RpcServer) DestroyEnvironment(cxt context.Context, req *pb.DestroyEnvir
 	}
 
 	if req.AllowInRunningState && env.CurrentState() == "RUNNING" {
-		err = env.TryTransition(environment.MakeTransition(m.state.taskman, pb.ControlEnvironmentRequest_STOP_ACTIVITY))
+		err = env.TryTransition(environment.NewStopActivityTransition(m.state.taskman))
 		if err != nil {
-			log.WithField("partition", env.Id().String()).
-				Warn("could not perform STOP transition for environment teardown, forcing")
+			log.WithError(err).
+				WithField("partition", env.Id().String()).
+				Warn("could not perform STOP transition for environment teardown, going to ERROR, then forcing")
+			the.EventWriterWithTopic(topic.Environment).WriteEvent(
+				environment.NewEnvGoErrorEvent(env, "STOP_ACTIVITY during environment destruction failed"),
+			)
+			err = env.TryTransition(environment.NewGoErrorTransition(m.state.taskman))
+			if err != nil {
+				environment.HandleFailedGoError(err, env)
+			}
 			reply, err = m.doTeardownAndCleanup(env, true /*force*/, false /*keepTasks*/)
 			return
 		}
@@ -746,10 +754,18 @@ func (m *RpcServer) DestroyEnvironment(cxt context.Context, req *pb.DestroyEnvir
 
 	// This might transition to STANDBY if needed, or do nothing if we're already there
 	if env.CurrentState() == "CONFIGURED" {
-		err = env.TryTransition(environment.MakeTransition(m.state.taskman, pb.ControlEnvironmentRequest_RESET))
+		err = env.TryTransition(environment.NewResetTransition(m.state.taskman))
 		if err != nil {
-			log.WithField("partition", env.Id().String()).
-				Warnf("cannot teardown environment in state %s, forcing", env.CurrentState())
+			log.WithError(err).
+				WithField("partition", env.Id().String()).
+				Warnf("cannot teardown environment in state %s, going to ERROR, then forcing", env.CurrentState())
+			the.EventWriterWithTopic(topic.Environment).WriteEvent(
+				environment.NewEnvGoErrorEvent(env, "RESET during environment destruction failed"),
+			)
+			err = env.TryTransition(environment.NewGoErrorTransition(m.state.taskman))
+			if err != nil {
+				environment.HandleFailedGoError(err, env)
+			}
 			reply, err = m.doTeardownAndCleanup(env, true /*force*/, false /*keepTasks*/)
 			return
 		}