refactor/fix: overhaul BeeSerde and msg buffers

* Serializer now requires a preallocated buffer in form of a &mut
  [u8]. This moves the responsibility to allocate the required memory
  to the user. Before it used a dynamically growing buffer (which
  would reallocate internally if needed). This change is crucial for
  future RDMA support.
* Since we now no longer use `BytesMut`, remove the bytes crate.
* The buffers handled by the store have a fixed size of 4 MiB. This is
  apparently the maximum size of a BeeMsg (see `WORKER_BUF(IN|OUT)_SIZE`
  in `Worker.h`). C++ server code also uses these fixed size.
* Generalize `msg_feature_flags` to a `Header` that can be
  modified from within the `Serializable` implementation and can be
  read out from within the `Deserializable` implementation.
* Collect all BeeMsg (de)serialization functions in one module and
  provide functions for header, body and both combined. The split
  is required because depending on where the data comes from / goes
  to different actions need to be taken. This also provides an easy
  interface for potential external users to handle BeeMsges.
* Remove the MsgBuf struct, instead just pass a `&mut [u8]` into the
dispatcher.
* Add documentation
* Various small code cleanups in BeeSerde and other locations

Author: rustybee42

Cargo.lock

@@ -11,7 +11,6 @@ publish = false
 
 [workspace.dependencies]
 anyhow = "1"
-bytes = "1"
 clap = { version = "4", features = ["derive"] }
 env_logger = "0"
 itertools = "0"

Cargo.toml

@@ -4,8 +4,8 @@ use crate::bee_msg::dispatch_request;
 use crate::license::LicenseVerifier;
 use crate::{ClientPulledStateNotification, StaticInfo};
 use anyhow::Result;
-use shared::conn::Pool;
 use shared::conn::msg_dispatch::*;
+use shared::conn::outgoing::Pool;
 use shared::run_state::WeakRunStateHandle;
 use shared::types::{NodeId, NodeType};
 use std::ops::Deref;

mgmtd/src/context.rs

@@ -82,7 +82,7 @@ targetStates=1
 fn check_target_states(f: &Path) -> Result<()> {
     let s = std::fs::read(f)?;
 
-    let mut des = Deserializer::new(&s, 0);
+    let mut des = Deserializer::new(&s);
     let states = des.map(
         false,
         |des| TargetId::deserialize(des),
@@ -184,7 +184,7 @@ struct ReadNodesResult {
 fn read_nodes(f: &Path) -> Result<ReadNodesResult> {
     let s = std::fs::read(f)?;
 
-    let mut des = Deserializer::new(&s, 0);
+    let mut des = Deserializer::new(&s);
     let version = des.u32()?;
     let root_id = des.u32()?;
     let root_mirrored = des.u8()?;
@@ -286,7 +286,7 @@ fn storage_targets(tx: &Transaction, targets_path: &Path) -> Result<()> {
 fn storage_pools(tx: &Transaction, f: &Path) -> Result<()> {
     let s = std::fs::read(f)?;
 
-    let mut des = Deserializer::new(&s, 0);
+    let mut des = Deserializer::new(&s);
     // Serialized as size_t, which should usually be 64 bit.
     let count = des.i64()?;
     let mut used_aliases = vec![];
@@ -412,7 +412,7 @@ fn quota_default_limits(tx: &Transaction, f: &Path, pool_id: PoolId) -> Result<(
         Err(err) => return Err(err.into()),
     };
 
-    let mut des = Deserializer::new(&s, 0);
+    let mut des = Deserializer::new(&s);
     let user_inode_limit = des.u64()?;
     let user_space_limit = des.u64()?;
     let group_inode_limit = des.u64()?;
@@ -494,7 +494,7 @@ fn quota_limits(
         Err(err) => return Err(err.into()),
     };
 
-    let mut des = Deserializer::new(&s, 0);
+    let mut des = Deserializer::new(&s);
     let limits = des.seq(false, |des| QuotaEntry::deserialize(des))?;
     des.finish()?;
 

mgmtd/src/db/import_v7.rs

@@ -19,7 +19,8 @@ use bee_msg::notify_nodes;
 use db::node_nic::ReplaceNic;
 use license::LicenseVerifier;
 use shared::bee_msg::target::RefreshTargetStates;
-use shared::conn::{Pool, incoming};
+use shared::conn::incoming;
+use shared::conn::outgoing::Pool;
 use shared::nic::Nic;
 use shared::run_state::{self, RunStateControl};
 use shared::types::{AuthSecret, MGMTD_UID, NicType, NodeId, NodeType};

mgmtd/src/lib.rs

@@ -11,7 +11,6 @@ publish.workspace = true
 bee_serde_derive = { path = "../bee_serde_derive" }
 
 anyhow = { workspace = true }
-bytes = { workspace = true }
 log = { workspace = true }
 pnet_datalink = "0"
 protobuf = { workspace = true, optional = true }

shared/Cargo.toml

@@ -2,12 +2,11 @@
 
 use crate::bee_serde::*;
 use crate::types::*;
-use anyhow::Result;
+use anyhow::{Context, Result, anyhow};
 use bee_serde_derive::BeeSerde;
 use std::collections::{HashMap, HashSet};
 
 pub mod buddy_group;
-pub mod header;
 pub mod misc;
 pub mod node;
 pub mod quota;
@@ -41,3 +40,168 @@ impl OpsErr {
     pub const AGAIN: Self = Self(22);
     pub const UNKNOWN_POOL: Self = Self(30);
 }
+
+/// The BeeMsg header
+#[derive(Clone, Debug, PartialEq, Eq, BeeSerde)]
+pub struct Header {
+    /// Total length of the serialized message, including the header itself
+    msg_len: u32,
+    /// Sometimes used for additional message specific payload and/or serialization info
+    pub msg_feature_flags: u16,
+    /// Sometimes used for additional message specific payload and/or serialization info
+    pub msg_compat_feature_flags: u8,
+    /// Sometimes used for additional message specific payload and/or serialization info
+    pub msg_flags: u8,
+    /// Fixed value to identify a BeeMsg header (see MSG_PREFIX below)
+    msg_prefix: u64,
+    /// Uniquely identifies the message type as defined in the C++ codebase in NetMessageTypes.h
+    msg_id: MsgId,
+    /// Sometimes used for additional message specific payload and/or serialization info
+    pub msg_target_id: TargetId,
+    /// Sometimes used for additional message specific payload and/or serialization info
+    pub msg_user_id: u32,
+    /// Mirroring related information
+    pub msg_seq: u64,
+    /// Mirroring related information
+    pub msg_seq_done: u64,
+}
+
+impl Header {
+    /// The serialized length of the header
+    pub const LEN: usize = 40;
+    /// Fixed value for identifying BeeMsges. In theory, this has some kind of version modifier
+    /// (thus the + 0), but it is unused
+    #[allow(clippy::identity_op)]
+    pub const MSG_PREFIX: u64 = (0x42474653 << 32) + 0;
+
+    /// The total length of the serialized message
+    pub fn msg_len(&self) -> usize {
+        self.msg_len as usize
+    }
+
+    /// The messages id
+    pub fn msg_id(&self) -> MsgId {
+        self.msg_id
+    }
+}
+
+impl Default for Header {
+    fn default() -> Self {
+        Self {
+            msg_len: 0,
+            msg_feature_flags: 0,
+            msg_compat_feature_flags: 0,
+            msg_flags: 0,
+            msg_prefix: Self::MSG_PREFIX,
+            msg_id: 0,
+            msg_target_id: 0,
+            msg_user_id: 0,
+            msg_seq: 0,
+            msg_seq_done: 0,
+        }
+    }
+}
+
+/// Serializes a BeeMsg body into the provided buffer.
+///
+/// The data is written from the beginning of the slice, it's up to the caller to pass the correct
+/// sub slice if space for the header should be reserved.
+///
+/// # Return value
+/// Returns the number of bytes written and the header modified by serialization function.
+pub fn serialize_body<M: Msg + Serializable>(msg: &M, buf: &mut [u8]) -> Result<(usize, Header)> {
+    let mut ser = Serializer::new(buf);
+    msg.serialize(&mut ser)
+        .context("BeeMsg body serialization failed")?;
+
+    Ok((ser.bytes_written(), ser.finish()))
+}
+
+/// Serializes a BeeMsg header into the provided buffer.
+///
+/// # Return value
+/// Returns the number of bytes written.
+pub fn serialize_header(header: &Header, buf: &mut [u8]) -> Result<usize> {
+    let mut ser_header = Serializer::new(buf);
+    header
+        .serialize(&mut ser_header)
+        .context("BeeMsg header serialization failed")?;
+
+    Ok(ser_header.bytes_written())
+}
+
+/// Serializes a complete BeeMsg (header + body) into the provided buffer.
+///
+/// # Return value
+/// Returns the number of bytes written.
+pub fn serialize<M: Msg + Serializable>(msg: &M, buf: &mut [u8]) -> Result<usize> {
+    let (written, mut header) = serialize_body(msg, &mut buf[Header::LEN..])?;
+
+    header.msg_len = (written + Header::LEN) as u32;
+    header.msg_id = M::ID;
+
+    let _ = serialize_header(&header, &mut buf[0..Header::LEN])?;
+
+    Ok(header.msg_len())
+}
+
+/// Deserializes a BeeMsg header from the provided buffer.
+///
+/// # Return value
+/// Returns the deserialized header.
+pub fn deserialize_header(buf: &[u8]) -> Result<Header> {
+    const CTX: &str = "BeeMsg header deserialization failed";
+
+    let header_buf = buf
+        .get(..Header::LEN)
+        .ok_or_else(|| {
+            anyhow!(
+                "Header buffer must be at least {} bytes big, got {}",
+                Header::LEN,
+                buf.len()
+            )
+        })
+        .context(CTX)?;
+
+    let mut des = Deserializer::new(header_buf);
+    let header = Header::deserialize(&mut des).context(CTX)?;
+    des.finish().context(CTX)?;
+
+    if header.msg_prefix != Header::MSG_PREFIX {
+        return Err(anyhow!(
+            "Invalid BeeMsg prefix: Must be {}, got {}",
+            Header::MSG_PREFIX,
+            header.msg_prefix
+        ))
+        .context(CTX);
+    }
+
+    Ok(header)
+}
+
+/// Deserializes a BeeMsg body from the provided buffer.
+///
+/// The data is read from the beginning of the slice, it's up to the caller to pass the correct
+/// sub slice if space for the header should be excluded from the source.
+///
+/// # Return value
+/// Returns the deserialized message.
+pub fn deserialize_body<M: Msg + Deserializable>(header: &Header, buf: &[u8]) -> Result<M> {
+    const CTX: &str = "BeeMsg body deserialization failed";
+
+    let mut des = Deserializer::with_header(&buf[0..(header.msg_len() - Header::LEN)], header);
+    let des_msg = M::deserialize(&mut des).context(CTX)?;
+    des.finish().context(CTX)?;
+
+    Ok(des_msg)
+}
+
+/// Deserializes a complete BeeMsg (header + body) from the provided buffer.
+///
+/// # Return value
+/// Returns the deserialized message.
+pub fn deserialize<M: Msg + Deserializable>(buf: &[u8]) -> Result<M> {
+    let header = deserialize_header(&buf[0..Header::LEN])?;
+    let msg = deserialize_body(&header, &buf[Header::LEN..])?;
+    Ok(msg)
+}