Fix packet count when talking with PAR2 peers

ethcore/light/src/net/mod.rs

@@ -75,14 +75,17 @@ const RECALCULATE_COSTS_INTERVAL: Duration = Duration::from_secs(60 * 60);
 // minimum interval between updates.
 const UPDATE_INTERVAL: Duration = Duration::from_millis(5000);
 
+/// Packet count for PIP.
+const PACKET_COUNT_V1: u8 = 9;
+
 /// Supported protocol versions.
-pub const PROTOCOL_VERSIONS: &'static [u8] = &[1];
+pub const PROTOCOL_VERSIONS: &'static [(u8, u8)] = &[
+	(1, PACKET_COUNT_V1),
+];
 
 /// Max protocol version.
 pub const MAX_PROTOCOL_VERSION: u8 = 1;
 
-/// Packet count for PIP.
-pub const PACKET_COUNT: u8 = 9;
 
 // packet ID definitions.
 mod packet {
@@ -111,9 +114,9 @@ mod packet {
 mod timeout {
 	use std::time::Duration;
 
-	pub const HANDSHAKE: Duration =  Duration::from_millis(2500);
-	pub const ACKNOWLEDGE_UPDATE: Duration = Duration::from_millis(5000);
-	pub const BASE: u64 = 1500; // base timeout for packet.
+	pub const HANDSHAKE: Duration =  Duration::from_millis(4_000);
+	pub const ACKNOWLEDGE_UPDATE: Duration = Duration::from_millis(5_000);
+	pub const BASE: u64 = 2500; // base timeout for packet.
 
 	// timeouts per request within packet.
 	pub const HEADERS: u64 = 250; // per header?
@@ -688,7 +691,7 @@ impl LightProtocol {
 			Err(e) => { punish(*peer, io, e); return }
 		};
 
-		if PROTOCOL_VERSIONS.iter().find(|x| **x == proto_version).is_none() {
+		if PROTOCOL_VERSIONS.iter().find(|x| x.0 == proto_version).is_none() {
 			punish(*peer, io, Error::UnsupportedProtocolVersion(proto_version));
 			return;
 		}

ethcore/sync/src/api.rs

@@ -33,7 +33,7 @@ use chain::{ChainSync, SyncStatus as EthSyncStatus};
 use std::net::{SocketAddr, AddrParseError};
 use std::str::FromStr;
 use parking_lot::RwLock;
-use chain::{ETH_PACKET_COUNT, SNAPSHOT_SYNC_PACKET_COUNT, ETH_PROTOCOL_VERSION_63, ETH_PROTOCOL_VERSION_62,
+use chain::{ETH_PROTOCOL_VERSION_63, ETH_PROTOCOL_VERSION_62,
 	PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_2, PAR_PROTOCOL_VERSION_3};
 use light::client::AsLightClient;
 use light::Provider;
@@ -202,18 +202,15 @@ pub struct AttachedProtocol {
 	pub handler: Arc<NetworkProtocolHandler + Send + Sync>,
 	/// 3-character ID for the protocol.
 	pub protocol_id: ProtocolId,
-	/// Packet count.
-	pub packet_count: u8,
-	/// Supported versions.
-	pub versions: &'static [u8],
+	/// Supported versions and their packet counts.
+	pub versions: &'static [(u8, u8)],
 }
 
 impl AttachedProtocol {
 	fn register(&self, network: &NetworkService) {
 		let res = network.register_protocol(
 			self.handler.clone(),
 			self.protocol_id,
-			self.packet_count,
 			self.versions
 		);
 
@@ -459,15 +456,15 @@ impl ChainNotify for EthSync {
 			Err(err) => warn!("Error starting network: {}", err),
 			_ => {},
 		}
-		self.network.register_protocol(self.eth_handler.clone(), self.subprotocol_name, ETH_PACKET_COUNT, &[ETH_PROTOCOL_VERSION_62, ETH_PROTOCOL_VERSION_63])
+		self.network.register_protocol(self.eth_handler.clone(), self.subprotocol_name, &[ETH_PROTOCOL_VERSION_62, ETH_PROTOCOL_VERSION_63])
 			.unwrap_or_else(|e| warn!("Error registering ethereum protocol: {:?}", e));
 		// register the warp sync subprotocol
-		self.network.register_protocol(self.eth_handler.clone(), WARP_SYNC_PROTOCOL_ID, SNAPSHOT_SYNC_PACKET_COUNT, &[PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_2, PAR_PROTOCOL_VERSION_3])
+		self.network.register_protocol(self.eth_handler.clone(), WARP_SYNC_PROTOCOL_ID, &[PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_2, PAR_PROTOCOL_VERSION_3])
 			.unwrap_or_else(|e| warn!("Error registering snapshot sync protocol: {:?}", e));
 
 		// register the light protocol.
 		if let Some(light_proto) = self.light_proto.as_ref().map(|x| x.clone()) {
-			self.network.register_protocol(light_proto, self.light_subprotocol_name, ::light::net::PACKET_COUNT, ::light::net::PROTOCOL_VERSIONS)
+			self.network.register_protocol(light_proto, self.light_subprotocol_name, ::light::net::PROTOCOL_VERSIONS)
 				.unwrap_or_else(|e| warn!("Error registering light client protocol: {:?}", e));
 		}
 
@@ -827,7 +824,7 @@ impl ManageNetwork for LightSync {
 
 		let light_proto = self.proto.clone();
 
-		self.network.register_protocol(light_proto, self.subprotocol_name, ::light::net::PACKET_COUNT, ::light::net::PROTOCOL_VERSIONS)
+		self.network.register_protocol(light_proto, self.subprotocol_name, ::light::net::PROTOCOL_VERSIONS)
 			.unwrap_or_else(|e| warn!("Error registering light client protocol: {:?}", e));
 
 		for proto in &self.attached_protos { proto.register(&self.network) }

ethcore/sync/src/chain/handler.rs

@@ -45,7 +45,6 @@ use super::{
 	MAX_NEW_BLOCK_AGE,
 	MAX_NEW_HASHES,
 	PAR_PROTOCOL_VERSION_1,
-	PAR_PROTOCOL_VERSION_2,
 	PAR_PROTOCOL_VERSION_3,
 	BLOCK_BODIES_PACKET,
 	BLOCK_HEADERS_PACKET,
@@ -641,8 +640,11 @@ impl SyncHandler {
 			trace!(target: "sync", "Peer {} network id mismatch (ours: {}, theirs: {})", peer_id, sync.network_id, peer.network_id);
 			return Ok(());
 		}
-		if (warp_protocol && peer.protocol_version != PAR_PROTOCOL_VERSION_1 && peer.protocol_version != PAR_PROTOCOL_VERSION_2 && peer.protocol_version != PAR_PROTOCOL_VERSION_3)
-			|| (!warp_protocol && peer.protocol_version != ETH_PROTOCOL_VERSION_63 && peer.protocol_version != ETH_PROTOCOL_VERSION_62) {
+
+		if false
+			|| (warp_protocol && (peer.protocol_version < PAR_PROTOCOL_VERSION_1.0 || peer.protocol_version > PAR_PROTOCOL_VERSION_3.0))
+			|| (!warp_protocol && (peer.protocol_version < ETH_PROTOCOL_VERSION_62.0 || peer.protocol_version > ETH_PROTOCOL_VERSION_63.0))
+		{
 			io.disable_peer(peer_id);
 			trace!(target: "sync", "Peer {} unsupported eth protocol ({})", peer_id, peer.protocol_version);
 			return Ok(());

ethcore/sync/src/chain/mod.rs

@@ -127,15 +127,15 @@ known_heap_size!(0, PeerInfo);
 pub type PacketDecodeError = DecoderError;
 
 /// 63 version of Ethereum protocol.
-pub const ETH_PROTOCOL_VERSION_63: u8 = 63;
+pub const ETH_PROTOCOL_VERSION_63: (u8, u8) = (63, 0x11);
 /// 62 version of Ethereum protocol.
-pub const ETH_PROTOCOL_VERSION_62: u8 = 62;
-/// 1 version of Parity protocol.
-pub const PAR_PROTOCOL_VERSION_1: u8 = 1;
+pub const ETH_PROTOCOL_VERSION_62: (u8, u8) = (62, 0x11);
+/// 1 version of Parity protocol and the packet count.
+pub const PAR_PROTOCOL_VERSION_1: (u8, u8) = (1, 0x15);
 /// 2 version of Parity protocol (consensus messages added).
-pub const PAR_PROTOCOL_VERSION_2: u8 = 2;
+pub const PAR_PROTOCOL_VERSION_2: (u8, u8) = (2, 0x16);
 /// 3 version of Parity protocol (private transactions messages added).
-pub const PAR_PROTOCOL_VERSION_3: u8 = 3;
+pub const PAR_PROTOCOL_VERSION_3: (u8, u8) = (3, 0x18);
 
 pub const MAX_BODIES_TO_SEND: usize = 256;
 pub const MAX_HEADERS_TO_SEND: usize = 512;
@@ -169,8 +169,6 @@ pub const NODE_DATA_PACKET: u8 = 0x0e;
 pub const GET_RECEIPTS_PACKET: u8 = 0x0f;
 pub const RECEIPTS_PACKET: u8 = 0x10;
 
-pub const ETH_PACKET_COUNT: u8 = 0x11;
-
 pub const GET_SNAPSHOT_MANIFEST_PACKET: u8 = 0x11;
 pub const SNAPSHOT_MANIFEST_PACKET: u8 = 0x12;
 pub const GET_SNAPSHOT_DATA_PACKET: u8 = 0x13;
@@ -179,8 +177,6 @@ pub const CONSENSUS_DATA_PACKET: u8 = 0x15;
 const PRIVATE_TRANSACTION_PACKET: u8 = 0x16;
 const SIGNED_PRIVATE_TRANSACTION_PACKET: u8 = 0x17;
 
-pub const SNAPSHOT_SYNC_PACKET_COUNT: u8 = 0x18;
-
 const MAX_SNAPSHOT_CHUNKS_DOWNLOAD_AHEAD: usize = 3;
 
 const WAIT_PEERS_TIMEOUT: Duration = Duration::from_secs(5);
@@ -453,7 +449,7 @@ impl ChainSync {
 		let last_imported_number = self.new_blocks.last_imported_block_number();
 		SyncStatus {
 			state: self.state.clone(),
-			protocol_version: ETH_PROTOCOL_VERSION_63,
+			protocol_version: ETH_PROTOCOL_VERSION_63.0,
 			network_id: self.network_id,
 			start_block_number: self.starting_block,
 			last_imported_block_number: Some(last_imported_number),
@@ -855,7 +851,7 @@ impl ChainSync {
 	fn send_status(&mut self, io: &mut SyncIo, peer: PeerId) -> Result<(), network::Error> {
 		let warp_protocol_version = io.protocol_version(&WARP_SYNC_PROTOCOL_ID, peer);
 		let warp_protocol = warp_protocol_version != 0;
-		let protocol = if warp_protocol { warp_protocol_version } else { ETH_PROTOCOL_VERSION_63 };
+		let protocol = if warp_protocol { warp_protocol_version } else { ETH_PROTOCOL_VERSION_63.0 };
 		trace!(target: "sync", "Sending status to {}, protocol version {}", peer, protocol);
 		let mut packet = RlpStream::new_list(if warp_protocol { 7 } else { 5 });
 		let chain = io.chain().chain_info();
@@ -1019,11 +1015,11 @@ impl ChainSync {
 	}
 
 	fn get_consensus_peers(&self) -> Vec<PeerId> {
-		self.peers.iter().filter_map(|(id, p)| if p.protocol_version >= PAR_PROTOCOL_VERSION_2 { Some(*id) } else { None }).collect()
+		self.peers.iter().filter_map(|(id, p)| if p.protocol_version >= PAR_PROTOCOL_VERSION_2.0 { Some(*id) } else { None }).collect()
 	}
 
 	fn get_private_transaction_peers(&self) -> Vec<PeerId> {
-		self.peers.iter().filter_map(|(id, p)| if p.protocol_version >= PAR_PROTOCOL_VERSION_3 { Some(*id) } else { None }).collect()
+		self.peers.iter().filter_map(|(id, p)| if p.protocol_version >= PAR_PROTOCOL_VERSION_3.0 { Some(*id) } else { None }).collect()
 	}
 
 	/// Maintain other peers. Send out any new blocks and transactions

ethcore/sync/src/chain/requester.rs

@@ -28,7 +28,7 @@ use super::{
 	BlockSet,
 	ChainSync,
 	PeerAsking,
-	ETH_PACKET_COUNT,
+	ETH_PROTOCOL_VERSION_63,
 	GET_BLOCK_BODIES_PACKET,
 	GET_BLOCK_HEADERS_PACKET,
 	GET_RECEIPTS_PACKET,
@@ -140,7 +140,8 @@ impl SyncRequester {
 			}
 			peer.asking = asking;
 			peer.ask_time = Instant::now();
-			let result = if packet_id >= ETH_PACKET_COUNT {
+			// TODO [ToDr] This seems quite fragile. Be careful when protocol is updated.
+			let result = if packet_id >= ETH_PROTOCOL_VERSION_63.1 {
 				io.send_protocol(WARP_SYNC_PROTOCOL_ID, peer_id, packet_id, packet)
 			} else {
 				io.send(peer_id, packet_id, packet)

ethcore/sync/src/light_sync/response.rs

@@ -16,7 +16,7 @@
 
 //! Helpers for decoding and verifying responses for headers.
 
-use ethcore::{self, encoded, header::Header};
+use ethcore::{encoded, header::Header};
 use ethereum_types::H256;
 use light::request::{HashOrNumber, CompleteHeadersRequest as HeadersRequest};
 use rlp::DecoderError;

ethcore/sync/src/tests/helpers.rs

@@ -134,11 +134,11 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
 	}
 
 	fn eth_protocol_version(&self, _peer: PeerId) -> u8 {
-		ETH_PROTOCOL_VERSION_63
+		ETH_PROTOCOL_VERSION_63.0
 	}
 
 	fn protocol_version(&self, protocol: &ProtocolId, peer_id: PeerId) -> u8 {
-		if protocol == &WARP_SYNC_PROTOCOL_ID { PAR_PROTOCOL_VERSION_3 } else { self.eth_protocol_version(peer_id) }
+		if protocol == &WARP_SYNC_PROTOCOL_ID { PAR_PROTOCOL_VERSION_3.0 } else { self.eth_protocol_version(peer_id) }
 	}
 
 	fn chain_overlay(&self) -> &RwLock<HashMap<BlockNumber, Bytes>> {

parity/whisper.rs

@@ -89,15 +89,13 @@ pub fn setup(target_pool_size: usize, protos: &mut Vec<AttachedProtocol>)
 
 	protos.push(AttachedProtocol {
 		handler: net.clone() as Arc<_>,
-		packet_count: whisper_net::PACKET_COUNT,
 		versions: whisper_net::SUPPORTED_VERSIONS,
 		protocol_id: whisper_net::PROTOCOL_ID,
 	});
 
 	// parity-only extensions to whisper.
 	protos.push(AttachedProtocol {
 		handler: Arc::new(whisper_net::ParityExtensions),
-		packet_count: whisper_net::PACKET_COUNT,
 		versions: whisper_net::SUPPORTED_VERSIONS,
 		protocol_id: whisper_net::PARITY_PROTOCOL_ID,
 	});

util/network-devp2p/src/host.rs

@@ -79,7 +79,9 @@ const NODE_TABLE_TIMEOUT: Duration = Duration::from_secs(300);
 #[derive(Debug, PartialEq, Eq)]
 /// Protocol info
 pub struct CapabilityInfo {
+	/// Protocol ID
 	pub protocol: ProtocolId,
+	/// Protocol version
 	pub version: u8,
 	/// Total number of packet IDs this protocol support.
 	pub packet_count: u8,
@@ -161,6 +163,7 @@ impl<'s> NetworkContextTrait for NetworkContext<'s> {
 	}
 
 	fn disconnect_peer(&self, peer: PeerId) {
+		trace!(target: "network", "Peer disconnect requested: {}", peer);
 		self.io.message(NetworkIoMessage::Disconnect(peer))
 			.unwrap_or_else(|e| warn!("Error sending network IO message: {:?}", e));
 	}
@@ -722,11 +725,24 @@ impl Host {
 							// Outgoing connections are allowed as long as their count is <= min_peers
 							// Incoming connections are allowed to take all of the max_peers reserve, or at most half of the slots.
 							let max_ingress = max(max_peers - min_peers, min_peers / 2);
-							if reserved_only ||
-								(s.info.originated && egress_count > min_peers) ||
-								(!s.info.originated && ingress_count > max_ingress) {
-								// only proceed if the connecting peer is reserved.
-								if !self.reserved_nodes.read().contains(&id) {
+							if reserved_only
+								|| (s.info.originated && egress_count > min_peers)
+								|| (!s.info.originated && ingress_count > max_ingress)
+							{
+								// We didn't start the connection, but the node is known to us
+								// So eventually we will attempt to connect to it as well.
+								let is_incoming_but_known = !s.info.originated && self.nodes.read().contains(&id);
+
+								if is_incoming_but_known {
+									warn!(target: "network", "Allowing incoming connection from a known node.");
+								}
+								// only proceed if the connecting peer is reserved or is known
+								if !is_incoming_but_known && !self.reserved_nodes.read().contains(&id) {
+									trace!(
+										target: "network",
+										"Rejected {} session: TooManyPeers",
+										if s.info.originated { "outbound" } else { "inbound" }
+									);
 									s.disconnect(io, DisconnectReason::TooManyPeers);
 									kill = true;
 									break;
@@ -990,7 +1006,6 @@ impl IoHandler<NetworkIoMessage> for Host {
 				ref handler,
 				ref protocol,
 				ref versions,
-				ref packet_count,
 			} => {
 				let h = handler.clone();
 				let reserved = self.reserved_nodes.read();
@@ -1000,8 +1015,12 @@ impl IoHandler<NetworkIoMessage> for Host {
 				);
 				self.handlers.write().insert(*protocol, h);
 				let mut info = self.info.write();
-				for v in versions {
-					info.capabilities.push(CapabilityInfo { protocol: *protocol, version: *v, packet_count: *packet_count });
+				for &(version, packet_count) in versions {
+					info.capabilities.push(CapabilityInfo {
+						protocol: *protocol,
+						version,
+						packet_count,
+					});
 				}
 			},
 			NetworkIoMessage::AddTimer {

util/network-devp2p/src/lib.rs

@@ -49,7 +49,7 @@
 //! fn main () {
 //! 	let mut service = NetworkService::new(NetworkConfiguration::new_local(), None).expect("Error creating network service");
 //! 	service.start().expect("Error starting service");
-//! 	service.register_protocol(Arc::new(MyHandler), *b"myp", 1, &[1u8]);
+//! 	service.register_protocol(Arc::new(MyHandler), *b"myp", &[(1u8, 1u8)]);
 //!
 //! 	// Wait for quit condition
 //! 	// ...

util/network-devp2p/src/service.rs

@@ -67,12 +67,17 @@ impl NetworkService {
 	}
 
 	/// Regiter a new protocol handler with the event loop.
-	pub fn register_protocol(&self, handler: Arc<NetworkProtocolHandler + Send + Sync>, protocol: ProtocolId, packet_count: u8, versions: &[u8]) -> Result<(), Error> {
+	pub fn register_protocol(
+		&self,
+		handler: Arc<NetworkProtocolHandler + Send + Sync>,
+		protocol: ProtocolId,
+		// version id + packet count
+		versions: &[(u8, u8)]
+	) -> Result<(), Error> {
 		self.io_service.send_message(NetworkIoMessage::AddHandler {
-			handler: handler,
-			protocol: protocol,
+			handler,
+			protocol,
 			versions: versions.to_vec(),
-			packet_count: packet_count,
 		})?;
 		Ok(())
 	}

util/network-devp2p/tests/tests.rs

@@ -52,7 +52,7 @@ impl TestProtocol {
 	/// Creates and register protocol with the network service
 	pub fn register(service: &mut NetworkService, drop_session: bool) -> Arc<TestProtocol> {
 		let handler = Arc::new(TestProtocol::new(drop_session));
-		service.register_protocol(handler.clone(), *b"tst", 1, &[42u8, 43u8]).expect("Error registering test protocol handler");
+		service.register_protocol(handler.clone(), *b"tst", &[(42u8, 1u8), (43u8, 1u8)]).expect("Error registering test protocol handler");
 		handler
 	}
 
@@ -104,7 +104,7 @@ impl NetworkProtocolHandler for TestProtocol {
 fn net_service() {
 	let service = NetworkService::new(NetworkConfiguration::new_local(), None).expect("Error creating network service");
 	service.start().unwrap();
-	service.register_protocol(Arc::new(TestProtocol::new(false)), *b"myp", 1, &[1u8]).unwrap();
+	service.register_protocol(Arc::new(TestProtocol::new(false)), *b"myp", &[(1u8, 1u8)]).unwrap();
 }
 
 #[test]

util/network/src/lib.rs

@@ -64,10 +64,8 @@ pub enum NetworkIoMessage {
 		handler: Arc<NetworkProtocolHandler + Sync>,
 		/// Protocol Id.
 		protocol: ProtocolId,
-		/// Supported protocol versions.
-		versions: Vec<u8>,
-		/// Number of packet IDs reserved by the protocol.
-		packet_count: u8,
+		/// Supported protocol versions and number of packet IDs reserved by the protocol (packet count).
+		versions: Vec<(u8, u8)>,
 	},
 	/// Register a new protocol timer
 	AddTimer {

whisper/cli/src/main.rs

@@ -218,10 +218,10 @@ fn execute<S, I>(command: I) -> Result<(), Error> where I: IntoIterator<Item=S>,
 	network.start()?;
 
 	// Attach whisper protocol to the network service
-	network.register_protocol(whisper_network_handler.clone(), whisper::net::PROTOCOL_ID, whisper::net::PACKET_COUNT,
+	network.register_protocol(whisper_network_handler.clone(), whisper::net::PROTOCOL_ID,
 							  whisper::net::SUPPORTED_VERSIONS)?;
 	network.register_protocol(Arc::new(whisper::net::ParityExtensions), whisper::net::PARITY_PROTOCOL_ID,
-							  whisper::net::PACKET_COUNT, whisper::net::SUPPORTED_VERSIONS)?;
+							  whisper::net::SUPPORTED_VERSIONS)?;
 
 	// Request handler
 	let mut io = MetaIoHandler::default();