feature: add saga persistence layer

go.mod

@@ -33,6 +33,7 @@ require (
 require (
 	github.com/agiledragon/gomonkey v2.0.2+incompatible
 	github.com/agiledragon/gomonkey/v2 v2.9.0
+	github.com/mattn/go-sqlite3 v1.14.19
 )
 
 require (

go.sum

@@ -515,6 +515,8 @@ github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27k
 github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
 github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
+github.com/mattn/go-sqlite3 v1.14.19 h1:fhGleo2h1p8tVChob4I9HpmVFIAkKGpiukdrgQbWfGI=
+github.com/mattn/go-sqlite3 v1.14.19/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
 github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=

pkg/remoting/loadbalance/loadbalance.go

@@ -37,6 +37,8 @@ func Select(loadBalanceType string, sessions *sync.Map, xid string) getty.Sessio
 		return RandomLoadBalance(sessions, xid)
 	case xidLoadBalance:
 		return XidLoadBalance(sessions, xid)
+	case roundRobinLoadBalance:
+		return RoundRobinLoadBalance(sessions, xid)
 	default:
 		return RandomLoadBalance(sessions, xid)
 	}

pkg/remoting/loadbalance/round_robin_loadbalance.go

@@ -0,0 +1,69 @@
+/*
+ * 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 loadbalance
+
+import (
+	"math"
+	"sort"
+	"sync"
+	"sync/atomic"
+
+	getty "github.com/apache/dubbo-getty"
+)
+
+var sequence int32
+
+func RoundRobinLoadBalance(sessions *sync.Map, s string) getty.Session {
+	// collect sync.Map adderToSession
+	// filter out closed session instance
+	adderToSession := make(map[string]getty.Session, 0)
+	// map has no sequence, we should sort it to make sure the sequence is always the same
+	adders := make([]string, 0)
+	sessions.Range(func(key, value interface{}) bool {
+		session := key.(getty.Session)
+		if session.IsClosed() {
+			sessions.Delete(key)
+		} else {
+			adderToSession[session.RemoteAddr()] = session
+			adders = append(adders, session.RemoteAddr())
+		}
+		return true
+	})
+	sort.Strings(adders)
+	// adderToSession eq 0 means there are no available session
+	if len(adderToSession) == 0 {
+		return nil
+	}
+	index := getPositiveSequence() % len(adderToSession)
+	return adderToSession[adders[index]]
+}
+
+func getPositiveSequence() int {
+	for {
+		current := atomic.LoadInt32(&sequence)
+		var next int32
+		if current == math.MaxInt32 {
+			next = 0
+		} else {
+			next = current + 1
+		}
+		if atomic.CompareAndSwapInt32(&sequence, current, next) {
+			return int(current)
+		}
+	}
+}

pkg/remoting/loadbalance/round_robin_loadbalance_test.go

@@ -0,0 +1,100 @@
+/*
+ * 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 loadbalance
+
+import (
+	"fmt"
+	"math"
+	"sync"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/seata/seata-go/pkg/remoting/mock"
+)
+
+func TestRoundRobinLoadBalance_Normal(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	sessions := &sync.Map{}
+
+	for i := 0; i < 10; i++ {
+		session := mock.NewMockTestSession(ctrl)
+		session.EXPECT().IsClosed().Return(i == 2).AnyTimes()
+		session.EXPECT().RemoteAddr().Return(fmt.Sprintf("%d", i)).AnyTimes()
+		sessions.Store(session, fmt.Sprintf("session-%d", i+1))
+	}
+
+	for i := 0; i < 10; i++ {
+		if i == 2 {
+			continue
+		}
+		result := RoundRobinLoadBalance(sessions, "some_xid")
+		assert.Equal(t, fmt.Sprintf("%d", i), result.RemoteAddr())
+		assert.NotNil(t, result)
+		assert.False(t, result.IsClosed())
+	}
+}
+
+func TestRoundRobinLoadBalance_OverSequence(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	sessions := &sync.Map{}
+	sequence = math.MaxInt32
+
+	for i := 0; i < 10; i++ {
+		session := mock.NewMockTestSession(ctrl)
+		session.EXPECT().IsClosed().Return(false).AnyTimes()
+		session.EXPECT().RemoteAddr().Return(fmt.Sprintf("%d", i)).AnyTimes()
+		sessions.Store(session, fmt.Sprintf("session-%d", i+1))
+	}
+
+	for i := 0; i < 10; i++ {
+		// over sequence here
+		if i == 0 {
+			result := RoundRobinLoadBalance(sessions, "some_xid")
+			assert.Equal(t, "7", result.RemoteAddr())
+			assert.NotNil(t, result)
+			assert.False(t, result.IsClosed())
+			continue
+		}
+		result := RoundRobinLoadBalance(sessions, "some_xid")
+		assert.Equal(t, fmt.Sprintf("%d", i-1), result.RemoteAddr())
+		assert.NotNil(t, result)
+		assert.False(t, result.IsClosed())
+	}
+}
+
+func TestRoundRobinLoadBalance_All_Closed(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	sessions := &sync.Map{}
+	for i := 0; i < 10; i++ {
+		session := mock.NewMockTestSession(ctrl)
+		session.EXPECT().IsClosed().Return(true).AnyTimes()
+		sessions.Store(session, fmt.Sprintf("session-%d", i+1))
+	}
+	if result := RoundRobinLoadBalance(sessions, "some_xid"); result != nil {
+		t.Errorf("Expected nil, actual got %+v", result)
+	}
+}
+
+func TestRoundRobinLoadBalance_Empty(t *testing.T) {
+	sessions := &sync.Map{}
+	if result := RoundRobinLoadBalance(sessions, "some_xid"); result != nil {
+		t.Errorf("Expected nil, actual got %+v", result)
+	}
+}

pkg/saga/statemachine/constant/constant.go

@@ -13,6 +13,7 @@ const (
 	VarNameIsAsyncExecution             string = "_is_async_execution_"
 	VarNameStateInst                    string = "_current_state_instance_"
 	SeqEntityStateMachineInst           string = "STATE_MACHINE_INST"
+	SeqEntityStateInst                  string = "STATE_INST"
 	VarNameBusinesskey                  string = "_business_key_"
 	VarNameParentId                     string = "_parent_id_"
 	StateTypeServiceTask                string = "ServiceTask"

pkg/saga/statemachine/engine/process_ctrl/process_context.go

@@ -47,6 +47,12 @@ type ProcessContextImpl struct {
 	instruction Instruction
 }
 
+func NewProcessContextImpl() *ProcessContextImpl {
+	return &ProcessContextImpl{
+		mp: make(map[string]interface{}),
+	}
+}
+
 func (p *ProcessContextImpl) GetVariable(name string) interface{} {
 	p.mu.RLock()
 	defer p.mu.RUnlock()

pkg/saga/statemachine/engine/serializer/serializer.go

@@ -0,0 +1,42 @@
+package serializer
+
+import (
+	"bytes"
+	"encoding/gob"
+	"encoding/json"
+	"github.com/pkg/errors"
+)
+
+type ParamsSerializer struct{}
+
+func (ParamsSerializer) Serialize(object any) (string, error) {
+	result, err := json.Marshal(object)
+	return string(result), err
+}
+
+func (ParamsSerializer) Deserialize(object string) (any, error) {
+	var result any
+	err := json.Unmarshal([]byte(object), &result)
+	return result, err
+}
+
+type ErrorSerializer struct{}
+
+func (ErrorSerializer) Serialize(object error) ([]byte, error) {
+	var buffer bytes.Buffer
+	encoder := gob.NewEncoder(&buffer)
+	if object != nil {
+		err := encoder.Encode(object.Error())
+		return buffer.Bytes(), err
+	}
+	return nil, nil
+}
+
+func (ErrorSerializer) Deserialize(object []byte) (error, error) {
+	var errorMsg string
+	buffer := bytes.NewReader(object)
+	encoder := gob.NewDecoder(buffer)
+	err := encoder.Decode(&errorMsg)
+
+	return errors.New(errorMsg), err
+}

pkg/saga/statemachine/engine/serializer/serializer_test.go

@@ -0,0 +1,17 @@
+package serializer
+
+import (
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestErrorSerializer(t *testing.T) {
+	serializer := ErrorSerializer{}
+	expected := errors.New("This is a test error")
+	serialized, err := serializer.Serialize(expected)
+	assert.Nil(t, err)
+	actual, err := serializer.Deserialize(serialized)
+	assert.Nil(t, err)
+	assert.Equal(t, expected.Error(), actual.Error())
+}

pkg/saga/statemachine/engine/store/db/db.go

@@ -0,0 +1,102 @@
+package db
+
+import (
+	"database/sql"
+	"github.com/pkg/errors"
+	"github.com/seata/seata-go/pkg/util/log"
+)
+
+const TimeLayout = "2006-01-02 15:04:05.999999999-07:00"
+
+type ExecStatement[T any] func(obj T, stmt *sql.Stmt) (int64, error)
+
+type ScanRows[T any] func(rows *sql.Rows) (T, error)
+
+type Store struct {
+	db *sql.DB
+}
+
+func SelectOne[T any](db *sql.DB, sql string, fn ScanRows[T], args ...any) (T, error) {
+	var result T
+	log.Debugf("Preparing SQL: %s", sql)
+	stmt, err := db.Prepare(sql)
+	defer stmt.Close()
+	if err != nil {
+		return result, err
+	}
+
+	log.Debugf("setting params to Stmt: %v", args)
+	rows, err := stmt.Query(args...)
+	defer rows.Close()
+	if err != nil {
+		return result, nil
+	}
+
+	if rows.Next() {
+		return fn(rows)
+	}
+	return result, errors.New("no target selected")
+}
+
+func SelectList[T any](db *sql.DB, sql string, fn ScanRows[T], args ...any) ([]T, error) {
+	result := make([]T, 0)
+
+	log.Debugf("Preparing SQL: %s", sql)
+	stmt, err := db.Prepare(sql)
+	defer stmt.Close()
+	if err != nil {
+		return result, err
+	}
+
+	log.Debugf("setting params to Stmt: %v", args)
+	rows, err := stmt.Query(args...)
+	defer rows.Close()
+	if err != nil {
+		return result, err
+	}
+
+	for rows.Next() {
+		obj, err := fn(rows)
+		if err != nil {
+			return result, err
+		}
+		result = append(result, obj)
+	}
+
+	return result, nil
+}
+
+func ExecuteUpdate[T any](db *sql.DB, sql string, fn ExecStatement[T], obj T) (int64, error) {
+	log.Debugf("Preparing SQL: %s", sql)
+	stmt, err := db.Prepare(sql)
+	defer stmt.Close()
+	if err != nil {
+		return 0, err
+	}
+
+	log.Debugf("setting params to Stmt: %v", obj)
+
+	rowsAffected, err := fn(obj, stmt)
+	if err != nil {
+		return rowsAffected, err
+	}
+
+	return rowsAffected, nil
+}
+
+func ExecuteUpdateArgs(db *sql.DB, sql string, args ...any) (int64, error) {
+	log.Debugf("Preparing SQL: %s", sql)
+	stmt, err := db.Prepare(sql)
+	defer stmt.Close()
+	if err != nil {
+		return 0, err
+	}
+
+	log.Debugf("setting params to Stmt: %v", args)
+
+	result, err := stmt.Exec(args...)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}