BUG: Fixed transactional issue with workflowexecutions
This commit is contained in:
@@ -683,7 +683,7 @@ func handleCreateFile(resp http.ResponseWriter, request *http.Request) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if strings.Contains(curfile.Filename, "/") || strings.Contains(curfile.Filename, `"`) || strings.Contains(curfile.Filename, "..") {
|
if strings.Contains(curfile.Filename, "/") || strings.Contains(curfile.Filename, `"`) || strings.Contains(curfile.Filename, "..") || strings.Contains(curfile.Filename, "~") {
|
||||||
resp.WriteHeader(401)
|
resp.WriteHeader(401)
|
||||||
resp.Write([]byte(`{"success": false, "reason": "Invalid characters in filename"}`))
|
resp.Write([]byte(`{"success": false, "reason": "Invalid characters in filename"}`))
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -954,6 +954,31 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
runWorkflowExecutionTransaction(0, workflowExecution.ExecutionId, actionResult, resp)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Will make sure transactions are always ran for an execution. This is recursive if it fails. Allowed to fail up to 5 times
|
||||||
|
func runWorkflowExecutionTransaction(attempts int64, workflowExecutionId string, actionResult ActionResult, resp http.ResponseWriter) {
|
||||||
|
ctx := context.Background()
|
||||||
|
// Should start a tx for the execution here
|
||||||
|
tx, err := dbclient.NewTransaction(ctx)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("client.NewTransaction: %v", err)
|
||||||
|
resp.WriteHeader(401)
|
||||||
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed creating transaction"}`)))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
key := datastore.NameKey("workflowexecution", workflowExecutionId, nil)
|
||||||
|
workflowExecution := &WorkflowExecution{}
|
||||||
|
if err := tx.Get(key, workflowExecution); err != nil {
|
||||||
|
log.Printf("tx.Get bug: %v", err)
|
||||||
|
tx.Rollback()
|
||||||
|
resp.WriteHeader(401)
|
||||||
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting the workflow key"}`)))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
if actionResult.Status == "ABORTED" || actionResult.Status == "FAILURE" {
|
if actionResult.Status == "ABORTED" || actionResult.Status == "FAILURE" {
|
||||||
log.Printf("Actionresult is %s. Should set workflowExecution and exit all running functions", actionResult.Status)
|
log.Printf("Actionresult is %s. Should set workflowExecution and exit all running functions", actionResult.Status)
|
||||||
|
|
||||||
@@ -1228,18 +1253,32 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
err = setWorkflowExecution(ctx, *workflowExecution)
|
// Transactions: https://cloud.google.com/datastore/docs/concepts/transactions#datastore-datastore-transactional-update-go
|
||||||
if err != nil {
|
// Prevents timing issues
|
||||||
//workflowExecution.Result = "Error setting workflow: result too large"
|
//ExecutionId
|
||||||
//workflowExecution.Status = "FINISHED"
|
if _, err := tx.Put(key, workflowExecution); err != nil {
|
||||||
//workflowExecution.CompletedAt = int64(time.Now().Unix())
|
tx.Rollback()
|
||||||
|
log.Printf("[ERROR] tx.Put bug: %v", err)
|
||||||
|
|
||||||
log.Printf("Error saving workflow execution actionresult setting: %s", err)
|
|
||||||
resp.WriteHeader(401)
|
resp.WriteHeader(401)
|
||||||
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed setting workflowexecution actionresult: %s"}`, err)))
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed setting workflowexecution actionresult: %s"}`, err)))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if _, err = tx.Commit(); err != nil {
|
||||||
|
log.Printf("[ERROR] tx.Commit: %v", err)
|
||||||
|
|
||||||
|
if attempts >= 0 {
|
||||||
|
resp.WriteHeader(401)
|
||||||
|
resp.Write([]byte(`{"success": false}`))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
attempts += 1
|
||||||
|
runWorkflowExecutionTransaction(attempts, workflowExecutionId, actionResult, resp)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
resp.WriteHeader(200)
|
resp.WriteHeader(200)
|
||||||
resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
|
resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user