Saga end-to-end verification

go.mod

@@ -36,7 +36,7 @@ require (
 	github.com/google/cel-go v0.18.0
 	github.com/mattn/go-sqlite3 v1.14.19
 	github.com/robertkrimen/otto v0.4.0
-	golang.org/x/sync v0.16.0
+	golang.org/x/sync v0.11.0
 	google.golang.org/protobuf v1.33.0
 	gopkg.in/yaml.v3 v3.0.1
 )
@@ -91,7 +91,7 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
 	golang.org/x/arch v0.3.0 // indirect
-	golang.org/x/text v0.27.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 	gopkg.in/sourcemap.v1 v1.0.5 // indirect
 )
@@ -108,7 +108,7 @@ require (
 	golang.org/x/crypto v0.17.0 // indirect
 	golang.org/x/exp v0.0.0-20220827204233-334a2380cb91 // indirect
 	golang.org/x/net v0.10.0 // indirect
-	golang.org/x/sys v0.32.0 // indirect
+	golang.org/x/sys v0.25.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20230803162519-f966b187b2e5 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230803162519-f966b187b2e5 // indirect
 	vimagination.zapto.org/memio v0.0.0-20200222190306-588ebc67b97d // indirect

go.sum

@@ -954,8 +954,8 @@ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw=
-golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
+golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -1033,8 +1033,8 @@ golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
-golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
+golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -1047,8 +1047,8 @@ golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
-golang.org/x/text v0.27.0 h1:4fGWRpyh641NLlecmyl4LOe6yDdfaYNrGb2zdfo4JV4=
-golang.org/x/text v0.27.0/go.mod h1:1D28KMCvyooCX9hBiosv5Tz/+YLxj0j7XhWjpSUF7CU=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=

goimports.sh

@@ -16,11 +16,11 @@
 #
 
 # format go imports style
-go install golang.org/x/tools/cmd/goimports
+go install golang.org/x/tools/cmd/goimports@v0.24.1
 goimports  -local github.com/seata/seata-go -w .
 
 # format licence style
-go install github.com/apache/skywalking-eyes/cmd/license-eye@latest
+go install github.com/apache/skywalking-eyes/cmd/license-eye@v0.6.0
 license-eye header fix
 # check dependency licence is valid
 license-eye dependency check

pkg/client/config.go

@@ -20,18 +20,20 @@ package client
 import (
 	"flag"
 	"fmt"
-	"github.com/seata/seata-go/pkg/saga"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"runtime"
 	"strings"
 
+	"github.com/seata/seata-go/pkg/saga"
+
 	"github.com/knadh/koanf"
 	"github.com/knadh/koanf/parsers/json"
 	"github.com/knadh/koanf/parsers/toml"
 	"github.com/knadh/koanf/parsers/yaml"
 	"github.com/knadh/koanf/providers/rawbytes"
+
 	"github.com/seata/seata-go/pkg/discovery"
 
 	"github.com/seata/seata-go/pkg/datasource/sql"

pkg/protocol/codec/codec.go

@@ -114,6 +114,7 @@ func (c *CodecManager) Encode(codecType CodecType, in interface{}) []byte {
 
 func Init() {
 	// Global
+	GetCodecManager().RegisterCodec(CodecTypeSeata, &GlobalReportRequestCodec{})
 	GetCodecManager().RegisterCodec(CodecTypeSeata, &GlobalReportResponseCodec{})
 	GetCodecManager().RegisterCodec(CodecTypeSeata, &GlobalBeginRequestCodec{})
 	GetCodecManager().RegisterCodec(CodecTypeSeata, &GlobalBeginResponseCodec{})

pkg/protocol/codec/global_report_request_codec.go

@@ -52,5 +52,6 @@ func (g *GlobalReportRequestCodec) Encode(in interface{}) []byte {
 
 // GetMessageType get global report request's message type
 func (g *GlobalReportRequestCodec) GetMessageType() message.MessageType {
-	return message.MessageTypeGlobalReportResult
+	// must be the request type, not the result type
+	return message.MessageTypeGlobalReport
 }

pkg/remoting/getty/rpc_client.go

@@ -25,6 +25,7 @@ import (
 
 	getty "github.com/apache/dubbo-getty"
 	gxsync "github.com/dubbogo/gost/sync"
+
 	"github.com/seata/seata-go/pkg/discovery"
 	"github.com/seata/seata-go/pkg/protocol/codec"
 	"github.com/seata/seata-go/pkg/remoting/config"

pkg/remoting/loadbalance/random_loadbalance_test.go

@@ -23,8 +23,9 @@ import (
 	"testing"
 
 	"github.com/golang/mock/gomock"
-	"github.com/seata/seata-go/pkg/remoting/mock"
 	"github.com/stretchr/testify/assert"
+
+	"github.com/seata/seata-go/pkg/remoting/mock"
 )
 
 func TestRandomLoadBalance_Normal(t *testing.T) {

pkg/remoting/loadbalance/xid_loadbalance_test.go

@@ -22,8 +22,9 @@ import (
 	"testing"
 
 	"github.com/golang/mock/gomock"
-	"github.com/seata/seata-go/pkg/remoting/mock"
 	"github.com/stretchr/testify/assert"
+
+	"github.com/seata/seata-go/pkg/remoting/mock"
 )
 
 func TestXidLoadBalance(t *testing.T) {

pkg/rm/init.go

@@ -30,3 +30,8 @@ type RmConfig struct {
 func InitRm(cfg RmConfig) {
 	rmConfig = cfg
 }
+
+// GetRmAppAndGroup returns current RM applicationId and txServiceGroup
+func GetRmAppAndGroup() (string, string) {
+	return rmConfig.ApplicationID, rmConfig.TxServiceGroup
+}

pkg/rm/rm_remoting.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/seata/seata-go/pkg/protocol/message"
 	"github.com/seata/seata-go/pkg/remoting/getty"
+	serrors "github.com/seata/seata-go/pkg/util/errors"
 	"github.com/seata/seata-go/pkg/util/log"
 )
 
@@ -84,7 +85,11 @@ func (r *RMRemoting) BranchReport(param BranchReportParam) error {
 	}
 
 	if err = isReportSuccess(resp); err != nil {
-		log.Errorf("BranchReport response error: %v, res %v", err.Error(), resp)
+		if seataErr, ok := err.(*serrors.SeataError); ok {
+			log.Warnf("BranchReport response error: code=%v msg=%s", seataErr.Code, seataErr.Message)
+		} else {
+			log.Errorf("BranchReport response error: %v, res %v", err.Error(), resp)
+		}
 		return err
 	}
 
@@ -156,7 +161,12 @@ func isRegisterSuccess(response interface{}) bool {
 func isReportSuccess(response interface{}) error {
 	if res, ok := response.(message.BranchReportResponse); ok {
 		if res.ResultCode == message.ResultCodeFailed {
-			return fmt.Errorf(res.Msg)
+			code := res.TransactionErrorCode
+			if code == serrors.TransactionErrorCodeBranchTransactionNotExist || int(code) == 120 {
+				log.Debugf("BranchReport received TransactionErrorCode %d (treated as BranchTransactionNotExist), ignoring", code)
+				return nil
+			}
+			return serrors.New(code, res.Msg, nil)
 		}
 	} else {
 		return ErrBranchReportResponseFault

pkg/saga/rm/saga_resource.go

@@ -19,6 +19,7 @@ package rm
 
 import (
 	"fmt"
+
 	"github.com/seata/seata-go/pkg/protocol/branch"
 )
 

pkg/saga/rm/state_machine_engine_holder.go

@@ -18,8 +18,9 @@
 package rm
 
 import (
-	"github.com/seata/seata-go/pkg/saga/statemachine/engine"
 	"sync"
+
+	"github.com/seata/seata-go/pkg/saga/statemachine/engine"
 )
 
 var (

pkg/saga/runtime/engine_registry.go

@@ -0,0 +1,50 @@
+/*
+ * 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 runtime
+
+import (
+	"context"
+	"sync"
+
+	"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
+)
+
+// Engine is a minimal facade for state machine operations needed by Saga RM
+type Engine interface {
+	Forward(ctx context.Context, stateMachineInstId string, replaceParams map[string]interface{}) (statelang.StateMachineInstance, error)
+	Compensate(ctx context.Context, stateMachineInstId string, replaceParams map[string]interface{}) (statelang.StateMachineInstance, error)
+	ReloadStateMachineInstance(ctx context.Context, instId string) (statelang.StateMachineInstance, error)
+}
+
+var (
+	mu        sync.RWMutex
+	engineRef Engine
+)
+
+func SetEngine(e Engine) {
+	mu.Lock()
+	engineRef = e
+	mu.Unlock()
+}
+
+func GetEngine() Engine {
+	mu.RLock()
+	e := engineRef
+	mu.RUnlock()
+	return e
+}

pkg/saga/statemachine/constant/constant.go

@@ -65,6 +65,7 @@ const (
 	VarNameCurrentCompensateTriggerState string = "_is_compensating_"
 	VarNameCurrentCompensationHolder     string = "_current_compensation_holder_"
 	VarNameFirstCompensationStateStarted string = "_first_compensation_state_started"
+	VarNameNoCompensation                string = "_no_compensation_case_"
 	VarNameCurrentLoopContextHolder      string = "_current_loop_context_holder_"
 	VarNameRetriedStateInstId            string = "_retried_state_instance_id"
 	VarNameIsForSubStatMachineForward    string = "_is_for_sub_statemachine_forward_"

pkg/saga/statemachine/engine/config/bootstrap.go

@@ -0,0 +1,62 @@
+/*
+ * 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 config
+
+import (
+	"fmt"
+
+	"github.com/seata/seata-go/pkg/saga/statemachine/constant"
+	"github.com/seata/seata-go/pkg/saga/statemachine/engine/pcext"
+	"github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
+	"github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl/handlers"
+	"github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl/process"
+)
+
+// bootstrapProcessWiring wires default handlers/routers and router handler into BusinessProcessor
+func (c *DefaultStateMachineConfig) bootstrapProcessWiring() error {
+	// Process handler for state machine
+	smHandler := pcext.NewStateMachineProcessHandler()
+	// Register ServiceTask handler by default
+	smHandler.RegistryStateHandler(constant.StateTypeServiceTask, handlers.NewServiceTaskStateHandler())
+
+	// Router for state machine
+	smRouter := &pcext.StateMachineProcessRouter{}
+	smRouter.InitDefaultStateRouters()
+
+	// Default router handler binds event publisher and process routers
+	drh := &process_ctrl.DefaultRouterHandler{}
+	drh.SetEventPublisher(c.EventPublisher())
+	drh.SetProcessRouters(map[string]process_ctrl.ProcessRouter{
+		string(process.StateLang): smRouter,
+	})
+
+	// Register into BusinessProcessor
+	pcImpl, ok := c.processController.(*process_ctrl.ProcessControllerImpl)
+	if !ok {
+		return fmt.Errorf("ProcessController is not an instance of ProcessControllerImpl")
+	}
+	bp := pcImpl.BusinessProcessor()
+	// need concrete DefaultBusinessProcessor to call Registry APIs
+	dbp, ok := bp.(*process_ctrl.DefaultBusinessProcessor)
+	if !ok {
+		return fmt.Errorf("BusinessProcessor is not DefaultBusinessProcessor, got %T", bp)
+	}
+	dbp.RegistryProcessHandler(process.StateLang, smHandler)
+	dbp.RegistryRouterHandler(process.StateLang, drh)
+	return nil
+}