redo: release callbacks after size rotation

Release post-flush callbacks as soon as size-rotated redo files are durable.

Clear invoked callback slots to avoid retaining receivers through slice capacity, while preserving callback order across concurrent memory-backend uploads.

Author: wlwilliamx

pkg/redo/writer/file/file.go

@@ -228,24 +228,33 @@ func (w *Writer) Run(ctx context.Context) error {
 // Write implement write interface
 // TODO: more general api with fileName generated by caller
 func (w *Writer) Write(rawData []byte) (int, error) {
+	n, _, err := w.writeRawData(rawData)
+	return n, err
+}
+
+// writeRawData reports whether writing rawData first made the previous file
+// durable through a size-triggered rotation.
+func (w *Writer) writeRawData(rawData []byte) (int, bool, error) {
 	w.Lock()
 	defer w.Unlock()
 
 	writeLen := int64(len(rawData))
 	if writeLen > w.cfg.MaxLogSizeInBytes() {
-		return 0, errors.ErrRedoFileSizeExceed.GenWithStackByArgs(writeLen, w.cfg.MaxLogSizeInBytes())
+		return 0, false, errors.ErrRedoFileSizeExceed.GenWithStackByArgs(writeLen, w.cfg.MaxLogSizeInBytes())
 	}
 
 	if w.file == nil {
 		if err := w.openNew(); err != nil {
-			return 0, err
+			return 0, false, err
 		}
 	}
 
+	rotated := false
 	if w.size+writeLen > w.cfg.MaxLogSizeInBytes() {
 		if err := w.rotate(); err != nil {
-			return 0, err
+			return 0, false, err
 		}
+		rotated = true
 	}
 
 	if w.maxCommitTS.Load() < w.eventCommitTS.Load() {
@@ -254,7 +263,7 @@ func (w *Writer) Write(rawData []byte) (int, error) {
 	// ref: https://github.com/etcd-io/etcd/pull/5250
 	lenField, padBytes := writer.EncodeFrameSize(len(rawData))
 	if err := w.writeUint64(lenField, w.uint64buf); err != nil {
-		return 0, err
+		return 0, rotated, err
 	}
 
 	if padBytes != 0 {
@@ -263,12 +272,12 @@ func (w *Writer) Write(rawData []byte) (int, error) {
 
 	n, err := w.bw.Write(rawData)
 	if err != nil {
-		return 0, err
+		return 0, rotated, err
 	}
 	w.metricWriteBytes.Add(float64(n))
 	w.size += int64(n)
 
-	return n, err
+	return n, rotated, nil
 }
 
 // AdvanceTs implement Advance interface
@@ -327,25 +336,22 @@ func (w *Writer) GetInputCh() chan writer.RedoEvent {
 	return w.inputCh
 }
 
-func (w *Writer) write(event writer.RedoEvent) error {
+func (w *Writer) write(event writer.RedoEvent) (bool, error) {
 	rl := event.ToRedoLog()
 	if rl.Type == commonEvent.RedoLogTypeDDL {
 		rl.RedoDDL.SetTableSchemaStore(w.tableSchemaStore)
 	}
 	data, err := codec.MarshalRedoLog(rl, nil)
 	if err != nil {
-		return errors.WrapError(errors.ErrMarshalFailed, err)
+		return false, errors.WrapError(errors.ErrMarshalFailed, err)
 	}
 	w.AdvanceTs(rl.GetCommitTs())
-	_, err = w.Write(data)
-	if err != nil {
-		return err
-	}
-	return nil
+	_, rotated, err := w.writeRawData(data)
+	return rotated, err
 }
 
 func (w *Writer) SyncWrite(event writer.RedoEvent) error {
-	err := w.write(event)
+	_, err := w.write(event)
 	if err != nil {
 		return err
 	}
@@ -364,16 +370,22 @@ func (w *Writer) encode(ctx context.Context) error {
 	num := 0
 	flushBatchSize := w.cfg.FlushBatchSize()
 	cacheEventPostFlush := make([]func(), 0)
+	runCachedPostFlush := func() {
+		for i, fn := range cacheEventPostFlush {
+			// Clear the slot before invocation so retained slice capacity does not
+			// keep the callback receiver alive after it has been acknowledged.
+			cacheEventPostFlush[i] = nil
+			fn()
+		}
+		cacheEventPostFlush = cacheEventPostFlush[:0]
+	}
 	flush := func() error {
 		err := w.Flush()
 		if err != nil {
 			return err
 		}
-		for _, fn := range cacheEventPostFlush {
-			fn()
-		}
+		runCachedPostFlush()
 		num = 0
-		cacheEventPostFlush = cacheEventPostFlush[:0]
 		return nil
 	}
 	for {
@@ -386,7 +398,13 @@ func (w *Writer) encode(ctx context.Context) error {
 				return errors.Trace(err)
 			}
 		case e := <-w.inputCh:
-			err := w.write(e)
+			rotated, err := w.write(e)
+			if rotated {
+				// The previous file is durable before the current event is written,
+				// so only callbacks accumulated before this event can run here.
+				runCachedPostFlush()
+				num = 0
+			}
 			if err != nil {
 				return err
 			}

pkg/redo/writer/file/file_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/pingcap/ticdc/pkg/fsutil"
 	"github.com/pingcap/ticdc/pkg/metrics"
 	"github.com/pingcap/ticdc/pkg/redo"
+	"github.com/pingcap/ticdc/pkg/redo/codec"
 	"github.com/pingcap/ticdc/pkg/redo/writer"
 	"github.com/pingcap/ticdc/pkg/util"
 	"github.com/pingcap/ticdc/pkg/uuid"
@@ -503,6 +504,71 @@ func TestRunFlushesOnBatchBoundaryAndExecutesPostFlush(t *testing.T) {
 	require.NoError(t, w.Close())
 }
 
+// TestRunReleasesCallbacksAfterSizeRotation writes one event that exactly fills
+// a local redo file, confirms its callback remains pending, then writes a second
+// event to rotate the first file. The first callback must run after that durable
+// rotation while the second event remains unacknowledged in the current file.
+func TestRunReleasesCallbacksAfterSizeRotation(t *testing.T) {
+	dir := t.TempDir()
+	firstCallbackDone := make(chan struct{})
+	firstEvent := &pevent.RedoRowEvent{
+		StartTs:  1,
+		CommitTs: 1,
+		Callback: func() {
+			close(firstCallbackDone)
+		},
+	}
+	encodedFirstEvent, err := codec.MarshalRedoLog(firstEvent.ToRedoLog(), nil)
+	require.NoError(t, err)
+	_, padBytes := writer.EncodeFrameSize(len(encodedFirstEvent))
+	maxLogSizeInBytes := int64(8 + len(encodedFirstEvent) + padBytes)
+
+	w, err := newWriter(&localFileConfig{
+		dir:               dir,
+		maxLogSizeInBytes: maxLogSizeInBytes,
+		flushIntervalInMs: int64(time.Hour / time.Millisecond),
+		flushBatchSize:    0,
+		flushWorkerNum:    1,
+	}, redo.RedoRowLogFileType, nil)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	runErrCh := make(chan error, 1)
+	go func() {
+		runErrCh <- w.Run(ctx)
+	}()
+
+	w.GetInputCh() <- firstEvent
+	require.Eventually(t, func() bool {
+		return w.eventCommitTS.Load() == firstEvent.CommitTs
+	}, 10*time.Second, 10*time.Millisecond)
+	select {
+	case <-firstCallbackDone:
+		require.FailNow(t, "callback ran before the file became durable")
+	default:
+	}
+
+	secondCallbackCount := atomic.NewInt64(0)
+	secondEvent := &pevent.RedoRowEvent{
+		StartTs:  2,
+		CommitTs: 2,
+		Callback: func() {
+			secondCallbackCount.Inc()
+		},
+	}
+	w.GetInputCh() <- secondEvent
+	select {
+	case <-firstCallbackDone:
+	case <-time.After(10 * time.Second):
+		require.FailNow(t, "callback was not released after size rotation")
+	}
+	require.Zero(t, secondCallbackCount.Load())
+
+	cancel()
+	require.ErrorIs(t, <-runErrCh, context.Canceled)
+	require.NoError(t, w.Close())
+}
+
 // TestRunDisablesCountBasedFlushWithZero writes more rows than the old fixed
 // boundary while using a long ticker interval, then verifies the file backend
 // processes them without executing callbacks through a row-count flush.

pkg/redo/writer/memory/file_worker.go

@@ -47,6 +47,8 @@ type fileCache struct {
 	filename string
 	flushed  chan struct{}
 	writer   *dataWriter
+
+	postFlushCallbacks []func()
 }
 
 type dataWriter struct {
@@ -83,6 +85,22 @@ func (f *fileCache) markFlushed() {
 	}
 }
 
+func (f *fileCache) addPostFlushCallback(callback func()) {
+	if callback != nil {
+		f.postFlushCallbacks = append(f.postFlushCallbacks, callback)
+	}
+}
+
+// runPostFlushCallbacks clears each slot before invocation so the retained
+// slice capacity cannot keep callback receivers alive after the file is durable.
+func (f *fileCache) runPostFlushCallbacks() {
+	for i, callback := range f.postFlushCallbacks {
+		f.postFlushCallbacks[i] = nil
+		callback()
+	}
+	f.postFlushCallbacks = nil
+}
+
 type fileWorkerGroup struct {
 	cfg           *writer.Config
 	op            *writer.LogWriterOptions
@@ -214,23 +232,27 @@ func (f *fileWorkerGroup) bgWriteLogs(
 	defer ticker.Stop()
 	num := 0
 	flushBatchSize := f.cfg.FlushBatchSize()
-	cacheEventPostFlush := make([]func(), 0)
 	flush := func() error {
 		err := f.flushAll(egCtx)
 		if err != nil {
 			return err
 		}
-		for _, fn := range cacheEventPostFlush {
-			fn()
-		}
 		num = 0
-		cacheEventPostFlush = cacheEventPostFlush[:0]
 		return nil
 	}
 	for {
+		// A size-rotated file can finish independently of the current file.
+		// Release only the durable prefix to preserve input callback order.
+		f.releaseFlushedFiles()
+		var firstRotatedFileFlushed <-chan struct{}
+		if len(f.files) > 1 {
+			firstRotatedFileFlushed = f.files[0].flushed
+		}
 		select {
 		case <-egCtx.Done():
 			return errors.Trace(egCtx.Err())
+		case <-firstRotatedFileFlushed:
+			continue
 		case <-ticker.C:
 			err := flush()
 			if err != nil {
@@ -241,20 +263,20 @@ func (f *fileWorkerGroup) bgWriteLogs(
 				log.Error("inputCh of redo file worker is closed unexpectedly")
 				return errors.ErrUnexpected.FastGenByArgs("inputCh of redo file worker is closed unexpectedly")
 			}
-			err := f.writeToCache(egCtx, event)
+			rotated, err := f.writeToCache(egCtx, event)
 			if err != nil {
 				return errors.Trace(err)
 			}
+			if rotated {
+				num = 0
+			}
 			num++
 			// Zero leaves file size and the periodic ticker as the only flush triggers.
 			if flushBatchSize > 0 && num >= flushBatchSize {
 				err := flush()
 				if err != nil {
 					return errors.Trace(err)
 				}
-				event.PostFlush()
-			} else {
-				cacheEventPostFlush = append(cacheEventPostFlush, event.PostFlush)
 			}
 		}
 	}
@@ -322,46 +344,48 @@ func (f *fileWorkerGroup) newFileCache(data []byte, commitTs common.Ts) *fileCac
 
 func (f *fileWorkerGroup) writeToCache(
 	egCtx context.Context, event *polymorphicRedoEvent,
-) (err error) {
+) (rotated bool, err error) {
 	commitTs := event.commitTs
 	data := event.data
 	if len(data) == 0 {
-		return errors.ErrUnexpected.FastGenByArgs("encoded redo event data is empty")
+		return false, errors.ErrUnexpected.FastGenByArgs("encoded redo event data is empty")
 	}
 	writeLen := int64(len(data))
 	if writeLen > f.cfg.MaxLogSizeInBytes() {
 		// TODO: maybe we need to deal with the oversized commonEvent.
-		return errors.ErrRedoFileSizeExceed.GenWithStackByArgs(writeLen, f.cfg.MaxLogSizeInBytes())
+		return false, errors.ErrRedoFileSizeExceed.GenWithStackByArgs(writeLen, f.cfg.MaxLogSizeInBytes())
 	}
 	defer f.metricWriteBytes.Add(float64(writeLen))
 
 	if len(f.files) == 0 {
 		file := f.newFileCache(data, commitTs)
 		if file == nil {
-			return errors.ErrRedoWriterStopped.FastGenByArgs("failed to create file cache")
+			return false, errors.ErrRedoWriterStopped.FastGenByArgs("failed to create file cache")
 		}
+		file.addPostFlushCallback(event.callback)
 		f.files = append(f.files, file)
-		return nil
+		return false, nil
 	}
 
 	file := f.files[len(f.files)-1]
 	if file.fileSize+writeLen > f.cfg.MaxLogSizeInBytes() {
 		select {
 		case <-egCtx.Done():
-			return errors.Trace(egCtx.Err())
+			return false, errors.Trace(egCtx.Err())
 		case f.flushCh <- file:
 		}
 		file := f.newFileCache(data, commitTs)
 		if file == nil {
-			return errors.ErrRedoWriterStopped.FastGenByArgs("failed to create file cache")
+			return false, errors.ErrRedoWriterStopped.FastGenByArgs("failed to create file cache")
 		}
+		file.addPostFlushCallback(event.callback)
 		f.files = append(f.files, file)
-		return nil
+		return true, nil
 	}
 
 	_, err = file.writer.Write(data)
 	if err != nil {
-		return err
+		return false, err
 	}
 
 	file.fileSize += writeLen
@@ -371,7 +395,24 @@ func (f *fileWorkerGroup) writeToCache(
 	if commitTs < file.minCommitTs {
 		file.minCommitTs = commitTs
 	}
-	return nil
+	file.addPostFlushCallback(event.callback)
+	return false, nil
+}
+
+// releaseFlushedFiles invokes callbacks for the durable prefix of rotated
+// files. The last file is still writable and must remain pending until a flush.
+func (f *fileWorkerGroup) releaseFlushedFiles() {
+	for len(f.files) > 1 {
+		file := f.files[0]
+		select {
+		case <-file.flushed:
+			file.runPostFlushCallbacks()
+			f.files[0] = nil
+			f.files = f.files[1:]
+		default:
+			return
+		}
+	}
 }
 
 func (f *fileWorkerGroup) flushAll(egCtx context.Context) error {
@@ -393,6 +434,10 @@ func (f *fileWorkerGroup) flushAll(egCtx context.Context) error {
 			return errors.Trace(err)
 		}
 	}
+	for _, file := range f.files {
+		file.runPostFlushCallbacks()
+	}
+	clear(f.files)
 	f.files = f.files[:0]
 	return nil
 }