From 195afe71a536fcc20a98b4d2ebac35b2af270f30 Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Thu, 23 Jul 2026 10:38:25 -0400 Subject: [PATCH 1/3] Migrate source consumers to SourceRuntime Have the source-side components resolve the current source connection from the Ferry's SourceRuntime instead of holding a captured *sql.DB, so a future source master failover can repoint them all by swapping the runtime rather than editing each consumer. Consumers migrated (each keeps its static DB as a fallback when no runtime is configured, so behavior is unchanged today): - CursorConfig: NewCursor binds a cursor to the runtime's current handle at creation time. A cursor keeps that handle for its whole scan (a scan must be consistent against one server); only cursors created after a swap observe the new source. - DataIterator: resolves the source once at the start of Run. - InlineVerifier: currentSourceDB resolves the runtime handle and, on a detected swap, resets the source statement cache (cached statements are bound to the previous DB); the DB and its matching cache are returned together under a lock so concurrent workers stay consistent. - IterativeVerifier: currentSourceDB resolves the runtime handle for fingerprint/compressed-hash reads; its CursorConfig also carries the runtime for pagination scans. - ChecksumTableVerifier: currentSourceDB resolves the runtime handle for CHECKSUM TABLE reads. Ferry wires f.sourceRuntime into each constructor. No swap happens yet (that arrives with failover), so the runtime always resolves to the same initial handle and behavior is unchanged. --- cursor.go | 25 +++++++- data_iterator.go | 18 +++++- ferry.go | 10 ++- inline_verifier.go | 37 ++++++++++- iterative_verifier.go | 23 ++++++- source_runtime_migration_test.go | 106 +++++++++++++++++++++++++++++++ verifier.go | 20 +++++- 7 files changed, 226 insertions(+), 13 deletions(-) create mode 100644 source_runtime_migration_test.go diff --git a/cursor.go b/cursor.go index 3c83a1ea..f533eec4 100644 --- a/cursor.go +++ b/cursor.go @@ -36,6 +36,15 @@ type CursorConfig struct { DB *sql.DB Throttler Throttler + // SourceRuntime, when set, is the authoritative source of the current + // source connection. NewCursor resolves the cursor's DB from it at cursor + // creation time so that cursors started after a source master failover run + // against the promoted writer. When nil, the static DB field is used. A + // cursor captures the resolved handle for the duration of its scan (a scan + // must be consistent against a single server); only newly created cursors + // observe a swap. + SourceRuntime *SourceRuntime + ColumnsToSelect []string BuildSelect func([]string, *TableSchema, PaginationKey, uint64) (squirrel.SelectBuilder, error) // BatchSize is a pointer to the BatchSize in Config.UpdatableConfig which can be independently updated from this code. @@ -45,10 +54,24 @@ type CursorConfig struct { ReadRetries int } +// resolvedDB returns the source connection to use for a new cursor: the current +// handle from SourceRuntime when configured, otherwise the static DB. +func (c *CursorConfig) resolvedDB() *sql.DB { + if c.SourceRuntime != nil { + if db := c.SourceRuntime.DB(); db != nil { + return db + } + } + return c.DB +} + // returns a new Cursor with an embedded copy of itself func (c *CursorConfig) NewCursor(table *TableSchema, startPaginationKey, maxPaginationKey PaginationKey) *Cursor { + cfg := *c + // Bind the cursor to the current source connection for the whole scan. + cfg.DB = c.resolvedDB() return &Cursor{ - CursorConfig: *c, + CursorConfig: cfg, Table: table, MaxPaginationKey: maxPaginationKey, RowLock: true, diff --git a/data_iterator.go b/data_iterator.go index 0586fd8a..7fd6ea53 100644 --- a/data_iterator.go +++ b/data_iterator.go @@ -8,7 +8,11 @@ import ( ) type DataIterator struct { - DB *sql.DB + DB *sql.DB + // SourceRuntime, when set, provides the current source connection so a data + // iteration run started after a source master failover targets the promoted + // writer. When nil, DB is used. Resolved once at the start of Run. + SourceRuntime *SourceRuntime Concurrency int SelectFingerprint bool @@ -38,8 +42,18 @@ func (d *DataIterator) Run(tables []*TableSchema) { d.StateTracker = NewStateTracker(0) } + // Resolve the source connection for this run. When a SourceRuntime is + // configured it is authoritative (so a run started after a failover uses the + // promoted writer); otherwise fall back to the static DB. + db := d.DB + if d.SourceRuntime != nil { + if current := d.SourceRuntime.DB(); current != nil { + db = current + } + } + d.logger.WithField("tablesCount", len(tables)).Info("starting data iterator run") - tablesWithData, emptyTables, err := MaxPaginationKeys(d.DB, tables, d.logger) + tablesWithData, emptyTables, err := MaxPaginationKeys(db, tables, d.logger) if err != nil { d.ErrorHandler.Fatal("data_iterator", err) } diff --git a/ferry.go b/ferry.go index 4d822ebd..8de2b23f 100644 --- a/ferry.go +++ b/ferry.go @@ -102,13 +102,15 @@ func (f *Ferry) NewDataIterator() *DataIterator { dataIterator := &DataIterator{ DB: f.SourceDB, + SourceRuntime: f.sourceRuntime, Concurrency: f.Config.DataIterationConcurrency, SelectFingerprint: f.Config.VerifierType == VerifierTypeInline, ErrorHandler: f.ErrorHandler, CursorConfig: &CursorConfig{ - DB: f.SourceDB, - Throttler: f.Throttler, + DB: f.SourceDB, + SourceRuntime: f.sourceRuntime, + Throttler: f.Throttler, BatchSize: &f.Config.UpdatableConfig.DataIterationBatchSize, BatchSizePerTableOverride: f.Config.DataIterationBatchSizePerTableOverride, @@ -220,6 +222,7 @@ func (f *Ferry) NewChecksumTableVerifier() *ChecksumTableVerifier { return &ChecksumTableVerifier{ SourceDB: f.SourceDB, + SourceRuntime: f.sourceRuntime, TargetDB: f.TargetDB, DatabaseRewrites: f.Config.DatabaseRewrites, TableRewrites: f.Config.TableRewrites, @@ -239,6 +242,7 @@ func (f *Ferry) NewInlineVerifier() *InlineVerifier { return &InlineVerifier{ SourceDB: f.SourceDB, + SourceRuntime: f.sourceRuntime, TargetDB: f.TargetDB, DatabaseRewrites: f.Config.DatabaseRewrites, TableRewrites: f.Config.TableRewrites, @@ -312,6 +316,7 @@ func (f *Ferry) NewIterativeVerifier() (*IterativeVerifier, error) { v := &IterativeVerifier{ CursorConfig: &CursorConfig{ DB: f.SourceDB, + SourceRuntime: f.sourceRuntime, BatchSize: &f.Config.UpdatableConfig.DataIterationBatchSize, BatchSizePerTableOverride: f.Config.DataIterationBatchSizePerTableOverride, ReadRetries: f.Config.DBReadRetries, @@ -319,6 +324,7 @@ func (f *Ferry) NewIterativeVerifier() (*IterativeVerifier, error) { BinlogStreamer: f.BinlogStreamer, SourceDB: f.SourceDB, + SourceRuntime: f.sourceRuntime, TargetDB: f.TargetDB, CompressionVerifier: compressionVerifier, diff --git a/inline_verifier.go b/inline_verifier.go index 909b3f90..7bf347a1 100644 --- a/inline_verifier.go +++ b/inline_verifier.go @@ -257,7 +257,11 @@ type InlineVerifierMismatches struct { } type InlineVerifier struct { - SourceDB *sql.DB + SourceDB *sql.DB + // SourceRuntime, when set, is the authoritative source of the current + // source connection (see currentSourceDB). It lets inline verification + // follow a source master failover. When nil, SourceDB is used directly. + SourceRuntime *SourceRuntime TargetDB *sql.DB DatabaseRewrites map[string]string TableRewrites map[string]string @@ -274,7 +278,10 @@ type InlineVerifier struct { sourceStmtCache *StmtCache targetStmtCache *StmtCache - logger Logger + // sourceDBMu guards SourceDB and sourceStmtCache when currentSourceDB + // resolves the runtime handle and may reset the cache on a failover swap. + sourceDBMu sync.Mutex + logger Logger // Used only for the ControlServer initiated VerifyDuringCutover backgroundVerificationResultAndStatus VerificationResultAndStatus @@ -524,8 +531,32 @@ func (v *InlineVerifier) VerifyDuringCutover() (VerificationResult, error) { }, nil } +// currentSourceDB returns the source connection inline verification should use, +// resolving it from SourceRuntime when configured so a failover is followed. If +// the source handle changed since the last call, the source statement cache is +// reset because cached prepared statements are bound to the previous DB. +// currentSourceDB returns the source connection inline verification should use, +// resolving it from SourceRuntime when configured so a failover is followed. If +// the source handle changed since the last call, the source statement cache is +// reset because cached prepared statements are bound to the previous DB. The +// resolved DB and its matching statement cache are returned together under a +// lock so concurrent verification workers stay consistent. +func (v *InlineVerifier) currentSourceDB() (*sql.DB, *StmtCache) { + v.sourceDBMu.Lock() + defer v.sourceDBMu.Unlock() + + if v.SourceRuntime != nil { + if current := v.SourceRuntime.DB(); current != nil && current != v.SourceDB { + v.SourceDB = current + v.sourceStmtCache = NewStmtCache() + } + } + return v.SourceDB, v.sourceStmtCache +} + func (v *InlineVerifier) getFingerprintDataFromSourceDb(schemaName, tableName string, tx *sql.Tx, table *TableSchema, paginationKeys []interface{}) (map[string][]byte, map[string]map[string][]byte, error) { - return v.getFingerprintDataFromDb(v.SourceDB, v.sourceStmtCache, schemaName, tableName, tx, table, paginationKeys) + db, stmtCache := v.currentSourceDB() + return v.getFingerprintDataFromDb(db, stmtCache, schemaName, tableName, tx, table, paginationKeys) } func (v *InlineVerifier) getFingerprintDataFromTargetDb(schemaName, tableName string, tx *sql.Tx, table *TableSchema, paginationKeys []interface{}) (map[string][]byte, map[string]map[string][]byte, error) { diff --git a/iterative_verifier.go b/iterative_verifier.go index 0ba1b07c..92f26ade 100644 --- a/iterative_verifier.go +++ b/iterative_verifier.go @@ -120,7 +120,11 @@ type IterativeVerifier struct { BinlogStreamer *BinlogStreamer TableSchemaCache TableSchemaCache SourceDB *sql.DB - TargetDB *sql.DB + // SourceRuntime, when set, provides the current source connection so + // iterative verification follows a source master failover. When nil, + // SourceDB is used. See currentSourceDB. + SourceRuntime *SourceRuntime + TargetDB *sql.DB Tables []*TableSchema IgnoredTables []string @@ -289,6 +293,19 @@ func (v *IterativeVerifier) Result() (VerificationResultAndStatus, error) { return v.verificationResultAndStatus, v.verificationErr } +// currentSourceDB resolves the source connection from SourceRuntime when +// configured (so verification follows a source master failover), otherwise the +// static SourceDB. Unlike the inline verifier there is no persistent +// source-bound statement cache to reset: GetHashes prepares per call. +func (v *IterativeVerifier) currentSourceDB() *sql.DB { + if v.SourceRuntime != nil { + if db := v.SourceRuntime.DB(); db != nil { + return db + } + } + return v.SourceDB +} + func (v *IterativeVerifier) GetHashes(db *sql.DB, schemaName, tableName, paginationKeyColumn string, columns []schema.TableColumn, paginationKeys []interface{}) (map[string][]byte, error) { sql, args, err := GetMd5HashesSql(schemaName, tableName, paginationKeyColumn, columns, paginationKeys) if err != nil { @@ -595,7 +612,7 @@ func (v *IterativeVerifier) compareFingerprints(paginationKeys []interface{}, ta go func() { defer wg.Done() sourceErr = WithRetries(5, 0, v.logger, "get fingerprints from source db", func() (err error) { - sourceHashes, err = v.GetHashes(v.SourceDB, table.Schema, table.Name, table.GetPaginationColumn().Name, v.columnsToVerify(table), paginationKeys) + sourceHashes, err = v.GetHashes(v.currentSourceDB(), table.Schema, table.Name, table.GetPaginationColumn().Name, v.columnsToVerify(table), paginationKeys) return }) }() @@ -627,7 +644,7 @@ func (v *IterativeVerifier) compareFingerprints(paginationKeys []interface{}, ta } func (v *IterativeVerifier) compareCompressedHashes(targetDb, targetTable string, table *TableSchema, paginationKeys []interface{}) ([]string, error) { - sourceHashes, err := v.CompressionVerifier.GetCompressedHashes(v.SourceDB, table.Schema, table.Name, table.GetPaginationColumn().Name, v.columnsToVerify(table), paginationKeys) + sourceHashes, err := v.CompressionVerifier.GetCompressedHashes(v.currentSourceDB(), table.Schema, table.Name, table.GetPaginationColumn().Name, v.columnsToVerify(table), paginationKeys) if err != nil { return nil, err } diff --git a/source_runtime_migration_test.go b/source_runtime_migration_test.go new file mode 100644 index 00000000..2b447337 --- /dev/null +++ b/source_runtime_migration_test.go @@ -0,0 +1,106 @@ +package ghostferry + +import ( + "testing" + + sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// These tests verify that source consumers resolve the current source +// connection from a SourceRuntime when one is configured, and fall back to the +// static DB otherwise. sql.DB values are sentinels; no query is executed. + +func TestCursorConfigResolvedDB(t *testing.T) { + staticDB := &sql.DB{Marginalia: "static"} + + // No runtime: uses static DB. + c := &CursorConfig{DB: staticDB} + assert.Same(t, staticDB, c.resolvedDB()) + + // Runtime with a current DB: uses runtime DB. + runtimeDB := &sql.DB{Marginalia: "runtime"} + c.SourceRuntime = NewSourceRuntime(runtimeDB, runtimeTestConfig("h")) + assert.Same(t, runtimeDB, c.resolvedDB()) + + // Runtime present but nil DB: falls back to static DB. + c.SourceRuntime = NewSourceRuntime(nil, nil) + assert.Same(t, staticDB, c.resolvedDB()) +} + +func TestNewCursorBindsToResolvedDB(t *testing.T) { + staticDB := &sql.DB{Marginalia: "static"} + runtimeDB := &sql.DB{Marginalia: "runtime"} + rt := NewSourceRuntime(staticDB, runtimeTestConfig("old")) + + cfg := &CursorConfig{DB: staticDB, SourceRuntime: rt} + table := &TableSchema{} + + // Cursor created now binds to the current (static) handle. + cursor := cfg.NewCursor(table, NewUint64Key(0), NewUint64Key(10)) + assert.Same(t, staticDB, cursor.DB) + + // After a runtime swap, a newly created cursor binds to the new handle, + // while the already-created cursor keeps its handle (scan consistency). + _, err := rt.Replace(runtimeTestConfig("new"), nil) + require.NoError(t, err) + newHandle := rt.DB() + + cursor2 := cfg.NewCursor(table, NewUint64Key(0), NewUint64Key(10)) + assert.Same(t, newHandle, cursor2.DB) + assert.Same(t, staticDB, cursor.DB, "existing cursor must keep its handle") + _ = runtimeDB +} + +func TestInlineVerifierCurrentSourceDBResetsStmtCacheOnSwap(t *testing.T) { + oldDB := &sql.DB{Marginalia: "old"} + rt := NewSourceRuntime(oldDB, runtimeTestConfig("old")) + v := &InlineVerifier{ + SourceDB: oldDB, + SourceRuntime: rt, + sourceStmtCache: NewStmtCache(), + } + + origCache := v.sourceStmtCache + db, cache := v.currentSourceDB() + assert.Same(t, oldDB, db) + assert.Same(t, origCache, cache, "no swap: cache unchanged") + + // Swap the runtime; the verifier must pick up the new DB and reset its + // statement cache (statements were bound to the old DB). + _, err := rt.Replace(runtimeTestConfig("new"), nil) + require.NoError(t, err) + newHandle := rt.DB() + + db, cache = v.currentSourceDB() + assert.Same(t, newHandle, db) + assert.NotSame(t, origCache, cache, "swap must reset the source statement cache") +} + +func TestInlineVerifierCurrentSourceDBFallsBackWithoutRuntime(t *testing.T) { + db := &sql.DB{Marginalia: "static"} + v := &InlineVerifier{SourceDB: db, sourceStmtCache: NewStmtCache()} + got, _ := v.currentSourceDB() + assert.Same(t, db, got) +} + +func TestIterativeVerifierCurrentSourceDB(t *testing.T) { + staticDB := &sql.DB{Marginalia: "static"} + v := &IterativeVerifier{SourceDB: staticDB} + assert.Same(t, staticDB, v.currentSourceDB()) + + runtimeDB := &sql.DB{Marginalia: "runtime"} + v.SourceRuntime = NewSourceRuntime(runtimeDB, runtimeTestConfig("h")) + assert.Same(t, runtimeDB, v.currentSourceDB()) +} + +func TestChecksumVerifierCurrentSourceDB(t *testing.T) { + staticDB := &sql.DB{Marginalia: "static"} + v := &ChecksumTableVerifier{SourceDB: staticDB} + assert.Same(t, staticDB, v.currentSourceDB()) + + runtimeDB := &sql.DB{Marginalia: "runtime"} + v.SourceRuntime = NewSourceRuntime(runtimeDB, runtimeTestConfig("h")) + assert.Same(t, runtimeDB, v.currentSourceDB()) +} diff --git a/verifier.go b/verifier.go index 2db8d392..a57fea16 100644 --- a/verifier.go +++ b/verifier.go @@ -97,7 +97,11 @@ type ChecksumTableVerifier struct { DatabaseRewrites map[string]string TableRewrites map[string]string SourceDB *sql.DB - TargetDB *sql.DB + // SourceRuntime, when set, provides the current source connection so + // checksum verification follows a source master failover. When nil, + // SourceDB is used. See currentSourceDB. + SourceRuntime *SourceRuntime + TargetDB *sql.DB started *AtomicBoolean @@ -108,6 +112,18 @@ type ChecksumTableVerifier struct { wg *sync.WaitGroup } +// currentSourceDB resolves the source connection from SourceRuntime when +// configured (so verification follows a source master failover), otherwise the +// static SourceDB. +func (v *ChecksumTableVerifier) currentSourceDB() *sql.DB { + if v.SourceRuntime != nil { + if db := v.SourceRuntime.DB(); db != nil { + return db + } + } + return v.SourceDB +} + func (v *ChecksumTableVerifier) VerifyBeforeCutover() error { // All verification occurs in cutover for this verifier. return nil @@ -151,7 +167,7 @@ func (v *ChecksumTableVerifier) VerifyDuringCutover() (VerificationResult, error go func() { defer wg.Done() query := fmt.Sprintf("CHECKSUM TABLE %s EXTENDED", sourceTable) - sourceRow := v.SourceDB.QueryRow(query) + sourceRow := v.currentSourceDB().QueryRow(query) sourceChecksum, sourceErr = v.fetchChecksumValueFromRow(sourceRow) }() From 4dbd3da9dd0e9d1bb284da443782e1877622e1c4 Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Tue, 22 Sep 2026 20:01:50 +0200 Subject: [PATCH 2/3] Resolve source runtime at point of use in data iterator and verifiers The MaxTableSizeSorter still queried DataIterator.DB directly, so table sizing ignored a SourceRuntime swap even though pagination bounds and cursors already followed it. Add DataIterator.currentSourceDB (same shape as the verifier resolvers) and route both the pagination-bound read and the sorter query through it. The sorter resolves on each Sort call, so a swap between the two metadata reads is observed. IterativeVerifier.SanityCheckParameters and ChecksumTableVerifier.StartInBackground validated the static SourceDB rather than the handle they actually use; check currentSourceDB() so a runtime-only source is valid and a missing effective source still fails before any work starts. Error messages are unchanged. Comments in ferry.go and source_runtime.go claiming consumers were not yet migrated are obsolete; describe the current ownership (metadata reads and verification resolve the runtime, cursors snapshot at creation) and state the inline verifier's retained-cache lifetime once instead of twice. Replace the pointer-identity migration tests with consumer-observable regressions backed by a database/sql/driver fixture: sorter ordering and post-swap failure, cursor binding across a real Replace, inline statement cache lifetime through swap and CloseRetired, and verifier validation over every source/target boundary. Each test was shown to fail against the corresponding regression via go test -overlay. --- data_iterator.go | 29 +- data_iterator_sorter.go | 2 +- inline_verifier.go | 14 +- iterative_verifier.go | 2 +- source_runtime_migration_test.go | 500 ++++++++++++++++++++++++++----- verifier.go | 2 +- 6 files changed, 458 insertions(+), 91 deletions(-) diff --git a/data_iterator.go b/data_iterator.go index 7fd6ea53..5bef0c99 100644 --- a/data_iterator.go +++ b/data_iterator.go @@ -9,9 +9,10 @@ import ( type DataIterator struct { DB *sql.DB - // SourceRuntime, when set, provides the current source connection so a data - // iteration run started after a source master failover targets the promoted - // writer. When nil, DB is used. Resolved once at the start of Run. + // SourceRuntime, when set, provides the current source connection for the + // data iterator's own metadata reads (pagination bounds, table sizing) so + // they follow a source master failover. When nil, DB is used. Cursors + // resolve their own handle at creation via CursorConfig. SourceRuntime *SourceRuntime Concurrency int SelectFingerprint bool @@ -32,6 +33,18 @@ type TableMaxPaginationKey struct { MaxPaginationKey PaginationKey } +// currentSourceDB resolves the source connection for the iterator's metadata +// reads from SourceRuntime when configured, otherwise the static DB. It is +// resolved at each point of use so a swap between reads is observed. +func (d *DataIterator) currentSourceDB() *sql.DB { + if d.SourceRuntime != nil { + if db := d.SourceRuntime.DB(); db != nil { + return db + } + } + return d.DB +} + func (d *DataIterator) Run(tables []*TableSchema) { d.logger = LogWithField("tag", "data_iterator") @@ -42,15 +55,7 @@ func (d *DataIterator) Run(tables []*TableSchema) { d.StateTracker = NewStateTracker(0) } - // Resolve the source connection for this run. When a SourceRuntime is - // configured it is authoritative (so a run started after a failover uses the - // promoted writer); otherwise fall back to the static DB. - db := d.DB - if d.SourceRuntime != nil { - if current := d.SourceRuntime.DB(); current != nil { - db = current - } - } + db := d.currentSourceDB() d.logger.WithField("tablesCount", len(tables)).Info("starting data iterator run") tablesWithData, emptyTables, err := MaxPaginationKeys(db, tables, d.logger) diff --git a/data_iterator_sorter.go b/data_iterator_sorter.go index 1a293106..b5f36400 100644 --- a/data_iterator_sorter.go +++ b/data_iterator_sorter.go @@ -60,7 +60,7 @@ func (s *MaxTableSizeSorter) Sort(unorderedTables map[*TableSchema]PaginationKey strings.Join(databaseSchemas, `", "`), strings.Join(tableNames, `", "`), ) - rows, err := s.DataIterator.DB.Query(query) + rows, err := s.DataIterator.currentSourceDB().Query(query) if err != nil { return orderedTables, err diff --git a/inline_verifier.go b/inline_verifier.go index 7bf347a1..c13926d5 100644 --- a/inline_verifier.go +++ b/inline_verifier.go @@ -533,14 +533,12 @@ func (v *InlineVerifier) VerifyDuringCutover() (VerificationResult, error) { // currentSourceDB returns the source connection inline verification should use, // resolving it from SourceRuntime when configured so a failover is followed. If -// the source handle changed since the last call, the source statement cache is -// reset because cached prepared statements are bound to the previous DB. -// currentSourceDB returns the source connection inline verification should use, -// resolving it from SourceRuntime when configured so a failover is followed. If -// the source handle changed since the last call, the source statement cache is -// reset because cached prepared statements are bound to the previous DB. The -// resolved DB and its matching statement cache are returned together under a -// lock so concurrent verification workers stay consistent. +// the source handle changed since the last call, a fresh source statement cache +// is installed because cached prepared statements are bound to the previous DB. +// The resolved DB and its matching statement cache are returned together under +// a lock so concurrent verification workers stay consistent. The previous cache +// is not closed here: in-flight workers may still hold its (db, cache) pair, and +// its statements are released with the retired DB by SourceRuntime.CloseRetired. func (v *InlineVerifier) currentSourceDB() (*sql.DB, *StmtCache) { v.sourceDBMu.Lock() defer v.sourceDBMu.Unlock() diff --git a/iterative_verifier.go b/iterative_verifier.go index 92f26ade..a9da8e25 100644 --- a/iterative_verifier.go +++ b/iterative_verifier.go @@ -157,7 +157,7 @@ func (v *IterativeVerifier) SanityCheckParameters() error { return errors.New("BinlogStreamer must not be nil") } - if v.SourceDB == nil { + if v.currentSourceDB() == nil { return errors.New("SourceDB must not be nil") } diff --git a/source_runtime_migration_test.go b/source_runtime_migration_test.go index 2b447337..6bd16b18 100644 --- a/source_runtime_migration_test.go +++ b/source_runtime_migration_test.go @@ -1,106 +1,470 @@ package ghostferry import ( + "context" + sqlorig "database/sql" + "database/sql/driver" + "errors" + "io" "testing" sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/go-mysql-org/go-mysql/schema" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -// These tests verify that source consumers resolve the current source -// connection from a SourceRuntime when one is configured, and fall back to the -// static DB otherwise. sql.DB values are sentinels; no query is executed. +// These tests verify what source consumers observe when a SourceRuntime is +// configured: metadata reads and verification use the current runtime handle, +// new cursors bind to it at creation, and existing cursors / in-flight inline +// verification keep the handle they started with. No MySQL server is needed: +// fixture DBs answer every query with canned rows, and SourceRuntime.Replace +// opens a lazy (undialed) handle which tests close before querying so that a +// consumer routed to the replacement fails deterministically. -func TestCursorConfigResolvedDB(t *testing.T) { - staticDB := &sql.DB{Marginalia: "static"} +var errUnsupportedSourceRuntimeTestOp = errors.New("unsupported source runtime test operation") - // No runtime: uses static DB. - c := &CursorConfig{DB: staticDB} - assert.Same(t, staticDB, c.resolvedDB()) +// sourceRuntimeQueryConnector backs a *sql.DB that returns the configured rows +// for every query, whether issued directly or through a prepared statement. +type sourceRuntimeQueryConnector struct { + columns []string + data [][]driver.Value +} + +func (c *sourceRuntimeQueryConnector) Connect(ctx context.Context) (driver.Conn, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return &sourceRuntimeQueryConn{connector: c}, nil +} + +func (c *sourceRuntimeQueryConnector) Driver() driver.Driver { return sourceRuntimeQueryDriver{} } + +func (c *sourceRuntimeQueryConnector) rows() *sourceRuntimeQueryRows { + return &sourceRuntimeQueryRows{columns: c.columns, data: c.data} +} + +type sourceRuntimeQueryDriver struct{} + +func (sourceRuntimeQueryDriver) Open(string) (driver.Conn, error) { + return nil, errUnsupportedSourceRuntimeTestOp +} + +type sourceRuntimeQueryConn struct { + connector *sourceRuntimeQueryConnector +} + +func (c *sourceRuntimeQueryConn) Prepare(string) (driver.Stmt, error) { + return &sourceRuntimeQueryStmt{connector: c.connector}, nil +} + +func (c *sourceRuntimeQueryConn) Close() error { return nil } + +func (c *sourceRuntimeQueryConn) Begin() (driver.Tx, error) { + return nil, errUnsupportedSourceRuntimeTestOp +} + +func (c *sourceRuntimeQueryConn) QueryContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Rows, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return c.connector.rows(), nil +} + +type sourceRuntimeQueryStmt struct { + connector *sourceRuntimeQueryConnector +} + +func (s *sourceRuntimeQueryStmt) Close() error { return nil } +func (s *sourceRuntimeQueryStmt) NumInput() int { return -1 } + +func (s *sourceRuntimeQueryStmt) Exec([]driver.Value) (driver.Result, error) { + return nil, errUnsupportedSourceRuntimeTestOp +} + +func (s *sourceRuntimeQueryStmt) Query([]driver.Value) (driver.Rows, error) { + return s.connector.rows(), nil +} + +type sourceRuntimeQueryRows struct { + columns []string + data [][]driver.Value + next int +} + +func (r *sourceRuntimeQueryRows) Columns() []string { return r.columns } +func (r *sourceRuntimeQueryRows) Close() error { return nil } + +func (r *sourceRuntimeQueryRows) Next(dest []driver.Value) error { + if r.next >= len(r.data) { + return io.EOF + } + copy(dest, r.data[r.next]) + r.next++ + return nil +} + +// newSourceRuntimeQueryDB returns an open *sql.DB whose every query yields the +// given columns and rows. It is closed when the test ends. +func newSourceRuntimeQueryDB(t *testing.T, columns []string, data [][]driver.Value) *sql.DB { + t.Helper() + db := &sql.DB{DB: sqlorig.OpenDB(&sourceRuntimeQueryConnector{columns: columns, data: data})} + t.Cleanup(func() { _ = db.Close() }) + return db +} + +// replaceWithClosedHandle swaps rt to a fresh lazily-opened handle and closes +// that handle immediately, so any consumer that resolves the current source +// after the swap fails deterministically without dialing. It requires the +// swap to have actually installed a new handle so a no-op Replace cannot pass. +func replaceWithClosedHandle(t *testing.T, rt *SourceRuntime, host string) *sql.DB { + t.Helper() + previous := rt.DB() + replacement, err := rt.Replace(runtimeTestConfig(host), nil) + require.NoError(t, err) + require.NotNil(t, replacement) + require.NotSame(t, previous, replacement, "Replace must install a new handle") + require.Same(t, replacement, rt.DB()) + t.Cleanup(rt.CloseRetired) + t.Cleanup(func() { _ = replacement.Close() }) + require.NoError(t, replacement.Close()) + return replacement +} + +// numericReviewTable is a review.t table paginated on a numeric id column. +func numericReviewTable() *TableSchema { + columns := []schema.TableColumn{{Name: "id", Type: schema.TYPE_NUMBER}} + return &TableSchema{ + Table: &schema.Table{Schema: "review", Name: "t", Columns: columns}, + PaginationKeyColumn: &columns[0], + PaginationKeyIndex: 0, + } +} + +func newTestCursorConfig(db *sql.DB, rt *SourceRuntime) *CursorConfig { + batchSize := uint64(1) + return &CursorConfig{DB: db, SourceRuntime: rt, BatchSize: &batchSize, ReadRetries: 1} +} + +func newSingleRowDB(t *testing.T) *sql.DB { + t.Helper() + return newSourceRuntimeQueryDB(t, []string{"id"}, [][]driver.Value{{int64(1)}}) +} + +func scanCursor(cursor *Cursor) ([]RowData, error) { + var rows []RowData + err := cursor.Each(func(batch *RowBatch) error { + rows = append(rows, batch.Values()...) + return nil + }) + return rows, err +} + +var singleRow = []RowData{{int64(1)}} + +// --------------------------------------------------------------------------- +// MaxTableSizeSorter - // Runtime with a current DB: uses runtime DB. - runtimeDB := &sql.DB{Marginalia: "runtime"} - c.SourceRuntime = NewSourceRuntime(runtimeDB, runtimeTestConfig("h")) - assert.Same(t, runtimeDB, c.resolvedDB()) +var sorterColumns = []string{"TABLE_NAME", "TABLE_SCHEMA"} - // Runtime present but nil DB: falls back to static DB. - c.SourceRuntime = NewSourceRuntime(nil, nil) - assert.Same(t, staticDB, c.resolvedDB()) +func sorterInput() (map[*TableSchema]PaginationKey, *TableSchema, *TableSchema) { + small := &TableSchema{Table: &schema.Table{Schema: "review", Name: "small"}} + large := &TableSchema{Table: &schema.Table{Schema: "review", Name: "large"}} + return map[*TableSchema]PaginationKey{ + small: NewUint64Key(10), + large: NewUint64Key(20), + }, small, large } +func assertSortedTables(t *testing.T, sorted []TableMaxPaginationKey, expected ...*TableSchema) { + t.Helper() + require.Len(t, sorted, len(expected)) + for i, table := range expected { + assert.Same(t, table, sorted[i].Table, "position %d", i) + } + for _, entry := range sorted { + switch entry.Table.Name { + case "small": + assert.Equal(t, NewUint64Key(10), entry.MaxPaginationKey) + case "large": + assert.Equal(t, NewUint64Key(20), entry.MaxPaginationKey) + default: + t.Fatalf("unexpected table %s", entry.Table.Name) + } + } +} + +func TestMaxTableSizeSorterUsesSourceRuntime(t *testing.T) { + input, small, large := sorterInput() + + staticDB := newSourceRuntimeQueryDB(t, sorterColumns, [][]driver.Value{ + {"small", "review"}, + {"large", "review"}, + }) + runtimeDB := newSourceRuntimeQueryDB(t, sorterColumns, [][]driver.Value{ + {"large", "review"}, + {"small", "review"}, + }) + rt := NewSourceRuntime(runtimeDB, runtimeTestConfig("runtime")) + + iterator := &DataIterator{DB: staticDB, SourceRuntime: rt} + sorter := MaxTableSizeSorter{DataIterator: iterator} + + // The runtime's ordering wins over the static DB's. + sorted, err := sorter.Sort(input) + require.NoError(t, err) + assertSortedTables(t, sorted, large, small) + + // The sorter resolves the source on every call: once the runtime points at + // an unusable handle, sorting fails rather than falling back to the static + // DB, which is still open. + replaceWithClosedHandle(t, rt, "replacement") + _, err = sorter.Sort(input) + require.Error(t, err) +} + +func TestMaxTableSizeSorterStaticFallback(t *testing.T) { + input, small, large := sorterInput() + + cases := map[string]*SourceRuntime{ + "no runtime": nil, + "runtime with nil DB": NewSourceRuntime(nil, nil), + } + for name, rt := range cases { + t.Run(name, func(t *testing.T) { + staticDB := newSourceRuntimeQueryDB(t, sorterColumns, [][]driver.Value{ + {"large", "review"}, + {"small", "review"}, + }) + sorter := MaxTableSizeSorter{DataIterator: &DataIterator{DB: staticDB, SourceRuntime: rt}} + + sorted, err := sorter.Sort(input) + require.NoError(t, err) + assertSortedTables(t, sorted, large, small) + }) + } +} + +// --------------------------------------------------------------------------- +// Cursor + func TestNewCursorBindsToResolvedDB(t *testing.T) { - staticDB := &sql.DB{Marginalia: "static"} - runtimeDB := &sql.DB{Marginalia: "runtime"} - rt := NewSourceRuntime(staticDB, runtimeTestConfig("old")) + table := numericReviewTable() + initialDB := newSingleRowDB(t) + rt := NewSourceRuntime(initialDB, runtimeTestConfig("old")) + cfg := newTestCursorConfig(initialDB, rt) - cfg := &CursorConfig{DB: staticDB, SourceRuntime: rt} - table := &TableSchema{} + // A cursor created before the swap keeps scanning the server it started on. + before := cfg.NewCursorWithoutRowLock(table, NewUint64Key(0), NewUint64Key(1)) - // Cursor created now binds to the current (static) handle. - cursor := cfg.NewCursor(table, NewUint64Key(0), NewUint64Key(10)) - assert.Same(t, staticDB, cursor.DB) + replaceWithClosedHandle(t, rt, "new") - // After a runtime swap, a newly created cursor binds to the new handle, - // while the already-created cursor keeps its handle (scan consistency). - _, err := rt.Replace(runtimeTestConfig("new"), nil) + rows, err := scanCursor(before) require.NoError(t, err) - newHandle := rt.DB() + assert.Equal(t, singleRow, rows, "existing cursor must keep its handle") - cursor2 := cfg.NewCursor(table, NewUint64Key(0), NewUint64Key(10)) - assert.Same(t, newHandle, cursor2.DB) - assert.Same(t, staticDB, cursor.DB, "existing cursor must keep its handle") - _ = runtimeDB + // A cursor created after the swap uses the current (closed) runtime handle + // rather than the still-open static DB. + after := cfg.NewCursorWithoutRowLock(table, NewUint64Key(0), NewUint64Key(1)) + rows, err = scanCursor(after) + require.Error(t, err) + assert.Empty(t, rows) } -func TestInlineVerifierCurrentSourceDBResetsStmtCacheOnSwap(t *testing.T) { - oldDB := &sql.DB{Marginalia: "old"} - rt := NewSourceRuntime(oldDB, runtimeTestConfig("old")) - v := &InlineVerifier{ - SourceDB: oldDB, - SourceRuntime: rt, - sourceStmtCache: NewStmtCache(), +func TestCursorRuntimeFallback(t *testing.T) { + table := numericReviewTable() + + cases := map[string]*SourceRuntime{ + "no runtime": nil, + "runtime with nil DB": NewSourceRuntime(nil, nil), + } + for name, rt := range cases { + t.Run(name, func(t *testing.T) { + cfg := newTestCursorConfig(newSingleRowDB(t), rt) + cursor := cfg.NewCursorWithoutRowLock(table, NewUint64Key(0), NewUint64Key(1)) + + rows, err := scanCursor(cursor) + require.NoError(t, err) + assert.Equal(t, singleRow, rows) + }) } +} + +// --------------------------------------------------------------------------- +// InlineVerifier + +var ( + fingerprintColumns = []string{"id", "__ghostferry_row_md5"} + oldFingerprint = []byte("old-fingerprint") +) + +func newFingerprintDB(t *testing.T) *sql.DB { + t.Helper() + return newSourceRuntimeQueryDB(t, fingerprintColumns, [][]driver.Value{{int64(1), oldFingerprint}}) +} + +func TestInlineVerifierSourceRuntimeCacheLifetime(t *testing.T) { + table := numericReviewTable() + key := NewUint64Key(1).String() + paginationKeys := []interface{}{uint64(1)} + + oldDB := newFingerprintDB(t) + rt := NewSourceRuntime(oldDB, runtimeTestConfig("old")) + v := &InlineVerifier{SourceDB: oldDB, SourceRuntime: rt, sourceStmtCache: NewStmtCache()} + + // Prime the source statement cache against the old DB. + fingerprints, _, err := v.getFingerprintDataFromSourceDb("review", "t", nil, table, paginationKeys) + require.NoError(t, err) + assert.Equal(t, oldFingerprint, fingerprints[key]) + + // An in-flight worker holds the (db, cache) pair it resolved before the swap. + inFlightDB, inFlightCache := v.currentSourceDB() + + replaceWithClosedHandle(t, rt, "new") - origCache := v.sourceStmtCache - db, cache := v.currentSourceDB() - assert.Same(t, oldDB, db) - assert.Same(t, origCache, cache, "no swap: cache unchanged") + // New work follows the runtime to the (closed) replacement. Reusing the + // query-keyed cache primed against the old DB would incorrectly succeed. + _, _, err = v.getFingerprintDataFromSourceDb("review", "t", nil, table, paginationKeys) + require.Error(t, err) - // Swap the runtime; the verifier must pick up the new DB and reset its - // statement cache (statements were bound to the old DB). - _, err := rt.Replace(runtimeTestConfig("new"), nil) + // The in-flight pair remains usable until retired handles are released. + fingerprints, _, err = v.getFingerprintDataFromDb(inFlightDB, inFlightCache, "review", "t", nil, table, paginationKeys) require.NoError(t, err) - newHandle := rt.DB() + assert.Equal(t, oldFingerprint, fingerprints[key]) - db, cache = v.currentSourceDB() - assert.Same(t, newHandle, db) - assert.NotSame(t, origCache, cache, "swap must reset the source statement cache") + rt.CloseRetired() + _, _, err = v.getFingerprintDataFromDb(inFlightDB, inFlightCache, "review", "t", nil, table, paginationKeys) + require.Error(t, err) } -func TestInlineVerifierCurrentSourceDBFallsBackWithoutRuntime(t *testing.T) { - db := &sql.DB{Marginalia: "static"} - v := &InlineVerifier{SourceDB: db, sourceStmtCache: NewStmtCache()} - got, _ := v.currentSourceDB() - assert.Same(t, db, got) +func TestInlineVerifierSourceFallback(t *testing.T) { + table := numericReviewTable() + key := NewUint64Key(1).String() + + cases := map[string]*SourceRuntime{ + "no runtime": nil, + "runtime with nil DB": NewSourceRuntime(nil, nil), + } + for name, rt := range cases { + t.Run(name, func(t *testing.T) { + v := &InlineVerifier{SourceDB: newFingerprintDB(t), SourceRuntime: rt, sourceStmtCache: NewStmtCache()} + + fingerprints, _, err := v.getFingerprintDataFromSourceDb("review", "t", nil, table, []interface{}{uint64(1)}) + require.NoError(t, err) + assert.Equal(t, oldFingerprint, fingerprints[key]) + }) + } +} + +// --------------------------------------------------------------------------- +// Verifier validation + +// sourceSelection enumerates the source configurations a verifier can be given. +type sourceSelection struct { + name string + static bool // provide a working static SourceDB + runtime bool // provide a SourceRuntime + runtimeDB bool // ... whose handle is a working DB (false: nil handle) + target bool // provide a working TargetDB + ok bool } -func TestIterativeVerifierCurrentSourceDB(t *testing.T) { - staticDB := &sql.DB{Marginalia: "static"} - v := &IterativeVerifier{SourceDB: staticDB} - assert.Same(t, staticDB, v.currentSourceDB()) +var sourceSelections = []sourceSelection{ + {name: "runtime only", runtime: true, runtimeDB: true, target: true, ok: true}, + {name: "static only", static: true, target: true, ok: true}, + {name: "nil-handle runtime with static fallback", static: true, runtime: true, target: true, ok: true}, + {name: "no source", target: true}, + {name: "nil-handle runtime without fallback", runtime: true, target: true}, + {name: "no target", static: true, runtime: true, runtimeDB: true}, +} - runtimeDB := &sql.DB{Marginalia: "runtime"} - v.SourceRuntime = NewSourceRuntime(runtimeDB, runtimeTestConfig("h")) - assert.Same(t, runtimeDB, v.currentSourceDB()) +func (s sourceSelection) build(t *testing.T, newDB func(*testing.T) *sql.DB) (static *sql.DB, rt *SourceRuntime, target *sql.DB) { + t.Helper() + if s.static { + static = newDB(t) + } + if s.runtime { + var db *sql.DB + if s.runtimeDB { + db = newDB(t) + } + rt = NewSourceRuntime(db, runtimeTestConfig("runtime")) + } + if s.target { + target = newDB(t) + } + return } -func TestChecksumVerifierCurrentSourceDB(t *testing.T) { - staticDB := &sql.DB{Marginalia: "static"} - v := &ChecksumTableVerifier{SourceDB: staticDB} - assert.Same(t, staticDB, v.currentSourceDB()) +func TestIterativeVerifierSourceRuntimeValidation(t *testing.T) { + for _, tc := range sourceSelections { + t.Run(tc.name, func(t *testing.T) { + static, rt, target := tc.build(t, newSingleRowDB) + v := &IterativeVerifier{ + SourceDB: static, + SourceRuntime: rt, + TargetDB: target, + CursorConfig: newTestCursorConfig(static, rt), + BinlogStreamer: &BinlogStreamer{}, + Concurrency: 1, + } + + err := v.Initialize() + if tc.ok { + require.NoError(t, err) + } else { + require.Error(t, err) + } + }) + } +} + +var checksumColumns = []string{"Table", "Checksum"} + +func newChecksumDB(checksum int64) func(*testing.T) *sql.DB { + return func(t *testing.T) *sql.DB { + t.Helper() + return newSourceRuntimeQueryDB(t, checksumColumns, [][]driver.Value{{"review.t", checksum}}) + } +} + +func runChecksumVerifier(t *testing.T, static *sql.DB, rt *SourceRuntime, target *sql.DB) error { + t.Helper() + v := &ChecksumTableVerifier{ + Tables: []*TableSchema{numericReviewTable()}, + SourceDB: static, + SourceRuntime: rt, + TargetDB: target, + } + + if err := v.StartInBackground(); err != nil { + return err + } + v.Wait() + + result, err := v.Result() + require.NoError(t, err) + assert.True(t, result.DataCorrect) + return nil +} + +func TestChecksumVerifierSourceRuntimeValidation(t *testing.T) { + for _, tc := range sourceSelections { + t.Run(tc.name, func(t *testing.T) { + static, rt, target := tc.build(t, newChecksumDB(11)) + err := runChecksumVerifier(t, static, rt, target) + if tc.ok { + require.NoError(t, err) + } else { + require.Error(t, err) + } + }) + } - runtimeDB := &sql.DB{Marginalia: "runtime"} - v.SourceRuntime = NewSourceRuntime(runtimeDB, runtimeTestConfig("h")) - assert.Same(t, runtimeDB, v.currentSourceDB()) + t.Run("runtime over static", func(t *testing.T) { + static := newChecksumDB(22)(t) + rt := NewSourceRuntime(newChecksumDB(11)(t), runtimeTestConfig("runtime")) + target := newChecksumDB(11)(t) + require.NoError(t, runChecksumVerifier(t, static, rt, target)) + }) } diff --git a/verifier.go b/verifier.go index a57fea16..13939bf1 100644 --- a/verifier.go +++ b/verifier.go @@ -226,7 +226,7 @@ func (v *ChecksumTableVerifier) fetchChecksumValueFromRow(row *sqlorig.Row) (int } func (v *ChecksumTableVerifier) StartInBackground() error { - if v.SourceDB == nil || v.TargetDB == nil { + if v.currentSourceDB() == nil || v.TargetDB == nil { return errors.New("must specify source and target db") } From 67fc962d8b3e0f01134a7091889b5e067b72c84b Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Tue, 13 Oct 2026 14:14:08 +0200 Subject: [PATCH 3/3] Trim verbose comments --- cursor.go | 9 --------- data_iterator.go | 10 ++-------- inline_verifier.go | 17 +++------------- iterative_verifier.go | 11 ++--------- source_runtime_migration_test.go | 34 ++++++-------------------------- verifier.go | 10 ++-------- 6 files changed, 15 insertions(+), 76 deletions(-) diff --git a/cursor.go b/cursor.go index f533eec4..7f3eef07 100644 --- a/cursor.go +++ b/cursor.go @@ -36,13 +36,6 @@ type CursorConfig struct { DB *sql.DB Throttler Throttler - // SourceRuntime, when set, is the authoritative source of the current - // source connection. NewCursor resolves the cursor's DB from it at cursor - // creation time so that cursors started after a source master failover run - // against the promoted writer. When nil, the static DB field is used. A - // cursor captures the resolved handle for the duration of its scan (a scan - // must be consistent against a single server); only newly created cursors - // observe a swap. SourceRuntime *SourceRuntime ColumnsToSelect []string @@ -54,8 +47,6 @@ type CursorConfig struct { ReadRetries int } -// resolvedDB returns the source connection to use for a new cursor: the current -// handle from SourceRuntime when configured, otherwise the static DB. func (c *CursorConfig) resolvedDB() *sql.DB { if c.SourceRuntime != nil { if db := c.SourceRuntime.DB(); db != nil { diff --git a/data_iterator.go b/data_iterator.go index 5bef0c99..80e8fa1c 100644 --- a/data_iterator.go +++ b/data_iterator.go @@ -8,11 +8,7 @@ import ( ) type DataIterator struct { - DB *sql.DB - // SourceRuntime, when set, provides the current source connection for the - // data iterator's own metadata reads (pagination bounds, table sizing) so - // they follow a source master failover. When nil, DB is used. Cursors - // resolve their own handle at creation via CursorConfig. + DB *sql.DB SourceRuntime *SourceRuntime Concurrency int SelectFingerprint bool @@ -33,9 +29,7 @@ type TableMaxPaginationKey struct { MaxPaginationKey PaginationKey } -// currentSourceDB resolves the source connection for the iterator's metadata -// reads from SourceRuntime when configured, otherwise the static DB. It is -// resolved at each point of use so a swap between reads is observed. +// currentSourceDB resolves the source per call, not at startup, so a runtime swap is observed between reads. func (d *DataIterator) currentSourceDB() *sql.DB { if d.SourceRuntime != nil { if db := d.SourceRuntime.DB(); db != nil { diff --git a/inline_verifier.go b/inline_verifier.go index c13926d5..4251a222 100644 --- a/inline_verifier.go +++ b/inline_verifier.go @@ -257,10 +257,7 @@ type InlineVerifierMismatches struct { } type InlineVerifier struct { - SourceDB *sql.DB - // SourceRuntime, when set, is the authoritative source of the current - // source connection (see currentSourceDB). It lets inline verification - // follow a source master failover. When nil, SourceDB is used directly. + SourceDB *sql.DB SourceRuntime *SourceRuntime TargetDB *sql.DB DatabaseRewrites map[string]string @@ -278,8 +275,7 @@ type InlineVerifier struct { sourceStmtCache *StmtCache targetStmtCache *StmtCache - // sourceDBMu guards SourceDB and sourceStmtCache when currentSourceDB - // resolves the runtime handle and may reset the cache on a failover swap. + // sourceDBMu guards SourceDB and sourceStmtCache. sourceDBMu sync.Mutex logger Logger @@ -531,14 +527,7 @@ func (v *InlineVerifier) VerifyDuringCutover() (VerificationResult, error) { }, nil } -// currentSourceDB returns the source connection inline verification should use, -// resolving it from SourceRuntime when configured so a failover is followed. If -// the source handle changed since the last call, a fresh source statement cache -// is installed because cached prepared statements are bound to the previous DB. -// The resolved DB and its matching statement cache are returned together under -// a lock so concurrent verification workers stay consistent. The previous cache -// is not closed here: in-flight workers may still hold its (db, cache) pair, and -// its statements are released with the retired DB by SourceRuntime.CloseRetired. +// currentSourceDB returns the source DB and its matching statement cache; on a handle swap the cache is replaced (statements are bound to the old DB) but not closed, as in-flight workers may still hold it. func (v *InlineVerifier) currentSourceDB() (*sql.DB, *StmtCache) { v.sourceDBMu.Lock() defer v.sourceDBMu.Unlock() diff --git a/iterative_verifier.go b/iterative_verifier.go index a9da8e25..b5dd5e45 100644 --- a/iterative_verifier.go +++ b/iterative_verifier.go @@ -120,11 +120,8 @@ type IterativeVerifier struct { BinlogStreamer *BinlogStreamer TableSchemaCache TableSchemaCache SourceDB *sql.DB - // SourceRuntime, when set, provides the current source connection so - // iterative verification follows a source master failover. When nil, - // SourceDB is used. See currentSourceDB. - SourceRuntime *SourceRuntime - TargetDB *sql.DB + SourceRuntime *SourceRuntime + TargetDB *sql.DB Tables []*TableSchema IgnoredTables []string @@ -293,10 +290,6 @@ func (v *IterativeVerifier) Result() (VerificationResultAndStatus, error) { return v.verificationResultAndStatus, v.verificationErr } -// currentSourceDB resolves the source connection from SourceRuntime when -// configured (so verification follows a source master failover), otherwise the -// static SourceDB. Unlike the inline verifier there is no persistent -// source-bound statement cache to reset: GetHashes prepares per call. func (v *IterativeVerifier) currentSourceDB() *sql.DB { if v.SourceRuntime != nil { if db := v.SourceRuntime.DB(); db != nil { diff --git a/source_runtime_migration_test.go b/source_runtime_migration_test.go index 6bd16b18..30a75c66 100644 --- a/source_runtime_migration_test.go +++ b/source_runtime_migration_test.go @@ -14,18 +14,12 @@ import ( "github.com/stretchr/testify/require" ) -// These tests verify what source consumers observe when a SourceRuntime is -// configured: metadata reads and verification use the current runtime handle, -// new cursors bind to it at creation, and existing cursors / in-flight inline -// verification keep the handle they started with. No MySQL server is needed: -// fixture DBs answer every query with canned rows, and SourceRuntime.Replace -// opens a lazy (undialed) handle which tests close before querying so that a -// consumer routed to the replacement fails deterministically. +// These tests need no MySQL: fixture DBs answer every query with canned rows, +// and a swapped-in handle is closed up front so consumers routed to it fail fast. var errUnsupportedSourceRuntimeTestOp = errors.New("unsupported source runtime test operation") -// sourceRuntimeQueryConnector backs a *sql.DB that returns the configured rows -// for every query, whether issued directly or through a prepared statement. +// sourceRuntimeQueryConnector returns the configured rows for every query, prepared or not. type sourceRuntimeQueryConnector struct { columns []string data [][]driver.Value @@ -104,8 +98,6 @@ func (r *sourceRuntimeQueryRows) Next(dest []driver.Value) error { return nil } -// newSourceRuntimeQueryDB returns an open *sql.DB whose every query yields the -// given columns and rows. It is closed when the test ends. func newSourceRuntimeQueryDB(t *testing.T, columns []string, data [][]driver.Value) *sql.DB { t.Helper() db := &sql.DB{DB: sqlorig.OpenDB(&sourceRuntimeQueryConnector{columns: columns, data: data})} @@ -113,10 +105,7 @@ func newSourceRuntimeQueryDB(t *testing.T, columns []string, data [][]driver.Val return db } -// replaceWithClosedHandle swaps rt to a fresh lazily-opened handle and closes -// that handle immediately, so any consumer that resolves the current source -// after the swap fails deterministically without dialing. It requires the -// swap to have actually installed a new handle so a no-op Replace cannot pass. +// replaceWithClosedHandle swaps in a fresh handle and closes it, so consumers resolving the source after the swap fail deterministically. func replaceWithClosedHandle(t *testing.T, rt *SourceRuntime, host string) *sql.DB { t.Helper() previous := rt.DB() @@ -131,7 +120,6 @@ func replaceWithClosedHandle(t *testing.T, rt *SourceRuntime, host string) *sql. return replacement } -// numericReviewTable is a review.t table paginated on a numeric id column. func numericReviewTable() *TableSchema { columns := []schema.TableColumn{{Name: "id", Type: schema.TYPE_NUMBER}} return &TableSchema{ @@ -210,14 +198,11 @@ func TestMaxTableSizeSorterUsesSourceRuntime(t *testing.T) { iterator := &DataIterator{DB: staticDB, SourceRuntime: rt} sorter := MaxTableSizeSorter{DataIterator: iterator} - // The runtime's ordering wins over the static DB's. sorted, err := sorter.Sort(input) require.NoError(t, err) assertSortedTables(t, sorted, large, small) - // The sorter resolves the source on every call: once the runtime points at - // an unusable handle, sorting fails rather than falling back to the static - // DB, which is still open. + // The static DB is still open, so this failing proves the sorter resolves the source per call. replaceWithClosedHandle(t, rt, "replacement") _, err = sorter.Sort(input) require.Error(t, err) @@ -254,7 +239,6 @@ func TestNewCursorBindsToResolvedDB(t *testing.T) { rt := NewSourceRuntime(initialDB, runtimeTestConfig("old")) cfg := newTestCursorConfig(initialDB, rt) - // A cursor created before the swap keeps scanning the server it started on. before := cfg.NewCursorWithoutRowLock(table, NewUint64Key(0), NewUint64Key(1)) replaceWithClosedHandle(t, rt, "new") @@ -263,8 +247,6 @@ func TestNewCursorBindsToResolvedDB(t *testing.T) { require.NoError(t, err) assert.Equal(t, singleRow, rows, "existing cursor must keep its handle") - // A cursor created after the swap uses the current (closed) runtime handle - // rather than the still-open static DB. after := cfg.NewCursorWithoutRowLock(table, NewUint64Key(0), NewUint64Key(1)) rows, err = scanCursor(after) require.Error(t, err) @@ -317,17 +299,14 @@ func TestInlineVerifierSourceRuntimeCacheLifetime(t *testing.T) { require.NoError(t, err) assert.Equal(t, oldFingerprint, fingerprints[key]) - // An in-flight worker holds the (db, cache) pair it resolved before the swap. inFlightDB, inFlightCache := v.currentSourceDB() replaceWithClosedHandle(t, rt, "new") - // New work follows the runtime to the (closed) replacement. Reusing the - // query-keyed cache primed against the old DB would incorrectly succeed. + // A stale cache would reuse the old DB's statement and incorrectly succeed. _, _, err = v.getFingerprintDataFromSourceDb("review", "t", nil, table, paginationKeys) require.Error(t, err) - // The in-flight pair remains usable until retired handles are released. fingerprints, _, err = v.getFingerprintDataFromDb(inFlightDB, inFlightCache, "review", "t", nil, table, paginationKeys) require.NoError(t, err) assert.Equal(t, oldFingerprint, fingerprints[key]) @@ -359,7 +338,6 @@ func TestInlineVerifierSourceFallback(t *testing.T) { // --------------------------------------------------------------------------- // Verifier validation -// sourceSelection enumerates the source configurations a verifier can be given. type sourceSelection struct { name string static bool // provide a working static SourceDB diff --git a/verifier.go b/verifier.go index 13939bf1..4047e65c 100644 --- a/verifier.go +++ b/verifier.go @@ -97,11 +97,8 @@ type ChecksumTableVerifier struct { DatabaseRewrites map[string]string TableRewrites map[string]string SourceDB *sql.DB - // SourceRuntime, when set, provides the current source connection so - // checksum verification follows a source master failover. When nil, - // SourceDB is used. See currentSourceDB. - SourceRuntime *SourceRuntime - TargetDB *sql.DB + SourceRuntime *SourceRuntime + TargetDB *sql.DB started *AtomicBoolean @@ -112,9 +109,6 @@ type ChecksumTableVerifier struct { wg *sync.WaitGroup } -// currentSourceDB resolves the source connection from SourceRuntime when -// configured (so verification follows a source master failover), otherwise the -// static SourceDB. func (v *ChecksumTableVerifier) currentSourceDB() *sql.DB { if v.SourceRuntime != nil { if db := v.SourceRuntime.DB(); db != nil {