diff --git a/ferry.go b/ferry.go index 60048d9a..a0ff1299 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) +}