fix: supervisor deadlock

elfo-core/src/object.rs

@@ -147,13 +147,11 @@ pub(crate) trait GroupHandle: Send + Sync + 'static {
 /// Possible sequences of calls:
 /// * `done()`, if handled by a supervisor
 /// * `empty()`, if no relevant actors in a group
-/// * `visit_last()`, if only one relevant actor in a group
-/// * `visit()`, `visit()`, .., `visit_last()`
+/// * `visit()`, otherwise
 pub trait GroupVisitor {
     fn done(&mut self);
     fn empty(&mut self, envelope: Envelope);
-    fn visit(&mut self, object: &ObjectArc, envelope: &Envelope);
-    fn visit_last(&mut self, object: &ObjectArc, envelope: Envelope);
+    fn visit(&mut self, object: &ObjectArc, envelope: Envelope);
 }
 
 // === SendGroupVisitor ===
@@ -266,12 +264,7 @@ impl GroupVisitor for SendGroupVisitor<'_> {
         self.extra = Some(envelope);
     }
 
-    fn visit(&mut self, object: &ObjectArc, envelope: &Envelope) {
-        let envelope = self.extra.take().unwrap_or_else(|| envelope.duplicate());
-        self.try_send(object, envelope);
-    }
-
-    fn visit_last(&mut self, object: &ObjectArc, envelope: Envelope) {
+    fn visit(&mut self, object: &ObjectArc, envelope: Envelope) {
         self.try_send(object, envelope);
     }
 }
@@ -328,12 +321,7 @@ impl GroupVisitor for TrySendGroupVisitor {
         self.extra = Some(envelope);
     }
 
-    fn visit(&mut self, object: &ObjectArc, envelope: &Envelope) {
-        let envelope = self.extra.take().unwrap_or_else(|| envelope.duplicate());
-        self.try_send(object, envelope);
-    }
-
-    fn visit_last(&mut self, object: &ObjectArc, envelope: Envelope) {
+    fn visit(&mut self, object: &ObjectArc, envelope: Envelope) {
         self.try_send(object, envelope);
     }
 }

elfo-core/src/supervisor.rs

@@ -16,7 +16,7 @@ use crate::{
     actor::{Actor, ActorMeta, ActorStatus},
     config::{AnyConfig, Config, SystemConfig},
     context::Context,
-    envelope::Envelope,
+    envelope::{Envelope, EnvelopeOwned},
     exec::{Exec, ExecResult},
     group::{RestartMode, RestartPolicy, TerminationPolicy},
     message::Request,
@@ -218,22 +218,32 @@ where
 
         match outcome {
             Outcome::Unicast(key) => match get_or_spawn!(self, key) {
-                Some(object) => visitor.visit_last(&object, envelope),
+                Some(object) => visitor.visit(&object, envelope),
                 None => visitor.empty(envelope),
             },
             Outcome::GentleUnicast(key) => match self.objects.get(&key) {
-                Some(object) => visitor.visit_last(&object, envelope),
+                Some(object) => visitor.visit(&object, envelope),
                 None => visitor.empty(envelope),
             },
             Outcome::Multicast(list) => {
-                let iter = list.into_iter().filter_map(|key| get_or_spawn!(self, key));
-                self.visit_multiple(envelope, visitor, iter);
+                let keys = list
+                    .into_iter()
+                    .filter(|key| get_or_spawn!(self, key.clone()).is_some())
+                    .collect();
+                self.visit_multiple(envelope, visitor, keys);
             }
             Outcome::GentleMulticast(list) => {
-                let iter = list.into_iter().filter_map(|key| self.objects.get(&key));
-                self.visit_multiple(envelope, visitor, iter);
+                let keys = list
+                    .into_iter()
+                    .filter(|key| self.objects.get(key).is_some())
+                    .collect();
+                self.visit_multiple(envelope, visitor, keys);
             }
-            Outcome::Broadcast => self.visit_multiple(envelope, visitor, self.objects.iter()),
+            Outcome::Broadcast => self.visit_multiple(
+                envelope,
+                visitor,
+                self.objects.iter().map(|o| o.key().clone()).collect(),
+            ),
             Outcome::Discard => visitor.empty(envelope),
             Outcome::Default => unreachable!("must be altered earlier"),
         }
@@ -243,20 +253,16 @@ where
         &self,
         envelope: Envelope,
         visitor: &mut dyn GroupVisitor,
-        iter: impl Iterator<Item = impl Deref<Target = ObjectArc>>,
+        keys: Vec<R::Key>,
     ) {
-        let mut iter = iter.peekable();
-
-        if iter.peek().is_none() {
-            return visitor.empty(envelope);
-        }
-
-        loop {
-            let object = iter.next().unwrap();
-            if iter.peek().is_none() {
-                return visitor.visit_last(&object, envelope);
+        let len = keys.len();
+        for (index, key) in keys.into_iter().enumerate() {
+            let item = self.objects.get(&key).unwrap();
+            if index + 1 == len {
+                visitor.visit(&item, envelope);
+                break;
             } else {
-                visitor.visit(&object, &envelope);
+                visitor.visit(&item, envelope.duplicate());
             }
         }
     }

elfo-network/src/worker/mod.rs

@@ -585,13 +585,7 @@ impl SocketReader {
                 // TODO: maybe emit some metric?
             }
 
-            fn visit(&mut self, object: &ObjectArc, envelope: &Envelope) {
-                let envelope = envelope.duplicate();
-                self.this
-                    .do_handle_message(self.flows, object, envelope, true);
-            }
-
-            fn visit_last(&mut self, object: &ObjectArc, envelope: Envelope) {
+            fn visit(&mut self, object: &ObjectArc, envelope: Envelope) {
                 self.this
                     .do_handle_message(self.flows, object, envelope, true);
             }