Repository: meilisearch/meilisearch
File: crates/index-scheduler/src/scheduler/process_batch.rs (Rust)

Code before:
            } else {
                tracing::debug!(
                    "Not rollbacking an upgrade targetting the earlier version v{}.{}.{}",
                    bin_major,
                    bin_minor,
                    bin_patch
                )
            }
        }

        // 3. If we are cancelling a compaction task, remove the tempfiles after incomplete compactions
        for compaction_task in &tasks_to_cancel & compaction_tasks {
            progress.update_progress(TaskCancelationProgress::CleaningCompactionLeftover);
            let task = self.queue.tasks.get_task(rtxn, compaction_task)?.unwrap();
            let Some(Details::IndexCompaction {
                index_uid,
                pre_compaction_size: _,
                post_compaction_size: _,
            }) = task.details
            else {
                unreachable!("wrong details for compaction task {compaction_task}")
            };

            let index_path = match self.index_mapper.index_mapping.get(rtxn, &index_uid)? {
                Some(index_uuid) => self.index_mapper.index_path(index_uuid),
                None => return Err(Error::IndexNotFound(index_uid)),
            };

            if let Err(e) = remove_file(index_path.join(DATA_MDB_COPY_NAME)) {
                match e.kind() {
                    ErrorKind::NotFound => (),
                    _ => return Err(Error::IoError(e)),
                }
            }
        }

        // 4. We now have a list of tasks to cancel, cancel them
        let (task_progress, progress_obj) = AtomicTaskStep::new(tasks_to_cancel.len() as u32);
        progress.update_progress(progress_obj);

        let mut tasks = self.queue.tasks.get_existing_tasks(
            rtxn,
            tasks_to_cancel.iter().inspect(|_| {
                task_progress.fetch_add(1, Ordering::Relaxed);
            }),
        )?;

        progress.update_progress(TaskCancelationProgress::UpdatingTasks);
        let (task_progress, progress_obj) = AtomicTaskStep::new(tasks_to_cancel.len() as u32);
        progress.update_progress(progress_obj);
        for task in tasks.iter_mut() {

Review comment (suggestion):
should that be an error? Like, if we're cancelling a task that is for an index that no longer exists, I think no cleanup should be necessary, but no error either

Code after:
            } else {
                tracing::debug!(
                    "Not rollbacking an upgrade targetting the earlier version v{}.{}.{}",
                    bin_major,
                    bin_minor,
                    bin_patch
                )
            }
        }

        // 3. If we are cancelling a compaction task, remove the tempfiles after incomplete compactions
        for compaction_task in &tasks_to_cancel & compaction_tasks {
            progress.update_progress(TaskCancelationProgress::CleaningCompactionLeftover);
            let task = self.queue.tasks.get_task(rtxn, compaction_task)?.unwrap();
            let Some(Details::IndexCompaction {
                index_uid,
                pre_compaction_size: _,
                post_compaction_size: _,
            }) = task.details
            else {
                unreachable!("wrong details for compaction task {compaction_task}")
            };

            let index_path = match self.index_mapper.index_mapping.get(rtxn, &index_uid)? {
                Some(index_uuid) => self.index_mapper.index_path(index_uuid),
                None => continue,
            };

            if let Err(e) = remove_file(index_path.join(DATA_MDB_COPY_NAME)) {
                match e.kind() {
                    ErrorKind::NotFound => (),
                    _ => return Err(Error::IoError(e)),
                }
            }
        }

        // 4. We now have a list of tasks to cancel, cancel them
        let (task_progress, progress_obj) = AtomicTaskStep::new(tasks_to_cancel.len() as u32);
        progress.update_progress(progress_obj);

        let mut tasks = self.queue.tasks.get_existing_tasks(
            rtxn,
            tasks_to_cancel.iter().inspect(|_| {
                task_progress.fetch_add(1, Ordering::Relaxed);
            }),
        )?;

        progress.update_progress(TaskCancelationProgress::UpdatingTasks);
        let (task_progress, progress_obj) = AtomicTaskStep::new(tasks_to_cancel.len() as u32);
        progress.update_progress(progress_obj);
        for task in tasks.iter_mut() {