From 3be312a58e33d7b3047d61cab321ac3c600a8f5f Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Thu, 23 Jul 2026 10:28:43 -0400 Subject: [PATCH 1/3] Introduce SourceRuntime to own the source connection Ghostferry's source connection and its DatabaseConfig are currently copied by value into every consumer (data iterator, cursors, verifiers, ad-hoc SHOW CREATE queries). That is fine while the source never changes, but a source master failover needs to repoint every consumer at a newly promoted writer atomically, which today would mean hand-editing a field on each one. Introduce SourceRuntime as the future single owner of "the current source": it holds the source *sql.DB and *DatabaseConfig behind concurrency-safe accessors and offers Replace to swap in a new source (opening the new connection and retaining the old handle rather than closing it, since in-flight cursors and cached statements may still reference it) plus CloseRetired to release retired handles at teardown. This change only introduces the type and has the Ferry construct/hold one alongside SourceDB in Initialize; no consumer is migrated yet, so behavior is unchanged. Follow-up changes migrate consumers to the runtime and then use Replace to implement master failover. --- ferry.go | 18 +++++++ source_runtime.go | 104 +++++++++++++++++++++++++++++++++++++++++ source_runtime_test.go | 80 +++++++++++++++++++++++++++++++ 3 files changed, 202 insertions(+) create mode 100644 source_runtime.go create mode 100644 source_runtime_test.go diff --git a/ferry.go b/ferry.go index 1074bcc2..fe34bbc9 100644 --- a/ferry.go +++ b/ferry.go @@ -54,6 +54,13 @@ type Ferry struct { SourceDB *sql.DB TargetDB *sql.DB + // sourceRuntime owns the current source connection and its DatabaseConfig. + // It is introduced as the future single source of truth for "the current + // source" so a master failover can repoint every consumer atomically. It is + // initialised alongside SourceDB in Initialize. Consumers are not yet + // migrated to it; SourceDB / Config.Source remain authoritative for now. + sourceRuntime *SourceRuntime + ControlServer *ControlServer BinlogStreamer *BinlogStreamer @@ -91,6 +98,12 @@ type Ferry struct { rowCopyCompleteCh chan struct{} } +// SourceRuntime returns the Ferry's source runtime, which owns the current +// source connection and config. It is populated during Initialize. +func (f *Ferry) SourceRuntime() *SourceRuntime { + return f.sourceRuntime +} + func (f *Ferry) NewDataIterator() *DataIterator { f.ensureInitialized() @@ -367,6 +380,11 @@ func (f *Ferry) Initialize() (err error) { return err } + // Wrap the source connection in a SourceRuntime. This is currently only the + // owner of the initial handle; consumers still read SourceDB directly. A + // later change migrates consumers to the runtime and uses it for failover. + f.sourceRuntime = NewSourceRuntime(f.SourceDB, f.Config.Source) + err = f.checkConnection("source", f.SourceDB) if err != nil { f.logger.WithError(err).Error("source connection checking failed") diff --git a/source_runtime.go b/source_runtime.go new file mode 100644 index 00000000..ceeff36c --- /dev/null +++ b/source_runtime.go @@ -0,0 +1,104 @@ +package ghostferry + +import ( + "sync" + + sql "github.com/Shopify/ghostferry/sqlwrapper" +) + +// SourceRuntime owns the live connection to the source MySQL server and the +// DatabaseConfig it was opened from. +// +// It exists so that the rest of Ghostferry can talk about "the current source" +// as a single, swappable thing rather than each component holding its own copy +// of a *sql.DB and a *DatabaseConfig. Today the source never changes during a +// run, but a source master failover needs to atomically repoint every source +// consumer at a newly promoted writer. Centralising ownership here is the +// prerequisite for doing that without hand-editing a field on every consumer. +// +// A SourceRuntime is safe for concurrent use. Readers obtain the current handle +// via DB()/Config(); a swap (Replace) is serialised against those readers. +// +// This type is intentionally introduced without migrating any consumers yet: +// the Ferry constructs and holds one, but components continue to read +// Ferry.SourceDB / Config.Source directly. A follow-up change migrates the +// consumers to depend on the runtime, and a further change uses Replace to +// implement failover. +type SourceRuntime struct { + mu sync.RWMutex + db *sql.DB + config *DatabaseConfig + + // retired holds source DB handles replaced by a swap. They are not closed + // at swap time because in-flight cursors and cached prepared statements may + // still reference them; CloseRetired releases them once no source work is + // running (e.g. at teardown). + retired []*sql.DB +} + +// NewSourceRuntime creates a SourceRuntime around an already-open source +// connection and the config it was opened from. Both may be nil for tests that +// only exercise Replace. +func NewSourceRuntime(db *sql.DB, config *DatabaseConfig) *SourceRuntime { + return &SourceRuntime{ + db: db, + config: config, + } +} + +// DB returns the current source connection. +func (r *SourceRuntime) DB() *sql.DB { + r.mu.RLock() + defer r.mu.RUnlock() + return r.db +} + +// Config returns the current source DatabaseConfig. +func (r *SourceRuntime) Config() *DatabaseConfig { + r.mu.RLock() + defer r.mu.RUnlock() + return r.config +} + +// Replace opens a new source connection from the given config and installs it +// as the current source, returning the newly opened *sql.DB. +// +// The previously current handle is retired (retained, not closed) so callers +// that still hold a reference to it — an in-flight cursor, a cached prepared +// statement — do not see it closed out from under them. Use CloseRetired at a +// safe point to release retired handles. +// +// 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) { + newDB, err := config.SqlDB(logger) + if err != nil { + return nil, err + } + + r.mu.Lock() + defer r.mu.Unlock() + + if r.db != nil { + r.retired = append(r.retired, r.db) + } + r.db = newDB + r.config = config + return newDB, nil +} + +// CloseRetired closes every retired source handle and clears the retired list. +// It is best-effort and intended to be called during teardown, once no source +// cursors or verification passes remain in flight. +func (r *SourceRuntime) CloseRetired() { + r.mu.Lock() + retired := r.retired + r.retired = nil + r.mu.Unlock() + + for _, db := range retired { + if db != nil { + _ = db.Close() + } + } +} diff --git a/source_runtime_test.go b/source_runtime_test.go new file mode 100644 index 00000000..cf421ca4 --- /dev/null +++ b/source_runtime_test.go @@ -0,0 +1,80 @@ +package ghostferry + +import ( + "testing" + + sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// runtimeTestConfig is a DatabaseConfig SqlDB can open lazily (sql.Open does +// not dial), so Replace can install a real handle without a live server. +func runtimeTestConfig(host string) *DatabaseConfig { + return &DatabaseConfig{Host: host, Port: 3306, User: "root", Net: "tcp"} +} + +func TestSourceRuntimeAccessors(t *testing.T) { + db := &sql.DB{Marginalia: "initial"} + cfg := runtimeTestConfig("initial-host") + + rt := NewSourceRuntime(db, cfg) + assert.Same(t, db, rt.DB()) + assert.Same(t, cfg, rt.Config()) +} + +func TestSourceRuntimeReplaceInstallsNewSourceAndRetiresOld(t *testing.T) { + oldDB := &sql.DB{Marginalia: "old"} + oldCfg := runtimeTestConfig("old-host") + rt := NewSourceRuntime(oldDB, oldCfg) + + newCfg := runtimeTestConfig("new-host") + newDB, err := rt.Replace(newCfg, nil) + require.NoError(t, err) + require.NotNil(t, newDB) + + // The runtime now serves the new handle/config. + assert.Same(t, newDB, rt.DB()) + assert.Same(t, newCfg, rt.Config()) + assert.NotSame(t, oldDB, rt.DB()) + + // The old handle is retired, not closed here. + assert.Contains(t, rt.retired, oldDB) +} + +func TestSourceRuntimeReplaceRetainsAllPriorHandles(t *testing.T) { + rt := NewSourceRuntime(&sql.DB{Marginalia: "gen0"}, runtimeTestConfig("h0")) + + _, err := rt.Replace(runtimeTestConfig("h1"), nil) + require.NoError(t, err) + _, err = rt.Replace(runtimeTestConfig("h2"), nil) + require.NoError(t, err) + + // Two swaps retire two handles (gen0 and the h1 handle). + assert.Len(t, rt.retired, 2) +} + +func TestSourceRuntimeCloseRetiredDrains(t *testing.T) { + // Use real (lazily-opened, undialed) handles so Close has an underlying DB. + dbA, err := runtimeTestConfig("a").SqlDB(nil) + require.NoError(t, err) + dbB, err := runtimeTestConfig("b").SqlDB(nil) + require.NoError(t, err) + + rt := NewSourceRuntime(nil, nil) + rt.retired = []*sql.DB{dbA, dbB} + + assert.NotPanics(t, func() { rt.CloseRetired() }) + assert.Empty(t, rt.retired, "retired handles must be drained") +} + +func TestSourceRuntimeNilInitial(t *testing.T) { + rt := NewSourceRuntime(nil, nil) + assert.Nil(t, rt.DB()) + assert.Nil(t, rt.Config()) + + // Replacing from a nil initial does not retire anything. + _, err := rt.Replace(runtimeTestConfig("h1"), nil) + require.NoError(t, err) + assert.Empty(t, rt.retired) +} From 4e778cea4cea202d093358a7e1f224b715b1b965 Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Tue, 22 Sep 2026 18:38:57 +0200 Subject: [PATCH 2/3] Improve test suite --- source_runtime_test.go | 40 ++++++++++++++++++++++++++++++++-------- 1 file changed, 32 insertions(+), 8 deletions(-) diff --git a/source_runtime_test.go b/source_runtime_test.go index cf421ca4..7191f205 100644 --- a/source_runtime_test.go +++ b/source_runtime_test.go @@ -1,6 +1,7 @@ package ghostferry import ( + "context" "testing" sql "github.com/Shopify/ghostferry/sqlwrapper" @@ -14,6 +15,17 @@ func runtimeTestConfig(host string) *DatabaseConfig { return &DatabaseConfig{Host: host, Port: 3306, User: "root", Net: "tcp"} } +// isClosed reports whether db rejects work because it was closed. A cancelled +// context makes an open lazily-dialed handle fail fast without dialing. +func isClosed(t *testing.T, db *sql.DB) bool { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err := db.PingContext(ctx) + require.Error(t, err) + return err != context.Canceled +} + func TestSourceRuntimeAccessors(t *testing.T) { db := &sql.DB{Marginalia: "initial"} cfg := runtimeTestConfig("initial-host") @@ -54,18 +66,30 @@ func TestSourceRuntimeReplaceRetainsAllPriorHandles(t *testing.T) { assert.Len(t, rt.retired, 2) } -func TestSourceRuntimeCloseRetiredDrains(t *testing.T) { - // Use real (lazily-opened, undialed) handles so Close has an underlying DB. - dbA, err := runtimeTestConfig("a").SqlDB(nil) +func TestSourceRuntimeCloseRetiredClosesOnlyRetiredHandles(t *testing.T) { + initial, err := runtimeTestConfig("gen0").SqlDB(nil) + require.NoError(t, err) + rt := NewSourceRuntime(initial, runtimeTestConfig("gen0")) + + replaced, err := rt.Replace(runtimeTestConfig("gen1"), nil) require.NoError(t, err) - dbB, err := runtimeTestConfig("b").SqlDB(nil) + current, err := rt.Replace(runtimeTestConfig("gen2"), nil) require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, current.Close()) }) - rt := NewSourceRuntime(nil, nil) - rt.retired = []*sql.DB{dbA, dbB} + // Retiring must not close handles that in-flight work may still use. + assert.False(t, isClosed(t, initial)) + assert.False(t, isClosed(t, replaced)) + + rt.CloseRetired() + assert.True(t, isClosed(t, initial), "initial handle must be closed") + assert.True(t, isClosed(t, replaced), "intermediate handle must be closed") + assert.False(t, isClosed(t, current), "current handle must stay open") + assert.Same(t, current, rt.DB()) - assert.NotPanics(t, func() { rt.CloseRetired() }) - assert.Empty(t, rt.retired, "retired handles must be drained") + // Draining is idempotent and never touches the current handle. + rt.CloseRetired() + assert.False(t, isClosed(t, current)) } func TestSourceRuntimeNilInitial(t *testing.T) { From 0dd03fe8cae415d45835dc6d89f3d29bf3284734 Mon Sep 17 00:00:00 2001 From: Leszek Zalewski Date: Tue, 13 Oct 2026 14:15:13 +0200 Subject: [PATCH 3/3] Trim verbose comments --- ferry.go | 10 --------- source_runtime.go | 48 +++++++----------------------------------- source_runtime_test.go | 12 ++--------- 3 files changed, 10 insertions(+), 60 deletions(-) diff --git a/ferry.go b/ferry.go index fe34bbc9..4d822ebd 100644 --- a/ferry.go +++ b/ferry.go @@ -54,11 +54,6 @@ type Ferry struct { SourceDB *sql.DB TargetDB *sql.DB - // sourceRuntime owns the current source connection and its DatabaseConfig. - // It is introduced as the future single source of truth for "the current - // source" so a master failover can repoint every consumer atomically. It is - // initialised alongside SourceDB in Initialize. Consumers are not yet - // migrated to it; SourceDB / Config.Source remain authoritative for now. sourceRuntime *SourceRuntime ControlServer *ControlServer @@ -98,8 +93,6 @@ type Ferry struct { rowCopyCompleteCh chan struct{} } -// SourceRuntime returns the Ferry's source runtime, which owns the current -// source connection and config. It is populated during Initialize. func (f *Ferry) SourceRuntime() *SourceRuntime { return f.sourceRuntime } @@ -380,9 +373,6 @@ func (f *Ferry) Initialize() (err error) { return err } - // Wrap the source connection in a SourceRuntime. This is currently only the - // owner of the initial handle; consumers still read SourceDB directly. A - // later change migrates consumers to the runtime and uses it for failover. f.sourceRuntime = NewSourceRuntime(f.SourceDB, f.Config.Source) err = f.checkConnection("source", f.SourceDB) diff --git a/source_runtime.go b/source_runtime.go index ceeff36c..dba3cce9 100644 --- a/source_runtime.go +++ b/source_runtime.go @@ -6,39 +6,18 @@ import ( sql "github.com/Shopify/ghostferry/sqlwrapper" ) -// SourceRuntime owns the live connection to the source MySQL server and the -// DatabaseConfig it was opened from. -// -// It exists so that the rest of Ghostferry can talk about "the current source" -// as a single, swappable thing rather than each component holding its own copy -// of a *sql.DB and a *DatabaseConfig. Today the source never changes during a -// run, but a source master failover needs to atomically repoint every source -// consumer at a newly promoted writer. Centralising ownership here is the -// prerequisite for doing that without hand-editing a field on every consumer. -// -// A SourceRuntime is safe for concurrent use. Readers obtain the current handle -// via DB()/Config(); a swap (Replace) is serialised against those readers. -// -// This type is intentionally introduced without migrating any consumers yet: -// the Ferry constructs and holds one, but components continue to read -// Ferry.SourceDB / Config.Source directly. A follow-up change migrates the -// consumers to depend on the runtime, and a further change uses Replace to -// implement failover. +// SourceRuntime owns the source connection and its config; safe for concurrent use. type SourceRuntime struct { mu sync.RWMutex db *sql.DB config *DatabaseConfig - // retired holds source DB handles replaced by a swap. They are not closed - // at swap time because in-flight cursors and cached prepared statements may - // still reference them; CloseRetired releases them once no source work is - // running (e.g. at teardown). + // retired holds replaced handles, kept open because in-flight work may + // still reference them; released by CloseRetired. retired []*sql.DB } -// NewSourceRuntime creates a SourceRuntime around an already-open source -// connection and the config it was opened from. Both may be nil for tests that -// only exercise Replace. +// NewSourceRuntime wraps an already-open source connection and its config. func NewSourceRuntime(db *sql.DB, config *DatabaseConfig) *SourceRuntime { return &SourceRuntime{ db: db, @@ -46,30 +25,20 @@ func NewSourceRuntime(db *sql.DB, config *DatabaseConfig) *SourceRuntime { } } -// DB returns the current source connection. func (r *SourceRuntime) DB() *sql.DB { r.mu.RLock() defer r.mu.RUnlock() return r.db } -// Config returns the current source DatabaseConfig. func (r *SourceRuntime) Config() *DatabaseConfig { r.mu.RLock() defer r.mu.RUnlock() return r.config } -// Replace opens a new source connection from the given config and installs it -// as the current source, returning the newly opened *sql.DB. -// -// The previously current handle is retired (retained, not closed) so callers -// that still hold a reference to it — an in-flight cursor, a cached prepared -// statement — do not see it closed out from under them. Use CloseRetired at a -// safe point to release retired handles. -// -// If opening the new connection fails, the current source is left unchanged and -// the error is returned. +// Replace opens a new connection from config and installs it as current, +// retiring (not closing) the previous handle; on failure the source is unchanged. func (r *SourceRuntime) Replace(config *DatabaseConfig, logger Logger) (*sql.DB, error) { newDB, err := config.SqlDB(logger) if err != nil { @@ -87,9 +56,8 @@ func (r *SourceRuntime) Replace(config *DatabaseConfig, logger Logger) (*sql.DB, return newDB, nil } -// CloseRetired closes every retired source handle and clears the retired list. -// It is best-effort and intended to be called during teardown, once no source -// cursors or verification passes remain in flight. +// CloseRetired closes the handles retired by Replace; callers must ensure no +// source work is still in flight. func (r *SourceRuntime) CloseRetired() { r.mu.Lock() retired := r.retired diff --git a/source_runtime_test.go b/source_runtime_test.go index 7191f205..037e7b1d 100644 --- a/source_runtime_test.go +++ b/source_runtime_test.go @@ -9,14 +9,12 @@ import ( "github.com/stretchr/testify/require" ) -// runtimeTestConfig is a DatabaseConfig SqlDB can open lazily (sql.Open does -// not dial), so Replace can install a real handle without a live server. +// SqlDB opens lazily (no dial), so tests get real handles without a server. func runtimeTestConfig(host string) *DatabaseConfig { return &DatabaseConfig{Host: host, Port: 3306, User: "root", Net: "tcp"} } -// isClosed reports whether db rejects work because it was closed. A cancelled -// context makes an open lazily-dialed handle fail fast without dialing. +// isClosed: a cancelled context makes PingContext fail fast without dialing. func isClosed(t *testing.T, db *sql.DB) bool { t.Helper() ctx, cancel := context.WithCancel(context.Background()) @@ -45,12 +43,10 @@ func TestSourceRuntimeReplaceInstallsNewSourceAndRetiresOld(t *testing.T) { require.NoError(t, err) require.NotNil(t, newDB) - // The runtime now serves the new handle/config. assert.Same(t, newDB, rt.DB()) assert.Same(t, newCfg, rt.Config()) assert.NotSame(t, oldDB, rt.DB()) - // The old handle is retired, not closed here. assert.Contains(t, rt.retired, oldDB) } @@ -62,7 +58,6 @@ func TestSourceRuntimeReplaceRetainsAllPriorHandles(t *testing.T) { _, err = rt.Replace(runtimeTestConfig("h2"), nil) require.NoError(t, err) - // Two swaps retire two handles (gen0 and the h1 handle). assert.Len(t, rt.retired, 2) } @@ -77,7 +72,6 @@ func TestSourceRuntimeCloseRetiredClosesOnlyRetiredHandles(t *testing.T) { require.NoError(t, err) t.Cleanup(func() { require.NoError(t, current.Close()) }) - // Retiring must not close handles that in-flight work may still use. assert.False(t, isClosed(t, initial)) assert.False(t, isClosed(t, replaced)) @@ -87,7 +81,6 @@ func TestSourceRuntimeCloseRetiredClosesOnlyRetiredHandles(t *testing.T) { assert.False(t, isClosed(t, current), "current handle must stay open") assert.Same(t, current, rt.DB()) - // Draining is idempotent and never touches the current handle. rt.CloseRetired() assert.False(t, isClosed(t, current)) } @@ -97,7 +90,6 @@ func TestSourceRuntimeNilInitial(t *testing.T) { assert.Nil(t, rt.DB()) assert.Nil(t, rt.Config()) - // Replacing from a nil initial does not retire anything. _, err := rt.Replace(runtimeTestConfig("h1"), nil) require.NoError(t, err) assert.Empty(t, rt.retired)