fix infinite loop in split cache (#5334)

* always call truncate_candidate_list from a consistent state

* Update quickwit/quickwit-storage/src/split_cache/split_table.rs

Co-authored-by: François Massot <[email protected]>

---------

Co-authored-by: François Massot <[email protected]>

quickwit/quickwit-storage/src/split_cache/split_table.rs

@@ -206,9 +206,10 @@ impl SplitTable {
     fn insert(&mut self, split_info: SplitInfo) {
         let was_not_in_queue = match split_info.status {
             Status::Candidate { .. } => {
-                let was_not_in_queue = self.candidate_splits.insert(split_info.split_key);
+                // we truncate *before* inserting, otherwise way may end up in an inconsistent
+                // state which make truncate_candidate_list loop indefinitely
                 self.truncate_candidate_list();
-                was_not_in_queue
+                self.candidate_splits.insert(split_info.split_key)
             }
             Status::Downloading { .. } => self.downloading_splits.insert(split_info.split_key),
             Status::OnDisk { num_bytes } => {
@@ -224,6 +225,8 @@ impl SplitTable {
                 self.on_disk_splits.insert(split_info.split_key)
             }
         };
+        // this is fine to do in an inconsistent state, the last entry will just be ignored while
+        // gcing
         self.gc_downloading_splits_if_necessary();
         assert!(was_not_in_queue);
         let split_ulid_was_absent = self
@@ -323,7 +326,8 @@ impl SplitTable {
 
     /// Make sure we have at most `MAX_CANDIDATES` candidate splits.
     fn truncate_candidate_list(&mut self) {
-        while self.candidate_splits.len() > MAX_NUM_CANDIDATES {
+        // we remove one more to make place for one candidate about to be inserted
+        while self.candidate_splits.len() >= MAX_NUM_CANDIDATES {
             let worst_candidate = self.candidate_splits.first().unwrap().split_ulid;
             self.remove(worst_candidate);
         }