tantivy: backfill today's converged files

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::<VecDeque<_>>();
                 queues.push((pid, queue));

src/database/mod.rs

@@ -2940,12 +2940,12 @@ pub struct Database {
     repair_pass_permit: Arc<tokio::sync::Semaphore>,
 }
 
-/// 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<String>, 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<String>, 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
 }
 

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)
 }