refactor: replace grace period with slot-based operator doppelgänger detection

Refactors the operator doppelgänger protection system to use slot comparison
instead of time-based grace period, improving startup time and reliability.

Changes:

State Management:
- Replace DoppelgangerState enum with AtomicBool for simpler state tracking
- Add startup_slot field to track service initialization time
- Remove grace period logic (saves 411 seconds at startup)

Detection Logic:
- Implement slot-based detection: msg_slot > startup_slot indicates twin
- Messages at or before startup_slot are ignored (our own old messages)
- Add extract_message_slot() helper supporting both QBFT and PartialSignatureMessages
- No race conditions possible since all outgoing messages blocked during monitoring

Benefits:
- Faster startup: eliminates 411-second grace period wait time
- More reliable: slot comparison is deterministic, unaffected by network delays
- Simpler code: fewer states, cleaner logic with lock-free atomic operations
- Better edge case handling: restart in same slot, network delays, clock skew

Testing:
- Replace time-based async tests with slot-based sync tests
- All 8 tests pass verifying correct twin detection behavior

Author: diegomrsantos

Cargo.lock

@@ -495,8 +495,8 @@ pub struct Node {
     #[clap(
         long,
         help = "Enable operator doppelgänger protection. When enabled, the node blocks all \
-                outgoing messages during a grace period (to let old messages expire), then \
-                monitors for messages with its operator ID. Shuts down if a twin is detected \
+                outgoing messages and monitors the network for messages signed with its operator ID \
+                that reference slots after startup. Shuts down if a twin operator is detected \
                 to prevent QBFT protocol violations. Enabled by default.",
         display_order = 0,
         default_value_t = true,
@@ -508,9 +508,9 @@ pub struct Node {
     #[clap(
         long,
         value_name = "EPOCHS",
-        help = "Number of epochs to monitor for twin operators after the grace period. \
-                During monitoring, outgoing messages remain blocked and the node listens \
-                for messages with its operator ID to detect duplicate instances.",
+        help = "Number of epochs to monitor for twin operators using slot-based detection. \
+                During monitoring, outgoing messages remain blocked and the node checks incoming \
+                messages for slots after startup to detect duplicate operator instances.",
         display_order = 0,
         default_value_t = 2,
         requires = "operator_dg"

anchor/client/src/cli.rs

@@ -437,8 +437,16 @@ impl Client {
 
         // Create operator doppelgänger protection if enabled (will be started after sync)
         let doppelganger_service = if config.operator_dg && config.impostor.is_none() {
+            // Get current slot for slot-based detection baseline
+            let current_slot = slot_clock.now().ok_or_else(|| {
+                "Failed to get current slot for doppelgänger protection".to_string()
+            })?;
+            // Convert types::Slot to ssv_types::Slot
+            let startup_slot = ssv_types::Slot::new(current_slot.as_u64());
+
             Some(Arc::new(OperatorDoppelgangerService::new(
                 operator_id.clone(),
+                startup_slot,
                 E::slots_per_epoch(),
                 Duration::from_secs(spec.seconds_per_slot),
                 executor.shutdown_sender(),

anchor/client/src/lib.rs

@@ -54,10 +54,10 @@ impl<S: SlotClock + 'static, D: DutiesProvider> MessageSender for Arc<NetworkMes
         additional_message_callback: Option<Box<MessageCallback>>,
     ) -> Result<(), Error> {
         // Check if doppelgänger protection is active - block outgoing messages
-        // This includes both grace period (waiting for old messages to expire) and
-        // monitoring period (actively detecting twins)
+        // During the entire monitoring period, we block all outgoing messages to prevent
+        // competition with potential twin operators
         if let Some(dg) = &self.doppelganger_service
-            && dg.is_active()
+            && dg.is_monitoring()
         {
             trace!("Dropping message send - doppelgänger protection active");
             return Ok(());
@@ -109,10 +109,10 @@ impl<S: SlotClock + 'static, D: DutiesProvider> MessageSender for Arc<NetworkMes
 
     fn send(&self, message: SignedSSVMessage, committee_id: CommitteeId) -> Result<(), Error> {
         // Check if doppelgänger protection is active - block outgoing messages
-        // This includes both grace period (waiting for old messages to expire) and
-        // monitoring period (actively detecting twins)
+        // During the entire monitoring period, we block all outgoing messages to prevent
+        // competition with potential twin operators
         if let Some(dg) = &self.doppelganger_service
-            && dg.is_active()
+            && dg.is_monitoring()
         {
             trace!("Dropping message send - doppelgänger protection active");
             return Ok(());

anchor/message_sender/src/network.rs

@@ -5,13 +5,14 @@ edition = { workspace = true }
 
 [dependencies]
 database = { workspace = true }
+ethereum_ssz = { workspace = true }
 futures = { workspace = true }
-message_validator = { workspace = true }
 parking_lot = { workspace = true }
 ssv_types = { workspace = true }
 task_executor = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+types = { workspace = true }
 
 [dev-dependencies]
 async-channel = { workspace = true }