Repository: slawlor/ractor
Pull request #320 (2025-01-09 12:04:28)

Make latency measurement in ping message robust against time drift
Time drift happens and time going backwards is common on NTP enabled servers. When time drift happens it's not worth it to kill the node session.

Change the timestamp payload to use duration against NodeSession specific Instant epoch initialized at creation time. This ensures the timestamp increases monotonically. It can only go backwards when the payload is malformed.

Note - this makes the timestamp in the payload an offset from UNIX_EPOCH. I think it is fine and nobody should be reading the timestamp for anything useful.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/slawlor/ractor/pull/320?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) Report
Attention: Patch coverage is `42.85714%` with `8 lines` in your changes missing coverage. Please review.
> Project coverage is 82.08%. Comparing base [(`b0cd1a0`)](https://app.codecov.io/gh/slawlor/ractor/commit/b0cd1a09d9c3218ecd061d99b11f8713c5a8ba5c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) to head [(`4319ce2`)](https://app.codecov.io/gh/slawlor/ractor/commit/4319ce21b3d232dc303b393676cdd62370fa2b12?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor).

| [Files with missing lines](https://app.codecov.io/gh/slawlor/ractor/pull/320?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) | Patch % | Lines |
|---|---|---|
| [ractor\_cluster/src/node/node\_session/mod.rs](https://app.codecov.io/gh/slawlor/ractor/pull/320?src=pr&el=tree&filepath=ractor_cluster%2Fsrc%2Fnode%2Fnode_session%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor#diff-cmFjdG9yX2NsdXN0ZXIvc3JjL25vZGUvbm9kZV9zZXNzaW9uL21vZC5ycw==) | 0.00% | [8 Missing :warning: ](https://app.codecov.io/gh/slawlor/ractor/pull/320?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #320      +/-   ##
==========================================
- Coverage   82.08%   82.08%   -0.01%     
==========================================
  Files          61       61              
  Lines       12198    12206       +8     
==========================================
+ Hits        10013    10019       +6     
- Misses       2185     2187       +2     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/slawlor/ractor/pull/320?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor).


Patch:
diff --git a/ractor_cluster/src/node/node_session/mod.rs b/ractor_cluster/src/node/node_session/mod.rs
index 7218442..addf35d 100644
--- a/ractor_cluster/src/node/node_session/mod.rs
+++ b/ractor_cluster/src/node/node_session/mod.rs
@@ -10,6 +10,7 @@
 use std::collections::{HashMap, HashSet};
 use std::convert::TryInto;
 use std::net::SocketAddr;
+use std::time::{Instant, SystemTime};
 
 use ractor::message::SerializedMessage;
 use ractor::pg::{get_scoped_local_members, which_scopes_and_groups, GroupChangeMessage};
@@ -500,10 +501,10 @@ impl NodeSession {
                         .expect("Timestamp missing in Pong")
                         .try_into()
                         .expect("Failed to convert Pong(Timestamp) to SystemTime");
-                    let delta_ms = std::time::SystemTime::now()
-                        .duration_since(ts)
-                        .expect("Time went backwards")
-                        .as_millis();
+                    let inst = ts
+                        .duration_since(SystemTime::UNIX_EPOCH)
+                        .expect("Time went backwards");
+                    let delta_ms = (state.epoch.elapsed() - inst).as_millis();
                     tracing::debug!("Ping -> Pong took {delta_ms}ms");
                     if delta_ms > 50 {
                         tracing::warn!(
@@ -755,6 +756,7 @@ pub struct NodeSessionState {
     tcp: Option<ActorRef<crate::net::session::SessionMessage>>,
     peer_addr: SocketAddr,
     local_addr: SocketAddr,
+    epoch: Instant,
     name: Option<auth_protocol::NameMessage>,
     auth: AuthenticationState,
     remote_actors: HashMap<u64, ActorRef<RemoteActorMessage>>,
@@ -798,14 +800,15 @@ impl NodeSessionState {
     }
 
     fn schedule_tcp_ping(&self) {
+        let epoch = self.epoch;
         if let Some(tcp) = &self.tcp {
             #[allow(clippy::let_underscore_future)]
-            let _ = tcp.send_after(Self::get_send_delay(), || {
+            let _ = tcp.send_after(Self::get_send_delay(), move || {
                 let ping = control_protocol::ControlMessage {
                     msg: Some(control_protocol::control_message::Msg::Ping(
                         control_protocol::Ping {
                             timestamp: Some(prost_types::Timestamp::from(
-                                std::time::SystemTime::now(),
+                                SystemTime::UNIX_EPOCH + epoch.elapsed(),
                             )),
                         },
                     )),
@@ -861,6 +864,7 @@ impl Actor for NodeSession {
             remote_actors: HashMap::new(),
             peer_addr,
             local_addr,
+            epoch: Instant::now(),
         };
 
         // If a client-connection, startup the handshake


Test patch:
diff --git a/ractor_cluster/src/node/node_session/tests.rs b/ractor_cluster/src/node/node_session/tests.rs
index 0160b10..30ae7f0 100644
--- a/ractor_cluster/src/node/node_session/tests.rs
+++ b/ractor_cluster/src/node/node_session/tests.rs
@@ -9,6 +9,7 @@ use std::sync::{
     atomic::{AtomicU8, Ordering},
     Arc,
 };
+use std::time::Instant;
 
 use ractor::concurrency::sleep;
 
@@ -113,6 +114,7 @@ async fn node_sesison_client_auth_success() {
         name: None,
         remote_actors: HashMap::new(),
         tcp: None,
+        epoch: Instant::now(),
     };
 
     // Client sends their name, Server responds with Ok
@@ -257,6 +259,7 @@ async fn node_session_client_auth_session_state_failures() {
         name: None,
         remote_actors: HashMap::new(),
         tcp: None,
+        epoch: Instant::now(),
     };
 
     // Client sends their name, Server responds with Ok
@@ -386,6 +389,7 @@ async fn node_session_server_auth_success() {
         name: None,
         remote_actors: HashMap::new(),
         tcp: None,
+        epoch: Instant::now(),
     };
 
     // Client sends their name
@@ -479,6 +483,7 @@ async fn node_session_server_auth_session_state_failures() {
         name: None,
         remote_actors: HashMap::new(),
         tcp: None,
+        epoch: Instant::now(),
     };
 
     // Other session continues, this one dies
@@ -627,6 +632,7 @@ async fn node_session_handle_node_msg() {
         name: None,
         remote_actors: HashMap::new(),
         tcp: None,
+        epoch: Instant::now(),
     };
     // add the "remote" actor
     state
@@ -723,6 +729,7 @@ async fn node_session_handle_control() {
         name: None,
         remote_actors: HashMap::new(),
         tcp: None,
+        epoch: Instant::now(),
     };
 
     // check spawn creates a remote actor