diff --git a/config/quickwit.yaml b/config/quickwit.yaml index 2cacb3bca4a..4ff3bb8dfa2 100644 --- a/config/quickwit.yaml +++ b/config/quickwit.yaml @@ -203,6 +203,9 @@ indexer: # max_num_bytes: 1G # max_num_splits: 10000 # num_concurrent_downloads: 1 +# # Opt-in: refuse to cache a split larger than max_num_bytes (served from +# # object storage instead). Defaults to false (no per-split size guard). +# skip_oversized_splits: false # -------------------------------- Jaeger settings -------------------------------- jaeger: diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index f3c771eec4c..a835a5b675d 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -362,6 +362,13 @@ pub struct SplitCacheLimits { pub num_concurrent_downloads: NonZeroU32, #[serde(default = "SplitCacheLimits::default_max_file_descriptors")] pub max_file_descriptors: NonZeroU32, + /// Opt-in per-split size guard. When `true`, the searcher split cache refuses + /// to download a split larger than `max_num_bytes`, reserves the bytes of + /// in-flight downloads against the budget, and advances past candidates it + /// cannot make room for. Defaults to `false`, which preserves the historical + /// behavior (no size guard: any reported split can be downloaded). + #[serde(default)] + pub skip_oversized_splits: bool, } impl SplitCacheLimits { diff --git a/quickwit/quickwit-indexing/src/actors/uploader.rs b/quickwit/quickwit-indexing/src/actors/uploader.rs index 44dbf9d2aa1..09203755f13 100644 --- a/quickwit/quickwit-indexing/src/actors/uploader.rs +++ b/quickwit/quickwit-indexing/src/actors/uploader.rs @@ -338,6 +338,10 @@ impl Handler for Uploader { report_splits.push(ReportSplit { storage_uri: split_store.remote_uri().to_string(), split_id: packaged_split.split_id_str().to_string(), + // The footer sits at the end of the split file, so its end + // offset equals the split file size. The searcher split + // cache uses this to skip caching splits larger than its budget. + num_bytes: split_streamer.footer_range.end, }); split_metadata_list.push(split_metadata); @@ -1058,6 +1062,9 @@ mod tests { let split = &report_splits.report_splits[0]; assert_eq!(split.storage_uri, "ram:///"); assert_eq!(split.split_id, SPLIT_ULID_STR); + // The reported size is the split file size (footer end offset), which the + // searcher split cache uses to skip caching oversized splits. + assert!(split.num_bytes > 0); universe.assert_quit().await; Ok(()) } diff --git a/quickwit/quickwit-proto/protos/quickwit/search.proto b/quickwit/quickwit-proto/protos/quickwit/search.proto index 4d060d6e1be..bedf561c544 100644 --- a/quickwit/quickwit-proto/protos/quickwit/search.proto +++ b/quickwit/quickwit-proto/protos/quickwit/search.proto @@ -105,6 +105,10 @@ message ReportSplit { string split_id = 2; // The storage uri. This URI does NOT include the split id. string storage_uri = 1; + // Size of the split file in bytes (i.e. `footer_offsets.end`). + // A value of 0 means the size is unknown to the reporter; the split + // cache treats an unknown size as "fits" so behavior is unchanged. + uint64 num_bytes = 3; } message ReportSplitsRequest { diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs index a9109c1ad11..7c21fb082c3 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.search.rs @@ -43,6 +43,11 @@ pub struct ReportSplit { /// The storage uri. This URI does NOT include the split id. #[prost(string, tag = "1")] pub storage_uri: ::prost::alloc::string::String, + /// Size of the split file in bytes (i.e. `footer_offsets.end`). + /// A value of 0 means the size is unknown to the reporter; the split + /// cache treats an unknown size as "fits" so behavior is unchanged. + #[prost(uint64, tag = "3")] + pub num_bytes: u64, } #[derive(serde::Serialize, serde::Deserialize, utoipa::ToSchema)] #[derive(Clone, PartialEq, ::prost::Message)] diff --git a/quickwit/quickwit-search/src/leaf.rs b/quickwit/quickwit-search/src/leaf.rs index 874cbe111c3..32c096bc4e5 100644 --- a/quickwit/quickwit-search/src/leaf.rs +++ b/quickwit/quickwit-search/src/leaf.rs @@ -179,6 +179,15 @@ pub(crate) async fn open_split_bundle( // This is before the bundle storage: at this point, this storage is reading `.split` files. let index_storage_with_split_cache = if let Some(split_cache) = searcher_context.split_cache_opt.as_ref() { + // Register the split together with its file size (the footer end + // offset) so the download guard can skip it when it is larger than + // the cache budget. This is the only size signal for pre-existing + // splits that no indexer ever reported a size for. + split_cache.report_split_size( + split_and_footer_offsets.split_id.clone().into(), + index_storage.uri(), + split_and_footer_offsets.split_footer_end, + ); SearchSplitCache::wrap_storage(split_cache.clone(), index_storage.clone()) } else { index_storage.clone() diff --git a/quickwit/quickwit-storage/src/metrics.rs b/quickwit/quickwit-storage/src/metrics.rs index 446a45f60d4..e2f77cfc310 100644 --- a/quickwit/quickwit-storage/src/metrics.rs +++ b/quickwit/quickwit-storage/src/metrics.rs @@ -335,6 +335,19 @@ pub static PREDICATE_CACHE: LazyLock = pub(crate) static SEARCHER_SPLIT_CACHE: LazyLock = LazyLock::new(|| CacheMetrics::for_component("searcher_split")); +/// Number of times a split download opportunity was skipped because the split +/// is larger than the entire split-cache byte budget (`split_cache.max_num_bytes`). +/// +/// Such a split can never coexist with any other split and would evict the whole +/// cache without ever fitting, so it is left to be served by the cold-storage +/// warmup path. A steadily rising counter means a searcher keeps re-evaluating a +/// hot oversized split; it is a "skip events" rate, not a distinct-split count. +pub(crate) static SEARCHER_SPLIT_CACHE_DOWNLOADS_SKIPPED_TOO_LARGE: LazyCounter = lazy_counter!( + name: "searcher_split_cache_downloads_skipped_too_large_total", + description: "Number of split downloads skipped because the split is larger than the split cache byte budget.", + subsystem: "storage", +); + /// Cache metrics for short-lived byte range caches (used during leaf search /// and caching directory warmup). pub static SHORTLIVED_CACHE: LazyLock = diff --git a/quickwit/quickwit-storage/src/split_cache/download_task.rs b/quickwit/quickwit-storage/src/split_cache/download_task.rs index 1f58b691459..25ef9650528 100644 --- a/quickwit/quickwit-storage/src/split_cache/download_task.rs +++ b/quickwit/quickwit-storage/src/split_cache/download_task.rs @@ -32,6 +32,7 @@ async fn download_split( split_id, storage_uri, living_token: _, + num_bytes: _, } = candidate_split; let split_filename = split_file(split_id); let target_filepath = root_path.join(&split_filename); diff --git a/quickwit/quickwit-storage/src/split_cache/mod.rs b/quickwit/quickwit-storage/src/split_cache/mod.rs index 57569060697..aff04326d93 100644 --- a/quickwit/quickwit-storage/src/split_cache/mod.rs +++ b/quickwit/quickwit-storage/src/split_cache/mod.rs @@ -94,7 +94,8 @@ impl SearchSplitCache { let mut split_table = SplitTable::with_limits_and_existing_splits(limits, existing_splits); // In case of a setting change, it could be useful to evict some splits on startup. - let splits_to_remove_res = split_table.make_room_for_split_if_necessary(u64::MAX); + // No specific split is incoming here; `0` just trims the table to the current limits. + let splits_to_remove_res = split_table.make_room_for_split_if_necessary(u64::MAX, 0); if let Ok(splits_to_remove) = splits_to_remove_res { info!( num_splits = splits_to_remove.len(), @@ -144,10 +145,26 @@ impl SearchSplitCache { error!(storage_uri=%report_split.storage_uri, "received invalid storage uri: ignoring"); continue; }; - split_table.report(split_id, storage_uri); + split_table.report(split_id, storage_uri, report_split.num_bytes); } } + /// Informs the cache that a split is about to be searched, together with its + /// file size in bytes (the split footer end offset). + /// + /// This is the only size signal for splits that no indexer reported a size + /// for — most importantly pre-existing oversized splits. Those are otherwise + /// discovered via `get_split_file`/`touch` without a size and would bypass + /// the download guard entirely. Attaching the size here lets the guard skip + /// them just like freshly-reported oversized splits. + /// + /// It is a no-op unless the `skip_oversized_splits` guard is enabled (and the + /// size is known), so the default download path is unaffected. + pub fn report_split_size(&self, split_id: SplitId, storage_uri: &Uri, num_bytes: u64) { + let mut split_table = self.split_table.lock().unwrap(); + split_table.report_split_size_from_search(split_id, storage_uri.clone(), num_bytes); + } + // Returns a split guard object. As long as it is not dropped, the // split won't be evinced from the cache. async fn get_split_file(&self, split_id: SplitId, storage_uri: &Uri) -> Option { diff --git a/quickwit/quickwit-storage/src/split_cache/split_table.rs b/quickwit/quickwit-storage/src/split_cache/split_table.rs index 4e191bc47fa..507667d650f 100644 --- a/quickwit/quickwit-storage/src/split_cache/split_table.rs +++ b/quickwit/quickwit-storage/src/split_cache/split_table.rs @@ -21,7 +21,7 @@ use quickwit_common::uri::Uri; use quickwit_config::SplitCacheLimits; use quickwit_proto::types::SplitId; -use crate::metrics::SEARCHER_SPLIT_CACHE; +use crate::metrics::{SEARCHER_SPLIT_CACHE, SEARCHER_SPLIT_CACHE_DOWNLOADS_SKIPPED_TOO_LARGE}; type LastAccessDate = u64; @@ -60,7 +60,7 @@ impl Eq for SplitKey {} #[derive(Clone, Debug)] enum Status { Candidate(CandidateSplit), - Downloading { alive_token: Weak<()> }, + Downloading { alive_token: Weak<()>, num_bytes: u64 }, OnDisk { num_bytes: u64 }, } @@ -97,8 +97,15 @@ pub struct SplitInfo { /// - downloading_splits /// - candidate_splits. /// -/// It is possible for the split table size in bytes to exceed its limits, by at -/// most one split. +/// By default the split table size in bytes may exceed its limits by at most one +/// split (the incoming split's bytes are not reserved before its download +/// starts). +/// +/// When `SplitCacheLimits::skip_oversized_splits` is enabled, splits larger than +/// `max_num_bytes` are never downloaded, and the incoming split plus all +/// in-flight downloads are reserved against the budget — so on-disk usage stays +/// within `max_num_bytes`, exceeded only transiently by downloads still in +/// flight (each already checked to fit). pub struct SplitTable { on_disk_splits: BTreeSet, downloading_splits: BTreeSet, @@ -169,7 +176,7 @@ impl SplitTable { }; let is_in_queue = split_queue.remove(&split_info.split_key); assert!(is_in_queue); - if let Status::Downloading { alive_token } = &split_info.status + if let Status::Downloading { alive_token, .. } = &split_info.status && alive_token.strong_count() == 0 { return None; @@ -186,7 +193,7 @@ impl SplitTable { let mut splits_to_remove = Vec::new(); for split in &self.downloading_splits { if let Some(split_info) = self.split_to_status.get(&split.split_id) - && let Status::Downloading { alive_token } = &split_info.status + && let Status::Downloading { alive_token, .. } = &split_info.status && alive_token.strong_count() == 0 { splits_to_remove.push(split.split_id.clone()); @@ -255,6 +262,9 @@ impl SplitTable { storage_uri: storage_uri.clone(), split_id, living_token: Arc::new(()), + // The size is not known on the search read path; a later + // `report` fills it in. Unknown (0) is treated as "fits". + num_bytes: 0, }), } } @@ -302,10 +312,20 @@ impl SplitTable { }); } - pub(crate) fn report(&mut self, split_id: SplitId, storage_uri: Uri) { + pub(crate) fn report(&mut self, split_id: SplitId, storage_uri: Uri, num_bytes: u64) { let origin_time = self.origin_time; self.mutate_split(split_id, move |split_info_opt, split_id| { - if let Some(split_info) = split_info_opt { + if let Some(mut split_info) = split_info_opt { + // The split is already known. Attach the size to a candidate that + // was first discovered without one (e.g. via `touch`), but never + // overwrite a known size, and never disturb downloading/on-disk + // splits (they already carry an accurate size). + if let Status::Candidate(candidate_split) = &mut split_info.status + && candidate_split.num_bytes == 0 + && num_bytes > 0 + { + candidate_split.num_bytes = num_bytes; + } return split_info; } SplitInfo { @@ -318,11 +338,42 @@ impl SplitTable { storage_uri, split_id, living_token: Arc::new(()), + num_bytes, }), } }); } + /// Registers a split size discovered on the search path. + /// + /// This is a no-op unless the size guard is enabled and the size is known + /// (`num_bytes > 0`), so the default download path is never affected by an + /// extra candidate registration. + pub(crate) fn report_split_size_from_search( + &mut self, + split_id: SplitId, + storage_uri: Uri, + num_bytes: u64, + ) { + if !self.limits.skip_oversized_splits || num_bytes == 0 { + return; + } + // Only attach a size to a candidate (or a not-yet-known split). Splits + // that are already downloading or on disk already carry an accurate size; + // routing them through `report` (a remove + reinsert) would wrongly bump + // the cache's eviction counters for what is actually a cache hit. + let is_candidate_or_unknown = matches!( + self.split_to_status.get(&split_id), + None | Some(SplitInfo { + status: Status::Candidate(_), + .. + }) + ); + if is_candidate_or_unknown { + self.report(split_id, storage_uri, num_bytes); + } + } + /// Make sure we have at most `MAX_CANDIDATES` candidate splits. fn truncate_candidate_list(&mut self) { // we remove one more to make place for one candidate about to be inserted @@ -348,30 +399,98 @@ impl SplitTable { return None; }; let alive_token = Arc::downgrade(&candidate_split.living_token); + // Carry the split size into the downloading state so its bytes are + // reserved against the budget while the download is in flight. + let num_bytes = candidate_split.num_bytes; self.insert(SplitInfo { split_key: split_info.split_key, - status: Status::Downloading { alive_token }, + status: Status::Downloading { + alive_token, + num_bytes, + }, }); Some(candidate_split) } + /// Returns the hottest candidate (highest last-accessed, ties broken by + /// split id). Used by the default download path (size guard disabled). fn best_candidate(&self) -> Option { self.candidate_splits.last().cloned() } - fn is_out_of_limits(&self) -> bool { - if self.on_disk_splits.is_empty() { - return false; + /// Returns the known size in bytes of a candidate split, or 0 if the split + /// is unknown, or not currently a candidate, or was reported without a size. + /// + /// A size of 0 is treated as "fits" by the download guard, so a candidate + /// without a known size behaves exactly as it did before the guard existed. + fn candidate_num_bytes(&self, split_id: &SplitId) -> u64 { + match self.split_to_status.get(split_id) { + Some(SplitInfo { + status: Status::Candidate(candidate_split), + .. + }) => candidate_split.num_bytes, + _ => 0, } - if self.on_disk_splits.len() + self.downloading_splits.len() - >= self.limits.max_num_splits.get() as usize - { - return true; + } + + /// Sum of the reported sizes of the splits currently being downloaded. + /// + /// These bytes are not yet on disk but will be shortly, so they must be + /// reserved against the budget to avoid overshooting when several downloads + /// run concurrently (`num_concurrent_downloads > 1`). A downloading split + /// with an unknown size (0) contributes nothing, matching legacy behavior. + fn downloading_bytes(&self) -> u64 { + self.downloading_splits + .iter() + .map(|key| match self.split_to_status.get(&key.split_id) { + // Only count downloads that are still alive. A failed download + // keeps a dead `Downloading` entry until it is garbage-collected; + // reserving its bytes would needlessly block other downloads. + Some(SplitInfo { + status: Status::Downloading { + alive_token, + num_bytes, + }, + .. + }) if alive_token.strong_count() > 0 => *num_bytes, + _ => 0, + }) + .sum() + } + + /// Returns true if the table is (or, with the size guard, would be) over its + /// byte or split-count budget. + /// + /// With the guard disabled (default), room is judged only on what is already + /// on disk: neither the incoming split nor in-flight downloads are reserved, + /// so the table may exceed its limits by at most one split (per the type + /// invariant). Eviction can only reclaim on-disk splits, so an empty disk is + /// never reported as over limit. + /// + /// With the guard enabled, the incoming split (`incoming_bytes`) and every + /// in-flight download are reserved against the budget. Counting in-flight + /// downloads matters when `num_concurrent_downloads > 1`, where two + /// concurrent sub-budget splits could otherwise both pass and overshoot once + /// they land. + fn would_exceed_limits_with(&self, incoming_bytes: u64) -> bool { + if !self.limits.skip_oversized_splits { + if self.on_disk_splits.is_empty() { + return false; + } + if self.on_disk_splits.len() + self.downloading_splits.len() + >= self.limits.max_num_splits.get() as usize + { + return true; + } + return self.on_disk_bytes > self.limits.max_num_bytes.as_u64(); } - if self.on_disk_bytes > self.limits.max_num_bytes.as_u64() { + if self.on_disk_splits.len() + self.downloading_splits.len() + 1 + > self.limits.max_num_splits.get() as usize + { return true; } - false + self.on_disk_bytes + self.downloading_bytes() + incoming_bytes + > self.limits.max_num_bytes.as_u64() } /// Evicts splits to reach the target limits. @@ -384,9 +503,10 @@ impl SplitTable { pub(crate) fn make_room_for_split_if_necessary( &mut self, last_access_date: LastAccessDate, + incoming_bytes: u64, ) -> Result, NoRoomAvailable> { let mut split_infos = Vec::new(); - while self.is_out_of_limits() { + while self.would_exceed_limits_with(incoming_bytes) { // We clone the oldest split's key so we can drop the immutable borrow on // `on_disk_splits` before calling `remove`, which needs `&mut self`. let oldest_split_key_opt: Option = self.on_disk_splits.first().cloned(); @@ -400,7 +520,7 @@ impl SplitTable { break; } } - if self.is_out_of_limits() { + if self.would_exceed_limits_with(incoming_bytes) { // We are still out of limits. // Let's not go through with the eviction, and reinsert the splits. for split_info in split_infos { @@ -415,17 +535,65 @@ impl SplitTable { } } + /// Selects the next split download opportunity. + /// + /// Default behavior (size guard disabled): take the single hottest candidate + /// and download it; if room cannot be made for it, do nothing this round. + /// + /// With the size guard enabled, candidates are instead scanned hottest-first + /// for the best *fitting* candidate that can also make room for itself. Two + /// kinds of candidate are skipped in favour of the next one rather than + /// aborting the scan — otherwise the downloader would stall on a single + /// permanently-blocked (but permanently hot) split and starve the cache: + /// - larger than the whole budget (`max_num_bytes`): can never coexist with + /// anything, so it is skipped (and counted) and left to the cold-storage + /// warmup path; + /// - fits, but cannot make room (its eviction would remove a *fresher* + /// on-disk split): a colder candidate that fits without that eviction is + /// preferred. + /// + /// Candidates with an unknown size (0) are treated as fitting. pub(crate) fn find_download_opportunity(&mut self) -> Option { - let best_candidate_split_key = self.best_candidate()?; - let splits_to_delete: Vec = self - .make_room_for_split_if_necessary(best_candidate_split_key.last_accessed) - .ok()?; - let split_to_download: CandidateSplit = - self.start_download(&best_candidate_split_key.split_id)?; - Some(DownloadOpportunity { - splits_to_delete, - split_to_download, - }) + if !self.limits.skip_oversized_splits { + let best_candidate_split_key = self.best_candidate()?; + let splits_to_delete: Vec = self + .make_room_for_split_if_necessary(best_candidate_split_key.last_accessed, 0) + .ok()?; + let split_to_download: CandidateSplit = + self.start_download(&best_candidate_split_key.split_id)?; + return Some(DownloadOpportunity { + splits_to_delete, + split_to_download, + }); + } + let max_num_bytes = self.limits.max_num_bytes.as_u64(); + // Snapshot the candidate keys (hottest first) so we can mutate the table + // (evict / start a download) while iterating. `make_room` only touches + // on-disk splits and reinserts them on failure, so the snapshot stays + // valid across a skipped candidate. + let candidate_keys: Vec = + self.candidate_splits.iter().rev().cloned().collect(); + for candidate_key in candidate_keys { + let incoming_bytes = self.candidate_num_bytes(&candidate_key.split_id); + if incoming_bytes > max_num_bytes { + SEARCHER_SPLIT_CACHE_DOWNLOADS_SKIPPED_TOO_LARGE.inc(); + continue; + } + let Ok(splits_to_delete) = + self.make_room_for_split_if_necessary(candidate_key.last_accessed, incoming_bytes) + else { + // Could not make room without evicting a fresher split; try the + // next-best fitting candidate. + continue; + }; + let split_to_download: CandidateSplit = + self.start_download(&candidate_key.split_id)?; + return Some(DownloadOpportunity { + splits_to_delete, + split_to_download, + }); + } + None } #[cfg(test)] @@ -442,6 +610,12 @@ pub(crate) struct CandidateSplit { pub storage_uri: Uri, pub split_id: SplitId, pub living_token: Arc<()>, + /// Size of the split file in bytes, or 0 if unknown. + /// + /// An unknown size (0) is treated as "fits" by the download guard, so a + /// candidate discovered without a size behaves exactly as it did before + /// the guard existed. + pub num_bytes: u64, } pub(crate) struct DownloadOpportunity { @@ -491,14 +665,15 @@ mod tests { max_num_splits: NonZeroU32::new(1).unwrap(), num_concurrent_downloads: NonZeroU32::new(1).unwrap(), max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: false, }, Default::default(), ); let split_ids = sorted_split_ids(2); let split_id1 = split_ids[0].clone(); let split_id2 = split_ids[1].clone(); - split_table.report(split_id1, Uri::for_test(TEST_STORAGE_URI)); - split_table.report(split_id2.clone(), Uri::for_test(TEST_STORAGE_URI)); + split_table.report(split_id1, Uri::for_test(TEST_STORAGE_URI), 0); + split_table.report(split_id2.clone(), Uri::for_test(TEST_STORAGE_URI), 0); let candidate = split_table.best_candidate().unwrap(); assert_eq!(candidate.split_id, split_id2); } @@ -511,14 +686,15 @@ mod tests { max_num_splits: NonZeroU32::new(1).unwrap(), num_concurrent_downloads: NonZeroU32::new(1).unwrap(), max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: false, }, Default::default(), ); let split_ids = sorted_split_ids(2); let split_id1 = split_ids[0].clone(); let split_id2 = split_ids[1].clone(); - split_table.report(split_id1.clone(), Uri::for_test(TEST_STORAGE_URI)); - split_table.report(split_id2, Uri::for_test(TEST_STORAGE_URI)); + split_table.report(split_id1.clone(), Uri::for_test(TEST_STORAGE_URI), 0); + split_table.report(split_id2, Uri::for_test(TEST_STORAGE_URI), 0); let num_bytes_opt = split_table.touch(split_id1.clone(), &Uri::for_test("s3://test1/")); assert!(num_bytes_opt.is_none()); let candidate = split_table.best_candidate().unwrap(); @@ -533,11 +709,12 @@ mod tests { max_num_splits: NonZeroU32::new(1).unwrap(), num_concurrent_downloads: NonZeroU32::new(1).unwrap(), max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: false, }, Default::default(), ); let split_id1 = new_test_split_id(); - split_table.report(split_id1.clone(), Uri::for_test(TEST_STORAGE_URI)); + split_table.report(split_id1.clone(), Uri::for_test(TEST_STORAGE_URI), 0); assert_eq!(split_table.num_bytes(), 0); let download = split_table.start_download(&split_id1); assert!(download.is_some()); @@ -549,7 +726,7 @@ mod tests { Some(10_000_000) ); let split_id2 = new_test_split_id(); - split_table.report(split_id2.clone(), Uri::for_test("s3://test`/")); + split_table.report(split_id2.clone(), Uri::for_test("s3://test`/"), 0); let download = split_table.start_download(&split_id2); assert!(download.is_some()); assert!(split_table.start_download(&split_id2).is_none()); @@ -566,6 +743,7 @@ mod tests { max_num_splits: NonZeroU32::new(30).unwrap(), num_concurrent_downloads: NonZeroU32::new(1).unwrap(), max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: false, }, Default::default(), ); @@ -579,11 +757,11 @@ mod tests { (split_ids[5].clone(), 300_000), ]; for (split_id, num_bytes) in &splits { - split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI)); + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 0); split_table.register_as_downloaded(split_id.clone(), *num_bytes); } let new_split_id = new_test_split_id(); - split_table.report(new_split_id.clone(), Uri::for_test(TEST_STORAGE_URI)); + split_table.report(new_split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 0); let DownloadOpportunity { splits_to_delete, split_to_download, @@ -607,6 +785,7 @@ mod tests { max_num_splits: NonZeroU32::new(5).unwrap(), num_concurrent_downloads: NonZeroU32::new(1).unwrap(), max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: false, }, Default::default(), ); @@ -620,11 +799,11 @@ mod tests { (split_ids[5].clone(), 300_000), ]; for (split_id, num_bytes) in &splits { - split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI)); + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 0); split_table.register_as_downloaded(split_id.clone(), *num_bytes); } let new_split_id = new_test_split_id(); - split_table.report(new_split_id.clone(), Uri::for_test(TEST_STORAGE_URI)); + split_table.report(new_split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 0); let DownloadOpportunity { splits_to_delete, split_to_download, @@ -644,14 +823,15 @@ mod tests { max_num_splits: NonZeroU32::new(5).unwrap(), num_concurrent_downloads: NonZeroU32::new(1).unwrap(), max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: false, }, Default::default(), ); let split_id = new_test_split_id(); - split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI)); + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 0); let candidate = split_table.start_download(&split_id).unwrap(); // This report should be cancelled as we have a download currently running. - split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI)); + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 0); assert!(split_table.start_download(&split_id).is_none()); std::mem::drop(candidate); @@ -660,7 +840,7 @@ mod tests { assert!(split_table.start_download(&split_id).is_none()); // This report should be considered as our candidate (and its alive token has been dropped) - split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI)); + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 0); let candidate2 = split_table.start_download(&split_id).unwrap(); assert_eq!(candidate2.split_id, split_id); @@ -674,12 +854,13 @@ mod tests { max_num_splits: NonZeroU32::new(5).unwrap(), num_concurrent_downloads: NonZeroU32::new(1).unwrap(), max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: false, }, Default::default(), ); for i in 1..2_000 { let split_id = new_test_split_id(); - split_table.report(split_id, Uri::for_test(TEST_STORAGE_URI)); + split_table.report(split_id, Uri::for_test(TEST_STORAGE_URI), 0); assert_eq!( split_table.candidate_splits.len(), i.min(super::MAX_NUM_CANDIDATES) @@ -696,6 +877,7 @@ mod tests { max_num_splits: NonZeroU32::new(2).unwrap(), num_concurrent_downloads: NonZeroU32::new(1).unwrap(), max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: false, }, Default::default(), ); @@ -707,6 +889,7 @@ mod tests { storage_uri: Uri::for_test(TEST_STORAGE_URI), split_id: split_id.clone(), living_token: Arc::new(()), + num_bytes: 0, }; let split_info = SplitInfo { split_key: SplitKey { @@ -722,4 +905,301 @@ mod tests { super::MAX_NUM_CANDIDATES ); } + + #[test] + fn test_skip_download_of_oversized_split_and_advance() { + // A split larger than the whole cache budget must never be downloaded, and + // the downloader must advance to the next-best *fitting* candidate rather + // than stalling on the (permanently hot) oversized split. + let mut split_table = SplitTable::with_limits_and_existing_splits( + SplitCacheLimits { + max_num_bytes: ByteSize::mb(1), + max_num_splits: NonZeroU32::new(30).unwrap(), + num_concurrent_downloads: NonZeroU32::new(1).unwrap(), + max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: true, + }, + Default::default(), + ); + let split_ids = sorted_split_ids(2); + let small_split_id = split_ids[0].clone(); + let oversized_split_id = split_ids[1].clone(); + // Report the fitting split first, then the oversized one. Reported later + // and with the larger id, the oversized split is the hottest candidate. + split_table.report( + small_split_id.clone(), + Uri::for_test(TEST_STORAGE_URI), + 100_000, + ); + split_table.report( + oversized_split_id.clone(), + Uri::for_test(TEST_STORAGE_URI), + 2_000_000, + ); + assert_eq!( + split_table.best_candidate().unwrap().split_id, + oversized_split_id + ); + // The guard skips the oversized split and downloads the fitting one. + let opportunity = split_table.find_download_opportunity().unwrap(); + assert_eq!(opportunity.split_to_download.split_id, small_split_id); + // With only the oversized split left as a candidate there is nothing to + // download, but crucially the oversized split is never selected. + assert!(split_table.find_download_opportunity().is_none()); + } + + #[test] + fn test_oversized_only_candidate_is_never_downloaded() { + let mut split_table = SplitTable::with_limits_and_existing_splits( + SplitCacheLimits { + max_num_bytes: ByteSize::mb(1), + max_num_splits: NonZeroU32::new(30).unwrap(), + num_concurrent_downloads: NonZeroU32::new(1).unwrap(), + max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: true, + }, + Default::default(), + ); + let split_id = new_test_split_id(); + split_table.report(split_id, Uri::for_test(TEST_STORAGE_URI), 5_000_000); + assert!(split_table.find_download_opportunity().is_none()); + } + + #[test] + fn test_eviction_accounts_for_incoming_split_size() { + // Eviction must make room for the *incoming* split's bytes, not merely + // trim already-on-disk bytes, so on-disk usage never transiently exceeds + // the budget. + let mut split_table = SplitTable::with_limits_and_existing_splits( + SplitCacheLimits { + max_num_bytes: ByteSize::mb(1), + max_num_splits: NonZeroU32::new(30).unwrap(), + num_concurrent_downloads: NonZeroU32::new(1).unwrap(), + max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: true, + }, + Default::default(), + ); + let split_ids = sorted_split_ids(3); + // 3 x 300_000 = 900_000 on disk, under the 1_000_000 budget. + for split_id in &split_ids { + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 0); + split_table.register_as_downloaded(split_id.clone(), 300_000); + } + assert_eq!(split_table.num_bytes(), 900_000); + // A new 300_000-byte split: 900_000 + 300_000 > 1_000_000, so the oldest + // on-disk split must be evicted first. A size-unaware check (as before) + // would see 900_000 <= 1_000_000 and evict nothing. + let new_split_id = new_test_split_id(); + split_table.report( + new_split_id.clone(), + Uri::for_test(TEST_STORAGE_URI), + 300_000, + ); + let DownloadOpportunity { + splits_to_delete, + split_to_download, + } = split_table.find_download_opportunity().unwrap(); + assert_eq!(split_to_download.split_id, new_split_id); + assert_eq!(&splits_to_delete[..], &[split_ids[0].clone()]); + // The evicted split's bytes are already reclaimed; completing the download + // keeps on-disk usage within budget. + assert_eq!(split_table.num_bytes(), 600_000); + split_table.register_as_downloaded(new_split_id, 300_000); + assert_eq!(split_table.num_bytes(), 900_000); + } + + #[test] + fn test_unknown_size_candidate_is_not_skipped() { + // A candidate reported without a size (0) behaves exactly as before the + // guard existed: it is still selected for download, even under a tiny + // budget. + let mut split_table = SplitTable::with_limits_and_existing_splits( + SplitCacheLimits { + max_num_bytes: ByteSize::kb(1), + max_num_splits: NonZeroU32::new(30).unwrap(), + num_concurrent_downloads: NonZeroU32::new(1).unwrap(), + max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: true, + }, + Default::default(), + ); + let split_id = new_test_split_id(); + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 0); + let opportunity = split_table.find_download_opportunity().unwrap(); + assert_eq!(opportunity.split_to_download.split_id, split_id); + } + + #[test] + fn test_report_attaches_size_to_sizeless_candidate() { + let mut split_table = SplitTable::with_limits_and_existing_splits( + SplitCacheLimits { + max_num_bytes: ByteSize::kb(1), + max_num_splits: NonZeroU32::new(30).unwrap(), + num_concurrent_downloads: NonZeroU32::new(1).unwrap(), + max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: true, + }, + Default::default(), + ); + let split_id = new_test_split_id(); + // Discovered on the search path first: the size is unknown. + assert!( + split_table + .touch(split_id.clone(), &Uri::for_test(TEST_STORAGE_URI)) + .is_none() + ); + assert_eq!(split_table.candidate_num_bytes(&split_id), 0); + // A later report attaches the real size. + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 4_096); + assert_eq!(split_table.candidate_num_bytes(&split_id), 4_096); + // A second report never overwrites an already-known size. + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 9_999); + assert_eq!(split_table.candidate_num_bytes(&split_id), 4_096); + } + + #[test] + fn test_size_attached_after_touch_enables_guard() { + // Models the search path (Option A): a pre-existing oversized split is + // first discovered via `touch` with no size — so it would be downloaded — + // then its real size is attached (as `SearchSplitCache::report_split_size` + // does at split-open time). The guard must then skip it. + let mut split_table = SplitTable::with_limits_and_existing_splits( + SplitCacheLimits { + max_num_bytes: ByteSize::mb(1), + max_num_splits: NonZeroU32::new(30).unwrap(), + num_concurrent_downloads: NonZeroU32::new(1).unwrap(), + max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: true, + }, + Default::default(), + ); + let split_id = new_test_split_id(); + // Discovered via a search, size unknown: still a valid candidate. + assert!( + split_table + .touch(split_id.clone(), &Uri::for_test(TEST_STORAGE_URI)) + .is_none() + ); + assert_eq!(split_table.candidate_num_bytes(&split_id), 0); + // The real (oversized) size is attached from the split's footer offsets. + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 5_000_000); + assert_eq!(split_table.candidate_num_bytes(&split_id), 5_000_000); + // Now known to exceed the 1 MB budget, so it is never downloaded. + assert!(split_table.find_download_opportunity().is_none()); + } + + #[test] + fn test_concurrent_downloads_reserve_bytes() { + // With more than one download slot, an in-flight download's bytes must be + // reserved against the budget so two sub-budget splits can't both start + // and overshoot once they land. + let mut split_table = SplitTable::with_limits_and_existing_splits( + SplitCacheLimits { + max_num_bytes: ByteSize::mb(1), + max_num_splits: NonZeroU32::new(30).unwrap(), + num_concurrent_downloads: NonZeroU32::new(2).unwrap(), + max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: true, + }, + Default::default(), + ); + let split_ids = sorted_split_ids(2); + // Each fits alone (600K <= 1M) but not together (1.2M > 1M). + split_table.report(split_ids[0].clone(), Uri::for_test(TEST_STORAGE_URI), 600_000); + split_table.report(split_ids[1].clone(), Uri::for_test(TEST_STORAGE_URI), 600_000); + // First download starts; its bytes are now reserved as "in flight". The + // returned opportunity keeps the split's living token alive. + let first = split_table.find_download_opportunity(); + assert!(first.is_some()); + // The second download must not start: 600K in flight + 600K incoming > 1M. + assert!(split_table.find_download_opportunity().is_none()); + } + + #[test] + fn test_advances_to_next_candidate_when_room_cannot_be_made() { + // If the hottest fitting candidate can only make room by evicting a + // *fresher* on-disk split, the downloader must fall through to a colder + // candidate that fits without eviction rather than stalling. + let mut split_table = SplitTable::with_limits_and_existing_splits( + SplitCacheLimits { + max_num_bytes: ByteSize::mb(1), + max_num_splits: NonZeroU32::new(10).unwrap(), + num_concurrent_downloads: NonZeroU32::new(1).unwrap(), + max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: true, + }, + Default::default(), + ); + // A small on-disk split, freshly accessed so eviction must not touch it + // (a `touch` sets `last_accessed` to now, far fresher than the reported + // candidates below, which are stamped ~10 minutes in the past). + let on_disk_id = new_test_split_id(); + split_table.report(on_disk_id.clone(), Uri::for_test(TEST_STORAGE_URI), 0); + split_table.register_as_downloaded(on_disk_id.clone(), 100_000); + split_table.touch(on_disk_id.clone(), &Uri::for_test(TEST_STORAGE_URI)); + // Two candidates, hotter one first by id: + // - hot C1 (larger id): fits by size, but making room would evict the + // fresher on-disk split -> cannot make room; + // - cold C2 (smaller id): fits in the free space with no eviction. + let split_ids = sorted_split_ids(2); + let cold_c2 = split_ids[0].clone(); + let hot_c1 = split_ids[1].clone(); + split_table.report(cold_c2.clone(), Uri::for_test(TEST_STORAGE_URI), 50_000); + split_table.report(hot_c1.clone(), Uri::for_test(TEST_STORAGE_URI), 950_000); + let opportunity = split_table.find_download_opportunity().unwrap(); + // Advanced past C1 (could not make room) to C2. + assert_eq!(opportunity.split_to_download.split_id, cold_c2); + // The fresher on-disk split was preserved. + assert!(opportunity.splits_to_delete.is_empty()); + } + + #[test] + fn test_default_flow_ignores_split_size() { + // With the guard disabled (the default), the reported split size is + // ignored: an oversized split is still selected for download, exactly as + // before this feature existed. + let mut split_table = SplitTable::with_limits_and_existing_splits( + SplitCacheLimits { + max_num_bytes: ByteSize::kb(500), + max_num_splits: NonZeroU32::new(30).unwrap(), + num_concurrent_downloads: NonZeroU32::new(1).unwrap(), + max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: false, + }, + Default::default(), + ); + let split_id = new_test_split_id(); + // 5 MB split, far larger than the 500 KB budget. + split_table.report(split_id.clone(), Uri::for_test(TEST_STORAGE_URI), 5_000_000); + let opportunity = split_table.find_download_opportunity().unwrap(); + assert_eq!(opportunity.split_to_download.split_id, split_id); + } + + #[test] + fn test_failed_download_bytes_not_reserved() { + // A failed download leaves a dead `Downloading` entry until it is + // garbage-collected. Its bytes must not stay reserved, or an unrelated + // candidate that would otherwise fit could be blocked indefinitely. + let mut split_table = SplitTable::with_limits_and_existing_splits( + SplitCacheLimits { + max_num_bytes: ByteSize::mb(1), + max_num_splits: NonZeroU32::new(30).unwrap(), + num_concurrent_downloads: NonZeroU32::new(2).unwrap(), + max_file_descriptors: NonZeroU32::new(100).unwrap(), + skip_oversized_splits: true, + }, + Default::default(), + ); + let split_ids = sorted_split_ids(2); + split_table.report(split_ids[0].clone(), Uri::for_test(TEST_STORAGE_URI), 600_000); + split_table.report(split_ids[1].clone(), Uri::for_test(TEST_STORAGE_URI), 600_000); + // Start the first download, then simulate its failure by dropping the + // returned guard (its living token is the only strong reference). + let first = split_table.find_download_opportunity().unwrap(); + drop(first); + // The dead download's 600K must no longer be reserved, so the second + // 600K candidate can now start (600K <= 1M). + assert!(split_table.find_download_opportunity().is_some()); + } }