vtctl/workflow: stop reverse replication before Complete drops sources by harshit2017 · Pull Request #20188 · vitessio/vitess

harshit2017 · 2026-05-26T18:06:06Z

Description

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.

Related Issue(s)

Fixes #20135

Checklist

"Backport to:" labels have been added if this change should be back-ported to release branches
If this change is to be back-ported to previous releases, a justification is included in the PR description
Tests were added or are not required
Did the new or modified tests pass consistently locally and on CI?
Documentation was added or is not required

Deployment Notes

AI Disclosure

Tests were written by AI

vitess-bot · 2026-05-26T18:06:38Z

Copilot

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

This PR adds safety checks and shutdown behavior for reverse VReplication during workflow completion to prevent completing (and potentially removing source data) while reverse streams are unhealthy or not drained.

Changes:

Validate reverse workflow stream states during doValidateWorkflowHasCompleted.
Add stopAndDrainReverseVReplication to stop reverse streams after waiting for them to catch up to target primary positions.
Update workflow completion tests to expect reverse stream stop queries and add an error-state test case.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
go/vt/vtctl/workflow/utils.go	Adds validation of reverse workflow stream states during completion validation.
go/vt/vtctl/workflow/traffic_switcher.go	Implements draining (wait-for-pos) and stopping reverse VReplication streams prior to source removal.
go/vt/vtctl/workflow/switcher_interface.go	Extends the switcher interface with reverse drain/stop capability.
go/vt/vtctl/workflow/switcher_dry_run.go	Adds dry-run logging for reverse drain/stop behavior.
go/vt/vtctl/workflow/switcher.go	Wires switcher call-through to trafficSwitcher implementation.
go/vt/vtctl/workflow/server.go	Calls reverse drain/stop before dropping sources (unless ignoreSourceKeyspace).
go/vt/vtctl/workflow/server_test.go	Updates expected queries to include stopping reverse streams; adds error-state coverage.
go/vt/vtctl/workflow/framework_test.go	Adjusts test TM client behavior to better model reverse workflows.

+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
+	})
+}


+		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
+				}
+			}
+		}


+	// reverseReplicationDrainTimeout bounds how long Complete waits for reverse
+	// streams to catch up to the target primary position before stopping them.
+	reverseReplicationDrainTimeout = 30 * time.Second


promptless · 2026-05-26T18:10:21Z

Promptless prepared a documentation update related to this change.

Triggered by PR #20188

This PR adds validation of reverse workflow state before MoveTables Complete proceeds. The documentation update explains the new safety behavior: Complete now validates that reverse replication streams are not in Error or Copying state, waits for them to catch up, and stops them before removing source tables.

Review: Document reverse workflow validation in MoveTables Complete

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 vitessio#20135 Signed-off-by: Harshit Katyal <harshit1121998@gmail.com> Signed-off-by: Harshit <harshit1121998@gmail.com>

…sCompleted Remove the redundant outer WaitGroup; ForAllSources and ForAllTargets already run callbacks concurrently and wait internally. Propagate ForAllSources aggregate errors into the error recorder instead of discarding them, and return the error from validateReverseWorkflowForComplete. Signed-off-by: Harshit <harshit1121998@gmail.com>

Copilot

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 5 comments.

+				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))


harshit2017 · 2026-05-27T03:04:54Z

+		res, err := ts.ws.tmc.ReadVReplicationWorkflow(ctx, source.GetPrimary().Tablet, &tabletmanagerdatapb.ReadVReplicationWorkflowRequest{
+			Workflow: ts.ReverseWorkflowName(),
+		})
+		if err != nil {
+			rec.RecordError(err)
+			return nil
+		}


+func (ts *trafficSwitcher) stopAndDrainReverseVReplication(ctx context.Context, waitTime time.Duration) error {
+	ctx, cancel := context.WithTimeout(ctx, waitTime)
+	defer cancel()


+			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 err := ts.TabletManagerClient().VReplicationWaitForPos(ctx, source.GetPrimary().Tablet, stream.Id, pos); err != nil {
+					return err
+				}


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

mattlord · 2026-05-27T03:05:27Z

 			},
-			expectedSourceQueries: []*queryResult{
-				{
+			expectedSourceQueries: append(slices.Clone(stopReverseStreamQueries),


Why are these changes necessary? It's best to eliminate unnecessary changes if we can.

mattlord · 2026-05-27T03:09:11Z

 }

+func (ts *trafficSwitcher) stopAndDrainReverseVReplication(ctx context.Context, waitTime time.Duration) error {
+	targetPositions := make(map[string]string)


I don't think that we should do all of this work to "drain" the stream. The tables are normally deleted. In this case we are renaming them. But new data is ONLY in the keyspace where they were moved. We are holding onto locks and can potentially introduce a whole new set of edge cases and failure scenarios by doing this work and having it take a long time, time out, or error.

I think that we should simply stop the workflow / delete the record for the reverse workflow. i.e. I think it's better to rename the function to just stop, and then only do that, stop it.

mattlord · 2026-05-27T03:11:22Z

 	return nil
 }

+func validateReverseWorkflowForComplete(ctx context.Context, ts *trafficSwitcher, rec *concurrency.AllErrorRecorder) error {


Why do we need to validate it? Why can't we simply stop it / clean it up?

codecov · 2026-05-27T03:21:14Z

Codecov Report

❌ Patch coverage is 54.25532% with 43 lines in your changes missing coverage. Please review.
✅ Project coverage is 72.66%. Comparing base (70c7a72) to head (091b5b8).
⚠️ Report is 277 commits behind head on main.

Files with missing lines	Patch %	Lines
go/vt/vtctl/workflow/traffic_switcher.go	56.60%	23 Missing ⚠️
go/vt/vtctl/workflow/utils.go	53.12%	15 Missing ⚠️
go/vt/vtctl/workflow/switcher_dry_run.go	0.00%	4 Missing ⚠️
go/vt/vtctl/workflow/server.go	66.66%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff             @@
##             main   #20188       +/-   ##
===========================================
+ Coverage   69.67%   72.66%    +2.99%     
===========================================
  Files        1614       22     -1592     
  Lines      216793     7742   -209051     
===========================================
- Hits       151044     5626   -145418     
+ Misses      65749     2116    -63633

Flag	Coverage Δ
partial	`72.66% <54.25%> (?)`

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Copilot AI review requested due to automatic review settings May 26, 2026 18:06

harshit2017 requested review from beingnoble03, mattlord, rohit-nayak-ps and shlomi-noach as code owners May 26, 2026 18:06

github-actions Bot added this to the v25.0.0 milestone May 26, 2026

Copilot AI reviewed May 26, 2026

View reviewed changes

harshit2017 mentioned this pull request May 26, 2026

Bug Report: race condition in MoveTables ... Complete --rename-tables #20135

Open

harshit2017 force-pushed the fix-complete-reverse-drain branch from d02e4a8 to 33916df Compare May 26, 2026 18:25

Copilot AI review requested due to automatic review settings May 26, 2026 18:37

Copilot AI reviewed May 26, 2026

View reviewed changes

vtctl/workflow: harden reverse workflow validation and drain on Complete

091b5b8

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

mattlord reviewed May 27, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

vtctl/workflow: stop reverse replication before Complete drops sources#20188

vtctl/workflow: stop reverse replication before Complete drops sources#20188
harshit2017 wants to merge 3 commits into
vitessio:mainfrom
harshit2017:fix-complete-reverse-drain

harshit2017 commented May 26, 2026

Uh oh!

vitess-bot Bot commented May 26, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

promptless Bot commented May 26, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

harshit2017 May 27, 2026

Uh oh!

mattlord May 27, 2026

Uh oh!

mattlord May 27, 2026

Uh oh!

mattlord May 27, 2026

Uh oh!

codecov Bot commented May 27, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

harshit2017 commented May 26, 2026

Description

Related Issue(s)

Checklist

Deployment Notes

AI Disclosure

Uh oh!

vitess-bot Bot commented May 26, 2026

Review Checklist

General

Tests

Documentation

New flags

If a workflow is added or modified:

Backward compatibility

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

promptless Bot commented May 26, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

harshit2017 May 27, 2026

Choose a reason for hiding this comment

Uh oh!

mattlord May 27, 2026

Choose a reason for hiding this comment

Uh oh!

mattlord May 27, 2026

Choose a reason for hiding this comment

Uh oh!

mattlord May 27, 2026

Choose a reason for hiding this comment

Uh oh!

codecov Bot commented May 27, 2026

Codecov Report

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants