Skip to main content

blue_lang_proc/
lib.rs

1//! Processes and supervision — OTP's structure over blue's scheduler.
2//!
3//! This is the layer that makes *let it crash* a strategy rather than a
4//! slogan, and it is only buildable because two things landed underneath it:
5//!
6//! - **bounded, catchable failure** — a runaway returns a typed
7//!   `VmError::BudgetExhausted` instead of aborting the OS process, so a
8//!   supervisor has something to catch;
9//! - **preemption** — `Vm::step` parks at a quantum with the frame stack
10//!   intact, so one process cannot starve its siblings.
11//!
12//! Both are the *same reduction counter*, read as a total ceiling and as a
13//! per-slice ceiling.
14//!
15//! ## The isolation this does and does not provide
16//!
17//! **Stated first, because overstating it would be the easy mistake.** Each
18//! process gets its **own `Vm`** — its own value stack, frame stack,
19//! handlers and local cells. That is real isolation of *control state*: one
20//! process's crash cannot corrupt another's stack, and a runaway is
21//! contained.
22//!
23//! **Each process also owns its own `Interpreter`**, so it owns its own
24//! globals. A definition or a mutation in one process is invisible to its
25//! siblings, and a restart discards them — which is what makes let-it-crash a
26//! guarantee rather than a hope: the restarted process cannot inherit the
27//! corrupted state that killed it.
28//!
29//! Messages are **deep-copied on send** ([`mailbox::deep_copy`]). `Value` is
30//! `Arc`-based, so a naive send would share structure between processes and
31//! quietly reintroduce the coupling the separate interpreters removed.
32//!
33//! The cost is stated rather than hidden: one interpreter per process means
34//! the primitives and the Lisp stdlib are installed per process, and
35//! [`Supervisor::interpreters_built`] counts it. That is the same trade the
36//! test framework makes for per-test isolation, and for the same reason —
37//! shared state makes order significant.
38//!
39//! This is `sumika` at the *environment* level. It is **not** a per-process
40//! GC heap: allocation still goes through Rust's allocator and reclamation is
41//! still `Arc` refcounting, so there is no independent per-process collection
42//! pause. `theory/BLUE.md` §V records that as the remaining distance.
43//!
44//! ## Strategy names follow `caixa`, deliberately
45//!
46//! `caixa-core`'s `Supervisor` kind already carries `estrategia` with
47//! `OneForOne` / `OneForAll` / `RestForOne` / `SimpleOneForOne`, plus
48//! max-restarts and a restart window. The org rule is one vocabulary per
49//! concept, so [`Strategy`] uses those names rather than inventing a second
50//! set.
51
52#![deny(unsafe_code)]
53
54use std::sync::Arc;
55
56use tatara_lisp_eval::vm::{Budget, Chunk, Progress, Vm, VmError};
57use tatara_lisp_eval::{ErrorObj, EvalError, Interpreter, Value};
58
59pub mod bisect;
60pub mod code;
61pub mod image;
62pub mod mailbox;
63pub mod node;
64pub mod pattern;
65pub mod receipt;
66pub mod scheduling;
67#[cfg(target_vendor = "apple")]
68mod seam;
69pub mod trace;
70pub mod violation;
71pub use bisect::{bisect, Bisected};
72pub use mailbox::{copy_data, deep_copy, install_process_primitives, System};
73pub use node::{
74    install, Carrier, ClockKind, Meter, Node, NodeConfig, CARRIED_WORDS, PROCESS_WORDS, ROOT,
75};
76pub use receipt::{
77    Frame, HostIo, Identities, Place, Places, Receipt, ReceiptSink, Source, Subject,
78};
79pub use scheduling::{IdleLane, Scheduling};
80pub use trace::{replay, Journal, Replayed, Trace};
81pub use violation::{Class, Violation};
82
83#[must_use]
84pub fn raised(kind: &str, message: String, data: Value, at: tatara_lisp::Span) -> EvalError {
85    EvalError::User {
86        value: Value::Error(Arc::new(ErrorObj {
87            tag: Arc::from(kind),
88            message: Arc::from(message),
89            data: vec![(Value::Keyword(Arc::from("data")), data)],
90        })),
91        at,
92    }
93}
94
95#[must_use]
96pub fn provided_name(word: &str) -> String {
97    format!("%provided/{word}")
98}
99
100/// A process identifier.
101#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
102pub struct Pid(pub u64);
103
104/// Why a process stopped.
105#[derive(Clone, Debug, PartialEq)]
106pub enum ExitReason {
107    /// Ran to completion.
108    Normal,
109    /// A typed runtime failure. Carries the rendered error.
110    Crashed(String),
111    /// The execution budget ran out — a runaway. **This is the case that
112    /// was previously an OS-process abort**, and it is the reason
113    /// supervision is possible at all.
114    Runaway(String),
115    /// Ended from outside: an exit signal from a linked process, `exit(pid,
116    /// reason)`, or a supervisor shutting its children down. Carries the
117    /// rendered reason.
118    Killed(String),
119    /// A fault the runtime declares unrecoverable, such as a capability
120    /// breach. Links and monitors see it; a supervisor never restarts it.
121    Fatal(String),
122}
123
124/// The kind of the refusal an evaluation or a process meets when it holds more
125/// live bytes than its bound (`max_alloc_bytes`).
126pub const MEMORY_EXHAUSTED: &str = "memory-exhausted";
127
128/// What a memory refusal says: the bound it passed.
129#[must_use]
130pub fn memory_budget(bound: u64) -> String {
131    format!("live-byte budget of {bound} exceeded")
132}
133
134/// A memory refusal as one line: its kind, then [`memory_budget`].
135#[must_use]
136pub fn memory_refusal(bound: u64) -> String {
137    format!("{MEMORY_EXHAUSTED}: {}", memory_budget(bound))
138}
139
140#[must_use]
141pub fn dispatch_refusal(limit: usize) -> String {
142    format!("execution budget exhausted: steps per dispatch (limit {limit})")
143}
144
145impl ExitReason {
146    /// A process that held more than `bound` live bytes: a runaway, which its
147    /// links, monitors and supervisor see, and a supervisor restarts.
148    #[must_use]
149    pub fn memory_exhausted(bound: u64) -> Self {
150        Self::Runaway(memory_refusal(bound))
151    }
152
153    #[must_use]
154    pub fn dispatch_exhausted(limit: usize) -> Self {
155        Self::Runaway(dispatch_refusal(limit))
156    }
157
158    pub fn is_abnormal(&self) -> bool {
159        !matches!(self, ExitReason::Normal)
160    }
161
162    pub fn is_fatal(&self) -> bool {
163        matches!(self, ExitReason::Fatal(_))
164    }
165
166    /// Does a supervisor restart a child that exited this way?
167    pub fn is_restartable(&self) -> bool {
168        self.is_abnormal() && !self.is_fatal()
169    }
170}
171
172/// No `PartialEq`: `tatara_lisp_eval::Value` does not implement it, and
173/// wrapping it in a comparable shim here would invent an equality the
174/// runtime does not define. Callers ask the questions they actually mean —
175/// [`ProcState::is_runnable`], [`ProcState::done_int`].
176#[derive(Clone, Debug)]
177pub enum ProcState {
178    Runnable,
179    /// Parked inside `receive` with an empty mailbox. Becomes `Runnable`
180    /// again the moment a message arrives — a *free* wait, costing no fuel.
181    Blocked,
182    Done(Value),
183    Exited(ExitReason),
184}
185
186impl ProcState {
187    pub fn is_runnable(&self) -> bool {
188        matches!(self, ProcState::Runnable)
189    }
190
191    pub fn is_blocked(&self) -> bool {
192        matches!(self, ProcState::Blocked)
193    }
194
195    pub fn is_done(&self) -> bool {
196        matches!(self, ProcState::Done(_))
197    }
198
199    /// The integer this process produced, if it finished with one.
200    pub fn done_int(&self) -> Option<i64> {
201        match self {
202            ProcState::Done(Value::Int(v)) => Some(*v),
203            _ => None,
204        }
205    }
206
207    pub fn exit_reason(&self) -> Option<&ExitReason> {
208        match self {
209            ProcState::Exited(r) => Some(r),
210            _ => None,
211        }
212    }
213}
214
215/// One process: a program, its own VM, and **its own interpreter**.
216pub struct Proc<H> {
217    pub pid: Pid,
218    /// The code. Retained so a supervisor can restart from a clean VM.
219    chunk: Arc<Chunk>,
220    vm: Vm,
221    /// This process's own globals. The reason a sibling cannot see its
222    /// definitions, and the reason a restart discards them.
223    interp: Interpreter<H>,
224    /// Rebuilds the interpreter on restart. A closure rather than a stored
225    /// clone: `Interpreter` holds registered closures, and a restart must
226    /// produce a *fresh* environment rather than a copy of a possibly-corrupt
227    /// one.
228    build: Arc<dyn Fn() -> Interpreter<H> + Send + Sync>,
229    budget: Budget,
230    pub state: ProcState,
231    /// How many times this process has been restarted.
232    pub restarts: usize,
233}
234
235impl<H> Proc<H> {
236    pub(crate) fn new(
237        pid: Pid,
238        chunk: Arc<Chunk>,
239        budget: Budget,
240        build: Arc<dyn Fn() -> Interpreter<H> + Send + Sync>,
241    ) -> Self {
242        let interp = build();
243        Self {
244            pid,
245            chunk,
246            vm: Vm::with_budget(budget),
247            interp,
248            build,
249            budget,
250            state: ProcState::Runnable,
251            restarts: 0,
252        }
253    }
254
255    /// Discard all control state and start over from the same code.
256    ///
257    /// A fresh `Vm` is the whole restart: there is nothing to unwind,
258    /// because the crashed process's stack was never shared.
259    pub(crate) fn restart(&mut self) {
260        self.vm = Vm::with_budget(self.budget);
261        // A FRESH interpreter, not a saved copy. The whole point of
262        // let-it-crash is that the restarted process cannot inherit the state
263        // that killed it.
264        self.interp = (self.build)();
265        self.state = ProcState::Runnable;
266        self.restarts += 1;
267    }
268
269    fn is_runnable(&self) -> bool {
270        self.state.is_runnable()
271    }
272
273    /// Give this process one quantum and record what it did.
274    ///
275    /// The one place a VM step becomes a process state, so the Rust-level
276    /// [`Supervisor`] and the blue-level [`Node`] cannot disagree about what a
277    /// slice means.
278    pub(crate) fn slice(&mut self, host: &mut H) -> Slice
279    where
280        H: 'static,
281    {
282        let chunk = self.chunk.clone();
283        match self.vm.step(chunk, &mut self.interp, host) {
284            Ok(Progress::Yielded) => Slice::Yielded,
285            Ok(Progress::Blocked) => {
286                self.state = ProcState::Blocked;
287                Slice::Blocked
288            }
289            Ok(Progress::Done(v)) => {
290                self.state = ProcState::Done(v.clone());
291                Slice::Done(v)
292            }
293            Err(e) => {
294                let reason = classify(&e);
295                self.state = ProcState::Exited(reason.clone());
296                Slice::Exited(reason)
297            }
298        }
299    }
300}
301
302/// What one scheduling slice did.
303pub(crate) enum Slice {
304    Yielded,
305    Blocked,
306    Done(Value),
307    Exited(ExitReason),
308}
309
310/// What the scheduler needs to know about the host, and nothing more.
311///
312/// The supervisor schedules; it does not know what a mailbox is. It only
313/// asks *"can this parked process proceed now?"* and *"is there anything
314/// pending that could make one proceed?"* — which is exactly enough to wake
315/// a blocked process and to tell a live wait from a deadlock.
316///
317/// Keeping it a trait rather than hard-coding [`System`] means a host with a
318/// different readiness source — a timer, a socket, a market feed — plugs in
319/// without touching the scheduler.
320pub trait Readiness {
321    /// May this parked process now make progress?
322    fn is_ready(&self, pid: Pid) -> bool;
323
324    /// Is there any pending work at all? When every process is parked and
325    /// this is false, nothing will ever change: that is a deadlock, and
326    /// spinning on it forever is the failure mode this question exists to
327    /// prevent.
328    fn any_pending(&self) -> bool;
329
330    /// The scheduler is about to give `pid` a slice. A host whose primitives
331    /// are process-relative — `receive`, `self` — needs this to answer them.
332    /// Default: nothing.
333    fn set_current(&mut self, _pid: Pid) {}
334
335    /// `pid` is about to restart, so its per-process host state should be
336    /// discarded. Default: nothing.
337    fn on_restart(&mut self, _pid: Pid) {}
338}
339
340/// A host with no readiness source of its own. Nothing ever becomes ready,
341/// so a process that parks under `()` is immediately a deadlock — which is
342/// the honest answer, since there is nothing to deliver.
343impl Readiness for () {
344    fn is_ready(&self, _pid: Pid) -> bool {
345        false
346    }
347    fn any_pending(&self) -> bool {
348        false
349    }
350}
351
352/// Restart strategy. Names follow `caixa-core`'s `estrategia`.
353#[derive(Clone, Copy, Debug, PartialEq, Eq)]
354pub enum Strategy {
355    /// Restart only the process that exited.
356    OneForOne,
357    /// Restart every child.
358    OneForAll,
359    /// Restart the exited child and every child declared after it.
360    RestForOne,
361}
362
363impl Strategy {
364    /// The positions a strategy restarts when the child at `exited` of `len`
365    /// exits abnormally. Shared by [`Supervisor`] and the blue-level
366    /// supervisors in [`node`], so the three strategies mean one thing.
367    #[must_use]
368    pub fn targets(self, exited: usize, len: usize) -> std::ops::Range<usize> {
369        match self {
370            Strategy::OneForOne => exited..exited + 1,
371            Strategy::OneForAll => 0..len,
372            Strategy::RestForOne => exited..len,
373        }
374    }
375
376    /// The blue spelling: `:one_for_one`, `:one_for_all`, `:rest_for_one`.
377    #[must_use]
378    pub fn from_keyword(word: &str) -> Option<Self> {
379        match word {
380            "one_for_one" => Some(Strategy::OneForOne),
381            "one_for_all" => Some(Strategy::OneForAll),
382            "rest_for_one" => Some(Strategy::RestForOne),
383            _ => None,
384        }
385    }
386}
387
388/// A record of what the supervisor did, so a caller can assert on
389/// behaviour rather than infer it from final state.
390#[derive(Clone, Debug, PartialEq)]
391pub enum Event {
392    Exited {
393        pid: Pid,
394        reason: ExitReason,
395    },
396    Restarted {
397        pid: Pid,
398    },
399    /// Restart intensity exceeded: the supervisor gave up rather than
400    /// restarting in a tight loop forever.
401    GaveUp {
402        pid: Pid,
403        restarts: usize,
404    },
405    /// A process parked waiting for a message.
406    Blocked {
407        pid: Pid,
408    },
409    /// A parked process was woken by an arriving message.
410    Woke {
411        pid: Pid,
412    },
413    /// Every remaining process is parked and nothing is pending, so no
414    /// message can ever arrive. Reported rather than spun on, because a
415    /// deadlocked switch that *looks* busy is far worse than one that says
416    /// which processes are stuck.
417    Deadlocked {
418        blocked: Vec<Pid>,
419    },
420}
421
422/// A supervisor over an ordered set of children.
423/// A [`Supervisor::spawn`] factory that forks one prototype instead of
424/// rebuilding the standard library per process and per restart.
425///
426/// The isolation guarantee is unchanged, and that is why this is a fork rather
427/// than a shared interpreter. Each child gets a private writable frame, so its
428/// `define`s are invisible to every sibling; the inherited frames sit below its
429/// write floor, so a `set!` reaching one raises `SetSealed` rather than
430/// mutating state a sibling can observe; and a restart is a *new* fork, so the
431/// incarnation that died leaves nothing behind.
432///
433/// ```ignore
434/// let prototype = Arc::new(blue_lang_runtime::interpreter(&mut System::new()));
435/// let pid = sup.spawn(chunk, budget, forking(&prototype));
436/// ```
437///
438/// Build the prototype, install everything on it, *then* fork. A `define` the
439/// prototype runs after a child forks is visible to that child — the frames are
440/// shared, which is the whole point.
441#[must_use]
442pub fn forking<H: 'static>(
443    prototype: &Arc<Interpreter<H>>,
444) -> impl Fn() -> Interpreter<H> + Send + Sync + 'static
445where
446    Interpreter<H>: Send + Sync,
447{
448    let prototype = Arc::clone(prototype);
449    move || prototype.fork()
450}
451
452pub struct Supervisor<H> {
453    procs: Vec<Proc<H>>,
454    strategy: Strategy,
455    /// Maximum restarts of any single child before the supervisor gives up.
456    ///
457    /// Without this a crash-on-start child restarts forever, which is a
458    /// busy loop wearing a supervision tree's clothes.
459    max_restarts: usize,
460    next_pid: u64,
461    pub events: Vec<Event>,
462    /// Interpreters constructed — one per spawn plus one per restart. Reported
463    /// so the cost of per-process isolation is visible rather than hidden.
464    pub interpreters_built: usize,
465}
466
467impl<H: 'static> Supervisor<H> {
468    pub fn new(strategy: Strategy, max_restarts: usize) -> Self {
469        Self {
470            procs: Vec::new(),
471            strategy,
472            max_restarts,
473            next_pid: 1,
474            events: Vec::new(),
475            interpreters_built: 0,
476        }
477    }
478
479    /// Add a child. Order matters for [`Strategy::RestForOne`].
480    ///
481    /// `build` constructs this process's interpreter, and is called again on
482    /// every restart. It is a factory rather than a value because a restart
483    /// must produce a *fresh* environment: handing over one pre-built
484    /// interpreter would mean a crashed process restarts into the globals that
485    /// killed it.
486    ///
487    /// **Prefer [`forking`] to a closure that builds from scratch.** A factory
488    /// calling `interpreter_hostless()` evaluates the whole Lisp standard
489    /// library on every spawn *and every restart* — ~945µs each, for an
490    /// identical result — and a supervision tree under a crash loop pays it
491    /// repeatedly. `forking` gives the same guarantee (a restart gets a clean
492    /// private frame, and cannot write the prototype's globals) for ~29µs.
493    #[allow(clippy::doc_markdown)]
494    pub fn spawn<F>(&mut self, chunk: Arc<Chunk>, budget: Budget, build: F) -> Pid
495    where
496        F: Fn() -> Interpreter<H> + Send + Sync + 'static,
497    {
498        let pid = Pid(self.next_pid);
499        self.next_pid += 1;
500        self.procs
501            .push(Proc::new(pid, chunk, budget, Arc::new(build)));
502        self.interpreters_built += 1;
503        pid
504    }
505
506    pub fn state_of(&self, pid: Pid) -> Option<&ProcState> {
507        self.procs.iter().find(|p| p.pid == pid).map(|p| &p.state)
508    }
509
510    /// Does this process's own environment bind `name`?
511    ///
512    /// Exposed because "a restart discards the environment" is otherwise
513    /// untestable: a counter proves a rebuild was *requested*, not that the old
514    /// bindings are gone. Also useful when debugging a supervision tree.
515    pub fn proc_binds(&self, pid: Pid, name: &str) -> Option<bool> {
516        self.procs
517            .iter()
518            .find(|p| p.pid == pid)
519            .map(|p| p.interp.lookup_global(name).is_some())
520    }
521
522    pub fn restarts_of(&self, pid: Pid) -> Option<usize> {
523        self.procs.iter().find(|p| p.pid == pid).map(|p| p.restarts)
524    }
525
526    /// Is anything still runnable?
527    /// Is there anything to run *right now*?
528    ///
529    /// A blocked process counts only if the host says it is ready. A blocked
530    /// process with nothing pending is not work — it is a deadlock, and
531    /// [`Supervisor::round`] reports it.
532    pub fn has_work(&self, host: &H) -> bool
533    where
534        H: Readiness,
535    {
536        self.procs
537            .iter()
538            .any(|p| p.is_runnable() || (p.state.is_blocked() && host.is_ready(p.pid)))
539    }
540
541    /// Processes parked with nothing to wake them.
542    pub fn deadlocked(&self, host: &H) -> Vec<Pid>
543    where
544        H: Readiness,
545    {
546        if host.any_pending() {
547            return Vec::new();
548        }
549        self.procs
550            .iter()
551            .filter(|p| p.state.is_blocked())
552            .map(|p| p.pid)
553            .collect()
554    }
555
556    /// Run one scheduling round: give every runnable process one quantum,
557    /// then apply the restart strategy to anything that exited abnormally.
558    ///
559    /// Round-robin, so a long-running child cannot starve a short one —
560    /// that is preemption's contribution, and it holds without any child
561    /// cooperating.
562    pub fn round(&mut self, host: &mut H)
563    where
564        H: Readiness,
565    {
566        let mut exited: Vec<(usize, ExitReason)> = Vec::new();
567
568        // Wake first, so a message that arrived during the previous round is
569        // acted on in this one rather than a round later.
570        for idx in 0..self.procs.len() {
571            if self.procs[idx].state.is_blocked() && host.is_ready(self.procs[idx].pid) {
572                self.procs[idx].state = ProcState::Runnable;
573                let pid = self.procs[idx].pid;
574                self.events.push(Event::Woke { pid });
575            }
576        }
577
578        for idx in 0..self.procs.len() {
579            if !self.procs[idx].is_runnable() {
580                continue;
581            }
582            // Tell the host whose slice this is, so `receive` and `self`
583            // resolve to the right process.
584            host.set_current(self.procs[idx].pid);
585            let pid = self.procs[idx].pid;
586            match self.procs[idx].slice(host) {
587                Slice::Yielded => {}
588                Slice::Blocked => self.events.push(Event::Blocked { pid }),
589                Slice::Done(_) => self.events.push(Event::Exited {
590                    pid,
591                    reason: ExitReason::Normal,
592                }),
593                Slice::Exited(reason) => {
594                    self.events.push(Event::Exited {
595                        pid,
596                        reason: reason.clone(),
597                    });
598                    exited.push((idx, reason));
599                }
600            }
601        }
602
603        for (idx, reason) in exited {
604            if reason.is_restartable() {
605                self.apply_strategy(idx, host);
606            }
607        }
608    }
609
610    /// Drive rounds until nothing is runnable, or `max_rounds` elapse.
611    ///
612    /// The bound is a test-and-operator safety net, not a semantic: a
613    /// supervision tree that cannot settle should say so rather than spin.
614    pub fn run_to_quiescence(&mut self, host: &mut H, max_rounds: usize) -> usize
615    where
616        H: Readiness,
617    {
618        let mut rounds = 0;
619        while self.has_work(host) && rounds < max_rounds {
620            self.round(host);
621            rounds += 1;
622        }
623        // Settling with processes still parked is a deadlock, not quiescence.
624        // Saying so once, here, is what keeps a stuck system from reading as
625        // a finished one.
626        let stuck = self.deadlocked(host);
627        if !stuck.is_empty() {
628            self.events.push(Event::Deadlocked { blocked: stuck });
629        }
630        rounds
631    }
632
633    fn apply_strategy(&mut self, exited_idx: usize, host: &mut H)
634    where
635        H: Readiness,
636    {
637        let targets = self.strategy.targets(exited_idx, self.procs.len());
638
639        // Intensity is checked on the CHILD THAT EXITED, not on the
640        // siblings a strategy sweeps up: a sibling restarted as collateral
641        // has not itself misbehaved.
642        if self.procs[exited_idx].restarts >= self.max_restarts {
643            let pid = self.procs[exited_idx].pid;
644            let restarts = self.procs[exited_idx].restarts;
645            self.events.push(Event::GaveUp { pid, restarts });
646            return;
647        }
648
649        for idx in targets {
650            if self.procs[idx]
651                .state
652                .exit_reason()
653                .is_some_and(ExitReason::is_fatal)
654            {
655                continue;
656            }
657            self.procs[idx].restart();
658            self.interpreters_built += 1;
659            let pid = self.procs[idx].pid;
660            // A restarted process starts clean on the host side too. In
661            // particular its mailbox is emptied: the messages in flight were
662            // addressed to the incarnation that died, and replaying them into
663            // the fresh one is how a poison message kills a process forever.
664            host.on_restart(pid);
665            self.events.push(Event::Restarted { pid });
666        }
667    }
668}
669
670pub(crate) fn classify(e: &VmError) -> ExitReason {
671    match e {
672        VmError::BudgetExhausted { .. } => ExitReason::Runaway(e.to_string()),
673        VmError::Eval(fatal @ tatara_lisp_eval::EvalError::Fatal { .. }) => {
674            ExitReason::Fatal(fatal.short_message())
675        }
676        other => ExitReason::Crashed(other.to_string()),
677    }
678}
679
680#[cfg(test)]
681mod tests {
682    use super::*;
683    use tatara_lisp_eval::vm::compile_program;
684
685    fn compile(src: &str) -> Arc<Chunk> {
686        let forms = tatara_lisp::read_spanned(src).expect("parse");
687        Arc::new(compile_program(&forms).expect("compile"))
688    }
689
690    fn blue(src: &str) -> Arc<Chunk> {
691        let forms = blue_lang_syntax::parse_program(src).expect("parse blue");
692        let text = forms
693            .iter()
694            .map(|f| f.to_string())
695            .collect::<Vec<_>>()
696            .join("\n");
697        compile(&text)
698    }
699
700    /// Each process builds its own. Passed as a factory, so a restart gets a
701    /// fresh one.
702    fn build() -> Interpreter<()> {
703        blue_lang_runtime::interpreter_hostless()
704    }
705
706    fn preemptive(q: usize) -> Budget {
707        Budget::preemptive(q)
708    }
709
710    // ---- the payoff: a runaway is contained and restarted -------------
711
712    /// **This is what was impossible before.** A runaway previously aborted
713    /// the OS process, so there was nothing to supervise. Now it exits with
714    /// a typed reason and the supervisor restarts it.
715    #[test]
716    fn a_runaway_is_contained_and_restarted() {
717        let mut sup = Supervisor::new(Strategy::OneForOne, 3);
718        let spin = compile("(define (spin n) (spin (+ n 1))) (spin 0)");
719        let pid = sup.spawn(
720            spin,
721            Budget {
722                fuel: Some(2_000),
723                max_depth: None,
724                quantum: Some(200),
725            },
726            build,
727        );
728        sup.run_to_quiescence(&mut (), 200);
729
730        assert!(
731            sup.events.iter().any(|e| matches!(
732                e,
733                Event::Exited {
734                    reason: ExitReason::Runaway(_),
735                    ..
736                }
737            )),
738            "the runaway must exit with a Runaway reason: {:?}",
739            sup.events
740        );
741        assert!(
742            sup.events
743                .iter()
744                .any(|e| matches!(e, Event::Restarted { .. })),
745            "the supervisor must have restarted it"
746        );
747        assert!(sup.restarts_of(pid).unwrap() > 0);
748    }
749
750    /// And it must eventually GIVE UP rather than restart forever — a
751    /// crash-on-start child otherwise becomes a busy loop in a supervision
752    /// tree's clothes.
753    #[test]
754    fn restart_intensity_stops_an_unfixable_child() {
755        let mut sup = Supervisor::new(Strategy::OneForOne, 2);
756        let spin = compile("(define (spin n) (spin (+ n 1))) (spin 0)");
757        sup.spawn(
758            spin,
759            Budget {
760                fuel: Some(500),
761                max_depth: None,
762                quantum: Some(100),
763            },
764            build,
765        );
766        sup.run_to_quiescence(&mut (), 500);
767
768        assert!(
769            sup.events.iter().any(|e| matches!(e, Event::GaveUp { .. })),
770            "the supervisor must give up: {:?}",
771            sup.events
772        );
773        assert!(!sup.has_work(&()), "nothing should still be runnable");
774    }
775
776    // ---- normal completion -------------------------------------------
777
778    #[test]
779    fn a_healthy_process_finishes_and_is_not_restarted() {
780        let mut sup = Supervisor::new(Strategy::OneForOne, 3);
781        let pid = sup.spawn(blue("1 + 2"), preemptive(50), build);
782        sup.run_to_quiescence(&mut (), 100);
783
784        assert_eq!(sup.state_of(pid).and_then(ProcState::done_int), Some(3));
785        assert_eq!(
786            sup.restarts_of(pid),
787            Some(0),
788            "a healthy child must not restart"
789        );
790    }
791
792    /// Round-robin fairness at the supervisor level: a long child must not
793    /// prevent a short one from finishing.
794    #[test]
795    fn a_long_child_does_not_starve_a_short_one() {
796        let mut sup = Supervisor::new(Strategy::OneForOne, 3);
797        let long = sup.spawn(
798            blue("def loop(n)\n  if n == 0\n    1\n  else\n    loop(n - 1)\n  end\nend\nloop(400)"),
799            preemptive(20),
800            build,
801        );
802        let short = sup.spawn(blue("2 + 3"), preemptive(20), build);
803
804        // One round is enough for the SHORT child; the long one is still going.
805        sup.round(&mut ());
806        assert_eq!(sup.state_of(short).and_then(ProcState::done_int), Some(5));
807        assert!(
808            sup.state_of(long).is_some_and(ProcState::is_runnable),
809            "the long child should still be running"
810        );
811
812        sup.run_to_quiescence(&mut (), 1000);
813        assert_eq!(sup.state_of(long).and_then(ProcState::done_int), Some(1));
814    }
815
816    // ---- the three strategies ------------------------------------------
817
818    fn crashing() -> Arc<Chunk> {
819        compile("(define (spin n) (spin (+ n 1))) (spin 0)")
820    }
821
822    /// Fuel BELOW the quantum, so the crasher exhausts *inside* its first
823    /// slice. With fuel above the quantum it merely yields, and a
824    /// single-round strategy test would see no exit at all.
825    fn tiny_budget() -> Budget {
826        Budget {
827            fuel: Some(60),
828            max_depth: None,
829            quantum: Some(1_000),
830        }
831    }
832
833    #[test]
834    fn one_for_one_restarts_only_the_child_that_exited() {
835        let mut sup = Supervisor::new(Strategy::OneForOne, 5);
836        let a = sup.spawn(blue("1"), preemptive(50), build);
837        let bad = sup.spawn(crashing(), tiny_budget(), build);
838        let c = sup.spawn(blue("2"), preemptive(50), build);
839        sup.round(&mut ());
840
841        assert_eq!(sup.restarts_of(bad), Some(1), "the crasher must restart");
842        assert_eq!(sup.restarts_of(a), Some(0), "a sibling must NOT restart");
843        assert_eq!(sup.restarts_of(c), Some(0), "a sibling must NOT restart");
844    }
845
846    #[test]
847    fn one_for_all_restarts_every_child() {
848        let mut sup = Supervisor::new(Strategy::OneForAll, 5);
849        let a = sup.spawn(crashing(), tiny_budget(), build);
850        let b = sup.spawn(blue("1"), preemptive(50), build);
851        let c = sup.spawn(blue("2"), preemptive(50), build);
852        sup.round(&mut ());
853
854        assert_eq!(sup.restarts_of(a), Some(1));
855        assert_eq!(
856            sup.restarts_of(b),
857            Some(1),
858            "one_for_all must restart siblings"
859        );
860        assert_eq!(
861            sup.restarts_of(c),
862            Some(1),
863            "one_for_all must restart siblings"
864        );
865    }
866
867    fn breaching_build() -> Interpreter<()> {
868        let mut interp = build();
869        interp.register_fn(
870            "breach",
871            tatara_lisp_eval::ffi::Arity::Exact(0),
872            |_: &[Value], _: &mut (), at| {
873                Err(tatara_lisp_eval::EvalError::fatal(
874                    "capability-breach",
875                    "`read_file` reaches the host (filesystem)",
876                    at,
877                ))
878            },
879        );
880        interp
881    }
882
883    fn breaching() -> Arc<Chunk> {
884        compile("(try (breach) (catch (e) 0))")
885    }
886
887    #[test]
888    fn a_fatal_child_is_never_restarted_and_no_sibling_restarts_for_it() {
889        let mut sup = Supervisor::new(Strategy::OneForAll, 5);
890        let bad = sup.spawn(breaching(), preemptive(50), breaching_build);
891        let ok = sup.spawn(blue("1 + 2"), preemptive(50), build);
892        sup.run_to_quiescence(&mut (), 100);
893
894        assert!(
895            sup.events.iter().any(|e| matches!(
896                e,
897                Event::Exited { pid, reason: ExitReason::Fatal(m) } if *pid == bad && m.contains("capability-breach")
898            )),
899            "{:?}",
900            sup.events
901        );
902        assert_eq!(
903            sup.restarts_of(bad),
904            Some(0),
905            "a fatal child must not restart"
906        );
907        assert_eq!(
908            sup.restarts_of(ok),
909            Some(0),
910            "a sibling must not restart for it"
911        );
912        assert_eq!(sup.state_of(ok).and_then(ProcState::done_int), Some(3));
913    }
914
915    #[test]
916    fn a_fatal_child_is_not_swept_up_when_a_sibling_restarts() {
917        let mut sup = Supervisor::new(Strategy::OneForAll, 5);
918        let bad = sup.spawn(breaching(), preemptive(50), breaching_build);
919        sup.round(&mut ());
920        let crasher = sup.spawn(crashing(), tiny_budget(), build);
921        sup.round(&mut ());
922
923        assert_eq!(sup.restarts_of(crasher), Some(1));
924        assert_eq!(
925            sup.restarts_of(bad),
926            Some(0),
927            "one_for_all must leave a fatal child down"
928        );
929        assert!(sup
930            .state_of(bad)
931            .and_then(ProcState::exit_reason)
932            .is_some_and(ExitReason::is_fatal));
933    }
934
935    #[test]
936    fn only_a_fatal_exit_is_abnormal_and_not_restartable() {
937        let s = String::new;
938        for (r, restartable) in [
939            (ExitReason::Normal, false),
940            (ExitReason::Crashed(s()), true),
941            (ExitReason::Runaway(s()), true),
942            (ExitReason::Killed(s()), true),
943            (ExitReason::Fatal(s()), false),
944        ] {
945            assert_eq!(r.is_restartable(), restartable, "{r:?}");
946            assert_eq!(r.is_fatal(), matches!(r, ExitReason::Fatal(_)), "{r:?}");
947        }
948        assert!(ExitReason::Fatal(s()).is_abnormal());
949    }
950
951    /// `rest_for_one` restarts the exited child and everything declared
952    /// AFTER it — the ones that may depend on it — and leaves the earlier
953    /// ones alone.
954    #[test]
955    fn rest_for_one_restarts_the_child_and_its_successors_only() {
956        let mut sup = Supervisor::new(Strategy::RestForOne, 5);
957        let before = sup.spawn(blue("1"), preemptive(50), build);
958        let bad = sup.spawn(crashing(), tiny_budget(), build);
959        let after = sup.spawn(blue("2"), preemptive(50), build);
960        sup.round(&mut ());
961
962        assert_eq!(
963            sup.restarts_of(before),
964            Some(0),
965            "a predecessor must not restart"
966        );
967        assert_eq!(sup.restarts_of(bad), Some(1));
968        assert_eq!(sup.restarts_of(after), Some(1), "a successor must restart");
969    }
970
971    // ---- anti-vacuity --------------------------------------------------
972
973    /// The supervisor must be able to observe a normal exit as normal. If
974    /// everything looked abnormal, the restart tests would pass by always
975    /// restarting.
976    #[test]
977    fn a_normal_exit_is_not_treated_as_a_crash() {
978        let mut sup = Supervisor::new(Strategy::OneForAll, 5);
979        let a = sup.spawn(blue("1"), preemptive(50), build);
980        let b = sup.spawn(blue("2"), preemptive(50), build);
981        sup.run_to_quiescence(&mut (), 100);
982
983        assert!(
984            !sup.events
985                .iter()
986                .any(|e| matches!(e, Event::Restarted { .. })),
987            "a normal exit must not trigger one_for_all: {:?}",
988            sup.events
989        );
990        assert_eq!(sup.restarts_of(a), Some(0));
991        assert_eq!(sup.restarts_of(b), Some(0));
992    }
993
994    /// One process's crash must not corrupt another's control state — the
995    /// isolation claim, tested directly.
996    #[test]
997    fn a_crash_does_not_disturb_a_siblings_result() {
998        let mut sup = Supervisor::new(Strategy::OneForOne, 1);
999        let _bad = sup.spawn(crashing(), tiny_budget(), build);
1000        let good = sup.spawn(blue("def sum(n, acc)\n  if n == 0\n    acc\n  else\n    sum(n - 1, acc + n)\n  end\nend\nsum(10, 0)"), preemptive(15), build);
1001        sup.run_to_quiescence(&mut (), 500);
1002
1003        assert_eq!(
1004            sup.state_of(good).and_then(ProcState::done_int),
1005            Some(55),
1006            "the healthy child's answer must survive a sibling's crash"
1007        );
1008    }
1009
1010    /// And blue source really does run under supervision — not just
1011    /// hand-written tatara-lisp.
1012    #[test]
1013    fn blue_source_runs_under_supervision() {
1014        let mut sup = Supervisor::new(Strategy::OneForOne, 3);
1015        let pid = sup.spawn(
1016            blue(
1017                "def fact(n)\n  if n < 2\n    1\n  else\n    n * fact(n - 1)\n  end\nend\nfact(6)",
1018            ),
1019            preemptive(30),
1020            build,
1021        );
1022        sup.run_to_quiescence(&mut (), 1000);
1023        assert_eq!(sup.state_of(pid).and_then(ProcState::done_int), Some(720));
1024    }
1025}