Repository: tikv/tikv
File: components/cdc/src/delegate.rs (Rust)

Code before:
            send(downstream);
        }
    }

    /// `txn_extra_op` returns a shared flag which is accessed in TiKV's
    /// transaction layer to determine whether to capture modifications' old
    /// value or not. Unsubscribing all downstreams or calling
    /// `Delegate::stop` will store it with `TxnExtraOp::Noop`.
    ///
    /// NOTE: Dropping a `Delegate` won't update this flag.
    pub fn txn_extra_op(&self) -> &AtomicCell<TxnExtraOp> {
        self.txn_extra_op.as_ref()
    }

    /// Try advance and broadcast resolved ts.
    pub(crate) fn on_min_ts(
        &mut self,
        min_ts: TimeStamp,
        current_ts: TimeStamp,
        connections: &HashMap<ConnId, Conn>,
        advance: &mut Advance,
    ) {
        let locks = match &self.lock_tracker {
            LockTracker::Prepared { locks, .. } => locks,
            _ => {
                advance.blocked_on_scan += 1;
                let now = Instant::now_coarse();
                let elapsed = now.duration_since(self.created);
                if elapsed > WARN_LAG_THRESHOLD
                    && now.duration_since(self.last_lag_warn) > WARN_LAG_INTERVAL
                {
                    warn!(
                        "cdc region scan locks too slow";
                        "region_id" => self.region_id,
                        "elapsed" => ?elapsed,
                        "stage" => ?self.lock_tracker,
                    );
                    self.last_lag_warn = now;
                }
                return;
            }
        };

        let mut handle_downstream = |downstream: &mut Downstream| -> Option<TimeStamp> {
            if !downstream.state.load().ready_for_advancing_ts() {
                advance.blocked_on_scan += 1;
                return None;
            }
            advance.scan_finished += 1;

            if downstream.lock_heap.is_none() {

Review comment (suggestion):
It's better to be `advance.blocked_on_scan += self.downstreams.len()`.

We can treat all downstreams as uninitialized.

Or we can iterate downstreams to update `advance.blocked_on_scan`.

Code after:
            send(downstream);
        }
    }

    /// `txn_extra_op` returns a shared flag which is accessed in TiKV's
    /// transaction layer to determine whether to capture modifications' old
    /// value or not. Unsubscribing all downstreams or calling
    /// `Delegate::stop` will store it with `TxnExtraOp::Noop`.
    ///
    /// NOTE: Dropping a `Delegate` won't update this flag.
    pub fn txn_extra_op(&self) -> &AtomicCell<TxnExtraOp> {
        self.txn_extra_op.as_ref()
    }

    /// Try advance and broadcast resolved ts.
    pub(crate) fn on_min_ts(
        &mut self,
        min_ts: TimeStamp,
        current_ts: TimeStamp,
        connections: &HashMap<ConnId, Conn>,
        advance: &mut Advance,
    ) {
        let locks = match &self.lock_tracker {
            LockTracker::Prepared { locks, .. } => locks,
            _ => {
                advance.blocked_on_scan += self.downstreams.len();
                let now = Instant::now_coarse();
                let elapsed = now.duration_since(self.created);
                if elapsed > WARN_LAG_THRESHOLD
                    && now.duration_since(self.last_lag_warn) > WARN_LAG_INTERVAL
                {
                    warn!(
                        "cdc region scan locks too slow";
                        "region_id" => self.region_id,
                        "elapsed" => ?elapsed,
                        "stage" => ?self.lock_tracker,
                    );
                    self.last_lag_warn = now;
                }
                return;
            }
        };

        let mut handle_downstream = |downstream: &mut Downstream| -> Option<TimeStamp> {
            if !downstream.state.load().ready_for_advancing_ts() {
                advance.blocked_on_scan += 1;
                return None;
            }
            advance.scan_finished += 1;

            if downstream.lock_heap.is_none() {