Merge pull request #51 from cschleiden/workflow-errors

Transition workflow to an error state when encountering a panic or error

Author: cschleiden

internal/sync/coroutine.go

@@ -85,7 +85,8 @@ func newState() *coState {
 	return &coState{
 		blocking: make(chan bool, 1),
 		unblock:  make(chan bool),
-		logger:   log.New(io.Discard, "[co]", log.LstdFlags),
+		// Mostly used while debugging issues, default to discarding log messages
+		logger: log.New(io.Discard, "[co]", log.LstdFlags),
 		// logger:            log.New(os.Stderr, fmt.Sprintf("[co %v]", i), log.Lmsgprefix|log.Ltime),
 		deadlockDetection: DeadlockDetection,
 	}

internal/tester/tester_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cschleiden/go-workflows/internal/sync"
 	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -37,7 +38,7 @@ func Test_WorkflowBlocked(t *testing.T) {
 }
 
 func workflowBlocked(ctx workflow.Context) error {
-	var f workflow.Future[int]
+	f := sync.NewFuture[int]()
 	f.Get(ctx)
 
 	return nil

internal/workflow/executor.go

@@ -67,6 +67,8 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 
 	e.workflowState.ClearCommands()
 
+	skipNewEvents := false
+
 	if t.LastSequenceID > e.lastSequenceID {
 		e.logger.Debug("Task has newer history than current state, fetching and replaying history", "task_sequence_id", t.LastSequenceID, "sequence_id", e.lastSequenceID)
 
@@ -77,7 +79,10 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 
 		if err := e.replayHistory(h); err != nil {
 			e.logger.Error("Error while replaying history", "error", err)
-			return nil, fmt.Errorf("replaying history: %w", err)
+
+			// Fail workflow with an error. Skip executing new events, but still go through the commands
+			e.workflowCompleted(nil, err)
+			skipNewEvents = true
 		}
 
 		if t.LastSequenceID != e.lastSequenceID {
@@ -87,18 +92,24 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 		return nil, fmt.Errorf("task has older history than current state, cannot execute")
 	}
 
-	// Always pad the received events with WorkflowTaskStarted/Finished evnets to indicate the execution
+	// Always add a WorkflowTaskStarted event before executing new tasks
 	toExecute := []history.Event{e.createNewEvent(history.EventType_WorkflowTaskStarted, &history.WorkflowTaskStartedAttributes{})}
+	executedEvents := toExecute
+
 	toExecute = append(toExecute, t.NewEvents...)
 
 	// Execute new events received from the backend
-	if err := e.executeNewEvents(toExecute); err != nil {
-		e.logger.Error("Error while executing new events", "error", err)
-		return nil, fmt.Errorf("executing new events: %w", err)
-	}
+	if !skipNewEvents {
+		var err error
+		executedEvents, err = e.executeNewEvents(toExecute)
+		if err != nil {
+			e.logger.Error("Error while executing new events", "error", err)
 
-	executedEvents := toExecute
+			e.workflowCompleted(nil, err)
+		}
+	}
 
+	// Process any commands added while executing new events
 	completed, newCommandEvents, activityEvents, workflowEvents, err := e.processCommands(ctx, t)
 	if err != nil {
 		return nil, fmt.Errorf("processing commands: %w", err)
@@ -115,6 +126,8 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 		"task_id", t.ID,
 		"instance_id", t.WorkflowInstance.InstanceID,
 		"executed", len(executedEvents),
+		"last_sequence_id", e.lastSequenceID,
+		"completed", e.workflow.Completed(),
 	)
 
 	return &ExecutionResult{
@@ -129,7 +142,7 @@ func (e *executor) replayHistory(history []history.Event) error {
 	e.workflowState.SetReplaying(true)
 	for _, event := range history {
 		if err := e.executeEvent(event); err != nil {
-			return fmt.Errorf("executeing history event: %w", err)
+			return err
 		}
 
 		e.lastSequenceID = event.SequenceID
@@ -138,22 +151,20 @@ func (e *executor) replayHistory(history []history.Event) error {
 	return nil
 }
 
-func (e *executor) executeNewEvents(newEvents []history.Event) error {
+func (e *executor) executeNewEvents(newEvents []history.Event) ([]history.Event, error) {
 	e.workflowState.SetReplaying(false)
 
-	for _, event := range newEvents {
+	for i, event := range newEvents {
 		if err := e.executeEvent(event); err != nil {
-			return fmt.Errorf("executing event: %w", err)
+			return newEvents[:i], err
 		}
 	}
 
 	if e.workflow.Completed() {
-		if err := e.workflowCompleted(e.workflow.Result(), e.workflow.Error()); err != nil {
-			return err
-		}
+		e.workflowCompleted(e.workflow.Result(), e.workflow.Error())
 	}
 
-	return nil
+	return newEvents, nil
 }
 
 func (e *executor) Close() {
@@ -373,13 +384,11 @@ func (e *executor) handleSideEffectResult(event history.Event, a *history.SideEf
 	return e.workflow.Continue(e.workflowCtx)
 }
 
-func (e *executor) workflowCompleted(result payload.Payload, err error) error {
+func (e *executor) workflowCompleted(result payload.Payload, err error) {
 	eventId := e.workflowState.GetNextScheduleEventID()
 
 	cmd := command.NewCompleteWorkflowCommand(eventId, result, err)
 	e.workflowState.AddCommand(&cmd)
-
-	return nil
 }
 
 func (e *executor) processCommands(ctx context.Context, t *task.Workflow) (bool, []history.Event, []history.Event, []history.WorkflowEvent, error) {

internal/workflow/executor_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/command"
 	"github.com/cschleiden/go-workflows/internal/converter"
 	"github.com/cschleiden/go-workflows/internal/core"
+	"github.com/cschleiden/go-workflows/internal/fn"
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/logger"
 	"github.com/cschleiden/go-workflows/internal/payload"
@@ -29,14 +30,14 @@ func (t *testHistoryProvider) GetWorkflowInstanceHistory(ctx context.Context, in
 	return t.history, nil
 }
 
-func newExecutor(r *Registry, i *core.WorkflowInstance, historyProvider WorkflowHistoryProvider) *executor {
+func newExecutor(r *Registry, i *core.WorkflowInstance, workflow interface{}, historyProvider WorkflowHistoryProvider) *executor {
 	logger := logger.NewDefaultLogger()
 	s := workflowstate.NewWorkflowState(i, logger, clock.New())
 	wfCtx, cancel := sync.WithCancel(workflowstate.WithWorkflowState(sync.Background(), s))
 
 	return &executor{
 		registry:          r,
-		workflow:          NewWorkflow(reflect.ValueOf(workflow1)),
+		workflow:          NewWorkflow(reflect.ValueOf(workflow)),
 		historyProvider:   historyProvider,
 		workflowState:     s,
 		workflowCtx:       wfCtx,
@@ -52,15 +53,13 @@ func activity1(ctx context.Context, r int) (int, error) {
 	return r, nil
 }
 
-var workflowHits int
-
-func workflow1(ctx sync.Context) error {
-	workflowHits++
-
-	return nil
-}
-
 func Test_ExecuteWorkflow(t *testing.T) {
+	var workflowHits int
+	workflow1 := func(ctx sync.Context) error {
+		workflowHits++
+		return nil
+	}
+
 	r := NewRegistry()
 
 	r.RegisterWorkflow(workflow1)
@@ -74,14 +73,14 @@ func Test_ExecuteWorkflow(t *testing.T) {
 				time.Now(),
 				history.EventType_WorkflowExecutionStarted,
 				&history.ExecutionStartedAttributes{
-					Name:   "workflow1",
+					Name:   fn.Name(workflow1),
 					Inputs: []payload.Payload{},
 				},
 			),
 		},
 	}
 
-	e := newExecutor(r, task.WorkflowInstance, &testHistoryProvider{})
+	e := newExecutor(r, task.WorkflowInstance, workflow1, &testHistoryProvider{})
 
 	_, err := e.ExecuteTask(context.Background(), task)
 	require.NoError(t, err)
@@ -125,13 +124,13 @@ func Test_ReplayWorkflowWithActivityResult(t *testing.T) {
 		LastSequenceID:   3,
 	}
 
-	e := newExecutor(r, task.WorkflowInstance, &testHistoryProvider{[]history.Event{
+	e := newExecutor(r, task.WorkflowInstance, workflowWithActivity, &testHistoryProvider{[]history.Event{
 		history.NewHistoryEvent(
 			1,
 			time.Now(),
 			history.EventType_WorkflowExecutionStarted,
 			&history.ExecutionStartedAttributes{
-				Name:   "workflowWithActivity",
+				Name:   fn.Name(workflowWithActivity),
 				Inputs: []payload.Payload{inputs},
 			},
 		),
@@ -181,14 +180,14 @@ func Test_ExecuteWorkflowWithActivityCommand(t *testing.T) {
 				time.Now(),
 				history.EventType_WorkflowExecutionStarted,
 				&history.ExecutionStartedAttributes{
-					Name:   "workflowWithActivity",
+					Name:   fn.Name(workflowWithActivity),
 					Inputs: []payload.Payload{},
 				},
 			),
 		},
 	}
 
-	e := newExecutor(r, task.WorkflowInstance, &testHistoryProvider{})
+	e := newExecutor(r, task.WorkflowInstance, workflowWithActivity, &testHistoryProvider{})
 
 	e.ExecuteTask(context.Background(), task)
 
@@ -237,14 +236,14 @@ func Test_ExecuteWorkflowWithTimer(t *testing.T) {
 				time.Now(),
 				history.EventType_WorkflowExecutionStarted,
 				&history.ExecutionStartedAttributes{
-					Name:   "workflowWithTimer",
+					Name:   fn.Name(workflowWithTimer),
 					Inputs: []payload.Payload{},
 				},
 			),
 		},
 	}
 
-	e := newExecutor(r, task.WorkflowInstance, &testHistoryProvider{})
+	e := newExecutor(r, task.WorkflowInstance, workflowWithTimer, &testHistoryProvider{})
 
 	e.ExecuteTask(context.Background(), task)
 
@@ -293,14 +292,14 @@ func Test_ExecuteWorkflowWithSelector(t *testing.T) {
 				time.Now(),
 				history.EventType_WorkflowExecutionStarted,
 				&history.ExecutionStartedAttributes{
-					Name:   "workflowWithSelector",
+					Name:   fn.Name(workflowWithSelector),
 					Inputs: []payload.Payload{},
 				},
 			),
 		},
 	}
 
-	e := newExecutor(r, task.WorkflowInstance, &testHistoryProvider{})
+	e := newExecutor(r, task.WorkflowInstance, workflowWithSelector, &testHistoryProvider{})
 
 	e.ExecuteTask(context.Background(), task)
 
@@ -330,7 +329,7 @@ func Test_ExecuteNewEvents(t *testing.T) {
 				time.Now(),
 				history.EventType_WorkflowExecutionStarted,
 				&history.ExecutionStartedAttributes{
-					Name:   "workflowWithActivity",
+					Name:   fn.Name(workflowWithActivity),
 					Inputs: []payload.Payload{inputs},
 				},
 			),
@@ -346,7 +345,7 @@ func Test_ExecuteNewEvents(t *testing.T) {
 		},
 	}
 
-	e := newExecutor(r, oldTask.WorkflowInstance, &testHistoryProvider{[]history.Event{}})
+	e := newExecutor(r, oldTask.WorkflowInstance, workflowWithActivity, &testHistoryProvider{[]history.Event{}})
 
 	taskResult, err := e.ExecuteTask(context.Background(), oldTask)
 
@@ -385,21 +384,21 @@ func Test_ExecuteNewEvents(t *testing.T) {
 	require.Len(t, e.workflowState.Commands(), 1)
 }
 
-var workflowSignalHits int
+func Test_ExecuteWorkflowWithSignal(t *testing.T) {
+	r := NewRegistry()
 
-func workflowWithSignal1(ctx sync.Context) error {
-	c := wf.NewSignalChannel[string](ctx, "signal1")
-	c.Receive(ctx)
+	var workflowSignalHits int
 
-	workflowSignalHits++
+	workflowWithSignal := func(ctx sync.Context) error {
+		c := wf.NewSignalChannel[string](ctx, "signal1")
+		c.Receive(ctx)
 
-	return nil
-}
+		workflowSignalHits++
 
-func Test_ExecuteWorkflowWithSignal(t *testing.T) {
-	r := NewRegistry()
+		return nil
+	}
 
-	r.RegisterWorkflow(workflowWithSignal1)
+	r.RegisterWorkflow(workflowWithSignal)
 
 	s, err := converter.DefaultConverter.To("")
 	require.NoError(t, err)
@@ -412,7 +411,7 @@ func Test_ExecuteWorkflowWithSignal(t *testing.T) {
 				time.Now(),
 				history.EventType_WorkflowExecutionStarted,
 				&history.ExecutionStartedAttributes{
-					Name:   "workflowWithSignal1",
+					Name:   fn.Name(workflowWithSignal),
 					Inputs: []payload.Payload{},
 				},
 			),
@@ -427,7 +426,7 @@ func Test_ExecuteWorkflowWithSignal(t *testing.T) {
 		},
 	}
 
-	e := newExecutor(r, task.WorkflowInstance, &testHistoryProvider{})
+	e := newExecutor(r, task.WorkflowInstance, workflowWithSignal, &testHistoryProvider{})
 
 	_, err = e.ExecuteTask(context.Background(), task)
 	require.NoError(t, err)
@@ -437,6 +436,40 @@ func Test_ExecuteWorkflowWithSignal(t *testing.T) {
 	require.Len(t, e.workflowState.Commands(), 1)
 }
 
+func Test_CompletesWorkflowOnError(t *testing.T) {
+	r := NewRegistry()
+
+	workflowPanic := func(ctx sync.Context) error {
+		panic("wf error")
+	}
+
+	r.RegisterWorkflow(workflowPanic)
+
+	task1 := &task.Workflow{
+		ID:               "taskid",
+		WorkflowInstance: core.NewWorkflowInstance("instanceID", "executionID"),
+		NewEvents: []history.Event{
+			history.NewPendingEvent(
+				time.Now(),
+				history.EventType_WorkflowExecutionStarted,
+				&history.ExecutionStartedAttributes{
+					Name:   fn.Name(workflowPanic),
+					Inputs: []payload.Payload{},
+				},
+			),
+		},
+	}
+
+	historyProvider := &testHistoryProvider{[]history.Event{}}
+	e := newExecutor(r, task1.WorkflowInstance, workflowPanic, historyProvider)
+
+	r1, err := e.ExecuteTask(context.Background(), task1)
+	require.NoError(t, err)
+	require.True(t, e.workflow.Completed())
+	require.Len(t, e.workflowState.Commands(), 1)
+	require.True(t, r1.Completed)
+}
+
 func Test_ClearCommandsBetweenRuns(t *testing.T) {
 	r := NewRegistry()
 
@@ -461,7 +494,7 @@ func Test_ClearCommandsBetweenRuns(t *testing.T) {
 	}
 
 	historyProvider := &testHistoryProvider{[]history.Event{}}
-	e := newExecutor(r, task1.WorkflowInstance, historyProvider)
+	e := newExecutor(r, task1.WorkflowInstance, workflowWithActivity, historyProvider)
 
 	r1, err := e.ExecuteTask(context.Background(), task1)
 	require.NoError(t, err)