logpuller: fix stale sent request after fast region error (#4557)

close #4472

Author: lidezhu

logservice/logpuller/region_req_cache.go

@@ -191,6 +191,9 @@ func (c *requestCache) markStopped(subID SubscriptionID, regionID uint64) {
 	}
 
 	delete(regionReqs, regionID)
+	if len(regionReqs) == 0 {
+		delete(c.sentRequests.regionReqs, subID)
+	}
 	c.markDone()
 }
 

logservice/logpuller/region_req_cache_test.go

@@ -311,3 +311,26 @@ func TestRequestCacheMarkSent_DuplicateReleaseSlot(t *testing.T) {
 	require.True(t, cache.resolve(region.subscribedSpan.subID, region.verID.GetID()))
 	require.Equal(t, 0, cache.getPendingCount())
 }
+
+func TestRequestCacheMarkStopped_ReleasesSlot(t *testing.T) {
+	cache := newRequestCache(10)
+	ctx := context.Background()
+
+	region := createTestRegionInfo(1, 1)
+
+	ok, err := cache.add(ctx, region, false)
+	require.True(t, ok)
+	require.NoError(t, err)
+	require.Equal(t, 1, cache.getPendingCount())
+
+	req, err := cache.pop(ctx)
+	require.NoError(t, err)
+
+	cache.markSent(req)
+	require.Equal(t, 1, cache.getPendingCount())
+	require.Contains(t, cache.sentRequests.regionReqs, req.regionInfo.subscribedSpan.subID)
+
+	cache.markStopped(req.regionInfo.subscribedSpan.subID, req.regionInfo.verID.GetID())
+	require.Equal(t, 0, cache.getPendingCount())
+	require.NotContains(t, cache.sentRequests.regionReqs, req.regionInfo.subscribedSpan.subID)
+}

logservice/logpuller/region_request_worker.go

@@ -406,11 +406,25 @@ func (s *regionRequestWorker) processRegionSendTask(
 			state := newRegionFeedState(region, uint64(subID), s)
 			state.start()
 			s.addRegionState(subID, region.verID.GetID(), state)
+			// Mark the request as sent before sending it.
+			// Otherwise there is a race with the receiver goroutine:
+			//  1. addRegionState makes the region visible to error handling.
+			//  2. doSend sends the request.
+			//  3. the receiver goroutine may receive a region error immediately.
+			//  4. markStopped runs before markSent, so requestCache.markStopped cannot
+			//     find the request in sentRequests.
+			//  5. the sender goroutine then calls markSent and leaves a stale sent
+			//     request behind, even though the region has already been
+			//     unlocked/rescheduled.
+			//
+			// Tracking the request before Send keeps requestedRegions and
+			// sentRequests visible in the same order and avoids leaving stale
+			// requests in cleanup.
+			s.requestCache.markSent(regionReq)
 			if err := doSend(s.createRegionRequest(region)); err != nil {
-				s.requestCache.markDone()
+				state.markStopped(err)
 				return err
 			}
-			s.requestCache.markSent(regionReq)
 		}
 		regionReq, err = fetchMoreReq()
 		if err != nil {

logservice/logpuller/region_request_worker_test.go

@@ -17,9 +17,40 @@ import (
 	"context"
 	"testing"
 
+	"github.com/pingcap/errors"
+	"github.com/pingcap/kvproto/pkg/cdcpb"
+	"github.com/pingcap/kvproto/pkg/metapb"
+	"github.com/pingcap/ticdc/logservice/logpuller/regionlock"
 	"github.com/stretchr/testify/require"
+	"github.com/tikv/client-go/v2/tikv"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
 )
 
+type mockEventFeedV2Client struct {
+	sendErr error
+}
+
+func (m *mockEventFeedV2Client) Send(*cdcpb.ChangeDataRequest) error   { return m.sendErr }
+func (m *mockEventFeedV2Client) Recv() (*cdcpb.ChangeDataEvent, error) { return nil, nil }
+func (m *mockEventFeedV2Client) Header() (metadata.MD, error)          { return metadata.MD{}, nil }
+func (m *mockEventFeedV2Client) Trailer() metadata.MD                  { return metadata.MD{} }
+func (m *mockEventFeedV2Client) CloseSend() error                      { return nil }
+func (m *mockEventFeedV2Client) Context() context.Context              { return context.Background() }
+func (m *mockEventFeedV2Client) SendMsg(any) error                     { return nil }
+func (m *mockEventFeedV2Client) RecvMsg(any) error                     { return nil }
+
+func prepareRegionForSendTest(region regionInfo) regionInfo {
+	region.rpcCtx = &tikv.RPCContext{
+		Meta: &metapb.Region{
+			RegionEpoch: &metapb.RegionEpoch{Version: 1, ConfVer: 1},
+		},
+	}
+	region.lockedRangeState = &regionlock.LockedRangeState{}
+	region.lockedRangeState.ResolvedTs.Store(100)
+	return region
+}
+
 func TestRegionStatesOperation(t *testing.T) {
 	worker := &regionRequestWorker{}
 	worker.requestedRegions.subscriptions = make(map[SubscriptionID]regionFeedStates)
@@ -64,3 +95,71 @@ func TestClearPendingRegionsReleaseSlotForPreFetchedRegion(t *testing.T) {
 	require.Nil(t, worker.preFetchForConnecting)
 	require.Equal(t, 0, worker.requestCache.getPendingCount())
 }
+
+func TestClearPendingRegionsDoesNotReturnStoppedSentRegion(t *testing.T) {
+	worker := &regionRequestWorker{
+		requestCache: newRequestCache(10),
+	}
+	worker.requestedRegions.subscriptions = make(map[SubscriptionID]regionFeedStates)
+
+	ctx := context.Background()
+	region := createTestRegionInfo(1, 1)
+
+	ok, err := worker.requestCache.add(ctx, region, false)
+	require.NoError(t, err)
+	require.True(t, ok)
+
+	req, err := worker.requestCache.pop(ctx)
+	require.NoError(t, err)
+
+	state := newRegionFeedState(req.regionInfo, uint64(req.regionInfo.subscribedSpan.subID), worker)
+	state.start()
+	worker.addRegionState(req.regionInfo.subscribedSpan.subID, req.regionInfo.verID.GetID(), state)
+
+	// Simulate the race we are fixing in processRegionSendTask:
+	// once a request is visible in sentRequests, a fast region error may mark the
+	// region stopped before worker cleanup runs. In that case, markStopped should
+	// remove the sent request immediately, so clearPendingRegions must not return
+	// the stale region again during worker shutdown.
+	worker.requestCache.markSent(req)
+	state.markStopped(errors.New("send request to store error"))
+	worker.takeRegionState(req.regionInfo.subscribedSpan.subID, req.regionInfo.verID.GetID())
+
+	require.Equal(t, 0, worker.requestCache.getPendingCount())
+	require.Empty(t, worker.clearPendingRegions())
+}
+
+func TestProcessRegionSendTaskSendFailureCleansSentRequest(t *testing.T) {
+	worker := &regionRequestWorker{
+		requestCache: newRequestCache(10),
+		store:        &requestedStore{storeAddr: "store-1"},
+		client:       &subscriptionClient{},
+	}
+	worker.requestedRegions.subscriptions = make(map[SubscriptionID]regionFeedStates)
+
+	ctx := context.Background()
+	region := prepareRegionForSendTest(createTestRegionInfo(1, 1))
+
+	ok, err := worker.requestCache.add(ctx, region, false)
+	require.NoError(t, err)
+	require.True(t, ok)
+	require.Equal(t, 1, worker.requestCache.getPendingCount())
+
+	req, err := worker.requestCache.pop(ctx)
+	require.NoError(t, err)
+	worker.preFetchForConnecting = new(regionInfo)
+	*worker.preFetchForConnecting = req.regionInfo
+
+	sendErr := errors.New("send failed")
+	conn := &ConnAndClient{
+		Client: &mockEventFeedV2Client{sendErr: sendErr},
+		Conn:   &grpc.ClientConn{},
+	}
+
+	err = worker.processRegionSendTask(ctx, conn)
+	require.ErrorIs(t, err, sendErr)
+	require.Equal(t, 0, worker.requestCache.getPendingCount())
+	require.Empty(t, worker.requestCache.sentRequests.regionReqs)
+	state := worker.getRegionState(req.regionInfo.subscribedSpan.subID, req.regionInfo.verID.GetID())
+	require.True(t, state == nil || state.isStale(), "region state should be removed or marked stale after send failure")
+}