From 6196023d8d465facd9cee2758cee800489534a0e Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Tue, 21 Jul 2026 20:51:49 +0200 Subject: [PATCH 1/4] Add GTID binlog streaming start/stop (behind mode flag) Wire the GTID coordinate mode into BinlogStreamer so a fresh run can stream from and stop at a GTID set, while file/position stays the default and unchanged. - BinlogStreamer gains BinlogCoordinateMode plus GTID tracking of the streamed, resumable, and stop GTID sets. - Add ConnectBinlogStreamerToMysqlWithCoordinate and ...FromCoordinate that dispatch to StartSync (file/pos) or StartSyncGTID (gtid), with coordinate-type validation. - Track committed GTID via XIDEvent.GSet and record the pre-transaction set as the resumable point at GTIDEvent; clone to avoid aliasing go-mysql's mutable set. - Replace the Run loop stop condition with shouldContinueStreaming, which uses position compare for file/pos and executed-set containment for GTID. FlushAndStop records the GTID stop set. - Ferry passes the mode to its streamers and uses coordinate-based connect/state updates. GTID resume-from-state is explicitly rejected until GTID state persistence lands. - Add internal unit tests for mode dispatch, stop logic, and validation. File/position behavior is unchanged; GTID streaming works for fresh runs. GTID resume and verifier/progress/replica-catchup wiring follow in later stages. --- binlog_streamer.go | 195 +++++++++++++++++++++++++++++++++-- binlog_streamer_gtid_test.go | 118 +++++++++++++++++++++ ferry.go | 46 +++++---- 3 files changed, 334 insertions(+), 25 deletions(-) create mode 100644 binlog_streamer_gtid_test.go diff --git a/binlog_streamer.go b/binlog_streamer.go index d762370d..77b5f9ff 100644 --- a/binlog_streamer.go +++ b/binlog_streamer.go @@ -45,10 +45,24 @@ type BinlogStreamer struct { DatabaseRewrites map[string]string TableRewrites map[string]string + // BinlogCoordinateMode selects whether this streamer tracks file/position + // or GTID coordinates. An empty value is treated as file/position for + // backwards compatibility. + BinlogCoordinateMode BinlogCoordinateType + lastStreamedBinlogPosition mysql.Position lastResumableBinlogPosition mysql.Position stopAtBinlogPosition mysql.Position + // GTID tracking, only maintained when BinlogCoordinateMode is + // BinlogCoordinateGTID. lastStreamedGTIDSet is the committed GTID set seen + // so far. lastResumableGTIDSet is the committed GTID set at the last + // transaction boundary (a safe resume point). stopAtGTIDSet is the target + // executed set to stop at during cutover. + lastStreamedGTIDSet mysql.GTIDSet + lastResumableGTIDSet mysql.GTIDSet + stopAtGTIDSet mysql.GTIDSet + lastProcessedEventTime time.Time lastLagMetricEmittedTime time.Time @@ -148,6 +162,94 @@ func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlFrom(startFromBinlogPositio return s.lastStreamedBinlogPosition, err } +// coordinateMode returns the effective coordinate mode, treating the empty +// value as file/position for backwards compatibility. +func (s *BinlogStreamer) coordinateMode() BinlogCoordinateType { + if s.BinlogCoordinateMode == "" { + return BinlogCoordinateFilePosition + } + return s.BinlogCoordinateMode +} + +// ConnectBinlogStreamerToMysqlWithCoordinate starts streaming from the current +// server coordinate for the configured BinlogCoordinateMode. It is the +// coordinate-typed counterpart of ConnectBinlogStreamerToMysql. +func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlWithCoordinate() (BinlogCoordinate, error) { + s.ensureLogger() + + switch s.coordinateMode() { + case BinlogCoordinateGTID: + coord, err := ReadCurrentGTIDCoordinate(s.DB) + if err != nil { + s.logger.WithError(err).Error("failed to read current executed GTID set") + return BinlogCoordinate{}, err + } + return s.ConnectBinlogStreamerToMysqlFromCoordinate(coord) + default: + pos, err := s.ConnectBinlogStreamerToMysql() + if err != nil { + return BinlogCoordinate{}, err + } + return NewFilePositionCoordinate(pos), nil + } +} + +// ConnectBinlogStreamerToMysqlFromCoordinate starts streaming from the given +// coordinate. The coordinate type must match the streamer's configured +// BinlogCoordinateMode. +func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlFromCoordinate(startFrom BinlogCoordinate) (BinlogCoordinate, error) { + s.ensureLogger() + + switch s.coordinateMode() { + case BinlogCoordinateGTID: + if !startFrom.IsGTID() { + return BinlogCoordinate{}, fmt.Errorf("binlog streamer in GTID mode requires a GTID coordinate, got %q", startFrom.Type) + } + return s.connectBinlogStreamerFromGTID(startFrom) + default: + if !startFrom.IsFilePosition() { + return BinlogCoordinate{}, fmt.Errorf("binlog streamer in file/position mode requires a file/position coordinate, got %q", startFrom.Type) + } + pos, err := s.ConnectBinlogStreamerToMysqlFrom(startFrom.Position()) + if err != nil { + return BinlogCoordinate{}, err + } + return NewFilePositionCoordinate(pos), nil + } +} + +func (s *BinlogStreamer) connectBinlogStreamerFromGTID(startFrom BinlogCoordinate) (BinlogCoordinate, error) { + err := s.createBinlogSyncer() + if err != nil { + return BinlogCoordinate{}, err + } + + gtidSet, err := startFrom.ParsedGTIDSet() + if err != nil { + s.logger.WithError(err).Error("failed to parse starting GTID set") + return BinlogCoordinate{}, err + } + + // Seed both streamed and resumable GTID sets to the starting set. Clone so + // later mutations from event tracking never alias the starting value. + s.lastStreamedGTIDSet = gtidSet.Clone() + s.lastResumableGTIDSet = gtidSet.Clone() + + s.logger.WithFields(Fields{ + "gtid_set": gtidSet.String(), + "host": s.DBConfig.Host, + "port": s.DBConfig.Port, + }).Info("starting binlog streaming from GTID set") + + s.binlogStreamer, err = s.binlogSyncer.StartSyncGTID(gtidSet) + if err != nil { + s.logger.WithError(err).Error("unable to start binlog streamer from GTID set") + return BinlogCoordinate{}, err + } + + return NewGTIDCoordinate(s.lastStreamedGTIDSet.String()), nil +} + // the default event handler is called for replication binLogEvents that do not have a // separate event Handler registered. @@ -227,6 +329,28 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query // last transaction es.isEventPositionResumable = true + // GTID tracking. go-mysql maintains the current GTID set internally and + // attaches it to XIDEvent.GSet at commit boundaries, so we do not need + // to reconstruct it from raw GTIDEvent SID/GNO. + if s.coordinateMode() == BinlogCoordinateGTID { + switch tev := ev.Event.(type) { + case *replication.GTIDEvent: + // Start of a transaction. A safe resume point is the committed + // set that existed BEFORE this transaction, so that an + // interruption replays the whole in-flight transaction. + if s.lastStreamedGTIDSet != nil { + s.lastResumableGTIDSet = s.lastStreamedGTIDSet.Clone() + } + case *replication.XIDEvent: + // End of a transaction. GSet is the committed GTID set through + // this transaction. Clone to avoid aliasing go-mysql's mutable + // internal set. + if tev.GSet != nil { + s.lastStreamedGTIDSet = tev.GSet.Clone() + } + } + } + // Here we also reset the query event as we are either at the beginning // or the end of the current/next transaction. As such, the query will be // reset following the next RowsQueryEvent before the corresponding RowsEvent(s) @@ -235,6 +359,32 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query return query, err } +// shouldContinueStreaming reports whether the Run loop should keep streaming. +// +// It keeps streaming until a stop has been requested AND the stop coordinate +// has been reached. For file/position that is a position comparison; for GTID +// it is executed-set containment (we have streamed a committed set that +// contains the target stop set). +func (s *BinlogStreamer) shouldContinueStreaming() bool { + if !s.stopRequested { + return true + } + + if s.coordinateMode() == BinlogCoordinateGTID { + if s.stopAtGTIDSet == nil { + // Stop requested but no target recorded yet; keep going until it is. + return true + } + if s.lastStreamedGTIDSet == nil { + return true + } + // Continue while we have NOT yet reached the stop set. + return !s.lastStreamedGTIDSet.Contain(s.stopAtGTIDSet) + } + + return s.lastStreamedBinlogPosition.Compare(s.stopAtBinlogPosition) < 0 +} + func (s *BinlogStreamer) Run() { s.ensureLogger() @@ -242,6 +392,7 @@ func (s *BinlogStreamer) Run() { s.logger.WithFields(Fields{ "stopAtBinlogPosition": s.stopAtBinlogPosition, "lastStreamedBinlogPosition": s.lastStreamedBinlogPosition, + "coordinateMode": s.coordinateMode(), }).Info("exiting binlog streamer") s.binlogSyncer.Close() }() @@ -252,7 +403,7 @@ func (s *BinlogStreamer) Run() { currentFilename := s.lastStreamedBinlogPosition.Name es.nextFilename = s.lastStreamedBinlogPosition.Name s.logger.Info("starting binlog streamer") - for !s.stopRequested || (s.stopRequested && s.lastStreamedBinlogPosition.Compare(s.stopAtBinlogPosition) < 0) { + for s.shouldContinueStreaming() { currentFilename = es.nextFilename var ev *replication.BinlogEvent var timedOut bool @@ -341,8 +492,15 @@ func (s *BinlogStreamer) GetLastStreamedBinlogPosition() mysql.Position { } // GetLastStreamedBinlogCoordinate is the coordinate-typed counterpart of -// GetLastStreamedBinlogPosition. +// GetLastStreamedBinlogPosition. It returns a coordinate matching the +// streamer's configured BinlogCoordinateMode. func (s *BinlogStreamer) GetLastStreamedBinlogCoordinate() BinlogCoordinate { + if s.coordinateMode() == BinlogCoordinateGTID { + if s.lastStreamedGTIDSet == nil { + return NewGTIDCoordinate("") + } + return NewGTIDCoordinate(s.lastStreamedGTIDSet.String()) + } return NewFilePositionCoordinate(s.lastStreamedBinlogPosition) } @@ -352,11 +510,34 @@ func (s *BinlogStreamer) IsAlmostCaughtUp() bool { func (s *BinlogStreamer) FlushAndStop() { s.logger.Info("requesting binlog streamer to stop") - // Must first read the binlog position before requesting stop - // Otherwise there is a race condition where the stopRequested is - // set to True but the TargetPosition is nil, which would cause - // the BinlogStreamer to immediately exit, as it thinks that it has - // passed the initial target position. + // Must first read the stop coordinate before requesting stop. + // Otherwise there is a race condition where stopRequested is set to true + // but the stop coordinate is still nil/zero, which would cause the + // BinlogStreamer to immediately exit, as it thinks that it has already + // passed the stop coordinate. + if s.coordinateMode() == BinlogCoordinateGTID { + err := WithRetries(100, 600*time.Millisecond, s.logger, "read current executed GTID set", func() error { + gtidSet, err := ReadExecutedGTIDSet(s.DB) + if err != nil { + return err + } + parsed, err := mysql.ParseMysqlGTIDSet(gtidSet) + if err != nil { + return err + } + s.stopAtGTIDSet = parsed + return nil + }) + + if err != nil { + s.ErrorHandler.Fatal("binlog_streamer", err) + } + s.logger.WithField("stop_at_gtid_set", s.stopAtGTIDSet.String()).Info("current stop GTID set was recorded") + + s.stopRequested = true + return + } + err := WithRetries(100, 600*time.Millisecond, s.logger, "read current binlog position", func() error { var err error s.stopAtBinlogPosition, err = ShowMasterStatusBinlogPosition(s.DB) diff --git a/binlog_streamer_gtid_test.go b/binlog_streamer_gtid_test.go new file mode 100644 index 00000000..fd22a61e --- /dev/null +++ b/binlog_streamer_gtid_test.go @@ -0,0 +1,118 @@ +package ghostferry + +import ( + "testing" + + "github.com/go-mysql-org/go-mysql/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + gtidSetLower = "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-57" + gtidSetTarget = "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-100" + gtidSetPast = "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-150" +) + +func mustParseGTID(t *testing.T, s string) mysql.GTIDSet { + t.Helper() + set, err := mysql.ParseMysqlGTIDSet(s) + require.NoError(t, err) + return set +} + +func TestCoordinateModeDefaultsToFilePosition(t *testing.T) { + s := &BinlogStreamer{} + assert.Equal(t, BinlogCoordinateFilePosition, s.coordinateMode()) + + s.BinlogCoordinateMode = BinlogCoordinateGTID + assert.Equal(t, BinlogCoordinateGTID, s.coordinateMode()) +} + +func TestShouldContinueStreaming_FilePosition(t *testing.T) { + s := &BinlogStreamer{} + + // No stop requested: always continue. + assert.True(t, s.shouldContinueStreaming()) + + s.stopRequested = true + s.stopAtBinlogPosition = mysql.Position{Name: "mysql-bin.000010", Pos: 100} + + // Streamed position behind stop: continue. + s.lastStreamedBinlogPosition = mysql.Position{Name: "mysql-bin.000010", Pos: 50} + assert.True(t, s.shouldContinueStreaming()) + + // Streamed position reached stop: stop. + s.lastStreamedBinlogPosition = mysql.Position{Name: "mysql-bin.000010", Pos: 100} + assert.False(t, s.shouldContinueStreaming()) +} + +func TestShouldContinueStreaming_GTID(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + + // No stop requested: always continue. + assert.True(t, s.shouldContinueStreaming()) + + s.stopRequested = true + + // Stop requested but no target recorded yet: keep going. + assert.True(t, s.shouldContinueStreaming()) + + s.stopAtGTIDSet = mustParseGTID(t, gtidSetTarget) + + // No streamed set yet: keep going. + assert.True(t, s.shouldContinueStreaming()) + + // Streamed set does not yet contain target: continue. + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetLower) + assert.True(t, s.shouldContinueStreaming()) + + // Streamed set exactly reaches target: stop. + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetTarget) + assert.False(t, s.shouldContinueStreaming()) + + // Streamed set past target: stop. + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetPast) + assert.False(t, s.shouldContinueStreaming()) +} + +func TestGetLastStreamedBinlogCoordinate_GTIDMode(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + + // Nil streamed set yields an empty (zero) GTID coordinate. + coord := s.GetLastStreamedBinlogCoordinate() + assert.True(t, coord.IsGTID()) + assert.True(t, coord.IsZero()) + + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetTarget) + coord = s.GetLastStreamedBinlogCoordinate() + assert.True(t, coord.IsGTID()) + assert.Equal(t, gtidSetTarget, coord.GTIDSet) +} + +func TestGetLastStreamedBinlogCoordinate_FilePositionMode(t *testing.T) { + s := &BinlogStreamer{} + s.lastStreamedBinlogPosition = mysql.Position{Name: "mysql-bin.000010", Pos: 100} + + coord := s.GetLastStreamedBinlogCoordinate() + assert.True(t, coord.IsFilePosition()) + assert.Equal(t, "mysql-bin.000010", coord.Position().Name) + assert.Equal(t, uint32(100), coord.Position().Pos) +} + +func TestConnectBinlogStreamerFromCoordinate_TypeMismatch(t *testing.T) { + // GTID mode with a file/position coordinate must be rejected before any DB + // interaction. + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + _, err := s.ConnectBinlogStreamerToMysqlFromCoordinate( + NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 4}), + ) + require.Error(t, err) + assert.Contains(t, err.Error(), "GTID mode requires a GTID coordinate") + + // File/position mode with a GTID coordinate must also be rejected. + s2 := &BinlogStreamer{} + _, err = s2.ConnectBinlogStreamerToMysqlFromCoordinate(NewGTIDCoordinate(gtidSetTarget)) + require.Error(t, err) + assert.Contains(t, err.Error(), "file/position mode requires a file/position coordinate") +} diff --git a/ferry.go b/ferry.go index 9666576c..95fa5c78 100644 --- a/ferry.go +++ b/ferry.go @@ -147,15 +147,16 @@ func (f *Ferry) newBinlogStreamer(db *sql.DB, dbConf *DatabaseConfig, schemaRewr f.ensureInitialized() return &BinlogStreamer{ - DB: db, - DBConfig: dbConf, - MyServerId: f.Config.MyServerId, - ErrorHandler: f.ErrorHandler, - Filter: f.CopyFilter, - TableSchema: f.Tables, - LogTag: logTag, - DatabaseRewrites: schemaRewrites, - TableRewrites: tableRewrites, + DB: db, + DBConfig: dbConf, + MyServerId: f.Config.MyServerId, + ErrorHandler: f.ErrorHandler, + Filter: f.CopyFilter, + TableSchema: f.Tables, + LogTag: logTag, + DatabaseRewrites: schemaRewrites, + TableRewrites: tableRewrites, + BinlogCoordinateMode: f.Config.BinlogCoordinateMode, } } @@ -588,14 +589,21 @@ 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. - var sourcePos siddontangmysql.Position - var targetPos siddontangmysql.Position + // 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 { - sourcePos, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlFrom(f.StateToResumeFrom.MinSourceBinlogPosition()) + sourceCoord, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlFromCoordinate(f.StateToResumeFrom.MinSourceBinlogCoordinate()) } else { - sourcePos, err = f.BinlogStreamer.ConnectBinlogStreamerToMysql() + sourceCoord, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlWithCoordinate() } if err != nil { return err @@ -603,9 +611,11 @@ func (f *Ferry) Start() error { if !f.Config.SkipTargetVerification { if f.StateToResumeFrom != nil && f.StateToResumeFrom.LastStoredBinlogPositionForTargetVerifier != zeroPosition { - targetPos, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysqlFrom(f.StateToResumeFrom.LastStoredBinlogPositionForTargetVerifier) + targetCoord, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysqlFromCoordinate( + NewFilePositionCoordinate(f.StateToResumeFrom.LastStoredBinlogPositionForTargetVerifier), + ) } else { - targetPos, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysql() + targetCoord, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysqlWithCoordinate() } } if err != nil { @@ -615,13 +625,13 @@ func (f *Ferry) Start() error { // If we don't set this now, there is a race condition where Ghostferry // is terminated with some rows copied but no binlog events are written. // This guarentees that we are able to restart from a valid location. - f.StateTracker.UpdateLastResumableSourceBinlogPosition(sourcePos) + f.StateTracker.UpdateLastResumableSourceBinlogCoordinate(sourceCoord) if f.inlineVerifier != nil { - f.StateTracker.UpdateLastResumableSourceBinlogPositionForInlineVerifier(sourcePos) + f.StateTracker.UpdateLastResumableSourceBinlogCoordinateForInlineVerifier(sourceCoord) } if !f.Config.SkipTargetVerification { - f.StateTracker.UpdateLastResumableBinlogPositionForTargetVerifier(targetPos) + f.StateTracker.UpdateLastResumableBinlogCoordinateForTargetVerifier(targetCoord) } return nil From 63abedcb326f418bbccdf70fd083b3f28e559773 Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Wed, 22 Jul 2026 02:10:58 +0200 Subject: [PATCH 2/4] Drive stream-stop check through coordinate HasReached shouldContinueStreaming no longer branches on coordinate type: it asks GetLastStreamedBinlogCoordinate().HasReached(GetStopBinlogCoordinate()), delegating the file/position-vs-GTID mechanics to BinlogCoordinate. Add GetStopBinlogCoordinate so both the streamed and stop roles are exposed as coordinates. A zero stop coordinate means "no target recorded yet", matching the previous guard. --- binlog_streamer.go | 43 +++++++++++++++++++++++++++++-------------- 1 file changed, 29 insertions(+), 14 deletions(-) diff --git a/binlog_streamer.go b/binlog_streamer.go index 77b5f9ff..d018d70c 100644 --- a/binlog_streamer.go +++ b/binlog_streamer.go @@ -362,27 +362,29 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query // shouldContinueStreaming reports whether the Run loop should keep streaming. // // It keeps streaming until a stop has been requested AND the stop coordinate -// has been reached. For file/position that is a position comparison; for GTID -// it is executed-set containment (we have streamed a committed set that -// contains the target stop set). +// has been reached. The "have we reached the stop coordinate?" question is +// answered by BinlogCoordinate.HasReached, so this method is identical for +// file/position and GTID; the representation-specific mechanics live on the +// coordinate type. func (s *BinlogStreamer) shouldContinueStreaming() bool { if !s.stopRequested { return true } - if s.coordinateMode() == BinlogCoordinateGTID { - if s.stopAtGTIDSet == nil { - // Stop requested but no target recorded yet; keep going until it is. - return true - } - if s.lastStreamedGTIDSet == nil { - return true - } - // Continue while we have NOT yet reached the stop set. - return !s.lastStreamedGTIDSet.Contain(s.stopAtGTIDSet) + stop := s.GetStopBinlogCoordinate() + if stop.IsZero() { + // Stop requested but no target recorded yet; keep going until it is. + return true } - return s.lastStreamedBinlogPosition.Compare(s.stopAtBinlogPosition) < 0 + reached, err := s.GetLastStreamedBinlogCoordinate().HasReached(stop) + if err != nil { + // A mismatch or parse error should not silently stop the stream; log + // and keep going so a spurious error can't truncate replication. + s.logger.WithError(err).Warn("could not evaluate stop coordinate; continuing to stream") + return true + } + return !reached } func (s *BinlogStreamer) Run() { @@ -504,6 +506,19 @@ func (s *BinlogStreamer) GetLastStreamedBinlogCoordinate() BinlogCoordinate { return NewFilePositionCoordinate(s.lastStreamedBinlogPosition) } +// GetStopBinlogCoordinate returns the recorded stop coordinate matching the +// streamer's configured BinlogCoordinateMode. It is zero until FlushAndStop has +// recorded a stop target. +func (s *BinlogStreamer) GetStopBinlogCoordinate() BinlogCoordinate { + if s.coordinateMode() == BinlogCoordinateGTID { + if s.stopAtGTIDSet == nil { + return NewGTIDCoordinate("") + } + return NewGTIDCoordinate(s.stopAtGTIDSet.String()) + } + return NewFilePositionCoordinate(s.stopAtBinlogPosition) +} + func (s *BinlogStreamer) IsAlmostCaughtUp() bool { return time.Now().Sub(s.lastProcessedEventTime) < caughtUpThreshold } From c1095e74ef1f64b8f6c126bc85ff8aca599f826e Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Wed, 22 Jul 2026 15:47:39 +0200 Subject: [PATCH 3/4] Rename ...FromCoordinate to ...SinceCoordinate "From" read too much like the "With" (start-from-current) variant. "Since" makes it clear the streamer starts since the given coordinate. --- binlog_streamer.go | 6 +++--- binlog_streamer_gtid_test.go | 6 +++--- ferry.go | 4 ++-- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/binlog_streamer.go b/binlog_streamer.go index d018d70c..abf09a0c 100644 --- a/binlog_streamer.go +++ b/binlog_streamer.go @@ -184,7 +184,7 @@ func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlWithCoordinate() (BinlogCoo s.logger.WithError(err).Error("failed to read current executed GTID set") return BinlogCoordinate{}, err } - return s.ConnectBinlogStreamerToMysqlFromCoordinate(coord) + return s.ConnectBinlogStreamerToMysqlSinceCoordinate(coord) default: pos, err := s.ConnectBinlogStreamerToMysql() if err != nil { @@ -194,10 +194,10 @@ func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlWithCoordinate() (BinlogCoo } } -// ConnectBinlogStreamerToMysqlFromCoordinate starts streaming from the given +// ConnectBinlogStreamerToMysqlSinceCoordinate starts streaming since the given // coordinate. The coordinate type must match the streamer's configured // BinlogCoordinateMode. -func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlFromCoordinate(startFrom BinlogCoordinate) (BinlogCoordinate, error) { +func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlSinceCoordinate(startFrom BinlogCoordinate) (BinlogCoordinate, error) { s.ensureLogger() switch s.coordinateMode() { diff --git a/binlog_streamer_gtid_test.go b/binlog_streamer_gtid_test.go index fd22a61e..6c608312 100644 --- a/binlog_streamer_gtid_test.go +++ b/binlog_streamer_gtid_test.go @@ -100,11 +100,11 @@ func TestGetLastStreamedBinlogCoordinate_FilePositionMode(t *testing.T) { assert.Equal(t, uint32(100), coord.Position().Pos) } -func TestConnectBinlogStreamerFromCoordinate_TypeMismatch(t *testing.T) { +func TestConnectBinlogStreamerSinceCoordinate_TypeMismatch(t *testing.T) { // GTID mode with a file/position coordinate must be rejected before any DB // interaction. s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} - _, err := s.ConnectBinlogStreamerToMysqlFromCoordinate( + _, err := s.ConnectBinlogStreamerToMysqlSinceCoordinate( NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 4}), ) require.Error(t, err) @@ -112,7 +112,7 @@ func TestConnectBinlogStreamerFromCoordinate_TypeMismatch(t *testing.T) { // File/position mode with a GTID coordinate must also be rejected. s2 := &BinlogStreamer{} - _, err = s2.ConnectBinlogStreamerToMysqlFromCoordinate(NewGTIDCoordinate(gtidSetTarget)) + _, err = s2.ConnectBinlogStreamerToMysqlSinceCoordinate(NewGTIDCoordinate(gtidSetTarget)) require.Error(t, err) assert.Contains(t, err.Error(), "file/position mode requires a file/position coordinate") } diff --git a/ferry.go b/ferry.go index 95fa5c78..010ecc14 100644 --- a/ferry.go +++ b/ferry.go @@ -601,7 +601,7 @@ func (f *Ferry) Start() error { var err error if f.StateToResumeFrom != nil { - sourceCoord, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlFromCoordinate(f.StateToResumeFrom.MinSourceBinlogCoordinate()) + sourceCoord, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlSinceCoordinate(f.StateToResumeFrom.MinSourceBinlogCoordinate()) } else { sourceCoord, err = f.BinlogStreamer.ConnectBinlogStreamerToMysqlWithCoordinate() } @@ -611,7 +611,7 @@ func (f *Ferry) Start() error { if !f.Config.SkipTargetVerification { if f.StateToResumeFrom != nil && f.StateToResumeFrom.LastStoredBinlogPositionForTargetVerifier != zeroPosition { - targetCoord, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysqlFromCoordinate( + targetCoord, err = f.TargetVerifier.BinlogStreamer.ConnectBinlogStreamerToMysqlSinceCoordinate( NewFilePositionCoordinate(f.StateToResumeFrom.LastStoredBinlogPositionForTargetVerifier), ) } else { From e3d40fbc4591cadb8585d8750a309ed27a2b4be2 Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Wed, 22 Jul 2026 20:45:51 +0200 Subject: [PATCH 4/4] Fix GTID stop hangs and validate gtid_mode on connect - shouldContinueStreaming no longer treats an empty stop coordinate as "unset". FlushAndStop always records the stop target before setting stopRequested, so an empty executed GTID set on a fresh source is a valid target that is immediately reached, rather than hanging cutover. - Advance the streamed GTID set on DDL/admin QueryEvents (which commit without an XIDEvent), skipping transaction-control statements (BEGIN). Previously a cutover whose stop target ended in a DDL could hang. - Validate @@GLOBAL.gtid_mode=ON for the source (and target when target verification is enabled) during Ferry.Initialize when in GTID mode, so misconfiguration fails fast instead of late during streaming. - Add unit tests for the empty stop target, QueryEvent advancement, and transaction-control detection. --- binlog_streamer.go | 44 ++++++++++++++++++++++--- binlog_streamer_gtid_test.go | 63 +++++++++++++++++++++++++++++++++--- ferry.go | 16 +++++++++ 3 files changed, 115 insertions(+), 8 deletions(-) diff --git a/binlog_streamer.go b/binlog_streamer.go index abf09a0c..0dea42ab 100644 --- a/binlog_streamer.go +++ b/binlog_streamer.go @@ -6,6 +6,7 @@ import ( sqlorig "database/sql" "errors" "fmt" + "strings" "time" sql "github.com/Shopify/ghostferry/sqlwrapper" @@ -302,6 +303,29 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query s.logger.WithError(err).Error("failed to handle rows event") s.ErrorHandler.Fatal("binlog_streamer", err) } + case *replication.QueryEvent: + // DDL and administrative statements (CREATE TABLE, GRANT, etc.) commit + // via a QueryEvent rather than an XIDEvent, so their GTID is only + // visible here. Without this, the streamed GTID set would never + // advance past such a statement and a cutover whose stop target + // includes it would hang forever. go-mysql attaches the current + // executed GTID set (including this statement's GTID) to the event. + // + // Transaction-control statements ("BEGIN") also arrive as QueryEvents + // but do NOT commit anything; their GTID is captured at the closing + // XIDEvent instead, so we must skip them here to avoid advancing the + // streamed set before the transaction's rows have been applied. + if s.coordinateMode() == BinlogCoordinateGTID { + qe := ev.Event.(*replication.QueryEvent) + if qe.GSet != nil && !isTransactionControlQuery(qe.Query) { + // A DDL/admin statement is its own transaction; the pre-statement + // committed set is a safe resume point. + if s.lastStreamedGTIDSet != nil { + s.lastResumableGTIDSet = s.lastStreamedGTIDSet.Clone() + } + s.lastStreamedGTIDSet = qe.GSet.Clone() + } + } case *replication.XIDEvent, *replication.GTIDEvent: // With regards to DMLs, we see (at least) the following sequence // of events in the binlog stream: @@ -371,11 +395,13 @@ func (s *BinlogStreamer) shouldContinueStreaming() bool { return true } + // Once stopRequested is set, FlushAndStop has already recorded the stop + // coordinate. We must NOT treat a zero/empty stop coordinate as "not yet + // recorded": on a fresh source the executed GTID set (or binlog position) + // can legitimately be empty, and an empty GTID set is a valid stop target + // that any streamed set already contains. Deriving presence from IsZero() + // here would hang cutover forever in that case. stop := s.GetStopBinlogCoordinate() - if stop.IsZero() { - // Stop requested but no target recorded yet; keep going until it is. - return true - } reached, err := s.GetLastStreamedBinlogCoordinate().HasReached(stop) if err != nil { @@ -519,6 +545,16 @@ func (s *BinlogStreamer) GetStopBinlogCoordinate() BinlogCoordinate { return NewFilePositionCoordinate(s.stopAtBinlogPosition) } +// isTransactionControlQuery reports whether a QueryEvent query is a +// transaction-control statement that does not itself commit data (BEGIN). +// Such statements must not advance the committed GTID set; the enclosing +// transaction commits at its XIDEvent. Note COMMIT/ROLLBACK are normally +// represented as XIDEvents for InnoDB, but are treated defensively here too. +func isTransactionControlQuery(query []byte) bool { + q := strings.ToUpper(strings.TrimSpace(string(query))) + return q == "BEGIN" || q == "COMMIT" || q == "ROLLBACK" +} + func (s *BinlogStreamer) IsAlmostCaughtUp() bool { return time.Now().Sub(s.lastProcessedEventTime) < caughtUpThreshold } diff --git a/binlog_streamer_gtid_test.go b/binlog_streamer_gtid_test.go index 6c608312..178c59e5 100644 --- a/binlog_streamer_gtid_test.go +++ b/binlog_streamer_gtid_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/go-mysql-org/go-mysql/mysql" + "github.com/go-mysql-org/go-mysql/replication" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -21,6 +22,47 @@ func mustParseGTID(t *testing.T, s string) mysql.GTIDSet { return set } +func TestIsTransactionControlQuery(t *testing.T) { + assert.True(t, isTransactionControlQuery([]byte("BEGIN"))) + assert.True(t, isTransactionControlQuery([]byte(" begin "))) + assert.True(t, isTransactionControlQuery([]byte("COMMIT"))) + assert.True(t, isTransactionControlQuery([]byte("ROLLBACK"))) + assert.False(t, isTransactionControlQuery([]byte("CREATE TABLE t (id int)"))) + assert.False(t, isTransactionControlQuery([]byte("GRANT ALL ON *.* TO 'x'@'%'"))) +} + +// TestQueryEventAdvancesStreamedGTID verifies that a DDL/admin QueryEvent +// (which commits without an XIDEvent) advances the streamed GTID set, while a +// BEGIN QueryEvent does not. Without this, a cutover whose stop target includes +// a trailing DDL would hang forever. +func TestQueryEventAdvancesStreamedGTID(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + s.logger = LogWithField("tag", "test") + + ddlSet := mustParseGTID(t, gtidSetTarget) + + // A DDL QueryEvent carrying the executed set advances lastStreamedGTIDSet. + ddlEvent := &replication.BinlogEvent{ + Header: &replication.EventHeader{LogPos: 100}, + Event: &replication.QueryEvent{Query: []byte("CREATE TABLE t (id int)"), GSet: ddlSet}, + } + es := &BinlogEventState{} + _, err := s.defaultEventHandler(ddlEvent, nil, es) + require.NoError(t, err) + require.NotNil(t, s.lastStreamedGTIDSet) + assert.Equal(t, gtidSetTarget, s.lastStreamedGTIDSet.String()) + + // A BEGIN QueryEvent must NOT advance the streamed set. + before := s.lastStreamedGTIDSet.String() + beginEvent := &replication.BinlogEvent{ + Header: &replication.EventHeader{LogPos: 200}, + Event: &replication.QueryEvent{Query: []byte("BEGIN"), GSet: mustParseGTID(t, gtidSetPast)}, + } + _, err = s.defaultEventHandler(beginEvent, nil, es) + require.NoError(t, err) + assert.Equal(t, before, s.lastStreamedGTIDSet.String(), "BEGIN must not advance the streamed GTID set") +} + func TestCoordinateModeDefaultsToFilePosition(t *testing.T) { s := &BinlogStreamer{} assert.Equal(t, BinlogCoordinateFilePosition, s.coordinateMode()) @@ -54,10 +96,6 @@ func TestShouldContinueStreaming_GTID(t *testing.T) { assert.True(t, s.shouldContinueStreaming()) s.stopRequested = true - - // Stop requested but no target recorded yet: keep going. - assert.True(t, s.shouldContinueStreaming()) - s.stopAtGTIDSet = mustParseGTID(t, gtidSetTarget) // No streamed set yet: keep going. @@ -76,6 +114,23 @@ func TestShouldContinueStreaming_GTID(t *testing.T) { assert.False(t, s.shouldContinueStreaming()) } +// TestShouldContinueStreaming_GTIDEmptyStopTarget guards the fresh-source +// case: an empty executed GTID set is a valid stop target (not "unset"), and +// any streamed set — including an empty one — has already reached it, so the +// stream must stop rather than hang. +func TestShouldContinueStreaming_GTIDEmptyStopTarget(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + s.stopRequested = true + s.stopAtGTIDSet = mustParseGTID(t, "") // empty executed set on a fresh source + + // Empty streamed set has reached the empty stop target: stop. + assert.False(t, s.shouldContinueStreaming()) + + // A non-empty streamed set also trivially contains the empty target: stop. + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetTarget) + assert.False(t, s.shouldContinueStreaming()) +} + func TestGetLastStreamedBinlogCoordinate_GTIDMode(t *testing.T) { s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} diff --git a/ferry.go b/ferry.go index 010ecc14..8087a91a 100644 --- a/ferry.go +++ b/ferry.go @@ -379,6 +379,13 @@ func (f *Ferry) Initialize() (err error) { return err } + if f.Config.BinlogCoordinateMode == BinlogCoordinateGTID { + if err = CheckServerGTIDModeEnabled(f.SourceDB); err != nil { + f.logger.WithError(err).Error("source db is not configured for GTID mode") + return err + } + } + f.TargetDB, err = f.Target.SqlDB(f.logger.WithField("dbname", "target")) if err != nil { f.logger.WithError(err).Error("failed to connect to target database") @@ -391,6 +398,15 @@ func (f *Ferry) Initialize() (err error) { return err } + // The target binlog streamer (used by the target verifier) streams in the + // same coordinate mode, so the target must also have GTID mode enabled. + if f.Config.BinlogCoordinateMode == BinlogCoordinateGTID && !f.Config.SkipTargetVerification { + if err = CheckServerGTIDModeEnabled(f.TargetDB); err != nil { + f.logger.WithError(err).Error("target db is not configured for GTID mode") + return err + } + } + if !f.Config.AllowSuperUserOnReadOnly { isReplica, err := CheckDbIsAReplica(f.TargetDB) if err != nil {