From fae7941b2d13dc9cd1ba8282aefe2798a70c7c74 Mon Sep 17 00:00:00 2001 From: Shane K Moore <41407272+shane-moore@users.noreply.github.com> Date: Sun, 26 Apr 2026 08:25:00 -0700 Subject: [PATCH] Gloas ptc duties beacon node response (#8415) Co-Authored-By: shane-moore Co-Authored-By: Eitan Seri-Levi Co-Authored-By: Eitan Seri-Levi Co-Authored-By: dapplion <35266934+dapplion@users.noreply.github.com> --- beacon_node/beacon_chain/src/beacon_chain.rs | 44 ++++- beacon_node/http_api/src/lib.rs | 10 + beacon_node/http_api/src/ptc_duties.rs | 182 +++++++++++++++++++ beacon_node/http_api/src/validator/mod.rs | 38 +++- beacon_node/http_api/tests/tests.rs | 120 +++++++++++- consensus/types/src/state/beacon_state.rs | 21 +++ 6 files changed, 411 insertions(+), 4 deletions(-) create mode 100644 beacon_node/http_api/src/ptc_duties.rs diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs index b556e6d849..bfe1b404e0 100644 --- a/beacon_node/beacon_chain/src/beacon_chain.rs +++ b/beacon_node/beacon_chain/src/beacon_chain.rs @@ -84,8 +84,8 @@ use crate::{ use bls::{PublicKey, PublicKeyBytes, Signature}; use eth2::beacon_response::ForkVersionedResponse; use eth2::types::{ - EventKind, SseBlobSidecar, SseBlock, SseDataColumnSidecar, SseExtendedPayloadAttributes, - SseHead, + EventKind, PtcDuty, SseBlobSidecar, SseBlock, SseDataColumnSidecar, + SseExtendedPayloadAttributes, SseHead, }; use execution_layer::{ BlockProposalContents, BlockProposalContentsType, BuilderParams, ChainHealth, ExecutionLayer, @@ -1719,6 +1719,46 @@ impl BeaconChain { Ok((duties, dependent_root, execution_status)) } + /// Get PTC duties for validators at a given epoch. + /// + /// TODO(gloas): per-validator `get_ptc_assignment` makes this O(N * slots_per_epoch * PTCSize). + /// A future ptc cache (or a single-pass `ptc_window` walk) can drop this to + /// O(slots_per_epoch * PTCSize + N). + pub fn compute_ptc_duties( + &self, + state: &BeaconState, + epoch: Epoch, + validator_indices: &[u64], + dependent_block_root: Hash256, + ) -> Result<(Vec>, Hash256), Error> { + // The ptc_window only covers previous, current, and next epochs. + let relative_epoch = RelativeEpoch::from_epoch(state.current_epoch(), epoch) + .map_err(Error::IncorrectStateForAttestation)?; + + let dependent_root = + state.attester_shuffling_decision_root(dependent_block_root, relative_epoch)?; + + let pubkey_cache = self.validator_pubkey_cache.read(); + + let duties = validator_indices + .iter() + .map(|&validator_index| -> Result, Error> { + let Some(&pubkey) = pubkey_cache.get_pubkey_bytes(validator_index as usize) else { + return Ok(None); + }; + let slot_opt = + state.get_ptc_assignment(validator_index as usize, epoch, &self.spec)?; + Ok(slot_opt.map(|slot| PtcDuty { + validator_index, + slot, + pubkey, + })) + }) + .collect::, _>>()?; + + Ok((duties, dependent_root)) + } + pub fn get_aggregated_attestation( &self, attestation: AttestationRef, diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs index 0be631c057..bd80dd1e82 100644 --- a/beacon_node/http_api/src/lib.rs +++ b/beacon_node/http_api/src/lib.rs @@ -19,6 +19,7 @@ mod metrics; mod peer; mod produce_block; mod proposer_duties; +mod ptc_duties; mod publish_attestations; mod publish_blocks; mod standard_block_rewards; @@ -2560,6 +2561,14 @@ pub fn serve( task_spawner_filter.clone(), ); + // POST validator/duties/ptc/{epoch} + let post_validator_duties_ptc = post_validator_duties_ptc( + eth_v1.clone(), + chain_filter.clone(), + not_while_syncing_filter.clone(), + task_spawner_filter.clone(), + ); + // POST validator/duties/sync/{epoch} let post_validator_duties_sync = post_validator_duties_sync( eth_v1.clone(), @@ -3410,6 +3419,7 @@ pub fn serve( .uor(post_beacon_rewards_attestations) .uor(post_beacon_rewards_sync_committee) .uor(post_validator_duties_attester) + .uor(post_validator_duties_ptc) .uor(post_validator_duties_sync) .uor(post_validator_aggregate_and_proofs) .uor(post_validator_contribution_and_proofs) diff --git a/beacon_node/http_api/src/ptc_duties.rs b/beacon_node/http_api/src/ptc_duties.rs new file mode 100644 index 0000000000..f727b84004 --- /dev/null +++ b/beacon_node/http_api/src/ptc_duties.rs @@ -0,0 +1,182 @@ +//! Contains the handler for the `POST validator/duties/ptc/{epoch}` endpoint. + +use crate::state_id::StateId; +use beacon_chain::{BeaconChain, BeaconChainError, BeaconChainTypes}; +use eth2::types::{self as api_types, PtcDuty}; +use slot_clock::SlotClock; +use state_processing::state_advance::partial_state_advance; +use types::{BeaconState, ChainSpec, Epoch, EthSpec, Hash256}; + +type ApiDuties = api_types::DutiesResponse>; + +pub fn ptc_duties( + request_epoch: Epoch, + request_indices: &[u64], + chain: &BeaconChain, +) -> Result { + let current_epoch = chain + .slot_clock + .now_or_genesis() + .map(|slot| slot.epoch(T::EthSpec::slots_per_epoch())) + .ok_or(BeaconChainError::UnableToReadSlot) + .map_err(warp_utils::reject::unhandled_error)?; + + let tolerant_current_epoch = if chain.slot_clock.is_prior_to_genesis().unwrap_or(true) { + current_epoch + } else { + chain + .slot_clock + .now_with_future_tolerance(chain.spec.maximum_gossip_clock_disparity()) + .ok_or_else(|| { + warp_utils::reject::custom_server_error("unable to read slot clock".into()) + })? + .epoch(T::EthSpec::slots_per_epoch()) + }; + + let is_within_clock_tolerance = request_epoch == current_epoch + || request_epoch == current_epoch + 1 + || request_epoch == tolerant_current_epoch + 1; + + if is_within_clock_tolerance { + let head_epoch = chain + .canonical_head + .cached_head() + .snapshot + .beacon_state + .current_epoch(); + + let head_can_serve_request = request_epoch == head_epoch || request_epoch == head_epoch + 1; + + if head_can_serve_request { + compute_ptc_duties_from_cached_head(request_epoch, request_indices, chain) + } else { + compute_ptc_duties_from_state(request_epoch, request_indices, chain) + } + } else if request_epoch > current_epoch + 1 { + Err(warp_utils::reject::custom_bad_request(format!( + "request epoch {} is more than one epoch past the current epoch {}", + request_epoch, current_epoch + ))) + } else { + compute_ptc_duties_from_state(request_epoch, request_indices, chain) + } +} + +fn compute_ptc_duties_from_cached_head( + request_epoch: Epoch, + request_indices: &[u64], + chain: &BeaconChain, +) -> Result { + let (cached_head, execution_status) = chain + .canonical_head + .head_and_execution_status() + .map_err(warp_utils::reject::unhandled_error)?; + let state = &cached_head.snapshot.beacon_state; + let head_block_root = cached_head.head_block_root(); + + let (duties, dependent_root) = chain + .compute_ptc_duties(state, request_epoch, request_indices, head_block_root) + .map_err(warp_utils::reject::unhandled_error)?; + + convert_to_api_response( + duties, + dependent_root, + execution_status.is_optimistic_or_invalid(), + ) +} + +fn compute_ptc_duties_from_state( + request_epoch: Epoch, + request_indices: &[u64], + chain: &BeaconChain, +) -> Result { + let state_opt = { + let (cached_head, execution_status) = chain + .canonical_head + .head_and_execution_status() + .map_err(warp_utils::reject::unhandled_error)?; + let head = &cached_head.snapshot; + + if head.beacon_state.current_epoch() <= request_epoch { + Some(( + head.beacon_state_root(), + head.beacon_state.clone(), + execution_status.is_optimistic_or_invalid(), + )) + } else { + None + } + }; + + let (state, execution_optimistic) = + if let Some((state_root, mut state, execution_optimistic)) = state_opt { + ensure_state_knows_ptc_duties_for_epoch( + &mut state, + state_root, + request_epoch, + &chain.spec, + )?; + (state, execution_optimistic) + } else { + let (state, execution_optimistic, _finalized) = + StateId::from_slot(request_epoch.start_slot(T::EthSpec::slots_per_epoch())) + .state(chain)?; + (state, execution_optimistic) + }; + + if !(state.current_epoch() == request_epoch || state.current_epoch() + 1 == request_epoch) { + return Err(warp_utils::reject::custom_server_error(format!( + "state epoch {} not suitable for request epoch {}", + state.current_epoch(), + request_epoch + ))); + } + + let (duties, dependent_root) = chain + .compute_ptc_duties( + &state, + request_epoch, + request_indices, + chain.genesis_block_root, + ) + .map_err(warp_utils::reject::unhandled_error)?; + + convert_to_api_response(duties, dependent_root, execution_optimistic) +} + +fn ensure_state_knows_ptc_duties_for_epoch( + state: &mut BeaconState, + state_root: Hash256, + target_epoch: Epoch, + spec: &ChainSpec, +) -> Result<(), warp::reject::Rejection> { + if state.current_epoch() > target_epoch { + return Err(warp_utils::reject::custom_server_error(format!( + "state epoch {} is later than target epoch {}", + state.current_epoch(), + target_epoch + ))); + } else if state.current_epoch() + 1 < target_epoch { + let target_slot = target_epoch + .saturating_sub(1_u64) + .start_slot(E::slots_per_epoch()); + + partial_state_advance(state, Some(state_root), target_slot, spec) + .map_err(BeaconChainError::from) + .map_err(warp_utils::reject::unhandled_error)?; + } + + Ok(()) +} + +fn convert_to_api_response( + duties: Vec>, + dependent_root: Hash256, + execution_optimistic: bool, +) -> Result { + Ok(api_types::DutiesResponse { + dependent_root, + execution_optimistic: Some(execution_optimistic), + data: duties.into_iter().flatten().collect(), + }) +} diff --git a/beacon_node/http_api/src/validator/mod.rs b/beacon_node/http_api/src/validator/mod.rs index 7349aa4db0..27fe5de6e7 100644 --- a/beacon_node/http_api/src/validator/mod.rs +++ b/beacon_node/http_api/src/validator/mod.rs @@ -7,7 +7,7 @@ use crate::utils::{ ResponseFilter, TaskSpawnerFilter, ValidatorSubscriptionTxFilter, publish_network_message, }; use crate::version::{V1, V2, V3, unsupported_version_rejection}; -use crate::{StateId, attester_duties, proposer_duties, sync_committees}; +use crate::{StateId, attester_duties, proposer_duties, ptc_duties, sync_committees}; use beacon_chain::attestation_verification::VerifiedAttestation; use beacon_chain::{AttestationError, BeaconChain, BeaconChainError, BeaconChainTypes}; use bls::PublicKeyBytes; @@ -168,6 +168,42 @@ pub fn post_validator_duties_attester( .boxed() } +// POST validator/duties/ptc/{epoch} +pub fn post_validator_duties_ptc( + eth_v1: EthV1Filter, + chain_filter: ChainFilter, + not_while_syncing_filter: NotWhileSyncingFilter, + task_spawner_filter: TaskSpawnerFilter, +) -> ResponseFilter { + eth_v1 + .and(warp::path("validator")) + .and(warp::path("duties")) + .and(warp::path("ptc")) + .and(warp::path::param::().or_else(|_| async { + Err(warp_utils::reject::custom_bad_request( + "Invalid epoch".to_string(), + )) + })) + .and(warp::path::end()) + .and(not_while_syncing_filter.clone()) + .and(warp_utils::json::json()) + .and(task_spawner_filter.clone()) + .and(chain_filter.clone()) + .then( + |epoch: Epoch, + not_synced_filter: Result<(), Rejection>, + indices: ValidatorIndexData, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P0, move || { + not_synced_filter?; + ptc_duties::ptc_duties(epoch, &indices.0, &chain) + }) + }, + ) + .boxed() +} + // GET validator/aggregate_attestation?attestation_data_root,slot pub fn get_validator_aggregate_attestation( any_version: AnyVersionFilter, diff --git a/beacon_node/http_api/tests/tests.rs b/beacon_node/http_api/tests/tests.rs index 2dd4c28040..aac3384fbd 100644 --- a/beacon_node/http_api/tests/tests.rs +++ b/beacon_node/http_api/tests/tests.rs @@ -3474,7 +3474,6 @@ impl ApiTester { self } - // TODO(EIP-7732): Add test_get_validator_duties_ptc function to test PTC duties endpoint pub async fn test_get_validator_duties_proposer_v2(self) -> Self { let current_epoch = self.chain.epoch().unwrap(); @@ -3598,6 +3597,17 @@ impl ApiTester { "should not get attester duties outside of tolerance" ); + assert_eq!( + self.client + .post_validator_duties_ptc(next_epoch, &[0]) + .await + .unwrap_err() + .status() + .map(Into::into), + Some(400), + "should not get ptc duties outside of tolerance" + ); + self.chain.slot_clock.set_current_time( current_epoch_start - self.chain.spec.maximum_gossip_clock_disparity(), ); @@ -3621,6 +3631,88 @@ impl ApiTester { .await .expect("should get attester duties within tolerance"); + self.client + .post_validator_duties_ptc(next_epoch, &[0]) + .await + .expect("should get ptc duties within tolerance"); + + self + } + + pub async fn test_get_validator_duties_ptc(self) -> Self { + let current_epoch = self.chain.epoch().unwrap().as_u64(); + + let half = current_epoch / 2; + let first = current_epoch - half; + let last = current_epoch + half; + + for epoch in first..=last { + for indices in self.interesting_validator_indices() { + let epoch = Epoch::from(epoch); + + // The endpoint does not allow getting duties past the next epoch. + if epoch > current_epoch + 1 { + assert_eq!( + self.client + .post_validator_duties_ptc(epoch, indices.as_slice()) + .await + .unwrap_err() + .status() + .map(Into::into), + Some(400) + ); + continue; + } + + let results = self + .client + .post_validator_duties_ptc(epoch, indices.as_slice()) + .await + .unwrap(); + + let dependent_root = self + .chain + .block_root_at_slot( + (epoch - 1).start_slot(E::slots_per_epoch()) - 1, + WhenSlotSkipped::Prev, + ) + .unwrap() + .unwrap_or(self.chain.head_beacon_block_root()); + + assert_eq!(results.dependent_root, dependent_root); + + let result_duties = results.data; + + let state = self + .chain + .state_at_slot( + epoch.start_slot(E::slots_per_epoch()), + StateSkipConfig::WithStateRoots, + ) + .unwrap(); + + let expected_duties: Vec = indices + .iter() + .filter_map(|&validator_index| { + let validator = state.validators().get(validator_index as usize)?; + let slot = state + .get_ptc_assignment(validator_index as usize, epoch, &self.chain.spec) + .unwrap()?; + Some(PtcDuty { + pubkey: validator.pubkey, + validator_index, + slot, + }) + }) + .collect(); + + assert_eq!( + result_duties, expected_duties, + "ptc duties should exactly match state assignments" + ); + } + } + self } @@ -7871,6 +7963,9 @@ async fn get_light_client_finality_update() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn get_validator_duties_early() { + if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) { + return; + } ApiTester::new() .await .test_get_validator_duties_early() @@ -7936,6 +8031,29 @@ async fn get_validator_duties_proposer_v2_with_skip_slots() { .await; } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn get_validator_duties_ptc() { + if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) { + return; + } + ApiTester::new_with_hard_forks() + .await + .test_get_validator_duties_ptc() + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn get_validator_duties_ptc_with_skip_slots() { + if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) { + return; + } + ApiTester::new_with_hard_forks() + .await + .skip_slots(E::slots_per_epoch() * 2) + .test_get_validator_duties_ptc() + .await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn block_production() { ApiTester::new().await.test_block_production().await; diff --git a/consensus/types/src/state/beacon_state.rs b/consensus/types/src/state/beacon_state.rs index 7e2b3096a8..7ed3121d6e 100644 --- a/consensus/types/src/state/beacon_state.rs +++ b/consensus/types/src/state/beacon_state.rs @@ -3198,6 +3198,27 @@ impl BeaconState { Ok(hash(&preimage)) } + /// Find the first slot in the given epoch where the validator is assigned to the PTC. + /// + /// Returns `Ok(Some(slot))` if the validator is in the PTC for any slot in the epoch, + /// `Ok(None)` if the validator is not in the PTC for this epoch. + /// + /// This iterates through all slots in the epoch, so it's O(slots_per_epoch) per validator. + pub fn get_ptc_assignment( + &self, + validator_index: usize, + epoch: Epoch, + spec: &ChainSpec, + ) -> Result, BeaconStateError> { + for slot in epoch.slot_iter(E::slots_per_epoch()) { + let ptc = self.get_ptc(slot, spec)?; + if ptc.0.contains(&validator_index) { + return Ok(Some(slot)); + } + } + Ok(None) + } + /// Return size indices sampled by effective balance, using indices as candidates. /// /// If shuffle_indices is True, candidate indices are themselves sampled from indices