feat: add NanoID actor refs and REPL bindings

Require explicit behavior selection on spawn, expose randomized actor references, and add named REPL bindings, completion, history navigation, and protocol v2 routing.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Steve Biedermann 2026-11-02 15:19:25 +01:00
parent d96d603286
commit 4f6e7ccfa2
16 changed files with 1496 additions and 555 deletions

View File

@ -46,6 +46,10 @@ All notable changes to this project will be documented in this file.
- Execution-model decision and Phase 1 runtime documentation. - Execution-model decision and Phase 1 runtime documentation.
- Lifecycle tests proving failed-message retry from committed state and - Lifecycle tests proving failed-message retry from committed state and
stop/restart preservation of reserved-first FIFO order. 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, - Runnable Phase 0 experiment pinned to Rune 0.14.2, covering script execution,
native calls, async suspension, instruction budgets, frame inspection, 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 - Interactive CLI commands now use the versioned protocol dispatcher instead
of calling the image API directly. 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 - Actor implementation IDs now identify compiled source/configuration content
rather than using sequential process-local numbers. rather than using sequential process-local numbers.
- Expanded the architecture specification with crate ownership and dependency - Expanded the architecture specification with crate ownership and dependency

299
Cargo.lock generated
View File

@ -41,6 +41,12 @@ version = "1.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
[[package]]
name = "bitflags"
version = "2.13.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06"
[[package]] [[package]]
name = "blake3" name = "blake3"
version = "1.8.7" version = "1.8.7"
@ -70,6 +76,21 @@ version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" 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]] [[package]]
name = "codespan-reporting" name = "codespan-reporting"
version = "0.11.1" version = "0.11.1"
@ -77,7 +98,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3538270d33cc669650c4b093848450d380def10c331d38c768e34cac80576e6e" checksum = "3538270d33cc669650c4b093848450d380def10c331d38c768e34cac80576e6e"
dependencies = [ dependencies = [
"termcolor", "termcolor",
"unicode-width", "unicode-width 0.1.14",
] ]
[[package]] [[package]]
@ -101,6 +122,39 @@ version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" 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]] [[package]]
name = "find-msvc-tools" name = "find-msvc-tools"
version = "0.1.15" version = "0.1.15"
@ -146,6 +200,26 @@ dependencies = [
"windows-result", "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]] [[package]]
name = "itoa" name = "itoa"
version = "1.0.18" version = "1.0.18"
@ -164,6 +238,12 @@ version = "0.2.191"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a" checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a"
[[package]]
name = "linux-raw-sys"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"
[[package]] [[package]]
name = "log" name = "log"
version = "0.4.35" version = "0.4.35"
@ -230,13 +310,43 @@ dependencies = [
"syn 2.0.119", "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]] [[package]]
name = "nu-ansi-term" name = "nu-ansi-term"
version = "0.50.3" version = "0.50.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
dependencies = [ dependencies = [
"windows-sys", "windows-sys 0.61.2",
] ]
[[package]] [[package]]
@ -353,6 +463,15 @@ version = "1.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" 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]] [[package]]
name = "proc-macro2" name = "proc-macro2"
version = "1.0.107" version = "1.0.107"
@ -371,6 +490,46 @@ dependencies = [
"proc-macro2", "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]] [[package]]
name = "regex-automata" name = "regex-automata"
version = "0.4.18" version = "0.4.18"
@ -394,12 +553,14 @@ version = "0.0.0"
dependencies = [ dependencies = [
"runa-protocol", "runa-protocol",
"runa-server", "runa-server",
"rustyline",
] ]
[[package]] [[package]]
name = "runa-actor" name = "runa-actor"
version = "0.0.0" version = "0.0.0"
dependencies = [ dependencies = [
"nanoid",
"runa-core", "runa-core",
] ]
@ -566,12 +727,47 @@ dependencies = [
"syn 2.0.119", "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]] [[package]]
name = "rustversion" name = "rustversion"
version = "1.0.23" version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" 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]] [[package]]
name = "ryu" name = "ryu"
version = "1.0.23" version = "1.0.23"
@ -767,12 +963,30 @@ version = "1.0.27"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563"
[[package]]
name = "unicode-segmentation"
version = "1.13.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8"
[[package]] [[package]]
name = "unicode-width" name = "unicode-width"
version = "0.1.14" version = "0.1.14"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" 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]] [[package]]
name = "valuable" name = "valuable"
version = "0.1.1" version = "0.1.1"
@ -785,13 +999,19 @@ version = "0.9.5"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" 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]] [[package]]
name = "winapi-util" name = "winapi-util"
version = "0.1.11" version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [ dependencies = [
"windows-sys", "windows-sys 0.61.2",
] ]
[[package]] [[package]]
@ -809,6 +1029,15 @@ dependencies = [
"windows-link", "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]] [[package]]
name = "windows-sys" name = "windows-sys"
version = "0.61.2" version = "0.61.2"
@ -818,6 +1047,70 @@ dependencies = [
"windows-link", "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]] [[package]]
name = "zerocopy" name = "zerocopy"
version = "0.8.63" version = "0.8.63"

View File

@ -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 the image regardless of how clients connect. Remote network access requires
authentication and authorization, which are not implemented. 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 ```text
runa> spawn 0 runa> let sender = spawn counter 0
spawned ActorId(1) sender = @<actor-reference>
runa> spawn 10 runa> let receiver = spawn counter 10
spawned ActorId(2) receiver = @<actor-reference>
runa> send 1 1 2 runa> send sender 1 5 --to receiver
message queued message queued
runa> tick 100 runa> tick 100
runa> inspect 1 runa> inspect sender
runa> inspect 2 runa> inspect receiver
runa> events runa> events
``` ```
The sender commits state `1` and sends a message to actor 2, which commits Actor references are generated NanoIDs rather than sequential numbers. The
state `15`. The CLI also supports publishing Rune source files as named, `let` form binds each spawned actor to a local name; references can also be
versioned behaviors, viewing history, rolling back a behavior reference, entered directly with an `@` prefix. The sender commits state `1` and sends a
installing the current version on an idle actor, and cooperatively pausing and message to the receiver, which commits state `15`. The REPL provides Tab
resuming or cancelling actor invocations between scheduler ticks. Runtime completion and Up/Down command history. It also supports publishing Rune source
users can also supply an explicit state migration closure during installation; files as named, versioned behaviors, viewing history, rolling back a behavior
the CLI's install command preserves state as-is. Use `help` in the session for reference, installing the current version on an idle actor, and cooperatively
the complete command list. 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 See [the feasibility report](docs/rune-feasibility.md) for verified Rune
capabilities and limitations, and [the runtime guide](docs/runtime.md) for capabilities and limitations, and [the runtime guide](docs/runtime.md) for
the runtime contract. the runtime contract.

View File

@ -4,4 +4,5 @@ version = "0.0.0"
edition = "2024" edition = "2024"
[dependencies] [dependencies]
nanoid = "0.4"
runa-core = { path = "../runa-core" } runa-core = { path = "../runa-core" }

View File

@ -1,9 +1,72 @@
//! Value contracts for the single-worker actor prototype. //! Value contracts for the single-worker actor prototype.
use runa_core::ImplementationId; use runa_core::ImplementationId;
use std::fmt;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ActorId(pub u64); 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<Self> {
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)] #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct MailboxId(pub u64); pub struct MailboxId(pub u64);
@ -12,6 +75,7 @@ pub struct MailboxId(pub u64);
pub struct Message { pub struct Message {
pub kind: i64, pub kind: i64,
pub value: i64, pub value: i64,
pub target_ref: Option<ActorRef>,
} }
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
@ -26,7 +90,7 @@ pub enum Status {
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct ActorSnapshot { pub struct ActorSnapshot {
pub id: ActorId, pub actor_ref: ActorRef,
pub mailbox: MailboxId, pub mailbox: MailboxId,
pub implementation: ImplementationId, pub implementation: ImplementationId,
pub incarnation: u64, pub incarnation: u64,
@ -46,14 +110,14 @@ pub struct ExecutionSnapshot {
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub enum Event { pub enum Event {
Spawned(ActorId), Spawned(ActorRef),
Committed(ActorId), Committed(ActorRef),
Failed(ActorId, String), Failed(ActorRef, String),
Stopped(ActorId), Stopped(ActorRef),
Restarted(ActorId), Restarted(ActorRef),
Paused(ActorId), Paused(ActorRef),
Resumed(ActorId), Resumed(ActorRef),
InvocationCancelled(ActorId), InvocationCancelled(ActorRef),
BehaviorInstalled(ActorId, ImplementationId), BehaviorInstalled(ActorRef, ImplementationId),
StateMigrated(ActorId, i64, i64), StateMigrated(ActorRef, i64, i64),
} }

View File

@ -3,7 +3,7 @@
use std::num::NonZeroUsize; use std::num::NonZeroUsize;
use std::sync::Arc; 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_code::{BehaviorSnapshot, CodeStore};
use runa_core::ImplementationId; use runa_core::ImplementationId;
use runa_runtime::{Runtime, RuntimeResult}; use runa_runtime::{Runtime, RuntimeResult};
@ -74,7 +74,7 @@ impl ImageApi {
behavior_name: &str, behavior_name: &str,
state: i64, state: i64,
capacity: NonZeroUsize, capacity: NonZeroUsize,
) -> RuntimeResult<ActorId> { ) -> RuntimeResult<ActorRef> {
let behavior = self let behavior = self
.code .code
.behavior(behavior_name) .behavior(behavior_name)
@ -82,7 +82,7 @@ impl ImageApi {
self.runtime.spawn(behavior.current, state, capacity) 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) self.runtime.send(actor, message)
} }
@ -92,7 +92,7 @@ impl ImageApi {
} }
} }
pub fn inspect(&self, actor: ActorId) -> RuntimeResult<ActorSnapshot> { pub fn inspect(&self, actor: ActorRef) -> RuntimeResult<ActorSnapshot> {
self.runtime.inspect(actor) self.runtime.inspect(actor)
} }
@ -104,29 +104,29 @@ impl ImageApi {
self.runtime.drain_events() self.runtime.drain_events()
} }
pub fn stop(&mut self, actor: ActorId) -> RuntimeResult<()> { pub fn stop(&mut self, actor: ActorRef) -> RuntimeResult<()> {
self.runtime.stop(actor) self.runtime.stop(actor)
} }
pub fn restart(&mut self, actor: ActorId) -> RuntimeResult<()> { pub fn restart(&mut self, actor: ActorRef) -> RuntimeResult<()> {
self.runtime.restart(actor) self.runtime.restart(actor)
} }
pub fn pause(&mut self, actor: ActorId) -> RuntimeResult<()> { pub fn pause(&mut self, actor: ActorRef) -> RuntimeResult<()> {
self.runtime.pause(actor) self.runtime.pause(actor)
} }
pub fn resume(&mut self, actor: ActorId) -> RuntimeResult<()> { pub fn resume(&mut self, actor: ActorRef) -> RuntimeResult<()> {
self.runtime.resume(actor) self.runtime.resume(actor)
} }
pub fn cancel(&mut self, actor: ActorId) -> RuntimeResult<()> { pub fn cancel(&mut self, actor: ActorRef) -> RuntimeResult<()> {
self.runtime.cancel(actor) self.runtime.cancel(actor)
} }
pub fn install( pub fn install(
&mut self, &mut self,
actor: ActorId, actor: ActorRef,
behavior_name: &str, behavior_name: &str,
) -> RuntimeResult<ImplementationId> { ) -> RuntimeResult<ImplementationId> {
self.install_with_migration(actor, behavior_name, Ok) self.install_with_migration(actor, behavior_name, Ok)
@ -134,7 +134,7 @@ impl ImageApi {
pub fn install_with_migration( pub fn install_with_migration(
&mut self, &mut self,
actor: ActorId, actor: ActorRef,
behavior_name: &str, behavior_name: &str,
migrate: impl FnOnce(i64) -> RuntimeResult<i64>, migrate: impl FnOnce(i64) -> RuntimeResult<i64>,
) -> RuntimeResult<ImplementationId> { ) -> RuntimeResult<ImplementationId> {
@ -162,14 +162,21 @@ mod tests {
let behavior = image.publish_behavior( let behavior = image.publish_behavior(
"counter", "counter",
None, None,
"pub fn handle(state, kind, value) { (state + value, []) }", "pub fn handle(state, kind, value, target_ref) { (state + value, []) }",
)?; )?;
assert_eq!( assert_eq!(
image.behavior_source("counter", behavior.revision)?, 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))?; 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)); image.tick(100, capacity(10));
assert_eq!(image.inspect(actor)?.state, 12); assert_eq!(image.inspect(actor)?.state, 12);
@ -178,7 +185,7 @@ mod tests {
image image
.events() .events()
.iter() .iter()
.any(|event| matches!(event, Event::Committed(id) if *id == actor)) .any(|event| matches!(event, Event::Committed(id) if id == &actor))
); );
Ok(()) Ok(())
} }
@ -189,14 +196,14 @@ mod tests {
let first = image.publish_behavior( let first = image.publish_behavior(
"counter", "counter",
None, None,
"pub fn handle(state, kind, value) { (state + value, []) }", "pub fn handle(state, kind, value, target_ref) { (state + value, []) }",
)?; )?;
assert!( assert!(
image image
.publish_behavior( .publish_behavior(
"counter", "counter",
None, None,
"pub fn handle(state, kind, value) { (state - value, []) }" "pub fn handle(state, kind, value, target_ref) { (state - value, []) }"
) )
.is_err() .is_err()
); );
@ -210,12 +217,12 @@ mod tests {
let first = image.publish_behavior( let first = image.publish_behavior(
"counter", "counter",
None, None,
"pub fn handle(state, kind, value) { (state + value, []) }", "pub fn handle(state, kind, value, target_ref) { (state + value, []) }",
)?; )?;
image.publish_behavior( image.publish_behavior(
"counter", "counter",
Some(first.revision), 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 actor = image.spawn("counter", 5, capacity(2))?;
let implementation = image.install_with_migration(actor, "counter", |state| { let implementation = image.install_with_migration(actor, "counter", |state| {

View File

@ -2,7 +2,7 @@
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
pub const PROTOCOL_VERSION: u16 = 1; pub const PROTOCOL_VERSION: u16 = 2;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
@ -60,18 +60,23 @@ impl Request {
return Err(invalid_request("mailbox_capacity must be positive")); return Err(invalid_request("mailbox_capacity must be positive"));
} }
} }
Command::Send { actor_id, .. } Command::Send { actor_ref, message } => {
| Command::InspectActor { actor_id } validate_actor_ref(actor_ref)?;
| Command::StopActor { actor_id } if let Some(target_ref) = &message.target_ref {
| Command::RestartActor { actor_id } validate_actor_ref(target_ref)?;
| Command::PauseActor { actor_id } }
| Command::ResumeActor { actor_id } }
| Command::CancelActor { actor_id } => validate_actor_id(*actor_id)?, 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 { Command::InstallBehavior {
actor_id, actor_ref,
behavior_name, behavior_name,
} => { } => {
validate_actor_id(*actor_id)?; validate_actor_ref(actor_ref)?;
validate_name(behavior_name)?; validate_name(behavior_name)?;
} }
Command::Tick { Command::Tick {
@ -115,7 +120,7 @@ pub enum Command {
mailbox_capacity: usize, mailbox_capacity: usize,
}, },
Send { Send {
actor_id: u64, actor_ref: String,
message: MessageDto, message: MessageDto,
}, },
Tick { Tick {
@ -123,27 +128,27 @@ pub enum Command {
instruction_budget: usize, instruction_budget: usize,
}, },
InspectActor { InspectActor {
actor_id: u64, actor_ref: String,
}, },
ListActors, ListActors,
GetEvents, GetEvents,
StopActor { StopActor {
actor_id: u64, actor_ref: String,
}, },
RestartActor { RestartActor {
actor_id: u64, actor_ref: String,
}, },
PauseActor { PauseActor {
actor_id: u64, actor_ref: String,
}, },
ResumeActor { ResumeActor {
actor_id: u64, actor_ref: String,
}, },
CancelActor { CancelActor {
actor_id: u64, actor_ref: String,
}, },
InstallBehavior { InstallBehavior {
actor_id: u64, actor_ref: String,
behavior_name: String, behavior_name: String,
}, },
} }
@ -163,7 +168,7 @@ pub enum Reply {
Behavior(BehaviorDto), Behavior(BehaviorDto),
Behaviors(Vec<BehaviorDto>), Behaviors(Vec<BehaviorDto>),
BehaviorSource { source: String }, BehaviorSource { source: String },
ActorSpawned { actor_id: u64 }, ActorSpawned { actor_ref: String },
Actor(ActorSnapshotDto), Actor(ActorSnapshotDto),
Actors(Vec<ActorSnapshotDto>), Actors(Vec<ActorSnapshotDto>),
Events(Vec<EventDto>), Events(Vec<EventDto>),
@ -182,36 +187,36 @@ pub struct ProtocolEvent {
#[serde(tag = "event", content = "data", rename_all = "snake_case")] #[serde(tag = "event", content = "data", rename_all = "snake_case")]
pub enum EventDto { pub enum EventDto {
Spawned { Spawned {
actor_id: u64, actor_ref: String,
}, },
Committed { Committed {
actor_id: u64, actor_ref: String,
}, },
Failed { Failed {
actor_id: u64, actor_ref: String,
message: String, message: String,
}, },
Stopped { Stopped {
actor_id: u64, actor_ref: String,
}, },
Restarted { Restarted {
actor_id: u64, actor_ref: String,
}, },
Paused { Paused {
actor_id: u64, actor_ref: String,
}, },
Resumed { Resumed {
actor_id: u64, actor_ref: String,
}, },
InvocationCancelled { InvocationCancelled {
actor_id: u64, actor_ref: String,
}, },
BehaviorInstalled { BehaviorInstalled {
actor_id: u64, actor_ref: String,
implementation_id: String, implementation_id: String,
}, },
StateMigrated { StateMigrated {
actor_id: u64, actor_ref: String,
previous_state: i64, previous_state: i64,
current_state: i64, current_state: i64,
}, },
@ -233,17 +238,19 @@ pub struct BehaviorVersionDto {
pub implementation_id: String, pub implementation_id: String,
} }
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub struct MessageDto { pub struct MessageDto {
pub kind: i64, pub kind: i64,
pub value: i64, pub value: i64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub target_ref: Option<String>,
} }
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub struct ActorSnapshotDto { pub struct ActorSnapshotDto {
pub actor_id: u64, pub actor_ref: String,
pub mailbox_id: u64, pub mailbox_id: u64,
pub implementation_id: String, pub implementation_id: String,
pub incarnation: u64, pub incarnation: u64,
@ -307,9 +314,15 @@ fn validate_name(name: &str) -> Result<(), ProtocolError> {
Ok(()) Ok(())
} }
fn validate_actor_id(actor_id: u64) -> Result<(), ProtocolError> { fn validate_actor_ref(actor_ref: &str) -> Result<(), ProtocolError> {
if actor_id == 0 { if actor_ref.len() != 21
return Err(invalid_request("actor_id must be positive")); || !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(()) Ok(())
} }
@ -347,8 +360,12 @@ mod tests {
#[test] #[test]
fn request_uses_a_stable_tagged_wire_shape() { fn request_uses_a_stable_tagged_wire_shape() {
let value = serde_json::to_value(request(Command::Send { let value = serde_json::to_value(request(Command::Send {
actor_id: 7, actor_ref: "xK8_pQ2-Lm5_012345678".to_owned(),
message: MessageDto { kind: 2, value: 9 }, message: MessageDto {
kind: 2,
value: 9,
target_ref: None,
},
})) }))
.unwrap(); .unwrap();
assert_eq!( assert_eq!(
@ -359,7 +376,7 @@ mod tests {
"command": { "command": {
"operation": "send", "operation": "send",
"parameters": { "parameters": {
"actor_id": 7, "actor_ref": "xK8_pQ2-Lm5_012345678",
"message": { "kind": 2, "value": 9 } "message": { "kind": 2, "value": 9 }
} }
} }
@ -395,7 +412,7 @@ mod tests {
} }
#[test] #[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); let mut request = request(Command::GetEvents);
request.request_id = " ".to_owned(); request.request_id = " ".to_owned();
assert_eq!( assert_eq!(
@ -404,7 +421,35 @@ mod tests {
); );
request.request_id = "req-2".to_owned(); 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!( assert_eq!(
request.validate().unwrap_err().code, request.validate().unwrap_err().code,
ErrorCode::InvalidRequest ErrorCode::InvalidRequest
@ -428,7 +473,7 @@ mod tests {
let event = ProtocolEvent { let event = ProtocolEvent {
version: PROTOCOL_VERSION, version: PROTOCOL_VERSION,
event: EventDto::StateMigrated { event: EventDto::StateMigrated {
actor_id: 3, actor_ref: "actor_ref_000000000000".to_owned(),
previous_state: 4, previous_state: 4,
current_state: 40, current_state: 40,
}, },

View File

@ -6,12 +6,12 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
use std::task::{Context, Wake, Waker}; 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_code::CompiledArtifact;
use runa_core::ImplementationId; use runa_core::ImplementationId;
use runa_rune::{ExperimentResult, Invocation, Slice}; use runa_rune::{ExperimentResult, Invocation, Slice};
type WireTransition = (i64, Vec<(u64, i64, i64)>); type WireTransition = (i64, Vec<(String, i64, i64)>);
pub type RuntimeResult<T> = ExperimentResult<T>; pub type RuntimeResult<T> = ExperimentResult<T>;
struct WakeFlag(AtomicBool); struct WakeFlag(AtomicBool);
@ -36,9 +36,9 @@ struct Actor {
#[derive(Default)] #[derive(Default)]
pub struct Runtime { pub struct Runtime {
actors: BTreeMap<ActorId, Actor>, actors: BTreeMap<ActorRef, Actor>,
implementations: BTreeMap<ImplementationId, Arc<CompiledArtifact>>, implementations: BTreeMap<ImplementationId, Arc<CompiledArtifact>>,
next_actor: u64, next_mailbox: u64,
events: Vec<Event>, events: Vec<Event>,
} }
@ -63,21 +63,24 @@ impl Runtime {
implementation: ImplementationId, implementation: ImplementationId,
state: i64, state: i64,
capacity: NonZeroUsize, capacity: NonZeroUsize,
) -> ExperimentResult<ActorId> { ) -> ExperimentResult<ActorRef> {
if !self.implementations.contains_key(&implementation) { if !self.implementations.contains_key(&implementation) {
return Err("unknown implementation".into()); return Err("unknown implementation".into());
} }
self.next_actor = self self.next_mailbox = self
.next_actor .next_mailbox
.checked_add(1) .checked_add(1)
.ok_or("actor IDs exhausted")?; .ok_or("mailbox IDs exhausted")?;
let id = ActorId(self.next_actor); let mut actor_ref = ActorRef::new();
while self.actors.contains_key(&actor_ref) {
actor_ref = ActorRef::new();
}
self.actors.insert( self.actors.insert(
id, actor_ref,
Actor { Actor {
snapshot: ActorSnapshot { snapshot: ActorSnapshot {
id, actor_ref,
mailbox: MailboxId(id.0), mailbox: MailboxId(self.next_mailbox),
implementation, implementation,
incarnation: 1, incarnation: 1,
state, state,
@ -93,11 +96,11 @@ impl Runtime {
paused_from: None, paused_from: None,
}, },
); );
self.events.push(Event::Spawned(id)); self.events.push(Event::Spawned(actor_ref));
Ok(id) 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")?; let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
if matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { if matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) {
return Err("actor is not accepting messages".into()); return Err("actor is not accepting messages".into());
@ -111,7 +114,7 @@ impl Runtime {
Ok(()) Ok(())
} }
pub fn inspect(&self, id: ActorId) -> ExperimentResult<ActorSnapshot> { pub fn inspect(&self, id: ActorRef) -> ExperimentResult<ActorSnapshot> {
let actor = self.actors.get(&id).ok_or("unknown actor")?; let actor = self.actors.get(&id).ok_or("unknown actor")?;
let mut snapshot = actor.snapshot.clone(); let mut snapshot = actor.snapshot.clone();
snapshot.execution = actor.active.as_ref().map(|invocation| { snapshot.execution = actor.active.as_ref().map(|invocation| {
@ -145,7 +148,7 @@ impl Runtime {
.collect() .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")?; let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
actor.active = None; actor.active = None;
actor.paused_from = None; actor.paused_from = None;
@ -155,7 +158,7 @@ impl Runtime {
} }
/// Pause scheduling between worker ticks without discarding an invocation. /// 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")?; let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
if matches!( if matches!(
actor.snapshot.status, actor.snapshot.status,
@ -170,7 +173,7 @@ impl Runtime {
} }
/// Resume from the exact runnable or awaiting state saved by `pause`. /// 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 actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
let status = actor.paused_from.take().ok_or("actor is not paused")?; let status = actor.paused_from.take().ok_or("actor is not paused")?;
actor.snapshot.status = status; actor.snapshot.status = status;
@ -181,7 +184,7 @@ impl Runtime {
/// Cancel an active handler and requeue its reserved message for retry. /// Cancel an active handler and requeue its reserved message for retry.
/// ///
/// This does not roll back side effects already performed by native code. /// 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")?; let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
if actor.active.is_none() { if actor.active.is_none() {
return Err("actor has no active invocation to cancel".into()); return Err("actor has no active invocation to cancel".into());
@ -199,7 +202,7 @@ impl Runtime {
Ok(()) 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")?; let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
if !matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { if !matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) {
return Err("stop the actor before restarting".into()); return Err("stop the actor before restarting".into());
@ -221,7 +224,7 @@ impl Runtime {
/// Install an immutable implementation between handler invocations. /// Install an immutable implementation between handler invocations.
pub fn install( pub fn install(
&mut self, &mut self,
id: ActorId, id: ActorRef,
implementation: ImplementationId, implementation: ImplementationId,
) -> ExperimentResult<()> { ) -> ExperimentResult<()> {
self.install_with_migration(id, implementation, Ok) self.install_with_migration(id, implementation, Ok)
@ -231,7 +234,7 @@ impl Runtime {
/// operation. A migration error leaves both unchanged. /// operation. A migration error leaves both unchanged.
pub fn install_with_migration( pub fn install_with_migration(
&mut self, &mut self,
id: ActorId, id: ActorRef,
implementation: ImplementationId, implementation: ImplementationId,
migrate: impl FnOnce(i64) -> ExperimentResult<i64>, migrate: impl FnOnce(i64) -> ExperimentResult<i64>,
) -> ExperimentResult<()> { ) -> ExperimentResult<()> {
@ -265,18 +268,18 @@ impl Runtime {
std::mem::take(&mut self.events) std::mem::take(&mut self.events)
} }
fn fail(&mut self, id: ActorId, error: String) { fn fail(&mut self, id: &ActorRef, error: String) {
let actor = self.actors.get_mut(&id).expect("scheduled actor exists"); let actor = self.actors.get_mut(id).expect("scheduled actor exists");
actor.active = None; actor.active = None;
actor.snapshot.status = Status::Failed(error.clone()); 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 (state, effects) = transition;
let mut counts = BTreeMap::<ActorId, usize>::new(); let mut counts = BTreeMap::<ActorRef, usize>::new();
for &(target, _, _) in &effects { for (target, _, _) in &effects {
let target = ActorId(target); let target = ActorRef::parse(target).ok_or("effect targets invalid actor reference")?;
let actor = self let actor = self
.actors .actors
.get(&target) .get(&target)
@ -286,24 +289,29 @@ impl Runtime {
} }
let count = counts.entry(target).or_default(); let count = counts.entry(target).or_default();
*count += 1; *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() { if actor.mailbox.len() + retained + *count > actor.capacity.get() {
return Err("effect exceeds mailbox capacity".into()); 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.state = state;
actor.snapshot.reserved = None; actor.snapshot.reserved = None;
actor.snapshot.status = Status::Idle; actor.snapshot.status = Status::Idle;
actor.active = None; actor.active = None;
for (target, kind, value) in effects { for (target, kind, value) in effects {
let target = ActorRef::parse(&target).expect("validated target");
self.actors self.actors
.get_mut(&ActorId(target)) .get_mut(&target)
.expect("validated target") .expect("validated target")
.mailbox .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(()) Ok(())
} }
@ -327,11 +335,18 @@ impl Runtime {
match Invocation::start( match Invocation::start(
program.vm(), program.vm(),
"handle", "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), Ok(invocation) => actor.active = Some(invocation),
Err(error) => { Err(error) => {
self.fail(id, error.to_string()); self.fail(&id, error.to_string());
continue; continue;
} }
} }
@ -350,13 +365,13 @@ impl Runtime {
.poll_slice(&mut cx, budget); .poll_slice(&mut cx, budget);
match outcome { match outcome {
Slice::Completed(transition) => { Slice::Completed(transition) => {
if let Err(error) = self.commit(id, transition) { if let Err(error) = self.commit(&id, transition) {
self.fail(id, error.to_string()); self.fail(&id, error.to_string());
} }
} }
Slice::Awaiting => actor.snapshot.status = Status::Awaiting, Slice::Awaiting => actor.snapshot.status = Status::Awaiting,
Slice::BudgetExhausted => actor.snapshot.status = Status::Running, 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 mut runtime = Runtime::default();
let code = publish( let code = publish(
&mut runtime, &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 first = runtime.spawn(code, 0, capacity(4))?;
let second = runtime.spawn(code, 10, capacity(4))?; let second = runtime.spawn(code, 10, capacity(4))?;
@ -388,7 +403,8 @@ mod tests {
first, first,
Message { Message {
kind: 1, kind: 1,
value: i64::try_from(second.0)?, value: 0,
target_ref: Some(second),
}, },
)?; )?;
for _ in 0..100 { for _ in 0..100 {
@ -411,23 +427,48 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let runaway = publish( let runaway = publish(
&mut runtime, &mut runtime,
"pub fn handle(state, kind, value) { loop {} }", "pub fn handle(state, kind, value, target_ref) { loop {} }",
)?; )?;
let healthy = publish( let healthy = publish(
&mut runtime, &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 first = runtime.spawn(runaway, 0, capacity(1))?;
let second = runtime.spawn(healthy, 0, capacity(2))?; let second = runtime.spawn(healthy, 0, capacity(2))?;
runtime.send(first, Message { kind: 0, value: 1 })?; runtime.send(
runtime.send(second, Message { kind: 0, value: 42 })?; first,
Message {
kind: 0,
value: 1,
target_ref: None,
},
)?;
runtime.send(
second,
Message {
kind: 0,
value: 42,
target_ref: None,
},
)?;
for _ in 0..100 { for _ in 0..100 {
runtime.tick(capacity(5)); runtime.tick(capacity(5));
} }
assert_eq!(runtime.inspect(first)?.status, Status::Running); assert_eq!(runtime.inspect(first)?.status, Status::Running);
assert_eq!(runtime.inspect(first)?.state, 0); assert_eq!(runtime.inspect(first)?.state, 0);
assert_eq!(runtime.inspect(second)?.state, 42); 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)?; runtime.stop(first)?;
assert_eq!(runtime.inspect(first)?.status, Status::Stopped); assert_eq!(runtime.inspect(first)?.status, Status::Stopped);
runtime.restart(first)?; runtime.restart(first)?;
@ -442,10 +483,17 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let implementation = publish( let implementation = publish(
&mut runtime, &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))?; 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)); runtime.tick(capacity(3));
let running = runtime.inspect(actor)?; let running = runtime.inspect(actor)?;
@ -467,7 +515,14 @@ mod tests {
runtime.cancel(actor)?; runtime.cancel(actor)?;
let cancelled = runtime.inspect(actor)?; let cancelled = runtime.inspect(actor)?;
assert_eq!(cancelled.execution, None); 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(()) Ok(())
} }
@ -476,14 +531,15 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let code = publish( let code = publish(
&mut runtime, &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))?; let actor = runtime.spawn(code, 5, capacity(4))?;
runtime.send( runtime.send(
actor, actor,
Message { Message {
kind: 0, kind: 0,
value: i64::try_from(actor.0)?, value: 0,
target_ref: Some(actor),
}, },
)?; )?;
for _ in 0..100 { for _ in 0..100 {
@ -539,14 +595,28 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let code = runtime.publish( let code = runtime.publish(
CodeStore::default().compile_with_module( 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", "test-gate:v1",
module, module,
)?, )?,
)?; )?;
let actor = runtime.spawn(code, 10, capacity(2))?; let actor = runtime.spawn(code, 10, capacity(2))?;
runtime.send(actor, Message { kind: 1, value: 0 })?; runtime.send(
runtime.send(actor, Message { kind: 0, value: 3 })?; actor,
Message {
kind: 1,
value: 0,
target_ref: None,
},
)?;
runtime.send(
actor,
Message {
kind: 0,
value: 3,
target_ref: None,
},
)?;
for _ in 0..100 { for _ in 0..100 {
runtime.tick(capacity(5)); runtime.tick(capacity(5));
} }
@ -554,7 +624,14 @@ mod tests {
assert_eq!(pending.status, Status::Awaiting); assert_eq!(pending.status, Status::Awaiting);
assert_eq!(pending.state, 10); assert_eq!(pending.state, 10);
assert_eq!(pending.queued.len(), 1); 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); let polls = gate.polls.load(Ordering::SeqCst);
for _ in 0..10 { for _ in 0..10 {
runtime.tick(capacity(5)); runtime.tick(capacity(5));
@ -586,7 +663,14 @@ mod tests {
gate.ready.store(false, Ordering::SeqCst); gate.ready.store(false, Ordering::SeqCst);
gate.dropped.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 { for _ in 0..100 {
runtime.tick(capacity(5)); runtime.tick(capacity(5));
} }
@ -597,7 +681,14 @@ mod tests {
assert_eq!(cancelled.mailbox, pending.mailbox); assert_eq!(cancelled.mailbox, pending.mailbox);
assert_eq!(cancelled.state, 20); assert_eq!(cancelled.state, 20);
assert_eq!(cancelled.status, Status::Idle); 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()); assert!(cancelled.reserved.is_none());
Ok(()) Ok(())
} }
@ -607,10 +698,17 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let implementation = publish( let implementation = publish(
&mut runtime, &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))?; 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)); runtime.tick(capacity(1));
let before_pause = runtime.inspect(actor)?; let before_pause = runtime.inspect(actor)?;
assert_eq!(before_pause.status, Status::Running); assert_eq!(before_pause.status, Status::Running);
@ -655,12 +753,30 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let code = publish( let code = publish(
&mut runtime, &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))?; let actor = runtime.spawn(code, 5, capacity(1))?;
runtime.send(actor, Message { kind: 0, value: 0 })?; runtime.send(
assert!(runtime.send(actor, Message { kind: 0, value: 1 }).is_err()); actor,
assert!(runtime.inspect(ActorId(999)).is_err()); 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 { for _ in 0..100 {
runtime.tick(capacity(5)); runtime.tick(capacity(5));
} }
@ -680,14 +796,15 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let code = publish( let code = publish(
&mut runtime, &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))?; let actor = runtime.spawn(code, 5, capacity(1))?;
runtime.send( runtime.send(
actor, actor,
Message { Message {
kind: 0, kind: 0,
value: i64::try_from(actor.0)?, value: 0,
target_ref: Some(actor),
}, },
)?; )?;
for _ in 0..100 { for _ in 0..100 {
@ -704,9 +821,19 @@ mod tests {
#[test] #[test]
fn invalid_transition_shape_fails_without_committing() -> ExperimentResult<()> { fn invalid_transition_shape_fails_without_committing() -> ExperimentResult<()> {
let mut runtime = Runtime::default(); 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))?; 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 { for _ in 0..100 {
runtime.tick(capacity(5)); runtime.tick(capacity(5));
} }
@ -720,16 +847,25 @@ mod tests {
#[test] #[test]
fn behavior_installation_waits_for_handler_boundary() -> ExperimentResult<()> { fn behavior_installation_waits_for_handler_boundary() -> ExperimentResult<()> {
let mut code = CodeStore::default(); let mut code = CodeStore::default();
let first = code.compile("pub fn handle(state, kind, value) { (state + value, []) }")?; let first =
let second = code.compile("pub fn handle(state, kind, value, target_ref) { (state + value, []) }")?;
code.compile("pub fn handle(state, kind, value) { (state + value * 10, []) }")?; let second = code.compile(
"pub fn handle(state, kind, value, target_ref) { (state + value * 10, []) }",
)?;
let first_id = first.implementation_id; let first_id = first.implementation_id;
let second_id = second.implementation_id; let second_id = second.implementation_id;
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
runtime.publish(first)?; runtime.publish(first)?;
runtime.publish(second)?; runtime.publish(second)?;
let actor = runtime.spawn(first_id, 0, capacity(2))?; 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)); runtime.tick(capacity(1));
let active = runtime.inspect(actor)?; let active = runtime.inspect(actor)?;
assert_eq!(active.implementation, first_id); assert_eq!(active.implementation, first_id);
@ -738,7 +874,14 @@ mod tests {
runtime.tick(capacity(10)); runtime.tick(capacity(10));
} }
assert_eq!(runtime.inspect(actor)?.state, 1); 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)?; runtime.install(actor, second_id)?;
assert_eq!(runtime.inspect(actor)?.implementation, second_id); assert_eq!(runtime.inspect(actor)?.implementation, second_id);
for _ in 0..100 { for _ in 0..100 {
@ -761,11 +904,11 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let old_implementation = publish( let old_implementation = publish(
&mut runtime, &mut runtime,
"pub fn handle(state, kind, value) { (state + value, []) }", "pub fn handle(state, kind, value, target_ref) { (state + value, []) }",
)?; )?;
let new_implementation = publish( let new_implementation = publish(
&mut runtime, &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))?; let actor = runtime.spawn(old_implementation, 7, capacity(2))?;
@ -790,15 +933,29 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let old_implementation = publish( let old_implementation = publish(
&mut runtime, &mut runtime,
"pub fn handle(state, kind, value) { (state + value, []) }", "pub fn handle(state, kind, value, target_ref) { (state + value, []) }",
)?; )?;
let new_implementation = publish( let new_implementation = publish(
&mut runtime, &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))?; let actor = runtime.spawn(old_implementation, 0, capacity(4))?;
runtime.send(actor, Message { kind: 0, value: 1 })?; runtime.send(
runtime.send(actor, Message { kind: 0, value: 2 })?; actor,
Message {
kind: 0,
value: 1,
target_ref: None,
},
)?;
runtime.send(
actor,
Message {
kind: 0,
value: 2,
target_ref: None,
},
)?;
runtime.tick(capacity(1)); runtime.tick(capacity(1));
assert_eq!(runtime.inspect(actor)?.status, Status::Running); assert_eq!(runtime.inspect(actor)?.status, Status::Running);
@ -808,12 +965,30 @@ mod tests {
assert_eq!(cancelled.state, 0); assert_eq!(cancelled.state, 0);
assert_eq!( assert_eq!(
cancelled.queued, 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()); assert!(cancelled.reserved.is_none());
runtime.install(actor, new_implementation)?; 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 { for _ in 0..100 {
runtime.tick(capacity(10)); runtime.tick(capacity(10));
} }
@ -834,11 +1009,11 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let old_implementation = publish( let old_implementation = publish(
&mut runtime, &mut runtime,
"pub fn handle(state, kind, value) { (state + value, []) }", "pub fn handle(state, kind, value, target_ref) { (state + value, []) }",
)?; )?;
let new_implementation = publish( let new_implementation = publish(
&mut runtime, &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))?; let actor = runtime.spawn(old_implementation, 7, capacity(2))?;
@ -869,14 +1044,21 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let old_implementation = publish( let old_implementation = publish(
&mut runtime, &mut runtime,
"pub fn handle(state, kind, value) { (state + value, []) }", "pub fn handle(state, kind, value, target_ref) { (state + value, []) }",
)?; )?;
let new_implementation = publish( let new_implementation = publish(
&mut runtime, &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))?; 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)); runtime.tick(capacity(1));
let called = Cell::new(false); let called = Cell::new(false);
@ -908,14 +1090,22 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let code = runtime.publish( let code = runtime.publish(
CodeStore::default().compile_with_module( 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", "test-fail-once:v1",
module, module,
)?, )?,
)?; )?;
let actor = runtime.spawn(code, 10, capacity(3))?; let actor = runtime.spawn(code, 10, capacity(3))?;
let first = Message { kind: 1, value: 2 }; let first = Message {
let second = Message { kind: 1, value: 3 }; kind: 1,
value: 2,
target_ref: None,
};
let second = Message {
kind: 1,
value: 3,
target_ref: None,
};
runtime.send(actor, first)?; runtime.send(actor, first)?;
runtime.send(actor, second)?; runtime.send(actor, second)?;
for _ in 0..100 { for _ in 0..100 {
@ -927,7 +1117,18 @@ mod tests {
assert_eq!(failed.state, 10); assert_eq!(failed.state, 10);
assert_eq!(failed.reserved, Some(first)); assert_eq!(failed.reserved, Some(first));
assert_eq!(failed.queued, vec![second]); 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!( assert!(
runtime runtime
.drain_events() .drain_events()
@ -937,7 +1138,7 @@ mod tests {
runtime.restart(actor)?; runtime.restart(actor)?;
let restarted = runtime.inspect(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.mailbox, failed.mailbox);
assert_eq!(restarted.implementation, failed.implementation); assert_eq!(restarted.implementation, failed.implementation);
assert_eq!(restarted.incarnation, failed.incarnation + 1); assert_eq!(restarted.incarnation, failed.incarnation + 1);
@ -961,12 +1162,24 @@ mod tests {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();
let code = publish( let code = publish(
&mut runtime, &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 actor = runtime.spawn(code, 0, capacity(3))?;
let first = Message { kind: 0, value: 1 }; let first = Message {
let second = Message { kind: 0, value: 2 }; kind: 0,
let third = Message { kind: 0, value: 3 }; 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, first)?;
runtime.send(actor, second)?; runtime.send(actor, second)?;
runtime.send(actor, third)?; runtime.send(actor, third)?;
@ -984,7 +1197,18 @@ mod tests {
assert_eq!(stopped.state, 0); assert_eq!(stopped.state, 0);
assert_eq!(stopped.reserved, Some(first)); assert_eq!(stopped.reserved, Some(first));
assert_eq!(stopped.queued, vec![second, third]); 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)?; runtime.restart(actor)?;
let restarted = runtime.inspect(actor)?; let restarted = runtime.inspect(actor)?;

View File

@ -6,7 +6,7 @@ use std::num::NonZeroUsize;
use std::thread; use std::thread;
use std::time::Duration; 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_image_api::ImageApi;
use runa_protocol::{ use runa_protocol::{
ActorSnapshotDto, ActorStatusDto, BehaviorDto, BehaviorVersionDto, Command, ErrorCode, ActorSnapshotDto, ActorStatusDto, BehaviorDto, BehaviorVersionDto, Command, ErrorCode,
@ -339,16 +339,18 @@ impl Server {
.ok_or_else(|| invalid_request("mailbox_capacity must be positive"))?; .ok_or_else(|| invalid_request("mailbox_capacity must be positive"))?;
self.image self.image
.spawn(&behavior_name, state, capacity) .spawn(&behavior_name, state, capacity)
.map(|actor_id| Reply::ActorSpawned { .map(|actor_ref| Reply::ActorSpawned {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
}) })
.map_err(api_error) .map_err(api_error)
} }
Command::Send { actor_id, message } => self Command::Send { actor_ref, message } => {
.image let actor_ref = parse_actor_ref(&actor_ref)?;
.send(ActorId(actor_id), message_from_dto(message)) self.image
.map(|()| Reply::Acknowledged) .send(actor_ref, message_from_dto(message)?)
.map_err(api_error), .map(|()| Reply::Acknowledged)
.map_err(api_error)
}
Command::Tick { Command::Tick {
ticks, ticks,
instruction_budget, instruction_budget,
@ -358,53 +360,67 @@ impl Server {
self.image.tick(ticks, budget); self.image.tick(ticks, budget);
Ok(Reply::TickCompleted { ticks }) Ok(Reply::TickCompleted { ticks })
} }
Command::InspectActor { actor_id } => self Command::InspectActor { actor_ref } => {
.image let actor_ref = parse_actor_ref(&actor_ref)?;
.inspect(ActorId(actor_id)) self.image
.map(|snapshot| actor_snapshot_dto(&snapshot)) .inspect(actor_ref)
.map(Reply::Actor) .map(|snapshot| actor_snapshot_dto(&snapshot))
.map_err(api_error), .map(Reply::Actor)
.map_err(api_error)
}
Command::ListActors => Ok(Reply::Actors( Command::ListActors => Ok(Reply::Actors(
self.image.actors().iter().map(actor_snapshot_dto).collect(), self.image.actors().iter().map(actor_snapshot_dto).collect(),
)), )),
Command::GetEvents => Ok(Reply::Events( Command::GetEvents => Ok(Reply::Events(
self.image.events().iter().map(event_dto).collect(), self.image.events().iter().map(event_dto).collect(),
)), )),
Command::StopActor { actor_id } => self Command::StopActor { actor_ref } => {
.image let actor_ref = parse_actor_ref(&actor_ref)?;
.stop(ActorId(actor_id)) self.image
.map(|()| Reply::Acknowledged) .stop(actor_ref)
.map_err(api_error), .map(|()| Reply::Acknowledged)
Command::RestartActor { actor_id } => self .map_err(api_error)
.image }
.restart(ActorId(actor_id)) Command::RestartActor { actor_ref } => {
.map(|()| Reply::Acknowledged) let actor_ref = parse_actor_ref(&actor_ref)?;
.map_err(api_error), self.image
Command::PauseActor { actor_id } => self .restart(actor_ref)
.image .map(|()| Reply::Acknowledged)
.pause(ActorId(actor_id)) .map_err(api_error)
.map(|()| Reply::Acknowledged) }
.map_err(api_error), Command::PauseActor { actor_ref } => {
Command::ResumeActor { actor_id } => self let actor_ref = parse_actor_ref(&actor_ref)?;
.image self.image
.resume(ActorId(actor_id)) .pause(actor_ref)
.map(|()| Reply::Acknowledged) .map(|()| Reply::Acknowledged)
.map_err(api_error), .map_err(api_error)
Command::CancelActor { actor_id } => self }
.image Command::ResumeActor { actor_ref } => {
.cancel(ActorId(actor_id)) let actor_ref = parse_actor_ref(&actor_ref)?;
.map(|()| Reply::Acknowledged) self.image
.map_err(api_error), .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 { Command::InstallBehavior {
actor_id, actor_ref,
behavior_name, behavior_name,
} => self } => {
.image let actor_ref = parse_actor_ref(&actor_ref)?;
.install(ActorId(actor_id), &behavior_name) self.image
.map(|implementation_id| Reply::BehaviorInstalled { .install(actor_ref, &behavior_name)
implementation_id: implementation_id.to_string(), .map(|implementation_id| Reply::BehaviorInstalled {
}) implementation_id: implementation_id.to_string(),
.map_err(api_error), })
.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 { fn actor_snapshot_dto(snapshot: &runa_actor::ActorSnapshot) -> ActorSnapshotDto {
ActorSnapshotDto { ActorSnapshotDto {
actor_id: snapshot.id.0, actor_ref: snapshot.actor_ref.as_str().to_owned(),
mailbox_id: snapshot.mailbox.0, mailbox_id: snapshot.mailbox.0,
implementation_id: snapshot.implementation.to_string(), implementation_id: snapshot.implementation.to_string(),
incarnation: snapshot.incarnation, incarnation: snapshot.incarnation,
@ -454,49 +470,57 @@ fn message_dto(message: Message) -> MessageDto {
MessageDto { MessageDto {
kind: message.kind, kind: message.kind,
value: message.value, value: message.value,
target_ref: message
.target_ref
.map(|actor_ref| actor_ref.as_str().to_owned()),
} }
} }
fn message_from_dto(message: MessageDto) -> Message { fn message_from_dto(message: MessageDto) -> Result<Message, ProtocolError> {
Message { Ok(Message {
kind: message.kind, kind: message.kind,
value: message.value, value: message.value,
} target_ref: message
.target_ref
.as_deref()
.map(parse_actor_ref)
.transpose()?,
})
} }
fn event_dto(event: &Event) -> EventDto { fn event_dto(event: &Event) -> EventDto {
match event { match event {
Event::Spawned(actor_id) => EventDto::Spawned { Event::Spawned(actor_ref) => EventDto::Spawned {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
}, },
Event::Committed(actor_id) => EventDto::Committed { Event::Committed(actor_ref) => EventDto::Committed {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
}, },
Event::Failed(actor_id, message) => EventDto::Failed { Event::Failed(actor_ref, message) => EventDto::Failed {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
message: message.clone(), message: message.clone(),
}, },
Event::Stopped(actor_id) => EventDto::Stopped { Event::Stopped(actor_ref) => EventDto::Stopped {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
}, },
Event::Restarted(actor_id) => EventDto::Restarted { Event::Restarted(actor_ref) => EventDto::Restarted {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
}, },
Event::Paused(actor_id) => EventDto::Paused { Event::Paused(actor_ref) => EventDto::Paused {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
}, },
Event::Resumed(actor_id) => EventDto::Resumed { Event::Resumed(actor_ref) => EventDto::Resumed {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
}, },
Event::InvocationCancelled(actor_id) => EventDto::InvocationCancelled { Event::InvocationCancelled(actor_ref) => EventDto::InvocationCancelled {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
}, },
Event::BehaviorInstalled(actor_id, implementation_id) => EventDto::BehaviorInstalled { Event::BehaviorInstalled(actor_ref, implementation_id) => EventDto::BehaviorInstalled {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
implementation_id: implementation_id.to_string(), implementation_id: implementation_id.to_string(),
}, },
Event::StateMigrated(actor_id, previous_state, current_state) => EventDto::StateMigrated { Event::StateMigrated(actor_ref, previous_state, current_state) => EventDto::StateMigrated {
actor_id: actor_id.0, actor_ref: actor_ref.as_str().to_owned(),
previous_state: *previous_state, previous_state: *previous_state,
current_state: *current_state, current_state: *current_state,
}, },
@ -533,6 +557,10 @@ fn invalid_request(message: impl Into<String>) -> ProtocolError {
ProtocolError::new(ErrorCode::InvalidRequest, message) ProtocolError::new(ErrorCode::InvalidRequest, message)
} }
fn parse_actor_ref(value: &str) -> Result<ActorRef, ProtocolError> {
ActorRef::parse(value).ok_or_else(|| invalid_request("actor_ref must be a valid Nano ID"))
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
@ -552,7 +580,8 @@ mod tests {
let response = server.dispatch(request(Command::PublishBehavior { let response = server.dispatch(request(Command::PublishBehavior {
name: "counter".to_owned(), name: "counter".to_owned(),
expected_revision: None, 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() { match response.result.unwrap() {
Reply::Behavior(behavior) => behavior, Reply::Behavior(behavior) => behavior,
@ -572,15 +601,20 @@ mod tests {
state: 5, state: 5,
mailbox_capacity: 2, mailbox_capacity: 2,
})); }));
let actor_id = match spawned.result.unwrap() { let actor_ref = match spawned.result.unwrap() {
Reply::ActorSpawned { actor_id } => actor_id, Reply::ActorSpawned { actor_ref } => actor_ref,
other => panic!("unexpected reply: {other:?}"), other => panic!("unexpected reply: {other:?}"),
}; };
assert_eq!(actor_id, 1); assert_eq!(actor_ref.len(), 21);
assert!(!actor_ref.parse::<u64>().is_ok());
let sent = server.dispatch(request(Command::Send { let sent = server.dispatch(request(Command::Send {
actor_id, actor_ref: actor_ref.clone(),
message: MessageDto { kind: 0, value: 7 }, message: MessageDto {
kind: 0,
value: 7,
target_ref: None,
},
})); }));
assert_eq!(sent.result, Ok(Reply::Acknowledged)); assert_eq!(sent.result, Ok(Reply::Acknowledged));
server.dispatch(request(Command::Tick { server.dispatch(request(Command::Tick {
@ -588,7 +622,9 @@ mod tests {
instruction_budget: 10, 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() { match inspected.result.unwrap() {
Reply::Actor(snapshot) => assert_eq!(snapshot.state, 12), Reply::Actor(snapshot) => assert_eq!(snapshot.state, 12),
other => panic!("unexpected reply: {other:?}"), other => panic!("unexpected reply: {other:?}"),
@ -597,7 +633,7 @@ mod tests {
let events = server.dispatch(request(Command::GetEvents)); let events = server.dispatch(request(Command::GetEvents));
match events.result.unwrap() { match events.result.unwrap() {
Reply::Events(events) => assert!(events.iter().any( 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:?}"), other => panic!("unexpected reply: {other:?}"),
} }
@ -619,14 +655,16 @@ mod tests {
ErrorCode::UnsupportedVersion 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); assert_eq!(missing.result.unwrap_err().code, ErrorCode::NotFound);
publish(&mut server); publish(&mut server);
let conflict = server.dispatch(request(Command::PublishBehavior { let conflict = server.dispatch(request(Command::PublishBehavior {
name: "counter".to_owned(), name: "counter".to_owned(),
expected_revision: Some(99), 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); assert_eq!(conflict.result.unwrap_err().code, ErrorCode::Conflict);
} }
@ -638,7 +676,8 @@ mod tests {
let updated = server.dispatch(request(Command::PublishBehavior { let updated = server.dispatch(request(Command::PublishBehavior {
name: "counter".to_owned(), name: "counter".to_owned(),
expected_revision: Some(first.revision), 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(_)))); assert!(matches!(updated.result, Ok(Reply::Behavior(_))));
@ -649,11 +688,12 @@ mod tests {
assert_eq!( assert_eq!(
source.result, source.result,
Ok(Reply::BehaviorSource { 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 { .dispatch(request(Command::Spawn {
behavior_name: "counter".to_owned(), behavior_name: "counter".to_owned(),
state: 3, state: 3,
@ -662,11 +702,11 @@ mod tests {
.result .result
.unwrap() .unwrap()
{ {
Reply::ActorSpawned { actor_id } => actor_id, Reply::ActorSpawned { actor_ref } => actor_ref,
other => panic!("unexpected reply: {other:?}"), other => panic!("unexpected reply: {other:?}"),
}; };
let installed = server.dispatch(request(Command::InstallBehavior { let installed = server.dispatch(request(Command::InstallBehavior {
actor_id, actor_ref,
behavior_name: "counter".to_owned(), behavior_name: "counter".to_owned(),
})); }));
assert!(matches!( assert!(matches!(
@ -680,7 +720,8 @@ mod tests {
let publish_request = request(Command::PublishBehavior { let publish_request = request(Command::PublishBehavior {
name: "counter".to_owned(), name: "counter".to_owned(),
expected_revision: None, 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 { let mut spawn_request = request(Command::Spawn {
behavior_name: "counter".to_owned(), behavior_name: "counter".to_owned(),
@ -717,7 +758,10 @@ mod tests {
vec!["test-request", "spawn-request", "list-request"] vec!["test-request", "spawn-request", "list-request"]
); );
assert!(matches!(responses[0].result, Ok(Reply::Behavior(_)))); 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!( assert!(matches!(
&responses[2].result, &responses[2].result,
Ok(Reply::Actors(actors)) if actors.len() == 1 && actors[0].state == 4 Ok(Reply::Actors(actors)) if actors.len() == 1 && actors[0].state == 4

View File

@ -23,7 +23,8 @@ fn local_client_launches_server_and_operates_the_same_live_image() {
.request(Command::PublishBehavior { .request(Command::PublishBehavior {
name: "counter".to_owned(), name: "counter".to_owned(),
expected_revision: None, 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(); .unwrap();
assert_eq!(published.request_id, "runa-client-1"); assert_eq!(published.request_id, "runa-client-1");
@ -37,12 +38,20 @@ fn local_client_launches_server_and_operates_the_same_live_image() {
}) })
.unwrap(); .unwrap();
assert_eq!(spawned.request_id, "runa-client-2"); 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 client
.request(Command::Send { .request(Command::Send {
actor_id: 1, actor_ref: actor_ref.clone(),
message: MessageDto { kind: 0, value: 7 }, message: MessageDto {
kind: 0,
value: 7,
target_ref: None,
},
}) })
.unwrap(); .unwrap();
client client
@ -51,9 +60,7 @@ fn local_client_launches_server_and_operates_the_same_live_image() {
instruction_budget: 10, instruction_budget: 10,
}) })
.unwrap(); .unwrap();
let inspected = client let inspected = client.request(Command::InspectActor { actor_ref }).unwrap();
.request(Command::InspectActor { actor_id: 1 })
.unwrap();
assert!(matches!( assert!(matches!(
inspected.result, inspected.result,
Ok(Reply::Actor(actor)) if actor.state == 12 Ok(Reply::Actor(actor)) if actor.state == 12

View File

@ -6,3 +6,4 @@ edition = "2024"
[dependencies] [dependencies]
runa-protocol = { path = "../runa-protocol" } runa-protocol = { path = "../runa-protocol" }
runa-server = { path = "../runa-server" } runa-server = { path = "../runa-server" }
rustyline = "15.0"

View File

@ -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 std::num::NonZeroUsize;
use runa_protocol::{Command, MessageDto, PROTOCOL_VERSION, Reply, Request, Response}; use runa_protocol::{Command, MessageDto, PROTOCOL_VERSION, Reply, Request, Response};
use runa_server::Server; 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 MAILBOX_CAPACITY: usize = 64;
const DEFAULT_TICK_BUDGET: usize = 32; const DEFAULT_TICK_BUDGET: usize = 32;
const DEFAULT_BEHAVIOR: &str = "counter"; const COUNTER_BEHAVIOR: &str = "counter";
#[derive(Default)]
struct ReplState {
bindings: BTreeMap<String, String>,
}
#[derive(Default)]
struct ReplHelper {
behaviors: Vec<String>,
bindings: Vec<(String, String)>,
}
impl ReplHelper {
fn set_context(&mut self, behaviors: Vec<String>, bindings: &BTreeMap<String, String>) {
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<String> {
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<String> = 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<String> {
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<Self::Candidate>)> {
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<dyn std::error::Error>> { fn main() -> Result<(), Box<dyn std::error::Error>> {
let mut server = Server::new(); let mut server = Server::new();
let behavior = match dispatch( let behavior = match dispatch(
&mut server, &mut server,
Command::PublishBehavior { Command::PublishBehavior {
name: DEFAULT_BEHAVIOR.to_owned(), name: COUNTER_BEHAVIOR.to_owned(),
expected_revision: None, expected_revision: None,
source: include_str!("../../../runa-std/actors/counter.rn").to_owned(), source: include_str!("../../../runa-std/actors/counter.rn").to_owned(),
}, },
@ -27,22 +152,18 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
"Loaded behavior `{}` revision {}.", "Loaded behavior `{}` revision {}.",
behavior.name, behavior.revision behavior.name, behavior.revision
); );
println!("Type `help` for commands."); println!("Type `help` for commands. Use Up/Down for history and Tab for completion.");
run_repl( run_interactive_repl(&mut server)?;
io::stdin().lock(),
io::stdout().lock(),
&mut server,
DEFAULT_BEHAVIOR,
)?;
Ok(()) Ok(())
} }
#[cfg(test)]
fn run_repl<R: BufRead, W: Write>( fn run_repl<R: BufRead, W: Write>(
mut input: R, mut input: R,
mut output: W, mut output: W,
server: &mut Server, server: &mut Server,
default_behavior: &str,
) -> io::Result<()> { ) -> io::Result<()> {
let mut state = ReplState::default();
let mut line = String::new(); let mut line = String::new();
loop { loop {
write!(output, "runa> ")?; write!(output, "runa> ")?;
@ -52,129 +173,202 @@ fn run_repl<R: BufRead, W: Write>(
writeln!(output)?; writeln!(output)?;
break; break;
} }
let mut words = line.split_whitespace(); if !execute_line(&line, &mut output, server, &mut state)? {
let Some(command) = words.next() else { break;
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::<u64, 1>(&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?;
} }
Ok(()) Ok(())
} }
fn run_interactive_repl(server: &mut Server) -> Result<(), Box<dyn std::error::Error>> {
let mut editor = Editor::<ReplHelper, DefaultHistory>::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<W: Write>(
line: &str,
output: &mut W,
server: &mut Server,
state: &mut ReplState,
) -> io::Result<bool> {
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<W: Write>( fn lifecycle_command<W: Write>(
output: &mut W, output: &mut W,
args: &[&str], args: &[&str],
server: &mut Server, server: &mut Server,
operation: impl FnOnce(u64) -> Command, state: &ReplState,
operation: impl FnOnce(String) -> Command,
) -> io::Result<()> { ) -> io::Result<()> {
match parse_exact::<u64, 1>(args) { match parse_actor_argument(args, state) {
Ok([id]) => match dispatch(server, operation(id)) { Ok(actor_ref) => match dispatch(server, operation(actor_ref)) {
Ok(Reply::Acknowledged) => writeln!(output, "ok"), Ok(Reply::Acknowledged) => writeln!(output, "ok"),
Ok(other) => unexpected_reply(output, other), Ok(other) => unexpected_reply(output, other),
Err(error) => writeln!(output, "error: {error}"), Err(error) => writeln!(output, "error: {error}"),
@ -183,56 +377,66 @@ fn lifecycle_command<W: Write>(
} }
} }
fn spawn_command<W: Write>( fn spawn_command<W: Write>(output: &mut W, args: &[&str], server: &mut Server) -> io::Result<()> {
output: &mut W, let [behavior_name, initial_state] = args else {
args: &[&str], return writeln!(
default_behavior: &str, output,
"error: usage: spawn <behavior-name> <initial-state>"
);
};
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, server: &mut Server,
) -> io::Result<()> { ) -> Result<String, String> {
let (state_arg, behavior_name) = match args { let state = initial_state
[state] => (*state, default_behavior), .parse()
[state, behavior_name] => (*state, *behavior_name), .map_err(|_| format!("invalid initial state: `{initial_state}`"))?;
_ => { match dispatch(
return writeln!(
output,
"error: usage: spawn <initial-state> [behavior-name]"
);
}
};
let state = match state_arg.parse() {
Ok(state) => state,
Err(_) => return writeln!(output, "error: invalid initial state: `{state_arg}`"),
};
let response = dispatch(
server, server,
Command::Spawn { Command::Spawn {
behavior_name: behavior_name.to_owned(), behavior_name: behavior_name.to_owned(),
state, state,
mailbox_capacity: MAILBOX_CAPACITY, mailbox_capacity: MAILBOX_CAPACITY,
}, },
); )? {
match response { Reply::ActorSpawned { actor_ref } => Ok(actor_ref),
Ok(Reply::ActorSpawned { actor_id }) => { other => Err(format!("unexpected protocol response: {other:?}")),
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}"),
} }
} }
fn parse_actor_argument(args: &[&str], state: &ReplState) -> Result<String, String> {
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<W: Write>(output: &mut W, args: &[&str], server: &mut Server) -> io::Result<()> { fn publish_command<W: Write>(output: &mut W, args: &[&str], server: &mut Server) -> io::Result<()> {
let [name, expected_revision, source_path] = args else { let [name, expected_revision, source_path] = args else {
return writeln!( return writeln!(
@ -339,18 +543,26 @@ fn rollback_command<W: Write>(
} }
} }
fn install_command<W: Write>(output: &mut W, args: &[&str], server: &mut Server) -> io::Result<()> { fn install_command<W: Write>(
let [actor_id, name] = args else { output: &mut W,
return writeln!(output, "error: usage: install <actor-id> <behavior-name>"); args: &[&str],
server: &mut Server,
state: &ReplState,
) -> io::Result<()> {
let [actor_arg, name] = args else {
return writeln!(
output,
"error: usage: install <actor-ref-or-name> <behavior-name>"
);
}; };
let actor_id = match actor_id.parse::<u64>() { let actor_ref = match parse_actor_argument(&[*actor_arg], state) {
Ok(id) => id, Ok(actor_ref) => actor_ref,
Err(_) => return writeln!(output, "error: invalid actor ID"), Err(error) => return writeln!(output, "error: {error}"),
}; };
let implementation = match dispatch( let implementation = match dispatch(
server, server,
Command::InstallBehavior { Command::InstallBehavior {
actor_id, actor_ref: actor_ref.clone(),
behavior_name: (*name).to_owned(), behavior_name: (*name).to_owned(),
}, },
) { ) {
@ -367,8 +579,8 @@ fn install_command<W: Write>(output: &mut W, args: &[&str], server: &mut Server)
Ok(Reply::Behavior(behavior)) => { Ok(Reply::Behavior(behavior)) => {
writeln!( writeln!(
output, output,
"installed `{}` revision {} on actor {} ({implementation})", "installed `{}` revision {} on actor @{} ({implementation})",
behavior.name, behavior.revision, actor_id behavior.name, behavior.revision, actor_ref
) )
} }
Ok(other) => unexpected_reply(output, other), Ok(other) => unexpected_reply(output, other),
@ -389,23 +601,6 @@ fn unexpected_reply<W: Write>(output: &mut W, reply: Reply) -> io::Result<()> {
writeln!(output, "error: unexpected protocol response: {reply:?}") writeln!(output, "error: unexpected protocol response: {reply:?}")
} }
fn parse_exact<T: std::str::FromStr, const N: usize>(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::<T>()
.map_err(|_| format!("invalid argument {}: `{arg}`", index + 1))
})
.collect::<Result<Vec<_>, _>>()?;
parsed
.try_into()
.map_err(|_| "invalid argument count".to_owned())
}
fn parse_tick_args(args: &[&str]) -> Result<(usize, NonZeroUsize), String> { fn parse_tick_args(args: &[&str]) -> Result<(usize, NonZeroUsize), String> {
match args { match args {
[count] => Ok(( [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> { fn parse_send_args(
if args.len() != 3 { args: &[&str],
return Err(format!("expected 3 arguments, got {}", args.len())); state: &ReplState,
} ) -> Result<(String, i64, i64, Option<String>), String> {
let actor = args[0] 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 <actor> <kind> <value> [--to <target>]".to_owned()),
};
let actor_ref = parse_actor_argument(&[actor_arg], state)?;
let kind = kind_arg
.parse() .parse()
.map_err(|_| format!("invalid actor ID: `{}`", args[0]))?; .map_err(|_| format!("invalid message kind: `{kind_arg}`"))?;
let kind = args[1] let value = value_arg
.parse() .parse()
.map_err(|_| format!("invalid message kind: `{}`", args[1]))?; .map_err(|_| format!("invalid message value: `{value_arg}`"))?;
let value = args[2] let target_ref = target_arg
.parse() .map(|target| parse_actor_argument(&[target], state))
.map_err(|_| format!("invalid message value: `{}`", args[2]))?; .transpose()?;
Ok((actor, kind, value)) Ok((actor_ref, kind, value, target_ref))
} }
fn print_help<W: Write>(output: &mut W) -> io::Result<()> { fn print_help<W: Write>(output: &mut W) -> io::Result<()> {
writeln!( writeln!(
output, output,
"Commands:\n\ "Commands:\n\
spawn <initial-state> [behavior-name]\n\ let <name> = spawn <behavior-name> <initial-state>\n\
spawn <behavior-name> <initial-state>\n\
behaviors\n\ behaviors\n\
publish <behavior-name> <new|expected-revision> <source-file>\n\ publish <behavior-name> <new|expected-revision> <source-file>\n\
history <behavior-name>\n\ history <behavior-name>\n\
rollback <behavior-name> <expected-revision> <target-revision>\n\ rollback <behavior-name> <expected-revision> <target-revision>\n\
install <actor-id> <behavior-name>\n\ install <actor-ref-or-name> <behavior-name>\n\
send <actor-id> <kind> <value>\n\ send <actor-ref-or-name> <kind> <value> [--to <actor-ref-or-name>]\n\
tick <count> [instruction-budget]\n\ tick <count> [instruction-budget]\n\
inspect <actor-id>\n\ inspect <actor-ref-or-name>\n\
actors\n\ actors\n\
events\n\ events\n\
stop <actor-id>\n\ stop <actor-ref-or-name>\n\
restart <actor-id>\n\ restart <actor-ref-or-name>\n\
pause <actor-id>\n\ pause <actor-ref-or-name>\n\
resume <actor-id>\n\ resume <actor-ref-or-name>\n\
cancel <actor-id>\n\ cancel <actor-ref-or-name>\n\
help\n\ help\n\
quit" quit"
) )
@ -478,7 +680,7 @@ mod tests {
dispatch( dispatch(
&mut server, &mut server,
Command::PublishBehavior { Command::PublishBehavior {
name: DEFAULT_BEHAVIOR.to_owned(), name: COUNTER_BEHAVIOR.to_owned(),
expected_revision: None, expected_revision: None,
source: include_str!("../../../runa-std/actors/counter.rn").to_owned(), 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<dyn std::error::Error>> { fn repl_drives_actor_messages_and_inspection() -> Result<(), Box<dyn std::error::Error>> {
let mut server = server(); let mut server = server();
let input = io::Cursor::new( 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(); 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)?; let output = String::from_utf8(output)?;
assert!(output.contains("actor_id: 1")); assert!(output.contains("sender = @"));
assert!(output.contains("actor_id: 2")); assert!(output.contains("receiver = @"));
assert!(output.contains("state: 15")); assert!(output.contains("state: 15"));
assert!(output.contains("Committed {")); assert!(output.contains("Committed {"));
assert!(output.contains("actor_id: 2")); assert!(output.contains("actor_ref: \""));
Ok(()) Ok(())
} }
#[test] #[test]
fn repl_reports_bad_input_and_continues() -> Result<(), Box<dyn std::error::Error>> { fn repl_reports_bad_input_and_continues() -> Result<(), Box<dyn std::error::Error>> {
let mut server = server(); 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(); 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)?; let output = String::from_utf8(output)?;
assert!(output.contains("usage: spawn <behavior-name> <initial-state>"));
assert!(output.contains("invalid initial state")); assert!(output.contains("invalid initial state"));
assert!(output.contains("unknown actor")); assert!(output.contains("unknown actor"));
assert!(output.contains("tick budget must be positive")); assert!(output.contains("tick budget must be positive"));
@ -527,35 +732,51 @@ mod tests {
"/../../runa-std/actors/counter.rn" "/../../runa-std/actors/counter.rn"
); );
let input = io::Cursor::new(format!( 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(); 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)?; let output = String::from_utf8(output)?;
assert!(output.contains("published `counter` revision 2")); assert!(output.contains("published `counter` revision 2"));
assert!(output.contains("revision 1:")); assert!(output.contains("revision 1:"));
assert!(output.contains("revision 2:")); assert!(output.contains("revision 2:"));
assert!(output.contains("rolled back `counter`")); assert!(output.contains("rolled back `counter`"));
assert!(output.contains("spawned ActorId(1) with `counter` revision 3")); assert!(output.contains("actor = @"));
assert!(output.contains("installed `counter` revision 3 on actor 1")); assert!(output.contains("installed `counter` revision 3 on actor @"));
assert!(output.contains("Paused {")); assert!(output.contains("Paused {"));
assert!(output.contains("Resumed {")); assert!(output.contains("Resumed {"));
assert!(output.contains("InvocationCancelled {")); assert!(output.contains("InvocationCancelled {"));
let behavior = match dispatch( let behavior = match dispatch(
&mut server, &mut server,
Command::GetBehavior { Command::GetBehavior {
name: DEFAULT_BEHAVIOR.to_owned(), name: COUNTER_BEHAVIOR.to_owned(),
}, },
)? { )? {
Reply::Behavior(behavior) => behavior, Reply::Behavior(behavior) => behavior,
other => return Err(format!("unexpected response: {other:?}").into()), other => return Err(format!("unexpected response: {other:?}").into()),
}; };
assert_eq!(behavior.revision, 3); 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, Reply::Actor(actor) => actor,
other => return Err(format!("unexpected response: {other:?}").into()), other => return Err(format!("unexpected response: {other:?}").into()),
}; };
assert_eq!(actor.implementation_id, behavior.current_implementation_id); assert_eq!(actor.implementation_id, behavior.current_implementation_id);
Ok(()) 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"]);
}
} }

View File

@ -71,19 +71,25 @@ An actor comprises distinct components:
```text ```text
Actor { Actor {
id: ActorId, // stable logical identity actor_ref: ActorRef, // opaque public reference
mailbox: MailboxId, // independent queue; may survive restart mailbox: MailboxId, // independent queue; may survive restart
state: StateRecord, // committed private state state: StateRecord, // committed private state
behavior: ImplementationId, // immutable handler implementation behavior: ImplementationId, // immutable handler implementation
incarnation: Generation, // changes on restart incarnation: Generation, // changes on restart
status: ActorStatus, status: ActorStatus,
supervisor: Option<ActorId>, supervisor: Option<ActorRef>,
} }
``` ```
### 4.1 Identity ### 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 ### 4.2 Mailbox
@ -304,7 +310,7 @@ The diagram is conceptual; Cargo dependencies should be kept minimal. A dependen
| Crate | Owns | May depend on | | 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-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-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 | | `runa-rune` | Rune compilation, VM execution, suspension, interruption hooks, language-specific diagnostics | `runa-core`, Rune libraries |

View File

@ -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 address. The TCP listener is unauthenticated and is intended only for trusted
local development. 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 non-empty `request_id` used to correlate the response. A server must reject
unsupported versions with `unsupported_version` rather than interpreting a unsupported versions with `unsupported_version` rather than interpreting a
request using a different contract. Events carry the protocol version and do request using a different contract. Events carry the protocol version and do
@ -19,19 +19,23 @@ Requests have a tagged command object:
```json ```json
{ {
"version": 1, "version": 2,
"request_id": "req-1", "request_id": "req-1",
"command": { "command": {
"operation": "send", "operation": "send",
"parameters": { "parameters": {
"actor_id": 7, "actor_ref": "xK8_pQ2-Lm5_012345678",
"message": { "kind": 2, "value": 9 } "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, retrieval, actor spawn, send, tick, inspect, actor listing, event retrieval,
stop, restart, pause, resume, cancellation, and behavior installation. stop, restart, pause, resume, cancellation, and behavior installation.
Migration closures are intentionally excluded: they are host callbacks, not 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 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 error. Events use a separate versioned envelope. Actor and behavior snapshots
are wire DTOs, not serialized runtime objects; implementation identities are 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` failures to the defined error codes and otherwise uses `runtime_failure`
rather than claiming a more specific category. 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 The `runa-client` crate launches the stdio server locally and validates
response protocol versions and request IDs. response protocol versions and request IDs.
The current validation checks the protocol version, request ID, positive The current validation checks the protocol version, request ID, actor-reference
actor/revision identifiers where required, non-empty behavior names, mailbox shape, positive revisions where required, non-empty behavior names, mailbox
capacity, and instruction budget. Behavior-name grammar, actor existence, capacity, and instruction budget. Behavior-name grammar, actor existence,
current behavior revisions, mailbox capacity at send time, and all runtime current behavior revisions, mailbox capacity at send time, and all runtime
invariants remain authoritative at the application-service boundary. invariants remain authoritative at the application-service boundary.

View File

@ -6,41 +6,44 @@ Start the interactive, in-memory runtime session with:
mise run run mise run run
``` ```
The executable loads [counter.rn](../runa-std/actors/counter.rn) as the default The executable publishes [counter.rn](../runa-std/actors/counter.rn) as the
behavior. Commands include `spawn`, `send`, `tick`, `inspect`, `actors`, `counter` behavior. Commands include `spawn`, `send`, `tick`, `inspect`, `actors`,
`events`, `stop`, `restart`, `pause`, `resume`, `cancel`, `behaviors`, `events`, `stop`, `restart`, `pause`, `resume`, `cancel`, `behaviors`,
`publish`, `history`, `rollback`, and `install`; enter `help` to see usage. `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 `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 expected current revision for an update. `rollback` requires the expected
current revision and a target historical revision. `install` selects a named 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 behavior's current implementation and fails if the actor is not at an idle
handler boundary. handler boundary. `spawn` requires a behavior name and initial state, in that
`spawn` accepts an optional behavior name and otherwise uses `counter`. All order. All commands operate on the same process-local runtime and code store.
commands operate on the same process-local runtime and code store. This CLI is The CLI uses the versioned protocol dispatcher, also available over stdio or
for local development, not a versioned external API. the loopback-only TCP listener for local development.
Example: Example:
```text ```text
runa> spawn 0 runa> let sender = spawn counter 0
spawned ActorId(1) sender = @<actor-reference>
runa> spawn 10 runa> let receiver = spawn counter 10
spawned ActorId(2) receiver = @<actor-reference>
runa> send 1 1 2 runa> send sender 1 5 --to receiver
message queued message queued
runa> tick 100 runa> tick 100
runa> inspect 1 runa> inspect sender
runa> inspect 2 runa> inspect receiver
``` ```
The sender commits state 1 and sends a structured message; the receiver commits The sender commits state 1 and sends a structured message; the receiver commits
state 15. `tick <count> [instruction-budget]` explicitly advances the scheduler. state 15. Actor references are 21-character NanoIDs; names are local REPL
The default per-actor slice budget is 32 instructions. bindings. Tab completes commands, behaviors, and bound actor references, while
Up/Down navigate command history. `tick <count> [instruction-budget]`
explicitly advances the scheduler. The default per-actor slice budget is 32
instructions.
## Current contracts ## Current contracts
- `runa-actor` owns actor, mailbox and implementation IDs, structured messages, - `runa-actor` owns actor references, mailbox and implementation IDs, structured
snapshots, statuses and lifecycle events. messages, snapshots, statuses and lifecycle events.
- `runa-core` owns stable source and implementation identity types. - `runa-core` owns stable source and implementation identity types.
- `runa-code` preserves original source, derives identities, caches compiled - `runa-code` preserves original source, derives identities, caches compiled
artifacts, and provides an in-memory named behavior registry with revisions. 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 - `runa-protocol` defines versioned transport-neutral request, response, and
event DTOs; it does not dispatch requests or access runtime state. event DTOs; it does not dispatch requests or access runtime state.
- `runa-server` dispatches those requests through `runa-image-api` and maps - `runa-server` dispatches those requests through `runa-image-api` and maps
results to protocol replies; its executable uses newline-delimited JSON over results to protocol replies; its executable accepts newline-delimited JSON
stdin/stdout with a 1 MiB request-line limit, not a network listener. 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 - `runa` provides an interactive CLI client using protocol requests through the
dispatcher. dispatcher.
@ -66,8 +69,8 @@ published source for a selected behavior revision.
An implementation is published once under an ID. Actors reference that exact An implementation is published once under an ID. Actors reference that exact
implementation. Implementation IDs use source content and the compile implementation. Implementation IDs use source content and the compile
fingerprint; actor and mailbox IDs remain process-local counters. Actor IDs and fingerprint; actor references are random NanoIDs and mailbox IDs remain
incarnation counters use checked increments. process-local counters. Incarnation counters use checked increments.
`CodeStore::publish_behavior` compiles source before appending a behavior `CodeStore::publish_behavior` compiles source before appending a behavior
revision. Creation expects no existing revision; updates require the caller's 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. change. The CLI currently creates a fresh `CodeStore` when it starts.
Prototype state is an `i64`; messages contain integer `kind` and `value` 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 ```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 The returned effect vector describes sends; handlers cannot access the runtime
or directly mutate committed state. General schema/versioned values and or directly mutate committed state. General schema/versioned values and
capability-bearing contexts are future work. capability-bearing contexts are future work.
## Scheduling and lifecycle ## 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. FIFO queues are bounded, including the reserved message in capacity accounting.
A queued message is reserved when its invocation starts. Subsequent messages A queued message is reserved when its invocation starts. Subsequent messages
stay queued, including while the actor awaits. A wake marks an awaiting actor 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. for unknown, unavailable, or full actors.
Stopping drops the invocation, retaining committed state, mailbox and reserved Stopping drops the invocation, retaining committed state, mailbox and reserved
message. Restarting a stopped or failed actor preserves its actor ID, mailbox message. Restarting a stopped or failed actor preserves its actor reference,
ID, implementation ID, committed state, and queued-message order; it increments mailbox ID, implementation ID, committed state, and queued-message order; it
its incarnation and puts the reserved message at the front of the queue. 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 Restart is explicit. Repeated failure of the same reserved message can therefore
block later messages until an operator changes the implementation or otherwise block later messages until an operator changes the implementation or otherwise
resolves it; there is no automatic poison-message policy or supervision. resolves it; there is no automatic poison-message policy or supervision.
@ -208,11 +219,10 @@ capacity, and successful message effects.
## Limits ## Limits
This is not yet a full image: no persistent behavior registry or mailboxes, This is not yet a full image: no persistent behavior registry or mailboxes,
transactional outbox, supervisors, API server, GUI, or automatic image transactional outbox, supervisors, GUI, or automatic image bootstrap. The
bootstrap. The behavior registry and implementation installation are behavior registry and implementation installation are process-local;
process-local; installation requires an idle handler boundary and does not installation requires an idle handler boundary and does not upgrade active
upgrade active execution. The example default script is bundled through execution. The example counter script is bundled through `include_str!`.
`include_str!`.
Messages are not accepted by stopped/failed actors. Messages are not accepted by stopped/failed actors.
Native functions must be nonblocking. Dropping a future proves cleanup only Native functions must be nonblocking. Dropping a future proves cleanup only

View File

@ -1,6 +1,6 @@
pub fn handle(state, kind, value) { pub fn handle(state, kind, value, target_ref) {
if kind == 1 { if kind == 1 {
(state + 1, [(value, 0, 5)]) (state + 1, [(target_ref, 0, 5)])
} else { } else {
(state + value, []) (state + value, [])
} }