Merge pull request #303 from cschleiden/pending-future-output

Track pending futures with a name

Author: cschleiden

internal/workflow/executor.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log/slog"
 	"reflect"
+	"slices"
 
 	"github.com/benbjohnson/clock"
 	"github.com/cschleiden/go-workflows/backend"
@@ -264,8 +265,15 @@ func (e *executor) executeNewEvents(newEvents []*history.Event) ([]*history.Even
 	if e.workflow.Completed() {
 		// TODO: Is this too early? We haven't committed some of the commands
 		if e.workflowState.HasPendingFutures() {
-			e.logger.Error("workflow completed, but there are still pending futures")
-			panic("workflow completed, but there are still pending futures")
+			var pending []string
+			pf := e.workflowState.PendingFutureNames()
+			for id, name := range pf {
+				pending = append(pending, fmt.Sprintf("%d-%s", id, name))
+			}
+			slices.Sort(pending)
+
+			e.logger.Error("workflow completed, but there are still pending futures", "pending", pending)
+			panic(fmt.Sprintf("workflow completed, but there are still pending futures: %s", pending))
 		}
 
 		if canErr, ok := e.workflow.Error().(*continueasnew.Error); ok {
@@ -411,7 +419,7 @@ func (e *executor) handleActivityCompleted(event *history.Event, a *history.Acti
 		return fmt.Errorf("could not find pending future for activity completion")
 	}
 
-	err := f(a.Result, nil)
+	err := f.Set(a.Result, nil)
 	if err != nil {
 		return fmt.Errorf("setting activity completed result: %w", err)
 	}
@@ -440,7 +448,7 @@ func (e *executor) handleActivityFailed(event *history.Event, a *history.Activit
 	}
 
 	actErr := workflowerrors.ToError(a.Error)
-	if err := f(nil, actErr); err != nil {
+	if err := f.Set(nil, actErr); err != nil {
 		return fmt.Errorf("setting activity failed result: %w", err)
 	}
 
@@ -483,7 +491,7 @@ func (e *executor) handleTimerFired(event *history.Event, a *history.TimerFiredA
 		return nil
 	}
 
-	if err := f(nil, nil); err != nil {
+	if err := f.Set(nil, nil); err != nil {
 		return fmt.Errorf("setting timer fired result: %w", err)
 	}
 
@@ -523,7 +531,7 @@ func (e *executor) handleTimerCanceled(event *history.Event, a *history.TimerCan
 		return nil
 	}
 
-	if err := f(nil, sync.Canceled); err != nil {
+	if err := f.Set(nil, sync.Canceled); err != nil {
 		return fmt.Errorf("setting timer canceled result: %w", err)
 	}
 
@@ -580,7 +588,7 @@ func (e *executor) handleSubWorkflowFailed(event *history.Event, a *history.SubW
 
 	wfErr := workflowerrors.ToError(a.Error)
 
-	if err := f(nil, wfErr); err != nil {
+	if err := f.Set(nil, wfErr); err != nil {
 		return fmt.Errorf("setting sub workflow failed result: %w", err)
 	}
 
@@ -606,7 +614,7 @@ func (e *executor) handleSubWorkflowCompleted(event *history.Event, a *history.S
 		return errors.New("no pending future found for sub workflow completed event")
 	}
 
-	if err := f(a.Result, nil); err != nil {
+	if err := f.Set(a.Result, nil); err != nil {
 		return fmt.Errorf("setting sub workflow completed result: %w", err)
 	}
 
@@ -651,7 +659,7 @@ func (e *executor) handleSideEffectResult(event *history.Event, a *history.SideE
 		return errors.New("no pending future found for side effect result event")
 	}
 
-	if err := f(a.Result, nil); err != nil {
+	if err := f.Set(a.Result, nil); err != nil {
 		return fmt.Errorf("setting side effect result result: %w", err)
 	}
 

internal/workflow/executor_test.go

@@ -580,6 +580,35 @@ func Test_Executor(t *testing.T) {
 				require.True(t, e.workflow.Completed())
 			},
 		},
+		{
+			name: "Pending futures result in panic",
+			f: func(t *testing.T, r *registry.Registry, e *executor, i *core.WorkflowInstance, hp *testHistoryProvider) {
+				subworkflow := func(ctx wf.Context) error {
+					return nil
+				}
+
+				workflow := func(ctx wf.Context) error {
+					// Start but not wait for sub-workflow
+					wf.CreateSubWorkflowInstance[any](ctx, wf.SubWorkflowOptions{
+						InstanceID: "subworkflow",
+					}, subworkflow)
+
+					// Schedule but not wait for timer
+					wf.ScheduleTimer(ctx, time.Second*2)
+
+					return nil
+				}
+
+				r.RegisterWorkflow(workflow)
+				r.RegisterWorkflow(subworkflow)
+
+				task := startWorkflowTask("instanceID", workflow)
+
+				require.PanicsWithValue(t, "workflow completed, but there are still pending futures: [1-subworkflow:1 2-timer:2s]", func() {
+					e.ExecuteTask(context.Background(), task)
+				})
+			},
+		},
 	}
 
 	for _, tt := range tests {

internal/workflowstate/workflowstate.go

@@ -18,26 +18,32 @@ type key int
 
 var workflowCtxKey key
 
-type DecodingSettable func(v payload.Payload, err error) error
+type DecodingSettable struct {
+	Set  func(v payload.Payload, err error) error
+	Name string
+}
 
 // Use this to track futures for the workflow state. It's required to map the generic Future interface
 // to a type without type parameters.
-func AsDecodingSettable[T any](cv converter.Converter, f sync.SettableFuture[T]) DecodingSettable {
-	return func(v payload.Payload, err error) error {
-		if f.HasValue() {
-			return fmt.Errorf("future already has value")
-		}
+func AsDecodingSettable[T any](cv converter.Converter, name string, f sync.SettableFuture[T]) *DecodingSettable {
+	return &DecodingSettable{
+		Name: name,
+		Set: func(v payload.Payload, err error) error {
+			if f.HasValue() {
+				return fmt.Errorf("future already has value")
+			}
 
-		var t T
-		if v != nil {
-			if err := cv.From(v, &t); err != nil {
-				return fmt.Errorf("failed to decode future: %v", err)
+			var t T
+			if v != nil {
+				if err := cv.From(v, &t); err != nil {
+					return fmt.Errorf("failed to decode future: %v", err)
+				}
 			}
-		}
 
-		f.Set(t, err)
+			f.Set(t, err)
 
-		return nil
+			return nil
+		},
 	}
 }
 
@@ -50,7 +56,7 @@ type WfState struct {
 	instance        *core.WorkflowInstance
 	scheduleEventID int64
 	commands        []command.Command
-	pendingFutures  map[int64]DecodingSettable
+	pendingFutures  map[int64]*DecodingSettable
 	replaying       bool
 
 	pendingSignals map[string][]payload.Payload
@@ -67,7 +73,7 @@ func NewWorkflowState(instance *core.WorkflowInstance, logger *slog.Logger, cloc
 		instance:        instance,
 		commands:        []command.Command{},
 		scheduleEventID: 1,
-		pendingFutures:  map[int64]DecodingSettable{},
+		pendingFutures:  map[int64]*DecodingSettable{},
 
 		pendingSignals: map[string][]payload.Payload{},
 		signalChannels: make(map[string]*signalChannel),
@@ -96,15 +102,24 @@ func (wf *WfState) GetNextScheduleEventID() int64 {
 	return scheduleEventID
 }
 
-func (wf *WfState) TrackFuture(scheduleEventID int64, f DecodingSettable) {
+func (wf *WfState) TrackFuture(scheduleEventID int64, f *DecodingSettable) {
 	wf.pendingFutures[scheduleEventID] = f
 }
 
+func (wf *WfState) PendingFutureNames() map[int64]string {
+	result := map[int64]string{}
+	for id, f := range wf.pendingFutures {
+		result[id] = f.Name
+	}
+
+	return result
+}
+
 func (wf *WfState) HasPendingFutures() bool {
 	return len(wf.pendingFutures) > 0
 }
 
-func (wf *WfState) FutureByScheduleEventID(scheduleEventID int64) (DecodingSettable, bool) {
+func (wf *WfState) FutureByScheduleEventID(scheduleEventID int64) (*DecodingSettable, bool) {
 	f, ok := wf.pendingFutures[scheduleEventID]
 	return f, ok
 }

internal/workflowstate/workflowstate_test.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/benbjohnson/clock"
 	"github.com/cschleiden/go-workflows/backend/converter"
-	"github.com/cschleiden/go-workflows/backend/payload"
 	"github.com/cschleiden/go-workflows/core"
 	"github.com/cschleiden/go-workflows/internal/sync"
 	"github.com/google/uuid"
@@ -21,12 +20,7 @@ func Test_PendingFutures(t *testing.T) {
 	require.False(t, wfState.HasPendingFutures())
 
 	f := sync.NewFuture[int]()
-	wfState.TrackFuture(1, func(v payload.Payload, err error) error {
-		var r int
-		require.NoError(t, converter.DefaultConverter.From(v, &r))
-		f.Set(r, err)
-		return nil
-	})
+	wfState.TrackFuture(1, AsDecodingSettable[int](converter.DefaultConverter, "f", f))
 
 	require.True(t, wfState.HasPendingFutures())
 

workflow/activity.go

@@ -73,7 +73,7 @@ func executeActivity[TResult any](ctx Context, options ActivityOptions, attempt
 
 	cmd := command.NewScheduleActivityCommand(scheduleEventID, name, inputs, metadata)
 	wfState.AddCommand(cmd)
-	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(cv, f))
+	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(cv, fmt.Sprintf("activity: %s", name), f))
 
 	ctx, span := workflowtracer.Tracer(ctx).Start(ctx,
 		fmt.Sprintf("ExecuteActivity: %s", name),

workflow/sideeffect.go

@@ -28,7 +28,7 @@ func SideEffect[TResult any](ctx Context, f func(ctx Context) TResult) Future[TR
 	scheduleEventID := wfState.GetNextScheduleEventID()
 
 	cv := contextvalue.Converter(ctx)
-	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(cv, future))
+	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(cv, "sideeffect", future))
 
 	cmd := command.NewSideEffectCommand(scheduleEventID)
 	wfState.AddCommand(cmd)

workflow/subworkflow.go

@@ -98,7 +98,7 @@ func createSubWorkflowInstance[TResult any](ctx Context, options SubWorkflowOpti
 	cmd := command.NewScheduleSubWorkflowCommand(scheduleEventID, wfState.Instance(), options.InstanceID, workflowName, inputs, metadata)
 
 	wfState.AddCommand(cmd)
-	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(cv, f))
+	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(cv, fmt.Sprintf("subworkflow:%s", workflowName), f))
 
 	// Check if the channel is cancelable
 	if c, cancelable := ctx.Done().(sync.CancelChannel); cancelable {

workflow/timer.go

@@ -1,6 +1,7 @@
 package workflow
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/cschleiden/go-workflows/internal/command"
@@ -30,7 +31,7 @@ func ScheduleTimer(ctx Context, delay time.Duration) Future[any] {
 
 	timerCmd := command.NewScheduleTimerCommand(scheduleEventID, at)
 	wfState.AddCommand(timerCmd)
-	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(contextvalue.Converter(ctx), f))
+	wfState.TrackFuture(scheduleEventID, workflowstate.AsDecodingSettable(contextvalue.Converter(ctx), fmt.Sprintf("timer:%v", delay), f))
 
 	cancelReceiver := &sync.Receiver[struct{}]{
 		Receive: func(v struct{}, ok bool) {

workflow/timer_test.go

@@ -42,7 +42,7 @@ func Test_Timer_Cancellation(t *testing.T) {
 	cmd.Done()
 	fs, ok := state.FutureByScheduleEventID(1)
 	require.True(t, ok)
-	fs(nil, nil)
+	fs.Set(nil, nil)
 
 	c.Execute()
 	require.False(t, c.Finished())