Merge remote-tracking branch 'origin/main' into snapshots

meilisearch-http/src/index/search.rs

@@ -41,6 +41,7 @@ pub struct SearchQuery {
 pub struct SearchResult {
     pub hits: Vec<Map<String, Value>>,
     pub nb_hits: u64,
+    pub exhaustive_nb_hits: bool,
     pub query: String,
     pub limit: usize,
     pub offset: usize,
@@ -107,6 +108,7 @@ impl Index {
         };
 
         let result = SearchResult {
+            exhaustive_nb_hits: false, // not implemented yet
             hits: documents,
             nb_hits,
             query: query.q.clone().unwrap_or_default(),

meilisearch-http/src/index_controller/mod.rs

@@ -17,6 +17,7 @@ use milli::update::{IndexDocumentsMethod, UpdateFormat};
 use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc;
 use tokio::time::sleep;
+use uuid::Uuid;
 
 use crate::index::{Document, SearchQuery, SearchResult};
 use crate::index::{Facets, Settings, UpdateResult};
@@ -29,6 +30,7 @@ use uuid_resolver::UuidResolverHandle;
 
 use snapshot::SnapshotService;
 pub use updates::{Failed, Processed, Processing};
+use uuid_resolver::UuidError;
 
 pub type UpdateStatus = updates::UpdateStatus<UpdateMeta, UpdateResult, String>;
 
@@ -119,41 +121,51 @@ impl IndexController {
         uid: String,
         method: milli::update::IndexDocumentsMethod,
         format: milli::update::UpdateFormat,
-        mut payload: Payload,
+        payload: Payload,
         primary_key: Option<String>,
     ) -> anyhow::Result<UpdateStatus> {
-        let uuid = self.uuid_resolver.get_or_create(uid).await?;
+        let perform_update = |uuid| async move {
-        let meta = UpdateMeta::DocumentsAddition {
+            let meta = UpdateMeta::DocumentsAddition {
-            method,
+                method,
-            format,
+                format,
-            primary_key,
+                primary_key,
+            };
+            let (sender, receiver) = mpsc::channel(10);
+
+            // It is necessary to spawn a local task to send the payload to the update handle to
+            // prevent dead_locking between the update_handle::update that waits for the update to be
+            // registered and the update_actor that waits for the the payload to be sent to it.
+            tokio::task::spawn_local(async move {
+                payload
+                    .map(|bytes| {
+                        bytes.map_err(|e| {
+                            Box::new(e) as Box<dyn std::error::Error + Sync + Send + 'static>
+                        })
+                    })
+                    .for_each(|r| async {
+                        let _ = sender.send(r).await;
+                    })
+                    .await
+            });
+
+            // This must be done *AFTER* spawning the task.
+            self.update_handle.update(meta, receiver, uuid).await
         };
-        let (sender, receiver) = mpsc::channel(10);
 
-        // It is necessary to spawn a local task to senf the payload to the update handle to
+        match self.uuid_resolver.get(uid).await {
-        // prevent dead_locking between the update_handle::update that waits for the update to be
+            Ok(uuid) => Ok(perform_update(uuid).await?),
-        // registered and the update_actor that waits for the the payload to be sent to it.
+            Err(UuidError::UnexistingIndex(name)) => {
-        tokio::task::spawn_local(async move {
+                let uuid = Uuid::new_v4();
-            while let Some(bytes) = payload.next().await {
+                let status = perform_update(uuid).await?;
-                match bytes {
+                self.uuid_resolver.insert(name, uuid).await?;
-                    Ok(bytes) => {
+                Ok(status)
-                        let _ = sender.send(Ok(bytes)).await;
-                    }
-                    Err(e) => {
-                        let error: Box<dyn std::error::Error + Sync + Send + 'static> = Box::new(e);
-                        let _ = sender.send(Err(error)).await;
-                    }
-                }
             }
-        });
+            Err(e) => Err(e.into()),
-
+        }
-        // This must be done *AFTER* spawning the task.
-        let status = self.update_handle.update(meta, receiver, uuid).await?;
-        Ok(status)
     }
 
     pub async fn clear_documents(&self, uid: String) -> anyhow::Result<UpdateStatus> {
-        let uuid = self.uuid_resolver.resolve(uid).await?;
+        let uuid = self.uuid_resolver.get(uid).await?;
         let meta = UpdateMeta::ClearDocuments;
         let (_, receiver) = mpsc::channel(1);
         let status = self.update_handle.update(meta, receiver, uuid).await?;
@@ -165,7 +177,7 @@ impl IndexController {
         uid: String,
         document_ids: Vec<String>,
     ) -> anyhow::Result<UpdateStatus> {
-        let uuid = self.uuid_resolver.resolve(uid).await?;
+        let uuid = self.uuid_resolver.get(uid).await?;
         let meta = UpdateMeta::DeleteDocuments;
         let (sender, receiver) = mpsc::channel(10);
 
@@ -185,26 +197,23 @@ impl IndexController {
         settings: Settings,
         create: bool,
     ) -> anyhow::Result<UpdateStatus> {
-        let uuid = if create {
+        let perform_udpate = |uuid| async move {
-            let uuid = self.uuid_resolver.get_or_create(uid).await?;
+            let meta = UpdateMeta::Settings(settings);
-            // We need to create the index upfront, since it would otherwise only be created when
+            // Nothing so send, drop the sender right away, as not to block the update actor.
-            // the update is processed. This would make calls to GET index to fail until the update
+            let (_, receiver) = mpsc::channel(1);
-            // is complete. Since this is get or create, we ignore the error when the index already
+            self.update_handle.update(meta, receiver, uuid).await
-            // exists.
-            match self.index_handle.create_index(uuid, None).await {
-                Ok(_) | Err(index_actor::IndexError::IndexAlreadyExists) => (),
-                Err(e) => return Err(e.into()),
-            }
-            uuid
-        } else {
-            self.uuid_resolver.resolve(uid).await?
         };
-        let meta = UpdateMeta::Settings(settings);
-        // Nothing so send, drop the sender right away, as not to block the update actor.
-        let (_, receiver) = mpsc::channel(1);
 
-        let status = self.update_handle.update(meta, receiver, uuid).await?;
+        match self.uuid_resolver.get(uid).await {
-        Ok(status)
+            Ok(uuid) => Ok(perform_udpate(uuid).await?),
+            Err(UuidError::UnexistingIndex(name)) if create => {
+                let uuid = Uuid::new_v4();
+                let status = perform_udpate(uuid).await?;
+                self.uuid_resolver.insert(name, uuid).await?;
+                Ok(status)
+            }
+            Err(e) => Err(e.into()),
+        }
     }
 
     pub async fn create_index(
@@ -233,13 +242,13 @@ impl IndexController {
     }
 
     pub async fn update_status(&self, uid: String, id: u64) -> anyhow::Result<UpdateStatus> {
-        let uuid = self.uuid_resolver.resolve(uid).await?;
+        let uuid = self.uuid_resolver.get(uid).await?;
         let result = self.update_handle.update_status(uuid, id).await?;
         Ok(result)
     }
 
     pub async fn all_update_status(&self, uid: String) -> anyhow::Result<Vec<UpdateStatus>> {
-        let uuid = self.uuid_resolver.resolve(uid).await?;
+        let uuid = self.uuid_resolver.get(uid).await?;
         let result = self.update_handle.get_all_updates_status(uuid).await?;
         Ok(result)
     }
@@ -263,7 +272,7 @@ impl IndexController {
     }
 
     pub async fn settings(&self, uid: String) -> anyhow::Result<Settings> {
-        let uuid = self.uuid_resolver.resolve(uid.clone()).await?;
+        let uuid = self.uuid_resolver.get(uid.clone()).await?;
         let settings = self.index_handle.settings(uuid).await?;
         Ok(settings)
     }
@@ -275,7 +284,7 @@ impl IndexController {
         limit: usize,
         attributes_to_retrieve: Option<Vec<String>>,
     ) -> anyhow::Result<Vec<Document>> {
-        let uuid = self.uuid_resolver.resolve(uid.clone()).await?;
+        let uuid = self.uuid_resolver.get(uid.clone()).await?;
         let documents = self
             .index_handle
             .documents(uuid, offset, limit, attributes_to_retrieve)
@@ -289,7 +298,7 @@ impl IndexController {
         doc_id: String,
         attributes_to_retrieve: Option<Vec<String>>,
     ) -> anyhow::Result<Document> {
-        let uuid = self.uuid_resolver.resolve(uid.clone()).await?;
+        let uuid = self.uuid_resolver.get(uid.clone()).await?;
         let document = self
             .index_handle
             .document(uuid, doc_id, attributes_to_retrieve)
@@ -306,7 +315,7 @@ impl IndexController {
             bail!("Can't change the index uid.")
         }
 
-        let uuid = self.uuid_resolver.resolve(uid.clone()).await?;
+        let uuid = self.uuid_resolver.get(uid.clone()).await?;
         let meta = self.index_handle.update_index(uuid, index_settings).await?;
         let meta = IndexMetadata {
             name: uid.clone(),
@@ -317,13 +326,13 @@ impl IndexController {
     }
 
     pub async fn search(&self, uid: String, query: SearchQuery) -> anyhow::Result<SearchResult> {
-        let uuid = self.uuid_resolver.resolve(uid).await?;
+        let uuid = self.uuid_resolver.get(uid).await?;
         let result = self.index_handle.search(uuid, query).await?;
         Ok(result)
     }
 
     pub async fn get_index(&self, uid: String) -> anyhow::Result<IndexMetadata> {
-        let uuid = self.uuid_resolver.resolve(uid.clone()).await?;
+        let uuid = self.uuid_resolver.get(uid.clone()).await?;
         let meta = self.index_handle.get_index_meta(uuid).await?;
         let meta = IndexMetadata {
             name: uid.clone(),