Explicitly switch between timer modes

Author: cschleiden

tester/options.go

@@ -0,0 +1,28 @@
+package tester
+
+import (
+	"time"
+
+	"github.com/cschleiden/go-workflows/internal/converter"
+	"github.com/cschleiden/go-workflows/log"
+)
+
+type WorkflowTesterOption func(*options)
+
+func WithLogger(logger log.Logger) WorkflowTesterOption {
+	return func(o *options) {
+		o.Logger = logger
+	}
+}
+
+func WithConverter(converter converter.Converter) WorkflowTesterOption {
+	return func(o *options) {
+		o.Converter = converter
+	}
+}
+
+func WithTestTimeout(timeout time.Duration) WorkflowTesterOption {
+	return func(o *options) {
+		o.TestTimeout = timeout
+	}
+}

tester/tester.go

@@ -45,15 +45,29 @@ type testTimer struct {
 	// ScheduleEventID is the ID of the schedule event for this timer
 	ScheduleEventID int64
 
-	// At is the time this timer is scheduled for
+	// At is the time this timer is scheduled for in test time
 	At time.Time
 
+	// WallClockAt is the time this timer is scheduled for in wall-clock time
+	WallClockAt time.Time
+
 	// Callback is called when the timer should fire.
-	Callback func()
+	Callback *func()
+
+	TimerEvent *history.WorkflowEvent
 
 	wallClockTimer *clock.Timer
 }
 
+func (tt *testTimer) fire() *history.WorkflowEvent {
+	if tt.Callback != nil {
+		(*tt.Callback)()
+		return nil
+	}
+
+	return tt.TimerEvent
+}
+
 type testWorkflow struct {
 	instance      *core.WorkflowInstance
 	history       []*history.Event
@@ -93,11 +107,16 @@ type workflowTester[TResult any] struct {
 	workflowHistory []*history.Event
 	clock           *clock.Mock
 	wallClock       clock.Clock
-	startTime       time.Time
 
-	sync.Map
-	timers    []*testTimer
-	nextTimer *testTimer
+	// Wall-clock start time of the workflow test run
+	startTime time.Time
+
+	timers         []*testTimer
+	wallClockTimer *clock.Timer
+
+	// timerWallClockStart time.Time
+	timerMode timeMode
+
 	callbacks chan func() *history.WorkflowEvent
 
 	subWorkflowListener func(*core.WorkflowInstance, string)
@@ -111,26 +130,6 @@ type workflowTester[TResult any] struct {
 	converter converter.Converter
 }
 
-type WorkflowTesterOption func(*options)
-
-func WithLogger(logger log.Logger) WorkflowTesterOption {
-	return func(o *options) {
-		o.Logger = logger
-	}
-}
-
-func WithConverter(converter converter.Converter) WorkflowTesterOption {
-	return func(o *options) {
-		o.Converter = converter
-	}
-}
-
-func WithTestTimeout(timeout time.Duration) WorkflowTesterOption {
-	return func(o *options) {
-		o.TestTimeout = timeout
-	}
-}
-
 func NewWorkflowTester[TResult any](wf interface{}, opts ...WorkflowTesterOption) *workflowTester[TResult] {
 	if err := margs.ReturnTypeMatch[TResult](wf); err != nil {
 		panic(fmt.Sprintf("workflow return type does not match: %s", err))
@@ -177,6 +176,7 @@ func NewWorkflowTester[TResult any](wf interface{}, opts ...WorkflowTesterOption
 
 		timers:    make([]*testTimer, 0),
 		callbacks: make(chan func() *history.WorkflowEvent, 1024),
+		timerMode: TM_TimeTravel,
 
 		logger:    options.Logger.With("source", "tester"),
 		tracer:    tracer,
@@ -205,8 +205,9 @@ func (wt *workflowTester[TResult]) Registry() *workflow.Registry {
 
 func (wt *workflowTester[TResult]) ScheduleCallback(delay time.Duration, callback func()) {
 	wt.timers = append(wt.timers, &testTimer{
-		At:       wt.clock.Now().Add(delay),
-		Callback: callback,
+		At:         wt.clock.Now().Add(delay),
+		Callback:   &callback,
+		TimerEvent: nil,
 	})
 }
 
@@ -332,34 +333,8 @@ func (wt *workflowTester[TResult]) Execute(args ...interface{}) {
 			}
 
 			// No callbacks, try to fire any pending timers
-			if len(wt.timers) > 0 && wt.nextTimer == nil {
-				// Take first timer and execute it
-				t := wt.timers[0]
-				wt.timers = wt.timers[1:]
-
-				// If there are no running activities, we can time-travel to the next timer and execute it. Otherwise, if
-				// there are running activities, only fire the timer if it is due.
-				runningActivities := atomic.LoadInt32(&wt.runningActivities)
-				if runningActivities > 0 {
-					// Wall-clock mode
-					wt.logger.Debug("Scheduling wall-clock timer", "at", t.At)
-
-					wt.nextTimer = t
-
-					remainingTime := wt.clock.Until(t.At)
-					t.wallClockTimer = wt.wallClock.AfterFunc(remainingTime, func() {
-						t.Callback()
-						wt.nextTimer = nil
-					})
-				} else {
-					// Time-travel mode
-					wt.logger.Debug("Advancing workflow clock to fire timer", "to", t.At)
-
-					// Advance workflow clock and fire the timer
-					wt.clock.Set(t.At)
-					t.Callback()
-				}
-
+			if wt.fireTimer() {
+				// Timer fired
 				continue
 			}
 
@@ -382,6 +357,88 @@ func (wt *workflowTester[TResult]) Execute(args ...interface{}) {
 	}
 }
 
+func (wt *workflowTester[TResult]) fireTimer() bool {
+	if len(wt.timers) == 0 {
+		// No timers to fire
+		return false
+	}
+
+	// Determine mode we should be in and transition if it doesn't match the current one
+	newMode := wt.newTimerMode()
+	if wt.timerMode != newMode {
+		wt.logger.Debug("Transitioning timer mode", "from", wt.timerMode, "to", newMode)
+
+		// Transition timer mode
+		switch newMode {
+		case TM_TimeTravel:
+			if wt.wallClockTimer != nil {
+				wt.wallClockTimer.Stop()
+				wt.wallClockTimer = nil
+			}
+
+		case TM_WallClock:
+			// Going from time-travel to wall-clock mode. Nothing to do here.
+		}
+
+		wt.timerMode = newMode
+	}
+
+	switch wt.timerMode {
+	case TM_TimeTravel:
+		{
+			// Pop first timer and execute it
+			t := wt.timers[0]
+			wt.timers = wt.timers[1:]
+
+			wt.logger.Debug("Advancing workflow clock to fire timer", "to", t.At)
+
+			// Advance workflow clock and fire the timer
+			wt.clock.Set(t.At)
+			wt.callbacks <- t.fire
+			return true
+		}
+
+	case TM_WallClock:
+		{
+			t := wt.timers[0]
+
+			wt.logger.Debug("Scheduling wall-clock timer", "at", t.WallClockAt)
+
+			// wt.nextTimer = t
+
+			if wt.wallClock.Now().After(t.WallClockAt) {
+				// Fire timer
+				wt.timers = wt.timers[1:]
+				wt.callbacks <- t.fire
+
+				return true
+			} else if wt.wallClockTimer == nil {
+				// Schedule timer
+				wt.wallClockTimer = wt.wallClock.AfterFunc(t.WallClockAt.Sub(wt.wallClock.Now()), func() {
+					wt.callbacks <- func() *history.WorkflowEvent {
+						// Remove timer
+						wt.timers = wt.timers[1:]
+						wt.wallClockTimer = nil
+
+						return t.fire()
+					}
+				})
+			}
+		}
+	}
+
+	return false
+}
+
+func (wt *workflowTester[TResult]) newTimerMode() timeMode {
+	runningActivities := atomic.LoadInt32(&wt.runningActivities)
+	if runningActivities > 0 {
+		return TM_WallClock
+	}
+
+	return TM_TimeTravel
+}
+
 func (wt *workflowTester[TResult]) sendEvent(wfi *core.WorkflowInstance, event *history.Event) {
 	w := wt.getWorkflow(wfi)
 
@@ -550,13 +607,10 @@ func (wt *workflowTester[TResult]) scheduleTimer(instance *core.WorkflowInstance
 		Instance:        instance,
 		ScheduleEventID: event.ScheduleEventID,
 		At:              e.At,
-		Callback: func() {
-			wt.callbacks <- func() *history.WorkflowEvent {
-				return &history.WorkflowEvent{
-					WorkflowInstance: instance,
-					HistoryEvent:     event,
-				}
-			}
+		WallClockAt:     wt.wallClock.Now().Add(e.At.Sub(wt.clock.Now())),
+		TimerEvent: &history.WorkflowEvent{
+			WorkflowInstance: instance,
+			HistoryEvent:     event,
 		},
 	})
 

tester/tester_activity_test.go

@@ -60,6 +60,8 @@ func Test_LongActivity(t *testing.T) {
 
 func Test_ActivityRaceWithSignal(t *testing.T) {
 	activity1 := func() (string, error) {
+		time.Sleep(200 * time.Millisecond)
+
 		return "activity", nil
 	}
 
@@ -90,9 +92,7 @@ func Test_ActivityRaceWithSignal(t *testing.T) {
 
 	tester := NewWorkflowTester[string](wf, WithTestTimeout(time.Second*3))
 
-	tester.OnActivity(activity1).Run(func(args mock.Arguments) {
-		time.Sleep(200 * time.Millisecond)
-	}).Return("activity", nil)
+	tester.OnActivity(activity1).Return("activity", nil)
 
 	tester.ScheduleCallback(time.Millisecond*100, func() {
 		tester.SignalWorkflow("signal", "stop")

tester/tester_logger_test.go

@@ -0,0 +1,85 @@
+package tester
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/cschleiden/go-workflows/internal/logger"
+	"github.com/cschleiden/go-workflows/log"
+)
+
+type debugLogger struct {
+	defaultFields []interface{}
+	lines         *[]string
+	l             log.Logger
+}
+
+func newDebugLogger() *debugLogger {
+	lines := []string{}
+	return &debugLogger{
+		lines:         &lines,
+		defaultFields: []interface{}{},
+		l:             logger.NewDefaultLogger(),
+	}
+}
+
+func (dl *debugLogger) hasLine(msg string) bool {
+	for _, line := range *dl.lines {
+		if strings.Contains(line, msg) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (dl *debugLogger) formatFields(level, msg string, fields ...interface{}) string {
+	var result []string
+
+	result = append(result, fmt.Sprintf("|%s| %s", level, msg))
+
+	for i := 0; i < len(dl.defaultFields)/2; i++ {
+		result = append(result, fmt.Sprintf("%v=%v", dl.defaultFields[i*2], dl.defaultFields[i*2+1]))
+	}
+
+	for i := 0; i < len(fields)/2; i++ {
+		result = append(result, fmt.Sprintf("%v=%v", fields[i*2], fields[i*2+1]))
+	}
+
+	return strings.Join(result, " ")
+}
+
+func (dl *debugLogger) addLine(level, msg string, fields ...interface{}) {
+	// Persist for debugging
+	*dl.lines = append(*dl.lines, dl.formatFields(level, msg, fields...))
+}
+
+func (dl *debugLogger) Debug(msg string, fields ...interface{}) {
+	dl.addLine("DEBUG", msg, fields...)
+	dl.l.Debug(msg, fields...)
+}
+
+func (dl *debugLogger) Error(msg string, fields ...interface{}) {
+	dl.addLine("ERROR", msg, fields...)
+	dl.l.Error(msg, fields...)
+}
+
+func (dl *debugLogger) Panic(msg string, fields ...interface{}) {
+	dl.addLine("PANIC", msg, fields...)
+	dl.l.Panic(msg, fields...)
+}
+
+func (dl *debugLogger) Warn(msg string, fields ...interface{}) {
+	dl.addLine("WARN", msg, fields...)
+	dl.l.Warn(msg, fields...)
+}
+
+func (dl *debugLogger) With(fields ...interface{}) log.Logger {
+	return &debugLogger{
+		lines:         dl.lines, // Keep this here
+		defaultFields: append(dl.defaultFields, fields...),
+		l:             dl.l.With(fields...),
+	}
+}
+
+var _ log.Logger = (*debugLogger)(nil)