diff --git a/crates/temper-platform/tests/common/dst.rs b/crates/temper-platform/tests/common/dst.rs index eac244313..186bca30f 100644 --- a/crates/temper-platform/tests/common/dst.rs +++ b/crates/temper-platform/tests/common/dst.rs @@ -17,6 +17,8 @@ pub fn new_sim( max_ticks, faults, max_actions_per_actor, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }) } diff --git a/crates/temper-runtime/src/scheduler/core.rs b/crates/temper-runtime/src/scheduler/core.rs index 75b5cf3b4..3d99cb63c 100644 --- a/crates/temper-runtime/src/scheduler/core.rs +++ b/crates/temper-runtime/src/scheduler/core.rs @@ -5,6 +5,8 @@ use std::collections::{BTreeMap, BinaryHeap, VecDeque}; use super::rng::DeterministicRng; use super::types::{FaultConfig, SimActorState, SimMessage, SimTime}; +const DEFAULT_MAILBOX_BUDGET_PER_ACTOR: usize = 4_096; + /// The deterministic simulation scheduler. /// /// Drives message delivery in a controlled, reproducible order. @@ -19,6 +21,10 @@ pub struct SimScheduler { /// Per-actor mailbox of delivered (ready to process) messages. /// BTreeMap ensures deterministic iteration order. mailboxes: BTreeMap>, + /// Next mailbox position for deterministic, starvation-free draining. + next_mailbox_index: usize, + /// Maximum ready messages retained for one actor before failing fast. + mailbox_budget_per_actor: usize, /// Actor states. BTreeMap ensures deterministic iteration order /// (critical for reproducible crash selection). actor_states: BTreeMap, @@ -37,11 +43,26 @@ pub struct SimScheduler { impl SimScheduler { /// Create a new simulation scheduler with the given seed and fault config. pub fn new(seed: u64, fault_config: FaultConfig) -> Self { + Self::with_mailbox_budget(seed, fault_config, DEFAULT_MAILBOX_BUDGET_PER_ACTOR) + } + + /// Create a scheduler with an explicit per-actor ready-mailbox budget. + pub fn with_mailbox_budget( + seed: u64, + fault_config: FaultConfig, + mailbox_budget_per_actor: usize, + ) -> Self { + assert!( + mailbox_budget_per_actor > 0, + "mailbox budget per actor must be positive" + ); Self { rng: DeterministicRng::new(seed), current_time: 0, pending: BinaryHeap::new(), mailboxes: BTreeMap::new(), + next_mailbox_index: 0, + mailbox_budget_per_actor, actor_states: BTreeMap::new(), fault_config, next_msg_id: 0, @@ -135,25 +156,30 @@ 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 and enqueue every message now due in its target mailbox. + /// + /// [`drain_ready`](Self::drain_ready) is the sole ownership transfer from + /// scheduler mailboxes to a simulation driver. + 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 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 + let msg = self.pending.pop().expect("pending message was peeked"); let to = msg.to.clone(); // Check if target actor is running 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()); + let mailbox = self.mailboxes.entry(to.clone()).or_default(); + assert!( + mailbox.len() < self.mailbox_budget_per_actor, + "ready mailbox budget exhausted for actor '{to}'" + ); + mailbox.push_back(msg.clone()); self.delivered.push(msg); } Some(SimActorState::Crashed) => { @@ -189,13 +215,58 @@ impl SimScheduler { .insert(running[idx].clone(), SimActorState::Crashed); } } + } + + /// Remove up to `message_budget` ready messages in deterministic order. + /// + /// Actors are visited in cyclic lexicographic [`BTreeMap`] order and each + /// actor's messages retain FIFO order. A drained message cannot be returned + /// again, and a small budget cannot permanently starve a later mailbox. + pub fn drain_ready(&mut self, message_budget: usize) -> Vec { + assert!(message_budget > 0, "message budget must be positive"); + + let actor_ids: Vec = self.mailboxes.keys().cloned().collect(); + if actor_ids.is_empty() { + return Vec::new(); + } + + let mut ready = Vec::new(); + let mut index = self.next_mailbox_index % actor_ids.len(); + let mut empty_mailboxes_seen = 0; + while ready.len() < message_budget && empty_mailboxes_seen < actor_ids.len() { + let actor_id = &actor_ids[index]; + let mailbox = self + .mailboxes + .get_mut(actor_id) + .expect("actor id came from mailbox keys"); + if let Some(message) = mailbox.pop_front() { + ready.push(message); + empty_mailboxes_seen = 0; + } else { + empty_mailboxes_seen += 1; + } + index = (index + 1) % actor_ids.len(); + } + self.next_mailbox_index = index; - delivered_this_tick + debug_assert!(ready.len() <= message_budget); + ready } - /// Take the next message from an actor's mailbox. + /// Take the next message from one actor's mailbox. + /// + /// Simulation drivers should prefer [`drain_ready`](Self::drain_ready) so + /// ordering and budgets are shared. This actor-specific consumer remains + /// available for direct scheduler users. pub fn receive(&mut self, actor_id: &str) -> Option { - self.mailboxes.get_mut(actor_id).and_then(|q| q.pop_front()) + self.mailboxes + .get_mut(actor_id) + .and_then(VecDeque::pop_front) + } + + /// Return whether any actor mailbox owns a ready message. + pub fn has_ready_messages(&self) -> bool { + self.mailboxes.values().any(|mailbox| !mailbox.is_empty()) } /// Check if the simulation has no more pending messages. @@ -203,7 +274,10 @@ impl SimScheduler { self.pending.is_empty() && self.mailboxes.values().all(|q| q.is_empty()) } - /// Run until quiescent or max ticks reached. Returns total ticks. + /// Advance until quiescent or `max_ticks` is reached. + /// + /// Ready mailbox messages remain owned by the scheduler for a caller to + /// consume with [`receive`](Self::receive) or [`drain_ready`](Self::drain_ready). pub fn run_until_quiescent(&mut self, max_ticks: u64) -> u64 { for _ in 0..max_ticks { if self.is_quiescent() { @@ -256,269 +330,5 @@ impl SimScheduler { } #[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_basic_message_delivery() { - let mut sched = SimScheduler::new(1, FaultConfig::none()); - sched.register_actor("actor-a"); - sched.register_actor("actor-b"); - - sched.send("actor-a", "actor-b", "Ping", "{}"); - assert_eq!(sched.total_delivered(), 0); - - sched.tick(); // deliver - assert_eq!(sched.total_delivered(), 1); - - let msg = sched.receive("actor-b").unwrap(); - assert_eq!(msg.msg_type, "Ping"); - assert_eq!(msg.from, "actor-a"); - } - - #[test] - fn test_message_ordering_is_deterministic() { - // Run the same scenario twice with the same seed → same delivery order - fn run_scenario(seed: u64) -> Vec { - let mut sched = SimScheduler::new(seed, FaultConfig::light()); - sched.register_actor("a"); - sched.register_actor("b"); - - for i in 0..10 { - sched.send("a", "b", &format!("msg-{i}"), "{}"); - } - - sched.run_until_quiescent(100); - - sched - .delivered_log() - .iter() - .map(|m| m.msg_type.clone()) - .collect() - } - - let run1 = run_scenario(42); - let run2 = run_scenario(42); - assert_eq!(run1, run2, "Same seed must produce same delivery order"); - } - - #[test] - fn test_different_seeds_may_produce_different_order() { - fn run_scenario(seed: u64) -> Vec { - let mut sched = SimScheduler::new(seed, FaultConfig::light()); - sched.register_actor("a"); - sched.register_actor("b"); - - for i in 0..20 { - sched.send("a", "b", &format!("msg-{i}"), "{}"); - } - - sched.run_until_quiescent(100); - sched - .delivered_log() - .iter() - .map(|m| m.msg_type.clone()) - .collect() - } - - let run1 = run_scenario(42); - let run2 = run_scenario(999); - // With light faults (10% delay), different seeds should likely produce different orders - // This isn't guaranteed for every pair, but is overwhelmingly likely with 20 messages - assert_ne!( - run1, run2, - "Different seeds should usually produce different orders" - ); - } - - #[test] - fn test_fault_injection_message_drop() { - let config = FaultConfig { - message_drop_prob: 1.0, // Drop everything - ..FaultConfig::none() - }; - let mut sched = SimScheduler::new(42, config); - sched.register_actor("a"); - sched.register_actor("b"); - - sched.send("a", "b", "Important", "{}"); - sched.tick(); - - assert_eq!(sched.total_delivered(), 0); - assert_eq!(sched.total_dropped(), 1); - } - - #[test] - fn test_fault_injection_actor_crash() { - let config = FaultConfig { - actor_crash_prob: 1.0, // Crash after every tick - ..FaultConfig::none() - }; - let mut sched = SimScheduler::new(42, config); - sched.register_actor("a"); - sched.register_actor("b"); - - sched.send("a", "b", "msg", "{}"); - sched.tick(); - - // Message should be delivered (crash happens AFTER delivery) - assert_eq!(sched.total_delivered(), 1); - - // But one of the actors should now be crashed - let crashed = sched - .actor_states - .values() - .filter(|s| **s == SimActorState::Crashed) - .count(); - assert!(crashed > 0, "Should have at least one crashed actor"); - } - - #[test] - fn test_message_to_crashed_actor_is_dropped() { - let mut sched = SimScheduler::new(42, FaultConfig::none()); - sched.register_actor("a"); - sched.register_actor("b"); - - // Manually crash actor-b - sched - .actor_states - .insert("b".to_string(), SimActorState::Crashed); - - sched.send("a", "b", "msg", "{}"); - sched.tick(); - - assert_eq!(sched.total_delivered(), 0); - assert_eq!(sched.total_dropped(), 1); - } - - #[test] - fn test_quiescence_detection() { - let mut sched = SimScheduler::new(1, FaultConfig::none()); - sched.register_actor("a"); - - assert!(sched.is_quiescent()); - - sched.send("a", "a", "self-msg", "{}"); - assert!(!sched.is_quiescent()); - - sched.tick(); - // Message delivered to mailbox — not quiescent until consumed - sched.receive("a"); - assert!(sched.is_quiescent()); - } - - #[test] - fn test_run_until_quiescent() { - let mut sched = SimScheduler::new(1, FaultConfig::none()); - sched.register_actor("a"); - sched.register_actor("b"); - - sched.send("a", "b", "msg-1", "{}"); - sched.send("a", "b", "msg-2", "{}"); - sched.send("a", "b", "msg-3", "{}"); - - let ticks = sched.run_until_quiescent(100); - assert!(ticks <= 100); - assert_eq!(sched.total_delivered(), 3); - } - - #[test] - fn test_message_delay_increases_delivery_time() { - let config = FaultConfig { - message_delay_prob: 1.0, // Always delay - max_delay_ticks: 5, - ..FaultConfig::none() - }; - let mut sched = SimScheduler::new(42, config); - sched.register_actor("a"); - sched.register_actor("b"); - - sched.send("a", "b", "delayed", "{}"); - - // Tick 1: message not yet delivered (delayed) - sched.tick(); - let delivered_at_1 = sched.total_delivered(); - - // Run more ticks - sched.run_until_quiescent(20); - assert_eq!( - sched.total_delivered(), - 1, - "Message should eventually arrive" - ); - if delivered_at_1 == 0 { - assert!( - sched.current_time() > 1, - "Delivery should be delayed beyond tick 1" - ); - } - } - - #[test] - fn test_heavy_faults_simulation_completes() { - // Even with heavy faults, simulation should complete without panic - let mut sched = SimScheduler::new(12345, FaultConfig::heavy()); - for i in 0..5 { - sched.register_actor(&format!("actor-{i}")); - } - - // Send 50 messages between random actors - let mut rng = super::super::DeterministicRng::new(67890); - for _ in 0..50 { - let from = format!("actor-{}", rng.next_bound(5)); - let to = format!("actor-{}", rng.next_bound(5)); - sched.send(&from, &to, "msg", "{}"); - } - - sched.run_until_quiescent(200); - - // Just verify it completed without panic and some messages got through - let total = sched.total_delivered() + sched.total_dropped(); - assert!(total > 0, "Should have processed some messages"); - } - - #[test] - fn test_send_at_delivers_at_specified_time() { - let mut sched = SimScheduler::new(1, FaultConfig::none()); - sched.register_actor("a"); - sched.register_actor("b"); - - // Schedule a message at time 5 - sched.send_at("a", "b", "Scheduled", "{}", 5); - - // Ticks 1-4: nothing delivered - for _ in 1..5 { - sched.tick(); - assert_eq!( - sched.total_delivered(), - 0, - "should not deliver before deliver_at" - ); - } - - // Tick 5: message delivered - sched.tick(); - assert_eq!(sched.total_delivered(), 1); - - let msg = sched.receive("b").unwrap(); - assert_eq!(msg.msg_type, "Scheduled"); - assert_eq!(msg.deliver_at, 5); - } - - #[test] - fn test_send_at_respects_message_drop() { - let config = FaultConfig { - message_drop_prob: 1.0, - ..FaultConfig::none() - }; - let mut sched = SimScheduler::new(42, config); - sched.register_actor("a"); - sched.register_actor("b"); - - sched.send_at("a", "b", "Scheduled", "{}", 3); - sched.run_until_quiescent(10); - - assert_eq!(sched.total_delivered(), 0); - assert_eq!(sched.total_dropped(), 1); - } -} +#[path = "core/tests.rs"] +mod tests; diff --git a/crates/temper-runtime/src/scheduler/core/tests.rs b/crates/temper-runtime/src/scheduler/core/tests.rs new file mode 100644 index 000000000..83b351692 --- /dev/null +++ b/crates/temper-runtime/src/scheduler/core/tests.rs @@ -0,0 +1,357 @@ +use std::collections::BTreeSet; + +use super::*; + +fn drain_until_quiescent(sched: &mut SimScheduler, max_ticks: u64) -> u64 { + for _ in 0..max_ticks { + if sched.is_quiescent() { + break; + } + sched.tick(); + sched.drain_ready(1_024); + } + sched.ticks +} + +#[test] +fn test_basic_message_delivery() { + let mut sched = SimScheduler::new(1, FaultConfig::none()); + sched.register_actor("actor-a"); + sched.register_actor("actor-b"); + + sched.send("actor-a", "actor-b", "Ping", "{}"); + assert_eq!(sched.total_delivered(), 0); + + sched.tick(); // deliver + assert_eq!(sched.total_delivered(), 1); + + let msg = sched.drain_ready(1).pop().unwrap(); + assert_eq!(msg.msg_type, "Ping"); + assert_eq!(msg.from, "actor-a"); + assert!(sched.drain_ready(1).is_empty()); +} + +#[test] +fn test_message_ordering_is_deterministic() { + // Run the same scenario twice with the same seed → same delivery order + fn run_scenario(seed: u64) -> Vec { + let mut sched = SimScheduler::new(seed, FaultConfig::light()); + sched.register_actor("a"); + sched.register_actor("b"); + + for i in 0..10 { + sched.send("a", "b", &format!("msg-{i}"), "{}"); + } + + drain_until_quiescent(&mut sched, 100); + + sched + .delivered_log() + .iter() + .map(|m| m.msg_type.clone()) + .collect() + } + + let run1 = run_scenario(42); + let run2 = run_scenario(42); + assert_eq!(run1, run2, "Same seed must produce same delivery order"); +} + +#[test] +fn test_different_seeds_may_produce_different_order() { + fn run_scenario(seed: u64) -> Vec { + let mut sched = SimScheduler::new(seed, FaultConfig::light()); + sched.register_actor("a"); + sched.register_actor("b"); + + for i in 0..20 { + sched.send("a", "b", &format!("msg-{i}"), "{}"); + } + + drain_until_quiescent(&mut sched, 100); + sched + .delivered_log() + .iter() + .map(|m| m.msg_type.clone()) + .collect() + } + + let run1 = run_scenario(42); + let run2 = run_scenario(999); + // With light faults (10% delay), different seeds should likely produce different orders + // This isn't guaranteed for every pair, but is overwhelmingly likely with 20 messages + assert_ne!( + run1, run2, + "Different seeds should usually produce different orders" + ); +} + +#[test] +fn test_fault_injection_message_drop() { + let config = FaultConfig { + message_drop_prob: 1.0, // Drop everything + ..FaultConfig::none() + }; + let mut sched = SimScheduler::new(42, config); + sched.register_actor("a"); + sched.register_actor("b"); + + sched.send("a", "b", "Important", "{}"); + sched.tick(); + + assert_eq!(sched.total_delivered(), 0); + assert_eq!(sched.total_dropped(), 1); +} + +#[test] +fn test_fault_injection_actor_crash() { + let config = FaultConfig { + actor_crash_prob: 1.0, // Crash after every tick + ..FaultConfig::none() + }; + let mut sched = SimScheduler::new(42, config); + sched.register_actor("a"); + sched.register_actor("b"); + + sched.send("a", "b", "msg", "{}"); + sched.tick(); + + // Message should be delivered (crash happens AFTER delivery) + assert_eq!(sched.total_delivered(), 1); + + // But one of the actors should now be crashed + let crashed = sched + .actor_states + .values() + .filter(|s| **s == SimActorState::Crashed) + .count(); + assert!(crashed > 0, "Should have at least one crashed actor"); +} + +#[test] +fn test_message_to_crashed_actor_is_dropped() { + let mut sched = SimScheduler::new(42, FaultConfig::none()); + sched.register_actor("a"); + sched.register_actor("b"); + + // Manually crash actor-b + sched + .actor_states + .insert("b".to_string(), SimActorState::Crashed); + + sched.send("a", "b", "msg", "{}"); + sched.tick(); + + assert_eq!(sched.total_delivered(), 0); + assert_eq!(sched.total_dropped(), 1); +} + +#[test] +fn test_quiescence_detection() { + let mut sched = SimScheduler::new(1, FaultConfig::none()); + sched.register_actor("a"); + + assert!(sched.is_quiescent()); + + sched.send("a", "a", "self-msg", "{}"); + assert!(!sched.is_quiescent()); + + sched.tick(); + // Message delivered to mailbox — not quiescent until consumed + sched.drain_ready(1); + assert!(sched.is_quiescent()); +} + +#[test] +fn test_budgeted_drain_preserves_ready_messages() { + let mut sched = SimScheduler::new(1, FaultConfig::none()); + sched.register_actor("a"); + sched.register_actor("b"); + + sched.send("a", "b", "msg-1", "{}"); + sched.send("a", "b", "msg-2", "{}"); + sched.send("a", "b", "msg-3", "{}"); + + sched.tick(); + let first = sched.drain_ready(2); + assert_eq!(first.len(), 2); + assert_eq!(sched.mailbox_depth("b"), 1); + + let second = sched.drain_ready(2); + assert_eq!(second.len(), 1); + assert_eq!(sched.mailbox_depth("b"), 0); + assert!(sched.drain_ready(2).is_empty()); + assert_eq!(sched.total_delivered(), 3); +} + +#[test] +#[should_panic(expected = "ready mailbox budget exhausted for actor 'b'")] +fn test_ready_mailbox_budget_fails_fast() { + let mut sched = SimScheduler::with_mailbox_budget(1, FaultConfig::none(), 1); + sched.register_actor("a"); + sched.register_actor("b"); + sched.send("a", "b", "first", "{}"); + sched.send("a", "b", "second", "{}"); + sched.tick(); +} + +#[test] +fn test_drain_ready_transfers_ownership_once_in_actor_order() { + let mut sched = SimScheduler::new(1, FaultConfig::none()); + sched.register_actor("actor-b"); + sched.register_actor("actor-a"); + + sched.send("driver", "actor-b", "ForB", "{}"); + sched.send("driver", "actor-a", "ForA", "{}"); + sched.send("driver", "actor-b", "ForB2", "{}"); + sched.send("driver", "actor-a", "ForA2", "{}"); + sched.tick(); + + let owners: Vec = (0..4) + .map(|_| sched.drain_ready(1).pop().unwrap().to) + .collect(); + assert_eq!(owners, vec!["actor-a", "actor-b", "actor-a", "actor-b"]); + assert!(sched.drain_ready(2).is_empty()); + assert!(sched.is_quiescent()); +} + +#[test] +fn test_delayed_delivery_is_exactly_once_across_replay_seeds() { + for seed in 0..16 { + let mut sched = SimScheduler::new( + seed, + FaultConfig { + message_delay_prob: 1.0, + max_delay_ticks: 5, + ..FaultConfig::none() + }, + ); + sched.register_actor("actor-a"); + sched.register_actor("actor-b"); + for id in 0..32 { + let actor = if id % 2 == 0 { "actor-a" } else { "actor-b" }; + sched.send("driver", actor, "Apply", "{}"); + } + + let mut received_ids = BTreeSet::new(); + for _ in 0..16 { + sched.tick(); + for message in sched.drain_ready(3) { + assert!( + received_ids.insert(message.id), + "seed {seed} returned message {} twice", + message.id + ); + } + } + while !sched.is_quiescent() { + for message in sched.drain_ready(3) { + assert!(received_ids.insert(message.id)); + } + } + + assert_eq!(received_ids.len(), 32, "seed {seed} lost a delivery"); + assert_eq!(sched.total_delivered(), 32); + assert_eq!(sched.total_dropped(), 0); + } +} + +#[test] +fn test_message_delay_increases_delivery_time() { + let config = FaultConfig { + message_delay_prob: 1.0, // Always delay + max_delay_ticks: 5, + ..FaultConfig::none() + }; + let mut sched = SimScheduler::new(42, config); + sched.register_actor("a"); + sched.register_actor("b"); + + sched.send("a", "b", "delayed", "{}"); + + // Tick 1: message not yet delivered (delayed) + sched.tick(); + let delivered_at_1 = sched.total_delivered(); + + // Run more ticks + drain_until_quiescent(&mut sched, 20); + assert_eq!( + sched.total_delivered(), + 1, + "Message should eventually arrive" + ); + if delivered_at_1 == 0 { + assert!( + sched.current_time() > 1, + "Delivery should be delayed beyond tick 1" + ); + } +} + +#[test] +fn test_heavy_faults_simulation_completes() { + // Even with heavy faults, simulation should complete without panic + let mut sched = SimScheduler::new(12345, FaultConfig::heavy()); + for i in 0..5 { + sched.register_actor(&format!("actor-{i}")); + } + + // Send 50 messages between random actors + let mut rng = super::super::DeterministicRng::new(67890); + for _ in 0..50 { + let from = format!("actor-{}", rng.next_bound(5)); + let to = format!("actor-{}", rng.next_bound(5)); + sched.send(&from, &to, "msg", "{}"); + } + + drain_until_quiescent(&mut sched, 200); + + // Just verify it completed without panic and some messages got through + let total = sched.total_delivered() + sched.total_dropped(); + assert!(total > 0, "Should have processed some messages"); +} + +#[test] +fn test_send_at_delivers_at_specified_time() { + let mut sched = SimScheduler::new(1, FaultConfig::none()); + sched.register_actor("a"); + sched.register_actor("b"); + + // Schedule a message at time 5 + sched.send_at("a", "b", "Scheduled", "{}", 5); + + // Ticks 1-4: nothing delivered + for _ in 1..5 { + sched.tick(); + assert_eq!( + sched.total_delivered(), + 0, + "should not deliver before deliver_at" + ); + } + + // Tick 5: message delivered + sched.tick(); + assert_eq!(sched.total_delivered(), 1); + + let msg = sched.drain_ready(1).pop().unwrap(); + assert_eq!(msg.msg_type, "Scheduled"); + assert_eq!(msg.deliver_at, 5); +} + +#[test] +fn test_send_at_respects_message_drop() { + let config = FaultConfig { + message_drop_prob: 1.0, + ..FaultConfig::none() + }; + let mut sched = SimScheduler::new(42, config); + sched.register_actor("a"); + sched.register_actor("b"); + + sched.send_at("a", "b", "Scheduled", "{}", 3); + drain_until_quiescent(&mut sched, 10); + + assert_eq!(sched.total_delivered(), 0); + assert_eq!(sched.total_dropped(), 1); +} diff --git a/crates/temper-runtime/src/scheduler/mod.rs b/crates/temper-runtime/src/scheduler/mod.rs index bc5e84a5d..8468201aa 100644 --- a/crates/temper-runtime/src/scheduler/mod.rs +++ b/crates/temper-runtime/src/scheduler/mod.rs @@ -22,7 +22,7 @@ pub use id_gen::{DeterministicIdGen, RealIdGen, SimIdGen}; pub use rng::DeterministicRng; pub use sim_actor_system::{ ActorInvariantViolation, RunRecord, SimActorResult, SimActorSystem, SimActorSystemConfig, - SimIntegrationResponses, + SimExecutionError, SimIntegrationResponses, }; pub use sim_handler::{CompareOp, SimActorHandler, SpecAssert, SpecInvariant}; pub use types::{FaultConfig, SimActorState, SimMessage, SimTime}; diff --git a/crates/temper-runtime/src/scheduler/sim_actor_system.rs b/crates/temper-runtime/src/scheduler/sim_actor_system.rs index 8009277c4..581cc8add 100644 --- a/crates/temper-runtime/src/scheduler/sim_actor_system.rs +++ b/crates/temper-runtime/src/scheduler/sim_actor_system.rs @@ -1,16 +1,9 @@ -//! Deterministic actor simulation system. -//! -//! [`SimActorSystem`] bridges [`SimScheduler`] and real actor handlers -//! ([`SimActorHandler`]). It runs real `TransitionTable::evaluate()` through -//! the scheduler with seed-controlled everything. -//! -//! Two modes: -//! - **Scripted**: call `step()` with specific (actor, action, params) tuples -//! - **Random**: call `run_random()` to explore randomly with fault injection -//! -//! Invariants are checked after every successful transition. - -use std::collections::BTreeMap; +//! Deterministic actor simulation through [`SimScheduler`] and real +//! [`SimActorHandler`] implementations. Scripted mode uses [`SimActorSystem::step`]; +//! random mode uses [`SimActorSystem::run_random`] with fault injection. +//! Invariants are checked after each successful transition. + +use std::collections::{BTreeMap, VecDeque}; use std::sync::Arc; use super::clock::{LogicalClock, SimClock}; @@ -19,40 +12,13 @@ use super::id_gen::DeterministicIdGen; use super::sim_handler::SimActorHandler; use super::{DeterministicRng, FaultConfig, SimScheduler}; -/// Configures how integration callbacks are delivered in simulation. -/// -/// Maps `(entity_type, trigger_name)` → callback action name. When a simulated -/// entity emits a custom effect matching a trigger, the system auto-schedules -/// the configured callback action on the next tick. This lets DST explore both -/// success and failure paths without executing real WASM modules. -#[derive(Debug, Clone, Default)] -pub struct SimIntegrationResponses { - /// Maps (entity_type, trigger_name) → callback action name. - responses: BTreeMap<(String, String), String>, -} - -impl SimIntegrationResponses { - /// Create an empty integration response map. - pub fn new() -> Self { - Self::default() - } - - /// Configure a success callback for a trigger. - pub fn on_trigger(mut self, entity_type: &str, trigger: &str, callback_action: &str) -> Self { - self.responses.insert( - (entity_type.to_string(), trigger.to_string()), - callback_action.to_string(), - ); - self - } - - /// Look up the callback action for a trigger. - pub fn get_callback(&self, entity_type: &str, trigger: &str) -> Option<&str> { - self.responses - .get(&(entity_type.to_string(), trigger.to_string())) - .map(|s| s.as_str()) - } -} +mod invariant_eval; +use invariant_eval::evaluate_spec_assert; +mod callbacks; +mod random_budget; +use random_budget::{release, reserve}; +mod recording; +pub use callbacks::{SimExecutionError, SimIntegrationResponses}; /// Configuration for a [`SimActorSystem`] run. #[derive(Debug, Clone)] @@ -65,6 +31,10 @@ pub struct SimActorSystemConfig { pub faults: FaultConfig, /// Maximum actions per actor in random mode. pub max_actions_per_actor: usize, + /// Maximum ready messages transferred in one bounded drain batch. + pub message_batch_budget: usize, + /// Maximum integration callbacks executed in one deterministic cascade. + pub reaction_budget_per_tick: usize, } impl Default for SimActorSystemConfig { @@ -74,6 +44,8 @@ impl Default for SimActorSystemConfig { max_ticks: 500, faults: FaultConfig::light(), max_actions_per_actor: 50, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, } } } @@ -95,11 +67,8 @@ pub struct ActorInvariantViolation { pub tick: u64, } -/// Complete recording of a simulation run for determinism comparison. -/// -/// Captures every state transition, every event, and every final state so that -/// two runs with the same seed can be compared for byte-exact equality. -/// This is the FoundationDB principle: same seed MUST produce identical output. +/// Complete transition, event, invariant, and final-state recording for +/// byte-exact comparison of runs using the same seed. #[derive(Debug, Clone, PartialEq, Eq)] pub struct RunRecord { /// Seed used. @@ -117,7 +86,7 @@ pub struct RunRecord { /// Result of a simulation run. #[derive(Debug, Clone)] pub struct SimActorResult { - /// Whether all invariants held. + /// Whether all invariants held and the simulation driver completed cleanly. pub all_invariants_held: bool, /// Seed used (for replay). pub seed: u64, @@ -129,6 +98,8 @@ pub struct SimActorResult { pub dropped: u64, /// Invariant violations found. pub violations: Vec, + /// Callback or driver failures that invalidate the run. + pub execution_errors: Vec, /// Final state per actor: (actor_id, status, item_count, event_count). pub actor_states: Vec<(String, String, usize, usize)>, } @@ -136,15 +107,14 @@ pub struct SimActorResult { /// Invariant checker function signature. pub type InvariantChecker = Box Option>; -/// The deterministic actor simulation system. -/// -/// Runs real [`SimActorHandler`] instances through [`SimScheduler`] with -/// full determinism: logical clock, deterministic UUIDs, seed-controlled -/// fault injection. +/// Runs real handlers through a logical clock, deterministic UUIDs, and +/// seed-controlled scheduler fault injection. pub struct SimActorSystem { config: SimActorSystemConfig, actors: BTreeMap>, action_counts: BTreeMap, + random_in_flight_actions: BTreeMap, + observed_scheduler_drops: usize, scheduler: SimScheduler, clock: Arc, _id_gen: Arc, @@ -152,6 +122,7 @@ pub struct SimActorSystem { rng: DeterministicRng, invariant_checker: Option, violations: Vec, + execution_errors: Vec, total_transitions: u64, total_messages: u64, /// Recorded transitions for RunRecord: (tick, actor_id, action, from_status, to_status). @@ -161,22 +132,36 @@ pub struct SimActorSystem { /// Integration callback configuration for WASM trigger simulation. integration_responses: SimIntegrationResponses, /// Pending integration callbacks to deliver: (actor_id, callback_action). - pending_integration_callbacks: Vec<(String, String)>, + pending_integration_callbacks: VecDeque<(String, String)>, } impl SimActorSystem { /// Create a new simulation system with the given config. pub fn new(config: SimActorSystemConfig) -> Self { + assert!( + config.message_batch_budget > 0, + "message batch budget must be positive" + ); + assert!( + config.reaction_budget_per_tick > 0, + "reaction budget per tick must be positive" + ); let clock = Arc::new(LogicalClock::new()); let id_gen = Arc::new(DeterministicIdGen::new(config.seed)); let guard = install_sim_context(clock.clone(), id_gen.clone()); - let scheduler = SimScheduler::new(config.seed, config.faults.clone()); + let mailbox_budget = usize::try_from(config.max_ticks) + .expect("maximum ticks must fit the platform address space") + .max(1); + let scheduler = + SimScheduler::with_mailbox_budget(config.seed, config.faults.clone(), mailbox_budget); let rng = DeterministicRng::new(config.seed.wrapping_add(7)); Self { config, actors: BTreeMap::new(), action_counts: BTreeMap::new(), + random_in_flight_actions: BTreeMap::new(), + observed_scheduler_drops: 0, scheduler, clock, _id_gen: id_gen, @@ -184,12 +169,13 @@ impl SimActorSystem { rng, invariant_checker: None, violations: Vec::new(), + execution_errors: Vec::new(), total_transitions: 0, total_messages: 0, recorded_transitions: Vec::new(), recorded_invariants: Vec::new(), integration_responses: SimIntegrationResponses::new(), - pending_integration_callbacks: Vec::new(), + pending_integration_callbacks: VecDeque::new(), } } @@ -199,6 +185,7 @@ impl SimActorSystem { handler.init().expect("actor init should succeed"); self.actors.insert(id.to_string(), handler); self.action_counts.insert(id.to_string(), 0); + self.random_in_flight_actions.insert(id.to_string(), 0); } /// Set a custom invariant checker. @@ -219,71 +206,74 @@ impl SimActorSystem { self.integration_responses = responses; } - // =================================================================== - // Scripted Mode - // =================================================================== - /// Execute a specific action on a specific actor. /// - /// Returns the actor's state as JSON on success, or an error string. + /// Returns the actor's state as JSON when both the primary action and its + /// callback cascade succeed. If a callback fails, the primary action has + /// already committed and this returns an error describing the callback; + /// callers must not retry the primary action as though it were rolled back. pub fn step( &mut self, actor_id: &str, action: &str, params: &str, ) -> Result { - let handler = self - .actors - .get_mut(actor_id) - .ok_or_else(|| format!("Unknown actor: {actor_id}"))?; - - let status_before = handler.current_status(); + self.ensure_execution_active()?; self.clock.advance(); self.total_messages += 1; + let result = self.apply_action(actor_id, action, params)?; + self.deliver_integration_callbacks(&mut 0)?; + Ok(result) + } - let result = handler.handle_message(action, params); - - match &result { - Ok(_) => { - let status_after = handler.current_status(); - let item_count = handler.current_item_count(); - let tick = self.clock.tick(); - - // Only count as transition if status or items actually changed - let count = self.action_counts.get_mut(actor_id).unwrap(); // ci-ok: actor always in action_counts - *count += 1; - self.total_transitions += 1; - - // Record the transition - self.recorded_transitions.push(( - tick, - actor_id.to_string(), - action.to_string(), - status_before.clone(), - status_after.clone(), - )); - - // Check invariants - self.check_invariants( - actor_id, - action, - &status_before, - &status_after, - item_count, - tick, - ); + /// Apply one actor action without recursively delivering its callbacks. + fn apply_action( + &mut self, + actor_id: &str, + action: &str, + params: &str, + ) -> Result { + let (status_before, result, status_after, item_count) = { + let handler = self + .actors + .get_mut(actor_id) + .ok_or_else(|| format!("Unknown actor: {actor_id}"))?; + + let status_before = handler.current_status(); + let result = handler.handle_message(action, params); + let status_after = handler.current_status(); + let item_count = handler.current_item_count(); + (status_before, result, status_after, item_count) + }; - // Schedule integration callbacks for any custom effects - self.schedule_integration_callbacks(actor_id); - } - Err(_) => { - // Failed action — invariants should still hold on unchanged state - } - } + if result.is_ok() { + let tick = self.clock.tick(); - // Deliver any pending integration callbacks - if !self.pending_integration_callbacks.is_empty() { - self.deliver_integration_callbacks(); + let count = self + .action_counts + .get_mut(actor_id) + .expect("registered actor"); + *count += 1; + self.total_transitions += 1; + + self.recorded_transitions.push(( + tick, + actor_id.to_string(), + action.to_string(), + status_before.clone(), + status_after.clone(), + )); + + self.check_invariants( + actor_id, + action, + &status_before, + &status_after, + item_count, + tick, + ); + + self.schedule_integration_callbacks(actor_id); } result @@ -341,7 +331,12 @@ impl SimActorSystem { .unwrap_or_default() } - /// Whether there are any violations. + /// Whether any spec-invariant violations were recorded. + /// + /// This reflects only invariant-check failures. For the full simulation + /// health gate (invariants **and** callback/delivery errors) use + /// [`execution_errors`](Self::execution_errors) in addition, or inspect + /// [`SimActorResult::all_invariants_held`] from a [`run_random`](Self::run_random) call. pub fn has_violations(&self) -> bool { !self.violations.is_empty() } @@ -351,16 +346,21 @@ impl SimActorSystem { &self.violations } - // =================================================================== - // Random Mode - // =================================================================== + /// Get callback and driver failures collected during random simulation. + pub fn execution_errors(&self) -> &[SimExecutionError] { + &self.execution_errors + } /// Run random exploration with fault injection. /// /// The RNG picks actors and actions. The scheduler delays/drops/crashes. /// Invariants are checked after every successful transition. pub fn run_random(&mut self) -> SimActorResult { - for _tick in 0..self.config.max_ticks { + if self.ensure_execution_active().is_err() { + return self.result_snapshot(); + } + + 'simulation: for tick in 0..self.config.max_ticks { if self.actors.is_empty() { break; } @@ -371,217 +371,56 @@ impl SimActorSystem { let actor_id = actor_ids[actor_idx].clone(); // Check action budget - let count = self.action_counts.get(&actor_id).copied().unwrap_or(0); - if count >= self.config.max_actions_per_actor { - continue; - } - - // Get valid actions - let valid = { - let handler = self.actors.get(&actor_id).unwrap(); // ci-ok: actor_id from self.actors.keys() - handler.valid_actions() - }; - - if valid.is_empty() { - continue; // Terminal state - } - - // Pick a random valid action - let action_idx = self.rng.next_bound(valid.len()); - let action = valid[action_idx].clone(); - - // Execute through the scheduler for fault injection - self.scheduler.send("sim-driver", &actor_id, &action, "{}"); - self.total_messages += 1; - - let delivered = self.scheduler.tick(); - self.clock.advance(); + let completed = self.action_counts.get(&actor_id).copied().unwrap_or(0); + let in_flight = self + .random_in_flight_actions + .get(&actor_id) + .copied() + .unwrap_or(0); + if completed + in_flight < self.config.max_actions_per_actor { + let valid = { + let handler = self.actors.get(&actor_id).expect("selected actor"); + handler.valid_actions() + }; - // Process delivered messages - 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 - } - } + if !valid.is_empty() { + let action_idx = self.rng.next_bound(valid.len()); + let action = valid[action_idx].clone(); + self.scheduler.send("sim-driver", &actor_id, &action, "{}"); + reserve(&mut self.random_in_flight_actions, &actor_id); + self.total_messages += 1; } } - // Deliver any pending integration callbacks - if !self.pending_integration_callbacks.is_empty() { - self.deliver_integration_callbacks(); - } - - // Drain any remaining scheduled messages + // Logical time and ready delivery progress independently of whether + // a new action was eligible this tick. self.scheduler.tick(); - } - - let actor_states: Vec<_> = self - .actors - .iter() - .map(|(id, h)| { - ( - id.clone(), - h.current_status(), - h.current_item_count(), - h.event_count(), - ) - }) - .collect(); - - SimActorResult { - all_invariants_held: self.violations.is_empty(), - seed: self.config.seed, - transitions: self.total_transitions, - messages: self.total_messages, - dropped: self.scheduler.total_dropped() as u64, - violations: self.violations.clone(), - actor_states, - } - } - - /// Run random exploration and return a full [`RunRecord`] alongside the result. - /// - /// This is the recording variant of [`run_random()`]. The `RunRecord` captures - /// every transition, every event, and every final state for determinism - /// comparison. Two calls with the same seed MUST produce identical records. - pub fn run_random_recorded(&mut self) -> (SimActorResult, RunRecord) { - let result = self.run_random(); - - // Collect events from each actor - let events: BTreeMap> = self - .actors - .iter() - .map(|(id, handler)| { - let events_val = handler.events_json(); - let event_strings = match events_val { - serde_json::Value::Array(arr) => arr - .iter() - .map(|v| serde_json::to_string(v).unwrap_or_default()) - .collect(), - _ => Vec::new(), - }; - (id.clone(), event_strings) - }) - .collect(); - - // Collect final states with counters serialized as JSON - let final_states: Vec<_> = self - .actors - .iter() - .map(|(id, handler)| { - let status = handler.current_status(); - let item_count = handler.current_item_count(); - let event_count = handler.event_count(); - // Serialize the full events_json as a proxy for counters - // since SimActorHandler doesn't expose counters directly. - // The events contain all state change details. - let counters_json = - serde_json::to_string(&handler.events_json()).unwrap_or_default(); - (id.clone(), status, item_count, event_count, counters_json) - }) - .collect(); - - let record = RunRecord { - seed: self.config.seed, - transitions: self.recorded_transitions.clone(), - events, - final_states, - invariant_results: self.recorded_invariants.clone(), - }; - - (result, record) - } - - // =================================================================== - // Integration callback scheduling - // =================================================================== - - /// Check for pending integration callbacks and schedule them. - /// - /// After a successful action, the handler may have emitted custom effects - /// (integration triggers). This method looks up configured callbacks and - /// queues them for delivery on the next tick. - fn schedule_integration_callbacks(&mut self, actor_id: &str) { - let handler = match self.actors.get(actor_id) { - Some(h) => h, - None => return, - }; - - let callbacks = handler.pending_callbacks(); - if callbacks.is_empty() { - return; - } - - // Derive entity_type from actor_id (convention: "EntityType:EntityId" or just id) - // For simplicity, check against all registered entity_type patterns. - for trigger in &callbacks { - // Try matching with the actor_id as-is for the entity_type lookup - if let Some(callback_action) = - self.integration_responses.get_callback(actor_id, trigger) - { - self.pending_integration_callbacks - .push((actor_id.to_string(), callback_action.to_string())); + self.clock.advance(); + for dropped in &self.scheduler.dropped_log()[self.observed_scheduler_drops..] { + if dropped.from == "sim-driver" { + release(&mut self.random_in_flight_actions, &dropped.to, "dropped"); + } } - // Also try splitting on ':' (e.g., "Order:o1" → entity_type = "Order") - else if let Some(colon_pos) = actor_id.find(':') { - let entity_type = &actor_id[..colon_pos]; - if let Some(callback_action) = self - .integration_responses - .get_callback(entity_type, trigger) - { - self.pending_integration_callbacks - .push((actor_id.to_string(), callback_action.to_string())); + self.observed_scheduler_drops = self.scheduler.dropped_log().len(); + let mut reactions = 0; + loop { + let delivered = self.scheduler.drain_ready(self.config.message_batch_budget); + for msg in &delivered { + release(&mut self.random_in_flight_actions, &msg.to, "delivered"); + let _ = self.apply_action(&msg.to, &msg.msg_type, &msg.payload); + } + if self.deliver_integration_callbacks(&mut reactions).is_err() { + break 'simulation; + } + if tick + 1 < self.config.max_ticks || !self.scheduler.has_ready_messages() { + break; } } } - } - /// Deliver any pending integration callbacks by executing them as actions. - 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, "{}"); - } + self.result_snapshot() } - // =================================================================== - // Invariant checking - // =================================================================== - fn check_invariants( &mut self, actor_id: &str, @@ -642,148 +481,6 @@ impl SimActorSystem { } } -/// Evaluate a [`SpecAssert`] against handler state. Returns `true` if the -/// assertion holds, `false` if violated. Recurses through `And`/`Or`. -fn evaluate_spec_assert( - assert: &super::sim_handler::SpecAssert, - handler: &dyn super::sim_handler::SimActorHandler, - when: &[String], - status_before: &str, - status_after: &str, - item_count: usize, -) -> bool { - use super::sim_handler::{CompareOp, SpecAssert}; - - match assert { - SpecAssert::CounterPositive { var } => { - if var == "items" { - item_count > 0 - } else { - true // Unknown counter: not in scope for invariant checking here. - } - } - SpecAssert::NoFurtherTransitions => { - // Holds unless status_before was a terminal state in `when`. - !when.iter().any(|s| s == status_before) - } - SpecAssert::OrderingConstraint { before, after } => { - if status_after == after.as_str() { - let events = handler.events_json(); - if let Some(arr) = events.as_array() { - arr.iter().any(|e| { - e.get("to_status").and_then(|s| s.as_str()) == Some(before.as_str()) - }) - } else { - true - } - } else { - true - } - } - SpecAssert::NeverState { state } => status_after != state.as_str(), - SpecAssert::CounterCompare { var, op, value } => { - let counter_val = if var == "items" { item_count } else { 0 }; - match op { - CompareOp::Gt => counter_val > *value, - CompareOp::Gte => counter_val >= *value, - CompareOp::Lt => counter_val < *value, - CompareOp::Lte => counter_val <= *value, - CompareOp::Eq => counter_val == *value, - } - } - SpecAssert::BoolRequired { var, expect } => { - handler.bool_field(var).unwrap_or(false) == *expect - } - SpecAssert::And(parts) => parts.iter().all(|p| { - evaluate_spec_assert(p, handler, when, status_before, status_after, item_count) - }), - SpecAssert::Or(parts) => parts.iter().any(|p| { - evaluate_spec_assert(p, handler, when, status_before, status_after, item_count) - }), - } -} - #[cfg(test)] -mod tests { - use super::*; - - #[test] - fn integration_responses_empty_returns_none() { - let responses = SimIntegrationResponses::new(); - assert!(responses.get_callback("Order", "payment_trigger").is_none()); - } - - #[test] - fn integration_responses_on_trigger_and_get_callback() { - let responses = SimIntegrationResponses::new() - .on_trigger("Order", "payment_trigger", "ConfirmPayment") - .on_trigger("Invoice", "send_trigger", "MarkSent"); - - assert_eq!( - responses.get_callback("Order", "payment_trigger"), - Some("ConfirmPayment") - ); - assert_eq!( - responses.get_callback("Invoice", "send_trigger"), - Some("MarkSent") - ); - assert!(responses.get_callback("Order", "send_trigger").is_none()); - assert!( - responses - .get_callback("Unknown", "payment_trigger") - .is_none() - ); - } - - #[test] - fn integration_responses_overwrite() { - let responses = SimIntegrationResponses::new() - .on_trigger("Order", "trigger", "ActionA") - .on_trigger("Order", "trigger", "ActionB"); - - assert_eq!(responses.get_callback("Order", "trigger"), Some("ActionB")); - } - - #[test] - fn config_default_values() { - let config = SimActorSystemConfig::default(); - assert_eq!(config.seed, 42); - assert_eq!(config.max_ticks, 500); - assert_eq!(config.max_actions_per_actor, 50); - } - - #[test] - fn run_record_equality() { - let r1 = RunRecord { - seed: 42, - transitions: vec![( - 1, - "a".into(), - "Submit".into(), - "Draft".into(), - "Submitted".into(), - )], - events: BTreeMap::new(), - final_states: vec![], - invariant_results: vec![], - }; - let r2 = r1.clone(); - assert_eq!(r1, r2); - } - - #[test] - fn run_record_inequality_on_seed() { - let r1 = RunRecord { - seed: 42, - transitions: vec![], - events: BTreeMap::new(), - final_states: vec![], - invariant_results: vec![], - }; - let r2 = RunRecord { - seed: 99, - ..r1.clone() - }; - assert_ne!(r1, r2); - } -} +#[path = "sim_actor_system/tests.rs"] +mod tests; diff --git a/crates/temper-runtime/src/scheduler/sim_actor_system/callbacks.rs b/crates/temper-runtime/src/scheduler/sim_actor_system/callbacks.rs new file mode 100644 index 000000000..460ebac70 --- /dev/null +++ b/crates/temper-runtime/src/scheduler/sim_actor_system/callbacks.rs @@ -0,0 +1,122 @@ +//! Deterministic integration callback configuration and delivery. + +use std::collections::BTreeMap; + +use super::SimActorSystem; +use crate::scheduler::SimClock; + +/// Configures how integration callbacks are delivered in simulation. +#[derive(Debug, Clone, Default)] +pub struct SimIntegrationResponses { + responses: BTreeMap<(String, String), String>, +} + +impl SimIntegrationResponses { + /// Create an empty integration response map. + pub fn new() -> Self { + Self::default() + } + + /// Configure a callback action for a trigger. + pub fn on_trigger(mut self, entity_type: &str, trigger: &str, callback_action: &str) -> Self { + self.responses.insert( + (entity_type.to_string(), trigger.to_string()), + callback_action.to_string(), + ); + self + } + + /// Look up the callback action for a trigger. + pub fn get_callback(&self, entity_type: &str, trigger: &str) -> Option<&str> { + self.responses + .get(&(entity_type.to_string(), trigger.to_string())) + .map(String::as_str) + } +} + +/// A simulation-driver failure that invalidates a run. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SimExecutionError { + /// Actor whose action failed. + pub actor_id: String, + /// Action that failed. + pub action: String, + /// Error returned by the actor handler or driver budget. + pub description: String, + /// Logical tick at failure. + pub tick: u64, +} + +impl SimActorSystem { + pub(super) fn ensure_execution_active(&self) -> Result<(), String> { + let Some(error) = self.execution_errors.last() else { + return Ok(()); + }; + Err(format!( + "simulation run is invalid after execution error at tick {}: {}", + error.tick, error.description + )) + } + + pub(super) fn schedule_integration_callbacks(&mut self, actor_id: &str) { + let Some(handler) = self.actors.get(actor_id) else { + return; + }; + + for trigger in handler.pending_callbacks() { + let callback_action = self + .integration_responses + .get_callback(actor_id, &trigger) + .or_else(|| { + actor_id.find(':').and_then(|colon| { + self.integration_responses + .get_callback(&actor_id[..colon], &trigger) + }) + }); + + if let Some(callback_action) = callback_action { + self.pending_integration_callbacks + .push_back((actor_id.to_string(), callback_action.to_string())); + } + } + } + + pub(super) fn deliver_integration_callbacks( + &mut self, + reactions: &mut usize, + ) -> Result<(), String> { + while let Some((actor_id, callback_action)) = self.pending_integration_callbacks.pop_front() + { + if *reactions == self.config.reaction_budget_per_tick { + let description = + format!("integration callback budget exhausted after {reactions} reactions"); + return self.invalidate_callback_cascade(actor_id, callback_action, description); + } + + *reactions += 1; + if let Err(error) = self.apply_action(&actor_id, &callback_action, "{}") { + let description = format!( + "integration callback '{callback_action}' failed for '{actor_id}': {error}" + ); + return self.invalidate_callback_cascade(actor_id, callback_action, description); + } + } + Ok(()) + } + + fn invalidate_callback_cascade( + &mut self, + actor_id: String, + action: String, + description: String, + ) -> Result<(), String> { + self.pending_integration_callbacks.clear(); + self.execution_errors.push(SimExecutionError { + actor_id, + action, + description: description.clone(), + tick: self.clock.tick(), + }); + Err(description) + } +} diff --git a/crates/temper-runtime/src/scheduler/sim_actor_system/invariant_eval.rs b/crates/temper-runtime/src/scheduler/sim_actor_system/invariant_eval.rs new file mode 100644 index 000000000..d719845d6 --- /dev/null +++ b/crates/temper-runtime/src/scheduler/sim_actor_system/invariant_eval.rs @@ -0,0 +1,58 @@ +//! Spec-derived invariant evaluation for actor simulation. + +use super::super::sim_handler::{CompareOp, SimActorHandler, SpecAssert}; + +/// Evaluate a [`SpecAssert`] against handler state. Returns `true` if the +/// assertion holds, `false` if violated. Recurses through `And`/`Or`. +pub(super) fn evaluate_spec_assert( + assert: &SpecAssert, + handler: &dyn SimActorHandler, + when: &[String], + status_before: &str, + status_after: &str, + item_count: usize, +) -> bool { + match assert { + SpecAssert::CounterPositive { var } => { + if var == "items" { + item_count > 0 + } else { + true + } + } + SpecAssert::NoFurtherTransitions => !when.iter().any(|state| state == status_before), + SpecAssert::OrderingConstraint { before, after } => { + if status_after == after.as_str() { + let events = handler.events_json(); + events.as_array().is_none_or(|events| { + events.iter().any(|event| { + event.get("to_status").and_then(|status| status.as_str()) + == Some(before.as_str()) + }) + }) + } else { + true + } + } + SpecAssert::NeverState { state } => status_after != state.as_str(), + SpecAssert::CounterCompare { var, op, value } => { + let counter_value = if var == "items" { item_count } else { 0 }; + match op { + CompareOp::Gt => counter_value > *value, + CompareOp::Gte => counter_value >= *value, + CompareOp::Lt => counter_value < *value, + CompareOp::Lte => counter_value <= *value, + CompareOp::Eq => counter_value == *value, + } + } + SpecAssert::BoolRequired { var, expect } => { + handler.bool_field(var).unwrap_or(false) == *expect + } + SpecAssert::And(parts) => parts.iter().all(|part| { + evaluate_spec_assert(part, handler, when, status_before, status_after, item_count) + }), + SpecAssert::Or(parts) => parts.iter().any(|part| { + evaluate_spec_assert(part, handler, when, status_before, status_after, item_count) + }), + } +} diff --git a/crates/temper-runtime/src/scheduler/sim_actor_system/random_budget.rs b/crates/temper-runtime/src/scheduler/sim_actor_system/random_budget.rs new file mode 100644 index 000000000..16a816281 --- /dev/null +++ b/crates/temper-runtime/src/scheduler/sim_actor_system/random_budget.rs @@ -0,0 +1,22 @@ +//! Random-driver action reservation accounting. + +use std::collections::BTreeMap; + +pub(super) fn reserve(in_flight_actions: &mut BTreeMap, actor_id: &str) { + let in_flight = in_flight_actions + .get_mut(actor_id) + .expect("registered reservation target"); + *in_flight += 1; +} + +pub(super) fn release( + in_flight_actions: &mut BTreeMap, + actor_id: &str, + outcome: &str, +) { + let in_flight = in_flight_actions + .get_mut(actor_id) + .expect("registered reservation target"); + assert!(*in_flight > 0, "{outcome} action must own a reservation"); + *in_flight -= 1; +} diff --git a/crates/temper-runtime/src/scheduler/sim_actor_system/recording.rs b/crates/temper-runtime/src/scheduler/sim_actor_system/recording.rs new file mode 100644 index 000000000..1e120d69c --- /dev/null +++ b/crates/temper-runtime/src/scheduler/sim_actor_system/recording.rs @@ -0,0 +1,82 @@ +//! Full deterministic run recording. + +use std::collections::BTreeMap; + +use super::{RunRecord, SimActorResult, SimActorSystem}; + +impl SimActorSystem { + pub(super) fn result_snapshot(&self) -> SimActorResult { + let actor_states = self + .actors + .iter() + .map(|(id, handler)| { + ( + id.clone(), + handler.current_status(), + handler.current_item_count(), + handler.event_count(), + ) + }) + .collect(); + + SimActorResult { + all_invariants_held: self.violations.is_empty() && self.execution_errors.is_empty(), + seed: self.config.seed, + transitions: self.total_transitions, + messages: self.total_messages, + dropped: self.scheduler.total_dropped() as u64, + violations: self.violations.clone(), + execution_errors: self.execution_errors.clone(), + actor_states, + } + } + + /// Run random exploration and return a full [`RunRecord`] alongside the result. + /// + /// The record captures every transition, event, and final state. Two calls + /// with the same seed must produce identical records. + pub fn run_random_recorded(&mut self) -> (SimActorResult, RunRecord) { + let result = self.run_random(); + + let events: BTreeMap> = self + .actors + .iter() + .map(|(id, handler)| { + let event_strings = match handler.events_json() { + serde_json::Value::Array(events) => events + .iter() + .map(|event| serde_json::to_string(event).unwrap_or_default()) + .collect(), + _ => Vec::new(), + }; + (id.clone(), event_strings) + }) + .collect(); + + let final_states = self + .actors + .iter() + .map(|(id, handler)| { + let counters_json = + serde_json::to_string(&handler.events_json()).unwrap_or_default(); + ( + id.clone(), + handler.current_status(), + handler.current_item_count(), + handler.event_count(), + counters_json, + ) + }) + .collect(); + + let record = RunRecord { + seed: self.config.seed, + transitions: self.recorded_transitions.clone(), + events, + final_states, + invariant_results: self.recorded_invariants.clone(), + }; + + (result, record) + } +} diff --git a/crates/temper-runtime/src/scheduler/sim_actor_system/tests.rs b/crates/temper-runtime/src/scheduler/sim_actor_system/tests.rs new file mode 100644 index 000000000..492a1a760 --- /dev/null +++ b/crates/temper-runtime/src/scheduler/sim_actor_system/tests.rs @@ -0,0 +1,296 @@ +use super::*; + +struct CallbackFailureHandler { + status: String, + pending_callbacks: Vec, +} + +impl CallbackFailureHandler { + fn new() -> Self { + Self { + status: "Ready".to_string(), + pending_callbacks: Vec::new(), + } + } +} + +impl SimActorHandler for CallbackFailureHandler { + fn init(&mut self) -> Result { + Ok(serde_json::json!({"status": self.status})) + } + + fn handle_message(&mut self, action: &str, _params: &str) -> Result { + match action { + "Start" => { + self.status = "Started".to_string(); + self.pending_callbacks = vec!["integration".to_string()]; + Ok(serde_json::json!({"status": self.status})) + } + "Callback" => Err("callback rejected".to_string()), + "Complete" => { + self.pending_callbacks.clear(); + Ok(serde_json::json!({"status": self.status})) + } + "Loop" => Ok(serde_json::json!({"status": self.status})), + _ => Err(format!("unknown action: {action}")), + } + } + + fn current_status(&self) -> String { + self.status.clone() + } + + fn current_item_count(&self) -> usize { + 0 + } + + fn event_count(&self) -> usize { + 0 + } + + fn valid_actions(&self) -> Vec { + if self.status == "Ready" { + vec!["Start".to_string()] + } else { + Vec::new() + } + } + + fn events_json(&self) -> serde_json::Value { + serde_json::json!([]) + } + + fn pending_callbacks(&self) -> Vec { + self.pending_callbacks.clone() + } +} + +#[test] +fn integration_responses_empty_returns_none() { + let responses = SimIntegrationResponses::new(); + assert!(responses.get_callback("Order", "payment_trigger").is_none()); +} + +#[test] +fn integration_responses_on_trigger_and_get_callback() { + let responses = SimIntegrationResponses::new() + .on_trigger("Order", "payment_trigger", "ConfirmPayment") + .on_trigger("Invoice", "send_trigger", "MarkSent"); + + assert_eq!( + responses.get_callback("Order", "payment_trigger"), + Some("ConfirmPayment") + ); + assert_eq!( + responses.get_callback("Invoice", "send_trigger"), + Some("MarkSent") + ); + assert!(responses.get_callback("Order", "send_trigger").is_none()); + assert!( + responses + .get_callback("Unknown", "payment_trigger") + .is_none() + ); +} + +#[test] +fn integration_responses_overwrite() { + let responses = SimIntegrationResponses::new() + .on_trigger("Order", "trigger", "ActionA") + .on_trigger("Order", "trigger", "ActionB"); + + assert_eq!(responses.get_callback("Order", "trigger"), Some("ActionB")); +} + +#[test] +fn config_default_values() { + let config = SimActorSystemConfig::default(); + assert_eq!(config.seed, 42); + assert_eq!(config.max_ticks, 500); + assert_eq!(config.max_actions_per_actor, 50); + assert_eq!(config.message_batch_budget, 1_024); + assert_eq!(config.reaction_budget_per_tick, 1_024); +} + +#[test] +fn callback_failure_is_returned_and_invalidates_random_run() { + let config = SimActorSystemConfig { + seed: 1, + max_ticks: 2, + faults: FaultConfig { + message_delay_prob: 1.0, + max_delay_ticks: 2, + ..FaultConfig::none() + }, + max_actions_per_actor: 1, + message_batch_budget: 1, + reaction_budget_per_tick: 1, + }; + let responses = SimIntegrationResponses::new().on_trigger("Job", "integration", "Callback"); + + let mut scripted = SimActorSystem::new(config.clone()); + scripted.register_actor("Job:1", Box::new(CallbackFailureHandler::new())); + scripted.set_integration_responses(responses.clone()); + let error = scripted.step("Job:1", "Start", "{}").unwrap_err(); + assert!(error.contains("callback rejected")); + assert_eq!(scripted.execution_errors().len(), 1); + let transitions_after_failure = scripted.total_transitions; + let retry_error = scripted.step("Job:1", "Complete", "{}").unwrap_err(); + assert!(retry_error.contains("simulation run is invalid")); + assert_eq!(scripted.total_transitions, transitions_after_failure); + let tick_after_failure = scripted.clock.tick(); + let messages_after_failure = scripted.total_messages; + let resumed = scripted.run_random(); + assert!(!resumed.all_invariants_held); + assert_eq!(scripted.clock.tick(), tick_after_failure); + assert_eq!(scripted.total_transitions, transitions_after_failure); + assert_eq!(scripted.total_messages, messages_after_failure); + + let mut random = SimActorSystem::new(config); + random.register_actor("Job:1", Box::new(CallbackFailureHandler::new())); + random.set_integration_responses(responses); + let result = random.run_random(); + assert!(!result.all_invariants_held); + assert_eq!( + result.messages, 1, + "the delayed action owns one reservation" + ); + assert_eq!(result.execution_errors.len(), 1); + assert!( + result.execution_errors[0] + .description + .contains("callback rejected") + ); + let tick_after_random_failure = random.clock.tick(); + let resumed_random = random.run_random(); + assert_eq!(random.clock.tick(), tick_after_random_failure); + assert_eq!(resumed_random.transitions, result.transitions); + assert_eq!(resumed_random.messages, result.messages); +} + +#[test] +fn final_tick_drains_every_due_message_when_batch_exceeds_budget() { + let config = SimActorSystemConfig { + seed: 3, + max_ticks: 2, + faults: FaultConfig { + message_delay_prob: 1.0, + max_delay_ticks: 2, + ..FaultConfig::none() + }, + max_actions_per_actor: 2, + message_batch_budget: 1, + reaction_budget_per_tick: 1, + }; + let mut sim = SimActorSystem::new(config); + sim.register_actor("Job:1", Box::new(CallbackFailureHandler::new())); + + let result = sim.run_random(); + + assert_eq!(result.dropped, 0); + assert_eq!(result.messages, 2); + assert_eq!( + result.transitions, 2, + "every message due on the final tick must leave scheduler ownership" + ); +} + +#[test] +fn final_tick_batches_share_one_reaction_budget() { + let config = SimActorSystemConfig { + seed: 3, + max_ticks: 2, + faults: FaultConfig { + message_delay_prob: 1.0, + max_delay_ticks: 2, + ..FaultConfig::none() + }, + max_actions_per_actor: 2, + message_batch_budget: 1, + reaction_budget_per_tick: 1, + }; + let mut sim = SimActorSystem::new(config); + sim.register_actor("Job:1", Box::new(CallbackFailureHandler::new())); + sim.set_integration_responses(SimIntegrationResponses::new().on_trigger( + "Job", + "integration", + "Complete", + )); + + let result = sim.run_random(); + + assert!(!result.all_invariants_held); + assert_eq!(result.messages, 2); + assert_eq!(result.transitions, 3); + assert_eq!(result.execution_errors.len(), 1); + assert!( + result.execution_errors[0] + .description + .contains("budget exhausted after 1 reactions") + ); +} + +#[test] +fn callback_cascade_fails_when_reaction_budget_is_exhausted() { + let config = SimActorSystemConfig { + reaction_budget_per_tick: 2, + ..Default::default() + }; + let mut sim = SimActorSystem::new(config); + sim.register_actor("Job:1", Box::new(CallbackFailureHandler::new())); + sim.set_integration_responses(SimIntegrationResponses::new().on_trigger( + "Job", + "integration", + "Loop", + )); + + let error = sim.step("Job:1", "Start", "{}").unwrap_err(); + assert!(error.contains("budget exhausted after 2 reactions")); + assert_eq!(sim.execution_errors().len(), 1); + let transitions_after_failure = sim.total_transitions; + let retry_error = sim.step("Job:1", "Complete", "{}").unwrap_err(); + assert!(retry_error.contains("simulation run is invalid")); + assert_eq!(sim.total_transitions, transitions_after_failure); + let tick_after_failure = sim.clock.tick(); + let messages_after_failure = sim.total_messages; + let resumed = sim.run_random(); + assert!(!resumed.all_invariants_held); + assert_eq!(sim.clock.tick(), tick_after_failure); + assert_eq!(sim.total_transitions, transitions_after_failure); + assert_eq!(sim.total_messages, messages_after_failure); +} + +#[test] +fn run_record_equality() { + let r1 = RunRecord { + seed: 42, + transitions: vec![( + 1, + "a".into(), + "Submit".into(), + "Draft".into(), + "Submitted".into(), + )], + events: BTreeMap::new(), + final_states: vec![], + invariant_results: vec![], + }; + let r2 = r1.clone(); + assert_eq!(r1, r2); +} + +#[test] +fn run_record_inequality_on_seed() { + let r1 = RunRecord { + seed: 42, + transitions: vec![], + events: BTreeMap::new(), + final_states: vec![], + invariant_results: vec![], + }; + let r2 = RunRecord { + seed: 99, + ..r1.clone() + }; + assert_ne!(r1, r2); +} diff --git a/crates/temper-server/src/observe/verification/simulation.rs b/crates/temper-server/src/observe/verification/simulation.rs index 64a71bc7b..bef5008f9 100644 --- a/crates/temper-server/src/observe/verification/simulation.rs +++ b/crates/temper-server/src/observe/verification/simulation.rs @@ -36,6 +36,7 @@ pub(crate) async fn handle_run_simulation( num_actors: 3, max_actions_per_actor: 20, max_counter: 2, + message_batch_budget: 1_024, faults: temper_runtime::scheduler::FaultConfig::light(), }; temper_verify::run_simulation_from_ioa(&ioa_source, &config) diff --git a/crates/temper-server/tests/gmail_oauth_dst.rs b/crates/temper-server/tests/gmail_oauth_dst.rs index f597da7b2..a23c2c42c 100644 --- a/crates/temper-server/tests/gmail_oauth_dst.rs +++ b/crates/temper-server/tests/gmail_oauth_dst.rs @@ -230,6 +230,8 @@ fn random_no_faults() { max_ticks: 200, faults: FaultConfig::none(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -256,6 +258,8 @@ fn random_light_faults() { max_ticks: 300, faults: FaultConfig::light(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -281,6 +285,8 @@ fn random_heavy_faults() { max_ticks: 500, faults: FaultConfig::heavy(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -309,6 +315,8 @@ fn run_determinism_trial(seed: u64) -> Vec<(String, String, usize, usize)> { max_ticks: 300, faults: FaultConfig::light(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -360,6 +368,8 @@ fn multi_seed_sweep() { max_ticks: 100, faults: FaultConfig::light(), max_actions_per_actor: 20, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); diff --git a/crates/temper-server/tests/reaction_cascade.rs b/crates/temper-server/tests/reaction_cascade.rs index 977d5e809..8c10ab5c1 100644 --- a/crates/temper-server/tests/reaction_cascade.rs +++ b/crates/temper-server/tests/reaction_cascade.rs @@ -81,6 +81,8 @@ fn sim_config() -> SimActorSystemConfig { max_ticks: 100, faults: FaultConfig::none(), max_actions_per_actor: 20, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, } } diff --git a/crates/temper-verify/src/cascade.rs b/crates/temper-verify/src/cascade.rs index b57b7bd3f..c08e5475b 100644 --- a/crates/temper-verify/src/cascade.rs +++ b/crates/temper-verify/src/cascade.rs @@ -453,6 +453,7 @@ impl VerificationCascade { num_actors: 3, max_actions_per_actor: 20, max_counter: self.max_counter, + message_batch_budget: 1_024, faults: FaultConfig::light(), }; diff --git a/crates/temper-verify/src/simulation.rs b/crates/temper-verify/src/simulation.rs index f68cb27a1..3463552c2 100644 --- a/crates/temper-verify/src/simulation.rs +++ b/crates/temper-verify/src/simulation.rs @@ -33,6 +33,8 @@ pub struct SimConfig { pub max_actions_per_actor: usize, /// Maximum counter value for bounded model checking. pub max_counter: usize, + /// Maximum ready messages transferred in one bounded drain batch. + pub message_batch_budget: usize, /// Fault injection configuration. pub faults: FaultConfig, } @@ -45,6 +47,7 @@ impl Default for SimConfig { num_actors: 3, max_actions_per_actor: 20, max_counter: 2, + message_batch_budget: 1_024, faults: FaultConfig::none(), } } @@ -153,12 +156,21 @@ pub fn run_multi_seed_simulation_from_ioa( } fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationResult { - let mut sched = SimScheduler::new(config.seed, config.faults.clone()); + assert!( + config.message_batch_budget > 0, + "message batch budget must be positive" + ); + let mailbox_budget = usize::try_from(config.max_ticks) + .expect("maximum ticks must fit the platform address space") + .max(1); + let mut sched = + SimScheduler::with_mailbox_budget(config.seed, config.faults.clone(), mailbox_budget); let mut rng = DeterministicRng::new(config.seed.wrapping_add(1)); // Initialize actors let mut actor_states: Vec<(String, TemperModelState)> = Vec::new(); let mut actor_action_counts: Vec = Vec::new(); + let mut actor_in_flight_actions: Vec = Vec::new(); for i in 0..config.num_actors { let actor_id = format!("entity-{i}"); @@ -166,11 +178,13 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes let initial = model.init_states()[0].clone(); actor_states.push((actor_id, initial)); actor_action_counts.push(0); + actor_in_flight_actions.push(0); } let mut violations = Vec::new(); let mut total_transitions: u64 = 0; let mut total_messages: u64 = 0; + let mut observed_scheduler_drops = 0; // Main simulation loop for tick in 0..config.max_ticks { @@ -179,66 +193,84 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes } let actor_idx = rng.next_bound(actor_states.len()); - let (ref actor_id, ref current_state) = actor_states[actor_idx]; - - if actor_action_counts[actor_idx] >= config.max_actions_per_actor { - continue; - } - - if sched.actor_state(actor_id) == Some(&SimActorState::Crashed) { - continue; + let actor_id = actor_states[actor_idx].0.clone(); + let can_schedule = actor_action_counts[actor_idx] + actor_in_flight_actions[actor_idx] + < config.max_actions_per_actor + && sched.actor_state(&actor_id) != Some(&SimActorState::Crashed); + + if can_schedule { + let mut valid_actions = Vec::new(); + model.actions(&actor_states[actor_idx].1, &mut valid_actions); + + if !valid_actions.is_empty() { + let action_idx = rng.next_bound(valid_actions.len()); + let action = valid_actions[action_idx].clone(); + sched.send( + "sim-driver", + &actor_id, + &action.name, + &serde_json::to_string(&action).unwrap_or_default(), + ); + actor_in_flight_actions[actor_idx] += 1; + total_messages += 1; + } } - let mut valid_actions = Vec::new(); - model.actions(current_state, &mut valid_actions); - - if valid_actions.is_empty() { - continue; + // Logical time and ready delivery progress independently of whether a + // new action was eligible this tick. + sched.tick(); + for dropped in &sched.dropped_log()[observed_scheduler_drops..] { + if dropped.from == "sim-driver" + && let Some(idx) = actor_states.iter().position(|(id, _)| id == &dropped.to) + { + assert!( + actor_in_flight_actions[idx] > 0, + "dropped action must own a reservation" + ); + actor_in_flight_actions[idx] -= 1; + } } + observed_scheduler_drops = sched.dropped_log().len(); + loop { + let delivered = sched.drain_ready(config.message_batch_budget); - let action_idx = rng.next_bound(valid_actions.len()); - let action = valid_actions[action_idx].clone(); - - let action_name = action.name.clone(); - sched.send( - "sim-driver", - actor_id, - &action_name, - &serde_json::to_string(&action).unwrap_or_default(), - ); - 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 }; + 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]; - - let action: TemperModelAction = match serde_json::from_str(&msg.payload) { - Ok(a) => a, - Err(_) => continue, - }; - - 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, + assert!( + actor_in_flight_actions[idx] > 0, + "delivered action must own a reservation" ); - - actor_states[idx].1 = new_state; - actor_action_counts[idx] += 1; - total_transitions += 1; + actor_in_flight_actions[idx] -= 1; + + let (ref target_id, ref state_before) = actor_states[idx]; + + let action: TemperModelAction = match serde_json::from_str(&msg.payload) { + Ok(a) => a, + Err(_) => continue, + }; + + 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; + } + } + if tick + 1 < config.max_ticks || !sched.has_ready_messages() { + break; } } - - sched.tick(); } // Post-simulation liveness checks @@ -413,171 +445,5 @@ fn sim_kind_violated( } #[cfg(test)] -mod tests { - use super::*; - - const ORDER_IOA: &str = include_str!("../../../test-fixtures/specs/order.ioa.toml"); - - #[test] - fn test_simulation_no_faults() { - let config = SimConfig { - seed: 42, - max_ticks: 200, - num_actors: 3, - max_actions_per_actor: 15, - max_counter: 2, - faults: FaultConfig::none(), - }; - - let result = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); - assert!( - result.all_invariants_held, - "No invariant violations expected without faults, got: {:?}", - result.violations - ); - assert!( - result.total_transitions > 0, - "Should have applied some transitions" - ); - } - - #[test] - fn test_simulation_light_faults() { - let config = SimConfig { - seed: 123, - max_ticks: 300, - num_actors: 3, - max_actions_per_actor: 20, - max_counter: 2, - faults: FaultConfig::light(), - }; - - let result = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); - assert!( - result.all_invariants_held, - "No invariant violations expected with light faults, got: {:?}", - result.violations - ); - } - - #[test] - fn test_simulation_heavy_faults() { - let config = SimConfig { - seed: 456, - max_ticks: 300, - num_actors: 5, - max_actions_per_actor: 15, - max_counter: 2, - faults: FaultConfig::heavy(), - }; - - let result = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); - assert!( - result.all_invariants_held, - "Invariants must hold even under heavy faults, got: {:?}", - result.violations - ); - assert!( - result.total_dropped > 0 || result.total_messages > 0, - "Should have processed messages" - ); - } - - #[test] - fn test_simulation_is_reproducible() { - let config = SimConfig { - seed: 999, - max_ticks: 100, - num_actors: 2, - max_actions_per_actor: 10, - max_counter: 2, - faults: FaultConfig::light(), - }; - - let result1 = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); - let result2 = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); - - assert_eq!( - result1.total_transitions, result2.total_transitions, - "Same seed must produce same number of transitions" - ); - assert_eq!( - result1.total_messages, result2.total_messages, - "Same seed must produce same number of messages" - ); - - for (i, ((id1, s1), (id2, s2))) in result1 - .actor_final_states - .iter() - .zip(result2.actor_final_states.iter()) - .enumerate() - { - assert_eq!(id1, id2, "Actor {i} ID mismatch"); - assert_eq!(s1.status, s2.status, "Actor {i} status mismatch"); - assert_eq!(s1.counters, s2.counters, "Actor {i} counters mismatch"); - } - } - - #[test] - fn test_simulation_different_seeds_diverge() { - let config1 = SimConfig::default().with_seed(42); - let config2 = SimConfig::default().with_seed(9999); - - let result1 = run_simulation_from_ioa(ORDER_IOA, &config1).unwrap(); - let result2 = run_simulation_from_ioa(ORDER_IOA, &config2).unwrap(); - - assert!(result1.total_transitions > 0); - assert!(result2.total_transitions > 0); - } - - #[test] - fn test_multi_seed_simulation() { - let config = SimConfig { - seed: 1, - max_ticks: 100, - num_actors: 2, - max_actions_per_actor: 10, - max_counter: 2, - faults: FaultConfig::light(), - }; - - let results = run_multi_seed_simulation_from_ioa(ORDER_IOA, &config, 10).unwrap(); - assert_eq!(results.len(), 10); - - for (i, result) in results.iter().enumerate() { - assert!( - result.all_invariants_held, - "Seed {} failed with violations: {:?}", - result.seed, result.violations - ); - assert_eq!(result.seed, 1 + i as u64); - } - } - - #[test] - fn test_simulation_result_contains_final_states() { - let config = SimConfig { - seed: 77, - max_ticks: 50, - num_actors: 2, - max_actions_per_actor: 5, - max_counter: 2, - faults: FaultConfig::none(), - }; - - let result = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); - assert_eq!(result.actor_final_states.len(), 2); - - let model = build_model_from_ioa(ORDER_IOA, config.max_counter).unwrap(); - - for (id, state) in &result.actor_final_states { - assert!(id.starts_with("entity-")); - assert!( - model.states.contains(&state.status), - "Status '{}' not in spec states {:?}", - state.status, - model.states - ); - } - } -} +#[path = "simulation/tests.rs"] +mod tests; diff --git a/crates/temper-verify/src/simulation/tests.rs b/crates/temper-verify/src/simulation/tests.rs new file mode 100644 index 000000000..c8caf874f --- /dev/null +++ b/crates/temper-verify/src/simulation/tests.rs @@ -0,0 +1,231 @@ +use super::*; + +const ORDER_IOA: &str = include_str!("../../../../test-fixtures/specs/order.ioa.toml"); + +#[test] +fn test_simulation_no_faults() { + let config = SimConfig { + seed: 42, + max_ticks: 200, + num_actors: 3, + max_actions_per_actor: 15, + max_counter: 2, + message_batch_budget: 1_024, + faults: FaultConfig::none(), + }; + + let result = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); + assert!( + result.all_invariants_held, + "No invariant violations expected without faults, got: {:?}", + result.violations + ); + assert!( + result.total_transitions > 0, + "Should have applied some transitions" + ); +} + +#[test] +fn test_simulation_light_faults() { + let config = SimConfig { + seed: 123, + max_ticks: 300, + num_actors: 3, + max_actions_per_actor: 20, + max_counter: 2, + message_batch_budget: 1_024, + faults: FaultConfig::light(), + }; + + let result = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); + assert!( + result.all_invariants_held, + "No invariant violations expected with light faults, got: {:?}", + result.violations + ); +} + +#[test] +fn test_simulation_heavy_faults() { + let config = SimConfig { + seed: 456, + max_ticks: 300, + num_actors: 5, + max_actions_per_actor: 15, + max_counter: 2, + message_batch_budget: 1_024, + faults: FaultConfig::heavy(), + }; + + let result = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); + assert!( + result.all_invariants_held, + "Invariants must hold even under heavy faults, got: {:?}", + result.violations + ); + assert!( + result.total_dropped > 0 || result.total_messages > 0, + "Should have processed messages" + ); +} + +#[test] +fn delayed_message_due_on_final_tick_is_delivered() { + let config = SimConfig { + seed: 1, + max_ticks: 2, + num_actors: 1, + max_actions_per_actor: 1, + max_counter: 2, + message_batch_budget: 1, + faults: FaultConfig { + message_delay_prob: 1.0, + max_delay_ticks: 2, + message_drop_prob: 0.0, + actor_crash_prob: 0.0, + actor_restart_prob: 0.0, + }, + }; + + let result = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); + + assert_eq!(result.total_dropped, 0, "the message was not fault-dropped"); + assert_eq!( + result.total_messages, 1, + "the delayed action must reserve its per-actor budget" + ); + assert_eq!( + result.total_transitions, 1, + "the delivery due on the final tick must be applied exactly once" + ); +} + +#[test] +fn final_tick_drains_every_due_message_when_batch_exceeds_budget() { + let config = SimConfig { + seed: 3, + max_ticks: 2, + num_actors: 1, + max_actions_per_actor: 2, + max_counter: 2, + message_batch_budget: 1, + faults: FaultConfig { + message_delay_prob: 1.0, + max_delay_ticks: 2, + message_drop_prob: 0.0, + actor_crash_prob: 0.0, + actor_restart_prob: 0.0, + }, + }; + + let result = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); + + assert_eq!(result.total_dropped, 0); + assert_eq!(result.total_messages, 2); + assert_eq!( + result.total_transitions, 2, + "every message due on the final tick must leave scheduler ownership" + ); +} + +#[test] +fn test_simulation_is_reproducible() { + let config = SimConfig { + seed: 999, + max_ticks: 100, + num_actors: 2, + max_actions_per_actor: 10, + max_counter: 2, + message_batch_budget: 1_024, + faults: FaultConfig::light(), + }; + + let result1 = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); + let result2 = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); + + assert_eq!( + result1.total_transitions, result2.total_transitions, + "Same seed must produce same number of transitions" + ); + assert_eq!( + result1.total_messages, result2.total_messages, + "Same seed must produce same number of messages" + ); + + for (i, ((id1, s1), (id2, s2))) in result1 + .actor_final_states + .iter() + .zip(result2.actor_final_states.iter()) + .enumerate() + { + assert_eq!(id1, id2, "Actor {i} ID mismatch"); + assert_eq!(s1.status, s2.status, "Actor {i} status mismatch"); + assert_eq!(s1.counters, s2.counters, "Actor {i} counters mismatch"); + } +} + +#[test] +fn test_simulation_different_seeds_diverge() { + let config1 = SimConfig::default().with_seed(42); + let config2 = SimConfig::default().with_seed(9999); + + let result1 = run_simulation_from_ioa(ORDER_IOA, &config1).unwrap(); + let result2 = run_simulation_from_ioa(ORDER_IOA, &config2).unwrap(); + + assert!(result1.total_transitions > 0); + assert!(result2.total_transitions > 0); +} + +#[test] +fn test_multi_seed_simulation() { + let config = SimConfig { + seed: 1, + max_ticks: 100, + num_actors: 2, + max_actions_per_actor: 10, + max_counter: 2, + message_batch_budget: 1_024, + faults: FaultConfig::light(), + }; + + let results = run_multi_seed_simulation_from_ioa(ORDER_IOA, &config, 10).unwrap(); + assert_eq!(results.len(), 10); + + for (i, result) in results.iter().enumerate() { + assert!( + result.all_invariants_held, + "Seed {} failed with violations: {:?}", + result.seed, result.violations + ); + assert_eq!(result.seed, 1 + i as u64); + } +} + +#[test] +fn test_simulation_result_contains_final_states() { + let config = SimConfig { + seed: 77, + max_ticks: 50, + num_actors: 2, + max_actions_per_actor: 5, + max_counter: 2, + message_batch_budget: 1_024, + faults: FaultConfig::none(), + }; + + let result = run_simulation_from_ioa(ORDER_IOA, &config).unwrap(); + assert_eq!(result.actor_final_states.len(), 2); + + let model = build_model_from_ioa(ORDER_IOA, config.max_counter).unwrap(); + + for (id, state) in &result.actor_final_states { + assert!(id.starts_with("entity-")); + assert!( + model.states.contains(&state.status), + "Status '{}' not in spec states {:?}", + state.status, + model.states + ); + } +} diff --git a/docs/adrs/0177-single-owner-simulation-delivery.md b/docs/adrs/0177-single-owner-simulation-delivery.md new file mode 100644 index 000000000..b9cd3b1a2 --- /dev/null +++ b/docs/adrs/0177-single-owner-simulation-delivery.md @@ -0,0 +1,97 @@ +# ADR-0177: Single-owner simulation delivery + +- Status: Accepted +- Date: 2026-07-14 +- Deciders: Temper core maintainers +- Related: + - ARN-236: Simulation delayed-message ownership correctness + - `crates/temper-runtime/src/scheduler/core.rs` + - `crates/temper-runtime/src/scheduler/sim_actor_system.rs` + - `crates/temper-verify/src/simulation.rs` + +## Context + +The deterministic scheduler currently gives every due message two owners. `SimScheduler::tick` enqueues a message in the target mailbox and also returns a clone. The runtime actor simulator and the verifier simulator process the returned clone without consuming the mailbox entry. Both drivers then perform a second tick whose returned messages are ignored. A delivery can therefore be applied while its mailbox copy remains queued, or be moved from the pending heap into a mailbox by the ignored tick and never be applied. + +Integration callback delivery has a related truthfulness gap. The actor simulator recursively invokes callback actions and discards their errors, so a rejected callback can still yield a successful simulation result. + +These paths make deterministic simulation disagree with the delivery contract it claims to verify. The correction spans the shared scheduler and both simulation drivers, so it requires an architectural decision before implementation. + +## Decision + +### A scheduler tick never transfers processing ownership + +`SimScheduler::tick` advances logical time, applies deterministic faults, and moves due messages from the pending heap into actor mailboxes. It does not return message clones. + +### One deterministic mailbox drain owns processing + +The scheduler exposes one budgeted drain operation. It consumes ready messages by cycling through `BTreeMap` actor order and preserving FIFO order within each actor. The cursor carries across drains so a small budget cannot permanently starve a later mailbox. Both the runtime actor simulator and verifier simulator process only messages returned by this consuming drain. No driver may inspect a delivery through a parallel return path. + +The drain accepts a message batch budget. A tick budget already bounds elapsed logical time; runtime and verifier configurations make the maximum transfer per drain call explicit. Reaching a batch budget preserves undrained messages in scheduler ownership rather than dropping them. Ordinary ticks perform one batch; the terminal tick repeats bounded batches until every message already due at that logical time is consumed. The ready-mailbox retention budget bounds that final flush without advancing time or admitting new actions. + +Scheduler mailboxes also have an explicit per-actor retention budget. Drivers derive it from their maximum tick budget, which bounds the number of actions they can enqueue. Direct scheduler users receive a conservative default and fail fast if they exceed it. + +### Reactions are iterative, budgeted, and fallible + +Integration callbacks are drained iteratively from their reaction queue under an explicit per-tick reaction budget shared by every message batch in that tick, including terminal-flush batches. Callback rejection is recorded as a simulation execution error and makes the run unsuccessful. Callback dispatch does not recursively start another independent drain. + +In scripted mode, a primary action commits before its callbacks run. A callback error therefore reports partial completion and must not be interpreted as rollback or as permission to retry the primary action. + +### Simulation success includes delivery execution + +A successful simulation requires both invariant preservation and absence of delivery/callback execution errors. Results retain explicit error evidence so callers can distinguish a modeled invariant violation from a driver failure. + +## Rollout Plan + +1. Add behavioral regressions against the current competing-ownership behavior. +2. Change the scheduler contract and migrate both in-repository simulation drivers together. +3. Add deterministic delayed-delivery and callback-failure coverage, then run the full verification cascade. + +## Readiness Gates + +- A processed message leaves no mailbox clone behind. +- Multiple actors and multiple due messages drain in reproducible actor/FIFO order. +- A delivery becoming due on the final observed tick is consumed, not discarded. +- Callback rejection is visible in the result and fails verification. +- Runtime and verifier compile against the same consuming scheduler API. + +## Consequences + +### Positive + +- Every ready message has one processing owner and one consumption point. +- Delivery order remains deterministic and replayable. +- Budget exhaustion defers work without silently losing it. +- Simulation results report callback rejection truthfully. + +### Negative + +- Callers that used the vector returned by `tick` must migrate to the consuming drain. +- A small message batch budget requires more bounded drain calls on the terminal tick, but cannot discard work already due at the simulation horizon. + +### Risks + +- A drain order change can alter existing seeded traces. Deterministic replay remains stable after the contract change, and regression tests pin the new ordering. +- Too-small budgets can reduce exploration depth. Defaults cover the configured actor/action bounds, and remaining work is retained rather than discarded. + +### DST Compliance + +- Scheduler collections remain `BTreeMap`, `BinaryHeap`, and `VecDeque`; no nondeterministic iteration is introduced. +- Logical time and the seeded scheduler RNG remain the only time and randomness sources. +- No threads, wall-clock calls, ambient I/O, or new determinism exceptions are introduced. + +## Non-Goals + +- Changing production actor mailbox semantics. +- Changing the probability or ordering of configured scheduler faults. +- Executing real external integrations inside deterministic simulation. + +## Alternatives Considered + +1. **Process only the vector returned by `tick` and remove mailboxes** — Rejected because the scheduler's mailbox is the natural ownership boundary and is required for quiescence and budget-preserving delivery. +2. **Keep both paths and explicitly remove returned clones from mailboxes** — Rejected because clone correlation adds a compensating protocol while preserving two owners. +3. **Let each simulator implement its own mailbox iteration** — Rejected because independent drivers can drift again and duplicate ordering/budget logic. + +## Rollback Policy + +Revert the scheduler and both driver migrations together. A partial rollback is invalid because it would restore competing ownership or leave one simulator unable to consume deliveries. diff --git a/reference-apps/crucible/tests/crucible_dst.rs b/reference-apps/crucible/tests/crucible_dst.rs index 090fee270..ed763979f 100644 --- a/reference-apps/crucible/tests/crucible_dst.rs +++ b/reference-apps/crucible/tests/crucible_dst.rs @@ -557,6 +557,8 @@ fn random_all_entities_no_faults() { max_ticks: 200, faults: FaultConfig::none(), max_actions_per_actor: 20, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -648,6 +650,8 @@ fn run_determinism_trial(seed: u64) -> Vec<(String, String, usize, usize)> { max_ticks: 200, faults: FaultConfig::light(), max_actions_per_actor: 20, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); diff --git a/reference-apps/ecommerce/tests/ecommerce_dst.rs b/reference-apps/ecommerce/tests/ecommerce_dst.rs index 73427de8b..7330a3345 100644 --- a/reference-apps/ecommerce/tests/ecommerce_dst.rs +++ b/reference-apps/ecommerce/tests/ecommerce_dst.rs @@ -407,6 +407,8 @@ fn random_order_no_faults() { max_ticks: 200, faults: FaultConfig::none(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -436,6 +438,8 @@ fn random_all_entities_light_faults() { max_ticks: 300, faults: FaultConfig::light(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -475,6 +479,8 @@ fn random_all_entities_heavy_faults() { max_ticks: 500, faults: FaultConfig::heavy(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -517,6 +523,8 @@ fn run_determinism_trial(seed: u64) -> Vec<(String, String, usize, usize)> { max_ticks: 300, faults: FaultConfig::light(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -582,6 +590,8 @@ fn multi_seed_sweep_all_entities() { max_ticks: 100, faults: FaultConfig::light(), max_actions_per_actor: 20, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); diff --git a/reference-apps/ecommerce/tests/interactive_demo.rs b/reference-apps/ecommerce/tests/interactive_demo.rs index 4ca95ffc3..9bda2a106 100644 --- a/reference-apps/ecommerce/tests/interactive_demo.rs +++ b/reference-apps/ecommerce/tests/interactive_demo.rs @@ -249,6 +249,8 @@ fn interactive_full_pipeline() { max_ticks: 300, faults: FaultConfig::heavy(), max_actions_per_actor: 25, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -291,6 +293,8 @@ fn interactive_full_pipeline() { max_ticks: 200, faults: FaultConfig::light(), max_actions_per_actor: 20, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); let handler = EntityActorHandler::new( diff --git a/reference-apps/oncall/tests/interactive_demo.rs b/reference-apps/oncall/tests/interactive_demo.rs index e46dfd7fc..47ad0c500 100644 --- a/reference-apps/oncall/tests/interactive_demo.rs +++ b/reference-apps/oncall/tests/interactive_demo.rs @@ -167,6 +167,8 @@ fn interactive_full_triage_pipeline() { max_ticks: 200, faults: FaultConfig::light(), max_actions_per_actor: 20, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); diff --git a/reference-apps/oncall/tests/oncall_dst.rs b/reference-apps/oncall/tests/oncall_dst.rs index aec520d55..700ff83ed 100644 --- a/reference-apps/oncall/tests/oncall_dst.rs +++ b/reference-apps/oncall/tests/oncall_dst.rs @@ -406,6 +406,8 @@ fn random_page_light_faults() { max_ticks: 200, faults: FaultConfig::light(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -439,6 +441,8 @@ fn random_all_entities_heavy_faults() { max_ticks: 500, faults: FaultConfig::heavy(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -490,6 +494,8 @@ fn random_multi_seed_sweep() { max_ticks: 100, faults: FaultConfig::light(), max_actions_per_actor: 20, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config); @@ -540,6 +546,8 @@ fn run_determinism_trial(seed: u64) -> Vec<(String, String, usize, usize)> { max_ticks: 300, faults: FaultConfig::light(), max_actions_per_actor: 30, + message_batch_budget: 1_024, + reaction_budget_per_tick: 1_024, }; let mut sim = SimActorSystem::new(config);