Checkpoint only fully applied binlog transactions

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

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

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 {

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

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

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
 	}

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