timeout server pings

5 files changed, 70 insertions, 22 deletions

diff --git a/mumctl/src/main.rs b/mumctl/src/main.rs
index 07e5064..4ce7fc1 100644
--- a/mumctl/src/main.rs
+++ b/mumctl/src/main.rs
@@ -515,16 +515,21 @@ fn match_server_command(server_command: Server, config: &mut Config) -> Result<(
                 .collect();
             for (server, response) in config.servers.iter().zip(queries) {
                 match response.join().unwrap() {
-                    Ok(Ok(response)) => {
-                        if let Some(CommandResponse::ServerStatus {
-                            users, max_users, ..
-                        }) = response
+                    Ok(Ok(Some(response))) => {
+                        if let CommandResponse::ServerStatus {
+                            users,
+                            max_users,
+                            ..
+                        } = response
                         {
-                            println!("{} [{}/{}]", server.name, users, max_users)
+                            println!("{} [{}/{}]", server.name, users, max_users);
                         } else {
-                            unreachable!()
+                            unreachable!();
                         }
                     }
+                    Ok(Ok(None)) => {
+                        println!("{} offline", server.name);
+                    }
                     Ok(Err(e)) => {
                         error!("{}", e);
                         return Err(e)?;
diff --git a/mumd/src/command.rs b/mumd/src/command.rs
index 1337dce..e1c91ab 100644
--- a/mumd/src/command.rs
+++ b/mumd/src/command.rs
@@ -57,7 +57,9 @@ pub async fn handle(
                             id,
                             addr,
                             Box::new(move |packet| {
-                                response_sender.send(converter(packet)).unwrap();
+                                if let Some(packet) = packet {
+                                    response_sender.send(converter(packet)).unwrap();
+                                }
                             }),
                         ));
                         if res.is_err() {
diff --git a/mumd/src/main.rs b/mumd/src/main.rs
index d7bc2c0..341300e 100644
--- a/mumd/src/main.rs
+++ b/mumd/src/main.rs
@@ -109,7 +109,10 @@ async fn receive_commands(
 
                     sender.send((command, tx)).unwrap();
 
-                    let response = rx.await.unwrap();
+                    let response = match rx.await {
+                        Ok(r) => r,
+                        Err(_) => Ok(None),
+                    };
                     let mut serialized = BytesMut::new();
                     bincode::serialize_into((&mut serialized).writer(), &response).unwrap();
 
diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs
index d267007..59620a3 100644
--- a/mumd/src/network/udp.rs
+++ b/mumd/src/network/udp.rs
@@ -10,18 +10,18 @@ use mumble_protocol::crypt::ClientCryptState;
 use mumble_protocol::ping::{PingPacket, PongPacket};
 use mumble_protocol::voice::VoicePacket;
 use mumble_protocol::Serverbound;
-use std::collections::HashMap;
+use std::collections::{hash_map::Entry, HashMap};
 use std::convert::TryFrom;
 use std::net::{Ipv6Addr, SocketAddr};
 use std::sync::{atomic::{AtomicU64, Ordering}, Arc, RwLock};
-use tokio::{join, net::UdpSocket};
-use tokio::sync::{mpsc, watch, Mutex};
+use tokio::{join, net::UdpSocket, select};
+use tokio::sync::{mpsc, oneshot, watch, Mutex};
 use tokio::time::{interval, Duration};
 use tokio_util::udp::UdpFramed;
 
 use super::{run_until, VoiceStreamType};
 
-pub type PingRequest = (u64, SocketAddr, Box<dyn FnOnce(PongPacket)>);
+pub type PingRequest = (u64, SocketAddr, Box<dyn FnOnce(Option<PongPacket>) + Send>);
 
 type UdpSender = SplitSink<UdpFramed<ClientCryptState>, (VoicePacket<Serverbound>, SocketAddr)>;
 type UdpReceiver = SplitStream<UdpFramed<ClientCryptState>>;
@@ -226,32 +226,70 @@ pub async fn handle_pings(
         .await
         .expect("Failed to bind UDP socket");
 
-    let pending = Mutex::new(HashMap::new());
+    let pending = Arc::new(Mutex::new(HashMap::new()));
 
-    let sender_handle = async {
+    let sender = async {
         while let Some((id, socket_addr, handle)) = ping_request_receiver.recv().await {
             debug!("Sending ping {} to {}", id, socket_addr);
             let packet = PingPacket { id };
             let packet: [u8; 12] = packet.into();
             udp_socket.send_to(&packet, &socket_addr).await.unwrap();
-            pending.lock().await.insert(id, handle);
+            let (tx, rx) = oneshot::channel();
+            match pending.lock().await.entry(id) {
+                Entry::Occupied(_) => {
+                    warn!("Tried to send duplicate ping {}", id);
+                    continue;
+                }
+                Entry::Vacant(v) => {
+                    v.insert(tx);
+                }
+            }
+
+            tokio::spawn(async move {
+                let rx = rx.fuse();
+                let timeout = async {
+                    tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await;
+                };
+                handle(select! {
+                    r = rx => match r {
+                        Ok(r) => Some(r),
+                        Err(_) => {
+                            warn!("Ping response sender dropped");
+                            None
+                        }
+                    },
+                    _ = timeout => None,
+                });
+            });
         }
     };
 
-    let receiver_handle = async {
+    let receiver = async {
         let mut buf = vec![0; 24];
+
         while let Ok(read) = udp_socket.recv(&mut buf).await {
-            assert_eq!(read, 24);
+            if read != 24 {
+                warn!("Ping response had length {}, expected 24", read);
+                continue;
+            }
+            assert_eq!(read, 24); // just checked
 
             let packet = PongPacket::try_from(buf.as_slice()).unwrap();
 
-            if let Some(handler) = pending.lock().await.remove(&packet.id) {
-                handler(packet);
+            match pending.lock().await.entry(packet.id) {
+                Entry::Occupied(o) => {
+                    let id = *o.key();
+                    if o.remove().send(packet).is_err() {
+                        debug!("Received response to ping {} too late", id);
+                    }
+                }
+                Entry::Vacant(v) => {
+                    warn!("Received ping {} that we didn't send", v.key());
+                }
             }
         }
     };
 
     debug!("Waiting for ping requests");
-
-    join!(sender_handle, receiver_handle);
+    join!(sender, receiver);
 }
diff --git a/mumd/src/state.rs b/mumd/src/state.rs
index b52b330..1f7331a 100644
--- a/mumd/src/state.rs
+++ b/mumd/src/state.rs
@@ -43,7 +43,7 @@ pub enum ExecutionContext {
     Now(Box<dyn FnOnce() -> mumlib::error::Result<Option<CommandResponse>>>),
     Ping(
         Box<dyn FnOnce() -> mumlib::error::Result<SocketAddr>>,
-        Box<dyn FnOnce(PongPacket) -> mumlib::error::Result<Option<CommandResponse>>>,
+        Box<dyn FnOnce(PongPacket) -> mumlib::error::Result<Option<CommandResponse>> + Send>,
     ),
 }