Fixed a broken part of the condition system which didn't look for skipped/failed nodes

backend/app_sdk/app_base.py

@@ -2466,6 +2466,23 @@ class AppBase:
                     continue
 
                 matching_branches += 1
+
+                # Find if previous is skipped or failed. Skipped != correct branch
+                try:
+                    should_skip = False
+                    for res in fullexecution["results"]:
+                        if res["action"]["id"] == branch["source_id"]:
+                            if res["status"] == "FAILURE" or res["status"] == "SKIPPED":
+                                should_skip = True 
+
+                            break
+
+                    if should_skip:
+                        continue
+                except Exception as e:
+                    self.logger.info("[WARNING] Failed handling check of if parent is skipped") 
+
+
                 # Remove anything without a condition
                 try:
                     if (branch["conditions"]) == 0 or branch["conditions"] == None:
@@ -2475,6 +2492,8 @@ class AppBase:
                     correct_branches += 1
                     continue
 
+                # FIXME: Check if the previous node has a result or not 
+
                 self.logger.info("[DEBUG] Relevant conditions: %s" % branch["conditions"])
                 successful_conditions = []
                 failed_conditions = []
@@ -2535,6 +2554,8 @@ class AppBase:
             if matching_branches > 0 and correct_branches > 0:
                 return True, ""
 
+            # FIXME: Check if previous branches are at all finished
+
             self.logger.info("[DEBUG] Correct branches vs matching branches: %d vs %d" % (correct_branches, matching_branches))
             return False, {"success": False, "reason": "Minimum of one branch's conditions must be correct to continue. Total: %d of %d" % (correct_branches, matching_branches)}
 

backend/go-app/main.go

@@ -4075,7 +4075,7 @@ func runInitEs(ctx context.Context) {
 
 	// FIXME: Have this for all envs in all orgs (loop and find).
 	if len(parsedApikey) > 0 {
-		cleanupSchedule := 600
+		cleanupSchedule := 300
 		environments := []string{"Shuffle"}
 		log.Printf("[DEBUG] Starting schedule setup for execution cleanup every %d seconds. Running first immediately.", cleanupSchedule)
 		cleanupJob := func() func() {

functions/onprem/orborus/orborus.go

@@ -495,23 +495,24 @@ func deployWorker(image string, identifier string, env []string, executionReques
 		// FIXME: Should we handle replies properly?
 		// In certain cases, a workflow may e.g. be aborted already. If it's aborted, that returns
 		// a 401 from the worker, which returns an error here
-		err := sendWorkerRequest(executionRequest)
-		if err != nil {
-			log.Printf("[ERROR] Failed worker request for %s: %s", executionRequest.ExecutionId, err)
+		go sendWorkerRequest(executionRequest)
+		//err := sendWorkerRequest(executionRequest)
+		//if err != nil {
+		//	log.Printf("[ERROR] Failed worker request for %s: %s", executionRequest.ExecutionId, err)
 
-			if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
-				workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
-				deployServiceWorkers(workerImage)
+		//	if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
+		//		workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
+		//		deployServiceWorkers(workerImage)
 
-				time.Sleep(time.Duration(10) * time.Second)
-				err = sendWorkerRequest(executionRequest)
-			}
-		}
+		//		time.Sleep(time.Duration(10) * time.Second)
+		//		err = sendWorkerRequest(executionRequest)
+		//	}
+		//}
 
-		if err == nil {
-			// FIXME: Readd this? Removed for rerun reasons
-			// executionIds = append(executionIds, executionRequest.ExecutionId)
-		}
+		//if err == nil {
+		//	// FIXME: Readd this? Removed for rerun reasons
+		//	// executionIds = append(executionIds, executionRequest.ExecutionId)
+		//}
 		//}()
 
 		return nil
@@ -1327,21 +1328,36 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error {
 		streamUrl = fmt.Sprintf("%s:33333/api/v1/execute", parsedBaseurl)
 	}
 
+	client := &http.Client{}
 	req, err := http.NewRequest(
 		"POST",
 		streamUrl,
 		bytes.NewBuffer([]byte(data)),
 	)
-
-	client := &http.Client{}
 	if err != nil {
 		log.Printf("[ERROR] Failed creating worker request: %s", err)
+		if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
+			workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
+			deployServiceWorkers(workerImage)
+
+			time.Sleep(time.Duration(10) * time.Second)
+			//err = sendWorkerRequest(executionRequest)
+		}
+
 		return err
 	}
 
 	newresp, err := client.Do(req)
 	if err != nil {
 		log.Printf("[ERROR] Error running worker request: %s", err)
+		if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
+			workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
+			deployServiceWorkers(workerImage)
+
+			time.Sleep(time.Duration(10) * time.Second)
+			//err = sendWorkerRequest(executionRequest)
+		}
+
 		return err
 	}
 
@@ -1353,6 +1369,13 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error {
 
 	if newresp.StatusCode != 200 {
 		log.Printf("[ERROR] Error running worker request - status code is %d, not 200. Body: %s", newresp.StatusCode, string(body))
+
+		workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
+		deployServiceWorkers(workerImage)
+
+		time.Sleep(time.Duration(10) * time.Second)
+		//err = sendWorkerRequest(executionRequest)
+
 		return errors.New(fmt.Sprintf("Bad statuscode from worker: %d - expecting 200", newresp.StatusCode))
 	}