move Prepare outof CallWithDB

dtmcli/msg.go

@@ -44,13 +44,15 @@ func (s *Msg) Submit() error {
 func (s *Msg) PrepareAndSubmit(queryPrepared string, db *sql.DB, busiCall BarrierBusiFunc) error {
 	bb, err := BarrierFrom(s.TransType, s.Gid, "00", "msg") // a special barrier for msg QueryPrepared
 	if err == nil {
-		err = bb.CallWithDB(db, func(tx *sql.Tx) error {
+		err = s.Prepare(queryPrepared)
-			err := busiCall(tx)
+	}
-			if err == nil {
+	if err == nil {
-				err = s.Prepare(queryPrepared)
+		defer func() {
+			if err != nil && bb.QueryPrepared(db) == ErrFailure {
+				_ = dtmimp.TransCallDtm(&s.TransBase, s, "abort")
 			}
-			return err
+		}()
-		})
+		err = bb.CallWithDB(db, busiCall)
 	}
 	if err == nil {
 		err = s.Submit()

dtmsvr/api.go

@@ -46,6 +46,10 @@ func svcPrepare(t *TransGlobal) interface{} {
 
 func svcAbort(t *TransGlobal) interface{} {
 	dbt := GetTransGlobal(t.Gid)
+	if dbt.TransType == "msg" && dbt.Status == dtmcli.StatusPrepared {
+		dbt.changeStatus(dtmcli.StatusFailed)
+		return nil
+	}
 	if t.TransType != "xa" && t.TransType != "tcc" || dbt.Status != dtmcli.StatusPrepared && dbt.Status != dtmcli.StatusAborting {
 		return fmt.Errorf("trans type: '%s' current status '%s', cannot abort. %w", dbt.TransType, dbt.Status, dtmcli.ErrFailure)
 	}