[WIP] memory quota optimization

logservice/eventstore/event_store.go

@@ -460,7 +460,7 @@ func (e *eventStore) Close(_ context.Context) error {
 }
 
 func (e *eventStore) RegisterDispatcher(
-	_ common.ChangeFeedID,
+	changefeedID common.ChangeFeedID,
 	dispatcherID common.DispatcherID,
 	dispatcherSpan *heartbeatpb.TableSpan,
 	startTs uint64,
@@ -678,7 +678,15 @@ func (e *eventStore) RegisterDispatcher(
 	serverConfig := config.GetGlobalServerConfig()
 	resolvedTsAdvanceInterval := int64(serverConfig.KVClient.AdvanceIntervalInMs)
 	// Note: don't hold any lock when call Subscribe
-	e.subClient.Subscribe(subStat.subID, *dispatcherSpan, startTs, consumeKVEvents, advanceResolvedTs, resolvedTsAdvanceInterval, bdrMode)
+	e.subClient.Subscribe(
+		subStat.subID,
+		logpuller.NewChangefeedSubscriptionMeta(changefeedID),
+		*dispatcherSpan,
+		startTs,
+		consumeKVEvents,
+		advanceResolvedTs,
+		resolvedTsAdvanceInterval,
+		bdrMode)
 	log.Info("new subscription created",
 		zap.Stringer("dispatcherID", dispatcherID),
 		zap.Uint64("startTs", startTs),

logservice/eventstore/event_store_test.go

@@ -121,13 +121,15 @@ func (s *mockSubscriptionClient) AllocSubscriptionID() logpuller.SubscriptionID
 
 func (s *mockSubscriptionClient) Subscribe(
 	subID logpuller.SubscriptionID,
+	meta logpuller.SubscriptionMeta,
 	span heartbeatpb.TableSpan,
 	startTs uint64,
 	consumeKVEvents func(raw []common.RawKVEntry, wakeCallback func()) bool,
 	advanceResolvedTs func(ts uint64),
 	advanceInterval int64,
 	bdrMode bool,
 ) {
+	_ = meta
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.subscriptions[subID] = &mockSubscriptionStat{

logservice/logpuller/errors.go

@@ -38,6 +38,17 @@ func (e *rpcCtxUnavailableErr) Error() string {
 		e.verID.GetID(), e.verID.GetVer(), e.verID.GetConfVer())
 }
 
+type rpcCtxChangedError struct {
+	verID tikv.RegionVerID
+	from  string
+	to    string
+}
+
+func (e *rpcCtxChangedError) Error() string {
+	return fmt.Sprintf("rpcCtx for region %v changed from %s to %s. ver:%v, confver:%v",
+		e.verID.GetID(), e.from, e.to, e.verID.GetVer(), e.verID.GetConfVer())
+}
+
 type getStoreErr struct{}
 
 func (e *getStoreErr) Error() string { return "get store error" }