Implement eth/65 (EIP-2464)

ethcore/client-traits/src/lib.rs

@@ -285,7 +285,10 @@ pub trait BlockChainClient:
 	fn list_storage(&self, id: BlockId, account: &Address, after: Option<&H256>, count: Option<u64>) -> Option<Vec<H256>>;
 
 	/// Get transaction with given hash.
-	fn transaction(&self, id: TransactionId) -> Option<LocalizedTransaction>;
+	fn block_transaction(&self, id: TransactionId) -> Option<LocalizedTransaction>;
+
+	/// Get pool transaction with a given hash.
+	fn queued_transaction(&self, hash: H256) -> Option<Arc<VerifiedTransaction>>;
 
 	/// Get uncle with given id.
 	fn uncle(&self, id: UncleId) -> Option<encoded::Header>;

ethcore/src/client/client.rs

@@ -1898,10 +1898,14 @@ impl BlockChainClient for Client {
 		Some(keys)
 	}
 
-	fn transaction(&self, id: TransactionId) -> Option<LocalizedTransaction> {
+	fn block_transaction(&self, id: TransactionId) -> Option<LocalizedTransaction> {
 		self.transaction_address(id).and_then(|address| self.chain.read().transaction(&address))
 	}
 
+	fn queued_transaction(&self, hash: H256) -> Option<Arc<VerifiedTransaction>> {
+		self.importer.miner.transaction(&hash)
+	}
+
 	fn uncle(&self, id: UncleId) -> Option<encoded::Header> {
 		let index = id.position;
 		self.block_body(id.block).and_then(|body| body.view().uncle_rlp_at(index))

ethcore/src/test_helpers/test_client.rs

@@ -737,9 +737,12 @@ impl BlockChainClient for TestBlockChainClient {
 	fn list_storage(&self, _id: BlockId, _account: &Address, _after: Option<&H256>, _count: Option<u64>) -> Option<Vec<H256>> {
 		None
 	}
-	fn transaction(&self, _id: TransactionId) -> Option<LocalizedTransaction> {
+	fn block_transaction(&self, _id: TransactionId) -> Option<LocalizedTransaction> {
 		None	// Simple default.
 	}
+	fn queued_transaction(&self, _hash: H256) -> Option<Arc<VerifiedTransaction>> {
+		None
+	}
 
 	fn uncle(&self, _id: UncleId) -> Option<encoded::Header> {
 		None	// Simple default.

ethcore/sync/Cargo.toml

@@ -36,6 +36,7 @@ rlp = "0.4.5"
 snapshot = { path = "../snapshot" }
 trace-time = "0.1"
 triehash-ethereum = { version = "0.2", path = "../../util/triehash-ethereum" }
+transaction-pool = "2"
 
 [dev-dependencies]
 env_logger = "0.5"

ethcore/sync/src/chain/handler.rs

@@ -26,14 +26,12 @@ use crate::{
 		sync_packet::{
 			PacketInfo,
 			SyncPacket::{
-				self, BlockBodiesPacket, BlockHeadersPacket, NewBlockHashesPacket, NewBlockPacket,
-				PrivateStatePacket, PrivateTransactionPacket, ReceiptsPacket, SignedPrivateTransactionPacket,
-				SnapshotDataPacket, SnapshotManifestPacket, StatusPacket,
+				self, *,
 			}
 		},
 		BlockSet, ChainSync, ForkConfirmation, PacketProcessError, PeerAsking, PeerInfo, SyncRequester,
-		SyncState, ETH_PROTOCOL_VERSION_63, ETH_PROTOCOL_VERSION_64, MAX_NEW_BLOCK_AGE, MAX_NEW_HASHES,
-		PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_3, PAR_PROTOCOL_VERSION_4,
+		SyncState, ETH_PROTOCOL_VERSION_63, ETH_PROTOCOL_VERSION_64, ETH_PROTOCOL_VERSION_65,
+		MAX_NEW_BLOCK_AGE, MAX_NEW_HASHES, PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_3, PAR_PROTOCOL_VERSION_4,
 	}
 };
 
@@ -53,6 +51,7 @@ use common_types::{
 	verification::Unverified,
 	snapshot::{ManifestData, RestorationStatus},
 };
+use transaction_pool::VerifiedTransaction;
 
 
 /// The Chain Sync Handler: handles responses from peers
@@ -70,6 +69,8 @@ impl SyncHandler {
 				ReceiptsPacket => SyncHandler::on_peer_block_receipts(sync, io, peer, &rlp),
 				NewBlockPacket => SyncHandler::on_peer_new_block(sync, io, peer, &rlp),
 				NewBlockHashesPacket => SyncHandler::on_peer_new_hashes(sync, io, peer, &rlp),
+				NewPooledTransactionHashesPacket => SyncHandler::on_peer_new_pooled_transactions(sync, io, peer, &rlp),
+				PooledTransactionsPacket => SyncHandler::on_peer_pooled_transactions(sync, io, peer, &rlp),
 				SnapshotManifestPacket => SyncHandler::on_snapshot_manifest(sync, io, peer, &rlp),
 				SnapshotDataPacket => SyncHandler::on_snapshot_data(sync, io, peer, &rlp),
 				PrivateTransactionPacket => SyncHandler::on_private_transaction(sync, io, peer, &rlp),
@@ -595,9 +596,11 @@ impl SyncHandler {
 			difficulty,
 			latest_hash,
 			genesis,
+			unsent_pooled_hashes: if eth_protocol_version >= ETH_PROTOCOL_VERSION_65.0 { Some(io.chain().transactions_to_propagate().into_iter().map(|tx| *tx.hash()).collect()) } else { None },
 			asking: PeerAsking::Nothing,
 			asking_blocks: Vec::new(),
 			asking_hash: None,
+			asking_pooled_transactions: if eth_protocol_version >= ETH_PROTOCOL_VERSION_65.0 { Some(Vec::new()) } else { None },
 			asking_private_state: None,
 			ask_time: Instant::now(),
 			last_sent_transactions: Default::default(),
@@ -656,7 +659,7 @@ impl SyncHandler {
 
 		if false
 			|| (warp_protocol && (peer.protocol_version < PAR_PROTOCOL_VERSION_1.0 || peer.protocol_version > PAR_PROTOCOL_VERSION_4.0))
-			|| (!warp_protocol && (peer.protocol_version < ETH_PROTOCOL_VERSION_63.0 || peer.protocol_version > ETH_PROTOCOL_VERSION_64.0))
+			|| (!warp_protocol && (peer.protocol_version < ETH_PROTOCOL_VERSION_63.0 || peer.protocol_version > ETH_PROTOCOL_VERSION_65.0))
 		{
 			trace!(target: "sync", "Peer {} unsupported eth protocol ({})", peer_id, peer.protocol_version);
 			return Err(DownloaderImportError::Invalid);
@@ -703,6 +706,61 @@ impl SyncHandler {
 		Ok(())
 	}
 
+	/// Called when peer sends us a set of new pooled transactions
+	pub fn on_peer_new_pooled_transactions(sync: &mut ChainSync, io: &mut dyn SyncIo, peer_id: PeerId, tx_rlp: &Rlp) -> Result<(), DownloaderImportError> {
+		for item in tx_rlp {
+			let hash = item.as_val::<H256>().map_err(|_| DownloaderImportError::Invalid)?;
+
+			if io.chain().queued_transaction(hash).is_none() {
+				let unfetched = sync.unfetched_pooled_transactions.entry(hash).or_insert_with(|| super::UnfetchedTransaction {
+					announcer: peer_id,
+					next_fetch: Instant::now(),
+					tries: 0,
+				});
+
+				// Only reset the budget if we hear from multiple sources
+				if unfetched.announcer != peer_id {
+					unfetched.next_fetch = Instant::now();
+					unfetched.tries = 0;
+				}
+			}
+		}
+
+		Ok(())
+	}
+
+	/// Called when peer sends us a list of pooled transactions
+	pub fn on_peer_pooled_transactions(sync: &ChainSync, io: &mut dyn SyncIo, peer_id: PeerId, tx_rlp: &Rlp) -> Result<(), DownloaderImportError> {
+		let peer = match sync.peers.get(&peer_id).filter(|p| p.can_sync()) {
+			Some(peer) => peer,
+			None => {
+				trace!(target: "sync", "{} Ignoring transactions from unconfirmed/unknown peer", peer_id);
+				return Ok(());
+			}
+		};
+
+		// TODO: actually check against asked hashes
+		let item_count = tx_rlp.item_count()?;
+		if let Some(p) = &peer.asking_pooled_transactions {
+			if item_count > p.len() {
+				trace!(target: "sync", "{} Peer sent us more transactions than was supposed to", peer_id);
+				return Err(DownloaderImportError::Invalid);
+			}
+		} else {
+			trace!(target: "sync", "{} Peer sent us pooled transactions but does not declare support for them", peer_id);
+			return Err(DownloaderImportError::Invalid);
+		}
+		trace!(target: "sync", "{:02} -> PooledTransactions ({} entries)", peer_id, item_count);
+		let mut transactions = Vec::with_capacity(item_count);
+		for i in 0 .. item_count {
+			let rlp = tx_rlp.at(i)?;
+			let tx = rlp.as_raw().to_vec();
+			transactions.push(tx);
+		}
+		io.chain().queue_transactions(transactions, peer_id);
+		Ok(())
+	}
+
 	/// Called when peer sends us signed private transaction packet
 	fn on_signed_private_transaction(sync: &mut ChainSync, _io: &mut dyn SyncIo, peer_id: PeerId, r: &Rlp) -> Result<(), DownloaderImportError> {
 		if !sync.peers.get(&peer_id).map_or(false, |p| p.can_sync()) {

ethcore/sync/src/chain/mod.rs

@@ -166,6 +166,8 @@ impl From<DecoderError> for PacketProcessError {
 	}
 }
 
+/// Version 65 of the Ethereum protocol and number of packet IDs reserved by the protocol (packet count).
+pub const ETH_PROTOCOL_VERSION_65: (u8, u8) = (65, 0x11);
 /// Version 64 of the Ethereum protocol and number of packet IDs reserved by the protocol (packet count).
 pub const ETH_PROTOCOL_VERSION_64: (u8, u8) = (64, 0x11);
 /// Version 63 of the Ethereum protocol and number of packet IDs reserved by the protocol (packet count).
@@ -217,6 +219,7 @@ const STATUS_TIMEOUT: Duration = Duration::from_secs(10);
 const HEADERS_TIMEOUT: Duration = Duration::from_secs(15);
 const BODIES_TIMEOUT: Duration = Duration::from_secs(20);
 const RECEIPTS_TIMEOUT: Duration = Duration::from_secs(10);
+const POOLED_TRANSACTIONS_TIMEOUT: Duration = Duration::from_secs(10);
 const FORK_HEADER_TIMEOUT: Duration = Duration::from_secs(3);
 /// Max time to wait for the Snapshot Manifest packet to arrive from a peer after it's being asked.
 const SNAPSHOT_MANIFEST_TIMEOUT: Duration = Duration::from_secs(5);
@@ -318,6 +321,7 @@ pub enum PeerAsking {
 	BlockHeaders,
 	BlockBodies,
 	BlockReceipts,
+	PooledTransactions,
 	SnapshotManifest,
 	SnapshotData,
 	PrivateState,
@@ -352,6 +356,8 @@ pub struct PeerInfo {
 	network_id: u64,
 	/// Peer best block hash
 	latest_hash: H256,
+	/// Unpropagated tx pool hashes
+	unsent_pooled_hashes: Option<H256FastSet>,
 	/// Peer total difficulty if known
 	difficulty: Option<U256>,
 	/// Type of data currently being requested by us from a peer.
@@ -360,6 +366,8 @@ pub struct PeerInfo {
 	asking_blocks: Vec<H256>,
 	/// Holds requested header hash if currently requesting block header by hash
 	asking_hash: Option<H256>,
+	/// Holds requested transaction IDs
+	asking_pooled_transactions: Option<Vec<H256>>,
 	/// Holds requested private state hash
 	asking_private_state: Option<H256>,
 	/// Holds requested snapshot chunk hash if any.
@@ -669,6 +677,13 @@ enum PeerState {
 	SameBlock
 }
 
+#[derive(Clone, MallocSizeOf)]
+struct UnfetchedTransaction {
+	announcer: PeerId,
+	next_fetch: Instant,
+	tries: usize,
+}
+
 /// Blockchain sync handler.
 /// See module documentation for more details.
 #[derive(MallocSizeOf)]
@@ -708,6 +723,8 @@ pub struct ChainSync {
 	sync_start_time: Option<Instant>,
 	/// Transactions propagation statistics
 	transactions_stats: TransactionsStats,
+	/// Transactions whose hash has been announced, but that we have not fetched
+	unfetched_pooled_transactions: H256FastMap<UnfetchedTransaction>,
 	/// Enable ancient block downloading
 	download_old_blocks: bool,
 	/// Shared private tx service.
@@ -751,6 +768,7 @@ impl ChainSync {
 			snapshot: Snapshot::new(),
 			sync_start_time: None,
 			transactions_stats: TransactionsStats::default(),
+			unfetched_pooled_transactions: Default::default(),
 			private_tx_handler,
 			warp_sync: config.warp_sync,
 			status_sinks: Vec::new()
@@ -764,7 +782,7 @@ impl ChainSync {
 		let last_imported_number = self.new_blocks.last_imported_block_number();
 		SyncStatus {
 			state: self.state.clone(),
-			protocol_version: ETH_PROTOCOL_VERSION_64.0,
+			protocol_version: ETH_PROTOCOL_VERSION_65.0,
 			network_id: self.network_id,
 			start_block_number: self.starting_block,
 			last_imported_block_number: Some(last_imported_number),
@@ -798,8 +816,17 @@ impl ChainSync {
 
 	/// Updates the set of transactions recently sent to this peer to avoid spamming.
 	pub fn transactions_received(&mut self, txs: &[UnverifiedTransaction], peer_id: PeerId) {
-		if let Some(peer_info) = self.peers.get_mut(&peer_id) {
-			peer_info.last_sent_transactions.extend(txs.iter().map(|tx| tx.hash()));
+		for (id, peer) in &mut self.peers {
+			let hashes = txs.iter().map(|tx| tx.hash());
+			if *id == peer_id {
+				peer.last_sent_transactions.extend(hashes);
+			} else if let Some(s) = &mut peer.unsent_pooled_hashes {
+				s.extend(hashes);
+			}
+		}
+
+		for tx in txs {
+			self.unfetched_pooled_transactions.remove(&tx.hash());
 		}
 	}
 
@@ -1149,6 +1176,48 @@ impl ChainSync {
 						}
 					}
 
+					// get the peer to give us at least some of announced but unfetched transactions
+					if !self.unfetched_pooled_transactions.is_empty() {
+						if let Some(s) = &mut self.peers.get_mut(&peer_id).expect("this is always an active peer; qed").asking_pooled_transactions {
+							let now = Instant::now();
+
+							let mut new_asking_pooled_transactions = s.iter().copied().collect::<HashSet<_>>();
+							let mut remaining_unfetched_pooled_transactions = self.unfetched_pooled_transactions.clone();
+							for (hash, mut item) in self.unfetched_pooled_transactions.drain() {
+								if new_asking_pooled_transactions.len() >= 256 {
+									// can't request any more transactions
+									break;
+								}
+
+								// if enough time has passed since last attempt...
+								if item.next_fetch < now {
+									// ...queue this hash for requesting
+									new_asking_pooled_transactions.insert(hash);
+									item.tries += 1;
+
+									// if we just started asking for it, queue it to be asked later on again
+									if item.tries < 5 {
+										item.next_fetch = now + (POOLED_TRANSACTIONS_TIMEOUT / 2);
+										remaining_unfetched_pooled_transactions.insert(hash, item);
+									} else {
+										// ...otherwise we assume this transaction does not exist and remove its hash from request queue
+										remaining_unfetched_pooled_transactions.remove(&hash);
+									}
+								}
+							}
+
+							let new_asking_pooled_transactions = new_asking_pooled_transactions.into_iter().collect::<Vec<_>>();
+							SyncRequester::request_pooled_transactions(self, io, peer_id, &new_asking_pooled_transactions);
+
+							self.peers.get_mut(&peer_id).expect("this is always an active peer; qed").asking_pooled_transactions = Some(new_asking_pooled_transactions);
+							self.unfetched_pooled_transactions = remaining_unfetched_pooled_transactions;
+
+							return;
+						} else {
+							trace!(target: "sync", "Skipping transaction fetch for peer {} as they don't support eth/65", peer_id);
+						}
+					}
+
 					// Only ask for old blocks if the peer has an equal or higher difficulty
 					let equal_or_higher_difficulty = peer_difficulty.map_or(true, |pd| pd >= syncing_difficulty);
 
@@ -1340,6 +1409,7 @@ impl ChainSync {
 				PeerAsking::BlockHeaders => elapsed > HEADERS_TIMEOUT,
 				PeerAsking::BlockBodies => elapsed > BODIES_TIMEOUT,
 				PeerAsking::BlockReceipts => elapsed > RECEIPTS_TIMEOUT,
+				PeerAsking::PooledTransactions => elapsed > POOLED_TRANSACTIONS_TIMEOUT,
 				PeerAsking::Nothing => false,
 				PeerAsking::ForkHeader => elapsed > FORK_HEADER_TIMEOUT,
 				PeerAsking::SnapshotManifest => elapsed > SNAPSHOT_MANIFEST_TIMEOUT,
@@ -1668,10 +1738,12 @@ pub mod tests {
 				genesis: H256::zero(),
 				network_id: 0,
 				latest_hash: peer_latest_hash,
+				unsent_pooled_hashes: Some(Default::default()),
 				difficulty: None,
 				asking: PeerAsking::Nothing,
 				asking_blocks: Vec::new(),
 				asking_hash: None,
+				asking_pooled_transactions: Some(Vec::new()),
 				asking_private_state: None,
 				ask_time: Instant::now(),
 				last_sent_transactions: Default::default(),