searchengines-web/MeiliSearch
searchengines-web/MeiliSearch
1
0
Fork 0
mirror of https://github.com/meilisearch/MeiliSearch synced 2024-06-06 12:57:48 +02:00
Code Issues Releases Wiki Activity

Merge #2447

Browse Source 
2447: move index uid in task content r=Kerollmops a=MarinPostma

this pr moves the index_uid from the `Task` to the `TaskContent`. This is because the task can now have content that do not target a particular index.


Co-authored-by: ad hoc <postma.marin@protonmail.com>
This commit is contained in:
bors[bot] 2022-06-02 13:54:09 +00:00 committed by GitHub
commit 953a209f02
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 452 additions and 345 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
meilisearch-http/src/routes/tasks.rs
View File

@ -38,9 +38,9 @@ fn task_type_matches_content(type_: &TaskType, content: &TaskContent) -> bool {
matches!((type_, content), matches!((type_, content),
(TaskType::IndexCreation, TaskContent::IndexCreation { .. }) (TaskType::IndexCreation, TaskContent::IndexCreation { .. })
| (TaskType::IndexUpdate, TaskContent::IndexUpdate { .. }) | (TaskType::IndexUpdate, TaskContent::IndexUpdate { .. })
| (TaskType::IndexDeletion, TaskContent::IndexDeletion) | (TaskType::IndexDeletion, TaskContent::IndexDeletion { .. })
| (TaskType::DocumentAdditionOrUpdate, TaskContent::DocumentAddition { .. }) | (TaskType::DocumentAdditionOrUpdate, TaskContent::DocumentAddition { .. })
| (TaskType::DocumentDeletion, TaskContent::DocumentDeletion(_)) | (TaskType::DocumentDeletion, TaskContent::DocumentDeletion{ .. })
| (TaskType::SettingsUpdate, TaskContent::SettingsUpdate { .. }) | (TaskType::SettingsUpdate, TaskContent::SettingsUpdate { .. })
) )
} }

26
meilisearch-http/src/task.rs
View File

@ -30,9 +30,9 @@ impl From<TaskContent> for TaskType {
match other { match other {
TaskContent::IndexCreation { .. } => TaskType::IndexCreation, TaskContent::IndexCreation { .. } => TaskType::IndexCreation,
TaskContent::IndexUpdate { .. } => TaskType::IndexUpdate, TaskContent::IndexUpdate { .. } => TaskType::IndexUpdate,
TaskContent::IndexDeletion => TaskType::IndexDeletion, TaskContent::IndexDeletion { .. } => TaskType::IndexDeletion,
TaskContent::DocumentAddition { .. } => TaskType::DocumentAdditionOrUpdate, TaskContent::DocumentAddition { .. } => TaskType::DocumentAdditionOrUpdate,
TaskContent::DocumentDeletion(_) => TaskType::DocumentDeletion, TaskContent::DocumentDeletion { .. } => TaskType::DocumentDeletion,
TaskContent::SettingsUpdate { .. } => TaskType::SettingsUpdate, TaskContent::SettingsUpdate { .. } => TaskType::SettingsUpdate,
TaskContent::Dump { .. } => TaskType::DumpCreation, TaskContent::Dump { .. } => TaskType::DumpCreation,
} }
@ -203,9 +203,9 @@ pub struct TaskView {
impl From<Task> for TaskView { impl From<Task> for TaskView {
fn from(task: Task) -> Self { fn from(task: Task) -> Self {
let index_uid = task.index_uid().map(String::from);
let Task { let Task {
id, id,
index_uid,
content, content,
events, events,
} = task; } = task;
@ -221,20 +221,26 @@ impl From<Task> for TaskView {
(TaskType::DocumentAdditionOrUpdate, Some(details)) (TaskType::DocumentAdditionOrUpdate, Some(details))
} }
TaskContent::DocumentDeletion(DocumentDeletion::Ids(ids)) => ( TaskContent::DocumentDeletion {
deletion: DocumentDeletion::Ids(ids),
..
} => (
TaskType::DocumentDeletion, TaskType::DocumentDeletion,
Some(TaskDetails::DocumentDeletion { Some(TaskDetails::DocumentDeletion {
received_document_ids: ids.len(), received_document_ids: ids.len(),
deleted_documents: None, deleted_documents: None,
}), }),
), ),
TaskContent::DocumentDeletion(DocumentDeletion::Clear) => ( TaskContent::DocumentDeletion {
deletion: DocumentDeletion::Clear,
..
} => (
TaskType::DocumentDeletion, TaskType::DocumentDeletion,
Some(TaskDetails::ClearAll { Some(TaskDetails::ClearAll {
deleted_documents: None, deleted_documents: None,
}), }),
), ),
TaskContent::IndexDeletion => ( TaskContent::IndexDeletion { .. } => (
TaskType::IndexDeletion, TaskType::IndexDeletion,
Some(TaskDetails::ClearAll { Some(TaskDetails::ClearAll {
deleted_documents: None, deleted_documents: None,
@ -244,11 +250,11 @@ impl From<Task> for TaskView {
TaskType::SettingsUpdate, TaskType::SettingsUpdate,
Some(TaskDetails::Settings { settings }), Some(TaskDetails::Settings { settings }),
), ),
TaskContent::IndexCreation { primary_key } => ( TaskContent::IndexCreation { primary_key, .. } => (
TaskType::IndexCreation, TaskType::IndexCreation,
Some(TaskDetails::IndexInfo { primary_key }), Some(TaskDetails::IndexInfo { primary_key }),
), ),
TaskContent::IndexUpdate { primary_key } => ( TaskContent::IndexUpdate { primary_key, .. } => (
TaskType::IndexUpdate, TaskType::IndexUpdate,
Some(TaskDetails::IndexInfo { primary_key }), Some(TaskDetails::IndexInfo { primary_key }),
), ),
@ -353,7 +359,7 @@ impl From<Task> for TaskView {
Self { Self {
uid: id, uid: id,
index_uid: index_uid.map(|u| u.into_inner()), index_uid,
status, status,
task_type, task_type,
details, details,
@ -402,7 +408,7 @@ impl From<Task> for SummarizedTaskView {
Self { Self {
task_uid: other.id, task_uid: other.id,
index_uid: other.index_uid.map(|u| u.into_inner()), index_uid: other.index_uid().map(String::from),
status: TaskStatus::Enqueued, status: TaskStatus::Enqueued,
task_type: other.content.into(), task_type: other.content.into(),
enqueued_at, enqueued_at,

16
meilisearch-lib/src/dump/compat/v3.rs
View File

@ -4,10 +4,10 @@ use serde::{Deserialize, Serialize};
use time::OffsetDateTime; use time::OffsetDateTime;
use uuid::Uuid; use uuid::Uuid;
use super::v4::{Task, TaskEvent}; use super::v4::{Task, TaskContent, TaskEvent};
use crate::index::{Settings, Unchecked}; use crate::index::{Settings, Unchecked};
use crate::index_resolver::IndexUid; use crate::index_resolver::IndexUid;
use crate::tasks::task::{DocumentDeletion, TaskContent, TaskId, TaskResult}; use crate::tasks::task::{DocumentDeletion, TaskId, TaskResult};
use super::v2; use super::v2;
@ -59,9 +59,9 @@ pub enum Update {
ClearDocuments, ClearDocuments,
} }
impl From<Update> for TaskContent { impl From<Update> for super::v4::TaskContent {
fn from(other: Update) -> Self { fn from(update: Update) -> Self {
match other { match update {
Update::DeleteDocuments(ids) => { Update::DeleteDocuments(ids) => {
TaskContent::DocumentDeletion(DocumentDeletion::Ids(ids)) TaskContent::DocumentDeletion(DocumentDeletion::Ids(ids))
} }
@ -186,10 +186,10 @@ impl Failed {
impl From<(UpdateStatus, String, TaskId)> for Task { impl From<(UpdateStatus, String, TaskId)> for Task {
fn from((update, uid, task_id): (UpdateStatus, String, TaskId)) -> Self { fn from((update, uid, task_id): (UpdateStatus, String, TaskId)) -> Self {
// Dummy task // Dummy task
let mut task = Task { let mut task = super::v4::Task {
id: task_id, id: task_id,
index_uid: IndexUid::new(uid).unwrap(), index_uid: IndexUid::new_unchecked(uid),
content: TaskContent::IndexDeletion, content: super::v4::TaskContent::IndexDeletion,
events: Vec::new(), events: Vec::new(),
}; };

84
meilisearch-lib/src/dump/compat/v4.rs
View File

@ -1,9 +1,14 @@
use meilisearch_error::ResponseError; use meilisearch_error::ResponseError;
use milli::update::IndexDocumentsMethod;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use time::OffsetDateTime; use time::OffsetDateTime;
use uuid::Uuid;
use crate::index::{Settings, Unchecked};
use crate::tasks::batch::BatchId; use crate::tasks::batch::BatchId;
use crate::tasks::task::{TaskContent, TaskEvent as NewTaskEvent, TaskId, TaskResult}; use crate::tasks::task::{
DocumentDeletion, TaskContent as NewTaskContent, TaskEvent as NewTaskEvent, TaskId, TaskResult,
};
use crate::IndexUid; use crate::IndexUid;
#[derive(Debug, Serialize, Deserialize)] #[derive(Debug, Serialize, Deserialize)]
@ -18,8 +23,7 @@ impl From<Task> for crate::tasks::task::Task {
fn from(other: Task) -> Self { fn from(other: Task) -> Self {
Self { Self {
id: other.id, id: other.id,
index_uid: Some(other.index_uid), content: NewTaskContent::from((other.index_uid, other.content)),
content: other.content,
events: other.events.into_iter().map(Into::into).collect(), events: other.events.into_iter().map(Into::into).collect(),
} }
} }
@ -65,3 +69,77 @@ impl From<TaskEvent> for NewTaskEvent {
} }
} }
} }
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
#[allow(clippy::large_enum_variant)]
pub enum TaskContent {
DocumentAddition {
content_uuid: Uuid,
merge_strategy: IndexDocumentsMethod,
primary_key: Option<String>,
documents_count: usize,
allow_index_creation: bool,
},
DocumentDeletion(DocumentDeletion),
SettingsUpdate {
settings: Settings<Unchecked>,
/// Indicates whether the task was a deletion
is_deletion: bool,
allow_index_creation: bool,
},
IndexDeletion,
IndexCreation {
primary_key: Option<String>,
},
IndexUpdate {
primary_key: Option<String>,
},
Dump {
uid: String,
},
}
impl From<(IndexUid, TaskContent)> for NewTaskContent {
fn from((index_uid, content): (IndexUid, TaskContent)) -> Self {
match content {
TaskContent::DocumentAddition {
content_uuid,
merge_strategy,
primary_key,
documents_count,
allow_index_creation,
} => NewTaskContent::DocumentAddition {
index_uid,
content_uuid,
merge_strategy,
primary_key,
documents_count,
allow_index_creation,
},
TaskContent::DocumentDeletion(deletion) => NewTaskContent::DocumentDeletion {
index_uid,
deletion,
},
TaskContent::SettingsUpdate {
settings,
is_deletion,
allow_index_creation,
} => NewTaskContent::SettingsUpdate {
index_uid,
settings,
is_deletion,
allow_index_creation,
},
TaskContent::IndexDeletion => NewTaskContent::IndexDeletion { index_uid },
TaskContent::IndexCreation { primary_key } => NewTaskContent::IndexCreation {
index_uid,
primary_key,
},
TaskContent::IndexUpdate { primary_key } => NewTaskContent::IndexUpdate {
index_uid,
primary_key,
},
TaskContent::Dump { uid } => NewTaskContent::Dump { uid },
}
}
}

42
meilisearch-lib/src/index_controller/mod.rs
View File

@ -356,12 +356,16 @@ where
} }
pub async fn register_update(&self, uid: String, update: Update) -> Result<Task> { pub async fn register_update(&self, uid: String, update: Update) -> Result<Task> {
let uid = IndexUid::new(uid)?; let index_uid = IndexUid::new(uid)?;
let content = match update { let content = match update {
Update::DeleteDocuments(ids) => { Update::DeleteDocuments(ids) => TaskContent::DocumentDeletion {
TaskContent::DocumentDeletion(DocumentDeletion::Ids(ids)) index_uid,
} deletion: DocumentDeletion::Ids(ids),
Update::ClearDocuments => TaskContent::DocumentDeletion(DocumentDeletion::Clear), },
Update::ClearDocuments => TaskContent::DocumentDeletion {
index_uid,
deletion: DocumentDeletion::Clear,
},
Update::Settings { Update::Settings {
settings, settings,
is_deletion, is_deletion,
@ -370,6 +374,7 @@ where
settings, settings,
is_deletion, is_deletion,
allow_index_creation, allow_index_creation,
index_uid,
}, },
Update::DocumentAddition { Update::DocumentAddition {
mut payload, mut payload,
@ -409,14 +414,21 @@ where
primary_key, primary_key,
documents_count, documents_count,
allow_index_creation, allow_index_creation,
index_uid,
} }
} }
Update::DeleteIndex => TaskContent::IndexDeletion, Update::DeleteIndex => TaskContent::IndexDeletion { index_uid },
Update::CreateIndex { primary_key } => TaskContent::IndexCreation { primary_key }, Update::CreateIndex { primary_key } => TaskContent::IndexCreation {
Update::UpdateIndex { primary_key } => TaskContent::IndexUpdate { primary_key }, primary_key,
index_uid,
},
Update::UpdateIndex { primary_key } => TaskContent::IndexUpdate {
primary_key,
index_uid,
},
}; };
let task = self.task_store.register(Some(uid), content).await?; let task = self.task_store.register(content).await?;
self.scheduler.read().await.notify(); self.scheduler.read().await.notify();
Ok(task) Ok(task)
@ -425,7 +437,7 @@ where
pub async fn register_dump_task(&self) -> Result<Task> { pub async fn register_dump_task(&self) -> Result<Task> {
let uid = dump::generate_uid(); let uid = dump::generate_uid();
let content = TaskContent::Dump { uid }; let content = TaskContent::Dump { uid };
let task = self.task_store.register(None, content).await?; let task = self.task_store.register(content).await?;
self.scheduler.read().await.notify(); self.scheduler.read().await.notify();
Ok(task) Ok(task)
} }
@ -569,13 +581,7 @@ where
// Check if the currently indexing update is from our index. // Check if the currently indexing update is from our index.
let is_indexing = processing_tasks let is_indexing = processing_tasks
.first() .first()
.map(|task| { .map_or(false, |task| task.index_uid().map_or(false, |u| u == uid));
task.index_uid
.as_ref()
.map(|u| u.as_str() == uid)
.unwrap_or(false)
})
.unwrap_or_default();
let index = self.index_resolver.get_index(uid).await?; let index = self.index_resolver.get_index(uid).await?;
let mut stats = spawn_blocking(move || index.stats()).await??; let mut stats = spawn_blocking(move || index.stats()).await??;
@ -610,7 +616,7 @@ where
// Check if the currently indexing update is from our index. // Check if the currently indexing update is from our index.
stats.is_indexing = processing_tasks stats.is_indexing = processing_tasks
.first() .first()
.and_then(|p| p.index_uid.as_ref().map(|u| u.as_str() == index_uid)) .and_then(|p| p.index_uid().map(|u| u == index_uid))
.or(Some(false)); .or(Some(false));
indexes.insert(index_uid, stats); indexes.insert(index_uid, stats);

414
meilisearch-lib/src/index_resolver/mod.rs
View File

@ -58,7 +58,6 @@ impl IndexUid {
} }
} }
#[cfg(test)]
pub fn new_unchecked(s: impl AsRef<str>) -> Self { pub fn new_unchecked(s: impl AsRef<str>) -> Self {
Self(s.as_ref().to_string()) Self(s.as_ref().to_string())
} }
@ -151,13 +150,13 @@ where
match tasks.first() { match tasks.first() {
Some(Task { Some(Task {
index_uid: Some(ref index_uid),
id, id,
content: content:
TaskContent::DocumentAddition { TaskContent::DocumentAddition {
merge_strategy, merge_strategy,
primary_key, primary_key,
allow_index_creation, allow_index_creation,
index_uid,
.. ..
}, },
.. ..
@ -227,12 +226,14 @@ where
} }
pub async fn process_task(&self, task: &Task) -> Result<TaskResult> { pub async fn process_task(&self, task: &Task) -> Result<TaskResult> {
let index_uid = task.index_uid.clone();
match &task.content { match &task.content {
TaskContent::DocumentAddition { .. } => panic!("updates should be handled by batch"), TaskContent::DocumentAddition { .. } => panic!("updates should be handled by batch"),
TaskContent::DocumentDeletion(DocumentDeletion::Ids(ids)) => { TaskContent::DocumentDeletion {
deletion: DocumentDeletion::Ids(ids),
index_uid,
} => {
let ids = ids.clone(); let ids = ids.clone();
let index = self.get_index(index_uid.unwrap().into_inner()).await?; let index = self.get_index(index_uid.clone().into_inner()).await?;
let DocumentDeletionResult { let DocumentDeletionResult {
deleted_documents, .. deleted_documents, ..
@ -240,8 +241,11 @@ where
Ok(TaskResult::DocumentDeletion { deleted_documents }) Ok(TaskResult::DocumentDeletion { deleted_documents })
} }
TaskContent::DocumentDeletion(DocumentDeletion::Clear) => { TaskContent::DocumentDeletion {
let index = self.get_index(index_uid.unwrap().into_inner()).await?; deletion: DocumentDeletion::Clear,
index_uid,
} => {
let index = self.get_index(index_uid.clone().into_inner()).await?;
let deleted_documents = spawn_blocking(move || -> IndexResult<u64> { let deleted_documents = spawn_blocking(move || -> IndexResult<u64> {
let number_documents = index.stats()?.number_of_documents; let number_documents = index.stats()?.number_of_documents;
index.clear_documents()?; index.clear_documents()?;
@ -255,12 +259,12 @@ where
settings, settings,
is_deletion, is_deletion,
allow_index_creation, allow_index_creation,
index_uid,
} => { } => {
let index = if *is_deletion || !*allow_index_creation { let index = if *is_deletion || !*allow_index_creation {
self.get_index(index_uid.unwrap().into_inner()).await? self.get_index(index_uid.clone().into_inner()).await?
} else { } else {
self.get_or_create_index(index_uid.unwrap(), task.id) self.get_or_create_index(index_uid.clone(), task.id).await?
.await?
}; };
let settings = settings.clone(); let settings = settings.clone();
@ -268,8 +272,8 @@ where
Ok(TaskResult::Other) Ok(TaskResult::Other)
} }
TaskContent::IndexDeletion => { TaskContent::IndexDeletion { index_uid } => {
let index = self.delete_index(index_uid.unwrap().into_inner()).await?; let index = self.delete_index(index_uid.clone().into_inner()).await?;
let deleted_documents = spawn_blocking(move || -> IndexResult<u64> { let deleted_documents = spawn_blocking(move || -> IndexResult<u64> {
Ok(index.stats()?.number_of_documents) Ok(index.stats()?.number_of_documents)
@ -278,8 +282,11 @@ where
Ok(TaskResult::ClearAll { deleted_documents }) Ok(TaskResult::ClearAll { deleted_documents })
} }
TaskContent::IndexCreation { primary_key } => { TaskContent::IndexCreation {
let index = self.create_index(index_uid.unwrap(), task.id).await?; primary_key,
index_uid,
} => {
let index = self.create_index(index_uid.clone(), task.id).await?;
if let Some(primary_key) = primary_key { if let Some(primary_key) = primary_key {
let primary_key = primary_key.clone(); let primary_key = primary_key.clone();
@ -288,8 +295,11 @@ where
Ok(TaskResult::Other) Ok(TaskResult::Other)
} }
TaskContent::IndexUpdate { primary_key } => { TaskContent::IndexUpdate {
let index = self.get_index(index_uid.unwrap().into_inner()).await?; primary_key,
index_uid,
} => {
let index = self.get_index(index_uid.clone().into_inner()).await?;
if let Some(primary_key) = primary_key { if let Some(primary_key) = primary_key {
let primary_key = primary_key.clone(); let primary_key = primary_key.clone();
@ -411,193 +421,193 @@ where
#[cfg(test)] #[cfg(test)]
mod test { mod test {
use std::{collections::BTreeMap, vec::IntoIter}; // use std::{collections::BTreeMap, vec::IntoIter};
//
use super::*; // use super::*;
//
use futures::future::ok; // use futures::future::ok;
use milli::update::{DocumentAdditionResult, IndexDocumentsMethod}; // use milli::update::{DocumentAdditionResult, IndexDocumentsMethod};
use nelson::Mocker; // use nelson::Mocker;
use proptest::prelude::*; // use proptest::prelude::*;
//
use crate::{ // use crate::{
index::{ // index::{
error::{IndexError, Result as IndexResult}, // error::{IndexError, Result as IndexResult},
Checked, IndexMeta, IndexStats, Settings, // Checked, IndexMeta, IndexStats, Settings,
}, // },
tasks::{batch::Batch, BatchHandler}, // tasks::{batch::Batch, BatchHandler},
}; // };
use index_store::MockIndexStore; // use index_store::MockIndexStore;
use meta_store::MockIndexMetaStore; // use meta_store::MockIndexMetaStore;
// TODO: ignoring this test, it has become too complex to maintain, and rather implement // TODO: ignoring this test, it has become too complex to maintain, and rather implement
// handler logic test. // handler logic test.
proptest! { // proptest! {
#[test] // #[test]
#[ignore] // #[ignore]
fn test_process_task( // fn test_process_task(
task in any::<Task>().prop_filter("IndexUid should be Some", |s| s.index_uid.is_some()), // task in any::<Task>().prop_filter("IndexUid should be Some", |s| s.index_uid.is_some()),
index_exists in any::<bool>(), // index_exists in any::<bool>(),
index_op_fails in any::<bool>(), // index_op_fails in any::<bool>(),
any_int in any::<u64>(), // any_int in any::<u64>(),
) { // ) {
actix_rt::System::new().block_on(async move { // actix_rt::System::new().block_on(async move {
let uuid = Uuid::new_v4(); // let uuid = Uuid::new_v4();
let mut index_store = MockIndexStore::new(); // let mut index_store = MockIndexStore::new();
//
let mocker = Mocker::default(); // let mocker = Mocker::default();
//
// Return arbitrary data from index call. // // Return arbitrary data from index call.
match &task.content { // match &task.content {
TaskContent::DocumentAddition{primary_key, ..} => { // TaskContent::DocumentAddition{primary_key, ..} => {
let result = move || if !index_op_fails { // let result = move || if !index_op_fails {
Ok(DocumentAdditionResult { indexed_documents: any_int, number_of_documents: any_int }) // Ok(DocumentAdditionResult { indexed_documents: any_int, number_of_documents: any_int })
} else { // } else {
// return this error because it's easy to generate... // // return this error because it's easy to generate...
Err(IndexError::DocumentNotFound("a doc".into())) // Err(IndexError::DocumentNotFound("a doc".into()))
}; // };
if primary_key.is_some() { // if primary_key.is_some() {
mocker.when::<String, IndexResult<IndexMeta>>("update_primary_key") // mocker.when::<String, IndexResult<IndexMeta>>("update_primary_key")
.then(move |_| Ok(IndexMeta{ created_at: OffsetDateTime::now_utc(), updated_at: OffsetDateTime::now_utc(), primary_key: None })); // .then(move |_| Ok(IndexMeta{ created_at: OffsetDateTime::now_utc(), updated_at: OffsetDateTime::now_utc(), primary_key: None }));
} // }
mocker.when::<(IndexDocumentsMethod, Option<String>, UpdateFileStore, IntoIter<Uuid>), IndexResult<DocumentAdditionResult>>("update_documents") // mocker.when::<(IndexDocumentsMethod, Option<String>, UpdateFileStore, IntoIter<Uuid>), IndexResult<DocumentAdditionResult>>("update_documents")
.then(move |(_, _, _, _)| result()); // .then(move |(_, _, _, _)| result());
} // }
TaskContent::SettingsUpdate{..} => { // TaskContent::SettingsUpdate{..} => {
let result = move || if !index_op_fails { // let result = move || if !index_op_fails {
Ok(()) // Ok(())
} else { // } else {
// return this error because it's easy to generate... // // return this error because it's easy to generate...
Err(IndexError::DocumentNotFound("a doc".into())) // Err(IndexError::DocumentNotFound("a doc".into()))
}; // };
mocker.when::<&Settings<Checked>, IndexResult<()>>("update_settings") // mocker.when::<&Settings<Checked>, IndexResult<()>>("update_settings")
.then(move |_| result()); // .then(move |_| result());
} // }
TaskContent::DocumentDeletion(DocumentDeletion::Ids(_ids)) => { // TaskContent::DocumentDeletion(DocumentDeletion::Ids(_ids)) => {
let result = move || if !index_op_fails { // let result = move || if !index_op_fails {
Ok(DocumentDeletionResult { deleted_documents: any_int as u64, remaining_documents: any_int as u64 }) // Ok(DocumentDeletionResult { deleted_documents: any_int as u64, remaining_documents: any_int as u64 })
} else { // } else {
// return this error because it's easy to generate... // // return this error because it's easy to generate...
Err(IndexError::DocumentNotFound("a doc".into())) // Err(IndexError::DocumentNotFound("a doc".into()))
}; // };
//
mocker.when::<&[String], IndexResult<DocumentDeletionResult>>("delete_documents") // mocker.when::<&[String], IndexResult<DocumentDeletionResult>>("delete_documents")
.then(move |_| result()); // .then(move |_| result());
}, // },
TaskContent::DocumentDeletion(DocumentDeletion::Clear) => { // TaskContent::DocumentDeletion(DocumentDeletion::Clear) => {
let result = move || if !index_op_fails { // let result = move || if !index_op_fails {
Ok(()) // Ok(())
} else { // } else {
// return this error because it's easy to generate... // // return this error because it's easy to generate...
Err(IndexError::DocumentNotFound("a doc".into())) // Err(IndexError::DocumentNotFound("a doc".into()))
}; // };
mocker.when::<(), IndexResult<()>>("clear_documents") // mocker.when::<(), IndexResult<()>>("clear_documents")
.then(move |_| result()); // .then(move |_| result());
}, // },
TaskContent::IndexDeletion => { // TaskContent::IndexDeletion => {
mocker.when::<(), ()>("close") // mocker.when::<(), ()>("close")
.times(index_exists as usize) // .times(index_exists as usize)
.then(move |_| ()); // .then(move |_| ());
} // }
TaskContent::IndexUpdate { primary_key } // TaskContent::IndexUpdate { primary_key }
| TaskContent::IndexCreation { primary_key } => { // | TaskContent::IndexCreation { primary_key } => {
if primary_key.is_some() { // if primary_key.is_some() {
let result = move || if !index_op_fails { // let result = move || if !index_op_fails {
Ok(IndexMeta{ created_at: OffsetDateTime::now_utc(), updated_at: OffsetDateTime::now_utc(), primary_key: None }) // Ok(IndexMeta{ created_at: OffsetDateTime::now_utc(), updated_at: OffsetDateTime::now_utc(), primary_key: None })
} else { // } else {
// return this error because it's easy to generate... // // return this error because it's easy to generate...
Err(IndexError::DocumentNotFound("a doc".into())) // Err(IndexError::DocumentNotFound("a doc".into()))
}; // };
mocker.when::<String, IndexResult<IndexMeta>>("update_primary_key") // mocker.when::<String, IndexResult<IndexMeta>>("update_primary_key")
.then(move |_| result()); // .then(move |_| result());
} // }
} // }
TaskContent::Dump { .. } => { } // TaskContent::Dump { .. } => { }
} // }
//
mocker.when::<(), IndexResult<IndexStats>>("stats") // mocker.when::<(), IndexResult<IndexStats>>("stats")
.then(|()| Ok(IndexStats { size: 0, number_of_documents: 0, is_indexing: Some(false), field_distribution: BTreeMap::new() })); // .then(|()| Ok(IndexStats { size: 0, number_of_documents: 0, is_indexing: Some(false), field_distribution: BTreeMap::new() }));
//
let index = Index::mock(mocker); // let index = Index::mock(mocker);
//
match &task.content { // match &task.content {
// an unexisting index should trigger an index creation in the folllowing cases: // // an unexisting index should trigger an index creation in the folllowing cases:
TaskContent::DocumentAddition { allow_index_creation: true, .. } // TaskContent::DocumentAddition { allow_index_creation: true, .. }
| TaskContent::SettingsUpdate { allow_index_creation: true, is_deletion: false, .. } // | TaskContent::SettingsUpdate { allow_index_creation: true, is_deletion: false, .. }
| TaskContent::IndexCreation { .. } if !index_exists => { // | TaskContent::IndexCreation { .. } if !index_exists => {
index_store // index_store
.expect_create() // .expect_create()
.once() // .once()
.withf(move |&found| !index_exists || found == uuid) // .withf(move |&found| !index_exists || found == uuid)
.returning(move |_| Box::pin(ok(index.clone()))); // .returning(move |_| Box::pin(ok(index.clone())));
}, // },
TaskContent::IndexDeletion => { // TaskContent::IndexDeletion => {
index_store // index_store
.expect_delete() // .expect_delete()
// this is called only if the index.exists // // this is called only if the index.exists
.times(index_exists as usize) // .times(index_exists as usize)
.withf(move |&found| !index_exists || found == uuid) // .withf(move |&found| !index_exists || found == uuid)
.returning(move |_| Box::pin(ok(Some(index.clone())))); // .returning(move |_| Box::pin(ok(Some(index.clone()))));
} // }
// if index already exists, create index will return an error // // if index already exists, create index will return an error
TaskContent::IndexCreation { .. } if index_exists => (), // TaskContent::IndexCreation { .. } if index_exists => (),
TaskContent::Dump { .. } => (), // TaskContent::Dump { .. } => (),
// The index exists and get should be called // // The index exists and get should be called
_ if index_exists => { // _ if index_exists => {
index_store // index_store
.expect_get() // .expect_get()
.once() // .once()
.withf(move |&found| found == uuid) // .withf(move |&found| found == uuid)
.returning(move |_| Box::pin(ok(Some(index.clone())))); // .returning(move |_| Box::pin(ok(Some(index.clone()))));
}, // },
// the index doesn't exist and shouldn't be created, the uuidstore will return an error, and get_index will never be called. // // the index doesn't exist and shouldn't be created, the uuidstore will return an error, and get_index will never be called.
_ => (), // _ => (),
} // }
//
let mut uuid_store = MockIndexMetaStore::new(); // let mut uuid_store = MockIndexMetaStore::new();
uuid_store // uuid_store
.expect_get() // .expect_get()
.returning(move |uid| { // .returning(move |uid| {
Box::pin(ok((uid, index_exists.then(|| crate::index_resolver::meta_store::IndexMeta {uuid, creation_task_id: 0 })))) // Box::pin(ok((uid, index_exists.then(|| crate::index_resolver::meta_store::IndexMeta {uuid, creation_task_id: 0 }))))
}); // });
//
// we sould only be creating an index if the index doesn't alredy exist // // we sould only be creating an index if the index doesn't alredy exist
uuid_store // uuid_store
.expect_insert() // .expect_insert()
.withf(move |_, _| !index_exists) // .withf(move |_, _| !index_exists)
.returning(|_, _| Box::pin(ok(()))); // .returning(|_, _| Box::pin(ok(())));
//
uuid_store // uuid_store
.expect_delete() // .expect_delete()
.times(matches!(task.content, TaskContent::IndexDeletion) as usize) // .times(matches!(task.content, TaskContent::IndexDeletion) as usize)
.returning(move |_| Box::pin(ok(index_exists.then(|| crate::index_resolver::meta_store::IndexMeta { uuid, creation_task_id: 0})))); // .returning(move |_| Box::pin(ok(index_exists.then(|| crate::index_resolver::meta_store::IndexMeta { uuid, creation_task_id: 0}))));
//
let mocker = Mocker::default(); // let mocker = Mocker::default();
let update_file_store = UpdateFileStore::mock(mocker); // let update_file_store = UpdateFileStore::mock(mocker);
let index_resolver = IndexResolver::new(uuid_store, index_store, update_file_store); // let index_resolver = IndexResolver::new(uuid_store, index_store, update_file_store);
//
let batch = Batch { id: Some(1), created_at: OffsetDateTime::now_utc(), content: crate::tasks::batch::BatchContent::IndexUpdate(task.clone()) }; // let batch = Batch { id: Some(1), created_at: OffsetDateTime::now_utc(), content: crate::tasks::batch::BatchContent::IndexUpdate(task.clone()) };
if index_resolver.accept(&batch) { // if index_resolver.accept(&batch) {
let result = index_resolver.process_batch(batch).await; // let result = index_resolver.process_batch(batch).await;
//
// Test for some expected output scenarios: // // Test for some expected output scenarios:
// Index creation and deletion cannot fail because of a failed index op, since they // // Index creation and deletion cannot fail because of a failed index op, since they
// don't perform index ops. // // don't perform index ops.
if index_op_fails && !matches!(task.content, TaskContent::IndexDeletion | TaskContent::IndexCreation { primary_key: None } | TaskContent::IndexUpdate { primary_key: None } | TaskContent::Dump { .. }) // if index_op_fails && !matches!(task.content, TaskContent::IndexDeletion | TaskContent::IndexCreation { primary_key: None } | TaskContent::IndexUpdate { primary_key: None } | TaskContent::Dump { .. })
|| (index_exists && matches!(task.content, TaskContent::IndexCreation { .. })) // || (index_exists && matches!(task.content, TaskContent::IndexCreation { .. }))
|| (!index_exists && matches!(task.content, TaskContent::IndexDeletion // || (!index_exists && matches!(task.content, TaskContent::IndexDeletion
| TaskContent::DocumentDeletion(_) // | TaskContent::DocumentDeletion(_)
| TaskContent::SettingsUpdate { is_deletion: true, ..} // | TaskContent::SettingsUpdate { is_deletion: true, ..}
| TaskContent::SettingsUpdate { allow_index_creation: false, ..} // | TaskContent::SettingsUpdate { allow_index_creation: false, ..}
| TaskContent::DocumentAddition { allow_index_creation: false, ..} // | TaskContent::DocumentAddition { allow_index_creation: false, ..}
| TaskContent::IndexUpdate { .. } )) // | TaskContent::IndexUpdate { .. } ))
{ // {
assert!(matches!(result.content.first().unwrap().events.last().unwrap(), TaskEvent::Failed { .. }), "{:?}", result); // assert!(matches!(result.content.first().unwrap().events.last().unwrap(), TaskEvent::Failed { .. }), "{:?}", result);
} else { // } else {
assert!(matches!(result.content.first().unwrap().events.last().unwrap(), TaskEvent::Succeeded { .. }), "{:?}", result); // assert!(matches!(result.content.first().unwrap().events.last().unwrap(), TaskEvent::Succeeded { .. }), "{:?}", result);
} // }
} // }
}); // });
} // }
} // }
} }

4
meilisearch-lib/src/tasks/handlers/index_resolver_handler.rs
View File

@ -55,6 +55,7 @@ mod test {
task::{Task, TaskContent}, task::{Task, TaskContent},
}; };
use crate::update_file_store::{Result as FileStoreResult, UpdateFileStore}; use crate::update_file_store::{Result as FileStoreResult, UpdateFileStore};
use crate::IndexUid;
use super::*; use super::*;
use milli::update::IndexDocumentsMethod; use milli::update::IndexDocumentsMethod;
@ -103,13 +104,13 @@ mod test {
let task = Task { let task = Task {
id: 1, id: 1,
index_uid: None,
content: TaskContent::DocumentAddition { content: TaskContent::DocumentAddition {
content_uuid, content_uuid,
merge_strategy: IndexDocumentsMethod::ReplaceDocuments, merge_strategy: IndexDocumentsMethod::ReplaceDocuments,
primary_key: None, primary_key: None,
documents_count: 100, documents_count: 100,
allow_index_creation: true, allow_index_creation: true,
index_uid: IndexUid::new_unchecked("test"),
}, },
events: Vec::new(), events: Vec::new(),
}; };
@ -130,7 +131,6 @@ mod test {
let task = Task { let task = Task {
id: 1, id: 1,
index_uid: None,
content: TaskContent::Dump { content: TaskContent::Dump {
uid: String::from("hello"), uid: String::from("hello"),
}, },

4
meilisearch-lib/src/tasks/handlers/mod.rs
View File

@ -17,9 +17,9 @@ mod test {
TaskContent::DocumentAddition { .. } => { TaskContent::DocumentAddition { .. } => {
BatchContent::DocumentsAdditionBatch(vec![task]) BatchContent::DocumentsAdditionBatch(vec![task])
} }
TaskContent::DocumentDeletion(_) TaskContent::DocumentDeletion { .. }
| TaskContent::SettingsUpdate { .. } | TaskContent::SettingsUpdate { .. }
| TaskContent::IndexDeletion | TaskContent::IndexDeletion { .. }
| TaskContent::IndexCreation { .. } | TaskContent::IndexCreation { .. }
| TaskContent::IndexUpdate { .. } => BatchContent::IndexUpdate(task), | TaskContent::IndexUpdate { .. } => BatchContent::IndexUpdate(task),
TaskContent::Dump { .. } => BatchContent::Dump(task), TaskContent::Dump { .. } => BatchContent::Dump(task),

84
meilisearch-lib/src/tasks/scheduler.rs
View File

@ -131,6 +131,22 @@ enum TaskListIdentifier {
Dump, Dump,
} }
impl From<&Task> for TaskListIdentifier {
fn from(task: &Task) -> Self {
match &task.content {
TaskContent::DocumentAddition { index_uid, .. }
| TaskContent::DocumentDeletion { index_uid, .. }
| TaskContent::SettingsUpdate { index_uid, .. }
| TaskContent::IndexDeletion { index_uid }
| TaskContent::IndexCreation { index_uid, .. }
| TaskContent::IndexUpdate { index_uid, .. } => {
TaskListIdentifier::Index(index_uid.as_str().to_string())
}
TaskContent::Dump { .. } => TaskListIdentifier::Dump,
}
}
}
#[derive(Default)] #[derive(Default)]
struct TaskQueue { struct TaskQueue {
/// Maps index uids to their TaskList, for quick access /// Maps index uids to their TaskList, for quick access
@ -142,11 +158,8 @@ struct TaskQueue {
impl TaskQueue { impl TaskQueue {
fn insert(&mut self, task: Task) { fn insert(&mut self, task: Task) {
let id = task.id; let id = task.id;
let uid = match task.index_uid { let uid = TaskListIdentifier::from(&task);
Some(uid) => TaskListIdentifier::Index(uid.into_inner()),
None if matches!(task.content, TaskContent::Dump { .. }) => TaskListIdentifier::Dump,
None => unreachable!("invalid task state"),
};
let kind = match task.content { let kind = match task.content {
TaskContent::DocumentAddition { TaskContent::DocumentAddition {
documents_count, documents_count,
@ -163,9 +176,9 @@ impl TaskQueue {
number: documents_count, number: documents_count,
}, },
TaskContent::Dump { .. } => TaskType::Dump, TaskContent::Dump { .. } => TaskType::Dump,
TaskContent::DocumentDeletion(_) TaskContent::DocumentDeletion { .. }
| TaskContent::SettingsUpdate { .. } | TaskContent::SettingsUpdate { .. }
| TaskContent::IndexDeletion | TaskContent::IndexDeletion { .. }
| TaskContent::IndexCreation { .. } | TaskContent::IndexCreation { .. }
| TaskContent::IndexUpdate { .. } => TaskType::IndexUpdate, | TaskContent::IndexUpdate { .. } => TaskType::IndexUpdate,
_ => unreachable!("unhandled task type"), _ => unreachable!("unhandled task type"),
@ -528,25 +541,25 @@ mod test {
use super::*; use super::*;
fn gen_task(id: TaskId, index_uid: Option<&str>, content: TaskContent) -> Task { fn gen_task(id: TaskId, content: TaskContent) -> Task {
Task { Task {
id, id,
index_uid: index_uid.map(IndexUid::new_unchecked),
content, content,
events: vec![], events: vec![],
} }
} }
#[test] #[test]
#[rustfmt::skip]
fn register_updates_multiples_indexes() { fn register_updates_multiples_indexes() {
let mut queue = TaskQueue::default(); let mut queue = TaskQueue::default();
queue.insert(gen_task(0, Some("test1"), TaskContent::IndexDeletion)); queue.insert(gen_task(0, TaskContent::IndexDeletion { index_uid: IndexUid::new_unchecked("test1") }));
queue.insert(gen_task(1, Some("test2"), TaskContent::IndexDeletion)); queue.insert(gen_task(1, TaskContent::IndexDeletion { index_uid: IndexUid::new_unchecked("test2") }));
queue.insert(gen_task(2, Some("test2"), TaskContent::IndexDeletion)); queue.insert(gen_task(2, TaskContent::IndexDeletion { index_uid: IndexUid::new_unchecked("test2") }));
queue.insert(gen_task(3, Some("test2"), TaskContent::IndexDeletion)); queue.insert(gen_task(3, TaskContent::IndexDeletion { index_uid: IndexUid::new_unchecked("test2") }));
queue.insert(gen_task(4, Some("test1"), TaskContent::IndexDeletion)); queue.insert(gen_task(4, TaskContent::IndexDeletion { index_uid: IndexUid::new_unchecked("test1") }));
queue.insert(gen_task(5, Some("test1"), TaskContent::IndexDeletion)); queue.insert(gen_task(5, TaskContent::IndexDeletion { index_uid: IndexUid::new_unchecked("test1") }));
queue.insert(gen_task(6, Some("test2"), TaskContent::IndexDeletion)); queue.insert(gen_task(6, TaskContent::IndexDeletion { index_uid: IndexUid::new_unchecked("test2") }));
let test1_tasks = queue let test1_tasks = queue
.head_mut(|tasks| tasks.drain().map(|t| t.id).collect::<Vec<_>>()) .head_mut(|tasks| tasks.drain().map(|t| t.id).collect::<Vec<_>>())
@ -564,31 +577,30 @@ mod test {
assert!(queue.queue.is_empty()); assert!(queue.queue.is_empty());
} }
#[test] fn gen_doc_addition_task_content(index_uid: &str) -> TaskContent {
fn test_make_batch() { TaskContent::DocumentAddition {
let mut queue = TaskQueue::default();
let content = TaskContent::DocumentAddition {
content_uuid: Uuid::new_v4(), content_uuid: Uuid::new_v4(),
merge_strategy: IndexDocumentsMethod::ReplaceDocuments, merge_strategy: IndexDocumentsMethod::ReplaceDocuments,
primary_key: Some("test".to_string()), primary_key: Some("test".to_string()),
documents_count: 0, documents_count: 0,
allow_index_creation: true, allow_index_creation: true,
}; index_uid: IndexUid::new_unchecked(index_uid),
queue.insert(gen_task(0, Some("test1"), content.clone())); }
queue.insert(gen_task(1, Some("test2"), content.clone())); }
queue.insert(gen_task(2, Some("test2"), TaskContent::IndexDeletion));
queue.insert(gen_task(3, Some("test2"), content.clone())); #[test]
queue.insert(gen_task(4, Some("test1"), content.clone())); #[rustfmt::skip]
queue.insert(gen_task(5, Some("test1"), TaskContent::IndexDeletion)); fn test_make_batch() {
queue.insert(gen_task(6, Some("test2"), content.clone())); let mut queue = TaskQueue::default();
queue.insert(gen_task(7, Some("test1"), content)); queue.insert(gen_task(0, gen_doc_addition_task_content("test1")));
queue.insert(gen_task( queue.insert(gen_task(1, gen_doc_addition_task_content("test2")));
8, queue.insert(gen_task(2, TaskContent::IndexDeletion { index_uid: IndexUid::new_unchecked("test2")}));
None, queue.insert(gen_task(3, gen_doc_addition_task_content("test2")));
TaskContent::Dump { queue.insert(gen_task(4, gen_doc_addition_task_content("test1")));
uid: "adump".to_owned(), queue.insert(gen_task(5, TaskContent::IndexDeletion { index_uid: IndexUid::new_unchecked("test1")}));
}, queue.insert(gen_task(6, gen_doc_addition_task_content("test2")));
)); queue.insert(gen_task(7, gen_doc_addition_task_content("test1")));
queue.insert(gen_task(8, TaskContent::Dump { uid: "adump".to_owned() }));
let config = SchedulerConfig::default(); let config = SchedulerConfig::default();

38
meilisearch-lib/src/tasks/task.rs
View File

@ -5,10 +5,8 @@ use time::OffsetDateTime;
use uuid::Uuid; use uuid::Uuid;
use super::batch::BatchId; use super::batch::BatchId;
use crate::{ use crate::index::{Settings, Unchecked};
index::{Settings, Unchecked}, use crate::index_resolver::IndexUid;
index_resolver::IndexUid,
};
pub type TaskId = u32; pub type TaskId = u32;
@ -90,13 +88,6 @@ pub struct Task {
/// then this is None /// then this is None
// TODO: when next forward breaking dumps, it would be a good idea to move this field inside of // TODO: when next forward breaking dumps, it would be a good idea to move this field inside of
// the TaskContent. // the TaskContent.
#[cfg_attr(
test,
proptest(
strategy = "proptest::option::weighted(proptest::option::Probability::new(0.99), IndexUid::arbitrary())"
)
)]
pub index_uid: Option<IndexUid>,
pub content: TaskContent, pub content: TaskContent,
pub events: Vec<TaskEvent>, pub events: Vec<TaskEvent>,
} }
@ -123,6 +114,18 @@ impl Task {
_ => None, _ => None,
} }
} }
pub fn index_uid(&self) -> Option<&str> {
match &self.content {
TaskContent::DocumentAddition { index_uid, .. }
| TaskContent::DocumentDeletion { index_uid, .. }
| TaskContent::SettingsUpdate { index_uid, .. }
| TaskContent::IndexDeletion { index_uid }
| TaskContent::IndexCreation { index_uid, .. }
| TaskContent::IndexUpdate { index_uid, .. } => Some(index_uid.as_str()),
TaskContent::Dump { .. } => None,
}
}
} }
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
@ -137,6 +140,7 @@ pub enum DocumentDeletion {
#[allow(clippy::large_enum_variant)] #[allow(clippy::large_enum_variant)]
pub enum TaskContent { pub enum TaskContent {
DocumentAddition { DocumentAddition {
index_uid: IndexUid,
#[cfg_attr(test, proptest(value = "Uuid::new_v4()"))] #[cfg_attr(test, proptest(value = "Uuid::new_v4()"))]
content_uuid: Uuid, content_uuid: Uuid,
#[cfg_attr(test, proptest(strategy = "test::index_document_method_strategy()"))] #[cfg_attr(test, proptest(strategy = "test::index_document_method_strategy()"))]
@ -145,18 +149,26 @@ pub enum TaskContent {
documents_count: usize, documents_count: usize,
allow_index_creation: bool, allow_index_creation: bool,
}, },
DocumentDeletion(DocumentDeletion), DocumentDeletion {
index_uid: IndexUid,
deletion: DocumentDeletion,
},
SettingsUpdate { SettingsUpdate {
index_uid: IndexUid,
settings: Settings<Unchecked>, settings: Settings<Unchecked>,
/// Indicates whether the task was a deletion /// Indicates whether the task was a deletion
is_deletion: bool, is_deletion: bool,
allow_index_creation: bool, allow_index_creation: bool,
}, },
IndexDeletion, IndexDeletion {
index_uid: IndexUid,
},
IndexCreation { IndexCreation {
index_uid: IndexUid,
primary_key: Option<String>, primary_key: Option<String>,
}, },
IndexUpdate { IndexUpdate {
index_uid: IndexUid,
primary_key: Option<String>, primary_key: Option<String>,
}, },
Dump { Dump {

33
meilisearch-lib/src/tasks/task_store/mod.rs
View File

@ -14,7 +14,6 @@ use super::error::TaskError;
use super::scheduler::Processing; use super::scheduler::Processing;
use super::task::{Task, TaskContent, TaskId}; use super::task::{Task, TaskContent, TaskId};
use super::Result; use super::Result;
use crate::index_resolver::IndexUid;
use crate::tasks::task::TaskEvent; use crate::tasks::task::TaskEvent;
use crate::update_file_store::UpdateFileStore; use crate::update_file_store::UpdateFileStore;
@ -32,11 +31,11 @@ pub struct TaskFilter {
impl TaskFilter { impl TaskFilter {
fn pass(&self, task: &Task) -> bool { fn pass(&self, task: &Task) -> bool {
match task.index_uid { match task.index_uid() {
Some(ref index_uid) => self Some(index_uid) => self
.indexes .indexes
.as_ref() .as_ref()
.map_or(true, |indexes| indexes.contains(index_uid.as_str())), .map_or(true, |indexes| indexes.contains(index_uid)),
None => false, None => false,
} }
} }
@ -75,11 +74,7 @@ impl TaskStore {
Ok(Self { store }) Ok(Self { store })
} }
pub async fn register( pub async fn register(&self, content: TaskContent) -> Result<Task> {
&self,
index_uid: Option<IndexUid>,
content: TaskContent,
) -> Result<Task> {
debug!("registering update: {:?}", content); debug!("registering update: {:?}", content);
let store = self.store.clone(); let store = self.store.clone();
let task = tokio::task::spawn_blocking(move || -> Result<Task> { let task = tokio::task::spawn_blocking(move || -> Result<Task> {
@ -88,7 +83,6 @@ impl TaskStore {
let created_at = TaskEvent::Created(OffsetDateTime::now_utc()); let created_at = TaskEvent::Created(OffsetDateTime::now_utc());
let task = Task { let task = Task {
id: next_task_id, id: next_task_id,
index_uid,
content, content,
events: vec![created_at], events: vec![created_at],
}; };
@ -273,7 +267,10 @@ impl TaskStore {
#[cfg(test)] #[cfg(test)]
pub mod test { pub mod test {
use crate::tasks::{scheduler::Processing, task_store::store::test::tmp_env}; use crate::{
tasks::{scheduler::Processing, task_store::store::test::tmp_env},
IndexUid,
};
use super::*; use super::*;
@ -359,13 +356,9 @@ pub mod test {
} }
} }
pub async fn register( pub async fn register(&self, content: TaskContent) -> Result<Task> {
&self,
index_uid: Option<IndexUid>,
content: TaskContent,
) -> Result<Task> {
match self { match self {
Self::Real(s) => s.register(index_uid, content).await, Self::Real(s) => s.register(content).await,
Self::Mock(_m) => todo!(), Self::Mock(_m) => todo!(),
} }
} }
@ -393,8 +386,10 @@ pub mod test {
let gen_task = |id: TaskId| Task { let gen_task = |id: TaskId| Task {
id, id,
index_uid: Some(IndexUid::new_unchecked("test")), content: TaskContent::IndexCreation {
content: TaskContent::IndexCreation { primary_key: None }, primary_key: None,
index_uid: IndexUid::new_unchecked("test"),
},
events: Vec::new(), events: Vec::new(),
}; };

48
meilisearch-lib/src/tasks/task_store/store.rs
View File

@ -77,7 +77,7 @@ impl Store {
pub fn put(&self, txn: &mut RwTxn, task: &Task) -> Result<()> { pub fn put(&self, txn: &mut RwTxn, task: &Task) -> Result<()> {
self.tasks.put(txn, &BEU32::new(task.id), task)?; self.tasks.put(txn, &BEU32::new(task.id), task)?;
// only add the task to the indexes index if it has an index_uid // only add the task to the indexes index if it has an index_uid
if let Some(index_uid) = &task.index_uid { if let Some(index_uid) = task.index_uid() {
let mut tasks_set = self let mut tasks_set = self
.index_uid_task_ids .index_uid_task_ids
.get(txn, index_uid)? .get(txn, index_uid)?
@ -287,8 +287,9 @@ pub mod test {
let tasks = (0..100) let tasks = (0..100)
.map(|_| Task { .map(|_| Task {
id: rand::random(), id: rand::random(),
index_uid: Some(IndexUid::new_unchecked("test")), content: TaskContent::IndexDeletion {
content: TaskContent::IndexDeletion, index_uid: IndexUid::new_unchecked("test"),
},
events: vec![], events: vec![],
}) })
.collect::<Vec<_>>(); .collect::<Vec<_>>();
@ -318,15 +319,17 @@ pub mod test {
let task_1 = Task { let task_1 = Task {
id: 1, id: 1,
index_uid: Some(IndexUid::new_unchecked("test")), content: TaskContent::IndexDeletion {
content: TaskContent::IndexDeletion, index_uid: IndexUid::new_unchecked("test"),
},
events: vec![], events: vec![],
}; };
let task_2 = Task { let task_2 = Task {
id: 0, id: 0,
index_uid: Some(IndexUid::new_unchecked("test1")), content: TaskContent::IndexDeletion {
content: TaskContent::IndexDeletion, index_uid: IndexUid::new_unchecked("test1"),
},
events: vec![], events: vec![],
}; };
@ -341,29 +344,21 @@ pub mod test {
txn.abort().unwrap(); txn.abort().unwrap();
assert_eq!(tasks.len(), 1); assert_eq!(tasks.len(), 1);
assert_eq!( assert_eq!(tasks.first().as_ref().unwrap().index_uid().unwrap(), "test");
tasks
.first()
.as_ref()
.unwrap()
.index_uid
.as_ref()
.unwrap()
.as_str(),
"test"
);
// same thing but invert the ids // same thing but invert the ids
let task_1 = Task { let task_1 = Task {
id: 0, id: 0,
index_uid: Some(IndexUid::new_unchecked("test")), content: TaskContent::IndexDeletion {
content: TaskContent::IndexDeletion, index_uid: IndexUid::new_unchecked("test"),
},
events: vec![], events: vec![],
}; };
let task_2 = Task { let task_2 = Task {
id: 1, id: 1,
index_uid: Some(IndexUid::new_unchecked("test1")), content: TaskContent::IndexDeletion {
content: TaskContent::IndexDeletion, index_uid: IndexUid::new_unchecked("test1"),
},
events: vec![], events: vec![],
}; };
@ -378,14 +373,7 @@ pub mod test {
assert_eq!(tasks.len(), 1); assert_eq!(tasks.len(), 1);
assert_eq!( assert_eq!(
&*tasks &*tasks.first().as_ref().unwrap().index_uid().unwrap(),
.first()
.as_ref()
.unwrap()
.index_uid
.as_ref()
.unwrap()
.as_str(),
"test" "test"
); );
} }