From ad63e286d9c973c7b8164da2c64a7b68ed6ea3b1 Mon Sep 17 00:00:00 2001 From: Steve Biedermann Date: Mon, 2 Nov 2026 16:56:03 +0100 Subject: [PATCH] feat!: add structured actor state and messages Introduce owned portable values with distinct actor references, bounded Rune conversion, JSON REPL literals, and nested reference completion. Preserve isolation and atomic transitions across state, messages, effects, migration, inspection, and TCP clients. BREAKING CHANGE: protocol v3 replaces integer state and kind/value/target_ref messages with tagged values. Rune handlers now take (state, message) and return (new_state, [(actor_reference, message)]). The REPL send command accepts one JSON message. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 10 + Cargo.lock | 9 +- README.md | 13 +- crates/runa-actor/Cargo.toml | 1 - crates/runa-actor/src/lib.rs | 78 +-- crates/runa-code/src/lib.rs | 2 +- crates/runa-core/Cargo.toml | 2 + crates/runa-core/src/lib.rs | 3 + crates/runa-core/src/value.rs | 170 ++++++ crates/runa-image-api/src/lib.rs | 53 +- crates/runa-protocol/Cargo.toml | 1 + crates/runa-protocol/src/lib.rs | 127 +++-- crates/runa-rune/Cargo.toml | 1 + crates/runa-rune/src/lib.rs | 22 +- crates/runa-rune/src/value.rs | 185 +++++++ crates/runa-runtime/src/lib.rs | 663 ++++++++++------------- crates/runa-server/src/lib.rs | 82 ++- crates/runa-server/tests/local_client.rs | 99 +++- crates/runa/Cargo.toml | 2 + crates/runa/src/main.rs | 330 +++++++++-- docs/architecture.md | 13 +- docs/protocol.md | 52 +- docs/runtime.md | 78 ++- runa-std/actors/counter.rn | 10 +- 24 files changed, 1311 insertions(+), 695 deletions(-) create mode 100644 crates/runa-core/src/value.rs create mode 100644 crates/runa-rune/src/value.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c612f6..dc7d321 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ All notable changes to this project will be documented in this file. ### Added +- Portable structured actor values: null, booleans, signed 64-bit integers, + strings, lists, records, and distinct NanoID actor references. +- JSON state/message literals in the REPL, including named `$actor` references, + `$record` escaping, and actor-reference completion inside nested values. +- Value conversion limits and tests for nested-state isolation, copied sends, + cyclic/unsupported Rune output, and all-or-nothing structured transitions. - Reusable loopback TCP protocol client with bounded messages, correlation checks, explicit connection failures, and no automatic request retries. - Multi-client and reconnect tests proving that actors remain in the @@ -74,6 +80,10 @@ All notable changes to this project will be documented in this file. ### Changed +- Protocol v3 and Rune's two-argument `handle(state, message)` replace the + temporary integer-only state and kind/value/target_ref message envelope. + Returned sends now contain `(actor_reference, message)` pairs; migrations, + snapshots, and events preserve structured values. - The REPL now connects to the localhost server rather than embedding its own image. Start the server first; `--connect` selects a loopback endpoint. - TCP server startup publishes the bundled counter behavior once; connecting diff --git a/Cargo.lock b/Cargo.lock index 6fb2cfe..d09080f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -552,16 +552,17 @@ name = "runa" version = "0.0.0" dependencies = [ "runa-client", + "runa-core", "runa-protocol", "runa-server", "rustyline", + "serde_json", ] [[package]] name = "runa-actor" version = "0.0.0" dependencies = [ - "nanoid", "runa-core", ] @@ -586,6 +587,10 @@ dependencies = [ [[package]] name = "runa-core" version = "0.0.0" +dependencies = [ + "nanoid", + "serde", +] [[package]] name = "runa-image-api" @@ -601,6 +606,7 @@ dependencies = [ name = "runa-protocol" version = "0.0.0" dependencies = [ + "runa-core", "serde", "serde_json", ] @@ -609,6 +615,7 @@ dependencies = [ name = "runa-rune" version = "0.0.0" dependencies = [ + "runa-core", "rune", ] diff --git a/README.md b/README.md index 48a3911..0319b2d 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,7 @@ runa> let sender = spawn counter 0 sender = @ runa> let receiver = spawn counter 10 receiver = @ -runa> send sender 1 5 --to receiver +runa> send sender {"op": "send", "value": 5, "target": {"$actor": "receiver"}} message queued runa> tick 100 runa> inspect sender @@ -111,6 +111,15 @@ 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. +State and messages support null, booleans, signed 64-bit integers, strings, +lists, records, and distinct actor-reference values. `spawn ` +sets initial state; `send ` sends any supported value. JSON +objects of the form `{"$actor": "name-or-@reference"}` embed an actor reference, +including inside a record or list. See the [runtime guide](docs/runtime.md) +for Rune conversions and value limits. This is protocol version 3; old +four-argument Rune handlers and kind/value/target message envelopes must be +updated to the two-argument handler contract. + See [the feasibility report](docs/rune-feasibility.md) for verified Rune capabilities and limitations, and [the runtime guide](docs/runtime.md) for the runtime contract. @@ -142,7 +151,7 @@ test task runs `cargo test --workspace`. ├── Cargo.toml ├── crates/ │ ├── runa/ # Main binary -│ ├── runa-core/ # Shared stable identity types +│ ├── runa-core/ # Shared stable identities and portable actor values │ ├── runa-actor/ # Actor value contracts │ ├── runa-code/ # Source identities and compile cache │ ├── runa-rune/ # Phase 0 Rune experiments diff --git a/crates/runa-actor/Cargo.toml b/crates/runa-actor/Cargo.toml index 8bfcb87..5bd419c 100644 --- a/crates/runa-actor/Cargo.toml +++ b/crates/runa-actor/Cargo.toml @@ -4,5 +4,4 @@ version = "0.0.0" edition = "2024" [dependencies] -nanoid = "0.4" runa-core = { path = "../runa-core" } diff --git a/crates/runa-actor/src/lib.rs b/crates/runa-actor/src/lib.rs index e3cc9df..a03f4c9 100644 --- a/crates/runa-actor/src/lib.rs +++ b/crates/runa-actor/src/lib.rs @@ -1,82 +1,12 @@ //! Value contracts for the single-worker actor prototype. use runa_core::ImplementationId; -use std::fmt; - -#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] -pub struct ActorRef([u8; 21]); - -impl ActorRef { - pub fn new() -> Self { - Self::parse(&nanoid::nanoid!()).expect("Nano ID uses the supported alphabet") - } - - pub fn as_str(&self) -> &str { - std::str::from_utf8(&self.0).expect("Nano ID is valid UTF-8") - } - - pub fn parse(value: &str) -> Option { - if value.len() != 21 - || !value - .bytes() - .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) - { - return None; - } - let mut bytes = [0; 21]; - bytes.copy_from_slice(value.as_bytes()); - Some(Self(bytes)) - } -} - -impl Default for ActorRef { - fn default() -> Self { - Self::new() - } -} - -impl fmt::Debug for ActorRef { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(formatter, "@{}", self.as_str()) - } -} - -impl fmt::Display for ActorRef { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter.write_str(self.as_str()) - } -} - -#[cfg(test)] -mod tests { - use super::ActorRef; - - #[test] - fn generated_actor_references_are_parseable_and_display_without_prefix() { - let actor_ref = ActorRef::new(); - let parsed = ActorRef::parse(actor_ref.as_str()).expect("generated reference is valid"); - - assert_eq!(parsed, actor_ref); - assert_eq!(actor_ref.to_string().len(), 21); - assert_eq!(format!("{actor_ref:?}"), format!("@{actor_ref}")); - } - - #[test] - fn parsing_rejects_references_outside_the_url_safe_nanoid_shape() { - assert!(ActorRef::parse("too-short").is_none()); - assert!(ActorRef::parse("abcdefghijklmnopqrstu!").is_none()); - } -} +pub use runa_core::{ActorRef, Value}; #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub struct MailboxId(pub u64); -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct Message { - pub kind: i64, - pub value: i64, - pub target_ref: Option, -} +pub type Message = Value; #[derive(Debug, Clone, PartialEq, Eq)] pub enum Status { @@ -94,7 +24,7 @@ pub struct ActorSnapshot { pub mailbox: MailboxId, pub implementation: ImplementationId, pub incarnation: u64, - pub state: i64, + pub state: Value, pub status: Status, pub execution: Option, pub queued: Vec, @@ -119,5 +49,5 @@ pub enum Event { Resumed(ActorRef), InvocationCancelled(ActorRef), BehaviorInstalled(ActorRef, ImplementationId), - StateMigrated(ActorRef, i64, i64), + StateMigrated(ActorRef, Value, Value), } diff --git a/crates/runa-code/src/lib.rs b/crates/runa-code/src/lib.rs index d9b4b79..ac6c6f8 100644 --- a/crates/runa-code/src/lib.rs +++ b/crates/runa-code/src/lib.rs @@ -25,7 +25,7 @@ impl CompileFingerprint { rune_version: runa_rune::RUNE_VERSION.to_owned(), compiler_settings: "default;features:default".to_owned(), native_modules: runa_rune::COMPILER_CONFIGURATION_ID.to_owned(), - runtime_abi: "runa-runtime-abi:v1".to_owned(), + runtime_abi: "runa-runtime-abi:v3:structured-values".to_owned(), target: env!("RUNA_BUILD_TARGET").to_owned(), } } diff --git a/crates/runa-core/Cargo.toml b/crates/runa-core/Cargo.toml index b283998..e66a696 100644 --- a/crates/runa-core/Cargo.toml +++ b/crates/runa-core/Cargo.toml @@ -4,3 +4,5 @@ version = "0.0.0" edition = "2024" [dependencies] +nanoid = "0.4" +serde = { version = "1", features = ["derive"] } diff --git a/crates/runa-core/src/lib.rs b/crates/runa-core/src/lib.rs index d0a04e8..58fc82a 100644 --- a/crates/runa-core/src/lib.rs +++ b/crates/runa-core/src/lib.rs @@ -2,6 +2,9 @@ use std::fmt; +mod value; +pub use value::{ActorRef, MAX_VALUE_BYTES, MAX_VALUE_DEPTH, MAX_VALUE_NODES, Value}; + macro_rules! digest_id { ($name:ident, $prefix:literal) => { #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] diff --git a/crates/runa-core/src/value.rs b/crates/runa-core/src/value.rs new file mode 100644 index 0000000..8ac5221 --- /dev/null +++ b/crates/runa-core/src/value.rs @@ -0,0 +1,170 @@ +use std::collections::BTreeMap; +use std::fmt; + +use serde::{Deserialize, Serialize}; + +pub const MAX_VALUE_DEPTH: usize = 32; +pub const MAX_VALUE_NODES: usize = 4096; +pub const MAX_VALUE_BYTES: usize = 1024 * 1024; + +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +#[serde(try_from = "String", into = "String")] +pub struct ActorRef([u8; 21]); + +impl ActorRef { + pub fn new() -> Self { + Self::parse(&nanoid::nanoid!()).expect("Nano ID uses the supported alphabet") + } + + pub fn as_str(&self) -> &str { + std::str::from_utf8(&self.0).expect("Nano ID is valid UTF-8") + } + + pub fn parse(value: &str) -> Option { + if value.len() != 21 + || !value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) + { + return None; + } + let mut bytes = [0; 21]; + bytes.copy_from_slice(value.as_bytes()); + Some(Self(bytes)) + } +} + +impl Default for ActorRef { + fn default() -> Self { + Self::new() + } +} + +impl TryFrom for ActorRef { + type Error = String; + + fn try_from(value: String) -> Result { + Self::parse(&value) + .ok_or_else(|| "actor reference must be a 21-character URL-safe NanoID".to_owned()) + } +} + +impl From for String { + fn from(value: ActorRef) -> Self { + value.to_string() + } +} + +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()) + } +} + +/// Owned actor data, never a shared Rune heap object or execution handle. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde( + tag = "type", + content = "data", + rename_all = "snake_case", + deny_unknown_fields +)] +pub enum Value { + Null, + Bool(bool), + Integer(i64), + String(String), + List(Vec), + Record(BTreeMap), + ActorRef(ActorRef), +} + +impl Value { + pub fn validate(&self) -> Result<(), String> { + fn visit( + value: &Value, + depth: usize, + nodes: &mut usize, + bytes: &mut usize, + ) -> Result<(), String> { + *nodes += 1; + if depth > MAX_VALUE_DEPTH || *nodes > MAX_VALUE_NODES { + return Err("actor value exceeds depth or node limit".to_owned()); + } + match value { + Value::String(value) => *bytes += value.len(), + Value::List(values) => { + for value in values { + visit(value, depth + 1, nodes, bytes)?; + } + } + Value::Record(values) => { + for (key, value) in values { + *bytes += key.len(); + visit(value, depth + 1, nodes, bytes)?; + } + } + _ => {} + } + if *bytes > MAX_VALUE_BYTES { + return Err("actor value exceeds string byte limit".to_owned()); + } + Ok(()) + } + visit(self, 0, &mut 0, &mut 0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn generated_references_round_trip_and_are_distinct_values() { + let reference = ActorRef::new(); + assert_eq!(ActorRef::parse(reference.as_str()), Some(reference)); + assert_eq!(format!("{reference:?}"), format!("@{reference}")); + assert_ne!( + Value::ActorRef(reference), + Value::String(reference.to_string()) + ); + assert!(ActorRef::parse("too-short").is_none()); + assert!(ActorRef::parse("abcdefghijklmnopqrst!").is_none()); + } + + #[test] + fn value_limits_are_inclusive() { + let mut value = Value::Null; + for _ in 0..MAX_VALUE_DEPTH { + value = Value::List(vec![value]); + } + assert!(value.validate().is_ok()); + assert!(Value::List(vec![value]).validate().is_err()); + assert!( + Value::List(vec![Value::Null; MAX_VALUE_NODES - 1]) + .validate() + .is_ok() + ); + assert!( + Value::List(vec![Value::Null; MAX_VALUE_NODES]) + .validate() + .is_err() + ); + assert!( + Value::String("x".repeat(MAX_VALUE_BYTES)) + .validate() + .is_ok() + ); + assert!( + Value::String("x".repeat(MAX_VALUE_BYTES + 1)) + .validate() + .is_err() + ); + } +} diff --git a/crates/runa-image-api/src/lib.rs b/crates/runa-image-api/src/lib.rs index 72cee16..4effc6e 100644 --- a/crates/runa-image-api/src/lib.rs +++ b/crates/runa-image-api/src/lib.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use runa_actor::{ActorRef, ActorSnapshot, Event, Message}; use runa_code::{BehaviorSnapshot, CodeStore}; -use runa_core::ImplementationId; +use runa_core::{ImplementationId, Value}; use runa_runtime::{Runtime, RuntimeResult}; pub struct ImageApi { @@ -72,7 +72,7 @@ impl ImageApi { pub fn spawn( &mut self, behavior_name: &str, - state: i64, + state: Value, capacity: NonZeroUsize, ) -> RuntimeResult { let behavior = self @@ -136,7 +136,7 @@ impl ImageApi { &mut self, actor: ActorRef, behavior_name: &str, - migrate: impl FnOnce(i64) -> RuntimeResult, + migrate: impl FnOnce(Value) -> RuntimeResult, ) -> RuntimeResult { let behavior = self .code @@ -152,6 +152,17 @@ impl ImageApi { mod tests { use super::*; + fn test_message(kind: i64, value: i64, target: Option) -> Value { + Value::Record(std::collections::BTreeMap::from([ + ("kind".to_owned(), Value::Integer(kind)), + ("value".to_owned(), Value::Integer(value)), + ( + "target".to_owned(), + target.map_or(Value::Null, Value::ActorRef), + ), + ])) + } + fn capacity(n: usize) -> NonZeroUsize { NonZeroUsize::new(n).expect("capacity must be positive") } @@ -162,24 +173,17 @@ mod tests { let behavior = image.publish_behavior( "counter", None, - "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", + "pub fn handle(state, message) { (state + message.value, []) }", )?; assert_eq!( image.behavior_source("counter", behavior.revision)?, - "pub fn handle(state, kind, value, target_ref) { (state + value, []) }" + "pub fn handle(state, message) { (state + message.value, []) }" ); - let actor = image.spawn("counter", 5, capacity(2))?; - image.send( - actor, - Message { - kind: 0, - value: 7, - target_ref: None, - }, - )?; + let actor = image.spawn("counter", Value::Integer(5), capacity(2))?; + image.send(actor, test_message(0, 7, None))?; image.tick(100, capacity(10)); - assert_eq!(image.inspect(actor)?.state, 12); + assert_eq!(image.inspect(actor)?.state, Value::Integer(12)); assert_eq!(image.inspect(actor)?.implementation, behavior.current); assert!( image @@ -196,14 +200,14 @@ mod tests { let first = image.publish_behavior( "counter", None, - "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", + "pub fn handle(state, message) { (state + message.value, []) }", )?; assert!( image .publish_behavior( "counter", None, - "pub fn handle(state, kind, value, target_ref) { (state - value, []) }" + "pub fn handle(state, message) { (state - message.value, []) }" ) .is_err() ); @@ -217,19 +221,24 @@ mod tests { let first = image.publish_behavior( "counter", None, - "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", + "pub fn handle(state, message) { (state + message.value, []) }", )?; image.publish_behavior( "counter", Some(first.revision), - "pub fn handle(state, kind, value, target_ref) { (state * value, []) }", + "pub fn handle(state, message) { (state * message.value, []) }", )?; - let actor = image.spawn("counter", 5, capacity(2))?; + let actor = image.spawn("counter", Value::Integer(5), capacity(2))?; let implementation = image.install_with_migration(actor, "counter", |state| { - Ok(state.checked_mul(10).ok_or("migration overflow")?) + let Value::Integer(state) = state else { + return Err("expected integer".into()); + }; + Ok(Value::Integer( + state.checked_mul(10).ok_or("migration overflow")?, + )) })?; - assert_eq!(image.inspect(actor)?.state, 50); + assert_eq!(image.inspect(actor)?.state, Value::Integer(50)); assert_eq!(image.inspect(actor)?.implementation, implementation); Ok(()) } diff --git a/crates/runa-protocol/Cargo.toml b/crates/runa-protocol/Cargo.toml index 93b25c4..b23884d 100644 --- a/crates/runa-protocol/Cargo.toml +++ b/crates/runa-protocol/Cargo.toml @@ -4,6 +4,7 @@ version = "0.0.0" edition = "2024" [dependencies] +runa-core = { path = "../runa-core" } serde = { version = "1.0.229", features = ["derive"] } [dev-dependencies] diff --git a/crates/runa-protocol/src/lib.rs b/crates/runa-protocol/src/lib.rs index d648f81..08f858c 100644 --- a/crates/runa-protocol/src/lib.rs +++ b/crates/runa-protocol/src/lib.rs @@ -1,8 +1,9 @@ //! Versioned, transport-neutral wire contracts for Runa clients and servers. +pub use runa_core::Value; use serde::{Deserialize, Serialize}; -pub const PROTOCOL_VERSION: u16 = 2; +pub const PROTOCOL_VERSION: u16 = 3; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -53,18 +54,17 @@ impl Request { Command::Spawn { behavior_name, mailbox_capacity, - .. + state, } => { validate_name(behavior_name)?; + state.validate().map_err(invalid_request)?; if *mailbox_capacity == 0 { return Err(invalid_request("mailbox_capacity must be positive")); } } Command::Send { actor_ref, message } => { validate_actor_ref(actor_ref)?; - if let Some(target_ref) = &message.target_ref { - validate_actor_ref(target_ref)?; - } + message.validate().map_err(invalid_request)?; } Command::InspectActor { actor_ref } | Command::StopActor { actor_ref } @@ -116,7 +116,7 @@ pub enum Command { }, Spawn { behavior_name: String, - state: i64, + state: Value, mailbox_capacity: usize, }, Send { @@ -217,8 +217,8 @@ pub enum EventDto { }, StateMigrated { actor_ref: String, - previous_state: i64, - current_state: i64, + previous_state: Value, + current_state: Value, }, } @@ -238,14 +238,7 @@ pub struct BehaviorVersionDto { pub implementation_id: String, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct MessageDto { - pub kind: i64, - pub value: i64, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub target_ref: Option, -} +pub type MessageDto = Value; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -254,7 +247,7 @@ pub struct ActorSnapshotDto { pub mailbox_id: u64, pub implementation_id: String, pub incarnation: u64, - pub state: i64, + pub state: Value, pub status: ActorStatusDto, pub execution: Option, pub queued: Vec, @@ -349,6 +342,17 @@ fn invalid_request(message: impl Into) -> ProtocolError { mod tests { use super::*; + fn test_message(kind: i64, value: i64, target: Option) -> Value { + Value::Record(std::collections::BTreeMap::from([ + ("kind".to_owned(), Value::Integer(kind)), + ("value".to_owned(), Value::Integer(value)), + ( + "target".to_owned(), + target.map_or(Value::Null, Value::ActorRef), + ), + ])) + } + fn request(command: Command) -> Request { Request { version: PROTOCOL_VERSION, @@ -361,11 +365,7 @@ mod tests { fn request_uses_a_stable_tagged_wire_shape() { let value = serde_json::to_value(request(Command::Send { actor_ref: "xK8_pQ2-Lm5_012345678".to_owned(), - message: MessageDto { - kind: 2, - value: 9, - target_ref: None, - }, + message: Value::Integer(9), })) .unwrap(); assert_eq!( @@ -377,7 +377,7 @@ mod tests { "operation": "send", "parameters": { "actor_ref": "xK8_pQ2-Lm5_012345678", - "message": { "kind": 2, "value": 9 } + "message": { "type": "integer", "data": 9 } } } }) @@ -388,7 +388,7 @@ mod tests { fn requests_validate_versions_and_required_positive_values() { let mut request = request(Command::Spawn { behavior_name: "counter".to_owned(), - state: 0, + state: Value::Integer(0), mailbox_capacity: 1, }); assert_eq!(request.validate(), Ok(())); @@ -402,7 +402,7 @@ mod tests { request.version = PROTOCOL_VERSION; request.command = Command::Spawn { behavior_name: "counter".to_owned(), - state: 0, + state: Value::Integer(0), mailbox_capacity: 0, }; assert_eq!( @@ -431,29 +431,15 @@ mod tests { } #[test] - fn send_target_reference_is_optional_and_validated_when_present() { - let mut request = request(Command::Send { + fn nested_references_round_trip_and_malformed_references_are_rejected() { + let 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()), - }, + message: test_message(2, 9, runa_core::ActorRef::parse("Hj2_9sM0-pQ4_7aBcD3eF")), }); assert_eq!(request.validate(), Ok(())); - - request.command = Command::Send { - actor_ref: "xK8_pQ2-Lm5_012345678".to_owned(), - message: MessageDto { - kind: 2, - value: 9, - target_ref: Some("bad".to_owned()), - }, - }; - assert_eq!( - request.validate().unwrap_err().code, - ErrorCode::InvalidRequest - ); + let encoded = serde_json::to_string(&request).unwrap(); + assert_eq!(serde_json::from_str::(&encoded).unwrap(), request); + assert!(serde_json::from_str::(r#"{"type":"actor_ref","data":"bad"}"#).is_err()); } #[test] @@ -468,14 +454,61 @@ mod tests { assert_eq!(decoded, response); } + #[test] + fn structured_values_have_unambiguous_wire_types_and_reject_invalid_data() { + let reference = runa_core::ActorRef::new(); + let value = Value::Record(std::collections::BTreeMap::from([( + "values".to_owned(), + Value::List(vec![ + Value::Null, + Value::Bool(false), + Value::Integer(i64::MAX), + Value::String(reference.to_string()), + Value::ActorRef(reference), + ]), + )])); + let encoded = serde_json::to_string(&value).unwrap(); + assert_eq!(serde_json::from_str::(&encoded).unwrap(), value); + for encoded in [ + r#"{"type":"integer","data":1.5}"#, + r#"{"type":"integer","data":9223372036854775808}"#, + r#"{"type":"actor_ref","data":"invalid"}"#, + r#"{"type":"unknown","data":1}"#, + r#"{"type":"bool","data":true,"extra":1}"#, + ] { + assert!(serde_json::from_str::(encoded).is_err(), "{encoded}"); + } + let invalid = Value::List(vec![Value::Null; runa_core::MAX_VALUE_NODES]); + let spawn = request(Command::Spawn { + behavior_name: "echo".to_owned(), + state: invalid.clone(), + mailbox_capacity: 2, + }); + assert_eq!( + spawn.validate().unwrap_err().code, + ErrorCode::InvalidRequest + ); + let send = request(Command::Send { + actor_ref: reference.to_string(), + message: invalid, + }); + assert_eq!(send.validate().unwrap_err().code, ErrorCode::InvalidRequest); + let mut old = request(Command::ListActors); + old.version = 2; + assert_eq!( + old.validate().unwrap_err().code, + ErrorCode::UnsupportedVersion + ); + } + #[test] fn event_envelopes_are_versioned_and_serializable() { let event = ProtocolEvent { version: PROTOCOL_VERSION, event: EventDto::StateMigrated { actor_ref: "actor_ref_000000000000".to_owned(), - previous_state: 4, - current_state: 40, + previous_state: Value::Integer(4), + current_state: Value::Integer(40), }, }; let encoded = serde_json::to_string(&event).unwrap(); diff --git a/crates/runa-rune/Cargo.toml b/crates/runa-rune/Cargo.toml index d2e7f6c..62f2185 100644 --- a/crates/runa-rune/Cargo.toml +++ b/crates/runa-rune/Cargo.toml @@ -4,4 +4,5 @@ version = "0.0.0" edition = "2024" [dependencies] +runa-core = { path = "../runa-core" } rune = "=0.14.2" diff --git a/crates/runa-rune/src/lib.rs b/crates/runa-rune/src/lib.rs index 124fdae..9ff85db 100644 --- a/crates/runa-rune/src/lib.rs +++ b/crates/runa-rune/src/lib.rs @@ -14,10 +14,12 @@ use rune::{Context, Diagnostics, Module, Source, Sources, Vm}; pub const RUNE_VERSION: &str = "0.14.2"; pub const COMPILER_CONFIGURATION_ID: &str = - "default-modules; native-functions:v1:native_double,native_suspend"; + "default-modules; native-functions:v2:native_double,native_suspend; runa::ActorReference:v1"; mod execution; pub use execution::{Invocation, InvocationProgress, Slice}; +mod value; +pub use value::{WireTransition, value_from_rune, value_to_rune}; /// Errors remain visible to the caller; compilation also emits source diagnostics. pub type ExperimentResult = Result>; @@ -58,6 +60,24 @@ impl CompiledProgram { pub fn compile_with_module(source: &str, module: Module) -> ExperimentResult { let mut context = Context::with_default_modules()?; + let mut values = Module::with_crate("runa")?; + values.ty::()?; + values.associated_function( + &rune::runtime::Protocol::PARTIAL_EQ, + |left: &value::ActorReference, right: &value::ActorReference| left.0 == right.0, + )?; + values.associated_function( + &rune::runtime::Protocol::EQ, + |left: &value::ActorReference, right: &value::ActorReference| left.0 == right.0, + )?; + values.associated_function( + &rune::runtime::Protocol::CLONE, + |reference: &value::ActorReference| reference.clone(), + )?; + values.associated_function("clone", |reference: &value::ActorReference| { + reference.clone() + })?; + context.install(values)?; context.install(module)?; let mut sources = Sources::new(); diff --git a/crates/runa-rune/src/value.rs b/crates/runa-rune/src/value.rs new file mode 100644 index 0000000..7277d4b --- /dev/null +++ b/crates/runa-rune/src/value.rs @@ -0,0 +1,185 @@ +use runa_core::{ActorRef, MAX_VALUE_BYTES, MAX_VALUE_DEPTH, MAX_VALUE_NODES, Value}; +use rune::runtime::{Object, Vec as RuneVec}; +use rune::{Any, TypeHash, Value as RuneValue}; + +use crate::ExperimentResult; + +pub type WireTransition = (RuneValue, Vec<(RuneValue, RuneValue)>); + +#[derive(Any, Clone)] +#[rune(item = ::runa)] +pub(crate) struct ActorReference(pub ActorRef); + +/// Allocate a fresh Rune object graph for each invocation input. +pub fn value_to_rune(value: &Value) -> ExperimentResult { + value.validate()?; + fn convert(value: &Value) -> ExperimentResult { + Ok(match value { + Value::Null => RuneValue::from(()), + Value::Bool(value) => rune::to_value(*value)?, + Value::Integer(value) => rune::to_value(*value)?, + Value::String(value) => rune::to_value(value.as_str())?, + Value::ActorRef(value) => RuneValue::new(ActorReference(*value))?, + Value::List(values) => { + let mut result = RuneVec::with_capacity(values.len())?; + for value in values { + result.push(convert(value)?)?; + } + RuneValue::new(result)? + } + Value::Record(values) => { + let mut result = Object::with_capacity(values.len())?; + for (key, value) in values { + result.insert( + rune::alloc::String::try_from(key.as_str())?, + convert(value)?, + )?; + } + RuneValue::new(result)? + } + }) + } + convert(value) +} + +/// Copy only portable data out of Rune; cycles and unsupported objects fail. +pub fn value_from_rune(value: &RuneValue) -> ExperimentResult { + fn convert( + value: &RuneValue, + depth: usize, + nodes: &mut usize, + bytes: &mut usize, + ) -> ExperimentResult { + *nodes += 1; + if depth > MAX_VALUE_DEPTH || *nodes > MAX_VALUE_NODES { + return Err("Rune actor value exceeds depth or node limit (possibly cyclic)".into()); + } + + let hash = value.type_hash(); + let result = if value.into_unit().is_ok() { + Value::Null + } else if hash == bool::HASH { + Value::Bool(value.as_bool()?) + } else if hash == i64::HASH { + Value::Integer(value.as_integer()?) + } else if hash == rune::alloc::String::HASH { + let value = value.borrow_string_ref()?; + *bytes += value.len(); + if *bytes > MAX_VALUE_BYTES { + return Err("Rune actor value exceeds string byte limit".into()); + } + Value::String(value.to_string()) + } else if hash == ActorReference::HASH { + Value::ActorRef(value.borrow_ref::()?.0) + } else if hash == RuneVec::HASH { + let values = value.borrow_ref::()?; + let mut result = Vec::new(); + for value in values.iter() { + result.push(convert(value, depth + 1, nodes, bytes)?); + } + Value::List(result) + } else if hash == Object::HASH { + let values = value.borrow_ref::()?; + let mut result = std::collections::BTreeMap::new(); + for (key, value) in values.iter() { + *bytes += key.len(); + if *bytes > MAX_VALUE_BYTES { + return Err("Rune actor value exceeds string byte limit".into()); + } + result.insert(key.to_string(), convert(value, depth + 1, nodes, bytes)?); + } + Value::Record(result) + } else { + return Err(format!("unsupported Rune actor value: {}", value.type_info()).into()); + }; + Ok(result) + } + convert(value, 0, &mut 0, &mut 0) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::CompiledProgram; + use std::collections::BTreeMap; + + #[test] + fn portable_values_round_trip_without_turning_strings_into_references() -> ExperimentResult<()> + { + let actor = ActorRef::new(); + let value = Value::Record(BTreeMap::from([( + "values".to_owned(), + Value::List(vec![ + Value::Null, + Value::Bool(true), + Value::Integer(i64::MIN), + Value::Integer(i64::MAX), + Value::String("hello world".to_owned()), + Value::ActorRef(actor), + Value::String(actor.to_string()), + Value::List(vec![]), + Value::Record(BTreeMap::new()), + ]), + )])); + assert_eq!(value_from_rune(&value_to_rune(&value)?)?, value); + Ok(()) + } + + #[test] + fn rune_mutations_do_not_alias_host_values_or_other_invocation_inputs() -> ExperimentResult<()> + { + let original = Value::Record(BTreeMap::from([( + "items".to_owned(), + Value::List(vec![Value::Integer(1)]), + )])); + let program = + CompiledProgram::compile("pub fn change(state) { state.items.push(2); state }")?; + let first = value_to_rune(&original)?; + let second = value_to_rune(&original)?; + let result = program.vm().call(["change"], (first,))?; + assert_eq!(value_from_rune(&second)?, original); + assert_ne!(value_from_rune(&result)?, original); + Ok(()) + } + + #[test] + fn references_can_be_compared_cloned_and_retained_in_rune_records() -> ExperimentResult<()> { + let actor = ActorRef::new(); + let other = ActorRef::new(); + let program = CompiledProgram::compile( + "pub fn check(left, right) { #{same: left == right, reference: left.clone()} }", + )?; + for (right, expected) in [(actor, true), (other, false)] { + let result = program.vm().call( + ["check"], + ( + value_to_rune(&Value::ActorRef(actor))?, + value_to_rune(&Value::ActorRef(right))?, + ), + )?; + assert_eq!( + value_from_rune(&result)?, + Value::Record(BTreeMap::from([ + ("same".to_owned(), Value::Bool(expected)), + ("reference".to_owned(), Value::ActorRef(actor)), + ])) + ); + } + Ok(()) + } + + #[test] + fn cyclic_and_unsupported_rune_values_are_explicit_errors() -> ExperimentResult<()> { + for source in [ + "pub fn make() { 1.5 }", + "pub fn make() { (1, 2) }", + "pub fn make() { Some(1) }", + "pub fn make() { let x = []; x.push(x); x }", + ] { + let program = CompiledProgram::compile(source)?; + let value = program.vm().call(["make"], ())?; + assert!(value_from_rune(&value).is_err(), "{source}"); + } + Ok(()) + } +} diff --git a/crates/runa-runtime/src/lib.rs b/crates/runa-runtime/src/lib.rs index 6661e55..a31e568 100644 --- a/crates/runa-runtime/src/lib.rs +++ b/crates/runa-runtime/src/lib.rs @@ -8,10 +8,11 @@ use std::task::{Context, Wake, Waker}; use runa_actor::{ActorRef, ActorSnapshot, Event, ExecutionSnapshot, MailboxId, Message, Status}; use runa_code::CompiledArtifact; -use runa_core::ImplementationId; -use runa_rune::{ExperimentResult, Invocation, Slice}; +use runa_core::{ImplementationId, MAX_VALUE_NODES, Value}; +use runa_rune::{ + ExperimentResult, Invocation, Slice, WireTransition, value_from_rune, value_to_rune, +}; -type WireTransition = (i64, Vec<(String, i64, i64)>); pub type RuntimeResult = ExperimentResult; struct WakeFlag(AtomicBool); @@ -61,12 +62,13 @@ impl Runtime { pub fn spawn( &mut self, implementation: ImplementationId, - state: i64, + state: Value, capacity: NonZeroUsize, ) -> ExperimentResult { if !self.implementations.contains_key(&implementation) { return Err("unknown implementation".into()); } + state.validate()?; self.next_mailbox = self .next_mailbox .checked_add(1) @@ -101,6 +103,7 @@ impl Runtime { } pub fn send(&mut self, id: ActorRef, message: Message) -> ExperimentResult<()> { + message.validate()?; let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { return Err("actor is not accepting messages".into()); @@ -125,7 +128,7 @@ impl Runtime { call_depth: progress.call_depth, } }); - snapshot.queued = actor.mailbox.iter().copied().collect(); + snapshot.queued = actor.mailbox.iter().cloned().collect(); Ok(snapshot) } @@ -142,7 +145,7 @@ impl Runtime { call_depth: progress.call_depth, } }); - snapshot.queued = actor.mailbox.iter().copied().collect(); + snapshot.queued = actor.mailbox.iter().cloned().collect(); snapshot }) .collect() @@ -236,7 +239,7 @@ impl Runtime { &mut self, id: ActorRef, implementation: ImplementationId, - migrate: impl FnOnce(i64) -> ExperimentResult, + migrate: impl FnOnce(Value) -> ExperimentResult, ) -> ExperimentResult<()> { if !self.implementations.contains_key(&implementation) { return Err("unknown implementation".into()); @@ -248,12 +251,13 @@ impl Runtime { { return Err("behavior can only be installed at an idle handler boundary".into()); } - let previous_state = current.snapshot.state; - let migrated_state = - migrate(previous_state).map_err(|error| format!("state migration failed: {error}"))?; + let previous_state = current.snapshot.state.clone(); + let migrated_state = migrate(previous_state.clone()) + .map_err(|error| format!("state migration failed: {error}"))?; + migrated_state.validate()?; let actor = self.actors.get_mut(&id).expect("actor checked above"); - actor.snapshot.state = migrated_state; + actor.snapshot.state = migrated_state.clone(); actor.snapshot.implementation = implementation; self.events .push(Event::BehaviorInstalled(id, implementation)); @@ -277,19 +281,31 @@ impl Runtime { fn commit(&mut self, id: &ActorRef, transition: WireTransition) -> ExperimentResult<()> { let (state, effects) = transition; + let state = value_from_rune(&state)?; + if effects.len() > MAX_VALUE_NODES { + return Err("transition exceeds effect limit".into()); + } + let effects = effects + .iter() + .map(|(target, message)| { + let Value::ActorRef(target) = value_from_rune(target)? else { + return Err("effect target must be an actor-reference value".into()); + }; + Ok((target, value_from_rune(message)?)) + }) + .collect::>>()?; let mut counts = BTreeMap::::new(); - for (target, _, _) in &effects { - let target = ActorRef::parse(target).ok_or("effect targets invalid actor reference")?; + for (target, _) in &effects { let actor = self .actors - .get(&target) + .get(target) .ok_or("effect targets unknown actor")?; if matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { return Err("effect targets unavailable actor".into()); } - let count = counts.entry(target).or_default(); + let count = counts.entry(*target).or_default(); *count += 1; - let retained = usize::from(target != *id && actor.snapshot.reserved.is_some()); + let retained = usize::from(target != id && actor.snapshot.reserved.is_some()); if actor.mailbox.len() + retained + *count > actor.capacity.get() { return Err("effect exceeds mailbox capacity".into()); } @@ -299,17 +315,12 @@ impl Runtime { actor.snapshot.reserved = None; actor.snapshot.status = Status::Idle; actor.active = None; - for (target, kind, value) in effects { - let target = ActorRef::parse(&target).expect("validated target"); + for (target, message) in effects { self.actors .get_mut(&target) .expect("validated target") .mailbox - .push_back(Message { - kind, - value, - target_ref: None, - }); + .push_back(message); } self.events.push(Event::Committed(*id)); Ok(()) @@ -330,20 +341,14 @@ impl Runtime { let Some(message) = actor.mailbox.pop_front() else { continue; }; - actor.snapshot.reserved = Some(message); + actor.snapshot.reserved = Some(message.clone()); let program = self.implementations[&actor.snapshot.implementation].program(); - match Invocation::start( - program.vm(), - "handle", - ( - actor.snapshot.state, - message.kind, - message.value, - message - .target_ref - .map_or_else(String::new, |target_ref| target_ref.to_string()), - ), - ) { + let invocation = (|| -> ExperimentResult<_> { + let state = value_to_rune(&actor.snapshot.state)?; + let message = value_to_rune(&message)?; + Ok(Invocation::start(program.vm(), "handle", (state, message))?) + })(); + match invocation { Ok(invocation) => actor.active = Some(invocation), Err(error) => { self.fail(&id, error.to_string()); @@ -380,8 +385,123 @@ impl Runtime { #[cfg(test)] mod tests { use super::*; + + fn test_message(kind: i64, value: i64, target: Option) -> Value { + Value::Record(std::collections::BTreeMap::from([ + ("kind".to_owned(), Value::Integer(kind)), + ("value".to_owned(), Value::Integer(value)), + ( + "target".to_owned(), + target.map_or(Value::Null, Value::ActorRef), + ), + ])) + } + use runa_code::CodeStore; + #[test] + fn nested_state_and_messages_are_private_and_sends_copy_owned_values() -> ExperimentResult<()> { + let mut runtime = Runtime::default(); + let receiver_code = publish( + &mut runtime, + "pub fn handle(state, message) { message.items.push(9); (message, []) }", + )?; + let sender_code = publish( + &mut runtime, + "pub fn handle(state, message) { state.items.push(message.text); (state, [(message.target, state)]) }", + )?; + let initial = Value::Record(BTreeMap::from([( + "items".to_owned(), + Value::List(vec![Value::Bool(true), Value::Null]), + )])); + let receiver = runtime.spawn(receiver_code, Value::Null, capacity(2))?; + let sender = runtime.spawn(sender_code, initial.clone(), capacity(2))?; + let message = Value::Record(BTreeMap::from([ + ("target".to_owned(), Value::ActorRef(receiver)), + ("text".to_owned(), Value::String("hello world".to_owned())), + ])); + runtime.send(sender, message.clone())?; + runtime.tick(capacity(1)); + assert_eq!(runtime.inspect(sender)?.state, initial); + assert_eq!(runtime.inspect(sender)?.reserved, Some(message)); + for _ in 0..100 { + runtime.tick(capacity(32)); + } + let sender_state = Value::Record(BTreeMap::from([( + "items".to_owned(), + Value::List(vec![ + Value::Bool(true), + Value::Null, + Value::String("hello world".to_owned()), + ]), + )])); + let receiver_state = Value::Record(BTreeMap::from([( + "items".to_owned(), + Value::List(vec![ + Value::Bool(true), + Value::Null, + Value::String("hello world".to_owned()), + Value::Integer(9), + ]), + )])); + assert_eq!(runtime.inspect(sender)?.state, sender_state); + assert_eq!(runtime.inspect(receiver)?.state, receiver_state); + Ok(()) + } + + #[test] + fn mutated_drafts_and_invalid_structured_effects_never_partially_commit() -> ExperimentResult<()> + { + for source in [ + "pub fn handle(state, message) { state.items.push(2); panic(\"failed\"); (state, []) }", + "pub fn handle(state, message) { state.items.push(2); (state, [(message, state), (\"not a reference\", state)]) }", + "pub fn handle(state, message) { state.items.push(2); (state, [(message, state), (message, 1.5)]) }", + "pub fn handle(state, message) { state.items.push(2); (Some(state), [(message, state)]) }", + "pub fn handle(state, message) { state.items.push(state); (state, []) }", + ] { + let mut runtime = Runtime::default(); + let code = publish(&mut runtime, source)?; + let initial = Value::Record(BTreeMap::from([( + "items".to_owned(), + Value::List(vec![Value::Integer(1)]), + )])); + let actor = runtime.spawn(code, initial.clone(), capacity(4))?; + runtime.send(actor, Value::ActorRef(actor))?; + for _ in 0..100 { + runtime.tick(capacity(32)); + } + let snapshot = runtime.inspect(actor)?; + assert_eq!(snapshot.state, initial, "{source}"); + assert_eq!(snapshot.reserved, Some(Value::ActorRef(actor)), "{source}"); + assert!(snapshot.queued.is_empty(), "{source}"); + assert!(matches!(snapshot.status, Status::Failed(_)), "{source}"); + } + Ok(()) + } + + #[test] + fn invalid_input_and_migration_values_leave_the_image_unchanged() -> ExperimentResult<()> { + let mut runtime = Runtime::default(); + let code = publish( + &mut runtime, + "pub fn handle(state, message) { (message, []) }", + )?; + let invalid = Value::List(vec![Value::Null; MAX_VALUE_NODES]); + assert!(runtime.spawn(code, invalid.clone(), capacity(2)).is_err()); + assert!(runtime.actors().is_empty()); + let actor = runtime.spawn(code, Value::Bool(true), capacity(2))?; + assert!(runtime.send(actor, invalid.clone()).is_err()); + assert!( + runtime + .install_with_migration(actor, code, |_| Ok(invalid)) + .is_err() + ); + let snapshot = runtime.inspect(actor)?; + assert_eq!(snapshot.state, Value::Bool(true)); + assert!(snapshot.queued.is_empty()); + Ok(()) + } + fn capacity(n: usize) -> NonZeroUsize { NonZeroUsize::new(n).unwrap() } @@ -395,23 +515,16 @@ mod tests { let mut runtime = Runtime::default(); let code = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { if kind == 1 { (state + 1, [(target_ref, 0, 5)]) } else { (state + value, []) } }", - )?; - let first = runtime.spawn(code, 0, capacity(4))?; - let second = runtime.spawn(code, 10, capacity(4))?; - runtime.send( - first, - Message { - kind: 1, - value: 0, - target_ref: Some(second), - }, + "pub fn handle(state, message) { if message.kind == 1 { (state + 1, [(message.target, #{kind: 0, value: 5, target: ()})]) } else { (state + message.value, []) } }", )?; + let first = runtime.spawn(code, Value::Integer(0), capacity(4))?; + let second = runtime.spawn(code, Value::Integer(10), capacity(4))?; + runtime.send(first, test_message(1, 0, Some(second)))?; for _ in 0..100 { runtime.tick(capacity(10)); } - assert_eq!(runtime.inspect(first)?.state, 1); - assert_eq!(runtime.inspect(second)?.state, 15); + assert_eq!(runtime.inspect(first)?.state, Value::Integer(1)); + assert_eq!(runtime.inspect(second)?.state, Value::Integer(15)); assert!(runtime.inspect(first)?.reserved.is_none()); assert!( runtime @@ -425,50 +538,22 @@ mod tests { #[test] fn runaway_does_not_starve_healthy_actor() -> ExperimentResult<()> { let mut runtime = Runtime::default(); - let runaway = publish( - &mut runtime, - "pub fn handle(state, kind, value, target_ref) { loop {} }", - )?; + let runaway = publish(&mut runtime, "pub fn handle(state, message) { loop {} }")?; let healthy = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", - )?; - let first = runtime.spawn(runaway, 0, capacity(1))?; - let second = runtime.spawn(healthy, 0, capacity(2))?; - runtime.send( - first, - Message { - kind: 0, - value: 1, - target_ref: None, - }, - )?; - runtime.send( - second, - Message { - kind: 0, - value: 42, - target_ref: None, - }, + "pub fn handle(state, message) { (state + message.value, []) }", )?; + let first = runtime.spawn(runaway, Value::Integer(0), capacity(1))?; + let second = runtime.spawn(healthy, Value::Integer(0), capacity(2))?; + runtime.send(first, test_message(0, 1, None))?; + runtime.send(second, test_message(0, 42, None))?; for _ in 0..100 { runtime.tick(capacity(5)); } assert_eq!(runtime.inspect(first)?.status, Status::Running); - assert_eq!(runtime.inspect(first)?.state, 0); - assert_eq!(runtime.inspect(second)?.state, 42); - assert!( - runtime - .send( - first, - Message { - kind: 0, - value: 2, - target_ref: None - } - ) - .is_err() - ); + assert_eq!(runtime.inspect(first)?.state, Value::Integer(0)); + assert_eq!(runtime.inspect(second)?.state, Value::Integer(42)); + assert!(runtime.send(first, test_message(0, 2, None)).is_err()); runtime.stop(first)?; assert_eq!(runtime.inspect(first)?.status, Status::Stopped); runtime.restart(first)?; @@ -481,19 +566,9 @@ mod tests { fn execution_snapshot_reports_live_progress_and_is_stable_while_paused() -> ExperimentResult<()> { let mut runtime = Runtime::default(); - let implementation = publish( - &mut runtime, - "pub fn handle(state, kind, value, target_ref) { loop {} }", - )?; - let actor = runtime.spawn(implementation, 0, capacity(2))?; - runtime.send( - actor, - Message { - kind: 0, - value: 1, - target_ref: None, - }, - )?; + let implementation = publish(&mut runtime, "pub fn handle(state, message) { loop {} }")?; + let actor = runtime.spawn(implementation, Value::Integer(0), capacity(2))?; + runtime.send(actor, test_message(0, 1, None))?; runtime.tick(capacity(3)); let running = runtime.inspect(actor)?; @@ -515,14 +590,7 @@ mod tests { runtime.cancel(actor)?; let cancelled = runtime.inspect(actor)?; assert_eq!(cancelled.execution, None); - assert_eq!( - cancelled.queued, - vec![Message { - kind: 0, - value: 1, - target_ref: None - }] - ); + assert_eq!(cancelled.queued, vec![test_message(0, 1, None)]); Ok(()) } @@ -531,22 +599,15 @@ mod tests { let mut runtime = Runtime::default(); let code = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (99, [(target_ref, 0, 1), (\"invalid\", 0, 2)]) }", - )?; - let actor = runtime.spawn(code, 5, capacity(4))?; - runtime.send( - actor, - Message { - kind: 0, - value: 0, - target_ref: Some(actor), - }, + "pub fn handle(state, message) { (99, [(message.target, #{kind: 0, value: 1, target: ()}), (\"invalid\", #{kind: 0, value: 2, target: ()})]) }", )?; + let actor = runtime.spawn(code, Value::Integer(5), capacity(4))?; + runtime.send(actor, test_message(0, 0, Some(actor)))?; for _ in 0..100 { runtime.tick(capacity(10)); } let snapshot = runtime.inspect(actor)?; - assert_eq!(snapshot.state, 5); + assert_eq!(snapshot.state, Value::Integer(5)); assert!(matches!(snapshot.status, Status::Failed(_))); assert!(snapshot.queued.is_empty()); assert!(snapshot.reserved.is_some()); @@ -595,43 +656,22 @@ mod tests { let mut runtime = Runtime::default(); let code = runtime.publish( CodeStore::default().compile_with_module( - "pub async fn handle(state, kind, value, target_ref) { if kind == 1 { let x = wait().await; (state + x, []) } else { (state + value, []) } }", + "pub async fn handle(state, message) { if message.kind == 1 { let x = wait().await; (state + x, []) } else { (state + message.value, []) } }", "test-gate:v1", module, )?, )?; - let actor = runtime.spawn(code, 10, capacity(2))?; - runtime.send( - actor, - Message { - kind: 1, - value: 0, - target_ref: None, - }, - )?; - runtime.send( - actor, - Message { - kind: 0, - value: 3, - target_ref: None, - }, - )?; + let actor = runtime.spawn(code, Value::Integer(10), capacity(2))?; + runtime.send(actor, test_message(1, 0, None))?; + runtime.send(actor, test_message(0, 3, None))?; for _ in 0..100 { runtime.tick(capacity(5)); } let pending = runtime.inspect(actor)?; assert_eq!(pending.status, Status::Awaiting); - assert_eq!(pending.state, 10); + assert_eq!(pending.state, Value::Integer(10)); assert_eq!(pending.queued.len(), 1); - assert_eq!( - pending.reserved, - Some(Message { - kind: 1, - value: 0, - target_ref: None - }) - ); + assert_eq!(pending.reserved, Some(test_message(1, 0, None))); let polls = gate.polls.load(Ordering::SeqCst); for _ in 0..10 { runtime.tick(capacity(5)); @@ -656,21 +696,14 @@ mod tests { runtime.tick(capacity(5)); } let completed = runtime.inspect(actor)?; - assert_eq!(completed.state, 20); + assert_eq!(completed.state, Value::Integer(20)); assert_eq!(completed.status, Status::Idle); assert!(completed.reserved.is_none()); assert!(completed.queued.is_empty()); gate.ready.store(false, Ordering::SeqCst); gate.dropped.store(false, Ordering::SeqCst); - runtime.send( - actor, - Message { - kind: 1, - value: 0, - target_ref: None, - }, - )?; + runtime.send(actor, test_message(1, 0, None))?; for _ in 0..100 { runtime.tick(capacity(5)); } @@ -679,16 +712,9 @@ mod tests { assert!(gate.dropped.load(Ordering::SeqCst)); let cancelled = runtime.inspect(actor)?; assert_eq!(cancelled.mailbox, pending.mailbox); - assert_eq!(cancelled.state, 20); + assert_eq!(cancelled.state, Value::Integer(20)); assert_eq!(cancelled.status, Status::Idle); - assert_eq!( - cancelled.queued, - vec![Message { - kind: 1, - value: 0, - target_ref: None - }] - ); + assert_eq!(cancelled.queued, vec![test_message(1, 0, None)]); assert!(cancelled.reserved.is_none()); Ok(()) } @@ -698,21 +724,14 @@ mod tests { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", - )?; - let actor = runtime.spawn(implementation, 10, capacity(2))?; - runtime.send( - actor, - Message { - kind: 0, - value: 7, - target_ref: None, - }, + "pub fn handle(state, message) { (state + message.value, []) }", )?; + let actor = runtime.spawn(implementation, Value::Integer(10), capacity(2))?; + runtime.send(actor, test_message(0, 7, None))?; runtime.tick(capacity(1)); let before_pause = runtime.inspect(actor)?; assert_eq!(before_pause.status, Status::Running); - assert_eq!(before_pause.state, 10); + assert_eq!(before_pause.state, Value::Integer(10)); assert!(before_pause.reserved.is_some()); runtime.pause(actor)?; @@ -732,7 +751,7 @@ mod tests { } let completed = runtime.inspect(actor)?; assert_eq!(completed.status, Status::Idle); - assert_eq!(completed.state, 17); + assert_eq!(completed.state, Value::Integer(17)); assert!(completed.reserved.is_none()); let events = runtime.drain_events(); assert!( @@ -753,35 +772,17 @@ mod tests { let mut runtime = Runtime::default(); let code = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state / value, []) }", + "pub fn handle(state, message) { (state / message.value, []) }", )?; - let actor = runtime.spawn(code, 5, capacity(1))?; - runtime.send( - actor, - Message { - kind: 0, - value: 0, - target_ref: None, - }, - )?; - assert!( - runtime - .send( - actor, - Message { - kind: 0, - value: 1, - target_ref: None - } - ) - .is_err() - ); + let actor = runtime.spawn(code, Value::Integer(5), capacity(1))?; + runtime.send(actor, test_message(0, 0, None))?; + assert!(runtime.send(actor, test_message(0, 1, None)).is_err()); assert!(runtime.inspect(ActorRef::new()).is_err()); for _ in 0..100 { runtime.tick(capacity(5)); } assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_))); - assert_eq!(runtime.inspect(actor)?.state, 5); + assert_eq!(runtime.inspect(actor)?.state, Value::Integer(5)); assert!( runtime .drain_events() @@ -796,23 +797,16 @@ mod tests { let mut runtime = Runtime::default(); let code = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (99, [(target_ref, 0, 1), (target_ref, 0, 2)]) }", - )?; - let actor = runtime.spawn(code, 5, capacity(1))?; - runtime.send( - actor, - Message { - kind: 0, - value: 0, - target_ref: Some(actor), - }, + "pub fn handle(state, message) { (99, [(message.target, #{kind: 0, value: 1, target: ()}), (message.target, #{kind: 0, value: 2, target: ()})]) }", )?; + let actor = runtime.spawn(code, Value::Integer(5), capacity(1))?; + runtime.send(actor, test_message(0, 0, Some(actor)))?; for _ in 0..100 { runtime.tick(capacity(5)); } let snapshot = runtime.inspect(actor)?; assert!(matches!(snapshot.status, Status::Failed(_))); - assert_eq!(snapshot.state, 5); + assert_eq!(snapshot.state, Value::Integer(5)); assert!(snapshot.queued.is_empty()); assert!(snapshot.reserved.is_some()); Ok(()) @@ -821,25 +815,15 @@ mod tests { #[test] fn invalid_transition_shape_fails_without_committing() -> ExperimentResult<()> { let mut runtime = Runtime::default(); - let code = publish( - &mut runtime, - "pub fn handle(state, kind, value, target_ref) { 99 }", - )?; - let actor = runtime.spawn(code, 5, capacity(1))?; - runtime.send( - actor, - Message { - kind: 0, - value: 1, - target_ref: None, - }, - )?; + let code = publish(&mut runtime, "pub fn handle(state, message) { 99 }")?; + let actor = runtime.spawn(code, Value::Integer(5), capacity(1))?; + runtime.send(actor, test_message(0, 1, None))?; for _ in 0..100 { runtime.tick(capacity(5)); } let snapshot = runtime.inspect(actor)?; assert!(matches!(snapshot.status, Status::Failed(_))); - assert_eq!(snapshot.state, 5); + assert_eq!(snapshot.state, Value::Integer(5)); assert!(snapshot.reserved.is_some()); Ok(()) } @@ -848,24 +832,16 @@ mod tests { fn behavior_installation_waits_for_handler_boundary() -> ExperimentResult<()> { let mut code = CodeStore::default(); let first = - code.compile("pub fn handle(state, kind, value, target_ref) { (state + value, []) }")?; - let second = code.compile( - "pub fn handle(state, kind, value, target_ref) { (state + value * 10, []) }", - )?; + code.compile("pub fn handle(state, message) { (state + message.value, []) }")?; + let second = + code.compile("pub fn handle(state, message) { (state + message.value * 10, []) }")?; let first_id = first.implementation_id; let second_id = second.implementation_id; let mut runtime = Runtime::default(); runtime.publish(first)?; runtime.publish(second)?; - let actor = runtime.spawn(first_id, 0, capacity(2))?; - runtime.send( - actor, - Message { - kind: 0, - value: 1, - target_ref: None, - }, - )?; + let actor = runtime.spawn(first_id, Value::Integer(0), capacity(2))?; + runtime.send(actor, test_message(0, 1, None))?; runtime.tick(capacity(1)); let active = runtime.inspect(actor)?; assert_eq!(active.implementation, first_id); @@ -873,22 +849,15 @@ mod tests { for _ in 0..100 { runtime.tick(capacity(10)); } - assert_eq!(runtime.inspect(actor)?.state, 1); - runtime.send( - actor, - Message { - kind: 0, - value: 2, - target_ref: None, - }, - )?; + assert_eq!(runtime.inspect(actor)?.state, Value::Integer(1)); + runtime.send(actor, test_message(0, 2, None))?; runtime.install(actor, second_id)?; assert_eq!(runtime.inspect(actor)?.implementation, second_id); for _ in 0..100 { runtime.tick(capacity(10)); } let updated = runtime.inspect(actor)?; - assert_eq!(updated.state, 21); + assert_eq!(updated.state, Value::Integer(21)); assert_eq!(updated.implementation, second_id); assert!( runtime @@ -904,26 +873,31 @@ mod tests { let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", + "pub fn handle(state, message) { (state + message.value, []) }", )?; let new_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state * value, []) }", + "pub fn handle(state, message) { (state * message.value, []) }", )?; - let actor = runtime.spawn(old_implementation, 7, capacity(2))?; + let actor = runtime.spawn(old_implementation, Value::Integer(7), capacity(2))?; runtime.install_with_migration(actor, new_implementation, |state| { - Ok(state.checked_mul(10).ok_or("state migration overflow")?) + let Value::Integer(state) = state else { + return Err("expected integer".into()); + }; + Ok(Value::Integer( + state.checked_mul(10).ok_or("state migration overflow")?, + )) })?; let migrated = runtime.inspect(actor)?; assert_eq!(migrated.implementation, new_implementation); - assert_eq!(migrated.state, 70); + assert_eq!(migrated.state, Value::Integer(70)); assert!( runtime .drain_events() .iter() - .any(|event| matches!(event, Event::StateMigrated(id, 7, 70) if *id == actor)) + .any(|event| matches!(event, Event::StateMigrated(id, Value::Integer(7), Value::Integer(70)) if *id == actor)) ); Ok(()) } @@ -933,67 +907,35 @@ mod tests { let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", + "pub fn handle(state, message) { (state + message.value, []) }", )?; let new_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state * 10 + value, []) }", - )?; - let actor = runtime.spawn(old_implementation, 0, capacity(4))?; - runtime.send( - actor, - Message { - kind: 0, - value: 1, - target_ref: None, - }, - )?; - runtime.send( - actor, - Message { - kind: 0, - value: 2, - target_ref: None, - }, + "pub fn handle(state, message) { (state * 10 + message.value, []) }", )?; + let actor = runtime.spawn(old_implementation, Value::Integer(0), capacity(4))?; + runtime.send(actor, test_message(0, 1, None))?; + runtime.send(actor, test_message(0, 2, None))?; runtime.tick(capacity(1)); assert_eq!(runtime.inspect(actor)?.status, Status::Running); runtime.cancel(actor)?; let cancelled = runtime.inspect(actor)?; assert_eq!(cancelled.status, Status::Idle); - assert_eq!(cancelled.state, 0); + assert_eq!(cancelled.state, Value::Integer(0)); assert_eq!( cancelled.queued, - vec![ - Message { - kind: 0, - value: 1, - target_ref: None - }, - Message { - kind: 0, - value: 2, - target_ref: None - } - ] + vec![test_message(0, 1, None), test_message(0, 2, None)] ); assert!(cancelled.reserved.is_none()); runtime.install(actor, new_implementation)?; - runtime.send( - actor, - Message { - kind: 0, - value: 3, - target_ref: None, - }, - )?; + runtime.send(actor, test_message(0, 3, None))?; for _ in 0..100 { runtime.tick(capacity(10)); } let completed = runtime.inspect(actor)?; - assert_eq!(completed.state, 123); + assert_eq!(completed.state, Value::Integer(123)); assert_eq!(completed.implementation, new_implementation); assert!( runtime @@ -1009,13 +951,13 @@ mod tests { let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", + "pub fn handle(state, message) { (state + message.value, []) }", )?; let new_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state * value, []) }", + "pub fn handle(state, message) { (state * message.value, []) }", )?; - let actor = runtime.spawn(old_implementation, 7, capacity(2))?; + let actor = runtime.spawn(old_implementation, Value::Integer(7), capacity(2))?; assert!( runtime @@ -1027,7 +969,7 @@ mod tests { let unchanged = runtime.inspect(actor)?; assert_eq!(unchanged.implementation, old_implementation); - assert_eq!(unchanged.state, 7); + assert_eq!(unchanged.state, Value::Integer(7)); assert!( !runtime .drain_events() @@ -1044,21 +986,14 @@ mod tests { let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state + value, []) }", + "pub fn handle(state, message) { (state + message.value, []) }", )?; let new_implementation = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state * value, []) }", - )?; - let actor = runtime.spawn(old_implementation, 7, capacity(2))?; - runtime.send( - actor, - Message { - kind: 0, - value: 1, - target_ref: None, - }, + "pub fn handle(state, message) { (state * message.value, []) }", )?; + let actor = runtime.spawn(old_implementation, Value::Integer(7), capacity(2))?; + runtime.send(actor, test_message(0, 1, None))?; runtime.tick(capacity(1)); let called = Cell::new(false); @@ -1066,7 +1001,10 @@ mod tests { runtime .install_with_migration(actor, new_implementation, |state| { called.set(true); - Ok(state + 100) + let Value::Integer(state) = state else { + return Err("expected integer".into()); + }; + Ok(Value::Integer(state + 100)) }) .is_err() ); @@ -1090,45 +1028,26 @@ mod tests { let mut runtime = Runtime::default(); let code = runtime.publish( CodeStore::default().compile_with_module( - "fn check() { if fail_once() { 1 / 0 } else { 0 } } pub fn handle(state, kind, value, target_ref) { let _ = check(); (state + value, []) }", + "fn check() { if fail_once() { 1 / 0 } else { 0 } } pub fn handle(state, message) { let _ = check(); (state + message.value, []) }", "test-fail-once:v1", module, )?, )?; - let actor = runtime.spawn(code, 10, capacity(3))?; - let first = Message { - kind: 1, - value: 2, - target_ref: None, - }; - let second = Message { - kind: 1, - value: 3, - target_ref: None, - }; - runtime.send(actor, first)?; - runtime.send(actor, second)?; + let actor = runtime.spawn(code, Value::Integer(10), capacity(3))?; + let first = test_message(1, 2, None); + let second = test_message(1, 3, None); + runtime.send(actor, first.clone())?; + runtime.send(actor, second.clone())?; for _ in 0..100 { runtime.tick(capacity(5)); } let failed = runtime.inspect(actor)?; assert!(matches!(failed.status, Status::Failed(_))); - assert_eq!(failed.state, 10); - assert_eq!(failed.reserved, Some(first)); - assert_eq!(failed.queued, vec![second]); - assert!( - runtime - .send( - actor, - Message { - kind: 1, - value: 4, - target_ref: None - } - ) - .is_err() - ); + assert_eq!(failed.state, Value::Integer(10)); + assert_eq!(failed.reserved, Some(first.clone())); + assert_eq!(failed.queued, vec![second.clone()]); + assert!(runtime.send(actor, test_message(1, 4, None)).is_err()); assert!( runtime .drain_events() @@ -1142,15 +1061,15 @@ mod tests { assert_eq!(restarted.mailbox, failed.mailbox); assert_eq!(restarted.implementation, failed.implementation); assert_eq!(restarted.incarnation, failed.incarnation + 1); - assert_eq!(restarted.state, 10); - assert_eq!(restarted.queued, vec![first, second]); + assert_eq!(restarted.state, Value::Integer(10)); + assert_eq!(restarted.queued, vec![first.clone(), second.clone()]); assert_eq!(restarted.reserved, None); for _ in 0..100 { runtime.tick(capacity(5)); } let completed = runtime.inspect(actor)?; - assert_eq!(completed.state, 15, "{completed:?}"); + assert_eq!(completed.state, Value::Integer(15), "{completed:?}"); assert_eq!(completed.status, Status::Idle); assert_eq!(completed.queued, Vec::::new()); assert_eq!(completed.reserved, None); @@ -1162,64 +1081,44 @@ mod tests { let mut runtime = Runtime::default(); let code = publish( &mut runtime, - "pub fn handle(state, kind, value, target_ref) { (state * 10 + value, []) }", + "pub fn handle(state, message) { (state * 10 + message.value, []) }", )?; - let actor = runtime.spawn(code, 0, capacity(3))?; - let first = Message { - kind: 0, - value: 1, - target_ref: None, - }; - let second = Message { - kind: 0, - value: 2, - target_ref: None, - }; - let third = Message { - kind: 0, - value: 3, - target_ref: None, - }; - runtime.send(actor, first)?; - runtime.send(actor, second)?; - runtime.send(actor, third)?; + let actor = runtime.spawn(code, Value::Integer(0), capacity(3))?; + let first = test_message(0, 1, None); + let second = test_message(0, 2, None); + let third = test_message(0, 3, None); + runtime.send(actor, first.clone())?; + runtime.send(actor, second.clone())?; + runtime.send(actor, third.clone())?; runtime.tick(capacity(1)); let running = runtime.inspect(actor)?; assert_eq!(running.status, Status::Running); - assert_eq!(running.state, 0); - assert_eq!(running.reserved, Some(first)); - assert_eq!(running.queued, vec![second, third]); + assert_eq!(running.state, Value::Integer(0)); + assert_eq!(running.reserved, Some(first.clone())); + assert_eq!(running.queued, vec![second.clone(), third.clone()]); runtime.stop(actor)?; let stopped = runtime.inspect(actor)?; assert_eq!(stopped.status, Status::Stopped); - assert_eq!(stopped.state, 0); - assert_eq!(stopped.reserved, Some(first)); - assert_eq!(stopped.queued, vec![second, third]); - assert!( - runtime - .send( - actor, - Message { - kind: 0, - value: 4, - target_ref: None - } - ) - .is_err() - ); + assert_eq!(stopped.state, Value::Integer(0)); + assert_eq!(stopped.reserved, Some(first.clone())); + assert_eq!(stopped.queued, vec![second.clone(), third.clone()]); + assert!(runtime.send(actor, test_message(0, 4, None)).is_err()); runtime.restart(actor)?; let restarted = runtime.inspect(actor)?; - assert_eq!(restarted.state, 0); + assert_eq!(restarted.state, Value::Integer(0)); assert_eq!(restarted.reserved, None); - assert_eq!(restarted.queued, vec![first, second, third]); + assert_eq!( + restarted.queued, + vec![first.clone(), second.clone(), third.clone()] + ); for _ in 0..100 { runtime.tick(capacity(5)); } let completed = runtime.inspect(actor)?; - assert_eq!(completed.state, 123); + assert_eq!(completed.state, Value::Integer(123)); assert_eq!(completed.status, Status::Idle); assert_eq!(completed.queued, Vec::::new()); assert!(completed.reserved.is_none()); diff --git a/crates/runa-server/src/lib.rs b/crates/runa-server/src/lib.rs index 4ec3e5f..fdd709b 100644 --- a/crates/runa-server/src/lib.rs +++ b/crates/runa-server/src/lib.rs @@ -6,12 +6,11 @@ use std::num::NonZeroUsize; use std::thread; use std::time::Duration; -use runa_actor::{ActorRef, Event, Message, Status}; +use runa_actor::{ActorRef, Event, Status}; use runa_image_api::ImageApi; use runa_protocol::{ ActorSnapshotDto, ActorStatusDto, BehaviorDto, BehaviorVersionDto, Command, ErrorCode, - EventDto, ExecutionSnapshotDto, MessageDto, PROTOCOL_VERSION, ProtocolError, Reply, Request, - Response, + EventDto, ExecutionSnapshotDto, PROTOCOL_VERSION, ProtocolError, Reply, Request, Response, }; pub const MAX_REQUEST_LINE_BYTES: usize = 1024 * 1024; @@ -347,7 +346,7 @@ impl Server { Command::Send { actor_ref, message } => { let actor_ref = parse_actor_ref(&actor_ref)?; self.image - .send(actor_ref, message_from_dto(message)?) + .send(actor_ref, message) .map(|()| Reply::Acknowledged) .map_err(api_error) } @@ -447,7 +446,7 @@ fn actor_snapshot_dto(snapshot: &runa_actor::ActorSnapshot) -> ActorSnapshotDto mailbox_id: snapshot.mailbox.0, implementation_id: snapshot.implementation.to_string(), incarnation: snapshot.incarnation, - state: snapshot.state, + state: snapshot.state.clone(), status: match &snapshot.status { Status::Idle => ActorStatusDto::Idle, Status::Running => ActorStatusDto::Running, @@ -461,33 +460,11 @@ fn actor_snapshot_dto(snapshot: &runa_actor::ActorSnapshot) -> ActorSnapshotDto instruction_pointer: execution.instruction_pointer, call_depth: execution.call_depth, }), - queued: snapshot.queued.iter().copied().map(message_dto).collect(), - reserved: snapshot.reserved.map(message_dto), + queued: snapshot.queued.clone(), + reserved: snapshot.reserved.clone(), } } -fn message_dto(message: Message) -> MessageDto { - MessageDto { - kind: message.kind, - value: message.value, - target_ref: message - .target_ref - .map(|actor_ref| actor_ref.as_str().to_owned()), - } -} - -fn message_from_dto(message: MessageDto) -> Result { - Ok(Message { - kind: message.kind, - value: message.value, - target_ref: message - .target_ref - .as_deref() - .map(parse_actor_ref) - .transpose()?, - }) -} - fn event_dto(event: &Event) -> EventDto { match event { Event::Spawned(actor_ref) => EventDto::Spawned { @@ -521,8 +498,8 @@ fn event_dto(event: &Event) -> EventDto { }, Event::StateMigrated(actor_ref, previous_state, current_state) => EventDto::StateMigrated { actor_ref: actor_ref.as_str().to_owned(), - previous_state: *previous_state, - current_state: *current_state, + previous_state: previous_state.clone(), + current_state: current_state.clone(), }, } } @@ -564,6 +541,19 @@ fn parse_actor_ref(value: &str) -> Result { #[cfg(test)] mod tests { use super::*; + use runa_protocol::Value; + + fn test_message(kind: i64, value: i64, target: Option) -> Value { + Value::Record(std::collections::BTreeMap::from([ + ("kind".to_owned(), Value::Integer(kind)), + ("value".to_owned(), Value::Integer(value)), + ( + "target".to_owned(), + target.map_or(Value::Null, Value::ActorRef), + ), + ])) + } + use runa_protocol::{Command, PROTOCOL_VERSION}; use std::io::{BufRead, BufReader, Cursor}; use std::net::Shutdown; @@ -580,8 +570,7 @@ mod tests { let response = server.dispatch(request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: None, - source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }" - .to_owned(), + source: "pub fn handle(state, message) { (state + message.value, []) }".to_owned(), })); match response.result.unwrap() { Reply::Behavior(behavior) => behavior, @@ -598,7 +587,7 @@ mod tests { let spawned = server.dispatch(request(Command::Spawn { behavior_name: "counter".to_owned(), - state: 5, + state: Value::Integer(5), mailbox_capacity: 2, })); let actor_ref = match spawned.result.unwrap() { @@ -610,11 +599,7 @@ mod tests { let sent = server.dispatch(request(Command::Send { actor_ref: actor_ref.clone(), - message: MessageDto { - kind: 0, - value: 7, - target_ref: None, - }, + message: test_message(0, 7, None), })); assert_eq!(sent.result, Ok(Reply::Acknowledged)); server.dispatch(request(Command::Tick { @@ -626,7 +611,7 @@ mod tests { actor_ref: actor_ref.clone(), })); match inspected.result.unwrap() { - Reply::Actor(snapshot) => assert_eq!(snapshot.state, 12), + Reply::Actor(snapshot) => assert_eq!(snapshot.state, Value::Integer(12)), other => panic!("unexpected reply: {other:?}"), } @@ -664,7 +649,7 @@ mod tests { let conflict = server.dispatch(request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: Some(99), - source: "pub fn handle(state, kind, value, target_ref) { (state, []) }".to_owned(), + source: "pub fn handle(state, message) { (state, []) }".to_owned(), })); assert_eq!(conflict.result.unwrap_err().code, ErrorCode::Conflict); } @@ -676,8 +661,7 @@ mod tests { let updated = server.dispatch(request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: Some(first.revision), - source: "pub fn handle(state, kind, value, target_ref) { (state * value, []) }" - .to_owned(), + source: "pub fn handle(state, message) { (state * message.value, []) }".to_owned(), })); assert!(matches!(updated.result, Ok(Reply::Behavior(_)))); @@ -688,15 +672,14 @@ mod tests { assert_eq!( source.result, Ok(Reply::BehaviorSource { - source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }" - .to_owned() + source: "pub fn handle(state, message) { (state + message.value, []) }".to_owned() }) ); let actor_ref = match server .dispatch(request(Command::Spawn { behavior_name: "counter".to_owned(), - state: 3, + state: Value::Integer(3), mailbox_capacity: 1, })) .result @@ -720,12 +703,11 @@ mod tests { let publish_request = request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: None, - source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }" - .to_owned(), + source: "pub fn handle(state, message) { (state + message.value, []) }".to_owned(), }); let mut spawn_request = request(Command::Spawn { behavior_name: "counter".to_owned(), - state: 4, + state: Value::Integer(4), mailbox_capacity: 2, }); spawn_request.request_id = "spawn-request".to_owned(); @@ -764,7 +746,7 @@ mod tests { )); assert!(matches!( &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 == Value::Integer(4) )); } diff --git a/crates/runa-server/tests/local_client.rs b/crates/runa-server/tests/local_client.rs index a6cc489..f5558da 100644 --- a/crates/runa-server/tests/local_client.rs +++ b/crates/runa-server/tests/local_client.rs @@ -1,10 +1,11 @@ +use runa_protocol::Value; use std::io::{BufRead, BufReader, Write}; use std::net::{Shutdown, SocketAddr, TcpStream}; use std::process::{Child, Command as ProcessCommand, Stdio}; use std::time::Duration; use runa_client::{LocalClient, TcpClient}; -use runa_protocol::{Command, MessageDto, PROTOCOL_VERSION, Reply, Request, Response}; +use runa_protocol::{Command, PROTOCOL_VERSION, Reply, Request, Response}; struct ServerProcess(Child); @@ -23,8 +24,7 @@ fn local_client_launches_server_and_operates_the_same_live_image() { .request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: None, - source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }" - .to_owned(), + source: "pub fn handle(state, message) { (state + message.value, []) }".to_owned(), }) .unwrap(); assert_eq!(published.request_id, "runa-client-1"); @@ -33,7 +33,7 @@ fn local_client_launches_server_and_operates_the_same_live_image() { let spawned = client .request(Command::Spawn { behavior_name: "counter".to_owned(), - state: 5, + state: Value::Integer(5), mailbox_capacity: 2, }) .unwrap(); @@ -47,11 +47,7 @@ fn local_client_launches_server_and_operates_the_same_live_image() { client .request(Command::Send { actor_ref: actor_ref.clone(), - message: MessageDto { - kind: 0, - value: 7, - target_ref: None, - }, + message: test_message(0, 7, None), }) .unwrap(); client @@ -63,7 +59,7 @@ fn local_client_launches_server_and_operates_the_same_live_image() { let inspected = client.request(Command::InspectActor { actor_ref }).unwrap(); assert!(matches!( inspected.result, - Ok(Reply::Actor(actor)) if actor.state == 12 + Ok(Reply::Actor(actor)) if actor.state == Value::Integer(12) )); assert!(client.shutdown().unwrap().success()); } @@ -110,7 +106,7 @@ fn tcp_server_accepts_local_protocol_requests() { let actor_ref = match first .request(Command::Spawn { behavior_name: "counter".to_owned(), - state: 5, + state: Value::Integer(5), mailbox_capacity: 2, }) .unwrap() @@ -122,7 +118,7 @@ fn tcp_server_accepts_local_protocol_requests() { }; assert!(matches!(second.request(Command::InspectActor { actor_ref: actor_ref.clone(), - }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == 5)); + }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == Value::Integer(5))); assert!(matches!( first .request(Command::RollbackBehavior { @@ -150,11 +146,10 @@ fn tcp_server_accepts_local_protocol_requests() { reconnected .request(Command::Send { actor_ref: actor_ref.clone(), - message: MessageDto { - kind: 0, - value: 7, - target_ref: None, - }, + message: Value::Record(std::collections::BTreeMap::from([ + ("op".to_owned(), Value::String("add".to_owned())), + ("value".to_owned(), Value::Integer(7)), + ])), }) .unwrap() .result @@ -168,6 +163,72 @@ fn tcp_server_accepts_local_protocol_requests() { .result .unwrap(); assert!(matches!(reconnected.request(Command::InspectActor { - actor_ref, - }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == 12)); + actor_ref: actor_ref.clone(), + }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == Value::Integer(12))); + + reconnected + .request(Command::PublishBehavior { + name: "echo".to_owned(), + expected_revision: None, + source: "pub fn handle(state, message) { (message, []) }".to_owned(), + }) + .unwrap() + .result + .unwrap(); + let nested = Value::Record(std::collections::BTreeMap::from([( + "items".to_owned(), + Value::List(vec![ + Value::Null, + Value::Bool(true), + Value::Integer(-42), + Value::String("hello world".to_owned()), + Value::ActorRef(runa_actor::ActorRef::parse(&actor_ref).unwrap()), + ]), + )])); + let echo = match reconnected + .request(Command::Spawn { + behavior_name: "echo".to_owned(), + state: nested.clone(), + mailbox_capacity: 2, + }) + .unwrap() + .result + .unwrap() + { + Reply::ActorSpawned { actor_ref } => actor_ref, + other => panic!("unexpected spawn: {other:?}"), + }; + assert!(matches!(reconnected.request(Command::InspectActor { + actor_ref: echo.clone(), + }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == nested)); + reconnected + .request(Command::Send { + actor_ref: echo.clone(), + message: nested.clone(), + }) + .unwrap() + .result + .unwrap(); + reconnected + .request(Command::Tick { + ticks: 100, + instruction_budget: 10, + }) + .unwrap() + .result + .unwrap(); + assert!(matches!(reconnected.request(Command::InspectActor { + actor_ref: echo, + }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == nested && actor.reserved.is_none())); +} + +fn test_message(kind: i64, value: i64, target: Option) -> Value { + Value::Record(std::collections::BTreeMap::from([ + ("kind".to_owned(), Value::Integer(kind)), + ("value".to_owned(), Value::Integer(value)), + ( + "target".to_owned(), + target.map_or(Value::Null, Value::ActorRef), + ), + ])) } diff --git a/crates/runa/Cargo.toml b/crates/runa/Cargo.toml index 64eaef6..df16c00 100644 --- a/crates/runa/Cargo.toml +++ b/crates/runa/Cargo.toml @@ -5,8 +5,10 @@ edition = "2024" [dependencies] runa-client = { path = "../runa-client" } +runa-core = { path = "../runa-core" } runa-protocol = { path = "../runa-protocol" } rustyline = "15.0" +serde_json = "1" [dev-dependencies] runa-server = { path = "../runa-server" } diff --git a/crates/runa/src/main.rs b/crates/runa/src/main.rs index 69d1707..1e9b377 100644 --- a/crates/runa/src/main.rs +++ b/crates/runa/src/main.rs @@ -6,7 +6,8 @@ use std::net::SocketAddr; use std::num::NonZeroUsize; use runa_client::TcpClient; -use runa_protocol::{Command, MessageDto, Reply}; +use runa_core::ActorRef; +use runa_protocol::{Command, Reply, Value}; #[cfg(test)] use runa_server::Server; use rustyline::completion::{Completer, Pair}; @@ -94,9 +95,14 @@ impl ReplHelper { "stop", "tick", ]; - let start = line[..position] - .rfind(char::is_whitespace) - .map_or(0, |index| index + 1); + let (start, reference_value) = Self::completion_start(line, position); + if reference_value { + return self + .actor_candidates() + .into_iter() + .filter(|option| option.starts_with(&line[start..position])) + .collect(); + } let previous: Vec<_> = line[..start].split_whitespace().collect(); let options: Vec = match previous.as_slice() { [] => COMMANDS @@ -114,9 +120,7 @@ impl ReplHelper { | ["resume"] | ["cancel"] | ["install"] => self.actor_candidates(), - ["send", _, _, _] => vec!["--to".to_owned()], - ["send", _, _, _, "--to"] => self.actor_candidates(), - ["install", _] => self.behaviors.clone(), + ["install", _] | ["history"] | ["rollback"] | ["publish"] => self.behaviors.clone(), _ => Vec::new(), }; let prefix = &line[start..position]; @@ -126,6 +130,28 @@ impl ReplHelper { .collect() } + fn completion_start(line: &str, position: usize) -> (usize, bool) { + let before_cursor = &line[..position]; + if let Some(marker) = before_cursor.rfind("\"$actor\"") { + let suffix = &before_cursor[marker + "\"$actor\"".len()..]; + if let Some(value) = suffix.trim_start().strip_prefix(':') { + let value = value.trim_start(); + if let Some(content) = value.strip_prefix('"') + && !content.contains('"') + { + return (position - content.len(), true); + } + } + } + ( + before_cursor + .char_indices() + .rfind(|(_, character)| character.is_whitespace()) + .map_or(0, |(index, character)| index + character.len_utf8()), + false, + ) + } + fn actor_candidates(&self) -> Vec { self.bindings .iter() @@ -143,9 +169,7 @@ impl Completer for ReplHelper { position: usize, _context: &Context<'_>, ) -> rustyline::Result<(usize, Vec)> { - let start = line[..position] - .rfind(char::is_whitespace) - .map_or(0, |index| index + 1); + let (start, _) = ReplHelper::completion_start(line, position); let candidates = self .candidates(line, position) .into_iter() @@ -260,7 +284,9 @@ fn execute_line( if words.is_empty() { return Ok(true); } - if let ["let", name, "=", "spawn", behavior_name, initial_state] = words.as_slice() { + if let ["let", name, "=", "spawn", behavior_name, initial_state @ ..] = words.as_slice() + && !initial_state.is_empty() + { if !is_binding_name(name) { writeln!(output, "error: invalid actor binding name `{name}`")?; return Ok(true); @@ -269,7 +295,7 @@ fn execute_line( writeln!(output, "error: actor binding `{name}` already exists")?; return Ok(true); } - match spawn_actor(behavior_name, initial_state, server) { + match spawn_actor(behavior_name, after_words(line, 5), server, state) { Ok(actor_ref) => { writeln!(output, "{name} = @{actor_ref}")?; state.bindings.insert((*name).to_owned(), actor_ref); @@ -282,7 +308,7 @@ fn execute_line( let args = &words[1..]; let result = match command { "help" => print_help(output), - "spawn" => spawn_command(output, args, server), + "spawn" => spawn_command(output, line, server, state), "behaviors" if args.is_empty() => match dispatch(server, Command::GetBehaviors) { Ok(Reply::Behaviors(behaviors)) => { for behavior in behaviors { @@ -313,19 +339,9 @@ fn execute_line( 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, - }, - }, - ) { + "send" => match parse_send_args(line, state) { + Ok((actor_ref, message)) => { + match dispatch(server, Command::Send { actor_ref, message }) { Ok(Reply::Acknowledged) => writeln!(output, "message queued"), Ok(other) => unexpected_reply(output, other), Err(error) => writeln!(output, "error: {error}"), @@ -412,16 +428,24 @@ fn lifecycle_command( fn spawn_command( output: &mut W, - args: &[&str], + line: &str, server: &mut dyn Connection, + state: &ReplState, ) -> io::Result<()> { - let [behavior_name, initial_state] = args else { + let args: Vec<_> = line.split_whitespace().collect(); + let [_, behavior_name, initial_state @ ..] = args.as_slice() else { return writeln!( output, "error: usage: spawn " ); }; - match spawn_actor(behavior_name, initial_state, server) { + if initial_state.is_empty() { + return writeln!( + output, + "error: usage: spawn " + ); + } + match spawn_actor(behavior_name, after_words(line, 2), server, state) { Ok(actor_ref) => writeln!(output, "spawned @{actor_ref} from `{behavior_name}`"), Err(error) => writeln!(output, "error: {error}"), } @@ -431,10 +455,10 @@ fn spawn_actor( behavior_name: &str, initial_state: &str, server: &mut dyn Connection, + repl: &ReplState, ) -> Result { - let state = initial_state - .parse() - .map_err(|_| format!("invalid initial state: `{initial_state}`"))?; + let state = parse_value(initial_state, repl) + .map_err(|error| format!("invalid initial state: {error}"))?; match dispatch( server, Command::Spawn { @@ -663,26 +687,86 @@ fn parse_tick_args(args: &[&str]) -> Result<(usize, NonZeroUsize), String> { } } -fn parse_send_args( - args: &[&str], - state: &ReplState, -) -> Result<(String, i64, i64, Option), String> { - let (actor_arg, kind_arg, value_arg, target_arg) = match args { - [actor, kind, value] => (*actor, *kind, *value, None), - [actor, kind, value, "--to", target] => (*actor, *kind, *value, Some(*target)), - _ => return Err("usage: send [--to ]".to_owned()), - }; +fn parse_send_args(line: &str, state: &ReplState) -> Result<(String, Value), String> { + let actor_arg = line + .split_whitespace() + .nth(1) + .ok_or("usage: send ")?; let actor_ref = parse_actor_argument(&[actor_arg], state)?; - let kind = kind_arg - .parse() - .map_err(|_| format!("invalid message kind: `{kind_arg}`"))?; - let value = value_arg - .parse() - .map_err(|_| format!("invalid message value: `{value_arg}`"))?; - let target_ref = target_arg - .map(|target| parse_actor_argument(&[target], state)) - .transpose()?; - Ok((actor_ref, kind, value, target_ref)) + Ok((actor_ref, parse_value(after_words(line, 2), state)?)) +} + +fn after_words(mut line: &str, count: usize) -> &str { + for _ in 0..count { + line = line.trim_start(); + line = line + .find(char::is_whitespace) + .map_or("", |index| &line[index..]); + } + line.trim() +} + +fn parse_value(input: &str, state: &ReplState) -> Result { + if input.len() > runa_core::MAX_VALUE_BYTES { + return Err("JSON value exceeds input byte limit".to_owned()); + } + fn record( + values: serde_json::Map, + state: &ReplState, + ) -> Result { + values + .into_iter() + .map(|(key, value)| Ok((key, convert(value, state)?))) + .collect::, String>>() + .map(Value::Record) + } + fn convert(value: serde_json::Value, state: &ReplState) -> Result { + Ok(match value { + serde_json::Value::Null => Value::Null, + serde_json::Value::Bool(value) => Value::Bool(value), + serde_json::Value::Number(value) => Value::Integer( + value + .as_i64() + .ok_or("numbers must be signed 64-bit integers")?, + ), + serde_json::Value::String(value) => Value::String(value), + serde_json::Value::Array(values) => Value::List( + values + .into_iter() + .map(|value| convert(value, state)) + .collect::>()?, + ), + serde_json::Value::Object(mut values) + if values.len() == 1 && values.contains_key("$actor") => + { + let serde_json::Value::String(reference) = + values.remove("$actor").expect("key checked") + else { + return Err( + "$actor must contain a binding name or actor reference string".to_owned(), + ); + }; + let reference = parse_actor_argument(&[&reference], state)?; + Value::ActorRef(ActorRef::parse(&reference).expect("reference validated")) + } + serde_json::Value::Object(mut values) + if values.len() == 1 && values.contains_key("$record") => + { + let serde_json::Value::Object(values) = + values.remove("$record").expect("key checked") + else { + return Err("$record must contain a JSON object".to_owned()); + }; + record(values, state)? + } + serde_json::Value::Object(values) => record(values, state)?, + }) + } + let value = + serde_json::from_str(input).map_err(|error| format!("invalid JSON value: {error}"))?; + let value = convert(value, state)?; + value.validate()?; + Ok(value) } fn print_help(output: &mut W) -> io::Result<()> { @@ -696,7 +780,8 @@ fn print_help(output: &mut W) -> io::Result<()> { history \n\ rollback \n\ install \n\ - send [--to ]\n\ + send \n\ + State/messages: JSON values; {{\"$actor\":\"name-or-@ref\"}} embeds an actor reference.\n\ tick [instruction-budget]\n\ inspect \n\ actors\n\ @@ -715,6 +800,101 @@ fn print_help(output: &mut W) -> io::Result<()> { mod tests { use super::*; + #[test] + fn value_parser_preserves_json_shapes_and_distinguishes_references() { + let reference = ActorRef::new(); + let state = ReplState { + bindings: BTreeMap::from([("receiver".to_owned(), reference.to_string())]), + }; + let input = + r#"{"text": "hello world", "items": [null, true, -42, {"$actor": "receiver"}]}"#; + assert_eq!( + parse_value(input, &state).unwrap(), + Value::Record(BTreeMap::from([ + ("text".to_owned(), Value::String("hello world".to_owned())), + ( + "items".to_owned(), + Value::List(vec![ + Value::Null, + Value::Bool(true), + Value::Integer(-42), + Value::ActorRef(reference) + ]) + ), + ])) + ); + assert_eq!( + parse_value(r#"{"$record": {"$actor": "literal"}}"#, &state).unwrap(), + Value::Record(BTreeMap::from([( + "$actor".to_owned(), + Value::String("literal".to_owned()) + )])) + ); + for input in [ + "1.5", + "9223372036854775808", + r#"{"$actor": "missing"}"#, + r#"{"$actor": 1}"#, + "null trailing", + ] { + assert!(parse_value(input, &state).is_err(), "{input}"); + } + } + + #[test] + fn repl_spawns_and_sends_nested_values_without_losing_whitespace() + -> Result<(), Box> { + let mut server = Server::new(); + dispatch( + &mut server, + Command::PublishBehavior { + name: "echo".to_owned(), + expected_revision: None, + source: "pub fn handle(state, message) { (message, []) }".to_owned(), + }, + )?; + let mut state = ReplState::default(); + let mut output = Vec::new(); + execute_line( + r#"let echo = spawn echo {"text": "hello world", "items": [true, null]}"#, + &mut output, + &mut server, + &mut state, + )?; + let actor_ref = state.bindings.get("echo").ok_or("spawn failed")?.clone(); + let Reply::Actor(snapshot) = dispatch( + &mut server, + Command::InspectActor { + actor_ref: actor_ref.clone(), + }, + )? + else { + return Err("unexpected inspect reply".into()); + }; + assert_eq!( + snapshot.state, + parse_value(r#"{"text":"hello world","items":[true,null]}"#, &state)? + ); + execute_line( + r#"send echo {"nested": [{"$actor": "echo"}, "two spaces"]}"#, + &mut output, + &mut server, + &mut state, + )?; + execute_line("tick 100", &mut output, &mut server, &mut state)?; + let Reply::Actor(snapshot) = dispatch(&mut server, Command::InspectActor { actor_ref })? + else { + return Err("unexpected inspect reply".into()); + }; + assert_eq!( + snapshot.state, + parse_value(r#"{"nested": [{"$actor":"echo"}, "two spaces"]}"#, &state)? + ); + assert_eq!(snapshot.status, runa_protocol::ActorStatusDto::Idle); + assert!(!String::from_utf8(output)?.contains("error:")); + Ok(()) + } + #[test] fn connection_arguments_require_an_explicit_socket_address() { assert_eq!( @@ -749,7 +929,9 @@ mod tests { let mut client = TcpClient::connect(address)?; let mut output = Vec::new(); run_repl( - io::Cursor::new("let actor = spawn counter 5\nsend actor 0 7\ntick 100\nquit\n"), + io::Cursor::new( + "let actor = spawn counter 5\nsend actor {\"op\":\"add\",\"value\":7}\ntick 100\nquit\n", + ), &mut output, &mut client, )?; @@ -770,8 +952,18 @@ mod tests { )?; let output = String::from_utf8(output)?; assert!(output.contains(actor_ref)); - assert!(output.contains("state: 12")); assert!(!output.contains("error:")); + let Reply::Actor(snapshot) = dispatch( + &mut client, + Command::InspectActor { + actor_ref: actor_ref.to_owned(), + }, + )? + else { + return Err("unexpected inspect reply".into()); + }; + assert_eq!(snapshot.state, Value::Integer(12)); + assert_eq!(snapshot.status, runa_protocol::ActorStatusDto::Idle); drop(client); worker.join().expect("TCP image worker completed"); Ok(()) @@ -795,16 +987,27 @@ mod tests { fn repl_drives_actor_messages_and_inspection() -> Result<(), Box> { let mut server = server(); let input = io::Cursor::new( - "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 sender = spawn counter 0\nlet receiver = spawn counter 10\nsend sender {\"op\": \"send\", \"value\": 5, \"target\": {\"$actor\": \"receiver\"}}\ntick 100\ninspect sender\ninspect receiver\nevents\nquit\n", ); let mut output = Vec::new(); run_repl(input, &mut output, &mut server)?; let output = String::from_utf8(output)?; assert!(output.contains("sender = @")); assert!(output.contains("receiver = @")); - assert!(output.contains("state: 15")); + assert!(!output.contains("error:"), "{output}"); assert!(output.contains("Committed {")); assert!(output.contains("actor_ref: \"")); + let Reply::Actors(actors) = dispatch(&mut server, Command::ListActors)? else { + return Err("unexpected actors reply".into()); + }; + assert_eq!(actors.len(), 2); + assert!( + actors + .iter() + .all(|actor| actor.status == runa_protocol::ActorStatusDto::Idle) + ); + assert!(actors.iter().any(|actor| actor.state == Value::Integer(1))); + assert!(actors.iter().any(|actor| actor.state == Value::Integer(15))); Ok(()) } @@ -812,7 +1015,7 @@ mod tests { fn repl_reports_bad_input_and_continues() -> Result<(), Box> { let mut server = server(); let input = io::Cursor::new( - "spawn 0\nspawn counter nope\nsend 000000000000000000000 0 1\ntick 1 0\nunknown\nquit\n", + "spawn 0\nspawn counter nope\nsend 000000000000000000000 {\"op\":\"add\",\"value\":1}\ntick 1 0\nunknown\nquit\n", ); let mut output = Vec::new(); run_repl(input, &mut output, &mut server)?; @@ -834,7 +1037,7 @@ mod tests { "/../../runa-std/actors/counter.rn" ); let input = io::Cursor::new(format!( - "behaviors\npublish counter 1 {source_path}\nhistory counter\nrollback counter 2 1\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" + "behaviors\npublish counter 1 {source_path}\nhistory counter\nrollback counter 2 1\nlet actor = spawn counter 0\ninstall actor counter\nsend actor {{\"op\":\"add\",\"value\":1}}\ntick 1 1\ncancel actor\npause actor\nresume actor\ninspect actor\nevents\nquit\n" )); let mut output = Vec::new(); run_repl(input, &mut output, &mut server)?; @@ -880,5 +1083,16 @@ mod tests { 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"]); + let unicode_whitespace = "inspect\u{2003}se"; + assert_eq!( + helper.candidates(unicode_whitespace, unicode_whitespace.len()), + vec!["sender"] + ); + let line = r#"send sender {"target": {"$actor": "se"#; + assert_eq!(helper.candidates(line, line.len()), vec!["sender"]); + assert_eq!( + ReplHelper::completion_start(line, line.len()), + (line.len() - 2, true) + ); } } diff --git a/docs/architecture.md b/docs/architecture.md index 75556ca..574a0c7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -310,7 +310,7 @@ The diagram is conceptual; Cargo dependencies should be kept minimal. A dependen | Crate | Owns | May depend on | | ---------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------- | -| `runa-core` | Small, stable value types shared across independent layers: 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: actor references, source and implementation IDs, portable actor values, common error identifiers | Minimal external libraries only | | `runa-code` | Original source retention, versioned source/implementation hashing, compile-input fingerprints, in-memory compiled artifact cache | `runa-core`, `runa-rune`, BLAKE3 | | `runa-actor` | Mailbox and state-transition semantics, message reservation, acknowledgment, effects, actor lifecycle definitions | `runa-core` | | `runa-rune` | Rune compilation, VM execution, suspension, interruption hooks, language-specific diagnostics | `runa-core`, Rune libraries | @@ -439,11 +439,14 @@ This is a **target layout**, not a request to generate empty crates. For Phase 0 The current runtime prototype supports implementation installation only at an idle handler boundary, cooperative pause/resume between scheduler ticks, and an -explicit host-provided migration closure for its temporary `i64` actor state. +explicit host-provided migration closure for structured actor state. Actor snapshots expose active VM instruction progress, instruction pointer, -and call-frame depth. Structured state schemas, version compatibility checks, -Rune-authored migration functions, source mapping, and local-variable -inspection remain future work. +and call-frame depth. State and messages are owned portable trees of null, +booleans, integers, strings, lists, records, and distinct actor references. +Rune handlers receive `(state, message)` and return state plus explicit sends; +Rune heap objects never become shared committed actor data. State schemas, +version compatibility checks, Rune-authored migration functions, source +mapping, and local-variable inspection remain future work. ### Phase 5 — Image API diff --git a/docs/protocol.md b/docs/protocol.md index ce46352..5e688f5 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -7,9 +7,9 @@ when started with `--listen`, over TCP bound to an IPv4 or IPv6 loopback address. The TCP listener is unauthenticated and is intended only for trusted local development. -## Version 2 +## Version 3 -Every request and response carries `version: 2`. Requests also carry a +Every request and response carries `version: 3`. Requests also carry a non-empty `request_id` used to correlate the response. A server must reject unsupported versions with `unsupported_version` rather than interpreting a request using a different contract. Events carry the protocol version and do @@ -19,23 +19,26 @@ Requests have a tagged command object: ```json { - "version": 2, + "version": 3, "request_id": "req-1", "command": { "operation": "send", "parameters": { "actor_ref": "xK8_pQ2-Lm5_012345678", "message": { - "kind": 2, - "value": 9, - "target_ref": "Hj2_9sM0-pQ4_7aBcD3eF" + "type": "record", + "data": { + "op": { "type": "string", "data": "send" }, + "value": { "type": "integer", "data": 9 }, + "target": { "type": "actor_ref", "data": "Hj2_9sM0-pQ4_7aBcD3eF" } + } } } } } ``` -The version 2 operations are behavior publication, rollback, lookup, source +The version 3 operations are behavior publication, rollback, lookup, source retrieval, actor spawn, send, tick, inspect, actor listing, event retrieval, stop, restart, pause, resume, cancellation, and behavior installation. Migration closures are intentionally excluded: they are host callbacks, not @@ -45,11 +48,40 @@ Responses echo the request ID and carry either a typed reply or a protocol error. Events use a separate versioned envelope. Actor and behavior snapshots are wire DTOs, not serialized runtime objects; implementation identities are represented by their stable display strings. 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 +21-character URL-safe NanoID string carried in `actor_ref`. The server maps known service failures to the defined error codes and otherwise uses `runtime_failure` rather than claiming a more specific category. +### Portable values + +Spawn state, send messages, snapshot state/mailbox/reservations, and migration +event states use the same tagged value shape: + +| Value | JSON | +| --- | --- | +| Null | `{"type":"null"}` | +| Boolean | `{"type":"bool","data":true}` | +| Signed 64-bit integer | `{"type":"integer","data":42}` | +| String | `{"type":"string","data":"hello"}` | +| List | `{"type":"list","data":[, ...]}` | +| Record | `{"type":"record","data":{"key":, ...}}` | +| Actor reference | `{"type":"actor_ref","data":"<21-character-NanoID>"}` | + +References are distinct from strings. Their shape is validated even when +nested; existence is checked when a reference is used as a send destination, +not merely stored as data. The old message envelope and protocol v2 are not +accepted. The REPL's natural JSON and `$actor`/`$record` conveniences are +client syntax, not wire encodings. + +Each state/message permits a maximum nesting depth of 32 (root depth 0), +4,096 nodes, and 1 MiB of UTF-8 string/key bytes. Requests are also subject to +the transport's 1 MiB total line limit, including tags and envelope. +Unsupported value tags, malformed references, floats, and out-of-range integers +fail decoding; well-formed requests exceeding value limits receive +`invalid_request`. Invalid Rune transition output fails the actor without +acknowledging the reserved message, changing committed state, or enqueueing +any of its sends. + `get_events` drains and returns the events currently accumulated by this single-worker image. It is not a durable subscription or replayable event log. Both transports limit each request line to 1 MiB before parsing. Malformed @@ -76,7 +108,7 @@ publishes the bundled `counter` behavior on image startup, not on connection. The stdio executable continues to start with an empty image. The current validation checks the protocol version, request ID, actor-reference -shape, positive revisions where required, non-empty behavior names, mailbox +shape, structured-value limits, positive revisions where required, non-empty behavior names, mailbox capacity, and instruction budget. Behavior-name grammar, actor existence, current behavior revisions, mailbox capacity at send time, and all runtime invariants remain authoritative at the application-service boundary. diff --git a/docs/runtime.md b/docs/runtime.md index 80ad181..d9d4901 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -32,7 +32,7 @@ runa> let sender = spawn counter 0 sender = @ runa> let receiver = spawn counter 10 receiver = @ -runa> send sender 1 5 --to receiver +runa> send sender {"op": "send", "value": 5, "target": {"$actor": "receiver"}} message queued runa> tick 100 runa> inspect sender @@ -46,11 +46,27 @@ Up/Down navigate command history. `tick [instruction-budget]` explicitly advances the scheduler. The default per-actor slice budget is 32 instructions. +State and messages use JSON literals in the REPL, preserving whitespace inside +strings. `spawn ` and `let = spawn ` +require explicit initial state; `send ` takes one message value. +For example, `send receiver {"op": "add", "value": 7}` adds to the bundled +counter. Its supported operations are `add` and `send`; unknown operations +fail explicitly. + +The sole-key object `{"$actor": "receiver"}` resolves a local binding to a +distinct actor-reference value. A literal `@` or raw NanoID string +can be used in place of the binding. Ordinary strings are never implicitly +converted to references. To encode a literal record that would otherwise +match a marker, wrap it in `{"$record": {...}}`. The wrapper escapes only that +record; nested values still use the normal conversion rules. Tab completion +also offers references inside a `"$actor"` string. + ## Current contracts - `runa-actor` owns actor references, mailbox and implementation IDs, structured messages, snapshots, statuses and lifecycle events. -- `runa-core` owns stable source and implementation identity types. +- `runa-core` owns stable source/implementation identities, actor references, + and the portable owned value model shared by the runtime and protocol. - `runa-code` preserves original source, derives identities, caches compiled artifacts, and provides an in-memory named behavior registry with revisions. - `runa-rune` owns compilation and typed bounded invocation execution. @@ -94,7 +110,7 @@ when it is idle, has no active invocation, and has no reserved input. Queued messages may remain; they will use the newly installed implementation. Active or awaiting handlers keep their original implementation, and installation attempts during execution fail explicitly so the caller can retry at the next -handler boundary. `install` preserves the current `i64` state unchanged. +handler boundary. `install` preserves the current structured state unchanged. Runtime callers that need a state transformation can use `install_with_migration`, which receives the committed state and computes its @@ -103,8 +119,10 @@ requires the same idle safe point; if the migration returns an error, neither value changes and no installation event is emitted. A successful changed state emits `StateMigrated` as well as `BehaviorInstalled`. This prototype accepts a host-provided migration closure only: it does not infer schema -compatibility or execute migration code from Rune source. Its state contract -is still `i64`, so structured schema/version migration remains future work. +compatibility or execute migration code from Rune source. The closure receives +an owned copy of the structured state; an invalid result also leaves the +original state and implementation unchanged. Schema/version policy remains +future work. `pause` prevents an actor from being scheduled on subsequent worker ticks and retains its active invocation, reserved message, and prior `Running`, `Awaiting`, @@ -140,24 +158,48 @@ The cache is in-memory only. Invalid source is retained in the current identity is not a claim that Rune bytecode is portable or safe to persist. `CodeStore::compile_with_module` requires an explicit stable module identity; callers must change it when registered native functions or their contracts -change. The CLI currently creates a fresh `CodeStore` when it starts. +change. The server creates a fresh `CodeStore` when its image starts. -Prototype state is an `i64`; messages contain integer `kind` and `value` -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: +State and messages are owned `Value` trees: null, boolean, signed 64-bit +integer, UTF-8 string, list, string-keyed record, or actor reference. 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 interface: ```text -handle(state, kind, value, target_ref) - -> (new_state, [(target_ref, kind, value), ...]) +handle(state, message) -> (new_state, [(target_actor_ref, message), ...]) ``` -`target_ref` is an empty string when the incoming message has no routed target. -The returned effect vector describes sends; handlers cannot access the runtime -or directly mutate committed state. General schema/versioned values and -capability-bearing contexts are future work. +The former `kind`, `value`, and `target_ref` envelope is removed. Actors choose +their own message records. For example: + +```rune +pub fn handle(state, message) { + state.items.push(message.text); + (state, [(message.reply_to, #{accepted: true})]) +} +``` + +The runtime allocates a fresh Rune object graph for each state and message +input, and copies portable data back out before committing. A Rune handler can +mutate its invocation-local draft but cannot mutate committed state, the +reserved input, or another actor's data. Copies made for separate effect +messages do not share mutable Rune objects. + +Null maps to Rune unit `()`, lists to Rune vectors, records to anonymous +objects (`#{...}`), and references to an opaque native `runa::ActorReference`. +Effect targets must be references, not lookalike strings. References support +equality and cloning in Rune and can be retained and forwarded through +state/messages; this is not authentication. +Floats, nonempty tuples, options, closures, futures, custom Rune structs, and +other native objects are not portable actor data and fail conversion. Cyclic +or excessively deep Rune output also fails without committing. + +Each state/message is limited to depth 32 (root depth 0), 4,096 value nodes +(including the root), and 1 MiB of UTF-8 string/key bytes. A transition permits +at most 4,096 send effects. Transport line-size limits apply separately. +These are conversion limits, not a Rune heap quota or hostile-code sandbox. +Schema/version policy and capability-bearing contexts remain future work. ## Scheduling and lifecycle diff --git a/runa-std/actors/counter.rn b/runa-std/actors/counter.rn index b3c11c0..d36bd99 100644 --- a/runa-std/actors/counter.rn +++ b/runa-std/actors/counter.rn @@ -1,7 +1,9 @@ -pub fn handle(state, kind, value, target_ref) { - if kind == 1 { - (state + 1, [(target_ref, 0, 5)]) +pub fn handle(state, message) { + if message.op == "send" { + (state + 1, [(message.target, #{op: "add", value: message.value})]) + } else if message.op == "add" { + (state + message.value, []) } else { - (state + value, []) + panic("unknown counter operation") } }