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

Code before:
            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(
                    std::time::Instant::now(),
                    Duration::from_secs(CDC_WATCHDOG_CHECK_INTERVAL_SECS),
                )
                .compat();

            loop {
                tokio::select! {
                    _ = &mut forward_exit_rx => {
                        info!("cdc connection forward exit signal received, stopping watchdog");
                        break;
                    }
                    _ = interval.next() => {
                        let elapsed = last_flush_time_clone.load().elapsed();

                        // Check if last flush was more than the warning threshold
                        if elapsed > Duration::from_secs(CDC_IDLE_WARNING_THRESHOLD_SECS) {
                            warn!("cdc connection idle too long";
                                  "downstream" => peer_clone.clone(),
                                  "conn_id" => ?conn_id,
                                  "seconds_since_last_flush" => elapsed.as_secs());
                        }

                        let _idle_threshold = CDC_IDLE_DEREGISTER_THRESHOLD_SECS;

                        #[cfg(feature = "failpoints")]
                        let _idle_threshold = {

Review comment (suggestion):
```suggestion
                    Instant::now(),
```

Code after:
            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(
                    Instant::now(),
                    Duration::from_secs(CDC_WATCHDOG_CHECK_INTERVAL_SECS),
                )
                .compat();

            loop {
                tokio::select! {
                    _ = &mut forward_exit_rx => {
                        info!("cdc connection forward exit signal received, stopping watchdog");
                        break;
                    }
                    _ = interval.next() => {
                        let elapsed = last_flush_time_clone.load().elapsed();

                        // Check if last flush was more than the warning threshold
                        if elapsed > Duration::from_secs(CDC_IDLE_WARNING_THRESHOLD_SECS) {
                            warn!("cdc connection idle too long";
                                  "downstream" => peer_clone.clone(),
                                  "conn_id" => ?conn_id,
                                  "seconds_since_last_flush" => elapsed.as_secs());
                        }

                        let _idle_threshold = CDC_IDLE_DEREGISTER_THRESHOLD_SECS;

                        #[cfg(feature = "failpoints")]