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

Code before:
                    task.details = task.details.map(|d| d.to_failed());

                    #[cfg(test)]
                    self.maybe_fail(tests::FailureLocation::UpdatingTaskAfterProcessBatchFailure)?;

                    tracing::info!("Batch failed {}", error);

                    self.update_task(&mut wtxn, &task)
                        .map_err(|e| Error::TaskDatabaseUpdate(Box::new(e)))?;
                }
            }
        }

        let processed = self.processing_tasks.write().unwrap().stop_processing();

        #[cfg(test)]
        self.maybe_fail(tests::FailureLocation::CommittingWtxn)?;

        wtxn.commit().map_err(Error::HeedTransaction)?;

        // Once the tasks are commited, we should delete all the update files associated ASAP to avoid leaking files in case of a restart
        tracing::debug!("Deleting the upadate files");

        ids.into_par_iter().try_for_each(|id| -> Result<()> {
            let rtxn = self.read_txn()?;
            let task = self
                .get_task(&rtxn, id)
                .map_err(|e| Error::TaskDatabaseUpdate(Box::new(e)))?
                .ok_or(Error::CorruptedTaskQueue)?;
            if let Err(e) = self.delete_persisted_task_data(&task) {
                tracing::error!(
                    "Failure to delete the content files associated with task {}. Error: {e}",
                    task.uid
                );
            }
            Ok(())
        })?;

        // We shouldn't crash the tick function if we can't send data to the webhook.
        let _ = self.notify_webhook(&processed);

        #[cfg(test)]
        self.breakpoint(Breakpoint::AfterProcessing);

        Ok(TickOutcome::TickAgain(processed_tasks))
    }

    /// Once the tasks changes have been commited we must send all the tasks that were updated to our webhook if there is one.
    fn notify_webhook(&self, updated: &RoaringBitmap) -> Result<()> {
        if let Some(ref url) = self.webhook_url {
            struct TaskReader<'a, 'b> {

Review comment (performance):
it would be better to create the read txn once (using the appropriate feature). Opening a lot of transactions in parallel can be slower than using the same one on different threads. What do you think?

Code after:

                    #[cfg(test)]
                    self.maybe_fail(tests::FailureLocation::UpdatingTaskAfterProcessBatchFailure)?;

                    tracing::info!("Batch failed {}", error);

                    self.update_task(&mut wtxn, &task)
                        .map_err(|e| Error::TaskDatabaseUpdate(Box::new(e)))?;
                }
            }
        }

        let processed = self.processing_tasks.write().unwrap().stop_processing();

        #[cfg(test)]
        self.maybe_fail(tests::FailureLocation::CommittingWtxn)?;

        wtxn.commit().map_err(Error::HeedTransaction)?;

        // Once the tasks are commited, we should delete all the update files associated ASAP to avoid leaking files in case of a restart
        tracing::debug!("Deleting the upadate files");

        //We take one read transaction **per thread**. Then, every thread is going to pull out new IDs from the roaring bitmap with the help of an atomic shared index into the bitmap
        let idx = AtomicU32::new(0);
        (0..current_num_threads()).into_par_iter().try_for_each(|_| -> Result<()> {
            let rtxn = self.read_txn()?;
            while let Some(id) = ids.select(idx.fetch_add(1, Ordering::Relaxed)) {
                let task = self
                    .get_task(&rtxn, id)
                    .map_err(|e| Error::TaskDatabaseUpdate(Box::new(e)))?
                    .ok_or(Error::CorruptedTaskQueue)?;
                if let Err(e) = self.delete_persisted_task_data(&task) {
                    tracing::error!(
                        "Failure to delete the content files associated with task {}. Error: {e}",
                        task.uid
                    );
                }
            }
            Ok(())
        })?;

        // We shouldn't crash the tick function if we can't send data to the webhook.
        let _ = self.notify_webhook(&processed);

        #[cfg(test)]
        self.breakpoint(Breakpoint::AfterProcessing);

        Ok(TickOutcome::TickAgain(processed_tasks))
    }

    /// Once the tasks changes have been commited we must send all the tasks that were updated to our webhook if there is one.