From f461d6aa410d4a93fe75d2c9dd1d8d4e9eb8c763 Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Tue, 28 Jul 2026 10:13:12 +0200 Subject: [PATCH 1/2] Add GTID master failover recovery When the source master fails over, the binlog streamer previously died: a non-timeout GetEvent error went straight to ErrorHandler.Fatal, and go-mysql's syncer only reconnects to the same host. This adds opt-in, GTID-only recovery that reconnects the whole run to the promoted writer. Built on SourceRuntime: failover is an in-process resume onto a new source session rather than a hand-repoint of each consumer. - SourceReconnector is the seam between the streamer ("I lost the source; here's the GTID set I must not fall behind") and the Ferry, which owns the SourceRuntime. ferrySourceReconnector resolves the new writer, then SourceRuntime.ReplaceValidated opens one connection, validates it, and only publishes it to consumers if valid. Because every source consumer already reads from the SourceRuntime, that single swap repoints the whole run. - validateFailoverTarget fails closed: the candidate must be a writer (not read_only), have gtid_mode=ON, and its executed set must contain everything already applied downstream -- the committed streamed set plus any in-flight transaction's GTID (tracked from GTIDEvent/XIDEvent). If a cutover stop target is recorded, the candidate must contain it too, or streaming would hang. An undecodable in-flight GTID aborts recovery rather than validating against an under-approximated set. - On success the streamer rebuilds its syncer against the promoted writer and restarts StartSyncGTID from the last resumable set (replaying the interrupted transaction). DB/DBConfig swaps are under connMu; FlushAndStop reads currentDB() so cutover records the stop coordinate from the new writer. On recovery exhaustion the Run loop fatals and returns. - Bounds go-mysql's same-host reconnect retries (MaxReconnectAttempts, default 3) when failover is enabled. Without this the syncer retries the dead host forever and GetEvent only ever returns context timeouts, so recovery would never trigger. Behavior is unchanged when failover is disabled. Tunable via SyncerMaxReconnectAttempts. Config.MasterFailoverRecovery is validated (requires gtid mode + resolver) and wired onto the source streamer only; the target verifier streams from the stable target and does not fail over. --- binlog_coordinate.go | 61 +++++++ binlog_streamer.go | 323 ++++++++++++++++++++++++++++++++--- binlog_streamer_gtid_test.go | 30 ++++ config.go | 18 ++ ferry.go | 9 +- ferry_failover.go | 69 ++++++++ ferry_failover_test.go | 70 ++++++++ go.mod | 2 +- master_failover.go | 154 +++++++++++++++++ master_failover_test.go | 145 ++++++++++++++++ source_runtime.go | 21 +++ source_runtime_test.go | 36 ++++ test/go/config_test.go | 29 ++++ 13 files changed, 945 insertions(+), 22 deletions(-) create mode 100644 ferry_failover.go create mode 100644 ferry_failover_test.go create mode 100644 master_failover.go create mode 100644 master_failover_test.go diff --git a/binlog_coordinate.go b/binlog_coordinate.go index 2a989fdb..63dcecf8 100644 --- a/binlog_coordinate.go +++ b/binlog_coordinate.go @@ -261,6 +261,67 @@ func intersectGTIDSets(a, b mysql.GTIDSet) (mysql.GTIDSet, error) { return result, nil } +// unionGTIDStringInto returns a new GTID set equal to base with the GTID(s) in +// add merged in. base may be nil (treated as empty). It never mutates base. It +// is used by failover recovery to fold an in-flight transaction's GTID into the +// already-applied set before validating a candidate master. +func unionGTIDStringInto(base mysql.GTIDSet, add string) (mysql.GTIDSet, error) { + var result *mysql.MysqlGTIDSet + if base == nil { + empty, err := mysql.ParseMysqlGTIDSet("") + if err != nil { + return nil, err + } + result = empty.(*mysql.MysqlGTIDSet) + } else { + clone, ok := base.Clone().(*mysql.MysqlGTIDSet) + if !ok { + return nil, fmt.Errorf("GTID union requires a MySQL GTID set, got %T", base) + } + result = clone + } + + if err := result.Update(add); err != nil { + return nil, fmt.Errorf("GTID union: updating with %q: %w", add, err) + } + return result, nil +} + +// unionGTIDSets returns a new GTID set equal to a with b merged in. Either may +// be nil (treated as empty). It never mutates its inputs. It is used by failover +// recovery to require a candidate master to contain both the applied set and the +// cutover stop target. +func unionGTIDSets(a, b mysql.GTIDSet) (mysql.GTIDSet, error) { + var result *mysql.MysqlGTIDSet + if a == nil { + empty, err := mysql.ParseMysqlGTIDSet("") + if err != nil { + return nil, err + } + result = empty.(*mysql.MysqlGTIDSet) + } else { + clone, ok := a.Clone().(*mysql.MysqlGTIDSet) + if !ok { + return nil, fmt.Errorf("GTID union requires MySQL GTID sets, got %T", a) + } + result = clone + } + + if b != nil { + bMysql, ok := b.(*mysql.MysqlGTIDSet) + if !ok { + return nil, fmt.Errorf("GTID union requires MySQL GTID sets, got %T", b) + } + for _, uuidSet := range bMysql.Sets { + // Clone so result never aliases b's internal UUIDSet (AddSet stores + // the pointer directly when the SID is not already present). + result.AddSet(uuidSet.Clone()) + } + } + + return result, nil +} + // serializedBinlogCoordinate is the on-disk / on-wire shape of a // BinlogCoordinate. It is deliberately explicit and self-describing so that a // future GTID variant can be added as additional fields without breaking diff --git a/binlog_streamer.go b/binlog_streamer.go index 2819f938..72a9f3ca 100644 --- a/binlog_streamer.go +++ b/binlog_streamer.go @@ -52,6 +52,16 @@ type BinlogStreamer struct { // backwards compatibility. BinlogCoordinateMode BinlogCoordinateType + // MasterFailoverRecovery, when set together with SourceReconnector, enables + // automatic reconnection to a new source master on connection loss. Honored + // only in GTID mode (GTID sets are server-independent). Nil disables it. + MasterFailoverRecovery *MasterFailoverRecoveryConfig + + // SourceReconnector reconnects the source after a lost connection. It is + // supplied by the Ferry, which owns the SourceRuntime and therefore how the + // whole run repoints at the promoted writer. Required for failover recovery. + SourceReconnector SourceReconnector + lastStreamedBinlogPosition mysql.Position lastResumableBinlogPosition mysql.Position stopAtBinlogPosition mysql.Position @@ -74,6 +84,17 @@ type BinlogStreamer struct { lastResumableGTIDSet mysql.GTIDSet stopAtGTIDSet mysql.GTIDSet + // inFlightGTID is the GTID of the transaction currently streamed but not yet + // committed (set at GTIDEvent, cleared at its XIDEvent). Its rows may already + // have been emitted downstream before lastStreamedGTIDSet advances, so + // failover validation folds it into the applied set. Only touched by the Run + // goroutine. + inFlightGTID string + + // connMu guards DB and DBConfig, swapped by the Run goroutine during + // failover recovery while FlushAndStop (another goroutine) reads them. + connMu sync.Mutex + lastProcessedEventTime time.Time lastLagMetricEmittedTime time.Time @@ -100,29 +121,42 @@ func (s *BinlogStreamer) ensureLogger() { func (s *BinlogStreamer) createBinlogSyncer() error { var err error - var tlsConfig *tls.Config - if s.DBConfig.TLS != nil { - tlsConfig, err = s.DBConfig.TLS.BuildConfig() + if s.MyServerId == 0 { + s.MyServerId, err = s.generateNewServerId() if err != nil { + s.logger.WithError(err).Error("could not generate unique server_id") return err } } - if s.MyServerId == 0 { - s.MyServerId, err = s.generateNewServerId() + syncer, err := s.newBinlogSyncerFor(s.DBConfig, s.MyServerId) + if err != nil { + return err + } + s.binlogSyncer = syncer + return nil +} + +// newBinlogSyncerFor builds a BinlogSyncer targeting the given DB config and +// server id without mutating streamer state, so failover recovery can construct +// a candidate syncer before committing to it. +func (s *BinlogStreamer) newBinlogSyncerFor(dbConf *DatabaseConfig, serverID uint32) (*replication.BinlogSyncer, error) { + var tlsConfig *tls.Config + if dbConf.TLS != nil { + var err error + tlsConfig, err = dbConf.TLS.BuildConfig() if err != nil { - s.logger.WithError(err).Error("could not generate unique server_id") - return err + return nil, err } } syncerConfig := replication.BinlogSyncerConfig{ - ServerID: s.MyServerId, - Host: s.DBConfig.Host, - Port: s.DBConfig.Port, - User: s.DBConfig.User, - Password: s.DBConfig.Pass, + ServerID: serverID, + Host: dbConf.Host, + Port: dbConf.Port, + User: dbConf.User, + Password: dbConf.Pass, TLSConfig: tlsConfig, UseDecimal: true, UseFloatWithTrailingZero: true, @@ -130,8 +164,17 @@ func (s *BinlogStreamer) createBinlogSyncer() error { Logger: NewSlogLogger(s.logger), } - s.binlogSyncer = replication.NewBinlogSyncer(syncerConfig) - return nil + // When master-failover recovery is enabled, bound go-mysql's internal + // reconnect-to-the-same-host retries. Otherwise the syncer would retry the + // dead host forever and GetEvent would only ever surface context timeouts, + // so our failover recovery (which reconnects to a *different* host) would + // never be triggered. With the default (0) behavior is unchanged: infinite + // retries against the configured host, matching pre-failover Ghostferry. + if s.failoverRecoveryEnabled() { + syncerConfig.MaxReconnectAttempts = s.MasterFailoverRecovery.syncerMaxReconnectAttempts() + } + + return replication.NewBinlogSyncer(syncerConfig), nil } func (s *BinlogStreamer) ConnectBinlogStreamerToMysql() (mysql.Position, error) { @@ -261,6 +304,184 @@ func (s *BinlogStreamer) connectBinlogStreamerFromGTID(startFrom BinlogCoordinat return NewGTIDCoordinate(s.lastStreamedGTIDString()), nil } +// failoverRecoveryEnabled reports whether automatic master-failover recovery is +// configured and usable. It requires GTID mode and a reconnector. +func (s *BinlogStreamer) failoverRecoveryEnabled() bool { + return s.MasterFailoverRecovery != nil && + s.SourceReconnector != nil && + s.coordinateMode() == BinlogCoordinateGTID +} + +// currentDB returns the active source DB handle under the connection lock. Used +// by goroutines other than Run (FlushAndStop) so they observe a consistent +// handle across a failover swap. +func (s *BinlogStreamer) currentDB() *sql.DB { + s.connMu.Lock() + defer s.connMu.Unlock() + return s.DB +} + +// currentDBConfig returns the active DBConfig under the connection lock. +func (s *BinlogStreamer) currentDBConfig() *DatabaseConfig { + s.connMu.Lock() + defer s.connMu.Unlock() + return s.DBConfig +} + +// appliedGTIDSet returns everything the streamer has emitted downstream: the +// committed streamed set plus any in-flight transaction's GTID (whose rows may +// already have reached listeners before the commit advanced the streamed set). +// Failover validation must ensure a candidate contains this whole set. +// +// It fails closed: if an in-flight GTID is present but cannot be folded in, it +// returns an error rather than an under-approximated set, so recovery refuses +// to validate a candidate against an incomplete applied set (which could let a +// diverging master through). +func (s *BinlogStreamer) appliedGTIDSet() (mysql.GTIDSet, error) { + // Snapshot the streamed set under gtidMu; it is mutated by the XID/DDL paths + // and read by Progress() concurrently. + applied := s.lastStreamedGTIDClone() + if s.inFlightGTID != "" { + merged, err := unionGTIDStringInto(applied, s.inFlightGTID) + if err != nil { + return nil, fmt.Errorf("folding in-flight GTID %q into applied set: %w", s.inFlightGTID, err) + } + applied = merged + } + return applied, nil +} + +// recoverFromMasterFailover reconnects the source after a lost connection and +// restarts streaming from the last resumable GTID set. It is only meaningful in +// GTID mode; the caller must guard with failoverRecoveryEnabled. +// +// Each attempt asks the SourceReconnector for a validated connection to the new +// writer (the Ferry implements this via SourceRuntime.Replace, so the whole run +// repoints in one place), then rebuilds the binlog syncer against it and +// restarts StartSyncGTID from the resumable set (replaying the interrupted +// transaction). Returns an error only when recovery is exhausted. +func (s *BinlogStreamer) recoverFromMasterFailover(cause error) error { + cfg := s.MasterFailoverRecovery + + resumeSet := s.resumableGTIDClone() + appliedSet, err := s.appliedGTIDSet() + if err != nil { + // Fail closed: without a trustworthy applied set we cannot safely + // validate a candidate, so refuse to recover. + return fmt.Errorf("master failover recovery: %w (original error: %v)", err, cause) + } + + // If a cutover stop target has been recorded, the promoted master must also + // contain it; otherwise streaming would wait forever for GTIDs the new + // writer will never produce. Fold it into the set the candidate must + // contain. + if stopSet := s.stopGTIDClone(); stopSet != nil { + merged, mergeErr := unionGTIDSets(appliedSet, stopSet) + if mergeErr != nil { + return fmt.Errorf("master failover recovery: folding stop target into required set: %w (original error: %v)", mergeErr, cause) + } + appliedSet = merged + } + + previous := s.currentDBConfig() + previousHost, previousPort := "", uint16(0) + if previous != nil { + previousHost, previousPort = previous.Host, previous.Port + } + + s.logger.WithFields(Fields{ + "error": cause.Error(), + "resume_set": gtidSetString(resumeSet), + "applied_set": gtidSetString(appliedSet), + "dead_host": previousHost, + "dead_port": previousPort, + "max_attempts": cfg.MaxAttempts, + }).Warn("source connection lost; attempting master failover recovery") + + for attempt := 1; cfg.MaxAttempts == 0 || attempt <= cfg.MaxAttempts; attempt++ { + newDB, newConfig, err := s.SourceReconnector.Reconnect(previous, appliedSet) + if err != nil { + s.logger.WithError(err).WithField("attempt", attempt).Warn("failover: source reconnect failed") + time.Sleep(cfg.retryWait()) + continue + } + + newSyncer, newStreamer, err := s.buildSyncerFromGTID(newConfig, newDB, resumeSet) + if err != nil { + s.logger.WithError(err).WithField("attempt", attempt).Warn("failover: could not restart streaming on new master") + time.Sleep(cfg.retryWait()) + continue + } + + oldSyncer := s.binlogSyncer + s.connMu.Lock() + s.DB = newDB + s.DBConfig = newConfig + s.binlogSyncer = newSyncer + s.binlogStreamer = newStreamer + s.connMu.Unlock() + // Re-seed the streamed/resumable GTID sets under gtidMu so a concurrent + // Progress() read never observes a torn set during the swap. + s.seedGTIDSets(cloneOrEmpty(resumeSet)) + s.inFlightGTID = "" + + if oldSyncer != nil { + oldSyncer.Close() + } + + s.logger.WithFields(Fields{ + "new_host": newConfig.Host, + "new_port": newConfig.Port, + "resume_set": gtidSetString(resumeSet), + "attempt": attempt, + }).Info("master failover recovery succeeded; streaming resumed on new master") + return nil + } + + return fmt.Errorf("master failover recovery exhausted after %d attempts: %w", cfg.MaxAttempts, cause) +} + +// buildSyncerFromGTID constructs (without installing) a fresh binlog syncer +// against the given DB config and starts a GTID stream from resumeSet. It has +// no side effects on installed state, so callers can discard the result on a +// later failure. The server id is generated against db (the new master). +func (s *BinlogStreamer) buildSyncerFromGTID(dbConf *DatabaseConfig, db *sql.DB, resumeSet mysql.GTIDSet) (*replication.BinlogSyncer, *replication.BinlogStreamer, error) { + serverID, err := generateNewServerIdOn(db, s.logger) + if err != nil { + return nil, nil, err + } + + syncer, err := s.newBinlogSyncerFor(dbConf, serverID) + if err != nil { + return nil, nil, err + } + + streamer, err := syncer.StartSyncGTID(cloneOrEmpty(resumeSet)) + if err != nil { + syncer.Close() + return nil, nil, err + } + return syncer, streamer, nil +} + +// gtidSetString renders a possibly-nil GTID set for logging. +func gtidSetString(set mysql.GTIDSet) string { + if set == nil { + return "" + } + return set.String() +} + +// cloneOrEmpty clones set, returning a fresh empty MySQL GTID set when set is +// nil (a valid starting point for a fresh source). +func cloneOrEmpty(set mysql.GTIDSet) mysql.GTIDSet { + if set == nil { + empty, _ := mysql.ParseMysqlGTIDSet("") + return empty + } + return set.Clone() +} + // the default event handler is called for replication binLogEvents that do not have a // separate event Handler registered. @@ -372,6 +593,18 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query // set that existed BEFORE this transaction, so that an // interruption replays the whole in-flight transaction. s.setResumableToStreamed() + // Record this transaction's GTID as in-flight: its rows may be + // emitted downstream before the closing XIDEvent advances + // lastStreamedGTIDSet, so failover validation must ensure a + // candidate master contains it. GTIDNext returns the single-GTID + // set (uuid:gno) for this transaction. inFlightGTID is only + // touched by this (streaming) goroutine, so it needs no lock. + if nextSet, uerr := tev.GTIDNext(); uerr == nil { + s.inFlightGTID = nextSet.String() + } else { + s.logger.WithError(uerr).Warn("could not decode in-flight GTID; failover validation may under-approximate applied set") + s.inFlightGTID = "" + } case *replication.XIDEvent: // End of a transaction. GSet is the committed GTID set through // this transaction. setLastStreamedGTIDSet clones under gtidMu to @@ -380,6 +613,8 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query if tev.GSet != nil { s.setLastStreamedGTIDSet(tev.GSet) } + // The transaction committed; nothing is in flight now. + s.inFlightGTID = "" } } @@ -453,11 +688,28 @@ func (s *BinlogStreamer) Run() { ev, err = s.binlogStreamer.GetEvent(ctx) if err == context.DeadlineExceeded { timedOut = true - } else if err != nil { - s.ErrorHandler.Fatal("binlog_streamer", err) } }() + if err != nil && err != context.DeadlineExceeded { + // A non-timeout GetEvent error means the source connection was lost. + // In GTID mode with failover recovery enabled, reconnect to the new + // master and continue; otherwise this is fatal. + if s.failoverRecoveryEnabled() { + if recoverErr := s.recoverFromMasterFailover(err); recoverErr != nil { + s.ErrorHandler.Fatal("binlog_streamer", recoverErr) + return + } + // Recovered: reset per-iteration state so no stale RowsQueryEvent + // annotation or filename crosses from the old stream. + es = BinlogEventState{nextFilename: s.lastStreamedBinlogPosition.Name} + query = nil + continue + } + s.ErrorHandler.Fatal("binlog_streamer", err) + return + } + if timedOut { s.lastProcessedEventTime = time.Now() continue @@ -595,6 +847,28 @@ func (s *BinlogStreamer) resumableGTIDClone() mysql.GTIDSet { return s.lastResumableGTIDSet.Clone() } +// lastStreamedGTIDClone returns a mutable clone of the streamed set under the +// lock, or nil when unset. +func (s *BinlogStreamer) lastStreamedGTIDClone() mysql.GTIDSet { + s.gtidMu.RLock() + defer s.gtidMu.RUnlock() + if s.lastStreamedGTIDSet == nil { + return nil + } + return s.lastStreamedGTIDSet.Clone() +} + +// stopGTIDClone returns a mutable clone of the stop set under the lock, or nil +// when unset. +func (s *BinlogStreamer) stopGTIDClone() mysql.GTIDSet { + s.gtidMu.RLock() + defer s.gtidMu.RUnlock() + if s.stopAtGTIDSet == nil { + return nil + } + return s.stopAtGTIDSet.Clone() +} + // streamedGTIDCoordinate and resumableGTIDCoordinate return GTID coordinates // snapshotted under gtidMu, for stamping onto DML events. They read the sets // while holding the lock so the snapshot never races the streaming goroutine's @@ -666,7 +940,7 @@ func (s *BinlogStreamer) FlushAndStop() { // 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) + gtidSet, err := ReadExecutedGTIDSet(s.currentDB()) if err != nil { return err } @@ -691,7 +965,7 @@ func (s *BinlogStreamer) FlushAndStop() { err := WithRetries(100, 600*time.Millisecond, s.logger, "read current binlog position", func() error { var err error - s.stopAtBinlogPosition, err = ShowMasterStatusBinlogPosition(s.DB) + s.stopAtBinlogPosition, err = ShowMasterStatusBinlogPosition(s.currentDB()) return err }) @@ -822,12 +1096,19 @@ func (s *BinlogStreamer) handleRowsEvent(ev *replication.BinlogEvent, query []by } func (s *BinlogStreamer) generateNewServerId() (uint32, error) { + return generateNewServerIdOn(s.DB, s.logger) +} + +// generateNewServerIdOn generates a server id not already in use on the given +// server. It takes an explicit DB so failover recovery can target a server +// other than the currently-installed one. +func generateNewServerIdOn(db *sql.DB, logger Logger) (uint32, error) { var id uint32 for { id = randomServerId() - exists, err := idExistsOnServer(id, s.DB) + exists, err := idExistsOnServer(id, db) if err != nil { return 0, err } @@ -835,7 +1116,9 @@ func (s *BinlogStreamer) generateNewServerId() (uint32, error) { break } - s.logger.WithField("server_id", id).Warn("server_id was taken, retrying") + if logger != nil { + logger.WithField("server_id", id).Warn("server_id was taken, retrying") + } } return id, nil diff --git a/binlog_streamer_gtid_test.go b/binlog_streamer_gtid_test.go index bffd4a09..3c8ce27e 100644 --- a/binlog_streamer_gtid_test.go +++ b/binlog_streamer_gtid_test.go @@ -6,6 +6,7 @@ import ( "github.com/go-mysql-org/go-mysql/mysql" "github.com/go-mysql-org/go-mysql/replication" + "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -64,6 +65,35 @@ func TestQueryEventAdvancesStreamedGTID(t *testing.T) { assert.Equal(t, before, s.lastStreamedGTIDSet.String(), "BEGIN must not advance the streamed GTID set") } +// TestInFlightGTIDTracking verifies a GTIDEvent records the in-flight +// transaction's GTID and the closing XIDEvent clears it. Failover validation +// relies on this to reject a candidate missing an emitted-but-uncommitted txn. +func TestInFlightGTIDTracking(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + s.logger = LogWithField("tag", "test") + + sid, err := uuid.Parse("3e11fa47-71ca-11e1-9e33-c80aa9429562") + require.NoError(t, err) + es := &BinlogEventState{} + + gtidEvent := &replication.BinlogEvent{ + Header: &replication.EventHeader{LogPos: 100}, + Event: &replication.GTIDEvent{SID: sid[:], GNO: 101}, + } + _, err = s.defaultEventHandler(gtidEvent, nil, es) + require.NoError(t, err) + assert.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:101", s.inFlightGTID) + + xidEvent := &replication.BinlogEvent{ + Header: &replication.EventHeader{LogPos: 200}, + Event: &replication.XIDEvent{GSet: mustParseGTID(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-101")}, + } + _, err = s.defaultEventHandler(xidEvent, nil, es) + require.NoError(t, err) + assert.Equal(t, "", s.inFlightGTID, "XIDEvent must clear the in-flight GTID") + assert.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-101", s.lastStreamedGTIDSet.String()) +} + func TestCoordinateModeDefaultsToFilePosition(t *testing.T) { s := &BinlogStreamer{} assert.Equal(t, BinlogCoordinateFilePosition, s.coordinateMode()) diff --git a/config.go b/config.go index 4cf9dce0..c6efb42f 100644 --- a/config.go +++ b/config.go @@ -707,6 +707,15 @@ type Config struct { // additional coordinate sources. BinlogCoordinateMode BinlogCoordinateType + // MasterFailoverRecovery, when set, enables automatic reconnection of the + // source to a new master writer when the source connection is lost (a + // topology failover). It requires BinlogCoordinateMode "gtid" (GTID + // coordinates are server-independent; file/position cannot be carried across + // a failover). The Resolver is supplied by the embedding application to + // discover the new writer. Nil keeps the previous behavior: a lost source + // connection is fatal. + MasterFailoverRecovery *MasterFailoverRecoveryConfig + // This config is necessary for inline verification for a special case of // Ghostferry: // @@ -885,6 +894,15 @@ func (c *Config) ValidateConfig() error { return fmt.Errorf("invalid BinlogCoordinateMode %q, must be %q or %q", c.BinlogCoordinateMode, BinlogCoordinateFilePosition, BinlogCoordinateGTID) } + if c.MasterFailoverRecovery != nil { + if c.BinlogCoordinateMode != BinlogCoordinateGTID { + return fmt.Errorf("MasterFailoverRecovery requires BinlogCoordinateMode %q, got %q", BinlogCoordinateGTID, c.BinlogCoordinateMode) + } + if c.MasterFailoverRecovery.Resolver == nil { + return fmt.Errorf("MasterFailoverRecovery requires a Resolver") + } + } + if c.VerifierType == VerifierTypeIterative { if err := c.IterativeVerifierConfig.Validate(); err != nil { return fmt.Errorf("IterativeVerifierConfig invalid: %v", err) diff --git a/ferry.go b/ferry.go index 6c8c3a74..4f963e24 100644 --- a/ferry.go +++ b/ferry.go @@ -141,7 +141,14 @@ func (f *Ferry) NewDataIteratorWithoutStateTracker() *DataIterator { } func (f *Ferry) NewSourceBinlogStreamer() *BinlogStreamer { - return f.newBinlogStreamer(f.SourceDB, f.Config.Source, nil, nil, "source_binlog_streamer") + streamer := f.newBinlogStreamer(f.SourceDB, f.Config.Source, nil, nil, "source_binlog_streamer") + // Master-failover recovery only applies to the source stream: the target + // verifier streams from the (stable) target and does not fail over. The + // reconnector repoints the whole run via SourceRuntime when a failover + // occurs. + streamer.MasterFailoverRecovery = f.Config.MasterFailoverRecovery + streamer.SourceReconnector = f.newSourceReconnector() + return streamer } func (f *Ferry) NewTargetBinlogStreamer() (*BinlogStreamer, error) { diff --git a/ferry_failover.go b/ferry_failover.go new file mode 100644 index 00000000..33fcdd50 --- /dev/null +++ b/ferry_failover.go @@ -0,0 +1,69 @@ +package ghostferry + +import ( + sql "github.com/Shopify/ghostferry/sqlwrapper" + + "github.com/go-mysql-org/go-mysql/mysql" +) + +// ferrySourceReconnector implements SourceReconnector for the Ferry. It is the +// single place a source master failover repoints the whole run: it resolves the +// promoted writer, opens one Ferry-owned connection to it via SourceRuntime +// (which atomically becomes the source for every consumer that starts new work +// afterwards — data iterator, verifiers, ...), validates it is a safe target, +// and hands the connection back to the binlog streamer. +// +// Because every source consumer already reads its connection from the +// SourceRuntime, there is nothing to hand-repoint here: the Replace is the swap. +type ferrySourceReconnector struct { + ferry *Ferry + resolver MasterWriterResolver +} + +// Reconnect satisfies SourceReconnector. previous is the config the streamer +// last used; appliedGTIDSet is everything already emitted downstream, which the +// promoted writer must contain. +func (r *ferrySourceReconnector) Reconnect(previous *DatabaseConfig, appliedGTIDSet mysql.GTIDSet) (*sql.DB, *DatabaseConfig, error) { + candidate, err := r.resolver.ResolveCurrentMaster(previous) + if err != nil { + return nil, nil, err + } + if candidate == nil { + return nil, nil, errNilCandidate + } + + // Install the candidate as the current source ONLY if it validates. Fail + // closed: the promoted writer must be a writable server with GTID mode on + // and must contain everything already applied downstream. A candidate that + // fails validation is never published to consumers. On success the old + // handle is retired (retained, not closed), since in-flight cursors / cached + // statements may still reference it. + newDB, err := r.ferry.sourceRuntime.ReplaceValidated( + candidate, + r.ferry.logger.WithField("dbname", "source_failover"), + func(db *sql.DB) error { + return validateFailoverTarget(db, candidate, appliedGTIDSet) + }, + ) + if err != nil { + return nil, nil, err + } + + r.ferry.logger.WithFields(Fields{ + "new_host": candidate.Host, + "new_port": candidate.Port, + }).Info("source repointed at promoted master after failover") + return newDB, candidate, nil +} + +// newSourceReconnector builds the reconnector wired to this Ferry, or nil when +// failover recovery is not configured. +func (f *Ferry) newSourceReconnector() SourceReconnector { + if f.Config.MasterFailoverRecovery == nil || f.Config.MasterFailoverRecovery.Resolver == nil { + return nil + } + return &ferrySourceReconnector{ + ferry: f, + resolver: f.Config.MasterFailoverRecovery.Resolver, + } +} diff --git a/ferry_failover_test.go b/ferry_failover_test.go new file mode 100644 index 00000000..272c69bb --- /dev/null +++ b/ferry_failover_test.go @@ -0,0 +1,70 @@ +package ghostferry + +import ( + "errors" + "testing" + + sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func reconnectorTestConfig(host string) *DatabaseConfig { + return &DatabaseConfig{Host: host, Port: 3306, User: "root", Net: "tcp"} +} + +func TestNewSourceReconnectorNilWhenDisabled(t *testing.T) { + f := &Ferry{Config: &Config{}} + assert.Nil(t, f.newSourceReconnector()) + + f.Config.MasterFailoverRecovery = &MasterFailoverRecoveryConfig{} + assert.Nil(t, f.newSourceReconnector(), "nil resolver disables reconnector") +} + +func TestReconnectPropagatesResolverError(t *testing.T) { + wantErr := errors.New("no master found") + f := &Ferry{ + Config: &Config{Source: reconnectorTestConfig("old")}, + logger: LogWithField("tag", "test"), + sourceRuntime: NewSourceRuntime(&sql.DB{Marginalia: "old"}, reconnectorTestConfig("old")), + } + r := &ferrySourceReconnector{ + ferry: f, + resolver: MasterWriterResolverFunc(func(_ *DatabaseConfig) (*DatabaseConfig, error) { return nil, wantErr }), + } + + _, _, err := r.Reconnect(reconnectorTestConfig("old"), nil) + assert.ErrorIs(t, err, wantErr) +} + +func TestReconnectRejectsNilCandidate(t *testing.T) { + f := &Ferry{ + Config: &Config{Source: reconnectorTestConfig("old")}, + logger: LogWithField("tag", "test"), + sourceRuntime: NewSourceRuntime(&sql.DB{Marginalia: "old"}, reconnectorTestConfig("old")), + } + r := &ferrySourceReconnector{ + ferry: f, + resolver: MasterWriterResolverFunc(func(_ *DatabaseConfig) (*DatabaseConfig, error) { return nil, nil }), + } + + _, _, err := r.Reconnect(reconnectorTestConfig("old"), nil) + assert.ErrorIs(t, err, errNilCandidate) + + // The runtime must not have been swapped when the candidate is nil. + assert.Equal(t, "old", f.sourceRuntime.Config().Host) +} + +func TestNewSourceReconnectorWiredWhenConfigured(t *testing.T) { + f := &Ferry{ + Config: &Config{ + MasterFailoverRecovery: &MasterFailoverRecoveryConfig{ + Resolver: MasterWriterResolverFunc(func(_ *DatabaseConfig) (*DatabaseConfig, error) { return nil, nil }), + }, + }, + } + r := f.newSourceReconnector() + require.NotNil(t, r) + _, ok := r.(*ferrySourceReconnector) + assert.True(t, ok) +} diff --git a/go.mod b/go.mod index 0dbdbfd0..bd8d0b00 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/go-mysql-org/go-mysql v1.13.0 github.com/go-sql-driver/mysql v1.7.1 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db + github.com/google/uuid v1.3.0 github.com/gorilla/mux v1.6.1 github.com/rs/zerolog v1.35.0 github.com/shopspring/decimal v1.2.0 @@ -20,7 +21,6 @@ require ( github.com/Microsoft/go-winio v0.5.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/goccy/go-json v0.10.2 // indirect - github.com/google/uuid v1.3.0 // indirect github.com/gorilla/context v1.1.1 // indirect github.com/klauspost/compress v1.17.8 // indirect github.com/lann/builder v0.0.0-20180216234317-1b87b36280d0 // indirect diff --git a/master_failover.go b/master_failover.go new file mode 100644 index 00000000..74e7a89c --- /dev/null +++ b/master_failover.go @@ -0,0 +1,154 @@ +package ghostferry + +import ( + "errors" + "fmt" + "time" + + sql "github.com/Shopify/ghostferry/sqlwrapper" + + "github.com/go-mysql-org/go-mysql/mysql" +) + +// errNilCandidate is returned when a resolver yields no candidate master. +var errNilCandidate = errors.New("failover: resolver returned a nil candidate master") + +// MasterWriterResolver resolves the current source master writer after a +// suspected failover. Implementations typically consult an external topology +// service (orchestrator, a service-discovery endpoint, etc.) to find the host +// that is now the writable primary. +// +// ResolveCurrentMaster receives the DatabaseConfig the source was last +// connected to (so the implementation can, for example, exclude the now-dead +// host) and returns the connection details of the host that should now be +// streamed from. Returning an error causes recovery to be retried after a +// backoff. +type MasterWriterResolver interface { + ResolveCurrentMaster(previous *DatabaseConfig) (*DatabaseConfig, error) +} + +// MasterWriterResolverFunc adapts a plain function to MasterWriterResolver. +type MasterWriterResolverFunc func(previous *DatabaseConfig) (*DatabaseConfig, error) + +func (f MasterWriterResolverFunc) ResolveCurrentMaster(previous *DatabaseConfig) (*DatabaseConfig, error) { + return f(previous) +} + +// MasterFailoverRecoveryConfig configures automatic reconnection to a new +// source master when the source connection is lost. +// +// Failover recovery is only supported in GTID binlog coordinate mode: GTID sets +// are server-independent, so a resume set valid on the old master is meaningful +// on the new one. File/position coordinates are per-host and cannot be carried +// across a failover, so recovery is refused in that mode. +type MasterFailoverRecoveryConfig struct { + // Resolver discovers the new master writer. Required; recovery is disabled + // when nil. + Resolver MasterWriterResolver + + // MaxAttempts bounds how many times recovery is attempted for a single + // disconnect before giving up and surfacing a fatal error. Zero means retry + // indefinitely. + MaxAttempts int + + // RetryWait is how long to wait between recovery attempts. Defaults to + // DefaultFailoverRetryWait when zero. + RetryWait time.Duration + + // SyncerMaxReconnectAttempts bounds how many times go-mysql's binlog syncer + // retries reconnecting to the SAME (now-dead) host before giving up and + // surfacing the error, which is what triggers our reconnect-to-a-new-host + // recovery. Without a bound the syncer retries the dead host forever and the + // failure never surfaces. Defaults to DefaultSyncerMaxReconnectAttempts. + SyncerMaxReconnectAttempts int +} + +// DefaultSyncerMaxReconnectAttempts bounds go-mysql's same-host reconnect +// retries when failover recovery is enabled, so a lost source surfaces as a +// GetEvent error (triggering failover) within a few seconds. +const DefaultSyncerMaxReconnectAttempts = 3 + +func (c *MasterFailoverRecoveryConfig) syncerMaxReconnectAttempts() int { + if c.SyncerMaxReconnectAttempts <= 0 { + return DefaultSyncerMaxReconnectAttempts + } + return c.SyncerMaxReconnectAttempts +} + +// DefaultFailoverRetryWait is the wait between failover recovery attempts when +// MasterFailoverRecoveryConfig.RetryWait is unset. +const DefaultFailoverRetryWait = 500 * time.Millisecond + +func (c *MasterFailoverRecoveryConfig) retryWait() time.Duration { + if c.RetryWait <= 0 { + return DefaultFailoverRetryWait + } + return c.RetryWait +} + +// SourceReconnector reconnects the source after the binlog streamer loses its +// connection. It is the single seam between the streamer (which only knows "I +// lost the source and here is the safe GTID set I must not fall behind") and +// the Ferry (which owns the SourceRuntime and therefore how the whole run +// repoints at a promoted writer). +// +// Reconnect is called on the streamer's Run goroutine when a non-timeout +// GetEvent error occurs. appliedGTIDSet is everything the streamer has already +// emitted downstream (committed set plus any in-flight transaction); the +// implementation must guarantee the returned source contains it, or fail +// closed, so recovery never resumes against a master that lost applied +// transactions (which would silently diverge the target). +// +// On success it returns the new source connection and the config it was opened +// from; the streamer rebuilds its binlog syncer against them. Returning an +// error means this attempt failed and the streamer will retry (subject to +// MaxAttempts). +type SourceReconnector interface { + Reconnect(previous *DatabaseConfig, appliedGTIDSet mysql.GTIDSet) (*sql.DB, *DatabaseConfig, error) +} + +// validateFailoverTarget verifies that db (already opened against candidate) is +// a safe source to resume streaming from: +// +// - it is a writer (@@read_only = OFF), not a demoted master or replica; +// - GTID mode is enabled (@@GLOBAL.gtid_mode = ON); and +// - its executed GTID set contains appliedSet, i.e. everything already +// emitted downstream. A candidate missing any applied transaction is +// rejected (fail closed) so recovery cannot silently diverge the target. +// +// It does not close db; the caller owns its lifecycle. +func validateFailoverTarget(db *sql.DB, candidate *DatabaseConfig, appliedSet mysql.GTIDSet) error { + if candidate == nil { + return fmt.Errorf("failover: nil candidate master config") + } + + isReadOnly, err := CheckDbIsAReplica(db) + if err != nil { + return fmt.Errorf("failover: checking candidate master %s:%d read_only: %w", candidate.Host, candidate.Port, err) + } + if isReadOnly { + return fmt.Errorf("failover: candidate master %s:%d is read_only (not a writer); rejecting", candidate.Host, candidate.Port) + } + + if err := CheckServerGTIDModeEnabled(db); err != nil { + return fmt.Errorf("failover: candidate master %s:%d rejected: %w", candidate.Host, candidate.Port, err) + } + + candidateSetStr, err := ReadExecutedGTIDSet(db) + if err != nil { + return fmt.Errorf("failover: reading candidate master executed GTID set: %w", err) + } + candidateSet, err := mysql.ParseMysqlGTIDSet(candidateSetStr) + if err != nil { + return fmt.Errorf("failover: parsing candidate master executed GTID set %q: %w", candidateSetStr, err) + } + + if appliedSet != nil && !candidateSet.Contain(appliedSet) { + return fmt.Errorf( + "failover: candidate master %s:%d executed set %q does not contain already-applied set %q; resuming there would diverge from the source", + candidate.Host, candidate.Port, candidateSet.String(), appliedSet.String(), + ) + } + + return nil +} diff --git a/master_failover_test.go b/master_failover_test.go new file mode 100644 index 00000000..f42a2832 --- /dev/null +++ b/master_failover_test.go @@ -0,0 +1,145 @@ +package ghostferry + +import ( + "errors" + "testing" + "time" + + sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/go-mysql-org/go-mysql/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFailoverRecoveryEnabled(t *testing.T) { + reconnector := &fakeReconnector{} + resolver := MasterWriterResolverFunc(func(_ *DatabaseConfig) (*DatabaseConfig, error) { + return nil, nil + }) + _ = resolver + + // Disabled with no config. + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + assert.False(t, s.failoverRecoveryEnabled()) + + // Config but no reconnector. + s.MasterFailoverRecovery = &MasterFailoverRecoveryConfig{} + assert.False(t, s.failoverRecoveryEnabled()) + + // Config + reconnector but file/position mode. + s.BinlogCoordinateMode = BinlogCoordinateFilePosition + s.SourceReconnector = reconnector + assert.False(t, s.failoverRecoveryEnabled()) + + // GTID mode + config + reconnector: enabled. + s.BinlogCoordinateMode = BinlogCoordinateGTID + assert.True(t, s.failoverRecoveryEnabled()) +} + +func TestFailoverRetryWaitDefault(t *testing.T) { + c := &MasterFailoverRecoveryConfig{} + assert.Equal(t, DefaultFailoverRetryWait, c.retryWait()) + c.RetryWait = 2 * time.Second + assert.Equal(t, 2*time.Second, c.retryWait()) +} + +func TestMasterWriterResolverFunc(t *testing.T) { + want := &DatabaseConfig{Host: "new", Port: 3306} + prev := &DatabaseConfig{Host: "old", Port: 3306} + var got *DatabaseConfig + resolver := MasterWriterResolverFunc(func(p *DatabaseConfig) (*DatabaseConfig, error) { + got = p + return want, nil + }) + result, err := resolver.ResolveCurrentMaster(prev) + require.NoError(t, err) + assert.Same(t, want, result) + assert.Same(t, prev, got) +} + +func TestAppliedGTIDSetFoldsInFlight(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + s.logger = LogWithField("tag", "test") + + // Committed set only. + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetTarget) // :1-100 + applied, err := s.appliedGTIDSet() + require.NoError(t, err) + assert.Equal(t, gtidSetTarget, applied.String()) + + // With an in-flight GTID, the applied set extends to include it. + s.inFlightGTID = "3e11fa47-71ca-11e1-9e33-c80aa9429562:101" + applied, err = s.appliedGTIDSet() + require.NoError(t, err) + assert.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-101", applied.String()) + + // A malformed in-flight GTID fails closed. + s.inFlightGTID = "not-a-gtid" + _, err = s.appliedGTIDSet() + assert.Error(t, err) +} + +func TestGtidSetStringNil(t *testing.T) { + assert.Equal(t, "", gtidSetString(nil)) + assert.Equal(t, gtidSetTarget, gtidSetString(mustParseGTID(t, gtidSetTarget))) +} + +func TestCloneOrEmpty(t *testing.T) { + empty := cloneOrEmpty(nil) + require.NotNil(t, empty) + assert.Equal(t, "", empty.String()) + + set := mustParseGTID(t, gtidSetTarget) + clone := cloneOrEmpty(set) + assert.Equal(t, gtidSetTarget, clone.String()) + require.NoError(t, clone.(*mysql.MysqlGTIDSet).Update("3e11fa47-71ca-11e1-9e33-c80aa9429562:101")) + assert.Equal(t, gtidSetTarget, set.String(), "clone must not alias source") +} + +func TestUnionGTIDStringInto(t *testing.T) { + res, err := unionGTIDStringInto(nil, "3e11fa47-71ca-11e1-9e33-c80aa9429562:101") + require.NoError(t, err) + assert.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:101", res.String()) + + base := mustParseGTID(t, gtidSetTarget) // :1-100 + merged, err := unionGTIDStringInto(base, "3e11fa47-71ca-11e1-9e33-c80aa9429562:101") + require.NoError(t, err) + assert.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-101", merged.String()) + assert.Equal(t, gtidSetTarget, base.String(), "base must not be mutated") + + // The safety property: a candidate missing the in-flight GTID fails containment. + candidateMissing := mustParseGTID(t, gtidSetTarget) + assert.False(t, candidateMissing.Contain(merged)) + candidateHas := mustParseGTID(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-101") + assert.True(t, candidateHas.Contain(merged)) +} + +func TestUnionGTIDSets(t *testing.T) { + // nil inputs yield empty. + res, err := unionGTIDSets(nil, nil) + require.NoError(t, err) + assert.Equal(t, "", res.String()) + + applied := mustParseGTID(t, gtidSetTarget) // :1-100 + stop := mustParseGTID(t, gtidSetPast) // :1-150 + merged, err := unionGTIDSets(applied, stop) + require.NoError(t, err) + assert.Equal(t, gtidSetPast, merged.String()) + // Inputs unchanged. + assert.Equal(t, gtidSetTarget, applied.String()) + assert.Equal(t, gtidSetPast, stop.String()) + + // A candidate must contain the union of applied + stop target. + candidateShort := mustParseGTID(t, gtidSetTarget) // only :1-100 + assert.False(t, candidateShort.Contain(merged), "candidate missing stop-target GTIDs must fail") +} + +// fakeReconnector is a test double for SourceReconnector. +type fakeReconnector struct { + calls int +} + +func (f *fakeReconnector) Reconnect(_ *DatabaseConfig, _ mysql.GTIDSet) (*sql.DB, *DatabaseConfig, error) { + f.calls++ + return nil, nil, errors.New("not implemented") +} diff --git a/source_runtime.go b/source_runtime.go index ceeff36c..1a564482 100644 --- a/source_runtime.go +++ b/source_runtime.go @@ -71,11 +71,32 @@ func (r *SourceRuntime) Config() *DatabaseConfig { // If opening the new connection fails, the current source is left unchanged and // the error is returned. func (r *SourceRuntime) Replace(config *DatabaseConfig, logger Logger) (*sql.DB, error) { + return r.ReplaceValidated(config, logger, nil) +} + +// ReplaceValidated opens a new source connection from the given config, runs an +// optional validate hook against it, and only installs it as the current source +// if validation succeeds. This keeps the swap fail-closed: a candidate that +// fails validation is never published to consumers. +// +// If opening the connection or the validate hook fails, the newly opened handle +// is closed and the current source is left unchanged. On success the previous +// handle is retired (retained, not closed) as with Replace. +func (r *SourceRuntime) ReplaceValidated(config *DatabaseConfig, logger Logger, validate func(*sql.DB) error) (*sql.DB, error) { newDB, err := config.SqlDB(logger) if err != nil { return nil, err } + if validate != nil { + if err := validate(newDB); err != nil { + // The candidate is unsafe; do not publish it. Close the handle we + // opened for validation. + _ = newDB.Close() + return nil, err + } + } + r.mu.Lock() defer r.mu.Unlock() diff --git a/source_runtime_test.go b/source_runtime_test.go index cf421ca4..5cb14880 100644 --- a/source_runtime_test.go +++ b/source_runtime_test.go @@ -1,6 +1,7 @@ package ghostferry import ( + "errors" "testing" sql "github.com/Shopify/ghostferry/sqlwrapper" @@ -68,6 +69,41 @@ func TestSourceRuntimeCloseRetiredDrains(t *testing.T) { assert.Empty(t, rt.retired, "retired handles must be drained") } +func TestSourceRuntimeReplaceValidatedFailureDoesNotPublish(t *testing.T) { + oldDB := &sql.DB{Marginalia: "old"} + oldCfg := runtimeTestConfig("old-host") + rt := NewSourceRuntime(oldDB, oldCfg) + + validateErr := errors.New("candidate rejected") + _, err := rt.ReplaceValidated(runtimeTestConfig("bad-host"), nil, func(_ *sql.DB) error { + return validateErr + }) + assert.ErrorIs(t, err, validateErr) + + // The current source must be unchanged and nothing retired. + assert.Same(t, oldDB, rt.DB()) + assert.Same(t, oldCfg, rt.Config()) + assert.Empty(t, rt.retired) +} + +func TestSourceRuntimeReplaceValidatedSuccessPublishes(t *testing.T) { + oldDB := &sql.DB{Marginalia: "old"} + rt := NewSourceRuntime(oldDB, runtimeTestConfig("old-host")) + + validated := false + newCfg := runtimeTestConfig("new-host") + newDB, err := rt.ReplaceValidated(newCfg, nil, func(db *sql.DB) error { + validated = true + assert.NotNil(t, db) + return nil + }) + require.NoError(t, err) + assert.True(t, validated) + assert.Same(t, newDB, rt.DB()) + assert.Same(t, newCfg, rt.Config()) + assert.Contains(t, rt.retired, oldDB) +} + func TestSourceRuntimeNilInitial(t *testing.T) { rt := NewSourceRuntime(nil, nil) assert.Nil(t, rt.DB()) diff --git a/test/go/config_test.go b/test/go/config_test.go index 50b8d0f1..c015ba2f 100644 --- a/test/go/config_test.go +++ b/test/go/config_test.go @@ -220,6 +220,35 @@ func (this *ConfigTestSuite) TestBinlogCoordinateModeInvalidIsRejected() { this.Require().EqualError(err, `invalid BinlogCoordinateMode "banana", must be "file_position" or "gtid"`) } +func (this *ConfigTestSuite) TestMasterFailoverRecoveryRequiresGTIDMode() { + this.config.BinlogCoordinateMode = ghostferry.BinlogCoordinateFilePosition + this.config.MasterFailoverRecovery = &ghostferry.MasterFailoverRecoveryConfig{ + Resolver: ghostferry.MasterWriterResolverFunc(func(_ *ghostferry.DatabaseConfig) (*ghostferry.DatabaseConfig, error) { + return nil, nil + }), + } + err := this.config.ValidateConfig() + this.Require().EqualError(err, `MasterFailoverRecovery requires BinlogCoordinateMode "gtid", got "file_position"`) +} + +func (this *ConfigTestSuite) TestMasterFailoverRecoveryRequiresResolver() { + this.config.BinlogCoordinateMode = ghostferry.BinlogCoordinateGTID + this.config.MasterFailoverRecovery = &ghostferry.MasterFailoverRecoveryConfig{} + err := this.config.ValidateConfig() + this.Require().EqualError(err, "MasterFailoverRecovery requires a Resolver") +} + +func (this *ConfigTestSuite) TestMasterFailoverRecoveryValidInGTIDMode() { + this.config.BinlogCoordinateMode = ghostferry.BinlogCoordinateGTID + this.config.MasterFailoverRecovery = &ghostferry.MasterFailoverRecoveryConfig{ + Resolver: ghostferry.MasterWriterResolverFunc(func(_ *ghostferry.DatabaseConfig) (*ghostferry.DatabaseConfig, error) { + return nil, nil + }), + } + err := this.config.ValidateConfig() + this.Require().Nil(err) +} + func TestConfig(t *testing.T) { testhelpers.SetupTest() suite.Run(t, new(ConfigTestSuite)) From c94372da73ea4e15cbf7cd121a6dc64ce1a530ce Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Tue, 28 Jul 2026 10:13:37 +0200 Subject: [PATCH 2/2] Add Go integration test for master failover recovery Drives a full source master failover and asserts Ghostferry reconnects to the promoted writer and finishes the move against it. The test controls ordering directly (no status-callback timing races): - sets mysql-3 (port 29293, already in the compose files) up as a GTID replica of the source, aligning gtid_purged so it holds the real seed and later writes; - runs the ferry, and only AFTER row copy completes (so the data iterator is done reading the source) writes extra rows, waits for mysql-3 to replicate them, promotes mysql-3, and stops the source container; - synchronizes on SourceRuntime().Config().Port flipping to the promoted port before driving cutover, so cutover records its stop coordinate from the new writer; - asserts the promoted master and target are identical and that the ferry did not fatal. GTID-only (skips otherwise) and requires docker/podman to stop the source container (skips otherwise). start-mysql.sh now also starts/waits for mysql-3. --- .github/workflows/start-mysql.sh | 11 +- .gitignore | 3 + test/go/master_failover_integration_test.go | 424 ++++++++++++++++++++ 3 files changed, 435 insertions(+), 3 deletions(-) create mode 100644 test/go/master_failover_integration_test.go diff --git a/.github/workflows/start-mysql.sh b/.github/workflows/start-mysql.sh index 89b423f7..56c8ef61 100755 --- a/.github/workflows/start-mysql.sh +++ b/.github/workflows/start-mysql.sh @@ -1,12 +1,15 @@ #!/bin/bash set -xe +# mysql-3 is the promoted writer used by the master-failover integration test. +# It is started alongside the source/target so that test can run; other tests +# ignore it. if [ "$MYSQL_VERSION" == "8.0" ]; then - docker compose -f docker-compose_8.0.yml up -d mysql-1 mysql-2 + docker compose -f docker-compose_8.0.yml up -d mysql-1 mysql-2 mysql-3 elif [ "$MYSQL_VERSION" == "8.4" ]; then - docker compose -f docker-compose_8.4.yml up -d mysql-1 mysql-2 + docker compose -f docker-compose_8.4.yml up -d mysql-1 mysql-2 mysql-3 else - docker compose up -d mysql-1 mysql-2 + docker compose up -d mysql-1 mysql-2 mysql-3 fi MAX_ATTEMPTS=60 @@ -38,6 +41,8 @@ wait_for_configuration () { wait_for_version "ghostferry-mysql-1-1" wait_for_version "ghostferry-mysql-2-1" +wait_for_version "ghostferry-mysql-3-1" wait_for_configuration 29291 wait_for_configuration 29292 +wait_for_configuration 29293 diff --git a/.gitignore b/.gitignore index 3e50b25c..1c572c52 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,6 @@ build/ .idea .ipynb_checkpoints + +# Local test tmpdir (macOS TMPDIR workaround) +.tmp_ghostferry/ diff --git a/test/go/master_failover_integration_test.go b/test/go/master_failover_integration_test.go new file mode 100644 index 00000000..b7175794 --- /dev/null +++ b/test/go/master_failover_integration_test.go @@ -0,0 +1,424 @@ +package test + +import ( + "fmt" + "os" + "os/exec" + "strings" + "sync" + "testing" + "time" + + sqlorig "database/sql" + + "github.com/Shopify/ghostferry" + sqlwrapper "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/Shopify/ghostferry/testhelpers" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The promoted-writer ("future master") MySQL. It exists in the docker-compose +// files as mysql-3 and is started by .github/workflows/start-mysql.sh. There is +// no N3_PORT helper, so hardcode it (matching the Ruby harness convention). +const futureMasterPort = 29293 + +// TestMasterFailoverRecoveryReconnectsToPromotedMaster drives a full source +// master failover and asserts Ghostferry reconnects to the promoted writer and +// finishes the move against it. +// +// It is GTID-only (failover recovery requires server-independent coordinates) +// and needs a container tool (docker/podman) to stop the source, so it skips +// when either precondition is absent. +func TestMasterFailoverRecoveryReconnectsToPromotedMaster(t *testing.T) { + if binlogModeFromEnv() != ghostferry.BinlogCoordinateGTID { + t.Skip("master failover recovery is only supported in gtid mode") + } + tool := containerTool(t) + if tool == "" { + t.Skip("master failover test requires docker or podman to stop the source container") + } + + sourcePort := testhelpers.TestSourcePort + targetPort := testhelpers.TestTargetPort + + sourceDB := openDB(t, sourcePort) + defer sourceDB.Close() + targetDB := openDB(t, targetPort) + defer targetDB.Close() + promotedDB := openDB(t, uint64(futureMasterPort)) + defer promotedDB.Close() + + // Clean slate on all three. + dropTestDBs(t, sourceDB) + dropTestDBs(t, targetDB) + dropTestDBs(t, promotedDB) + + fm := &futureMaster{t: t, db: promotedDB, sourceDB: sourceDB, containerTool: tool} + // Ensure the promoted server is restored even if the test fails. + defer fm.restoreSourceContainer() + defer fm.reset() + + // mysql-3 replicates the source BEFORE seeding, so it holds the real seed + // data and later writes; a failover to it then cannot lose applied + // transactions. + fm.setupAsReplicaOfSource() + + testhelpers.SeedInitialData(sourceDB, "gftest", "table1", 30) + testhelpers.SeedInitialData(targetDB, "gftest", "table1", 0) + + promoted := &ghostferry.DatabaseConfig{ + Host: "127.0.0.1", + Port: uint16(futureMasterPort), + Net: "tcp", + User: "root", + Pass: "", + Collation: "utf8mb4_unicode_ci", + Params: map[string]string{"charset": "utf8mb4"}, + } + + ferry := testhelpers.NewTestFerry() + ferry.Config.BinlogCoordinateMode = ghostferry.BinlogCoordinateGTID + ferry.Config.MasterFailoverRecovery = &ghostferry.MasterFailoverRecoveryConfig{ + Resolver: ghostferry.MasterWriterResolverFunc(func(_ *ghostferry.DatabaseConfig) (*ghostferry.DatabaseConfig, error) { + return promoted, nil + }), + MaxAttempts: 20, + RetryWait: 250 * time.Millisecond, + } + require.NoError(t, ferry.Config.ValidateConfig()) + + errHandler := &testhelpers.ErrorHandler{} + ferry.ErrorHandler = errHandler + + require.NoError(t, ferry.Initialize()) + require.NoError(t, ferry.Start()) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + ferry.Run() + }() + + // Wait until row copy is fully complete before touching the source, so the + // data iterator is no longer reading it when we take it down. (Failover + // during row copy is out of scope: in-flight cursors are bound to their + // connection for scan consistency.) + ferry.WaitUntilRowCopyIsComplete() + + // Now fail the source over while the binlog streamer is still live and has + // no stop target yet: + // 1. write extra rows to the source and wait for mysql-3 to replicate them, + // so they exist on the promoted master but have NOT been streamed yet; + // 2. promote mysql-3 and stop the old source container. + // The streamer then finds the old source gone and must recover onto mysql-3 + // to stream those rows before cutover completes. + insertRows(t, sourceDB, 50) + fm.waitUntilCaughtUpToSource() + fm.promote() + fm.stopSourceContainer() + + // Wait for the streamer to actually recover onto the promoted master before + // driving cutover, so cutover records its stop coordinate from the new + // writer. (In Go we control the flow directly, so we can synchronize here + // rather than racing the cutover path against recovery.) + waitUntilSourceRepointed(t, ferry, uint16(futureMasterPort), errHandler) + + ferry.FlushBinlogAndStopStreaming() + wg.Wait() + ferry.StopTargetVerifier() + + require.NoError(t, errHandler.LastError, "ghostferry should not have fataled") + + // The promoted master (mysql-3) is now the source of truth; the target must + // be identical to it. + assertTablesIdentical(t, promotedDB, targetDB, "gftest", "table1") + + // Recovery must actually have run and repointed the source. + assert.NotEqual(t, uint16(sourcePort), ferry.SourceRuntime().Config().Port, + "the ferry source should have been repointed away from the dead master") + assert.Equal(t, uint16(futureMasterPort), ferry.SourceRuntime().Config().Port, + "the ferry source should now be the promoted master") +} + +// waitUntilSourceRepointed blocks until the ferry's source runtime has been +// swapped to the expected (promoted) port, or the ferry fataled, or a timeout. +func waitUntilSourceRepointed(t *testing.T, ferry *testhelpers.TestFerry, wantPort uint16, errHandler *testhelpers.ErrorHandler) { + t.Helper() + deadline := time.Now().Add(60 * time.Second) + for time.Now().Before(deadline) { + if errHandler.LastError != nil { + t.Fatalf("ghostferry fataled during failover recovery: %v", errHandler.LastError) + } + if cfg := ferry.SourceRuntime().Config(); cfg != nil && cfg.Port == wantPort { + return + } + time.Sleep(200 * time.Millisecond) + } + t.Fatalf("source was not repointed to port %d within timeout", wantPort) +} + +// ---- helpers ---- + +func binlogModeFromEnv() ghostferry.BinlogCoordinateType { + mode := os.Getenv("GHOSTFERRY_BINLOG_COORDINATE_MODE") + if mode == "" { + return ghostferry.BinlogCoordinateFilePosition + } + return ghostferry.BinlogCoordinateType(mode) +} + +func openDB(t *testing.T, port uint64) *sqlwrapper.DB { + t.Helper() + dsn := fmt.Sprintf("root@tcp(127.0.0.1:%d)/?charset=utf8mb4&collation=utf8mb4_unicode_ci&interpolateParams=true", port) + db, err := sqlwrapper.Open("mysql", dsn, "") + require.NoError(t, err) + require.NoError(t, db.Ping()) + return db +} + +func dropTestDBs(t *testing.T, db *sqlwrapper.DB) { + t.Helper() + for _, name := range testhelpers.ApplicableTestDbs { + _, err := db.Exec(fmt.Sprintf("DROP DATABASE IF EXISTS `%s`", name)) + require.NoError(t, err) + } +} + +func insertRows(t *testing.T, db *sqlwrapper.DB, count int) { + t.Helper() + for i := 0; i < count; i++ { + _, err := db.Exec("INSERT INTO gftest.table1 (id, data) VALUES (?, ?)", nil, testhelpers.RandData()) + require.NoError(t, err) + } +} + +func assertTablesIdentical(t *testing.T, a, b *sqlwrapper.DB, dbName, table string) { + t.Helper() + countA := tableRowCount(t, a, dbName, table) + countB := tableRowCount(t, b, dbName, table) + assert.Greater(t, countA, 0, "promoted master should have rows") + assert.Equal(t, countA, countB, "promoted master and target row counts differ") + + checksumA := tableChecksum(t, a, dbName, table) + checksumB := tableChecksum(t, b, dbName, table) + assert.Equal(t, checksumA, checksumB, "promoted master and target checksums differ") +} + +func tableRowCount(t *testing.T, db *sqlwrapper.DB, dbName, table string) int { + t.Helper() + var n int + require.NoError(t, db.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM `%s`.`%s`", dbName, table)).Scan(&n)) + return n +} + +func tableChecksum(t *testing.T, db *sqlwrapper.DB, dbName, table string) int64 { + t.Helper() + var name string + var checksum sqlorig.NullInt64 + require.NoError(t, db.QueryRow(fmt.Sprintf("CHECKSUM TABLE `%s`.`%s`", dbName, table)).Scan(&name, &checksum)) + return checksum.Int64 +} + +// containerTool returns "docker" or "podman" if one is on PATH and can see the +// source container, else "". +func containerTool(t *testing.T) string { + for _, tool := range []string{"docker", "podman"} { + if _, err := exec.LookPath(tool); err != nil { + continue + } + if sourceContainerName(tool) != "" { + return tool + } + } + return "" +} + +func sourceContainerName(tool string) string { + out, err := exec.Command(tool, "ps", "-a", "--format", "{{.Names}}").Output() + if err != nil { + return "" + } + names := strings.Split(string(out), "\n") + for _, candidate := range []string{"ghostferry_mysql-1_1", "ghostferry-mysql-1-1"} { + for _, n := range names { + if strings.TrimSpace(n) == candidate { + return candidate + } + } + } + return "" +} + +// futureMaster manages mysql-3 as a promotable replica of the source. +type futureMaster struct { + t *testing.T + db *sqlwrapper.DB + sourceDB *sqlwrapper.DB + containerTool string +} + +func (f *futureMaster) exec(db *sqlwrapper.DB, query string) { + f.t.Helper() + _, err := db.Exec(query) + require.NoError(f.t, err, "query: %s", query) +} + +func (f *futureMaster) execIgnore(db *sqlwrapper.DB, query string) { + _, _ = db.Exec(query) +} + +func (f *futureMaster) sourceExecutedGTIDSet() string { + f.t.Helper() + var gtid string + require.NoError(f.t, f.sourceDB.QueryRow("SELECT @@GLOBAL.gtid_executed").Scan(>id)) + return strings.Join(strings.Fields(gtid), "") +} + +func (f *futureMaster) resetBinaryLogsStatement() string { + var version string + require.NoError(f.t, f.db.QueryRow("SELECT VERSION()").Scan(&version)) + if versionAtLeast(version, 8, 4) { + return "RESET BINARY LOGS AND GTIDS" + } + return "RESET MASTER" +} + +func (f *futureMaster) setupAsReplicaOfSource() { + f.t.Helper() + f.execIgnore(f.db, "STOP REPLICA") + f.execIgnore(f.db, "RESET REPLICA ALL") + f.execIgnore(f.db, "SET GLOBAL read_only = OFF") + dropTestDBs(f.t, f.db) + + sourceGTID := f.sourceExecutedGTIDSet() + f.exec(f.db, f.resetBinaryLogsStatement()) + if sourceGTID != "" { + f.exec(f.db, fmt.Sprintf("SET GLOBAL gtid_purged = '%s'", sourceGTID)) + } + f.exec(f.db, "CHANGE REPLICATION SOURCE TO SOURCE_HOST='mysql-1', SOURCE_PORT=3306, SOURCE_USER='root', SOURCE_AUTO_POSITION=1") + f.exec(f.db, "START REPLICA") + f.waitUntilReplicaRunning() +} + +func (f *futureMaster) waitUntilReplicaRunning() { + f.t.Helper() + deadline := time.Now().Add(20 * time.Second) + for { + io, sql := f.replicaRunning() + if io && sql { + return + } + if time.Now().After(deadline) { + f.t.Fatalf("future master replica did not start in time") + } + time.Sleep(200 * time.Millisecond) + } +} + +func (f *futureMaster) replicaRunning() (ioRunning, sqlRunning bool) { + rows, err := f.db.Query("SHOW REPLICA STATUS") + require.NoError(f.t, err) + defer rows.Close() + + cols, err := rows.Columns() + require.NoError(f.t, err) + if !rows.Next() { + return false, false + } + vals := make([]interface{}, len(cols)) + ptrs := make([]interface{}, len(cols)) + for i := range vals { + ptrs[i] = &vals[i] + } + require.NoError(f.t, rows.Scan(ptrs...)) + for i, c := range cols { + s := asString(vals[i]) + if c == "Replica_IO_Running" { + ioRunning = s == "Yes" + } + if c == "Replica_SQL_Running" { + sqlRunning = s == "Yes" + } + } + return ioRunning, sqlRunning +} + +func (f *futureMaster) waitUntilCaughtUpToSource() { + f.t.Helper() + sourceGTID := f.sourceExecutedGTIDSet() + if sourceGTID == "" { + return + } + var result sqlorig.NullInt64 + require.NoError(f.t, f.db.QueryRow("SELECT WAIT_FOR_EXECUTED_GTID_SET(?, 20)", sourceGTID).Scan(&result)) + require.Equal(f.t, int64(0), result.Int64, "future master did not catch up to source in time") +} + +func (f *futureMaster) promote() { + f.t.Helper() + f.waitUntilCaughtUpToSource() + f.exec(f.db, "STOP REPLICA") + f.exec(f.db, "RESET REPLICA ALL") + f.exec(f.db, "SET GLOBAL read_only = OFF") + f.execIgnore(f.db, "SET GLOBAL super_read_only = OFF") +} + +func (f *futureMaster) stopSourceContainer() { + f.t.Helper() + name := sourceContainerName(f.containerTool) + require.NotEmpty(f.t, name, "source container not found") + require.NoError(f.t, exec.Command(f.containerTool, "stop", name).Run()) +} + +func (f *futureMaster) restoreSourceContainer() { + name := sourceContainerName(f.containerTool) + if name == "" { + return + } + _ = exec.Command(f.containerTool, "start", name).Run() + // Wait until the source accepts connections again for later tests. + deadline := time.Now().Add(180 * time.Second) + for time.Now().Before(deadline) { + db, err := sqlwrapper.Open("mysql", fmt.Sprintf("root@tcp(127.0.0.1:%d)/", testhelpers.TestSourcePort), "") + if err == nil { + pingErr := db.Ping() + db.Close() + if pingErr == nil { + return + } + } + time.Sleep(500 * time.Millisecond) + } +} + +func (f *futureMaster) reset() { + f.execIgnore(f.db, "STOP REPLICA") + f.execIgnore(f.db, "RESET REPLICA ALL") + f.execIgnore(f.db, "SET GLOBAL read_only = OFF") + for _, name := range testhelpers.ApplicableTestDbs { + f.execIgnore(f.db, fmt.Sprintf("DROP DATABASE IF EXISTS `%s`", name)) + } +} + +func asString(v interface{}) string { + switch t := v.(type) { + case []byte: + return string(t) + case string: + return t + case nil: + return "" + default: + return fmt.Sprintf("%v", t) + } +} + +func versionAtLeast(version string, major, minor int) bool { + var maj, min int + if _, err := fmt.Sscanf(version, "%d.%d", &maj, &min); err != nil { + return false + } + return maj > major || (maj == major && min >= minor) +}