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") } }