refactor: align operator doppelgänger grace period with message TTL window

Synchronizes the operator doppelgänger grace period with the complete
message validation TTL formula to prevent false positives from
late-arriving messages after restart.

Changes:
- Export LATE_MESSAGE_MARGIN from message_validator for reuse
- Calculate grace period using full TTL: (slots_per_epoch +
  LATE_SLOT_ALLOWANCE) × slot_duration + LATE_MESSAGE_MARGIN
- Remove obsolete OPERATOR_DOPPELGANGER_GRACE_PERIOD_SECS from network
- Simplify client startup by moving grace period calculation into service
- Update tests to dynamically compute durations from service config

Grace period on mainnet: (32 + 2) × 12s + 3s = 411 seconds
This matches the complete message acceptance window for Committee and
Aggregator roles, eliminating the theoretical 3-second vulnerability.

Author: diegomrsantos

Cargo.lock

@@ -92,20 +92,7 @@ fn start_operator_doppelganger(
     wait_epochs: u64,
     executor: &TaskExecutor,
 ) {
-    info!(
-        wait_epochs = wait_epochs,
-        grace_period_secs = network::OPERATOR_DOPPELGANGER_GRACE_PERIOD_SECS,
-        "Operator doppelgänger: starting monitoring period"
-    );
-
-    // Spawn background task to end monitoring after grace period + wait epochs
-    // Grace period prevents false positives from receiving our own old messages after
-    // restart (they remain in gossip cache for ~4.2s)
-    service.spawn_monitor_task(
-        Duration::from_secs(network::OPERATOR_DOPPELGANGER_GRACE_PERIOD_SECS),
-        wait_epochs,
-        executor,
-    );
+    service.clone().spawn_monitor_task(wait_epochs, executor);
 }
 
 impl Client {

anchor/client/src/lib.rs

@@ -582,8 +582,15 @@ pub(crate) fn validate_beacon_duty(
 /// clockErrorTolerance is the maximum amount of clock error we expect to see between nodes.
 const CLOCK_ERROR_TOLERANCE: Duration = Duration::from_millis(50);
 /// lateMessageMargin is the duration past a message's TTL in which it is still considered valid.
-const LATE_MESSAGE_MARGIN: Duration = Duration::from_secs(3);
-const LATE_SLOT_ALLOWANCE: u64 = 2;
+///
+/// This margin is added to the deadline calculation after converting slot-based TTL to time.
+/// The full message acceptance window is: (ttl_slots × slot_duration) + LATE_MESSAGE_MARGIN
+pub const LATE_MESSAGE_MARGIN: Duration = Duration::from_secs(3);
+/// Number of slots added to TTL windows for late message acceptance
+///
+/// Used in calculating message acceptance deadlines for Committee and Aggregator roles.
+/// The actual TTL is: slots_per_epoch + LATE_SLOT_ALLOWANCE
+pub const LATE_SLOT_ALLOWANCE: u64 = 2;
 
 /// Validates that the message's slot timing is correct
 pub(crate) fn validate_slot_time(

anchor/message_validator/src/lib.rs

@@ -26,16 +26,6 @@ pub const GOSSIPSUB_HEARTBEAT_INTERVAL_MILLIS: u64 = 700;
 /// Messages remain in mcache for: history_length × heartbeat_interval (6 × 700ms = 4.2s)
 pub const GOSSIPSUB_HISTORY_LENGTH: usize = 6;
 
-/// Operator doppelgänger grace period in seconds
-///
-/// Wait after startup before detecting twins to prevent false positives from own old messages.
-/// Automatically derived as: (gossip_cache_window + 1 second buffer)
-/// Default: 5s (cache window is 4.2s)
-pub const OPERATOR_DOPPELGANGER_GRACE_PERIOD_SECS: u64 = {
-    let cache_window_millis = GOSSIPSUB_HISTORY_LENGTH as u64 * GOSSIPSUB_HEARTBEAT_INTERVAL_MILLIS;
-    cache_window_millis / 1000 + 1
-};
-
 /// Custom message ID function matching Go-SSV implementation.
 /// Uses xxhash64 of the full message to ensure uniqueness across operators.
 ///

anchor/network/src/behaviour.rs

@@ -11,7 +11,6 @@ mod peer_manager;
 mod scoring;
 mod transport;
 
-pub use behaviour::OPERATOR_DOPPELGANGER_GRACE_PERIOD_SECS;
 pub use config::{Config, DEFAULT_DISC_PORT, DEFAULT_QUIC_PORT, DEFAULT_TCP_PORT};
 pub use network::Network;
 pub use network_utils::listen_addr::{ListenAddr, ListenAddress};

anchor/network/src/lib.rs

@@ -6,6 +6,7 @@ edition = { workspace = true }
 [dependencies]
 database = { workspace = true }
 futures = { workspace = true }
+message_validator = { workspace = true }
 parking_lot = { workspace = true }
 ssv_types = { workspace = true }
 task_executor = { workspace = true }

anchor/operator_doppelganger/Cargo.toml

@@ -115,39 +115,40 @@ impl OperatorDoppelgangerService {
     }
 
     /// Spawn a background task to end monitoring after the configured wait period
-    ///
-    /// The task sleeps for the grace period plus the monitoring duration, then transitions
-    /// to active mode.
-    ///
-    /// # Arguments
-    ///
-    /// * `grace_period` - Duration to wait before starting twin detection. Should be slightly
-    ///   longer than the gossip message cache window (history_length × heartbeat_interval ≈ 4.2s)
-    ///   to ensure our own old messages have expired from the network. See `DoppelgangerState`
-    ///   documentation for details on why this prevents false positives.
-    /// * `wait_epochs` - Number of epochs to monitor for doppelgängers after grace period ends
-    pub fn spawn_monitor_task(
-        self: Arc<Self>,
-        grace_period: Duration,
-        wait_epochs: u64,
-        executor: &TaskExecutor,
-    ) {
-        // Calculate monitoring duration (after grace period)
-        let monitoring_duration =
-            Duration::from_secs(wait_epochs * self.slots_per_epoch * self.slot_duration.as_secs());
+    pub fn spawn_monitor_task(self: Arc<Self>, wait_epochs: u64, executor: &TaskExecutor) {
+        // Grace period must match the full message TTL window to prevent false positives
+        // from late-arriving messages after restart.
+        //
+        // Full TTL = (slots_per_epoch + LATE_SLOT_ALLOWANCE) × slot_duration + LATE_MESSAGE_MARGIN
+        // This matches the complete deadline calculation in message validation.
+        let grace_period_slots = self.slots_per_epoch + message_validator::LATE_SLOT_ALLOWANCE;
+        let grace_period_secs = grace_period_slots * self.slot_duration.as_secs();
+        let grace_period =
+            Duration::from_secs(grace_period_secs) + message_validator::LATE_MESSAGE_MARGIN;
+
+        let monitoring_slots = wait_epochs * self.slots_per_epoch;
+        let monitoring_secs = monitoring_slots * self.slot_duration.as_secs();
+        let monitoring_duration = Duration::from_secs(monitoring_secs);
 
         executor.spawn_without_exit(
             async move {
-                // Wait for grace period - prevents false positives from own old messages
+                info!(
+                    grace_period_slots = grace_period_slots,
+                    grace_period_secs = grace_period.as_secs(),
+                    "Operator doppelgänger: entering grace period"
+                );
                 tokio::time::sleep(grace_period).await;
 
-                // Grace period complete - start detecting twins
+                info!(
+                    monitoring_epochs = wait_epochs,
+                    monitoring_secs = monitoring_duration.as_secs(),
+                    "Operator doppelgänger: grace period complete, starting monitoring"
+                );
                 self.state.write().end_grace_period();
 
-                // Wait for monitoring period to complete
                 tokio::time::sleep(monitoring_duration).await;
 
-                // Monitoring complete - stop checking for doppelgängers
+                info!("Operator doppelgänger: monitoring period complete");
                 self.end_monitoring_period();
             },
             "doppelganger-monitor",
@@ -273,25 +274,26 @@ mod tests {
         TaskExecutor::new(handle, exit, shutdown_tx, "doppelganger_test".into())
     }
 
-    /// Helper to spawn monitor task and advance time past grace period
-    ///
-    /// Returns the service in monitoring mode with grace period complete,
-    /// ready for twin detection tests.
     async fn spawn_and_advance_past_grace_period(
         service: Arc<OperatorDoppelgangerService>,
         executor: &TaskExecutor,
     ) {
-        let grace_period = Duration::from_secs(5);
         let wait_epochs = 2;
 
         // Spawn monitor task
-        service
-            .clone()
-            .spawn_monitor_task(grace_period, wait_epochs, executor);
+        service.clone().spawn_monitor_task(wait_epochs, executor);
 
         // Give the spawned task a chance to start
         tokio::task::yield_now().await;
 
+        // Calculate grace period from service configuration
+        // Grace period = (slots_per_epoch + LATE_SLOT_ALLOWANCE) × slot_duration +
+        // LATE_MESSAGE_MARGIN
+        let grace_period_slots = service.slots_per_epoch + message_validator::LATE_SLOT_ALLOWANCE;
+        let grace_period =
+            Duration::from_secs(grace_period_slots * service.slot_duration.as_secs())
+                + message_validator::LATE_MESSAGE_MARGIN;
+
         // Advance time past grace period
         tokio::time::advance(grace_period).await;
 
@@ -430,12 +432,9 @@ mod tests {
         let committee_id = CommitteeId([1u8; 32]);
         let executor = create_test_executor();
 
-        // Spawn the monitor task with grace period
-        let grace_period = Duration::from_secs(5);
+        // Spawn the monitor task
         let wait_epochs = 2;
-        service
-            .clone()
-            .spawn_monitor_task(grace_period, wait_epochs, &executor);
+        service.clone().spawn_monitor_task(wait_epochs, &executor);
 
         // Still in grace period (don't advance time)
         assert!(!service.is_monitoring());
@@ -505,17 +504,23 @@ mod tests {
         let executor = create_test_executor();
 
         // Spawn the monitor task
-        let grace_period = Duration::from_secs(5);
         let wait_epochs = 2;
-        service
-            .clone()
-            .spawn_monitor_task(grace_period, wait_epochs, &executor);
+        service.clone().spawn_monitor_task(wait_epochs, &executor);
 
         // Give the spawned task a chance to start
         tokio::task::yield_now().await;
 
-        // Calculate monitoring duration
-        let monitoring_duration = Duration::from_secs(wait_epochs * 12); // epochs * (slots_per_epoch=1) * (slot_duration=12s)
+        // Calculate durations from service configuration
+        // Grace period = (slots_per_epoch + LATE_SLOT_ALLOWANCE) × slot_duration +
+        // LATE_MESSAGE_MARGIN
+        let grace_period_slots = service.slots_per_epoch + message_validator::LATE_SLOT_ALLOWANCE;
+        let grace_period =
+            Duration::from_secs(grace_period_slots * service.slot_duration.as_secs())
+                + message_validator::LATE_MESSAGE_MARGIN;
+
+        let monitoring_slots = wait_epochs * service.slots_per_epoch;
+        let monitoring_duration =
+            Duration::from_secs(monitoring_slots * service.slot_duration.as_secs());
 
         // Advance time past grace period first
         tokio::time::advance(grace_period).await;
@@ -532,11 +537,11 @@ mod tests {
         let (signed_message, qbft_message) =
             create_test_message(committee_id, vec![OperatorId(1)], 10, 0);
 
-        // This should NOT detect a twin (monitoring period ended via timer)
+        // This should NOT detect a twin (monitoring period completed via timer)
         let result = service.is_doppelganger(&signed_message, Some(&qbft_message));
         assert!(
             !result,
-            "Message after monitoring period should NOT detect twin"
+            "Message after monitoring period should NOT detect twin (monitoring complete)"
         );
     }
 }