@@ -159,6 +159,8 @@ func (engine *Engine) executeStep(ctx context.Context, instance *WorkflowInstanc
159159 output , stepErr = engine .executeJoin (ctx , instance , step , stepDef )
160160 case StepTypeParallel :
161161 output , stepErr = engine .executeFork (ctx , instance , step , stepDef )
162+ case StepTypeSavePoint :
163+ output = step .Input
162164 default :
163165 stepErr = fmt .Errorf ("unsupported step type: %s" , stepDef .Type )
164166 }
@@ -410,6 +412,18 @@ func (engine *Engine) handleStepFailure(
410412 return fmt .Errorf ("notify join steps: %w" , err )
411413 }
412414
415+ // Try to rollback to save point before handling failure
416+ def , err := engine .store .GetWorkflowDefinition (ctx , instance .WorkflowID )
417+ if err == nil {
418+ if rollbackErr := engine .rollbackToSavePoint (ctx , instance .ID , step .StepName , def ); rollbackErr != nil {
419+ // Log rollback error but continue with failure handling
420+ _ = engine .store .LogEvent (ctx , instance .ID , & step .ID , EventStepFailed , map [string ]any {
421+ KeyStepName : step .StepName ,
422+ KeyError : fmt .Sprintf ("rollback failed: %v" , rollbackErr ),
423+ })
424+ }
425+ }
426+
413427 if stepDef .OnFailure != "" {
414428 return engine .enqueueNextSteps (ctx , instance .ID , []string {stepDef .OnFailure }, step .Input )
415429 }
@@ -664,3 +678,158 @@ func (engine *Engine) validateDefinition(def *WorkflowDefinition) error {
664678
665679 return nil
666680}
681+
682+ func (engine * Engine ) rollbackToSavePoint (
683+ ctx context.Context ,
684+ instanceID int64 ,
685+ failedStepName string ,
686+ def * WorkflowDefinition ,
687+ ) error {
688+ savePointName := engine .findNearestSavePoint (failedStepName , def )
689+ if savePointName == "" {
690+ return engine .rollbackAllSteps (ctx , instanceID , failedStepName , def )
691+ }
692+
693+ return engine .rollbackStepsToSavePoint (ctx , instanceID , failedStepName , savePointName , def )
694+ }
695+
696+ func (engine * Engine ) findNearestSavePoint (stepName string , def * WorkflowDefinition ) string {
697+ visited := make (map [string ]bool )
698+
699+ for stepName != "" {
700+ if visited [stepName ] {
701+ break // Prevent infinite loops
702+ }
703+ visited [stepName ] = true
704+
705+ stepDef , ok := def .Definition .Steps [stepName ]
706+ if ! ok {
707+ break
708+ }
709+
710+ if stepDef .Type == StepTypeSavePoint {
711+ return stepName
712+ }
713+
714+ stepName = stepDef .Prev
715+ }
716+
717+ return ""
718+ }
719+
720+ func (engine * Engine ) rollbackAllSteps (
721+ ctx context.Context ,
722+ instanceID int64 ,
723+ failedStepName string ,
724+ def * WorkflowDefinition ,
725+ ) error {
726+ steps , err := engine .store .GetStepsByInstance (ctx , instanceID )
727+ if err != nil {
728+ return fmt .Errorf ("get steps by instance: %w" , err )
729+ }
730+
731+ for _ , step := range steps {
732+ if step .Status == StepStatusCompleted {
733+ if err := engine .rollbackStep (ctx , step , def ); err != nil {
734+ return fmt .Errorf ("rollback step %s: %w" , step .StepName , err )
735+ }
736+ }
737+ }
738+
739+ return nil
740+ }
741+
742+ func (engine * Engine ) rollbackStepsToSavePoint (
743+ ctx context.Context ,
744+ instanceID int64 ,
745+ failedStepName , savePointName string ,
746+ def * WorkflowDefinition ,
747+ ) error {
748+ steps , err := engine .store .GetStepsByInstance (ctx , instanceID )
749+ if err != nil {
750+ return fmt .Errorf ("get steps by instance: %w" , err )
751+ }
752+
753+ stepMap := make (map [string ]* WorkflowStep )
754+ for _ , step := range steps {
755+ stepMap [step .StepName ] = step
756+ }
757+
758+ return engine .rollbackStepChain (ctx , failedStepName , savePointName , def , stepMap )
759+ }
760+
761+ func (engine * Engine ) rollbackStepChain (
762+ ctx context.Context ,
763+ currentStep , savePointName string ,
764+ def * WorkflowDefinition ,
765+ stepMap map [string ]* WorkflowStep ,
766+ ) error {
767+ if currentStep == savePointName {
768+ return nil // Reached save point
769+ }
770+
771+ stepDef , ok := def .Definition .Steps [currentStep ]
772+ if ! ok {
773+ return fmt .Errorf ("step definition not found: %s" , currentStep )
774+ }
775+
776+ if step , exists := stepMap [currentStep ]; exists && step .Status == StepStatusCompleted {
777+ if err := engine .rollbackStep (ctx , step , def ); err != nil {
778+ return fmt .Errorf ("rollback step %s: %w" , currentStep , err )
779+ }
780+ }
781+
782+ // Handle parallel steps (fork branches)
783+ if stepDef .Type == StepTypeFork || stepDef .Type == StepTypeParallel {
784+ for _ , parallelStepName := range stepDef .Parallel {
785+ if err := engine .rollbackStepChain (ctx , parallelStepName , savePointName , def , stepMap ); err != nil {
786+ return err
787+ }
788+ }
789+ }
790+
791+ // Continue with a previous step
792+ if stepDef .Prev != "" {
793+ return engine .rollbackStepChain (ctx , stepDef .Prev , savePointName , def , stepMap )
794+ }
795+
796+ return nil
797+ }
798+
799+ func (engine * Engine ) rollbackStep (ctx context.Context , step * WorkflowStep , def * WorkflowDefinition ) error {
800+ stepDef , ok := def .Definition .Steps [step .StepName ]
801+ if ! ok {
802+ return fmt .Errorf ("step definition not found: %s" , step .StepName )
803+ }
804+
805+ handler , exists := engine .handlers [stepDef .OnFailure ]
806+ if ! exists {
807+ return fmt .Errorf ("handler not found: %s" , stepDef .Handler )
808+ }
809+
810+ stepCtx := & executionContext {
811+ instanceID : step .InstanceID ,
812+ stepName : step .StepName ,
813+ retryCount : step .RetryCount ,
814+ variables : stepDef .Metadata ,
815+ }
816+
817+ // Execute the handler in compensation mode
818+ _ , err := handler .Execute (ctx , stepCtx , step .Input )
819+ if err != nil {
820+ return fmt .Errorf ("execute compensation for step %q: %w" , step .StepName , err )
821+ }
822+
823+ // Update step status to rolled back
824+ if err := engine .store .UpdateStep (ctx , step .ID , StepStatusRolledBack , step .Input , nil ); err != nil {
825+ return fmt .Errorf ("update step status: %w" , err )
826+ }
827+
828+ _ = engine .store .LogEvent (ctx , step .InstanceID , & step .ID , EventStepFailed , map [string ]any {
829+ KeyStepName : step .StepName ,
830+ KeyStepType : step .StepType ,
831+ KeyError : "step rolled back" ,
832+ })
833+
834+ return nil
835+ }
0 commit comments