Merge pull request #118 from cschleiden/fix-error-during-replay

Ensure errors encountered during replay can be saved & Fix SideEffect determinism check

Author: cschleiden

backend/redis/events.go

@@ -31,7 +31,7 @@ func addEventToStreamP(ctx context.Context, p redis.Pipeliner, streamKey string,
 // ARGV[1] - event data as serialized strings
 var addEventsToStreamCmd = redis.NewScript(`
 	local msgID = ""
-	for i = 1, #ARGV,2 do
+	for i = 1, #ARGV, 2 do
 		msgID = redis.call("XADD", KEYS[1], ARGV[i], "event", ARGV[i + 1])
 	end
 	return msgID
@@ -45,6 +45,8 @@ func addEventsToHistoryStreamP(ctx context.Context, p redis.Pipeliner, streamKey
 			return err
 		}
 
+		// log.Println("addEventsToHistoryStreamP:", event.SequenceID, string(eventData))
+
 		eventsData = append(eventsData, historyID(event.SequenceID))
 		eventsData = append(eventsData, string(eventData))
 	}

backend/test/e2e.go

@@ -20,8 +20,9 @@ import (
 
 func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b TestBackend)) {
 	tests := []struct {
-		name string
-		f    func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend)
+		name         string
+		withoutCache bool // If set, test will only be run when the cache is disabled
+		f            func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend)
 	}{
 		{
 			name: "SimpleWorkflow",
@@ -125,6 +126,35 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 				require.ErrorContains(t, err, "converting activity inputs: mismatched argument count: expected 2, got 1")
 			},
 		},
+		{
+			name: "SideEffect_Simple",
+			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {
+				i := 2
+				wf := func(ctx workflow.Context) (int, error) {
+					r1, _ := workflow.SideEffect(ctx, func(ctx workflow.Context) int {
+						i++
+						return i
+					}).Get(ctx)
+
+					// Do something to force the task to end
+					workflow.Sleep(ctx, time.Millisecond*1)
+
+					r2, _ := workflow.SideEffect(ctx, func(ctx workflow.Context) int {
+						i++
+						return i
+					}).Get(ctx)
+
+					return r1 + r2, nil
+				}
+				register(t, ctx, w, []interface{}{wf}, nil)
+
+				instance := runWorkflow(t, ctx, c, wf)
+
+				r, err := client.GetWorkflowResult[int](ctx, c, instance, time.Second*5)
+				require.NoError(t, err)
+				require.Equal(t, 7, r)
+			},
+		},
 		{
 			name: "SubWorkflow_Simple",
 			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {
@@ -363,10 +393,46 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 				require.Len(t, futureEvents, 0, "no future events should be scheduled")
 			},
 		},
+		{
+			name:         "NonDeterminism",
+			withoutCache: true,
+			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {
+				i := 0
+				wf := func(ctx workflow.Context) (int, error) {
+					var r int
+
+					i++
+					if i%2 == 0 {
+						r, _ = workflow.SideEffect(ctx, func(ctx workflow.Context) int {
+							return 1
+						}).Get(ctx)
+					} else {
+						workflow.Sleep(ctx, time.Millisecond*1)
+					}
+
+					// Do something to force the task to end
+					workflow.Sleep(ctx, time.Millisecond*1)
+
+					return r, nil
+				}
+				register(t, ctx, w, []interface{}{wf}, nil)
+
+				instance := runWorkflow(t, ctx, c, wf)
+
+				r, err := client.GetWorkflowResult[int](ctx, c, instance, time.Second*5)
+				require.NoError(t, err)
+				require.Equal(t, 0, r)
+			},
+		},
 	}
 
 	run := func(suffix string, workerOptions *worker.Options) {
 		for _, tt := range tests {
+			if tt.withoutCache && workerOptions.WorkflowExecutorCache != nil {
+				// Skip test
+				continue
+			}
+
 			t.Run(tt.name+suffix, func(t *testing.T) {
 				b := setup()
 				ctx := context.Background()

internal/command/sideeffect.go

@@ -14,16 +14,19 @@ type SideEffectCommand struct {
 
 var _ Command = (*SideEffectCommand)(nil)
 
-func NewSideEffectCommand(id int64, result payload.Payload) *SideEffectCommand {
+func NewSideEffectCommand(id int64) *SideEffectCommand {
 	return &SideEffectCommand{
 		command: command{
 			state: CommandState_Pending,
 			id:    id,
 		},
-		result: result,
 	}
 }
 
+func (c *SideEffectCommand) SetResult(result payload.Payload) {
+	c.result = result
+}
+
 func (c *SideEffectCommand) Type() string {
 	return "SideEffect"
 }

internal/workflow/executor.go

@@ -115,6 +115,9 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 			// Fail workflow with an error. Skip executing new events, but still go through the commands
 			e.workflowCompleted(nil, err)
 			skipNewEvents = true
+
+			// With an error occurred during replay, we need to ensure new events don't get duplicate sequence ids
+			e.lastSequenceID = t.LastSequenceID
 		} else if t.LastSequenceID != e.lastSequenceID {
 			logger.Error("After replaying history, task still has newer history than current state", "task_sequence_id", t.LastSequenceID, "local_sequence_id", e.lastSequenceID)
 
@@ -494,12 +497,12 @@ func (e *executor) handleSignalReceived(event history.Event, a *history.SignalRe
 func (e *executor) handleSideEffectResult(event history.Event, a *history.SideEffectResultAttributes) error {
 	c := e.workflowState.CommandByScheduleEventID(event.ScheduleEventID)
 	if c == nil {
-		return fmt.Errorf("previous workflow execution scheduled a sub workflow")
+		return fmt.Errorf("previous workflow execution scheduled a side effect")
 	}
 
 	sec, ok := c.(*command.SideEffectCommand)
 	if !ok {
-		return fmt.Errorf("previous workflow execution scheduled a sub workflow, not: %v", c.Type())
+		return fmt.Errorf("previous workflow execution scheduled a side effect, not: %v", c.Type())
 	}
 
 	sec.Done()

workflow/sideeffect.go

@@ -22,26 +22,23 @@ func SideEffect[TResult any](ctx Context, f func(ctx Context) TResult) Future[TR
 	wfState := workflowstate.WorkflowState(ctx)
 	scheduleEventID := wfState.GetNextScheduleEventID()
 
-	if Replaying(ctx) {
-		// There has to be a message in the history with the result, create a new future
-		// and block on it
-		wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(future))
-		return future
-	}
+	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(future))
 
-	// Execute side effect
-	r := f(ctx)
+	cmd := command.NewSideEffectCommand(scheduleEventID)
+	wfState.AddCommand(cmd)
 
-	// Create command to add it to the history
-	payload, err := converter.DefaultConverter.To(r)
-	if err != nil {
-		future.Set(*new(TResult), err)
-	}
+	if !Replaying(ctx) {
+		// Execute side effect
+		r := f(ctx)
 
-	cmd := command.NewSideEffectCommand(scheduleEventID, payload)
-	wfState.AddCommand(cmd)
+		payload, err := converter.DefaultConverter.To(r)
+		if err != nil {
+			future.Set(*new(TResult), err)
+		}
 
-	future.Set(r, nil)
+		cmd.SetResult(payload)
+		future.Set(r, nil)
+	}
 
 	return future
 }