From fb68eaef2b808964e9f936062c7af8f3bbc7fea7 Mon Sep 17 00:00:00 2001 From: fernandodeluret Date: Wed, 26 Aug 2026 02:15:11 +0000 Subject: [PATCH 1/2] `feat` Make cache not point to tcp chunks --- crates/api/src/http/streaming.rs | 18 ++++++++++++++---- crates/api/src/modules/cache/mod.rs | 10 ++++++---- 2 files changed, 20 insertions(+), 8 deletions(-) diff --git a/crates/api/src/http/streaming.rs b/crates/api/src/http/streaming.rs index dd1c0e2..d0964b5 100644 --- a/crates/api/src/http/streaming.rs +++ b/crates/api/src/http/streaming.rs @@ -310,10 +310,20 @@ fn drain_pending_into_cache( Vec::with_capacity(pending_fresh.len() + pending_cached.len()); new_pairs.append(pending_cached); for (pk, range) in pending_fresh.drain(..) { - // `Bytes::slice` is O(1): pointer math + atomic refcount inc on - // the same allocation backing `frozen`. The slice keeps that - // allocation alive for as long as the cache entry holds it. - new_pairs.push((pk, frozen.slice(range))); + // Copy the freshly-serialized account into its own tight allocation + // instead of retaining a `frozen.slice(range)`. A slice would be O(1) + // but keeps the *entire* ~64 KB streaming chunk (`STREAM_BUFFER_PREALLOC`) + // alive for as long as the cache entry holds it, since `Bytes` frees the + // backing allocation only when its last view drops. Under sustained GPA + // caching where each refresh re-serializes only a few accounts, those + // few fresh accounts would each pin a whole chunk, so the resident cache + // grows far beyond the accounted `GpaCache::size` (which only counts + // `bytes.len()`). Copying costs one memcpy per newly-cached account but + // makes retained memory track the accounted size. Cache hits stay + // zero-copy: they reuse the compact `Bytes` already held in the entry + // (carried through `pending_cached` above), and the response body is + // still written from `frozen` directly, so this does not affect it. + new_pairs.push((pk, Bytes::copy_from_slice(&frozen[range]))); } gpa_processor.update_new_accounts_for_query(new_pairs, cache_hits); diff --git a/crates/api/src/modules/cache/mod.rs b/crates/api/src/modules/cache/mod.rs index 2c7df40..6a23666 100644 --- a/crates/api/src/modules/cache/mod.rs +++ b/crates/api/src/modules/cache/mod.rs @@ -349,10 +349,12 @@ impl GpaProcessor { #[derive(Debug, Clone)] pub struct CachedQuery { /// JSON-encoded account bytes keyed by pubkey. Stored as `Bytes` so that - /// on a future cache hit we can append the slice directly into the next - /// response's `BytesMut` (just a memcpy) with no re-serialization, and so - /// that fresh slices coming out of the streaming pipeline share storage - /// with the response chunks they were carved out of. + /// on a future cache hit we can append it directly into the next response's + /// `BytesMut` (just a memcpy) with no re-serialization. Each `Bytes` owns a + /// tight, per-account allocation: freshly-serialized accounts are copied out + /// of the shared streaming chunk when inserted (see `drain_pending_into_cache` + /// in `http::streaming`) so a cache entry never pins a whole ~64 KB chunk, + /// which keeps retained memory in line with the accounted [`Self::size`]. pub accounts: Arc>, /// Slot for which the cached query was served. pub slot: u64, From 7816c8a5da20a891b5a222dc6eaf6cd53bf29653 Mon Sep 17 00:00:00 2001 From: fernandodeluret Date: Wed, 26 Aug 2026 02:17:04 +0000 Subject: [PATCH 2/2] `feat` Move cache finalize_query outside the main request task, make locked time not do any deallocation, optimize path for small not cached requests --- crates/api/src/http/streaming.rs | 4 +- crates/api/src/metrics.rs | 55 +++++ crates/api/src/modules/cache/mod.rs | 332 +++++++++++++++++++--------- crates/snapshot/src/sidecar.rs | 3 +- 4 files changed, 290 insertions(+), 104 deletions(-) diff --git a/crates/api/src/http/streaming.rs b/crates/api/src/http/streaming.rs index d0964b5..0286db6 100644 --- a/crates/api/src/http/streaming.rs +++ b/crates/api/src/http/streaming.rs @@ -226,10 +226,8 @@ pub async fn gpa_streaming_response_body( json_span.record("json_bytes", json_bytes as i64); json_span.record("total_wall_time", gpa_global_start_time.elapsed().as_millis() as i64); - // Commit the accumulated `(pubkey, bytes)` pairs as the new cached query. - let finalize_query_start_time = Instant::now(); + // Hand the accumulated `(pubkey, bytes)` pairs to the background finalize thread json_span.in_scope(|| gpa_processor.finalize_query()); - metrics::CLOUDBREAK_API_REQUEST_DURATION_MS.with_label_values(&["cache_finalize_query", gpa_processor.get_type()]).observe(finalize_query_start_time.elapsed().as_millis() as f64); // Close the JSON array yield Ok(Frame::data(Bytes::from(streaming_response_body_wrapper.end))); diff --git a/crates/api/src/metrics.rs b/crates/api/src/metrics.rs index 43cd7e8..1b0e1f3 100644 --- a/crates/api/src/metrics.rs +++ b/crates/api/src/metrics.rs @@ -158,6 +158,58 @@ lazy_static::lazy_static! { ), &["used"], ).unwrap(); + + /// Queries accepted for caching that are queued on, or running on, the + /// blocking pool. Expected to sit near 0; a sustained value means insertion + /// is falling behind the requests producing it, which delays entries becoming + /// visible and therefore lowers the hit rate. + pub static ref CLOUDBREAK_GPA_CACHE_FINALIZE_INFLIGHT_JOBS: IntGauge = IntGauge::new( + "cloudbreak_gpa_cache_finalize_inflight_jobs", + "GPA queries currently queued or running on the blocking pool waiting to be inserted into the cache" + ).unwrap(); + + /// Payload bytes held by the in-flight finalize jobs above. This memory is + /// resident but not yet accounted in `cloudbreak_gpa_cache_size_bytes`, so it + /// is the amount by which that gauge under-reports the cache's real footprint. + pub static ref CLOUDBREAK_GPA_CACHE_FINALIZE_INFLIGHT_BYTES: IntGauge = IntGauge::new( + "cloudbreak_gpa_cache_finalize_inflight_bytes", + "Payload bytes of GPA queries currently queued or running on the blocking pool waiting to be inserted into the cache" + ).unwrap(); + + /// Queries that reached the finalize job but were not inserted, labelled by + /// `reason`: `stale_slot` (a newer version of the query was already cached by + /// the time the job ran) or `cleanup_failed` (could not free enough space). + /// A non-trivial `stale_slot` rate means finalize is running too far behind + /// the requests that produced it. + pub static ref CLOUDBREAK_GPA_CACHE_FINALIZE_SKIPPED_TOTAL: IntCounterVec = IntCounterVec::new( + Opts::new( + "cloudbreak_gpa_cache_finalize_skipped_total", + "GPA queries dropped by the finalize job without being cached, labelled by reason" + ), + &["reason"], + ).unwrap(); +} + +/// Tracks a GPA cache insertion while it is queued on, or running on, the +/// blocking pool. Follows the same guard pattern as [`InFlightRequestGuard`] so +/// the gauges are corrected even if the job panics. +pub struct FinalizeInFlightGuard { + bytes: i64, +} + +impl FinalizeInFlightGuard { + pub fn new(bytes: i64) -> Self { + CLOUDBREAK_GPA_CACHE_FINALIZE_INFLIGHT_JOBS.inc(); + CLOUDBREAK_GPA_CACHE_FINALIZE_INFLIGHT_BYTES.add(bytes); + Self { bytes } + } +} + +impl Drop for FinalizeInFlightGuard { + fn drop(&mut self) { + CLOUDBREAK_GPA_CACHE_FINALIZE_INFLIGHT_JOBS.dec(); + CLOUDBREAK_GPA_CACHE_FINALIZE_INFLIGHT_BYTES.sub(self.bytes); + } } /// We use a guard to increment the in-flight requests metric when a request starts and @@ -241,6 +293,9 @@ pub fn setup_metrics(config: &ApiConfig) -> anyhow::Result<()> { register!(CLOUDBREAK_GPA_CACHE_MAX_BYTES); register!(CLOUDBREAK_GPA_CACHE_EVICTIONS_TOTAL); register!(CLOUDBREAK_GPA_CACHE_EVICTED_BYTES_TOTAL); + register!(CLOUDBREAK_GPA_CACHE_FINALIZE_INFLIGHT_JOBS); + register!(CLOUDBREAK_GPA_CACHE_FINALIZE_INFLIGHT_BYTES); + register!(CLOUDBREAK_GPA_CACHE_FINALIZE_SKIPPED_TOTAL); // Per-client-IP egress bandwidth (peak gauge + throughput histogram). // Optional, off by default: registers its collectors and starts the 1s diff --git a/crates/api/src/modules/cache/mod.rs b/crates/api/src/modules/cache/mod.rs index 6a23666..264002a 100644 --- a/crates/api/src/modules/cache/mod.rs +++ b/crates/api/src/modules/cache/mod.rs @@ -21,13 +21,190 @@ use std::collections::HashMap; use std::sync::Arc; use std::sync::Mutex; use std::sync::RwLock; -use tokio::time::Instant; +use tracing_opentelemetry::OpenTelemetrySpanExt; use crate::error::RpcError; use crate::methods::program; use crate::methods::program::GpaDbQueryInput; use crate::metrics; +/// A query that has been accepted for caching, carrying everything needed to +/// install it. Built on the request path, executed on the blocking pool by +/// [`FinalizeJob::run`]. +struct FinalizeJob { + cache: Arc>, + normalized_query: NormalizedQuery, + accounts: Vec<(Pubkey, Bytes)>, + query_bytes: u64, + slot: u64, + cache_hits: u64, + /// The entry this request read from, if any. Carried so that the worker — + /// not the request — holds the last reference to the map it is about to + /// replace, and therefore pays to free it. See [`FinalizeJob::run`]. + previous_query: Option, + /// Trace context of the request that produced this query, so the worker's + /// span still hangs off the originating trace. + parent_cx: opentelemetry::Context, +} + +impl FinalizeJob { + /// Hands the insertion to the blocking pool and returns immediately. + /// + /// Installing a query is `O(accounts)`: building the account map, taking the + /// cache write lock, and deallocating the version being replaced. The client + /// is not waiting on any of it, so none of it belongs on the request path. + /// The join handle is dropped on purpose — there is nothing to await, and + /// losing an insertion is harmless because the next request repopulates it. + fn spawn(self) { + let inflight = metrics::FinalizeInFlightGuard::new(self.query_bytes as i64); + + tokio::task::spawn_blocking(move || { + let started = std::time::Instant::now(); + + self.run(); + + // Timed out here so every exit path in `run` is counted. + metrics::CLOUDBREAK_API_REQUEST_DURATION_MS + .with_label_values(&["cache_finalize_query", "cached"]) + .observe(started.elapsed().as_micros() as f64); + + // Also what moves the guard into the closure: without this the guard + // would drop as soon as `spawn` returns and the gauges would report + // nothing as in flight. + drop(inflight); + }); + } + + /// Installs the query in the cache. Runs on the blocking pool, never on a + /// request. + /// + /// The declaration order of `new_entry`, `replaced` and `cache_guard` is + /// load-bearing: Rust drops locals in reverse order, so the write lock is + /// always released before any account map is deallocated, on every exit path + /// including the early returns. + fn run(self) { + let start_time = std::time::Instant::now(); + let Self { + cache, + normalized_query, + accounts, + query_bytes, + slot, + cache_hits, + previous_query, + parent_cx, + } = self; + + let span = tracing::info_span!( + "gpa_cache_finalize_query", + cache_hits = cache_hits as i64, + query_bytes = query_bytes as i64, + query_accounts = accounts.len() as i64, + wall_time = tracing::field::Empty, + locked_micros = tracing::field::Empty, + ); + // Must happen before `enter()`: `set_parent` rejects a span that has + // already been started. Errors only mean there is no OpenTelemetry layer + // or the span was filtered out, in which case there is no trace to attach + // to anyway, so they are ignored (as in the crate's own examples). + let _ = span.set_parent(parent_cx); + let _span_guard = span.enter(); + + let new_entry = CachedQuery { + accounts: Arc::new(accounts.into_iter().collect()), + slot, + size: query_bytes, + cache_hits, + }; + + // Versions of this query that are no longer reachable from the cache. + // `previous_query` goes in first so this job holds the last reference to + // the map being replaced. + let mut replaced: Vec = Vec::new(); + replaced.extend(previous_query); + + let start_locked_time = std::time::Instant::now(); + let mut cache_guard = cache.write().expect("can't lock gpa cache rwlock"); + let lock_held_start = std::time::Instant::now(); + + // This runs behind the request that produced the query, so a newer + // version may already be cached. Installing an older snapshot would be + // internally consistent but would force the next request to refresh more + // accounts, so discard this one instead. + if cache_guard + .queries + .get(&normalized_query) + .is_some_and(|current| current.slot >= slot) + { + metrics::CLOUDBREAK_GPA_CACHE_FINALIZE_SKIPPED_TOTAL + .with_label_values(&["stale_slot"]) + .inc(); + return; + } + + // Cleanup cache if needed + if let Some(bytes_freed) = cache_guard.cleanup_old_queries(query_bytes, &mut replaced) + && bytes_freed < query_bytes + { + tracing::error!(target: "gpa_cache", "Failed to cleanup old queries, not enough bytes freed {}", query_bytes - bytes_freed); + metrics::CLOUDBREAK_GPA_CACHE_FINALIZE_SKIPPED_TOTAL + .with_label_values(&["cleanup_failed"]) + .inc(); + return; + } + + // Insert the query into the main map (replacing the older query if existed) + let older_query = cache_guard + .queries + .insert(normalized_query.clone(), new_entry); + + // Update map size, crediting back the bytes of the query we just + // replaced (if any) so the counter tracks what is actually held. + cache_guard.size += query_bytes; + if let Some(older_query) = &older_query { + cache_guard.size = cache_guard.size.saturating_sub(older_query.size); + } + + // Mirror the same accounting for pinned bytes: credit the new query if + // it is pinned, and credit back the replaced query if it was pinned. + if cache_guard.is_pinned_size(query_bytes) { + cache_guard.pinned_size += query_bytes; + } + if let Some(older_query) = &older_query + && cache_guard.is_pinned_size(older_query.size) + { + cache_guard.pinned_size = cache_guard.pinned_size.saturating_sub(older_query.size); + } + + cache_guard.insert_query_for_slot( + normalized_query, + slot, + older_query.as_ref().map(|q| q.slot), + ); + replaced.extend(older_query); + + cache_guard.update_size_metrics(); + + // `finalize_query_locked` spans lock acquisition plus the critical + // section, so it also reflects time spent queueing behind other + // writers. `finalize_query_held` isolates the critical section itself. + let locked_micros = start_locked_time.elapsed().as_micros() as i64; + let held_micros = lock_held_start.elapsed().as_micros() as i64; + metrics::CLOUDBREAK_API_REQUEST_DURATION_MS + .with_label_values(&["finalize_query_locked", "cached"]) + .observe(locked_micros as f64); + metrics::CLOUDBREAK_API_REQUEST_DURATION_MS + .with_label_values(&["finalize_query_held", "cached"]) + .observe(held_micros as f64); + + drop(cache_guard); + drop(replaced); + + span.record("wall_time", start_time.elapsed().as_millis() as i64); + span.record("locked_micros", locked_micros); + } +} + #[derive(Debug, Clone)] pub struct GpaCache { /// Map of queries by their key, and stores the slot for which the query @@ -225,124 +402,68 @@ impl GpaProcessor { } } - /// Commit the accumulated `(pubkey, bytes)` pairs as the new `CachedQuery` + /// Hand the accumulated `(pubkey, bytes)` pairs to the background finalize + /// thread, which commits them as the new `CachedQuery`. /// /// If the GpaProcessor is `Standard`, this is a no-op. /// - /// It will only add the query to the cache if the query is larger than the - /// `config.min_bytes_per_query`. - /// - /// If the insertion gets the cache size above the `config.max_total_bytes`, - /// it will trigger the cache cleanup of oldest queries to ensure the cache - /// size stays within the configured limit . + /// Only queries larger than `config.min_bytes_per_query` are cached, and that + /// is decided here rather than on the worker: the overwhelming majority of + /// queries fall below the threshold, and rejecting them costs a single pass + /// over the accumulated pairs. Everything expensive — building the account + /// map, taking the cache write lock, cleaning up old queries to stay within + /// `config.max_total_bytes`, and freeing the replaced version — happens on + /// the blocking pool, off the request path. See [`FinalizeJob::run`]. pub fn finalize_query(&mut self) { - let start_time = Instant::now(); let Self::Cached { cache, + cached_query, normalized_query, new_accounts_for_query, new_slot, cache_hits, - cached_query: _, } = self else { return; }; - let finalize_query_span = tracing::info_span!( - "gpa_cache_finalize_query", - cache_hits = tracing::field::Empty, - query_bytes = tracing::field::Empty, - query_accounts = tracing::field::Empty, - wall_time = tracing::field::Empty, - locked_micros = tracing::field::Empty, - ); - let Some(normalized_query) = normalized_query.take() else { tracing::error!(target: "gpa_cache", "No normalized query found"); return; }; - let new_accounts_for_query = std::mem::take( + let accounts = std::mem::take( &mut *new_accounts_for_query .lock() .expect("new_accounts_for_query mutex poisoned"), ); - let mut query_bytes = 0; - let new_accounts_for_query_len = new_accounts_for_query.len(); - let accounts: HashMap = new_accounts_for_query - .into_iter() - .map(|(pubkey, bytes)| { - query_bytes += bytes.len() as u64; - (pubkey, bytes) - }) - .collect(); - - let new_entry = CachedQuery { - accounts: Arc::new(accounts), - slot: *new_slot, - size: query_bytes, - cache_hits: *cache_hits, - }; - - finalize_query_span.record("query_bytes", query_bytes as i64); - finalize_query_span.record("cache_hits", *cache_hits as i64); - finalize_query_span.record("query_accounts", new_accounts_for_query_len as i64); - - let start_locked_time = Instant::now(); - let mut cache_guard = cache.write().expect("can't lock gpa cache rwlock"); - - // If query is smaller than the min_bytes_per_query, don't cache it - if query_bytes < cache_guard.config.min_bytes_per_query as u64 { - finalize_query_span.record("wall_time", start_time.elapsed().as_millis() as i64); - return; - } - - // Cleanup cache if needed - if let Some(bytes_freed) = cache_guard.cleanup_old_queries(query_bytes) - && bytes_freed < query_bytes - { - tracing::error!(target: "gpa_cache", "Failed to cleanup old queries, not enough bytes freed {}", query_bytes - bytes_freed); - finalize_query_span.record("wall_time", start_time.elapsed().as_millis() as i64); + // Summing the encoded lengths is a cheap linear pass with no allocation, + // unlike building the map, so the threshold is checked first. The config + // is immutable after startup, so a read lock is enough and never blocks + // other readers. + let query_bytes: u64 = accounts.iter().map(|(_, bytes)| bytes.len() as u64).sum(); + let min_bytes_per_query = cache + .read() + .expect("gpa cache rwlock poisoned") + .config + .min_bytes_per_query as u64; + + if query_bytes < min_bytes_per_query { return; } - // Insert the query into the main map (replacing the older query if existed) - let older_query = cache_guard - .queries - .insert(normalized_query.clone(), new_entry); - - // Update map size, crediting back the bytes of the query we just - // replaced (if any) so the counter tracks what is actually held. - cache_guard.size += query_bytes; - if let Some(older_query) = &older_query { - cache_guard.size = cache_guard.size.saturating_sub(older_query.size); - } - - // Mirror the same accounting for pinned bytes: credit the new query if - // it is pinned, and credit back the replaced query if it was pinned. - if cache_guard.is_pinned_size(query_bytes) { - cache_guard.pinned_size += query_bytes; - } - if let Some(older_query) = &older_query - && cache_guard.is_pinned_size(older_query.size) - { - cache_guard.pinned_size = cache_guard.pinned_size.saturating_sub(older_query.size); + FinalizeJob { + cache: cache.clone(), + normalized_query, + accounts, + query_bytes, + slot: *new_slot, + cache_hits: *cache_hits, + previous_query: cached_query.take(), + parent_cx: tracing::Span::current().context(), } - - cache_guard.insert_query_for_slot(normalized_query.clone(), *new_slot, older_query); - - cache_guard.update_size_metrics(); - metrics::CLOUDBREAK_API_REQUEST_DURATION_MS - .with_label_values(&["finalize_query_locked", "cached"]) - .observe(start_locked_time.elapsed().as_micros() as f64); - - finalize_query_span.record("wall_time", start_time.elapsed().as_millis() as i64); - finalize_query_span.record( - "locked_micros", - start_locked_time.elapsed().as_micros() as i64, - ); + .spawn(); } } @@ -550,21 +671,23 @@ impl GpaCache { } /// It will first remove the query from the `queries_for_slot` bucket if it exists. + /// + /// Takes the replaced entry's slot rather than the entry itself so the caller + /// retains ownership and can deallocate it after dropping the write lock. pub fn insert_query_for_slot( &mut self, normalized_query: NormalizedQuery, slot: u64, - older_query: Option, + older_slot: Option, ) { // Remove old version of the query - if let Some(prev) = older_query { - let prev_slot = prev.slot; - if let Some(queries_list) = self.queries_for_slot.get_mut(&prev_slot) { - queries_list.retain(|q| q != &normalized_query); - // If there is no more queries for the slot, remove the slot from the map - if queries_list.is_empty() { - self.queries_for_slot.remove(&prev_slot); - } + if let Some(prev_slot) = older_slot + && let Some(queries_list) = self.queries_for_slot.get_mut(&prev_slot) + { + queries_list.retain(|q| q != &normalized_query); + // If there is no more queries for the slot, remove the slot from the map + if queries_list.is_empty() { + self.queries_for_slot.remove(&prev_slot); } } @@ -614,7 +737,14 @@ impl GpaCache { /// usage drops back under the cap. Because of pinning, cleanup may still free /// less than requested when the oldest slots hold mostly pinned queries that /// are within the cap. - pub fn cleanup_old_queries(&mut self, mut bytes_to_free: u64) -> Option { + /// + /// Evicted entries are appended to `evicted` rather than dropped here, so the + /// caller can deallocate them once the write lock is released. + pub fn cleanup_old_queries( + &mut self, + mut bytes_to_free: u64, + evicted: &mut Vec, + ) -> Option { let mut bytes_freed: u64 = 0; let available_bytes = match (self.config.max_total_bytes as u64).checked_sub(self.size) { @@ -699,6 +829,8 @@ impl GpaCache { crate::metrics::CLOUDBREAK_GPA_CACHE_EVICTED_BYTES_TOTAL .with_label_values(&[used]) .inc_by(cached.size); + + evicted.push(cached); } false }); diff --git a/crates/snapshot/src/sidecar.rs b/crates/snapshot/src/sidecar.rs index 9feb0d2..bdb3a25 100644 --- a/crates/snapshot/src/sidecar.rs +++ b/crates/snapshot/src/sidecar.rs @@ -502,7 +502,8 @@ pub fn unpack_compressed_snapshot>( }; if size != file_size { - tracing::warn!("size mismatch for id: {} and slot: {}", id, slot); + // Since 4.2.1 this is expected to happen for all files marked as "obsolete" + tracing::debug!(target: "unpack_compressed_snapshot", "size mismatch for id: {} and slot: {}", id, slot); } account_file_data.push(AccountFileData {