Rollback (apache#358)

* feat: rollback

* feat: add at rollback sampel

* feat: fix name

pkg/datasource/sql/undo/builder/mysql_update_undo_log_builder.go

@@ -151,7 +151,7 @@ func (u *MySQLUpdateUndoLogBuilder) buildAfterImageSQL(beforeImage *types.Record
 	if OnlyCareUpdateColumns {
 		for _, row := range beforeImage.Rows {
 			for _, column := range row.Columns {
-				selectFields += column.Name + separator
+				selectFields += column.ColumnName + separator
 			}
 		}
 		selectFields = strings.TrimSuffix(selectFields, separator)

sample/at/gin/client/main.go

@@ -53,7 +53,8 @@ func main() {
 func updateData(ctx context.Context) (re error) {
 	request := gorequest.New()
 	log.Infof("branch transaction begin")
-	request.Post(serverIpPort+"/updateData").
+
+	request.Post(serverIpPort+"/updateDataSuccess").
 		Set(constant.XidKey, tm.GetXID(ctx)).
 		End(func(response gorequest.Response, body string, errs []error) {
 			if len(errs) != 0 {

sample/at/gin/server/main.go

@@ -38,9 +38,9 @@ func main() {
 
 	r.Use(ginmiddleware.TransactionMiddleware())
 
-	r.POST("/updateData", func(c *gin.Context) {
+	r.POST("/updateDataSuccess", func(c *gin.Context) {
 		log.Infof("get tm updateData")
-		if err := updateData(c); err != nil {
+		if err := updateDataSuccess(c); err != nil {
 			c.JSON(http.StatusOK, "updateData failure")
 			return
 		}

sample/at/gin/server/service.go

@@ -38,10 +38,9 @@ func initService() {
 	}
 }
 
-// case1 : tm commit success
-func updateData(ctx context.Context) error {
+func updateDataSuccess(ctx context.Context) error {
 	sql := "update order_tbl set descs=? where id=?"
-	ret, err := db.ExecContext(ctx, sql, fmt.Sprintf("NewDescs-%d", time.Now().UnixMilli()), 1)
+	ret, err := db.ExecContext(ctx, sql, fmt.Sprintf("NewDescs1-%d", time.Now().UnixMilli()), 1)
 	if err != nil {
 		fmt.Printf("update failed, err:%v\n", err)
 		return nil

sample/at/rollback/client/main.go

@@ -0,0 +1,75 @@
+/*
+ * 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 main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"time"
+
+	"github.com/parnurzeal/gorequest"
+
+	"github.com/seata/seata-go/pkg/client"
+	"github.com/seata/seata-go/pkg/constant"
+	"github.com/seata/seata-go/pkg/tm"
+	"github.com/seata/seata-go/pkg/util/log"
+)
+
+var serverIpPort = "http://127.0.0.1:8080"
+var serverIpPort2 = "http://127.0.0.1:8081"
+
+func main() {
+	flag.Parse()
+	client.Init()
+
+	bgCtx, cancel := context.WithTimeout(context.Background(), time.Minute*10)
+	defer cancel()
+
+	transInfo := &tm.TransactionInfo{
+		Name:    "ATSampleLocalGlobalTx",
+		TimeOut: time.Second * 30,
+	}
+
+	if err := tm.WithGlobalTx(bgCtx, transInfo, updateData); err != nil {
+		panic(fmt.Sprintf("tm update data err, %v", err))
+	}
+}
+
+func updateData(ctx context.Context) (re error) {
+	request := gorequest.New()
+	log.Infof("branch transaction begin")
+
+	request.Post(serverIpPort+"/updateDataSuccess").
+		Set(constant.XidKey, tm.GetXID(ctx)).
+		End(func(response gorequest.Response, body string, errs []error) {
+			if len(errs) != 0 {
+				re = errs[0]
+			}
+		})
+
+	request.Post(serverIpPort2+"/updateDataFail").
+		Set(constant.XidKey, tm.GetXID(ctx)).
+		End(func(response gorequest.Response, body string, errs []error) {
+			if len(errs) != 0 {
+				re = errs[0]
+			}
+		})
+
+	return
+}

sample/at/rollback/rollback.sql

@@ -0,0 +1,83 @@
+-- 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.
+
+CREATE database if NOT EXISTS `seata_client1` default character set utf8mb4 collate utf8mb4_unicode_ci;
+USE `seata_client1`;
+
+SET NAMES utf8mb4;
+SET FOREIGN_KEY_CHECKS = 0;
+
+CREATE TABLE IF NOT EXISTS  `order_tbl` (
+    `id` int(11) NOT NULL AUTO_INCREMENT,
+    `user_id` varchar(255) DEFAULT NULL,
+    `commodity_code` varchar(255) DEFAULT NULL,
+    `count` int(11) DEFAULT '0',
+    `money` int(11) DEFAULT '0',
+    `descs` varchar(255) DEFAULT '',
+    PRIMARY KEY (`id`)
+    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
+
+INSERT INTO `seata_client1`.`order_tbl` (`id`, `user_id`, `commodity_code`, `count`, `money`, `descs`) VALUES (1, 'NO-100001', 'C100000', 100, 10, 'init desc');
+
+DROP TABLE IF EXISTS `undo_log`;
+
+CREATE TABLE `undo_log` (
+                            `id` bigint NOT NULL AUTO_INCREMENT,
+                            `branch_id` bigint NOT NULL,
+                            `xid` varchar(100) NOT NULL,
+                            `context` varchar(128) NOT NULL,
+                            `rollback_info` longblob NOT NULL,
+                            `log_status` int NOT NULL,
+                            `log_created` datetime NOT NULL,
+                            `log_modified` datetime NOT NULL,
+                            `ext` varchar(100) DEFAULT NULL,
+                            PRIMARY KEY (`id`),
+                            KEY `idx_unionkey` (`xid`,`branch_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8;
+
+
+CREATE database if NOT EXISTS `seata_client` default character set utf8mb4 collate utf8mb4_unicode_ci;
+USE `seata_client`;
+
+SET NAMES utf8mb4;
+SET FOREIGN_KEY_CHECKS = 0;
+
+CREATE TABLE IF NOT EXISTS  `order_tbl` (
+    `id` int(11) NOT NULL AUTO_INCREMENT,
+    `user_id` varchar(255) DEFAULT NULL,
+    `commodity_code` varchar(255) DEFAULT NULL,
+    `count` int(11) DEFAULT '0',
+    `money` int(11) DEFAULT '0',
+    `descs` varchar(255) DEFAULT '',
+    PRIMARY KEY (`id`)
+    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
+
+INSERT INTO `seata_client`.`order_tbl` (`id`, `user_id`, `commodity_code`, `count`, `money`, `descs`) VALUES (1, 'NO-100001', 'C100000', 100, 10, 'init desc');
+
+DROP TABLE IF EXISTS `undo_log`;
+
+CREATE TABLE `undo_log` (
+                            `id` bigint NOT NULL AUTO_INCREMENT,
+                            `branch_id` bigint NOT NULL,
+                            `xid` varchar(100) NOT NULL,
+                            `context` varchar(128) NOT NULL,
+                            `rollback_info` longblob NOT NULL,
+                            `log_status` int NOT NULL,
+                            `log_created` datetime NOT NULL,
+                            `log_modified` datetime NOT NULL,
+                            `ext` varchar(100) DEFAULT NULL,
+                            PRIMARY KEY (`id`),
+                            KEY `idx_unionkey` (`xid`,`branch_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8;

sample/at/rollback/server/main.go

@@ -0,0 +1,53 @@
+/*
+ * 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 main
+
+import (
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/seata/seata-go/pkg/client"
+	ginmiddleware "github.com/seata/seata-go/pkg/integration/gin"
+	"github.com/seata/seata-go/pkg/util/log"
+)
+
+func main() {
+	client.Init()
+	initService()
+
+	r := gin.Default()
+
+	// NOTE: when use gin，must set ContextWithFallback true when gin version >= 1.8.1
+	// r.ContextWithFallback = true
+
+	r.Use(ginmiddleware.TransactionMiddleware())
+
+	r.POST("/updateDataSuccess", func(c *gin.Context) {
+		log.Infof("get tm updateData")
+		if err := updateDataSuccess(c); err != nil {
+			c.JSON(http.StatusOK, "updateData failure")
+			return
+		}
+		c.JSON(http.StatusOK, "updateData ok")
+	})
+
+	if err := r.Run(":8080"); err != nil {
+		log.Fatalf("start tcc server fatal: %v", err)
+	}
+}

sample/at/rollback/server/service.go

@@ -0,0 +1,56 @@
+/*
+ * 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 main
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+
+	sql2 "github.com/seata/seata-go/pkg/datasource/sql"
+)
+
+var (
+	db *sql.DB
+)
+
+func initService() {
+	var err error
+	db, err = sql.Open(sql2.SeataATMySQLDriver, "root:123456@tcp(127.0.0.1:3306)/seata_client?multiStatements=true&interpolateParams=true")
+	if err != nil {
+		panic("init service error")
+	}
+}
+
+func updateDataSuccess(ctx context.Context) error {
+	sql := "update order_tbl set descs=? where id=?"
+	ret, err := db.ExecContext(ctx, sql, fmt.Sprintf("NewDescs100-%d", time.Now().UnixMilli()), 1)
+	if err != nil {
+		fmt.Printf("update failed, err:%v\n", err)
+		return nil
+	}
+
+	rows, err := ret.RowsAffected()
+	if err != nil {
+		fmt.Printf("update failed, err:%v\n", err)
+		return nil
+	}
+	fmt.Printf("update success： %d.\n", rows)
+	return nil
+}

sample/at/rollback/server2/main.go

@@ -0,0 +1,53 @@
+/*
+ * 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 main
+
+import (
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/seata/seata-go/pkg/client"
+	ginmiddleware "github.com/seata/seata-go/pkg/integration/gin"
+	"github.com/seata/seata-go/pkg/util/log"
+)
+
+func main() {
+	client.Init()
+	initService()
+
+	r := gin.Default()
+
+	// NOTE: when use gin，must set ContextWithFallback true when gin version >= 1.8.1
+	// r.ContextWithFallback = true
+
+	r.Use(ginmiddleware.TransactionMiddleware())
+
+	r.POST("/updateDataFail", func(c *gin.Context) {
+		log.Infof("get tm updateData")
+		if err := updateDataFail(c); err != nil {
+			c.JSON(http.StatusOK, "updateData failure")
+			return
+		}
+		c.JSON(http.StatusOK, "updateData ok")
+	})
+
+	if err := r.Run(":8081"); err != nil {
+		log.Fatalf("start tcc server fatal: %v", err)
+	}
+}