khive_db/pool/reader_counters.rs
1//! Reader acquisition, timeout, and completed-hold accounting.
2
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::time::Duration;
5
6#[cfg(doc)]
7use super::{ConnectionPool, ReaderGuard, WriterAcquisitionSnapshot};
8
9/// Why a caller is allowed to bypass the pooled reader queue.
10///
11/// This is deliberately a closed, crate-private list. Ordinary request reads
12/// have no variant: they must use [`ConnectionPool::reader`] and surface its
13/// bounded admission timeout without falling back to a fresh connection.
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15#[allow(dead_code)] // The closed ADR list includes infrastructure paths not instantiated today.
16pub(crate) enum StandaloneReaderPurpose {
17 /// ADR-005/ADR-091's explicit, multi-call deferred raw-SQL transaction.
18 ExplicitSqlReadTransaction,
19 /// Boot-time schema/model-registry inspection before a runtime pool can
20 /// own the read. Kept separate from request traffic in diagnostics.
21 BootSchemaProbe,
22 /// An operator diagnostic that requires a physically independent
23 /// snapshot. Kept separate from request traffic in diagnostics.
24 DiagnosticsIndependentSnapshot,
25}
26
27impl StandaloneReaderPurpose {
28 fn is_infrastructure(self) -> bool {
29 matches!(
30 self,
31 Self::BootSchemaProbe | Self::DiagnosticsIndependentSnapshot
32 )
33 }
34}
35
36/// Process-local reader acquisition and hold lifecycle since one
37/// [`ConnectionPool`] was constructed.
38///
39/// Counters are monotonic and reset only when the pool is reconstructed.
40/// `active_pooled_checkouts` is the point-in-time number of live
41/// [`ReaderGuard`] values. Hold duration includes connection reset or
42/// replacement on return, because the slot is not reusable before then.
43#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
44pub struct ReaderAcquisitionSnapshot {
45 /// Configured process-local admission budget shared by pooled readers and
46 /// the explicit raw-SQL read-transaction exception.
47 pub reader_admission_capacity: usize,
48 /// Point-in-time permits not held by either pooled readers or explicit
49 /// raw-SQL read transactions.
50 pub available_reader_admission_slots: usize,
51 /// Successful request-path reader acquisitions (pooled plus the explicit
52 /// raw-SQL transaction exception; infrastructure opens excluded).
53 pub acquisitions: u64,
54 /// Successful bounded pooled-reader checkouts.
55 pub pooled_checkouts: u64,
56 /// Successful request-path standalone opens from the closed exception
57 /// list (currently explicit raw-SQL deferred transactions only).
58 pub standalone_opens: u64,
59 /// Successful boot/diagnostic standalone opens, attributed separately so
60 /// request traffic cannot be inferred from infrastructure activity.
61 pub infrastructure_standalone_opens: u64,
62 /// Reader-admission waits that exhausted `checkout_timeout` before any
63 /// query began. Covers pooled checkout and the closed raw-SQL exception;
64 /// cooperative request cancellation is intentionally excluded.
65 pub checkout_timeouts: u64,
66 /// Queries on a checked-out pooled reader that SQLite refused with
67 /// `SQLITE_BUSY` after the connection's busy handler gave up (typed-store
68 /// reads, pooled raw-SQL reads, and [`ReaderGuard::query_row`]). Counted
69 /// after checkout succeeded, so it never overlaps `checkout_timeouts`;
70 /// `SQLITE_LOCKED` and cooperative cancellation are excluded. Writer
71 /// refusals are not counted here; the writer task's are in
72 /// [`WriterAcquisitionSnapshot::writer_task_begin_busy`].
73 pub busy_timeouts: u64,
74 /// Pooled checkouts live at the instant this snapshot was taken.
75 pub active_pooled_checkouts: u64,
76 /// High-water mark of concurrent pooled checkouts.
77 pub peak_active_pooled_checkouts: u64,
78 /// Pooled guards that completed their full return/reset lifecycle.
79 pub completed_pooled_checkouts: u64,
80 /// Longest completed checkout hold, including return/reset, in
81 /// microseconds. Diagnostic evidence only; never a test timing gate.
82 pub max_completed_hold_micros: u64,
83 /// The typed-store operation that held the checkout reported in
84 /// `max_completed_hold_micros`. `None` when that hold came from a route
85 /// that carries no operation name, which is itself the answer rather
86 /// than a missing reading (#2793).
87 pub max_completed_hold_operation: Option<&'static str>,
88 /// Dedicated reader connections discarded on return for age, checkout
89 /// count, failed cleanup/health checks, or an explicitly non-reusable lease.
90 /// Counts replacement attempts, including opens that fail; not pool shutdown.
91 pub reader_discards: u64,
92 /// A discarded pooled-reader return (expiry or reset/pristine-check failure)
93 /// whose replacement connection then also failed to open, permanently
94 /// shrinking the physical pool by one slot below `max_readers`. Logged at
95 /// `warn` when it happens; this counter makes the shrink observable in a
96 /// snapshot too, since the pool itself never re-grows on its own.
97 pub reader_replacement_open_failures: u64,
98}
99
100/// The longest completed pooled-reader hold and the operation that held it,
101/// kept under one lock so a snapshot cannot pair one checkout's duration with
102/// another's name.
103#[derive(Debug, Default, Clone, Copy)]
104struct LongestCompletedHold {
105 micros: u64,
106 operation: Option<&'static str>,
107}
108
109#[derive(Debug, Default)]
110pub(super) struct ReaderAcquisitionCounters {
111 pooled_checkouts: AtomicU64,
112 standalone_opens: AtomicU64,
113 infrastructure_standalone_opens: AtomicU64,
114 checkout_timeouts: AtomicU64,
115 busy_timeouts: AtomicU64,
116 active_pooled_checkouts: AtomicU64,
117 peak_active_pooled_checkouts: AtomicU64,
118 completed_pooled_checkouts: AtomicU64,
119 longest_completed_hold: parking_lot::Mutex<LongestCompletedHold>,
120 pub(super) reader_discards: AtomicU64,
121 reader_replacement_open_failures: AtomicU64,
122}
123
124impl ReaderAcquisitionCounters {
125 pub(super) fn record_pooled_checkout(&self) {
126 self.pooled_checkouts.fetch_add(1, Ordering::Relaxed);
127 let active = self
128 .active_pooled_checkouts
129 .fetch_add(1, Ordering::Relaxed)
130 .saturating_add(1);
131 self.peak_active_pooled_checkouts
132 .fetch_max(active, Ordering::Relaxed);
133 }
134
135 pub(super) fn record_checkout_timeout(&self) {
136 self.checkout_timeouts.fetch_add(1, Ordering::Relaxed);
137 }
138
139 pub(super) fn record_busy_timeout(&self) {
140 self.busy_timeouts.fetch_add(1, Ordering::Relaxed);
141 }
142
143 pub(super) fn record_reader_replacement_open_failure(&self) {
144 self.reader_replacement_open_failures
145 .fetch_add(1, Ordering::Relaxed);
146 }
147
148 pub(super) fn record_standalone_open(&self, purpose: StandaloneReaderPurpose) {
149 if purpose.is_infrastructure() {
150 self.infrastructure_standalone_opens
151 .fetch_add(1, Ordering::Relaxed);
152 } else {
153 self.standalone_opens.fetch_add(1, Ordering::Relaxed);
154 }
155 }
156
157 pub(super) fn record_checkout_completed(
158 &self,
159 hold: Duration,
160 operation: Option<&'static str>,
161 ) {
162 let previous = self.active_pooled_checkouts.fetch_sub(1, Ordering::Relaxed);
163 debug_assert!(previous > 0, "reader active-checkout counter underflow");
164 self.completed_pooled_checkouts
165 .fetch_add(1, Ordering::Relaxed);
166 let micros = u64::try_from(hold.as_micros()).unwrap_or(u64::MAX);
167 // The maximum and the name of what held it are one reading: taken
168 // apart they can report a duration from one checkout beside a label
169 // from another, which is worse than no label at all.
170 let mut longest = self.longest_completed_hold.lock();
171 if micros > longest.micros {
172 longest.micros = micros;
173 longest.operation = operation;
174 }
175 }
176
177 pub(super) fn snapshot(
178 &self,
179 reader_admission_capacity: usize,
180 available_reader_admission_slots: usize,
181 ) -> ReaderAcquisitionSnapshot {
182 let pooled_checkouts = self.pooled_checkouts.load(Ordering::Relaxed);
183 let standalone_opens = self.standalone_opens.load(Ordering::Relaxed);
184 let longest_completed_hold = *self.longest_completed_hold.lock();
185 ReaderAcquisitionSnapshot {
186 reader_admission_capacity,
187 available_reader_admission_slots,
188 acquisitions: pooled_checkouts.saturating_add(standalone_opens),
189 pooled_checkouts,
190 standalone_opens,
191 infrastructure_standalone_opens: self
192 .infrastructure_standalone_opens
193 .load(Ordering::Relaxed),
194 checkout_timeouts: self.checkout_timeouts.load(Ordering::Relaxed),
195 busy_timeouts: self.busy_timeouts.load(Ordering::Relaxed),
196 active_pooled_checkouts: self.active_pooled_checkouts.load(Ordering::Relaxed),
197 peak_active_pooled_checkouts: self.peak_active_pooled_checkouts.load(Ordering::Relaxed),
198 completed_pooled_checkouts: self.completed_pooled_checkouts.load(Ordering::Relaxed),
199 max_completed_hold_micros: longest_completed_hold.micros,
200 max_completed_hold_operation: longest_completed_hold.operation,
201 reader_discards: self.reader_discards.load(Ordering::Relaxed),
202 reader_replacement_open_failures: self
203 .reader_replacement_open_failures
204 .load(Ordering::Relaxed),
205 }
206 }
207}