Merge pull request #128 from cschleiden/cancellation-replay

Improve command state machine

Author: cschleiden

backend/test/e2e.go

@@ -392,6 +392,49 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 				require.Len(t, futureEvents, 0, "no future events should be scheduled")
 			},
 		},
+		{
+			name: "Timer_CancelTwice",
+			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {
+				a := func(ctx context.Context) error {
+					return nil
+				}
+				wf := func(ctx workflow.Context) error {
+					tctx, cancel := workflow.WithCancel(ctx)
+					f := workflow.ScheduleTimer(tctx, time.Second*10)
+
+					// Force the checkpoint before continuing the execution
+					workflow.ExecuteActivity[any](ctx, workflow.DefaultActivityOptions, a).Get(ctx)
+
+					// Cancel timer
+					cancel()
+
+					// Force another checkpoint
+					workflow.ExecuteActivity[any](ctx, workflow.DefaultActivityOptions, a).Get(ctx)
+
+					cancel()
+
+					// Force another checkpoint
+					workflow.ExecuteActivity[any](ctx, workflow.DefaultActivityOptions, a).Get(ctx)
+
+					if _, err := f.Get(ctx); err != nil && err != workflow.Canceled {
+						return err
+					}
+
+					return nil
+				}
+				register(t, ctx, w, []interface{}{wf}, []interface{}{a})
+
+				instance := runWorkflow(t, ctx, c, wf)
+				_, err := client.GetWorkflowResult[any](ctx, c, instance, time.Second*5)
+				require.NoError(t, err)
+
+				historyContains(ctx, t, b, instance, history.EventType_TimerScheduled, history.EventType_TimerCanceled)
+
+				futureEvents, err := b.GetFutureEvents(ctx)
+				require.NoError(t, err)
+				require.Len(t, futureEvents, 0, "no future events should be scheduled")
+			},
+		},
 		{
 			name: "Timer_CancelBeforeFiringRemovesFutureEvent",
 			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {
@@ -408,6 +451,9 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 					// Cancel timer
 					cancel()
 
+					// Force another checkpoint
+					workflow.ExecuteActivity[any](ctx, workflow.DefaultActivityOptions, a).Get(ctx)
+
 					if _, err := f.Get(ctx); err != nil && err != workflow.Canceled {
 						return err
 					}

docs/commands.drawio.png

@@ -0,0 +1,29 @@
+# Commands
+
+Commands encode decisions during the workflow execution. When a workflow like
+
+```go
+func Workflow(ctx workflow.Context) error {
+  _, err := workflow.ExecuteActivity(ctx, "Activity1").Get(ctx, nil)
+
+  return err
+}
+```
+
+is executed for the first time, a `ScheduleActivity` command is tracked and when the workflow reaches a checkpoint (it yields), this command is `Execute`d, and resulting `EventType_ActivityScheduled` event is processed by the backend. When the workflow is replayed, the command is also generated, but once the workflow reaches a checkpoint and history events are replayed, it is marked as `Committed` since we already have the `EventType_ActivityScheduled` message in the history.
+
+This will cause the command to **not** produce another event when the workflow yields again. Other operations are implemented in a similar way.
+
+
+## States & Transitions
+
+Commands are a simpel state machine and their states and transitions are the following:
+
+![](./commands.drawio.png)
+
+- Not all commands are cancelable. The ones which are are:
+  - ScheduleSubWorkflow
+  - ScheduleTimer
+
+- No all commands are committable, the ones which aren't are:
+  - Sideeffect

docs/commands.md

@@ -0,0 +1,44 @@
+package command
+
+type CancelableCommand interface {
+	Command
+
+	// Cancel cancels the command.
+	Cancel()
+
+	// HandleCancel handles a cancel event during replay
+	HandleCancel()
+}
+
+type cancelableCommand struct {
+	command
+}
+
+func (c *cancelableCommand) Cancel() {
+	switch c.state {
+	case CommandState_Pending, CommandState_Canceled:
+		c.state = CommandState_Canceled
+	case CommandState_Committed:
+		c.state = CommandState_CancelPending
+	default:
+		c.invalidStateTransition(CommandState_Canceled)
+	}
+}
+
+func (c *cancelableCommand) HandleCancel() {
+	switch c.state {
+	case CommandState_CancelPending:
+		c.state = CommandState_Canceled
+	default:
+		c.invalidStateTransition(CommandState_Canceled)
+	}
+}
+
+func (c *cancelableCommand) Done() {
+	switch c.state {
+	case CommandState_Committed, CommandState_Canceled:
+		c.state = CommandState_Done
+	default:
+		c.invalidStateTransition(CommandState_Done)
+	}
+}

internal/command/cancel_subworkflow.go

@@ -1,39 +1,25 @@
 package command
 
 import (
+	"fmt"
+
 	"github.com/benbjohnson/clock"
 	"github.com/cschleiden/go-workflows/internal/history"
 )
 
-type CommandState int
-
-//        ┌───────┐
-// ┌──────┤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
-	CommandState_Done
-)
-
 type Command interface {
 	ID() int64
 
 	Type() string
 
+	// State returns the current state of the command.
 	State() CommandState
 
-	Committed() bool
+	// Execute processes the command in its current state and moves it to the next state.
+	Execute(clock.Clock) *CommandResult
 
-	Commit(clock clock.Clock) *CommandResult
+	// Commit marks the command as committed without executing it.
+	Commit()
 
 	// Done marks the command as done. This transitions the state to done and indicates that the result
 	// of this command has been applied.
@@ -49,32 +35,43 @@ type CommandResult struct {
 }
 
 type command struct {
+	name string
+
 	state CommandState
 
 	id int64
 }
 
-func (c *command) commit() {
-	if c.state != CommandState_Pending {
-		panic("command already committed")
-	}
-
-	c.state = CommandState_Committed
-}
-
 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) Type() string {
+	return c.name
+}
+
+func (c *command) Commit() {
+	switch c.state {
+	case CommandState_Pending:
+		c.state = CommandState_Committed
+	default:
+		c.invalidStateTransition(CommandState_Committed)
+	}
+}
+
 func (c *command) Done() {
-	c.state = CommandState_Done
+	switch c.state {
+	case CommandState_Committed:
+		c.state = CommandState_Done
+	default:
+		c.invalidStateTransition(CommandState_Done)
+	}
+}
+
+func (c *command) invalidStateTransition(state CommandState) {
+	panic(fmt.Errorf("invalid state transition for command %s: %s -> %s", c.name, c.State().String(), state.String()))
 }

internal/command/cancel_timer.go

@@ -0,0 +1,26 @@
+package command
+
+import (
+	"testing"
+
+	"github.com/benbjohnson/clock"
+	"github.com/cschleiden/go-workflows/internal/history"
+	"github.com/stretchr/testify/require"
+)
+
+func assertExecuteNoEvent(t *testing.T, c Command, expectedState CommandState) {
+	r := c.Execute(clock.New())
+
+	require.Nil(t, r)
+}
+
+func assertExecuteWithEvent(t *testing.T, c Command, expectedState CommandState, expectedEventType history.EventType) *CommandResult {
+	r := c.Execute(clock.New())
+
+	require.NotNil(t, r)
+	require.Equal(t, expectedState, c.State())
+	require.Len(t, r.Events, 1)
+	require.Equal(t, expectedEventType, r.Events[0].Type)
+
+	return r
+}