move shard coordinator to use the new cluster singleton system & remo…

…ve `CoordinatorSpawner` actor

coerce/Cargo.toml

@@ -2,7 +2,7 @@
 name = "coerce"
 description = "Async actor runtime and distributed systems framework"
 license = "Apache-2.0"
-version = "0.8.11"
+version = "0.8.12"
 authors = ["Leon Hartley <[email protected]>"]
 edition = "2021"
 readme = "README.md"
@@ -20,6 +20,7 @@ full = [
     "actor-tracing",
     "actor-tracing-info",
     "client-auth-jwt",
+    "singleton",
 ]
 
 remote = [

coerce/src/actor/blocking.rs

@@ -38,9 +38,9 @@ impl ActorSystem {
             Ok(_) => Ok(actor_ref),
             Err(_e) => {
                 error!(
-                    "actor not started, actor_id={}, type={}",
-                    &id,
-                    A::type_name()
+                    actor_id = id.as_ref(),
+                    actor_type = A::type_name(),
+                    "actor not started",
                 );
                 Err(ActorRefErr::ActorStartFailed)
             }

coerce/src/actor/context.rs

@@ -385,27 +385,41 @@ fn on_context_dropped(
         ActorStatus::Started => {
             if let Some(true) = system_terminated {
                 debug!(
-                    "actor (id={}, type={}) has stopped due to system shutdown",
-                    actor_id, actor_type
+                    actor_id = actor_id.as_ref(),
+                    actor_type = actor_type,
+                    "actor stopped due to system shutdown",
                 );
             } else {
-                debug!("actor (id={}) has stopped unexpectedly", actor.0.actor_id());
+                debug!(
+                    actor_id = actor.0.actor_id().as_ref(),
+                    actor_type = actor_type,
+                    "actor stopped unexpectedly"
+                );
             }
         }
 
         ActorStatus::Stopping => {
             if let Some(true) = system_terminated {
-                trace!("actor (id={}) has stopped due to system shutdown", actor_id,);
+                trace!(
+                    actor_id = actor_id.as_ref(),
+                    actor_type = actor_type,
+                    "actor stopped due to system shutdown"
+                );
             } else {
                 debug!(
-                    "actor (id={}) was stopping but did not complete the stop procedure",
-                    actor_id,
+                    actor_id = actor_id.as_ref(),
+                    actor_type = actor_type,
+                    "actor was stopping but did not complete the stop procedure",
                 );
             }
         }
 
         ActorStatus::Stopped => {
-            debug!("actor (id={}) stopped, context dropped", actor_id);
+            debug!(
+                actor_id = actor_id.as_ref(),
+                actor_type = actor_type,
+                "actor stopped, context dropped"
+            );
         }
     }
 

coerce/src/actor/lifecycle.rs

@@ -61,7 +61,7 @@ impl ActorLoop {
             .new_context(system.clone(), Starting, actor_ref.clone().into())
             .with_parent(parent_ref);
 
-        trace!("[{}] starting", ctx.full_path());
+        trace!(actor = ctx.full_path().as_ref(), "actor starting");
 
         actor.started(&mut ctx).await;
         ActorMetrics::incr_actor_created(A::type_name());
@@ -72,7 +72,7 @@ impl ActorLoop {
 
         ctx.set_status(Started);
 
-        trace!("[{}] ready", ctx.full_path());
+        trace!(actor = ctx.full_path().as_ref(), "actor ready");
 
         if let Some(on_start) = on_start.take() {
             let _ = on_start.send(());
@@ -102,7 +102,11 @@ impl ActorLoop {
                     message_type = msg.name(),
                 );
 
-                trace!("[{}] received {}", ctx.full_path(), msg.name(),);
+                trace!(
+                    actor = ctx.full_path().as_ref(),
+                    msg_type = msg.name(),
+                    "actor message received"
+                );
 
                 let handle_fut = msg.handle(&mut actor, &mut ctx);
 
@@ -111,15 +115,19 @@ impl ActorLoop {
 
                 handle_fut.await;
 
-                trace!("[{}] processed {}", ctx.full_path(), msg.name());
+                trace!(
+                    actor = ctx.full_path().as_ref(),
+                    msg_type = msg.name(),
+                    "actor message processed"
+                );
             }
 
             if ctx.get_status() == &Stopping {
                 break;
             }
         }
 
-        trace!("[{}] stopping", ctx.full_path());
+        trace!(actor = ctx.full_path().as_ref(), "actor stopping");
 
         ctx.set_status(Stopping);
 
@@ -141,7 +149,7 @@ async fn actor_stopped<A: Actor>(
     if actor_type.is_tracked() {
         if let Some(system) = system.take() {
             if !system.is_terminated() {
-                trace!("de-registering actor {}", &actor_id);
+                trace!(actor_id = actor_id.as_ref(), "de-registering actor");
 
                 system
                     .scheduler()

coerce/src/remote/cluster/singleton/manager/mod.rs

@@ -6,9 +6,7 @@ use crate::actor::context::ActorContext;
 use crate::actor::message::{
     FromBytes, Handler, Message, MessageUnwrapErr, MessageWrapErr, ToBytes,
 };
-use crate::actor::{
-    Actor, ActorFactory, ActorId, ActorRef, ActorRefErr, IntoActor, LocalActorRef, ToActorId,
-};
+use crate::actor::{Actor, ActorFactory, ActorId, ActorRef, IntoActor, LocalActorRef, ToActorId};
 use crate::remote::cluster::node::NodeSelector;
 use crate::remote::cluster::singleton::factory::SingletonFactory;
 use crate::remote::cluster::singleton::manager::lease::LeaseAck;
@@ -19,7 +17,7 @@ use crate::remote::stream::system::{ClusterEvent, ClusterMemberUp, SystemEvent,
 use crate::remote::system::{NodeId, RemoteActorSystem};
 use crate::remote::RemoteActorRef;
 use std::collections::hash_map::Entry;
-use std::collections::{HashMap, HashSet, VecDeque};
+use std::collections::{HashMap, HashSet};
 use std::fmt::{Debug, Formatter};
 use std::mem;
 use std::time::Duration;
@@ -316,7 +314,7 @@ impl<F: SingletonFactory> Handler<Receive<SystemTopic>> for Manager<F> {
                                 },
                                 ctx,
                             )
-                                .await;
+                            .await;
                         }
                         _ => {}
                     }

coerce/src/remote/cluster/singleton/mod.rs

@@ -124,3 +124,12 @@ pub fn singleton<F: SingletonFactory>(
             actor_type
         ))
 }
+
+impl<A: Actor, F: SingletonFactory<Actor = A>> Clone for Singleton<A, F> {
+    fn clone(&self) -> Self {
+        Self {
+            manager: self.manager.clone(),
+            proxy: self.proxy.clone(),
+        }
+    }
+}

coerce/src/remote/cluster/singleton/proxy/mod.rs

@@ -66,19 +66,25 @@ impl<A: Actor> Handler<SingletonStarted<A>> for Proxy<A> {
 
         match &mut self.state {
             ProxyState::Buffered { request_queue } => {
-                debug!(
-                    buffered_msgs = request_queue.len(),
-                    actor_ref = format!("{}", &actor_ref),
-                    "emitting buffered messages",
-                );
-
-                while let Some(mut buffered) = request_queue.pop_front() {
-                    buffered.send(actor_ref.clone());
+                if request_queue.len() > 0 {
+                    debug!(
+                        buffered_msgs = request_queue.len(),
+                        actor_ref = format!("{}", &actor_ref),
+                        "emitting buffered messages",
+                    );
+
+                    while let Some(mut buffered) = request_queue.pop_front() {
+                        buffered.send(actor_ref.clone());
+                    }
                 }
             }
             _ => {}
         }
 
+        debug!(
+            singleton_actor = format!("{}", actor_ref),
+            "singleton proxy active - singleton started"
+        );
         self.state = ProxyState::Active { actor_ref };
     }
 }

coerce/src/remote/cluster/singleton/proxy/send.rs

@@ -27,6 +27,8 @@ impl<M: Message> Deliver<M> {
         let message = self.message.take().unwrap();
         let result_channel = self.result_channel.take().unwrap();
         tokio::spawn(async move {
+            debug!(msg_type = M::type_name(), "delivering message to singleton");
+
             let res = actor.send(message).await;
             result_channel.send(res)
         });