You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by xi...@apache.org on 2026/08/30 04:05:20 UTC
(incubator-seata-go) branch master updated: fix: align RM undo log delete codec with Java wire format (#1135)
This is an automated email from the ASF dual-hosted git repository.
Ethan-Xingyue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new 6bd31c77 fix: align RM undo log delete codec with Java wire format (#1135)
6bd31c77 is described below
commit 6bd31c77a01de46421f6d4369108b8661f3960ee
Author: Cocoyu <we...@gmail.com>
AuthorDate: Sat Aug 29 21:05:16 2026 -0700
fix: align RM undo log delete codec with Java wire format (#1135)
* fix: support RM delete undo log request
* fix ci
* feat: realize undo_log deletion logic
* add license header
* feat: add DeleteBatchSize to undo log Config
* fix: align RM undo log delete codec with Java wire format
* fix: surface real undo log delete failures instead of swallowing
* test: cover end-to-end undo log delete path and broadcast-skip branch
* test: fix data race in TestDayValueYAML by not mutating time.Local
* fix(undo): Fix security vulnerability during undo log deletion process
* refactor(protocol): Simplify the decoding logic of UndoLogDeleteRequest
* fix(remoting): Fixed the RowsAffected error handling issue when deleting undo logs
* fix (rm): Fix parameter validation issue in undo log deletion handler
* fix (rm): Fix parameter validation issue in undo log deletion handler
* ci: rerun integration tests
* fix: gate undo log delete to MySQL, cap batch rounds per request, and pin Java wire format with golden bytes
---------
Co-authored-by: everfid-ever <16...@users.noreply.github.com>
Co-authored-by: ThunGuo <te...@apache.org>
Co-authored-by: EVERFID <30...@qq.com>
Co-authored-by: Ethan <xi...@apache.org>
---
pkg/datasource/sql/undo/config.go | 2 +
pkg/protocol/codec/codec.go | 1 +
pkg/protocol/codec/undolog_delete_req_codec.go | 63 ++++
.../codec/undolog_delete_req_codec_test.go | 334 +++++++++++++++++++
pkg/protocol/message/request_message.go | 2 +-
pkg/remoting/processor/client/init.go | 1 +
.../client/rm_delete_undolog_processor.go | 177 ++++++++++
.../client/rm_delete_undolog_processor_test.go | 368 +++++++++++++++++++++
8 files changed, 947 insertions(+), 1 deletion(-)
diff --git a/pkg/datasource/sql/undo/config.go b/pkg/datasource/sql/undo/config.go
index 3a3b1118..850feed3 100644
--- a/pkg/datasource/sql/undo/config.go
+++ b/pkg/datasource/sql/undo/config.go
@@ -43,6 +43,7 @@ type Config struct {
LogTable string `yaml:"log-table" json:"log-table,omitempty" koanf:"log-table"`
OnlyCareUpdateColumns bool `yaml:"only-care-update-columns" json:"only-care-update-columns,omitempty" koanf:"only-care-update-columns"`
CompressConfig CompressConfig `yaml:"compress" json:"compress,omitempty" koanf:"compress"`
+ DeleteBatchSize int `yaml:"delete-batch-size" json:"delete-batch-size,omitempty" koanf:"delete-batch-size"`
}
func (u *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
@@ -50,6 +51,7 @@ func (u *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
f.StringVar(&u.LogSerialization, prefix+".log-serialization", "json", "Serialization method.")
f.StringVar(&u.LogTable, prefix+".log-table", "undo_log", "undo log table name.")
f.BoolVar(&u.OnlyCareUpdateColumns, prefix+".only-care-update-columns", true, "The switch for degrade check.")
+ f.IntVar(&u.DeleteBatchSize, prefix+".delete-batch-size", 1000, "The batch size when deleting expired undo log.")
u.CompressConfig.RegisterFlagsWithPrefix(prefix+".compress", f)
}
diff --git a/pkg/protocol/codec/codec.go b/pkg/protocol/codec/codec.go
index 2f9b3564..16ae6aa5 100644
--- a/pkg/protocol/codec/codec.go
+++ b/pkg/protocol/codec/codec.go
@@ -144,6 +144,7 @@ func Init() {
// RM
GetCodecManager().RegisterCodec(CodecTypeSeata, &RegisterRMRequestCodec{})
GetCodecManager().RegisterCodec(CodecTypeSeata, &RegisterRMResponseCodec{})
+ GetCodecManager().RegisterCodec(CodecTypeSeata, &UndoLogDeleteRequestCodec{})
// TM
GetCodecManager().RegisterCodec(CodecTypeSeata, &RegisterTMRequestCodec{})
diff --git a/pkg/protocol/codec/undolog_delete_req_codec.go b/pkg/protocol/codec/undolog_delete_req_codec.go
new file mode 100644
index 00000000..21dbc242
--- /dev/null
+++ b/pkg/protocol/codec/undolog_delete_req_codec.go
@@ -0,0 +1,63 @@
+/*
+ * 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 codec
+
+import (
+ "seata.apache.org/seata-go/v2/pkg/protocol/branch"
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+ "seata.apache.org/seata-go/v2/pkg/util/bytes"
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+type UndoLogDeleteRequestCodec struct{}
+
+func (u *UndoLogDeleteRequestCodec) Decode(in []byte) interface{} {
+ data := message.UndoLogDeleteRequest{}
+ buf := bytes.NewByteBuffer(in)
+
+ branchType, err := buf.ReadByte()
+ if err != nil {
+ log.Errorf("failed to decode UndoLogDeleteRequest branchType: %v", err)
+ return nil
+ }
+ data.BranchType = branch.BranchType(branchType)
+ data.ResourceId = bytes.ReadString16Length(buf)
+ saveDays, err := buf.ReadUint16()
+ if err != nil {
+ log.Errorf("failed to decode UndoLogDeleteRequest saveDays: %v", err)
+ return nil
+ }
+ data.SaveDays = int16(saveDays)
+
+ return data
+}
+
+func (u *UndoLogDeleteRequestCodec) Encode(in interface{}) []byte {
+ data := in.(message.UndoLogDeleteRequest)
+ buf := bytes.NewByteBuffer([]byte{})
+
+ buf.WriteByte(byte(data.BranchType))
+ bytes.WriteString16Length(data.ResourceId, buf)
+ buf.WriteUint16(uint16(data.SaveDays))
+
+ return buf.Bytes()
+}
+
+func (u *UndoLogDeleteRequestCodec) GetMessageType() message.MessageType {
+ return message.MessageTypeRmDeleteUndolog
+}
diff --git a/pkg/protocol/codec/undolog_delete_req_codec_test.go b/pkg/protocol/codec/undolog_delete_req_codec_test.go
new file mode 100644
index 00000000..d483e870
--- /dev/null
+++ b/pkg/protocol/codec/undolog_delete_req_codec_test.go
@@ -0,0 +1,334 @@
+/*
+ * 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 codec
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "seata.apache.org/seata-go/v2/pkg/protocol/branch"
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+)
+
+func TestUndoLogDeleteRequestCodec_Encode(t *testing.T) {
+ c := &UndoLogDeleteRequestCodec{}
+
+ tests := []struct {
+ name string
+ req message.UndoLogDeleteRequest
+ }{
+ {
+ name: "normal case with resource id",
+ req: message.UndoLogDeleteRequest{
+ ResourceId: "jdbc:mysql://localhost:3306/seata",
+ SaveDays: 7,
+ BranchType: branch.BranchTypeAT,
+ },
+ },
+ {
+ name: "empty resource id",
+ req: message.UndoLogDeleteRequest{
+ ResourceId: "",
+ SaveDays: 30,
+ BranchType: branch.BranchTypeAT,
+ },
+ },
+ {
+ name: "max save days",
+ req: message.UndoLogDeleteRequest{
+ ResourceId: "test-resource",
+ SaveDays: 32767, // max int16
+ BranchType: branch.BranchTypeAT,
+ },
+ },
+ {
+ name: "min save days",
+ req: message.UndoLogDeleteRequest{
+ ResourceId: "test-resource",
+ SaveDays: -32768, // min int16
+ BranchType: branch.BranchTypeAT,
+ },
+ },
+ {
+ name: "zero save days",
+ req: message.UndoLogDeleteRequest{
+ ResourceId: "test-resource",
+ SaveDays: 0,
+ BranchType: branch.BranchTypeAT,
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ encoded := c.Encode(tt.req)
+ assert.NotNil(t, encoded)
+ assert.Greater(t, len(encoded), 0, "encoded data should not be empty")
+ })
+ }
+}
+
+func TestUndoLogDeleteRequestCodec_Decode(t *testing.T) {
+ c := &UndoLogDeleteRequestCodec{}
+
+ tests := []struct {
+ name string
+ req message.UndoLogDeleteRequest
+ validate func(t *testing.T, decoded message.UndoLogDeleteRequest)
+ }{
+ {
+ name: "decode normal request",
+ req: message.UndoLogDeleteRequest{
+ ResourceId: "jdbc:mysql://localhost:3306/seata",
+ SaveDays: 7,
+ BranchType: branch.BranchTypeAT,
+ },
+ validate: func(t *testing.T, decoded message.UndoLogDeleteRequest) {
+ assert.Equal(t, "jdbc:mysql://localhost:3306/seata", decoded.ResourceId)
+ assert.Equal(t, int16(7), decoded.SaveDays)
+ assert.Equal(t, branch.BranchTypeAT, decoded.BranchType)
+ },
+ },
+ {
+ name: "decode request with empty resource id",
+ req: message.UndoLogDeleteRequest{
+ ResourceId: "",
+ SaveDays: 15,
+ BranchType: branch.BranchTypeAT,
+ },
+ validate: func(t *testing.T, decoded message.UndoLogDeleteRequest) {
+ assert.Equal(t, "", decoded.ResourceId)
+ assert.Equal(t, int16(15), decoded.SaveDays)
+ assert.Equal(t, branch.BranchTypeAT, decoded.BranchType)
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ encoded := c.Encode(tt.req)
+ assert.NotNil(t, encoded)
+
+ decoded := c.Decode(encoded)
+ assert.NotNil(t, decoded)
+
+ decodedReq, ok := decoded.(message.UndoLogDeleteRequest)
+ assert.True(t, ok, "decoded result should be UndoLogDeleteRequest type")
+
+ tt.validate(t, decodedReq)
+ })
+ }
+}
+
+func TestUndoLogDeleteRequestCodec_EncodeDecode(t *testing.T) {
+ c := &UndoLogDeleteRequestCodec{}
+
+ tests := []struct {
+ name string
+ req message.UndoLogDeleteRequest
+ }{
+ {
+ name: "round trip - normal case",
+ req: message.UndoLogDeleteRequest{
+ ResourceId: "jdbc:mysql://127.0.0.1:3306/test_db",
+ SaveDays: 10,
+ BranchType: branch.BranchTypeAT,
+ },
+ },
+ {
+ name: "round trip - empty resource id",
+ req: message.UndoLogDeleteRequest{
+ ResourceId: "",
+ SaveDays: 5,
+ BranchType: branch.BranchTypeAT,
+ },
+ },
+ {
+ name: "round trip - negative save days",
+ req: message.UndoLogDeleteRequest{
+ ResourceId: "test-resource",
+ SaveDays: -1,
+ BranchType: branch.BranchTypeAT,
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ encoded := c.Encode(tt.req)
+ assert.NotNil(t, encoded)
+
+ decoded := c.Decode(encoded)
+ assert.NotNil(t, decoded)
+
+ decodedReq, ok := decoded.(message.UndoLogDeleteRequest)
+ assert.True(t, ok)
+
+ assert.Equal(t, tt.req.ResourceId, decodedReq.ResourceId, "ResourceId should match")
+ assert.Equal(t, tt.req.SaveDays, decodedReq.SaveDays, "SaveDays should match")
+ assert.Equal(t, tt.req.BranchType, decodedReq.BranchType, "BranchType should match")
+ })
+ }
+}
+
+func TestUndoLogDeleteRequestCodec_GetMessageType(t *testing.T) {
+ c := &UndoLogDeleteRequestCodec{}
+ assert.Equal(t, message.MessageTypeRmDeleteUndolog, c.GetMessageType(),
+ "message type should be MessageTypeRmDeleteUndolog (111)")
+}
+
+// Byte vectors hand-derived from Java UndoLogDeleteRequestCodec#encode:
+// branchType ordinal (1 byte) + resourceId (uint16-BE length + UTF-8) + saveDays (int16-BE).
+func TestUndoLogDeleteRequestCodec_JavaWireFormat(t *testing.T) {
+ c := &UndoLogDeleteRequestCodec{}
+
+ tests := []struct {
+ name string
+ in []byte
+ want message.UndoLogDeleteRequest
+ }{
+ {
+ name: "AT mysql resource 7 days",
+ in: []byte{
+ 0x00,
+ 0x00, 0x21,
+ 'j', 'd', 'b', 'c', ':', 'm', 'y', 's', 'q', 'l', ':', '/', '/',
+ '1', '2', '7', '.', '0', '.', '0', '.', '1', ':', '3', '3', '0', '6',
+ '/', 's', 'e', 'a', 't', 'a',
+ 0x00, 0x07,
+ },
+ want: message.UndoLogDeleteRequest{
+ BranchType: branch.BranchTypeAT,
+ ResourceId: "jdbc:mysql://127.0.0.1:3306/seata",
+ SaveDays: 7,
+ },
+ },
+ {
+ name: "AT empty resource id 30 days",
+ in: []byte{0x00, 0x00, 0x00, 0x00, 0x1e},
+ want: message.UndoLogDeleteRequest{
+ BranchType: branch.BranchTypeAT,
+ ResourceId: "",
+ SaveDays: 30,
+ },
+ },
+ {
+ name: "AT negative save days two complement",
+ in: []byte{0x00, 0x00, 0x02, 'p', 'g', 0xff, 0xff},
+ want: message.UndoLogDeleteRequest{
+ BranchType: branch.BranchTypeAT,
+ ResourceId: "pg",
+ SaveDays: -1,
+ },
+ },
+ {
+ name: "XA ordinal 3 14 days",
+ in: []byte{0x03, 0x00, 0x02, 'x', 'a', 0x00, 0x0e},
+ want: message.UndoLogDeleteRequest{
+ BranchType: branch.BranchTypeXA,
+ ResourceId: "xa",
+ SaveDays: 14,
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ req, ok := c.Decode(tt.in).(message.UndoLogDeleteRequest)
+ assert.True(t, ok, "decoded result should be UndoLogDeleteRequest type")
+ assert.Equal(t, tt.want, req)
+ assert.Equal(t, tt.in, c.Encode(tt.want), "encode must reproduce the Java bytes exactly")
+ })
+ }
+}
+
+func TestUndoLogDeleteRequestCodec_DecodeMalformed(t *testing.T) {
+ c := &UndoLogDeleteRequestCodec{}
+
+ tests := []struct {
+ name string
+ in []byte
+ }{
+ {name: "empty input", in: []byte{}},
+ {name: "branch type only", in: []byte{0x00}},
+ {name: "missing save days", in: []byte{0x00, 0x00, 0x01, 'a'}},
+ {name: "truncated resourceId (declares 10 bytes, only 3 provided)", in: []byte{0x00, 0x00, 0x0A, 'a', 'b', 'c'}},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ assert.Nil(t, c.Decode(tt.in), "malformed input should be rejected with nil")
+ })
+ }
+}
+
+func TestUndoLogDeleteRequestCodec_Integration(t *testing.T) {
+ Init()
+
+ cm := GetCodecManager()
+ c := cm.GetCodec(CodecTypeSeata, message.MessageTypeRmDeleteUndolog)
+ assert.NotNil(t, c, "UndoLogDeleteRequest codec should be registered")
+
+ req := message.UndoLogDeleteRequest{
+ ResourceId: "integration-test-resource",
+ SaveDays: 20,
+ BranchType: branch.BranchTypeAT,
+ }
+
+ encoded := cm.Encode(CodecTypeSeata, req)
+ assert.NotNil(t, encoded)
+ assert.Greater(t, len(encoded), 2, "encoded data should include type code (2 bytes) + body")
+
+ decoded := cm.Decode(CodecTypeSeata, encoded)
+ assert.NotNil(t, decoded)
+
+ decodedReq, ok := decoded.(message.UndoLogDeleteRequest)
+ assert.True(t, ok)
+ assert.Equal(t, req.ResourceId, decodedReq.ResourceId)
+ assert.Equal(t, req.SaveDays, decodedReq.SaveDays)
+ assert.Equal(t, req.BranchType, decodedReq.BranchType)
+}
+
+func BenchmarkUndoLogDeleteRequestCodec_Encode(b *testing.B) {
+ c := &UndoLogDeleteRequestCodec{}
+ req := message.UndoLogDeleteRequest{
+ ResourceId: "jdbc:mysql://localhost:3306/seata",
+ SaveDays: 7,
+ BranchType: branch.BranchTypeAT,
+ }
+
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ _ = c.Encode(req)
+ }
+}
+
+func BenchmarkUndoLogDeleteRequestCodec_Decode(b *testing.B) {
+ c := &UndoLogDeleteRequestCodec{}
+ req := message.UndoLogDeleteRequest{
+ ResourceId: "jdbc:mysql://localhost:3306/seata",
+ SaveDays: 7,
+ BranchType: branch.BranchTypeAT,
+ }
+ encoded := c.Encode(req)
+
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ _ = c.Decode(encoded)
+ }
+}
diff --git a/pkg/protocol/message/request_message.go b/pkg/protocol/message/request_message.go
index c0d4a135..733207e5 100644
--- a/pkg/protocol/message/request_message.go
+++ b/pkg/protocol/message/request_message.go
@@ -131,7 +131,7 @@ func (req GlobalRollbackRequest) GetTypeCode() MessageType {
type UndoLogDeleteRequest struct {
ResourceId string
- SaveDays MessageType
+ SaveDays int16
BranchType model2.BranchType
}
diff --git a/pkg/remoting/processor/client/init.go b/pkg/remoting/processor/client/init.go
index 60461ab2..ca6c55fb 100644
--- a/pkg/remoting/processor/client/init.go
+++ b/pkg/remoting/processor/client/init.go
@@ -23,4 +23,5 @@ func RegisterProcessor() {
initOnResponse()
initBranchCommit()
initBranchRollback()
+ initDeleteUndoLog()
}
diff --git a/pkg/remoting/processor/client/rm_delete_undolog_processor.go b/pkg/remoting/processor/client/rm_delete_undolog_processor.go
new file mode 100644
index 00000000..7becc209
--- /dev/null
+++ b/pkg/remoting/processor/client/rm_delete_undolog_processor.go
@@ -0,0 +1,177 @@
+/*
+ * 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 client
+
+import (
+ "context"
+ "database/sql"
+ "fmt"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/undo"
+ "seata.apache.org/seata-go/v2/pkg/protocol/branch"
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+ "seata.apache.org/seata-go/v2/pkg/remoting/getty"
+ "seata.apache.org/seata-go/v2/pkg/rm"
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+const (
+ defaultDeleteBatchSize = 1000
+ maxDeleteBatchRounds = 100
+)
+
+func initDeleteUndoLog() {
+ getty.GetGettyClientHandlerInstance().RegisterProcessor(
+ message.MessageTypeRmDeleteUndolog, &rmDeleteUndoLogProcessor{})
+}
+
+type rmDeleteUndoLogProcessor struct{}
+
+func (r *rmDeleteUndoLogProcessor) Process(ctx context.Context, rpcMessage message.RpcMessage) error {
+ req, ok := rpcMessage.Body.(message.UndoLogDeleteRequest)
+ if !ok {
+ err := fmt.Errorf("invalid message body type: %T, expected message.UndoLogDeleteRequest", rpcMessage.Body)
+ log.Errorf("rmDeleteUndoLogProcessor: %v", err)
+ return err
+ }
+
+ if req.BranchType != branch.BranchTypeAT {
+ log.Infof("skip undo log delete for non-AT branch type: %v", req.BranchType)
+ return nil
+ }
+
+ if req.SaveDays <= 0 {
+ log.Errorf("skip undo log delete, invalid saveDays: resourceId=%s, saveDays=%d", req.ResourceId, req.SaveDays)
+ return fmt.Errorf("invalid saveDays %d for resourceId %s", req.SaveDays, req.ResourceId)
+ }
+
+ log.Infof("received undo log delete request: resourceId=%s, saveDays=%d",
+ req.ResourceId, req.SaveDays)
+
+ if err := r.deleteExpiredUndoLog(ctx, req); err != nil {
+ log.Errorf("failed to delete expired undo log: resourceId=%s, saveDays=%d, err=%v",
+ req.ResourceId, req.SaveDays, err)
+ return err
+ }
+
+ return nil
+}
+
+type dbResource interface {
+ GetDB() *sql.DB
+ GetDbType() types.DBType
+}
+
+func (r *rmDeleteUndoLogProcessor) deleteExpiredUndoLog(ctx context.Context, req message.UndoLogDeleteRequest) error {
+ resMgr, err := safeGetResourceManager(req.BranchType)
+ if err != nil {
+ log.Infof("skip undo log delete, no AT resource manager: %v", err)
+ return nil
+ }
+
+ val, ok := resMgr.GetCachedResources().Load(req.ResourceId)
+ if !ok {
+ log.Infof("skip undo log delete, resource not managed by this client: %s", req.ResourceId)
+ return nil
+ }
+
+ res, ok := val.(dbResource)
+ if !ok {
+ log.Warnf("skip undo log delete, resource does not implement dbResource: %s", req.ResourceId)
+ return nil
+ }
+
+ if res.GetDbType() != types.DBTypeMySQL {
+ log.Warnf("skip undo log delete, unsupported dbType %v, only MySQL supported: resourceId=%s",
+ res.GetDbType(), req.ResourceId)
+ return nil
+ }
+
+ conn, err := res.GetDB().Conn(ctx)
+ if err != nil {
+ return fmt.Errorf("get conn: %w", err)
+ }
+ defer conn.Close()
+
+ undoMgr, err := undo.GetUndoLogManager(res.GetDbType())
+ if err != nil {
+ return fmt.Errorf("get undo log manager for dbType %v: %w", res.GetDbType(), err)
+ }
+
+ exists, err := undoMgr.HasUndoLogTable(ctx, conn)
+ if err != nil {
+ return fmt.Errorf("check undo log table: %w", err)
+ }
+ if !exists {
+ log.Infof("undo_log table not exist, skip: resourceId=%s", req.ResourceId)
+ return nil
+ }
+
+ before := time.Now().AddDate(0, 0, -int(req.SaveDays))
+ return r.batchDeleteByLogCreated(ctx, conn, before)
+}
+
+func (r *rmDeleteUndoLogProcessor) batchDeleteByLogCreated(ctx context.Context, conn *sql.Conn, before time.Time) error {
+ undoLogTable := undo.UndoConfig.LogTable
+ if undoLogTable == "" {
+ undoLogTable = "undo_log"
+ }
+
+ batchSize := undo.UndoConfig.DeleteBatchSize
+ if batchSize <= 0 {
+ batchSize = defaultDeleteBatchSize
+ }
+
+ deleteSQL := fmt.Sprintf("DELETE FROM %s WHERE log_created <= ? LIMIT %d", undoLogTable, batchSize)
+
+ totalAffected := int64(0)
+ for round := 1; ; round++ {
+ result, err := conn.ExecContext(ctx, deleteSQL, before)
+ if err != nil {
+ return fmt.Errorf("exec delete: %w", err)
+ }
+ affected, err := result.RowsAffected()
+ if err != nil {
+ return fmt.Errorf("get rows affected: %w", err)
+ }
+ totalAffected += affected
+ if affected < int64(batchSize) {
+ break
+ }
+ if round >= maxDeleteBatchRounds {
+ log.Warnf("undo log delete stopped at round limit %d, leftover rows wait for the next request: before=%v, totalAffected=%d",
+ maxDeleteBatchRounds, before, totalAffected)
+ return nil
+ }
+ }
+
+ log.Infof("deleted expired undo log: before=%v, totalAffected=%d", before, totalAffected)
+ return nil
+}
+
+func safeGetResourceManager(bt branch.BranchType) (mgr rm.ResourceManager, err error) {
+ defer func() {
+ if r := recover(); r != nil {
+ err = fmt.Errorf("resource manager not registered for branch type %v: %v", bt, r)
+ }
+ }()
+ mgr = rm.GetRmCacheInstance().GetResourceManager(bt)
+ return
+}
diff --git a/pkg/remoting/processor/client/rm_delete_undolog_processor_test.go b/pkg/remoting/processor/client/rm_delete_undolog_processor_test.go
new file mode 100644
index 00000000..bfffacdf
--- /dev/null
+++ b/pkg/remoting/processor/client/rm_delete_undolog_processor_test.go
@@ -0,0 +1,368 @@
+/*
+ * 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 client
+
+import (
+ "context"
+ "database/sql"
+ "database/sql/driver"
+ "errors"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/DATA-DOG/go-sqlmock"
+ "github.com/stretchr/testify/assert"
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/undo"
+ mysqlundo "seata.apache.org/seata-go/v2/pkg/datasource/sql/undo/mysql"
+ "seata.apache.org/seata-go/v2/pkg/protocol/branch"
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+ "seata.apache.org/seata-go/v2/pkg/rm"
+)
+
+func TestProcess_InvalidBodyType(t *testing.T) {
+ p := &rmDeleteUndoLogProcessor{}
+ err := p.Process(context.Background(), message.RpcMessage{
+ Body: "not a UndoLogDeleteRequest",
+ })
+ assert.Error(t, err)
+}
+
+func TestProcess_NonATBranchType_Skipped(t *testing.T) {
+ p := &rmDeleteUndoLogProcessor{}
+ for _, bt := range []branch.BranchType{branch.BranchTypeXA, branch.BranchTypeTCC, branch.BranchTypeSAGA} {
+ err := p.Process(context.Background(), message.RpcMessage{
+ Body: message.UndoLogDeleteRequest{
+ ResourceId: "any-resource",
+ SaveDays: 7,
+ BranchType: bt,
+ },
+ })
+ assert.NoError(t, err, "branch type %v should be skipped silently", bt)
+ }
+}
+
+func TestProcess_ResourceNotFound_ReturnsNil(t *testing.T) {
+ p := &rmDeleteUndoLogProcessor{}
+ err := p.Process(context.Background(), message.RpcMessage{
+ Body: message.UndoLogDeleteRequest{
+ ResourceId: "jdbc:mysql://not-registered:3306/db",
+ SaveDays: 7,
+ BranchType: branch.BranchTypeAT,
+ },
+ })
+ // no AT resource manager / resource not managed by this client is a normal
+ // skip (the request is broadcast), so Process returns nil rather than an error
+ assert.NoError(t, err, "unmanaged resource should be skipped, not treated as an error")
+}
+
+type mockDBResource struct {
+ db *sql.DB
+ dbType types.DBType
+}
+
+func (m *mockDBResource) GetDB() *sql.DB { return m.db }
+func (m *mockDBResource) GetDbType() types.DBType { return m.dbType }
+
+// fakeATResourceManager is a minimal AT resource manager that only exposes a cached
+// resource, so the processor can reach the delete path in tests.
+type fakeATResourceManager struct {
+ rm.ResourceManager
+ cache *sync.Map
+}
+
+func (f *fakeATResourceManager) GetBranchType() branch.BranchType { return branch.BranchTypeAT }
+func (f *fakeATResourceManager) GetCachedResources() *sync.Map { return f.cache }
+
+// TestProcess_RealError_Propagates verifies that a real internal failure while
+// deleting (here: no undo log manager for the resource's db type) is surfaced by
+// Process instead of being swallowed, while unmanaged resources are still skipped.
+func TestProcess_RealError_Propagates(t *testing.T) {
+ undo.RegisterUndoLogManager(mysqlundo.NewUndoLogManager())
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ cache := &sync.Map{}
+ cache.Store("res-real-err", &mockDBResource{db: db, dbType: types.DBTypeMySQL})
+ rm.GetRmCacheInstance().RegisterResourceManager(&fakeATResourceManager{cache: cache})
+
+ mock.ExpectQuery("SELECT 1 FROM undo_log LIMIT 1").
+ WillReturnError(errors.New("check table failed"))
+
+ p := &rmDeleteUndoLogProcessor{}
+ err = p.Process(context.Background(), message.RpcMessage{
+ Body: message.UndoLogDeleteRequest{
+ ResourceId: "res-real-err",
+ SaveDays: 7,
+ BranchType: branch.BranchTypeAT,
+ },
+ })
+ assert.Error(t, err, "a real internal failure must propagate through Process")
+ assert.Contains(t, err.Error(), "check undo log table")
+ assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestProcess_EndToEnd_DeletesExpiredUndoLog(t *testing.T) {
+ undo.RegisterUndoLogManager(mysqlundo.NewUndoLogManager())
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ const resourceID = "jdbc:mysql://127.0.0.1:3306/seata"
+ cache := &sync.Map{}
+ cache.Store(resourceID, &mockDBResource{db: db, dbType: types.DBTypeMySQL})
+ rm.GetRmCacheInstance().RegisterResourceManager(&fakeATResourceManager{cache: cache})
+
+ mock.ExpectQuery("SELECT 1 FROM undo_log LIMIT 1").
+ WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
+ mock.ExpectExec("DELETE FROM undo_log WHERE log_created <= ?").
+ WithArgs(sqlmock.AnyArg()).
+ WillReturnResult(sqlmock.NewResult(0, defaultDeleteBatchSize))
+ mock.ExpectExec("DELETE FROM undo_log WHERE log_created <= ?").
+ WithArgs(sqlmock.AnyArg()).
+ WillReturnResult(sqlmock.NewResult(0, 10))
+
+ p := &rmDeleteUndoLogProcessor{}
+ err = p.Process(context.Background(), message.RpcMessage{
+ Body: message.UndoLogDeleteRequest{
+ ResourceId: resourceID,
+ SaveDays: 7,
+ BranchType: branch.BranchTypeAT,
+ },
+ })
+ assert.NoError(t, err, "end-to-end delete should succeed")
+ assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestProcess_NonPositiveSaveDays_RejectedAtEntry(t *testing.T) {
+ p := &rmDeleteUndoLogProcessor{}
+ for _, saveDays := range []int16{0, -1, -32768} {
+ err := p.Process(context.Background(), message.RpcMessage{
+ Body: message.UndoLogDeleteRequest{
+ ResourceId: "jdbc:mysql://127.0.0.1:3306/seata-savedays-guard",
+ SaveDays: saveDays,
+ BranchType: branch.BranchTypeAT,
+ },
+ })
+ assert.Error(t, err, "saveDays=%d must be rejected", saveDays)
+ }
+}
+
+func TestProcess_ResourceNotInCache_Skipped(t *testing.T) {
+ rm.GetRmCacheInstance().RegisterResourceManager(&fakeATResourceManager{cache: &sync.Map{}})
+
+ p := &rmDeleteUndoLogProcessor{}
+ err := p.Process(context.Background(), message.RpcMessage{
+ Body: message.UndoLogDeleteRequest{
+ ResourceId: "jdbc:mysql://another-client:3306/seata",
+ SaveDays: 7,
+ BranchType: branch.BranchTypeAT,
+ },
+ })
+ assert.NoError(t, err, "unmanaged resource on a broadcast request should be skipped, not error")
+}
+
+// TestProcess_UnsupportedDbType_Skipped verifies that a non-MySQL AT resource is
+// skipped before any SQL is issued: the DELETE ... LIMIT cleanup is a MySQL
+// dialect extension, and Postgres (which has a registered undo log manager, so
+// the earlier manager lookup would not stop it) rejects the statement on every
+// TC cleanup cycle. The processor must warn and skip instead.
+func TestProcess_UnsupportedDbType_Skipped(t *testing.T) {
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ const resourceID = "jdbc:postgresql://127.0.0.1:5432/seata"
+ cache := &sync.Map{}
+ cache.Store(resourceID, &mockDBResource{db: db, dbType: types.DBTypePostgreSQL})
+ rm.GetRmCacheInstance().RegisterResourceManager(&fakeATResourceManager{cache: cache})
+
+ p := &rmDeleteUndoLogProcessor{}
+ err = p.Process(context.Background(), message.RpcMessage{
+ Body: message.UndoLogDeleteRequest{
+ ResourceId: resourceID,
+ SaveDays: 7,
+ BranchType: branch.BranchTypeAT,
+ },
+ })
+ assert.NoError(t, err, "non-MySQL resource should be skipped, not error")
+ assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestBatchDeleteByLogCreated_DeletesRows(t *testing.T) {
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ before := time.Now().AddDate(0, 0, -7)
+
+ mock.ExpectExec("DELETE FROM undo_log WHERE log_created <= ?").
+ WithArgs(sqlmock.AnyArg()).
+ WillReturnResult(sqlmock.NewResult(0, 1000))
+ mock.ExpectExec("DELETE FROM undo_log WHERE log_created <= ?").
+ WithArgs(sqlmock.AnyArg()).
+ WillReturnResult(sqlmock.NewResult(0, 50))
+
+ conn, err := db.Conn(context.Background())
+ assert.NoError(t, err)
+ defer conn.Close()
+
+ p := &rmDeleteUndoLogProcessor{}
+ err = p.batchDeleteByLogCreated(context.Background(), conn, before)
+ assert.NoError(t, err)
+ assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestBatchDeleteByLogCreated_CustomTableName(t *testing.T) {
+ original := undo.UndoConfig.LogTable
+ undo.UndoConfig.LogTable = "my_undo_log"
+ defer func() { undo.UndoConfig.LogTable = original }()
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ mock.ExpectExec("DELETE FROM my_undo_log WHERE log_created <= ?").
+ WithArgs(sqlmock.AnyArg()).
+ WillReturnResult(sqlmock.NewResult(0, 0))
+
+ conn, err := db.Conn(context.Background())
+ assert.NoError(t, err)
+ defer conn.Close()
+
+ p := &rmDeleteUndoLogProcessor{}
+ err = p.batchDeleteByLogCreated(context.Background(), conn, time.Now())
+ assert.NoError(t, err)
+ assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestBatchDeleteByLogCreated_DBError(t *testing.T) {
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ mock.ExpectExec("DELETE FROM undo_log WHERE log_created <= ?").
+ WithArgs(sqlmock.AnyArg()).
+ WillReturnError(driver.ErrBadConn)
+
+ conn, err := db.Conn(context.Background())
+ assert.NoError(t, err)
+ defer conn.Close()
+
+ p := &rmDeleteUndoLogProcessor{}
+ err = p.batchDeleteByLogCreated(context.Background(), conn, time.Now())
+ assert.Error(t, err)
+}
+
+// TestBatchDeleteByLogCreated_RowsAffectedError verifies that an error from
+// RowsAffected() (e.g. driver returns -1 affected on a successful exec) is
+// surfaced instead of being swallowed. Before the fix, affected=-1 would
+// trigger an early `break` (since -1 < batchSize), silently leaving old
+// undo_log rows un-deleted while reporting success.
+func TestBatchDeleteByLogCreated_RowsAffectedError(t *testing.T) {
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ mock.ExpectExec("DELETE FROM undo_log WHERE log_created <= ?").
+ WithArgs(sqlmock.AnyArg()).
+ WillReturnResult(sqlmock.NewErrorResult(errors.New("driver does not support RowsAffected")))
+
+ conn, err := db.Conn(context.Background())
+ assert.NoError(t, err)
+ defer conn.Close()
+
+ p := &rmDeleteUndoLogProcessor{}
+ err = p.batchDeleteByLogCreated(context.Background(), conn, time.Now())
+ assert.Error(t, err, "RowsAffected error must propagate, not be swallowed")
+ assert.Contains(t, err.Error(), "rows affected")
+ assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func BenchmarkProcess_NonAT(b *testing.B) {
+ p := &rmDeleteUndoLogProcessor{}
+ msg := message.RpcMessage{
+ Body: message.UndoLogDeleteRequest{
+ ResourceId: "any",
+ SaveDays: 7,
+ BranchType: branch.BranchTypeXA,
+ },
+ }
+
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ _ = p.Process(context.Background(), msg)
+ }
+}
+
+func TestBatchDeleteByLogCreated_CustomBatchSize(t *testing.T) {
+ original := undo.UndoConfig.DeleteBatchSize
+ undo.UndoConfig.DeleteBatchSize = 2 // 小批次，方便验证循环
+ defer func() { undo.UndoConfig.DeleteBatchSize = original }()
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ // 第一批删了 2 行（等于 batchSize），继续
+ mock.ExpectExec("DELETE FROM undo_log WHERE log_created <= ?").
+ WithArgs(sqlmock.AnyArg()).
+ WillReturnResult(sqlmock.NewResult(0, 2))
+ // 第二批删了 1 行（小于 batchSize），退出
+ mock.ExpectExec("DELETE FROM undo_log WHERE log_created <= ?").
+ WithArgs(sqlmock.AnyArg()).
+ WillReturnResult(sqlmock.NewResult(0, 1))
+
+ conn, err := db.Conn(context.Background())
+ assert.NoError(t, err)
+ defer conn.Close()
+
+ p := &rmDeleteUndoLogProcessor{}
+ err = p.batchDeleteByLogCreated(context.Background(), conn, time.Now())
+ assert.NoError(t, err)
+ assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestBatchDeleteByLogCreated_RoundLimitStopsDraining(t *testing.T) {
+ original := undo.UndoConfig.DeleteBatchSize
+ undo.UndoConfig.DeleteBatchSize = 2
+ defer func() { undo.UndoConfig.DeleteBatchSize = original }()
+
+ db, mock, err := sqlmock.New()
+ assert.NoError(t, err)
+ defer db.Close()
+
+ for i := 0; i < maxDeleteBatchRounds; i++ {
+ mock.ExpectExec("DELETE FROM undo_log WHERE log_created <= ?").
+ WithArgs(sqlmock.AnyArg()).
+ WillReturnResult(sqlmock.NewResult(0, 2))
+ }
+
+ conn, err := db.Conn(context.Background())
+ assert.NoError(t, err)
+ defer conn.Close()
+
+ p := &rmDeleteUndoLogProcessor{}
+ err = p.batchDeleteByLogCreated(context.Background(), conn, time.Now())
+ assert.NoError(t, err)
+ assert.NoError(t, mock.ExpectationsWereMet())
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org