mongreldb-server 0.64.11

HTTP daemon for MongrelDB — serves SQL, native queries, and typed Kit API over HTTP for multi-process access.
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
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
//! Lightweight Prometheus-compatible metrics + slow-query instrumentation.
//!
//! No external metrics crate — counters are plain `AtomicU64`s and the
//! `/metrics` handler emits the [Prometheus text exposition format][1] by hand.
//! This keeps the daemon's dependency surface minimal and matches the
//! "sub-ms writes" ethos: the counters are bumped with `Relaxed` ordering (no
//! global fence on the hot path).
//!
//! [1]: https://github.com/prometheus/docs/blob/main/content/docs/instrumenting/exposition_formats.md

use std::sync::atomic::{AtomicU64, Ordering};

use mongreldb_core::engine::LookupMetricsSnapshot;

/// Stable label names for `hot_fallback_total{reason=...}`. Order MUST match
/// the indices baked into [`mongreldb_core::engine::hot_fallback_reason_index`].
const HOT_FALLBACK_REASON_LABELS: [&str; 9] = [
    "missing_mapping",
    "stale_row_id",
    "invisible_at_snapshot",
    "historical_snapshot",
    "tombstone",
    "ttl_expired",
    "primary_key_mismatch",
    "index_incomplete",
    "checkpoint_rejected",
];

/// Daemon-wide counters, bumped from every instrumented HTTP handler.
///
/// Stored as `Arc<Metrics>` inside `AppState` so all handlers share one set.
/// Every field is a monotonic counter unless noted (gauges are snapshot at
/// scrape time).
#[derive(Default)]
pub struct Metrics {
    /// Total `/sql` requests received (before execution).
    pub sql_queries: AtomicU64,
    /// `/sql` requests that returned an error.
    pub sql_errors: AtomicU64,
    /// `/sql` requests slower than the configured slow-query threshold.
    pub slow_queries: AtomicU64,
    /// Total single-row `PUT /tables/{name}/put` calls.
    pub puts: AtomicU64,
    /// Total `POST /tables/{name}/commit` calls.
    pub commits: AtomicU64,
    /// Total `POST /txn` atomic batch calls.
    pub txns: AtomicU64,
    pub sql_cancel_requests: AtomicU64,
    pub sql_cancelled: AtomicU64,
    pub sql_cancelled_by_reason: [AtomicU64; 6],
    pub sql_deadline_exceeded: AtomicU64,
    pub sql_commit_cancel_winner_cancel: AtomicU64,
    pub sql_commit_cancel_winner_commit: AtomicU64,
    pub sql_stuck_after_cancel: AtomicU64,
    pub sql_output_bytes: AtomicU64,
    pub sql_cancel_latency_micros: AtomicU64,
    pub sql_cancel_latency_count: AtomicU64,
}

impl Metrics {
    /// Bump `sql_queries`.
    #[inline]
    pub fn inc_sql_queries(&self) {
        self.sql_queries.fetch_add(1, Ordering::Relaxed);
    }

    /// Bump `sql_errors`.
    #[inline]
    pub fn inc_sql_errors(&self) {
        self.sql_errors.fetch_add(1, Ordering::Relaxed);
    }

    /// Bump `slow_queries`.
    #[inline]
    pub fn inc_slow_queries(&self) {
        self.slow_queries.fetch_add(1, Ordering::Relaxed);
    }

    /// Bump `puts`.
    #[inline]
    pub fn inc_puts(&self) {
        self.puts.fetch_add(1, Ordering::Relaxed);
    }

    /// Bump `commits`.
    #[inline]
    pub fn inc_commits(&self) {
        self.commits.fetch_add(1, Ordering::Relaxed);
    }

    /// Bump `txns`.
    #[inline]
    pub fn inc_txns(&self) {
        self.txns.fetch_add(1, Ordering::Relaxed);
    }

    pub fn inc_sql_cancel_requests(&self) {
        self.sql_cancel_requests.fetch_add(1, Ordering::Relaxed);
    }

    pub fn inc_sql_cancelled(&self, reason: mongreldb_core::CancellationReason) {
        self.sql_cancelled.fetch_add(1, Ordering::Relaxed);
        self.sql_cancelled_by_reason[reason as usize].fetch_add(1, Ordering::Relaxed);
    }

    pub fn inc_sql_deadline_exceeded(&self) {
        self.sql_deadline_exceeded.fetch_add(1, Ordering::Relaxed);
    }

    pub fn inc_sql_commit_cancel_winner_cancel(&self) {
        self.sql_commit_cancel_winner_cancel
            .fetch_add(1, Ordering::Relaxed);
    }

    pub fn inc_sql_commit_cancel_winner_commit(&self) {
        self.sql_commit_cancel_winner_commit
            .fetch_add(1, Ordering::Relaxed);
    }

    pub fn add_sql_stuck_after_cancel(&self, count: usize) {
        self.sql_stuck_after_cancel
            .fetch_add(count.min(u64::MAX as usize) as u64, Ordering::Relaxed);
    }

    pub fn add_sql_output_bytes(&self, bytes: usize) {
        self.sql_output_bytes
            .fetch_add(bytes.min(u64::MAX as usize) as u64, Ordering::Relaxed);
    }

    pub fn observe_sql_cancel_latency(&self, latency: std::time::Duration) {
        self.sql_cancel_latency_micros.fetch_add(
            latency.as_micros().min(u128::from(u64::MAX)) as u64,
            Ordering::Relaxed,
        );
        self.sql_cancel_latency_count
            .fetch_add(1, Ordering::Relaxed);
    }

    /// Render the current counter values as a Prometheus text-format body.
    ///
    /// `table_count` is passed in as a gauge (it is read off the live
    /// `Database` at scrape time rather than maintained as a counter).
    pub fn prometheus_text(
        &self,
        table_count: usize,
        registry: mongreldb_query::QueryRegistryStats,
        pre_cancel: (usize, usize),
    ) -> String {
        let (pre_cancel_entries, pre_cancel_bytes) = pre_cancel;
        let sql_queries = self.sql_queries.load(Ordering::Relaxed);
        let sql_errors = self.sql_errors.load(Ordering::Relaxed);
        let slow_queries = self.slow_queries.load(Ordering::Relaxed);
        let puts = self.puts.load(Ordering::Relaxed);
        let commits = self.commits.load(Ordering::Relaxed);
        let txns = self.txns.load(Ordering::Relaxed);
        let cancel_requests = self.sql_cancel_requests.load(Ordering::Relaxed);
        let cancelled = self.sql_cancelled.load(Ordering::Relaxed);
        let cancelled_by_reason = self
            .sql_cancelled_by_reason
            .each_ref()
            .map(|counter| counter.load(Ordering::Relaxed));
        let deadline_exceeded = self.sql_deadline_exceeded.load(Ordering::Relaxed);
        let race_cancel = self.sql_commit_cancel_winner_cancel.load(Ordering::Relaxed);
        let race_commit = self.sql_commit_cancel_winner_commit.load(Ordering::Relaxed);
        let stuck = self.sql_stuck_after_cancel.load(Ordering::Relaxed);
        let output_bytes = self.sql_output_bytes.load(Ordering::Relaxed);
        let cancel_latency_micros = self.sql_cancel_latency_micros.load(Ordering::Relaxed);
        let cancel_latency_count = self.sql_cancel_latency_count.load(Ordering::Relaxed);

        let mut out = String::with_capacity(1024);
        // mongreldb_sql_queries_total
        out.push_str("# HELP mongreldb_sql_queries_total Total /sql requests received.\n");
        out.push_str("# TYPE mongreldb_sql_queries_total counter\n");
        out.push_str(&format!("mongreldb_sql_queries_total {sql_queries}\n\n"));
        out.push_str("# TYPE mongreldb_sql_active_queries gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_active_queries {}\n\n",
            registry.active
        ));
        out.push_str("# TYPE mongreldb_sql_queued_queries gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_queued_queries {}\n\n",
            registry.queued
        ));
        out.push_str("# TYPE mongreldb_sql_cancel_requests_total counter\n");
        out.push_str(&format!(
            "mongreldb_sql_cancel_requests_total {cancel_requests}\n\n"
        ));
        out.push_str("# TYPE mongreldb_sql_cancelled_total counter\n");
        out.push_str(&format!("mongreldb_sql_cancelled_total {cancelled}\n"));
        for (index, reason) in [
            "none",
            "client_request",
            "deadline",
            "client_disconnected",
            "session_closed",
            "server_shutdown",
        ]
        .into_iter()
        .enumerate()
        {
            out.push_str(&format!(
                "mongreldb_sql_cancelled_total{{reason=\"{reason}\"}} {}\n",
                cancelled_by_reason[index]
            ));
        }
        out.push('\n');
        out.push_str("# TYPE mongreldb_sql_cancel_latency_seconds summary\n");
        out.push_str(&format!(
            "mongreldb_sql_cancel_latency_seconds_sum {}\n",
            cancel_latency_micros as f64 / 1_000_000.0
        ));
        out.push_str(&format!(
            "mongreldb_sql_cancel_latency_seconds_count {cancel_latency_count}\n\n"
        ));
        out.push_str("# TYPE mongreldb_sql_deadline_exceeded_total counter\n");
        out.push_str(&format!(
            "mongreldb_sql_deadline_exceeded_total {deadline_exceeded}\n\n"
        ));
        out.push_str("# TYPE mongreldb_sql_commit_cancel_races_total counter\n");
        out.push_str(&format!(
            "mongreldb_sql_commit_cancel_races_total{{winner=\"cancel\"}} {race_cancel}\n"
        ));
        out.push_str(&format!(
            "mongreldb_sql_commit_cancel_races_total{{winner=\"commit\"}} {race_commit}\n\n"
        ));
        out.push_str("# TYPE mongreldb_sql_registry_entries gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_entries {}\n\n",
            registry.active + registry.detailed + registry.compact
        ));
        out.push_str("# TYPE mongreldb_sql_registry_bytes gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_bytes {}\n\n",
            registry.detailed_bytes + registry.compact_bytes
        ));
        out.push_str("# TYPE mongreldb_sql_registry_active gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_active {}\n",
            registry.active
        ));
        out.push_str("# TYPE mongreldb_sql_registry_queued gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_queued {}\n",
            registry.queued
        ));
        out.push_str("# TYPE mongreldb_sql_registry_detailed gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_detailed {}\n",
            registry.detailed
        ));
        out.push_str("# TYPE mongreldb_sql_registry_compact gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_compact {}\n",
            registry.compact
        ));
        out.push_str("# TYPE mongreldb_sql_registry_detailed_bytes gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_detailed_bytes {}\n",
            registry.detailed_bytes
        ));
        out.push_str("# TYPE mongreldb_sql_registry_compact_bytes gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_compact_bytes {}\n",
            registry.compact_bytes
        ));
        out.push_str("# TYPE mongreldb_sql_registry_demotions_total counter\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_demotions_total {}\n",
            registry.demotions
        ));
        out.push_str("# TYPE mongreldb_sql_registry_compact_evictions_total counter\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_compact_evictions_total {}\n",
            registry.compact_evictions
        ));
        out.push_str("# TYPE mongreldb_sql_registry_rejections_total counter\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_rejections_total{{reason=\"active_limit\"}} {}\n",
            registry.active_rejections
        ));
        out.push_str("# TYPE mongreldb_sql_registry_oldest_compact_age_seconds gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_registry_oldest_compact_age_seconds {}\n\n",
            registry.oldest_compact_age.as_secs_f64()
        ));
        out.push_str("# TYPE mongreldb_sql_pre_cancel_entries gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_pre_cancel_entries {pre_cancel_entries}\n\n"
        ));
        out.push_str("# TYPE mongreldb_sql_pre_cancel_bytes gauge\n");
        out.push_str(&format!(
            "mongreldb_sql_pre_cancel_bytes {pre_cancel_bytes}\n\n"
        ));
        out.push_str("# TYPE mongreldb_sql_stuck_after_cancel_total counter\n");
        out.push_str(&format!(
            "mongreldb_sql_stuck_after_cancel_total {stuck}\n\n"
        ));
        out.push_str("# TYPE mongreldb_sql_output_bytes counter\n");
        out.push_str(&format!("mongreldb_sql_output_bytes {output_bytes}\n\n"));
        // mongreldb_sql_errors_total
        out.push_str("# HELP mongreldb_sql_errors_total /sql requests that returned an error.\n");
        out.push_str("# TYPE mongreldb_sql_errors_total counter\n");
        out.push_str(&format!("mongreldb_sql_errors_total {sql_errors}\n\n"));
        // mongreldb_slow_queries_total
        out.push_str(
            "# HELP mongreldb_slow_queries_total /sql requests above the slow-query threshold.\n",
        );
        out.push_str("# TYPE mongreldb_slow_queries_total counter\n");
        out.push_str(&format!("mongreldb_slow_queries_total {slow_queries}\n\n"));
        // mongreldb_puts_total
        out.push_str("# HELP mongreldb_puts_total Single-row put requests.\n");
        out.push_str("# TYPE mongreldb_puts_total counter\n");
        out.push_str(&format!("mongreldb_puts_total {puts}\n\n"));
        // mongreldb_commits_total
        out.push_str("# HELP mongreldb_commits_total Explicit table commit requests.\n");
        out.push_str("# TYPE mongreldb_commits_total counter\n");
        out.push_str(&format!("mongreldb_commits_total {commits}\n\n"));
        // mongreldb_txns_total
        out.push_str("# HELP mongreldb_txns_total Atomic /txn batch requests.\n");
        out.push_str("# TYPE mongreldb_txns_total counter\n");
        out.push_str(&format!("mongreldb_txns_total {txns}\n\n"));
        // mongreldb_tables (gauge)
        out.push_str("# HELP mongreldb_tables Current number of tables.\n");
        out.push_str("# TYPE mongreldb_tables gauge\n");
        out.push_str(&format!("mongreldb_tables {table_count}\n"));
        out
    }
}

/// Render the HOT-fallback counters from `snap` as a Prometheus text-format
/// block. Each metric carries one HELP + TYPE preamble followed by zero or
/// more sample lines. Callers append this after the SQL/cache block so the
/// final `/metrics` body remains a single valid exposition.
pub fn hot_lookup_metrics(snap: &LookupMetricsSnapshot) -> String {
    let mut out = String::with_capacity(1024);
    // hot_lookup_total{outcome=...}
    out.push_str("# HELP hot_lookup_total HOT primary-key lookups by outcome.\n");
    out.push_str("# TYPE hot_lookup_total counter\n");
    out.push_str(&format!(
        "hot_lookup_total{{outcome=\"hit\"}} {}\n",
        snap.hot_lookup_hit
    ));
    out.push_str(&format!(
        "hot_lookup_total{{outcome=\"fallback\"}} {}\n\n",
        snap.hot_lookup_fallback
    ));
    // hot_fallback_total{reason=...}
    out.push_str(
        "# HELP hot_fallback_total HOT lookups that fell back to the row scanner, by reason.\n",
    );
    out.push_str("# TYPE hot_fallback_total counter\n");
    for (idx, reason) in HOT_FALLBACK_REASON_LABELS.iter().enumerate() {
        out.push_str(&format!(
            "hot_fallback_total{{reason=\"{reason}\"}} {}\n",
            snap.hot_fallback_reasons[idx]
        ));
    }
    out.push('\n');
    // run/page/row counters (no labels).
    out.push_str("# HELP hot_fallback_overlay_versions_total Overlay row versions consulted during HOT fallback.\n");
    out.push_str("# TYPE hot_fallback_overlay_versions_total counter\n");
    out.push_str(&format!(
        "hot_fallback_overlay_versions_total {}\n\n",
        snap.hot_fallback_overlay_versions_total
    ));
    out.push_str(
        "# HELP hot_fallback_runs_considered_total Sorted runs considered during HOT fallback.\n",
    );
    out.push_str("# TYPE hot_fallback_runs_considered_total counter\n");
    out.push_str(&format!(
        "hot_fallback_runs_considered_total {}\n",
        snap.hot_fallback_runs_considered_total
    ));
    out.push_str("# TYPE hot_fallback_runs_opened_total counter\n");
    out.push_str(&format!(
        "hot_fallback_runs_opened_total {}\n",
        snap.hot_fallback_runs_opened_total
    ));
    out.push_str("# TYPE hot_fallback_pages_decoded_total counter\n");
    out.push_str(&format!(
        "hot_fallback_pages_decoded_total {}\n",
        snap.hot_fallback_pages_decoded_total
    ));
    out.push_str("# TYPE hot_fallback_rows_materialized_total counter\n");
    out.push_str(&format!(
        "hot_fallback_rows_materialized_total {}\n\n",
        snap.hot_fallback_rows_materialized_total
    ));
    // duration counters (reported as fractional seconds; raw nanoseconds live
    // on the per-table atomics).
    out.push_str(
        "# HELP hot_lookup_duration_seconds Cumulative time spent in HOT primary-key lookups.\n",
    );
    out.push_str("# TYPE hot_lookup_duration_seconds counter\n");
    out.push_str(&format!(
        "hot_lookup_duration_seconds {}\n",
        snap.hot_lookup_duration_nanos as f64 / 1e9
    ));
    out.push_str("# TYPE hot_fallback_duration_seconds counter\n");
    out.push_str(&format!(
        "hot_fallback_duration_seconds {}\n\n",
        snap.hot_fallback_duration_nanos as f64 / 1e9
    ));
    // mapping rebuild / checkpoint rejection.
    out.push_str(
        "# HELP hot_mapping_rebuild_total HOT->row-id mapping rebuilds triggered by stale state.\n",
    );
    out.push_str("# TYPE hot_mapping_rebuild_total counter\n");
    out.push_str(&format!(
        "hot_mapping_rebuild_total {}\n",
        snap.hot_mapping_rebuild_total
    ));
    out.push_str("# TYPE hot_checkpoint_rejected_total counter\n");
    out.push_str(&format!(
        "hot_checkpoint_rejected_total {}\n",
        snap.hot_checkpoint_rejected_total
    ));
    out
}

/// Aggregate `Table::lookup_metrics_snapshot()` over every live table on
/// `state.db()`. Per-reason arrays sum element-wise, scalar counters sum,
/// and the persist queue depth takes the per-table max. Tolerant of an
/// empty database (returns `LookupMetricsSnapshot::default()`).
pub fn aggregate_hot_metrics(state: &crate::AppState) -> LookupMetricsSnapshot {
    let names = state.db().table_names();
    if names.is_empty() {
        return LookupMetricsSnapshot::default();
    }
    let mut agg = LookupMetricsSnapshot::default();
    for name in &names {
        let Ok(handle) = state.db().table(name) else {
            continue;
        };
        let snap = handle.read().lookup_metrics_snapshot();
        agg.hot_lookup_hit += snap.hot_lookup_hit;
        agg.hot_lookup_fallback += snap.hot_lookup_fallback;
        agg.hot_lookup_fallback_overlay_rows += snap.hot_lookup_fallback_overlay_rows;
        agg.hot_lookup_fallback_runs += snap.hot_lookup_fallback_runs;
        agg.result_cache_memory_hit += snap.result_cache_memory_hit;
        agg.result_cache_disk_hit += snap.result_cache_disk_hit;
        agg.result_cache_miss += snap.result_cache_miss;
        agg.result_cache_persistent_write_us += snap.result_cache_persistent_write_us;
        agg.get_run_opened += snap.get_run_opened;
        agg.get_run_skipped += snap.get_run_skipped;
        agg.directory_lookup_hit += snap.directory_lookup_hit;
        agg.directory_lookup_fallback += snap.directory_lookup_fallback;
        agg.directory_incomplete += snap.directory_incomplete;
        agg.directory_run_readers_opened += snap.directory_run_readers_opened;
        agg.directory_early_stop_total += snap.directory_early_stop_total;
        agg.result_cache_persist_enqueued_total += snap.result_cache_persist_enqueued_total;
        agg.result_cache_persist_coalesced_total += snap.result_cache_persist_coalesced_total;
        agg.result_cache_persist_dropped_store_total +=
            snap.result_cache_persist_dropped_store_total;
        agg.result_cache_persist_remove_total += snap.result_cache_persist_remove_total;
        agg.result_cache_persist_stale_store_skipped_total +=
            snap.result_cache_persist_stale_store_skipped_total;
        agg.result_cache_persist_errors_total += snap.result_cache_persist_errors_total;
        agg.result_cache_persist_shutdown_abandoned_total +=
            snap.result_cache_persist_shutdown_abandoned_total;
        for i in 0..agg.hot_fallback_reasons.len() {
            agg.hot_fallback_reasons[i] += snap.hot_fallback_reasons[i];
        }
        agg.hot_fallback_overlay_versions_total += snap.hot_fallback_overlay_versions_total;
        agg.hot_fallback_runs_considered_total += snap.hot_fallback_runs_considered_total;
        agg.hot_fallback_runs_opened_total += snap.hot_fallback_runs_opened_total;
        agg.hot_fallback_pages_decoded_total += snap.hot_fallback_pages_decoded_total;
        agg.hot_fallback_rows_materialized_total += snap.hot_fallback_rows_materialized_total;
        agg.hot_lookup_duration_nanos += snap.hot_lookup_duration_nanos;
        agg.hot_fallback_duration_nanos += snap.hot_fallback_duration_nanos;
        agg.hot_mapping_rebuild_total += snap.hot_mapping_rebuild_total;
        agg.hot_checkpoint_rejected_total += snap.hot_checkpoint_rejected_total;
        // queue_depth is a gauge — track the high-water mark across tables.
        if snap.result_cache_persist_queue_depth > agg.result_cache_persist_queue_depth {
            agg.result_cache_persist_queue_depth = snap.result_cache_persist_queue_depth;
        }
    }
    agg
}

/// Read the slow-query threshold from the `MONGRELBL_SLOW_QUERY_MS` env var,
/// defaulting to 100 ms. Returns the threshold as a `Duration`.
pub fn slow_query_threshold() -> std::time::Duration {
    const DEFAULT_MS: u64 = 100;
    let ms = std::env::var("MONGRELBL_SLOW_QUERY_MS")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .unwrap_or(DEFAULT_MS);
    std::time::Duration::from_millis(ms.max(1))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn prometheus_text_contains_all_series_and_types() {
        let m = Metrics::default();
        m.inc_sql_queries();
        m.inc_sql_queries();
        m.inc_sql_errors();
        m.inc_slow_queries();
        m.inc_puts();
        m.inc_commits();
        m.inc_txns();
        let body = m.prometheus_text(
            3,
            mongreldb_query::QueryRegistryStats {
                active: 1,
                queued: 1,
                detailed: 1,
                detailed_bytes: 512,
                ..Default::default()
            },
            (1, 128),
        );
        // HELP/TYPE lines present for every series.
        assert!(body.contains("# TYPE mongreldb_sql_queries_total counter"));
        assert!(body.contains("# TYPE mongreldb_sql_errors_total counter"));
        assert!(body.contains("# TYPE mongreldb_slow_queries_total counter"));
        assert!(body.contains("# TYPE mongreldb_puts_total counter"));
        assert!(body.contains("# TYPE mongreldb_commits_total counter"));
        assert!(body.contains("# TYPE mongreldb_txns_total counter"));
        assert!(body.contains("# TYPE mongreldb_tables gauge"));
        assert!(body.contains("mongreldb_sql_pre_cancel_entries 1"));
        assert!(body.contains("mongreldb_sql_pre_cancel_bytes 128"));
        // Counter values reflect the bumps.
        assert!(body.contains("mongreldb_sql_queries_total 2"));
        assert!(body.contains("mongreldb_sql_errors_total 1"));
        assert!(body.contains("mongreldb_slow_queries_total 1"));
        assert!(body.contains("mongreldb_puts_total 1"));
        assert!(body.contains("mongreldb_commits_total 1"));
        assert!(body.contains("mongreldb_txns_total 1"));
        // Gauge value.
        assert!(body.contains("mongreldb_tables 3"));
    }

    #[test]
    fn default_threshold_is_100ms() {
        // Only assert the default when the env var is not set in the test
        // environment; if a developer sets it, respect their value.
        if std::env::var("MONGRELBL_SLOW_QUERY_MS").is_err() {
            assert_eq!(
                slow_query_threshold(),
                std::time::Duration::from_millis(100)
            );
        }
    }
}