fix: checking for split brain

internal/app/app.go

@@ -827,6 +827,12 @@ func (app *App) calcActiveNodes(clusterState, clusterStateDcs map[string]*NodeSt
 		app.logger.Warnf("failed to get master status %v", err)
 		return nil, err
 	}
+	muuid, err := masterNode.UUID()
+	if err != nil {
+		app.logger.Warnf("failed to get master uuid %v", err)
+		return nil, err
+	}
+
 	for host, node := range clusterState {
 		if host == master {
 			activeNodes = append(activeNodes, master)
@@ -869,7 +875,7 @@ func (app *App) calcActiveNodes(clusterState, clusterStateDcs map[string]*NodeSt
 			continue
 		}
 		sgtids := gtids.ParseGtidSet(sstatus.ExecutedGtidSet)
-		if !(sstatus.ReplicationState == mysql.ReplicationRunning && isGTIDLessOrEqual(sgtids, mgtids)) {
+		if sstatus.ReplicationState != mysql.ReplicationRunning || isSplitBrained(sgtids, mgtids, muuid) {
 			app.logger.Errorf("calc active nodes: %s is not replicating or splitbrained, deleting from active...", host)
 			continue
 		}

internal/app/util.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"time"
 
+	gomysql "github.com/go-mysql-org/go-mysql/mysql"
+	"github.com/google/uuid"
 	"github.com/yandex/mysync/internal/log"
 	"github.com/yandex/mysync/internal/mysql"
 	"github.com/yandex/mysync/internal/mysql/gtids"
@@ -233,6 +235,32 @@ func isGTIDLessOrEqual(slaveGtidSet, masterGtidSet gtids.GTIDSet) bool {
 	return masterGtidSet.Contain(slaveGtidSet) || masterGtidSet.Equal(slaveGtidSet)
 }
 
+func isSplitBrained(slaveGtidSet, masterGtidSet *gomysql.MysqlGTIDSet, masterUUID uuid.UUID) bool {
+	for _, masterSet := range masterGtidSet.Sets {
+		slaveSet, ok := slaveGtidSet.Sets[masterSet.SID.String()]
+		if !ok {
+			continue
+		}
+		if masterSet.Contain(slaveSet) || masterSet == slaveSet {
+			continue
+		}
+
+		if masterSet.SID == masterUUID {
+			continue
+		}
+
+		return true
+	}
+
+	for _, slaveSet := range slaveGtidSet.Sets {
+		if _, ok := masterGtidSet.Sets[slaveSet.SID.String()]; !ok {
+			return true
+		}
+	}
+
+	return false
+}
+
 func validatePriority(priority *int64) error {
 	if priority == nil || *priority >= 0 {
 		return nil

internal/app/util_test.go

@@ -327,3 +327,52 @@ func getLogger() *log.Logger {
 	}
 	return l
 }
+
+func TestIsSplitBrained(t *testing.T) {
+	masterGTID, _ := gomysql.ParseMysqlGTIDSet("6DBC0B04-4B09-43DC-86CC-9AF852DED919:1-100," +
+		"09978591-5754-4710-BF67-062880ABE1B4:1-100," +
+		"AA6890C8-69F8-4BC4-B3A5-5D3FEA8C28CF:1-100")
+	masterMySQLGTID := masterGTID.(*gomysql.MysqlGTIDSet)
+	masterUUID := masterMySQLGTID.Sets["6dbc0b04-4b09-43dc-86cc-9af852ded919"].SID
+
+	// equal gtids
+	slaveGTID, _ := gomysql.ParseMysqlGTIDSet("6DBC0B04-4B09-43DC-86CC-9AF852DED919:1-101," +
+		"09978591-5754-4710-BF67-062880ABE1B4:1-100," +
+		"AA6890C8-69F8-4BC4-B3A5-5D3FEA8C28CF:1-100")
+	slaveMySQLGTID := slaveGTID.(*gomysql.MysqlGTIDSet)
+	ok := isSplitBrained(slaveMySQLGTID, masterMySQLGTID, masterUUID)
+	require.False(t, ok)
+
+	// the replica is lagging behind the master
+	slaveGTID, _ = gomysql.ParseMysqlGTIDSet("6DBC0B04-4B09-43DC-86CC-9AF852DED919:1-99," +
+		"09978591-5754-4710-BF67-062880ABE1B4:1-100," +
+		"AA6890C8-69F8-4BC4-B3A5-5D3FEA8C28CF:1-100")
+	slaveMySQLGTID = slaveGTID.(*gomysql.MysqlGTIDSet)
+	ok = isSplitBrained(slaveMySQLGTID, masterMySQLGTID, masterUUID)
+	require.False(t, ok)
+
+	// the replica applied the transaction from the master before the master
+	slaveGTID, _ = gomysql.ParseMysqlGTIDSet("6DBC0B04-4B09-43DC-86CC-9AF852DED919:1-101," +
+		"09978591-5754-4710-BF67-062880ABE1B4:1-100," +
+		"AA6890C8-69F8-4BC4-B3A5-5D3FEA8C28CF:1-100")
+	slaveMySQLGTID = slaveGTID.(*gomysql.MysqlGTIDSet)
+	ok = isSplitBrained(slaveMySQLGTID, masterMySQLGTID, masterUUID)
+	require.False(t, ok)
+
+	// the replica applied a transaction not from the master
+	slaveGTID, _ = gomysql.ParseMysqlGTIDSet("6DBC0B04-4B09-43DC-86CC-9AF852DED919:1-100," +
+		"09978591-5754-4710-BF67-062880ABE1B4:1-100," +
+		"AA6890C8-69F8-4BC4-B3A5-5D3FEA8C28CF:1-101")
+	slaveMySQLGTID = slaveGTID.(*gomysql.MysqlGTIDSet)
+	ok = isSplitBrained(slaveMySQLGTID, masterMySQLGTID, masterUUID)
+	require.True(t, ok)
+
+	// the replica applied a new transaction not from the master
+	slaveGTID, _ = gomysql.ParseMysqlGTIDSet("6DBC0B04-4B09-43DC-86CC-9AF852DED919:1-101," +
+		"09978591-5754-4710-BF67-062880ABE1B4:1-100," +
+		"AA6890C8-69F8-4BC4-B3A5-5D3FEA8C28CF:1-100," +
+		"BB6890C8-69F8-4BC4-B3A5-5D3FEA8C28CF:1-100")
+	slaveMySQLGTID = slaveGTID.(*gomysql.MysqlGTIDSet)
+	ok = isSplitBrained(slaveMySQLGTID, masterMySQLGTID, masterUUID)
+	require.True(t, ok)
+}

internal/mysql/data.go

@@ -28,6 +28,10 @@ type readOnlyResult struct {
 	SuperReadOnly int `db:"SuperReadOnly"`
 }
 
+type ServerUUIDResult struct {
+	ServerUUID string `db:"server_uuid"`
+}
+
 // CascadeNodeConfiguration is a dcs node configuration for cascade mysql replica
 type CascadeNodeConfiguration struct {
 	// StreamFrom - is a host to stream from. Can be changed from CLI.

internal/mysql/gtids/wrapper.go

@@ -6,10 +6,10 @@ import (
 
 type GTIDSet = mysql.GTIDSet
 
-func ParseGtidSet(gtidset string) mysql.GTIDSet {
-	parsed, err := mysql.ParseGTIDSet(mysql.MySQLFlavor, gtidset)
+func ParseGtidSet(gtidset string) *mysql.MysqlGTIDSet {
+	parsed, err := mysql.ParseMysqlGTIDSet(gtidset)
 	if err != nil {
 		panic(err)
 	}
-	return parsed
+	return parsed.(*mysql.MysqlGTIDSet)
 }

internal/mysql/node.go

@@ -17,10 +17,11 @@ import (
 	"syscall"
 	"time"
 
+	mysql2 "github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-sql-driver/mysql"
+	"github.com/google/uuid"
 	"github.com/jmoiron/sqlx"
 	"github.com/shirou/gopsutil/v3/process"
-
 	"github.com/yandex/mysync/internal/config"
 	"github.com/yandex/mysync/internal/log"
 	"github.com/yandex/mysync/internal/mysql/gtids"
@@ -578,7 +579,7 @@ func (n *Node) GTIDExecuted() (*GTIDExecuted, error) {
 }
 
 // GTIDExecuted returns global transaction id executed
-func (n *Node) GTIDExecutedParsed() (gtids.GTIDSet, error) {
+func (n *Node) GTIDExecutedParsed() (*mysql2.MysqlGTIDSet, error) {
 	gtid, err := n.GTIDExecuted()
 	if err != nil {
 		return nil, err
@@ -602,6 +603,16 @@ func (n *Node) GetBinlogs() ([]Binlog, error) {
 	return binlogs, err
 }
 
+// UUID returns server_uuid
+func (n *Node) UUID() (uuid.UUID, error) {
+	var r ServerUUIDResult
+	err := n.queryRow(queryGetUUID, nil, &r)
+	if err != nil {
+		return uuid.UUID{}, err
+	}
+	return uuid.Parse(r.ServerUUID)
+}
+
 // IsReadOnly returns (true, true) if MySQL Node in (read-only, super-read-only) mode
 func (n *Node) IsReadOnly() (bool, bool, error) {
 	var ror readOnlyResult

internal/mysql/queries.go

@@ -6,6 +6,7 @@ const (
 	queryReplicaStatus                  = "replica_status"
 	queryGetVersion                     = "get_version"
 	queryGTIDExecuted                   = "gtid_executed"
+	queryGetUUID                        = "get_uuid"
 	queryShowBinaryLogs                 = "binary_logs"
 	queryReplicationLag                 = "replication_lag"
 	querySlaveHosts                     = "slave_hosts"
@@ -53,6 +54,7 @@ var DefaultQueries = map[string]string{
 	queryReplicaStatus:       `SHOW REPLICA STATUS FOR CHANNEL :channel`,
 	queryGetVersion:          `SELECT sys.version_major() AS MajorVersion, sys.version_minor() AS MinorVersion, sys.version_patch() AS PatchVersion`,
 	queryGTIDExecuted:        `SELECT @@GLOBAL.gtid_executed  as Executed_Gtid_Set`,
+	queryGetUUID:             `SELECT @@server_uuid as server_uuid`,
 	queryShowBinaryLogs:      `SHOW BINARY LOGS`,
 	querySlaveHosts:          `SHOW SLAVE HOSTS`,
 	queryReplicationLag:      ``,