merge queue: checking epbs (68acc2e) and #1125 together

anchor/validator_store/src/instrumentation.rs

@@ -23,12 +23,16 @@ pub mod checkpoints {
     pub const DUTY_FAILED: &str = "duty_failed";
 }
 
-/// Telemetry classification of `collect_signature` failures during PTC duties.
+/// Telemetry classification of `collect_signature` failures.
+///
+/// Shared by every single-validator signing path (PTC, ProposerPreferences, ...): the
+/// classification depends only on the generic `collect_signature` error, never on the role, so
+/// each caller maps these classes onto its own log message and reconstruction-failure metric.
 ///
 /// Returned as an enum rather than a label string because the class drives two independent
 /// effects in the caller: the log level and whether a reconstruction-failure metric is
 /// incremented at all.
-pub enum PtcFailureClass {
+pub enum CollectionFailureClass {
     /// The committee never reached the partial signature threshold. This surfaces as
     /// `QueueClosedError` because the collector is evicted after
     /// `SIGNATURE_COLLECTOR_RETAIN_SLOTS`, dropping the result channel while we await it, so it
@@ -41,22 +45,27 @@ pub enum PtcFailureClass {
     NonCollection,
 }
 
-pub fn classify_ptc_collection_failure(error: &Error) -> PtcFailureClass {
+pub fn classify_collection_failure(error: &Error) -> CollectionFailureClass {
     match error {
         // The inner match is deliberately wildcard-free so a future `CollectionError` variant
         // forces a conscious classification decision here at compile time.
         Error::SpecificError(SpecificError::SignatureCollectionFailed(collection_error)) => {
             match collection_error {
+                // `CollectionTimeout` is synthesized by `sign_proposer_preferences` when its
+                // bounded collection-wait deadline elapses; the collector itself never emits it. It
+                // shares the no-signature bucket with the `QueueClosedError` threshold-not-reached
+                // signal because a deadline elapse likewise means the threshold was not reached
+                // within the wait window.
                 CollectionError::QueueClosedError | CollectionError::CollectionTimeout => {
-                    PtcFailureClass::NoSignature
+                    CollectionFailureClass::NoSignature
                 }
                 CollectionError::QueueFullError
                 | CollectionError::OwnOperatorIdUnknown
                 | CollectionError::EmptySignature
-                | CollectionError::RecoverError(_) => PtcFailureClass::Infra,
+                | CollectionError::RecoverError(_) => CollectionFailureClass::Infra,
             }
         }
-        _ => PtcFailureClass::NonCollection,
+        _ => CollectionFailureClass::NonCollection,
     }
 }
 

anchor/validator_store/src/lib.rs

@@ -81,7 +81,7 @@ use validator_store::{
     ValidatorStore,
 };
 
-use crate::instrumentation::PtcFailureClass;
+use crate::instrumentation::CollectionFailureClass;
 
 /// Number of epochs of slashing protection history to keep.
 ///
@@ -99,6 +99,10 @@ const SELECTION_PROOF_LOG_NAME: &str = "selection proof";
 const SYNC_SELECTION_PROOF_LOG_NAME: &str = "sync selection proof";
 const SYNC_COMMITTEE_CONTRIBUTION_LOG_NAME: &str = "sync committee contribution";
 
+/// Upper bound, in slots, on how long `sign_proposer_preferences` waits for a validator's signature
+/// to be reconstructed.
+const PROPOSER_PREFERENCES_COLLECTION_TIMEOUT_SLOTS: u32 = 2;
+
 /// A request to collect a committee signature for a single validator.
 ///
 /// The shared fields (`validator`, `signing_root`) drive `collect_prepared_signatures`,
@@ -871,8 +875,8 @@ impl<T: SlotClock, E: EthSpec, C: ConsensusDecider<E> + 'static> AnchorValidator
         validator_pubkey: &PublicKeyBytes,
         slot: Slot,
     ) {
-        match instrumentation::classify_ptc_collection_failure(error) {
-            PtcFailureClass::NoSignature => {
+        match instrumentation::classify_collection_failure(error) {
+            CollectionFailureClass::NoSignature => {
                 warn!(
                     ?validator_pubkey,
                     %slot,
@@ -884,7 +888,7 @@ impl<T: SlotClock, E: EthSpec, C: ConsensusDecider<E> + 'static> AnchorValidator
                     &[metrics::PTC_FAILURE_NO_SIGNATURE],
                 );
             }
-            PtcFailureClass::Infra => {
+            CollectionFailureClass::Infra => {
                 error!(
                     ?validator_pubkey,
                     %slot,
@@ -896,7 +900,7 @@ impl<T: SlotClock, E: EthSpec, C: ConsensusDecider<E> + 'static> AnchorValidator
                     &[metrics::PTC_FAILURE_INFRA],
                 );
             }
-            PtcFailureClass::NonCollection => {
+            CollectionFailureClass::NonCollection => {
                 error!(
                     ?validator_pubkey,
                     %slot,
@@ -907,6 +911,55 @@ impl<T: SlotClock, E: EthSpec, C: ConsensusDecider<E> + 'static> AnchorValidator
         }
     }
 
+    /// Classify and report a ProposerPreferences signature-collection failure.
+    fn report_proposer_preferences_collection_failure(
+        &self,
+        error: &Error,
+        preferences: &ProposerPreferences,
+        signing_root: Hash256,
+    ) {
+        match instrumentation::classify_collection_failure(error) {
+            CollectionFailureClass::NoSignature => {
+                warn!(
+                    validator_index = preferences.validator_index,
+                    proposal_slot = %preferences.proposal_slot,
+                    target_gas_limit = preferences.target_gas_limit,
+                    dependent_root = ?preferences.dependent_root,
+                    ?signing_root,
+                    ?error,
+                    "Insufficient partial signatures to reconstruct ProposerPreferences; possible \
+                     causes: too few operators reached the threshold, partial-signature delivery \
+                     loss, or operators diverged on the signing root (target_gas_limit or \
+                     dependent_root)"
+                );
+                metrics::inc_counter_vec(
+                    &metrics::PROPOSER_PREFERENCES_RECONSTRUCTION_FAILURES,
+                    &[metrics::PROPOSER_PREFERENCES_FAILURE_INSUFFICIENT_PARTIAL_SIGNATURES],
+                );
+            }
+            CollectionFailureClass::Infra => {
+                error!(
+                    validator_index = preferences.validator_index,
+                    proposal_slot = %preferences.proposal_slot,
+                    ?error,
+                    "ProposerPreferences signature collection infrastructure failure"
+                );
+                metrics::inc_counter_vec(
+                    &metrics::PROPOSER_PREFERENCES_RECONSTRUCTION_FAILURES,
+                    &[metrics::PROPOSER_PREFERENCES_FAILURE_INFRA],
+                );
+            }
+            CollectionFailureClass::NonCollection => {
+                error!(
+                    validator_index = preferences.validator_index,
+                    proposal_slot = %preferences.proposal_slot,
+                    ?error,
+                    "Failed to sign ProposerPreferences"
+                );
+            }
+        }
+    }
+
     fn create_proposer_consensus_data_validator(
         &self,
         validator_pubkey: PublicKeyBytes,
@@ -3488,11 +3541,69 @@ impl<T: SlotClock, E: EthSpec, C: ConsensusDecider<E> + 'static> ValidatorStore
 
     async fn sign_proposer_preferences(
         &self,
-        _validator_pubkey: PublicKeyBytes,
-        _preferences: ProposerPreferences,
+        validator_pubkey: PublicKeyBytes,
+        preferences: ProposerPreferences,
     ) -> Result<SignedProposerPreferences, Error> {
-        // TODO(gloas)
-        Err(Error::SpecificError(SpecificError::Unsupported))
+        let (validator, cluster) = self.get_validator_and_cluster(validator_pubkey)?;
+
+        let epoch = preferences.proposal_slot.epoch(E::slots_per_epoch());
+        let domain = self.get_domain(epoch, Domain::ProposerPreferences);
+        let signing_root = preferences.signing_root(domain);
+
+        // Envelope slot = the duty's proposal_slot (SIP-94 §5/§7), NOT slot_clock.now() and NOT
+        // epoch-start. It becomes PartialSignatureMessages.slot on the wire, which peers accept via
+        // the ProposerPreferences earliness allowance and validate proposer-assignment against
+        // (#1062). A future proposal_slot also keeps the collector alive until that slot passes.
+        let proposal_slot = preferences.proposal_slot;
+
+        // Bound the wait so a no-quorum validator cannot head-of-line-block the LH service's
+        // sequential per-validator loop (see `PROPOSER_PREFERENCES_COLLECTION_TIMEOUT_SLOTS`).
+        let collection_timeout =
+            self.spec.get_slot_duration() * PROPOSER_PREFERENCES_COLLECTION_TIMEOUT_SLOTS;
+
+        // Map a deadline elapse to `CollectionTimeout` so it and any collector error share the
+        // single reporting/return path below.
+        let collected = match tokio::time::timeout(
+            collection_timeout,
+            self.collect_signature(
+                PartialSignatureKind::ProposerPreferences,
+                Role::ProposerPreferences,
+                CollectionMode::SingleValidator,
+                &validator,
+                &cluster,
+                signing_root,
+                proposal_slot,
+            ),
+        )
+        .await
+        {
+            Ok(result) => result,
+            Err(_elapsed) => Err(Error::SpecificError(
+                SpecificError::SignatureCollectionFailed(CollectionError::CollectionTimeout),
+            )),
+        };
+
+        let signature = match collected {
+            Ok(signature) => signature,
+            Err(err) => {
+                self.report_proposer_preferences_collection_failure(
+                    &err,
+                    &preferences,
+                    signing_root,
+                );
+                return Err(err);
+            }
+        };
+
+        validator_metrics::inc_counter_vec(
+            &metrics::SIGNED_PROPOSER_PREFERENCES_TOTAL,
+            &[validator_metrics::SUCCESS],
+        );
+
+        Ok(SignedProposerPreferences {
+            message: preferences,
+            signature,
+        })
     }
 }
 

anchor/validator_store/src/metrics.rs

@@ -27,6 +27,15 @@ pub static SIGNED_RANDAO_REVEALS_TOTAL: LazyLock<Result<IntCounterVec>> = LazyLo
     )
 });
 
+pub static SIGNED_PROPOSER_PREFERENCES_TOTAL: LazyLock<Result<IntCounterVec>> =
+    LazyLock::new(|| {
+        try_create_int_counter_vec(
+            "anchor_signed_proposer_preferences_total",
+            "Total count of ProposerPreferences signings",
+            &["status"],
+        )
+    });
+
 // ═══════════════════════════════════════════════════════════════════════════════
 // MetadataService metrics
 // ═══════════════════════════════════════════════════════════════════════════════
@@ -147,3 +156,22 @@ pub static PTC_RECONSTRUCTION_FAILURES: LazyLock<Result<IntCounterVec>> = LazyLo
         &["reason"],
     )
 });
+
+/// The committee never reached the partial-signature threshold for a ProposerPreferences signing.
+/// At the current collector granularity this single bucket covers every no-quorum cause — too few
+/// operators, partial-signature delivery loss, and operators diverging on the signing root
+/// (`target_gas_limit` / `dependent_root`) — which are indistinguishable at the wire. Mirrors
+/// `PTC_FAILURE_NO_SIGNATURE`.
+pub const PROPOSER_PREFERENCES_FAILURE_INSUFFICIENT_PARTIAL_SIGNATURES: &str =
+    "insufficient_partial_signatures";
+/// Local collection or reconstruction infrastructure fault.
+pub const PROPOSER_PREFERENCES_FAILURE_INFRA: &str = "infra";
+
+pub static PROPOSER_PREFERENCES_RECONSTRUCTION_FAILURES: LazyLock<Result<IntCounterVec>> =
+    LazyLock::new(|| {
+        try_create_int_counter_vec(
+            "anchor_proposer_preferences_reconstruction_failures_total",
+            "ProposerPreferences signature collection failures by reason",
+            &["reason"],
+        )
+    });

anchor/validator_store/src/testing/common.rs

@@ -125,10 +125,16 @@ pub(super) struct CapturedSignatureCall {
     pub(super) signing_root: Hash256,
 }
 
-/// Mock that captures calls and returns a canned infinity signature, or a configured failure.
+/// Mock that captures calls and returns a canned infinity signature, or a configured failure, or
+/// a future that never resolves (`hang`).
 struct MockSignatureCollector {
     captured: CapturedCalls,
     failure: Option<CollectionError>,
+    /// When `true`, `sign_and_collect` captures the call and then returns a future that never
+    /// resolves, modeling a quorum that never forms. This lets tests drive the production
+    /// `tokio::time::timeout` in `sign_proposer_preferences` to elapse; `hang` takes precedence
+    /// over `failure`.
+    hang: bool,
 }
 
 impl SignatureCollecting for MockSignatureCollector {
@@ -138,13 +144,19 @@ impl SignatureCollecting for MockSignatureCollector {
         requester: SignatureRequester,
         signing_data: ValidatorSigningData,
     ) -> Pin<Box<dyn Future<Output = Result<Arc<Signature>, CollectionError>> + Send + '_>> {
-        // Capture before failing so tests can assert that the collection attempt happened even
-        // when the configured outcome is an error.
+        // Capture before hanging/failing so tests can assert that the collection attempt happened
+        // even when the configured outcome is a stall or an error.
         self.captured.lock().push(CapturedSignatureCall {
             requester,
             metadata,
             signing_root: signing_data.root,
         });
+        if self.hang {
+            // Never resolves, so the caller only unblocks via the production collection timeout.
+            // `std::future::pending::<Result<Arc<Signature>, CollectionError>>()` is `Send`, which
+            // satisfies the returned future's bound.
+            return Box::pin(std::future::pending());
+        }
         if let Some(failure) = self.failure.clone() {
             return Box::pin(async move { Err(failure) });
         }
@@ -156,11 +168,13 @@ impl SignatureCollecting for MockSignatureCollector {
 /// Creates a mock signature collector and returns the shared captured calls handle.
 fn create_mock_collector(
     failure: Option<CollectionError>,
+    hang: bool,
 ) -> (Box<dyn SignatureCollecting>, CapturedCalls) {
     let captured: CapturedCalls = Arc::new(Mutex::new(Vec::new()));
     let mock = MockSignatureCollector {
         captured: Arc::clone(&captured),
         failure,
+        hang,
     };
     (Box::new(mock), captured)
 }
@@ -238,6 +252,10 @@ pub(super) fn create_committee_setup(
 pub(super) struct HarnessOptions {
     /// When set, every `sign_and_collect` call fails with this error after being captured.
     pub(super) collector_failure: Option<CollectionError>,
+    /// When `true`, every `sign_and_collect` call captures the call and then returns a future that
+    /// never resolves, modeling a quorum that never forms. Used to drive the production
+    /// collection-timeout path. Takes precedence over `collector_failure`.
+    pub(super) collector_hangs: bool,
     pub(super) disable_slashing_protection: bool,
     /// Chain spec to wire into the store. Defaults to `ChainSpec::mainnet()`, under which
     /// `TEST_SLOT` is pre-Electra. Tests that need a specific fork at `TEST_SLOT` (Electra or
@@ -252,6 +270,7 @@ impl Default for HarnessOptions {
     fn default() -> Self {
         Self {
             collector_failure: None,
+            collector_hangs: false,
             // Slashing protection is disabled by default because the harness never registers
             // validators in the slashing DB, which would fail block/attestation signing paths.
             disable_slashing_protection: true,
@@ -269,6 +288,18 @@ pub(super) fn gloas_at_genesis_spec() -> Arc<ChainSpec> {
     Arc::new(spec)
 }
 
+/// Builds a `ChainSpec` based on mainnet but with Gloas activated at `gloas_epoch` (leaving
+/// mainnet's other fork epochs untouched, so they stay far in the future). This places a fork
+/// boundary strictly inside a lookahead window: an epoch below `gloas_epoch` resolves to the
+/// genesis fork version while `gloas_epoch` and beyond resolve to the Gloas fork version, giving
+/// two byte-distinct signing domains on either side of the boundary. Used to make the
+/// "domain keyed on the *proposal* epoch, not the send epoch" assertion falsifiable.
+pub(super) fn gloas_at_epoch_spec(gloas_epoch: Epoch) -> Arc<ChainSpec> {
+    let mut spec = ChainSpec::mainnet();
+    spec.gloas_fork_epoch = Some(gloas_epoch);
+    Arc::new(spec)
+}
+
 /// Builds a `ChainSpec` based on mainnet but with Electra activated at genesis and Gloas
 /// disabled, so `TEST_SLOT` resolves to Electra (post-Electra, pre-Gloas). The committee QBFT
 /// decides over `BeaconVote` and the attestation index stays untouched at `0`.
@@ -325,7 +356,8 @@ impl ValidatorStoreTestHarness {
             "test",
         ));
 
-        let (mock_collector, captured_calls) = create_mock_collector(options.collector_failure);
+        let (mock_collector, captured_calls) =
+            create_mock_collector(options.collector_failure, options.collector_hangs);
 
         // Database
         let database = Arc::new(

anchor/validator_store/src/testing/mod.rs

@@ -4,3 +4,4 @@ mod committee_aggregate;
 mod committee_attestation;
 mod committee_attestation_gloas;
 mod payload_attestation;
+mod proposer_preferences;