From f06b1e9d4b6433bb678ba3359f7d9f0d150ade6b Mon Sep 17 00:00:00 2001
From: dongmen <414110582@qq.com>
Date: Wed, 19 Aug 2026 13:23:17 +0800
Subject: [PATCH] eventservice: keep transaction cursor inside scan window

---
 pkg/eventservice/event_broker.go      | 11 ++++++-----
 pkg/eventservice/event_broker_test.go | 10 +++++++---
 2 files changed, 13 insertions(+), 8 deletions(-)

diff --git a/pkg/eventservice/event_broker.go b/pkg/eventservice/event_broker.go
index 542556b88f..90db0d0163 100644
--- a/pkg/eventservice/event_broker.go
+++ b/pkg/eventservice/event_broker.go
@@ -522,13 +522,14 @@ func (c *eventBroker) getScanTaskRequestResult(task scanTask) scanTaskRequestRes
 		}
 	}
 
-	hasRowResume := len(request.Cursor.Position) != 0
-	// A published row cursor at C came from an earlier scan whose DDL and received
+	hasResumeCursor := request.Cursor.TxnStartTs != 0 || len(request.Cursor.Position) != 0
+	// A published cursor at C came from an earlier scan whose DDL and received
 	// resolved-ts bounds had already reached C. Since those bounds do not regress,
 	// only the adaptive scan window can move CommitTsEnd behind C. For example, if
 	// C=100 and the window caps the end at 80, restore the effective range to
-	// [100, 100] so scanning resumes after Position inside that transaction.
-	if hasRowResume && dataRange.CommitTsEnd < dataRange.CommitTsStart {
+	// [100, 100] so Position can resume rows inside a transaction, or TxnStartTs
+	// can resume later transactions sharing commit-ts C.
+	if hasResumeCursor && dataRange.CommitTsEnd < dataRange.CommitTsStart {
 		dataRange.CommitTsEnd = dataRange.CommitTsStart
 	}
 
@@ -536,7 +537,7 @@ func (c *eventBroker) getScanTaskRequestResult(task scanTask) scanTaskRequestRes
 		// A cursor makes [C, C] meaningful: Position resumes rows inside a
 		// transaction, while TxnStartTs resumes later transactions at the same C.
 		canResumeAtStart := dataRange.CommitTsEnd == dataRange.CommitTsStart &&
-			(hasRowResume || request.Cursor.TxnStartTs != 0)
+			hasResumeCursor
 		if canResumeAtStart || task.hasPendingLargeTxnState() {
 			result := scanTaskRequestResult{needScan: true, request: request}
 			if task.changefeedStat.lowLatencyMode {
diff --git a/pkg/eventservice/event_broker_test.go b/pkg/eventservice/event_broker_test.go
index 99498796a7..1032ab239e 100644
--- a/pkg/eventservice/event_broker_test.go
+++ b/pkg/eventservice/event_broker_test.go
@@ -760,7 +760,7 @@ func TestScanRangeCappedByScanWindow(t *testing.T) {
 	require.Equal(t, oracle.GoTimeToTS(baseTime.Add(defaultScanInterval)), result.request.Range.CommitTsEnd)
 }
 
-func TestGetScanTaskDataRangeEmptyAfterCappingDoesNotResetScanRange(t *testing.T) {
+func TestGetScanTaskRequestKeepsTxnCursorInsideShrunkWindow(t *testing.T) {
 	broker, _, _, _ := newEventBrokerForTest()
 	// Close the broker, so we can catch all message in the test.
 	broker.close()
@@ -785,8 +785,12 @@ func TestGetScanTaskDataRangeEmptyAfterCappingDoesNotResetScanRange(t *testing.T
 	changefeedStatus.minSentTs.Store(baseTs)
 	changefeedStatus.scanInterval.Store(int64(defaultScanInterval))
 
-	needScan, _ := broker.getScanTaskRequest(disp)
-	require.False(t, needScan)
+	needScan, request := broker.getScanTaskRequest(disp)
+	require.True(t, needScan)
+	require.Equal(t, commitStart, request.Range.CommitTsStart)
+	require.Equal(t, commitStart, request.Range.CommitTsEnd)
+	require.Equal(t, lastStartTs, request.Cursor.TxnStartTs)
+	require.Empty(t, request.Cursor.Position)
 	require.Equal(t, commitStart, disp.loadScanProgress().txnCommitTs)
 	require.Equal(t, lastStartTs, disp.loadScanProgress().txnStartTs)
 }