Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mergify merged 3 commits into from
Jul 30, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
44c949e
wiring for envelope proposer to qbft manager layer
jnhsigmap Jul 29, 2026
7193ad7
Merge branch 'epbs' into feat/1122/qbft-manager
jnhsigmap Jul 29, 2026
480dea8
add fork gating comment back
jnhsigmap Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 62 additions & 9 deletions anchor/qbft_manager/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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;
Expand Down Expand Up @@ -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>,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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());
}
}
}
Expand Down Expand Up @@ -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,
Expand Down
112 changes: 1 addition & 111 deletions anchor/qbft_manager/src/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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:?}"
);
}
}
}
Loading
Loading