diff --git a/crates/pulsing-actor/src/lib.rs b/crates/pulsing-actor/src/lib.rs index d4f82a38d..d56bb38a2 100644 --- a/crates/pulsing-actor/src/lib.rs +++ b/crates/pulsing-actor/src/lib.rs @@ -198,6 +198,7 @@ pub mod prelude { ActorSystem, ActorSystemCoreExt, ActorSystemOpsExt, ResolveOptions, SpawnOptions, SystemConfig, }; + pub use crate::system_actor::{ShmBackend, ShmManager, ShmRegionDescriptor, ShmStats}; pub use async_trait::async_trait; pub use serde::{Deserialize, Serialize}; } diff --git a/crates/pulsing-actor/src/system/handle.rs b/crates/pulsing-actor/src/system/handle.rs index 11cfcaf4a..d7612df44 100644 --- a/crates/pulsing-actor/src/system/handle.rs +++ b/crates/pulsing-actor/src/system/handle.rs @@ -4,6 +4,7 @@ use crate::actor::{ActorId, ActorPath, Envelope}; use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; +use std::time::Instant; use tokio::sync::mpsc; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; @@ -57,6 +58,9 @@ pub(crate) struct LocalActorHandle { /// Static metadata provided by the actor pub metadata: HashMap, + /// Local monotonic start time used by the authoritative actor catalog. + pub started_at: Instant, + /// Named actor path (if this is a named actor) pub named_path: Option, diff --git a/crates/pulsing-actor/src/system/handler.rs b/crates/pulsing-actor/src/system/handler.rs index 37f3607ce..9c44219ea 100644 --- a/crates/pulsing-actor/src/system/handler.rs +++ b/crates/pulsing-actor/src/system/handler.rs @@ -663,6 +663,7 @@ mod tests { cancel_token: tokio_util::sync::CancellationToken::new(), stats: Arc::new(ActorStats::default()), metadata: HashMap::new(), + started_at: std::time::Instant::now(), named_path: None, actor_id, }; @@ -693,6 +694,7 @@ mod tests { cancel_token: tokio_util::sync::CancellationToken::new(), stats: Arc::new(ActorStats::default()), metadata: HashMap::new(), + started_at: std::time::Instant::now(), named_path: None, actor_id, }; diff --git a/crates/pulsing-actor/src/system/lifecycle.rs b/crates/pulsing-actor/src/system/lifecycle.rs index afe32b98b..75034d648 100644 --- a/crates/pulsing-actor/src/system/lifecycle.rs +++ b/crates/pulsing-actor/src/system/lifecycle.rs @@ -95,6 +95,7 @@ impl ActorSystem { pub async fn shutdown(&self) -> Result<()> { tracing::info!("Shutting down actor system"); + self.node_lifecycle.begin_draining()?; self.cancel_token.cancel(); tokio::time::sleep(Duration::from_millis(100)).await; @@ -138,13 +139,25 @@ impl ActorSystem { self.registry.clear_lifecycle().await; + // System services perform their normal drain in `SystemActor::on_stop`, + // but ActorSystem is the final owner of host resources. Revoke every + // descriptor even if SystemActor was absent, failed, or timed out. + self.shm_manager.clear(); + { let cluster_guard = self.cluster.read().await; if let Some(cluster) = cluster_guard.as_ref() { - cluster.leave().await?; + if let Err(error) = cluster.leave().await { + let _ = self + .node_lifecycle + .transition(crate::system_actor::NodeState::Failed); + return Err(error); + } } } + self.node_lifecycle + .transition(crate::system_actor::NodeState::Stopped)?; tracing::info!("Actor system shutdown complete"); Ok(()) } @@ -216,3 +229,39 @@ impl ActorSystem { self.notify_monitor_actor_stopped(actor_name, actor_id, &reason_copy); } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::system::SystemConfig; + use crate::system_actor::{ShmStats, SYSTEM_ACTOR_PATH}; + use bytes::Bytes; + + #[tokio::test] + async fn shutdown_clears_host_shm_when_system_actor_cannot_stop() { + let system = ActorSystem::new(SystemConfig::standalone()).await.unwrap(); + let descriptor = system + .shm_manager() + .offer(Bytes::from_static(b"host-owned"), Duration::from_secs(30)); + assert_eq!( + &system.shm_manager().map(&descriptor).unwrap()[..], + b"host-owned" + ); + + let (_, local_id) = system + .registry + .remove_by_name(SYSTEM_ACTOR_PATH) + .expect("system actor must be registered"); + let (_, handle) = system + .registry + .remove_handle(&local_id) + .expect("system actor handle must be registered"); + handle.join_handle.abort(); + let _ = handle.join_handle.await; + + system.shutdown().await.unwrap(); + + assert_eq!(system.shm_manager().stats(), ShmStats::default()); + assert!(system.shm_manager().map(&descriptor).is_err()); + } +} diff --git a/crates/pulsing-actor/src/system/mod.rs b/crates/pulsing-actor/src/system/mod.rs index 8daeead95..67d9b3dbc 100644 --- a/crates/pulsing-actor/src/system/mod.rs +++ b/crates/pulsing-actor/src/system/mod.rs @@ -92,7 +92,7 @@ pub use registry::ActorRegistry; pub use traits::{ActorSystemCoreExt, ActorSystemOpsExt}; use crate::actor::{ - ActorId, ActorPath, ActorRef, ActorResolver, ActorSystemRef, Envelope, NodeId, StopReason, + ActorId, ActorPath, ActorRef, ActorResolver, ActorSystemRef, Message, NodeId, StopReason, }; use crate::cluster::{GossipBackend, HeadNodeBackend, NamingBackend}; use crate::error::{PulsingError, Result, RuntimeError}; @@ -100,13 +100,17 @@ use crate::performance_store::{ PerformanceSnapshot, PerformanceStore, DEFAULT_PERFORMANCE_HISTORY_CAPACITY, }; use crate::policies::{LoadBalancingPolicy, RoundRobinPolicy}; -use crate::system_actor::{BoxedActorFactory, SystemActor, SystemRef, SYSTEM_ACTOR_PATH}; +use crate::system_actor::{ + BoxedActorFactory, DefaultActorFactory, NodeLifecycle, NodeState, ShmManager, SystemActor, + SystemHost, SystemMessage, SystemRef, SystemResponse, SYSTEM_ACTOR_PATH, +}; use crate::transport::Http2Transport; use dashmap::DashMap; use handler::SystemMessageHandler; +use std::future::Future; use std::net::SocketAddr; use std::sync::{Arc, OnceLock}; -use tokio::sync::mpsc; +use std::time::Duration; use tokio::sync::RwLock; use tokio_util::sync::CancellationToken; @@ -152,9 +156,17 @@ pub struct ActorSystem { /// Ring buffer of recent `GetMetrics` snapshots (for local analysis / SQL adapters). pub(crate) performance_store: Arc, + + /// Node-level shared-memory control plane used by future same-host tensor backends. + pub(crate) shm_manager: Arc, + + /// Authoritative node control-plane lifecycle. + pub(crate) node_lifecycle: Arc, } impl ActorSystem { + const BOOTSTRAP_CLEANUP_TIMEOUT: Duration = Duration::from_secs(5); + /// Create a builder for configuring ActorSystem /// /// # Example @@ -168,6 +180,25 @@ impl ActorSystem { /// Create a new actor system pub async fn new(config: SystemConfig) -> Result> { + Self::new_inner(config, None).await + } + + /// Create an actor system with a custom actors-system service factory. + /// + /// The factory is installed while `system/core` is bootstrapped, before + /// the node becomes usable. This is the supported replacement for trying + /// to replace an already-running SystemActor after [`Self::new`]. + pub async fn new_with_system_actor_factory( + config: SystemConfig, + factory: BoxedActorFactory, + ) -> Result> { + Self::new_inner(config, Some(factory)).await + } + + async fn new_inner( + config: SystemConfig, + system_actor_factory: Option, + ) -> Result> { let cancel_token = CancellationToken::new(); let node_id = NodeId::generate(); let registry = Arc::new(ActorRegistry::new()); @@ -181,6 +212,8 @@ impl ActorSystem { > = Arc::new(OnceLock::new()); let performance_store = Arc::new(PerformanceStore::new(DEFAULT_PERFORMANCE_HISTORY_CAPACITY)); + let shm_manager = Arc::new(ShmManager::new()); + let node_lifecycle = Arc::new(NodeLifecycle::new()); // Create message handler (needs registry and cluster reference) let handler = SystemMessageHandler::new( @@ -234,23 +267,33 @@ impl ActorSystem { // Start backend backend.start(cancel_token.clone()); - // Join cluster if seed nodes provided (only for gossip mode) - if !config.seed_nodes.is_empty() && config.head_addr.is_none() && !config.is_head_node { - backend.join(config.seed_nodes).await?; + // Join cluster if seed nodes provided (only for gossip mode). + let join_result = if !config.seed_nodes.is_empty() + && config.head_addr.is_none() + && !config.is_head_node + { + backend.join(config.seed_nodes).await } else if config.head_addr.is_some() || config.is_head_node { - // For head node mode, join is handled internally - backend.join(Vec::new()).await?; + // For head node mode, join is handled internally. + backend.join(Vec::new()).await + } else { + Ok(()) + }; + if let Err(error) = join_result { + Self::cleanup_failed_bootstrap( + &node_lifecycle, + &shm_manager, + &cancel_token, + backend.leave(), + Self::BOOTSTRAP_CLEANUP_TIMEOUT, + ) + .await; + return Err(error); } crate::actor_store::init_actor_memtable(); crate::metrics_store::init_metrics_memtable(); crate::members_store::init_members_memtable(); - crate::members_store::upsert_member( - node_id, - actual_addr, - crate::cluster::NodeStatus::Online, - 0, - ); let system = Arc::new(Self { node_id, @@ -264,72 +307,75 @@ impl ActorSystem { node_load: Arc::new(DashMap::new()), system_monitor, performance_store, + shm_manager, + node_lifecycle, }); - // Start SystemActor - system.start_system_actor().await?; + // SystemRoot is part of bootstrap, not a runtime replacement. This + // keeps factory/service dependencies fixed before the node is ready. + let root_result = if let Some(factory) = system_actor_factory { + system.start_system_actor_with_factory(factory).await + } else { + system.start_system_actor().await + }; + if let Err(error) = root_result { + Self::cleanup_failed_bootstrap( + &system.node_lifecycle, + &system.shm_manager, + &system.cancel_token, + backend.leave(), + Self::BOOTSTRAP_CLEANUP_TIMEOUT, + ) + .await; + return Err(error); + } + + crate::members_store::upsert_member( + node_id, + actual_addr, + crate::cluster::NodeStatus::Online, + 0, + ); Ok(system) } - /// Start SystemActor (internal, called during system creation) - async fn start_system_actor(self: &Arc) -> Result<()> { - // Create senders snapshot for SystemRef - let local_actor_senders: Arc>> = - Arc::new(DashMap::new()); - for entry in self.registry.iter_actors() { - // Find name for this actor (reverse lookup from actor_names) - if let Some(name_entry) = self - .registry - .actor_names - .iter() - .find(|e| *e.value() == *entry.key()) - { - local_actor_senders.insert(name_entry.key().clone(), entry.sender.clone()); + async fn cleanup_failed_bootstrap( + lifecycle: &NodeLifecycle, + shm_manager: &ShmManager, + cancel_token: &CancellationToken, + leave: F, + leave_timeout: Duration, + ) where + F: Future>, + { + let _ = lifecycle.transition(NodeState::Failed); + match tokio::time::timeout(leave_timeout, leave).await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + tracing::warn!(%error, "Failed to leave cluster during bootstrap cleanup"); + } + Err(_) => { + tracing::warn!( + timeout_ms = leave_timeout.as_millis() as u64, + "Timed out leaving cluster during bootstrap cleanup" + ); } } + cancel_token.cancel(); + shm_manager.clear(); + } - // Create named_actor_paths snapshot - let named_actor_paths: Arc> = Arc::new(DashMap::new()); - for entry in self.registry.iter_named_paths() { - named_actor_paths.insert(entry.key().clone(), entry.value().clone()); - } - - let system_ref = Arc::new(SystemRef { - node_id: self.node_id, - addr: self.addr, - local_actors: local_actor_senders, - named_actor_paths, - }); - - let metrics = Arc::new(crate::system_actor::SystemMetrics::new()); - let sa_registry = Arc::new(crate::system_actor::ActorRegistry::new()); - let system_actor = SystemActor::with_default_factory_shared( - system_ref, - sa_registry.clone(), - metrics.clone(), - self.performance_store.clone(), - ) - .with_system_registry(self.registry.clone()); - - // Spawn as named actor with path "system" (use new_system to bypass namespace check) - let system_path = ActorPath::new_system(SYSTEM_ACTOR_PATH)?; - self.spawning() - .path(system_path) - .spawn(system_actor) - .await?; - - let _ = self.system_monitor.set((metrics, sa_registry)); - - // Note: The local_actors_ref and actor_names_ref are used internally, - // SystemRef snapshot may become stale for new actors but that's acceptable - // since SystemActor doesn't need real-time actor list - - tracing::debug!(path = SYSTEM_ACTOR_PATH, "SystemActor started"); - Ok(()) + /// Start SystemActor (internal, called during system creation) + async fn start_system_actor(self: &Arc) -> Result<()> { + self.spawn_system_actor(Box::new(DefaultActorFactory)).await } - /// Start SystemActor with custom factory (for Python extension) + /// Start SystemActor with a custom factory during bootstrap. + /// + /// `ActorSystem::new` already starts `system/core`; callers that need a + /// custom factory must use [`Self::new_with_system_actor_factory`]. This + /// method remains public for compatibility with custom bootstrap code. pub async fn start_system_actor_with_factory( self: &Arc, factory: BoxedActorFactory, @@ -341,45 +387,78 @@ impl ActorSystem { ))); } - // Create SystemRef (snapshot of named paths) - let named_paths_snapshot: Arc> = Arc::new(DashMap::new()); - for entry in self.registry.iter_named_paths() { - named_paths_snapshot.insert(entry.key().clone(), entry.value().clone()); - } - let system_ref = Arc::new(SystemRef { - node_id: self.node_id, - addr: self.addr, - local_actors: Arc::new(DashMap::new()), // Will be updated - named_actor_paths: named_paths_snapshot, - }); + self.spawn_system_actor(factory).await + } + async fn spawn_system_actor(self: &Arc, factory: BoxedActorFactory) -> Result<()> { + let system_ref = self.system_ref_compat(); let metrics = Arc::new(crate::system_actor::SystemMetrics::new()); - let sa_registry = Arc::new(crate::system_actor::ActorRegistry::new()); - let system_actor = SystemActor::new_shared( - system_ref, - factory, - sa_registry.clone(), + let legacy_registry = Arc::new(crate::system_actor::ActorRegistry::new()); + let host = SystemHost::hosted( + &system_ref, + Arc::downgrade(self), metrics.clone(), self.performance_store.clone(), - ) - .with_system_registry(self.registry.clone()); + self.shm_manager.clone(), + self.node_lifecycle.clone(), + ); + let system_actor = + SystemActor::new_hosted(factory, legacy_registry.clone(), metrics.clone(), host); - // Spawn as named actor (use new_system to bypass namespace check) let system_path = ActorPath::new_system(SYSTEM_ACTOR_PATH)?; - self.spawning() + let actor_ref = self + .spawning() .path(system_path) + .defer_cluster_publication() .spawn(system_actor) .await?; - let _ = self.system_monitor.set((metrics, sa_registry)); - - tracing::debug!( - path = SYSTEM_ACTOR_PATH, - "SystemActor started with custom factory" - ); + if let Err(error) = self.await_system_actor_ready(&actor_ref).await { + let _ = self.node_lifecycle.transition(NodeState::Failed); + let _ = self.stop(SYSTEM_ACTOR_PATH).await; + return Err(error); + } + if let Some(cluster) = self.cluster.read().await.as_ref() { + let path = ActorPath::new_system(SYSTEM_ACTOR_PATH)?; + cluster.register_named_actor(path).await; + cluster.register_actor(*actor_ref.id()).await; + } + let _ = self.system_monitor.set((metrics, legacy_registry)); + tracing::debug!(path = SYSTEM_ACTOR_PATH, "SystemActor ready"); Ok(()) } + fn system_ref_compat(&self) -> Arc { + Arc::new(SystemRef::new(self.node_id, self.addr)) + } + + async fn await_system_actor_ready(&self, actor_ref: &ActorRef) -> Result<()> { + let data = serde_json::to_vec(&SystemMessage::Ping) + .map_err(|error| PulsingError::from(RuntimeError::Serialization(error.to_string())))?; + let request = Message::Single { + msg_type: "SystemMessage".to_string(), + data, + }; + let response = + tokio::time::timeout(std::time::Duration::from_secs(5), actor_ref.send(request)) + .await + .map_err(|_| { + PulsingError::from(RuntimeError::Other( + "SystemActor readiness timed out".to_string(), + )) + })??; + let response: SystemResponse = response.parse()?; + if matches!(response, SystemResponse::Pong { .. }) + && self.node_lifecycle.state() == NodeState::Ready + { + Ok(()) + } else { + Err(PulsingError::from(RuntimeError::Other(format!( + "SystemActor did not become ready: {response:?}" + )))) + } + } + /// Get SystemActor reference pub async fn system(&self) -> Result { self.resolve_named(&ActorPath::new_system(SYSTEM_ACTOR_PATH)?, None) @@ -419,6 +498,14 @@ impl ActorSystem { &self.performance_store } + /// Node-scoped shared-memory control plane. + /// + /// The initial backend is in-process and provides the stable region/lease + /// contract required before enabling cross-process mappings. + pub fn shm_manager(&self) -> &Arc { + &self.shm_manager + } + /// Keep [`SystemActor`] `GetMetrics` / list in sync with real spawns (excludes `system/core`). pub(crate) fn notify_monitor_actor_spawned( &self, @@ -512,3 +599,72 @@ impl ActorResolver for ActorSystem { self.resolve_named_direct(path, None).await } } + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + use std::sync::atomic::{AtomicBool, Ordering}; + + #[tokio::test] + async fn failed_bootstrap_cleanup_leaves_then_cancels_and_clears_host_resources() { + let lifecycle = NodeLifecycle::new(); + let shm_manager = ShmManager::new(); + let descriptor = + shm_manager.offer(Bytes::from_static(b"bootstrap"), Duration::from_secs(30)); + let cancel_token = CancellationToken::new(); + let leave_called = Arc::new(AtomicBool::new(false)); + let leave_called_in_future = leave_called.clone(); + let cancel_observed_by_leave = cancel_token.clone(); + + ActorSystem::cleanup_failed_bootstrap( + &lifecycle, + &shm_manager, + &cancel_token, + async move { + assert!(!cancel_observed_by_leave.is_cancelled()); + leave_called_in_future.store(true, Ordering::Release); + Ok(()) + }, + Duration::from_millis(100), + ) + .await; + + assert!(leave_called.load(Ordering::Acquire)); + assert!(cancel_token.is_cancelled()); + assert_eq!(lifecycle.state(), NodeState::Failed); + assert_eq!( + shm_manager.stats(), + crate::system_actor::ShmStats::default() + ); + assert!(shm_manager.map(&descriptor).is_err()); + } + + #[tokio::test] + async fn failed_bootstrap_cleanup_continues_when_leave_fails() { + let lifecycle = NodeLifecycle::new(); + let shm_manager = ShmManager::new(); + shm_manager.offer(Bytes::from_static(b"bootstrap"), Duration::from_secs(30)); + let cancel_token = CancellationToken::new(); + + ActorSystem::cleanup_failed_bootstrap( + &lifecycle, + &shm_manager, + &cancel_token, + async { + Err(PulsingError::from(RuntimeError::Other( + "injected leave failure".to_string(), + ))) + }, + Duration::from_millis(100), + ) + .await; + + assert!(cancel_token.is_cancelled()); + assert_eq!(lifecycle.state(), NodeState::Failed); + assert_eq!( + shm_manager.stats(), + crate::system_actor::ShmStats::default() + ); + } +} diff --git a/crates/pulsing-actor/src/system/spawn.rs b/crates/pulsing-actor/src/system/spawn.rs index 6f3913560..2f87480ed 100644 --- a/crates/pulsing-actor/src/system/spawn.rs +++ b/crates/pulsing-actor/src/system/spawn.rs @@ -28,6 +28,7 @@ impl ActorSystem { path: Option, factory: F, options: SpawnOptions, + publish_to_cluster: bool, ) -> Result where F: FnMut() -> Result + Send + 'static, @@ -65,6 +66,7 @@ impl ActorSystem { cancel_token: actor_cancel, stats: stats.clone(), metadata: metadata.clone(), + started_at: std::time::Instant::now(), named_path: path.clone(), actor_id, }; @@ -90,17 +92,19 @@ impl ActorSystem { .register_named_path(name.clone(), name.clone()); // Register with cluster if available - if let Some(ref path) = path { - if let Some(cluster) = self.cluster.read().await.as_ref() { - if metadata.is_empty() { - cluster.register_named_actor(path.clone()).await; - } else { - cluster - .register_named_actor_full(path.clone(), actor_id, metadata.clone()) - .await; + if publish_to_cluster { + if let Some(ref path) = path { + if let Some(cluster) = self.cluster.read().await.as_ref() { + if metadata.is_empty() { + cluster.register_named_actor(path.clone()).await; + } else { + cluster + .register_named_actor_full(path.clone(), actor_id, metadata.clone()) + .await; + } + // So refer(actor_id) / lookup_actor can resolve this actor on any node + cluster.register_actor(actor_id).await; } - // So refer(actor_id) / lookup_actor can resolve this actor on any node - cluster.register_actor(actor_id).await; } } } else { diff --git a/crates/pulsing-actor/src/system/traits.rs b/crates/pulsing-actor/src/system/traits.rs index 1f4c5ea24..1c7aec52e 100644 --- a/crates/pulsing-actor/src/system/traits.rs +++ b/crates/pulsing-actor/src/system/traits.rs @@ -64,6 +64,7 @@ pub struct SpawnBuilder<'a> { name: Option, name_error: Option, options: SpawnOptions, + publish_to_cluster: bool, } impl<'a> SpawnBuilder<'a> { @@ -74,6 +75,7 @@ impl<'a> SpawnBuilder<'a> { name: None, name_error: None, options: SpawnOptions::default(), + publish_to_cluster: true, } } @@ -128,6 +130,12 @@ impl<'a> SpawnBuilder<'a> { self } + /// Delay cluster publication until a bootstrap actor acknowledges readiness. + pub(crate) fn defer_cluster_publication(mut self) -> Self { + self.publish_to_cluster = false; + self + } + /// Spawn the actor /// /// Accepts any type that implements `IntoActor`, including: @@ -175,11 +183,25 @@ impl<'a> SpawnBuilder<'a> { match self.name { Some(path) => { // Named actor: resolvable by name - ActorSystem::spawn_internal(self.system, Some(path), factory, self.options).await + ActorSystem::spawn_internal( + self.system, + Some(path), + factory, + self.options, + self.publish_to_cluster, + ) + .await } None => { // Anonymous actor: not resolvable - ActorSystem::spawn_internal(self.system, None, factory, self.options).await + ActorSystem::spawn_internal( + self.system, + None, + factory, + self.options, + self.publish_to_cluster, + ) + .await } } } diff --git a/crates/pulsing-actor/src/system_actor/builtin.rs b/crates/pulsing-actor/src/system_actor/builtin.rs new file mode 100644 index 000000000..c5e567f78 --- /dev/null +++ b/crates/pulsing-actor/src/system_actor/builtin.rs @@ -0,0 +1,311 @@ +//! Built-in system-service registrations. + +use super::host::{ActorControl, SystemHost}; +use super::lifecycle::{NodeLifecycle, NodeState}; +use super::service::{ + service_error, ActorsCommand, MetricsCommand, OperationAccess, OperationManifest, + RuntimeCommand, ShmCommand, StatelessComponent, SystemCommand, SystemComponent, + SystemRequestHandler, SystemServiceExposure, SystemServiceKind, SystemServiceManifest, + SystemServiceRegistration, ACTORS_SERVICE, METRICS_SERVICE, RUNTIME_SERVICE, SHM_SERVICE, +}; +use super::{ActorFactory, ShmManager, SystemMetrics, SystemResponse}; +use crate::error::Result; +use crate::performance_store::{PerformanceSnapshot, PerformanceStore}; +use async_trait::async_trait; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; + +const RUNTIME_OPERATIONS: &[OperationManifest] = &[ + operation("node_info", OperationAccess::Read, true), + operation("health", OperationAccess::Read, true), + operation("ping", OperationAccess::Read, true), +]; +const ACTOR_OPERATIONS: &[OperationManifest] = &[ + operation("list", OperationAccess::Read, true), + operation("get", OperationAccess::Read, true), + operation("create", OperationAccess::Admin, false), + operation("stop", OperationAccess::Operate, false), + operation("extension", OperationAccess::Admin, false), +]; +const METRICS_OPERATIONS: &[OperationManifest] = + &[operation("snapshot", OperationAccess::Read, true)]; +const SHM_OPERATIONS: &[OperationManifest] = &[operation("stats", OperationAccess::Read, true)]; + +const fn operation( + name: &'static str, + access: OperationAccess, + allowed_while_draining: bool, +) -> OperationManifest { + OperationManifest { + name, + access, + allowed_while_draining, + } +} + +pub(crate) fn registrations( + host: &SystemHost, + factory: Arc, +) -> Vec { + let stateless: Arc = Arc::new(StatelessComponent); + let shm = Arc::new(ShmService { + manager: host.shm_manager().clone(), + }); + vec![ + SystemServiceRegistration { + manifest: core_manifest( + RUNTIME_SERVICE, + SystemServiceExposure::AuthenticatedRemote, + RUNTIME_OPERATIONS, + ), + component: stateless.clone(), + handler: Arc::new(RuntimeService { + node_id: host.node_id(), + addr: host.addr(), + start_time: host.start_time(), + actors: host.actors(), + lifecycle: host.lifecycle(), + }), + }, + SystemServiceRegistration { + manifest: core_manifest( + ACTORS_SERVICE, + SystemServiceExposure::AuthenticatedRemote, + ACTOR_OPERATIONS, + ), + component: stateless.clone(), + handler: Arc::new(ActorsService { + actors: host.actors(), + factory, + }), + }, + SystemServiceRegistration { + manifest: core_manifest( + METRICS_SERVICE, + SystemServiceExposure::AuthenticatedRemote, + METRICS_OPERATIONS, + ), + component: stateless, + handler: Arc::new(MetricsService { + node_id: host.node_id(), + start_time: host.start_time(), + actors: host.actors(), + metrics: host.metrics(), + performance_store: host.performance_store(), + }), + }, + SystemServiceRegistration { + manifest: core_manifest( + SHM_SERVICE, + SystemServiceExposure::LocalOnly, + SHM_OPERATIONS, + ), + component: shm.clone(), + handler: shm, + }, + ] +} + +fn core_manifest( + id: super::service::SystemServiceId, + exposure: SystemServiceExposure, + operations: &'static [OperationManifest], +) -> SystemServiceManifest { + SystemServiceManifest { + id, + kind: SystemServiceKind::Core, + exposure, + operations, + } +} + +struct RuntimeService { + node_id: crate::actor::NodeId, + addr: SocketAddr, + start_time: Instant, + actors: Arc, + lifecycle: Arc, +} + +#[async_trait] +impl SystemRequestHandler for RuntimeService { + async fn handle(&self, command: SystemCommand) -> Result { + match command { + SystemCommand::Runtime(RuntimeCommand::NodeInfo) => Ok(SystemResponse::NodeInfo { + node_id: self.node_id.0, + addr: self.addr.to_string(), + uptime_secs: self.start_time.elapsed().as_secs(), + }), + SystemCommand::Runtime(RuntimeCommand::Health) => { + let state = self.lifecycle.state(); + Ok(SystemResponse::Health { + status: if state == NodeState::Ready { + "healthy".to_string() + } else { + state.as_str().to_string() + }, + actors_count: self.actors.count(), + uptime_secs: self.start_time.elapsed().as_secs(), + }) + } + SystemCommand::Runtime(RuntimeCommand::Ping) => Ok(SystemResponse::Pong { + node_id: self.node_id.0, + timestamp: SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64, + }), + _ => Err(service_error("runtime service received an invalid command")), + } + } +} + +struct ActorsService { + actors: Arc, + factory: Arc, +} + +#[async_trait] +impl SystemRequestHandler for ActorsService { + async fn handle(&self, command: SystemCommand) -> Result { + match command { + SystemCommand::Actors(ActorsCommand::List) => Ok(SystemResponse::ActorList { + actors: self.actors.list(), + }), + SystemCommand::Actors(ActorsCommand::Get { name }) => { + Ok(match self.actors.get(&name) { + Some(info) => SystemResponse::ActorInfo(info), + None => SystemResponse::Error { + message: format!("Actor not found: {name}"), + }, + }) + } + SystemCommand::Actors(ActorsCommand::Create { + actor_type, + name, + params, + public, + }) => { + let _ = (actor_type, name, params, public); + Ok(SystemResponse::Error { + message: "CreateActor not supported in pure Rust mode. Use Python extension." + .to_string(), + }) + } + SystemCommand::Actors(ActorsCommand::Stop { name }) => { + if self.actors.stop(&name).await? { + Ok(SystemResponse::Ok) + } else { + Ok(SystemResponse::Error { + message: format!("Actor not found: {name}"), + }) + } + } + SystemCommand::Actors(ActorsCommand::Extension { handler, payload }) => { + Ok(self.factory.handle_extension(&handler, payload).await) + } + _ => Err(service_error("actors service received an invalid command")), + } + } +} + +struct MetricsService { + node_id: crate::actor::NodeId, + start_time: Instant, + actors: Arc, + metrics: Arc, + performance_store: Arc, +} + +#[async_trait] +impl SystemRequestHandler for MetricsService { + async fn handle(&self, command: SystemCommand) -> Result { + match command { + SystemCommand::Metrics(MetricsCommand::Snapshot) => { + let actors_count = self.actors.count(); + let messages_total = self.actors.total_messages(); + let actors_created = self.metrics.actors_created(); + let actors_stopped = self.metrics.actors_stopped(); + let uptime_secs = self.start_time.elapsed().as_secs(); + let ts_unix_micros = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_micros() as u64) + .unwrap_or(0); + let node_id = self.node_id.to_string(); + + self.performance_store.record(PerformanceSnapshot { + ts_unix_micros, + node_id: node_id.clone(), + actors_count: actors_count as u64, + messages_total, + actors_created, + actors_stopped, + uptime_secs, + }); + crate::metrics_store::write_metrics_snapshot( + ts_unix_micros, + &node_id, + actors_count as u64, + messages_total, + actors_created, + actors_stopped, + uptime_secs, + ); + + Ok(SystemResponse::Metrics { + actors_count, + messages_total, + actors_created, + actors_stopped, + uptime_secs, + }) + } + _ => Err(service_error("metrics service received an invalid command")), + } + } +} + +struct ShmService { + manager: Arc, +} + +#[async_trait] +impl SystemComponent for ShmService { + async fn stop(&self) -> Result<()> { + self.manager.clear(); + Ok(()) + } +} + +#[async_trait] +impl SystemRequestHandler for ShmService { + async fn handle(&self, command: SystemCommand) -> Result { + match command { + SystemCommand::Shm(ShmCommand::Stats) => { + let stats = self.manager.stats(); + Ok(SystemResponse::ShmStats { + backend: self.manager.backend().as_str().to_string(), + regions: stats.regions, + published_regions: stats.published_regions, + active_leases: stats.active_leases, + bytes: stats.bytes, + }) + } + _ => Err(service_error("shm service received an invalid command")), + } + } +} + +pub(crate) fn prometheus_metrics(host: &SystemHost) -> crate::metrics::SystemMetrics { + let actors = host.actors(); + let metrics = host.metrics(); + crate::metrics::SystemMetrics { + node_id: host.node_id().0, + actors_count: actors.count(), + messages_total: actors.total_messages(), + actors_created: metrics.actors_created(), + actors_stopped: metrics.actors_stopped(), + cluster_members: std::collections::HashMap::new(), + } +} diff --git a/crates/pulsing-actor/src/system_actor/host.rs b/crates/pulsing-actor/src/system_actor/host.rs new file mode 100644 index 000000000..0733f1fa5 --- /dev/null +++ b/crates/pulsing-actor/src/system_actor/host.rs @@ -0,0 +1,297 @@ +//! Narrow host capabilities granted to system services. + +use super::lifecycle::{NodeLifecycle, NodeState}; +use super::{ActorInfo, ActorRegistry, ShmManager, SystemMetrics, SystemRef, SYSTEM_ACTOR_PATH}; +use crate::error::{PulsingError, Result, RuntimeError}; +use crate::performance_store::PerformanceStore; +use crate::system::ActorSystem; +use async_trait::async_trait; +use std::sync::atomic::Ordering; +use std::sync::{Arc, Weak}; +use std::time::Instant; + +/// Actor operations available to the node control plane. +#[async_trait] +pub(crate) trait ActorControl: Send + Sync { + fn list(&self) -> Vec; + fn get(&self, name: &str) -> Option; + fn count(&self) -> usize; + fn total_messages(&self) -> u64; + async fn stop(&self, name: &str) -> Result; +} + +/// Explicit capability view passed to system services. +/// +/// It contains no transport or cluster implementation details. Services can +/// only perform the node operations declared here. +pub(crate) struct SystemHost { + node_id: crate::actor::NodeId, + addr: std::net::SocketAddr, + start_time: Instant, + actors: Arc, + metrics: Arc, + performance_store: Arc, + shm_manager: Arc, + lifecycle: Arc, +} + +impl SystemHost { + pub(crate) fn standalone( + system_ref: &SystemRef, + registry: Arc, + metrics: Arc, + performance_store: Arc, + ) -> Self { + let actors = Arc::new(LegacyActorControl { + registry, + metrics: metrics.clone(), + }); + Self { + node_id: system_ref.node_id, + addr: system_ref.addr, + start_time: Instant::now(), + actors, + metrics, + performance_store, + shm_manager: Arc::new(ShmManager::new()), + lifecycle: Arc::new(NodeLifecycle::new()), + } + } + + pub(crate) fn hosted( + system_ref: &SystemRef, + system: Weak, + metrics: Arc, + performance_store: Arc, + shm_manager: Arc, + lifecycle: Arc, + ) -> Self { + Self { + node_id: system_ref.node_id, + addr: system_ref.addr, + start_time: Instant::now(), + actors: Arc::new(HostedActorControl { system }), + metrics, + performance_store, + shm_manager, + lifecycle, + } + } + + pub(crate) fn with_actor_control(mut self, actors: Arc) -> Self { + self.actors = actors; + self + } + + pub(crate) fn with_shm_manager(mut self, manager: Arc) -> Self { + self.shm_manager = manager; + self + } + + pub(crate) fn node_state(&self) -> NodeState { + self.lifecycle.state() + } + + pub(crate) fn node_id(&self) -> crate::actor::NodeId { + self.node_id + } + + pub(crate) fn addr(&self) -> std::net::SocketAddr { + self.addr + } + + pub(crate) fn start_time(&self) -> Instant { + self.start_time + } + + pub(crate) fn actors(&self) -> Arc { + self.actors.clone() + } + + pub(crate) fn metrics(&self) -> Arc { + self.metrics.clone() + } + + pub(crate) fn performance_store(&self) -> Arc { + self.performance_store.clone() + } + + pub(crate) fn shm_manager(&self) -> &Arc { + &self.shm_manager + } + + pub(crate) fn lifecycle(&self) -> Arc { + self.lifecycle.clone() + } +} + +struct LegacyActorControl { + registry: Arc, + metrics: Arc, +} + +#[async_trait] +impl ActorControl for LegacyActorControl { + fn list(&self) -> Vec { + self.registry.list_all() + } + + fn get(&self, name: &str) -> Option { + self.registry.get_info(name) + } + + fn count(&self) -> usize { + self.registry.count() + } + + fn total_messages(&self) -> u64 { + self.metrics.messages_total() + } + + async fn stop(&self, name: &str) -> Result { + Ok(self.registry.unregister(name).is_some()) + } +} + +pub(crate) struct RegistryActorControl { + registry: Arc, +} + +impl RegistryActorControl { + pub(crate) fn new(registry: Arc) -> Self { + Self { registry } + } +} + +#[async_trait] +impl ActorControl for RegistryActorControl { + fn list(&self) -> Vec { + actor_infos(&self.registry) + } + + fn get(&self, name: &str) -> Option { + actor_info(&self.registry, name) + } + + fn count(&self) -> usize { + actor_count(&self.registry) + } + + fn total_messages(&self) -> u64 { + total_messages(&self.registry) + } + + async fn stop(&self, _name: &str) -> Result { + Err(host_error( + "actor stop requires an ActorSystem-owned lifecycle capability", + )) + } +} + +struct HostedActorControl { + system: Weak, +} + +#[async_trait] +impl ActorControl for HostedActorControl { + fn list(&self) -> Vec { + self.system + .upgrade() + .map(|system| actor_infos(&system.registry)) + .unwrap_or_default() + } + + fn get(&self, name: &str) -> Option { + self.system + .upgrade() + .and_then(|system| actor_info(&system.registry, name)) + } + + fn count(&self) -> usize { + self.system + .upgrade() + .map(|system| actor_count(&system.registry)) + .unwrap_or_default() + } + + fn total_messages(&self) -> u64 { + self.system + .upgrade() + .map(|system| total_messages(&system.registry)) + .unwrap_or_default() + } + + async fn stop(&self, name: &str) -> Result { + if name == SYSTEM_ACTOR_PATH { + return Ok(false); + } + let system = self + .system + .upgrade() + .ok_or_else(|| host_error("ActorSystem is no longer available"))?; + let resolved_name = if system.registry.has_name(name) { + Some(name.to_string()) + } else if !name.contains('/') { + let prefixed = format!("actors/{name}"); + system.registry.has_name(&prefixed).then_some(prefixed) + } else { + None + }; + let Some(resolved_name) = resolved_name else { + return Ok(false); + }; + system.stop(&resolved_name).await?; + Ok(true) + } +} + +fn actor_infos(registry: &crate::system::registry::ActorRegistry) -> Vec { + registry + .actor_names + .iter() + .filter_map(|entry| actor_info(registry, entry.key())) + .collect() +} + +fn actor_info(registry: &crate::system::registry::ActorRegistry, name: &str) -> Option { + if name == SYSTEM_ACTOR_PATH { + return None; + } + let actor_id = registry.get_actor_id(name)?; + let handle = registry.get_handle(&actor_id)?; + let actor_type = handle + .metadata + .get("class") + .or_else(|| handle.metadata.get("type")) + .cloned() + .unwrap_or_else(|| "Actor".to_string()); + Some(ActorInfo { + name: name.to_string(), + actor_id: handle.actor_id.0, + actor_type, + uptime_secs: handle.started_at.elapsed().as_secs(), + metadata: handle.metadata.clone(), + }) +} + +fn actor_count(registry: &crate::system::registry::ActorRegistry) -> usize { + registry + .actor_names + .iter() + .filter(|entry| entry.key().as_str() != SYSTEM_ACTOR_PATH) + .count() +} + +fn total_messages(registry: &crate::system::registry::ActorRegistry) -> u64 { + registry + .iter_actors() + .map(|entry| entry.value().stats.message_count.load(Ordering::Relaxed)) + .sum() +} + +fn host_error(message: impl Into) -> PulsingError { + PulsingError::from(RuntimeError::Other(format!( + "System host capability error: {}", + message.into() + ))) +} diff --git a/crates/pulsing-actor/src/system_actor/lifecycle.rs b/crates/pulsing-actor/src/system_actor/lifecycle.rs new file mode 100644 index 000000000..76c3f2527 --- /dev/null +++ b/crates/pulsing-actor/src/system_actor/lifecycle.rs @@ -0,0 +1,137 @@ +//! Node control-plane lifecycle. + +use crate::error::{PulsingError, Result, RuntimeError}; +use std::sync::atomic::{AtomicU8, Ordering}; + +/// Observable state of the node control plane. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[repr(u8)] +pub(crate) enum NodeState { + Booting = 0, + Starting = 1, + Ready = 2, + Draining = 3, + Failed = 4, + Stopped = 5, +} + +impl NodeState { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Booting => "booting", + Self::Starting => "starting", + Self::Ready => "ready", + Self::Draining => "draining", + Self::Failed => "failed", + Self::Stopped => "stopped", + } + } + + const fn can_transition_to(self, next: Self) -> bool { + matches!( + (self, next), + ( + Self::Booting, + Self::Starting | Self::Draining | Self::Failed + ) | (Self::Starting, Self::Ready | Self::Draining | Self::Failed) + | (Self::Ready, Self::Draining | Self::Failed) + | (Self::Draining, Self::Stopped | Self::Failed) + | (Self::Failed, Self::Draining | Self::Stopped) + ) + } +} + +/// Shared, validated state machine for node bootstrap and shutdown. +pub(crate) struct NodeLifecycle { + state: AtomicU8, +} + +impl NodeLifecycle { + pub(crate) fn new() -> Self { + Self { + state: AtomicU8::new(NodeState::Booting as u8), + } + } + + pub(crate) fn state(&self) -> NodeState { + match self.state.load(Ordering::Acquire) { + 0 => NodeState::Booting, + 1 => NodeState::Starting, + 2 => NodeState::Ready, + 3 => NodeState::Draining, + 4 => NodeState::Failed, + 5 => NodeState::Stopped, + _ => unreachable!("node lifecycle contains an invalid state"), + } + } + + pub(crate) fn transition(&self, next: NodeState) -> Result<()> { + loop { + let current = self.state(); + if current == next { + return Ok(()); + } + if !current.can_transition_to(next) { + return Err(lifecycle_error(format!( + "invalid node lifecycle transition: {current:?} -> {next:?}" + ))); + } + if self + .state + .compare_exchange( + current as u8, + next as u8, + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_ok() + { + return Ok(()); + } + } + } + + /// Enter draining unless the node has already reached a terminal path. + pub(crate) fn begin_draining(&self) -> Result<()> { + match self.state() { + NodeState::Draining | NodeState::Stopped => Ok(()), + NodeState::Failed => self.transition(NodeState::Draining), + _ => self.transition(NodeState::Draining), + } + } +} + +impl Default for NodeLifecycle { + fn default() -> Self { + Self::new() + } +} + +fn lifecycle_error(message: impl Into) -> PulsingError { + PulsingError::from(RuntimeError::Other(format!( + "System lifecycle error: {}", + message.into() + ))) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn lifecycle_accepts_boot_and_shutdown_path() { + let lifecycle = NodeLifecycle::new(); + lifecycle.transition(NodeState::Starting).unwrap(); + lifecycle.transition(NodeState::Ready).unwrap(); + lifecycle.begin_draining().unwrap(); + lifecycle.transition(NodeState::Stopped).unwrap(); + assert_eq!(lifecycle.state(), NodeState::Stopped); + } + + #[test] + fn lifecycle_rejects_skipping_readiness() { + let lifecycle = NodeLifecycle::new(); + assert!(lifecycle.transition(NodeState::Ready).is_err()); + assert_eq!(lifecycle.state(), NodeState::Booting); + } +} diff --git a/crates/pulsing-actor/src/system_actor/messages.rs b/crates/pulsing-actor/src/system_actor/messages.rs index 450719fd3..e3be2a21a 100644 --- a/crates/pulsing-actor/src/system_actor/messages.rs +++ b/crates/pulsing-actor/src/system_actor/messages.rs @@ -63,6 +63,9 @@ pub enum SystemMessage { /// Get node info GetNodeInfo, + /// Get node-level shared-memory control-plane statistics. + GetShmStats, + /// Health check HealthCheck, @@ -145,6 +148,17 @@ pub enum SystemResponse { uptime_secs: u64, }, + /// Shared-memory control-plane status. `in_process` means that region + /// lifetime semantics are available but cross-process mapping is not yet + /// advertised by this node. + ShmStats { + backend: String, + regions: usize, + published_regions: usize, + active_leases: usize, + bytes: usize, + }, + /// Health status Health { /// Status @@ -171,6 +185,7 @@ pub struct ActorInfo { /// Actor name (also used as path for resolution) pub name: String, /// Actor ID (full UUID) + #[serde(with = "u128_as_string")] pub actor_id: u128, /// Actor type pub actor_type: String, diff --git a/crates/pulsing-actor/src/system_actor/mod.rs b/crates/pulsing-actor/src/system_actor/mod.rs index 3c5b1cb2a..b36ea00bc 100644 --- a/crates/pulsing-actor/src/system_actor/mod.rs +++ b/crates/pulsing-actor/src/system_actor/mod.rs @@ -18,21 +18,29 @@ //! let remote_sys = system.resolve_named(&ActorPath::new("system/core")?, Some(&node_id)).await?; //! ``` +mod builtin; mod factory; +mod host; +mod lifecycle; mod messages; +mod service; +mod shm; pub use factory::{ActorFactory, BoxedActorFactory, DefaultActorFactory}; +pub(crate) use host::SystemHost; +pub(crate) use lifecycle::{NodeLifecycle, NodeState}; pub use messages::{ActorInfo, ActorStatusInfo, SystemMessage, SystemResponse}; +pub use shm::{ShmBackend, ShmManager, ShmRegionDescriptor, ShmStats}; use crate::actor::{Actor, ActorContext, ActorId, Message}; use crate::error::{PulsingError, Result, RuntimeError}; use crate::metrics::SystemMetrics as PrometheusSystemMetrics; -use crate::performance_store::{PerformanceSnapshot, PerformanceStore}; +use crate::performance_store::PerformanceStore; use dashmap::DashMap; use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; -use std::time::{Instant, SystemTime, UNIX_EPOCH}; +use std::time::Instant; use tokio::sync::mpsc; /// Named path for SystemActor (system/core satisfies namespace/name format requirement) @@ -175,34 +183,54 @@ pub struct SystemRef { pub node_id: crate::actor::NodeId, /// Local address pub addr: std::net::SocketAddr, - /// Local actor senders + /// Legacy spawn-time projection. System services use host capabilities and + /// the authoritative ActorSystem registry instead. + #[deprecated( + since = "0.1.3", + note = "use ActorSystem actor APIs; ActorSystem bootstrap no longer populates this snapshot" + )] pub local_actors: Arc>>, - /// Named actor path mappings + /// Legacy spawn-time projection. System services use host capabilities and + /// the authoritative ActorSystem registry instead. + #[deprecated( + since = "0.1.3", + note = "use ActorSystem resolution APIs; ActorSystem bootstrap no longer populates this snapshot" + )] pub named_actor_paths: Arc>, } +impl SystemRef { + /// Construct the compatibility reference required by manually hosted + /// SystemActors. Actor and path snapshots are intentionally left empty. + #[allow(deprecated)] + pub fn new(node_id: crate::actor::NodeId, addr: std::net::SocketAddr) -> Self { + Self { + node_id, + addr, + local_actors: Arc::new(DashMap::new()), + named_actor_paths: Arc::new(DashMap::new()), + } + } +} + /// SystemActor - Built-in system actor for each ActorSystem pub struct SystemActor { - /// Actor registry + /// Compatibility projection retained for public Rust/Python APIs. registry: Arc, - /// Actor factory - factory: BoxedActorFactory, - - /// System metrics + /// Compatibility counters shared with ActorSystem monitoring. metrics: Arc, - /// System reference - system_ref: Arc, + /// Bootstrap extension factory retained so compatibility builders can + /// rebuild registrations before the actor starts. + factory: Arc, - /// Start time - start_time: Instant, + /// Narrow view of ActorSystem-owned resources. This is the only source used + /// by built-in service handlers. + host: host::SystemHost, - /// Ring buffer of recent `GetMetrics` samples - performance_store: Arc, - - /// System-level registry with per-actor stats (for accurate messages_total) - system_registry: Option>, + /// Governed service registry owned by this SystemRoot. + services: Arc, } impl SystemActor { @@ -234,15 +262,13 @@ impl SystemActor { metrics: Arc, performance_store: Arc, ) -> Self { - Self { + Self::new_shared( + system_ref, + Box::new(DefaultActorFactory), registry, - factory: Box::new(DefaultActorFactory), metrics, - system_ref, - start_time: Instant::now(), performance_store, - system_registry: None, - } + ) } /// Shared registry + metrics with a custom factory. @@ -253,26 +279,86 @@ impl SystemActor { metrics: Arc, performance_store: Arc, ) -> Self { + let host = host::SystemHost::standalone( + &system_ref, + registry.clone(), + metrics.clone(), + performance_store, + ); + Self::from_host(factory, registry, metrics, host) + } + + pub(crate) fn new_hosted( + factory: BoxedActorFactory, + registry: Arc, + metrics: Arc, + host: SystemHost, + ) -> Self { + Self::from_host(factory, registry, metrics, host) + } + + fn from_host( + factory: BoxedActorFactory, + registry: Arc, + metrics: Arc, + host: host::SystemHost, + ) -> Self { + let factory: Arc = Arc::from(factory); + let services = Self::build_services(&host, factory.clone()); Self { registry, - factory, metrics, - system_ref, - start_time: Instant::now(), - performance_store, - system_registry: None, + factory, + host, + services, } } - /// Attach the system-level registry for accurate global message counting. + fn build_services( + host: &host::SystemHost, + factory: Arc, + ) -> Arc { + let services = Arc::new(service::SystemServiceRegistry::new()); + for registration in builtin::registrations(host, factory) { + services + .register(registration) + .expect("built-in system service manifests must be valid and unique"); + } + services + } + + fn rebuild_services(&mut self) { + self.services = Self::build_services(&self.host, self.factory.clone()); + } + + /// Attach an authoritative host registry for read-only actor control. + /// + /// ActorSystem bootstrap uses [`Self::new_hosted`] so stop operations also + /// receive the full lifecycle capability. This builder remains for source + /// compatibility with manually constructed SystemActors. pub fn with_system_registry( mut self, reg: Arc, ) -> Self { - self.system_registry = Some(reg); + self.host = self + .host + .with_actor_control(Arc::new(host::RegistryActorControl::new(reg))); + self.rebuild_services(); self } + /// Use the ActorSystem-owned shared-memory control plane. + pub fn with_shm_manager(mut self, manager: Arc) -> Self { + self.host = self.host.with_shm_manager(manager); + self.rebuild_services(); + self + } + + /// Shared-memory control plane associated with this system actor. + pub fn shm_manager(&self) -> &Arc { + self.host.shm_manager() + } + /// Get registry (for Python bindings) pub fn registry(&self) -> &Arc { &self.registry @@ -283,95 +369,6 @@ impl SystemActor { &self.metrics } - /// Handle ListActors request - fn handle_list_actors(&self) -> SystemResponse { - SystemResponse::ActorList { - actors: self.registry.list_all(), - } - } - - /// Handle GetActor request - fn handle_get_actor(&self, name: &str) -> SystemResponse { - match self.registry.get_info(name) { - Some(info) => SystemResponse::ActorInfo(info), - None => SystemResponse::Error { - message: format!("Actor not found: {}", name), - }, - } - } - - /// Handle GetMetrics request - fn handle_get_metrics(&self) -> SystemResponse { - let actors_count = self.registry.count(); - let messages_total = self.total_messages(); - let actors_created = self.metrics.actors_created(); - let actors_stopped = self.metrics.actors_stopped(); - let uptime_secs = self.start_time.elapsed().as_secs(); - - let ts_unix_micros = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_micros() as u64) - .unwrap_or(0); - let node_id_str = self.system_ref.node_id.to_string(); - - self.performance_store.record(PerformanceSnapshot { - ts_unix_micros, - node_id: node_id_str.clone(), - actors_count: actors_count as u64, - messages_total, - actors_created, - actors_stopped, - uptime_secs, - }); - - crate::metrics_store::write_metrics_snapshot( - ts_unix_micros, - &node_id_str, - actors_count as u64, - messages_total, - actors_created, - actors_stopped, - uptime_secs, - ); - - SystemResponse::Metrics { - actors_count, - messages_total, - actors_created, - actors_stopped, - uptime_secs, - } - } - - /// Handle GetNodeInfo request - fn handle_get_node_info(&self) -> SystemResponse { - SystemResponse::NodeInfo { - node_id: self.system_ref.node_id.0, - addr: self.system_ref.addr.to_string(), - uptime_secs: self.start_time.elapsed().as_secs(), - } - } - - /// Handle HealthCheck request - fn handle_health_check(&self) -> SystemResponse { - SystemResponse::Health { - status: "healthy".to_string(), - actors_count: self.registry.count(), - uptime_secs: self.start_time.elapsed().as_secs(), - } - } - - /// Handle Ping request - fn handle_ping(&self) -> SystemResponse { - SystemResponse::Pong { - node_id: self.system_ref.node_id.0, - timestamp: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as u64, - } - } - /// Register a created actor (called externally) pub fn register_actor(&self, name: &str, actor_id: ActorId, actor_type: &str) { self.registry.register(name, actor_id, actor_type); @@ -387,28 +384,7 @@ impl SystemActor { /// Get Prometheus-compatible system metrics pub fn get_prometheus_metrics(&self) -> PrometheusSystemMetrics { - PrometheusSystemMetrics { - node_id: self.system_ref.node_id.0, - actors_count: self.registry.count(), - messages_total: self.total_messages(), - actors_created: self.metrics.actors_created(), - actors_stopped: self.metrics.actors_stopped(), - cluster_members: HashMap::new(), // Will be filled by caller with cluster info - } - } - - /// Sum message counts from all actors in the system-level registry, - /// falling back to `SystemMetrics::messages_total` (system-actor only) if unavailable. - fn total_messages(&self) -> u64 { - if let Some(reg) = &self.system_registry { - let mut total = 0u64; - for entry in reg.iter_actors() { - total += entry.value().stats.message_count.load(Ordering::Relaxed); - } - total - } else { - self.metrics.messages_total() - } + builtin::prometheus_metrics(&self.host) } /// Generate JSON error response @@ -432,10 +408,28 @@ impl Actor for SystemActor { meta.insert("type".to_string(), "SystemActor".to_string()); meta.insert("builtin".to_string(), "true".to_string()); meta.insert("path".to_string(), SYSTEM_ACTOR_PATH.to_string()); + meta.insert("control_plane".to_string(), "system-root".to_string()); + // Actor metadata is captured at spawn time, so advertise immutable + // service identities here. Runtime readiness belongs to runtime@1. + let service_ids = self + .services + .statuses() + .into_iter() + .map(|(manifest, _)| format!("{}@{}", manifest.id.namespace, manifest.id.major)) + .collect::>() + .join(","); + meta.insert("services".to_string(), service_ids); meta } async fn on_start(&mut self, ctx: &mut ActorContext) -> Result<()> { + let lifecycle = self.host.lifecycle(); + lifecycle.transition(NodeState::Starting)?; + if let Err(error) = self.services.start_all().await { + let _ = lifecycle.transition(NodeState::Failed); + return Err(error); + } + lifecycle.transition(NodeState::Ready)?; tracing::info!( actor_id = ?ctx.id(), path = SYSTEM_ACTOR_PATH, @@ -445,6 +439,12 @@ impl Actor for SystemActor { } async fn on_stop(&mut self, ctx: &mut ActorContext) -> Result<()> { + let lifecycle = self.host.lifecycle(); + lifecycle.begin_draining()?; + if let Err(error) = self.services.stop_all().await { + let _ = lifecycle.transition(NodeState::Failed); + return Err(error); + } tracing::info!( actor_id = ?ctx.id(), path = SYSTEM_ACTOR_PATH, @@ -475,41 +475,18 @@ impl Actor for SystemActor { } }; - let response = match sys_msg { - SystemMessage::ListActors => self.handle_list_actors(), - - SystemMessage::GetActor { name } => self.handle_get_actor(&name), - - SystemMessage::GetMetrics => self.handle_get_metrics(), - - SystemMessage::GetNodeInfo => self.handle_get_node_info(), - - SystemMessage::HealthCheck => self.handle_health_check(), - - SystemMessage::Ping => self.handle_ping(), - - // Actor creation handled by Python layer (via extension) - SystemMessage::CreateActor { .. } => SystemResponse::Error { - message: "CreateActor not supported in pure Rust mode. Use Python extension." - .to_string(), + let response = match self + .services + .dispatch( + service::SystemCommand::from_legacy(sys_msg), + self.host.node_state(), + ) + .await + { + Ok(response) => response, + Err(error) => SystemResponse::Error { + message: error.to_string(), }, - - SystemMessage::StopActor { name } => { - // Mark as stopped (actual stop handled by ActorSystem) - if self.registry.contains(&name) { - self.unregister_actor(&name); - SystemResponse::Ok - } else { - SystemResponse::Error { - message: format!("Actor not found: {}", name), - } - } - } - - SystemMessage::Extension { handler, payload } => { - // Call factory's extension handler - self.factory.handle_extension(&handler, payload).await - } }; // Use JSON serialization for response (for Python compatibility) diff --git a/crates/pulsing-actor/src/system_actor/service.rs b/crates/pulsing-actor/src/system_actor/service.rs new file mode 100644 index 000000000..7d0dd0e48 --- /dev/null +++ b/crates/pulsing-actor/src/system_actor/service.rs @@ -0,0 +1,556 @@ +//! System-service registration, lifecycle, and routing. +//! +//! A service is a governed node capability, not a resource owner. The host owns +//! resources, the component owns lifecycle hooks, and the handler owns request +//! semantics. Keeping those roles separate allows future Forge and agent +//! extensions without turning SystemRoot into a service locator. + +use super::lifecycle::NodeState; +use super::{SystemMessage, SystemResponse}; +use crate::error::{PulsingError, Result, RuntimeError}; +use async_trait::async_trait; +use std::sync::{Arc, RwLock}; + +/// Stable logical identity of a system service. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub(crate) struct SystemServiceId { + pub namespace: &'static str, + pub major: u16, +} + +/// Whether a service is built into the node or installed as an extension. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum SystemServiceKind { + Core, + #[allow(dead_code)] + Extension, +} + +/// Declared exposure policy. Enforcement happens at the protocol boundary +/// once a request carries an authenticated origin. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum SystemServiceExposure { + LocalOnly, + AuthenticatedRemote, +} + +/// Permission class of an operation. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum OperationAccess { + Read, + Operate, + Admin, +} + +/// Static operation contract used for policy and discovery. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct OperationManifest { + pub name: &'static str, + pub access: OperationAccess, + pub allowed_while_draining: bool, +} + +/// Static identity and policy contract for a system service. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct SystemServiceManifest { + pub id: SystemServiceId, + pub kind: SystemServiceKind, + pub exposure: SystemServiceExposure, + pub operations: &'static [OperationManifest], +} + +/// Observable lifecycle state of a registered service. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum SystemServiceState { + Registered, + Starting, + Ready, + Failed, + Stopping, + Stopped, +} + +/// Internal typed command set. New wire protocols adapt into this model while +/// the public legacy enum remains unchanged. +#[derive(Debug)] +pub(crate) enum SystemCommand { + Runtime(RuntimeCommand), + Actors(ActorsCommand), + Metrics(MetricsCommand), + Shm(ShmCommand), +} + +#[derive(Debug)] +pub(crate) enum RuntimeCommand { + NodeInfo, + Health, + Ping, +} + +#[derive(Debug)] +pub(crate) enum ActorsCommand { + List, + Get { + name: String, + }, + Create { + actor_type: String, + name: String, + params: serde_json::Value, + public: bool, + }, + Stop { + name: String, + }, + Extension { + handler: String, + payload: serde_json::Value, + }, +} + +#[derive(Debug)] +pub(crate) enum MetricsCommand { + Snapshot, +} + +#[derive(Debug)] +pub(crate) enum ShmCommand { + Stats, +} + +impl SystemCommand { + pub(crate) fn from_legacy(message: SystemMessage) -> Self { + match message { + SystemMessage::Ping => Self::Runtime(RuntimeCommand::Ping), + SystemMessage::GetNodeInfo => Self::Runtime(RuntimeCommand::NodeInfo), + SystemMessage::HealthCheck => Self::Runtime(RuntimeCommand::Health), + SystemMessage::ListActors => Self::Actors(ActorsCommand::List), + SystemMessage::GetActor { name } => Self::Actors(ActorsCommand::Get { name }), + SystemMessage::CreateActor { + actor_type, + name, + params, + public, + } => Self::Actors(ActorsCommand::Create { + actor_type, + name, + params, + public, + }), + SystemMessage::StopActor { name } => Self::Actors(ActorsCommand::Stop { name }), + SystemMessage::Extension { handler, payload } => { + Self::Actors(ActorsCommand::Extension { handler, payload }) + } + SystemMessage::GetMetrics => Self::Metrics(MetricsCommand::Snapshot), + SystemMessage::GetShmStats => Self::Shm(ShmCommand::Stats), + } + } + + pub(crate) fn target(&self) -> SystemServiceId { + match self { + Self::Runtime(_) => RUNTIME_SERVICE, + Self::Actors(_) => ACTORS_SERVICE, + Self::Metrics(_) => METRICS_SERVICE, + Self::Shm(_) => SHM_SERVICE, + } + } + + pub(crate) fn operation(&self) -> &'static str { + match self { + Self::Runtime(RuntimeCommand::NodeInfo) => "node_info", + Self::Runtime(RuntimeCommand::Health) => "health", + Self::Runtime(RuntimeCommand::Ping) => "ping", + Self::Actors(ActorsCommand::List) => "list", + Self::Actors(ActorsCommand::Get { .. }) => "get", + Self::Actors(ActorsCommand::Create { .. }) => "create", + Self::Actors(ActorsCommand::Stop { .. }) => "stop", + Self::Actors(ActorsCommand::Extension { .. }) => "extension", + Self::Metrics(MetricsCommand::Snapshot) => "snapshot", + Self::Shm(ShmCommand::Stats) => "stats", + } + } +} + +pub(crate) const RUNTIME_SERVICE: SystemServiceId = SystemServiceId { + namespace: "runtime", + major: 1, +}; +pub(crate) const ACTORS_SERVICE: SystemServiceId = SystemServiceId { + namespace: "actors", + major: 1, +}; +pub(crate) const METRICS_SERVICE: SystemServiceId = SystemServiceId { + namespace: "metrics", + major: 1, +}; +pub(crate) const SHM_SERVICE: SystemServiceId = SystemServiceId { + namespace: "shm", + major: 1, +}; + +/// Lifecycle hooks for a node component. +#[async_trait] +pub(crate) trait SystemComponent: Send + Sync { + async fn start(&self) -> Result<()> { + Ok(()) + } + + async fn stop(&self) -> Result<()> { + Ok(()) + } +} + +/// Request semantics for one registered service. +#[async_trait] +pub(crate) trait SystemRequestHandler: Send + Sync { + async fn handle(&self, command: SystemCommand) -> Result; +} + +pub(crate) struct StatelessComponent; + +#[async_trait] +impl SystemComponent for StatelessComponent {} + +/// Complete bootstrap registration for one service. +pub(crate) struct SystemServiceRegistration { + pub manifest: SystemServiceManifest, + pub component: Arc, + pub handler: Arc, +} + +struct RegisteredSystemService { + registration: SystemServiceRegistration, + state: SystemServiceState, +} + +/// Bootstrap-time registry and runtime router owned by SystemRoot. +/// +/// Registration is intentionally closed after startup. Runtime installation +/// will require a staged generation/rollback protocol rather than direct +/// mutation of this registry. +#[derive(Default)] +pub(crate) struct SystemServiceRegistry { + entries: RwLock>, +} + +impl SystemServiceRegistry { + pub(crate) fn new() -> Self { + Self::default() + } + + pub(crate) fn register(&self, registration: SystemServiceRegistration) -> Result<()> { + let manifest = ®istration.manifest; + let mut entries = self + .entries + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if entries + .iter() + .any(|entry| entry.registration.manifest.id == manifest.id) + { + return Err(service_error(format!( + "service {}@{} is already registered", + manifest.id.namespace, manifest.id.major + ))); + } + validate_manifest(manifest)?; + entries.push(RegisteredSystemService { + registration, + state: SystemServiceState::Registered, + }); + Ok(()) + } + + pub(crate) fn statuses(&self) -> Vec<(SystemServiceManifest, SystemServiceState)> { + self.entries + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .iter() + .map(|entry| (entry.registration.manifest.clone(), entry.state)) + .collect() + } + + pub(crate) async fn start_all(&self) -> Result<()> { + let components: Vec<_> = self + .entries + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .iter() + .map(|entry| { + ( + entry.registration.manifest.id, + entry.registration.component.clone(), + ) + }) + .collect(); + + for (id, component) in components { + self.transition_service( + id, + SystemServiceState::Registered, + SystemServiceState::Starting, + )?; + if let Err(error) = component.start().await { + let _ = self.set_state(id, SystemServiceState::Failed); + let _ = self.stop_ready().await; + return Err(error); + } + self.transition_service(id, SystemServiceState::Starting, SystemServiceState::Ready)?; + } + Ok(()) + } + + pub(crate) async fn stop_all(&self) -> Result<()> { + self.stop_ready().await + } + + pub(crate) async fn dispatch( + &self, + command: SystemCommand, + node_state: NodeState, + ) -> Result { + let target = command.target(); + let operation_name = command.operation(); + let (handler, operation) = { + let entries = self + .entries + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let entry = entries + .iter() + .find(|entry| entry.registration.manifest.id == target) + .ok_or_else(|| { + service_error(format!( + "service {}@{} is not registered", + target.namespace, target.major + )) + })?; + if entry.state != SystemServiceState::Ready { + return Err(service_error(format!( + "service {}@{} is not ready ({:?})", + target.namespace, target.major, entry.state + ))); + } + let operation = entry + .registration + .manifest + .operations + .iter() + .find(|operation| operation.name == operation_name) + .copied() + .ok_or_else(|| { + service_error(format!( + "operation {operation_name} is not declared by {}@{}", + target.namespace, target.major + )) + })?; + (entry.registration.handler.clone(), operation) + }; + + if node_state == NodeState::Draining && !operation.allowed_while_draining { + return Err(service_error(format!( + "operation {} is unavailable while the node is draining", + operation.name + ))); + } + handler.handle(command).await + } + + fn transition_service( + &self, + id: SystemServiceId, + expected: SystemServiceState, + next: SystemServiceState, + ) -> Result<()> { + let mut entries = self + .entries + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let entry = entries + .iter_mut() + .find(|entry| entry.registration.manifest.id == id) + .ok_or_else(|| service_error("registered service disappeared"))?; + if entry.state != expected { + return Err(service_error(format!( + "invalid service transition for {}@{}: {:?} -> {:?}", + id.namespace, id.major, entry.state, next + ))); + } + entry.state = next; + Ok(()) + } + + fn set_state(&self, id: SystemServiceId, state: SystemServiceState) -> Result<()> { + let mut entries = self + .entries + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let entry = entries + .iter_mut() + .find(|entry| entry.registration.manifest.id == id) + .ok_or_else(|| service_error("registered service disappeared"))?; + entry.state = state; + Ok(()) + } + + async fn stop_ready(&self) -> Result<()> { + let components: Vec<_> = self + .entries + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .iter() + .rev() + .filter(|entry| entry.state == SystemServiceState::Ready) + .map(|entry| { + ( + entry.registration.manifest.id, + entry.registration.component.clone(), + ) + }) + .collect(); + let mut failures = Vec::new(); + for (id, component) in components { + self.transition_service(id, SystemServiceState::Ready, SystemServiceState::Stopping)?; + match component.stop().await { + Ok(()) => self.transition_service( + id, + SystemServiceState::Stopping, + SystemServiceState::Stopped, + )?, + Err(error) => { + self.set_state(id, SystemServiceState::Failed)?; + failures.push(format!("{}@{}: {error}", id.namespace, id.major)); + } + } + } + if failures.is_empty() { + Ok(()) + } else { + Err(service_error(format!( + "failed to stop system services: {}", + failures.join("; ") + ))) + } + } +} + +fn validate_manifest(manifest: &SystemServiceManifest) -> Result<()> { + if manifest.id.namespace.is_empty() { + return Err(service_error("service namespace must not be empty")); + } + if manifest.id.major == 0 { + return Err(service_error("service major version must be non-zero")); + } + for (index, operation) in manifest.operations.iter().enumerate() { + if operation.name.is_empty() { + return Err(service_error("operation name must not be empty")); + } + if manifest.operations[..index] + .iter() + .any(|existing| existing.name == operation.name) + { + return Err(service_error(format!( + "operation {} is declared more than once", + operation.name + ))); + } + } + Ok(()) +} + +pub(crate) fn service_error(message: impl Into) -> PulsingError { + PulsingError::from(RuntimeError::Other(format!( + "System service error: {}", + message.into() + ))) +} + +#[cfg(test)] +mod tests { + use super::super::builtin; + use super::super::host::SystemHost; + use super::super::{ + ActorRegistry, DefaultActorFactory, SystemMetrics, SystemRef, SystemResponse, + }; + use super::*; + use crate::actor::NodeId; + use crate::performance_store::PerformanceStore; + use std::sync::Arc; + + fn test_host() -> SystemHost { + let system_ref = SystemRef::new(NodeId::generate(), "127.0.0.1:0".parse().unwrap()); + SystemHost::standalone( + &system_ref, + Arc::new(ActorRegistry::new()), + Arc::new(SystemMetrics::new()), + Arc::new(PerformanceStore::new(8)), + ) + } + + fn builtins(host: &SystemHost) -> Vec { + builtin::registrations(host, Arc::new(DefaultActorFactory)) + } + + #[test] + fn registry_rejects_duplicate_service_major() { + let registry = SystemServiceRegistry::new(); + let host = test_host(); + registry.register(builtins(&host).remove(0)).unwrap(); + assert!(registry.register(builtins(&host).remove(0)).is_err()); + } + + #[tokio::test] + async fn registry_requires_ready_and_honors_draining_policy() { + let registry = SystemServiceRegistry::new(); + let host = test_host(); + for registration in builtins(&host) { + registry.register(registration).unwrap(); + } + + assert!(registry + .dispatch( + SystemCommand::Runtime(RuntimeCommand::Ping), + host.node_state(), + ) + .await + .is_err()); + + host.lifecycle().transition(NodeState::Starting).unwrap(); + registry.start_all().await.unwrap(); + host.lifecycle().transition(NodeState::Ready).unwrap(); + assert!(matches!( + registry + .dispatch( + SystemCommand::Runtime(RuntimeCommand::Ping), + host.node_state(), + ) + .await + .unwrap(), + SystemResponse::Pong { .. } + )); + + host.lifecycle().begin_draining().unwrap(); + assert!(registry + .dispatch( + SystemCommand::Actors(ActorsCommand::Stop { + name: "actors/test".to_string(), + }), + host.node_state(), + ) + .await + .is_err()); + assert!(registry + .dispatch( + SystemCommand::Runtime(RuntimeCommand::Ping), + host.node_state(), + ) + .await + .is_ok()); + + registry.stop_all().await.unwrap(); + assert!(registry + .statuses() + .iter() + .all(|(_, state)| *state == SystemServiceState::Stopped)); + } +} diff --git a/crates/pulsing-actor/src/system_actor/shm.rs b/crates/pulsing-actor/src/system_actor/shm.rs new file mode 100644 index 000000000..659485f48 --- /dev/null +++ b/crates/pulsing-actor/src/system_actor/shm.rs @@ -0,0 +1,436 @@ +//! Shared-memory control-plane primitives. +//! +//! This module deliberately owns *lifetime* and *capability* semantics before +//! an OS specific mapping backend is introduced. The current backend is an +//! in-process, immutable `Bytes` region: cloning a mapping is zero-copy and +//! has the same lease and revocation rules a cross-process backend will need. +//! +//! A future POSIX/Windows shared-memory transport can replace the backing +//! implementation behind these descriptors without changing the rendezvous +//! (`offer`) or serving (`publish` / `open`) API. + +use crate::error::{PulsingError, Result, RuntimeError}; +use bytes::Bytes; +use std::collections::HashMap; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +/// The backing currently selected by a [`ShmManager`]. +/// +/// `InProcess` is intentionally explicit: it is a semantic foundation, not a +/// claim that an arbitrary peer process can already map this memory. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum ShmBackend { + /// `Bytes`-backed regions shared by handles in this ActorSystem process. + #[default] + InProcess, +} + +impl ShmBackend { + pub const fn as_str(self) -> &'static str { + match self { + Self::InProcess => "in_process", + } + } +} + +/// Opaque capability for a bounded part of a shared-memory region. +/// +/// A descriptor is valid only while its lease is active. `generation` is +/// included so future reusable OS allocations can reject stale descriptors. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ShmRegionDescriptor { + pub region_id: u64, + pub generation: u64, + pub offset: usize, + pub len: usize, + pub lease_id: u64, +} + +/// Snapshot for metrics and control-plane inspection. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct ShmStats { + pub regions: usize, + pub published_regions: usize, + pub active_leases: usize, + pub bytes: usize, +} + +/// Node-scoped region manager. +/// +/// The lock is only on metadata operations. A successful [`Self::map`] +/// returns a `Bytes` handle and does not keep the manager locked while users +/// consume the payload. +#[derive(Default)] +pub struct ShmManager { + state: Mutex, +} + +#[derive(Default)] +struct ShmState { + next_region_id: u64, + next_lease_id: u64, + regions: HashMap, + published_names: HashMap, +} + +struct Region { + generation: u64, + bytes: Bytes, + published_name: Option, + /// `None` represents a TTL that exceeds the platform `Instant` range and + /// is therefore treated as non-expiring rather than immediately expired. + leases: HashMap>, +} + +impl ShmManager { + pub fn new() -> Self { + Self::default() + } + + pub fn backend(&self) -> ShmBackend { + ShmBackend::InProcess + } + + /// Create a one-recipient region for the rendezvous/message-style API. + /// + /// Releasing (or expiring) the returned lease removes the region. + pub fn offer(&self, bytes: Bytes, lease_ttl: Duration) -> ShmRegionDescriptor { + let mut state = self.lock_state(); + reclaim_expired_locked(&mut state, Instant::now()); + let region_id = next_id(&mut state.next_region_id); + let generation = 1; + let lease_id = next_id(&mut state.next_lease_id); + let descriptor = ShmRegionDescriptor { + region_id, + generation, + offset: 0, + len: bytes.len(), + lease_id, + }; + let mut leases = HashMap::new(); + leases.insert(lease_id, deadline(lease_ttl)); + state.regions.insert( + region_id, + Region { + generation, + bytes, + published_name: None, + leases, + }, + ); + descriptor + } + + /// Publish a named region for the serve-style API. + /// + /// The name remains available until [`Self::unpublish`] is called. Each + /// [`Self::open`] creates an independent lease, so one consumer cannot + /// revoke another consumer's mapping. + pub fn publish(&self, name: impl Into, bytes: Bytes) -> Result<()> { + let name = name.into(); + if name.is_empty() { + return Err(shm_error("published region name must not be empty")); + } + + let mut state = self.lock_state(); + reclaim_expired_locked(&mut state, Instant::now()); + if state.published_names.contains_key(&name) { + return Err(shm_error(format!( + "shared-memory region '{name}' already exists" + ))); + } + let region_id = next_id(&mut state.next_region_id); + state.published_names.insert(name.clone(), region_id); + state.regions.insert( + region_id, + Region { + generation: 1, + bytes, + published_name: Some(name), + leases: HashMap::new(), + }, + ); + Ok(()) + } + + /// Open a published region and create a time-bounded consumer capability. + pub fn open(&self, name: &str, lease_ttl: Duration) -> Result { + let mut state = self.lock_state(); + reclaim_expired_locked(&mut state, Instant::now()); + let region_id = *state + .published_names + .get(name) + .ok_or_else(|| shm_error(format!("shared-memory region '{name}' was not found")))?; + let lease_id = next_id(&mut state.next_lease_id); + let region = state + .regions + .get_mut(®ion_id) + .expect("published region must exist"); + region.leases.insert(lease_id, deadline(lease_ttl)); + Ok(ShmRegionDescriptor { + region_id, + generation: region.generation, + offset: 0, + len: region.bytes.len(), + lease_id, + }) + } + + /// Resolve a descriptor to its zero-copy in-process mapping. + pub fn map(&self, descriptor: &ShmRegionDescriptor) -> Result { + let mut state = self.lock_state(); + let now = Instant::now(); + let remove = match state.regions.get_mut(&descriptor.region_id) { + Some(region) => { + if region.generation != descriptor.generation { + return Err(shm_error("stale shared-memory region descriptor")); + } + match region.leases.get(&descriptor.lease_id) { + Some(expires_at) + if expires_at + .as_ref() + .map(|expires_at| *expires_at > now) + .unwrap_or(true) => + { + let end = descriptor + .offset + .checked_add(descriptor.len) + .ok_or_else(|| shm_error("shared-memory descriptor range overflow"))?; + if end > region.bytes.len() { + return Err(shm_error("shared-memory descriptor range is invalid")); + } + return Ok(region.bytes.slice(descriptor.offset..end)); + } + Some(_) => { + region.leases.remove(&descriptor.lease_id); + region.leases.is_empty() && region.published_name.is_none() + } + None => return Err(shm_error("shared-memory lease is no longer active")), + } + } + None => return Err(shm_error("shared-memory region is no longer available")), + }; + if remove { + state.regions.remove(&descriptor.region_id); + } + Err(shm_error("shared-memory lease has expired")) + } + + /// Release a consumer capability. This is idempotent, which lets actor + /// teardown safely race with best-effort lease cleanup. + pub fn release(&self, descriptor: &ShmRegionDescriptor) { + let mut state = self.lock_state(); + let remove = if let Some(region) = state.regions.get_mut(&descriptor.region_id) { + if region.generation != descriptor.generation { + return; + } + region.leases.remove(&descriptor.lease_id); + region.leases.is_empty() && region.published_name.is_none() + } else { + false + }; + if remove { + state.regions.remove(&descriptor.region_id); + } + } + + /// Stop accepting new opens. Existing leases remain valid until released + /// or expired, which gives serving and rendezvous the same drain rule. + pub fn unpublish(&self, name: &str) -> Result<()> { + let mut state = self.lock_state(); + let region_id = state + .published_names + .remove(name) + .ok_or_else(|| shm_error(format!("shared-memory region '{name}' was not found")))?; + let remove = if let Some(region) = state.regions.get_mut(®ion_id) { + region.published_name = None; + region.leases.is_empty() + } else { + false + }; + if remove { + state.regions.remove(®ion_id); + } + Ok(()) + } + + /// Reclaim expired leases and return how many leases were removed. + pub fn reclaim_expired(&self) -> usize { + let mut state = self.lock_state(); + reclaim_expired_locked(&mut state, Instant::now()) + } + + pub fn stats(&self) -> ShmStats { + let mut state = self.lock_state(); + reclaim_expired_locked(&mut state, Instant::now()); + ShmStats { + regions: state.regions.len(), + published_regions: state.published_names.len(), + active_leases: state + .regions + .values() + .map(|region| region.leases.len()) + .sum(), + bytes: state + .regions + .values() + .map(|region| region.bytes.len()) + .sum(), + } + } + + /// Revoke every manager-owned descriptor during node shutdown. + /// + /// Existing cloned `Bytes` values remain memory-safe, but no descriptor can + /// be mapped again after this call. + pub fn clear(&self) { + let mut state = self.lock_state(); + state.regions.clear(); + state.published_names.clear(); + } + + fn lock_state(&self) -> std::sync::MutexGuard<'_, ShmState> { + self.state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} + +fn next_id(counter: &mut u64) -> u64 { + *counter = counter + .checked_add(1) + .expect("shared-memory identifier exhausted"); + *counter +} + +fn deadline(ttl: Duration) -> Option { + Instant::now().checked_add(ttl) +} + +fn reclaim_expired_locked(state: &mut ShmState, now: Instant) -> usize { + let mut reclaimed = 0; + let mut empty_unpublished = Vec::new(); + for (region_id, region) in &mut state.regions { + let before = region.leases.len(); + region.leases.retain(|_, expires_at| { + expires_at + .as_ref() + .map(|expires_at| *expires_at > now) + .unwrap_or(true) + }); + reclaimed += before - region.leases.len(); + if region.leases.is_empty() && region.published_name.is_none() { + empty_unpublished.push(*region_id); + } + } + for region_id in empty_unpublished { + state.regions.remove(®ion_id); + } + reclaimed +} + +fn shm_error(message: impl Into) -> PulsingError { + PulsingError::from(RuntimeError::Other(format!( + "Shared-memory error: {}", + message.into() + ))) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn offer_maps_zero_copy_and_release_revokes_it() { + let manager = ShmManager::new(); + let source = Bytes::from_static(b"tensor"); + let descriptor = manager.offer(source.clone(), Duration::from_secs(1)); + + let mapped = manager.map(&descriptor).unwrap(); + assert_eq!(mapped, source); + assert_eq!(mapped.as_ptr(), source.as_ptr()); + + manager.release(&descriptor); + assert!(manager.map(&descriptor).is_err()); + assert_eq!(manager.stats().regions, 0); + } + + #[test] + fn publish_open_and_unpublish_drain_existing_leases() { + let manager = ShmManager::new(); + manager + .publish("models/current", Bytes::from_static(b"weights")) + .unwrap(); + let descriptor = manager + .open("models/current", Duration::from_secs(1)) + .unwrap(); + + manager.unpublish("models/current").unwrap(); + assert!(manager + .open("models/current", Duration::from_secs(1)) + .is_err()); + assert_eq!(&manager.map(&descriptor).unwrap()[..], b"weights"); + + manager.release(&descriptor); + assert_eq!(manager.stats().regions, 0); + } + + #[test] + fn expired_offer_is_reclaimed() { + let manager = ShmManager::new(); + let descriptor = manager.offer(Bytes::from_static(b"x"), Duration::ZERO); + assert_eq!(manager.reclaim_expired(), 1); + assert!(manager.map(&descriptor).is_err()); + } + + #[test] + fn new_offer_opportunistically_reclaims_expired_regions() { + let manager = ShmManager::new(); + let expired = manager.offer(Bytes::from_static(b"old"), Duration::ZERO); + let active = manager.offer(Bytes::from_static(b"new"), Duration::from_secs(1)); + + assert!(manager.map(&expired).is_err()); + assert_eq!(&manager.map(&active).unwrap()[..], b"new"); + assert_eq!(manager.stats().regions, 1); + } + + #[test] + fn stats_excludes_expired_leases_but_retains_published_region() { + let manager = ShmManager::new(); + manager + .publish("models/current", Bytes::from_static(b"weights")) + .unwrap(); + manager.open("models/current", Duration::ZERO).unwrap(); + + assert_eq!( + manager.stats(), + ShmStats { + regions: 1, + published_regions: 1, + active_leases: 0, + bytes: 7, + } + ); + } + + #[test] + fn overflowing_ttl_does_not_expire_immediately() { + let manager = ShmManager::new(); + let descriptor = manager.offer(Bytes::from_static(b"x"), Duration::MAX); + assert_eq!(&manager.map(&descriptor).unwrap()[..], b"x"); + assert_eq!(manager.reclaim_expired(), 0); + } + + #[test] + fn clear_revokes_all_descriptors() { + let manager = ShmManager::new(); + let descriptor = manager.offer(Bytes::from_static(b"x"), Duration::from_secs(1)); + manager + .publish("models/current", Bytes::from_static(b"weights")) + .unwrap(); + manager.clear(); + assert_eq!(manager.stats(), ShmStats::default()); + assert!(manager.map(&descriptor).is_err()); + } +} diff --git a/crates/pulsing-actor/src/transport/http2/client.rs b/crates/pulsing-actor/src/transport/http2/client.rs index 74986b89d..464f13f73 100644 --- a/crates/pulsing-actor/src/transport/http2/client.rs +++ b/crates/pulsing-actor/src/transport/http2/client.rs @@ -15,7 +15,9 @@ use crate::tracing::{ }; use crate::transport::tensor::{ raw_tensor_transport_requested, read_raw_tensor_frame, record_tensor_http2_fallback, - write_raw_tensor_frame, RawTensorKind, TensorCopyModel, + write_raw_tensor_frame, RawTensorKind, TensorCopyModel, TensorTransportCapabilities, + TensorTransportLocality, TensorTransportPreference, TensorTransportRoute, + TensorTransportRouter, }; use bytes::Bytes; use futures::{Stream, StreamExt, TryStreamExt}; @@ -361,22 +363,30 @@ impl Http2Client { } pub(crate) fn tensor_copy_model(&self) -> TensorCopyModel { - if self.raw_tensor_enabled() { - TensorCopyModel::DirectTcp - } else { - TensorCopyModel::PackedHttp2Compatibility - } + self.tensor_route().copy_model() } fn raw_tensor_enabled(&self) -> bool { - if !raw_tensor_transport_requested() { - return false; - } + self.tensor_route() == TensorTransportRoute::DirectTcp + } + + fn tensor_route(&self) -> TensorTransportRoute { + let mut direct_tcp = raw_tensor_transport_requested(); #[cfg(feature = "tls")] if self.config.tls.is_some() { - return false; + direct_tcp = false; } - true + TensorTransportRouter::select( + TensorTransportPreference::from_environment(), + TensorTransportLocality::Remote, + TensorTransportCapabilities { + direct_tcp, + // Cross-process SHM has not been enabled. The node-level + // manager exists, but a peer capability handshake and OS + // mapper are required before this can become true. + shared_memory: false, + }, + ) } async fn take_tensor_connection(&self, addr: SocketAddr) -> Result { diff --git a/crates/pulsing-actor/src/transport/mod.rs b/crates/pulsing-actor/src/transport/mod.rs index 8fe4e29b1..665b795a5 100644 --- a/crates/pulsing-actor/src/transport/mod.rs +++ b/crates/pulsing-actor/src/transport/mod.rs @@ -24,4 +24,6 @@ pub use http2::{ }; pub use tensor::{ raw_tensor_transport_stats, RawTensorTransportStats, TensorCopyModel, TensorTransport, + TensorTransportCapabilities, TensorTransportLocality, TensorTransportPreference, + TensorTransportRoute, TensorTransportRouter, }; diff --git a/crates/pulsing-actor/src/transport/tensor.rs b/crates/pulsing-actor/src/transport/tensor.rs index 6a9bc4811..876fd6b9c 100644 --- a/crates/pulsing-actor/src/transport/tensor.rs +++ b/crates/pulsing-actor/src/transport/tensor.rs @@ -316,11 +316,8 @@ where pub(crate) fn raw_tensor_transport_requested() -> bool { !matches!( - std::env::var("PULSING_TENSOR_TRANSPORT") - .unwrap_or_else(|_| "auto".to_string()) - .to_ascii_lowercase() - .as_str(), - "http2" | "legacy" | "off" + TensorTransportPreference::from_environment(), + TensorTransportPreference::Http2Compatibility ) } @@ -487,6 +484,104 @@ pub enum TensorCopyModel { SharedMemory, } +/// Caller intent for selecting a tensor data plane. +/// +/// `SharedMemory` is never silently treated as a raw socket request: if no +/// compatible shared-memory backend is available, the router picks a safe +/// existing fallback instead. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TensorTransportPreference { + Auto, + DirectTcp, + Http2Compatibility, + SharedMemory, +} + +impl TensorTransportPreference { + pub fn from_environment() -> Self { + match std::env::var("PULSING_TENSOR_TRANSPORT") + .unwrap_or_else(|_| "auto".to_string()) + .to_ascii_lowercase() + .as_str() + { + "http2" | "legacy" | "off" => Self::Http2Compatibility, + "shm" | "shared_memory" | "shared-memory" => Self::SharedMemory, + "tcp" | "raw" | "direct_tcp" | "direct-tcp" => Self::DirectTcp, + _ => Self::Auto, + } + } +} + +/// How closely the sender and receiver are placed. Same host is not enough +/// for SHM: it still needs a negotiated mapping backend and compatible access +/// rights. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TensorTransportLocality { + SameProcess, + SameHost, + Remote, +} + +/// Capabilities known at a concrete send site. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct TensorTransportCapabilities { + pub direct_tcp: bool, + pub shared_memory: bool, +} + +/// Result of selecting a tensor data plane. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TensorTransportRoute { + InProcessSharedMemory, + DirectTcp, + Http2Compatibility, +} + +impl TensorTransportRoute { + pub fn copy_model(self) -> TensorCopyModel { + match self { + Self::InProcessSharedMemory => TensorCopyModel::SharedMemory, + Self::DirectTcp => TensorCopyModel::DirectTcp, + Self::Http2Compatibility => TensorCopyModel::PackedHttp2Compatibility, + } + } +} + +/// Transport policy separated from individual HTTP/2 and future SHM backends. +/// +/// The rule intentionally selects SHM only for a same-process, explicitly +/// available backend. A later same-host backend can extend this decision once +/// it has a peer capability handshake and OS mapping implementation. +#[derive(Clone, Copy, Debug, Default)] +pub struct TensorTransportRouter; + +impl TensorTransportRouter { + pub fn select( + preference: TensorTransportPreference, + locality: TensorTransportLocality, + capabilities: TensorTransportCapabilities, + ) -> TensorTransportRoute { + let shm_available = + locality == TensorTransportLocality::SameProcess && capabilities.shared_memory; + match preference { + TensorTransportPreference::Http2Compatibility => { + TensorTransportRoute::Http2Compatibility + } + TensorTransportPreference::SharedMemory if shm_available => { + TensorTransportRoute::InProcessSharedMemory + } + TensorTransportPreference::SharedMemory + | TensorTransportPreference::DirectTcp + | TensorTransportPreference::Auto + if capabilities.direct_tcp => + { + TensorTransportRoute::DirectTcp + } + _ => TensorTransportRoute::Http2Compatibility, + } + } +} + /// Backend boundary for opaque metadata plus contiguous tensor buffers. #[async_trait::async_trait] pub trait TensorTransport: Send + Sync { @@ -500,3 +595,44 @@ pub trait TensorTransport: Send + Sync { fn copy_model(&self) -> TensorCopyModel; } + +#[cfg(test)] +mod routing_tests { + use super::*; + + #[test] + fn shared_memory_requires_same_process_capability() { + let route = TensorTransportRouter::select( + TensorTransportPreference::SharedMemory, + TensorTransportLocality::SameHost, + TensorTransportCapabilities { + direct_tcp: true, + shared_memory: true, + }, + ); + assert_eq!(route, TensorTransportRoute::DirectTcp); + + let route = TensorTransportRouter::select( + TensorTransportPreference::SharedMemory, + TensorTransportLocality::SameProcess, + TensorTransportCapabilities { + direct_tcp: true, + shared_memory: true, + }, + ); + assert_eq!(route, TensorTransportRoute::InProcessSharedMemory); + } + + #[test] + fn existing_compatibility_mode_remains_authoritative() { + let route = TensorTransportRouter::select( + TensorTransportPreference::Http2Compatibility, + TensorTransportLocality::Remote, + TensorTransportCapabilities { + direct_tcp: true, + shared_memory: false, + }, + ); + assert_eq!(route, TensorTransportRoute::Http2Compatibility); + } +} diff --git a/crates/pulsing-actor/tests/unit/system/system_actor_tests.rs b/crates/pulsing-actor/tests/unit/system/system_actor_tests.rs index 8637e78f1..a875a65a1 100644 --- a/crates/pulsing-actor/tests/unit/system/system_actor_tests.rs +++ b/crates/pulsing-actor/tests/unit/system/system_actor_tests.rs @@ -5,7 +5,8 @@ use pulsing_actor::actor::ActorId; use pulsing_actor::prelude::*; use pulsing_actor::system_actor::{ - ActorInfo, ActorRegistry, SystemMessage, SystemMetrics, SystemResponse, SYSTEM_ACTOR_PATH, + ActorFactory, ActorInfo, ActorRegistry, SystemMessage, SystemMetrics, SystemResponse, + SYSTEM_ACTOR_PATH, }; use std::time::Duration; @@ -32,6 +33,42 @@ fn create_system_message(msg: &SystemMessage) -> Message { } } +struct BootstrapFactory; + +#[async_trait] +impl ActorFactory for BootstrapFactory { + fn supports(&self, _actor_type: &str) -> bool { + false + } + + fn supported_types(&self) -> Vec { + Vec::new() + } + + async fn handle_extension(&self, handler: &str, _payload: serde_json::Value) -> SystemResponse { + if handler == "bootstrap_test" { + SystemResponse::Ok + } else { + SystemResponse::Error { + message: "unknown test extension".to_string(), + } + } + } +} + +struct ManagedActor; + +#[async_trait] +impl Actor for ManagedActor { + async fn receive( + &mut self, + message: Message, + _ctx: &mut ActorContext, + ) -> pulsing_actor::error::Result { + Ok(message) + } +} + // ============================================================================ // SystemActor Path Tests // ============================================================================ @@ -65,6 +102,16 @@ fn test_system_message_list_actors_serialization() { assert!(matches!(parsed, SystemMessage::ListActors)); } +#[test] +fn test_system_message_get_shm_stats_serialization() { + let msg = SystemMessage::GetShmStats; + let json = serde_json::to_string(&msg).unwrap(); + assert!(json.contains("GetShmStats")); + + let parsed: SystemMessage = serde_json::from_str(&json).unwrap(); + assert!(matches!(parsed, SystemMessage::GetShmStats)); +} + #[test] fn test_system_message_get_actor_serialization() { let msg = SystemMessage::GetActor { @@ -197,6 +244,30 @@ async fn test_system_actor_auto_start() { system.shutdown().await.unwrap(); } +#[tokio::test] +async fn test_system_actor_factory_is_installed_during_bootstrap() { + let system = ActorSystem::new_with_system_actor_factory( + SystemConfig::standalone(), + Box::new(BootstrapFactory), + ) + .await + .unwrap(); + let sys_ref = system.system().await.unwrap(); + let response = sys_ref + .send(create_system_message(&SystemMessage::Extension { + handler: "bootstrap_test".to_string(), + payload: serde_json::Value::Null, + })) + .await + .unwrap(); + assert!(matches!( + parse_system_response(response), + SystemResponse::Ok + )); + + system.shutdown().await.unwrap(); +} + #[tokio::test] async fn test_system_actor_ping() { let system = create_test_system().await; @@ -264,6 +335,36 @@ async fn test_system_actor_get_node_info() { system.shutdown().await.unwrap(); } +#[tokio::test] +async fn test_system_actor_get_shm_stats() { + let system = create_test_system().await; + let sys_ref = system.system().await.unwrap(); + + let response = sys_ref + .send(create_system_message(&SystemMessage::GetShmStats)) + .await + .unwrap(); + let parsed = parse_system_response(response); + match parsed { + SystemResponse::ShmStats { + backend, + regions, + published_regions, + active_leases, + bytes, + } => { + assert_eq!(backend, "in_process"); + assert_eq!(regions, 0); + assert_eq!(published_regions, 0); + assert_eq!(active_leases, 0); + assert_eq!(bytes, 0); + } + _ => panic!("Expected ShmStats response"), + } + + system.shutdown().await.unwrap(); +} + #[tokio::test] async fn test_system_actor_get_metrics() { let system = create_test_system().await; @@ -332,6 +433,57 @@ async fn test_system_actor_list_actors() { system.shutdown().await.unwrap(); } +#[tokio::test] +async fn test_system_actor_reads_and_stops_authoritative_actor() { + let system = create_test_system().await; + system + .spawn_named("actors/managed", ManagedActor) + .await + .unwrap(); + let sys_ref = system.system().await.unwrap(); + + let response = sys_ref + .send(create_system_message(&SystemMessage::GetActor { + name: "actors/managed".to_string(), + })) + .await + .unwrap(); + assert!(matches!( + parse_system_response(response), + SystemResponse::ActorInfo(_) + )); + + let response = sys_ref + .send(create_system_message(&SystemMessage::StopActor { + name: "actors/managed".to_string(), + })) + .await + .unwrap(); + assert!(matches!( + parse_system_response(response), + SystemResponse::Ok + )); + assert!(system.local_actor_ref_by_name("actors/managed").is_none()); + + system.shutdown().await.unwrap(); +} + +#[tokio::test] +async fn test_shutdown_clears_shm_control_plane() { + let system = create_test_system().await; + system + .shm_manager() + .publish("models/current", bytes::Bytes::from_static(b"weights")) + .unwrap(); + assert_eq!(system.shm_manager().stats().regions, 1); + + system.shutdown().await.unwrap(); + assert_eq!( + system.shm_manager().stats(), + pulsing_actor::system_actor::ShmStats::default() + ); +} + #[tokio::test] async fn test_system_actor_get_actor_not_found() { let system = create_test_system().await; diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index 85cf517e7..415d794bc 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -132,6 +132,7 @@ plugins: Actor Addressing: Actor 寻址 HTTP2 Transport: HTTP2 传输 TensorMessage Transport: TensorMessage 快速传输 + System Actors: System Actor 机制 Load Sync: 负载同步 AS Actor Decorator: AS Actor 装饰器 Communication Evolution: 集群内通信技术演进 @@ -223,6 +224,7 @@ nav: - Implementation: - AS Actor Decorator: design/as-actor-decorator.md - TensorMessage Transport: design/tensor-message-transport.md + - System Actors: design/system-actors.md - Communication Evolution: design/cluster-communication-evolution.md - Out-Cluster Connect: design/out-cluster-connect.md - Forge: diff --git a/docs/src/design/system-actors.md b/docs/src/design/system-actors.md new file mode 100644 index 000000000..222207a13 --- /dev/null +++ b/docs/src/design/system-actors.md @@ -0,0 +1,229 @@ +# System Actor Design + +> Status: evolving. Authoritative host capabilities, the node state machine, +> service manifests, split component/handler roles, the typed legacy adapter, +> and SHM shutdown reclamation are implemented. The versioned external +> envelope, caller identity, and governed runtime installation remain target +> work. + +## Problem and decision + +Pulsing currently auto-starts `system/core`. It has gradually accumulated +actor queries, metrics, Python extension hooks, and SHM status. The result is +a growing enum plus an untyped `Extension { handler, payload }` escape hatch; +resource ownership is not always explicit, and some system state is represented +by startup snapshots. + +This is not yet a system-actor mechanism. It makes versioning, authorization, +remote control, plugin isolation, and authoritative state increasingly hard. + +**Decision:** system actors form the node control plane, not a general-purpose +business actor or a global singleton. They route small, auditable control +commands to lifecycle-managed system services. Data planes never copy payloads +through the control plane. + +## Boundaries + +| Term | Meaning | +|---|---| +| Host | `ActorSystem` and the resources it uniquely owns: registry, transport, cluster, SHM manager, and shutdown token. | +| SystemRoot | Stable node control-plane entry point. Its permanent standard path is `system/core`. | +| SystemService | A discoverable, governed, and upgradeable node control-plane capability. | +| Control plane | Metadata, commands, status, leases, and operation orchestration. | +| Data plane | Tensor bytes, raw TCP/HTTP2 bodies, or future SHM mappings. | +| Capability | The minimum host-owned resource granted to a service. | + +The host owns resources and final shutdown. Services receive explicit +capabilities rather than an unrestricted `ActorSystem` reference. A SHM service +may create/revoke descriptors and leases; it must not transport tensor bytes in +the SystemRoot mailbox. + +## Target architecture + +```mermaid +flowchart TB + Client["Control-plane client"] --> Root["SystemRoot\nsystem/core"] + Root --> Gate["auth, deadline, audit"] + Gate --> Directory["service directory\nnamespace + version"] + Directory --> Actors["actors service"] + Directory --> Metrics["metrics service"] + Directory --> Shm["shm service"] + Directory --> Runtime["runtime service"] + + subgraph Host["ActorSystem host"] + Registry["ActorRegistry"] + Transport["Transport / Cluster"] + ShmManager["ShmManager"] + Lifecycle["Lifecycle supervisor"] + end + Actors --> Registry + Metrics --> Registry + Shm --> ShmManager + Runtime --> Lifecycle + Data["Tensor / SHM data plane"] -. "bypasses root" .-> Transport +``` + +`system/core` is the only required root entry point. The logical identity of a +service is `@` (`actors@1`, `shm@1`); future isolated service +actors may use `system/`, but that path is not the discovery +contract. + +## Formal service contract + +The implementation separates static policy, lifecycle, and request semantics: + +```rust +pub struct SystemServiceManifest { + pub id: SystemServiceId, + pub kind: Core | Extension, + pub exposure: Exposure, + pub operations: &'static [OperationManifest], +} + +#[async_trait] +pub trait SystemComponent { + async fn start(&self) -> Result<()>; + async fn stop(&self) -> Result<()>; +} + +#[async_trait] +pub trait SystemRequestHandler { + async fn handle(&self, request: SystemRequest, ctx: RequestContext) + -> Result; +} + +pub struct SystemServiceRegistration { + pub manifest: SystemServiceManifest, + pub component: Arc, + pub handler: Arc, +} +``` + +`SystemHost` injects minimum capabilities into each component/handler during +bootstrap; handlers do not receive the complete host at runtime. The actors +service holds an `ActorControl` capability backed by the authoritative host registry; the +legacy public `SystemActor::registry()` remains a compatibility projection but +is not a source of truth for built-in services. Registration is unique on +`(namespace, major)` and becomes available only after component start. + +## Protocol + +All external control requests use a stable envelope. Services may retain typed +Rust request/response values behind it. + +```text +SystemRequest { + protocol: "pulsing.system", + version: "1.0", + request_id: UUID, + target: { namespace: "shm", major: 1 }, + operation: "stats", + deadline_unix_ms: optional u64, + body: bytes, +} +``` + +The fixed envelope allows SystemRoot to apply version validation, authorization, +body limits, deadlines, cancellation, rate limiting, tracing, and auditing +without knowing each service schema. Operation descriptors specify body +encoding; Rust/Python type names are never protocol identifiers. + +Replies carry the request id and service version. Failures use stable codes: +`INVALID_ARGUMENT`, `NOT_FOUND`, `CONFLICT`, `UNAVAILABLE`, +`DEADLINE_EXCEEDED`, `PERMISSION_DENIED`, and `INTERNAL`. Long-running work +returns an `OperationHandle`; root must not wait without bound in one actor +turn. + +## Lifecycle + +```mermaid +stateDiagram-v2 + [*] --> Constructing + Constructing --> Starting: host resources ready + Starting --> Ready: required services ready + Starting --> Failed: required service start failed + Ready --> Draining: shutdown or fatal dependency + Draining --> Stopping: reject new mutations + Stopping --> Stopped: reverse-order service stop + Failed --> Stopping + Stopped --> [*] +``` + +Bootstrap constructs host resources, creates SystemRoot and the directory, +starts required services in dependency order, and only then advertises the node +as ready to the cluster. Shutdown enters `Draining`, cancels operations, stops +services in reverse order, reclaims local resources such as SHM leases, +withdraws availability, then closes transport. + +Hosts requiring an actor-extension factory install it during bootstrap through +`ActorSystem::new_with_system_actor_factory(...)`; a factory must not replace +an already-running `system/core`. + +System queries must read the authoritative host registry or a versioned +read-only projection. Startup snapshots such as the current `SystemRef` actor +list are caches only, never control-plane truth. + +## Standard services + +| Service | Example operations | Default exposure | +|---|---|---| +| `runtime@1` | node info, health, readiness, operation status | authenticated read-only remote | +| `actors@1` | list, get, stop, spawn | read-only remote; mutation requires admin | +| `metrics@1` | snapshot, recent history | authenticated remote | +| `shm@1` | stats, publish, open, release, reclaim | local only | + +Future cross-process SHM requires peer capability negotiation, credential +binding, and mapping cleanup. The current `in_process` backend must not be +advertised as cross-process shared memory. + +## Forge and self-evolving agents + +High-frequency Forge tool execution remains in an in-process runtime or a +`ToolWorkerActor`; it does not pass through `system/core`. System services are +appropriate for the Forge control plane: runtime/worker creation, tool-schema +discovery, policy inspection, and diagnostics. + +A self-evolving agent must not insert an arbitrary `Arc` into the +active registry. Governed runtime installation follows: + +```text +Propose → Validate → Stage → Health Check + → Atomic Activate(generation) → Drain Old → Commit / Rollback +``` + +Extension manifests additionally declare dependencies, provenance, required +capabilities, and upgrade policy. Generated modules should normally run in an +actor subprocess, WASM runtime, or separate process; SystemRoot governs their +endpoint rather than granting unrestricted host access. + +## Security + +Remote control requires a transport-derived principal, per-operation `Read`, +`Operate`, or `Admin` permission, bounded request bodies, deadlines, and audit +records that omit payloads. Mutable remote operations are denied by default. +In-process plugins remain trusted code; capability injection reduces accidental +coupling, not same-process privilege. + +## Permanent entry point and legacy compatibility + +`system/core` is the permanent standard address of the node control plane, not +a legacy compatibility path. `ActorSystem::system()` and remote SystemRoot +resolution continue to target it as the protocol and service directory evolve. + +Compatibility applies to the existing `SystemMessage`, `SystemResponse`, Python +proxy, and their legacy request shapes. Those messages are adapted to the +service model without changing the permanent root address. + +Legacy mapping is straightforward: runtime messages (`Ping`, `GetNodeInfo`, +`HealthCheck`) target `runtime@1`; actor queries/mutations target `actors@1`; +`GetMetrics` targets `metrics@1`; and `GetShmStats` targets `shm@1`. New +features must not add variants to the legacy enum or use `Extension`. + +## Open product decisions + +- Whether remote callers may invoke `actors/stop`, spawn, or `shm/open`, and + which layer supplies their identity. +- Whether Python/Forge extensions are trusted in-process plugins or require + process isolation. +- Whether optional-service degradation is compatible with node readiness, or + every built-in service must be ready before cluster publication. diff --git a/docs/src/design/system-actors.zh.md b/docs/src/design/system-actors.zh.md new file mode 100644 index 000000000..2d2b4da31 --- /dev/null +++ b/docs/src/design/system-actors.zh.md @@ -0,0 +1,707 @@ +# System Actor 与 System Service 设计 + +> 本文定义 Pulsing 节点控制面的长期架构。文中的“当前实现”描述仓库现状， +> “正式契约”描述后续实现必须保持的稳定语义。本文不包含实施切分或验收计划。 + +## 1. 设计结论 + +Pulsing 将 **System Actor** 正式定义为承载节点基础设施职责的 actor，将 +**System Service** 定义为由节点控制面发现、治理和调用的逻辑能力。二者不是同一个概念： + +- **SystemRoot** 是每个节点唯一且必须存在的控制面入口，长期标准路径固定为 + `system/core`，由 Rust 实现； +- **System Service** 是挂载在 SystemRoot 后的逻辑能力，例如 `actors@1`、 + `metrics@1`、`shm@1`； +- **独立 System Actor** 是某个 service 可选的执行 endpoint，用于隔离状态、进程或语言 + runtime；默认情况下不要求“一个 service 对应一个 actor”； +- **数据面**不经过 SystemRoot。Tensor、SHM payload、工具执行结果和业务流量使用普通 + actor 通信或专用 transport。 + +Pulsing 默认只要求一个 Rust SystemRoot。Python、Forge 和自进化 Agent 可以提供 +extension service 的实现 endpoint，但不能替换 SystemRoot，也不能直接取得完整 +`ActorSystem` 的宿主权限。 + +## 2. 设计理念与出发点 + +### 2.1 为什么需要正式的 System Actor 机制 + +历史上的 `SystemActor` 同时承担 actor 查询、指标、健康检查、Python extension 和 SHM +状态查询。随着能力增加，它逐渐暴露出几个结构性问题： + +- 新能力只能继续增加 `SystemMessage` enum 和中心化 `match`； +- 资源所有者、生命周期管理者和请求处理者混在同一个对象中； +- 控制面可能读取启动时快照，而不是 `ActorSystem` 的权威状态； +- 远程身份、权限、deadline、审计和稳定错误没有统一落点； +- Python/Forge 只能通过无类型的 `Extension { handler, payload }` 扩展； +- 控制面协商与大 payload 传输之间缺少明确边界。 + +System Actor 机制的目的不是增加一层通用 RPC 框架，而是把节点基础设施中本来就存在的 +控制职责放入一个稳定、可治理的模型。 + +### 2.2 核心设计原则 + +#### 原则一：控制面与数据面分离 + +SystemRoot 只处理小型命令、元数据、状态查询和操作编排。大 payload 和高频业务消息始终 +绕过 SystemRoot： + +```text +控制面：publish/open/release descriptor、worker start/stop、policy query +数据面：tensor buffers、SHM mapping、tool execution、stream payload +``` + +SystemRoot 可以协商“如何传输”，但不亲自搬运数据。 + +#### 原则二：宿主拥有资源，service 只持有 capability + +`ActorSystem` 是节点资源的唯一所有者，包括 actor registry、transport、cluster、 +`ShmManager`、metrics store 和 shutdown token。 + +System service 不接收完整的 `ActorSystem`，而是在 bootstrap 时获得最小 capability： + +- actors service 获得 `ActorControl`； +- metrics service 获得只读 actor 统计、metrics 与 performance store； +- SHM service 获得 `ShmManager`； +- runtime service 获得节点 identity 与 lifecycle。 + +capability 注入减少偶然耦合和权限误用，但它不是同一地址空间内的安全沙箱。 + +#### 原则三：入口稳定，能力可扩展 + +`system/core` 是唯一强制存在的根入口。service 使用 +`@` 作为稳定逻辑身份，例如： + +```text +runtime@1 +actors@1 +metrics@1 +shm@1 +forge.runtime@1 +``` + +客户端依赖 service identity 和 operation contract，不依赖某个具体实现是否位于 +SystemRoot 进程内、另一个 actor、Python 子进程或 WASM runtime。 + +#### 原则四：Rust 承载核心路径，动态语言承载受治理扩展 + +SystemRoot、目录、鉴权、生命周期和默认 core services 使用 Rust 实现，避免 GIL、跨语言 +序列化和 Python runtime 故障进入节点关键路径。 + +Python 可以实现 extension service 的业务语义，但应通过独立 actor 或进程 endpoint +接入。高频 Python 工具执行和模型工作负载仍走普通 actor/data plane，不经过 +`system/core`。 + +#### 原则五：扩展必须可治理，而不是任意注入 + +运行中的 Agent 不得直接向活动目录插入任意 `Arc`，也不得获得完整 host +引用。扩展必须具有： + +- 稳定 identity 和版本； +- 来源与完整 manifest； +- 明确的 capability 需求； +- health、deadline、drain 和失败语义； +- 可替换的 generation； +- 独立或受信任的执行边界。 + +#### 原则六：固定标准入口，兼容旧消息协议 + +`system/core` 是长期标准入口，不属于 legacy compatibility 范畴，也没有迁移到其他路径的 +计划。需要兼容的是已有 `SystemMessage`、`SystemResponse` 和 Python proxy。旧消息通过 +adapter 路由到 service；新增能力使用 versioned envelope，不再给 `SystemMessage` +增加无限分支。 + +### 2.3 非目标 + +- 不把所有内部模块都暴露为远程 RPC； +- 不把每个 system service 都强制实现成独立 actor； +- 不让 SystemRoot 承载长时间计算或无限等待； +- 不通过 service directory 实现分布式一致性或跨节点事务； +- 不把进程内 trait object 包装宣传成安全隔离； +- 不允许 Python 或 Agent 替换节点的 Rust SystemRoot。 + +## 3. 核心概念 + +| 概念 | 定义 | 典型实例 | +|---|---|---| +| **System Host** | `ActorSystem` 及其拥有的节点资源，是唯一事实源 | registry、transport、cluster、SHM | +| **SystemRoot** | 节点控制面入口 actor，只负责协议与治理 | `system/core` | +| **System Actor** | 承担节点基础设施职责的 actor | SystemRoot、Python actor creation endpoint | +| **System Service** | 可发现、带版本、受生命周期管理的逻辑能力 | `actors@1`、`shm@1` | +| **Service Directory** | 按 `(namespace, major)` 管理 service 的目录与路由 | core/extension registrations | +| **Manifest** | service 的静态身份、operation、暴露和依赖契约 | `SystemServiceManifest` | +| **Component** | service 的资源启动与停止语义 | SHM cleanup、worker process lifecycle | +| **Handler** | service 的请求处理语义 | list actors、query metrics | +| **Endpoint** | handler 的实际执行位置 | in-process Rust、actor、Python process、WASM | +| **Capability** | Host 授予 service 的最小操作接口 | `ActorControl`、`ShmCapability` | +| **RequestContext** | Root 验证后传给 handler 的调用上下文 | principal、deadline、trace、cancellation | +| **Operation** | 可能跨越一次 mailbox turn 的受管理控制操作 | drain worker、批量 stop、activate generation | + +## 4. 总体架构 + +```mermaid +flowchart TB + Local["本地控制调用方"] + Remote["远程认证调用方"] + Legacy["Legacy SystemMessage"] + + Local --> Root + Remote --> Root + Legacy --> Adapter["Legacy Adapter"] + Adapter --> Root["SystemRoot\nsystem/core\nRust"] + + Root --> Boundary["协议校验 / Auth / Deadline / Audit"] + Boundary --> Directory["SystemServiceDirectory"] + + Directory --> Runtime["runtime@1"] + Directory --> Actors["actors@1"] + Directory --> Metrics["metrics@1"] + Directory --> Shm["shm@1"] + Directory --> Extension["Extension services"] + + subgraph Host["ActorSystem Host：唯一资源所有者"] + Lifecycle["Node Lifecycle"] + Registry["Actor Registry"] + Stores["Metrics / Performance Stores"] + ShmManager["ShmManager"] + Transport["Transport / Cluster"] + end + + Runtime --> Lifecycle + Actors --> Registry + Metrics --> Registry + Metrics --> Stores + Shm --> ShmManager + + Extension --> RustEndpoint["In-process Rust endpoint"] + Extension --> ActorEndpoint["Actor / Process endpoint"] + ActorEndpoint --> Python["Python / Forge / Agent"] + + Data["Tensor / SHM / Tool 数据面"] -. "绕过 SystemRoot" .-> Transport +``` + +### 4.1 SystemRoot + +当前代码中的 `SystemActor` 承担设计中的 SystemRoot 角色。它必须保持轻量，不实现具体 +业务能力。每个请求依次经过： + +1. 识别 legacy message 或 versioned envelope； +2. 校验协议、版本、大小和 deadline； +3. 从 transport 获得可信 caller principal； +4. 根据 manifest 检查 exposure 和 operation access； +5. 从 directory 解析一个 `Ready` service； +6. 构造 `RequestContext` 并调用 handler/endpoint； +7. 记录 trace、audit 和 metrics； +8. 返回稳定 reply 或 error。 + +SystemRoot 只持有 directory、节点生命周期和短期 operation registry，不持有长期业务资源。 + +### 4.2 System Host + +System Host 是 bootstrap capability source，而不是运行时 service locator。它负责： + +- 构造并最终释放节点资源； +- 为每个 service 注入最小 capability； +- 决定 required/optional service； +- 协调节点 readiness、draining 和 shutdown； +- 为 extension endpoint 提供受约束的 actor/process 创建能力。 + +service 不得在 handler 中反向取得完整 host。 + +### 4.3 Service Directory + +Directory 同时承担注册表、生命周期状态表和路由表： + +```rust +pub struct SystemServiceId { + pub namespace: String, + pub major: u16, +} + +pub struct SystemServiceRegistration { + pub manifest: SystemServiceManifest, + pub component: Arc, + pub handler: Arc, +} +``` + +目录遵守以下规则： + +- `(namespace, major)` 全局唯一； +- incompatible change 必须使用新的 major； +- 注册成功不代表可调用，只有 `Ready` service 才可 dispatch； +- required service 启动失败会使 SystemRoot 启动失败； +- optional service 失败不会阻止节点 Ready，但必须可观测； +- 停止顺序与依赖启动顺序相反； +- 运行时更新使用 generation activation，不直接修改活动 entry。 + +### 4.4 Manifest + +Manifest 是静态治理契约，而不是普通 metadata： + +```rust +pub struct SystemServiceManifest { + pub id: SystemServiceId, + pub kind: SystemServiceKind, // Core | Extension + pub exposure: Exposure, // LocalOnly | AuthenticatedRemote + pub operations: Vec, + pub dependencies: Vec, + pub required_capabilities: Vec, + pub source: ServiceSource, + pub upgrade_policy: UpgradePolicy, +} +``` + +operation 至少声明： + +- 稳定名称； +- `Read`、`Operate` 或 `Admin` access class； +- draining 时是否允许； +- request/reply content type 和大小上限； +- 默认 timeout； +- 是否可能返回 `OperationHandle`。 + +### 4.5 Component 与 Handler + +生命周期和请求处理必须分离： + +```rust +#[async_trait] +pub trait SystemComponent { + async fn start(&self) -> Result<()>; + async fn stop(&self) -> Result<()>; + async fn health(&self) -> ServiceHealth; +} + +#[async_trait] +pub trait SystemRequestHandler { + async fn handle( + &self, + request: SystemRequest, + context: RequestContext, + ) -> Result; +} +``` + +无状态 service 可以复用 stateless component。有资源的 service 必须在 component 中实现 +对称的 start/stop；start 部分失败时也必须回滚自身已创建的资源。 + +### 4.6 Endpoint Adapter + +Handler 可以有两种主要实现： + +| Endpoint | 适用场景 | 特性 | +|---|---|---| +| **In-process Rust** | core service、高频控制路径、宿主资源访问 | 延迟低；属于受信任代码 | +| **Actor/Process** | Python、Forge、自进化模块、强隔离 service | 可监督、可超时、可 drain；有一次消息边界 | + +Actor endpoint adapter 将标准 `SystemRequest` 转换为普通 actor message，并把 actor reply +转换回 `SystemReply`。SystemRoot 不感知 endpoint 的编程语言。 + +## 5. 请求协议 + +### 5.1 Versioned Envelope + +正式外部协议使用固定 envelope，service 的 request body 独立版本化： + +```text +SystemRequest { + protocol: "pulsing.system", + protocol_version: "1.0", + request_id: UUID, + target: { namespace: "shm", major: 1 }, + operation: "stats", + deadline_unix_ms: optional u64, + content_type: "application/json", + body: bytes +} +``` + +固定 envelope 让 Root 无需理解 service body，也能统一执行鉴权、限流、deadline、审计和 +路由。service major 独立演进，不要求所有能力共同修改一个全局 enum。 + +### 5.2 RequestContext + +`RequestContext` 只能由可信协议边界创建，至少包含： + +```text +RequestContext { + request_id, + principal, + origin: Local | Remote(node), + deadline, + cancellation, + trace_context, + granted_access +} +``` + +principal 来自 transport authentication，不能由 request body 自报。 + +### 5.3 Reply 与错误 + +成功 reply 回传相同 `request_id`、实际 service version、content type 和 body。错误使用 +稳定 code： + +| Code | 语义 | +|---|---| +| `INVALID_ARGUMENT` | envelope、schema、范围或状态非法 | +| `NOT_FOUND` | service、operation 或资源不存在 | +| `CONFLICT` | 名称冲突或状态转换冲突 | +| `UNAVAILABLE` | service 未 Ready、节点 draining 或依赖失败 | +| `DEADLINE_EXCEEDED` | deadline 到期 | +| `PERMISSION_DENIED` | caller 无权执行 | +| `RESOURCE_EXHAUSTED` | request、并发或资源额度超限 | +| `INTERNAL` | 非预期错误；只返回 trace id，不泄漏内部细节 | + +超过一个合理 mailbox turn 的工作返回 `OperationHandle`，由 runtime service 查询或取消； +SystemRoot 不进行无界等待。 + +## 6. 生命周期 + +### 6.1 节点生命周期 + +```mermaid +stateDiagram-v2 + [*] --> Booting + Booting --> Starting + Starting --> Ready + Starting --> Failed + Ready --> Draining + Ready --> Failed + Draining --> Stopped + Draining --> Failed + Failed --> Draining + Failed --> Stopped +``` + +节点只有在所有 required core services Ready 后才进入 Ready。cluster membership 可以先建立 +连接，但不得在 Ready 前把节点宣告为可承载控制请求或新业务流量。 + +### 6.2 Service 生命周期 + +```text +Registered → Starting → Ready → Stopping → Stopped + └────→ Failed ←─────────┘ +``` + +- `start()` 成功后才进入 Ready； +- start 失败必须回滚失败 component 自身，并逆序停止已 Ready 的依赖者； +- draining 时由 operation manifest 决定哪些只读请求仍可执行； +- `stop()` 失败记录为 service Failed，但不阻止其他 service 尝试停止； +- component health 可以在运行期使 service 降级或退出 Ready。 + +### 6.3 Endpoint 生命周期 + +Actor/process endpoint 的生命周期属于对应 component： + +- component start 创建或解析 endpoint，并完成 health handshake； +- handler 只向已 Ready endpoint dispatch； +- component stop 先停止接收新请求，再等待 in-flight request； +- 超时后取消或终止 endpoint； +- endpoint generation 替换必须先激活新版本，再 drain 旧版本。 + +## 7. 默认 System Actors + +Pulsing 不为每个 service 默认创建一个 actor。默认 actor 集保持最小： + +| Path | 实现 | 启动条件 | 定位 | +|---|---|---|---| +| `system/core` | Rust `SystemActor` / SystemRoot | 每个 `ActorSystem` 必须启动 | 唯一控制面入口、service 路由与生命周期协调 | +| `system/python_actor_service` | Python `PythonActorService` | 使用 Python 全局 runtime 时启动 | Python actor 创建兼容 endpoint | + +### 7.1 `system/core` + +`system/core` 是唯一 required System Actor，也是永久标准入口： + +- 必须由 Rust 实现； +- 不能被 Python、Forge 或用户 actor 替换； +- 不会在新协议成熟后更名或被其他 service path 取代； +- 不承载 tensor、SHM payload 或工具执行； +- Ready 前不发布自身 actor path； +- 对外契约是 system protocol，不是内部 Rust struct。 + +### 7.2 `system/python_actor_service` + +当前 Python runtime 会启动 `system/python_actor_service`，用于远程创建 Python actor 和查询 +Python class registry。它是条件性的兼容 System Actor，不属于 Rust core service directory。 + +长期设计中，它应作为 `actors@1` 的 Python actor-creation endpoint 被治理，而不是成为第二个 +控制面根入口。保留现有路径用于兼容，客户端的新能力发现应通过 service contract 完成。 + +### 7.3 不默认启动的基础设施 Actors + +下列 actor 可以由产品或 extension service 创建，但不属于每个 Pulsing 节点的默认集合： + +- Forge `ToolWorkerActor`； +- Forge event inbox、MCP hub、code-cell registry； +- scheduler、diagnostic collector； +- Agent 生成的 worker、WASM host 或 sandbox process。 + +它们使用普通 actor supervision、placement 和 transport；只有生命周期管理、发现和策略查询 +需要通过 System Service 控制面。 + +## 8. 默认 System Services + +每个节点默认注册四个 required core services： + +| Service | 主要职责 | 默认实现 | 默认暴露 | +|---|---|---|---| +| `runtime@1` | node identity、health、readiness、service/operation 状态 | In-process Rust | 本地；认证远程只读 | +| `actors@1` | actor 查询与受控生命周期操作 | In-process Rust + 可选语言 endpoint | 本地；远程默认只读 | +| `metrics@1` | 节点与 actor 指标快照、近期性能数据 | In-process Rust | 本地；认证远程只读 | +| `shm@1` | region/lease 控制面、统计与回收 | In-process Rust | `LocalOnly` | + +### 8.1 `runtime@1` + +稳定职责： + +- `node_info`：node id、地址、版本和 uptime； +- `health`：节点及 required service 健康状态； +- `ready`：节点是否可接受新控制和业务请求； +- `services/list`、`services/get`：service manifest 与 runtime status； +- `operations/get`、`operations/cancel`：长操作状态； +- `ping`：兼容连通性探测。 + +runtime service 只读取权威 lifecycle 和 directory，不维护第二份节点状态。 + +当前实现已经提供 `node_info`、`health` 和 `ping`；service discovery 与 operation registry +属于正式契约但尚未落地。 + +### 8.2 `actors@1` + +稳定职责： + +- `list`、`get`：查询 host 权威 actor registry； +- `stop`：经策略授权后执行真实 actor stop； +- `spawn`：使用受控 spawn capability 创建 actor； +- `types/list`：查询可创建 actor 类型或 endpoint； +- 管理 Python actor creation endpoint。 + +远程 `list/get` 可以在认证后开放；`spawn/stop` 默认需要 `Admin/Operate` 权限，不能因为调用 +到达 `system/core` 就自动获得授权。 + +现有 `Extension { handler, payload }` 只作为兼容 adapter；新增能力不得继续使用这一无类型 +旁路。 + +### 8.3 `metrics@1` + +稳定职责： + +- `snapshot`：当前 actor、message、lifecycle 和 transport 指标； +- `recent`：从 performance store 读取有界历史； +- `service_health`：各 service 与 endpoint 健康状态； +- `transport`：普通消息、tensor route 与 fallback 指标。 + +metrics service 只做轻量查询和快照，不在 SystemRoot mailbox 中执行昂贵聚合或无界扫描。 + +### 8.4 `shm@1` + +SHM service 是节点共享内存的控制面： + +- `stats`：backend、region、lease 和 bytes； +- `publish/unpublish`：serve 风格的命名 region 生命周期； +- `offer`：message/MPI 风格的一次 rendezvous； +- `open/map-info`：签发映射 descriptor； +- `release`：释放 consumer lease； +- `reclaim`：回收过期 lease 和空 region； +- `capabilities`：报告 backend、locality 和 mapping 能力。 + +SHM payload 永远不经过 SystemRoot。descriptor 必须是不可猜、可撤销、绑定 generation 和 +caller credential 的 capability；当前公开字段与顺序 ID 只适用于 in-process 语义原型， +不能直接作为跨进程安全协议。 + +当前实现提供 in-process `offer`、`publish/open`、`map/release`、机会性 `reclaim` 与 +shutdown clear。System service 执行正常 drain，ActorSystem 作为资源所有者执行最终 clear +兜底；通过 `system/core` 暂时只暴露 `stats`。 + +## 9. System Actor 与 System Service 的关系 + +System Service 是逻辑契约，System Actor 是一种执行载体： + +```text +一个 SystemRoot + ├─ 多个 in-process services + ├─ 多个 actor-backed services + └─ 多个 process/WASM-backed services +``` + +选择原则： + +- 需要直接访问 host capability、延迟敏感、状态简单：使用 in-process Rust service； +- 需要独立监督、故障隔离、动态升级：使用 actor/process endpoint； +- 需要 Python 或 Agent 生成代码：使用隔离 endpoint； +- 高频业务或大数据路径：不建 system service，使用普通 actor/data plane。 + +不应因为某个模块“很重要”就把它变成 System Actor，也不应因为某个能力叫 service 就给它 +创建独立 mailbox。 + +## 10. Python、Forge 与自进化 Agent + +### 10.1 Python + +Python 支持的目标形态是“实现 extension endpoint”，不是“实现 SystemRoot”： + +```mermaid +sequenceDiagram + participant C as Client + participant R as Rust SystemRoot + participant D as Service Directory + participant P as Python Actor Endpoint + + C->>R: versioned SystemRequest + R->>R: auth / deadline / audit + R->>D: resolve extension service + D->>P: actor request + P-->>D: actor reply + D-->>R: SystemReply + R-->>C: versioned reply +``` + +这使核心控制路径不受 GIL 影响。Python endpoint 卡住或崩溃时，Rust Root 仍可超时、熔断、 +drain 或重启对应进程。 + +### 10.2 Forge + +Forge 的高频工具执行继续由进程内 runtime 或 `ToolWorkerActor` 承载，不经过 +`system/core`。适合作为 System Service 的是低频控制能力： + +- runtime/worker 创建与停止； +- tool schema 与 capability 发现； +- sandbox、approval 和 execution policy 查询； +- worker health、diagnostic 与 generation 管理； +- plugin/extension 的受控激活。 + +`forge.runtime@1` 可以是 optional extension service，但不属于 Pulsing core 默认服务。 + +### 10.3 自进化 Agent + +自进化模块采用受治理 generation，而不是原地修改活动 service： + +```text +Propose → Validate → Stage → Health Check + → Atomic Activate(generation) + → Drain Old → Commit / Rollback +``` + +Agent 生成的实现优先运行在 actor 子进程、独立进程或 WASM 中。SystemRoot 保存 manifest、 +generation 和受监督 endpoint，不向生成代码授予任意宿主指针。 + +## 11. SHM 的统一模型 + +serve 风格和 message/MPI 风格并不是两个独立 subsystem，它们共享同一个 region/lease +模型： + +```text +Serve: + publish(name, region) → open(name) → lease → map → release + +Message: + offer(region) → descriptor/lease → send descriptor → map → release +``` + +二者只在 region 的发现方式上不同： + +- serve 通过稳定 name 发现； +- message 通过通信双方传递 descriptor 完成 rendezvous。 + +共同语义包括 generation、lease、TTL、revoke、drain、credential binding 和 cleanup。 +因此 SHM 适合作为 `shm@1` service 管理的 host resource；真正的 mapping 与 tensor bytes +仍属于数据面 backend。 + +## 12. 安全与隔离 + +### 12.1 默认策略 + +- `LocalOnly` 是强制策略，不是文档标签； +- 远程只读请求必须经过 transport authentication； +- 远程 mutating operation 默认拒绝，除非明确授予 `Operate/Admin`； +- principal 来自 transport，不接受 body 自报； +- Root 为每个 operation 执行独立 request size、timeout 和 concurrency limit； +- audit 记录 principal、request id、target、operation、结果和 trace id，不记录敏感 payload。 + +### 12.2 信任边界 + +- Rust core service 是进程内受信任代码； +- capability 限制 API 面，但不能阻止恶意进程内代码读取内存； +- Python/Forge/self-evolving code 默认视为不受信任 extension，优先进程隔离； +- actor mailbox 是并发与故障边界，不自动等于安全沙箱； +- SHM descriptor 必须绑定调用方身份，不能依赖可猜的数字 ID。 + +当前实现已经声明 exposure/access，但在 versioned request context 和 transport principal +落地前，这些字段仍不能被视为完整安全保证。 + +## 13. 长期稳定入口与兼容性 + +### 13.1 长期标准入口 + +`system/core` 是 Pulsing 节点控制面的永久标准地址。`ActorSystem::system()` 和远程 +SystemRoot resolve 始终解析到该路径。versioned envelope、service directory 或 endpoint +实现方式的演进均不会改变这一入口。 + +### 13.2 兼容层 + +以下接口作为已有用户 API 保持兼容： + +- 已有 `SystemMessage`/`SystemResponse` shape； +- Python `SystemActorProxy`； +- `system/python_actor_service` 兼容路径； +- 普通 actor spawn、ask、tell 和 transport API。 + +### 13.3 Legacy Adapter + +旧请求映射到默认 services： + +| Legacy message | Service operation | +|---|---| +| `Ping` | `runtime@1/ping` | +| `GetNodeInfo` | `runtime@1/node_info` | +| `HealthCheck` | `runtime@1/health` | +| `ListActors` | `actors@1/list` | +| `GetActor` | `actors@1/get` | +| `CreateActor` | `actors@1/spawn` 或 Python endpoint | +| `StopActor` | `actors@1/stop` | +| `GetMetrics` | `metrics@1/snapshot` | +| `GetShmStats` | `shm@1/stats` | +| `Extension` | 临时 compatibility adapter | + +Legacy adapter 只能映射旧能力，不能成为新增 service 的注册协议。 + +## 14. 当前实现与正式契约的边界 + +| 能力 | 当前状态 | +|---|---| +| Rust `system/core` | 已实现 | +| Host 权威 capability | 已实现 | +| 四个 core service 的内部 registry | 已实现 | +| Component/handler 分离 | 已实现 | +| Node/service lifecycle 基础状态机 | 已实现 | +| Ready 后发布 `system/core` actor path | 已实现 | +| Bootstrap 失败后的 leave/cancel/host cleanup | 已实现 | +| ActorSystem shutdown 的 SHM 最终回收兜底 | 已实现 | +| SHM 控制面操作时的机会性过期回收 | 已实现 | +| Legacy typed adapter | 已实现 | +| SHM in-process region/lease 语义 | 已实现 | +| Versioned external envelope | 尚未实现 | +| Transport principal 与强制 exposure/access | 尚未实现 | +| Service dependency、optional service、持续 health | 尚未实现 | +| Actor/process endpoint adapter | 尚未实现 | +| Python extension service registration | 尚未实现 | +| Runtime generation activation/rollback | 尚未实现 | +| 跨进程 SHM mapping backend | 尚未实现 | + +该边界用于防止把“已经确定的设计”误写成“已经可用的能力”。 + +## 15. 架构约束 + +后续实现必须保持以下约束： + +1. `ActorSystem` 始终是节点资源唯一所有者； +2. `system/core` 始终是唯一 required 控制面入口； +3. SystemRoot 始终由 Rust 实现，不允许动态语言替换； +4. core service 默认使用 Rust，extension 可以使用 actor/process endpoint； +5. service identity 不依赖 actor path 或语言类型名； +6. 控制面不传输大 payload； +7. manifest 中声明的 exposure/access 必须在 Root 强制执行； +8. service 只通过最小 capability 访问 host； +9. 动态扩展必须通过 generation 和受监督 endpoint； +10. legacy API 可以长期兼容，但新增能力不得扩大 legacy enum。 diff --git a/docs/src/design/tensor-message-transport.zh.md b/docs/src/design/tensor-message-transport.zh.md index a44eab1e3..a15ab167e 100644 --- a/docs/src/design/tensor-message-transport.zh.md +++ b/docs/src/design/tensor-message-transport.zh.md @@ -32,7 +32,8 @@ Pulsing **不解析也不转换** Tensor 的 dtype、shape、stride、字节序 当前实现不提供： - GPU Direct、CUDA IPC 或 GPU 到 GPU 传输； -- 同节点共享内存传输（传输抽象中已经预留）； +- 跨进程/跨容器的同节点共享内存传输（节点级控制面已经引入，但 OS mapping + backend 与 peer capability handshake 尚未启用）； - Tensor schema 编解码、dtype 转换或字节序转换； - checksum、压缩、raw TCP 加密或持久化存储语义； - raw 协议能力协商、request ID、透明重试或 raw 到 HTTP/2 自动 fallback； @@ -97,6 +98,27 @@ pub struct TensorMessage { `SharedMemory` copy model。它们是 Pulsing 的物理传输选择，不是 PulsingQueue 的存储后端。 Python 与 Rust 表示都把 metadata 当作不透明数据。 +## Shared-memory 控制面（基础实现） + +每个 `ActorSystem` 现在持有一个节点级 `ShmManager`，并由 `system/core` +SystemActor 共同拥有其生命周期。它同时覆盖两种上层风格： + +- `offer(bytes, ttl)`：消息式 rendezvous；返回一个只能在 lease 有效期内映射的 + descriptor。 +- `publish(name, bytes)` + `open(name, ttl)`：serve 式命名 region；`unpublish()` + 会阻止新的 open，但会等待已有 lease drain。 + +当前 backend 明确标记为 `in_process`：`map()` 返回 alias 同一份 immutable `Bytes` +storage 的零拷贝 handle。这先固定了 descriptor、range validation、lease、revoke 和 +drain 语义，不会把“同机”误判为“已经可以安全跨进程共享”。可以向 `system/core` +发送 `SystemMessage::GetShmStats` 查询控制面状态，或在 Rust 中通过 +`ActorSystem::shm_manager()` 使用它。 + +传输路由已经从 `Http2Client` 中抽出。它只会在共享内存 backend 被明确声明可用、并且 +locality 满足要求时选择 SHM；目前远端 tensor 仍沿用默认 raw TCP，TLS 或 +`PULSING_TENSOR_TRANSPORT=http2` 仍使用 HTTP/2 兼容路径。因此现有 +`TensorMessage`、`ask/tell` 和 wire protocol 都没有变化。 + ## 数据路径 ```mermaid diff --git a/python/pulsing/core/__init__.py b/python/pulsing/core/__init__.py index 205aa9426..9e6087bdb 100644 --- a/python/pulsing/core/__init__.py +++ b/python/pulsing/core/__init__.py @@ -46,8 +46,7 @@ def incr(self): self.value += 1; return self.value _native_core = sys.modules["pulsing._core"] _HAS_NATIVE_TENSOR_TRANSPORT = all( - hasattr(_native_core, name) - for name in ("TensorMessage", "tensor_transport_stats") + hasattr(_native_core, name) for name in ("TensorMessage", "tensor_transport_stats") ) if _HAS_NATIVE_TENSOR_TRANSPORT: TensorMessage = _native_core.TensorMessage diff --git a/python/pulsing/core/remote.py b/python/pulsing/core/remote.py index b004fab7f..9c2599be5 100644 --- a/python/pulsing/core/remote.py +++ b/python/pulsing/core/remote.py @@ -233,11 +233,7 @@ async def receive(self, msg) -> Any: if isinstance(msg, dict): method, args, kwargs, is_async_call = _unwrap_call(msg) - if ( - not method - or method.startswith("_") - or method in _RESERVED_WIRE_METHODS - ): + if not method or method.startswith("_") or method in _RESERVED_WIRE_METHODS: return _wrap_response(error=f"Invalid method: {method}") _MISSING = object() diff --git a/tests/python/test_tensor_message.py b/tests/python/test_tensor_message.py index 7f7b71240..e2eb0c38b 100644 --- a/tests/python/test_tensor_message.py +++ b/tests/python/test_tensor_message.py @@ -122,10 +122,7 @@ async def test_remote_actor_tensor_roundtrip_uses_selected_transport( assert after["raw_frames_sent"] >= before["raw_frames_sent"] + 2 assert after["active_copy_model"] == "direct_tcp" else: - assert ( - after["http2_fallback_frames"] - >= before["http2_fallback_frames"] + 1 - ) + assert after["http2_fallback_frames"] >= before["http2_fallback_frames"] + 1 assert after["active_copy_model"] == "packed_http2_compatibility" finally: if client is not None: @@ -214,8 +211,7 @@ async def test_raw_tensor_pool_reconnects_only_before_next_request(monkeypatch): after = pul.tensor_transport_stats() assert ( - after["raw_connections_accepted"] - >= before["raw_connections_accepted"] + 2 + after["raw_connections_accepted"] >= before["raw_connections_accepted"] + 2 ) finally: if client is not None: @@ -229,7 +225,9 @@ def __init__(self): async def receive_tensor(self, message): self.calls += 1 - return TensorMessage(message.metadata + b"-reply", message.buffers, message.version) + return TensorMessage( + message.metadata + b"-reply", message.buffers, message.version + ) @pytest.mark.asyncio