thunguo commented on code in PR #1135: URL: https://github.com/apache/incubator-seata-go/pull/1135#discussion_r3542830207
########## pkg/remoting/processor/client/rm_delete_undolog_processor.go: ########## @@ -0,0 +1,161 @@ +/* + * 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 + +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 + } + + 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 { + // no AT resource manager on this client, nothing to clean up + log.Infof("skip undo log delete, no AT resource manager: %v", err) + return nil + } + + val, ok := resMgr.GetCachedResources().Load(req.ResourceId) + if !ok { + // resource not managed by this client (normal for a broadcast request) + log.Infof("skip undo log delete, resource not managed by this client: %s", req.ResourceId) + return nil + } + + res, ok := val.(dbResource) + if !ok { + return fmt.Errorf("resource %s does not implement dbResource interface", req.ResourceId) + } + + 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 + } + + 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) + } + + 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 { + result, err := conn.ExecContext(ctx, deleteSQL, before) + if err != nil { + return fmt.Errorf("exec delete: %w", err) + } + affected, _ := result.RowsAffected() + totalAffected += affected + if affected < int64(batchSize) { + break + } + } + + log.Infof("deleted expired undo log: before=%v, totalAffected=%d", before, totalAffected) + return nil +} + +func safeGetResourceManager(bt branch.BranchType) (mgr rm.ResourceManager, err error) { Review Comment: 这里为什么这么写？ ########## 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) Review Comment: 编码处理的位置不要静默断言失败，err需要显示的抛出来 ########## pkg/remoting/processor/client/rm_delete_undolog_processor.go: ########## @@ -0,0 +1,161 @@ +/* + * 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 + +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 + } + + 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 { + // no AT resource manager on this client, nothing to clean up + log.Infof("skip undo log delete, no AT resource manager: %v", err) + return nil + } + + val, ok := resMgr.GetCachedResources().Load(req.ResourceId) + if !ok { + // resource not managed by this client (normal for a broadcast request) + log.Infof("skip undo log delete, resource not managed by this client: %s", req.ResourceId) + return nil + } + + res, ok := val.(dbResource) + if !ok { + return fmt.Errorf("resource %s does not implement dbResource interface", req.ResourceId) + } + + 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 + } + + 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) + } + + 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 { + result, err := conn.ExecContext(ctx, deleteSQL, before) + if err != nil { + return fmt.Errorf("exec delete: %w", err) + } + affected, _ := result.RowsAffected() Review Comment: 这里这个error不应该忽略吧，确认下`RowsAffected()` error的时候affected是什么值 ########## pkg/remoting/processor/client/rm_delete_undolog_processor.go: ########## @@ -0,0 +1,161 @@ +/* + * 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 + +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 + } + + 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 { + // no AT resource manager on this client, nothing to clean up + log.Infof("skip undo log delete, no AT resource manager: %v", err) + return nil + } + + val, ok := resMgr.GetCachedResources().Load(req.ResourceId) + if !ok { + // resource not managed by this client (normal for a broadcast request) + log.Infof("skip undo log delete, resource not managed by this client: %s", req.ResourceId) + return nil + } + + res, ok := val.(dbResource) + if !ok { + return fmt.Errorf("resource %s does not implement dbResource interface", req.ResourceId) Review Comment: 考虑下这里需要error吗？是不是warning然后`return nil`就好了，error上报TC会响应失败 ########## pkg/protocol/message/request_message.go: ########## @@ -131,7 +131,7 @@ func (req GlobalRollbackRequest) GetTypeCode() MessageType { type UndoLogDeleteRequest struct { ResourceId string - SaveDays MessageType + SaveDays int16 Review Comment: 这里的修改是必要的吗，不必要的话对于向外暴露的字段是不是保留原类型好一些 ########## pkg/remoting/processor/client/rm_delete_undolog_processor.go: ########## @@ -0,0 +1,161 @@ +/* + * 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 + +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 + } + + 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 { + // no AT resource manager on this client, nothing to clean up + log.Infof("skip undo log delete, no AT resource manager: %v", err) + return nil + } + + val, ok := resMgr.GetCachedResources().Load(req.ResourceId) + if !ok { + // resource not managed by this client (normal for a broadcast request) + log.Infof("skip undo log delete, resource not managed by this client: %s", req.ResourceId) + return nil + } + + res, ok := val.(dbResource) + if !ok { + return fmt.Errorf("resource %s does not implement dbResource interface", req.ResourceId) + } + + 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 + } + + if req.SaveDays <= 0 { Review Comment: 这个check的时机是不是可以提前到Process入口处，对于非法的应该直接拦掉 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
