mesh_remote: windows: fix messages near the small message limit

Cargo.lock

@@ -4092,6 +4092,7 @@ dependencies = [
 name = "mesh_remote"
 version = "0.0.0"
 dependencies = [
+ "event-listener",
  "futures",
  "futures-concurrency",
  "libc",

support/mesh/mesh_remote/Cargo.toml

@@ -31,6 +31,7 @@ libc.workspace = true
 socket2.workspace = true
 
 [dev-dependencies]
+event-listener.workspace = true
 test_with_tracing.workspace = true
 
 [target.'cfg(windows)'.dev-dependencies]

support/mesh/mesh_remote/src/alpc_node.rs

@@ -741,13 +741,15 @@ impl Connection {
 
 impl SendEvent for Connection {
     fn event(&self, event: OutgoingEvent<'_>) {
+        let len = event.len();
         match self.send_event(event) {
             Ok(_) => (),
             Err(err) => {
                 tracing::error!(
                     node = ?self.local_id,
                     remote_node = ?self.remote_id,
                     error = err.as_error(),
+                    len,
                     "error sending packet"
                 );
                 // Notify the connection task of the failure.
@@ -825,6 +827,18 @@ mod tests {
         node2.shutdown().await;
     }
 
+    #[async_test]
+    async fn test_message_sizes(driver: DefaultDriver) {
+        let (p1, p2) = mesh_node::local_node::Port::new_pair();
+        let (p3, p4) = mesh_node::local_node::Port::new_pair();
+        let node1 = AlpcNode::new(driver.clone()).unwrap();
+        let (invitation, _handle) = node1.invite(p2).unwrap();
+        let _node2 = AlpcNode::join(driver.clone(), invitation, p3).unwrap();
+
+        crate::test_common::test_message_sizes(p1, p4, 0..=super::MAX_SMALL_EVENT_SIZE + 0x1000)
+            .await;
+    }
+
     #[async_test]
     async fn test_three(driver: DefaultDriver) {
         let (p1, p2) = channel::<u32>();

support/mesh/mesh_remote/src/lib.rs

@@ -7,6 +7,7 @@ mod alpc_node;
 mod common;
 mod point_to_point;
 mod protocol;
+mod test_common;
 mod unix_node;
 
 #[cfg(windows)]

support/mesh/mesh_remote/src/test_common.rs

@@ -0,0 +1,68 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+#![cfg(test)]
+
+use mesh_node::local_node::HandlePortEvent;
+use mesh_node::local_node::Port;
+use mesh_node::message::Message;
+use std::ops::RangeInclusive;
+use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering::Relaxed;
+
+#[cfg_attr(not(any(windows, target_os = "linux")), expect(dead_code))]
+pub(crate) async fn test_message_sizes(left: Port, right: Port, range: RangeInclusive<usize>) {
+    let counter = Arc::new(Counter {
+        count: AtomicUsize::new(*range.start()),
+        event: event_listener::Event::new(),
+    });
+    let _right = right.set_handler(CountHandler(counter.clone()));
+    let buf = vec![0; *range.end()];
+    for i in range {
+        if i % 0x1000 == 0 {
+            tracing::info!(i, "at message");
+        }
+        left.send(Message::serialized(&buf[..i], Vec::new()));
+        loop {
+            let listener = counter.event.listen();
+            if counter.count.load(Relaxed) == i + 1 {
+                break;
+            }
+            listener.await;
+        }
+    }
+}
+
+struct Counter {
+    count: AtomicUsize,
+    event: event_listener::Event,
+}
+
+struct CountHandler(Arc<Counter>);
+
+impl HandlePortEvent for CountHandler {
+    fn message<'a>(
+        &mut self,
+        _control: &mut mesh_node::local_node::PortControl<'_, 'a>,
+        _message: Message<'a>,
+    ) -> Result<(), mesh_node::local_node::HandleMessageError> {
+        self.0.count.fetch_add(1, Relaxed);
+        self.0.event.notify(1);
+        Ok(())
+    }
+
+    fn close(&mut self, _control: &mut mesh_node::local_node::PortControl<'_, '_>) {}
+
+    fn fail(
+        &mut self,
+        _control: &mut mesh_node::local_node::PortControl<'_, '_>,
+        _err: mesh_node::local_node::NodeError,
+    ) {
+        panic!("failed after {} messages", self.0.count.load(Relaxed));
+    }
+
+    fn drain(&mut self) -> Vec<mesh_node::message::OwnedMessage> {
+        Vec::new()
+    }
+}

support/mesh/mesh_remote/src/unix_node.rs

@@ -1334,6 +1334,21 @@ mod tests {
         leader.shutdown().await;
     }
 
+    #[cfg(target_os = "linux")]
+    #[async_test]
+    async fn test_message_sizes(driver: DefaultDriver) {
+        let (p1, p2) = mesh_node::local_node::Port::new_pair();
+        let (p3, p4) = mesh_node::local_node::Port::new_pair();
+        let node1 = UnixNode::new(driver.clone());
+        let invitation = node1.invite(p2).await.unwrap();
+        let _node2 = UnixNode::join(driver.clone(), invitation, p3)
+            .await
+            .unwrap();
+
+        crate::test_common::test_message_sizes(p1, p4, 0..=super::MAX_SMALL_EVENT_SIZE + 0x1000)
+            .await;
+    }
+
     #[async_test]
     async fn test_dropped_shutdown(driver: DefaultDriver) {
         let leader = UnixNode::new(driver.clone());

support/pal/src/windows/alpc.rs

@@ -139,7 +139,7 @@ impl PortConfig {
                 | ALPC_PORFLG_ACCEPT_INDIRECT_HANDLES
                 | ALPC_PORFLG_ACCEPT_REQUESTS,
             DupObjectTypes: OB_ALL_OBJECT_TYPE_CODES,
-            MaxMessageLength: self.max_message_len,
+            MaxMessageLength: self.max_message_len + size_of::<PORT_MESSAGE>(),
             MaxPoolUsage: usize::MAX,
             MaxSectionSize: usize::MAX,
             MaxTotalSectionSize: usize::MAX,