rushdb_storage/
observability.rs

1//! Live store observability hooks: per-commit latency, backend lock
2//! contention, and the conflict window's footprint (C0, cloud-capacity F12).
3
4use std::sync::Arc;
5
6use super::Store;
7
8/// One timed step of the commit pipeline. Committer-side phases are
9/// observed once per commit; publisher-side phases once per publish group.
10#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
11pub enum CommitPhase {
12    /// Waiting for the admission (sequencer) lock.
13    AdmissionWait,
14    /// Running the caller's planner/build closure.
15    Build,
16    /// Deterministic expansion into ops, including slot-reservation retries.
17    Expand,
18    /// Enqueue until the durable acknowledgement (the end-to-end latency the
19    /// client waits for after expansion).
20    Durable,
21    /// Publisher: encoding a group's WAL records.
22    WalEncode,
23    /// Publisher: appending and fsyncing a group's WAL records.
24    WalAppendSync,
25    /// Publisher: applying a group's ops to the backend materialization.
26    BackendApply,
27    /// Publisher: advancing derived state for a group.
28    Derive,
29    /// Publisher: conflict-window recording, watermark advance and acks.
30    Publish,
31}
32
33impl CommitPhase {
34    /// Every phase, in pipeline order.
35    pub const ALL: [Self; 9] = [
36        Self::AdmissionWait,
37        Self::Build,
38        Self::Expand,
39        Self::Durable,
40        Self::WalEncode,
41        Self::WalAppendSync,
42        Self::BackendApply,
43        Self::Derive,
44        Self::Publish,
45    ];
46
47    /// Stable metric label for this phase.
48    #[must_use]
49    pub const fn label(self) -> &'static str {
50        match self {
51            Self::AdmissionWait => "admission_wait",
52            Self::Build => "build",
53            Self::Expand => "expand",
54            Self::Durable => "durable",
55            Self::WalEncode => "wal_encode",
56            Self::WalAppendSync => "wal_append_sync",
57            Self::BackendApply => "backend_apply",
58            Self::Derive => "derive",
59            Self::Publish => "publish",
60        }
61    }
62}
63
64/// Callback receiving each timed commit phase, in seconds.
65pub type CommitObserver = dyn Fn(CommitPhase, f64) + Send + Sync;
66
67/// Cumulative backend database-lock timings, summed over every shard.
68#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
69pub struct BackendLockTimes {
70    /// Nanoseconds spent waiting to acquire the lock.
71    pub wait_ns: u64,
72    /// Nanoseconds the lock was held.
73    pub hold_ns: u64,
74    /// Longest single hold, in nanoseconds.
75    pub hold_max_ns: u64,
76}
77
78/// What holds a backend's single writer: one redb write transaction, from
79/// `begin_write` to its commit. Every other writer waits meanwhile, so the
80/// longest holds are what stall other tenants' commits.
81#[derive(Debug, Clone, Copy, PartialEq, Eq)]
82pub enum WriterHolder {
83    /// Publishing commit groups into the materialization.
84    Publish,
85    /// One batch of history garbage collection.
86    HistoryGc,
87    /// A durable checkpoint (fsync).
88    Checkpoint,
89    /// Writes paused for a consistent copy (backup).
90    Pause,
91}
92
93impl WriterHolder {
94    /// Every holder.
95    pub const ALL: [Self; 4] = [
96        Self::Publish,
97        Self::HistoryGc,
98        Self::Checkpoint,
99        Self::Pause,
100    ];
101
102    /// Stable metric label.
103    #[must_use]
104    pub const fn label(self) -> &'static str {
105        match self {
106            Self::Publish => "publish",
107            Self::HistoryGc => "history_gc",
108            Self::Checkpoint => "checkpoint",
109            Self::Pause => "pause",
110        }
111    }
112
113    pub(crate) const fn index(self) -> usize {
114        match self {
115            Self::Publish => 0,
116            Self::HistoryGc => 1,
117            Self::Checkpoint => 2,
118            Self::Pause => 3,
119        }
120    }
121}
122
123/// Upper bounds, in seconds, of the writer-hold histogram buckets.
124pub const WRITER_HOLD_BOUNDS: [f64; 12] = [
125    0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
126];
127
128/// One holder's writer holds since the store opened.
129#[derive(Debug, Clone, Copy, Default, PartialEq)]
130pub struct WriterHolds {
131    /// Holds per bucket of [`WRITER_HOLD_BOUNDS`] (not cumulative); the last
132    /// slot counts holds above every bound.
133    pub counts: [u64; WRITER_HOLD_BOUNDS.len() + 1],
134    /// Total seconds held.
135    pub sum_seconds: f64,
136}
137
138impl WriterHolds {
139    /// Fold `other` into `self`.
140    #[must_use]
141    pub fn plus(mut self, other: &Self) -> Self {
142        for (total, count) in self.counts.iter_mut().zip(other.counts) {
143            *total = total.saturating_add(count);
144        }
145        self.sum_seconds += other.sum_seconds;
146        self
147    }
148}
149
150impl Store {
151    /// Install the callback that receives every timed commit phase.
152    ///
153    /// Lock poisoning fences the store until reopen.
154    pub fn set_commit_observer(&self, observer: Arc<CommitObserver>) {
155        *self.health.cleanup(self.commit_observer.write()) = Some(observer);
156    }
157
158    /// Report the time since `started` for `phase` to the installed observer.
159    pub(crate) fn observe_phase(&self, phase: CommitPhase, started: std::time::Instant) {
160        let observer = self.health.cleanup(self.commit_observer.read()).clone();
161        if let Some(observer) = observer {
162            observer(phase, started.elapsed().as_secs_f64());
163        }
164    }
165
166    /// Backend lock timings summed over all shards (live, no scan).
167    #[must_use]
168    pub fn backend_lock_times(&self) -> BackendLockTimes {
169        self.all_backends()
170            .iter()
171            .map(|backend| backend.lock_times())
172            .fold(BackendLockTimes::default(), |total, times| {
173                BackendLockTimes {
174                    wait_ns: total.wait_ns.saturating_add(times.wait_ns),
175                    hold_ns: total.hold_ns.saturating_add(times.hold_ns),
176                    hold_max_ns: total.hold_max_ns.max(times.hold_max_ns),
177                }
178            })
179    }
180
181    /// Writer holds by holder, summed over every shard (live, no scan).
182    #[must_use]
183    pub fn writer_holds(&self) -> Vec<(WriterHolder, WriterHolds)> {
184        let mut totals = [WriterHolds::default(); WriterHolder::ALL.len()];
185        for backend in self.all_backends() {
186            for (holder, holds) in backend.writer_holds() {
187                totals[holder.index()] = totals[holder.index()].plus(&holds);
188            }
189        }
190        WriterHolder::ALL.into_iter().zip(totals).collect()
191    }
192
193    /// Copy-on-write state of a durable store's data files, recorded when it
194    /// opened; `None` in memory.
195    #[must_use]
196    pub fn copy_on_write(&self) -> Option<crate::CopyOnWrite> {
197        self.health.cleanup(self.copy_on_write.lock()).clone()
198    }
199
200    /// Directory holding a durable store's files, or `None` in memory.
201    #[must_use]
202    pub fn data_dir(&self) -> Option<std::path::PathBuf> {
203        self.health
204            .cleanup(self.data_path.lock())
205            .as_deref()
206            .and_then(std::path::Path::parent)
207            .map(std::path::Path::to_path_buf)
208    }
209
210    /// Bytes currently retained by the in-memory conflict window.
211    #[must_use]
212    pub fn conflict_window_bytes(&self) -> usize {
213        self.health.cleanup(self.conflicts.lock()).bytes()
214    }
215}
rushdb_storage/observability.rs

rushdb_storage/
observability.rs
