Merge pull request #185 from cschleiden/cschleiden/test-timers

Explicitly switch between timer modes in test

Author: cschleiden

tester/options.go

@@ -0,0 +1,34 @@
+package tester
+
+import (
+	"time"
+
+	"github.com/cschleiden/go-workflows/internal/converter"
+	"github.com/cschleiden/go-workflows/log"
+)
+
+type options struct {
+	TestTimeout time.Duration
+	Logger      log.Logger
+	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
+	}
+}

tester/tester.go

@@ -38,6 +38,39 @@ func (t *testHistoryProvider) GetWorkflowInstanceHistory(ctx context.Context, in
 	return t.history, nil
 }
 
+type testTimer struct {
+	// Instance is the workflow instance this timer is for
+	Instance *core.WorkflowInstance
+
+	// ScheduleEventID is the ID of the schedule event for this timer
+	ScheduleEventID int64
+
+	// At is the time this timer is scheduled for in test time
+	At time.Time
+
+	// Callback is called when the timer should fire.
+	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
+	pendingEvents []*history.Event
+}
+
 type WorkflowTester[TResult any] interface {
 	// Now returns the current time of the simulated clock in the tester
 	Now() time.Time
@@ -68,34 +101,6 @@ type WorkflowTester[TResult any] interface {
 	ListenSubWorkflow(listener func(instance *core.WorkflowInstance, name string))
 }
 
-type testTimer struct {
-	// Instance is the workflow instance this timer is for
-	Instance *core.WorkflowInstance
-
-	// ScheduleEventID is the ID of the schedule event for this timer
-	ScheduleEventID int64
-
-	// At is the time this timer is scheduled for
-	At time.Time
-
-	// Callback is called when the timer should fire.
-	Callback func()
-
-	wallClockTimer *clock.Timer
-}
-
-type testWorkflow struct {
-	instance      *core.WorkflowInstance
-	history       []*history.Event
-	pendingEvents []*history.Event
-}
-
-type options struct {
-	TestTimeout time.Duration
-	Logger      log.Logger
-	Converter   converter.Converter
-}
-
 type workflowTester[TResult any] struct {
 	options *options
 
@@ -123,11 +128,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)
@@ -141,27 +151,9 @@ 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
-	}
-}
+var _ WorkflowTester[any] = (*workflowTester[any])(nil)
 
-func WithTestTimeout(timeout time.Duration) WorkflowTesterOption {
-	return func(o *options) {
-		o.TestTimeout = timeout
-	}
-}
-
-func NewWorkflowTester[TResult any](wf interface{}, opts ...WorkflowTesterOption) WorkflowTester[TResult] {
+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))
 	}
@@ -207,6 +199,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,
@@ -235,8 +228,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,
 	})
 }
 
@@ -362,34 +356,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
 			}
 
@@ -412,6 +380,86 @@ 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:
+		{
+			if wt.wallClockTimer != nil {
+				// Wall-clock timer already scheduled
+				return false
+			}
+
+			t := wt.timers[0]
+
+			wt.logger.Debug("Scheduling wall-clock timer", "at", t.At)
+
+			// Determine when this should run
+			remainingTime := t.At.Sub(wt.clock.Now())
+
+			// Schedule timer
+			wt.wallClockTimer = wt.wallClock.AfterFunc(remainingTime, 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)
 
@@ -580,13 +628,9 @@ 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,
-				}
-			}
+		TimerEvent: &history.WorkflowEvent{
+			WorkflowInstance: instance,
+			HistoryEvent:     event,
 		},
 	})