diff --git a/CHANGELOG.md b/CHANGELOG.md index 9221f09..73e034f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -46,6 +46,10 @@ All notable changes to this project will be documented in this file. - Execution-model decision and Phase 1 runtime documentation. - Lifecycle tests proving failed-message retry from committed state and stop/restart preservation of reserved-first FIFO order. +- NanoID-based public actor references, explicit behavior-and-state spawning, + named REPL actor bindings, tab completion, and command history navigation. +- NanoID-based public actor references, explicit behavior-and-state spawning, + named REPL actor bindings, tab completion, and command history navigation. - Runnable Phase 0 experiment pinned to Rune 0.14.2, covering script execution, native calls, async suspension, instruction budgets, frame inspection, @@ -68,6 +72,10 @@ All notable changes to this project will be documented in this file. - Interactive CLI commands now use the versioned protocol dispatcher instead of calling the image API directly. +- Protocol version 2 replaces numeric actor IDs with NanoID actor references + and adds optional target references for message effects. +- Protocol version 2 replaces numeric actor IDs with NanoID actor references + and adds optional target references for message effects. - Actor implementation IDs now identify compiled source/configuration content rather than using sequential process-local numbers. - Expanded the architecture specification with crate ownership and dependency diff --git a/Cargo.lock b/Cargo.lock index 2fd9c30..6318daf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -41,6 +41,12 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + [[package]] name = "blake3" version = "1.8.7" @@ -70,6 +76,21 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + +[[package]] +name = "clipboard-win" +version = "5.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "052af7878466feccda12ba7a6a67d019d712f296809559b47c12c8cc196d3a9a" +dependencies = [ + "error-code", +] + [[package]] name = "codespan-reporting" version = "0.11.1" @@ -77,7 +98,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3538270d33cc669650c4b093848450d380def10c331d38c768e34cac80576e6e" dependencies = [ "termcolor", - "unicode-width", + "unicode-width 0.1.14", ] [[package]] @@ -101,6 +122,39 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" +[[package]] +name = "endian-type" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c34f04666d835ff5d62e058c3995147c06f42fe86ff053337632bca83e42702d" + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "error-code" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5343afd4a8365a643ac588dab4cf234a190c7f6c88c9f6dd6ffe00837661b7" + +[[package]] +name = "fd-lock" +version = "4.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ce92ff622d6dadf7349484f42c93271a0d49b7cc4d466a936405bacbe10aa78" +dependencies = [ + "cfg-if", + "rustix", + "windows-sys 0.59.0", +] + [[package]] name = "find-msvc-tools" version = "0.1.15" @@ -146,6 +200,26 @@ dependencies = [ "windows-result", ] +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + +[[package]] +name = "home" +version = "0.5.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "itoa" version = "1.0.18" @@ -164,6 +238,12 @@ version = "0.2.191" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a" +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + [[package]] name = "log" version = "0.4.35" @@ -230,13 +310,43 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "nanoid" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ffa00dec017b5b1a8b7cf5e2c008bfda1aa7e0697ac1508b491fdf2622fb4d8" +dependencies = [ + "rand", +] + +[[package]] +name = "nibble_vec" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77a5d83df9f36fe23f0c3648c6bbb8b0298bb5f1939c8f2704431371f4b84d43" +dependencies = [ + "smallvec", +] + +[[package]] +name = "nix" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -353,6 +463,15 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "proc-macro2" version = "1.0.107" @@ -371,6 +490,46 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "radix_trie" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c069c179fcdc6a2fe24d8d18305cf085fdbd4f922c041943e203685d6a1c58fd" +dependencies = [ + "endian-type", + "nibble_vec", +] + +[[package]] +name = "rand" +version = "0.8.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e058c7de0b26af77780c769414d6257830bb240f3c38477dbc2c16e5f54d6d4c" +dependencies = [ + "libc", + "rand_chacha", + "rand_core", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom", +] + [[package]] name = "regex-automata" version = "0.4.18" @@ -394,12 +553,14 @@ version = "0.0.0" dependencies = [ "runa-protocol", "runa-server", + "rustyline", ] [[package]] name = "runa-actor" version = "0.0.0" dependencies = [ + "nanoid", "runa-core", ] @@ -566,12 +727,47 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "rustix" +version = "1.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5e76742e5294153d4dde1207dbec5bdd81eef6fabdc0d2118babadabfd858aa" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys 0.61.2", +] + [[package]] name = "rustversion" version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" +[[package]] +name = "rustyline" +version = "15.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ee1e066dc922e513bda599c6ccb5f3bb2b0ea5870a579448f2622993f0a9a2f" +dependencies = [ + "bitflags", + "cfg-if", + "clipboard-win", + "fd-lock", + "home", + "libc", + "log", + "memchr", + "nix", + "radix_trie", + "unicode-segmentation", + "unicode-width 0.2.3", + "utf8parse", + "windows-sys 0.59.0", +] + [[package]] name = "ryu" version = "1.0.23" @@ -767,12 +963,30 @@ version = "1.0.27" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" +[[package]] +name = "unicode-segmentation" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" + [[package]] name = "unicode-width" version = "0.1.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" +[[package]] +name = "unicode-width" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca718471876992059f251e33aafc334346ada08042b7661dd1fe313a9bced557" + +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + [[package]] name = "valuable" version = "0.1.1" @@ -785,13 +999,19 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "winapi-util" version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -809,6 +1029,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -818,6 +1047,70 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + [[package]] name = "zerocopy" version = "0.8.63" diff --git a/README.md b/README.md index ee751e2..11ab684 100644 --- a/README.md +++ b/README.md @@ -67,29 +67,33 @@ intended to connect over the network to this same live image; actors remain in the image regardless of how clients connect. Remote network access requires authentication and authorization, which are not implemented. -The CLI opens an interactive runtime session using the default counter behavior: +The CLI opens an interactive runtime session with the `counter` behavior +published: ```text -runa> spawn 0 -spawned ActorId(1) -runa> spawn 10 -spawned ActorId(2) -runa> send 1 1 2 +runa> let sender = spawn counter 0 +sender = @ +runa> let receiver = spawn counter 10 +receiver = @ +runa> send sender 1 5 --to receiver message queued runa> tick 100 -runa> inspect 1 -runa> inspect 2 +runa> inspect sender +runa> inspect receiver runa> events ``` -The sender commits state `1` and sends a message to actor 2, which commits -state `15`. The CLI also supports publishing Rune source files as named, -versioned behaviors, viewing history, rolling back a behavior reference, -installing the current version on an idle actor, and cooperatively pausing and -resuming or cancelling actor invocations between scheduler ticks. Runtime -users can also supply an explicit state migration closure during installation; -the CLI's install command preserves state as-is. Use `help` in the session for -the complete command list. +Actor references are generated NanoIDs rather than sequential numbers. The +`let` form binds each spawned actor to a local name; references can also be +entered directly with an `@` prefix. The sender commits state `1` and sends a +message to the receiver, which commits state `15`. The REPL provides Tab +completion and Up/Down command history. It also supports publishing Rune source +files as named, versioned behaviors, viewing history, rolling back a behavior +reference, installing the current version on an idle actor, and cooperatively +pausing and resuming or cancelling actor invocations between scheduler ticks. +Runtime users can also supply an explicit state migration closure during +installation; the CLI's install command preserves state as-is. Use `help` in +the session for the complete command list. See [the feasibility report](docs/rune-feasibility.md) for verified Rune capabilities and limitations, and [the runtime guide](docs/runtime.md) for the runtime contract. diff --git a/crates/runa-actor/Cargo.toml b/crates/runa-actor/Cargo.toml index 5bd419c..8bfcb87 100644 --- a/crates/runa-actor/Cargo.toml +++ b/crates/runa-actor/Cargo.toml @@ -4,4 +4,5 @@ version = "0.0.0" edition = "2024" [dependencies] +nanoid = "0.4" runa-core = { path = "../runa-core" } diff --git a/crates/runa-actor/src/lib.rs b/crates/runa-actor/src/lib.rs index 94ced20..e3cc9df 100644 --- a/crates/runa-actor/src/lib.rs +++ b/crates/runa-actor/src/lib.rs @@ -1,9 +1,72 @@ //! Value contracts for the single-worker actor prototype. use runa_core::ImplementationId; +use std::fmt; -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] -pub struct ActorId(pub u64); +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ActorRef([u8; 21]); + +impl ActorRef { + pub fn new() -> Self { + Self::parse(&nanoid::nanoid!()).expect("Nano ID uses the supported alphabet") + } + + pub fn as_str(&self) -> &str { + std::str::from_utf8(&self.0).expect("Nano ID is valid UTF-8") + } + + pub fn parse(value: &str) -> Option { + if value.len() != 21 + || !value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) + { + return None; + } + let mut bytes = [0; 21]; + bytes.copy_from_slice(value.as_bytes()); + Some(Self(bytes)) + } +} + +impl Default for ActorRef { + fn default() -> Self { + Self::new() + } +} + +impl fmt::Debug for ActorRef { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(formatter, "@{}", self.as_str()) + } +} + +impl fmt::Display for ActorRef { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.as_str()) + } +} + +#[cfg(test)] +mod tests { + use super::ActorRef; + + #[test] + fn generated_actor_references_are_parseable_and_display_without_prefix() { + let actor_ref = ActorRef::new(); + let parsed = ActorRef::parse(actor_ref.as_str()).expect("generated reference is valid"); + + assert_eq!(parsed, actor_ref); + assert_eq!(actor_ref.to_string().len(), 21); + assert_eq!(format!("{actor_ref:?}"), format!("@{actor_ref}")); + } + + #[test] + fn parsing_rejects_references_outside_the_url_safe_nanoid_shape() { + assert!(ActorRef::parse("too-short").is_none()); + assert!(ActorRef::parse("abcdefghijklmnopqrstu!").is_none()); + } +} #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub struct MailboxId(pub u64); @@ -12,6 +75,7 @@ pub struct MailboxId(pub u64); pub struct Message { pub kind: i64, pub value: i64, + pub target_ref: Option, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -26,7 +90,7 @@ pub enum Status { #[derive(Debug, Clone)] pub struct ActorSnapshot { - pub id: ActorId, + pub actor_ref: ActorRef, pub mailbox: MailboxId, pub implementation: ImplementationId, pub incarnation: u64, @@ -46,14 +110,14 @@ pub struct ExecutionSnapshot { #[derive(Debug, Clone)] pub enum Event { - Spawned(ActorId), - Committed(ActorId), - Failed(ActorId, String), - Stopped(ActorId), - Restarted(ActorId), - Paused(ActorId), - Resumed(ActorId), - InvocationCancelled(ActorId), - BehaviorInstalled(ActorId, ImplementationId), - StateMigrated(ActorId, i64, i64), + Spawned(ActorRef), + Committed(ActorRef), + Failed(ActorRef, String), + Stopped(ActorRef), + Restarted(ActorRef), + Paused(ActorRef), + Resumed(ActorRef), + InvocationCancelled(ActorRef), + BehaviorInstalled(ActorRef, ImplementationId), + StateMigrated(ActorRef, i64, i64), } diff --git a/crates/runa-image-api/src/lib.rs b/crates/runa-image-api/src/lib.rs index e214025..72cee16 100644 --- a/crates/runa-image-api/src/lib.rs +++ b/crates/runa-image-api/src/lib.rs @@ -3,7 +3,7 @@ use std::num::NonZeroUsize; use std::sync::Arc; -use runa_actor::{ActorId, ActorSnapshot, Event, Message}; +use runa_actor::{ActorRef, ActorSnapshot, Event, Message}; use runa_code::{BehaviorSnapshot, CodeStore}; use runa_core::ImplementationId; use runa_runtime::{Runtime, RuntimeResult}; @@ -74,7 +74,7 @@ impl ImageApi { behavior_name: &str, state: i64, capacity: NonZeroUsize, - ) -> RuntimeResult { + ) -> RuntimeResult { let behavior = self .code .behavior(behavior_name) @@ -82,7 +82,7 @@ impl ImageApi { self.runtime.spawn(behavior.current, state, capacity) } - pub fn send(&mut self, actor: ActorId, message: Message) -> RuntimeResult<()> { + pub fn send(&mut self, actor: ActorRef, message: Message) -> RuntimeResult<()> { self.runtime.send(actor, message) } @@ -92,7 +92,7 @@ impl ImageApi { } } - pub fn inspect(&self, actor: ActorId) -> RuntimeResult { + pub fn inspect(&self, actor: ActorRef) -> RuntimeResult { self.runtime.inspect(actor) } @@ -104,29 +104,29 @@ impl ImageApi { self.runtime.drain_events() } - pub fn stop(&mut self, actor: ActorId) -> RuntimeResult<()> { + pub fn stop(&mut self, actor: ActorRef) -> RuntimeResult<()> { self.runtime.stop(actor) } - pub fn restart(&mut self, actor: ActorId) -> RuntimeResult<()> { + pub fn restart(&mut self, actor: ActorRef) -> RuntimeResult<()> { self.runtime.restart(actor) } - pub fn pause(&mut self, actor: ActorId) -> RuntimeResult<()> { + pub fn pause(&mut self, actor: ActorRef) -> RuntimeResult<()> { self.runtime.pause(actor) } - pub fn resume(&mut self, actor: ActorId) -> RuntimeResult<()> { + pub fn resume(&mut self, actor: ActorRef) -> RuntimeResult<()> { self.runtime.resume(actor) } - pub fn cancel(&mut self, actor: ActorId) -> RuntimeResult<()> { + pub fn cancel(&mut self, actor: ActorRef) -> RuntimeResult<()> { self.runtime.cancel(actor) } pub fn install( &mut self, - actor: ActorId, + actor: ActorRef, behavior_name: &str, ) -> RuntimeResult { self.install_with_migration(actor, behavior_name, Ok) @@ -134,7 +134,7 @@ impl ImageApi { pub fn install_with_migration( &mut self, - actor: ActorId, + actor: ActorRef, behavior_name: &str, migrate: impl FnOnce(i64) -> RuntimeResult, ) -> RuntimeResult { @@ -162,14 +162,21 @@ mod tests { let behavior = image.publish_behavior( "counter", None, - "pub fn handle(state, kind, value) { (state + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", )?; assert_eq!( image.behavior_source("counter", behavior.revision)?, - "pub fn handle(state, kind, value) { (state + value, []) }" + "pub fn handle(state, kind, value, target_ref) { (state + value, []) }" ); let actor = image.spawn("counter", 5, capacity(2))?; - image.send(actor, Message { kind: 0, value: 7 })?; + image.send( + actor, + Message { + kind: 0, + value: 7, + target_ref: None, + }, + )?; image.tick(100, capacity(10)); assert_eq!(image.inspect(actor)?.state, 12); @@ -178,7 +185,7 @@ mod tests { image .events() .iter() - .any(|event| matches!(event, Event::Committed(id) if *id == actor)) + .any(|event| matches!(event, Event::Committed(id) if id == &actor)) ); Ok(()) } @@ -189,14 +196,14 @@ mod tests { let first = image.publish_behavior( "counter", None, - "pub fn handle(state, kind, value) { (state + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", )?; assert!( image .publish_behavior( "counter", None, - "pub fn handle(state, kind, value) { (state - value, []) }" + "pub fn handle(state, kind, value, target_ref) { (state - value, []) }" ) .is_err() ); @@ -210,12 +217,12 @@ mod tests { let first = image.publish_behavior( "counter", None, - "pub fn handle(state, kind, value) { (state + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", )?; image.publish_behavior( "counter", Some(first.revision), - "pub fn handle(state, kind, value) { (state * value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state * value, []) }", )?; let actor = image.spawn("counter", 5, capacity(2))?; let implementation = image.install_with_migration(actor, "counter", |state| { diff --git a/crates/runa-protocol/src/lib.rs b/crates/runa-protocol/src/lib.rs index ba82e97..d648f81 100644 --- a/crates/runa-protocol/src/lib.rs +++ b/crates/runa-protocol/src/lib.rs @@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize}; -pub const PROTOCOL_VERSION: u16 = 1; +pub const PROTOCOL_VERSION: u16 = 2; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -60,18 +60,23 @@ impl Request { return Err(invalid_request("mailbox_capacity must be positive")); } } - Command::Send { actor_id, .. } - | Command::InspectActor { actor_id } - | Command::StopActor { actor_id } - | Command::RestartActor { actor_id } - | Command::PauseActor { actor_id } - | Command::ResumeActor { actor_id } - | Command::CancelActor { actor_id } => validate_actor_id(*actor_id)?, + Command::Send { actor_ref, message } => { + validate_actor_ref(actor_ref)?; + if let Some(target_ref) = &message.target_ref { + validate_actor_ref(target_ref)?; + } + } + Command::InspectActor { actor_ref } + | Command::StopActor { actor_ref } + | Command::RestartActor { actor_ref } + | Command::PauseActor { actor_ref } + | Command::ResumeActor { actor_ref } + | Command::CancelActor { actor_ref } => validate_actor_ref(actor_ref)?, Command::InstallBehavior { - actor_id, + actor_ref, behavior_name, } => { - validate_actor_id(*actor_id)?; + validate_actor_ref(actor_ref)?; validate_name(behavior_name)?; } Command::Tick { @@ -115,7 +120,7 @@ pub enum Command { mailbox_capacity: usize, }, Send { - actor_id: u64, + actor_ref: String, message: MessageDto, }, Tick { @@ -123,27 +128,27 @@ pub enum Command { instruction_budget: usize, }, InspectActor { - actor_id: u64, + actor_ref: String, }, ListActors, GetEvents, StopActor { - actor_id: u64, + actor_ref: String, }, RestartActor { - actor_id: u64, + actor_ref: String, }, PauseActor { - actor_id: u64, + actor_ref: String, }, ResumeActor { - actor_id: u64, + actor_ref: String, }, CancelActor { - actor_id: u64, + actor_ref: String, }, InstallBehavior { - actor_id: u64, + actor_ref: String, behavior_name: String, }, } @@ -163,7 +168,7 @@ pub enum Reply { Behavior(BehaviorDto), Behaviors(Vec), BehaviorSource { source: String }, - ActorSpawned { actor_id: u64 }, + ActorSpawned { actor_ref: String }, Actor(ActorSnapshotDto), Actors(Vec), Events(Vec), @@ -182,36 +187,36 @@ pub struct ProtocolEvent { #[serde(tag = "event", content = "data", rename_all = "snake_case")] pub enum EventDto { Spawned { - actor_id: u64, + actor_ref: String, }, Committed { - actor_id: u64, + actor_ref: String, }, Failed { - actor_id: u64, + actor_ref: String, message: String, }, Stopped { - actor_id: u64, + actor_ref: String, }, Restarted { - actor_id: u64, + actor_ref: String, }, Paused { - actor_id: u64, + actor_ref: String, }, Resumed { - actor_id: u64, + actor_ref: String, }, InvocationCancelled { - actor_id: u64, + actor_ref: String, }, BehaviorInstalled { - actor_id: u64, + actor_ref: String, implementation_id: String, }, StateMigrated { - actor_id: u64, + actor_ref: String, previous_state: i64, current_state: i64, }, @@ -233,17 +238,19 @@ pub struct BehaviorVersionDto { pub implementation_id: String, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct MessageDto { pub kind: i64, pub value: i64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub target_ref: Option, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ActorSnapshotDto { - pub actor_id: u64, + pub actor_ref: String, pub mailbox_id: u64, pub implementation_id: String, pub incarnation: u64, @@ -307,9 +314,15 @@ fn validate_name(name: &str) -> Result<(), ProtocolError> { Ok(()) } -fn validate_actor_id(actor_id: u64) -> Result<(), ProtocolError> { - if actor_id == 0 { - return Err(invalid_request("actor_id must be positive")); +fn validate_actor_ref(actor_ref: &str) -> Result<(), ProtocolError> { + if actor_ref.len() != 21 + || !actor_ref + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) + { + return Err(invalid_request( + "actor_ref must be a 21-character URL-safe identifier", + )); } Ok(()) } @@ -347,8 +360,12 @@ mod tests { #[test] fn request_uses_a_stable_tagged_wire_shape() { let value = serde_json::to_value(request(Command::Send { - actor_id: 7, - message: MessageDto { kind: 2, value: 9 }, + actor_ref: "xK8_pQ2-Lm5_012345678".to_owned(), + message: MessageDto { + kind: 2, + value: 9, + target_ref: None, + }, })) .unwrap(); assert_eq!( @@ -359,7 +376,7 @@ mod tests { "command": { "operation": "send", "parameters": { - "actor_id": 7, + "actor_ref": "xK8_pQ2-Lm5_012345678", "message": { "kind": 2, "value": 9 } } } @@ -395,7 +412,7 @@ mod tests { } #[test] - fn requests_reject_empty_correlation_ids_and_zero_actor_ids() { + fn requests_reject_empty_correlation_ids_and_malformed_actor_refs() { let mut request = request(Command::GetEvents); request.request_id = " ".to_owned(); assert_eq!( @@ -404,7 +421,35 @@ mod tests { ); request.request_id = "req-2".to_owned(); - request.command = Command::StopActor { actor_id: 0 }; + request.command = Command::StopActor { + actor_ref: String::new(), + }; + assert_eq!( + request.validate().unwrap_err().code, + ErrorCode::InvalidRequest + ); + } + + #[test] + fn send_target_reference_is_optional_and_validated_when_present() { + let mut request = request(Command::Send { + actor_ref: "xK8_pQ2-Lm5_012345678".to_owned(), + message: MessageDto { + kind: 2, + value: 9, + target_ref: Some("Hj2_9sM0-pQ4_7aBcD3eF".to_owned()), + }, + }); + assert_eq!(request.validate(), Ok(())); + + request.command = Command::Send { + actor_ref: "xK8_pQ2-Lm5_012345678".to_owned(), + message: MessageDto { + kind: 2, + value: 9, + target_ref: Some("bad".to_owned()), + }, + }; assert_eq!( request.validate().unwrap_err().code, ErrorCode::InvalidRequest @@ -428,7 +473,7 @@ mod tests { let event = ProtocolEvent { version: PROTOCOL_VERSION, event: EventDto::StateMigrated { - actor_id: 3, + actor_ref: "actor_ref_000000000000".to_owned(), previous_state: 4, current_state: 40, }, diff --git a/crates/runa-runtime/src/lib.rs b/crates/runa-runtime/src/lib.rs index a9df515..6661e55 100644 --- a/crates/runa-runtime/src/lib.rs +++ b/crates/runa-runtime/src/lib.rs @@ -6,12 +6,12 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::task::{Context, Wake, Waker}; -use runa_actor::{ActorId, ActorSnapshot, Event, ExecutionSnapshot, MailboxId, Message, Status}; +use runa_actor::{ActorRef, ActorSnapshot, Event, ExecutionSnapshot, MailboxId, Message, Status}; use runa_code::CompiledArtifact; use runa_core::ImplementationId; use runa_rune::{ExperimentResult, Invocation, Slice}; -type WireTransition = (i64, Vec<(u64, i64, i64)>); +type WireTransition = (i64, Vec<(String, i64, i64)>); pub type RuntimeResult = ExperimentResult; struct WakeFlag(AtomicBool); @@ -36,9 +36,9 @@ struct Actor { #[derive(Default)] pub struct Runtime { - actors: BTreeMap, + actors: BTreeMap, implementations: BTreeMap>, - next_actor: u64, + next_mailbox: u64, events: Vec, } @@ -63,21 +63,24 @@ impl Runtime { implementation: ImplementationId, state: i64, capacity: NonZeroUsize, - ) -> ExperimentResult { + ) -> ExperimentResult { if !self.implementations.contains_key(&implementation) { return Err("unknown implementation".into()); } - self.next_actor = self - .next_actor + self.next_mailbox = self + .next_mailbox .checked_add(1) - .ok_or("actor IDs exhausted")?; - let id = ActorId(self.next_actor); + .ok_or("mailbox IDs exhausted")?; + let mut actor_ref = ActorRef::new(); + while self.actors.contains_key(&actor_ref) { + actor_ref = ActorRef::new(); + } self.actors.insert( - id, + actor_ref, Actor { snapshot: ActorSnapshot { - id, - mailbox: MailboxId(id.0), + actor_ref, + mailbox: MailboxId(self.next_mailbox), implementation, incarnation: 1, state, @@ -93,11 +96,11 @@ impl Runtime { paused_from: None, }, ); - self.events.push(Event::Spawned(id)); - Ok(id) + self.events.push(Event::Spawned(actor_ref)); + Ok(actor_ref) } - pub fn send(&mut self, id: ActorId, message: Message) -> ExperimentResult<()> { + pub fn send(&mut self, id: ActorRef, message: Message) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { return Err("actor is not accepting messages".into()); @@ -111,7 +114,7 @@ impl Runtime { Ok(()) } - pub fn inspect(&self, id: ActorId) -> ExperimentResult { + pub fn inspect(&self, id: ActorRef) -> ExperimentResult { let actor = self.actors.get(&id).ok_or("unknown actor")?; let mut snapshot = actor.snapshot.clone(); snapshot.execution = actor.active.as_ref().map(|invocation| { @@ -145,7 +148,7 @@ impl Runtime { .collect() } - pub fn stop(&mut self, id: ActorId) -> ExperimentResult<()> { + pub fn stop(&mut self, id: ActorRef) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; actor.active = None; actor.paused_from = None; @@ -155,7 +158,7 @@ impl Runtime { } /// Pause scheduling between worker ticks without discarding an invocation. - pub fn pause(&mut self, id: ActorId) -> ExperimentResult<()> { + pub fn pause(&mut self, id: ActorRef) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if matches!( actor.snapshot.status, @@ -170,7 +173,7 @@ impl Runtime { } /// Resume from the exact runnable or awaiting state saved by `pause`. - pub fn resume(&mut self, id: ActorId) -> ExperimentResult<()> { + pub fn resume(&mut self, id: ActorRef) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; let status = actor.paused_from.take().ok_or("actor is not paused")?; actor.snapshot.status = status; @@ -181,7 +184,7 @@ impl Runtime { /// Cancel an active handler and requeue its reserved message for retry. /// /// This does not roll back side effects already performed by native code. - pub fn cancel(&mut self, id: ActorId) -> ExperimentResult<()> { + pub fn cancel(&mut self, id: ActorRef) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if actor.active.is_none() { return Err("actor has no active invocation to cancel".into()); @@ -199,7 +202,7 @@ impl Runtime { Ok(()) } - pub fn restart(&mut self, id: ActorId) -> ExperimentResult<()> { + pub fn restart(&mut self, id: ActorRef) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if !matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { return Err("stop the actor before restarting".into()); @@ -221,7 +224,7 @@ impl Runtime { /// Install an immutable implementation between handler invocations. pub fn install( &mut self, - id: ActorId, + id: ActorRef, implementation: ImplementationId, ) -> ExperimentResult<()> { self.install_with_migration(id, implementation, Ok) @@ -231,7 +234,7 @@ impl Runtime { /// operation. A migration error leaves both unchanged. pub fn install_with_migration( &mut self, - id: ActorId, + id: ActorRef, implementation: ImplementationId, migrate: impl FnOnce(i64) -> ExperimentResult, ) -> ExperimentResult<()> { @@ -265,18 +268,18 @@ impl Runtime { std::mem::take(&mut self.events) } - fn fail(&mut self, id: ActorId, error: String) { - let actor = self.actors.get_mut(&id).expect("scheduled actor exists"); + fn fail(&mut self, id: &ActorRef, error: String) { + let actor = self.actors.get_mut(id).expect("scheduled actor exists"); actor.active = None; actor.snapshot.status = Status::Failed(error.clone()); - self.events.push(Event::Failed(id, error)); + self.events.push(Event::Failed(*id, error)); } - fn commit(&mut self, id: ActorId, transition: WireTransition) -> ExperimentResult<()> { + fn commit(&mut self, id: &ActorRef, transition: WireTransition) -> ExperimentResult<()> { let (state, effects) = transition; - let mut counts = BTreeMap::::new(); - for &(target, _, _) in &effects { - let target = ActorId(target); + let mut counts = BTreeMap::::new(); + for (target, _, _) in &effects { + let target = ActorRef::parse(target).ok_or("effect targets invalid actor reference")?; let actor = self .actors .get(&target) @@ -286,24 +289,29 @@ impl Runtime { } let count = counts.entry(target).or_default(); *count += 1; - let retained = usize::from(target != id && actor.snapshot.reserved.is_some()); + let retained = usize::from(target != *id && actor.snapshot.reserved.is_some()); if actor.mailbox.len() + retained + *count > actor.capacity.get() { return Err("effect exceeds mailbox capacity".into()); } } - let actor = self.actors.get_mut(&id).expect("scheduled actor exists"); + let actor = self.actors.get_mut(id).expect("scheduled actor exists"); actor.snapshot.state = state; actor.snapshot.reserved = None; actor.snapshot.status = Status::Idle; actor.active = None; for (target, kind, value) in effects { + let target = ActorRef::parse(&target).expect("validated target"); self.actors - .get_mut(&ActorId(target)) + .get_mut(&target) .expect("validated target") .mailbox - .push_back(Message { kind, value }); + .push_back(Message { + kind, + value, + target_ref: None, + }); } - self.events.push(Event::Committed(id)); + self.events.push(Event::Committed(*id)); Ok(()) } @@ -327,11 +335,18 @@ impl Runtime { match Invocation::start( program.vm(), "handle", - (actor.snapshot.state, message.kind, message.value), + ( + actor.snapshot.state, + message.kind, + message.value, + message + .target_ref + .map_or_else(String::new, |target_ref| target_ref.to_string()), + ), ) { Ok(invocation) => actor.active = Some(invocation), Err(error) => { - self.fail(id, error.to_string()); + self.fail(&id, error.to_string()); continue; } } @@ -350,13 +365,13 @@ impl Runtime { .poll_slice(&mut cx, budget); match outcome { Slice::Completed(transition) => { - if let Err(error) = self.commit(id, transition) { - self.fail(id, error.to_string()); + if let Err(error) = self.commit(&id, transition) { + self.fail(&id, error.to_string()); } } Slice::Awaiting => actor.snapshot.status = Status::Awaiting, Slice::BudgetExhausted => actor.snapshot.status = Status::Running, - Slice::Failed(error) => self.fail(id, error.to_string()), + Slice::Failed(error) => self.fail(&id, error.to_string()), } } } @@ -380,7 +395,7 @@ mod tests { let mut runtime = Runtime::default(); let code = publish( &mut runtime, - "pub fn handle(state, kind, value) { if kind == 1 { (state + 1, [(value, 0, 5)]) } else { (state + value, []) } }", + "pub fn handle(state, kind, value, target_ref) { if kind == 1 { (state + 1, [(target_ref, 0, 5)]) } else { (state + value, []) } }", )?; let first = runtime.spawn(code, 0, capacity(4))?; let second = runtime.spawn(code, 10, capacity(4))?; @@ -388,7 +403,8 @@ mod tests { first, Message { kind: 1, - value: i64::try_from(second.0)?, + value: 0, + target_ref: Some(second), }, )?; for _ in 0..100 { @@ -411,23 +427,48 @@ mod tests { let mut runtime = Runtime::default(); let runaway = publish( &mut runtime, - "pub fn handle(state, kind, value) { loop {} }", + "pub fn handle(state, kind, value, target_ref) { loop {} }", )?; let healthy = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", )?; let first = runtime.spawn(runaway, 0, capacity(1))?; let second = runtime.spawn(healthy, 0, capacity(2))?; - runtime.send(first, Message { kind: 0, value: 1 })?; - runtime.send(second, Message { kind: 0, value: 42 })?; + runtime.send( + first, + Message { + kind: 0, + value: 1, + target_ref: None, + }, + )?; + runtime.send( + second, + Message { + kind: 0, + value: 42, + target_ref: None, + }, + )?; for _ in 0..100 { runtime.tick(capacity(5)); } assert_eq!(runtime.inspect(first)?.status, Status::Running); assert_eq!(runtime.inspect(first)?.state, 0); assert_eq!(runtime.inspect(second)?.state, 42); - assert!(runtime.send(first, Message { kind: 0, value: 2 }).is_err()); + assert!( + runtime + .send( + first, + Message { + kind: 0, + value: 2, + target_ref: None + } + ) + .is_err() + ); runtime.stop(first)?; assert_eq!(runtime.inspect(first)?.status, Status::Stopped); runtime.restart(first)?; @@ -442,10 +483,17 @@ mod tests { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, - "pub fn handle(state, kind, value) { loop {} }", + "pub fn handle(state, kind, value, target_ref) { loop {} }", )?; let actor = runtime.spawn(implementation, 0, capacity(2))?; - runtime.send(actor, Message { kind: 0, value: 1 })?; + runtime.send( + actor, + Message { + kind: 0, + value: 1, + target_ref: None, + }, + )?; runtime.tick(capacity(3)); let running = runtime.inspect(actor)?; @@ -467,7 +515,14 @@ mod tests { runtime.cancel(actor)?; let cancelled = runtime.inspect(actor)?; assert_eq!(cancelled.execution, None); - assert_eq!(cancelled.queued, vec![Message { kind: 0, value: 1 }]); + assert_eq!( + cancelled.queued, + vec![Message { + kind: 0, + value: 1, + target_ref: None + }] + ); Ok(()) } @@ -476,14 +531,15 @@ mod tests { let mut runtime = Runtime::default(); let code = publish( &mut runtime, - "pub fn handle(state, kind, value) { (99, [(value, 0, 1), (999, 0, 2)]) }", + "pub fn handle(state, kind, value, target_ref) { (99, [(target_ref, 0, 1), (\"invalid\", 0, 2)]) }", )?; let actor = runtime.spawn(code, 5, capacity(4))?; runtime.send( actor, Message { kind: 0, - value: i64::try_from(actor.0)?, + value: 0, + target_ref: Some(actor), }, )?; for _ in 0..100 { @@ -539,14 +595,28 @@ mod tests { let mut runtime = Runtime::default(); let code = runtime.publish( CodeStore::default().compile_with_module( - "pub async fn handle(state, kind, value) { if kind == 1 { let x = wait().await; (state + x, []) } else { (state + value, []) } }", + "pub async fn handle(state, kind, value, target_ref) { if kind == 1 { let x = wait().await; (state + x, []) } else { (state + value, []) } }", "test-gate:v1", module, )?, )?; let actor = runtime.spawn(code, 10, capacity(2))?; - runtime.send(actor, Message { kind: 1, value: 0 })?; - runtime.send(actor, Message { kind: 0, value: 3 })?; + runtime.send( + actor, + Message { + kind: 1, + value: 0, + target_ref: None, + }, + )?; + runtime.send( + actor, + Message { + kind: 0, + value: 3, + target_ref: None, + }, + )?; for _ in 0..100 { runtime.tick(capacity(5)); } @@ -554,7 +624,14 @@ mod tests { assert_eq!(pending.status, Status::Awaiting); assert_eq!(pending.state, 10); assert_eq!(pending.queued.len(), 1); - assert_eq!(pending.reserved, Some(Message { kind: 1, value: 0 })); + assert_eq!( + pending.reserved, + Some(Message { + kind: 1, + value: 0, + target_ref: None + }) + ); let polls = gate.polls.load(Ordering::SeqCst); for _ in 0..10 { runtime.tick(capacity(5)); @@ -586,7 +663,14 @@ mod tests { gate.ready.store(false, Ordering::SeqCst); gate.dropped.store(false, Ordering::SeqCst); - runtime.send(actor, Message { kind: 1, value: 0 })?; + runtime.send( + actor, + Message { + kind: 1, + value: 0, + target_ref: None, + }, + )?; for _ in 0..100 { runtime.tick(capacity(5)); } @@ -597,7 +681,14 @@ mod tests { assert_eq!(cancelled.mailbox, pending.mailbox); assert_eq!(cancelled.state, 20); assert_eq!(cancelled.status, Status::Idle); - assert_eq!(cancelled.queued, vec![Message { kind: 1, value: 0 }]); + assert_eq!( + cancelled.queued, + vec![Message { + kind: 1, + value: 0, + target_ref: None + }] + ); assert!(cancelled.reserved.is_none()); Ok(()) } @@ -607,10 +698,17 @@ mod tests { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", )?; let actor = runtime.spawn(implementation, 10, capacity(2))?; - runtime.send(actor, Message { kind: 0, value: 7 })?; + runtime.send( + actor, + Message { + kind: 0, + value: 7, + target_ref: None, + }, + )?; runtime.tick(capacity(1)); let before_pause = runtime.inspect(actor)?; assert_eq!(before_pause.status, Status::Running); @@ -655,12 +753,30 @@ mod tests { let mut runtime = Runtime::default(); let code = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state / value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state / value, []) }", )?; let actor = runtime.spawn(code, 5, capacity(1))?; - runtime.send(actor, Message { kind: 0, value: 0 })?; - assert!(runtime.send(actor, Message { kind: 0, value: 1 }).is_err()); - assert!(runtime.inspect(ActorId(999)).is_err()); + runtime.send( + actor, + Message { + kind: 0, + value: 0, + target_ref: None, + }, + )?; + assert!( + runtime + .send( + actor, + Message { + kind: 0, + value: 1, + target_ref: None + } + ) + .is_err() + ); + assert!(runtime.inspect(ActorRef::new()).is_err()); for _ in 0..100 { runtime.tick(capacity(5)); } @@ -680,14 +796,15 @@ mod tests { let mut runtime = Runtime::default(); let code = publish( &mut runtime, - "pub fn handle(state, kind, value) { (99, [(value, 0, 1), (value, 0, 2)]) }", + "pub fn handle(state, kind, value, target_ref) { (99, [(target_ref, 0, 1), (target_ref, 0, 2)]) }", )?; let actor = runtime.spawn(code, 5, capacity(1))?; runtime.send( actor, Message { kind: 0, - value: i64::try_from(actor.0)?, + value: 0, + target_ref: Some(actor), }, )?; for _ in 0..100 { @@ -704,9 +821,19 @@ mod tests { #[test] fn invalid_transition_shape_fails_without_committing() -> ExperimentResult<()> { let mut runtime = Runtime::default(); - let code = publish(&mut runtime, "pub fn handle(state, kind, value) { 99 }")?; + let code = publish( + &mut runtime, + "pub fn handle(state, kind, value, target_ref) { 99 }", + )?; let actor = runtime.spawn(code, 5, capacity(1))?; - runtime.send(actor, Message { kind: 0, value: 1 })?; + runtime.send( + actor, + Message { + kind: 0, + value: 1, + target_ref: None, + }, + )?; for _ in 0..100 { runtime.tick(capacity(5)); } @@ -720,16 +847,25 @@ mod tests { #[test] fn behavior_installation_waits_for_handler_boundary() -> ExperimentResult<()> { let mut code = CodeStore::default(); - let first = code.compile("pub fn handle(state, kind, value) { (state + value, []) }")?; - let second = - code.compile("pub fn handle(state, kind, value) { (state + value * 10, []) }")?; + let first = + code.compile("pub fn handle(state, kind, value, target_ref) { (state + value, []) }")?; + let second = code.compile( + "pub fn handle(state, kind, value, target_ref) { (state + value * 10, []) }", + )?; let first_id = first.implementation_id; let second_id = second.implementation_id; let mut runtime = Runtime::default(); runtime.publish(first)?; runtime.publish(second)?; let actor = runtime.spawn(first_id, 0, capacity(2))?; - runtime.send(actor, Message { kind: 0, value: 1 })?; + runtime.send( + actor, + Message { + kind: 0, + value: 1, + target_ref: None, + }, + )?; runtime.tick(capacity(1)); let active = runtime.inspect(actor)?; assert_eq!(active.implementation, first_id); @@ -738,7 +874,14 @@ mod tests { runtime.tick(capacity(10)); } assert_eq!(runtime.inspect(actor)?.state, 1); - runtime.send(actor, Message { kind: 0, value: 2 })?; + runtime.send( + actor, + Message { + kind: 0, + value: 2, + target_ref: None, + }, + )?; runtime.install(actor, second_id)?; assert_eq!(runtime.inspect(actor)?.implementation, second_id); for _ in 0..100 { @@ -761,11 +904,11 @@ mod tests { let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", )?; let new_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state * value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state * value, []) }", )?; let actor = runtime.spawn(old_implementation, 7, capacity(2))?; @@ -790,15 +933,29 @@ mod tests { let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", )?; let new_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state * 10 + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state * 10 + value, []) }", )?; let actor = runtime.spawn(old_implementation, 0, capacity(4))?; - runtime.send(actor, Message { kind: 0, value: 1 })?; - runtime.send(actor, Message { kind: 0, value: 2 })?; + runtime.send( + actor, + Message { + kind: 0, + value: 1, + target_ref: None, + }, + )?; + runtime.send( + actor, + Message { + kind: 0, + value: 2, + target_ref: None, + }, + )?; runtime.tick(capacity(1)); assert_eq!(runtime.inspect(actor)?.status, Status::Running); @@ -808,12 +965,30 @@ mod tests { assert_eq!(cancelled.state, 0); assert_eq!( cancelled.queued, - vec![Message { kind: 0, value: 1 }, Message { kind: 0, value: 2 }] + vec![ + Message { + kind: 0, + value: 1, + target_ref: None + }, + Message { + kind: 0, + value: 2, + target_ref: None + } + ] ); assert!(cancelled.reserved.is_none()); runtime.install(actor, new_implementation)?; - runtime.send(actor, Message { kind: 0, value: 3 })?; + runtime.send( + actor, + Message { + kind: 0, + value: 3, + target_ref: None, + }, + )?; for _ in 0..100 { runtime.tick(capacity(10)); } @@ -834,11 +1009,11 @@ mod tests { let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", )?; let new_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state * value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state * value, []) }", )?; let actor = runtime.spawn(old_implementation, 7, capacity(2))?; @@ -869,14 +1044,21 @@ mod tests { let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", )?; let new_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state * value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state * value, []) }", )?; let actor = runtime.spawn(old_implementation, 7, capacity(2))?; - runtime.send(actor, Message { kind: 0, value: 1 })?; + runtime.send( + actor, + Message { + kind: 0, + value: 1, + target_ref: None, + }, + )?; runtime.tick(capacity(1)); let called = Cell::new(false); @@ -908,14 +1090,22 @@ mod tests { let mut runtime = Runtime::default(); let code = runtime.publish( CodeStore::default().compile_with_module( - "fn check() { if fail_once() { 1 / 0 } else { 0 } } pub fn handle(state, kind, value) { let _ = check(); (state + value, []) }", + "fn check() { if fail_once() { 1 / 0 } else { 0 } } pub fn handle(state, kind, value, target_ref) { let _ = check(); (state + value, []) }", "test-fail-once:v1", module, )?, )?; let actor = runtime.spawn(code, 10, capacity(3))?; - let first = Message { kind: 1, value: 2 }; - let second = Message { kind: 1, value: 3 }; + let first = Message { + kind: 1, + value: 2, + target_ref: None, + }; + let second = Message { + kind: 1, + value: 3, + target_ref: None, + }; runtime.send(actor, first)?; runtime.send(actor, second)?; for _ in 0..100 { @@ -927,7 +1117,18 @@ mod tests { assert_eq!(failed.state, 10); assert_eq!(failed.reserved, Some(first)); assert_eq!(failed.queued, vec![second]); - assert!(runtime.send(actor, Message { kind: 1, value: 4 }).is_err()); + assert!( + runtime + .send( + actor, + Message { + kind: 1, + value: 4, + target_ref: None + } + ) + .is_err() + ); assert!( runtime .drain_events() @@ -937,7 +1138,7 @@ mod tests { runtime.restart(actor)?; let restarted = runtime.inspect(actor)?; - assert_eq!(restarted.id, failed.id); + assert_eq!(restarted.actor_ref, failed.actor_ref); assert_eq!(restarted.mailbox, failed.mailbox); assert_eq!(restarted.implementation, failed.implementation); assert_eq!(restarted.incarnation, failed.incarnation + 1); @@ -961,12 +1162,24 @@ mod tests { let mut runtime = Runtime::default(); let code = publish( &mut runtime, - "pub fn handle(state, kind, value) { (state * 10 + value, []) }", + "pub fn handle(state, kind, value, target_ref) { (state * 10 + value, []) }", )?; let actor = runtime.spawn(code, 0, capacity(3))?; - let first = Message { kind: 0, value: 1 }; - let second = Message { kind: 0, value: 2 }; - let third = Message { kind: 0, value: 3 }; + let first = Message { + kind: 0, + value: 1, + target_ref: None, + }; + let second = Message { + kind: 0, + value: 2, + target_ref: None, + }; + let third = Message { + kind: 0, + value: 3, + target_ref: None, + }; runtime.send(actor, first)?; runtime.send(actor, second)?; runtime.send(actor, third)?; @@ -984,7 +1197,18 @@ mod tests { assert_eq!(stopped.state, 0); assert_eq!(stopped.reserved, Some(first)); assert_eq!(stopped.queued, vec![second, third]); - assert!(runtime.send(actor, Message { kind: 0, value: 4 }).is_err()); + assert!( + runtime + .send( + actor, + Message { + kind: 0, + value: 4, + target_ref: None + } + ) + .is_err() + ); runtime.restart(actor)?; let restarted = runtime.inspect(actor)?; diff --git a/crates/runa-server/src/lib.rs b/crates/runa-server/src/lib.rs index 40b2345..4ec3e5f 100644 --- a/crates/runa-server/src/lib.rs +++ b/crates/runa-server/src/lib.rs @@ -6,7 +6,7 @@ use std::num::NonZeroUsize; use std::thread; use std::time::Duration; -use runa_actor::{ActorId, Event, Message, Status}; +use runa_actor::{ActorRef, Event, Message, Status}; use runa_image_api::ImageApi; use runa_protocol::{ ActorSnapshotDto, ActorStatusDto, BehaviorDto, BehaviorVersionDto, Command, ErrorCode, @@ -339,16 +339,18 @@ impl Server { .ok_or_else(|| invalid_request("mailbox_capacity must be positive"))?; self.image .spawn(&behavior_name, state, capacity) - .map(|actor_id| Reply::ActorSpawned { - actor_id: actor_id.0, + .map(|actor_ref| Reply::ActorSpawned { + actor_ref: actor_ref.as_str().to_owned(), }) .map_err(api_error) } - Command::Send { actor_id, message } => self - .image - .send(ActorId(actor_id), message_from_dto(message)) - .map(|()| Reply::Acknowledged) - .map_err(api_error), + Command::Send { actor_ref, message } => { + let actor_ref = parse_actor_ref(&actor_ref)?; + self.image + .send(actor_ref, message_from_dto(message)?) + .map(|()| Reply::Acknowledged) + .map_err(api_error) + } Command::Tick { ticks, instruction_budget, @@ -358,53 +360,67 @@ impl Server { self.image.tick(ticks, budget); Ok(Reply::TickCompleted { ticks }) } - Command::InspectActor { actor_id } => self - .image - .inspect(ActorId(actor_id)) - .map(|snapshot| actor_snapshot_dto(&snapshot)) - .map(Reply::Actor) - .map_err(api_error), + Command::InspectActor { actor_ref } => { + let actor_ref = parse_actor_ref(&actor_ref)?; + self.image + .inspect(actor_ref) + .map(|snapshot| actor_snapshot_dto(&snapshot)) + .map(Reply::Actor) + .map_err(api_error) + } Command::ListActors => Ok(Reply::Actors( self.image.actors().iter().map(actor_snapshot_dto).collect(), )), Command::GetEvents => Ok(Reply::Events( self.image.events().iter().map(event_dto).collect(), )), - Command::StopActor { actor_id } => self - .image - .stop(ActorId(actor_id)) - .map(|()| Reply::Acknowledged) - .map_err(api_error), - Command::RestartActor { actor_id } => self - .image - .restart(ActorId(actor_id)) - .map(|()| Reply::Acknowledged) - .map_err(api_error), - Command::PauseActor { actor_id } => self - .image - .pause(ActorId(actor_id)) - .map(|()| Reply::Acknowledged) - .map_err(api_error), - Command::ResumeActor { actor_id } => self - .image - .resume(ActorId(actor_id)) - .map(|()| Reply::Acknowledged) - .map_err(api_error), - Command::CancelActor { actor_id } => self - .image - .cancel(ActorId(actor_id)) - .map(|()| Reply::Acknowledged) - .map_err(api_error), + Command::StopActor { actor_ref } => { + let actor_ref = parse_actor_ref(&actor_ref)?; + self.image + .stop(actor_ref) + .map(|()| Reply::Acknowledged) + .map_err(api_error) + } + Command::RestartActor { actor_ref } => { + let actor_ref = parse_actor_ref(&actor_ref)?; + self.image + .restart(actor_ref) + .map(|()| Reply::Acknowledged) + .map_err(api_error) + } + Command::PauseActor { actor_ref } => { + let actor_ref = parse_actor_ref(&actor_ref)?; + self.image + .pause(actor_ref) + .map(|()| Reply::Acknowledged) + .map_err(api_error) + } + Command::ResumeActor { actor_ref } => { + let actor_ref = parse_actor_ref(&actor_ref)?; + self.image + .resume(actor_ref) + .map(|()| Reply::Acknowledged) + .map_err(api_error) + } + Command::CancelActor { actor_ref } => { + let actor_ref = parse_actor_ref(&actor_ref)?; + self.image + .cancel(actor_ref) + .map(|()| Reply::Acknowledged) + .map_err(api_error) + } Command::InstallBehavior { - actor_id, + actor_ref, behavior_name, - } => self - .image - .install(ActorId(actor_id), &behavior_name) - .map(|implementation_id| Reply::BehaviorInstalled { - implementation_id: implementation_id.to_string(), - }) - .map_err(api_error), + } => { + let actor_ref = parse_actor_ref(&actor_ref)?; + self.image + .install(actor_ref, &behavior_name) + .map(|implementation_id| Reply::BehaviorInstalled { + implementation_id: implementation_id.to_string(), + }) + .map_err(api_error) + } } } } @@ -427,7 +443,7 @@ fn behavior_dto(behavior: runa_code::BehaviorSnapshot) -> BehaviorDto { fn actor_snapshot_dto(snapshot: &runa_actor::ActorSnapshot) -> ActorSnapshotDto { ActorSnapshotDto { - actor_id: snapshot.id.0, + actor_ref: snapshot.actor_ref.as_str().to_owned(), mailbox_id: snapshot.mailbox.0, implementation_id: snapshot.implementation.to_string(), incarnation: snapshot.incarnation, @@ -454,49 +470,57 @@ fn message_dto(message: Message) -> MessageDto { MessageDto { kind: message.kind, value: message.value, + target_ref: message + .target_ref + .map(|actor_ref| actor_ref.as_str().to_owned()), } } -fn message_from_dto(message: MessageDto) -> Message { - Message { +fn message_from_dto(message: MessageDto) -> Result { + Ok(Message { kind: message.kind, value: message.value, - } + target_ref: message + .target_ref + .as_deref() + .map(parse_actor_ref) + .transpose()?, + }) } fn event_dto(event: &Event) -> EventDto { match event { - Event::Spawned(actor_id) => EventDto::Spawned { - actor_id: actor_id.0, + Event::Spawned(actor_ref) => EventDto::Spawned { + actor_ref: actor_ref.as_str().to_owned(), }, - Event::Committed(actor_id) => EventDto::Committed { - actor_id: actor_id.0, + Event::Committed(actor_ref) => EventDto::Committed { + actor_ref: actor_ref.as_str().to_owned(), }, - Event::Failed(actor_id, message) => EventDto::Failed { - actor_id: actor_id.0, + Event::Failed(actor_ref, message) => EventDto::Failed { + actor_ref: actor_ref.as_str().to_owned(), message: message.clone(), }, - Event::Stopped(actor_id) => EventDto::Stopped { - actor_id: actor_id.0, + Event::Stopped(actor_ref) => EventDto::Stopped { + actor_ref: actor_ref.as_str().to_owned(), }, - Event::Restarted(actor_id) => EventDto::Restarted { - actor_id: actor_id.0, + Event::Restarted(actor_ref) => EventDto::Restarted { + actor_ref: actor_ref.as_str().to_owned(), }, - Event::Paused(actor_id) => EventDto::Paused { - actor_id: actor_id.0, + Event::Paused(actor_ref) => EventDto::Paused { + actor_ref: actor_ref.as_str().to_owned(), }, - Event::Resumed(actor_id) => EventDto::Resumed { - actor_id: actor_id.0, + Event::Resumed(actor_ref) => EventDto::Resumed { + actor_ref: actor_ref.as_str().to_owned(), }, - Event::InvocationCancelled(actor_id) => EventDto::InvocationCancelled { - actor_id: actor_id.0, + Event::InvocationCancelled(actor_ref) => EventDto::InvocationCancelled { + actor_ref: actor_ref.as_str().to_owned(), }, - Event::BehaviorInstalled(actor_id, implementation_id) => EventDto::BehaviorInstalled { - actor_id: actor_id.0, + Event::BehaviorInstalled(actor_ref, implementation_id) => EventDto::BehaviorInstalled { + actor_ref: actor_ref.as_str().to_owned(), implementation_id: implementation_id.to_string(), }, - Event::StateMigrated(actor_id, previous_state, current_state) => EventDto::StateMigrated { - actor_id: actor_id.0, + Event::StateMigrated(actor_ref, previous_state, current_state) => EventDto::StateMigrated { + actor_ref: actor_ref.as_str().to_owned(), previous_state: *previous_state, current_state: *current_state, }, @@ -533,6 +557,10 @@ fn invalid_request(message: impl Into) -> ProtocolError { ProtocolError::new(ErrorCode::InvalidRequest, message) } +fn parse_actor_ref(value: &str) -> Result { + ActorRef::parse(value).ok_or_else(|| invalid_request("actor_ref must be a valid Nano ID")) +} + #[cfg(test)] mod tests { use super::*; @@ -552,7 +580,8 @@ mod tests { let response = server.dispatch(request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: None, - source: "pub fn handle(state, kind, value) { (state + value, []) }".to_owned(), + source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }" + .to_owned(), })); match response.result.unwrap() { Reply::Behavior(behavior) => behavior, @@ -572,15 +601,20 @@ mod tests { state: 5, mailbox_capacity: 2, })); - let actor_id = match spawned.result.unwrap() { - Reply::ActorSpawned { actor_id } => actor_id, + let actor_ref = match spawned.result.unwrap() { + Reply::ActorSpawned { actor_ref } => actor_ref, other => panic!("unexpected reply: {other:?}"), }; - assert_eq!(actor_id, 1); + assert_eq!(actor_ref.len(), 21); + assert!(!actor_ref.parse::().is_ok()); let sent = server.dispatch(request(Command::Send { - actor_id, - message: MessageDto { kind: 0, value: 7 }, + actor_ref: actor_ref.clone(), + message: MessageDto { + kind: 0, + value: 7, + target_ref: None, + }, })); assert_eq!(sent.result, Ok(Reply::Acknowledged)); server.dispatch(request(Command::Tick { @@ -588,7 +622,9 @@ mod tests { instruction_budget: 10, })); - let inspected = server.dispatch(request(Command::InspectActor { actor_id })); + let inspected = server.dispatch(request(Command::InspectActor { + actor_ref: actor_ref.clone(), + })); match inspected.result.unwrap() { Reply::Actor(snapshot) => assert_eq!(snapshot.state, 12), other => panic!("unexpected reply: {other:?}"), @@ -597,7 +633,7 @@ mod tests { let events = server.dispatch(request(Command::GetEvents)); match events.result.unwrap() { Reply::Events(events) => assert!(events.iter().any( - |event| matches!(event, EventDto::Committed { actor_id: id } if *id == actor_id) + |event| matches!(event, EventDto::Committed { actor_ref: id } if id == &actor_ref) )), other => panic!("unexpected reply: {other:?}"), } @@ -619,14 +655,16 @@ mod tests { ErrorCode::UnsupportedVersion ); - let missing = server.dispatch(request(Command::InspectActor { actor_id: 42 })); + let missing = server.dispatch(request(Command::InspectActor { + actor_ref: "000000000000000000000".to_owned(), + })); assert_eq!(missing.result.unwrap_err().code, ErrorCode::NotFound); publish(&mut server); let conflict = server.dispatch(request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: Some(99), - source: "pub fn handle(state, kind, value) { (state, []) }".to_owned(), + source: "pub fn handle(state, kind, value, target_ref) { (state, []) }".to_owned(), })); assert_eq!(conflict.result.unwrap_err().code, ErrorCode::Conflict); } @@ -638,7 +676,8 @@ mod tests { let updated = server.dispatch(request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: Some(first.revision), - source: "pub fn handle(state, kind, value) { (state * value, []) }".to_owned(), + source: "pub fn handle(state, kind, value, target_ref) { (state * value, []) }" + .to_owned(), })); assert!(matches!(updated.result, Ok(Reply::Behavior(_)))); @@ -649,11 +688,12 @@ mod tests { assert_eq!( source.result, Ok(Reply::BehaviorSource { - source: "pub fn handle(state, kind, value) { (state + value, []) }".to_owned() + source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }" + .to_owned() }) ); - let actor_id = match server + let actor_ref = match server .dispatch(request(Command::Spawn { behavior_name: "counter".to_owned(), state: 3, @@ -662,11 +702,11 @@ mod tests { .result .unwrap() { - Reply::ActorSpawned { actor_id } => actor_id, + Reply::ActorSpawned { actor_ref } => actor_ref, other => panic!("unexpected reply: {other:?}"), }; let installed = server.dispatch(request(Command::InstallBehavior { - actor_id, + actor_ref, behavior_name: "counter".to_owned(), })); assert!(matches!( @@ -680,7 +720,8 @@ mod tests { let publish_request = request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: None, - source: "pub fn handle(state, kind, value) { (state + value, []) }".to_owned(), + source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }" + .to_owned(), }); let mut spawn_request = request(Command::Spawn { behavior_name: "counter".to_owned(), @@ -717,7 +758,10 @@ mod tests { vec!["test-request", "spawn-request", "list-request"] ); assert!(matches!(responses[0].result, Ok(Reply::Behavior(_)))); - assert_eq!(responses[1].result, Ok(Reply::ActorSpawned { actor_id: 1 })); + assert!(matches!( + &responses[1].result, + Ok(Reply::ActorSpawned { actor_ref }) if actor_ref.len() == 21 + )); assert!(matches!( &responses[2].result, Ok(Reply::Actors(actors)) if actors.len() == 1 && actors[0].state == 4 diff --git a/crates/runa-server/tests/local_client.rs b/crates/runa-server/tests/local_client.rs index 59201be..08e7591 100644 --- a/crates/runa-server/tests/local_client.rs +++ b/crates/runa-server/tests/local_client.rs @@ -23,7 +23,8 @@ fn local_client_launches_server_and_operates_the_same_live_image() { .request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: None, - source: "pub fn handle(state, kind, value) { (state + value, []) }".to_owned(), + source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }" + .to_owned(), }) .unwrap(); assert_eq!(published.request_id, "runa-client-1"); @@ -37,12 +38,20 @@ fn local_client_launches_server_and_operates_the_same_live_image() { }) .unwrap(); assert_eq!(spawned.request_id, "runa-client-2"); - assert_eq!(spawned.result, Ok(Reply::ActorSpawned { actor_id: 1 })); + let actor_ref = match spawned.result { + Ok(Reply::ActorSpawned { actor_ref }) => actor_ref, + other => panic!("unexpected spawn response: {other:?}"), + }; + assert_eq!(actor_ref.len(), 21); client .request(Command::Send { - actor_id: 1, - message: MessageDto { kind: 0, value: 7 }, + actor_ref: actor_ref.clone(), + message: MessageDto { + kind: 0, + value: 7, + target_ref: None, + }, }) .unwrap(); client @@ -51,9 +60,7 @@ fn local_client_launches_server_and_operates_the_same_live_image() { instruction_budget: 10, }) .unwrap(); - let inspected = client - .request(Command::InspectActor { actor_id: 1 }) - .unwrap(); + let inspected = client.request(Command::InspectActor { actor_ref }).unwrap(); assert!(matches!( inspected.result, Ok(Reply::Actor(actor)) if actor.state == 12 diff --git a/crates/runa/Cargo.toml b/crates/runa/Cargo.toml index efffb7e..751f324 100644 --- a/crates/runa/Cargo.toml +++ b/crates/runa/Cargo.toml @@ -6,3 +6,4 @@ edition = "2024" [dependencies] runa-protocol = { path = "../runa-protocol" } runa-server = { path = "../runa-server" } +rustyline = "15.0" diff --git a/crates/runa/src/main.rs b/crates/runa/src/main.rs index ac10122..23616c2 100644 --- a/crates/runa/src/main.rs +++ b/crates/runa/src/main.rs @@ -1,19 +1,144 @@ -use std::io::{self, BufRead, Write}; +use std::collections::BTreeMap; +#[cfg(test)] +use std::io::BufRead; +use std::io::{self, Write}; use std::num::NonZeroUsize; use runa_protocol::{Command, MessageDto, PROTOCOL_VERSION, Reply, Request, Response}; use runa_server::Server; +use rustyline::completion::{Completer, Pair}; +use rustyline::error::ReadlineError; +use rustyline::highlight::Highlighter; +use rustyline::hint::Hinter; +use rustyline::history::DefaultHistory; +use rustyline::validate::Validator; +use rustyline::{Context, Editor, Helper}; const MAILBOX_CAPACITY: usize = 64; const DEFAULT_TICK_BUDGET: usize = 32; -const DEFAULT_BEHAVIOR: &str = "counter"; +const COUNTER_BEHAVIOR: &str = "counter"; + +#[derive(Default)] +struct ReplState { + bindings: BTreeMap, +} + +#[derive(Default)] +struct ReplHelper { + behaviors: Vec, + bindings: Vec<(String, String)>, +} + +impl ReplHelper { + fn set_context(&mut self, behaviors: Vec, bindings: &BTreeMap) { + self.behaviors = behaviors; + self.bindings = bindings + .iter() + .map(|(name, actor_ref)| (name.clone(), actor_ref.clone())) + .collect(); + } + + fn candidates(&self, line: &str, position: usize) -> Vec { + const COMMANDS: &[&str] = &[ + "actors", + "behaviors", + "cancel", + "events", + "exit", + "help", + "history", + "inspect", + "install", + "let", + "pause", + "publish", + "quit", + "restart", + "resume", + "rollback", + "send", + "spawn", + "stop", + "tick", + ]; + let start = line[..position] + .rfind(char::is_whitespace) + .map_or(0, |index| index + 1); + let previous: Vec<_> = line[..start].split_whitespace().collect(); + let options: Vec = match previous.as_slice() { + [] => COMMANDS + .iter() + .map(|command| (*command).to_owned()) + .collect(), + ["let", _, "="] => vec!["spawn".to_owned()], + ["spawn"] | ["let", _, "=", "spawn"] => self.behaviors.clone(), + ["let", _, "=", "spawn", _] => Vec::new(), + ["send"] + | ["inspect"] + | ["stop"] + | ["restart"] + | ["pause"] + | ["resume"] + | ["cancel"] + | ["install"] => self.actor_candidates(), + ["send", _, _, _] => vec!["--to".to_owned()], + ["send", _, _, _, "--to"] => self.actor_candidates(), + ["install", _] => self.behaviors.clone(), + _ => Vec::new(), + }; + let prefix = &line[start..position]; + options + .into_iter() + .filter(|option| option.starts_with(prefix)) + .collect() + } + + fn actor_candidates(&self) -> Vec { + self.bindings + .iter() + .flat_map(|(name, actor_ref)| [name.clone(), format!("@{actor_ref}")]) + .collect() + } +} + +impl Completer for ReplHelper { + type Candidate = Pair; + + fn complete( + &self, + line: &str, + position: usize, + _context: &Context<'_>, + ) -> rustyline::Result<(usize, Vec)> { + let start = line[..position] + .rfind(char::is_whitespace) + .map_or(0, |index| index + 1); + let candidates = self + .candidates(line, position) + .into_iter() + .map(|candidate| Pair { + display: candidate.clone(), + replacement: candidate, + }) + .collect(); + Ok((start, candidates)) + } +} + +impl Hinter for ReplHelper { + type Hint = String; +} + +impl Highlighter for ReplHelper {} +impl Validator for ReplHelper {} +impl Helper for ReplHelper {} fn main() -> Result<(), Box> { let mut server = Server::new(); let behavior = match dispatch( &mut server, Command::PublishBehavior { - name: DEFAULT_BEHAVIOR.to_owned(), + name: COUNTER_BEHAVIOR.to_owned(), expected_revision: None, source: include_str!("../../../runa-std/actors/counter.rn").to_owned(), }, @@ -27,22 +152,18 @@ fn main() -> Result<(), Box> { "Loaded behavior `{}` revision {}.", behavior.name, behavior.revision ); - println!("Type `help` for commands."); - run_repl( - io::stdin().lock(), - io::stdout().lock(), - &mut server, - DEFAULT_BEHAVIOR, - )?; + println!("Type `help` for commands. Use Up/Down for history and Tab for completion."); + run_interactive_repl(&mut server)?; Ok(()) } +#[cfg(test)] fn run_repl( mut input: R, mut output: W, server: &mut Server, - default_behavior: &str, ) -> io::Result<()> { + let mut state = ReplState::default(); let mut line = String::new(); loop { write!(output, "runa> ")?; @@ -52,129 +173,202 @@ fn run_repl( writeln!(output)?; break; } - let mut words = line.split_whitespace(); - let Some(command) = words.next() else { - continue; - }; - let args: Vec<_> = words.collect(); - let result = match command { - "help" => print_help(&mut output), - "spawn" => spawn_command(&mut output, &args, default_behavior, server), - "behaviors" if args.is_empty() => match dispatch(server, Command::GetBehaviors) { - Ok(Reply::Behaviors(behaviors)) => { - for behavior in behaviors { - writeln!( - output, - "{} revision {} ({})", - behavior.name, behavior.revision, behavior.current_implementation_id - )?; - } - Ok(()) - } - Ok(other) => unexpected_reply(&mut output, other), - Err(error) => writeln!(output, "error: {error}"), - }, - "behaviors" => writeln!(output, "error: usage: behaviors"), - "publish" => publish_command(&mut output, &args, server), - "history" => history_command(&mut output, &args, server), - "rollback" => rollback_command(&mut output, &args, server), - "install" => install_command(&mut output, &args, server), - "actors" if args.is_empty() => match dispatch(server, Command::ListActors) { - Ok(Reply::Actors(actors)) => { - for actor in actors { - writeln!(output, "{actor:#?}")?; - } - Ok(()) - } - Ok(other) => unexpected_reply(&mut output, other), - Err(error) => writeln!(output, "error: {error}"), - }, - "actors" => writeln!(output, "error: usage: actors"), - "send" => match parse_send_args(&args) { - Ok((actor, kind, value)) => { - match dispatch( - server, - Command::Send { - actor_id: actor, - message: MessageDto { kind, value }, - }, - ) { - Ok(Reply::Acknowledged) => writeln!(output, "message queued"), - Ok(other) => unexpected_reply(&mut output, other), - Err(error) => writeln!(output, "error: {error}"), - } - } - Err(error) => writeln!(output, "error: {error}"), - }, - "tick" => match parse_tick_args(&args) { - Ok((count, budget)) => { - match dispatch( - server, - Command::Tick { - ticks: count, - instruction_budget: budget.get(), - }, - ) { - Ok(Reply::TickCompleted { ticks }) => { - writeln!(output, "ran {ticks} scheduler tick(s)") - } - Ok(other) => unexpected_reply(&mut output, other), - Err(error) => writeln!(output, "error: {error}"), - } - } - Err(error) => writeln!(output, "error: {error}"), - }, - "inspect" => match parse_exact::(&args) { - Ok([id]) => match dispatch(server, Command::InspectActor { actor_id: id }) { - Ok(Reply::Actor(actor)) => writeln!(output, "{actor:#?}"), - Ok(other) => unexpected_reply(&mut output, other), - Err(error) => writeln!(output, "error: {error}"), - }, - Err(error) => writeln!(output, "error: {error}"), - }, - "events" if args.is_empty() => match dispatch(server, Command::GetEvents) { - Ok(Reply::Events(events)) => { - for event in events { - writeln!(output, "{event:#?}")?; - } - Ok(()) - } - Ok(other) => unexpected_reply(&mut output, other), - Err(error) => writeln!(output, "error: {error}"), - }, - "events" => writeln!(output, "error: usage: events"), - "stop" => lifecycle_command(&mut output, &args, server, |actor_id| { - Command::StopActor { actor_id } - }), - "restart" => lifecycle_command(&mut output, &args, server, |actor_id| { - Command::RestartActor { actor_id } - }), - "pause" => lifecycle_command(&mut output, &args, server, |actor_id| { - Command::PauseActor { actor_id } - }), - "resume" => lifecycle_command(&mut output, &args, server, |actor_id| { - Command::ResumeActor { actor_id } - }), - "cancel" => lifecycle_command(&mut output, &args, server, |actor_id| { - Command::CancelActor { actor_id } - }), - "quit" | "exit" if args.is_empty() => break, - "quit" | "exit" => writeln!(output, "error: usage: quit"), - _ => writeln!(output, "error: unknown command `{command}`; type `help`"), - }; - result?; + if !execute_line(&line, &mut output, server, &mut state)? { + break; + } } Ok(()) } +fn run_interactive_repl(server: &mut Server) -> Result<(), Box> { + let mut editor = Editor::::new()?; + editor.set_helper(Some(ReplHelper::default())); + let mut state = ReplState::default(); + let mut output = io::stdout().lock(); + loop { + let behaviors = match dispatch(server, Command::GetBehaviors)? { + Reply::Behaviors(behaviors) => behaviors + .into_iter() + .map(|behavior| behavior.name) + .collect(), + other => return Err(format!("unexpected protocol response: {other:?}").into()), + }; + editor + .helper_mut() + .expect("REPL helper is installed") + .set_context(behaviors, &state.bindings); + match editor.readline("runa> ") { + Ok(line) => { + if !line.trim().is_empty() { + editor.add_history_entry(line.as_str())?; + } + if !execute_line(&line, &mut output, server, &mut state)? { + break; + } + } + Err(ReadlineError::Interrupted) => writeln!(output, "^C")?, + Err(ReadlineError::Eof) => { + writeln!(output)?; + break; + } + Err(error) => return Err(error.into()), + } + } + Ok(()) +} + +fn execute_line( + line: &str, + output: &mut W, + server: &mut Server, + state: &mut ReplState, +) -> io::Result { + let words: Vec<_> = line.split_whitespace().collect(); + if words.is_empty() { + return Ok(true); + } + if let ["let", name, "=", "spawn", behavior_name, initial_state] = words.as_slice() { + if !is_binding_name(name) { + writeln!(output, "error: invalid actor binding name `{name}`")?; + return Ok(true); + } + if state.bindings.contains_key(*name) { + writeln!(output, "error: actor binding `{name}` already exists")?; + return Ok(true); + } + match spawn_actor(behavior_name, initial_state, server) { + Ok(actor_ref) => { + writeln!(output, "{name} = @{actor_ref}")?; + state.bindings.insert((*name).to_owned(), actor_ref); + } + Err(error) => writeln!(output, "error: {error}")?, + } + return Ok(true); + } + let command = words[0]; + let args = &words[1..]; + let result = match command { + "help" => print_help(output), + "spawn" => spawn_command(output, args, server), + "behaviors" if args.is_empty() => match dispatch(server, Command::GetBehaviors) { + Ok(Reply::Behaviors(behaviors)) => { + for behavior in behaviors { + writeln!( + output, + "{} revision {} ({})", + behavior.name, behavior.revision, behavior.current_implementation_id + )?; + } + Ok(()) + } + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + }, + "behaviors" => writeln!(output, "error: usage: behaviors"), + "publish" => publish_command(output, args, server), + "history" => history_command(output, args, server), + "rollback" => rollback_command(output, args, server), + "install" => install_command(output, args, server, state), + "actors" if args.is_empty() => match dispatch(server, Command::ListActors) { + Ok(Reply::Actors(actors)) => { + for actor in actors { + writeln!(output, "{actor:#?}")?; + } + Ok(()) + } + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + }, + "actors" => writeln!(output, "error: usage: actors"), + "send" => match parse_send_args(args, state) { + Ok((actor_ref, kind, value, target_ref)) => { + match dispatch( + server, + Command::Send { + actor_ref, + message: MessageDto { + kind, + value, + target_ref, + }, + }, + ) { + Ok(Reply::Acknowledged) => writeln!(output, "message queued"), + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + } + } + Err(error) => writeln!(output, "error: {error}"), + }, + "tick" => match parse_tick_args(args) { + Ok((count, budget)) => { + match dispatch( + server, + Command::Tick { + ticks: count, + instruction_budget: budget.get(), + }, + ) { + Ok(Reply::TickCompleted { ticks }) => { + writeln!(output, "ran {ticks} scheduler tick(s)") + } + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + } + } + Err(error) => writeln!(output, "error: {error}"), + }, + "inspect" => match parse_actor_argument(args, state) { + Ok(actor_ref) => match dispatch(server, Command::InspectActor { actor_ref }) { + Ok(Reply::Actor(actor)) => writeln!(output, "{actor:#?}"), + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + }, + Err(error) => writeln!(output, "error: {error}"), + }, + "events" if args.is_empty() => match dispatch(server, Command::GetEvents) { + Ok(Reply::Events(events)) => { + for event in events { + writeln!(output, "{event:#?}")?; + } + Ok(()) + } + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + }, + "events" => writeln!(output, "error: usage: events"), + "stop" => lifecycle_command(output, args, server, state, |actor_ref| { + Command::StopActor { actor_ref } + }), + "restart" => lifecycle_command(output, args, server, state, |actor_ref| { + Command::RestartActor { actor_ref } + }), + "pause" => lifecycle_command(output, args, server, state, |actor_ref| { + Command::PauseActor { actor_ref } + }), + "resume" => lifecycle_command(output, args, server, state, |actor_ref| { + Command::ResumeActor { actor_ref } + }), + "cancel" => lifecycle_command(output, args, server, state, |actor_ref| { + Command::CancelActor { actor_ref } + }), + "quit" | "exit" if args.is_empty() => return Ok(false), + "quit" | "exit" => writeln!(output, "error: usage: quit"), + _ => writeln!(output, "error: unknown command `{command}`; type `help`"), + }; + result?; + Ok(true) +} + fn lifecycle_command( output: &mut W, args: &[&str], server: &mut Server, - operation: impl FnOnce(u64) -> Command, + state: &ReplState, + operation: impl FnOnce(String) -> Command, ) -> io::Result<()> { - match parse_exact::(args) { - Ok([id]) => match dispatch(server, operation(id)) { + match parse_actor_argument(args, state) { + Ok(actor_ref) => match dispatch(server, operation(actor_ref)) { Ok(Reply::Acknowledged) => writeln!(output, "ok"), Ok(other) => unexpected_reply(output, other), Err(error) => writeln!(output, "error: {error}"), @@ -183,56 +377,66 @@ fn lifecycle_command( } } -fn spawn_command( - output: &mut W, - args: &[&str], - default_behavior: &str, +fn spawn_command(output: &mut W, args: &[&str], server: &mut Server) -> io::Result<()> { + let [behavior_name, initial_state] = args else { + return writeln!( + output, + "error: usage: spawn " + ); + }; + match spawn_actor(behavior_name, initial_state, server) { + Ok(actor_ref) => writeln!(output, "spawned @{actor_ref} from `{behavior_name}`"), + Err(error) => writeln!(output, "error: {error}"), + } +} + +fn spawn_actor( + behavior_name: &str, + initial_state: &str, server: &mut Server, -) -> io::Result<()> { - let (state_arg, behavior_name) = match args { - [state] => (*state, default_behavior), - [state, behavior_name] => (*state, *behavior_name), - _ => { - return writeln!( - output, - "error: usage: spawn [behavior-name]" - ); - } - }; - let state = match state_arg.parse() { - Ok(state) => state, - Err(_) => return writeln!(output, "error: invalid initial state: `{state_arg}`"), - }; - let response = dispatch( +) -> Result { + let state = initial_state + .parse() + .map_err(|_| format!("invalid initial state: `{initial_state}`"))?; + match dispatch( server, Command::Spawn { behavior_name: behavior_name.to_owned(), state, mailbox_capacity: MAILBOX_CAPACITY, }, - ); - match response { - Ok(Reply::ActorSpawned { actor_id }) => { - let revision = match dispatch( - server, - Command::GetBehavior { - name: behavior_name.to_owned(), - }, - ) { - Ok(Reply::Behavior(behavior)) => behavior.revision, - Ok(other) => return unexpected_reply(output, other), - Err(error) => return writeln!(output, "error: {error}"), - }; - writeln!( - output, - "spawned ActorId({actor_id}) with `{behavior_name}` revision {revision}" - ) - } - Ok(other) => unexpected_reply(output, other), - Err(error) => writeln!(output, "error: {error}"), + )? { + Reply::ActorSpawned { actor_ref } => Ok(actor_ref), + other => Err(format!("unexpected protocol response: {other:?}")), } } +fn parse_actor_argument(args: &[&str], state: &ReplState) -> Result { + let [argument] = args else { + return Err(format!("expected 1 actor reference, got {}", args.len())); + }; + if let Some(actor_ref) = state.bindings.get(*argument) { + return Ok(actor_ref.clone()); + } + let actor_ref = argument.strip_prefix('@').unwrap_or(argument); + if actor_ref.len() != 21 + || !actor_ref + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) + { + return Err(format!( + "unknown actor binding or invalid actor reference `{argument}`" + )); + } + Ok(actor_ref.to_owned()) +} + +fn is_binding_name(name: &str) -> bool { + let mut characters = name.chars(); + matches!(characters.next(), Some(character) if character == '_' || character.is_ascii_alphabetic()) + && characters.all(|character| character == '_' || character.is_ascii_alphanumeric()) +} + fn publish_command(output: &mut W, args: &[&str], server: &mut Server) -> io::Result<()> { let [name, expected_revision, source_path] = args else { return writeln!( @@ -339,18 +543,26 @@ fn rollback_command( } } -fn install_command(output: &mut W, args: &[&str], server: &mut Server) -> io::Result<()> { - let [actor_id, name] = args else { - return writeln!(output, "error: usage: install "); +fn install_command( + output: &mut W, + args: &[&str], + server: &mut Server, + state: &ReplState, +) -> io::Result<()> { + let [actor_arg, name] = args else { + return writeln!( + output, + "error: usage: install " + ); }; - let actor_id = match actor_id.parse::() { - Ok(id) => id, - Err(_) => return writeln!(output, "error: invalid actor ID"), + let actor_ref = match parse_actor_argument(&[*actor_arg], state) { + Ok(actor_ref) => actor_ref, + Err(error) => return writeln!(output, "error: {error}"), }; let implementation = match dispatch( server, Command::InstallBehavior { - actor_id, + actor_ref: actor_ref.clone(), behavior_name: (*name).to_owned(), }, ) { @@ -367,8 +579,8 @@ fn install_command(output: &mut W, args: &[&str], server: &mut Server) Ok(Reply::Behavior(behavior)) => { writeln!( output, - "installed `{}` revision {} on actor {} ({implementation})", - behavior.name, behavior.revision, actor_id + "installed `{}` revision {} on actor @{} ({implementation})", + behavior.name, behavior.revision, actor_ref ) } Ok(other) => unexpected_reply(output, other), @@ -389,23 +601,6 @@ fn unexpected_reply(output: &mut W, reply: Reply) -> io::Result<()> { writeln!(output, "error: unexpected protocol response: {reply:?}") } -fn parse_exact(args: &[&str]) -> Result<[T; N], String> { - if args.len() != N { - return Err(format!("expected {N} argument(s), got {}", args.len())); - } - let parsed = args - .iter() - .enumerate() - .map(|(index, arg)| { - arg.parse::() - .map_err(|_| format!("invalid argument {}: `{arg}`", index + 1)) - }) - .collect::, _>>()?; - parsed - .try_into() - .map_err(|_| "invalid argument count".to_owned()) -} - fn parse_tick_args(args: &[&str]) -> Result<(usize, NonZeroUsize), String> { match args { [count] => Ok(( @@ -428,42 +623,49 @@ fn parse_tick_args(args: &[&str]) -> Result<(usize, NonZeroUsize), String> { } } -fn parse_send_args(args: &[&str]) -> Result<(u64, i64, i64), String> { - if args.len() != 3 { - return Err(format!("expected 3 arguments, got {}", args.len())); - } - let actor = args[0] +fn parse_send_args( + args: &[&str], + state: &ReplState, +) -> Result<(String, i64, i64, Option), String> { + let (actor_arg, kind_arg, value_arg, target_arg) = match args { + [actor, kind, value] => (*actor, *kind, *value, None), + [actor, kind, value, "--to", target] => (*actor, *kind, *value, Some(*target)), + _ => return Err("usage: send [--to ]".to_owned()), + }; + let actor_ref = parse_actor_argument(&[actor_arg], state)?; + let kind = kind_arg .parse() - .map_err(|_| format!("invalid actor ID: `{}`", args[0]))?; - let kind = args[1] + .map_err(|_| format!("invalid message kind: `{kind_arg}`"))?; + let value = value_arg .parse() - .map_err(|_| format!("invalid message kind: `{}`", args[1]))?; - let value = args[2] - .parse() - .map_err(|_| format!("invalid message value: `{}`", args[2]))?; - Ok((actor, kind, value)) + .map_err(|_| format!("invalid message value: `{value_arg}`"))?; + let target_ref = target_arg + .map(|target| parse_actor_argument(&[target], state)) + .transpose()?; + Ok((actor_ref, kind, value, target_ref)) } fn print_help(output: &mut W) -> io::Result<()> { writeln!( output, "Commands:\n\ - spawn [behavior-name]\n\ + let = spawn \n\ + spawn \n\ behaviors\n\ publish \n\ history \n\ rollback \n\ - install \n\ - send \n\ + install \n\ + send [--to ]\n\ tick [instruction-budget]\n\ - inspect \n\ + inspect \n\ actors\n\ events\n\ - stop \n\ - restart \n\ - pause \n\ - resume \n\ - cancel \n\ + stop \n\ + restart \n\ + pause \n\ + resume \n\ + cancel \n\ help\n\ quit" ) @@ -478,7 +680,7 @@ mod tests { dispatch( &mut server, Command::PublishBehavior { - name: DEFAULT_BEHAVIOR.to_owned(), + name: COUNTER_BEHAVIOR.to_owned(), expected_revision: None, source: include_str!("../../../runa-std/actors/counter.rn").to_owned(), }, @@ -491,26 +693,29 @@ mod tests { fn repl_drives_actor_messages_and_inspection() -> Result<(), Box> { let mut server = server(); let input = io::Cursor::new( - "spawn 0\nspawn 10\nsend 1 1 2\ntick 100\ninspect 1\ninspect 2\nevents\nquit\n", + "let sender = spawn counter 0\nlet receiver = spawn counter 10\nsend sender 1 5 --to receiver\ntick 100\ninspect sender\ninspect receiver\nevents\nquit\n", ); let mut output = Vec::new(); - run_repl(input, &mut output, &mut server, DEFAULT_BEHAVIOR)?; + run_repl(input, &mut output, &mut server)?; let output = String::from_utf8(output)?; - assert!(output.contains("actor_id: 1")); - assert!(output.contains("actor_id: 2")); + assert!(output.contains("sender = @")); + assert!(output.contains("receiver = @")); assert!(output.contains("state: 15")); assert!(output.contains("Committed {")); - assert!(output.contains("actor_id: 2")); + assert!(output.contains("actor_ref: \"")); Ok(()) } #[test] fn repl_reports_bad_input_and_continues() -> Result<(), Box> { let mut server = server(); - let input = io::Cursor::new("spawn nope\nsend 999 0 1\ntick 1 0\nunknown\nquit\n"); + let input = io::Cursor::new( + "spawn 0\nspawn counter nope\nsend 000000000000000000000 0 1\ntick 1 0\nunknown\nquit\n", + ); let mut output = Vec::new(); - run_repl(input, &mut output, &mut server, DEFAULT_BEHAVIOR)?; + run_repl(input, &mut output, &mut server)?; let output = String::from_utf8(output)?; + assert!(output.contains("usage: spawn ")); assert!(output.contains("invalid initial state")); assert!(output.contains("unknown actor")); assert!(output.contains("tick budget must be positive")); @@ -527,35 +732,51 @@ mod tests { "/../../runa-std/actors/counter.rn" ); let input = io::Cursor::new(format!( - "behaviors\npublish counter 1 {source_path}\nhistory counter\nrollback counter 2 1\nspawn 0 counter\ninstall 1 counter\nsend 1 0 1\ntick 1 1\ncancel 1\npause 1\nresume 1\ninspect 1\nevents\nquit\n" + "behaviors\npublish counter 1 {source_path}\nhistory counter\nrollback counter 2 1\nlet actor = spawn counter 0\ninstall actor counter\nsend actor 0 1\ntick 1 1\ncancel actor\npause actor\nresume actor\ninspect actor\nevents\nquit\n" )); let mut output = Vec::new(); - run_repl(input, &mut output, &mut server, DEFAULT_BEHAVIOR)?; + run_repl(input, &mut output, &mut server)?; let output = String::from_utf8(output)?; assert!(output.contains("published `counter` revision 2")); assert!(output.contains("revision 1:")); assert!(output.contains("revision 2:")); assert!(output.contains("rolled back `counter`")); - assert!(output.contains("spawned ActorId(1) with `counter` revision 3")); - assert!(output.contains("installed `counter` revision 3 on actor 1")); + assert!(output.contains("actor = @")); + assert!(output.contains("installed `counter` revision 3 on actor @")); assert!(output.contains("Paused {")); assert!(output.contains("Resumed {")); assert!(output.contains("InvocationCancelled {")); let behavior = match dispatch( &mut server, Command::GetBehavior { - name: DEFAULT_BEHAVIOR.to_owned(), + name: COUNTER_BEHAVIOR.to_owned(), }, )? { Reply::Behavior(behavior) => behavior, other => return Err(format!("unexpected response: {other:?}").into()), }; assert_eq!(behavior.revision, 3); - let actor = match dispatch(&mut server, Command::InspectActor { actor_id: 1 })? { + let actor_ref = match dispatch(&mut server, Command::ListActors)? { + Reply::Actors(actors) => actors[0].actor_ref.clone(), + other => return Err(format!("unexpected response: {other:?}").into()), + }; + let actor = match dispatch(&mut server, Command::InspectActor { actor_ref })? { Reply::Actor(actor) => actor, other => return Err(format!("unexpected response: {other:?}").into()), }; assert_eq!(actor.implementation_id, behavior.current_implementation_id); Ok(()) } + + #[test] + fn completion_offers_commands_behaviors_and_named_actor_refs() { + let mut helper = ReplHelper::default(); + let mut bindings = BTreeMap::new(); + bindings.insert("sender".to_owned(), "xK8_pQ2-Lm5_012345678".to_owned()); + helper.set_context(vec!["counter".to_owned()], &bindings); + + assert_eq!(helper.candidates("sp", 2), vec!["spawn"]); + assert_eq!(helper.candidates("spawn co", 8), vec!["counter"]); + assert_eq!(helper.candidates("inspect se", 10), vec!["sender"]); + } } diff --git a/docs/architecture.md b/docs/architecture.md index af39486..3d488f0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -71,19 +71,25 @@ An actor comprises distinct components: ```text Actor { - id: ActorId, // stable logical identity + actor_ref: ActorRef, // opaque public reference mailbox: MailboxId, // independent queue; may survive restart state: StateRecord, // committed private state behavior: ImplementationId, // immutable handler implementation incarnation: Generation, // changes on restart status: ActorStatus, - supervisor: Option, + supervisor: Option, } ``` ### 4.1 Identity -Actor IDs are opaque, never raw pointers. Use generation or other safeguards against stale references. A logical actor may be restarted with the same ID and mailbox but a new incarnation. Distinguish runtime-instance identity from actor identity if remote or persisted references are introduced. +Actor references are opaque, never raw pointers or sequential numbers. The +prototype uses 21-character URL-safe NanoIDs for public references. A logical +actor retains its reference and mailbox across restart but receives a new +incarnation. Random references are not an authentication or authorization +mechanism; network access remains loopback-only until access control exists. +Distinguish runtime-instance identity from actor identity if remote or +persisted references are introduced. ### 4.2 Mailbox @@ -304,7 +310,7 @@ The diagram is conceptual; Cargo dependencies should be kept minimal. A dependen | Crate | Owns | May depend on | | ---------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------- | -| `runa-core` | Small, stable value types shared across independent layers: actor IDs, source and implementation IDs, canonical message value types, common error identifiers | Minimal external libraries only | +| `runa-core` | Small, stable value types shared across independent layers: source and implementation IDs, canonical message value types, common error identifiers | Minimal external libraries only | | `runa-code` | Original source retention, versioned source/implementation hashing, compile-input fingerprints, in-memory compiled artifact cache | `runa-core`, `runa-rune`, BLAKE3 | | `runa-actor` | Mailbox and state-transition semantics, message reservation, acknowledgment, effects, actor lifecycle definitions | `runa-core` | | `runa-rune` | Rune compilation, VM execution, suspension, interruption hooks, language-specific diagnostics | `runa-core`, Rune libraries | diff --git a/docs/protocol.md b/docs/protocol.md index 54973bc..fda597f 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -7,9 +7,9 @@ when started with `--listen`, over TCP bound to an IPv4 or IPv6 loopback address. The TCP listener is unauthenticated and is intended only for trusted local development. -## Version 1 +## Version 2 -Every request and response carries `version: 1`. Requests also carry a +Every request and response carries `version: 2`. Requests also carry a non-empty `request_id` used to correlate the response. A server must reject unsupported versions with `unsupported_version` rather than interpreting a request using a different contract. Events carry the protocol version and do @@ -19,19 +19,23 @@ Requests have a tagged command object: ```json { - "version": 1, + "version": 2, "request_id": "req-1", "command": { "operation": "send", "parameters": { - "actor_id": 7, - "message": { "kind": 2, "value": 9 } + "actor_ref": "xK8_pQ2-Lm5_012345678", + "message": { + "kind": 2, + "value": 9, + "target_ref": "Hj2_9sM0-pQ4_7aBcD3eF" + } } } } ``` -The version 1 operations are behavior publication, rollback, lookup, source +The version 2 operations are behavior publication, rollback, lookup, source retrieval, actor spawn, send, tick, inspect, actor listing, event retrieval, stop, restart, pause, resume, cancellation, and behavior installation. Migration closures are intentionally excluded: they are host callbacks, not @@ -40,7 +44,9 @@ portable wire values. Responses echo the request ID and carry either a typed reply or a protocol error. Events use a separate versioned envelope. Actor and behavior snapshots are wire DTOs, not serialized runtime objects; implementation identities are -represented by their stable display strings. The server maps known service +represented by their stable display strings. Public actor identity is a +21-character URL-safe NanoID string carried in `actor_ref`; message `target_ref` +is optional and omitted when absent. The server maps known service failures to the defined error codes and otherwise uses `runtime_failure` rather than claiming a more specific category. @@ -58,8 +64,8 @@ must not be enabled until appropriate authentication and authorization exist. The `runa-client` crate launches the stdio server locally and validates response protocol versions and request IDs. -The current validation checks the protocol version, request ID, positive -actor/revision identifiers where required, non-empty behavior names, mailbox +The current validation checks the protocol version, request ID, actor-reference +shape, positive revisions where required, non-empty behavior names, mailbox capacity, and instruction budget. Behavior-name grammar, actor existence, current behavior revisions, mailbox capacity at send time, and all runtime invariants remain authoritative at the application-service boundary. diff --git a/docs/runtime.md b/docs/runtime.md index cb27449..58d2bd3 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -6,41 +6,44 @@ Start the interactive, in-memory runtime session with: mise run run ``` -The executable loads [counter.rn](../runa-std/actors/counter.rn) as the default -behavior. Commands include `spawn`, `send`, `tick`, `inspect`, `actors`, +The executable publishes [counter.rn](../runa-std/actors/counter.rn) as the +`counter` behavior. Commands include `spawn`, `send`, `tick`, `inspect`, `actors`, `events`, `stop`, `restart`, `pause`, `resume`, `cancel`, `behaviors`, `publish`, `history`, `rollback`, and `install`; enter `help` to see usage. `publish` reads a Rune source file and takes `new` for a first revision or the expected current revision for an update. `rollback` requires the expected current revision and a target historical revision. `install` selects a named behavior's current implementation and fails if the actor is not at an idle -handler boundary. -`spawn` accepts an optional behavior name and otherwise uses `counter`. All -commands operate on the same process-local runtime and code store. This CLI is -for local development, not a versioned external API. +handler boundary. `spawn` requires a behavior name and initial state, in that +order. All commands operate on the same process-local runtime and code store. +The CLI uses the versioned protocol dispatcher, also available over stdio or +the loopback-only TCP listener for local development. Example: ```text -runa> spawn 0 -spawned ActorId(1) -runa> spawn 10 -spawned ActorId(2) -runa> send 1 1 2 +runa> let sender = spawn counter 0 +sender = @ +runa> let receiver = spawn counter 10 +receiver = @ +runa> send sender 1 5 --to receiver message queued runa> tick 100 -runa> inspect 1 -runa> inspect 2 +runa> inspect sender +runa> inspect receiver ``` The sender commits state 1 and sends a structured message; the receiver commits -state 15. `tick [instruction-budget]` explicitly advances the scheduler. -The default per-actor slice budget is 32 instructions. +state 15. Actor references are 21-character NanoIDs; names are local REPL +bindings. Tab completes commands, behaviors, and bound actor references, while +Up/Down navigate command history. `tick [instruction-budget]` +explicitly advances the scheduler. The default per-actor slice budget is 32 +instructions. ## Current contracts -- `runa-actor` owns actor, mailbox and implementation IDs, structured messages, - snapshots, statuses and lifecycle events. +- `runa-actor` owns actor references, mailbox and implementation IDs, structured + messages, snapshots, statuses and lifecycle events. - `runa-core` owns stable source and implementation identity types. - `runa-code` preserves original source, derives identities, caches compiled artifacts, and provides an in-memory named behavior registry with revisions. @@ -52,8 +55,8 @@ The default per-actor slice budget is 32 instructions. - `runa-protocol` defines versioned transport-neutral request, response, and event DTOs; it does not dispatch requests or access runtime state. - `runa-server` dispatches those requests through `runa-image-api` and maps - results to protocol replies; its executable uses newline-delimited JSON over - stdin/stdout with a 1 MiB request-line limit, not a network listener. + results to protocol replies; its executable accepts newline-delimited JSON + over stdin/stdout or TCP bound to loopback, with a 1 MiB request-line limit. - `runa` provides an interactive CLI client using protocol requests through the dispatcher. @@ -66,8 +69,8 @@ published source for a selected behavior revision. An implementation is published once under an ID. Actors reference that exact implementation. Implementation IDs use source content and the compile -fingerprint; actor and mailbox IDs remain process-local counters. Actor IDs and -incarnation counters use checked increments. +fingerprint; actor references are random NanoIDs and mailbox IDs remain +process-local counters. Incarnation counters use checked increments. `CodeStore::publish_behavior` compiles source before appending a behavior revision. Creation expects no existing revision; updates require the caller's @@ -133,19 +136,26 @@ callers must change it when registered native functions or their contracts change. The CLI currently creates a fresh `CodeStore` when it starts. Prototype state is an `i64`; messages contain integer `kind` and `value` -fields. Rune handlers have this deliberately restricted interface: +fields, plus an optional target actor reference for a handler-produced send. +Public actor references are 21-character URL-safe NanoIDs. They are generated +when actors are spawned and remain stable across stop/restart; mailbox IDs are +separate internal counters. Rune handlers have this deliberately restricted +interface: ```text -handle(state, kind, value) -> (new_state, [(target_actor_id, kind, value), ...]) +handle(state, kind, value, target_ref) + -> (new_state, [(target_ref, kind, value), ...]) ``` +`target_ref` is an empty string when the incoming message has no routed target. The returned effect vector describes sends; handlers cannot access the runtime or directly mutate committed state. General schema/versioned values and capability-bearing contexts are future work. ## Scheduling and lifecycle -`tick(nonzero_budget)` gives each runnable actor one bounded slice in ID order. +`tick(nonzero_budget)` gives each runnable actor one bounded slice in actor +reference order. FIFO queues are bounded, including the reserved message in capacity accounting. A queued message is reserved when its invocation starts. Subsequent messages stay queued, including while the actor awaits. A wake marks an awaiting actor @@ -166,9 +176,10 @@ silently acknowledged or automatically retried. External sends return errors for unknown, unavailable, or full actors. Stopping drops the invocation, retaining committed state, mailbox and reserved -message. Restarting a stopped or failed actor preserves its actor ID, mailbox -ID, implementation ID, committed state, and queued-message order; it increments -its incarnation and puts the reserved message at the front of the queue. +message. Restarting a stopped or failed actor preserves its actor reference, +mailbox ID, implementation ID, committed state, and queued-message order; it +increments its incarnation and puts the reserved message at the front of the +queue. Restart is explicit. Repeated failure of the same reserved message can therefore block later messages until an operator changes the implementation or otherwise resolves it; there is no automatic poison-message policy or supervision. @@ -208,11 +219,10 @@ capacity, and successful message effects. ## Limits This is not yet a full image: no persistent behavior registry or mailboxes, -transactional outbox, supervisors, API server, GUI, or automatic image -bootstrap. The behavior registry and implementation installation are -process-local; installation requires an idle handler boundary and does not -upgrade active execution. The example default script is bundled through -`include_str!`. +transactional outbox, supervisors, GUI, or automatic image bootstrap. The +behavior registry and implementation installation are process-local; +installation requires an idle handler boundary and does not upgrade active +execution. The example counter script is bundled through `include_str!`. Messages are not accepted by stopped/failed actors. Native functions must be nonblocking. Dropping a future proves cleanup only diff --git a/runa-std/actors/counter.rn b/runa-std/actors/counter.rn index c6f9101..b3c11c0 100644 --- a/runa-std/actors/counter.rn +++ b/runa-std/actors/counter.rn @@ -1,6 +1,6 @@ -pub fn handle(state, kind, value) { +pub fn handle(state, kind, value, target_ref) { if kind == 1 { - (state + 1, [(value, 0, 5)]) + (state + 1, [(target_ref, 0, 5)]) } else { (state + value, []) }