Merge pull request #60 from cschleiden/improve-dx

cschleiden · web-flow · commit 9af048daf7d4 · 2022-05-02T08:33:47.000-07:00
Improve error messages when workflow replays yield different commands
diff --git a/internal/command/command.go b/internal/command/command.go
@@ -13,7 +13,7 @@ type CommandType int
 const (
 	_ CommandType = iota
 
-	CommandType_ScheduleActivityTask
+	CommandType_ScheduleActivity
 
 	CommandType_ScheduleSubWorkflow
 	CommandType_CancelSubWorkflow
@@ -26,6 +26,27 @@ const (
 	CommandType_CompleteWorkflow
 )
 
+func (ct CommandType) String() string {
+	switch ct {
+	case CommandType_ScheduleActivity:
+		return "ScheduleActivityTask"
+	case CommandType_ScheduleSubWorkflow:
+		return "ScheduleSubWorkflow"
+	case CommandType_CancelSubWorkflow:
+		return "CancelSubWorkflow"
+	case CommandType_ScheduleTimer:
+		return "ScheduleTimer"
+	case CommandType_CancelTimer:
+		return "CancelTimer"
+	case CommandType_SideEffect:
+		return "SideEffect"
+	case CommandType_CompleteWorkflow:
+		return "CompleteWorkflow"
+	}
+
+	return ""
+}
+
 type CommandState int
 
 const (
@@ -52,7 +73,7 @@ type ScheduleActivityTaskCommandAttr struct {
 func NewScheduleActivityTaskCommand(id int64, name string, inputs []payload.Payload) Command {
 	return Command{
 		ID:   id,
-		Type: CommandType_ScheduleActivityTask,
+		Type: CommandType_ScheduleActivity,
 		Attr: &ScheduleActivityTaskCommandAttr{
 			Name:   name,
 			Inputs: inputs,
diff --git a/internal/sync/coroutine.go b/internal/sync/coroutine.go
@@ -85,7 +85,7 @@ func newState() *coState {
 	return &coState{
 		blocking: make(chan bool, 1),
 		unblock:  make(chan bool),
-		// Mostly used while debugging issues, default to discarding log messages
+		// Only 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,
diff --git a/internal/sync/future.go b/internal/sync/future.go
@@ -12,7 +12,7 @@ type Future[T any] interface {
 type SettableFuture[T any] interface {
 	Future[T]
 
-	// Set stores the value and unblocks any waiting consumers
+	// Set stores the value
 	Set(v T, err error) error
 }
 
diff --git a/internal/workflow/executor.go b/internal/workflow/executor.go
@@ -262,12 +262,20 @@ func (e *executor) handleWorkflowTaskStarted(event history.Event, a *history.Wor
 
 func (e *executor) handleActivityScheduled(event history.Event, a *history.ActivityScheduledAttributes) error {
 	c := e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
-	if c != nil {
-		// Ensure the same activity was scheduled again
-		ca := c.Attr.(*command.ScheduleActivityTaskCommandAttr)
-		if a.Name != ca.Name {
-			return fmt.Errorf("previous workflow execution scheduled different type of activity: %s, %s", a.Name, ca.Name)
-		}
+
+	// Ensure activity
+	if c == nil {
+		return fmt.Errorf("previous workflow execution scheduled an activity which could not be found")
+	}
+
+	if c.Type != command.CommandType_ScheduleActivity {
+		return fmt.Errorf("previous workflow execution scheduled an activity, not: %v", c.Type)
+	}
+
+	// Ensure the same activity was scheduled again
+	ca := c.Attr.(*command.ScheduleActivityTaskCommandAttr)
+	if a.Name != ca.Name {
+		return fmt.Errorf("previous workflow execution scheduled different type of activity: %s, %s", a.Name, ca.Name)
 	}
 
 	return nil
@@ -276,10 +284,9 @@ func (e *executor) handleActivityScheduled(event history.Event, a *history.Activ
 func (e *executor) handleActivityCompleted(event history.Event, a *history.ActivityCompletedAttributes) error {
 	f, ok := e.workflowState.FutureByScheduleEventID(event.ScheduleEventID)
 	if !ok {
-		return nil
+		return fmt.Errorf("could not find pending future for activity completion")
 	}
 
-	e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
 	err := f(a.Result, nil)
 	if err != nil {
 		return fmt.Errorf("setting result: %w", err)
@@ -291,11 +298,9 @@ func (e *executor) handleActivityCompleted(event history.Event, a *history.Activ
 func (e *executor) handleActivityFailed(event history.Event, a *history.ActivityFailedAttributes) error {
 	f, ok := e.workflowState.FutureByScheduleEventID(event.ScheduleEventID)
 	if !ok {
-		return errors.New("no pending future found for activity failed event")
+		return errors.New("no pending future for activity failed event")
 	}
 
-	e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
-
 	if err := f(nil, errors.New(a.Reason)); err != nil {
 		return fmt.Errorf("setting result: %w", err)
 	}
@@ -304,7 +309,14 @@ func (e *executor) handleActivityFailed(event history.Event, a *history.Activity
 }
 
 func (e *executor) handleTimerScheduled(event history.Event, a *history.TimerScheduledAttributes) error {
-	e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
+	c := e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
+	if c == nil {
+		return fmt.Errorf("previous workflow execution scheduled a timer")
+	}
+
+	if c.Type != command.CommandType_ScheduleTimer {
+		return fmt.Errorf("previous workflow execution scheduled a timer, not: %v", c.Type)
+	}
 
 	return nil
 }
@@ -316,8 +328,6 @@ func (e *executor) handleTimerFired(event history.Event, a *history.TimerFiredAt
 		return nil
 	}
 
-	e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
-
 	if err := f(nil, nil); err != nil {
 		return fmt.Errorf("setting result: %w", err)
 	}
@@ -332,8 +342,6 @@ func (e *executor) handleTimerCanceled(event history.Event, a *history.TimerCanc
 		return nil
 	}
 
-	e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
-
 	if err := f(nil, nil); err != nil {
 		return fmt.Errorf("setting result: %w", err)
 	}
@@ -343,27 +351,35 @@ func (e *executor) handleTimerCanceled(event history.Event, a *history.TimerCanc
 
 func (e *executor) handleSubWorkflowScheduled(event history.Event, a *history.SubWorkflowScheduledAttributes) error {
 	c := e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
-	if c != nil {
-		ca := c.Attr.(*command.ScheduleSubWorkflowCommandAttr)
-		if a.Name != ca.Name {
-			return errors.New("previous workflow execution scheduled a different sub workflow")
-		}
+	if c == nil {
+		return fmt.Errorf("previous workflow execution scheduled a sub workflow")
+	}
+
+	if c.Type != command.CommandType_ScheduleSubWorkflow {
+		return fmt.Errorf("previous workflow execution scheduled a sub workflow, not: %v", c.Type)
+	}
 
-		// Set correct InstanceID here.
-		// TODO: see if we can provide better support for commands here and find a better place to store
-		// this message.
-		ca.Instance = a.SubWorkflowInstance
+	ca := c.Attr.(*command.ScheduleSubWorkflowCommandAttr)
+	if a.Name != ca.Name {
+		return fmt.Errorf("previous workflow execution scheduled different type of sub workflow: %s, %s", a.Name, ca.Name)
 	}
 
-	// TOOD: If the command cannot be found, raise error?
+	// Set correct InstanceID here.
+	// TODO: see if we can provide better support for commands here and find a better place to store
+	// this message.
+	ca.Instance = a.SubWorkflowInstance
 
 	return nil
 }
 
 func (e *executor) handleSubWorkflowCancellationRequest(event history.Event, a *history.SubWorkflowCancellationRequestedAttributes) error {
 	c := e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
-	if c != nil {
-		return fmt.Errorf("expected to find a sub workflow cancellation event, instead got: %v", event.Type)
+	if c == nil {
+		return fmt.Errorf("previous workflow execution cancelled a sub-workflow execution")
+	}
+
+	if c.Type != command.CommandType_CancelSubWorkflow {
+		return fmt.Errorf("previous workflow execution cancelled a sub-workflow execution, not: %v", c.Type)
 	}
 
 	return e.workflow.Continue(e.workflowCtx)
@@ -375,8 +391,6 @@ func (e *executor) handleSubWorkflowFailed(event history.Event, a *history.SubWo
 		return errors.New("no pending future found for sub workflow failed event")
 	}
 
-	e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
-
 	if err := f(nil, errors.New(a.Error)); err != nil {
 		return fmt.Errorf("setting result: %w", err)
 	}
@@ -390,8 +404,6 @@ func (e *executor) handleSubWorkflowCompleted(event history.Event, a *history.Su
 		return errors.New("no pending future found for sub workflow completed event")
 	}
 
-	e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
-
 	if err := f(a.Result, nil); err != nil {
 		return fmt.Errorf("setting result: %w", err)
 	}
@@ -403,8 +415,6 @@ func (e *executor) handleSignalReceived(event history.Event, a *history.SignalRe
 	// Send signal to workflow channel
 	workflowstate.ReceiveSignal(e.workflowCtx, e.workflowState, a.Name, a.Arg)
 
-	e.workflowState.RemoveCommandByEventID(event.ScheduleEventID)
-
 	return e.workflow.Continue(e.workflowCtx)
 }
 
@@ -441,7 +451,7 @@ func (e *executor) processCommands(ctx context.Context, t *task.Workflow) (bool,
 		c.State = command.CommandState_Committed
 
 		switch c.Type {
-		case command.CommandType_ScheduleActivityTask:
+		case command.CommandType_ScheduleActivity:
 			a := c.Attr.(*command.ScheduleActivityTaskCommandAttr)
 
 			scheduleActivityEvent := e.createNewEvent(
diff --git a/internal/workflow/executor_test.go b/internal/workflow/executor_test.go
@@ -200,7 +200,7 @@ func Test_ExecuteWorkflowWithActivityCommand(t *testing.T) {
 	require.Equal(t, command.Command{
 		ID:    1,
 		State: command.CommandState_Committed,
-		Type:  command.CommandType_ScheduleActivityTask,
+		Type:  command.CommandType_ScheduleActivity,
 		Attr: &command.ScheduleActivityTaskCommandAttr{
 			Name:   "activity1",
 			Inputs: []payload.Payload{inputs},
@@ -309,7 +309,7 @@ func Test_ExecuteWorkflowWithSelector(t *testing.T) {
 	require.Len(t, e.workflowState.Commands(), 2)
 
 	require.Equal(t, command.CommandType_ScheduleTimer, e.workflowState.Commands()[0].Type)
-	require.Equal(t, command.CommandType_ScheduleActivityTask, e.workflowState.Commands()[1].Type)
+	require.Equal(t, command.CommandType_ScheduleActivity, e.workflowState.Commands()[1].Type)
 }
 
 func Test_ExecuteNewEvents(t *testing.T) {

Original file line number	Diff line number	Diff line change
`@@ -12,7 +12,7 @@ type Future[T any] interface {`
`12`	`12`	`type SettableFuture[T any] interface {`
`13`	`13`	`Future[T]`
`14`	`14`
`15`		`- // Set stores the value and unblocks any waiting consumers`
	`15`	`+ // Set stores the value`
`16`	`16`	`Set(v T, err error) error`
`17`	`17`	`}`
`18`	`18`