[slack-22.0]: backports: vreplication/vstream bug fixes from v22.0.2

go/mysql/sqlerror/constants.go

@@ -84,11 +84,12 @@ const (
 	ERKeyDoesNotExist       = ErrorCode(1176)
 
 	// permissions
-	ERDBAccessDenied            = ErrorCode(1044)
-	ERAccessDeniedError         = ErrorCode(1045)
-	ERKillDenied                = ErrorCode(1095)
-	ERNoPermissionToCreateUsers = ErrorCode(1211)
-	ERSpecifiedAccessDenied     = ErrorCode(1227)
+	ERDBAccessDenied               = ErrorCode(1044)
+	ERAccessDeniedError            = ErrorCode(1045)
+	ERKillDenied                   = ErrorCode(1095)
+	ERNoPermissionToCreateUsers    = ErrorCode(1211)
+	ERSpecifiedAccessDenied        = ErrorCode(1227)
+	ERBinlogCreateRoutineNeedSuper = ErrorCode(1419)
 
 	// failed precondition
 	ERNoDb                          = ErrorCode(1046)

go/test/endtoend/vreplication/fk_test.go

@@ -29,6 +29,7 @@ import (
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/log"
+	vttablet "vitess.io/vitess/go/vt/vttablet/common"
 
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 )
@@ -101,6 +102,49 @@ func TestFKWorkflow(t *testing.T) {
 	waitForWorkflowState(t, vc, ksWorkflow, binlogdatapb.VReplicationWorkflowState_Running.String())
 	targetKs := vc.Cells[cellName].Keyspaces[targetKeyspace]
 	targetTab := targetKs.Shards["0"].Tablets[fmt.Sprintf("%s-%d", cellName, targetTabletId)].Vttablet
+
+	// Stop the LoadSimulator while we are testing for workflow error, so that
+	// we don't error out in the LoadSimulator as we will be shutting down source dbServer.
+	if withLoad {
+		cancel()
+		<-ch
+	}
+
+	sourceTab := vc.Cells[cellName].Keyspaces[sourceKeyspace].Shards["0"].Tablets[fmt.Sprintf("%s-%d", cellName, 100)]
+
+	// Stop the source database server to simulate an error during replication phase
+	// This should cause recoverable errors that atomic workflows should retry
+	// as it is already out of copy phase.
+	err := sourceTab.DbServer.Stop()
+	require.NoError(t, err)
+
+	// Give some time for the workflow to encounter errors and potentially retry
+	time.Sleep(2 * vttablet.GetDefaultVReplicationConfig().RetryDelay)
+
+	// Verify workflow is still running and hasn't terminated due to errors
+	waitForWorkflowState(t, vc, ksWorkflow, binlogdatapb.VReplicationWorkflowState_Running.String())
+
+	// Restart the source database to allow workflow to continue
+	err = sourceTab.DbServer.StartProvideInit(false)
+	require.NoError(t, err)
+
+	err = vc.VtctldClient.ExecuteCommand("SetWritable", fmt.Sprintf("%s-%d", cellName, 100), "true")
+	require.NoError(t, err)
+
+	// Restart the LoadSimulator.
+	if withLoad {
+		ctx, cancel = context.WithCancel(context.Background())
+		ls = newFKLoadSimulator(t, ctx)
+		defer func() {
+			select {
+			case <-ctx.Done():
+			default:
+				cancel()
+			}
+		}()
+		go ls.simulateLoad()
+	}
+
 	require.NotNil(t, targetTab)
 	catchup(t, targetTab, workflowName, "MoveTables")
 	vdiff(t, targetKeyspace, workflowName, cellName, nil)

go/vt/vtgate/vstream_manager.go

@@ -319,9 +319,19 @@ func (vsm *vstreamManager) GetTotalStreamDelay() int64 {
 
 func (vs *vstream) stream(ctx context.Context) error {
 	ctx, vs.cancel = context.WithCancel(ctx)
-	defer vs.cancel()
 
-	go vs.sendEvents(ctx)
+	vs.wg.Add(1)
+	go func() {
+		defer vs.wg.Done()
+
+		// sendEvents returns either if the given context has been canceled or if
+		// an error is returned from the callback. If the callback returns an error,
+		// we need to cancel the context to stop the other stream goroutines
+		// and to unblock the VStream call.
+		defer vs.cancel()
+
+		vs.sendEvents(ctx)
+	}()
 
 	// Make a copy first, because the ShardGtids list can change once streaming starts.
 	copylist := append(([]*binlogdatapb.ShardGtid)(nil), vs.vgtid.ShardGtids...)
@@ -358,6 +368,7 @@ func (vs *vstream) sendEvents(ctx context.Context) {
 		}
 		return nil
 	}
+
 	for {
 		select {
 		case <-ctx.Done():