vtctl/workflow: stop reverse replication before Complete drops sources

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 cell-0000000100",
+			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,74 @@ func (ts *trafficSwitcher) buildTenantPredicate(ctx context.Context) (*sqlparser
 	return tenantPredicate, nil
 }
 
+func (ts *trafficSwitcher) stopAndDrainReverseVReplication(ctx context.Context, waitTime time.Duration) error {
+	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
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, waitTime)
+	defer cancel()
+
+	return ts.ForAllSources(func(source *MigrationSource) error {
+		tabletAlias := topoproto.TabletAliasString(source.GetPrimary().GetAlias())
+		res, err := ts.ws.tmc.ReadVReplicationWorkflow(ctx, source.GetPrimary().Tablet, &tabletmanagerdatapb.ReadVReplicationWorkflowRequest{
+			Workflow: ts.ReverseWorkflowName(),
+		})
+		if err != nil {
+			return vterrors.Wrapf(err, "reading reverse workflow %s on %s", ts.ReverseWorkflowName(), tabletAlias)
+		}
+		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, tabletAlias)
+			}
+			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, tabletAlias, targetShard)
+			}
+			switch stream.State {
+			case binlogdatapb.VReplicationWorkflowState_Error:
+				return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION,
+					"reverse vreplication stream %d on %s is in error state", stream.Id, tabletAlias)
+			case binlogdatapb.VReplicationWorkflowState_Copying:
+				return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION,
+					"reverse vreplication stream %d on %s is still copying", stream.Id, tabletAlias)
+			case binlogdatapb.VReplicationWorkflowState_Stopped:
+				continue
+			case binlogdatapb.VReplicationWorkflowState_Running:
+				ts.Logger().Infof("Waiting for reverse stream %d on %s to catch up to target shard %s position %s",
+					stream.Id, tabletAlias, targetShard, pos)
+				if err := ts.TabletManagerClient().VReplicationWaitForPos(ctx, source.GetPrimary().Tablet, stream.Id, pos); err != nil {
+					return err
+				}
+			default:
+				return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION,
+					"reverse vreplication stream %d on %s is in state %s", stream.Id, tabletAlias, stream.State)
+			}
+			ts.Logger().Infof("Stopping reverse stream %d on %s for complete", stream.Id, tabletAlias)
+			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

@@ -548,21 +548,18 @@ func HashStreams(targetKeyspace string, targets map[string]*MigrationTarget) int
 }
 
 func doValidateWorkflowHasCompleted(ctx context.Context, ts *trafficSwitcher) error {
-	wg := sync.WaitGroup{}
 	rec := concurrency.AllErrorRecorder{}
 	if ts.MigrationType() == binlogdatapb.MigrationType_SHARDS {
-		_ = ts.ForAllSources(func(source *MigrationSource) error {
-			wg.Add(1)
+		if err := ts.ForAllSources(func(source *MigrationSource) error {
 			if source.GetShard().IsPrimaryServing {
 				rec.RecordError(fmt.Errorf("shard %s is still serving", source.GetShard().ShardName()))
 			}
-			wg.Done()
 			return nil
-		})
+		}); err != nil {
+			rec.RecordError(err)
+		}
 	} else {
-		_ = ts.ForAllTargets(func(target *MigrationTarget) error {
-			wg.Add(1)
-			defer wg.Done()
+		if err := ts.ForAllTargets(func(target *MigrationTarget) error {
 			res, err := ts.ws.tmc.ReadVReplicationWorkflow(ctx, target.GetPrimary().Tablet, &tabletmanagerdatapb.ReadVReplicationWorkflowRequest{
 				Workflow: ts.WorkflowName(),
 			})
@@ -577,9 +574,13 @@ func doValidateWorkflowHasCompleted(ctx context.Context, ts *trafficSwitcher) er
 				}
 			}
 			return nil
-		})
+		}); err != nil {
+			rec.RecordError(err)
+		}
+	}
+	if err := validateReverseWorkflowForComplete(ctx, ts, &rec); err != nil {
+		rec.RecordError(err)
 	}
-	wg.Wait()
 
 	if !ts.keepRoutingRules {
 		// Check if table is routable.
@@ -605,6 +606,38 @@ func doValidateWorkflowHasCompleted(ctx context.Context, ts *trafficSwitcher) er
 	return nil
 }
 
+func validateReverseWorkflowForComplete(ctx context.Context, ts *trafficSwitcher, rec *concurrency.AllErrorRecorder) error {
+	return ts.ForAllSources(func(source *MigrationSource) error {
+		tabletAlias := topoproto.TabletAliasString(source.GetPrimary().GetAlias())
+		res, err := ts.ws.tmc.ReadVReplicationWorkflow(ctx, source.GetPrimary().Tablet, &tabletmanagerdatapb.ReadVReplicationWorkflowRequest{
+			Workflow: ts.ReverseWorkflowName(),
+		})
+		if err != nil {
+			rec.RecordError(vterrors.Wrapf(err, "reading reverse workflow %s on %s", ts.ReverseWorkflowName(), tabletAlias))
+			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 %s",
+					stream.Id, tabletAlias))
+			case binlogdatapb.VReplicationWorkflowState_Copying:
+				rec.RecordError(fmt.Errorf("reverse vreplication stream %d is still copying on %s",
+					stream.Id, tabletAlias))
+			default:
+				rec.RecordError(fmt.Errorf("reverse vreplication stream %d is in state %s on %s",
+					stream.Id, stream.State, tabletAlias))
+			}
+		}
+		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"