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

Code before:
        let peer = ctx.peer();
        ctx.spawn(async move {
            if let Err(e) = recv_req.await {
                warn!("cdc receive failed"; "error" => ?e, "downstream" => peer, "conn_id" => ?conn_id);
            } else {
                info!("cdc receive closed"; "downstream" => peer, "conn_id" => ?conn_id);
            }

            let deregister = Deregister::Conn(conn_id);
            if let Err(e) = scheduler_dereg.schedule(Task::Deregister(deregister)) {
                error!("cdc deregister failed"; "error" => ?e, "conn_id" => ?conn_id);
            }
        });

        // 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(),

Review comment (suggestion):
It looks that `tikv_util::future::paired_future_callback();` can be used to send signal, it's used in the tikv-cdc.


Code after:

        let scheduler_dereg = self.scheduler.clone();
        let peer = ctx.peer();
        ctx.spawn(async move {
            if let Err(e) = recv_req.await {
                warn!("cdc receive failed"; "error" => ?e, "downstream" => peer, "conn_id" => ?conn_id);
            } else {
                info!("cdc receive closed"; "downstream" => peer, "conn_id" => ?conn_id);
            }

            let deregister = Deregister::Conn(conn_id);
            if let Err(e) = scheduler_dereg.schedule(Task::Deregister(deregister)) {
                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(