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

Code before:
        ctx.spawn(async move {
            #[cfg(feature = "failpoints")]
            sleep_before_drain_change_event().await;
            tokio::select! {
                _ = &mut cancel_rx => {
                    warn!("cdc send cancelled"; "downstream" => peer, "conn_id" => ?conn_id);
                    let status = RpcStatus::with_message(RpcStatusCode::UNKNOWN, "connection cancelled".to_string());
                    let _ = sink.fail(status).await;
                    return;
                }
                result = event_drain.forward(&mut sink, Some(&last_flush_time_for_forward)) => {
                    if let Err(e) = result {
                        warn!("cdc send failed"; "error" => ?e, "downstream" => peer, "conn_id" => ?conn_id);
                    } else {
                        info!("cdc send closed"; "downstream" => peer, "conn_id" => ?conn_id);
                    }
                    last_flush_time_for_forward.store(0, Ordering::Relaxed);
                }
            }
        });

        // Add watchdog loop to monitor connection activity using tokio runtime
        let pool = self.pool.clone();
        pool.spawn(async move {
            use std::time::{SystemTime, UNIX_EPOCH, Duration};
            use tokio::time::interval;

            // Create an interval that ticks every CDC_WATCHDOG_CHECK_INTERVAL_SECS
            let mut interval = interval(Duration::from_secs(CDC_WATCHDOG_CHECK_INTERVAL_SECS));

            loop {
                interval.tick().await;
                
                let current_time = match SystemTime::now().duration_since(UNIX_EPOCH) {
                    Ok(duration) => duration.as_secs(),
                    Err(_) => continue,
                };
                
                let last_flush = last_flush_time_for_watchdog.load(Ordering::Relaxed);
                let time_since_last_flush = current_time.saturating_sub(last_flush);

                if last_flush == 0 {
                    info!("cdc connection is closed, exiting watchdog loop"; "downstream" => peer_for_watchdog.clone(), "conn_id" => ?conn_id);
                }
                
                // Check if last flush was more than the warning threshold
                if time_since_last_flush > CDC_IDLE_WARNING_THRESHOLD_SECS {
                    warn!("cdc connection idle too long"; 
                          "downstream" => peer_for_watchdog.clone(), 
                          "conn_id" => ?conn_id, 
                          "seconds_since_last_flush" => time_since_last_flush);

Review comment (suggestion):
It's better to place use statements at the top of the function or module rather than inside a block.

Code after:
            sleep_before_drain_change_event().await;
            tokio::select! {
                _ = &mut cancel_rx => {
                    warn!("cdc send cancelled"; "downstream" => peer, "conn_id" => ?conn_id);
                    let status = RpcStatus::with_message(RpcStatusCode::UNKNOWN, "connection cancelled".to_string());
                    let _ = sink.fail(status).await;
                    CDC_ABORTED_CONNECTIONS.inc();
                }
                result = event_drain.forward(&mut sink, Some(&last_flush_time_for_forward)) => {
                    if let Err(e) = result {
                        warn!("cdc send failed"; "error" => ?e, "downstream" => peer, "conn_id" => ?conn_id);
                    } else {
                        info!("cdc send closed"; "downstream" => peer, "conn_id" => ?conn_id);
                    }
                    // Send signal when eventDrain.forward exits
                    let _ = forward_exit_tx.send(());
                }
            }
        });

        // Start watchdog to monitor connection activity
        Self::start_connection_watchdog(
            self.pool.clone(),
            last_flush_time_for_watchdog.clone(),
            peer_for_watchdog.clone(),
            conn_id,
            cancel_tx,
            forward_exit_rx,
        );
    }

    /// Start a watchdog to monitor CDC connection activity.
    ///
    /// This function creates a background task that periodically checks if the
    /// connection has been idle for too long and takes appropriate action
    /// (warning or aborting).
    fn start_connection_watchdog(
        pool: Arc<Worker>,
        last_flush_time: Arc<AtomicCell<Instant>>,
        peer: String,
        conn_id: ConnId,
        cancel_tx: tokio::sync::oneshot::Sender<()>,
        mut forward_exit_rx: tokio::sync::oneshot::Receiver<()>,
    ) {
        let last_flush_time_clone = last_flush_time.clone();
        let peer_clone = peer.clone();

        // Create a custom interval task that can be stopped
        let _ = pool.pool().spawn(async move {
            let mut interval = GLOBAL_TIMER_HANDLE
                .interval(