feat!: remove actors on stop and kill and default counter state
Remove actors after successful graceful shutdown or immediately on kill, including state, mailbox, logs, and source-path metadata. Keep failed actors available for inspection and repair. Preserve restart as an independent identity-retaining operation and mark removed VS Code snapshots read-only. Default the self-incrementing counter's missing state to zero on its first message and isolate finite control-test fixtures from the demo. BREAKING CHANGE: stop and kill permanently remove actors instead of retaining stopped instances. Removed references cannot be inspected, sent to, or restarted. Use restart directly to retain actor identity and state. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
parent
867eabb957
commit
f5decbdea4
11
CHANGELOG.md
11
CHANGELOG.md
|
|
@ -6,11 +6,18 @@ All notable changes to this project will be documented in this file.
|
|||
|
||||
### Changed
|
||||
|
||||
- **Breaking:** Successful graceful stop and immediate kill now remove actors,
|
||||
discarding their state, mailbox, and retained logs. Failed actors remain until
|
||||
killed. Restart remains an independent, identity-preserving operation.
|
||||
- Default the bundled self-incrementing counter's missing/null state to zero
|
||||
when it handles its first message.
|
||||
- Mark removed actors' open VS Code inspection snapshots read-only with
|
||||
`removed: true` while removing them from the actor list.
|
||||
- **Breaking:** Handler and shutdown-hook unit returns (including omitted returns)
|
||||
now retain the previously committed state instead of replacing it with null.
|
||||
Draft mutations are discarded; valid sends and message acknowledgement still commit.
|
||||
- Change the bundled counter's `send` operation to `start`, which queues an
|
||||
increment to itself using `ctx.this`.
|
||||
- Change the bundled counter to an `inc` operation that logs each increment
|
||||
and continuously queues another increment to itself using `ctx.this`.
|
||||
- **Breaking:** Protocol v5 adds actor log thresholds to snapshots, guarded
|
||||
editing operations, and opt-in log-history/follow operations and notifications.
|
||||
- Move image sources and actors into a dedicated Runa Activity Bar sidebar,
|
||||
|
|
|
|||
34
README.md
34
README.md
|
|
@ -17,7 +17,7 @@ addresses with an in-memory compile cache. Persistence, canonicalization,
|
|||
remote network access, and the graphical world are not implemented yet.
|
||||
A minimal VS Code extension can connect to the local image, list actors, and
|
||||
inspect state, queued/reserved messages, and execution telemetry. Inspection
|
||||
JSON is editable for inactive paused/stopped/failed actors; saving restores
|
||||
JSON is editable for inactive paused/failed actors; saving restores
|
||||
protected fields with a warning. The image owns mutable, directory-based Rune source files with
|
||||
independent commit history. The CLI exposes source writes, commit/history/rollback,
|
||||
reload/become, graceful shutdown, kill, restart, pause/resume, and cancellation through a trusted image
|
||||
|
|
@ -85,8 +85,8 @@ or run an image operation directly:
|
|||
|
||||
```sh
|
||||
cargo run -p runa-cli -- ps
|
||||
actor=$(cargo run -p runa-cli -- spawn actors/counter 1)
|
||||
cargo run -p runa-cli -- send "$actor" '{"op":"add","value":2}'
|
||||
actor=$(cargo run -p runa-cli -- spawn actors/counter)
|
||||
cargo run -p runa-cli -- send "$actor" '{"op":"inc"}'
|
||||
cargo run -p runa-cli -- tick 100
|
||||
```
|
||||
|
||||
|
|
@ -117,22 +117,22 @@ The server loads the bundled `source/actors/counter.rn` working file at startup.
|
|||
Connecting a client does not modify source or create a commit:
|
||||
|
||||
```text
|
||||
runa> let sender = spawn actors/counter 0
|
||||
sender = @<actor-reference>
|
||||
runa> let receiver = spawn actors/counter 10
|
||||
receiver = @<actor-reference>
|
||||
runa> send sender {"op": "send", "value": 5, "target": {"$actor": "receiver"}}
|
||||
runa> let counter = spawn actors/counter
|
||||
counter = @<actor-reference>
|
||||
runa> send counter {"op": "inc"}
|
||||
message queued
|
||||
runa> tick 100
|
||||
runa> inspect sender
|
||||
runa> inspect receiver
|
||||
runa> pause counter
|
||||
runa> inspect counter
|
||||
runa> events
|
||||
```
|
||||
|
||||
Actor references are generated NanoIDs rather than sequential numbers. The
|
||||
`let` form binds each spawned actor to a local name; references can also be
|
||||
entered directly with an `@` prefix. The sender commits state `1` and sends a
|
||||
message to the receiver, which commits state `15`. The REPL provides Tab
|
||||
entered directly with an `@` prefix. The counter treats missing/null state as
|
||||
zero on its first message, increments and logs its value, then queues another
|
||||
`inc` message to itself. It runs only when ticks advance; pause/resume controls
|
||||
its loop. An explicit integer initial state is preserved. The REPL provides Tab
|
||||
completion and Up/Down command history. `write actors/counter <local-file>`
|
||||
updates the image's working source. `commit actors/counter` records a snapshot
|
||||
with an automatically generated readable revision ID; `history actors/counter`
|
||||
|
|
@ -143,8 +143,11 @@ creating an actor; commits can preserve invalid work in progress.
|
|||
`reload <actor>` loads current source at its existing path, and
|
||||
`become <actor> <path>` switches scripts, preserving state and queued messages
|
||||
after the active handler completes. `stop` runs optional `on_shutdown(state)`
|
||||
after the active handler; `kill` discards it without shutdown. `restart` creates
|
||||
after the active handler and then removes the actor; `kill` removes it
|
||||
immediately without shutdown. Removal discards identity, state, mailbox, and
|
||||
retained logs. Failed actors remain inspectable until killed. `restart` creates
|
||||
a new incarnation with the same state and loaded script, without reloading.
|
||||
Restart is independent of stop/kill; a removed actor cannot be restarted.
|
||||
Pause/resume remain cooperative controls between scheduler slices. Use `help` in
|
||||
the session for the complete command list.
|
||||
State and messages support null, booleans, signed 64-bit integers, strings,
|
||||
|
|
@ -197,8 +200,9 @@ for connection restrictions, snapshot semantics, and tests.
|
|||
|
||||
## Repairing actors and reading logs
|
||||
|
||||
State and messages can be edited only on paused, stopped, or failed actors with
|
||||
no active invocation. A paused suspended handler must be killed first. State
|
||||
State and messages can be edited only on paused or failed actors with
|
||||
no active invocation. An active handler must finish or be cancelled through the
|
||||
runtime cancellation API; kill removes the actor rather than making it editable. State
|
||||
edits preserve messages; message edits preserve state. Neither restarts an actor.
|
||||
Reserved messages are retried first on restart; clearing the reserved array
|
||||
removes the failed input.
|
||||
|
|
|
|||
|
|
@ -224,7 +224,7 @@ struct Args {
|
|||
|
||||
#[derive(Debug, clap::Subcommand)]
|
||||
enum CliCommand {
|
||||
/// Replace committed state on a paused, stopped, or failed actor without active execution.
|
||||
/// Replace committed state on a paused or failed actor without active execution.
|
||||
SetState {
|
||||
actor: String,
|
||||
#[arg(allow_hyphen_values = true)]
|
||||
|
|
@ -283,9 +283,9 @@ enum CliCommand {
|
|||
History { path: String },
|
||||
/// Restore a source commit into the working file.
|
||||
Rollback { path: String, revision: String },
|
||||
/// Stop gracefully at a handler boundary.
|
||||
/// Stop gracefully and remove the actor after shutdown (requires scheduler ticks).
|
||||
Stop { actor: String },
|
||||
/// Kill the current invocation without shutdown.
|
||||
/// Remove the actor immediately, discarding state and messages without shutdown.
|
||||
Kill { actor: String },
|
||||
/// Restart with the same state and loaded script.
|
||||
Restart { actor: String },
|
||||
|
|
@ -1500,7 +1500,13 @@ mod tests {
|
|||
Command::WriteSource {
|
||||
path: COUNTER_BEHAVIOR.to_owned(),
|
||||
expected_source_id: None,
|
||||
source: include_str!("../../../runa-std/actors/counter.rn").to_owned(),
|
||||
source: r#"pub fn handle(ctx, state, sender, message) {
|
||||
if message.op == "send" {
|
||||
ctx.send(message.target, #{op: "add", value: message.value});
|
||||
state + 1
|
||||
} else { state + message.value }
|
||||
}"#
|
||||
.to_owned(),
|
||||
},
|
||||
)
|
||||
.expect("default behavior publishes");
|
||||
|
|
@ -1574,7 +1580,7 @@ mod tests {
|
|||
"/../../runa-std/actors/counter.rn"
|
||||
);
|
||||
let input = io::Cursor::new(format!(
|
||||
"sources\nwrite actors/counter {source_path}\ncommit actors/counter\nhistory actors/counter\nlet actor = spawn actors/counter 0\nbecome actor actors/counter\nreload actor\nsend actor {{\"op\":\"add\",\"value\":1}}\ntick 1 1\npause actor\nresume actor\nstop actor\ntick 100\nrestart actor\nkill actor\ninspect actor\nevents\nquit\n"
|
||||
"sources\nwrite actors/counter {source_path}\ncommit actors/counter\nhistory actors/counter\nlet actor = spawn actors/counter 0\nbecome actor actors/counter\nreload actor\nsend actor {{\"op\":\"inc\"}}\ntick 1 1\npause actor\nresume actor\nrestart actor\nstop actor\ntick 100\nlet doomed = spawn actors/counter 0\nkill doomed\nevents\nquit\n"
|
||||
));
|
||||
let mut output = Vec::new();
|
||||
run_repl(input, &mut output, &mut server)?;
|
||||
|
|
@ -1603,15 +1609,10 @@ mod tests {
|
|||
&mut server,
|
||||
)?;
|
||||
assert!(String::from_utf8(output)?.contains("restored `actors/counter`"));
|
||||
let actor_ref = match dispatch(&mut server, Command::ListActors)? {
|
||||
Reply::Actors(actors) => actors[0].actor_ref.clone(),
|
||||
match dispatch(&mut server, Command::ListActors)? {
|
||||
Reply::Actors(actors) => assert!(actors.is_empty()),
|
||||
other => return Err(format!("unexpected response: {other:?}").into()),
|
||||
};
|
||||
let actor = match dispatch(&mut server, Command::InspectActor { actor_ref })? {
|
||||
Reply::Actor(actor) => actor,
|
||||
other => return Err(format!("unexpected response: {other:?}").into()),
|
||||
};
|
||||
assert_eq!(actor.status, runa_protocol::ActorStatusDto::Stopped);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -229,6 +229,10 @@ impl ImageApi {
|
|||
|
||||
fn collect_runtime_events(&mut self) {
|
||||
for event in self.runtime.drain_events() {
|
||||
if let Event::Stopped(actor) = &event {
|
||||
self.actor_paths.remove(actor);
|
||||
self.pending_paths.remove(actor);
|
||||
}
|
||||
if let Event::BehaviorInstalled(actor, implementation) = &event
|
||||
&& self
|
||||
.pending_paths
|
||||
|
|
@ -251,6 +255,7 @@ impl ImageApi {
|
|||
if matches!(message, SystemMessage::Kill | SystemMessage::Restart) {
|
||||
self.pending_paths.remove(&actor);
|
||||
}
|
||||
self.collect_runtime_events();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -437,7 +442,7 @@ mod tests {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn killed_pending_become_preserves_the_installed_path() -> RuntimeResult<()> {
|
||||
fn restarting_pending_become_preserves_the_installed_path() -> RuntimeResult<()> {
|
||||
let mut image = ImageApi::new();
|
||||
image.write_source(
|
||||
"actors/first",
|
||||
|
|
@ -453,7 +458,6 @@ mod tests {
|
|||
let original = image.inspect(actor)?.implementation;
|
||||
image.r#become(actor, "actors/second")?;
|
||||
assert_eq!(image.actor_paths[&actor], "actors/first");
|
||||
image.system(actor, SystemMessage::Kill)?;
|
||||
image.restart(actor)?;
|
||||
assert_eq!(image.reload(actor)?, original);
|
||||
image.tick(1, capacity(100));
|
||||
|
|
@ -465,6 +469,11 @@ mod tests {
|
|||
assert_eq!(image.actor_paths[&actor], "actors/second");
|
||||
let current = image.inspect(actor)?.implementation;
|
||||
assert_eq!(image.reload(actor)?, current);
|
||||
image.system(actor, SystemMessage::Kill)?;
|
||||
assert!(image.inspect(actor).is_err());
|
||||
assert!(!image.actor_paths.contains_key(&actor));
|
||||
assert!(!image.pending_paths.contains_key(&actor));
|
||||
assert!(image.reload(actor).is_err());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -32,6 +32,11 @@ pub struct ActorLogs {
|
|||
}
|
||||
|
||||
impl ActorLogs {
|
||||
pub fn remove(&mut self, actor: ActorRef) {
|
||||
self.entries.remove(&actor);
|
||||
self.levels.remove(&actor);
|
||||
}
|
||||
|
||||
pub fn set_sink(&mut self, sink: impl Fn(LogEntry) + 'static) {
|
||||
self.sink = Some(Box::new(sink));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -170,14 +170,10 @@ impl Runtime {
|
|||
) -> ExperimentResult<()> {
|
||||
let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
|
||||
if actor.active.is_some()
|
||||
|| !matches!(
|
||||
actor.snapshot.status,
|
||||
Status::Paused | Status::Stopped | Status::Failed(_)
|
||||
)
|
||||
|| !matches!(actor.snapshot.status, Status::Paused | Status::Failed(_))
|
||||
{
|
||||
return Err(
|
||||
"editing requires a paused, stopped, or failed actor with no active invocation"
|
||||
.into(),
|
||||
"editing requires a paused or failed actor with no active invocation".into(),
|
||||
);
|
||||
}
|
||||
if let Some(state) = &state {
|
||||
|
|
@ -279,14 +275,8 @@ impl Runtime {
|
|||
}
|
||||
|
||||
pub fn kill(&mut self, id: ActorRef) -> ExperimentResult<()> {
|
||||
let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
|
||||
actor.active = None;
|
||||
actor.context.take_effects();
|
||||
actor.stopping = false;
|
||||
actor.shutdown = false;
|
||||
actor.pending_install = None;
|
||||
actor.paused_from = None;
|
||||
actor.snapshot.status = Status::Stopped;
|
||||
self.actors.remove(&id).ok_or("unknown actor")?;
|
||||
self.logs.borrow_mut().remove(id);
|
||||
self.events.push(Event::Stopped(id));
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -295,10 +285,7 @@ impl Runtime {
|
|||
match message {
|
||||
SystemMessage::Stop => self.stop(id),
|
||||
SystemMessage::Kill => self.kill(id),
|
||||
SystemMessage::Restart => {
|
||||
self.kill(id)?;
|
||||
self.restart(id)
|
||||
}
|
||||
SystemMessage::Restart => self.restart(id),
|
||||
SystemMessage::Pause => self.pause(id),
|
||||
SystemMessage::Resume => self.resume(id),
|
||||
SystemMessage::Reload(implementation) | SystemMessage::Become(implementation) => {
|
||||
|
|
@ -374,14 +361,16 @@ impl Runtime {
|
|||
|
||||
pub fn restart(&mut self, id: ActorRef) -> ExperimentResult<()> {
|
||||
let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
|
||||
if !matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) {
|
||||
return Err("stop the actor before restarting".into());
|
||||
}
|
||||
actor.snapshot.incarnation = actor
|
||||
.snapshot
|
||||
.incarnation
|
||||
.checked_add(1)
|
||||
.ok_or("incarnation exhausted")?;
|
||||
actor.active = None;
|
||||
actor.context.take_effects();
|
||||
actor.stopping = false;
|
||||
actor.shutdown = false;
|
||||
actor.pending_install = None;
|
||||
if let Some(message) = actor.snapshot.reserved.take() {
|
||||
actor.mailbox.push_front(Envelope {
|
||||
sender: actor.reserved_sender.take(),
|
||||
|
|
@ -484,12 +473,7 @@ impl Runtime {
|
|||
actor.snapshot.status = Status::Idle;
|
||||
actor.active = None;
|
||||
actor.reserved_sender = None;
|
||||
if actor.shutdown {
|
||||
actor.shutdown = false;
|
||||
actor.stopping = false;
|
||||
actor.snapshot.status = Status::Stopped;
|
||||
self.events.push(Event::Stopped(*id));
|
||||
}
|
||||
let shutdown = actor.shutdown;
|
||||
for effect in effects {
|
||||
self.actors
|
||||
.get_mut(&effect.recipient)
|
||||
|
|
@ -501,6 +485,9 @@ impl Runtime {
|
|||
});
|
||||
}
|
||||
self.events.push(Event::Committed(*id));
|
||||
if shutdown {
|
||||
self.kill(*id)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -519,9 +506,7 @@ impl Runtime {
|
|||
if actor.stopping {
|
||||
let program = self.implementations[&actor.snapshot.implementation].program();
|
||||
if !program.has_function("on_shutdown") {
|
||||
actor.snapshot.status = Status::Stopped;
|
||||
actor.stopping = false;
|
||||
self.events.push(Event::Stopped(id));
|
||||
self.kill(id).expect("scheduled actor exists");
|
||||
continue;
|
||||
}
|
||||
let invocation = (|| -> ExperimentResult<_> {
|
||||
|
|
@ -606,6 +591,70 @@ mod tests {
|
|||
use super::*;
|
||||
use runa_core::MAX_VALUE_NODES;
|
||||
|
||||
#[test]
|
||||
fn bundled_counter_defaults_missing_state_to_zero() -> ExperimentResult<()> {
|
||||
let mut runtime = Runtime::default();
|
||||
let implementation = publish(
|
||||
&mut runtime,
|
||||
include_str!("../../../runa-std/actors/counter.rn"),
|
||||
)?;
|
||||
for (initial, expected) in [(Value::Null, 1), (Value::Integer(10), 11)] {
|
||||
let actor = runtime.spawn(implementation, initial, capacity(2))?;
|
||||
runtime.send(
|
||||
actor,
|
||||
Value::Record(BTreeMap::from([(
|
||||
"op".to_owned(),
|
||||
Value::String("inc".to_owned()),
|
||||
)])),
|
||||
)?;
|
||||
for _ in 0..1000 {
|
||||
runtime.tick(capacity(1));
|
||||
if runtime.inspect(actor)?.state == Value::Integer(expected) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
let snapshot = runtime.inspect(actor)?;
|
||||
assert_eq!(snapshot.state, Value::Integer(expected));
|
||||
assert_eq!(snapshot.queued.len(), 1);
|
||||
assert_eq!(runtime.logs(&[actor], 0)?.len(), 1);
|
||||
runtime.kill(actor)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stop_removes_paused_actors_and_kill_discards_identity_mailbox_and_logs()
|
||||
-> ExperimentResult<()> {
|
||||
let mut runtime = Runtime::default();
|
||||
let implementation = publish(
|
||||
&mut runtime,
|
||||
"pub fn handle(ctx, state, sender, message) { ctx.info(\"started\"); loop {} }",
|
||||
)?;
|
||||
let actor = runtime.spawn(implementation, Value::Integer(5), capacity(3))?;
|
||||
runtime.send(actor, Value::Null)?;
|
||||
runtime.send(actor, Value::Integer(2))?;
|
||||
for _ in 0..10 {
|
||||
runtime.tick(capacity(32));
|
||||
}
|
||||
assert_eq!(runtime.logs(&[actor], 0)?.len(), 1);
|
||||
runtime.set_log_level(actor, LogLevel::Fatal)?;
|
||||
runtime.kill(actor)?;
|
||||
assert!(runtime.inspect(actor).is_err());
|
||||
assert!(runtime.restart(actor).is_err());
|
||||
assert!(runtime.send(actor, Value::Null).is_err());
|
||||
assert!(runtime.kill(actor).is_err());
|
||||
assert!(runtime.actors().is_empty());
|
||||
assert!(runtime.logs(&[], 0)?.is_empty());
|
||||
assert_eq!(runtime.logs.borrow().level(actor), LogLevel::Info);
|
||||
let idle = runtime.spawn(implementation, Value::Null, capacity(1))?;
|
||||
runtime.pause(idle)?;
|
||||
runtime.stop(idle)?;
|
||||
assert!(runtime.inspect(idle).is_ok());
|
||||
runtime.tick(capacity(32));
|
||||
assert!(runtime.inspect(idle).is_err());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn optional_state_retains_committed_state_and_commits_sends() -> ExperimentResult<()> {
|
||||
let mut runtime = Runtime::default();
|
||||
|
|
@ -617,7 +666,10 @@ mod tests {
|
|||
if message == 2 { return (); }
|
||||
#{value: 7}
|
||||
}
|
||||
pub fn on_shutdown(state) { state.value = 1000; }"#,
|
||||
pub fn on_shutdown(state) {
|
||||
if state.value != 7 { panic("incorrect shutdown state"); }
|
||||
state.value = 1000;
|
||||
}"#,
|
||||
)?;
|
||||
let initial = Value::Record(BTreeMap::from([("value".to_owned(), Value::Integer(3))]));
|
||||
let actor = runtime.spawn(implementation, initial.clone(), capacity(4))?;
|
||||
|
|
@ -650,10 +702,12 @@ mod tests {
|
|||
for _ in 0..50 {
|
||||
runtime.tick(capacity(32));
|
||||
}
|
||||
assert_eq!(runtime.inspect(actor)?.status, Status::Stopped);
|
||||
assert_eq!(
|
||||
runtime.inspect(actor)?.state,
|
||||
Value::Record(BTreeMap::from([("value".to_owned(), Value::Integer(7))]))
|
||||
assert!(runtime.inspect(actor).is_err());
|
||||
assert!(
|
||||
runtime
|
||||
.drain_events()
|
||||
.iter()
|
||||
.any(|event| matches!(event, Event::Stopped(id) if *id == actor))
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -749,7 +803,11 @@ mod tests {
|
|||
.is_err()
|
||||
);
|
||||
runtime.kill(active)?;
|
||||
runtime.edit_actor(active, Some(Value::Integer(9)), Some((None, vec![])))?;
|
||||
assert!(
|
||||
runtime
|
||||
.edit_actor(active, Some(Value::Integer(9)), Some((None, vec![])))
|
||||
.is_err()
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -812,8 +870,8 @@ mod tests {
|
|||
.all(|pair| pair[0].sequence < pair[1].sequence)
|
||||
);
|
||||
runtime.kill(second)?;
|
||||
assert_eq!(runtime.logs(&[second], 0)?.len(), 1000);
|
||||
assert_eq!(runtime.logs(&[], 0)?.len(), 1006);
|
||||
assert!(runtime.logs(&[second], 0).is_err());
|
||||
assert_eq!(runtime.logs(&[], 0)?.len(), 6);
|
||||
assert_eq!(emitted.borrow().len(), 1011);
|
||||
assert_eq!(emitted.borrow().last(), retained.last());
|
||||
Ok(())
|
||||
|
|
@ -916,7 +974,7 @@ mod tests {
|
|||
let mut runtime = Runtime::default();
|
||||
let implementation = publish(
|
||||
&mut runtime,
|
||||
"pub fn handle(ctx, state, sender, message) { let n = 0; while n < 20 { n += 1; } state + message } pub fn on_shutdown(state) { state + 100 }",
|
||||
"pub fn handle(ctx, state, sender, message) { let n = 0; while n < 20 { n += 1; } state + message } pub fn on_shutdown(state) { if state != 1 { panic(\"wrong state\"); } state + 100 }",
|
||||
)?;
|
||||
let actor = runtime.spawn(implementation, Value::Integer(0), capacity(3))?;
|
||||
runtime.send(actor, Value::Integer(1))?;
|
||||
|
|
@ -927,16 +985,14 @@ mod tests {
|
|||
for _ in 0..200 {
|
||||
runtime.tick(capacity(32));
|
||||
}
|
||||
let snapshot = runtime.inspect(actor)?;
|
||||
assert_eq!(snapshot.state, Value::Integer(101), "{snapshot:?}");
|
||||
assert_eq!(snapshot.status, Status::Stopped);
|
||||
assert_eq!(snapshot.queued, vec![Value::Integer(2)]);
|
||||
assert_eq!(snapshot.reserved, None);
|
||||
runtime.system(actor, SystemMessage::Restart)?;
|
||||
for _ in 0..200 {
|
||||
runtime.tick(capacity(32));
|
||||
}
|
||||
assert_eq!(runtime.inspect(actor)?.state, Value::Integer(103));
|
||||
assert!(runtime.inspect(actor).is_err());
|
||||
assert!(runtime.system(actor, SystemMessage::Restart).is_err());
|
||||
assert!(
|
||||
runtime
|
||||
.drain_events()
|
||||
.iter()
|
||||
.any(|event| matches!(event, Event::Stopped(id) if *id == actor))
|
||||
);
|
||||
let self_send = publish(
|
||||
&mut runtime,
|
||||
"pub fn handle(ctx, state, sender, message) { ctx.send(message, 42); state + 1 } pub fn on_shutdown(state) { state + 100 }",
|
||||
|
|
@ -948,10 +1004,7 @@ mod tests {
|
|||
for _ in 0..100 {
|
||||
runtime.tick(capacity(32));
|
||||
}
|
||||
let snapshot = runtime.inspect(actor)?;
|
||||
assert_eq!(snapshot.state, Value::Integer(101));
|
||||
assert_eq!(snapshot.status, Status::Stopped);
|
||||
assert_eq!(snapshot.queued, vec![Value::Integer(42)]);
|
||||
assert!(runtime.inspect(actor).is_err());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -975,11 +1028,11 @@ mod tests {
|
|||
runtime.tick(capacity(32));
|
||||
}
|
||||
assert_eq!(runtime.inspect(actor)?.state, Value::Integer(21));
|
||||
runtime.system(actor, SystemMessage::Kill)?;
|
||||
assert_eq!(runtime.inspect(actor)?.status, Status::Stopped);
|
||||
assert_eq!(runtime.inspect(actor)?.state, Value::Integer(21));
|
||||
runtime.system(actor, SystemMessage::Restart)?;
|
||||
assert_eq!(runtime.inspect(actor)?.implementation, second);
|
||||
assert_eq!(runtime.inspect(actor)?.state, Value::Integer(21));
|
||||
runtime.system(actor, SystemMessage::Kill)?;
|
||||
assert!(runtime.inspect(actor).is_err());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -1002,12 +1055,14 @@ mod tests {
|
|||
for _ in 0..100 {
|
||||
runtime.tick(capacity(32));
|
||||
}
|
||||
if failed {
|
||||
let snapshot = runtime.inspect(actor)?;
|
||||
assert_eq!(snapshot.state, Value::Integer(7));
|
||||
if failed {
|
||||
assert!(matches!(snapshot.status, Status::Failed(_)));
|
||||
runtime.kill(actor)?;
|
||||
assert!(runtime.inspect(actor).is_err());
|
||||
} else {
|
||||
assert_eq!(snapshot.status, Status::Stopped);
|
||||
assert!(runtime.inspect(actor).is_err());
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
|
|
@ -1184,11 +1239,11 @@ mod tests {
|
|||
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.kill(first)?;
|
||||
assert_eq!(runtime.inspect(first)?.status, Status::Stopped);
|
||||
runtime.restart(first)?;
|
||||
assert_eq!(runtime.inspect(first)?.incarnation, 2);
|
||||
assert_eq!(runtime.inspect(first)?.queued.len(), 1);
|
||||
runtime.kill(first)?;
|
||||
assert!(runtime.inspect(first).is_err());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -1713,7 +1768,7 @@ mod tests {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn stop_and_restart_preserve_reserved_and_queued_fifo_order() -> ExperimentResult<()> {
|
||||
fn restart_preserves_reserved_and_queued_fifo_order() -> ExperimentResult<()> {
|
||||
let mut runtime = Runtime::default();
|
||||
let code = publish(
|
||||
&mut runtime,
|
||||
|
|
@ -1734,14 +1789,6 @@ mod tests {
|
|||
assert_eq!(running.reserved, Some(first.clone()));
|
||||
assert_eq!(running.queued, vec![second.clone(), third.clone()]);
|
||||
|
||||
runtime.kill(actor)?;
|
||||
let stopped = runtime.inspect(actor)?;
|
||||
assert_eq!(stopped.status, Status::Stopped);
|
||||
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, Value::Integer(0));
|
||||
|
|
|
|||
|
|
@ -563,14 +563,13 @@ mod tests {
|
|||
ticks: 100,
|
||||
instruction_budget: 32,
|
||||
}));
|
||||
let Ok(Reply::Actor(snapshot)) = server
|
||||
let Err(error) = server
|
||||
.dispatch(request(Command::InspectActor { actor_ref }))
|
||||
.result
|
||||
else {
|
||||
panic!("inspect failed");
|
||||
panic!("removed actor should not be inspectable");
|
||||
};
|
||||
assert_eq!(snapshot.status, ActorStatusDto::Stopped);
|
||||
assert_eq!(snapshot.state, Value::Integer(107));
|
||||
assert_eq!(error.code, ErrorCode::NotFound);
|
||||
let Ok(Reply::SourceHistory(history)) = server
|
||||
.dispatch(request(Command::SourceHistory { path }))
|
||||
.result
|
||||
|
|
|
|||
|
|
@ -8,7 +8,34 @@ use runa_protocol::{Command, Reply};
|
|||
use tokio_tungstenite::tungstenite::{self, Message, client::IntoClientRequest};
|
||||
|
||||
fn start_server() -> (ServerProcess, SocketAddr) {
|
||||
start_server_with(&["--listen", "127.0.0.1:0"], "invalid")
|
||||
let (server, address) = start_server_with(&["--listen", "127.0.0.1:0"], "invalid");
|
||||
let mut client = WebSocketClient::connect(address).unwrap();
|
||||
install_counter_fixture(&mut client);
|
||||
client.close().unwrap();
|
||||
(server, address)
|
||||
}
|
||||
|
||||
fn install_counter_fixture(client: &mut WebSocketClient) {
|
||||
let Reply::Source(source) = client
|
||||
.request(Command::ReadSource {
|
||||
path: "actors/counter".to_owned(),
|
||||
})
|
||||
.unwrap()
|
||||
.result
|
||||
.unwrap()
|
||||
else {
|
||||
panic!("expected counter source");
|
||||
};
|
||||
client
|
||||
.request(Command::WriteSource {
|
||||
path: "actors/counter".to_owned(),
|
||||
expected_source_id: Some(source.source_id),
|
||||
source: "pub fn handle(ctx, state, sender, message) { state + message.value }"
|
||||
.to_owned(),
|
||||
})
|
||||
.unwrap()
|
||||
.result
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn start_server_with(args: &[&str], listen_env: &str) -> (ServerProcess, SocketAddr) {
|
||||
|
|
@ -278,6 +305,7 @@ fn websocket_server_accepts_local_protocol_requests() {
|
|||
let mut client = WebSocketClient::connect(address).unwrap();
|
||||
let response = client.request(Command::ListActors).unwrap();
|
||||
assert_eq!(response.result, Ok(Reply::Actors(Vec::new())));
|
||||
install_counter_fixture(&mut client);
|
||||
|
||||
let mut first = WebSocketClient::connect(address).unwrap();
|
||||
let mut second = WebSocketClient::connect(address).unwrap();
|
||||
|
|
|
|||
|
|
@ -126,7 +126,9 @@ change a buffered message. Timers and durable external requests remain planned.
|
|||
System lifecycle messages are a separate host-only control path: actors cannot
|
||||
construct stop, kill, restart, pause, resume, reload, or become commands through
|
||||
ordinary sends. Stop completes the current handler then invokes optional
|
||||
`on_shutdown(state)` before stopping; kill skips shutdown. Restart retains the
|
||||
`on_shutdown(state)` before removing the actor; kill removes it immediately
|
||||
without shutdown. Failed actors remain for inspection/repair until killed.
|
||||
Removal discards state, mailbox, and retained logs. Restart is a separate operation that retains the
|
||||
loaded script and state. Reload/become switch immutable implementations only
|
||||
after an active handler completes.
|
||||
|
||||
|
|
|
|||
|
|
@ -70,8 +70,13 @@ uncommitted; compilation failures do not spawn an actor.
|
|||
System message shape is `{kind: "stop"|"kill"|"restart"|"pause"|"resume"|"reload"}`
|
||||
or `{kind: "become", path: "actors/other"}`. This is a trusted host request;
|
||||
actor `ctx.send` and `ctx.send_from` accept only ordinary data. Stop runs optional
|
||||
`on_shutdown` after the active handler, kill drops it without shutdown, restart
|
||||
retains the loaded script and state, and reload/become defer installation to
|
||||
`on_shutdown` after the active handler and removes the actor on success. Kill
|
||||
removes the actor immediately without shutdown, including failed actors. Removed
|
||||
references return `not_found`; state, mailbox, and retained logs are discarded.
|
||||
Failed actors are not automatically removed. A `Stopped` event denotes removal,
|
||||
and `image_changed` invalidates actor lists. Restart is independent of stop/kill and
|
||||
retains identity, mailbox, the loaded script and state; removed actors cannot be
|
||||
restarted. Reload/become defer installation to
|
||||
the next handler boundary. The scheduler still advances only through `tick`.
|
||||
|
||||
Responses echo the request ID and carry either a typed reply or a protocol
|
||||
|
|
@ -161,7 +166,7 @@ the actor tree and open snapshot documents. Manual refresh remains available.
|
|||
null to leave it unchanged. `messages` is null to leave the mailbox unchanged,
|
||||
or `{reserved,queued}` where reserved is a tagged value or null for no
|
||||
reservation, and queued is an ordered array of tagged values. At least one
|
||||
edit is required. Only paused/stopped/failed actors with no active execution
|
||||
edit is required. Only paused/failed actors with no active execution
|
||||
are editable; values and mailbox capacity are validated atomically.
|
||||
- `set_log_level {actor_ref,level}`: level is one of `trace`, `debug`, `info`,
|
||||
`warn`, `error`, `fatal`. Both mutations return `acknowledged` and invalidate
|
||||
|
|
|
|||
|
|
@ -6,8 +6,9 @@
|
|||
`set-messages <actor> {"reserved":[],"queued":[]}` replaces the reserved and queued
|
||||
messages without changing state. The reserved array has zero or one element
|
||||
(so a null-valued message is distinguishable from no reservation).
|
||||
Both operations require paused, stopped, or failed status and no active invocation.
|
||||
Idle actors must be paused first; a paused active invocation must be killed first.
|
||||
Both operations require paused or failed status and no active invocation.
|
||||
Idle actors must be paused first; an active invocation must finish or be
|
||||
cancelled through the runtime API. Kill removes the actor rather than making it editable.
|
||||
All value and capacity checks finish before any changes are applied. Edits do
|
||||
not resume/restart an actor or change its implementation/incarnation.
|
||||
Unchanged messages at unchanged positions retain sender metadata; replaced
|
||||
|
|
@ -214,7 +215,7 @@ change. The server creates a fresh `CodeStore` when its image starts.
|
|||
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
|
||||
actors are spawned and remain stable across restart; stop/kill removes the actor. Mailbox IDs are
|
||||
separate internal counters. Rune handlers have this interface:
|
||||
|
||||
```text
|
||||
|
|
@ -227,7 +228,7 @@ or falling through without a value) keeps the previously committed state,
|
|||
not the invocation's mutable draft. Non-unit results replace state as before.
|
||||
Successful handlers still acknowledge the reserved message and commit validated
|
||||
context sends atomically. Invalid sends still fail the transition. A unit
|
||||
shutdown result stops the actor without changing its state. Unit cannot reset
|
||||
shutdown result finishes shutdown without changing state before the actor is removed. Unit cannot reset
|
||||
state to null anymore; host state editing can still set null explicitly.
|
||||
|
||||
The runtime uses internal envelopes containing sender and data, then passes
|
||||
|
|
@ -301,13 +302,18 @@ with cooperative instruction slices and commits a non-unit returned state,
|
|||
or retains committed state for a unit return. It does not
|
||||
drain queued input. A missing hook stops normally; a failing hook marks the actor
|
||||
failed and retains its last committed state. Stop unpauses a paused actor so
|
||||
shutdown can progress; explicit scheduler ticks are still needed.
|
||||
shutdown can progress; explicit scheduler ticks are still needed. Successful
|
||||
shutdown removes the actor, its state, queued/reserved messages, and retained
|
||||
logs/threshold. There is no persistent stopped actor. The `Stopped` lifecycle
|
||||
event records removal; subsequent inspection, sends, or restart return unknown actor.
|
||||
Failed actors (including shutdown failures) remain inspectable and repairable.
|
||||
Kill drops the invocation immediately between slices, skips shutdown, and
|
||||
retains committed state, mailbox and reserved message. Restart preserves its actor reference,
|
||||
removes the actor, including failed actors and all retained state/messages/logs.
|
||||
Restart is a separate operation that preserves its actor reference,
|
||||
mailbox ID, implementation ID, committed state, and queued-message order; it
|
||||
increments its incarnation and puts the reserved message at the front of the
|
||||
queue.
|
||||
It can also restart a running actor, killing the current invocation first.
|
||||
It can also restart a running actor, dropping the current invocation first without removing the actor.
|
||||
It does not reload working source. Repeated failure of the same reserved message can therefore
|
||||
block later messages until an operator changes the implementation or otherwise
|
||||
resolves it; there is no automatic poison-message policy or supervision.
|
||||
|
|
@ -320,9 +326,9 @@ resolves it; there is no automatic poison-message policy or supervision.
|
|||
| Handler returns unit / no state | Last committed state retained; draft discarded | Acknowledged | All validated sends enqueue | `Idle` |
|
||||
| Rune execution or transition decoding fails | Last committed state retained | Retained | None from the returned transition | `Failed` |
|
||||
| Any target is unknown, stopped, failed, or over capacity | Last committed state retained | Retained | No sends from the transition enqueue | `Failed` |
|
||||
| Graceful stop | Active handler then shutdown state commits | Active input acknowledged | Valid active-handler sends commit | `Stopped` after shutdown |
|
||||
| Kill during an invocation | Last committed state retained | Retained | None from incomplete invocation | `Stopped` |
|
||||
| Restart after stop or failure | Retained until retried | Moved to queue front | None until retry completes | `Idle` |
|
||||
| Graceful stop | Discarded after successful shutdown | Active input acknowledged; queued input discarded | Valid active-handler sends commit | Removed after shutdown |
|
||||
| Kill during an invocation | Discarded | Discarded | None from incomplete invocation | Removed |
|
||||
| Restart while alive or failed | Retained until retried | Moved to queue front | None until retry completes | `Idle` |
|
||||
| Pause between ticks | Last committed state retained | Retained | None while paused | `Paused` |
|
||||
| Resume a paused invocation | Retained until invocation completes | Retained | None until completion | Prior runnable/awaiting status |
|
||||
| Cancel an invocation | Last committed state retained | Moved to queue front | None from incomplete transition | `Idle` |
|
||||
|
|
@ -339,7 +345,7 @@ reservations and events. External/native callback effects may already have
|
|||
occurred even when the Rune invocation later fails or is stopped.
|
||||
|
||||
Lifecycle tests verify that failed and cancelled reservations retry from the
|
||||
last committed state, that stop/restart preserves reserved-first FIFO order,
|
||||
last committed state, that restart preserves reserved-first FIFO order,
|
||||
and that cancellation followed by installation retries queued input in order
|
||||
with the new implementation. Tests also cover that a running actor's queued
|
||||
messages are not processed until the reservation finishes. Transition tests
|
||||
|
|
@ -353,7 +359,11 @@ transactional outbox, supervisors, GUI, or automatic image bootstrap. The
|
|||
behavior registry and implementation installation are process-local;
|
||||
installation requires an idle handler boundary and does not upgrade active
|
||||
execution. The example counter script is bundled through `include_str!`.
|
||||
Messages are not accepted by stopped/failed actors.
|
||||
Messages are not accepted by failed or stopping actors, or removed references.
|
||||
The bundled counter defaults unit/null initial state to zero when it handles its
|
||||
first message. `#{op: "inc"}` increments and logs the state, and queues another
|
||||
increment to itself. Explicit integer initial state is preserved. No handler
|
||||
runs at spawn, so a no-state counter snapshot stays null until its first message.
|
||||
|
||||
Native functions must be nonblocking. Dropping a future proves cleanup only
|
||||
for the controlled test operation, not rollback of arbitrary external effects.
|
||||
|
|
|
|||
|
|
@ -1,9 +1,10 @@
|
|||
pub fn handle(ctx, state, sender, message) {
|
||||
if message.op == "start" {
|
||||
ctx.send(ctx.this, #{op: "add", value: 1});
|
||||
state + 1
|
||||
} else if message.op == "add" {
|
||||
state + message.value
|
||||
let state = match state { () => 0, state => state };
|
||||
if message.op == "inc" {
|
||||
let new_state = state + 1;
|
||||
ctx.info(`New State: ${new_state}`);
|
||||
ctx.send(ctx.this, #{op: "inc"});
|
||||
new_state
|
||||
} else {
|
||||
panic("unknown counter operation")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -38,7 +38,9 @@ In that host:
|
|||
5. Use the **Advance Scheduler** button in the Actors toolbar (or
|
||||
**Runa: Advance Scheduler**) to execute an explicit number of ticks
|
||||
(instruction budget 32 per actor). Send only queues a message; it does not
|
||||
execute it. For the counter, send `{"op":"add","value":7}`, then advance.
|
||||
execute it. For the bundled counter, send `{"op":"inc"}`, then advance.
|
||||
It defaults missing state to zero on its first message and queues further
|
||||
increments to itself. Pause/resume controls its loop.
|
||||
6. Run **Runa: Disconnect** to close only this client connection.
|
||||
|
||||
JSON input supports null, booleans, signed 64-bit integers, strings, arrays,
|
||||
|
|
@ -64,7 +66,7 @@ machine running the extension host, not necessarily the desktop.
|
|||
committed state, FIFO queued messages, the reserved message, status, and
|
||||
active-execution telemetry. Tagged values distinguish actor references from
|
||||
strings and preserve signed 64-bit integers exactly.
|
||||
- Documents can be edited and saved while the actor is paused, stopped, or failed
|
||||
- Documents can be edited and saved while the actor is paused or failed
|
||||
without an active invocation. Edit the tagged `actor.state`, `actor.reserved`,
|
||||
`actor.queued`, and `actor.log_level` fields. Saving applies changed writable
|
||||
fields; changes to identity, status, telemetry, connection metadata, and other
|
||||
|
|
@ -79,6 +81,12 @@ machine running the extension host, not necessarily the desktop.
|
|||
documents. Bursts are coalesced, and changes arriving during a fetch trigger
|
||||
a follow-up fetch. **Refresh Image** is also available on demand.
|
||||
- Open documents are retained and marked `connected: false` on disconnect.
|
||||
- Stop removes the actor after graceful shutdown (advance the scheduler); kill
|
||||
removes it immediately, including failed actors. Failure alone does not remove
|
||||
an actor. Removal discards state, mailbox, and retained logs. Removed actors
|
||||
disappear from the tree; open snapshots keep their last values with
|
||||
`removed: true` and become read-only. Restart preserves a live/failed actor's
|
||||
identity but cannot recover a removed actor.
|
||||
- Explicit disconnect, reconnect, and extension deactivation complete the
|
||||
WebSocket close handshake. Deactivation waits for outstanding closes; an
|
||||
unresponsive peer is forcibly disconnected after five seconds. Protocol
|
||||
|
|
@ -112,7 +120,7 @@ in the Runa output channel and an error notification; commands are never retried
|
|||
|
||||
Actor context menus also offer **Edit Actor State...** and
|
||||
**Edit Reserved and Queued Messages...**. They prefill JSON from the current
|
||||
snapshot; editing requires a paused, stopped, or failed actor with no active
|
||||
snapshot; editing requires a paused or failed actor with no active
|
||||
execution. Message input uses `{"reserved":[],"queued":[]}` with zero or one
|
||||
reserved message. Clearing the reservation removes the failed input that would
|
||||
otherwise retry on restart. Edits do not resume actors. **Set Actor Log Level...**
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ test(
|
|||
const image = await startImage();
|
||||
let client: ImageClient | undefined;
|
||||
t.after(async () => {
|
||||
client?.close();
|
||||
await client?.close();
|
||||
await image.stop();
|
||||
});
|
||||
client = await ImageClient.connect(image.address, (error) => assert.equal(error, undefined));
|
||||
|
|
@ -35,11 +35,6 @@ test(
|
|||
snapshot = await client.inspectActor(ref);
|
||||
assert.deepEqual(snapshot.state, parseValue("12"));
|
||||
assert.deepEqual(snapshot.queued, []);
|
||||
// Graceful stop completes only when the scheduler advances.
|
||||
await client.system(ref, { kind: "stop" });
|
||||
assert.notEqual((await client.inspectActor(ref)).status.status, "stopped");
|
||||
await client.tick(10);
|
||||
assert.equal((await client.inspectActor(ref)).status.status, "stopped");
|
||||
await client.system(ref, { kind: "restart" });
|
||||
snapshot = await client.inspectActor(ref);
|
||||
assert.equal(snapshot.incarnation, 2n);
|
||||
|
|
@ -58,10 +53,6 @@ test(
|
|||
client.system(ref, { kind: "become", path: "actors/missing" }),
|
||||
ProtocolError,
|
||||
);
|
||||
await client.system(ref, { kind: "kill" });
|
||||
assert.equal((await client.inspectActor(ref)).status.status, "stopped");
|
||||
await client.system(ref, { kind: "restart" });
|
||||
assert.equal((await client.inspectActor(ref)).incarnation, 4n);
|
||||
const payload = parseValue(`{"nested":[9223372036854775807,{"$actor":"${ref}"}]}`);
|
||||
await client.send(ref, payload);
|
||||
assert.deepEqual((await client.inspectActor(ref)).queued, [payload]);
|
||||
|
|
@ -71,5 +62,18 @@ test(
|
|||
/1 MiB/,
|
||||
);
|
||||
assert.equal((await client.listActors()).length, 1);
|
||||
await client.system(ref, { kind: "stop" });
|
||||
assert.equal((await client.inspectActor(ref)).actor_ref, ref);
|
||||
await client.tick(100);
|
||||
await assert.rejects(client.inspectActor(ref), /unknown actor/);
|
||||
await assert.rejects(client.system(ref, { kind: "restart" }), /unknown actor/);
|
||||
assert.deepEqual(await client.listActors(), []);
|
||||
const failed = await client.spawn("actors/counter", parseValue("0"));
|
||||
await client.send(failed, parseValue('"bad message"'));
|
||||
await client.tick(100);
|
||||
assert.equal((await client.inspectActor(failed)).status.status, "failed");
|
||||
await client.system(failed, { kind: "kill" });
|
||||
await assert.rejects(client.inspectActor(failed), /unknown actor/);
|
||||
assert.deepEqual(await client.listActors(), []);
|
||||
},
|
||||
);
|
||||
|
|
|
|||
|
|
@ -68,16 +68,6 @@ export async function run(): Promise<void> {
|
|||
snapshot = JSON.parse(controlled.getText()).actor;
|
||||
assert.equal(snapshot.state.data, 12);
|
||||
assert.equal(snapshot.queued.length, 0);
|
||||
await vscode.commands.executeCommand("runa.stop", selection);
|
||||
await vscode.commands.executeCommand("runa.inspect", selection);
|
||||
assert.notEqual(JSON.parse(controlled.getText()).actor.status.status, "stopped");
|
||||
const controlStopped = documentChanged(controlled, (text) =>
|
||||
text.includes('"status": "stopped"'),
|
||||
);
|
||||
// Graceful stop completes only when the scheduler advances.
|
||||
await vscode.commands.executeCommand("runa.tick", 10);
|
||||
await controlStopped;
|
||||
assert.equal(JSON.parse(controlled.getText()).actor.status.status, "stopped");
|
||||
const controlRestarted = documentChanged(
|
||||
controlled,
|
||||
(text) => JSON.parse(text).actor.incarnation === 2,
|
||||
|
|
@ -87,19 +77,31 @@ export async function run(): Promise<void> {
|
|||
snapshot = JSON.parse(controlled.getText()).actor;
|
||||
assert.equal(snapshot.incarnation, 2);
|
||||
assert.equal(snapshot.state.data, 12);
|
||||
const controlKilled = documentChanged(controlled, (text) =>
|
||||
text.includes('"status": "stopped"'),
|
||||
);
|
||||
await vscode.commands.executeCommand("runa.kill", selection);
|
||||
await controlKilled;
|
||||
const controlReloaded = documentChanged(
|
||||
controlled,
|
||||
(text) => JSON.parse(text).actor.incarnation === 3,
|
||||
);
|
||||
await vscode.commands.executeCommand("runa.restart", selection);
|
||||
await controlReloaded;
|
||||
await vscode.commands.executeCommand("runa.reload", selection);
|
||||
await vscode.commands.executeCommand("runa.tick", 10);
|
||||
await vscode.commands.executeCommand("runa.stop", selection);
|
||||
assert.equal(JSON.parse(controlled.getText()).removed, undefined);
|
||||
const controlStopped = documentChanged(controlled, (text) => JSON.parse(text).removed === true);
|
||||
await vscode.commands.executeCommand("runa.tick", 10);
|
||||
await controlStopped;
|
||||
assert.equal(
|
||||
(await vscode.workspace.fs.stat(controlled.uri)).permissions,
|
||||
vscode.FilePermission.Readonly,
|
||||
);
|
||||
const doomed = await vscode.commands.executeCommand<string>("runa.spawn", {
|
||||
path: "actors/counter",
|
||||
state: "0",
|
||||
});
|
||||
await vscode.commands.executeCommand("runa.inspect", { actor_ref: doomed });
|
||||
const killed = vscode.window.activeTextEditor?.document;
|
||||
assert.ok(killed);
|
||||
const controlKilled = documentChanged(killed, (text) => JSON.parse(text).removed === true);
|
||||
await vscode.commands.executeCommand("runa.kill", { actor_ref: doomed });
|
||||
await controlKilled;
|
||||
assert.equal(
|
||||
(await vscode.workspace.fs.stat(killed.uri)).permissions,
|
||||
vscode.FilePermission.Readonly,
|
||||
);
|
||||
const actorRef = await spawnCounter(image.address);
|
||||
await vscode.commands.executeCommand("runa.connect", image.address);
|
||||
await vscode.commands.executeCommand("runa.refresh");
|
||||
|
|
@ -210,7 +212,7 @@ export async function run(): Promise<void> {
|
|||
);
|
||||
await assert.rejects(async () => {
|
||||
await vscode.workspace.fs.writeFile(repaired.uri, Buffer.from(repaired.getText()));
|
||||
}, /paused, stopped, or failed/);
|
||||
}, /paused or failed/);
|
||||
console.log(
|
||||
"Runa extension host: spawn, send, system messages, ticks, inspection, automatic refresh, reconnect passed",
|
||||
);
|
||||
|
|
|
|||
|
|
@ -37,7 +37,6 @@ test(
|
|||
queued: [parseValue('"queued"')],
|
||||
});
|
||||
assert.deepEqual((await client.inspectActor(first)).state, parseValue("5"));
|
||||
await client.system(first, { kind: "kill" });
|
||||
await client.system(first, { kind: "restart" });
|
||||
await client.tick(100);
|
||||
const executable = resolve(__dirname, "../../../target/debug/runa-cli");
|
||||
|
|
@ -122,7 +121,7 @@ test(
|
|||
assert.equal(replCode, 0, replError);
|
||||
assert.doesNotMatch(replOutput, /^error:/im);
|
||||
assert.equal(replError, "");
|
||||
await client.system(first, { kind: "kill" });
|
||||
await client.system(first, { kind: "pause" });
|
||||
await run("set-state", first, "19");
|
||||
await run("set-messages", first, '{"reserved":[],"queued":["fixed"]}');
|
||||
assert.deepEqual((await client.inspectActor(first)).state, parseValue("19"));
|
||||
|
|
|
|||
|
|
@ -88,6 +88,8 @@ test(
|
|||
const actor = await run("spawn", "actors/copy", "-5");
|
||||
assert.match(actor, /^@[A-Za-z0-9_-]{21}$/);
|
||||
assert.match(await run("ps"), new RegExp(actor.slice(1)));
|
||||
await run("become", actor, "actors/counter");
|
||||
await run("tick", "1");
|
||||
assert.equal(await run("send", actor, '{"op":"add","value":7}'), "ok");
|
||||
await run("tick", "100", "32");
|
||||
assert.match(await run("inspect", actor), /Integer\(\s*2/);
|
||||
|
|
@ -96,11 +98,12 @@ test(
|
|||
}
|
||||
await run("become", actor, "actors/counter");
|
||||
await run("tick", "100");
|
||||
await run("stop", actor);
|
||||
await run("tick", "100");
|
||||
assert.match(await run("inspect", actor), /Stopped/);
|
||||
await run("restart", actor);
|
||||
await run("kill", actor);
|
||||
const stopped = await run("spawn", "actors/counter", "0");
|
||||
await run("stop", stopped);
|
||||
await run("tick", "100");
|
||||
assert.doesNotMatch(await run("ps"), new RegExp(stopped.slice(1)));
|
||||
assert.match(await run("events"), /Spawned/);
|
||||
for (const args of [
|
||||
["spawn", "actors/missing", "0"],
|
||||
|
|
|
|||
|
|
@ -1,10 +1,21 @@
|
|||
import { ChildProcess, spawn } from "node:child_process";
|
||||
import assert from "node:assert/strict";
|
||||
import { once } from "node:events";
|
||||
import { resolve } from "node:path";
|
||||
import { createInterface } from "node:readline";
|
||||
import { json } from "../test/fixtures";
|
||||
import WebSocket from "ws";
|
||||
import { decodeNotification, WEBSOCKET_SUBPROTOCOL } from "../src/protocol";
|
||||
import { ImageClient } from "../src/client";
|
||||
|
||||
// Control tests use a finite handler independent of the bundled self-messaging demo.
|
||||
export const COUNTER_SOURCE = `pub fn handle(ctx, state, sender, message) {
|
||||
if message.op == "send" {
|
||||
ctx.send(message.target, #{op: "add", value: message.value});
|
||||
state + 1
|
||||
} else if message.op == "add" { state + message.value }
|
||||
else { panic("unknown counter operation") }
|
||||
}`;
|
||||
|
||||
export async function startImage(): Promise<{
|
||||
address: string;
|
||||
|
|
@ -60,6 +71,18 @@ export async function startImage(): Promise<{
|
|||
await stop();
|
||||
throw error;
|
||||
}
|
||||
try {
|
||||
const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined));
|
||||
try {
|
||||
const source = await client.readSource("actors/counter");
|
||||
await client.writeSource("actors/counter", source.source_id, COUNTER_SOURCE);
|
||||
} finally {
|
||||
await client.close();
|
||||
}
|
||||
} catch (error) {
|
||||
await stop();
|
||||
throw error;
|
||||
}
|
||||
return { address, stop, diagnostics: () => diagnostics };
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -118,13 +118,13 @@
|
|||
},
|
||||
{
|
||||
"command": "runa.stop",
|
||||
"title": "Stop Actor (Graceful)",
|
||||
"title": "Stop Actor (Graceful Removal)",
|
||||
"category": "Runa",
|
||||
"enablement": "runa.connected"
|
||||
},
|
||||
{
|
||||
"command": "runa.kill",
|
||||
"title": "Kill Actor",
|
||||
"title": "Kill Actor (Remove Immediately)",
|
||||
"category": "Runa",
|
||||
"enablement": "runa.connected"
|
||||
},
|
||||
|
|
|
|||
|
|
@ -81,9 +81,7 @@ export function activate(context: vscode.ExtensionContext): { sourceTree: ImageS
|
|||
throw new Error("Inspection belongs to a disconnected session");
|
||||
const actor = await connection.client.inspectActor(actorRef);
|
||||
if (!canEditActor(actor))
|
||||
throw new Error(
|
||||
"Editing requires a paused, stopped, or failed actor without active execution",
|
||||
);
|
||||
throw new Error("Editing requires a paused or failed actor without active execution");
|
||||
connection.check();
|
||||
if (edit.state !== null || edit.messages !== null)
|
||||
await connection.client.editActor(actorRef, edit.state, edit.messages);
|
||||
|
|
@ -542,13 +540,8 @@ export function activate(context: vscode.ExtensionContext): { sourceTree: ImageS
|
|||
const actorRef = await chooseActor(selected);
|
||||
if (actorRef === undefined) return;
|
||||
const actor = await connection.client.inspectActor(actorRef);
|
||||
if (
|
||||
actor.execution !== null ||
|
||||
!["paused", "stopped", "failed"].includes(actor.status.status)
|
||||
) {
|
||||
throw new Error(
|
||||
"Editing requires a paused, stopped, or failed actor with no active invocation",
|
||||
);
|
||||
if (!canEditActor(actor)) {
|
||||
throw new Error("Editing requires a paused or failed actor with no active invocation");
|
||||
}
|
||||
const current =
|
||||
kind === "state"
|
||||
|
|
@ -599,7 +592,9 @@ export function activate(context: vscode.ExtensionContext): { sourceTree: ImageS
|
|||
await connection.client.system(actorRef, { kind });
|
||||
output.appendLine(
|
||||
kind === "stop"
|
||||
? `stop: @${actorRef} (graceful; completes when the scheduler advances)`
|
||||
? `stop: @${actorRef} (removes actor after graceful shutdown; advance the scheduler)`
|
||||
: kind === "kill"
|
||||
? `kill: @${actorRef} removed`
|
||||
: `${kind}: @${actorRef}`,
|
||||
);
|
||||
};
|
||||
|
|
|
|||
|
|
@ -370,9 +370,15 @@ export function formatSnapshot(
|
|||
endpoint: string,
|
||||
capturedAt: string,
|
||||
connected: boolean,
|
||||
removed = false,
|
||||
): string {
|
||||
const result = stringify(
|
||||
{ connection: { endpoint, connected }, captured_at: capturedAt, actor },
|
||||
{
|
||||
connection: { endpoint, connected },
|
||||
captured_at: capturedAt,
|
||||
...(removed ? { removed: true } : {}),
|
||||
actor,
|
||||
},
|
||||
null,
|
||||
2,
|
||||
);
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ import {
|
|||
} from "./protocol";
|
||||
|
||||
export function canEditActor(actor: ActorSnapshot): boolean {
|
||||
return actor.execution === null && ["paused", "stopped", "failed"].includes(actor.status.status);
|
||||
return actor.execution === null && ["paused", "failed"].includes(actor.status.status);
|
||||
}
|
||||
|
||||
export interface SnapshotEdit {
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ interface DocumentSnapshot {
|
|||
actor: ActorSnapshot;
|
||||
capturedAt: string;
|
||||
connected: boolean;
|
||||
removed: boolean;
|
||||
mtime: number;
|
||||
}
|
||||
|
||||
|
|
@ -35,6 +36,7 @@ export class SnapshotDocuments implements vscode.FileSystemProvider {
|
|||
actor,
|
||||
capturedAt: new Date().toISOString(),
|
||||
connected: true,
|
||||
removed: false,
|
||||
mtime: Math.max(Date.now(), (previous?.mtime ?? 0) + 1),
|
||||
});
|
||||
this.changed(uri);
|
||||
|
|
@ -43,9 +45,15 @@ export class SnapshotDocuments implements vscode.FileSystemProvider {
|
|||
|
||||
refresh(session: number, endpoint: string, actors: ActorSnapshot[]): void {
|
||||
const byRef = new Map(actors.map((actor) => [actor.actor_ref, actor]));
|
||||
for (const snapshot of this.documents.values()) {
|
||||
for (const [key, snapshot] of this.documents) {
|
||||
const actor = byRef.get(snapshot.actor.actor_ref);
|
||||
if (snapshot.session === session && actor) this.set(session, endpoint, actor);
|
||||
if (snapshot.session !== session) continue;
|
||||
if (actor) this.set(session, endpoint, actor);
|
||||
else if (!snapshot.removed) {
|
||||
snapshot.removed = true;
|
||||
snapshot.mtime++;
|
||||
this.changed(vscode.Uri.parse(key));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -75,6 +83,7 @@ export class SnapshotDocuments implements vscode.FileSystemProvider {
|
|||
snapshot.endpoint,
|
||||
snapshot.capturedAt,
|
||||
snapshot.connected,
|
||||
snapshot.removed,
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -88,7 +97,7 @@ export class SnapshotDocuments implements vscode.FileSystemProvider {
|
|||
mtime: snapshot.mtime,
|
||||
size: Buffer.byteLength(this.text(snapshot)),
|
||||
permissions:
|
||||
snapshot.connected && canEditActor(snapshot.actor)
|
||||
snapshot.connected && !snapshot.removed && canEditActor(snapshot.actor)
|
||||
? undefined
|
||||
: vscode.FilePermission.Readonly,
|
||||
};
|
||||
|
|
@ -105,9 +114,9 @@ export class SnapshotDocuments implements vscode.FileSystemProvider {
|
|||
|
||||
async writeFile(uri: vscode.Uri, content: Uint8Array): Promise<void> {
|
||||
const snapshot = this.snapshot(uri);
|
||||
if (!snapshot.connected || !canEditActor(snapshot.actor))
|
||||
if (!snapshot.connected || snapshot.removed || !canEditActor(snapshot.actor))
|
||||
throw vscode.FileSystemError.NoPermissions(
|
||||
"Actor must be paused, stopped, or failed with no active invocation",
|
||||
"Actor must be paused or failed with no active invocation, and not removed or disconnected",
|
||||
);
|
||||
const baseline = this.baselines.get(uri.toString());
|
||||
if (baseline === undefined)
|
||||
|
|
|
|||
Loading…
Reference in New Issue