marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1920 from yashsinghcodes/nightly

Browse Source 
fix: worker execution takeover for cloud actions
This commit is contained in:
Yash Singh
2026-01-13 22:30:19 +05:30
committed by GitHub
parent cabe871807 6921a63cc5
commit 9286bbcf2b
1 changed files with 35 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
functions/onprem/worker/worker.go
+35
View File
@@ -2652,6 +2652,41 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
}
}
if strings.ToLower(actionResult.Action.Environment) != environment && len(environment) > 0 {
log.Printf("[WARNING] Got an action for %s environment forwarding it to the backend", actionResult.Action.Environment)
streamUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl)
req, err := http.NewRequest(
"POST",
streamUrl,
bytes.NewBuffer([]byte(body)),
)
if err != nil {
log.Printf("[ERROR] Error building subflow (%s) request: %s", workflowExecution.ExecutionId, err)
return
}
newresp, err := topClient.Do(req)
if err != nil {
log.Printf("[ERROR] Error running subflow (%s) request: %s", workflowExecution.ExecutionId, err)
return
}
defer newresp.Body.Close()
if newresp.StatusCode != 200 {
body, err := ioutil.ReadAll(newresp.Body)
if err != nil {
log.Printf("[INFO][%s] Failed reading body after subflow request: %s", workflowExecution.ExecutionId, err)
return
} else {
log.Printf("[ERROR][%s] Failed forwarding subflow request of length %d\n: %s", workflowExecution.ExecutionId, len(actionResult.Result), string(body))
}
}
return
}
log.Printf("[DEBUG][%s] Action: Received, Label: '%s', Action: '%s', Status: %s, Run status: %s, Extra=Retry:%d", workflowExecution.ExecutionId, actionResult.Action.Label, actionResult.Action.AppName, actionResult.Status, workflowExecution.Status, retries)
// results = append(results, actionResult)