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

Code before:
                )
            }
        }

        // 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() {
            task.status = Status::Canceled;
            task.canceled_by = Some(cancel_task_id);
            task.details = task.details.as_ref().map(|d| d.to_failed());
            current_batch.processing(Some(task));
            task_progress.fetch_add(1, Ordering::Relaxed);
        }

Review comment (bug):
nitpick: should we return an error here, or maybe just log it?

Code after:
                )
            }
        }

        // 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() {
            task.status = Status::Canceled;
            task.canceled_by = Some(cancel_task_id);
            task.details = task.details.as_ref().map(|d| d.to_failed());
            current_batch.processing(Some(task));
            task_progress.fetch_add(1, Ordering::Relaxed);
        }