jetstreamer_plugin/
metrics.rs

1//! Global runtime metrics shared with frontends such as the CLI `--tui` mode.
2//!
3//! The plugin runner stamps per-thread activity as data flows through its handlers and
4//! records a structured snapshot of every stats pulse; a frontend polls these from its render
5//! loop instead of scraping log lines.
6
7use dashmap::DashMap;
8use once_cell::sync::Lazy;
9use std::sync::Mutex;
10use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
11use std::time::Instant;
12
13static ORIGIN: Lazy<Instant> = Lazy::new(Instant::now);
14static THREAD_COUNT: AtomicUsize = AtomicUsize::new(0);
15static THREAD_LAST_ACTIVITY_MS: Lazy<DashMap<usize, u64, ahash::RandomState>> =
16    Lazy::new(|| DashMap::with_hasher(ahash::RandomState::new()));
17static THREAD_TX_COUNTS: Lazy<DashMap<usize, u64, ahash::RandomState>> =
18    Lazy::new(|| DashMap::with_hasher(ahash::RandomState::new()));
19static LATEST_PULSE: Mutex<Option<PulseSnapshot>> = Mutex::new(None);
20static RUN_SLOT_RANGE: Mutex<Option<(u64, u64)>> = Mutex::new(None);
21static RESUME_COMMAND_TEMPLATE: Mutex<Option<String>> = Mutex::new(None);
22static DB_RETRIES: AtomicU64 = AtomicU64::new(0);
23
24/// Structured copy of the numbers a stats pulse logs, for frontends to render.
25#[derive(Clone, Debug, Default)]
26pub struct PulseSnapshot {
27    /// Overall progress in percent, clamped to `[0, 100]`.
28    pub progress_pct: f64,
29    /// Human-readable ETA, if computable.
30    pub eta: Option<String>,
31    /// Transactions per second measured between the last two pulses.
32    pub tps: f64,
33    /// Aggregate slots processed, capped to the run's total.
34    pub slots_processed: u64,
35    /// Aggregate blocks processed.
36    pub blocks_processed: u64,
37    /// Aggregate transactions processed.
38    pub transactions_processed: u64,
39    /// Aggregate entries processed.
40    pub entries_processed: u64,
41    /// Aggregate rewards processed.
42    pub rewards_processed: u64,
43    /// Total number of slots in the run's range.
44    pub total_slots: u64,
45    /// Seconds elapsed since the run started.
46    pub elapsed_secs: f64,
47}
48
49/// Milliseconds since metrics tracking began (a process-wide monotonic clock).
50pub fn now_ms() -> u64 {
51    ORIGIN.elapsed().as_millis() as u64
52}
53
54/// Prepares metrics for a new run with `thread_count` firehose threads.
55pub fn init(thread_count: usize) {
56    Lazy::force(&ORIGIN);
57    THREAD_COUNT.store(thread_count, Ordering::Relaxed);
58    THREAD_LAST_ACTIVITY_MS.clear();
59    THREAD_TX_COUNTS.clear();
60    *LATEST_PULSE.lock().unwrap() = None;
61    *RUN_SLOT_RANGE.lock().unwrap() = None;
62    DB_RETRIES.store(0, Ordering::Relaxed);
63}
64
65/// Records one retried ClickHouse write attempt.
66pub fn note_db_retry() {
67    DB_RETRIES.fetch_add(1, Ordering::Relaxed);
68}
69
70/// Total ClickHouse write retries this run.
71pub fn db_retry_count() -> u64 {
72    DB_RETRIES.load(Ordering::Relaxed)
73}
74
75/// Stores the process invocation with the range positional replaced by `{range}`, used to
76/// print an accurate resume command in fatal error messages. Set once at CLI parse time and
77/// deliberately not cleared by [`init`] (it describes the process, not the run).
78pub fn set_resume_command_template(template: String) {
79    *RESUME_COMMAND_TEMPLATE.lock().unwrap() = Some(template);
80}
81
82/// The resume command template recorded at CLI parse time, if any.
83pub fn resume_command_template() -> Option<String> {
84    RESUME_COMMAND_TEMPLATE.lock().unwrap().clone()
85}
86
87/// Records the half-open slot range `[start, end)` the current run covers.
88pub fn set_run_slot_range(start: u64, end: u64) {
89    *RUN_SLOT_RANGE.lock().unwrap() = Some((start, end));
90}
91
92/// The half-open slot range `[start, end)` of the current run, if one is active.
93pub fn run_slot_range() -> Option<(u64, u64)> {
94    *RUN_SLOT_RANGE.lock().unwrap()
95}
96
97/// Number of firehose threads in the current run.
98pub fn thread_count() -> usize {
99    THREAD_COUNT.load(Ordering::Relaxed)
100}
101
102/// Records that data flowed through `thread_id` just now.
103pub fn note_thread_activity(thread_id: usize) {
104    THREAD_LAST_ACTIVITY_MS.insert(thread_id, now_ms());
105}
106
107/// Records one processed transaction on `thread_id` (also stamps activity).
108pub fn note_thread_transaction(thread_id: usize) {
109    note_thread_activity(thread_id);
110    *THREAD_TX_COUNTS.entry(thread_id).or_insert(0) += 1;
111}
112
113/// Total transactions processed by `thread_id` so far.
114pub fn thread_tx_count(thread_id: usize) -> u64 {
115    THREAD_TX_COUNTS
116        .get(&thread_id)
117        .map(|count| *count)
118        .unwrap_or(0)
119}
120
121/// Milliseconds since data last flowed through `thread_id`, or `None` if the thread has not
122/// reported any data yet.
123pub fn thread_idle_ms(thread_id: usize) -> Option<u64> {
124    THREAD_LAST_ACTIVITY_MS
125        .get(&thread_id)
126        .map(|stamp| now_ms().saturating_sub(*stamp))
127}
128
129/// Stores the latest stats pulse.
130pub fn record_pulse(pulse: PulseSnapshot) {
131    *LATEST_PULSE.lock().unwrap() = Some(pulse);
132}
133
134/// Returns the most recent stats pulse, if any.
135pub fn latest_pulse() -> Option<PulseSnapshot> {
136    LATEST_PULSE.lock().unwrap().clone()
137}
jetstreamer_plugin/metrics.rs

jetstreamer_plugin/
metrics.rs
