hotshot_new_protocol/
leader_trace.rs

1//! Optional per-view tracing of the leader's duty.
2//!
3//! With no tracer registered each event site costs one `Option` check. Timestamps are
4//! wall-clock unix-epoch ns, the same clock as the bench's `MetricsCollector`, so the streams join on `view + ts_ns`.
5
6use std::{
7    fs::{self, File, OpenOptions},
8    io::{self, BufWriter, Write},
9    path::Path,
10    sync::{
11        Arc, Weak,
12        atomic::{AtomicBool, Ordering},
13    },
14    thread,
15    time::Duration,
16};
17
18use parking_lot::Mutex;
19use time::OffsetDateTime;
20use tracing::warn;
21
22pub type LeaderTracerHandle = Arc<dyn LeaderTracer>;
23
24pub trait LeaderTracer: Send + Sync + 'static {
25    fn record(&self, view: u64, event: LeaderEvent, ts_ns: i128);
26}
27
28/// Per-view events in rough order. Most fire on every node; the proposal and block builder events
29/// fire on the leader of the view only.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub enum LeaderEvent {
32    // V-1 events that trigger V duty.
33    ProposalValidatedVMinus1,
34    RequestBlockHeaderQueued,
35
36    // Block builder, per request.
37    /// The block builder accepts a request and spawns its task, before the empty-block delay.
38    BlockRequestStarted,
39    /// The block builder task is past the empty-block delay, before it builds the payload.
40    BlockPayloadStart,
41    /// The block builder starts `block::block_commitments`: payload bytes, then the VID encode in
42    /// parallel with transaction hashes and builder commitment. Built path only.
43    BlockCommitmentsStart,
44    /// `block_commitments` returned, `Ok` or not. Same span as `block_commitments_duration`.
45    BlockCommitmentsEnd,
46
47    // Coordinator applies the header and block.
48    HeaderCreatedApplied,
49    BlockBuiltApplied,
50
51    // Disperse the block and unicast the shares. `NsDisperse*` when the disperser encodes
52    // (`Dispersal::Encode`), `VidFragments*` when the block builder did (`Dispersal::Built`).
53    /// The disperser starts encoding the block. Encode path only.
54    NsDisperseStart,
55    /// The last bucket is encoded and its fragments assembled. Encode path only.
56    NsDisperseEnd,
57    /// The disperser starts on the block builder's shares: sign, bucket, assemble. Built path only.
58    VidFragmentsStart,
59    /// The last bucket's fragments are assembled. Built path only.
60    VidFragmentsEnd,
61    VidSharesUnicastStart,
62    VidSharesUnicastEnd,
63
64    // Replica work for V-1, in parallel.
65    Vote1VMinus1Arrived,
66    ThresholdShareReachedVMinus1,
67    /// Stamped on every attempt; with the first stamp kept, the recover span includes failed
68    /// attempts and the retry.
69    RecoverVMinus1Start,
70    /// Erasure decoding done; the rest until `RecoverVMinus1End` is the
71    /// single-threaded `from_bytes` and transaction commitments.
72    RecoverVMinus1DecodeEnd,
73    RecoverVMinus1End,
74
75    // cert1[V-1] formation; gates V's proposal.
76    /// Consensus first stores cert1 of V-1, however it was formed or received.
77    Cert1VMinus1Obtained,
78    Vote2VMinus1Queued,
79    // cert2 formation and finality.
80    /// Consensus first stores cert2 of V-1, however it was formed or received.
81    Cert2VMinus1Obtained,
82    LeafDecided,
83
84    // Build and sign V's proposal.
85    /// The proposal is assembled, before its commit is computed and signed.
86    ProposalAssembled,
87    ProposalQueued,
88
89    // Network sends of V's proposal and V-1 votes and certs.
90    ProposalBroadcastStart,
91    ProposalBroadcastEnd,
92    Vote2VMinus1BroadcastStart,
93    Vote2VMinus1BroadcastEnd,
94    Cert1VMinus1BroadcastStart,
95    Cert1VMinus1BroadcastEnd,
96    Vote1VMinus1BroadcastStart,
97    Vote1VMinus1BroadcastEnd,
98}
99
100impl LeaderEvent {
101    /// CSV event name. Downstream tools match on it; do not rename.
102    pub fn name(self) -> &'static str {
103        use LeaderEvent::*;
104        match self {
105            ProposalValidatedVMinus1 => "proposal_validated_v_minus_1",
106            RequestBlockHeaderQueued => "request_block_header_queued",
107            BlockRequestStarted => "block_request_started",
108            BlockPayloadStart => "block_payload_start",
109            BlockCommitmentsStart => "block_commitments_start",
110            BlockCommitmentsEnd => "block_commitments_end",
111            HeaderCreatedApplied => "header_created_applied",
112            BlockBuiltApplied => "block_built_applied",
113            NsDisperseStart => "ns_disperse_start",
114            NsDisperseEnd => "ns_disperse_end",
115            VidFragmentsStart => "vid_fragments_start",
116            VidFragmentsEnd => "vid_fragments_end",
117            VidSharesUnicastStart => "vid_shares_unicast_start",
118            VidSharesUnicastEnd => "vid_shares_unicast_end",
119            Vote1VMinus1Arrived => "vote1_v_minus_1_arrived",
120            ThresholdShareReachedVMinus1 => "threshold_share_reached_v_minus_1",
121            RecoverVMinus1Start => "recover_v_minus_1_start",
122            RecoverVMinus1DecodeEnd => "recover_v_minus_1_decode_end",
123            RecoverVMinus1End => "recover_v_minus_1_end",
124            Cert1VMinus1Obtained => "cert1_v_minus_1_obtained",
125            Vote2VMinus1Queued => "vote2_v_minus_1_queued",
126            Cert2VMinus1Obtained => "cert2_v_minus_1_obtained",
127            LeafDecided => "leaf_decided",
128            ProposalAssembled => "proposal_assembled",
129            ProposalQueued => "proposal_queued",
130            ProposalBroadcastStart => "proposal_broadcast_start",
131            ProposalBroadcastEnd => "proposal_broadcast_end",
132            Vote2VMinus1BroadcastStart => "vote2_v_minus_1_broadcast_start",
133            Vote2VMinus1BroadcastEnd => "vote2_v_minus_1_broadcast_end",
134            Cert1VMinus1BroadcastStart => "cert1_v_minus_1_broadcast_start",
135            Cert1VMinus1BroadcastEnd => "cert1_v_minus_1_broadcast_end",
136            Vote1VMinus1BroadcastStart => "vote1_v_minus_1_broadcast_start",
137            Vote1VMinus1BroadcastEnd => "vote1_v_minus_1_broadcast_end",
138        }
139    }
140}
141
142#[inline(always)]
143pub fn now_ns() -> i128 {
144    OffsetDateTime::now_utc().unix_timestamp_nanos()
145}
146
147const FLUSH_INTERVAL: Duration = Duration::from_secs(1);
148
149/// `LeaderTracer` appending `view,node_id,event,ts_ns` rows to a CSV file.
150///
151/// A background thread flushes every second, so a SIGKILL loses at most about 1s
152/// of rows. Appends to an existing file. The first I/O error disables the tracer.
153pub struct CsvLeaderTracer {
154    node_id: u64,
155    shared: Arc<Shared>,
156}
157
158struct Shared {
159    writer: Mutex<BufWriter<File>>,
160    failed: AtomicBool,
161}
162
163impl Shared {
164    fn fail(&self, err: io::Error) {
165        if !self.failed.swap(true, Ordering::Relaxed) {
166            warn!(%err, "leader trace write failed, disabling leader trace");
167        }
168    }
169
170    fn flush(&self) {
171        if self.failed.load(Ordering::Relaxed) {
172            return;
173        }
174        let result = self.writer.lock().flush();
175        if let Err(err) = result {
176            self.fail(err);
177        }
178    }
179}
180
181impl Drop for CsvLeaderTracer {
182    fn drop(&mut self) {
183        self.shared.flush();
184    }
185}
186
187impl CsvLeaderTracer {
188    pub fn new(node_id: u64, path: impl AsRef<Path>) -> io::Result<Self> {
189        let path = path.as_ref();
190        if let Some(parent) = path.parent() {
191            fs::create_dir_all(parent)?;
192        }
193        let mut file = OpenOptions::new().create(true).append(true).open(path)?;
194        if file.metadata()?.len() == 0 {
195            file.write_all(b"view,node_id,event,ts_ns\n")?;
196        }
197        let shared = Arc::new(Shared {
198            writer: Mutex::new(BufWriter::new(file)),
199            failed: AtomicBool::new(false),
200        });
201        spawn_flusher(Arc::downgrade(&shared))?;
202        Ok(Self { node_id, shared })
203    }
204}
205
206fn spawn_flusher(shared: Weak<Shared>) -> io::Result<()> {
207    thread::Builder::new()
208        .name("leader-trace-flush".into())
209        .spawn(move || {
210            loop {
211                thread::sleep(FLUSH_INTERVAL);
212                let Some(shared) = shared.upgrade() else {
213                    return;
214                };
215                shared.flush();
216            }
217        })?;
218    Ok(())
219}
220
221impl LeaderTracer for CsvLeaderTracer {
222    fn record(&self, view: u64, event: LeaderEvent, ts_ns: i128) {
223        if self.shared.failed.load(Ordering::Relaxed) {
224            return;
225        }
226        let row = format!("{view},{},{},{ts_ns}\n", self.node_id, event.name());
227        let result = self.shared.writer.lock().write_all(row.as_bytes());
228        if let Err(err) = result {
229            self.shared.fail(err);
230        }
231    }
232}
233
234pub trait AsViewU64 {
235    fn as_view_u64(&self) -> u64;
236}
237
238impl AsViewU64 for hotshot_types::data::ViewNumber {
239    fn as_view_u64(&self) -> u64 {
240        **self
241    }
242}
243
244impl AsViewU64 for u64 {
245    fn as_view_u64(&self) -> u64 {
246        *self
247    }
248}
249
250/// Records `$event` for `$view` when `$tracer` is `Some`.
251#[macro_export]
252macro_rules! trace_leader_event {
253    ($tracer:expr, $view:expr, $event:expr) => {
254        if let ::core::option::Option::Some(ref t) = $tracer {
255            let v = $crate::leader_trace::AsViewU64::as_view_u64(&$view);
256            t.record(v, $event, $crate::leader_trace::now_ns());
257        }
258    };
259}
260
261#[cfg(test)]
262mod tests {
263    use std::time::Instant;
264
265    use super::*;
266
267    const HEADER: &str = "view,node_id,event,ts_ns\n";
268
269    #[test]
270    fn row_reaches_disk_without_flush_or_drop() {
271        let dir = tempfile::tempdir().unwrap();
272        let path = dir.path().join("sub/trace.csv");
273        let tracer = CsvLeaderTracer::new(7, &path).unwrap();
274        tracer.record(1, LeaderEvent::NsDisperseStart, 100);
275        let expected = format!("{HEADER}1,7,ns_disperse_start,100\n");
276        let deadline = Instant::now() + Duration::from_secs(10);
277        while fs::read_to_string(&path).unwrap() != expected {
278            assert!(Instant::now() < deadline, "row not flushed within 10s");
279            thread::sleep(Duration::from_millis(20));
280        }
281    }
282
283    #[test]
284    fn reopen_appends_without_second_header() {
285        let dir = tempfile::tempdir().unwrap();
286        let path = dir.path().join("trace.csv");
287        let first = CsvLeaderTracer::new(0, &path).unwrap();
288        first.record(1, LeaderEvent::NsDisperseStart, 1);
289        drop(first);
290        let second = CsvLeaderTracer::new(0, &path).unwrap();
291        second.record(2, LeaderEvent::VidSharesUnicastEnd, 2);
292        drop(second);
293        assert_eq!(
294            fs::read_to_string(&path).unwrap(),
295            format!("{HEADER}1,0,ns_disperse_start,1\n2,0,vid_shares_unicast_end,2\n")
296        );
297    }
298}
hotshot_new_protocol/leader_trace.rs

hotshot_new_protocol/
leader_trace.rs
