refactor(measurement): enhance ICMP engine's async integration and robustness

Ensure IcmpEngine initializes within a Tokio runtime, improve timestamp validation, and refine probe management.

src/main.rs

@@ -243,6 +243,8 @@ fn wait_for_exit(error_rx: &flume::Receiver<anyhow::Error>) -> anyhow::Result<()
 }
 
 fn run(settings: &Settings) -> anyhow::Result<()> {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
     if settings.advanced_settings.dry_run {
         info!("*** MONITORING MODE ACTIVE — qdisc rates will NOT be changed ***");
     }
@@ -252,7 +254,10 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
 
     let (ie_txt, _) = flume::unbounded();
     let (binding_tx, binding_rx) = flume::unbounded();
-    let engine = IcmpEngine::new(start_time, &settings.network, ie_txt, binding_rx)?;
+    let engine = {
+        let _guard = rt.enter();
+        IcmpEngine::new(start_time, &settings.network, ie_txt, binding_rx)?
+    };
     binding_tx.send(vec![IcmpBinding::new(
         MeasurementStream {
             id: MeasurementStreamId(0),
@@ -265,7 +270,6 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         },
     )])?;
 
-    let rt = tokio::runtime::Runtime::new().unwrap();
     rt.spawn(engine.run());
 
     let ReflectorSetup {

src/measurement/icmp.rs

@@ -229,9 +229,7 @@ impl IcmpEngine {
                 .get_mut(&stream_id)
                 .expect("due binding disappeared");
 
-            while binding.next_probe_at <= now {
+            binding.next_probe_at = binding.schedule.next_probe_at(self.origin, now);
-                binding.next_probe_at += binding.schedule.period;
-            }
         }
 
         Ok(())
@@ -318,8 +316,6 @@ impl IcmpEngine {
                 receive,
                 transmit,
             } => {
-                const NON_STANDARD_TIMESTAMP: u32 = 1 << 31;
-
                 if identifier != self.identifier {
                     return Err(IcmpError::WrongIdentifier {
                         expected: self.identifier,
@@ -327,7 +323,10 @@ impl IcmpEngine {
                     });
                 }
 
-                if receive & NON_STANDARD_TIMESTAMP != 0 || transmit & NON_STANDARD_TIMESTAMP != 0 {
+                if originate >= MS_PER_DAY as u32
+                    || receive >= MS_PER_DAY as u32
+                    || transmit >= MS_PER_DAY as u32
+                {
                     return Err(IcmpError::InvalidPacket(
                         "non-standard RFC 792 timestamp".into(),
                     ));
@@ -365,8 +364,8 @@ impl IcmpEngine {
         }
     }
 
-    fn expire_inflight(&mut self, now: Instant) -> Result<(), IcmpError> {
+    async fn expire_inflight(&mut self, now: Instant) -> Result<(), IcmpError> {
-        const PROBE_TIMEOUT: Duration = Duration::from_secs(2);
+        const PROBE_TIMEOUT: Duration = Duration::from_secs(30);
 
         let expired = {
             let mut expired = Vec::new();
@@ -385,11 +384,12 @@ impl IcmpEngine {
 
         for probe in expired {
             self.observation_tx
-                .send(MeasurementEvent::Loss(MeasurementLoss {
+                .send_async(MeasurementEvent::Loss(MeasurementLoss {
                     stream: probe.stream,
                     started_at: probe.sent_at,
                     detected_at: now,
                 }))
+                .await
                 .map_err(|_| IcmpError::EventChannelClosed)?;
         }
 
@@ -432,7 +432,8 @@ impl IcmpEngine {
                     ) {
                         Ok(observation) => {
                             self.observation_tx
-                                .send(MeasurementEvent::Observation(observation))
+                                .send_async(MeasurementEvent::Observation(observation))
+                                .await
                                 .map_err(|_| IcmpError::EventChannelClosed)?;
                         }
 
@@ -449,7 +450,7 @@ impl IcmpEngine {
                 }
 
                 _ = housekeeping.tick() => {
-                    self.expire_inflight(Instant::now())?;
+                    self.expire_inflight(Instant::now()).await?;
                 }
             }
         }