Anchor pre-PR: Decouple eth2 (#6770)

* decouple `eth2` from `store` and `lighthouse_network`

Co-authored-by: Pawan Dhananjay <pawandhananjay@gmail.com>

* remove unused dependency

---------

Co-authored-by: Pawan Dhananjay <pawandhananjay@gmail.com>
Co-authored-by: Jimmy Chen <jimmy@sigmaprime.io>
This commit is contained in:
Daniel Knopik
2025-01-09 13:48:18 +01:00
committed by GitHub
parent 320843024c
commit 9a4768a771
14 changed files with 95 additions and 86 deletions

View File

@@ -12,7 +12,6 @@ arc-swap = "1.6.0"
builder_client = { path = "../builder_client" }
bytes = { workspace = true }
eth2 = { workspace = true }
eth2_network_config = { workspace = true }
ethereum_serde_utils = { workspace = true }
ethereum_ssz = { workspace = true }
ethers-core = { workspace = true }

View File

@@ -30,6 +30,7 @@ rand = { workspace = true }
safe_arith = { workspace = true }
sensitive_url = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
slog = { workspace = true }
slot_clock = { workspace = true }
state_processing = { workspace = true }
@@ -48,7 +49,6 @@ warp_utils = { workspace = true }
genesis = { workspace = true }
logging = { workspace = true }
proto_array = { workspace = true }
serde_json = { workspace = true }
[[test]]
name = "bn_http_api_tests"

View File

@@ -1,7 +1,17 @@
use beacon_chain::store::metadata::CURRENT_SCHEMA_VERSION;
use beacon_chain::{BeaconChain, BeaconChainTypes};
use eth2::lighthouse::DatabaseInfo;
use serde::Serialize;
use std::sync::Arc;
use store::{AnchorInfo, BlobInfo, Split, StoreConfig};
#[derive(Debug, Serialize)]
pub struct DatabaseInfo {
pub schema_version: u64,
pub config: StoreConfig,
pub split: Split,
pub anchor: AnchorInfo,
pub blob_info: BlobInfo,
}
pub fn info<T: BeaconChainTypes>(
chain: Arc<BeaconChain<T>>,

View File

@@ -3059,9 +3059,9 @@ pub fn serve<T: BeaconChainTypes>(
peer_id: peer_id.to_string(),
enr: peer_info.enr().map(|enr| enr.to_base64()),
last_seen_p2p_address: address,
direction: api_types::PeerDirection::from_connection_direction(dir),
state: api_types::PeerState::from_peer_connection_status(
peer_info.connection_status(),
direction: api_types::PeerDirection::from((*dir).clone()),
state: api_types::PeerState::from(
peer_info.connection_status().clone(),
),
}));
}
@@ -3104,10 +3104,9 @@ pub fn serve<T: BeaconChainTypes>(
// the eth2 API spec implies only peers we have been connected to at some point should be included.
if let Some(dir) = peer_info.connection_direction() {
let direction =
api_types::PeerDirection::from_connection_direction(dir);
let state = api_types::PeerState::from_peer_connection_status(
peer_info.connection_status(),
let direction = api_types::PeerDirection::from((*dir).clone());
let state = api_types::PeerState::from(
peer_info.connection_status().clone(),
);
let state_matches = query.state.as_ref().map_or(true, |states| {
@@ -3160,9 +3159,8 @@ pub fn serve<T: BeaconChainTypes>(
.read()
.peers()
.for_each(|(_, peer_info)| {
let state = api_types::PeerState::from_peer_connection_status(
peer_info.connection_status(),
);
let state =
api_types::PeerState::from(peer_info.connection_status().clone());
match state {
api_types::PeerState::Connected => connected += 1,
api_types::PeerState::Connecting => connecting += 1,
@@ -4175,15 +4173,18 @@ pub fn serve<T: BeaconChainTypes>(
|task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
task_spawner.blocking_json_task(Priority::P1, move || {
Ok(network_globals
.peers
.read()
.peers()
.map(|(peer_id, peer_info)| eth2::lighthouse::Peer {
let mut peers = vec![];
for (peer_id, peer_info) in network_globals.peers.read().peers() {
peers.push(eth2::lighthouse::Peer {
peer_id: peer_id.to_string(),
peer_info: peer_info.clone(),
})
.collect::<Vec<_>>())
peer_info: serde_json::to_value(peer_info).map_err(|e| {
warp_utils::reject::custom_not_found(format!(
"unable to serialize peer_info: {e:?}",
))
})?,
});
}
Ok(peers)
})
},
);
@@ -4199,15 +4200,18 @@ pub fn serve<T: BeaconChainTypes>(
|task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
task_spawner.blocking_json_task(Priority::P1, move || {
Ok(network_globals
.peers
.read()
.connected_peers()
.map(|(peer_id, peer_info)| eth2::lighthouse::Peer {
let mut peers = vec![];
for (peer_id, peer_info) in network_globals.peers.read().connected_peers() {
peers.push(eth2::lighthouse::Peer {
peer_id: peer_id.to_string(),
peer_info: peer_info.clone(),
})
.collect::<Vec<_>>())
peer_info: serde_json::to_value(peer_info).map_err(|e| {
warp_utils::reject::custom_not_found(format!(
"unable to serialize peer_info: {e:?}",
))
})?,
});
}
Ok(peers)
})
},
);

View File

@@ -35,6 +35,7 @@ use state_processing::per_slot_processing;
use state_processing::state_advance::partial_state_advance;
use std::convert::TryInto;
use std::sync::Arc;
use store::{AnchorInfo, Split};
use tokio::time::Duration;
use tree_hash::TreeHash;
use types::application_domain::ApplicationDomain;
@@ -5646,10 +5647,16 @@ impl ApiTester {
pub async fn test_get_lighthouse_database_info(self) -> Self {
let info = self.client.get_lighthouse_database_info().await.unwrap();
assert_eq!(info.anchor, self.chain.store.get_anchor_info());
assert_eq!(info.split, self.chain.store.get_split_info());
assert_eq!(
info.schema_version,
serde_json::from_value::<AnchorInfo>(info.get("anchor").unwrap().clone()).unwrap(),
self.chain.store.get_anchor_info()
);
assert_eq!(
serde_json::from_value::<Split>(info.get("split").unwrap().clone()).unwrap(),
self.chain.store.get_split_info()
);
assert_eq!(
serde_json::from_value::<u64>(info.get("schema_version").unwrap().clone()).unwrap(),
store::metadata::CURRENT_SCHEMA_VERSION.as_u64()
);

View File

@@ -13,6 +13,7 @@ directory = { workspace = true }
dirs = { workspace = true }
discv5 = { workspace = true }
either = { workspace = true }
eth2 = { workspace = true }
ethereum_ssz = { workspace = true }
ethereum_ssz_derive = { workspace = true }
fnv = { workspace = true }

View File

@@ -4,6 +4,7 @@ use super::sync_status::SyncStatus;
use crate::discovery::Eth2Enr;
use crate::{rpc::MetaData, types::Subnet};
use discv5::Enr;
use eth2::types::{PeerDirection, PeerState};
use libp2p::core::multiaddr::{Multiaddr, Protocol};
use serde::{
ser::{SerializeStruct, Serializer},
@@ -506,6 +507,15 @@ pub enum ConnectionDirection {
Outgoing,
}
impl From<ConnectionDirection> for PeerDirection {
fn from(direction: ConnectionDirection) -> Self {
match direction {
ConnectionDirection::Incoming => PeerDirection::Inbound,
ConnectionDirection::Outgoing => PeerDirection::Outbound,
}
}
}
/// Connection Status of the peer.
#[derive(Debug, Clone, Default)]
pub enum PeerConnectionStatus {
@@ -599,3 +609,14 @@ impl Serialize for PeerConnectionStatus {
}
}
}
impl From<PeerConnectionStatus> for PeerState {
fn from(status: PeerConnectionStatus) -> Self {
match status {
Connected { .. } => PeerState::Connected,
Dialing { .. } => PeerState::Connecting,
Disconnecting { .. } => PeerState::Disconnecting,
Disconnected { .. } | Banned { .. } | Unknown => PeerState::Disconnected,
}
}
}

View File

@@ -1,7 +1,6 @@
mod globals;
mod pubsub;
mod subnet;
mod sync_state;
mod topics;
use types::{BitVector, EthSpec};
@@ -11,10 +10,10 @@ pub type EnrSyncCommitteeBitfield<E> = BitVector<<E as EthSpec>::SyncCommitteeSu
pub type Enr = discv5::enr::Enr<discv5::enr::CombinedKey>;
pub use eth2::lighthouse::sync_state::{BackFillState, SyncState};
pub use globals::NetworkGlobals;
pub use pubsub::{PubsubMessage, SnappyTransform};
pub use subnet::{Subnet, SubnetDiscovery};
pub use sync_state::{BackFillState, SyncState};
pub use topics::{
attestation_sync_committee_topics, core_topics_to_subscribe, fork_core_topics,
subnet_from_topic_hash, GossipEncoding, GossipKind, GossipTopic, ALTAIR_CORE_TOPICS,

View File

@@ -1,113 +0,0 @@
use serde::{Deserialize, Serialize};
use types::Slot;
/// The current state of the node.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum SyncState {
/// The node is performing a long-range (batch) sync over a finalized chain.
/// In this state, parent lookups are disabled.
SyncingFinalized { start_slot: Slot, target_slot: Slot },
/// The node is performing a long-range (batch) sync over one or many head chains.
/// In this state parent lookups are disabled.
SyncingHead { start_slot: Slot, target_slot: Slot },
/// The node is undertaking a backfill sync. This occurs when a user has specified a trusted
/// state. The node first syncs "forward" by downloading blocks up to the current head as
/// specified by its peers. Once completed, the node enters this sync state and attempts to
/// download all required historical blocks.
BackFillSyncing { completed: usize, remaining: usize },
/// The node has completed syncing a finalized chain and is in the process of re-evaluating
/// which sync state to progress to.
SyncTransition,
/// The node is up to date with all known peers and is connected to at least one
/// fully synced peer. In this state, parent lookups are enabled.
Synced,
/// No useful peers are connected. Long-range sync's cannot proceed and we have no useful
/// peers to download parents for. More peers need to be connected before we can proceed.
Stalled,
}
#[derive(PartialEq, Debug, Clone, Serialize, Deserialize)]
/// The state of the backfill sync.
pub enum BackFillState {
/// The sync is partially completed and currently paused.
Paused,
/// We are currently backfilling.
Syncing,
/// A backfill sync has completed.
Completed,
/// Too many failed attempts at backfilling. Consider it failed.
Failed,
}
impl PartialEq for SyncState {
fn eq(&self, other: &Self) -> bool {
matches!(
(self, other),
(
SyncState::SyncingFinalized { .. },
SyncState::SyncingFinalized { .. }
) | (SyncState::SyncingHead { .. }, SyncState::SyncingHead { .. })
| (SyncState::Synced, SyncState::Synced)
| (SyncState::Stalled, SyncState::Stalled)
| (SyncState::SyncTransition, SyncState::SyncTransition)
| (
SyncState::BackFillSyncing { .. },
SyncState::BackFillSyncing { .. }
)
)
}
}
impl SyncState {
/// Returns a boolean indicating the node is currently performing a long-range sync.
pub fn is_syncing(&self) -> bool {
match self {
SyncState::SyncingFinalized { .. } => true,
SyncState::SyncingHead { .. } => true,
SyncState::SyncTransition => true,
// Backfill doesn't effect any logic, we consider this state, not syncing.
SyncState::BackFillSyncing { .. } => false,
SyncState::Synced => false,
SyncState::Stalled => false,
}
}
pub fn is_syncing_finalized(&self) -> bool {
match self {
SyncState::SyncingFinalized { .. } => true,
SyncState::SyncingHead { .. } => false,
SyncState::SyncTransition => false,
SyncState::BackFillSyncing { .. } => false,
SyncState::Synced => false,
SyncState::Stalled => false,
}
}
/// Returns true if the node is synced.
///
/// NOTE: We consider the node synced if it is fetching old historical blocks.
pub fn is_synced(&self) -> bool {
matches!(self, SyncState::Synced | SyncState::BackFillSyncing { .. })
}
/// Returns true if the node is *stalled*, i.e. has no synced peers.
///
/// Usually this state is treated as unsynced, except in some places where we make an exception
/// for single-node testnets where having 0 peers is desired.
pub fn is_stalled(&self) -> bool {
matches!(self, SyncState::Stalled)
}
}
impl std::fmt::Display for SyncState {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
SyncState::SyncingFinalized { .. } => write!(f, "Syncing Finalized Chain"),
SyncState::SyncingHead { .. } => write!(f, "Syncing Head Chain"),
SyncState::Synced { .. } => write!(f, "Synced"),
SyncState::Stalled { .. } => write!(f, "Stalled"),
SyncState::SyncTransition => write!(f, "Evaluating known peers"),
SyncState::BackFillSyncing { .. } => write!(f, "Syncing Historical Blocks"),
}
}
}