p2p.rs (3524B)
1 use std::{ 2 collections::BTreeMap, 3 net::SocketAddr, 4 sync::{Arc, Mutex as StdMutex}, 5 time::Duration, 6 }; 7 8 use tokio::{ 9 sync::{Mutex, mpsc}, 10 task::JoinHandle, 11 }; 12 13 use crate::app::{GossipEnvelope, SharedNode, SharedPeerBook}; 14 15 mod fetch; 16 mod handshake; 17 mod identity; 18 mod inbound_limiter; 19 mod line_codec; 20 mod metrics; 21 mod network; 22 mod peer_addr; 23 mod peer_status; 24 mod process; 25 mod session; 26 mod sync; 27 #[cfg(test)] 28 mod test_support; 29 mod writer; 30 pub use fetch::{fetch_peer_height, fetch_snapshot, fetch_snapshot_with_announcement}; 31 use fetch::{ 32 network_adjusted_time_ms, validate_blocks_extension, validate_snapshot_extension, 33 verify_block_vdf, 34 }; 35 #[cfg(test)] 36 use handshake::verify_advertised_peer_node_id; 37 use handshake::{ 38 forget_stale_self_peer, process_hello, process_hello_with_verification, record_peer_status, 39 }; 40 #[cfg(test)] 41 use identity::peer_verification_response_for_node_id; 42 use identity::{new_node_id, peer_verification_response}; 43 #[cfg(test)] 44 use identity::{new_verification_nonce, peer_verification_response_is_valid}; 45 use inbound_limiter::{InboundConnectionLimiter, InboundSessionPermit, InboundSessionRejection}; 46 use line_codec::{LimitedLineReader, parse_envelope, read_session_envelope}; 47 pub use metrics::P2pMetrics; 48 use metrics::P2pMetricsCounters; 49 use peer_addr::{ 50 inbound_error_counts_as_misbehavior, is_possible_fork_error, is_quiet_disconnect, 51 is_self_peer_address_for, next_reconnect_delay as next_reconnect_delay_with_max, 52 normalize_advertised_peer, 53 }; 54 use peer_status::PeerStatus; 55 use process::{process_envelope, respond_to_peer_verification_challenge}; 56 use session::{accept_loop, outbound_session, outbound_supervisor}; 57 #[cfg(test)] 58 use sync::catchup_payload_for_peer; 59 use sync::{ 60 apply_peer_list, envelopes_for_peer, maybe_request_catchup, push_catchup_to_peer, 61 write_peer_exchange, 62 }; 63 use writer::{write_envelope, write_payload}; 64 65 const MAX_BLOCK_BATCH: usize = 128; 66 const MAX_OBJECT_REQUESTS: usize = 128; 67 const MAX_INVENTORY_ITEMS: usize = 512; 68 const MAX_PEER_LIST: usize = 128; 69 const MAX_SNAPSHOT_BLOCKS: usize = 10_000; 70 const MAX_GOSSIP_LINE_BYTES: usize = 8 * 1024 * 1024; 71 const MAX_INBOUND_SESSIONS: usize = 64; 72 const MAX_INBOUND_SESSIONS_PER_IP: usize = 8; 73 const MAX_INBOUND_ACCEPTS_PER_IP_PER_WINDOW: usize = 24; 74 const INBOUND_ACCEPT_RATE_WINDOW_MS: u64 = 10_000; 75 const PEER_QUEUE_SIZE: usize = 256; 76 const STALE_INBOUND_PEER_RETENTION_MS: u64 = 60 * 60 * 1_000; 77 const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); 78 const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(5); 79 const SESSION_SYNC_INTERVAL: Duration = Duration::from_secs(2); 80 const PEER_EXCHANGE_INTERVAL: Duration = Duration::from_secs(30); 81 const JOIN_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); 82 const MAX_JOIN_RESPONSE_ENVELOPES: usize = 16; 83 const MAX_PEER_VERIFICATION_ENVELOPES: usize = 8; 84 const INITIAL_RECONNECT_DELAY: Duration = Duration::from_secs(1); 85 const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(30); 86 type OutboundBatch = Vec<GossipEnvelope>; 87 88 #[derive(Clone)] 89 pub struct GossipNetwork { 90 inner: Arc<GossipNetworkInner>, 91 } 92 93 struct GossipNetworkInner { 94 node: SharedNode, 95 peers: SharedPeerBook, 96 listen_addr: SocketAddr, 97 p2p_announce_addr: Mutex<Option<SocketAddr>>, 98 node_id: String, 99 accept_task: Mutex<Option<JoinHandle<()>>>, 100 sessions: Mutex<BTreeMap<String, mpsc::Sender<OutboundBatch>>>, 101 inbound_limiter: Arc<StdMutex<InboundConnectionLimiter>>, 102 metrics: P2pMetricsCounters, 103 } 104 105 #[cfg(test)] 106 mod tests;