GTID master failover recovery + integration test (GTID failover 3/3)

.github/workflows/start-mysql.sh

@@ -1,12 +1,15 @@
 #!/bin/bash
 set -xe
 
+# mysql-3 is the promoted writer used by the master-failover integration test.
+# It is started alongside the source/target so that test can run; other tests
+# ignore it.
 if [ "$MYSQL_VERSION" == "8.0" ]; then
-  docker compose -f docker-compose_8.0.yml up -d mysql-1 mysql-2
+  docker compose -f docker-compose_8.0.yml up -d mysql-1 mysql-2 mysql-3
 elif [ "$MYSQL_VERSION" == "8.4" ]; then
-  docker compose -f docker-compose_8.4.yml up -d mysql-1 mysql-2
+  docker compose -f docker-compose_8.4.yml up -d mysql-1 mysql-2 mysql-3
 else
-  docker compose up -d mysql-1 mysql-2
+  docker compose up -d mysql-1 mysql-2 mysql-3
 fi
 
 MAX_ATTEMPTS=60
@@ -38,6 +41,8 @@ wait_for_configuration () {
 
 wait_for_version "ghostferry-mysql-1-1"
 wait_for_version "ghostferry-mysql-2-1"
+wait_for_version "ghostferry-mysql-3-1"
 
 wait_for_configuration 29291
 wait_for_configuration 29292
+wait_for_configuration 29293

.gitignore

@@ -21,3 +21,6 @@ build/
 .idea
 
 .ipynb_checkpoints
+
+# Local test tmpdir (macOS TMPDIR workaround)
+.tmp_ghostferry/

binlog_coordinate.go

@@ -261,6 +261,67 @@ func intersectGTIDSets(a, b mysql.GTIDSet) (mysql.GTIDSet, error) {
 	return result, nil
 }
 
+// unionGTIDStringInto returns a new GTID set equal to base with the GTID(s) in
+// add merged in. base may be nil (treated as empty). It never mutates base. It
+// is used by failover recovery to fold an in-flight transaction's GTID into the
+// already-applied set before validating a candidate master.
+func unionGTIDStringInto(base mysql.GTIDSet, add string) (mysql.GTIDSet, error) {
+	var result *mysql.MysqlGTIDSet
+	if base == nil {
+		empty, err := mysql.ParseMysqlGTIDSet("")
+		if err != nil {
+			return nil, err
+		}
+		result = empty.(*mysql.MysqlGTIDSet)
+	} else {
+		clone, ok := base.Clone().(*mysql.MysqlGTIDSet)
+		if !ok {
+			return nil, fmt.Errorf("GTID union requires a MySQL GTID set, got %T", base)
+		}
+		result = clone
+	}
+
+	if err := result.Update(add); err != nil {
+		return nil, fmt.Errorf("GTID union: updating with %q: %w", add, err)
+	}
+	return result, nil
+}
+
+// unionGTIDSets returns a new GTID set equal to a with b merged in. Either may
+// be nil (treated as empty). It never mutates its inputs. It is used by failover
+// recovery to require a candidate master to contain both the applied set and the
+// cutover stop target.
+func unionGTIDSets(a, b mysql.GTIDSet) (mysql.GTIDSet, error) {
+	var result *mysql.MysqlGTIDSet
+	if a == nil {
+		empty, err := mysql.ParseMysqlGTIDSet("")
+		if err != nil {
+			return nil, err
+		}
+		result = empty.(*mysql.MysqlGTIDSet)
+	} else {
+		clone, ok := a.Clone().(*mysql.MysqlGTIDSet)
+		if !ok {
+			return nil, fmt.Errorf("GTID union requires MySQL GTID sets, got %T", a)
+		}
+		result = clone
+	}
+
+	if b != nil {
+		bMysql, ok := b.(*mysql.MysqlGTIDSet)
+		if !ok {
+			return nil, fmt.Errorf("GTID union requires MySQL GTID sets, got %T", b)
+		}
+		for _, uuidSet := range bMysql.Sets {
+			// Clone so result never aliases b's internal UUIDSet (AddSet stores
+			// the pointer directly when the SID is not already present).
+			result.AddSet(uuidSet.Clone())
+		}
+	}
+
+	return result, nil
+}
+
 // serializedBinlogCoordinate is the on-disk / on-wire shape of a
 // BinlogCoordinate. It is deliberately explicit and self-describing so that a
 // future GTID variant can be added as additional fields without breaking