query_runtime/attempt/
queue.rs

1//! Bounded best-effort report delivery and exact backend counters.
2#[cfg(test)]
3use std::sync::MutexGuard;
4use std::{
5    collections::VecDeque,
6    sync::{
7        Arc, Mutex,
8        atomic::{AtomicU64, Ordering},
9    },
10};
11
12use crate::observability::QueryAttemptReport;
13
14/// Process-lifetime counters for one backend's attempts. Each is monotonic
15/// and exact. `attempts_begun` is incremented under the observation permit
16/// that `begin_attempt` takes, and every begun attempt publishes one
17/// report, so `attempts_begun - reports_ready` is the number of attempts
18/// whose report is still pending.
19#[derive(Debug, Default)]
20pub struct AttemptBackendCounters {
21    pub(crate) attempts_begun: AtomicU64,
22    pub(crate) reports_ready: AtomicU64,
23    pub(crate) completion_queue_pushed: AtomicU64,
24    pub(crate) completion_queue_overflowed: AtomicU64,
25}
26
27/// Monotonic attempt and abandoned-report delivery counters for one backend.
28/// A report becomes ready after cleanup and sealing, independently of whether
29/// a caller retains a completion receiver.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub struct AttemptBackendCountersSnapshot {
32    /// Attempts created by this backend, including attempts rejected later.
33    pub attempts_begun: u64,
34    /// Attempts whose immutable terminal report has been published.
35    pub reports_ready: u64,
36    /// Reports routed into the abandoned-report queue with no active receiver.
37    pub completion_queue_pushed: u64,
38    /// Abandoned reports rejected because the configured queue was full.
39    pub completion_queue_overflowed: u64,
40}
41
42impl AttemptBackendCounters {
43    pub(crate) fn snapshot(&self) -> AttemptBackendCountersSnapshot {
44        AttemptBackendCountersSnapshot {
45            attempts_begun: self.attempts_begun.load(Ordering::Acquire),
46            reports_ready: self.reports_ready.load(Ordering::Acquire),
47            completion_queue_pushed: self.completion_queue_pushed.load(Ordering::Acquire),
48            completion_queue_overflowed: self.completion_queue_overflowed.load(Ordering::Acquire),
49        }
50    }
51}
52
53/// Reports whose attempt had no completion receiver when it published.
54///
55/// Routing is decided under the attempt lock at publication; this queue only
56/// stores. Capacity is fixed at construction, and a full queue rejects, which
57/// the publisher counts as `completion_queue_overflowed`.
58#[derive(Debug)]
59pub struct AttemptCompletionQueue {
60    capacity: usize,
61    queue: Mutex<VecDeque<Arc<QueryAttemptReport>>>,
62}
63
64impl AttemptCompletionQueue {
65    /// `None` for a zero capacity, so a disabled queue is an absent one.
66    pub fn new(capacity: usize) -> Option<Self> {
67        (capacity > 0).then(|| Self { capacity, queue: Mutex::new(VecDeque::new()) })
68    }
69
70    /// Returns `false` only when the queue is full; a contended lock waits.
71    pub(crate) fn try_push(&self, report: Arc<QueryAttemptReport>) -> bool {
72        // Blocking is bounded by one VecDeque push or drain. Waiting for the lock
73        // keeps a contended push from counting as a capacity overflow.
74        let mut queue = self.queue.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
75        if queue.len() == self.capacity {
76            false
77        } else {
78            queue.push_back(report);
79            true
80        }
81    }
82    pub(crate) fn drain(&self, max: usize) -> Vec<Arc<QueryAttemptReport>> {
83        let mut queue = self.queue.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
84        let n = max.min(queue.len());
85        queue.drain(..n).collect()
86    }
87
88    /// Holds the queue mutex so tests can prove that a contended push waits for
89    /// the lock and does not fail as if the queue were full.
90    #[cfg(test)]
91    pub(super) fn hold_lock_for_test(&self) -> MutexGuard<'_, VecDeque<Arc<QueryAttemptReport>>> {
92        self.queue.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
93    }
94}
query_runtime/attempt/queue.rs

query_runtime/attempt/
queue.rs
