Merge pull request #131 from cschleiden/panic-when-completing-pending-workflow

Panic when completing workflow with pending futures

Author: cschleiden

internal/workflow/executor.go

@@ -218,6 +218,11 @@ func (e *executor) executeNewEvents(newEvents []history.Event) ([]history.Event,
 	}
 
 	if e.workflow.Completed() {
+		// TODO: Is this too early? We haven't committed some of the commands
+		if e.workflowState.HasPendingFutures() {
+			panic("Workflow completed, but there are still pending futures")
+		}
+
 		e.workflowCompleted(e.workflow.Result(), e.workflow.Error())
 	}
 
@@ -356,6 +361,8 @@ func (e *executor) handleActivityCompleted(event history.Event, a *history.Activ
 		return fmt.Errorf("setting activity completed result: %w", err)
 	}
 
+	e.workflowState.RemoveFuture(event.ScheduleEventID)
+
 	c := e.workflowState.CommandByScheduleEventID(event.ScheduleEventID)
 	if c == nil {
 		return fmt.Errorf("previous workflow execution scheduled an activity which could not be found")
@@ -381,6 +388,8 @@ func (e *executor) handleActivityFailed(event history.Event, a *history.Activity
 		return fmt.Errorf("setting activity failed result: %w", err)
 	}
 
+	e.workflowState.RemoveFuture(event.ScheduleEventID)
+
 	c := e.workflowState.CommandByScheduleEventID(event.ScheduleEventID)
 	if c == nil {
 		return fmt.Errorf("previous workflow execution scheduled an activity which could not be found")
@@ -422,6 +431,8 @@ func (e *executor) handleTimerFired(event history.Event, a *history.TimerFiredAt
 		return fmt.Errorf("setting timer fired result: %w", err)
 	}
 
+	e.workflowState.RemoveFuture(event.ScheduleEventID)
+
 	c := e.workflowState.CommandByScheduleEventID(event.ScheduleEventID)
 	if c == nil {
 		return fmt.Errorf("no command found for timer fired event")
@@ -460,6 +471,8 @@ func (e *executor) handleTimerCanceled(event history.Event, a *history.TimerCanc
 		return fmt.Errorf("setting timer canceled result: %w", err)
 	}
 
+	e.workflowState.RemoveFuture(event.ScheduleEventID)
+
 	return e.workflow.Continue()
 }
 
@@ -513,6 +526,8 @@ func (e *executor) handleSubWorkflowFailed(event history.Event, a *history.SubWo
 		return fmt.Errorf("setting sub workflow failed result: %w", err)
 	}
 
+	e.workflowState.RemoveFuture(event.ScheduleEventID)
+
 	c := e.workflowState.CommandByScheduleEventID(event.ScheduleEventID)
 	if c == nil {
 		// TODO: Adjust
@@ -539,6 +554,8 @@ func (e *executor) handleSubWorkflowCompleted(event history.Event, a *history.Su
 		return fmt.Errorf("setting sub workflow completed result: %w", err)
 	}
 
+	e.workflowState.RemoveFuture(event.ScheduleEventID)
+
 	c := e.workflowState.CommandByScheduleEventID(event.ScheduleEventID)
 	if c == nil {
 		// TODO: Adjust
@@ -600,6 +617,8 @@ func (e *executor) handleSideEffectResult(event history.Event, a *history.SideEf
 		return fmt.Errorf("setting side effect result result: %w", err)
 	}
 
+	e.workflowState.RemoveFuture(event.ScheduleEventID)
+
 	return e.workflow.Continue()
 }
 

internal/workflow/executor_test.go

@@ -529,14 +529,16 @@ func Test_Executor(t *testing.T) {
 				workflow := func(ctx wf.Context) error {
 					swctx, cancel := wf.WithCancel(ctx)
 
-					wf.CreateSubWorkflowInstance[any](swctx, wf.SubWorkflowOptions{
+					f := wf.CreateSubWorkflowInstance[any](swctx, wf.SubWorkflowOptions{
 						InstanceID: "subworkflow",
 					}, subworkflow)
 
 					wf.Sleep(ctx, time.Millisecond)
 
 					cancel()
 
+					f.Get(ctx)
+
 					return nil
 				}
 
@@ -559,6 +561,7 @@ func Test_Executor(t *testing.T) {
 				result, err = e.ExecuteTask(context.Background(), continueTask("instanceID", []history.Event{
 					result.TimerEvents[0],
 				}, result.Executed[len(result.Executed)-1].SequenceID))
+
 				require.NoError(t, err)
 				require.Len(t, result.WorkflowEvents, 1, "Cancellation should have been requested")
 				require.Equal(t, history.EventType_WorkflowExecutionCanceled, result.WorkflowEvents[0].HistoryEvent.Type)
@@ -567,8 +570,19 @@ func Test_Executor(t *testing.T) {
 					subWorkflowInstance,
 					result.WorkflowEvents[0].WorkflowInstance)
 
+				require.Len(t, e.workflowState.Commands(), 2)
+
+				// Complete subworkflow
+				swr, _ := converter.DefaultConverter.To(nil)
+				hp.history = append(hp.history, result.Executed...)
+				result, err = e.ExecuteTask(context.Background(), continueTask("instanceID", []history.Event{
+					history.NewPendingEvent(time.Now(), history.EventType_SubWorkflowCompleted, &history.SubWorkflowCompletedAttributes{
+						Result: swr,
+					}, history.ScheduleEventID(1)),
+				}, result.Executed[len(result.Executed)-1].SequenceID))
+
+				require.NoError(t, err)
 				require.True(t, e.workflow.Completed())
-				require.Len(t, e.workflowState.Commands(), 3)
 			},
 		},
 	}

internal/workflowstate/workflowstate.go

@@ -100,6 +100,10 @@ func (wf *WfState) TrackFuture(scheduleEventID int64, f DecodingSettable) {
 	wf.pendingFutures[scheduleEventID] = f
 }
 
+func (wf *WfState) HasPendingFutures() bool {
+	return len(wf.pendingFutures) > 0
+}
+
 func (wf *WfState) FutureByScheduleEventID(scheduleEventID int64) (DecodingSettable, bool) {
 	f, ok := wf.pendingFutures[scheduleEventID]
 	return f, ok

internal/workflowstate/workflowstate_test.go

@@ -0,0 +1,36 @@
+package workflowstate
+
+import (
+	"testing"
+
+	"github.com/benbjohnson/clock"
+	"github.com/cschleiden/go-workflows/internal/converter"
+	"github.com/cschleiden/go-workflows/internal/core"
+	"github.com/cschleiden/go-workflows/internal/logger"
+	"github.com/cschleiden/go-workflows/internal/payload"
+	"github.com/cschleiden/go-workflows/internal/sync"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_PendingFutures(t *testing.T) {
+	i := core.NewWorkflowInstance(uuid.NewString(), "")
+
+	wfState := NewWorkflowState(i, logger.NewDefaultLogger(), clock.New())
+
+	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, nil)
+		return nil
+	})
+
+	require.True(t, wfState.HasPendingFutures())
+
+	wfState.RemoveFuture(1)
+
+	require.False(t, wfState.HasPendingFutures())
+}

samples/timer/timer.go

@@ -68,8 +68,7 @@ func Workflow1(ctx workflow.Context, msg string) (string, error) {
 
 	a1 := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1, 35, 12)
 
-	tctx, _ := workflow.WithCancel(ctx)
-	// cancel()
+	tctx, cancel := workflow.WithCancel(ctx)
 
 	workflow.Select(
 		ctx,
@@ -89,7 +88,7 @@ func Workflow1(ctx workflow.Context, msg string) (string, error) {
 			logger.Debug("Activity result", r)
 
 			// Cancel timer
-			// cancel()
+			cancel()
 		}),
 	)
 

workflow/sideeffect.go

@@ -38,6 +38,7 @@ func SideEffect[TResult any](ctx Context, f func(ctx Context) TResult) Future[TR
 
 		cmd.SetResult(payload)
 		future.Set(r, nil)
+		wfState.RemoveFuture(scheduleEventID)
 	}
 
 	return future