fix(validator): admit local proposer preferences

Honor exact assignments in the local duty producer when the complete
tracker rejects a preference-role message, allowing already-scheduled
signing work to collect peer shares.

Keep envelope validation unchanged and cover the exception through a
real receiver and signature-collector regression.

Related to #1295.

Author: shane-moore

Cargo.lock

@@ -11,6 +11,7 @@ path = "src/lib.rs"
 [dependencies]
 anchor_validator_store = { workspace = true }
 beacon_node_fallback = { workspace = true }
+bls = { workspace = true }
 clap = { workspace = true }
 cli = { workspace = true }
 database = { workspace = true }

anchor/client/Cargo.toml

@@ -4,6 +4,7 @@ mod metrics;
 mod notifier;
 
 use std::{
+    collections::HashMap,
     fs::File,
     io::Read,
     net::SocketAddr,
@@ -21,13 +22,14 @@ use beacon_node_fallback::{
     BeaconNodeFallback, CandidateBeaconNode, beacon_head_monitor::HeadEvent,
     start_fallback_updater_service,
 };
+use bls::PublicKeyBytes;
 use config::Config;
 use database::{NetworkDatabase, OwnOperatorId};
 use duties_tracker::{duties_tracker::DutiesTracker, voluntary_exit_tracker::VoluntaryExitTracker};
 use eth::{
     index_sync::start_validator_index_syncer, voluntary_exit_processor::start_exit_processor,
 };
-use eth2::{BeaconNodeHttpClient, Timeouts};
+use eth2::{BeaconNodeHttpClient, Timeouts, types::ProposerData};
 use message_receiver::NetworkMessageReceiver;
 use message_sender::{MessageSender, NetworkMessageSender, impostor::ImpostorMessageSender};
 use message_validator::Validator;
@@ -53,7 +55,7 @@ use tokio::{
     time::{Instant, interval, sleep},
 };
 use tracing::{debug, error, info, warn};
-use types::{EthSpec, Hash256};
+use types::{Epoch, EthSpec, Hash256, Slot};
 use validator_metrics::set_gauge;
 use validator_services::{
     attestation_service::AttestationServiceBuilder,
@@ -91,6 +93,21 @@ const MAX_HEAD_EVENT_QUEUE_LEN: usize = 1_024;
 
 pub struct Client {}
 
+fn local_proposer_assignment_at_slot(
+    proposers: &HashMap<Epoch, (Hash256, Vec<ProposerData>)>,
+    slots_per_epoch: u64,
+    slot: Slot,
+    validator_pubkey: &PublicKeyBytes,
+) -> bool {
+    proposers
+        .get(&slot.epoch(slots_per_epoch))
+        .is_some_and(|(_, duties)| {
+            duties
+                .iter()
+                .any(|duty| duty.slot == slot && duty.pubkey == *validator_pubkey)
+        })
+}
+
 impl Client {
     /// Runs the Anchor Client
     pub async fn run<E: EthSpec>(executor: TaskExecutor, config: Config) -> Result<(), String> {
@@ -693,6 +710,20 @@ impl Client {
                 .build()?,
         );
 
+        // Support preferences for duties the local producer sees even when the tracker disagrees.
+        // Neither cache is guaranteed to be newer, so this supplies positive evidence only.
+        let local_duties_service = duties_service.clone();
+        duties_tracker
+            .set_local_proposer_lookup(move |slot, pubkey| {
+                local_proposer_assignment_at_slot(
+                    &local_duties_service.proposers.read(),
+                    E::slots_per_epoch(),
+                    slot,
+                    pubkey,
+                )
+            })
+            .map_err(|e| e.to_string())?;
+
         // Update the metrics server.
         if let Some(ctx) = &http_metrics_shared_state {
             ctx.write().genesis_time = Some(genesis_time);
@@ -995,3 +1026,84 @@ pub fn load_pem_certificate<P: AsRef<Path>>(pem_path: P) -> Result<Certificate,
         .map_err(|e| format!("Unable to read certificate file: {e}"))?;
     Certificate::from_pem(&buf).map_err(|e| format!("Unable to parse certificate: {e}"))
 }
+
+#[cfg(test)]
+mod local_proposer_tests {
+    use super::*;
+
+    const SLOTS_PER_EPOCH: u64 = 32;
+    const DUTY_SLOT: Slot = Slot::new(SLOTS_PER_EPOCH);
+
+    #[test]
+    fn test_local_proposer_assignment_requires_exact_live_epoch_slot_and_pubkey() {
+        // Arrange: create the lookup before any local duty rows exist, as at client startup.
+        let proposers = Arc::new(RwLock::new(HashMap::new()));
+        let local_proposers = proposers.clone();
+        let lookup = move |slot, pubkey: &PublicKeyBytes| {
+            local_proposer_assignment_at_slot(
+                &local_proposers.read(),
+                SLOTS_PER_EPOCH,
+                slot,
+                pubkey,
+            )
+        };
+        let validator_x = bls::Keypair::random().pk.compress();
+        let validator_y = bls::Keypair::random().pk.compress();
+        let epoch = DUTY_SLOT.epoch(SLOTS_PER_EPOCH);
+        assert!(!lookup(DUTY_SLOT, &validator_x));
+
+        // Act: a matching row under the wrong epoch key must not count as local evidence.
+        proposers.write().insert(
+            epoch + 1,
+            (
+                Hash256::ZERO,
+                vec![ProposerData {
+                    pubkey: validator_x,
+                    validator_index: 0,
+                    slot: DUTY_SLOT,
+                }],
+            ),
+        );
+
+        // Assert: the query must select the duty slot's epoch before matching rows.
+        assert!(!lookup(DUTY_SLOT, &validator_x));
+
+        // Act: in the correct epoch, X has a different slot and Y has the requested slot.
+        proposers.write().insert(
+            epoch,
+            (
+                Hash256::ZERO,
+                vec![
+                    ProposerData {
+                        pubkey: validator_x,
+                        validator_index: 0,
+                        slot: DUTY_SLOT + 1,
+                    },
+                    ProposerData {
+                        pubkey: validator_y,
+                        validator_index: 1,
+                        slot: DUTY_SLOT,
+                    },
+                ],
+            ),
+        );
+
+        // Assert: neither row matches both the requested slot and validator pubkey.
+        assert!(!lookup(DUTY_SLOT, &validator_x));
+
+        // Act: a later local cache update supplies the exact row after lookup construction.
+        proposers
+            .write()
+            .get_mut(&epoch)
+            .unwrap()
+            .1
+            .push(ProposerData {
+                pubkey: validator_x,
+                validator_index: 0,
+                slot: DUTY_SLOT,
+            });
+
+        // Assert: the existing lookup reads the current cache, not a startup snapshot.
+        assert!(lookup(DUTY_SLOT, &validator_x));
+    }
+}

anchor/client/src/lib.rs

@@ -1,4 +1,7 @@
-use std::{future::Future, sync::Arc};
+use std::{
+    future::Future,
+    sync::{Arc, OnceLock},
+};
 
 use beacon_node_fallback::BeaconNodeFallback;
 use bls::PublicKeyBytes;
@@ -21,6 +24,8 @@ use crate::{
 /// Only retain `HISTORICAL_DUTIES_EPOCHS` duties prior to the current epoch.
 const HISTORICAL_DUTIES_EPOCHS: u64 = 2;
 
+type LocalProposerLookup = dyn Fn(Slot, &PublicKeyBytes) -> bool + Send + Sync;
+
 #[derive(Error, Debug)]
 pub enum Error {
     #[error("Unable to read the slot clock")]
@@ -33,6 +38,8 @@ pub enum Error {
     FailedToPollPtc(String),
     #[error("Invalid PTC duties: {0}")]
     InvalidPtcDuties(#[from] PtcScheduleError),
+    #[error("Local proposer lookup has already been configured")]
+    LocalProposerLookupAlreadySet,
 }
 
 pub struct DutiesTracker<T: SlotClock + 'static> {
@@ -50,6 +57,8 @@ pub struct DutiesTracker<T: SlotClock + 'static> {
     slot_clock: T,
     /// The network state receiver.
     network_state_rx: watch::Receiver<NetworkState>,
+    /// The local duty producer is initialized after this tracker and the network receiver.
+    local_proposer_lookup: OnceLock<Box<LocalProposerLookup>>,
 }
 
 impl<T: SlotClock + 'static> DutiesTracker<T> {
@@ -69,9 +78,21 @@ impl<T: SlotClock + 'static> DutiesTracker<T> {
             slots_per_epoch,
             slot_clock,
             network_state_rx,
+            local_proposer_lookup: OnceLock::new(),
         }
     }
 
+    /// Installs the local producer's live assignment lookup once during client initialization.
+    /// The lookup runs synchronously during message validation and must only read cached duties.
+    pub fn set_local_proposer_lookup(
+        &self,
+        lookup: impl Fn(Slot, &PublicKeyBytes) -> bool + Send + Sync + 'static,
+    ) -> Result<(), Error> {
+        self.local_proposer_lookup
+            .set(Box::new(lookup))
+            .map_err(|_| Error::LocalProposerLookupAlreadySet)
+    }
+
     async fn poll_sync_committee_duties(&self) -> Result<(), Error> {
         let sync_duties = &self.duties.sync_duties;
         let spec = &self.spec;
@@ -445,6 +466,16 @@ impl<T: SlotClock + 'static> DutiesProvider for DutiesTracker<T> {
         }
     }
 
+    fn local_proposer_assignment_at_slot(
+        &self,
+        slot: Slot,
+        validator_pubkey: &PublicKeyBytes,
+    ) -> bool {
+        self.local_proposer_lookup
+            .get()
+            .is_some_and(|lookup| lookup(slot, validator_pubkey))
+    }
+
     fn ptc_assignment_at_slot(
         &self,
         slot: Slot,
@@ -529,6 +560,31 @@ mod tests {
         )
     }
 
+    #[test]
+    fn test_local_proposer_lookup_is_optional_and_installed_only_once() {
+        // Arrange: the tracker starts without a local producer lookup.
+        let tracker = tracker_with_empty_network_state();
+        let pubkey = random_validator_pubkey();
+        let slot = Slot::new(SLOTS_PER_EPOCH);
+        assert!(!tracker.local_proposer_assignment_at_slot(slot, &pubkey));
+
+        // Act: install exact positive evidence, then attempt to replace the configured source.
+        tracker
+            .set_local_proposer_lookup(move |candidate_slot, candidate_pubkey| {
+                candidate_slot == slot && *candidate_pubkey == pubkey
+            })
+            .unwrap();
+        let replacement = tracker.set_local_proposer_lookup(|_, _| false);
+
+        // Assert: duplicate initialization is visible and preserves the original live lookup.
+        assert!(matches!(
+            replacement,
+            Err(Error::LocalProposerLookupAlreadySet)
+        ));
+        assert!(tracker.local_proposer_assignment_at_slot(slot, &pubkey));
+        assert!(!tracker.local_proposer_assignment_at_slot(slot + 1, &pubkey));
+    }
+
     /// A `BeaconNodeFallback` with a single candidate that is never actually contacted by these
     /// tests. It exists only to satisfy `DutiesTracker::new`.
     fn dummy_beacon_node_fallback(spec: Arc<ChainSpec>) -> BeaconNodeFallback<ManualSlotClock> {

anchor/duties_tracker/src/duties_tracker.rs

@@ -285,6 +285,16 @@ pub trait DutiesProvider: Sync + Send + 'static {
         validator_pubkey: &PublicKeyBytes,
     ) -> DutyAssignment;
 
+    /// Additional positive evidence from the local duty producer. This does not order the two
+    /// proposer views or change the retained complete schedule's assignments.
+    fn local_proposer_assignment_at_slot(
+        &self,
+        _slot: Slot,
+        _validator_pubkey: &PublicKeyBytes,
+    ) -> bool {
+        false
+    }
+
     /// Unknown unless a completed PTC fetch covered this validator in the slot's epoch.
     fn ptc_assignment_at_slot(&self, slot: Slot, validator_index: ValidatorIndex)
     -> DutyAssignment;

anchor/duties_tracker/src/lib.rs

@@ -20,3 +20,20 @@ thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
 types = { workspace = true }
+
+[dev-dependencies]
+axum = { workspace = true }
+beacon_node_fallback = { workspace = true }
+bls = { workspace = true }
+bls_lagrange = { workspace = true }
+database = { workspace = true, features = ["test-utils"] }
+duties_tracker = { workspace = true }
+eth2 = { workspace = true }
+ethereum_ssz = { workspace = true }
+fork = { workspace = true }
+message_sender = { workspace = true, features = ["testing"] }
+openssl = { workspace = true }
+sensitive_url = { workspace = true }
+subnet_service = { workspace = true }
+task_executor = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }

anchor/message_receiver/Cargo.toml

@@ -24,3 +24,6 @@ pub enum Error {
     #[error("Processor error: {0}")]
     Processor(#[from] processor::Error),
 }
+
+#[cfg(test)]
+mod proposer_view_tests;