redo: add DML two-stage ack

downstreamadapter/dispatchermanager/dispatcher_manager_redo.go

@@ -57,18 +57,17 @@ func initRedoComponet(
 	}
 	var err error
 	redoDispatcherMap := newDispatcherMap[*dispatcher.RedoDispatcher]()
-	redoSink, err := redo.New(ctx, changefeedID, manager.config.Consistent)
-	if err != nil {
-		return err
-	}
-	redoSchemaIDToDispatchers := dispatcher.NewSchemaIDToDispatchers()
-
 	totalQuota := manager.sinkQuota
 	consistentMemoryUsage := manager.config.Consistent.MemoryUsage
 	if consistentMemoryUsage == nil {
 		consistentMemoryUsage = config.GetDefaultReplicaConfig().Consistent.MemoryUsage
 	}
 	redoQuota := totalQuota * consistentMemoryUsage.MemoryQuotaPercentage / 100
+	redoSink, err := redo.New(ctx, changefeedID, manager.config.Consistent, redoQuota)
+	if err != nil {
+		return err
+	}
+	redoSchemaIDToDispatchers := dispatcher.NewSchemaIDToDispatchers()
 
 	manager.writePathMu.Lock()
 	if manager.writePathClosed.Load() {

downstreamadapter/sink/cloudstorage/buffer_manager.go

@@ -18,11 +18,11 @@ import (
 	"context"
 	"time"
 
-	"github.com/pingcap/ticdc/downstreamadapter/sink/cloudstorage/spool"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/metrics"
 	"github.com/pingcap/ticdc/pkg/cloudstorage"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/sink/spool"
 )
 
 const (

downstreamadapter/sink/cloudstorage/buffer_manager_test.go

@@ -18,12 +18,12 @@ import (
 	"testing"
 	"time"
 
-	"github.com/pingcap/ticdc/downstreamadapter/sink/cloudstorage/spool"
 	"github.com/pingcap/ticdc/pkg/cloudstorage"
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
+	"github.com/pingcap/ticdc/pkg/sink/spool"
 	"github.com/stretchr/testify/require"
 )
 

downstreamadapter/sink/cloudstorage/dml_writers.go

@@ -17,14 +17,14 @@ import (
 	"context"
 	"time"
 
-	"github.com/pingcap/ticdc/downstreamadapter/sink/cloudstorage/spool"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	sinkmetrics "github.com/pingcap/ticdc/downstreamadapter/sink/metrics"
 	"github.com/pingcap/ticdc/pkg/cloudstorage"
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/metrics"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
+	"github.com/pingcap/ticdc/pkg/sink/spool"
 	"github.com/pingcap/ticdc/utils/chann"
 	"github.com/pingcap/tidb/pkg/objstore/storeapi"
 	"go.uber.org/atomic"
@@ -68,6 +68,7 @@ func newDMLWriters(
 		changefeedID,
 		spool.WithRootDir(config.SpoolBaseDir),
 		spool.WithDiskQuotaBytes(config.SpoolDiskQuota),
+		spool.WithMetrics(newSpoolMetrics(changefeedID)),
 	)
 	if err != nil {
 		return nil, err

downstreamadapter/sink/cloudstorage/spool_metrics.go

@@ -0,0 +1,45 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cloudstorage
+
+import (
+	sinkmetrics "github.com/pingcap/ticdc/downstreamadapter/sink/metrics"
+	"github.com/pingcap/ticdc/pkg/common"
+	"github.com/pingcap/ticdc/pkg/sink/spool"
+)
+
+func newSpoolMetrics(changefeedID common.ChangeFeedID) *spool.Metrics {
+	keyspace := changefeedID.Keyspace()
+	changefeed := changefeedID.Name()
+	return &spool.Metrics{
+		MemoryBytes:        sinkmetrics.CloudStorageSpoolMemoryBytesGauge.WithLabelValues(keyspace, changefeed),
+		DiskBytes:          sinkmetrics.CloudStorageSpoolDiskBytesGauge.WithLabelValues(keyspace, changefeed),
+		PendingPostEnqueue: sinkmetrics.CloudStoragePendingPostEnqueueGauge.WithLabelValues(keyspace, changefeed),
+		DiskQuotaWaiters:   sinkmetrics.CloudStorageSpoolDiskQuotaWaitersGauge.WithLabelValues(keyspace, changefeed),
+		DiskQuotaWait:      sinkmetrics.CloudStorageSpoolDiskQuotaWaitDurationHistogram.WithLabelValues(keyspace, changefeed),
+		LoadedBytes:        sinkmetrics.CloudStorageLoadBytesHistogram.WithLabelValues(keyspace, changefeed),
+		RotatedCount:       sinkmetrics.CloudStorageRotateCountCounter.WithLabelValues(keyspace, changefeed),
+		SegmentCount:       sinkmetrics.CloudStorageSpoolSegmentCountGauge.WithLabelValues(keyspace, changefeed),
+		Close: func() {
+			sinkmetrics.CloudStorageSpoolMemoryBytesGauge.DeleteLabelValues(keyspace, changefeed)
+			sinkmetrics.CloudStorageSpoolDiskBytesGauge.DeleteLabelValues(keyspace, changefeed)
+			sinkmetrics.CloudStoragePendingPostEnqueueGauge.DeleteLabelValues(keyspace, changefeed)
+			sinkmetrics.CloudStorageSpoolDiskQuotaWaitersGauge.DeleteLabelValues(keyspace, changefeed)
+			sinkmetrics.CloudStorageSpoolDiskQuotaWaitDurationHistogram.DeleteLabelValues(keyspace, changefeed)
+			sinkmetrics.CloudStorageLoadBytesHistogram.DeleteLabelValues(keyspace, changefeed)
+			sinkmetrics.CloudStorageRotateCountCounter.DeleteLabelValues(keyspace, changefeed)
+			sinkmetrics.CloudStorageSpoolSegmentCountGauge.DeleteLabelValues(keyspace, changefeed)
+		},
+	}
+}

downstreamadapter/sink/cloudstorage/writer.go

@@ -20,12 +20,12 @@ import (
 	"time"
 
 	"github.com/pingcap/log"
-	"github.com/pingcap/ticdc/downstreamadapter/sink/cloudstorage/spool"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/metrics"
 	"github.com/pingcap/ticdc/pkg/cloudstorage"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/errors"
 	pmetrics "github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/pingcap/ticdc/pkg/sink/spool"
 	"github.com/pingcap/tidb/pkg/objstore/storeapi"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"

downstreamadapter/sink/cloudstorage/writer_test.go

@@ -26,14 +26,14 @@ import (
 	"testing"
 	"time"
 
-	"github.com/pingcap/ticdc/downstreamadapter/sink/cloudstorage/spool"
 	"github.com/pingcap/ticdc/pkg/cloudstorage"
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/metrics"
 	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
+	"github.com/pingcap/ticdc/pkg/sink/spool"
 	"github.com/pingcap/ticdc/pkg/util"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/objstore/objectio"

downstreamadapter/sink/helper/row_callback.go

@@ -21,10 +21,16 @@ import (
 // NewPostFlushRowCallback returns a row-level callback that triggers txn-level
 // PostFlush exactly once when the callback has been invoked totalCount times.
 func NewPostFlushRowCallback(event *event.DMLEvent, totalCount uint64) func() {
+	return NewRowCallback(totalCount, event.PostFlush)
+}
+
+// NewRowCallback returns a row-level callback that triggers callback exactly
+// once after it has been invoked totalCount times.
+func NewRowCallback(totalCount uint64, callback func()) func() {
 	var calledCount atomic.Uint64
 	return func() {
 		if calledCount.Inc() == totalCount {
-			event.PostFlush()
+			callback()
 		}
 	}
 }