feat: add explicit handler cancellation
Expose cancellation for active and paused handler invocations. Cancellation drops the invocation, preserves committed state, moves reserved input to the mailbox front for retry, and emits an event. Document that native side effects are not rolled back; test cancelling pending async work and retrying an input under a newly installed behavior. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
parent
bdc34d037c
commit
1916814c2e
|
|
@ -27,6 +27,8 @@ All notable changes to this project will be documented in this file.
|
||||||
invocations and emitting lifecycle events.
|
invocations and emitting lifecycle events.
|
||||||
- Explicit safe-point state migration during implementation installation, with
|
- Explicit safe-point state migration during implementation installation, with
|
||||||
atomic failure behavior and migration events.
|
atomic failure behavior and migration events.
|
||||||
|
- Explicit invocation cancellation that requeues the reserved message for
|
||||||
|
retry, exposed in the CLI and documented with native side-effect caveats.
|
||||||
- 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.
|
||||||
|
|
|
||||||
|
|
@ -63,10 +63,10 @@ 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,
|
versioned behaviors, viewing history, rolling back a behavior reference,
|
||||||
installing the current version on an idle actor, and cooperatively pausing and
|
installing the current version on an idle actor, and cooperatively pausing and
|
||||||
resuming actors between scheduler ticks. Runtime users can also supply an
|
resuming or cancelling actor invocations between scheduler ticks. Runtime
|
||||||
explicit state migration closure during installation; the CLI's install
|
users can also supply an explicit state migration closure during installation;
|
||||||
command preserves state as-is. Use `help` in the session for the complete
|
the CLI's install command preserves state as-is. Use `help` in the session for
|
||||||
command list.
|
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.
|
||||||
|
|
|
||||||
|
|
@ -45,6 +45,7 @@ pub enum Event {
|
||||||
Restarted(ActorId),
|
Restarted(ActorId),
|
||||||
Paused(ActorId),
|
Paused(ActorId),
|
||||||
Resumed(ActorId),
|
Resumed(ActorId),
|
||||||
|
InvocationCancelled(ActorId),
|
||||||
BehaviorInstalled(ActorId, ImplementationId),
|
BehaviorInstalled(ActorId, ImplementationId),
|
||||||
StateMigrated(ActorId, i64, i64),
|
StateMigrated(ActorId, i64, i64),
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -161,6 +161,27 @@ impl Runtime {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Cancel an active handler and requeue its reserved message for retry.
|
||||||
|
///
|
||||||
|
/// This does not roll back side effects already performed by native code.
|
||||||
|
pub fn cancel(&mut self, id: ActorId) -> ExperimentResult<()> {
|
||||||
|
let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
|
||||||
|
if actor.active.is_none() {
|
||||||
|
return Err("actor has no active invocation to cancel".into());
|
||||||
|
}
|
||||||
|
let message = actor
|
||||||
|
.snapshot
|
||||||
|
.reserved
|
||||||
|
.take()
|
||||||
|
.ok_or("active invocation has no reserved message")?;
|
||||||
|
actor.active = None;
|
||||||
|
actor.paused_from = None;
|
||||||
|
actor.mailbox.push_front(message);
|
||||||
|
actor.snapshot.status = Status::Idle;
|
||||||
|
self.events.push(Event::InvocationCancelled(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(_)) {
|
||||||
|
|
@ -518,14 +539,14 @@ mod tests {
|
||||||
runtime.tick(capacity(5));
|
runtime.tick(capacity(5));
|
||||||
}
|
}
|
||||||
assert_eq!(runtime.inspect(actor)?.status, Status::Awaiting);
|
assert_eq!(runtime.inspect(actor)?.status, Status::Awaiting);
|
||||||
runtime.stop(actor)?;
|
runtime.cancel(actor)?;
|
||||||
assert!(gate.dropped.load(Ordering::SeqCst));
|
assert!(gate.dropped.load(Ordering::SeqCst));
|
||||||
runtime.restart(actor)?;
|
let cancelled = runtime.inspect(actor)?;
|
||||||
let restarted = runtime.inspect(actor)?;
|
assert_eq!(cancelled.mailbox, pending.mailbox);
|
||||||
assert_eq!(restarted.mailbox, pending.mailbox);
|
assert_eq!(cancelled.state, 20);
|
||||||
assert_eq!(restarted.state, 20);
|
assert_eq!(cancelled.status, Status::Idle);
|
||||||
assert_eq!(restarted.incarnation, 2);
|
assert_eq!(cancelled.queued, vec![Message { kind: 1, value: 0 }]);
|
||||||
assert_eq!(restarted.queued.len(), 1);
|
assert!(cancelled.reserved.is_none());
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -712,6 +733,50 @@ mod tests {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn cancelled_input_retries_first_with_new_behavior() -> ExperimentResult<()> {
|
||||||
|
let mut runtime = Runtime::default();
|
||||||
|
let old_implementation = publish(
|
||||||
|
&mut runtime,
|
||||||
|
"pub fn handle(state, kind, value) { (state + value, []) }",
|
||||||
|
)?;
|
||||||
|
let new_implementation = publish(
|
||||||
|
&mut runtime,
|
||||||
|
"pub fn handle(state, kind, value) { (state * 10 + value, []) }",
|
||||||
|
)?;
|
||||||
|
let actor = runtime.spawn(old_implementation, 0, capacity(4))?;
|
||||||
|
runtime.send(actor, Message { kind: 0, value: 1 })?;
|
||||||
|
runtime.send(actor, Message { kind: 0, value: 2 })?;
|
||||||
|
runtime.tick(capacity(1));
|
||||||
|
|
||||||
|
assert_eq!(runtime.inspect(actor)?.status, Status::Running);
|
||||||
|
runtime.cancel(actor)?;
|
||||||
|
let cancelled = runtime.inspect(actor)?;
|
||||||
|
assert_eq!(cancelled.status, Status::Idle);
|
||||||
|
assert_eq!(cancelled.state, 0);
|
||||||
|
assert_eq!(
|
||||||
|
cancelled.queued,
|
||||||
|
vec![Message { kind: 0, value: 1 }, Message { kind: 0, value: 2 }]
|
||||||
|
);
|
||||||
|
assert!(cancelled.reserved.is_none());
|
||||||
|
|
||||||
|
runtime.install(actor, new_implementation)?;
|
||||||
|
runtime.send(actor, Message { kind: 0, value: 3 })?;
|
||||||
|
for _ in 0..100 {
|
||||||
|
runtime.tick(capacity(10));
|
||||||
|
}
|
||||||
|
let completed = runtime.inspect(actor)?;
|
||||||
|
assert_eq!(completed.state, 123);
|
||||||
|
assert_eq!(completed.implementation, new_implementation);
|
||||||
|
assert!(
|
||||||
|
runtime
|
||||||
|
.drain_events()
|
||||||
|
.iter()
|
||||||
|
.any(|event| matches!(event, Event::InvocationCancelled(id) if *id == actor))
|
||||||
|
);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn failed_state_migration_preserves_actor_state_and_implementation() -> ExperimentResult<()> {
|
fn failed_state_migration_preserves_actor_state_and_implementation() -> ExperimentResult<()> {
|
||||||
let mut runtime = Runtime::default();
|
let mut runtime = Runtime::default();
|
||||||
|
|
|
||||||
|
|
@ -135,6 +135,12 @@ fn run_repl<R: BufRead, W: Write>(
|
||||||
|runtime, id| runtime.resume(id),
|
|runtime, id| runtime.resume(id),
|
||||||
runtime,
|
runtime,
|
||||||
),
|
),
|
||||||
|
"cancel" => lifecycle_command(
|
||||||
|
&mut output,
|
||||||
|
&args,
|
||||||
|
|runtime, id| runtime.cancel(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`"),
|
||||||
|
|
@ -392,6 +398,7 @@ fn print_help<W: Write>(output: &mut W) -> io::Result<()> {
|
||||||
restart <actor-id>\n\
|
restart <actor-id>\n\
|
||||||
pause <actor-id>\n\
|
pause <actor-id>\n\
|
||||||
resume <actor-id>\n\
|
resume <actor-id>\n\
|
||||||
|
cancel <actor-id>\n\
|
||||||
help\n\
|
help\n\
|
||||||
quit"
|
quit"
|
||||||
)
|
)
|
||||||
|
|
@ -481,7 +488,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\npause 1\nresume 1\ninspect 1\nevents\nquit\n"
|
"behaviors\npublish counter 1 {source_path}\nhistory counter\nrollback counter 2 1\nspawn 0 counter\ninstall 1 counter\nsend 1 0 1\ntick 1 1\ncancel 1\npause 1\nresume 1\ninspect 1\nevents\nquit\n"
|
||||||
));
|
));
|
||||||
let mut output = Vec::new();
|
let mut output = Vec::new();
|
||||||
run_repl(
|
run_repl(
|
||||||
|
|
@ -500,6 +507,7 @@ mod tests {
|
||||||
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("Paused("));
|
||||||
assert!(output.contains("Resumed("));
|
assert!(output.contains("Resumed("));
|
||||||
|
assert!(output.contains("InvocationCancelled("));
|
||||||
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(())
|
||||||
|
|
|
||||||
|
|
@ -8,12 +8,13 @@ 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`, `pause`, `resume`, `behaviors`, `publish`,
|
`events`, `stop`, `restart`, `pause`, `resume`, `cancel`, `behaviors`,
|
||||||
`history`, `rollback`, and `install`; enter `help` to see usage. `publish` reads a Rune source file and
|
`publish`, `history`, `rollback`, and `install`; enter `help` to see usage.
|
||||||
takes `new` for a first revision or the expected current revision for an
|
`publish` reads a Rune source file and takes `new` for a first revision or the
|
||||||
update. `rollback` requires the expected current revision and a target
|
expected current revision for an update. `rollback` requires the expected
|
||||||
historical revision. `install` selects a named behavior's current implementation and fails if the
|
current revision and a target historical revision. `install` selects a named
|
||||||
actor is not at an idle handler boundary.
|
behavior's current implementation and fails if the actor is not at an idle
|
||||||
|
handler boundary.
|
||||||
`spawn` accepts an optional behavior name and otherwise uses `counter`. All
|
`spawn` accepts an optional behavior name and otherwise uses `counter`. All
|
||||||
commands operate on the same process-local runtime and code store. This CLI is
|
commands operate on the same process-local runtime and code store. This CLI is
|
||||||
for local development, not a versioned external API.
|
for local development, not a versioned external API.
|
||||||
|
|
@ -87,6 +88,14 @@ 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
|
to `tick` and cannot interrupt a currently executing Rune slice or native
|
||||||
callback. They provide cooperative scheduling control, not preemption.
|
callback. They provide cooperative scheduling control, not preemption.
|
||||||
|
|
||||||
|
`cancel` is distinct from `stop`: it drops an actor's active invocation, moves
|
||||||
|
its reserved message to the front of the mailbox, and returns the actor to
|
||||||
|
`Idle`. The input is therefore eligible for retry on the next tick, ahead of
|
||||||
|
messages that were already queued. Committed state is unchanged and no
|
||||||
|
transition effects are applied. Cancellation is also available for paused
|
||||||
|
invocations. It does not undo external side effects already performed by
|
||||||
|
native callbacks, so retries may repeat such effects.
|
||||||
|
|
||||||
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
|
||||||
|
|
@ -154,6 +163,7 @@ resolves it; there is no automatic poison-message policy or supervision.
|
||||||
| 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` |
|
| 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 |
|
| Resume a paused invocation | Retained until invocation completes | Retained | None until completion | Prior runnable/awaiting status |
|
||||||
|
| Cancel an invocation | Last committed state retained | Moved to queue front | None from incomplete transition | `Idle` |
|
||||||
|
|
||||||
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
|
||||||
|
|
@ -166,11 +176,13 @@ These are process-local guarantees only. A host crash loses state, mailboxes,
|
||||||
reservations and events. External/native callback effects may already have
|
reservations and events. External/native callback effects may already have
|
||||||
occurred even when the Rune invocation later fails or is stopped.
|
occurred even when the Rune invocation later fails or is stopped.
|
||||||
|
|
||||||
Lifecycle tests verify that a failed reservation retries from the last
|
Lifecycle tests verify that failed and cancelled reservations retry from the
|
||||||
committed state, that stop/restart preserves reserved-first FIFO order, and
|
last committed state, that stop/restart preserves reserved-first FIFO order,
|
||||||
that a running actor's queued messages are not processed until the reservation
|
and that cancellation followed by installation retries queued input in order
|
||||||
finishes. Transition tests cover malformed results, runtime failures, invalid
|
with the new implementation. Tests also cover that a running actor's queued
|
||||||
targets, aggregate capacity, and successful message effects.
|
messages are not processed until the reservation finishes. Transition tests
|
||||||
|
cover malformed results, runtime failures, invalid targets, aggregate
|
||||||
|
capacity, and successful message effects.
|
||||||
|
|
||||||
## Limits
|
## Limits
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue