fix: validate PTC duty assignments

anchor/duties_tracker/Cargo.toml

@@ -21,6 +21,8 @@ tracing = { workspace = true }
 types = { workspace = true }
 
 [dev-dependencies]
+axum = { workspace = true }
 database = { workspace = true, features = ["test-utils"] }
 openssl = { workspace = true }
 sensitive_url = { workspace = true }
+serde_json = { workspace = true }

anchor/duties_tracker/src/duties_tracker.rs

@@ -15,7 +15,7 @@ use types::{ChainSpec, Epoch, Slot};
 
 use crate::{
     Duties, DutiesProvider, DutyAssignment, MembershipKey, ProposerSchedule, ProposerScheduleError,
-    voluntary_exit_tracker::VoluntaryExitTracker,
+    PtcSchedule, PtcScheduleError, voluntary_exit_tracker::VoluntaryExitTracker,
 };
 
 /// Only retain `HISTORICAL_DUTIES_EPOCHS` duties prior to the current epoch.
@@ -29,6 +29,10 @@ pub enum Error {
     Arith(ArithError),
     #[error("Failed to poll proposers: {0}")]
     FailedToPollProposers(String),
+    #[error("Failed to poll PTC duties: {0}")]
+    FailedToPollPtc(String),
+    #[error("Invalid PTC duties: {0}")]
+    InvalidPtcDuties(#[from] PtcScheduleError),
 }
 
 pub struct DutiesTracker<T: SlotClock + 'static> {
@@ -260,6 +264,56 @@ impl<T: SlotClock + 'static> DutiesTracker<T> {
         Ok(())
     }
 
+    /// Replace the current PTC view using exactly the indices captured for this request.
+    async fn poll_ptc_duties(&self) -> Result<(), Error> {
+        let current_slot = self.slot_clock.now().ok_or(Error::UnableToReadSlotClock)?;
+        let current_epoch = current_slot.epoch(self.slots_per_epoch);
+
+        // Retain one previous epoch for PTC messages arriving after an epoch boundary.
+        self.duties
+            .ptc
+            .write()
+            .retain(|&epoch, _| epoch >= current_epoch.saturating_sub(1u64));
+
+        if self
+            .spec
+            .gloas_fork_epoch
+            .is_none_or(|gloas_epoch| current_epoch < gloas_epoch)
+        {
+            return Ok(());
+        }
+
+        // Release the database watch borrow before HTTP. Later additions remain unknown
+        // until a subsequent request includes them.
+        let validator_indices = self.network_state_rx.borrow().validator_indices();
+        if validator_indices.is_empty() {
+            self.duties.ptc.write().remove(&current_epoch);
+            return Ok(());
+        }
+
+        let response = self
+            .beacon_nodes
+            .first_success(|beacon_node| {
+                let indices = &validator_indices;
+                async move {
+                    beacon_node
+                        .post_validator_duties_ptc(current_epoch, indices)
+                        .await
+                }
+            })
+            .await
+            .map_err(|error| Error::FailedToPollPtc(error.to_string()))?;
+
+        let schedule = PtcSchedule::from_response(
+            current_epoch,
+            self.slots_per_epoch,
+            &validator_indices,
+            response,
+        )?;
+        self.duties.ptc.write().insert(current_epoch, schedule);
+        Ok(())
+    }
+
     pub fn start(self: Arc<Self>, executor: TaskExecutor) {
         let self_clone = self.clone();
         self_clone.spawn_polling_task(
@@ -272,6 +326,15 @@ impl<T: SlotClock + 'static> DutiesTracker<T> {
             executor.clone(),
         );
 
+        if self.spec.gloas_fork_epoch.is_some() {
+            self.clone().spawn_polling_task(
+                |tracker| async move { tracker.poll_ptc_duties().await },
+                "Failed to poll PTC duties",
+                "ptc_tracker",
+                executor.clone(),
+            );
+        }
+
         self.spawn_polling_task(
             |tracker| {
                 let tracker = tracker.clone();
@@ -381,6 +444,20 @@ impl<T: SlotClock + 'static> DutiesProvider for DutiesTracker<T> {
             None => DutyAssignment::Unknown,
         }
     }
+
+    fn ptc_assignment_at_slot(
+        &self,
+        slot: Slot,
+        validator_index: ValidatorIndex,
+    ) -> DutyAssignment {
+        self.duties
+            .ptc
+            .read()
+            .get(&slot.epoch(self.slots_per_epoch))
+            .map_or(DutyAssignment::Unknown, |schedule| {
+                schedule.assignment_at_slot(slot, validator_index.into())
+            })
+    }
 }
 
 /// Number of epochs to wait from the start of the period before actually fetching duties.
@@ -1042,3 +1119,7 @@ mod tests {
         );
     }
 }
+
+#[cfg(test)]
+#[path = "ptc_tests.rs"]
+mod ptc_tests;

anchor/duties_tracker/src/lib.rs

@@ -2,7 +2,7 @@ use std::collections::HashMap;
 
 use bls::PublicKeyBytes;
 use dashmap::DashMap;
-use eth2::types::{DutiesResponse, ProposerData};
+use eth2::types::{DutiesResponse, ProposerData, PtcDuty};
 use parking_lot::RwLock;
 use ssv_types::ValidatorIndex;
 use thiserror::Error;
@@ -181,13 +181,16 @@ pub struct Duties {
     pub proposers: RwLock<ProposerMap>,
     /// Map from validator index to sync committee duties.
     pub sync_duties: SyncCommitteePerPeriod,
+    /// PTC snapshots include every queried index, including validators with no duty.
+    pub(crate) ptc: RwLock<HashMap<Epoch, PtcSchedule>>,
 }
 
 impl Duties {
     pub fn new() -> Self {
         Self {
             proposers: RwLock::new(HashMap::new()),
             sync_duties: SyncCommitteePerPeriod::new(),
+            ptc: RwLock::new(HashMap::new()),
         }
     }
 }
@@ -199,19 +202,66 @@ impl Default for Duties {
 }
 
 /// Whether a validator holds a duty at a slot, as one atomic verdict over the stored duty view.
-///
-/// A retained view is never revoked by a failed or malformed refresh, a local registry change,
-/// `execution_optimistic`, or a reorg. It changes only when a complete schedule replaces it.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum DutyAssignment {
-    /// A complete fetched view assigns the validator at this slot.
+    /// A fetched view covering this validator assigns it at this slot.
     Assigned,
-    /// A complete fetched view proves the validator is not assigned at this slot.
+    /// A fetched view covering this validator proves it is not assigned at this slot.
     NotAssigned,
-    /// The view cannot answer: no completed fetch for the slot's epoch.
+    /// No fetched view covers this validator for the slot's epoch.
     Unknown,
 }
 
+/// A sparse PTC response together with its exact requested-index coverage.
+#[derive(Debug)]
+pub(crate) struct PtcSchedule {
+    assignments: HashMap<u64, Option<Slot>>,
+}
+
+#[derive(Debug, Error, PartialEq)]
+pub enum PtcScheduleError {
+    #[error("PTC duty slot {0} is outside epoch {1}")]
+    SlotOutOfEpoch(Slot, Epoch),
+    #[error("PTC response includes unrequested validator {0}")]
+    UnrequestedValidator(u64),
+    #[error("PTC response includes multiple duties for validator {0}")]
+    DuplicateValidator(u64),
+}
+
+impl PtcSchedule {
+    fn from_response(
+        epoch: Epoch,
+        slots_per_epoch: u64,
+        requested_indices: &[u64],
+        response: DutiesResponse<Vec<PtcDuty>>,
+    ) -> Result<Self, PtcScheduleError> {
+        let mut assignments: HashMap<_, _> = requested_indices
+            .iter()
+            .map(|&index| (index, None))
+            .collect();
+        for duty in response.data {
+            if duty.slot.epoch(slots_per_epoch) != epoch {
+                return Err(PtcScheduleError::SlotOutOfEpoch(duty.slot, epoch));
+            }
+            let assignment = assignments
+                .get_mut(&duty.validator_index)
+                .ok_or(PtcScheduleError::UnrequestedValidator(duty.validator_index))?;
+            if assignment.replace(duty.slot).is_some() {
+                return Err(PtcScheduleError::DuplicateValidator(duty.validator_index));
+            }
+        }
+        Ok(Self { assignments })
+    }
+
+    fn assignment_at_slot(&self, slot: Slot, validator_index: u64) -> DutyAssignment {
+        match self.assignments.get(&validator_index) {
+            Some(Some(assigned_slot)) if *assigned_slot == slot => DutyAssignment::Assigned,
+            Some(_) => DutyAssignment::NotAssigned,
+            None => DutyAssignment::Unknown,
+        }
+    }
+}
+
 pub trait DutiesProvider: Sync + Send + 'static {
     fn is_validator_in_sync_committee(
         &self,
@@ -226,9 +276,16 @@ pub trait DutiesProvider: Sync + Send + 'static {
 
     fn get_voluntary_exit_duty_count(&self, slot: Slot, pubkey: &PublicKeyBytes) -> u64;
 
+    /// A retained complete proposer view is not revoked by a failed or malformed refresh,
+    /// local registry changes, `execution_optimistic`, or a reorg. Only a complete replacement
+    /// changes its assignments.
     fn proposer_assignment_at_slot(
         &self,
         slot: Slot,
         validator_pubkey: &PublicKeyBytes,
     ) -> DutyAssignment;
+
+    /// 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;
 }