tantivy: serve bundle ranges from the foyer disk tier

src/main.rs

@@ -510,7 +510,10 @@ async fn async_main(cfg: &'static AppConfig) -> anyhow::Result<()> {
         let tcfg = Arc::new(cfg.tantivy.clone());
         let svc = Arc::new(timefusion::tantivy::search::TantivyIndexService::new(obj_store.clone(), tcfg.clone(), cfg.core.timefusion_data_dir.clone()));
         layer = layer.with_tantivy_indexer(timefusion::server::tantivy_index_callback(&db, Arc::clone(&svc)));
-        let search = Arc::new(timefusion::tantivy::search::TantivySearchService::new(obj_store, cfg.core.timefusion_data_dir.clone(), tcfg));
+        let search = Arc::new(
+            timefusion::tantivy::search::TantivySearchService::new(obj_store, cfg.core.timefusion_data_dir.clone(), tcfg)
+                .with_disk_cache(db.object_store_cache().cloned()),
+        );
         // Lets a publish seed the reader's cache and invalidate its manifest
         // in-process instead of round-tripping through S3.
         svc.with_reader(&search);

src/server/mod.rs

@@ -73,7 +73,10 @@ pub async fn bootstrap(cfg: Arc<AppConfig>) -> Result<Bootstrapped> {
         let tcfg = Arc::new(cfg.tantivy.clone());
         let svc = Arc::new(crate::tantivy::search::TantivyIndexService::new(obj_store.clone(), tcfg.clone(), cfg.core.timefusion_data_dir.clone()));
         layer = layer.with_tantivy_indexer(tantivy_index_callback(&db, Arc::clone(&svc)));
-        let search = Arc::new(crate::tantivy::search::TantivySearchService::new(obj_store, cfg.core.timefusion_data_dir.clone(), tcfg));
+        let search = Arc::new(
+            crate::tantivy::search::TantivySearchService::new(obj_store, cfg.core.timefusion_data_dir.clone(), tcfg)
+                .with_disk_cache(db.object_store_cache().cloned()),
+        );
         // Lets a publish seed the reader's cache and invalidate its manifest
         // in-process instead of round-tripping through S3.
         svc.with_reader(&search);

src/server/pg_compat.rs

@@ -1235,6 +1235,8 @@ impl StatsTableProvider {
                 "blob_fetch_bytes" => s.blob_fetch_bytes.load(Relaxed),
                 "range_reads" => s.range_reads.load(Relaxed),
                 "range_read_bytes" => s.range_read_bytes.load(Relaxed),
+                "range_disk_hits" => s.range_disk_hits.load(Relaxed),
+                "range_disk_hit_bytes" => s.range_disk_hit_bytes.load(Relaxed),
                 "range_read_fast_bytes" => s.range_read_fast_bytes.load(Relaxed),
                 "range_read_terms_bytes" => s.range_read_terms_bytes.load(Relaxed),
                 "range_read_store_bytes" => s.range_read_store_bytes.load(Relaxed),

src/storage.rs

@@ -507,6 +507,20 @@ impl foyer::EventListener for EvictionCounter {
 }
 
 impl SharedFoyerCache {
+    /// A range of an immutable object (e.g. a tantivy bundle) from the cache; `None` on a miss.
+    pub async fn get_immutable(&self, key: &str) -> Option<Bytes> {
+        let entry = self.cache.get(&key.to_owned()).await.ok()??;
+        let value = entry.value();
+        (!value.is_expired(self.config.ttl)).then(|| value.data.clone())
+    }
+
+    /// Cache an immutable object's range disk-only: it survives restarts without displacing
+    /// the in-memory working set.
+    pub fn put_immutable(&self, key: String, data: Bytes) {
+        let meta = ObjectMeta { location: Path::from(key.as_str()), last_modified: Utc::now(), size: data.len() as u64, e_tag: None, version: None };
+        self.cache.insert_with_properties(key, CacheValue::new(data, meta), HybridCacheProperties::default().with_location(Location::OnDisk));
+    }
+
     /// Create the process-wide data + metadata caches.
     pub async fn new(config: FoyerCacheConfig) -> anyhow::Result<Self> {
         info!(

src/tantivy/search.rs

@@ -52,6 +52,9 @@ pub struct SearchStats {
     /// Ranged reads of bundles in place of whole-blob installs.
     pub range_reads: AtomicU64,
     pub range_read_bytes: AtomicU64,
+    /// Bundle ranges served by the disk cache instead of the object store.
+    pub range_disk_hits: AtomicU64,
+    pub range_disk_hit_bytes: AtomicU64,
     /// `range_read_bytes` by index file kind; the rest is everything else.
     pub range_read_fast_bytes: AtomicU64,
     pub range_read_terms_bytes: AtomicU64,
@@ -179,6 +182,9 @@ pub struct TantivySearchService {
     /// (not global) so distinct object stores never cross-contaminate.
     manifests: DashMap<(String, String), CachedManifest>,
     block_cache: super::hotcache::BlockCache,
+    /// Disk tier under `block_cache`: bundles are immutable (every publish takes a fresh
+    /// generation path), so their ranges survive restarts there.
+    disk_cache: Option<Arc<crate::storage::SharedFoyerCache>>,
     /// Last time each cache dir was served to a query — the reaper's recency
     /// signal, since mmap reads don't reliably move a directory's atime. Dirs
     /// absent here fall back to dir mtime, which is their unpack time.
@@ -315,6 +321,7 @@ impl TantivySearchService {
             cache_root,
             readers: Mutex::new(LruCache::new(config.reader_cache_entries())),
             block_cache: super::hotcache::block_cache(config.timefusion_tantivy_block_cache_mb << 20),
+            disk_cache: None,
             config,
             stats: Default::default(),
             manifests: DashMap::new(),
@@ -328,6 +335,11 @@ impl TantivySearchService {
         }
     }
 
+    pub fn with_disk_cache(mut self, cache: Option<Arc<crate::storage::SharedFoyerCache>>) -> Self {
+        self.disk_cache = cache;
+        self
+    }
+
     /// Single-predicate convenience used by tests/tools.
     pub async fn search(self: &Arc<Self>, table: &str, project_id: &str, field: &str, query_str: &str) -> Result<Option<Vec<Hit>>> {
         let node = PredNode::Leaf(TextMatchPred { column: field.to_string(), query: query_str.to_string() });
@@ -927,7 +939,13 @@ impl TantivySearchService {
         }
         let path = ObjPath::from(blob_path);
         let head = self.object_store.get_range(&path, 0..head_len).await.with_context(|| format!("get head of {path}"))?;
-        let source = Arc::new(StoreSource { store: self.object_store.clone(), path, rt: tokio::runtime::Handle::current(), stats: self.stats.clone() });
+        let source = Arc::new(StoreSource {
+            store: self.object_store.clone(),
+            path,
+            rt: tokio::runtime::Handle::current(),
+            stats: self.stats.clone(),
+            disk_cache: self.disk_cache.clone(),
+        });
         let bundle = std::hash::BuildHasher::hash_one(&std::hash::BuildHasherDefault::<std::collections::hash_map::DefaultHasher>::default(), blob_path);
         let directory = super::hotcache::BundleDirectory::new(head, source)?.with_cache(bundle, self.block_cache.clone());
         let index = super::open_index_in(directory)?;
@@ -1148,6 +1166,7 @@ struct StoreSource {
     path: ObjPath,
     rt: tokio::runtime::Handle,
     stats: Arc<SearchStats>,
+    disk_cache: Option<Arc<crate::storage::SharedFoyerCache>>,
 }
 
 impl super::hotcache::BundleSource for StoreSource {
@@ -1164,9 +1183,25 @@ impl super::hotcache::BundleSource for StoreSource {
         self.read(range)
     }
     fn read(&self, range: std::ops::Range<u64>) -> std::io::Result<bytes::Bytes> {
-        SearchStats::add(&self.stats.range_reads, 1);
-        SearchStats::add(&self.stats.range_read_bytes, range.end - range.start);
-        let fetch = || self.rt.block_on(self.store.get_range(&self.path, range)).map_err(std::io::Error::other);
+        let key = format!("tantivy-bundle:{}:{}-{}", self.path, range.start, range.end);
+        let fetch = || {
+            self.rt.block_on(async {
+                if let Some(cache) = &self.disk_cache
+                    && let Some(bytes) = cache.get_immutable(&key).await
+                {
+                    SearchStats::add(&self.stats.range_disk_hits, 1);
+                    SearchStats::add(&self.stats.range_disk_hit_bytes, bytes.len() as u64);
+                    return Ok(bytes);
+                }
+                SearchStats::add(&self.stats.range_reads, 1);
+                SearchStats::add(&self.stats.range_read_bytes, range.end - range.start);
+                let bytes = self.store.get_range(&self.path, range.clone()).await.map_err(std::io::Error::other)?;
+                if let Some(cache) = &self.disk_cache {
+                    cache.put_immutable(key.clone(), bytes.clone());
+                }
+                Ok(bytes)
+            })
+        };
         match tokio::runtime::Handle::try_current() {
             Ok(_) => std::thread::scope(|s| s.spawn(fetch).join()).map_err(|_| std::io::Error::other("bundle fetch panicked"))?,
             Err(_) => fetch(),

tests/suite/tantivy_search_test.rs

@@ -948,6 +948,38 @@ async fn a_bundled_index_is_searched_by_range_reads_without_an_install() {
     }
 }
 
+/// A restarted reader (empty block cache) is served a bundle's ranges by the disk cache, not
+/// the object store: after a deploy every first search otherwise re-reads its columns from S3.
+#[tokio::test(flavor = "multi_thread")]
+async fn bundle_ranges_survive_a_restart_in_the_disk_cache() {
+    use timefusion::storage::{FoyerCacheConfig, SharedFoyerCache};
+    let bundled = TantivyConfig { timefusion_tantivy_bundle_writes: true, timefusion_tantivy_range_reads: true, ..prod_defaults() };
+    let store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+    let env = Env::new("otel_logs_and_spans", "p-bundle-disk", store.clone(), bundled.clone(), bundled.clone());
+    env.publish(&[(1_000_000, "a", "ERROR"), (1_000_001, "b", "INFO")], &["f1"]).await;
+    let disk = Arc::new(SharedFoyerCache::new(FoyerCacheConfig::test_config("bundle_disk_cache")).await.unwrap());
+    let reader = || {
+        let dir = TempDir::new().unwrap();
+        (Arc::new(TantivySearchService::new(store.clone(), dir.path().to_path_buf(), Arc::new(bundled.clone())).with_disk_cache(Some(disk.clone()))), dir)
+    };
+    let (first, _dir) = reader();
+    assert!(first.search_detailed(env.table, env.project, &level_error_node(), 100, None, false).await.unwrap().is_ok());
+    assert!(first.stats.range_reads.load(Relaxed) > 0, "the first reader fetches from the object store");
+    // Disk-only inserts land asynchronously.
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
+    loop {
+        let (restarted, _dir) = reader();
+        let r = restarted.search_detailed(env.table, env.project, &level_error_node(), 100, None, false).await.unwrap().unwrap();
+        assert_eq!(r.hits.iter().map(|h| h.id.as_str()).collect::<Vec<_>>(), ["a"]);
+        if restarted.stats.range_reads.load(Relaxed) == 0 {
+            assert!(restarted.stats.range_disk_hits.load(Relaxed) > 0);
+            break;
+        }
+        assert!(std::time::Instant::now() < deadline, "a restarted reader must be served from the disk cache");
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    }
+}
+
 /// Old `tar.zst` indexes are repacked as bundles in place, so searches stop skipping them
 /// as cold and read them by range instead.
 #[tokio::test(flavor = "multi_thread")]