linkrs_metrics/manager/
sync.rs

1//! Sync pipeline observability: outbox, generations, transport.
2use super::core::StatsManager;
3use super::metric_type::MetricType;
4
5/// Statistics Manager
6///
7/// Unified management of query metrics, query profiling and error statistics.
8#[derive(Debug, Clone, Copy)]
9pub struct OutboxState {
10    pub pending: u64,
11    pub retries: u64,
12    pub dead_lettered: u64,
13    pub leased: u64,
14    pub oldest_event_age_ms: u64,
15    pub frontier_lag: u64,
16    pub degraded: bool,
17}
18
19impl StatsManager {
20    pub fn record_sync_operation(&self, latency_ms: u64, success: bool) {
21        self.add_value(MetricType::SyncOperations);
22        self.add_value_with_amount(MetricType::SyncLatencyMs, latency_ms);
23        if !success {
24            self.add_value(MetricType::SyncErrors);
25        }
26    }
27
28    pub fn record_sync_error(&self) {
29        self.add_value(MetricType::SyncErrors);
30    }
31
32    pub fn set_sync_queue_depth(&self, depth: u64) {
33        self.set_value(MetricType::SyncQueueDepth, depth);
34    }
35
36    pub fn record_outbox_state(&self, state: OutboxState) {
37        self.set_value(MetricType::OutboxPending, state.pending);
38        self.set_value(MetricType::OutboxRetryCount, state.retries);
39        self.set_value(MetricType::OutboxDeadLetterCount, state.dead_lettered);
40        self.set_value(MetricType::OutboxLeasedCount, state.leased);
41        self.set_value(
42            MetricType::OutboxOldestEventAgeMs,
43            state.oldest_event_age_ms,
44        );
45        self.set_value(MetricType::OutboxFrontierLag, state.frontier_lag);
46        self.set_value(MetricType::OutboxDegraded, u64::from(state.degraded));
47    }
48
49    pub fn record_target_frontier_lag(&self, target: &str, lag: u64) {
50        // Per-target lag uses the `target:{name}` namespace so it never
51        // collides with `space_{id}` buckets. Both global and per-target
52        // values are point-in-time gauges and use overwrite semantics.
53        self.set_value(MetricType::TargetFrontierLag, lag);
54        self.set_space_metric_with_amount(
55            &Self::target_key(target),
56            MetricType::TargetFrontierLag,
57            lag,
58        );
59    }
60
61    pub fn record_generation_build(&self) {
62        self.add_value(MetricType::GenerationBuildCount);
63    }
64
65    pub fn record_generation_publish(&self) {
66        self.add_value(MetricType::GenerationPublishCount);
67    }
68
69    pub fn record_generation_rebuild_failure(&self) {
70        self.add_value(MetricType::GenerationRebuildFailures);
71    }
72
73    pub fn record_inconsistent_index_count(&self, count: u64) {
74        self.set_value(MetricType::InconsistentIndexCount, count);
75    }
76
77    pub fn record_rebuild_phase_latency(&self, target: &str, phase: &str, latency_ms: u64) {
78        // `rebuild:{target}:{phase}` namespace plus an ops count so means
79        // are computable.
80        self.add_value(MetricType::RebuildPhaseOps);
81        self.add_value_with_amount(MetricType::RebuildPhaseLatencyMs, latency_ms);
82        let key = Self::rebuild_key(target, phase);
83        self.add_space_metric(&key, MetricType::RebuildPhaseOps);
84        self.add_space_metric_with_amount(&key, MetricType::RebuildPhaseLatencyMs, latency_ms);
85    }
86
87    pub fn record_split(&self, success: bool) {
88        if success {
89            self.add_value(MetricType::SplitCount);
90        } else {
91            self.add_value(MetricType::SplitFailures);
92        }
93    }
94
95    pub fn record_reclaimed_index_files(&self, count: u64) {
96        self.add_value_with_amount(MetricType::ReclaimedIndexFiles, count);
97    }
98
99    pub fn set_manifest_state(&self, active_readers: u64, retired_generations: u64) {
100        self.set_value(MetricType::ManifestActiveReaders, active_readers);
101        self.set_value(MetricType::ManifestRetiredGenerations, retired_generations);
102    }
103
104    pub fn record_fence_failure(&self) {
105        self.add_value(MetricType::FenceFailures);
106    }
107
108    pub fn record_transport_latency(&self, latency_ms: u64) {
109        // Sum plus count so the mean is computable.
110        self.add_value(MetricType::TransportOps);
111        self.add_value_with_amount(MetricType::TransportLatencyMs, latency_ms);
112    }
113
114    pub fn record_materializer_latency(&self, latency_ms: u64) {
115        self.add_value(MetricType::MaterializerOps);
116        self.add_value_with_amount(MetricType::MaterializerLatencyMs, latency_ms);
117    }
118
119    pub fn set_snapshot_lag(&self, lag: u64) {
120        self.set_value(MetricType::SnapshotLag, lag);
121    }
122
123    // Index scan/write timing is covered by `record_index_operation`
124    // (per-space isolated); the global-only variants are removed to avoid
125    // implying a second uncovered path.
126}
linkrs_metrics/manager/sync.rs

linkrs_metrics/manager/
sync.rs
