diff --git a/src/actor.rs b/src/actor.rs index baf46fc..b1bccc3 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -443,6 +443,22 @@ impl Actor { pub fn new(inner: A) -> Self { Self { inner } } + + pub(crate) fn inner(&self) -> &A { + &self.inner + } + + pub(crate) fn replace_inner(&mut self, inner: A) { + self.inner = inner; + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ActorTypeMetadata { + pub actor_type_id: TypeId, + pub actor_type_name: &'static str, + pub message_type_id: TypeId, + pub message_type_name: &'static str, } /// Trait for type-erased actors — single-message handler. @@ -456,6 +472,12 @@ pub trait AnyActor: Send { /// Called on graceful stop, before removal. See [`ActorInterface::on_stop`]. fn on_stop(&mut self, _ctx: &Ctx) {} + + fn metadata(&self) -> ActorTypeMetadata; + + fn as_any(&self) -> &dyn Any; + + fn as_any_mut(&mut self) -> &mut dyn Any; } impl AnyActor for Actor @@ -493,6 +515,23 @@ where fn on_stop(&mut self, ctx: &Ctx) { self.inner.on_stop(ctx); } + + fn metadata(&self) -> ActorTypeMetadata { + ActorTypeMetadata { + actor_type_id: TypeId::of::(), + actor_type_name: std::any::type_name::(), + message_type_id: TypeId::of::(), + message_type_name: std::any::type_name::(), + } + } + + fn as_any(&self) -> &dyn Any { + self + } + + fn as_any_mut(&mut self) -> &mut dyn Any { + self + } } /// Unique token identifying a monitor subscription. diff --git a/src/admin.rs b/src/admin.rs new file mode 100644 index 0000000..d949c5e --- /dev/null +++ b/src/admin.rs @@ -0,0 +1,163 @@ +use crate::actor::{ActorAddress, ActorInterface, ActorTypeMetadata, AnyActor, Message}; +use crate::runtime::{Inbox, Runtime}; +use parking_lot::Mutex; +use std::any::Any; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; + +pub type AdminResult = Result; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AdminError { + ActorNotFound { + actor: ActorAddress, + }, + AddressMismatch { + requested: ActorAddress, + snapshot: ActorAddress, + }, + TypeMismatch { + expected_actor_type: &'static str, + expected_message_type: &'static str, + actual_actor_type: &'static str, + actual_message_type: &'static str, + }, + Timeout, +} + +pub struct RuntimeAdmin<'a> { + pub(crate) runtime: &'a Runtime, +} + +pub struct Admin { + pub(crate) inbox: Inbox>, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OperationResult { + pub applied: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ActorStatus { + pub started: bool, + pub suspended: bool, + pub stopping: bool, + pub poisoned: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ActorSummary { + pub address: ActorAddress, + pub actor_type: &'static str, + pub message_type: &'static str, + pub worker_id: usize, + pub parent: Option, + pub mailbox_depth: usize, + pub status: ActorStatus, + pub last_message_type: Option<&'static str>, + pub messages_handled: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ListActorsResponse { + pub actors: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct InspectActorResponse { + pub summary: ActorSummary, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ActorStateSnapshot { + pub actor: ActorAddress, + pub actor_type: &'static str, + pub message_type: &'static str, + pub actor_instance: A, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GetActorStateResponse { + pub state: ActorStateSnapshot, +} + +impl ActorStateSnapshot { + pub fn new(actor: ActorAddress, actor_instance: A) -> Self { + Self { + actor, + actor_type: std::any::type_name::(), + message_type: std::any::type_name::(), + actor_instance, + } + } +} + +impl Admin { + pub(crate) fn new(inbox: Inbox>) -> Self { + Self { inbox } + } + + pub fn try_recv(&self) -> Option> { + self.inbox.try_recv() + } + + pub fn recv_ticking(&self, rt: &Runtime, max_ticks: usize) -> AdminResult { + for _ in 0..max_ticks { + rt.tick(); + if let Some(resp) = self.inbox.try_recv() { + return resp; + } + } + Err(AdminError::Timeout) + } + + pub fn reply_addr(&self) -> &ActorAddress { + self.inbox.addr() + } +} + +pub(crate) type AdminBoxedReply = Box; + +pub(crate) enum AdminCommand { + ListActors { + acc: Arc, + }, + InspectActor { + actor: ActorAddress, + reply_to: ActorAddress, + }, + GetActorState { + actor: ActorAddress, + reply_to: ActorAddress, + get: Box< + dyn FnOnce(ActorAddress, &dyn AnyActor, ActorTypeMetadata) -> AdminBoxedReply + Send, + >, + not_found: Box AdminBoxedReply + Send>, + }, + ReplaceActorState { + actor: ActorAddress, + reply_to: ActorAddress, + replace: Box< + dyn FnOnce(&mut dyn AnyActor, ActorTypeMetadata) -> AdminResult + Send, + >, + }, + StopActor { + actor: ActorAddress, + reply_to: ActorAddress, + }, + SuspendActor { + actor: ActorAddress, + reply_to: ActorAddress, + }, + ResumeActor { + actor: ActorAddress, + reply_to: ActorAddress, + }, +} + +pub(crate) struct ListActorsAccumulator { + pub(crate) remaining: AtomicUsize, + pub(crate) summaries: Mutex>, + pub(crate) reply_to: ActorAddress, +} diff --git a/src/lib.rs b/src/lib.rs index 9396d65..3fe5533 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,4 +1,5 @@ pub mod actor; +pub mod admin; pub mod extension; pub mod process_observer; pub mod worker; diff --git a/src/runtime.rs b/src/runtime.rs index e9ed2ce..e9dffc9 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,15 +1,19 @@ use crate::Instant; -use std::any::Any; +use std::any::{Any, TypeId}; use std::cell::RefCell; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, OnceLock}; use std::thread::Thread; #[cfg(not(target_arch = "wasm32"))] use std::thread::{self, JoinHandle}; use crate::actor::{ - Actor, ActorAddress, ActorInterface, AnyActor, Environment, ExitValue, Message, ResumeSignal, - SpawnRequest, StopSignal, StopWithSignal, SystemInfo, + Actor, ActorAddress, ActorInterface, ActorTypeMetadata, AnyActor, Environment, ExitValue, + Message, ResumeSignal, SpawnRequest, StopSignal, StopWithSignal, SystemInfo, +}; +use crate::admin::{ + ActorStateSnapshot, Admin, AdminCommand, AdminError, AdminResult, GetActorStateResponse, + InspectActorResponse, ListActorsAccumulator, ListActorsResponse, OperationResult, RuntimeAdmin, }; use crate::channel::{Receiver, Sender}; // Re-export config types so existing code using `runtime::RuntimeConfig` still works @@ -106,6 +110,7 @@ pub struct Runtime { extension: Option>, transfer_txs: Vec>, spawn_txs: Vec>, + admin_txs: Vec>, placement: Placement, is_running: AtomicBool, worker_stats: Vec>, @@ -221,6 +226,7 @@ impl Runtime { let mut transfer_txs = Vec::with_capacity(num_workers); let mut spawn_txs = Vec::with_capacity(num_workers); + let mut admin_txs = Vec::with_capacity(num_workers); let mut worker_stats = Vec::with_capacity(num_workers); let mut workers = Vec::with_capacity(num_workers); @@ -233,9 +239,19 @@ impl Runtime { let spawn_tx = spawn_rx.new_sender(); spawn_txs.push(spawn_tx); + let admin_rx = Receiver::::new(config.channel_buffer_size); + let admin_tx = admin_rx.new_sender(); + admin_txs.push(admin_tx); + let stats = Arc::new(WorkerStats::new()); worker_stats.push(stats.clone()); - workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx, stats)); + workers.push(Worker::new( + WorkerId(i), + transfer_rx, + spawn_rx, + admin_rx, + stats, + )); } let placement = Placement::new(num_workers, worker_stats.clone()); @@ -250,6 +266,7 @@ impl Runtime { extension: None, transfer_txs, spawn_txs, + admin_txs, placement, is_running: AtomicBool::new(false), worker_stats, @@ -342,6 +359,10 @@ impl Runtime { self.extension.as_deref() } + pub fn admin(&self) -> RuntimeAdmin<'_> { + RuntimeAdmin { runtime: self } + } + /// Send a request and get a handle for the response. /// /// Creates a temporary inbox, calls `msg_builder` with the inbox's address @@ -590,6 +611,218 @@ impl Runtime { } } +impl RuntimeAdmin<'_> { + fn new_admin(&self) -> Result<(Admin, ActorAddress), Error> { + let inbox = self.runtime.new_inbox::>()?; + let reply_to = *inbox.addr(); + Ok((Admin::new(inbox), reply_to)) + } + + fn ready(&self, result: AdminResult) -> Result, Error> { + let (admin, reply_to) = self.new_admin::()?; + let _ = self + .runtime + .inbox_registry + .try_deliver(reply_to, Box::new(result)); + Ok(admin) + } + + pub fn list_actors(&self) -> Result, Error> { + let (admin, reply_to) = self.new_admin::()?; + let acc = Arc::new(ListActorsAccumulator { + remaining: AtomicUsize::new(self.runtime.admin_txs.len()), + summaries: parking_lot::Mutex::new(Vec::new()), + reply_to, + }); + + for (idx, tx) in self.runtime.admin_txs.iter().enumerate() { + tx.send(AdminCommand::ListActors { acc: acc.clone() }); + notify_worker(&self.runtime.worker_threads, idx); + } + + Ok(admin) + } + + pub fn inspect_actor(&self, actor: ActorAddress) -> Result, Error> { + let Some(wid) = self.runtime.address_map.lookup(&actor) else { + return self.ready::(Err(AdminError::ActorNotFound { actor })); + }; + let (admin, reply_to) = self.new_admin::()?; + let worker_idx = wid.as_usize(); + self.runtime.admin_txs[worker_idx].send(AdminCommand::InspectActor { actor, reply_to }); + notify_worker(&self.runtime.worker_threads, worker_idx); + Ok(admin) + } + + pub fn get_actor_state( + &self, + actor: ActorAddress, + ) -> Result>, Error> + where + A: ActorInterface + Clone + Sync, + { + let Some(wid) = self.runtime.address_map.lookup(&actor) else { + return self + .ready::>(Err(AdminError::ActorNotFound { actor })); + }; + let (admin, reply_to) = self.new_admin::>()?; + + let get = Box::new( + |actor: ActorAddress, + erased: &dyn AnyActor, + metadata: ActorTypeMetadata| + -> Box { + let expected_actor_type = std::any::type_name::(); + let expected_message_type = std::any::type_name::(); + if metadata.actor_type_id != TypeId::of::() + || metadata.message_type_id != TypeId::of::() + { + return Box::new(Err::, AdminError>( + AdminError::TypeMismatch { + expected_actor_type, + expected_message_type, + actual_actor_type: metadata.actor_type_name, + actual_message_type: metadata.message_type_name, + }, + )); + } + + let Some(typed) = erased.as_any().downcast_ref::>() else { + return Box::new(Err::, AdminError>( + AdminError::TypeMismatch { + expected_actor_type, + expected_message_type, + actual_actor_type: metadata.actor_type_name, + actual_message_type: metadata.message_type_name, + }, + )); + }; + + Box::new(Ok::, AdminError>( + GetActorStateResponse { + state: ActorStateSnapshot { + actor, + actor_type: metadata.actor_type_name, + message_type: metadata.message_type_name, + actor_instance: typed.inner().clone(), + }, + }, + )) + }, + ); + let not_found = Box::new(|actor| { + Box::new(Err::, AdminError>( + AdminError::ActorNotFound { actor }, + )) as Box + }); + + let worker_idx = wid.as_usize(); + self.runtime.admin_txs[worker_idx].send(AdminCommand::GetActorState { + actor, + reply_to, + get, + not_found, + }); + notify_worker(&self.runtime.worker_threads, worker_idx); + Ok(admin) + } + + pub fn replace_actor_state( + &self, + actor: ActorAddress, + state: ActorStateSnapshot, + ) -> Result, Error> + where + A: ActorInterface, + { + if state.actor != actor { + return self.ready::(Err(AdminError::AddressMismatch { + requested: actor, + snapshot: state.actor, + })); + } + + let Some(wid) = self.runtime.address_map.lookup(&actor) else { + return self.ready::(Err(AdminError::ActorNotFound { actor })); + }; + let (admin, reply_to) = self.new_admin::()?; + + let actor_instance = state.actor_instance; + let replace = Box::new( + move |erased: &mut dyn AnyActor, + metadata: ActorTypeMetadata| + -> AdminResult { + let expected_actor_type = std::any::type_name::(); + let expected_message_type = std::any::type_name::(); + if metadata.actor_type_id != TypeId::of::() + || metadata.message_type_id != TypeId::of::() + { + return Err(AdminError::TypeMismatch { + expected_actor_type, + expected_message_type, + actual_actor_type: metadata.actor_type_name, + actual_message_type: metadata.message_type_name, + }); + } + + let Some(typed) = erased.as_any_mut().downcast_mut::>() else { + return Err(AdminError::TypeMismatch { + expected_actor_type, + expected_message_type, + actual_actor_type: metadata.actor_type_name, + actual_message_type: metadata.message_type_name, + }); + }; + + typed.replace_inner(actor_instance); + Ok(OperationResult { applied: true }) + }, + ); + + let worker_idx = wid.as_usize(); + self.runtime.admin_txs[worker_idx].send(AdminCommand::ReplaceActorState { + actor, + reply_to, + replace, + }); + notify_worker(&self.runtime.worker_threads, worker_idx); + Ok(admin) + } + + pub fn stop_actor(&self, actor: ActorAddress) -> Result, Error> { + let Some(wid) = self.runtime.address_map.lookup(&actor) else { + return self.ready::(Err(AdminError::ActorNotFound { actor })); + }; + let (admin, reply_to) = self.new_admin::()?; + let worker_idx = wid.as_usize(); + self.runtime.admin_txs[worker_idx].send(AdminCommand::StopActor { actor, reply_to }); + notify_worker(&self.runtime.worker_threads, worker_idx); + Ok(admin) + } + + pub fn suspend_actor(&self, actor: ActorAddress) -> Result, Error> { + let Some(wid) = self.runtime.address_map.lookup(&actor) else { + return self.ready::(Err(AdminError::ActorNotFound { actor })); + }; + let (admin, reply_to) = self.new_admin::()?; + let worker_idx = wid.as_usize(); + self.runtime.admin_txs[worker_idx].send(AdminCommand::SuspendActor { actor, reply_to }); + notify_worker(&self.runtime.worker_threads, worker_idx); + Ok(admin) + } + + pub fn resume_actor(&self, actor: ActorAddress) -> Result, Error> { + let Some(wid) = self.runtime.address_map.lookup(&actor) else { + return self.ready::(Err(AdminError::ActorNotFound { actor })); + }; + let (admin, reply_to) = self.new_admin::()?; + let worker_idx = wid.as_usize(); + self.runtime.admin_txs[worker_idx].send(AdminCommand::ResumeActor { actor, reply_to }); + notify_worker(&self.runtime.worker_threads, worker_idx); + Ok(admin) + } +} + /// Wake a parked worker thread so it can process new work. /// No-op if the thread handle hasn't been registered yet (single-threaded tick mode). #[inline] diff --git a/src/worker.rs b/src/worker.rs index 087d4c1..4b43c75 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -11,6 +11,10 @@ use crate::actor::{ ActorAddress, AnyActor, ContextInner, Ctx, Environment, ExitValue, ResumeSignal, SpawnRequest, StopReason, StopSignal, StopWithSignal, SystemInfo, }; +use crate::admin::{ + ActorStatus, ActorSummary, AdminCommand, AdminError, AdminResult, InspectActorResponse, + ListActorsResponse, OperationResult, +}; use crate::channel::Receiver; use crate::delivery::{AddrBuildHasher, AddrMap, Envelope, TickContext, WorkerId}; use crate::stats::{ActorSnapshot, TickTiming, WorkerStats}; @@ -69,6 +73,7 @@ pub(crate) struct Worker { pub(crate) pool: ActorPool, transfer_rx: Receiver, spawn_rx: Receiver, + admin_rx: Receiver, stats: Arc, /// Reusable scratch buffer for building per-actor snapshots. snapshot_buf: Vec, @@ -84,6 +89,7 @@ impl Worker { id: WorkerId, transfer_rx: Receiver, spawn_rx: Receiver, + admin_rx: Receiver, stats: Arc, ) -> Self { Self { @@ -91,6 +97,7 @@ impl Worker { pool: ActorPool::new(), transfer_rx, spawn_rx, + admin_rx, stats, snapshot_buf: Vec::new(), worker_ext: None, @@ -132,6 +139,88 @@ impl Worker { did_work } + fn drain_admin(&mut self, tc: &TickContext) -> bool { + let mut did_work = false; + while let Some(cmd) = self.admin_rx.try_recv() { + did_work = true; + self.apply_admin_command(tc, cmd); + } + did_work + } + + fn send_admin_reply( + tc: &TickContext, + reply_to: ActorAddress, + result: AdminResult, + ) { + let _ = tc.inbox_registry.try_deliver(reply_to, Box::new(result)); + } + + fn apply_admin_command(&mut self, tc: &TickContext, cmd: AdminCommand) { + match cmd { + AdminCommand::ListActors { acc } => { + let mut local = Vec::new(); + self.pool.actor_summaries_into(self.id, &mut local); + { + let mut summaries = acc.summaries.lock(); + summaries.extend(local); + } + if acc.remaining.fetch_sub(1, Ordering::AcqRel) == 1 { + let actors = { + let mut summaries = acc.summaries.lock(); + std::mem::take(&mut *summaries) + }; + Self::send_admin_reply(tc, acc.reply_to, Ok(ListActorsResponse { actors })); + } + } + AdminCommand::InspectActor { actor, reply_to } => { + let result = self + .pool + .actor_summary(self.id, actor) + .map(|summary| InspectActorResponse { summary }); + Self::send_admin_reply(tc, reply_to, result); + } + AdminCommand::GetActorState { + actor, + reply_to, + get, + not_found, + } => { + let boxed = match self.pool.get_actor_erased(actor) { + Some(erased) => get(actor, erased, erased.metadata()), + None => not_found(actor), + }; + let _ = tc.inbox_registry.try_deliver(reply_to, boxed); + } + AdminCommand::ReplaceActorState { + actor, + reply_to, + replace, + } => { + let result = match self.pool.get_actor_erased_mut(actor) { + Some(erased) => { + let metadata = erased.metadata(); + replace(erased, metadata) + } + None => Err(AdminError::ActorNotFound { actor }), + }; + Self::send_admin_reply(tc, reply_to, result); + } + AdminCommand::StopActor { actor, reply_to } => { + let result = self.pool.stop_actor_admin(actor, &self.stats); + Self::send_admin_reply(tc, reply_to, result); + } + AdminCommand::SuspendActor { actor, reply_to } => { + let result = self.pool.suspend_actor_admin(actor); + Self::send_admin_reply(tc, reply_to, result); + } + AdminCommand::ResumeActor { actor, reply_to } => { + let result = self.pool.resume_actor_admin(actor); + Self::send_admin_reply(tc, reply_to, result); + } + } + } + /// Phase 7: clean up dead actors, deliver death notifications, GC extension state. fn cleanup_dead_actors(&mut self, tc: &TickContext) -> bool { let cleanup_pending: RefCell)>> = @@ -197,6 +286,7 @@ impl Worker { if !self.has_backlog && self.spawn_rx.is_empty() && self.transfer_rx.is_empty() + && self.admin_rx.is_empty() && !self .worker_ext .as_ref() @@ -221,7 +311,10 @@ impl Worker { } let t2 = Instant::now(); - // 2.5. Fire per-worker extension (e.g., timers) → deliver before tick_all + // 3. Drain admin queue → inspect or mutate worker-owned slots before handlers + did_work |= self.drain_admin(tc); + + // 4. Fire per-worker extension (e.g., timers) → deliver before tick_all let ext_msgs: Vec<_> = self .worker_ext .as_mut() @@ -232,7 +325,7 @@ impl Worker { did_work = true; } - // 3. Tick all actors with WorkerContext + // 5. Tick all actors with WorkerContext let pending_local: RefCell)>> = RefCell::new(Vec::new()); let stop_requests: RefCell> = RefCell::new(Vec::new()); @@ -275,12 +368,12 @@ impl Worker { ); } - // 4. Drain spawn queue again — actors spawned during step 3 + // 6. Drain spawn queue again — actors spawned during step 5 // must be in the pool before pending_local delivery. did_work |= self.drain_spawns(tc); let t4 = Instant::now(); - // 5. Drain pending_local buffer → deliver to local actors + // 7. Drain pending_local buffer → deliver to local actors let pending = pending_local.into_inner(); if !pending.is_empty() { did_work = true; @@ -289,7 +382,7 @@ impl Worker { self.pool.deliver(&addr, msg); } - // 5.5. Process worker extension requests from handlers (e.g., timer scheduling) + // 7.5. Process worker extension requests from handlers (e.g., timer scheduling) if let Some(ext) = &mut self.worker_ext { for request in worker_requests.into_inner() { ext.handle_request(request); @@ -298,7 +391,7 @@ impl Worker { let t5 = Instant::now(); - // 6. Publish stats (skip entirely when idle to avoid allocation + mutex) + // 8. Publish stats (skip entirely when idle to avoid allocation + mutex) if did_work { self.stats .num_actors @@ -344,7 +437,7 @@ impl Worker { ); } - // 7. Clean up poisoned and stopping actors + // 9. Clean up poisoned and stopping actors did_work |= self.cleanup_dead_actors(tc); self.has_backlog = did_work; @@ -545,6 +638,95 @@ impl ActorPool { } } + fn get_actor_erased(&self, addr: ActorAddress) -> Option<&dyn AnyActor> { + self.actors.get(&addr).map(|slot| slot.actor.as_ref()) + } + + fn get_actor_erased_mut( + &mut self, + addr: ActorAddress, + ) -> Option<&mut (dyn AnyActor + 'static)> { + self.actors.get_mut(&addr).map(|slot| slot.actor.as_mut()) + } + + fn actor_summary_from_slot( + worker_id: WorkerId, + address: ActorAddress, + slot: &ActorSlot, + ) -> ActorSummary { + let metadata = slot.actor.metadata(); + ActorSummary { + address, + actor_type: metadata.actor_type_name, + message_type: metadata.message_type_name, + worker_id: worker_id.as_usize(), + parent: slot.parent_addr, + mailbox_depth: slot.mailbox.len(), + status: ActorStatus { + started: slot.started, + suspended: slot.suspended, + stopping: slot.stopping, + poisoned: slot.poisoned, + }, + last_message_type: slot.last_msg_type, + messages_handled: slot.messages_processed, + } + } + + fn actor_summary(&self, worker_id: WorkerId, addr: ActorAddress) -> AdminResult { + self.actors + .get(&addr) + .map(|slot| Self::actor_summary_from_slot(worker_id, addr, slot)) + .ok_or(AdminError::ActorNotFound { actor: addr }) + } + + fn actor_summaries_into(&self, worker_id: WorkerId, out: &mut Vec) { + out.clear(); + out.extend( + self.actors + .iter() + .map(|(&addr, slot)| Self::actor_summary_from_slot(worker_id, addr, slot)), + ); + } + + fn suspend_actor_admin(&mut self, addr: ActorAddress) -> AdminResult { + match self.actors.get_mut(&addr) { + Some(slot) => { + slot.suspended = true; + Ok(OperationResult { applied: true }) + } + None => Err(AdminError::ActorNotFound { actor: addr }), + } + } + + fn resume_actor_admin(&mut self, addr: ActorAddress) -> AdminResult { + match self.actors.get_mut(&addr) { + Some(slot) => { + slot.suspended = false; + Ok(OperationResult { applied: true }) + } + None => Err(AdminError::ActorNotFound { actor: addr }), + } + } + + fn stop_actor_admin( + &mut self, + addr: ActorAddress, + stats: &WorkerStats, + ) -> AdminResult { + match self.actors.get_mut(&addr) { + Some(slot) => { + if !slot.stopping { + stats.stops.fetch_add(1, Ordering::Relaxed); + } + slot.stopping = true; + slot.mailbox.clear(); + Ok(OperationResult { applied: true }) + } + None => Err(AdminError::ActorNotFound { actor: addr }), + } + } + /// Tick all actors in the pool. Returns the number of messages processed. /// /// Each actor processes up to `budget` messages per tick (0 = unlimited). diff --git a/tests/runtime_admin.rs b/tests/runtime_admin.rs new file mode 100644 index 0000000..28ec3f7 --- /dev/null +++ b/tests/runtime_admin.rs @@ -0,0 +1,608 @@ +//! Runtime Admin API tests — inventory, typed actor state, lifecycle control, and scheduling. + +mod common; +use common::*; + +use std::collections::HashSet; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; + +use swactor::admin::{ActorStateSnapshot, AdminError, OperationResult}; +use swactor::config::RuntimeConfig; + +#[derive(Clone)] +struct AddAndReport { + delta: usize, + reply_to: ActorAddress, +} + +#[derive(Clone)] +struct ReplaceProbe { + value: usize, + started: Arc, + stopped: Arc, +} + +impl ActorInterface for ReplaceProbe { + type Incoming = AddAndReport; + type Response = Count; + + fn on_start(&mut self, _ctx: &Ctx) { + self.started.fetch_add(1, Ordering::SeqCst); + } + + fn on_stop(&mut self, _ctx: &Ctx) { + self.stopped.fetch_add(1, Ordering::SeqCst); + } + + fn handle(&mut self, ctx: &Ctx, msg: AddAndReport) { + self.value += msg.delta; + let _ = ctx.send(msg.reply_to, Count(self.value)); + } +} + +struct WrongProbe; + +impl ActorInterface for WrongProbe { + type Incoming = Ping; + type Response = Pong; + + fn handle(&mut self, _ctx: &Ctx, _msg: Ping) {} +} + +struct StopProbe { + started: Arc, + handled: Arc, + stopped: Arc, +} + +impl ActorInterface for StopProbe { + type Incoming = Ping; + type Response = Pong; + + fn on_start(&mut self, _ctx: &Ctx) { + self.started.fetch_add(1, Ordering::SeqCst); + } + + fn on_stop(&mut self, _ctx: &Ctx) { + self.stopped.fetch_add(1, Ordering::SeqCst); + } + + fn handle(&mut self, ctx: &Ctx, msg: Ping) { + self.handled.fetch_add(1, Ordering::SeqCst); + let _ = ctx.send(msg.reply_to, Pong); + } +} + +fn poll_admin( + admin: &swactor::admin::Admin, + timeout: Duration, +) -> Option> { + let start = Instant::now(); + while start.elapsed() < timeout { + if let Some(value) = admin.try_recv() { + return Some(value); + } + std::thread::sleep(Duration::from_millis(5)); + } + None +} + +fn poll_inbox(inbox: &Inbox, timeout: Duration) -> Option { + let start = Instant::now(); + while start.elapsed() < timeout { + if let Some(value) = inbox.try_recv() { + return Some(value); + } + std::thread::sleep(Duration::from_millis(5)); + } + None +} + +fn operation_applied() -> OperationResult { + OperationResult { applied: true } +} + +#[test] +fn ask_recv_ticking_delivers_reply_through_runtime_inbox() { + let rt = std_runtime(RuntimeConfig::default()); + let actor = rt.spawn(SelfAddrActor).unwrap(); + + let ask = rt + .ask::(actor, |reply_to| WhoAreYou { reply_to }) + .unwrap(); + + assert_eq!( + ask.try_recv(), + None, + "ask reply is not available before ticking" + ); + assert_eq!( + ask.recv_ticking(&rt, 5).unwrap(), + MyAddr(actor), + "recv_ticking drives the runtime inbox reply path" + ); +} + +#[test] +fn admin_list_and_inspect_report_actor_slot_metadata() { + let rt = std_runtime(RuntimeConfig::default()); + let ping_pong = rt.spawn(PingPongActor).unwrap(); + let counter = rt.spawn(CounterActor { count: 0 }).unwrap(); + rt.tick(); + + let count_inbox = rt.new_inbox::().unwrap(); + rt.send_to( + counter, + Increment { + reply_to: *count_inbox.addr(), + }, + ) + .unwrap(); + rt.send_to( + counter, + Increment { + reply_to: *count_inbox.addr(), + }, + ) + .unwrap(); + + assert_eq!(tick_until_recv(&rt, &count_inbox, 5), Some(Count(1))); + assert_eq!(tick_until_recv(&rt, &count_inbox, 5), Some(Count(2))); + + let response = rt + .admin() + .list_actors() + .unwrap() + .recv_ticking(&rt, 5) + .unwrap(); + + assert_eq!( + response + .actors + .iter() + .filter(|summary| summary.address == ping_pong) + .count(), + 1, + "ping-pong actor appears exactly once in inventory" + ); + assert_eq!( + response + .actors + .iter() + .filter(|summary| summary.address == counter) + .count(), + 1, + "counter actor appears exactly once in inventory" + ); + + let counter_summary = response + .actors + .iter() + .find(|summary| summary.address == counter) + .expect("counter summary missing"); + + assert_eq!(counter_summary.worker_id, 0); + assert_eq!(counter_summary.parent, None); + assert_eq!(counter_summary.mailbox_depth, 0); + assert!(counter_summary.status.started); + assert!(!counter_summary.status.suspended); + assert!(!counter_summary.status.stopping); + assert!(!counter_summary.status.poisoned); + assert_eq!(counter_summary.messages_handled, 2); + assert!(counter_summary.actor_type.ends_with("CounterActor")); + assert!(counter_summary.message_type.ends_with("Increment")); + + let inspect = rt + .admin() + .inspect_actor(counter) + .unwrap() + .recv_ticking(&rt, 5) + .unwrap(); + assert_eq!(inspect.summary, *counter_summary); + + let missing = ActorAddress::new_random(); + let missing_result = rt + .admin() + .inspect_actor(missing) + .unwrap() + .recv_ticking(&rt, 5); + assert!( + matches!(missing_result, Err(AdminError::ActorNotFound { actor }) if actor == missing), + "missing actor is reported through AdminResult" + ); +} + +#[test] +fn admin_get_and_replace_actor_state_preserves_slot_metadata() { + let rt = std_runtime(RuntimeConfig::default()); + let started = Arc::new(AtomicUsize::new(0)); + let stopped = Arc::new(AtomicUsize::new(0)); + let addr = rt + .spawn(ReplaceProbe { + value: 1, + started: started.clone(), + stopped: stopped.clone(), + }) + .unwrap(); + rt.tick(); + + assert_eq!(started.load(Ordering::SeqCst), 1); + assert_eq!(stopped.load(Ordering::SeqCst), 0); + + let count_inbox = rt.new_inbox::().unwrap(); + rt.send_to( + addr, + AddAndReport { + delta: 1, + reply_to: *count_inbox.addr(), + }, + ) + .unwrap(); + assert_eq!(tick_until_recv(&rt, &count_inbox, 5), Some(Count(2))); + + let state = rt + .admin() + .get_actor_state::(addr) + .unwrap() + .recv_ticking(&rt, 5) + .unwrap() + .state; + assert_eq!(state.actor, addr); + assert!(state.actor_type.ends_with("ReplaceProbe")); + assert!(state.message_type.ends_with("AddAndReport")); + assert_eq!(state.actor_instance.value, 2); + + let replacement = ActorStateSnapshot::new( + addr, + ReplaceProbe { + value: 100, + started: started.clone(), + stopped: stopped.clone(), + }, + ); + let replace_result = rt + .admin() + .replace_actor_state::(addr, replacement) + .unwrap() + .recv_ticking(&rt, 5) + .unwrap(); + assert_eq!(replace_result, operation_applied()); + assert_eq!( + started.load(Ordering::SeqCst), + 1, + "replacement does not call on_start" + ); + assert_eq!( + stopped.load(Ordering::SeqCst), + 0, + "replacement does not call on_stop" + ); + + rt.send_to( + addr, + AddAndReport { + delta: 1, + reply_to: *count_inbox.addr(), + }, + ) + .unwrap(); + assert_eq!(tick_until_recv(&rt, &count_inbox, 5), Some(Count(101))); + + let summary = rt + .admin() + .inspect_actor(addr) + .unwrap() + .recv_ticking(&rt, 5) + .unwrap() + .summary; + assert_eq!(summary.address, addr); + assert_eq!(summary.worker_id, 0); + assert_eq!( + summary.messages_handled, 2, + "state replacement preserves slot-owned message counters" + ); + + let stop_result = rt + .admin() + .stop_actor(addr) + .unwrap() + .recv_ticking(&rt, 5) + .unwrap(); + assert_eq!(stop_result, operation_applied()); + rt.tick(); + assert_eq!(stopped.load(Ordering::SeqCst), 1); +} + +#[test] +fn admin_replace_rejects_wrong_actor_type_and_wrong_snapshot_address() { + let rt = std_runtime(RuntimeConfig::default()); + let started = Arc::new(AtomicUsize::new(0)); + let stopped = Arc::new(AtomicUsize::new(0)); + let addr = rt + .spawn(ReplaceProbe { + value: 10, + started: started.clone(), + stopped: stopped.clone(), + }) + .unwrap(); + rt.tick(); + + let wrong_type_snapshot = ActorStateSnapshot::new(addr, WrongProbe); + let wrong_type = rt + .admin() + .replace_actor_state::(addr, wrong_type_snapshot) + .unwrap() + .recv_ticking(&rt, 5); + assert!( + matches!(wrong_type, Err(AdminError::TypeMismatch { .. })), + "wrong concrete actor type is rejected" + ); + + let wrong_addr = ActorAddress::new_random(); + let wrong_addr_snapshot = ActorStateSnapshot::new( + wrong_addr, + ReplaceProbe { + value: 50, + started: started.clone(), + stopped: stopped.clone(), + }, + ); + let wrong_address = rt + .admin() + .replace_actor_state::(addr, wrong_addr_snapshot) + .unwrap() + .recv_ticking(&rt, 5); + assert!( + matches!(wrong_address, Err(AdminError::AddressMismatch { requested, snapshot }) if requested == addr && snapshot == wrong_addr), + "snapshot address must match the target address" + ); + + let count_inbox = rt.new_inbox::().unwrap(); + rt.send_to( + addr, + AddAndReport { + delta: 1, + reply_to: *count_inbox.addr(), + }, + ) + .unwrap(); + assert_eq!( + tick_until_recv(&rt, &count_inbox, 5), + Some(Count(11)), + "failed replacements do not mutate the original actor state" + ); +} + +#[test] +fn admin_suspend_queues_messages_until_resume() { + let rt = std_runtime(RuntimeConfig::default()); + let counter = Arc::new(AtomicUsize::new(0)); + let addr = rt + .spawn(CountingPingActor { + counter: counter.clone(), + }) + .unwrap(); + rt.tick(); + + let suspend_result = rt + .admin() + .suspend_actor(addr) + .unwrap() + .recv_ticking(&rt, 5) + .unwrap(); + assert_eq!(suspend_result, operation_applied()); + + let pong_inbox = rt.new_inbox::().unwrap(); + for _ in 0..3 { + rt.send_to( + addr, + Ping { + reply_to: *pong_inbox.addr(), + }, + ) + .unwrap(); + } + tick_n(&rt, 5); + assert_eq!(counter.load(Ordering::SeqCst), 0); + assert_eq!(pong_inbox.try_recv(), None); + + let suspended = rt + .admin() + .inspect_actor(addr) + .unwrap() + .recv_ticking(&rt, 5) + .unwrap() + .summary; + assert!(suspended.status.suspended); + assert_eq!(suspended.mailbox_depth, 3); + + let resume_result = rt + .admin() + .resume_actor(addr) + .unwrap() + .recv_ticking(&rt, 5) + .unwrap(); + assert_eq!(resume_result, operation_applied()); + for _ in 0..3 { + assert_eq!(tick_until_recv(&rt, &pong_inbox, 5), Some(Pong)); + } + assert_eq!(counter.load(Ordering::SeqCst), 3); + + let resumed = rt + .admin() + .inspect_actor(addr) + .unwrap() + .recv_ticking(&rt, 5) + .unwrap() + .summary; + assert!(!resumed.status.suspended); + assert_eq!(resumed.mailbox_depth, 0); + assert_eq!(resumed.messages_handled, 3); +} + +#[test] +fn admin_stop_clears_pending_mailbox_without_calling_handle() { + let rt = std_runtime(RuntimeConfig::default()); + let started = Arc::new(AtomicUsize::new(0)); + let handled = Arc::new(AtomicUsize::new(0)); + let stopped = Arc::new(AtomicUsize::new(0)); + let addr = rt + .spawn(StopProbe { + started: started.clone(), + handled: handled.clone(), + stopped: stopped.clone(), + }) + .unwrap(); + rt.tick(); + assert_eq!(started.load(Ordering::SeqCst), 1); + + let pong_inbox = rt.new_inbox::().unwrap(); + for _ in 0..5 { + rt.send_to( + addr, + Ping { + reply_to: *pong_inbox.addr(), + }, + ) + .unwrap(); + } + + let stop_result = rt + .admin() + .stop_actor(addr) + .unwrap() + .recv_ticking(&rt, 5) + .unwrap(); + assert_eq!(stop_result, operation_applied()); + rt.tick(); + + assert_eq!(handled.load(Ordering::SeqCst), 0); + assert_eq!(stopped.load(Ordering::SeqCst), 1); + assert_eq!(pong_inbox.try_recv(), None); + assert!( + rt.send_to( + addr, + Ping { + reply_to: *pong_inbox.addr(), + }, + ) + .is_err(), + "admin-stopped actor is removed from normal send routing" + ); + + let inspect = rt.admin().inspect_actor(addr).unwrap().recv_ticking(&rt, 5); + assert!( + matches!(inspect, Err(AdminError::ActorNotFound { actor }) if actor == addr), + "admin-stopped actor is no longer inspectable" + ); +} + +#[test] +fn threaded_admin_suspend_resume_wakes_parked_worker() { + let rt = std_runtime(RuntimeConfig { + num_threads: 2, + ..Default::default() + }); + let counter = Arc::new(AtomicUsize::new(0)); + let addr = rt + .spawn(CountingPingActor { + counter: counter.clone(), + }) + .unwrap(); + let pong_inbox = rt.new_inbox::().unwrap(); + + let handle = rt.run().unwrap(); + std::thread::sleep(Duration::from_millis(50)); + + let suspended = handle.runtime.admin().suspend_actor(addr).unwrap(); + let suspended = poll_admin(&suspended, Duration::from_secs(1)); + + handle + .runtime + .send_to( + addr, + Ping { + reply_to: *pong_inbox.addr(), + }, + ) + .unwrap(); + let pong_while_suspended = poll_inbox(&pong_inbox, Duration::from_millis(100)); + let count_while_suspended = counter.load(Ordering::SeqCst); + + let resumed = handle.runtime.admin().resume_actor(addr).unwrap(); + let resumed = poll_admin(&resumed, Duration::from_secs(1)); + let pong_after_resume = poll_inbox(&pong_inbox, Duration::from_secs(1)); + let final_count = counter.load(Ordering::SeqCst); + + handle.shutdown(); + handle.join(); + + assert_eq!(suspended, Some(Ok(operation_applied()))); + assert_eq!(pong_while_suspended, None); + assert_eq!(count_while_suspended, 0); + assert_eq!(resumed, Some(Ok(operation_applied()))); + assert_eq!(pong_after_resume, Some(Pong)); + assert_eq!(final_count, 1); +} + +#[test] +fn admin_list_actors_aggregates_all_workers() { + let rt = std_runtime(RuntimeConfig { + num_threads: 4, + max_actors: 100, + ..Default::default() + }); + let mut addrs = Vec::new(); + for _ in 0..16 { + addrs.push(rt.spawn(CounterActor { count: 0 }).unwrap()); + } + + let handle = rt.run().unwrap(); + let start = Instant::now(); + let mut response = None; + while start.elapsed() < Duration::from_secs(1) { + let admin = handle.runtime.admin().list_actors().unwrap(); + if let Some(Ok(list)) = poll_admin(&admin, Duration::from_millis(100)) { + if list.actors.len() == addrs.len() { + response = Some(list); + break; + } + } + std::thread::sleep(Duration::from_millis(5)); + } + + handle.shutdown(); + handle.join(); + + let response = response.expect("admin list did not observe all spawned actors within timeout"); + let expected: HashSet<_> = addrs.iter().copied().collect(); + let actual: HashSet<_> = response + .actors + .iter() + .map(|summary| summary.address) + .collect(); + assert_eq!(actual, expected); + for addr in &addrs { + assert_eq!( + response + .actors + .iter() + .filter(|summary| summary.address == *addr) + .count(), + 1, + "actor {addr} appears exactly once in aggregated list" + ); + } + + let worker_ids: HashSet<_> = response + .actors + .iter() + .map(|summary| summary.worker_id) + .collect(); + assert!( + worker_ids.len() >= 2, + "aggregation should include actors from at least two workers, got {worker_ids:?}" + ); +}