[WIP] remove use buffer

logservice/logpuller/region_event_sink.go

@@ -39,9 +39,6 @@ func newRegionEventSink(
 	option := dynstream.NewOption()
 	// Note: it is max batch size of the kv sent from tikv(not committed rows)
 	option.BatchCount = 1024
-	// TODO: Set `UseBuffer` to true until we refactor the `regionEventHandler.Handle` method so that it doesn't call any method of the dynamic stream. Currently, if `UseBuffer` is set to false, there will be a deadlock:
-	// 	ds.handleLoop fetch events from `ch` -> regionEventHandler.Handle -> ds.RemovePath -> send event to `ch`
-	option.UseBuffer = true
 	ds := dynstream.NewParallelDynamicStream(
 		"log-puller",
 		&regionEventHandler{eventSink: sink, failureHandler: failureHandler},

logservice/logpuller/region_failure_handler.go

@@ -194,7 +194,8 @@ func (r *regionFailureHandler) Report(errInfo regionErrorInfo) {
 	if errInfo.subscribedSpan.rangeLock.UnlockRange(
 		errInfo.span.StartKey, errInfo.span.EndKey,
 		errInfo.verID.GetID(), errInfo.verID.GetVer(), errInfo.resolvedTs()) {
-		r.onTableDrained(errInfo.subscribedSpan)
+		// Defer span cleanup to Run so Report never calls back into dynstream.
+		r.cache.addDrainedSpan(errInfo.subscribedSpan)
 		return
 	}
 	r.cache.add(errInfo)
@@ -206,6 +207,9 @@ func (r *regionFailureHandler) Run(ctx context.Context) error {
 	defer r.cancelRecoveries()
 
 	handleCachedErrors := func() error {
+		for _, span := range r.cache.popDrainedSpans() {
+			r.onTableDrained(span)
+		}
 		for {
 			batch := r.cache.popBatch(errCacheBatchSize)
 			for _, errInfo := range batch {
@@ -383,8 +387,9 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr
 
 type errCache struct {
 	sync.Mutex
-	cache  []regionErrorInfo
-	notify chan struct{}
+	cache        []regionErrorInfo
+	drainedSpans []*subscribedSpan
+	notify       chan struct{}
 }
 
 const errCacheBatchSize = 1024
@@ -400,12 +405,31 @@ func (e *errCache) add(errInfo regionErrorInfo) {
 	e.Lock()
 	defer e.Unlock()
 	e.cache = append(e.cache, errInfo)
+	e.signal()
+}
+
+func (e *errCache) addDrainedSpan(span *subscribedSpan) {
+	e.Lock()
+	defer e.Unlock()
+	e.drainedSpans = append(e.drainedSpans, span)
+	e.signal()
+}
+
+func (e *errCache) signal() {
 	select {
 	case e.notify <- struct{}{}:
 	default:
 	}
 }
 
+func (e *errCache) popDrainedSpans() []*subscribedSpan {
+	e.Lock()
+	defer e.Unlock()
+	drainedSpans := e.drainedSpans
+	e.drainedSpans = nil
+	return drainedSpans
+}
+
 func (e *errCache) popBatch(limit int) []regionErrorInfo {
 	e.Lock()
 	defer e.Unlock()

logservice/logpuller/region_request_scheduler_test.go

@@ -207,6 +207,10 @@ func TestRegionRequestSchedulerSkipsStoppedSubscriptionBeforeCreatingStore(t *te
 	handler := newRegionFailureHandler(nil, func(rt *subscribedSpan) {
 		drainedCh <- rt
 	}, nil, nil)
+	handlerErrCh := make(chan error, 1)
+	go func() {
+		handlerErrCh <- handler.Run(ctx)
+	}()
 	scheduler := &regionRequestScheduler{
 		upstream: &upstreamHandle{
 			pd:          pdClient,
@@ -244,4 +248,5 @@ func TestRegionRequestSchedulerSkipsStoppedSubscriptionBeforeCreatingStore(t *te
 
 	cancel()
 	require.ErrorIs(t, <-errCh, context.Canceled)
+	require.ErrorIs(t, <-handlerErrCh, context.Canceled)
 }

logservice/logpuller/subscription_client_test.go

@@ -393,7 +393,25 @@ func TestRegionFailureHandlerQueuesCanceledError(t *testing.T) {
 	}, &requestCancelledErr{}))
 
 	require.Len(t, client.failureHandler.cache.cache, 1)
-	require.Nil(t, client.spanRegistry.Get(span.subID))
+	require.Same(t, span, client.spanRegistry.Get(span.subID))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	runDone := make(chan error, 1)
+	go func() {
+		runDone <- client.failureHandler.Run(ctx)
+	}()
+
+	require.Eventually(t, func() bool {
+		return client.spanRegistry.Get(span.subID) == nil
+	}, time.Second, 10*time.Millisecond)
+	cancel()
+	select {
+	case err := <-runDone:
+		require.ErrorIs(t, err, context.Canceled)
+	case <-time.After(time.Second):
+		t.Fatal("failure handler did not exit after context cancellation")
+	}
 }
 
 type mockDynamicStream struct{}