Merge pull request #109 from cschleiden/fix-uncached-command-handling

Fix uncached command handling

Author: cschleiden

.github/workflows/go.yml

@@ -38,7 +38,7 @@ jobs:
     - name: Tests
       run: |
         go install github.com/jstemmer/go-junit-report/v2@latest
-        go test -race -count 3 -v ./... 2>&1 | go-junit-report -set-exit-code -iocopy -out "${{ github.workspace }}/report.xml"
+        go test -race -count 2 -v ./... 2>&1 | go-junit-report -set-exit-code -iocopy -out "${{ github.workspace }}/report.xml"
 
 
     - name: Test Summary

backend/mysql/mysql.go

@@ -517,7 +517,7 @@ func (b *mysqlBackend) CompleteWorkflowTask(
 			if event.Type == history.EventType_WorkflowExecutionStarted {
 				a := event.Attributes.(*history.ExecutionStartedAttributes)
 				// Create new instance
-				if err := createInstance(ctx, tx, targetInstance, a.Metadata, true); err != nil {
+				if err := createInstance(ctx, tx, &targetInstance, a.Metadata, true); err != nil {
 					return err
 				}
 

backend/redis/workflow.go

@@ -210,7 +210,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 			if event.Type == history.EventType_WorkflowExecutionStarted {
 				// Create new instance
 				a := event.Attributes.(*history.ExecutionStartedAttributes)
-				if err := createInstanceP(ctx, p, targetInstance, a.Metadata, true); err != nil {
+				if err := createInstanceP(ctx, p, &targetInstance, a.Metadata, true); err != nil {
 					return err
 				}
 			}
@@ -222,7 +222,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 		}
 
 		// Try to queue workflow task
-		if targetInstance != instance {
+		if targetInstance != *instance {
 			if err := rb.workflowQueue.Enqueue(ctx, p, targetInstance.InstanceID, nil); err != nil {
 				return fmt.Errorf("enqueuing workflow task: %w", err)
 			}

backend/sqlite/sqlite.go

@@ -406,7 +406,7 @@ func (sb *sqliteBackend) CompleteWorkflowTask(
 			if event.Type == history.EventType_WorkflowExecutionStarted {
 				a := event.Attributes.(*history.ExecutionStartedAttributes)
 				// Create new instance
-				if err := createInstance(ctx, tx, targetInstance, a.Metadata, true); err != nil {
+				if err := createInstance(ctx, tx, &targetInstance, a.Metadata, true); err != nil {
 					return err
 				}
 

backend/test/e2e.go

@@ -11,6 +11,7 @@ import (
 	"github.com/cschleiden/go-workflows/client"
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
+	internalwf "github.com/cschleiden/go-workflows/internal/workflow"
 	"github.com/cschleiden/go-workflows/worker"
 	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/google/uuid"
@@ -166,7 +167,8 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 					return i * 2, nil
 				}
 
-				ch := make(chan struct{}, 1)
+				// Workflow will be executed multiple times, but the test will wait only once. Create buffered channel
+				ch := make(chan struct{}, 10)
 
 				wf := func(ctx workflow.Context) (int, error) {
 					swfs := make([]workflow.Future[int], 0)
@@ -177,6 +179,7 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 					// Unblock test. Should not do this in production code, but here we know that this will be executed in the same process.
 					ch <- struct{}{}
 
+					// Wait for subworkflows to complete
 					r := 0
 
 					for _, f := range swfs {
@@ -362,30 +365,60 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 		},
 	}
 
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			b := setup()
-			ctx := context.Background()
-			ctx, cancel := context.WithCancel(ctx)
+	run := func(suffix string, workerOptions *worker.Options) {
+		for _, tt := range tests {
+			t.Run(tt.name+suffix, func(t *testing.T) {
+				b := setup()
+				ctx := context.Background()
+				ctx, cancel := context.WithCancel(ctx)
 
-			c := client.New(b)
-			w := worker.New(b, &worker.DefaultWorkerOptions)
+				c := client.New(b)
+				w := worker.New(b, workerOptions)
 
-			t.Cleanup(func() {
-				cancel()
-				if err := w.WaitForCompletion(); err != nil {
-					log.Println("Worker did not stop in time")
-					t.FailNow()
-				}
+				t.Cleanup(func() {
+					cancel()
+					if err := w.WaitForCompletion(); err != nil {
+						log.Println("Worker did not stop in time")
+						t.FailNow()
+					}
 
-				if teardown != nil {
-					teardown(b)
-				}
-			})
+					if teardown != nil {
+						teardown(b)
+					}
+				})
 
-			tt.f(t, ctx, c, w, b)
-		})
+				tt.f(t, ctx, c, w, b)
+			})
+		}
 	}
+
+	options := worker.DefaultWorkerOptions
+
+	// Run with cache
+	run("", &options)
+
+	// Disable cache for this execution
+	options.WorkflowExecutorCache = &noopWorkflowExecutorCache{}
+	run("_without_cache", &options)
+}
+
+type noopWorkflowExecutorCache struct {
+}
+
+var _ internalwf.ExecutorCache = (*noopWorkflowExecutorCache)(nil)
+
+// Get implements workflow.ExecutorCache
+func (*noopWorkflowExecutorCache) Get(ctx context.Context, instance *core.WorkflowInstance) (internalwf.WorkflowExecutor, bool, error) {
+	return nil, false, nil
+}
+
+// StartEviction implements workflow.ExecutorCache
+func (*noopWorkflowExecutorCache) StartEviction(ctx context.Context) {
+}
+
+// Store implements workflow.ExecutorCache
+func (*noopWorkflowExecutorCache) Store(ctx context.Context, instance *core.WorkflowInstance, workflow internalwf.WorkflowExecutor) error {
+	return nil
 }
 
 func register(t *testing.T, ctx context.Context, w worker.Worker, workflows []interface{}, activities []interface{}) {

internal/command/cancel_subworkflow.go

@@ -40,6 +40,7 @@ func (c *CancelSubWorkflowCommand) Commit(clock clock.Clock) *CommandResult {
 				&history.SubWorkflowCancellationRequestedAttributes{
 					SubWorkflowInstance: c.SubWorkflowInstance,
 				},
+				history.ScheduleEventID(c.id),
 			),
 		},
 

internal/command/cancel_timer.go

@@ -5,16 +5,16 @@ import (
 	"github.com/cschleiden/go-workflows/internal/history"
 )
 
-type cancelTimerCommand struct {
+type CancelTimerCommand struct {
 	command
 
 	TimerScheduleEventID int64
 }
 
-var _ Command = (*cancelTimerCommand)(nil)
+var _ Command = (*CancelTimerCommand)(nil)
 
-func NewCancelTimerCommand(id int64, timerScheduleEventID int64) *cancelTimerCommand {
-	return &cancelTimerCommand{
+func NewCancelTimerCommand(id int64, timerScheduleEventID int64) *CancelTimerCommand {
+	return &CancelTimerCommand{
 		command: command{
 			state: CommandState_Pending,
 			id:    id,
@@ -23,11 +23,11 @@ func NewCancelTimerCommand(id int64, timerScheduleEventID int64) *cancelTimerCom
 	}
 }
 
-func (*cancelTimerCommand) Type() string {
+func (*CancelTimerCommand) Type() string {
 	return "CancelTimer"
 }
 
-func (c *cancelTimerCommand) Commit(clock clock.Clock) *CommandResult {
+func (c *CancelTimerCommand) Commit(clock clock.Clock) *CommandResult {
 	c.commit()
 
 	return &CommandResult{

internal/command/command.go

@@ -7,17 +7,17 @@ import (
 
 type CommandState int
 
-//  ┌───────┐
-//  │Pending├
-//  └───────┘
-//      ▼
-// ┌─────────┐
-// │Committed│
-// └─────────┘
-//      ▼
-//   ┌────┐
-//   │Done│
-//   └────┘
+//        ┌───────┐
+// ┌──────┤Pending│ - Command is pending, has just been added
+// │      └───────┘
+// │          ▼
+// │     ┌─────────┐
+// │     │Committed│ - Command has been committed. Its results (e.g., events) have been checkpointed
+// │     └─────────┘
+// │          ▼
+// │       ┌────┐
+// └──────►│Done│ - Command has been marked as done.
+//         └────┘
 const (
 	CommandState_Pending CommandState = iota
 	CommandState_Committed
@@ -27,15 +27,17 @@ const (
 type Command interface {
 	ID() int64
 
+	Type() string
+
+	State() CommandState
+
+	Committed() bool
+
 	Commit(clock clock.Clock) *CommandResult
 
 	// Done marks the command as done. This transitions the state to done and indicates that the result
 	// of this command has been applied.
 	Done()
-
-	State() CommandState
-
-	Type() string
 }
 
 type CommandResult struct {
@@ -64,10 +66,15 @@ func (c *command) ID() int64 {
 	return c.id
 }
 
+func (c *command) Committed() bool {
+	return c.state >= CommandState_Committed
+}
+
 func (c *command) State() CommandState {
 	return c.state
 }
 
+// Done marks the command as done
 func (c *command) Done() {
 	c.state = CommandState_Done
 }

internal/command/schedule_subworkflow.go

@@ -49,7 +49,7 @@ func (c *ScheduleSubWorkflowCommand) Commit(clock clock.Clock) *CommandResult {
 	c.commit()
 
 	return &CommandResult{
-		// Record scheduled sub-workflow
+		// Record scheduled sub-workflow for source workflow instance
 		Events: []history.Event{
 			history.NewPendingEvent(
 				clock.Now(),

internal/command/sideeffect.go

@@ -6,16 +6,16 @@ import (
 	"github.com/cschleiden/go-workflows/internal/payload"
 )
 
-type sideEffectCommand struct {
+type SideEffectCommand struct {
 	command
 
 	result payload.Payload
 }
 
-var _ Command = (*sideEffectCommand)(nil)
+var _ Command = (*SideEffectCommand)(nil)
 
-func NewSideEffectCommand(id int64, result payload.Payload) *sideEffectCommand {
-	return &sideEffectCommand{
+func NewSideEffectCommand(id int64, result payload.Payload) *SideEffectCommand {
+	return &SideEffectCommand{
 		command: command{
 			state: CommandState_Pending,
 			id:    id,
@@ -24,11 +24,11 @@ func NewSideEffectCommand(id int64, result payload.Payload) *sideEffectCommand {
 	}
 }
 
-func (c *sideEffectCommand) Type() string {
+func (c *SideEffectCommand) Type() string {
 	return "SideEffect"
 }
 
-func (c *sideEffectCommand) Commit(clock clock.Clock) *CommandResult {
+func (c *SideEffectCommand) Commit(clock clock.Clock) *CommandResult {
 	c.commit()
 
 	return &CommandResult{