GTID 4/5: Persist GTID coordinates in state, enable resume, test both modes in CI

.github/workflows/tests.yml

@@ -43,7 +43,15 @@ jobs:
   go-test:
     strategy:
       matrix:
+        # file_position is exercised on every supported MySQL version.
         mysql: ["5.7", "8.0", "8.4"]
+        binlog_coordinate_mode: ["file_position"]
+        include:
+          # GTID coordinate mode is only supported/tested on MySQL 8+.
+          - mysql: "8.0"
+            binlog_coordinate_mode: "gtid"
+          - mysql: "8.4"
+            binlog_coordinate_mode: "gtid"
 
     runs-on: ubuntu-latest
     timeout-minutes: 15
@@ -52,6 +60,7 @@ jobs:
     env:
       CI: "true"
       MYSQL_VERSION: ${{ matrix.mysql }}
+      GHOSTFERRY_BINLOG_COORDINATE_MODE: ${{ matrix.binlog_coordinate_mode }}
 
     steps:
       - uses: actions/checkout@v6.0.2
@@ -68,11 +77,21 @@ jobs:
   ruby-test:
     strategy:
       matrix:
+        # file_position is exercised on every supported MySQL version.
         mysql: ["5.7", "8.0", "8.4"]
         log_backend: ["logrus"]
+        binlog_coordinate_mode: ["file_position"]
         include:
           - mysql: "8.4"
             log_backend: "zerolog"
+            binlog_coordinate_mode: "file_position"
+          # GTID coordinate mode is only supported/tested on MySQL 8+.
+          - mysql: "8.0"
+            log_backend: "logrus"
+            binlog_coordinate_mode: "gtid"
+          - mysql: "8.4"
+            log_backend: "logrus"
+            binlog_coordinate_mode: "gtid"
 
     runs-on: ubuntu-latest
     timeout-minutes: 15
@@ -83,6 +102,7 @@ jobs:
       BUNDLE_WITHOUT: "development"
       MYSQL_VERSION: ${{ matrix.mysql }}
       GHOSTFERRY_LOG_BACKEND: ${{ matrix.log_backend }}
+      GHOSTFERRY_BINLOG_COORDINATE_MODE: ${{ matrix.binlog_coordinate_mode }}
 
     steps:
       - uses: actions/checkout@v6.0.2

binlog_coordinate.go

@@ -225,6 +225,42 @@ func (c BinlogCoordinate) String() string {
 	}
 }
 
+// intersectGTIDSets returns the intersection of two MySQL GTID sets, i.e. the
+// GTIDs present in both. go-mysql does not expose an intersection primitive, so
+// this computes A ∩ B = A - (A - B).
+//
+// It operates on clones and does not mutate its inputs. It is fail-closed:
+// rather than silently falling back to one side (which could advance a resume
+// floor past what a consumer durably processed and skip events), it returns an
+// error so the caller can refuse to resume.
+func intersectGTIDSets(a, b mysql.GTIDSet) (mysql.GTIDSet, error) {
+	aMysql, aOK := a.(*mysql.MysqlGTIDSet)
+	bMysql, bOK := b.(*mysql.MysqlGTIDSet)
+	if !aOK || !bOK {
+		return nil, fmt.Errorf("GTID intersection requires MySQL GTID sets, got %T and %T", a, b)
+	}
+
+	// diff = A - B
+	diff, ok := aMysql.Clone().(*mysql.MysqlGTIDSet)
+	if !ok {
+		return nil, fmt.Errorf("GTID intersection: unexpected clone type %T", aMysql.Clone())
+	}
+	if err := diff.Minus(*bMysql); err != nil {
+		return nil, fmt.Errorf("GTID intersection (A - B): %w", err)
+	}
+
+	// result = A - diff = A ∩ B
+	result, ok := aMysql.Clone().(*mysql.MysqlGTIDSet)
+	if !ok {
+		return nil, fmt.Errorf("GTID intersection: unexpected clone type %T", aMysql.Clone())
+	}
+	if err := result.Minus(*diff); err != nil {
+		return nil, fmt.Errorf("GTID intersection (A - diff): %w", err)
+	}
+
+	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

binlog_streamer.go

@@ -666,6 +666,23 @@ func (s *BinlogStreamer) handleRowsEvent(ev *replication.BinlogEvent, query []by
 		return err
 	}
 
+	// In GTID mode, stamp GTID coordinates onto the events so that downstream
+	// consumers (binlog writer, verifiers) advance GTID-based state rather than
+	// file/position. The resumable coordinate is the committed set BEFORE the
+	// current transaction, so an interruption replays the whole transaction.
+	if s.coordinateMode() == BinlogCoordinateGTID {
+		currentCoord := NewGTIDCoordinateFromSet(s.lastStreamedGTIDSet)
+		resumableCoord := NewGTIDCoordinateFromSet(s.lastResumableGTIDSet)
+		for _, dmlEv := range dmlEvs {
+			// SetCoordinates is an internal capability (coordinateStamper), not
+			// part of the exported DMLEvent interface; all built-in events
+			// satisfy it via DMLEventBase.
+			if stamper, ok := dmlEv.(coordinateStamper); ok {
+				stamper.SetCoordinates(currentCoord, resumableCoord)
+			}
+		}
+	}
+
 	events := make([]DMLEvent, 0)
 
 	for _, dmlEv := range dmlEvs {

binlog_writer.go

@@ -123,7 +123,7 @@ func (b *BinlogWriter) writeEvents(events []DMLEvent) error {
 	}
 
 	if b.StateTracker != nil {
-		b.StateTracker.UpdateLastResumableSourceBinlogPosition(events[len(events)-1].ResumableBinlogPosition())
+		b.StateTracker.UpdateLastResumableSourceBinlogCoordinate(events[len(events)-1].ResumableBinlogCoordinate())
 	}
 
 	b.lastProcessedEventTime = events[len(events)-1].Timestamp()

dml_events.go

@@ -42,14 +42,14 @@ type RowData []interface{}
 // https://github.com/Shopify/ghostferry/issues/165.
 //
 // In summary:
-// - This code receives values from both go-sql-driver/mysql and
-//   go-mysql-org/go-mysql.
-// - go-sql-driver/mysql gives us int64 for signed integer, and uint64 in a byte
-//   slice for unsigned integer.
-// - go-mysql-org/go-mysql gives us int64 for signed integer, and uint64 for
-//   unsigned integer.
-// - We currently make this function deal with both cases. In the future we can
-//   investigate alternative solutions.
+//   - This code receives values from both go-sql-driver/mysql and
+//     go-mysql-org/go-mysql.
+//   - go-sql-driver/mysql gives us int64 for signed integer, and uint64 in a byte
+//     slice for unsigned integer.
+//   - go-mysql-org/go-mysql gives us int64 for signed integer, and uint64 for
+//     unsigned integer.
+//   - We currently make this function deal with both cases. In the future we can
+//     investigate alternative solutions.
 func (r RowData) GetUint64(colIdx int) (uint64, error) {
 	u64, ok := Uint64Value(r[colIdx])
 	if ok {
@@ -88,13 +88,39 @@ type DMLEvent interface {
 	Timestamp() time.Time
 }
 
+// coordinateStamper is an internal capability used by the BinlogStreamer to
+// stamp non-file/position coordinates (e.g. GTID) onto events. It is
+// deliberately NOT part of the exported DMLEvent interface so that external
+// implementations of DMLEvent do not have to implement it. All built-in DML
+// events satisfy it via DMLEventBase.
+type coordinateStamper interface {
+	SetCoordinates(coordinate, resumableCoordinate BinlogCoordinate)
+}
+
 // The base of DMLEvent to provide the necessary methods.
 type DMLEventBase struct {
 	table        *TableSchema
 	pos          mysql.Position
 	resumablePos mysql.Position
 	query        []byte
 	timestamp    time.Time
+
+	// coordinate and resumableCoordinate optionally carry non-file/position
+	// coordinates (e.g. GTID). When set (non-nil), they take precedence over
+	// the file/position fields in the coordinate accessors. They are stamped by
+	// the BinlogStreamer in GTID mode.
+	coordinate          *BinlogCoordinate
+	resumableCoordinate *BinlogCoordinate
+}
+
+// SetCoordinates overrides the coordinate-typed accessors with explicit
+// coordinates. It is used by the BinlogStreamer to stamp GTID coordinates onto
+// events in GTID mode. The file/position accessors are unaffected.
+func (e *DMLEventBase) SetCoordinates(coordinate, resumableCoordinate BinlogCoordinate) {
+	c := coordinate
+	r := resumableCoordinate
+	e.coordinate = &c
+	e.resumableCoordinate = &r
 }
 
 func (e *DMLEventBase) Database() string {
@@ -118,10 +144,16 @@ func (e *DMLEventBase) ResumableBinlogPosition() mysql.Position {
 }
 
 func (e *DMLEventBase) BinlogCoordinate() BinlogCoordinate {
+	if e.coordinate != nil {
+		return *e.coordinate
+	}
 	return NewFilePositionCoordinate(e.pos)
 }
 
 func (e *DMLEventBase) ResumableBinlogCoordinate() BinlogCoordinate {
+	if e.resumableCoordinate != nil {
+		return *e.resumableCoordinate
+	}
 	return NewFilePositionCoordinate(e.resumablePos)
 }
 

ferry.go

@@ -605,19 +605,27 @@ func (f *Ferry) Start() error {
 	// miss some records that are inserted between the time the
 	// DataIterator determines the range of IDs to copy and the time that
 	// the starting binlog coordinates are determined.
-	// Resume-from-state currently only supports file/position coordinates.
-	// GTID-based resume is introduced in a later stage once GTID coordinates are
-	// persisted in the serialized state.
-	if f.StateToResumeFrom != nil && f.Config.BinlogCoordinateMode == BinlogCoordinateGTID {
-		return fmt.Errorf("resuming from state is not yet supported in GTID binlog coordinate mode")
-	}
-
 	var sourceCoord BinlogCoordinate
 	var targetCoord BinlogCoordinate
 
 	var err error
 	if f.StateToResumeFrom != nil {
-		sourceCoord, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlSinceCoordinate(f.StateToResumeFrom.MinSourceBinlogCoordinate())
+		// Fail fast if the serialized state was produced in a different binlog
+		// coordinate mode than the current run; resuming across modes would
+		// mis-route the streamer (e.g. seeking a GTID set against a
+		// file/position stream) and could silently skip or replay events.
+		if stateMode := f.StateToResumeFrom.CoordinateMode(); stateMode != f.Config.BinlogCoordinateMode {
+			return fmt.Errorf(
+				"cannot resume: state was saved in %q binlog coordinate mode but this run is configured for %q",
+				stateMode, f.Config.BinlogCoordinateMode,
+			)
+		}
+
+		resumeCoord, resumeErr := f.StateToResumeFrom.MinSourceBinlogCoordinate()
+		if resumeErr != nil {
+			return fmt.Errorf("computing resume coordinate: %w", resumeErr)
+		}
+		sourceCoord, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlSinceCoordinate(resumeCoord)
 	} else {
 		sourceCoord, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlWithCoordinate()
 	}
@@ -626,9 +634,13 @@ func (f *Ferry) Start() error {
 	}
 
 	if !f.Config.SkipTargetVerification {
-		if f.StateToResumeFrom != nil && f.StateToResumeFrom.LastStoredBinlogPositionForTargetVerifier != zeroPosition {
+		// Presence must be checked explicitly, not via IsZero(): in GTID mode an
+		// empty executed set is a valid target-verifier coordinate, and treating
+		// it as absent would reconnect the target verifier at the target's
+		// current position and skip events that occurred before the interrupt.
+		if f.StateToResumeFrom != nil && f.StateToResumeFrom.HasTargetVerifierBinlogCoordinate() {
 			targetCoord, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysqlSinceCoordinate(
-				NewFilePositionCoordinate(f.StateToResumeFrom.LastStoredBinlogPositionForTargetVerifier),
+				f.StateToResumeFrom.TargetVerifierBinlogCoordinate(),
 			)
 		} else {
 			targetCoord, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysqlWithCoordinate()
@@ -998,9 +1010,11 @@ func (f *Ferry) Progress() *Progress {
 
 	// Binlog Progress
 	s.LastSuccessfulBinlogPos = f.BinlogStreamer.lastStreamedBinlogPosition
+	s.LastSuccessfulBinlogCoordinate = f.BinlogStreamer.GetLastStreamedBinlogCoordinate()
 	s.BinlogStreamerLag = now.Sub(f.BinlogStreamer.lastProcessedEventTime).Seconds()
 	s.BinlogWriterLag = now.Sub(f.BinlogWriter.lastProcessedEventTime).Seconds()
 	s.FinalBinlogPos = f.BinlogStreamer.stopAtBinlogPosition
+	s.FinalBinlogCoordinate = f.BinlogStreamer.GetStopBinlogCoordinate()
 
 	if f.TargetVerifier != nil {
 		s.TargetBinlogStreamerLag = now.Sub(f.TargetVerifier.BinlogStreamer.lastProcessedEventTime).Seconds()

inline_verifier.go

@@ -750,7 +750,7 @@ func (v *InlineVerifier) binlogEventListener(evs []DMLEvent) error {
 	}
 
 	if v.StateTracker != nil {
-		v.StateTracker.UpdateLastResumableSourceBinlogPositionForInlineVerifier(evs[len(evs)-1].ResumableBinlogPosition())
+		v.StateTracker.UpdateLastResumableSourceBinlogCoordinateForInlineVerifier(evs[len(evs)-1].ResumableBinlogCoordinate())
 	}
 
 	return nil

progress.go

@@ -32,9 +32,15 @@ type Progress struct {
 
 	Tables                  map[string]TableProgress
 	LastSuccessfulBinlogPos mysql.Position
-	BinlogStreamerLag       float64 // This is the amount of seconds the binlog streamer is lagging by (seconds)
-	BinlogWriterLag         float64 // This is the amount of seconds the binlog writer is lagging by (seconds)
-	Throttled               bool
+	// LastSuccessfulBinlogCoordinate is the coordinate-typed counterpart of
+	// LastSuccessfulBinlogPos. It reflects the active BinlogCoordinateMode, so
+	// in GTID mode it carries the streamed GTID set. The legacy
+	// LastSuccessfulBinlogPos field is retained for backward compatibility and
+	// is only meaningful in file/position mode.
+	LastSuccessfulBinlogCoordinate BinlogCoordinate
+	BinlogStreamerLag              float64 // This is the amount of seconds the binlog streamer is lagging by (seconds)
+	BinlogWriterLag                float64 // This is the amount of seconds the binlog writer is lagging by (seconds)
+	Throttled                      bool
 
 	// if the TargetVerifier is enabled, we emit this lag, otherwise this number will be 0
 	TargetBinlogStreamerLag float64
@@ -51,6 +57,9 @@ type Progress struct {
 
 	// These are some variables that are only filled when CurrentState == done.
 	FinalBinlogPos mysql.Position
+	// FinalBinlogCoordinate is the coordinate-typed counterpart of
+	// FinalBinlogPos, reflecting the active BinlogCoordinateMode.
+	FinalBinlogCoordinate BinlogCoordinate
 
 	// A best estimate on the speed at which the copying is taking place. If
 	// there are large gaps in the PaginationKey space, this probably will be inaccurate.