[DBI-798] MySQL Connector: Allow passing server_id as a peer configuration parameter (#4525)

This PR allows setting `server_id` on MySQL/MariaDB peers. 

When it is present, CDC flows will use this value as the [replication
`server_id`](https://dev.mysql.com/doc/mysql-replication-excerpt/5.7/en/replication-options.html#:~:text=server_id%20is%20set%20to%200,replica%20in%20the%20replication%20topology.)
in the declared MySQL replica.

This is an optional field in the configuration and offers a
deterministic alternative to the default "random id" select upon CDC
flow initialization in the current implementation and in the new
implementation when the value is missing.

Note that if a peer has this option, it can not be re-used in different
mirrors. This PR also includes changes to prevent the creation of such
mirrors through validation.

<img width="1222" height="866" alt="image"
src="https://github.com/user-attachments/assets/ecdec95f-5285-46fd-8e1d-430dcecb5c7e"
/>

Commits are organized so they can be reviewed independently:

- bb949c0 Core functionality, adding
optional parameter to the gRPC specification of PeerDB API and
propagating it down to CDC initialization.
- 729225c e2e tests for this
functionality.
- 132c87e Add a validation step in the
create/validate API endpoints that prevents re-using the same MySQL peer
in two mirrors if it has `server_id` set.
- 35f565f API e2e tests for this
validation.
- 0950728 PeerDB UI changes to include
this field in MySQL peer configuration

Part of: https://linear.app/clickhouse/issue/DBI-798

Author: pfcoperez

flow/cmd/validate_mirror.go

@@ -8,6 +8,8 @@ import (
 	"regexp"
 	"slices"
 
+	"google.golang.org/protobuf/proto"
+
 	"github.com/PeerDB-io/peerdb/flow/connectors"
 	"github.com/PeerDB-io/peerdb/flow/generated/proto_conversions"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
@@ -102,6 +104,10 @@ func (h *FlowRequestHandler) validateCDCMirrorImpl(
 		}
 	}
 
+	if apiErr := h.checkSourcePeerReuse(ctx, connectionConfigs); apiErr != nil {
+		return nil, apiErr
+	}
+
 	srcConn, srcClose, err := connectors.GetByNameAs[connectors.MirrorSourceValidationConnector](
 		ctx, connectionConfigs.Env, h.pool, connectionConfigs.SourceName,
 	)
@@ -158,6 +164,73 @@ func (h *FlowRequestHandler) validateCDCMirrorImpl(
 	return &protos.ValidateCDCMirrorResponse{}, nil
 }
 
+// checkSourcePeerReuse rejects a CDC mirror whose MySQL source peer pins a fixed server_id while
+// that peer already backs another streaming CDC mirror. A fixed server_id can only be used by one
+// concurrent binlog connection, so sharing such a peer across mirrors makes their replicas collide
+// on the source. Peers without a fixed server_id fall back to a random per-connection id and are
+// safe to reuse.
+func (h *FlowRequestHandler) checkSourcePeerReuse(
+	ctx context.Context, cfg *protos.FlowConnectionConfigsCore,
+) APIError {
+	// A pinned server_id only matters for CDC; a snapshot-only mirror never opens a binlog connection.
+	if cfg.DoInitialSnapshot && cfg.InitialSnapshotOnly {
+		return nil
+	}
+
+	peer, err := connectors.LoadPeer(ctx, h.pool, cfg.SourceName)
+	if err != nil {
+		return NewInternalApiError(fmt.Errorf("failed to load source peer %s: %w", cfg.SourceName, err))
+	}
+	mysqlCfg := peer.GetMysqlConfig()
+	if mysqlCfg == nil || mysqlCfg.ServerId == nil {
+		return nil
+	}
+
+	// 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.
+	query := `
+	SELECT f.name name, f.config_proto config_proto
+	FROM flows f
+	JOIN peers p ON f.source_peer = p.id AND p.name = $1
+	WHERE f.name != $2 AND f.config_proto IS NOT NULL AND f.query_string IS NULL
+	`
+
+	rows, err := h.pool.Query(ctx, query, cfg.SourceName, cfg.FlowJobName)
+	if err != nil {
+		return NewInternalApiError(fmt.Errorf("failed to check source peer reuse for %s: %w", cfg.SourceName, err))
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var name string
+		var configBytes []byte
+		if err := rows.Scan(&name, &configBytes); err != nil {
+			return NewInternalApiError(fmt.Errorf("failed to read flow row while checking peer reuse: %w", err))
+		}
+		var existing protos.FlowConnectionConfigsCore
+		if err := proto.Unmarshal(configBytes, &existing); err != nil {
+			return NewInternalApiError(fmt.Errorf("failed to unmarshal config for flow %s: %w", name, err))
+		}
+		// A snapshot-only mirror never streams the binlog, so it cannot collide on server_id.
+		if existing.DoInitialSnapshot && existing.InitialSnapshotOnly {
+			continue
+		}
+		return NewFailedPreconditionApiError(
+			fmt.Errorf("source peer %q pins server_id=%d, which cannot be shared across mirrors; "+
+				"it is already used by CDC mirror %q. Use a distinct source peer (or one without a fixed server_id) for this mirror",
+				cfg.SourceName, mysqlCfg.GetServerId(), name),
+			NewMirrorErrorInfo(map[string]string{
+				common.ErrorMetadataOffendingField: "source_name",
+			}))
+	}
+	if err := rows.Err(); err != nil {
+		return NewInternalApiError(fmt.Errorf("failed to iterate flows while checking peer reuse for %s: %w", cfg.SourceName, err))
+	}
+
+	return nil
+}
+
 func (h *FlowRequestHandler) checkIfMirrorNameExists(ctx context.Context, mirrorName string) (bool, error) {
 	var nameExists bool
 	if err := h.pool.QueryRow(ctx, "SELECT EXISTS(SELECT * FROM flows WHERE name = $1)", mirrorName).Scan(&nameExists); err != nil {

flow/connectors/mysql/cdc.go

@@ -10,6 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"math"
 	"math/rand/v2"
 	"slices"
 	"sync/atomic"
@@ -330,9 +331,19 @@ func (c *MySqlConnector) startSyncer(ctx context.Context, env map[string]string)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get event cache count: %w", err)
 	}
-	//nolint:gosec
+
+	var serverId uint32
+	if c.config.ServerId != nil {
+		serverId = *c.config.ServerId
+	} else {
+		// If the configuration doesn't specify a server_id value, fallback to
+		// default behavior of generating a random server_id in the range [1000, MaxUint32).
+		// Range reference: https://dev.mysql.com/doc/refman/9.7/en/replication-options.html#sysvar_server_id
+		serverId = 1000 + rand.Uint32()%(math.MaxUint32-1000) //nolint:gosec // G404: server_id does not require cryptographic randomness
+	}
+
 	return replication.NewBinlogSyncer(replication.BinlogSyncerConfig{
-		ServerID:         rand.Uint32(),
+		ServerID:         serverId,
 		Flavor:           c.Flavor(),
 		Host:             config.Host,
 		Port:             uint16(config.Port),

flow/connectors/mysql/cdc_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	tidbmysql "github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel/metric"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
@@ -23,6 +24,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/otel_metrics"
+	mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
@@ -138,6 +140,38 @@ func TestIntegrationANSIQuotesDDLParsedFromBinlog(t *testing.T) {
 	}
 }
 
+// TestIntegrationConfiguredServerID verifies that an explicitly configured server_id is used for
+// the binlog connection instead of the legacy random fallback: the source must list a replica
+// registered with exactly that id.
+func TestIntegrationConfiguredServerID(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+	}{
+		{name: "mysql"},
+		{name: "mariadb"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			ctx := t.Context()
+			connector := newTestConnector(t, ctx, tc.name)
+
+			wantServerID := uint32(1987)
+			connector.config.ServerId = &wantServerID
+
+			// Establishing the stream sends COM_REGISTER_SLAVE with the configured server_id.
+			// Ref: https://github.com/mysql/mysql-server/blob/6b6d3ed3d5c6591b446276184642d7d0504ecc86/include/my_command.h#L74
+			startBinlogStream(t, ctx, connector)
+
+			// The source's replica list can lag briefly after registration, so retry until it shows up.
+			require.EventuallyWithT(t, func(c *assert.CollectT) {
+				found, err := mysql_validation.HasReplicaWithServerId(connector.conn.Load(), wantServerID)
+				assert.NoError(c, err)
+				assert.True(c, found, "source does not list a replica registered with server_id %d", wantServerID)
+			}, 15*time.Second, time.Second)
+		})
+	}
+}
+
 func TestParseSQLParsesTrailingNull(t *testing.T) {
 	for _, tc := range []struct {
 		name   string

flow/e2e/api_test.go

@@ -325,6 +325,81 @@ func (s APITestSuite) TestClickHouseMirrorValidation_Pass() {
 	require.NotNil(s.t, response)
 }
 
+// TestValidateCDCMirror_ServerIDPeerReuse verifies that validating a CDC mirror whose MySQL source
+// peer pins a fixed server_id is rejected when that peer already backs another streaming CDC mirror.
+func (s APITestSuite) TestValidateCDCMirror_ServerIDPeerReuse() {
+	mySource, ok := s.source.(*MySqlSource)
+	if !ok {
+		s.t.Skip("server_id peer-reuse validation is MySQL-specific")
+	}
+	ctx := s.t.Context()
+
+	// Source table backing the existing mirror.
+	require.NoError(s.t, s.source.Exec(ctx,
+		fmt.Sprintf("CREATE TABLE %s(id int primary key, val text)", AttachSchema(s, "serverid_reuse"))))
+
+	// A MySQL source peer that pins a fixed server_id. It clones the suite's MySQL config, so it
+	// points at the same (reachable) instance.
+	serverID := uint32(4224)
+	pinnedConfig := proto.CloneOf(mySource.Config)
+	pinnedConfig.ServerId = &serverID
+	peerName := "serverid_reuse_" + s.suffix
+	_, err := s.CreatePeer(ctx, &protos.CreatePeerRequest{
+		Peer: &protos.Peer{
+			Name:   peerName,
+			Type:   protos.DBType_MYSQL,
+			Config: &protos.Peer_MysqlConfig{MysqlConfig: pinnedConfig},
+		},
+	})
+	require.NoError(s.t, err)
+
+	// An existing streaming CDC mirror that already uses the pinned-server_id peer as its source.
+	existingGen := FlowConnectionGenerationConfig{
+		FlowJobName:      "serverid_reuse_existing_" + s.suffix,
+		TableNameMapping: map[string]string{AttachSchema(s, "serverid_reuse"): "serverid_reuse"},
+		Destination:      s.ch.Peer().Name,
+	}
+	existingCfg := existingGen.GenerateFlowConnectionConfigs(s)
+	existingCfg.SourceName = peerName
+	existingCfg.DoInitialSnapshot = true
+	createResp, err := s.CreateCDCFlow(ctx, &protos.CreateCDCFlowRequest{ConnectionConfigs: existingCfg})
+	require.NoError(s.t, err)
+	require.NotNil(s.t, createResp)
+
+	tc := NewTemporalClient(s.t)
+	env, err := GetPeerflow(ctx, s.catalog, tc, existingCfg.FlowJobName)
+	require.NoError(s.t, err)
+
+	// Validating a second CDC mirror on the same pinned-server_id peer must be rejected.
+	newGen := FlowConnectionGenerationConfig{
+		FlowJobName:      "serverid_reuse_new_" + s.suffix,
+		TableNameMapping: map[string]string{AttachSchema(s, "serverid_reuse"): "serverid_reuse"},
+		Destination:      s.ch.Peer().Name,
+	}
+	newCfg := newGen.GenerateFlowConnectionConfigs(s)
+	newCfg.SourceName = peerName
+	newCfg.DoInitialSnapshot = true
+
+	res, err := s.ValidateCDCMirror(ctx, &protos.CreateCDCFlowRequest{ConnectionConfigs: newCfg})
+	require.Nil(s.t, res)
+	require.Error(s.t, err)
+	st, ok := status.FromError(err)
+	require.True(s.t, ok)
+	require.Equal(s.t, codes.FailedPrecondition, st.Code())
+	require.Contains(s.t, st.Message(), "cannot be shared across mirrors")
+
+	// Tear down via gRPC: drop the mirror, then the now-unreferenced peer.
+	_, err = s.FlowStateChange(ctx, &protos.FlowStateChangeRequest{
+		FlowJobName:        existingCfg.FlowJobName,
+		RequestedFlowState: protos.FlowStatus_STATUS_TERMINATING,
+	})
+	require.NoError(s.t, err)
+	s.waitForFlowDropped(env, existingCfg.FlowJobName)
+
+	_, err = s.DropPeer(ctx, &protos.DropPeerRequest{PeerName: peerName})
+	require.NoError(s.t, err)
+}
+
 func (s APITestSuite) TestClickHouseMirrorValidation_NoPrimaryKey() {
 	switch s.source.(type) {
 	case *PostgresSource, *MySqlSource:

flow/pkg/mysql/validation.go

@@ -295,3 +295,38 @@ func IsVitess(conn *client.Conn) (bool, error) {
 	}
 	return true, nil // is a Vitess server
 }
+
+func HasReplicaWithServerId(conn *client.Conn, serverID uint32) (bool, error) {
+	// This check assumes that the number of replicas is not beyond the order of magnitude of thousands.
+	// This is a reasonable assumption given that services like RDS limit them to 15 and that
+	// this number is constrained by concurrent connections and main server resources.
+	rs, err := conn.Execute("SHOW REPLICAS")
+	if err != nil {
+		if rs, err = conn.Execute("SHOW SLAVE HOSTS"); err != nil {
+			return false, fmt.Errorf("failed to query source for registered replicas")
+		}
+	}
+
+	serverIDCol := -1
+	for i, field := range rs.Fields {
+		// Case insensitive match because the column name differs between MySQL and MariaDB.
+		if strings.EqualFold(string(field.Name), "Server_id") {
+			serverIDCol = i
+			break
+		}
+	}
+	if serverIDCol < 0 {
+		return false, fmt.Errorf("no Server_id column in replica list, got fields %v", rs.Fields)
+	}
+
+	for row := range rs.RowNumber() {
+		got, err := rs.GetInt(row, serverIDCol)
+		if err != nil {
+			return false, err
+		}
+		if uint32(got) == serverID {
+			return true, nil
+		}
+	}
+	return false, nil
+}

nexus/analyzer/src/lib.rs

@@ -1134,6 +1134,7 @@ fn parse_db_options(db_type: DbType, with_options: &[SqlOption]) -> anyhow::Resu
             }
             .into(),
             aws_auth: None,
+            server_id: opts.get("server_id").and_then(|s| s.parse::<u32>().ok()),
         }),
         DbType::DbtypeUnknown => return Ok(None),
     }))

protos/peers.proto

@@ -234,6 +234,7 @@ message MySqlConfig {
   MySqlAuthType auth_type = 15;
   optional AwsAuthenticationConfig aws_auth = 16;
   bool skip_cert_verification = 17;
+  optional uint32 server_id = 18;
 }
 
 message KafkaConfig {

ui/app/peers/create/[peerType]/helpers/my.ts

@@ -98,6 +98,29 @@ export const mysqlSetting: PeerSetting[] = [
       { value: 'MYSQL_FILEPOS', label: 'FilePos' },
     ],
   },
+  {
+    label: 'Server ID',
+    field: 'serverId',
+    type: 'number',
+    optional: true,
+    stateHandler: (value, setter) => {
+      const strValue = value as string;
+      if (!strValue) {
+        // Leave unset so a random server_id is generated on each CDC run (legacy behavior).
+        setter((curr) => {
+          const newCurr = { ...curr } as MySqlConfig;
+          delete newCurr.serverId;
+          return newCurr;
+        });
+      } else {
+        setter((curr) => ({ ...curr, serverId: parseInt(strValue, 10) }));
+      }
+    },
+    tips:
+      'Optional. Fixed server_id used for the MySQL replication (binlog) connection. ' +
+      'Leave blank to auto-generate a random id on each CDC run. A fixed server_id must be ' +
+      'unique among replicas, so a peer that sets it cannot be reused across multiple mirrors.',
+  },
   {
     label: 'Root Certificate',
     stateHandler: (value, setter) => {