Reject empty state decision (#353)

indeedeng · Oct 9, 2023 · dad8c61 · dad8c61
1 parent 5f34973
commit dad8c61
Show file tree

Hide file tree

Showing 7 changed files with 58 additions and 5 deletions.
diff --git a/integ/interstate_test.go b/integ/interstate_test.go
@@ -87,7 +87,7 @@ func doTestInterStateWorkflow(t *testing.T, backendType service.BackendType, con
 	}, history, "interstate test fail, %v", history)
 
 	assertions.Equal(iwfidl.COMPLETED, resp2.GetWorkflowStatus())
-	assertions.Equal(0, len(resp2.GetResults()))
+	assertions.Equal(1, len(resp2.GetResults()))
 	assertions.Equal(map[string]interface{}{
 		interstate.State21 + "received": interstate.TestVal1,
 		interstate.State31 + "received": interstate.TestVal2,

diff --git a/integ/wf_state_execute_api_fail_and_proceed_test.go b/integ/wf_state_execute_api_fail_and_proceed_test.go
@@ -103,5 +103,11 @@ func doTestStateExecuteApiFailAndProceed(t *testing.T, backendType service.Backe
 	assertions.Equalf(&iwfidl.WorkflowGetResponse{
 		WorkflowRunId:  startResp.GetWorkflowRunId(),
 		WorkflowStatus: iwfidl.COMPLETED,
+		Results: []iwfidl.StateCompletionOutput{
+			{
+				CompletedStateId:          wf_execute_api_fail_and_proceed.StateRecover,
+				CompletedStateExecutionId: wf_execute_api_fail_and_proceed.StateRecover + "-1",
+			},
+		},
 	}, resp, "response not expected")
 }
diff --git a/integ/wf_state_wait_until_api_fail_and_proceed_test.go b/integ/wf_state_wait_until_api_fail_and_proceed_test.go
@@ -102,5 +102,11 @@ func doTestStateApiFailAndProceed(t *testing.T, backendType service.BackendType,
 	assertions.Equalf(&iwfidl.WorkflowGetResponse{
 		WorkflowRunId:  startResp.GetWorkflowRunId(),
 		WorkflowStatus: iwfidl.COMPLETED,
+		Results: []iwfidl.StateCompletionOutput{
+			{
+				CompletedStateId:          wf_state_api_fail_and_proceed.State1,
+				CompletedStateExecutionId: wf_state_api_fail_and_proceed.State1 + "-1",
+			},
+		},
 	}, resp, "response not expected")
 }
diff --git a/integ/workflow/interstate/routers.go b/integ/workflow/interstate/routers.go
@@ -159,8 +159,13 @@ func (h *handler) ApiV1WorkflowStateDecide(c *gin.Context) {
 			h.invokeData[State31+"received"] = results.GetInterStateChannelResults()[0].GetValue()
 
 			c.JSON(http.StatusOK, iwfidl.WorkflowStateDecideResponse{
-				// old legacy dead end
-				StateDecision: &iwfidl.StateDecision{},
+				StateDecision: &iwfidl.StateDecision{
+					NextStates: []iwfidl.StateMovement{
+						{
+							StateId: service.GracefulCompletingWorkflowStateId,
+						},
+					},
+				},
 			})
 			return
 		}

diff --git a/integ/workflow/wf_execute_api_fail_and_proceed/routers.go b/integ/workflow/wf_execute_api_fail_and_proceed/routers.go
@@ -1,6 +1,7 @@
 package wf_execute_api_fail_and_proceed
 
 import (
+	"github.com/indeedeng/iwf/service"
 	"log"
 	"net/http"
 
@@ -56,7 +57,15 @@ func (h *handler) ApiV1WorkflowStateDecide(c *gin.Context) {
 	}
 	if req.WorkflowStateId == StateRecover {
 		if input.GetData() == InputData && input.GetEncoding() == InputDataEncoding {
-			c.JSON(http.StatusOK, iwfidl.WorkflowStateDecideResponse{})
+			c.JSON(http.StatusOK, iwfidl.WorkflowStateDecideResponse{
+				StateDecision: &iwfidl.StateDecision{
+					NextStates: []iwfidl.StateMovement{
+						{
+							StateId: service.GracefulCompletingWorkflowStateId,
+						},
+					},
+				},
+			})
 		} else {
 			panic("input is not correct: " + input.GetData() + ", " + input.GetEncoding())
 		}

diff --git a/integ/workflow/wf_state_api_fail_and_proceed/routers.go b/integ/workflow/wf_state_api_fail_and_proceed/routers.go
@@ -1,6 +1,7 @@
 package wf_state_api_fail_and_proceed
 
 import (
+	"github.com/indeedeng/iwf/service"
 	"log"
 	"net/http"
 
@@ -60,7 +61,15 @@ func (h *handler) ApiV1WorkflowStateDecide(c *gin.Context) {
 	if req.GetWorkflowType() == WorkflowType {
 		h.invokeHistory[req.GetWorkflowStateId()+"_decide"]++
 	}
-	c.JSON(http.StatusOK, iwfidl.WorkflowStateDecideResponse{})
+	c.JSON(http.StatusOK, iwfidl.WorkflowStateDecideResponse{
+		StateDecision: &iwfidl.StateDecision{
+			NextStates: []iwfidl.StateMovement{
+				{
+					StateId: service.GracefulCompletingWorkflowStateId,
+				},
+			},
+		},
+	})
 }
 
 func (h *handler) GetTestResult() (map[string]int64, map[string]interface{}) {

diff --git a/service/interpreter/activityImpl.go b/service/interpreter/activityImpl.go
@@ -113,9 +113,21 @@ func StateApiExecute(
 	if checkHttpError(err, httpResp) {
 		return nil, composeHttpError(provider, err, httpResp, string(iwfidl.STATE_API_FAIL_MAX_OUT_RETRY_ERROR_TYPE))
 	}
+
+	if err = checkStateDecisionFromResponse(resp); err != nil {
+		return nil, composeExecuteApiRespError(provider, err, resp)
+	}
+
 	return resp, nil
 }
 
+func checkStateDecisionFromResponse(resp *iwfidl.WorkflowStateDecideResponse) error {
+	if resp == nil || resp.StateDecision == nil || len(resp.StateDecision.NextStates) == 0 {
+		return fmt.Errorf("empty state decision is no longer supported. If it's from old SDKs then upgrade the SDK to newer versions")
+	}
+	return nil
+}
+
 func printDebugMsg(logger UnifiedLogger, err error, url string) {
 	debugMode := os.Getenv(service.EnvNameDebugMode)
 	if debugMode != "" {
@@ -129,6 +141,12 @@ func composeStartApiRespError(provider ActivityProvider, err error, resp *iwfidl
 		fmt.Sprintf("err msg: %v, response: %v", err, string(respStr)))
 }
 
+func composeExecuteApiRespError(provider ActivityProvider, err error, resp *iwfidl.WorkflowStateDecideResponse) error {
+	respStr, _ := resp.MarshalJSON()
+	return provider.NewApplicationError(string(iwfidl.STATE_API_FAIL_MAX_OUT_RETRY_ERROR_TYPE),
+		fmt.Sprintf("err msg: %v, response: %v", err, string(respStr)))
+}
+
 func checkHttpError(err error, httpResp *http.Response) bool {
 	if err != nil || (httpResp != nil && httpResp.StatusCode != http.StatusOK) {
 		return true