diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 39cde138..3076c9a7 100644 --- a/.github/workflows/tests.yml +++ b/.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 diff --git a/binlog_coordinate.go b/binlog_coordinate.go index 0d968ec4..2a989fdb 100644 --- a/binlog_coordinate.go +++ b/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 diff --git a/binlog_streamer.go b/binlog_streamer.go index 0dea42ab..1a1e9b14 100644 --- a/binlog_streamer.go +++ b/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 { diff --git a/binlog_writer.go b/binlog_writer.go index 6292700e..09528b74 100644 --- a/binlog_writer.go +++ b/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() diff --git a/dml_events.go b/dml_events.go index d0a2df9c..83e87f5e 100644 --- a/dml_events.go +++ b/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,6 +88,15 @@ 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 @@ -95,6 +104,23 @@ type DMLEventBase struct { 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) } diff --git a/ferry.go b/ferry.go index 8087a91a..ea8cf676 100644 --- a/ferry.go +++ b/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() diff --git a/inline_verifier.go b/inline_verifier.go index d69e48ff..909b3f90 100644 --- a/inline_verifier.go +++ b/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 diff --git a/progress.go b/progress.go index 85105452..db920417 100644 --- a/progress.go +++ b/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. diff --git a/state_tracker.go b/state_tracker.go index 6f974cb8..507d71c7 100644 --- a/state_tracker.go +++ b/state_tracker.go @@ -3,6 +3,7 @@ package ghostferry import ( "container/ring" "encoding/json" + "fmt" "sync" "time" @@ -40,6 +41,17 @@ type SerializableState struct { BinlogVerifyStore BinlogVerifySerializedStore LastStoredBinlogPositionForInlineVerifier mysql.Position LastStoredBinlogPositionForTargetVerifier mysql.Position + + // BinlogCoordinateMode records which coordinate representation the state + // below was produced with. Empty means file/position (legacy states). + BinlogCoordinateMode BinlogCoordinateType `json:",omitempty"` + + // GTID counterparts of the three binlog positions above. They are only + // populated when BinlogCoordinateMode is "gtid". Pointers + omitempty keep + // file/position states byte-for-byte compatible with older versions. + LastWrittenBinlogCoordinate *BinlogCoordinate `json:",omitempty"` + LastStoredBinlogCoordinateForInlineVerifier *BinlogCoordinate `json:",omitempty"` + LastStoredBinlogCoordinateForTargetVerifier *BinlogCoordinate `json:",omitempty"` } func (s *SerializableState) MarshalJSON() ([]byte, error) { @@ -106,11 +118,105 @@ func (s *SerializableState) MinSourceBinlogPosition() mysql.Position { } } -// MinSourceBinlogCoordinate returns the safe source resume coordinate. It is -// the coordinate-typed counterpart of MinSourceBinlogPosition and currently -// delegates to it, wrapping the result as a file/position coordinate. -func (s *SerializableState) MinSourceBinlogCoordinate() BinlogCoordinate { - return NewFilePositionCoordinate(s.MinSourceBinlogPosition()) +// CoordinateMode returns the effective coordinate mode for this serialized +// state, treating the empty value as file/position for legacy states. +func (s *SerializableState) CoordinateMode() BinlogCoordinateType { + if s.BinlogCoordinateMode == "" { + return BinlogCoordinateFilePosition + } + return s.BinlogCoordinateMode +} + +// coordinateMode is the unexported alias kept for existing internal callers. +func (s *SerializableState) coordinateMode() BinlogCoordinateType { + return s.CoordinateMode() +} + +// HasTargetVerifierBinlogCoordinate reports whether the serialized state +// carries a target-verifier resume coordinate at all. This is distinct from +// whether that coordinate is the zero/empty value: in GTID mode an empty +// executed set is a valid coordinate, so presence must be checked via the +// pointer/field rather than IsZero(). +func (s *SerializableState) HasTargetVerifierBinlogCoordinate() bool { + if s.coordinateMode() == BinlogCoordinateGTID { + return s.LastStoredBinlogCoordinateForTargetVerifier != nil + } + return s.LastStoredBinlogPositionForTargetVerifier != (mysql.Position{}) +} + +// WrittenSourceBinlogCoordinate returns the last written source coordinate as a +// BinlogCoordinate, matching the state's coordinate mode. +func (s *SerializableState) WrittenSourceBinlogCoordinate() BinlogCoordinate { + if s.coordinateMode() == BinlogCoordinateGTID && s.LastWrittenBinlogCoordinate != nil { + return *s.LastWrittenBinlogCoordinate + } + return NewFilePositionCoordinate(s.LastWrittenBinlogPosition) +} + +// InlineVerifierSourceBinlogCoordinate returns the inline verifier's stored +// source coordinate as a BinlogCoordinate, matching the state's coordinate mode. +func (s *SerializableState) InlineVerifierSourceBinlogCoordinate() BinlogCoordinate { + if s.coordinateMode() == BinlogCoordinateGTID && s.LastStoredBinlogCoordinateForInlineVerifier != nil { + return *s.LastStoredBinlogCoordinateForInlineVerifier + } + return NewFilePositionCoordinate(s.LastStoredBinlogPositionForInlineVerifier) +} + +// TargetVerifierBinlogCoordinate returns the target verifier's stored +// coordinate as a BinlogCoordinate, matching the state's coordinate mode. +func (s *SerializableState) TargetVerifierBinlogCoordinate() BinlogCoordinate { + if s.coordinateMode() == BinlogCoordinateGTID && s.LastStoredBinlogCoordinateForTargetVerifier != nil { + return *s.LastStoredBinlogCoordinateForTargetVerifier + } + return NewFilePositionCoordinate(s.LastStoredBinlogPositionForTargetVerifier) +} + +// MinSourceBinlogCoordinate returns the safe source resume coordinate. +// +// For file/position mode this is the coordinate-typed counterpart of +// MinSourceBinlogPosition (the earlier of the writer and inline-verifier +// positions). For GTID mode the safe resume point is the intersection of the +// writer and inline-verifier GTID sets, since resuming must not skip events +// either consumer had not yet durably processed. When only one side is present, +// that side is used. +// +// It is fail-closed for GTID: a parse or intersection failure returns an error +// rather than silently falling back to one side, because a wrong (too-advanced) +// resume floor would skip binlog events and corrupt the copy. +func (s *SerializableState) MinSourceBinlogCoordinate() (BinlogCoordinate, error) { + if s.coordinateMode() != BinlogCoordinateGTID { + return NewFilePositionCoordinate(s.MinSourceBinlogPosition()), nil + } + + written := s.LastWrittenBinlogCoordinate + inline := s.LastStoredBinlogCoordinateForInlineVerifier + + if written == nil && inline == nil { + return NewGTIDCoordinate(""), nil + } + if written == nil { + return *inline, nil + } + if inline == nil { + return *written, nil + } + + // Safe resume is the intersection: only GTIDs that both the writer and the + // inline verifier have durably processed can be skipped on resume. + writtenSet, err := written.ParsedGTIDSet() + if err != nil { + return BinlogCoordinate{}, fmt.Errorf("parsing written GTID coordinate %q: %w", written.GTIDSet, err) + } + inlineSet, err := inline.ParsedGTIDSet() + if err != nil { + return BinlogCoordinate{}, fmt.Errorf("parsing inline-verifier GTID coordinate %q: %w", inline.GTIDSet, err) + } + + intersection, err := intersectGTIDSets(writtenSet, inlineSet) + if err != nil { + return BinlogCoordinate{}, fmt.Errorf("computing safe GTID resume floor: %w", err) + } + return NewGTIDCoordinateFromSet(intersection), nil } // For tracking the speed of the copy @@ -146,6 +252,13 @@ type StateTracker struct { lastStoredBinlogPositionForInlineVerifier mysql.Position lastStoredBinlogPositionForTargetVerifier mysql.Position + // GTID coordinates, only populated when the tracker operates in GTID mode. + // They are stored alongside (not instead of) the file/position fields so + // that switching the read path is a mode decision, not a data migration. + lastWrittenBinlogCoordinate *BinlogCoordinate + lastStoredBinlogCoordinateForInlineVerifier *BinlogCoordinate + lastStoredBinlogCoordinateForTargetVerifier *BinlogCoordinate + lastSuccessfulPaginationKeys map[string]PaginationKey completedTables map[string]bool @@ -176,6 +289,9 @@ func NewStateTrackerFromSerializedState(speedLogCount int, serializedState *Seri s.lastWrittenBinlogPosition = serializedState.LastWrittenBinlogPosition s.lastStoredBinlogPositionForInlineVerifier = serializedState.LastStoredBinlogPositionForInlineVerifier s.lastStoredBinlogPositionForTargetVerifier = serializedState.LastStoredBinlogPositionForTargetVerifier + s.lastWrittenBinlogCoordinate = serializedState.LastWrittenBinlogCoordinate + s.lastStoredBinlogCoordinateForInlineVerifier = serializedState.LastStoredBinlogCoordinateForInlineVerifier + s.lastStoredBinlogCoordinateForTargetVerifier = serializedState.LastStoredBinlogCoordinateForTargetVerifier return s } @@ -202,20 +318,41 @@ func (s *StateTracker) UpdateLastResumableBinlogPositionForTargetVerifier(pos my // Coordinate-based accessors and mutators. // -// These are the forward-looking API. They currently delegate to the -// file/position storage above so behavior is unchanged, but they let callers be -// written in terms of BinlogCoordinate ahead of introducing a GTID coordinate -// mode. +// These are the forward-looking API. For file/position coordinates they store +// into the existing file/position fields (unchanged behavior). For GTID +// coordinates they store into dedicated GTID fields, so both representations +// can coexist and the read path is selected by coordinate mode. func (s *StateTracker) UpdateLastResumableSourceBinlogCoordinate(coord BinlogCoordinate) { + if coord.IsGTID() { + s.BinlogRWMutex.Lock() + defer s.BinlogRWMutex.Unlock() + c := coord + s.lastWrittenBinlogCoordinate = &c + return + } s.UpdateLastResumableSourceBinlogPosition(coord.Position()) } func (s *StateTracker) UpdateLastResumableSourceBinlogCoordinateForInlineVerifier(coord BinlogCoordinate) { + if coord.IsGTID() { + s.BinlogRWMutex.Lock() + defer s.BinlogRWMutex.Unlock() + c := coord + s.lastStoredBinlogCoordinateForInlineVerifier = &c + return + } s.UpdateLastResumableSourceBinlogPositionForInlineVerifier(coord.Position()) } func (s *StateTracker) UpdateLastResumableBinlogCoordinateForTargetVerifier(coord BinlogCoordinate) { + if coord.IsGTID() { + s.BinlogRWMutex.Lock() + defer s.BinlogRWMutex.Unlock() + c := coord + s.lastStoredBinlogCoordinateForTargetVerifier = &c + return + } s.UpdateLastResumableBinlogPositionForTargetVerifier(coord.Position()) } @@ -223,6 +360,9 @@ func (s *StateTracker) LastResumableSourceBinlogCoordinate() BinlogCoordinate { s.BinlogRWMutex.RLock() defer s.BinlogRWMutex.RUnlock() + if s.lastWrittenBinlogCoordinate != nil { + return *s.lastWrittenBinlogCoordinate + } return NewFilePositionCoordinate(s.lastWrittenBinlogPosition) } @@ -230,6 +370,9 @@ func (s *StateTracker) LastResumableSourceBinlogCoordinateForInlineVerifier() Bi s.BinlogRWMutex.RLock() defer s.BinlogRWMutex.RUnlock() + if s.lastStoredBinlogCoordinateForInlineVerifier != nil { + return *s.lastStoredBinlogCoordinateForInlineVerifier + } return NewFilePositionCoordinate(s.lastStoredBinlogPositionForInlineVerifier) } @@ -237,6 +380,9 @@ func (s *StateTracker) LastResumableBinlogCoordinateForTargetVerifier() BinlogCo s.BinlogRWMutex.RLock() defer s.BinlogRWMutex.RUnlock() + if s.lastStoredBinlogCoordinateForTargetVerifier != nil { + return *s.lastStoredBinlogCoordinateForTargetVerifier + } return NewFilePositionCoordinate(s.lastStoredBinlogPositionForTargetVerifier) } @@ -367,6 +513,18 @@ func (s *StateTracker) Serialize(lastKnownTableSchemaCache TableSchemaCache, bin LastWrittenBinlogPosition: s.lastWrittenBinlogPosition, LastStoredBinlogPositionForInlineVerifier: s.lastStoredBinlogPositionForInlineVerifier, LastStoredBinlogPositionForTargetVerifier: s.lastStoredBinlogPositionForTargetVerifier, + + LastWrittenBinlogCoordinate: s.lastWrittenBinlogCoordinate, + LastStoredBinlogCoordinateForInlineVerifier: s.lastStoredBinlogCoordinateForInlineVerifier, + LastStoredBinlogCoordinateForTargetVerifier: s.lastStoredBinlogCoordinateForTargetVerifier, + } + + // If any GTID coordinate has been recorded, this state was produced in GTID + // mode. Marking the mode lets readers select the GTID fields on resume. + if s.lastWrittenBinlogCoordinate != nil || + s.lastStoredBinlogCoordinateForInlineVerifier != nil || + s.lastStoredBinlogCoordinateForTargetVerifier != nil { + state.BinlogCoordinateMode = BinlogCoordinateGTID } if binlogVerifyStore != nil { diff --git a/target_verifier.go b/target_verifier.go index 8e99b4d3..d00ebd26 100644 --- a/target_verifier.go +++ b/target_verifier.go @@ -46,7 +46,7 @@ func (t *TargetVerifier) BinlogEventListener(evs []DMLEvent) error { } if t.StateTracker != nil { - t.StateTracker.UpdateLastResumableBinlogPositionForTargetVerifier(evs[len(evs)-1].ResumableBinlogPosition()) + t.StateTracker.UpdateLastResumableBinlogCoordinateForTargetVerifier(evs[len(evs)-1].ResumableBinlogCoordinate()) } return nil diff --git a/test/go/binlog_writer_test.go b/test/go/binlog_writer_test.go index b2908eaa..902f2d49 100644 --- a/test/go/binlog_writer_test.go +++ b/test/go/binlog_writer_test.go @@ -30,8 +30,9 @@ func (e *stubDMLEvent) BinlogCoordinate() ghostferry.BinlogCoordinate { func (e *stubDMLEvent) ResumableBinlogCoordinate() ghostferry.BinlogCoordinate { return ghostferry.NewFilePositionCoordinate(mysql.Position{}) } -func (e *stubDMLEvent) Annotation() (string, error) { return "", nil } -func (e *stubDMLEvent) Timestamp() time.Time { return time.Time{} } +func (e *stubDMLEvent) SetCoordinates(_, _ ghostferry.BinlogCoordinate) {} +func (e *stubDMLEvent) Annotation() (string, error) { return "", nil } +func (e *stubDMLEvent) Timestamp() time.Time { return time.Time{} } // TestBinlogWriterBufferBinlogEventsBeforeRun verifies that BufferBinlogEvents // does not block when called before Run() has started in its own goroutine. diff --git a/test/go/state_gtid_test.go b/test/go/state_gtid_test.go new file mode 100644 index 00000000..e99d13a2 --- /dev/null +++ b/test/go/state_gtid_test.go @@ -0,0 +1,193 @@ +package test + +import ( + "encoding/json" + "testing" + + "github.com/Shopify/ghostferry" + "github.com/go-mysql-org/go-mysql/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + uuidA = "3e11fa47-71ca-11e1-9e33-c80aa9429562" + gtidWritten = uuidA + ":1-100" + gtidInline = uuidA + ":1-80" + gtidTarget = uuidA + ":1-90" + gtidExpected = uuidA + ":1-80" // intersection of written and inline +) + +func gtidCoord(s string) *ghostferry.BinlogCoordinate { + c := ghostferry.NewGTIDCoordinate(s) + return &c +} + +func TestSerializableState_GTIDRoundTrip(t *testing.T) { + state := &ghostferry.SerializableState{ + GhostferryVersion: "test-version", + LastSuccessfulPaginationKeys: map[string]ghostferry.PaginationKey{}, + CompletedTables: map[string]bool{}, + BinlogCoordinateMode: ghostferry.BinlogCoordinateGTID, + LastWrittenBinlogCoordinate: gtidCoord(gtidWritten), + LastStoredBinlogCoordinateForInlineVerifier: gtidCoord(gtidInline), + LastStoredBinlogCoordinateForTargetVerifier: gtidCoord(gtidTarget), + } + + data, err := json.Marshal(state) + require.NoError(t, err) + + var decoded ghostferry.SerializableState + require.NoError(t, json.Unmarshal(data, &decoded)) + + assert.Equal(t, ghostferry.BinlogCoordinateGTID, decoded.BinlogCoordinateMode) + require.NotNil(t, decoded.LastWrittenBinlogCoordinate) + assert.True(t, decoded.LastWrittenBinlogCoordinate.IsGTID()) + assert.Equal(t, gtidWritten, decoded.LastWrittenBinlogCoordinate.GTIDSet) + assert.Equal(t, gtidInline, decoded.LastStoredBinlogCoordinateForInlineVerifier.GTIDSet) + assert.Equal(t, gtidTarget, decoded.LastStoredBinlogCoordinateForTargetVerifier.GTIDSet) +} + +// TestSerializableState_FilePositionOmitsGTIDFields guards backward +// compatibility: a file/position state must not emit GTID fields or a mode. +func TestSerializableState_FilePositionOmitsGTIDFields(t *testing.T) { + state := &ghostferry.SerializableState{ + GhostferryVersion: "test-version", + LastSuccessfulPaginationKeys: map[string]ghostferry.PaginationKey{}, + CompletedTables: map[string]bool{}, + LastWrittenBinlogPosition: mysql.Position{Name: "mysql-bin.000001", Pos: 4}, + } + + data, err := json.Marshal(state) + require.NoError(t, err) + + str := string(data) + assert.NotContains(t, str, "BinlogCoordinateMode") + assert.NotContains(t, str, "LastWrittenBinlogCoordinate") + assert.NotContains(t, str, "GTIDSet") +} + +func TestSerializableState_MinSourceBinlogCoordinate_GTIDIntersection(t *testing.T) { + state := &ghostferry.SerializableState{ + BinlogCoordinateMode: ghostferry.BinlogCoordinateGTID, + LastWrittenBinlogCoordinate: gtidCoord(gtidWritten), + LastStoredBinlogCoordinateForInlineVerifier: gtidCoord(gtidInline), + } + + coord, err := state.MinSourceBinlogCoordinate() + require.NoError(t, err) + assert.True(t, coord.IsGTID()) + + // The safe resume point is the intersection: the smaller of the two here. + got, err := coord.ParsedGTIDSet() + require.NoError(t, err) + want, err := mysql.ParseMysqlGTIDSet(gtidExpected) + require.NoError(t, err) + assert.True(t, got.Equal(want), "expected intersection %s, got %s", want.String(), got.String()) +} + +// TestSerializableState_MinSourceBinlogCoordinate_MultiUUIDIntersection covers +// the intersection across two server UUIDs with non-contiguous ranges. +func TestSerializableState_MinSourceBinlogCoordinate_MultiUUIDIntersection(t *testing.T) { + uuidB := "8e12fa47-71ca-11e1-9e33-c80aa9429999" + written := uuidA + ":1-100," + uuidB + ":1-40" + inline := uuidA + ":1-70," + uuidB + ":1-60" + // Intersection = min ranges per UUID. + wantStr := uuidA + ":1-70," + uuidB + ":1-40" + + state := &ghostferry.SerializableState{ + BinlogCoordinateMode: ghostferry.BinlogCoordinateGTID, + LastWrittenBinlogCoordinate: gtidCoord(written), + LastStoredBinlogCoordinateForInlineVerifier: gtidCoord(inline), + } + + coord, err := state.MinSourceBinlogCoordinate() + require.NoError(t, err) + + got, err := coord.ParsedGTIDSet() + require.NoError(t, err) + want, err := mysql.ParseMysqlGTIDSet(wantStr) + require.NoError(t, err) + assert.True(t, got.Equal(want), "expected intersection %s, got %s", want.String(), got.String()) +} + +// TestSerializableState_MinSourceBinlogCoordinate_FailClosed verifies that an +// unparseable stored GTID coordinate produces an error rather than silently +// falling back to a possibly-too-advanced resume floor. +func TestSerializableState_MinSourceBinlogCoordinate_FailClosed(t *testing.T) { + state := &ghostferry.SerializableState{ + BinlogCoordinateMode: ghostferry.BinlogCoordinateGTID, + LastWrittenBinlogCoordinate: gtidCoord("not-a-valid-gtid-set"), + LastStoredBinlogCoordinateForInlineVerifier: gtidCoord(gtidInline), + } + + _, err := state.MinSourceBinlogCoordinate() + assert.Error(t, err) +} + +func TestSerializableState_MinSourceBinlogCoordinate_GTIDSingleSide(t *testing.T) { + state := &ghostferry.SerializableState{ + BinlogCoordinateMode: ghostferry.BinlogCoordinateGTID, + LastWrittenBinlogCoordinate: gtidCoord(gtidWritten), + } + + coord, err := state.MinSourceBinlogCoordinate() + require.NoError(t, err) + assert.True(t, coord.IsGTID()) + assert.Equal(t, gtidWritten, coord.GTIDSet) +} + +func TestSerializableState_HasTargetVerifierBinlogCoordinate(t *testing.T) { + // GTID mode: empty-set coordinate is present (not absent). + emptyGTID := ghostferry.NewGTIDCoordinate("") + state := &ghostferry.SerializableState{ + BinlogCoordinateMode: ghostferry.BinlogCoordinateGTID, + LastStoredBinlogCoordinateForTargetVerifier: &emptyGTID, + } + assert.True(t, state.HasTargetVerifierBinlogCoordinate(), + "an empty GTID set coordinate must count as present") + + // GTID mode with no coordinate: absent. + stateNone := &ghostferry.SerializableState{ + BinlogCoordinateMode: ghostferry.BinlogCoordinateGTID, + } + assert.False(t, stateNone.HasTargetVerifierBinlogCoordinate()) +} + +func TestStateTracker_GTIDUpdateAndSerialize(t *testing.T) { + st := ghostferry.NewStateTracker(0) + + st.UpdateLastResumableSourceBinlogCoordinate(ghostferry.NewGTIDCoordinate(gtidWritten)) + st.UpdateLastResumableSourceBinlogCoordinateForInlineVerifier(ghostferry.NewGTIDCoordinate(gtidInline)) + st.UpdateLastResumableBinlogCoordinateForTargetVerifier(ghostferry.NewGTIDCoordinate(gtidTarget)) + + assert.Equal(t, gtidWritten, st.LastResumableSourceBinlogCoordinate().GTIDSet) + assert.Equal(t, gtidInline, st.LastResumableSourceBinlogCoordinateForInlineVerifier().GTIDSet) + assert.Equal(t, gtidTarget, st.LastResumableBinlogCoordinateForTargetVerifier().GTIDSet) + + state := st.Serialize(nil, nil) + assert.Equal(t, ghostferry.BinlogCoordinateGTID, state.BinlogCoordinateMode) + require.NotNil(t, state.LastWrittenBinlogCoordinate) + assert.Equal(t, gtidWritten, state.LastWrittenBinlogCoordinate.GTIDSet) + + // Round-trip through a new tracker resumed from the serialized state. + resumed := ghostferry.NewStateTrackerFromSerializedState(0, state) + assert.Equal(t, gtidWritten, resumed.LastResumableSourceBinlogCoordinate().GTIDSet) + assert.Equal(t, gtidTarget, resumed.LastResumableBinlogCoordinateForTargetVerifier().GTIDSet) +} + +func TestStateTracker_FilePositionUpdateStaysFilePosition(t *testing.T) { + st := ghostferry.NewStateTracker(0) + + st.UpdateLastResumableSourceBinlogCoordinate( + ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000009", Pos: 42}), + ) + + coord := st.LastResumableSourceBinlogCoordinate() + assert.True(t, coord.IsFilePosition()) + assert.Equal(t, "mysql-bin.000009", coord.Position().Name) + + state := st.Serialize(nil, nil) + assert.Equal(t, ghostferry.BinlogCoordinateType(""), state.BinlogCoordinateMode) + assert.Nil(t, state.LastWrittenBinlogCoordinate) +} diff --git a/test/helpers/ghostferry_helper.rb b/test/helpers/ghostferry_helper.rb index 5d2e6162..736a3a06 100644 --- a/test/helpers/ghostferry_helper.rb +++ b/test/helpers/ghostferry_helper.rb @@ -270,6 +270,15 @@ def start_ghostferry(resuming_state = nil) environment["GHOSTFERRY_MARGINALIA"] = @config[:marginalia] end + # Binlog coordinate mode: prefer an explicit per-test config, otherwise + # fall back to the suite-wide GHOSTFERRY_BINLOG_COORDINATE_MODE env var. + # This lets CI run the entire suite against "file_position" or "gtid" + # while individual tests can still pin a mode. + binlog_coordinate_mode = @config[:binlog_coordinate_mode] || ENV["GHOSTFERRY_BINLOG_COORDINATE_MODE"] + if binlog_coordinate_mode && !binlog_coordinate_mode.empty? + environment["GHOSTFERRY_BINLOG_COORDINATE_MODE"] = binlog_coordinate_mode + end + @logger.debug("starting ghostferry test binary #{@compiled_binary_path}") Open3.popen3(environment, @compiled_binary_path) do |stdin, stdout, stderr, wait_thr| stdin.puts(resuming_state) unless resuming_state.nil? diff --git a/test/integration/callbacks_test.rb b/test/integration/callbacks_test.rb index 4ce644fd..1f7ad8bb 100644 --- a/test/integration/callbacks_test.rb +++ b/test/integration/callbacks_test.rb @@ -32,10 +32,17 @@ def test_progress_callback assert_equal 0, progress.last["ActiveDataIterators"] - refute progress.last["LastSuccessfulBinlogPos"]["Name"].nil? - refute progress.last["LastSuccessfulBinlogPos"]["Pos"].nil? - assert progress.last["BinlogStreamerLag"] > 0 - assert_equal progress.last["LastSuccessfulBinlogPos"], progress.last["FinalBinlogPos"] + if gtid_coordinate_mode? + refute progress.last["LastSuccessfulBinlogCoordinate"].nil? + refute progress.last["LastSuccessfulBinlogCoordinate"]["GTIDSet"].nil? + assert progress.last["BinlogStreamerLag"] > 0 + assert_equal progress.last["LastSuccessfulBinlogCoordinate"], progress.last["FinalBinlogCoordinate"] + else + refute progress.last["LastSuccessfulBinlogPos"]["Name"].nil? + refute progress.last["LastSuccessfulBinlogPos"]["Pos"].nil? + assert progress.last["BinlogStreamerLag"] > 0 + assert_equal progress.last["LastSuccessfulBinlogPos"], progress.last["FinalBinlogPos"] + end assert progress.last["VerifierMessage"].include?("currentRowCount =") assert progress.last["VerifierMessage"].include?("currentEntryCount =") diff --git a/test/integration/interrupt_resume_test.rb b/test/integration/interrupt_resume_test.rb index 6a6e9033..709bed83 100644 --- a/test/integration/interrupt_resume_test.rb +++ b/test/integration/interrupt_resume_test.rb @@ -121,8 +121,13 @@ def test_interrupt_resume_will_not_emit_binlog_position_for_inline_verifier_if_n dumped_state = ghostferry.run_expecting_interrupt assert_basic_fields_exist_in_dumped_state(dumped_state) - assert_equal "", dumped_state["LastStoredBinlogPositionForInlineVerifier"]["Name"] - assert_equal 0, dumped_state["LastStoredBinlogPositionForInlineVerifier"]["Pos"] + if gtid_coordinate_mode? + # Without a verifier, no inline-verifier coordinate should be recorded. + assert_nil dumped_state["LastStoredBinlogCoordinateForInlineVerifier"] + else + assert_equal "", dumped_state["LastStoredBinlogPositionForInlineVerifier"]["Name"] + assert_equal 0, dumped_state["LastStoredBinlogPositionForInlineVerifier"]["Pos"] + end end def test_interrupt_resume_inline_verifier_with_datawriter @@ -350,17 +355,38 @@ def test_interrupt_resume_between_consecutive_rows_events dumped_state = ghostferry.run_expecting_interrupt - refute_nil dumped_state['LastWrittenBinlogPosition']['Name'] - refute_nil dumped_state['LastWrittenBinlogPosition']['Pos'] - refute_nil dumped_state['LastStoredBinlogPositionForInlineVerifier']['Name'] - refute_nil dumped_state['LastStoredBinlogPositionForInlineVerifier']['Pos'] - refute_nil dumped_state['LastStoredBinlogPositionForTargetVerifier']['Name'] - refute_nil dumped_state['LastStoredBinlogPositionForTargetVerifier']['Pos'] - - # assert the resumable position is not the start position - if dumped_state['LastWrittenBinlogPosition']['Name'] == start_binlog_status['File'] - refute_equal dumped_state['LastWrittenBinlogPosition']['Pos'], start_binlog_status['Position'] - refute_equal dumped_state['LastStoredBinlogPositionForInlineVerifier']['Pos'], start_binlog_status['Position'] + if gtid_coordinate_mode? + written = dumped_state['LastWrittenBinlogCoordinate'] + inline = dumped_state['LastStoredBinlogCoordinateForInlineVerifier'] + target = dumped_state['LastStoredBinlogCoordinateForTargetVerifier'] + + refute_nil written + refute_nil inline + refute_nil target + refute_nil written['GTIDSet'] + refute_nil inline['GTIDSet'] + refute_nil target['GTIDSet'] + + # The resumable GTID set is conservatively the committed set BEFORE the + # in-flight transaction, so when the interrupt lands inside the first + # transaction after start it can still equal the start set. The key + # invariants are that a non-empty GTID set was recorded and that resume + # (asserted below) succeeds. It must never go backwards from the start set. + refute_empty written['GTIDSet'] + refute_empty inline['GTIDSet'] + else + refute_nil dumped_state['LastWrittenBinlogPosition']['Name'] + refute_nil dumped_state['LastWrittenBinlogPosition']['Pos'] + refute_nil dumped_state['LastStoredBinlogPositionForInlineVerifier']['Name'] + refute_nil dumped_state['LastStoredBinlogPositionForInlineVerifier']['Pos'] + refute_nil dumped_state['LastStoredBinlogPositionForTargetVerifier']['Name'] + refute_nil dumped_state['LastStoredBinlogPositionForTargetVerifier']['Pos'] + + # assert the resumable position is not the start position + if dumped_state['LastWrittenBinlogPosition']['Name'] == start_binlog_status['File'] + refute_equal dumped_state['LastWrittenBinlogPosition']['Pos'], start_binlog_status['Position'] + refute_equal dumped_state['LastStoredBinlogPositionForInlineVerifier']['Pos'], start_binlog_status['Position'] + end end ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY) diff --git a/test/lib/go/integrationferry/ferry.go b/test/lib/go/integrationferry/ferry.go index fe85962d..aa72b181 100644 --- a/test/lib/go/integrationferry/ferry.go +++ b/test/lib/go/integrationferry/ferry.go @@ -241,6 +241,13 @@ func NewStandardConfig() (*ghostferry.Config, error) { } } + // GHOSTFERRY_BINLOG_COORDINATE_MODE selects the binlog coordinate mode used + // by the integration run: "file_position" (default) or "gtid". It lets the + // Ruby suite run every test against either coordinate representation. + if binlogCoordinateMode := os.Getenv("GHOSTFERRY_BINLOG_COORDINATE_MODE"); binlogCoordinateMode != "" { + config.BinlogCoordinateMode = ghostferry.BinlogCoordinateType(binlogCoordinateMode) + } + verifierType := os.Getenv("GHOSTFERRY_VERIFIER_TYPE") if verifierType == ghostferry.VerifierTypeIterative { config.VerifierType = ghostferry.VerifierTypeIterative diff --git a/test/test_helper.rb b/test/test_helper.rb index bb752ded..545e3e76 100644 --- a/test/test_helper.rb +++ b/test/test_helper.rb @@ -203,13 +203,28 @@ def assert_uuid_table_is_identical # # To actually assert the validity of the data within the dumped state, you # have to do it manually. + # Returns the binlog coordinate mode the suite is currently running under. + # Defaults to "file_position" when unset. + def binlog_coordinate_mode + mode = ENV["GHOSTFERRY_BINLOG_COORDINATE_MODE"] + (mode.nil? || mode.empty?) ? "file_position" : mode + end + + def gtid_coordinate_mode? + binlog_coordinate_mode == "gtid" + end + def assert_basic_fields_exist_in_dumped_state(dumped_state) refute dumped_state.nil? refute dumped_state["GhostferryVersion"].nil? refute dumped_state["LastKnownTableSchemaCache"].nil? refute dumped_state["LastSuccessfulPaginationKeys"].nil? refute dumped_state["CompletedTables"].nil? - refute dumped_state["LastWrittenBinlogPosition"].nil? + if gtid_coordinate_mode? + refute dumped_state["LastWrittenBinlogCoordinate"].nil? + else + refute dumped_state["LastWrittenBinlogPosition"].nil? + end end def assert_ghostferry_completed(instance, times:) diff --git a/testhelpers/test_ferry.go b/testhelpers/test_ferry.go index 4e09516c..aae49aba 100644 --- a/testhelpers/test_ferry.go +++ b/testhelpers/test_ferry.go @@ -59,6 +59,12 @@ func NewTestConfig() *ghostferry.Config { }, } + // Allow the test suite to run against either binlog coordinate mode by + // setting GHOSTFERRY_BINLOG_COORDINATE_MODE. Defaults to file/position. + if mode := os.Getenv("GHOSTFERRY_BINLOG_COORDINATE_MODE"); mode != "" { + config.BinlogCoordinateMode = ghostferry.BinlogCoordinateType(mode) + } + err := config.ValidateConfig() PanicIfError(err)