diff --git a/CHANGELOG.md b/CHANGELOG.md index b4cf082..fb6ff7d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -56,6 +56,12 @@ All notable changes to this project will be documented in this file. ### Added +- Explicit versioned disk image save/restore through the CLI, REPL, protocol and + VS Code commands. Preserve source/history, actor identities, committed state, + mailbox envelopes and retained logs; reject unfinished handlers, pause runnable + actors on save and restore their scheduling eligibility with fresh incarnations. + Snapshot files are server-local and never overwrite existing destinations. + - Native VS Code debugger attachment to failed actors for retained-state, reserved-message and failure inspection, plus DAP exception details for both existing and newly observed failures. Debug controls never retry failed actors; diff --git a/Cargo.lock b/Cargo.lock index 38201d7..89b8f49 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -322,6 +322,12 @@ version = "3.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b5343afd4a8365a643ac588dab4cf234a190c7f6c88c9f6dd6ffe00837661b7" +[[package]] +name = "fastrand" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" + [[package]] name = "fd-lock" version = "4.0.4" @@ -885,6 +891,7 @@ dependencies = [ "runa-image-api", "runa-protocol", "serde_json", + "tempfile", "tokio", "tokio-tungstenite", ] @@ -894,6 +901,7 @@ name = "runa-actor" version = "0.0.0" dependencies = [ "runa-core", + "serde", ] [[package]] @@ -929,6 +937,7 @@ dependencies = [ "runa-core", "runa-rune", "rune", + "serde", ] [[package]] @@ -947,6 +956,7 @@ dependencies = [ "runa-code", "runa-core", "runa-runtime", + "serde", ] [[package]] @@ -975,6 +985,7 @@ dependencies = [ "runa-core", "runa-rune", "rune", + "serde", ] [[package]] @@ -1254,6 +1265,19 @@ version = "0.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "00c99c9cda412afe293a6b962af651b4594161ba88c1affe7ef66459ea040a06" +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom 0.3.4", + "once_cell", + "rustix", + "windows-sys 0.61.2", +] + [[package]] name = "termcolor" version = "1.4.1" diff --git a/README.md b/README.md index 3302bcc..1050005 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ experiment using Rune 0.14.2: script execution, native calls, async suspension, instruction budgets, stepping, and compiled-code sharing. It also runs a minimal single-worker actor demonstration with bounded mailboxes and explicit message effects. Source and compiled implementation identities are BLAKE3 content -addresses with an in-memory compile cache. Persistence, canonicalization, +addresses with an in-memory compile cache. Automatic crash recovery, 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 @@ -39,10 +39,12 @@ gutter breakpoints, the call-stack view, and bounded toolbar stepping. Continue resumes the actor; scheduler ticks remain explicit. The Variables view exposes a read-only committed actor snapshot with expandable state and mailbox. Watch and Debug Console accept snapshot paths such as `actor/committed_state`; -they never execute actor code. Frame locals, Rune expression evaluation, -and persistent storage remain unimplemented. +they never execute actor code. Frame locals and Rune expression evaluation +remain unimplemented. Failed actors can also be attached for exception details and retained-snapshot inspection without restarting them; post-mortem stacks are not retained. +Explicit disk image save/restore is implemented at handler boundaries, including +sources/history, actor identities, committed state, mailboxes and retained logs. ## Design goals @@ -80,6 +82,31 @@ Ctrl+D also exit the REPL and complete the WebSocket close handshake. Reconnect and use `ps` to recover their references. REPL bindings and command history are session-local and are not restored on reconnect. +### Save and restore an image + +```sh +cargo run -p runa-cli -- save /path/to/image.runa.json +cargo run -p runa-cli -- restore /path/to/image.runa.json +``` + +The REPL also accepts `save ` and `restore `. VS Code provides +**Runa: Save Image...** and **Runa: Restore Image...**. Paths refer to the +server's filesystem, not the client's. Saving requires a new destination file. +The server captures all actors at a handler boundary and leaves runnable actors +paused after a successful save. Saving is rejected without changing the image +if any handler is unfinished, including paused or awaiting execution. + +Restore replaces the whole image after validating and recompiling the snapshot. +Actors that were runnable when saved resume scheduling eligibility; previously +paused and failed actors stay that way. Execution still requires explicit ticks. +Actor references survive, but incarnations advance and old debug sessions end. +Restore also refuses to replace an image with unfinished handlers. +Only saved data survives server exit; this is not an automatic checkpoint, +durable per-message transaction log, or execution-stack serialization. +See [runtime persistence guarantees](docs/runtime.md#image-save-and-restore). + +### Commands and connections + Both executables use Clap for `--help`, `--version`, and argument validation. `runa` listens on `127.0.0.1:3737` without arguments. Set `RUNA_LISTEN` to change the server address and `RUNA_CONNECT` to change the REPL address; explicit diff --git a/crates/runa-actor/Cargo.toml b/crates/runa-actor/Cargo.toml index 5bd419c..925581b 100644 --- a/crates/runa-actor/Cargo.toml +++ b/crates/runa-actor/Cargo.toml @@ -5,3 +5,4 @@ edition = "2024" [dependencies] runa-core = { path = "../runa-core" } +serde = { version = "1", features = ["derive"] } diff --git a/crates/runa-actor/src/lib.rs b/crates/runa-actor/src/lib.rs index 0a7117d..2b8b360 100644 --- a/crates/runa-actor/src/lib.rs +++ b/crates/runa-actor/src/lib.rs @@ -8,7 +8,8 @@ pub struct MailboxId(pub u64); pub type Message = Value; -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct Envelope { pub sender: Option, pub message: Message, @@ -26,7 +27,7 @@ pub enum SystemMessage { Become(ImplementationId), } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub enum Status { Idle, Running, diff --git a/crates/runa-cli/src/main.rs b/crates/runa-cli/src/main.rs index 9518043..0d11455 100644 --- a/crates/runa-cli/src/main.rs +++ b/crates/runa-cli/src/main.rs @@ -81,6 +81,8 @@ impl ReplHelper { fn candidates(&self, line: &str, position: usize) -> Vec { const COMMANDS: &[&str] = &[ + "save", + "restore", "set-state", "set-messages", "log-level", @@ -224,6 +226,10 @@ struct Args { #[derive(Debug, clap::Subcommand)] enum CliCommand { + /// Save the image to a new server-local file and pause runnable actors. + Save { file: String }, + /// Replace the image from a server-local file; resume previously runnable actors. + Restore { file: String }, /// Replace committed state on a paused or failed actor without active execution. SetState { actor: String, @@ -313,6 +319,8 @@ impl CliCommand { }) }; Ok(match self { + Self::Save { file } => Command::SaveImage { file }, + Self::Restore { file } => Command::RestoreImage { file }, Self::SetState { actor, state: json } => { edit_request(actor_ref(&actor)?, Some(parse_value(&json, &state)?), None)? } @@ -601,6 +609,26 @@ fn execute_line( } } "help" => print_help(output), + "save" | "restore" => { + let [file] = args else { + writeln!(output, "error: usage: {command} ")?; + return Ok(true); + }; + let request = if command == "save" { + Command::SaveImage { + file: (*file).to_owned(), + } + } else { + Command::RestoreImage { + file: (*file).to_owned(), + } + }; + match dispatch(server, request) { + Ok(Reply::Acknowledged) => writeln!(output, "{command} complete"), + Ok(other) => unexpected_reply(output, other), + Err(error) => writeln!(output, "error: {error}"), + } + } "spawn" => spawn_command(output, line, server, state), "sources" if args.is_empty() => match dispatch(server, Command::ListSources) { Ok(Reply::Sources(files)) => { @@ -1175,6 +1203,8 @@ fn print_help(output: &mut W) -> io::Result<()> { log-level \n\ logs [-f|--follow] [actor...]\n\ sources\n\ + save (pauses runnable actors; unfinished handlers reject save)\n\ + restore (replaces image; resumes previously runnable actors)\n\ write \n\ commit \n\ history \n\ diff --git a/crates/runa-code/Cargo.toml b/crates/runa-code/Cargo.toml index 5278607..1b6380a 100644 --- a/crates/runa-code/Cargo.toml +++ b/crates/runa-code/Cargo.toml @@ -9,3 +9,4 @@ nanoid = "0.4" runa-core = { path = "../runa-core" } runa-rune = { path = "../runa-rune" } rune = "=0.14.2" +serde = { version = "1", features = ["derive"] } diff --git a/crates/runa-code/src/lib.rs b/crates/runa-code/src/lib.rs index bd963c6..3ea0130 100644 --- a/crates/runa-code/src/lib.rs +++ b/crates/runa-code/src/lib.rs @@ -15,7 +15,8 @@ const REVISION_ALPHABET: &[char] = &[ 'w', 'x', 'y', 'z', '2', '3', '4', '5', '6', '7', '8', '9', ]; -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct CompileFingerprint { pub rune_version: String, pub compiler_settings: String, @@ -71,7 +72,8 @@ impl CompiledArtifact { } } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct BehaviorVersion { pub revision: u64, pub implementation_id: ImplementationId, @@ -85,14 +87,16 @@ pub struct BehaviorSnapshot { pub history: Vec, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct SourceFile { pub path: String, pub source_id: SourceId, pub source: String, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct SourceCommit { pub path: String, pub revision: String, @@ -109,7 +113,100 @@ pub struct CodeStore { source_commits: BTreeMap>, } +#[derive(Clone, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CodeSnapshot { + fingerprint: CompileFingerprint, + artifacts: Vec<(ImplementationId, String)>, + behaviors: BTreeMap>, + working_sources: BTreeMap, + source_commits: BTreeMap>, +} + impl CodeStore { + pub fn snapshot(&self) -> ExperimentResult { + let fingerprint = CompileFingerprint::current(); + if self + .artifacts + .values() + .any(|artifact| artifact.fingerprint() != &fingerprint) + { + return Err("snapshot cannot include custom native modules".into()); + } + Ok(CodeSnapshot { + fingerprint, + artifacts: self + .artifacts + .iter() + .map(|(id, artifact)| (*id, artifact.original_source().to_owned())) + .collect(), + behaviors: self.behaviors.clone(), + working_sources: self.working_sources.clone(), + source_commits: self.source_commits.clone(), + }) + } + + pub fn restore(snapshot: CodeSnapshot) -> ExperimentResult { + if snapshot.fingerprint != CompileFingerprint::current() { + return Err("snapshot compiler fingerprint is incompatible with this runtime".into()); + } + let mut store = Self::default(); + for (id, source) in snapshot.artifacts { + if store.artifacts.contains_key(&id) { + return Err("duplicate snapshot implementation".into()); + } + if store.compile(&source)?.implementation_id != id { + return Err("snapshot implementation identity mismatch".into()); + } + } + for (path, file) in &snapshot.working_sources { + validate_source_path(path)?; + if &file.path != path || source_identity(&file.source) != file.source_id { + return Err("snapshot working source identity mismatch".into()); + } + } + let mut revisions = std::collections::BTreeSet::new(); + for (path, history) in &snapshot.source_commits { + if !snapshot.working_sources.contains_key(path) { + return Err("snapshot history has no working source".into()); + } + for commit in history { + if &commit.path != path + || source_identity(&commit.source) != commit.source_id + || commit.revision.len() != 12 + || !commit + .revision + .chars() + .all(|c| REVISION_ALPHABET.contains(&c)) + || !revisions.insert(&commit.revision) + { + return Err("invalid snapshot source commit".into()); + } + } + } + for (name, history) in &snapshot.behaviors { + validate_behavior_name(name)?; + if history.is_empty() { + return Err("snapshot behavior has no revisions".into()); + } + for (index, version) in history.iter().enumerate() { + if version.revision != index as u64 + 1 + || !store.artifacts.contains_key(&version.implementation_id) + { + return Err("invalid snapshot behavior revision".into()); + } + } + } + store.behaviors = snapshot.behaviors; + store.working_sources = snapshot.working_sources; + store.source_commits = snapshot.source_commits; + Ok(store) + } + + pub fn artifacts(&self) -> impl Iterator> { + self.artifacts.values() + } + pub fn check_source(&self, source: &str) -> ExperimentResult> { CompiledProgram::check(source) } diff --git a/crates/runa-core/src/lib.rs b/crates/runa-core/src/lib.rs index 0dd51b4..c29320c 100644 --- a/crates/runa-core/src/lib.rs +++ b/crates/runa-core/src/lib.rs @@ -16,7 +16,9 @@ pub use value::{ActorRef, MAX_VALUE_BYTES, MAX_VALUE_DEPTH, MAX_VALUE_NODES, Val macro_rules! digest_id { ($name:ident, $prefix:literal) => { - #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] + #[derive( + Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize, + )] pub struct $name(pub [u8; 32]); impl fmt::Debug for $name { diff --git a/crates/runa-image-api/Cargo.toml b/crates/runa-image-api/Cargo.toml index ec5f724..bc75714 100644 --- a/crates/runa-image-api/Cargo.toml +++ b/crates/runa-image-api/Cargo.toml @@ -8,3 +8,4 @@ runa-actor = { path = "../runa-actor" } runa-code = { path = "../runa-code" } runa-core = { path = "../runa-core" } runa-runtime = { path = "../runa-runtime" } +serde = { version = "1", features = ["derive"] } diff --git a/crates/runa-image-api/src/lib.rs b/crates/runa-image-api/src/lib.rs index bcb0607..83191d1 100644 --- a/crates/runa-image-api/src/lib.rs +++ b/crates/runa-image-api/src/lib.rs @@ -18,6 +18,18 @@ pub struct ImageApi { events: Vec, } +#[derive(Clone, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ImageSnapshot { + version: u16, + code: runa_code::CodeSnapshot, + runtime: runa_runtime::RuntimeSnapshot, + actor_paths: BTreeMap, + pending_paths: BTreeMap, + temporary: BTreeMap, + logs: Vec, +} + pub struct ExecutionInspection { pub actor_ref: ActorRef, pub incarnation: u64, @@ -36,6 +48,77 @@ impl Default for ImageApi { } impl ImageApi { + /// Capture at a handler boundary. Pausing is committed only after the caller + /// has durably written the returned snapshot. + pub fn snapshot(&self) -> RuntimeResult { + Ok(ImageSnapshot { + version: 1, + code: self.code.snapshot()?, + runtime: self.runtime.snapshot()?, + actor_paths: self.actor_paths.clone(), + pending_paths: self.pending_paths.clone(), + temporary: self.temporary.clone(), + logs: self.runtime.logs(&[], 0)?, + }) + } + + pub fn pause_after_save(&mut self) -> RuntimeResult<()> { + self.runtime.pause_for_snapshot() + } + + pub fn restore(&mut self, saved: ImageSnapshot) -> RuntimeResult<()> { + // Replacement never discards active drafts or shutdown work. + self.runtime.require_snapshot_boundary()?; + if saved.version != 1 { + return Err("unsupported image snapshot version".into()); + } + let code = CodeStore::restore(saved.code)?; + let mut runtime = Runtime::default(); + for artifact in code.artifacts() { + runtime.publish(Arc::clone(artifact))?; + } + runtime.restore_actors(saved.runtime, &self.runtime)?; + runtime.restore_logs(saved.logs)?; + if saved.actor_paths.len() != runtime.actors().len() { + return Err("snapshot actor source bindings are incomplete".into()); + } + for (actor, path) in &saved.actor_paths { + runtime.inspect(*actor)?; + runa_code::validate_source_path(path)?; + } + for (actor, (path, implementation)) in &saved.pending_paths { + runtime.inspect(*actor)?; + runa_code::validate_source_path(path)?; + if runtime.pending_implementation(*actor)? != Some(*implementation) { + return Err("snapshot pending behavior is unavailable".into()); + } + } + for (actor, file) in &saved.temporary { + let snapshot = runtime.inspect(*actor)?; + runa_code::validate_source_path(&file.path)?; + if saved.actor_paths.get(actor) != Some(&file.path) + || runa_code::source_identity(&file.source) != file.source_id + || code + .artifact( + runtime + .pending_implementation(*actor)? + .unwrap_or(snapshot.implementation), + ) + .is_none_or(|artifact| artifact.source_id != file.source_id) + { + return Err("invalid snapshot temporary behavior".into()); + } + } + runtime.transfer_log_sink(&mut self.runtime); + self.code = code; + self.runtime = runtime; + self.actor_paths = saved.actor_paths; + self.pending_paths = saved.pending_paths; + self.temporary = saved.temporary; + self.events.clear(); + Ok(()) + } + pub fn new() -> Self { Self { code: CodeStore::default(), diff --git a/crates/runa-protocol/src/lib.rs b/crates/runa-protocol/src/lib.rs index 0444ac6..d539fd1 100644 --- a/crates/runa-protocol/src/lib.rs +++ b/crates/runa-protocol/src/lib.rs @@ -51,6 +51,13 @@ impl Request { } match &self.command { + Command::SaveImage { file } | Command::RestoreImage { file } => { + if file.trim().is_empty() || file.len() > 4096 || file.contains('\0') { + return Err(invalid_request( + "snapshot file must be a nonempty server-local path of at most 4096 bytes", + )); + } + } Command::SetBreakpoints { actor_ref, implementation_id, @@ -209,6 +216,12 @@ impl Request { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "operation", content = "parameters", rename_all = "snake_case")] pub enum Command { + SaveImage { + file: String, + }, + RestoreImage { + file: String, + }, SetBreakpoints { actor_ref: String, implementation_id: String, diff --git a/crates/runa-rune/src/context.rs b/crates/runa-rune/src/context.rs index 6e4c067..cbe01d3 100644 --- a/crates/runa-rune/src/context.rs +++ b/crates/runa-rune/src/context.rs @@ -32,6 +32,40 @@ pub struct ActorLogs { } impl ActorLogs { + pub fn sequence(&self) -> u64 { + self.sequence + } + + pub fn restore_sequence(&mut self, sequence: u64) -> ExperimentResult<()> { + if !self.entries.is_empty() { + return Err("log sequence restore requires empty history".into()); + } + self.sequence = sequence; + Ok(()) + } + + pub fn transfer_sink(&mut self, previous: &mut Self) { + self.sink = previous.sink.take(); + self.sequence = self.sequence.max(previous.sequence); + } + + pub fn restore_entries(&mut self, entries: Vec) -> ExperimentResult<()> { + let mut last = 0; + for entry in entries { + if entry.sequence <= last || entry.sequence > self.sequence || entry.message.len() > 512 + { + return Err("invalid snapshot log entry".into()); + } + last = entry.sequence; + let history = self.entries.entry(entry.actor_ref).or_default(); + if history.len() >= 1000 { + return Err("snapshot actor log exceeds retention limit".into()); + } + history.push_back(entry); + } + Ok(()) + } + pub fn remove(&mut self, actor: ActorRef) { self.entries.remove(&actor); self.levels.remove(&actor); diff --git a/crates/runa-runtime/Cargo.toml b/crates/runa-runtime/Cargo.toml index 21f5379..d4d356b 100644 --- a/crates/runa-runtime/Cargo.toml +++ b/crates/runa-runtime/Cargo.toml @@ -9,3 +9,4 @@ runa-code = { path = "../runa-code" } runa-core = { path = "../runa-core" } runa-rune = { path = "../runa-rune" } rune = "=0.14.2" +serde = { version = "1", features = ["derive"] } diff --git a/crates/runa-runtime/src/lib.rs b/crates/runa-runtime/src/lib.rs index 88e80c2..8e537fc 100644 --- a/crates/runa-runtime/src/lib.rs +++ b/crates/runa-runtime/src/lib.rs @@ -60,7 +60,230 @@ pub struct Runtime { logs: Rc>, } +#[derive(Clone, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RuntimeSnapshot { + actors: Vec, + next_mailbox: u64, + log_sequence: u64, +} + +#[derive(Clone, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct SavedActor { + actor_ref: ActorRef, + implementation: ImplementationId, + incarnation: u64, + state: Value, + status: Status, + capacity: usize, + mailbox_id: u64, + mailbox: Vec, + reserved: Option, + pending_install: Option, + log_level: LogLevel, + stopping: bool, + shutdown: bool, +} + impl Runtime { + pub fn pending_implementation( + &self, + actor: ActorRef, + ) -> ExperimentResult> { + Ok(self + .actors + .get(&actor) + .ok_or("unknown actor")? + .pending_install) + } + + pub fn require_snapshot_boundary(&self) -> ExperimentResult<()> { + if self.actors.values().any(|actor| { + actor.active.is_some() + || ((actor.stopping || actor.shutdown) + && !matches!(actor.snapshot.status, Status::Failed(_))) + }) { + return Err("cannot snapshot image: an actor has an unfinished handler or shutdown; finish it first".into()); + } + Ok(()) + } + + pub fn snapshot(&self) -> ExperimentResult { + self.require_snapshot_boundary()?; + let actors = self + .actors + .values() + .map(|actor| SavedActor { + actor_ref: actor.snapshot.actor_ref, + implementation: actor.snapshot.implementation, + incarnation: actor.snapshot.incarnation, + state: actor.snapshot.state.clone(), + status: actor.snapshot.status.clone(), + capacity: actor.capacity.get(), + mailbox_id: actor.snapshot.mailbox.0, + mailbox: actor.mailbox.iter().cloned().collect(), + reserved: actor.snapshot.reserved.clone().map(|message| Envelope { + sender: actor.reserved_sender, + message, + }), + pending_install: actor.pending_install, + log_level: actor.snapshot.log_level, + stopping: actor.stopping, + shutdown: actor.shutdown, + }) + .collect(); + Ok(RuntimeSnapshot { + actors, + next_mailbox: self.next_mailbox, + log_sequence: self.logs.borrow().sequence(), + }) + } + + pub fn restore_actors( + &mut self, + saved: RuntimeSnapshot, + previous: &Self, + ) -> ExperimentResult<()> { + if !self.actors.is_empty() { + return Err("restore requires a fresh runtime".into()); + } + let mut staged = Self { + implementations: self.implementations.clone(), + ..Self::default() + }; + staged.restore_actors_in_place(saved.actors, previous)?; + if saved.next_mailbox < staged.next_mailbox { + return Err("snapshot mailbox counter precedes a saved mailbox".into()); + } + staged.next_mailbox = saved.next_mailbox.max(previous.next_mailbox); + staged + .logs + .borrow_mut() + .restore_sequence(saved.log_sequence)?; + *self = staged; + Ok(()) + } + + fn restore_actors_in_place( + &mut self, + saved: Vec, + previous: &Self, + ) -> ExperimentResult<()> { + let mut mailboxes = BTreeSet::new(); + for actor in saved { + if self.actors.contains_key(&actor.actor_ref) + || actor.mailbox_id == 0 + || !mailboxes.insert(actor.mailbox_id) + { + return Err("duplicate or invalid snapshot actor/mailbox identity".into()); + } + if actor.incarnation == 0 + || !matches!( + actor.status, + Status::Idle | Status::Paused | Status::Failed(_) + ) + || (actor.reserved.is_some() && !matches!(actor.status, Status::Failed(_))) + || ((actor.stopping || actor.shutdown) + && !matches!(actor.status, Status::Failed(_))) + { + return Err("invalid snapshot actor lifecycle".into()); + } + let capacity = + NonZeroUsize::new(actor.capacity).ok_or("invalid snapshot mailbox capacity")?; + if actor.mailbox.len() + usize::from(actor.reserved.is_some()) > actor.capacity { + return Err("snapshot mailbox exceeds capacity".into()); + } + actor.state.validate()?; + for envelope in actor.mailbox.iter().chain(actor.reserved.iter()) { + envelope.message.validate()?; + } + if !self.implementations.contains_key(&actor.implementation) + || actor + .pending_install + .is_some_and(|id| !self.implementations.contains_key(&id)) + { + return Err("snapshot actor implementation is unavailable".into()); + } + let incarnation = previous + .actors + .get(&actor.actor_ref) + .map_or(actor.incarnation, |old| { + old.snapshot.incarnation.max(actor.incarnation) + }) + .checked_add(1) + .ok_or("actor incarnation exhausted")?; + self.next_mailbox = self.next_mailbox.max(actor.mailbox_id); + self.logs + .borrow_mut() + .set_level(actor.actor_ref, actor.log_level); + let (reserved, reserved_sender) = actor.reserved.map_or((None, None), |envelope| { + (Some(envelope.message), envelope.sender) + }); + self.actors.insert( + actor.actor_ref, + Actor { + snapshot: ActorSnapshot { + log_level: actor.log_level, + actor_ref: actor.actor_ref, + mailbox: MailboxId(actor.mailbox_id), + implementation: actor.implementation, + incarnation, + state: actor.state, + status: actor.status.clone(), + execution: None, + queued: Vec::new(), + reserved, + }, + mailbox: actor.mailbox.into(), + capacity, + active: None, + context: ActorContext::with_logs(actor.actor_ref, Rc::clone(&self.logs)), + reserved_sender, + stopping: actor.stopping, + shutdown: actor.shutdown, + pending_install: actor.pending_install, + wake: Arc::new(WakeFlag(AtomicBool::new(true))), + paused_from: matches!(actor.status, Status::Paused).then_some(Status::Idle), + debugger: DebuggerSnapshot::default(), + breakpoint_ips: BTreeSet::new(), + skip_breakpoint: None, + }, + ); + } + Ok(()) + } + + pub fn pause_for_snapshot(&mut self) -> ExperimentResult<()> { + self.require_snapshot_boundary()?; + let running: Vec<_> = self + .actors + .values() + .filter(|actor| matches!(actor.snapshot.status, Status::Idle)) + .map(|actor| actor.snapshot.actor_ref) + .collect(); + for actor in running { + self.pause(actor)?; + } + Ok(()) + } + + pub fn transfer_log_sink(&mut self, previous: &mut Self) { + self.logs + .borrow_mut() + .transfer_sink(&mut previous.logs.borrow_mut()); + } + + pub fn restore_logs(&mut self, entries: Vec) -> ExperimentResult<()> { + if entries + .iter() + .any(|entry| !self.actors.contains_key(&entry.actor_ref)) + { + return Err("snapshot logs reference an unknown actor".into()); + } + self.logs.borrow_mut().restore_entries(entries) + } + pub fn publish( &mut self, artifact: Arc, @@ -854,6 +1077,126 @@ fn breakpoint_hit( #[cfg(test)] mod tests { use super::*; + + #[test] + fn snapshots_preserve_fifo_senders_capacity_and_failed_reservation() -> ExperimentResult<()> { + let mut old = Runtime::default(); + let code = publish( + &mut old, + "pub fn handle(ctx, state, sender, message) { sender }", + )?; + let failing = publish( + &mut old, + "pub fn handle(ctx, state, sender, message) { panic(\"retained\") }", + )?; + let actor = old.spawn(code, Value::Null, capacity(2))?; + let failed = old.spawn(failing, Value::Null, capacity(2))?; + old.send_from(Some(actor), failed, Value::Null)?; + old.tick(capacity(1000)); + old.send_from(Some(failed), actor, Value::Integer(1))?; + old.send_from(Some(actor), actor, Value::Integer(2))?; + let saved = old.snapshot()?; + let reservation = saved + .actors + .iter() + .find(|a| a.actor_ref == failed) + .unwrap() + .reserved + .clone(); + assert_eq!( + reservation, + Some(Envelope { + sender: Some(actor), + message: Value::Null + }) + ); + let mut restored = Runtime::default(); + for artifact in old.implementations.values() { + restored.publish(Arc::clone(artifact))?; + } + restored.restore_actors(saved, &old)?; + assert!(restored.send(actor, Value::Null).is_err()); + restored.tick(capacity(1000)); + assert_eq!(restored.inspect(actor)?.state, Value::ActorRef(failed)); + restored.tick(capacity(1000)); + assert_eq!(restored.inspect(actor)?.state, Value::ActorRef(actor)); + assert_eq!(restored.inspect(failed)?.reserved, Some(Value::Null)); + restored.restart(failed)?; + assert_eq!( + restored + .snapshot()? + .actors + .iter() + .find(|a| a.actor_ref == failed) + .unwrap() + .mailbox[0] + .sender, + Some(actor) + ); + let another = restored.spawn(code, Value::Null, capacity(1))?; + assert!(restored.inspect(another)?.mailbox.0 > restored.inspect(actor)?.mailbox.0); + Ok(()) + } + + #[test] + fn awaiting_and_shutdown_work_prevent_snapshots() -> ExperimentResult<()> { + let mut runtime = Runtime::default(); + let mut module = rune::Module::new(); + module + .function("wait", std::future::pending::) + .build()?; + let code = runtime.publish(CodeStore::default().compile_with_module( + "pub async fn handle(ctx, state, sender, message) { wait().await; state }", + "snapshot-wait:v1", + module, + )?)?; + let actor = runtime.spawn(code, Value::Null, capacity(1))?; + runtime.send(actor, Value::Null)?; + runtime.tick(capacity(1000)); + assert_eq!(runtime.inspect(actor)?.status, Status::Awaiting); + assert!(runtime.snapshot().is_err()); + runtime.pause(actor)?; + assert!(runtime.snapshot().is_err()); + runtime.kill(actor)?; + let actor = runtime.spawn(code, Value::Null, capacity(1))?; + runtime.stop(actor)?; + assert!(runtime.snapshot().is_err()); + Ok(()) + } + + #[test] + fn snapshots_preserve_counters_after_actor_removal_and_reject_partial_restores() + -> ExperimentResult<()> { + let mut old = Runtime::default(); + let code = publish( + &mut old, + "pub fn handle(ctx, state, sender, message) { ctx.info(\"one\"); state }", + )?; + let actor = old.spawn(code, Value::Null, capacity(1))?; + old.send(actor, Value::Null)?; + old.tick(capacity(1000)); + old.kill(actor)?; + let saved = old.snapshot()?; + assert!(saved.actors.is_empty()); + assert_eq!(saved.next_mailbox, 1); + assert_eq!(saved.log_sequence, 1); + let mut restored = Runtime::default(); + restored.publish(Arc::clone(&old.implementations[&code]))?; + restored.restore_actors(saved, &old)?; + restored.restore_logs(Vec::new())?; + let actor = restored.spawn(code, Value::Null, capacity(1))?; + assert_eq!(restored.inspect(actor)?.mailbox.0, 2); + restored.send(actor, Value::Null)?; + restored.tick(capacity(1000)); + assert_eq!(restored.logs(&[], 0)?[0].sequence, 2); + let mut bad = restored.snapshot()?; + bad.actors.push(bad.actors[0].clone()); + let mut empty = Runtime::default(); + empty.publish(Arc::clone(&old.implementations[&code]))?; + assert!(empty.restore_actors(bad, &old).is_err()); + assert!(empty.actors().is_empty()); + Ok(()) + } use runa_core::MAX_VALUE_NODES; #[test] diff --git a/crates/runa/Cargo.toml b/crates/runa/Cargo.toml index 8d4ba06..e837dd1 100644 --- a/crates/runa/Cargo.toml +++ b/crates/runa/Cargo.toml @@ -11,6 +11,7 @@ runa-code = { path = "../runa-code" } runa-image-api = { path = "../runa-image-api" } runa-protocol = { path = "../runa-protocol" } serde_json = "1" +tempfile = "3" tokio = { version = "1", features = ["macros", "net", "rt", "sync", "time"] } tokio-tungstenite = "0.28" diff --git a/crates/runa/src/lib.rs b/crates/runa/src/lib.rs index bdce0c4..bff0ae8 100644 --- a/crates/runa/src/lib.rs +++ b/crates/runa/src/lib.rs @@ -5,6 +5,7 @@ use std::net::{SocketAddr, TcpListener}; use std::num::NonZeroUsize; mod network; +mod storage; pub use network::serve_websocket; use runa_actor::{ActorRef, Event, Status, SystemMessage}; @@ -65,6 +66,12 @@ impl Server { fn dispatch_command(&mut self, command: Command) -> Result { match command { + Command::SaveImage { file } => storage::save(&mut self.image, &file) + .map(|()| Reply::Acknowledged) + .map_err(api_error), + Command::RestoreImage { file } => storage::restore(&mut self.image, &file) + .map(|()| Reply::Acknowledged) + .map_err(api_error), Command::SetBreakpoints { actor_ref, implementation_id, diff --git a/crates/runa/src/storage.rs b/crates/runa/src/storage.rs new file mode 100644 index 0000000..6d1cbc2 --- /dev/null +++ b/crates/runa/src/storage.rs @@ -0,0 +1,261 @@ +use std::fs::File; +use std::io::{Read, Write}; +use std::path::Path; + +use runa_image_api::{ImageApi, ImageSnapshot}; + +const MAX_SNAPSHOT_BYTES: u64 = 64 * 1024 * 1024; + +pub(crate) fn save(image: &mut ImageApi, file: &str) -> Result<(), Box> { + let snapshot = image.snapshot()?; + let bytes = serde_json::to_vec_pretty(&snapshot)?; + if bytes.len() as u64 > MAX_SNAPSHOT_BYTES { + return Err("snapshot exceeds 64 MiB limit".into()); + } + let path = Path::new(file); + let parent = path + .parent() + .filter(|p| !p.as_os_str().is_empty()) + .unwrap_or(Path::new(".")); + let mut temporary = tempfile::NamedTempFile::new_in(parent)?; + temporary.write_all(&bytes)?; + temporary.as_file().sync_all()?; + // Never overwrite an existing snapshot, including a symlink. + temporary.persist_noclobber(path)?; + File::open(parent)?.sync_all()?; + image.pause_after_save()?; + Ok(()) +} + +pub(crate) fn restore(image: &mut ImageApi, file: &str) -> Result<(), Box> { + let mut bytes = Vec::new(); + File::open(file)? + .take(MAX_SNAPSHOT_BYTES + 1) + .read_to_end(&mut bytes)?; + if bytes.len() as u64 > MAX_SNAPSHOT_BYTES { + return Err("snapshot exceeds 64 MiB limit".into()); + } + let snapshot: ImageSnapshot = serde_json::from_slice(&bytes)?; + image.restore(snapshot) +} + +#[cfg(test)] +mod tests { + use super::*; + use runa_actor::Status; + use runa_protocol::Value; + use std::num::NonZeroUsize; + + const SOURCE: &str = r#"pub fn handle(ctx, state, sender, message) { + ctx.info("snapshot probe"); + if message == -999 { panic("saved failure") } + state + message + }"#; + + #[test] + fn roundtrip_preserves_source_history_loaded_code_actor_data_and_status() + -> Result<(), Box> { + let dir = tempfile::tempdir()?; + let path = dir.path().join("image.json"); + let file = path.to_str().ok_or("non-UTF8 test path")?; + let mut image = ImageApi::new(); + image.publish_behavior("actors/probe", None, SOURCE)?; + let commit = image.commit_source("actors/probe")?; + let capacity = NonZeroUsize::new(2).unwrap(); + let running = image.spawn("actors/probe", Value::Integer(i64::MAX), capacity)?; + let paused = image.spawn("actors/probe", Value::ActorRef(running), capacity)?; + image.pause(paused)?; + let failed = image.spawn("actors/probe", Value::Integer(7), capacity)?; + image.send(failed, Value::Integer(-999))?; + image.tick(1, NonZeroUsize::new(1000).unwrap()); + assert!(matches!(image.inspect(failed)?.status, Status::Failed(_))); + image.send(running, Value::Integer(-1))?; + let current = image.source("actors/probe").unwrap().source_id; + image.write_source("actors/probe", Some(current), "invalid working source")?; + let before = image.inspect(running)?; + save(&mut image, file)?; + assert_eq!(image.inspect(running)?.status, Status::Paused); + let mut restored = ImageApi::new(); + restore(&mut restored, file)?; + let after = restored.inspect(running)?; + assert_eq!(after.actor_ref, before.actor_ref); + assert_eq!(after.mailbox, before.mailbox); + assert_eq!(after.implementation, before.implementation); + assert_eq!(after.incarnation, before.incarnation + 1); + assert_eq!(after.state, before.state); + assert_eq!(after.queued, before.queued); + assert_eq!(after.status, Status::Idle); + assert_eq!(restored.inspect(paused)?.status, Status::Paused); + assert_eq!(restored.inspect(paused)?.state, Value::ActorRef(running)); + assert_eq!( + restored.inspect(failed)?.reserved, + Some(Value::Integer(-999)) + ); + assert_eq!( + restored.inspect(failed)?.status, + image.inspect(failed)?.status + ); + assert_eq!( + restored.source("actors/probe"), + image.source("actors/probe") + ); + assert_eq!( + restored.source_commit("actors/probe", &commit.revision), + Some(&commit) + ); + assert_eq!(restored.behaviors(), image.behaviors()); + assert_eq!(restored.logs(&[], 0)?, image.logs(&[], 0)?); + restored.tick(1, NonZeroUsize::new(1000).unwrap()); + assert_eq!( + restored.inspect(running)?.state, + Value::Integer(i64::MAX - 1) + ); + Ok(()) + } + + #[test] + fn unfinished_invocations_and_failed_writes_leave_image_and_files_unchanged() + -> Result<(), Box> { + let dir = tempfile::tempdir()?; + let path = dir.path().join("snapshot.json"); + let file = path.to_str().ok_or("non-UTF8 test path")?; + let mut image = ImageApi::new(); + image.write_source("actors/probe", None, SOURCE)?; + let actor = image.spawn( + "actors/probe", + Value::Integer(0), + NonZeroUsize::new(2).unwrap(), + )?; + image.send(actor, Value::Integer(1))?; + image.tick(1, NonZeroUsize::new(1).unwrap()); + assert!( + save(&mut image, file) + .unwrap_err() + .to_string() + .contains("unfinished") + ); + assert!(!path.exists()); + assert!(image.inspect(actor)?.execution.is_some()); + image.pause(actor)?; + assert!(save(&mut image, file).is_err()); + image.resume(actor)?; + image.tick(1, NonZeroUsize::new(1000).unwrap()); + std::fs::write(&path, "existing snapshot")?; + assert!(save(&mut image, file).is_err()); + assert_eq!(std::fs::read_to_string(&path)?, "existing snapshot"); + assert_eq!(image.inspect(actor)?.status, Status::Idle); + let missing = dir.path().join("missing/image.json"); + assert!(save(&mut image, missing.to_str().unwrap()).is_err()); + assert_eq!(image.inspect(actor)?.status, Status::Idle); + Ok(()) + } + + #[test] + fn temporary_and_pending_behaviors_restore_without_overwriting_shared_source() + -> Result<(), Box> { + let dir = tempfile::tempdir()?; + let path = dir.path().join("temporary.json"); + let file = path.to_str().unwrap(); + let mut image = ImageApi::new(); + let source = image.write_source("actors/probe", None, SOURCE)?; + image.write_source( + "actors/next", + None, + "pub fn handle(ctx, state, sender, message) { 88 }", + )?; + let actor = image.spawn( + "actors/probe", + Value::Integer(0), + NonZeroUsize::new(2).unwrap(), + )?; + image.write_actor_behavior( + actor, + source.source_id.to_string(), + "pub fn handle(ctx, state, sender, message) { 77 }", + )?; + let pending = image.spawn( + "actors/probe", + Value::Integer(0), + NonZeroUsize::new(2).unwrap(), + )?; + image.r#become(pending, "actors/next")?; + image.send(actor, Value::Null)?; + image.send(pending, Value::Null)?; + save(&mut image, file)?; + restore(&mut image, file)?; + assert!(image.has_temporary_behavior(actor)); + assert_eq!(image.source("actors/probe"), Some(&source)); + assert_eq!(image.pending_actor_path(pending), Some("actors/next")); + image.tick(1, NonZeroUsize::new(1000).unwrap()); + assert_eq!(image.inspect(actor)?.state, Value::Integer(77)); + assert_eq!(image.inspect(pending)?.state, Value::Integer(88)); + assert_eq!(image.actor_path(pending), Some("actors/next")); + assert_eq!(image.pending_actor_path(pending), None); + assert!(image.has_temporary_behavior(actor)); + Ok(()) + } + + #[test] + fn snapshot_read_limit_is_checked_before_parsing() -> Result<(), Box> { + let dir = tempfile::tempdir()?; + let path = dir.path().join("large.json"); + let file = File::create(&path)?; + file.set_len(MAX_SNAPSHOT_BYTES + 1)?; + let mut image = ImageApi::new(); + assert!( + restore(&mut image, path.to_str().unwrap()) + .unwrap_err() + .to_string() + .contains("64 MiB") + ); + file.set_len(MAX_SNAPSHOT_BYTES)?; + assert!( + !restore(&mut image, path.to_str().unwrap()) + .unwrap_err() + .to_string() + .contains("64 MiB") + ); + Ok(()) + } + + #[test] + fn malformed_incompatible_and_corrupted_snapshots_do_not_replace_image() + -> Result<(), Box> { + let dir = tempfile::tempdir()?; + let path = dir.path().join("bad.json"); + let file = path.to_str().unwrap(); + let mut image = ImageApi::new(); + image.write_source("actors/probe", None, SOURCE)?; + let actor = image.spawn( + "actors/probe", + Value::Integer(10), + NonZeroUsize::new(2).unwrap(), + )?; + let saved = serde_json::to_value(image.snapshot()?)?; + for mutate in [ + |s: &mut serde_json::Value| s["version"] = 999.into(), + |s: &mut serde_json::Value| s["code"]["fingerprint"]["rune_version"] = "wrong".into(), + |s: &mut serde_json::Value| s["runtime"]["actors"][0]["capacity"] = 0.into(), + |s: &mut serde_json::Value| s["runtime"]["actors"][0]["incarnation"] = u64::MAX.into(), + |s: &mut serde_json::Value| s["runtime"]["actors"][0]["status"] = "Running".into(), + |s: &mut serde_json::Value| s["runtime"]["next_mailbox"] = 0.into(), + |s: &mut serde_json::Value| s["actor_paths"] = serde_json::json!({}), + |s: &mut serde_json::Value| { + s["code"]["working_sources"]["actors/probe"]["source"] = "tampered".into() + }, + |s: &mut serde_json::Value| s["code"]["artifacts"][0][1] = "tampered".into(), + ] { + let mut bad = saved.clone(); + mutate(&mut bad); + std::fs::write(&path, serde_json::to_vec(&bad)?)?; + assert!(restore(&mut image, file).is_err()); + assert_eq!(image.inspect(actor)?.state, Value::Integer(10)); + assert_eq!(image.inspect(actor)?.incarnation, 1); + assert_eq!(image.inspect(actor)?.status, Status::Idle); + } + std::fs::write(&path, "{")?; + assert!(restore(&mut image, file).is_err()); + assert_eq!(image.actors().len(), 1); + Ok(()) + } +} diff --git a/docs/architecture.md b/docs/architecture.md index 526baf8..4f6b09c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -312,6 +312,19 @@ Implement persistence incrementally: Start with SQLite or another simple durable store plus Git-friendly source export. Define snapshot consistency explicitly. Never assume arbitrary Rune stacks, closures, and native resources can be serialized and restored. Preserve mailbox and state transaction invariants across recovery. +The first concrete persistence adapter is an explicit, versioned JSON image file, +not a database or transaction journal. Save/restore is serialized by the image +server and requires no unfinished invocation or nonfailed shutdown work. Save +retains pre-pause scheduling status and leaves runnable actors paused; restore +recompiles and validates all saved artifacts/data before replacement, then +restores runnable actors' scheduling eligibility while preserving paused/failed +statuses. Incarnations advance and old debug sessions end. Sources/history, +registry revisions, pending/temporary behavior bindings, committed state, +mailbox envelopes/capacities and retained logs are included. See +[runtime guarantees](runtime.md#image-save-and-restore) for exact scope. +SQLite-backed per-transition durability, supervised recovery and execution-stack +restoration remain planned. + ## 11. Workspace Boundaries and Dependency Direction The Rust workspace uses **acyclic, inward-pointing dependencies**. The executable (`runa`) is the composition root. Higher-level services depend on lower-level contracts; low-level crates never import graphical, transport, or application code. @@ -528,7 +541,8 @@ state/mailbox edits and live log configuration are implemented. Sources support creation, guarded deletion, and confirmed rename/move with actor path updates. Actor-specific temporary behavior documents compile on save without changing shared source; reload clears them and a confirmed action can copy them back to -working source. These behaviors remain in-memory, not durable. +working source. These behaviors remain live in-memory and can be included in explicit disk +image snapshots; there is no automatic per-edit durability. Structured Rune compilation errors/warnings now flow through the image API and protocol into VS Code Problems for open sources and temporary behaviors, including unsaved text. Validation does not mutate the image; source positions diff --git a/docs/protocol.md b/docs/protocol.md index 6e3a47b..3f49772 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -51,6 +51,8 @@ behavior aliases. They exclude the `source/` root and `.rn` extension. | Operation | Parameters | Reply kind/data | | --- | --- | --- | +| `save_image` | `file`: new server-local destination | `acknowledged` | +| `restore_image` | `file`: server-local snapshot to replace image | `acknowledged` | | `list_sources` | none | `sources`: file array | | `read_source` | `path` | `source`: file | | `check_source` | `source` (including unsaved text) | `source_checked`: `{diagnostics}` | @@ -70,6 +72,20 @@ behavior aliases. They exclude the `source/` root and `.rn` extension. | `step_source` | `actor_ref`, `implementation_id`, `instruction_budget`, optional `mode` | `debug_step_completed`: `{reason}` | | `system` | `actor_ref`, `message` | `acknowledged` | +`save_image` and `restore_image` are additive v5 operations. File paths must be +nonempty, contain no NUL and fit 4096 UTF-8 bytes. Snapshot contents are stored +on the server, not transferred through the 1 MiB WebSocket envelope. +Save refuses existing destinations and unfinished handlers, and leaves runnable +actors paused on success. Restore validates and recompiles the entire snapshot +before replacing an image that has no unfinished handlers. Saved runnable actors +resume eligibility; previously paused/failed actors retain their status. +Incarnations advance to invalidate old debugger sessions. Both successful +operations emit the normal coalescible `image_changed` notification. +File/validation failures are explicit protocol errors; neither operation is +retried automatically. See [persistence semantics](runtime.md#image-save-and-restore). +These privileged commands grant server-local file access and must be used only +on a trusted machine, just like code installation. + A file is `{path, source_id, source}` and a commit is `{revision, source_id}`. Commits generate readable random revision strings and capture working text without compilation. Rollback restores working source diff --git a/docs/runtime.md b/docs/runtime.md index 1c2a611..7c237fe 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -447,11 +447,58 @@ messages are not processed until the reservation finishes. Transition tests cover malformed results, runtime failures, invalid targets, aggregate capacity, and successful message effects. +## Image save and restore + +`save ` and `restore ` are available in the CLI/REPL and through +the trusted image protocol. Files are server-local, versioned JSON with a +64 MiB limit. Saving writes a temporary file in the destination directory, +syncs its data, publishes it without overwriting an existing destination, and +syncs the directory. The server serializes snapshot operations with all image +commands; no scheduler tick runs during capture or disk I/O. Successful saves +leave previously runnable actors paused. Failed capture/write operations do not +pause them. A final filesystem sync failure is reported explicitly; the +destination may already exist in that case. + +Snapshots contain exact source text and implementation identities, the compiler +fingerprint, working sources (even invalid text), source commit history, +behavior registry revisions, actor/source bindings, pending installs and +temporary behavior text. They also retain actor references, mailbox IDs, +committed state, FIFO message envelopes including senders, reserved failed +messages, capacity, lifecycle status, log thresholds and retained log entries. +Actor references to removed actors remain valid data, not newly live actors. + +Any unfinished invocation prevents saving, including paused or awaiting Rune +frames; pending nonfailed graceful shutdown work also prevents saving. +Failures whose invocations have already been discarded can be saved. +No draft, buffered effects, native resource or execution stack is serialized. +Snapshot capture records scheduling status before save-induced pauses. + +Restore recompiles exact artifacts with the current default native modules, +checks content identities, validates histories and actor data, then atomically +replaces the image. Malformed/incompatible snapshots and compilation failures +leave the current image untouched. Compiler/ABI/target fingerprint changes and +custom native-module artifacts are unsupported. An image with unfinished +invocations/shutdown work cannot be replaced. + +Runnable actors restore as idle and eligible to run; previously paused/failed +actors retain those statuses. Restore does not tick the scheduler or retry +failed reservations. Incarnations advance beyond both the saved incarnation +and any currently live incarnation of the same reference. This invalidates +old debug sessions. Breakpoints, execution inspection, pending runtime events, +client subscriptions, REPL bindings and editor session state are not stored. +Existing log subscribers remain connected; restored historical entries are not +replayed as new events, and subsequent log sequences exceed the live server's +previous sequence and restored retained entries. + +This is explicit point-in-time persistence, not automatic crash recovery, +durable message acknowledgement, an outbox or stack restoration. Message/state +changes after the last successful save can be lost on process exit. + ## Limits -This is not yet a full image: no persistent behavior registry or mailboxes, -transactional outbox, supervisors, GUI, or automatic image bootstrap. The -behavior registry and implementation installation are process-local; +This is not yet a full image: no per-transition durable mailboxes, +transactional outbox, supervisors, GUI, or automatic image bootstrap. +Behavior registry and mailbox data survive process exit only in explicit saves; 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 failed or stopping actors, or removed references. diff --git a/vscode/README.md b/vscode/README.md index 4d07146..e0e78e1 100644 --- a/vscode/README.md +++ b/vscode/README.md @@ -61,6 +61,29 @@ machine running the extension host, not necessarily the desktop. ## Snapshot semantics +### Disk image save/restore + +**Runa: Save Image...** prompts for a new file on the server's filesystem. +It saves the whole image at a handler boundary and leaves runnable actors +paused. Unfinished handlers, including paused/awaiting execution, reject saving +without changing the image. Existing destination files are never overwritten. +Unsaved editor buffers are not part of the image snapshot; save desired source +or actor edits first. + +**Runa: Restore Image...** prompts for a server-local snapshot and requires +confirmation before replacing the entire image. Validation/compilation errors +leave the current image unchanged; unfinished handlers also prevent replacement. +Restored actors previously eligible to run resume eligibility automatically, +while previously paused and failed actors retain their status. Scheduler ticks +are still explicit. Open actor files refresh and old-incarnation debug sessions +terminate. Sources/history, temporary behaviors, state, mailbox data, log +configuration and retained logs are included; editor buffers, breakpoints and +client session state are not. These snapshots do not serialize execution stacks +or provide automatic crash recovery. See +[runtime guarantees](../docs/runtime.md#image-save-and-restore). + +### Live actor inspection + - The tree shows actor directories with references, lifecycle status, and queued/reserved counts. Tooltips include implementation ID, incarnation, and failure text. - Actor directories contain: diff --git a/vscode/integration/host.ts b/vscode/integration/host.ts index 70c2058..14b18d7 100644 --- a/vscode/integration/host.ts +++ b/vscode/integration/host.ts @@ -11,6 +11,9 @@ import { parseValue } from "../src/values"; import { ExecutionDocuments, frameSelection } from "../src/executionProvider"; import { DebugMessage } from "../src/debugger"; import { DebugVariable } from "../src/debugVariables"; +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; export async function run(): Promise { const image = await startImage(); @@ -50,6 +53,7 @@ export async function run(): Promise { await sourceDebugging(image.address); await nativeDebugging(image.address); await editActorFiles(image.address); + await snapshotImage(image.address); const spawned = await vscode.commands.executeCommand("runa.spawn", { path: "actors/counter", state: "5", @@ -250,6 +254,54 @@ export async function run(): Promise { } } +async function snapshotImage(address: string): Promise { + const directory = await mkdtemp(join(tmpdir(), "runa-host-snapshot-")); + const file = join(directory, "image.json"); + const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined)); + try { + const actor = await client.spawn("actors/counter", parseValue("23")); + await vscode.commands.executeCommand("runa.saveImage", file); + assert.equal(JSON.parse(await readFile(file, "utf8")).version, 1); + assert.equal((await client.inspectActor(actor)).status.status, "paused"); + await vscode.commands.executeCommand("runa.inspect", { actor_ref: actor }); + const document = vscode.window.activeTextEditor?.document; + assert.ok(document); + assert.equal( + await vscode.commands.executeCommand("runa.debugActor", { actor_ref: actor }), + true, + ); + const changed = documentChanged( + document, + (text) => + JSON.parse(text).actor.incarnation === 2 && JSON.parse(text).actor.state.data === 23, + ); + const terminated = new Promise((resolve, reject) => { + const timer = setTimeout(() => { + subscription.dispose(); + reject(new Error("Old-incarnation debugger did not terminate after image restore")); + }, 5000); + const subscription = vscode.debug.onDidTerminateDebugSession((session) => { + if (session.configuration.actorRef === actor) { + clearTimeout(timer); + subscription.dispose(); + resolve(); + } + }); + }); + await client.editActor(actor, parseValue("99"), null); + await client.restoreImage(file); + await Promise.all([changed, terminated]); + assert.equal((await client.inspectActor(actor)).status.status, "idle"); + await client.system(actor, { kind: "kill" }); + console.log( + "Runa image snapshots: Save Image command, restored actor files and old-incarnation debugger termination passed", + ); + } finally { + await client.close(); + await rm(directory, { recursive: true }); + } +} + async function inspectExecution(address: string): Promise { const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined)); try { diff --git a/vscode/integration/persistence.test.ts b/vscode/integration/persistence.test.ts new file mode 100644 index 0000000..5d2c9f0 --- /dev/null +++ b/vscode/integration/persistence.test.ts @@ -0,0 +1,176 @@ +import assert from "node:assert/strict"; +import { execFile } from "node:child_process"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { promisify } from "node:util"; +import test from "node:test"; +import { once } from "node:events"; +import WebSocket from "ws"; +import { ImageClient } from "../src/client"; +import { parseValue } from "../src/values"; +import { startImage } from "./server"; +import { WEBSOCKET_SUBPROTOCOL } from "../src/protocol"; + +test( + "disk image snapshots survive process replacement and resume only previously runnable actors", + { timeout: 15000 }, + async (t) => { + const directory = await mkdtemp(join(tmpdir(), "runa-snapshot-")); + const file = join(directory, "image.json"); + t.after(() => rm(directory, { recursive: true })); + const first = await startImage(); + const client = await ImageClient.connect(first.address, (error) => + assert.equal(error, undefined), + ); + t.after(async () => { + await client.close(); + await first.stop(); + }); + const actor = await client.spawn("actors/counter", parseValue("10")); + const paused = await client.spawn("actors/counter", parseValue("42")); + await client.system(paused, { kind: "pause" }); + await client.send(actor, parseValue('{"op":"add","value":2}')); + const commit = await client.commitSource("actors/counter"); + const run = promisify(execFile); + const executable = resolve( + __dirname, + "../../../target/debug", + process.platform === "win32" ? "runa-cli.exe" : "runa-cli", + ); + await run(executable, ["--connect", first.address, "save", file], { timeout: 5000 }); + assert.equal((await client.inspectActor(actor)).status.status, "paused"); + const snapshot = JSON.parse(await readFile(file, "utf8")); + assert.equal(snapshot.version, 1); + await client.close(); + await first.stop(); + const second = await startImage(); + const restored = await ImageClient.connect(second.address, (error) => + assert.equal(error, undefined), + ); + t.after(async () => { + await restored.close(); + await second.stop(); + }); + await run(executable, ["--connect", second.address, "restore", file], { timeout: 5000 }); + assert.equal((await restored.inspectActor(actor)).incarnation, 2n); + assert.equal((await restored.inspectActor(actor)).status.status, "idle"); + assert.equal((await restored.inspectActor(paused)).status.status, "paused"); + assert.deepEqual(await restored.sourceHistory("actors/counter"), [commit]); + await restored.tick(1, 1000); + assert.deepEqual((await restored.inspectActor(actor)).state, parseValue("12")); + assert.deepEqual((await restored.inspectActor(paused)).state, parseValue("42")); + await restored.restoreImage(file); + assert.equal((await restored.inspectActor(actor)).incarnation, 3n); + assert.deepEqual((await restored.inspectActor(actor)).state, parseValue("10")); + await assert.rejects(restored.saveImage(file), /exist/i); + assert.equal((await restored.inspectActor(actor)).status.status, "idle"); + }, +); + +test("snapshot errors preserve active handlers and the connected image", async (t) => { + const directory = await mkdtemp(join(tmpdir(), "runa-snapshot-errors-")); + t.after(() => rm(directory, { recursive: true })); + const image = await startImage(); + const client = await ImageClient.connect(image.address, (error) => + assert.equal(error, undefined), + ); + t.after(async () => { + await client.close(); + await image.stop(); + }); + const actor = await client.spawn("actors/counter", parseValue("5")); + await client.send(actor, parseValue('{"op":"add","value":1}')); + await client.tick(1, 1); + const before = await client.inspectActor(actor); + assert.ok(before.execution); + const file = join(directory, "busy.json"); + await assert.rejects(client.saveImage(file), /unfinished/); + assert.deepEqual(await client.inspectActor(actor), before); + await assert.rejects(readFile(file), /ENOENT/); + await client.system(actor, { kind: "pause" }); + await assert.rejects(client.saveImage(file), /unfinished/); + await writeFile(file, '{"version":999}'); + await assert.rejects(client.restoreImage(file)); + assert.equal((await client.inspectActor(actor)).status.status, "paused"); + await client.system(actor, { kind: "resume" }); + await client.tick(1, 1000); + assert.deepEqual((await client.inspectActor(actor)).state, parseValue("6")); + await assert.rejects(client.restoreImage(file)); + assert.deepEqual((await client.inspectActor(actor)).state, parseValue("6")); +}); + +test( + "restore keeps log subscriptions and advances sequences without replaying history", + { timeout: 10000 }, + async (t) => { + const directory = await mkdtemp(join(tmpdir(), "runa-snapshot-logs-")); + t.after(() => rm(directory, { recursive: true })); + const image = await startImage(); + const client = await ImageClient.connect(image.address, (error) => + assert.equal(error, undefined), + ); + let subscriptionSocket: WebSocket | undefined; + t.after(async () => { + if (subscriptionSocket && subscriptionSocket.readyState !== WebSocket.CLOSED) { + const closed = once(subscriptionSocket, "close"); + subscriptionSocket.close(); + await closed; + } + await client.close(); + await image.stop(); + }); + await client.writeSource( + "actors/logger", + null, + "pub fn handle(ctx, state, sender, message) { ctx.info(message); state }", + ); + const actor = await client.spawn("actors/logger", parseValue("0")); + await client.send(actor, parseValue('"saved"')); + await client.tick(1, 1000); + const file = join(directory, "logs.json"); + await client.saveImage(file); + await client.system(actor, { kind: "resume" }); + await client.send(actor, parseValue('"discarded"')); + await client.tick(1, 1000); + const socket = new WebSocket(`ws://${image.address}/`, WEBSOCKET_SUBPROTOCOL); + subscriptionSocket = socket; + await once(socket, "open"); + const subscribed = once(socket, "message"); + socket.send( + JSON.stringify({ + version: 5, + request_id: "logs", + command: { operation: "logs", parameters: { actors: [actor], follow: true, after: 0 } }, + }), + ); + await subscribed; + const events: { sequence: number; message: string }[] = []; + const received = new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error("Restored runtime did not deliver future log notification")), + 5000, + ); + socket.on("message", (data) => { + const notification = JSON.parse(data.toString()); + if (notification.event?.event === "log") { + events.push(notification.event.data); + clearTimeout(timer); + resolve(); + } + }); + }); + await client.restoreImage(file); + await client.send(actor, parseValue('"future"')); + await client.tick(1, 1000); + await received; + assert.deepEqual( + events.map((entry) => [entry.sequence, entry.message]), + [[3, "future"]], + ); + assert.deepEqual( + (await client.logs(actor)).map((entry) => entry.message), + ["saved", "future"], + ); + }, +); diff --git a/vscode/package.json b/vscode/package.json index 62fbc85..dfc4948 100644 --- a/vscode/package.json +++ b/vscode/package.json @@ -279,6 +279,18 @@ "category": "Runa", "enablement": "runa.connected" }, + { + "command": "runa.saveImage", + "title": "Save Image...", + "category": "Runa", + "enablement": "runa.connected" + }, + { + "command": "runa.restoreImage", + "title": "Restore Image...", + "category": "Runa", + "enablement": "runa.connected" + }, { "command": "runa.tick", "title": "Advance Scheduler", diff --git a/vscode/src/client.ts b/vscode/src/client.ts index e11edda..632d535 100644 --- a/vscode/src/client.ts +++ b/vscode/src/client.ts @@ -427,6 +427,23 @@ export class ImageClient { await this.request({ operation: "tick", parameters: { ticks, instruction_budget: budget } }); } + async saveImage(file: string): Promise { + await this.snapshotFile("save_image", file); + } + + async restoreImage(file: string): Promise { + await this.snapshotFile("restore_image", file); + } + + private async snapshotFile( + operation: "save_image" | "restore_image", + file: string, + ): Promise { + if (!file.trim() || Buffer.byteLength(file, "utf8") > 4096 || file.includes("\0")) + throw new Error("Snapshot file must be a nonempty server-local path of at most 4096 bytes"); + await this.request({ operation, parameters: { file } }); + } + private request(command: Command): Promise { const request = this.queue.then( () => diff --git a/vscode/src/extension.ts b/vscode/src/extension.ts index 005173f..dc88c63 100644 --- a/vscode/src/extension.ts +++ b/vscode/src/extension.ts @@ -815,6 +815,36 @@ export function activate(context: vscode.ExtensionContext): { } }); + const snapshotFile = async (restore: boolean, supplied?: string) => { + const connection = active(); + const file = + supplied ?? + (await vscode.window.showInputBox({ + title: restore ? "Restore Image" : "Save Image", + prompt: "File path on the Runa server's filesystem (not the editor filesystem)", + placeHolder: "/path/to/image.runa.json", + })); + if (file === undefined) return; + connection.check(); + if (restore) { + const answer = await vscode.window.showWarningMessage( + "Replace the entire image from this file? Unsaved image changes will be lost. Previously runnable actors will resume; unfinished handlers prevent restore.", + { modal: true }, + "Restore Image", + ); + if (answer !== "Restore Image") return; + connection.check(); + await connection.client.restoreImage(file); + } else { + await connection.client.saveImage(file); + } + connection.check(); + await refresh(); + output.appendLine( + `${restore ? "Restored" : "Saved"} image: ${file}${restore ? "" : "; runnable actors are now paused"}`, + ); + }; + const sourceUri = async (target?: SourceTarget): Promise => { const connection = active(); const path = await sourceTarget(target, connection.endpoint, "Select a source file"); @@ -1042,6 +1072,8 @@ export function activate(context: vscode.ExtensionContext): { }), command("runa.connect", connect), command("runa.disconnect", disconnect), + command("runa.saveImage", (file?: string) => snapshotFile(false, file)), + command("runa.restoreImage", (file?: string) => snapshotFile(true, file)), command("runa.refresh", refresh), command("runa.inspect", inspect), command("runa.inspectExecution", inspectExecution), diff --git a/vscode/src/protocol.ts b/vscode/src/protocol.ts index 745f115..2188ca4 100644 --- a/vscode/src/protocol.ts +++ b/vscode/src/protocol.ts @@ -128,6 +128,7 @@ export interface DebuggerSnapshot { } export type Command = + | { operation: "save_image" | "restore_image"; parameters: { file: string } } | { operation: "set_breakpoints"; parameters: { actor_ref: string; implementation_id: string; lines: number[] }; @@ -651,6 +652,8 @@ export function decodeResponse(line: string, requestId: string, command: Command } if ( (command.operation === "send" || + command.operation === "save_image" || + command.operation === "restore_image" || command.operation === "system" || command.operation === "set_log_level" || command.operation === "delete_source" || diff --git a/vscode/test/client.test.ts b/vscode/test/client.test.ts index 4ce5bf4..72d2373 100644 --- a/vscode/test/client.test.ts +++ b/vscode/test/client.test.ts @@ -25,6 +25,33 @@ test("only numeric loopback endpoints and valid ports are accepted", () => { } }); +test("image snapshot commands validate server-local paths and propagate failures", async (t) => { + const operations: string[] = []; + const address = await fakeImage(t, (request, socket) => { + operations.push(request.command.operation); + assert.equal(request.command.parameters?.file, "/server/image.json"); + if (request.command.operation === "restore_image") + socket.send( + json({ + version: 5, + request_id: request.request_id, + result: { Err: { code: "runtime_failure", message: "unfinished handler" } }, + }), + ); + else socket.send(response(request, { kind: "acknowledged" })); + }); + const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined)); + t.after(() => client.close()); + for (const file of ["", " ", "\0", "a".repeat(4097), "é".repeat(2049)]) { + await assert.rejects(client.saveImage(file), /server-local path/); + await assert.rejects(client.restoreImage(file), /server-local path/); + } + assert.deepEqual(operations, []); + await client.saveImage("/server/image.json"); + await assert.rejects(client.restoreImage("/server/image.json"), /unfinished handler/); + assert.deepEqual(operations, ["save_image", "restore_image"]); +}); + test("source step modes are sent explicitly and retain the default into behavior", async (t) => { const modes: unknown[] = []; const address = await fakeImage(t, (request, socket) => {