Kernels v2 (variable size) (#3034)

* wip

* exhaustive match

* write with fixed v1 strategy when writing for hashing

* local protocol version is 2

* cleanup "size" tests that exercise v1 vs v2 vs default protocol versions

* add proto version to Connected! log msg

* cleanup docs

* negotiate protocol version min(local, peer) when doing hand/shake

api/src/handlers/pool_api.rs

@@ -75,8 +75,8 @@ impl PoolPushHandler {
 						.map_err(|e| ErrorKind::RequestError(format!("Bad request: {}", e)).into())
 				})
 				.and_then(move |tx_bin| {
-					// TODO - pass protocol version in via the api call?
-					let version = ProtocolVersion::local();
+					// All wallet api interaction explicitly uses protocol version 1 for now.
+					let version = ProtocolVersion(1);
 
 					ser::deserialize(&mut &tx_bin[..], version)
 						.map_err(|e| ErrorKind::RequestError(format!("Bad request: {}", e)).into())

core/src/core/transaction.rs

@@ -20,8 +20,8 @@ use crate::core::{committed, Committed};
 use crate::keychain::{self, BlindingFactor};
 use crate::libtx::secp_ser;
 use crate::ser::{
-	self, read_multi, FixedLength, PMMRable, Readable, Reader, VerifySortedAndUnique, Writeable,
-	Writer,
+	self, read_multi, FixedLength, PMMRable, ProtocolVersion, Readable, Reader,
+	VerifySortedAndUnique, Writeable, Writer,
 };
 use crate::util;
 use crate::util::secp;
@@ -92,22 +92,33 @@ impl KernelFeatures {
 		let msg = secp::Message::from_slice(&hash.as_bytes())?;
 		Ok(msg)
 	}
-}
 
-impl Writeable for KernelFeatures {
-	/// Still only supporting protocol version v1 serialization.
-	/// Always include fee, defaulting to 0, and lock_height, defaulting to 0, for all feature variants.
-	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
+	/// Write tx kernel features out in v1 protocol format.
+	/// Always include the fee and lock_height, writing 0 value if unused.
+	fn write_v1<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
+		let (fee, lock_height) = match self {
+			KernelFeatures::Plain { fee } => (*fee, 0),
+			KernelFeatures::Coinbase => (0, 0),
+			KernelFeatures::HeightLocked { fee, lock_height } => (*fee, *lock_height),
+		};
+		writer.write_u8(self.as_u8())?;
+		writer.write_u64(fee)?;
+		writer.write_u64(lock_height)?;
+		Ok(())
+	}
+
+	/// Write tx kernel features out in v2 protocol format.
+	/// These are variable sized based on feature variant.
+	/// Only write fee out for feature variants that support it.
+	/// Only write lock_height out for feature variants that support it.
+	fn write_v2<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
 		match self {
 			KernelFeatures::Plain { fee } => {
 				writer.write_u8(self.as_u8())?;
 				writer.write_u64(*fee)?;
-				writer.write_u64(0)?;
 			}
 			KernelFeatures::Coinbase => {
 				writer.write_u8(self.as_u8())?;
-				writer.write_u64(0)?;
-				writer.write_u64(0)?;
 			}
 			KernelFeatures::HeightLocked { fee, lock_height } => {
 				writer.write_u8(self.as_u8())?;
@@ -117,33 +128,48 @@ impl Writeable for KernelFeatures {
 		}
 		Ok(())
 	}
-}
 
-impl Readable for KernelFeatures {
-	/// Still only supporting protocol version v1 serialization.
-	/// Always read both fee and lock_height, regardless of feature variant.
-	/// These will be 0 values if not applicable, but bytes must still be read and verified.
-	fn read(reader: &mut dyn Reader) -> Result<KernelFeatures, ser::Error> {
-		let features = match reader.read_u8()? {
+	// Always read feature byte, 8 bytes for fee and 8 bytes for lock height.
+	// Fee and lock height may be unused for some kernel variants but we need
+	// to read these bytes and verify they are 0 if unused.
+	fn read_v1(reader: &mut dyn Reader) -> Result<KernelFeatures, ser::Error> {
+		let feature_byte = reader.read_u8()?;
+		let fee = reader.read_u64()?;
+		let lock_height = reader.read_u64()?;
+
+		let features = match feature_byte {
 			KernelFeatures::PLAIN_U8 => {
-				let fee = reader.read_u64()?;
-				let lock_height = reader.read_u64()?;
 				if lock_height != 0 {
 					return Err(ser::Error::CorruptedData);
 				}
 				KernelFeatures::Plain { fee }
 			}
 			KernelFeatures::COINBASE_U8 => {
-				let fee = reader.read_u64()?;
 				if fee != 0 {
 					return Err(ser::Error::CorruptedData);
 				}
-				let lock_height = reader.read_u64()?;
 				if lock_height != 0 {
 					return Err(ser::Error::CorruptedData);
 				}
 				KernelFeatures::Coinbase
 			}
+			KernelFeatures::HEIGHT_LOCKED_U8 => KernelFeatures::HeightLocked { fee, lock_height },
+			_ => {
+				return Err(ser::Error::CorruptedData);
+			}
+		};
+		Ok(features)
+	}
+
+	// V2 kernels only expect bytes specific to each variant.
+	// Coinbase kernels have no associated fee and we do not serialize a fee for these.
+	fn read_v2(reader: &mut dyn Reader) -> Result<KernelFeatures, ser::Error> {
+		let features = match reader.read_u8()? {
+			KernelFeatures::PLAIN_U8 => {
+				let fee = reader.read_u64()?;
+				KernelFeatures::Plain { fee }
+			}
+			KernelFeatures::COINBASE_U8 => KernelFeatures::Coinbase,
 			KernelFeatures::HEIGHT_LOCKED_U8 => {
 				let fee = reader.read_u64()?;
 				let lock_height = reader.read_u64()?;
@@ -157,6 +183,32 @@ impl Readable for KernelFeatures {
 	}
 }
 
+impl Writeable for KernelFeatures {
+	/// Protocol version may increment rapidly for other unrelated changes.
+	/// So we match on ranges here and not specific version values.
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
+		// Care must be exercised when writing for hashing purposes.
+		// All kernels are hashed using original v1 serialization strategy.
+		if writer.serialization_mode() == ser::SerializationMode::Hash {
+			return self.write_v1(writer);
+		}
+
+		match writer.protocol_version().value() {
+			0..=1 => self.write_v1(writer),
+			2..=ProtocolVersion::MAX => self.write_v2(writer),
+		}
+	}
+}
+
+impl Readable for KernelFeatures {
+	fn read(reader: &mut dyn Reader) -> Result<KernelFeatures, ser::Error> {
+		match reader.protocol_version().value() {
+			0..=1 => KernelFeatures::read_v1(reader),
+			2..=ProtocolVersion::MAX => KernelFeatures::read_v2(reader),
+		}
+	}
+}
+
 /// Errors thrown by Transaction validation
 #[derive(Clone, Eq, Debug, PartialEq, Serialize, Deserialize)]
 pub enum Error {
@@ -275,12 +327,6 @@ impl ::std::hash::Hash for TxKernel {
 
 impl Writeable for TxKernel {
 	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
-		// We have access to the protocol version here.
-		// This may be a protocol version based on a peer connection
-		// or the version used locally for db storage.
-		// We can handle version specific serialization here.
-		let _version = writer.protocol_version();
-
 		self.features.write(writer)?;
 		self.excess.write(writer)?;
 		self.excess_sig.write(writer)?;
@@ -290,12 +336,6 @@ impl Writeable for TxKernel {
 
 impl Readable for TxKernel {
 	fn read(reader: &mut dyn Reader) -> Result<TxKernel, ser::Error> {
-		// We have access to the protocol version here.
-		// This may be a protocol version based on a peer connection
-		// or the version used locally for db storage.
-		// We can handle version specific deserialization here.
-		let _version = reader.protocol_version();
-
 		Ok(TxKernel {
 			features: KernelFeatures::read(reader)?,
 			excess: Commitment::read(reader)?,

core/src/global.rs

@@ -38,8 +38,8 @@ use crate::util::RwLock;
 /// We negotiate compatible versions with each peer via Hand/Shake.
 /// Note: We also use a specific (possible different) protocol version
 /// for both the backend database and MMR data files.
-/// This one is p2p layer specific.
-pub const PROTOCOL_VERSION: u32 = 1;
+/// This defines the p2p layer protocol version for this node.
+pub const PROTOCOL_VERSION: u32 = 2;
 
 /// Automated testing edge_bits
 pub const AUTOMATED_TESTING_MIN_EDGE_BITS: u8 = 10;

core/src/ser.rs

@@ -289,7 +289,18 @@ where
 pub struct ProtocolVersion(pub u32);
 
 impl ProtocolVersion {
+	/// The max protocol version supported.
+	pub const MAX: u32 = std::u32::MAX;
+
+	/// Protocol version as u32 to allow for convenient exhaustive matching on values.
+	pub fn value(&self) -> u32 {
+		self.0
+	}
+
 	/// Our default "local" protocol version.
+	/// This protocol version is provided to peers as part of the Hand/Shake
+	/// negotiation in the p2p layer. Connected peers will negotiate a suitable
+	/// protocol version for serialization/deserialization of p2p messages.
 	pub fn local() -> ProtocolVersion {
 		ProtocolVersion(PROTOCOL_VERSION)
 	}

core/tests/block.rs

@@ -269,8 +269,7 @@ fn empty_block_serialized_size() {
 	let b = new_block(vec![], &keychain, &builder, &prev, &key_id);
 	let mut vec = Vec::new();
 	ser::serialize_default(&mut vec, &b).expect("serialization failed");
-	let target_len = 1_112;
-	assert_eq!(vec.len(), target_len);
+	assert_eq!(vec.len(), 1_096);
 }
 
 #[test]
@@ -284,8 +283,7 @@ fn block_single_tx_serialized_size() {
 	let b = new_block(vec![&tx1], &keychain, &builder, &prev, &key_id);
 	let mut vec = Vec::new();
 	ser::serialize_default(&mut vec, &b).expect("serialization failed");
-	let target_len = 2_694;
-	assert_eq!(vec.len(), target_len);
+	assert_eq!(vec.len(), 2_670);
 }
 
 #[test]
@@ -299,8 +297,7 @@ fn empty_compact_block_serialized_size() {
 	let cb: CompactBlock = b.into();
 	let mut vec = Vec::new();
 	ser::serialize_default(&mut vec, &cb).expect("serialization failed");
-	let target_len = 1_120;
-	assert_eq!(vec.len(), target_len);
+	assert_eq!(vec.len(), 1_104);
 }
 
 #[test]
@@ -315,8 +312,7 @@ fn compact_block_single_tx_serialized_size() {
 	let cb: CompactBlock = b.into();
 	let mut vec = Vec::new();
 	ser::serialize_default(&mut vec, &cb).expect("serialization failed");
-	let target_len = 1_126;
-	assert_eq!(vec.len(), target_len);
+	assert_eq!(vec.len(), 1_110);
 }
 
 #[test]
@@ -333,10 +329,27 @@ fn block_10_tx_serialized_size() {
 	let prev = BlockHeader::default();
 	let key_id = ExtKeychain::derive_key_id(1, 1, 0, 0, 0);
 	let b = new_block(txs.iter().collect(), &keychain, &builder, &prev, &key_id);
-	let mut vec = Vec::new();
-	ser::serialize_default(&mut vec, &b).expect("serialization failed");
-	let target_len = 16_932;
-	assert_eq!(vec.len(), target_len,);
+
+	// Default protocol version.
+	{
+		let mut vec = Vec::new();
+		ser::serialize_default(&mut vec, &b).expect("serialization failed");
+		assert_eq!(vec.len(), 16_836);
+	}
+
+	// Explicit protocol version 1
+	{
+		let mut vec = Vec::new();
+		ser::serialize(&mut vec, ser::ProtocolVersion(1), &b).expect("serialization failed");
+		assert_eq!(vec.len(), 16_932);
+	}
+
+	// Explicit protocol version 2
+	{
+		let mut vec = Vec::new();
+		ser::serialize(&mut vec, ser::ProtocolVersion(2), &b).expect("serialization failed");
+		assert_eq!(vec.len(), 16_836);
+	}
 }
 
 #[test]
@@ -356,8 +369,7 @@ fn compact_block_10_tx_serialized_size() {
 	let cb: CompactBlock = b.into();
 	let mut vec = Vec::new();
 	ser::serialize_default(&mut vec, &cb).expect("serialization failed");
-	let target_len = 1_180;
-	assert_eq!(vec.len(), target_len,);
+	assert_eq!(vec.len(), 1_164);
 }
 
 #[test]

core/tests/core.rs

@@ -38,10 +38,27 @@ use std::sync::Arc;
 #[test]
 fn simple_tx_ser() {
 	let tx = tx2i1o();
-	let mut vec = Vec::new();
-	ser::serialize_default(&mut vec, &tx).expect("serialization failed");
-	let target_len = 955;
-	assert_eq!(vec.len(), target_len,);
+
+	// Default protocol version.
+	{
+		let mut vec = Vec::new();
+		ser::serialize_default(&mut vec, &tx).expect("serialization failed");
+		assert_eq!(vec.len(), 947);
+	}
+
+	// Explicit protocol version 1.
+	{
+		let mut vec = Vec::new();
+		ser::serialize(&mut vec, ser::ProtocolVersion(1), &tx).expect("serialization failed");
+		assert_eq!(vec.len(), 955);
+	}
+
+	// Explicit protocol version 2.
+	{
+		let mut vec = Vec::new();
+		ser::serialize(&mut vec, ser::ProtocolVersion(2), &tx).expect("serialization failed");
+		assert_eq!(vec.len(), 947);
+	}
 }
 
 #[test]

core/tests/vec_backend.rs

@@ -109,7 +109,7 @@ impl<T: PMMRable> Backend<T> for VecBackend<T> {
 		unimplemented!()
 	}
 
-	fn leaf_pos_iter(&self) -> Box<Iterator<Item = u64> + '_> {
+	fn leaf_pos_iter(&self) -> Box<dyn Iterator<Item = u64> + '_> {
 		unimplemented!()
 	}
 

p2p/src/handshake.rs

@@ -46,6 +46,7 @@ pub struct Handshake {
 	/// ok).
 	genesis: Hash,
 	config: P2PConfig,
+	protocol_version: ProtocolVersion,
 }
 
 impl Handshake {
@@ -56,9 +57,22 @@ impl Handshake {
 			addrs: Arc::new(RwLock::new(VecDeque::with_capacity(ADDRS_CAP))),
 			genesis,
 			config,
+			protocol_version: ProtocolVersion::local(),
 		}
 	}
 
+	/// Select a protocol version here that we know is supported by both us and the remote peer.
+	///
+	/// Current strategy is to simply use `min(local, remote)`.
+	///
+	/// We can enforce "minimum" protocol version here in the future
+	/// by raising an error and forcing the connection to close.
+	///
+	fn negotiate_protocol_version(&self, other: ProtocolVersion) -> Result<ProtocolVersion, Error> {
+		let version = std::cmp::min(self.protocol_version, other);
+		Ok(version)
+	}
+
 	pub fn initiate(
 		&self,
 		capabilities: Capabilities,
@@ -73,11 +87,8 @@ impl Handshake {
 			Err(e) => return Err(Error::Connection(e)),
 		};
 
-		// Using our default "local" protocol version.
-		let version = ProtocolVersion::local();
-
 		let hand = Hand {
-			version,
+			version: self.protocol_version,
 			capabilities,
 			nonce,
 			genesis: self.genesis,
@@ -88,22 +99,23 @@ impl Handshake {
 		};
 
 		// write and read the handshake response
-		write_message(conn, hand, Type::Hand, version)?;
+		write_message(conn, hand, Type::Hand, self.protocol_version)?;
 
-		// Note: We have to read the Shake message *before* we know which protocol
-		// version our peer supports (it is in the shake message itself).
-		let shake: Shake = read_message(conn, version, Type::Shake)?;
+		let shake: Shake = read_message(conn, self.protocol_version, Type::Shake)?;
 		if shake.genesis != self.genesis {
 			return Err(Error::GenesisMismatch {
 				us: self.genesis,
 				peer: shake.genesis,
 			});
 		}
+
+		let negotiated_version = self.negotiate_protocol_version(shake.version)?;
+
 		let peer_info = PeerInfo {
 			capabilities: shake.capabilities,
 			user_agent: shake.user_agent,
 			addr: peer_addr,
-			version: shake.version,
+			version: negotiated_version,
 			live_info: Arc::new(RwLock::new(PeerLiveInfo::new(shake.total_difficulty))),
 			direction: Direction::Outbound,
 		};
@@ -115,11 +127,12 @@ impl Handshake {
 		}
 
 		debug!(
-			"Connected! Cumulative {} offered from {:?} {:?} {:?}",
+			"Connected! Cumulative {} offered from {:?}, {:?}, {:?}, {:?}",
 			shake.total_difficulty.to_num(),
 			peer_info.addr,
+			peer_info.version,
 			peer_info.user_agent,
-			peer_info.capabilities
+			peer_info.capabilities,
 		);
 		// when more than one protocol version is supported, choosing should go here
 		Ok(peer_info)
@@ -131,11 +144,7 @@ impl Handshake {
 		total_difficulty: Difficulty,
 		conn: &mut TcpStream,
 	) -> Result<PeerInfo, Error> {
-		// Note: We read the Hand message *before* we know which protocol version
-		// is supported by our peer (in the Hand message).
-		let version = ProtocolVersion::local();
-
-		let hand: Hand = read_message(conn, version, Type::Hand)?;
+		let hand: Hand = read_message(conn, self.protocol_version, Type::Hand)?;
 
 		// all the reasons we could refuse this connection for
 		if hand.genesis != self.genesis {
@@ -158,12 +167,14 @@ impl Handshake {
 			}
 		}
 
+		let negotiated_version = self.negotiate_protocol_version(hand.version)?;
+
 		// all good, keep peer info
 		let peer_info = PeerInfo {
 			capabilities: hand.capabilities,
 			user_agent: hand.user_agent,
 			addr: resolve_peer_addr(hand.sender_addr, &conn),
-			version: hand.version,
+			version: negotiated_version,
 			live_info: Arc::new(RwLock::new(PeerLiveInfo::new(hand.total_difficulty))),
 			direction: Direction::Inbound,
 		};
@@ -178,14 +189,14 @@ impl Handshake {
 
 		// send our reply with our info
 		let shake = Shake {
-			version,
+			version: self.protocol_version,
 			capabilities: capab,
 			genesis: self.genesis,
 			total_difficulty: total_difficulty,
 			user_agent: USER_AGENT.to_string(),
 		};
 
-		write_message(conn, shake, Type::Shake, version)?;
+		write_message(conn, shake, Type::Shake, negotiated_version)?;
 		trace!("Success handshake with {}.", peer_info.addr);
 
 		Ok(peer_info)