fix: tolerate unknown proposer duties without spawning QBFT

anchor/client/src/lib.rs

@@ -562,7 +562,6 @@ impl Client {
                     network_tx: network_tx.clone(),
                     private_key: key.clone(),
                     operator_id: operator_id.clone(),
-                    validator: Some(message_validator.clone()),
                     is_synced: is_synced.clone(),
                     subnet_service: subnet_service.clone(),
                 },

anchor/common/ssv_types/src/msgid.rs

@@ -98,6 +98,21 @@ impl Role {
         self.max_round().is_some()
     }
 
+    /// Returns true if this role's duty is bound to a single validator's proposal slot, so the
+    /// proposer-duty assignment for that slot decides whether a message can have a duty at all.
+    pub fn is_proposer_scoped(self) -> bool {
+        match self {
+            Role::Proposer | Role::ProposerPreferences | Role::EnvelopeProposer => true,
+            Role::Committee
+            | Role::Aggregator
+            | Role::SyncCommittee
+            | Role::ValidatorRegistration
+            | Role::VoluntaryExit
+            | Role::PTCAttester
+            | Role::AggregatorCommittee => false,
+        }
+    }
+
     /// monotonicSlotRole reports whether a role's signer advances through slots one at a time, so a
     /// message for a slot below the signer's max is stale and must be rejected. False for
     /// committee roles (state is slot-keyed across many validators) and for proposer

anchor/duties_tracker/src/duties_tracker.rs

@@ -10,7 +10,7 @@ use task_executor::TaskExecutor;
 use thiserror::Error;
 use tokio::{sync::watch, time::sleep};
 use tracing::{debug, error, trace, warn};
-use types::{ChainSpec, Epoch, Slot};
+use types::{ChainSpec, Slot};
 
 use crate::{
     Duties, DutiesProvider, DutyAssignment, MembershipKey,
@@ -317,25 +317,6 @@ impl<T: SlotClock + 'static> DutiesProvider for DutiesTracker<T> {
             .is_validator_in_sync_committee(committee_period, validator_index.into())
     }
 
-    fn is_epoch_known_for_proposers(&self, epoch: Epoch) -> bool {
-        self.duties.proposers.read().contains_key(&epoch)
-    }
-
-    fn is_validator_proposer_at_slot(&self, slot: Slot, validator_index: ValidatorIndex) -> bool {
-        let epoch = slot.epoch(self.slots_per_epoch);
-        let validator_index: u64 = validator_index.into();
-        self.duties
-            .proposers
-            .read()
-            .get(&epoch)
-            .map(|proposers| {
-                proposers.iter().any(|proposer_data| {
-                    proposer_data.slot == slot && proposer_data.validator_index == validator_index
-                })
-            })
-            .unwrap_or_default()
-    }
-
     fn get_voluntary_exit_duty_count(&self, slot: Slot, pubkey: &PublicKeyBytes) -> u64 {
         self.voluntary_exit_tracker.get_duty_count(slot, pubkey)
     }
@@ -469,7 +450,7 @@ mod tests {
     // ==================== proposer_assignment_at_slot ====================
 
     #[test]
-    fn test_proposer_assignment_at_slot_returns_some_true_for_assigned_pubkey() {
+    fn test_proposer_assignment_at_slot_returns_assigned_for_assigned_pubkey() {
         // Assigned pubkey AT its slot in a fetched epoch -> Assigned.
         let tracker = tracker_with_empty_network_state();
         let epoch = Epoch::new(0);
@@ -485,8 +466,7 @@ mod tests {
     }
 
     #[test]
-    fn test_proposer_assignment_at_slot_returns_some_false_for_unassigned_pubkey_in_fetched_epoch()
-    {
+    fn test_proposer_assignment_returns_not_assigned_for_unassigned_pubkey() {
         // Different (unassigned) pubkey, same fetched epoch and slot -> NotAssigned.
         let tracker = tracker_with_empty_network_state();
         let epoch = Epoch::new(0);
@@ -503,7 +483,7 @@ mod tests {
     }
 
     #[test]
-    fn test_proposer_assignment_at_slot_returns_some_false_for_assigned_pubkey_at_different_slot() {
+    fn test_proposer_assignment_at_slot_returns_not_assigned_at_different_slot() {
         // The assignment is bound to the exact slot: the assigned pubkey queried at a DIFFERENT
         // slot within the SAME fetched epoch must return NotAssigned (not Assigned). This is the
         // slot-bind case that guards against matching on pubkey alone.
@@ -526,7 +506,7 @@ mod tests {
     }
 
     #[test]
-    fn test_proposer_assignment_at_slot_returns_none_for_unfetched_epoch() {
+    fn test_proposer_assignment_at_slot_returns_unknown_for_unfetched_epoch() {
         // A slot whose epoch has not been fetched -> Unknown, regardless of pubkey.
         let tracker = tracker_with_empty_network_state();
         let fetched_epoch = Epoch::new(0);

anchor/duties_tracker/src/lib.rs

@@ -88,8 +88,11 @@ type ProposerMap = HashMap<Epoch, Vec<ProposerData>>;
 
 #[derive(Debug)]
 pub struct Duties {
-    /// Maps an epoch to all *local* proposers in this epoch. Notably, this does not contain
-    /// proposals for any validators which are not registered locally.
+    /// Maps an epoch to the network-wide proposer duties returned by the beacon node.
+    ///
+    /// Entries are not filtered by Anchor's local validator registry. This coverage is required
+    /// for a missing pubkey and slot pair to be authoritative once the response is validated as
+    /// complete.
     pub proposers: RwLock<ProposerMap>,
     /// Map from validator index to sync committee duties.
     pub sync_duties: SyncCommitteePerPeriod,
@@ -128,10 +131,6 @@ pub trait DutiesProvider: Sync + Send + 'static {
         validator_index: ValidatorIndex,
     ) -> bool;
 
-    fn is_epoch_known_for_proposers(&self, epoch: Epoch) -> bool;
-
-    fn is_validator_proposer_at_slot(&self, slot: Slot, validator_index: ValidatorIndex) -> bool;
-
     fn get_voluntary_exit_duty_count(&self, slot: Slot, pubkey: &PublicKeyBytes) -> u64;
 
     fn proposer_assignment_at_slot(

anchor/message_receiver/src/lib.rs

@@ -4,7 +4,7 @@ use libp2p::{
     PeerId,
     gossipsub::{Message, MessageId},
 };
-pub use message_validator::TopicContext;
+pub use message_validator::ParsedTopic;
 use thiserror::Error;
 
 pub use crate::{NetworkMessageReceiver, manager::*};
@@ -15,7 +15,7 @@ pub trait MessageReceiver {
         propagation_source: PeerId,
         message_id: MessageId,
         message: Message,
-        topic_context: TopicContext,
+        parsed_topic: ParsedTopic,
     ) -> Result<(), Error>;
 }
 

anchor/message_receiver/src/manager.rs

@@ -6,8 +6,7 @@ use libp2p::{
     gossipsub::{Message, MessageAcceptance, MessageId},
 };
 use message_validator::{
-    DutiesProvider, TopicContext, ValidatedMessage, ValidatedSSVMessage, ValidationResult,
-    Validator,
+    DutiesProvider, ParsedTopic, ValidatedMessage, ValidatedSSVMessage, ValidationResult, Validator,
 };
 use operator_doppelganger::OperatorDoppelgangerService;
 use qbft_manager::QbftManager;
@@ -72,15 +71,15 @@ impl<E: types::EthSpec, S: SlotClock + 'static, D: DutiesProvider> MessageReceiv
         propagation_source: PeerId,
         message_id: MessageId,
         message: Message,
-        topic_context: TopicContext,
+        parsed_topic: ParsedTopic,
     ) -> Result<(), crate::Error> {
         let receiver = self.clone();
         self.processor.urgent_consensus.send_blocking(
             move || {
                 let span = debug_span!("message_receiver", msg=%message_id);
                 let _enter = span.enter();
 
-                let result = receiver.validator.validate(&message.data, &topic_context);
+                let result = receiver.validator.validate(&message.data, &parsed_topic);
 
                 let mut action = MessageAcceptance::from(&result);
 
@@ -180,10 +179,19 @@ impl<E: types::EthSpec, S: SlotClock + 'static, D: DutiesProvider> MessageReceiv
 
                 match ssv_message {
                     ValidatedSSVMessage::QbftMessage(qbft_message) => {
-                        if let Err(err) = receiver
-                            .qbft_manager
-                            .receive_data(signed_ssv_message, qbft_message)
-                        {
+                        // Re-query the proposer assignment immediately before dispatch so an
+                        // earlier validation observation cannot authorize later instance
+                        // allocation.
+                        let result = receiver.qbft_manager.receive_network_message(
+                            signed_ssv_message,
+                            qbft_message,
+                            |slot, validator_pubkey| {
+                                receiver
+                                    .validator
+                                    .proposer_assignment_at_slot(slot, validator_pubkey)
+                            },
+                        );
+                        if let Err(err) = result {
                             error!(gossipsub_message_id = ?message_id, ssv_msg_id = ?msg_id, ?err, "Unable to receive QBFT message");
                         }
                     }

anchor/message_sender/src/network.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use database::OwnOperatorId;
-use message_validator::{DutiesProvider, MessageAcceptance, TopicContext, Validator};
+use message_validator::validate_outbound_message;
 use openssl::{
     hash::MessageDigest,
     pkey::{PKey, Private},
@@ -15,38 +15,36 @@ use ssv_types::{
 use ssz::Encode;
 use subnet_service::SubnetService;
 use tokio::sync::{mpsc, mpsc::error::TrySendError, watch};
-use tracing::{debug, error, trace, warn};
+use tracing::{error, trace, warn};
 
 use crate::{Error, MessageCallback, MessageSender, SigningError};
 
 const SIGNER_NAME: &str = "message_sign_and_send";
 const SENDER_NAME: &str = "message_send";
 
 /// Configuration for creating a NetworkMessageSender
-pub struct NetworkMessageSenderConfig<S: SlotClock, D: DutiesProvider> {
+pub struct NetworkMessageSenderConfig<S: SlotClock> {
     pub processor: processor::Senders,
     /// Channel to send messages to the network. Tuple of (topic string, message bytes).
     /// Per SIP-43, the topic is determined by the message's slot.
     pub network_tx: mpsc::Sender<(String, Vec<u8>)>,
     pub private_key: Rsa<Private>,
     pub operator_id: OwnOperatorId,
-    pub validator: Option<Arc<Validator<S, D>>>,
     pub is_synced: watch::Receiver<bool>,
     pub subnet_service: Arc<SubnetService<S>>,
 }
 
-pub struct NetworkMessageSender<S: SlotClock, D: DutiesProvider> {
+pub struct NetworkMessageSender<S: SlotClock> {
     processor: processor::Senders,
     /// Channel to send messages to the network. Tuple of (topic string, message bytes).
     network_tx: mpsc::Sender<(String, Vec<u8>)>,
     private_key: PKey<Private>,
     operator_id: OwnOperatorId,
-    validator: Option<Arc<Validator<S, D>>>,
     is_synced: watch::Receiver<bool>,
     subnet_service: Arc<SubnetService<S>>,
 }
 
-impl<S: SlotClock + 'static, D: DutiesProvider> MessageSender for Arc<NetworkMessageSender<S, D>> {
+impl<S: SlotClock + 'static> MessageSender for Arc<NetworkMessageSender<S>> {
     fn sign_and_send(
         &self,
         message: UnsignedSSVMessage,
@@ -118,54 +116,36 @@ impl<S: SlotClock + 'static, D: DutiesProvider> MessageSender for Arc<NetworkMes
     }
 }
 
-impl<S: SlotClock + 'static, D: DutiesProvider> NetworkMessageSender<S, D> {
-    pub fn new(config: NetworkMessageSenderConfig<S, D>) -> Result<Arc<Self>, String> {
+impl<S: SlotClock + 'static> NetworkMessageSender<S> {
+    pub fn new(config: NetworkMessageSenderConfig<S>) -> Result<Arc<Self>, String> {
         let private_key = PKey::from_rsa(config.private_key)
             .map_err(|err| format!("Failed to create PKey from RSA: {err}"))?;
         Ok(Arc::new(Self {
             processor: config.processor,
             network_tx: config.network_tx,
             private_key,
             operator_id: config.operator_id,
-            validator: config.validator,
             is_synced: config.is_synced,
             subnet_service: config.subnet_service,
         }))
     }
 
     fn do_send(&self, message: SignedSSVMessage, committee_id: CommitteeId) {
-        let message_bytes = message.as_ssz_bytes();
-
-        // For outgoing messages, we use default TopicContext (no topic validation)
-        // since we're just doing a sanity check on our own message content
-        if let Some(validator) = self.validator.as_ref()
-            && let Err(err) = validator
-                .validate(&message_bytes, &TopicContext::default())
-                .as_result()
-        {
-            // `Reject` is more severe and can be punished by other peers. We should not have
-            // created this message ever, while `Ignore` can be triggered simply because the message
-            // is irrelevant by now.
-            if let MessageAcceptance::Reject = MessageAcceptance::from(err) {
-                warn!(?err, "Validation of outgoing message failed (Reject)");
-                debug!(msg = %message, "Failing message");
-            } else {
-                debug!(?err, "Validation of outgoing message failed (Ignore)");
-            }
-            return;
-        }
-
-        // Extract slot from message for slot-based topic routing (per SIP-43)
-        let message_slot = match message.ssv_message().extract_slot() {
-            Some(slot) => slot,
-            None => {
-                warn!(
+        let message_slot = match validate_outbound_message(&message) {
+            Ok(slot) => slot,
+            Err(err) => {
+                let ssv_message = message.ssv_message();
+                error!(
+                    ?err,
                     ?committee_id,
-                    "Cannot extract slot from message for topic routing"
+                    ssv_msg_id = ?ssv_message.msg_id(),
+                    msg_type = ?ssv_message.msg_type(),
+                    "Stateless validation of outgoing message failed"
                 );
                 return;
             }
         };
+        let message_bytes = message.as_ssz_bytes();
 
         // Use subnet service for slot-based subnet calculation
         let subnet = match self