Framing support for realtime network messages (#1972)

- Sizes addes for each message types & subtypes
- New count field added to message header extensions (4 bit, max size 15) for confirm_req/confirm_ack by hash
- Node ID handshake header flags functions moved to nano::message_header instead from nano::node_id_handshake
- Tests expanded to check count extensions field

nano/core_test/block.cpp

@@ -312,78 +312,6 @@ TEST (block, publish_req_serialization)
 	ASSERT_EQ (*req.block, *req2.block);
 }
 
-TEST (block, confirm_req_serialization)
-{
-	nano::keypair key1;
-	nano::keypair key2;
-	auto block (std::make_shared<nano::send_block> (0, key2.pub, 200, nano::keypair ().prv, 2, 3));
-	nano::confirm_req req (block);
-	std::vector<uint8_t> bytes;
-	{
-		nano::vectorstream stream (bytes);
-		req.serialize (stream);
-	}
-	auto error (false);
-	nano::bufferstream stream2 (bytes.data (), bytes.size ());
-	nano::message_header header (error, stream2);
-	nano::confirm_req req2 (error, stream2, header);
-	ASSERT_FALSE (error);
-	ASSERT_EQ (req, req2);
-	ASSERT_EQ (*req.block, *req2.block);
-}
-
-TEST (block, confirm_req_hash_serialization)
-{
-	nano::keypair key1;
-	nano::keypair key2;
-	nano::send_block block (1, key2.pub, 200, nano::keypair ().prv, 2, 3);
-	nano::confirm_req req (block.hash (), block.root ());
-	std::vector<uint8_t> bytes;
-	{
-		nano::vectorstream stream (bytes);
-		req.serialize (stream);
-	}
-	auto error (false);
-	nano::bufferstream stream2 (bytes.data (), bytes.size ());
-	nano::message_header header (error, stream2);
-	nano::confirm_req req2 (error, stream2, header);
-	ASSERT_FALSE (error);
-	ASSERT_EQ (req, req2);
-	ASSERT_EQ (req.roots_hashes, req2.roots_hashes);
-}
-
-TEST (block, confirm_req_hash_batch_serialization)
-{
-	nano::keypair key;
-	nano::keypair representative;
-	std::vector<std::pair<nano::block_hash, nano::block_hash>> roots_hashes;
-	nano::state_block open (key.pub, 0, representative.pub, 2, 4, key.prv, key.pub, 5);
-	roots_hashes.push_back (std::make_pair (open.hash (), open.root ()));
-	for (auto i (roots_hashes.size ()); i < 7; i++)
-	{
-		nano::keypair key1;
-		nano::keypair previous;
-		nano::state_block block (key1.pub, previous.pub, representative.pub, 2, 4, key1.prv, key1.pub, 5);
-		roots_hashes.push_back (std::make_pair (block.hash (), block.root ()));
-	}
-	roots_hashes.push_back (std::make_pair (open.hash (), open.root ()));
-	nano::confirm_req req (roots_hashes);
-	std::vector<uint8_t> bytes;
-	{
-		nano::vectorstream stream (bytes);
-		req.serialize (stream);
-	}
-	auto error (false);
-	nano::bufferstream stream2 (bytes.data (), bytes.size ());
-	nano::message_header header (error, stream2);
-	nano::confirm_req req2 (error, stream2, header);
-	ASSERT_FALSE (error);
-	ASSERT_EQ (req, req2);
-	ASSERT_EQ (req.roots_hashes, req2.roots_hashes);
-	ASSERT_EQ (req.roots_hashes, roots_hashes);
-	ASSERT_EQ (req2.roots_hashes, roots_hashes);
-}
-
 TEST (state_block, serialization)
 {
 	nano::keypair key1;

nano/core_test/message.cpp

@@ -82,4 +82,116 @@ TEST (message, confirm_ack_serialization)
 	nano::confirm_ack con2 (error, stream2, header);
 	ASSERT_FALSE (error);
 	ASSERT_EQ (con1, con2);
+	ASSERT_EQ (header.block_type (), nano::block_type::send);
+}
+
+TEST (message, confirm_ack_hash_serialization)
+{
+	std::vector<nano::block_hash> hashes;
+	for (auto i (hashes.size ()); i < 12; i++)
+	{
+		nano::keypair key1;
+		nano::keypair previous;
+		nano::state_block block (key1.pub, previous.pub, key1.pub, 2, 4, key1.prv, key1.pub, 5);
+		hashes.push_back (block.hash ());
+	}
+	nano::keypair representative1;
+	auto vote (std::make_shared<nano::vote> (representative1.pub, representative1.prv, 0, hashes));
+	nano::confirm_ack con1 (vote);
+	std::vector<uint8_t> bytes;
+	{
+		nano::vectorstream stream1 (bytes);
+		con1.serialize (stream1);
+	}
+	nano::bufferstream stream2 (bytes.data (), bytes.size ());
+	bool error (false);
+	nano::message_header header (error, stream2);
+	nano::confirm_ack con2 (error, stream2, header);
+	ASSERT_FALSE (error);
+	ASSERT_EQ (con1, con2);
+	std::vector<nano::block_hash> vote_blocks;
+	for (auto block : con2.vote->blocks)
+	{
+		vote_blocks.push_back (boost::get<nano::block_hash> (block));
+	}
+	ASSERT_EQ (hashes, vote_blocks);
+	// Check overflow with 12 hashes
+	ASSERT_EQ (header.count_get (), hashes.size ());
+	ASSERT_EQ (header.block_type (), nano::block_type::not_a_block);
+}
+
+TEST (message, confirm_req_serialization)
+{
+	nano::keypair key1;
+	nano::keypair key2;
+	auto block (std::make_shared<nano::send_block> (0, key2.pub, 200, nano::keypair ().prv, 2, 3));
+	nano::confirm_req req (block);
+	std::vector<uint8_t> bytes;
+	{
+		nano::vectorstream stream (bytes);
+		req.serialize (stream);
+	}
+	auto error (false);
+	nano::bufferstream stream2 (bytes.data (), bytes.size ());
+	nano::message_header header (error, stream2);
+	nano::confirm_req req2 (error, stream2, header);
+	ASSERT_FALSE (error);
+	ASSERT_EQ (req, req2);
+	ASSERT_EQ (*req.block, *req2.block);
+}
+
+TEST (message, confirm_req_hash_serialization)
+{
+	nano::keypair key1;
+	nano::keypair key2;
+	nano::send_block block (1, key2.pub, 200, nano::keypair ().prv, 2, 3);
+	nano::confirm_req req (block.hash (), block.root ());
+	std::vector<uint8_t> bytes;
+	{
+		nano::vectorstream stream (bytes);
+		req.serialize (stream);
+	}
+	auto error (false);
+	nano::bufferstream stream2 (bytes.data (), bytes.size ());
+	nano::message_header header (error, stream2);
+	nano::confirm_req req2 (error, stream2, header);
+	ASSERT_FALSE (error);
+	ASSERT_EQ (req, req2);
+	ASSERT_EQ (req.roots_hashes, req2.roots_hashes);
+	ASSERT_EQ (header.block_type (), nano::block_type::not_a_block);
+	ASSERT_EQ (header.count_get (), req.roots_hashes.size ());
+}
+
+TEST (message, confirm_req_hash_batch_serialization)
+{
+	nano::keypair key;
+	nano::keypair representative;
+	std::vector<std::pair<nano::block_hash, nano::block_hash>> roots_hashes;
+	nano::state_block open (key.pub, 0, representative.pub, 2, 4, key.prv, key.pub, 5);
+	roots_hashes.push_back (std::make_pair (open.hash (), open.root ()));
+	for (auto i (roots_hashes.size ()); i < 7; i++)
+	{
+		nano::keypair key1;
+		nano::keypair previous;
+		nano::state_block block (key1.pub, previous.pub, representative.pub, 2, 4, key1.prv, key1.pub, 5);
+		roots_hashes.push_back (std::make_pair (block.hash (), block.root ()));
+	}
+	roots_hashes.push_back (std::make_pair (open.hash (), open.root ()));
+	nano::confirm_req req (roots_hashes);
+	std::vector<uint8_t> bytes;
+	{
+		nano::vectorstream stream (bytes);
+		req.serialize (stream);
+	}
+	auto error (false);
+	nano::bufferstream stream2 (bytes.data (), bytes.size ());
+	nano::message_header header (error, stream2);
+	nano::confirm_req req2 (error, stream2, header);
+	ASSERT_FALSE (error);
+	ASSERT_EQ (req, req2);
+	ASSERT_EQ (req.roots_hashes, req2.roots_hashes);
+	ASSERT_EQ (req.roots_hashes, roots_hashes);
+	ASSERT_EQ (req2.roots_hashes, roots_hashes);
+	ASSERT_EQ (header.block_type (), nano::block_type::not_a_block);
+	ASSERT_EQ (header.count_get (), req.roots_hashes.size ());
 }

nano/node/bootstrap.cpp

@@ -1966,6 +1966,38 @@ void nano::bootstrap_server::receive_header_action (boost::system::error_code co
 					});
 					break;
 				}
+				case nano::message_type::publish:
+				{
+					auto this_l (shared_from_this ());
+					socket->async_read (receive_buffer, header.payload_length_bytes (), [this_l, header](boost::system::error_code const & ec, size_t size_a) {
+						this_l->receive_publish_action (ec, size_a, header);
+					});
+					break;
+				}
+				case nano::message_type::confirm_ack:
+				{
+					auto this_l (shared_from_this ());
+					socket->async_read (receive_buffer, header.payload_length_bytes (), [this_l, header](boost::system::error_code const & ec, size_t size_a) {
+						this_l->receive_confirm_ack_action (ec, size_a, header);
+					});
+					break;
+				}
+				case nano::message_type::confirm_req:
+				{
+					auto this_l (shared_from_this ());
+					socket->async_read (receive_buffer, header.payload_length_bytes (), [this_l, header](boost::system::error_code const & ec, size_t size_a) {
+						this_l->receive_confirm_req_action (ec, size_a, header);
+					});
+					break;
+				}
+				case nano::message_type::node_id_handshake:
+				{
+					auto this_l (shared_from_this ());
+					socket->async_read (receive_buffer, header.payload_length_bytes (), [this_l, header](boost::system::error_code const & ec, size_t size_a) {
+						this_l->receive_node_id_handshake_action (ec, size_a, header);
+					});
+					break;
+				}
 				default:
 				{
 					if (node->config.logging.network_logging ())
@@ -2025,34 +2057,12 @@ void nano::bootstrap_server::receive_bulk_pull_account_action (boost::system::er
 	}
 }
 
-void nano::bootstrap_server::receive_keepalive_action (boost::system::error_code const & ec, size_t size_a, nano::message_header const & header_a)
-{
-	if (!ec)
-	{
-		auto error (false);
-		nano::bufferstream stream (receive_buffer->data (), header_a.payload_length_bytes ());
-		std::unique_ptr<nano::keepalive> request (new nano::keepalive (error, stream, header_a));
-		if (!error)
-		{
-			add_request (std::unique_ptr<nano::message> (request.release ()));
-			receive ();
-		}
-	}
-	else
-	{
-		if (node->config.logging.network_keepalive_logging ())
-		{
-			node->logger.try_log (boost::str (boost::format ("Error receiving keepalive from: %1%") % ec.message ()));
-		}
-	}
-}
-
 void nano::bootstrap_server::receive_frontier_req_action (boost::system::error_code const & ec, size_t size_a, nano::message_header const & header_a)
 {
 	if (!ec)
 	{
 		auto error (false);
-		nano::bufferstream stream (receive_buffer->data (), header_a.payload_length_bytes ());
+		nano::bufferstream stream (receive_buffer->data (), size_a);
 		std::unique_ptr<nano::frontier_req> request (new nano::frontier_req (error, stream, header_a));
 		if (!error)
 		{
@@ -2073,6 +2083,107 @@ void nano::bootstrap_server::receive_frontier_req_action (boost::system::error_c
 	}
 }
 
+void nano::bootstrap_server::receive_keepalive_action (boost::system::error_code const & ec, size_t size_a, nano::message_header const & header_a)
+{
+	if (!ec)
+	{
+		auto error (false);
+		nano::bufferstream stream (receive_buffer->data (), size_a);
+		std::unique_ptr<nano::keepalive> request (new nano::keepalive (error, stream, header_a));
+		if (!error)
+		{
+			add_request (std::unique_ptr<nano::message> (request.release ()));
+			receive ();
+		}
+	}
+	else
+	{
+		if (node->config.logging.network_keepalive_logging ())
+		{
+			node->logger.try_log (boost::str (boost::format ("Error receiving keepalive: %1%") % ec.message ()));
+		}
+	}
+}
+
+void nano::bootstrap_server::receive_publish_action (boost::system::error_code const & ec, size_t size_a, nano::message_header const & header_a)
+{
+	if (!ec)
+	{
+		auto error (false);
+		nano::bufferstream stream (receive_buffer->data (), size_a);
+		std::unique_ptr<nano::publish> request (new nano::publish (error, stream, header_a));
+		if (!error)
+		{
+			add_request (std::unique_ptr<nano::message> (request.release ()));
+			receive ();
+		}
+	}
+	else
+	{
+		if (node->config.logging.network_message_logging ())
+		{
+			node->logger.try_log (boost::str (boost::format ("Error receiving publish: %1%") % ec.message ()));
+		}
+	}
+}
+
+void nano::bootstrap_server::receive_confirm_req_action (boost::system::error_code const & ec, size_t size_a, nano::message_header const & header_a)
+{
+	if (!ec)
+	{
+		auto error (false);
+		nano::bufferstream stream (receive_buffer->data (), size_a);
+		std::unique_ptr<nano::confirm_req> request (new nano::confirm_req (error, stream, header_a));
+		if (!error)
+		{
+			add_request (std::unique_ptr<nano::message> (request.release ()));
+			receive ();
+		}
+	}
+	else if (node->config.logging.network_message_logging ())
+	{
+		node->logger.try_log (boost::str (boost::format ("Error receiving confirm_req: %1%") % ec.message ()));
+	}
+}
+
+void nano::bootstrap_server::receive_confirm_ack_action (boost::system::error_code const & ec, size_t size_a, nano::message_header const & header_a)
+{
+	if (!ec)
+	{
+		auto error (false);
+		nano::bufferstream stream (receive_buffer->data (), size_a);
+		std::unique_ptr<nano::confirm_ack> request (new nano::confirm_ack (error, stream, header_a));
+		if (!error)
+		{
+			add_request (std::unique_ptr<nano::message> (request.release ()));
+			receive ();
+		}
+	}
+	else if (node->config.logging.network_message_logging ())
+	{
+		node->logger.try_log (boost::str (boost::format ("Error receiving confirm_ack: %1%") % ec.message ()));
+	}
+}
+
+void nano::bootstrap_server::receive_node_id_handshake_action (boost::system::error_code const & ec, size_t size_a, nano::message_header const & header_a)
+{
+	if (!ec)
+	{
+		auto error (false);
+		nano::bufferstream stream (receive_buffer->data (), size_a);
+		std::unique_ptr<nano::node_id_handshake> request (new nano::node_id_handshake (error, stream, header_a));
+		if (!error)
+		{
+			add_request (std::unique_ptr<nano::message> (request.release ()));
+			receive ();
+		}
+	}
+	else if (node->config.logging.network_node_id_handshake_logging ())
+	{
+		node->logger.try_log (boost::str (boost::format ("Error receiving node_id_handshake: %1%") % ec.message ()));
+	}
+}
+
 void nano::bootstrap_server::add_request (std::unique_ptr<nano::message> message_a)
 {
 	std::lock_guard<std::mutex> lock (mutex);

nano/node/bootstrap.hpp

@@ -297,6 +297,10 @@ public:
 	void receive_bulk_pull_account_action (boost::system::error_code const &, size_t, nano::message_header const &);
 	void receive_frontier_req_action (boost::system::error_code const &, size_t, nano::message_header const &);
 	void receive_keepalive_action (boost::system::error_code const &, size_t, nano::message_header const &);
+	void receive_publish_action (boost::system::error_code const &, size_t, nano::message_header const &);
+	void receive_confirm_req_action (boost::system::error_code const &, size_t, nano::message_header const &);
+	void receive_confirm_ack_action (boost::system::error_code const &, size_t, nano::message_header const &);
+	void receive_node_id_handshake_action (boost::system::error_code const &, size_t, nano::message_header const &);
 	void add_request (std::unique_ptr<nano::message>);
 	void finish_request ();
 	void run_next ();

nano/node/common.cpp

@@ -7,6 +7,7 @@
 #include <boost/endian/conversion.hpp>
 
 std::bitset<16> constexpr nano::message_header::block_type_mask;
+std::bitset<16> constexpr nano::message_header::count_mask;
 
 nano::message_header::message_header (nano::message_type type_a) :
 version_max (nano::protocol_version),
@@ -85,6 +86,25 @@ void nano::message_header::block_type_set (nano::block_type type_a)
 	extensions |= std::bitset<16> (static_cast<unsigned long long> (type_a) << 8);
 }
 
+uint8_t nano::message_header::count_get () const
+{
+	return ((extensions & count_mask) >> 12).to_ullong ();
+}
+
+void nano::message_header::count_set (uint8_t count_a)
+{
+	assert (count_a < 16);
+	extensions &= ~count_mask;
+	extensions |= std::bitset<16> (static_cast<unsigned long long> (count_a) << 12);
+}
+
+void nano::message_header::flag_set (uint8_t flag_a)
+{
+	// Flags from 8 are block_type & count
+	assert (flag_a < 8);
+	extensions.set (flag_a, true);
+}
+
 bool nano::message_header::bulk_pull_is_count_present () const
 {
 	auto result (false);
@@ -95,7 +115,32 @@ bool nano::message_header::bulk_pull_is_count_present () const
 			result = true;
 		}
 	}
+	return result;
+}
 
+bool nano::message_header::node_id_handshake_is_query () const
+{
+	auto result (false);
+	if (type == nano::message_type::node_id_handshake)
+	{
+		if (extensions.test (node_id_handshake_query_flag))
+		{
+			result = true;
+		}
+	}
+	return result;
+}
+
+bool nano::message_header::node_id_handshake_is_response () const
+{
+	auto result (false);
+	if (type == nano::message_type::node_id_handshake)
+	{
+		if (extensions.test (node_id_handshake_response_flag))
+		{
+			result = true;
+		}
+	}
 	return result;
 }
 
@@ -124,8 +169,22 @@ size_t nano::message_header::payload_length_bytes () const
 		{
 			return nano::keepalive::size;
 		}
-		// Add realtime network messages once they get framing support; currently the
+		case nano::message_type::publish:
-		// realtime messages all fit in a datagram from which they're deserialized.
+		{
+			return nano::block::size (block_type ());
+		}
+		case nano::message_type::confirm_ack:
+		{
+			return nano::confirm_ack::size (block_type (), count_get ());
+		}
+		case nano::message_type::confirm_req:
+		{
+			return nano::confirm_req::size (block_type (), count_get ());
+		}
+		case nano::message_type::node_id_handshake:
+		{
+			return nano::node_id_handshake::size (*this);
+		}
 		default:
 		{
 			assert (false);
@@ -498,6 +557,8 @@ roots_hashes (roots_hashes_a)
 {
 	// not_a_block (1) block type for hashes + roots request
 	header.block_type_set (nano::block_type::not_a_block);
+	assert (roots_hashes.size () < 16);
+	header.count_set (roots_hashes.size ());
 }
 
 nano::confirm_req::confirm_req (nano::block_hash const & hash_a, nano::block_hash const & root_a) :
@@ -507,6 +568,7 @@ roots_hashes (std::vector<std::pair<nano::block_hash, nano::block_hash>> (1, std
 	assert (!roots_hashes.empty ());
 	// not_a_block (1) block type for hashes + roots request
 	header.block_type_set (nano::block_type::not_a_block);
+	header.count_set (roots_hashes.size ());
 }
 
 void nano::confirm_req::visit (nano::message_visitor & visitor_a) const
@@ -606,6 +668,20 @@ std::string nano::confirm_req::roots_string () const
 	return result;
 }
 
+size_t nano::confirm_req::size (nano::block_type type_a, size_t count)
+{
+	size_t result (0);
+	if (type_a != nano::block_type::invalid && type_a != nano::block_type::not_a_block)
+	{
+		result = nano::block::size (type_a);
+	}
+	else if (type_a == nano::block_type::not_a_block)
+	{
+		result = sizeof (uint8_t) + count * (sizeof (nano::uint256_union) + sizeof (nano::block_hash));
+	}
+	return result;
+}
+
 nano::confirm_ack::confirm_ack (bool & error_a, nano::stream & stream_a, nano::message_header const & header_a, nano::vote_uniquer * uniquer_a) :
 message (header_a),
 vote (std::make_shared<nano::vote> (error_a, stream_a, header.block_type ()))
@@ -625,6 +701,8 @@ vote (vote_a)
 	if (first_vote_block.which ())
 	{
 		header.block_type_set (nano::block_type::not_a_block);
+		assert (vote_a->blocks.size () < 16);
+		header.count_set (vote_a->blocks.size ());
 	}
 	else
 	{
@@ -650,6 +728,20 @@ void nano::confirm_ack::visit (nano::message_visitor & visitor_a) const
 	visitor_a.confirm_ack (*this);
 }
 
+size_t nano::confirm_ack::size (nano::block_type type_a, size_t count)
+{
+	size_t result (sizeof (nano::account) + sizeof (nano::signature) + sizeof (uint64_t));
+	if (type_a != nano::block_type::invalid && type_a != nano::block_type::not_a_block)
+	{
+		result += nano::block::size (type_a);
+	}
+	else if (type_a == nano::block_type::not_a_block)
+	{
+		result += count * sizeof (nano::block_hash);
+	}
+	return result;
+}
+
 nano::frontier_req::frontier_req () :
 message (nano::message_type::frontier_req)
 {
@@ -869,9 +961,6 @@ void nano::bulk_push::visit (nano::message_visitor & visitor_a) const
 	visitor_a.bulk_push (*this);
 }
 
-size_t constexpr nano::node_id_handshake::query_flag;
-size_t constexpr nano::node_id_handshake::response_flag;
-
 nano::node_id_handshake::node_id_handshake (bool & error_a, nano::stream & stream_a, nano::message_header const & header_a) :
 message (header_a),
 query (boost::none),
@@ -887,11 +976,11 @@ response (response)
 {
 	if (query)
 	{
-		set_query_flag (true);
+		header.flag_set (nano::message_header::node_id_handshake_query_flag);
 	}
 	if (response)
 	{
-		set_response_flag (true);
+		header.flag_set (nano::message_header::node_id_handshake_response_flag);
 	}
 }
 
@@ -915,14 +1004,14 @@ bool nano::node_id_handshake::deserialize (nano::stream & stream_a)
 	auto error (false);
 	try
 	{
-		if (is_query_flag ())
+		if (header.node_id_handshake_is_query ())
 		{
 			nano::uint256_union query_hash;
 			read (stream_a, query_hash);
 			query = query_hash;
 		}
 
-		if (is_response_flag ())
+		if (header.node_id_handshake_is_response ())
 		{
 			nano::account response_account;
 			read (stream_a, response_account);
@@ -945,31 +1034,30 @@ bool nano::node_id_handshake::operator== (nano::node_id_handshake const & other_
 	return result;
 }
 
-bool nano::node_id_handshake::is_query_flag () const
-{
-	return header.extensions.test (query_flag);
-}
-
-void nano::node_id_handshake::set_query_flag (bool value_a)
-{
-	header.extensions.set (query_flag, value_a);
-}
-
-bool nano::node_id_handshake::is_response_flag () const
-{
-	return header.extensions.test (response_flag);
-}
-
-void nano::node_id_handshake::set_response_flag (bool value_a)
-{
-	header.extensions.set (response_flag, value_a);
-}
-
 void nano::node_id_handshake::visit (nano::message_visitor & visitor_a) const
 {
 	visitor_a.node_id_handshake (*this);
 }
 
+size_t nano::node_id_handshake::size () const
+{
+	return size (header);
+}
+
+size_t nano::node_id_handshake::size (nano::message_header const & header_a)
+{
+	size_t result (0);
+	if (header_a.node_id_handshake_is_query ())
+	{
+		result = sizeof (nano::uint256_union);
+	}
+	if (header_a.node_id_handshake_is_response ())
+	{
+		result += sizeof (nano::account) + sizeof (nano::signature);
+	}
+	return result;
+}
+
 nano::message_visitor::~message_visitor ()
 {
 }

nano/node/common.hpp

@@ -186,19 +186,27 @@ public:
 	bool deserialize (nano::stream &);
 	nano::block_type block_type () const;
 	void block_type_set (nano::block_type);
+	uint8_t count_get () const;
+	void count_set (uint8_t);
 	uint8_t version_max;
 	uint8_t version_using;
 	uint8_t version_min;
 	nano::message_type type;
 	std::bitset<16> extensions;
 
-	static size_t constexpr bulk_pull_count_present_flag = 0;
+	void flag_set (uint8_t);
+	static uint8_t constexpr bulk_pull_count_present_flag = 0;
 	bool bulk_pull_is_count_present () const;
+	static uint8_t constexpr node_id_handshake_query_flag = 0;
+	static uint8_t constexpr node_id_handshake_response_flag = 1;
+	bool node_id_handshake_is_query () const;
+	bool node_id_handshake_is_response () const;
 
 	/** Size of the payload in bytes. For some messages, the payload size is based on header flags. */
 	size_t payload_length_bytes () const;
 
 	static std::bitset<16> constexpr block_type_mask = std::bitset<16> (0x0f00);
+	static std::bitset<16> constexpr count_mask = std::bitset<16> (0xf000);
 };
 class message
 {
@@ -289,6 +297,7 @@ public:
 	std::shared_ptr<nano::block> block;
 	std::vector<std::pair<nano::block_hash, nano::block_hash>> roots_hashes;
 	std::string roots_string () const;
+	static size_t size (nano::block_type, size_t = 0);
 };
 class confirm_ack final : public message
 {
@@ -299,6 +308,7 @@ public:
 	void visit (nano::message_visitor &) const override;
 	bool operator== (nano::confirm_ack const &) const;
 	std::shared_ptr<nano::vote> vote;
+	static size_t size (nano::block_type, size_t = 0);
 };
 class frontier_req final : public message
 {
@@ -363,14 +373,10 @@ public:
 	bool deserialize (nano::stream &);
 	void visit (nano::message_visitor &) const override;
 	bool operator== (nano::node_id_handshake const &) const;
-	bool is_query_flag () const;
-	void set_query_flag (bool);
-	bool is_response_flag () const;
-	void set_response_flag (bool);
 	boost::optional<nano::uint256_union> query;
 	boost::optional<std::pair<nano::account, nano::signature>> response;
-	static size_t constexpr query_flag = 0;
+	size_t size () const;
-	static size_t constexpr response_flag = 1;
+	static size_t size (nano::message_header const &);
 };
 class message_visitor
 {