logpuller: add region runtime registry scaffold

logservice/logpuller/region_event_handler.go

@@ -252,6 +252,7 @@ func (h *regionEventHandler) handleRegionError(state *regionFeedState) {
 
 func handleEventEntries(span *subscribedSpan, state *regionFeedState, entries *cdcpb.Event_Entries_) {
 	regionID, _, _ := state.getRegionMeta()
+	state.updateRuntimeLastEvent(time.Now())
 	assembleRowEvent := func(regionID uint64, entry *cdcpb.Event_Row) common.RawKVEntry {
 		var opType common.OpType
 		switch entry.GetOpType() {
@@ -277,6 +278,7 @@ func handleEventEntries(span *subscribedSpan, state *regionFeedState, entries *c
 		switch entry.Type {
 		case cdcpb.Event_INITIALIZED:
 			state.setInitialized()
+			state.markRuntimeReplicating(time.Now())
 			log.Debug("region is initialized",
 				zap.Int64("tableID", span.span.TableID),
 				zap.Uint64("regionID", regionID),
@@ -364,6 +366,7 @@ func handleResolvedTs(span *subscribedSpan, state *regionFeedState, resolvedTs u
 	}
 
 	state.updateResolvedTs(resolvedTs)
+	state.updateRuntimeResolvedTs(resolvedTs, time.Now())
 
 	ts := uint64(0)
 	shouldAdvance := false

logservice/logpuller/region_request_worker.go

@@ -61,6 +61,25 @@ type regionRequestWorker struct {
 	}
 }
 
+func (s *regionRequestWorker) runtimeRegistry() *regionRuntimeRegistry {
+	if s.client == nil {
+		return nil
+	}
+	return s.client.regionRuntimeRegistry
+}
+
+func (s *regionRequestWorker) markRegionRuntimeEnqueued(region regionInfo, now time.Time) {
+	if registry := s.runtimeRegistry(); registry != nil && region.runtimeKey.isValid() {
+		registry.setRequestEnqueueTime(region.runtimeKey, now)
+	}
+}
+
+func (s *regionRequestWorker) markRegionRuntimeSent(region regionInfo, now time.Time) {
+	if registry := s.runtimeRegistry(); registry != nil && region.runtimeKey.isValid() {
+		registry.markWaitInitialized(region.runtimeKey, s.workerID, now)
+	}
+}
+
 func newRegionRequestWorker(
 	ctx context.Context,
 	client *subscriptionClient,
@@ -431,6 +450,7 @@ func (s *regionRequestWorker) processRegionSendTask(
 			// sentRequests visible in the same order and avoids leaving stale
 			// requests in cleanup.
 			s.requestCache.markSent(regionReq)
+			s.markRegionRuntimeSent(region, time.Now())
 			if err := doSend(s.createRegionRequest(region)); err != nil {
 				state.markStopped(err)
 				return err
@@ -511,7 +531,11 @@ func (s *regionRequestWorker) clearRegionStates() map[SubscriptionID]regionFeedS
 // add adds a region request to the worker's cache
 // It blocks if the cache is full until there's space or ctx is cancelled
 func (s *regionRequestWorker) add(ctx context.Context, region regionInfo, force bool) (bool, error) {
-	return s.requestCache.add(ctx, region, force)
+	ok, err := s.requestCache.add(ctx, region, force)
+	if ok && err == nil {
+		s.markRegionRuntimeEnqueued(region, time.Now())
+	}
+	return ok, err
 }
 
 func (s *regionRequestWorker) clearPendingRegions() []regionInfo {