feat: add cooperative actor pause and resume

Add pause/resume lifecycle operations that retain active Rune invocations and restore their prior runnable or awaiting state. Expose both operations in the CLI and document the between-tick, non-preemptive guarantee.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Steve Biedermann 2026-11-02 12:35:29 +01:00
parent c857853cf5
commit 42a5884f45
6 changed files with 122 additions and 7 deletions

View File

@ -23,6 +23,8 @@ All notable changes to this project will be documented in this file.
boundaries. boundaries.
- CLI commands for named behavior publishing, history, rollback, and - CLI commands for named behavior publishing, history, rollback, and
installation, plus `mise run run` to launch the local executable. 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. - Execution-model decision and Phase 1 runtime documentation.
- Lifecycle tests proving failed-message retry from committed state and - Lifecycle tests proving failed-message retry from committed state and
stop/restart preservation of reserved-first FIFO order. stop/restart preservation of reserved-first FIFO order.

View File

@ -61,9 +61,10 @@ runa> events
The sender commits state `1` and sends a message to actor 2, which commits 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, state `15`. The CLI also supports publishing Rune source files as named,
versioned behaviors, viewing history, rolling back a behavior reference, and versioned behaviors, viewing history, rolling back a behavior reference,
installing the current version on an idle actor. Use `help` in the session for installing the current version on an idle actor, and cooperatively pausing and
the complete command list. 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 See [the feasibility report](docs/rune-feasibility.md) for verified Rune
capabilities and limitations, and [the runtime guide](docs/runtime.md) for capabilities and limitations, and [the runtime guide](docs/runtime.md) for
the runtime contract. the runtime contract.

View File

@ -19,6 +19,7 @@ pub enum Status {
Idle, Idle,
Running, Running,
Awaiting, Awaiting,
Paused,
Stopped, Stopped,
Failed(String), Failed(String),
} }
@ -42,5 +43,7 @@ pub enum Event {
Failed(ActorId, String), Failed(ActorId, String),
Stopped(ActorId), Stopped(ActorId),
Restarted(ActorId), Restarted(ActorId),
Paused(ActorId),
Resumed(ActorId),
BehaviorInstalled(ActorId, ImplementationId), BehaviorInstalled(ActorId, ImplementationId),
} }

View File

@ -31,6 +31,7 @@ struct Actor {
capacity: NonZeroUsize, capacity: NonZeroUsize,
active: Option<Invocation<WireTransition>>, active: Option<Invocation<WireTransition>>,
wake: Arc<WakeFlag>, wake: Arc<WakeFlag>,
paused_from: Option<Status>,
} }
#[derive(Default)] #[derive(Default)]
@ -88,6 +89,7 @@ impl Runtime {
capacity, capacity,
active: None, active: None,
wake: Arc::new(WakeFlag(AtomicBool::new(true))), wake: Arc::new(WakeFlag(AtomicBool::new(true))),
paused_from: None,
}, },
); );
self.events.push(Event::Spawned(id)); self.events.push(Event::Spawned(id));
@ -129,11 +131,36 @@ impl Runtime {
pub fn stop(&mut self, id: ActorId) -> ExperimentResult<()> { pub fn stop(&mut self, id: ActorId) -> ExperimentResult<()> {
let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
actor.active = None; actor.active = None;
actor.paused_from = None;
actor.snapshot.status = Status::Stopped; actor.snapshot.status = Status::Stopped;
self.events.push(Event::Stopped(id)); self.events.push(Event::Stopped(id));
Ok(()) 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<()> { pub fn restart(&mut self, id: ActorId) -> ExperimentResult<()> {
let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
if !matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { if !matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) {
@ -148,6 +175,7 @@ impl Runtime {
actor.mailbox.push_front(message); actor.mailbox.push_front(message);
} }
actor.snapshot.status = Status::Idle; actor.snapshot.status = Status::Idle;
actor.paused_from = None;
self.events.push(Event::Restarted(id)); self.events.push(Event::Restarted(id));
Ok(()) Ok(())
} }
@ -225,7 +253,10 @@ impl Runtime {
let ids: Vec<_> = self.actors.keys().copied().collect(); let ids: Vec<_> = self.actors.keys().copied().collect();
for id in ids { for id in ids {
let actor = self.actors.get_mut(&id).expect("scheduled actor exists"); 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; continue;
} }
if actor.active.is_none() { if actor.active.is_none() {
@ -435,6 +466,8 @@ mod tests {
runtime.tick(capacity(5)); runtime.tick(capacity(5));
} }
assert_eq!(gate.polls.load(Ordering::SeqCst), polls); 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.ready.store(true, Ordering::SeqCst);
gate.waker gate.waker
.lock() .lock()
@ -442,6 +475,12 @@ mod tests {
.take() .take()
.expect("registered wake") .expect("registered wake")
.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 { for _ in 0..100 {
runtime.tick(capacity(5)); runtime.tick(capacity(5));
} }
@ -469,6 +508,54 @@ mod tests {
Ok(()) 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] #[test]
fn mailbox_capacity_and_runtime_failures_are_explicit() -> ExperimentResult<()> { fn mailbox_capacity_and_runtime_failures_are_explicit() -> ExperimentResult<()> {
let mut runtime = Runtime::default(); let mut runtime = Runtime::default();

View File

@ -126,6 +126,15 @@ fn run_repl<R: BufRead, W: Write>(
|runtime, id| runtime.restart(id), |runtime, id| runtime.restart(id),
runtime, 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" if args.is_empty() => break,
"quit" | "exit" => writeln!(output, "error: usage: quit"), "quit" | "exit" => writeln!(output, "error: usage: quit"),
_ => writeln!(output, "error: unknown command `{command}`; type `help`"), _ => writeln!(output, "error: unknown command `{command}`; type `help`"),
@ -381,6 +390,8 @@ fn print_help<W: Write>(output: &mut W) -> io::Result<()> {
events\n\ events\n\
stop <actor-id>\n\ stop <actor-id>\n\
restart <actor-id>\n\ restart <actor-id>\n\
pause <actor-id>\n\
resume <actor-id>\n\
help\n\ help\n\
quit" quit"
) )
@ -470,7 +481,7 @@ mod tests {
"/../../runa-std/actors/counter.rn" "/../../runa-std/actors/counter.rn"
); );
let input = io::Cursor::new(format!( 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(); let mut output = Vec::new();
run_repl( run_repl(
@ -487,6 +498,8 @@ mod tests {
assert!(output.contains("rolled back `counter`")); assert!(output.contains("rolled back `counter`"));
assert!(output.contains("spawned ActorId(1) with `counter` revision 3")); assert!(output.contains("spawned ActorId(1) with `counter` revision 3"));
assert!(output.contains("installed `counter` revision 3 on actor 1")); 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!(code.behavior(DEFAULT_BEHAVIOR).unwrap().revision, 3);
assert_eq!(runtime.inspect(ActorId(1))?.implementation, implementation); assert_eq!(runtime.inspect(ActorId(1))?.implementation, implementation);
Ok(()) Ok(())

View File

@ -8,8 +8,8 @@ mise run run
The executable loads [counter.rn](../runa-std/actors/counter.rn) as the default The executable loads [counter.rn](../runa-std/actors/counter.rn) as the default
behavior. Commands include `spawn`, `send`, `tick`, `inspect`, `actors`, behavior. Commands include `spawn`, `send`, `tick`, `inspect`, `actors`,
`events`, `stop`, `restart`, `behaviors`, `publish`, `history`, `rollback`, and `events`, `stop`, `restart`, `pause`, `resume`, `behaviors`, `publish`,
`install`; enter `help` to see usage. `publish` reads a Rune source file and `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 takes `new` for a first revision or the expected current revision for an
update. `rollback` requires the expected current revision and a target update. `rollback` requires the expected current revision and a target
historical revision. `install` selects a named behavior's current 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 attempts during execution fail explicitly so the caller can retry at the next
handler boundary. 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 The source identity hashes the exact original UTF-8 bytes with the versioned
`runa-source:v1` domain. Whitespace/comment canonicalization is intentionally `runa-source:v1` domain. Whitespace/comment canonicalization is intentionally
not attempted; textually different source has a different identity. The 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` | | 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` | | 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` | | 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 Mailbox capacity counts queued plus reserved messages. Before applying a
transition, the runtime validates all effects, including aggregate sends to a transition, the runtime validates all effects, including aggregate sends to a