//! Single-worker, in-memory actor runtime. No durability or preemption guarantees. use std::cell::RefCell; use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::num::NonZeroUsize; use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::task::{Context, Wake, Waker}; use runa_actor::{ ActorRef, ActorSnapshot, Envelope, Event, ExecutionSnapshot, MailboxId, Message, Status, SystemMessage, }; use runa_code::CompiledArtifact; use runa_core::{ DebugStopReason, DebuggerSnapshot, ImplementationId, LogEntry, LogLevel, SourceBreakpoint, Value, }; use runa_rune::{ ActorContext, ActorLogs, ExperimentResult, Invocation, Slice, value_from_rune, value_to_rune, }; pub type RuntimeResult = ExperimentResult; struct WakeFlag(AtomicBool); impl Wake for WakeFlag { fn wake(self: Arc) { self.0.store(true, Ordering::SeqCst); } fn wake_by_ref(self: &Arc) { self.0.store(true, Ordering::SeqCst); } } struct Actor { snapshot: ActorSnapshot, mailbox: VecDeque, capacity: NonZeroUsize, active: Option>, context: ActorContext, reserved_sender: Option, stopping: bool, shutdown: bool, pending_install: Option, wake: Arc, paused_from: Option, debugger: DebuggerSnapshot, breakpoint_ips: BTreeSet, skip_breakpoint: Option, } #[derive(Default)] pub struct Runtime { actors: BTreeMap, implementations: BTreeMap>, next_mailbox: u64, events: Vec, 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, ) -> ExperimentResult { let id = artifact.implementation_id; if let Some(existing) = self.implementations.get(&id) { if existing.source_id != artifact.source_id { return Err("implementation identity collision".into()); } return Ok(id); } self.implementations.insert(id, artifact); Ok(id) } pub fn spawn( &mut self, implementation: ImplementationId, state: Value, capacity: NonZeroUsize, ) -> ExperimentResult { if !self.implementations.contains_key(&implementation) { return Err("unknown implementation".into()); } state.validate()?; self.next_mailbox = self .next_mailbox .checked_add(1) .ok_or("mailbox IDs exhausted")?; let mut actor_ref = ActorRef::new(); while self.actors.contains_key(&actor_ref) { actor_ref = ActorRef::new(); } self.actors.insert( actor_ref, Actor { snapshot: ActorSnapshot { log_level: LogLevel::Info, actor_ref, mailbox: MailboxId(self.next_mailbox), implementation, incarnation: 1, state, status: Status::Idle, execution: None, queued: Vec::new(), reserved: None, }, mailbox: VecDeque::new(), capacity, active: None, context: ActorContext::with_logs(actor_ref, Rc::clone(&self.logs)), reserved_sender: None, stopping: false, shutdown: false, pending_install: None, wake: Arc::new(WakeFlag(AtomicBool::new(true))), paused_from: None, debugger: DebuggerSnapshot::default(), breakpoint_ips: BTreeSet::new(), skip_breakpoint: None, }, ); self.events.push(Event::Spawned(actor_ref)); Ok(actor_ref) } pub fn send(&mut self, id: ActorRef, message: Message) -> ExperimentResult<()> { self.send_from(None, id, message) } pub fn send_from( &mut self, sender: Option, id: ActorRef, message: Message, ) -> ExperimentResult<()> { message.validate()?; let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if actor.stopping || matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { return Err("actor is not accepting messages".into()); } if actor.mailbox.len() + usize::from(actor.snapshot.reserved.is_some()) >= actor.capacity.get() { return Err("mailbox full".into()); } actor.mailbox.push_back(Envelope { sender, message }); Ok(()) } pub fn inspect(&self, id: ActorRef) -> ExperimentResult { let actor = self.actors.get(&id).ok_or("unknown actor")?; let mut snapshot = actor.snapshot.clone(); snapshot.execution = actor.active.as_ref().map(|invocation| { let progress = invocation.progress(); ExecutionSnapshot { vm_steps: progress.vm_steps, instruction_pointer: progress.instruction_pointer, call_depth: progress.call_depth, } }); snapshot.queued = actor .mailbox .iter() .map(|envelope| envelope.message.clone()) .collect(); Ok(snapshot) } pub fn inspect_execution( &self, id: ActorRef, ) -> ExperimentResult> { let actor = self.actors.get(&id).ok_or("unknown actor")?; let Some(invocation) = &actor.active else { return Ok(None); }; let artifact = self .implementations .get(&actor.snapshot.implementation) .ok_or("unknown implementation")?; Ok(Some(artifact.program().inspect_stack( artifact.original_source(), &invocation.stack(), invocation.progress().vm_steps, ))) } pub fn debugger(&self, id: ActorRef) -> ExperimentResult { Ok(self .actors .get(&id) .ok_or("unknown actor")? .debugger .clone()) } pub fn set_breakpoints( &mut self, id: ActorRef, expected: ImplementationId, lines: &[usize], ) -> ExperimentResult> { if lines.len() > 128 || lines.iter().any(|line| *line > u32::MAX as usize) { return Err("at most 128 breakpoint lines in the 32-bit line range are allowed".into()); } let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if actor.snapshot.implementation != expected { return Err("implementation conflict: loaded behavior changed".into()); } let artifact = &self.implementations[&expected]; let breakpoints = artifact.program().source_breakpoints(lines); actor.breakpoint_ips = breakpoints .iter() .flat_map(|bp| bp.instruction_pointers.iter().copied()) .collect(); actor.debugger.breakpoints = breakpoints.clone(); Ok(breakpoints) } /// Advance only an already active paused actor, retaining its draft and effects. pub fn step_source( &mut self, id: ActorRef, expected: ImplementationId, budget: NonZeroUsize, ) -> ExperimentResult { self.step_source_mode(id, expected, budget, runa_core::StepMode::Into) } pub fn step_source_mode( &mut self, id: ActorRef, expected: ImplementationId, budget: NonZeroUsize, mode: runa_core::StepMode, ) -> ExperimentResult { if budget.get() > 10_000 { return Err("debug step budget must not exceed 10000".into()); } let actor = self.actors.get(&id).ok_or("unknown actor")?; if actor.snapshot.implementation != expected { return Err("implementation conflict: loaded behavior changed".into()); } if actor.snapshot.status != Status::Paused || actor.active.is_none() { return Err("source stepping requires a paused actor with an active invocation".into()); } let origin = self.inspect_execution(id)?.expect("active invocation"); let origin_line = origin.frames[0].range.map(|range| range.start.line); let actor = self.actors.get_mut(&id).expect("actor checked"); let invocation = actor.active.as_ref().expect("active invocation"); actor.skip_breakpoint = Some(invocation.progress().instruction_pointer); actor.debugger.stop_reason = None; let waker = Waker::from(Arc::clone(&actor.wake)); let mut cx = Context::from_waker(&waker); let mut reason = DebugStopReason::Budget; for _ in 0..budget.get() { let actor = self.actors.get_mut(&id).expect("actor checked"); let artifact = &self.implementations[&actor.snapshot.implementation]; let invocation = actor.active.as_mut().expect("active invocation"); if breakpoint_hit( actor.skip_breakpoint.take(), &actor.breakpoint_ips, invocation, ) { reason = DebugStopReason::Breakpoint; break; } match invocation.poll_slice(&mut cx, NonZeroUsize::new(1).unwrap()) { Slice::BudgetExhausted => { if breakpoint_hit(None, &actor.breakpoint_ips, invocation) { reason = DebugStopReason::Breakpoint; break; } let stack = invocation.stack(); if artifact .program() .current_source_line(&stack) .is_some_and(|line| match mode { runa_core::StepMode::Into => { stack.call_depth != origin.call_depth || Some(line) != origin_line } runa_core::StepMode::Over => { stack.call_depth < origin.call_depth || (stack.call_depth == origin.call_depth && Some(line) != origin_line) } runa_core::StepMode::Out => stack.call_depth < origin.call_depth, }) { reason = DebugStopReason::Step; break; } } Slice::Awaiting => { reason = DebugStopReason::Awaiting; break; } Slice::Completed(value) => { match self.commit(&id, value) { Ok(()) => { reason = if self.actors.contains_key(&id) { DebugStopReason::Completed } else { DebugStopReason::Removed } } Err(error) => { self.fail(&id, error.to_string()); reason = DebugStopReason::Failed; } } break; } Slice::Failed(error) => { self.fail(&id, error.to_string()); reason = DebugStopReason::Failed; break; } } } if let Some(actor) = self.actors.get_mut(&id) { actor.debugger.stop_reason = Some(reason); if !matches!(actor.snapshot.status, Status::Failed(_)) { actor.paused_from = Some(if actor.active.is_none() { Status::Idle } else if reason == DebugStopReason::Awaiting { Status::Awaiting } else { Status::Running }); actor.snapshot.status = Status::Paused; self.events.push(Event::Paused(id)); } } Ok(reason) } pub fn edit_actor( &mut self, id: ActorRef, state: Option, messages: Option<(Option, Vec)>, ) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if actor.active.is_some() || !matches!(actor.snapshot.status, Status::Paused | Status::Failed(_)) { return Err( "editing requires a paused or failed actor with no active invocation".into(), ); } if let Some(state) = &state { state.validate()?; } if let Some((reserved, queued)) = &messages { for message in reserved.iter().chain(queued) { message.validate()?; } if queued.len() + usize::from(reserved.is_some()) > actor.capacity.get() { return Err("mailbox full".into()); } } if let Some(state) = state { actor.snapshot.state = state; } let Some((reserved, queued)) = messages else { return Ok(()); }; // Preserve sender metadata for unchanged messages, including the failed reservation. if actor.snapshot.reserved != reserved { actor.reserved_sender = None; } actor.snapshot.reserved = reserved; let previous = std::mem::take(&mut actor.mailbox); actor.mailbox = queued .into_iter() .enumerate() .map(|(index, message)| { let sender = previous .get(index) .filter(|old| old.message == message) .and_then(|old| old.sender); Envelope { sender, message } }) .collect(); Ok(()) } pub fn set_log_level(&mut self, id: ActorRef, level: LogLevel) -> ExperimentResult<()> { self.actors .get_mut(&id) .ok_or("unknown actor")? .snapshot .log_level = level; self.logs.borrow_mut().set_level(id, level); Ok(()) } pub fn log_level(&self, id: ActorRef) -> ExperimentResult { self.actors.get(&id).ok_or("unknown actor")?; Ok(self.logs.borrow().level(id)) } pub fn logs(&self, actors: &[ActorRef], after: u64) -> ExperimentResult> { for actor in actors { self.actors.get(actor).ok_or("unknown actor")?; } Ok(self.logs.borrow().entries(actors, after)) } pub fn set_log_sink(&mut self, sink: impl Fn(LogEntry) + 'static) { self.logs.borrow_mut().set_sink(sink); } pub fn actors(&self) -> Vec { self.actors .values() .map(|actor| { let mut snapshot = actor.snapshot.clone(); snapshot.execution = actor.active.as_ref().map(|invocation| { let progress = invocation.progress(); ExecutionSnapshot { vm_steps: progress.vm_steps, instruction_pointer: progress.instruction_pointer, call_depth: progress.call_depth, } }); snapshot.queued = actor .mailbox .iter() .map(|envelope| envelope.message.clone()) .collect(); snapshot }) .collect() } pub fn stop(&mut self, id: ActorRef) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { return Err("actor is not running".into()); } actor.stopping = true; if let Some(status) = actor.paused_from.take() { actor.snapshot.status = status; } if actor.debugger.stop_reason == Some(DebugStopReason::Breakpoint) { actor.skip_breakpoint = actor .active .as_ref() .map(|invocation| invocation.progress().instruction_pointer); } actor.debugger.stop_reason = None; Ok(()) } pub fn kill(&mut self, id: ActorRef) -> ExperimentResult<()> { self.actors.remove(&id).ok_or("unknown actor")?; self.logs.borrow_mut().remove(id); self.events.push(Event::Stopped(id)); Ok(()) } pub fn system(&mut self, id: ActorRef, message: SystemMessage) -> ExperimentResult<()> { match message { SystemMessage::Stop => self.stop(id), SystemMessage::Kill => self.kill(id), SystemMessage::Restart => self.restart(id), SystemMessage::Pause => self.pause(id), SystemMessage::Resume => self.resume(id), SystemMessage::Reload(implementation) | SystemMessage::Become(implementation) => { self.schedule_install(id, implementation) } } } pub fn schedule_install( &mut self, id: ActorRef, implementation: ImplementationId, ) -> ExperimentResult<()> { if !self.implementations.contains_key(&implementation) { return Err("unknown implementation".into()); } let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if actor.stopping || matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { return Err("actor is not running".into()); } actor.pending_install = Some(implementation); Ok(()) } /// Schedule a boundary install, or replace a failed actor's inactive implementation. pub fn set_actor_behavior( &mut self, id: ActorRef, implementation: ImplementationId, ) -> ExperimentResult<()> { if !self.implementations.contains_key(&implementation) { return Err("unknown implementation".into()); } let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; if actor.stopping { return Err("actor is stopping".into()); } if !matches!(actor.snapshot.status, Status::Failed(_)) { actor.pending_install = Some(implementation); } else { actor.pending_install = None; actor.snapshot.implementation = implementation; clear_debugger(actor); self.events .push(Event::BehaviorInstalled(id, implementation)); } Ok(()) } /// Pause scheduling between worker ticks without discarding an invocation. pub fn pause(&mut self, id: ActorRef) -> 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; actor.debugger.stop_reason = None; self.events.push(Event::Paused(id)); Ok(()) } /// Resume from the exact runnable or awaiting state saved by `pause`. pub fn resume(&mut self, id: ActorRef) -> 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; if actor.debugger.stop_reason == Some(DebugStopReason::Breakpoint) { actor.skip_breakpoint = actor .active .as_ref() .map(|invocation| invocation.progress().instruction_pointer); } actor.debugger.stop_reason = None; self.events.push(Event::Resumed(id)); 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: ActorRef) -> 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.debugger.stop_reason = None; actor.skip_breakpoint = None; actor.paused_from = None; actor.context.take_effects(); actor.mailbox.push_front(Envelope { sender: actor.reserved_sender.take(), message, }); actor.snapshot.status = Status::Idle; self.events.push(Event::InvocationCancelled(id)); Ok(()) } pub fn restart(&mut self, id: ActorRef) -> ExperimentResult<()> { let actor = self.actors.get_mut(&id).ok_or("unknown actor")?; actor.snapshot.incarnation = actor .snapshot .incarnation .checked_add(1) .ok_or("incarnation exhausted")?; actor.active = None; actor.debugger.stop_reason = None; actor.skip_breakpoint = None; actor.context.take_effects(); actor.stopping = false; actor.shutdown = false; actor.pending_install = None; if let Some(message) = actor.snapshot.reserved.take() { actor.mailbox.push_front(Envelope { sender: actor.reserved_sender.take(), message, }); } actor.snapshot.status = Status::Idle; actor.paused_from = None; self.events.push(Event::Restarted(id)); Ok(()) } /// Install an immutable implementation between handler invocations. pub fn install( &mut self, id: ActorRef, implementation: ImplementationId, ) -> ExperimentResult<()> { self.install_with_migration(id, implementation, Ok) } /// Migrate committed state and install an implementation as one safe-point /// operation. A migration error leaves both unchanged. pub fn install_with_migration( &mut self, id: ActorRef, implementation: ImplementationId, migrate: impl FnOnce(Value) -> ExperimentResult, ) -> ExperimentResult<()> { if !self.implementations.contains_key(&implementation) { return Err("unknown implementation".into()); } let current = self.actors.get(&id).ok_or("unknown actor")?; if current.active.is_some() || current.snapshot.reserved.is_some() || current.snapshot.status != Status::Idle { return Err("behavior can only be installed at an idle handler boundary".into()); } let previous_state = current.snapshot.state.clone(); let migrated_state = migrate(previous_state.clone()) .map_err(|error| format!("state migration failed: {error}"))?; migrated_state.validate()?; let actor = self.actors.get_mut(&id).expect("actor checked above"); actor.snapshot.state = migrated_state.clone(); actor.snapshot.implementation = implementation; clear_debugger(actor); self.events .push(Event::BehaviorInstalled(id, implementation)); if migrated_state != previous_state { self.events .push(Event::StateMigrated(id, previous_state, migrated_state)); } Ok(()) } pub fn drain_events(&mut self) -> Vec { std::mem::take(&mut self.events) } fn fail(&mut self, id: &ActorRef, error: String) { let actor = self.actors.get_mut(id).expect("scheduled actor exists"); actor.active = None; actor.debugger.stop_reason = None; actor.skip_breakpoint = None; actor.paused_from = None; actor.context.take_effects(); actor.snapshot.status = Status::Failed(error.clone()); self.events.push(Event::Failed(*id, error)); } fn commit(&mut self, id: &ActorRef, state: rune::Value) -> ExperimentResult<()> { let state = if state.into_unit().is_ok() { None } else { Some(value_from_rune(&state)?) }; let effects = self.actors[id].context.take_effects(); let mut counts = BTreeMap::::new(); for effect in &effects { let target = &effect.recipient; let actor = self .actors .get(target) .ok_or("effect targets unknown actor")?; if (actor.stopping && target != id) || matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) { return Err("effect targets unavailable actor".into()); } let count = counts.entry(*target).or_default(); *count += 1; let retained = usize::from(target != id && actor.snapshot.reserved.is_some()); if actor.mailbox.len() + retained + *count > actor.capacity.get() { return Err("effect exceeds mailbox capacity".into()); } } let actor = self.actors.get_mut(id).expect("scheduled actor exists"); if let Some(state) = state { actor.snapshot.state = state; } actor.snapshot.reserved = None; actor.snapshot.status = Status::Idle; actor.active = None; actor.reserved_sender = None; let shutdown = actor.shutdown; for effect in effects { self.actors .get_mut(&effect.recipient) .expect("validated target") .mailbox .push_back(Envelope { sender: effect.sender, message: effect.message, }); } self.events.push(Event::Committed(*id)); if shutdown { self.kill(*id)?; } Ok(()) } /// Give each runnable actor one bounded slice. Awaiting actors require a wake. pub fn tick(&mut self, budget: NonZeroUsize) { 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::Paused | Status::Stopped | Status::Failed(_) ) { continue; } if actor.active.is_none() { if actor.stopping { let program = self.implementations[&actor.snapshot.implementation].program(); if !program.has_function("on_shutdown") { self.kill(id).expect("scheduled actor exists"); continue; } let invocation = (|| -> ExperimentResult<_> { Ok(Invocation::start( program.vm(), "on_shutdown", (value_to_rune(&actor.snapshot.state)?,), )?) })(); match invocation { Ok(invocation) => { actor.active = Some(invocation); actor.shutdown = true; actor.context = ActorContext::with_logs(id, Rc::clone(&self.logs)); } Err(error) => { self.fail(&id, error.to_string()); continue; } } } else { if let Some(implementation) = actor.pending_install.take() { actor.snapshot.implementation = implementation; clear_debugger(actor); self.events .push(Event::BehaviorInstalled(id, implementation)); } let Some(message) = actor.mailbox.pop_front() else { continue; }; actor.snapshot.reserved = Some(message.message.clone()); actor.reserved_sender = message.sender; actor.context = ActorContext::with_logs(id, Rc::clone(&self.logs)); let program = self.implementations[&actor.snapshot.implementation].program(); let invocation = (|| -> ExperimentResult<_> { let state = value_to_rune(&actor.snapshot.state)?; let sender = value_to_rune(&message.sender.map_or(Value::Null, Value::ActorRef))?; let message = value_to_rune(&message.message)?; Ok(Invocation::start( program.vm(), "handle", (actor.context.clone(), state, sender, message), )?) })(); match invocation { Ok(invocation) => actor.active = Some(invocation), Err(error) => { self.fail(&id, error.to_string()); continue; } } } actor.snapshot.status = Status::Running; actor.debugger.stop_reason = None; } if actor.snapshot.status == Status::Awaiting && !actor.wake.0.load(Ordering::SeqCst) { continue; } actor.wake.0.store(false, Ordering::SeqCst); let waker = Waker::from(Arc::clone(&actor.wake)); let mut cx = Context::from_waker(&waker); let mut outcome = Slice::BudgetExhausted; for _ in 0..budget.get() { let invocation = actor.active.as_mut().expect("active invocation"); if breakpoint_hit( actor.skip_breakpoint.take(), &actor.breakpoint_ips, invocation, ) { actor.paused_from = Some(Status::Running); actor.snapshot.status = Status::Paused; actor.debugger.stop_reason = Some(DebugStopReason::Breakpoint); self.events.push(Event::Paused(id)); break; } let slice_budget = if actor.breakpoint_ips.is_empty() { budget } else { NonZeroUsize::new(1).unwrap() }; outcome = invocation.poll_slice(&mut cx, slice_budget); if !matches!(outcome, Slice::BudgetExhausted) || actor.breakpoint_ips.is_empty() { break; } } if actor.snapshot.status == Status::Paused { continue; } match outcome { Slice::Completed(transition) => { if let Err(error) = self.commit(&id, transition) { self.fail(&id, error.to_string()); } } Slice::Awaiting => actor.snapshot.status = Status::Awaiting, Slice::BudgetExhausted => actor.snapshot.status = Status::Running, Slice::Failed(error) => self.fail(&id, error.to_string()), } } } } fn clear_debugger(actor: &mut Actor) { actor.debugger = DebuggerSnapshot::default(); actor.breakpoint_ips.clear(); actor.skip_breakpoint = None; } fn breakpoint_hit( skip: Option, breakpoints: &BTreeSet, invocation: &Invocation, ) -> bool { let ip = invocation.progress().instruction_pointer; skip != Some(ip) && breakpoints.contains(&ip) && !invocation.stack().awaiting } #[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] fn debugger_step_over_skips_nested_calls_and_step_out_returns_to_caller() -> ExperimentResult<()> { use runa_core::StepMode; let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) {\n let next = outer(state);\n next + 1\n}\nfn outer(value) {\n let result = inner(value);\n result + 1\n}\nfn inner(value) {\n value + 1\n}", )?; let actor = runtime.spawn(code, Value::Integer(3), capacity(2))?; runtime.set_breakpoints(actor, code, &[1])?; runtime.send(actor, Value::Null)?; runtime.tick(capacity(100)); runtime.set_breakpoints(actor, code, &[])?; assert_eq!( runtime.step_source_mode(actor, code, capacity(1000), StepMode::Over)?, DebugStopReason::Step ); let over = runtime.inspect_execution(actor)?.unwrap(); assert_eq!(over.call_depth, 0); assert_eq!(over.frames[0].function.as_deref(), Some("handle")); assert_eq!(over.frames[0].range.unwrap().start.line, 2); assert_eq!(runtime.inspect(actor)?.state, Value::Integer(3)); assert_eq!( runtime.step_source_mode(actor, code, capacity(1000), StepMode::Out)?, DebugStopReason::Completed ); assert_eq!(runtime.inspect(actor)?.state, Value::Integer(6)); assert_eq!(runtime.inspect(actor)?.status, Status::Paused); runtime.restart(actor)?; runtime.set_breakpoints(actor, code, &[8])?; runtime.send(actor, Value::Null)?; runtime.tick(capacity(100)); assert_eq!(runtime.inspect_execution(actor)?.unwrap().call_depth, 2); runtime.set_breakpoints(actor, code, &[])?; assert_eq!( runtime.step_source_mode(actor, code, capacity(1000), StepMode::Out)?, DebugStopReason::Step ); let out = runtime.inspect_execution(actor)?.unwrap(); assert_eq!(out.call_depth, 1); assert_eq!(out.frames[0].function.as_deref(), Some("outer")); assert_eq!(runtime.inspect(actor)?.state, Value::Integer(6)); assert_eq!( runtime.step_source_mode(actor, code, capacity(1000), StepMode::Out)?, DebugStopReason::Step ); assert_eq!(runtime.inspect_execution(actor)?.unwrap().call_depth, 0); Ok(()) } #[test] fn debugger_step_modes_respect_nested_breakpoints_and_recursive_depth() -> ExperimentResult<()> { use runa_core::StepMode; for mode in [StepMode::Over, StepMode::Out] { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) {\n recurse(4)\n}\nfn recurse(n) {\n if n == 0 { return 7; }\n let value = recurse(n - 1);\n value + 1\n}", )?; let actor = runtime.spawn(code, Value::Integer(0), capacity(2))?; runtime.set_breakpoints(actor, code, &[1])?; runtime.send(actor, Value::Null)?; runtime.tick(capacity(100)); runtime.set_breakpoints(actor, code, &[5])?; assert_eq!( runtime.step_source_mode(actor, code, capacity(1000), mode)?, DebugStopReason::Breakpoint ); assert_eq!(runtime.inspect_execution(actor)?.unwrap().call_depth, 1); runtime.set_breakpoints(actor, code, &[])?; assert_eq!( runtime.step_source_mode(actor, code, capacity(1000), StepMode::Over)?, DebugStopReason::Step ); let over = runtime.inspect_execution(actor)?.unwrap(); assert_eq!( over.call_depth, 1, "recursive calls must not count as returning to the origin frame" ); assert_eq!(over.frames[0].range.unwrap().start.line, 6); assert_eq!(runtime.inspect(actor)?.state, Value::Integer(0)); assert_eq!( runtime.step_source_mode(actor, code, capacity(1000), StepMode::Out)?, DebugStopReason::Step ); assert_eq!(runtime.inspect_execution(actor)?.unwrap().call_depth, 0); } Ok(()) } #[test] fn debugger_over_and_out_keep_budget_await_and_failure_boundaries() -> ExperimentResult<()> { use runa_core::StepMode; for mode in [StepMode::Over, StepMode::Out] { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) {\n spin()\n}\nfn spin() { loop {} }", )?; let actor = runtime.spawn(code, Value::Null, capacity(2))?; runtime.set_breakpoints(actor, code, &[1])?; runtime.send(actor, Value::Null)?; runtime.tick(capacity(100)); let before = runtime.inspect_execution(actor)?.unwrap().vm_steps; assert_eq!( runtime.step_source_mode(actor, code, capacity(20), mode)?, DebugStopReason::Budget ); assert_eq!( runtime.inspect_execution(actor)?.unwrap().vm_steps, before + 20 ); assert_eq!(runtime.inspect(actor)?.status, Status::Paused); let code = publish( &mut runtime, "pub async fn handle(ctx, state, sender, message) {\n native_suspend(7).await;\n 42\n}", )?; let actor = runtime.spawn(code, Value::Integer(1), capacity(2))?; runtime.set_breakpoints(actor, code, &[1])?; runtime.send(actor, Value::Null)?; runtime.tick(capacity(100)); runtime.set_breakpoints(actor, code, &[])?; assert_eq!( runtime.step_source_mode(actor, code, capacity(100), mode)?, DebugStopReason::Awaiting ); assert_eq!(runtime.inspect(actor)?.state, Value::Integer(1)); assert_eq!(runtime.inspect(actor)?.status, Status::Paused); runtime.resume(actor)?; runtime.tick(capacity(100)); assert_eq!(runtime.inspect(actor)?.state, Value::Integer(42)); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) {\n crash()\n}\nfn crash() { 1 / 0 }", )?; let actor = runtime.spawn(code, Value::Integer(1), capacity(2))?; runtime.set_breakpoints(actor, code, &[1])?; runtime.send(actor, Value::Null)?; runtime.tick(capacity(100)); runtime.set_breakpoints(actor, code, &[])?; assert_eq!( runtime.step_source_mode(actor, code, capacity(100), mode)?, DebugStopReason::Failed ); assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_))); assert_eq!(runtime.inspect(actor)?.reserved, Some(Value::Null)); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state }\npub fn on_shutdown(state) {\n state + 1\n}", )?; let actor = runtime.spawn(code, Value::Integer(1), capacity(2))?; runtime.set_breakpoints(actor, code, &[2])?; runtime.stop(actor)?; runtime.tick(capacity(100)); runtime.set_breakpoints(actor, code, &[])?; let mut removed = false; for _ in 0..10 { if runtime.step_source_mode(actor, code, capacity(100), mode)? == DebugStopReason::Removed { removed = true; break; } } assert!(removed); assert!(runtime.inspect(actor).is_err()); } Ok(()) } #[test] fn debugger_breakpoints_preserve_transactions_and_step_only_one_actor() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let source = "pub fn handle(ctx, state, sender, message) {\n ctx.send(ctx.this, 99);\n let next = nested(state);\n next\n}\nfn nested(value) {\n value + 1\n}"; let code = publish(&mut runtime, source)?; let actor = runtime.spawn(code, Value::Integer(3), capacity(4))?; let other = runtime.spawn(code, Value::Integer(10), capacity(4))?; runtime.send(actor, Value::Null)?; let verified = runtime.set_breakpoints(actor, code, &[2, 100, 2])?; assert_eq!(verified.len(), 2); assert!(!verified[0].instruction_pointers.is_empty()); assert!(verified[1].instruction_pointers.is_empty()); runtime.tick(capacity(1000)); let before = runtime.inspect(actor)?; assert_eq!(before.status, Status::Paused); assert_eq!(before.state, Value::Integer(3)); assert_eq!(before.reserved, Some(Value::Null)); assert!( before.queued.is_empty(), "draft send is not visible before commit" ); assert_eq!( runtime.debugger(actor)?.stop_reason, Some(DebugStopReason::Breakpoint) ); assert_eq!( runtime.inspect_execution(actor)?.unwrap().frames[0] .range .unwrap() .start .line, 2 ); runtime.send(other, Value::Null)?; let other_before = runtime.inspect(other)?; let origin = runtime.inspect_execution(actor)?.unwrap().vm_steps; let reason = runtime.step_source(actor, code, capacity(1000))?; assert_eq!(reason, DebugStopReason::Step); let stepped = runtime.inspect_execution(actor)?.unwrap(); assert!(stepped.vm_steps > origin); assert_eq!(stepped.frames[0].function.as_deref(), Some("nested")); assert_eq!(runtime.inspect(actor)?.state, Value::Integer(3)); assert_eq!(runtime.inspect(other)?.queued, other_before.queued); assert_eq!(runtime.inspect(other)?.execution, other_before.execution); let mut completed = false; for _ in 0..20 { if runtime.step_source(actor, code, capacity(1000))? == DebugStopReason::Completed { completed = true; break; } } assert!(completed); let after = runtime.inspect(actor)?; assert_eq!(after.status, Status::Paused); assert_eq!(after.state, Value::Integer(4)); assert_eq!(after.queued, [Value::Integer(99)]); assert!(after.reserved.is_none()); assert!(after.execution.is_none()); assert!(runtime.step_source(actor, code, capacity(1000)).is_err()); runtime.resume(actor)?; assert_eq!(runtime.inspect(actor)?.status, Status::Idle); Ok(()) } #[test] fn debugger_resume_crosses_breakpoint_once_and_loop_can_hit_again() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) {\n let n = 0;\n loop {\n n += 1;\n }\n}", )?; let actor = runtime.spawn(code, Value::Null, capacity(2))?; runtime.set_breakpoints(actor, code, &[3])?; runtime.send(actor, Value::Null)?; runtime.tick(capacity(100)); let stopped = runtime.inspect_execution(actor)?.unwrap().vm_steps; assert_eq!(runtime.inspect(actor)?.status, Status::Paused); runtime.tick(capacity(100)); assert_eq!(runtime.inspect_execution(actor)?.unwrap().vm_steps, stopped); runtime.resume(actor)?; runtime.tick(capacity(100)); assert_eq!(runtime.inspect(actor)?.status, Status::Paused); assert!(runtime.inspect_execution(actor)?.unwrap().vm_steps > stopped); runtime.set_breakpoints(actor, code, &[])?; runtime.resume(actor)?; runtime.tick(capacity(100)); assert_eq!(runtime.inspect(actor)?.status, Status::Running); Ok(()) } #[test] fn debugger_bounded_steps_handle_same_line_loops_awaits_failure_and_shutdown() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { loop {} }", )?; let actor = runtime.spawn(code, Value::Null, capacity(2))?; runtime.send(actor, Value::Null)?; runtime.tick(capacity(1)); runtime.pause(actor)?; let before = runtime.inspect_execution(actor)?.unwrap().vm_steps; assert_eq!( runtime.step_source(actor, code, capacity(5))?, DebugStopReason::Budget ); assert_eq!( runtime.inspect_execution(actor)?.unwrap().vm_steps, before + 5 ); assert_eq!(runtime.inspect(actor)?.status, Status::Paused); assert!(runtime.step_source(actor, code, capacity(10_001)).is_err()); let code = publish( &mut runtime, "pub async fn handle(ctx, state, sender, message) {\n native_suspend(7).await;\n 42\n}", )?; let actor = runtime.spawn(code, Value::Integer(1), capacity(2))?; runtime.set_breakpoints(actor, code, &[1])?; runtime.send(actor, Value::Null)?; runtime.tick(capacity(100)); runtime.set_breakpoints(actor, code, &[])?; assert_eq!( runtime.step_source(actor, code, capacity(100))?, DebugStopReason::Awaiting ); assert_eq!(runtime.inspect(actor)?.status, Status::Paused); assert!(runtime.inspect_execution(actor)?.unwrap().awaiting); for _ in 0..20 { if runtime.step_source(actor, code, capacity(100))? == DebugStopReason::Completed { break; } } assert_eq!(runtime.inspect(actor)?.state, Value::Integer(42)); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) {\n 1 / 0\n}", )?; let actor = runtime.spawn(code, Value::Null, capacity(2))?; runtime.set_breakpoints(actor, code, &[1])?; runtime.send(actor, Value::Null)?; runtime.tick(capacity(100)); runtime.set_breakpoints(actor, code, &[])?; assert_eq!( runtime.step_source(actor, code, capacity(100))?, DebugStopReason::Failed ); assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_))); assert_eq!(runtime.inspect(actor)?.reserved, Some(Value::Null)); assert!(runtime.resume(actor).is_err()); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state }\npub fn on_shutdown(state) {\n state + 1\n}", )?; let actor = runtime.spawn(code, Value::Integer(0), capacity(2))?; assert!( !runtime.set_breakpoints(actor, code, &[2])?[0] .instruction_pointers .is_empty() ); runtime.stop(actor)?; runtime.tick(capacity(100)); assert_eq!(runtime.inspect(actor)?.status, Status::Paused); runtime.set_breakpoints(actor, code, &[])?; let mut removed = false; for _ in 0..10 { if runtime.step_source(actor, code, capacity(100))? == DebugStopReason::Removed { removed = true; break; } } assert!(removed); assert!(runtime.inspect(actor).is_err()); Ok(()) } #[test] fn debugger_configuration_is_version_bound_and_resets_only_at_install() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) {\n state + 1\n}", )?; let new = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state + 2 }", )?; let actor = runtime.spawn(code, Value::Integer(0), capacity(2))?; runtime.set_breakpoints(actor, code, &[1])?; assert!(runtime.set_breakpoints(actor, new, &[0]).is_err()); assert!( runtime.step_source(actor, code, capacity(100)).is_err(), "idle actor must not start a handler implicitly" ); runtime.send(actor, Value::Null)?; runtime.tick(capacity(100)); runtime.schedule_install(actor, new)?; assert_eq!(runtime.debugger(actor)?.breakpoints.len(), 1); assert!(runtime.step_source(actor, new, capacity(100)).is_err()); runtime.restart(actor)?; assert_eq!( runtime.debugger(actor)?.breakpoints.len(), 1, "same loaded version retains breakpoints across restart" ); runtime.tick(capacity(100)); assert_eq!(runtime.inspect(actor)?.status, Status::Paused); runtime.set_breakpoints(actor, code, &[])?; runtime.resume(actor)?; runtime.tick(capacity(100)); runtime.schedule_install(actor, new)?; runtime.set_breakpoints(actor, code, &[1])?; runtime.tick(capacity(100)); assert_eq!(runtime.inspect(actor)?.implementation, new); assert_eq!(runtime.debugger(actor)?, DebuggerSnapshot::default()); assert!(runtime.set_breakpoints(actor, code, &[]).is_err()); Ok(()) } #[test] fn bundled_counter_defaults_missing_state_to_zero() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, include_str!("../../../runa-std/actors/counter.rn"), )?; for (initial, expected) in [(Value::Null, 1), (Value::Integer(10), 11)] { let actor = runtime.spawn(implementation, initial, capacity(2))?; runtime.send( actor, Value::Record(BTreeMap::from([( "op".to_owned(), Value::String("inc".to_owned()), )])), )?; for _ in 0..1000 { runtime.tick(capacity(1)); if runtime.inspect(actor)?.state == Value::Integer(expected) { break; } } let snapshot = runtime.inspect(actor)?; assert_eq!(snapshot.state, Value::Integer(expected)); assert_eq!(snapshot.queued.len(), 1); assert_eq!(runtime.logs(&[actor], 0)?.len(), 1); runtime.kill(actor)?; } Ok(()) } #[test] fn stop_removes_paused_actors_and_kill_discards_identity_mailbox_and_logs() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { ctx.info(\"started\"); loop {} }", )?; let actor = runtime.spawn(implementation, Value::Integer(5), capacity(3))?; runtime.send(actor, Value::Null)?; runtime.send(actor, Value::Integer(2))?; for _ in 0..10 { runtime.tick(capacity(32)); } assert_eq!(runtime.logs(&[actor], 0)?.len(), 1); runtime.set_log_level(actor, LogLevel::Fatal)?; runtime.kill(actor)?; assert!(runtime.inspect(actor).is_err()); assert!(runtime.restart(actor).is_err()); assert!(runtime.send(actor, Value::Null).is_err()); assert!(runtime.kill(actor).is_err()); assert!(runtime.actors().is_empty()); assert!(runtime.logs(&[], 0)?.is_empty()); assert_eq!(runtime.logs.borrow().level(actor), LogLevel::Info); let idle = runtime.spawn(implementation, Value::Null, capacity(1))?; runtime.pause(idle)?; runtime.stop(idle)?; assert!(runtime.inspect(idle).is_ok()); runtime.tick(capacity(32)); assert!(runtime.inspect(idle).is_err()); Ok(()) } #[test] fn optional_state_retains_committed_state_and_commits_sends() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, r#"pub fn handle(ctx, state, sender, message) { state.value = 999; if message == 1 { ctx.send(ctx.this, 2); return; } if message == 2 { return (); } #{value: 7} } pub fn on_shutdown(state) { if state.value != 7 { panic("incorrect shutdown state"); } state.value = 1000; }"#, )?; let initial = Value::Record(BTreeMap::from([("value".to_owned(), Value::Integer(3))])); let actor = runtime.spawn(implementation, initial.clone(), capacity(4))?; runtime.send(actor, Value::Integer(1))?; for _ in 0..50 { runtime.tick(capacity(1)); } let snapshot = runtime.inspect(actor)?; assert_eq!(snapshot.state, initial); assert_eq!(snapshot.status, Status::Idle); assert!(snapshot.reserved.is_none()); assert!(snapshot.queued.is_empty()); assert_eq!( runtime .drain_events() .iter() .filter(|event| matches!(event, Event::Committed(id) if *id == actor)) .count(), 2 ); runtime.send(actor, Value::Integer(3))?; for _ in 0..50 { runtime.tick(capacity(32)); } assert_eq!( runtime.inspect(actor)?.state, Value::Record(BTreeMap::from([("value".to_owned(), Value::Integer(7))])) ); runtime.stop(actor)?; for _ in 0..50 { runtime.tick(capacity(32)); } assert!(runtime.inspect(actor).is_err()); assert!( runtime .drain_events() .iter() .any(|event| matches!(event, Event::Stopped(id) if *id == actor)) ); Ok(()) } #[test] fn omitted_state_still_validates_effects_and_handles_null_state() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { ctx.send(message, 1); }", )?; let actor = runtime.spawn(implementation, Value::Integer(5), capacity(2))?; runtime.send(actor, Value::ActorRef(ActorRef::new()))?; for _ in 0..20 { runtime.tick(capacity(32)); } let snapshot = runtime.inspect(actor)?; assert_eq!(snapshot.state, Value::Integer(5)); assert!(matches!(snapshot.status, Status::Failed(_))); assert!(snapshot.reserved.is_some()); let implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) {}", )?; let actor = runtime.spawn(implementation, Value::Null, capacity(2))?; runtime.send(actor, Value::Null)?; for _ in 0..10 { runtime.tick(capacity(32)); } assert_eq!(runtime.inspect(actor)?.state, Value::Null); assert_eq!(runtime.inspect(actor)?.status, Status::Idle); assert!(runtime.inspect(actor)?.reserved.is_none()); Ok(()) } #[test] fn editing_is_guarded_atomic_and_can_repair_a_failed_reservation() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state + message.value }", )?; let actor = runtime.spawn(implementation, Value::Integer(1), capacity(2))?; assert!( runtime .edit_actor(actor, Some(Value::Integer(9)), None) .is_err() ); runtime.send(actor, Value::Null)?; runtime.send(actor, Value::Integer(4))?; for _ in 0..10 { runtime.tick(capacity(32)); } assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_))); assert!( runtime .edit_actor( actor, Some(Value::Integer(9)), Some((Some(Value::Null), vec![Value::Null, Value::Null])) ) .is_err() ); assert_eq!(runtime.inspect(actor)?.state, Value::Integer(1)); runtime.edit_actor(actor, Some(Value::Integer(10)), None)?; assert_eq!(runtime.inspect(actor)?.queued, vec![Value::Integer(4)]); let message = Value::Record(BTreeMap::from([("value".to_owned(), Value::Integer(2))])); runtime.edit_actor(actor, None, Some((Some(message), vec![])))?; runtime.restart(actor)?; for _ in 0..10 { runtime.tick(capacity(32)); } assert_eq!(runtime.inspect(actor)?.state, Value::Integer(12)); runtime.pause(actor)?; runtime.edit_actor( actor, Some(Value::Integer(20)), Some((None, vec![Value::Null])), )?; assert_eq!(runtime.inspect(actor)?.status, Status::Paused); runtime.resume(actor)?; let runaway = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { loop {} }", )?; let active = runtime.spawn(runaway, Value::Null, capacity(2))?; runtime.send(active, Value::Null)?; runtime.tick(capacity(1)); runtime.pause(active)?; assert!( runtime .edit_actor(active, Some(Value::Integer(9)), Some((None, vec![]))) .is_err() ); runtime.kill(active)?; assert!( runtime .edit_actor(active, Some(Value::Integer(9)), Some((None, vec![]))) .is_err() ); Ok(()) } #[test] fn logs_are_immediate_filtered_ordered_bounded_and_survive_failure() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let emitted = Rc::new(RefCell::new(Vec::new())); let sink = emitted.clone(); runtime.set_log_sink(move |entry| sink.borrow_mut().push(entry)); let implementation = publish( &mut runtime, r#"pub fn handle(ctx, state, sender, message) { ctx.trace("trace"); ctx.debug("debug"); ctx.info("info"); ctx.warn("warn"); ctx.error("error"); ctx.fatal("fatal"); ctx.log("info", "generic"); panic("failed after logging") }"#, )?; let actor = runtime.spawn(implementation, Value::Integer(1), capacity(2))?; assert_eq!(runtime.log_level(actor)?, LogLevel::Info); runtime.send(actor, Value::Null)?; for _ in 0..10 { runtime.tick(capacity(32)); } let entries = runtime.logs(&[actor], 0)?; assert_eq!( entries .iter() .map(|entry| entry.message.as_str()) .collect::>(), ["info", "warn", "error", "fatal", "generic"] ); assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_))); runtime.set_log_level(actor, LogLevel::Fatal)?; runtime.restart(actor)?; for _ in 0..10 { runtime.tick(capacity(32)); } assert_eq!( runtime .logs(&[actor], entries.last().unwrap().sequence)? .len(), 1 ); let busy = publish( &mut runtime, r#"pub fn handle(ctx, state, sender, message) { let n = 0; while n < 1005 { ctx.info("entry"); n += 1; } loop {} }"#, )?; let second = runtime.spawn(busy, Value::Null, capacity(2))?; runtime.send(second, Value::Null)?; for _ in 0..100 { runtime.tick(capacity(1000)); } let retained = runtime.logs(&[second], 0)?; assert_eq!(retained.len(), 1000); assert!( retained .windows(2) .all(|pair| pair[0].sequence < pair[1].sequence) ); runtime.kill(second)?; assert!(runtime.logs(&[second], 0).is_err()); assert_eq!(runtime.logs(&[], 0)?.len(), 6); assert_eq!(emitted.borrow().len(), 1011); assert_eq!(emitted.borrow().last(), retained.last()); Ok(()) } #[test] fn context_sends_copy_at_call_and_are_discarded_on_kill() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let receiver_code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { message }", )?; let receiver = runtime.spawn(receiver_code, Value::Null, capacity(4))?; let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { let data = #{items: []}; ctx.send(state, data); data.items.push(7); data }", )?; let actor = runtime.spawn(code, Value::ActorRef(receiver), capacity(2))?; runtime.send(actor, Value::Null)?; for _ in 0..100 { runtime.tick(capacity(32)); } assert_eq!( runtime.inspect(receiver)?.state, Value::Record(BTreeMap::from([("items".to_owned(), Value::List(vec![])),])) ); let runaway = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { ctx.send(state, 999); loop {} }", )?; let killed = runtime.spawn(runaway, Value::ActorRef(receiver), capacity(2))?; runtime.send(killed, Value::Null)?; for _ in 0..100 { runtime.tick(capacity(32)); } runtime.system(killed, SystemMessage::Kill)?; for _ in 0..100 { runtime.tick(capacity(32)); } assert_eq!( runtime.inspect(receiver)?.state, Value::Record(BTreeMap::from([("items".to_owned(), Value::List(vec![])),])) ); Ok(()) } #[test] fn sender_envelopes_support_replies_and_forwarding_without_message_metadata() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let sink = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { #{sender: sender, data: message} }", )?; let forward = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { ctx.send_from(sender, state, message); state }", )?; let origin = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { ctx.send(state, message); state }", )?; let sink_ref = runtime.spawn(sink, Value::Null, capacity(4))?; let forward_ref = runtime.spawn(forward, Value::ActorRef(sink_ref), capacity(4))?; let origin_ref = runtime.spawn(origin, Value::ActorRef(forward_ref), capacity(4))?; runtime.send(origin_ref, Value::String("stop".to_owned()))?; for _ in 0..100 { runtime.tick(capacity(32)); } assert_eq!( runtime.inspect(sink_ref)?.state, Value::Record(BTreeMap::from([ ("sender".to_owned(), Value::ActorRef(origin_ref)), ("data".to_owned(), Value::String("stop".to_owned())), ])) ); assert_eq!(runtime.inspect(sink_ref)?.status, Status::Idle); let reply = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { ctx.send(sender, message); state }", )?; let reply_ref = runtime.spawn(reply, Value::Null, capacity(4))?; runtime.send_from(Some(sink_ref), reply_ref, Value::Integer(42))?; for _ in 0..100 { runtime.tick(capacity(32)); } assert_eq!( runtime.inspect(sink_ref)?.state, Value::Record(BTreeMap::from([ ("sender".to_owned(), Value::ActorRef(reply_ref)), ("data".to_owned(), Value::Integer(42)), ])) ); Ok(()) } #[test] fn graceful_stop_finishes_handler_then_runs_shutdown_without_draining_mailbox() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { let n = 0; while n < 20 { n += 1; } state + message } pub fn on_shutdown(state) { if state != 1 { panic(\"wrong state\"); } state + 100 }", )?; let actor = runtime.spawn(implementation, Value::Integer(0), capacity(3))?; runtime.send(actor, Value::Integer(1))?; runtime.send(actor, Value::Integer(2))?; runtime.tick(capacity(1)); runtime.system(actor, SystemMessage::Stop)?; assert!(runtime.send(actor, Value::Null).is_err()); for _ in 0..200 { runtime.tick(capacity(32)); } assert!(runtime.inspect(actor).is_err()); assert!(runtime.system(actor, SystemMessage::Restart).is_err()); assert!( runtime .drain_events() .iter() .any(|event| matches!(event, Event::Stopped(id) if *id == actor)) ); let self_send = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { ctx.send(message, 42); state + 1 } pub fn on_shutdown(state) { state + 100 }", )?; let actor = runtime.spawn(self_send, Value::Integer(0), capacity(2))?; runtime.send(actor, Value::ActorRef(actor))?; runtime.tick(capacity(1)); runtime.system(actor, SystemMessage::Stop)?; for _ in 0..100 { runtime.tick(capacity(32)); } assert!(runtime.inspect(actor).is_err()); Ok(()) } #[test] fn kill_skips_shutdown_and_reload_waits_for_active_handler() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let first = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state + message } pub fn on_shutdown(state) { panic(\"must not run\") }", )?; let second = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state + message * 10 }", )?; let actor = runtime.spawn(first, Value::Integer(0), capacity(3))?; runtime.send(actor, Value::Integer(1))?; runtime.send(actor, Value::Integer(2))?; runtime.tick(capacity(1)); runtime.system(actor, SystemMessage::Reload(second))?; for _ in 0..100 { runtime.tick(capacity(32)); } assert_eq!(runtime.inspect(actor)?.state, Value::Integer(21)); runtime.system(actor, SystemMessage::Restart)?; assert_eq!(runtime.inspect(actor)?.implementation, second); assert_eq!(runtime.inspect(actor)?.state, Value::Integer(21)); runtime.system(actor, SystemMessage::Kill)?; assert!(runtime.inspect(actor).is_err()); Ok(()) } #[test] fn absent_and_failing_shutdown_hooks_are_explicit() -> ExperimentResult<()> { let mut runtime = Runtime::default(); for (hook, failed) in [ ("", false), ( "pub fn on_shutdown(state) { panic(\"shutdown failed\") }", true, ), ] { let implementation = publish( &mut runtime, &format!("pub fn handle(ctx, state, sender, message) {{ state }} {hook}"), )?; let actor = runtime.spawn(implementation, Value::Integer(7), capacity(1))?; runtime.system(actor, SystemMessage::Stop)?; for _ in 0..100 { runtime.tick(capacity(32)); } if failed { let snapshot = runtime.inspect(actor)?; assert_eq!(snapshot.state, Value::Integer(7)); assert!(matches!(snapshot.status, Status::Failed(_))); runtime.kill(actor)?; assert!(runtime.inspect(actor).is_err()); } else { assert!(runtime.inspect(actor).is_err()); } } Ok(()) } fn test_message(kind: i64, value: i64, target: Option) -> Value { Value::Record(std::collections::BTreeMap::from([ ("kind".to_owned(), Value::Integer(kind)), ("value".to_owned(), Value::Integer(value)), ( "target".to_owned(), target.map_or(Value::Null, Value::ActorRef), ), ])) } use runa_code::CodeStore; #[test] fn nested_state_and_messages_are_private_and_sends_copy_owned_values() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let receiver_code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { message.items.push(9); message }", )?; let sender_code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state.items.push(message.text); ctx.send(message.target, state); state }", )?; let initial = Value::Record(BTreeMap::from([( "items".to_owned(), Value::List(vec![Value::Bool(true), Value::Null]), )])); let receiver = runtime.spawn(receiver_code, Value::Null, capacity(2))?; let sender = runtime.spawn(sender_code, initial.clone(), capacity(2))?; let message = Value::Record(BTreeMap::from([ ("target".to_owned(), Value::ActorRef(receiver)), ("text".to_owned(), Value::String("hello world".to_owned())), ])); runtime.send(sender, message.clone())?; runtime.tick(capacity(1)); assert_eq!(runtime.inspect(sender)?.state, initial); assert_eq!(runtime.inspect(sender)?.reserved, Some(message)); for _ in 0..100 { runtime.tick(capacity(32)); } let sender_state = Value::Record(BTreeMap::from([( "items".to_owned(), Value::List(vec![ Value::Bool(true), Value::Null, Value::String("hello world".to_owned()), ]), )])); let receiver_state = Value::Record(BTreeMap::from([( "items".to_owned(), Value::List(vec![ Value::Bool(true), Value::Null, Value::String("hello world".to_owned()), Value::Integer(9), ]), )])); assert_eq!(runtime.inspect(sender)?.state, sender_state); assert_eq!(runtime.inspect(receiver)?.state, receiver_state); Ok(()) } #[test] fn mutated_drafts_and_invalid_structured_effects_never_partially_commit() -> ExperimentResult<()> { for source in [ "pub fn handle(ctx, state, sender, message) { state.items.push(2); panic(\"failed\"); state }", "pub fn handle(ctx, state, sender, message) { state.items.push(2); ctx.send(message, state); ctx.send(\"not a reference\", state); state }", "pub fn handle(ctx, state, sender, message) { state.items.push(2); ctx.send(message, state); ctx.send(message, 1.5); state }", "pub fn handle(ctx, state, sender, message) { state.items.push(2); ctx.send(message, state); Some(state) }", "pub fn handle(ctx, state, sender, message) { state.items.push(state); state }", ] { let mut runtime = Runtime::default(); let code = publish(&mut runtime, source)?; let initial = Value::Record(BTreeMap::from([( "items".to_owned(), Value::List(vec![Value::Integer(1)]), )])); let actor = runtime.spawn(code, initial.clone(), capacity(4))?; runtime.send(actor, Value::ActorRef(actor))?; for _ in 0..100 { runtime.tick(capacity(32)); } let snapshot = runtime.inspect(actor)?; assert_eq!(snapshot.state, initial, "{source}"); assert_eq!(snapshot.reserved, Some(Value::ActorRef(actor)), "{source}"); assert!(snapshot.queued.is_empty(), "{source}"); assert!(matches!(snapshot.status, Status::Failed(_)), "{source}"); } Ok(()) } #[test] fn invalid_input_and_migration_values_leave_the_image_unchanged() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { message }", )?; let invalid = Value::List(vec![Value::Null; MAX_VALUE_NODES]); assert!(runtime.spawn(code, invalid.clone(), capacity(2)).is_err()); assert!(runtime.actors().is_empty()); let actor = runtime.spawn(code, Value::Bool(true), capacity(2))?; assert!(runtime.send(actor, invalid.clone()).is_err()); assert!( runtime .install_with_migration(actor, code, |_| Ok(invalid)) .is_err() ); let snapshot = runtime.inspect(actor)?; assert_eq!(snapshot.state, Value::Bool(true)); assert!(snapshot.queued.is_empty()); Ok(()) } fn capacity(n: usize) -> NonZeroUsize { NonZeroUsize::new(n).unwrap() } fn publish(runtime: &mut Runtime, source: &str) -> ExperimentResult { runtime.publish(CodeStore::default().compile(source)?) } #[test] fn two_actors_isolation_and_effects() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { if message.kind == 1 { ctx.send(message.target, #{kind: 0, value: 5, target: ()}); state + 1 } else { state + message.value } }", )?; let first = runtime.spawn(code, Value::Integer(0), capacity(4))?; let second = runtime.spawn(code, Value::Integer(10), capacity(4))?; runtime.send(first, test_message(1, 0, Some(second)))?; for _ in 0..100 { runtime.tick(capacity(10)); } assert_eq!(runtime.inspect(first)?.state, Value::Integer(1)); assert_eq!(runtime.inspect(second)?.state, Value::Integer(15)); assert!(runtime.inspect(first)?.reserved.is_none()); assert!( runtime .drain_events() .iter() .any(|e| matches!(e, Event::Committed(id) if *id == second)) ); Ok(()) } #[test] fn runaway_does_not_starve_healthy_actor() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let runaway = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { loop {} }", )?; let healthy = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state + message.value }", )?; let first = runtime.spawn(runaway, Value::Integer(0), capacity(1))?; let second = runtime.spawn(healthy, Value::Integer(0), capacity(2))?; runtime.send(first, test_message(0, 1, None))?; runtime.send(second, test_message(0, 42, None))?; for _ in 0..100 { runtime.tick(capacity(5)); } assert_eq!(runtime.inspect(first)?.status, Status::Running); assert_eq!(runtime.inspect(first)?.state, Value::Integer(0)); assert_eq!(runtime.inspect(second)?.state, Value::Integer(42)); assert!(runtime.send(first, test_message(0, 2, None)).is_err()); runtime.restart(first)?; assert_eq!(runtime.inspect(first)?.incarnation, 2); assert_eq!(runtime.inspect(first)?.queued.len(), 1); runtime.kill(first)?; assert!(runtime.inspect(first).is_err()); Ok(()) } #[test] fn execution_snapshot_reports_live_progress_and_is_stable_while_paused() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { loop {} }", )?; let actor = runtime.spawn(implementation, Value::Integer(0), capacity(2))?; runtime.send(actor, test_message(0, 1, None))?; runtime.tick(capacity(3)); let running = runtime.inspect(actor)?; assert_eq!(running.status, Status::Running); let progress = running .execution .ok_or("running actor has no execution snapshot")?; assert!(progress.vm_steps > 0); assert!(running.reserved.is_some()); runtime.pause(actor)?; for _ in 0..5 { runtime.tick(capacity(10)); } let paused = runtime.inspect(actor)?; assert_eq!(paused.status, Status::Paused); assert_eq!(paused.execution, Some(progress)); runtime.cancel(actor)?; let cancelled = runtime.inspect(actor)?; assert_eq!(cancelled.execution, None); assert_eq!(cancelled.queued, vec![test_message(0, 1, None)]); Ok(()) } #[test] fn invalid_effect_does_not_commit_state_or_partial_sends() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { ctx.send(message.target, #{kind: 0, value: 1, target: ()}); ctx.send(\"invalid\", #{kind: 0, value: 2, target: ()}); 99 }", )?; let actor = runtime.spawn(code, Value::Integer(5), capacity(4))?; runtime.send(actor, test_message(0, 0, Some(actor)))?; for _ in 0..100 { runtime.tick(capacity(10)); } let snapshot = runtime.inspect(actor)?; assert_eq!(snapshot.state, Value::Integer(5)); assert!(matches!(snapshot.status, Status::Failed(_))); assert!(snapshot.queued.is_empty()); assert!(snapshot.reserved.is_some()); Ok(()) } #[test] fn awaiting_retains_message_and_state_until_woken() -> ExperimentResult<()> { use std::future::Future; use std::pin::Pin; use std::sync::Mutex; use std::task::Poll; #[derive(Default)] struct Gate { ready: AtomicBool, dropped: AtomicBool, polls: std::sync::atomic::AtomicUsize, waker: Mutex>, } struct Wait(Arc); impl Future for Wait { type Output = i64; fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { self.0.polls.fetch_add(1, Ordering::SeqCst); if self.0.ready.load(Ordering::SeqCst) { Poll::Ready(7) } else { *self.0.waker.lock().unwrap() = Some(cx.waker().clone()); Poll::Pending } } } impl Drop for Wait { fn drop(&mut self) { self.0.dropped.store(true, Ordering::SeqCst); } } let gate = Arc::new(Gate::default()); let native_gate = Arc::clone(&gate); let mut module = rune::Module::new(); module .function("wait", move || Wait(Arc::clone(&native_gate))) .build()?; let mut runtime = Runtime::default(); let code = runtime.publish( CodeStore::default().compile_with_module( "pub async fn handle(ctx, state, sender, message) { if message.kind == 1 { let x = wait().await; state + x } else { state + message.value } }", "test-gate:v1", module, )?, )?; let actor = runtime.spawn(code, Value::Integer(10), capacity(2))?; runtime.send(actor, test_message(1, 0, None))?; runtime.send(actor, test_message(0, 3, None))?; for _ in 0..100 { runtime.tick(capacity(5)); } let pending = runtime.inspect(actor)?; assert_eq!(pending.status, Status::Awaiting); assert_eq!(pending.state, Value::Integer(10)); assert_eq!(pending.queued.len(), 1); assert_eq!(pending.reserved, Some(test_message(1, 0, None))); let polls = gate.polls.load(Ordering::SeqCst); for _ in 0..10 { 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() .unwrap() .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)); } let completed = runtime.inspect(actor)?; assert_eq!(completed.state, Value::Integer(20)); assert_eq!(completed.status, Status::Idle); assert!(completed.reserved.is_none()); assert!(completed.queued.is_empty()); gate.ready.store(false, Ordering::SeqCst); gate.dropped.store(false, Ordering::SeqCst); runtime.send(actor, test_message(1, 0, None))?; for _ in 0..100 { runtime.tick(capacity(5)); } assert_eq!(runtime.inspect(actor)?.status, Status::Awaiting); runtime.cancel(actor)?; assert!(gate.dropped.load(Ordering::SeqCst)); let cancelled = runtime.inspect(actor)?; assert_eq!(cancelled.mailbox, pending.mailbox); assert_eq!(cancelled.state, Value::Integer(20)); assert_eq!(cancelled.status, Status::Idle); assert_eq!(cancelled.queued, vec![test_message(1, 0, None)]); assert!(cancelled.reserved.is_none()); 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(ctx, state, sender, message) { state + message.value }", )?; let actor = runtime.spawn(implementation, Value::Integer(10), capacity(2))?; runtime.send(actor, test_message(0, 7, None))?; runtime.tick(capacity(1)); let before_pause = runtime.inspect(actor)?; assert_eq!(before_pause.status, Status::Running); assert_eq!(before_pause.state, Value::Integer(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, Value::Integer(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(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state / message.value }", )?; let actor = runtime.spawn(code, Value::Integer(5), capacity(1))?; runtime.send(actor, test_message(0, 0, None))?; assert!(runtime.send(actor, test_message(0, 1, None)).is_err()); assert!(runtime.inspect(ActorRef::new()).is_err()); for _ in 0..100 { runtime.tick(capacity(5)); } assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_))); assert_eq!(runtime.inspect(actor)?.state, Value::Integer(5)); assert!( runtime .drain_events() .iter() .any(|e| matches!(e, Event::Failed(id, _) if *id == actor)) ); Ok(()) } #[test] fn aggregate_self_sends_respect_capacity_without_partial_commit() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { ctx.send(message.target, #{kind: 0, value: 1, target: ()}); ctx.send(message.target, #{kind: 0, value: 2, target: ()}); 99 }", )?; let actor = runtime.spawn(code, Value::Integer(5), capacity(1))?; runtime.send(actor, test_message(0, 0, Some(actor)))?; for _ in 0..100 { runtime.tick(capacity(5)); } let snapshot = runtime.inspect(actor)?; assert!(matches!(snapshot.status, Status::Failed(_))); assert_eq!(snapshot.state, Value::Integer(5)); assert!(snapshot.queued.is_empty()); assert!(snapshot.reserved.is_some()); Ok(()) } #[test] fn invalid_transition_shape_fails_without_committing() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { Some(99) }", )?; let actor = runtime.spawn(code, Value::Integer(5), capacity(1))?; runtime.send(actor, test_message(0, 1, None))?; for _ in 0..100 { runtime.tick(capacity(5)); } let snapshot = runtime.inspect(actor)?; assert!(matches!(snapshot.status, Status::Failed(_))); assert_eq!(snapshot.state, Value::Integer(5)); assert!(snapshot.reserved.is_some()); Ok(()) } #[test] fn behavior_installation_waits_for_handler_boundary() -> ExperimentResult<()> { let mut code = CodeStore::default(); let first = code.compile("pub fn handle(ctx, state, sender, message) { state + message.value }")?; let second = code .compile("pub fn handle(ctx, state, sender, message) { state + message.value * 10 }")?; let first_id = first.implementation_id; let second_id = second.implementation_id; let mut runtime = Runtime::default(); runtime.publish(first)?; runtime.publish(second)?; let actor = runtime.spawn(first_id, Value::Integer(0), capacity(2))?; runtime.send(actor, test_message(0, 1, None))?; runtime.tick(capacity(1)); let active = runtime.inspect(actor)?; assert_eq!(active.implementation, first_id); assert!(runtime.install(actor, second_id).is_err()); for _ in 0..100 { runtime.tick(capacity(10)); } assert_eq!(runtime.inspect(actor)?.state, Value::Integer(1)); runtime.send(actor, test_message(0, 2, None))?; runtime.install(actor, second_id)?; assert_eq!(runtime.inspect(actor)?.implementation, second_id); for _ in 0..100 { runtime.tick(capacity(10)); } let updated = runtime.inspect(actor)?; assert_eq!(updated.state, Value::Integer(21)); assert_eq!(updated.implementation, second_id); assert!( runtime .drain_events() .iter() .any(|event| matches!(event, Event::BehaviorInstalled(id, version) if *id == actor && *version == second_id)) ); Ok(()) } #[test] fn behavior_installation_migrates_state_atomically() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state + message.value }", )?; let new_implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state * message.value }", )?; let actor = runtime.spawn(old_implementation, Value::Integer(7), capacity(2))?; runtime.install_with_migration(actor, new_implementation, |state| { let Value::Integer(state) = state else { return Err("expected integer".into()); }; Ok(Value::Integer( state.checked_mul(10).ok_or("state migration overflow")?, )) })?; let migrated = runtime.inspect(actor)?; assert_eq!(migrated.implementation, new_implementation); assert_eq!(migrated.state, Value::Integer(70)); assert!( runtime .drain_events() .iter() .any(|event| matches!(event, Event::StateMigrated(id, Value::Integer(7), Value::Integer(70)) if *id == actor)) ); 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(ctx, state, sender, message) { state + message.value }", )?; let new_implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state * 10 + message.value }", )?; let actor = runtime.spawn(old_implementation, Value::Integer(0), capacity(4))?; runtime.send(actor, test_message(0, 1, None))?; runtime.send(actor, test_message(0, 2, None))?; 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, Value::Integer(0)); assert_eq!( cancelled.queued, vec![test_message(0, 1, None), test_message(0, 2, None)] ); assert!(cancelled.reserved.is_none()); runtime.install(actor, new_implementation)?; runtime.send(actor, test_message(0, 3, None))?; for _ in 0..100 { runtime.tick(capacity(10)); } let completed = runtime.inspect(actor)?; assert_eq!(completed.state, Value::Integer(123)); assert_eq!(completed.implementation, new_implementation); assert!( runtime .drain_events() .iter() .any(|event| matches!(event, Event::InvocationCancelled(id) if *id == actor)) ); Ok(()) } #[test] fn failed_state_migration_preserves_actor_state_and_implementation() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state + message.value }", )?; let new_implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state * message.value }", )?; let actor = runtime.spawn(old_implementation, Value::Integer(7), capacity(2))?; assert!( runtime .install_with_migration(actor, new_implementation, |_| { Err("unsupported state version".into()) }) .is_err() ); let unchanged = runtime.inspect(actor)?; assert_eq!(unchanged.implementation, old_implementation); assert_eq!(unchanged.state, Value::Integer(7)); assert!( !runtime .drain_events() .iter() .any(|event| matches!(event, Event::BehaviorInstalled(id, _) if *id == actor)) ); Ok(()) } #[test] fn migration_is_not_run_outside_an_idle_handler_boundary() -> ExperimentResult<()> { use std::cell::Cell; let mut runtime = Runtime::default(); let old_implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state + message.value }", )?; let new_implementation = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state * message.value }", )?; let actor = runtime.spawn(old_implementation, Value::Integer(7), capacity(2))?; runtime.send(actor, test_message(0, 1, None))?; runtime.tick(capacity(1)); let called = Cell::new(false); assert!( runtime .install_with_migration(actor, new_implementation, |state| { called.set(true); let Value::Integer(state) = state else { return Err("expected integer".into()); }; Ok(Value::Integer(state + 100)) }) .is_err() ); assert!(!called.get()); assert_eq!(runtime.inspect(actor)?.implementation, old_implementation); Ok(()) } #[test] fn restart_retries_failed_reservation_from_last_committed_state() -> ExperimentResult<()> { use std::sync::atomic::AtomicBool; let fail_once = Arc::new(AtomicBool::new(true)); let native_fail_once = Arc::clone(&fail_once); let mut module = rune::Module::new(); module .function("fail_once", move || { native_fail_once.swap(false, Ordering::SeqCst) }) .build()?; let mut runtime = Runtime::default(); let code = runtime.publish( CodeStore::default().compile_with_module( "fn check() { if fail_once() { 1 / 0 } else { 0 } } pub fn handle(ctx, state, sender, message) { let _ = check(); state + message.value }", "test-fail-once:v1", module, )?, )?; let actor = runtime.spawn(code, Value::Integer(10), capacity(3))?; let first = test_message(1, 2, None); let second = test_message(1, 3, None); runtime.send(actor, first.clone())?; runtime.send(actor, second.clone())?; for _ in 0..100 { runtime.tick(capacity(5)); } let failed = runtime.inspect(actor)?; assert!(matches!(failed.status, Status::Failed(_))); assert_eq!(failed.state, Value::Integer(10)); assert_eq!(failed.reserved, Some(first.clone())); assert_eq!(failed.queued, vec![second.clone()]); assert!(runtime.send(actor, test_message(1, 4, None)).is_err()); assert!( runtime .drain_events() .iter() .any(|event| matches!(event, Event::Failed(id, _) if *id == actor)) ); runtime.restart(actor)?; let restarted = runtime.inspect(actor)?; assert_eq!(restarted.actor_ref, failed.actor_ref); assert_eq!(restarted.mailbox, failed.mailbox); assert_eq!(restarted.implementation, failed.implementation); assert_eq!(restarted.incarnation, failed.incarnation + 1); assert_eq!(restarted.state, Value::Integer(10)); assert_eq!(restarted.queued, vec![first.clone(), second.clone()]); assert_eq!(restarted.reserved, None); for _ in 0..100 { runtime.tick(capacity(5)); } let completed = runtime.inspect(actor)?; assert_eq!(completed.state, Value::Integer(15), "{completed:?}"); assert_eq!(completed.status, Status::Idle); assert_eq!(completed.queued, Vec::::new()); assert_eq!(completed.reserved, None); Ok(()) } #[test] fn restart_preserves_reserved_and_queued_fifo_order() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let code = publish( &mut runtime, "pub fn handle(ctx, state, sender, message) { state * 10 + message.value }", )?; let actor = runtime.spawn(code, Value::Integer(0), capacity(3))?; let first = test_message(0, 1, None); let second = test_message(0, 2, None); let third = test_message(0, 3, None); runtime.send(actor, first.clone())?; runtime.send(actor, second.clone())?; runtime.send(actor, third.clone())?; runtime.tick(capacity(1)); let running = runtime.inspect(actor)?; assert_eq!(running.status, Status::Running); assert_eq!(running.state, Value::Integer(0)); assert_eq!(running.reserved, Some(first.clone())); assert_eq!(running.queued, vec![second.clone(), third.clone()]); runtime.restart(actor)?; let restarted = runtime.inspect(actor)?; assert_eq!(restarted.state, Value::Integer(0)); assert_eq!(restarted.reserved, None); assert_eq!( restarted.queued, vec![first.clone(), second.clone(), third.clone()] ); for _ in 0..100 { runtime.tick(capacity(5)); } let completed = runtime.inspect(actor)?; assert_eq!(completed.state, Value::Integer(123)); assert_eq!(completed.status, Status::Idle); assert_eq!(completed.queued, Vec::::new()); assert!(completed.reserved.is_none()); Ok(()) } #[test] fn context_this_sends_to_self_and_cannot_be_reassigned() -> ExperimentResult<()> { let mut runtime = Runtime::default(); let implementation = publish( &mut runtime, r#" pub fn handle(ctx, state, sender, message) { if message == 1 { ctx.send(ctx.this, 2); } state + message } "#, )?; let actor = runtime.spawn(implementation, Value::Integer(0), capacity(4))?; runtime.send(actor, Value::Integer(1))?; for _ in 0..20 { runtime.tick(capacity(32)); } assert_eq!(runtime.inspect(actor)?.state, Value::Integer(3)); assert_eq!(runtime.inspect(actor)?.status, Status::Idle); let invalid = publish( &mut runtime, r#" pub fn handle(ctx, state, sender, message) { ctx.this = sender; state } "#, )?; let actor = runtime.spawn(invalid, Value::Null, capacity(2))?; runtime.send(actor, Value::Null)?; for _ in 0..10 { runtime.tick(capacity(32)); } assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_))); Ok(()) } }