ledgence_worker_api/
metrics.rs

1//! Closed-vocabulary operational observations, independent of any exporter SDK.
2//!
3//! The existing tracing dispatch carries these events to optional adapters. They
4//! are not logs, spans, durable accounting, or execution authority. Exporters must
5//! aggregate without doing I/O in the observation path.
6use std::time::Instant;
7
8pub const METRIC_TARGET: &str = "ledgence::metrics";
9
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11#[repr(u64)]
12pub enum Metric {
13    HttpDuration,
14    DatabaseDuration,
15    ExecutionDuration,
16    PreparationDuration,
17    CallbackDuration,
18    RecoveryDuration,
19    QueueAge,
20    CallbackAge,
21    CacheLookup,
22    ProcessSelection,
23    DatabaseRetry,
24    RecoveryExpired,
25    ConsumerSlots,
26    ExecutingPrograms,
27}
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub enum MetricKind {
30    Histogram,
31    Counter,
32    UpDownCounter,
33}
34impl Metric {
35    pub const ALL: [Self; 14] = [
36        Self::HttpDuration,
37        Self::DatabaseDuration,
38        Self::ExecutionDuration,
39        Self::PreparationDuration,
40        Self::CallbackDuration,
41        Self::RecoveryDuration,
42        Self::QueueAge,
43        Self::CallbackAge,
44        Self::CacheLookup,
45        Self::ProcessSelection,
46        Self::DatabaseRetry,
47        Self::RecoveryExpired,
48        Self::ConsumerSlots,
49        Self::ExecutingPrograms,
50    ];
51    pub fn from_id(id: u64) -> Option<Self> {
52        Self::ALL.get(usize::try_from(id).ok()?).copied()
53    }
54    pub fn name(self) -> &'static str {
55        match self {
56            Self::HttpDuration => "ledgence.http.request.duration",
57            Self::DatabaseDuration => "ledgence.database.operation.duration",
58            Self::ExecutionDuration => "ledgence.worker.execution.duration",
59            Self::PreparationDuration => "ledgence.worker.preparation.duration",
60            Self::CallbackDuration => "ledgence.completion.delivery.duration",
61            Self::RecoveryDuration => "ledgence.recovery.scan.duration",
62            Self::QueueAge => "ledgence.task.claim.queue_age",
63            Self::CallbackAge => "ledgence.completion.delivery.age",
64            Self::CacheLookup => "ledgence.worker.cache.lookup",
65            Self::ProcessSelection => "ledgence.worker.process.selection",
66            Self::DatabaseRetry => "ledgence.database.operation.retry",
67            Self::RecoveryExpired => "ledgence.recovery.expired",
68            Self::ConsumerSlots => "ledgence.worker.consumer.occupied",
69            Self::ExecutingPrograms => "ledgence.worker.execution.active",
70        }
71    }
72    pub fn kind(self) -> MetricKind {
73        match self {
74            Self::CacheLookup
75            | Self::ProcessSelection
76            | Self::DatabaseRetry
77            | Self::RecoveryExpired => MetricKind::Counter,
78            Self::ConsumerSlots | Self::ExecutingPrograms => MetricKind::UpDownCounter,
79            _ => MetricKind::Histogram,
80        }
81    }
82    pub fn unit(self) -> &'static str {
83        match self.kind() {
84            MetricKind::Histogram => "s",
85            _ => "1",
86        }
87    }
88    pub fn accepts(self, outcome: MetricOutcome) -> bool {
89        use MetricOutcome::*;
90        match self {
91            Self::HttpDuration => matches!(outcome, Ok | ClientError | ServerError | Cancelled),
92            Self::DatabaseDuration | Self::PreparationDuration | Self::RecoveryDuration => {
93                matches!(outcome, Ok | Failed | Cancelled)
94            }
95            Self::ExecutionDuration => matches!(outcome, Ok | Failed | RuntimeError | Cancelled),
96            Self::CallbackDuration => matches!(outcome, Ok | Retry | Cancelled),
97            Self::QueueAge => matches!(outcome, Integrated | External),
98            Self::CallbackAge
99            | Self::DatabaseRetry
100            | Self::RecoveryExpired
101            | Self::ConsumerSlots
102            | Self::ExecutingPrograms => outcome == None,
103            Self::CacheLookup => matches!(outcome, Hit | Miss),
104            Self::ProcessSelection => matches!(outcome, Reused | Started),
105        }
106    }
107    /// Only fixed enum dimensions are emitted. Negative values are valid only
108    /// for slot releases; nonfinite values and invalid combinations are ignored.
109    #[inline]
110    pub fn record(self, value: f64, outcome: MetricOutcome) {
111        if value.is_finite()
112            && (value >= 0.0 || self.kind() == MetricKind::UpDownCounter)
113            && self.accepts(outcome)
114        {
115            tracing::event!(target: "ledgence::metrics", tracing::Level::DEBUG, metric = self as u64, value, outcome = outcome as u64);
116        }
117    }
118}
119
120#[derive(Debug, Clone, Copy, PartialEq, Eq)]
121#[repr(u64)]
122pub enum MetricOutcome {
123    None,
124    Ok,
125    Failed,
126    RuntimeError,
127    Cancelled,
128    ClientError,
129    ServerError,
130    Hit,
131    Miss,
132    Reused,
133    Started,
134    Retry,
135    Integrated,
136    External,
137}
138impl MetricOutcome {
139    pub const ALL: [Self; 14] = [
140        Self::None,
141        Self::Ok,
142        Self::Failed,
143        Self::RuntimeError,
144        Self::Cancelled,
145        Self::ClientError,
146        Self::ServerError,
147        Self::Hit,
148        Self::Miss,
149        Self::Reused,
150        Self::Started,
151        Self::Retry,
152        Self::Integrated,
153        Self::External,
154    ];
155    pub fn from_id(id: u64) -> Option<Self> {
156        Self::ALL.get(usize::try_from(id).ok()?).copied()
157    }
158    pub fn name(self) -> &'static str {
159        match self {
160            Self::None => "none",
161            Self::Ok => "ok",
162            Self::Failed => "failed",
163            Self::RuntimeError => "runtime_error",
164            Self::Cancelled => "cancelled",
165            Self::ClientError => "client_error",
166            Self::ServerError => "server_error",
167            Self::Hit => "hit",
168            Self::Miss => "miss",
169            Self::Reused => "reused",
170            Self::Started => "started",
171            Self::Retry => "retry",
172            Self::Integrated => "integrated",
173            Self::External => "external",
174        }
175    }
176}
177
178/// Records cancellation when a future is dropped before an explicit finish.
179/// Captures the original dispatch so task migration/drop cannot split a series.
180pub struct MetricTimer {
181    metric: Metric,
182    started: Option<(Instant, tracing::Dispatch)>,
183    outcome: MetricOutcome,
184}
185impl MetricTimer {
186    pub fn start(metric: Metric) -> Self {
187        Self {
188            metric,
189            started: enabled().then(|| {
190                (
191                    Instant::now(),
192                    tracing::dispatcher::get_default(Clone::clone),
193                )
194            }),
195            outcome: MetricOutcome::Cancelled,
196        }
197    }
198    pub fn finish(mut self, outcome: MetricOutcome) {
199        self.outcome = outcome;
200    }
201}
202impl Drop for MetricTimer {
203    fn drop(&mut self) {
204        if let Some((started, dispatch)) = &self.started {
205            tracing::dispatcher::with_default(dispatch, || {
206                self.metric
207                    .record(started.elapsed().as_secs_f64(), self.outcome)
208            });
209        }
210    }
211}
212
213/// Own alongside the actual semaphore permit, including retained cleanup work.
214pub struct MetricGuard {
215    metric: Metric,
216    dispatch: Option<tracing::Dispatch>,
217}
218impl MetricGuard {
219    pub fn consumer() -> Self {
220        Self::new(Metric::ConsumerSlots)
221    }
222    pub fn execution() -> Self {
223        Self::new(Metric::ExecutingPrograms)
224    }
225    fn new(metric: Metric) -> Self {
226        let dispatch = enabled().then(|| tracing::dispatcher::get_default(Clone::clone));
227        if let Some(dispatch) = &dispatch {
228            tracing::dispatcher::with_default(dispatch, || metric.record(1.0, MetricOutcome::None));
229        }
230        Self { metric, dispatch }
231    }
232}
233impl Drop for MetricGuard {
234    fn drop(&mut self) {
235        if let Some(dispatch) = &self.dispatch {
236            tracing::dispatcher::with_default(dispatch, || {
237                self.metric.record(-1.0, MetricOutcome::None)
238            });
239        }
240    }
241}
242#[inline]
243fn enabled() -> bool {
244    tracing::enabled!(target: "ledgence::metrics", tracing::Level::DEBUG)
245}
ledgence_worker_api/metrics.rs

ledgence_worker_api/
metrics.rs
