/build/source/nativelink-sim/src/checker.rs

Source
// Copyright 2026 Trace Machina, Inc. All rights reserved.
//
// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    See LICENSE file for details
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! The invariants, checked over a finished trace. Each is a pure function
//! of the records and the facts of the run, so a test can hand one a trace
//! it altered and see it fire.

use core::fmt;
use std::collections::{BTreeSet, HashMap, HashSet};

use crate::trace::{Event, Record, Stage};

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum Invariant {
    /// No result lost: a result a worker reported reaches every client still
    /// attached to the action.
    I1,
    /// At most one dispatch of an action at a time: a second needs the first
    /// declined, finished, killed or its worker gone in between.
    I2,
    /// Every dispatch is answered (accepted or declined, when the worker
    /// acknowledges), and an accepted one ends in exactly one result, a kill
    /// or the worker's loss.
    I3,
    /// What the scheduler has executing equals what the workers run, at
    /// quiescence.
    I4,
    /// Order: a dispatch never passes over a waiting action of higher
    /// priority, or of equal priority submitted earlier.
    I5,
    /// Liveness bounds: abandoned actions stop being dispatched, and a
    /// silent worker is evicted, within their bounds.
    I6,
    /// Retry budgets: an action is executed no more often than its budgets
    /// allow.
    I7,
    /// No stall: the final settle reaches quiescence inside its budget.
    I8,
    /// A worker never runs more than its `max_inflight_tasks`.
    I9,
    /// A client sees stages in order; the only back-edge is Executing to
    /// Queued.
    I10,
    /// Store commands between two matching passes stay inside the declared
    /// bound.
    I11,
}

impl fmt::Display for Invariant {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Debug::fmt(self, f)
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Violation {
    pub invariant: Invariant,
    pub at_ms: u64,
    pub detail: String,
}

impl fmt::Display for Violation {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "{} at {} ms: {}",
            self.invariant, self.at_ms, self.detail
        )
    }
}

/// What the checker needs to know about a run beyond its trace.
#[derive(Clone, Debug, Default)]
pub struct Facts {
    /// `max_inflight_tasks` by worker name.
    pub max_inflight: HashMap<String, u64>,
    /// Whether each action label is cacheable.
    pub cacheable: HashMap<String, bool>,
    /// The most executions one action may get; 0 does not check.
    pub dispatch_budget: u64,
    /// After every client of an action left, how long it may still be
    /// dispatched.
    pub abandon_bound_ms: u64,
    /// How long a silent worker may stay before it is evicted.
    pub evict_bound_ms: u64,
    /// Silences: worker, start and end.
    pub silences: Vec<(String, u64, u64)>,
    /// Check I5.
    pub ordered: bool,
    /// The bound I11 checks, and what the run measured.
    pub command_bound: Option<u64>,
    pub max_commands_per_pass: u64,
}

/// Every violation in `records`, in invariant order.
pub fn check(records: &[Record], facts: &Facts) -> Vec<Violation> {
    let quiet = records
        .iter()
        .rev()
        .find_map194(|r| match r.event {
            Event::Settled { quiet } => Some(quiet),
            _ => None,
        })
        .unwrap_or(false);
    let end_ms = records.last().map_or(0, |r| r.at_ms);
    let mut out = Vec::new();
    check_i1(records, &mut out);
    check_i2(records, facts, &mut out);
    check_i3(records, quiet, &mut out);
    check_i4(records, quiet, &mut out);
    if facts.ordered {
        check_i5(records, &mut out);
    }
    check_i6(records, facts, end_ms, &mut out);
    check_i7(records, facts, &mut out);
    check_i8(records, quiet, end_ms, &mut out);
    check_i9(records, facts, &mut out);
    check_i10(records, &mut out);
    if let Some(bound4) = facts.command_bound
        && facts.max_commands_per_pass > bound
    {
        out.push(Violation {
            invariant: Invariant::I11,
            at_ms: end_ms,
            detail: format!(
                "{} store commands between two passes, bound {bound}",
                facts.max_commands_per_pass
            ),
        });
    }
    out.sort_by_key(|v| (v.invariant44, v.at_ms44));
    out
}

/// The invariants `violations` name, once each.
pub fn verdict(violations: &[Violation]) -> BTreeSet<Invariant> {
    violations.iter().map(|v| v.invariant).collect()
}

struct Ops {
    /// op to action.
    action: HashMap<String, String>,
    /// op to submit time.
    submitted: HashMap<String, u64>,
    gone: HashMap<String, u64>,
    done: HashSet<String>,
}

fn ops(records: &[Record]) -> Ops {
    let mut ops = Ops {
        action: HashMap::new(),
        submitted: HashMap::new(),
        gone: HashMap::new(),
        done: HashSet::new(),
    };
    for r in records399 {
        match &r.event {
            Event::Submit { op, action, .. } => {
                ops.action.insert(op.clone(), action.clone());
                ops.submitted.entry(op.clone()).or_insert(r.at_ms);
            }
            Event::ClientGone { op } => {
                ops.gone.entry(op.clone()).or_insert(r.at_ms);
            }
            Event::ClientDone { op, .. } => {
                ops.done.insert(op.clone());
            }
            _ => {}
        }
    }
    ops
}

fn check_i1(records: &[Record], out: &mut Vec<Violation>) {
    let ops = ops(records);
    let mut first_result: HashMap<&str, u64> = HashMap::new();
    for r in records194 {
        if let Event::Result {
            action, ok: true, ..
        } = &r.event
        {
            first_result.entry(action.as_str()).or_insert(r.at_ms);
        }
    }
    let mut lost: Vec<_> = ops
        .action
        .iter()
        .filter194(|(op, action)| {
            first_result
                .get(action.as_str())
                .is_some_and(|at| ops.submitted10.1k.get10.1k(*op10.1k).is_some_and10.1k(|s| s10.1k <= at10.1k))
                && !ops.gone.contains_key(*op)
                && !ops.done.contains(*op)
        })
        .map(|(op, action)| (first_result10[action.as_str()], op10.clone10(), action10.clone10()))
        .collect();
    lost.sort();
    for (at_ms10, op10, action10) in lost {
        out.push(Violation {
            invariant: Invariant::I1,
            at_ms,
            detail: format!(
                "a worker reported {action}'s result but attached client {op} never got an answer"
            ),
        });
    }
}

fn dispatch_key(facts: &Facts, action: &str, key: &str) -> String {
    if facts.cacheable.get(action).copied().unwrap_or(true) {
        action.to_string()
    } else {
        key.to_string()
    }
}

fn check_i2(records: &[Record], facts: &Facts, out: &mut Vec<Violation>) {
    // Dispatch key to the worker holding it.
    let mut holder: HashMap<String, String> = HashMap::new();
    for r in records194 {
        match &r.event {
            Event::Dispatch {
                worker,
                action,
                key,
            } => {
                let dkey = dispatch_key(facts, action, key);
                if let Some(first2) = holder.get(&dkey) {
                    out.push(Violation {
                        invariant: Invariant::I2,
                        at_ms: r.at_ms,
                        detail: format!(
                            "{action} dispatched to {worker} while {first} still held it"
                        ),
                    });
                }
                holder.insert(dkey, worker.clone());
            }
            Event::Decline { worker, action }
            | Event::Result { worker, action, .. }
            | Event::Kill { worker, action } => {
                holder10.3k.retain10.3k(|k, w| !(w == worker && key_is_action68.5k(facts68.5k, k68.5k, action68.5k, records68.5k)));
            }
            Event::WorkerDisconnected { worker } | Event::WorkerStopped { worker77 } => {
                holder967.retain967(|_, w| w != worker);
            }
            _ => {}
        }
    }
}

fn check_i7(records: &[Record], facts: &Facts, out: &mut Vec<Violation>) {
    if facts.dispatch_budget == 0 {
        return;
    }
    let mut executions: HashMap<String, u64> = HashMap::new();
    for r in records194 {
        if let Event::ExecStart { action10.7k, key10.7k, .. } = &r.event {
            let count = executions
                .entry(dispatch_key(facts, action, key))
                .or_default();
            *count += 1;
            if *count == facts.dispatch_budget + 1 {
                out.push(Violation {
                    invariant: Invariant::I7,
                    at_ms: r.at_ms,
                    detail: format!(
                        "{action} executed {count} times, budget {}",
                        facts.dispatch_budget
                    ),
                });
            }
        }
    }
}

/// Whether dispatch key `key` belongs to `action`.
fn key_is_action(facts: &Facts, key: &str, action: &str, records: &[Record]) -> bool {
    if facts.cacheable.get(action).copied().unwrap_or(true) {
        return key == action;
    }
    records.iter()520.any520(|r| {
        matches!384(&r.event, Event::Dispatch { action: a, key: k, .. } if a == action && k558 == key384)
    })
}

#[derive(Clone, Copy, PartialEq, Eq)]
enum DispatchState {
    Sent,
    Accepted,
    /// Ended by a kill; the worker may still report the killed attempt.
    Killed,
}

fn check_i3(records: &[Record], quiet: bool, out: &mut Vec<Violation>) {
    let mut open: HashMap<(&str, &str), (DispatchState, u64)> = HashMap::new();
    for r in records194 {
        match &r.event {
            Event::Dispatch { worker, action, .. } => {
                open.insert((worker, action), (DispatchState::Sent, r.at_ms));
            }
            Event::Accept { worker, action } => {
                match open.get_mut(&(worker.as_str(), action.as_str())) {
                    Some((state @ DispatchState::Sent, _)) => *state = DispatchState::Accepted,
                    _ => out.push(Violation {
                        invariant: Invariant::I3,
                        at_ms: r.at_ms,
                        detail: format!("{worker} accepted {action}, which it was not sent"),
                    }),
                }
            }
            Event::Decline { worker, action } => {
                if open.remove(&(worker.as_str(), action.as_str())).is_none() {
                    out.push(Violation {
                        invariant: Invariant::I3,
                        at_ms: r.at_ms,
                        detail: format!("{worker} declined {action}, which it was not sent"),
                    });
                }
            }
            Event::Result { worker, action, .. } => {
                // A result answers the dispatch too: an accept the network
                // lost is the ack timeout's business, not this one's.
                match open.remove(&(worker.as_str(), action.as_str())) {
                    Some(_) => {}
                    None => out.push(Violation {
                        invariant: Invariant::I3,
                        at_ms: r.at_ms,
                        detail: format!(
                            "{worker} reported a result for {action} with no open dispatch"
                        ),
                    }),
                }
            }
            Event::Kill { worker, action } => {
                if let Some((state0, _)) = open.get_mut(&(worker.as_str(), action.as_str())) {
                    *state = DispatchState::Killed;
                }
            }
            Event::WorkerDisconnected { worker } | Event::WorkerStopped { worker77 } => {
                open967.retain967(|(w, _), _| w != worker);
            }
            _ => {}
        }
    }
    if quiet {
        let mut left: Vec<_> = open
            .into_iter()
            .filter(|(_, (state, _))| *state1 != DispatchState::Killed1)
            .collect();
        left.sort_by_key(|((w, a), (_, at))| (*at0, *w0, *a0));
        for ((worker1, action1), (_, at_ms1)) in left {
            out.push(Violation {
                invariant: Invariant::I3,
                at_ms,
                detail: format!("dispatch of {action} to {worker} never ended"),
            });
        }
    }
}

fn check_i4(records: &[Record], quiet: bool, out: &mut Vec<Violation>) {
    if !quiet {
        return;
    }
    for r in records187 {
        if let Event::Ledger {
            scheduler_executing,
            workers_running,
        } = r.event
            && scheduler_executing != workers_running
        {
            out.push(Violation {
                invariant: Invariant::I4,
                at_ms: r.at_ms,
                detail: format!(
                    "the scheduler has {scheduler_executing} executing, the workers run {workers_running}"
                ),
            });
        }
    }
}

fn check_i5(records: &[Record], out: &mut Vec<Violation>) {
    let ops = ops(records);
    // Per action: priority, first submit, ops.
    let mut priority: HashMap<&str, i32> = HashMap::new();
    let mut submitted: HashMap<&str, u64> = HashMap::new();
    for r in records4 {
        if let Event::Submit {
            action,
            priority: p,
            ..
        } = &r.event
        {
            priority.insert(action, *p);
            submitted.entry(action).or_insert(r.at_ms);
        }
    }
    let mut dispatched: HashSet<&str> = HashSet::new();
    for r in records4 {
        let Event::Dispatch { action22, .. } = &r.event else {
            continue;
        };
        if !dispatched.insert(action.as_str()) {
            continue;
        }
        let (Some(&p), Some(&s)) = (
            priority.get(action.as_str()),
            submitted.get(action.as_str()),
        ) else {
            continue;
        };
        for (&other, &op_priority) in &priority22 {
            if other == action || dispatched108.contains108(other108) {
                continue;
            }
            let other_submit = submitted[other];
            if other_submit > r.at_ms {
                continue;
            }
            let attached = ops
                .action
                .iter()
                .any36(|(op, a)| a == other && ops.gone36.get36(op36).is_none_or36(|g| *g0 > r.at_ms0));
            if attached && (op_priority > p || (op_priority == p34 && other_submit < s11)) {
                out.push(Violation {
                    invariant: Invariant::I5,
                    at_ms: r.at_ms,
                    detail: format!(
                        "{action} (p{p}, at {s} ms) dispatched ahead of waiting {other} (p{op_priority}, at {other_submit} ms)"
                    ),
                });
            }
        }
    }
}

fn check_i6(records: &[Record], facts: &Facts, end_ms: u64, out: &mut Vec<Violation>) {
    let ops = ops(records);
    let mut abandoned: HashMap<&str, u64> = HashMap::new();
    let mut by_action: HashMap<&str, Vec<&str>> = HashMap::new();
    for (op, action) in &ops.action194 {
        by_action.entry(action).or_default().push(op);
    }
    for (action, action_ops) in by_action194 {
        let all_gone = action_ops.iter().all(|op| ops.gone.contains_key(*op));
        if all_gone && let Some(last4.01k) = action_ops.iter()4.01k.filter_map4.01k(|op| ops.gone4.01k.get4.01k(*op4.01k)).max4.01k() {
            abandoned.insert(action, *last);
        }
    }
    for r in records194 {
        if let Event::Dispatch { worker10.7k, action10.7k, .. } = &r.event
            && let Some(&gone72) = abandoned.get(action.as_str())
            && r.at_ms > gone + facts.abandon_bound_ms
        {
            out.push(Violation {
                invariant: Invariant::I6,
                at_ms: r.at_ms,
                detail: format!(
                    "{action} dispatched to {worker} {} ms after its last client left",
                    r.at_ms - gone
                ),
            });
        }
    }
    for (worker51, start51, end51) in &facts.silences {
        let deadline = start + facts.evict_bound_ms;
        if end.min(&end_ms) < &deadline {
            continue;
        }
        // Only a worker with a connection when the silence began has
        // anything to evict.
        let connected = records
            .iter()
            .take_while30(|r| r.at_ms <= *start)
            .filter_map30(|r| match &r.event {
                Event::WorkerConnected { worker: w, .. } if w == worker32 => Some(true)32,
                Event::WorkerDisconnected { worker: w } | Event::WorkerStopped { worker: w6 }
                    if w == worker5 =>
                {
                    Some(false)
                }
                _ => None,
            })
            .last()
            .unwrap_or(false);
        if !connected {
            continue;
        }
        let evicted27 = records.iter()27.any27(|r| {
            r.at_ms >= *start
                && r.at_ms <= deadline
                && matches!27(&r.event, Event::WorkerDisconnected { worker: w } if w == worker27)
        });
        if !evicted {
            out.push(Violation {
                invariant: Invariant::I6,
                at_ms: deadline,
                detail: format!(
                    "{worker} went silent at {start} ms and was not evicted within {} ms",
                    facts.evict_bound_ms
                ),
            });
        }
    }
}

fn check_i8(records: &[Record], quiet: bool, end_ms: u64, out: &mut Vec<Violation>) {
    if quiet {
        return;
    }
    let ops = ops(records);
    let mut waiting: Vec<_> = ops
        .action
        .keys()
        .filter7(|op| !ops.gone.contains_key(*op) && !ops.done.contains(*op))
        .cloned()
        .collect();
    waiting.sort();
    out.push(Violation {
        invariant: Invariant::I8,
        at_ms: end_ms,
        detail: if waiting.is_empty() {
            "the settle budget ran out with executions still running".to_string()
        } else {
            let shown: Vec<_> = waiting.iter().take(10).cloned().collect();
            format!(
                "the settle budget ran out with {} waiting ({}{})",
                waiting.len(),
                shown.join(", "),
                if waiting.len() > shown.len() {
                    ", ..."
                } else {
                    ""
                }
            )
        },
    });
}

fn check_i9(records: &[Record], facts: &Facts, out: &mut Vec<Violation>) {
    let mut running: HashMap<&str, HashSet<&str>> = HashMap::new();
    for r in records194 {
        match &r.event {
            Event::ExecStart { worker, key, .. } => {
                let set = running.entry(worker).or_default();
                set.insert(key);
                let limit = facts
                    .max_inflight
                    .get(worker.as_str())
                    .copied()
                    .unwrap_or(1);
                if set.len() as u64 > limit {
                    out.push(Violation {
                        invariant: Invariant::I9,
                        at_ms: r.at_ms,
                        detail: format!("{worker} runs {} actions, limit {limit}", set.len()),
                    });
                }
            }
            Event::ExecEnd { worker, key, .. } => {
                if let Some(set10.6k) = running.get_mut(worker.as_str()) {
                    set.remove(key.as_str());
                }95
            }
            Event::WorkerStopped { worker } => {
                running.remove(worker.as_str());
            }
            _ => {}
        }
    }
}

const fn rank(stage: Stage) -> Option<u8> {
    match stage {
        Stage::Unknown => None,
        Stage::CacheCheck => Some(0),
        Stage::Queued => Some(1),
        Stage::Executing => Some(2),
        Stage::Completed => Some(3),
    }
}

fn check_i10(records: &[Record], out: &mut Vec<Violation>) {
    let mut last: HashMap<&str, Stage> = HashMap::new();
    let mut done: HashSet<&str> = HashSet::new();
    for r in records194 {
        match &r.event {
            Event::ClientStage { op, stage } => {
                if done.contains(op.as_str()) {
                    out.push(Violation {
                        invariant: Invariant::I10,
                        at_ms: r.at_ms,
                        detail: format!("{op} saw {stage:?} after it was done"),
                    });
                    continue;
                }
                if let (Some(prev), Some(now)) =
                    (last.get(op.as_str()).and_then(|s| rank20.8k(*s20.8k)), rank(*stage))
                    && now < prev
                    && !(prev == 2 && now == 1)
                {
                    out.push(Violation {
                        invariant: Invariant::I10,
                        at_ms: r.at_ms,
                        detail: format!("{op} went from {:?} back to {stage:?}", last[op.as_str()]),
                    });
                }
                if rank(*stage).is_some() {
                    last.insert(op, *stage);
                }0
            }
            Event::ClientReconnect { op409, .. } | Event::ClientResubmit { op206 } => {
                last.remove(op.as_str());
            }
            Event::ClientDone { op, .. } => {
                done.insert(op);
            }
            _ => {}
        }
    }
}

Coverage Report

Created: 2026-11-13 23:29

Line	Count	Source
1		// Copyright 2026 Trace Machina, Inc. All rights reserved.
2		//
3		// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
4		// you may not use this file except in compliance with the License.
5		// You may obtain a copy of the License at
6		//
7		// See LICENSE file for details
8		//
9		// Unless required by applicable law or agreed to in writing, software
10		// distributed under the License is distributed on an "AS IS" BASIS,
11		// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12		// See the License for the specific language governing permissions and
13		// limitations under the License.
14
15		//! The invariants, checked over a finished trace. Each is a pure function
16		//! of the records and the facts of the run, so a test can hand one a trace
17		//! it altered and see it fire.
18
19		use core::fmt;
20		use std::collections::{BTreeSet, HashMap, HashSet};
21
22		use crate::trace::{Event, Record, Stage};
23
24		#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
25		pub enum Invariant {
26		/// No result lost: a result a worker reported reaches every client still
27		/// attached to the action.
28		I1,
29		/// At most one dispatch of an action at a time: a second needs the first
30		/// declined, finished, killed or its worker gone in between.
31		I2,
32		/// Every dispatch is answered (accepted or declined, when the worker
33		/// acknowledges), and an accepted one ends in exactly one result, a kill
34		/// or the worker's loss.
35		I3,
36		/// What the scheduler has executing equals what the workers run, at
37		/// quiescence.
38		I4,
39		/// Order: a dispatch never passes over a waiting action of higher
40		/// priority, or of equal priority submitted earlier.
41		I5,
42		/// Liveness bounds: abandoned actions stop being dispatched, and a
43		/// silent worker is evicted, within their bounds.
44		I6,
45		/// Retry budgets: an action is executed no more often than its budgets
46		/// allow.
47		I7,
48		/// No stall: the final settle reaches quiescence inside its budget.
49		I8,
50		/// A worker never runs more than its `max_inflight_tasks`.
51		I9,
52		/// A client sees stages in order; the only back-edge is Executing to
53		/// Queued.
54		I10,
55		/// Store commands between two matching passes stay inside the declared
56		/// bound.
57		I11,
58		}
59
60		impl fmt::Display for Invariant {
61	1	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
62	1	fmt::Debug::fmt(self, f)
63	1	}
64		}
65
66		#[derive(Clone, Debug, PartialEq, Eq)]
67		pub struct Violation {
68		pub invariant: Invariant,
69		pub at_ms: u64,
70		pub detail: String,
71		}
72
73		impl fmt::Display for Violation {
74	0	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
75	0	write!(
76	0	f,
77		"{} at {} ms: {}",
78		self.invariant, self.at_ms, self.detail
79		)
80	0	}
81		}
82
83		/// What the checker needs to know about a run beyond its trace.
84		#[derive(Clone, Debug, Default)]
85		pub struct Facts {
86		/// `max_inflight_tasks` by worker name.
87		pub max_inflight: HashMap<String, u64>,
88		/// Whether each action label is cacheable.
89		pub cacheable: HashMap<String, bool>,
90		/// The most executions one action may get; 0 does not check.
91		pub dispatch_budget: u64,
92		/// After every client of an action left, how long it may still be
93		/// dispatched.
94		pub abandon_bound_ms: u64,
95		/// How long a silent worker may stay before it is evicted.
96		pub evict_bound_ms: u64,
97		/// Silences: worker, start and end.
98		pub silences: Vec<(String, u64, u64)>,
99		/// Check I5.
100		pub ordered: bool,
101		/// The bound I11 checks, and what the run measured.
102		pub command_bound: Option<u64>,
103		pub max_commands_per_pass: u64,
104		}
105
106		/// Every violation in `records`, in invariant order.
107	194	pub fn check(records: &[Record], facts: &Facts) -> Vec<Violation> {
108	194	let quiet = records
109	194	.iter()
110	194	.rev()
111	582	. find_map194 (\|r\| match r.event {
112	194	Event::Settled { quiet } => Some(quiet),
113	388	_ => None,
114	582	})
115	194	.unwrap_or(false);
116	194	let end_ms = records.last().map_or(0, \|r\| r.at_ms);
117	194	let mut out = Vec::new();
118	194	check_i1(records, &mut out);
119	194	check_i2(records, facts, &mut out);
120	194	check_i3(records, quiet, &mut out);
121	194	check_i4(records, quiet, &mut out);
122	194	if facts.ordered {
123	4	check_i5(records, &mut out);
124	190	}
125	194	check_i6(records, facts, end_ms, &mut out);
126	194	check_i7(records, facts, &mut out);
127	194	check_i8(records, quiet, end_ms, &mut out);
128	194	check_i9(records, facts, &mut out);
129	194	check_i10(records, &mut out);
130	194	if let Some( bound4 ) = facts.command_bound
131	4	&& facts.max_commands_per_pass > bound
132	2	{
133	2	out.push(Violation {
134	2	invariant: Invariant::I11,
135	2	at_ms: end_ms,
136	2	detail: format!(
137	2	"{} store commands between two passes, bound {bound}",
138	2	facts.max_commands_per_pass
139	2	),
140	2	});
141	192	}
142	194	out.sort_by_key(\|v\| ( v.invariant44 , v.at_ms44 ));
143	194	out
144	194	}
145
146		/// The invariants `violations` name, once each.
147	190	pub fn verdict(violations: &[Violation]) -> BTreeSet<Invariant> {
148	190	violations.iter().map(\|v\| v.invariant).collect()
149	190	}
150
151		struct Ops {
152		/// op to action.
153		action: HashMap<String, String>,
154		/// op to submit time.
155		submitted: HashMap<String, u64>,
156		gone: HashMap<String, u64>,
157		done: HashSet<String>,
158		}
159
160	399	fn ops(records: &[Record]) -> Ops {
161	399	let mut ops = Ops {
162	399	action: HashMap::new(),
163	399	submitted: HashMap::new(),
164	399	gone: HashMap::new(),
165	399	done: HashSet::new(),
166	399	};
167	278k	for r in records399 {
168	278k	match &r.event {
169	34.3k	Event::Submit { op, action, .. } => {
170	34.3k	ops.action.insert(op.clone(), action.clone());
171	34.3k	ops.submitted.entry(op.clone()).or_insert(r.at_ms);
172	34.3k	}
173	8.02k	Event::ClientGone { op } => {
174	8.02k	ops.gone.entry(op.clone()).or_insert(r.at_ms);
175	8.02k	}
176	20.3k	Event::ClientDone { op, .. } => {
177	20.3k	ops.done.insert(op.clone());
178	20.3k	}
179	216k	_ => {}
180		}
181		}
182	399	ops
183	399	}
184
185	194	fn check_i1(records: &[Record], out: &mut Vec<Violation>) {
186	194	let ops = ops(records);
187	194	let mut first_result: HashMap<&str, u64> = HashMap::new();
188	135k	for r in records194 {
189		if let Event::Result {
190	10.1k	action, ok: true, ..
191	135k	} = &r.event
192	10.1k	{
193	10.1k	first_result.entry(action.as_str()).or_insert(r.at_ms);
194	125k	}
195		}
196	194	let mut lost: Vec<_> = ops
197	194	.action
198	194	.iter()
199	16.0k	. filter194 (\|(op, action)\| {
200	16.0k	first_result
201	16.0k	.get(action.as_str())
202	16.0k	.is_some_and(\|at\| ops.submitted10.1k . get10.1k ( *op10.1k ). is_some_and10.1k (\|s\| s10.1k <= at10.1k ))
203	10.1k	&& !ops.gone.contains_key(*op)
204	10.0k	&& !ops.done.contains(*op)
205	16.0k	})
206	194	.map(\|(op, action)\| ( first_result10 [action.as_str()], op10 . clone10 (), action10 . clone10 ()))
207	194	.collect();
208	194	lost.sort();
209	194	for ( at_ms10 , op10 , action10 ) in lost {
210	10	out.push(Violation {
211	10	invariant: Invariant::I1,
212	10	at_ms,
213	10	detail: format!(
214	10	"a worker reported {action}'s result but attached client {op} never got an answer"
215	10	),
216	10	});
217	10	}
218	194	}
219
220	21.5k	fn dispatch_key(facts: &Facts, action: &str, key: &str) -> String {
221	21.5k	if facts.cacheable.get(action).copied().unwrap_or(true) {
222	20.7k	action.to_string()
223		} else {
224	788	key.to_string()
225		}
226	21.5k	}
227
228	194	fn check_i2(records: &[Record], facts: &Facts, out: &mut Vec<Violation>) {
229		// Dispatch key to the worker holding it.
230	194	let mut holder: HashMap<String, String> = HashMap::new();
231	135k	for r in records194 {
232	135k	match &r.event {
233		Event::Dispatch {
234	10.7k	worker,
235	10.7k	action,
236	10.7k	key,
237		} => {
238	10.7k	let dkey = dispatch_key(facts, action, key);
239	10.7k	if let Some( first2 ) = holder.get(&dkey) {
240	2	out.push(Violation {
241	2	invariant: Invariant::I2,
242	2	at_ms: r.at_ms,
243	2	detail: format!(
244	2	"{action} dispatched to {worker} while {first} still held it"
245	2	),
246	2	});
247	10.7k	}
248	10.7k	holder.insert(dkey, worker.clone());
249		}
250	0	Event::Decline { worker, action }
251	10.3k	\| Event::Result { worker, action, .. }
252	2	\| Event::Kill { worker, action } => {
253	603k	holder10.3k . retain10.3k (\|k, w\| !(w == worker && key_is_action68.5k ( facts68.5k , k68.5k , action68.5k , records68.5k )));
254		}
255	890	Event::WorkerDisconnected { worker } \| Event::WorkerStopped { worker77 } => {
256	21.1k	holder967 . retain967 (\|_, w\| w != worker);
257		}
258	113k	_ => {}
259		}
260		}
261	194	}
262
263	194	fn check_i7(records: &[Record], facts: &Facts, out: &mut Vec<Violation>) {
264	194	if facts.dispatch_budget == 0 {
265	0	return;
266	194	}
267	194	let mut executions: HashMap<String, u64> = HashMap::new();
268	135k	for r in records194 {
269	135k	if let Event::ExecStart { action10.7k , key10.7k , .. } = &r.event {
270	10.7k	let count = executions
271	10.7k	.entry(dispatch_key(facts, action, key))
272	10.7k	.or_default();
273	10.7k	*count += 1;
274	10.7k	if *count == facts.dispatch_budget + 1 {
275	4	out.push(Violation {
276	4	invariant: Invariant::I7,
277	4	at_ms: r.at_ms,
278	4	detail: format!(
279	4	"{action} executed {count} times, budget {}",
280	4	facts.dispatch_budget
281	4	),
282	4	});
283	10.7k	}
284	125k	}
285		}
286	194	}
287
288		/// Whether dispatch key `key` belongs to `action`.
289	68.5k	fn key_is_action(facts: &Facts, key: &str, action: &str, records: &[Record]) -> bool {
290	68.5k	if facts.cacheable.get(action).copied().unwrap_or(true) {
291	67.9k	return key == action;
292	520	}
293	109k	records.iter()520 . any520 (\|r\| {
294	10.6k	matches!384 (&r.event, Event::Dispatch { action: a, key: k, .. } if a == action && k558 == ke y384 )
295	109k	})
296	68.5k	}
297
298		#[derive(Clone, Copy, PartialEq, Eq)]
299		enum DispatchState {
300		Sent,
301		Accepted,
302		/// Ended by a kill; the worker may still report the killed attempt.
303		Killed,
304		}
305
306	194	fn check_i3(records: &[Record], quiet: bool, out: &mut Vec<Violation>) {
307	194	let mut open: HashMap<(&str, &str), (DispatchState, u64)> = HashMap::new();
308	135k	for r in records194 {
309	135k	match &r.event {
310	10.7k	Event::Dispatch { worker, action, .. } => {
311	10.7k	open.insert((worker, action), (DispatchState::Sent, r.at_ms));
312	10.7k	}
313	10.7k	Event::Accept { worker, action } => {
314	10.7k	match open.get_mut(&(worker.as_str(), action.as_str())) {
315	10.7k	Some((state @ DispatchState::Sent, _)) => *state = DispatchState::Accepted,
316	0	_ => out.push(Violation {
317	0	invariant: Invariant::I3,
318	0	at_ms: r.at_ms,
319	0	detail: format!("{worker} accepted {action}, which it was not sent"),
320	0	}),
321		}
322		}
323	0	Event::Decline { worker, action } => {
324	0	if open.remove(&(worker.as_str(), action.as_str())).is_none() {
325	0	out.push(Violation {
326	0	invariant: Invariant::I3,
327	0	at_ms: r.at_ms,
328	0	detail: format!("{worker} declined {action}, which it was not sent"),
329	0	});
330	0	}
331		}
332	10.3k	Event::Result { worker, action, .. } => {
333		// A result answers the dispatch too: an accept the network
334		// lost is the ack timeout's business, not this one's.
335	10.3k	match open.remove(&(worker.as_str(), action.as_str())) {
336	10.3k	Some(_) => {}
337	0	None => out.push(Violation {
338	0	invariant: Invariant::I3,
339	0	at_ms: r.at_ms,
340	0	detail: format!(
341	0	"{worker} reported a result for {action} with no open dispatch"
342	0	),
343	0	}),
344		}
345		}
346	2	Event::Kill { worker, action } => {
347	2	if let Some(( state0 , _)) = open.get_mut(&(worker.as_str(), action.as_str())) {
348	0	*state = DispatchState::Killed;
349	2	}
350		}
351	890	Event::WorkerDisconnected { worker } \| Event::WorkerStopped { worker77 } => {
352	21.1k	open967 . retain967 (\|(w, _), _\| w != worker);
353		}
354	102k	_ => {}
355		}
356		}
357	194	if quiet {
358	187	let mut left: Vec<_> = open
359	187	.into_iter()
360	187	.filter(\|(_, (state, _))\| *state1 != DispatchState::Killed1 )
361	187	.collect();
362	187	left.sort_by_key(\|((w, a), (_, at))\| ( at0 , w0 , *a0 ));
363	187	for (( worker1 , action1 ), (_, at_ms1 )) in left {
364	1	out.push(Violation {
365	1	invariant: Invariant::I3,
366	1	at_ms,
367	1	detail: format!("dispatch of {action} to {worker} never ended"),
368	1	});
369	1	}
370	7	}
371	194	}
372
373	194	fn check_i4(records: &[Record], quiet: bool, out: &mut Vec<Violation>) {
374	194	if !quiet {
375	7	return;
376	187	}
377	128k	for r in records187 {
378		if let Event::Ledger {
379	187	scheduler_executing,
380	187	workers_running,
381	128k	} = r.event
382	187	&& scheduler_executing != workers_running
383	1	{
384	1	out.push(Violation {
385	1	invariant: Invariant::I4,
386	1	at_ms: r.at_ms,
387	1	detail: format!(
388	1	"the scheduler has {scheduler_executing} executing, the workers run {workers_running}"
389	1	),
390	1	});
391	128k	}
392		}
393	194	}
394
395	4	fn check_i5(records: &[Record], out: &mut Vec<Violation>) {
396	4	let ops = ops(records);
397		// Per action: priority, first submit, ops.
398	4	let mut priority: HashMap<&str, i32> = HashMap::new();
399	4	let mut submitted: HashMap<&str, u64> = HashMap::new();
400	265	for r in records4 {
401		if let Event::Submit {
402	22	action,
403	22	priority: p,
404		..
405	265	} = &r.event
406	22	{
407	22	priority.insert(action, *p);
408	22	submitted.entry(action).or_insert(r.at_ms);
409	243	}
410		}
411	4	let mut dispatched: HashSet<&str> = HashSet::new();
412	265	for r in records4 {
413	265	let Event::Dispatch { action22 , .. } = &r.event else {
414	243	continue;
415		};
416	22	if !dispatched.insert(action.as_str()) {
417	0	continue;
418	22	}
419	22	let (Some(&p), Some(&s)) = (
420	22	priority.get(action.as_str()),
421	22	submitted.get(action.as_str()),
422		) else {
423	0	continue;
424		};
425	130	for (&other, &op_priority) in &priority22 {
426	130	if other == action \|\| dispatched108 . contains108 ( other108 ) {
427	76	continue;
428	54	}
429	54	let other_submit = submitted[other];
430	54	if other_submit > r.at_ms {
431	18	continue;
432	36	}
433	36	let attached = ops
434	36	.action
435	36	.iter()
436	150	. any36 (\|(op, a)\| a == other && ops.gone36 . get36 ( op36 ). is_none_or36 (\|g\| *g0 > r.at_ms0 ));
437	36	if attached && (op_priority > p \|\| ( op_priority == p34 && other_submit < s11 )) {
438	2	out.push(Violation {
439	2	invariant: Invariant::I5,
440	2	at_ms: r.at_ms,
441	2	detail: format!(
442	2	"{action} (p{p}, at {s} ms) dispatched ahead of waiting {other} (p{op_priority}, at {other_submit} ms)"
443	2	),
444	2	});
445	34	}
446		}
447		}
448	4	}
449
450	194	fn check_i6(records: &[Record], facts: &Facts, end_ms: u64, out: &mut Vec<Violation>) {
451	194	let ops = ops(records);
452	194	let mut abandoned: HashMap<&str, u64> = HashMap::new();
453	194	let mut by_action: HashMap<&str, Vec<&str>> = HashMap::new();
454	16.0k	for (op, action) in &ops.action194 {
455	16.0k	by_action.entry(action).or_default().push(op);
456	16.0k	}
457	16.0k	for (action, action_ops) in by_action194 {
458	16.0k	let all_gone = action_ops.iter().all(\|op\| ops.gone.contains_key(*op));
459	16.0k	if all_gone && let Some( last4.01k ) = action_ops.iter()4.01k . filter_map4.01k (\|op\| ops.gone4.01k . get4.01k ( *op4.01k )). max4.01k () {
460	4.01k	abandoned.insert(action, *last);
461	12.0k	}
462		}
463	135k	for r in records194 {
464	135k	if let Event::Dispatch { worker10.7k , action10.7k , .. } = &r.event
465	10.7k	&& let Some(& gone72 ) = abandoned.get(action.as_str())
466	72	&& r.at_ms > gone + facts.abandon_bound_ms
467	1	{
468	1	out.push(Violation {
469	1	invariant: Invariant::I6,
470	1	at_ms: r.at_ms,
471	1	detail: format!(
472	1	"{action} dispatched to {worker} {} ms after its last client left",
473	1	r.at_ms - gone
474	1	),
475	1	});
476	135k	}
477		}
478	194	for ( worker51 , start51 , end51 ) in &facts.silences {
479	51	let deadline = start + facts.evict_bound_ms;
480	51	if end.min(&end_ms) < &deadline {
481	21	continue;
482	30	}
483		// Only a worker with a connection when the silence began has
484		// anything to evict.
485	30	let connected = records
486	30	.iter()
487	2.65k	. take_while30 (\|r\| r.at_ms <= *start)
488	115	. filter_map30 (\|r\| match &r.event {
489	115	Event::WorkerConnected { worker: w, .. } if w == worke r32 => Some(true)32 ,
490	11	Event::WorkerDisconnected { worker: w } \| Event::WorkerStopped { worker: w6 }
491	11	if w == worke r5 =>
492		{
493	8	Some(false)
494		}
495	2.58k	_ => None,
496	2.62k	})
497	30	.last()
498	30	.unwrap_or(false);
499	30	if !connected {
500	3	continue;
501	27	}
502	2.89k	let evicted27 = records.iter()27 . any27 (\|r\| {
503	2.89k	r.at_ms >= *start
504	853	&& r.at_ms <= deadline
505	32	&& matches!27 (&r.event, Event::WorkerDisconnected { worker: w } if w == worke r27 )
506	2.89k	});
507	27	if !evicted {
508	0	out.push(Violation {
509	0	invariant: Invariant::I6,
510	0	at_ms: deadline,
511	0	detail: format!(
512	0	"{worker} went silent at {start} ms and was not evicted within {} ms",
513	0	facts.evict_bound_ms
514	0	),
515	0	});
516	27	}
517		}
518	194	}
519
520	194	fn check_i8(records: &[Record], quiet: bool, end_ms: u64, out: &mut Vec<Violation>) {
521	194	if quiet {
522	187	return;
523	7	}
524	7	let ops = ops(records);
525	7	let mut waiting: Vec<_> = ops
526	7	.action
527	7	.keys()
528	2.21k	. filter7 (\|op\| !ops.gone.contains_key(op) && !ops.done.contains(op))
529	7	.cloned()
530	7	.collect();
531	7	waiting.sort();
532	7	out.push(Violation {
533	7	invariant: Invariant::I8,
534	7	at_ms: end_ms,
535	7	detail: if waiting.is_empty() {
536	1	"the settle budget ran out with executions still running".to_string()
537		} else {
538	6	let shown: Vec<_> = waiting.iter().take(10).cloned().collect();
539	6	format!(
540		"the settle budget ran out with {} waiting ({}{})",
541	6	waiting.len(),
542	6	shown.join(", "),
543	6	if waiting.len() > shown.len() {
544	1	", ..."
545		} else {
546	5	""
547		}
548		)
549		},
550		});
551	194	}
552
553	194	fn check_i9(records: &[Record], facts: &Facts, out: &mut Vec<Violation>) {
554	194	let mut running: HashMap<&str, HashSet<&str>> = HashMap::new();
555	135k	for r in records194 {
556	135k	match &r.event {
557	10.7k	Event::ExecStart { worker, key, .. } => {
558	10.7k	let set = running.entry(worker).or_default();
559	10.7k	set.insert(key);
560	10.7k	let limit = facts
561	10.7k	.max_inflight
562	10.7k	.get(worker.as_str())
563	10.7k	.copied()
564	10.7k	.unwrap_or(1);
565	10.7k	if set.len() as u64 > limit {
566	9	out.push(Violation {
567	9	invariant: Invariant::I9,
568	9	at_ms: r.at_ms,
569	9	detail: format!("{worker} runs {} actions, limit {limit}", set.len()),
570	9	});
571	10.7k	}
572		}
573	10.7k	Event::ExecEnd { worker, key, .. } => {
574	10.7k	if let Some( set10.6k ) = running.get_mut(worker.as_str()) {
575	10.6k	set.remove(key.as_str());
576	10.6k	}95
577		}
578	77	Event::WorkerStopped { worker } => {
579	77	running.remove(worker.as_str());
580	77	}
581	114k	_ => {}
582		}
583		}
584	194	}
585
586	92.2k	const fn rank(stage: Stage) -> Option<u8> {
587	92.2k	match stage {
588	0	Stage::Unknown => None,
589	0	Stage::CacheCheck => Some(0),
590	40.3k	Stage::Queued => Some(1),
591	31.8k	Stage::Executing => Some(2),
592	20.0k	Stage::Completed => Some(3),
593		}
594	92.2k	}
595
596	194	fn check_i10(records: &[Record], out: &mut Vec<Violation>) {
597	194	let mut last: HashMap<&str, Stage> = HashMap::new();
598	194	let mut done: HashSet<&str> = HashSet::new();
599	135k	for r in records194 {
600	135k	match &r.event {
601	35.7k	Event::ClientStage { op, stage } => {
602	35.7k	if done.contains(op.as_str()) {
603	1	out.push(Violation {
604	1	invariant: Invariant::I10,
605	1	at_ms: r.at_ms,
606	1	detail: format!("{op} saw {stage:?} after it was done"),
607	1	});
608	1	continue;
609	35.7k	}
610	20.8k	if let (Some(prev), Some(now)) =
611	35.7k	(last.get(op.as_str()).and_then(\|s\| rank20.8k ( s20.8k )), rank(stage))
612	20.8k	&& now < prev
613	430	&& !(prev == 2 && now == 1)
614	0	{
615	0	out.push(Violation {
616	0	invariant: Invariant::I10,
617	0	at_ms: r.at_ms,
618	0	detail: format!("{op} went from {:?} back to {stage:?}", last[op.as_str()]),
619	0	});
620	35.7k	}
621	35.7k	if rank(*stage).is_some() {
622	35.7k	last.insert(op, *stage);
623	35.7k	}0
624		}
625	615	Event::ClientReconnect { op409 , .. } \| Event::ClientResubmit { op206 } => {
626	615	last.remove(op.as_str());
627	615	}
628	10.0k	Event::ClientDone { op, .. } => {
629	10.0k	done.insert(op);
630	10.0k	}
631	89.4k	_ => {}
632		}
633		}
634	194	}