Merge pull request #216 from memorysafety/handle-interface-change

Handle interface change

Author: rnijveld

Cargo.lock

@@ -21,6 +21,7 @@ serde_json = "1.0.82"
 sentry = { version = "0.27.0", optional = true }
 sentry-tracing = { version = "0.27.0", optional = true }
 rand = "0.8.5"
+libc = "0.2.126"
 
 [dev-dependencies]
 ntp-proto = { path = "../ntp-proto", features=["ext-test"]}

ntp-daemon/Cargo.toml

@@ -1,4 +1,4 @@
-use std::{future::Future, marker::PhantomData, ops::ControlFlow, pin::Pin, sync::Arc};
+use std::{future::Future, marker::PhantomData, pin::Pin, sync::Arc};
 
 use ntp_proto::{
     IgnoreReason, NtpClock, NtpHeader, NtpInstant, NtpTimestamp, Peer, PeerSnapshot, ReferenceId,
@@ -47,6 +47,8 @@ impl ResetEpoch {
 pub enum MsgForSystem {
     /// Received a Kiss-o'-Death and must demobilize
     MustDemobilize(PeerIndex),
+    /// Experienced a network issue and must be restarted
+    NetworkIssue(PeerIndex),
     /// Received an acceptable packet and made a new peer snapshot
     /// A new measurement should try to trigger a clock select
     NewMeasurement(PeerIndex, ResetEpoch, PeerSnapshot),
@@ -100,6 +102,18 @@ pub(crate) struct PeerTask<C: 'static + NtpClock + Send, T: Wait> {
     reset_epoch: ResetEpoch,
 }
 
+#[derive(Debug)]
+enum PollResult {
+    Ok,
+    NetworkGone,
+}
+
+#[derive(Debug)]
+enum PacketResult {
+    Ok,
+    Demobilize,
+}
+
 impl<C, T> PeerTask<C, T>
 where
     C: 'static + NtpClock + Send,
@@ -120,7 +134,7 @@ where
             .reset(self.last_poll_sent + poll_interval);
     }
 
-    async fn handle_poll(&mut self, poll_wait: &mut Pin<&mut T>) {
+    async fn handle_poll(&mut self, poll_wait: &mut Pin<&mut T>) -> PollResult {
         let system_snapshot = *self.channels.system_snapshots.read().await;
         let packet = self.peer.generate_poll_message(system_snapshot);
 
@@ -145,7 +159,17 @@ where
 
         if let Err(error) = self.socket.send(&packet.serialize()).await {
             warn!(?error, "poll message could not be sent");
+
+            match error.raw_os_error() {
+                Some(libc::EHOSTDOWN)
+                | Some(libc::EHOSTUNREACH)
+                | Some(libc::ENETDOWN)
+                | Some(libc::ENETUNREACH) => return PollResult::NetworkGone,
+                _ => {}
+            }
         }
+
+        PollResult::Ok
     }
 
     async fn handle_packet(
@@ -154,7 +178,7 @@ where
         packet: NtpHeader,
         send_timestamp: NtpTimestamp,
         recv_timestamp: NtpTimestamp,
-    ) -> ControlFlow<(), ()> {
+    ) -> PacketResult {
         let ntp_instant = NtpInstant::now();
 
         let system_snapshot = *self.channels.system_snapshots.read().await;
@@ -184,14 +208,14 @@ where
                 let msg = MsgForSystem::MustDemobilize(self.index);
                 self.channels.msg_for_system_sender.send(msg).await.ok();
 
-                return ControlFlow::Break(());
+                return PacketResult::Demobilize;
             }
             Err(ignore_reason) => {
                 debug!(?ignore_reason, "packet ignored");
             }
         }
 
-        ControlFlow::Continue(())
+        PacketResult::Ok
     }
 
     async fn run(&mut self, mut poll_wait: Pin<&mut T>) {
@@ -200,7 +224,13 @@ where
 
             tokio::select! {
                 () = &mut poll_wait => {
-                    self.handle_poll(&mut poll_wait).await;
+                    match self.handle_poll(&mut poll_wait).await {
+                        PollResult::Ok => {},
+                        PollResult::NetworkGone => {
+                            self.channels.msg_for_system_sender.send(MsgForSystem::NetworkIssue(self.index)).await.ok();
+                            break;
+                        }
+                    }
                 },
                 result = self.channels.reset.changed() => {
                     if let Ok(()) = result {
@@ -214,19 +244,26 @@ where
                     }
                 }
                 result = self.socket.recv(&mut buf) => {
-                    let send_timestamp = match self.last_send_timestamp {
-                        Some(ts) => ts,
-                        None => {
-                            warn!("we received a message without having sent one; discarding");
-                            continue;
-                        }
-                    };
-
-                    if let Some((packet, recv_timestamp)) = accept_packet(result, &buf) {
-                        match self.handle_packet(&mut poll_wait, packet, send_timestamp, recv_timestamp).await{
-                            ControlFlow::Continue(_) => continue,
-                            ControlFlow::Break(_) => break,
-                        }
+                    match accept_packet(result, &buf) {
+                        AcceptResult::Accept(packet, recv_timestamp) => {
+                            let send_timestamp = match self.last_send_timestamp {
+                                Some(ts) => ts,
+                                None => {
+                                    warn!("we received a message without having sent one; discarding");
+                                    continue;
+                                }
+                            };
+
+                            match self.handle_packet(&mut poll_wait, packet, send_timestamp, recv_timestamp).await {
+                                PacketResult::Ok => {},
+                                PacketResult::Demobilize => break,
+                            }
+                        },
+                        AcceptResult::NetworkGone => {
+                            self.channels.msg_for_system_sender.send(MsgForSystem::NetworkIssue(self.index)).await.ok();
+                            break;
+                        },
+                        AcceptResult::Ignore => {},
                     }
                 },
             }
@@ -239,17 +276,26 @@ where
     C: 'static + NtpClock + Send,
 {
     #[instrument(skip(clock, channels))]
-    pub async fn spawn<A: ToSocketAddrs + std::fmt::Debug>(
+    pub fn spawn<A: ToSocketAddrs + std::fmt::Debug + Send + Sync + 'static>(
         index: PeerIndex,
         addr: A,
         clock: C,
+        network_wait_period: std::time::Duration,
         mut channels: PeerChannels,
-    ) -> std::io::Result<tokio::task::JoinHandle<()>> {
-        let socket = UdpSocket::new("0.0.0.0:0", addr).await?;
-        let our_id = ReferenceId::from_ip(socket.as_ref().local_addr().unwrap().ip());
-        let peer_id = ReferenceId::from_ip(socket.as_ref().peer_addr().unwrap().ip());
+    ) -> tokio::task::JoinHandle<()> {
+        tokio::spawn(async move {
+            let socket = loop {
+                match UdpSocket::new("0.0.0.0:0", &addr).await {
+                    Ok(socket) => break socket,
+                    Err(error) => {
+                        warn!(?error, "Could not open socket");
+                        tokio::time::sleep(network_wait_period).await;
+                    }
+                }
+            };
+            let our_id = ReferenceId::from_ip(socket.as_ref().local_addr().unwrap().ip());
+            let peer_id = ReferenceId::from_ip(socket.as_ref().peer_addr().unwrap().ip());
 
-        let handle = tokio::spawn(async move {
             let local_clock_time = NtpInstant::now();
             let peer = Peer::new(our_id, peer_id, local_clock_time);
 
@@ -273,16 +319,21 @@ where
             };
 
             process.run(poll_wait).await
-        });
-
-        Ok(handle)
+        })
     }
 }
 
+#[derive(Debug)]
+enum AcceptResult {
+    Accept(NtpHeader, NtpTimestamp),
+    Ignore,
+    NetworkGone,
+}
+
 fn accept_packet(
     result: Result<(usize, Option<NtpTimestamp>), std::io::Error>,
     buf: &[u8; 48],
-) -> Option<(NtpHeader, NtpTimestamp)> {
+) -> AcceptResult {
     match result {
         Ok((size, Some(recv_timestamp))) => {
             // Note: packets are allowed to be bigger when including extensions.
@@ -292,26 +343,32 @@ fn accept_packet(
             if size < 48 {
                 warn!(expected = 48, actual = size, "received packet is too small");
 
-                None
+                AcceptResult::Ignore
             } else {
                 match NtpHeader::deserialize(buf) {
-                    Ok(packet) => Some((packet, recv_timestamp)),
+                    Ok(packet) => AcceptResult::Accept(packet, recv_timestamp),
                     Err(e) => {
                         warn!("received invalid packet: {}", e);
-                        None
+                        AcceptResult::Ignore
                     }
                 }
             }
         }
         Ok((size, None)) => {
             warn!(?size, "received a packet without a timestamp");
 
-            None
+            AcceptResult::Ignore
         }
         Err(receive_error) => {
             warn!(?receive_error, "could not receive packet");
 
-            None
+            match receive_error.raw_os_error() {
+                Some(libc::EHOSTDOWN)
+                | Some(libc::EHOSTUNREACH)
+                | Some(libc::ENETDOWN)
+                | Some(libc::ENETUNREACH) => AcceptResult::NetworkGone,
+                _ => AcceptResult::Ignore,
+            }
         }
     }
 }
@@ -502,15 +559,14 @@ mod tests {
             PeerIndex::from_inner(0),
             "127.0.0.1:8003",
             TestClock {},
+            std::time::Duration::from_secs(60),
             PeerChannels {
                 msg_for_system_sender,
                 system_snapshots,
                 system_config,
                 reset,
             },
-        )
-        .await
-        .unwrap();
+        );
 
         let peer_epoch = match msg_for_system_receiver.recv().await.unwrap() {
             MsgForSystem::UpdatedSnapshot(_, peer_epoch, _) => peer_epoch,

ntp-daemon/src/peer.rs

@@ -8,6 +8,8 @@ use crate::{
 use ntp_proto::{NtpClock, PeerSnapshot};
 use tokio::task::JoinHandle;
 
+const NETWORK_WAIT_PERIOD: std::time::Duration = std::time::Duration::from_secs(60);
+
 #[derive(Debug, Clone, Copy)]
 pub enum PeerStatus {
     /// We are waiting for the first snapshot from this peer _in the current reset epoch_.
@@ -72,17 +74,27 @@ impl<C: NtpClock> Peers<C> {
         }
     }
 
-    pub async fn add_peer(&mut self, config: PeerConfig) -> std::io::Result<JoinHandle<()>> {
+    fn add_peer_internal(&mut self, config: Arc<PeerConfig>) -> JoinHandle<()> {
         let index = self.indexer.get();
         let addr = config.addr.clone();
         self.peers.insert(
             index,
             PeerData {
                 status: PeerStatus::NoMeasurement,
-                config: Arc::new(config),
+                config,
             },
         );
-        PeerTask::spawn(index, &addr, self.clock.clone(), self.channels.clone()).await
+        PeerTask::spawn(
+            index,
+            addr,
+            self.clock.clone(),
+            NETWORK_WAIT_PERIOD,
+            self.channels.clone(),
+        )
+    }
+
+    pub async fn add_peer(&mut self, config: PeerConfig) -> JoinHandle<()> {
+        self.add_peer_internal(Arc::new(config))
     }
 
     #[cfg(test)]
@@ -151,6 +163,11 @@ impl<C: NtpClock> Peers<C> {
                     self.peers.get_mut(&index).unwrap().status = PeerStatus::Measurement(snapshot);
                 }
             }
+            MsgForSystem::NetworkIssue(index) => {
+                // Restart the peer reusing its configuration.
+                let config = self.peers.remove(&index).unwrap().config;
+                self.add_peer_internal(config);
+            }
         }
     }
 

ntp-daemon/src/peer_manager.rs

@@ -50,7 +50,7 @@ pub async fn spawn(
         UnixNtpClock::new(),
     );
     for peer_config in peer_configs.iter() {
-        peers.add_peer(peer_config.to_owned()).await?;
+        peers.add_peer(peer_config.to_owned()).await;
     }
     let peers = Arc::new(tokio::sync::RwLock::new(peers));