VDiff2: Auto retry to continue on error

go/mysql/constants.go

@@ -672,6 +672,47 @@ func IsConnLostDuringQuery(err error) bool {
 	return false
 }
 
+// IsEphemeralError returns true if the error is ephemeral and the caller should
+// retry if possible. Note: non-SQL errors are always treated as ephemeral.
+func IsEphemeralError(err error) bool {
+	if sqlErr, ok := err.(*SQLError); ok {
+		en := sqlErr.Number()
+		switch en {
+		case
+			CRConnectionError,
+			CRConnHostError,
+			CRMalformedPacket,
+			CRNamedPipeStateError,
+			CRServerLost,
+			CRSSLConnectionError,
+			ERCantCreateThread,
+			ERDiskFull,
+			ERForcingClose,
+			ERGotSignal,
+			ERHostIsBlocked,
+			ERLockTableFull,
+			ERInnodbReadOnly,
+			ERInternalError,
+			ERLockDeadlock,
+			ERLockWaitTimeout,
+			EROutOfMemory,
+			EROutOfResources,
+			EROutOfSortMemory,
+			ERQueryInterrupted,
+			ERServerIsntAvailable,
+			ERServerShutdown,
+			ERTooManyUserConnections,
+			ERUnknownError,
+			ERUserLimitReached:
+			return true
+		default:
+			return false
+		}
+	}
+	// If it's not an sqlError then we assume it's ephemeral
+	return true
+}
+
 // IsTooManyConnectionsErr returns true if the error is due to too many connections.
 func IsTooManyConnectionsErr(err error) bool {
 	if sqlErr, ok := err.(*SQLError); ok {

go/test/endtoend/reparent/utils/utils.go

@@ -514,10 +514,11 @@ func GetShardReplicationPositions(t *testing.T, clusterInstance *cluster.LocalPr
 }
 
 func WaitForReplicationToStart(t *testing.T, clusterInstance *cluster.LocalProcessCluster, keyspaceName, shardName string, tabletCnt int, doPrint bool) {
-	tck := time.NewTicker(500 * time.Millisecond)
+	tkr := time.NewTicker(500 * time.Millisecond)
+	defer tkr.Stop()
 	for {
 		select {
-		case <-tck.C:
+		case <-tkr.C:
 			strArray := GetShardReplicationPositions(t, clusterInstance, KeyspaceName, shardName, true)
 			if len(strArray) == tabletCnt && strings.Contains(strArray[0], "primary") { // primary first
 				return
@@ -575,13 +576,13 @@ func CheckReparentFromOutside(t *testing.T, clusterInstance *cluster.LocalProces
 // WaitForReplicationPosition waits for tablet B to catch up to the replication position of tablet A.
 func WaitForReplicationPosition(t *testing.T, tabletA *cluster.Vttablet, tabletB *cluster.Vttablet) error {
 	posA, _ := cluster.GetPrimaryPosition(t, *tabletA, Hostname)
-	timeout := time.Now().Add(5 * time.Second)
+	timeout := time.Now().Add(replicationWaitTimeout)
 	for time.Now().Before(timeout) {
 		posB, _ := cluster.GetPrimaryPosition(t, *tabletB, Hostname)
 		if positionAtLeast(t, tabletB, posA, posB) {
 			return nil
 		}
-		time.Sleep(100 * time.Millisecond)
+		time.Sleep(500 * time.Millisecond)
 	}
 	return fmt.Errorf("failed to catch up on replication position")
 }

go/test/endtoend/vreplication/vdiff2_test.go

@@ -31,47 +31,66 @@ type testCase struct {
 	tables                        string
 	workflow                      string
 	tabletBaseID                  int
-	resume                        bool   // test resume functionality with this workflow
-	resumeInsert                  string // if testing resume, what new rows should be diff'd
-	testCLIErrors                 bool   // test CLI errors against this workflow
+	autoRetryError                bool // if true, test auto retry on error against this workflow
+	// If testing auto retry on error, what new rows should be diff'd. These rows must have a PK > all initial rows.
+	retryInsert string
+	resume      bool // test resume functionality with this workflow
+	// If testing resume, what new rows should be diff'd. These rows must have a PK > all initial rows and retry rows.
+	resumeInsert  string
+	testCLIErrors bool // test CLI errors against this workflow
 }
 
+const (
+	sqlSimulateError = `update _vt.vdiff as vd, _vt.vdiff_table as vdt set vd.state = 'error', vdt.state = 'error', vd.completed_at = NULL,
+						vd.last_error = 'vttablet: rpc error: code = Unknown desc = (errno 1213) (sqlstate 40001): Deadlock found when trying to get lock; try restarting transaction'
+						where vd.vdiff_uuid = %s and vd.id = vdt.vdiff_id`
+	sqlAnalyzeTable = `analyze table %s`
+)
+
 var testCases = []*testCase{
 	{
-		name:          "MoveTables/unsharded to two shards",
-		workflow:      "p1c2",
-		typ:           "MoveTables",
-		sourceKs:      "product",
-		targetKs:      "customer",
-		sourceShards:  "0",
-		targetShards:  "-80,80-",
-		tabletBaseID:  200,
-		tables:        "customer,Lead,Lead-1",
-		resume:        true,
-		resumeInsert:  `insert into customer(cid, name, typ) values(12345678, 'Testy McTester', 'soho')`,
-		testCLIErrors: true, // test for errors in the simplest workflow
+		name:           "MoveTables/unsharded to two shards",
+		workflow:       "p1c2",
+		typ:            "MoveTables",
+		sourceKs:       "product",
+		targetKs:       "customer",
+		sourceShards:   "0",
+		targetShards:   "-80,80-",
+		tabletBaseID:   200,
+		tables:         "customer,Lead,Lead-1",
+		autoRetryError: true,
+		retryInsert:    `insert into customer(cid, name, typ) values(91234, 'Testy McTester', 'soho')`,
+		resume:         true,
+		resumeInsert:   `insert into customer(cid, name, typ) values(92234, 'Testy McTester (redux)', 'enterprise')`,
+		testCLIErrors:  true, // test for errors in the simplest workflow
 	},
 	{
-		name:         "Reshard Merge/split 2 to 3",
-		workflow:     "c2c3",
-		typ:          "Reshard",
-		sourceKs:     "customer",
-		targetKs:     "customer",
-		sourceShards: "-80,80-",
-		targetShards: "-40,40-a0,a0-",
-		tabletBaseID: 400,
-		resume:       true,
-		resumeInsert: `insert into customer(cid, name, typ) values(987654321, 'Testy McTester Jr.', 'enterprise'), (987654322, 'Testy McTester III', 'enterprise')`,
+		name:           "Reshard Merge/split 2 to 3",
+		workflow:       "c2c3",
+		typ:            "Reshard",
+		sourceKs:       "customer",
+		targetKs:       "customer",
+		sourceShards:   "-80,80-",
+		targetShards:   "-40,40-a0,a0-",
+		tabletBaseID:   400,
+		autoRetryError: true,
+		retryInsert:    `insert into customer(cid, name, typ) values(93234, 'Testy McTester Jr', 'enterprise'), (94234, 'Testy McTester II', 'enterprise')`,
+		resume:         true,
+		resumeInsert:   `insert into customer(cid, name, typ) values(95234, 'Testy McTester III', 'enterprise')`,
 	},
 	{
-		name:         "Reshard/merge 3 to 1",
-		workflow:     "c3c1",
-		typ:          "Reshard",
-		sourceKs:     "customer",
-		targetKs:     "customer",
-		sourceShards: "-40,40-a0,a0-",
-		targetShards: "0",
-		tabletBaseID: 700,
+		name:           "Reshard/merge 3 to 1",
+		workflow:       "c3c1",
+		typ:            "Reshard",
+		sourceKs:       "customer",
+		targetKs:       "customer",
+		sourceShards:   "-40,40-a0,a0-",
+		targetShards:   "0",
+		tabletBaseID:   700,
+		autoRetryError: true,
+		retryInsert:    `insert into customer(cid, name, typ) values(96234, 'Testy McTester IV', 'enterprise')`,
+		resume:         true,
+		resumeInsert:   `insert into customer(cid, name, typ) values(97234, 'Testy McTester V', 'enterprise'), (98234, 'Testy McTester VI', 'enterprise')`,
 	},
 }
 
@@ -105,12 +124,13 @@ func TestVDiff2(t *testing.T) {
 	verifyClusterHealth(t, vc)
 
 	insertInitialData(t)
-
 	// Insert null and empty enum values for testing vdiff comparisons for those values.
 	// If we add this to the initial data list, the counts in several other tests will need to change
 	query := `insert into customer(cid, name, typ, sport) values(1001, null, 'soho','')`
 	execVtgateQuery(t, vtgateConn, fmt.Sprintf("%s:%s", sourceKs, sourceShards[0]), query)
 
+	generateMoreCustomers(t, sourceKs, 100)
+
 	_, err := vc.AddKeyspace(t, cells, targetKs, strings.Join(targetShards, ","), customerVSchema, customerSchema, 0, 0, 200, targetKsOpts)
 	require.NoError(t, err)
 	for _, shard := range targetShards {
@@ -149,16 +169,17 @@ func testWorkflow(t *testing.T, vc *VitessCluster, tc *testCase, cells []*Cell)
 	for _, shard := range arrTargetShards {
 		tab := vc.getPrimaryTablet(t, tc.targetKs, shard)
 		catchup(t, tab, tc.workflow, tc.typ)
+		updateTableStats(t, tab, tc.tables) // need to do this in order to test progress reports
 	}
+
 	vdiff(t, tc.targetKs, tc.workflow, cells[0].Name, true, true, nil)
 
+	if tc.autoRetryError {
+		testAutoRetryError(t, tc, cells[0].Name)
+	}
+
 	if tc.resume {
-		expectedRows := int64(0)
-		if tc.resumeInsert != "" {
-			res := execVtgateQuery(t, vtgateConn, tc.sourceKs, tc.resumeInsert)
-			expectedRows = int64(res.RowsAffected)
-		}
-		vdiff2Resume(t, tc.targetKs, tc.workflow, cells[0].Name, expectedRows)
+		testResume(t, tc, cells[0].Name)
 	}
 
 	// This is done here so that we have a valid workflow to test the commands against
@@ -224,3 +245,69 @@ func testNoOrphanedData(t *testing.T, keyspace, workflow string, shards []string
 		}
 	})
 }
+
+func testResume(t *testing.T, tc *testCase, cells string) {
+	t.Run("Resume", func(t *testing.T) {
+		ksWorkflow := fmt.Sprintf("%s.%s", tc.targetKs, tc.workflow)
+
+		// confirm the last VDiff is in the expected completed state
+		uuid, output := performVDiff2Action(t, ksWorkflow, cells, "show", "last", false)
+		jsonOutput := getVDiffInfo(output)
+		require.Equal(t, "completed", jsonOutput.State)
+		// save the number of rows compared in previous runs
+		rowsCompared := jsonOutput.RowsCompared
+		ogTime := time.Now() // the completed_at should be later than this after resuming
+
+		expectedNewRows := int64(0)
+		if tc.resumeInsert != "" {
+			res := execVtgateQuery(t, vtgateConn, tc.sourceKs, tc.resumeInsert)
+			expectedNewRows = int64(res.RowsAffected)
+		}
+		expectedRows := rowsCompared + expectedNewRows
+
+		// confirm that the VDiff was resumed, able to complete, and we compared the
+		// expected number of rows in total (original run and resume)
+		uuid, _ = performVDiff2Action(t, ksWorkflow, cells, "resume", uuid, false)
+		info := waitForVDiff2ToComplete(t, ksWorkflow, cells, uuid, ogTime)
+		require.False(t, info.HasMismatch)
+		require.Equal(t, expectedRows, info.RowsCompared)
+	})
+}
+
+func testAutoRetryError(t *testing.T, tc *testCase, cells string) {
+	t.Run("Auto retry on error", func(t *testing.T) {
+		ksWorkflow := fmt.Sprintf("%s.%s", tc.targetKs, tc.workflow)
+
+		// confirm the last VDiff is in the expected completed state
+		uuid, output := performVDiff2Action(t, ksWorkflow, cells, "show", "last", false)
+		jsonOutput := getVDiffInfo(output)
+		require.Equal(t, "completed", jsonOutput.State)
+		// save the number of rows compared in the first run
+		rowsCompared := jsonOutput.RowsCompared
+		ogTime := time.Now() // the completed_at should be later than this upon retry
+
+		// create new data since original VDiff run -- if requested -- to confirm that the rows
+		// compared is cumulative
+		expectedNewRows := int64(0)
+		if tc.retryInsert != "" {
+			res := execVtgateQuery(t, vtgateConn, tc.sourceKs, tc.retryInsert)
+			expectedNewRows = int64(res.RowsAffected)
+		}
+		expectedRows := rowsCompared + expectedNewRows
+
+		// update the VDiff to simulate an ephemeral error having occurred
+		for _, shard := range strings.Split(tc.targetShards, ",") {
+			tab := vc.getPrimaryTablet(t, tc.targetKs, shard)
+			res, err := tab.QueryTabletWithDB(fmt.Sprintf(sqlSimulateError, encodeString(uuid)), "vt_"+tc.targetKs)
+			require.NoError(t, err)
+			// should have updated the vdiff record and at least one vdiff_table record
+			require.GreaterOrEqual(t, int(res.RowsAffected), 2)
+		}
+
+		// confirm that the VDiff was retried, able to complete, and we compared the expected
+		// number of rows in total (original run and retry)
+		info := waitForVDiff2ToComplete(t, ksWorkflow, cells, uuid, ogTime)
+		require.False(t, info.HasMismatch)
+		require.Equal(t, expectedRows, info.RowsCompared)
+	})
+}