From e705efe52f6510f8841e1d6a900afef7fb04ef78 Mon Sep 17 00:00:00 2001
From: Yang Keao <yangkeao@chunibyo.icu>
Date: Thu, 26 Sep 2024 22:16:22 +0800
Subject: [PATCH] add delay to avoid breaking consistency for add-foreign-key
 and async commit

Signed-off-by: Yang Keao <yangkeao@chunibyo.icu>
---
 pkg/ddl/foreign_key.go                 |  3 ++
 tests/realtikvtest/txntest/BUILD.bazel |  1 +
 tests/realtikvtest/txntest/txn_test.go | 65 ++++++++++++++++++++++++++
 3 files changed, 69 insertions(+)

diff --git a/pkg/ddl/foreign_key.go b/pkg/ddl/foreign_key.go
index 3721e28d679da..cb4b723bad714 100644
--- a/pkg/ddl/foreign_key.go
+++ b/pkg/ddl/foreign_key.go
@@ -20,6 +20,7 @@ import (
 	"strings"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	infoschemactx "github.com/pingcap/tidb/pkg/infoschema/context"
 	"github.com/pingcap/tidb/pkg/meta"
@@ -67,11 +68,13 @@ func (w *worker) onCreateForeignKey(jobCtx *jobContext, t *meta.Meta, job *model
 		job.SchemaState = model.StateWriteOnly
 		return ver, nil
 	case model.StateWriteOnly:
+		delayForAsyncCommit()
 		err = checkForeignKeyConstrain(w, job.SchemaName, tblInfo.Name.L, fkInfo, fkCheck)
 		if err != nil {
 			job.State = model.JobStateRollingback
 			return ver, err
 		}
+		failpoint.InjectCall("afterCheckForeignKeyConstrain")
 		tblInfo.ForeignKeys[len(tblInfo.ForeignKeys)-1].State = model.StateWriteReorganization
 		ver, err = updateVersionAndTableInfo(jobCtx, t, job, tblInfo, true)
 		if err != nil {
diff --git a/tests/realtikvtest/txntest/BUILD.bazel b/tests/realtikvtest/txntest/BUILD.bazel
index bd5b9d43a01f2..01cd9d87ad593 100644
--- a/tests/realtikvtest/txntest/BUILD.bazel
+++ b/tests/realtikvtest/txntest/BUILD.bazel
@@ -21,6 +21,7 @@ go_test(
         "//pkg/store/driver",
         "//pkg/testkit",
         "//pkg/util",
+        "//pkg/util/sqlexec",
         "//tests/realtikvtest",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
diff --git a/tests/realtikvtest/txntest/txn_test.go b/tests/realtikvtest/txntest/txn_test.go
index a4193164fea95..b9d0ec09ab7af 100644
--- a/tests/realtikvtest/txntest/txn_test.go
+++ b/tests/realtikvtest/txntest/txn_test.go
@@ -19,6 +19,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -29,6 +30,7 @@ import (
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/pingcap/tidb/pkg/util/sqlexec"
 	"github.com/pingcap/tidb/tests/realtikvtest"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/client-go/v2/tikvrpc"
@@ -532,3 +534,66 @@ func TestCheckTxnStatusOnOptimisticTxnBreakConsistency(t *testing.T) {
 	tk2.MustExec("admin check table t2")
 	tk2.MustQuery("select * from t2 order by id").Check(testkit.Rows("1 10", "2 11"))
 }
+
+func TestDMLWithAddForeignKey(t *testing.T) {
+	defer config.RestoreFunc()()
+	config.UpdateGlobal(func(conf *config.Config) {
+		conf.TiKVClient.AsyncCommit.SafeWindow = 10 * time.Second
+		conf.TiKVClient.AsyncCommit.AllowedClockDrift = 500 * time.Millisecond
+	})
+
+	store := realtikvtest.CreateMockStoreAndSetup(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("set global tidb_enable_1pc='OFF';")
+	tk.MustExec("set global tidb_enable_metadata_lock='OFF';")
+	tk.MustExec("set global tidb_enable_async_commit='ON'")
+
+	tkDML := testkit.NewTestKit(t, store)
+	tkDML.MustExec("use test")
+
+	tkDDL := testkit.NewTestKit(t, store)
+	tkDDL.MustExec("use test")
+	tkDDL.MustExec("create table parent (id int primary key, val int, index(val));")
+	tkDDL.MustExec("create table child (id int primary key, val int, index(val));")
+
+	// The fail path of this test is:
+	// tk:     INSERT -> ... -> Wait                                       -> PreWrite -> ... -> Async Commit -> Wait    -> ... -> Success.
+	// tkDDL:            DDL -> StateWriteOnly -> checkForeignKeyConstrain -> DDL                             -> Success
+	// After fixing, either the `tkDDL` or `tk` will fail.
+	failpoint.Enable("tikvclient/beforePrewrite", "pause")
+	defer failpoint.Disable("tikvclient/beforePrewrite")
+	failpoint.EnableCall("github.com/pingcap/tidb/pkg/ddl/afterCheckForeignKeyConstrain", func() {
+		require.NoError(t, failpoint.Disable("tikvclient/beforePrewrite"))
+	})
+	defer failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/afterCheckForeignKeyConstrain")
+	failpoint.Enable("tikvclient/asyncCommitDoNothing", "pause")
+	defer failpoint.Disable("tikvclient/asyncCommitDoNothing")
+
+	var wg sync.WaitGroup
+	var errDML, errDDL error
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+
+		var rs sqlexec.RecordSet
+		rs, errDML = tkDML.Exec("insert into child values (1, 1)")
+		if rs != nil {
+			rs.Close()
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+
+		var rs sqlexec.RecordSet
+		rs, errDDL = tkDDL.Exec("alter table child add foreign key fk(val) references parent (val);")
+		if rs != nil {
+			rs.Close()
+		}
+		require.NoError(t, failpoint.Disable("tikvclient/asyncCommitDoNothing"))
+	}()
+
+	wg.Wait()
+
+	require.True(t, errDML != nil || errDDL != nil)
+}