eventservice: keep transaction cursor inside scan window

pkg/eventservice/event_broker.go

@@ -522,21 +522,22 @@ 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
 	}
 
 	if dataRange.CommitTsEnd <= dataRange.CommitTsStart {
 		// 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 {

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)
 }