Merge pull request #6 from Azure/haitao/withRetry

Haitao/with retry

Author: haitch

README.md

@@ -10,7 +10,7 @@ AsyncJob aiming to help you organize code in dependencyGraph(DAG), instead of a
 **Step** is a individual code block which can be executed and have inputs, output.
 - a step would be started once all it's dependency is finished.
 - output of a step can be feed into next step as input, type is checked by go generics.
-- step is wrapped in [AsyncTask](github.com/Azure/go-asynctask) with strongType info preserved
+- step is wrapped in [AsyncTask](https://github.com/Azure/go-asynctask) with strongType info preserved
 - you can feed parameters as a step as well.
 
 # Usage
@@ -38,40 +38,57 @@ AsyncJob aiming to help you organize code in dependencyGraph(DAG), instead of a
 	# summarize
 	StepAfterBoth(bCtx, job, "summarize", qery1ResultTask, qery2ResultTask, jobLib.SummarizeQueryResult)
 
-	# visualize the job
-	dotGraph := job.Visualize()
-	fmt.Println(dotGraph)
-
 	# execute job
 	job.Start(context.Background())
 	job.Wait(context.WithTimeout(context.Background(), 10*time.Second))
 ```
 
 ### visualize of a job
-tried https://github.com/hashicorp/terraform/tree/main/internal/dag, which doesn't have own go module, but terraform go module have too much dependencies.
-baking a inhouse one.
+```
+	# visualize the job
+	dotGraph := job.Visualize()
+	fmt.Println(dotGraph)
+```
 
 ```
 digraph {
-	compound = "true"
 	newrank = "true"
-	subgraph "root" {
-		"[root] [Start]" -> "[root] getConnection"
-		"[root] [Start]" -> "[root] param_query1"
-		"[root] [Start]" -> "[root] param_query2"
-		"[root] [Start]" -> "[root] param_table1"
-		"[root] [Start]" -> "[root] param_table2"
-		"[root] getConnection" -> "[root] getTableClient1"
-		"[root] getConnection" -> "[root] getTableClient2"
-		"[root] getTableClient1" -> "[root] queryTable1"
-		"[root] getTableClient2" -> "[root] queryTable2"
-		"[root] param_query1" -> "[root] queryTable1"
-		"[root] param_query2" -> "[root] queryTable2"
-		"[root] param_table1" -> "[root] getTableClient1"
-		"[root] param_table2" -> "[root] getTableClient2"
-		"[root] queryTable1" -> "[root] summarize"
-		"[root] queryTable2" -> "[root] summarize"
-	}
+		param_table1 [label="table1" shape=hexagon style=filled tooltip="Type: param\nName: table1\nState: completed\nStartAt: 2022-11-03T00:56:30.006196-07:00\nDuration: 12.657µs" fillcolor=green] 
+		param_query1 [label="query1" shape=hexagon style=filled tooltip="Type: param\nName: query1\nState: completed\nStartAt: 2022-11-03T00:56:30.0062-07:00\nDuration: 17.013µs" fillcolor=green] 
+		root_job [label="job" shape=triangle style=filled tooltip="Type: root\nName: job\nState: completed\nStartAt: 2022-11-03T00:56:30.006183-07:00\nDuration: 3.695µs" fillcolor=green] 
+		param_query2 [label="query2" shape=hexagon style=filled tooltip="Type: param\nName: query2\nState: completed\nStartAt: 2022-11-03T00:56:30.006197-07:00\nDuration: 13.781µs" fillcolor=green] 
+		task_getTableClient1 [label="getTableClient1" shape=box style=filled tooltip="Type: task\nName: getTableClient1\nState: completed\nStartAt: 2022-11-03T00:56:30.006304-07:00\nDuration: 34.652µs" fillcolor=green] 
+		task_queryTable1 [label="queryTable1" shape=box style=filled tooltip="Type: task\nName: queryTable1\nState: completed\nStartAt: 2022-11-03T00:56:30.006349-07:00\nDuration: 3.217443247s" fillcolor=green] 
+		param_table2 [label="table2" shape=hexagon style=filled tooltip="Type: param\nName: table2\nState: completed\nStartAt: 2022-11-03T00:56:30.006199-07:00\nDuration: 15.632µs" fillcolor=green] 
+		task_getTableClient2 [label="getTableClient2" shape=box style=filled tooltip="Type: task\nName: getTableClient2\nState: completed\nStartAt: 2022-11-03T00:56:30.00631-07:00\nDuration: 51.872µs" fillcolor=green] 
+		task_queryTable2 [label="queryTable2" shape=box style=filled tooltip="Type: task\nName: queryTable2\nState: completed\nStartAt: 2022-11-03T00:56:30.006377-07:00\nDuration: 67.814µs" fillcolor=green] 
+		task_emailNotification [label="emailNotification" shape=box style=filled tooltip="Type: task\nName: emailNotification\nState: completed\nStartAt: 2022-11-03T00:56:33.223952-07:00\nDuration: 3.92µs" fillcolor=green] 
+		param_serverName [label="serverName" shape=hexagon style=filled tooltip="Type: param\nName: serverName\nState: completed\nStartAt: 2022-11-03T00:56:30.006198-07:00\nDuration: 14.638µs" fillcolor=green] 
+		task_getConnection [label="getConnection" shape=box style=filled tooltip="Type: task\nName: getConnection\nState: completed\nStartAt: 2022-11-03T00:56:30.006231-07:00\nDuration: 62.234µs" fillcolor=green] 
+		task_checkAuth [label="checkAuth" shape=box style=filled tooltip="Type: task\nName: checkAuth\nState: completed\nStartAt: 2022-11-03T00:56:30.006212-07:00\nDuration: 650ns" fillcolor=green] 
+		task_summarize [label="summarize" shape=box style=filled tooltip="Type: task\nName: summarize\nState: completed\nStartAt: 2022-11-03T00:56:33.22392-07:00\nDuration: 4.325µs" fillcolor=green] 
+        
+		param_table1 -> task_getTableClient1 [style=bold tooltip="Time: 2022-11-03T00:56:30.006304-07:00" color=green] 
+		param_query1 -> task_queryTable1 [style=bold tooltip="Time: 2022-11-03T00:56:30.006349-07:00" color=green] 
+		param_table2 -> task_getTableClient2 [style=bold tooltip="Time: 2022-11-03T00:56:30.00631-07:00" color=green] 
+		task_getTableClient2 -> task_queryTable2 [style=bold tooltip="Time: 2022-11-03T00:56:30.006377-07:00" color=green] 
+		param_query2 -> task_queryTable2 [style=bold tooltip="Time: 2022-11-03T00:56:30.006377-07:00" color=green] 
+		task_queryTable2 -> task_summarize [style=bold tooltip="Time: 2022-11-03T00:56:33.22392-07:00" color=green] 
+		root_job -> param_serverName [style=bold tooltip="Time: 2022-11-03T00:56:30.006198-07:00" color=green] 
+		root_job -> task_checkAuth [style=bold tooltip="Time: 2022-11-03T00:56:30.006212-07:00" color=green] 
+		root_job -> param_table1 [style=bold tooltip="Time: 2022-11-03T00:56:30.006196-07:00" color=green] 
+		root_job -> param_query1 [style=bold tooltip="Time: 2022-11-03T00:56:30.0062-07:00" color=green] 
+		root_job -> param_table2 [style=bold tooltip="Time: 2022-11-03T00:56:30.006199-07:00" color=green] 
+		root_job -> param_query2 [style=bold tooltip="Time: 2022-11-03T00:56:30.006197-07:00" color=green] 
+		param_serverName -> task_getConnection [style=bold tooltip="Time: 2022-11-03T00:56:30.006231-07:00" color=green] 
+		task_getTableClient1 -> task_queryTable1 [style=bold tooltip="Time: 2022-11-03T00:56:30.006349-07:00" color=green] 
+		task_queryTable1 -> task_summarize [style=bold tooltip="Time: 2022-11-03T00:56:33.22392-07:00" color=green] 
+		task_summarize -> task_emailNotification [style=bold tooltip="Time: 2022-11-03T00:56:33.223952-07:00" color=green] 
+		task_getConnection -> task_getTableClient1 [style=bold tooltip="Time: 2022-11-03T00:56:30.006304-07:00" color=green] 
+		task_getConnection -> task_getTableClient2 [style=bold tooltip="Time: 2022-11-03T00:56:30.00631-07:00" color=green] 
+		task_checkAuth -> task_queryTable1 [style=bold tooltip="Time: 2022-11-03T00:56:30.006349-07:00" color=green] 
+		task_checkAuth -> task_queryTable2 [style=bold tooltip="Time: 2022-11-03T00:56:30.006377-07:00" color=green] 
+
 }
 ```
-![visualize job graph](media/graphviz.svg)
+![visualize job graph](media/asyncjob.svg)

error.go

@@ -0,0 +1,46 @@
+package asyncjob
+
+import (
+	"fmt"
+)
+
+type JobErrorCode string
+
+const (
+	ErrPrecedentStepFailure JobErrorCode = "precedent step failed"
+	ErrStepFailed           JobErrorCode = "current step failed"
+)
+
+func (code JobErrorCode) Error() string {
+	return string(code)
+}
+
+type JobError struct {
+	Code      JobErrorCode
+	StepError error
+	StepName  string
+	Message   string
+}
+
+func newJobError(code JobErrorCode, message string) *JobError {
+	return &JobError{Code: code, Message: message}
+}
+
+func newStepError(stepName string, stepErr error) *JobError {
+	return &JobError{Code: ErrStepFailed, StepName: stepName, StepError: stepErr}
+}
+
+func (je *JobError) Error() string {
+	if je.Code == ErrStepFailed && je.StepError != nil {
+		return fmt.Sprintf("step %q failed: %s", je.StepName, je.StepError.Error())
+	}
+	return je.Code.Error() + ": " + je.Message
+}
+
+func (je *JobError) Unwrap() error {
+	if je.Code == ErrStepFailed {
+		return je.StepError
+	}
+
+	return je.Code
+}

graph/template.go

@@ -10,43 +10,9 @@ import (
 var digraphTemplate = template.Must(template.New("digraph").Parse(digraphTemplateText))
 
 const digraphTemplateText = `digraph {
-	compound = "true"
 	newrank = "true"
-	subgraph "root" {
 {{ range $node := $.Nodes}}		{{$node.ID}} [label="{{$node.Name}}" shape={{$node.Shape}} style={{$node.Style}} tooltip="{{$node.Tooltip}}" fillcolor={{$node.FillColor}}] 
 {{ end }}        
 {{ range $edge := $.Edges}}		{{$edge.FromNodeID}} -> {{$edge.ToNodeID}} [style={{$edge.Style}} tooltip="{{$edge.Tooltip}}" color={{$edge.Color}}] 
-{{ end }}        
-	}
+{{ end }}
 }`
-
-/* ideal output
-digraph G {
-  jobroot [shape=triangle style=filled fillcolor=gold tooltip="State: Failed\nDuration:5s"]
-  param_servername [label="servername" shape=doublecircle style=filled fillcolor=green tooltip="Value: dummy.server.io"]
-  param_table1 [label="table1" shape=doublecircle style=filled fillcolor=green tooltip="Value: table1"]
-  param_query1 [label="query1" shape=doublecircle style=filled fillcolor=green tooltip="Value: select * from table1"]
-  param_table2 [label="table2" shape=doublecircle style=filled fillcolor=green tooltip="Value: table2"]
-  param_query2 [label="query2" shape=doublecircle style=filled fillcolor=green tooltip="Value: select * from table2"]
-  jobroot -> param_servername [tooltip="time:2022-10-28T21:16:07Z"]
-  param_servername -> func_getConnection
-  func_getConnection [label="getConnection" shape=ellipse style=filled fillcolor=green  tooltip="State: Finished\nDuration:1s"]
-  func_query1 [label="query1" shape=ellipse style=filled fillcolor=green tooltip="State: Finished\nDuration:2s"]
-  func_query2 [label="query2" shape=ellipse style=filled fillcolor=red tooltip="State: Failed\nDuration:2s"]
-  jobroot -> param_table1 [style=bold color=green tooltip="time:2022-10-28T21:16:07Z"]
-  param_table1 -> func_query1 [tooltip="time:2022-10-28T21:16:07Z"]
-  jobroot -> param_query1 [tooltip="time:2022-10-28T21:16:07Z"]
-  param_query1 -> func_query1
-  jobroot -> param_table2 [tooltip="time:2022-10-28T21:16:07Z"]
-  param_table2 -> func_query2
-  jobroot -> param_query2 [tooltip="time:2022-10-28T21:16:07Z"]
-  param_query2 -> func_query2
-  func_getConnection -> func_query1
-  func_query1 -> func_summarize
-  func_getConnection -> func_query2
-  func_query2 -> func_summarize [color=red]
-  func_summarize [label="summarize" shape=ellipse style=filled fillcolor=red tooltip="State: Blocked"]
-  func_email [label="email" shape=ellipse style=filled tooltip="State: Pending"]
-  func_summarize -> func_email [style=dotted]
-}
-*/

graph_node.go

@@ -33,7 +33,7 @@ func (sn *stepNode) getShape() string {
 	case stepTypeRoot:
 		return "triangle"
 	case stepTypeParam:
-		return "doublecircle"
+		return "hexagon"
 	case stepTypeTask:
 		return "box"
 	default:

job.go

@@ -116,18 +116,28 @@ func AddStep[T any](bCtx context.Context, j *Job, stepName string, stepFunc asyn
 			step.executionData.StartTime = time.Now()
 			step.state = StepStateFailed
 			step.executionData.Duration = 0 */
-			return nil, err
+			return nil, newJobError(ErrPrecedentStepFailure, "")
 		}
 		step.executionData.StartTime = time.Now()
 		step.state = StepStateRunning
-		result, err := stepFunc(j.runtimeCtx)
+
+		var result *T
+		var err error
+		if step.executionOptions.RetryPolicy != nil {
+			step.executionData.Retried = &RetryReport{}
+			result, err = newRetryer(step.executionOptions.RetryPolicy, step.executionData.Retried, func() (*T, error) { return stepFunc(j.runtimeCtx) }).Run()
+		} else {
+			result, err = stepFunc(j.runtimeCtx)
+		}
+
 		if err != nil {
 			step.state = StepStateFailed
 		} else {
 			step.state = StepStateCompleted
 		}
+
 		step.executionData.Duration = time.Since(step.executionData.StartTime)
-		return result, err
+		return result, newStepError(stepName, err)
 	}
 
 	step.task = asynctask.Start(bCtx, instrumentedFunc)
@@ -171,18 +181,27 @@ func StepAfter[T, S any](bCtx context.Context, j *Job, stepName string, parentSt
 			step.executionData.StartTime = time.Now()
 			step.state = StepStateFailed
 			step.executionData.Duration = 0 */
-			return nil, err
+			return nil, newJobError(ErrPrecedentStepFailure, "")
 		}
 		step.executionData.StartTime = time.Now()
 		step.state = StepStateRunning
-		result, err := stepFunc(j.runtimeCtx, t)
+		var result *S
+		var err error
+		if step.executionOptions.RetryPolicy != nil {
+			step.executionData.Retried = &RetryReport{}
+			result, err = newRetryer(step.executionOptions.RetryPolicy, step.executionData.Retried, func() (*S, error) { return stepFunc(j.runtimeCtx, t) }).Run()
+		} else {
+			result, err = stepFunc(j.runtimeCtx, t)
+		}
+
 		if err != nil {
 			step.state = StepStateFailed
 		} else {
 			step.state = StepStateCompleted
 		}
+
 		step.executionData.Duration = time.Since(step.executionData.StartTime)
-		return result, err
+		return result, newStepError(stepName, err)
 	}
 
 	step.task = asynctask.ContinueWith(bCtx, parentStep.task, instrumentedFunc)
@@ -228,19 +247,29 @@ func StepAfterBoth[T, S, R any](bCtx context.Context, j *Job, stepName string, p
 			step.executionData.StartTime = time.Now()
 			step.state = StepStateFailed
 			step.executionData.Duration = 0 */
-			return nil, err
+			return nil, newJobError(ErrPrecedentStepFailure, "")
 		}
 
 		step.executionData.StartTime = time.Now()
 		step.state = StepStateRunning
-		result, err := stepFunc(j.runtimeCtx, t, s)
+
+		var result *R
+		var err error
+		if step.executionOptions.RetryPolicy != nil {
+			step.executionData.Retried = &RetryReport{}
+			result, err = newRetryer(step.executionOptions.RetryPolicy, step.executionData.Retried, func() (*R, error) { return stepFunc(j.runtimeCtx, t, s) }).Run()
+		} else {
+			result, err = stepFunc(j.runtimeCtx, t, s)
+		}
+
 		if err != nil {
 			step.state = StepStateFailed
 		} else {
 			step.state = StepStateCompleted
 		}
+
 		step.executionData.Duration = time.Since(step.executionData.StartTime)
-		return result, err
+		return result, newStepError(stepName, err)
 	}
 
 	step.task = asynctask.AfterBoth(bCtx, parentStepT.task, parentStepS.task, instrumentedFunc)