refactor: phase metadata service across full slot duration (#781)

Required for committee-based partial signature collection. The full list of attesting and sync committee validators must be available at slot start before produce_selection_proof is called, which was not possible with the previous single-phase metadata approach.


  - Refactor MetadataService to use a phased approach that spans the full slot duration, rather than computing all metadata at a single point
- Split metadata into three distinct phases: VotingAssignments at slot start, VotingContext at 1/3 slot, and AggregationAssignments at 2/3 slot
- Add event-driven coordination between MetadataService and DutiesService using watch channels to ensure duties are polled before reading them
- Update lighthouse dependencies to a fork that exposes the new duties service polling events
- Update produce_selection_proof and produce_sync_selection_proof to use committee-based partial signature collection after the Boole fork
- These methods now request signatures from all committee operators rather than just the validator's operators
- VotingAssignments provides the expected message count for committee-based batching
- Update lighthouse dependencies to a fork that exposes the new duties service polling events


Co-Authored-By: shane-moore <skm1790@gmail.com>

Author: shane-moore

Cargo.lock

@@ -542,6 +542,7 @@ impl Client {
             spec.clone(),
             genesis_validators_root,
             config.impostor.is_none().then_some(key),
+            fork_schedule.clone(),
             config.gas_limit,
             config.builder_boost_factor,
             config.prefer_builder_proposals,

anchor/client/src/lib.rs

@@ -28,7 +28,7 @@ use types::{
     ForkName, Hash256, Slot, SyncCommitteeContribution,
 };
 
-use crate::{ValidatorIndex, message::*};
+use crate::{CommitteeId, ValidatorIndex, message::*, partial_sig::PartialSignatureKind};
 //                          UnsignedSSVMessage
 //            ----------------------------------------------
 //            |                                            |
@@ -861,6 +861,39 @@ impl QbftData for BeaconVote {
     }
 }
 
+/// Identifies a batch of pre-consensus selection proofs for a committee.
+/// All operators compute the same hash for a given `(slot, committee_id)` pair, ensuring
+/// consistent batching across the network.
+///
+/// Unlike `BeaconVote::hash()` which hashes decided consensus data, this hash is an
+/// artificial correlation identifier since pre-consensus selection proofs have different
+/// signing roots (attestation vs sync committee selection proofs use different domains).
+#[derive(Debug, Clone)]
+pub struct SelectionProofBatchId {
+    pub slot: Slot,
+    pub committee_id: CommitteeId,
+}
+
+impl SelectionProofBatchId {
+    pub fn new(slot: Slot, committee_id: CommitteeId) -> Self {
+        Self { slot, committee_id }
+    }
+
+    /// Compute deterministic hash for batching correlation.
+    ///
+    /// The hash includes the SSZ encoding of `PartialSignatureKind::AggregatorCommitteePartialSig`
+    /// as a domain separator to prevent collision with other hashes in the system.
+    pub fn hash(&self) -> Hash256 {
+        let mut hasher = Sha256::new();
+        // Domain separator: SSZ encoding of the partial signature kind
+        hasher.update(PartialSignatureKind::AggregatorCommitteePartialSig.as_ssz_bytes());
+        hasher.update(self.slot.as_u64().to_le_bytes());
+        hasher.update(self.committee_id.0);
+
+        Hash256::from_slice(&hasher.finalize())
+    }
+}
+
 pub struct BeaconVoteValidator<E: EthSpec> {
     slot: Slot,
     // `None` if slashing protection is disabled via CLI.
@@ -1897,4 +1930,75 @@ mod tests {
             err => panic!("Expected DifferentCheckpoint error, got: {:?}", err),
         }
     }
+
+    // ═══════════════════════════════════════════════════════════════════════════════
+    // SelectionProofBatchId Tests
+    // ═══════════════════════════════════════════════════════════════════════════════
+
+    #[test]
+    fn test_selection_proof_batch_id_same_inputs_same_hash() {
+        let slot = Slot::new(12345);
+        let committee_id = CommitteeId::from([0u8; 32]);
+
+        // Compute hash twice with same inputs
+        let batch_id1 = SelectionProofBatchId::new(slot, committee_id);
+        let batch_id2 = SelectionProofBatchId::new(slot, committee_id);
+
+        // Same inputs should produce same hash
+        assert_eq!(batch_id1.hash(), batch_id2.hash());
+    }
+
+    #[test]
+    fn test_selection_proof_batch_id_different_slots() {
+        let committee_id = CommitteeId::from([1u8; 32]);
+
+        // Different slots
+        let slot1 = Slot::new(12345);
+        let slot2 = Slot::new(12346);
+
+        let batch_id1 = SelectionProofBatchId::new(slot1, committee_id);
+        let batch_id2 = SelectionProofBatchId::new(slot2, committee_id);
+
+        // Different slots should produce different hashes
+        assert_ne!(batch_id1.hash(), batch_id2.hash());
+    }
+
+    #[test]
+    fn test_selection_proof_batch_id_different_committees() {
+        let slot = Slot::new(12345);
+
+        // Different committee IDs
+        let committee_id1 = CommitteeId::from([1u8; 32]);
+        let committee_id2 = CommitteeId::from([2u8; 32]);
+
+        let batch_id1 = SelectionProofBatchId::new(slot, committee_id1);
+        let batch_id2 = SelectionProofBatchId::new(slot, committee_id2);
+
+        // Different committees should produce different hashes
+        assert_ne!(batch_id1.hash(), batch_id2.hash());
+    }
+
+    #[test]
+    fn test_selection_proof_batch_id_deterministic_across_operators() {
+        // This test verifies that the hash is deterministic and identical
+        // across different operators for the same inputs
+
+        let slot = Slot::new(42);
+        let committee_bytes = [
+            0x12, 0x34, 0x56, 0x78, 0x9a, 0xbc, 0xde, 0xf0, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66,
+            0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff, 0x00, 0x10, 0x20, 0x30, 0x40,
+            0x50, 0x60, 0x70, 0x80,
+        ];
+        let committee_id = CommitteeId::from(committee_bytes);
+
+        let batch_id = SelectionProofBatchId::new(slot, committee_id);
+        let expected_hash = batch_id.hash();
+
+        // Simulate computing on different "operators" (same calculation repeated)
+        for _ in 0..3 {
+            let batch_id = SelectionProofBatchId::new(slot, committee_id);
+            // All operators should get the same hash
+            assert_eq!(batch_id.hash(), expected_hash);
+        }
+    }
 }

anchor/common/ssv_types/src/consensus.rs

@@ -311,7 +311,7 @@ pub trait QbftDecidable: QbftData<Hash = Hash256> + Send + Sync + 'static {
             dashmap::Entry::Occupied(entry) => entry.get().clone(),
             dashmap::Entry::Vacant(entry) => {
                 // There is not an instance running yet, store the sender and spawn a new instance
-                // with the reeiver
+                // with the receiver
                 let (tx, rx) = mpsc::unbounded_channel();
                 let span = debug_span!("qbft_instance", instance_id = ?entry.key());
                 let tx = entry.insert(tx);

anchor/qbft_manager/src/lib.rs

@@ -10,6 +10,7 @@ bls = { workspace = true }
 database = { workspace = true }
 eth2 = { workspace = true }
 ethereum_ssz = { workspace = true }
+fork = { workspace = true }
 futures = { workspace = true }
 hex = { workspace = true }
 lru = { workspace = true }