feat: admit one envelope dissemination per signer and sign by binding

anchor/common/dissemination_store/Cargo.toml

@@ -10,3 +10,7 @@ parking_lot = { workspace = true }
 ssv_types = { workspace = true }
 tokio = { workspace = true }
 types = { workspace = true }
+
+[dev-dependencies]
+# `start_paused` in the wait tests: the deadline is the only thing they wait on.
+tokio = { workspace = true, features = ["test-util"] }

anchor/message_receiver/src/manager.rs

@@ -199,14 +199,19 @@ impl<E: types::EthSpec, S: SlotClock + 'static, D: DutiesProvider> MessageReceiv
                             error!(gossipsub_message_id = ?message_id, ssv_msg_id = ?msg_id, ?err, "Unable to receive partial signature message");
                         }
                     }
-                    ValidatedSSVMessage::EnvelopeDissemination(dissemination) => {
+                    ValidatedSSVMessage::EnvelopeDissemination {
+                        signer,
+                        dissemination,
+                    } => {
                         // Validation admits the class only for validator-scoped role-9 message
                         // IDs, so the duty executor is always a validator public key.
                         match msg_id.duty_executor() {
                             Some(DutyExecutor::Validator(validator_pubkey)) => {
-                                receiver
-                                    .dissemination_store
-                                    .insert(validator_pubkey, dissemination);
+                                receiver.dissemination_store.insert(
+                                    validator_pubkey,
+                                    signer,
+                                    dissemination,
+                                );
                             }
                             _ => {
                                 error!(gossipsub_message_id = ?message_id, ssv_msg_id = ?msg_id, "Envelope dissemination without a validator duty executor");

anchor/message_validator/src/dissemination.rs

@@ -16,18 +16,19 @@ use crate::{
 /// The class is structural-only at this layer: the inner envelope must SSZ-decode as a
 /// blinded execution payload envelope (shape, Reject-class), but the checks binding it to
 /// the block-QBFT decision are runner concerns, and validation never judges the envelope's
-/// content against the decision. Dedup is first-valid per (`MessageId`, slot): the first
-/// message passing all other rules is recorded, and further dissemination messages for the
+/// content against the decision. Dedup is one per (`MessageId`, signer, slot): a signer's
+/// first message passing all other rules is recorded, and its further disseminations for the
 /// tuple are Ignore regardless of content or peer (an honest origin retry can repeat one
 /// after the recipient's gossip duplicate cache expires, so repetition does not prove peer
-/// fault).
+/// fault). Another committee member's dissemination for the same slot is admitted on its own
+/// budget.
 ///
-/// First-valid means first STRUCTURALLY valid, by design: SIP-94 accepts that a Byzantine
-/// committee member can consume a slot's dissemination budget with a decision-unbound or
-/// payload-unbound (but well-formed) carrier, costing at most one missed self-build reveal
-/// (never a wrong payload on chain). Do not move semantic rejection into a
-/// replacement-capable store; the named hardening for that trade is sign-all, a protocol
-/// change.
+/// Admitting one per signer is what lets the runner pick by content (SIP-94 §6: it signs the
+/// first dissemination that passes the decision bindings, not the first that arrives), so a
+/// Byzantine committee member cannot cost the reveal merely by winning the race with a
+/// well-formed but decision-unbound carrier. Forwarding stays bounded because only committee
+/// members pass validation. The residual is a binding-passing forgery, whose `payload_root` no
+/// operator can check; the named hardening for that is sign-all, a protocol change.
 pub(crate) fn validate_envelope_dissemination(
     validation_context: ValidationContext<impl SlotClock>,
     duty_state: &mut DutyState,
@@ -92,10 +93,13 @@ pub(crate) fn validate_envelope_dissemination(
         duty_provider.clone(),
     )?;
 
-    // Rule: first-valid dedup per (`MessageId`, slot), signer-independent. Ignore-class.
-    if duty_state.is_dissemination_recorded(slot) {
+    // Rule: dedup per (`MessageId`, signer, slot). Ignore-class.
+    if duty_state
+        .get_or_create_operator(&signer)
+        .is_dissemination_recorded(slot)
+    {
         return Err(ValidationFailure::RelayedDuplicateMessage {
-            got: format!("envelope dissemination for slot {slot}"),
+            got: format!("envelope dissemination for slot {slot} from operator {signer}"),
         });
     }
 
@@ -108,11 +112,14 @@ pub(crate) fn validate_envelope_dissemination(
 
     verify_single_signer(&validation_context, signer)?;
 
-    // Record only after every other rule passed, so a rejected message cannot consume the
-    // slot's single dissemination budget.
-    duty_state.record_dissemination(slot, &signer);
+    // Record only after every other rule passed, so a rejected message cannot consume this
+    // signer's dissemination budget for the slot.
+    operator_state.record_dissemination(slot);
 
-    Ok(ValidatedSSVMessage::EnvelopeDissemination(dissemination))
+    Ok(ValidatedSSVMessage::EnvelopeDissemination {
+        signer,
+        dissemination,
+    })
 }
 
 #[cfg(test)]
@@ -286,7 +293,9 @@ mod tests {
         );
 
         match result {
-            Ok(ValidatedSSVMessage::EnvelopeDissemination(d)) => {
+            Ok(ValidatedSSVMessage::EnvelopeDissemination {
+                dissemination: d, ..
+            }) => {
                 assert_eq!(d.slot, Slot::new(TEST_SLOT));
             }
             other => panic!("expected accepted dissemination, got {other:?}"),
@@ -325,12 +334,10 @@ mod tests {
     }
 
     #[test]
-    fn second_dissemination_for_slot_ignored_regardless_of_signer() {
-        let (committee_info, private_key, mut map) = four_node_committee_and_keypair();
-        // A second operator with its own key, so the dedup is proven signer-independent.
-        let (private_key_2, public_key_2) = generate_test_key_pair();
-        map.insert(OperatorId(2), public_key_2);
+    fn second_dissemination_from_the_same_signer_ignored() {
+        let (committee_info, private_key, map) = four_node_committee_and_keypair();
         let mut duty_state = DutyState::new(64);
+        let provider = Arc::new(MockDutiesProvider::default());
 
         let first =
             create_signed_dissemination(Role::EnvelopeProposer, OperatorId(1), &private_key);
@@ -342,15 +349,12 @@ mod tests {
             1,
             Some(0),
         );
-        validate_envelope_dissemination(
-            ctx,
-            &mut duty_state,
-            Arc::new(MockDutiesProvider::default()),
-        )
-        .expect("first dissemination must be accepted");
+        validate_envelope_dissemination(ctx, &mut duty_state, provider.clone())
+            .expect("first dissemination must be accepted");
 
+        // Same signer, same slot: the signer's budget for the tuple is spent.
         let second =
-            create_signed_dissemination(Role::EnvelopeProposer, OperatorId(2), &private_key_2);
+            create_signed_dissemination(Role::EnvelopeProposer, OperatorId(1), &private_key);
         let ctx = create_dissemination_context(
             &second,
             &committee_info,
@@ -359,24 +363,59 @@ mod tests {
             1,
             Some(0),
         );
-        let result = validate_envelope_dissemination(
-            ctx,
-            &mut duty_state,
-            Arc::new(MockDutiesProvider::default()),
-        );
+        let result = validate_envelope_dissemination(ctx, &mut duty_state, provider);
 
         match &result {
             Err(failure @ ValidationFailure::RelayedDuplicateMessage { .. }) => {
                 assert_eq!(
                     MessageAcceptance::from(failure),
                     MessageAcceptance::Ignore,
-                    "a further dissemination for a recorded slot must be Ignore, not Reject"
+                    "a signer's further dissemination for a recorded slot must be Ignore, not Reject: an honest retry can repeat one after the gossip duplicate cache expires"
                 );
             }
             other => panic!("expected RelayedDuplicateMessage, got {other:?}"),
         }
     }
 
+    #[test]
+    fn dissemination_from_a_second_signer_accepted_for_the_same_slot() {
+        // SIP-94 §7 dedup is per (`MessageId`, signer, slot). Admitting one carrier per
+        // committee member is what lets the runner choose by content instead of by arrival, so
+        // a first carrier that fails the decision bindings cannot cost the slot's reveal.
+        let (committee_info, private_key, mut map) = four_node_committee_and_keypair();
+        let (private_key_2, public_key_2) = generate_test_key_pair();
+        map.insert(OperatorId(2), public_key_2);
+        let mut duty_state = DutyState::new(64);
+        let provider = Arc::new(MockDutiesProvider::default());
+
+        let first =
+            create_signed_dissemination(Role::EnvelopeProposer, OperatorId(1), &private_key);
+        let ctx = create_dissemination_context(
+            &first,
+            &committee_info,
+            Role::EnvelopeProposer,
+            &map,
+            1,
+            Some(0),
+        );
+        validate_envelope_dissemination(ctx, &mut duty_state, provider.clone())
+            .expect("first dissemination must be accepted");
+
+        let second =
+            create_signed_dissemination(Role::EnvelopeProposer, OperatorId(2), &private_key_2);
+        let ctx = create_dissemination_context(
+            &second,
+            &committee_info,
+            Role::EnvelopeProposer,
+            &map,
+            1,
+            Some(0),
+        );
+
+        validate_envelope_dissemination(ctx, &mut duty_state, provider)
+            .expect("a second committee member's dissemination must be accepted on its own budget");
+    }
+
     #[test]
     fn two_signers_rejected() {
         let (committee_info, private_key, map) = four_node_committee_and_keypair();

anchor/message_validator/src/duty_state.rs

@@ -52,11 +52,6 @@ pub(crate) struct DutyState {
     operators: HashMap<OperatorId, OperatorState>,
     /// The number of slots for which state is stored (defines the size of the circular buffer)
     stored_slot_count: usize,
-    /// Slot-indexed ring recording whether an envelope dissemination was already accepted for a
-    /// slot of this `MessageId` (SIP-94 §7 first-valid dedup: one dissemination per
-    /// (`MessageId`, slot), signer-independent). Allocated on first record: only
-    /// `Role::EnvelopeProposer` message IDs ever populate it.
-    disseminated_slots: Vec<Option<Slot>>,
 }
 
 impl DutyState {
@@ -65,30 +60,6 @@ impl DutyState {
         Self {
             operators: HashMap::new(),
             stored_slot_count,
-            disseminated_slots: Vec::new(),
-        }
-    }
-
-    /// True if a dissemination was already recorded for `slot` (SIP-94 §7: further
-    /// dissemination messages for the tuple are Ignore, regardless of content or peer).
-    pub(crate) fn is_dissemination_recorded(&self, slot: Slot) -> bool {
-        !self.disseminated_slots.is_empty()
-            && self.disseminated_slots[slot.as_usize() % self.disseminated_slots.len()]
-                == Some(slot)
-    }
-
-    /// Records the accepted dissemination for `slot` and creates the sender's signer state so
-    /// the duty counts toward `signer`'s per-epoch ring occupancy.
-    pub(crate) fn record_dissemination(&mut self, slot: Slot, signer: &OperatorId) {
-        if self.disseminated_slots.is_empty() {
-            self.disseminated_slots = vec![None; self.stored_slot_count];
-        }
-        let index = slot.as_usize() % self.disseminated_slots.len();
-        self.disseminated_slots[index] = Some(slot);
-
-        let operator_state = self.get_or_create_operator(signer);
-        if operator_state.is_first_message_for_duty(slot) {
-            operator_state.set_signer_state(&slot, SignerState::new(slot, FIRST_ROUND));
         }
     }
 
@@ -249,6 +220,29 @@ impl OperatorState {
         self.get_signer_state(&slot).is_none()
     }
 
+    /// True if this signer already had a dissemination accepted for `slot` (SIP-94 §7 dedup,
+    /// one per (`MessageId`, signer, slot): further disseminations for the tuple are Ignore,
+    /// regardless of content or peer). A ring entry held by a different slot reads as absent,
+    /// like every other per-slot lookup here.
+    pub(crate) fn is_dissemination_recorded(&self, slot: Slot) -> bool {
+        self.get_signer_state(&slot)
+            .is_some_and(|state| state.dissemination_recorded)
+    }
+
+    /// Records this signer's accepted dissemination for `slot`, creating its signer state
+    /// when the dissemination is the duty's first message so the duty counts toward the
+    /// per-epoch ring occupancy and advances `max_slot`, exactly as a first partial signature
+    /// would. An existing state is kept, not replaced: a partial signature may have arrived
+    /// first, and its counts must survive.
+    pub(crate) fn record_dissemination(&mut self, slot: Slot) {
+        if self.is_first_message_for_duty(slot) {
+            self.set_signer_state(&slot, SignerState::new(slot, FIRST_ROUND));
+        }
+        if let Some(state) = self.get_signer_state_mut(&slot) {
+            state.dissemination_recorded = true;
+        }
+    }
+
     /// Updates the SignerState for the given slot.
     ///
     /// If a state already exists and the incoming consensus round is higher,
@@ -311,6 +305,15 @@ pub(crate) struct SignerState {
     pub(crate) proposal_hash: Option<[u8; 32]>,
     /// A set of CommitteeIds indicating which committees have already been seen.
     seen_signers: HashSet<CommitteeId>,
+    /// True once an envelope dissemination from this signer was accepted for this slot
+    /// (SIP-94 §7). Only `Role::EnvelopeProposer` message IDs ever set it.
+    ///
+    /// Nothing clears it, and that rests on role 9 having no consensus path: `OperatorState::
+    /// update` is the one writer that replaces a `SignerState` outright, and it is reachable
+    /// only for QBFT roles, which `Role::EnvelopeProposer` is not (`max_round()` is `None`, so
+    /// consensus messages for it are rejected before any state update). A future role that both
+    /// disseminates and runs QBFT would need this bit carried across the replacement.
+    dissemination_recorded: bool,
     /// Accepted signing roots for the root-budgeted kinds, boxed and lazily allocated on the
     /// first such packet: every role's ring entries share this struct, but only
     /// `Role::ProposerPreferences` messages can ever populate it
@@ -337,6 +340,7 @@ impl SignerState {
             message_counts: MessageCounts::default(),
             proposal_hash: None,
             seen_signers: HashSet::new(),
+            dissemination_recorded: false,
             root_budgets: None,
         }
     }
@@ -415,6 +419,30 @@ mod tests {
         tests::{QbftMessageBuilder, create_signed_consensus_message},
     };
 
+    /// The dissemination flag lives in a slot ring, so a read for a slot that merely aliases a
+    /// recorded one must not inherit its flag. Without the ring's stored-slot filter an honest
+    /// dissemination exactly `stored_slot_count` slots later would be silently Ignored, and the
+    /// monotonic-slot guard would not catch it because the incoming slot is higher.
+    #[test]
+    fn dissemination_record_does_not_alias_across_the_ring() {
+        const RING: usize = 64;
+        let mut duty_state = DutyState::new(RING);
+        let recorded = Slot::new(5);
+        let aliased = recorded + RING as u64;
+
+        let operator_state = duty_state.get_or_create_operator(&OperatorId(1));
+        operator_state.record_dissemination(recorded);
+
+        assert!(
+            operator_state.is_dissemination_recorded(recorded),
+            "the recorded slot must read as recorded"
+        );
+        assert!(
+            !operator_state.is_dissemination_recorded(aliased),
+            "slot {aliased} shares a ring index with the recorded slot {recorded} and must read as absent"
+        );
+    }
+
     #[test]
     fn test_duty_state_update() {
         let mut duty_state = DutyState::new(10);

anchor/message_validator/src/lib.rs

@@ -329,7 +329,12 @@ impl From<SignedSSVMessageError> for ValidationFailure {
 pub enum ValidatedSSVMessage {
     QbftMessage(QbftMessage),
     PartialSignatureMessages(PartialSignatureMessages),
-    EnvelopeDissemination(EnvelopeDissemination),
+    /// The carrier plus the committee member that signed it. Validation proves there is
+    /// exactly one signer, and the envelope duty names it when rejecting a candidate.
+    EnvelopeDissemination {
+        signer: OperatorId,
+        dissemination: EnvelopeDissemination,
+    },
 }
 
 #[derive(Debug)]