@@ -668,6 +668,16 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
668668 _ = events .EmitExecutionStartedEvent (ctx , loggerLabels , triggerEvent .ID , executionID )
669669 e .metrics .With ("workflowID" , e .cfg .WorkflowID , "workflowName" , e .cfg .WorkflowName .String ()).IncrementWorkflowExecutionStartedCounter (ctx )
670670
671+ // Track execution error for deferred event emission
672+ var execErr error
673+ defer func () {
674+ _ = events .EmitExecutionFinishedEvent (ctx , loggerLabels , executionStatus , executionID , execErr , lggr )
675+ e .cfg .Hooks .OnExecutionFinished (executionID , executionStatus )
676+ if execErr != nil {
677+ e .cfg .Hooks .OnExecutionError (execErr .Error ())
678+ }
679+ }()
680+
671681 var timeProvider TimeProvider = & types.LocalTimeProvider {}
672682 if ! e .cfg .UseLocalTimeProvider {
673683 timeProvider = NewDonTimeProvider (e .cfg .DonTimeStore , e .cfg .WorkflowID , lggr )
@@ -676,18 +686,23 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
676686 moduleExecuteMaxResponseSizeBytes , err := e .cfg .LocalLimiters .ExecutionResponse .Limit (ctx )
677687 if err != nil {
678688 lggr .Errorw ("Failed to get execution response size limit" , "err" , err )
689+ executionStatus = store .StatusErrored
690+ execErr = err
679691 return
680692 }
681693 if moduleExecuteMaxResponseSizeBytes < 0 {
682- lggr .Errorf ("invalid moduleExecuteMaxResponseSizeBytes; must not be negative: %d" , moduleExecuteMaxResponseSizeBytes )
694+ execErr = fmt .Errorf ("invalid moduleExecuteMaxResponseSizeBytes; must not be negative: %d" , moduleExecuteMaxResponseSizeBytes )
695+ lggr .Errorw (execErr .Error ())
696+ executionStatus = store .StatusErrored
683697 return
684698 }
685699 execHelper := & ExecutionHelper {
686700 Engine : e , WorkflowExecutionID : executionID , UserLogChan : userLogChan ,
687701 TimeProvider : timeProvider , SecretsFetcher : e .secretsFetcher (executionID ),
688702 }
689703 execHelper .initLimiters (e .cfg .LocalLimiters )
690- result , execErr := e .cfg .Module .Execute (execCtx , & sdkpb.ExecuteRequest {
704+ var result * sdkpb.ExecutionResult
705+ result , execErr = e .cfg .Module .Execute (execCtx , & sdkpb.ExecuteRequest {
691706 Request : & sdkpb.ExecuteRequest_Trigger {
692707 Trigger : & sdkpb.Trigger {
693708 Id : tid ,
@@ -730,11 +745,7 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
730745 } else {
731746 e .metrics .UpdateWorkflowErrorDurationHistogram (ctx , int64 (executionDuration .Seconds ()))
732747 }
733-
734748 executionLogger .Errorw ("Workflow execution failed with module execution error" , "status" , executionStatus , "durationMs" , executionDuration .Milliseconds (), "err" , execErr )
735- _ = events .EmitExecutionFinishedEvent (ctx , loggerLabels , executionStatus , executionID , lggr )
736- e .cfg .Hooks .OnExecutionFinished (executionID , executionStatus )
737- e .cfg .Hooks .OnExecutionError (execErr .Error ())
738749 return
739750 }
740751
@@ -744,22 +755,18 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
744755
745756 if len (result .GetError ()) > 0 {
746757 executionStatus = store .StatusErrored
758+ execErr = errors .New (result .GetError ())
747759 e .metrics .UpdateWorkflowErrorDurationHistogram (ctx , int64 (executionDuration .Seconds ()))
748760 e .metrics .With ("workflowID" , e .cfg .WorkflowID , "workflowName" , e .cfg .WorkflowName .String ()).IncrementWorkflowExecutionFailedCounter (ctx )
749761 executionLogger .Errorw ("Workflow execution failed" , "status" , executionStatus , "durationMs" , executionDuration .Milliseconds (), "error" , result .GetError ())
750- _ = events .EmitExecutionFinishedEvent (ctx , loggerLabels , executionStatus , executionID , lggr )
751- e .cfg .Hooks .OnExecutionFinished (executionID , executionStatus )
752- e .cfg .Hooks .OnExecutionError (result .GetError ())
753762 return
754763 }
755764
756765 executionStatus = store .StatusCompleted
757766 executionLogger .Infow ("Workflow execution finished successfully" , "durationMs" , executionDuration .Milliseconds ())
758- _ = events .EmitExecutionFinishedEvent (ctx , loggerLabels , executionStatus , executionID , lggr )
759767 e .metrics .UpdateWorkflowCompletedDurationHistogram (ctx , int64 (executionDuration .Seconds ()))
760768 e .metrics .With ("workflowID" , e .cfg .WorkflowID , "workflowName" , e .cfg .WorkflowName .String ()).IncrementWorkflowExecutionSucceededCounter (ctx )
761769 e .cfg .Hooks .OnResultReceived (result )
762- e .cfg .Hooks .OnExecutionFinished (executionID , executionStatus )
763770}
764771
765772func (e * Engine ) secretsFetcher (phaseID string ) SecretsFetcher {
0 commit comments