2022-09-07 06:10:14 +08:00
|
|
|
//! Utility functions on the DBs. Mainly getter and setters.
|
|
|
|
|
2022-10-27 15:41:32 +08:00
|
|
|
use std::collections::{BTreeSet, HashSet};
|
2022-10-19 18:59:12 +08:00
|
|
|
use std::ops::Bound;
|
|
|
|
|
2024-11-15 00:31:02 +08:00
|
|
|
use meilisearch_types::batches::{Batch, BatchId};
|
2023-11-23 01:21:19 +08:00
|
|
|
use meilisearch_types::heed::types::DecodeIgnore;
|
2022-10-21 00:00:07 +08:00
|
|
|
use meilisearch_types::heed::{Database, RoTxn, RwTxn};
|
2023-11-23 01:21:19 +08:00
|
|
|
use meilisearch_types::milli::CboRoaringBitmapCodec;
|
2022-10-26 18:57:29 +08:00
|
|
|
use meilisearch_types::tasks::{Details, IndexSwap, Kind, KindWithContent, Status};
|
2022-10-18 17:02:46 +08:00
|
|
|
use roaring::{MultiOps, RoaringBitmap};
|
2022-10-19 18:59:12 +08:00
|
|
|
use time::OffsetDateTime;
|
2022-09-07 06:10:14 +08:00
|
|
|
|
2024-11-15 00:31:02 +08:00
|
|
|
use crate::{Error, IndexScheduler, ProcessingTasks, Result, Task, TaskId, BEI128};
|
|
|
|
|
|
|
|
/// This structure contains all the information required to write a batch in the database without reading the tasks.
|
|
|
|
/// It'll stay in RAM so it must be small.
|
|
|
|
pub(crate) struct CachedBatch {
|
|
|
|
uid: BatchId,
|
|
|
|
statuses: HashSet<Status>,
|
|
|
|
kinds: HashSet<Kind>,
|
|
|
|
indexes: HashSet<String>,
|
|
|
|
canceled_by: HashSet<TaskId>,
|
|
|
|
oldest_enqueued_at: Option<OffsetDateTime>,
|
|
|
|
earliest_enqueued_at: Option<OffsetDateTime>,
|
|
|
|
started_at: OffsetDateTime,
|
|
|
|
finished_at: OffsetDateTime,
|
|
|
|
}
|
|
|
|
|
|
|
|
impl CachedBatch {
|
|
|
|
pub fn new(uid: BatchId, started_at: OffsetDateTime, finished_at: OffsetDateTime) -> Self {
|
|
|
|
Self {
|
|
|
|
uid,
|
|
|
|
statuses: HashSet::default(),
|
|
|
|
kinds: HashSet::default(),
|
|
|
|
indexes: HashSet::default(),
|
|
|
|
canceled_by: HashSet::default(),
|
|
|
|
oldest_enqueued_at: None,
|
|
|
|
earliest_enqueued_at: None,
|
|
|
|
started_at,
|
|
|
|
finished_at,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn update(&mut self, task: &Task) {
|
|
|
|
self.statuses.insert(task.status);
|
|
|
|
self.kinds.insert(task.kind.as_kind());
|
|
|
|
self.indexes.extend(task.indexes().iter().map(|s| s.to_string()));
|
|
|
|
if let Some(canceled_by) = task.canceled_by {
|
|
|
|
self.canceled_by.insert(canceled_by);
|
|
|
|
}
|
|
|
|
self.oldest_enqueued_at =
|
|
|
|
Some(self.oldest_enqueued_at.map_or(task.enqueued_at, |oldest_enqueued_at| {
|
|
|
|
task.enqueued_at.min(oldest_enqueued_at)
|
|
|
|
}));
|
|
|
|
self.earliest_enqueued_at =
|
|
|
|
Some(self.earliest_enqueued_at.map_or(task.enqueued_at, |earliest_enqueued_at| {
|
|
|
|
task.enqueued_at.max(earliest_enqueued_at)
|
|
|
|
}));
|
|
|
|
}
|
|
|
|
}
|
2022-09-07 06:10:14 +08:00
|
|
|
|
|
|
|
impl IndexScheduler {
|
2022-10-17 18:58:20 +08:00
|
|
|
pub(crate) fn all_task_ids(&self, rtxn: &RoTxn) -> Result<RoaringBitmap> {
|
2022-10-22 22:35:42 +08:00
|
|
|
enum_iterator::all().map(|s| self.get_status(rtxn, s)).union()
|
2022-10-17 18:58:20 +08:00
|
|
|
}
|
2022-10-18 17:02:46 +08:00
|
|
|
|
2024-11-15 00:31:02 +08:00
|
|
|
pub(crate) fn all_batch_ids(&self, rtxn: &RoTxn) -> Result<RoaringBitmap> {
|
|
|
|
enum_iterator::all().map(|s| self.get_batch_status(rtxn, s)).union()
|
|
|
|
}
|
|
|
|
|
2022-09-07 17:21:53 +08:00
|
|
|
pub(crate) fn last_task_id(&self, rtxn: &RoTxn) -> Result<Option<TaskId>> {
|
2023-11-23 01:21:19 +08:00
|
|
|
Ok(self.all_tasks.remap_data_type::<DecodeIgnore>().last(rtxn)?.map(|(k, _)| k + 1))
|
2022-09-07 17:21:53 +08:00
|
|
|
}
|
|
|
|
|
|
|
|
pub(crate) fn next_task_id(&self, rtxn: &RoTxn) -> Result<TaskId> {
|
|
|
|
Ok(self.last_task_id(rtxn)?.unwrap_or_default())
|
|
|
|
}
|
|
|
|
|
2024-11-13 18:27:12 +08:00
|
|
|
pub(crate) fn next_batch_id(&self, rtxn: &RoTxn) -> Result<BatchId> {
|
|
|
|
Ok(self
|
|
|
|
.all_batches
|
|
|
|
.remap_data_type::<DecodeIgnore>()
|
|
|
|
.last(rtxn)?
|
|
|
|
.map(|(k, _)| k + 1)
|
|
|
|
.unwrap_or_default())
|
|
|
|
}
|
|
|
|
|
2022-09-07 06:10:14 +08:00
|
|
|
pub(crate) fn get_task(&self, rtxn: &RoTxn, task_id: TaskId) -> Result<Option<Task>> {
|
2023-11-23 01:21:19 +08:00
|
|
|
Ok(self.all_tasks.get(rtxn, &task_id)?)
|
2022-09-07 06:10:14 +08:00
|
|
|
}
|
|
|
|
|
2024-11-15 00:31:02 +08:00
|
|
|
pub(crate) fn get_batch(&self, rtxn: &RoTxn, batch_id: BatchId) -> Result<Option<Batch>> {
|
|
|
|
Ok(self.all_batches.get(rtxn, &batch_id)?)
|
|
|
|
}
|
|
|
|
|
|
|
|
pub(crate) fn write_batch(&self, wtxn: &mut RwTxn, batch: CachedBatch) -> Result<()> {
|
|
|
|
self.all_batches.put(
|
|
|
|
wtxn,
|
|
|
|
&batch.uid,
|
|
|
|
&Batch {
|
|
|
|
uid: batch.uid,
|
|
|
|
started_at: batch.started_at,
|
|
|
|
finished_at: Some(batch.finished_at),
|
|
|
|
},
|
|
|
|
)?;
|
|
|
|
|
|
|
|
for status in batch.statuses {
|
|
|
|
self.update_batch_status(wtxn, status, |bitmap| {
|
|
|
|
bitmap.insert(batch.uid);
|
|
|
|
})?;
|
|
|
|
}
|
|
|
|
|
|
|
|
for kind in batch.kinds {
|
|
|
|
self.update_batch_kind(wtxn, kind, |bitmap| {
|
|
|
|
bitmap.insert(batch.uid);
|
|
|
|
})?;
|
|
|
|
}
|
|
|
|
|
|
|
|
for index in batch.indexes {
|
|
|
|
self.update_batch_index(wtxn, &index, |bitmap| {
|
|
|
|
bitmap.insert(batch.uid);
|
|
|
|
})?;
|
|
|
|
}
|
|
|
|
if let Some(enqueued_at) = batch.oldest_enqueued_at {
|
|
|
|
insert_task_datetime(wtxn, self.batch_enqueued_at, enqueued_at, batch.uid)?;
|
|
|
|
}
|
|
|
|
if let Some(enqueued_at) = batch.earliest_enqueued_at {
|
|
|
|
insert_task_datetime(wtxn, self.batch_enqueued_at, enqueued_at, batch.uid)?;
|
|
|
|
}
|
|
|
|
insert_task_datetime(wtxn, self.batch_started_at, batch.started_at, batch.uid)?;
|
|
|
|
insert_task_datetime(wtxn, self.batch_finished_at, batch.finished_at, batch.uid)?;
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2024-11-13 18:27:12 +08:00
|
|
|
/// Convert an iterator to a `Vec` of tasks. The tasks MUST exist or a
|
|
|
|
/// `CorruptedTaskQueue` error will be throwed.
|
|
|
|
pub(crate) fn get_existing_tasks_with_batch_id(
|
|
|
|
&self,
|
|
|
|
rtxn: &RoTxn,
|
|
|
|
batch_id: BatchId,
|
|
|
|
tasks: impl IntoIterator<Item = TaskId>,
|
|
|
|
) -> Result<Vec<Task>> {
|
|
|
|
tasks
|
|
|
|
.into_iter()
|
|
|
|
.map(|task_id| {
|
|
|
|
self.get_task(rtxn, task_id)
|
|
|
|
.and_then(|task| task.ok_or(Error::CorruptedTaskQueue))
|
|
|
|
.map(|task| task.with_batch_id(batch_id))
|
|
|
|
})
|
|
|
|
.collect::<Result<_>>()
|
|
|
|
}
|
|
|
|
|
2022-09-07 06:10:14 +08:00
|
|
|
/// Convert an iterator to a `Vec` of tasks. The tasks MUST exist or a
|
|
|
|
/// `CorruptedTaskQueue` error will be throwed.
|
|
|
|
pub(crate) fn get_existing_tasks(
|
|
|
|
&self,
|
|
|
|
rtxn: &RoTxn,
|
|
|
|
tasks: impl IntoIterator<Item = TaskId>,
|
|
|
|
) -> Result<Vec<Task>> {
|
|
|
|
tasks
|
|
|
|
.into_iter()
|
|
|
|
.map(|task_id| {
|
2022-10-21 00:00:07 +08:00
|
|
|
self.get_task(rtxn, task_id).and_then(|task| task.ok_or(Error::CorruptedTaskQueue))
|
2022-09-07 06:10:14 +08:00
|
|
|
})
|
|
|
|
.collect::<Result<_>>()
|
|
|
|
}
|
|
|
|
|
2024-11-15 00:31:02 +08:00
|
|
|
/// Convert an iterator to a `Vec` of batches. The batches MUST exist or a
|
|
|
|
/// `CorruptedTaskQueue` error will be throwed.
|
|
|
|
pub(crate) fn get_existing_batches(
|
|
|
|
&self,
|
|
|
|
rtxn: &RoTxn,
|
|
|
|
tasks: impl IntoIterator<Item = BatchId>,
|
|
|
|
) -> Result<Vec<Batch>> {
|
|
|
|
tasks
|
|
|
|
.into_iter()
|
|
|
|
.map(|batch_id| {
|
|
|
|
self.get_batch(rtxn, batch_id)
|
|
|
|
.and_then(|task| task.ok_or(Error::CorruptedTaskQueue))
|
|
|
|
})
|
|
|
|
.collect::<Result<_>>()
|
|
|
|
}
|
|
|
|
|
2022-09-16 07:58:08 +08:00
|
|
|
pub(crate) fn update_task(&self, wtxn: &mut RwTxn, task: &Task) -> Result<()> {
|
2022-10-21 00:00:07 +08:00
|
|
|
let old_task = self.get_task(wtxn, task.uid)?.ok_or(Error::CorruptedTaskQueue)?;
|
2022-09-08 02:08:07 +08:00
|
|
|
|
2022-09-26 19:46:34 +08:00
|
|
|
debug_assert_eq!(old_task.uid, task.uid);
|
2024-11-15 00:31:02 +08:00
|
|
|
debug_assert!(old_task.batch_uid.is_none() && task.batch_uid.is_some());
|
2022-09-26 19:46:34 +08:00
|
|
|
|
2024-11-15 00:31:02 +08:00
|
|
|
// TODO: This shouldn't ever happen, we should assert it
|
2022-09-26 19:46:34 +08:00
|
|
|
if old_task == *task {
|
|
|
|
return Ok(());
|
|
|
|
}
|
|
|
|
|
2022-09-08 02:08:07 +08:00
|
|
|
if old_task.status != task.status {
|
2022-09-08 02:44:33 +08:00
|
|
|
self.update_status(wtxn, old_task.status, |bitmap| {
|
2022-09-08 02:08:07 +08:00
|
|
|
bitmap.remove(task.uid);
|
|
|
|
})?;
|
2022-09-08 02:44:33 +08:00
|
|
|
self.update_status(wtxn, task.status, |bitmap| {
|
2022-09-08 02:08:07 +08:00
|
|
|
bitmap.insert(task.uid);
|
|
|
|
})?;
|
|
|
|
}
|
|
|
|
|
|
|
|
if old_task.kind.as_kind() != task.kind.as_kind() {
|
2022-09-08 02:44:33 +08:00
|
|
|
self.update_kind(wtxn, old_task.kind.as_kind(), |bitmap| {
|
2022-09-08 02:08:07 +08:00
|
|
|
bitmap.remove(task.uid);
|
|
|
|
})?;
|
2022-09-08 02:44:33 +08:00
|
|
|
self.update_kind(wtxn, task.kind.as_kind(), |bitmap| {
|
2022-09-08 02:08:07 +08:00
|
|
|
bitmap.insert(task.uid);
|
|
|
|
})?;
|
|
|
|
}
|
|
|
|
|
2022-10-19 23:29:51 +08:00
|
|
|
assert_eq!(
|
|
|
|
old_task.enqueued_at, task.enqueued_at,
|
2022-10-19 22:18:00 +08:00
|
|
|
"Cannot update a task's enqueued_at time"
|
|
|
|
);
|
2022-10-19 18:59:12 +08:00
|
|
|
if old_task.started_at != task.started_at {
|
2022-10-21 00:00:07 +08:00
|
|
|
assert!(old_task.started_at.is_none(), "Cannot update a task's started_at time");
|
2022-10-19 18:59:12 +08:00
|
|
|
if let Some(started_at) = task.started_at {
|
|
|
|
insert_task_datetime(wtxn, self.started_at, started_at, task.uid)?;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if old_task.finished_at != task.finished_at {
|
2022-10-21 00:00:07 +08:00
|
|
|
assert!(old_task.finished_at.is_none(), "Cannot update a task's finished_at time");
|
2022-10-19 18:59:12 +08:00
|
|
|
if let Some(finished_at) = task.finished_at {
|
|
|
|
insert_task_datetime(wtxn, self.finished_at, finished_at, task.uid)?;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-11-23 01:21:19 +08:00
|
|
|
self.all_tasks.put(wtxn, &task.uid, task)?;
|
2022-09-08 02:08:07 +08:00
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2022-10-13 17:07:36 +08:00
|
|
|
/// Returns the whole set of tasks that belongs to this index.
|
|
|
|
pub(crate) fn index_tasks(&self, rtxn: &RoTxn, index: &str) -> Result<RoaringBitmap> {
|
2022-10-03 21:29:37 +08:00
|
|
|
Ok(self.index_tasks.get(rtxn, index)?.unwrap_or_default())
|
2022-09-07 06:10:14 +08:00
|
|
|
}
|
|
|
|
|
2022-09-07 06:22:58 +08:00
|
|
|
pub(crate) fn update_index(
|
2022-09-07 06:10:14 +08:00
|
|
|
&self,
|
|
|
|
wtxn: &mut RwTxn,
|
2022-09-07 06:22:58 +08:00
|
|
|
index: &str,
|
2022-09-08 02:44:33 +08:00
|
|
|
f: impl Fn(&mut RoaringBitmap),
|
2022-09-07 06:10:14 +08:00
|
|
|
) -> Result<()> {
|
2022-10-13 17:07:36 +08:00
|
|
|
let mut tasks = self.index_tasks(wtxn, index)?;
|
2022-09-08 02:44:33 +08:00
|
|
|
f(&mut tasks);
|
2022-10-20 15:56:43 +08:00
|
|
|
if tasks.is_empty() {
|
|
|
|
self.index_tasks.delete(wtxn, index)?;
|
|
|
|
} else {
|
|
|
|
self.index_tasks.put(wtxn, index, &tasks)?;
|
|
|
|
}
|
2022-09-07 06:22:58 +08:00
|
|
|
|
|
|
|
Ok(())
|
2022-09-07 06:10:14 +08:00
|
|
|
}
|
|
|
|
|
2024-11-15 00:31:02 +08:00
|
|
|
/// Returns the whole set of batches that belongs to this index.
|
|
|
|
pub(crate) fn index_batches(&self, rtxn: &RoTxn, index: &str) -> Result<RoaringBitmap> {
|
|
|
|
Ok(self.batch_index_tasks.get(rtxn, index)?.unwrap_or_default())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub(crate) fn update_batch_index(
|
|
|
|
&self,
|
|
|
|
wtxn: &mut RwTxn,
|
|
|
|
index: &str,
|
|
|
|
f: impl Fn(&mut RoaringBitmap),
|
|
|
|
) -> Result<()> {
|
|
|
|
let mut batches = self.index_batches(wtxn, index)?;
|
|
|
|
f(&mut batches);
|
|
|
|
if batches.is_empty() {
|
|
|
|
self.batch_index_tasks.delete(wtxn, index)?;
|
|
|
|
} else {
|
|
|
|
self.batch_index_tasks.put(wtxn, index, &batches)?;
|
|
|
|
}
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2022-09-07 06:22:58 +08:00
|
|
|
pub(crate) fn get_status(&self, rtxn: &RoTxn, status: Status) -> Result<RoaringBitmap> {
|
2022-10-24 21:12:01 +08:00
|
|
|
Ok(self.status.get(rtxn, &status)?.unwrap_or_default())
|
2022-09-07 06:10:14 +08:00
|
|
|
}
|
|
|
|
|
2022-09-07 06:22:58 +08:00
|
|
|
pub(crate) fn put_status(
|
2022-09-07 06:10:14 +08:00
|
|
|
&self,
|
|
|
|
wtxn: &mut RwTxn,
|
2022-09-07 06:22:58 +08:00
|
|
|
status: Status,
|
2022-09-07 06:10:14 +08:00
|
|
|
bitmap: &RoaringBitmap,
|
|
|
|
) -> Result<()> {
|
2022-09-07 06:22:58 +08:00
|
|
|
Ok(self.status.put(wtxn, &status, bitmap)?)
|
2022-09-07 06:10:14 +08:00
|
|
|
}
|
|
|
|
|
|
|
|
pub(crate) fn update_status(
|
|
|
|
&self,
|
|
|
|
wtxn: &mut RwTxn,
|
|
|
|
status: Status,
|
2022-09-08 02:44:33 +08:00
|
|
|
f: impl Fn(&mut RoaringBitmap),
|
2022-09-07 06:10:14 +08:00
|
|
|
) -> Result<()> {
|
2022-10-03 21:29:37 +08:00
|
|
|
let mut tasks = self.get_status(wtxn, status)?;
|
2022-09-08 02:44:33 +08:00
|
|
|
f(&mut tasks);
|
2022-09-07 06:10:14 +08:00
|
|
|
self.put_status(wtxn, status, &tasks)?;
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2024-11-15 00:31:02 +08:00
|
|
|
pub(crate) fn get_batch_status(&self, rtxn: &RoTxn, status: Status) -> Result<RoaringBitmap> {
|
|
|
|
Ok(self.batch_status.get(rtxn, &status)?.unwrap_or_default())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub(crate) fn put_batch_status(
|
|
|
|
&self,
|
|
|
|
wtxn: &mut RwTxn,
|
|
|
|
status: Status,
|
|
|
|
bitmap: &RoaringBitmap,
|
|
|
|
) -> Result<()> {
|
|
|
|
Ok(self.batch_status.put(wtxn, &status, bitmap)?)
|
|
|
|
}
|
|
|
|
|
|
|
|
pub(crate) fn update_batch_status(
|
|
|
|
&self,
|
|
|
|
wtxn: &mut RwTxn,
|
|
|
|
status: Status,
|
|
|
|
f: impl Fn(&mut RoaringBitmap),
|
|
|
|
) -> Result<()> {
|
|
|
|
let mut tasks = self.get_batch_status(wtxn, status)?;
|
|
|
|
f(&mut tasks);
|
|
|
|
self.put_batch_status(wtxn, status, &tasks)?;
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2022-09-07 06:22:58 +08:00
|
|
|
pub(crate) fn get_kind(&self, rtxn: &RoTxn, kind: Kind) -> Result<RoaringBitmap> {
|
2022-10-03 21:29:37 +08:00
|
|
|
Ok(self.kind.get(rtxn, &kind)?.unwrap_or_default())
|
2022-09-07 06:22:58 +08:00
|
|
|
}
|
|
|
|
|
|
|
|
pub(crate) fn put_kind(
|
|
|
|
&self,
|
|
|
|
wtxn: &mut RwTxn,
|
|
|
|
kind: Kind,
|
|
|
|
bitmap: &RoaringBitmap,
|
|
|
|
) -> Result<()> {
|
|
|
|
Ok(self.kind.put(wtxn, &kind, bitmap)?)
|
|
|
|
}
|
|
|
|
|
2022-09-07 06:10:14 +08:00
|
|
|
pub(crate) fn update_kind(
|
|
|
|
&self,
|
|
|
|
wtxn: &mut RwTxn,
|
|
|
|
kind: Kind,
|
2022-09-08 02:44:33 +08:00
|
|
|
f: impl Fn(&mut RoaringBitmap),
|
2022-09-07 06:10:14 +08:00
|
|
|
) -> Result<()> {
|
2022-10-03 21:29:37 +08:00
|
|
|
let mut tasks = self.get_kind(wtxn, kind)?;
|
2022-09-08 02:44:33 +08:00
|
|
|
f(&mut tasks);
|
2022-09-07 06:10:14 +08:00
|
|
|
self.put_kind(wtxn, kind, &tasks)?;
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
2024-11-15 00:31:02 +08:00
|
|
|
|
|
|
|
pub(crate) fn get_batch_kind(&self, rtxn: &RoTxn, kind: Kind) -> Result<RoaringBitmap> {
|
|
|
|
Ok(self.batch_kind.get(rtxn, &kind)?.unwrap_or_default())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub(crate) fn put_batch_kind(
|
|
|
|
&self,
|
|
|
|
wtxn: &mut RwTxn,
|
|
|
|
kind: Kind,
|
|
|
|
bitmap: &RoaringBitmap,
|
|
|
|
) -> Result<()> {
|
|
|
|
Ok(self.batch_kind.put(wtxn, &kind, bitmap)?)
|
|
|
|
}
|
|
|
|
|
|
|
|
pub(crate) fn update_batch_kind(
|
|
|
|
&self,
|
|
|
|
wtxn: &mut RwTxn,
|
|
|
|
kind: Kind,
|
|
|
|
f: impl Fn(&mut RoaringBitmap),
|
|
|
|
) -> Result<()> {
|
|
|
|
let mut tasks = self.get_batch_kind(wtxn, kind)?;
|
|
|
|
f(&mut tasks);
|
|
|
|
self.put_batch_kind(wtxn, kind, &tasks)?;
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
2022-09-07 06:10:14 +08:00
|
|
|
}
|
2022-10-19 18:59:12 +08:00
|
|
|
|
|
|
|
pub(crate) fn insert_task_datetime(
|
|
|
|
wtxn: &mut RwTxn,
|
2023-11-23 01:21:19 +08:00
|
|
|
database: Database<BEI128, CboRoaringBitmapCodec>,
|
2022-10-19 18:59:12 +08:00
|
|
|
time: OffsetDateTime,
|
|
|
|
task_id: TaskId,
|
|
|
|
) -> Result<()> {
|
2023-11-23 01:21:19 +08:00
|
|
|
let timestamp = time.unix_timestamp_nanos();
|
2022-10-22 22:35:42 +08:00
|
|
|
let mut task_ids = database.get(wtxn, ×tamp)?.unwrap_or_default();
|
2022-10-19 18:59:12 +08:00
|
|
|
task_ids.insert(task_id);
|
2022-10-25 21:30:36 +08:00
|
|
|
database.put(wtxn, ×tamp, &RoaringBitmap::from_iter(task_ids))?;
|
2022-10-19 18:59:12 +08:00
|
|
|
Ok(())
|
|
|
|
}
|
2022-10-19 22:13:14 +08:00
|
|
|
|
2022-10-19 18:59:12 +08:00
|
|
|
pub(crate) fn remove_task_datetime(
|
|
|
|
wtxn: &mut RwTxn,
|
2023-11-23 01:21:19 +08:00
|
|
|
database: Database<BEI128, CboRoaringBitmapCodec>,
|
2022-10-19 18:59:12 +08:00
|
|
|
time: OffsetDateTime,
|
|
|
|
task_id: TaskId,
|
|
|
|
) -> Result<()> {
|
2023-11-23 01:21:19 +08:00
|
|
|
let timestamp = time.unix_timestamp_nanos();
|
2022-10-22 22:35:42 +08:00
|
|
|
if let Some(mut existing) = database.get(wtxn, ×tamp)? {
|
2022-10-19 18:59:12 +08:00
|
|
|
existing.remove(task_id);
|
|
|
|
if existing.is_empty() {
|
|
|
|
database.delete(wtxn, ×tamp)?;
|
|
|
|
} else {
|
2022-10-25 21:30:36 +08:00
|
|
|
database.put(wtxn, ×tamp, &RoaringBitmap::from_iter(existing))?;
|
2022-10-19 18:59:12 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
2022-10-19 22:13:14 +08:00
|
|
|
|
2024-11-15 00:31:02 +08:00
|
|
|
// TODO: Rename the function since it also applies to batches
|
2022-10-19 18:59:12 +08:00
|
|
|
pub(crate) fn keep_tasks_within_datetimes(
|
|
|
|
rtxn: &RoTxn,
|
|
|
|
tasks: &mut RoaringBitmap,
|
2023-11-23 01:21:19 +08:00
|
|
|
database: Database<BEI128, CboRoaringBitmapCodec>,
|
2022-10-19 18:59:12 +08:00
|
|
|
after: Option<OffsetDateTime>,
|
|
|
|
before: Option<OffsetDateTime>,
|
|
|
|
) -> Result<()> {
|
|
|
|
let (start, end) = match (&after, &before) {
|
|
|
|
(None, None) => return Ok(()),
|
|
|
|
(None, Some(before)) => (Bound::Unbounded, Bound::Excluded(*before)),
|
|
|
|
(Some(after), None) => (Bound::Excluded(*after), Bound::Unbounded),
|
|
|
|
(Some(after), Some(before)) => (Bound::Excluded(*after), Bound::Excluded(*before)),
|
|
|
|
};
|
|
|
|
let mut collected_task_ids = RoaringBitmap::new();
|
2023-11-23 01:21:19 +08:00
|
|
|
let start = map_bound(start, |b| b.unix_timestamp_nanos());
|
|
|
|
let end = map_bound(end, |b| b.unix_timestamp_nanos());
|
2022-10-22 22:35:42 +08:00
|
|
|
let iter = database.range(rtxn, &(start, end))?;
|
2022-10-19 18:59:12 +08:00
|
|
|
for r in iter {
|
|
|
|
let (_timestamp, task_ids) = r?;
|
|
|
|
collected_task_ids |= task_ids;
|
|
|
|
}
|
|
|
|
*tasks &= collected_task_ids;
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
// TODO: remove when Bound::map ( https://github.com/rust-lang/rust/issues/86026 ) is available on stable
|
2022-10-24 19:32:46 +08:00
|
|
|
pub(crate) fn map_bound<T, U>(bound: Bound<T>, map: impl FnOnce(T) -> U) -> Bound<U> {
|
2022-10-19 18:59:12 +08:00
|
|
|
match bound {
|
|
|
|
Bound::Included(x) => Bound::Included(map(x)),
|
|
|
|
Bound::Excluded(x) => Bound::Excluded(map(x)),
|
|
|
|
Bound::Unbounded => Bound::Unbounded,
|
|
|
|
}
|
|
|
|
}
|
2022-10-17 22:30:18 +08:00
|
|
|
|
|
|
|
pub fn swap_index_uid_in_task(task: &mut Task, swap: (&str, &str)) {
|
|
|
|
use KindWithContent as K;
|
|
|
|
let mut index_uids = vec![];
|
|
|
|
match &mut task.kind {
|
2022-10-22 00:03:10 +08:00
|
|
|
K::DocumentAdditionOrUpdate { index_uid, .. } => index_uids.push(index_uid),
|
2024-05-08 21:26:21 +08:00
|
|
|
K::DocumentEdition { index_uid, .. } => index_uids.push(index_uid),
|
2022-10-17 22:30:18 +08:00
|
|
|
K::DocumentDeletion { index_uid, .. } => index_uids.push(index_uid),
|
2023-03-07 17:02:04 +08:00
|
|
|
K::DocumentDeletionByFilter { index_uid, .. } => index_uids.push(index_uid),
|
2022-10-17 22:30:18 +08:00
|
|
|
K::DocumentClear { index_uid } => index_uids.push(index_uid),
|
2022-10-22 00:03:10 +08:00
|
|
|
K::SettingsUpdate { index_uid, .. } => index_uids.push(index_uid),
|
2022-10-17 22:30:18 +08:00
|
|
|
K::IndexDeletion { index_uid } => index_uids.push(index_uid),
|
|
|
|
K::IndexCreation { index_uid, .. } => index_uids.push(index_uid),
|
|
|
|
K::IndexUpdate { index_uid, .. } => index_uids.push(index_uid),
|
|
|
|
K::IndexSwap { swaps } => {
|
2022-10-26 18:57:29 +08:00
|
|
|
for IndexSwap { indexes: (lhs, rhs) } in swaps.iter_mut() {
|
2022-10-22 22:35:42 +08:00
|
|
|
if lhs == swap.0 || lhs == swap.1 {
|
2022-10-17 22:30:18 +08:00
|
|
|
index_uids.push(lhs);
|
|
|
|
}
|
2022-10-22 22:35:42 +08:00
|
|
|
if rhs == swap.0 || rhs == swap.1 {
|
2022-10-17 22:30:18 +08:00
|
|
|
index_uids.push(rhs);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2022-10-25 01:08:15 +08:00
|
|
|
K::TaskCancelation { .. }
|
|
|
|
| K::TaskDeletion { .. }
|
|
|
|
| K::DumpCreation { .. }
|
2022-10-25 16:44:58 +08:00
|
|
|
| K::SnapshotCreation => (),
|
2022-10-17 22:30:18 +08:00
|
|
|
};
|
2022-10-25 16:58:55 +08:00
|
|
|
if let Some(Details::IndexSwap { swaps }) = &mut task.details {
|
2022-10-26 18:57:29 +08:00
|
|
|
for IndexSwap { indexes: (lhs, rhs) } in swaps.iter_mut() {
|
2022-10-25 16:58:55 +08:00
|
|
|
if lhs == swap.0 || lhs == swap.1 {
|
|
|
|
index_uids.push(lhs);
|
|
|
|
}
|
|
|
|
if rhs == swap.0 || rhs == swap.1 {
|
|
|
|
index_uids.push(rhs);
|
2022-10-24 14:12:03 +08:00
|
|
|
}
|
2022-10-25 15:48:51 +08:00
|
|
|
}
|
2022-10-24 14:12:03 +08:00
|
|
|
}
|
2022-10-17 22:30:18 +08:00
|
|
|
for index_uid in index_uids {
|
2022-10-22 22:35:42 +08:00
|
|
|
if index_uid == swap.0 {
|
2024-05-14 23:20:57 +08:00
|
|
|
swap.1.clone_into(index_uid);
|
2022-10-22 22:35:42 +08:00
|
|
|
} else if index_uid == swap.1 {
|
2024-05-14 23:20:57 +08:00
|
|
|
swap.0.clone_into(index_uid);
|
2022-10-17 22:30:18 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2022-10-25 22:10:14 +08:00
|
|
|
|
|
|
|
/// Remove references to task ids that are greater than the id of the given task.
|
|
|
|
pub(crate) fn filter_out_references_to_newer_tasks(task: &mut Task) {
|
|
|
|
let new_nbr_of_matched_tasks = match &mut task.kind {
|
|
|
|
KindWithContent::TaskCancelation { tasks, .. }
|
|
|
|
| KindWithContent::TaskDeletion { tasks, .. } => {
|
|
|
|
tasks.remove_range(task.uid..);
|
|
|
|
tasks.len()
|
|
|
|
}
|
|
|
|
_ => return,
|
|
|
|
};
|
2022-10-27 00:03:48 +08:00
|
|
|
if let Some(
|
|
|
|
Details::TaskCancelation { matched_tasks, .. }
|
|
|
|
| Details::TaskDeletion { matched_tasks, .. },
|
|
|
|
) = &mut task.details
|
|
|
|
{
|
|
|
|
*matched_tasks = new_nbr_of_matched_tasks;
|
2022-10-25 22:10:14 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-10-27 15:41:32 +08:00
|
|
|
pub(crate) fn check_index_swap_validity(task: &Task) -> Result<()> {
|
|
|
|
let swaps =
|
|
|
|
if let KindWithContent::IndexSwap { swaps } = &task.kind { swaps } else { return Ok(()) };
|
|
|
|
let mut all_indexes = HashSet::new();
|
|
|
|
let mut duplicate_indexes = BTreeSet::new();
|
|
|
|
for IndexSwap { indexes: (lhs, rhs) } in swaps {
|
|
|
|
for name in [lhs, rhs] {
|
|
|
|
let is_new = all_indexes.insert(name);
|
|
|
|
if !is_new {
|
|
|
|
duplicate_indexes.insert(name);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if !duplicate_indexes.is_empty() {
|
|
|
|
if duplicate_indexes.len() == 1 {
|
|
|
|
return Err(Error::SwapDuplicateIndexFound(
|
|
|
|
duplicate_indexes.into_iter().next().unwrap().clone(),
|
|
|
|
));
|
|
|
|
} else {
|
|
|
|
return Err(Error::SwapDuplicateIndexesFound(
|
|
|
|
duplicate_indexes.into_iter().cloned().collect(),
|
|
|
|
));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2022-11-29 17:38:27 +08:00
|
|
|
/// Clamp the provided value to be a multiple of system page size.
|
|
|
|
pub fn clamp_to_page_size(size: usize) -> usize {
|
|
|
|
size / page_size::get() * page_size::get()
|
|
|
|
}
|
|
|
|
|
2022-10-20 19:11:50 +08:00
|
|
|
#[cfg(test)]
|
|
|
|
impl IndexScheduler {
|
|
|
|
/// Asserts that the index scheduler's content is internally consistent.
|
|
|
|
pub fn assert_internally_consistent(&self) {
|
|
|
|
let rtxn = self.env.read_txn().unwrap();
|
|
|
|
for task in self.all_tasks.iter(&rtxn).unwrap() {
|
|
|
|
let (task_id, task) = task.unwrap();
|
|
|
|
let task_index_uid = task.index_uid().map(ToOwned::to_owned);
|
|
|
|
|
|
|
|
let Task {
|
|
|
|
uid,
|
2024-11-13 18:27:12 +08:00
|
|
|
/// We should iterate over the list of batch to ensure this task is effectively in the right batch
|
|
|
|
batch_uid,
|
2022-10-20 19:11:50 +08:00
|
|
|
enqueued_at,
|
|
|
|
started_at,
|
|
|
|
finished_at,
|
|
|
|
error: _,
|
|
|
|
canceled_by,
|
|
|
|
details,
|
|
|
|
status,
|
|
|
|
kind,
|
2022-10-25 21:35:06 +08:00
|
|
|
} = task;
|
2022-10-20 19:11:50 +08:00
|
|
|
assert_eq!(uid, task.uid);
|
|
|
|
if let Some(task_index_uid) = &task_index_uid {
|
|
|
|
assert!(self
|
|
|
|
.index_tasks
|
|
|
|
.get(&rtxn, task_index_uid.as_str())
|
|
|
|
.unwrap()
|
|
|
|
.unwrap()
|
|
|
|
.contains(task.uid));
|
|
|
|
}
|
2023-11-23 19:07:35 +08:00
|
|
|
let db_enqueued_at =
|
|
|
|
self.enqueued_at.get(&rtxn, &enqueued_at.unix_timestamp_nanos()).unwrap().unwrap();
|
2022-10-20 19:11:50 +08:00
|
|
|
assert!(db_enqueued_at.contains(task_id));
|
|
|
|
if let Some(started_at) = started_at {
|
|
|
|
let db_started_at = self
|
|
|
|
.started_at
|
2023-11-23 19:07:35 +08:00
|
|
|
.get(&rtxn, &started_at.unix_timestamp_nanos())
|
2022-10-20 19:11:50 +08:00
|
|
|
.unwrap()
|
|
|
|
.unwrap();
|
|
|
|
assert!(db_started_at.contains(task_id));
|
|
|
|
}
|
|
|
|
if let Some(finished_at) = finished_at {
|
|
|
|
let db_finished_at = self
|
|
|
|
.finished_at
|
2023-11-23 19:07:35 +08:00
|
|
|
.get(&rtxn, &finished_at.unix_timestamp_nanos())
|
2022-10-20 19:11:50 +08:00
|
|
|
.unwrap()
|
|
|
|
.unwrap();
|
|
|
|
assert!(db_finished_at.contains(task_id));
|
|
|
|
}
|
|
|
|
if let Some(canceled_by) = canceled_by {
|
|
|
|
let db_canceled_tasks = self.get_status(&rtxn, Status::Canceled).unwrap();
|
2022-10-25 17:42:14 +08:00
|
|
|
assert!(db_canceled_tasks.contains(uid));
|
2022-10-20 19:11:50 +08:00
|
|
|
let db_canceling_task = self.get_task(&rtxn, canceled_by).unwrap().unwrap();
|
|
|
|
assert_eq!(db_canceling_task.status, Status::Succeeded);
|
|
|
|
match db_canceling_task.kind {
|
|
|
|
KindWithContent::TaskCancelation { query: _, tasks } => {
|
|
|
|
assert!(tasks.contains(uid));
|
|
|
|
}
|
|
|
|
_ => panic!(),
|
|
|
|
}
|
|
|
|
}
|
2022-10-27 00:54:15 +08:00
|
|
|
if let Some(details) = details {
|
|
|
|
match details {
|
2022-10-27 00:27:43 +08:00
|
|
|
Details::IndexSwap { swaps: sw1 } => {
|
|
|
|
if let KindWithContent::IndexSwap { swaps: sw2 } = &kind {
|
2022-10-24 14:12:03 +08:00
|
|
|
assert_eq!(&sw1, sw2);
|
|
|
|
}
|
2022-10-27 00:27:43 +08:00
|
|
|
}
|
2022-10-20 19:11:50 +08:00
|
|
|
Details::DocumentAdditionOrUpdate { received_documents, indexed_documents } => {
|
|
|
|
assert_eq!(kind.as_kind(), Kind::DocumentAdditionOrUpdate);
|
2022-11-28 23:27:41 +08:00
|
|
|
match indexed_documents {
|
|
|
|
Some(indexed_documents) => {
|
2023-01-24 00:32:13 +08:00
|
|
|
assert!(matches!(
|
|
|
|
status,
|
|
|
|
Status::Succeeded | Status::Failed | Status::Canceled
|
|
|
|
));
|
|
|
|
match status {
|
|
|
|
Status::Succeeded => assert!(indexed_documents <= received_documents),
|
|
|
|
Status::Failed | Status::Canceled => assert_eq!(indexed_documents, 0),
|
2024-05-08 21:26:21 +08:00
|
|
|
status => panic!("DocumentAddition can't have an indexed_documents set if it's {}", status),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
None => {
|
|
|
|
assert!(matches!(status, Status::Enqueued | Status::Processing))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
Details::DocumentEdition { edited_documents, .. } => {
|
|
|
|
assert_eq!(kind.as_kind(), Kind::DocumentEdition);
|
|
|
|
match edited_documents {
|
|
|
|
Some(edited_documents) => {
|
|
|
|
assert!(matches!(
|
|
|
|
status,
|
|
|
|
Status::Succeeded | Status::Failed | Status::Canceled
|
|
|
|
));
|
|
|
|
match status {
|
|
|
|
Status::Succeeded => (),
|
|
|
|
Status::Failed | Status::Canceled => assert_eq!(edited_documents, 0),
|
|
|
|
status => panic!("DocumentEdition can't have an edited_documents set if it's {}", status),
|
2023-01-24 00:32:13 +08:00
|
|
|
}
|
2022-11-28 23:27:41 +08:00
|
|
|
}
|
|
|
|
None => {
|
2023-01-24 00:32:13 +08:00
|
|
|
assert!(matches!(status, Status::Enqueued | Status::Processing))
|
2022-11-28 23:27:41 +08:00
|
|
|
}
|
2022-10-20 19:11:50 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
Details::SettingsUpdate { settings: _ } => {
|
|
|
|
assert_eq!(kind.as_kind(), Kind::SettingsUpdate);
|
|
|
|
}
|
|
|
|
Details::IndexInfo { primary_key: pk1 } => match &kind {
|
|
|
|
KindWithContent::IndexCreation { index_uid, primary_key: pk2 }
|
|
|
|
| KindWithContent::IndexUpdate { index_uid, primary_key: pk2 } => {
|
|
|
|
self.index_tasks
|
|
|
|
.get(&rtxn, index_uid.as_str())
|
|
|
|
.unwrap()
|
|
|
|
.unwrap()
|
|
|
|
.contains(uid);
|
|
|
|
assert_eq!(&pk1, pk2);
|
|
|
|
}
|
|
|
|
_ => panic!(),
|
|
|
|
},
|
2022-10-25 17:02:26 +08:00
|
|
|
Details::DocumentDeletion {
|
2022-11-28 23:27:41 +08:00
|
|
|
provided_ids: received_document_ids,
|
2022-10-25 17:02:26 +08:00
|
|
|
deleted_documents,
|
|
|
|
} => {
|
2023-02-09 18:29:13 +08:00
|
|
|
assert_eq!(kind.as_kind(), Kind::DocumentDeletion);
|
2023-02-15 00:45:46 +08:00
|
|
|
let (index_uid, documents_ids) =
|
|
|
|
if let KindWithContent::DocumentDeletion {
|
|
|
|
ref index_uid,
|
|
|
|
ref documents_ids,
|
|
|
|
} = kind
|
|
|
|
{
|
|
|
|
(index_uid, documents_ids)
|
|
|
|
} else {
|
|
|
|
unreachable!()
|
|
|
|
};
|
2023-02-09 18:29:13 +08:00
|
|
|
assert_eq!(&task_index_uid.unwrap(), index_uid);
|
|
|
|
|
|
|
|
match status {
|
|
|
|
Status::Enqueued | Status::Processing => (),
|
|
|
|
Status::Succeeded => {
|
|
|
|
assert!(deleted_documents.unwrap() <= received_document_ids as u64);
|
|
|
|
assert!(documents_ids.len() == received_document_ids);
|
|
|
|
}
|
|
|
|
Status::Failed | Status::Canceled => {
|
|
|
|
assert!(deleted_documents == Some(0));
|
|
|
|
assert!(documents_ids.len() == received_document_ids);
|
2022-10-20 19:11:50 +08:00
|
|
|
}
|
|
|
|
}
|
2023-03-07 17:02:04 +08:00
|
|
|
}
|
|
|
|
Details::DocumentDeletionByFilter { deleted_documents, original_filter: _ } => {
|
2023-05-30 21:18:52 +08:00
|
|
|
assert_eq!(kind.as_kind(), Kind::DocumentDeletion);
|
2023-03-07 17:02:04 +08:00
|
|
|
let (index_uid, _) = if let KindWithContent::DocumentDeletionByFilter {
|
|
|
|
ref index_uid,
|
|
|
|
ref filter_expr,
|
|
|
|
} = kind
|
|
|
|
{
|
|
|
|
(index_uid, filter_expr)
|
|
|
|
} else {
|
|
|
|
unreachable!()
|
|
|
|
};
|
|
|
|
assert_eq!(&task_index_uid.unwrap(), index_uid);
|
|
|
|
|
|
|
|
match status {
|
|
|
|
Status::Enqueued | Status::Processing => (),
|
|
|
|
Status::Succeeded => {
|
|
|
|
assert!(deleted_documents.is_some());
|
|
|
|
}
|
|
|
|
Status::Failed | Status::Canceled => {
|
|
|
|
assert!(deleted_documents == Some(0));
|
|
|
|
}
|
|
|
|
}
|
2022-10-20 19:11:50 +08:00
|
|
|
}
|
|
|
|
Details::ClearAll { deleted_documents } => {
|
|
|
|
assert!(matches!(
|
|
|
|
kind.as_kind(),
|
|
|
|
Kind::DocumentDeletion | Kind::IndexDeletion
|
|
|
|
));
|
|
|
|
if deleted_documents.is_some() {
|
|
|
|
assert_eq!(status, Status::Succeeded);
|
|
|
|
} else {
|
|
|
|
assert_ne!(status, Status::Succeeded);
|
|
|
|
}
|
|
|
|
}
|
2022-11-28 23:27:41 +08:00
|
|
|
Details::TaskCancelation { matched_tasks, canceled_tasks, original_filter } => {
|
2022-10-20 19:11:50 +08:00
|
|
|
if let Some(canceled_tasks) = canceled_tasks {
|
|
|
|
assert_eq!(status, Status::Succeeded);
|
|
|
|
assert!(canceled_tasks <= matched_tasks);
|
|
|
|
match &kind {
|
|
|
|
KindWithContent::TaskCancelation { query, tasks } => {
|
2022-11-28 23:27:41 +08:00
|
|
|
assert_eq!(query, &original_filter);
|
2022-10-20 19:11:50 +08:00
|
|
|
assert_eq!(tasks.len(), matched_tasks);
|
|
|
|
}
|
|
|
|
_ => panic!(),
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
assert_ne!(status, Status::Succeeded);
|
|
|
|
}
|
|
|
|
}
|
2022-11-28 23:27:41 +08:00
|
|
|
Details::TaskDeletion { matched_tasks, deleted_tasks, original_filter } => {
|
2022-10-20 19:11:50 +08:00
|
|
|
if let Some(deleted_tasks) = deleted_tasks {
|
|
|
|
assert_eq!(status, Status::Succeeded);
|
|
|
|
assert!(deleted_tasks <= matched_tasks);
|
|
|
|
match &kind {
|
|
|
|
KindWithContent::TaskDeletion { query, tasks } => {
|
2022-11-28 23:27:41 +08:00
|
|
|
assert_eq!(query, &original_filter);
|
2022-10-20 19:11:50 +08:00
|
|
|
assert_eq!(tasks.len(), matched_tasks);
|
|
|
|
}
|
|
|
|
_ => panic!(),
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
assert_ne!(status, Status::Succeeded);
|
|
|
|
}
|
|
|
|
}
|
2022-11-28 23:27:41 +08:00
|
|
|
Details::Dump { dump_uid: _ } => {
|
|
|
|
assert_eq!(kind.as_kind(), Kind::DumpCreation);
|
2022-10-20 19:11:50 +08:00
|
|
|
}
|
2022-10-27 00:54:15 +08:00
|
|
|
}
|
2022-10-20 19:11:50 +08:00
|
|
|
}
|
|
|
|
|
|
|
|
assert!(self.get_status(&rtxn, status).unwrap().contains(uid));
|
|
|
|
assert!(self.get_kind(&rtxn, kind.as_kind()).unwrap().contains(uid));
|
|
|
|
|
2022-10-27 00:27:43 +08:00
|
|
|
if let KindWithContent::DocumentAdditionOrUpdate { content_file, .. } = kind {
|
|
|
|
match status {
|
2022-10-20 19:11:50 +08:00
|
|
|
Status::Enqueued | Status::Processing => {
|
2023-01-24 23:17:23 +08:00
|
|
|
assert!(self
|
2023-01-25 01:34:36 +08:00
|
|
|
.file_store
|
|
|
|
.all_uuids()
|
|
|
|
.unwrap()
|
|
|
|
.any(|uuid| uuid.as_ref().unwrap() == &content_file),
|
|
|
|
"Could not find uuid `{content_file}` in the file_store. Available uuids are {:?}.",
|
2023-01-25 18:20:15 +08:00
|
|
|
self.file_store.all_uuids().unwrap().collect::<std::result::Result<Vec<_>, file_store::Error>>().unwrap(),
|
2023-01-24 00:32:13 +08:00
|
|
|
);
|
2022-10-20 19:11:50 +08:00
|
|
|
}
|
|
|
|
Status::Succeeded | Status::Failed | Status::Canceled => {
|
2023-01-24 23:17:23 +08:00
|
|
|
assert!(self
|
|
|
|
.file_store
|
|
|
|
.all_uuids()
|
|
|
|
.unwrap()
|
2023-01-25 01:34:36 +08:00
|
|
|
.all(|uuid| uuid.as_ref().unwrap() != &content_file));
|
2022-10-20 19:11:50 +08:00
|
|
|
}
|
2022-10-27 00:27:43 +08:00
|
|
|
}
|
2022-10-20 19:11:50 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2023-02-15 19:29:36 +08:00
|
|
|
|
|
|
|
pub fn dichotomic_search(start_point: usize, mut is_good: impl FnMut(usize) -> bool) -> usize {
|
|
|
|
let mut biggest_good = None;
|
|
|
|
let mut smallest_bad = None;
|
|
|
|
let mut current = start_point;
|
|
|
|
loop {
|
|
|
|
let is_good = is_good(current);
|
|
|
|
|
|
|
|
(biggest_good, smallest_bad, current) = match (biggest_good, smallest_bad, is_good) {
|
|
|
|
(None, None, false) => (None, Some(current), current / 2),
|
|
|
|
(None, None, true) => (Some(current), None, current * 2),
|
|
|
|
(None, Some(smallest_bad), true) => {
|
|
|
|
(Some(current), Some(smallest_bad), (current + smallest_bad) / 2)
|
|
|
|
}
|
|
|
|
(None, Some(_), false) => (None, Some(current), current / 2),
|
|
|
|
(Some(_), None, true) => (Some(current), None, current * 2),
|
|
|
|
(Some(biggest_good), None, false) => {
|
|
|
|
(Some(biggest_good), Some(current), (biggest_good + current) / 2)
|
|
|
|
}
|
|
|
|
(Some(_), Some(smallest_bad), true) => {
|
|
|
|
(Some(current), Some(smallest_bad), (smallest_bad + current) / 2)
|
|
|
|
}
|
|
|
|
(Some(biggest_good), Some(_), false) => {
|
|
|
|
(Some(biggest_good), Some(current), (biggest_good + current) / 2)
|
|
|
|
}
|
|
|
|
};
|
|
|
|
if current == 0 {
|
|
|
|
return current;
|
|
|
|
}
|
|
|
|
if smallest_bad.is_some() && biggest_good.is_some() && biggest_good >= Some(current) {
|
|
|
|
return current;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|