Copilot commented on code in PR #1135: URL: https://github.com/apache/incubator-seata-go/pull/1135#discussion_r3536434417
########## pkg/remoting/processor/client/rm_delete_undolog_processor.go: ########## @@ -0,0 +1,156 @@ +/* + * 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) + Review Comment: `SaveDays` is used later to compute `before := time.Now().AddDate(0, 0, -int(req.SaveDays))`. For `SaveDays <= 0` this produces a cutoff in the future (or today), which can delete far more undo logs than intended. Add an explicit guard to skip non-positive values (as the PR description notes this is invalid). ########## pkg/remoting/processor/client/rm_delete_undolog_processor.go: ########## @@ -0,0 +1,156 @@ +/* + * 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) + } Review Comment: `batchDeleteByLogCreated` builds `DELETE ... WHERE ... <= ? LIMIT ...`, which is MySQL-specific (placeholder style and `DELETE ... LIMIT`). If a non-MySQL `DBType` is managed by the client (e.g. PostgreSQL), this path will error and bubble up from `Process`. Consider explicitly skipping unsupported `DBType`s until dialect-specific cleanup SQL is implemented. ########## 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) Review Comment: `Decode` reads `ResourceId` via `bytes.ReadString16Length(buf)`, which ignores short reads and errors. That can undermine the goal of rejecting malformed bodies; it’s safer for this codec to explicitly read and validate the 2-byte length and the subsequent bytes before continuing. -- 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]
