Gloas fix proposer preferences for validators added mid-epoch (sigp#9660)

## Issue Addressed

Proposer preferences were treated as published for an entire epoch and dependent root. A validator added mid-epoch, for example through the keymanager API or by a downstream DVT client, can appear in refreshed proposer duties under the same dependent root but would not have its preferences published.

## Proposed Changes

Track successfully published proposer duties individually, allowing newly discovered validators and duties that failed to sign or publish to be retried without re-publishing successful duties.

Author: shane-moore

validator_client/validator_services/src/proposer_preferences_service.rs

@@ -1,15 +1,21 @@
 use crate::duties_service::DutiesService;
 use beacon_node_fallback::BeaconNodeFallback;
+use bls::PublicKeyBytes;
 use eth2::types::ProposerData;
 use slot_clock::SlotClock;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::ops::Deref;
 use std::sync::Arc;
 use task_executor::TaskExecutor;
 use tokio::time::sleep;
 use tracing::{debug, error, info, warn};
-use types::{ChainSpec, Epoch, EthSpec, ForkName, Hash256, ProposerPreferences};
-use validator_store::ValidatorStore;
+use types::{ChainSpec, Epoch, EthSpec, ForkName, Hash256, ProposerPreferences, Slot};
+use validator_store::{ProposalData, ValidatorStore};
+
+/// `(validator_index, proposal_slot)` duties already published, keyed by epoch and dependent
+/// root. Sets for stale roots persist until their epoch is pruned, so a dependent root that
+/// recurs within an epoch is not re-published.
+type PublishedPreferences = HashMap<(Epoch, Hash256), HashSet<(u64, Slot)>>;
 
 pub struct Inner<S, T> {
     duties_service: Arc<DutiesService<S, T>>,
@@ -68,7 +74,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> ProposerPreferencesSer
         let executor = self.executor.clone();
 
         let interval_fut = async move {
-            let mut published_preferences: HashMap<Epoch, Hash256> = HashMap::new();
+            let mut published_preferences = PublishedPreferences::new();
 
             loop {
                 let Some(current_slot) = self.slot_clock.now() else {
@@ -95,11 +101,15 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> ProposerPreferencesSer
     }
 
     /// Publish proposer preferences for `current_epoch` and `current_epoch + 1`.
-    /// Will only publish preferences for a given epoch once per dependent root.
+    ///
+    /// Each proposer duty is published once per epoch and dependent root, so a validator added
+    /// after an epoch's duties were first published is still picked up, and a dependent root
+    /// change re-publishes the epoch's duties. Duties that are missing proposal data or that
+    /// fail to sign or publish are retried on later polls without re-publishing their siblings.
     async fn poll_and_publish_preferences(
         &self,
         current_epoch: Epoch,
-        published_preferences: &mut HashMap<Epoch, Hash256>,
+        published_preferences: &mut PublishedPreferences,
     ) {
         for (epoch, fork_name) in [
             (
@@ -123,63 +133,36 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> ProposerPreferencesSer
                 }
             };
 
-            if published_preferences.get(&epoch) == Some(&dependent_root) {
+            let published = published_preferences
+                .entry((epoch, dependent_root))
+                .or_default();
+
+            let preferences_to_sign =
+                preferences_to_publish(dependent_root, &duties, published, |pubkey| {
+                    self.validator_store.proposal_data(pubkey)
+                });
+
+            if preferences_to_sign.is_empty() {
                 continue;
             }
 
-            if self
-                .publish_proposer_preferences(epoch, fork_name, dependent_root, duties)
-                .await
-            {
-                published_preferences.insert(epoch, dependent_root);
-            }
+            let newly_published = self
+                .publish_proposer_preferences(epoch, fork_name, preferences_to_sign)
+                .await;
+            published.extend(newly_published);
         }
 
-        published_preferences.retain(|epoch, _| *epoch >= current_epoch);
+        published_preferences.retain(|(epoch, _), _| *epoch >= current_epoch);
     }
 
+    /// Sign and publish `preferences_to_sign`, returning the `(validator_index, slot)` pairs
+    /// that were signed and included in a successfully published batch.
     async fn publish_proposer_preferences(
         &self,
         epoch: Epoch,
         fork_name: ForkName,
-        dependent_root: Hash256,
-        duties: Vec<ProposerData>,
-    ) -> bool {
-        let preferences_to_sign: Vec<_> = {
-            let mut result = vec![];
-            for duty in &duties {
-                let Some(proposal_data) = self.validator_store.proposal_data(&duty.pubkey) else {
-                    warn!(
-                        validator = ?duty.pubkey,
-                        "Missing proposal data for proposer preferences"
-                    );
-                    continue;
-                };
-                let Some(fee_recipient) = proposal_data.fee_recipient else {
-                    warn!(
-                        validator = ?duty.pubkey,
-                        "Missing fee recipient for proposer preferences"
-                    );
-                    continue;
-                };
-                result.push((
-                    duty.pubkey,
-                    ProposerPreferences {
-                        dependent_root,
-                        proposal_slot: duty.slot,
-                        validator_index: duty.validator_index,
-                        fee_recipient,
-                        target_gas_limit: proposal_data.gas_limit,
-                    },
-                ));
-            }
-            result
-        };
-
-        if preferences_to_sign.is_empty() {
-            return false;
-        }
-
+        preferences_to_sign: Vec<(PublicKeyBytes, ProposerPreferences)>,
+    ) -> Vec<(u64, Slot)> {
         debug!(
             %epoch,
             count = preferences_to_sign.len(),
@@ -205,7 +188,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> ProposerPreferencesSer
         }
 
         if signed.is_empty() {
-            return false;
+            return vec![];
         }
 
         let count = signed.len();
@@ -250,16 +233,231 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> ProposerPreferencesSer
                     %count,
                     "Successfully published proposer preferences"
                 );
-                true
+                signed
+                    .iter()
+                    .map(|preferences| {
+                        (
+                            preferences.message.validator_index,
+                            preferences.message.proposal_slot,
+                        )
+                    })
+                    .collect()
             }
             Err(e) => {
                 error!(
                     error = %e,
                     %epoch,
                     "Failed to publish proposer preferences"
                 );
-                false
+                vec![]
             }
         }
     }
 }
+
+/// Build the proposer preferences that still need publishing: one per proposer duty whose
+/// `(validator_index, slot)` is not in `published` and whose proposal data is available.
+fn preferences_to_publish(
+    dependent_root: Hash256,
+    duties: &[ProposerData],
+    published: &HashSet<(u64, Slot)>,
+    proposal_data: impl Fn(&PublicKeyBytes) -> Option<ProposalData>,
+) -> Vec<(PublicKeyBytes, ProposerPreferences)> {
+    duties
+        .iter()
+        .filter(|duty| !published.contains(&(duty.validator_index, duty.slot)))
+        .filter_map(|duty| {
+            let Some(proposal_data) = proposal_data(&duty.pubkey) else {
+                warn!(
+                    validator = ?duty.pubkey,
+                    "Missing proposal data for proposer preferences"
+                );
+                return None;
+            };
+            let Some(fee_recipient) = proposal_data.fee_recipient else {
+                warn!(
+                    validator = ?duty.pubkey,
+                    "Missing fee recipient for proposer preferences"
+                );
+                return None;
+            };
+            Some((
+                duty.pubkey,
+                ProposerPreferences {
+                    dependent_root,
+                    proposal_slot: duty.slot,
+                    validator_index: duty.validator_index,
+                    fee_recipient,
+                    target_gas_limit: proposal_data.gas_limit,
+                },
+            ))
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use types::Address;
+
+    const FEE_RECIPIENT: Address = Address::repeat_byte(42);
+    const GAS_LIMIT: u64 = 30_000_000;
+
+    fn pubkey(i: u8) -> PublicKeyBytes {
+        PublicKeyBytes::deserialize(&[i; 48]).unwrap()
+    }
+
+    fn duty(i: u8, slot: u64) -> ProposerData {
+        ProposerData {
+            pubkey: pubkey(i),
+            validator_index: i as u64,
+            slot: Slot::new(slot),
+        }
+    }
+
+    fn proposal_data(fee_recipient: Option<Address>) -> ProposalData {
+        ProposalData {
+            validator_index: None,
+            fee_recipient,
+            gas_limit: GAS_LIMIT,
+            builder_proposals: false,
+        }
+    }
+
+    fn default_proposal_data(_: &PublicKeyBytes) -> Option<ProposalData> {
+        Some(proposal_data(Some(FEE_RECIPIENT)))
+    }
+
+    /// Run the state transitions of one poll (select unpublished duties, then record them as
+    /// published on success) and return how many duties were published. Mirrors the map
+    /// bookkeeping in `poll_and_publish_preferences` around the signing effect.
+    fn publish_round(
+        published_preferences: &mut PublishedPreferences,
+        epoch: Epoch,
+        dependent_root: Hash256,
+        duties: &[ProposerData],
+    ) -> usize {
+        let published = published_preferences
+            .entry((epoch, dependent_root))
+            .or_default();
+        let to_sign =
+            preferences_to_publish(dependent_root, duties, published, default_proposal_data);
+        let count = to_sign.len();
+        published.extend(
+            to_sign
+                .iter()
+                .map(|(_, preferences)| (preferences.validator_index, preferences.proposal_slot)),
+        );
+        count
+    }
+
+    #[test]
+    fn new_validator_under_same_root_yields_only_new_validator() {
+        let dependent_root = Hash256::repeat_byte(1);
+        let duties = [duty(1, 1), duty(2, 2), duty(3, 3)];
+        let published = HashSet::from([(1, Slot::new(1)), (2, Slot::new(2))]);
+
+        let result =
+            preferences_to_publish(dependent_root, &duties, &published, default_proposal_data);
+
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].0, pubkey(3));
+        assert_eq!(result[0].1.proposal_slot, Slot::new(3));
+    }
+
+    #[test]
+    fn built_preferences_carry_duty_fields() {
+        let dependent_root = Hash256::repeat_byte(7);
+        let duties = [duty(1, 5)];
+
+        let result = preferences_to_publish(
+            dependent_root,
+            &duties,
+            &HashSet::new(),
+            default_proposal_data,
+        );
+
+        assert_eq!(result.len(), 1);
+        let (pubkey_out, preferences) = &result[0];
+        assert_eq!(*pubkey_out, pubkey(1));
+        assert_eq!(preferences.dependent_root, dependent_root);
+        assert_eq!(preferences.proposal_slot, Slot::new(5));
+        assert_eq!(preferences.validator_index, 1);
+        assert_eq!(preferences.fee_recipient, FEE_RECIPIENT);
+        assert_eq!(preferences.target_gas_limit, GAS_LIMIT);
+    }
+
+    #[test]
+    fn missing_data_excluded_and_retried() {
+        let dependent_root = Hash256::repeat_byte(1);
+        // Validator 1 has no fee recipient yet, validator 2 is ready, validator 3 has no
+        // proposal data at all.
+        let duties = [duty(1, 1), duty(2, 2), duty(3, 3)];
+        let mut published = HashSet::new();
+
+        let result =
+            preferences_to_publish(
+                dependent_root,
+                &duties,
+                &published,
+                |pubkey_in| match pubkey_in {
+                    pk if *pk == pubkey(1) => Some(proposal_data(None)),
+                    pk if *pk == pubkey(3) => None,
+                    _ => Some(proposal_data(Some(FEE_RECIPIENT))),
+                },
+            );
+
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].0, pubkey(2));
+
+        // A later poll, after validator 1's fee recipient becomes available and validator 2's
+        // preferences were published. Validator 3 still has no proposal data.
+        published.insert((2, Slot::new(2)));
+        let result = preferences_to_publish(dependent_root, &duties, &published, |pubkey_in| {
+            (*pubkey_in != pubkey(3)).then(|| proposal_data(Some(FEE_RECIPIENT)))
+        });
+
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].0, pubkey(1));
+    }
+
+    #[test]
+    fn multi_slot_proposer_yields_one_entry_per_duty() {
+        let dependent_root = Hash256::repeat_byte(1);
+        let duties = [duty(1, 10), duty(1, 11)];
+        let published = HashSet::from([(1, Slot::new(10))]);
+
+        let result =
+            preferences_to_publish(dependent_root, &duties, &published, default_proposal_data);
+
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].0, pubkey(1));
+        assert_eq!(result[0].1.proposal_slot, Slot::new(11));
+    }
+
+    #[test]
+    fn dependent_root_flip_flop_does_not_resign() {
+        let epoch = Epoch::new(0);
+        let root_a = Hash256::repeat_byte(1);
+        let root_b = Hash256::repeat_byte(2);
+        let duties = [duty(1, 1), duty(2, 2)];
+        let mut published_preferences = PublishedPreferences::new();
+
+        // Publish under the original root, then re-publish under the reorged root.
+        assert_eq!(
+            publish_round(&mut published_preferences, epoch, root_a, &duties),
+            2
+        );
+        assert_eq!(
+            publish_round(&mut published_preferences, epoch, root_b, &duties),
+            2
+        );
+
+        // The head reorgs back to the original root. Its published set was retained under its own
+        // key, so nothing is signed again.
+        assert_eq!(
+            publish_round(&mut published_preferences, epoch, root_a, &duties),
+            0
+        );
+    }
+}