Api cache optimizations

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)));
@@ -310,10 +308,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);

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