Repository: tikv/tikv
File: src/server/raft_client.rs (Rust)

Code before:
}

struct RaftCall<R, M, B> {
    sender: AsyncRaftSender<R, M, B>,
    receiver: ClientCStreamReceiver<Done>,
    lifetime: Option<oneshot::Sender<RaftCallRes>>,
    store_id: u64,
}

impl<R, M, B> RaftCall<R, M, B>
where
    R: RaftExtension + Unpin + 'static,
    B: Buffer<OutputMessage = M> + Unpin,
{
    async fn poll(&mut self) {
        let res = futures::join!(&mut self.sender, &mut self.receiver);
        if let (Ok(()), Ok(Done { .. })) = res {
            info!("connection close"; "store_id" => self.store_id, "addr" => %self.sender.addr);
            if let Some(tx) = self.lifetime.take() {
                let _ = tx.send(RaftCallRes::Disconnected);
            }
            return;
        }

        let (sink_err, recv_err) = (res.0.err(), res.1.err());
        warn!("connection aborted"; "store_id" => self.store_id, "sink_error" => ?sink_err, "receiver_err" => ?recv_err, "addr" => %self.sender.addr);
        if let Some(tx) = self.lifetime.take() {
            let should_fallback = [sink_err, recv_err]
                .iter()
                .any(|e| e.as_ref().is_some_and(grpc_error_is_unimplemented));

            let res = if should_fallback {
                // Asks backend to fallback.
                RaftCallRes::Fallback
            } else {
                RaftCallRes::Disconnected
            };
            let _ = tx.send(res);
        }
    }
}

#[derive(Clone)]
pub struct ConnectionBuilder<S, R> {
    env: Arc<Environment>,
    cfg: Arc<VersionTrack<Config>>,
    security_mgr: Arc<SecurityManager>,
    resolver: S,
    router: R,
    snap_scheduler: Scheduler<SnapTask>,
    loads: Arc<ThreadLoadPool>,

Review comment (suggestion):
```suggestion
        warn!(
            "connection aborted";
            "store_id" => self.store_id,
            "sink_error" => ?sink_err,
            "receiver_err" => ?recv_err,
            "addr" => %self.sender.addr
        );
```

Code after:
}

struct RaftCall<R, M, B> {
    sender: AsyncRaftSender<R, M, B>,
    receiver: ClientCStreamReceiver<Done>,
    lifetime: Option<oneshot::Sender<RaftCallRes>>,
    store_id: u64,
}

impl<R, M, B> RaftCall<R, M, B>
where
    R: RaftExtension + Unpin + 'static,
    B: Buffer<OutputMessage = M> + Unpin,
{
    async fn poll(&mut self) {
        let res = futures::join!(&mut self.sender, &mut self.receiver);
        if let (Ok(()), Ok(Done { .. })) = res {
            info!("connection close"; "store_id" => self.store_id, "addr" => %self.sender.addr);
            if let Some(tx) = self.lifetime.take() {
                let _ = tx.send(RaftCallRes::Disconnected);
            }
            return;
        }

        let (sink_err, recv_err) = (res.0.err(), res.1.err());
        warn!(
            "connection aborted";
            "store_id" => self.store_id,
            "sink_error" => ?sink_err,
            "receiver_err" => ?recv_err,
            "addr" => %self.sender.addr,
        );
        if let Some(tx) = self.lifetime.take() {
            let should_fallback = [sink_err, recv_err]
                .iter()
                .any(|e| e.as_ref().is_some_and(grpc_error_is_unimplemented));

            let res = if should_fallback {
                // Asks backend to fallback.
                RaftCallRes::Fallback
            } else {
                RaftCallRes::Disconnected
            };
            let _ = tx.send(res);
        }
    }
}

#[derive(Clone)]
pub struct ConnectionBuilder<S, R> {
    env: Arc<Environment>,