Rework workflow step scheduling to improve stacktraces

The initial implementation caused them to be flaky when working
with loops.

automation/service/session.go

@@ -376,7 +376,7 @@ func (svc *session) stateChangeHandler(ctx context.Context) wfexec.StateChangeHa
 			frame.ElapsedTime = uint(frame.CreatedAt.Sub(ses.RuntimeStacktrace[0].CreatedAt) / time.Millisecond)
 		}
 
-		ses.RuntimeStacktrace = append(ses.RuntimeStacktrace, frame)
+		ses.AppendRuntimeStacktrace(frame)
 
 		switch i {
 		case wfexec.SessionPrompted:

automation/types/session.go

@@ -103,15 +103,15 @@ func NewSession(s *wfexec.Session) *Session {
 	}
 }
 
-func (s Session) Exec(ctx context.Context, step wfexec.Step, input *expr.Vars) error {
+func (s *Session) Exec(ctx context.Context, step wfexec.Step, input *expr.Vars) error {
 	return s.session.Exec(ctx, step, input)
 }
 
-func (s Session) Resume(ctx context.Context, stateID uint64, input *expr.Vars) (*wfexec.ResumedPrompt, error) {
+func (s *Session) Resume(ctx context.Context, stateID uint64, input *expr.Vars) (*wfexec.ResumedPrompt, error) {
 	return s.session.Resume(ctx, stateID, input)
 }
 
-func (s Session) PendingPrompts(ownerId uint64) []*wfexec.PendingPrompt {
+func (s *Session) PendingPrompts(ownerId uint64) []*wfexec.PendingPrompt {
 	return s.session.UserPendingPrompts(ownerId)
 }
 
@@ -155,6 +155,13 @@ func (s *Session) Apply(ssp SessionStartParams) {
 	}
 }
 
+func (s *Session) AppendRuntimeStacktrace(frame *wfexec.Frame) {
+	s.l.RLock()
+	defer s.l.RUnlock()
+
+	s.RuntimeStacktrace = append(s.RuntimeStacktrace, frame)
+}
+
 func (s *Session) CopyRuntimeStacktrace() {
 	s.l.RLock()
 	defer s.l.RUnlock()

pkg/wfexec/session.go

@@ -326,15 +326,24 @@ func (s *Session) Resume(ctx context.Context, stateId uint64, input *expr.Vars)
 	return p.toResumed(), nil
 }
 
-func (s *Session) enqueue(ctx context.Context, st *State) error {
+func (s *Session) canEnqueue(st *State) error {
 	if st == nil {
 		return fmt.Errorf("state is nil")
 	}
 
-	if st.step == nil {
+	// when the step is completed right away, it is considered as special
+	if st.step == nil && st.completed == nil {
 		return fmt.Errorf("state step is nil")
 	}
 
+	return nil
+}
+
+func (s *Session) enqueue(ctx context.Context, st *State) error {
+	if err := s.canEnqueue(st); err != nil {
+		return err
+	}
+
 	if st.stateId == 0 {
 		st.stateId = nextID()
 	}
@@ -448,7 +457,18 @@ func (s *Session) worker(ctx context.Context) {
 			s.execLock <- struct{}{}
 
 			go func() {
-				err := s.exec(ctx, st)
+				defer func() {
+					// remove protection that prevents multiple
+					// steps executing at the same time
+					<-s.execLock
+				}()
+
+				var (
+					err error
+					log = s.log.With(zap.Uint64("stateID", st.stateId))
+				)
+
+				nxt, err := s.exec(ctx, log, st)
 				if err != nil && st.err == nil {
 					// override the error from the execution
 					st.err = err
@@ -456,10 +476,6 @@ func (s *Session) worker(ctx context.Context) {
 
 				st.completed = now()
 
-				// remove protection that prevents multiple
-				// steps executing at the same time
-				<-s.execLock
-
 				status := s.Status()
 				if st.err != nil {
 					st.err = fmt.Errorf(
@@ -493,6 +509,18 @@ func (s *Session) worker(ctx context.Context) {
 				)
 
 				s.eventHandler(status, st, s)
+
+				for _, n := range nxt {
+					if n.step != nil {
+						log.Debug("next step queued", zap.Uint64("nextStepId", n.step.ID()))
+					} else {
+						log.Debug("next step queued", zap.Uint64("nextStepId", 0))
+					}
+					if err = s.enqueue(ctx, n); err != nil {
+						log.Error("unable to enqueue", zap.Error(err))
+						return
+					}
+				}
 			}()
 
 		case err := <-s.qErr:
@@ -547,7 +575,7 @@ func (s *Session) queueScheduledSuspended() {
 }
 
 // executes single step, resolves response and schedule following steps for execution
-func (s *Session) exec(ctx context.Context, st *State) (err error) {
+func (s *Session) exec(ctx context.Context, log *zap.Logger, st *State) (nxt []*State, err error) {
 	st.created = *now()
 
 	defer func() {
@@ -579,8 +607,6 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 		scope  = (&expr.Vars{}).MustMerge(st.scope)
 
 		currLoop = st.loopCurr()
-
-		log = s.log.With(zap.Uint64("stateID", st.stateId))
 	)
 
 	if st.step != nil {
@@ -614,7 +640,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 		if st.err != nil {
 			if st.errHandler == nil {
 				// no error handler set
-				return st.err
+				return nil, st.err
 			}
 
 			// handling error with error handling
@@ -631,11 +657,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 			eh := st.errHandler
 			st.errHandler = nil
 			st.errHandled = true
-			if err = s.enqueue(ctx, st.Next(eh, scope)); err != nil {
-				log.Warn("unable to queue", zap.Error(err))
-			}
-
-			return nil
+			return []*State{st.Next(eh, scope)}, nil
 		}
 
 		switch l := result.(type) {
@@ -647,7 +669,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 			)
 			n, result, st.err = l.Next(ctx, scope)
 			if st.err != nil {
-				return st.err
+				return nil, st.err
 			}
 
 			if n == nil {
@@ -683,7 +705,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 		case *loopBreak:
 			st.action = "loop break"
 			if currLoop == nil {
-				return fmt.Errorf("break step not inside a loop")
+				return nil, fmt.Errorf("break step not inside a loop")
 			}
 
 			// jump out of the loop
@@ -693,7 +715,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 		case *loopContinue:
 			st.action = "loop continue"
 			if currLoop == nil {
-				return fmt.Errorf("continue step not inside a loop")
+				return nil, fmt.Errorf("continue step not inside a loop")
 			}
 
 			// jump back to iterator
@@ -710,9 +732,10 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 			st.action = "termination"
 			// terminate all activities, all delayed tasks and exit right away
 			log.Debug("termination", zap.Int("delayed", len(s.delayed)))
+			s.mux.Lock()
 			s.delayed = nil
-			s.qState <- FinalState(s, scope)
-			return
+			s.mux.Unlock()
+			return []*State{FinalState(s, scope)}, nil
 
 		case *delayed:
 			st.action = "delayed"
@@ -731,7 +754,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 		case *prompted:
 			st.action = "prompted"
 			if result.ownerId == 0 {
-				return fmt.Errorf("without an owner")
+				return nil, fmt.Errorf("without an owner")
 			}
 
 			result.state = st
@@ -753,7 +776,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 			st.next = Steps{result}
 
 		default:
-			return fmt.Errorf("unknown exec response type %T", result)
+			return nil, fmt.Errorf("unknown exec response type %T", result)
 		}
 	}
 
@@ -767,7 +790,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 		// do a quick sanity check
 		cc := s.g.Children(st.step)
 		if len(cc) > 0 && !cc.Contains(st.next...) {
-			return fmt.Errorf("inconsistent relationship")
+			return nil, fmt.Errorf("inconsistent relationship")
 		}
 	}
 
@@ -781,19 +804,21 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
 	if len(st.next) == 0 {
 		log.Debug("zero paths, finalizing")
 		// using state to transport results and complete the worker loop
-		s.qState <- FinalState(s, scope)
-		return
+		return []*State{FinalState(s, scope)}, nil
 	}
 
-	for _, step := range st.next {
-		log.Debug("next step queued", zap.Uint64("nextStepId", step.ID()))
-		if err = s.enqueue(ctx, st.Next(step, scope)); err != nil {
+	nxt = make([]*State, len(st.next))
+	for i, step := range st.next {
+		nn := st.Next(step, scope)
+		if err = s.canEnqueue(nn); err != nil {
 			log.Error("unable to queue", zap.Error(err))
 			return
 		}
+
+		nxt[i] = nn
 	}
 
-	return
+	return nxt, nil
 }
 
 func SetWorkerInterval(i time.Duration) SessionOpt {

tests/workflows/0010_stacktrace_test.go

@@ -0,0 +1,55 @@
+package workflows
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	autTypes "github.com/cortezaproject/corteza-server/automation/types"
+	"github.com/stretchr/testify/require"
+)
+
+func Test0010_stacktrace(t *testing.T) {
+	var (
+		ctx = bypassRBAC(context.Background())
+		req = require.New(t)
+	)
+
+	req.NoError(defStore.TruncateComposeRecords(ctx, nil))
+	req.NoError(defStore.TruncateComposeModules(ctx))
+	req.NoError(defStore.TruncateComposeNamespaces(ctx))
+
+	loadScenario(ctx, t)
+
+	for rep := 0; rep < 11; rep++ {
+		t.Run(fmt.Sprintf("iteration %d", rep), func(t *testing.T) {
+			var (
+				_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+			)
+
+			// 6x iterator, 5x continue, 1x terminator, 1x completed
+			req.Len(trace, 13)
+
+			steps := []uint64{
+				10,
+				11,
+				10,
+				11,
+				10,
+				11,
+				10,
+				11,
+				10,
+				11,
+
+				10,
+				12,
+				0,
+			}
+
+			for i := 0; i < 13; i++ {
+				req.Equal(steps[i], trace[i].StepID)
+			}
+		})
+	}
+}

tests/workflows/0011_termination_explicit_test.go

@@ -0,0 +1,27 @@
+package workflows
+
+import (
+	"context"
+	"testing"
+
+	autTypes "github.com/cortezaproject/corteza-server/automation/types"
+	"github.com/stretchr/testify/require"
+)
+
+func Test0011_termination_explicit(t *testing.T) {
+	var (
+		ctx = bypassRBAC(context.Background())
+		req = require.New(t)
+	)
+
+	loadScenario(ctx, t)
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	req.Len(trace, 3)
+	req.Equal(uint64(10), trace[0].StepID)
+	req.Equal(uint64(11), trace[1].StepID)
+	req.Equal(uint64(0), trace[2].StepID)
+}

tests/workflows/0012_termination_implicit_test.go

@@ -0,0 +1,26 @@
+package workflows
+
+import (
+	"context"
+	"testing"
+
+	autTypes "github.com/cortezaproject/corteza-server/automation/types"
+	"github.com/stretchr/testify/require"
+)
+
+func Test0012_termination_implicit(t *testing.T) {
+	var (
+		ctx = bypassRBAC(context.Background())
+		req = require.New(t)
+	)
+
+	loadScenario(ctx, t)
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	req.Len(trace, 2)
+	req.Equal(uint64(10), trace[0].StepID)
+	req.Equal(uint64(0), trace[1].StepID)
+}

tests/workflows/0013_gateway_exclusive_test.go

@@ -0,0 +1,48 @@
+package workflows
+
+import (
+	"context"
+	"testing"
+
+	autTypes "github.com/cortezaproject/corteza-server/automation/types"
+	"github.com/stretchr/testify/require"
+)
+
+func Test0013_gateway_exclusive(t *testing.T) {
+	var (
+		ctx = bypassRBAC(context.Background())
+		req = require.New(t)
+	)
+
+	loadScenario(ctx, t)
+
+	t.Run("first path match", func(t *testing.T) {
+		_, trace := mustExecWorkflow(ctx, t, "case1", autTypes.WorkflowExecParams{})
+
+		req.Len(trace, 4)
+		req.Equal(uint64(10), trace[0].StepID)
+		req.Equal(uint64(11), trace[1].StepID)
+		req.Equal(uint64(14), trace[2].StepID)
+		req.Equal(uint64(0), trace[3].StepID)
+	})
+
+	t.Run("second path match", func(t *testing.T) {
+		_, trace := mustExecWorkflow(ctx, t, "case2", autTypes.WorkflowExecParams{})
+
+		req.Len(trace, 4)
+		req.Equal(uint64(10), trace[0].StepID)
+		req.Equal(uint64(12), trace[1].StepID)
+		req.Equal(uint64(14), trace[2].StepID)
+		req.Equal(uint64(0), trace[3].StepID)
+	})
+
+	t.Run("default path match", func(t *testing.T) {
+		_, trace := mustExecWorkflow(ctx, t, "case3", autTypes.WorkflowExecParams{})
+
+		req.Len(trace, 4)
+		req.Equal(uint64(10), trace[0].StepID)
+		req.Equal(uint64(13), trace[1].StepID)
+		req.Equal(uint64(14), trace[2].StepID)
+		req.Equal(uint64(0), trace[3].StepID)
+	})
+}

tests/workflows/testdata/S0010_stacktrace/data_model.yaml

@@ -0,0 +1,25 @@
+namespaces:
+  ns1:
+    name: ns1 name
+
+modules:
+  mod1:
+    name: mod1 name
+    fields:
+      f1:
+        label: f1 label
+        kind: String
+        required: false
+
+records:
+  mod1:
+    - values:
+        f1: 1
+    - values:
+        f1: 2
+    - values:
+        f1: 3
+    - values:
+        f1: 4
+    - values:
+        f1: 5

tests/workflows/testdata/S0010_stacktrace/workflow.yaml

@@ -0,0 +1,29 @@
+workflows:
+  testing:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: iterator
+        ref: composeRecordsEach
+        arguments:
+          - { "target": "module", "value": "mod1", "type": "Handle" }
+          - { "target": "namespace", "value": "ns1", "type": "Handle" }
+        results:
+          - { "target": "r", "expr": "record" }
+          - { "target": "i", "expr": "index" }
+          - { "target": "t", "expr": "total" }
+
+      - stepID: 11
+        kind: continue
+
+      - stepID: 12
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11 }
+      - { parentID: 10, childID: 12 }

tests/workflows/testdata/S0011_termination_explicit/workflow.yaml

@@ -0,0 +1,19 @@
+workflows:
+  testing:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: expressions
+        arguments:
+          - { "target": "test", "value": "42", "type": "Any" }
+
+      - stepID: 11
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11 }

tests/workflows/testdata/S0012_termination_implicit/workflow.yaml

@@ -0,0 +1,13 @@
+workflows:
+  testing:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: expressions
+        arguments:
+          - { "target": "test", "value": "42", "type": "Any" }

tests/workflows/testdata/S0013_gateway_exclusive/workflow.yaml

@@ -0,0 +1,120 @@
+workflows:
+  case1:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: gateway
+        ref: excl
+
+      - stepID: 11
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 1", type: "String"}
+      - stepID: 12
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 2", type: "String"}
+      - stepID: 13
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 3", type: "String"}
+
+      - stepID: 14
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11, expr: "true == true" }
+      - { parentID: 10, childID: 12, expr: "true == true" }
+      - { parentID: 10, childID: 13, expr: "" }
+
+      - { parentID: 11, childID: 14 }
+      - { parentID: 12, childID: 14 }
+      - { parentID: 13, childID: 14 }
+
+  case2:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: gateway
+        ref: excl
+
+      - stepID: 11
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 1", type: "String"}
+      - stepID: 12
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 2", type: "String"}
+      - stepID: 13
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 3", type: "String"}
+
+      - stepID: 14
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11, expr: "true == false" }
+      - { parentID: 10, childID: 12, expr: "true == true" }
+      - { parentID: 10, childID: 13, expr: "" }
+
+      - { parentID: 11, childID: 14 }
+      - { parentID: 12, childID: 14 }
+      - { parentID: 13, childID: 14 }
+
+  case3:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: gateway
+        ref: excl
+
+      - stepID: 11
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 1", type: "String"}
+      - stepID: 12
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 2", type: "String"}
+      - stepID: 13
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 3", type: "String"}
+
+      - stepID: 14
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11, expr: "true == false" }
+      - { parentID: 10, childID: 12, expr: "true == false" }
+      - { parentID: 10, childID: 13, expr: "" }
+
+      - { parentID: 11, childID: 14 }
+      - { parentID: 12, childID: 14 }
+      - { parentID: 13, childID: 14 }

tests/workflows/testdata/S0014_gateway_fork/workflow.yaml

@@ -0,0 +1,46 @@
+workflows:
+  case1:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: gateway
+        ref: fork
+
+      - stepID: 11
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 1", type: "String"}
+      - stepID: 12
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 2", type: "String"}
+      - stepID: 13
+        kind: function
+        ref: logInfo
+        arguments:
+          - {target: "message", value: "log 3", type: "String"}
+
+      - stepID: 14
+        kind: gateway
+        ref: join
+
+      - stepID: 15
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11 }
+      - { parentID: 10, childID: 12 }
+      - { parentID: 10, childID: 13 }
+
+      - { parentID: 11, childID: 14 }
+      - { parentID: 12, childID: 14 }
+      - { parentID: 13, childID: 14 }
+
+      - { parentID: 14, childID: 15 }