Merge pull request #1720 from yashsinghcodes/customer

fix: requesting backend if we cannot find the executionId on cache

functions/onprem/worker/worker.go

@@ -2336,6 +2336,54 @@ func buildEnvVars(envMap map[string]string) []corev1.EnvVar {
 	}
 	return envVars
 }
+func getWorkerBackendExecution(auth string, executionId string) (*shuffle.WorkflowExecution, error) {
+	backendUrl := os.Getenv("BASE_URL")
+	if len(backendUrl) == 0 {
+		backendUrl = "http://shuffle-backend:5001"
+	}
+
+	var workflowExecution *shuffle.WorkflowExecution
+
+	streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", backendUrl)
+	topClient := shuffle.GetExternalClient(backendUrl)
+	requestData := shuffle.ActionResult {
+		Authorization: auth,
+		ExecutionId: executionId,
+	}
+
+	data, err := json.Marshal(requestData)
+	if err != nil {
+		return workflowExecution, err
+	}
+
+	req, err := http.NewRequest(
+		"POST",
+		streamResultUrl,
+		bytes.NewBuffer([]byte(data)),
+	)
+
+	newresp, err := topClient.Do(req)
+	if err != nil {
+		return workflowExecution, err
+	}
+
+	defer newresp.Body.Close()
+	if newresp.StatusCode != 200 {
+		return workflowExecution, errors.New(fmt.Sprintf("Got bad status code from backend %d", newresp.StatusCode))
+	}
+
+	body, err := ioutil.ReadAll(newresp.Body)
+	if err != nil {
+		return workflowExecution, err
+	}
+
+	err = json.Unmarshal(body, &workflowExecution)
+	if err != nil {
+		return workflowExecution, err
+	}
+
+	return workflowExecution, nil
+}
 
 func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
 	if request.Body == nil {
@@ -2379,12 +2427,16 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
 	}
 
 	workflowExecution, err := shuffle.GetWorkflowExecution(ctx, actionResult.ExecutionId)
+	if err != nil {
+		log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (1): %s", actionResult.ExecutionId, err)
+		workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId)
 		if err != nil {
 			log.Printf("[ERROR][%s] Failed getting execution (workflowqueue) %s: %s", actionResult.ExecutionId, actionResult.ExecutionId, err)
 			resp.WriteHeader(500)
 			resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution ID %s because it doesn't exist locally."}`, actionResult.ExecutionId)))
 			return
 		}
+	}
 
 	if workflowExecution.Authorization != actionResult.Authorization {
 		log.Printf("[ERROR][%s] Bad authorization key when updating node (workflowQueue). Want: %s, Have: %s", actionResult.ExecutionId, workflowExecution.Authorization, actionResult.Authorization)
@@ -2428,12 +2480,16 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
 func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workflowExecutionId string, actionResult shuffle.ActionResult, resp http.ResponseWriter) {
 	//log.Printf("[DEBUG][%s] IN WORKFLOWEXECUTION SUB!", actionResult.ExecutionId)
 	workflowExecution, err := shuffle.GetWorkflowExecution(ctx, workflowExecutionId)
+	if err != nil {
+		log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (2): %s", actionResult.ExecutionId, err)
+		workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId)
 		if err != nil {
 			log.Printf("[ERROR] Failed getting execution cache: %s", err)
 			resp.WriteHeader(400)
 			resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution"}`)))
 			return
 		}
+	}
 
 	resultLength := len(workflowExecution.Results)
 	setExecution := true
@@ -2482,12 +2538,16 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl
 		log.Printf("[DEBUG] Rerunning transaction? %s", err)
 		if strings.Contains(fmt.Sprintf("%s", err), "Rerun this transaction") {
 			workflowExecution, err := shuffle.GetWorkflowExecution(ctx, workflowExecutionId)
+			if err != nil {
+				log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (3): %s", actionResult.ExecutionId, err)
+				workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId)
 				if err != nil {
 					log.Printf("[ERROR][%s] Failed getting execution cache (2): %s", workflowExecution.ExecutionId, err)
 					resp.WriteHeader(400)
 					resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution (2)"}`)))
 					return
 				}
+			}
 
 			resultLength = len(workflowExecution.Results)
 			setExecution = true
@@ -2796,11 +2856,15 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
 	ctx := context.Background()
 	workflowExecution, err := shuffle.GetWorkflowExecution(ctx, actionResult.ExecutionId)
 	if err != nil {
-		log.Printf("[INFO] Failed getting execution (streamresult) %s: %s", actionResult.ExecutionId, err)
+		log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (4): %s", actionResult.ExecutionId, err)
+		workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId)
+		if err != nil {
+			log.Printf("[ERROR] Failed getting execution (streamresult) %s: %s", actionResult.ExecutionId, err)
 			resp.WriteHeader(400)
 			resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key or execution_id might not exist."}`)))
 			return
 		}
+	}
 
 	// Authorization is done here
 	if workflowExecution.Authorization != actionResult.Authorization {