Update source positions once we know all writes are done.

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

go/vt/vtctl/workflow/server.go

@@ -3498,6 +3498,13 @@ func (s *Server) switchWrites(ctx context.Context, req *vtctldatapb.WorkflowSwit
 			}
 		}
 
+		// Ensure that the source positions are correct now that writes are stopped, the streams were
+		// stopped (e.g. intra-keyspace materializations that write on the source), and any in progress
+		// writes are done.
+		if err := ts.updateSourcePositions(ctx); err != nil {
+			return handleError("failed to update source replication positions", err)
+		}
+
 		if err := confirmKeyspaceLocksHeld(); err != nil {
 			return handleError("locks were lost", err)
 		}

go/vt/vtctl/workflow/traffic_switcher.go

@@ -1319,6 +1319,17 @@ func (ts *trafficSwitcher) gatherPositions(ctx context.Context) error {
 	})
 }
 
+// updateSourcePositions will update the Position for all migration sources.
+func (ts *trafficSwitcher) updateSourcePositions(ctx context.Context) error {
+	err := ts.ForAllSources(func(source *MigrationSource) error {
+		var err error
+		source.Position, err = ts.ws.tmc.PrimaryPosition(ctx, source.GetPrimary().Tablet)
+		ts.Logger().Infof("Updated position for source %v:%v: %v", ts.SourceKeyspaceName(), source.GetShard().ShardName(), source.Position)
+		return err
+	})
+	return err
+}
+
 func (ts *trafficSwitcher) isSequenceParticipating(ctx context.Context) (bool, error) {
 	vschema, err := ts.TopoServer().GetVSchema(ctx, ts.targetKeyspace)
 	if err != nil {