From bb73bbb21ec1cca33be968695095eca3e633ba66 Mon Sep 17 00:00:00 2001 From: Bastian Bartmann Date: Mon, 28 Sep 2026 17:42:36 +0200 Subject: [PATCH] Checkpoint only fully applied binlog transactions Route ordered completion markers through the existing DML queues and persist a separate applied-transaction watermark only after preceding SQL commits. Preserve batching and reject acknowledgements after failed delivery or migration cancellation. Make reader operations and checkpoint SQL cancellation-aware. Cover multi-row events, MySQL/MariaDB GTID cloning, partial-apply replay, failed batches, and cancellation during both checkpoint polling and an in-flight INSERT. --- go/binlog/binlog_entry.go | 5 +- go/binlog/cancellation_test.go | 115 ++++++++ go/binlog/gomysql_reader.go | 49 +++- go/binlog/transaction_marker_test.go | 74 +++++ go/binlog/transaction_test_helpers_test.go | 47 ++++ go/logic/applier.go | 11 +- go/logic/applier_test.go | 9 +- go/logic/binlog_checkpoint_test.go | 311 +++++++++++++++++++++ go/logic/checkpoint_cancellation_test.go | 191 +++++++++++++ go/logic/migrator.go | 93 ++++-- go/logic/streamer.go | 72 ++++- go/logic/streamer_test.go | 45 +++ go/logic/transaction_marker_test.go | 155 ++++++++++ 13 files changed, 1128 insertions(+), 49 deletions(-) create mode 100644 go/binlog/cancellation_test.go create mode 100644 go/binlog/transaction_marker_test.go create mode 100644 go/binlog/transaction_test_helpers_test.go create mode 100644 go/logic/binlog_checkpoint_test.go create mode 100644 go/logic/checkpoint_cancellation_test.go create mode 100644 go/logic/transaction_marker_test.go diff --git a/go/binlog/binlog_entry.go b/go/binlog/binlog_entry.go index 7620281d2..142e99cab 100644 --- a/go/binlog/binlog_entry.go +++ b/go/binlog/binlog_entry.go @@ -13,8 +13,9 @@ import ( // BinlogEntry describes an entry in the binary log type BinlogEntry struct { - Coordinates mysql.BinlogCoordinates - DmlEvent *BinlogDMLEvent + Coordinates mysql.BinlogCoordinates + DmlEvent *BinlogDMLEvent + TransactionComplete bool } // NewBinlogEntryAt creates an empty, ready to go BinlogEntry object diff --git a/go/binlog/cancellation_test.go b/go/binlog/cancellation_test.go new file mode 100644 index 000000000..c4c83a4e4 --- /dev/null +++ b/go/binlog/cancellation_test.go @@ -0,0 +1,115 @@ +package binlog + +import ( + "context" + "testing" + "time" + + "github.com/github/gh-ost/go/base" + "github.com/github/gh-ost/go/mysql" + "github.com/go-mysql-org/go-mysql/replication" + "github.com/stretchr/testify/require" +) + +func TestTransactionMarkerSendCanceled(t *testing.T) { + assertTransactionMarkerSendCanceled(t, nil) +} + +func assertTransactionMarkerSendCanceled(t *testing.T, wrapEvents func([]*replication.BinlogEvent) []*replication.BinlogEvent) { + t.Helper() + reader := transactionTestReader(false) + defer reader.migrationContext.CancelContext() + previous := mysql.NewFileBinlogCoordinates("mysql-bin.000001", 4) + reader.LastTrxCoords = previous.Clone() + row := &replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: replication.WRITE_ROWS_EVENTv2, LogPos: 200}, + Event: &replication.RowsEvent{ + Table: &replication.TableMapEvent{Schema: []byte("testdb"), Table: []byte("testing")}, + Rows: [][]interface{}{{1}}, + }, + } + commit := &replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: replication.XID_EVENT, LogPos: 220}, + Event: &replication.XIDEvent{}, + } + events := []*replication.BinlogEvent{row, commit} + if wrapEvents != nil { + events = wrapEvents(events) + } + for _, event := range events { + require.NoError(t, reader.binlogStreamer.AddEventToStreamer(event)) + } + entries := make(chan *BinlogEntry) + done := make(chan error, 1) + go func() { + err := reader.StreamEvents(func() bool { return false }, entries) + _ = base.SendWithContext(t.Context(), done, err) + }() + select { + case entry := <-entries: + require.NotNil(t, entry.DmlEvent) + case <-time.After(time.Second): + t.Fatal("row was not delivered") + } + // Wait until the commit position has been read, then stop consuming. + // The reader must propagate a canceled completion-marker send. + require.Eventually(t, func() bool { + return reader.GetCurrentBinlogCoordinates().(*mysql.FileBinlogCoordinates).LogPos == 220 + }, time.Second, time.Millisecond) + reader.migrationContext.CancelContext() + select { + case err := <-done: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(time.Second): + t.Fatal("completion marker send ignored cancellation") + } + require.True(t, reader.transactionHasRows, "failed marker delivery must retain pending state") + require.True(t, previous.Equals(reader.LastTrxCoords), "failed delivery must not advance reconnect coordinates") +} + +func TestRowsSendCanceled(t *testing.T) { + reader := transactionTestReader(false) + defer reader.migrationContext.CancelContext() + event := &replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: replication.WRITE_ROWS_EVENTv2}, + Event: &replication.RowsEvent{ + Table: &replication.TableMapEvent{Schema: []byte("testdb"), Table: []byte("testing")}, + Rows: [][]interface{}{{1}, {2}}, + }, + } + require.NoError(t, reader.binlogStreamer.AddEventToStreamer(event)) + entries := make(chan *BinlogEntry) + done := make(chan error, 1) + go func() { + err := reader.StreamEvents(func() bool { return false }, entries) + _ = base.SendWithContext(t.Context(), done, err) + }() + select { + case <-entries: + case <-time.After(time.Second): + t.Fatal("first row was not delivered") + } + reader.migrationContext.CancelContext() + select { + case err := <-done: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(time.Second): + t.Fatal("blocked row send ignored cancellation") + } +} + +func TestIdleReaderCanceled(t *testing.T) { + reader := transactionTestReader(false) + reader.migrationContext.CancelContext() + done := make(chan error, 1) + go func() { + err := reader.StreamEvents(func() bool { return false }, make(chan *BinlogEntry)) + _ = base.SendWithContext(t.Context(), done, err) + }() + select { + case err := <-done: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(time.Second): + t.Fatal("idle reader ignored cancellation") + } +} diff --git a/go/binlog/gomysql_reader.go b/go/binlog/gomysql_reader.go index e139726fd..96bdc39f8 100644 --- a/go/binlog/gomysql_reader.go +++ b/go/binlog/gomysql_reader.go @@ -15,8 +15,6 @@ import ( "time" - "context" - gomysql "github.com/go-mysql-org/go-mysql/mysql" "github.com/go-mysql-org/go-mysql/replication" ) @@ -47,8 +45,11 @@ type GoMySQLReader struct { currentCoordinates mysql.BinlogCoordinates currentCoordinatesMutex *sync.Mutex // LastTrxCoords are the coordinates of the last transaction completely read. - // If using the file coordinates it is binlog position of the transaction's XID event. + // They are for streamer reconnects, not durable applier checkpoints. + // For file coordinates, this is the end position of the transaction's XID event. LastTrxCoords mysql.BinlogCoordinates + + transactionHasRows bool } func NewGoMySQLReader(migrationContext *base.MigrationContext, rowsEventFilters ...RowsEventFilterFunc) *GoMySQLReader { @@ -147,15 +148,43 @@ func (gmr *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent // decides whether action is taken synchronously (meaning we wait before // next iteration) or asynchronously (we keep pushing more events) // In reality, reads will be synchronous - entriesChannel <- binlogEntry + if err := base.SendWithContext(gmr.migrationContext.GetContext(), entriesChannel, binlogEntry); err != nil { + return err + } + gmr.transactionHasRows = true } return nil } +func (gmr *GoMySQLReader) completeTransaction(event *replication.XIDEvent, entriesChannel chan<- *BinlogEntry) error { + var coords mysql.BinlogCoordinates + if gmr.migrationContext.UseGTIDs && event.GSet != nil { + coords = (&mysql.GTIDBinlogCoordinates{GTIDSet: event.GSet}).Clone() + } else { + // File positions use the current event's coordinates. For GTID + // events parsed without the syncer, fall back to the coordinates + // already updated by the preceding GTID event. + coords = gmr.GetCurrentBinlogCoordinates() + } + if gmr.transactionHasRows { + marker := &BinlogEntry{ + Coordinates: coords.Clone(), + TransactionComplete: true, + } + if err := base.SendWithContext(gmr.migrationContext.GetContext(), entriesChannel, marker); err != nil { + return err + } + gmr.transactionHasRows = false + } + // Reconnect only after all rows and their completion marker were delivered. + gmr.LastTrxCoords = coords + return nil +} + // StreamEvents func (gmr *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChannel chan<- *BinlogEntry) error { for !canStopStreaming() { - ev, err := gmr.binlogStreamer.GetEvent(context.Background()) + ev, err := gmr.binlogStreamer.GetEvent(gmr.migrationContext.GetContext()) if err != nil { return err } @@ -214,14 +243,8 @@ func (gmr *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChan gmr.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", coords.LogFile, int64(ev.Header.LogPos), event.NextLogName) gmr.currentCoordinatesMutex.Unlock() case *replication.XIDEvent: - if gmr.migrationContext.UseGTIDs { - // event.GSet is the full executed GTID set maintained by the - // syncer (MysqlGTIDSet or MariadbGTIDSet depending on flavor). - if event.GSet != nil { - gmr.LastTrxCoords = &mysql.GTIDBinlogCoordinates{GTIDSet: event.GSet} - } - } else { - gmr.LastTrxCoords = gmr.currentCoordinates.Clone() + if err := gmr.completeTransaction(event, entriesChannel); err != nil { + return err } case *replication.RowsEvent: if err := gmr.handleRowsEvent(ev, event, entriesChannel); err != nil { diff --git a/go/binlog/transaction_marker_test.go b/go/binlog/transaction_marker_test.go new file mode 100644 index 000000000..efc152a38 --- /dev/null +++ b/go/binlog/transaction_marker_test.go @@ -0,0 +1,74 @@ +package binlog + +import ( + "testing" + + "github.com/github/gh-ost/go/mysql" + "github.com/go-mysql-org/go-mysql/replication" + "github.com/stretchr/testify/require" +) + +func TestTransactionMarkerClonesGTIDSet(t *testing.T) { + for _, test := range []struct { + flavor, committed, next string + }{ + {mysql.MySQLFlavor, "00000000-0000-0000-0000-000000000001:1", "00000000-0000-0000-0000-000000000001:2"}, + {mysql.MariaDBFlavor, "0-1-1", "0-1-2"}, + } { + t.Run(test.flavor, func(t *testing.T) { + reader := transactionTestReader(true) + defer reader.migrationContext.CancelContext() + coords, err := mysql.NewGTIDBinlogCoordinates(test.flavor, test.committed) + require.NoError(t, err) + reader.currentCoordinates = coords.Clone() + rows := &replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: replication.WRITE_ROWS_EVENTv2}, + Event: &replication.RowsEvent{ + Table: &replication.TableMapEvent{Schema: []byte("testdb"), Table: []byte("testing")}, + Rows: [][]interface{}{{1}, {2}}, + }, + } + commit := &replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: replication.XID_EVENT}, + Event: &replication.XIDEvent{GSet: coords.GTIDSet}, + } + entries := transactionTestStream(t, reader, rows, commit) + require.Len(t, entries, 3) + require.True(t, entries[2].TransactionComplete) + require.NoError(t, coords.GTIDSet.Update(test.next)) + require.Equal(t, test.committed, reader.LastTrxCoords.String()) + require.Equal(t, test.committed, entries[2].Coordinates.String()) + }) + } +} + +func TestMultiRowEventDoesNotCompleteTransaction(t *testing.T) { + reader := transactionTestReader(false) + rows := &replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: replication.WRITE_ROWS_EVENTv2, LogPos: 200, EventSize: 100}, + Event: &replication.RowsEvent{ + Table: &replication.TableMapEvent{Schema: []byte("testdb"), Table: []byte("testing")}, + Rows: [][]interface{}{{1}, {2}, {3}}, + }, + } + // Stop before XID. All rows share the event end position, but none of + // them may independently acknowledge that position for checkpointing. + entries := transactionTestStream(t, reader, rows) + require.Len(t, entries, 3) + for _, entry := range entries { + require.Equal(t, int64(200), entry.Coordinates.(*mysql.FileBinlogCoordinates).LogPos) + require.False(t, entry.TransactionComplete) + } + require.Nil(t, reader.LastTrxCoords) + + commit := &replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: replication.XID_EVENT, LogPos: 220, EventSize: 20}, + Event: &replication.XIDEvent{}, + } + entries = transactionTestStream(t, reader, commit) + require.Len(t, entries, 1) + require.True(t, entries[0].TransactionComplete) + require.Nil(t, entries[0].DmlEvent) + require.Equal(t, int64(220), entries[0].Coordinates.(*mysql.FileBinlogCoordinates).LogPos) + require.Equal(t, reader.LastTrxCoords, entries[0].Coordinates) +} diff --git a/go/binlog/transaction_test_helpers_test.go b/go/binlog/transaction_test_helpers_test.go new file mode 100644 index 000000000..f916e26e9 --- /dev/null +++ b/go/binlog/transaction_test_helpers_test.go @@ -0,0 +1,47 @@ +package binlog + +import ( + "sync" + "testing" + + "github.com/github/gh-ost/go/base" + "github.com/github/gh-ost/go/mysql" + "github.com/go-mysql-org/go-mysql/replication" + "github.com/stretchr/testify/require" +) + +func transactionTestReader(useGTIDs bool) *GoMySQLReader { + ctx := base.NewMigrationContext() + ctx.UseGTIDs = useGTIDs + ctx.DatabaseName = "testdb" + ctx.OriginalTableName = "testing" + var coords mysql.BinlogCoordinates = &mysql.FileBinlogCoordinates{LogFile: "mysql-bin.000001", LogPos: 4} + if useGTIDs { + coords = &mysql.GTIDBinlogCoordinates{} + } + return &GoMySQLReader{ + migrationContext: ctx, + binlogStreamer: replication.NewBinlogStreamer(), + currentCoordinates: coords, + currentCoordinatesMutex: &sync.Mutex{}, + } +} + +func transactionTestStream(t *testing.T, reader *GoMySQLReader, events ...*replication.BinlogEvent) []*BinlogEntry { + t.Helper() + for _, event := range events { + require.NoError(t, reader.binlogStreamer.AddEventToStreamer(event)) + } + entries := make(chan *BinlogEntry, 32) + iterations := 0 + require.NoError(t, reader.StreamEvents(func() bool { + iterations++ + return iterations > len(events) + }, entries)) + close(entries) + var result []*BinlogEntry + for entry := range entries { + result = append(result, entry) + } + return result +} diff --git a/go/logic/applier.go b/go/logic/applier.go index d1c8d4716..34d481c52 100644 --- a/go/logic/applier.go +++ b/go/logic/applier.go @@ -75,8 +75,9 @@ type Applier struct { finishedMigrating int64 name string - CurrentCoordinatesMutex sync.Mutex - CurrentCoordinates mysql.BinlogCoordinates + CurrentCoordinatesMutex sync.Mutex + CurrentCoordinates mysql.BinlogCoordinates + AppliedTransactionCoordinates mysql.BinlogCoordinates LastIterationRangeMutex sync.Mutex LastIterationRangeMinValues *sql.ColumnValues @@ -882,8 +883,8 @@ func (apl *Applier) WriteChangelogState(value string) (string, error) { return apl.WriteAndLogChangelog("state", value) } -// WriteCheckpoints writes a checkpoint to the _ghk table. -func (apl *Applier) WriteCheckpoint(chk *Checkpoint) (int64, error) { +// WriteCheckpoint writes a checkpoint to the _ghk table. +func (apl *Applier) WriteCheckpoint(ctx context.Context, chk *Checkpoint) (int64, error) { var insertId int64 uniqueKeyArgs := sqlutils.Args(chk.IterationRangeMin.AbstractValues()...) uniqueKeyArgs = append(uniqueKeyArgs, chk.IterationRangeMax.AbstractValues()...) @@ -893,7 +894,7 @@ func (apl *Applier) WriteCheckpoint(chk *Checkpoint) (int64, error) { } args := sqlutils.Args(chk.LastTrxCoords.String(), chk.Iteration, chk.RowsCopied, chk.DMLApplied, chk.IsCutover) args = append(args, uniqueKeyArgs...) - res, err := apl.db.Exec(query, args...) + res, err := apl.db.ExecContext(ctx, query, args...) if err != nil { return insertId, err } diff --git a/go/logic/applier_test.go b/go/logic/applier_test.go index f1fa28bc8..79310fa7d 100644 --- a/go/logic/applier_test.go +++ b/go/logic/applier_test.go @@ -957,13 +957,20 @@ func (suite *ApplierTestSuite) TestWriteCheckpoint() { DMLApplied: 200000, IsCutover: true, } - id, err := applier.WriteCheckpoint(chk) + id, err := applier.WriteCheckpoint(ctx, chk) suite.Require().NoError(err) suite.Require().Equal(int64(1), id) + // A canceled write must leave the last durable checkpoint untouched. + canceledCtx, cancel := context.WithCancel(ctx) + cancel() + _, err = applier.WriteCheckpoint(canceledCtx, chk) + suite.Require().ErrorIs(err, context.Canceled) + gotChk, err := applier.ReadLastCheckpoint() suite.Require().NoError(err) + suite.Require().Equal(id, gotChk.Id) suite.Require().Equal(chk.Iteration, gotChk.Iteration) suite.Require().Equal(chk.LastTrxCoords.String(), gotChk.LastTrxCoords.String()) suite.Require().Equal(chk.IterationRangeMin.String(), gotChk.IterationRangeMin.String()) diff --git a/go/logic/binlog_checkpoint_test.go b/go/logic/binlog_checkpoint_test.go new file mode 100644 index 000000000..0d6030e50 --- /dev/null +++ b/go/logic/binlog_checkpoint_test.go @@ -0,0 +1,311 @@ +package logic + +import ( + "context" + gosql "database/sql" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/github/gh-ost/go/base" + "github.com/github/gh-ost/go/binlog" + "github.com/github/gh-ost/go/mysql" + "github.com/github/gh-ost/go/sql" + "github.com/stretchr/testify/require" + "github.com/testcontainers/testcontainers-go" + testmysql "github.com/testcontainers/testcontainers-go/modules/mysql" +) + +// queueCheckpointTestEvents uses the production reader, dispatcher and listeners, +// but leaves consumption of the apply queue under the test's control. +// +//nolint:contextcheck // Streamer APIs use MigrationContext; the test deadline closes the reader. +func queueCheckpointTestEvents(t *testing.T, ctx context.Context, migrationContext *base.MigrationContext, migrator *Migrator, start, end mysql.BinlogCoordinates) { + t.Helper() + reader := binlog.NewGoMySQLReader(migrationContext, func(database, table string) bool { + return database == testMysqlDatabase && table == testMysqlTableName + }) + defer reader.Close() + require.NoError(t, reader.ConnectBinlogStreamer(start)) + streamer := NewEventsStreamer(migrationContext) + streamer.binlogReader = reader + migrator.eventsStreamer = streamer + require.NoError(t, migrator.addDMLEventsListener()) + + delivered := make(chan struct{}, 1) + streamer.AddTransactionCompleteListener(func(entry *binlog.BinlogEntry) error { + if entry.Coordinates.Equals(end) { + delivered <- struct{}{} + } + return nil + }) + done := make(chan error, 1) + go func() { + done <- streamer.StreamEvents(func() bool { + return reader.LastTrxCoords != nil && reader.LastTrxCoords.Equals(end) + }) + }() + select { + case err := <-done: + require.NoError(t, err) + case <-ctx.Done(): + reader.Close() + <-done + t.Fatal("timed out reading transaction") + } + // The reader has stopped sending. Let the dispatcher drain and exit. + close(streamer.eventsChannel) + select { + case <-delivered: + case <-ctx.Done(): + t.Fatal("timed out dispatching transaction completion") + } +} + +//nolint:contextcheck // Migrator APIs use MigrationContext's internal context; SQL/checkpoint calls use the test deadline. +func TestBinlogCheckpointWaitsForTransactionIntegration(t *testing.T) { + if testing.Short() { + t.Skip("requires MySQL 8.0") + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + container, err := testmysql.Run(ctx, testMysqlContainerImage, + testmysql.WithDatabase(testMysqlDatabase), + testmysql.WithUsername(testMysqlUser), + testmysql.WithPassword(testMysqlPass), + testcontainers.WithCmdArgs("--server-id=1", "--gtid-mode=ON", "--enforce-gtid-consistency=ON"), + ) + require.NoError(t, err) + defer testcontainers.TerminateContainer(container) + dsn, err := container.ConnectionString(ctx) + require.NoError(t, err) + db, err := gosql.Open("mysql", dsn) + require.NoError(t, err) + defer db.Close() + conn, err := db.Conn(ctx) + require.NoError(t, err) + defer conn.Close() + for _, query := range []string{ + "SET SESSION time_zone='+00:00'", + "CREATE TABLE testing (id INT PRIMARY KEY, amount DECIMAL(5,2), created_at TIMESTAMP, padding TEXT)", + } { + _, err := conn.ExecContext(ctx, query) + require.NoError(t, err) + } + status := func() (*mysql.FileBinlogCoordinates, *mysql.GTIDBinlogCoordinates) { + var file, doDB, ignoreDB, gtid string + var pos int64 + require.NoError(t, conn.QueryRowContext(ctx, "SHOW MASTER STATUS").Scan(&file, &pos, &doDB, &ignoreDB, >id)) + coords, err := mysql.NewGTIDBinlogCoordinates(mysql.MySQLFlavor, gtid) + require.NoError(t, err) + return mysql.NewFileBinlogCoordinates(file, pos), coords + } + type transaction struct { + name string + startFile, endFile *mysql.FileBinlogCoordinates + startGTID, endGTID *mysql.GTIDBinlogCoordinates + } + var transactions []transaction + for _, compression := range []string{"OFF"} { + _, err := conn.ExecContext(ctx, "SET SESSION binlog_transaction_compression="+compression) + require.NoError(t, err) + startFile, startGTID := status() + tx, err := conn.BeginTx(ctx, nil) + require.NoError(t, err) + for _, query := range []string{ + "INSERT INTO testing VALUES (1, 123.45, '2023-11-14 22:13:20', REPEAT('x', 10000))", + "UPDATE testing SET amount=456.78 WHERE id=1", + "DELETE FROM testing WHERE id=1", + } { + _, err := tx.ExecContext(ctx, query) + require.NoError(t, err) + } + require.NoError(t, tx.Commit()) + endFile, endGTID := status() + transactions = append(transactions, transaction{compression, startFile, endFile, startGTID, endGTID}) + } + config, err := getTestConnectionConfig(ctx, container) + require.NoError(t, err) + + for _, transaction := range transactions { + for _, mode := range []struct { + useGTIDs bool + cancelAt string + }{ + {false, "none"}, {true, "none"}, + {false, "apply"}, {true, "apply"}, + {false, "insert"}, {true, "insert"}, + } { + useGTIDs := mode.useGTIDs + t.Run(fmt.Sprintf("compression=%s/GTID=%t/cancel=%s", transaction.name, useGTIDs, mode.cancelAt), func(t *testing.T) { + migrationContext := newTestMigrationContext() + defer migrationContext.CancelContext() + migrationContext.Checkpoint = true + migrationContext.UseGTIDs = useGTIDs + // Replaying an already-applied INSERT uses INSERT IGNORE. Duplicate + // warnings are expected on resume; use the default production policy. + migrationContext.PanicOnWarnings = false + migrationContext.ApplierConnectionConfig = config + migrationContext.InspectorConnectionConfig = config + require.NoError(t, migrationContext.SetConnectionConfig("innodb")) + migrationContext.SetDMLBatchSize(1) + columns := sql.NewColumnList([]string{"id", "amount", "created_at", "padding"}) + migrationContext.OriginalTableColumns = columns + migrationContext.SharedColumns = columns + migrationContext.MappedSharedColumns = columns + uniqueKeyColumns := sql.NewColumnList([]string{"id"}) + uniqueKeyColumns.GetColumn("id").MySQLType = "int" + migrationContext.UniqueKey = &sql.UniqueKey{Name: "PRIMARY", NameInGhostTable: "PRIMARY", Columns: *uniqueKeyColumns} + _, err := conn.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s LIKE %s", getTestGhostTableName(), getTestTableName())) + require.NoError(t, err) + defer func() { + _, err := conn.ExecContext(ctx, fmt.Sprintf("DROP TABLE IF EXISTS %s", getTestGhostTableName())) + require.NoError(t, err) + }() + + migrator := NewMigrator(migrationContext, "test") + migrator.applier = NewApplier(migrationContext) + require.NoError(t, migrator.applier.InitDBConnections()) + defer migrator.applier.Teardown() + migrator.applier.LastIterationRangeMinValues = sql.ToColumnValues([]interface{}{int64(1)}) + migrator.applier.LastIterationRangeMaxValues = sql.ToColumnValues([]interface{}{int64(1)}) + require.NoError(t, migrator.applier.prepareQueries()) + require.NoError(t, migrator.applier.CreateCheckpointTable()) + defer func() { require.NoError(t, migrator.applier.DropCheckpointTable()) }() + start, end := transaction.startFile.Clone(), transaction.endFile.Clone() + if useGTIDs { + start, end = transaction.startGTID.Clone(), transaction.endGTID.Clone() + } + seed := &Checkpoint{ + LastTrxCoords: start.Clone(), + IterationRangeMin: migrator.applier.LastIterationRangeMinValues.Clone(), + IterationRangeMax: migrator.applier.LastIterationRangeMaxValues.Clone(), + } + seedID, err := migrator.applier.WriteCheckpoint(ctx, seed) + require.NoError(t, err) + migrator.applier.AppliedTransactionCoordinates = start.Clone() + queueCheckpointTestEvents(t, ctx, migrationContext, migrator, start, end) + require.Len(t, migrator.applyEventsQueue, 4) + first := <-migrator.applyEventsQueue + require.NoError(t, migrator.onApplyEventStruct(first)) + require.Len(t, migrator.applyEventsQueue, 3, "two DMLs and the completion marker remain pending") + + assertCheckpointBlocked := func() { + checkpointCtx, cancelCheckpoint := context.WithTimeout(ctx, 100*time.Millisecond) + defer cancelCheckpoint() + checkpoint, err := migrator.Checkpoint(checkpointCtx) + require.ErrorIs(t, err, context.DeadlineExceeded, "premature checkpoint: %+v", checkpoint) + stored, err := migrator.applier.ReadLastCheckpoint() + require.NoError(t, err) + require.Equal(t, seedID, stored.Id, "failed checkpoint must not replace the previous durable checkpoint") + require.True(t, stored.LastTrxCoords.Equals(start)) + } + assertCheckpointBlocked() + switch mode.cancelAt { + case "apply": + assertCanceledTransactionCheckpoint(t, ctx, migrationContext, migrator, seedID) + return + case "insert": + for len(migrator.applyEventsQueue) > 0 { + require.NoError(t, migrator.onApplyEventStruct(<-migrator.applyEventsQueue)) + } + require.True(t, migrator.applier.AppliedTransactionCoordinates.Equals(end)) + assertCanceledCheckpointInsert(t, ctx, migrationContext, migrator, seedID) + return + } + + // Simulate a crash after the first applied row: discard volatile + // queue contents, then restart from the last persisted checkpoint. + for len(migrator.applyEventsQueue) > 0 { + <-migrator.applyEventsQueue + } + stored, err := migrator.applier.ReadLastCheckpoint() + require.NoError(t, err) + queueCheckpointTestEvents(t, ctx, migrationContext, migrator, stored.LastTrxCoords, end) + require.Len(t, migrator.applyEventsQueue, 4) + var transactionRows []*applyEventStruct + for len(migrator.applyEventsQueue) > 1 { + row := <-migrator.applyEventsQueue + transactionRows = append(transactionRows, row) + require.NoError(t, migrator.onApplyEventStruct(row)) + } + // Even the final row cannot independently acknowledge its commit. + require.True(t, migrator.applier.AppliedTransactionCoordinates.Equals(start)) + if !useGTIDs && transaction.name == "OFF" { + // Sample a real row-event position P between the previous commit A + // and this transaction's XID B. A must block; B must be persisted + // rather than P once its marker is acknowledged. + sampled := first.coords.Clone() + require.True(t, start.SmallerThan(sampled)) + require.True(t, sampled.SmallerThan(end)) + reader := binlog.NewGoMySQLReader(migrationContext, nil) + defer reader.Close() + require.NoError(t, reader.ConnectBinlogStreamer(sampled)) + migrator.eventsStreamer.binlogReader = reader + } + assertCheckpointBlocked() + marker := <-migrator.applyEventsQueue + require.True(t, marker.transactionComplete) + require.NoError(t, migrator.onApplyEventStruct(marker)) + checkpoint, err := migrator.Checkpoint(ctx) + require.NoError(t, err) + require.True(t, checkpoint.LastTrxCoords.Equals(end)) + stored, err = migrator.applier.ReadLastCheckpoint() + require.NoError(t, err) + require.True(t, stored.LastTrxCoords.Equals(end)) + var remaining int + require.NoError(t, conn.QueryRowContext(ctx, fmt.Sprintf("SELECT COUNT(*) FROM %s", getTestGhostTableName())).Scan(&remaining)) + require.Zero(t, remaining, "resuming after the partial apply must not skip the DELETE") + + // Resume again from the completed checkpoint and apply later changes. + _, err = conn.ExecContext(ctx, "INSERT INTO testing VALUES (100, 12.34, '2023-11-14 22:13:20', 'after checkpoint') ON DUPLICATE KEY UPDATE amount=amount+1") + require.NoError(t, err) + resumeFile, resumeGTID := status() + resumeEnd := mysql.BinlogCoordinates(resumeFile) + if useGTIDs { + resumeEnd = resumeGTID + } + queueCheckpointTestEvents(t, ctx, migrationContext, migrator, stored.LastTrxCoords, resumeEnd) + for len(migrator.applyEventsQueue) > 0 { + require.NoError(t, migrator.onApplyEventStruct(<-migrator.applyEventsQueue)) + } + require.True(t, migrator.applier.AppliedTransactionCoordinates.Equals(resumeEnd)) + var sourceRows, ghostRows string + query := "SELECT GROUP_CONCAT(CONCAT(id, ':', amount) ORDER BY id) FROM %s" + require.NoError(t, conn.QueryRowContext(ctx, fmt.Sprintf(query, getTestTableName())).Scan(&sourceRows)) + require.NoError(t, conn.QueryRowContext(ctx, fmt.Sprintf(query, getTestGhostTableName())).Scan(&ghostRows)) + require.Equal(t, sourceRows, ghostRows) + + // Batch across two source-transaction markers, ending inside the + // second transaction. Only the first may be acknowledged yet. + migrationContext.SetDMLBatchSize(4) + migrator.applier.AppliedTransactionCoordinates = start.Clone() + nextMarker := &applyEventStruct{coords: resumeEnd.Clone(), transactionComplete: true} + for _, event := range []*applyEventStruct{transactionRows[1], transactionRows[2], marker, transactionRows[0], transactionRows[1], transactionRows[2], nextMarker} { + migrator.applyEventsQueue <- event + } + appliedBefore := atomic.LoadInt64(&migrationContext.TotalDMLEventsApplied) + require.NoError(t, migrator.onApplyEventStruct(transactionRows[0])) + require.Equal(t, appliedBefore+4, atomic.LoadInt64(&migrationContext.TotalDMLEventsApplied)) + require.Len(t, migrator.applyEventsQueue, 3) + require.True(t, migrator.applier.AppliedTransactionCoordinates.Equals(end)) + require.NoError(t, migrator.onApplyEventStruct(<-migrator.applyEventsQueue)) + require.Empty(t, migrator.applyEventsQueue) + require.True(t, migrator.applier.AppliedTransactionCoordinates.Equals(resumeEnd)) + + // A marker collected while assembling a batch cannot acknowledge + // anything if the destination SQL transaction fails. + _, err = conn.ExecContext(ctx, fmt.Sprintf("DROP TABLE %s", getTestGhostTableName())) + require.NoError(t, err) + migrationContext.SetDefaultNumRetries(1) + migrationContext.PanicAbort = make(chan error, 1) + migrator.applier.AppliedTransactionCoordinates = end.Clone() + migrator.applyEventsQueue <- nextMarker + migrator.applyEventsQueue <- transactionRows[1] + require.Error(t, migrator.onApplyEventStruct(first)) + require.True(t, migrator.applier.AppliedTransactionCoordinates.Equals(end)) + }) + } + } +} diff --git a/go/logic/checkpoint_cancellation_test.go b/go/logic/checkpoint_cancellation_test.go new file mode 100644 index 000000000..b5d719c8e --- /dev/null +++ b/go/logic/checkpoint_cancellation_test.go @@ -0,0 +1,191 @@ +package logic + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/github/gh-ost/go/base" + "github.com/github/gh-ost/go/binlog" + "github.com/github/gh-ost/go/metrics" + "github.com/github/gh-ost/go/sql" + drivermysql "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Signal when Checkpoint is waiting for its sampled position to be applied, +// without relying on sleeps to arrange the cancellation interleaving. +type checkpointWaitMetrics struct { + metrics.Emitter + waiting chan struct{} + once sync.Once +} + +func (m *checkpointWaitMetrics) Histogram(name string, _ float64, tags ...string) { + if name == "sleep.duration_milliseconds" && len(tags) == 1 && tags[0] == "stage:replica_wait" { + m.once.Do(func() { close(m.waiting) }) + } +} + +// The caller applied the INSERT and seeded an older durable checkpoint. The +// UPDATE, DELETE and completion marker are still pending. Replay their delivery +// under cancellation to exercise the failed-enqueue/marker interleaving. +// +//nolint:contextcheck // Migrator and dispatcher APIs use MigrationContext's internal context. +func assertCanceledTransactionCheckpoint(t *testing.T, ctx context.Context, migrationContext *base.MigrationContext, migrator *Migrator, seedID int64) { + t.Helper() + var pending []*applyEventStruct + for len(migrator.applyEventsQueue) > 0 { + pending = append(pending, <-migrator.applyEventsQueue) + } + require.Len(t, pending, 3) + marker := pending[2] + require.True(t, marker.transactionComplete) + previous := migrator.applier.AppliedTransactionCoordinates.Clone() + + wait := &checkpointWaitMetrics{Emitter: metrics.Noop, waiting: make(chan struct{})} + migrationContext.Metrics = wait + checkpointCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := migrator.Checkpoint(checkpointCtx) + _ = base.SendWithContext(ctx, done, err) + }() + select { + case <-wait.waiting: + case <-checkpointCtx.Done(): + t.Fatal("checkpoint never waited for the pending transaction") + } + + migrationContext.CancelContext() + for _, event := range pending[:2] { + err := migrator.eventsStreamer.notifyListeners(&binlog.BinlogEntry{Coordinates: event.coords, DmlEvent: event.dmlEvent}) + require.ErrorIs(t, err, context.Canceled) + } + require.ErrorIs(t, migrator.eventsStreamer.notifyListeners(&binlog.BinlogEntry{Coordinates: marker.coords, TransactionComplete: true}), context.Canceled) + require.Empty(t, migrator.applyEventsQueue) + // Also reject a marker that was already dequeued before cancellation, + // e.g. by a worker that passed its loop's abort check before throttling. + require.ErrorIs(t, migrator.onApplyEventStruct(marker), context.Canceled) + require.True(t, previous.Equals(migrator.applier.AppliedTransactionCoordinates)) + select { + case err := <-done: + require.ErrorIs(t, err, context.Canceled) + case <-checkpointCtx.Done(): + t.Fatal("migration cancellation did not interrupt the pending checkpoint") + } + require.NoError(t, checkpointCtx.Err(), "checkpoint's caller has not canceled") + _, err := migrator.Checkpoint(checkpointCtx) + require.ErrorIs(t, err, context.Canceled, "new checkpoints must also reject an aborted migration") + stored, err := migrator.applier.ReadLastCheckpoint() + require.NoError(t, err) + require.Equal(t, seedID, stored.Id) + require.True(t, stored.LastTrxCoords.Equals(previous)) + var remaining int + require.NoError(t, migrator.applier.db.QueryRowContext(ctx, fmt.Sprintf("SELECT COUNT(*) FROM %s", getTestGhostTableName())).Scan(&remaining)) + require.Equal(t, 1, remaining, "the unapplied DELETE must still be replayed from the safe checkpoint") +} + +// Block the production checkpoint INSERT with a real MySQL table lock. Unlike +// cancellation during polling, this requires the migration cancellation bridge +// to reach an already-running ExecContext with an independent caller context. +func assertCanceledCheckpointInsert(t *testing.T, ctx context.Context, migrationContext *base.MigrationContext, migrator *Migrator, seedID int64) { + t.Helper() + previous, err := migrator.applier.ReadLastCheckpoint() + require.NoError(t, err) + require.Equal(t, seedID, previous.Id) + checkpointTable := fmt.Sprintf("%s.%s", sql.EscapeName(migrationContext.DatabaseName), sql.EscapeName(migrationContext.GetCheckpointTableName())) + blocker, err := migrator.applier.db.Conn(ctx) + require.NoError(t, err) + defer blocker.Close() + _, err = blocker.ExecContext(ctx, "LOCK TABLES "+checkpointTable+" READ") + require.NoError(t, err) + + checkpointCtx, cancelCheckpoint := context.WithTimeout(ctx, 30*time.Second) + var checkpointConnectionID int64 + done := make(chan error, 1) + workerDone := make(chan struct{}) + go func() { + defer close(workerDone) + _, err := migrator.Checkpoint(checkpointCtx) + _ = base.SendWithContext(ctx, done, err) + }() + defer func() { + // Release both waits on assertion failures, including mutation tests + // that remove cancellation propagation into the SQL operation. + cancelCheckpoint() + // Driver cancellation closes the client connection, but MySQL may + // retain a lock-blocked statement. Kill our test connection before + // unlocking so that cleanup cannot accidentally execute its INSERT. + if checkpointConnectionID != 0 { + _, err := migrator.applier.db.ExecContext(ctx, fmt.Sprintf("KILL CONNECTION %d", checkpointConnectionID)) + var mysqlErr *drivermysql.MySQLError + if !errors.As(err, &mysqlErr) || mysqlErr.Number != 1094 { // ER_NO_SUCH_THREAD: already disconnected + assert.NoError(t, err) + } + } + _, err := blocker.ExecContext(ctx, "UNLOCK TABLES") + assert.NoError(t, err) + select { + case <-workerDone: + case <-ctx.Done(): + t.Error("checkpoint worker did not exit during cleanup") + } + }() + + require.EventuallyWithT(t, func(collect *assert.CollectT) { + // Observing the server-side lock wait proves that Checkpoint passed + // its abort/polling checks and the INSERT is actually executing. + err := migrator.applier.db.QueryRowContext(ctx, ` + SELECT ID FROM information_schema.PROCESSLIST + WHERE DB = ? AND INFO LIKE '%insert /* gh-ost */%' + AND LOCATE(?, INFO) > 0 AND STATE = 'Waiting for table metadata lock'`, + migrationContext.DatabaseName, checkpointTable).Scan(&checkpointConnectionID) + assert.NoError(collect, err) + }, 5*time.Second, 10*time.Millisecond, "checkpoint INSERT never entered the table-lock wait") + select { + case err := <-done: + t.Fatalf("checkpoint returned before cancellation: %v", err) + default: + } + + migrationContext.CancelContext() + select { + case err := <-done: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(5 * time.Second): + t.Fatal("migration cancellation did not interrupt the blocked checkpoint INSERT") + } + require.NoError(t, checkpointCtx.Err(), "checkpoint's caller must remain live") + + // READ permits this SELECT while still preventing the pending INSERT. + // Check before cleanup: neither the lock nor the caller has been canceled + // to make the checkpoint call return. + stored, err := migrator.applier.ReadLastCheckpoint() + require.NoError(t, err) + require.Equal(t, previous.Id, stored.Id) + require.True(t, previous.LastTrxCoords.Equals(stored.LastTrxCoords)) +} + +func TestCheckpointLoopStopsOnCancellation(t *testing.T) { + migrationContext := newTestMigrationContext() + defer migrationContext.CancelContext() + migrationContext.CheckpointIntervalSeconds = 3600 + migrator := NewMigrator(migrationContext, "test") + done := make(chan struct{}) + go func() { + defer close(done) + migrator.checkpointLoop() + }() + migrationContext.CancelContext() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("checkpoint loop waited for its ticker after cancellation") + } +} diff --git a/go/logic/migrator.go b/go/logic/migrator.go index f2f6b3f20..5c72bdb98 100644 --- a/go/logic/migrator.go +++ b/go/logic/migrator.go @@ -51,9 +51,10 @@ type lockProcessedStruct struct { } type applyEventStruct struct { - writeFunc *tableWriteFunc - dmlEvent *binlog.BinlogDMLEvent - coords mysql.BinlogCoordinates + writeFunc *tableWriteFunc + dmlEvent *binlog.BinlogDMLEvent + coords mysql.BinlogCoordinates + transactionComplete bool } func newApplyEventStructByFunc(writeFunc *tableWriteFunc) *applyEventStruct { @@ -1567,6 +1568,9 @@ func (mgtr *Migrator) initiateStreaming() error { // addDMLEventsListener begins listening for binlog events on the original table, // and creates & enqueues a write task per such event. func (mgtr *Migrator) addDMLEventsListener() error { + // Register before the DML listener so every subsequently enqueued row + // has a completion marker following it through the same FIFO queues. + mgtr.eventsStreamer.AddTransactionCompleteListener(mgtr.onTransactionComplete) err := mgtr.eventsStreamer.AddListener( false, mgtr.migrationContext.DatabaseName, @@ -1580,6 +1584,14 @@ func (mgtr *Migrator) addDMLEventsListener() error { return err } +// onTransactionComplete enqueues an acknowledgement, not another SQL write. +// The applier may batch across these markers, but cannot acknowledge them +// until the batch containing their preceding DML has successfully committed. +func (mgtr *Migrator) onTransactionComplete(entry *binlog.BinlogEntry) error { + marker := &applyEventStruct{coords: entry.Coordinates.Clone(), transactionComplete: true} + return base.SendWithContext(mgtr.migrationContext.GetContext(), mgtr.applyEventsQueue, marker) +} + // initiateThrottler kicks in the throttling collection and the throttling checks. func (mgtr *Migrator) initiateThrottler() { mgtr.throttler = NewThrottler(mgtr.migrationContext, mgtr.applier, mgtr.inspector, mgtr.appVersion) @@ -1754,7 +1766,17 @@ func (mgtr *Migrator) iterateChunks() error { } func (mgtr *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error { + // Cancellation may happen after the worker's loop check, while throttled. + // A standalone marker must not bypass the abort checks used by SQL tasks. + if err := mgtr.checkAbort(); err != nil { + return err + } handleNonDMLEventStruct := func(eventStruct *applyEventStruct) error { + if eventStruct.transactionComplete { + mgtr.applier.CurrentCoordinatesMutex.Lock() + mgtr.applier.AppliedTransactionCoordinates = eventStruct.coords.Clone() + mgtr.applier.CurrentCoordinatesMutex.Unlock() + } if eventStruct.writeFunc != nil { if err := mgtr.retryOperation(*eventStruct.writeFunc); err != nil { return mgtr.migrationContext.Log.Errore(err) @@ -1769,16 +1791,19 @@ func (mgtr *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error { dmlEvents := [](*binlog.BinlogDMLEvent){} dmlEvents = append(dmlEvents, eventStruct.dmlEvent) var nonDmlStructToApply *applyEventStruct + var completedTransaction *applyEventStruct availableEvents := len(mgtr.applyEventsQueue) batchSize := int(atomic.LoadInt64(&mgtr.migrationContext.DMLBatchSize)) - if availableEvents > batchSize-1 { - // The "- 1" is because we already consumed one event: the original event that led to this function getting called. - // So, if DMLBatchSize==1 we wish to not process any further events - availableEvents = batchSize - 1 - } - for i := 0; i < availableEvents; i++ { + for i := 0; i < availableEvents && len(dmlEvents) < batchSize; i++ { additionalStruct := <-mgtr.applyEventsQueue + if additionalStruct.transactionComplete { + // Markers need no SQL and don't count toward DMLBatchSize. + // Acknowledge only after this entire batch succeeds, never while + // merely collecting rows from the queue. + completedTransaction = additionalStruct + continue + } if additionalStruct.dmlEvent == nil { // Not a DML. We don't group this, and we don't batch any further nonDmlStructToApply = additionalStruct @@ -1796,6 +1821,9 @@ func (mgtr *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error { // update applier coordinates mgtr.applier.CurrentCoordinatesMutex.Lock() mgtr.applier.CurrentCoordinates = eventStruct.coords + if completedTransaction != nil { + mgtr.applier.AppliedTransactionCoordinates = completedTransaction.coords.Clone() + } mgtr.applier.CurrentCoordinatesMutex.Unlock() if nonDmlStructToApply != nil { @@ -1810,9 +1838,20 @@ func (mgtr *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error { } // Checkpoint attempts to write a checkpoint of the Migrator's current state. -// It gets the binlog coordinates of the last received trx and waits until the -// applier reaches that trx. At that point it's safe to resume from these coordinates. +// It samples reader progress and waits for an applied transaction boundary +// covering that progress. Row/batch progress alone is not a safe resume point. func (mgtr *Migrator) Checkpoint(ctx context.Context) (*Checkpoint, error) { + if err := mgtr.checkAbort(); err != nil { + return nil, err + } + // Observe both the caller's deadline and migration cancellation, including + // while executing the checkpoint INSERT rather than only while polling. + ctx, cancel := context.WithCancel(ctx) + //nolint:contextcheck // Merge migration cancellation with the caller-derived checkpoint context. + stopCancel := context.AfterFunc(mgtr.migrationContext.GetContext(), cancel) + defer stopCancel() + defer cancel() + coords := mgtr.eventsStreamer.GetCurrentBinlogCoordinates() mgtr.applier.LastIterationRangeMutex.Lock() if mgtr.applier.LastIterationRangeMaxValues == nil || mgtr.applier.LastIterationRangeMinValues == nil { @@ -1823,19 +1862,25 @@ func (mgtr *Migrator) Checkpoint(ctx context.Context) (*Checkpoint, error) { Iteration: mgtr.migrationContext.GetIteration(), IterationRangeMin: mgtr.applier.LastIterationRangeMinValues.Clone(), IterationRangeMax: mgtr.applier.LastIterationRangeMaxValues.Clone(), - LastTrxCoords: coords, RowsCopied: atomic.LoadInt64(&mgtr.migrationContext.TotalRowsCopied), DMLApplied: atomic.LoadInt64(&mgtr.migrationContext.TotalDMLEventsApplied), } mgtr.applier.LastIterationRangeMutex.Unlock() for { + if err := mgtr.checkAbort(); err != nil { + return nil, err + } if err := ctx.Err(); err != nil { return nil, err } mgtr.applier.CurrentCoordinatesMutex.Lock() - if coords.SmallerThanOrEquals(mgtr.applier.CurrentCoordinates) { - id, err := mgtr.applier.WriteCheckpoint(chk) + applied := mgtr.applier.AppliedTransactionCoordinates + if applied != nil && coords.SmallerThanOrEquals(applied) { + // Persist the acknowledged commit boundary, not the sampled reader + // position, which may be inside a row event or transaction. + chk.LastTrxCoords = applied.Clone() + id, err := mgtr.applier.WriteCheckpoint(ctx, chk) chk.Id = id mgtr.applier.CurrentCoordinatesMutex.Unlock() return chk, err @@ -1843,7 +1888,11 @@ func (mgtr *Migrator) Checkpoint(ctx context.Context) (*Checkpoint, error) { mgtr.applier.CurrentCoordinatesMutex.Unlock() sleepDuration := 500 * time.Millisecond metrics.RecordSleep(mgtr.migrationContext.Metrics, "replica_wait", sleepDuration) - time.Sleep(sleepDuration) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(sleepDuration): + } } } @@ -1871,7 +1920,7 @@ func (mgtr *Migrator) CheckpointAfterCutOver() (*Checkpoint, error) { } mgtr.applier.LastIterationRangeMutex.Unlock() - id, err := mgtr.applier.WriteCheckpoint(chk) + id, err := mgtr.applier.WriteCheckpoint(mgtr.migrationContext.GetContext(), chk) chk.Id = id return chk, err } @@ -1883,7 +1932,15 @@ func (mgtr *Migrator) checkpointLoop() { } checkpointInterval := time.Duration(mgtr.migrationContext.CheckpointIntervalSeconds) * time.Second ticker := time.NewTicker(checkpointInterval) - for t := range ticker.C { + defer ticker.Stop() + migrationCtx := mgtr.migrationContext.GetContext() + for { + var t time.Time + select { + case <-migrationCtx.Done(): + return + case t = <-ticker.C: + } if atomic.LoadInt64(&mgtr.finishedMigrating) > 0 || atomic.LoadInt64(&mgtr.migrationContext.CutOverCompleteFlag) > 0 { return } @@ -1891,7 +1948,7 @@ func (mgtr *Migrator) checkpointLoop() { continue } mgtr.migrationContext.Log.Infof("starting checkpoint at %+v", t) - ctx, cancel := context.WithTimeout(context.Background(), checkpointTimeout) + ctx, cancel := context.WithTimeout(migrationCtx, checkpointTimeout) chk, err := mgtr.Checkpoint(ctx) if err != nil { if errors.Is(err, context.DeadlineExceeded) { diff --git a/go/logic/streamer.go b/go/logic/streamer.go index 7d3d00120..355c882be 100644 --- a/go/logic/streamer.go +++ b/go/logic/streamer.go @@ -40,6 +40,7 @@ type EventsStreamer struct { migrationContext *base.MigrationContext initialBinlogCoordinates mysql.BinlogCoordinates listeners [](*BinlogEventListener) + transactionListeners []func(*binlog.BinlogEntry) error listenersMutex *sync.Mutex eventsChannel chan *binlog.BinlogEntry binlogReader *binlog.GoMySQLReader @@ -80,6 +81,14 @@ func (es *EventsStreamer) AddListener( return nil } +// AddTransactionCompleteListener registers a synchronous commit listener. +// Consumers must also use synchronous DML listeners to preserve queue order. +func (es *EventsStreamer) AddTransactionCompleteListener(listener func(*binlog.BinlogEntry) error) { + es.listenersMutex.Lock() + defer es.listenersMutex.Unlock() + es.transactionListeners = append(es.transactionListeners, listener) +} + // shouldDecodeRowsEvent returns true when at least one listener is registered for // the table on which the rows event operates. This is used by the binlog parser // after it decodes the row event header/table map, but before it decodes row data. @@ -99,11 +108,34 @@ func (es *EventsStreamer) shouldDecodeRowsEvent(databaseName, tableName string) return false } -// notifyListeners will notify relevant listeners with given DML event. Only -// listeners registered for changes on the table on which the DML operates are notified. -func (es *EventsStreamer) notifyListeners(binlogEntry *binlog.BinlogEntry) { +// notifyListeners dispatches table-filtered DML or a transaction completion +// marker. Synchronous listeners run on the same dispatcher to preserve order. +// Failed delivery is fatal: continuing could acknowledge a transaction whose +// preceding DML never reached the apply queue. +func (es *EventsStreamer) notifyListeners(binlogEntry *binlog.BinlogEntry) (err error) { es.listenersMutex.Lock() defer es.listenersMutex.Unlock() + defer func() { + if err != nil { + es.migrationContext.SetAbortError(err) + es.migrationContext.CancelContext() + } + }() + if err := es.migrationContext.GetAbortError(); err != nil { + return err + } + if err := es.migrationContext.GetContext().Err(); err != nil { + return err + } + + if binlogEntry.TransactionComplete { + for _, listener := range es.transactionListeners { + if err := listener(binlogEntry); err != nil { + return err + } + } + return nil + } for _, listener := range es.listeners { listener := listener @@ -117,10 +149,11 @@ func (es *EventsStreamer) notifyListeners(binlogEntry *binlog.BinlogEntry) { go func() { listener.onDmlEvent(binlogEntry) }() - } else { - listener.onDmlEvent(binlogEntry) + } else if err := listener.onDmlEvent(binlogEntry); err != nil { + return err } } + return nil } func (es *EventsStreamer) InitDBConnections() (err error) { @@ -216,19 +249,32 @@ func (es *EventsStreamer) readCurrentMariaDBGTIDCoordinates() error { // StreamEvents will begin streaming events. It will be blocking, so should be // executed by a goroutine func (es *EventsStreamer) StreamEvents(canStopStreaming func() bool) error { + ctx := es.migrationContext.GetContext() go func() { - for binlogEntry := range es.eventsChannel { - if binlogEntry.DmlEvent != nil { - es.notifyListeners(binlogEntry) + for { + select { + case <-ctx.Done(): + return + case binlogEntry, ok := <-es.eventsChannel: + if !ok { + return + } + if binlogEntry.DmlEvent != nil || binlogEntry.TransactionComplete { + if err := es.notifyListeners(binlogEntry); err != nil { + return + } + } } } }() // The next should block and execute forever, unless there's a serious error. var successiveFailures int var reconnectCoords mysql.BinlogCoordinates - ctx := es.migrationContext.GetContext() for { - // Check for context cancellation each iteration + // Preserve a listener's failure rather than hiding it as cancellation. + if err := es.migrationContext.GetAbortError(); err != nil { + return err + } if err := ctx.Err(); err != nil { return err } @@ -239,6 +285,12 @@ func (es *EventsStreamer) StreamEvents(canStopStreaming func() bool) error { // of the last trx that was read completely from the streamer. // Since row event application is idempotent, it's OK if we reapply some events. if err := es.binlogReader.StreamEvents(canStopStreaming, es.eventsChannel); err != nil { + if abortErr := es.migrationContext.GetAbortError(); abortErr != nil { + return abortErr + } + if ctx.Err() != nil { + return ctx.Err() + } if canStopStreaming() { return nil } diff --git a/go/logic/streamer_test.go b/go/logic/streamer_test.go index 5b28c47c2..4871e2365 100644 --- a/go/logic/streamer_test.go +++ b/go/logic/streamer_test.go @@ -3,10 +3,12 @@ package logic import ( "context" gosql "database/sql" + "errors" "fmt" "testing" "time" + "github.com/github/gh-ost/go/base" "github.com/github/gh-ost/go/binlog" "github.com/stretchr/testify/suite" "github.com/testcontainers/testcontainers-go" @@ -140,6 +142,49 @@ func (suite *EventsStreamerTestSuite) TestStreamEvents() { suite.Require().Len(dmlEvents, 3) } +func (suite *EventsStreamerTestSuite) TestStreamEventsStopsOnListenerError() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err := suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY)", getTestTableName())) + suite.Require().NoError(err) + config, err := getTestConnectionConfig(ctx, suite.mysqlContainer) + suite.Require().NoError(err) + migrationContext := newTestMigrationContext() + defer migrationContext.CancelContext() + migrationContext.ApplierConnectionConfig = config + migrationContext.InspectorConnectionConfig = config + suite.Require().NoError(migrationContext.SetConnectionConfig("innodb")) + streamer := NewEventsStreamer(migrationContext) + suite.Require().NoError(streamer.InitDBConnections()) + defer streamer.Close() + defer streamer.Teardown() + failure := errors.New("DML enqueue failed") + suite.Require().NoError(streamer.AddListener(false, testMysqlDatabase, testMysqlTableName, func(*binlog.BinlogEntry) error { + return failure + })) + streamer.AddTransactionCompleteListener(func(*binlog.BinlogEntry) error { + suite.T().Error("marker delivered after failed DML") + return nil + }) + done := make(chan error, 1) + go func() { + //nolint:contextcheck // StreamEvents uses the migration context, canceled by listener failure. + err := streamer.StreamEvents(func() bool { return false }) + _ = base.SendWithContext(ctx, done, err) + }() + // Multiple rows also exercise cancellation of reader sends after the + // dispatcher exits, rather than only cancellation of GetEvent. + _, err = suite.db.ExecContext(ctx, fmt.Sprintf("INSERT INTO %s VALUES (1),(2),(3),(4),(5)", getTestTableName())) + suite.Require().NoError(err) + select { + case err := <-done: + suite.Require().ErrorIs(err, failure) + case <-ctx.Done(): + suite.T().Fatal("streamer did not stop after listener failure") + } + suite.Require().ErrorIs(migrationContext.GetAbortError(), failure) +} + func (suite *EventsStreamerTestSuite) TestStreamEventsAutomaticallyReconnects() { ctx := context.Background() _, err := suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, name VARCHAR(255))", getTestTableName())) diff --git a/go/logic/transaction_marker_test.go b/go/logic/transaction_marker_test.go new file mode 100644 index 000000000..519156056 --- /dev/null +++ b/go/logic/transaction_marker_test.go @@ -0,0 +1,155 @@ +package logic + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/github/gh-ost/go/binlog" + "github.com/github/gh-ost/go/mysql" + "github.com/github/gh-ost/go/sql" + "github.com/stretchr/testify/require" +) + +func TestTransactionMarkersFollowSynchronousDMLListeners(t *testing.T) { + ctx := newTestMigrationContext() + streamer := NewEventsStreamer(ctx) + migrator := NewMigrator(ctx, "test") + migrator.eventsStreamer = streamer + migrator.applier = NewApplier(ctx) + require.NoError(t, migrator.addDMLEventsListener()) + coords := mysql.NewFileBinlogCoordinates("mysql-bin.000001", 1234) + for i := 0; i < 3; i++ { + require.NoError(t, streamer.notifyListeners(&binlog.BinlogEntry{ + Coordinates: coords, + DmlEvent: binlog.NewBinlogDMLEvent(ctx.DatabaseName, ctx.OriginalTableName, binlog.InsertDML), + })) + } + require.NoError(t, streamer.notifyListeners(&binlog.BinlogEntry{Coordinates: coords, TransactionComplete: true})) + require.Len(t, migrator.applyEventsQueue, 4) + require.Nil(t, migrator.applier.AppliedTransactionCoordinates) + for i := 0; i < 3; i++ { + event := <-migrator.applyEventsQueue + require.NotNil(t, event.dmlEvent) + require.False(t, event.transactionComplete) + } + marker := <-migrator.applyEventsQueue + require.True(t, marker.transactionComplete) + require.Nil(t, marker.dmlEvent) +} + +func TestTransactionMarkerRejectedAfterAbort(t *testing.T) { + for _, abortErr := range []error{context.Canceled, errors.New("migration aborted")} { + t.Run(abortErr.Error(), func(t *testing.T) { + ctx := newTestMigrationContext() + defer ctx.CancelContext() + migrator := NewMigrator(ctx, "test") + migrator.applier = NewApplier(ctx) + previous := mysql.NewFileBinlogCoordinates("mysql-bin.000001", 100) + migrator.applier.AppliedTransactionCoordinates = previous.Clone() + marker := &applyEventStruct{coords: mysql.NewFileBinlogCoordinates("mysql-bin.000001", 200), transactionComplete: true} + if !errors.Is(abortErr, context.Canceled) { + ctx.SetAbortError(abortErr) + } + ctx.CancelContext() + require.ErrorIs(t, migrator.onApplyEventStruct(marker), abortErr) + require.True(t, previous.Equals(migrator.applier.AppliedTransactionCoordinates)) + }) + } +} + +func TestFailedSynchronousListenerStopsTransactionDelivery(t *testing.T) { + for _, failMarker := range []bool{false, true} { + t.Run(fmt.Sprintf("marker=%t", failMarker), func(t *testing.T) { + ctx := newTestMigrationContext() + defer ctx.CancelContext() + streamer := NewEventsStreamer(ctx) + failure := errors.New("listener rejected event") + called := 0 + fail := func(*binlog.BinlogEntry) error { called++; return failure } + unexpected := func(*binlog.BinlogEntry) error { t.Error("delivered after listener failure"); return nil } + entry := &binlog.BinlogEntry{TransactionComplete: failMarker} + if failMarker { + streamer.AddTransactionCompleteListener(fail) + } else { + entry.DmlEvent = binlog.NewBinlogDMLEvent(ctx.DatabaseName, ctx.OriginalTableName, binlog.DeleteDML) + require.NoError(t, streamer.AddListener(false, ctx.DatabaseName, ctx.OriginalTableName, fail)) + require.NoError(t, streamer.AddListener(false, ctx.DatabaseName, ctx.OriginalTableName, unexpected)) + } + streamer.AddTransactionCompleteListener(unexpected) + require.ErrorIs(t, streamer.notifyListeners(entry), failure) + require.ErrorIs(t, ctx.GetAbortError(), failure) + require.ErrorIs(t, ctx.GetContext().Err(), context.Canceled) + require.ErrorIs(t, streamer.notifyListeners(&binlog.BinlogEntry{TransactionComplete: true}), failure) + require.Equal(t, 1, called) + }) + } +} + +func TestCanceledDispatcherRejectsReadyQueue(t *testing.T) { + ctx := newTestMigrationContext() + migrator := NewMigrator(ctx, "test") + migrator.eventsStreamer = NewEventsStreamer(ctx) + require.NoError(t, migrator.addDMLEventsListener()) + ctx.CancelContext() + coords := mysql.NewFileBinlogCoordinates("mysql-bin.000001", 100) + for _, entry := range []*binlog.BinlogEntry{ + {Coordinates: coords, DmlEvent: binlog.NewBinlogDMLEvent(ctx.DatabaseName, ctx.OriginalTableName, binlog.DeleteDML)}, + {Coordinates: coords, TransactionComplete: true}, + } { + require.ErrorIs(t, migrator.eventsStreamer.notifyListeners(entry), context.Canceled) + } + require.Empty(t, migrator.applyEventsQueue) +} + +func TestTransactionMarkerClonesGTIDCoordinates(t *testing.T) { + for _, test := range []struct { + flavor string + committed string + next string + }{ + {mysql.MySQLFlavor, "00000000-0000-0000-0000-000000000001:1", "00000000-0000-0000-0000-000000000001:2"}, + {mysql.MariaDBFlavor, "0-1-1", "0-1-2"}, + } { + t.Run(test.flavor, func(t *testing.T) { + ctx := newTestMigrationContext() + ctx.UseGTIDs = true + migrator := NewMigrator(ctx, "test") + migrator.applier = NewApplier(ctx) + coords, err := mysql.NewGTIDBinlogCoordinates(test.flavor, test.committed) + require.NoError(t, err) + require.NoError(t, migrator.onTransactionComplete(&binlog.BinlogEntry{Coordinates: coords, TransactionComplete: true})) + require.Nil(t, migrator.applier.AppliedTransactionCoordinates, "enqueueing is not applying") + marker := <-migrator.applyEventsQueue + require.NoError(t, coords.GTIDSet.Update(test.next)) + require.NoError(t, migrator.onApplyEventStruct(marker)) + require.Equal(t, test.committed, migrator.applier.AppliedTransactionCoordinates.String()) + // The acknowledged watermark also owns its own copy of the marker. + require.NoError(t, marker.coords.(*mysql.GTIDBinlogCoordinates).GTIDSet.Update(test.next)) + require.Equal(t, test.committed, migrator.applier.AppliedTransactionCoordinates.String()) + }) + } +} + +func TestHeartbeatDoesNotAcknowledgeItsTransaction(t *testing.T) { + ctx := newTestMigrationContext() + migrator := NewMigrator(ctx, "test") + migrator.applier = NewApplier(ctx) + coords := mysql.NewFileBinlogCoordinates("mysql-bin.000001", 1234) + heartbeat := &binlog.BinlogEntry{ + Coordinates: coords, + DmlEvent: &binlog.BinlogDMLEvent{ + DML: binlog.InsertDML, + NewColumnValues: sql.ToColumnValues([]interface{}{1, 0, "heartbeat", time.Now().Format(time.RFC3339Nano)}), + }, + } + require.NoError(t, migrator.onChangelogHeartbeatEvent(heartbeat)) + require.NoError(t, migrator.onApplyEventStruct(<-migrator.applyEventsQueue)) + require.True(t, migrator.applier.CurrentCoordinates.Equals(coords)) + require.Nil(t, migrator.applier.AppliedTransactionCoordinates, "a heartbeat row may precede more rows in the same transaction") + require.NoError(t, migrator.onTransactionComplete(&binlog.BinlogEntry{Coordinates: coords, TransactionComplete: true})) + require.NoError(t, migrator.onApplyEventStruct(<-migrator.applyEventsQueue)) + require.True(t, migrator.applier.AppliedTransactionCoordinates.Equals(coords)) +}