[DNM] Branch for CSE testing

deployments/Dockerfile

@@ -4,7 +4,7 @@ WORKDIR /go/src/github.com/pingcap/ticdc
 COPY . .
 ENV CDC_ENABLE_VENDOR=0
 ARG NEXT_GEN
-RUN if [ -n "${NEXT_GEN}" ]; then NEXT_GEN=1 make cdc; else make cdc; fi
+RUN if [ -n "${NEXT_GEN}" ]; then NEXT_GEN=1 LEGACY_SAFEPOINT=1 make cdc; else make cdc; fi
 
 FROM alpine:3.15
 RUN apk add --no-cache tzdata bash curl socat

downstreamadapter/sink/mysql/causality/barrier_test.go

@@ -0,0 +1,183 @@
+// 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 causality
+
+import (
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+type testBarrier struct {
+	mu        sync.Mutex
+	done      chan struct{}
+	err       error
+	remaining int
+	doneFuncs []func()
+}
+
+func newTestBarrier(workerCount int) *testBarrier {
+	return &testBarrier{done: make(chan struct{}), remaining: workerCount}
+}
+
+func (b *testBarrier) Ack(int) {
+	b.mu.Lock()
+	if b.remaining == 0 {
+		b.mu.Unlock()
+		return
+	}
+	b.remaining--
+	if b.remaining > 0 {
+		b.mu.Unlock()
+		return
+	}
+	doneFuncs := b.doneFuncs
+	b.doneFuncs = nil
+	close(b.done)
+	b.mu.Unlock()
+	for _, f := range doneFuncs {
+		f()
+	}
+}
+
+func (b *testBarrier) Fail(err error) {
+	b.mu.Lock()
+	if b.remaining == 0 {
+		b.mu.Unlock()
+		return
+	}
+	b.err = err
+	b.remaining = 0
+	doneFuncs := b.doneFuncs
+	b.doneFuncs = nil
+	close(b.done)
+	b.mu.Unlock()
+	for _, f := range doneFuncs {
+		f()
+	}
+}
+
+func (b *testBarrier) OnDone(f func()) {
+	b.mu.Lock()
+	if b.remaining == 0 {
+		b.mu.Unlock()
+		f()
+		return
+	}
+	b.doneFuncs = append(b.doneFuncs, f)
+	b.mu.Unlock()
+}
+
+func TestBroadcastBarrierEnqueuesOneTokenPerWriter(t *testing.T) {
+	detector := New(4, TxnCacheOption{Count: 2, Size: 1, BlockStrategy: BlockStrategyWaitEmpty}, testChangefeedID())
+	barrier := newTestBarrier(2)
+
+	require.NoError(t, detector.BroadcastBarrier(barrier))
+
+	for i := 0; i < 2; i++ {
+		items, ok := detector.GetOutChByCacheID(i).GetMultipleNoGroup(make([]WriterItem, 0, 1))
+		require.True(t, ok)
+		require.Len(t, items, 1)
+		require.True(t, barrier == items[0].Barrier)
+		items[0].Barrier.Ack(i)
+	}
+
+	require.Eventually(t, func() bool {
+		select {
+		case <-barrier.done:
+			return true
+		default:
+			return false
+		}
+	}, time.Second, 10*time.Millisecond)
+}
+
+func TestBroadcastBarrierReturnsErrorWhenDetectorClosed(t *testing.T) {
+	detector := New(4, TxnCacheOption{Count: 1, Size: 1, BlockStrategy: BlockStrategyWaitEmpty}, testChangefeedID())
+	detector.CloseNotifiedNodes()
+
+	err := detector.BroadcastBarrier(newTestBarrier(1))
+	require.Error(t, err)
+}
+
+func TestRemovalOnlyFenceDoesNotResolveDependersOnAssignment(t *testing.T) {
+	assigned := false
+	fence := &Node{id: genNextNodeID(), assignedTo: unassigned, resolveByRemovalOnly: true}
+	fence.RandCacheID = func() cacheID { return 0 }
+	fence.TrySendToTxnCache = func(cacheID) bool { return true }
+	fence.OnNotified = func(callback func()) { callback() }
+
+	depender := &Node{id: genNextNodeID(), assignedTo: unassigned}
+	depender.RandCacheID = func() cacheID { return 0 }
+	depender.TrySendToTxnCache = func(cacheID) bool {
+		assigned = true
+		return true
+	}
+	depender.OnNotified = func(callback func()) { callback() }
+
+	depender.dependOn(map[int64]*Node{fence.nodeID(): fence})
+	fence.maybeResolve()
+	require.False(t, assigned)
+
+	fence.remove()
+	require.True(t, assigned)
+}
+
+func TestTxnCacheForceAddBypassesBlockedCache(t *testing.T) {
+	cache := newTxnCache(TxnCacheOption{Count: 1, Size: 1, BlockStrategy: BlockStrategyWaitEmpty})
+	require.True(t, cache.add(NewDMLItem(nil)))
+	require.True(t, cache.add(NewDMLItem(nil)))
+	require.False(t, cache.add(NewDMLItem(nil)))
+
+	barrier := newTestBarrier(1)
+	require.True(t, cache.forceAdd(NewBarrierItem(barrier)))
+
+	items, ok := cache.out().GetMultipleNoGroup(make([]WriterItem, 0, 3))
+	require.True(t, ok)
+	require.Len(t, items, 3)
+	require.True(t, barrier == items[2].Barrier)
+}
+
+func TestTxnCacheForceAddFailsWhenClosed(t *testing.T) {
+	cache := newTxnCache(TxnCacheOption{Count: 1, Size: 1, BlockStrategy: BlockStrategyWaitEmpty})
+	cache.out().Close()
+
+	barrier := newTestBarrier(1)
+	require.False(t, cache.forceAdd(NewBarrierItem(barrier)))
+	barrier.Fail(errors.New("closed"))
+	require.Error(t, barrier.err)
+}
+
+func BenchmarkTxnCacheAddDMLItem(b *testing.B) {
+	cache := newTxnCache(TxnCacheOption{Count: 1, Size: 4096, BlockStrategy: BlockStrategyWaitAvailable})
+	buffer := make([]WriterItem, 0, 1024)
+	b.ReportAllocs()
+
+	for i := 0; i < b.N; i++ {
+		if !cache.add(NewDMLItem(nil)) {
+			b.Fatal("cache unexpectedly rejected DML item")
+		}
+		if (i+1)%cap(buffer) == 0 {
+			var ok bool
+			buffer, ok = cache.out().GetMultipleNoGroup(buffer)
+			if !ok {
+				b.Fatal("cache closed")
+			}
+			buffer = buffer[:0]
+		}
+	}
+}

downstreamadapter/sink/mysql/causality/conflict_detector.go

@@ -15,6 +15,7 @@
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	"github.com/pingcap/log"
@@ -51,6 +52,9 @@
 
 	changefeedID                 common.ChangeFeedID
 	metricConflictDetectDuration prometheus.Observer
+
+	admissionMu sync.Mutex
+	activeFence *Node
 }
 
 // New creates a new ConflictDetector.
@@ -99,6 +103,9 @@
 // NOTE: if multiple threads access this concurrently,
 // ConflictKeys must be sorted by the slot index.
 func (d *ConflictDetector) Add(event *commonEvent.DMLEvent) {
+	d.admissionMu.Lock()
+	defer d.admissionMu.Unlock()
+
 	start := time.Now()
 	hashes := ConflictKeys(event)
 	node := d.slots.AllocNode(hashes)
@@ -118,27 +125,88 @@
 	node.RandCacheID = func() int64 {
 		return d.nextCacheID.Add(1) % int64(len(d.resolvedTxnCaches))
 	}
-	node.OnNotified = func(callback func()) {
-		if !d.notifyGuardWaitGroup.AddIf(func() bool { return !d.notifyClosed.Load() }) {
-			return
-		}
-		defer d.notifyGuardWaitGroup.Done()
-
-		d.notifiedNodes.Push(callback)
-	}
-	d.slots.Add(node)
+	node.OnNotified = d.onNodeNotified
+	extraDependencies := d.activeFenceDependency()
+	d.slots.AddWithDependencies(node, extraDependencies)
 }
 
 // sendToCache should not call txn.Callback if it returns an error.
 func (d *ConflictDetector) sendToCache(event *commonEvent.DMLEvent, id int64) bool {
 	cache := d.resolvedTxnCaches[id]
-	ok := cache.add(event)
+	ok := cache.add(NewDMLItem(event))
 	return ok
 }
 
+// BroadcastBarrier installs a removal-only fence after all DMLs admitted so far
+// and broadcasts one barrier token to every writer queue after that fence resolves.
+func (d *ConflictDetector) BroadcastBarrier(barrier Barrier) error {
+	if d.notifyClosed.Load() {
+		return errors.ErrMySQLTxnError.GenWithStackByArgs("broadcast barrier on closed conflict detector")
+	}
+
+	d.admissionMu.Lock()
+
+	dependencyNodes := make(map[int64]*Node)
+	for _, node := range d.slots.SnapshotTailNodes() {
+		dependencyNodes[node.nodeID()] = node
+	}
+	for id, node := range d.activeFenceDependency() {
+		dependencyNodes[id] = node
+	}
+
+	fence := &Node{
+		id:                   genNextNodeID(),
+		assignedTo:           unassigned,
+		resolveByRemovalOnly: true,
+	}
+	fence.TrySendToTxnCache = func(cacheID) bool {
+		item := NewBarrierItem(barrier)
+		for _, cache := range d.resolvedTxnCaches {
+			if !cache.forceAdd(item) {
+				err := errors.ErrMySQLTxnError.GenWithStackByArgs("broadcast barrier to closed DML writer queue")
+				go barrier.Fail(err)
+				return true
+			}
+		}
+		return true
+	}
+	fence.RandCacheID = func() cacheID { return 0 }
+	fence.OnNotified = d.onNodeNotified
+
+	d.activeFence = fence
+	fence.dependOn(dependencyNodes)
+	d.admissionMu.Unlock()
+
+	barrier.OnDone(func() {
+		d.admissionMu.Lock()
+		if d.activeFence == fence {
+			d.activeFence = nil
+		}
+		d.admissionMu.Unlock()
+		fence.remove()
+	})
+	return nil
+}
+
+func (d *ConflictDetector) activeFenceDependency() map[int64]*Node {
+	if d.activeFence == nil {
+		return nil
+	}
+	return map[int64]*Node{d.activeFence.nodeID(): d.activeFence}
+}
+
+func (d *ConflictDetector) onNodeNotified(callback func()) {
+	if !d.notifyGuardWaitGroup.AddIf(func() bool { return !d.notifyClosed.Load() }) {
+		return
+	}
+	defer d.notifyGuardWaitGroup.Done()
+
+	d.notifiedNodes.Push(callback)
+}
+
 // GetOutChByCacheID returns the output channel by cacheID.
 // Note txns in single cache should be executed sequentially.
-func (d *ConflictDetector) GetOutChByCacheID(id int) *chann.UnlimitedChannel[*commonEvent.DMLEvent, any] {
+func (d *ConflictDetector) GetOutChByCacheID(id int) *chann.UnlimitedChannel[WriterItem, any] {
 	return d.resolvedTxnCaches[id].out()
 }
 

downstreamadapter/sink/mysql/causality/helper_test.go

@@ -26,6 +26,10 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func testChangefeedID() common.ChangeFeedID {
+	return common.NewChangefeedID4Test("test", "test")
+}
+
 func TestGenKeyListUsesSchemaIndexWithVirtualGeneratedColumn(t *testing.T) {
 	t.Parallel()
 

downstreamadapter/sink/mysql/causality/node.go

@@ -65,8 +65,9 @@ type Node struct {
 	// Following fields are protected by `mu`.
 	mu sync.Mutex
 
-	assignedTo cacheID
-	removed    bool
+	assignedTo           cacheID
+	removed              bool
+	resolveByRemovalOnly bool
 
 	// dependers is an ordered set for all nodes that
 	// conflict with the current node.
@@ -99,7 +100,7 @@ func (n *Node) dependOn(dependencyNodes map[int64]*Node) {
 		target.mu.Lock()
 		defer target.mu.Unlock()
 
-		if target.assignedTo != unassigned {
+		if target.assignedTo != unassigned && !n.resolveByRemovalOnly && !target.resolveByRemovalOnly {
 			// The target has already been assigned to a cache.
 			// In this case, record the cache ID in `resolvedList`, and this node
 			// probably can be sent to the same cache and executed sequentially.
@@ -170,9 +171,11 @@ func (n *Node) tryAssignTo(cacheID int64) bool {
 	if n.dependers != nil {
 		// `mu` must be holded during accessing dependers.
 		n.dependers.Ascend(func(node *Node) bool {
-			resolvedDependencies := atomic.AddInt32(&node.resolvedDependencies, 1)
-			atomic.StoreInt64(&node.resolvedList[resolvedDependencies-1], n.assignedTo)
-			node.OnNotified(node.maybeResolve)
+			if !n.resolveByRemovalOnly {
+				resolvedDependencies := atomic.AddInt32(&node.resolvedDependencies, 1)
+				atomic.StoreInt64(&node.resolvedList[resolvedDependencies-1], n.assignedTo)
+				node.OnNotified(node.maybeResolve)
+			}
 			return true
 		})
 	}
@@ -215,6 +218,10 @@ func (n *Node) tryResolve() (int64, bool) {
 		return assignedToAny, true
 	}
 
+	if n.resolveByRemovalOnly {
+		return unassigned, false
+	}
+
 	resolvedDependencies := atomic.LoadInt32(&n.resolvedDependencies)
 	if resolvedDependencies == n.totalDependencies {
 		firstDep := atomic.LoadInt64(&n.resolvedList[0])