diff --git a/Cargo.lock b/Cargo.lock index 52eef1f..945a03c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3300,6 +3300,7 @@ name = "wasm" version = "0.1.0" dependencies = [ "swactor", + "swactor-std", "wasm-bindgen", ] diff --git a/crates/std/Cargo.toml b/crates/std/Cargo.toml index 99e4b2a..5b063c8 100644 --- a/crates/std/Cargo.toml +++ b/crates/std/Cargo.toml @@ -5,8 +5,9 @@ edition = "2024" [features] default = ["getrandom"] -getrandom = ["dep:getrandom"] +getrandom = ["dep:getrandom", "swactor/getrandom"] +wasm = ["swactor/wasm"] [dependencies] -swactor = { path = "../.." } +swactor = { path = "../..", default-features = false } getrandom = { version = "0.2", optional = true } diff --git a/crates/std/src/router.rs b/crates/std/src/router.rs index c0d2f89..3202d6c 100644 --- a/crates/std/src/router.rs +++ b/crates/std/src/router.rs @@ -102,10 +102,20 @@ impl Router { Some(live[idx]) } RoutingStrategy::Random => { - let mut buf = [0u8; 8]; - getrandom::getrandom(&mut buf).expect("getrandom failed"); - let r = u64::from_ne_bytes(buf) as usize; - Some(live[r % live.len()]) + #[cfg(feature = "getrandom")] + { + let mut buf = [0u8; 8]; + getrandom::getrandom(&mut buf).expect("getrandom failed"); + let r = u64::from_ne_bytes(buf) as usize; + Some(live[r % live.len()]) + } + #[cfg(not(feature = "getrandom"))] + { + // Fallback to round-robin when getrandom is unavailable (wasm) + let idx = self.rr_index % live.len(); + self.rr_index = self.rr_index.wrapping_add(1); + Some(live[idx]) + } } RoutingStrategy::Broadcast => None, // handled separately } diff --git a/crates/std/src/runtime_ext.rs b/crates/std/src/runtime_ext.rs index 1f7ab67..f75b0d7 100644 --- a/crates/std/src/runtime_ext.rs +++ b/crates/std/src/runtime_ext.rs @@ -17,6 +17,9 @@ fn get_ext(rt: &Runtime) -> &StdExtension { /// Provides `spawn_named`, `where_is`, `unregister`, and `registered_names` /// via the [`StdExtension`] name registry. pub trait RuntimeNaming { + /// Register a name for an already-spawned actor. Returns `Err` if name is taken. + fn register_name(&self, name: impl Into, addr: ActorAddress) -> Result<(), Error>; + /// Spawn an actor with a registered name, returning its address. fn spawn_named(&self, name: impl Into, actor: A) -> Result; @@ -31,6 +34,10 @@ pub trait RuntimeNaming { } impl RuntimeNaming for Runtime { + fn register_name(&self, name: impl Into, addr: ActorAddress) -> Result<(), Error> { + get_ext(self).name_registry.register(name.into(), addr) + } + fn spawn_named(&self, name: impl Into, actor: A) -> Result { let name = name.into(); let addr = self.spawn(actor)?; diff --git a/crates/wasm/Cargo.toml b/crates/wasm/Cargo.toml index 445b85c..20787f6 100644 --- a/crates/wasm/Cargo.toml +++ b/crates/wasm/Cargo.toml @@ -8,4 +8,5 @@ crate-type = ["cdylib"] [dependencies] swactor = { path = "../..", default-features = false, features = ["wasm"] } +swactor-std = { path = "../std", default-features = false, features = ["wasm"] } wasm-bindgen = "0.2" diff --git a/crates/wasm/src/lib.rs b/crates/wasm/src/lib.rs index 7bf53cb..36fcf4f 100644 --- a/crates/wasm/src/lib.rs +++ b/crates/wasm/src/lib.rs @@ -1,7 +1,10 @@ +use std::sync::Arc; + use wasm_bindgen::prelude::*; -use swactor::actor::{ActorAddress, ActorInterface}; +use swactor::actor::{ActorAddress, ActorExited, ActorInterface}; use swactor::runtime::{Ctx, Inbox, Runtime, RuntimeConfig}; +use swactor_std::{CtxGroups, RuntimeNaming, RuntimeGroups, StdExtension}; // ─── Core JS-facing types ─────────────────────────────────────────────────── @@ -66,13 +69,31 @@ impl WasmInboxBytes { } } +/// Inbox that receives string values (used for death notifications, etc.). +#[wasm_bindgen] +pub struct WasmInboxString { + inner: Inbox, +} + +#[wasm_bindgen] +impl WasmInboxString { + pub fn addr(&self) -> WasmAddr { + WasmAddr(*self.inner.addr()) + } + + /// Poll for the next string. Returns `undefined` when empty. + pub fn try_recv(&self) -> Option { + self.inner.try_recv() + } +} + // ─── Runtime ──────────────────────────────────────────────────────────────── /// The browser-facing swactor runtime. /// -/// Wraps `swactor::Runtime` in single-threaded mode. Actors are spawned via -/// dedicated spawn functions (one per actor type). The runtime is driven by -/// calling `tick()` — either manually or from a `setTimeout(0)` loop. +/// Wraps `swactor::Runtime` in single-threaded mode with StdExtension installed +/// (naming, monitoring, groups). Actors are spawned via dedicated spawn functions +/// (one per actor type). The runtime is driven by calling `tick()`. #[wasm_bindgen] pub struct WasmRuntime { rt: Runtime, @@ -85,7 +106,8 @@ impl WasmRuntime { let rt = Runtime::new(RuntimeConfig { num_threads: 1, ..RuntimeConfig::default() - }); + }) + .with_extension(Arc::new(StdExtension::new())); Self { rt } } @@ -113,6 +135,13 @@ impl WasmRuntime { } } + /// Create an inbox that receives strings. + pub fn new_inbox_string(&self) -> WasmInboxString { + WasmInboxString { + inner: self.rt.new_inbox().expect("new_inbox_string"), + } + } + /// Send a u32 to an actor. Returns false if the address is invalid. pub fn send_u32(&self, addr: &WasmAddr, value: u32) -> bool { self.rt.send_to(addr.0, value).is_ok() @@ -132,6 +161,67 @@ impl WasmRuntime { pub fn uptime_ms(&self) -> f64 { self.rt.stats().uptime_ms as f64 } + + // ─── Naming ───────────────────────────────────────────────────────────── + + /// Register a name for an actor address. Returns false if the name is taken. + pub fn register_name(&self, name: &str, addr: &WasmAddr) -> bool { + self.rt.register_name(name.to_string(), addr.0).is_ok() + } + + /// Look up an actor address by name. Returns undefined if not found. + pub fn where_is(&self, name: &str) -> Option { + self.rt.where_is(name).map(WasmAddr) + } + + /// Unregister a name. Returns the address it was bound to, or undefined. + pub fn unregister_name(&self, name: &str) -> Option { + self.rt.unregister(name).map(WasmAddr) + } + + /// Return all registered actor names as a comma-separated string. + pub fn registered_names(&self) -> String { + self.rt.registered_names().join(",") + } + + // ─── Groups ───────────────────────────────────────────────────────────── + + /// Add an actor to a named group. + pub fn join_group(&self, addr: &WasmAddr, group: &str) { + self.rt.join_group(addr.0, group.to_string()); + } + + /// Remove an actor from a named group. + pub fn leave_group(&self, addr: &WasmAddr, group: &str) { + self.rt.leave_group(addr.0, group); + } + + /// Broadcast a u32 message to all members of a group. Returns count sent. + pub fn publish_to_group_u32(&self, group: &str, msg: u32) -> usize { + self.rt.publish_to(group, msg) + } + + /// Number of actors in a group. + pub fn group_member_count(&self, group: &str) -> usize { + self.rt.group_members(group).len() + } + + /// Return all group names as a comma-separated string. + pub fn group_names(&self) -> String { + self.rt.groups().join(",") + } + + // ─── Stats ────────────────────────────────────────────────────────────── + + /// Total messages processed across all workers. + pub fn total_messages(&self) -> f64 { + self.rt.stats().workers.iter().map(|w| w.messages_processed).sum::() as f64 + } + + /// Total panics across all workers. + pub fn total_panics(&self) -> f64 { + self.rt.stats().workers.iter().map(|w| w.panics).sum::() as f64 + } } impl WasmRuntime { @@ -174,6 +264,52 @@ impl ActorInterface for Relay { } } +/// A sentinel actor that watches a target and reports its death to an inbox. +/// +/// Uses the std monitoring extension (CtxMonitoring::monitor). When the target +/// dies, the sentinel receives a `Down` message and sends the dead actor's +/// string representation to the report inbox, then stops itself. +struct Sentinel { + target: ActorAddress, + report_to: ActorAddress, +} + +impl ActorInterface for Sentinel { + type Incoming = (); + type Response = (); + + fn handle(&mut self, _ctx: &Ctx, _msg: ()) {} + + fn on_start(&mut self, ctx: &Ctx) { + ctx.watch(self.target); + } + + fn on_actor_exit(&mut self, ctx: &Ctx, exited: ActorExited) { + let msg = format!("{}:{:?}", exited.addr, exited.reason); + let _ = ctx.send(self.report_to, msg); + ctx.stop_self(); + } +} + +/// A group member that joins a named group and forwards u32 messages to a report inbox. +struct GroupMember { + group: String, + report_to: ActorAddress, +} + +impl ActorInterface for GroupMember { + type Incoming = u32; + type Response = (); + + fn on_start(&mut self, ctx: &Ctx) { + ctx.join_group(self.group.clone()); + } + + fn handle(&mut self, ctx: &Ctx, msg: u32) { + let _ = ctx.send(self.report_to, msg); + } +} + /// Spawn a counter that accumulates u32 values and reports running totals /// to the given inbox address. #[wasm_bindgen] @@ -197,3 +333,31 @@ pub fn spawn_relay(rt: &WasmRuntime, target: &WasmAddr) -> WasmAddr { .expect("spawn relay"); WasmAddr(addr) } + +/// Spawn a sentinel that watches a target actor and reports its death +/// to the given string inbox. +#[wasm_bindgen] +pub fn spawn_sentinel(rt: &WasmRuntime, target: &WasmAddr, report_to: &WasmInboxString) -> WasmAddr { + let addr = rt + .rt + .spawn(Sentinel { + target: target.0, + report_to: *report_to.inner.addr(), + }) + .expect("spawn sentinel"); + WasmAddr(addr) +} + +/// Spawn a group member that joins the given group and forwards u32 messages +/// to the report inbox. +#[wasm_bindgen] +pub fn spawn_group_member(rt: &WasmRuntime, group: &str, report_to: &WasmAddr) -> WasmAddr { + let addr = rt + .rt + .spawn(GroupMember { + group: group.to_string(), + report_to: report_to.0, + }) + .expect("spawn group_member"); + WasmAddr(addr) +} diff --git a/crates/wasm/test.mjs b/crates/wasm/test.mjs index 7948b6b..7c8c287 100644 --- a/crates/wasm/test.mjs +++ b/crates/wasm/test.mjs @@ -3,6 +3,8 @@ import { WasmAddr, spawn_counter, spawn_relay, + spawn_sentinel, + spawn_group_member, } from "./pkg/wasm.js"; let passed = 0; @@ -153,6 +155,192 @@ function drainInbox(inbox) { rt.free(); } +// ---- naming: register and resolve ----------------------------------------- +{ + console.log("test: naming — register_name and where_is"); + const rt = new WasmRuntime(); + const inbox = rt.new_inbox_u32(); + const c = spawn_counter(rt, inbox.addr()); + rt.tick(); // drain spawn + + const ok = rt.register_name("my_counter", c); + assert(ok, "register_name succeeds"); + + const found = rt.where_is("my_counter"); + assert(found !== undefined, "where_is finds registered actor"); + assertEq(found.toString(), c.toString(), "where_is returns correct address"); + + const notFound = rt.where_is("nonexistent"); + assert(notFound === undefined, "where_is returns undefined for unknown name"); + + found.free(); + inbox.free(); + rt.free(); +} + +// ---- naming: unregister --------------------------------------------------- +{ + console.log("test: naming — unregister_name"); + const rt = new WasmRuntime(); + const inbox = rt.new_inbox_u32(); + const c = spawn_counter(rt, inbox.addr()); + rt.tick(); + + rt.register_name("temp", c); + const prev = rt.unregister_name("temp"); + assert(prev !== undefined, "unregister returns previous address"); + assertEq(prev.toString(), c.toString(), "unregister returns correct address"); + + const gone = rt.where_is("temp"); + assert(gone === undefined, "name no longer resolves after unregister"); + + prev.free(); + inbox.free(); + rt.free(); +} + +// ---- naming: registered_names --------------------------------------------- +{ + console.log("test: naming — registered_names"); + const rt = new WasmRuntime(); + const inbox = rt.new_inbox_u32(); + const a = spawn_counter(rt, inbox.addr()); + const b = spawn_counter(rt, inbox.addr()); + rt.tick(); + + rt.register_name("alpha", a); + rt.register_name("beta", b); + const names = rt.registered_names().split(",").sort(); + assertEq(names, ["alpha", "beta"], "registered_names lists all names"); + + inbox.free(); + rt.free(); +} + +// ---- naming: duplicate name rejected -------------------------------------- +{ + console.log("test: naming — duplicate name rejected"); + const rt = new WasmRuntime(); + const inbox = rt.new_inbox_u32(); + const a = spawn_counter(rt, inbox.addr()); + const b = spawn_counter(rt, inbox.addr()); + rt.tick(); + + const ok1 = rt.register_name("unique", a); + const ok2 = rt.register_name("unique", b); + assert(ok1, "first registration succeeds"); + assert(!ok2, "duplicate registration fails"); + + inbox.free(); + rt.free(); +} + +// ---- groups: join and broadcast ------------------------------------------- +{ + console.log("test: groups — join_group and publish_to_group_u32"); + const rt = new WasmRuntime(); + const inbox = rt.new_inbox_u32(); + const a = spawn_group_member(rt, "workers", inbox.addr()); + const b = spawn_group_member(rt, "workers", inbox.addr()); + rt.tick(); // spawn + on_start (join group) + + assertEq(rt.group_member_count("workers"), 2, "two members in group"); + + rt.publish_to_group_u32("workers", 42); + rt.tick(); // group members receive + rt.tick(); // group members forward to inbox + + const results = drainInbox(inbox); + assertEq(results.length, 2, "both members received broadcast"); + assert(results.every((v) => v === 42), "correct value broadcast"); + + inbox.free(); + rt.free(); +} + +// ---- groups: leave -------------------------------------------------------- +{ + console.log("test: groups — leave_group"); + const rt = new WasmRuntime(); + const inbox = rt.new_inbox_u32(); + const a = spawn_group_member(rt, "pool", inbox.addr()); + const b = spawn_group_member(rt, "pool", inbox.addr()); + rt.tick(); // spawn + on_start + + assertEq(rt.group_member_count("pool"), 2, "two members before leave"); + rt.leave_group(a, "pool"); + assertEq(rt.group_member_count("pool"), 1, "one member after leave"); + + inbox.free(); + rt.free(); +} + +// ---- groups: group_names -------------------------------------------------- +{ + console.log("test: groups — group_names"); + const rt = new WasmRuntime(); + const inbox = rt.new_inbox_u32(); + spawn_group_member(rt, "alpha", inbox.addr()); + spawn_group_member(rt, "beta", inbox.addr()); + rt.tick(); // spawn + join + + const names = rt.group_names().split(",").sort(); + assertEq(names, ["alpha", "beta"], "group_names lists all groups"); + + inbox.free(); + rt.free(); +} + +// ---- watching: sentinel detects death ------------------------------------- +{ + console.log("test: watching — sentinel reports actor death"); + const rt = new WasmRuntime(); + const inbox = rt.new_inbox_u32(); + const deathInbox = rt.new_inbox_string(); + + const target = spawn_counter(rt, inbox.addr()); + const sentinel = spawn_sentinel(rt, target, deathInbox); + rt.tick(); // spawn + on_start (watch) + + rt.stop_actor(target); + // tick to process stop, cleanup, and deliver death notification + for (let i = 0; i < 5; i++) rt.tick(); + + const notification = deathInbox.try_recv(); + assert(notification !== undefined, "sentinel received death notification"); + assert( + typeof notification === "string" && notification.length > 0, + "notification is a non-empty string" + ); + + inbox.free(); + deathInbox.free(); + rt.free(); +} + +// ---- stats: total_messages ------------------------------------------------ +{ + console.log("test: stats — total_messages"); + const rt = new WasmRuntime(); + const inbox = rt.new_inbox_u32(); + const c = spawn_counter(rt, inbox.addr()); + rt.send_u32(c, 1); + rt.send_u32(c, 2); + rt.send_u32(c, 3); + rt.tick(); + assert(rt.total_messages() >= 3, "total_messages counts processed messages"); + inbox.free(); + rt.free(); +} + +// ---- stats: total_panics starts at zero ----------------------------------- +{ + console.log("test: stats — total_panics starts at zero"); + const rt = new WasmRuntime(); + assertEq(rt.total_panics(), 0, "no panics initially"); + rt.free(); +} + // ---- results -------------------------------------------------------------- console.log(`\n${passed} passed, ${failed} failed`); if (failed > 0) process.exit(1); diff --git a/docs/development_history/in-browser/FEATURE_PARITY.md b/docs/development_history/in-browser/FEATURE_PARITY.md new file mode 100644 index 0000000..2ac323b --- /dev/null +++ b/docs/development_history/in-browser/FEATURE_PARITY.md @@ -0,0 +1,75 @@ +# Feature Parity — Development History + +> Stage 4 of the in-browser swactor runtime. Enables swactor-std extensions +> (naming, monitoring, groups) and core actor watching in the wasm crate. + +--- + +## Changes + +### swactor-std wasm compilation + +- Added `wasm` feature to `crates/std/Cargo.toml` (forwards to `swactor/wasm`) +- Changed swactor dependency to `default-features = false`, forwarding `getrandom` + feature when active (`getrandom = ["dep:getrandom", "swactor/getrandom"]`) +- Cfg-gated `getrandom::getrandom()` call in `router.rs` `RoutingStrategy::Random` + — falls back to round-robin when `getrandom` feature is disabled (wasm mode) + +### RuntimeNaming: register_name + +- Added `register_name(name, addr)` method to `RuntimeNaming` trait and impl + — allows registering a name for an already-spawned actor from outside the runtime + — complements existing `spawn_named` (which spawns + registers atomically) + +### Core watching fix: StopSignal death notifications + +- Fixed gap in `worker.rs` tick_all: externally-stopped actors (via `rt.stop_actor()`) + were not added to the `deaths` list, so core WatchRegistry (phase 5b) never fired + for them. Added `deaths.push((addr, ExitReason::Stopped))` when StopSignal is + intercepted (line 737). All 140 existing native tests continue to pass. + +### WasmRuntime: StdExtension + new APIs + +- `WasmRuntime::new()` now installs `StdExtension` automatically +- New inbox type: `WasmInboxString` for receiving string notifications +- **Naming API**: `register_name`, `where_is`, `unregister_name`, `registered_names` +- **Groups API**: `join_group`, `leave_group`, `publish_to_group_u32`, + `group_member_count`, `group_names` +- **Stats API**: `total_messages`, `total_panics` (returned as f64 for JS compat) +- New demo actors: + - `Sentinel` — watches a target via `ctx.watch()`, reports death to string inbox + - `GroupMember` — joins a group on start, forwards u32 messages to report inbox + +### Design Decisions + +| # | Decision | Rationale | +|---|----------|-----------| +| 1 | StdExtension always installed | Browser runtime should have full naming/groups by default | +| 2 | Stats as f64, not u64 | wasm-bindgen maps u64 to BigInt which JSON.stringify rejects | +| 3 | Sentinel actor for watching | Demonstrates core watching from JS without exposing Watch API directly | +| 4 | register_name on RuntimeNaming | Needed for post-spawn registration from JS (no actor context available) | +| 5 | Round-robin fallback for Random routing | wasm mode disables getrandom; graceful degradation preferred | + +## Test Coverage + +22 new assertions across 10 new test scenarios (30 total, from 10): + +| Test | Scenario | +|------|----------| +| naming — register_name and where_is | Register name, resolve, verify not-found returns undefined | +| naming — unregister_name | Unregister returns previous addr, name no longer resolves | +| naming — registered_names | Lists all registered names as CSV | +| naming — duplicate name rejected | Second registration with same name fails | +| groups — join_group and publish_to_group_u32 | Two members receive broadcast message | +| groups — leave_group | Member count decreases after leave | +| groups — group_names | Lists all active group names | +| watching — sentinel reports actor death | Stop target → sentinel receives death notification | +| stats — total_messages | Counts processed messages across workers | +| stats — total_panics starts at zero | Fresh runtime has zero panics | + +## Verification + +- `cargo test -p swactor -p swactor-std` — 157 native tests pass (no regressions) +- `cargo build --target wasm32-unknown-unknown -p wasm` — compiles +- `wasm-pack build --target nodejs` in `crates/wasm/` — builds pkg/ +- `node test.mjs` in `crates/wasm/` — 30/30 tests pass diff --git a/src/worker.rs b/src/worker.rs index 446dc66..884d247 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -735,6 +735,7 @@ impl ActorPool { slot.stopping = true; stats.stops.fetch_add(1, Ordering::Relaxed); slot.mailbox.clear(); + deaths.push((addr, ExitReason::Stopped)); #[cfg(feature = "tracing")] tracing::info!(actor_addr = %addr, "actor.stop_requested"); break;