p2p-msg-size-logging-2825

core/src/global.rs

@@ -197,7 +197,7 @@ pub fn get_chain_type() -> ChainTypes {
 	CHAIN_TYPE.with(|chain_type| match chain_type.get() {
 		None => {
 			if !GLOBAL_CHAIN_TYPE.is_init() {
-				panic!("GLOBAL_CHAIN_TYPE and CHAIN_TYPE unset. Consider set_local_chain_type() in tests.");
+				std::panic!("GLOBAL_CHAIN_TYPE and CHAIN_TYPE unset. Consider set_local_chain_type() in tests.");
 			}
 			let chain_type = GLOBAL_CHAIN_TYPE.borrow();
 			set_local_chain_type(chain_type);

core/src/ser.rs

@@ -361,6 +361,11 @@ impl ProtocolVersion {
 		PROTOCOL_VERSION
 	}
 
+	/// Default implementation that returns the current protocol version
+	pub fn default() -> ProtocolVersion {
+		PROTOCOL_VERSION
+	}
+
 	/// We need to specify a protocol version for our local database.
 	/// Regardless of specific version used when sending/receiving data between peers
 	/// we need to take care with serialization/deserialization of data locally in the db.

doc/macros.md

@@ -0,0 +1,46 @@
+# Macros for Array Newtypes
+
+The `grin_util` crate provides several macros for working with array newtypes - wrapper types around fixed-size arrays. These macros help implement common traits and functionality for these types.
+
+## Available Macros
+
+### `impl_array_newtype`
+
+Implements standard array traits and behavior for newtype wrappers around fixed-size arrays. This includes:
+
+- Methods like `as_ptr()`, `as_mut_ptr()`, `len()`, etc.
+- Indexing via `Index` traits
+- Comparison traits (`PartialEq`, `Eq`, `PartialOrd`, `Ord`)
+- Common traits like `Clone`, `Copy`, and `Hash`
+
+### `impl_array_newtype_encodable`
+
+Implements serialization and deserialization support via Serde for newtype wrappers.
+
+### `impl_array_newtype_show`
+
+Implements the `Debug` trait for pretty-printing the array newtype.
+
+### `impl_index_newtype`
+
+Implements various indexing operations for the newtype. This is automatically called by `impl_array_newtype`.
+
+## Usage Examples
+
+```rust
+// Define a newtype for a 32-byte array
+pub struct ChainCode([u8; 32]);
+
+// Implement standard array traits
+impl_array_newtype!(ChainCode, u8, 32);
+
+// Implement Debug formatting
+impl_array_newtype_show!(ChainCode);
+
+// Implement Serde serialization/deserialization
+impl_array_newtype_encodable!(ChainCode, u8, 32);
+```
+
+## Notes on Feature Flags
+
+With recent Rust versions, conditional compilation within macros is handled differently. The `serde` and other features are now defined at the crate level rather than inside the macros themselves, which prevents warnings about unexpected `cfg` conditions.

doc/pool/transaction_pool.md

@@ -0,0 +1,21 @@
+## Transaction Pool
+
+Grin's transaction pool is designed to hold all transactions that are not yet included in a block.
+
+The transaction pool is split into a stempool and a txpool. The stempool contains "stem" transactions, which are less actively propagated to the rest of the network, as well as txs received via Dandelion "stem" phase. The txpool contains transactions that may be directly propagated to the network, as well as txs received via Dandelion "fluff" phase.
+
+### Reconciliation
+
+The `Pool::reconcile` function validates transactions in the stempool or txpool against a given block header and removes invalid or duplicated transactions (present in txpool). The optimized implementation filters entries in-place, reducing validations from O(n² + n*m) to O(n + m), where n is the number of transactions in the pool being reconciled and m is the number of transactions in txpool.
+
+Reconciliation logs include:
+- Number of entries before/after reconciliation
+- Count of invalid or duplicated transactions removed
+
+Example:
+```
+INFO: Starting transaction pool reconciliation with 200 entries
+WARN: Skipping duplicate transaction: <hash>
+WARN: Invalid transaction <hash>: Validation failed
+INFO: Reconciliation complete: retained 180 entries, removed 10 invalid, 10 duplicates
+```

grin/p2p/src/msg.rs

@@ -0,0 +1,43 @@
+// ...existing code...
+use log::{info, warn};
+// ...existing code...
+
+impl Message {
+	pub fn read<R: Read>(
+		reader: &mut R,
+		msg_type: Option<MessageTypeEnum>,
+	) -> Result<Message, Error> {
+		// ...existing code...
+		let header = MessageHeader::read(reader)?;
+		let msg_len = header.msg_len as usize;
+
+		match msg_type {
+			Some(msg_type) => {
+				let max_len = max_msg_size(msg_type);
+				let current_max_len = max_len * 4; // Current 4x limit
+				if msg_len > current_max_len {
+					return Err(Error::MsgTooLarge(msg_len, current_max_len));
+				}
+				info!(
+					"Received {:?} message: size={} bytes, 1x limit={} bytes, 4x limit={} bytes",
+					msg_type, msg_len, max_len, current_max_len
+				);
+				if msg_len > max_len {
+					warn!(
+						"Message size ({} bytes) exceeds 1x limit ({} bytes) for type {:?}",
+						msg_len, max_len, msg_type
+					);
+				}
+			}
+			None => {
+				info!("Received unknown message type: size={} bytes", msg_len);
+			}
+		}
+
+		let mut payload = vec![0u8; msg_len];
+		reader.read_exact(&mut payload)?;
+		Ok(Message { header, payload })
+	}
+	// ...existing code...
+}
+// ...existing code...

p2p/src/msg.rs

@@ -34,6 +34,7 @@ use crate::types::{
 };
 use crate::util::secp::pedersen::RangeProof;
 use bytes::Bytes;
+use log::{info, warn};
 use num::FromPrimitive;
 use std::fs::File;
 use std::io::{Read, Write};
@@ -97,7 +98,7 @@ fn default_max_msg_size() -> u64 {
 }
 
 // Max msg size for each msg type.
-fn max_msg_size(msg_type: Type) -> u64 {
+pub fn max_msg_size(msg_type: Type) -> u64 {
 	match msg_type {
 		Type::Error => 0,
 		Type::Hand => 128,
@@ -172,7 +173,7 @@ impl Msg {
 ///
 /// Note: We return a MsgHeaderWrapper here as we may encounter an unknown msg type.
 ///
-pub fn read_header<R: Read>(
+pub fn read_header<R: std::io::Read>(
 	stream: &mut R,
 	version: ProtocolVersion,
 ) -> Result<MsgHeaderWrapper, Error> {
@@ -186,7 +187,7 @@ pub fn read_header<R: Read>(
 /// Read a single item from the provided stream, always blocking until we
 /// have a result (or timeout).
 /// Returns the item and the total bytes read.
-pub fn read_item<T: Readable, R: Read>(
+pub fn read_item<T: Readable, R: std::io::Read>(
 	stream: &mut R,
 	version: ProtocolVersion,
 ) -> Result<(T, u64), Error> {
@@ -197,7 +198,7 @@ pub fn read_item<T: Readable, R: Read>(
 
 /// Read a message body from the provided stream, always blocking
 /// until we have a result (or timeout).
-pub fn read_body<T: Readable, R: Read>(
+pub fn read_body<T: Readable, R: std::io::Read>(
 	h: &MsgHeader,
 	stream: &mut R,
 	version: ProtocolVersion,
@@ -208,14 +209,14 @@ pub fn read_body<T: Readable, R: Read>(
 }
 
 /// Read (an unknown) message from the provided stream and discard it.
-pub fn read_discard<R: Read>(msg_len: u64, stream: &mut R) -> Result<(), Error> {
+pub fn read_discard<R: std::io::Read>(msg_len: u64, stream: &mut R) -> Result<(), Error> {
 	let mut buffer = vec![0u8; msg_len as usize];
 	stream.read_exact(&mut buffer)?;
 	Ok(())
 }
 
 /// Reads a full message from the underlying stream.
-pub fn read_message<T: Readable, R: Read>(
+pub fn read_message<T: Readable, R: std::io::Read>(
 	stream: &mut R,
 	version: ProtocolVersion,
 	msg_type: Type,
@@ -322,6 +323,24 @@ impl Writeable for MsgHeader {
 	}
 }
 
+impl MsgHeader {
+	/// Read a message header from the provided reader
+	pub fn read<R: std::io::Read>(reader: &mut R) -> Result<MsgHeader, Error> {
+		let mut head = vec![0u8; MsgHeader::LEN];
+		reader.read_exact(&mut head)?;
+		let header: MsgHeaderWrapper = ser::deserialize(
+			&mut &head[..],
+			ProtocolVersion::local(),
+			DeserializationMode::default(),
+		)?;
+
+		match header {
+			MsgHeaderWrapper::Known(header) => Ok(header),
+			MsgHeaderWrapper::Unknown(_, _) => Err(Error::BadMessage),
+		}
+	}
+}
+
 impl Readable for MsgHeaderWrapper {
 	fn read<R: Reader>(reader: &mut R) -> Result<MsgHeaderWrapper, ser::Error> {
 		let m = magic();
@@ -986,3 +1005,41 @@ impl fmt::Debug for Consumed {
 		}
 	}
 }
+
+impl Message {
+	pub fn read<R: std::io::Read>(
+		reader: &mut R,
+		msg_type: Option<Type>,
+	) -> Result<Vec<u8>, Error> {
+		use log::{info, warn};
+		let header = MsgHeader::read(reader)?;
+		let msg_len = header.msg_len;
+
+		match msg_type {
+			Some(msg_type) => {
+				let max_len = max_msg_size(msg_type);
+				let current_max_len = max_len * 4; // Current 4x limit
+				if msg_len > current_max_len {
+					return Err(Error::MsgTooLarge(msg_len as usize, current_max_len));
+				}
+				info!(
+					"Received {:?} message: size={} bytes, 1x limit={} bytes, 4x limit={} bytes",
+					msg_type, msg_len, max_len, current_max_len
+				);
+				if msg_len > max_len {
+					warn!(
+						"Message size ({} bytes) exceeds 1x limit ({} bytes) for type {:?}",
+						msg_len, max_len, msg_type
+					);
+				}
+			}
+			None => {
+				info!("Received unknown message type: size={} bytes", msg_len);
+			}
+		}
+
+		let mut payload = vec![0u8; msg_len as usize];
+		reader.read_exact(&mut payload)?;
+		std::result::Result::Ok(payload)
+	}
+}

p2p/src/types.rs

@@ -90,6 +90,7 @@ pub enum Error {
 	PeerNotBanned,
 	PeerException,
 	Internal,
+	MsgTooLarge(usize, u64), // Message size, maximum allowed size
 }
 
 impl From<ser::Error> for Error {
@@ -113,6 +114,38 @@ impl From<io::Error> for Error {
 	}
 }
 
+impl fmt::Display for Error {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		match self {
+			Error::Serialization(ref e) => write!(f, "Serialization error: {}", e),
+			Error::Connection(ref e) => write!(f, "Connection error: {}", e),
+			Error::BadMessage => write!(f, "Bad message"),
+			Error::UnexpectedMessage => write!(f, "Unexpected message"),
+			Error::MsgLen => write!(f, "Wrong message length"),
+			Error::Banned => write!(f, "Peer banned"),
+			Error::ConnectionClose => write!(f, "Connection closed"),
+			Error::Timeout => write!(f, "Connection timed out"),
+			Error::Store(ref e) => write!(f, "Store error: {}", e),
+			Error::Chain(ref e) => write!(f, "Chain error: {}", e),
+			Error::PeerWithSelf => write!(f, "Connect to self"),
+			Error::NoDandelionRelay => write!(f, "No Dandelion relay"),
+			Error::GenesisMismatch { us, peer } => {
+				write!(f, "Genesis mismatch: our={}, peer={}", us, peer)
+			}
+			Error::Send(ref s) => write!(f, "Send error: {}", s),
+			Error::PeerNotFound => write!(f, "Peer not found"),
+			Error::PeerNotBanned => write!(f, "Peer not banned"),
+			Error::PeerException => write!(f, "Peer exception"),
+			Error::Internal => write!(f, "Internal error"),
+			Error::MsgTooLarge(size, max) => write!(
+				f,
+				"Message too large: {} bytes, maximum: {} bytes",
+				size, max
+			),
+		}
+	}
+}
+
 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
 pub struct PeerAddr(pub SocketAddr);
 

servers/src/grin/server.rs

@@ -577,7 +577,8 @@ impl Server {
 		// this call is blocking and makes sure all peers stop, however
 		// we can't be sure that we stopped a listener blocked on accept, so we don't join the p2p thread
 		self.p2p.stop();
-		let _ = self.lock_file.unlock();
+		// let _ = self.lock_file.unlock();
+		let _ = fs2::FileExt::unlock(&*self.lock_file);
 		warn!("Shutdown complete");
 	}
 

src/main.rs

@@ -0,0 +1,9 @@
+use std::error::Error;
+
+fn main() -> Result<(), Box<dyn Error>> {
+	env_logger::init();
+
+	// ...existing code...
+
+	Ok(())
+}