Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pfcoperez wants to merge 2 commits into
base: main
Choose a base branch
from
+29 −0
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
19c30f3
DBI-798: Safer check on mirror validation with MySQL sources setting …
pfcoperez Jul 10, 2026
5b0ffb0
Reorder code
pfcoperez Jul 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions flow/cmd/validate_mirror.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ 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"
Expand Down Expand Up @@ -228,6 +229,25 @@ func (h *FlowRequestHandler) checkSourcePeerReuse(
return NewInternalApiError(fmt.Errorf("failed to iterate flows while checking peer reuse for %s: %w", cfg.SourceName, err))
}

// Beyond other PeerDB mirrors (checked above), 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 := mysqlConn.HasReplicaWithServerId(ctx, *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",
}))
}

return nil
}

Expand Down
9 changes: 9 additions & 0 deletions flow/connectors/mysql/validate.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,15 @@ func (c *MySqlConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.F
return nil
}

// HasReplicaWithServerId reports whether the source lists a replica registered with serverID.
func (c *MySqlConnector) HasReplicaWithServerId(ctx context.Context, serverID uint32) (bool, error) {
conn, err := c.connect(ctx)
if err != nil {
return false, fmt.Errorf("failed to connect: %w", err)
}
return mysql_validation.HasReplicaWithServerId(conn, serverID)
}

func (c *MySqlConnector) ValidateCheck(ctx context.Context) error {
if c.config.Flavor == protos.MySqlFlavor_MYSQL_UNKNOWN {
return fmt.Errorf("flavor is set to unknown")
Expand Down
Loading