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

Code before:
        });

        // Create atomic variable to track last flush time
        let last_flush_time = Arc::new(AtomicCell::new(Instant::now()));
        let last_flush_time_for_forward = last_flush_time.clone();
        let last_flush_time_for_watchdog = last_flush_time.clone();
        let peer_for_watchdog = ctx.peer().to_string();

        let peer = ctx.peer();
        let (cancel_tx, mut cancel_rx) = tokio::sync::oneshot::channel::<()>();

        // Create cancelCh for eventDrain.forward exit signal
        let (forward_exit_tx, forward_exit_rx) = tokio::sync::oneshot::channel::<()>();
        let forward_exit_tx_for_forward = forward_exit_tx;

        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;
                }
                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_for_forward.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

Review comment (suggestion):
It looks that we can only let the forward exit, and then the drain is dropped to release the memory quota.

recv_req cannot returned if the gRPC-rs pool is stuck.

Code after:
                error!("cdc deregister failed"; "error" => ?e, "conn_id" => ?conn_id);
            }
        });

        let last_flush_time = Arc::new(AtomicCell::new(Instant::now()));
        let last_flush_time_for_forward = last_flush_time.clone();
        let last_flush_time_for_watchdog = last_flush_time.clone();
        let peer_for_watchdog = ctx.peer().to_string();

        let peer = ctx.peer();
        let (cancel_tx, mut cancel_rx) = tokio::sync::oneshot::channel::<()>();

        // Create cancelCh for eventDrain.forward exit signal
        let (forward_exit_tx, forward_exit_rx) = tokio::sync::oneshot::channel::<()>();

        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;
                    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