update: Allow save and repeat for standard steps

Author: andrewwormald

record.go

@@ -1,6 +1,9 @@
 package workflow
 
-import "time"
+import (
+	"strconv"
+	"time"
+)
 
 // Record is the cornerstone of Workflow. Record must always be wire compatible with no generics as it's intended
 // purpose is to be the persisted data structure of a Run.
@@ -74,6 +77,32 @@ type Meta struct {
 	// It provides a line trace or path that helps track where the execution was initiated,
 	// offering context for debugging or auditing purposes by capturing the origin of the workflow trigger.
 	TraceOrigin string
+
+	// UpdateType indicates the type of update that is performed relating to the state being updated in the RecordStore
+	// and whether it results in a new event being created. UpdateType is useful for understanding the nature of
+	// the change made to the record and the intended consequences of that change.
+	UpdateType UpdateType
+}
+
+type UpdateType int
+
+const (
+	// UpdateTypeDefault indicates a standard update to the record that results in the state being updated and a new
+	// corresponding event is created.
+	UpdateTypeDefault UpdateType = 0
+	// UpdateTypeStateOnly indicates an update that only modifies the RunState of the record without creating a new event.
+	UpdateTypeStateOnly = 1
+)
+
+func (u UpdateType) String() string {
+	switch u {
+	case UpdateTypeDefault:
+		return "Default"
+	case UpdateTypeStateOnly:
+		return "State Only"
+	default:
+		return "UpdateType(" + strconv.FormatInt(int64(u), 10) + ")"
+	}
 }
 
 // TypedRecord differs from Record in that it contains a Typed Object and Typed Status

run.go

@@ -44,6 +44,10 @@ func (r *Run[Type, Status]) Cancel(ctx context.Context, reason string) (Status,
 	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) {
 	var t Type
 	err := Unmarshal(wr.Object, &t)

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

status.go

@@ -11,6 +11,10 @@ func skipUpdate[Status StatusType](status Status) bool {
 	return ok
 }
 
+func isSaveAndRepeat[Status StatusType](status Status) bool {
+	return SkipType(status) == SkipTypeSaveAndRepeat
+}
+
 func skipUpdateDescription[Status StatusType](status Status) string {
 	description, ok := skipConfig[SkipType(status)]
 	if !ok {
@@ -25,6 +29,7 @@ type SkipType int
 var (
 	SkipTypeDefault        SkipType = 0
 	SkipTypeRunStateUpdate SkipType = -1
+	SkipTypeSaveAndRepeat  SkipType = -2
 )
 
 // skipConfig holds the skip values and descriptions as documentation as to what they mean.

testing.go

@@ -28,9 +28,10 @@ func TriggerCallbackOn[Type any, Status StatusType, Payload any](
 		panic("*workflow.Workflow required for testing utility functions")
 	}
 
-	_ = waitFor(t, w, foreignID, func(r *Record) (bool, error) {
+	_, err := waitFor(t, w, foreignID, func(r *Record) (bool, error) {
 		return r.Status == int(waitForStatus), nil
 	})
+	require.NoError(t, err)
 
 	b, err := json.Marshal(p)
 	require.NoError(t, err)
@@ -109,12 +110,13 @@ func Require[Type any, Status StatusType](
 		return
 	}
 
-	wr := waitFor(t, w, foreignID, func(r *Record) (bool, error) {
+	wr, err := waitFor(t, w, foreignID, func(r *Record) (bool, error) {
 		return r.Status == int(waitForStatus), nil
 	})
+	require.NoError(t, err)
 
 	var actual Type
-	err := Unmarshal(wr.Object, &actual)
+	err = Unmarshal(wr.Object, &actual)
 	require.NoError(t, err)
 
 	// Due to nuances in encoding libraries such as json with the ability to implement custom
@@ -147,27 +149,28 @@ func WaitFor[Type any, Status StatusType](
 		panic("*workflow.Workflow required for testing utility functions")
 	}
 
-	waitFor(t, w, foreignID, func(r *Record) (bool, error) {
+	_, err := waitFor(t, w, foreignID, func(r *Record) (bool, error) {
 		run, err := buildRun[Type, Status](w.recordStore.Store, r)
 		require.NoError(t, err)
 
 		return fn(run)
 	})
+	require.NoError(t, err)
 }
 
 func waitFor[Type any, Status StatusType](
 	t testing.TB,
 	w *Workflow[Type, Status],
 	foreignID string,
 	fn func(r *Record) (bool, error),
-) *Record {
+) (*Record, error) {
 	testingStore, ok := w.recordStore.(TestingRecordStore)
 	if !ok {
 		panic("TestingRecordStore implementation for record store dependency required")
 	}
 
 	var runID string
-	for runID == "" {
+	for runID == "" && w.ctx.Err() == nil {
 		latest, err := w.recordStore.Latest(context.Background(), w.Name(), foreignID)
 		if errors.Is(err, ErrRecordNotFound) {
 			continue
@@ -182,7 +185,7 @@ func waitFor[Type any, Status StatusType](
 	// testingStore.SetSnapshotOffset(w.name, foreignID, runID, 0)
 
 	var wr Record
-	for wr.RunID == "" {
+	for wr.RunID == "" && w.ctx.Err() == nil {
 		snapshots := testingStore.Snapshots(w.Name(), foreignID, runID)
 		for _, r := range snapshots {
 			ok, err := fn(r)
@@ -194,7 +197,7 @@ func waitFor[Type any, Status StatusType](
 		}
 	}
 
-	return &wr
+	return &wr, w.ctx.Err()
 }
 
 // NewTestingRun should be used when testing logic that defines a workflow.Run as a parameter. This is usually the
@@ -255,10 +258,11 @@ func WithDeleteDataFn(deleteData func(ctx context.Context) error) TestingRunOpti
 }
 
 type testingRunStateController struct {
-	pause      func(ctx context.Context) error
-	cancel     func(ctx context.Context) error
-	resume     func(ctx context.Context) error
-	deleteData func(ctx context.Context) error
+	pause         func(ctx context.Context) error
+	cancel        func(ctx context.Context) error
+	resume        func(ctx context.Context) error
+	deleteData    func(ctx context.Context) error
+	saveAndRepeat func(ctx context.Context) error
 }
 
 func (c *testingRunStateController) Pause(ctx context.Context, reason string) error {
@@ -293,4 +297,12 @@ func (c *testingRunStateController) DeleteData(ctx context.Context, reason strin
 	return c.deleteData(ctx)
 }
 
+func (c *testingRunStateController) SaveAndRepeat(ctx context.Context) error {
+	if c.saveAndRepeat == nil {
+		return nil
+	}
+
+	return c.saveAndRepeat(ctx)
+}
+
 var _ RunStateController = (*testingRunStateController)(nil)

update.go

@@ -58,6 +58,12 @@ func newUpdater[Type any, Status StatusType](
 			return nil
 		}
 
+		// Save and repeat skips transition validation and keeps the current status.
+		if isSaveAndRepeat(next) {
+			updatedRecord.Status = int(current)
+			return updateRecord(ctx, store, updatedRecord, record.RunState, current.String())
+		}
+
 		err = validateTransition(current, next, graph)
 		if err != nil {
 			return err

workflow.go

@@ -272,6 +272,9 @@ func runOnce(
 	if errors.Is(err, context.Canceled) {
 		// Exit cleanly if error returned is cancellation of context
 		return err
+	} else if errors.Is(err, context.DeadlineExceeded) {
+		// Exit cleanly if context's deadline was exceeded
+		return err
 	} else if err != nil {
 		logger.Error(ctx, fmt.Errorf("run error [role=%s], [process=%s]: %v", role, processName, err))
 
@@ -287,6 +290,9 @@ func runOnce(
 		// Context can be cancelled by the role scheduler and thus return nil to attempt to gain the role again
 		// and if the parent context was cancelled then that will exit safely.
 		return nil
+	} else if errors.Is(err, context.DeadlineExceeded) {
+		// Exit cleanly if context's deadline was exceeded
+		return err
 	} else if err != nil {
 		logger.Error(ctx, fmt.Errorf("run error [role=%s], [process=%s]: %v", role, processName, err))
 		metrics.ProcessErrors.WithLabelValues(workflowName, processName).Inc()

workflow_test.go

@@ -231,9 +231,7 @@ func BenchmarkWorkflow(b *testing.B) {
 
 func benchmarkWorkflow(b *testing.B, numberOfSteps int) {
 	ctx, cancel := context.WithCancel(context.Background())
-	b.Cleanup(func() {
-		cancel()
-	})
+	b.Cleanup(cancel)
 
 	bldr := workflow.NewBuilder[MyType, status]("benchmark")
 
@@ -729,3 +727,61 @@ func TestExpectedProcesses_outboxDisabled(t *testing.T) {
 		require.Truef(t, expected[process], "process '%s' is missing expected value", process)
 	}
 }
+
+func TestSaveAndRepeat(t *testing.T) {
+	iterations := 100
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	type Custom struct {
+		Count      int
+		Collection []string
+	}
+
+	bldr := workflow.NewBuilder[Custom, status]("save and repeat")
+	bldr.AddStep(StatusStart, func(ctx context.Context, r *workflow.Run[Custom, status]) (status, error) {
+		r.Object.Count++
+		r.Object.Collection = append(r.Object.Collection, fmt.Sprintf("item-%d", r.Object.Count))
+
+		if r.Object.Count < iterations {
+			return r.SaveAndRepeat()
+		}
+
+		return StatusEnd, nil
+	}, StatusEnd)
+
+	recordStore := memrecordstore.New()
+	clock := clock_testing.NewFakeClock(time.Now())
+	wf := bldr.Build(
+		memstreamer.New(),
+		recordStore,
+		memrolescheduler.New(),
+		workflow.WithClock(clock),
+		workflow.WithDefaultOptions(
+			workflow.PollingFrequency(time.Nanosecond),
+		),
+		workflow.WithOutboxOptions(
+			workflow.OutboxPollingFrequency(time.Nanosecond),
+		),
+		workflow.WithDebugMode(),
+	)
+
+	wf.Run(ctx)
+	t.Cleanup(wf.Stop)
+
+	fid := "custom-save-and-repeat"
+	_, err := wf.Trigger(ctx, fid)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var expected []string
+	for i := range iterations {
+		expected = append(expected, fmt.Sprintf("item-%d", i+1))
+	}
+
+	workflow.Require(t, wf, fid, StatusEnd, Custom{
+		Count:      iterations,
+		Collection: expected,
+	})
+}