diff --git a/src/database/index.rs b/src/database/index.rs index 64dba325..08ffbd53 100644 --- a/src/database/index.rs +++ b/src/database/index.rs @@ -473,7 +473,11 @@ impl Database { for (pid, mut uris) in by_pid { // Counted before the hot-tail skip, so the gauge reports the true total. uncovered_total = uncovered_total.saturating_add(uris.len() as u64); - skipped_today = skipped_today.saturating_add(drop_hot_partition(&mut uris, skip_today.as_deref())); + // A file at the hot-tail target is never repacked, so indexing it cannot race the + // packer; until then carried-forward entries leave it searched as hundreds. + let target = self.config.maintenance.timefusion_light_optimize_target_size.max(1) as u64; + let converged = |uri: &str| parquet_rel_of_uri(uri).and_then(|rel| sizes.get(rel)).is_some_and(|size| *size >= target); + skipped_today = skipped_today.saturating_add(drop_hot_partition(&mut uris, skip_today.as_deref(), converged)); sort_backfill_uris_newest_first(&mut uris); let queue = uris.into_iter().filter_map(|uri| Some((parquet_rel_of_uri(&uri)?.to_string(), uri))).collect::>(); queues.push((pid, queue)); diff --git a/src/database/mod.rs b/src/database/mod.rs index d69f16b9..af6c5024 100644 --- a/src/database/mod.rs +++ b/src/database/mod.rs @@ -2940,12 +2940,12 @@ pub struct Database { repair_pass_permit: Arc, } -/// Drop the hot (today) partition from a backfill queue, returning how many URIs were removed. -/// `marker` is `None` when the skip is disabled. -fn drop_hot_partition(uris: &mut Vec, marker: Option<&str>) -> u64 { +/// Drop the hot (today) partition from a backfill queue, except the files `converged` names, +/// returning how many URIs were removed. `marker` is `None` when the skip is disabled. +fn drop_hot_partition(uris: &mut Vec, marker: Option<&str>, converged: impl Fn(&str) -> bool) -> u64 { let Some(marker) = marker else { return 0 }; let before = uris.len(); - uris.retain(|uri| !uri.contains(marker)); + uris.retain(|uri| !uri.contains(marker) || converged(uri)); (before - uris.len()) as u64 } diff --git a/src/database/tests.rs b/src/database/tests.rs index 0fb54327..31485405 100644 --- a/src/database/tests.rs +++ b/src/database/tests.rs @@ -4186,13 +4186,14 @@ fn tantivy_backfill_prioritizes_recent_partitions() { /// The hot-tail skip: today's partition leaves the backfill queue, every other date /// stays. Today's files are rewritten within hours and already covered at birth by /// the flush callback and the inline-reindex paths. -#[test_case::test_case(None => (3, 0) ; "disabled keeps the whole queue")] -#[test_case::test_case(Some("date=2026-08-22") => (2, 1) ; "today is dropped")] -#[test_case::test_case(Some("date=2026-08-2") => (0, 3) ; "a prefix marker is the caller's error, not silently ignored")] -#[test_case::test_case(Some("date=2026-01-01") => (3, 0) ; "a date with no files drops nothing")] +#[test_case::test_case(None => (4, 0) ; "disabled keeps the whole queue")] +#[test_case::test_case(Some("date=2026-08-22") => (3, 1) ; "today is dropped except its converged file")] +#[test_case::test_case(Some("date=2026-08-2") => (1, 3) ; "a prefix marker is the caller's error, not silently ignored")] +#[test_case::test_case(Some("date=2026-01-01") => (4, 0) ; "a date with no files drops nothing")] fn tantivy_backfill_drops_the_hot_partition(marker: Option<&str>) -> (usize, u64) { - let mut uris = ["date=2026-08-22/a.parquet", "date=2026-08-21/b.parquet", "date=2026-08-20/c.parquet"].map(String::from).to_vec(); - let dropped = super::drop_hot_partition(&mut uris, marker); + let mut uris = + ["date=2026-08-22/a.parquet", "date=2026-08-22/converged.parquet", "date=2026-08-21/b.parquet", "date=2026-08-20/c.parquet"].map(String::from).to_vec(); + let dropped = super::drop_hot_partition(&mut uris, marker, |uri| uri.contains("converged")); (uris.len(), dropped) }