vtctl/workflow: stop reverse replication before Complete drops sources

MoveTables Complete could rename or drop source tables while reverse
vreplication was still applying, causing permanent errno 1146 errors.
Validate reverse workflow state, wait for catchup, stop streams, then
remove source tables.

Fixes #20135

Signed-off-by: Harshit Katyal <harshit1121998@gmail.com>
Signed-off-by: Harshit <harshit1121998@gmail.com>

Author: harshit2017

go/vt/vtctl/workflow/framework_test.go

@@ -365,7 +365,9 @@ func (tmc *testTMClient) ReadVReplicationWorkflow(ctx context.Context, tablet *t
 		}
 	}
 	blsKs := tmc.env.sourceKeyspace
-	if tmc.reverse.Load() && tablet.Keyspace == tmc.env.sourceKeyspace.KeyspaceName {
+	if strings.HasSuffix(req.Workflow, reverseSuffix) && tablet.Keyspace == tmc.env.sourceKeyspace.KeyspaceName {
+		blsKs = tmc.env.targetKeyspace
+	} else if tmc.reverse.Load() && tablet.Keyspace == tmc.env.sourceKeyspace.KeyspaceName {
 		blsKs = tmc.env.targetKeyspace
 	}
 	for i, shard := range blsKs.ShardNames {
@@ -380,7 +382,9 @@ func (tmc *testTMClient) ReadVReplicationWorkflow(ctx context.Context, tablet *t
 				},
 			},
 		}
-		if tmc.frozen.Load() {
+		if strings.HasSuffix(req.Workflow, reverseSuffix) {
+			stream.State = binlogdatapb.VReplicationWorkflowState_Running
+		} else if tmc.frozen.Load() {
 			stream.Message = Frozen
 		}
 		res.Streams = append(res.Streams, stream)

go/vt/vtctl/workflow/server.go

@@ -2513,6 +2513,11 @@ func (s *Server) dropSources(ctx context.Context, ts *trafficSwitcher, removalTy
 			return nil, err
 		}
 	}
+	if !wopts.ignoreSourceKeyspace {
+		if err := sw.stopAndDrainReverseVReplication(ctx, reverseReplicationDrainTimeout); err != nil {
+			return nil, err
+		}
+	}
 	if !keepData {
 		switch ts.MigrationType() {
 		case binlogdatapb.MigrationType_TABLES:

go/vt/vtctl/workflow/server_test.go

@@ -199,6 +199,16 @@ func TestMoveTablesComplete(t *testing.T) {
 	defer cancel()
 
 	workflowName := "wf1"
+	stopReverseStreamQueries := []*queryResult{
+		{
+			query:  "update _vt.vreplication set state='Stopped', message='stopped for complete' where id=1",
+			result: &querypb.QueryResult{},
+		},
+		{
+			query:  "update _vt.vreplication set state='Stopped', message='stopped for complete' where id=2",
+			result: &querypb.QueryResult{},
+		},
+	}
 	table1Name := "t1"
 	table2Name := "t1_2"
 	table3Name := "t1_3"
@@ -259,25 +269,25 @@ func TestMoveTablesComplete(t *testing.T) {
 				TargetKeyspace: targetKeyspaceName,
 				Workflow:       workflowName,
 			},
-			expectedSourceQueries: []*queryResult{
-				{
+			expectedSourceQueries: append(slices.Clone(stopReverseStreamQueries),
+				&queryResult{
 					query:  fmt.Sprintf("drop table `vt_%s`.`%s`", sourceKeyspaceName, table1Name),
 					result: &querypb.QueryResult{},
 				},
-				{
+				&queryResult{
 					query:  fmt.Sprintf("drop table `vt_%s`.`%s`", sourceKeyspaceName, table2Name),
 					result: &querypb.QueryResult{},
 				},
-				{
+				&queryResult{
 					query:  fmt.Sprintf("drop table `vt_%s`.`%s`", sourceKeyspaceName, table3Name),
 					result: &querypb.QueryResult{},
 				},
-				{
+				&queryResult{
 					query: fmt.Sprintf("delete from _vt.vreplication where db_name = 'vt_%s' and workflow = '%s'",
 						sourceKeyspaceName, ReverseWorkflowName(workflowName)),
 					result: &querypb.QueryResult{},
 				},
-			},
+			),
 			expectedTargetQueries: []*queryResult{
 				{
 					query: fmt.Sprintf("delete from _vt.vreplication where db_name = 'vt_%s' and workflow = '%s'",
@@ -306,13 +316,13 @@ func TestMoveTablesComplete(t *testing.T) {
 				KeepRoutingRules: true,
 				KeepData:         new(true),
 			},
-			expectedSourceQueries: []*queryResult{
-				{
+			expectedSourceQueries: append(slices.Clone(stopReverseStreamQueries),
+				&queryResult{
 					query: fmt.Sprintf("delete from _vt.vreplication where db_name = 'vt_%s' and workflow = '%s'",
 						sourceKeyspaceName, ReverseWorkflowName(workflowName)),
 					result: &querypb.QueryResult{},
 				},
-			},
+			),
 			expectedTargetQueries: []*queryResult{
 				{
 					query: fmt.Sprintf("delete from _vt.vreplication where db_name = 'vt_%s' and workflow = '%s'",
@@ -343,25 +353,25 @@ func TestMoveTablesComplete(t *testing.T) {
 				Workflow:       workflowName,
 				RenameTables:   true,
 			},
-			expectedSourceQueries: []*queryResult{
-				{
+			expectedSourceQueries: append(slices.Clone(stopReverseStreamQueries),
+				&queryResult{
 					query:  fmt.Sprintf("rename table `vt_%s`.`%s` TO `vt_%s`.`_%s_old`", sourceKeyspaceName, table1Name, sourceKeyspaceName, table1Name),
 					result: &querypb.QueryResult{},
 				},
-				{
+				&queryResult{
 					query:  fmt.Sprintf("rename table `vt_%s`.`%s` TO `vt_%s`.`_%s_old`", sourceKeyspaceName, table2Name, sourceKeyspaceName, table2Name),
 					result: &querypb.QueryResult{},
 				},
-				{
+				&queryResult{
 					query:  fmt.Sprintf("rename table `vt_%s`.`%s` TO `vt_%s`.`_%s_old`", sourceKeyspaceName, table3Name, sourceKeyspaceName, table3Name),
 					result: &querypb.QueryResult{},
 				},
-				{
+				&queryResult{
 					query: fmt.Sprintf("delete from _vt.vreplication where db_name = 'vt_%s' and workflow = '%s'",
 						sourceKeyspaceName, ReverseWorkflowName(workflowName)),
 					result: &querypb.QueryResult{},
 				},
-			},
+			),
 			expectedTargetQueries: []*queryResult{
 				{
 					query: fmt.Sprintf("delete from _vt.vreplication where db_name = 'vt_%s' and workflow = '%s'",
@@ -374,6 +384,46 @@ func TestMoveTablesComplete(t *testing.T) {
 					workflowName, targetKeyspaceName),
 			},
 		},
+		{
+			name: "reverse workflow in error state",
+			sourceKeyspace: &testKeyspace{
+				KeyspaceName: sourceKeyspaceName,
+				ShardNames:   []string{"0"},
+			},
+			targetKeyspace: &testKeyspace{
+				KeyspaceName: targetKeyspaceName,
+				ShardNames:   []string{"-80", "80-"},
+			},
+			req: &vtctldatapb.MoveTablesCompleteRequest{
+				TargetKeyspace: targetKeyspaceName,
+				Workflow:       workflowName,
+			},
+			preFunc: func(t *testing.T, env *testEnv) {
+				for _, tablet := range env.tablets[sourceKeyspaceName] {
+					env.tmc.expectReadVReplicationWorkflowRequest(tablet.Alias.Uid, &readVReplicationWorkflowRequestResponse{
+						req: &tabletmanagerdatapb.ReadVReplicationWorkflowRequest{
+							Workflow: ReverseWorkflowName(workflowName),
+						},
+						res: &tabletmanagerdatapb.ReadVReplicationWorkflowResponse{
+							Workflow: ReverseWorkflowName(workflowName),
+							Streams: []*tabletmanagerdatapb.ReadVReplicationWorkflowResponse_Stream{
+								{
+									Id:    1,
+									State: binlogdatapb.VReplicationWorkflowState_Error,
+									Bls: &binlogdatapb.BinlogSource{
+										Keyspace: targetKeyspaceName,
+										Shard:    "-80",
+									},
+								},
+							},
+						},
+					})
+				}
+			},
+			wantErr: "reverse vreplication stream 1 is in error state on tablet 100",
+			postFunc: func(t *testing.T, env *testEnv) {
+			},
+		},
 		{
 			name: "ignore source keyspace",
 			sourceKeyspace: &testKeyspace{

go/vt/vtctl/workflow/switcher.go

@@ -116,6 +116,10 @@ func (r *switcher) waitForCatchup(ctx context.Context, filteredReplicationWaitTi
 	return r.ts.waitForCatchup(ctx, filteredReplicationWaitTime)
 }
 
+func (r *switcher) stopAndDrainReverseVReplication(ctx context.Context, waitTime time.Duration) error {
+	return r.ts.stopAndDrainReverseVReplication(ctx, waitTime)
+}
+
 func (r *switcher) stopSourceWrites(ctx context.Context) error {
 	return r.ts.stopSourceWrites(ctx)
 }

go/vt/vtctl/workflow/switcher_dry_run.go

@@ -261,6 +261,12 @@ func (dr *switcherDryRun) waitForCatchup(ctx context.Context, filteredReplicatio
 	return nil
 }
 
+func (dr *switcherDryRun) stopAndDrainReverseVReplication(ctx context.Context, waitTime time.Duration) error {
+	dr.drLog.Logf("Stop and drain reverse vreplication workflow %s for up to %v before removing source data",
+		dr.ts.ReverseWorkflowName(), waitTime)
+	return nil
+}
+
 func (dr *switcherDryRun) stopSourceWrites(ctx context.Context) error {
 	sources := maps.Values(dr.ts.Sources())
 	// Sort the slice for deterministic output.

go/vt/vtctl/workflow/switcher_interface.go

@@ -31,6 +31,7 @@ type iswitcher interface {
 	stopStreams(ctx context.Context, sm *StreamMigrator) ([]string, error)
 	stopSourceWrites(ctx context.Context) error
 	waitForCatchup(ctx context.Context, filteredReplicationWaitTime time.Duration) error
+	stopAndDrainReverseVReplication(ctx context.Context, waitTime time.Duration) error
 	migrateStreams(ctx context.Context, sm *StreamMigrator) error
 	createReverseVReplication(ctx context.Context) error
 	createJournals(ctx context.Context, sourceWorkflows []string) error

go/vt/vtctl/workflow/traffic_switcher.go

@@ -75,6 +75,12 @@ const (
 	// operation too.
 	shardTabletRefreshTimeout = time.Duration(30 * time.Second)
 
+	// reverseReplicationDrainTimeout bounds how long Complete waits for reverse
+	// streams to catch up to the target primary position before stopping them.
+	reverseReplicationDrainTimeout = 30 * time.Second
+
+	stoppedForComplete = "stopped for complete"
+
 	// Use pt-osc's naming convention, this format also ensures vstreamer ignores such tables.
 	renameTableTemplate = "_%.59s_old" // limit table name to 64 characters
 
@@ -1029,6 +1035,64 @@ func (ts *trafficSwitcher) buildTenantPredicate(ctx context.Context) (*sqlparser
 	return tenantPredicate, nil
 }
 
+func (ts *trafficSwitcher) stopAndDrainReverseVReplication(ctx context.Context, waitTime time.Duration) error {
+	ctx, cancel := context.WithTimeout(ctx, waitTime)
+	defer cancel()
+
+	targetPositions := make(map[string]string)
+	if err := ts.ForAllTargets(func(target *MigrationTarget) error {
+		pos, err := ts.TabletManagerClient().PrimaryPosition(ctx, target.GetPrimary().Tablet)
+		if err != nil {
+			return err
+		}
+		targetPositions[target.GetShard().ShardName()] = pos
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	return ts.ForAllSources(func(source *MigrationSource) error {
+		res, err := ts.ws.tmc.ReadVReplicationWorkflow(ctx, source.GetPrimary().Tablet, &tabletmanagerdatapb.ReadVReplicationWorkflowRequest{
+			Workflow: ts.ReverseWorkflowName(),
+		})
+		if err != nil {
+			return err
+		}
+		if res == nil || len(res.Streams) == 0 {
+			return nil
+		}
+		for _, stream := range res.Streams {
+			if stream.Bls == nil {
+				return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "reverse vreplication stream %d on %s has no binlog source",
+					stream.Id, topoproto.TabletAliasString(source.GetPrimary().GetAlias()))
+			}
+			targetShard := stream.Bls.Shard
+			pos, ok := targetPositions[targetShard]
+			if !ok {
+				return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION,
+					"reverse vreplication stream %d on %s reads from unknown target shard %s",
+					stream.Id, topoproto.TabletAliasString(source.GetPrimary().GetAlias()), targetShard)
+			}
+			if stream.State == binlogdatapb.VReplicationWorkflowState_Running {
+				ts.Logger().Infof("Waiting for reverse stream %d on %s to catch up to target shard %s position %s",
+					stream.Id, topoproto.TabletAliasString(source.GetPrimary().GetAlias()), targetShard, pos)
+				if err := ts.TabletManagerClient().VReplicationWaitForPos(ctx, source.GetPrimary().Tablet, stream.Id, pos); err != nil {
+					return err
+				}
+			}
+			if stream.State != binlogdatapb.VReplicationWorkflowState_Stopped {
+				ts.Logger().Infof("Stopping reverse stream %d on %s for complete",
+					stream.Id, topoproto.TabletAliasString(source.GetPrimary().GetAlias()))
+				if _, err := ts.TabletManagerClient().VReplicationExec(ctx, source.GetPrimary().Tablet,
+					binlogplayer.StopVReplication(stream.Id, stoppedForComplete)); err != nil {
+					return err
+				}
+			}
+		}
+		return nil
+	})
+}
+
 func (ts *trafficSwitcher) waitForCatchup(ctx context.Context, filteredReplicationWaitTime time.Duration) error {
 	ctx, cancel := context.WithTimeout(ctx, filteredReplicationWaitTime)
 	defer cancel()

go/vt/vtctl/workflow/utils.go

@@ -579,6 +579,7 @@ func doValidateWorkflowHasCompleted(ctx context.Context, ts *trafficSwitcher) er
 			return nil
 		})
 	}
+	validateReverseWorkflowForComplete(ctx, ts, &wg, &rec)
 	wg.Wait()
 
 	if !ts.keepRoutingRules {
@@ -605,6 +606,39 @@ func doValidateWorkflowHasCompleted(ctx context.Context, ts *trafficSwitcher) er
 	return nil
 }
 
+func validateReverseWorkflowForComplete(ctx context.Context, ts *trafficSwitcher, wg *sync.WaitGroup, rec *concurrency.AllErrorRecorder) {
+	_ = ts.ForAllSources(func(source *MigrationSource) error {
+		wg.Add(1)
+		defer wg.Done()
+		res, err := ts.ws.tmc.ReadVReplicationWorkflow(ctx, source.GetPrimary().Tablet, &tabletmanagerdatapb.ReadVReplicationWorkflowRequest{
+			Workflow: ts.ReverseWorkflowName(),
+		})
+		if err != nil {
+			rec.RecordError(err)
+			return nil
+		}
+		if res == nil || len(res.Streams) == 0 {
+			return nil
+		}
+		for _, stream := range res.Streams {
+			switch stream.State {
+			case binlogdatapb.VReplicationWorkflowState_Running,
+				binlogdatapb.VReplicationWorkflowState_Stopped:
+			case binlogdatapb.VReplicationWorkflowState_Error:
+				rec.RecordError(fmt.Errorf("reverse vreplication stream %d is in error state on tablet %d",
+					stream.Id, source.GetPrimary().Alias.Uid))
+			case binlogdatapb.VReplicationWorkflowState_Copying:
+				rec.RecordError(fmt.Errorf("reverse vreplication stream %d is still copying on tablet %d",
+					stream.Id, source.GetPrimary().Alias.Uid))
+			default:
+				rec.RecordError(fmt.Errorf("reverse vreplication stream %d is in state %s on tablet %d",
+					stream.Id, stream.State, source.GetPrimary().Alias.Uid))
+			}
+		}
+		return nil
+	})
+}
+
 // ReverseWorkflowName returns the "reversed" name of a workflow. For a
 // "forward" workflow, this is the workflow name with "_reverse" appended, and
 // for a "reversed" workflow, this is the workflow name with the "_reverse"