velesdb-core 1.14.4

High-performance vector database engine written in Rust
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
//! Performance metrics for graph operations (EPIC-019 US-006).
//!
//! Provides low-overhead, thread-safe metrics for monitoring:
//! - Operation counters (inserts, deletes, traversals)
//! - Latency histograms
//! - Memory usage estimates
//!
//! Metrics use atomic operations with relaxed ordering for minimal overhead (~1-5ns per op).

// Reason: Numeric casts in metrics are intentional:
// - All casts are for histogram bucketing and latency calculations
// - f64/u64 conversions for computing percentiles and averages
// - Values bounded by practical limits (bucket counts, durations)
// - Precision loss acceptable for metrics (approximate by design)
#![allow(clippy::cast_precision_loss)]
#![allow(clippy::cast_possible_truncation)]

#[cfg(test)]
mod tests;

use std::fmt::Write;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

/// Latency histogram buckets (milliseconds).
const BUCKET_BOUNDS_MS: [u64; 9] = [1, 5, 10, 50, 100, 500, 1000, 5000, 10000];

/// Simple latency histogram with fixed buckets.
///
/// Buckets: <1ms, <5ms, <10ms, <50ms, <100ms, <500ms, <1s, <5s, <10s, ≥10s
#[derive(Debug, Default)]
pub struct LatencyHistogram {
    /// Bucket counts [<1ms, <5ms, <10ms, <50ms, <100ms, <500ms, <1s, <5s, <10s, ≥10s]
    buckets: [AtomicU64; 10],
    /// Sum of all observed durations in nanoseconds
    sum_ns: AtomicU64,
    /// Total number of observations
    count: AtomicU64,
}

impl LatencyHistogram {
    /// Creates a new empty histogram.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Records a duration observation.
    ///
    /// # Note
    ///
    /// For extremely large durations (> 584 years), nanoseconds are capped at u64::MAX
    /// to prevent truncation. This is acceptable since such durations indicate a bug.
    pub fn observe(&self, duration: Duration) {
        // Cap at u64::MAX for durations > 584 years (u128 -> u64 truncation protection)
        let ns_u128 = duration.as_nanos();
        let ns = if ns_u128 > u128::from(u64::MAX) {
            u64::MAX
        } else {
            ns_u128 as u64
        };
        self.sum_ns.fetch_add(ns, Ordering::Relaxed);
        self.count.fetch_add(1, Ordering::Relaxed);

        // Same protection for milliseconds (though less likely to overflow)
        let ms_u128 = duration.as_millis();
        let ms = if ms_u128 > u128::from(u64::MAX) {
            u64::MAX
        } else {
            ms_u128 as u64
        };
        let bucket_idx = BUCKET_BOUNDS_MS
            .iter()
            .position(|&bound| ms < bound)
            .unwrap_or(9);
        self.buckets[bucket_idx].fetch_add(1, Ordering::Relaxed);
    }

    /// Returns the total count of observations.
    #[must_use]
    pub fn count(&self) -> u64 {
        self.count.load(Ordering::Relaxed)
    }

    /// Returns the sum of all durations in nanoseconds.
    #[must_use]
    pub fn sum_ns(&self) -> u64 {
        self.sum_ns.load(Ordering::Relaxed)
    }

    /// Returns the average duration in nanoseconds.
    #[must_use]
    pub fn avg_ns(&self) -> f64 {
        let count = self.count();
        if count == 0 {
            0.0
        } else {
            self.sum_ns() as f64 / count as f64
        }
    }

    /// Returns bucket counts as an array.
    #[must_use]
    pub fn bucket_counts(&self) -> [u64; 10] {
        let mut counts = [0u64; 10];
        for (i, bucket) in self.buckets.iter().enumerate() {
            counts[i] = bucket.load(Ordering::Relaxed);
        }
        counts
    }

    /// Resets all counters to zero.
    pub fn reset(&self) {
        self.sum_ns.store(0, Ordering::Relaxed);
        self.count.store(0, Ordering::Relaxed);
        for bucket in &self.buckets {
            bucket.store(0, Ordering::Relaxed);
        }
    }
}

/// Graph-specific performance metrics.
///
/// Thread-safe counters and histograms for monitoring graph operations.
///
/// # Example
///
/// ```rust,ignore
/// use velesdb_core::collection::graph::GraphMetrics;
/// use std::time::Instant;
///
/// let metrics = GraphMetrics::new();
///
/// // Record an edge insertion
/// let start = Instant::now();
/// // ... perform insertion ...
/// metrics.record_edge_insert(start.elapsed());
///
/// // Get statistics
/// println!("Total edges inserted: {}", metrics.edge_inserts_total());
/// println!("Avg insert latency: {:.2}µs", metrics.edge_insert_latency.avg_ns() / 1000.0);
/// ```
#[derive(Debug, Default)]
pub struct GraphMetrics {
    // Node counters
    nodes_total: AtomicU64,
    node_inserts_total: AtomicU64,
    node_deletes_total: AtomicU64,

    // Edge counters
    edges_total: AtomicU64,
    edge_inserts_total: AtomicU64,
    edge_deletes_total: AtomicU64,

    // Traversal counters
    traversals_total: AtomicU64,
    traversal_nodes_visited: AtomicU64,

    // Latency histograms
    /// Edge insertion latency histogram
    pub edge_insert_latency: LatencyHistogram,
    /// Edge deletion latency histogram
    pub edge_delete_latency: LatencyHistogram,
    /// Traversal latency histogram
    pub traversal_latency: LatencyHistogram,
    /// Query latency histogram
    pub query_latency: LatencyHistogram,
}

impl GraphMetrics {
    /// Creates a new metrics instance with all counters at zero.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    // =========================================================================
    // Node metrics
    // =========================================================================

    /// Records a node insertion.
    pub fn record_node_insert(&self) {
        self.node_inserts_total.fetch_add(1, Ordering::Relaxed);
        self.nodes_total.fetch_add(1, Ordering::Relaxed);
    }

    /// Records a node deletion.
    ///
    /// Uses saturating subtraction to prevent underflow if called
    /// more times than `record_node_insert`.
    pub fn record_node_delete(&self) {
        self.node_deletes_total.fetch_add(1, Ordering::Relaxed);
        // Saturating sub: load, compute, compare-exchange loop
        loop {
            let current = self.nodes_total.load(Ordering::Relaxed);
            let new_val = current.saturating_sub(1);
            if self
                .nodes_total
                .compare_exchange_weak(current, new_val, Ordering::Relaxed, Ordering::Relaxed)
                .is_ok()
            {
                break;
            }
        }
    }

    /// Returns total node count.
    #[must_use]
    pub fn nodes_total(&self) -> u64 {
        self.nodes_total.load(Ordering::Relaxed)
    }

    /// Returns total node insertions.
    #[must_use]
    pub fn node_inserts_total(&self) -> u64 {
        self.node_inserts_total.load(Ordering::Relaxed)
    }

    // =========================================================================
    // Edge metrics
    // =========================================================================

    /// Records an edge insertion with latency.
    pub fn record_edge_insert(&self, latency: Duration) {
        self.edge_inserts_total.fetch_add(1, Ordering::Relaxed);
        self.edges_total.fetch_add(1, Ordering::Relaxed);
        self.edge_insert_latency.observe(latency);
    }

    /// Records an edge deletion with latency.
    ///
    /// Uses saturating subtraction to prevent underflow.
    pub fn record_edge_delete(&self, latency: Duration) {
        self.edge_deletes_total.fetch_add(1, Ordering::Relaxed);
        // Saturating sub to prevent underflow
        loop {
            let current = self.edges_total.load(Ordering::Relaxed);
            let new_val = current.saturating_sub(1);
            if self
                .edges_total
                .compare_exchange_weak(current, new_val, Ordering::Relaxed, Ordering::Relaxed)
                .is_ok()
            {
                break;
            }
        }
        self.edge_delete_latency.observe(latency);
    }

    /// Returns total edge count.
    #[must_use]
    pub fn edges_total(&self) -> u64 {
        self.edges_total.load(Ordering::Relaxed)
    }

    /// Returns total edge insertions.
    #[must_use]
    pub fn edge_inserts_total(&self) -> u64 {
        self.edge_inserts_total.load(Ordering::Relaxed)
    }

    /// Returns total edge deletions.
    #[must_use]
    pub fn edge_deletes_total(&self) -> u64 {
        self.edge_deletes_total.load(Ordering::Relaxed)
    }

    // =========================================================================
    // Traversal metrics
    // =========================================================================

    /// Records a traversal with latency and nodes visited.
    pub fn record_traversal(&self, latency: Duration, nodes_visited: u64) {
        self.traversals_total.fetch_add(1, Ordering::Relaxed);
        self.traversal_nodes_visited
            .fetch_add(nodes_visited, Ordering::Relaxed);
        self.traversal_latency.observe(latency);
    }

    /// Returns total traversal count.
    #[must_use]
    pub fn traversals_total(&self) -> u64 {
        self.traversals_total.load(Ordering::Relaxed)
    }

    /// Returns total nodes visited across all traversals.
    #[must_use]
    pub fn traversal_nodes_visited(&self) -> u64 {
        self.traversal_nodes_visited.load(Ordering::Relaxed)
    }

    // =========================================================================
    // Query metrics
    // =========================================================================

    /// Records a query latency.
    pub fn record_query(&self, latency: Duration) {
        self.query_latency.observe(latency);
    }

    // =========================================================================
    // Export
    // =========================================================================

    /// Exports metrics in Prometheus text format.
    #[must_use]
    pub fn to_prometheus(&self) -> String {
        let mut output = String::with_capacity(2048);

        // Node metrics
        output.push_str("# HELP velesdb_graph_nodes_total Current number of nodes\n");
        output.push_str("# TYPE velesdb_graph_nodes_total gauge\n");
        let _ = writeln!(output, "velesdb_graph_nodes_total {}\n", self.nodes_total());

        output.push_str("# HELP velesdb_graph_node_inserts_total Total node insertions\n");
        output.push_str("# TYPE velesdb_graph_node_inserts_total counter\n");
        let _ = writeln!(
            output,
            "velesdb_graph_node_inserts_total {}\n",
            self.node_inserts_total()
        );

        // Edge metrics
        output.push_str("# HELP velesdb_graph_edges_total Current number of edges\n");
        output.push_str("# TYPE velesdb_graph_edges_total gauge\n");
        let _ = writeln!(output, "velesdb_graph_edges_total {}\n", self.edges_total());

        output.push_str("# HELP velesdb_graph_edge_inserts_total Total edge insertions\n");
        output.push_str("# TYPE velesdb_graph_edge_inserts_total counter\n");
        let _ = writeln!(
            output,
            "velesdb_graph_edge_inserts_total {}\n",
            self.edge_inserts_total()
        );

        // Latency histograms
        Self::append_histogram_prometheus(&mut output, "edge_insert", &self.edge_insert_latency);
        Self::append_histogram_prometheus(&mut output, "traversal", &self.traversal_latency);

        // Traversal metrics
        output.push_str("# HELP velesdb_graph_traversals_total Total traversals executed\n");
        output.push_str("# TYPE velesdb_graph_traversals_total counter\n");
        let _ = writeln!(
            output,
            "velesdb_graph_traversals_total {}\n",
            self.traversals_total()
        );

        output
    }

    fn append_histogram_prometheus(output: &mut String, name: &str, histogram: &LatencyHistogram) {
        let bucket_bounds = [
            "0.001", "0.005", "0.01", "0.05", "0.1", "0.5", "1", "5", "10", "+Inf",
        ];
        let counts = histogram.bucket_counts();
        let mut cumulative = 0u64;

        let _ = writeln!(
            output,
            "# HELP velesdb_graph_{}_duration_seconds {} latency histogram",
            name,
            name.replace('_', " ")
        );
        let _ = writeln!(
            output,
            "# TYPE velesdb_graph_{name}_duration_seconds histogram"
        );

        for (i, &bound) in bucket_bounds.iter().enumerate() {
            cumulative += counts[i];
            let _ = writeln!(
                output,
                "velesdb_graph_{name}_duration_seconds_bucket{{le=\"{bound}\"}} {cumulative}",
            );
        }

        let _ = writeln!(
            output,
            "velesdb_graph_{}_duration_seconds_sum {}",
            name,
            histogram.sum_ns() as f64 / 1_000_000_000.0
        );
        let _ = writeln!(
            output,
            "velesdb_graph_{}_duration_seconds_count {}\n",
            name,
            histogram.count()
        );
    }

    /// Resets all metrics to zero.
    pub fn reset(&self) {
        self.nodes_total.store(0, Ordering::Relaxed);
        self.node_inserts_total.store(0, Ordering::Relaxed);
        self.node_deletes_total.store(0, Ordering::Relaxed);
        self.edges_total.store(0, Ordering::Relaxed);
        self.edge_inserts_total.store(0, Ordering::Relaxed);
        self.edge_deletes_total.store(0, Ordering::Relaxed);
        self.traversals_total.store(0, Ordering::Relaxed);
        self.traversal_nodes_visited.store(0, Ordering::Relaxed);
        self.edge_insert_latency.reset();
        self.edge_delete_latency.reset();
        self.traversal_latency.reset();
        self.query_latency.reset();
    }
}