reddb_server/runtime/
claim_telemetry.rs

1//! Concurrent claim telemetry.
2//!
3//! Process-local counters for `UPDATE ... CLAIM` observability. Labels are
4//! bounded to `(collection, model)`; no predicate values, ids, consumers, or
5//! query text enter metric cardinality.
6
7use std::collections::BTreeMap;
8use std::sync::atomic::{AtomicU64, Ordering};
9use std::sync::Mutex;
10
11#[derive(Debug, Default)]
12struct CounterCell {
13    value: AtomicU64,
14}
15
16/// Materialised snapshot returned to `/metrics` and tests.
17#[derive(Debug, Clone, Default)]
18pub struct ClaimTelemetrySnapshot {
19    pub attempts: Vec<((String, String), u64)>,
20    pub successful: Vec<((String, String), u64)>,
21    pub misses: Vec<((String, String), u64)>,
22    pub skipped_locked: Vec<((String, String), u64)>,
23}
24
25#[derive(Debug, Default)]
26pub(crate) struct ClaimTelemetryCounters {
27    attempts: Mutex<BTreeMap<(String, String), CounterCell>>,
28    successful: Mutex<BTreeMap<(String, String), CounterCell>>,
29    misses: Mutex<BTreeMap<(String, String), CounterCell>>,
30    skipped_locked: Mutex<BTreeMap<(String, String), CounterCell>>,
31}
32
33impl ClaimTelemetryCounters {
34    pub(crate) fn record_attempt(&self, collection: &str, model: &str) {
35        increment(&self.attempts, collection, model, 1);
36    }
37
38    pub(crate) fn record_successful(&self, collection: &str, model: &str, count: u64) {
39        increment(&self.successful, collection, model, count);
40    }
41
42    pub(crate) fn record_miss(&self, collection: &str, model: &str) {
43        increment(&self.misses, collection, model, 1);
44    }
45
46    pub(crate) fn record_skipped_locked(&self, collection: &str, model: &str, count: u64) {
47        increment(&self.skipped_locked, collection, model, count);
48    }
49
50    pub(crate) fn snapshot(&self) -> ClaimTelemetrySnapshot {
51        ClaimTelemetrySnapshot {
52            attempts: snapshot_counter(&self.attempts),
53            successful: snapshot_counter(&self.successful),
54            misses: snapshot_counter(&self.misses),
55            skipped_locked: snapshot_counter(&self.skipped_locked),
56        }
57    }
58}
59
60fn increment(
61    counter: &Mutex<BTreeMap<(String, String), CounterCell>>,
62    collection: &str,
63    model: &str,
64    count: u64,
65) {
66    if count == 0 {
67        return;
68    }
69    let key = (collection.to_string(), model.to_string());
70    let mut map = counter.lock().unwrap_or_else(|p| p.into_inner());
71    map.entry(key)
72        .or_default()
73        .value
74        .fetch_add(count, Ordering::Relaxed);
75}
76
77fn snapshot_counter(
78    counter: &Mutex<BTreeMap<(String, String), CounterCell>>,
79) -> Vec<((String, String), u64)> {
80    let map = counter.lock().unwrap_or_else(|p| p.into_inner());
81    map.iter()
82        .map(|(k, v)| (k.clone(), v.value.load(Ordering::Relaxed)))
83        .collect()
84}
reddb_server/runtime/claim_telemetry.rs

reddb_server/runtime/
claim_telemetry.rs
