diff --git a/src/main.rs b/src/main.rs index ab081cc9..54d766cd 100644 --- a/src/main.rs +++ b/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); diff --git a/src/server/mod.rs b/src/server/mod.rs index 97726261..572cf919 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -73,7 +73,10 @@ pub async fn bootstrap(cfg: Arc) -> Result { 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); diff --git a/src/server/pg_compat.rs b/src/server/pg_compat.rs index 25160c5a..cdb8d1d5 100644 --- a/src/server/pg_compat.rs +++ b/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), diff --git a/src/storage.rs b/src/storage.rs index 1389ba97..daecbc12 100644 --- a/src/storage.rs +++ b/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 { + 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 { info!( diff --git a/src/tantivy/search.rs b/src/tantivy/search.rs index ff312d53..dbacb32c 100644 --- a/src/tantivy/search.rs +++ b/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>, /// 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>) -> Self { + self.disk_cache = cache; + self + } + /// Single-predicate convenience used by tests/tools. pub async fn search(self: &Arc, table: &str, project_id: &str, field: &str, query_str: &str) -> Result>> { 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::::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, + disk_cache: Option>, } 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) -> std::io::Result { - 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(), diff --git a/tests/suite/tantivy_search_test.rs b/tests/suite/tantivy_search_test.rs index 634a7c2c..71d7ecdc 100644 --- a/tests/suite/tantivy_search_test.rs +++ b/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 = 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::>(), ["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")]