update: Allow save and repeat for standard steps (#154)

* update: Allow save and repeat for standard steps

* add testing for testingRunStateController impl

* make skip types unexported

* panic if user uses internally reserved node values

Author: andrewwormald

builder.go

@@ -30,7 +30,7 @@ func NewBuilder[Type any, Status StatusType](name string) *Builder[Type, Status]
 			consumers:     make(map[Status]consumerConfig[Type, Status]),
 			callback:      make(map[Status][]callback[Type, Status]),
 			timeouts:      make(map[Status]timeouts[Type, Status]),
-			statusGraph:   graph.New(),
+			statusGraph:   graph.New(allSkipTypes()...),
 			errorCounter:  errorcounter.New(),
 			internalState: make(map[string]State),
 			logger: &logger{

builder_internal_test.go

@@ -561,3 +561,21 @@ func TestOnComplete(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, testErr, actualFn(nil, nil))
 }
+
+func TestPanicForUsingReservedNodeValues(t *testing.T) {
+	b := NewBuilder[string, testStatus]("")
+
+	for _, skipType := range allSkipTypes() {
+		require.PanicsWithValue(t,
+			"cannot use reserved node as 'from' node",
+			func() {
+				b.AddStep(testStatus(skipType), nil, statusEnd)
+			}, "Using reserved nodes should panic")
+
+		require.PanicsWithValue(t,
+			"cannot use reserved node as 'to' node",
+			func() {
+				b.AddStep(statusStart, nil, testStatus(skipType))
+			}, "Using reserved nodes should panic")
+	}
+}

callback_internal_test.go

@@ -136,7 +136,7 @@ func TestProcessCallback(t *testing.T) {
 
 		callbackFn := CallbackFunc[string, testStatus](func(ctx context.Context, r *Run[string, testStatus], reader io.Reader) (testStatus, error) {
 			calls["callbackFunc"] += 1
-			return testStatus(SkipTypeDefault), nil
+			return testStatus(skipTypeDefault), nil
 		})
 
 		updater := func(ctx context.Context, current testStatus, next testStatus, record *Run[string, testStatus], workingVersion uint) error {

internal/graph/graph.go

@@ -1,13 +1,21 @@
 package graph
 
-import "slices"
+import (
+	"slices"
+)
+
+func New(reservedNodes ...int) *Graph {
+	rs := make(map[int]bool)
+	for _, node := range reservedNodes {
+		rs[node] = true
+	}
 
-func New() *Graph {
 	return &Graph{
-		graph:      make(map[int][]int),
-		starting:   make(map[int]bool),
-		terminal:   make(map[int]bool),
-		validNodes: make(map[int]bool),
+		graph:         make(map[int][]int),
+		starting:      make(map[int]bool),
+		terminal:      make(map[int]bool),
+		validNodes:    make(map[int]bool),
+		reservedNodes: rs,
 	}
 }
 
@@ -17,9 +25,20 @@ type Graph struct {
 	starting   map[int]bool
 	terminal   map[int]bool
 	validNodes map[int]bool
+
+	// reservedNode is for values that are reserved for the user and cannot be used as nodes in the graph.
+	reservedNodes map[int]bool
 }
 
 func (g *Graph) AddTransition(from int, to int) {
+	if _, reserved := g.reservedNodes[from]; reserved {
+		panic("cannot use reserved node as 'from' node")
+	}
+
+	if _, reserved := g.reservedNodes[to]; reserved {
+		panic("cannot use reserved node as 'to' node")
+	}
+
 	if _, ok := g.validNodes[from]; !ok {
 		g.nodeOrder = append(g.nodeOrder, from)
 	}

internal/graph/graph_test.go

@@ -58,3 +58,36 @@ func TestGraph(t *testing.T) {
 	expectedNodes := []int{1, 2, 3, 4, 5}
 	require.Equal(t, expectedNodes, actualNodes)
 }
+
+func TestGraphReservedNodes(t *testing.T) {
+	g := graph.New(-1, 99)
+	// Ensure no panic on adding transitions that are not reserved nodes
+	g.AddTransition(1, 2)
+	require.True(t, g.IsTerminal(2))
+	require.True(t, g.IsValid(1))
+	require.True(t, g.IsValid(2))
+
+	require.PanicsWithValue(t,
+		"cannot use reserved node as 'from' node",
+		func() {
+			g.AddTransition(-1, 1)
+		}, "Using reserved nodes should panic")
+
+	require.PanicsWithValue(t,
+		"cannot use reserved node as 'to' node",
+		func() {
+			g.AddTransition(1, -1)
+		}, "Using reserved nodes should panic")
+
+	require.PanicsWithValue(t,
+		"cannot use reserved node as 'from' node",
+		func() {
+			g.AddTransition(99, 1)
+		}, "Using reserved nodes should panic")
+
+	require.PanicsWithValue(t,
+		"cannot use reserved node as 'to' node",
+		func() {
+			g.AddTransition(1, 99)
+		}, "Using reserved nodes should panic")
+}

run.go

@@ -24,12 +24,12 @@ func (r *Run[Type, Status]) Pause(ctx context.Context, reason string) (Status, e
 		return 0, err
 	}
 
-	return Status(SkipTypeRunStateUpdate), nil
+	return Status(skipTypeRunStateUpdate), nil
 }
 
 // Skip is a util function to skip the update and move on to the next event (consumer) or execution (callback)
 func (r *Run[Type, Status]) Skip() (Status, error) {
-	return Status(SkipTypeDefault), nil
+	return Status(skipTypeDefault), nil
 }
 
 // Cancel is intended to be used inside a workflow process where (Status, error) are the return signature. This allows
@@ -41,7 +41,11 @@ func (r *Run[Type, Status]) Cancel(ctx context.Context, reason string) (Status,
 		return 0, err
 	}
 
-	return Status(SkipTypeRunStateUpdate), nil
+	return Status(skipTypeRunStateUpdate), nil
+}
+
+func (r *Run[Type, Status]) SaveAndRepeat() (Status, error) {
+	return Status(skipTypeSaveAndRepeat), nil
 }
 
 func buildRun[Type any, Status StatusType](store storeFunc, wr *Record) (*Run[Type, Status], error) {

run_test.go renamed to run_internal_test.go

@@ -1,31 +1,29 @@
-package workflow_test
+package workflow
 
 import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
-
-	"github.com/luno/workflow"
 )
 
 func TestNewTestingRun(t *testing.T) {
-	r := workflow.NewTestingRun[string, status](t, workflow.Record{}, "test")
+	r := NewTestingRun[string, testStatus](t, Record{}, "test")
 	ctx := t.Context()
 
 	pauseStatus, err := r.Pause(ctx, "")
 	require.NoError(t, err)
-	require.Equal(t, status(workflow.SkipTypeRunStateUpdate), pauseStatus)
+	require.Equal(t, testStatus(skipTypeRunStateUpdate), pauseStatus)
 
 	cancelStatus, err := r.Cancel(ctx, "")
 	require.NoError(t, err)
-	require.Equal(t, status(workflow.SkipTypeRunStateUpdate), cancelStatus)
+	require.Equal(t, testStatus(skipTypeRunStateUpdate), cancelStatus)
 }
 
 func TestNewTestingRun_requiresTestingParam(t *testing.T) {
 	require.PanicsWithValue(t,
 		"Cannot use NewTestingRun without *testing.T parameter",
 		func() {
-			_ = workflow.NewTestingRun[string, status](nil, workflow.Record{}, "test")
+			_ = NewTestingRun[string, testStatus](nil, Record{}, "test")
 		},
 	)
 }

runstate.go

@@ -68,7 +68,7 @@ func (rs RunState) Stopped() bool {
 	}
 }
 
-// RunStateController allows the interaction with a specific workflow record.
+// RunStateController allows the interaction with a specific workflow run.
 type RunStateController interface {
 	// Pause will take the workflow run specified and move it into a temporary state where it will no longer be processed.
 	// A paused workflow run can be resumed by calling Resume. ErrUnableToPause is returned when a workflow is not in a

runstate_internal_test.go

@@ -2,26 +2,117 @@ package workflow
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 )
 
-func TestNoopRunStateController(t *testing.T) {
-	ctrl := testingRunStateController{}
+func TestTestingRunStateController(t *testing.T) {
+	ctx := context.Background()
 
-	ctx := t.Context()
-	err := ctrl.Pause(ctx, "")
-	require.NoError(t, err)
+	t.Run("methods return nil error when no functions set", func(t *testing.T) {
+		ctrl := testingRunStateController{}
 
-	err = ctrl.Resume(ctx)
-	require.NoError(t, err)
+		err := ctrl.Pause(ctx, "test reason")
+		require.NoError(t, err)
 
-	err = ctrl.Cancel(ctx, "")
-	require.NoError(t, err)
+		err = ctrl.Resume(ctx)
+		require.NoError(t, err)
 
-	err = ctrl.DeleteData(ctx, "")
-	require.NoError(t, err)
+		err = ctrl.Cancel(ctx, "test reason")
+		require.NoError(t, err)
+
+		err = ctrl.DeleteData(ctx, "test reason")
+		require.NoError(t, err)
+
+		err = ctrl.SaveAndRepeat(ctx)
+		require.NoError(t, err)
+	})
+
+	t.Run("methods call provided functions", func(t *testing.T) {
+		var pauseCalled, resumeCalled, cancelCalled, deleteDataCalled, saveAndRepeatCalled bool
+
+		ctrl := testingRunStateController{
+			pause: func(ctx context.Context) error {
+				pauseCalled = true
+				return nil
+			},
+			resume: func(ctx context.Context) error {
+				resumeCalled = true
+				return nil
+			},
+			cancel: func(ctx context.Context) error {
+				cancelCalled = true
+				return nil
+			},
+			deleteData: func(ctx context.Context) error {
+				deleteDataCalled = true
+				return nil
+			},
+			saveAndRepeat: func(ctx context.Context) error {
+				saveAndRepeatCalled = true
+				return nil
+			},
+		}
+
+		err := ctrl.Pause(ctx, "test reason")
+		require.NoError(t, err)
+		require.True(t, pauseCalled)
+
+		err = ctrl.Resume(ctx)
+		require.NoError(t, err)
+		require.True(t, resumeCalled)
+
+		err = ctrl.Cancel(ctx, "test reason")
+		require.NoError(t, err)
+		require.True(t, cancelCalled)
+
+		err = ctrl.DeleteData(ctx, "test reason")
+		require.NoError(t, err)
+		require.True(t, deleteDataCalled)
+
+		err = ctrl.SaveAndRepeat(ctx)
+		require.NoError(t, err)
+		require.True(t, saveAndRepeatCalled)
+	})
+
+	t.Run("methods propagate errors", func(t *testing.T) {
+		testErr := errors.New("test error")
+
+		ctrl := testingRunStateController{
+			pause: func(ctx context.Context) error {
+				return testErr
+			},
+			resume: func(ctx context.Context) error {
+				return testErr
+			},
+			cancel: func(ctx context.Context) error {
+				return testErr
+			},
+			deleteData: func(ctx context.Context) error {
+				return testErr
+			},
+			saveAndRepeat: func(ctx context.Context) error {
+				return testErr
+			},
+		}
+
+		err := ctrl.Pause(ctx, "test reason")
+		require.Equal(t, testErr, err)
+
+		err = ctrl.Resume(ctx)
+		require.Equal(t, testErr, err)
+
+		err = ctrl.Cancel(ctx, "test reason")
+		require.Equal(t, testErr, err)
+
+		err = ctrl.DeleteData(ctx, "test reason")
+		require.Equal(t, testErr, err)
+
+		err = ctrl.SaveAndRepeat(ctx)
+		require.Equal(t, testErr, err)
+	})
 }
 
 func TestRunStateControllerTransitions(t *testing.T) {

status.go

@@ -7,28 +7,43 @@ type StatusType interface {
 }
 
 func skipUpdate[Status StatusType](status Status) bool {
-	_, ok := skipConfig[SkipType(status)]
-	return ok
+	return skipType(status) == skipTypeDefault || skipType(status) == skipTypeRunStateUpdate
+}
+
+func isSaveAndRepeat[Status StatusType](status Status) bool {
+	return skipType(status) == skipTypeSaveAndRepeat
 }
 
 func skipUpdateDescription[Status StatusType](status Status) string {
-	description, ok := skipConfig[SkipType(status)]
+	description, ok := skipConfig[skipType(status)]
 	if !ok {
 		return "Unknown skip reason '" + status.String() + "'"
 	}
 
 	return description
 }
 
-type SkipType int
+type skipType int
 
 var (
-	SkipTypeDefault        SkipType = 0
-	SkipTypeRunStateUpdate SkipType = -1
+	skipTypeDefault        skipType = 0
+	skipTypeRunStateUpdate skipType = -1
+	skipTypeSaveAndRepeat  skipType = -2
+	skipTypeSentinel       skipType = -3
 )
 
+func allSkipTypes() []int {
+	var all []int
+	for i := skipTypeDefault; i > skipTypeSentinel; i-- {
+		all = append(all, int(i))
+	}
+
+	return all
+}
+
 // skipConfig holds the skip values and descriptions as documentation as to what they mean.
-var skipConfig = map[SkipType]string{
-	SkipTypeDefault:        "Zero status with nil error value should result in a skip",
-	SkipTypeRunStateUpdate: "Internal run state update taken place. Skip normal newUpdater",
+var skipConfig = map[skipType]string{
+	skipTypeDefault:        "Zero status with nil error value should result in a skip",
+	skipTypeRunStateUpdate: "Internal run state update taken place. Skip normal newUpdater",
+	skipTypeSaveAndRepeat:  "Save and repeat operation: record saved and re-queued for processing",
 }