diff --git a/anchor/duties_tracker/src/duties_tracker.rs b/anchor/duties_tracker/src/duties_tracker.rs index 772303c0f..79179109e 100644 --- a/anchor/duties_tracker/src/duties_tracker.rs +++ b/anchor/duties_tracker/src/duties_tracker.rs @@ -13,7 +13,7 @@ use tracing::{debug, error, trace, warn}; use types::{ChainSpec, Epoch, Slot}; use crate::{ - Duties, DutiesProvider, DutyAssignment, MembershipKey, + Duties, DutiesProvider, DutyAssignment, MembershipKey, ProposerSchedule, voluntary_exit_tracker::VoluntaryExitTracker, }; @@ -205,7 +205,8 @@ impl DutiesTracker { Ok(()) } - /// Download the proposer duties for the current epoch. + /// Download the proposer duties for the current and next epoch, retaining only + /// responses that validate as complete schedules. async fn poll_beacon_proposers(&self) -> Result<(), Error> { let current_slot = self.slot_clock.now().ok_or(Error::UnableToReadSlotClock)?; let current_epoch = current_slot.epoch(self.slots_per_epoch); @@ -215,18 +216,26 @@ impl DutiesTracker { match self .beacon_nodes .first_success(|beacon_node| async move { - beacon_node.get_validator_duties_proposer(epoch).await + beacon_node.get_validator_duties_proposer_v2(epoch).await }) .await - .map(|response| response.data) { - Ok(proposer_duties) => { + Ok(response) => { trace!( - num_proposer_duties = proposer_duties.len(), + num_proposer_duties = response.data.len(), "Downloaded proposer duties" ); - self.duties.proposers.write().insert(epoch, proposer_duties); + match ProposerSchedule::from_response(epoch, self.slots_per_epoch, response) { + Ok(schedule) => { + self.duties.proposers.write().insert(epoch, schedule); + } + Err(e) => warn!( + %epoch, + error = %e, + "Discarding malformed proposer duties; retaining prior view" + ), + } } Err(e) => last_err = Some(Error::FailedToPollProposers(e.to_string())), }; @@ -328,8 +337,8 @@ impl DutiesProvider for DutiesTracker { .proposers .read() .get(&epoch) - .map(|proposers| { - proposers.iter().any(|proposer_data| { + .map(|schedule| { + schedule.duties().iter().any(|proposer_data| { proposer_data.slot == slot && proposer_data.validator_index == validator_index }) }) @@ -347,8 +356,9 @@ impl DutiesProvider for DutiesTracker { ) -> DutyAssignment { let epoch = slot.epoch(self.slots_per_epoch); match self.duties.proposers.read().get(&epoch) { - Some(proposers) => { - if proposers + Some(schedule) => { + if schedule + .duties() .iter() .any(|d| d.slot == slot && d.pubkey == *validator_pubkey) { @@ -372,15 +382,19 @@ mod tests { use std::time::Duration; use beacon_node_fallback::{ApiTopic, BeaconNodeFallback, CandidateBeaconNode, Config}; - use bls::{Keypair, PublicKeyBytes}; + use bls::{FixedBytesExtended, Keypair, PublicKeyBytes}; use database::NetworkDatabase; - use eth2::{BeaconNodeHttpClient, Timeouts, types::ProposerData}; + use eth2::{ + BeaconNodeHttpClient, Timeouts, + types::{DutiesResponse, ProposerData}, + }; use openssl::rsa::Rsa; use sensitive_url::SensitiveUrl; use slot_clock::{ManualSlotClock, SlotClock}; - use types::{ChainSpec, Epoch, Slot}; + use types::{ChainSpec, Epoch, Hash256, Slot}; use super::*; + use crate::ProposerScheduleError::{DuplicateSlot, SlotOutOfEpoch, WrongLength}; /// Slots per epoch used by these tests. Kept small and independent of any real fork schedule. const SLOTS_PER_EPOCH: u64 = 32; @@ -451,10 +465,17 @@ mod tests { PublicKeyBytes::from(Keypair::random().pk) } - /// Seeds `tracker.duties.proposers[epoch]` with `data` exactly as `poll_beacon_proposers` - /// does (a single unfiltered `insert` of the whole `response.data`). + /// Seeds `tracker.duties.proposers[epoch]` with `data` by wrapping the raw duties in a + /// `ProposerSchedule` via the test-only `from_duties_unchecked` ctor. This deliberately + /// bypasses the completeness/validity checks that `from_response` enforces, so the + /// `proposer_assignment_at_slot` tests below can seed intentionally partial schedules that a + /// validated response could never produce. fn seed_epoch(tracker: &DutiesTracker, epoch: Epoch, data: Vec) { - tracker.duties.proposers.write().insert(epoch, data); + tracker + .duties + .proposers + .write() + .insert(epoch, ProposerSchedule::from_duties_unchecked(data)); } /// Builds a `ProposerData` for the given slot/pubkey with an arbitrary validator index. @@ -466,6 +487,40 @@ mod tests { } } + /// Epoch used by the `from_response` matrix. Deliberately NON-ZERO: at epoch 0 the epoch's + /// start slot is 0, so a regression that forgot to rebase slots onto the epoch + /// (`slot.checked_sub(start_slot)`) would still pass. Epoch 3 forces `start_slot == 96`, + /// so any such bug turns these tests RED. + const RESPONSE_EPOCH: u64 = 3; + + /// Distinctive, clearly non-zero `dependent_root` seed for the metadata-retention test. It + /// differs from `from_duties_unchecked`'s `Hash256::zero()` default, so a dropped-metadata + /// regression in `from_response` is observable rather than masked by a matching default. + const DISTINCTIVE_ROOT_SEED: u64 = 0xDEAD_BEEF; + + /// Builds a COMPLETE schedule for `epoch`: exactly one `proposer` entry for every slot in + /// `[start_slot, start_slot + SLOTS_PER_EPOCH)`, each with a fresh random pubkey. Entries are + /// emitted in REVERSED slot order so acceptance tests prove `from_response` is + /// order-independent (it must not assume the input is pre-sorted). + fn complete_epoch_duties(epoch: Epoch) -> Vec { + let start = epoch.start_slot(SLOTS_PER_EPOCH).as_u64(); + (0..SLOTS_PER_EPOCH) + .rev() + .map(|offset| proposer(Slot::new(start + offset), random_validator_pubkey())) + .collect() + } + + /// Wraps `data` in a `DutiesResponse` carrying placeholder metadata (`Hash256::zero()` / + /// `None`). Used only by the validation-matrix cases whose result (a `WrongLength`, + /// `SlotOutOfEpoch`, or `DuplicateSlot` error) is decided before any metadata is read. + fn duties_response(data: Vec) -> DutiesResponse> { + DutiesResponse { + dependent_root: Hash256::zero(), + execution_optimistic: None, + data, + } + } + // ==================== proposer_assignment_at_slot ==================== #[test] @@ -546,4 +601,189 @@ mod tests { "a slot in an unfetched epoch must return Unknown" ); } + + // ==================== ProposerSchedule::from_response ==================== + + #[test] + fn test_from_response_accepts_complete_unsorted_schedule() { + // regression caught: a completeness/uniqueness check that assumes the input is pre-sorted + // (e.g. compares the i-th duty's slot to `start_slot + i`) would reject this valid + // schedule. Arrange: a full 32-entry epoch-3 schedule supplied in REVERSED slot + // order. + let epoch = Epoch::new(RESPONSE_EPOCH); + let response = duties_response(complete_epoch_duties(epoch)); + + // Act + let schedule = ProposerSchedule::from_response(epoch, SLOTS_PER_EPOCH, response) + .expect("a complete, in-epoch, duplicate-free schedule must be accepted"); + + // Assert: exactly all SLOTS_PER_EPOCH duties are retained. + assert_eq!( + schedule.duties().len(), + SLOTS_PER_EPOCH as usize, + "an accepted schedule must retain exactly SLOTS_PER_EPOCH duties" + ); + } + + #[test] + fn test_from_response_retains_dependent_root_and_execution_optimistic() { + // regression caught: `from_response` dropping or defaulting the response metadata (e.g. + // constructing `Self` with `Hash256::zero()` / `None` instead of copying + // `response.dependent_root` / `response.execution_optimistic`). The distinctive non-zero + // root and `Some(true)` below differ from those defaults, so a dropped-metadata bug is + // visible rather than masked. This MUST go through `from_response` (not the unchecked + // ctor, which would make the assertion vacuous). + // Arrange: a complete schedule carried by a response with DISTINCTIVE non-default metadata. + let epoch = Epoch::new(RESPONSE_EPOCH); + let dependent_root = Hash256::from_low_u64_be(DISTINCTIVE_ROOT_SEED); + let response = DutiesResponse { + dependent_root, + execution_optimistic: Some(true), + data: complete_epoch_duties(epoch), + }; + + // Act + let schedule = ProposerSchedule::from_response(epoch, SLOTS_PER_EPOCH, response) + .expect("a complete schedule must be accepted"); + + // Assert: both metadata fields survive verbatim. + assert_eq!( + schedule.dependent_root(), + dependent_root, + "dependent_root must be copied through from_response verbatim" + ); + assert_eq!( + schedule.execution_optimistic(), + Some(true), + "execution_optimistic must be copied through from_response verbatim" + ); + } + + #[test] + fn test_from_response_rejects_too_short_schedule() { + // regression caught: a length check using `<` / `>=` instead of `!=`, or dropped entirely, + // would let a short (incomplete) schedule through as if it were complete. + // Arrange: 31 entries (drop the last slot of the complete set). + let epoch = Epoch::new(RESPONSE_EPOCH); + let mut data = complete_epoch_duties(epoch); + data.pop(); + + // Act + let result = ProposerSchedule::from_response(epoch, SLOTS_PER_EPOCH, duties_response(data)); + + // Assert: exact variant AND payload. `unwrap_err` sidesteps the `Ok` type lacking + // `PartialEq` while still pinning the error precisely. + assert_eq!( + result.unwrap_err(), + WrongLength { + expected: 32, + actual: 31 + }, + "31 duties must be rejected as WrongLength {{ expected: 32, actual: 31 }}" + ); + } + + #[test] + fn test_from_response_rejects_too_long_schedule() { + // regression caught: a length check missing its upper bound (only `len < expected`) would + // accept an over-long schedule. WrongLength must also preempt the duplicate scan here. + // Arrange: 33 entries — the complete 32 plus one extra (duplicated) in-range slot. + let epoch = Epoch::new(RESPONSE_EPOCH); + let mut data = complete_epoch_duties(epoch); + data.push(proposer( + epoch.start_slot(SLOTS_PER_EPOCH), + random_validator_pubkey(), + )); + + // Act + let result = ProposerSchedule::from_response(epoch, SLOTS_PER_EPOCH, duties_response(data)); + + // Assert: exact variant AND payload (length checked before any per-slot scan). + assert_eq!( + result.unwrap_err(), + WrongLength { + expected: 32, + actual: 33 + }, + "33 duties must be rejected as WrongLength {{ expected: 32, actual: 33 }}" + ); + } + + #[test] + fn test_from_response_rejects_empty_schedule() { + // regression caught: treating an empty response as a valid (vacuously complete) schedule. + // Arrange: zero entries. + let epoch = Epoch::new(RESPONSE_EPOCH); + + // Act + let result = + ProposerSchedule::from_response(epoch, SLOTS_PER_EPOCH, duties_response(Vec::new())); + + // Assert: exact variant AND payload. + assert_eq!( + result.unwrap_err(), + WrongLength { + expected: 32, + actual: 0 + }, + "an empty schedule must be rejected as WrongLength {{ expected: 32, actual: 0 }}" + ); + } + + #[test] + fn test_from_response_rejects_slot_from_another_epoch() { + // regression caught: a missing/incorrect epoch-bounds check — e.g. no upper `offset < + // slots_per_epoch` guard, or a missing `checked_sub(start_slot)` rebase — that would admit + // a duty slot belonging to a neighbouring epoch. The non-zero RESPONSE_EPOCH is essential: + // at epoch 0 a missing rebase would be invisible. + // Arrange: 32 entries — 31 distinct in-range slots plus exactly ONE slot from epoch 4. + // That foreign slot is the ONLY defect, so `SlotOutOfEpoch` is the only reachable error + // regardless of iteration order (the remaining 31 slots are distinct and in-range). + let epoch = Epoch::new(RESPONSE_EPOCH); + let foreign_slot = Epoch::new(RESPONSE_EPOCH + 1).start_slot(SLOTS_PER_EPOCH); + let mut data = complete_epoch_duties(epoch); + data[0].slot = foreign_slot; + + // Act + let result = ProposerSchedule::from_response(epoch, SLOTS_PER_EPOCH, duties_response(data)); + + // Assert: exact out-of-epoch slot AND epoch. + assert_eq!( + result.unwrap_err(), + SlotOutOfEpoch(foreign_slot, epoch), + "a duty slot from epoch 4 must be rejected as SlotOutOfEpoch(foreign_slot, epoch 3)" + ); + } + + #[test] + fn test_from_response_rejects_duplicate_slot_which_also_covers_missing_slot() { + // regression caught: dropping the per-slot uniqueness (`seen[offset]`) check, which would + // silently accept a schedule that duplicates one slot and (by pigeonhole) omits another. + // There is no `MissingSlot` variant, and none is needed: with `len == 32`, all slots + // in-range, and no duplicate, the schedule is necessarily complete — so a missing slot can + // ONLY surface as a `DuplicateSlot`. This case therefore doubles as the missing-slot test. + // Arrange: 32 entries, all in-range; overwrite one entry's slot with another's so + // `dup_slot` appears twice and `missing_slot` is absent. Length stays 32 so + // WrongLength cannot preempt. + let epoch = Epoch::new(RESPONSE_EPOCH); + let mut data = complete_epoch_duties(epoch); + let dup_slot = data[0].slot; + let missing_slot = data[1].slot; + assert_ne!( + dup_slot, missing_slot, + "precondition: the duplicated and omitted slots must differ" + ); + data[1].slot = dup_slot; + + // Act + let result = ProposerSchedule::from_response(epoch, SLOTS_PER_EPOCH, duties_response(data)); + + // Assert: exact duplicated slot. `dup_slot` is identical for whichever of the two matching + // entries is scanned second, so the payload is deterministic regardless of iteration order. + assert_eq!( + result.unwrap_err(), + DuplicateSlot(dup_slot), + "a slot appearing twice (with another absent) must be rejected as DuplicateSlot(dup_slot)" + ); + } } diff --git a/anchor/duties_tracker/src/lib.rs b/anchor/duties_tracker/src/lib.rs index 3eea0b849..bdb2c6c9d 100644 --- a/anchor/duties_tracker/src/lib.rs +++ b/anchor/duties_tracker/src/lib.rs @@ -2,10 +2,11 @@ use std::collections::HashMap; use bls::PublicKeyBytes; use dashmap::DashMap; -use eth2::types::ProposerData; +use eth2::types::{DutiesResponse, ProposerData}; use parking_lot::RwLock; use ssv_types::ValidatorIndex; -use types::{Epoch, Slot}; +use thiserror::Error; +use types::{Epoch, Hash256, Slot}; pub mod duties_tracker; pub mod voluntary_exit_tracker; @@ -84,12 +85,99 @@ impl SyncCommitteePerPeriod { } } -type ProposerMap = HashMap>; +type ProposerMap = HashMap; + +/// A proposer schedule validated as complete for one epoch (one duty per slot), +/// retained with the v2 response metadata used for refresh decisions and diagnostics. +#[derive(Debug, Clone)] +pub struct ProposerSchedule { + dependent_root: Hash256, + execution_optimistic: Option, + duties: Vec, +} + +#[derive(Debug, Error, PartialEq)] +pub enum ProposerScheduleError { + #[error("expected {expected} proposer duties, got {actual}")] + WrongLength { expected: usize, actual: usize }, + #[error("duty slot {0} is outside epoch {1}")] + SlotOutOfEpoch(Slot, Epoch), + #[error("slot {0} has more than one proposer duty")] + DuplicateSlot(Slot), +} + +impl ProposerSchedule { + /// Validate a v2 proposer-duties `response` as a complete schedule for `epoch`. + pub fn from_response( + epoch: Epoch, + slots_per_epoch: u64, + response: DutiesResponse>, + ) -> Result { + let expected = slots_per_epoch as usize; + if response.data.len() != expected { + return Err(ProposerScheduleError::WrongLength { + expected, + actual: response.data.len(), + }); + } + + let start_slot = epoch.start_slot(slots_per_epoch).as_u64(); + let mut seen = vec![false; expected]; + for duty in &response.data { + let offset = duty + .slot + .as_u64() + .checked_sub(start_slot) + .filter(|offset| *offset < slots_per_epoch) + .ok_or(ProposerScheduleError::SlotOutOfEpoch(duty.slot, epoch))? + as usize; + if seen[offset] { + return Err(ProposerScheduleError::DuplicateSlot(duty.slot)); + } + seen[offset] = true; + } + + Ok(Self { + dependent_root: response.dependent_root, + execution_optimistic: response.execution_optimistic, + duties: response.data, + }) + } + + pub fn duties(&self) -> &[ProposerData] { + &self.duties + } + + pub fn dependent_root(&self) -> Hash256 { + self.dependent_root + } + + pub fn execution_optimistic(&self) -> Option { + self.execution_optimistic + } +} + +#[cfg(test)] +impl ProposerSchedule { + /// Build a schedule from raw duties WITHOUT the completeness/validity checks that + /// `from_response` enforces. Test-only: lets `proposer_assignment_at_slot` tests seed + /// deliberately partial schedules that could never come from a validated response. + pub(crate) fn from_duties_unchecked(duties: Vec) -> Self { + // Brings `Hash256::zero()` (a `FixedBytesExtended` method) into scope for this + // test-only constructor without adding a non-test import to the crate. + use bls::FixedBytesExtended; + Self { + dependent_root: Hash256::zero(), + execution_optimistic: None, + duties, + } + } +} #[derive(Debug)] pub struct Duties { - /// Maps an epoch to all *local* proposers in this epoch. Notably, this does not contain - /// proposals for any validators which are not registered locally. + /// Maps an epoch to its validated complete proposer schedule (the full schedule for + /// every validator, not filtered by the local registry). pub proposers: RwLock, /// Map from validator index to sync committee duties. pub sync_duties: SyncCommitteePerPeriod,