Merge pull request #42 from cschleiden/remove-pending-commands

Remove commands after they've been executed

Author: cschleiden

internal/workflow/executor.go

@@ -64,6 +64,8 @@ func NewExecutor(logger log.Logger, registry *Registry, historyProvider Workflow
 }
 
 func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*ExecutionResult, error) {
+	e.workflowState.ClearCommands()
+
 	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)
 

internal/workflow/executor_test.go

@@ -436,3 +436,61 @@ func Test_ExecuteWorkflowWithSignal(t *testing.T) {
 	require.True(t, e.workflow.Completed())
 	require.Len(t, e.workflowState.Commands(), 1)
 }
+
+func Test_ClearCommandsBetweenRuns(t *testing.T) {
+	r := NewRegistry()
+
+	workflowActivityHit = 0
+
+	r.RegisterWorkflow(workflowWithActivity)
+	r.RegisterActivity(activity1)
+
+	task1 := &task.Workflow{
+		ID:               "oldtaskid",
+		WorkflowInstance: core.NewWorkflowInstance("instanceID", "executionID"),
+		NewEvents: []history.Event{
+			history.NewPendingEvent(
+				time.Now(),
+				history.EventType_WorkflowExecutionStarted,
+				&history.ExecutionStartedAttributes{
+					Name:   "workflowWithActivity",
+					Inputs: []payload.Payload{},
+				},
+			),
+		},
+	}
+
+	historyProvider := &testHistoryProvider{[]history.Event{}}
+	e := newExecutor(r, task1.WorkflowInstance, historyProvider)
+
+	r1, err := e.ExecuteTask(context.Background(), task1)
+	require.NoError(t, err)
+	require.Equal(t, 1, workflowActivityHit)
+	require.False(t, e.workflow.Completed())
+	require.Len(t, e.workflowState.Commands(), 1)
+
+	historyProvider.history = r1.Executed
+
+	task2 := &task.Workflow{
+		ID:               "oldtaskid",
+		WorkflowInstance: core.NewWorkflowInstance("instanceID", "executionID"),
+		NewEvents: []history.Event{
+			history.NewPendingEvent(
+				time.Now(),
+				history.EventType_SignalReceived,
+				&history.SignalReceivedAttributes{
+					Name: "signalr`",
+					Arg:  []byte("arg"),
+				},
+			),
+		},
+		LastSequenceID: 4,
+	}
+
+	r2, err := e.ExecuteTask(context.Background(), task2)
+	require.NoError(t, err)
+	require.Equal(t, 1, workflowActivityHit)
+	require.False(t, e.workflow.Completed())
+	require.Len(t, e.workflowState.Commands(), 0)
+	require.Len(t, r2.Executed, 3)
+}

samples/timer/timer.go

@@ -3,8 +3,6 @@ package main
 import (
 	"context"
 	"log"
-	"os"
-	"os/signal"
 	"time"
 
 	"github.com/cschleiden/go-workflows/backend"
@@ -27,18 +25,16 @@ func main() {
 	}
 
 	// Run worker
-	go RunWorker(ctx, b)
+	w := RunWorker(ctx, b)
 
 	// Start workflow via client
 	c := client.New(b)
 
 	startWorkflow(ctx, c)
 
-	c2 := make(chan os.Signal, 1)
-	signal.Notify(c2, os.Interrupt)
-	<-c2
-
 	cancel()
+
+	w.Stop()
 }
 
 func startWorkflow(ctx context.Context, c client.Client) {
@@ -49,10 +45,20 @@ func startWorkflow(ctx context.Context, c client.Client) {
 		panic("could not start workflow")
 	}
 
-	log.Println("Started workflow", wf.InstanceID)
+	for i := 0; i < 10; i++ {
+		time.Sleep(time.Millisecond * 200)
+		c.SignalWorkflow(ctx, wf.InstanceID, "Hello world", "value")
+	}
+
+	result, err := client.GetWorkflowResult[string](ctx, c, wf, time.Second*10)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	log.Println("Workflow finished. Result:", result)
 }
 
-func RunWorker(ctx context.Context, mb backend.Backend) {
+func RunWorker(ctx context.Context, mb backend.Backend) worker.Worker {
 	w := worker.New(mb, nil)
 
 	w.RegisterWorkflow(Workflow1)
@@ -62,6 +68,8 @@ func RunWorker(ctx context.Context, mb backend.Backend) {
 	if err := w.Start(ctx); err != nil {
 		panic("could not start worker")
 	}
+
+	return w
 }
 
 func Workflow1(ctx workflow.Context, msg string) (string, error) {
@@ -72,12 +80,11 @@ func Workflow1(ctx workflow.Context, msg string) (string, error) {
 	a1 := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1, 35, 12)
 
 	tctx, _ := workflow.WithCancel(ctx)
-	t := workflow.ScheduleTimer(tctx, 2*time.Second)
 	// cancel()
 
 	workflow.Select(
 		ctx,
-		workflow.Await(t, func(ctx workflow.Context, f workflow.Future[struct{}]) {
+		workflow.Await(workflow.ScheduleTimer(tctx, 2*time.Second), func(ctx workflow.Context, f workflow.Future[struct{}]) {
 			if _, err := f.Get(ctx); err != nil {
 				logger.Debug("Timer canceled")
 			} else {