[Bug]: EventListener will lose event since it's not cancellation safe.

[Phoenix500526]

Description about the bug

When using select! in a loop to receive messages from multiple sources, we should make sure that the receive call is cancellation safe to avoid losing messages. Unfortunately, EventListener is not cancellation safe. That means some implementation in our code may lose event due to the select! cancel it.

Here is the task method in watch_server.rs, and its implementation lead to events loss.

    async fn task<ST, W>(
        next_id_gen: Arc<WatchIdGenerator>,
        kv_watcher: Arc<W>,
        res_tx: mpsc::Sender<Result<WatchResponse, tonic::Status>>,
        mut req_rx: ST,
        header_gen: Arc<HeaderGenerator>,
        watch_progress_notify_interval: Duration,
    ) where
        ST: Stream<Item = Result<WatchRequest, tonic::Status>> + Unpin,
        W: KvWatcherOps,
    {
        let (event_tx, mut event_rx) = mpsc::channel(CHANNEL_SIZE);
        let stop_notify = Arc::new(Event::new());
        let mut watch_handle = WatchHandle::new(
            kv_watcher,
            res_tx,
            event_tx,
            Arc::clone(&stop_notify),
            next_id_gen,
            header_gen,
        );
        let mut ticker = tokio::time::interval(watch_progress_notify_interval);
        loop {
            tokio::select! {
                req = req_rx.next() => {
                    if let Some(req) = req {
                        match req {
                            Ok(req) => {
                                watch_handle.handle_watch_request(req).await;
                            }
                            Err(e) => {
                                warn!("Receive WatchRequest error {:?}", e);
                                break;
                            }
                        }
                    } else {
                        warn!("Watch client closes connection");
                        break;
                    }
                }
                event = event_rx.recv() => {
                    if let Some(event) = event {
                        watch_handle.handle_watch_event(event).await;
                    } else {
                        panic!("Watch event sender is closed");
                    }
                }
                _ = ticker.tick() => {
                    watch_handle.handle_tick_progress().await;
                }
                _ = stop_notify.listen() => {
                    break;
                }
            }
        }
    }
}

FYI: https://docs.rs/tokio/1.28.2/tokio/macro.select.html

Version

0.4.1 (Default)

Relevant log output

No response

Code of Conduct

• I agree to follow this project's Code of Conduct