Draft fix

Signed-off-by: Matt Lord <mattalord@gmail.com>

Author: mattlord

go/vt/vtctl/workflow/server.go

@@ -3174,6 +3174,19 @@ func (s *Server) switchWrites(ctx context.Context, req *vtctldatapb.WorkflowSwit
 			return 0, sw.logs(), nil
 		}
 
+		// We stop writes on the source before stopping the streams so that the catchup
+		// time is lessened and other workflows that we have to migrate such as
+		// materialize workflows that are within a single keyspace (source and target)
+		// also have a chance to catch up as well as those are internally generated
+		// GTIDs within the shard. For materialization streams that we migrate where
+		// the source and target are the keyspace being resharded, we wait for those
+		// to catchup in the stopStreams path before we actually stop them.
+		ts.Logger().Infof("Stopping source writes")
+		if err := sw.stopSourceWrites(ctx); err != nil {
+			sw.cancelMigration(ctx, sm)
+			return handleError(fmt.Sprintf("failed to stop writes in the %s keyspace", ts.SourceKeyspaceName()), err)
+		}
+
 		ts.Logger().Infof("Stopping streams")
 		sourceWorkflows, err = sw.stopStreams(ctx, sm)
 		if err != nil {
@@ -3186,12 +3199,6 @@ func (s *Server) switchWrites(ctx context.Context, req *vtctldatapb.WorkflowSwit
 			return handleError("failed to stop the workflow streams", err)
 		}
 
-		ts.Logger().Infof("Stopping source writes")
-		if err := sw.stopSourceWrites(ctx); err != nil {
-			sw.cancelMigration(ctx, sm)
-			return handleError(fmt.Sprintf("failed to stop writes in the %s keyspace", ts.SourceKeyspaceName()), err)
-		}
-
 		if ts.MigrationType() == binlogdatapb.MigrationType_TABLES {
 			ts.Logger().Infof("Executing LOCK TABLES on source tables %d times", lockTablesCycles)
 			// Doing this twice with a pause in-between to catch any writes that may have raced in between

go/vt/vtctl/workflow/stream_migrator.go

@@ -250,7 +250,7 @@ func (sm *StreamMigrator) LegacyStopStreams(ctx context.Context) ([]string, erro
 	return sm.legacyVerifyStreamPositions(ctx, positions)
 }
 
-// StopStreams stops streams
+// StopStreams stops streams.
 func (sm *StreamMigrator) StopStreams(ctx context.Context) ([]string, error) {
 	if sm.streams == nil {
 		return nil, nil
@@ -684,6 +684,26 @@ func (sm *StreamMigrator) stopSourceStreams(ctx context.Context) error {
 			return nil
 		}
 
+		// For materialize workflows where the source and target are both the keyspace
+		// that is being resharded, we need to wait for those to catchup as well.
+		// New writes have already been blocked on the source, but the materialization
+		// workflow(s) may still need to catchup with writes that happend just before
+		// writes were stopped on the source.
+		for _, vrs := range tabletStreams {
+			if vrs.WorkflowType == binlogdatapb.VReplicationWorkflowType_Materialize && vrs.BinlogSource.Keyspace == sm.ts.TargetKeyspaceName() {
+				tablet := source.GetPrimary().Tablet
+				pos, err := sm.ts.TabletManagerClient().PrimaryPosition(ctx, tablet)
+				if err != nil {
+					return err
+				}
+				sm.ts.Logger().Infof("Waiting for Materialization workflow %s on %v/%v to reach position %v, starting from position %s",
+					vrs.Workflow, sm.ts.SourceKeyspaceName(), vrs.BinlogSource.Shard, pos, vrs.Position)
+				if err := sm.ts.TabletManagerClient().VReplicationWaitForPos(ctx, tablet, vrs.ID, pos); err != nil {
+					return err
+				}
+			}
+		}
+
 		query := fmt.Sprintf("update _vt.vreplication set state='Stopped', message='for cutover' where id in %s", VReplicationStreams(tabletStreams).Values())
 		_, err := sm.ts.TabletManagerClient().VReplicationExec(ctx, source.GetPrimary().Tablet, query)
 		if err != nil {
@@ -925,6 +945,9 @@ func (sm *StreamMigrator) createTargetStreams(ctx context.Context, tmpl []*VRepl
 				// 1 to 1 in this scenario so we use the target shard's name and primary
 				// tablet's position for the source.
 				vrs.BinlogSource.Shard = target.GetShard().ShardName()
+				// TODO: the problem is that the materialize stream may still need GTIDs
+				// from the OLD shards at this point... so we could miss writes that
+				// occurred on the source table(s) just before the switch.
 				vrs.Position, err = binlogplayer.DecodePosition(target.Position)
 				if err != nil {
 					return err

go/vt/vtctl/workflow/traffic_switcher.go

@@ -939,7 +939,7 @@ func (ts *trafficSwitcher) waitForCatchup(ctx context.Context, filteredReplicati
 	}); err != nil {
 		return err
 	}
-	// all targets have caught up, record their positions for setting up reverse workflows
+	// All targets have caught up, record their positions for setting up reverse workflows.
 	return ts.ForAllTargets(func(target *MigrationTarget) error {
 		var err error
 		target.Position, err = ts.TabletManagerClient().PrimaryPosition(ctx, target.GetPrimary().Tablet)