add shadow-compatible quinn-udp fallback

lean_client/Cargo.toml

@@ -311,6 +311,9 @@ name = "lean_client"
 version = "0.1.0"
 edition = { workspace = true }
 
+[features]
+shadow-integration = ["xmss/shadow-integration"]
+
 [dependencies]
 anyhow = { workspace = true }
 bls = { workspace = true }
@@ -341,3 +344,6 @@ tikv-jemallocator = { workspace = true }
 [profile.release]
 lto = true
 codegen-units = 1
+
+[profile.shadow]
+inherits = "release"

lean_client/Makefile

@@ -126,6 +126,39 @@ docker-local: ./target/x86_64-unknown-linux-gnu/release/lean_client
 		$(DOCKER_TAGS) \
 		.
 
+### Shadow-simulator support
+# `rust/patch/quinn-udp` is a vendored fallback-only build of quinn-udp so QUIC
+# runs under the Shadow simulator (which does not emulate sendmsg cmsg / GRO /
+# GSO / ECN). Applied only for the shadow-* targets via cargo's --config
+# override, so prod / devnet builds see the crates.io version.
+SHADOW_CONFIG := --config 'patch.crates-io."quinn-udp".path="rust/patch/quinn-udp"'
+SHADOW_FEATURES := --features shadow-integration
+
+.PHONY: shadow-build
+shadow-build:
+	cargo build --profile shadow $(SHADOW_FEATURES) $(SHADOW_CONFIG)
+
+.PHONY: shadow-x86_64-unknown-linux-gnu
+shadow-x86_64-unknown-linux-gnu:
+	CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUSTFLAGS="-C target-cpu=x86-64-v3" \
+	LEAN_REPO_ROOT=$(LEAN_REPO_ROOT) cross build --bin lean_client --target x86_64-unknown-linux-gnu --profile shadow $(SHADOW_FEATURES) $(SHADOW_CONFIG)
+
+.PHONY: shadow-aarch64-unknown-linux-gnu
+shadow-aarch64-unknown-linux-gnu:
+	LEAN_REPO_ROOT=$(LEAN_REPO_ROOT) cross build --bin lean_client --target aarch64-unknown-linux-gnu --profile shadow $(SHADOW_FEATURES) $(SHADOW_CONFIG)
+
+.PHONY: shadow-docker-local
+shadow-docker-local: shadow-x86_64-unknown-linux-gnu
+	@mkdir -p ./bin/amd64
+	@cp ./target/x86_64-unknown-linux-gnu/shadow/lean_client ./bin/amd64/lean_client
+	docker build \
+		--file Dockerfile \
+		--build-arg COMMIT_SHA=$(COMMIT_SHA) \
+		--build-arg BUILD_DATE=$(BUILD_DATE) \
+		--build-arg GIT_BRANCH=$(GIT_BRANCH) \
+		$(DOCKER_TAGS) \
+		.
+
 .PHONY: help
 help:
 	@awk '/^## / { \

lean_client/networking/src/network/service.rs

@@ -636,9 +636,8 @@ where
                         METRICS
                             .get()
                             .map(|m| m.lean_gossip_block_size_bytes.observe(data_len as f64));
-                        info!(block_root = %signed_block.block.hash_tree_root(), "received block via gossip");
-
                         let slot = signed_block.block.slot.0;
+                        info!(slot, block_root = %signed_block.block.hash_tree_root(), "received block via gossip");
 
                         if let Err(err) = self
                             .chain_message_sink

lean_client/rust/patch/quinn-udp/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "quinn-udp"
+version = "0.5.14"
+edition = "2021"
+license = "MIT OR Apache-2.0"
+description = "UDP sockets for QUIC — Shadow-compatible fallback-only build"
+
+[features]
+default = ["tracing", "log"]
+log = ["tracing/log"]
+direct-log = ["dep:log"]
+
+[dependencies]
+log = { version = "0.4", optional = true }
+tracing = { version = "0.1", optional = true }
+
+[target.'cfg(not(all(target_family = "wasm", target_os = "unknown")))'.dependencies]
+socket2 = { version = "0.6" }
+
+[lib]
+bench = false

lean_client/rust/patch/quinn-udp/src/fallback.rs

@@ -0,0 +1,126 @@
+use std::{
+    io::{self, IoSliceMut},
+    sync::Mutex,
+    time::Instant,
+};
+
+use super::{IO_ERROR_LOG_INTERVAL, RecvMeta, Transmit, UdpSockRef, log_sendmsg_error};
+
+/// Fallback UDP socket interface that stubs out all special functionality
+///
+/// Used when a better implementation is not available for a particular target, at the cost of
+/// reduced performance compared to that enabled by some target-specific interfaces.
+#[derive(Debug)]
+pub struct UdpSocketState {
+    last_send_error: Mutex<Instant>,
+}
+
+impl UdpSocketState {
+    pub fn new(socket: UdpSockRef<'_>) -> io::Result<Self> {
+        socket.0.set_nonblocking(true)?;
+        let now = Instant::now();
+        Ok(Self {
+            last_send_error: Mutex::new(now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now)),
+        })
+    }
+
+    /// Sends a [`Transmit`] on the given socket.
+    ///
+    /// This function will only ever return errors of kind [`io::ErrorKind::WouldBlock`].
+    /// All other errors will be logged and converted to `Ok`.
+    ///
+    /// UDP transmission errors are considered non-fatal because higher-level protocols must
+    /// employ retransmits and timeouts anyway in order to deal with UDP's unreliable nature.
+    /// Thus, logging is most likely the only thing you can do with these errors.
+    ///
+    /// If you would like to handle these errors yourself, use [`UdpSocketState::try_send`]
+    /// instead.
+    pub fn send(&self, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> {
+        match send(socket, transmit) {
+            Ok(()) => Ok(()),
+            Err(e) if e.kind() == io::ErrorKind::WouldBlock => Err(e),
+            Err(e) => {
+                log_sendmsg_error(&self.last_send_error, e, transmit);
+                Ok(())
+            }
+        }
+    }
+
+    /// Sends a [`Transmit`] on the given socket without any additional error handling.
+    pub fn try_send(&self, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> {
+        send(socket, transmit)
+    }
+
+    pub fn recv(
+        &self,
+        socket: UdpSockRef<'_>,
+        bufs: &mut [IoSliceMut<'_>],
+        meta: &mut [RecvMeta],
+    ) -> io::Result<usize> {
+        // Safety: both `IoSliceMut` and `MaybeUninitSlice` promise to have the
+        // same layout, that of `iovec`/`WSABUF`. Furthermore `recv_vectored`
+        // promises to not write unitialised bytes to the `bufs` and pass it
+        // directly to the `recvmsg` system call, so this is safe.
+        let bufs = unsafe {
+            &mut *(bufs as *mut [IoSliceMut<'_>] as *mut [socket2::MaybeUninitSlice<'_>])
+        };
+        let (len, _flags, addr) = socket.0.recv_from_vectored(bufs)?;
+        meta[0] = RecvMeta {
+            len,
+            stride: len,
+            addr: addr.as_socket().unwrap(),
+            ecn: None,
+            dst_ip: None,
+        };
+        Ok(1)
+    }
+
+    #[inline]
+    pub fn max_gso_segments(&self) -> usize {
+        1
+    }
+
+    #[inline]
+    pub fn gro_segments(&self) -> usize {
+        1
+    }
+
+    /// Resize the send buffer of `socket` to `bytes`
+    #[inline]
+    pub fn set_send_buffer_size(&self, socket: UdpSockRef<'_>, bytes: usize) -> io::Result<()> {
+        socket.0.set_send_buffer_size(bytes)
+    }
+
+    /// Resize the receive buffer of `socket` to `bytes`
+    #[inline]
+    pub fn set_recv_buffer_size(&self, socket: UdpSockRef<'_>, bytes: usize) -> io::Result<()> {
+        socket.0.set_recv_buffer_size(bytes)
+    }
+
+    /// Get the size of the `socket` send buffer
+    #[inline]
+    pub fn send_buffer_size(&self, socket: UdpSockRef<'_>) -> io::Result<usize> {
+        socket.0.send_buffer_size()
+    }
+
+    /// Get the size of the `socket` receive buffer
+    #[inline]
+    pub fn recv_buffer_size(&self, socket: UdpSockRef<'_>) -> io::Result<usize> {
+        socket.0.recv_buffer_size()
+    }
+
+    #[inline]
+    pub fn may_fragment(&self) -> bool {
+        true
+    }
+}
+
+fn send(socket: UdpSockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> {
+    socket.0.send_to(
+        transmit.contents,
+        &socket2::SockAddr::from(transmit.destination),
+    )?;
+    Ok(())
+}
+
+pub(crate) const BATCH_SIZE: usize = 1;

lean_client/rust/patch/quinn-udp/src/lib.rs

@@ -0,0 +1,159 @@
+//! quinn-udp — Shadow-compatible fallback-only build
+//!
+//! This is a vendored replacement for quinn-udp 0.5.14 that always uses the
+//! simple fallback UDP path (send_to/recv_from_vectored) instead of the
+//! Linux-optimized unix.rs path. This is required for Shadow network simulator
+//! compatibility, since Shadow does not fully emulate sendmsg/recvmsg cmsg,
+//! GRO/GSO, or ECN socket options.
+//!
+//! See: https://shadow.github.io/
+#![warn(unreachable_pub)]
+#![warn(clippy::use_self)]
+
+use std::net::{IpAddr, Ipv6Addr, SocketAddr};
+#[cfg(unix)]
+use std::os::unix::io::AsFd;
+#[cfg(windows)]
+use std::os::windows::io::AsSocket;
+use std::{
+    sync::Mutex,
+    time::{Duration, Instant},
+};
+
+// Always use fallback — no unix.rs, no windows.rs, no cmsg
+#[path = "fallback.rs"]
+mod imp;
+
+#[allow(unused_imports, unused_macros)]
+mod log {
+    #[cfg(all(feature = "direct-log", not(feature = "tracing")))]
+    pub(crate) use log::{debug, error, info, trace, warn};
+
+    #[cfg(feature = "tracing")]
+    pub(crate) use tracing::{debug, error, info, trace, warn};
+
+    #[cfg(not(any(feature = "direct-log", feature = "tracing")))]
+    mod no_op {
+        macro_rules! trace    ( ($($tt:tt)*) => {{}} );
+        macro_rules! debug    ( ($($tt:tt)*) => {{}} );
+        macro_rules! info     ( ($($tt:tt)*) => {{}} );
+        macro_rules! log_warn ( ($($tt:tt)*) => {{}} );
+        macro_rules! error    ( ($($tt:tt)*) => {{}} );
+
+        pub(crate) use {debug, error, info, log_warn as warn, trace};
+    }
+
+    #[cfg(not(any(feature = "direct-log", feature = "tracing")))]
+    pub(crate) use no_op::*;
+}
+
+pub use imp::UdpSocketState;
+
+/// Number of UDP packets to send/receive at a time
+pub const BATCH_SIZE: usize = imp::BATCH_SIZE;
+
+/// Metadata for a single buffer filled with bytes received from the network
+#[derive(Debug, Copy, Clone)]
+pub struct RecvMeta {
+    pub addr: SocketAddr,
+    pub len: usize,
+    pub stride: usize,
+    pub ecn: Option<EcnCodepoint>,
+    pub dst_ip: Option<IpAddr>,
+}
+
+impl Default for RecvMeta {
+    fn default() -> Self {
+        Self {
+            addr: SocketAddr::new(Ipv6Addr::UNSPECIFIED.into(), 0),
+            len: 0,
+            stride: 0,
+            ecn: None,
+            dst_ip: None,
+        }
+    }
+}
+
+/// An outgoing packet
+#[derive(Debug, Clone)]
+pub struct Transmit<'a> {
+    pub destination: SocketAddr,
+    pub ecn: Option<EcnCodepoint>,
+    pub contents: &'a [u8],
+    pub segment_size: Option<usize>,
+    pub src_ip: Option<IpAddr>,
+}
+
+/// Log at most 1 IO error per minute
+const IO_ERROR_LOG_INTERVAL: Duration = std::time::Duration::from_secs(60);
+
+#[cfg(any(feature = "tracing", feature = "direct-log"))]
+fn log_sendmsg_error(
+    last_send_error: &Mutex<Instant>,
+    err: impl core::fmt::Debug,
+    transmit: &Transmit,
+) {
+    let now = Instant::now();
+    let last_send_error = &mut *last_send_error.lock().expect("poisoned lock");
+    if now.saturating_duration_since(*last_send_error) > IO_ERROR_LOG_INTERVAL {
+        *last_send_error = now;
+        log::warn!(
+            "sendmsg error: {:?}, Transmit: {{ destination: {:?}, src_ip: {:?}, ecn: {:?}, len: {:?}, segment_size: {:?} }}",
+            err,
+            transmit.destination,
+            transmit.src_ip,
+            transmit.ecn,
+            transmit.contents.len(),
+            transmit.segment_size
+        );
+    }
+}
+
+#[cfg(not(any(feature = "tracing", feature = "direct-log")))]
+fn log_sendmsg_error(_: &Mutex<Instant>, _: impl core::fmt::Debug, _: &Transmit) {}
+
+/// A borrowed UDP socket
+pub struct UdpSockRef<'a>(socket2::SockRef<'a>);
+
+#[cfg(unix)]
+impl<'s, S> From<&'s S> for UdpSockRef<'s>
+where
+    S: AsFd,
+{
+    fn from(socket: &'s S) -> Self {
+        Self(socket.into())
+    }
+}
+
+#[cfg(windows)]
+impl<'s, S> From<&'s S> for UdpSockRef<'s>
+where
+    S: AsSocket,
+{
+    fn from(socket: &'s S) -> Self {
+        Self(socket.into())
+    }
+}
+
+/// Explicit congestion notification codepoint
+#[repr(u8)]
+#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+pub enum EcnCodepoint {
+    Ect0 = 0b10,
+    Ect1 = 0b01,
+    Ce = 0b11,
+}
+
+impl EcnCodepoint {
+    pub fn from_bits(x: u8) -> Option<Self> {
+        use EcnCodepoint::*;
+        Some(match x & 0b11 {
+            0b10 => Ect0,
+            0b01 => Ect1,
+            0b11 => Ce,
+            _ => {
+                return None;
+            }
+        })
+    }
+}