Merge #2981 #2996

2981: Move index swap error handling from meilisearch-http to index-scheduler r=irevoire a=loiclec

And make index_not_found error asynchronous, since we can't know whether the index will exist by the time the index swap task is processed.

Improve the index-swap test to verify that future tasks are not swapped and to test the new error messages that were introduced.

## Related issue
https://github.com/meilisearch/meilisearch/issues/2973


2996: Get rids of the unecessary tasks when an index_uid is specified r=Kerollmops a=irevoire



Co-authored-by: Loïc Lecrenier <loic.lecrenier@me.com>
Co-authored-by: Tamo <tamo@meilisearch.com>
Co-authored-by: Irevoire <tamo@meilisearch.com>

index-scheduler/src/lib.rs

@@ -54,6 +54,7 @@ use utils::{filter_out_references_to_newer_tasks, keep_tasks_within_datetimes, m
 use uuid::Uuid;
 
 use crate::index_mapper::IndexMapper;
+use crate::utils::check_index_swap_validity;
 
 pub(crate) type BEI128 =
     meilisearch_types::heed::zerocopy::I128<meilisearch_types::heed::byteorder::BE>;
@@ -589,10 +590,12 @@ impl IndexScheduler {
     ) -> Result<RoaringBitmap> {
         let mut tasks = self.get_task_ids(rtxn, query)?;
 
-        // If the query contains a list of index_uid, then we must exclude IndexSwap tasks
-        // from the result (because it is not publicly associated with any index)
+        // If the query contains a list of `index_uid`, then we must exclude all the kind that
+        // arn't associated to one and only one index.
         if query.index_uid.is_some() {
-            tasks -= self.get_kind(rtxn, Kind::IndexSwap)?
+            for kind in enum_iterator::all::<Kind>().filter(|kind| !kind.related_to_one_index()) {
+                tasks -= self.get_kind(rtxn, kind)?;
+            }
         }
 
         // Any task that is internally associated with a non-authorized index
@@ -671,6 +674,10 @@ impl IndexScheduler {
         // For deletion and cancelation tasks, we want to make extra sure that they
         // don't attempt to delete/cancel tasks that are newer than themselves.
         filter_out_references_to_newer_tasks(&mut task);
+        // If the register task is an index swap task, verify that it is well-formed
+        // (that it does not contain duplicate indexes).
+        check_index_swap_validity(&task)?;
+
         // Get rid of the mutability.
         let task = task;
 
@@ -1586,17 +1593,17 @@ mod tests {
             })
             .unwrap();
         index_scheduler.assert_internally_consistent();
-
-        handle.wait_till(Breakpoint::AfterProcessing);
-        index_scheduler.assert_internally_consistent();
-        snapshot!(snapshot_index_scheduler(&index_scheduler), name: "first_swap_processed");
-
         index_scheduler
             .register(KindWithContent::IndexSwap {
                 swaps: vec![IndexSwap { indexes: ("a".to_owned(), "c".to_owned()) }],
             })
             .unwrap();
         index_scheduler.assert_internally_consistent();
+        snapshot!(snapshot_index_scheduler(&index_scheduler), name: "two_swaps_registered");
+
+        handle.wait_till(Breakpoint::AfterProcessing);
+        index_scheduler.assert_internally_consistent();
+        snapshot!(snapshot_index_scheduler(&index_scheduler), name: "first_swap_processed");
 
         handle.wait_till(Breakpoint::AfterProcessing);
         index_scheduler.assert_internally_consistent();
@@ -1607,6 +1614,57 @@ mod tests {
         snapshot!(snapshot_index_scheduler(&index_scheduler), name: "third_empty_swap_processed");
     }
 
+    #[test]
+    fn swap_indexes_errors() {
+        let (index_scheduler, handle) = IndexScheduler::test(true, vec![]);
+
+        let to_enqueue = [
+            index_creation_task("a", "id"),
+            index_creation_task("b", "id"),
+            index_creation_task("c", "id"),
+            index_creation_task("d", "id"),
+        ];
+
+        for task in to_enqueue {
+            let _ = index_scheduler.register(task).unwrap();
+            index_scheduler.assert_internally_consistent();
+        }
+        handle.advance_n_batch(4);
+        index_scheduler.assert_internally_consistent();
+
+        let first_snap = snapshot_index_scheduler(&index_scheduler);
+        snapshot!(first_snap, name: "initial_tasks_processed");
+
+        let err = index_scheduler
+            .register(KindWithContent::IndexSwap {
+                swaps: vec![
+                    IndexSwap { indexes: ("a".to_owned(), "b".to_owned()) },
+                    IndexSwap { indexes: ("b".to_owned(), "a".to_owned()) },
+                ],
+            })
+            .unwrap_err();
+        snapshot!(format!("{err}"), @"Indexes must be declared only once during a swap. `a`, `b` were specified several times.");
+
+        index_scheduler.assert_internally_consistent();
+        let second_snap = snapshot_index_scheduler(&index_scheduler);
+        assert_eq!(first_snap, second_snap);
+
+        // Index `e` does not exist, but we don't check its existence yet
+        index_scheduler
+            .register(KindWithContent::IndexSwap {
+                swaps: vec![
+                    IndexSwap { indexes: ("a".to_owned(), "b".to_owned()) },
+                    IndexSwap { indexes: ("c".to_owned(), "e".to_owned()) },
+                    IndexSwap { indexes: ("d".to_owned(), "f".to_owned()) },
+                ],
+            })
+            .unwrap();
+        handle.advance_n_batch(1);
+        // Now the first swap should have an error message saying `e` and `f` do not exist
+        index_scheduler.assert_internally_consistent();
+        snapshot!(snapshot_index_scheduler(&index_scheduler), name: "first_swap_failed");
+    }
+
     #[test]
     fn document_addition_and_index_deletion_on_unexisting_index() {
         let (index_scheduler, handle) = IndexScheduler::test(true, vec![]);