feat(cache): admit reads by access intent

changelog.d/3083.added.md

@@ -0,0 +1,13 @@
+- **The read cache admits by access intent.** `ravel-cache` gains
+  `AccessIntent` (`Interactive`, `BulkScan`, `Maintenance`) and
+  intent-taking variants of the RAM, tiered and S3-FIFO read and admit paths.
+  A `BulkScan` read is admitted to a bulk region that keeps a fixed,
+  hash-chosen subset of a repeated scan larger than the cache and never evicts
+  the main queue, so an interactive hot set survives the scan. A bulk read the
+  region declines when it arrives is not copied and evicts nothing, and one it
+  accepts is copied with the cache unlocked, so interactive reads do not wait
+  on the copy. The admission copy counters count every copy made, including
+  one whose entry is then not stored. No production caller sets `BulkScan`
+  until the SQL executor does in #3084, so this change leaves cache behaviour
+  as it was. The caching guide now describes the caps as S3-FIFO rather than
+  LRU (#2738).

crates/ravel-cache/src/lib.rs

@@ -90,6 +90,11 @@
 //! single-flight collapses) and [`tiered::TieredCache::disk_metrics`], so a
 //! disk-served hit is visible on the disk tier rather than hidden behind a
 //! RAM-only miss.
+//!
+//! Every read carries an [`AccessIntent`], `Interactive` unless a
+//! `_with_intent` method names another. The RAM tier admits a `BulkScan`
+//! read into a bulk region that serves a stable subset of a repeated scan
+//! larger than the cache, without evicting the main queue (see [`s3fifo`]).
 
 mod cache;
 mod clock;
@@ -107,11 +112,38 @@ pub use cache::Cache;
 pub use clock::{Clock, SystemClock};
 pub use disk::{DiskCache, ReclaimReport, RootEntryKind, classify_root_entry, reclaim_legacy};
 pub use key::{CacheKey, PinnedIdentity};
-pub use limits::{CacheLimits, DEFAULT_MAX_ENTRY_AGE_NS, DEFAULT_SWEEP_INTERVAL_NS};
-pub use metrics::{CacheMetrics, CacheMetricsSnapshot};
+pub use limits::{
+    CacheLimits, DEFAULT_BULK_SHARE, DEFAULT_MAX_ENTRY_AGE_NS, DEFAULT_SWEEP_INTERVAL_NS,
+};
+pub use metrics::{CacheMetrics, CacheMetricsSnapshot, IntentCounters, IntentMetricsSnapshot};
 pub use single_flight::{Role, SingleFlight, SingleFlightError};
 pub use tiered::{ReadOutcome, Source, TieredCache};
 
+/// What a read is for, which decides how the RAM tier admits and promotes
+/// its bytes (ADR-3072 decision 3). It is admission policy only: it never
+/// skips a read and never changes what a hit returns. The `_with_intent`
+/// methods of [`Cache`] and [`TieredCache`] take it; every other method
+/// reads as [`AccessIntent::Interactive`]. The disk tier applies the
+/// interactive policy whatever the intent.
+///
+/// No caller outside this crate passes anything but `Interactive` yet;
+/// issue #3084 is the change that has the SQL executor set `BulkScan`.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
+pub enum AccessIntent {
+    /// A read whose bytes may be read again soon: the S3-FIFO policy as it
+    /// was before intents existed.
+    #[default]
+    Interactive,
+    /// A read of a statement that reads every object of a tenant. A hit does
+    /// not count toward promotion, a ghost hit does not promote, and an
+    /// admission goes to the bulk region, which keeps a stable subset of a
+    /// repeated scan larger than itself and never evicts the main queue.
+    BulkScan,
+    /// A read by a maintenance pass. Admitted to the small queue only and
+    /// never promoted by its own hits or by a ghost hit. Nothing sets it yet.
+    Maintenance,
+}
+
 #[cfg(test)]
 #[allow(clippy::unwrap_used, clippy::expect_used)]
 mod tests {
@@ -457,6 +489,82 @@ mod tests {
         }
     }
 
+    /// Hits and bytes served are split by intent and sum to the totals, a bulk
+    /// hit leaves its entry in the bulk region, and the bookkeeping gauge
+    /// matches the cache's own figure.
+    ///
+    /// FLIP: recording `record_hit` instead of `record_hit_with_intent` in
+    /// `Inner::get` leaves every per-intent counter at 0.
+    #[test]
+    fn hits_are_counted_under_their_intent() {
+        let limits = CacheLimits::new(64 * 1024 * 1024, 10_000, 16 * 1024 * 1024);
+        let cache: Cache<&'static str> = Cache::new(limits);
+        let bulk = test_key(1);
+        let interactive = test_key(2);
+        let maintenance = test_key(3);
+        cache.insert_with_intent(bulk, Bytes::from(vec![1u8; 300]), AccessIntent::BulkScan);
+        cache.insert(interactive, Bytes::from(vec![2u8; 200]));
+        cache.insert_with_intent(
+            maintenance,
+            Bytes::from(vec![3u8; 100]),
+            AccessIntent::Maintenance,
+        );
+        assert_eq!(cache.bulk_region_bytes(), 300);
+
+        assert!(
+            cache
+                .get_with_intent(&bulk, AccessIntent::BulkScan)
+                .is_some()
+        );
+        assert!(
+            cache
+                .get_with_intent(&bulk, AccessIntent::BulkScan)
+                .is_some()
+        );
+        assert!(cache.get(&interactive).is_some());
+        assert!(
+            cache
+                .get_with_intent(&maintenance, AccessIntent::Maintenance)
+                .is_some()
+        );
+
+        let intents = cache.metrics().intent_snapshot();
+        assert_eq!(
+            intents.for_intent(AccessIntent::BulkScan),
+            IntentCounters {
+                hits: 2,
+                bytes_served: 600
+            }
+        );
+        assert_eq!(
+            intents.interactive,
+            IntentCounters {
+                hits: 1,
+                bytes_served: 200
+            }
+        );
+        assert_eq!(
+            intents.maintenance,
+            IntentCounters {
+                hits: 1,
+                bytes_served: 100
+            }
+        );
+        let totals = cache.metrics().snapshot();
+        assert_eq!(totals.hits, 4);
+        assert_eq!(totals.bytes_served, 900);
+        assert_eq!(
+            cache.bulk_region_bytes(),
+            300,
+            "a bulk hit stays in the region"
+        );
+        assert!(intents.bulk_bookkeeping_bytes > 0);
+        assert_eq!(
+            intents.bulk_bookkeeping_bytes,
+            cache.bulk_bookkeeping_bytes()
+        );
+    }
+
     #[test]
     fn bounds_respected_under_insertion_pressure_and_oversized_entry_not_admitted() {
         let limits = CacheLimits::new(10 * 1024, 5, 4096);

crates/ravel-cache/src/limits.rs

@@ -26,6 +26,10 @@ pub const DEFAULT_MAX_ENTRY_AGE_NS: u64 = 23 * 60 * 60 * 1_000_000_000;
 /// tiers for the same reason the max-age is.
 pub const DEFAULT_SWEEP_INTERVAL_NS: u64 = 60 * 60 * 1_000_000_000;
 
+/// Default share of `max_bytes` the bulk region holds against interactive
+/// admissions (ADR-3072 decision 3b).
+pub const DEFAULT_BULK_SHARE: f64 = 0.5;
+
 /// Bounds on a cache tier. The first three are enforced independently by both
 /// tiers: an entry that would only blow `max_entry_bytes` is rejected even if
 /// `max_bytes` has room left, and an insert that fits `max_entry_bytes` can
@@ -39,6 +43,9 @@ pub const DEFAULT_SWEEP_INTERVAL_NS: u64 = 60 * 60 * 1_000_000_000;
 /// that drops over-age idle entries. Together they bound how long raw bytes of
 /// an erased subject persist on a query node -- on local disk (ADR-0064)
 /// and in RAM alike.
+///
+/// `bulk_share` applies to the RAM tier only, since only
+/// [`crate::Cache`] admits [`crate::AccessIntent::BulkScan`] reads.
 #[derive(Debug, Clone, Copy)]
 pub struct CacheLimits {
     /// Total bytes across every resident entry.
@@ -61,6 +68,14 @@ pub struct CacheLimits {
     /// entry ages out within at most this interval past `max_entry_age_ns`.
     /// Defaults to [`DEFAULT_SWEEP_INTERVAL_NS`] (1 h).
     pub sweep_interval_ns: u64,
+    /// The share of `max_bytes`, from 0.0 to 1.0, that
+    /// [`crate::AccessIntent::BulkScan`] admissions hold against interactive
+    /// ones: interactive admissions reclaim bulk bytes above it first, and
+    /// bulk admissions evict small-queue entries only while below it. Bulk
+    /// entries may also fill bytes the interactive queues leave unused.
+    /// Values outside the range are clamped to it, and NaN reads as
+    /// [`DEFAULT_BULK_SHARE`], the default.
+    pub bulk_share: f64,
 }
 
 impl CacheLimits {
@@ -74,6 +89,7 @@ impl CacheLimits {
             max_entry_bytes,
             max_entry_age_ns: DEFAULT_MAX_ENTRY_AGE_NS,
             sweep_interval_ns: DEFAULT_SWEEP_INTERVAL_NS,
+            bulk_share: DEFAULT_BULK_SHARE,
         }
     }
 
@@ -93,4 +109,22 @@ impl CacheLimits {
         self.sweep_interval_ns = sweep_interval_ns;
         self
     }
+
+    /// Overrides the bulk region's share of `max_bytes`. Builder-style, like
+    /// [`CacheLimits::with_max_entry_age_ns`].
+    pub fn with_bulk_share(mut self, bulk_share: f64) -> Self {
+        self.bulk_share = bulk_share;
+        self
+    }
+
+    /// `bulk_share` as the bulk region applies it: clamped to 0.0..=1.0,
+    /// and [`DEFAULT_BULK_SHARE`] for NaN, which `f64::clamp` passes
+    /// through.
+    pub(crate) fn effective_bulk_share(&self) -> f64 {
+        if self.bulk_share.is_nan() {
+            DEFAULT_BULK_SHARE
+        } else {
+            self.bulk_share.clamp(0.0, 1.0)
+        }
+    }
 }

crates/ravel-cache/src/metrics.rs

@@ -8,6 +8,8 @@
 
 use std::sync::atomic::{AtomicU64, Ordering};
 
+use crate::AccessIntent;
+
 /// Shared counters for one `Cache`, held behind an `Arc` and cloned out
 /// with [`crate::Cache::metrics`] so a scrape path can read them without
 /// touching the cache itself.
@@ -24,6 +26,18 @@ pub struct CacheMetrics {
     disk_entries_expired_max_age: AtomicU64,
     admission_copies: AtomicU64,
     admission_copied_bytes: AtomicU64,
+    /// Indexed by [`intent_index`].
+    intent_hits: [AtomicU64; 3],
+    intent_bytes_served: [AtomicU64; 3],
+    bulk_bookkeeping_bytes: AtomicU64,
+}
+
+fn intent_index(intent: AccessIntent) -> usize {
+    match intent {
+        AccessIntent::Interactive => 0,
+        AccessIntent::BulkScan => 1,
+        AccessIntent::Maintenance => 2,
+    }
 }
 
 impl CacheMetrics {
@@ -32,6 +46,44 @@ impl CacheMetrics {
         self.bytes_served.fetch_add(bytes, Ordering::Relaxed);
     }
 
+    /// A RAM-tier hit, counted in [`Self::record_hit`]'s totals and under its
+    /// intent. The disk tier records plain hits and no intent.
+    pub(crate) fn record_hit_with_intent(&self, intent: AccessIntent, bytes: u64) {
+        self.record_hit(bytes);
+        let index = intent_index(intent);
+        self.intent_hits[index].fetch_add(1, Ordering::Relaxed);
+        self.intent_bytes_served[index].fetch_add(bytes, Ordering::Relaxed);
+    }
+
+    /// Publishes the bulk region's current bookkeeping bytes, a gauge. Stores
+    /// only when the value changed, so a lookup on a cache with no bulk
+    /// region writes nothing.
+    pub(crate) fn set_bulk_bookkeeping_bytes(&self, bytes: u64) {
+        if self.bulk_bookkeeping_bytes.load(Ordering::Relaxed) != bytes {
+            self.bulk_bookkeeping_bytes.store(bytes, Ordering::Relaxed);
+        }
+    }
+
+    /// Hits and bytes served split by [`AccessIntent`], and the bulk region's
+    /// bookkeeping bytes. Read on its own, like [`Self::admission_copies`],
+    /// because [`CacheMetricsSnapshot`] is built field by field outside this
+    /// crate. A scrape, not a consistent cut, as [`Self::snapshot`] is.
+    pub fn intent_snapshot(&self) -> IntentMetricsSnapshot {
+        let counters = |intent| {
+            let index = intent_index(intent);
+            IntentCounters {
+                hits: self.intent_hits[index].load(Ordering::Relaxed),
+                bytes_served: self.intent_bytes_served[index].load(Ordering::Relaxed),
+            }
+        };
+        IntentMetricsSnapshot {
+            interactive: counters(AccessIntent::Interactive),
+            bulk_scan: counters(AccessIntent::BulkScan),
+            maintenance: counters(AccessIntent::Maintenance),
+            bulk_bookkeeping_bytes: self.bulk_bookkeeping_bytes.load(Ordering::Relaxed),
+        }
+    }
+
     pub(crate) fn record_miss(&self) {
         self.misses.fetch_add(1, Ordering::Relaxed);
     }
@@ -74,17 +126,19 @@ impl CacheMetrics {
             .fetch_add(1, Ordering::Relaxed);
     }
 
-    /// A RAM-tier admission stored an exact-size copy of the value rather than
-    /// the value itself, because the value did not solely own an allocation of
-    /// exactly its own length (see `Cache::insert`).
+    /// A RAM-tier admission made an exact-size copy of the value to store in
+    /// place of the value itself, because the value did not solely own an
+    /// allocation of exactly its own length (see `Cache::insert`). Recorded
+    /// where the copy is made, whether or not the entry is then stored.
     pub(crate) fn record_admission_copy(&self, bytes: u64) {
         self.admission_copies.fetch_add(1, Ordering::Relaxed);
         self.admission_copied_bytes
             .fetch_add(bytes, Ordering::Relaxed);
     }
 
-    /// RAM-tier admissions that stored a copy rather than the value given
-    /// (`ravel_cache_admission_copies_total`). Read on its own rather than
+    /// RAM-tier admissions that copied the value given, whether or not the
+    /// copy was then stored (`ravel_cache_admission_copies_total`). Read on
+    /// its own rather than
     /// through [`Self::snapshot`] because [`CacheMetricsSnapshot`] is built
     /// field by field outside this crate.
     pub fn admission_copies(&self) -> u64 {
@@ -98,7 +152,7 @@ impl CacheMetrics {
     }
 
     /// Point-in-time copy of every counter but the two admission-copy ones
-    /// above. Not atomic across fields:
+    /// and the per-intent ones above. Not atomic across fields:
     /// concurrent calls may land between two loads, so a snapshot can show
     /// `hits` from a hair later than `misses`. It is a scrape, not a
     /// consistent cut, and nothing correctness-bearing reads it.
@@ -133,3 +187,35 @@ pub struct CacheMetricsSnapshot {
     pub disk_errors_degraded_to_misses: u64,
     pub disk_entries_expired_max_age: u64,
 }
+
+/// Hits and bytes served for one [`AccessIntent`].
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct IntentCounters {
+    pub hits: u64,
+    pub bytes_served: u64,
+}
+
+/// Point-in-time copy of [`CacheMetrics`]' per-intent counters, from
+/// [`CacheMetrics::intent_snapshot`]. On a RAM tier the three intents' hits
+/// sum to [`CacheMetricsSnapshot::hits`]; a disk tier records no intent and
+/// reads zero here.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct IntentMetricsSnapshot {
+    pub interactive: IntentCounters,
+    pub bulk_scan: IntentCounters,
+    pub maintenance: IntentCounters,
+    /// The bulk region's bookkeeping bytes, charged inside `max_bytes`; a
+    /// gauge, not a counter.
+    pub bulk_bookkeeping_bytes: u64,
+}
+
+impl IntentMetricsSnapshot {
+    /// The counters for `intent`.
+    pub fn for_intent(&self, intent: AccessIntent) -> IntentCounters {
+        match intent {
+            AccessIntent::Interactive => self.interactive,
+            AccessIntent::BulkScan => self.bulk_scan,
+            AccessIntent::Maintenance => self.maintenance,
+        }
+    }
+}