diff --git a/CHANGELOG.md b/CHANGELOG.md index bb099f0..e9f5c6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,8 @@ All notable changes to this project will be documented in this file. boundaries. - CLI commands for named behavior publishing, history, rollback, and installation, plus `mise run run` to launch the local executable. +- Cooperative actor pause/resume between scheduler ticks, preserving active + invocations and emitting lifecycle events. - Execution-model decision and Phase 1 runtime documentation. - Lifecycle tests proving failed-message retry from committed state and stop/restart preservation of reserved-first FIFO order. diff --git a/README.md b/README.md index 1fdaacd..833a6c8 100644 --- a/README.md +++ b/README.md @@ -61,9 +61,10 @@ runa> events The sender commits state `1` and sends a message to actor 2, which commits state `15`. The CLI also supports publishing Rune source files as named, -versioned behaviors, viewing history, rolling back a behavior reference, and -installing the current version on an idle actor. Use `help` in the session for -the complete command list. +versioned behaviors, viewing history, rolling back a behavior reference, +installing the current version on an idle actor, and cooperatively pausing and +resuming actors between scheduler ticks. Use `help` in the session for the +complete command list. See [the feasibility report](docs/rune-feasibility.md) for verified Rune capabilities and limitations, and [the runtime guide](docs/runtime.md) for the runtime contract. diff --git a/crates/runa-actor/src/lib.rs b/crates/runa-actor/src/lib.rs index af70e82..fcc7761 100644 --- a/crates/runa-actor/src/lib.rs +++ b/crates/runa-actor/src/lib.rs @@ -19,6 +19,7 @@ pub enum Status { Idle, Running, Awaiting, + Paused, Stopped, Failed(String), } @@ -42,5 +43,7 @@ pub enum Event { Failed(ActorId, String), Stopped(ActorId), Restarted(ActorId), + Paused(ActorId), + Resumed(ActorId), BehaviorInstalled(ActorId, ImplementationId), } diff --git a/crates/runa-runtime/src/lib.rs b/crates/runa-runtime/src/lib.rs index 51e5c47..aa002f3 100644 --- a/crates/runa-runtime/src/lib.rs +++ b/crates/runa-runtime/src/lib.rs @@ -31,6 +31,7 @@ struct Actor { capacity: NonZeroUsize, active: Option>, wake: Arc, + paused_from: Option, } #[derive(Default)] @@ -88,6 +89,7 @@ impl Runtime { capacity, active: None, wake: Arc::new(WakeFlag(AtomicBool::new(true))), + paused_from: None, }, ); self.events.push(Event::Spawned(id)); @@ -129,11 +131,36 @@ impl Runtime { pub fn stop(&mut self, id: ActorId) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; actor.active = None; + actor.paused_from = None; actor.snapshot.status = Status::Stopped; self.events.push(Event::Stopped(id)); Ok(()) } + /// Pause scheduling between worker ticks without discarding an invocation. + pub fn pause(&mut self, id: ActorId) -> ExperimentResult<()> { + let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; + if matches!( + actor.snapshot.status, + Status::Paused | Status::Stopped | Status::Failed(_) + ) { + return Err("actor cannot be paused in its current status".into()); + } + actor.paused_from = Some(actor.snapshot.status.clone()); + actor.snapshot.status = Status::Paused; + self.events.push(Event::Paused(id)); + Ok(()) + } + + /// Resume from the exact runnable or awaiting state saved by `pause`. + pub fn resume(&mut self, id: ActorId) -> ExperimentResult<()> { + let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; + let status = actor.paused_from.take().ok_or("actor is not paused")?; + actor.snapshot.status = status; + self.events.push(Event::Resumed(id)); + Ok(()) + } + pub fn restart(&mut self, id: ActorId) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if !matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { @@ -148,6 +175,7 @@ impl Runtime { actor.mailbox.push_front(message); } actor.snapshot.status = Status::Idle; + actor.paused_from = None; self.events.push(Event::Restarted(id)); Ok(()) } @@ -225,7 +253,10 @@ impl Runtime { let ids: Vec<_> = self.actors.keys().copied().collect(); for id in ids { let actor = self.actors.get_mut(&id).expect("scheduled actor exists"); - if matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { + if matches!( + actor.snapshot.status, + Status::Paused | Status::Stopped | Status::Failed(_) + ) { continue; } if actor.active.is_none() { @@ -435,6 +466,8 @@ mod tests { runtime.tick(capacity(5)); } assert_eq!(gate.polls.load(Ordering::SeqCst), polls); + runtime.pause(actor)?; + assert_eq!(runtime.inspect(actor)?.status, Status::Paused); gate.ready.store(true, Ordering::SeqCst); gate.waker .lock() @@ -442,6 +475,12 @@ mod tests { .take() .expect("registered wake") .wake(); + for _ in 0..10 { + runtime.tick(capacity(5)); + } + assert_eq!(gate.polls.load(Ordering::SeqCst), polls); + assert_eq!(runtime.inspect(actor)?.status, Status::Paused); + runtime.resume(actor)?; for _ in 0..100 { runtime.tick(capacity(5)); } @@ -469,6 +508,54 @@ mod tests { Ok(()) } + #[test] + fn pause_preserves_active_invocation_and_resume_continues_it() -> ExperimentResult<()> { + let mut runtime = Runtime::default(); + let implementation = publish( + &mut runtime, + "pub fn handle(state, kind, value) { (state + value, []) }", + )?; + let actor = runtime.spawn(implementation, 10, capacity(2))?; + runtime.send(actor, Message { kind: 0, value: 7 })?; + runtime.tick(capacity(1)); + let before_pause = runtime.inspect(actor)?; + assert_eq!(before_pause.status, Status::Running); + assert_eq!(before_pause.state, 10); + assert!(before_pause.reserved.is_some()); + + runtime.pause(actor)?; + assert!(runtime.pause(actor).is_err()); + for _ in 0..10 { + runtime.tick(capacity(10)); + } + let paused = runtime.inspect(actor)?; + assert_eq!(paused.status, Status::Paused); + assert_eq!(paused.state, before_pause.state); + assert_eq!(paused.reserved, before_pause.reserved); + + runtime.resume(actor)?; + assert!(runtime.resume(actor).is_err()); + for _ in 0..100 { + runtime.tick(capacity(10)); + } + let completed = runtime.inspect(actor)?; + assert_eq!(completed.status, Status::Idle); + assert_eq!(completed.state, 17); + assert!(completed.reserved.is_none()); + let events = runtime.drain_events(); + assert!( + events + .iter() + .any(|event| matches!(event, Event::Paused(id) if *id == actor)) + ); + assert!( + events + .iter() + .any(|event| matches!(event, Event::Resumed(id) if *id == actor)) + ); + Ok(()) + } + #[test] fn mailbox_capacity_and_runtime_failures_are_explicit() -> ExperimentResult<()> { let mut runtime = Runtime::default(); diff --git a/crates/runa/src/main.rs b/crates/runa/src/main.rs index eb02b5b..fd85d6c 100644 --- a/crates/runa/src/main.rs +++ b/crates/runa/src/main.rs @@ -126,6 +126,15 @@ fn run_repl( |runtime, id| runtime.restart(id), runtime, ), + "pause" => { + lifecycle_command(&mut output, &args, |runtime, id| runtime.pause(id), runtime) + } + "resume" => lifecycle_command( + &mut output, + &args, + |runtime, id| runtime.resume(id), + runtime, + ), "quit" | "exit" if args.is_empty() => break, "quit" | "exit" => writeln!(output, "error: usage: quit"), _ => writeln!(output, "error: unknown command `{command}`; type `help`"), @@ -381,6 +390,8 @@ fn print_help(output: &mut W) -> io::Result<()> { events\n\ stop \n\ restart \n\ + pause \n\ + resume \n\ help\n\ quit" ) @@ -470,7 +481,7 @@ mod tests { "/../../runa-std/actors/counter.rn" ); let input = io::Cursor::new(format!( - "behaviors\npublish counter 1 {source_path}\nhistory counter\nrollback counter 2 1\nspawn 0 counter\ninstall 1 counter\ninspect 1\nquit\n" + "behaviors\npublish counter 1 {source_path}\nhistory counter\nrollback counter 2 1\nspawn 0 counter\ninstall 1 counter\npause 1\nresume 1\ninspect 1\nevents\nquit\n" )); let mut output = Vec::new(); run_repl( @@ -487,6 +498,8 @@ mod tests { assert!(output.contains("rolled back `counter`")); assert!(output.contains("spawned ActorId(1) with `counter` revision 3")); assert!(output.contains("installed `counter` revision 3 on actor 1")); + assert!(output.contains("Paused(")); + assert!(output.contains("Resumed(")); assert_eq!(code.behavior(DEFAULT_BEHAVIOR).unwrap().revision, 3); assert_eq!(runtime.inspect(ActorId(1))?.implementation, implementation); Ok(()) diff --git a/docs/runtime.md b/docs/runtime.md index adfaf1c..92362cb 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -8,8 +8,8 @@ mise run run The executable loads [counter.rn](../runa-std/actors/counter.rn) as the default behavior. Commands include `spawn`, `send`, `tick`, `inspect`, `actors`, -`events`, `stop`, `restart`, `behaviors`, `publish`, `history`, `rollback`, and -`install`; enter `help` to see usage. `publish` reads a Rune source file and +`events`, `stop`, `restart`, `pause`, `resume`, `behaviors`, `publish`, +`history`, `rollback`, and `install`; enter `help` to see usage. `publish` reads a Rune source file and takes `new` for a first revision or the expected current revision for an update. `rollback` requires the expected current revision and a target historical revision. `install` selects a named behavior's current @@ -70,6 +70,13 @@ or awaiting handlers keep their original implementation, and installation attempts during execution fail explicitly so the caller can retry at the next handler boundary. +`pause` prevents an actor from being scheduled on subsequent worker ticks and +retains its active invocation, reserved message, and prior `Running`, `Awaiting`, +or `Idle` status. `resume` restores that status; an awaiting invocation resumes +only if its future has been woken. These operations take effect between calls +to `tick` and cannot interrupt a currently executing Rune slice or native +callback. They provide cooperative scheduling control, not preemption. + The source identity hashes the exact original UTF-8 bytes with the versioned `runa-source:v1` domain. Whitespace/comment canonicalization is intentionally not attempted; textually different source has a different identity. The @@ -135,6 +142,8 @@ resolves it; there is no automatic poison-message policy or supervision. | Any target is unknown, stopped, failed, or over capacity | Last committed state retained | Retained | No sends from the transition enqueue | `Failed` | | Stop during an invocation | Last committed state retained | Retained | None from the incomplete transition | `Stopped` | | Restart after stop or failure | 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 | Mailbox capacity counts queued plus reserved messages. Before applying a transition, the runtime validates all effects, including aggregate sends to a