Merge pull request #100 from cschleiden/better-commands

Use command state machine

Author: cschleiden

backend/backend.go

@@ -52,7 +52,7 @@ type Backend interface {
 	// completed or other workflow instances.
 	CompleteWorkflowTask(
 		ctx context.Context, task *task.Workflow, instance *workflow.Instance, state WorkflowState,
-		executedEvents []history.Event, activityEvents []history.Event, workflowEvents []history.WorkflowEvent) error
+		executedEvents, activityEvents, timerEvents []history.Event, workflowEvents []history.WorkflowEvent) error
 
 	// GetActivityTask returns a pending activity task or nil if there are no pending activities
 	GetActivityTask(ctx context.Context) (*task.Activity, error)

backend/mock_Backend.go

@@ -407,8 +407,7 @@ func (b *mysqlBackend) CompleteWorkflowTask(
 	task *task.Workflow,
 	instance *workflow.Instance,
 	state backend.WorkflowState,
-	executedEvents []history.Event,
-	activityEvents []history.Event,
+	executedEvents, activityEvents, timerEvents []history.Event,
 	workflowEvents []history.WorkflowEvent,
 ) error {
 	tx, err := b.db.BeginTx(ctx, &sql.TxOptions{
@@ -475,6 +474,11 @@ func (b *mysqlBackend) CompleteWorkflowTask(
 		}
 	}
 
+	// Timer events
+	if err := insertPendingEvents(ctx, tx, instance.InstanceID, timerEvents); err != nil {
+		return fmt.Errorf("scheduling timers: %w", err)
+	}
+
 	for _, event := range executedEvents {
 		switch event.Type {
 		case history.EventType_TimerCanceled:

backend/mysql/mysql.go

@@ -163,8 +163,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 	task *task.Workflow,
 	instance *core.WorkflowInstance,
 	state backend.WorkflowState,
-	executedEvents []history.Event,
-	activityEvents []history.Event,
+	executedEvents, activityEvents, timerEvents []history.Event,
 	workflowEvents []history.WorkflowEvent,
 ) error {
 	instanceState, err := readInstance(ctx, rb.rdb, instance.InstanceID)
@@ -189,6 +188,13 @@ func (rb *redisBackend) CompleteWorkflowTask(
 		}
 	}
 
+	// Scheduler timers
+	for _, timerEvent := range timerEvents {
+		if err := addFutureEventP(ctx, p, instance, &timerEvent); err != nil {
+			return err
+		}
+	}
+
 	// Send new workflow events to the respective streams
 	groupedEvents := eventsByWorkflowInstance(workflowEvents)
 	for targetInstance, events := range groupedEvents {
@@ -200,27 +206,17 @@ func (rb *redisBackend) CompleteWorkflowTask(
 		}
 
 		// Insert pending events for target instance
-		msgAdded := false
-
 		for _, event := range events {
 			event := event
 
-			if event.VisibleAt != nil {
-				if err := addFutureEventP(ctx, p, targetInstance, &event); err != nil {
-					return err
-				}
-			} else {
-				// Add pending event to stream
-				if err := addEventToStreamP(ctx, p, pendingEventsKey(targetInstance.InstanceID), &event); err != nil {
-					return err
-				}
-
-				msgAdded = true
+			// Add pending event to stream
+			if err := addEventToStreamP(ctx, p, pendingEventsKey(targetInstance.InstanceID), &event); err != nil {
+				return err
 			}
 		}
 
 		// If any pending message was added, try to queue workflow task
-		if targetInstance != instance && msgAdded {
+		if targetInstance != instance {
 			if err := rb.workflowQueue.Enqueue(ctx, p, targetInstance.InstanceID, nil); err != nil {
 				if err != errTaskAlreadyInQueue {
 					return fmt.Errorf("adding instance to locked instances set: %w", err)

backend/redis/workflow.go

@@ -84,7 +84,7 @@ func scanEvent(row Scanner) (history.Event, error) {
 	return historyEvent, nil
 }
 
-func insertNewEvents(ctx context.Context, tx *sql.Tx, instanceID string, newEvents []history.Event) error {
+func insertPendingEvents(ctx context.Context, tx *sql.Tx, instanceID string, newEvents []history.Event) error {
 	return insertEvents(ctx, tx, "pending_events", instanceID, newEvents)
 }
 

backend/sqlite/events.go

@@ -76,7 +76,7 @@ func (sb *sqliteBackend) CreateWorkflowInstance(ctx context.Context, m history.W
 	}
 
 	// Initial history is empty, store only new events
-	if err := insertNewEvents(ctx, tx, m.WorkflowInstance.InstanceID, []history.Event{m.HistoryEvent}); err != nil {
+	if err := insertPendingEvents(ctx, tx, m.WorkflowInstance.InstanceID, []history.Event{m.HistoryEvent}); err != nil {
 		return fmt.Errorf("inserting new event: %w", err)
 	}
 
@@ -143,7 +143,7 @@ func (sb *sqliteBackend) CancelWorkflowInstance(ctx context.Context, instance *w
 		return err
 	}
 
-	if err := insertNewEvents(ctx, tx, instanceID, []history.Event{*event}); err != nil {
+	if err := insertPendingEvents(ctx, tx, instanceID, []history.Event{*event}); err != nil {
 		return fmt.Errorf("inserting cancellation event: %w", err)
 	}
 
@@ -200,7 +200,7 @@ func (sb *sqliteBackend) SignalWorkflow(ctx context.Context, instanceID string,
 		return backend.ErrInstanceNotFound
 	}
 
-	if err := insertNewEvents(ctx, tx, instanceID, []history.Event{event}); err != nil {
+	if err := insertPendingEvents(ctx, tx, instanceID, []history.Event{event}); err != nil {
 		return fmt.Errorf("inserting signal event: %w", err)
 	}
 
@@ -303,8 +303,7 @@ func (sb *sqliteBackend) CompleteWorkflowTask(
 	task *task.Workflow,
 	instance *workflow.Instance,
 	state backend.WorkflowState,
-	executedEvents []history.Event,
-	activityEvents []history.Event,
+	executedEvents, activityEvents, timerEvents []history.Event,
 	workflowEvents []history.WorkflowEvent,
 ) error {
 	tx, err := sb.db.BeginTx(ctx, nil)
@@ -365,6 +364,11 @@ func (sb *sqliteBackend) CompleteWorkflowTask(
 		}
 	}
 
+	// Timer events
+	if err := insertPendingEvents(ctx, tx, instance.InstanceID, timerEvents); err != nil {
+		return fmt.Errorf("scheduling timers: %w", err)
+	}
+
 	for _, event := range executedEvents {
 		switch event.Type {
 		case history.EventType_TimerCanceled:
@@ -393,7 +397,7 @@ func (sb *sqliteBackend) CompleteWorkflowTask(
 		}
 
 		// Insert pending events for target instance
-		if err := insertNewEvents(ctx, tx, targetInstance.InstanceID, events); err != nil {
+		if err := insertPendingEvents(ctx, tx, targetInstance.InstanceID, events); err != nil {
 			return fmt.Errorf("inserting messages: %w", err)
 		}
 	}
@@ -511,7 +515,7 @@ func (sb *sqliteBackend) CompleteActivityTask(ctx context.Context, instance *wor
 	}
 
 	// Insert new event generated during this workflow execution
-	if err := insertNewEvents(ctx, tx, instance.InstanceID, []history.Event{event}); err != nil {
+	if err := insertPendingEvents(ctx, tx, instance.InstanceID, []history.Event{event}); err != nil {
 		return fmt.Errorf("inserting new events for completed activity: %w", err)
 	}
 

backend/sqlite/sqlite.go

@@ -128,11 +128,11 @@ func BackendTest(t *testing.T, setup func() TestBackend, teardown func(b TestBac
 				require.NotNil(t, tk)
 
 				// Complete workflow task
-				err = b.CompleteWorkflowTask(ctx, tk, wfi, backend.WorkflowStateActive, tk.NewEvents, []history.Event{}, []history.WorkflowEvent{})
+				err = b.CompleteWorkflowTask(ctx, tk, wfi, backend.WorkflowStateActive, tk.NewEvents, []history.Event{}, []history.Event{}, []history.WorkflowEvent{})
 				require.NoError(t, err)
 
 				// Task is already completed, this should error
-				err = b.CompleteWorkflowTask(ctx, tk, wfi, backend.WorkflowStateActive, tk.NewEvents, []history.Event{}, []history.WorkflowEvent{})
+				err = b.CompleteWorkflowTask(ctx, tk, wfi, backend.WorkflowStateActive, tk.NewEvents, []history.Event{}, []history.Event{}, []history.WorkflowEvent{})
 				require.Error(t, err)
 			},
 		},
@@ -171,7 +171,7 @@ func BackendTest(t *testing.T, setup func() TestBackend, teardown func(b TestBac
 
 				workflowEvents := []history.WorkflowEvent{}
 
-				err = b.CompleteWorkflowTask(ctx, task, wfi, backend.WorkflowStateActive, events, activityEvents, workflowEvents)
+				err = b.CompleteWorkflowTask(ctx, task, wfi, backend.WorkflowStateActive, events, activityEvents, []history.Event{}, workflowEvents)
 				require.NoError(t, err)
 
 				time.Sleep(time.Second)
@@ -213,7 +213,7 @@ func BackendTest(t *testing.T, setup func() TestBackend, teardown func(b TestBac
 					events[i].SequenceID = sequenceID
 				}
 
-				err = b.CompleteWorkflowTask(ctx, task, wfi, backend.WorkflowStateFinished, events, []history.Event{}, []history.WorkflowEvent{})
+				err = b.CompleteWorkflowTask(ctx, task, wfi, backend.WorkflowStateFinished, events, []history.Event{}, []history.Event{}, []history.WorkflowEvent{})
 				require.NoError(t, err)
 
 				time.Sleep(time.Second)
@@ -278,7 +278,7 @@ func BackendTest(t *testing.T, setup func() TestBackend, teardown func(b TestBac
 				// Simulate context and sub-workflow cancellation
 				task, err := b.GetWorkflowTask(ctx)
 				require.NoError(t, err)
-				err = b.CompleteWorkflowTask(ctx, task, instance, backend.WorkflowStateActive, task.NewEvents, []history.Event{}, []history.WorkflowEvent{
+				err = b.CompleteWorkflowTask(ctx, task, instance, backend.WorkflowStateActive, task.NewEvents, []history.Event{}, []history.Event{}, []history.WorkflowEvent{
 					{
 						WorkflowInstance: subInstance1,
 						HistoryEvent: history.NewHistoryEvent(1, time.Now(), history.EventType_WorkflowExecutionCanceled, &history.SubWorkflowCancellationRequestedAttributes{
@@ -323,6 +323,6 @@ func startWorkflow(t *testing.T, ctx context.Context, b backend.Backend, c clien
 	task, err := b.GetWorkflowTask(ctx)
 	require.NoError(t, err)
 
-	err = b.CompleteWorkflowTask(ctx, task, instance, backend.WorkflowStateActive, task.NewEvents, []history.Event{}, []history.WorkflowEvent{})
+	err = b.CompleteWorkflowTask(ctx, task, instance, backend.WorkflowStateActive, task.NewEvents, []history.Event{}, []history.Event{}, []history.WorkflowEvent{})
 	require.NoError(t, err)
 }

backend/test/backendtest.go

@@ -91,7 +91,11 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {
 				a := func(context.Context) error { return nil }
 				wf := func(ctx workflow.Context) (int, error) {
-					return workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, a).Get(ctx)
+					return workflow.ExecuteActivity[int](ctx, workflow.ActivityOptions{
+						RetryOptions: workflow.RetryOptions{
+							MaxAttempts: 1,
+						},
+					}, a).Get(ctx)
 				}
 				register(t, ctx, w, []interface{}{wf}, nil)
 
@@ -106,7 +110,11 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {
 				a := func(context.Context, int, int) error { return nil }
 				wf := func(ctx workflow.Context) (int, error) {
-					return workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, a, 42).Get(ctx)
+					return workflow.ExecuteActivity[int](ctx, workflow.ActivityOptions{
+						RetryOptions: workflow.RetryOptions{
+							MaxAttempts: 1,
+						},
+					}, a, 42).Get(ctx)
 				}
 				register(t, ctx, w, []interface{}{wf}, []interface{}{a})
 

backend/test/e2e.go

@@ -51,7 +51,7 @@ func Test_Client_GetWorkflowResultSuccess(t *testing.T) {
 		mockClock.Add(time.Second)
 	})
 	b.On("GetWorkflowInstanceState", mock.Anything, instance).Return(backend.WorkflowStateFinished, nil)
-	b.On("GetWorkflowInstanceHistory", mock.Anything, instance).Return([]history.Event{
+	b.On("GetWorkflowInstanceHistory", mock.Anything, instance, (*int64)(nil)).Return([]history.Event{
 		history.NewHistoryEvent(1, time.Now(), history.EventType_WorkflowExecutionStarted, &history.ExecutionStartedAttributes{}),
 		history.NewHistoryEvent(2, time.Now(), history.EventType_WorkflowExecutionFinished, &history.ExecutionCompletedAttributes{
 			Result: r,

client/client_test.go

@@ -0,0 +1,54 @@
+package command
+
+import (
+	"github.com/benbjohnson/clock"
+	"github.com/cschleiden/go-workflows/internal/core"
+	"github.com/cschleiden/go-workflows/internal/history"
+)
+
+type CancelSubWorkflowCommand struct {
+	command
+
+	SubWorkflowInstance *core.WorkflowInstance
+}
+
+var _ Command = (*CancelSubWorkflowCommand)(nil)
+
+func NewCancelSubWorkflowCommand(id int64, subWorkflowInstance *core.WorkflowInstance) *CancelSubWorkflowCommand {
+	return &CancelSubWorkflowCommand{
+		command: command{
+			state: CommandState_Pending,
+			id:    id,
+		},
+		SubWorkflowInstance: subWorkflowInstance,
+	}
+}
+
+func (*CancelSubWorkflowCommand) Type() string {
+	return "CancelSubworkflow"
+}
+
+func (c *CancelSubWorkflowCommand) Commit(clock clock.Clock) *CommandResult {
+	c.commit()
+
+	return &CommandResult{
+		// Record that cancellation was requested
+		Events: []history.Event{
+			history.NewPendingEvent(
+				clock.Now(),
+				history.EventType_SubWorkflowCancellationRequested,
+				&history.SubWorkflowCancellationRequestedAttributes{
+					SubWorkflowInstance: c.SubWorkflowInstance,
+				},
+			),
+		},
+
+		// Send cancellation event to sub-workflow
+		WorkflowEvents: []history.WorkflowEvent{
+			{
+				WorkflowInstance: c.SubWorkflowInstance,
+				HistoryEvent:     history.NewWorkflowCancellationEvent(clock.Now()),
+			},
+		},
+	}
+}