mongreldb_sim/
scenario.rs

1//! Scenario runner: schedule driving, fault orchestration, the event
2//! log, and failure artifacts (spec section 9.5, FND-005).
3//!
4//! A [`Scenario`] wires a [`Runtime`], a [`Network`], and one
5//! [`VirtualDisk`] per node together and drives the schedule to
6//! completion. While any task is ready, a seeded pick chooses the next
7//! step; when nothing can run, the clock advances on idle to the next
8//! timer or message delivery. A scenario that can neither run nor
9//! advance while tasks remain is reported as a deadlock; a step budget
10//! guards against livelock.
11//!
12//! Every failure (deadlock, step limit, or a caught task panic)
13//! persists the seed and the full event log as JSON so CI can archive
14//! the exact repro — see [`failure_dir`].
15
16use crate::clock::{Micros, SkewSchedule};
17use crate::disk::VirtualDisk;
18use crate::network::{DeliveryOutcome, DropReason, LinkConfig, Network, NetworkStats, NodeId};
19use crate::rng::Seed;
20use crate::runtime::{Runtime, TaskBody, TaskId};
21use serde::{Deserialize, Serialize};
22use std::any::Any;
23use std::collections::{BTreeMap, BTreeSet};
24use std::panic::{catch_unwind, resume_unwind, AssertUnwindSafe};
25use std::path::{Path, PathBuf};
26use std::sync::atomic::{AtomicU32, Ordering};
27use std::{env, fs, io};
28
29/// One recorded step of a simulation run, in execution order. Two runs
30/// of the same scenario with the same seed produce identical sequences.
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32pub enum Event {
33    /// A task was spawned for a node process.
34    TaskSpawned {
35        /// New task id.
36        task: TaskId,
37        /// Owning node.
38        node: NodeId,
39        /// Process name.
40        name: String,
41    },
42    /// A task finished.
43    TaskDone {
44        /// Finished task id.
45        task: TaskId,
46        /// Owning node.
47        node: NodeId,
48        /// Process name.
49        name: String,
50    },
51    /// A message entered the flight queue.
52    MessageSent {
53        /// Sender.
54        from: NodeId,
55        /// Destination.
56        to: NodeId,
57        /// Sequence number.
58        seq: u64,
59        /// Scheduled delivery time.
60        deliver_at: Micros,
61    },
62    /// A message never reached its destination.
63    MessageDropped {
64        /// Sender.
65        from: NodeId,
66        /// Destination.
67        to: NodeId,
68        /// Sequence number.
69        seq: u64,
70        /// Why it was dropped.
71        reason: DropReason,
72    },
73    /// A send scheduled an extra copy.
74    MessageDuplicated {
75        /// Sender.
76        from: NodeId,
77        /// Destination.
78        to: NodeId,
79        /// Sequence number of the original copy.
80        seq: u64,
81    },
82    /// A send undercut a previously scheduled send on the same link.
83    MessageReordered {
84        /// Sender.
85        from: NodeId,
86        /// Destination.
87        to: NodeId,
88        /// Sequence number.
89        seq: u64,
90    },
91    /// A message reached a destination inbox.
92    MessageDelivered {
93        /// Sender.
94        from: NodeId,
95        /// Destination.
96        to: NodeId,
97        /// Sequence number.
98        seq: u64,
99    },
100    /// A write landed in a file's pending stage.
101    DiskWrite {
102        /// Owning node.
103        node: NodeId,
104        /// File path.
105        path: String,
106        /// Bytes written.
107        len: usize,
108    },
109    /// A write was truncated (torn write).
110    DiskTornWrite {
111        /// Owning node.
112        node: NodeId,
113        /// File path.
114        path: String,
115        /// Bytes requested.
116        requested: usize,
117        /// Bytes actually written.
118        written: usize,
119    },
120    /// An injected write failure fired.
121    DiskWriteFailed {
122        /// Owning node.
123        node: NodeId,
124        /// File path.
125        path: String,
126    },
127    /// Pending bytes became durable.
128    DiskFsync {
129        /// Owning node.
130        node: NodeId,
131        /// File path.
132        path: String,
133        /// Total durable bytes after the sync.
134        durable_len: usize,
135    },
136    /// An injected fsync failure fired.
137    DiskFsyncFailed {
138        /// Owning node.
139        node: NodeId,
140        /// File path.
141        path: String,
142    },
143    /// A node process crashed; volatile state was dropped.
144    Crash {
145        /// Crashed node.
146        node: NodeId,
147    },
148    /// A crashed node process restarted.
149    Restart {
150        /// Restarted node.
151        node: NodeId,
152    },
153    /// Connectivity between two node sets was cut.
154    Partition {
155        /// First side.
156        group_a: Vec<NodeId>,
157        /// Second side.
158        group_b: Vec<NodeId>,
159    },
160    /// All partition cuts were removed.
161    Healed,
162    /// A node's skew schedule was replaced.
163    SkewUpdated {
164        /// Affected node.
165        node: NodeId,
166    },
167    /// The virtual clock advanced on idle.
168    ClockAdvanced {
169        /// Previous time.
170        from: Micros,
171        /// New time.
172        to: Micros,
173    },
174    /// An application-level record from `NodeContext::log`.
175    Custom {
176        /// Logging node.
177        node: NodeId,
178        /// Free-form message.
179        message: String,
180    },
181}
182
183/// Why a scenario run failed.
184#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
185pub enum ScenarioError {
186    /// No task can run, no message is in flight, no timer is pending,
187    /// and unfinished tasks remain.
188    #[error("deadlock detected at t={at_micros}: {waiting} task(s) waiting, none runnable")]
189    DeadlockDetected {
190        /// Virtual time of the detection.
191        at_micros: Micros,
192        /// Tasks still unfinished.
193        waiting: usize,
194    },
195    /// The step budget ran out (probable livelock).
196    #[error("step limit {limit} exceeded (possible livelock)")]
197    StepLimitExceeded {
198        /// The budget that was exceeded.
199        limit: u64,
200    },
201    /// A task body panicked; the panic is re-raised after the failure
202    /// artifact is persisted.
203    #[error("task panicked: {0}")]
204    TaskPanicked(String),
205}
206
207/// Summary of a completed run.
208#[derive(Debug, Clone, PartialEq, Eq)]
209pub struct RunOutcome {
210    /// The seed the run used.
211    pub seed: Seed,
212    /// Task steps executed.
213    pub steps: u64,
214    /// Final virtual time.
215    pub sim_time_micros: Micros,
216}
217
218/// A registered node process: its name plus the factory that rebuilds
219/// its task body on (re)start.
220type ProcessFactory = (String, Box<dyn FnMut() -> TaskBody>);
221
222/// A wired-up simulation: runtime, network, per-node disks, process
223/// factories for restart, and the event log.
224pub struct Scenario {
225    seed: Seed,
226    runtime: Runtime,
227    network: Network,
228    disks: BTreeMap<NodeId, VirtualDisk>,
229    factories: BTreeMap<NodeId, ProcessFactory>,
230    crashed: BTreeSet<NodeId>,
231    events: Vec<Event>,
232}
233
234impl Scenario {
235    /// A scenario with default link behavior and no nodes.
236    pub fn new(seed: Seed) -> Self {
237        Self {
238            seed,
239            runtime: Runtime::new(seed),
240            network: Network::new(LinkConfig::default()),
241            disks: BTreeMap::new(),
242            factories: BTreeMap::new(),
243            crashed: BTreeSet::new(),
244            events: Vec::new(),
245        }
246    }
247
248    /// The seed this scenario runs with.
249    pub fn seed(&self) -> Seed {
250        self.seed
251    }
252
253    /// Current virtual time.
254    pub fn now(&self) -> Micros {
255        self.runtime.clock().now()
256    }
257
258    /// The recorded event log, in execution order.
259    pub fn events(&self) -> &[Event] {
260        &self.events
261    }
262
263    /// Cumulative network counters.
264    pub fn network_stats(&self) -> NetworkStats {
265        self.network.stats()
266    }
267
268    /// A node's disk, if the node exists.
269    pub fn disk(&self, node: NodeId) -> Option<&VirtualDisk> {
270        self.disks.get(&node)
271    }
272
273    /// A node's disk for fault configuration (created on demand).
274    pub fn disk_mut(&mut self, node: NodeId) -> &mut VirtualDisk {
275        self.disks.entry(node).or_default()
276    }
277
278    /// Registers a node process. The factory produces the task body; it
279    /// is called now and again on every restart, so anything captured by
280    /// the returned closure is volatile state that a crash discards.
281    /// Durable state belongs on the node's disk.
282    pub fn add_node(
283        &mut self,
284        node: NodeId,
285        name: &str,
286        mut factory: impl FnMut() -> TaskBody + 'static,
287    ) {
288        self.disks.entry(node).or_default();
289        let body = factory();
290        let task = self.runtime.spawn(node, name, body);
291        self.events.push(Event::TaskSpawned {
292            task,
293            node,
294            name: name.to_string(),
295        });
296        self.factories
297            .insert(node, (name.to_string(), Box::new(factory)));
298    }
299
300    /// Overrides one directed link's behavior.
301    pub fn set_link_config(&mut self, from: NodeId, to: NodeId, config: LinkConfig) {
302        self.network.set_link_config(from, to, config);
303    }
304
305    /// Installs a clock skew schedule for a node.
306    pub fn set_skew(&mut self, node: NodeId, schedule: SkewSchedule) {
307        self.runtime.clock_mut().set_skew(node, schedule);
308        self.events.push(Event::SkewUpdated { node });
309    }
310
311    /// Cuts connectivity between two node sets until [`Scenario::heal`].
312    pub fn partition(
313        &mut self,
314        group_a: impl IntoIterator<Item = NodeId>,
315        group_b: impl IntoIterator<Item = NodeId>,
316    ) {
317        let group_a: Vec<NodeId> = group_a.into_iter().collect();
318        let group_b: Vec<NodeId> = group_b.into_iter().collect();
319        self.network
320            .partition(group_a.iter().copied(), group_b.iter().copied());
321        self.events.push(Event::Partition { group_a, group_b });
322    }
323
324    /// Restores full connectivity.
325    pub fn heal(&mut self) {
326        self.network.heal();
327        self.events.push(Event::Healed);
328    }
329
330    /// Crashes a node: its tasks (volatile state) and inbox are dropped
331    /// and its disk loses every un-fsynced byte. In-flight messages to
332    /// the node are discarded on delivery.
333    pub fn crash_node(&mut self, node: NodeId) {
334        if !self.crashed.insert(node) {
335            return;
336        }
337        self.runtime.remove_tasks_of(node);
338        self.network.clear_inbox(node);
339        if let Some(disk) = self.disks.get_mut(&node) {
340            disk.crash();
341        }
342        self.events.push(Event::Crash { node });
343    }
344
345    /// Restarts a crashed node by re-invoking its process factory. Only
346    /// nodes registered with [`Scenario::add_node`] can restart.
347    pub fn restart_node(&mut self, node: NodeId) {
348        if !self.crashed.remove(&node) {
349            return;
350        }
351        self.disks.entry(node).or_default();
352        let (name, factory) = self
353            .factories
354            .get_mut(&node)
355            .expect("no process registered for node");
356        let name = name.clone();
357        let body = factory();
358        let task = self.runtime.spawn(node, &name, body);
359        self.events.push(Event::Restart { node });
360        self.events.push(Event::TaskSpawned { task, node, name });
361    }
362
363    /// Drives the schedule to completion with no mid-run chaos.
364    ///
365    /// On failure the seed and event log are persisted to
366    /// [`failure_dir`]; a task panic is re-raised after persisting.
367    pub fn run(&mut self, max_steps: u64) -> Result<RunOutcome, ScenarioError> {
368        self.run_with(max_steps, |_| {})
369    }
370
371    /// Drives the schedule to completion. `chaos` runs before every
372    /// scheduling decision with full mutable access to the scenario, so
373    /// tests can partition, heal, crash, and restart at chosen virtual
374    /// times or in reaction to recorded events.
375    pub fn run_with(
376        &mut self,
377        max_steps: u64,
378        mut chaos: impl FnMut(&mut Scenario),
379    ) -> Result<RunOutcome, ScenarioError> {
380        let outcome = catch_unwind(AssertUnwindSafe(|| self.run_inner(max_steps, &mut chaos)));
381        match outcome {
382            Ok(Ok(report)) => Ok(report),
383            Ok(Err(error)) => {
384                self.persist_failure(&error);
385                Err(error)
386            }
387            Err(payload) => {
388                self.persist_failure(&ScenarioError::TaskPanicked(panic_message(
389                    payload.as_ref(),
390                )));
391                resume_unwind(payload)
392            }
393        }
394    }
395
396    fn run_inner(
397        &mut self,
398        max_steps: u64,
399        chaos: &mut dyn FnMut(&mut Scenario),
400    ) -> Result<RunOutcome, ScenarioError> {
401        let mut steps = 0u64;
402        loop {
403            chaos(self);
404            self.runtime.wake_due();
405
406            if let Some(task) = self.runtime.pick_ready() {
407                steps += 1;
408                if steps > max_steps {
409                    return Err(ScenarioError::StepLimitExceeded { limit: max_steps });
410                }
411                let node = self.runtime.task_node(task);
412                let disk = self.disks.entry(node).or_default();
413                self.runtime
414                    .step(task, &mut self.network, disk, &mut self.events);
415                continue;
416            }
417
418            let next = self
419                .network
420                .next_delivery()
421                .into_iter()
422                .chain(self.runtime.next_wake())
423                .min();
424            match next {
425                Some(target) => {
426                    let from = self.now();
427                    self.runtime
428                        .clock_mut()
429                        .advance_to(target)
430                        .expect("event times are monotonic");
431                    self.events.push(Event::ClockAdvanced { from, to: target });
432                    self.deliver_due();
433                }
434                None => {
435                    if self.runtime.all_done() {
436                        return Ok(RunOutcome {
437                            seed: self.seed,
438                            steps,
439                            sim_time_micros: self.now(),
440                        });
441                    }
442                    return Err(ScenarioError::DeadlockDetected {
443                        at_micros: self.now(),
444                        waiting: self.runtime.unfinished_count(),
445                    });
446                }
447            }
448        }
449    }
450
451    fn deliver_due(&mut self) {
452        let now = self.now();
453        let deliveries = self.network.deliver_due(now, &self.crashed);
454        let mut woke = BTreeSet::new();
455        for delivery in deliveries {
456            match delivery.outcome {
457                DeliveryOutcome::Delivered => {
458                    self.events.push(Event::MessageDelivered {
459                        from: delivery.from,
460                        to: delivery.to,
461                        seq: delivery.seq,
462                    });
463                    woke.insert(delivery.to);
464                }
465                DeliveryOutcome::Discarded(reason) => {
466                    self.events.push(Event::MessageDropped {
467                        from: delivery.from,
468                        to: delivery.to,
469                        seq: delivery.seq,
470                        reason,
471                    });
472                }
473            }
474        }
475        for node in woke {
476            self.runtime.wake_receivers(node);
477        }
478    }
479
480    fn persist_failure(&self, error: &ScenarioError) {
481        let dir = failure_dir();
482        if let Err(io_error) = persist_failure_report(&dir, self.seed, error, &self.events) {
483            eprintln!(
484                "mongreldb-sim: could not persist failure artifact to {}: {io_error}",
485                dir.display()
486            );
487        }
488    }
489}
490
491fn panic_message(payload: &dyn Any) -> String {
492    if let Some(message) = payload.downcast_ref::<&str>() {
493        (*message).to_string()
494    } else if let Some(message) = payload.downcast_ref::<String>() {
495        message.clone()
496    } else {
497        "unknown panic payload".to_string()
498    }
499}
500
501/// Where failed-seed artifacts are written: `$MONGRELDB_SIM_FAILURES`,
502/// defaulting to `target/sim-failures/`.
503pub fn failure_dir() -> PathBuf {
504    env::var("MONGRELDB_SIM_FAILURES")
505        .map_or_else(|_| PathBuf::from("target/sim-failures"), PathBuf::from)
506}
507
508static FAILURE_COUNTER: AtomicU32 = AtomicU32::new(0);
509
510/// Writes a `{seed, error, events}` JSON artifact for a failed run and
511/// returns its path.
512pub fn persist_failure_report(
513    dir: &Path,
514    seed: Seed,
515    error: &ScenarioError,
516    events: &[Event],
517) -> io::Result<PathBuf> {
518    fs::create_dir_all(dir)?;
519    let ordinal = FAILURE_COUNTER.fetch_add(1, Ordering::Relaxed);
520    let path = dir.join(format!("sim-failure-{}-{ordinal}.json", seed.get()));
521    let report = serde_json::json!({
522        "seed": seed.get(),
523        "error": error.to_string(),
524        "event_count": events.len(),
525        "events": events,
526    });
527    fs::write(&path, serde_json::to_string_pretty(&report)?)?;
528    Ok(path)
529}
530
531#[cfg(test)]
532mod tests {
533    use super::*;
534    use crate::runtime::TaskState;
535
536    const A: NodeId = NodeId(1);
537
538    #[test]
539    fn deadlock_is_detected() {
540        let mut scenario = Scenario::new(Seed::new(1));
541        scenario.add_node(A, "waiter", || Box::new(|_| TaskState::WaitForMessage));
542        let error = scenario.run_inner(1_000, &mut |_| {}).unwrap_err();
543        assert!(matches!(
544            error,
545            ScenarioError::DeadlockDetected { waiting: 1, .. }
546        ));
547    }
548
549    #[test]
550    fn step_limit_guards_livelock() {
551        let mut scenario = Scenario::new(Seed::new(2));
552        scenario.add_node(A, "spinner", || Box::new(|_| TaskState::Yield));
553        let error = scenario.run_inner(100, &mut |_| {}).unwrap_err();
554        assert_eq!(error, ScenarioError::StepLimitExceeded { limit: 100 });
555    }
556
557    #[test]
558    fn happy_path_run_records_events() {
559        let mut scenario = Scenario::new(Seed::new(3));
560        scenario.add_node(A, "lonely", || {
561            Box::new(|ctx| {
562                ctx.log("hello");
563                TaskState::Done
564            })
565        });
566        let outcome = scenario.run(1_000).unwrap();
567        assert_eq!(outcome.seed, Seed::new(3));
568        assert_eq!(outcome.steps, 1);
569        assert!(scenario
570            .events()
571            .iter()
572            .any(|event| matches!(event, Event::Custom { message, .. } if message == "hello")));
573    }
574
575    #[test]
576    fn persist_failure_report_writes_seed_file() {
577        let dir =
578            env::temp_dir().join(format!("mongreldb-sim-persist-test-{}", std::process::id()));
579        let _ = fs::remove_dir_all(&dir);
580        let error = ScenarioError::StepLimitExceeded { limit: 5 };
581        let events = vec![Event::Healed];
582        let path = persist_failure_report(&dir, Seed::new(77), &error, &events).unwrap();
583
584        let written: serde_json::Value =
585            serde_json::from_str(&fs::read_to_string(&path).unwrap()).unwrap();
586        assert_eq!(written["seed"], 77);
587        assert!(written["error"].as_str().unwrap().contains("step limit"));
588        assert_eq!(written["events"], serde_json::json!(["Healed"]));
589        let _ = fs::remove_dir_all(&dir);
590    }
591}
mongreldb_sim/scenario.rs

mongreldb_sim/
scenario.rs
