/build/source/nativelink-sim/src/trace.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.

//! Everything that crosses a boundary the harness owns, in the order it
//! happened: the client streams, the worker connections, the simulated
//! executions, the faults and the clock. Names are the scenario's own
//! (`op`, `action`, `worker`), never the ids the servers mint, so two runs
//! of one scenario can be compared line by line.

use core::fmt;
use std::sync::Arc;

use parking_lot::Mutex;

use crate::clock::SimClock;

/// The stage a client saw on its `Execute` or `WaitExecution` stream.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum Stage {
    Unknown,
    CacheCheck,
    Queued,
    Executing,
    Completed,
}

impl Stage {
    pub(crate) const fn from_proto(value: i32) -> Self {
        match value {
            1 => Self::CacheCheck,
            2 => Self::Queued,
            3 => Self::Executing,
            4 => Self::Completed,
            _ => Self::Unknown,
        }
    }
}

/// How a client's operation ended.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum Outcome {
    /// The action ran and exited with this code.
    Exited(i32),
    /// The operation failed with this gRPC code.
    Failed(i32),
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Event {
    /// A client called `Execute` for `action`.
    Submit {
        op: String,
        action: String,
        priority: i32,
    },
    ClientStage {
        op: String,
        stage: Stage,
    },
    ClientDone {
        op: String,
        outcome: Outcome,
    },
    /// The client dropped its stream for good.
    ClientGone {
        op: String,
    },
    /// The stream ended or failed before the operation was done; the
    /// client re-subscribes with `WaitExecution`, as Bazel does.
    ClientReconnect {
        op: String,
        code: i32,
    },
    /// `WaitExecution` did not know the operation; the client calls
    /// `Execute` again.
    ClientResubmit {
        op: String,
    },
    /// The scheduler sent `StartExecute`. `key` is the scheduler's
    /// operation id, which a requeue keeps and a new execution does not.
    Dispatch {
        worker: String,
        action: String,
        key: String,
    },
    Accept {
        worker: String,
        action: String,
    },
    Decline {
        worker: String,
        action: String,
    },
    /// The worker released the slot (`ExecuteComplete`).
    Complete {
        worker: String,
        action: String,
    },
    /// The worker reported a result: `ok` is a completed stage, otherwise
    /// an internal error.
    Result {
        worker: String,
        action: String,
        ok: bool,
    },
    /// The scheduler asked the worker to kill the operation.
    Kill {
        worker: String,
        action: String,
    },
    ExecStart {
        worker: String,
        action: String,
        key: String,
    },
    ExecEnd {
        worker: String,
        action: String,
        key: String,
        killed: bool,
    },
    /// The worker's `kill_all`, after it lost its connection.
    KillAll {
        worker: String,
    },
    WorkerStarted {
        worker: String,
    },
    /// The scheduler answered the connection; `dispatch_ack` says the worker
    /// acknowledges dispatches.
    WorkerConnected {
        worker: String,
        dispatch_ack: bool,
    },
    /// The scheduler's stream to the worker ended (eviction, restart,
    /// a cut link).
    WorkerDisconnected {
        worker: String,
    },
    /// The worker process is gone.
    WorkerStopped {
        worker: String,
    },
    Fault {
        what: String,
    },
    SchedulerUp,
    SchedulerDown,
    /// What the clock reads jumped forward; timers did not move.
    ClockJump {
        ms: u64,
    },
    /// What the scheduler and the workers held at the end: operations the
    /// scheduler has executing, and executions the workers are running.
    Ledger {
        scheduler_executing: usize,
        workers_running: usize,
    },
    Settled {
        quiet: bool,
    },
}

impl fmt::Display for Event {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Submit {
                op,
                action,
                priority,
            } => write!(f, "submit {op} {action} p{priority}"),
            Self::ClientStage { op, stage } => write!(f, "stage {op} {stage:?}"),
            Self::ClientDone { op, outcome } => write!(f, "done {op} {outcome:?}"),
            Self::ClientGone { op } => write!(f, "gone {op}"),
            Self::ClientReconnect { op, code } => write!(f, "reconnect {op} code {code}"),
            Self::ClientResubmit { op } => write!(f, "resubmit {op}"),
            Self::Dispatch {
                worker,
                action,
                key,
            } => write!(f, "dispatch {worker} {action} {key}"),
            Self::Accept { worker, action } => write!(f, "accept {worker} {action}"),
            Self::Decline { worker, action } => write!(f, "decline {worker} {action}"),
            Self::Complete { worker, action } => write!(f, "complete {worker} {action}"),
            Self::Result { worker, action, ok } => write!(f, "result {worker} {action} ok={ok}"),
            Self::Kill { worker, action } => write!(f, "kill {worker} {action}"),
            Self::ExecStart {
                worker,
                action,
                key,
            } => write!(f, "exec-start {worker} {action} {key}"),
            Self::ExecEnd {
                worker,
                action,
                key,
                killed,
            } => write!(f, "exec-end {worker} {action} {key} killed={killed}"),
            Self::KillAll { worker } => write!(f, "kill-all {worker}"),
            Self::WorkerStarted { worker } => write!(f, "worker-started {worker}"),
            Self::WorkerConnected {
                worker,
                dispatch_ack,
            } => write!(f, "worker-connected {worker} ack={dispatch_ack}"),
            Self::WorkerDisconnected { worker } => write!(f, "worker-disconnected {worker}"),
            Self::WorkerStopped { worker } => write!(f, "worker-stopped {worker}"),
            Self::Fault { what } => write!(f, "fault {what}"),
            Self::SchedulerUp => write!(f, "scheduler-up"),
            Self::SchedulerDown => write!(f, "scheduler-down"),
            Self::ClockJump { ms } => write!(f, "clock-jump +{ms}ms"),
            Self::Ledger {
                scheduler_executing,
                workers_running,
            } => write!(
                f,
                "ledger scheduler={scheduler_executing} workers={workers_running}"
            ),
            Self::Settled { quiet } => write!(f, "settled quiet={quiet}"),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Record {
    pub at_ms: u64,
    pub event: Event,
}

impl fmt::Display for Record {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{:>9} {}", self.at_ms, self.event)
    }
}

/// The shared trace. Cheap to clone; every component holds one.
#[derive(Clone, Debug, Default)]
pub struct Trace(Arc<Mutex<Vec<Record>>>);

impl Trace {
    pub fn record(&self, event: Event) {
        self.0.lock().push(Record {
            at_ms: SimClock::now_ms(),
            event,
        });
    }

    pub fn records(&self) -> Vec<Record> {
        self.0.lock().clone()
    }

    pub fn len(&self) -> usize {
        self.0.lock().len()
    }

    pub fn is_empty(&self) -> bool {
        self.0.lock().is_empty()
    }
}

/// The trace as text, one record per line.
pub fn render(records: &[Record]) -> String {
    let mut out = String::new();
    for record in records11 {
        out.push_str(&record.to_string());
        out.push('\n');
    }
    out
}

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		//! Everything that crosses a boundary the harness owns, in the order it
16		//! happened: the client streams, the worker connections, the simulated
17		//! executions, the faults and the clock. Names are the scenario's own
18		//! (`op`, `action`, `worker`), never the ids the servers mint, so two runs
19		//! of one scenario can be compared line by line.
20
21		use core::fmt;
22		use std::sync::Arc;
23
24		use parking_lot::Mutex;
25
26		use crate::clock::SimClock;
27
28		/// The stage a client saw on its `Execute` or `WaitExecution` stream.
29		#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
30		pub enum Stage {
31		Unknown,
32		CacheCheck,
33		Queued,
34		Executing,
35		Completed,
36		}
37
38		impl Stage {
39	40.9k	pub(crate) const fn from_proto(value: i32) -> Self {
40	40.9k	match value {
41	0	1 => Self::CacheCheck,
42	18.6k	2 => Self::Queued,
43	12.2k	3 => Self::Executing,
44	9.96k	4 => Self::Completed,
45	0	_ => Self::Unknown,
46		}
47	40.9k	}
48		}
49
50		/// How a client's operation ended.
51		#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
52		pub enum Outcome {
53		/// The action ran and exited with this code.
54		Exited(i32),
55		/// The operation failed with this gRPC code.
56		Failed(i32),
57		}
58
59		#[derive(Clone, Debug, PartialEq, Eq)]
60		pub enum Event {
61		/// A client called `Execute` for `action`.
62		Submit {
63		op: String,
64		action: String,
65		priority: i32,
66		},
67		ClientStage {
68		op: String,
69		stage: Stage,
70		},
71		ClientDone {
72		op: String,
73		outcome: Outcome,
74		},
75		/// The client dropped its stream for good.
76		ClientGone {
77		op: String,
78		},
79		/// The stream ended or failed before the operation was done; the
80		/// client re-subscribes with `WaitExecution`, as Bazel does.
81		ClientReconnect {
82		op: String,
83		code: i32,
84		},
85		/// `WaitExecution` did not know the operation; the client calls
86		/// `Execute` again.
87		ClientResubmit {
88		op: String,
89		},
90		/// The scheduler sent `StartExecute`. `key` is the scheduler's
91		/// operation id, which a requeue keeps and a new execution does not.
92		Dispatch {
93		worker: String,
94		action: String,
95		key: String,
96		},
97		Accept {
98		worker: String,
99		action: String,
100		},
101		Decline {
102		worker: String,
103		action: String,
104		},
105		/// The worker released the slot (`ExecuteComplete`).
106		Complete {
107		worker: String,
108		action: String,
109		},
110		/// The worker reported a result: `ok` is a completed stage, otherwise
111		/// an internal error.
112		Result {
113		worker: String,
114		action: String,
115		ok: bool,
116		},
117		/// The scheduler asked the worker to kill the operation.
118		Kill {
119		worker: String,
120		action: String,
121		},
122		ExecStart {
123		worker: String,
124		action: String,
125		key: String,
126		},
127		ExecEnd {
128		worker: String,
129		action: String,
130		key: String,
131		killed: bool,
132		},
133		/// The worker's `kill_all`, after it lost its connection.
134		KillAll {
135		worker: String,
136		},
137		WorkerStarted {
138		worker: String,
139		},
140		/// The scheduler answered the connection; `dispatch_ack` says the worker
141		/// acknowledges dispatches.
142		WorkerConnected {
143		worker: String,
144		dispatch_ack: bool,
145		},
146		/// The scheduler's stream to the worker ended (eviction, restart,
147		/// a cut link).
148		WorkerDisconnected {
149		worker: String,
150		},
151		/// The worker process is gone.
152		WorkerStopped {
153		worker: String,
154		},
155		Fault {
156		what: String,
157		},
158		SchedulerUp,
159		SchedulerDown,
160		/// What the clock reads jumped forward; timers did not move.
161		ClockJump {
162		ms: u64,
163		},
164		/// What the scheduler and the workers held at the end: operations the
165		/// scheduler has executing, and executions the workers are running.
166		Ledger {
167		scheduler_executing: usize,
168		workers_running: usize,
169		},
170		Settled {
171		quiet: bool,
172		},
173		}
174
175		impl fmt::Display for Event {
176	2.35k	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
177	2.35k	match self {
178		Self::Submit {
179	141	op,
180	141	action,
181	141	priority,
182	141	} => write!(f, "submit {op} {action} p{priority}"),
183	583	Self::ClientStage { op, stage } => write!(f, "stage {op} {stage:?}"),
184	141	Self::ClientDone { op, outcome } => write!(f, "done {op} {outcome:?}"),
185	0	Self::ClientGone { op } => write!(f, "gone {op}"),
186	0	Self::ClientReconnect { op, code } => write!(f, "reconnect {op} code {code}"),
187	0	Self::ClientResubmit { op } => write!(f, "resubmit {op}"),
188		Self::Dispatch {
189	221	worker,
190	221	action,
191	221	key,
192	221	} => write!(f, "dispatch {worker} {action} {key}"),
193	221	Self::Accept { worker, action } => write!(f, "accept {worker} {action}"),
194	0	Self::Decline { worker, action } => write!(f, "decline {worker} {action}"),
195	141	Self::Complete { worker, action } => write!(f, "complete {worker} {action}"),
196	141	Self::Result { worker, action, ok } => write!(f, "result {worker} {action} ok={ok}"),
197	0	Self::Kill { worker, action } => write!(f, "kill {worker} {action}"),
198		Self::ExecStart {
199	221	worker,
200	221	action,
201	221	key,
202	221	} => write!(f, "exec-start {worker} {action} {key}"),
203		Self::ExecEnd {
204	221	worker,
205	221	action,
206	221	key,
207	221	killed,
208	221	} => write!(f, "exec-end {worker} {action} {key} killed={killed}"),
209	20	Self::KillAll { worker } => write!(f, "kill-all {worker}"),
210	61	Self::WorkerStarted { worker } => write!(f, "worker-started {worker}"),
211		Self::WorkerConnected {
212	81	worker,
213	81	dispatch_ack,
214	81	} => write!(f, "worker-connected {worker} ack={dispatch_ack}"),
215	40	Self::WorkerDisconnected { worker } => write!(f, "worker-disconnected {worker}"),
216	20	Self::WorkerStopped { worker } => write!(f, "worker-stopped {worker}"),
217	20	Self::Fault { what } => write!(f, "fault {what}"),
218	21	Self::SchedulerUp => write!(f, "scheduler-up"),
219	21	Self::SchedulerDown => write!(f, "scheduler-down"),
220	0	Self::ClockJump { ms } => write!(f, "clock-jump +{ms}ms"),
221		Self::Ledger {
222	21	scheduler_executing,
223	21	workers_running,
224	21	} => write!(
225	21	f,
226		"ledger scheduler={scheduler_executing} workers={workers_running}"
227		),
228	21	Self::Settled { quiet } => write!(f, "settled quiet={quiet}"),
229		}
230	2.35k	}
231		}
232
233		#[derive(Clone, Debug, PartialEq, Eq)]
234		pub struct Record {
235		pub at_ms: u64,
236		pub event: Event,
237		}
238
239		impl fmt::Display for Record {
240	1.18k	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
241	1.18k	write!(f, "{:>9} {}", self.at_ms, self.event)
242	1.18k	}
243		}
244
245		/// The shared trace. Cheap to clone; every component holds one.
246		#[derive(Clone, Debug, Default)]
247		pub struct Trace(Arc<Mutex<Vec<Record>>>);
248
249		impl Trace {
250	134k	pub fn record(&self, event: Event) {
251	134k	self.0.lock().push(Record {
252	134k	at_ms: SimClock::now_ms(),
253	134k	event,
254	134k	});
255	134k	}
256
257	364	pub fn records(&self) -> Vec<Record> {
258	364	self.0.lock().clone()
259	364	}
260
261	30.7k	pub fn len(&self) -> usize {
262	30.7k	self.0.lock().len()
263	30.7k	}
264
265	0	pub fn is_empty(&self) -> bool {
266	0	self.0.lock().is_empty()
267	0	}
268		}
269
270		/// The trace as text, one record per line.
271	11	pub fn render(records: &[Record]) -> String {
272	11	let mut out = String::new();
273	1.18k	for record in records11 {
274	1.18k	out.push_str(&record.to_string());
275	1.18k	out.push('\n');
276	1.18k	}
277	11	out
278	11	}