This commit addresses an issue where acknowledgments (ACKs) were sometimes sent to the master before binlog events were fully written and fsynced to disk during backup operations. Sending ACKs prematurely in semi-synchronous replication could lead to data loss if the replica fails after sending the ACK but before persisting the event.

Key changes:

- Introduced an `EventHandler` interface with a `HandleEvent` method for
  processing binlog events. This allows custom event handling logic to
  be injected into the replication stream.

- Added an `eventHandler` field to `BinlogSyncer` and provided a
  `SetEventHandler` method to assign an event handler. This enables
  `BinlogSyncer` to delegate event processing to the assigned handler.

- Implemented `BackupEventHandler` which writes binlog events to disk
  and ensures that each event is fsynced before returning. This ensures
  data durability before ACKs are sent.

- Modified the `onStream` method in `BinlogSyncer` to separate event
  parsing (`parseEvent`) from event handling and ACK sending
  (`handleEventAndACK`). This adheres to the single-responsibility
  principle and makes the code cleaner.

- Moved state updates (e.g., updating `b.nextPos`) and GTID set handling
  from `parseEvent` to `handleEventAndACK` to avoid side effects during
  parsing.

- Ensured that ACKs are sent only after the event has been fully
  processed and fsynced by sending the ACK in `handleEventAndACK` after
  event handling.

Author: Dylan Terry

replication/backup.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"os"
 	"path"
+	"sync"
 	"time"
 
 	. "github.com/go-mysql-org/go-mysql/mysql"
@@ -41,77 +42,143 @@ func (b *BinlogSyncer) StartBackupWithHandler(p Position, timeout time.Duration,
 	// Force use raw mode
 	b.parser.SetRawMode(true)
 
+	// Set up the backup event handler
+	backupHandler := &BackupEventHandler{
+		handler: handler,
+	}
+
+	if b.cfg.SyncMode == SyncModeSync {
+		// Set the event handler in BinlogSyncer for synchronous mode
+		b.SetEventHandler(backupHandler)
+	}
+
 	s, err := b.StartSync(p)
 	if err != nil {
 		return errors.Trace(err)
 	}
 
-	var filename string
-	var offset uint32
-
-	var w io.WriteCloser
 	defer func() {
-		var closeErr error
-		if w != nil {
-			closeErr = w.Close()
-		}
-		if retErr == nil {
-			retErr = closeErr
+		b.SetEventHandler(nil) // Reset the event handler
+		if backupHandler.w != nil {
+			closeErr := backupHandler.w.Close()
+			if retErr == nil {
+				retErr = closeErr
+			}
 		}
 	}()
 
-	for {
-		ctx, cancel := context.WithTimeout(context.Background(), timeout)
-		e, err := s.GetEvent(ctx)
-		cancel()
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
 
-		if err == context.DeadlineExceeded {
+	if b.cfg.SyncMode == SyncModeSync {
+		// Synchronous mode: wait for completion or error
+		select {
+		case <-ctx.Done():
 			return nil
-		}
-
-		if err != nil {
+		case <-b.ctx.Done():
+			return nil
+		case err := <-s.ech:
 			return errors.Trace(err)
 		}
+	} else {
+		// Asynchronous mode: consume events from the streamer
+		for {
+			select {
+			case <-ctx.Done():
+				return nil
+			case <-b.ctx.Done():
+				return nil
+			case err := <-s.ech:
+				return errors.Trace(err)
+			case e := <-s.ch:
+				err = backupHandler.HandleEvent(e)
+				if err != nil {
+					return errors.Trace(err)
+				}
+			}
+		}
+	}
+}
 
-		offset = e.Header.LogPos
+// BackupEventHandler handles writing events for backup
+type BackupEventHandler struct {
+	handler     func(binlogFilename string) (io.WriteCloser, error)
+	w           io.WriteCloser
+	mutex       sync.Mutex
+	fsyncedChan chan struct{}
+	eventCount  int // eventCount used for testing
 
-		if e.Header.EventType == ROTATE_EVENT {
-			rotateEvent := e.Event.(*RotateEvent)
-			filename = string(rotateEvent.NextLogName)
+	filename string
+}
 
-			if e.Header.Timestamp == 0 || offset == 0 {
-				// fake rotate event
-				continue
-			}
-		} else if e.Header.EventType == FORMAT_DESCRIPTION_EVENT {
-			// FormateDescriptionEvent is the first event in binlog, we will close old one and create a new
+func (h *BackupEventHandler) HandleEvent(e *BinlogEvent) error {
+	h.mutex.Lock()
+	defer h.mutex.Unlock()
 
-			if w != nil {
-				if err = w.Close(); err != nil {
-					w = nil
-					return errors.Trace(err)
-				}
-			}
+	var err error
 
-			if len(filename) == 0 {
-				return errors.Errorf("empty binlog filename for FormateDescriptionEvent")
-			}
+	// Update the offset
+	offset := e.Header.LogPos
 
-			w, err = handler(filename)
-			if err != nil {
-				return errors.Trace(err)
-			}
+	if e.Header.EventType == ROTATE_EVENT {
+		rotateEvent := e.Event.(*RotateEvent)
+		h.filename = string(rotateEvent.NextLogName)
 
-			// write binlog header fe'bin'
-			if _, err = w.Write(BinLogFileHeader); err != nil {
+		if e.Header.Timestamp == 0 || offset == 0 {
+			// Fake rotate event, skip processing
+			return nil
+		}
+	} else if e.Header.EventType == FORMAT_DESCRIPTION_EVENT {
+		// Close the current writer and open a new one
+		if h.w != nil {
+			if err = h.w.Close(); err != nil {
+				h.w = nil
 				return errors.Trace(err)
 			}
 		}
 
-		if n, err := w.Write(e.RawData); err != nil {
+		if len(h.filename) == 0 {
+			return errors.Errorf("empty binlog filename for FormatDescriptionEvent")
+		}
+
+		h.w, err = h.handler(h.filename)
+		if err != nil {
+			return errors.Trace(err)
+		}
+
+		// Write binlog header fe'bin'
+		_, err = h.w.Write(BinLogFileHeader)
+		if err != nil {
 			return errors.Trace(err)
-		} else if n != len(e.RawData) {
+		}
+	}
+
+	// Write raw event data to the current writer
+	if h.w != nil {
+		n, err := h.w.Write(e.RawData)
+		if err != nil {
+			return errors.Trace(err)
+		}
+		if n != len(e.RawData) {
 			return errors.Trace(io.ErrShortWrite)
 		}
+
+		// Perform Sync if the writer supports it
+		if f, ok := h.w.(*os.File); ok {
+			if err := f.Sync(); err != nil {
+				return errors.Trace(err)
+			}
+			// Signal that fsync has completed
+			if h.fsyncedChan != nil {
+				h.fsyncedChan <- struct{}{}
+			}
+		}
+	} else {
+		// If writer is nil and event is not FORMAT_DESCRIPTION_EVENT, we can't write
+		// This should not happen if events are in expected order
+		return errors.New("writer is not initialized")
 	}
+
+	h.eventCount++
+	return nil
 }

replication/backup_test.go

@@ -47,3 +47,49 @@ func (t *testSyncerSuite) TestStartBackupEndInGivenTime() {
 		t.T().Fatal("time out error")
 	}
 }
+
+// TestAsyncBackup runs the backup process in asynchronous mode and verifies binlog file creation.
+func (t *testSyncerSuite) TestAsyncBackup() {
+	testSyncModeBackup(t, SyncModeAsync)
+}
+
+// TestSyncBackup runs the backup process in synchronous mode and verifies binlog file creation.
+func (t *testSyncerSuite) TestSyncBackup() {
+	testSyncModeBackup(t, SyncModeSync)
+}
+
+// testSyncModeBackup is a helper function that runs the backup process for a given sync mode and checks if binlog files are written correctly.
+func testSyncModeBackup(t *testSyncerSuite, syncMode SyncMode) {
+	t.setupTest(mysql.MySQLFlavor)
+	t.b.cfg.SemiSyncEnabled = false // Ensure semi-sync is disabled
+	t.b.cfg.SyncMode = syncMode     // Set the sync mode
+
+	binlogDir := "./var"
+	os.RemoveAll(binlogDir)
+	timeout := 3 * time.Second
+
+	done := make(chan bool)
+
+	// Start the backup process in a goroutine
+	go func() {
+		err := t.b.StartBackup(binlogDir, mysql.Position{Name: "", Pos: uint32(0)}, timeout)
+		require.NoError(t.T(), err)
+		done <- true
+	}()
+
+	failTimeout := 2 * timeout
+	ctx, cancel := context.WithTimeout(context.Background(), failTimeout)
+	defer cancel()
+
+	// Wait for the backup to complete or timeout
+	select {
+	case <-done:
+		// Check if binlog files are written to the specified directory
+		files, err := os.ReadDir(binlogDir)
+		require.NoError(t.T(), err, "Failed to read binlog directory")
+		require.Greater(t.T(), len(files), 0, "Binlog files were not written to the directory")
+		t.T().Logf("Backup completed successfully in %v mode with %d binlog file(s).", syncMode, len(files))
+	case <-ctx.Done():
+		t.T().Fatalf("Timeout error during backup in %v mode.", syncMode)
+	}
+}