Introduce SourceRuntime to own the source connection (GTID failover 1/3)

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")

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()
+		}
+	}
+}

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)
+}