Repository: apache/incubator-seata-go

Issue:
Feature: saga Interface optimization
Saga is a distributed transaction processing mode used to ensure consistency in distributed systems.The state machine mode of saga implemented in seata java, it`s needs to be support in seata go.

history saga interface impl need to optimization,The user interface needs to be simplified to make it easier to understand, and the interface between processes also needs to be simpler.

Hints from issue comments:
@wt-better: pls assign to me

Solution patch:
diff --git a/pkg/client/config.go b/pkg/client/config.go
index bd84852..c4fce0f 100644
--- a/pkg/client/config.go
+++ b/pkg/client/config.go
@@ -20,6 +20,7 @@ package client
 import (
 	"flag"
 	"fmt"
+	"github.com/seata/seata-go/pkg/saga"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -84,6 +85,8 @@ type Config struct {
 	TransportConfig   remoteConfig.TransportConfig `yaml:"transport" json:"transport" koanf:"transport"`
 	ServiceConfig     discovery.ServiceConfig      `yaml:"service" json:"service" koanf:"service"`
 	RegistryConfig    discovery.RegistryConfig     `yaml:"registry" json:"registry" koanf:"registry"`
+
+	SagaConfig saga.Config `yaml:"saga" json:"saga" koanf:"saga"`
 }
 
 func (c *Config) RegisterFlags(f *flag.FlagSet) {
@@ -102,6 +105,7 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
 	c.TransportConfig.RegisterFlagsWithPrefix("transport", f)
 	c.RegistryConfig.RegisterFlagsWithPrefix("registry", f)
 	c.ServiceConfig.RegisterFlagsWithPrefix("service", f)
+	c.SagaConfig.RegisterFlagsWithPrefix("saga", f)
 }
 
 type loaderConf struct {
diff --git a/pkg/saga/config.go b/pkg/saga/config.go
new file mode 100644
index 0000000..dd72bc6
--- /dev/null
+++ b/pkg/saga/config.go
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package saga
+
+import (
+	"flag"
+
+	"github.com/seata/seata-go/pkg/saga/statemachine"
+)
+
+type Config struct {
+	StateMachine *statemachine.StateMachineObject `yaml:"state-machine" json:"state-machine" koanf:"state-machine"`
+}
+
+func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	if cfg.StateMachine != nil {
+		cfg.StateMachine.RegisterFlagsWithPrefix(prefix+".state-machine", f)
+	}
+}
diff --git a/pkg/saga/statemachine/constant/constant.go b/pkg/saga/statemachine/constant/constant.go
index 72af09e..6446f76 100644
--- a/pkg/saga/statemachine/constant/constant.go
+++ b/pkg/saga/statemachine/constant/constant.go
@@ -64,9 +64,12 @@ const (
 	VarNameCurrentCompensationHolder     string = "_current_compensation_holder_"
 	VarNameFirstCompensationStateStarted string = "_first_compensation_state_started"
 	VarNameCurrentLoopContextHolder      string = "_current_loop_context_holder_"
+	VarNameRetriedStateInstId            string = "_retried_state_instance_id"
+	VarNameIsForSubStatMachineForward    string = "_is_for_sub_statemachine_forward_"
 	// TODO: this lock in process context only has one, try to add more to add concurrent
 	VarNameProcessContextMutexLock string = "_current_context_mutex_lock"
 	VarNameFailEndStateFlag        string = "_fail_end_state_flag_"
+	VarNameGlobalTx                string = "_global_transaction_"
 	// end region
 
 	// region of loop
@@ -79,10 +82,12 @@ const (
 	// end region
 
 	// region others
+	SeqEntityStateMachine     string = "STATE_MACHINE"
 	SeqEntityStateMachineInst string = "STATE_MACHINE_INST"
 	SeqEntityStateInst        string = "STATE_INST"
 	OperationNameForward      string = "forward"
 	LoopStateNamePattern      string = "-loop-"
+	SagaTransNamePrefix       string = "$Saga_"
 	// end region
 
 	SeperatorParentId string = ":"
diff --git a/pkg/saga/statemachine/engine/core/default_statemachine_config.go b/pkg/saga/statemachine/engine/core/default_statemachine_config.go
index fa2ef15..cd118f5 100644
--- a/pkg/saga/statemachine/engine/core/default_statemachine_config.go
+++ b/pkg/saga/statemachine/engine/core/default_statemachine_config.go
@@ -25,16 +25,24 @@ import (
 )
 
 const (
-	DefaultTransOperTimeout     = 60000 * 30
-	DefaultServiceInvokeTimeout = 60000 * 5
+	DefaultTransOperTimeout                      = 60000 * 30
+	DefaultServiceInvokeTimeout                  = 60000 * 5
+	DefaultClientSagaRetryPersistModeUpdate      = false
+	DefaultClientSagaCompensatePersistModeUpdate = false
+	DefaultClientReportSuccessEnable             = false
+	DefaultClientSagaBranchRegisterEnable        = true
 )
 
 type DefaultStateMachineConfig struct {
 	// Configuration
-	transOperationTimeout int
-	serviceInvokeTimeout  int
-	charset               string
-	defaultTenantId       string
+	transOperationTimeout           int
+	serviceInvokeTimeout            int
+	charset                         string
+	defaultTenantId                 string
+	sagaRetryPersistModeUpdate      bool
+	sagaCompensatePersistModeUpdate bool
+	sagaBranchRegisterEnable        bool
+	rmReportSuccessEnable           bool
 
 	// Components
 
@@ -202,13 +210,49 @@ func (c *DefaultStateMachineConfig) ServiceInvokeTimeout() int {
 	return c.serviceInvokeTimeout
 }
 
+func (c *DefaultStateMachineConfig) IsSagaRetryPersistModeUpdate() bool {
+	return c.sagaRetryPersistModeUpdate
+}
+
+func (c *DefaultStateMachineConfig) SetSagaRetryPersistModeUpdate(sagaRetryPersistModeUpdate bool) {
+	c.sagaRetryPersistModeUpdate = sagaRetryPersistModeUpdate
+}
+
+func (c *DefaultStateMachineConfig) IsSagaCompensatePersistModeUpdate() bool {
+	return c.sagaCompensatePersistModeUpdate
+}
+
+func (c *DefaultStateMachineConfig) SetSagaCompensatePersistModeUpdate(sagaCompensatePersistModeUpdate bool) {
+	c.sagaCompensatePersistModeUpdate = sagaCompensatePersistModeUpdate
+}
+
+func (c *DefaultStateMachineConfig) IsSagaBranchRegisterEnable() bool {
+	return c.sagaBranchRegisterEnable
+}
+
+func (c *DefaultStateMachineConfig) SetSagaBranchRegisterEnable(sagaBranchRegisterEnable bool) {
+	c.sagaBranchRegisterEnable = sagaBranchRegisterEnable
+}
+
+func (c *DefaultStateMachineConfig) IsRmReportSuccessEnable() bool {
+	return c.rmReportSuccessEnable
+}
+
+func (c *DefaultStateMachineConfig) SetRmReportSuccessEnable(rmReportSuccessEnable bool) {
+	c.rmReportSuccessEnable = rmReportSuccessEnable
+}
+
 func NewDefaultStateMachineConfig() *DefaultStateMachineConfig {
 	c := &DefaultStateMachineConfig{
-		transOperationTimeout: DefaultTransOperTimeout,
-		serviceInvokeTimeout:  DefaultServiceInvokeTimeout,
-		charset:               "UTF-8",
-		defaultTenantId:       "000001",
-		componentLock:         &sync.Mutex{},
+		transOperationTimeout:           DefaultTransOperTimeout,
+		serviceInvokeTimeout:            DefaultServiceInvokeTimeout,
+		charset:                         "UTF-8",
+		defaultTenantId:                 "000001",
+		sagaRetryPersistModeUpdate:      DefaultClientSagaRetryPersistModeUpdate,
+		sagaCompensatePersistModeUpdate: DefaultClientSagaCompensatePersistModeUpdate,
+		sagaBranchRegisterEnable:        DefaultClientSagaBranchRegisterEnable,
+		rmReportSuccessEnable:           DefaultClientReportSuccessEnable,
+		componentLock:                   &sync.Mutex{},
 	}
 
 	// TODO: init config
diff --git a/pkg/saga/statemachine/engine/core/process_ctrl_statemachine_engine.go b/pkg/saga/statemachine/engine/core/process_ctrl_statemachine_engine.go
index 1b59123..93f2dd4 100644
--- a/pkg/saga/statemachine/engine/core/process_ctrl_statemachine_engine.go
+++ b/pkg/saga/statemachine/engine/core/process_ctrl_statemachine_engine.go
@@ -41,15 +41,99 @@ func NewProcessCtrlStateMachineEngine() *ProcessCtrlStateMachineEngine {
 	}
 }
 
-func (p ProcessCtrlStateMachineEngine) Start(ctx context.Context, stateMachineName string, tenantId string, startParams map[string]interface{}) (statelang.StateMachineInstance, error) {
+func (p ProcessCtrlStateMachineEngine) Start(ctx context.Context, stateMachineName string, tenantId string,
+	startParams map[string]interface{}) (statelang.StateMachineInstance, error) {
 	return p.startInternal(ctx, stateMachineName, tenantId, "", startParams, false, nil)
 }
 
+func (p ProcessCtrlStateMachineEngine) StartAsync(ctx context.Context, stateMachineName string, tenantId string,
+	startParams map[string]interface{}, callback CallBack) (statelang.StateMachineInstance, error) {
+	return p.startInternal(ctx, stateMachineName, tenantId, "", startParams, true, callback)
+}
+
+func (p ProcessCtrlStateMachineEngine) StartWithBusinessKey(ctx context.Context, stateMachineName string,
+	tenantId string, businessKey string, startParams map[string]interface{}) (statelang.StateMachineInstance, error) {
+	return p.startInternal(ctx, stateMachineName, tenantId, businessKey, startParams, false, nil)
+}
+
+func (p ProcessCtrlStateMachineEngine) StartWithBusinessKeyAsync(ctx context.Context, stateMachineName string,
+	tenantId string, businessKey string, startParams map[string]interface{}, callback CallBack) (statelang.StateMachineInstance, error) {
+	return p.startInternal(ctx, stateMachineName, tenantId, businessKey, startParams, true, callback)
+}
+
+func (p ProcessCtrlStateMachineEngine) Forward(ctx context.Context, stateMachineInstId string,
+	replaceParams map[string]interface{}) (statelang.StateMachineInstance, error) {
+	return p.forwardInternal(ctx, stateMachineInstId, replaceParams, false, false, nil)
+}
+
+func (p ProcessCtrlStateMachineEngine) ForwardAsync(ctx context.Context, stateMachineInstId string, replaceParams map[string]interface{}, callback CallBack) (statelang.StateMachineInstance, error) {
+	return p.forwardInternal(ctx, stateMachineInstId, replaceParams, false, true, callback)
+}
+
 func (p ProcessCtrlStateMachineEngine) Compensate(ctx context.Context, stateMachineInstId string,
 	replaceParams map[string]any) (statelang.StateMachineInstance, error) {
 	return p.compensateInternal(ctx, stateMachineInstId, replaceParams, false, nil)
 }
 
+func (p ProcessCtrlStateMachineEngine) CompensateAsync(ctx context.Context, stateMachineInstId string, replaceParams map[string]interface{}, callback CallBack) (statelang.StateMachineInstance, error) {
+	return p.compensateInternal(ctx, stateMachineInstId, replaceParams, true, callback)
+}
+
+func (p ProcessCtrlStateMachineEngine) SkipAndForward(ctx context.Context, stateMachineInstId string, replaceParams map[string]interface{}) (statelang.StateMachineInstance, error) {
+	return p.forwardInternal(ctx, stateMachineInstId, replaceParams, true, false, nil)
+}
+
+func (p ProcessCtrlStateMachineEngine) SkipAndForwardAsync(ctx context.Context, stateMachineInstId string, callback CallBack) (statelang.StateMachineInstance, error) {
+	return p.forwardInternal(ctx, stateMachineInstId, nil, true, true, callback)
+}
+
+func (p ProcessCtrlStateMachineEngine) GetStateMachineConfig() StateMachineConfig {
+	return p.StateMachineConfig
+}
+
+func (p ProcessCtrlStateMachineEngine) ReloadStateMachineInstance(ctx context.Context, instId string) (statelang.StateMachineInstance, error) {
+	inst, err := p.StateMachineConfig.StateLogStore().GetStateMachineInstance(instId)
+	if err != nil {
+		return nil, err
+	}
+	if inst != nil {
+		stateMachine := inst.StateMachine()
+		if stateMachine == nil {
+			stateMachine, err = p.StateMachineConfig.StateMachineRepository().GetStateMachineById(inst.MachineID())
+			if err != nil {
+				return nil, err
+			}
+			inst.SetStateMachine(stateMachine)
+		}
+		if stateMachine == nil {
+			return nil, exception.NewEngineExecutionException(seataErrors.ObjectNotExists,
+				"StateMachine[id:"+inst.MachineID()+"] not exist.", nil)
+		}
+
+		stateList := inst.StateList()
+		if len(stateList) == 0 {
+			stateList, err = p.StateMachineConfig.StateLogStore().GetStateInstanceListByMachineInstanceId(instId)
+			if err != nil {
+				return nil, err
+			}
+			if len(stateList) > 0 {
+				for _, tmpStateInstance := range stateList {
+					inst.PutState(tmpStateInstance.ID(), tmpStateInstance)
+				}
+			}
+		}
+
+		if len(inst.EndParams()) == 0 {
+			endParams, err := p.replayContextVariables(ctx, inst)
+			if err != nil {
+				return nil, err
+			}
+			inst.SetEndParams(endParams)
+		}
+	}
+	return inst, nil
+}
+
 func (p ProcessCtrlStateMachineEngine) startInternal(ctx context.Context, stateMachineName string, tenantId string,
 	businessKey string, startParams map[string]interface{}, async bool, callback CallBack) (statelang.StateMachineInstance, error) {
 	if tenantId == "" {
@@ -104,6 +188,184 @@ func (p ProcessCtrlStateMachineEngine) startInternal(ctx context.Context, stateM
 	return stateMachineInstance, nil
 }
 
+func (p ProcessCtrlStateMachineEngine) forwardInternal(ctx context.Context, stateMachineInstId string,
+	replaceParams map[string]interface{}, skip bool, async bool, callback CallBack) (statelang.StateMachineInstance, error) {
+	stateMachineInstance, err := p.reloadStateMachineInstance(ctx, stateMachineInstId)
+	if err != nil {
+		return nil, err
+	}
+
+	if stateMachineInstance == nil {
+		return nil, exception.NewEngineExecutionException(seataErrors.StateMachineInstanceNotExists, "StateMachineInstance is not exists", nil)
+	}
+
+	if stateMachineInstance.Status() == statelang.SU && stateMachineInstance.CompensationStatus() == "" {
+		return stateMachineInstance, nil
+	}
+
+	acceptStatus := []statelang.ExecutionStatus{statelang.FA, statelang.UN, statelang.RU}
+	if _, err := p.checkStatus(ctx, stateMachineInstance, acceptStatus, nil, stateMachineInstance.Status(), "", "forward"); err != nil {
+		return nil, err
+	}
+
+	actList := stateMachineInstance.StateList()
+	if len(actList) == 0 {
+		return nil, exception.NewEngineExecutionException(seataErrors.OperationDenied,
+			fmt.Sprintf("StateMachineInstance[id:%s] has no stateInstance, please start a new StateMachine execution instead", stateMachineInstId), nil)
+	}
+
+	lastForwardState, err := p.findOutLastForwardStateInstance(actList)
+	if err != nil {
+		return nil, err
+	}
+	if lastForwardState == nil {
+		return nil, exception.NewEngineExecutionException(seataErrors.OperationDenied,
+			fmt.Sprintf("StateMachineInstance[id:%s] Cannot find last forward execution stateInstance", stateMachineInstId), nil)
+	}
+
+	contextBuilder := NewProcessContextBuilder().
+		WithProcessType(process.StateLang).
+		WithOperationName(constant.OperationNameForward).
+		WithAsyncCallback(callback).
+		WithStateMachineInstance(stateMachineInstance).
+		WithStateInstance(lastForwardState).
+		WithStateMachineConfig(p.StateMachineConfig).
+		WithStateMachineEngine(p).
+		WithIsAsyncExecution(async)
+
+	context := contextBuilder.Build()
+
+	contextVariables, err := p.getStateMachineContextVariables(ctx, stateMachineInstance)
+	if err != nil {
+		return nil, err
+	}
+
+	if replaceParams != nil {
+		for k, v := range replaceParams {
+			contextVariables[k] = v
+		}
+	}
+	p.putBusinesskeyToContextariables(stateMachineInstance, contextVariables)
+
+	concurrentContextVariables := p.copyMap(contextVariables)
+
+	context.SetVariable(constant.VarNameStateMachineContext, concurrentContextVariables)
+	stateMachineInstance.SetContext(concurrentContextVariables)
+
+	originStateName := GetOriginStateName(lastForwardState)
+	lastState := stateMachineInstance.StateMachine().State(originStateName)
+	loop := GetLoopConfig(ctx, context, lastState)
+	if loop != nil && lastForwardState.Status() == statelang.SU {
+		lastForwardState = p.findOutLastNeedForwardStateInstance(ctx, context)
+	}
+
+	context.SetVariable(lastForwardState.Name()+constant.VarNameRetriedStateInstId, lastForwardState.ID())
+	if lastForwardState.Type() == constant.StateTypeSubStateMachine && lastForwardState.CompensationStatus() != statelang.SU {
+		context.SetVariable(constant.VarNameIsForSubStatMachineForward, true)
+	}
+
+	if lastForwardState.Status() != statelang.SU {
+		lastForwardState.SetIgnoreStatus(true)
+	}
+
+	inst := NewStateInstruction(stateMachineInstance.StateMachine().Name(), stateMachineInstance.TenantID())
+	if skip || lastForwardState.Status() == statelang.SU {
+		next := ""
+		curState := stateMachineInstance.StateMachine().State(GetOriginStateName(lastForwardState))
+		if taskState, ok := curState.(*state.AbstractTaskState); ok {
+			next = taskState.Next()
+		}
+		if next == "" {
+			log.Warn(fmt.Sprintf("Last Forward execution StateInstance was succeed, and it has not Next State, skip forward operation"))
+			return stateMachineInstance, nil
+		}
+		inst.SetStateName(next)
+	} else {
+		if lastForwardState.Status() == statelang.RU && !IsTimeout(lastForwardState.StartedTime(), p.StateMachineConfig.ServiceInvokeTimeout()) {
+			return nil, exception.NewEngineExecutionException(seataErrors.OperationDenied,
+				fmt.Sprintf("State [%s] is running, operation[forward] denied", lastForwardState.Name()), nil)
+		}
+		inst.SetStateName(GetOriginStateName(lastForwardState))
+	}
+	context.SetInstruction(inst)
+
+	stateMachineInstance.SetStatus(statelang.RU)
+	stateMachineInstance.SetRunning(true)
+
+	log.Info(fmt.Sprintf("Operation [forward] started  stateMachineInstance[id:%s]", stateMachineInstance.ID()))
+
+	if stateMachineInstance.StateMachine().IsPersist() {
+		if err := p.StateMachineConfig.StateLogStore().RecordStateMachineRestarted(ctx, stateMachineInstance, context); err != nil {
+			return nil, err
+		}
+	}
+
+	curState, err := inst.GetState(context)
+	if err != nil {
+		return nil, err
+	}
+	loop = GetLoopConfig(ctx, context, curState)
+	if loop != nil {
+		inst.SetTemporaryState(state.NewLoopStartStateImpl())
+	}
+
+	if async {
+		if _, err := p.StateMachineConfig.AsyncEventPublisher().PushEvent(ctx, context); err != nil {
+			return nil, err
+		}
+	} else {
+		if _, err := p.StateMachineConfig.EventPublisher().PushEvent(ctx, context); err != nil {
+			return nil, err
+		}
+	}
+
+	return stateMachineInstance, nil
+}
+
+func (p ProcessCtrlStateMachineEngine) findOutLastForwardStateInstance(stateInstanceList []statelang.StateInstance) (statelang.StateInstance, error) {
+	var lastForwardStateInstance statelang.StateInstance
+	var err error
+	for i := len(stateInstanceList) - 1; i >= 0; i-- {
+		stateInstance := stateInstanceList[i]
+		if !stateInstance.IsForCompensation() {
+			if stateInstance.CompensationStatus() == statelang.SU {
+				continue
+			}
+
+			if stateInstance.Type() == constant.StateTypeSubStateMachine {
+				finalState := stateInstance
+				for finalState.StateIDRetriedFor() != "" {
+					if finalState, err = p.StateMachineConfig.StateLogStore().GetStateInstance(finalState.StateIDRetriedFor(),
+						finalState.MachineInstanceID()); err != nil {
+						return nil, err
+					}
+				}
+
+				subInst, _ := p.StateMachineConfig.StateLogStore().GetStateMachineInstanceByParentId(GenerateParentId(finalState))
+				if len(subInst) > 0 {
+					if subInst[0].CompensationStatus() == statelang.SU {
+						continue
+					}
+
+					if subInst[0].CompensationStatus() == statelang.UN {
+						return nil, exception.NewEngineExecutionException(seataErrors.ForwardInvalid,
+							"Last forward execution state instance is SubStateMachine and compensation status is [UN], Operation[forward] denied, stateInstanceId:"+stateInstance.ID(),
+							nil)
+					}
+				}
+			} else if stateInstance.CompensationStatus() == statelang.UN {
+				return nil, exception.NewEngineExecutionException(seataErrors.ForwardInvalid,
+					"Last forward execution state instance compensation status is [UN], Operation[forward] denied, stateInstanceId:"+stateInstance.ID(),
+					nil)
+			}
+
+			lastForwardStateInstance = stateInstance
+			break
+		}
+	}
+	return lastForwardStateInstance, nil
+}
+
 // copyMap not deep copy, so best practice: Don’t pass by reference
 func (p ProcessCtrlStateMachineEngine) copyMap(startParams map[string]interface{}) map[string]interface{} {
 	copyMap := make(map[string]interface{}, len(startParams))
@@ -439,3 +701,17 @@ func (p ProcessCtrlStateMachineEngine) nullSafeCopy(srcMap map[string]any, destM
 		}
 	}
 }
+
+func (p ProcessCtrlStateMachineEngine) findOutLastNeedForwardStateInstance(ctx context.Context, processContext ProcessContext) statelang.StateInstance {
+	stateMachineInstance := processContext.GetVariable(constant.VarNameStateMachineInst).(statelang.StateMachineInstance)
+	lastForwardState := processContext.GetVariable(constant.VarNameStateInst).(statelang.StateInstance)
+
+	actList := stateMachineInstance.StateList()
+	for i := len(actList) - 1; i >= 0; i-- {
+		stateInstance := actList[i]
+		if GetOriginStateName(stateInstance) == GetOriginStateName(lastForwardState) && stateInstance.Status() != statelang.SU {
+			return stateInstance
+		}
+	}
+	return lastForwardState
+}
diff --git a/pkg/saga/statemachine/engine/core/statemachine_engine.go b/pkg/saga/statemachine/engine/core/statemachine_engine.go
index 75cdb1a..6ad3620 100644
--- a/pkg/saga/statemachine/engine/core/statemachine_engine.go
+++ b/pkg/saga/statemachine/engine/core/statemachine_engine.go
@@ -23,8 +23,33 @@ import (
 )
 
 type StateMachineEngine interface {
+	// Start starts a state machine instance
 	Start(ctx context.Context, stateMachineName string, tenantId string, startParams map[string]interface{}) (statelang.StateMachineInstance, error)
-	Compensate(ctx context.Context, stateMachineInstId string, replaceParams map[string]any) (statelang.StateMachineInstance, error)
+	// StartAsync start a state machine instance asynchronously
+	StartAsync(ctx context.Context, stateMachineName string, tenantId string, startParams map[string]interface{},
+		callback CallBack) (statelang.StateMachineInstance, error)
+	// StartWithBusinessKey starts a state machine instance with a business key
+	StartWithBusinessKey(ctx context.Context, stateMachineName string, tenantId string, businessKey string,
+		startParams map[string]interface{}) (statelang.StateMachineInstance, error)
+	// StartWithBusinessKeyAsync starts a state machine instance with a business key asynchronously
+	StartWithBusinessKeyAsync(ctx context.Context, stateMachineName string, tenantId string, businessKey string,
+		startParams map[string]interface{}, callback CallBack) (statelang.StateMachineInstance, error)
+	// Forward  restart a failed state machine instance
+	Forward(ctx context.Context, stateMachineInstId string, replaceParams map[string]interface{}) (statelang.StateMachineInstance, error)
+	// ForwardAsync restart a failed state machine instance asynchronously
+	ForwardAsync(ctx context.Context, stateMachineInstId string, replaceParams map[string]interface{}, callback CallBack) (statelang.StateMachineInstance, error)
+	// Compensate compensate a state machine instance
+	Compensate(ctx context.Context, stateMachineInstId string, replaceParams map[string]interface{}) (statelang.StateMachineInstance, error)
+	// CompensateAsync compensate a state machine instance asynchronously
+	CompensateAsync(ctx context.Context, stateMachineInstId string, replaceParams map[string]interface{}, callback CallBack) (statelang.StateMachineInstance, error)
+	// SkipAndForward skips the current failed state instance and restarts the state machine instance
+	SkipAndForward(ctx context.Context, stateMachineInstId string, replaceParams map[string]interface{}) (statelang.StateMachineInstance, error)
+	// SkipAndForwardAsync skips the current failed state instance and restarts the state machine instance asynchronously
+	SkipAndForwardAsync(ctx context.Context, stateMachineInstId string, callback CallBack) (statelang.StateMachineInstance, error)
+	// GetStateMachineConfig gets the state machine configurations
+	GetStateMachineConfig() StateMachineConfig
+	// ReloadStateMachineInstance reloads a state machine instance
+	ReloadStateMachineInstance(ctx context.Context, instId string) (statelang.StateMachineInstance, error)
 }
 
 type CallBack interface {
diff --git a/pkg/saga/statemachine/engine/core/statemachine_store.go b/pkg/saga/statemachine/engine/core/statemachine_store.go
index f47065d..7e0a9e9 100644
--- a/pkg/saga/statemachine/engine/core/statemachine_store.go
+++ b/pkg/saga/statemachine/engine/core/statemachine_store.go
@@ -55,6 +55,8 @@ type StateLogStore interface {
 	GetStateInstance(stateInstanceId string, stateMachineInstanceId string) (statelang.StateInstance, error)
 
 	GetStateInstanceListByMachineInstanceId(stateMachineInstanceId string) ([]statelang.StateInstance, error)
+
+	ClearUp(context ProcessContext)
 }
 
 type StateMachineRepository interface {
diff --git a/pkg/saga/statemachine/engine/core/utils.go b/pkg/saga/statemachine/engine/core/utils.go
index 49f03e0..afa1fd6 100644
--- a/pkg/saga/statemachine/engine/core/utils.go
+++ b/pkg/saga/statemachine/engine/core/utils.go
@@ -98,6 +98,14 @@ func (p *ProcessContextBuilder) WithIsAsyncExecution(async bool) *ProcessContext
 	return p
 }
 
+func (p *ProcessContextBuilder) WithStateInstance(state statelang.StateInstance) *ProcessContextBuilder {
+	if state != nil {
+		p.processContext.SetVariable(constant.VarNameStateInst, state)
+	}
+
+	return p
+}
+
 func (p *ProcessContextBuilder) Build() ProcessContext {
 	return p.processContext
 }
diff --git a/pkg/saga/statemachine/engine/invoker/http_invoker.go b/pkg/saga/statemachine/engine/invoker/http_invoker.go
new file mode 100644
index 0000000..c23531a
--- /dev/null
+++ b/pkg/saga/statemachine/engine/invoker/http_invoker.go
@@ -0,0 +1,220 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package invoker
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"reflect"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/seata/seata-go/pkg/saga/statemachine/statelang/state"
+	"github.com/seata/seata-go/pkg/util/log"
+)
+
+const errHttpCode = 400
+
+type HTTPInvoker struct {
+	clientsMapLock sync.Mutex
+	clients        map[string]HTTPClient
+}
+
+func NewHTTPInvoker() *HTTPInvoker {
+	return &HTTPInvoker{
+		clients: make(map[string]HTTPClient),
+	}
+}
+
+func (h *HTTPInvoker) RegisterClient(serviceName string, client HTTPClient) {
+	h.clientsMapLock.Lock()
+	defer h.clientsMapLock.Unlock()
+	h.clients[serviceName] = client
+}
+
+func (h *HTTPInvoker) GetClient(serviceName string) HTTPClient {
+	h.clientsMapLock.Lock()
+	defer h.clientsMapLock.Unlock()
+	return h.clients[serviceName]
+}
+
+func (h *HTTPInvoker) Invoke(ctx context.Context, input []any, service state.ServiceTaskState) (output []reflect.Value, err error) {
+	serviceTaskStateImpl := service.(*state.ServiceTaskStateImpl)
+	client := h.GetClient(serviceTaskStateImpl.ServiceName())
+	if client == nil {
+		return nil, fmt.Errorf("no http client %s for service task state", serviceTaskStateImpl.ServiceName())
+	}
+
+	if serviceTaskStateImpl.IsAsync() {
+		go func() {
+			_, err := client.Call(ctx, serviceTaskStateImpl, input)
+			if err != nil {
+				log.Errorf("invoke Service[%s].%s failed, err is %s", serviceTaskStateImpl.ServiceName(),
+					serviceTaskStateImpl.ServiceMethod(), err)
+			}
+		}()
+		return nil, nil
+	}
+
+	return client.Call(ctx, serviceTaskStateImpl, input)
+}
+
+func (h *HTTPInvoker) Close(ctx context.Context) error {
+	return nil
+}
+
+type HTTPClient interface {
+	Call(ctx context.Context, serviceTaskStateImpl *state.ServiceTaskStateImpl, input []any) ([]reflect.Value, error)
+}
+
+type HTTPClientImpl struct {
+	serviceName string
+	baseURL     string
+	client      *http.Client
+}
+
+func NewHTTPClient(serviceName string, baseURL string, client *http.Client) *HTTPClientImpl {
+	if client == nil {
+		client = &http.Client{
+			Timeout: time.Second * 30,
+		}
+	}
+	return &HTTPClientImpl{
+		serviceName: serviceName,
+		baseURL:     baseURL,
+		client:      client,
+	}
+}
+
+func (h *HTTPClientImpl) Call(ctx context.Context, serviceTaskStateImpl *state.ServiceTaskStateImpl, input []any) ([]reflect.Value, error) {
+	retryCountMap := make(map[state.Retry]int)
+	for {
+		res, err, shouldRetry := func() (res []reflect.Value, resErr error, shouldRetry bool) {
+			defer func() {
+				if r := recover(); r != nil {
+					errStr := fmt.Sprintf("%v", r)
+					retry := h.matchRetry(serviceTaskStateImpl, errStr)
+					resErr = errors.New(errStr)
+					if retry != nil {
+						shouldRetry = h.needRetry(serviceTaskStateImpl, retryCountMap, retry, resErr)
+					}
+				}
+			}()
+
+			reqBody, err := json.Marshal(input)
+			if err != nil {
+				return nil, err, false
+			}
+
+			req, err := http.NewRequestWithContext(ctx,
+				serviceTaskStateImpl.ServiceMethod(),
+				h.baseURL+serviceTaskStateImpl.Name(),
+				bytes.NewBuffer(reqBody))
+			if err != nil {
+				return nil, err, false
+			}
+
+			req.Header.Set("Content-Type", "application/json")
+
+			resp, err := h.client.Do(req)
+			if err != nil {
+				retry := h.matchRetry(serviceTaskStateImpl, err.Error())
+				if retry != nil {
+					return nil, err, h.needRetry(serviceTaskStateImpl, retryCountMap, retry, err)
+				}
+				return nil, err, false
+			}
+			defer resp.Body.Close()
+
+			body, err := io.ReadAll(resp.Body)
+			if err != nil {
+				return nil, err, false
+			}
+
+			if resp.StatusCode >= errHttpCode {
+				errStr := fmt.Sprintf("HTTP error: %d - %s", resp.StatusCode, string(body))
+				retry := h.matchRetry(serviceTaskStateImpl, errStr)
+				if retry != nil {
+					return nil, errors.New(errStr), h.needRetry(serviceTaskStateImpl, retryCountMap, retry, err)
+				}
+				return nil, errors.New(errStr), false
+			}
+
+			return []reflect.Value{
+				reflect.ValueOf(string(body)),
+				reflect.Zero(reflect.TypeOf((*error)(nil)).Elem()),
+			}, nil, false
+		}()
+
+		if !shouldRetry {
+			if err != nil {
+				return nil, fmt.Errorf("invoke Service[%s] failed, not satisfy retry config, the last err is %s",
+					serviceTaskStateImpl.ServiceName(), err)
+			}
+			return res, nil
+		}
+	}
+}
+
+func (h *HTTPClientImpl) matchRetry(impl *state.ServiceTaskStateImpl, str string) state.Retry {
+	if impl.Retry() != nil {
+		for _, retry := range impl.Retry() {
+			if retry.Exceptions() != nil {
+				for _, exception := range retry.Exceptions() {
+					if strings.Contains(str, exception) {
+						return retry
+					}
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func (h *HTTPClientImpl) needRetry(impl *state.ServiceTaskStateImpl, countMap map[state.Retry]int, retry state.Retry, err error) bool {
+	attempt, exist := countMap[retry]
+	if !exist {
+		countMap[retry] = 0
+	}
+
+	if attempt >= retry.MaxAttempt() {
+		return false
+	}
+
+	intervalSecond := retry.IntervalSecond()
+	backoffRate := retry.BackoffRate()
+	var currentInterval int64
+	if attempt == 0 {
+		currentInterval = int64(intervalSecond * 1000)
+	} else {
+		currentInterval = int64(intervalSecond * backoffRate * float64(attempt) * 1000)
+	}
+
+	log.Warnf("invoke service[%s.%s] failed, will retry after %s millis, current retry count: %s, current err: %s",
+		impl.ServiceName(), impl.ServiceMethod(), currentInterval, attempt, err)
+
+	time.Sleep(time.Duration(currentInterval) * time.Millisecond)
+	countMap[retry] = attempt + 1
+	return true
+}
diff --git a/pkg/saga/statemachine/statelang/parser/statemachine_config_parser.go b/pkg/saga/statemachine/statelang/parser/statemachine_config_parser.go
index 63276e2..a33bbde 100644
--- a/pkg/saga/statemachine/statelang/parser/statemachine_config_parser.go
+++ b/pkg/saga/statemachine/statelang/parser/statemachine_config_parser.go
@@ -19,16 +19,20 @@ package parser
 
 import (
 	"bytes"
-	"encoding/json"
 	"fmt"
-	"gopkg.in/yaml.v3"
+	"github.com/seata/seata-go/pkg/saga/statemachine"
 	"io"
 	"os"
+
+	"github.com/knadh/koanf"
+	"github.com/knadh/koanf/parsers/json"
+	"github.com/knadh/koanf/parsers/yaml"
+	"github.com/knadh/koanf/providers/rawbytes"
 )
 
 // ConfigParser is a general configuration parser interface, used to agree on the implementation of different types of parsers
 type ConfigParser interface {
-	Parse(configContent []byte) (*StateMachineObject, error)
+	Parse(configContent []byte) (*statemachine.StateMachineObject, error)
 }
 
 type JSONConfigParser struct{}
@@ -37,16 +41,21 @@ func NewJSONConfigParser() *JSONConfigParser {
 	return &JSONConfigParser{}
 }
 
-func (p *JSONConfigParser) Parse(configContent []byte) (*StateMachineObject, error) {
+func (p *JSONConfigParser) Parse(configContent []byte) (*statemachine.StateMachineObject, error) {
 	if configContent == nil || len(configContent) == 0 {
 		return nil, fmt.Errorf("empty JSON config content")
 	}
 
-	var stateMachineObject StateMachineObject
-	if err := json.Unmarshal(configContent, &stateMachineObject); err != nil {
+	k := koanf.New(".")
+	if err := k.Load(rawbytes.Provider(configContent), json.Parser()); err != nil {
 		return nil, fmt.Errorf("failed to parse JSON config content: %w", err)
 	}
 
+	var stateMachineObject statemachine.StateMachineObject
+	if err := k.Unmarshal("", &stateMachineObject); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal JSON config to struct: %w", err)
+	}
+
 	return &stateMachineObject, nil
 }
 
@@ -56,16 +65,21 @@ func NewYAMLConfigParser() *YAMLConfigParser {
 	return &YAMLConfigParser{}
 }
 
-func (p *YAMLConfigParser) Parse(configContent []byte) (*StateMachineObject, error) {
+func (p *YAMLConfigParser) Parse(configContent []byte) (*statemachine.StateMachineObject, error) {
 	if configContent == nil || len(configContent) == 0 {
 		return nil, fmt.Errorf("empty YAML config content")
 	}
 
-	var stateMachineObject StateMachineObject
-	if err := yaml.Unmarshal(configContent, &stateMachineObject); err != nil {
+	k := koanf.New(".")
+	if err := k.Load(rawbytes.Provider(configContent), yaml.Parser()); err != nil {
 		return nil, fmt.Errorf("failed to parse YAML config content: %w", err)
 	}
 
+	var stateMachineObject statemachine.StateMachineObject
+	if err := k.Unmarshal("", &stateMachineObject); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal YAML config to struct: %w", err)
+	}
+
 	return &stateMachineObject, nil
 }
 
@@ -102,18 +116,20 @@ func (p *StateMachineConfigParser) ReadConfigFile(configFilePath string) ([]byte
 }
 
 func (p *StateMachineConfigParser) getParser(content []byte) (ConfigParser, error) {
-	var obj interface{}
-	if err := json.Unmarshal(content, &obj); err == nil {
+	k := koanf.New(".")
+	if err := k.Load(rawbytes.Provider(content), json.Parser()); err == nil {
 		return NewJSONConfigParser(), nil
 	}
-	if err := yaml.Unmarshal(content, &obj); err == nil {
+
+	k = koanf.New(".")
+	if err := k.Load(rawbytes.Provider(content), yaml.Parser()); err == nil {
 		return NewYAMLConfigParser(), nil
 	}
 
 	return nil, fmt.Errorf("unsupported config file format")
 }
 
-func (p *StateMachineConfigParser) Parse(content []byte) (*StateMachineObject, error) {
+func (p *StateMachineConfigParser) Parse(content []byte) (*statemachine.StateMachineObject, error) {
 	parser, err := p.getParser(content)
 	if err != nil {
 		return nil, err
@@ -121,16 +137,3 @@ func (p *StateMachineConfigParser) Parse(content []byte) (*StateMachineObject, e
 
 	return parser.Parse(content)
 }
-
-type StateMachineObject struct {
-	Name                        string                 `json:"Name" yaml:"Name"`
-	Comment                     string                 `json:"Comment" yaml:"Comment"`
-	Version                     string                 `json:"Version" yaml:"Version"`
-	StartState                  string                 `json:"StartState" yaml:"StartState"`
-	RecoverStrategy             string                 `json:"RecoverStrategy" yaml:"RecoverStrategy"`
-	Persist                     bool                   `json:"IsPersist" yaml:"IsPersist"`
-	RetryPersistModeUpdate      bool                   `json:"IsRetryPersistModeUpdate" yaml:"IsRetryPersistModeUpdate"`
-	CompensatePersistModeUpdate bool                   `json:"IsCompensatePersistModeUpdate" yaml:"IsCompensatePersistModeUpdate"`
-	Type                        string                 `json:"Type" yaml:"Type"`
-	States                      map[string]interface{} `json:"States" yaml:"States"`
-}
diff --git a/pkg/saga/statemachine/statelang/parser/statemachine_json_parser.go b/pkg/saga/statemachine/statelang/parser/statemachine_json_parser.go
index c3003dd..37ceaae 100644
--- a/pkg/saga/statemachine/statelang/parser/statemachine_json_parser.go
+++ b/pkg/saga/statemachine/statelang/parser/statemachine_json_parser.go
@@ -106,14 +106,13 @@ func (stateMachineParser JSONStateMachineParser) Parse(content string) (statelan
 }
 
 func (stateMachineParser JSONStateMachineParser) setForCompensation(stateValue statelang.State, stateMachine *statelang.StateMachineImpl) {
-	switch stateValue.Type() {
-	case stateValue.Type():
+	if stateValue.Type() == constant.StateTypeServiceTask {
 		serviceTaskStateImpl, ok := stateValue.(*state.ServiceTaskStateImpl)
 		if ok {
 			if serviceTaskStateImpl.CompensateState() != "" {
 				compState := stateMachine.States()[serviceTaskStateImpl.CompensateState()]
 				if stateMachineParser.isTaskState(compState.Type()) {
-					compStateImpl, ok := compState.(state.ServiceTaskStateImpl)
+					compStateImpl, ok := compState.(*state.ServiceTaskStateImpl)
 					if ok {
 						compStateImpl.SetForCompensation(true)
 					}
diff --git a/pkg/saga/statemachine/statelang/statemachine_instance.go b/pkg/saga/statemachine/statelang/statemachine_instance.go
index b794e10..9ba2c06 100644
--- a/pkg/saga/statemachine/statelang/statemachine_instance.go
+++ b/pkg/saga/statemachine/statelang/statemachine_instance.go
@@ -72,6 +72,10 @@ type StateMachineInstance interface {
 
 	SetStatus(status ExecutionStatus)
 
+	StateMap() map[string]StateInstance
+
+	SetStateMap(stateMap map[string]StateInstance)
+
 	CompensationStatus() ExecutionStatus
 
 	SetCompensationStatus(compensationStatus ExecutionStatus)
@@ -234,6 +238,14 @@ func (s *StateMachineInstanceImpl) SetStatus(status ExecutionStatus) {
 	s.status = status
 }
 
+func (s *StateMachineInstanceImpl) StateMap() map[string]StateInstance {
+	return s.stateMap
+}
+
+func (s *StateMachineInstanceImpl) SetStateMap(stateMap map[string]StateInstance) {
+	s.stateMap = stateMap
+}
+
 func (s *StateMachineInstanceImpl) CompensationStatus() ExecutionStatus {
 	return s.compensationStatus
 }
diff --git a/pkg/saga/statemachine/statemachine.go b/pkg/saga/statemachine/statemachine.go
new file mode 100644
index 0000000..087f212
--- /dev/null
+++ b/pkg/saga/statemachine/statemachine.go
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package statemachine
+
+import (
+	"flag"
+)
+
+type StateMachineObject struct {
+	Name                        string                 `json:"Name" yaml:"Name"`
+	Comment                     string                 `json:"Comment" yaml:"Comment"`
+	Version                     string                 `json:"Version" yaml:"Version"`
+	StartState                  string                 `json:"StartState" yaml:"StartState"`
+	RecoverStrategy             string                 `json:"RecoverStrategy" yaml:"RecoverStrategy"`
+	Persist                     bool                   `json:"IsPersist" yaml:"IsPersist"`
+	RetryPersistModeUpdate      bool                   `json:"IsRetryPersistModeUpdate" yaml:"IsRetryPersistModeUpdate"`
+	CompensatePersistModeUpdate bool                   `json:"IsCompensatePersistModeUpdate" yaml:"IsCompensatePersistModeUpdate"`
+	Type                        string                 `json:"Type" yaml:"Type"`
+	States                      map[string]interface{} `json:"States" yaml:"States"`
+}
+
+func (smo *StateMachineObject) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	f.StringVar(&smo.Name, prefix+".name", "", "State machine name.")
+	f.StringVar(&smo.Comment, prefix+".comment", "", "State machine comment.")
+	f.StringVar(&smo.Version, prefix+".version", "1.0", "State machine version.")
+	f.StringVar(&smo.StartState, prefix+".start-state", "", "State machine start state.")
+	f.StringVar(&smo.RecoverStrategy, prefix+".recover-strategy", "", "State machine recovery strategy.")
+	f.BoolVar(&smo.Persist, prefix+".persist", false, "Whether to persist state machine.")
+	f.BoolVar(&smo.RetryPersistModeUpdate, prefix+".retry-persist-mode-update", false, "Whether to use update mode for retry persistence.")
+	f.BoolVar(&smo.CompensatePersistModeUpdate, prefix+".compensate-persist-mode-update", false, "Whether to use update mode for compensate persistence.")
+	f.StringVar(&smo.Type, prefix+".type", "", "State machine type.")
+}
diff --git a/pkg/saga/statemachine/store/db/statelang.go b/pkg/saga/statemachine/store/db/statelang.go
index f0c25e3..82e4fd2 100644
--- a/pkg/saga/statemachine/store/db/statelang.go
+++ b/pkg/saga/statemachine/store/db/statelang.go
@@ -104,7 +104,7 @@ func scanRowsToStateMachine(rows *sql.Rows) (statelang.StateMachine, error) {
 	if recoverStrategy != "" {
 		stateMachine.SetRecoverStrategy(statelang.RecoverStrategy(recoverStrategy))
 	}
-	stateMachine.SetTenantId(t)
+	stateMachine.SetTenantId(tenantId)
 	stateMachine.SetStatus(statelang.StateMachineStatus(status))
 	return stateMachine, nil
 }
diff --git a/pkg/saga/statemachine/store/db/statelog.go b/pkg/saga/statemachine/store/db/statelog.go
index ba60c61..a9235db 100644
--- a/pkg/saga/statemachine/store/db/statelog.go
+++ b/pkg/saga/statemachine/store/db/statelog.go
@@ -21,17 +21,24 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
 	"github.com/pkg/errors"
+	constant2 "github.com/seata/seata-go/pkg/constant"
+	"github.com/seata/seata-go/pkg/protocol/branch"
+	"github.com/seata/seata-go/pkg/protocol/message"
+	"github.com/seata/seata-go/pkg/rm"
 	"github.com/seata/seata-go/pkg/saga/statemachine/constant"
 	"github.com/seata/seata-go/pkg/saga/statemachine/engine/core"
 	"github.com/seata/seata-go/pkg/saga/statemachine/engine/sequence"
 	"github.com/seata/seata-go/pkg/saga/statemachine/engine/serializer"
 	"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
+	"github.com/seata/seata-go/pkg/saga/statemachine/statelang/state"
+	"github.com/seata/seata-go/pkg/tm"
 	"github.com/seata/seata-go/pkg/util/log"
-	"regexp"
-	"strconv"
-	"strings"
-	"time"
 )
 
 const (
@@ -124,14 +131,19 @@ func (s *StateLogStore) RecordStateMachineStarted(ctx context.Context, machineIn
 	parentId := machineInstance.ParentID()
 
 	if parentId == "" {
-		//TODO begin transaction
+		// begin transaction
+		err = s.beginTransaction(ctx, machineInstance, context)
+		if err != nil {
+			return err
+		}
 	}
 
 	if machineInstance.ID() == "" && s.seqGenerator != nil {
 		machineInstance.SetID(s.seqGenerator.GenerateId(constant.SeqEntityStateMachineInst, ""))
 	}
 
-	//TODO bind SAGA branch type
+	// bind SAGA branch type
+	context.SetVariable(constant2.BranchTypeKey, branch.BranchTypeSAGA)
 
 	serializedStartParams, err := s.paramsSerializer.Serialize(machineInstance.StartParams())
 	if err != nil {
@@ -149,14 +161,48 @@ func (s *StateLogStore) RecordStateMachineStarted(ctx context.Context, machineIn
 	return nil
 }
 
+func (s *StateLogStore) beginTransaction(ctx context.Context, machineInstance statelang.StateMachineInstance, context core.ProcessContext) error {
+	cfg, ok := context.GetVariable(constant.VarNameStateMachineConfig).(core.StateMachineConfig)
+	if !ok {
+		return errors.New("begin transaction fail, stateMachineConfig is required in context")
+	}
+
+	defer func() {
+		isAsync, ok := context.GetVariable(constant.VarNameIsAsyncExecution).(bool)
+		if ok && isAsync {
+			s.ClearUp(context)
+		}
+	}()
+
+	tm.SetTxRole(ctx, tm.Launcher)
+	tm.SetTxStatus(ctx, message.GlobalStatusUnKnown)
+	tm.SetTxName(ctx, constant.SagaTransNamePrefix+machineInstance.StateMachine().Name())
+
+	err := tm.GetGlobalTransactionManager().Begin(ctx, time.Duration(cfg.TransOperationTimeout()))
+	if err != nil {
+		return err
+	}
+
+	machineInstance.SetID(tm.GetXID(ctx))
+	return nil
+}
+
 func (s *StateLogStore) RecordStateMachineFinished(ctx context.Context, machineInstance statelang.StateMachineInstance,
 	context core.ProcessContext) error {
 	if machineInstance == nil {
 		return nil
 	}
 
+	defer func() {
+		s.ClearUp(context)
+	}()
+
 	endParams := machineInstance.EndParams()
+	if endParams != nil {
+		delete(endParams, constant.VarNameGlobalTx)
+	}
 
+	// if success, clear exception
 	if statelang.SU == machineInstance.Status() && machineInstance.Exception() != nil {
 		machineInstance.SetException(nil)
 	}
@@ -183,10 +229,86 @@ func (s *StateLogStore) RecordStateMachineFinished(ctx context.Context, machineI
 		return nil
 	}
 
-	//TODO check if timeout or else report transaction finished
+	// check if timeout or else report transaction finished
+	cfg, ok := context.GetVariable(constant.VarNameStateMachineConfig).(core.StateMachineConfig)
+	if !ok {
+		return errors.New("stateMachineConfig is required in context")
+	}
+
+	if core.IsTimeout(machineInstance.UpdatedTime(), cfg.TransOperationTimeout()) {
+		log.Warnf("StateMachineInstance[%s] is execution timeout, skip report transaction finished to server.", machineInstance.ID())
+	} else if machineInstance.ParentID() == "" {
+		//if parentId is not null, machineInstance is a SubStateMachine, do not report global transaction.
+		err = s.reportTransactionFinished(ctx, machineInstance, context)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *StateLogStore) reportTransactionFinished(ctx context.Context, machineInstance statelang.StateMachineInstance, context core.ProcessContext) error {
+	var err error
+	defer func() {
+		s.ClearUp(context)
+		if err != nil {
+			log.Errorf("Report transaction finish to server error: %v, StateMachine: %s, XID: %s, Reason: %s",
+				err, machineInstance.StateMachine().Name(), machineInstance.ID(), err.Error())
+		}
+	}()
+
+	globalTransaction, err := s.getGlobalTransaction(machineInstance, context)
+	if err != nil {
+		log.Errorf("Failed to get global transaction: %v", err)
+		return err
+	}
+
+	var globalStatus message.GlobalStatus
+	if statelang.SU == machineInstance.Status() && machineInstance.CompensationStatus() == "" {
+		globalStatus = message.GlobalStatusCommitted
+	} else if statelang.SU == machineInstance.CompensationStatus() {
+		globalStatus = message.GlobalStatusRollbacked
+	} else if statelang.FA == machineInstance.CompensationStatus() || statelang.UN == machineInstance.CompensationStatus() {
+		globalStatus = message.GlobalStatusRollbackRetrying
+	} else if statelang.FA == machineInstance.Status() && machineInstance.CompensationStatus() == "" {
+		globalStatus = message.GlobalStatusFinished
+	} else if statelang.UN == machineInstance.Status() && machineInstance.CompensationStatus() == "" {
+		globalStatus = message.GlobalStatusCommitRetrying
+	} else {
+		globalStatus = message.GlobalStatusUnKnown
+	}
+
+	globalTransaction.TxStatus = globalStatus
+	_, err = tm.GetGlobalTransactionManager().GlobalReport(ctx, globalTransaction)
+	if err != nil {
+		return err
+	}
 	return nil
 }
 
+func (s *StateLogStore) getGlobalTransaction(machineInstance statelang.StateMachineInstance, context core.ProcessContext) (*tm.GlobalTransaction, error) {
+	globalTransaction, ok := context.GetVariable(constant.VarNameGlobalTx).(*tm.GlobalTransaction)
+	if ok {
+		return globalTransaction, nil
+	}
+
+	var xid string
+	parentId := machineInstance.ParentID()
+	if parentId == "" {
+		xid = machineInstance.ID()
+	} else {
+		xid = parentId[:strings.LastIndex(parentId, constant.SeperatorParentId)]
+	}
+	globalTransaction = &tm.GlobalTransaction{
+		Xid:      xid,
+		TxStatus: message.GlobalStatusUnKnown,
+		TxRole:   tm.Launcher,
+	}
+
+	context.SetVariable(constant.VarNameGlobalTx, globalTransaction)
+	return globalTransaction, nil
+}
+
 func (s *StateLogStore) RecordStateMachineRestarted(ctx context.Context, machineInstance statelang.StateMachineInstance,
 	context core.ProcessContext) error {
 	if machineInstance == nil {
@@ -211,18 +333,25 @@ func (s *StateLogStore) RecordStateStarted(ctx context.Context, stateInstance st
 	if stateInstance == nil {
 		return nil
 	}
-	isUpdateMode := s.isUpdateMode(stateInstance, context)
+	isUpdateMode, err := s.isUpdateMode(stateInstance, context)
+	if err != nil {
+		return err
+	}
 
+	// if this state is for retry, do not register branch
 	if stateInstance.StateIDRetriedFor() != "" {
 		if isUpdateMode {
 			stateInstance.SetID(stateInstance.StateIDRetriedFor())
 		} else {
+			// generate id by default
 			stateInstance.SetID(s.generateRetryStateInstanceId(stateInstance))
 		}
 	} else if stateInstance.StateIDCompensatedFor() != "" {
+		// if this state is for compensation, do not register branch
 		stateInstance.SetID(s.generateCompensateStateInstanceId(stateInstance, isUpdateMode))
 	} else {
-		//TODO register branch
+		// register branch
+		s.branchRegister(stateInstance, context)
 	}
 
 	if stateInstance.ID() == "" && s.seqGenerator != nil {
@@ -252,9 +381,45 @@ func (s *StateLogStore) RecordStateStarted(ctx context.Context, stateInstance st
 	return nil
 }
 
-func (s *StateLogStore) isUpdateMode(instance statelang.StateInstance, context core.ProcessContext) bool {
-	//TODO implement me, add forward logic
-	return false
+func (s *StateLogStore) isUpdateMode(stateInstance statelang.StateInstance, context core.ProcessContext) (bool, error) {
+	cfg, ok := context.GetVariable(constant.VarNameStateMachineConfig).(core.DefaultStateMachineConfig)
+	if !ok {
+		return false, errors.New("stateMachineConfig is required in context")
+	}
+
+	instruction, ok := context.GetInstruction().(*core.StateInstruction)
+	if !ok {
+		return false, errors.New("stateInstruction is required in processContext")
+	}
+	instructionState, err := instruction.GetState(context)
+	if err != nil {
+		return false, err
+	}
+	taskState, _ := instructionState.(*state.AbstractTaskState)
+	stateMachine := stateInstance.StateMachineInstance().StateMachine()
+
+	if stateInstance.StateIDRetriedFor() != "" {
+		if taskState != nil && taskState.RetryPersistModeUpdate() {
+			return taskState.RetryPersistModeUpdate(), nil
+		} else if stateMachine.IsRetryPersistModeUpdate() {
+			return stateMachine.IsRetryPersistModeUpdate(), nil
+		}
+		return cfg.IsSagaRetryPersistModeUpdate(), nil
+	} else if stateInstance.StateIDCompensatedFor() != "" {
+		// find if this compensate has been executed
+		stateList := stateInstance.StateMachineInstance().StateList()
+		for _, instance := range stateList {
+			if instance.IsForCompensation() && instance.Name() == stateInstance.Name() {
+				if taskState != nil && taskState.CompensatePersistModeUpdate() {
+					return taskState.CompensatePersistModeUpdate(), nil
+				} else if stateMachine.IsCompensatePersistModeUpdate() {
+					return stateMachine.IsCompensatePersistModeUpdate(), nil
+				}
+				return cfg.IsSagaCompensatePersistModeUpdate(), nil
+			}
+		}
+	}
+	return false, nil
 }
 
 func (s *StateLogStore) generateRetryStateInstanceId(stateInstance statelang.StateInstance) string {
@@ -293,6 +458,49 @@ func (s *StateLogStore) generateCompensateStateInstanceId(stateInstance statelan
 	return fmt.Sprintf("%s-%d", originalCompensateStateInstId, maxIndex)
 }
 
+func (s *StateLogStore) branchRegister(stateInstance statelang.StateInstance, context core.ProcessContext) error {
+	cfg, ok := context.GetVariable(constant.VarNameStateMachineConfig).(core.DefaultStateMachineConfig)
+	if !ok {
+		return errors.New("stateMachineConfig is required in context")
+	}
+
+	if !cfg.IsSagaBranchRegisterEnable() {
+		log.Debugf("sagaBranchRegisterEnable = false, skip branch report. state[%s]", stateInstance.Name())
+		return nil
+	}
+
+	//Register branch
+	var err error
+	machineInstance := stateInstance.StateMachineInstance()
+	defer func() {
+		if err != nil {
+			log.Errorf("Branch transaction failure. StateMachine: %s, XID: %s, State: %s, stateId: %s, err: %v",
+				machineInstance.StateMachine().Name(), machineInstance.ID(), stateInstance.Name(), stateInstance.ID(), err)
+		}
+	}()
+
+	globalTransaction, err := s.getGlobalTransaction(machineInstance, context)
+	if err != nil {
+		return err
+	}
+	if globalTransaction == nil {
+		err = errors.New("Global transaction is not exists")
+		return err
+	}
+
+	branchId, err := rm.GetRMRemotingInstance().BranchRegister(rm.BranchRegisterParam{
+		BranchType: branch.BranchTypeSAGA,
+		ResourceId: machineInstance.StateMachine().Name() + "#" + stateInstance.Name(),
+		Xid:        globalTransaction.Xid,
+	})
+	if err != nil {
+		return err
+	}
+
+	stateInstance.SetID(strconv.FormatInt(branchId, 10))
+	return nil
+}
+
 func (s *StateLogStore) getIdIndex(stateInstanceId string, separator string) int {
 	if stateInstanceId != "" {
 		start := strings.LastIndex(stateInstanceId, separator)
@@ -332,11 +540,124 @@ func (s *StateLogStore) RecordStateFinished(ctx context.Context, stateInstance s
 		return err
 	}
 
-	//TODO report branch
+	// A switch to skip branch report on branch success, in order to optimize performance
+	cfg, ok := context.GetVariable(constant.VarNameStateMachineConfig).(core.DefaultStateMachineConfig)
+	if !(ok && !cfg.IsRmReportSuccessEnable() && statelang.SU == stateInstance.Status()) {
+		err = s.branchReport(stateInstance, context)
+		return err
+	}
+
 	return nil
 
 }
 
+func (s *StateLogStore) branchReport(stateInstance statelang.StateInstance, context core.ProcessContext) error {
+	cfg, ok := context.GetVariable(constant.VarNameStateMachineConfig).(core.DefaultStateMachineConfig)
+	if ok && !cfg.IsSagaBranchRegisterEnable() {
+		log.Debugf("sagaBranchRegisterEnable = false, skip branch report. state[%s]", stateInstance.Name())
+		return nil
+	}
+
+	var branchStatus branch.BranchStatus
+	// find out the original state instance, only the original state instance is registered on the server,
+	// and its status should be reported.
+	var originalStateInst statelang.StateInstance
+	if stateInstance.StateIDRetriedFor() != "" {
+		isUpdateMode, err := s.isUpdateMode(stateInstance, context)
+		if err != nil {
+			return err
+		}
+
+		if isUpdateMode {
+			originalStateInst = stateInstance
+		} else {
+			originalStateInst = s.findOutOriginalStateInstanceOfRetryState(stateInstance)
+		}
+
+		if statelang.SU == stateInstance.Status() {
+			branchStatus = branch.BranchStatusPhasetwoCommitted
+		} else if statelang.FA == stateInstance.Status() || statelang.UN == stateInstance.Status() {
+			branchStatus = branch.BranchStatusPhaseoneFailed
+		} else {
+			branchStatus = branch.BranchStatusUnknown
+		}
+	} else if stateInstance.StateIDCompensatedFor() != "" {
+		isUpdateMode, err := s.isUpdateMode(stateInstance, context)
+		if err != nil {
+			return err
+		}
+
+		if isUpdateMode {
+			originalStateInst = stateInstance.StateMachineInstance().StateMap()[stateInstance.StateIDCompensatedFor()]
+		} else {
+			originalStateInst = s.findOutOriginalStateInstanceOfCompensateState(stateInstance)
+		}
+	}
+
+	if originalStateInst == nil {
+		originalStateInst = stateInstance
+	}
+
+	if branchStatus == branch.BranchStatusUnknown {
+		if statelang.SU == originalStateInst.Status() && originalStateInst.CompensationStatus() == "" {
+			branchStatus = branch.BranchStatusPhasetwoCommitted
+		} else if statelang.SU == originalStateInst.CompensationStatus() {
+			branchStatus = branch.BranchStatusPhasetwoRollbacked
+		} else if statelang.FA == originalStateInst.CompensationStatus() || statelang.UN == originalStateInst.CompensationStatus() {
+			branchStatus = branch.BranchStatusPhasetwoRollbackFailedRetryable
+		} else if (statelang.FA == originalStateInst.Status() || statelang.UN == originalStateInst.Status()) && originalStateInst.CompensationStatus() == "" {
+			branchStatus = branch.BranchStatusPhaseoneFailed
+		} else {
+			branchStatus = branch.BranchStatusUnknown
+		}
+	}
+
+	var err error
+	defer func() {
+		if err != nil {
+			log.Errorf("Report branch status to server error:%s, StateMachine:%s, StateName:%s, XID:%s, branchId:%s, branchStatus:%s, err:%v",
+				err.Error(), originalStateInst.StateMachineInstance().StateMachine().Name(), originalStateInst.Name(),
+				originalStateInst.StateMachineInstance().ID(), originalStateInst.ID(), branchStatus, err)
+		}
+	}()
+
+	globalTransaction, err := s.getGlobalTransaction(stateInstance.StateMachineInstance(), context)
+	if err != nil {
+		return err
+	}
+	if globalTransaction == nil {
+		err = errors.New("Global transaction is not exists")
+		return err
+	}
+
+	branchId, err := strconv.ParseInt(originalStateInst.ID(), 10, 0)
+	err = rm.GetRMRemotingInstance().BranchReport(rm.BranchReportParam{
+		BranchType: branch.BranchTypeSAGA,
+		Xid:        globalTransaction.Xid,
+		BranchId:   branchId,
+		Status:     branchStatus,
+	})
+	return err
+}
+
+func (s *StateLogStore) findOutOriginalStateInstanceOfRetryState(stateInstance statelang.StateInstance) statelang.StateInstance {
+	stateMap := stateInstance.StateMachineInstance().StateMap()
+	originalStateInst := stateMap[stateInstance.StateIDRetriedFor()]
+	for originalStateInst.StateIDRetriedFor() != "" {
+		originalStateInst = stateMap[stateInstance.StateIDRetriedFor()]
+	}
+	return originalStateInst
+}
+
+func (s *StateLogStore) findOutOriginalStateInstanceOfCompensateState(stateInstance statelang.StateInstance) statelang.StateInstance {
+	stateMap := stateInstance.StateMachineInstance().StateMap()
+	originalStateInst := stateMap[stateInstance.StateIDCompensatedFor()]
+	for originalStateInst.StateIDRetriedFor() != "" {
+		originalStateInst = stateMap[stateInstance.StateIDRetriedFor()]
+	}
+	return originalStateInst
+}
+
 func (s *StateLogStore) GetStateMachineInstance(stateMachineInstanceId string) (statelang.StateMachineInstance, error) {
 	stateMachineInstance, err := SelectOne(s.db, s.getStateMachineInstanceByIdSql, scanRowsToStateMachineInstance,
 		stateMachineInstanceId)
@@ -437,7 +758,7 @@ func (s *StateLogStore) GetStateInstanceListByMachineInstanceId(stateMachineInst
 	if lastStateInstance.EndTime().IsZero() {
 		lastStateInstance.SetStatus(statelang.RU)
 	}
-	//TODO add forward and compensate logic
+
 	originStateMap := make(map[string]statelang.StateInstance)
 	compensatedStateMap := make(map[string]statelang.StateInstance)
 	retriedStateMap := make(map[string]statelang.StateInstance)
@@ -522,6 +843,11 @@ func (s *StateLogStore) SetSeqGenerator(seqGenerator sequence.SeqGenerator) {
 	s.seqGenerator = seqGenerator
 }
 
+func (s *StateLogStore) ClearUp(context core.ProcessContext) {
+	context.RemoveVariable(constant2.XidKey)
+	context.RemoveVariable(constant2.BranchTypeKey)
+}
+
 func execStateMachineInstanceStatementForInsert(obj statelang.StateMachineInstance, stmt *sql.Stmt) (int64, error) {
 	result, err := stmt.Exec(
 		obj.ID(),
diff --git a/pkg/saga/statemachine/store/repository/state_log_repository.go b/pkg/saga/statemachine/store/repository/state_log_repository.go
new file mode 100644
index 0000000..a9d970c
--- /dev/null
+++ b/pkg/saga/statemachine/store/repository/state_log_repository.go
@@ -0,0 +1,141 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements.  See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License.  You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+ */
+
+package repository
+
+import (
+	"context"
+	"database/sql"
+	"github.com/pkg/errors"
+
+	"github.com/seata/seata-go/pkg/saga/statemachine/engine/core"
+	"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
+	"github.com/seata/seata-go/pkg/saga/statemachine/store/db"
+)
+
+var (
+	stateLogRepositoryImpl *StateLogRepositoryImpl
+)
+
+type StateLogRepositoryImpl struct {
+	stateLogStore *db.StateLogStore
+}
+
+func NewStateLogRepositoryImpl(hsqldb *sql.DB, tablePrefix string) *StateLogRepositoryImpl {
+	if stateLogRepositoryImpl == nil {
+		stateLogRepositoryImpl = &StateLogRepositoryImpl{
+			stateLogStore: db.NewStateLogStore(hsqldb, tablePrefix),
+		}
+	}
+
+	return stateLogRepositoryImpl
+}
+
+func (s *StateLogRepositoryImpl) RecordStateMachineStarted(
+	ctx context.Context,
+	machineInstance statelang.StateMachineInstance,
+	processContext core.ProcessContext,
+) error {
+	if s.stateLogStore == nil {
+		return errors.New("stateLogStore is not initialized")
+	}
+	return s.stateLogStore.RecordStateMachineStarted(ctx, machineInstance, processContext)
+}
+
+func (s *StateLogRepositoryImpl) RecordStateMachineFinished(
+	ctx context.Context,
+	machineInstance statelang.StateMachineInstance,
+	processContext core.ProcessContext,
+) error {
+	if s.stateLogStore == nil {
+		return errors.New("stateLogStore is not initialized")
+	}
+	return s.stateLogStore.RecordStateMachineFinished(ctx, machineInstance, processContext)
+}
+
+func (s *StateLogRepositoryImpl) RecordStateMachineRestarted(
+	ctx context.Context,
+	machineInstance statelang.StateMachineInstance,
+	processContext core.ProcessContext,
+) error {
+	if s.stateLogStore == nil {
+		return errors.New("stateLogStore is not initialized")
+	}
+	return s.stateLogStore.RecordStateMachineRestarted(ctx, machineInstance, processContext)
+}
+
+func (s *StateLogRepositoryImpl) RecordStateStarted(
+	ctx context.Context,
+	stateInstance statelang.StateInstance,
+	processContext core.ProcessContext,
+) error {
+	if s.stateLogStore == nil {
+		return errors.New("stateLogStore is not initialized")
+	}
+	return s.stateLogStore.RecordStateStarted(ctx, stateInstance, processContext)
+}
+
+func (s *StateLogRepositoryImpl) RecordStateFinished(
+	ctx context.Context,
+	stateInstance statelang.StateInstance,
+	processContext core.ProcessContext,
+) error {
+	if s.stateLogStore == nil {
+		return errors.New("stateLogStore is not initialized")
+	}
+	return s.stateLogStore.RecordStateFinished(ctx, stateInstance, processContext)
+}
+
+func (s *StateLogRepositoryImpl) GetStateMachineInstance(stateMachineInstanceId string) (statelang.StateMachineInstance, error) {
+	if s.stateLogStore == nil {
+		return nil, errors.New("stateLogStore is not initialized")
+	}
+	return s.stateLogStore.GetStateMachineInstance(stateMachineInstanceId)
+}
+
+func (s *StateLogRepositoryImpl) GetStateMachineInstanceByBusinessKey(businessKey, tenantId string) (statelang.StateMachineInstance, error) {
+	if s.stateLogStore == nil {
+		return nil, errors.New("stateLogStore is not initialized")
+	}
+	return s.stateLogStore.GetStateMachineInstanceByBusinessKey(businessKey, tenantId)
+}
+
+func (s *StateLogRepositoryImpl) QueryStateMachineInstanceByParentId(parentId string) ([]statelang.StateMachineInstance, error) {
+	if s.stateLogStore == nil {
+		return nil, errors.New("stateLogStore is not initialized")
+	}
+	return s.stateLogStore.GetStateMachineInstanceByParentId(parentId)
+}
+
+func (s *StateLogRepositoryImpl) GetStateInstance(stateInstanceId, machineInstId string) (statelang.StateInstance, error) {
+	if s.stateLogStore == nil {
+		return nil, errors.New("stateLogStore is not initialized")
+	}
+	return s.stateLogStore.GetStateInstance(stateInstanceId, machineInstId)
+}
+
+func (s *StateLogRepositoryImpl) QueryStateInstanceListByMachineInstanceId(stateMachineInstanceId string) ([]statelang.StateInstance, error) {
+	if s.stateLogStore == nil {
+		return nil, errors.New("stateLogStore is not initialized")
+	}
+	return s.stateLogStore.GetStateInstanceListByMachineInstanceId(stateMachineInstanceId)
+
+}
+
+func (s *StateLogRepositoryImpl) SetStateLogStore(stateLogStore *db.StateLogStore) {
+	s.stateLogStore = stateLogStore
+}
diff --git a/pkg/saga/statemachine/store/repository/state_machine_repository.go b/pkg/saga/statemachine/store/repository/state_machine_repository.go
index ea37171..da31406 100644
--- a/pkg/saga/statemachine/store/repository/state_machine_repository.go
+++ b/pkg/saga/statemachine/store/repository/state_machine_repository.go
@@ -18,34 +18,221 @@
 package repository
 
 import (
-	"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
 	"io"
+	"sync"
+	"time"
+
+	"github.com/seata/seata-go/pkg/saga/statemachine/constant"
+	"github.com/seata/seata-go/pkg/saga/statemachine/engine/sequence"
+	"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
+	"github.com/seata/seata-go/pkg/saga/statemachine/statelang/parser"
+	"github.com/seata/seata-go/pkg/saga/statemachine/store/db"
+	"github.com/seata/seata-go/pkg/util/log"
+)
+
+const (
+	DefaultJsonParser = "fastjson"
+)
+
+var (
+	stateMachineRepositoryImpl     *StateMachineRepositoryImpl
+	onceStateMachineRepositoryImpl sync.Once
 )
 
 type StateMachineRepositoryImpl struct {
+	stateMachineMapById            map[string]statelang.StateMachine
+	stateMachineMapByNameAndTenant map[string]statelang.StateMachine
+
+	stateLangStore  *db.StateLangStore
+	seqGenerator    sequence.SeqGenerator
+	defaultTenantId string
+	jsonParserName  string
+	charset         string
+	mutex           *sync.Mutex
+}
+
+func GetStateMachineRepositoryImpl() *StateMachineRepositoryImpl {
+	if stateMachineRepositoryImpl == nil {
+		onceStateMachineRepositoryImpl.Do(func() {
+			//TODO get charset by config
+			//TODO charset is not use
+			//TODO using json parser
+			stateMachineRepositoryImpl = &StateMachineRepositoryImpl{
+				stateMachineMapById:            make(map[string]statelang.StateMachine),
+				stateMachineMapByNameAndTenant: make(map[string]statelang.StateMachine),
+				seqGenerator:                   sequence.NewUUIDSeqGenerator(),
+				jsonParserName:                 DefaultJsonParser,
+				charset:                        "UTF-8",
+				mutex:                          &sync.Mutex{},
+			}
+		})
+	}
+
+	return stateMachineRepositoryImpl
+}
+
+func (s *StateMachineRepositoryImpl) GetStateMachineById(stateMachineId string) (statelang.StateMachine, error) {
+	stateMachine := s.stateMachineMapById[stateMachineId]
+	if stateMachine == nil && s.stateLangStore != nil {
+		s.mutex.Lock()
+		defer s.mutex.Unlock()
+
+		stateMachine = s.stateMachineMapById[stateMachineId]
+		if stateMachine == nil {
+			oldStateMachine, err := s.stateLangStore.GetStateMachineById(stateMachineId)
+			if err != nil {
+				return oldStateMachine, err
+			}
+
+			parseStatMachine, err := parser.NewJSONStateMachineParser().Parse(oldStateMachine.Content())
+			if err != nil {
+				return oldStateMachine, err
+			}
+
+			oldStateMachine.SetStartState(parseStatMachine.StartState())
+			for key, val := range parseStatMachine.States() {
+				oldStateMachine.States()[key] = val
+			}
+
+			s.stateMachineMapById[stateMachineId] = oldStateMachine
+			s.stateMachineMapByNameAndTenant[oldStateMachine.Name()+"_"+oldStateMachine.TenantId()] = oldStateMachine
+			return oldStateMachine, nil
+		}
+	}
+	return stateMachine, nil
+}
+
+func (s *StateMachineRepositoryImpl) GetStateMachineByNameAndTenantId(stateMachineName string, tenantId string) (statelang.StateMachine, error) {
+	return s.GetLastVersionStateMachine(stateMachineName, tenantId)
+}
+
+func (s *StateMachineRepositoryImpl) GetLastVersionStateMachine(stateMachineName string, tenantId string) (statelang.StateMachine, error) {
+	key := stateMachineName + "_" + tenantId
+	stateMachine := s.stateMachineMapByNameAndTenant[key]
+	if stateMachine == nil && s.stateLangStore != nil {
+		s.mutex.Lock()
+		defer s.mutex.Unlock()
+
+		stateMachine = s.stateMachineMapById[key]
+		if stateMachine == nil {
+			oldStateMachine, err := s.stateLangStore.GetLastVersionStateMachine(stateMachineName, tenantId)
+			if err != nil {
+				return oldStateMachine, err
+			}
+
+			parseStatMachine, err := parser.NewJSONStateMachineParser().Parse(oldStateMachine.Content())
+			if err != nil {
+				return oldStateMachine, err
+			}
+
+			oldStateMachine.SetStartState(parseStatMachine.StartState())
+			for key, val := range parseStatMachine.States() {
+				oldStateMachine.States()[key] = val
+			}
+
+			s.stateMachineMapById[oldStateMachine.ID()] = oldStateMachine
+			s.stateMachineMapByNameAndTenant[key] = oldStateMachine
+			return oldStateMachine, nil
+		}
+	}
+	return stateMachine, nil
+}
+
+func (s *StateMachineRepositoryImpl) RegistryStateMachine(machine statelang.StateMachine) error {
+	stateMachineName := machine.Name()
+	tenantId := machine.TenantId()
+
+	if s.stateLangStore != nil {
+		oldStateMachine, err := s.stateLangStore.GetLastVersionStateMachine(stateMachineName, tenantId)
+		if err != nil {
+			return err
+		}
+
+		if oldStateMachine != nil {
+			if oldStateMachine.Content() == machine.Content() && machine.Version() != "" && machine.Version() == oldStateMachine.Version() {
+				log.Debugf("StateMachine[%s] is already exist a same version", stateMachineName)
+				machine.SetID(oldStateMachine.ID())
+				machine.SetCreateTime(oldStateMachine.CreateTime())
+
+				s.stateMachineMapById[machine.ID()] = machine
+				s.stateMachineMapByNameAndTenant[machine.Name()+"_"+machine.TenantId()] = machine
+				return nil
+			}
+		}
+
+		if machine.ID() == "" {
+			machine.SetID(s.seqGenerator.GenerateId(constant.SeqEntityStateMachine, ""))
+		}
+
+		machine.SetCreateTime(time.Now())
+
+		err = s.stateLangStore.StoreStateMachine(machine)
+		if err != nil {
+			return err
+		}
+	}
+
+	if machine.ID() == "" {
+		machine.SetID(s.seqGenerator.GenerateId(constant.SeqEntityStateMachine, ""))
+	}
+
+	s.stateMachineMapById[machine.ID()] = machine
+	s.stateMachineMapByNameAndTenant[machine.Name()+"_"+machine.TenantId()] = machine
+	return nil
+}
+
+func (s *StateMachineRepositoryImpl) RegistryStateMachineByReader(reader io.Reader) error {
+	jsonByte, err := io.ReadAll(reader)
+	if err != nil {
+		return err
+	}
+
+	json := string(jsonByte)
+	parseStatMachine, err := parser.NewJSONStateMachineParser().Parse(json)
+	if err != nil {
+		return err
+	}
+
+	if parseStatMachine == nil {
+		return nil
+	}
+
+	parseStatMachine.SetContent(json)
+	s.RegistryStateMachine(parseStatMachine)
+
+	log.Debugf("===== StateMachine Loaded: %s", json)
+
+	return nil
+}
+
+func (s *StateMachineRepositoryImpl) SetStateLangStore(stateLangStore *db.StateLangStore) {
+	s.stateLangStore = stateLangStore
+}
+
+func (s *StateMachineRepositoryImpl) SetSeqGenerator(seqGenerator sequence.SeqGenerator) {
+	s.seqGenerator = seqGenerator
+}
+
+func (s *StateMachineRepositoryImpl) SetCharset(charset string) {
+	s.charset = charset
 }
 
-func (s StateMachineRepositoryImpl) GetStateMachineById(stateMachineId string) (statelang.StateMachine, error) {
-	//TODO implement me
-	panic("implement me")
+func (s *StateMachineRepositoryImpl) GetCharset() string {
+	return s.charset
 }
 
-func (s StateMachineRepositoryImpl) GetStateMachineByNameAndTenantId(stateMachineName string, tenantId string) (statelang.StateMachine, error) {
-	//TODO implement me
-	panic("implement me")
+func (s *StateMachineRepositoryImpl) SetDefaultTenantId(defaultTenantId string) {
+	s.defaultTenantId = defaultTenantId
 }
 
-func (s StateMachineRepositoryImpl) GetLastVersionStateMachine(stateMachineName string, tenantId string) (statelang.StateMachine, error) {
-	//TODO implement me
-	panic("implement me")
+func (s *StateMachineRepositoryImpl) GetDefaultTenantId() string {
+	return s.defaultTenantId
 }
 
-func (s StateMachineRepositoryImpl) RegistryStateMachine(machine statelang.StateMachine) error {
-	//TODO implement me
-	panic("implement me")
+func (s *StateMachineRepositoryImpl) SetJsonParserName(jsonParserName string) {
+	s.jsonParserName = jsonParserName
 }
 
-func (s StateMachineRepositoryImpl) RegistryStateMachineByReader(reader io.Reader) error {
-	//TODO implement me
-	panic("implement me")
+func (s *StateMachineRepositoryImpl) GetJsonParserName() string {
+	return s.jsonParserName
 }
diff --git a/pkg/tm/global_transaction.go b/pkg/tm/global_transaction.go
index 8ee554e..b8bb28c 100644
--- a/pkg/tm/global_transaction.go
+++ b/pkg/tm/global_transaction.go
@@ -151,3 +151,28 @@ func (g *GlobalTransactionManager) Rollback(ctx context.Context, gtr *GlobalTran
 
 	return nil
 }
+
+// GlobalReport Global report.
+func (g *GlobalTransactionManager) GlobalReport(ctx context.Context, gtr *GlobalTransaction) (message.GlobalStatus, error) {
+	if gtr.Xid == "" {
+		return message.GlobalStatusUnKnown, fmt.Errorf("GlobalReport xid should not be empty")
+	}
+
+	req := message.GlobalReportRequest{
+		AbstractGlobalEndRequest: message.AbstractGlobalEndRequest{
+			Xid: gtr.Xid,
+		},
+		GlobalStatus: gtr.TxStatus,
+	}
+	res, err := getty.GetGettyRemotingClient().SendSyncRequest(req)
+	if err != nil {
+		log.Errorf("GlobalBeginRequest  error %v", err)
+		return message.GlobalStatusUnKnown, err
+	}
+	if res == nil || res.(message.GlobalReportResponse).ResultCode == message.ResultCodeFailed {
+		log.Errorf("GlobalReportRequest result is empty or result code is failed, res %v", res)
+		return message.GlobalStatusUnKnown, fmt.Errorf("GlobalReportRequest result is empty or result code is failed.")
+	}
+	log.Infof("GlobalReportRequest success, res %v", res)
+	return res.(message.GlobalReportResponse).GlobalStatus, nil
+}
diff --git a/pkg/util/errors/code.go b/pkg/util/errors/code.go
index 9b15e72..5d265df 100644
--- a/pkg/util/errors/code.go
+++ b/pkg/util/errors/code.go
@@ -111,4 +111,6 @@ const (
 	InvalidParameter
 	// OperationDenied Operation denied
 	OperationDenied
+	// ForwardInvalid Forward invalid
+	ForwardInvalid
 )



Test patch:
diff --git a/pkg/saga/statemachine/engine/invoker/http_invoker_test.go b/pkg/saga/statemachine/engine/invoker/http_invoker_test.go
new file mode 100644
index 0000000..062aac0
--- /dev/null
+++ b/pkg/saga/statemachine/engine/invoker/http_invoker_test.go
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package invoker
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/seata/seata-go/pkg/saga/statemachine/statelang/state"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestHTTPInvokerInvokeSucceedWithOutRetry(t *testing.T) {
+	// create test server
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var input []interface{}
+		err := json.NewDecoder(r.Body).Decode(&input)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte(input[0].(string)))
+	}))
+	defer server.Close()
+
+	// create HTTP Invoker
+	invoker := NewHTTPInvoker()
+	client := NewHTTPClient("test", server.URL+"/", &http.Client{})
+	invoker.RegisterClient("test", client)
+
+	// invoke
+	ctx := context.Background()
+	values, err := invoker.Invoke(ctx, []any{"hello"}, newHTTPServiceTaskState())
+
+	// verify
+	assert.NoError(t, err)
+	assert.NotNil(t, values)
+	assert.Equal(t, "hello", values[0].Interface())
+}
+
+func TestHTTPInvokerInvokeWithRetry(t *testing.T) {
+	attemptCount := 0
+	// create test server
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		attemptCount++
+		if attemptCount < 2 {
+			w.WriteHeader(http.StatusInternalServerError)
+			w.Write([]byte("fail"))
+			return
+		}
+		var input []interface{}
+		json.NewDecoder(r.Body).Decode(&input)
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte(input[0].(string)))
+	}))
+	defer server.Close()
+
+	// create HTTP Invoker
+	invoker := NewHTTPInvoker()
+	client := NewHTTPClient("test", server.URL+"/", &http.Client{})
+	invoker.RegisterClient("test", client)
+
+	// invoker
+	ctx := context.Background()
+	values, err := invoker.Invoke(ctx, []any{"hello"}, newHTTPServiceTaskStateWithRetry())
+
+	// verify
+	assert.NoError(t, err)
+	assert.NotNil(t, values)
+	assert.Equal(t, "hello", values[0].Interface())
+	assert.Equal(t, 2, attemptCount)
+}
+
+func TestHTTPInvokerInvokeFailedInRetry(t *testing.T) {
+	// create test server
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+		w.Write([]byte("fail"))
+	}))
+	defer server.Close()
+
+	// create HTTP Invoker
+	invoker := NewHTTPInvoker()
+	client := NewHTTPClient("test", server.URL+"/", &http.Client{})
+	invoker.RegisterClient("test", client)
+
+	// invoker
+	ctx := context.Background()
+	_, err := invoker.Invoke(ctx, []any{"hello"}, newHTTPServiceTaskStateWithRetry())
+
+	// verify
+	assert.Error(t, err)
+}
+
+func TestHTTPInvokerAsyncInvoke(t *testing.T) {
+	called := false
+	// create test server
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		called = true
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte("success"))
+	}))
+	defer server.Close()
+
+	// create HTTP Invoker
+	invoker := NewHTTPInvoker()
+	client := NewHTTPClient("test", server.URL+"/", &http.Client{})
+	invoker.RegisterClient("test", client)
+
+	// async invoke
+	ctx := context.Background()
+	taskState := newHTTPServiceTaskStateWithAsync()
+	_, err := invoker.Invoke(ctx, []any{"hello"}, taskState)
+
+	// verify
+	assert.NoError(t, err)
+	time.Sleep(100 * time.Millisecond)
+	assert.True(t, called)
+}
+
+func newHTTPServiceTaskState() state.ServiceTaskState {
+	serviceTaskStateImpl := state.NewServiceTaskStateImpl()
+	serviceTaskStateImpl.SetName("test")
+	serviceTaskStateImpl.SetIsAsync(false)
+	serviceTaskStateImpl.SetServiceName("test")
+	serviceTaskStateImpl.SetServiceType("HTTP")
+	serviceTaskStateImpl.SetServiceMethod("POST")
+	return serviceTaskStateImpl
+}
+
+func newHTTPServiceTaskStateWithAsync() state.ServiceTaskState {
+	serviceTaskStateImpl := state.NewServiceTaskStateImpl()
+	serviceTaskStateImpl.SetName("test")
+	serviceTaskStateImpl.SetIsAsync(true)
+	serviceTaskStateImpl.SetServiceName("test")
+	serviceTaskStateImpl.SetServiceType("HTTP")
+	serviceTaskStateImpl.SetServiceMethod("POST")
+	return serviceTaskStateImpl
+}
+
+func newHTTPServiceTaskStateWithRetry() state.ServiceTaskState {
+	serviceTaskStateImpl := state.NewServiceTaskStateImpl()
+	serviceTaskStateImpl.SetName("test")
+	serviceTaskStateImpl.SetIsAsync(false)
+	serviceTaskStateImpl.SetServiceName("test")
+	serviceTaskStateImpl.SetServiceType("HTTP")
+	serviceTaskStateImpl.SetServiceMethod("POST")
+
+	retryImpl := &state.RetryImpl{}
+	retryImpl.SetExceptions([]string{"fail"})
+	retryImpl.SetIntervalSecond(1)
+	retryImpl.SetMaxAttempt(3)
+	retryImpl.SetBackoffRate(0.9)
+	serviceTaskStateImpl.SetRetry([]state.Retry{retryImpl})
+	return serviceTaskStateImpl
+}
diff --git a/pkg/saga/statemachine/statelang/parser/statemachine_config_parser_test.go b/pkg/saga/statemachine/statelang/parser/statemachine_config_parser_test.go
index d901296..e634fc2 100644
--- a/pkg/saga/statemachine/statelang/parser/statemachine_config_parser_test.go
+++ b/pkg/saga/statemachine/statelang/parser/statemachine_config_parser_test.go
@@ -18,6 +18,7 @@
 package parser
 
 import (
+	"github.com/seata/seata-go/pkg/saga/statemachine"
 	"github.com/stretchr/testify/assert"
 	"testing"
 )
@@ -26,39 +27,39 @@ func TestStateMachineConfigParser_Parse(t *testing.T) {
 	parser := NewStateMachineConfigParser()
 
 	tests := []struct {
-		name                       string
-		configFilePath             string
-		expectedStateMachineObject *StateMachineObject
+		name           string
+		configFilePath string
+		expectedObject *statemachine.StateMachineObject
 	}{
 		{
-			name:                       "JSON Simple 1",
-			configFilePath:             "../../../../../testdata/saga/statelang/simple_statelang_with_choice.json",
-			expectedStateMachineObject: GetStateMachineObject1("json"),
+			name:           "JSON Simple 1",
+			configFilePath: "../../../../../testdata/saga/statelang/simple_statelang_with_choice.json",
+			expectedObject: GetStateMachineObject1("json"),
 		},
 		{
-			name:                       "JSON Simple 2",
-			configFilePath:             "../../../../../testdata/saga/statelang/simple_statemachine.json",
-			expectedStateMachineObject: GetStateMachineObject2("json"),
+			name:           "JSON Simple 2",
+			configFilePath: "../../../../../testdata/saga/statelang/simple_statemachine.json",
+			expectedObject: GetStateMachineObject2("json"),
 		},
 		{
-			name:                       "JSON Simple 3",
-			configFilePath:             "../../../../../testdata/saga/statelang/state_machine_new_designer.json",
-			expectedStateMachineObject: GetStateMachineObject3("json"),
+			name:           "JSON Simple 3",
+			configFilePath: "../../../../../testdata/saga/statelang/state_machine_new_designer.json",
+			expectedObject: GetStateMachineObject3("json"),
 		},
 		{
-			name:                       "YAML Simple 1",
-			configFilePath:             "../../../../../testdata/saga/statelang/simple_statelang_with_choice.yaml",
-			expectedStateMachineObject: GetStateMachineObject1("yaml"),
+			name:           "YAML Simple 1",
+			configFilePath: "../../../../../testdata/saga/statelang/simple_statelang_with_choice.yaml",
+			expectedObject: GetStateMachineObject1("yaml"),
 		},
 		{
-			name:                       "YAML Simple 2",
-			configFilePath:             "../../../../../testdata/saga/statelang/simple_statemachine.yaml",
-			expectedStateMachineObject: GetStateMachineObject2("yaml"),
+			name:           "YAML Simple 2",
+			configFilePath: "../../../../../testdata/saga/statelang/simple_statemachine.yaml",
+			expectedObject: GetStateMachineObject2("yaml"),
 		},
 		{
-			name:                       "YAML Simple 3",
-			configFilePath:             "../../../../../testdata/saga/statelang/state_machine_new_designer.yaml",
-			expectedStateMachineObject: GetStateMachineObject3("yaml"),
+			name:           "YAML Simple 3",
+			configFilePath: "../../../../../testdata/saga/statelang/state_machine_new_designer.yaml",
+			expectedObject: GetStateMachineObject3("yaml"),
 		},
 	}
 
@@ -72,18 +73,18 @@ func TestStateMachineConfigParser_Parse(t *testing.T) {
 			if err != nil {
 				t.Error("parse fail: " + err.Error())
 			}
-			assert.Equal(t, tt.expectedStateMachineObject, object)
+			assert.Equal(t, tt.expectedObject, object)
 		})
 	}
 }
 
-func GetStateMachineObject1(format string) *StateMachineObject {
+func GetStateMachineObject1(format string) *statemachine.StateMachineObject {
 	switch format {
 	case "json":
 	case "yaml":
 	}
 
-	return &StateMachineObject{
+	return &statemachine.StateMachineObject{
 		Name:       "simpleChoiceTestStateMachine",
 		Comment:    "带条件分支的测试状态机定义",
 		StartState: "FirstState",
@@ -123,7 +124,7 @@ func GetStateMachineObject1(format string) *StateMachineObject {
 	}
 }
 
-func GetStateMachineObject2(format string) *StateMachineObject {
+func GetStateMachineObject2(format string) *statemachine.StateMachineObject {
 	var retryMap map[string]interface{}
 
 	switch format {
@@ -147,7 +148,7 @@ func GetStateMachineObject2(format string) *StateMachineObject {
 		}
 	}
 
-	return &StateMachineObject{
+	return &statemachine.StateMachineObject{
 		Name:       "simpleTestStateMachine",
 		Comment:    "测试状态机定义",
 		StartState: "FirstState",
@@ -282,7 +283,7 @@ func GetStateMachineObject2(format string) *StateMachineObject {
 	}
 }
 
-func GetStateMachineObject3(format string) *StateMachineObject {
+func GetStateMachineObject3(format string) *statemachine.StateMachineObject {
 	var (
 		boundsMap1 map[string]interface{}
 		boundsMap2 map[string]interface{}
@@ -685,7 +686,7 @@ func GetStateMachineObject3(format string) *StateMachineObject {
 		}
 	}
 
-	return &StateMachineObject{
+	return &statemachine.StateMachineObject{
 		Name:                        "StateMachineNewDesigner",
 		Comment:                     "This state machine is modeled by designer tools.",
 		Version:                     "0.0.1",
diff --git a/pkg/saga/statemachine/statelang/parser/statemachine_json_parser_test.go b/pkg/saga/statemachine/statelang/parser/statemachine_json_parser_test.go
index 2cc8280..2150875 100644
--- a/pkg/saga/statemachine/statelang/parser/statemachine_json_parser_test.go
+++ b/pkg/saga/statemachine/statelang/parser/statemachine_json_parser_test.go
@@ -18,9 +18,18 @@
 package parser
 
 import (
+	"os"
 	"testing"
 )
 
+func readFileContent(filePath string) (string, error) {
+	content, err := os.ReadFile(filePath)
+	if err != nil {
+		return "", err
+	}
+	return string(content), nil
+}
+
 func TestParseChoice(t *testing.T) {
 	parser := NewJSONStateMachineParser()
 
@@ -40,7 +49,12 @@ func TestParseChoice(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			_, err := parser.Parse(tt.configFilePath)
+			content, err := readFileContent(tt.configFilePath)
+			if err != nil {
+				t.Error("read file fail: " + err.Error())
+				return
+			}
+			_, err = parser.Parse(content)
 			if err != nil {
 				t.Error("parse fail: " + err.Error())
 			}
@@ -67,7 +81,12 @@ func TestParseServiceTaskForSimpleStateMachine(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			_, err := parser.Parse(tt.configFilePath)
+			content, err := readFileContent(tt.configFilePath)
+			if err != nil {
+				t.Error("read file fail: " + err.Error())
+				return
+			}
+			_, err = parser.Parse(content)
 			if err != nil {
 				t.Error("parse fail: " + err.Error())
 			}
@@ -94,7 +113,12 @@ func TestParseServiceTaskForNewDesigner(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			_, err := parser.Parse(tt.configFilePath)
+			content, err := readFileContent(tt.configFilePath)
+			if err != nil {
+				t.Error("read file fail: " + err.Error())
+				return
+			}
+			_, err = parser.Parse(content)
 			if err != nil {
 				t.Error("parse fail: " + err.Error())
 			}
diff --git a/pkg/saga/statemachine/store/db/statelog_test.go b/pkg/saga/statemachine/store/db/statelog_test.go
index fb15b8c..1445f2e 100644
--- a/pkg/saga/statemachine/store/db/statelog_test.go
+++ b/pkg/saga/statemachine/store/db/statelog_test.go
@@ -57,6 +57,12 @@ func mockMachineInstance(stateMachineName string) statelang.StateMachineInstance
 	return inst
 }
 
+func mockStateMachineConfig(context core.ProcessContext) core.StateMachineConfig {
+	cfg := core.NewDefaultStateMachineConfig()
+	context.SetVariable(constant.VarNameStateMachineConfig, cfg)
+	return cfg
+}
+
 func TestStateLogStore_RecordStateMachineStarted(t *testing.T) {
 	prepareDB()
 
@@ -65,6 +71,7 @@ func TestStateLogStore_RecordStateMachineStarted(t *testing.T) {
 	expected := mockMachineInstance(stateMachineName)
 	expected.SetBusinessKey("test_started")
 	ctx := mockProcessContext(stateMachineName, expected)
+	mockStateMachineConfig(ctx)
 	err := stateLogStore.RecordStateMachineStarted(context.Background(), expected, ctx)
 	assert.Nil(t, err)
 	actual, err := stateLogStore.GetStateMachineInstance(expected.ID())
diff --git a/pkg/saga/statemachine/store/repository/state_machine_repository_test.go b/pkg/saga/statemachine/store/repository/state_machine_repository_test.go
new file mode 100644
index 0000000..cfd9a12
--- /dev/null
+++ b/pkg/saga/statemachine/store/repository/state_machine_repository_test.go
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package repository
+
+import (
+	"database/sql"
+	"github.com/seata/seata-go/pkg/saga/statemachine/statelang/parser"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
+	"github.com/seata/seata-go/pkg/saga/statemachine/store/db"
+)
+
+var (
+	oncePrepareDB sync.Once
+	testdb        *sql.DB
+)
+
+func prepareDB() {
+	oncePrepareDB.Do(func() {
+		var err error
+		testdb, err = sql.Open("sqlite3", ":memory:")
+		query_, err := os.ReadFile("../../../../../testdata/sql/saga/sqlite_init.sql")
+		initScript := string(query_)
+		if err != nil {
+			panic(err)
+		}
+		if _, err := testdb.Exec(initScript); err != nil {
+			panic(err)
+		}
+	})
+}
+
+func loadStateMachineByYaml() string {
+	query, _ := os.ReadFile("../../../../../testdata/saga/statelang/simple_statemachine.json")
+	return string(query)
+}
+
+func TestStateMachineInMemory(t *testing.T) {
+	const stateMachineId, stateMachineName, tenantId = "simpleStateMachine", "simpleStateMachine", "test"
+	stateMachine := statelang.NewStateMachineImpl()
+	stateMachine.SetID(stateMachineId)
+	stateMachine.SetName(stateMachineName)
+	stateMachine.SetTenantId(tenantId)
+	stateMachine.SetComment("This is a test state machine")
+	stateMachine.SetCreateTime(time.Now())
+
+	repository := GetStateMachineRepositoryImpl()
+
+	err := repository.RegistryStateMachine(stateMachine)
+	assert.Nil(t, err)
+
+	machineById, err := repository.GetStateMachineById(stateMachine.ID())
+	assert.Nil(t, err)
+	assert.Equal(t, stateMachine.Name(), machineById.Name())
+	assert.Equal(t, stateMachine.TenantId(), machineById.TenantId())
+	assert.Equal(t, stateMachine.Comment(), machineById.Comment())
+	assert.Equal(t, stateMachine.CreateTime().UnixNano(), machineById.CreateTime().UnixNano())
+
+	machineByNameAndTenantId, err := repository.GetLastVersionStateMachine(stateMachine.Name(), stateMachine.TenantId())
+	assert.Nil(t, err)
+	assert.Equal(t, stateMachine.ID(), machineByNameAndTenantId.ID())
+	assert.Equal(t, stateMachine.Comment(), machineById.Comment())
+	assert.Equal(t, stateMachine.CreateTime().UnixNano(), machineById.CreateTime().UnixNano())
+}
+
+func TestStateMachineInDb(t *testing.T) {
+	prepareDB()
+
+	const tenantId = "test"
+	yaml := loadStateMachineByYaml()
+	stateMachine, err := parser.NewJSONStateMachineParser().Parse(yaml)
+	assert.Nil(t, err)
+	stateMachine.SetTenantId(tenantId)
+	stateMachine.SetContent(yaml)
+
+	repository := GetStateMachineRepositoryImpl()
+	repository.SetStateLangStore(db.NewStateLangStore(testdb, "seata_"))
+
+	err = repository.RegistryStateMachine(stateMachine)
+	assert.Nil(t, err)
+
+	repository.stateMachineMapById[stateMachine.ID()] = nil
+	machineById, err := repository.GetStateMachineById(stateMachine.ID())
+	assert.Nil(t, err)
+	assert.Equal(t, stateMachine.Name(), machineById.Name())
+	assert.Equal(t, stateMachine.TenantId(), machineById.TenantId())
+	assert.Equal(t, stateMachine.Comment(), machineById.Comment())
+	assert.Equal(t, stateMachine.CreateTime().UnixNano(), machineById.CreateTime().UnixNano())
+
+	repository.stateMachineMapByNameAndTenant[stateMachine.Name()+"_"+stateMachine.TenantId()] = nil
+	machineByNameAndTenantId, err := repository.GetLastVersionStateMachine(stateMachine.Name(), stateMachine.TenantId())
+	assert.Nil(t, err)
+	assert.Equal(t, stateMachine.ID(), machineByNameAndTenantId.ID())
+	assert.Equal(t, stateMachine.Comment(), machineById.Comment())
+	assert.Equal(t, stateMachine.CreateTime().UnixNano(), machineById.CreateTime().UnixNano())
+}