1use std::any::Any;
101use std::any::TypeId;
102use std::collections::BTreeMap;
103use std::collections::HashMap;
104use std::collections::HashSet;
105use std::collections::VecDeque;
106use std::fmt;
107use std::future::Future;
108use std::ops::Deref;
109use std::panic;
110use std::panic::AssertUnwindSafe;
111use std::panic::Location as PanicLocation;
112use std::pin::Pin;
113use std::ptr;
114use std::sync::Arc;
115use std::sync::Condvar;
116use std::sync::Mutex;
117use std::sync::OnceLock;
118use std::sync::RwLock;
119use std::sync::Weak;
120use std::sync::atomic::AtomicBool;
121use std::sync::atomic::AtomicU64;
122use std::sync::atomic::AtomicUsize;
123use std::sync::atomic::Ordering;
124use std::time::Duration;
125use std::time::Instant;
126use std::time::SystemTime;
127
128use async_trait::async_trait;
129use dashmap::DashMap;
130use dashmap::DashSet;
131use dashmap::mapref::entry::Entry;
132use dashmap::mapref::multiple::RefMulti;
133use enum_as_inner::EnumAsInner;
134use futures::FutureExt;
135use hyperactor_config::Flattrs;
136use hyperactor_telemetry::ActorStatusEvent;
137use hyperactor_telemetry::generate_actor_status_event_id;
138use hyperactor_telemetry::hash_to_u64;
139use hyperactor_telemetry::notify_actor_status_changed;
140use hyperactor_telemetry::notify_message;
141use hyperactor_telemetry::notify_message_status;
142use hyperactor_telemetry::recorder::Recording;
143use serde::Deserialize;
144use serde::Serialize;
145use tokio::sync::Notify;
146use tokio::sync::mpsc;
147use tokio::sync::oneshot;
148use tokio::sync::watch;
149use tokio::task::JoinHandle;
150use tracing::Instrument;
151use tracing::Span;
152use typeuri::Named;
153use uuid::Uuid;
154use wirevalue::TypeInfo;
155
156use crate as hyperactor;
157use crate::Actor;
158use crate::ActorAddr;
159use crate::ActorRef;
160use crate::Addr;
161use crate::Data;
162use crate::Handler;
163use crate::Location;
164use crate::Message;
165use crate::PortAddr;
166use crate::PortRef;
167use crate::ProcAddr;
168use crate::ProcId;
169use crate::RemoteMessage;
170use crate::actor::ActorError;
171use crate::actor::ActorErrorKind;
172use crate::actor::ActorHandle;
173use crate::actor::ActorStatus;
174use crate::actor::ActorStoppingReason;
175use crate::actor::AnyActorHandle;
176use crate::actor::Binds;
177use crate::actor::HandlerInfo;
178use crate::actor::MessageStatusReporting;
179use crate::actor::Referable;
180use crate::actor::RemoteHandles;
181use crate::actor::Signal;
182use crate::actor::StopMode;
183use crate::actor_local::ActorLocalStorage;
184use crate::channel;
185use crate::channel::ChannelAddr;
186use crate::channel::ChannelError;
187use crate::client::Client;
188use crate::client::ClientActor;
189use crate::config;
190use crate::context;
191use crate::context::Mailbox as _;
192use crate::endpoint::Endpoint as _;
193use crate::environment::ActorEnvironment;
194use crate::gateway::Gateway;
195use crate::id::ActorId;
196use crate::id::Label;
197use crate::id::Uid;
198use crate::introspect::IntrospectMessage;
199use crate::introspect::IntrospectResult;
200use crate::mailbox::BoxedMailboxSender;
201use crate::mailbox::DeliveryFailure;
202use crate::mailbox::DialMailboxRouter;
203use crate::mailbox::IntoBoxedMailboxSender as _;
204use crate::mailbox::Mailbox;
205use crate::mailbox::MailboxMuxer;
206use crate::mailbox::MailboxSender;
207use crate::mailbox::MessageEnvelope;
208use crate::mailbox::OncePortHandle;
209use crate::mailbox::OncePortReceiver;
210use crate::mailbox::PortGone;
211use crate::mailbox::PortHandle;
212use crate::mailbox::PortReceiver;
213use crate::mailbox::PortSender as _;
214use crate::mailbox::TransportFailure;
215use crate::mailbox::TransportFailureReason;
216use crate::mailbox::Undeliverable;
217use crate::mailbox::UndeliverableReason;
218use crate::metrics::ACTOR_MESSAGE_HANDLER_DURATION;
219use crate::metrics::ACTOR_MESSAGE_QUEUE_SIZE;
220use crate::metrics::ACTOR_MESSAGES_RECEIVED;
221use crate::port::Port;
222use crate::subject::AsSubject as _;
223
224tokio::task_local! {
225 static CURRENT_TASK_PROC: Proc;
226}
227
228pub const LEGACY_SERVICE_PROC_NAME: &str = "service";
233
234fn wall_clock_epoch_ms() -> u64 {
238 std::time::SystemTime::now()
239 .duration_since(std::time::UNIX_EPOCH)
240 .unwrap_or_default()
241 .as_millis() as u64
242}
243
244pub(crate) struct ProcQueueStats {
251 running_total: AtomicU64,
255 high_water_mark: AtomicU64,
257 last_nonzero_epoch_ms: AtomicU64,
263 clock: fn() -> u64,
266}
267
268impl ProcQueueStats {
269 fn new() -> Self {
270 Self {
271 running_total: AtomicU64::new(0),
272 high_water_mark: AtomicU64::new(0),
273 last_nonzero_epoch_ms: AtomicU64::new(0),
274 clock: wall_clock_epoch_ms,
275 }
276 }
277
278 #[cfg(test)]
280 fn with_clock(clock: fn() -> u64) -> Self {
281 Self {
282 running_total: AtomicU64::new(0),
283 high_water_mark: AtomicU64::new(0),
284 last_nonzero_epoch_ms: AtomicU64::new(0),
285 clock,
286 }
287 }
288
289 fn now_ms(&self) -> u64 {
291 (self.clock)()
292 }
293
294 pub(crate) fn running_total(&self) -> u64 {
296 self.running_total.load(Ordering::Relaxed)
297 }
298
299 pub(crate) fn high_water_mark(&self) -> u64 {
301 self.high_water_mark.load(Ordering::Relaxed)
302 }
303
304 pub(crate) fn last_nonzero_age_ms(&self) -> Option<u64> {
309 let ts = self.last_nonzero_epoch_ms.load(Ordering::Relaxed);
310 if ts == 0 {
311 return None;
312 }
313 Some(self.now_ms().saturating_sub(ts))
314 }
315}
316
317fn account_enqueue(queue_depth: &AtomicU64, proc_stats: &ProcQueueStats, actor_id: &str) {
327 queue_depth.fetch_add(1, Ordering::Relaxed);
328 let new_total = proc_stats.running_total.fetch_add(1, Ordering::Relaxed) + 1;
329 proc_stats
331 .high_water_mark
332 .fetch_max(new_total, Ordering::Relaxed);
333 proc_stats
335 .last_nonzero_epoch_ms
336 .store(proc_stats.now_ms(), Ordering::Relaxed);
337 ACTOR_MESSAGE_QUEUE_SIZE.add(
338 1,
339 hyperactor_telemetry::kv_pairs!("actor_id" => actor_id.to_owned()),
340 );
341}
342
343fn account_dequeue(queue_depth: &AtomicU64, proc_stats: &ProcQueueStats, actor_id: &str) {
353 queue_depth.fetch_sub(1, Ordering::Relaxed);
354 let prev_total = proc_stats.running_total.fetch_sub(1, Ordering::Relaxed);
355 if prev_total > 1 {
359 proc_stats
360 .last_nonzero_epoch_ms
361 .store(proc_stats.now_ms(), Ordering::Relaxed);
362 }
363 ACTOR_MESSAGE_QUEUE_SIZE.add(
364 -1,
365 hyperactor_telemetry::kv_pairs!("actor_id" => actor_id.to_owned()),
366 );
367}
368
369fn account_cancel_enqueue(queue_depth: &AtomicU64, proc_stats: &ProcQueueStats, actor_id: &str) {
379 queue_depth.fetch_sub(1, Ordering::Relaxed);
380 proc_stats.running_total.fetch_sub(1, Ordering::Relaxed);
381 ACTOR_MESSAGE_QUEUE_SIZE.add(
382 -1,
383 hyperactor_telemetry::kv_pairs!("actor_id" => actor_id.to_owned()),
384 );
385}
386
387use crate::monitor::EndpointId;
388use crate::ordering::DeliveryProgress;
389use crate::ordering::SEQ_INFO;
390use crate::ordering::SeqInfo;
391use crate::ordering::Sequencer;
392use crate::panic_handler;
393use crate::sequenced::SequencedEnvelope;
394use crate::sequenced::SequencedReceiver;
395use crate::sequenced::sequenced_unbounded_with_buffering;
396use crate::supervision::ActorSupervisionEvent;
397
398#[derive(Clone)]
405pub struct Proc {
406 inner: Arc<ProcState>,
407}
408
409#[cfg(not(hyperactor_verify_auto_traits))]
422mod _send_sync_shortcut {
423 use super::*;
424
425 unsafe impl Send for ProcState {}
427 unsafe impl Sync for ProcState {}
429 unsafe impl Send for InstanceCellState {}
431 unsafe impl Sync for InstanceCellState {}
433}
434
435const _: fn() = || {
436 fn assert<T: Send + Sync + ?Sized>() {}
437 assert::<ProcState>();
438 assert::<InstanceCellState>();
439};
440
441impl fmt::Debug for Proc {
442 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
443 f.debug_struct("Proc")
444 .field("proc_id", &self.inner.proc_id)
445 .finish()
446 }
447}
448
449struct ProcState {
450 proc_id: ProcId,
453
454 gateway: Gateway,
456
457 proc_muxer: MailboxMuxer,
460
461 reserved_roots: DashSet<crate::id::Uid>,
465
466 reserved_child_uids: DashSet<crate::id::Uid>,
469
470 actors: DashMap<ActorId, ActorEntry>,
475
476 root_instances: DashMap<ActorId, WeakInstanceCell>,
478
479 queue_stats: Arc<ProcQueueStats>,
484
485 terminated_snapshots: DashMap<ActorId, TerminatedSnapshot>,
490
491 supervision_coordinator_port: OnceLock<PortRef<ActorSupervisionEvent>>,
496
497 supervision_coordinator_actor_id: OnceLock<ActorAddr>,
500
501 mailbox_server_handle: std::sync::Mutex<Option<crate::mailbox::MailboxServerHandle>>,
506
507 _attached_proc_guard: OnceLock<crate::gateway::AttachedProcGuard>,
515}
516
517#[derive(EnumAsInner)]
519enum ActorEntry {
520 Pending(Vec<(Instant, StatusMessage)>),
524 Running {
527 cell: WeakInstanceCell,
528 status: watch::Receiver<ActorStatus>,
530 },
531 Tombstoned(ActorStatus),
533}
534
535impl ActorEntry {
536 fn is_running_instance(&self, state: &InstanceCellState) -> bool {
539 self.as_running()
540 .is_some_and(|(cell, _)| ptr::eq(cell.inner.as_ptr(), state))
541 }
542}
543
544struct TerminatedSnapshot {
545 actor_addr: ActorAddr,
546 payload: crate::introspect::IntrospectResult,
547}
548
549#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Named)]
551#[expect(
552 clippy::large_enum_variant,
553 reason = "keep the existing GetStatus wire representation stable while adding smaller subscription requests"
554)]
555pub enum StatusMessage {
556 GetStatus {
558 reply: crate::OncePortRef<Option<ActorStatus>>,
561 delivery: Option<(
563 Uuid,
564 EndpointId,
565 crate::OncePortRef<DeliveryProgressResponse>,
566 )>,
567 },
568 SubscribeTerminal {
578 subscription_id: Uuid,
580 subscriber: PortRef<Option<ActorStatus>>,
583 },
584 UnsubscribeTerminal {
587 subscription_id: Uuid,
589 },
590}
591wirevalue::register_type!(StatusMessage);
592
593const MAX_TERMINAL_STATUS_SUBSCRIBERS: usize = 100;
594
595#[derive(Default)]
597struct TerminalSubscriptions {
598 active: VecDeque<(Uuid, PortRef<Option<ActorStatus>>)>,
601 cancelled: HashSet<Uuid>,
603}
604
605#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Named)]
607pub enum DeliveryProgressResponse {
608 ActorGone,
610 Incomplete,
613 Progress(DeliveryProgress),
615}
616wirevalue::register_type!(DeliveryProgressResponse);
617
618struct StatusSender(WeakProc);
619
620impl StatusSender {
621 fn new(weak_proc: WeakProc) -> Self {
622 Self(weak_proc)
623 }
624}
625
626#[async_trait]
627impl MailboxSender for StatusSender {
628 fn post_unchecked(
629 &self,
630 envelope: MessageEnvelope,
631 return_handle: PortHandle<Undeliverable<MessageEnvelope>>,
632 ) {
633 let message = match envelope.deserialized() {
634 Ok(message) => message,
635 Err(err) => {
636 let target = envelope.dest().clone();
637 let failure =
638 DeliveryFailure::new(UndeliverableReason::Transport(TransportFailure::new(
639 target,
640 TransportFailureReason::LinkUnavailable(format!(
641 "status message deserialization failed: {err}"
642 )),
643 )));
644 envelope.undeliverable(failure, return_handle);
645 return;
646 }
647 };
648
649 let Some(proc) = self.0.upgrade() else {
650 let failure = DeliveryFailure::new(UndeliverableReason::PortGone(PortGone::new(
651 envelope.dest().clone(),
652 envelope.data().typename().map(str::to_string),
653 )));
654 envelope.undeliverable(failure, return_handle);
655 return;
656 };
657
658 proc.post_status(envelope.dest().actor_id().clone(), message);
659 }
660}
661
662impl Drop for ProcState {
663 fn drop(&mut self) {
664 let proc_addr = self.proc_addr();
668 tracing::info!(
669 subject = %proc_addr.subject(),
670 name = "ProcStatus",
671 status = "Dropped"
672 );
673 }
674}
675
676impl ProcState {
677 fn default_location(&self) -> Location {
679 self.gateway.default_location()
680 }
681
682 fn set_default_location(&self, location: Location) {
683 self.gateway.set_default_location(location)
684 }
685
686 fn proc_addr(&self) -> ProcAddr {
687 ProcAddr::new(self.proc_id.clone(), self.gateway.default_location())
688 }
689}
690
691pub struct ActorInstance<A: Actor> {
696 pub instance: Instance<A>,
698 pub handle: ActorHandle<A>,
700 pub supervision: mpsc::UnboundedReceiver<ActorSupervisionEvent>,
702 pub signal: mpsc::UnboundedReceiver<Signal>,
704 pub work: ActorWorkReceiver<A>,
706}
707
708pub struct ActorWorkReceiver<A: Actor> {
710 inner: SequencedReceiver<SequencedEnvelope<WorkCell<A>>>,
711}
712
713impl<A: Actor> fmt::Debug for ActorWorkReceiver<A> {
714 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
715 f.debug_struct("ActorWorkReceiver").finish_non_exhaustive()
716 }
717}
718
719impl<A: Actor> ActorWorkReceiver<A> {
720 fn new(inner: SequencedReceiver<SequencedEnvelope<WorkCell<A>>>) -> Self {
721 Self { inner }
722 }
723
724 pub async fn recv(&mut self) -> Option<WorkCell<A>> {
726 self.inner.recv().await
727 }
728
729 pub fn try_recv(&mut self) -> Result<WorkCell<A>, mpsc::error::TryRecvError> {
731 self.inner.try_recv()
732 }
733}
734
735pub struct Builder<State = GlobalGateway> {
737 proc_id: Option<ProcId>,
738 state: State,
739}
740
741pub struct GlobalGateway;
743
744pub struct SharedGateway {
746 gateway: Gateway,
747}
748
749pub struct PrivateGateway {
751 forwarder: BoxedMailboxSender,
752}
753
754impl Builder<GlobalGateway> {
755 pub fn new() -> Self {
757 Self {
758 proc_id: None,
759 state: GlobalGateway,
760 }
761 }
762
763 pub fn shared_gateway(self, gateway: Gateway) -> Builder<SharedGateway> {
765 Builder {
766 proc_id: self.proc_id,
767 state: SharedGateway { gateway },
768 }
769 }
770
771 pub fn private_gateway(self, forwarder: BoxedMailboxSender) -> Builder<PrivateGateway> {
773 Builder {
774 proc_id: self.proc_id,
775 state: PrivateGateway { forwarder },
776 }
777 }
778
779 pub fn build(self) -> Result<Proc, anyhow::Error> {
787 let proc_id = self
788 .proc_id
789 .unwrap_or_else(|| ProcId::new(Uid::anonymous(), None));
790 if is_legacy_pseudo_singleton_proc_id(&proc_id) {
791 anyhow::bail!(
792 "legacy pseudo-singleton proc id '{}' is reserved for host-scoped construction; use Builder::shared_gateway",
793 proc_id
794 );
795 }
796 Ok(Proc::from_parts_unchecked(
797 proc_id,
798 Gateway::global().clone(),
799 ))
800 }
801}
802
803impl<State> Builder<State> {
804 pub fn proc_id(mut self, proc_id: ProcId) -> Self {
806 self.proc_id = Some(proc_id);
807 self
808 }
809
810 fn build_proc(proc_id: Option<ProcId>, gateway: Gateway) -> Result<Proc, anyhow::Error> {
811 let proc_id = proc_id.unwrap_or_else(ProcId::anonymous);
812 Ok(Proc::from_parts_unchecked(proc_id, gateway))
813 }
814}
815
816impl Builder<SharedGateway> {
817 pub fn build(self) -> Result<Proc, anyhow::Error> {
821 let Builder {
822 proc_id,
823 state: SharedGateway { gateway },
824 } = self;
825 Self::build_proc(proc_id, gateway)
826 }
827}
828
829impl Builder<PrivateGateway> {
830 pub fn build(self) -> Result<Proc, anyhow::Error> {
832 let Builder {
833 proc_id,
834 state: PrivateGateway { forwarder },
835 } = self;
836 let gateway = Gateway::configured(channel::reserve_local_addr().into(), forwarder);
837 Self::build_proc(proc_id, gateway)
838 }
839}
840
841impl Proc {
842 fn from_parts_unchecked(proc_id: ProcId, gateway: Gateway) -> Self {
843 let proc_addr = ProcAddr::new(proc_id.clone(), gateway.default_location());
844 tracing::info!(
845 subject = %proc_addr.subject(),
846 name = "ProcStatus",
847 status = "Created"
848 );
849
850 let proc = Self {
851 inner: Arc::new(ProcState {
852 proc_id: proc_id.clone(),
853 gateway: gateway.clone(),
854 proc_muxer: MailboxMuxer::new(),
855 reserved_roots: DashSet::new(),
856 reserved_child_uids: DashSet::new(),
857 actors: DashMap::new(),
858 root_instances: DashMap::new(),
859 queue_stats: Arc::new(ProcQueueStats::new()),
860 terminated_snapshots: DashMap::new(),
861 supervision_coordinator_port: OnceLock::new(),
862 supervision_coordinator_actor_id: OnceLock::new(),
863 mailbox_server_handle: std::sync::Mutex::new(None),
864 _attached_proc_guard: OnceLock::new(),
865 }),
866 };
867 let bound = proc
868 .inner
869 .proc_muxer
870 .bind_status(StatusSender::new(proc.downgrade()));
871 assert!(
872 bound,
873 "fresh proc muxer must not have a status control port"
874 );
875 let guard = gateway.attach_proc(&proc);
879 proc.inner
880 ._attached_proc_guard
881 .set(guard)
882 .expect("fresh ProcState's attached-proc guard slot is empty");
883 proc
884 }
885
886 fn post_status(&self, actor_id: ActorId, message: StatusMessage) {
894 let timeout = hyperactor_config::global::get(config::PENDING_ACTOR_DELIVERY_TIMEOUT);
895 let deadline = Instant::now() + timeout;
896 match self.inner.actors.entry(actor_id.clone()) {
897 Entry::Occupied(mut entry) => {
898 if let Some(pending) = entry.get_mut().as_pending_mut() {
899 return pending.push((deadline, message));
900 }
901 }
902 Entry::Vacant(entry) if !timeout.is_zero() => {
903 entry.insert(ActorEntry::Pending(vec![(deadline, message)]));
904 crate::init::get_runtime().spawn(Self::expire_pending_status(
905 self.downgrade(),
906 actor_id,
907 deadline,
908 ));
909 return;
910 }
911 Entry::Vacant(_) => {}
912 }
913 self.handle_status(&actor_id, message);
914 }
915
916 async fn expire_pending_status(proc: WeakProc, actor_id: ActorId, mut deadline: Instant) {
920 loop {
921 tokio::time::sleep_until(deadline.into()).await;
922 let Some(proc) = proc.upgrade() else {
923 return;
924 };
925 let (expired, next) = {
926 let Entry::Occupied(mut entry) = proc.inner.actors.entry(actor_id.clone()) else {
927 return;
928 };
929 let Some(pending) = entry.get_mut().as_pending_mut() else {
930 return;
931 };
932 let now = Instant::now();
933 let (expired, remaining): (Vec<_>, Vec<_>) = std::mem::take(pending)
934 .into_iter()
935 .partition(|(deadline, _)| *deadline <= now);
936 let next = remaining.iter().map(|(deadline, _)| *deadline).min();
937 if next.is_some() {
938 *pending = remaining;
939 } else {
940 entry.remove();
941 }
942 (expired, next)
943 };
944 for (_, message) in expired {
945 proc.handle_status(&actor_id, message);
946 }
947 let Some(next) = next else {
948 return;
949 };
950 deadline = next;
951 }
952 }
953
954 fn handle_status(&self, actor_id: &ActorId, message: StatusMessage) {
955 match message {
956 StatusMessage::GetStatus { reply, delivery } => {
957 let status = self.status_for_actor(actor_id);
958 if let Err(err) = self.serialize_and_send_once(
959 reply,
960 status,
961 crate::mailbox::monitored_return_handle(),
962 ) {
963 tracing::error!("status reply failed: {err}");
964 }
965 if let Some((session_id, destination, reply)) = delivery {
966 let progress =
967 self.delivery_progress_for_actor(actor_id, session_id, &destination);
968 if let Err(err) = self.serialize_and_send_once(
969 reply,
970 progress,
971 crate::mailbox::monitored_return_handle(),
972 ) {
973 tracing::error!("delivery progress reply failed: {err}");
974 }
975 }
976 }
977 StatusMessage::SubscribeTerminal {
978 subscription_id,
979 subscriber,
980 } => {
981 self.subscribe_terminal_status(actor_id, subscription_id, subscriber);
982 }
983 StatusMessage::UnsubscribeTerminal { subscription_id } => {
984 self.unsubscribe_terminal_status(actor_id, subscription_id);
985 }
986 }
987 }
988
989 fn status_for_actor(&self, actor_id: &ActorId) -> Option<ActorStatus> {
990 match self.inner.actors.get(actor_id)?.value() {
991 ActorEntry::Pending(_) => None,
992 ActorEntry::Running { status, .. } => Some(status.borrow().clone()),
993 ActorEntry::Tombstoned(status) => Some(status.clone()),
994 }
995 }
996
997 fn delivery_progress_for_actor(
998 &self,
999 actor_id: &ActorId,
1000 session_id: Uuid,
1001 destination: &EndpointId,
1002 ) -> DeliveryProgressResponse {
1003 let Some(cell) = self.get_instance_by_id(actor_id) else {
1004 return DeliveryProgressResponse::ActorGone;
1005 };
1006
1007 match cell.delivery_progress_for_session(destination, session_id) {
1008 Some(progress) => DeliveryProgressResponse::Progress(progress),
1009 None => DeliveryProgressResponse::Incomplete,
1010 }
1011 }
1012
1013 fn send_terminal_status(
1014 &self,
1015 actor_addr: &ActorAddr,
1016 subscriber: PortRef<Option<ActorStatus>>,
1017 status: Option<ActorStatus>,
1018 ) {
1019 let envelope = MessageEnvelope::serialize(
1020 actor_addr.clone(),
1021 subscriber.into_port_addr(),
1022 &status,
1023 Flattrs::new(),
1024 );
1025 let Ok(mut envelope) = envelope else {
1026 tracing::error!(actor_id = %actor_addr, "failed to serialize terminal status notification");
1027 return;
1028 };
1029 envelope.set_header(SEQ_INFO, SeqInfo::Direct);
1030 envelope.set_return_undeliverable(false);
1034 self.post(envelope, crate::mailbox::monitored_return_handle());
1035 }
1036
1037 fn subscribe_terminal_status(
1038 &self,
1039 actor_id: &ActorId,
1040 subscription_id: Uuid,
1041 subscriber: PortRef<Option<ActorStatus>>,
1042 ) {
1043 if let Some(cell) = self.get_instance_by_id(actor_id) {
1044 cell.subscribe_terminal_status(subscription_id, subscriber);
1045 return;
1046 }
1047
1048 let status = self.status_for_actor(actor_id);
1049 let actor_addr = ActorAddr::new(actor_id.clone(), self.default_location());
1050 self.send_terminal_status(&actor_addr, subscriber, status);
1051 }
1052
1053 fn unsubscribe_terminal_status(&self, actor_id: &ActorId, subscription_id: Uuid) {
1054 if let Some(cell) = self.get_instance_by_id(actor_id) {
1055 cell.unsubscribe_terminal_status(subscription_id);
1056 }
1057 }
1058
1059 #[cfg(test)]
1060 pub(crate) fn terminal_status_subscriber_count(&self, actor_id: &ActorId) -> usize {
1061 self.get_instance_by_id(actor_id)
1062 .map_or(0, |cell| cell.terminal_status_subscriber_count())
1063 }
1064
1065 fn from_parts(proc_id: ProcId, gateway: Gateway) -> Self {
1066 assert_not_legacy_pseudo_singleton_proc_id(&proc_id);
1067 Self::from_parts_unchecked(proc_id, gateway)
1068 }
1069
1070 pub fn legacy_service_pseudo_singleton(
1073 addr: ChannelAddr,
1074 forwarder: BoxedMailboxSender,
1075 ) -> Self {
1076 Self::legacy_service_pseudo_singleton_on_gateway(Gateway::configured(
1077 addr.into(),
1078 forwarder,
1079 ))
1080 }
1081
1082 pub fn legacy_service_pseudo_singleton_on_gateway(gateway: Gateway) -> Self {
1085 Self::legacy_pseudo_singleton_on_gateway(LEGACY_SERVICE_PROC_NAME, gateway)
1086 }
1087
1088 fn legacy_pseudo_singleton_on_gateway(name: &'static str, gateway: Gateway) -> Self {
1089 let proc_id = ProcId::singleton(Label::strip(name));
1090 Self::from_parts_unchecked(proc_id, gateway)
1091 }
1092
1093 pub fn anonymous() -> Self {
1095 Self::builder()
1096 .build()
1097 .expect("anonymous proc builder is valid")
1098 }
1099
1100 pub fn instance(label: impl AsRef<str>) -> Self {
1102 Self::builder()
1103 .proc_id(ProcId::instance(Label::strip(label.as_ref())))
1104 .build()
1105 .expect("instance proc builder is valid")
1106 }
1107
1108 pub fn singleton(name: impl AsRef<str>) -> Self {
1110 Self::builder()
1111 .proc_id(ProcId::singleton(Label::strip(name.as_ref())))
1112 .build()
1113 .expect("singleton proc builder is valid")
1114 }
1115
1116 pub fn isolated() -> Self {
1118 Self::builder()
1119 .shared_gateway(Gateway::isolated())
1120 .build()
1121 .expect("isolated proc builder is valid")
1122 }
1123
1124 pub fn builder() -> Builder {
1126 Builder::new()
1127 }
1128
1129 pub fn configured(proc_id: impl Into<ProcAddr>, forwarder: BoxedMailboxSender) -> Self {
1131 let proc_addr = proc_id.into();
1132 Self::from_parts(
1133 proc_addr.id().clone(),
1134 Gateway::configured(proc_addr.location().clone(), forwarder),
1135 )
1136 }
1137
1138 pub fn direct(addr: ChannelAddr, name: String) -> Result<Self, ChannelError> {
1143 let (addr, rx) = channel::serve(addr)?;
1144 let proc_id = ProcAddr::instance(addr, name);
1145 let proc = Self::builder()
1146 .proc_id(proc_id.id().clone())
1147 .shared_gateway(Gateway::configured(
1148 proc_id.location().clone(),
1149 DialMailboxRouter::new().into_boxed(),
1150 ))
1151 .build()
1152 .expect("direct proc builder is valid");
1153 let handle = proc.gateway().serve_rx(rx);
1154 *proc.inner.mailbox_server_handle.lock().unwrap() = Some(handle);
1155 Ok(proc)
1156 }
1157
1158 pub fn set_supervision_coordinator(
1161 &self,
1162 port: PortRef<ActorSupervisionEvent>,
1163 ) -> Result<(), anyhow::Error> {
1164 let actor_ref: ActorAddr = port.port_addr().actor_addr();
1165 self.state()
1166 .supervision_coordinator_port
1167 .set(port)
1168 .map_err(|existing| anyhow::anyhow!("coordinator port is already set to {existing}"))?;
1169 let _ = self.state().supervision_coordinator_actor_id.set(actor_ref);
1170 Ok(())
1171 }
1172
1173 pub fn supervision_coordinator_actor_addr(&self) -> Option<&ActorAddr> {
1176 self.state().supervision_coordinator_actor_id.get()
1177 }
1178
1179 pub fn handle_unhandled_supervision_event(
1182 &self,
1183 cx: &impl context::Actor,
1184 event: ActorSupervisionEvent,
1185 ) {
1186 let result = match self.state().supervision_coordinator_port.get() {
1187 Some(port) => {
1188 port.post(cx, event.clone());
1189 Ok(())
1190 }
1191 None => {
1192 if !event.is_error() {
1193 return;
1196 }
1197 Err(anyhow::anyhow!(
1198 "coordinator port is not set for proc {}",
1199 self.proc_addr(),
1200 ))
1201 }
1202 };
1203 if let Err(err) = result {
1204 if !event.is_error() {
1205 tracing::debug!(
1208 subject = %self.proc_addr().subject(),
1209 "dropping non-error supervision event {}: {:?}",
1210 event,
1211 err
1212 );
1213 return;
1214 }
1215 tracing::error!(
1216 subject = %self.proc_addr().subject(),
1217 "could not propagate supervision event {} due to error: {:?}: crashing",
1218 event,
1219 err
1220 );
1221
1222 std::process::exit(1);
1223 }
1224 }
1225
1226 pub fn proc_id(&self) -> &ProcId {
1228 &self.state().proc_id
1229 }
1230
1231 pub fn default_location(&self) -> Location {
1233 self.state().default_location()
1234 }
1235
1236 pub fn set_default_location(&self, location: Location) {
1238 self.state().set_default_location(location)
1239 }
1240
1241 pub fn proc_addr(&self) -> ProcAddr {
1243 self.state().proc_addr()
1244 }
1245
1246 pub fn gateway(&self) -> Gateway {
1248 self.state().gateway.clone()
1249 }
1250
1251 pub fn global() -> Self {
1253 static GLOBAL_PROC: OnceLock<Proc> = OnceLock::new();
1254 GLOBAL_PROC
1255 .get_or_init(|| {
1256 let label = global_proc_label();
1257 Proc::instance(label.as_str())
1258 })
1259 .clone()
1260 }
1261
1262 pub fn current() -> Self {
1267 CURRENT_TASK_PROC
1268 .try_with(Clone::clone)
1269 .unwrap_or_else(|_| Self::global())
1270 }
1271
1272 async fn with_current<F>(&self, future: F) -> F::Output
1273 where
1274 F: Future,
1275 {
1276 CURRENT_TASK_PROC.scope(self.clone(), future).await
1277 }
1278
1279 pub fn forwarder(&self) -> BoxedMailboxSender {
1282 self.state().gateway.forwarder()
1283 }
1284
1285 pub fn muxer(&self) -> &MailboxMuxer {
1288 &self.inner.proc_muxer
1289 }
1290
1291 fn state(&self) -> &ProcState {
1293 self.inner.as_ref()
1294 }
1295
1296 pub fn attach(&self, name: &str) -> Result<Mailbox, anyhow::Error> {
1298 let actor_id: ActorAddr = self.allocate_root_id(name)?;
1299 Ok(self.bind_mailbox(actor_id))
1300 }
1301
1302 pub fn attach_child(&self, parent_id: &ActorAddr) -> Result<Mailbox, anyhow::Error> {
1304 let actor_id: ActorAddr = self.allocate_anonymous_child_id(parent_id);
1305 Ok(self.bind_mailbox(actor_id))
1306 }
1307
1308 fn bind_mailbox(&self, actor_id: ActorAddr) -> Mailbox {
1310 let mbox = Mailbox::new(actor_id);
1311
1312 self.state().proc_muxer.bind_mailbox(mbox.clone());
1315 mbox
1316 }
1317
1318 pub fn attach_actor<R, M>(
1321 &self,
1322 name: &str,
1323 ) -> Result<(Client, ActorRef<R>, PortReceiver<M>), anyhow::Error>
1324 where
1325 M: RemoteMessage,
1326 R: Referable + RemoteHandles<M>,
1327 {
1328 let client = self.client(name);
1329 let (_handle, rx) = client.bind_handler_port::<M>();
1330 let actor_ref = ActorRef::attest(client.self_addr().clone());
1331 Ok((client, actor_ref, rx))
1332 }
1333
1334 pub fn spawn<A: Actor>(&self, actor: A) -> ActorHandle<A> {
1336 let actor_id: ActorAddr = self.allocate_root_type::<A>();
1337 self.spawn_inner(actor_id, actor, None, ActorEnvironment::default())
1338 }
1339
1340 pub fn spawn_with_label<A: Actor>(&self, label: &str, actor: A) -> ActorHandle<A> {
1345 let actor_id: ActorAddr = self.allocate_root_label(label);
1346 self.spawn_inner(actor_id, actor, None, ActorEnvironment::default())
1347 }
1348
1349 pub fn spawn_with_uid<A: Actor>(
1356 &self,
1357 uid: Uid,
1358 actor: A,
1359 ) -> Result<ActorHandle<A>, anyhow::Error> {
1360 self.spawn_with_uid_in_environment(uid, actor, ActorEnvironment::default())
1361 }
1362
1363 pub fn spawn_bound_with_uid<A: Actor + Binds<A>>(
1372 &self,
1373 uid: Uid,
1374 actor: A,
1375 ) -> Result<ActorHandle<A>, anyhow::Error> {
1376 self.spawn_bound_with_uid_in_environment(uid, actor, ActorEnvironment::default())
1377 }
1378
1379 pub(crate) fn spawn_with_uid_in_environment<A: Actor>(
1387 &self,
1388 uid: Uid,
1389 actor: A,
1390 environment: ActorEnvironment,
1391 ) -> Result<ActorHandle<A>, anyhow::Error> {
1392 let actor_id: ActorAddr = self.allocate_root_uid(uid)?;
1393 Ok(self.spawn_inner(actor_id, actor, None, environment))
1394 }
1395
1396 pub(crate) fn spawn_bound_with_uid_in_environment<A: Actor + Binds<A>>(
1399 &self,
1400 uid: Uid,
1401 actor: A,
1402 environment: ActorEnvironment,
1403 ) -> Result<ActorHandle<A>, anyhow::Error> {
1404 let actor_id: ActorAddr = self.allocate_root_uid(uid)?;
1405 Ok(self.spawn_bound_inner(actor_id, actor, None, environment))
1406 }
1407
1408 fn spawn_inner<A: Actor>(
1411 &self,
1412 actor_id: ActorAddr,
1413 actor: A,
1414 parent: Option<InstanceCell>,
1415 environment: ActorEnvironment,
1416 ) -> ActorHandle<A> {
1417 let (instance, receivers) =
1418 Instance::new(self.clone(), actor_id, false, parent, environment);
1419 instance.start(actor, receivers)
1420 }
1421
1422 fn spawn_bound_inner<A: Actor + Binds<A>>(
1426 &self,
1427 actor_id: ActorAddr,
1428 actor: A,
1429 parent: Option<InstanceCell>,
1430 environment: ActorEnvironment,
1431 ) -> ActorHandle<A> {
1432 let (instance, receivers) =
1433 Instance::new_unpublished(self.clone(), actor_id, false, parent, environment);
1434 instance.inner.cell.bind::<A, A>(&instance.inner.ports);
1435 instance.publish();
1436 instance.start(actor, receivers)
1437 }
1438
1439 pub fn client(&self, label: &str) -> Client {
1444 let actor_id = self.allocate_client_id(label);
1445 let (instance, _receivers) = Instance::<ClientActor>::new(
1446 self.clone(),
1447 actor_id,
1448 false,
1449 None,
1450 ActorEnvironment::default(),
1451 );
1452 instance.change_status(ActorStatus::Client);
1453 Client::new(instance)
1454 }
1455
1456 pub fn introspectable_instance(
1469 &self,
1470 name: &str,
1471 ) -> Result<(Instance<()>, ActorHandle<()>), anyhow::Error> {
1472 let actor_id: ActorAddr = self.allocate_root_id(name)?;
1473 let (instance, receivers) = Instance::new(
1474 self.clone(),
1475 actor_id,
1476 false,
1477 None,
1478 ActorEnvironment::default(),
1479 );
1480 let handle = ActorHandle::new(instance.inner.cell.clone(), instance.inner.ports.clone());
1481 instance.change_status(ActorStatus::Client);
1482 instance.spawn_detached_introspect(receivers.introspect);
1483 Ok((instance, handle))
1484 }
1485
1486 pub fn actor_instance<A: Actor>(&self, name: &str) -> Result<ActorInstance<A>, anyhow::Error> {
1492 self.actor_instance_in_environment(name, ActorEnvironment::default())
1493 }
1494
1495 pub fn actor_instance_in_environment<A: Actor>(
1501 &self,
1502 name: &str,
1503 environment: ActorEnvironment,
1504 ) -> Result<ActorInstance<A>, anyhow::Error> {
1505 let actor_id: ActorAddr = self.allocate_root_id(name)?;
1506 let span = tracing::debug_span!(
1507 "actor_instance",
1508 subject = %actor_id.subject(),
1509 );
1510 let _guard = span.enter();
1511 let (instance, receivers) =
1512 Instance::new(self.clone(), actor_id.clone(), false, None, environment);
1513 let handle = ActorHandle::new(instance.inner.cell.clone(), instance.inner.ports.clone());
1514 instance.change_status(ActorStatus::Client);
1515
1516 instance.spawn_detached_introspect(receivers.introspect);
1517
1518 let (signal_rx, supervision_rx) = receivers.actor_loop.unwrap();
1519 Ok(ActorInstance {
1520 instance,
1521 handle,
1522 supervision: supervision_rx,
1523 signal: signal_rx,
1524 work: receivers.work,
1525 })
1526 }
1527
1528 pub fn traverse<F>(&self, f: &mut F)
1530 where
1531 F: FnMut(&InstanceCell, usize),
1532 {
1533 for cell in self.root_instances() {
1534 cell.traverse(f);
1535 }
1536 }
1537
1538 pub fn queue_depth_total(&self) -> u64 {
1540 self.state().queue_stats.running_total()
1541 }
1542
1543 pub fn queue_depth_high_water_mark(&self) -> u64 {
1545 self.state().queue_stats.high_water_mark()
1546 }
1547
1548 pub fn last_nonzero_queue_depth_age_ms(&self) -> Option<u64> {
1550 self.state().queue_stats.last_nonzero_age_ms()
1551 }
1552
1553 pub fn get_instance(&self, actor_id: &ActorAddr) -> Option<InstanceCell> {
1555 self.get_instance_by_id(actor_id.id())
1556 }
1557
1558 pub fn get_instance_by_id(&self, actor_id: &ActorId) -> Option<InstanceCell> {
1560 let entry = self.state().actors.get(actor_id)?;
1561 let (cell, _) = entry.as_running()?;
1562 cell.upgrade()
1563 }
1564
1565 fn root_instances(&self) -> Vec<InstanceCell> {
1567 self.state()
1568 .root_instances
1569 .iter()
1570 .filter_map(|entry| entry.value().upgrade())
1571 .collect()
1572 }
1573
1574 pub fn all_actor_ids(&self) -> Vec<ActorAddr> {
1583 let cells: Vec<_> = self
1586 .state()
1587 .actors
1588 .iter()
1589 .filter_map(|entry| {
1590 let (cell, status) = entry.as_running()?;
1591 if !status.borrow().is_terminal() {
1592 cell.upgrade()
1593 } else {
1594 None
1595 }
1596 })
1597 .collect();
1598 cells.iter().map(|cell| cell.actor_addr().clone()).collect()
1599 }
1600
1601 pub fn all_instance_keys(&self) -> Vec<ActorId> {
1611 self.state()
1612 .actors
1613 .iter()
1614 .filter(|entry| entry.is_running())
1615 .map(|entry| entry.key().clone())
1616 .collect()
1617 }
1618
1619 pub fn terminated_snapshot(
1621 &self,
1622 actor_id: &ActorAddr,
1623 ) -> Option<crate::introspect::IntrospectResult> {
1624 self.state()
1625 .terminated_snapshots
1626 .get(actor_id.id())
1627 .map(|entry| entry.value().payload.clone())
1628 }
1629
1630 pub fn all_terminated_actor_ids(&self) -> Vec<ActorAddr> {
1632 self.state()
1633 .terminated_snapshots
1634 .iter()
1635 .map(|entry| entry.value().actor_addr.clone())
1636 .collect()
1637 }
1638
1639 fn child_instance(&self, parent: InstanceCell) -> (Instance<()>, ActorHandle<()>) {
1641 let actor_id = self.allocate_anonymous_child_id(parent.actor_addr());
1642 let span = tracing::debug_span!(
1643 "child_actor_instance",
1644 subject = %actor_id.subject(),
1645 );
1646 let _guard = span.enter();
1647
1648 let environment = parent.actor_environment().clone();
1649 let (instance, _receivers) =
1650 Instance::new(self.clone(), actor_id, false, Some(parent), environment);
1651 let handle = ActorHandle::new(instance.inner.cell.clone(), instance.inner.ports.clone());
1654 instance.change_status(ActorStatus::Client);
1655 (instance, handle)
1656 }
1657
1658 pub(crate) fn spawn_child<A: Actor>(&self, parent: InstanceCell, actor: A) -> ActorHandle<A> {
1664 let actor_id = self.allocate_child_id::<A>(parent.actor_addr());
1665 let environment = parent.actor_environment().clone();
1666 self.spawn_inner(actor_id, actor, Some(parent), environment)
1667 }
1668
1669 pub(crate) fn spawn_child_with_uid<A: Actor>(
1671 &self,
1672 parent: InstanceCell,
1673 uid: Uid,
1674 actor: A,
1675 ) -> Result<ActorHandle<A>, anyhow::Error> {
1676 let environment = parent.actor_environment().clone();
1677 self.spawn_child_with_uid_in_environment(parent, uid, actor, environment)
1678 }
1679
1680 pub(crate) fn spawn_child_with_uid_in_environment<A: Actor>(
1683 &self,
1684 parent: InstanceCell,
1685 uid: Uid,
1686 actor: A,
1687 environment: ActorEnvironment,
1688 ) -> Result<ActorHandle<A>, anyhow::Error> {
1689 let actor_id = self.ensure_child_uid(parent.actor_addr(), uid)?;
1690 Ok(self.spawn_inner(actor_id, actor, Some(parent), environment))
1691 }
1692
1693 pub(crate) fn spawn_child_bound_with_uid_in_environment<A: Actor + Binds<A>>(
1696 &self,
1697 parent: InstanceCell,
1698 uid: Uid,
1699 actor: A,
1700 environment: ActorEnvironment,
1701 ) -> Result<ActorHandle<A>, anyhow::Error> {
1702 let actor_id = self.ensure_child_uid(parent.actor_addr(), uid)?;
1703 Ok(self.spawn_bound_inner(actor_id, actor, Some(parent), environment))
1704 }
1705
1706 pub(crate) fn spawn_named_child<A: Actor>(
1710 &self,
1711 parent: InstanceCell,
1712 name: &str,
1713 actor: A,
1714 ) -> ActorHandle<A> {
1715 let actor_id = self.allocate_named_child_id(parent.actor_addr(), name);
1716 let environment = parent.actor_environment().clone();
1717 self.spawn_inner(actor_id, actor, Some(parent), environment)
1718 }
1719
1720 pub fn abort_root_actor(&self, root: &ActorId) -> Option<impl Future<Output = ActorAddr>> {
1724 self.get_instance_by_id(root).map(|cell| {
1725 let actor_addr = cell.actor_addr().clone();
1726 let r1 = actor_addr.clone();
1727 let r2 = actor_addr;
1728 async move {
1732 tokio::task::spawn_blocking(move || {
1733 let h = cell.inner.actor_task_handle.wait();
1734 tracing::debug!("{}: aborting {:?}", r1, h);
1735 h.abort();
1736 })
1737 .await
1738 .unwrap();
1739 r2
1740 }
1741 })
1742 }
1743
1744 pub fn stop_actor(
1747 &self,
1748 actor_id: &ActorId,
1749 reason: String,
1750 ) -> Option<watch::Receiver<ActorStatus>> {
1751 let cell = match self
1756 .state()
1757 .actors
1758 .get(actor_id)
1759 .and_then(|entry| entry.as_running().map(|(cell, _)| cell.upgrade()))
1760 {
1761 Some(cell) => cell,
1762 None => {
1763 tracing::error!(subject = %self.proc_addr().subject(), "no actor {} found", actor_id);
1764 return None;
1765 }
1766 }; match cell {
1768 None => None, Some(cell) => {
1770 tracing::info!("sending stop signal to {}", cell.actor_addr());
1771 if let Err(err) = cell.signal(Signal::DrainAndStop(reason)) {
1772 tracing::error!(
1773 "failed to send stop signal to uid {}: {:?}",
1774 cell.uid(),
1775 err
1776 );
1777 None
1778 } else {
1779 Some(cell.status().clone())
1780 }
1781 }
1782 }
1783 }
1784
1785 #[hyperactor::instrument(fields(subject = self.proc_addr().subject().to_string()))]
1789 pub async fn destroy_and_wait(
1790 &mut self,
1791 timeout: Duration,
1792 reason: &str,
1793 ) -> Result<(Vec<ActorAddr>, Vec<ActorAddr>), anyhow::Error> {
1794 tracing::debug!("proc stopping");
1795
1796 let coordinator_id = self.supervision_coordinator_actor_addr().cloned();
1797
1798 let mut statuses = HashMap::new();
1801 for actor_id in self
1802 .root_instances()
1803 .into_iter()
1804 .filter(|cell| !matches!(*cell.status().borrow(), ActorStatus::Client))
1805 .map(|cell| cell.actor_addr().clone())
1806 .collect::<Vec<_>>()
1807 {
1808 if coordinator_id.as_ref() == Some(&actor_id) {
1809 continue;
1810 }
1811 if let Some(status) = self.stop_actor(actor_id.id(), reason.to_string()) {
1812 statuses.insert(actor_id, status);
1813 }
1814 }
1815 tracing::debug!("non-coordinator actors stopped");
1816
1817 let waits: Vec<_> = statuses
1818 .iter_mut()
1819 .map(|(actor_id, root)| {
1820 let actor_id = actor_id.clone();
1821 async move {
1822 tokio::time::timeout(
1823 timeout,
1824 root.wait_for(|state: &ActorStatus| state.is_terminal()),
1825 )
1826 .await
1827 .ok()
1828 .map(|_| actor_id)
1829 }
1830 })
1831 .collect();
1832
1833 let results = futures::future::join_all(waits).await;
1834 let mut stopped_actors: Vec<_> = results
1835 .iter()
1836 .filter_map(|actor_id| actor_id.as_ref())
1837 .cloned()
1838 .collect();
1839 let aborted_actors: Vec<_> = statuses
1840 .iter()
1841 .filter(|(actor_id, _)| !stopped_actors.contains(actor_id))
1842 .map(|(actor_id, _)| {
1843 let f = self.abort_root_actor(actor_id.id());
1844 async move {
1845 let _ = if let Some(f) = f { Some(f.await) } else { None };
1846 actor_id.clone()
1851 }
1852 })
1853 .collect();
1854 let mut aborted_actors = futures::future::join_all(aborted_actors).await;
1855
1856 if let Some(ref coord_id) = coordinator_id
1862 && let Some(mut status) = self.stop_actor(coord_id.id(), reason.to_string())
1863 {
1864 let stopped =
1865 tokio::time::timeout(timeout, status.wait_for(|s: &ActorStatus| s.is_terminal()))
1866 .await
1867 .is_ok();
1868 if stopped {
1869 stopped_actors.push(coord_id.clone());
1870 } else {
1871 if let Some(f) = self.abort_root_actor(coord_id.id()) {
1872 f.await;
1873 }
1874 aborted_actors.push(coord_id.clone());
1875 }
1876 }
1877
1878 let flush_timeout = hyperactor_config::global::get(crate::config::FORWARDER_FLUSH_TIMEOUT);
1885 let gateway = self.gateway();
1886 match tokio::time::timeout(flush_timeout, gateway.flush()).await {
1887 Ok(Err(err)) => {
1888 tracing::warn!("gateway flush failed during proc exit: {:?}", err);
1889 }
1890 Err(_elapsed) => {
1891 tracing::warn!("gateway flush timed out during proc exit");
1892 }
1893 Ok(Ok(())) => {}
1894 }
1895
1896 tracing::info!(
1897 "destroy_and_wait: {} actors stopped, {} actors aborted",
1898 stopped_actors.len(),
1899 aborted_actors.len()
1900 );
1901 Ok((stopped_actors, aborted_actors))
1902 }
1903
1904 pub fn resolve_actor_ref<R: Actor + Referable>(
1923 &self,
1924 actor_ref: &ActorRef<R>,
1925 ) -> Option<ActorHandle<R>> {
1926 let cell = self.get_instance_by_id(actor_ref.actor_addr().id())?;
1927 if cell.status().borrow().is_terminal() {
1931 return None;
1932 }
1933 cell.downcast_handle()
1934 }
1935
1936 fn allocate_root_id(&self, name: &str) -> Result<ActorAddr, anyhow::Error> {
1940 self.reserve_root(Uid::singleton(Label::strip(name)))
1941 }
1942
1943 fn allocate_root_type<A: Actor>(&self) -> ActorAddr {
1945 self.root_addr(Uid::instance(default_actor_label::<A>()))
1946 }
1947
1948 fn allocate_root_label(&self, label: &str) -> ActorAddr {
1950 self.root_addr(Uid::instance(Label::strip(label)))
1951 }
1952
1953 fn allocate_root_uid(&self, uid: Uid) -> Result<ActorAddr, anyhow::Error> {
1955 self.reserve_root(uid)
1956 }
1957
1958 fn allocate_client_id(&self, label: &str) -> ActorAddr {
1959 let actor_id = if label.is_empty() {
1960 ActorId::anonymous(self.proc_id().clone())
1961 } else {
1962 ActorId::instance(Label::strip(label), self.proc_id().clone())
1963 };
1964 ActorAddr::new(actor_id, self.default_location())
1965 }
1966
1967 fn reserve_root(&self, uid: Uid) -> Result<ActorAddr, anyhow::Error> {
1968 let actor_id = ActorId::new(uid.clone(), self.proc_id().clone(), None);
1969 if !self.state().reserved_roots.insert(uid) {
1970 anyhow::bail!("an actor with id '{}' has already been spawned", actor_id)
1971 }
1972 Ok(ActorAddr::new(actor_id, self.default_location()))
1973 }
1974
1975 fn root_addr(&self, uid: Uid) -> ActorAddr {
1976 ActorAddr::new(
1977 ActorId::new(uid, self.proc_id().clone(), None),
1978 self.default_location(),
1979 )
1980 }
1981
1982 pub(crate) fn allocate_anonymous_child_id(&self, parent_id: &ActorAddr) -> ActorAddr {
1984 assert_eq!(parent_id.proc_id(), self.proc_id());
1985 ActorAddr::new(
1986 ActorId::anonymous(self.proc_id().clone()),
1987 self.default_location(),
1988 )
1989 }
1990
1991 pub(crate) fn allocate_child_id<A: Actor>(&self, parent_id: &ActorAddr) -> ActorAddr {
1993 assert_eq!(parent_id.proc_id(), self.proc_id());
1994 let actor_id = ActorId::instance(default_actor_label::<A>(), self.proc_id().clone());
1995 ActorAddr::new(actor_id, self.default_location())
1996 }
1997
1998 fn ensure_child_uid(
2000 &self,
2001 parent_id: &ActorAddr,
2002 uid: Uid,
2003 ) -> Result<ActorAddr, anyhow::Error> {
2004 assert_eq!(parent_id.proc_id(), self.proc_id());
2005 let actor_id = ActorId::new(uid.clone(), self.proc_id().clone(), None);
2006 let actor_addr = ActorAddr::new(actor_id, self.default_location());
2007 if !self.state().reserved_child_uids.insert(uid) {
2008 anyhow::bail!("an actor with id {} has already been spawned", actor_addr);
2009 }
2010 Ok(actor_addr)
2011 }
2012
2013 pub(crate) fn allocate_named_child_id(&self, parent_id: &ActorAddr, name: &str) -> ActorAddr {
2015 assert_eq!(parent_id.proc_id(), self.proc_id());
2016 let proc_id = self.proc_id().clone();
2017 let actor_id = crate::id::ActorId::instance(crate::id::Label::strip(name), proc_id);
2018 ActorAddr::new(actor_id, self.default_location())
2019 }
2020
2021 pub fn downgrade(&self) -> WeakProc {
2023 WeakProc::new(self)
2024 }
2025
2026 pub async fn flush(&self) -> Result<(), anyhow::Error> {
2029 self.gateway().flush().await
2030 }
2031
2032 pub async fn join_mailbox_server(&self) {
2041 let handle = self.inner.mailbox_server_handle.lock().unwrap().take();
2042 if let Some(handle) = handle {
2043 handle.stop("proc shutting down");
2044 let _ = handle.await;
2045 }
2046 }
2047
2048 pub(crate) fn is_local_delivery_target(&self, dest_proc: &ProcAddr) -> bool {
2049 self.is_local_delivery_target_at(dest_proc, &[self.default_location()])
2050 }
2051
2052 pub(crate) fn is_local_delivery_target_at(
2053 &self,
2054 dest_proc: &ProcAddr,
2055 local_locations: &[Location],
2056 ) -> bool {
2057 dest_proc.id() == self.proc_id()
2058 && (!requires_location_for_local_delivery_identity(dest_proc.id())
2059 || local_locations
2060 .iter()
2061 .any(|location| dest_proc.location() == location))
2062 }
2063}
2064
2065fn requires_location_for_local_delivery_identity(proc_id: &ProcId) -> bool {
2066 is_legacy_pseudo_singleton_proc_id(proc_id)
2071}
2072
2073fn default_actor_label<A>() -> Label {
2074 let type_name = std::any::type_name::<A>();
2075 let type_name = type_name
2076 .split_once('<')
2077 .map_or(type_name, |(base, _)| base);
2078 Label::strip(type_name.rsplit("::").next().unwrap_or(type_name))
2079}
2080
2081fn global_proc_label() -> Label {
2082 let hostname = hostname::get().expect("hostname should be available");
2083 global_proc_label_from(&hostname.to_string_lossy(), std::process::id())
2084}
2085
2086fn global_proc_label_from(hostname: &str, pid: u32) -> Label {
2087 let short_hostname = hostname
2088 .split_once('.')
2089 .map_or(hostname, |(short, _)| short);
2090 Label::strip(&format!("{}-{}", short_hostname, pid))
2091}
2092
2093fn assert_not_legacy_pseudo_singleton_proc_id(proc_id: &ProcId) {
2094 if is_legacy_pseudo_singleton_proc_id(proc_id) {
2095 panic!(
2096 "legacy pseudo-singleton proc id '{}' must be constructed with a dedicated Proc constructor",
2097 proc_id
2098 );
2099 }
2100}
2101
2102fn is_legacy_pseudo_singleton_proc_id(proc_id: &ProcId) -> bool {
2103 matches!(
2104 proc_id.uid(),
2105 Uid::Singleton(label) if is_legacy_pseudo_singleton_label(label)
2106 )
2107}
2108
2109fn is_legacy_pseudo_singleton_label(label: &Label) -> bool {
2110 label.as_str() == LEGACY_SERVICE_PROC_NAME
2111}
2112
2113#[async_trait]
2114impl MailboxSender for Proc {
2115 fn post_unchecked(
2116 &self,
2117 envelope: MessageEnvelope,
2118 return_handle: PortHandle<Undeliverable<MessageEnvelope>>,
2119 ) {
2120 let dest_proc = envelope.dest().actor_addr().proc_addr();
2121 if self.is_local_delivery_target(&dest_proc) {
2122 self.state().proc_muxer.post(envelope, return_handle);
2123 return;
2124 }
2125 self.state().gateway.post(envelope, return_handle);
2131 }
2132
2133 async fn flush(&self) -> Result<(), anyhow::Error> {
2134 self.gateway().flush().await
2135 }
2136}
2137
2138#[derive(Clone, Debug)]
2140pub struct WeakProc(Weak<ProcState>);
2141
2142impl WeakProc {
2143 fn new(proc: &Proc) -> Self {
2144 Self(Arc::downgrade(&proc.inner))
2145 }
2146
2147 pub fn upgrade(&self) -> Option<Proc> {
2149 self.0.upgrade().map(|inner| Proc { inner })
2150 }
2151
2152 pub(crate) fn ptr_eq(&self, other: &WeakProc) -> bool {
2154 self.0.ptr_eq(&other.0)
2155 }
2156}
2157
2158#[async_trait]
2159impl MailboxSender for WeakProc {
2160 fn post_unchecked(
2161 &self,
2162 envelope: MessageEnvelope,
2163 return_handle: PortHandle<Undeliverable<MessageEnvelope>>,
2164 ) {
2165 match self.upgrade() {
2166 Some(proc) => proc.post(envelope, return_handle),
2167 None => {
2168 let target = envelope.dest().clone();
2169 let failure =
2170 DeliveryFailure::new(UndeliverableReason::Transport(TransportFailure::new(
2171 target,
2172 TransportFailureReason::LinkUnavailable("proc is gone".to_string()),
2173 )));
2174 envelope.undeliverable(failure, return_handle)
2175 }
2176 }
2177 }
2178
2179 async fn flush(&self) -> Result<(), anyhow::Error> {
2180 match self.upgrade() {
2181 Some(proc) => proc.flush().await,
2182 None => Ok(()),
2183 }
2184 }
2185}
2186
2187pub struct WorkCell<A: Actor + Send>(
2190 Box<
2191 dyn for<'a> FnOnce(
2192 &'a mut A,
2193 &'a Instance<A>,
2194 )
2195 -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + 'a + Send>>
2196 + Send
2197 + Sync,
2198 >,
2199);
2200
2201impl<A: Actor + Send> WorkCell<A> {
2202 fn new(
2204 f: impl for<'a> FnOnce(
2205 &'a mut A,
2206 &'a Instance<A>,
2207 )
2208 -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + 'a + Send>>
2209 + Send
2210 + Sync
2211 + 'static,
2212 ) -> Self {
2213 Self(Box::new(f))
2214 }
2215
2216 pub fn handle<'a>(
2218 self,
2219 actor: &'a mut A,
2220 instance: &'a Instance<A>,
2221 ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + 'a>> {
2222 (self.0)(actor, instance)
2223 }
2224}
2225
2226pub struct Context<'a, A: Actor> {
2228 instance: &'a Instance<A>,
2229 headers: Flattrs,
2230}
2231
2232impl<'a, A: Actor> Context<'a, A> {
2233 pub fn new(instance: &'a Instance<A>, headers: Flattrs) -> Self {
2235 Self { instance, headers }
2236 }
2237
2238 pub fn headers(&self) -> &Flattrs {
2240 &self.headers
2241 }
2242}
2243
2244impl<A: Actor> Deref for Context<'_, A> {
2245 type Target = Instance<A>;
2246
2247 fn deref(&self) -> &Self::Target {
2248 self.instance
2249 }
2250}
2251
2252pub struct Instance<A: Actor> {
2256 inner: Arc<InstanceState<A>>,
2257}
2258
2259impl<A: Actor> fmt::Debug for Instance<A> {
2260 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2261 f.debug_struct("Instance").field("inner", &"..").finish()
2262 }
2263}
2264
2265struct InstanceState<A: Actor> {
2266 proc: Proc,
2268
2269 cell: InstanceCell,
2271
2272 mailbox: Mailbox,
2274
2275 ports: Arc<HandlerPorts<A>>,
2276
2277 delayed_posts: DelayedPosts<A>,
2279
2280 introspect_shutdown_tx: Mutex<Option<oneshot::Sender<ActorStatus>>>,
2282
2283 introspect_task_handle: Mutex<Option<JoinHandle<()>>>,
2285
2286 detached_introspect_shutdown_tx: Mutex<Option<oneshot::Sender<ActorStatus>>>,
2290
2291 id: Uuid,
2293
2294 sequencer: Sequencer,
2296
2297 instance_locals: ActorLocalStorage,
2299}
2300
2301struct ActorStopped {
2302 reason: String,
2303 stop_mode: StopMode,
2304}
2305
2306#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2307enum ChildTeardown {
2308 Cooperative(StopMode),
2309 Kill,
2310}
2311
2312impl ChildTeardown {
2313 fn from_run_result(result: &Result<ActorStopped, ActorError>) -> Self {
2314 match result {
2315 Ok(stopped) => Self::Cooperative(stopped.stop_mode),
2316 Err(err) if matches!(err.kind.as_ref(), ActorErrorKind::Aborted(_)) => Self::Kill,
2317 Err(_) => Self::Cooperative(StopMode::Stop),
2318 }
2319 }
2320
2321 fn cooperative_signal(mode: StopMode) -> Signal {
2322 match mode {
2323 StopMode::DrainAndStop => Signal::DrainAndStop("parent draining".to_string()),
2324 StopMode::Stop => Signal::Stop("parent stopping".to_string()),
2325 }
2326 }
2327}
2328
2329type DelayedPost<A> = Box<dyn FnOnce(&Instance<A>) + Send>;
2330
2331trait PostAfterEndpoint<A: Actor, M: Message>: Send {
2332 fn endpoint_location(&self) -> crate::EndpointLocation;
2333
2334 fn into_delayed_post(self, message: M) -> DelayedPost<A>;
2335}
2336
2337impl<A, M> PostAfterEndpoint<A, M> for &Instance<A>
2338where
2339 A: Actor + Handler<M>,
2340 M: Message,
2341{
2342 fn endpoint_location(&self) -> crate::EndpointLocation {
2343 crate::EndpointLocation::Actor(self.self_addr().clone())
2344 }
2345
2346 fn into_delayed_post(self, message: M) -> DelayedPost<A> {
2347 let dest = self.clone_for_py();
2348 Box::new(move |this| crate::Endpoint::post(&dest, this, message))
2349 }
2350}
2351
2352impl<A, M> PostAfterEndpoint<A, M> for &Context<'_, A>
2353where
2354 A: Actor + Handler<M>,
2355 M: Message,
2356{
2357 fn endpoint_location(&self) -> crate::EndpointLocation {
2358 crate::EndpointLocation::Actor(self.self_addr().clone())
2359 }
2360
2361 fn into_delayed_post(self, message: M) -> DelayedPost<A> {
2362 let dest = self.clone_for_py();
2363 Box::new(move |this| crate::Endpoint::post(&dest, this, message))
2364 }
2365}
2366
2367impl<A, M> PostAfterEndpoint<A, M> for Instance<A>
2368where
2369 A: Actor + Handler<M>,
2370 M: Message,
2371{
2372 fn endpoint_location(&self) -> crate::EndpointLocation {
2373 crate::EndpointLocation::Actor(self.self_addr().clone())
2374 }
2375
2376 fn into_delayed_post(self, message: M) -> DelayedPost<A> {
2377 Box::new(move |this| crate::Endpoint::post(&self, this, message))
2378 }
2379}
2380
2381impl<A, B, M> PostAfterEndpoint<A, M> for ActorHandle<B>
2382where
2383 A: Actor,
2384 B: Actor + Handler<M>,
2385 M: Message,
2386{
2387 fn endpoint_location(&self) -> crate::EndpointLocation {
2388 crate::Endpoint::endpoint_location(&self)
2389 }
2390
2391 fn into_delayed_post(self, message: M) -> DelayedPost<A> {
2392 Box::new(move |this| crate::Endpoint::post(&self, this, message))
2393 }
2394}
2395
2396impl<A, M> PostAfterEndpoint<A, M> for PortHandle<M>
2397where
2398 A: Actor,
2399 M: Message,
2400{
2401 fn endpoint_location(&self) -> crate::EndpointLocation {
2402 crate::Endpoint::endpoint_location(&self)
2403 }
2404
2405 fn into_delayed_post(self, message: M) -> DelayedPost<A> {
2406 Box::new(move |this| crate::Endpoint::post(&self, this, message))
2407 }
2408}
2409
2410impl<A, M> PostAfterEndpoint<A, M> for OncePortHandle<M>
2411where
2412 A: Actor,
2413 M: Message,
2414{
2415 fn endpoint_location(&self) -> crate::EndpointLocation {
2416 crate::Endpoint::endpoint_location(self)
2417 }
2418
2419 fn into_delayed_post(self, message: M) -> DelayedPost<A> {
2420 Box::new(move |this| crate::Endpoint::post(self, this, message))
2421 }
2422}
2423
2424impl<A, B, M> PostAfterEndpoint<A, M> for ActorRef<B>
2425where
2426 A: Actor,
2427 B: Referable + RemoteHandles<M>,
2428 M: RemoteMessage,
2429{
2430 fn endpoint_location(&self) -> crate::EndpointLocation {
2431 crate::Endpoint::endpoint_location(&self)
2432 }
2433
2434 fn into_delayed_post(self, message: M) -> DelayedPost<A> {
2435 Box::new(move |this| crate::Endpoint::post(&self, this, message))
2436 }
2437}
2438
2439impl<A, M> PostAfterEndpoint<A, M> for crate::PortRef<M>
2440where
2441 A: Actor,
2442 M: RemoteMessage,
2443{
2444 fn endpoint_location(&self) -> crate::EndpointLocation {
2445 crate::Endpoint::endpoint_location(&self)
2446 }
2447
2448 fn into_delayed_post(self, message: M) -> DelayedPost<A> {
2449 Box::new(move |this| crate::Endpoint::post(&self, this, message))
2450 }
2451}
2452
2453impl<A, M> PostAfterEndpoint<A, M> for crate::OncePortRef<M>
2454where
2455 A: Actor,
2456 M: RemoteMessage,
2457{
2458 fn endpoint_location(&self) -> crate::EndpointLocation {
2459 crate::Endpoint::endpoint_location(self)
2460 }
2461
2462 fn into_delayed_post(self, message: M) -> DelayedPost<A> {
2463 Box::new(move |this| crate::Endpoint::post(self, this, message))
2464 }
2465}
2466
2467struct DelayedPosts<A: Actor> {
2468 ingress: Arc<DelayedPostIngressGate>,
2469 state: Mutex<DelayedPostState<A>>,
2470 notify: Notify,
2471}
2472
2473struct DelayedPostState<A: Actor> {
2474 queue: BTreeMap<(tokio::time::Instant, u64), DelayedPost<A>>,
2475 next_order: u64,
2476}
2477
2478impl<A: Actor> DelayedPosts<A> {
2479 fn new() -> Self {
2480 Self {
2481 ingress: Arc::new(DelayedPostIngressGate::new()),
2482 state: Mutex::new(DelayedPostState {
2483 queue: BTreeMap::new(),
2484 next_order: 0,
2485 }),
2486 notify: Notify::new(),
2487 }
2488 }
2489
2490 fn push(&self, deadline: tokio::time::Instant, post: DelayedPost<A>) {
2491 let mut state = self.state.lock().unwrap();
2492 let order = state.next_order;
2493 state.next_order = state.next_order.wrapping_add(1);
2494 state.queue.insert((deadline, order), post);
2495 drop(state);
2496 self.notify.notify_one();
2497 }
2498
2499 fn next_deadline(&self) -> Option<tokio::time::Instant> {
2500 self.state
2501 .lock()
2502 .unwrap()
2503 .queue
2504 .keys()
2505 .next()
2506 .map(|(deadline, _)| *deadline)
2507 }
2508
2509 fn pop_due(&self, now: tokio::time::Instant) -> Vec<DelayedPost<A>> {
2510 let mut posts = Vec::new();
2511 let mut state = self.state.lock().unwrap();
2512 while let Some((&(deadline, _), _)) = state.queue.first_key_value() {
2513 if deadline > now {
2514 break;
2515 }
2516 let (_, post) = state.queue.pop_first().expect("delayed post should exist");
2517 posts.push(post);
2518 }
2519 posts
2520 }
2521
2522 fn drain(&self) {
2523 self.ingress.drain();
2524 }
2525
2526 fn is_draining(&self) -> bool {
2527 self.ingress.is_draining()
2528 }
2529}
2530
2531const DELAYED_POST_INGRESS_DRAINING: usize = 1usize << (usize::BITS as usize - 1);
2532const DELAYED_POST_INGRESS_ACTIVE_MASK: usize = !DELAYED_POST_INGRESS_DRAINING;
2533
2534struct DelayedPostIngressGate {
2535 state: AtomicUsize,
2536 wait_lock: Mutex<()>,
2537 drained: Condvar,
2538}
2539
2540struct DelayedPostIngressGuard {
2541 gate: Arc<DelayedPostIngressGate>,
2542}
2543
2544impl DelayedPostIngressGate {
2545 fn new() -> Self {
2546 Self {
2547 state: AtomicUsize::new(0),
2548 wait_lock: Mutex::new(()),
2549 drained: Condvar::new(),
2550 }
2551 }
2552
2553 fn try_enter(self: &Arc<Self>) -> Result<DelayedPostIngressGuard, ()> {
2554 let mut state = self.state.load(Ordering::Acquire);
2555 loop {
2556 if state & DELAYED_POST_INGRESS_DRAINING != 0 {
2557 return Err(());
2558 }
2559
2560 let active = state & DELAYED_POST_INGRESS_ACTIVE_MASK;
2561 assert!(
2562 active < DELAYED_POST_INGRESS_ACTIVE_MASK,
2563 "too many active delayed post sends"
2564 );
2565
2566 match self.state.compare_exchange_weak(
2567 state,
2568 state + 1,
2569 Ordering::AcqRel,
2570 Ordering::Acquire,
2571 ) {
2572 Ok(_) => {
2573 return Ok(DelayedPostIngressGuard {
2574 gate: Arc::clone(self),
2575 });
2576 }
2577 Err(next_state) => state = next_state,
2578 }
2579 }
2580 }
2581
2582 fn drain(&self) {
2583 let mut state = self.state.load(Ordering::Acquire);
2584 loop {
2585 if state & DELAYED_POST_INGRESS_DRAINING != 0 {
2586 break;
2587 }
2588 match self.state.compare_exchange_weak(
2589 state,
2590 state | DELAYED_POST_INGRESS_DRAINING,
2591 Ordering::AcqRel,
2592 Ordering::Acquire,
2593 ) {
2594 Ok(_) => break,
2595 Err(next_state) => state = next_state,
2596 }
2597 }
2598
2599 let mut wait_guard = self.wait_lock.lock().unwrap();
2600 while self.state.load(Ordering::Acquire) & DELAYED_POST_INGRESS_ACTIVE_MASK != 0 {
2601 wait_guard = self.drained.wait(wait_guard).unwrap();
2602 }
2603 }
2604
2605 fn is_draining(&self) -> bool {
2606 self.state.load(Ordering::Acquire) & DELAYED_POST_INGRESS_DRAINING != 0
2607 }
2608}
2609
2610impl Drop for DelayedPostIngressGuard {
2611 fn drop(&mut self) {
2612 let previous = self.gate.state.fetch_sub(1, Ordering::AcqRel);
2613 assert!(
2614 previous & DELAYED_POST_INGRESS_ACTIVE_MASK != 0,
2615 "delayed post ingress active count underflow"
2616 );
2617 if previous & DELAYED_POST_INGRESS_DRAINING != 0
2618 && previous & DELAYED_POST_INGRESS_ACTIVE_MASK == 1
2619 {
2620 let _wait_guard = self.gate.wait_lock.lock().unwrap();
2621 self.gate.drained.notify_all();
2622 }
2623 }
2624}
2625
2626impl<A: Actor> InstanceState<A> {
2627 fn self_addr(&self) -> &ActorAddr {
2628 self.mailbox.actor_addr()
2629 }
2630}
2631
2632impl<A: Actor> Drop for InstanceState<A> {
2633 fn drop(&mut self) {
2634 let terminal_status = ActorStatus::Stopped("instance is dropped".into());
2635 if let Some(shutdown_tx) = self.detached_introspect_shutdown_tx.lock().unwrap().take() {
2636 let _ = shutdown_tx.send(terminal_status.clone());
2637 return;
2638 }
2639
2640 if let Some(shutdown_tx) = self.introspect_shutdown_tx.lock().unwrap().take() {
2641 let _ = shutdown_tx.send(terminal_status.clone());
2642 }
2643 let _ = self.introspect_task_handle.lock().unwrap().take();
2644
2645 self.cell.publish_dropped_status(terminal_status);
2646 }
2647}
2648
2649pub struct InstanceReceivers<A: Actor> {
2656 actor_loop: Option<(
2659 mpsc::UnboundedReceiver<Signal>,
2660 mpsc::UnboundedReceiver<ActorSupervisionEvent>,
2661 )>,
2662 work: ActorWorkReceiver<A>,
2664 introspect: PortReceiver<IntrospectMessage>,
2666}
2667
2668impl<A: Actor> Instance<A> {
2669 fn new(
2675 proc: Proc,
2676 actor_id: ActorAddr,
2677 detached: bool,
2678 parent: Option<InstanceCell>,
2679 actor_environment: ActorEnvironment,
2680 ) -> (Self, InstanceReceivers<A>) {
2681 let (instance, receivers) =
2682 Self::new_unpublished(proc, actor_id, detached, parent, actor_environment);
2683 instance.publish();
2684 (instance, receivers)
2685 }
2686
2687 fn new_unpublished(
2690 proc: Proc,
2691 actor_id: ActorAddr,
2692 detached: bool,
2693 parent: Option<InstanceCell>,
2694 actor_environment: ActorEnvironment,
2695 ) -> (Self, InstanceReceivers<A>) {
2696 let mailbox = Mailbox::new(actor_id.clone());
2698 let enable_buffering =
2699 hyperactor_config::global::get(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER);
2700 let (work_tx, work_rx) = sequenced_unbounded_with_buffering(enable_buffering);
2701 let inbound_ordering_snapshot_handle = work_rx.snapshot_handle();
2702 let queue_depth = Arc::new(AtomicU64::new(0));
2703 let proc_stats = Arc::clone(&proc.state().queue_stats);
2704 let ports: Arc<HandlerPorts<A>> = Arc::new(HandlerPorts::new(
2705 mailbox.clone(),
2706 work_tx,
2707 enable_buffering,
2708 Arc::clone(&queue_depth),
2709 proc_stats,
2710 ));
2711 let (status_tx, status_rx) = watch::channel(ActorStatus::Created);
2712
2713 let actor_type = match TypeInfo::of::<A>() {
2714 Some(info) => ActorType::Named(info),
2715 None => ActorType::Anonymous(std::any::type_name::<A>()),
2716 };
2717 let actor_loop_ports = if detached {
2718 None
2719 } else {
2720 let (signal_tx, signal_receiver) = mpsc::unbounded_channel::<Signal>();
2721 let (supervision_tx, supervision_receiver) =
2722 mpsc::unbounded_channel::<ActorSupervisionEvent>();
2723 Some((
2724 (signal_tx, supervision_tx),
2725 (signal_receiver, supervision_receiver),
2726 ))
2727 };
2728
2729 let (actor_loop, actor_loop_receivers) = actor_loop_ports.unzip();
2730
2731 let (introspect_port, introspect_receiver) = mailbox.open_port::<IntrospectMessage>();
2737 introspect_port.bind_control_port(crate::port::ControlPort::Introspect);
2738
2739 let instance_id = Uuid::now_v7();
2740
2741 let inbound_ordering_snapshot: Option<
2744 Box<dyn Fn() -> crate::ordering::OrderingSnapshot + Send + Sync>,
2745 > = Some(Box::new(move || {
2746 inbound_ordering_snapshot_handle.snapshot()
2747 }));
2748
2749 let cell = InstanceCell::new(
2750 actor_id,
2751 instance_id,
2752 actor_type,
2753 actor_environment,
2754 proc.clone(),
2755 mailbox.clone(),
2756 actor_loop,
2757 status_tx,
2758 status_rx,
2759 parent,
2760 ports.clone(),
2761 queue_depth,
2762 inbound_ordering_snapshot,
2763 );
2764 let actor_id = hash_to_u64(cell.actor_addr().id());
2765 notify_actor_status_changed(ActorStatusEvent {
2766 id: generate_actor_status_event_id(actor_id),
2767 timestamp: cell.created_at(),
2768 actor_id,
2769 new_status: ActorStatus::Created.arm().unwrap_or("unknown").to_string(),
2770 reason: None,
2771 });
2772 let inner = Arc::new(InstanceState {
2773 proc,
2774 cell,
2775 mailbox,
2776 ports,
2777 delayed_posts: DelayedPosts::new(),
2778 introspect_shutdown_tx: Mutex::new(None),
2779 introspect_task_handle: Mutex::new(None),
2780 detached_introspect_shutdown_tx: Mutex::new(None),
2781 sequencer: Sequencer::new(instance_id),
2782 id: instance_id,
2783 instance_locals: ActorLocalStorage::new(),
2784 });
2785 (
2786 Self { inner },
2787 InstanceReceivers {
2788 actor_loop: actor_loop_receivers,
2789 work: ActorWorkReceiver::new(work_rx),
2790 introspect: introspect_receiver,
2791 },
2792 )
2793 }
2794
2795 fn publish(&self) {
2798 self.inner
2799 .proc
2800 .state()
2801 .proc_muxer
2802 .bind_mailbox(self.inner.mailbox.clone());
2803 }
2804
2805 fn spawn_introspect(&self, receiver: PortReceiver<IntrospectMessage>) {
2806 let (shutdown_tx, shutdown_rx) = oneshot::channel();
2807 let handle = tokio::spawn(crate::introspect::serve_introspect(
2808 self.inner.cell.clone(),
2809 receiver,
2810 shutdown_rx,
2811 ));
2812
2813 let mut shutdown = self.inner.introspect_shutdown_tx.lock().unwrap();
2814 assert!(
2815 shutdown.is_none(),
2816 "introspection shutdown handle already set"
2817 );
2818 let mut task = self.inner.introspect_task_handle.lock().unwrap();
2819 assert!(task.is_none(), "introspection task already set");
2820 *shutdown = Some(shutdown_tx);
2821 *task = Some(handle);
2822 }
2823
2824 fn spawn_detached_introspect(&self, receiver: PortReceiver<IntrospectMessage>) {
2825 let (detached_shutdown_tx, detached_shutdown_rx) = oneshot::channel::<ActorStatus>();
2826 let (introspect_shutdown_tx, introspect_shutdown_rx) = oneshot::channel();
2827 let introspect_handle = tokio::spawn(crate::introspect::serve_introspect(
2828 self.inner.cell.clone(),
2829 receiver,
2830 introspect_shutdown_rx,
2831 ));
2832
2833 let cell = self.inner.cell.clone();
2834 tokio::spawn(async move {
2835 let Ok(terminal_status) = detached_shutdown_rx.await else {
2836 return;
2837 };
2838 let _ = introspect_shutdown_tx.send(terminal_status.clone());
2839 if let Err(err) = introspect_handle.await {
2840 tracing::debug!("introspect task join failed: {:?}", err);
2841 }
2842 cell.publish_dropped_status(terminal_status);
2843 });
2844
2845 let mut shutdown = self.inner.detached_introspect_shutdown_tx.lock().unwrap();
2846 assert!(
2847 shutdown.is_none(),
2848 "detached introspection shutdown handle already set"
2849 );
2850 *shutdown = Some(detached_shutdown_tx);
2851 }
2852
2853 fn signal_introspect_stop(&self, terminal_status: ActorStatus) -> Option<JoinHandle<()>> {
2854 if let Some(shutdown_tx) = self.inner.introspect_shutdown_tx.lock().unwrap().take() {
2855 let _ = shutdown_tx.send(terminal_status);
2856 }
2857 self.inner.introspect_task_handle.lock().unwrap().take()
2858 }
2859
2860 async fn stop_introspect(&self, terminal_status: ActorStatus) {
2861 if let Some(handle) = self.signal_introspect_stop(terminal_status)
2862 && let Err(err) = handle.await
2863 {
2864 tracing::debug!("introspect task join failed: {:?}", err);
2865 }
2866 }
2867
2868 #[track_caller]
2871 pub fn change_status(&self, new: ActorStatus) {
2872 self.inner.cell.change_status(new);
2873 }
2874
2875 fn is_terminal(&self) -> bool {
2876 self.inner.cell.status().borrow().is_terminal()
2877 }
2878
2879 fn is_stopping(&self) -> bool {
2880 self.inner.cell.status().borrow().is_stopping()
2881 }
2882
2883 pub fn self_addr(&self) -> &ActorAddr {
2885 self.inner.self_addr()
2886 }
2887
2888 pub(crate) fn report_delivery_failure(&self, report: crate::mailbox::DeliveryFailureReport) {
2890 static REPORT_WARNED_MAILBOXES: OnceLock<DashSet<ActorAddr>> = OnceLock::new();
2891
2892 let mailbox = &self.inner.mailbox;
2893 mailbox.ensure_return_handler();
2894 let return_handle = mailbox.bound_return_handle().unwrap_or_else(|| {
2895 let actor_id = mailbox.actor_addr();
2896 if REPORT_WARNED_MAILBOXES
2897 .get_or_init(DashSet::new)
2898 .insert(actor_id.clone())
2899 {
2900 let bt = std::backtrace::Backtrace::force_capture();
2901 tracing::warn!(
2902 actor_id = ?actor_id,
2903 backtrace = ?bt,
2904 "actor attempted to report delivery failure without binding Undeliverable<MessageEnvelope>"
2905 );
2906 }
2907 crate::mailbox::monitored_return_handle()
2908 });
2909
2910 if let Err(error) =
2911 return_handle.try_post(self, crate::mailbox::Undeliverable::report(report.clone()))
2912 {
2913 tracing::error!(
2914 sender = %report.sender,
2915 dest = %report.dest,
2916 message_type = report.message_type.as_deref().unwrap_or("unknown"),
2917 error = %report.error_msg().unwrap_or_default(),
2918 return_error = %error,
2919 "delivery failure report could not be returned"
2920 );
2921 }
2922 }
2923
2924 pub fn introspect_payload(&self) -> crate::introspect::IntrospectResult {
2940 crate::introspect::live_actor_payload(&self.inner.cell)
2941 }
2942
2943 pub fn recording_span(&self) -> tracing::Span {
2947 use crate::subject::AsSubject;
2948 self.inner
2949 .cell
2950 .recording()
2951 .span(&self.self_addr().subject().to_string())
2952 }
2953
2954 pub fn publish_attrs(&self, attrs: hyperactor_config::Attrs) {
2962 #[cfg(debug_assertions)]
2963 {
2964 use std::collections::HashSet;
2965 use std::sync::OnceLock;
2966
2967 use hyperactor_config::attrs::AttrKeyInfo;
2968
2969 static INTROSPECT_KEYS: OnceLock<HashSet<&'static str>> = OnceLock::new();
2970 let allowed = INTROSPECT_KEYS.get_or_init(|| {
2971 inventory::iter::<AttrKeyInfo>()
2972 .filter(|info| info.meta.get(hyperactor_config::INTROSPECT).is_some())
2973 .map(|info| info.name)
2974 .collect()
2975 });
2976 for (name, _) in attrs.iter() {
2977 debug_assert!(
2978 allowed.contains(name),
2979 "publish_attrs: key {:?} is not tagged with INTROSPECT",
2980 name
2981 );
2982 }
2983 }
2984 self.inner.cell.set_published_attrs(attrs);
2985 }
2986
2987 pub fn publish_attr<T: hyperactor_config::AttrValue>(
2993 &self,
2994 key: hyperactor_config::Key<T>,
2995 value: T,
2996 ) {
2997 debug_assert!(
2998 key.attrs().get(hyperactor_config::INTROSPECT).is_some(),
2999 "publish_attr called with non-introspection key: {}",
3000 key.name()
3001 );
3002 self.inner.cell.merge_published_attr(key, value);
3003 }
3004
3005 pub fn set_attrs_snapshot(
3015 &self,
3016 callback: impl Fn() -> hyperactor_config::Attrs + Send + Sync + 'static,
3017 ) {
3018 self.inner.cell.set_attrs_snapshot(callback);
3019 }
3020
3021 pub fn set_system(&self) {
3024 self.inner
3025 .cell
3026 .inner
3027 .is_system
3028 .store(true, Ordering::Relaxed);
3029 }
3030
3031 pub fn set_query_child_handler(
3040 &self,
3041 handler: impl (Fn(&Addr) -> IntrospectResult) + Send + Sync + 'static,
3042 ) {
3043 self.inner.cell.set_query_child_handler(handler);
3044 }
3045
3046 pub fn stop(&self, reason: &str) -> Result<(), ActorError> {
3048 tracing::info!(
3049 actor_id = %self.inner.cell.actor_addr(),
3050 reason,
3051 "instance stop called",
3052 );
3053 self.inner.cell.signal(Signal::Stop(reason.to_string()))
3054 }
3055
3056 pub fn drain_and_stop(&self, reason: &str) -> Result<(), ActorError> {
3058 tracing::info!(
3059 actor_id = %self.inner.cell.actor_addr(),
3060 reason,
3061 "instance drain_and_stop called",
3062 );
3063 self.inner
3064 .cell
3065 .signal(Signal::DrainAndStop(reason.to_string()))
3066 }
3067
3068 pub fn kill(&self, reason: &str) -> Result<(), ActorError> {
3070 tracing::info!(
3071 actor_id = %self.inner.cell.actor_addr(),
3072 reason,
3073 "instance kill called",
3074 );
3075 self.inner.cell.signal(Signal::Kill(reason.to_string()))
3076 }
3077
3078 pub fn abort(&self, reason: &str) -> Result<(), ActorError> {
3080 tracing::info!(
3081 actor_id = %self.inner.cell.actor_addr(),
3082 reason,
3083 "instance abort called",
3084 );
3085 self.kill(reason)
3086 }
3087
3088 pub fn close(&self) {
3090 self.inner.delayed_posts.drain();
3091 self.inner.mailbox.drain();
3092 }
3093
3094 pub fn status(&self) -> watch::Receiver<ActorStatus> {
3096 self.inner.cell.status().clone()
3097 }
3098
3099 pub(crate) fn close_client(&self, reason: &str) {
3100 let status = ActorStatus::Stopped(reason.to_string());
3101 self.inner.mailbox.close(status.clone());
3102 self.change_status(status);
3103 }
3104
3105 pub fn exit(&self, reason: &str) -> Result<(), ActorError> {
3107 self.inner
3108 .cell
3109 .signal(Signal::ExitRequested(reason.to_string()))
3110 }
3111
3112 pub fn exit_after_drain(&self, reason: &str) -> Result<(), ActorError> {
3119 let this = self.clone_for_py();
3120 let reason = reason.to_string();
3121 let work = WorkCell::new(move |_actor: &mut A, _instance: &Instance<A>| {
3122 Box::pin(async move {
3123 this.exit(&reason).map_err(anyhow::Error::from)?;
3124 Ok(())
3125 })
3126 });
3127 self.enqueue_runtime_work(work)
3128 }
3129
3130 pub fn open_port<M: Message>(&self) -> (PortHandle<M>, PortReceiver<M>) {
3135 self.inner.mailbox.open_port()
3136 }
3137
3138 pub(crate) fn bind_handler_enqueue_port<M: RemoteMessage>(
3140 &self,
3141 enqueue: impl Fn(Flattrs, M) -> Result<(), anyhow::Error> + Send + Sync + 'static,
3142 ) where
3143 A: Handler<M>,
3144 {
3145 self.inner.ports.bind_enqueue(enqueue);
3146 }
3147
3148 pub fn open_once_port<M: Message>(&self) -> (OncePortHandle<M>, OncePortReceiver<M>) {
3152 self.inner.mailbox.open_once_port()
3153 }
3154
3155 #[doc(hidden)]
3157 pub fn signal_sender(&self) -> mpsc::UnboundedSender<Signal> {
3158 self.inner.cell.signal_sender()
3159 }
3160
3161 pub fn locals(&self) -> &ActorLocalStorage {
3163 &self.inner.instance_locals
3164 }
3165
3166 pub fn post(&self, port_id: impl Into<PortAddr>, headers: Flattrs, message: wirevalue::Any) {
3168 let port_id: PortAddr = port_id.into();
3169 <Self as context::MailboxExt>::post(
3170 self,
3171 port_id,
3172 headers,
3173 message,
3174 true,
3175 context::SeqInfoPolicy::AssignNew,
3176 )
3177 }
3178
3179 #[doc(hidden)]
3185 pub fn post_with_external_seq_info(
3186 &self,
3187 port_id: impl Into<PortAddr>,
3188 headers: Flattrs,
3189 message: wirevalue::Any,
3190 ) {
3191 self.post_with_external_seq_info_and_return(port_id, headers, message, true)
3192 }
3193
3194 #[doc(hidden)]
3201 pub fn post_with_external_seq_info_and_return(
3202 &self,
3203 port_id: impl Into<PortAddr>,
3204 headers: Flattrs,
3205 message: wirevalue::Any,
3206 return_undeliverable: bool,
3207 ) {
3208 <Self as context::MailboxExt>::post(
3209 self,
3210 port_id.into(),
3211 headers,
3212 message,
3213 return_undeliverable,
3214 context::SeqInfoPolicy::AllowExternal,
3215 )
3216 }
3217
3218 fn enqueue_runtime_work(&self, work: WorkCell<A>) -> Result<(), ActorError> {
3219 let actor_id_str = self.self_addr().to_string();
3220 account_enqueue(
3221 &self.inner.cell.inner.queue_depth,
3222 &self.inner.proc.state().queue_stats,
3223 &actor_id_str,
3224 );
3225 let result = self
3226 .inner
3227 .ports
3228 .workq
3229 .send(SequencedEnvelope::new(SeqInfo::Direct, None, work))
3230 .map_err(anyhow::Error::from);
3231 if result.is_err() {
3232 account_cancel_enqueue(
3233 &self.inner.cell.inner.queue_depth,
3234 &self.inner.proc.state().queue_stats,
3235 &actor_id_str,
3236 );
3237 }
3238 result.map_err(|err| ActorError::new(self.self_addr(), ActorErrorKind::processing(err)))
3239 }
3240
3241 pub fn self_client() -> &'static Client {
3247 static CLIENT: OnceLock<Client> = OnceLock::new();
3248 CLIENT.get_or_init(|| Proc::global().client("self_message_client"))
3249 }
3250
3251 #[allow(private_bounds)]
3257 pub fn post_after<D, M>(&self, dest: D, message: M, delay: Duration)
3258 where
3259 M: Message,
3260 D: PostAfterEndpoint<A, M>,
3261 {
3262 let dest_location = dest.endpoint_location();
3263 if matches!(*self.inner.cell.status().borrow(), ActorStatus::Client) {
3264 self.report_delivery_failure(
3265 crate::mailbox::DeliveryFailureReport::link_unavailable::<M>(
3266 self.mailbox().actor_addr().clone(),
3267 dest_location,
3268 "delayed posts require an actor runtime",
3269 ),
3270 );
3271 return;
3272 }
3273 let Ok(_guard) = self.inner.delayed_posts.ingress.try_enter() else {
3274 self.report_delivery_failure(
3275 crate::mailbox::DeliveryFailureReport::link_unavailable::<M>(
3276 self.mailbox().actor_addr().clone(),
3277 dest_location,
3278 "actor runtime is stopping",
3279 ),
3280 );
3281 return;
3282 };
3283 if self.is_stopping() || self.is_terminal() {
3284 self.report_delivery_failure(
3285 crate::mailbox::DeliveryFailureReport::link_unavailable::<M>(
3286 self.mailbox().actor_addr().clone(),
3287 dest_location,
3288 "actor runtime is stopping",
3289 ),
3290 );
3291 return;
3292 }
3293
3294 self.inner.delayed_posts.push(
3295 tokio::time::Instant::now() + delay,
3296 dest.into_delayed_post(message),
3297 );
3298 }
3299
3300 fn start(self, actor: A, receivers: InstanceReceivers<A>) -> ActorHandle<A> {
3303 let instance_cell = self.inner.cell.clone();
3304 let actor_id = self.inner.cell.actor_addr().clone();
3305 let actor_handle = ActorHandle::new(self.inner.cell.clone(), self.inner.ports.clone());
3306
3307 let ports = Arc::downgrade(&self.inner.ports);
3308 self.inner.mailbox.set_return_handler_hook(move || {
3309 if let Some(ports) = ports.upgrade() {
3310 ports.bind::<Undeliverable<MessageEnvelope>>();
3311 }
3312 });
3313
3314 self.spawn_introspect(receivers.introspect);
3318
3319 let actor_loop_receivers = receivers
3320 .actor_loop
3321 .expect("non-detached instance must have actor loop receivers");
3322 let actor_task_handle = A::spawn_server_task(
3323 panic_handler::with_backtrace_tracking(self.serve(
3324 actor,
3325 actor_loop_receivers,
3326 receivers.work,
3327 ))
3328 .instrument(Span::current()),
3329 );
3330 tracing::debug!("{}: spawned with {:?}", actor_id, actor_task_handle);
3331 instance_cell
3332 .inner
3333 .actor_task_handle
3334 .set(actor_task_handle)
3335 .unwrap_or_else(|_| panic!("{}: task handle store failed", actor_id));
3336
3337 actor_handle
3338 }
3339
3340 async fn serve(
3341 mut self,
3342 mut actor: A,
3343 actor_loop_receivers: (
3344 mpsc::UnboundedReceiver<Signal>,
3345 mpsc::UnboundedReceiver<ActorSupervisionEvent>,
3346 ),
3347 mut work_rx: ActorWorkReceiver<A>,
3348 ) {
3349 let result = self
3350 .run_actor_tree(&mut actor, actor_loop_receivers, &mut work_rx)
3351 .await;
3352
3353 assert!(self.is_stopping());
3354 let (terminal_status, event) = match result {
3360 Ok(stop_reason) => {
3361 let status = ActorStatus::Stopped(stop_reason);
3362 let event = ActorSupervisionEvent::new(
3363 self.inner.cell.actor_addr().clone(),
3364 actor.display_name(),
3365 status.clone(),
3366 None,
3367 );
3368 (status, Some(event))
3369 }
3370 Err(err) => match *err.kind {
3371 ActorErrorKind::UnhandledSupervisionEvent(box event) => {
3372 assert!(event.actor_status.is_terminal());
3374 let status = event.actor_status.clone();
3375 (status, Some(event))
3376 }
3377 ActorErrorKind::SyntheticSupervision(synthetic) => {
3378 let local_fence = synthetic.local_fence.clone();
3379 let error_kind = ActorErrorKind::SyntheticSupervision(synthetic);
3380 let status = ActorStatus::Failed(error_kind);
3381 let event = ActorSupervisionEvent::new(
3382 self.inner.cell.actor_addr().clone(),
3383 actor.display_name(),
3384 status.clone(),
3385 None,
3386 )
3387 .with_local_fence(local_fence);
3388 (status, Some(event))
3389 }
3390 _ => {
3391 let error_kind = ActorErrorKind::Generic(err.kind.to_string());
3392 let status = ActorStatus::Failed(error_kind);
3393 let event = ActorSupervisionEvent::new(
3394 self.inner.cell.actor_addr().clone(),
3395 actor.display_name(),
3396 status.clone(),
3397 None,
3398 );
3399 (status, Some(event))
3400 }
3401 },
3402 };
3403
3404 let current_status = self.inner.cell.status().borrow().clone();
3405 let event_to_deliver = if current_status.is_zombie() {
3408 Some(ActorSupervisionEvent::new(
3409 self.inner.cell.actor_addr().clone(),
3410 actor.display_name(),
3411 current_status,
3412 None,
3413 ))
3414 } else {
3415 event.clone()
3416 };
3417
3418 self.mailbox().close(terminal_status.clone());
3419 if let Some(event) = &event {
3421 *self.inner.cell.inner.supervision_event.lock().unwrap() = Some(event.clone());
3422 }
3423
3424 if let Some(parent) = self.inner.cell.maybe_unlink_parent() {
3429 if let Some(event) = event_to_deliver {
3430 parent.send_supervision_event_or_crash(event);
3432 }
3433 if let Err(err) = parent.signal(Signal::ChildStopped(self.inner.cell.uid().clone())) {
3436 tracing::error!(
3437 "{}: failed to send stop message to parent uid {}: {:?}",
3438 self.self_addr(),
3439 parent.uid(),
3440 err
3441 );
3442 }
3443 } else {
3444 if let Some(event) = event_to_deliver {
3450 self.inner
3451 .proc
3452 .handle_unhandled_supervision_event(&self, event);
3453 }
3454 }
3455
3456 self.stop_introspect(terminal_status.clone()).await;
3457 self.change_status(terminal_status);
3458 }
3459
3460 async fn run_actor_tree(
3464 &mut self,
3465 actor: &mut A,
3466 mut actor_loop_receivers: (
3467 mpsc::UnboundedReceiver<Signal>,
3468 mpsc::UnboundedReceiver<ActorSupervisionEvent>,
3469 ),
3470 work_rx: &mut ActorWorkReceiver<A>,
3471 ) -> Result<String, ActorError> {
3472 let mut did_panic = false;
3478 let result = match AssertUnwindSafe(self.run(actor, &mut actor_loop_receivers, work_rx))
3479 .catch_unwind()
3480 .await
3481 {
3482 Ok(result) => result,
3483 Err(_) => {
3484 did_panic = true;
3485 let panic_info = panic_handler::take_panic_info()
3486 .map(|info| info.to_string())
3487 .unwrap_or_else(|e| format!("Cannot take backtrace due to: {:?}", e));
3488 Err(ActorError::new(
3489 self.self_addr(),
3490 ActorErrorKind::panic(anyhow::anyhow!(panic_info)),
3491 ))
3492 }
3493 };
3494
3495 assert!(!self.is_terminal());
3496 if !self.inner.cell.status().borrow().is_zombie() {
3499 self.change_status(ActorStatus::stopping());
3500 }
3501 if let Err(err) = &result {
3502 tracing::error!("{}: actor failure: {}", self.self_addr(), err);
3503 }
3504
3505 let mut to_unlink = Vec::new();
3508 let child_signal = match ChildTeardown::from_run_result(&result) {
3509 ChildTeardown::Cooperative(mode) => ChildTeardown::cooperative_signal(mode),
3510 ChildTeardown::Kill => {
3511 ChildTeardown::cooperative_signal(StopMode::Stop)
3514 }
3515 };
3516 for child in self.inner.cell.child_iter() {
3517 if let Err(err) = child.value().signal(child_signal.clone()) {
3518 tracing::error!(
3519 "{}: failed to send stop signal to child pid {}: {:?}",
3520 self.self_addr(),
3521 child.key(),
3522 err
3523 );
3524 to_unlink.push(child.value().clone());
3525 }
3526 }
3527 for child in to_unlink {
3529 self.inner.cell.unlink(&child);
3530 }
3531
3532 let (mut signal_receiver, _) = actor_loop_receivers;
3533 while self.inner.cell.child_count() > 0 {
3534 match tokio::time::timeout(Duration::from_millis(500), signal_receiver.recv()).await {
3535 Ok(Some(Signal::ChildStopped(uid))) => {
3536 assert!(self.inner.cell.get_child(&uid).is_none());
3537 }
3538 Ok(Some(_)) => {}
3541 Ok(None) => {
3542 self.inner.cell.unlink_all();
3547 break;
3548 }
3549 Err(_) => {
3550 tracing::warn!(
3551 "timeout waiting for ChildStopped signal from child on actor: {}, ignoring",
3552 self.self_addr()
3553 );
3554 self.inner.cell.unlink_all();
3557 break;
3558 }
3559 }
3560 }
3561 let cleanup_result = if !did_panic {
3567 let cleanup_timeout = hyperactor_config::global::get(config::CLEANUP_TIMEOUT);
3568 match tokio::time::timeout(
3569 cleanup_timeout,
3570 self.inner
3571 .proc
3572 .with_current(actor.cleanup(self, result.as_ref().err())),
3573 )
3574 .await
3575 {
3576 Ok(Ok(x)) => Ok(x),
3577 Ok(Err(e)) => Err(ActorError::new(
3578 self.self_addr(),
3579 ActorErrorKind::cleanup(e),
3580 )),
3581 Err(e) => Err(ActorError::new(
3582 self.self_addr(),
3583 ActorErrorKind::cleanup(e.into()),
3584 )),
3585 }
3586 } else {
3587 Ok(())
3588 };
3589 if let Err(ref actor_err) = result {
3590 if let Err(ref err) = cleanup_result {
3593 tracing::warn!(
3594 cleanup_err = %err,
3595 %actor_err,
3596 "ignoring cleanup error after actor error",
3597 );
3598 }
3599 }
3600 result.and_then(|stopped| cleanup_result.map(|_| stopped.reason))
3603 }
3604
3605 async fn run(
3612 &mut self,
3613 actor: &mut A,
3614 actor_loop_receivers: &mut (
3615 mpsc::UnboundedReceiver<Signal>,
3616 mpsc::UnboundedReceiver<ActorSupervisionEvent>,
3617 ),
3618 work_rx: &mut ActorWorkReceiver<A>,
3619 ) -> Result<ActorStopped, ActorError> {
3620 let (signal_receiver, supervision_event_receiver) = actor_loop_receivers;
3621 let mut stop_mode = StopMode::Stop;
3622
3623 self.change_status(ActorStatus::Initializing);
3624 self.inner
3625 .proc
3626 .with_current(actor.init(self))
3627 .await
3628 .map_err(|err| ActorError::new(self.self_addr(), ActorErrorKind::init(err)))?;
3629 let actor_id_str = self.self_addr().to_string();
3630 let stop_reason = 'messages: loop {
3631 if !self.is_stopping() {
3632 self.change_status(ActorStatus::Idle);
3633 }
3634 let next_delayed_deadline = self.inner.delayed_posts.next_deadline();
3635 let metric_pairs = hyperactor_telemetry::kv_pairs!("actor_id" => actor_id_str.clone());
3636 tokio::select! {
3637 biased;
3638 signal = signal_receiver.recv() => {
3639 let signal = signal.ok_or_else(|| {
3640 ActorError::new(self.self_addr(), ActorErrorKind::SignalChannelClosed)
3641 })?;
3642 tracing::debug!("received signal {signal:?}");
3643 match signal {
3644 Signal::Stop(reason) => {
3645 stop_mode = StopMode::Stop;
3646 self.change_status(ActorStatus::stopping());
3647 self.inner
3648 .proc
3649 .with_current(actor.handle_stop(self, StopMode::Stop, &reason))
3650 .await
3651 .map_err(|err| ActorError::new(self.self_addr(), ActorErrorKind::processing(err)))?;
3652 },
3653 Signal::DrainAndStop(reason) => {
3654 stop_mode = StopMode::DrainAndStop;
3655 self.change_status(ActorStatus::stopping());
3656 self.inner
3657 .proc
3658 .with_current(actor.handle_stop(self, StopMode::DrainAndStop, &reason))
3659 .await
3660 .map_err(|err| ActorError::new(self.self_addr(), ActorErrorKind::processing(err)))?;
3661 },
3662 Signal::ChildStopped(uid) => {
3663 assert!(self.inner.cell.get_child(&uid).is_none());
3664 },
3665 Signal::ExitRequested(reason) => {
3666 break 'messages reason;
3667 }
3668 Signal::Kill(reason) => {
3669 return Err(ActorError { actor_id: Box::new(self.self_addr().clone()), kind: Box::new(ActorErrorKind::Aborted(reason)) });
3670 }
3671 }
3672 }
3673 work = work_rx.recv() => {
3674 ACTOR_MESSAGES_RECEIVED.add(1, metric_pairs);
3675 account_dequeue(&self.inner.cell.inner.queue_depth, &self.inner.proc.state().queue_stats, &actor_id_str);
3676 let _timer = ACTOR_MESSAGE_HANDLER_DURATION.start(metric_pairs);
3677 let work = work.expect("inconsistent work queue state");
3678 if let Err(err) = work.handle(actor, self).await {
3679 while let Ok(supervision_event) = supervision_event_receiver.try_recv() {
3680 self.handle_supervision_event(actor, supervision_event).await?;
3681 }
3682 let kind = ActorErrorKind::processing(err);
3683 return Err(ActorError {
3684 actor_id: Box::new(self.self_addr().clone()),
3685 kind: Box::new(kind),
3686 });
3687 }
3688 }
3689 _ = self.inner.delayed_posts.notify.notified(), if !self.is_stopping() && !self.inner.delayed_posts.is_draining() => {
3690 }
3691 _ = async {
3692 match next_delayed_deadline {
3693 Some(deadline) => tokio::time::sleep_until(deadline).await,
3694 None => std::future::pending::<()>().await,
3695 }
3696 }, if !self.is_stopping() && !self.inner.delayed_posts.is_draining() && next_delayed_deadline.is_some() => {
3697 let now = tokio::time::Instant::now();
3698 if let Ok(_guard) = self.inner.delayed_posts.ingress.try_enter() {
3699 for post in self.inner.delayed_posts.pop_due(now) {
3700 post(self);
3701 }
3702 }
3703 }
3704 Some(supervision_event) = supervision_event_receiver.recv() => {
3705 self.handle_supervision_event(actor, supervision_event).await?;
3706 }
3707 }
3708 self.inner
3709 .cell
3710 .inner
3711 .num_processed_messages
3712 .fetch_add(1, Ordering::SeqCst);
3713 };
3714 tracing::debug!(
3715 actor_id = %self.self_addr(),
3716 reason = stop_reason,
3717 "exited actor loop",
3718 );
3719 Ok(ActorStopped {
3720 reason: stop_reason,
3721 stop_mode,
3722 })
3723 }
3724
3725 pub async fn handle_supervision_event(
3727 &self,
3728 actor: &mut A,
3729 supervision_event: ActorSupervisionEvent,
3730 ) -> Result<(), ActorError> {
3731 if supervision_event.is_locally_cancelled() {
3732 return Ok(());
3733 }
3734
3735 match self
3737 .inner
3738 .proc
3739 .with_current(actor.handle_supervision_event(self, &supervision_event))
3740 .await
3741 {
3742 Ok(true) => {
3743 Ok(())
3745 }
3746 Ok(false) => {
3747 let kind = ActorErrorKind::UnhandledSupervisionEvent(Box::new(supervision_event));
3748 Err(ActorError::new(self.self_addr(), kind))
3749 }
3750 Err(err) => {
3751 let kind = ActorErrorKind::ErrorDuringHandlingSupervision(
3754 err.to_string(),
3755 Box::new(supervision_event),
3756 );
3757 Err(ActorError::new(self.self_addr(), kind))
3758 }
3759 }
3760 }
3761
3762 async unsafe fn handle_message<M: Message>(
3766 &self,
3767 actor: &mut A,
3768 type_info: Option<&'static TypeInfo>,
3769 headers: Flattrs,
3770 message: M,
3771 ) -> Result<(), anyhow::Error>
3772 where
3773 A: Handler<M>,
3774 {
3775 let handler_info = match type_info {
3777 Some(info) => {
3778 let arm = unsafe { info.arm_unchecked(&message as *const M as *const ()) };
3780 HandlerInfo::from_static(info.typename(), arm)
3781 }
3782 None => {
3783 HandlerInfo::from_static(std::any::type_name::<M>(), None)
3785 }
3786 };
3787
3788 let endpoint = type_info.and_then(|info| {
3789 unsafe { info.endpoint_name(&message as *const M as *const ()) }
3791 });
3792
3793 self.handle_message_with_handler_info(actor, handler_info, headers, message, endpoint)
3795 .await
3796 }
3797
3798 #[tracing::instrument(level = "debug", name = "handle_message", skip_all, fields(message_type = %handler_info))]
3802 async fn handle_message_with_handler_info<M: Message>(
3803 &self,
3804 actor: &mut A,
3805 handler_info: HandlerInfo,
3806 headers: Flattrs,
3807 message: M,
3808 endpoint: Option<String>,
3809 ) -> Result<(), anyhow::Error>
3810 where
3811 A: Handler<M>,
3812 {
3813 let now = std::time::SystemTime::now();
3814 let handler_info = Some(handler_info);
3815 self.change_status(ActorStatus::Processing(now, handler_info.clone()));
3816 crate::mailbox::headers::log_message_latency_if_sampling(&headers, self.self_addr());
3817
3818 let message_id = headers.get(crate::mailbox::headers::TELEMETRY_MESSAGE_ID);
3819
3820 if let Some(message_id) = message_id {
3821 let from_actor_id = headers
3822 .get(crate::mailbox::headers::SENDER_ACTOR_ID_HASH)
3823 .unwrap_or(0);
3824 let to_actor_id = hash_to_u64(self.self_addr().id());
3825 let port_index = headers.get(crate::mailbox::headers::TELEMETRY_PORT_INDEX);
3826
3827 notify_message(hyperactor_telemetry::MessageEvent {
3828 timestamp: now,
3829 id: message_id,
3830 from_actor_id,
3831 to_actor_id,
3832 endpoint,
3833 port_index,
3834 });
3835
3836 notify_message_status(hyperactor_telemetry::MessageStatusEvent {
3837 timestamp: now,
3838 id: hyperactor_telemetry::generate_status_event_id(message_id),
3839 message_id,
3840 status: "active".to_string(),
3841 });
3842 }
3843
3844 *self.inner.cell.inner.last_message_handler.write().unwrap() = handler_info;
3846
3847 let context = Context::new(self, headers);
3848 let start = Instant::now();
3852 let subject_str = self.self_addr().subject().to_string();
3853 let result = self
3854 .inner
3855 .proc
3856 .with_current(actor.handle(&context, message))
3857 .instrument(self.inner.cell.inner.recording.span(&subject_str))
3858 .await;
3859 let elapsed_us = start.elapsed().as_micros() as u64;
3860 self.inner
3861 .cell
3862 .inner
3863 .total_processing_time_us
3864 .fetch_add(elapsed_us, Ordering::SeqCst);
3865
3866 let terminal_status = match (&result, A::message_status_reporting()) {
3867 (Err(_), _) => Some("failed"),
3868 (Ok(_), MessageStatusReporting::Immediate) => Some("complete"),
3869 (Ok(_), MessageStatusReporting::Deferred) => None,
3870 };
3871 if let (Some(message_id), Some(status)) = (message_id, terminal_status) {
3872 notify_message_status(hyperactor_telemetry::MessageStatusEvent {
3873 timestamp: std::time::SystemTime::now(),
3874 id: hyperactor_telemetry::generate_status_event_id(message_id),
3875 message_id,
3876 status: status.to_string(),
3877 });
3878 }
3879
3880 result
3881 }
3882
3883 pub fn spawn<C: Actor>(&self, actor: C) -> ActorHandle<C> {
3885 self.inner.proc.spawn_child(self.inner.cell.clone(), actor)
3886 }
3887
3888 pub fn spawn_with_name<C: Actor>(&self, name: &str, actor: C) -> ActorHandle<C> {
3892 self.inner
3893 .proc
3894 .spawn_named_child(self.inner.cell.clone(), name, actor)
3895 }
3896
3897 pub fn spawn_with_label<C: Actor>(&self, label: &str, actor: C) -> ActorHandle<C> {
3902 self.inner
3903 .proc
3904 .spawn_named_child(self.inner.cell.clone(), label, actor)
3905 }
3906
3907 pub fn spawn_with_uid<C: Actor>(&self, uid: Uid, actor: C) -> anyhow::Result<ActorHandle<C>> {
3913 self.inner
3914 .proc
3915 .spawn_child_with_uid(self.inner.cell.clone(), uid, actor)
3916 }
3917
3918 pub fn child(&self) -> (Instance<()>, ActorHandle<()>) {
3920 self.inner.proc.child_instance(self.inner.cell.clone())
3921 }
3922
3923 pub async fn gspawn(&self, actor_type: &str, params: Data) -> anyhow::Result<AnyActorHandle> {
3928 self.gspawn_uid(actor_type, crate::id::Uid::anonymous(), params)
3929 .await
3930 }
3931
3932 pub async fn gspawn_uid(
3937 &self,
3938 actor_type: &str,
3939 uid: crate::id::Uid,
3940 params: Data,
3941 ) -> anyhow::Result<AnyActorHandle> {
3942 crate::actor::remote::Remote::global()
3943 .gspawn_child(
3944 &self.inner.proc,
3945 self.inner.cell.clone(),
3946 actor_type,
3947 uid,
3948 params,
3949 )
3950 .await
3951 }
3952
3953 #[doc(hidden)]
3959 pub async fn gspawn_uid_in_environment(
3960 &self,
3961 actor_type: &str,
3962 uid: crate::id::Uid,
3963 params: Data,
3964 environment: ActorEnvironment,
3965 ) -> anyhow::Result<AnyActorHandle> {
3966 crate::actor::remote::Remote::global()
3967 .gspawn_child_in_environment(
3968 &self.inner.proc,
3969 self.inner.cell.clone(),
3970 actor_type,
3971 uid,
3972 params,
3973 environment,
3974 )
3975 .await
3976 }
3977
3978 pub fn port<M: Message>(&self) -> PortHandle<M>
3981 where
3982 A: Handler<M>,
3983 {
3984 self.inner.ports.get()
3985 }
3986
3987 pub fn handle(&self) -> ActorHandle<A> {
3989 ActorHandle::new(self.inner.cell.clone(), Arc::clone(&self.inner.ports))
3990 }
3991
3992 pub fn bind<R: Binds<A>>(&self) -> ActorRef<R> {
3994 self.inner.cell.bind(self.inner.ports.as_ref())
3995 }
3996
3997 #[doc(hidden)]
3999 pub fn mailbox_for_py(&self) -> &Mailbox {
4000 &self.inner.mailbox
4001 }
4002
4003 pub fn proc(&self) -> &Proc {
4005 &self.inner.proc
4006 }
4007
4008 pub fn actor_environment(&self) -> &ActorEnvironment {
4014 self.inner.cell.actor_environment()
4015 }
4016
4017 #[doc(hidden)]
4021 pub fn clone_for_py(&self) -> Self {
4022 Self {
4023 inner: Arc::clone(&self.inner),
4024 }
4025 }
4026
4027 fn actor_task_handle(&self) -> Option<&JoinHandle<()>> {
4029 self.inner.cell.inner.actor_task_handle.get()
4030 }
4031
4032 pub fn sequencer(&self) -> &Sequencer {
4034 &self.inner.sequencer
4035 }
4036
4037 #[doc(hidden)]
4046 pub fn debug_skip_next_ordering_seq(&self, dest: &PortAddr, count: u64) {
4047 let sequencer = self.sequencer();
4048 for _ in 0..count {
4049 let _ = sequencer.assign_seq(dest);
4050 }
4051 }
4052
4053 pub fn instance_id(&self) -> Uuid {
4055 self.inner.id
4056 }
4057
4058 pub fn parent_handle<P: Actor>(&self) -> Option<ActorHandle<P>> {
4060 let parent_cell = self.inner.cell.parent()?;
4061 let ports = if let Ok(ports) = parent_cell.inner.ports.clone().downcast() {
4062 ports
4063 } else {
4064 return None;
4065 };
4066 Some(ActorHandle::new(parent_cell, ports))
4067 }
4068}
4069
4070impl Instance<ClientActor> {
4071 pub(crate) fn child_client(&self) -> Client {
4072 let actor_id = self
4073 .inner
4074 .proc
4075 .allocate_anonymous_child_id(self.inner.cell.actor_addr());
4076 let (instance, _receivers) = Instance::new(
4077 self.inner.proc.clone(),
4078 actor_id,
4079 false,
4080 Some(self.inner.cell.clone()),
4081 self.inner.cell.actor_environment().clone(),
4082 );
4083 instance.change_status(ActorStatus::Client);
4084 Client::new(instance)
4085 }
4086}
4087
4088impl<A: Actor> context::Mailbox for Instance<A> {
4089 fn mailbox(&self) -> &Mailbox {
4090 &self.inner.mailbox
4091 }
4092}
4093
4094impl<A: Actor> context::Mailbox for Context<'_, A> {
4095 fn mailbox(&self) -> &Mailbox {
4096 &self.instance.inner.mailbox
4097 }
4098}
4099
4100impl<A: Actor> context::Mailbox for &Instance<A> {
4101 fn mailbox(&self) -> &Mailbox {
4102 &self.inner.mailbox
4103 }
4104}
4105
4106impl<A: Actor> context::Mailbox for &Context<'_, A> {
4107 fn mailbox(&self) -> &Mailbox {
4108 &self.instance.inner.mailbox
4109 }
4110}
4111
4112impl<A: Actor> context::Actor for Instance<A> {
4113 type A = A;
4114 fn instance(&self) -> &Instance<A> {
4115 self
4116 }
4117}
4118
4119impl<A: Actor> context::Actor for Context<'_, A> {
4120 type A = A;
4121 fn instance(&self) -> &Instance<A> {
4122 self
4123 }
4124
4125 fn headers(&self) -> &Flattrs {
4126 Context::headers(self)
4127 }
4128}
4129
4130impl<A: Actor> context::Actor for &Instance<A> {
4131 type A = A;
4132 fn instance(&self) -> &Instance<A> {
4133 self
4134 }
4135}
4136
4137impl<A: Actor> context::Actor for &Context<'_, A> {
4138 type A = A;
4139 fn instance(&self) -> &Instance<A> {
4140 self
4141 }
4142
4143 fn headers(&self) -> &Flattrs {
4144 Context::headers(self)
4145 }
4146}
4147
4148impl<A, M> crate::Endpoint<M> for &Instance<A>
4149where
4150 A: Actor + Handler<M>,
4151 M: Message,
4152{
4153 fn endpoint_location(&self) -> crate::EndpointLocation {
4154 crate::EndpointLocation::Actor(self.self_addr().clone())
4155 }
4156
4157 fn post<C>(self, cx: &C, message: M)
4158 where
4159 C: context::Actor,
4160 {
4161 let port = self.port();
4162 crate::Endpoint::post(&port, cx, message)
4163 }
4164}
4165
4166impl<A, M> crate::Endpoint<M> for &Context<'_, A>
4167where
4168 A: Actor + Handler<M>,
4169 M: Message,
4170{
4171 fn endpoint_location(&self) -> crate::EndpointLocation {
4172 crate::EndpointLocation::Actor(self.self_addr().clone())
4173 }
4174
4175 fn post<C>(self, cx: &C, message: M)
4176 where
4177 C: context::Actor,
4178 {
4179 crate::Endpoint::post(self.instance, cx, message)
4180 }
4181}
4182
4183impl<A, M> crate::Endpoint<M> for Instance<A>
4184where
4185 A: Actor + Handler<M>,
4186 M: Message,
4187{
4188 fn endpoint_location(&self) -> crate::EndpointLocation {
4189 crate::EndpointLocation::Actor(self.self_addr().clone())
4190 }
4191
4192 fn post<C>(self, cx: &C, message: M)
4193 where
4194 C: context::Actor,
4195 {
4196 crate::Endpoint::post(&self, cx, message)
4197 }
4198}
4199
4200impl Instance<()> {
4201 pub fn bind_handler_port<M: RemoteMessage>(&self) -> (PortHandle<M>, PortReceiver<M>) {
4203 assert!(
4204 self.actor_task_handle().is_none(),
4205 "can only bind handler port on instance with no running actor task"
4206 );
4207 self.inner.mailbox.bind_handler_port()
4208 }
4209}
4210
4211#[derive(Debug)]
4212enum ActorType {
4213 Named(&'static TypeInfo),
4214 Anonymous(&'static str),
4215}
4216
4217impl ActorType {
4218 fn type_name(&self) -> &str {
4219 match self {
4220 ActorType::Named(info) => info.typename(),
4221 ActorType::Anonymous(name) => name,
4222 }
4223 }
4224}
4225
4226#[derive(Clone)]
4232pub struct InstanceCell {
4233 inner: Arc<InstanceCellState>,
4234}
4235
4236impl fmt::Debug for InstanceCell {
4237 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
4238 f.debug_struct("InstanceCell")
4239 .field("actor_id", &self.inner.actor_id)
4240 .field("actor_type", &self.inner.actor_type)
4241 .finish()
4242 }
4243}
4244
4245struct InstanceCellState {
4246 actor_id: ActorAddr,
4248
4249 actor_environment: ActorEnvironment,
4251
4252 instance_id: Uuid,
4256
4257 actor_type: ActorType,
4259
4260 proc: Proc,
4262
4263 mailbox: Mailbox,
4265
4266 actor_loop: Option<(
4268 mpsc::UnboundedSender<Signal>,
4269 mpsc::UnboundedSender<ActorSupervisionEvent>,
4270 )>,
4271
4272 status_tx: watch::Sender<ActorStatus>,
4274
4275 status: watch::Receiver<ActorStatus>,
4277
4278 terminal_status_subscriptions: Mutex<TerminalSubscriptions>,
4280
4281 parent: WeakInstanceCell,
4283
4284 children: DashMap<crate::id::Uid, InstanceCell>,
4286
4287 actor_task_handle: OnceLock<JoinHandle<()>>,
4289
4290 exported_named_ports: DashMap<Port, &'static str>,
4292
4293 num_processed_messages: AtomicU64,
4295
4296 created_at: SystemTime,
4298
4299 last_message_handler: RwLock<Option<HandlerInfo>>,
4301
4302 total_processing_time_us: AtomicU64,
4304
4305 queue_depth: Arc<AtomicU64>,
4316
4317 recording: Recording,
4320
4321 published_attrs: RwLock<Option<hyperactor_config::Attrs>>,
4330
4331 query_child_handler: RwLock<Option<Box<dyn (Fn(&Addr) -> IntrospectResult) + Send + Sync>>>,
4339
4340 supervision_event: std::sync::Mutex<Option<crate::supervision::ActorSupervisionEvent>>,
4345
4346 is_system: AtomicBool,
4350
4351 ports: Arc<dyn Any + Send + Sync>,
4354
4355 inbound_ordering_snapshot:
4370 Option<Box<dyn Fn() -> crate::ordering::OrderingSnapshot + Send + Sync>>,
4371
4372 actor_attrs_snapshot: RwLock<Option<Box<dyn Fn() -> hyperactor_config::Attrs + Send + Sync>>>,
4385}
4386
4387impl InstanceCellState {
4388 fn maybe_unlink_parent(&self) -> Option<InstanceCell> {
4391 self.parent
4392 .upgrade()
4393 .filter(|parent| parent.inner.unlink(self))
4394 }
4395
4396 fn unlink(&self, child: &InstanceCellState) -> bool {
4398 assert_eq!(self.actor_id.proc_id(), child.actor_id.proc_id());
4399 self.children.remove(child.actor_id.uid()).is_some()
4400 }
4401}
4402
4403fn select_eviction_candidates(
4416 entries: &[(ActorAddr, Option<String>)],
4417 excess: usize,
4418) -> Vec<ActorAddr> {
4419 let mut clean: Vec<&ActorAddr> = Vec::new();
4420 let mut failed: Vec<(&ActorAddr, &str)> = Vec::new();
4421 for (id, occurred_at) in entries {
4422 match occurred_at {
4423 Some(ts) => failed.push((id, ts.as_str())),
4424 None => clean.push(id),
4425 }
4426 }
4427
4428 let mut to_remove: Vec<ActorAddr> = Vec::new();
4429 let mut remaining = excess;
4430
4431 for id in clean {
4433 if remaining == 0 {
4434 break;
4435 }
4436 to_remove.push(id.clone());
4437 remaining -= 1;
4438 }
4439
4440 if remaining > 0 {
4442 failed.sort_by(|a, b| b.1.cmp(a.1));
4443 for (id, _) in failed.into_iter().take(remaining) {
4444 to_remove.push(id.clone());
4445 }
4446 }
4447
4448 to_remove
4449}
4450
4451#[derive(Copy, Clone, Debug, Eq, PartialEq)]
4452enum IllegalStatusChange {
4453 LinkedZombie,
4454 TerminalRewrite,
4455}
4456
4457#[derive(Copy, Clone, Debug, Eq, PartialEq)]
4458enum StatusChange {
4459 Apply,
4460 Ignore,
4461 Illegal(IllegalStatusChange),
4462}
4463
4464fn classify_status_change(old: &ActorStatus, new: &ActorStatus) -> StatusChange {
4465 if old == new || old.is_zombie() && !new.is_terminal() || old.is_terminal() && new.is_zombie() {
4468 StatusChange::Ignore
4469 } else if old.is_terminal() {
4470 StatusChange::Illegal(IllegalStatusChange::TerminalRewrite)
4471 } else {
4472 StatusChange::Apply
4473 }
4474}
4475
4476impl InstanceCell {
4477 #[allow(clippy::too_many_arguments)]
4480 fn new(
4481 actor_id: ActorAddr,
4482 instance_id: Uuid,
4483 actor_type: ActorType,
4484 actor_environment: ActorEnvironment,
4485 proc: Proc,
4486 mailbox: Mailbox,
4487 actor_loop: Option<(
4488 mpsc::UnboundedSender<Signal>,
4489 mpsc::UnboundedSender<ActorSupervisionEvent>,
4490 )>,
4491 status_tx: watch::Sender<ActorStatus>,
4492 status: watch::Receiver<ActorStatus>,
4493 parent: Option<InstanceCell>,
4494 ports: Arc<dyn Any + Send + Sync>,
4495 queue_depth: Arc<AtomicU64>,
4496 inbound_ordering_snapshot: Option<
4497 Box<dyn Fn() -> crate::ordering::OrderingSnapshot + Send + Sync>,
4498 >,
4499 ) -> Self {
4500 let is_root_instance = parent.is_none();
4501 let _ais = actor_id.to_string();
4502 let cell = Self {
4503 inner: Arc::new(InstanceCellState {
4504 actor_id: actor_id.clone(),
4505 instance_id,
4506 actor_type,
4507 actor_environment,
4508 proc: proc.clone(),
4509 mailbox,
4510 actor_loop,
4511 status_tx,
4512 status,
4513 terminal_status_subscriptions: Mutex::new(TerminalSubscriptions::default()),
4514 parent: parent.map_or_else(WeakInstanceCell::new, |cell| cell.downgrade()),
4515 children: DashMap::new(),
4516 actor_task_handle: OnceLock::new(),
4517 exported_named_ports: DashMap::new(),
4518 num_processed_messages: AtomicU64::new(0),
4519 created_at: std::time::SystemTime::now(),
4520 last_message_handler: RwLock::new(None),
4521 total_processing_time_us: AtomicU64::new(0),
4522 queue_depth,
4523 recording: hyperactor_telemetry::recorder().record(64),
4524 published_attrs: RwLock::new(None),
4525 query_child_handler: RwLock::new(None),
4526 supervision_event: std::sync::Mutex::new(None),
4527 is_system: AtomicBool::new(false),
4528 ports,
4529 inbound_ordering_snapshot,
4530 actor_attrs_snapshot: RwLock::new(None),
4531 }),
4532 };
4533 cell.maybe_link_parent();
4534 let previous = proc.inner.actors.insert(
4536 actor_id.id().clone(),
4537 ActorEntry::Running {
4538 cell: cell.downgrade(),
4539 status: cell.status().clone(),
4540 },
4541 );
4542 if is_root_instance {
4543 proc.inner
4544 .root_instances
4545 .insert(actor_id.id().clone(), cell.downgrade());
4546 }
4547 if let Some(ActorEntry::Pending(pending)) = previous {
4548 for (_, message) in pending {
4549 proc.handle_status(actor_id.id(), message);
4550 }
4551 }
4552 cell
4553 }
4554
4555 fn wrap(inner: Arc<InstanceCellState>) -> Self {
4556 Self { inner }
4557 }
4558
4559 pub fn actor_addr(&self) -> &ActorAddr {
4561 &self.inner.actor_id
4562 }
4563
4564 pub(crate) fn actor_environment(&self) -> &ActorEnvironment {
4566 &self.inner.actor_environment
4567 }
4568
4569 pub(crate) fn proc(&self) -> &Proc {
4571 &self.inner.proc
4572 }
4573
4574 pub(crate) fn uid(&self) -> &crate::id::Uid {
4576 self.inner.actor_id.uid()
4577 }
4578
4579 #[allow(dead_code)]
4581 pub(crate) fn actor_task_handle(&self) -> Option<&JoinHandle<()>> {
4582 self.inner.actor_task_handle.get()
4583 }
4584
4585 pub fn status(&self) -> &watch::Receiver<ActorStatus> {
4587 &self.inner.status
4588 }
4589
4590 fn subscribe_terminal_status(
4591 &self,
4592 subscription_id: Uuid,
4593 subscriber: PortRef<Option<ActorStatus>>,
4594 ) {
4595 let mut subscriptions = self.inner.terminal_status_subscriptions.lock().unwrap();
4596 if subscriptions.cancelled.remove(&subscription_id) {
4597 return;
4598 }
4599 let status = self.status().borrow().clone();
4600 if !status.is_terminal() {
4601 if subscriptions.active.len() == MAX_TERMINAL_STATUS_SUBSCRIBERS {
4602 subscriptions.active.pop_front();
4603 }
4604 subscriptions
4605 .active
4606 .push_back((subscription_id, subscriber));
4607 return;
4608 }
4609
4610 drop(subscriptions);
4611 self.proc()
4612 .send_terminal_status(self.actor_addr(), subscriber, Some(status));
4613 }
4614
4615 fn unsubscribe_terminal_status(&self, subscription_id: Uuid) {
4616 let mut subscriptions = self.inner.terminal_status_subscriptions.lock().unwrap();
4617 match subscriptions
4618 .active
4619 .iter()
4620 .position(|(id, _)| *id == subscription_id)
4621 {
4622 Some(index) => {
4623 subscriptions.active.remove(index);
4624 }
4625 None => {
4626 subscriptions.cancelled.insert(subscription_id);
4627 }
4628 }
4629 }
4630
4631 fn notify_terminal_status(&self, status: ActorStatus) {
4632 let subscriptions =
4633 std::mem::take(&mut *self.inner.terminal_status_subscriptions.lock().unwrap());
4634 for (_, subscriber) in subscriptions.active {
4635 self.proc()
4636 .send_terminal_status(self.actor_addr(), subscriber, Some(status.clone()));
4637 }
4638 }
4639
4640 #[cfg(test)]
4641 fn terminal_status_subscriber_count(&self) -> usize {
4642 self.inner
4643 .terminal_status_subscriptions
4644 .lock()
4645 .unwrap()
4646 .active
4647 .len()
4648 }
4649
4650 #[track_caller]
4653 fn change_status(&self, new: ActorStatus) {
4654 let mut old_status = None;
4655 let mut illegal_status = None;
4656 let changed = self.inner.status_tx.send_if_modified(|status| {
4657 let old = status.clone();
4658 let mut status_change = classify_status_change(&old, &new);
4659 if status_change == StatusChange::Apply && new.is_zombie() && self.parent().is_some() {
4660 status_change = StatusChange::Illegal(IllegalStatusChange::LinkedZombie);
4661 }
4662
4663 match status_change {
4664 StatusChange::Apply => {}
4665 StatusChange::Ignore => return false,
4666 StatusChange::Illegal(reason) => {
4667 illegal_status = Some((old, reason));
4670 return false;
4671 }
4672 }
4673
4674 old_status = Some(old);
4675 *status = new.clone();
4676 true
4677 });
4678
4679 if let Some((old, reason)) = illegal_status {
4680 match reason {
4681 IllegalStatusChange::LinkedZombie => {
4682 panic!(
4683 "zombie actor must be detached from parent before status update: actor_id={}, prev_status={}, status={}",
4684 self.actor_addr(),
4685 old,
4686 new
4687 );
4688 }
4689 IllegalStatusChange::TerminalRewrite => {
4690 panic!(
4691 "actor changing status illegally, only allow non-terminal -> non-terminal \
4692 and non-terminal -> terminal statuses. actor_id={}, prev_status={}, status={}",
4693 self.actor_addr(),
4694 old,
4695 new
4696 );
4697 }
4698 }
4699 }
4700
4701 if !changed {
4702 return;
4703 }
4704 if new.is_terminal() {
4705 self.notify_terminal_status(new.clone());
4706 }
4707 let old = old_status.expect("status change should capture previous status");
4708 if (old.is_idle() && new.is_processing())
4711 || (old.is_processing() && new.is_idle())
4712 || old == new
4713 {
4714 return;
4715 }
4716
4717 let actor_id = hash_to_u64(self.actor_addr().id());
4718 let new_status = new.arm().unwrap_or("unknown");
4719 let change_reason = match &new {
4720 ActorStatus::Failed(reason) => Some(reason.to_string()),
4721 ActorStatus::Stopping(ActorStoppingReason::Zombie(reason)) => Some(reason.clone()),
4722 ActorStatus::Stopped(reason) => Some(reason.clone()),
4723 _ => None,
4724 };
4725
4726 tracing::info!(
4727 name = "ActorStatus",
4728 actor_id = %self.actor_addr(),
4729 actor_name = self.actor_addr().log_name(),
4730 status = new_status,
4731 prev_status = old.arm().unwrap_or("unknown"),
4732 caller = %PanicLocation::caller(),
4733 change_reason = change_reason.as_deref().unwrap_or(""),
4734 );
4735 notify_actor_status_changed(ActorStatusEvent {
4736 id: generate_actor_status_event_id(actor_id),
4737 timestamp: std::time::SystemTime::now(),
4738 actor_id,
4739 new_status: new_status.to_string(),
4740 reason: change_reason,
4741 });
4742 }
4743
4744 fn publish_dropped_status(&self, terminal_status: ActorStatus) {
4745 let actor_addr = self.actor_addr().clone();
4746 let changed = self.inner.status_tx.send_if_modified(|status| {
4747 if status.is_terminal() {
4748 false
4749 } else {
4750 tracing::info!(
4751 name = "ActorStatus",
4752 actor_id = %actor_addr,
4753 actor_name = actor_addr.log_name(),
4754 status = "Stopped",
4755 prev_status = status.arm().unwrap_or("unknown"),
4756 "instance is dropped",
4757 );
4758 *status = terminal_status.clone();
4759 true
4760 }
4761 });
4762 if changed {
4763 self.notify_terminal_status(terminal_status);
4764 }
4765 }
4766
4767 pub fn supervision_event(&self) -> Option<crate::supervision::ActorSupervisionEvent> {
4773 self.inner.supervision_event.lock().unwrap().clone()
4774 }
4775
4776 fn signal_sender(&self) -> mpsc::UnboundedSender<Signal> {
4777 self.inner
4778 .actor_loop
4779 .as_ref()
4780 .map(|(signal_tx, _)| signal_tx.clone())
4781 .unwrap_or_else(|| panic!("{} has no runtime signal sender", self.actor_addr()))
4782 }
4783
4784 pub fn signal(&self, signal: Signal) -> Result<(), ActorError> {
4786 if let Some((signal_tx, _)) = &self.inner.actor_loop {
4787 signal_tx.send(signal).map_err(|_| {
4788 ActorError::new(self.actor_addr(), ActorErrorKind::SignalChannelClosed)
4789 })
4790 } else {
4791 tracing::warn!(
4792 "{}: attempted to send signal {} to detached actor",
4793 self.inner.actor_id,
4794 signal
4795 );
4796 Ok(())
4797 }
4798 }
4799
4800 pub fn send_supervision_event_or_crash(&self, event: ActorSupervisionEvent) {
4809 match &self.inner.actor_loop {
4810 Some((_, supervision_tx)) => {
4811 if let Err(err) = supervision_tx.send(event.clone()) {
4812 if !event.is_error() {
4813 tracing::debug!(
4819 "{}: dropping non-error supervision event {}: {:?}",
4820 self.actor_addr(),
4821 event,
4822 err
4823 );
4824 return;
4825 }
4826 tracing::error!(
4827 "{}: failed to send supervision event to actor: {:?}. Crash the process.",
4828 self.actor_addr(),
4829 err
4830 );
4831 std::process::exit(1);
4832 }
4833 }
4834 None => {
4835 if !event.is_error() {
4836 tracing::debug!(
4837 "{}: dropping non-error supervision event {} to detached actor",
4838 self.actor_addr(),
4839 event,
4840 );
4841 return;
4842 }
4843 tracing::error!(
4844 "{}: failed: {}: cannot send supervision event to detached actor: crashing",
4845 self.actor_addr(),
4846 event,
4847 );
4848 std::process::exit(1);
4849 }
4850 }
4851 }
4852
4853 pub fn downgrade(&self) -> WeakInstanceCell {
4855 WeakInstanceCell {
4856 inner: Arc::downgrade(&self.inner),
4857 }
4858 }
4859
4860 fn link(&self, child: InstanceCell) {
4862 assert_eq!(self.actor_addr().proc_id(), child.actor_addr().proc_id());
4863 self.inner.children.insert(child.uid().clone(), child);
4864 }
4865
4866 fn unlink(&self, child: &InstanceCell) {
4868 assert_eq!(self.actor_addr().proc_id(), child.actor_addr().proc_id());
4869 self.inner.children.remove(child.uid());
4870 }
4871
4872 fn unlink_all(&self) {
4874 self.inner.children.clear();
4875 }
4876
4877 fn maybe_link_parent(&self) {
4879 if let Some(parent) = self.inner.parent.upgrade() {
4880 parent.link(self.clone());
4881 }
4882 }
4883
4884 fn maybe_unlink_parent(&self) -> Option<InstanceCell> {
4887 self.inner.maybe_unlink_parent()
4888 }
4889
4890 fn child_iter(&self) -> impl Iterator<Item = RefMulti<'_, crate::id::Uid, InstanceCell>> {
4893 self.inner.children.iter()
4894 }
4895
4896 pub fn child_count(&self) -> usize {
4898 self.inner.children.len()
4899 }
4900
4901 pub fn child_actor_ids(&self) -> Vec<ActorAddr> {
4903 self.inner
4904 .children
4905 .iter()
4906 .map(|entry| entry.value().actor_addr().clone())
4907 .collect()
4908 }
4909
4910 fn get_child(&self, uid: &crate::id::Uid) -> Option<InstanceCell> {
4912 self.inner.children.get(uid).map(|child| child.clone())
4913 }
4914
4915 pub fn recording(&self) -> &Recording {
4917 &self.inner.recording
4918 }
4919
4920 pub fn created_at(&self) -> SystemTime {
4922 self.inner.created_at
4923 }
4924
4925 pub fn num_processed_messages(&self) -> u64 {
4927 self.inner.num_processed_messages.load(Ordering::SeqCst)
4928 }
4929
4930 pub fn last_message_handler(&self) -> Option<HandlerInfo> {
4932 self.inner.last_message_handler.read().unwrap().clone()
4933 }
4934
4935 pub fn total_processing_time_us(&self) -> u64 {
4937 self.inner.total_processing_time_us.load(Ordering::SeqCst)
4938 }
4939
4940 pub fn queue_depth(&self) -> u64 {
4942 self.inner.queue_depth.load(Ordering::Relaxed)
4943 }
4944
4945 pub fn instance_id(&self) -> Uuid {
4948 self.inner.instance_id
4949 }
4950
4951 pub fn inbound_ordering_snapshot(&self) -> Option<crate::ordering::OrderingSnapshot> {
4956 self.inner.inbound_ordering_snapshot.as_ref().map(|f| f())
4957 }
4958
4959 pub fn delivery_progress_for_session(
4961 &self,
4962 destination: &EndpointId,
4963 session_id: Uuid,
4964 ) -> Option<DeliveryProgress> {
4965 let snapshot = match destination {
4966 EndpointId::Handler => self.inbound_ordering_snapshot(),
4967 EndpointId::Port(port) => self.inner.mailbox.ordering_snapshot(port),
4968 }?;
4969 snapshot.delivery_progress_for_session(session_id)
4970 }
4971
4972 pub fn set_attrs_snapshot(
4978 &self,
4979 callback: impl Fn() -> hyperactor_config::Attrs + Send + Sync + 'static,
4980 ) {
4981 *self.inner.actor_attrs_snapshot.write().unwrap() = Some(Box::new(callback));
4982 }
4983
4984 pub fn actor_attrs_snapshot(&self) -> Option<hyperactor_config::Attrs> {
4989 let guard = self.inner.actor_attrs_snapshot.read().unwrap();
4990 let callback = guard.as_ref()?;
4991 let attrs = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback))
4992 .map_err(|_| {
4993 tracing::warn!(
4994 actor_id = %self.actor_addr(),
4995 "actor_attrs_snapshot callback panicked; omitting actor attrs",
4996 );
4997 })
4998 .ok()?;
4999 #[cfg(debug_assertions)]
5004 for (name, _) in attrs.iter() {
5005 debug_assert!(
5006 inventory::iter::<hyperactor_config::attrs::AttrKeyInfo>().any(|info| {
5007 info.name == name && info.meta.get(hyperactor_config::INTROSPECT).is_some()
5008 }),
5009 "actor_attrs_snapshot callback returned non-INTROSPECT key `{name}`; \
5010 actor introspection keys must carry @meta(INTROSPECT)"
5011 );
5012 }
5013 Some(attrs)
5014 }
5015
5016 pub fn parent(&self) -> Option<InstanceCell> {
5018 self.inner.parent.upgrade()
5019 }
5020
5021 pub fn actor_type_name(&self) -> &str {
5023 self.inner.actor_type.type_name()
5024 }
5025
5026 pub fn set_published_attrs(&self, attrs: hyperactor_config::Attrs) {
5028 *self.inner.published_attrs.write().unwrap() = Some(attrs);
5029 }
5030
5031 pub fn merge_published_attr<T: hyperactor_config::AttrValue>(
5034 &self,
5035 key: hyperactor_config::Key<T>,
5036 value: T,
5037 ) {
5038 self.inner
5039 .published_attrs
5040 .write()
5041 .unwrap()
5042 .get_or_insert_with(hyperactor_config::Attrs::new)
5043 .set(key, value);
5044 }
5045
5046 pub fn published_attrs(&self) -> Option<hyperactor_config::Attrs> {
5048 self.inner.published_attrs.read().unwrap().clone()
5049 }
5050
5051 pub fn set_query_child_handler(
5059 &self,
5060 handler: impl (Fn(&Addr) -> IntrospectResult) + Send + Sync + 'static,
5061 ) {
5062 *self.inner.query_child_handler.write().unwrap() = Some(Box::new(handler));
5063 }
5064
5065 pub fn query_child(&self, child_ref: &Addr) -> Option<IntrospectResult> {
5067 let guard = self.inner.query_child_handler.read().unwrap();
5068 guard.as_ref().map(|handler| handler(child_ref))
5069 }
5070
5071 pub fn is_system(&self) -> bool {
5073 self.inner.is_system.load(Ordering::Relaxed)
5074 }
5075
5076 pub fn store_terminated_snapshot(&self, payload: crate::introspect::IntrospectResult) {
5086 let snapshots = &self.inner.proc.inner.terminated_snapshots;
5087 snapshots.insert(
5088 self.actor_addr().id().clone(),
5089 TerminatedSnapshot {
5090 actor_addr: self.actor_addr().clone(),
5091 payload,
5092 },
5093 );
5094 let max = hyperactor_config::global::get(crate::config::TERMINATED_SNAPSHOT_RETENTION);
5095 let excess = snapshots.len().saturating_sub(max);
5096 if excess > 0 {
5097 let entries: Vec<_> = snapshots
5099 .iter()
5100 .map(|entry| {
5101 let occurred_at = serde_json::from_str::<hyperactor_config::Attrs>(
5102 &entry.value().payload.attrs,
5103 )
5104 .ok()
5105 .and_then(|attrs| {
5106 attrs
5108 .get(crate::introspect::FAILURE_ERROR_MESSAGE)
5109 .cloned()?;
5110 attrs
5112 .get(crate::introspect::FAILURE_OCCURRED_AT)
5113 .map(|t| humantime::format_rfc3339(*t).to_string())
5114 });
5115 (entry.value().actor_addr.clone(), occurred_at)
5116 })
5117 .collect();
5118
5119 for key in select_eviction_candidates(&entries, excess) {
5120 snapshots.remove(key.id());
5121 }
5122 }
5123 }
5124
5125 pub(crate) fn bind<A: Actor, R: Binds<A>>(&self, ports: &HandlerPorts<A>) -> ActorRef<R> {
5128 <R as Binds<A>>::bind(ports);
5129 ports.bind::<Undeliverable<MessageEnvelope>>();
5135 for entry in ports.bound.iter() {
5137 self.inner
5138 .exported_named_ports
5139 .insert(entry.key().clone(), entry.value());
5140 }
5141 ActorRef::attest(ActorAddr::new(
5142 self.actor_addr().id().clone(),
5143 self.inner.proc.default_location(),
5144 ))
5145 }
5146
5147 pub(crate) fn downcast_handle<A: Actor>(&self) -> Option<ActorHandle<A>> {
5149 let ports = Arc::clone(&self.inner.ports)
5150 .downcast::<HandlerPorts<A>>()
5151 .ok()?;
5152 Some(ActorHandle::new(self.clone(), ports))
5153 }
5154
5155 pub fn traverse<F>(&self, f: &mut F)
5159 where
5160 F: FnMut(&InstanceCell, usize),
5161 {
5162 self.traverse_inner(0, f);
5163 }
5164
5165 fn traverse_inner<F>(&self, depth: usize, f: &mut F)
5166 where
5167 F: FnMut(&InstanceCell, usize),
5168 {
5169 f(self, depth);
5170 let mut children: Vec<_> = self.child_iter().map(|r| r.value().clone()).collect();
5172 children.sort_by_key(|c| c.uid().clone());
5173 for child in children {
5174 child.traverse_inner(depth + 1, f);
5175 }
5176 }
5177}
5178
5179impl Drop for InstanceCellState {
5180 fn drop(&mut self) {
5181 if let Some(parent) = self.maybe_unlink_parent() {
5182 tracing::debug!(
5183 "instance {} was dropped with parent {} still linked",
5184 self.actor_id,
5185 parent.actor_addr()
5186 );
5187 }
5188 let status = self.status.borrow().clone();
5189 match self.proc.inner.actors.entry(self.actor_id.id().clone()) {
5190 Entry::Occupied(mut entry) if entry.get().is_running_instance(self) => {
5191 if status.is_terminal() {
5192 entry.insert(ActorEntry::Tombstoned(status));
5193 } else {
5194 entry.remove();
5195 }
5196 }
5197 _ => tracing::error!("instance {} was dropped but not in proc", self.actor_id),
5198 }
5199 self.proc.inner.root_instances.remove(self.actor_id.id());
5200 }
5201}
5202
5203#[derive(Debug, Clone)]
5206pub struct WeakInstanceCell {
5207 inner: Weak<InstanceCellState>,
5208}
5209
5210impl WeakInstanceCell {
5211 fn new() -> Self {
5213 Self { inner: Weak::new() }
5214 }
5215
5216 pub fn upgrade(&self) -> Option<InstanceCell> {
5218 self.inner.upgrade().map(InstanceCell::wrap)
5219 }
5220}
5221
5222pub struct HandlerPorts<A: Actor> {
5227 ports: DashMap<TypeId, Box<dyn Any + Send + Sync + 'static>>,
5228 bound: DashMap<Port, &'static str>,
5229 mailbox: Mailbox,
5230 workq: mpsc::UnboundedSender<SequencedEnvelope<WorkCell<A>>>,
5231 enable_buffering: bool,
5232 queue_depth: Arc<AtomicU64>,
5234 proc_stats: Arc<ProcQueueStats>,
5236}
5237
5238impl<A: Actor> HandlerPorts<A> {
5239 fn new(
5240 mailbox: Mailbox,
5241 workq: mpsc::UnboundedSender<SequencedEnvelope<WorkCell<A>>>,
5242 enable_buffering: bool,
5243 queue_depth: Arc<AtomicU64>,
5244 proc_stats: Arc<ProcQueueStats>,
5245 ) -> Self {
5246 Self {
5247 ports: DashMap::new(),
5248 bound: DashMap::new(),
5249 mailbox,
5250 workq,
5251 enable_buffering,
5252 queue_depth,
5253 proc_stats,
5254 }
5255 }
5256
5257 pub(crate) fn get<M: Message>(&self) -> PortHandle<M>
5259 where
5260 A: Handler<M>,
5261 {
5262 let key = TypeId::of::<M>();
5263 match self.ports.entry(key) {
5264 Entry::Vacant(entry) => {
5265 assert!(
5269 !crate::ordering::is_bypass_workq_type_id(key),
5270 "cannot provision bypass-workq port {} through `Ports::get`; \
5271 it must be pre-registered via `open_message_port` in `Instance::new`",
5272 std::any::type_name::<M>()
5273 );
5274
5275 let type_info = TypeInfo::get_by_typeid(key);
5276 let workq = self.workq.clone();
5277 let enable_buffering = self.enable_buffering;
5278 let actor_id = self.mailbox.actor_addr().to_string();
5279 let enqueue_depth = Arc::clone(&self.queue_depth);
5280 let enqueue_proc_stats = Arc::clone(&self.proc_stats);
5281 let enqueue = move |headers: Flattrs, msg: M| {
5291 let seq_info = if enable_buffering {
5296 match headers.get(SEQ_INFO) {
5297 Some(seq_info) => seq_info,
5298 None => {
5299 let error_msg = format!(
5300 "in enqueue func for {}, buffering is enabled, but SEQ_INFO is not set for message type {}",
5301 actor_id,
5302 std::any::type_name::<M>(),
5303 );
5304 tracing::error!(error_msg);
5305 return Err(anyhow::anyhow!(error_msg));
5306 }
5307 }
5308 } else {
5309 SeqInfo::Direct
5310 };
5311 if !seq_info.is_valid() {
5312 let error_msg = format!(
5313 "in enqueue func for {}, got seq 0 for message type {}",
5314 actor_id,
5315 std::any::type_name::<M>(),
5316 );
5317 tracing::error!(error_msg);
5318 return Err(anyhow::anyhow!(error_msg));
5319 }
5320 let sender = headers.get(crate::mailbox::headers::SENDER_ACTOR_ID);
5321
5322 let work = WorkCell::new(move |actor: &mut A, instance: &Instance<A>| {
5323 Box::pin(async move {
5324 unsafe {
5326 instance
5327 .handle_message(actor, type_info, headers, msg)
5328 .await
5329 }
5330 })
5331 });
5332 account_enqueue(&enqueue_depth, &enqueue_proc_stats, &actor_id);
5339 let result = workq
5342 .send(SequencedEnvelope::new(seq_info, sender, work))
5343 .map_err(anyhow::Error::from);
5344 if result.is_err() {
5345 account_cancel_enqueue(&enqueue_depth, &enqueue_proc_stats, &actor_id);
5346 }
5347 result
5348 };
5349 let port = self.mailbox.open_handler_enqueue_port(enqueue);
5350 entry.insert(Box::new(port.clone()));
5351 port
5352 }
5353 Entry::Occupied(entry) => {
5354 let port = entry.get();
5355 port.downcast_ref::<PortHandle<M>>().unwrap().clone()
5356 }
5357 }
5358 }
5359
5360 pub fn bind<M: RemoteMessage>(&self)
5362 where
5363 A: Handler<M>,
5364 {
5365 let port = Port::handler::<M>();
5366 match self.bound.entry(port.clone()) {
5367 Entry::Vacant(entry) => {
5368 let _ = self.get::<M>().bind();
5369 entry.insert(M::typename());
5370 }
5371 Entry::Occupied(entry) => {
5372 assert_eq!(
5373 *entry.get(),
5374 M::typename(),
5375 "bind {}: port {} already bound to type {}",
5376 M::typename(),
5377 port,
5378 entry.get(),
5379 );
5380 }
5381 }
5382 }
5383
5384 fn bind_enqueue<M: RemoteMessage>(
5385 &self,
5386 enqueue: impl Fn(Flattrs, M) -> Result<(), anyhow::Error> + Send + Sync + 'static,
5387 ) where
5388 A: Handler<M>,
5389 {
5390 let key = TypeId::of::<M>();
5391 match self.ports.entry(key) {
5392 Entry::Vacant(entry) => {
5393 let port = self.mailbox.open_handler_enqueue_port(enqueue);
5394 entry.insert(Box::new(port));
5395 }
5396 Entry::Occupied(_) => panic!("handler port {} already provisioned", M::typename()),
5397 }
5398 self.bind::<M>();
5399 }
5400}
5401
5402#[cfg(test)]
5403mod tests {
5404 use std::assert_matches;
5405 use std::sync::atomic::AtomicBool;
5406 use std::sync::mpsc as std_mpsc;
5407
5408 use hyperactor_macros::export;
5409 use hyperactor_telemetry::EntityEvent;
5410 use hyperactor_telemetry::TraceEvent;
5411 use hyperactor_telemetry::TraceEventSink;
5412 use serde_json::json;
5413 use timed_test::async_timed_test;
5414 use tokio::sync::Barrier;
5415 use tokio::sync::oneshot;
5416 use tracing::Level;
5417 use tracing_subscriber::layer::SubscriberExt;
5418 use tracing_test::internal::logs_with_scope_contain;
5419
5420 use super::*;
5421 use crate as hyperactor;
5423 use crate::HandleClient;
5424 use crate::Handler;
5425 use crate::OncePortRef;
5426 use crate::PortRef;
5427 use crate::channel::ChannelTransport;
5428 use crate::mailbox::DeliveryFailureKind;
5429 use crate::mailbox::InvalidReferenceReason;
5430 use crate::mailbox::MailboxClient;
5431 use crate::mailbox::PanickingMailboxSender;
5432 use crate::port::Port;
5433 use crate::testing::ids::test_actor_id;
5434 use crate::testing::proc_supervison::ProcSupervisionCoordinator;
5435 use crate::testing::process_assertion::assert_termination;
5436
5437 #[derive(Debug, Default)]
5438 #[export]
5439 struct TestActor;
5440
5441 impl Actor for TestActor {}
5442
5443 hyperactor::behavior!(UnitBehavior, ());
5444
5445 #[derive(Debug)]
5446 struct UnitActor;
5447
5448 impl Actor for UnitActor {}
5449
5450 #[async_trait]
5451 impl Handler<()> for UnitActor {
5452 async fn handle(&mut self, _cx: &Context<Self>, _message: ()) -> anyhow::Result<()> {
5453 Ok(())
5454 }
5455 }
5456
5457 #[derive(Debug)]
5458 struct LocalOnlyActor {
5459 observed: Option<oneshot::Sender<bool>>,
5460 }
5461
5462 #[async_trait]
5463 impl Actor for LocalOnlyActor {
5464 async fn init(&mut self, this: &Instance<Self>) -> anyhow::Result<()> {
5465 let (dest, _dest_rx) = this.open_port::<()>();
5466 (&dest).post(this, ());
5467 self.observed
5468 .take()
5469 .expect("local binding observation should only be sent once")
5470 .send(this.inner.mailbox.bound_return_handle().is_some())
5471 .expect("local binding observation receiver should remain open");
5472 Ok(())
5473 }
5474 }
5475
5476 #[derive(Debug)]
5477 struct InitSendActor {
5478 dest: PortRef<()>,
5479 }
5480
5481 #[async_trait]
5482 impl Actor for InitSendActor {
5483 async fn init(&mut self, this: &Instance<Self>) -> anyhow::Result<()> {
5484 self.dest.post(this, ());
5485 Ok(())
5486 }
5487 }
5488
5489 #[derive(Debug)]
5490 struct InitUndeliverableActor {
5491 dest: PortRef<()>,
5492 observed: Option<OncePortRef<()>>,
5493 }
5494
5495 #[async_trait]
5496 impl Actor for InitUndeliverableActor {
5497 async fn init(&mut self, this: &Instance<Self>) -> anyhow::Result<()> {
5498 self.dest.post(this, ());
5499 Ok(())
5500 }
5501
5502 async fn handle_delivery_failure_event(
5503 &mut self,
5504 this: &Instance<Self>,
5505 _undeliverable: Undeliverable<MessageEnvelope>,
5506 ) -> anyhow::Result<()> {
5507 self.observed
5508 .take()
5509 .expect("delivery failure should only be reported once")
5510 .post(this, ());
5511 Ok(())
5512 }
5513 }
5514
5515 #[derive(Debug)]
5516 struct InitLocalUndeliverableActor {
5517 dest: PortHandle<()>,
5518 observed: Option<OncePortRef<()>>,
5519 }
5520
5521 #[async_trait]
5522 impl Actor for InitLocalUndeliverableActor {
5523 async fn init(&mut self, this: &Instance<Self>) -> anyhow::Result<()> {
5524 (&self.dest).post(this, ());
5525 Ok(())
5526 }
5527
5528 async fn handle_delivery_failure_event(
5529 &mut self,
5530 this: &Instance<Self>,
5531 _undeliverable: Undeliverable<MessageEnvelope>,
5532 ) -> anyhow::Result<()> {
5533 self.observed
5534 .take()
5535 .expect("delivery failure should only be reported once")
5536 .post(this, ());
5537 Ok(())
5538 }
5539 }
5540
5541 struct ActorStatusSink {
5542 actor_id: u64,
5543 sender: std_mpsc::Sender<ActorStatusEvent>,
5544 }
5545
5546 impl TraceEventSink for ActorStatusSink {
5547 fn consume(&mut self, event: &TraceEvent) -> Result<(), anyhow::Error> {
5548 if let TraceEvent::Entity(EntityEvent::ActorStatus(status)) = event
5549 && status.actor_id == self.actor_id
5550 {
5551 let _ = self.sender.send(status.clone());
5552 }
5553 Ok(())
5554 }
5555
5556 fn flush(&mut self) -> Result<(), anyhow::Error> {
5557 Ok(())
5558 }
5559 }
5560
5561 async fn get_status(client: &Client, actor_addr: &ActorAddr) -> Option<ActorStatus> {
5562 let (reply_port, reply_rx) = client.open_once_port::<Option<ActorStatus>>();
5563 actor_addr.status_port().post(
5564 client,
5565 StatusMessage::GetStatus {
5566 reply: reply_port.bind(),
5567 delivery: None,
5568 },
5569 );
5570 tokio::time::timeout(Duration::from_secs(5), reply_rx.recv())
5571 .await
5572 .expect("status reply should arrive")
5573 .expect("status reply port should remain open")
5574 }
5575
5576 #[tokio::test]
5577 async fn terminal_status_subscription_notifies_after_actor_stops() {
5578 let proc = Proc::isolated();
5579 let client = proc.client("client");
5580 let actor = proc.spawn_with_label("target", TestActor);
5581 let (subscriber, mut status_rx) = client.open_port::<Option<ActorStatus>>();
5582 actor.actor_addr().status_port().post(
5583 &client,
5584 StatusMessage::SubscribeTerminal {
5585 subscription_id: Uuid::new_v4(),
5586 subscriber: subscriber.bind(),
5587 },
5588 );
5589
5590 assert!(
5591 tokio::time::timeout(Duration::from_millis(20), status_rx.recv())
5592 .await
5593 .is_err(),
5594 "terminal status subscription should remain quiet while the actor is alive"
5595 );
5596
5597 actor.drain_and_stop("test").unwrap();
5598 let status = tokio::time::timeout(Duration::from_secs(1), status_rx.recv())
5599 .await
5600 .expect("terminal status notification should arrive")
5601 .expect("terminal status subscriber should remain open");
5602 assert_matches!(status, Some(ActorStatus::Stopped(reason)) if reason == "test");
5603 }
5604
5605 #[tokio::test]
5606 async fn terminal_status_subscription_can_be_removed() {
5607 let proc = Proc::isolated();
5608 let client = proc.client("client");
5609 let actor = proc.spawn_with_label("target", TestActor);
5610 let (subscriber, mut status_rx) = client.open_port::<Option<ActorStatus>>();
5611 let subscription_id = Uuid::new_v4();
5612 actor.actor_addr().status_port().post(
5613 &client,
5614 StatusMessage::SubscribeTerminal {
5615 subscription_id,
5616 subscriber: subscriber.bind(),
5617 },
5618 );
5619 actor.actor_addr().status_port().post(
5620 &client,
5621 StatusMessage::UnsubscribeTerminal { subscription_id },
5622 );
5623
5624 actor.drain_and_stop("test").unwrap();
5625 actor.await;
5626 assert!(
5627 tokio::time::timeout(Duration::from_millis(20), status_rx.recv())
5628 .await
5629 .is_err(),
5630 "removed terminal status subscriber should not be notified"
5631 );
5632 }
5633
5634 #[tokio::test]
5635 async fn terminal_status_subscription_drops_stale_subscriber_after_notification() {
5636 let proc = Proc::isolated();
5637 let client = proc.client("client");
5638 let actor = proc.spawn_with_label("target", TestActor);
5639 let actor_id = actor.actor_addr().id().clone();
5640 let (subscriber, status_rx) = client.open_port::<Option<ActorStatus>>();
5641 actor.actor_addr().status_port().post(
5642 &client,
5643 StatusMessage::SubscribeTerminal {
5644 subscription_id: Uuid::new_v4(),
5645 subscriber: subscriber.bind(),
5646 },
5647 );
5648 assert_eq!(
5649 proc.terminal_status_subscriber_count(&actor_id),
5650 1,
5651 "live actor should retain its terminal status subscriber"
5652 );
5653
5654 drop(status_rx);
5655 actor.drain_and_stop("test").unwrap();
5656 assert_matches!(actor.await, ActorStatus::Stopped(reason) if reason == "test");
5657 assert_eq!(
5658 proc.terminal_status_subscriber_count(&actor_id),
5659 0,
5660 "terminal transition should clear a stale subscriber before sending"
5661 );
5662 }
5663
5664 #[tokio::test]
5665 async fn terminal_status_subscription_notifies_for_terminal_and_missing_actors() {
5666 let config = hyperactor_config::global::lock();
5667 let _config_guard = config.override_key(
5668 crate::config::PENDING_ACTOR_DELIVERY_TIMEOUT,
5669 Duration::from_millis(100),
5670 );
5671 let proc = Proc::isolated();
5672 let client = proc.client("client");
5673 let actor = proc.spawn_with_label("target", TestActor);
5674 let actor_addr = actor.actor_addr().clone();
5675 actor.drain_and_stop("test").unwrap();
5676 assert_matches!(actor.await, ActorStatus::Stopped(reason) if reason == "test");
5677
5678 for (actor_addr, expected) in [
5679 (actor_addr, Some(ActorStatus::Stopped("test".to_string()))),
5680 (proc.proc_addr().actor_addr("missing"), None),
5681 ] {
5682 let (subscriber, mut status_rx) = client.open_port::<Option<ActorStatus>>();
5683 actor_addr.status_port().post(
5684 &client,
5685 StatusMessage::SubscribeTerminal {
5686 subscription_id: Uuid::new_v4(),
5687 subscriber: subscriber.bind(),
5688 },
5689 );
5690 let status = tokio::time::timeout(Duration::from_secs(1), status_rx.recv())
5691 .await
5692 .expect("terminal status notification should arrive")
5693 .expect("terminal status subscriber should remain open");
5694 assert_eq!(status, expected);
5695 }
5696 }
5697
5698 #[test]
5699 fn classify_status_change_table() {
5700 let cases = [
5701 (
5702 ActorStatus::Idle,
5703 ActorStatus::Processing(SystemTime::UNIX_EPOCH, None),
5704 StatusChange::Apply,
5705 "idle to processing",
5706 ),
5707 (
5708 ActorStatus::Processing(SystemTime::UNIX_EPOCH, None),
5709 ActorStatus::Idle,
5710 StatusChange::Apply,
5711 "processing to idle",
5712 ),
5713 (
5714 ActorStatus::Idle,
5715 ActorStatus::Stopped("done".to_string()),
5716 StatusChange::Apply,
5717 "non-terminal to terminal",
5718 ),
5719 (
5720 ActorStatus::stopping(),
5721 ActorStatus::zombie("hard kill did not finish"),
5722 StatusChange::Apply,
5723 "non-terminal to zombie",
5724 ),
5725 (
5726 ActorStatus::zombie("hard kill did not finish"),
5727 ActorStatus::Stopped("done".to_string()),
5728 StatusChange::Apply,
5729 "zombie to terminal",
5730 ),
5731 (
5732 ActorStatus::zombie("hard kill did not finish"),
5733 ActorStatus::Idle,
5734 StatusChange::Ignore,
5735 "zombie to idle",
5736 ),
5737 (
5738 ActorStatus::zombie("hard kill did not finish"),
5739 ActorStatus::stopping(),
5740 StatusChange::Ignore,
5741 "zombie to stopping",
5742 ),
5743 (
5744 ActorStatus::Stopped("done".to_string()),
5745 ActorStatus::zombie("hard kill did not finish"),
5746 StatusChange::Ignore,
5747 "terminal to zombie",
5748 ),
5749 (
5750 ActorStatus::Stopped("done".to_string()),
5751 ActorStatus::Idle,
5752 StatusChange::Illegal(IllegalStatusChange::TerminalRewrite),
5753 "terminal to idle",
5754 ),
5755 (
5756 ActorStatus::Stopped("done".to_string()),
5757 ActorStatus::Stopped("other".to_string()),
5758 StatusChange::Illegal(IllegalStatusChange::TerminalRewrite),
5759 "terminal to terminal",
5760 ),
5761 (
5762 ActorStatus::Idle,
5763 ActorStatus::Idle,
5764 StatusChange::Ignore,
5765 "same status",
5766 ),
5767 ];
5768
5769 for (old, new, expected, label) in cases {
5770 assert_eq!(
5771 classify_status_change(&old, &new),
5772 expected,
5773 "{label}: {old} -> {new}"
5774 );
5775 }
5776 }
5777
5778 #[test]
5779 fn instance_creation_emits_created_status() {
5780 hyperactor_telemetry::initialize_logging_for_test();
5781 let proc = Proc::isolated();
5782 let actor_addr = proc.allocate_root_type::<TestActor>();
5783 let actor_id = hash_to_u64(actor_addr.id());
5784 let (sender, receiver) = std_mpsc::channel();
5785 hyperactor_telemetry::register_sink(Box::new(ActorStatusSink { actor_id, sender }));
5786
5787 let (instance, _receivers) =
5788 Instance::<TestActor>::new(proc, actor_addr, false, None, ActorEnvironment::default());
5789
5790 assert_eq!(*instance.status().borrow(), ActorStatus::Created);
5791 let event = receiver
5792 .recv_timeout(Duration::from_secs(5))
5793 .expect("created status should be emitted during instance construction");
5794 assert_eq!(event.actor_id, actor_id);
5795 assert_eq!(event.timestamp, instance.inner.cell.created_at());
5796 assert_eq!(event.new_status, "Created");
5797 assert_eq!(event.reason, None);
5798 }
5799
5800 #[derive(Debug)]
5801 struct ChildLabelActor;
5802
5803 impl Actor for ChildLabelActor {}
5804
5805 #[async_timed_test(timeout_secs = 30)]
5806 async fn test_status_control_port_reports_live_actor() {
5807 let proc = Proc::isolated();
5808 let client = proc.client("client");
5809 let handle = proc.spawn(TestActor);
5810
5811 let mut status_rx = handle.status();
5812 status_rx
5813 .wait_for(|status| matches!(status, ActorStatus::Idle))
5814 .await
5815 .expect("actor should become idle");
5816
5817 let status = get_status(&client, handle.actor_addr()).await;
5818 assert_eq!(status, Some(ActorStatus::Idle));
5819
5820 handle.drain_and_stop("test").unwrap();
5821 handle.await;
5822 }
5823
5824 #[async_timed_test(timeout_secs = 30)]
5825 async fn test_local_only_actor_leaves_return_handler_unbound() {
5826 let proc = Proc::isolated();
5827 let (observed, observed_rx) = oneshot::channel();
5828 let handle = proc.spawn(LocalOnlyActor {
5829 observed: Some(observed),
5830 });
5831
5832 assert!(
5833 !observed_rx
5834 .await
5835 .expect("actor should report its return-handler state"),
5836 "successful local unbound sends should not bind the return handler"
5837 );
5838 assert!(
5839 handle.cell().inner.mailbox.bound_return_handle().is_none(),
5840 "local-only actor should remain unbound"
5841 );
5842
5843 handle.drain_and_stop("test").unwrap();
5844 handle.await;
5845 }
5846
5847 #[async_timed_test(timeout_secs = 30)]
5848 async fn test_empty_behavior_binding_binds_return_handler() {
5849 let proc = Proc::isolated();
5850 let handle = proc.spawn(TestActor);
5851
5852 let _actor_ref: ActorRef<TestActor> = handle.bind();
5853 assert!(
5854 handle.cell().inner.mailbox.bound_return_handle().is_some(),
5855 "producing an actor ref should bind the return handler"
5856 );
5857
5858 handle.drain_and_stop("test").unwrap();
5859 handle.await;
5860 }
5861
5862 #[expect(
5863 clippy::await_holding_invalid_type,
5864 reason = "tracing_test::traced_test macro expansion holds tracing::span::Entered across awaits; can't be fixed in our code"
5865 )]
5866 #[tracing_test::traced_test]
5867 #[async_timed_test(timeout_secs = 30)]
5868 async fn test_remote_send_lazily_binds_return_handler() {
5869 let proc = Proc::isolated();
5870 let client = proc.client("client");
5871 let (dest, mut dest_rx) = client.open_port::<()>();
5872 let handle = proc.spawn(InitSendActor { dest: dest.bind() });
5873
5874 dest_rx
5875 .recv()
5876 .await
5877 .expect("destination should receive the initialization message");
5878 assert!(
5879 handle.cell().inner.mailbox.bound_return_handle().is_some(),
5880 "an outbound remote send should bind the actor's return handler"
5881 );
5882 assert!(
5883 !logs_with_scope_contain(
5884 "hyperactor::context",
5885 "mailbox attempted to post a message without binding Undeliverable<MessageEnvelope>"
5886 ),
5887 "an outbound remote send should not reach the stack-capturing fallback"
5888 );
5889
5890 handle.drain_and_stop("test").unwrap();
5891 handle.await;
5892 }
5893
5894 #[async_timed_test(timeout_secs = 30)]
5895 async fn test_remote_delivery_failure_lazily_binds_return_handler() {
5896 let proc = Proc::isolated();
5897 let client = proc.client("client");
5898 let (dest, dest_rx) = client.open_port::<()>();
5899 let (observed, observed_rx) = client.open_once_port::<()>();
5900 drop(dest_rx);
5901
5902 let handle = proc.spawn(InitUndeliverableActor {
5903 dest: dest.bind(),
5904 observed: Some(observed.bind()),
5905 });
5906
5907 observed_rx
5908 .recv()
5909 .await
5910 .expect("actor should receive the delivery failure from its init post");
5911 assert!(
5912 handle.cell().inner.mailbox.bound_return_handle().is_some(),
5913 "a remote delivery failure should use the actor's return handler"
5914 );
5915
5916 handle.drain_and_stop("test").unwrap();
5917 handle.await;
5918 }
5919
5920 #[async_timed_test(timeout_secs = 30)]
5921 async fn test_local_delivery_failure_lazily_binds_return_handler() {
5922 let proc = Proc::isolated();
5923 let client = proc.client("client");
5924 let (dest, dest_rx) = client.open_port::<()>();
5925 let (observed, observed_rx) = client.open_once_port::<()>();
5926 drop(dest_rx);
5927
5928 let handle = proc.spawn(InitLocalUndeliverableActor {
5929 dest,
5930 observed: Some(observed.bind()),
5931 });
5932
5933 observed_rx
5934 .recv()
5935 .await
5936 .expect("actor should receive its local delivery failure");
5937 assert!(
5938 handle.cell().inner.mailbox.bound_return_handle().is_some(),
5939 "a local delivery failure should use the actor's return handler"
5940 );
5941
5942 handle.drain_and_stop("test").unwrap();
5943 handle.await;
5944 }
5945
5946 #[async_timed_test(timeout_secs = 30)]
5947 async fn test_behavior_binding_binds_return_handler() {
5948 let proc = Proc::isolated();
5949 let handle = proc.spawn(UnitActor);
5950
5951 assert!(
5952 handle.cell().inner.mailbox.bound_return_handle().is_none(),
5953 "the return handler should remain lazy before behavior binding"
5954 );
5955 let _actor_ref: ActorRef<UnitBehavior> = handle.bind();
5956 assert!(
5957 handle.cell().inner.mailbox.bound_return_handle().is_some(),
5958 "producing an actor ref should bind the return handler"
5959 );
5960
5961 handle.drain_and_stop("test").unwrap();
5962 handle.await;
5963 }
5964
5965 #[async_timed_test(timeout_secs = 30)]
5966 async fn test_status_control_port_reports_tombstoned_actor() {
5967 let proc = Proc::isolated();
5968 let client = proc.client("client");
5969 let handle = proc.spawn(TestActor);
5970 let actor_addr = handle.actor_addr().clone();
5971
5972 handle.drain_and_stop("test").unwrap();
5973 let terminal_status = handle.await;
5974
5975 assert_eq!(
5976 get_status(&client, &actor_addr).await,
5977 Some(terminal_status.clone())
5978 );
5979
5980 while !proc
5981 .state()
5982 .actors
5983 .get(actor_addr.id())
5984 .is_some_and(|entry| entry.is_tombstoned())
5985 {
5986 tokio::time::sleep(Duration::from_millis(10)).await;
5987 }
5988 assert_eq!(
5989 get_status(&client, &actor_addr).await,
5990 Some(terminal_status),
5991 "a dropped actor should report its terminal status"
5992 );
5993 }
5994
5995 #[async_timed_test(timeout_secs = 30)]
5996 async fn test_status_control_port_reports_unknown_actor() {
5997 let config = hyperactor_config::global::lock();
5998 let _config_guard = config.override_key(
5999 crate::config::PENDING_ACTOR_DELIVERY_TIMEOUT,
6000 Duration::from_millis(100),
6001 );
6002 let proc = Proc::isolated();
6003 let client = proc.client("client");
6004 let missing = proc.root_addr(Uid::instance(Label::strip("missing")));
6005
6006 assert_eq!(get_status(&client, &missing).await, None);
6007 }
6008
6009 #[async_timed_test(timeout_secs = 30)]
6010 async fn test_status_control_port_waits_for_actor_to_start() {
6011 let config = hyperactor_config::global::lock();
6012 let _config_guard = config.override_key(
6013 crate::config::PENDING_ACTOR_DELIVERY_TIMEOUT,
6014 Duration::from_secs(10),
6015 );
6016 let proc = Proc::isolated();
6017 let client = proc.client("client");
6018 let uid = Uid::instance(Label::strip("pending"));
6019 let pending = proc.root_addr(uid.clone());
6020
6021 let (reply_port, reply_rx) = client.open_once_port::<Option<ActorStatus>>();
6022 pending.status_port().post(
6023 &client,
6024 StatusMessage::GetStatus {
6025 reply: reply_port.bind(),
6026 delivery: None,
6027 },
6028 );
6029 let _actor = proc.spawn_with_uid(uid, TestActor).unwrap();
6030
6031 let status = reply_rx.recv().await.unwrap();
6032 assert_matches!(status, Some(status) if !status.is_terminal());
6033 }
6034
6035 #[async_timed_test(timeout_secs = 30)]
6036 async fn test_status_control_port_times_out_each_message() {
6037 let timeout = Duration::from_millis(500);
6038 let config = hyperactor_config::global::lock();
6039 let _config_guard =
6040 config.override_key(crate::config::PENDING_ACTOR_DELIVERY_TIMEOUT, timeout);
6041 let proc = Proc::isolated();
6042 let client = proc.client("client");
6043 let missing = proc.root_addr(Uid::instance(Label::strip("missing")));
6044 let post_get_status = || {
6045 let (reply_port, reply_rx) = client.open_once_port::<Option<ActorStatus>>();
6046 missing.status_port().post(
6047 &client,
6048 StatusMessage::GetStatus {
6049 reply: reply_port.bind(),
6050 delivery: None,
6051 },
6052 );
6053 reply_rx
6054 };
6055
6056 let first_rx = post_get_status();
6057 tokio::time::sleep(timeout / 2).await;
6058 let second_posted = Instant::now();
6059 let second_rx = post_get_status();
6060
6061 assert_eq!(first_rx.recv().await.unwrap(), None);
6062 assert_eq!(second_rx.recv().await.unwrap(), None);
6063 assert!(
6064 second_posted.elapsed() >= timeout,
6065 "a message should wait out its own timeout, not the first pending message's"
6066 );
6067 }
6068
6069 #[async_timed_test(timeout_secs = 30)]
6070 async fn test_terminal_unsubscribe_before_subscribe_for_pending_actor() {
6071 let config = hyperactor_config::global::lock();
6072 let _config_guard = config.override_key(
6073 crate::config::PENDING_ACTOR_DELIVERY_TIMEOUT,
6074 Duration::from_secs(10),
6075 );
6076 let proc = Proc::isolated();
6077 let client = proc.client("client");
6078 let uid = Uid::instance(Label::strip("pending"));
6079 let pending = proc.root_addr(uid.clone());
6080 let (kept, _kept_rx) = client.open_port::<Option<ActorStatus>>();
6081 let (cancelled, _cancelled_rx) = client.open_port::<Option<ActorStatus>>();
6082 let cancelled_id = Uuid::new_v4();
6083
6084 for message in [
6085 StatusMessage::UnsubscribeTerminal {
6086 subscription_id: cancelled_id,
6087 },
6088 StatusMessage::SubscribeTerminal {
6089 subscription_id: Uuid::new_v4(),
6090 subscriber: kept.bind(),
6091 },
6092 StatusMessage::SubscribeTerminal {
6093 subscription_id: cancelled_id,
6094 subscriber: cancelled.bind(),
6095 },
6096 ] {
6097 pending.status_port().post(&client, message);
6098 }
6099 let actor = proc.spawn_with_uid(uid, TestActor).unwrap();
6100
6101 assert_eq!(
6102 proc.terminal_status_subscriber_count(actor.actor_addr().id()),
6103 1,
6104 "a subscription cancelled before it arrived should not be activated"
6105 );
6106 }
6107
6108 #[tokio::test]
6109 async fn test_terminal_unsubscribe_before_subscribe_for_live_actor() {
6110 let proc = Proc::isolated();
6111 let client = proc.client("client");
6112 let actor = proc.spawn_with_label("target", TestActor);
6113 let actor_id = actor.actor_addr().id().clone();
6114 let (subscriber, mut status_rx) = client.open_port::<Option<ActorStatus>>();
6115 let subscription_id = Uuid::new_v4();
6116
6117 for message in [
6118 StatusMessage::UnsubscribeTerminal { subscription_id },
6119 StatusMessage::SubscribeTerminal {
6120 subscription_id,
6121 subscriber: subscriber.bind(),
6122 },
6123 ] {
6124 actor.actor_addr().status_port().post(&client, message);
6125 }
6126 assert_eq!(
6127 proc.terminal_status_subscriber_count(&actor_id),
6128 0,
6129 "a subscription cancelled before it arrived should not be activated"
6130 );
6131
6132 actor.drain_and_stop("test").unwrap();
6133 actor.await;
6134 assert!(
6135 tokio::time::timeout(Duration::from_millis(20), status_rx.recv())
6136 .await
6137 .is_err(),
6138 "cancelled terminal status subscriber should not be notified"
6139 );
6140 }
6141
6142 #[derive(Debug)]
6143 struct DelayedSelfActor {
6144 ready: Option<OncePortRef<()>>,
6145 fired: Option<OncePortRef<()>>,
6146 delay: Duration,
6147 }
6148
6149 #[derive(Debug)]
6150 struct DelayedSelfTick;
6151
6152 #[async_trait]
6153 impl Actor for DelayedSelfActor {
6154 async fn init(&mut self, this: &Instance<Self>) -> anyhow::Result<()> {
6155 if let Some(ready) = self.ready.take() {
6156 ready.post(this, ());
6157 }
6158 this.post_after(this, DelayedSelfTick, self.delay);
6159 Ok(())
6160 }
6161 }
6162
6163 #[async_trait]
6164 impl Handler<DelayedSelfTick> for DelayedSelfActor {
6165 async fn handle(
6166 &mut self,
6167 cx: &crate::Context<Self>,
6168 _message: DelayedSelfTick,
6169 ) -> anyhow::Result<()> {
6170 if let Some(fired) = self.fired.take() {
6171 fired.post(cx, ());
6172 }
6173 Ok(())
6174 }
6175 }
6176
6177 #[derive(Debug)]
6178 struct DelayedPortActor {
6179 reply: Option<PortRef<u64>>,
6180 delay: Duration,
6181 }
6182
6183 #[async_trait]
6184 impl Actor for DelayedPortActor {
6185 async fn init(&mut self, this: &Instance<Self>) -> anyhow::Result<()> {
6186 this.post_after(
6187 self.reply.take().expect("reply port should be present"),
6188 123u64,
6189 self.delay,
6190 );
6191 Ok(())
6192 }
6193 }
6194
6195 #[derive(Handler, HandleClient, Debug)]
6196 enum TestActorMessage {
6197 Reply(oneshot::Sender<()>),
6198 Wait(oneshot::Sender<()>, oneshot::Receiver<()>),
6199 Forward(ActorHandle<TestActor>, Box<TestActorMessage>),
6200 Noop(),
6201 Fail(anyhow::Error),
6202 Panic(String),
6203 Spawn(oneshot::Sender<ActorHandle<TestActor>>),
6204 }
6205
6206 impl TestActor {
6207 async fn spawn_child(
6208 cx: &impl context::Actor,
6209 parent: &ActorHandle<TestActor>,
6210 ) -> ActorHandle<TestActor> {
6211 let (tx, rx) = oneshot::channel();
6212 parent.post(cx, TestActorMessage::Spawn(tx));
6213 rx.await.unwrap()
6214 }
6215 }
6216
6217 #[test]
6218 fn test_proc_identity_constructors() {
6219 let anonymous = Proc::anonymous();
6220 assert!(
6221 matches!(anonymous.proc_id().uid(), crate::id::Uid::Instance(_, None)),
6222 "anonymous proc must have an unlabeled instance id"
6223 );
6224 assert_eq!(anonymous.proc_id().label(), None);
6225
6226 let instance = Proc::instance("worker");
6227 assert!(
6228 matches!(
6229 instance.proc_id().uid(),
6230 crate::id::Uid::Instance(_, Some(label)) if label.as_str() == "worker"
6231 ),
6232 "instance proc must have a labeled instance id"
6233 );
6234 assert_eq!(
6235 instance.proc_id().label().map(|label| label.as_str()),
6236 Some("worker")
6237 );
6238
6239 let singleton = Proc::singleton("controller");
6240 assert!(
6241 matches!(
6242 singleton.proc_id().uid(),
6243 crate::id::Uid::Singleton(label) if label.as_str() == "controller"
6244 ),
6245 "singleton proc must have a singleton id"
6246 );
6247 assert_eq!(
6248 singleton.proc_id().label().map(|label| label.as_str()),
6249 Some("controller")
6250 );
6251 }
6252
6253 #[test]
6254 fn test_default_actor_label_uses_label_compatible_type_basename() {
6255 assert_eq!(default_actor_label::<TestActor>().as_str(), "testactor");
6256 assert_eq!(
6257 default_actor_label::<std::collections::HashMap<String, u64>>().as_str(),
6258 "hashmap"
6259 );
6260 assert_eq!(default_actor_label::<()>().as_str(), "nil");
6261 }
6262
6263 #[test]
6264 fn test_global_proc_label_uses_short_hostname_and_pid() {
6265 assert_eq!(
6266 global_proc_label_from("devvm34959.nha0.facebook.com", 123555).as_str(),
6267 "devvm34959-123555"
6268 );
6269 assert_eq!(
6270 global_proc_label_from("DevVM34959.nha0.facebook.com", 7).as_str(),
6271 "devvm34959-7"
6272 );
6273 }
6274
6275 #[async_timed_test(timeout_secs = 30)]
6276 async fn test_spawn_uses_actor_type_label_for_root_actor() {
6277 let proc = Proc::isolated();
6278 let handle = proc.spawn(TestActor);
6279
6280 assert_eq!(
6281 handle.actor_addr().label().map(Label::as_str),
6282 Some("testactor")
6283 );
6284 assert!(matches!(
6285 handle.actor_addr().uid(),
6286 Uid::Instance(_, Some(label)) if label.as_str() == "testactor"
6287 ));
6288
6289 handle.drain_and_stop("test").unwrap();
6290 handle.await;
6291 }
6292
6293 #[async_timed_test(timeout_secs = 30)]
6294 async fn test_spawn_uses_actor_type_label_for_child_actor() {
6295 let proc = Proc::isolated();
6296 let parent = proc.spawn(TestActor);
6297 let child = proc.spawn_child(parent.cell().clone(), ChildLabelActor);
6298
6299 assert!(!child.actor_addr().is_root());
6300 assert_eq!(
6301 child.actor_addr().label().map(Label::as_str),
6302 Some("childlabelactor")
6303 );
6304 assert!(matches!(
6305 child.actor_addr().uid(),
6306 Uid::Instance(_, Some(label)) if label.as_str() == "childlabelactor"
6307 ));
6308
6309 child.drain_and_stop("test").unwrap();
6310 parent.drain_and_stop("test").unwrap();
6311 child.await;
6312 parent.await;
6313 }
6314
6315 #[async_timed_test(timeout_secs = 30)]
6316 async fn test_root_tracking_does_not_depend_on_singleton_uids() {
6317 let proc = Proc::isolated();
6318 let parent = proc.spawn(TestActor);
6319 let child = proc.spawn_child(parent.cell().clone(), TestActor);
6320
6321 assert!(parent.actor_addr().uid().is_instance());
6322 assert!(
6323 proc.state()
6324 .root_instances
6325 .contains_key(parent.actor_addr().id())
6326 );
6327 assert!(
6328 !proc
6329 .state()
6330 .root_instances
6331 .contains_key(child.actor_addr().id())
6332 );
6333
6334 let mut traversed = Vec::new();
6335 proc.traverse(&mut |cell, _depth| {
6336 traversed.push(cell.actor_addr().id().clone());
6337 });
6338 assert!(traversed.contains(parent.actor_addr().id()));
6339 assert!(traversed.contains(child.actor_addr().id()));
6340
6341 child.drain_and_stop("test").unwrap();
6342 parent.drain_and_stop("test").unwrap();
6343 child.await;
6344 parent.await;
6345 }
6346
6347 #[derive(Debug)]
6350 #[hyperactor::export(handlers = [u64])]
6351 struct Recorder {
6352 seen: mpsc::UnboundedSender<u64>,
6353 }
6354
6355 #[async_trait]
6356 impl Actor for Recorder {
6357 async fn init(&mut self, this: &Instance<Self>) -> anyhow::Result<()> {
6358 this.bind::<Self>();
6359 Ok(())
6360 }
6361 }
6362
6363 #[async_trait]
6364 impl Handler<u64> for Recorder {
6365 async fn handle(&mut self, _cx: &Context<Self>, value: u64) -> anyhow::Result<()> {
6366 self.seen
6367 .send(value)
6368 .expect("test receiver should remain open");
6369 Ok(())
6370 }
6371 }
6372
6373 fn post_to_unspawned_recorder(
6377 proc: &Proc,
6378 value: u64,
6379 ) -> (
6380 PortHandle<Undeliverable<MessageEnvelope>>,
6381 PortReceiver<Undeliverable<MessageEnvelope>>,
6382 ) {
6383 let dest = proc
6384 .proc_addr()
6385 .actor_addr("recorder")
6386 .port_addr(Port::handler::<u64>());
6387 let mut envelope =
6388 MessageEnvelope::serialize(test_actor_id("0", "sender"), dest, &value, Flattrs::new())
6389 .expect("serialize");
6390 envelope.set_header(SEQ_INFO, SeqInfo::Direct);
6391 let (return_handle, return_rx) = Mailbox::new(test_actor_id("0", "returns"))
6392 .open_port::<Undeliverable<MessageEnvelope>>();
6393 proc.post(envelope, return_handle.clone());
6394 (return_handle, return_rx)
6395 }
6396
6397 #[async_timed_test(timeout_secs = 30)]
6398 async fn test_spawn_bound_with_uid_delivers_messages_sent_before_spawn() {
6399 let config = hyperactor_config::global::lock();
6400 let _timeout = config.override_key(
6401 crate::config::PENDING_ACTOR_DELIVERY_TIMEOUT,
6402 Duration::from_secs(30),
6403 );
6404 let proc = Proc::isolated();
6405 let (_return_handle, mut return_rx) = post_to_unspawned_recorder(&proc, 7);
6406
6407 let (seen, mut seen_rx) = mpsc::unbounded_channel();
6408 let handle = proc
6409 .spawn_bound_with_uid(
6410 Uid::singleton(Label::new("recorder").unwrap()),
6411 Recorder { seen },
6412 )
6413 .unwrap();
6414
6415 assert_eq!(
6416 seen_rx.recv().await,
6417 Some(7),
6418 "the message buffered before spawn should be handled"
6419 );
6420 assert!(
6421 return_rx.try_recv().unwrap().is_none(),
6422 "no message should be returned"
6423 );
6424
6425 handle.drain_and_stop("test").unwrap();
6426 handle.await;
6427 }
6428
6429 #[async_timed_test(timeout_secs = 30)]
6433 async fn test_spawn_with_uid_returns_messages_sent_before_spawn() {
6434 let config = hyperactor_config::global::lock();
6435 let _timeout = config.override_key(
6436 crate::config::PENDING_ACTOR_DELIVERY_TIMEOUT,
6437 Duration::from_secs(30),
6438 );
6439 let proc = Proc::isolated();
6440 let (_return_handle, mut return_rx) = post_to_unspawned_recorder(&proc, 7);
6441
6442 let (seen, _seen_rx) = mpsc::unbounded_channel();
6443 let handle = proc
6444 .spawn_with_uid(
6445 Uid::singleton(Label::new("recorder").unwrap()),
6446 Recorder { seen },
6447 )
6448 .unwrap();
6449
6450 let undelivered = tokio::time::timeout(Duration::from_secs(10), return_rx.recv())
6451 .await
6452 .expect("timed out waiting for the returned message")
6453 .expect("return port closed")
6454 .into_message()
6455 .expect("expected a returned envelope");
6456 let root_failure = undelivered
6457 .root_delivery_failure()
6458 .expect("expected a root delivery failure");
6459 let DeliveryFailureKind::InvalidReference(invalid_reference) = &root_failure.kind else {
6460 panic!("expected an invalid reference, got {root_failure}");
6461 };
6462 assert_eq!(
6463 invalid_reference.reason,
6464 InvalidReferenceReason::HandlerNotBound
6465 );
6466
6467 handle.drain_and_stop("test").unwrap();
6468 handle.await;
6469 }
6470
6471 #[async_timed_test(timeout_secs = 30)]
6472 async fn test_client_spawn_api_labels_and_explicit_uid() {
6473 let proc = Proc::isolated();
6474 let client = proc.client("client");
6475
6476 let spawned = client.spawn(TestActor);
6477 assert_eq!(
6478 spawned.actor_addr().label().map(Label::as_str),
6479 Some("testactor")
6480 );
6481
6482 let labeled = client.spawn_with_label("custom", TestActor);
6483 assert_eq!(
6484 labeled.actor_addr().label().map(Label::as_str),
6485 Some("custom")
6486 );
6487
6488 let uid = Uid::instance(Label::new("explicit").unwrap());
6489 let explicit = client.spawn_with_uid(uid.clone(), TestActor).unwrap();
6490 assert_eq!(explicit.actor_addr().uid(), &uid);
6491
6492 let child = client.child();
6493 assert!(!child.self_addr().is_root());
6494 assert!(matches!(child.self_addr().uid(), Uid::Instance(_, None)));
6495 assert_eq!(child.self_addr().label(), None);
6496 let child_spawned = child.spawn(TestActor);
6497 assert_eq!(
6498 child_spawned.actor_addr().label().map(Label::as_str),
6499 Some("testactor")
6500 );
6501
6502 spawned.drain_and_stop("test").unwrap();
6503 labeled.drain_and_stop("test").unwrap();
6504 explicit.drain_and_stop("test").unwrap();
6505 child_spawned.drain_and_stop("test").unwrap();
6506 spawned.await;
6507 labeled.await;
6508 explicit.await;
6509 child_spawned.await;
6510 }
6511
6512 #[test]
6513 fn test_current_proc_uses_stable_global_proc_outside_actor_context() {
6514 let first = Proc::current();
6515 let second = Proc::current();
6516
6517 assert_eq!(first.proc_id(), second.proc_id());
6518 assert_eq!(
6519 ProcAddr::new(
6520 first.proc_id().clone(),
6521 Gateway::current().default_location()
6522 ),
6523 first.proc_addr()
6524 );
6525 }
6526
6527 #[async_trait]
6528 #[crate::handle(TestActorMessage)]
6529 impl TestActorMessageHandler for TestActor {
6530 async fn reply(
6531 &mut self,
6532 _cx: &crate::Context<Self>,
6533 sender: oneshot::Sender<()>,
6534 ) -> Result<(), anyhow::Error> {
6535 sender.send(()).unwrap();
6536 Ok(())
6537 }
6538
6539 async fn wait(
6540 &mut self,
6541 _cx: &crate::Context<Self>,
6542 sender: oneshot::Sender<()>,
6543 receiver: oneshot::Receiver<()>,
6544 ) -> Result<(), anyhow::Error> {
6545 sender.send(()).unwrap();
6546 receiver.await.unwrap();
6547 Ok(())
6548 }
6549
6550 async fn forward(
6551 &mut self,
6552 cx: &crate::Context<Self>,
6553 destination: ActorHandle<TestActor>,
6554 message: Box<TestActorMessage>,
6555 ) -> Result<(), anyhow::Error> {
6556 destination.post(cx, *message);
6558 Ok(())
6559 }
6560
6561 async fn noop(&mut self, _cx: &crate::Context<Self>) -> Result<(), anyhow::Error> {
6562 Ok(())
6563 }
6564
6565 async fn fail(
6566 &mut self,
6567 _cx: &crate::Context<Self>,
6568 err: anyhow::Error,
6569 ) -> Result<(), anyhow::Error> {
6570 Err(err)
6571 }
6572
6573 async fn panic(
6574 &mut self,
6575 _cx: &crate::Context<Self>,
6576 err_msg: String,
6577 ) -> Result<(), anyhow::Error> {
6578 panic!("{}", err_msg);
6579 }
6580
6581 async fn spawn(
6582 &mut self,
6583 cx: &crate::Context<Self>,
6584 reply: oneshot::Sender<ActorHandle<TestActor>>,
6585 ) -> Result<(), anyhow::Error> {
6586 let handle = cx.spawn(TestActor);
6587 reply.send(handle).unwrap();
6588 Ok(())
6589 }
6590 }
6591
6592 #[derive(Debug)]
6593 struct CurrentProcActor;
6594
6595 impl Actor for CurrentProcActor {}
6596
6597 #[derive(Handler, Debug)]
6598 enum CurrentProcMessage {
6599 Check(oneshot::Sender<CurrentProcSnapshot>),
6600 }
6601
6602 #[derive(Debug)]
6603 struct CurrentProcSnapshot {
6604 current_proc_id: ProcId,
6605 current_gateway_proc_addr: ProcAddr,
6606 spawned_handle: ActorHandle<TestActor>,
6607 client_proc_id: ProcId,
6608 }
6609
6610 #[async_trait]
6611 #[crate::handle(CurrentProcMessage)]
6612 impl CurrentProcMessageHandler for CurrentProcActor {
6613 async fn check(
6614 &mut self,
6615 _cx: &crate::Context<Self>,
6616 reply: oneshot::Sender<CurrentProcSnapshot>,
6617 ) -> Result<(), anyhow::Error> {
6618 let current = Proc::current();
6619 let spawned_handle = crate::spawn(TestActor);
6620 let client = crate::client("current_client");
6621 reply
6622 .send(CurrentProcSnapshot {
6623 current_proc_id: current.proc_id().clone(),
6624 current_gateway_proc_addr: ProcAddr::new(
6625 current.proc_id().clone(),
6626 Gateway::current().default_location(),
6627 ),
6628 spawned_handle,
6629 client_proc_id: client.self_addr().proc_id().clone(),
6630 })
6631 .unwrap();
6632 Ok(())
6633 }
6634 }
6635
6636 #[tokio::test]
6637 async fn test_current_proc_tracks_actor_context() {
6638 let proc = Proc::isolated();
6639 let client = proc.client("client");
6640 let actor = proc.spawn(CurrentProcActor);
6641 let (tx, rx) = oneshot::channel();
6642
6643 crate::Endpoint::post(&actor, &client, CurrentProcMessage::Check(tx));
6644 let snapshot = rx.await.unwrap();
6645
6646 assert_eq!(&snapshot.current_proc_id, proc.proc_id());
6647 assert_eq!(snapshot.current_gateway_proc_addr, proc.proc_addr());
6648 assert_eq!(
6649 snapshot.spawned_handle.actor_addr().proc_id(),
6650 proc.proc_id()
6651 );
6652 assert_eq!(&snapshot.client_proc_id, proc.proc_id());
6653
6654 snapshot
6655 .spawned_handle
6656 .drain_and_stop("test complete")
6657 .unwrap();
6658 snapshot.spawned_handle.await;
6659 actor.drain_and_stop("test complete").unwrap();
6660 actor.await;
6661 }
6662
6663 #[expect(
6664 clippy::await_holding_invalid_type,
6665 reason = "tracing_test::traced_test macro expansion holds tracing::span::Entered across awaits; can't be fixed in our code"
6666 )]
6667 #[tracing_test::traced_test]
6668 #[async_timed_test(timeout_secs = 30)]
6669 async fn test_spawn_actor() {
6670 let proc = Proc::isolated();
6671 let client = proc.client("client");
6672 let handle = proc.spawn(TestActor);
6673
6674 assert!(logs_contain(
6676 format!(
6677 "{}: spawned with {:?}",
6678 handle.actor_addr(),
6679 handle.cell().actor_task_handle().unwrap(),
6680 )
6681 .as_str()
6682 ));
6683
6684 let mut state = handle.status().clone();
6685
6686 let (tx, rx) = oneshot::channel::<()>();
6689 handle.post(&client, TestActorMessage::Reply(tx));
6690 rx.await.unwrap();
6691
6692 state
6693 .wait_for(|state: &ActorStatus| matches!(*state, ActorStatus::Idle))
6694 .await
6695 .unwrap();
6696
6697 let (enter_tx, enter_rx) = oneshot::channel::<()>();
6699 let (exit_tx, exit_rx) = oneshot::channel::<()>();
6700
6701 handle.post(&client, TestActorMessage::Wait(enter_tx, exit_rx));
6702 enter_rx.await.unwrap();
6703 assert_matches!(*state.borrow(), ActorStatus::Processing(instant, _) if instant <= std::time::SystemTime::now());
6704 exit_tx.send(()).unwrap();
6705
6706 state
6707 .wait_for(|state| matches!(*state, ActorStatus::Idle))
6708 .await
6709 .unwrap();
6710
6711 handle.drain_and_stop("test").unwrap();
6712 handle.await;
6713 assert_matches!(&*state.borrow(), ActorStatus::Stopped(reason) if reason == "test");
6714 }
6715
6716 #[async_timed_test(timeout_secs = 30)]
6717 async fn test_proc_actors_messaging() {
6718 let proc = Proc::isolated();
6719 let client = proc.client("client");
6720 let first = proc.spawn_with_label::<TestActor>("first", TestActor);
6721 let second = proc.spawn_with_label::<TestActor>("second", TestActor);
6722 let (tx, rx) = oneshot::channel::<()>();
6723 let reply_message = TestActorMessage::Reply(tx);
6724 first.post(
6725 &client,
6726 TestActorMessage::Forward(second, Box::new(reply_message)),
6727 );
6728 rx.await.unwrap();
6729 }
6730
6731 #[tokio::test]
6736 async fn test_post_routes_by_proc_id() {
6737 use crate::mailbox::monitored_return_handle;
6738
6739 #[derive(Clone)]
6740 struct CountingSender(Arc<AtomicUsize>);
6741
6742 #[async_trait]
6743 impl MailboxSender for CountingSender {
6744 fn post_unchecked(
6745 &self,
6746 _envelope: MessageEnvelope,
6747 _return_handle: PortHandle<Undeliverable<MessageEnvelope>>,
6748 ) {
6749 self.0.fetch_add(1, Ordering::SeqCst);
6750 }
6751 }
6752
6753 let local_addr = ChannelAddr::Local(1);
6756 let remote_addr = ChannelAddr::Local(2);
6757
6758 let proc_local = ProcAddr::instance(local_addr.clone(), "shared");
6759 let proc_same_id_other_location =
6760 ProcAddr::new(proc_local.id().clone(), remote_addr.into());
6761 let proc_other_id_same_location = ProcAddr::instance(local_addr, "other");
6762 assert_eq!(
6763 proc_local.id(),
6764 proc_same_id_other_location.id(),
6765 "test setup: both procs must share a ProcId"
6766 );
6767 assert_ne!(
6768 proc_local.id(),
6769 proc_other_id_same_location.id(),
6770 "test setup: the remote proc must have a distinct ProcId"
6771 );
6772
6773 let forwarded = Arc::new(AtomicUsize::new(0));
6774 let proc = Proc::configured(
6775 proc_local.clone(),
6776 BoxedMailboxSender::new(CountingSender(forwarded.clone())),
6777 );
6778 let sender = test_actor_id("sender", "client");
6779
6780 let local_dest = proc_local.actor_addr("worker").port_addr(Port::from(1234));
6782 proc.post(
6783 MessageEnvelope::new(
6784 sender.clone(),
6785 local_dest,
6786 wirevalue::Any::serialize(&1u64).unwrap(),
6787 Flattrs::new(),
6788 ),
6789 monitored_return_handle(),
6790 );
6791 assert_eq!(forwarded.load(Ordering::SeqCst), 0);
6792
6793 let same_id_other_location_dest = proc_same_id_other_location
6795 .actor_addr("worker")
6796 .port_addr(Port::from(1234));
6797 proc.post(
6798 MessageEnvelope::new(
6799 sender.clone(),
6800 same_id_other_location_dest,
6801 wirevalue::Any::serialize(&1u64).unwrap(),
6802 Flattrs::new(),
6803 ),
6804 monitored_return_handle(),
6805 );
6806 assert_eq!(forwarded.load(Ordering::SeqCst), 0);
6807
6808 let other_id_same_location_dest = proc_other_id_same_location
6810 .actor_addr("worker")
6811 .port_addr(Port::from(1234));
6812 proc.post(
6813 MessageEnvelope::new(
6814 sender,
6815 other_id_same_location_dest,
6816 wirevalue::Any::serialize(&1u64).unwrap(),
6817 Flattrs::new(),
6818 ),
6819 monitored_return_handle(),
6820 );
6821 assert_eq!(forwarded.load(Ordering::SeqCst), 1);
6822 }
6823
6824 #[tokio::test]
6829 async fn test_mailbox_ext_post_stamps_sender_actor_id() {
6830 use crate::mailbox::headers::SENDER_ACTOR_ID;
6831
6832 #[derive(typeuri::Named)]
6833 struct DestHandlerMsg;
6834
6835 #[derive(Clone, Default)]
6836 struct CapturingSender(Arc<Mutex<Vec<MessageEnvelope>>>);
6837
6838 #[async_trait]
6839 impl MailboxSender for CapturingSender {
6840 fn post_unchecked(
6841 &self,
6842 envelope: MessageEnvelope,
6843 _return_handle: PortHandle<Undeliverable<MessageEnvelope>>,
6844 ) {
6845 self.0.lock().unwrap().push(envelope);
6846 }
6847 }
6848
6849 let proc_addr = ProcAddr::instance(ChannelAddr::Local(1), "stamping_test");
6850 let captured: Arc<Mutex<Vec<MessageEnvelope>>> = Arc::new(Mutex::new(Vec::new()));
6851 let proc = Proc::configured(
6852 proc_addr,
6853 BoxedMailboxSender::new(CapturingSender(captured.clone())),
6854 );
6855
6856 let client = proc.client("client");
6857 let client_addr = client.mailbox().actor_addr().clone();
6858
6859 let remote_dest = ProcAddr::instance(ChannelAddr::Local(2), "remote")
6862 .actor_addr("worker")
6863 .port_addr(Port::handler::<DestHandlerMsg>());
6864
6865 <Client as context::MailboxExt>::post(
6869 &client,
6870 remote_dest,
6871 Flattrs::new(),
6872 wirevalue::Any::serialize(&1u64).unwrap(),
6873 false,
6874 context::SeqInfoPolicy::AssignNew,
6875 );
6876
6877 let captured = captured.lock().unwrap();
6878 assert_eq!(
6879 captured.len(),
6880 1,
6881 "exactly one envelope should be forwarded"
6882 );
6883 assert_eq!(
6884 captured[0].headers().get(SENDER_ACTOR_ID),
6885 Some(client_addr),
6886 "MailboxExt::post must stamp SENDER_ACTOR_ID with the client's actor_addr"
6887 );
6888 }
6889
6890 #[test]
6891 fn test_local_delivery_legacy_service_compares_full_proc_addr() {
6892 let local = ProcAddr::singleton(ChannelAddr::Local(1), LEGACY_SERVICE_PROC_NAME);
6893 let same_id_other_location =
6894 ProcAddr::singleton(ChannelAddr::Local(2), LEGACY_SERVICE_PROC_NAME);
6895 let proc = Proc::legacy_service_pseudo_singleton(
6896 ChannelAddr::Local(1),
6897 BoxedMailboxSender::new(PanickingMailboxSender),
6898 );
6899
6900 assert_eq!(local.id(), same_id_other_location.id());
6901 assert!(proc.is_local_delivery_target(&local));
6902 assert!(!proc.is_local_delivery_target(&same_id_other_location));
6903
6904 let shared = ProcAddr::singleton(ChannelAddr::Local(1), "shared");
6905 let shared_other_location = ProcAddr::singleton(ChannelAddr::Local(2), "shared");
6906 let proc = Proc::configured(
6907 shared.clone(),
6908 BoxedMailboxSender::new(PanickingMailboxSender),
6909 );
6910 assert!(proc.is_local_delivery_target(&shared_other_location));
6911
6912 let service_instance = ProcAddr::instance(ChannelAddr::Local(1), "service");
6913 let service_instance_other_location =
6914 ProcAddr::new(service_instance.id().clone(), ChannelAddr::Local(2).into());
6915 let proc = Proc::configured(
6916 service_instance,
6917 BoxedMailboxSender::new(PanickingMailboxSender),
6918 );
6919 assert!(proc.is_local_delivery_target(&service_instance_other_location));
6920 }
6921
6922 #[test]
6923 fn test_legacy_service_pseudo_singleton_uses_dedicated_constructor() {
6924 let result = std::panic::catch_unwind(|| {
6925 Proc::configured(
6926 ProcAddr::singleton(ChannelAddr::Local(1), LEGACY_SERVICE_PROC_NAME),
6927 BoxedMailboxSender::new(PanickingMailboxSender),
6928 );
6929 });
6930 assert!(result.is_err());
6931
6932 let service = Proc::legacy_service_pseudo_singleton(
6933 ChannelAddr::Local(1),
6934 BoxedMailboxSender::new(PanickingMailboxSender),
6935 );
6936 assert_eq!(
6937 service.proc_addr().id().uid().to_string(),
6938 LEGACY_SERVICE_PROC_NAME
6939 );
6940 }
6941
6942 #[tokio::test]
6943 async fn test_mailbox_muxer_delivers_by_actor_id() {
6944 use crate::mailbox::PortLocation;
6945 use crate::mailbox::monitored_return_handle;
6946
6947 let proc = Proc::isolated();
6948 let instance = proc.client("worker");
6949 let (port, mut receiver) = instance.bind_handler_port::<u64>();
6950
6951 let PortLocation::Bound(default_dest) = port.location() else {
6952 panic!("handler port must be bound");
6953 };
6954 let alternate_dest =
6955 PortAddr::new(default_dest.id().clone(), ChannelAddr::Local(9876).into());
6956
6957 proc.post(
6958 MessageEnvelope::serialize(
6959 test_actor_id("sender", "client"),
6960 alternate_dest,
6961 &123u64,
6962 Flattrs::new(),
6963 )
6964 .unwrap(),
6965 monitored_return_handle(),
6966 );
6967
6968 assert_eq!(receiver.recv().await.unwrap(), 123);
6969 }
6970
6971 #[test]
6972 fn test_default_location_changes_new_bindings_not_lookup() {
6973 let proc = Proc::isolated();
6974 let gateway = proc.gateway();
6975 let client = proc.client("worker");
6976
6977 let first_ref: ActorRef<()> = client.bind();
6978 let new_location = ChannelAddr::Local(9876).into();
6979 gateway.set_default_location(new_location);
6980 let second_ref: ActorRef<()> = client.bind();
6981
6982 assert_eq!(first_ref.actor_addr().id(), second_ref.actor_addr().id());
6983 assert_ne!(
6984 first_ref.actor_addr().location(),
6985 second_ref.actor_addr().location()
6986 );
6987 assert_eq!(second_ref.actor_addr().location(), &proc.default_location());
6988 assert_eq!(proc.default_location(), gateway.default_location());
6989 assert_eq!(
6990 proc.proc_addr(),
6991 ProcAddr::new(proc.proc_id().clone(), gateway.default_location())
6992 );
6993 assert!(proc.get_instance(second_ref.actor_addr()).is_some());
6994 }
6995
6996 #[async_timed_test(timeout_secs = 10)]
7001 async fn test_default_location_concurrent_with_bind() {
7002 let proc = Proc::isolated();
7003 let gateway = proc.gateway();
7004 let handle = proc.client("worker");
7005
7006 let loc_a: Location = ChannelAddr::Local(40001).into();
7007 let loc_b: Location = ChannelAddr::Local(40002).into();
7008
7009 gateway.set_default_location(loc_a.clone());
7014
7015 let barrier = std::sync::Arc::new(Barrier::new(2));
7016
7017 let setter = {
7018 let gateway = gateway.clone();
7019 let loc_a = loc_a.clone();
7020 let loc_b = loc_b.clone();
7021 let barrier = barrier.clone();
7022 tokio::spawn(async move {
7023 barrier.wait().await;
7024 for i in 0..100 {
7025 let loc = if i % 2 == 0 {
7026 loc_a.clone()
7027 } else {
7028 loc_b.clone()
7029 };
7030 gateway.set_default_location(loc);
7031 tokio::task::yield_now().await;
7032 }
7033 })
7034 };
7035
7036 let binder = {
7037 let handle = handle.clone();
7043 let barrier = barrier.clone();
7044 tokio::spawn(async move {
7045 barrier.wait().await;
7046 let mut refs = Vec::with_capacity(100);
7047 for _ in 0..100 {
7048 refs.push(handle.bind::<()>());
7049 tokio::task::yield_now().await;
7050 }
7051 refs
7052 })
7053 };
7054
7055 setter.await.unwrap();
7056 let refs = binder.await.unwrap();
7057
7058 for r in &refs {
7060 let loc = r.actor_addr().location();
7061 assert!(
7062 loc == &loc_a || loc == &loc_b,
7063 "ref location {loc:?} is neither {loc_a:?} nor {loc_b:?}",
7064 );
7065 }
7066
7067 for r in &refs {
7069 assert!(
7070 proc.get_instance(r.actor_addr()).is_some(),
7071 "ref {:?} no longer resolves",
7072 r.actor_addr(),
7073 );
7074 }
7075 }
7076
7077 #[test]
7078 fn test_builder_procs_can_share_gateway_with_distinct_ids() {
7079 let gateway = Gateway::new();
7080 let first = Proc::builder()
7081 .proc_id(ProcId::instance(Label::strip("first")))
7082 .shared_gateway(gateway.clone())
7083 .build()
7084 .unwrap();
7085 let second = Proc::builder()
7086 .proc_id(ProcId::instance(Label::strip("second")))
7087 .shared_gateway(gateway.clone())
7088 .build()
7089 .unwrap();
7090
7091 assert_ne!(first.proc_id(), second.proc_id());
7092 assert_eq!(first.default_location(), second.default_location());
7093
7094 let new_location = ChannelAddr::Local(9876).into();
7095 gateway.set_default_location(new_location);
7096
7097 assert_eq!(first.default_location(), gateway.default_location());
7098 assert_eq!(second.default_location(), gateway.default_location());
7099 assert_eq!(
7100 first.proc_addr(),
7101 ProcAddr::new(first.proc_id().clone(), gateway.default_location())
7102 );
7103 assert_eq!(
7104 second.proc_addr(),
7105 ProcAddr::new(second.proc_id().clone(), gateway.default_location())
7106 );
7107 }
7108
7109 #[test]
7110 fn test_isolated_procs_use_distinct_gateways() {
7111 let first = Proc::isolated();
7112 let second = Proc::isolated();
7113 let second_location = second.default_location();
7114
7115 first
7116 .gateway()
7117 .set_default_location(ChannelAddr::Local(9876).into());
7118
7119 assert_ne!(first.proc_id(), second.proc_id());
7120 assert_ne!(first.default_location(), second_location);
7121 assert_eq!(second.default_location(), second_location);
7122 }
7123
7124 #[tokio::test]
7125 async fn test_gateway_serve_updates_location_and_stops() {
7126 use crate::mailbox::PortLocation;
7127 use crate::mailbox::monitored_return_handle;
7128
7129 let proc = Proc::isolated();
7130 let gateway = proc.gateway();
7131 let initial_location = proc.default_location();
7132 let client = proc.client("client");
7133 let (port, mut receiver) = client.bind_handler_port::<u64>();
7134 let PortLocation::Bound(default_dest) = port.location() else {
7135 panic!("handler port must be bound");
7136 };
7137
7138 async fn send_to_location(
7139 location: Location,
7140 default_dest: &PortAddr,
7141 value: u64,
7142 receiver: &mut PortReceiver<u64>,
7143 ) {
7144 let dest = PortAddr::new(default_dest.id().clone(), location.clone());
7145 let sender = MailboxClient::dial(location.addr().clone()).unwrap();
7146 sender.post(
7147 MessageEnvelope::serialize(
7148 test_actor_id("sender", "client"),
7149 dest,
7150 &value,
7151 Flattrs::new(),
7152 )
7153 .unwrap(),
7154 monitored_return_handle(),
7155 );
7156 sender.flush().await.unwrap();
7157 let received = tokio::time::timeout(Duration::from_secs(5), receiver.recv())
7158 .await
7159 .unwrap()
7160 .unwrap();
7161 assert_eq!(received, value);
7162 }
7163
7164 let mut server =
7165 Gateway::serve(&gateway, ChannelAddr::any(ChannelTransport::Local)).unwrap();
7166
7167 assert_eq!(proc.default_location(), initial_location);
7168 assert_eq!(proc.default_location(), gateway.default_location());
7169 assert_eq!(
7170 proc.proc_addr(),
7171 ProcAddr::new(proc.proc_id().clone(), gateway.default_location())
7172 );
7173 send_to_location(initial_location.clone(), &default_dest, 1, &mut receiver).await;
7174
7175 let mut next_server =
7176 Gateway::serve(&gateway, ChannelAddr::any(ChannelTransport::Local)).unwrap();
7177 let next_location = proc.default_location();
7178
7179 assert_ne!(proc.default_location(), initial_location);
7180 assert_eq!(proc.default_location(), gateway.default_location());
7181 assert_eq!(
7182 proc.proc_addr(),
7183 ProcAddr::new(proc.proc_id().clone(), gateway.default_location())
7184 );
7185 send_to_location(next_location.clone(), &default_dest, 2, &mut receiver).await;
7186 send_to_location(initial_location.clone(), &default_dest, 3, &mut receiver).await;
7187
7188 next_server.stop("test complete");
7189 next_server.join().await.unwrap();
7190
7191 assert_eq!(proc.default_location(), initial_location);
7192 assert_eq!(proc.default_location(), gateway.default_location());
7193 assert!(MailboxClient::dial(next_location.addr().clone()).is_err());
7194 send_to_location(initial_location.clone(), &default_dest, 4, &mut receiver).await;
7195
7196 server.stop("test complete");
7197 server.join().await.unwrap();
7198
7199 assert_eq!(proc.default_location(), initial_location);
7200 assert_eq!(proc.default_location(), gateway.default_location());
7201 assert!(MailboxClient::dial(initial_location.addr().clone()).is_err());
7202 }
7203
7204 #[tokio::test]
7205 async fn test_direct_proc_server_stops_via_join_mailbox_server() {
7206 let proc = Proc::direct(
7207 ChannelAddr::any(ChannelTransport::Local),
7208 "direct".to_string(),
7209 )
7210 .unwrap();
7211
7212 assert_eq!(
7213 proc.proc_addr(),
7214 ProcAddr::new(proc.proc_id().clone(), proc.gateway().default_location())
7215 );
7216
7217 proc.join_mailbox_server().await;
7218 }
7219
7220 #[tokio::test]
7221 async fn test_local_only_gateway_returns_undeliverable_messages() {
7222 let proc = Proc::isolated();
7223 let client = proc.client("client");
7224 let (return_handle, mut undeliverable_rx) =
7225 client.open_port::<Undeliverable<MessageEnvelope>>();
7226 let remote_proc = ProcAddr::instance(ChannelAddr::Local(1234), "remote");
7227 let remote_dest = remote_proc.actor_addr("worker").port_addr(Port::from(0));
7228
7229 proc.post(
7230 MessageEnvelope::serialize(
7231 test_actor_id("sender", "client"),
7232 remote_dest.clone(),
7233 &123u64,
7234 Flattrs::new(),
7235 )
7236 .unwrap(),
7237 return_handle,
7238 );
7239
7240 let Undeliverable::Returned(envelope) = undeliverable_rx.recv().await.unwrap() else {
7241 panic!("expected returned message");
7242 };
7243 assert_eq!(envelope.dest(), &remote_dest);
7244 }
7245
7246 #[tokio::test]
7247 async fn test_gateway_attach_peer_routes_via_through_sender() {
7248 use tokio::sync::mpsc::UnboundedSender;
7249 use tokio::sync::mpsc::unbounded_channel;
7250
7251 use crate::mailbox::IntoBoxedMailboxSender;
7252 use crate::mailbox::MailboxSender as _;
7253 use crate::mailbox::monitored_return_handle;
7254
7255 #[derive(Clone)]
7256 struct MpscSender(UnboundedSender<MessageEnvelope>);
7257
7258 #[async_trait]
7259 impl crate::mailbox::MailboxSender for MpscSender {
7260 fn post_unchecked(
7261 &self,
7262 envelope: MessageEnvelope,
7263 _return_handle: PortHandle<Undeliverable<MessageEnvelope>>,
7264 ) {
7265 self.0.send(envelope).unwrap();
7266 }
7267 }
7268
7269 let proc = Proc::isolated();
7270 let gateway = proc.gateway();
7271
7272 let remote_uid = Uid::Instance(0xabc123, Some(Label::strip("remote")));
7276 let remote_inner_addr = ChannelAddr::Local(1234);
7277 let remote_location =
7278 Location::from(remote_inner_addr.clone()).with_via(remote_uid.clone());
7279 let remote_proc = ProcAddr::new(ProcId::new(remote_uid.clone(), None), remote_location);
7280 let remote_dest = remote_proc.actor_addr("worker").port_addr(Port::from(0));
7281
7282 let (tx, mut rx) = unbounded_channel::<MessageEnvelope>();
7283 let route_guard = gateway
7284 .attach_peer(remote_uid.clone(), MpscSender(tx).into_boxed())
7285 .unwrap();
7286
7287 gateway.post(
7288 MessageEnvelope::serialize(
7289 test_actor_id("sender", "client"),
7290 remote_dest.clone(),
7291 &42u64,
7292 Flattrs::new(),
7293 )
7294 .unwrap(),
7295 monitored_return_handle(),
7296 );
7297
7298 let routed = tokio::time::timeout(Duration::from_secs(5), rx.recv())
7299 .await
7300 .expect("attach_peer did not route envelope before timeout")
7301 .expect("recording sender closed unexpectedly");
7302 assert_eq!(routed.dest(), &remote_dest);
7305 assert_eq!(routed.next_hop().id(), remote_dest.id());
7306 assert_eq!(routed.next_hop().location().addr(), &remote_inner_addr);
7307 assert!(
7308 !routed.next_hop().location().is_via(),
7309 "outer via hop must be peeled from the next hop"
7310 );
7311
7312 drop(route_guard);
7313
7314 let client = proc.client("client");
7315 let (return_handle, mut undeliverable_rx) =
7316 client.open_port::<Undeliverable<MessageEnvelope>>();
7317 gateway.post(
7318 MessageEnvelope::serialize(
7319 test_actor_id("sender", "client"),
7320 remote_dest.clone(),
7321 &43u64,
7322 Flattrs::new(),
7323 )
7324 .unwrap(),
7325 return_handle,
7326 );
7327 let Undeliverable::Returned(envelope) = undeliverable_rx.recv().await.unwrap() else {
7328 panic!("expected Undeliverable::Returned variant");
7329 };
7330 assert_eq!(envelope.dest(), &remote_dest);
7331 }
7332
7333 #[derive(Debug, Default)]
7334 #[export]
7335 struct LookupTestActor;
7336
7337 impl Actor for LookupTestActor {}
7338
7339 #[derive(Handler, HandleClient, Debug)]
7340 enum LookupTestMessage {
7341 ActorExists(ActorRef<TestActor>, #[reply] OncePortRef<bool>),
7342 }
7343
7344 #[async_trait]
7345 #[crate::handle(LookupTestMessage)]
7346 impl LookupTestMessageHandler for LookupTestActor {
7347 async fn actor_exists(
7348 &mut self,
7349 cx: &crate::Context<Self>,
7350 actor_ref: ActorRef<TestActor>,
7351 ) -> Result<bool, anyhow::Error> {
7352 Ok(actor_ref.downcast_handle(cx).is_some())
7353 }
7354 }
7355
7356 #[async_timed_test(timeout_secs = 30)]
7357 async fn test_actor_lookup() {
7358 let proc = Proc::isolated();
7359 let client = proc.client("client");
7360
7361 let target_actor = proc.spawn(TestActor);
7362 let target_actor_ref = target_actor.bind();
7363 let lookup_actor = proc.spawn(LookupTestActor);
7364
7365 assert!(
7366 lookup_actor
7367 .actor_exists(&client, target_actor_ref.clone())
7368 .await
7369 .unwrap()
7370 );
7371
7372 assert!(
7374 !lookup_actor
7375 .actor_exists(
7376 &client,
7377 ActorRef::attest(target_actor.actor_addr().anonymous_child())
7378 )
7379 .await
7380 .unwrap()
7381 );
7382 assert!(
7384 !lookup_actor
7385 .actor_exists(&client, ActorRef::attest(lookup_actor.actor_addr().clone()))
7386 .await
7387 .unwrap()
7388 );
7389
7390 target_actor.drain_and_stop("test").unwrap();
7391 target_actor.await;
7392
7393 assert!(
7394 !lookup_actor
7395 .actor_exists(&client, target_actor_ref)
7396 .await
7397 .unwrap()
7398 );
7399
7400 lookup_actor.drain_and_stop("test").unwrap();
7401 lookup_actor.await;
7402 }
7403
7404 fn validate_link(child: &InstanceCell, parent: &InstanceCell) {
7405 assert_eq!(
7406 child.actor_addr().proc_addr(),
7407 parent.actor_addr().proc_addr()
7408 );
7409 assert_eq!(child.parent().unwrap().actor_addr(), parent.actor_addr());
7410 assert_matches!(
7411 parent.inner.children.get(child.uid()),
7412 Some(node) if node.actor_addr() == child.actor_addr()
7413 );
7414 }
7415
7416 #[expect(
7417 clippy::await_holding_invalid_type,
7418 reason = "tracing_test::traced_test macro expansion holds tracing::span::Entered across awaits; can't be fixed in our code"
7419 )]
7420 #[tracing_test::traced_test]
7421 #[async_timed_test(timeout_secs = 30)]
7422 async fn test_spawn_child() {
7423 let proc = Proc::isolated();
7424 let client = proc.client("client");
7425
7426 let first = proc.spawn_with_label::<TestActor>("first", TestActor);
7427 let second = TestActor::spawn_child(&client, &first).await;
7428 let third = TestActor::spawn_child(&client, &second).await;
7429
7430 assert!(logs_with_scope_contain(
7432 "hyperactor::proc",
7433 format!(
7434 "{}: spawned with {:?}",
7435 first.actor_addr(),
7436 first.cell().actor_task_handle().unwrap()
7437 )
7438 .as_str()
7439 ));
7440 assert!(logs_with_scope_contain(
7441 "hyperactor::proc",
7442 format!(
7443 "{}: spawned with {:?}",
7444 second.actor_addr(),
7445 second.cell().actor_task_handle().unwrap()
7446 )
7447 .as_str()
7448 ));
7449 assert!(logs_with_scope_contain(
7450 "hyperactor::proc",
7451 format!(
7452 "{}: spawned with {:?}",
7453 third.actor_addr(),
7454 third.cell().actor_task_handle().unwrap()
7455 )
7456 .as_str()
7457 ));
7458
7459 assert_eq!(first.actor_addr().proc_addr(), proc.proc_addr());
7461 assert_eq!(second.actor_addr().proc_addr(), proc.proc_addr());
7462 assert_eq!(third.actor_addr().proc_addr(), proc.proc_addr());
7463
7464 validate_link(third.cell(), second.cell());
7466 validate_link(second.cell(), first.cell());
7467 assert!(first.cell().parent().is_none());
7468
7469 let third_cell = third.cell().clone();
7472 third.drain_and_stop("test").unwrap();
7473 third.await;
7474 assert!(third_cell.inner.children.is_empty());
7475 assert_eq!(
7476 third_cell.parent().unwrap().actor_addr(),
7477 second.actor_addr()
7478 );
7479 assert!(second.cell().get_child(third_cell.uid()).is_none());
7480 drop(third_cell);
7481 validate_link(second.cell(), first.cell());
7482
7483 let second_cell = second.cell().clone();
7484 second.drain_and_stop("test").unwrap();
7485 second.await;
7486 assert!(second_cell.inner.children.is_empty());
7487 drop(second_cell);
7488
7489 let first_cell = first.cell().clone();
7490 first.drain_and_stop("test").unwrap();
7491 first.await;
7492 assert!(first_cell.inner.children.is_empty());
7493 }
7494
7495 #[async_timed_test(timeout_secs = 30)]
7496 async fn expired_parent_link_is_not_a_root() {
7497 let proc = Proc::isolated();
7498 let client = proc.client("client");
7499
7500 let parent = proc.spawn_with_label::<TestActor>("parent", TestActor);
7501 let child = TestActor::spawn_child(&client, &parent).await;
7502 let child_cell = child.cell().clone();
7503 let child_addr = child_cell.actor_addr().clone();
7504
7505 assert!(
7506 proc.state()
7507 .root_instances
7508 .contains_key(parent.actor_addr().id())
7509 );
7510 assert!(!proc.state().root_instances.contains_key(child_addr.id()));
7511
7512 parent.drain_and_stop("test").unwrap();
7513 parent.await;
7514 drop(child);
7515
7516 for i in 0..1000 {
7517 if child_cell.parent().is_none() {
7518 break;
7519 }
7520 if i < 50 {
7521 tokio::task::yield_now().await;
7522 } else {
7523 tokio::time::sleep(Duration::from_millis(50)).await;
7524 }
7525 }
7526 assert!(
7527 child_cell.parent().is_none(),
7528 "parent cell should be released once the parent stops"
7529 );
7530
7531 assert!(!proc.state().root_instances.contains_key(child_addr.id()));
7533 }
7534
7535 #[async_timed_test(timeout_secs = 30)]
7536 async fn zombie_status_sets_unless_terminal() {
7537 let proc = Proc::isolated();
7538 let alive = proc.spawn_with_label::<TestActor>("alive", TestActor);
7539 alive
7540 .status()
7541 .clone()
7542 .wait_for(ActorStatus::is_idle)
7543 .await
7544 .unwrap();
7545 alive
7546 .cell()
7547 .change_status(ActorStatus::zombie("hard kill did not finish"));
7548 assert!(alive.cell().status().borrow().is_zombie());
7549
7550 alive.cell().change_status(ActorStatus::Idle);
7551 assert!(alive.cell().status().borrow().is_zombie());
7552
7553 let stopped = proc.spawn_with_label::<TestActor>("stopped", TestActor);
7554 let stopped_cell = stopped.cell().clone();
7555 stopped.drain_and_stop("test").unwrap();
7556 let terminal_status = stopped.await;
7557 assert!(terminal_status.is_terminal());
7558
7559 stopped_cell.change_status(ActorStatus::zombie("hard kill did not finish"));
7560 assert_eq!(*stopped_cell.status().borrow(), terminal_status);
7561
7562 alive.drain_and_stop("test").unwrap();
7563 alive.await;
7564 }
7565
7566 #[async_timed_test(timeout_secs = 30)]
7567 async fn change_status_rejects_zombie_for_linked_child() {
7568 let proc = Proc::isolated();
7569 let client = proc.client("client");
7570 let parent = proc.spawn_with_label::<TestActor>("parent", TestActor);
7571 let child = TestActor::spawn_child(&client, &parent).await;
7572 child
7573 .status()
7574 .clone()
7575 .wait_for(ActorStatus::is_idle)
7576 .await
7577 .unwrap();
7578
7579 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
7580 child
7581 .cell()
7582 .change_status(ActorStatus::zombie("hard kill did not finish"));
7583 }));
7584 assert!(
7585 result.is_err(),
7586 "`change_status` must reject zombie status for linked children"
7587 );
7588
7589 parent.drain_and_stop("test").unwrap();
7590 assert_matches!(parent.await, ActorStatus::Stopped(reason) if reason == "test");
7591 assert_matches!(child.await, ActorStatus::Stopped(reason) if reason == "parent draining");
7592 }
7593
7594 #[async_timed_test(timeout_secs = 30)]
7595 async fn zombie_status_does_not_complete_actor_handle() {
7596 let proc = Proc::isolated();
7597 let handle = proc.spawn_with_label::<TestActor>("alive", TestActor);
7598 handle
7599 .status()
7600 .clone()
7601 .wait_for(ActorStatus::is_idle)
7602 .await
7603 .unwrap();
7604 handle
7605 .cell()
7606 .change_status(ActorStatus::zombie("hard kill did not finish"));
7607
7608 let await_handle = handle.clone();
7609 let result = tokio::time::timeout(
7610 Duration::from_millis(100),
7611 async move { await_handle.await },
7612 )
7613 .await;
7614 assert!(
7615 result.is_err(),
7616 "zombie actor must not complete ActorHandle::await"
7617 );
7618 handle.drain_and_stop("test").unwrap();
7619 assert_matches!(handle.await, ActorStatus::Stopped(reason) if reason == "test");
7620 }
7621
7622 #[async_timed_test(timeout_secs = 60)]
7624 async fn zombie_task_failure_stores_terminal_event_and_reports_zombie() {
7625 let proc = Proc::isolated();
7626 let client = proc.client("client");
7627 let (mut reported_event, _coordinator) =
7628 ProcSupervisionCoordinator::set(&proc).await.unwrap();
7629 let handle = proc.spawn_with_label::<TestActor>("alive", TestActor);
7630 let actor_id = handle.actor_addr().clone();
7631 let cell = handle.cell().clone();
7632 handle
7633 .status()
7634 .clone()
7635 .wait_for(ActorStatus::is_idle)
7636 .await
7637 .unwrap();
7638 handle
7639 .cell()
7640 .change_status(ActorStatus::zombie("hard kill did not finish"));
7641
7642 handle
7643 .fail(&client, anyhow::anyhow!("zombie failure"))
7644 .await
7645 .unwrap();
7646 let terminal_status = handle.await;
7647 assert!(
7648 terminal_status.is_failed(),
7649 "zombie task failure should reach a failed terminal status"
7650 );
7651
7652 let event = cell
7653 .supervision_event()
7654 .expect("failed zombie must store its terminal supervision event");
7655 assert!(
7656 event.actor_status.is_failed(),
7657 "zombie terminal event should preserve the failed status"
7658 );
7659 assert_eq!(event.actor_status, terminal_status);
7660
7661 let propagated = tokio::time::timeout(Duration::from_secs(1), reported_event.recv())
7662 .await
7663 .expect("zombie supervision event should arrive");
7664 assert!(
7665 propagated.actor_status.is_zombie(),
7666 "zombie task failure should report its zombie lifecycle verdict"
7667 );
7668 assert!(
7669 !propagated.is_error(),
7670 "zombie supervision event should remain non-error"
7671 );
7672
7673 let view = wait_for_terminated_actor_view(&proc, &actor_id).await;
7674 assert_eq!(
7675 view.status, "failed",
7676 "zombie terminal snapshot must be failed"
7677 );
7678 let failure = view
7679 .failure
7680 .expect("failed zombie snapshot must retain failure information");
7681 assert_eq!(failure.root_cause_actor, actor_id);
7682 assert!(
7683 !failure.is_propagated,
7684 "direct zombie failure must not be marked propagated"
7685 );
7686 }
7687
7688 #[async_timed_test(timeout_secs = 60)]
7690 async fn zombie_then_clean_stop_snapshot_has_no_failure() {
7691 let proc = Proc::isolated();
7692 let _client = proc.client("client");
7693 let handle = proc.spawn_with_label::<TestActor>("alive", TestActor);
7694 let actor_id = handle.actor_addr().clone();
7695 let cell = handle.cell().clone();
7696 handle
7697 .status()
7698 .clone()
7699 .wait_for(ActorStatus::is_idle)
7700 .await
7701 .unwrap();
7702 cell.change_status(ActorStatus::zombie("hard kill did not finish"));
7703 assert!(cell.status().borrow().is_zombie());
7704
7705 handle.drain_and_stop("test").unwrap();
7706 assert_matches!(handle.await, ActorStatus::Stopped(reason) if reason == "test");
7707
7708 let view = wait_for_terminated_actor_view(&proc, &actor_id).await;
7709 assert_eq!(view.status, "stopped", "zombie clean stop must be stopped");
7710 assert!(
7711 view.failure.is_none(),
7712 "zombie clean stop must not manufacture failure attrs"
7713 );
7714 }
7715
7716 #[async_timed_test(timeout_secs = 30)]
7717 async fn test_child_lifecycle() {
7718 let proc = Proc::isolated();
7719 let client = proc.client("client");
7720
7721 let root = proc.spawn_with_label::<TestActor>("root", TestActor);
7722 let root_1 = TestActor::spawn_child(&client, &root).await;
7723 let root_2 = TestActor::spawn_child(&client, &root).await;
7724 let root_2_1 = TestActor::spawn_child(&client, &root_2).await;
7725
7726 root.drain_and_stop("test").unwrap();
7727 root.await;
7728
7729 for actor in [root_1, root_2, root_2_1] {
7730 assert!(
7731 actor
7732 .port::<TestActorMessage>()
7733 .try_post(&client, TestActorMessage::Noop())
7734 .is_err()
7735 );
7736 assert_matches!(actor.await, ActorStatus::Stopped(reason) if reason == "parent draining");
7737 }
7738 }
7739
7740 #[derive(Debug)]
7741 struct DrainCountingActor {
7742 handled: Arc<AtomicUsize>,
7743 }
7744
7745 impl Actor for DrainCountingActor {}
7746
7747 #[derive(Handler, Debug)]
7748 enum DrainCountingMessage {
7749 Block(oneshot::Sender<()>, oneshot::Receiver<()>),
7750 Count(),
7751 }
7752
7753 #[async_trait]
7754 #[crate::handle(DrainCountingMessage)]
7755 impl DrainCountingMessageHandler for DrainCountingActor {
7756 async fn block(
7757 &mut self,
7758 _cx: &crate::Context<Self>,
7759 entered: oneshot::Sender<()>,
7760 release: oneshot::Receiver<()>,
7761 ) -> Result<(), anyhow::Error> {
7762 entered.send(()).unwrap();
7763 release.await.unwrap();
7764 Ok(())
7765 }
7766
7767 async fn count(&mut self, _cx: &crate::Context<Self>) -> Result<(), anyhow::Error> {
7768 self.handled.fetch_add(1, Ordering::SeqCst);
7769 Ok(())
7770 }
7771 }
7772
7773 async fn block_and_queue_counts(
7774 client: &Client,
7775 handle: &ActorHandle<DrainCountingActor>,
7776 count: usize,
7777 ) -> oneshot::Sender<()> {
7778 let (entered_tx, entered_rx) = oneshot::channel();
7779 let (release_tx, release_rx) = oneshot::channel();
7780 handle.post(client, DrainCountingMessage::Block(entered_tx, release_rx));
7781 entered_rx.await.unwrap();
7782 for _ in 0..count {
7783 handle.post(client, DrainCountingMessage::Count());
7784 }
7785 release_tx
7786 }
7787
7788 #[async_timed_test(timeout_secs = 30)]
7789 async fn child_inherits_drain_mode() {
7790 let proc = Proc::isolated();
7791 let client = proc.client("client");
7792 let parent = proc.spawn(TestActor);
7793 let handled = Arc::new(AtomicUsize::new(0));
7794 let child = proc.spawn_child(
7795 parent.cell().clone(),
7796 DrainCountingActor {
7797 handled: handled.clone(),
7798 },
7799 );
7800 let release = block_and_queue_counts(&client, &child, 5).await;
7801
7802 parent.drain_and_stop("test").unwrap();
7803 release.send(()).unwrap();
7804
7805 assert_matches!(parent.await, ActorStatus::Stopped(reason) if reason == "test");
7806 assert_matches!(child.await, ActorStatus::Stopped(reason) if reason == "parent draining");
7807 assert_eq!(handled.load(Ordering::SeqCst), 5);
7808 }
7809
7810 #[async_timed_test(timeout_secs = 30)]
7811 async fn tree_inherits_drain_mode() {
7812 let proc = Proc::isolated();
7813 let client = proc.client("client");
7814 let grandparent = proc.spawn(TestActor);
7815 let parent = proc.spawn_child(grandparent.cell().clone(), TestActor);
7816 let handled = Arc::new(AtomicUsize::new(0));
7817 let grandchild = proc.spawn_child(
7818 parent.cell().clone(),
7819 DrainCountingActor {
7820 handled: handled.clone(),
7821 },
7822 );
7823 let release = block_and_queue_counts(&client, &grandchild, 7).await;
7824
7825 grandparent.drain_and_stop("test").unwrap();
7826 release.send(()).unwrap();
7827
7828 assert_matches!(grandparent.await, ActorStatus::Stopped(reason) if reason == "test");
7829 assert_matches!(parent.await, ActorStatus::Stopped(reason) if reason == "parent draining");
7830 assert_matches!(grandchild.await, ActorStatus::Stopped(reason) if reason == "parent draining");
7831 assert_eq!(handled.load(Ordering::SeqCst), 7);
7832 }
7833
7834 #[derive(Debug)]
7835 struct StopFailingActor;
7836
7837 #[async_trait]
7838 impl Actor for StopFailingActor {
7839 async fn handle_stop(
7840 &mut self,
7841 _this: &Instance<Self>,
7842 _mode: StopMode,
7843 _reason: &str,
7844 ) -> Result<(), anyhow::Error> {
7845 Err(anyhow::anyhow!("stop failed"))
7846 }
7847 }
7848
7849 #[async_timed_test(timeout_secs = 30)]
7850 async fn failed_drain_stop_stops_children_immediately() {
7851 let proc = Proc::isolated();
7852 let (_reported, _coordinator) = ProcSupervisionCoordinator::set(&proc).await.unwrap();
7853 let parent = proc.spawn(StopFailingActor);
7854 let child = proc.spawn_child(parent.cell().clone(), TestActor);
7855
7856 parent.drain_and_stop("test").unwrap();
7857
7858 assert_matches!(child.await, ActorStatus::Stopped(reason) if reason == "parent stopping");
7859 assert_matches!(
7860 parent.await,
7861 ActorStatus::Failed(err) if err.to_string().contains("stop failed")
7862 );
7863 }
7864
7865 #[derive(Debug)]
7866 struct DeferredStopActor {
7867 stop_started: Arc<tokio::sync::Notify>,
7868 release_stop: Arc<tokio::sync::Notify>,
7869 }
7870
7871 #[async_trait]
7872 impl Actor for DeferredStopActor {
7873 async fn handle_stop(
7874 &mut self,
7875 this: &Instance<Self>,
7876 mode: StopMode,
7877 reason: &str,
7878 ) -> Result<(), anyhow::Error> {
7879 let this = this.clone_for_py();
7880 let release_stop = Arc::clone(&self.release_stop);
7881 let reason = reason.to_string();
7882 this.close();
7883 self.stop_started.notify_one();
7884 tokio::spawn(async move {
7885 release_stop.notified().await;
7886 match mode {
7887 StopMode::Stop => this.exit(&reason).unwrap(),
7888 StopMode::DrainAndStop => this.exit_after_drain(&reason).unwrap(),
7889 }
7890 });
7891 Ok(())
7892 }
7893 }
7894
7895 #[async_trait]
7896 impl Handler<()> for DeferredStopActor {
7897 async fn handle(&mut self, _cx: &crate::Context<Self>, _message: ()) -> anyhow::Result<()> {
7898 Ok(())
7899 }
7900 }
7901
7902 #[async_timed_test(timeout_secs = 30)]
7903 async fn test_handle_stop_can_defer_exit() {
7904 let proc = Proc::isolated();
7905 let stop_started = Arc::new(tokio::sync::Notify::new());
7906 let release_stop = Arc::new(tokio::sync::Notify::new());
7907 let handle = proc.spawn(DeferredStopActor {
7908 stop_started: Arc::clone(&stop_started),
7909 release_stop: Arc::clone(&release_stop),
7910 });
7911
7912 let mut status = handle.status();
7913 handle.stop("test").unwrap();
7914 stop_started.notified().await;
7915 status.wait_for(ActorStatus::is_stopping).await.unwrap();
7916
7917 release_stop.notify_one();
7918 assert_matches!(handle.await, ActorStatus::Stopped(reason) if reason == "test");
7919 }
7920
7921 #[async_timed_test(timeout_secs = 30)]
7922 async fn test_drain_and_stop_closes_handler_ingress() {
7923 let proc = Proc::isolated();
7924 let client = proc.client("client");
7925 let stop_started = Arc::new(tokio::sync::Notify::new());
7926 let release_stop = Arc::new(tokio::sync::Notify::new());
7927 let handle = proc.spawn(DeferredStopActor {
7928 stop_started: Arc::clone(&stop_started),
7929 release_stop: Arc::clone(&release_stop),
7930 });
7931
7932 handle.drain_and_stop("test").unwrap();
7933 stop_started.notified().await;
7934
7935 let err = handle.port::<()>().try_post(&client, ()).unwrap_err();
7938 assert_matches!(err.kind(), crate::mailbox::MailboxSenderErrorKind::Closed);
7939
7940 release_stop.notify_one();
7941 assert_matches!(handle.await, ActorStatus::Stopped(reason) if reason == "test");
7942 }
7943
7944 #[async_timed_test(timeout_secs = 30)]
7945 async fn test_parent_failure() {
7946 let proc = Proc::isolated();
7947 let client = proc.client("client");
7948 let (_reported, _coordinator) = ProcSupervisionCoordinator::set(&proc).await.unwrap();
7951
7952 let root = proc.spawn_with_label::<TestActor>("root", TestActor);
7953 let root_1 = TestActor::spawn_child(&client, &root).await;
7954 let root_2 = TestActor::spawn_child(&client, &root).await;
7955 let root_2_1 = TestActor::spawn_child(&client, &root_2).await;
7956
7957 root_2.post(
7958 &client,
7959 TestActorMessage::Fail(anyhow::anyhow!("some random failure")),
7960 );
7961 let _root_2_actor_id = root_2.actor_addr().clone();
7962 assert_matches!(
7963 root_2.await,
7964 ActorStatus::Failed(err) if err.to_string() == "some random failure"
7965 );
7966
7967 assert_matches!(
7971 root.await,
7972 ActorStatus::Failed(err) if err.to_string().contains("some random failure")
7973 );
7974 assert_matches!(root_2_1.await, ActorStatus::Stopped(_));
7975 assert_matches!(root_1.await, ActorStatus::Stopped(_));
7976 }
7977
7978 #[async_timed_test(timeout_secs = 30)]
7979 async fn test_multi_handler() {
7980 #[derive(Debug)]
7984 struct TestActor(Arc<AtomicUsize>);
7985
7986 #[async_trait]
7987 impl Actor for TestActor {}
7988
7989 #[async_trait]
7990 impl Handler<OncePortHandle<PortHandle<usize>>> for TestActor {
7991 async fn handle(
7992 &mut self,
7993 cx: &crate::Context<Self>,
7994 message: OncePortHandle<PortHandle<usize>>,
7995 ) -> anyhow::Result<()> {
7996 message.post(cx, cx.port());
7997 Ok(())
7998 }
7999 }
8000
8001 #[async_trait]
8002 impl Handler<usize> for TestActor {
8003 async fn handle(
8004 &mut self,
8005 _cx: &crate::Context<Self>,
8006 message: usize,
8007 ) -> anyhow::Result<()> {
8008 self.0.fetch_add(message, Ordering::SeqCst);
8009 Ok(())
8010 }
8011 }
8012
8013 let proc = Proc::isolated();
8014 let state = Arc::new(AtomicUsize::new(0));
8015 let actor = TestActor(state.clone());
8016 let handle = proc.spawn(actor);
8017 let client = proc.client("client");
8018 let (tx, rx) = client.open_once_port();
8019 handle.post(&client, tx);
8020 let usize_handle = rx.recv().await.unwrap();
8021 usize_handle.post(&client, 123);
8022
8023 handle.drain_and_stop("test").unwrap();
8024 handle.await;
8025
8026 assert_eq!(state.load(Ordering::SeqCst), 123);
8027 }
8028
8029 #[async_timed_test(timeout_secs = 30)]
8030 async fn test_post_after_self_message() {
8031 let proc = Proc::isolated();
8032 let client = proc.client("client");
8033 let (ready, ready_rx) = client.open_once_port();
8034 let (fired, fired_rx) = client.open_once_port();
8035 let delay = Duration::from_millis(50);
8036 let start = tokio::time::Instant::now();
8037 let handle = proc.spawn(DelayedSelfActor {
8038 ready: Some(ready.bind()),
8039 fired: Some(fired.bind()),
8040 delay,
8041 });
8042
8043 ready_rx.recv().await.unwrap();
8044 fired_rx.recv().await.unwrap();
8045
8046 assert!(start.elapsed() >= delay);
8047 handle.drain_and_stop("test").unwrap();
8048 handle.await;
8049 }
8050
8051 #[async_timed_test(timeout_secs = 30)]
8052 async fn test_post_after_port_ref() {
8053 let proc = Proc::isolated();
8054 let client = proc.client("client");
8055 let (reply, mut reply_rx) = client.open_port();
8056 let delay = Duration::from_millis(50);
8057 let start = tokio::time::Instant::now();
8058 let handle = proc.spawn(DelayedPortActor {
8059 reply: Some(reply.bind()),
8060 delay,
8061 });
8062
8063 assert_eq!(reply_rx.recv().await.unwrap(), 123);
8064 assert!(start.elapsed() >= delay);
8065 handle.drain_and_stop("test").unwrap();
8066 handle.await;
8067 }
8068
8069 #[async_timed_test(timeout_secs = 30)]
8070 async fn test_post_after_discards_pending_messages_on_shutdown() {
8071 let proc = Proc::isolated();
8072 let client = proc.client("client");
8073 let (ready, ready_rx) = client.open_once_port();
8074 let (fired, fired_rx) = client.open_once_port();
8075 let handle = proc.spawn(DelayedSelfActor {
8076 ready: Some(ready.bind()),
8077 fired: Some(fired.bind()),
8078 delay: Duration::from_secs(60),
8079 });
8080
8081 ready_rx.recv().await.unwrap();
8082 handle.drain_and_stop("test").unwrap();
8083 assert_matches!(handle.await, ActorStatus::Stopped(reason) if reason == "test");
8084
8085 let result = tokio::time::timeout(Duration::from_millis(100), fired_rx.recv()).await;
8086 assert!(!matches!(result, Ok(Ok(()))));
8087 }
8088
8089 #[async_timed_test(timeout_secs = 30)]
8090 async fn test_actor_panic() {
8091 panic_handler::set_panic_hook();
8093
8094 let proc = Proc::isolated();
8095 let (_reported, _coordinator) = ProcSupervisionCoordinator::set(&proc).await.unwrap();
8098
8099 let client = proc.client("client");
8100 let actor_handle = proc.spawn(TestActor);
8101 actor_handle
8102 .panic(&client, "some random failure".to_string())
8103 .await
8104 .unwrap();
8105 let actor_status = actor_handle.await;
8106
8107 assert_matches!(actor_status, ActorStatus::Failed(_));
8111 if let ActorStatus::Failed(err) = actor_status {
8112 let error_msg = err.to_string();
8113 assert!(error_msg.contains("some random failure"));
8115 assert!(error_msg.contains("library/std/src/panicking.rs"));
8119 }
8120 }
8121
8122 #[async_timed_test(timeout_secs = 30)]
8140 async fn test_local_supervision_propagation() {
8141 hyperactor_telemetry::initialize_logging_for_test();
8142
8143 #[derive(Debug)]
8144 struct TestActor {
8145 handled: Arc<AtomicBool>,
8146 notify: Arc<tokio::sync::Notify>,
8147 should_handle: bool,
8148 }
8149
8150 #[async_trait]
8151 impl Actor for TestActor {
8152 async fn handle_supervision_event(
8153 &mut self,
8154 _this: &Instance<Self>,
8155 _event: &ActorSupervisionEvent,
8156 ) -> Result<bool, anyhow::Error> {
8157 if !self.should_handle {
8158 return Ok(false);
8159 }
8160
8161 tracing::error!(
8162 "{}: supervision event received: {:?}",
8163 _this.self_addr(),
8164 _event
8165 );
8166 self.handled.store(true, Ordering::SeqCst);
8167 self.notify.notify_one();
8168 Ok(true)
8169 }
8170 }
8171
8172 #[async_trait]
8173 impl Handler<String> for TestActor {
8174 async fn handle(
8175 &mut self,
8176 cx: &crate::Context<Self>,
8177 message: String,
8178 ) -> anyhow::Result<()> {
8179 tracing::info!("{} received message: {}", cx.self_addr(), message);
8180 Err(anyhow::anyhow!(message))
8181 }
8182 }
8183
8184 let make_actor = |handled: &Arc<AtomicBool>, should_handle: bool| TestActor {
8185 handled: handled.clone(),
8186 notify: Arc::new(tokio::sync::Notify::new()),
8187 should_handle,
8188 };
8189
8190 let proc = Proc::isolated();
8191 let client = proc.client("client");
8192 let (mut reported_event, _coordinator) =
8193 ProcSupervisionCoordinator::set(&proc).await.unwrap();
8194
8195 let root_state = Arc::new(AtomicBool::new(false));
8196 let root_1_state = Arc::new(AtomicBool::new(false));
8197 let root_1_notify = Arc::new(tokio::sync::Notify::new());
8198 let root_1_1_state = Arc::new(AtomicBool::new(false));
8199 let root_1_1_1_state = Arc::new(AtomicBool::new(false));
8200 let root_2_state = Arc::new(AtomicBool::new(false));
8201 let root_2_1_state = Arc::new(AtomicBool::new(false));
8202
8203 let root = proc.spawn_with_label::<TestActor>("root", make_actor(&root_state, false));
8204 let root_1 = proc.spawn_child::<TestActor>(
8205 root.cell().clone(),
8206 TestActor {
8207 handled: root_1_state.clone(),
8208 notify: root_1_notify.clone(),
8209 should_handle: true, },
8211 );
8212 let root_1_1 = proc
8213 .spawn_child::<TestActor>(root_1.cell().clone(), make_actor(&root_1_1_state, false));
8214 let root_1_1_1 = proc.spawn_child::<TestActor>(
8215 root_1_1.cell().clone(),
8216 make_actor(&root_1_1_1_state, false),
8217 );
8218 let root_2 = proc.spawn_with_label::<TestActor>("root_2", make_actor(&root_2_state, false));
8221 let root_2_1 = proc
8222 .spawn_child::<TestActor>(root_2.cell().clone(), make_actor(&root_2_1_state, false));
8223
8224 root_1_1_1.post(&client, "some random failure".to_string());
8227
8228 let root_2_1_id = root_2_1.actor_addr().clone();
8231 root_2_1.post(&client, "some random failure".to_string());
8232
8233 root_1_notify.notified().await;
8237
8238 let event = reported_event.recv().await;
8241 assert_eq!(event.actor_id, root_2_1_id);
8242
8243 assert!(!root_state.load(Ordering::SeqCst));
8244 assert!(root_1_state.load(Ordering::SeqCst));
8245 assert!(!root_1_1_state.load(Ordering::SeqCst));
8246 assert!(!root_1_1_1_state.load(Ordering::SeqCst));
8247 assert!(!root_2_state.load(Ordering::SeqCst));
8248 assert!(!root_2_1_state.load(Ordering::SeqCst));
8249 }
8250
8251 #[async_timed_test(timeout_secs = 30)]
8252 async fn test_instance() {
8253 #[derive(Debug, Default)]
8254 struct TestActor;
8255
8256 impl Actor for TestActor {}
8257
8258 #[async_trait]
8259 impl Handler<(String, PortRef<String>)> for TestActor {
8260 async fn handle(
8261 &mut self,
8262 cx: &crate::Context<Self>,
8263 (message, port): (String, PortRef<String>),
8264 ) -> anyhow::Result<()> {
8265 port.post(cx, message);
8266 Ok(())
8267 }
8268 }
8269
8270 let proc = Proc::isolated();
8271
8272 let client = proc.client("my_test_actor");
8273 let status = client.status();
8274
8275 let child_actor = client.spawn(TestActor);
8276
8277 let (port, mut receiver) = client.open_port();
8278 child_actor.post(&client, ("hello".to_string(), port.bind()));
8279
8280 let message = receiver.recv().await.unwrap();
8281 assert_eq!(message, "hello");
8282
8283 child_actor.drain_and_stop("test").unwrap();
8284 child_actor.await;
8285
8286 assert_eq!(*status.borrow(), ActorStatus::Client);
8287 drop(client);
8288 assert_matches!(*status.borrow(), ActorStatus::Stopped(_));
8289 }
8290
8291 #[cfg_attr(target_os = "macos", ignore = "tokio runtime fork assertion on macOS")]
8294 #[tokio::test]
8295 async fn test_proc_terminate_without_coordinator() {
8296 if std::env::var("CARGO_TEST").is_ok() {
8297 eprintln!("test skipped as it hangs when run by cargo in sandcastle");
8298 return;
8299 }
8300
8301 let process = async {
8302 let proc = Proc::isolated();
8303 let root = proc.spawn_with_label("root", TestActor);
8307 let client = proc.client("client");
8308 root.fail(&client, anyhow::anyhow!("some random failure"))
8309 .await
8310 .unwrap();
8311 tokio::time::sleep(Duration::from_secs(30)).await;
8315 };
8316
8317 assert_termination(|| process, 1).await.unwrap();
8318 }
8319
8320 fn trace_and_block(fut: impl Future) {
8321 tracing::subscriber::with_default(
8322 tracing_subscriber::Registry::default().with(hyperactor_telemetry::recorder().layer()),
8323 || {
8324 tokio::runtime::Builder::new_current_thread()
8325 .enable_all()
8326 .build()
8327 .unwrap()
8328 .block_on(fut)
8329 },
8330 );
8331 }
8332
8333 #[test]
8334 fn test_handler_logging() {
8335 #[derive(Debug, Default)]
8336 struct LoggingActor;
8337
8338 impl Actor for LoggingActor {}
8339
8340 impl LoggingActor {
8341 async fn wait(cx: &impl context::Actor, handle: &ActorHandle<Self>) {
8342 let barrier = Arc::new(Barrier::new(2));
8343 handle.post(cx, barrier.clone());
8344 barrier.wait().await;
8345 }
8346 }
8347
8348 #[async_trait]
8349 impl Handler<String> for LoggingActor {
8350 async fn handle(
8351 &mut self,
8352 _cx: &crate::Context<Self>,
8353 message: String,
8354 ) -> anyhow::Result<()> {
8355 tracing::info!("{}", message);
8356 Ok(())
8357 }
8358 }
8359
8360 #[async_trait]
8361 impl Handler<u64> for LoggingActor {
8362 async fn handle(
8363 &mut self,
8364 _cx: &crate::Context<Self>,
8365 message: u64,
8366 ) -> anyhow::Result<()> {
8367 tracing::event!(Level::INFO, number = message);
8368 Ok(())
8369 }
8370 }
8371
8372 #[async_trait]
8373 impl Handler<Arc<Barrier>> for LoggingActor {
8374 async fn handle(
8375 &mut self,
8376 _cx: &crate::Context<Self>,
8377 message: Arc<Barrier>,
8378 ) -> anyhow::Result<()> {
8379 message.wait().await;
8380 Ok(())
8381 }
8382 }
8383
8384 #[async_trait]
8385 impl Handler<Arc<(Barrier, Barrier)>> for LoggingActor {
8386 #[expect(
8387 clippy::await_holding_invalid_type,
8388 reason = "tracing_test::traced_test macro expansion holds tracing::span::Entered across awaits; can't be fixed in our code"
8389 )]
8390 async fn handle(
8391 &mut self,
8392 _cx: &crate::Context<Self>,
8393 barriers: Arc<(Barrier, Barrier)>,
8394 ) -> anyhow::Result<()> {
8395 let inner = tracing::span!(Level::INFO, "child_span");
8396 let _inner_guard = inner.enter();
8397 barriers.0.wait().await;
8398 barriers.1.wait().await;
8399 Ok(())
8400 }
8401 }
8402
8403 trace_and_block(async {
8404 let proc = Proc::isolated();
8405 let client = proc.client("client");
8406 let handle = hyperactor::spawn(LoggingActor).into_guard();
8407 handle.post(&client, "hello world".to_string());
8408 handle.post(&client, "hello world again".to_string());
8409 handle.post(&client, 123u64);
8410
8411 LoggingActor::wait(&client, &handle).await;
8412
8413 let events = handle.cell().inner.recording.tail();
8414 assert_eq!(events.len(), 3);
8415 assert_eq!(events[0].json_value(), json!({ "message": "hello world" }));
8416 assert_eq!(
8417 events[1].json_value(),
8418 json!({ "message": "hello world again" })
8419 );
8420 assert_eq!(events[2].json_value(), json!({ "number": 123 }));
8421
8422 let stacks = {
8423 let barriers = Arc::new((Barrier::new(2), Barrier::new(2)));
8424 handle.post(&client, Arc::clone(&barriers));
8425 barriers.0.wait().await;
8426 let stacks = handle.cell().inner.recording.stacks();
8427 barriers.1.wait().await;
8428 stacks
8429 };
8430 assert_eq!(stacks.len(), 1);
8431 assert_eq!(stacks[0].len(), 1);
8432 assert_eq!(stacks[0][0].name(), "child_span");
8433 })
8434 }
8435
8436 #[async_timed_test(timeout_secs = 30)]
8437 async fn test_mailbox_closed_with_owner_stopped_reason() {
8438 let proc = Proc::isolated();
8439 let client = proc.client("client");
8440 let actor_handle = proc.spawn(TestActor);
8441
8442 let handle_for_send = actor_handle.clone();
8444
8445 actor_handle.drain_and_stop("healthy shutdown").unwrap();
8447 actor_handle.await;
8448
8449 let result = handle_for_send
8451 .port::<TestActorMessage>()
8452 .try_post(&client, TestActorMessage::Noop());
8453
8454 assert!(result.is_err(), "send should fail when actor is stopped");
8455 let err = result.unwrap_err();
8456 assert_matches!(
8457 err.kind(),
8458 crate::mailbox::MailboxSenderErrorKind::Mailbox(mailbox_err)
8459 if matches!(
8460 mailbox_err.kind(),
8461 crate::mailbox::MailboxErrorKind::OwnerTerminated(ActorStatus::Stopped(reason)) if reason == "healthy shutdown"
8462 )
8463 );
8464 }
8465
8466 #[async_timed_test(timeout_secs = 30)]
8467 async fn test_mailbox_closed_with_owner_failed_reason() {
8468 let proc = Proc::isolated();
8469 let client = proc.client("client");
8470 let (_reported, _coordinator) = ProcSupervisionCoordinator::set(&proc).await.unwrap();
8473
8474 let actor_handle = proc.spawn(TestActor);
8475
8476 let handle_for_send = actor_handle.clone();
8478
8479 actor_handle.post(
8481 &client,
8482 TestActorMessage::Fail(anyhow::anyhow!("intentional failure")),
8483 );
8484 actor_handle.await;
8485
8486 let result = handle_for_send
8488 .port::<TestActorMessage>()
8489 .try_post(&client, TestActorMessage::Noop());
8490
8491 assert!(result.is_err(), "send should fail when actor has failed");
8492 let err = result.unwrap_err();
8493 assert_matches!(
8494 err.kind(),
8495 crate::mailbox::MailboxSenderErrorKind::Mailbox(mailbox_err)
8496 if matches!(
8497 mailbox_err.kind(),
8498 crate::mailbox::MailboxErrorKind::OwnerTerminated(ActorStatus::Failed(ActorErrorKind::Generic(msg)))
8499 if msg.contains("intentional failure")
8500 )
8501 );
8502 }
8503
8504 async fn wait_for_terminated_snapshot(
8508 proc: &Proc,
8509 actor_id: &ActorAddr,
8510 ) -> crate::introspect::IntrospectResult {
8511 for i in 0..1000 {
8515 if let Some(snapshot) = proc.terminated_snapshot(actor_id) {
8516 return snapshot;
8517 }
8518 if i < 50 {
8519 tokio::task::yield_now().await;
8520 } else {
8521 tokio::time::sleep(Duration::from_millis(50)).await;
8522 }
8523 }
8524 panic!("timed out waiting for terminated snapshot for {}", actor_id);
8525 }
8526
8527 async fn wait_for_terminated_actor_view(
8533 proc: &Proc,
8534 actor_id: &ActorAddr,
8535 ) -> crate::introspect::ActorAttrsView {
8536 let snapshot = wait_for_terminated_snapshot(proc, actor_id).await;
8537 let attrs: hyperactor_config::Attrs =
8538 serde_json::from_str(&snapshot.attrs).expect("snapshot attrs must be valid");
8539 crate::introspect::ActorAttrsView::from_attrs(&attrs)
8540 .expect("terminated snapshot must decode as an actor view (producer-side FI-3/IA-4)")
8541 }
8542
8543 #[async_timed_test(timeout_secs = 60)]
8552 async fn test_terminated_snapshot_stored_on_stop() {
8553 let proc = Proc::isolated();
8554 let _client = proc.client("client");
8555
8556 let handle = proc.spawn(TestActor);
8557 let actor_id = handle.actor_addr().clone();
8558
8559 assert!(proc.terminated_snapshot(&actor_id).is_none());
8561 assert!(!proc.all_terminated_actor_ids().contains(&actor_id));
8562
8563 handle.drain_and_stop("test").unwrap();
8565 handle.await;
8566
8567 let view = wait_for_terminated_actor_view(&proc, &actor_id).await;
8570 assert_eq!(view.status, "stopped", "expected stopped status");
8571 assert!(
8572 view.failure.is_none(),
8573 "clean stop must not carry failure attrs"
8574 );
8575
8576 assert!(proc.all_terminated_actor_ids().contains(&actor_id));
8578 assert!(
8579 !proc.all_actor_ids().contains(&actor_id),
8580 "stopped actor should not appear in live actor IDs"
8581 );
8582 }
8583
8584 #[async_timed_test(timeout_secs = 60)]
8591 async fn test_terminated_snapshot_stored_on_failure() {
8592 let proc = Proc::isolated();
8593 let client = proc.client("client");
8594 let (_reported_event, _coordinator) = ProcSupervisionCoordinator::set(&proc).await.unwrap();
8596
8597 let handle = proc.spawn(TestActor);
8598 let actor_id = handle.actor_addr().clone();
8599
8600 handle.post(&client, TestActorMessage::Fail(anyhow::anyhow!("boom")));
8602 handle.await;
8603
8604 let view = wait_for_terminated_actor_view(&proc, &actor_id).await;
8605 assert_eq!(view.status, "failed", "expected failed status");
8606 assert!(
8607 view.failure.is_some(),
8608 "failed actor snapshot must carry failure attrs"
8609 );
8610 }
8611
8612 #[async_timed_test(timeout_secs = 30)]
8614 async fn test_supervision_event_stored_on_failure() {
8615 let proc = Proc::isolated();
8616 let client = proc.client("client");
8617 ProcSupervisionCoordinator::set(&proc).await.unwrap();
8618
8619 let handle = proc.spawn(TestActor);
8620 let actor_id = handle.actor_addr().clone();
8621 let cell = handle.cell().clone();
8622
8623 handle.post(&client, TestActorMessage::Fail(anyhow::anyhow!("boom")));
8624 handle.await;
8625
8626 let event = cell
8627 .supervision_event()
8628 .expect("failed actor must have supervision_event");
8629 assert_eq!(event.actor_id, actor_id);
8630 assert!(event.actor_status.is_failed());
8631 assert_eq!(event.actually_failing_actor().unwrap().actor_id, actor_id);
8633 }
8634
8635 #[async_timed_test(timeout_secs = 30)]
8637 async fn test_supervision_event_on_clean_stop() {
8638 let proc = Proc::isolated();
8639 let _client = proc.client("client");
8640
8641 let handle = proc.spawn(TestActor);
8642 let cell = handle.cell().clone();
8643
8644 handle.drain_and_stop("test").unwrap();
8645 handle.await;
8646
8647 let event = cell
8648 .supervision_event()
8649 .expect("clean stop must store supervision event");
8650 assert!(
8651 matches!(event.actor_status, ActorStatus::Stopped(_)),
8652 "expected Stopped status, got {:?}",
8653 event.actor_status
8654 );
8655 assert!(!event.is_error());
8656 }
8657
8658 #[async_timed_test(timeout_secs = 30)]
8659 async fn test_supervision_coordinator_receives_clean_stop() {
8660 let proc = Proc::isolated();
8661 let _client = proc.client("client");
8662 let (mut reported_event, _coordinator_handle) =
8663 ProcSupervisionCoordinator::set(&proc).await.unwrap();
8664
8665 let handle = proc.spawn(TestActor);
8666 let actor_id = handle.actor_addr().clone();
8667
8668 handle.drain_and_stop("test").unwrap();
8669 handle.await;
8670
8671 let event = reported_event.recv().await;
8672 assert_eq!(event.actor_id, actor_id);
8673 assert!(
8674 matches!(event.actor_status, ActorStatus::Stopped(_)),
8675 "expected Stopped status, got {:?}",
8676 event.actor_status
8677 );
8678 assert!(!event.is_error());
8679 }
8680
8681 #[async_timed_test(timeout_secs = 30)]
8682 async fn test_coordinator_shuts_down_last_during_destroy() {
8683 let mut proc = Proc::isolated();
8684 let _client = proc.client("client");
8685 let (mut reported_event, _coordinator_handle) =
8686 ProcSupervisionCoordinator::set(&proc).await.unwrap();
8687
8688 let mut actor_ids = Vec::new();
8690 for i in 0..3 {
8691 let handle = proc.spawn_with_label::<TestActor>(&format!("actor_{i}"), TestActor);
8692 actor_ids.push(handle.actor_addr().clone());
8693 }
8694
8695 proc.destroy_and_wait(Duration::from_secs(5), "test")
8700 .await
8701 .unwrap();
8702
8703 let mut received_ids = Vec::new();
8705 for _ in 0..actor_ids.len() {
8706 let event = reported_event.recv().await;
8707 assert!(
8708 matches!(event.actor_status, ActorStatus::Stopped(_)),
8709 "expected Stopped, got {:?}",
8710 event.actor_status
8711 );
8712 received_ids.push(event.actor_id);
8713 }
8714 received_ids.sort();
8715 actor_ids.sort();
8716 assert_eq!(received_ids, actor_ids);
8717 }
8718
8719 #[async_timed_test(timeout_secs = 30)]
8721 async fn test_supervision_event_on_propagated_failure() {
8722 let proc = Proc::isolated();
8723 let client = proc.client("client");
8724 ProcSupervisionCoordinator::set(&proc).await.unwrap();
8725
8726 let parent = proc.spawn_with_label::<TestActor>("parent", TestActor);
8727 let parent_cell = parent.cell().clone();
8728 let (tx, rx) = oneshot::channel();
8730 parent.post(&client, TestActorMessage::Spawn(tx));
8731 let child = rx.await.unwrap();
8732 let child_id = child.actor_addr().clone();
8733
8734 child.post(
8737 &client,
8738 TestActorMessage::Fail(anyhow::anyhow!("child boom")),
8739 );
8740 parent.await;
8741
8742 let event = parent_cell.supervision_event();
8743 assert!(
8744 event.is_some(),
8745 "parent must have supervision_event from propagated failure"
8746 );
8747 let event = event.unwrap();
8748 assert_eq!(event.actually_failing_actor().unwrap().actor_id, child_id);
8750 }
8751
8752 #[async_timed_test(timeout_secs = 30)]
8761 async fn test_resolve_actor_ref_none_for_terminal_actor() {
8762 let proc = Proc::isolated();
8763 let _client = proc.client("client");
8764
8765 let handle = proc.spawn(TestActor);
8766 let actor_ref: ActorRef<TestActor> = handle.bind();
8767
8768 assert!(
8770 proc.resolve_actor_ref(&actor_ref).is_some(),
8771 "live actor should be resolvable"
8772 );
8773
8774 handle.drain_and_stop("test").unwrap();
8775 handle.await;
8776
8777 assert!(
8780 proc.resolve_actor_ref(&actor_ref).is_none(),
8781 "terminal actor must not be resolvable"
8782 );
8783 }
8784
8785 #[async_timed_test(timeout_secs = 60)]
8787 async fn test_terminated_snapshot_has_failure_info() {
8788 let proc = Proc::isolated();
8789 let client = proc.client("client");
8790 let (_reported_event, _coordinator) = ProcSupervisionCoordinator::set(&proc).await.unwrap();
8791
8792 let handle = proc.spawn(TestActor);
8793 let actor_id = handle.actor_addr().clone();
8794
8795 handle.post(&client, TestActorMessage::Fail(anyhow::anyhow!("kaboom")));
8796 handle.await;
8797
8798 let view = wait_for_terminated_actor_view(&proc, &actor_id).await;
8799 assert_eq!(view.status, "failed", "expected failed status");
8800 let failure = view
8801 .failure
8802 .expect("failed actor snapshot must carry failure attrs");
8803 assert!(!failure.error_message.is_empty());
8804 assert_eq!(failure.root_cause_actor, actor_id);
8805 assert!(
8806 !failure.is_propagated,
8807 "direct failure must not be marked propagated"
8808 );
8809 }
8810
8811 #[async_timed_test(timeout_secs = 60)]
8813 async fn test_propagated_failure_info() {
8814 let proc = Proc::isolated();
8815 let client = proc.client("client");
8816 let (_reported_event, _coordinator) = ProcSupervisionCoordinator::set(&proc).await.unwrap();
8817
8818 let parent = proc.spawn_with_label::<TestActor>("parent", TestActor);
8819 let parent_id = parent.actor_addr().clone();
8820
8821 let (tx, rx) = oneshot::channel();
8822 parent.post(&client, TestActorMessage::Spawn(tx));
8823 let child = rx.await.unwrap();
8824 let child_id = child.actor_addr().clone();
8825
8826 child.post(
8827 &client,
8828 TestActorMessage::Fail(anyhow::anyhow!("child fail")),
8829 );
8830 parent.await;
8831
8832 let view = wait_for_terminated_actor_view(&proc, &parent_id).await;
8833 assert_eq!(view.status, "failed", "expected failed status");
8834 let failure = view
8835 .failure
8836 .expect("propagated failure snapshot must carry failure attrs");
8837 assert_eq!(failure.root_cause_actor, child_id);
8838 assert!(
8839 failure.is_propagated,
8840 "child-caused failure must be marked propagated"
8841 );
8842 }
8843
8844 #[async_timed_test(timeout_secs = 30)]
8846 async fn test_spawn_with_name_creates_descriptive_name() {
8847 let proc = Proc::isolated();
8848 let root = proc.spawn_with_label::<TestActor>("root", TestActor);
8849 let handle = proc.spawn_named_child(root.cell().clone(), "my_controller", TestActor);
8850 assert_eq!(
8851 handle.actor_addr().label().unwrap().as_str(),
8852 "my_controller"
8853 );
8854 assert!(!handle.actor_addr().is_root());
8855 }
8856
8857 #[async_timed_test(timeout_secs = 30)]
8859 async fn test_spawn_with_name_increments_index() {
8860 let proc = Proc::isolated();
8861 let root = proc.spawn_with_label::<TestActor>("root", TestActor);
8862 let first = proc.spawn_named_child(root.cell().clone(), "my_controller", TestActor);
8863 let second = proc.spawn_named_child(root.cell().clone(), "my_controller", TestActor);
8864 assert_ne!(first.actor_addr().uid(), second.actor_addr().uid());
8865 }
8866
8867 #[async_timed_test(timeout_secs = 30)]
8870 async fn test_spawn_with_name_preserves_supervision() {
8871 let proc = Proc::isolated();
8872 let root = proc.spawn_with_label::<TestActor>("root", TestActor);
8873 let child = proc.spawn_named_child(root.cell().clone(), "supervised_child", TestActor);
8874 let child_cell = child.cell();
8875 let parent = child_cell.parent().expect("child must have parent");
8876 assert_eq!(parent.actor_addr(), root.actor_addr());
8877 }
8878
8879 #[async_timed_test(timeout_secs = 30)]
8881 async fn test_spawn_unchanged() {
8882 let proc = Proc::isolated();
8883 let root = proc.spawn_with_label::<TestActor>("root", TestActor);
8884 let child = proc.spawn_child(root.cell().clone(), TestActor);
8885 assert!(!child.actor_addr().is_root());
8886 }
8887
8888 #[async_timed_test(timeout_secs = 30)]
8890 async fn test_spawn_with_name_different_names_different_pids() {
8891 let proc = Proc::isolated();
8892 let root = proc.spawn_with_label::<TestActor>("root", TestActor);
8893 let a = proc.spawn_named_child(root.cell().clone(), "controller_a", TestActor);
8894 let b = proc.spawn_named_child(root.cell().clone(), "controller_b", TestActor);
8895 assert_ne!(a.actor_addr().uid(), b.actor_addr().uid());
8896 assert_eq!(a.actor_addr().label().unwrap().as_str(), "controller_a");
8897 assert_eq!(b.actor_addr().label().unwrap().as_str(), "controller_b");
8898 }
8899
8900 #[async_timed_test(timeout_secs = 30)]
8902 async fn test_spawn_with_name_no_child_overwrite() {
8903 let proc = Proc::isolated();
8904 let root = proc.spawn_with_label::<TestActor>("root", TestActor);
8905 let _a = proc.spawn_named_child(root.cell().clone(), "ctrl", TestActor);
8906 let _b = proc.spawn_named_child(root.cell().clone(), "ctrl", TestActor);
8907 let _c = proc.spawn_child(root.cell().clone(), TestActor);
8908 assert_eq!(root.cell().child_count(), 3);
8909 }
8910
8911 #[async_timed_test(timeout_secs = 30)]
8913 async fn test_spawn_with_name_does_not_pollute_roots() {
8914 let proc = Proc::isolated();
8915 let root = proc.spawn_with_label::<TestActor>("root", TestActor);
8916 let _child = proc.spawn_named_child(root.cell().clone(), "foo", TestActor);
8917 let _root = proc.spawn_with_label::<TestActor>("foo", TestActor);
8920 }
8921
8922 #[async_timed_test(timeout_secs = 30)]
8924 async fn test_ai3_controller_actor_ids_unique_across_parents_same_proc() {
8925 let proc = Proc::isolated();
8926 let parent_a = proc.spawn_with_label::<TestActor>("parent_a", TestActor);
8927 let parent_b = proc.spawn_with_label::<TestActor>("parent_b", TestActor);
8928
8929 let ctrl_a =
8931 proc.spawn_named_child(parent_a.cell().clone(), "controller_mesh_a", TestActor);
8932 let ctrl_b =
8933 proc.spawn_named_child(parent_b.cell().clone(), "controller_mesh_b", TestActor);
8934
8935 assert_ne!(
8936 ctrl_a.actor_addr(),
8937 ctrl_b.actor_addr(),
8938 "controller ActorAddrs must be unique across parents"
8939 );
8940 }
8941
8942 #[async_timed_test(timeout_secs = 30)]
8944 async fn test_ai3_no_controller_overwrite_in_parent_or_proc_maps() {
8945 let proc = Proc::isolated();
8946 let parent_a = proc.spawn_with_label::<TestActor>("parent_a", TestActor);
8947 let parent_b = proc.spawn_with_label::<TestActor>("parent_b", TestActor);
8948
8949 let ctrl_a =
8950 proc.spawn_named_child(parent_a.cell().clone(), "controller_mesh_a", TestActor);
8951 let ctrl_b =
8952 proc.spawn_named_child(parent_b.cell().clone(), "controller_mesh_b", TestActor);
8953
8954 assert!(
8956 proc.get_instance(ctrl_a.actor_addr()).is_some(),
8957 "ctrl_a must be resolvable"
8958 );
8959 assert!(
8960 proc.get_instance(ctrl_b.actor_addr()).is_some(),
8961 "ctrl_b must be resolvable"
8962 );
8963 assert_eq!(parent_a.cell().child_count(), 1);
8965 assert_eq!(parent_b.cell().child_count(), 1);
8966 }
8967
8968 #[async_timed_test(timeout_secs = 60)]
8970 async fn test_stopped_snapshot_has_no_failure_info() {
8971 let proc = Proc::isolated();
8972 let _client = proc.client("client");
8973
8974 let handle = proc.spawn(TestActor);
8975 let actor_id = handle.actor_addr().clone();
8976
8977 handle.drain_and_stop("test").unwrap();
8978 handle.await;
8979
8980 let view = wait_for_terminated_actor_view(&proc, &actor_id).await;
8981 assert_eq!(view.status, "stopped", "expected stopped status");
8982 assert!(
8983 view.failure.is_none(),
8984 "stopped actor must not have failure attrs"
8985 );
8986 }
8987
8988 #[async_timed_test(timeout_secs = 10)]
8995 async fn test_queue_depth_increment_decrement() {
8996 let proc = Proc::isolated();
8997 let client = proc.client("client");
8998 let handle = proc.spawn_with_label("qd_test", TestActor);
8999 let actor_ref: crate::ActorRef<TestActor> = handle.bind();
9000 let actor_id = actor_ref.actor_addr().clone();
9001
9002 let cell = proc.get_instance(&actor_id).expect("actor must exist");
9004 assert_eq!(cell.queue_depth(), 0, "initial queue depth should be 0");
9005
9006 let (reply_tx, reply_rx) = oneshot::channel();
9009 let (gate_tx, gate_rx) = oneshot::channel::<()>();
9010 handle.wait(&client, reply_tx, gate_rx).await.unwrap();
9011
9012 reply_rx.await.unwrap();
9014
9015 let (reply2_tx, reply2_rx) = oneshot::channel();
9017 handle.reply(&client, reply2_tx).await.unwrap();
9018
9019 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
9021
9022 let depth = cell.queue_depth();
9024 assert!(
9025 depth >= 1,
9026 "expected queue depth >= 1 while actor is busy, got {depth}"
9027 );
9028
9029 let _ = gate_tx.send(());
9031
9032 reply2_rx.await.unwrap();
9034
9035 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
9037
9038 let depth = cell.queue_depth();
9040 assert_eq!(
9041 depth, 0,
9042 "queue depth should return to 0 after all messages handled"
9043 );
9044 }
9045
9046 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, typeuri::Named)]
9053 struct BufferTestMsg;
9054
9055 #[derive(Debug, Default)]
9056 #[hyperactor::export(handlers = [BufferTestMsg])]
9057 struct BufferTestActor;
9058
9059 #[async_trait]
9060 impl Actor for BufferTestActor {}
9061
9062 #[async_trait]
9063 impl Handler<BufferTestMsg> for BufferTestActor {
9064 async fn handle(
9065 &mut self,
9066 _cx: &crate::Context<Self>,
9067 _msg: BufferTestMsg,
9068 ) -> anyhow::Result<()> {
9069 Ok(())
9070 }
9071 }
9072
9073 #[async_timed_test(timeout_secs = 30)]
9088 async fn test_inbound_ordering_snapshot_callback_publishes_session() {
9089 let config = hyperactor_config::global::lock();
9095 let _g = config.override_key(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER, true);
9096
9097 let proc = Proc::isolated();
9098 let client = proc.client("client");
9099 let handle = proc.spawn_with_label("a", BufferTestActor);
9100 let actor_id = handle.actor_addr().clone();
9101
9102 let _actor_ref: crate::ActorRef<BufferTestActor> = handle.bind();
9107
9108 let handler_port = actor_id.port_addr(Port::handler::<BufferTestMsg>());
9112 let _ = client.sequencer().assign_seq(&handler_port);
9117
9118 for _ in 0..3 {
9122 handle.post(&client, BufferTestMsg);
9123 }
9124 let cell = proc.get_instance(&actor_id).expect("actor exists");
9129 let snapshot = tokio::time::timeout(Duration::from_secs(5), async {
9130 loop {
9131 let snapshot = cell
9132 .inbound_ordering_snapshot()
9133 .expect("snapshot callback should be installed for live actors");
9134 if snapshot
9135 .sessions
9136 .first()
9137 .is_some_and(|session| session.buffered_count == 3)
9138 {
9139 break snapshot;
9140 }
9141 tokio::task::yield_now().await;
9142 }
9143 })
9144 .await
9145 .expect("receiver-local sequencing state should be populated");
9146
9147 assert!(snapshot.enabled, "buffering enabled by override");
9148 assert_eq!(snapshot.sessions.len(), 1, "one client session expected");
9149 let session = &snapshot.sessions[0];
9150 assert_eq!(session.expected_next_seq, 1);
9151 assert_eq!(session.buffered_count, 3);
9152 assert_eq!(session.oldest_buffered_seq, Some(2));
9153 assert_eq!(session.newest_buffered_seq, Some(4));
9154 assert_eq!(
9155 session.sender.as_ref(),
9156 Some(client.mailbox().actor_addr()),
9157 "session owner should be the posting client",
9158 );
9159
9160 let payload = crate::introspect::live_actor_payload(&cell);
9164 let attrs: hyperactor_config::Attrs =
9165 serde_json::from_str(&payload.attrs).expect("payload.attrs is well-formed JSON");
9166 let view = crate::introspect::ActorAttrsView::from_attrs(&attrs)
9167 .expect("attrs decode through ActorAttrsView");
9168 let view_snapshot = view
9169 .inbound_ordering
9170 .as_ref()
9171 .expect("INBOUND_ORDERING attr should be set by build_actor_attrs");
9172 assert_eq!(view_snapshot, &snapshot);
9173
9174 let _: u64 = cell.queue_depth();
9177 let _: u64 = view.queue_depth;
9178 }
9179
9180 #[async_timed_test(timeout_secs = 30)]
9181 async fn test_status_message_reports_delivery_progress() {
9182 let config = hyperactor_config::global::lock();
9183 let _g = config.override_key(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER, true);
9184
9185 let proc = Proc::isolated();
9186 let client = proc.client("client");
9187 let handle = proc.spawn_with_label("a", BufferTestActor);
9188 let actor_id = handle.actor_addr().clone();
9189 let _actor_ref: crate::ActorRef<BufferTestActor> = handle.bind();
9190 let handler_port = actor_id.port_addr(Port::handler::<BufferTestMsg>());
9191
9192 for _ in 0..3 {
9193 handle.post(&client, BufferTestMsg);
9194 }
9195
9196 assert_eq!(client.sequencer().last_sent(&handler_port), Some(3));
9197
9198 let progress = tokio::time::timeout(Duration::from_secs(5), async {
9199 loop {
9200 let (status_reply, status_reply_rx) =
9201 client.open_once_port::<Option<ActorStatus>>();
9202 let (delivery_reply, delivery_reply_rx) =
9203 client.open_once_port::<DeliveryProgressResponse>();
9204 actor_id.status_port().post(
9205 &client,
9206 StatusMessage::GetStatus {
9207 reply: status_reply.bind(),
9208 delivery: Some((
9209 client.sequencer().session_id(),
9210 EndpointId::Handler,
9211 delivery_reply.bind(),
9212 )),
9213 },
9214 );
9215 status_reply_rx
9216 .recv()
9217 .await
9218 .expect("status reply should arrive")
9219 .expect("actor should exist");
9220 let progress = delivery_reply_rx
9221 .recv()
9222 .await
9223 .expect("delivery progress reply should arrive");
9224 let DeliveryProgressResponse::Progress(progress) = progress else {
9225 tokio::task::yield_now().await;
9226 continue;
9227 };
9228 if progress.largest_dequeueable_sequence >= 3 {
9229 break progress;
9230 }
9231 tokio::task::yield_now().await;
9232 }
9233 })
9234 .await
9235 .expect("delivery progress should advance");
9236
9237 assert_eq!(
9238 progress,
9239 DeliveryProgress {
9240 largest_dequeueable_sequence: 3,
9241 }
9242 );
9243 }
9244
9245 #[async_timed_test(timeout_secs = 10)]
9249 async fn test_proc_queue_depth_aggregation_under_pressure() {
9250 let proc = Proc::isolated();
9251 let client = proc.client("client");
9252
9253 let h1 = proc.spawn_with_label("a1", TestActor);
9255 let h2 = proc.spawn_with_label("a2", TestActor);
9256
9257 let (reply1, rx1) = oneshot::channel();
9259 let (gate1, grx1) = oneshot::channel::<()>();
9260 h1.wait(&client, reply1, grx1).await.unwrap();
9261 rx1.await.unwrap();
9262
9263 let (reply2, rx2) = oneshot::channel();
9264 let (gate2, grx2) = oneshot::channel::<()>();
9265 h2.wait(&client, reply2, grx2).await.unwrap();
9266 rx2.await.unwrap();
9267
9268 h1.noop(&client).await.unwrap();
9270 h1.noop(&client).await.unwrap();
9271 h2.noop(&client).await.unwrap();
9272
9273 let aggregate = || -> (u64, u64) {
9276 let mut total: u64 = 0;
9277 let mut max: u64 = 0;
9278 for actor_id in proc.all_instance_keys() {
9279 if let Some(cell) = proc.get_instance_by_id(&actor_id) {
9280 let depth = cell.queue_depth();
9281 total = total.saturating_add(depth);
9282 max = max.max(depth);
9283 }
9284 }
9285 (total, max)
9286 };
9287
9288 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
9291 loop {
9292 let (total, max) = aggregate();
9293 if total >= 3 {
9294 assert!(max >= 1, "expected max >= 1, got {max}");
9295 assert!(max <= total, "PD-1: max ({max}) <= total ({total})");
9296 break;
9297 }
9298 assert!(
9299 tokio::time::Instant::now() < deadline,
9300 "timed out waiting for queue depth >= 3, got {total}",
9301 );
9302 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
9303 }
9304
9305 let _ = gate1.send(());
9307 let _ = gate2.send(());
9308
9309 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
9311 loop {
9312 let (total, _) = aggregate();
9313 if total == 0 {
9314 break;
9315 }
9316 assert!(
9317 tokio::time::Instant::now() < deadline,
9318 "timed out waiting for queue depth to return to 0, got {total}",
9319 );
9320 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
9321 }
9322 }
9323
9324 #[async_timed_test(timeout_secs = 5)]
9329 async fn test_retained_queue_stats_cold_start() {
9330 let proc = Proc::isolated();
9331 assert_eq!(proc.queue_depth_total(), 0);
9332 assert_eq!(proc.queue_depth_high_water_mark(), 0);
9333 assert_eq!(proc.last_nonzero_queue_depth_age_ms(), None);
9334 }
9335
9336 #[async_timed_test(timeout_secs = 10)]
9339 async fn test_retained_queue_stats_burst_then_drain() {
9340 let proc = Proc::isolated();
9341 let client = proc.client("client");
9342 let h = proc.spawn_with_label("ret_test", TestActor);
9343
9344 let (ready_tx, ready_rx) = oneshot::channel();
9346 let (gate_tx, gate_rx) = oneshot::channel::<()>();
9347 h.wait(&client, ready_tx, gate_rx).await.unwrap();
9348 ready_rx.await.unwrap();
9349
9350 h.noop(&client).await.unwrap();
9352 h.noop(&client).await.unwrap();
9353
9354 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
9356 loop {
9357 let hwm = proc.queue_depth_high_water_mark();
9358 if hwm >= 2 {
9359 assert!(hwm >= proc.queue_depth_total());
9361 let age = proc.last_nonzero_queue_depth_age_ms();
9363 assert!(
9364 age.is_some(),
9365 "last-nonzero should be Some while pressure is active"
9366 );
9367 assert!(age.unwrap() < 2000, "last-nonzero age should be near zero");
9368 break;
9369 }
9370 assert!(
9371 tokio::time::Instant::now() < deadline,
9372 "timed out waiting for watermark >= 2",
9373 );
9374 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
9375 }
9376
9377 let _ = gate_tx.send(());
9379 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
9380 loop {
9381 if proc.queue_depth_total() == 0 {
9382 break;
9383 }
9384 assert!(
9385 tokio::time::Instant::now() < deadline,
9386 "timed out waiting for total to drain",
9387 );
9388 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
9389 }
9390
9391 assert!(
9393 proc.queue_depth_high_water_mark() >= 2,
9394 "watermark should retain the peak after drain",
9395 );
9396
9397 let age = proc.last_nonzero_queue_depth_age_ms();
9399 assert!(age.is_some(), "last-nonzero should be Some after pressure");
9400 }
9401
9402 #[test]
9406 fn test_last_nonzero_refreshed_on_dequeue_deterministic() {
9407 use std::sync::atomic::AtomicU64;
9408
9409 static FAKE_NOW: AtomicU64 = AtomicU64::new(0);
9410 fn fake_clock() -> u64 {
9411 FAKE_NOW.load(Ordering::Relaxed)
9412 }
9413
9414 let stats = ProcQueueStats::with_clock(fake_clock);
9415 let depth = Arc::new(AtomicU64::new(0));
9416
9417 assert_eq!(stats.last_nonzero_age_ms(), None);
9419
9420 FAKE_NOW.store(1000, Ordering::Relaxed);
9422 account_enqueue(&depth, &stats, "a");
9423 account_enqueue(&depth, &stats, "a");
9424 assert_eq!(stats.running_total(), 2);
9425 assert_eq!(stats.high_water_mark(), 2);
9426
9427 FAKE_NOW.store(2000, Ordering::Relaxed);
9429 assert_eq!(stats.last_nonzero_age_ms(), Some(1000));
9430
9431 FAKE_NOW.store(3000, Ordering::Relaxed);
9434 account_dequeue(&depth, &stats, "a");
9435 assert_eq!(stats.running_total(), 1);
9436
9437 FAKE_NOW.store(4000, Ordering::Relaxed);
9441 assert_eq!(stats.last_nonzero_age_ms(), Some(1000));
9442
9443 FAKE_NOW.store(5000, Ordering::Relaxed);
9447 account_dequeue(&depth, &stats, "a");
9448 assert_eq!(stats.running_total(), 0);
9449
9450 FAKE_NOW.store(6000, Ordering::Relaxed);
9452 assert_eq!(stats.last_nonzero_age_ms(), Some(3000));
9453
9454 assert_eq!(stats.high_water_mark(), 2);
9456 }
9457
9458 #[test]
9464 fn test_account_cancel_enqueue_restores_counters() {
9465 let stats = ProcQueueStats::new();
9466 let depth = Arc::new(AtomicU64::new(0));
9467
9468 account_enqueue(&depth, &stats, "a");
9469 assert_eq!(stats.running_total(), 1);
9470 assert_eq!(depth.load(Ordering::Relaxed), 1);
9471
9472 account_cancel_enqueue(&depth, &stats, "a");
9473 assert_eq!(
9474 stats.running_total(),
9475 0,
9476 "cancel must restore running_total"
9477 );
9478 assert_eq!(
9479 depth.load(Ordering::Relaxed),
9480 0,
9481 "cancel must restore queue_depth"
9482 );
9483
9484 assert_eq!(stats.high_water_mark(), 1);
9486
9487 account_enqueue(&depth, &stats, "a");
9490 assert_eq!(stats.running_total(), 1);
9491 }
9492
9493 #[test]
9494 fn child_teardown_distinguishes_kill_from_failure() {
9495 let actor_addr = test_actor_id("proc", "actor");
9496
9497 let stopped = Ok(ActorStopped {
9498 reason: "test".to_string(),
9499 stop_mode: StopMode::DrainAndStop,
9500 });
9501 assert_eq!(
9502 ChildTeardown::from_run_result(&stopped),
9503 ChildTeardown::Cooperative(StopMode::DrainAndStop)
9504 );
9505
9506 let killed = Err(ActorError::new(
9507 &actor_addr,
9508 ActorErrorKind::Aborted("test kill".to_string()),
9509 ));
9510 assert_eq!(ChildTeardown::from_run_result(&killed), ChildTeardown::Kill);
9511
9512 let failed = Err(ActorError::new(
9513 &actor_addr,
9514 ActorErrorKind::Generic("test failure".to_string()),
9515 ));
9516 assert_eq!(
9517 ChildTeardown::from_run_result(&failed),
9518 ChildTeardown::Cooperative(StopMode::Stop)
9519 );
9520 }
9521
9522 hyperactor_config::attrs::declare_attrs! {
9523 attr AENV_TEST_TAG: u64;
9524 }
9525
9526 const AENV_SENTINEL: u64 = 0xA0FE;
9529
9530 #[derive(Debug)]
9535 #[hyperactor::export(())]
9536 struct EnvProbe {
9537 reply: PortRef<u64>,
9538 }
9539
9540 #[async_trait]
9541 impl Actor for EnvProbe {
9542 async fn init(&mut self, this: &Instance<Self>) -> Result<(), anyhow::Error> {
9543 let observed = this.actor_environment().get(AENV_TEST_TAG).unwrap_or(0);
9544 self.reply.post(this, observed);
9545 Ok(())
9546 }
9547 }
9548
9549 #[async_trait]
9550 impl Handler<()> for EnvProbe {
9551 async fn handle(&mut self, _cx: &Context<Self>, _message: ()) -> anyhow::Result<()> {
9552 Ok(())
9553 }
9554 }
9555
9556 #[async_trait]
9557 impl crate::RemoteSpawn for EnvProbe {
9558 type Params = PortRef<u64>;
9559
9560 async fn new(reply: PortRef<u64>, _environment: &ActorEnvironment) -> anyhow::Result<Self> {
9561 Ok(Self { reply })
9562 }
9563 }
9564
9565 crate::register_spawnable!(EnvProbe);
9566
9567 #[async_timed_test(timeout_secs = 30)]
9569 async fn aenv_ordinary_instances_start_empty() {
9570 let proc = Proc::isolated();
9571 let client = proc.client("client");
9572 assert!(
9573 context::Actor::instance(&client)
9574 .actor_environment()
9575 .is_empty()
9576 );
9577
9578 let (tx, mut rx) = client.open_port::<u64>();
9581 let _root = proc.spawn(EnvProbe { reply: tx.bind() });
9582 assert_eq!(rx.recv().await.unwrap(), 0);
9583 }
9584
9585 #[async_timed_test(timeout_secs = 30)]
9588 async fn aenv_local_children_inherit_environment() {
9589 let proc = Proc::isolated();
9590 let client = proc.client("client");
9591
9592 let mut seed = ActorEnvironment::default();
9593 seed.set(AENV_TEST_TAG, AENV_SENTINEL)
9594 .expect("seed actor environment");
9595 let parent = proc
9596 .actor_instance_in_environment::<EnvProbe>("seeded", seed.clone())
9597 .expect("create seeded root instance");
9598 assert_eq!(parent.instance.actor_environment(), &seed);
9599
9600 assert_eq!(parent.instance.child().0.actor_environment(), &seed);
9602
9603 let client_parent = proc
9604 .actor_instance_in_environment::<ClientActor>("seeded_client", seed.clone())
9605 .expect("create seeded client instance");
9606 let child_client = client_parent.instance.child_client();
9607 assert_eq!(
9608 context::Actor::instance(&child_client).actor_environment(),
9609 &seed,
9610 );
9611
9612 let (tx, mut rx) = client.open_port::<u64>();
9614 let _ordinary = parent.instance.spawn(EnvProbe { reply: tx.bind() });
9615 assert_eq!(rx.recv().await.unwrap(), AENV_SENTINEL);
9616
9617 let (tx, mut rx) = client.open_port::<u64>();
9619 let _named = parent
9620 .instance
9621 .spawn_with_label("named", EnvProbe { reply: tx.bind() });
9622 assert_eq!(rx.recv().await.unwrap(), AENV_SENTINEL);
9623
9624 let (tx, mut rx) = client.open_port::<u64>();
9626 let _uid = parent
9627 .instance
9628 .spawn_with_uid(crate::id::Uid::anonymous(), EnvProbe { reply: tx.bind() })
9629 .unwrap();
9630 assert_eq!(rx.recv().await.unwrap(), AENV_SENTINEL);
9631
9632 let (tx, mut rx) = client.open_port::<u64>();
9634 let params = bincode::serde::encode_to_vec(tx.bind(), bincode::config::legacy()).unwrap();
9635 let type_name = crate::actor::remote::Remote::global()
9636 .name_of::<EnvProbe>()
9637 .expect("EnvProbe is registered");
9638 let _gspawned = parent.instance.gspawn(type_name, params).await.unwrap();
9639 assert_eq!(rx.recv().await.unwrap(), AENV_SENTINEL);
9640 }
9641}