From 743f898ae16e0834b43ab189cca3090cd38978d1 Mon Sep 17 00:00:00 2001 From: Anandh Somasundaram Date: Thu, 1 Oct 2026 18:37:06 +0000 Subject: [PATCH 1/3] fix(core): keep the metrics drain from holding the write lock over the points The flush calls MetricsCollector::drain once a minute. Its first phase held the collector's write lock while it partitioned every buffered data point (about 120 s of them) into old and new vectors. Every request records its points through the same lock, so the server served almost nothing for the whole partition: about 0.2 to 0.5 s at 16000 to 38000 GetItem/s on a 16 vCPU host, 7 points per request. The drain now takes the lock twice and copies no buffered point under it. The first step swaps the map out. The split by cutoff runs on the swapped-out map without the lock. The second step swaps the younger points back and appends the points recorded meanwhile, moving only their chunks. The drain takes and keeps the same points as before, in the same order. A mutex serializes drains; record and prune never take it. While the split runs, an in-memory query does not see the buffered points; the server reads its metrics from the store. Each key's points are kept in chunks of at most 1024, with the oldest and newest timestamp of each chunk. The split moves whole chunks and splits only the chunks that straddle the cutoff. A record call can now reallocate at most one chunk under the lock, not the whole vector of its key. The prune uses the same split, so its pass under the lock is now over chunks, not points. The flush buckets are summed per key, so a drained point no longer clones its key. Tests check that the write lock is free and every point is still unsplit when the split starts, that drains are serialized, and that the second step moves only the chunks recorded in between, the same with 4000 and 400000 points per key. They also compare the drained, kept, pruned, and aggregated points with a plain partition by cutoff, with points recorded out of order. Assisted-by: pi claude-opus-5-5 --- crates/core/src/metrics/accumulator.rs | 201 ++++++++++ crates/core/src/metrics/collector.rs | 97 +---- crates/core/src/metrics/collector_query.rs | 175 ++++++--- crates/core/src/metrics/drain_tests.rs | 407 +++++++++++++++++++++ crates/core/src/metrics/mod.rs | 1 + 5 files changed, 742 insertions(+), 139 deletions(-) create mode 100644 crates/core/src/metrics/accumulator.rs create mode 100644 crates/core/src/metrics/drain_tests.rs diff --git a/crates/core/src/metrics/accumulator.rs b/crates/core/src/metrics/accumulator.rs new file mode 100644 index 000000000..ffdec3344 --- /dev/null +++ b/crates/core/src/metrics/accumulator.rs @@ -0,0 +1,201 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Per-key point storage for `MetricsCollector`. +//! +//! Points are kept in chunks of at most [`CHUNK_CAP`]. Taking the old points +//! out moves whole chunks and splits only the chunks that straddle the cutoff, +//! and a `record` call reallocates at most one chunk. + +use std::collections::VecDeque; +use std::time::{Instant, SystemTime}; + +use super::collector::window_cutoff; +use super::types::TimeWindow; + +/// Most points one chunk holds: the largest reallocation a `record` call can +/// cause, and the most points a drain splits one by one per straddling chunk. +pub(super) const CHUNK_CAP: usize = 1024; + +/// Capacity of a key's first chunk; most keys never fill one. +const FIRST_CHUNK_CAP: usize = 64; + +/// A single data point recorded for a metric. +#[derive(Debug, Clone)] +pub(super) struct DataPoint { + pub(super) value: f64, + pub(super) timestamp: Instant, + /// Wall-clock time for DB persistence. Truncated to minute boundary on flush. + pub(super) wall_time: SystemTime, +} + +/// Points in record order, with the range of their timestamps. +#[derive(Debug)] +struct Chunk { + points: Vec, + oldest: Instant, + newest: Instant, +} + +impl Chunk { + fn new(point: DataPoint, capacity: usize) -> Self { + let mut points = Vec::with_capacity(capacity); + let (oldest, newest) = (point.timestamp, point.timestamp); + points.push(point); + Self { + points, + oldest, + newest, + } + } + + fn push(&mut self, point: DataPoint) { + self.oldest = self.oldest.min(point.timestamp); + self.newest = self.newest.max(point.timestamp); + self.points.push(point); + } + + /// Moves the points that `is_old` selects out of the chunk, keeping order. + fn split_off_old(&mut self, is_old: impl Fn(Instant) -> bool) -> Vec { + let (old, young): (Vec<_>, Vec<_>) = + self.points.drain(..).partition(|p| is_old(p.timestamp)); + self.points = young; + if let Some(first) = self.points.first() { + let start = (first.timestamp, first.timestamp); + (self.oldest, self.newest) = self.points.iter().fold(start, |(lo, hi), p| { + (lo.min(p.timestamp), hi.max(p.timestamp)) + }); + } + old + } +} + +/// Accumulator for a single metric+dimension combination. +#[derive(Debug)] +pub(super) struct Accumulator { + chunks: VecDeque, +} + +impl Accumulator { + pub(super) fn new() -> Self { + Self { + chunks: VecDeque::new(), + } + } + + pub(super) fn record(&mut self, value: f64, now: Instant, wall_time: SystemTime) { + let point = DataPoint { + value, + timestamp: now, + wall_time, + }; + match self.chunks.back_mut() { + Some(chunk) if chunk.points.len() < CHUNK_CAP => chunk.push(point), + Some(_) => self.chunks.push_back(Chunk::new(point, CHUNK_CAP)), + None => self.chunks.push_back(Chunk::new(point, FIRST_CHUNK_CAP)), + } + } + + pub(super) fn is_empty(&self) -> bool { + self.chunks.is_empty() + } + + /// All points, in record order. + pub(super) fn points(&self) -> impl Iterator { + self.chunks.iter().flat_map(|c| c.points.iter()) + } + + /// Number of chunks held. + pub(super) fn chunk_count(&self) -> usize { + self.chunks.len() + } + + /// Appends the points of `newer`, recorded after every point held here. + /// Moves chunk headers only: amortized O(chunks of `newer`). + pub(super) fn append(&mut self, mut newer: Accumulator) { + self.chunks.append(&mut newer.chunks); + } + + /// Removes and returns the points at or before `cutoff`, in record order. + pub(super) fn take_expired(&mut self, cutoff: Instant) -> Vec> { + self.take_old(|t| t <= cutoff) + } + + /// Prune points older than the retention window (1 day). + pub(super) fn prune(&mut self, cutoff: Instant) { + self.take_old(|t| t < cutoff); + } + + /// Moves out every point that `is_old` selects. Chunks whose newest point + /// is old move whole; only chunks holding both old and young points are + /// split. Points arrive almost in time order, so that is about one chunk. + fn take_old(&mut self, is_old: impl Fn(Instant) -> bool) -> Vec> { + let mut out = Vec::new(); + while self.chunks.front().is_some_and(|c| is_old(c.newest)) { + if let Some(chunk) = self.chunks.pop_front() { + out.push(chunk.points); + } + } + let mut emptied = false; + for chunk in self.chunks.iter_mut().filter(|c| is_old(c.oldest)) { + if is_old(chunk.newest) { + out.push(std::mem::take(&mut chunk.points)); + } else { + out.push(chunk.split_off_old(&is_old)); + } + emptied |= chunk.points.is_empty(); + } + if emptied { + self.chunks.retain(|c| !c.points.is_empty()); + } + out + } + + pub(super) fn snapshot(&self, window: TimeWindow, now: Instant) -> Option { + let cutoff = window_cutoff(window, now); + let values: Vec = match cutoff { + Some(c) => self + .points() + .filter(|p| p.timestamp >= c) + .map(|p| p.value) + .collect(), + None => self.points().map(|p| p.value).collect(), + }; + + if values.is_empty() { + return None; + } + + let sum: f64 = values.iter().sum(); + #[allow(clippy::cast_possible_truncation)] + let count = values.len() as u64; + let min = values.iter().copied().fold(f64::INFINITY, f64::min); + let max = values.iter().copied().fold(f64::NEG_INFINITY, f64::max); + + Some(AccumulatorSnapshot { + sum, + count, + min, + max, + values, + }) + } + + /// Chunk lengths and capacities, for tests of the storage bound. + #[cfg(test)] + pub(super) fn chunk_shapes(&self) -> Vec<(usize, usize)> { + self.chunks + .iter() + .map(|c| (c.points.len(), c.points.capacity())) + .collect() + } +} + +/// Snapshot of an accumulator's data for a given time window. +pub(super) struct AccumulatorSnapshot { + pub(super) sum: f64, + pub(super) count: u64, + pub(super) min: f64, + pub(super) max: f64, + pub(super) values: Vec, +} diff --git a/crates/core/src/metrics/collector.rs b/crates/core/src/metrics/collector.rs index 56d416a25..689b900a5 100755 --- a/crates/core/src/metrics/collector.rs +++ b/crates/core/src/metrics/collector.rs @@ -5,87 +5,12 @@ //! per-key aggregation. use std::collections::HashMap; -use std::sync::RwLock; +use std::sync::{Mutex, RwLock}; use std::time::{Duration, Instant, SystemTime}; +use super::accumulator::Accumulator; use super::types::{LatencySegments, MetricName, TimeWindow}; -/// A single data point recorded for a metric. -#[derive(Debug, Clone)] -pub(super) struct DataPoint { - pub(super) value: f64, - pub(super) timestamp: Instant, - /// Wall-clock time for DB persistence. Truncated to minute boundary on flush. - pub(super) wall_time: SystemTime, -} - -/// Accumulator for a single metric+dimension combination. -#[derive(Debug)] -pub(super) struct Accumulator { - pub(super) points: Vec, -} - -impl Accumulator { - fn new() -> Self { - Self { - points: Vec::with_capacity(64), - } - } - - fn record(&mut self, value: f64, now: Instant, wall_time: SystemTime) { - self.points.push(DataPoint { - value, - timestamp: now, - wall_time, - }); - } - - /// Prune points older than the retention window (1 day). - fn prune(&mut self, cutoff: Instant) { - self.points.retain(|p| p.timestamp >= cutoff); - } - - pub(super) fn snapshot(&self, window: TimeWindow, now: Instant) -> Option { - let cutoff = window_cutoff(window, now); - let values: Vec = match cutoff { - Some(c) => self - .points - .iter() - .filter(|p| p.timestamp >= c) - .map(|p| p.value) - .collect(), - None => self.points.iter().map(|p| p.value).collect(), - }; - - if values.is_empty() { - return None; - } - - let sum: f64 = values.iter().sum(); - #[allow(clippy::cast_possible_truncation)] - let count = values.len() as u64; - let min = values.iter().copied().fold(f64::INFINITY, f64::min); - let max = values.iter().copied().fold(f64::NEG_INFINITY, f64::max); - - Some(AccumulatorSnapshot { - sum, - count, - min, - max, - values, - }) - } -} - -/// Snapshot of an accumulator's data for a given time window. -pub(super) struct AccumulatorSnapshot { - pub(super) sum: f64, - pub(super) count: u64, - pub(super) min: f64, - pub(super) max: f64, - pub(super) values: Vec, -} - /// Returns `None` for `AllTime` (no cutoff), or `Some(cutoff)` for bounded windows. pub(super) fn window_cutoff(window: TimeWindow, now: Instant) -> Option { let dur = match window { @@ -115,6 +40,9 @@ pub struct MetricsCollector { pub(super) data: RwLock>, /// Per-operation latency segment breakdowns for the console deep-dive. pub(super) segments: RwLock>, + /// Serializes drains, so each one takes and keeps points in record order. + /// `record` and `prune` never take it. + pub(super) drain_lock: Mutex<()>, } /// A single latency segment data point with metadata. @@ -134,6 +62,7 @@ impl MetricsCollector { Self { data: RwLock::new(HashMap::new()), segments: RwLock::new(Vec::with_capacity(256)), + drain_lock: Mutex::new(()), } } @@ -152,8 +81,16 @@ impl MetricsCollector { index_name: index_name.map(ToOwned::to_owned), operation: operation.map(ToOwned::to_owned), }; - let now = Instant::now(); - let wall_time = SystemTime::now(); + self.record_point(key, value, Instant::now(), SystemTime::now()); + } + + pub(super) fn record_point( + &self, + key: MetricKey, + value: f64, + now: Instant, + wall_time: SystemTime, + ) { // Poisoned lock: silently skip rather than panic (async safety rule). let Ok(mut map) = self.data.write() else { return; @@ -409,7 +346,7 @@ impl MetricsCollector { for acc in map.values_mut() { acc.prune(cutoff); } - map.retain(|_, acc| !acc.points.is_empty()); + map.retain(|_, acc| !acc.is_empty()); drop(map); // Prune segment points too. if let Ok(mut segs) = self.segments.write() { diff --git a/crates/core/src/metrics/collector_query.rs b/crates/core/src/metrics/collector_query.rs index c932ccde3..143f7b30c 100755 --- a/crates/core/src/metrics/collector_query.rs +++ b/crates/core/src/metrics/collector_query.rs @@ -6,11 +6,14 @@ //! Split from `collector.rs` to keep both files under the 500-line limit. use std::collections::HashMap; +use std::collections::hash_map::Entry; use std::time::{Duration, Instant, SystemTime}; -use super::collector::{MetricsCollector, window_cutoff}; +use super::accumulator::{Accumulator, DataPoint}; +use super::collector::{MetricKey, MetricsCollector, window_cutoff}; use super::types::{ - Dimension, MetricName, MetricSnapshot, MetricsQuery, OperationSegments, Percentiles, TimeWindow, + Dimension, FlushBucket, MetricName, MetricSnapshot, MetricsQuery, OperationSegments, + Percentiles, TimeWindow, }; /// Accumulator for per-operation segment sums: (auth, authz, throttle, dispatch, response, total, count). @@ -18,6 +21,9 @@ type SegmentAccum = (f64, f64, f64, f64, f64, f64, u64); impl MetricsCollector { /// Query metrics and return snapshots. + /// + /// While a [`drain`](Self::drain) splits its points, the buffered points + /// are out of the map and this query does not see them. #[must_use] pub fn query(&self, params: &MetricsQuery) -> Vec { let now = Instant::now(); @@ -91,68 +97,44 @@ impl MetricsCollector { /// `FlushBucket`s for DB persistence. Points newer than `age` are kept /// in memory for the next flush cycle. /// - /// The write lock is held only for the partition step; aggregation into - /// buckets happens outside the lock to avoid blocking `record_*` calls. - pub fn drain(&self, age: Duration) -> Vec { + /// The drain takes the write lock twice: to swap the map out, and to put + /// the younger points back ahead of the points recorded meanwhile. Neither + /// step touches the points held, so `record_*` calls wait a bounded time + /// however many points are buffered. Between the two steps an in-memory + /// `query` does not see the buffered points; the server reads metrics + /// from its store. Concurrent drains run one at a time. The split and the + /// aggregation are CPU-bound: async callers should run the drain on a + /// blocking thread. + pub fn drain(&self, age: Duration) -> Vec { let cutoff = Instant::now().checked_sub(age).unwrap_or(Instant::now()); + aggregate(self.drain_points(cutoff, |_| {}).0) + } - // Phase 1: under write lock, partition old points out and collect them. - let drained: Vec<( - super::collector::MetricKey, - Vec, - )> = { + /// The points of a drain, and the chunks its second locked step moved. + /// `between` sees the map taken by the first locked step, before the + /// split, without the write lock. + pub(super) fn drain_points( + &self, + cutoff: Instant, + between: impl FnOnce(&PointMap), + ) -> (Drained, usize) { + let _serial = self + .drain_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut held = { let Ok(mut map) = self.data.write() else { - return Vec::new(); + return (Vec::new(), 0); }; - let mut result = Vec::new(); - for (key, acc) in map.iter_mut() { - let (old, new): (Vec<_>, Vec<_>) = - acc.points.drain(..).partition(|p| p.timestamp <= cutoff); - acc.points = new; - if !old.is_empty() { - result.push((key.clone(), old)); - } - } - map.retain(|_, acc| !acc.points.is_empty()); - result + std::mem::take(&mut *map) }; - - // Phase 2: aggregate outside the lock. - let mut buckets: HashMap<(super::collector::MetricKey, i64), (f64, u64, f64, f64)> = - HashMap::new(); - for (key, points) in drained { - for dp in points { - let minute = truncate_to_minute(dp.wall_time); - let e = buckets.entry((key.clone(), minute)).or_insert(( - 0.0, - 0, - f64::INFINITY, - f64::NEG_INFINITY, - )); - e.0 += dp.value; - e.1 += 1; - e.2 = e.2.min(dp.value); - e.3 = e.3.max(dp.value); - } - } - - buckets - .into_iter() - .map( - |((key, minute), (sum, count, min, max))| super::types::FlushBucket { - bucket: SystemTime::UNIX_EPOCH - + Duration::from_secs(u64::try_from(minute).unwrap_or(0)), - metric: key.metric, - table_name: key.table_name.unwrap_or_default(), - index_name: key.index_name.unwrap_or_default(), - operation: key.operation.unwrap_or_default(), - sum, - count, - min, - max, - }, - ) - .collect() + between(&held); + let drained = take_expired(&mut held, cutoff); + let Ok(mut map) = self.data.write() else { + return (drained, 0); + }; + let moved = reattach(&mut map, held); + (drained, moved) } /// Query average latency segments for the console deep-dive. @@ -238,3 +220,78 @@ fn truncate_to_minute(t: SystemTime) -> i64 { let secs_i64 = secs as i64; secs_i64 - (secs_i64 % 60) } + +/// Drained points per key, in record order. +type Drained = Vec<(MetricKey, Vec>)>; + +pub(super) type PointMap = HashMap; + +/// Removes every point at or before `cutoff` from `held`, which the caller +/// has taken out of the collector, and drops the keys left empty. +pub(super) fn take_expired(held: &mut PointMap, cutoff: Instant) -> Drained { + let mut drained = Vec::new(); + for (key, acc) in held.iter_mut() { + let old = acc.take_expired(cutoff); + if !old.is_empty() { + drained.push((key.clone(), old)); + } + } + held.retain(|_, acc| !acc.is_empty()); + drained +} + +/// The second locked step of a drain: makes `young` the map again and appends +/// the points recorded since the first step. Returns the chunks it moved, which +/// are only the chunks of those new points. +pub(super) fn reattach(map: &mut PointMap, young: PointMap) -> usize { + let recorded = std::mem::replace(map, young); + let mut moved = 0; + for (key, acc) in recorded { + moved += acc.chunk_count(); + match map.entry(key) { + Entry::Occupied(mut e) => e.get_mut().append(acc), + Entry::Vacant(e) => { + e.insert(acc); + } + } + } + moved +} + +/// Sums drained points into 1-minute buckets per key, in record order. +pub(super) fn aggregate(drained: Drained) -> Vec { + let mut out = Vec::new(); + for (key, chunks) in drained { + let mut by_minute: HashMap = HashMap::new(); + for dp in chunks.iter().flatten() { + let e = by_minute + .entry(truncate_to_minute(dp.wall_time)) + .or_insert((0.0, 0, f64::INFINITY, f64::NEG_INFINITY)); + e.0 += dp.value; + e.1 += 1; + e.2 = e.2.min(dp.value); + e.3 = e.3.max(dp.value); + } + out.extend( + by_minute + .into_iter() + .map(|(minute, (sum, count, min, max))| FlushBucket { + bucket: SystemTime::UNIX_EPOCH + + Duration::from_secs(u64::try_from(minute).unwrap_or(0)), + metric: key.metric, + table_name: key.table_name.clone().unwrap_or_default(), + index_name: key.index_name.clone().unwrap_or_default(), + operation: key.operation.clone().unwrap_or_default(), + sum, + count, + min, + max, + }), + ); + } + out +} + +#[cfg(test)] +#[path = "drain_tests.rs"] +mod tests; diff --git a/crates/core/src/metrics/drain_tests.rs b/crates/core/src/metrics/drain_tests.rs new file mode 100644 index 000000000..bc350541e --- /dev/null +++ b/crates/core/src/metrics/drain_tests.rs @@ -0,0 +1,407 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Tests for the drain: what it moves under the write lock, and that it drains +//! and keeps exactly the points a plain partition by cutoff would. + +use super::*; +use crate::metrics::accumulator::CHUNK_CAP; +use crate::metrics::types::{MetricsQuery, TimeWindow}; + +/// Small deterministic generator (xorshift64) for timestamp jitter. +struct Rng(u64); + +impl Rng { + fn next(&mut self) -> u64 { + self.0 ^= self.0 << 13; + self.0 ^= self.0 >> 7; + self.0 ^= self.0 << 17; + self.0 + } +} + +/// One recorded point, as the reference model sees it. +#[derive(Debug, Clone, Copy, PartialEq)] +struct Pt { + value: f64, + timestamp: Instant, + wall_time: SystemTime, +} + +impl From<&DataPoint> for Pt { + fn from(p: &DataPoint) -> Self { + Self { + value: p.value, + timestamp: p.timestamp, + wall_time: p.wall_time, + } + } +} + +fn keys(n: usize) -> Vec { + (0..n) + .map(|i| MetricKey { + metric: MetricName::SuccessfulRequestLatency, + table_name: Some(format!("table-{i}")), + index_name: None, + operation: Some("GetItem".to_owned()), + }) + .collect() +} + +/// Records `per_key` points for every key, interleaved across keys like the +/// request path does. Point `i` is at `base + i` microseconds, moved by up to +/// `jitter_us` either way, and its wall time advances 1 ms per point. +/// Returns the points per key in record order. +fn fill( + c: &MetricsCollector, + keys: &[MetricKey], + per_key: usize, + base: Instant, + jitter_us: u64, + rng: &mut Rng, +) -> Vec> { + let wall0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_800_000_000); + let mut recorded = vec![Vec::with_capacity(per_key); keys.len()]; + for i in 0..per_key { + for (k, key) in keys.iter().enumerate() { + let shift = if jitter_us == 0 { + 0 + } else { + rng.next() % (2 * jitter_us + 1) + }; + let micros = (i as u64 + jitter_us).saturating_sub(shift); + let p = Pt { + value: (i * keys.len() + k) as f64, + timestamp: base + Duration::from_micros(micros), + wall_time: wall0 + Duration::from_millis(i as u64), + }; + c.record_point(key.clone(), p.value, p.timestamp, p.wall_time); + recorded[k].push(p); + } + } + recorded +} + +fn drain_at(c: &MetricsCollector, cutoff: Instant) -> Drained { + c.drain_points(cutoff, |_| {}).0 +} + +fn kept(c: &MetricsCollector, key: &MetricKey) -> Vec { + let map = c.data.read().expect("lock"); + map.get(key) + .map(|acc| acc.points().map(Pt::from).collect()) + .unwrap_or_default() +} + +fn drained_of(drained: &Drained, key: &MetricKey) -> Vec { + drained + .iter() + .filter(|(k, _)| k == key) + .flat_map(|(_, chunks)| chunks.iter().flatten().map(Pt::from)) + .collect() +} + +/// Under the lock a drain moves no buffered point: the first step swaps the +/// map out and releases the lock before the split, and the second moves back +/// only the chunks recorded in between, the same count with 100 times more +/// points held. `moved` is reported by the code under test, so this proves the +/// structure of the second step, not its time. +#[test] +fn drain_moves_only_newly_recorded_chunks_under_the_lock() { + let keys = keys(7); + let mut moved_by_size = Vec::new(); + for per_key in [4_000, 400_000] { + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, per_key, base, 0, &mut Rng(1)); + let half = (per_key / 2) as u64; + let late = base + Duration::from_secs(10); + let fresh = MetricKey { + operation: Some("PutItem".to_owned()), + ..keys[0].clone() + }; + let wall = SystemTime::UNIX_EPOCH; + let (drained, moved) = c.drain_points(base + Duration::from_micros(half), |held| { + // Every point is still in the taken map (the split has not run) + // while the write lock is free and the collector's map is empty. + let unsplit: usize = held.values().map(|acc| acc.points().count()).sum(); + assert_eq!( + unsplit, + keys.len() * per_key, + "split ran before the write lock was released" + ); + assert!(c.data.try_write().is_ok(), "lock held before the split"); + assert!(c.data.read().expect("lock").is_empty()); + assert!(c.drain_lock.try_lock().is_err(), "drains not serialized"); + for v in [1.0, 2.0, 3.0] { + c.record_point(keys[0].clone(), -v, late, wall); + } + c.record_point(fresh.clone(), -4.0, late, wall); + }); + moved_by_size.push(moved); + + let drained_points: usize = drained + .iter() + .flat_map(|(_, chunks)| chunks.iter()) + .map(Vec::len) + .sum(); + assert_eq!(drained_points, keys.len() * (per_key / 2 + 1)); + // The points recorded during the drain follow the kept ones. + let mut want = recorded[0][per_key / 2 + 1..].to_vec(); + for v in [1.0, 2.0, 3.0] { + want.push(Pt { + value: -v, + timestamp: late, + wall_time: wall, + }); + } + assert_eq!(kept(&c, &keys[0]), want); + assert_eq!(kept(&c, &fresh).len(), 1); + assert_eq!(kept(&c, &keys[1]), recorded[1][per_key / 2 + 1..].to_vec()); + } + assert_eq!(moved_by_size, vec![2, 2], "chunks moved under the lock"); +} + +/// Points recorded out of order (the request path reads the clock before it +/// takes the lock) are drained and kept exactly as a partition would. +#[test] +fn drain_matches_a_partition_with_out_of_order_points() { + let keys = keys(3); + let per_key = 3 * CHUNK_CAP + 17; + for (seed, cut) in [(7, 0), (11, 500), (13, CHUNK_CAP), (17, 2_000), (19, 5_000)] { + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, per_key, base, 300, &mut Rng(seed)); + let cutoff = base + Duration::from_micros(cut as u64); + let drained = drain_at(&c, cutoff); + + for (k, key) in keys.iter().enumerate() { + let (old, young): (Vec, Vec) = + recorded[k].iter().partition(|p| p.timestamp <= cutoff); + assert_eq!(drained_of(&drained, key), old, "drained, cutoff {cut}"); + assert_eq!(kept(&c, key), young, "kept, cutoff {cut}"); + } + } +} + +/// A second drain at a later cutoff takes the next points, and a drain of +/// everything leaves no key behind. +#[test] +fn successive_drains_take_every_point_once() { + let keys = keys(2); + let per_key = 5 * CHUNK_CAP; + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, per_key, base, 50, &mut Rng(3)); + + let mut seen = vec![Vec::new(); keys.len()]; + for cut in [1_000_u64, 1_001, 2_500, 4_000, 10_000] { + let drained = drain_at(&c, base + Duration::from_micros(cut)); + for (k, key) in keys.iter().enumerate() { + seen[k].extend(drained_of(&drained, key)); + } + } + for k in 0..keys.len() { + let mut want = recorded[k].clone(); + want.sort_by_key(|p| p.timestamp); + seen[k].sort_by_key(|p| p.timestamp); + assert_eq!(seen[k], want); + } + assert!(c.data.read().expect("lock").is_empty()); +} + +type BucketRow = ( + String, + String, + String, + String, + SystemTime, + u64, + u64, + u64, + u64, +); + +fn rows(buckets: Vec) -> Vec { + let mut rows: Vec = buckets + .into_iter() + .map(|b| { + ( + b.metric.to_string(), + b.table_name, + b.index_name, + b.operation, + b.bucket, + b.sum.to_bits(), + b.count, + b.min.to_bits(), + b.max.to_bits(), + ) + }) + .collect(); + rows.sort_by(|a, b| (&a.0, &a.1, &a.2, &a.3, a.4).cmp(&(&b.0, &b.1, &b.2, &b.3, b.4))); + rows +} + +/// The flush buckets are the per-minute sum, count, min, and max of the +/// drained points, summed in record order. +#[test] +fn drain_buckets_match_a_per_minute_reference() { + let keys = keys(3); + let per_key = 150_000; // 150 s of wall time per key: three minute buckets + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, per_key, base, 200, &mut Rng(5)); + let cutoff = base + Duration::from_micros(100_000); + let drained = drain_at(&c, cutoff); + + let mut want = Vec::new(); + for (k, key) in keys.iter().enumerate() { + let mut by_minute: Vec<(i64, (f64, u64, f64, f64))> = Vec::new(); + for p in recorded[k].iter().filter(|p| p.timestamp <= cutoff) { + let minute = truncate_to_minute(p.wall_time); + if by_minute.last().is_none_or(|(m, _)| *m != minute) { + by_minute.push((minute, (0.0, 0, f64::INFINITY, f64::NEG_INFINITY))); + } + let e = &mut by_minute.last_mut().expect("bucket").1; + e.0 += p.value; + e.1 += 1; + e.2 = e.2.min(p.value); + e.3 = e.3.max(p.value); + } + for (minute, (sum, count, min, max)) in by_minute { + want.push(FlushBucket { + bucket: SystemTime::UNIX_EPOCH + Duration::from_secs(minute as u64), + metric: key.metric, + table_name: key.table_name.clone().unwrap_or_default(), + index_name: String::new(), + operation: key.operation.clone().unwrap_or_default(), + sum, + count, + min, + max, + }); + } + } + assert_eq!(rows(aggregate(drained)), rows(want)); +} + +/// `drain(ZERO)`, the shutdown flush, drains every point recorded before it. +#[test] +fn drain_zero_takes_everything() { + let c = MetricsCollector::new(); + for i in 0..(2 * CHUNK_CAP + 5) { + c.record_latency(Some("T"), "GetItem", i as f64); + c.record_request_count("GetItem"); + } + let buckets = c.drain(Duration::ZERO); + let count: u64 = buckets.iter().map(|b| b.count).sum(); + assert_eq!(count, 2 * (2 * CHUNK_CAP as u64 + 5)); + assert!(c.data.read().expect("lock").is_empty()); +} + +/// `record` never grows a chunk past `CHUNK_CAP`, so its largest reallocation +/// under the lock is one chunk, not every point of the key. +#[test] +fn record_keeps_every_chunk_within_the_cap() { + let c = MetricsCollector::new(); + for i in 0..(10 * CHUNK_CAP + 3) { + c.record_latency(Some("T"), "GetItem", i as f64); + } + let map = c.data.read().expect("lock"); + let acc = map.values().next().expect("one key"); + let shapes = acc.chunk_shapes(); + assert_eq!(shapes.len(), 11); + for (len, capacity) in &shapes { + assert!( + *len <= CHUNK_CAP && *capacity <= CHUNK_CAP, + "{len}/{capacity}" + ); + } + let values: Vec = acc.points().map(|p| p.value).collect(); + let want: Vec = (0..(10 * CHUNK_CAP + 3)).map(|i| i as f64).collect(); + assert_eq!(values, want, "points out of record order"); +} + +/// The prune keeps exactly the points at or after its cutoff. +#[test] +fn prune_matches_a_retain_with_out_of_order_points() { + let keys = keys(2); + let per_key = 4 * CHUNK_CAP; + for cut in [0_u64, 777, 2_048, 3_333, 9_999] { + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, per_key, base, 100, &mut Rng(23)); + let cutoff = base + Duration::from_micros(cut); + for acc in c.data.write().expect("lock").values_mut() { + acc.prune(cutoff); + } + for (k, key) in keys.iter().enumerate() { + let want: Vec = recorded[k] + .iter() + .copied() + .filter(|p| p.timestamp >= cutoff) + .collect(); + assert_eq!(kept(&c, key), want, "cutoff {cut}"); + } + } +} + +/// An in-memory query after a drain reports the same sum, count, min, and +/// max as the kept points, summed in record order. +#[test] +fn query_after_drain_reports_the_kept_points() { + let keys = keys(1); + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, 3 * CHUNK_CAP, base, 40, &mut Rng(29)); + let cutoff = base + Duration::from_micros(1_500); + drain_at(&c, cutoff); + + let young: Vec = recorded[0] + .iter() + .filter(|p| p.timestamp > cutoff) + .map(|p| p.value) + .collect(); + let snaps = c.query(&MetricsQuery { + window: Some(TimeWindow::AllTime), + ..Default::default() + }); + assert_eq!(snaps.len(), 1); + let s = &snaps[0]; + assert_eq!(s.count, young.len() as u64); + assert_eq!(s.sum.to_bits(), young.iter().sum::().to_bits()); + assert_eq!(s.min, young.iter().copied().fold(f64::INFINITY, f64::min)); + assert_eq!( + s.max, + young.iter().copied().fold(f64::NEG_INFINITY, f64::max) + ); +} + +/// A fully old chunk behind a chunk that holds a young point is taken whole, +/// and the result still matches a partition. +#[test] +fn drain_takes_a_fully_old_chunk_behind_a_young_point() { + let key = keys(1).remove(0); + let c = MetricsCollector::new(); + let base = Instant::now(); + let wall = SystemTime::UNIX_EPOCH; + let mut recorded = Vec::new(); + for i in 0..(2 * CHUNK_CAP) { + // One late-read young point at the head of the first chunk. + let micros = if i == 0 { 10_000 } else { i as u64 }; + let p = Pt { + value: i as f64, + timestamp: base + Duration::from_micros(micros), + wall_time: wall, + }; + c.record_point(key.clone(), p.value, p.timestamp, p.wall_time); + recorded.push(p); + } + let cutoff = base + Duration::from_micros(5_000); + let drained = drain_at(&c, cutoff); + let (old, young): (Vec, Vec) = recorded.iter().partition(|p| p.timestamp <= cutoff); + assert_eq!(drained_of(&drained, &key), old); + assert_eq!(kept(&c, &key), young); +} diff --git a/crates/core/src/metrics/mod.rs b/crates/core/src/metrics/mod.rs index 4db7c40fd..bf1ff0272 100755 --- a/crates/core/src/metrics/mod.rs +++ b/crates/core/src/metrics/mod.rs @@ -10,6 +10,7 @@ //! //! REQ-OBS-005: Metrics endpoint. +mod accumulator; mod collector; mod collector_query; mod types; From b6f4f8ccb5f61f476036579971fde052ab602a4b Mon Sep 17 00:00:00 2001 From: Anandh Somasundaram Date: Thu, 1 Oct 2026 18:37:06 +0000 Subject: [PATCH 2/3] fix(server): run the metrics drain on a blocking thread The flush worker called MetricsCollector::drain inline on a tokio worker thread. The drain splits and aggregates every point it takes, 0.2 to 0.7 s of CPU at 10000 to 39000 GetItem/s, and that worker runs no other task meanwhile. Run the drain with spawn_blocking. A drain task that fails is logged and counted as a worker error for the cycle. Assisted-by: pi claude-opus-5-5 --- crates/server/src/workers.rs | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/crates/server/src/workers.rs b/crates/server/src/workers.rs index 372fdb45a..f519f91c3 100755 --- a/crates/server/src/workers.rs +++ b/crates/server/src/workers.rs @@ -198,7 +198,19 @@ pub(crate) async fn metrics_flush_worker( Duration::ZERO }; let cycle_start = std::time::Instant::now(); - let buckets = metrics.drain(drain_age); + // The drain aggregates every point it takes: keep it off the runtime + // workers so they keep polling sockets while it runs. + let drain_metrics = Arc::clone(&metrics); + let mut errored = false; + let buckets = + match tokio::task::spawn_blocking(move || drain_metrics.drain(drain_age)).await { + Ok(buckets) => buckets, + Err(e) => { + tracing::warn!("Metrics drain task failed: {e}"); + errored = true; + Vec::new() + } + }; if !buckets.is_empty() { let rows: Vec = buckets .iter() @@ -240,13 +252,13 @@ pub(crate) async fn metrics_flush_worker( let _ = store.insert_metrics(&rows).await; } // Prune old DB rows. - let mut errored = false; if let Err(e) = store.prune_metrics(RETENTION).await { tracing::warn!("Failed to prune old metrics from DB: {e:?}"); - metrics.record_worker_error(QuerySource::MetricsFlush); errored = true; } - if !errored { + if errored { + metrics.record_worker_error(QuerySource::MetricsFlush); + } else { #[allow(clippy::cast_precision_loss)] let cycle_us = cycle_start.elapsed().as_micros() as f64; metrics.record_worker_success(QuerySource::MetricsFlush, cycle_us); From 30102b3229ff2d074b40652abaf2938aa6f99d30 Mon Sep 17 00:00:00 2001 From: Anandh Somasundaram Date: Fri, 9 Oct 2026 15:27:31 +0000 Subject: [PATCH 3/3] test(core): check the drain lock state after the split too The drain test checked the write lock and the drain mutex only before the split. A change that took the write lock again around the split, or released the drain mutex right after that check, still passed. A second check after the split now asserts that the write lock is still free and the drain mutex is still held, and records a point there that must land after the kept ones. Both changes now fail the test. The drain doc said concurrent drains run one at a time, but the aggregation runs after the mutex is released. It now says that the swap, split, and put-back are what run one at a time. Both from review by @robinnsc. Assisted-by: pi claude-opus-5-5 --- crates/core/src/metrics/collector_query.rs | 15 +++--- crates/core/src/metrics/drain_tests.rs | 61 ++++++++++++++-------- 2 files changed, 48 insertions(+), 28 deletions(-) diff --git a/crates/core/src/metrics/collector_query.rs b/crates/core/src/metrics/collector_query.rs index 143f7b30c..7ce9af6ff 100755 --- a/crates/core/src/metrics/collector_query.rs +++ b/crates/core/src/metrics/collector_query.rs @@ -102,21 +102,23 @@ impl MetricsCollector { /// step touches the points held, so `record_*` calls wait a bounded time /// however many points are buffered. Between the two steps an in-memory /// `query` does not see the buffered points; the server reads metrics - /// from its store. Concurrent drains run one at a time. The split and the + /// from its store. The swap, split, and put-back of concurrent drains run + /// one at a time; their aggregation can overlap. The split and the /// aggregation are CPU-bound: async callers should run the drain on a /// blocking thread. pub fn drain(&self, age: Duration) -> Vec { let cutoff = Instant::now().checked_sub(age).unwrap_or(Instant::now()); - aggregate(self.drain_points(cutoff, |_| {}).0) + aggregate(self.drain_points(cutoff, |_| {}, |_| {}).0) } /// The points of a drain, and the chunks its second locked step moved. - /// `between` sees the map taken by the first locked step, before the - /// split, without the write lock. + /// `before_split` and `after_split` see the map taken by the first locked + /// step on each side of the split, without the write lock. pub(super) fn drain_points( &self, cutoff: Instant, - between: impl FnOnce(&PointMap), + before_split: impl FnOnce(&PointMap), + after_split: impl FnOnce(&PointMap), ) -> (Drained, usize) { let _serial = self .drain_lock @@ -128,8 +130,9 @@ impl MetricsCollector { }; std::mem::take(&mut *map) }; - between(&held); + before_split(&held); let drained = take_expired(&mut held, cutoff); + after_split(&held); let Ok(mut map) = self.data.write() else { return (drained, 0); }; diff --git a/crates/core/src/metrics/drain_tests.rs b/crates/core/src/metrics/drain_tests.rs index bc350541e..2945d339c 100644 --- a/crates/core/src/metrics/drain_tests.rs +++ b/crates/core/src/metrics/drain_tests.rs @@ -84,7 +84,7 @@ fn fill( } fn drain_at(c: &MetricsCollector, cutoff: Instant) -> Drained { - c.drain_points(cutoff, |_| {}).0 + c.drain_points(cutoff, |_| {}, |_| {}).0 } fn kept(c: &MetricsCollector, key: &MetricKey) -> Vec { @@ -103,10 +103,11 @@ fn drained_of(drained: &Drained, key: &MetricKey) -> Vec { } /// Under the lock a drain moves no buffered point: the first step swaps the -/// map out and releases the lock before the split, and the second moves back -/// only the chunks recorded in between, the same count with 100 times more -/// points held. `moved` is reported by the code under test, so this proves the -/// structure of the second step, not its time. +/// map out and releases the lock before the split, the lock stays free and the +/// drain mutex stays held until the split is done, and the second step moves +/// back only the chunks recorded in between, the same count with 100 times +/// more points held. `moved` is reported by the code under test, so this +/// proves the structure of the second step, not its time. #[test] fn drain_moves_only_newly_recorded_chunks_under_the_lock() { let keys = keys(7); @@ -122,23 +123,39 @@ fn drain_moves_only_newly_recorded_chunks_under_the_lock() { ..keys[0].clone() }; let wall = SystemTime::UNIX_EPOCH; - let (drained, moved) = c.drain_points(base + Duration::from_micros(half), |held| { - // Every point is still in the taken map (the split has not run) - // while the write lock is free and the collector's map is empty. - let unsplit: usize = held.values().map(|acc| acc.points().count()).sum(); - assert_eq!( - unsplit, - keys.len() * per_key, - "split ran before the write lock was released" - ); - assert!(c.data.try_write().is_ok(), "lock held before the split"); - assert!(c.data.read().expect("lock").is_empty()); - assert!(c.drain_lock.try_lock().is_err(), "drains not serialized"); - for v in [1.0, 2.0, 3.0] { - c.record_point(keys[0].clone(), -v, late, wall); - } - c.record_point(fresh.clone(), -4.0, late, wall); - }); + let (drained, moved) = c.drain_points( + base + Duration::from_micros(half), + |held| { + // Every point is still in the taken map (the split has not + // run) while the write lock is free and the collector's map is + // empty. + let unsplit: usize = held.values().map(|acc| acc.points().count()).sum(); + assert_eq!( + unsplit, + keys.len() * per_key, + "split ran before the write lock was released" + ); + assert!(c.data.try_write().is_ok(), "lock held before the split"); + assert!(c.data.read().expect("lock").is_empty()); + assert!(c.drain_lock.try_lock().is_err(), "drains not serialized"); + for v in [1.0, 2.0] { + c.record_point(keys[0].clone(), -v, late, wall); + } + c.record_point(fresh.clone(), -4.0, late, wall); + }, + |held| { + // The split is done: only the younger points are left. The + // write lock is still free and the drain mutex still held. + let young: usize = held.values().map(|acc| acc.points().count()).sum(); + assert_eq!(young, keys.len() * (per_key - per_key / 2 - 1)); + assert!(c.data.try_write().is_ok(), "lock held over the split"); + assert!( + c.drain_lock.try_lock().is_err(), + "drain mutex released before the split ended" + ); + c.record_point(keys[0].clone(), -3.0, late, wall); + }, + ); moved_by_size.push(moved); let drained_points: usize = drained