@andresilva review

Author: clabby

actor/src/lib.rs

@@ -179,7 +179,11 @@ stability_scope!(ALPHA {
             futures::future::pending()
         }
 
-        /// Runs at the end of each iteration.
+        /// Runs at the end of each iteration that dispatched a message.
+        ///
+        /// Not guaranteed to be called after every single message. When
+        /// multiple read-only messages are batched concurrently,
+        /// `postprocess` may only run after the full batch completes.
         fn postprocess(
             &mut self,
             _context: &mut E,

actor/src/service/builder.rs

@@ -98,7 +98,7 @@ impl<A> ServiceBuilder<A> {
     /// `capacity` is per-lane queue depth and must be non-zero.
     pub fn with_lane<L>(self, lane: L, capacity: NonZeroUsize) -> MultiLaneServiceBuilder<A, L>
     where
-        L: Copy + Ord + Send + 'static,
+        L: Ord + Send + 'static,
     {
         MultiLaneServiceBuilder {
             actor: self.actor,
@@ -110,7 +110,7 @@ impl<A> ServiceBuilder<A> {
     /// Add an unbounded lane, transitioning to a [`MultiLaneUnboundedServiceBuilder`].
     pub fn with_unbounded_lane<L>(self, lane: L) -> MultiLaneUnboundedServiceBuilder<A, L>
     where
-        L: Copy + Ord + Send + 'static,
+        L: Ord + Send + 'static,
     {
         MultiLaneUnboundedServiceBuilder {
             actor: self.actor,
@@ -186,7 +186,7 @@ impl<A> ServiceBuilder<A> {
 /// Configures a multi-lane actor service loop with bounded lanes.
 pub struct MultiLaneServiceBuilder<A, L>
 where
-    L: Copy + Ord + Send + 'static,
+    L: Ord + Send + 'static,
 {
     actor: A,
     lanes: Vec<(L, NonZeroUsize)>,
@@ -195,7 +195,7 @@ where
 
 impl<A, L> MultiLaneServiceBuilder<A, L>
 where
-    L: Copy + Ord + Send + 'static,
+    L: Ord + Send + 'static,
 {
     /// Add another bounded lane.
     ///
@@ -245,7 +245,7 @@ where
 /// Configures a multi-lane actor service loop with unbounded lanes.
 pub struct MultiLaneUnboundedServiceBuilder<A, L>
 where
-    L: Copy + Ord + Send + 'static,
+    L: Ord + Send + 'static,
 {
     actor: A,
     lanes: Vec<L>,
@@ -254,7 +254,7 @@ where
 
 impl<A, L> MultiLaneUnboundedServiceBuilder<A, L>
 where
-    L: Copy + Ord + Send + 'static,
+    L: Ord + Send + 'static,
 {
     /// Add another unbounded lane.
     pub fn with_unbounded_lane(mut self, lane: L) -> Self {

actor/src/service/driver.rs

@@ -203,39 +203,52 @@ where
                 let external = self
                     .actor
                     .on_external(self.context.as_present_mut(), &mut args);
-                Self::recv_event(&mut self.shutdown, &mut self.lanes, pin!(external)).await
-            };
-            match event {
-                LoopEvent::Shutdown => {
-                    self.shutdown_gracefully(&mut args, &mut reads, "shutdown signal received")
-                        .await;
-                    return;
+                let external = pin!(external);
+                let recv = Self::recv_event(&mut self.shutdown, &mut self.lanes, external);
+                if reads.is_empty() {
+                    Some(recv.await)
+                } else {
+                    select! {
+                        event = recv => Some(event),
+                        result = reads.next() => {
+                            result.and_then(|is_fatal| is_fatal.then_some(LoopEvent::Shutdown))
+                        },
+                    }
                 }
-                LoopEvent::Mailbox(Some(message)) => match message.into_ingress_envelope() {
-                    IngressEnvelope::ReadOnly(message) => {
-                        self.handle_read_only(&args, &mut reads, message);
+            };
+            if let Some(event) = event {
+                match event {
+                    LoopEvent::Shutdown => {
+                        self.shutdown_gracefully(&mut args, &mut reads, "shutdown signal received")
+                            .await;
+                        return;
                     }
-                    IngressEnvelope::ReadWrite(message) => {
+                    LoopEvent::Mailbox(Some(message)) => match message.into_ingress_envelope() {
+                        IngressEnvelope::ReadOnly(message) => {
+                            self.handle_read_only(&args, &mut reads, message);
+                        }
+                        IngressEnvelope::ReadWrite(message) => {
+                            if self.handle_read_write(&mut args, &mut reads, message).await {
+                                return;
+                            }
+                        }
+                    },
+                    LoopEvent::External(Some(message)) => {
                         if self.handle_read_write(&mut args, &mut reads, message).await {
                             return;
                         }
                     }
-                },
-                LoopEvent::External(Some(message)) => {
-                    if self.handle_read_write(&mut args, &mut reads, message).await {
+                    LoopEvent::Mailbox(None) | LoopEvent::External(None) => {
+                        self.shutdown_gracefully(
+                            &mut args,
+                            &mut reads,
+                            "ingress source closed, shutting down actor",
+                        )
+                        .await;
                         return;
                     }
                 }
-                LoopEvent::Mailbox(None) | LoopEvent::External(None) => {
-                    self.shutdown_gracefully(
-                        &mut args,
-                        &mut reads,
-                        "ingress source closed, shutting down actor",
-                    )
-                    .await;
-                    return;
-                }
-            };
+            }
 
             self.actor
                 .postprocess(self.context.as_present_mut(), &mut args)