feat: add RETRY_NOW state to enable looping over a step (#190)

RETRY_NOW state can be used inside a step condition, to loop over the same step, and re-execute it multiple time.
To fully control the looping, it's possible to configure the max_retry value to -1, in order to generate controlled infinite loop.

Signed-off-by: Alexandre Szymocha <[email protected]>
Signed-off-by: Romain Beuque <[email protected]>
Co-authored-by: Romain Beuque <[email protected]>

Author: qexk

engine/engine.go

@@ -396,9 +396,9 @@ func resolve(dbp zesty.DBProvider, res *resolution.Resolution, t *task.Task, sm
 				executedSteps[s.Name] = true
 			}
 
-			debugLogger.Debugf("Engine: resolve() %s loop, step %s result: %s", res.PublicID, s.Name, s.State)
+			debugLogger.Debugf("Engine: resolve() %s loop, step %s (#%d) result: %s", res.PublicID, s.Name, s.TryCount, s.State)
 
-			// uptate done step count
+			// update done step count
 			// ignore foreach iterations for global done count
 			if s.IsFinal() && !s.IsChild() {
 				t.StepsDone++
@@ -821,6 +821,16 @@ func availableSteps(modifiedSteps map[string]bool, res *resolution.Resolution, e
 			candidateSteps[s] = struct{}{}
 		}
 	}
+
+	// force RETRY_NOW steps to be evaluated again
+	for name, ok := range executedSteps {
+		// second check is necessary when dealing with Foreach children
+		if ok && res.Steps[name] != nil && res.Steps[name].State == step.StateRetryNow {
+			candidateSteps[name] = struct{}{}
+			delete(executedSteps, name)
+		}
+	}
+
 	// looping on just created steps from an EXPANDED step, to verify if they are eligible
 	// (in case we had modifiedSteps at the same time)
 	for _, s := range expandedSteps {

engine/engine_test.go

@@ -578,6 +578,44 @@ func TestMetadata(t *testing.T) {
 	assert.Equal(t, "NOTFOUND", notfoundState)
 }
 
+func TestRetryNow(t *testing.T) {
+	expectedResult := "0 sep 1 sep 1 sep 2 sep 3 sep 5 sep 8 sep 13 sep 21 sep 34 sep 55 sep 89 sep 144"
+	res, err := createResolution("retryNowState.yaml", map[string]interface{}{
+		"N":         12.0,
+		"separator": " sep ",
+	}, nil)
+	assert.Nil(t, err)
+	assert.NotNil(t, res)
+
+	res, err = runResolution(res)
+	assert.NotNil(t, res)
+	assert.Nil(t, err)
+	assert.Equal(t, resolution.StateDone, res.State)
+
+	assert.Equal(t, step.StateDone, res.Steps["fibonacci"].State)
+	assert.Equal(t, step.StateDone, res.Steps["join"].State)
+
+	output := res.Steps["join"].Output.(map[string]interface{})
+	assert.Equal(t, expectedResult, output["str"])
+}
+
+func TestRetryNowMaxRetry(t *testing.T) {
+	expected := "42"
+	res, err := createResolution("retryNowMaxRetry.yaml", map[string]interface{}{}, nil)
+	assert.Nil(t, err)
+	assert.NotNil(t, res)
+
+	res, err = runResolution(res)
+	assert.NotNil(t, res)
+	assert.Nil(t, err)
+	assert.Equal(t, resolution.StateBlockedFatal, res.State)
+
+	assert.Equal(t, step.StateFatalError, res.Steps["infinite"].State)
+	assert.Equal(t, res.Steps["infinite"].TryCount, res.Steps["infinite"].MaxRetries+1)
+
+	assert.Equal(t, expected, res.Steps["infinite"].Output)
+}
+
 func TestForeach(t *testing.T) {
 	res, err := createResolution("foreach.yaml", map[string]interface{}{
 		"list": []interface{}{"a", "b", "c"},

engine/step/step.go

@@ -40,6 +40,7 @@ const (
 	StateCrashed       = "CRASHED"
 	StatePrune         = "PRUNE"
 	StateToRetry       = "TO_RETRY"
+	StateRetryNow      = "RETRY_NOW"
 	StateAfterrunError = "AFTERRUN_ERROR"
 
 	// steps that carry a foreach list of arguments
@@ -50,12 +51,14 @@ const (
 	stepRefThis = utask.This
 
 	defaultMaxRetries = 10000
+
+	maxExecutionDelay = time.Duration(20) * time.Second
 )
 
 var (
-	builtinStates            = []string{StateTODO, StateRunning, StateDone, StateClientError, StateServerError, StateFatalError, StateCrashed, StatePrune, StateToRetry, StateAfterrunError, StateAny, StateExpanded}
-	stepConditionValidStates = []string{StateDone, StatePrune, StateToRetry, StateFatalError, StateClientError}
-	runnableStates           = []string{StateTODO, StateServerError, StateClientError, StateFatalError, StateCrashed, StateToRetry, StateAfterrunError, StateExpanded} // everything but RUNNING, DONE, PRUNE
+	builtinStates            = []string{StateTODO, StateRunning, StateDone, StateClientError, StateServerError, StateFatalError, StateCrashed, StatePrune, StateToRetry, StateRetryNow, StateAfterrunError, StateAny, StateExpanded}
+	stepConditionValidStates = []string{StateDone, StatePrune, StateToRetry, StateRetryNow, StateFatalError, StateClientError}
+	runnableStates           = []string{StateTODO, StateServerError, StateClientError, StateFatalError, StateCrashed, StateToRetry, StateRetryNow, StateAfterrunError, StateExpanded} // everything but RUNNING, DONE, PRUNE
 	retriableStates          = []string{StateServerError, StateToRetry, StateAfterrunError}
 )
 
@@ -90,10 +93,11 @@ type Step struct {
 	State          string                  `json:"state,omitempty"`
 	// hints about ETA latency, async, for retrier to define strategy
 	// how often VS how many times
-	RetryPattern string    `json:"retry_pattern,omitempty"` // seconds, minutes, hours
-	TryCount     int       `json:"try_count,omitempty"`
-	MaxRetries   int       `json:"max_retries,omitempty"`
-	LastRun      time.Time `json:"last_run,omitempty"`
+	RetryPattern   string        `json:"retry_pattern,omitempty"` // seconds, minutes, hours
+	TryCount       int           `json:"try_count,omitempty"`
+	MaxRetries     int           `json:"max_retries,omitempty"`
+	LastRun        time.Time     `json:"last_run,omitempty"`
+	ExecutionDelay time.Duration `json:"execution_delay,omitempty"`
 
 	// flow control
 	Dependencies []string               `json:"dependencies,omitempty"`
@@ -321,7 +325,10 @@ func Run(st *Step, baseConfig map[string]json.RawMessage, stepValues *values.Val
 	if st.MaxRetries == 0 {
 		st.MaxRetries = defaultMaxRetries
 	}
-	if st.TryCount > st.MaxRetries {
+
+	// we can set "max_retries" to a negative number to have full control
+	// over the repetition of a step with a check condition using RETRY_NOW
+	if st.MaxRetries > 0 && st.TryCount > st.MaxRetries {
 		st.State = StateFatalError
 		st.Error = fmt.Sprintf("Step reached max retries %d: %s", st.MaxRetries, st.Error)
 		go noopStep(st, stepChan)
@@ -372,6 +379,8 @@ func Run(st *Step, baseConfig map[string]json.RawMessage, stepValues *values.Val
 	go func() {
 		defer wg.Done()
 
+		time.Sleep(st.ExecutionDelay)
+
 		// Wait the prehook execution is done
 		preHookWg.Wait()
 
@@ -533,6 +542,13 @@ func (st *Step) ValidAndNormalize(name string, baseConfigs map[string]json.RawMe
 		}
 	}
 
+	// valid execution delay
+	if st.ExecutionDelay < 0 || st.ExecutionDelay > maxExecutionDelay {
+		return errors.NewNotValid(nil,
+			fmt.Sprintf("execution_delay: expected %s to be a duration between 0s and %s",
+				st.ExecutionDelay, maxExecutionDelay))
+	}
+
 	// valid retry pattern, accept empty
 	switch st.RetryPattern {
 	case "", RetrySeconds, RetryMinutes, RetryHours:

engine/templates_tests/retryNowMaxRetry.yaml

@@ -0,0 +1,22 @@
+name: loopConditionMaxRetry
+description: stop an infinite loop using max_retry
+title_format: "[test] infinite loop condition with max_retry"
+
+auto_runnable: true
+
+steps:
+    infinite:
+        description: counts to infinity
+        max_retries: 41
+        conditions:
+        - type: check
+          if:
+          - value: 0
+            operator: EQ
+            expected: 0
+          then:
+            this: RETRY_NOW
+        action:
+            type: echo
+            configuration:
+                output: '{{ add (default 0 .step.this.output) 1 }}'

engine/templates_tests/retryNowState.yaml

@@ -0,0 +1,52 @@
+name: retryNowState
+description: generate and format a list of the first N fibonacci numbers
+title_format: "[test] RETRY_NOW state"
+
+inputs:
+    - name: 'N'
+      description: how many numbers to generate
+      type: number
+      regex: '\d+'
+    - name: separator
+      description: what to put between the numbers
+      type: string
+      default: ', '
+
+steps:
+    fibonacci:
+        description: generate the list
+        max_retries: -1
+        conditions:
+        - type: check
+          if:
+          - value: '{{ len (fromJson .step.this.output.list) }}'
+            operator: LE
+            expected: '{{ .input.N }}'
+          then:
+            this: RETRY_NOW
+        action:
+            type: echo
+            configuration:
+                output:
+                    tail0: '{{ default "0" .step.this.output.tail1 }}'
+                    tail1: '{{ default "1" (add .step.this.output.tail0 .step.this.output.tail1) }}'
+                    list: '{{ append (fromJson (default "[]" .step.this.output.list)) (default "0" .step.this.output.tail1) | toJson }}'
+
+    join:
+        description: join the numbers using the separator
+        dependencies:
+        - fibonacci
+        conditions:
+        - type: check
+          if:
+          - value: '{{ len (fromJson .step.this.output.list) }}'
+            operator: GT
+            expected: 1
+          then:
+            this: RETRY_NOW
+        action:
+            type: echo
+            configuration:
+                output:
+                    list: '{{ default .step.fibonacci.output.list .step.this.output.list | fromJson | rest | toJson }}'
+                    str: '{{ default (.step.fibonacci.output.list | fromJson | first) .step.this.output.str }}{{ .input.separator }}{{ default .step.fibonacci.output.list .step.this.output.list | fromJson | rest | first }}'