From 66f830313f34ac49a0690a425c8a94ae9c6a2a69 Mon Sep 17 00:00:00 2001 From: Roger Yen <16497194+rogeryen@users.noreply.github.com> Date: Fri, 9 Oct 2026 13:11:02 -0400 Subject: [PATCH 1/2] sharding: prototype bounded cutover dependency copies by logical key --- docs/cutover-dependencies.md | 154 ++++++++++ sharding/config.go | 6 + sharding/cutover_dependencies.go | 427 ++++++++++++++++++++++++++ sharding/cutover_dependencies_test.go | 296 ++++++++++++++++++ sharding/filter.go | 16 +- sharding/sharding.go | 17 + 6 files changed, 910 insertions(+), 6 deletions(-) create mode 100644 docs/cutover-dependencies.md create mode 100644 sharding/cutover_dependencies.go create mode 100644 sharding/cutover_dependencies_test.go diff --git a/docs/cutover-dependencies.md b/docs/cutover-dependencies.md new file mode 100644 index 000000000..202835e66 --- /dev/null +++ b/docs/cutover-dependencies.md @@ -0,0 +1,154 @@ +--- +layout: default +title: Cutover dependency copy (POC) +--- + +# Cutover dependency copy (POC) + +This opt-in prototype copies small dependency sets during sharding cutover. +It separates three keys: + +- The sharding key selects reference rows. +- Join columns select dependency rows. +- A source pagination key orders the copy. A separate logical key identifies the destination row. + +The feature does not require a sharding-key column on the dependency table. +It does not replace `JoinedTables` or change that option's behavior. + +## Caller contract + +**Do not enable this prototype until the caller can enforce these conditions.** + +- `CutoverLock` drains in-flight writes and prevents new writes to the selected + references and dependencies until cutover completes. +- The same protection covers destination dependency updates and cleanup. + A shop lock alone does not prove that pod-wide jobs honor this contract. +- All required source data exists before cutover. This feature cannot recover + dependencies that were lost before the move. +- Reference tables use InnoDB. Source and target dependency tables use InnoDB, + have matching column definitions, and have no target triggers or generated columns. +- Each dependency has a single integer primary key for pagination and a separate + non-nullable, full-column unique logical key on both databases. +- Query plans and the configured row, byte, and time budgets fit the write-stop window. +- Selected reference rows follow the normal migration path. The caller must not + exclude them from normal copying while it relies on them at the destination. + +Ghostferry checks the schema restrictions at cutover. It cannot establish the +application's lock or reference-copy contract from the configuration. + +## Example + +This example uses fictional `installations`, `versions`, and `modules` tables. +The sharding key is `tenant_id` and the selected tenant is `1`. +Add this section to an otherwise valid sharding configuration with a real +`CutoverLock` callback: + +```json +{ + "CutoverDependencies": { + "MaxRowsPerTable": 1000, + "MaxBytes": 16777216, + "TimeoutSeconds": 10, + "Tables": [ + { + "Table": "versions", + "ReferenceTable": "installations", + "PaginationColumn": "id", + "IdentityColumns": ["app_id", "app_version_id"], + "JoinColumns": [ + {"Column": "app_id", "ReferenceColumn": "app_id"}, + {"Column": "app_version_id", "ReferenceColumn": "deployment_id"} + ], + "ReferenceEquals": {"development": true}, + "Equals": {"release_id": 0}, + "RequireMatch": true + }, + { + "Table": "modules", + "ReferenceTable": "installations", + "PaginationColumn": "id", + "IdentityColumns": ["app_id", "app_version_id", "module_uuid"], + "JoinColumns": [ + {"Column": "app_id", "ReferenceColumn": "app_id"}, + {"Column": "app_version_id", "ReferenceColumn": "deployment_id"} + ], + "ReferenceEquals": {"development": true}, + "Equals": {"release_id": 0} + } + ] + } +} +``` + +`ReferenceEquals` and `Equals` support parameterized scalar equality predicates. +A null value means `IS NULL` through MySQL's null-safe equality operator. +`RequireMatch` requires at least one matching dependency for every selected +reference. It does not assert a module count. The example permits a version with +zero modules but rejects a missing version. + +These explicit dependency tables remain outside normal row copying, binlog +streaming, and the pagination-based verifiers, even if they have a sharding-key +column. `IncludedTables` and `IgnoredTables` govern the normal copy; they do not +cancel an explicit dependency declaration. Do not also declare a dependency in +`JoinedTables` or `PrimaryKeyTables`. Dependency chains are not supported. + +## Cutover sequence + +1. Obtain the existing cutover lock callback's acknowledgment. +2. Wait for the configured source replica to catch up, then flush binlog writes. +3. Complete the existing joined-table delta copy. +4. Open a repeatable-read source snapshot and one target transaction. +5. Select dependencies through the final source references, in batches of 100. +6. Insert each source row with its existing surrogate key. +7. If an insert encounters a duplicate key, leave the destination unchanged. +8. Read the destination by logical key with `FOR UPDATE`. +9. Compare every column except the pagination key byte-for-byte, including nullness. +10. Commit all dependency tables together only after every comparison succeeds. +11. Continue normal verification and cutover. + +For example, source `(id=100, app_id=7, app_version_id=42)` can match destination +`(id=900, app_id=7, app_version_id=42)`. The destination keeps `id=900`. +An unrelated row with `id=100` does not count as a successful copy. A missing +logical identity or any other payload difference aborts the dependency transaction. + +The prototype does not overwrite payload differences, delete destination rows, +allocate replacement IDs, or rewrite references. Timestamps also participate in +verification. This deliberately rejects ambiguous cases instead of choosing a +winner for shared data. + +## Limits and failure behavior + +- `MaxRowsPerTable` must be between 1 and 100000. +- `MaxBytes` bounds the total source field bytes across all dependency tables. + It is not a hard bound on driver buffers or one oversized row allocation. +- `TimeoutSeconds` must be between 1 and 300. It bounds the dependency phase, + not the entire cutover. +- An error rolls back the dependency transaction and takes the existing fatal + error path. The success unlock callback must not run. +- The caller retains responsibility for failure recovery and lock release. +- The dependency phase restarts from the beginning on retry; identical existing + rows make the operation idempotent. +- If a later cutover phase fails, the committed dependency rows remain. This + prototype does not remove them, since they might have other owners. +- A successful copy does not protect against writers after the dependency + transaction commits. The caller contract must hold through route switch. + +## Validation and remaining work + +The MySQL tests cover composite joins, duplicate references, unrelated tenants, +released-row exclusion, multiple modules per version, different destination IDs, +repeat copies, binary/null payloads, keyset pagination, missing dependencies, +surrogate collisions, payload mismatches, schema restrictions, budgets, and a +blocked-write deadline. End-to-end tests cover a final write in the lock callback, +normal target verification, and no success unlock on conflict. + +This remains a POC. Before production use: + +- Confirm all writer and cleanup paths honor the caller contract on both sides. +- Add an application-level lock test with a concurrent development request. +- Validate real query plans and cutover latency with representative data. +- Design an explicit payload-conflict policy if strict equality is too restrictive. +- Add source-replica and failure/retry scenario coverage. +- Add version-gated caller configuration. This patch enables no application tables. + +The feature applies only to Ghostferry's MySQL sharding path, not other migration backends. diff --git a/sharding/config.go b/sharding/config.go index a3e6ab741..0db4ca060 100644 --- a/sharding/config.go +++ b/sharding/config.go @@ -23,6 +23,8 @@ type Config struct { JoinedTables map[string][]JoinTable + CutoverDependencies *CutoverDependenciesConfig + // IgnoredTables and IncludedTables are mutually exclusive. Specifying both is an error. IgnoredTables []string IncludedTables []string @@ -40,6 +42,10 @@ func (c *Config) ValidateConfig() error { return fmt.Errorf("IgnoredTables and IncludedTables cannot be defined at the same time.") } + if err := c.validateCutoverDependencies(); err != nil { + return err + } + if c.RunFerryFromReplica && c.SourceReplicationMaster != nil { if err := c.SourceReplicationMaster.Validate(); err != nil { return err diff --git a/sharding/cutover_dependencies.go b/sharding/cutover_dependencies.go new file mode 100644 index 000000000..88fc4a30e --- /dev/null +++ b/sharding/cutover_dependencies.go @@ -0,0 +1,427 @@ +package sharding + +import ( + "bytes" + "context" + "database/sql" + "errors" + "fmt" + "reflect" + "regexp" + "sort" + "strings" + "time" + + "github.com/Shopify/ghostferry" + "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/go-sql-driver/mysql" +) + +// CutoverDependenciesConfig is an opt-in, bounded copy for dependencies whose +// writers the caller must stop through CutoverLock. These tables are not streamed. +type CutoverDependenciesConfig struct { + Tables []CutoverDependency + MaxRowsPerTable int + MaxBytes int64 + TimeoutSeconds int +} + +type CutoverDependency struct { + Table string + ReferenceTable string + PaginationColumn string + IdentityColumns []string + JoinColumns []DependencyJoin + ReferenceEquals map[string]interface{} + Equals map[string]interface{} + RequireMatch bool +} + +type DependencyJoin struct { + Column string + ReferenceColumn string +} + +var dependencyIdentifier = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`) + +func (c *Config) validateCutoverDependencies() error { + d := c.CutoverDependencies + if d == nil { + return nil + } + if len(d.Tables) == 0 || d.MaxRowsPerTable <= 0 || d.MaxRowsPerTable > 100000 || d.MaxBytes <= 0 || d.TimeoutSeconds <= 0 || d.TimeoutSeconds > 300 { + return fmt.Errorf("CutoverDependencies requires tables, MaxRowsPerTable (1..100000), MaxBytes and TimeoutSeconds (1..300)") + } + if c.CutoverLock.URI == "" { + return fmt.Errorf("CutoverDependencies requires a CutoverLock callback that stops dependency writers") + } + names := map[string]bool{} + for _, t := range d.Tables { + if names[t.Table] { + return fmt.Errorf("duplicate cutover dependency %q", t.Table) + } + names[t.Table] = true + if len(t.JoinColumns) == 0 || len(t.IdentityColumns) == 0 || t.Table == t.ReferenceTable { + return fmt.Errorf("dependency %q requires a distinct reference table, joins and identity columns", t.Table) + } + if _, ok := c.JoinedTables[t.Table]; ok { + return fmt.Errorf("dependency %q cannot also be a JoinedTable", t.Table) + } + for _, name := range c.PrimaryKeyTables { + if name == t.Table { + return fmt.Errorf("dependency %q cannot also be a PrimaryKeyTable", t.Table) + } + } + identifiers := []string{c.SourceDB, c.TargetDB, c.ShardingKey, t.Table, t.ReferenceTable, t.PaginationColumn} + seen := map[string]bool{} + for _, name := range t.IdentityColumns { + if seen[name] || name == t.PaginationColumn { + return fmt.Errorf("dependency %q requires distinct logical identity columns separate from pagination", t.Table) + } + seen[name] = true + identifiers = append(identifiers, name) + } + for _, join := range t.JoinColumns { + identifiers = append(identifiers, join.Column, join.ReferenceColumn) + } + for _, conditions := range []map[string]interface{}{t.ReferenceEquals, t.Equals} { + for name, value := range conditions { + identifiers = append(identifiers, name) + switch value.(type) { + case nil, string, bool, float64, int, int64: + default: + return fmt.Errorf("dependency %q condition %q must be a scalar", t.Table, name) + } + } + } + for _, name := range identifiers { + if !dependencyIdentifier.MatchString(name) { + return fmt.Errorf("invalid dependency identifier %q", name) + } + } + } + for _, t := range d.Tables { + if names[t.ReferenceTable] { + return fmt.Errorf("dependency chains are not supported: %q", t.ReferenceTable) + } + } + return nil +} + +func (r *ShardingFerry) copyCutoverDependencies() error { + config := r.config.CutoverDependencies + if config == nil { + return nil + } + ctx, cancel := context.WithTimeout(context.Background(), time.Duration(config.TimeoutSeconds)*time.Second) + defer cancel() + source, err := r.Ferry.SourceDB.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true}) + if err != nil { + return err + } + defer source.Rollback() + target, err := r.Ferry.TargetDB.BeginTx(ctx, nil) + if err != nil { + return err + } + defer target.Rollback() + var copiedBytes int64 + for _, table := range config.Tables { + if err := r.copyDependency(ctx, source, target, table, &copiedBytes); err != nil { + return fmt.Errorf("cutover dependency %s: %w", table.Table, err) + } + } + return target.Commit() +} + +type dependencyColumn struct { + Name string + Type string + Nullable string + Extra string + Collation string +} + +func dependencyColumns(ctx context.Context, tx *sql.Tx, database, table string) ([]dependencyColumn, error) { + rows, err := tx.QueryContext(ctx, "SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, EXTRA, COALESCE(COLLATION_NAME, '') FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? ORDER BY ORDINAL_POSITION", database, table) + if err != nil { + return nil, err + } + defer rows.Close() + var columns []dependencyColumn + for rows.Next() { + var c dependencyColumn + if err := rows.Scan(&c.Name, &c.Type, &c.Nullable, &c.Extra, &c.Collation); err != nil { + return nil, err + } + columns = append(columns, c) + } + return columns, rows.Err() +} + +func dependencyHasUniqueKey(ctx context.Context, tx *sql.Tx, database, table string, columns []string, primary bool) (bool, error) { + rows, err := tx.QueryContext(ctx, "SELECT INDEX_NAME, COLUMN_NAME, SUB_PART FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND NON_UNIQUE = 0 ORDER BY INDEX_NAME, SEQ_IN_INDEX", database, table) + if err != nil { + return false, err + } + defer rows.Close() + keys := map[string][]string{} + for rows.Next() { + var index string + var column sql.NullString + var prefix sql.NullInt64 + if err := rows.Scan(&index, &column, &prefix); err != nil { + return false, err + } + if prefix.Valid || !column.Valid { + keys[index] = append(keys[index], "") + } else { + keys[index] = append(keys[index], column.String) + } + } + if err := rows.Err(); err != nil { + return false, err + } + for index, key := range keys { + if (!primary || index == "PRIMARY") && reflect.DeepEqual(key, columns) { + return true, nil + } + } + return false, nil +} + +func (r *ShardingFerry) dependencySchema(ctx context.Context, source, target *sql.Tx, table CutoverDependency) ([]string, error) { + for _, side := range []struct { + tx *sql.Tx + database, table string + }{ + {source, r.config.SourceDB, table.Table}, + {source, r.config.SourceDB, table.ReferenceTable}, + {target, r.config.TargetDB, table.Table}, + } { + var engine string + if err := side.tx.QueryRowContext(ctx, "SELECT ENGINE FROM information_schema.TABLES WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?", side.database, side.table).Scan(&engine); err != nil { + return nil, err + } + if engine != "InnoDB" { + return nil, fmt.Errorf("dependency and reference tables must use InnoDB") + } + } + var triggers int + if err := target.QueryRowContext(ctx, "SELECT COUNT(*) FROM information_schema.TRIGGERS WHERE EVENT_OBJECT_SCHEMA = ? AND EVENT_OBJECT_TABLE = ?", r.config.TargetDB, table.Table).Scan(&triggers); err != nil { + return nil, err + } + if triggers != 0 { + return nil, fmt.Errorf("target dependency triggers are not supported") + } + sourceColumns, err := dependencyColumns(ctx, source, r.config.SourceDB, table.Table) + if err != nil { + return nil, err + } + targetColumns, err := dependencyColumns(ctx, target, r.config.TargetDB, table.Table) + if err != nil { + return nil, err + } + if len(sourceColumns) == 0 || !reflect.DeepEqual(sourceColumns, targetColumns) { + return nil, fmt.Errorf("source and target column definitions must match") + } + names := make([]string, len(sourceColumns)) + for i, col := range sourceColumns { + names[i] = col.Name + if strings.Contains(col.Extra, "VIRTUAL GENERATED") || strings.Contains(col.Extra, "STORED GENERATED") { + return nil, fmt.Errorf("generated columns are not supported") + } + if col.Name == table.PaginationColumn && !strings.Contains(col.Type, "int") { + return nil, fmt.Errorf("pagination column must be an integer") + } + for _, identity := range table.IdentityColumns { + if col.Name == identity && col.Nullable != "NO" { + return nil, fmt.Errorf("logical identity columns must be non-nullable") + } + } + } + for _, side := range []struct { + tx *sql.Tx + database string + }{{source, r.config.SourceDB}, {target, r.config.TargetDB}} { + for _, key := range []struct { + columns []string + primary bool + }{{[]string{table.PaginationColumn}, true}, {table.IdentityColumns, false}} { + ok, err := dependencyHasUniqueKey(ctx, side.tx, side.database, table.Table, key.columns, key.primary) + if err != nil { + return nil, err + } + if !ok { + return nil, fmt.Errorf("%s requires a full unique logical key and a single-column pagination primary key", side.database) + } + } + } + return names, nil +} + +func dependencyEquals(alias string, values map[string]interface{}, args *[]interface{}) []string { + names := make([]string, 0, len(values)) + for name := range values { + names = append(names, name) + } + sort.Strings(names) + predicates := make([]string, 0, len(names)) + for _, name := range names { + predicates = append(predicates, alias+"."+ghostferry.QuoteField(name)+" <=> ?") + *args = append(*args, values[name]) + } + return predicates +} + +func (r *ShardingFerry) dependencyPredicates(table CutoverDependency) ([]string, []interface{}) { + predicates := []string{"r." + ghostferry.QuoteField(r.config.ShardingKey) + " = ?"} + args := []interface{}{r.config.ShardingValue} + for _, join := range table.JoinColumns { + predicates = append(predicates, "d."+ghostferry.QuoteField(join.Column)+" = r."+ghostferry.QuoteField(join.ReferenceColumn)) + } + predicates = append(predicates, dependencyEquals("r", table.ReferenceEquals, &args)...) + predicates = append(predicates, dependencyEquals("d", table.Equals, &args)...) + return predicates, args +} + +func dependencyReadRow(scanner interface{ Scan(...interface{}) error }, count int) ([][]byte, error) { + values := make([][]byte, count) + pointers := make([]interface{}, count) + for i := range values { + pointers[i] = &values[i] + } + err := scanner.Scan(pointers...) + return values, err +} + +func (r *ShardingFerry) copyDependency(ctx context.Context, source, target *sql.Tx, table CutoverDependency, copiedBytes *int64) error { + columns, err := r.dependencySchema(ctx, source, target, table) + if err != nil { + return err + } + sourceName := ghostferry.QuotedTableNameFromString(r.config.SourceDB, table.Table) + referenceName := ghostferry.QuotedTableNameFromString(r.config.SourceDB, table.ReferenceTable) + targetName := ghostferry.QuotedTableNameFromString(r.config.TargetDB, table.Table) + predicates, args := r.dependencyPredicates(table) + if table.RequireMatch { + // A missing required dependency must not turn an empty SELECT into success. + refArgs := []interface{}{r.config.ShardingValue} + refPredicates := []string{"r." + ghostferry.QuoteField(r.config.ShardingKey) + " = ?"} + refPredicates = append(refPredicates, dependencyEquals("r", table.ReferenceEquals, &refArgs)...) + inner := make([]string, 0, len(table.JoinColumns)) + for _, join := range table.JoinColumns { + inner = append(inner, "d."+ghostferry.QuoteField(join.Column)+" = r."+ghostferry.QuoteField(join.ReferenceColumn)) + } + inner = append(inner, dependencyEquals("d", table.Equals, &refArgs)...) + query := "SELECT EXISTS (SELECT 1 FROM " + referenceName + " r WHERE " + strings.Join(refPredicates, " AND ") + " AND NOT EXISTS (SELECT 1 FROM " + sourceName + " d WHERE " + strings.Join(inner, " AND ") + "))" + var missing bool + if err := source.QueryRowContext(ctx, query, refArgs...).Scan(&missing); err != nil { + return err + } + if missing { + return fmt.Errorf("required source dependency is missing") + } + } + selectColumns := make([]string, len(columns)) + insertColumns := ghostferry.QuoteFields(columns) + paginationIndex := 0 + for i, name := range columns { + selectColumns[i] = "BINARY d." + ghostferry.QuoteField(name) + if name == table.PaginationColumn { + paginationIndex = i + } + } + // Bounded keyset iteration uses the source surrogate key, not the logical key. + var last []byte + copied := 0 + for { + query := "SELECT " + strings.Join(selectColumns, ",") + " FROM " + sourceName + " d WHERE EXISTS (SELECT 1 FROM " + referenceName + " r WHERE " + strings.Join(predicates, " AND ") + ")" + batchArgs := append([]interface{}{}, args...) + if last != nil { + query += " AND d." + ghostferry.QuoteField(table.PaginationColumn) + " > ?" + batchArgs = append(batchArgs, last) + } + query += " ORDER BY d." + ghostferry.QuoteField(table.PaginationColumn) + " LIMIT 100" + rows, err := source.QueryContext(ctx, query, batchArgs...) + if err != nil { + return err + } + batchCount := 0 + for rows.Next() { + values, err := dependencyReadRow(rows, len(columns)) + if err != nil { + rows.Close() + return err + } + copied++ + for _, value := range values { + *copiedBytes += int64(len(value)) + } + if copied > r.config.CutoverDependencies.MaxRowsPerTable || *copiedBytes > r.config.CutoverDependencies.MaxBytes { + rows.Close() + return fmt.Errorf("dependency copy exceeded row or byte budget") + } + if err := r.insertAndVerifyDependency(ctx, target, table, targetName, columns, insertColumns, values); err != nil { + rows.Close() + return err + } + last = values[paginationIndex] + batchCount++ + } + err = rows.Err() + rows.Close() + if err != nil { + return err + } + if batchCount < 100 { + r.logger.WithField("table", table.Table).WithField("rows", copied).Info("verified cutover dependency rows; transaction pending") + return nil + } + } +} + +func (r *ShardingFerry) insertAndVerifyDependency(ctx context.Context, target *sql.Tx, table CutoverDependency, targetName string, columns, insertColumns []string, values [][]byte) error { + args := make([]interface{}, len(values)) + for i, value := range values { + if value != nil { + args[i] = value + } + } + query := "INSERT INTO " + targetName + " (" + strings.Join(insertColumns, ",") + ") VALUES (" + strings.TrimSuffix(strings.Repeat("?,", len(columns)), ",") + ")" + _, err := target.ExecContext(ctx, sqlwrapper.AnnotateStmt(query, r.Ferry.TargetDB.Marginalia), args...) + if err != nil { + var duplicate *mysql.MySQLError + if !errors.As(err, &duplicate) || duplicate.Number != 1062 { + return err + } + } + predicates := make([]string, 0, len(table.IdentityColumns)) + identityArgs := make([]interface{}, 0, len(table.IdentityColumns)) + for _, identity := range table.IdentityColumns { + predicates = append(predicates, ghostferry.QuoteField(identity)+" = ?") + for i, name := range columns { + if name == identity { + identityArgs = append(identityArgs, args[i]) + } + } + } + selected := make([]string, len(columns)) + for i, name := range columns { + selected[i] = "BINARY " + ghostferry.QuoteField(name) + } + query = "SELECT " + strings.Join(selected, ",") + " FROM " + targetName + " WHERE " + strings.Join(predicates, " AND ") + " FOR UPDATE" + actual, err := dependencyReadRow(target.QueryRowContext(ctx, query, identityArgs...), len(columns)) + if err != nil { + return fmt.Errorf("destination logical identity not found after insert (possible surrogate-key collision): %w", err) + } + for i, name := range columns { + if name == table.PaginationColumn { + continue + } + if (actual[i] == nil) != (values[i] == nil) || !bytes.Equal(actual[i], values[i]) { + return fmt.Errorf("destination payload mismatch in column %s", name) + } + } + return nil +} diff --git a/sharding/cutover_dependencies_test.go b/sharding/cutover_dependencies_test.go new file mode 100644 index 000000000..e50b5283f --- /dev/null +++ b/sharding/cutover_dependencies_test.go @@ -0,0 +1,296 @@ +package sharding + +import ( + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Shopify/ghostferry" + "github.com/go-mysql-org/go-mysql/schema" + + "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/Shopify/ghostferry/testhelpers" + "github.com/stretchr/testify/require" +) + +func dependencyTestConfig() *Config { + return &Config{ + Config: testhelpers.NewTestConfig(), + SourceDB: "gftest_dependency_source", TargetDB: "gftest_dependency_target", + ShardingKey: "tenant_id", ShardingValue: 1, + CutoverDependencies: &CutoverDependenciesConfig{ + MaxRowsPerTable: 1000, MaxBytes: 1024 * 1024, TimeoutSeconds: 10, + Tables: []CutoverDependency{ + {Table: "versions", ReferenceTable: "installations", PaginationColumn: "id", + IdentityColumns: []string{"app_id", "app_version_id"}, + JoinColumns: []DependencyJoin{{"app_id", "app_id"}, {"app_version_id", "deployment_id"}}, + ReferenceEquals: map[string]interface{}{"development": true}, Equals: map[string]interface{}{"release_id": 0}, RequireMatch: true}, + {Table: "modules", ReferenceTable: "installations", PaginationColumn: "id", + IdentityColumns: []string{"app_id", "app_version_id", "module_uuid"}, + JoinColumns: []DependencyJoin{{"app_id", "app_id"}, {"app_version_id", "deployment_id"}}, + ReferenceEquals: map[string]interface{}{"development": true}, Equals: map[string]interface{}{"release_id": 0}}, + }, + }, + } +} + +func dependencyFixture(t *testing.T) *ShardingFerry { + t.Helper() + config := dependencyTestConfig() + config.CutoverLock.URI = "http://unused.invalid/lock" + ferry, err := NewFerry(config) + require.NoError(t, err) + ferry.Ferry.SourceDB, err = config.Source.SqlDB(nil) + require.NoError(t, err) + ferry.Ferry.TargetDB, err = config.Target.SqlDB(nil) + require.NoError(t, err) + for _, side := range []struct { + db *sqlwrapper.DB + name string + }{{ferry.Ferry.SourceDB, config.SourceDB}, {ferry.Ferry.TargetDB, config.TargetDB}} { + db, name := side.db, side.name + _, err := db.Exec("DROP DATABASE IF EXISTS " + name) + require.NoError(t, err) + _, err = db.Exec("CREATE DATABASE " + name) + require.NoError(t, err) + t.Cleanup(func() { db.Exec("DROP DATABASE IF EXISTS " + name); db.Close() }) + for _, ddl := range []string{ + "CREATE TABLE %s.installations (id bigint NOT NULL PRIMARY KEY, tenant_id bigint NOT NULL, app_id bigint NOT NULL, deployment_id bigint NOT NULL, development boolean NOT NULL, KEY (tenant_id, id)) ENGINE=InnoDB", + "CREATE TABLE %s.versions (id bigint NOT NULL AUTO_INCREMENT PRIMARY KEY, app_id bigint NOT NULL, app_version_id bigint NOT NULL, release_id bigint NOT NULL, payload blob, UNIQUE KEY logical_identity (app_id, app_version_id)) ENGINE=InnoDB", + "CREATE TABLE %s.modules (id bigint NOT NULL AUTO_INCREMENT PRIMARY KEY, app_id bigint NOT NULL, app_version_id bigint NOT NULL, module_uuid varchar(100) NOT NULL, release_id bigint NOT NULL, payload blob, UNIQUE KEY logical_identity (app_id, app_version_id, module_uuid)) ENGINE=InnoDB", + } { + _, err := db.Exec(fmt.Sprintf(ddl, name)) + require.NoError(t, err) + } + } + return ferry +} + +func dependencyExec(t *testing.T, db *sqlwrapper.DB, query string, args ...interface{}) { + t.Helper() + _, err := db.Exec(query, args...) + require.NoError(t, err) +} + +func dependencyCount(t *testing.T, db *sqlwrapper.DB, table string) int { + t.Helper() + var count int + require.NoError(t, db.QueryRow("SELECT COUNT(*) FROM "+table).Scan(&count)) + return count +} + +func TestCutoverDependenciesPreserveLogicalIdentity(t *testing.T) { + r := dependencyFixture(t) + source, target := r.Ferry.SourceDB, r.Ferry.TargetDB + dependencyExec(t, source, "INSERT INTO gftest_dependency_source.installations VALUES (1,1,7,42,1),(2,1,7,42,1),(3,2,7,43,1),(4,1,7,44,0)") + dependencyExec(t, source, "INSERT INTO gftest_dependency_source.versions VALUES (100,7,42,0,?),(101,7,43,0,'other shop'),(102,7,44,10,'released')", []byte{'a', 0, 'b'}) + dependencyExec(t, source, "INSERT INTO gftest_dependency_source.modules VALUES (200,7,42,'a',0,NULL),(201,7,42,'b',0,''),(202,7,43,'a',0,'other shop')") + dependencyExec(t, target, "INSERT INTO gftest_dependency_target.versions VALUES (900,7,42,0,?)", []byte{'a', 0, 'b'}) + dependencyExec(t, target, "INSERT INTO gftest_dependency_target.modules VALUES (901,7,42,'a',0,NULL)") + for i := 0; i < 2; i++ { + require.NoError(t, r.copyCutoverDependencies()) + } + var id int + require.NoError(t, target.QueryRow("SELECT id FROM gftest_dependency_target.versions WHERE app_id=7 AND app_version_id=42").Scan(&id)) + require.Equal(t, 900, id) + require.NoError(t, target.QueryRow("SELECT id FROM gftest_dependency_target.modules WHERE module_uuid='a'").Scan(&id)) + require.Equal(t, 901, id) + require.Equal(t, 1, dependencyCount(t, target, "gftest_dependency_target.versions")) + require.Equal(t, 2, dependencyCount(t, target, "gftest_dependency_target.modules")) +} + +func TestCutoverDependenciesAbortAtomically(t *testing.T) { + for _, name := range []string{"payload", "surrogate_collision", "missing_source", "row_budget", "byte_budget", "null_is_not_empty"} { + t.Run(name, func(t *testing.T) { + r := dependencyFixture(t) + source, target := r.Ferry.SourceDB, r.Ferry.TargetDB + dependencyExec(t, source, "INSERT INTO gftest_dependency_source.installations VALUES (1,1,7,42,1)") + dependencyExec(t, source, "INSERT INTO gftest_dependency_source.versions VALUES (100,7,42,0,'version')") + dependencyExec(t, source, "INSERT INTO gftest_dependency_source.modules VALUES (200,7,42,'a',0,'source'),(201,7,42,'b',0,'second')") + want := "" + switch name { + case "payload": + dependencyExec(t, target, "INSERT INTO gftest_dependency_target.modules VALUES (900,7,42,'a',0,'different')") + want = "payload mismatch" + case "surrogate_collision": + dependencyExec(t, target, "INSERT INTO gftest_dependency_target.modules VALUES (200,8,99,'a',0,'unrelated')") + want = "surrogate-key collision" + case "missing_source": + dependencyExec(t, source, "DELETE FROM gftest_dependency_source.versions") + want = "required source dependency is missing" + case "row_budget": + r.config.CutoverDependencies.MaxRowsPerTable = 1 + want = "budget" + case "byte_budget": + r.config.CutoverDependencies.MaxBytes = 1 + want = "budget" + case "null_is_not_empty": + dependencyExec(t, source, "UPDATE gftest_dependency_source.modules SET payload=NULL WHERE id=200") + dependencyExec(t, target, "INSERT INTO gftest_dependency_target.modules VALUES (900,7,42,'a',0,'')") + want = "payload mismatch" + } + before := dependencyCount(t, target, "gftest_dependency_target.modules") + require.ErrorContains(t, r.copyCutoverDependencies(), want) + require.Zero(t, dependencyCount(t, target, "gftest_dependency_target.versions")) + require.Equal(t, before, dependencyCount(t, target, "gftest_dependency_target.modules")) + }) + } +} + +func TestCutoverDependenciesKeysetPagination(t *testing.T) { + r := dependencyFixture(t) + dependencyExec(t, r.Ferry.SourceDB, "INSERT INTO gftest_dependency_source.installations VALUES (1,1,7,42,1)") + dependencyExec(t, r.Ferry.SourceDB, "INSERT INTO gftest_dependency_source.versions VALUES (-5,7,42,0,'version')") + for i := 1; i <= 205; i++ { + dependencyExec(t, r.Ferry.SourceDB, "INSERT INTO gftest_dependency_source.modules VALUES (?,7,42,?,0,'module')", i, fmt.Sprint(i)) + } + require.NoError(t, r.copyCutoverDependencies()) + require.Equal(t, 205, dependencyCount(t, r.Ferry.TargetDB, "gftest_dependency_target.modules")) +} + +func TestCutoverDependenciesSchemaValidation(t *testing.T) { + for _, change := range []string{ + "ALTER TABLE gftest_dependency_target.versions DROP INDEX logical_identity", + "ALTER TABLE gftest_dependency_target.versions ADD extra_column int", + "DROP TABLE gftest_dependency_target.versions", + "ALTER TABLE gftest_dependency_target.versions ENGINE=MyISAM", + "CREATE TRIGGER gftest_dependency_target.dependency_trigger BEFORE INSERT ON gftest_dependency_target.versions FOR EACH ROW SET NEW.payload='changed'", + } { + t.Run(change, func(t *testing.T) { + r := dependencyFixture(t) + dependencyExec(t, r.Ferry.TargetDB, change) + require.Error(t, r.copyCutoverDependencies()) + }) + } +} + +func TestCutoverDependenciesConfigValidation(t *testing.T) { + cases := map[string]func(*Config){ + "no callback": func(c *Config) { c.CutoverLock.URI = "" }, + "no budget": func(c *Config) { c.CutoverDependencies.MaxRowsPerTable = 0 }, + "no bytes": func(c *Config) { c.CutoverDependencies.MaxBytes = 0 }, + "no deadline": func(c *Config) { c.CutoverDependencies.TimeoutSeconds = 0 }, + "empty tables": func(c *Config) { c.CutoverDependencies.Tables = nil }, + "duplicate": func(c *Config) { + c.CutoverDependencies.Tables = append(c.CutoverDependencies.Tables, c.CutoverDependencies.Tables[0]) + }, + "joined": func(c *Config) { + c.JoinedTables = map[string][]JoinTable{"versions": {{TableName: "installations", JoinColumn: "deployment_id"}}} + }, + "primary": func(c *Config) { c.PrimaryKeyTables = []string{"versions"} }, + "identifier": func(c *Config) { c.CutoverDependencies.Tables[0].Table = "versions`" }, + "identity": func(c *Config) { c.CutoverDependencies.Tables[0].IdentityColumns = []string{"id"} }, + "join": func(c *Config) { c.CutoverDependencies.Tables[0].JoinColumns = nil }, + "chain": func(c *Config) { c.CutoverDependencies.Tables[1].ReferenceTable = "versions" }, + } + for name, modify := range cases { + t.Run(name, func(t *testing.T) { + c := dependencyTestConfig() + c.CutoverLock.URI = "http://unused.invalid/lock" + modify(c) + _, err := NewFerry(c) + require.Error(t, err) + }) + } +} + +func TestCutoverDependenciesRunAfterLock(t *testing.T) { + for _, conflict := range []bool{false, true} { + t.Run(fmt.Sprint("conflict=", conflict), func(t *testing.T) { + r := dependencyFixture(t) + dependencyExec(t, r.Ferry.SourceDB, "INSERT INTO gftest_dependency_source.installations VALUES (1,1,7,41,1),(2,2,7,99,1)") + dependencyExec(t, r.Ferry.SourceDB, "INSERT INTO gftest_dependency_source.versions VALUES (100,7,41,0,'old')") + if conflict { + dependencyExec(t, r.Ferry.TargetDB, "INSERT INTO gftest_dependency_target.versions VALUES (900,7,42,0,'conflict')") + } + locked, unlocked := false, false + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + switch req.URL.Path { + case "/lock": + locked = true + // This last in-flight write completes before the callback acknowledges the lock. + dependencyExec(t, r.Ferry.SourceDB, "INSERT INTO gftest_dependency_source.versions VALUES (101,7,42,0,'final')") + dependencyExec(t, r.Ferry.SourceDB, "INSERT INTO gftest_dependency_source.modules VALUES (201,7,42,'a',0,'final')") + dependencyExec(t, r.Ferry.SourceDB, "UPDATE gftest_dependency_source.installations SET deployment_id=42 WHERE id=1") + case "/unlock": + unlocked = true + require.Equal(t, 1, dependencyCount(t, r.Ferry.TargetDB, "gftest_dependency_target.modules")) + default: + t.Errorf("unexpected callback %s", req.URL.Path) + } + })) + defer server.Close() + r.config.CutoverLock.URI = server.URL + "/lock" + r.config.CutoverUnlock.URI = server.URL + "/unlock" + r.config.SkipTargetVerification = false + require.NoError(t, r.Initialize()) + t.Cleanup(func() { r.Ferry.SourceDB.Close(); r.Ferry.TargetDB.Close() }) + errors := &testhelpers.ErrorHandler{} + r.Ferry.ErrorHandler = errors + require.NoError(t, r.Start()) + r.Run() + require.True(t, locked) + if conflict { + require.ErrorContains(t, errors.LastError, "payload mismatch") + require.False(t, unlocked) + r.Ferry.StopTargetVerifier() + } else { + require.NoError(t, errors.LastError) + require.True(t, unlocked) + require.Equal(t, 1, dependencyCount(t, r.Ferry.TargetDB, "gftest_dependency_target.versions")) + var version int + require.NoError(t, r.Ferry.TargetDB.QueryRow("SELECT deployment_id FROM gftest_dependency_target.installations WHERE id=1").Scan(&version)) + require.Equal(t, 42, version) + require.Equal(t, 1, dependencyCount(t, r.Ferry.TargetDB, "gftest_dependency_target.installations")) + } + }) + } +} + +func TestCutoverDependenciesDeadline(t *testing.T) { + r := dependencyFixture(t) + r.config.CutoverDependencies.TimeoutSeconds = 1 + dependencyExec(t, r.Ferry.SourceDB, "INSERT INTO gftest_dependency_source.installations VALUES (1,1,7,42,1)") + dependencyExec(t, r.Ferry.SourceDB, "INSERT INTO gftest_dependency_source.versions VALUES (100,7,42,0,'version')") + dependencyExec(t, r.Ferry.TargetDB, "INSERT INTO gftest_dependency_target.versions VALUES (900,7,42,0,'version')") + blocker, err := r.Ferry.TargetDB.Begin() + require.NoError(t, err) + defer blocker.Rollback() + _, err = blocker.Exec("UPDATE gftest_dependency_target.versions SET payload='locked' WHERE id=900") + require.NoError(t, err) + start := time.Now() + require.Error(t, r.copyCutoverDependencies()) + require.Less(t, time.Since(start), 5*time.Second) +} + +func TestCutoverDependenciesExcludedFromNormalCopy(t *testing.T) { + config := dependencyTestConfig() + config.CutoverLock.URI = "http://unused.invalid/lock" + r, err := NewFerry(config) + require.NoError(t, err) + tables, err := r.config.TableFilter.ApplicableTables([]*ghostferry.TableSchema{ + {Table: &schema.Table{Name: "versions", Columns: []schema.TableColumn{{Name: "tenant_id"}}}}, + {Table: &schema.Table{Name: "installations", Columns: []schema.TableColumn{{Name: "tenant_id"}}}}, + }) + require.NoError(t, err) + require.Len(t, tables, 1) + require.Equal(t, "installations", tables[0].Name) +} + +func TestCutoverDependenciesNoReferences(t *testing.T) { + r := dependencyFixture(t) + dependencyExec(t, r.Ferry.SourceDB, "INSERT INTO gftest_dependency_source.versions VALUES (100,7,42,0,'orphan')") + dependencyExec(t, r.Ferry.TargetDB, "INSERT INTO gftest_dependency_target.versions VALUES (900,8,99,0,'unrelated')") + require.NoError(t, r.copyCutoverDependencies()) + require.Equal(t, 1, dependencyCount(t, r.Ferry.TargetDB, "gftest_dependency_target.versions")) + var id int + require.NoError(t, r.Ferry.TargetDB.QueryRow("SELECT id FROM gftest_dependency_target.versions").Scan(&id)) + require.Equal(t, 900, id) +} + +func TestCutoverDependenciesDisabled(t *testing.T) { + r := &ShardingFerry{config: &Config{}} + require.NoError(t, r.copyCutoverDependencies()) +} diff --git a/sharding/filter.go b/sharding/filter.go index a29145b49..e81f14690 100644 --- a/sharding/filter.go +++ b/sharding/filter.go @@ -264,12 +264,13 @@ const ( ) type ShardedTableFilter struct { - SourceShard string - ShardingKey string - JoinedTables map[string][]JoinTable - Type ShardedTableFilterType - Tables []*regexp.Regexp - PrimaryKeyTables map[string]struct{} + SourceShard string + ShardingKey string + JoinedTables map[string][]JoinTable + Type ShardedTableFilterType + Tables []*regexp.Regexp + PrimaryKeyTables map[string]struct{} + CutoverDependencyTables map[string]struct{} } func (s *ShardedTableFilter) isIgnoreFilter() bool { @@ -286,6 +287,9 @@ func (s *ShardedTableFilter) ApplicableDatabases(dbs []string) ([]string, error) func (s *ShardedTableFilter) ApplicableTables(tables []*ghostferry.TableSchema) (applicable []*ghostferry.TableSchema, err error) { for _, table := range tables { + if _, dependency := s.CutoverDependencyTables[table.Name]; dependency { + continue + } if (s.isIgnoreFilter() && s.isPresent(table)) || (s.isIncludeFilter() && !s.isPresent(table)) { continue } diff --git a/sharding/sharding.go b/sharding/sharding.go index 4a08efbc9..951d544de 100644 --- a/sharding/sharding.go +++ b/sharding/sharding.go @@ -52,6 +52,14 @@ func NewFerry(config *Config) (*ShardingFerry, error) { } } + if config.CutoverDependencies != nil { + dependencyTables := map[string]struct{}{} + for _, table := range config.CutoverDependencies.Tables { + dependencyTables[table.Table] = struct{}{} + } + tableFilter.(*ShardedTableFilter).CutoverDependencyTables = dependencyTables + } + config.TableFilter = tableFilter if err := config.ValidateConfig(); err != nil { @@ -158,6 +166,15 @@ func (r *ShardingFerry) Run() { r.Ferry.ErrorHandler.Fatal("sharding.delta_copy", err) } + metrics.Measure("CopyCutoverDependencies", nil, 1.0, func() { + err = r.copyCutoverDependencies() + }) + if err != nil { + r.logger.WithError(err).Error("failed to copy and verify cutover dependencies") + r.Ferry.ErrorHandler.Fatal("sharding.cutover_dependencies", err) + return + } + var verificationResult ghostferry.VerificationResult metrics.Measure("VerifyCutover", nil, 1.0, func() { verificationResult, err = r.Ferry.Verifier.VerifyDuringCutover() From 2ba246abd9350fd7024d9a00bfd1a54a7c8c7ca9 Mon Sep 17 00:00:00 2001 From: Roger Yen <16497194+rogeryen@users.noreply.github.com> Date: Fri, 9 Oct 2026 13:22:37 -0400 Subject: [PATCH 2/2] test: include cutover dependency tests in the MySQL CI matrix --- Makefile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Makefile b/Makefile index 4098bf8a3..da9757219 100644 --- a/Makefile +++ b/Makefile @@ -64,7 +64,7 @@ test-go: curl -sL $(GOTESTSUM_URL) | tar -xz -C ./bin gotestsum; \ fi - ulimit -n 1024 && ./bin/gotestsum --format short-verbose ./test/go ./copydb/test ./sharding/test -count 1 -p 1 -failfast + ulimit -n 1024 && ./bin/gotestsum --format short-verbose ./test/go ./copydb/test ./sharding ./sharding/test -count 1 -p 1 -failfast go test -v test-ruby: