DBI-798: Safer check on mirror validation with MySQL sources setting server_id

It adds a check on the utilization in the source DB of the server_id.

Author: pfcoperez

flow/cmd/validate_mirror.go

@@ -11,10 +11,12 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/PeerDB-io/peerdb/flow/connectors"
+	connmysql "github.com/PeerDB-io/peerdb/flow/connectors/mysql"
 	"github.com/PeerDB-io/peerdb/flow/generated/proto_conversions"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/pkg/common"
+	mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
@@ -186,6 +188,25 @@ func (h *FlowRequestHandler) checkSourcePeerReuse(
 		return nil
 	}
 
+	// beyond other PeerDB mirrors (checked below), the pinned server_id must not be already registered
+	// as a replica on the source DB itself.
+	mysqlConn, mysqlClose, err := connectors.GetAs[*connmysql.MySqlConnector](ctx, cfg.Env, peer)
+	if err != nil {
+		return NewInternalApiError(fmt.Errorf("failed to create MySQL connector for source peer %s: %w", cfg.SourceName, err))
+	}
+	defer mysqlClose(ctx)
+
+	if inUse, err := mysql_validation.HasReplicaWithServerId(mysqlConn.Conn(), *mysqlCfg.ServerId); err != nil {
+		return NewInternalApiError(fmt.Errorf("failed to check replicas registered on source peer %s: %w", cfg.SourceName, err))
+	} else if inUse {
+		return NewFailedPreconditionApiError(
+			fmt.Errorf("source peer %q pins server_id=%d, which is already in use by a replica registered on the source database",
+				cfg.SourceName, *mysqlCfg.ServerId),
+			NewMirrorErrorInfo(map[string]string{
+				common.ErrorMetadataOffendingField: "source_name",
+			}))
+	}
+
 	// CDC flows for this source peer other than the mirror being validated.
 	// query_string IS NULL filters out QRep flows, which do
 	// not stream the binlog.