a3s_vec/
stats.rs

1//! Collection telemetry and health snapshots.
2
3use crate::collection::CollectionResourceLimits;
4use crate::config::IoBackend;
5use serde::{Deserialize, Serialize};
6use std::sync::atomic::{AtomicU64, Ordering};
7
8#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
9pub struct IndexStat {
10    pub name: String,
11    pub index_type: crate::types::IndexType,
12    pub completeness: f32,
13    pub source_revision: u64,
14    pub document_count: u64,
15    /// Deterministic encoded index payload estimate. This excludes allocator,
16    /// map-node, and authoritative document-storage overhead.
17    #[serde(default)]
18    pub estimated_payload_bytes: Option<u64>,
19    pub state: String,
20}
21
22#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
23pub struct StatsSnapshot {
24    pub collection_name: String,
25    pub revision: u64,
26    pub doc_count: u64,
27    pub query_count: u64,
28    pub fts_query_count: u64,
29    pub fts_index_query_count: u64,
30    pub ann_query_count: u64,
31    /// Successful ANN queries whose immutable Vamana/DiskANN base was traversed
32    /// from the native sector sidecar instead of the in-memory graph.
33    #[serde(default)]
34    pub diskann_query_count: u64,
35    /// Successful Vamana/`DiskANN` sidecar queries served from an immutable
36    /// mmap snapshot instead of positioned file reads.
37    #[serde(default)]
38    pub diskann_mmap_query_count: u64,
39    /// Native sidecar sectors staged in request-local caches by successful
40    /// positioned or mmap-snapshot graph queries.
41    #[serde(default)]
42    pub diskann_sector_read_count: u64,
43    pub exact_query_count: u64,
44    pub filtered_query_count: u64,
45    pub scalar_index_query_count: u64,
46    pub radius_query_count: u64,
47    pub candidates_scanned: u64,
48    pub indexed_field_count: usize,
49    pub indexes: Vec<IndexStat>,
50    /// Whether this handle restored its index generation from the optional
51    /// on-disk derived-index cache during `Collection::open`.
52    #[serde(default)]
53    pub index_cache_hit: bool,
54    /// Resolved sidecar backend for this collection handle.
55    #[serde(default)]
56    pub io_backend: crate::config::IoBackend,
57    pub read_only: bool,
58    pub wal_active_seq: u64,
59    pub wal_checkpoint_seq: u64,
60    pub wal_ops_since_checkpoint: u64,
61    pub wal_bytes_since_checkpoint: u64,
62    /// Deterministic serialized size of the authoritative document map.
63    #[serde(default)]
64    pub accounted_document_bytes: u64,
65    /// Sum of deterministic derived-index payload estimates.
66    #[serde(default)]
67    pub estimated_index_bytes: u64,
68    /// Authoritative document accounting plus derived-index estimates.
69    #[serde(default)]
70    pub accounted_bytes: u64,
71    /// Collection-local limits captured when this handle was opened.
72    #[serde(default)]
73    pub resource_limits: CollectionResourceLimits,
74    /// Operations rejected by this handle's resource policy.
75    #[serde(default)]
76    pub resource_limit_rejections: u64,
77}
78
79/// Readiness assessment for one collection handle.
80#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
81#[serde(rename_all = "snake_case")]
82pub enum CollectionHealthStatus {
83    /// The authoritative and derived generations agree and every configured
84    /// index is complete.
85    Healthy,
86    /// Authoritative storage is usable, but at least one derived index is
87    /// incomplete, missing, or stale.
88    Degraded,
89    /// The in-memory and durable authoritative revisions disagree.
90    Unhealthy,
91    /// The shared collection handle has been closed.
92    Closed,
93}
94
95/// Machine-readable collection readiness and checkpoint-lag snapshot.
96#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
97pub struct CollectionHealth {
98    pub status: CollectionHealthStatus,
99    pub revision: u64,
100    pub storage_revision: u64,
101    pub doc_count: u64,
102    pub index_count: usize,
103    pub ready_index_count: usize,
104    pub read_only: bool,
105    /// WAL work awaiting an authoritative snapshot checkpoint. Pending WAL is
106    /// normal for interval/manual durability and does not by itself degrade
107    /// readiness.
108    pub checkpoint_pending: bool,
109    pub wal_ops_since_checkpoint: u64,
110    pub wal_bytes_since_checkpoint: u64,
111    /// Whether an explicitly owned background maintenance runtime currently
112    /// holds this collection's single scheduler claim.
113    pub maintenance_active: bool,
114    /// Stable, actionable explanations for non-healthy states.
115    pub reasons: Vec<String>,
116}
117
118impl CollectionHealth {
119    /// Returns true only while the collection is open and fully ready.
120    pub fn is_healthy(&self) -> bool {
121        self.status == CollectionHealthStatus::Healthy
122    }
123}
124
125#[derive(Clone, Copy)]
126pub(crate) struct CollectionHealthInput<'a> {
127    pub is_open: bool,
128    pub revision: u64,
129    pub storage_revision: u64,
130    pub doc_count: u64,
131    pub indexes: &'a [IndexStat],
132    pub read_only: bool,
133    pub wal_ops_since_checkpoint: u64,
134    pub wal_bytes_since_checkpoint: u64,
135    pub maintenance_active: bool,
136}
137
138pub(crate) fn assess_collection_health(input: CollectionHealthInput<'_>) -> CollectionHealth {
139    let ready_index_count = input
140        .indexes
141        .iter()
142        .filter(|index| {
143            index.state == "ready"
144                && is_complete(index.completeness)
145                && index.source_revision == input.revision
146        })
147        .count();
148    let mut reasons = Vec::new();
149    let status = if input.is_open {
150        let mut status = CollectionHealthStatus::Healthy;
151        if input.storage_revision != input.revision {
152            reasons.push(format!(
153                "storage revision {} does not match collection revision {}",
154                input.storage_revision, input.revision
155            ));
156            status = CollectionHealthStatus::Unhealthy;
157        }
158        for index in input.indexes {
159            if index.state != "ready" {
160                reasons.push(format!(
161                    "index '{}' is in '{}' state",
162                    index.name, index.state
163                ));
164            } else if !is_complete(index.completeness) {
165                reasons.push(format!(
166                    "index '{}' completeness is {}",
167                    index.name, index.completeness
168                ));
169            } else if index.source_revision != input.revision {
170                reasons.push(format!(
171                    "index '{}' source revision {} does not match collection revision {}",
172                    index.name, index.source_revision, input.revision
173                ));
174            } else {
175                continue;
176            }
177            if status == CollectionHealthStatus::Healthy {
178                status = CollectionHealthStatus::Degraded;
179            }
180        }
181        status
182    } else {
183        reasons.push("collection is closed".to_string());
184        CollectionHealthStatus::Closed
185    };
186
187    CollectionHealth {
188        status,
189        revision: input.revision,
190        storage_revision: input.storage_revision,
191        doc_count: input.doc_count,
192        index_count: input.indexes.len(),
193        ready_index_count,
194        read_only: input.read_only,
195        checkpoint_pending: input.wal_ops_since_checkpoint > 0
196            || input.wal_bytes_since_checkpoint > 0,
197        wal_ops_since_checkpoint: input.wal_ops_since_checkpoint,
198        wal_bytes_since_checkpoint: input.wal_bytes_since_checkpoint,
199        maintenance_active: input.maintenance_active,
200        reasons,
201    }
202}
203
204fn is_complete(completeness: f32) -> bool {
205    (completeness - 1.0).abs() <= f32::EPSILON
206}
207
208#[derive(Debug, Default)]
209pub(crate) struct StatsRegistry {
210    pub query_count: AtomicU64,
211    pub fts_query_count: AtomicU64,
212    pub fts_index_query_count: AtomicU64,
213    pub ann_query_count: AtomicU64,
214    pub diskann_query_count: AtomicU64,
215    pub diskann_mmap_query_count: AtomicU64,
216    pub diskann_sector_read_count: AtomicU64,
217    pub exact_query_count: AtomicU64,
218    pub filtered_query_count: AtomicU64,
219    pub scalar_index_query_count: AtomicU64,
220    pub radius_query_count: AtomicU64,
221    pub candidates_scanned: AtomicU64,
222    pub resource_limit_rejections: AtomicU64,
223}
224
225#[derive(Debug, Clone, Copy)]
226pub(crate) struct QueryObservation {
227    pub kind: QueryKind,
228    pub diskann_io_backend: Option<IoBackend>,
229    pub diskann_sector_reads: u64,
230    pub filtered: bool,
231    pub index_usage: IndexUsage,
232    pub radius: bool,
233    pub candidates: u64,
234}
235
236#[derive(Debug, Clone, Copy, PartialEq, Eq)]
237pub(crate) enum IndexUsage {
238    None,
239    Scalar,
240    Fts,
241    ScalarFts,
242}
243
244impl IndexUsage {
245    pub(crate) fn new(scalar: bool, fts: bool) -> Self {
246        match (scalar, fts) {
247            (false, false) => Self::None,
248            (true, false) => Self::Scalar,
249            (false, true) => Self::Fts,
250            (true, true) => Self::ScalarFts,
251        }
252    }
253
254    fn scalar(self) -> bool {
255        matches!(self, Self::Scalar | Self::ScalarFts)
256    }
257
258    fn fts(self) -> bool {
259        matches!(self, Self::Fts | Self::ScalarFts)
260    }
261}
262
263#[derive(Debug, Clone, Copy, PartialEq, Eq)]
264pub(crate) enum QueryKind {
265    Exact,
266    Fts,
267    Ann,
268    AnnFts,
269}
270
271impl StatsRegistry {
272    pub fn record_query(&self, observation: QueryObservation) {
273        self.query_count.fetch_add(1, Ordering::Relaxed);
274        if matches!(observation.kind, QueryKind::Fts | QueryKind::AnnFts) {
275            self.fts_query_count.fetch_add(1, Ordering::Relaxed);
276        }
277        if observation.index_usage.fts() {
278            self.fts_index_query_count.fetch_add(1, Ordering::Relaxed);
279        }
280        if matches!(observation.kind, QueryKind::Ann | QueryKind::AnnFts) {
281            self.ann_query_count.fetch_add(1, Ordering::Relaxed);
282        }
283        if let Some(io_backend) = observation.diskann_io_backend {
284            self.diskann_query_count.fetch_add(1, Ordering::Relaxed);
285            if io_backend == IoBackend::Mmap {
286                self.diskann_mmap_query_count
287                    .fetch_add(1, Ordering::Relaxed);
288            }
289            self.diskann_sector_read_count
290                .fetch_add(observation.diskann_sector_reads, Ordering::Relaxed);
291        }
292        self.exact_query_count.fetch_add(1, Ordering::Relaxed);
293        if observation.filtered {
294            self.filtered_query_count.fetch_add(1, Ordering::Relaxed);
295        }
296        if observation.index_usage.scalar() {
297            self.scalar_index_query_count
298                .fetch_add(1, Ordering::Relaxed);
299        }
300        if observation.radius {
301            self.radius_query_count.fetch_add(1, Ordering::Relaxed);
302        }
303        self.candidates_scanned
304            .fetch_add(observation.candidates, Ordering::Relaxed);
305    }
306
307    pub fn record_resource_limit_rejection(&self) {
308        self.resource_limit_rejections
309            .fetch_add(1, Ordering::Relaxed);
310    }
311}
312
313#[cfg(test)]
314mod tests {
315    use super::*;
316    use crate::types::IndexType;
317
318    fn index(source_revision: u64, completeness: f32, state: &str) -> IndexStat {
319        IndexStat {
320            name: "embedding".to_string(),
321            index_type: IndexType::Hnsw,
322            completeness,
323            source_revision,
324            document_count: 2,
325            estimated_payload_bytes: Some(128),
326            state: state.to_string(),
327        }
328    }
329
330    fn assess(indexes: &[IndexStat], storage_revision: u64) -> CollectionHealth {
331        assess_collection_health(CollectionHealthInput {
332            is_open: true,
333            revision: 7,
334            storage_revision,
335            doc_count: 2,
336            indexes,
337            read_only: false,
338            wal_ops_since_checkpoint: 0,
339            wal_bytes_since_checkpoint: 0,
340            maintenance_active: false,
341        })
342    }
343
344    #[test]
345    fn stale_or_non_finite_index_completeness_is_degraded() {
346        let stale = assess(&[index(6, 1.0, "ready")], 7);
347        assert_eq!(stale.status, CollectionHealthStatus::Degraded);
348        assert_eq!(stale.ready_index_count, 0);
349        assert!(stale.reasons[0].contains("source revision 6"));
350
351        let incomplete = assess(&[index(7, f32::NAN, "ready")], 7);
352        assert_eq!(incomplete.status, CollectionHealthStatus::Degraded);
353        assert_eq!(incomplete.ready_index_count, 0);
354        assert!(incomplete.reasons[0].contains("completeness"));
355    }
356
357    #[test]
358    fn authoritative_revision_disagreement_is_unhealthy() {
359        let health = assess(&[index(7, 1.0, "ready")], 6);
360        assert_eq!(health.status, CollectionHealthStatus::Unhealthy);
361        assert_eq!(health.ready_index_count, 1);
362        assert!(health.reasons[0].contains("storage revision 6"));
363    }
364
365    #[test]
366    fn closed_status_is_explicit_even_for_a_consistent_snapshot() {
367        let ready = [index(7, 1.0, "ready")];
368        let health = assess_collection_health(CollectionHealthInput {
369            is_open: false,
370            revision: 7,
371            storage_revision: 7,
372            doc_count: 2,
373            indexes: &ready,
374            read_only: false,
375            wal_ops_since_checkpoint: 0,
376            wal_bytes_since_checkpoint: 0,
377            maintenance_active: false,
378        });
379        assert_eq!(health.status, CollectionHealthStatus::Closed);
380        assert_eq!(health.ready_index_count, 1);
381        assert!(!health.is_healthy());
382    }
383}
a3s_vec/stats.rs

a3s_vec/
stats.rs
