feat(qbft_manager): wire EnvelopeProposer per-proposer QBFT instances (#1188)

Co-Authored-By: jnhsigmap <jason.harris@sigmaprime.io>

Author: jnhsigmap

anchor/qbft_manager/src/lib.rs

@@ -14,8 +14,8 @@ use slot_clock::SlotClock;
 use ssv_types::{
     CommitteeId, IndexSet, OperatorId,
     consensus::{
-        AggregatorCommitteeConsensusData, BeaconVote, GloasBeaconVote, ProposerConsensusData,
-        QbftData, QbftDataValidator,
+        AggregatorCommitteeConsensusData, BeaconVote, EnvelopeConsensusData, GloasBeaconVote,
+        ProposerConsensusData, QbftData, QbftDataValidator,
     },
     domain_type::DomainType,
     message::SignedSSVMessage,
@@ -30,7 +30,7 @@ use tokio::{
     },
     time::{Instant, sleep},
 };
-use tracing::{Instrument, debug, debug_span, error, warn};
+use tracing::{Instrument, debug_span, error, warn};
 use types::{ChainSpec, Epoch, EthSpec, Hash256, Slot};
 
 use crate::instance::qbft_instance;
@@ -96,6 +96,14 @@ pub enum ValidatorDutyKind {
     SyncCommitteeAggregator,
 }
 
+/// Unique identifier for an envelope-proposer QBFT instance (SIP-94 §6). Envelope
+/// signing is a single per-slot duty, so no `ValidatorDutyKind` discriminator.
+#[derive(Debug, Clone, Hash, PartialEq, Eq)]
+pub struct EnvelopeProposerInstanceId {
+    pub validator: PublicKeyBytes,
+    pub instance_height: InstanceHeight,
+}
+
 // Message that is passed around the QbftManager
 pub struct QbftMessage<D: QbftData> {
     pub kind: QbftMessageKind<D>,
@@ -148,6 +156,8 @@ pub struct QbftManager<E: EthSpec, S: SlotClock> {
     // QBFT instances for AggregatorCommitteeConsensusData
     aggregator_committee_instances:
         Map<AggregatorCommitteeInstanceId, AggregatorCommitteeConsensusData<E>>,
+    // QBFT instances voting on Gloas self-build envelope consensus data (SIP-94 §6)
+    envelope_consensus_data_instances: Map<EnvelopeProposerInstanceId, EnvelopeConsensusData>,
     // Utility to sign and serialize network messages
     message_sender: Arc<dyn MessageSender>,
     // Number of slots per epoch
@@ -178,6 +188,7 @@ impl<E: EthSpec, S: SlotClock + Clone + 'static> QbftManager<E, S> {
             beacon_vote_instances: DashMap::new(),
             gloas_beacon_vote_instances: DashMap::new(),
             aggregator_committee_instances: DashMap::new(),
+            envelope_consensus_data_instances: DashMap::new(),
             message_sender,
             slots_per_epoch,
             fork_schedule,
@@ -201,6 +212,12 @@ impl<E: EthSpec, S: SlotClock + Clone + 'static> QbftManager<E, S> {
         self.fork_schedule.active_fork_config(epoch).domain_type
     }
 
+    /// Whether the Ethereum Gloas (ePBS) fork is active at `slot`, per the consensus
+    /// spec. Distinct from the SSV protocol `fork_schedule`.
+    fn gloas_enabled_at_slot(&self, slot: Slot) -> bool {
+        self.spec.fork_name_at_slot::<E>(slot).gloas_enabled()
+    }
+
     // Decide a brand new qbft instance
     pub async fn decide_instance<D: QbftDecidable<E>>(
         &self,
@@ -283,9 +300,24 @@ impl<E: EthSpec, S: SlotClock + Clone + 'static> QbftManager<E, S> {
                     Some(Role::Aggregator) => ValidatorDutyKind::Aggregator,
                     Some(Role::SyncCommittee) => ValidatorDutyKind::SyncCommitteeAggregator,
                     Some(Role::EnvelopeProposer) => {
-                        // TODO: wire EnvelopeProposer instance routing (#1122)
-                        debug!(?msg_id, "EnvelopeProposer routing not yet wired");
-                        return Err(QbftError::RoleNotActive);
+                        let slot = types::Slot::new(qbft_message.height);
+                        // Defense in depth behind `validate_role_for_fork`: envelope QBFT
+                        // exists only post-Gloas.
+                        if !self.gloas_enabled_at_slot(slot) {
+                            warn!(%slot, "Ignoring EnvelopeProposer message before Gloas fork");
+                            return Err(QbftError::RoleNotActive);
+                        }
+                        let id = EnvelopeProposerInstanceId {
+                            validator,
+                            instance_height,
+                        };
+                        return self.pass_to_instance::<EnvelopeConsensusData>(
+                            id,
+                            WrappedQbftMessage {
+                                signed_message: full_message,
+                                qbft_message,
+                            },
+                        );
                     }
                     // Committee roles use DutyExecutor::Committee, not Validator
                     Some(Role::Committee | Role::AggregatorCommittee)
@@ -324,9 +356,8 @@ impl<E: EthSpec, S: SlotClock + Clone + 'static> QbftManager<E, S> {
                             qbft_message,
                         };
 
-                        // Gate the Gloas beacon-vote shape on Ethereum's Gloas (ePBS) fork,
-                        // read from the consensus spec, rather than an SSV-internal fork.
-                        if self.spec.fork_name_at_slot::<E>(slot).gloas_enabled() {
+                        // Gate the Gloas beacon-vote shape on Ethereum's Gloas (ePBS) fork using Ethereum consensus spec.
+                        if self.gloas_enabled_at_slot(slot) {
                             self.pass_to_instance::<GloasBeaconVote>(id, wrapped)
                         } else {
                             self.pass_to_instance::<BeaconVote>(id, wrapped)
@@ -416,6 +447,8 @@ impl<E: EthSpec, S: SlotClock + Clone + 'static> QbftManager<E, S> {
                 .retain(|k, _| *k.instance_height >= cutoff.as_usize());
             self.aggregator_committee_instances
                 .retain(|k, _| *k.instance_height >= cutoff.as_usize());
+            self.envelope_consensus_data_instances
+                .retain(|k, _| *k.instance_height >= cutoff.as_usize());
         }
     }
 }
@@ -560,6 +593,26 @@ impl<E: EthSpec> QbftDecidable<E> for AggregatorCommitteeConsensusData<E> {
     }
 }
 
+impl<E: EthSpec> QbftDecidable<E> for EnvelopeConsensusData {
+    type Id = EnvelopeProposerInstanceId;
+
+    fn get_map<S: SlotClock>(manager: &QbftManager<E, S>) -> &Map<Self::Id, Self> {
+        &manager.envelope_consensus_data_instances
+    }
+
+    fn instance_height(&self, id: &Self::Id) -> InstanceHeight {
+        id.instance_height
+    }
+
+    fn message_id(domain: &DomainType, id: &Self::Id) -> MessageId {
+        MessageId::new(
+            domain,
+            Role::EnvelopeProposer,
+            &DutyExecutor::Validator(id.validator),
+        )
+    }
+}
+
 #[derive(Debug, Clone)]
 pub enum QbftError {
     QueueClosedError,

anchor/qbft_manager/src/tests.rs

@@ -40,6 +40,7 @@ use super::{
 use crate::instance::qbft_instance;
 
 mod aggregator_tests;
+mod envelope_dispatch_tests;
 mod gloas_dispatch_tests;
 mod setup;
 mod timeout_tests;
@@ -961,115 +962,4 @@ mod manager_tests {
             result
         );
     }
-
-    #[tokio::test]
-    async fn envelope_proposer_any_executor_decodes_as_validator_transient_role_not_active() {
-        // `MessageId::new` starts from a zeroed 56-byte buffer and writes the executor bytes
-        // in place: `DutyExecutor::Validator` fills bytes 8..56, `DutyExecutor::Committee` fills
-        // bytes 24..56. Because `PublicKeyBytes::empty()` and `CommitteeId([0; 32])` are both
-        // all-zero, each executor writes ONLY zeros into an already-zero buffer, so role 9
-        // (`Role::EnvelopeProposer`) encodes to the identical 56-byte `MessageId` regardless of
-        // the executor passed in.
-        //
-        // `MessageId::duty_executor` then selects the executor purely from the role, and role 9
-        // is hard-wired to the `Validator` arm (bytes 8..55). So `receive_data` always enters the
-        // `Some(DutyExecutor::Validator(_))` branch and hits the `Some(Role::EnvelopeProposer)`
-        // arm, which returns `QbftError::RoleNotActive` unconditionally (routing not wired,
-        // TODO #1122). A committee-executor `EnvelopeProposer` is therefore unconstructable, and
-        // the `Role::EnvelopeProposer` listing in the `DutyExecutor::Committee` arm is unreachable
-        // for role 9 — it exists only for match exhaustiveness, so its `InconsistentMessageId` can
-        // never fire here.
-        use fork::{Fork, ForkSchedule};
-        use message_sender::testing::MockMessageSender;
-        use ssv_types::{
-            RSA_SIGNATURE_SIZE,
-            consensus::{QbftMessage, QbftMessageType},
-            message::{MsgType, SSVMessage, SignedSSVMessage},
-        };
-        use ssz::Encode;
-
-        // Arrange: build the manager once. The fork/spec values are irrelevant here — the
-        // `Validator`/`EnvelopeProposer` arm returns before consulting either.
-        let setup = setup_test(1);
-        let fork_schedule = ForkSchedule::new(Fork::Boole, DomainType::default(), "test");
-        let config = processor::Config {
-            max_workers: 4,
-            queue_size: Default::default(),
-        };
-        let senders = processor::spawn(config, setup.executor);
-        let (network_tx, _network_rx) = mpsc::unbounded_channel();
-
-        let manager = QbftManager::<types::MainnetEthSpec, _>::new(
-            senders,
-            OperatorId(1).into(),
-            setup.clock,
-            Arc::new(MockMessageSender::new(network_tx, OperatorId(1))),
-            NonZeroU64::new(32).expect("slots_per_epoch is non-zero"),
-            Arc::new(fork_schedule),
-            Arc::new(types::ChainSpec::mainnet()),
-        )
-        .expect("Manager creation should succeed");
-
-        // Constructs an `EnvelopeProposer` message for the given executor, returning the signed
-        // message plus its `QbftMessage`. Avoids duplicating the construction block per executor.
-        let build = |executor: &DutyExecutor| -> (SignedSSVMessage, QbftMessage) {
-            let msg_id = MessageId::new(&DomainType([0; 4]), Role::EnvelopeProposer, executor);
-            let qbft_message = QbftMessage {
-                qbft_message_type: QbftMessageType::Proposal,
-                height: 100,
-                round: 1,
-                identifier: (&msg_id).into(),
-                root: Hash256::from([0u8; 32]),
-                data_round: 1,
-                round_change_justification: ssv_types::VariableList::empty(),
-                prepare_justification: ssv_types::VariableList::empty(),
-            };
-            let ssv_msg = SSVMessage::new(
-                MsgType::SSVConsensusMsgType,
-                msg_id,
-                qbft_message.as_ssz_bytes(),
-            )
-            .expect("SSVMessage creation should succeed");
-            let signed_msg = SignedSSVMessage::new(
-                vec![[0xAA; RSA_SIGNATURE_SIZE]],
-                vec![OperatorId(1)],
-                ssv_msg,
-                vec![],
-            )
-            .expect("SignedSSVMessage creation should succeed");
-            (signed_msg, qbft_message)
-        };
-
-        // Make the byte-identity invariant explicit: both executors encode to the same 56 bytes.
-        let validator_msg_id = MessageId::new(
-            &DomainType([0; 4]),
-            Role::EnvelopeProposer,
-            &DutyExecutor::Validator(bls::PublicKeyBytes::empty()),
-        );
-        let committee_msg_id = MessageId::new(
-            &DomainType([0; 4]),
-            Role::EnvelopeProposer,
-            &DutyExecutor::Committee(CommitteeId([0; 32])),
-        );
-        assert_eq!(
-            validator_msg_id.as_ref(),
-            committee_msg_id.as_ref(),
-            "validator- and committee-executor `EnvelopeProposer` must encode to the same 56 bytes for role 9"
-        );
-
-        // Act + Assert: both executor variants route through the same
-        // `Validator`/`EnvelopeProposer` arm and return transient `RoleNotActive` (routing
-        // not wired), never `InconsistentMessageId`.
-        for executor in [
-            DutyExecutor::Validator(bls::PublicKeyBytes::empty()),
-            DutyExecutor::Committee(CommitteeId([0; 32])),
-        ] {
-            let (signed_msg, qbft_message) = build(&executor);
-            let result = manager.receive_data(signed_msg, qbft_message);
-            assert!(
-                matches!(result, Err(QbftError::RoleNotActive)),
-                "`EnvelopeProposer` always decodes as `Validator` and must be transient `RoleNotActive`, got: {result:?}"
-            );
-        }
-    }
 }