Remove the timeout for synchronous backup, revert the timeout move to return the behavior to 30 days _between_ events, restore some comments, use struct instead of bool as recommended, add a note about SynchronousEventHandler and the parseEvent return values

Author: Dylan Terry

replication/backup.go

@@ -23,7 +23,7 @@ func (b *BinlogSyncer) StartBackup(backupDir string, p Position, timeout time.Du
 			return os.OpenFile(path.Join(backupDir, filename), os.O_CREATE|os.O_WRONLY, 0644)
 		})
 	} else {
-		return b.StartSynchronousBackup(p, timeout)
+		return b.StartSynchronousBackup(p)
 	}
 }
 
@@ -68,10 +68,10 @@ func (b *BinlogSyncer) StartBackupWithHandler(p Position, timeout time.Duration,
 		}
 	}()
 
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-
 	for {
+		ctx, cancel := context.WithTimeout(context.Background(), timeout)
+		defer cancel()
+
 		select {
 		case <-ctx.Done():
 			return nil
@@ -89,27 +89,17 @@ func (b *BinlogSyncer) StartBackupWithHandler(p Position, timeout time.Duration,
 }
 
 // StartSynchronousBackup starts the backup process using the SynchronousEventHandler in the BinlogSyncerConfig.
-func (b *BinlogSyncer) StartSynchronousBackup(p Position, timeout time.Duration) error {
+func (b *BinlogSyncer) StartSynchronousBackup(p Position) error {
 	if b.cfg.SynchronousEventHandler == nil {
 		return errors.New("SynchronousEventHandler must be set in BinlogSyncerConfig to use StartSynchronousBackup")
 	}
 
-	if timeout == 0 {
-		timeout = 30 * 3600 * 24 * time.Second // Long timeout by default
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-
 	s, err := b.StartSync(p)
 	if err != nil {
 		return errors.Trace(err)
 	}
 
-	// Wait for the binlog syncer to finish or encounter an error
 	select {
-	case <-ctx.Done():
-		return nil
 	case <-b.ctx.Done():
 		return nil
 	case err := <-s.ech:
@@ -138,6 +128,7 @@ func (h *BackupEventHandler) HandleEvent(e *BinlogEvent) error {
 		rotateEvent := e.Event.(*RotateEvent)
 		h.filename = string(rotateEvent.NextLogName)
 		if e.Header.Timestamp == 0 || offset == 0 {
+			// fake rotate event
 			return nil
 		}
 	} else if e.Header.EventType == FORMAT_DESCRIPTION_EVENT {
@@ -157,6 +148,7 @@ func (h *BackupEventHandler) HandleEvent(e *BinlogEvent) error {
 			return errors.Trace(err)
 		}
 
+		// Write binlog header 0xfebin
 		_, err = h.w.Write(BinLogFileHeader)
 		if err != nil {
 			return errors.Trace(err)

replication/backup_test.go

@@ -32,12 +32,12 @@ func (t *testSyncerSuite) TestStartBackupEndInGivenTime() {
 	os.RemoveAll(binlogDir)
 	timeout := 2 * time.Second
 
-	done := make(chan bool)
+	done := make(chan struct{})
 
 	go func() {
 		err := t.b.StartBackup(binlogDir, mysql.Position{Name: "", Pos: uint32(0)}, timeout)
 		require.NoError(t.T(), err)
-		done <- true
+		close(done)
 	}()
 	failTimeout := 5 * timeout
 	ctx, cancel := context.WithTimeout(context.Background(), failTimeout)

replication/binlogsyncer.go

@@ -128,6 +128,7 @@ type BinlogSyncerConfig struct {
 
 	// SynchronousEventHandler is used for synchronous event handling.
 	// This should not be used together with StartBackupWithHandler.
+	// If this is not nil, GetEvent does not need to be called.
 	SynchronousEventHandler EventHandler
 }
 
@@ -805,24 +806,26 @@ func (b *BinlogSyncer) onStream(s *BinlogStreamer) {
 
 // parseEvent parses the raw data into a BinlogEvent.
 // It only handles parsing and does not perform any side effects.
-func (b *BinlogSyncer) parseEvent(data []byte) (*BinlogEvent, bool, error) {
+// Returns the parsed BinlogEvent, a boolean indicating if an ACK is needed, and an error if the
+// parsing fails
+func (b *BinlogSyncer) parseEvent(data []byte) (event *BinlogEvent, needACK bool, err error) {
 	// Skip OK byte (0x00)
 	data = data[1:]
 
-	needACK := false
+	needACK = false
 	if b.cfg.SemiSyncEnabled && data[0] == SemiSyncIndicator {
 		needACK = data[1] == 0x01
 		// Skip semi-sync header
 		data = data[2:]
 	}
 
 	// Parse the event using the BinlogParser
-	e, err := b.parser.Parse(data)
+	event, err = b.parser.Parse(data)
 	if err != nil {
 		return nil, false, errors.Trace(err)
 	}
 
-	return e, needACK, nil
+	return event, needACK, nil
 }
 
 // handleEventAndACK processes an event and sends an ACK if necessary.