diff --git a/crates/temper-runtime/src/scheduler/core.rs b/crates/temper-runtime/src/scheduler/core.rs index 75b5cf3b4..99855487a 100644 --- a/crates/temper-runtime/src/scheduler/core.rs +++ b/crates/temper-runtime/src/scheduler/core.rs @@ -135,14 +135,23 @@ impl SimScheduler { }); } - /// Advance one tick: deliver all messages due at current_time + 1. - /// Returns the messages delivered this tick. - pub fn tick(&mut self) -> Vec { + /// Advance one tick: enqueue all messages due at the new current time + /// into their target mailboxes. + /// + /// ARN-236 single-ownership contract: `tick` advances logical time and + /// enqueues ONLY — it does not hand deliveries to the caller. The one + /// consumption path is [`Self::drain_ready`] (or [`Self::receive`] for a + /// single actor), which REMOVES messages from mailboxes. A message is + /// therefore owned by exactly one place at every instant: the pending + /// queue, a mailbox, or the consumer that drained it. (Previously `tick` + /// both enqueued and returned clones; drivers processed the clones, + /// mailboxes grew forever, and deliveries surfaced by a discarded tick + /// were lost.) + pub fn tick(&mut self) { self.current_time += 1; self.ticks += 1; - let mut delivered_this_tick = Vec::new(); - // Deliver all messages due at or before current time + // Enqueue all messages due at or before current time while let Some(msg) = self.pending.peek() { if msg.deliver_at <= self.current_time { let msg = self.pending.pop().unwrap(); // ci-ok: guarded by peek() above @@ -152,9 +161,8 @@ impl SimScheduler { let actor_state = self.actor_states.get(&to).cloned(); match actor_state { Some(SimActorState::Running) => { - self.mailboxes.entry(to).or_default().push_back(msg.clone()); - delivered_this_tick.push(msg.clone()); - self.delivered.push(msg); + self.delivered.push(msg.clone()); + self.mailboxes.entry(to).or_default().push_back(msg); } Some(SimActorState::Crashed) => { // Actor is crashed — message is lost (or could be re-queued) @@ -175,7 +183,7 @@ impl SimScheduler { } } - // Maybe crash an actor after delivery + // Maybe crash an actor after enqueue if self.rng.chance(self.fault_config.actor_crash_prob) { let running: Vec = self .actor_states @@ -189,8 +197,25 @@ impl SimScheduler { .insert(running[idx].clone(), SimActorState::Crashed); } } + } - delivered_this_tick + /// Remove and return every queued message, in deterministic order + /// (actor id order, FIFO within each mailbox). + /// + /// This is the single delivery-consumption path for drivers (ARN-236): + /// a drained message has left the scheduler entirely and is applied by + /// the caller exactly once. + /// + /// `drain_ready` does not consult `actor_states`: a driver that ticks + /// several times before draining would apply messages to actors that + /// crashed after enqueue. Both current drivers drain immediately after + /// every tick, so the drained set is exactly that tick's enqueues. + pub fn drain_ready(&mut self) -> Vec { + let mut ready = Vec::new(); + for queue in self.mailboxes.values_mut() { + ready.extend(queue.drain(..)); + } + ready } /// Take the next message from an actor's mailbox. @@ -204,6 +229,11 @@ impl SimScheduler { } /// Run until quiescent or max ticks reached. Returns total ticks. + /// + /// With enqueue-only `tick` (ARN-236) nothing drains inside this loop, + /// so once any message is enqueued it terminates via the tick bound, + /// never via quiescence. Its callers (scheduler unit tests) drain via + /// [`Self::receive`]/[`Self::drain_ready`] after it returns. pub fn run_until_quiescent(&mut self, max_ticks: u64) -> u64 { for _ in 0..max_ticks { if self.is_quiescent() { diff --git a/crates/temper-runtime/src/scheduler/sim_actor_system.rs b/crates/temper-runtime/src/scheduler/sim_actor_system.rs index 8009277c4..f33d8c229 100644 --- a/crates/temper-runtime/src/scheduler/sim_actor_system.rs +++ b/crates/temper-runtime/src/scheduler/sim_actor_system.rs @@ -17,7 +17,7 @@ use super::clock::{LogicalClock, SimClock}; use super::context::{SimContextGuard, install_sim_context}; use super::id_gen::DeterministicIdGen; use super::sim_handler::SimActorHandler; -use super::{DeterministicRng, FaultConfig, SimScheduler}; +use super::{DeterministicRng, FaultConfig, SimMessage, SimScheduler}; /// Configures how integration callbacks are delivered in simulation. /// @@ -355,6 +355,49 @@ impl SimActorSystem { // Random Mode // =================================================================== + /// Apply one drained message to its actor: run the handler, record the + /// transition, check invariants, and schedule integration callbacks. + /// The exactly-once application step of the ARN-236 ownership model. + fn apply_delivered_message(&mut self, msg: &SimMessage) { + if let Some(handler) = self.actors.get_mut(&msg.to) { + let status_before = handler.current_status(); + + match handler.handle_message(&msg.msg_type, &msg.payload) { + Ok(_) => { + let status_after = handler.current_status(); + let item_count = handler.current_item_count(); + let tick = self.clock.tick(); + *self.action_counts.get_mut(&msg.to).unwrap() += 1; // ci-ok: actor always in action_counts + self.total_transitions += 1; + + // Record the transition + self.recorded_transitions.push(( + tick, + msg.to.clone(), + msg.msg_type.clone(), + status_before.clone(), + status_after.clone(), + )); + + self.check_invariants( + &msg.to, + &msg.msg_type, + &status_before, + &status_after, + item_count, + tick, + ); + + // Schedule integration callbacks for any custom effects + self.schedule_integration_callbacks(&msg.to); + } + Err(_) => { + // Action failed — expected for invalid transitions + } + } + } + } + /// Run random exploration with fault injection. /// /// The RNG picks actors and actions. The scheduler delays/drops/crashes. @@ -394,57 +437,38 @@ impl SimActorSystem { self.scheduler.send("sim-driver", &actor_id, &action, "{}"); self.total_messages += 1; - let delivered = self.scheduler.tick(); + self.scheduler.tick(); self.clock.advance(); - // Process delivered messages + // Single-ownership delivery (ARN-236): drain_ready removes the + // messages from their mailboxes; each is applied exactly once. + let delivered = self.scheduler.drain_ready(); for msg in &delivered { - if let Some(handler) = self.actors.get_mut(&msg.to) { - let status_before = handler.current_status(); - - match handler.handle_message(&msg.msg_type, &msg.payload) { - Ok(_) => { - let status_after = handler.current_status(); - let item_count = handler.current_item_count(); - let tick = self.clock.tick(); - *self.action_counts.get_mut(&msg.to).unwrap() += 1; // ci-ok: actor always in action_counts - self.total_transitions += 1; - - // Record the transition - self.recorded_transitions.push(( - tick, - msg.to.clone(), - msg.msg_type.clone(), - status_before.clone(), - status_after.clone(), - )); - - self.check_invariants( - &msg.to, - &msg.msg_type, - &status_before, - &status_after, - item_count, - tick, - ); - - // Schedule integration callbacks for any custom effects - self.schedule_integration_callbacks(&msg.to); - } - Err(_) => { - // Action failed — expected for invalid transitions - } - } - } + self.apply_delivered_message(msg); } // Deliver any pending integration callbacks if !self.pending_integration_callbacks.is_empty() { self.deliver_integration_callbacks(); } + } - // Drain any remaining scheduled messages + // Flush the schedule: delay faults can push deliveries past the last + // driver iteration. Budgeted (never unbounded), and every flushed + // message goes through the same exactly-once drain path — previously + // a bare tick() discarded these deliveries entirely (ARN-236). + let mut flush_budget = self.config.max_ticks; + while !self.scheduler.is_quiescent() && flush_budget > 0 { + flush_budget -= 1; self.scheduler.tick(); + self.clock.advance(); + let delivered = self.scheduler.drain_ready(); + for msg in &delivered { + self.apply_delivered_message(msg); + } + if !self.pending_integration_callbacks.is_empty() { + self.deliver_integration_callbacks(); + } } let actor_states: Vec<_> = self @@ -569,12 +593,32 @@ impl SimActorSystem { } /// Deliver any pending integration callbacks by executing them as actions. + /// + /// A rejected callback is part of the simulation result (ARN-236): the + /// configured callback action failing to apply means the simulated + /// integration contract is broken, and the run must say so rather than + /// stay green. fn deliver_integration_callbacks(&mut self) { let callbacks: Vec<(String, String)> = self.pending_integration_callbacks.drain(..).collect(); for (actor_id, callback_action) in callbacks { // Execute the callback as a regular step (this checks invariants too) - let _ = self.step(&actor_id, &callback_action, "{}"); + if let Err(error) = self.step(&actor_id, &callback_action, "{}") { + let tick = self.clock.tick(); + let status = self + .actors + .get(&actor_id) + .map(|h| h.current_status()) + .unwrap_or_default(); + self.violations.push(ActorInvariantViolation { + actor_id, + action: callback_action, + status_before: status.clone(), + status_after: status, + description: format!("integration callback rejected: {error}"), + tick, + }); + } } } @@ -705,8 +749,172 @@ fn evaluate_spec_assert( #[cfg(test)] mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + use super::*; + // ── ARN-236: delayed-message ownership properties ───────────────────── + // + // Scheduler::tick() both enqueues a due message into the target mailbox + // AND returns a clone; the drivers process the returned clones and never + // drain mailboxes, and each loop iteration ends with a bare tick() whose + // returned deliveries are discarded. Consequences these tests pin: + // processed messages remain queued forever, deliveries surfaced only by + // the trailing tick are never applied, and a failing integration + // callback still yields a green run. + + /// Accepts every action, counts applications, and (optionally) emits a + /// callback trigger whose configured action always fails. + struct CountingHandler { + applications: Arc, + emit_trigger: bool, + fired: std::cell::Cell, + } + + impl SimActorHandler for CountingHandler { + fn init(&mut self) -> Result { + Ok(serde_json::json!({"status": "Ready"})) + } + fn handle_message( + &mut self, + action: &str, + _params: &str, + ) -> Result { + if action == "AlwaysFails" { + return Err("callback action rejected".to_string()); + } + self.applications.fetch_add(1, Ordering::SeqCst); + if self.emit_trigger { + self.fired.set(true); + } + Ok(serde_json::json!({"status": "Ready"})) + } + fn current_status(&self) -> String { + "Ready".to_string() + } + fn current_item_count(&self) -> usize { + 0 + } + fn event_count(&self) -> usize { + self.applications.load(Ordering::SeqCst) + } + fn valid_actions(&self) -> Vec { + vec!["Step".to_string()] + } + fn events_json(&self) -> serde_json::Value { + serde_json::json!([]) + } + fn pending_callbacks(&self) -> Vec { + if self.fired.take() { + vec!["boom_trigger".to_string()] + } else { + Vec::new() + } + } + } + + fn counting_system(seed: u64, faults: FaultConfig) -> (SimActorSystem, Arc) { + let applications = Arc::new(AtomicUsize::new(0)); + let config = SimActorSystemConfig { + seed, + max_ticks: 200, + faults, + max_actions_per_actor: 30, + }; + let mut system = SimActorSystem::new(config); + system.register_actor( + "counter", + Box::new(CountingHandler { + applications: applications.clone(), + emit_trigger: false, + fired: std::cell::Cell::new(false), + }), + ); + (system, applications) + } + + /// No processed message may remain queued: after a run, every mailbox is + /// empty and the scheduler is quiescent. + #[test] + fn arn236_no_processed_message_remains_queued() { + let (mut system, _applications) = counting_system(7, FaultConfig::none()); + let result = system.run_random(); + assert!(result.messages > 0, "the run must exercise messages"); + + assert_eq!( + system.scheduler.mailbox_depth("counter"), + 0, + "a processed message must not remain queued in its mailbox \ + (single-ownership: applied messages are consumed, not cloned)" + ); + assert!( + system.scheduler.is_quiescent(), + "after a fault-free run every delivered message must be consumed" + ); + } + + /// Every scheduled message is applied exactly once — deliveries surfaced + /// by the loop's trailing tick must not be silently discarded. With + /// message delays (no drops, no crashes), every sent message is + /// eventually due, so applications must equal sends across all seeds. + #[test] + fn arn236_every_scheduled_message_is_applied_exactly_once() { + for seed in 0..50u64 { + let faults = FaultConfig { + message_delay_prob: 0.5, + max_delay_ticks: 8, + message_drop_prob: 0.0, + actor_crash_prob: 0.0, + actor_restart_prob: 0.0, + }; + let (mut system, applications) = counting_system(seed, faults); + let result = system.run_random(); + let applied = applications.load(Ordering::SeqCst); + assert_eq!( + applied as u64, result.messages, + "seed {seed}: every scheduled message must be applied exactly \ + once ({} sent, {applied} applied) — a delivery surfaced only \ + by the trailing tick must not be discarded, and none may \ + apply twice", + result.messages + ); + } + } + + /// A failing integration callback must fail the run, not vanish. + #[test] + fn arn236_callback_failure_is_part_of_the_simulation_result() { + let applications = Arc::new(AtomicUsize::new(0)); + let config = SimActorSystemConfig { + seed: 11, + max_ticks: 50, + faults: FaultConfig::none(), + max_actions_per_actor: 3, + }; + let mut system = SimActorSystem::new(config); + system.set_integration_responses(SimIntegrationResponses::new().on_trigger( + "counter", + "boom_trigger", + "AlwaysFails", + )); + system.register_actor( + "counter", + Box::new(CountingHandler { + applications: applications.clone(), + emit_trigger: true, + fired: std::cell::Cell::new(false), + }), + ); + + let result = system.run_random(); + assert!( + !result.all_invariants_held || !result.violations.is_empty(), + "a rejected integration callback must surface in the simulation \ + result — a green run that silently discarded a callback failure \ + does not faithfully exercise the schedule" + ); + } + #[test] fn integration_responses_empty_returns_none() { let responses = SimIntegrationResponses::new(); diff --git a/crates/temper-verify/src/simulation.rs b/crates/temper-verify/src/simulation.rs index f68cb27a1..4e511f630 100644 --- a/crates/temper-verify/src/simulation.rs +++ b/crates/temper-verify/src/simulation.rs @@ -9,6 +9,8 @@ //! - Any failure is reproducible by replaying the same seed //! - Specification invariants are checked after every transition +use std::collections::{BTreeMap, BTreeSet}; + use temper_runtime::scheduler::{DeterministicRng, FaultConfig, SimActorState, SimScheduler}; use stateright::Model; @@ -159,29 +161,45 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes // Initialize actors let mut actor_states: Vec<(String, TemperModelState)> = Vec::new(); let mut actor_action_counts: Vec = Vec::new(); + // Statuses each actor has EVER been in (ARN-236): liveness `reaches` is + // an eventually-visited property along the trace, not a state-at-horizon + // property — a cyclic spec (Resolve → Reopen) satisfies "eventually + // resolved" the moment it visits a target, wherever the random walk + // happens to stop. (The pre-fix final-state check only passed because + // lost trailing deliveries biased where traces ended.) + let mut visited_statuses: BTreeMap> = BTreeMap::new(); for i in 0..config.num_actors { let actor_id = format!("entity-{i}"); sched.register_actor(&actor_id); let initial = model.init_states()[0].clone(); + visited_statuses + .entry(actor_id.clone()) + .or_default() + .insert(model.initial_status.clone()); actor_states.push((actor_id, initial)); actor_action_counts.push(0); } - let mut violations = Vec::new(); - let mut total_transitions: u64 = 0; + let mut run = ModelRunState { + actor_states, + actor_action_counts, + violations: Vec::new(), + total_transitions: 0, + visited_statuses, + }; let mut total_messages: u64 = 0; // Main simulation loop for tick in 0..config.max_ticks { - if actor_states.is_empty() { + if run.actor_states.is_empty() { break; } - let actor_idx = rng.next_bound(actor_states.len()); - let (ref actor_id, ref current_state) = actor_states[actor_idx]; + let actor_idx = rng.next_bound(run.actor_states.len()); + let (ref actor_id, ref current_state) = run.actor_states[actor_idx]; - if actor_action_counts[actor_idx] >= config.max_actions_per_actor { + if run.actor_action_counts[actor_idx] >= config.max_actions_per_actor { continue; } @@ -208,52 +226,45 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes ); total_messages += 1; - let delivered = sched.tick(); - - for msg in &delivered { - let target_idx = actor_states.iter().position(|(id, _)| id == &msg.to); - let Some(idx) = target_idx else { continue }; - - let (ref target_id, ref state_before) = actor_states[idx]; + sched.tick(); - let action: TemperModelAction = match serde_json::from_str(&msg.payload) { - Ok(a) => a, - Err(_) => continue, - }; + // Single-ownership delivery (ARN-236): drain_ready removes messages + // from mailboxes; each is applied to the model exactly once. + let delivered = sched.drain_ready(); - if let Some(new_state) = model.next_state(state_before, action.clone()) { - check_invariants_on_state( - model, - target_id, - &action.name, - state_before, - &new_state, - tick, - &mut violations, - ); - - actor_states[idx].1 = new_state; - actor_action_counts[idx] += 1; - total_transitions += 1; - } + for msg in &delivered { + apply_to_model(model, &mut run, tick, msg); } + } + // Flush the schedule: delay faults can push deliveries past the last + // loop iteration. Budgeted; every flushed message runs through the same + // exactly-once path (previously a bare tick() discarded them — ARN-236). + let mut flush_budget = config.max_ticks; + let mut tick = config.max_ticks.saturating_sub(1); + while !sched.is_quiescent() && flush_budget > 0 { + flush_budget -= 1; + tick += 1; sched.tick(); + for msg in &sched.drain_ready() { + apply_to_model(model, &mut run, tick, msg); + } } // Post-simulation liveness checks - let liveness_violations = check_liveness_post_simulation(model, &actor_states); + let liveness_violations = + check_liveness_post_simulation(model, &run.actor_states, &run.visited_statuses); SimulationResult { - all_invariants_held: violations.is_empty(), + all_invariants_held: run.violations.is_empty(), ticks: config.max_ticks.min(sched.current_time()), - total_transitions, + total_transitions: run.total_transitions, total_messages, total_dropped: sched.total_dropped() as u64, - violations, + violations: run.violations, liveness_violations, seed: config.seed, - actor_final_states: actor_states, + actor_final_states: run.actor_states, } } @@ -265,6 +276,7 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes fn check_liveness_post_simulation( model: &TemperModel, actor_states: &[(String, TemperModelState)], + visited_statuses: &BTreeMap>, ) -> Vec { let mut violations = Vec::new(); @@ -292,15 +304,24 @@ fn check_liveness_post_simulation( if targets.is_empty() { continue; } - // If the actor started from a "from" state, it should have - // reached a target state by the end of simulation. + // Eventually-visited along the trace (ARN-236): the actor + // satisfies `reaches` the moment it has EVER been in a + // target status. Checking only the horizon final state + // wrongly flags cyclic specs (Resolve -> Reopen) whose + // random walk stops mid-cycle — and only ever passed + // before because lost trailing deliveries biased where + // traces ended. let started_from = from.is_empty() || from.contains(&model.initial_status); - if started_from && !targets.contains(&final_state.status) { + let visited_target = visited_statuses + .get(actor_id) + .is_some_and(|seen| targets.iter().any(|t| seen.contains(t))); + if started_from && !visited_target { violations.push(LivenessViolation { actor_id: actor_id.clone(), property: live.name.clone(), description: format!( - "actor did not reach target states {:?}, stuck at '{}'", + "actor never reached target states {:?} at any point in the \ + trace, ending at '{}'", targets, final_state.status ), final_state: final_state.clone(), @@ -314,6 +335,63 @@ fn check_liveness_post_simulation( violations } +/// Mutable per-run simulation state shared by the driver loop, the flush, +/// and the post-simulation checks (ARN-236). +struct ModelRunState { + /// Current state per actor. + actor_states: Vec<(String, TemperModelState)>, + /// Applied-action count per actor (parallel to `actor_states`). + actor_action_counts: Vec, + /// Invariant violations found so far. + violations: Vec, + /// Total successful transitions. + total_transitions: u64, + /// Statuses each actor has ever been in — the eventually-visited data + /// `reaches` liveness checks against. + visited_statuses: BTreeMap>, +} + +/// Apply one drained message to the model: parse the action, advance the +/// target actor's state, check invariants, and update counters. The +/// exactly-once application step shared by the driver loop and the flush +/// (ARN-236). +fn apply_to_model( + model: &TemperModel, + run: &mut ModelRunState, + tick: u64, + msg: &temper_runtime::scheduler::SimMessage, +) { + let target_idx = run.actor_states.iter().position(|(id, _)| id == &msg.to); + let Some(idx) = target_idx else { return }; + + let (ref target_id, ref state_before) = run.actor_states[idx]; + + let action: TemperModelAction = match serde_json::from_str(&msg.payload) { + Ok(a) => a, + Err(_) => return, + }; + + if let Some(new_state) = model.next_state(state_before, action.clone()) { + check_invariants_on_state( + model, + target_id, + &action.name, + state_before, + &new_state, + tick, + &mut run.violations, + ); + + run.actor_states[idx].1 = new_state; + run.visited_statuses + .entry(run.actor_states[idx].0.clone()) + .or_default() + .insert(run.actor_states[idx].1.status.clone()); + run.actor_action_counts[idx] += 1; + run.total_transitions += 1; + } +} + /// Check invariants on a state using the model's resolved invariants. /// /// All invariant data comes from the spec — no hardcoded entity knowledge. @@ -418,6 +496,112 @@ mod tests { const ORDER_IOA: &str = include_str!("../../../test-fixtures/specs/order.ioa.toml"); + /// Cyclic spec: Resolve moves to the target, Reopen leaves it again. + /// `reaches` liveness must be satisfied by EVER visiting the target, + /// wherever the bounded random walk happens to stop (ARN-236). + const CYCLIC_IOA: &str = r#" +[automaton] +name = "Loop" +states = ["Open", "Resolved"] +initial = "Open" + +[[action]] +name = "Resolve" +kind = "input" +from = ["Open"] +to = "Resolved" +hint = "Resolve it." + +[[action]] +name = "Reopen" +kind = "input" +from = ["Resolved"] +to = "Open" +hint = "Reopen it." + +[[liveness]] +name = "EventuallyResolved" +from = ["Open"] +reaches = ["Resolved"] +"#; + + /// The target status exists but no action ever moves into it: the + /// `reaches` property is genuinely unreachable and must still violate. + const UNREACHABLE_IOA: &str = r#" +[automaton] +name = "Stuck" +states = ["Open", "Parked", "Resolved"] +initial = "Open" + +[[action]] +name = "Park" +kind = "input" +from = ["Open"] +to = "Parked" +hint = "Park it." + +[[action]] +name = "Unpark" +kind = "input" +from = ["Parked"] +to = "Open" +hint = "Unpark it." + +[[liveness]] +name = "EventuallyResolved" +from = ["Open"] +reaches = ["Resolved"] +"#; + + fn liveness_config(seed: u64) -> SimConfig { + SimConfig { + seed, + max_ticks: 200, + num_actors: 2, + max_actions_per_actor: 20, + max_counter: 2, + faults: FaultConfig::none(), + } + } + + /// ARN-236: a cyclic model that VISITS the target mid-trace satisfies + /// `reaches`, even when the walk stops outside the target. + #[test] + fn reaches_liveness_satisfied_by_ever_visiting_the_target() { + for seed in [7u64, 21, 99] { + let result = run_simulation_from_ioa(CYCLIC_IOA, &liveness_config(seed)).unwrap(); + assert!( + result.total_transitions > 0, + "seed {seed}: the walk must actually move" + ); + let reaches_violations: Vec<_> = result + .liveness_violations + .iter() + .filter(|v| v.property == "EventuallyResolved") + .collect(); + assert!( + reaches_violations.is_empty(), + "seed {seed}: a trace that visited the target must satisfy \ + `reaches`, got: {reaches_violations:?}" + ); + } + } + + /// ARN-236 (the not-weakened direction): a target no action can ever + /// enter must still violate `reaches`. + #[test] + fn reaches_liveness_still_fails_when_target_is_never_visited() { + let result = run_simulation_from_ioa(UNREACHABLE_IOA, &liveness_config(7)).unwrap(); + assert!( + result + .liveness_violations + .iter() + .any(|v| v.property == "EventuallyResolved"), + "a genuinely unreachable target must still violate, got: {:?}", + result.liveness_violations + ); + } + #[test] fn test_simulation_no_faults() { let config = SimConfig { diff --git a/docs/adrs/0165-sim-delivery-single-ownership.md b/docs/adrs/0165-sim-delivery-single-ownership.md new file mode 100644 index 000000000..44f27f0c5 --- /dev/null +++ b/docs/adrs/0165-sim-delivery-single-ownership.md @@ -0,0 +1,84 @@ +# ADR-0165: Simulation Delivery Has a Single Ownership Path + +## Status + +Accepted (2026-07-14) + +(Numbered 0165: 0156–0164 are claimed by concurrently open arena branches.) + +## Context + +The deterministic scheduler exposed two incompatible delivery contracts: +`Scheduler::tick()` both enqueued a due message into the target mailbox AND +returned a clone. The simulation drivers (temper-runtime's +`SimActorSystem::run_random`, temper-verify's model-checking driver) +processed the returned clones and never drained mailboxes — `receive()` had +zero production callers — and each loop iteration ended with a bare +`tick()` whose returned deliveries were discarded. Consequences (ARN-236, +all reproduced by seeded tests): every processed message remained queued in +its mailbox forever; deliveries surfaced only by the trailing tick (delayed +messages coming due after the last driver iteration) were silently lost — +seed 0 of the regression sweep loses 14 of 44 sends; and a rejected +integration callback (`let _ = self.step(...)`) left the run green. DST and +model-check results therefore did not faithfully exercise the schedule they +claimed to. + +## Decision + +1. **`tick()` advances logical time and enqueues only.** It returns + nothing. A message is owned by exactly one place at every instant: the + pending queue, a mailbox, or the consumer that drained it. +2. **One consumption path.** `drain_ready()` removes and returns all queued + messages in deterministic order (actor-id order, FIFO per mailbox); + `receive()` remains for single-actor consumption. Drivers apply each + drained message exactly once (`apply_delivered_message`, extracted so + the main loop and the flush share one application path). The clone-return + processing path is deleted. +3. **Budgeted schedule flush.** After the driver's action loop, a bounded + flush (budget = `max_ticks`) ticks and drains until quiescent, so + delay-faulted deliveries due after the last iteration are applied through + the same exactly-once path instead of being discarded. +4. **Callback failures are part of the result.** A rejected integration + callback is recorded as a violation (`integration callback rejected: + ...`) on the simulation result; a run that discards one cannot stay + green. Both drivers share the runtime scheduler, so the verifier and + runtime exercise the same delivery contract. + +## Consequences + +- Exactly-once delivery is now a tested property: a 50-seed sweep with delay + faults asserts applications == sends; a fault-free run asserts empty + mailboxes and quiescence; a rejected callback asserts a non-green result. +- `is_quiescent()` (pending + mailboxes empty) is now reachable in driver + runs — previously mailboxes never emptied, so quiescence was unreachable + by construction. +- Runs get slightly longer traces: deliveries that were silently lost are + now applied (that is the point). Seeds produce different — now correct — + transition sequences than before; recorded-run comparisons across this + change are not byte-compatible, which is expected for a semantics fix. +- **L2 `reaches` liveness is now eventually-visited along the trace, not + final-state-at-horizon.** The corrected traces exposed that the old check + wrongly flagged cyclic specs (the Ticket fixture's Resolve → Reopen cycle + "failed" TicketEventuallyResolved whenever the random walk stopped + mid-cycle) — it had only ever passed because lost trailing deliveries + biased where traces ended. Each actor now records every status it visits; + `reaches` is satisfied the moment a target status was ever visited. Specs + that genuinely never reach a target still fail, and `no_deadlock` + liveness is unchanged. +- `Scheduler::run_until_quiescent` keeps its shape (tick in a bounded loop), + but with tick enqueue-only nothing drains during it — once anything is + enqueued it degrades to "tick `max_ticks` times" and terminates via the + bound, never via quiescence. Its only callers are scheduler unit tests, + which drain after it returns. + +## Alternatives Considered + +- **Making `tick()` return owned (non-cloned) messages and deleting + mailboxes from the delivery path**: also a single-owner model, but it + removes per-actor mailbox semantics (depth inspection, per-actor + `receive`, crash-time mailbox behavior) that other tests and fault + injection rely on. Enqueue-then-drain keeps those observables. +- **Draining inside `tick()` (tick returns owned mailbox contents)**: + conflates time advance with consumption; a driver that wants to tick + several times before applying (e.g. coalescing) couldn't. Separate verbs + keep the contract explicit.