diff --git a/internal/workflow/config.go b/internal/workflow/config.go index 90b0b9b..e485bb1 100644 --- a/internal/workflow/config.go +++ b/internal/workflow/config.go @@ -10,6 +10,20 @@ import ( "go.temporal.io/sdk/workflow" ) +// terminationContext returns a context safe for running terminal bookkeeping +// activities (status updates, termination events). When the workflow has been +// cancelled, the supplied context is already cancelled and any activity started +// on it fails immediately -- which would leave the instance/stage rows stuck +// "running" and skip the termination event. In that case a disconnected context +// is returned so the bookkeeping still runs. +func terminationContext(ctx workflow.Context) workflow.Context { + if ctx.Err() == nil { + return ctx + } + disconnected, _ := workflow.NewDisconnectedContext(ctx) + return disconnected +} + type RawStage map[string]map[string]any type Config struct { @@ -85,16 +99,20 @@ func (c *Config) run(ctx workflow.Context, instance Instance, variables map[stri } stage.SetTerminated(runError, workflow.Now(ctx).Round(time.Nanosecond)) - err = workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ + // Record the stage termination on a context that survives cancellation, + // otherwise a cancelled stage would never be marked terminated. + cleanupCtx := terminationContext(ctx) + + err = workflow.ExecuteActivity(workflow.WithActivityOptions(cleanupCtx, workflow.ActivityOptions{ StartToCloseTimeout: 10 * time.Second, - }), UpdateStageActivity, stage).Get(ctx, nil) + }), UpdateStageActivity, stage).Get(cleanupCtx, nil) if err != nil { return err } - err = workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ + err = workflow.ExecuteActivity(workflow.WithActivityOptions(cleanupCtx, workflow.ActivityOptions{ StartToCloseTimeout: 10 * time.Second, - }), SendWorkflowStageTerminationEventActivity, instance, stage).Get(ctx, nil) + }), SendWorkflowStageTerminationEventActivity, instance, stage).Get(cleanupCtx, nil) if err != nil { return err } diff --git a/internal/workflow/run.go b/internal/workflow/run.go index 267da96..1bc0262 100644 --- a/internal/workflow/run.go +++ b/internal/workflow/run.go @@ -74,16 +74,21 @@ func (w Workflows) Run(ctx workflow.Context, i Input, instance Instance) error { instance.SetTerminated(workflow.Now(ctx)) } - err = workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ + // Record the instance termination on a context that survives cancellation, + // otherwise a cancelled run would never be marked terminated and no + // termination event would be published. + cleanupCtx := terminationContext(ctx) + + err = workflow.ExecuteActivity(workflow.WithActivityOptions(cleanupCtx, workflow.ActivityOptions{ StartToCloseTimeout: 10 * time.Second, - }), UpdateInstanceActivity, instance).Get(ctx, nil) + }), UpdateInstanceActivity, instance).Get(cleanupCtx, nil) if err != nil { return err } - err = workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ + err = workflow.ExecuteActivity(workflow.WithActivityOptions(cleanupCtx, workflow.ActivityOptions{ StartToCloseTimeout: 10 * time.Second, - }), SendWorkflowTerminationEventActivity, instance).Get(ctx, nil) + }), SendWorkflowTerminationEventActivity, instance).Get(cleanupCtx, nil) if err != nil { return err }