From 66a852347357018b9d18569f39968bef13de8563 Mon Sep 17 00:00:00 2001 From: Developer Date: Thu, 12 Feb 2026 13:06:25 +0000 Subject: [PATCH] feat: named actor registry with auto-cleanup on death (Cycle 12) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add NameRegistry (String → ActorAddress) to delivery.rs with forward and reverse maps for O(1) lookup and cleanup. Actors can be spawned with names via rt.spawn_named() / ctx.spawn_named(), looked up via where_is(), and names are automatically freed when actors stop or panic. Co-Authored-By: Claude Opus 4.6 --- CLAUDE/notes/progress.md | 34 +++++- src/actor.rs | 27 +++++ src/delivery.rs | 57 +++++++++++ src/runtime.rs | 46 ++++++++- src/worker.rs | 9 ++ tests/runtime_api.rs | 216 ++++++++++++++++++++++++++++++++++++++- 6 files changed, 384 insertions(+), 5 deletions(-) diff --git a/CLAUDE/notes/progress.md b/CLAUDE/notes/progress.md index 755af5b..c7cd6c5 100644 --- a/CLAUDE/notes/progress.md +++ b/CLAUDE/notes/progress.md @@ -2,7 +2,7 @@ ## Current Stage: Phase 1 — Research + First Improvement Cycle -### Status: Cycle 11 COMPLETE +### Status: Cycle 12 COMPLETE ## Plan Overview 1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅ @@ -124,6 +124,38 @@ - `stop_nonexistent_actor_returns_error` — stop on bad address returns Err - **Result**: 82 tests pass, all workspace compiles +### Cycle 12: Named Actor Registry +- **Research**: Studied named actor/service discovery across Erlang (register/2, whereis/1, global, pg), + Actix (Registry, SystemRegistry — TypeId keys), Bastion (hierarchy-based), Ractor (String keys, DashMap, + global static), xactor (TypeId singleton), Akka (Receptionist, ServiceKey[T]) + - Key findings: TypeId keys (Actix/xactor) don't fit swactor's type-erased model; global static + (Ractor) breaks multi-runtime scenarios; Erlang's register/whereis is the gold standard + - Decision: String keys, RwLock (matches existing AddressMap/InboxRegistry pattern), + per-runtime scope, error on collision, auto-unregister on death +- **Implementation**: `NameRegistry` in delivery.rs with forward + reverse maps + - `NameRegistry`: `RwLock>` + `RwLock>` + - Forward map for O(1) name→addr lookup, reverse map for O(1) addr→name cleanup + - Added to `Runtime` as `Arc`, threaded through `TickContext` + - Runtime API: `spawn_named(name, actor)`, `where_is(name)`, `unregister(name)`, `registered_names()` + - Ctx API: `spawn_named(name, actor)`, `where_is(name)` — usable from inside handlers + - `ContextInner` trait extended: `where_is()` + `register_name()` (private, supports both Runtime and WorkerContext) + - Auto-unregister on death: `cleanup_dead` phase calls `name_registry.unregister_by_addr()` for each dead actor + - Name reservation is immediate (before spawn queue push) — prevents TOCTOU race + - Collision returns `Err("Name already registered")` — original binding preserved +- **Tests**: 11 new behavioral tests + - `named_actor_lookup_returns_spawn_address` — spawn_named → where_is roundtrip + - `named_actor_receives_messages_via_lookup` — send to looked-up address works + - `duplicate_name_returns_error` — collision error, original preserved + - `where_is_returns_none_for_unknown_name` — nonexistent name → None + - `name_auto_unregistered_on_actor_death` — stop_actor → name freed + - `name_can_be_reused_after_actor_death` — death → respawn with same name + - `name_auto_unregistered_on_panic` — panic → name freed + - `registered_names_lists_all` — all registered names returned + - `manual_unregister_frees_name_but_actor_lives` — unregister doesn't kill actor + - `ctx_where_is_resolves_inside_handler` — where_is from handler context + - `ctx_spawn_named_registers_from_handler` — spawn_named from handler context +- **Result**: 106 tests pass (99 behavioral + 7 proptest), all workspace compiles, zero warnings + ### Cycle 11: Property-Based Testing (proptest + fuzz extension) - **Research**: Studied testing approaches across tokio (loom), Erlang (PropEr, QuickCheck, Concuerror), Rust property-based testing (proptest vs quickcheck), cargo-fuzz, and actor-specific testing patterns. diff --git a/src/actor.rs b/src/actor.rs index 8e151d8..aa5694c 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -178,6 +178,10 @@ pub trait ContextInner { fn request_stop(&self, addr: ActorAddress); /// Schedule a timer (one-shot or interval). fn schedule_timer(&self, request: TimerRequest); + /// Look up an actor address by registered name. + fn where_is(&self, name: &str) -> Option; + /// Register a name → address mapping. Returns `Err` if the name is taken. + fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), Error>; } /// Actor syscall interface — passed to `ActorInterface::handle()`. @@ -249,6 +253,29 @@ impl<'a> Ctx<'a> { }); } + /// Look up an actor address by its registered name. + /// + /// Returns `None` if no actor is registered under that name. + pub fn where_is(&self, name: &str) -> Option { + self.inner.where_is(name) + } + + /// Spawn a new actor with a registered name. + /// + /// The name is reserved immediately (before the actor starts processing). + /// Returns `Err` if the name is already taken. + pub fn spawn_named( + &self, + name: impl Into, + actor: A, + ) -> Result { + let addr = ActorAddress::new_random(); + self.inner.register_name(name.into(), addr)?; + let boxed: Box = Box::new(Actor::new(actor)); + self.inner.spawn_any(addr, boxed); + Ok(addr) + } + /// Spawn a restartable actor. On panic, recreated via `factory` up to /// `max_restarts` times before permanent poisoning. pub fn spawn_restartable( diff --git a/src/delivery.rs b/src/delivery.rs index 37b12b8..4b49a0d 100644 --- a/src/delivery.rs +++ b/src/delivery.rs @@ -189,6 +189,7 @@ pub(crate) struct TickContext<'a> { pub(crate) placement: &'a Placement, pub(crate) inbox_registry: &'a InboxRegistry, pub(crate) config: &'a RuntimeConfig, + pub(crate) name_registry: &'a NameRegistry, pub(crate) stats_hook: Option<&'a dyn crate::stats::StatsHook>, /// Thread handles for waking parked workers on cross-worker sends. pub(crate) worker_threads: &'a [OnceLock], @@ -198,6 +199,62 @@ pub(crate) struct TickContext<'a> { pub(crate) transport_router: Option<&'a crate::transport::TransportRouter>, } +// ─── Name Registry ────────────────────────────────────────────────────────── + +/// Named actor registry — maps human-readable names to actor addresses. +/// +/// `RwLock` — same pattern as `AddressMap`. Write-rare (spawn/death), +/// read-often (lookup). A reverse map enables O(1) cleanup on actor death. +pub(crate) struct NameRegistry { + names: RwLock>, + reverse: RwLock>, +} + +impl NameRegistry { + pub fn new() -> Self { + Self { + names: RwLock::new(HashMap::new()), + reverse: RwLock::new(HashMap::new()), + } + } + + /// Register a name → address mapping. Returns `Err` if the name is already taken. + pub fn register(&self, name: String, addr: ActorAddress) -> Result<(), crate::Error> { + let mut names = self.names.write().unwrap(); + if names.contains_key(&name) { + return Err(crate::Error::from("Name already registered")); + } + names.insert(name.clone(), addr); + drop(names); + self.reverse.write().unwrap().insert(addr, name); + Ok(()) + } + + /// Look up an actor address by name. + pub fn lookup(&self, name: &str) -> Option { + self.names.read().unwrap().get(name).copied() + } + + /// Unregister a name, returning the address it was bound to. + pub fn unregister(&self, name: &str) -> Option { + let addr = self.names.write().unwrap().remove(name)?; + self.reverse.write().unwrap().remove(&addr); + Some(addr) + } + + /// Remove a name by address (called on actor death for auto-cleanup). + pub fn unregister_by_addr(&self, addr: &ActorAddress) { + if let Some(name) = self.reverse.write().unwrap().remove(addr) { + self.names.write().unwrap().remove(&name); + } + } + + /// Return all registered names. + pub fn registered_names(&self) -> Vec { + self.names.read().unwrap().keys().cloned().collect() + } +} + impl<'a> TickContext<'a> { /// Route a message whose destination is not in the local address map. /// Tries inbox registry, then remote transport, then falls back to inbox error. diff --git a/src/runtime.rs b/src/runtime.rs index 0cf44c1..bc330bd 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -9,7 +9,7 @@ use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopS use crate::channel::{Receiver, Sender}; // Re-export config types so existing code using `runtime::RuntimeConfig` still works pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig}; -use crate::delivery::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId}; +use crate::delivery::{AddressMap, Envelope, InboxRegistry, NameRegistry, Placement, TickContext, WorkerId}; use crate::stats::{StatsHook, WorkerStats}; // Re-export stats types so existing code using `runtime::*` still works pub use crate::stats::{RuntimeStats, WorkerInfo}; @@ -62,6 +62,7 @@ pub struct Runtime { config: RuntimeConfig, address_map: Arc, inbox_registry: Arc, + name_registry: Arc, transfer_txs: Vec>, spawn_txs: Vec)>>, placement: Placement, @@ -119,6 +120,7 @@ impl Runtime { let address_map = Arc::new(AddressMap::with_capacity(config.max_actors)); let inbox_registry = Arc::new(InboxRegistry::new()); + let name_registry = Arc::new(NameRegistry::new()); let mut transfer_txs = Vec::with_capacity(num_workers); let mut spawn_txs = Vec::with_capacity(num_workers); @@ -156,6 +158,7 @@ impl Runtime { config, address_map, inbox_registry, + name_registry, transfer_txs, spawn_txs, placement, @@ -226,6 +229,38 @@ impl Runtime { Ok(addr) } + /// Spawn an actor with a registered name, returns its address. + /// + /// The name is reserved immediately. Returns `Err` if the name is already taken. + pub fn spawn_named( + &self, + name: impl Into, + actor: A, + ) -> Result { + let addr = ActorAddress::new_random(); + self.name_registry.register(name.into(), addr)?; + let worker_id = self.placement.next_worker(); + self.address_map.insert(addr, worker_id); + let boxed: Box = Box::new(Actor::new(actor)); + self.spawn_txs[worker_id.as_usize()].send((addr, boxed)); + Ok(addr) + } + + /// Look up an actor address by its registered name. + pub fn where_is(&self, name: &str) -> Option { + self.name_registry.lookup(name) + } + + /// Unregister a name. Returns the address it was bound to, or `None`. + pub fn unregister(&self, name: &str) -> Option { + self.name_registry.unregister(name) + } + + /// Return all currently registered actor names. + pub fn registered_names(&self) -> Vec { + self.name_registry.registered_names() + } + /// Send a message to an actor address pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { let result = self.send_any(addr, Box::new(msg)); @@ -256,6 +291,7 @@ impl Runtime { placement: &self.placement, inbox_registry: &self.inbox_registry, config: &self.config, + name_registry: &self.name_registry, stats_hook: self.stats_hook.as_deref(), worker_threads: &self.worker_threads, #[cfg(feature = "transport")] @@ -456,4 +492,12 @@ impl ContextInner for Runtime { // Use rt.send_to() with a delay loop instead. eprintln!("swactor: schedule_timer called outside worker context — ignored"); } + + fn where_is(&self, name: &str) -> Option { + self.name_registry.lookup(name) + } + + fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), Error> { + self.name_registry.register(name, addr) + } } diff --git a/src/worker.rs b/src/worker.rs index 37ca80d..3583fc5 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -321,6 +321,7 @@ impl Worker { if !dead.is_empty() { for addr in &dead { tc.address_map.remove(addr); + tc.name_registry.unregister_by_addr(addr); } // Re-publish num_actors after cleanup so stats reflect removal self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); @@ -418,6 +419,14 @@ impl ContextInner for WorkerContext<'_> { fn schedule_timer(&self, request: TimerRequest) { self.timer_requests.borrow_mut().push(request); } + + fn where_is(&self, name: &str) -> Option { + self.tc.name_registry.lookup(name) + } + + fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), crate::Error> { + self.tc.name_registry.register(name, addr) + } } struct ActorSlot { diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index d529fc3..c31ba4f 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -2684,9 +2684,9 @@ impl ActorInterface for TimerStartActor { } /// Actor that schedules a one-shot timer when it receives a Forward message. -struct DelayEchoActor; +struct DelayPingPongActor; -impl ActorInterface for DelayEchoActor { +impl ActorInterface for DelayPingPongActor { type Incoming = Forward; type Response = Done; @@ -2751,7 +2751,7 @@ fn handler_can_schedule_one_shot_timer() { let rt = Runtime::new(RuntimeConfig::default()); let inbox = rt.new_inbox::().unwrap(); - let addr = rt.spawn(DelayEchoActor).unwrap(); + let addr = rt.spawn(DelayPingPongActor).unwrap(); let _ = rt.send_to(addr, Forward { value: 42, reply_to: *inbox.addr() }); rt.tick(); // process Forward, schedule timer (delay=3) @@ -2876,3 +2876,213 @@ fn timer_with_zero_delay_fires_next_tick() { rt.tick(); // timer fires assert!(inbox.try_recv().is_some(), "zero-delay timer fires on next tick"); } + +// ── Named Actor Registry ──────────────────────────────────────────────────── + +/// Given a named actor is spawned, +/// when I look it up by name, +/// then I get the same address that spawn returned. +#[test] +fn named_actor_lookup_returns_spawn_address() { + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn_named("greeter", PingPongActor).unwrap(); + assert_eq!(rt.where_is("greeter"), Some(addr)); +} + +/// Given a named actor exists, +/// when I send a message to the looked-up address, +/// then the actor receives and processes it. +#[test] +fn named_actor_receives_messages_via_lookup() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + let addr = rt.spawn_named("ponger", PingPongActor).unwrap(); + assert_eq!(rt.where_is("ponger"), Some(addr)); + + rt.send_to(addr, Ping { reply_to: *inbox.addr() }).unwrap(); + rt.tick(); + assert!(inbox.try_recv().is_some(), "named actor should process message"); +} + +/// Given a name is already registered, +/// when I try to spawn another actor with the same name, +/// then I get an error and the original binding is preserved. +#[test] +fn duplicate_name_returns_error() { + let rt = Runtime::new(RuntimeConfig::default()); + let first_addr = rt.spawn_named("singleton", PingPongActor).unwrap(); + let result = rt.spawn_named("singleton", PingPongActor); + assert!(result.is_err(), "duplicate name should fail"); + assert_eq!(rt.where_is("singleton"), Some(first_addr), "original binding preserved"); +} + +/// Given no actors are registered, +/// when I look up a nonexistent name, +/// then I get None. +#[test] +fn where_is_returns_none_for_unknown_name() { + let rt = Runtime::new(RuntimeConfig::default()); + assert_eq!(rt.where_is("ghost"), None); +} + +/// Given a named actor is stopped, +/// when the next tick runs cleanup, +/// then the name is automatically unregistered. +#[test] +fn name_auto_unregistered_on_actor_death() { + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn_named("ephemeral", PingPongActor).unwrap(); + rt.tick(); // on_start + + rt.stop_actor(addr).unwrap(); + rt.tick(); // process StopSignal + cleanup + + assert_eq!(rt.where_is("ephemeral"), None, "name should be freed after stop"); +} + +/// Given a named actor died and its name was freed, +/// when I spawn a new actor with the same name, +/// then registration succeeds with a new address. +#[test] +fn name_can_be_reused_after_actor_death() { + let rt = Runtime::new(RuntimeConfig::default()); + let first = rt.spawn_named("worker", PingPongActor).unwrap(); + rt.tick(); + rt.stop_actor(first).unwrap(); + rt.tick(); // cleanup frees the name + + let second = rt.spawn_named("worker", PingPongActor).unwrap(); + assert_ne!(first, second, "new actor should have a different address"); + assert_eq!(rt.where_is("worker"), Some(second)); +} + +/// Given a named actor panics (and is not restartable), +/// when the next tick runs cleanup, +/// then the name is freed. +#[test] +fn name_auto_unregistered_on_panic() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + let _addr = rt.spawn_named("fragile", PanicActor).unwrap(); + rt.tick(); // on_start + + rt.send_to(_addr, PanicMsg).unwrap(); + rt.tick(); // panic → poison → cleanup + + assert_eq!(rt.where_is("fragile"), None, "name freed after panic"); + // Can reuse the name + let _new = rt.spawn_named("fragile", PingPongActor).unwrap(); + assert!(rt.where_is("fragile").is_some()); + drop(inbox); +} + +/// Given multiple named actors are registered, +/// when I call registered_names(), +/// then all names are returned. +#[test] +fn registered_names_lists_all() { + let rt = Runtime::new(RuntimeConfig::default()); + rt.spawn_named("alpha", PingPongActor).unwrap(); + rt.spawn_named("beta", PingPongActor).unwrap(); + rt.spawn_named("gamma", PingPongActor).unwrap(); + + let mut names = rt.registered_names(); + names.sort(); + assert_eq!(names, vec!["alpha", "beta", "gamma"]); +} + +/// Given a named actor exists, +/// when I manually unregister the name, +/// then the name is freed but the actor continues running. +#[test] +fn manual_unregister_frees_name_but_actor_lives() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + let addr = rt.spawn_named("temp-name", PingPongActor).unwrap(); + rt.tick(); // on_start + + let removed = rt.unregister("temp-name"); + assert_eq!(removed, Some(addr)); + assert_eq!(rt.where_is("temp-name"), None, "name freed"); + + // Actor still alive and can receive messages + rt.send_to(addr, Ping { reply_to: *inbox.addr() }).unwrap(); + rt.tick(); + assert!(inbox.try_recv().is_some(), "actor still processes messages"); +} + +/// An actor that looks up a peer by name using ctx.where_is(). +struct NameLookupActor { + target_name: &'static str, + reply_to: ActorAddress, +} + +impl ActorInterface for NameLookupActor { + type Incoming = Ping; + type Response = (); + fn handle(&mut self, ctx: &Ctx, _msg: Ping) { + if let Some(peer) = ctx.where_is(self.target_name) { + ctx.send(self.reply_to, MyAddr(peer)).unwrap(); + } + } +} + +/// Given a named actor exists, +/// when another actor calls ctx.where_is() from inside a handler, +/// then it resolves the correct address. +#[test] +fn ctx_where_is_resolves_inside_handler() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + let target = rt.spawn_named("target", PingPongActor).unwrap(); + + let looker = rt.spawn(NameLookupActor { + target_name: "target", + reply_to: *inbox.addr(), + }).unwrap(); + + rt.tick(); // on_start + rt.send_to(looker, Ping { reply_to: ActorAddress::default() }).unwrap(); + rt.tick(); // handle → where_is → send + rt.tick(); // deliver reply + + let result = inbox.try_recv(); + assert_eq!(result, Some(MyAddr(target)), "ctx.where_is found the named actor"); +} + +/// An actor that spawns a named child using ctx.spawn_named(). +struct NamedSpawnerActor { + reply_to: ActorAddress, +} + +impl ActorInterface for NamedSpawnerActor { + type Incoming = Ping; + type Response = (); + fn handle(&mut self, ctx: &Ctx, _msg: Ping) { + match ctx.spawn_named("child", PingPongActor) { + Ok(addr) => { ctx.send(self.reply_to, MyAddr(addr)).unwrap(); } + Err(_) => {} + } + } +} + +/// Given an actor calls ctx.spawn_named("child", ...), +/// when the child is spawned, +/// then where_is("child") returns the correct address. +#[test] +fn ctx_spawn_named_registers_from_handler() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + let spawner = rt.spawn(NamedSpawnerActor { + reply_to: *inbox.addr(), + }).unwrap(); + + rt.tick(); // on_start + rt.send_to(spawner, Ping { reply_to: ActorAddress::default() }).unwrap(); + rt.tick(); // handle → spawn_named + rt.tick(); // deliver reply + + let child_addr = inbox.try_recv().expect("should receive child address"); + assert_eq!(rt.where_is("child"), Some(child_addr.0), "name registered from handler"); +}