fathomdb_engine/
lifecycle.rs

1//! Lifecycle observability data types.
2//!
3//! Pure data types and the subscriber boundary trait. Public type shape,
4//! phase semantics, diagnostic source/category taxonomy, counter snapshot
5//! key set, profile record shape, and stress-failure payload are owned by
6//! `dev/design/lifecycle.md`.
7
8use std::collections::BTreeMap;
9use std::sync::atomic::{AtomicU64, Ordering};
10use std::sync::{Arc, Mutex, Weak};
11
12use crate::CounterSnapshot;
13
14/// Lifecycle phase tag.
15///
16/// Five-value enum locked by AC-001 / AC-008 and `dev/design/lifecycle.md`
17/// § Phase enum.
18#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
19pub enum Phase {
20    Started,
21    Slow,
22    Heartbeat,
23    Finished,
24    Failed,
25}
26
27/// Origin of a structured diagnostic.
28///
29/// Pinned by `dev/design/lifecycle.md` § Diagnostic source and category.
30#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
31pub enum EventSource {
32    Engine,
33    SqliteInternal,
34}
35
36/// Stable diagnostic category.
37///
38/// `Writer`, `Search`, `Admin`, `Error` pair with `EventSource::Engine`.
39/// `Corruption`, `Recovery`, `Io` pair with `EventSource::SqliteInternal`.
40#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
41pub enum EventCategory {
42    Writer,
43    Search,
44    Admin,
45    Error,
46    Corruption,
47    Recovery,
48    Io,
49}
50
51/// Public lifecycle event payload.
52///
53/// The required public shape in 0.6.0 is the typed phase + source + category
54/// triple. Producing surfaces may attach additional structured operation
55/// identity or timing context per `dev/design/lifecycle.md` § Public event
56/// contract.
57///
58/// `code` carries a stable machine-readable identifier for events that
59/// represent a concrete failure: SQLite-internal events use the SQLite
60/// extended-code name (e.g. `"SQLITE_SCHEMA"`, `"SQLITE_BUSY"`); engine
61/// errors use the stable `EngineError::stable_code` value (e.g.
62/// `"StorageError"`, `"WriteValidationError"` — matching the binding
63/// matrix in `dev/design/errors.md`). Non-error events leave `code`
64/// `None`. AC-021 dispatches on `code` rather than counting all error
65/// events — without it the test cannot distinguish `SQLITE_SCHEMA` from
66/// any other engine error.
67#[derive(Debug, Clone)]
68pub struct Event {
69    pub phase: Phase,
70    pub source: EventSource,
71    pub category: EventCategory,
72    pub code: Option<&'static str>,
73}
74
75/// Host-routed subscriber boundary.
76///
77/// `dev/design/lifecycle.md` § Host-routed diagnostics requires that all
78/// engine and SQLite-internal diagnostics flow through the host's chosen
79/// subscriber. No private sink, no stderr fallback.
80///
81/// `on_profile` and `on_slow_statement` are default-no-op so existing
82/// subscribers compile unchanged. Their payload shapes are owned by
83/// `dev/design/lifecycle.md` § Per-statement profiling and § Slow and
84/// heartbeat policy. A statement that crosses the slow threshold emits
85/// both a [`SlowStatement`] signal here AND a `Phase::Slow` lifecycle
86/// event via `on_event`, per the design's two-correlated-facts contract.
87pub trait Subscriber: Send + Sync {
88    fn on_event(&self, event: &Event);
89
90    fn on_profile(&self, _record: &ProfileRecord) {}
91
92    fn on_slow_statement(&self, _signal: &SlowStatement) {}
93
94    fn on_stress_failure(&self, _context: &StressFailureContext) {}
95}
96
97/// Statement-level slow signal.
98///
99/// `dev/design/lifecycle.md` § Slow and heartbeat policy: when a
100/// statement crosses the configured threshold, "a slow signal must
101/// surface and identify the statement that crossed the threshold."
102/// `statement` is the SQL text of the slow statement (or a synthetic
103/// label for non-SQL operations such as the `search` / `write` outer
104/// envelope). `wall_clock_ms` is the measured duration.
105#[derive(Debug, Clone)]
106pub struct SlowStatement {
107    pub statement: String,
108    pub wall_clock_ms: u64,
109}
110
111/// Engine-side registry of attached subscribers.
112///
113/// Holds attached subscribers behind a `Mutex<Vec<...>>`. Dispatch fans
114/// the event out to every live subscriber. Drop of a [`Subscription`]
115/// detaches that subscriber by id.
116#[derive(Default)]
117pub(crate) struct SubscriberRegistry {
118    next_id: AtomicU64,
119    entries: Mutex<Vec<(u64, Arc<dyn Subscriber>)>>,
120}
121
122impl std::fmt::Debug for SubscriberRegistry {
123    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
124        let count = self.entries.lock().map(|e| e.len()).unwrap_or(0);
125        f.debug_struct("SubscriberRegistry").field("subscribers", &count).finish()
126    }
127}
128
129impl SubscriberRegistry {
130    pub(crate) fn new() -> Self {
131        Self::default()
132    }
133
134    pub(crate) fn attach(self: &Arc<Self>, subscriber: Arc<dyn Subscriber>) -> Subscription {
135        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
136        if let Ok(mut entries) = self.entries.lock() {
137            entries.push((id, subscriber));
138        }
139        Subscription { id, registry: Arc::downgrade(self) }
140    }
141
142    pub(crate) fn attach_persistent(&self, subscriber: Arc<dyn Subscriber>) {
143        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
144        if let Ok(mut entries) = self.entries.lock() {
145            entries.push((id, subscriber));
146        }
147    }
148
149    fn detach(&self, id: u64) {
150        if let Ok(mut entries) = self.entries.lock() {
151            entries.retain(|(eid, _)| *eid != id);
152        }
153    }
154
155    pub(crate) fn dispatch(&self, event: &Event) {
156        for sub in self.snapshot() {
157            sub.on_event(event);
158        }
159    }
160
161    pub(crate) fn dispatch_profile(&self, record: &ProfileRecord) {
162        for sub in self.snapshot() {
163            sub.on_profile(record);
164        }
165    }
166
167    pub(crate) fn dispatch_slow_statement(&self, signal: &SlowStatement) {
168        for sub in self.snapshot() {
169            sub.on_slow_statement(signal);
170        }
171    }
172
173    pub(crate) fn dispatch_stress_failure(&self, context: &StressFailureContext) {
174        for sub in self.snapshot() {
175            sub.on_stress_failure(context);
176        }
177    }
178
179    fn snapshot(&self) -> Vec<Arc<dyn Subscriber>> {
180        // Snapshot the subscriber list so callbacks may not call back into the
181        // registry while we hold the lock.
182        match self.entries.lock() {
183            Ok(entries) => entries.iter().map(|(_, s)| Arc::clone(s)).collect(),
184            Err(_) => Vec::new(),
185        }
186    }
187}
188
189/// Handle returned by `Engine::subscribe`.
190///
191/// Dropping the handle detaches the subscriber. Subscriber payload
192/// semantics are owned by `dev/design/lifecycle.md` and
193/// `dev/design/migrations.md`.
194#[derive(Debug)]
195pub struct Subscription {
196    id: u64,
197    registry: Weak<SubscriberRegistry>,
198}
199
200impl Drop for Subscription {
201    fn drop(&mut self) {
202        if let Some(registry) = self.registry.upgrade() {
203            registry.detach(self.id);
204        }
205    }
206}
207
208/// Per-statement profile record shape.
209///
210/// Field set locked by AC-005b / `dev/design/lifecycle.md` § Per-statement
211/// profiling. `cache_delta` is signed because cache counters can decrease
212/// across a statement window when SQLite evicts.
213#[derive(Debug, Clone, Copy, Eq, PartialEq)]
214pub struct ProfileRecord {
215    pub wall_clock_ms: u64,
216    pub step_count: u64,
217    pub cache_delta: i64,
218}
219
220/// Projection-status enum surfaced by the projection-status query.
221///
222/// Locked by AC-010.
223#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
224pub enum ProjectionStatus {
225    Pending,
226    Failed,
227    UpToDate,
228}
229
230/// Stress-failure context payload.
231///
232/// Required field set locked by AC-009 / REQ-007. The payload exists so
233/// stress / robustness failure events do not degrade into ad hoc free-text
234/// metadata; consumers must be able to reach all four fields without
235/// message parsing.
236#[derive(Debug, Clone)]
237pub struct StressFailureContext {
238    pub thread_group_id: u64,
239    pub op_kind: String,
240    pub last_error_chain: Vec<String>,
241    pub projection_state: String,
242}
243
244/// Internal cumulative counters backing [`CounterSnapshot`].
245///
246/// Public snapshot key set is owned by `dev/design/lifecycle.md` § Public
247/// key set. Snapshotting performs only atomic loads and a map clone — it
248/// must not perturb counters (AC-004c).
249#[derive(Debug)]
250pub(crate) struct Counters {
251    queries: AtomicU64,
252    writes: AtomicU64,
253    write_rows: AtomicU64,
254    admin_ops: AtomicU64,
255    cache_hit: AtomicU64,
256    cache_miss: AtomicU64,
257    errors_by_code: Mutex<BTreeMap<String, u64>>,
258}
259
260impl Counters {
261    pub(crate) fn new() -> Self {
262        Self {
263            queries: AtomicU64::new(0),
264            writes: AtomicU64::new(0),
265            write_rows: AtomicU64::new(0),
266            admin_ops: AtomicU64::new(0),
267            cache_hit: AtomicU64::new(0),
268            cache_miss: AtomicU64::new(0),
269            errors_by_code: Mutex::new(BTreeMap::new()),
270        }
271    }
272
273    pub(crate) fn record_write(&self, rows: u64) {
274        self.writes.fetch_add(1, Ordering::Relaxed);
275        self.write_rows.fetch_add(rows, Ordering::Relaxed);
276    }
277
278    pub(crate) fn record_query(&self) {
279        self.queries.fetch_add(1, Ordering::Relaxed);
280    }
281
282    pub(crate) fn record_admin(&self) {
283        self.admin_ops.fetch_add(1, Ordering::Relaxed);
284    }
285
286    pub(crate) fn record_error(&self, code: &str) {
287        if let Ok(mut map) = self.errors_by_code.lock() {
288            *map.entry(code.to_string()).or_insert(0) += 1;
289        }
290    }
291
292    #[allow(dead_code)]
293    pub(crate) fn record_cache_hit(&self) {
294        self.cache_hit.fetch_add(1, Ordering::Relaxed);
295    }
296
297    #[allow(dead_code)]
298    pub(crate) fn record_cache_miss(&self) {
299        self.cache_miss.fetch_add(1, Ordering::Relaxed);
300    }
301
302    pub(crate) fn snapshot(&self) -> CounterSnapshot {
303        // Treat poisoned lock as zero-error snapshot — prefer non-perturbing read over panic.
304        let errors_by_code = self.errors_by_code.lock().map(|map| map.clone()).unwrap_or_default();
305        CounterSnapshot {
306            queries: self.queries.load(Ordering::Relaxed),
307            writes: self.writes.load(Ordering::Relaxed),
308            write_rows: self.write_rows.load(Ordering::Relaxed),
309            errors_by_code,
310            admin_ops: self.admin_ops.load(Ordering::Relaxed),
311            cache_hit: self.cache_hit.load(Ordering::Relaxed),
312            cache_miss: self.cache_miss.load(Ordering::Relaxed),
313        }
314    }
315}
fathomdb_engine/lifecycle.rs

fathomdb_engine/
lifecycle.rs
