snowykami/meilisearch
1
0
Fork 0
mirror of https://github.com/meilisearch/meilisearch.git synced 2025-11-22 04:36:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

WIP

Browse Source
This commit is contained in:
Louis Dureuil
2025-10-14 09:31:41 +02:00
parent 316b4c047f
commit 415a4de376
27 changed files with 2154 additions and 591 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
crates/dump/src/lib.rs
View File

@@ -9,8 +9,9 @@ use meilisearch_types::error::ResponseError;
use meilisearch_types::keys::Key;
use meilisearch_types::milli::update::IndexDocumentsMethod;
use meilisearch_types::settings::Unchecked;
use meilisearch_types::tasks::enterprise_edition::network::{NetworkTopologyChange, TaskNetwork};
use meilisearch_types::tasks::{
Details, ExportIndexSettings, IndexSwap, KindWithContent, Status, Task, TaskId, TaskNetwork,
Details, ExportIndexSettings, IndexSwap, KindWithContent, Status, Task, TaskId,
};
use meilisearch_types::InstanceUid;
use roaring::RoaringBitmap;
@@ -161,6 +162,7 @@ pub enum KindDump {
IndexCompaction {
index_uid: String,
},
NetworkTopologyChange(NetworkTopologyChange),
}
impl From<Task> for TaskDump {
@@ -246,6 +248,9 @@ impl From<KindWithContent> for KindDump {
KindWithContent::IndexCompaction { index_uid } => {
KindDump::IndexCompaction { index_uid }
}
KindWithContent::NetworkTopologyChange(network_topology_change) => {
KindDump::NetworkTopologyChange(network_topology_change)
}
}
}
}
@@ -554,7 +559,8 @@ pub(crate) mod test {
Network {
local: Some("myself".to_string()),
remotes: maplit::btreemap! {"other".to_string() => Remote { url: "http://test".to_string(), search_api_key: Some("apiKey".to_string()), write_api_key: Some("docApiKey".to_string()) }},
sharding: false,
leader: None,
version: Default::default(),
}
}

3
crates/index-scheduler/src/dump.rs
View File

@@ -237,6 +237,9 @@ impl<'a> Dump<'a> {
KindDump::IndexCompaction { index_uid } => {
KindWithContent::IndexCompaction { index_uid }
}
KindDump::NetworkTopologyChange(network_topology_change) => {
KindWithContent::NetworkTopologyChange(network_topology_change)
}
},
};

5
crates/index-scheduler/src/index_mapper/mod.rs
View File

@@ -361,6 +361,11 @@ impl IndexMapper {
Ok(())
}
/// The number of indexes in the database
pub fn index_count(&self, rtxn: &RoTxn) -> Result<u64> {
Ok(self.index_mapping.len(rtxn)?)
}
/// Return an index, may open it if it wasn't already opened.
pub fn index(&self, rtxn: &RoTxn, name: &str) -> Result<Index> {
if let Some((current_name, current_index)) =

7
crates/index-scheduler/src/insta_snapshot.rs
View File

@@ -21,7 +21,7 @@ pub fn snapshot_index_scheduler(scheduler: &IndexScheduler) -> String {
let IndexScheduler {
cleanup_enabled: _,
experimental_no_edition_2024_for_dumps: _,
processing_tasks,
runtime_tasks,
env,
version,
queue,
@@ -49,7 +49,7 @@ pub fn snapshot_index_scheduler(scheduler: &IndexScheduler) -> String {
snap.push_str(&format!("index scheduler running on version {indx_sched_version:?}\n"));
}
let processing = processing_tasks.read().unwrap().clone();
let processing = runtime_tasks.read().unwrap().processing.clone();
snap.push_str(&format!("### Autobatching Enabled = {}\n", scheduler.autobatching_enabled));
snap.push_str(&format!(
"### Processing batch {:?}:\n",
@@ -320,6 +320,9 @@ fn snapshot_details(d: &Details) -> String {
Details::IndexCompaction { index_uid, pre_compaction_size, post_compaction_size } => {
format!("{{ index_uid: {index_uid:?}, pre_compaction_size: {pre_compaction_size:?}, post_compaction_size: {post_compaction_size:?} }}")
}
Details::NetworkTopologyChange { moved_documents, received_documents, message } => {
format!("{{ moved_documents: {moved_documents:?}, received_documents: {received_documents:?}, message: {message:?}")
}
}
}

44
crates/index-scheduler/src/lib.rs
View File

@@ -68,10 +68,10 @@ use meilisearch_types::milli::vector::{
};
use meilisearch_types::milli::{self, Index};
use meilisearch_types::task_view::TaskView;
use meilisearch_types::tasks::{KindWithContent, Task, TaskNetwork};
use meilisearch_types::tasks::enterprise_edition::network::TaskNetwork;
use meilisearch_types::tasks::{KindWithContent, Task};
use meilisearch_types::webhooks::{Webhook, WebhooksDumpView, WebhooksView};
use milli::vector::db::IndexEmbeddingConfig;
use processing::ProcessingTasks;
pub use queue::Query;
use queue::Queue;
use roaring::RoaringBitmap;
@@ -82,6 +82,7 @@ use uuid::Uuid;
use versioning::Versioning;
use crate::index_mapper::IndexMapper;
use crate::processing::RuntimeTasks;
use crate::utils::clamp_to_page_size;
pub(crate) type BEI128 = I128<BE>;
@@ -163,7 +164,7 @@ pub struct IndexScheduler {
pub(crate) env: Env<WithoutTls>,
/// The list of tasks currently processing
pub(crate) processing_tasks: Arc<RwLock<ProcessingTasks>>,
pub(crate) runtime_tasks: Arc<RwLock<RuntimeTasks>>,
/// A database containing only the version of the index-scheduler
pub version: versioning::Versioning,
@@ -222,7 +223,7 @@ impl IndexScheduler {
fn private_clone(&self) -> IndexScheduler {
IndexScheduler {
env: self.env.clone(),
processing_tasks: self.processing_tasks.clone(),
runtime_tasks: self.runtime_tasks.clone(),
version: self.version.clone(),
queue: self.queue.private_clone(),
scheduler: self.scheduler.private_clone(),
@@ -318,7 +319,7 @@ impl IndexScheduler {
// allow unreachable_code to get rids of the warning in the case of a test build.
let this = Self {
processing_tasks: Arc::new(RwLock::new(ProcessingTasks::new())),
runtime_tasks: Arc::new(RwLock::new(RuntimeTasks::new())),
version,
queue,
scheduler: Scheduler::new(&options, auth_env),
@@ -609,19 +610,19 @@ impl IndexScheduler {
/// 3. The number of times the properties appeared.
pub fn get_stats(&self) -> Result<BTreeMap<String, BTreeMap<String, u64>>> {
let rtxn = self.read_txn()?;
self.queue.get_stats(&rtxn, &self.processing_tasks.read().unwrap())
self.queue.get_stats(&rtxn, &self.runtime_tasks.read().unwrap().processing)
}
// Return true if there is at least one task that is processing.
pub fn is_task_processing(&self) -> Result<bool> {
Ok(!self.processing_tasks.read().unwrap().processing.is_empty())
Ok(!self.runtime_tasks.read().unwrap().processing.processing.is_empty())
}
/// Return true iff there is at least one task associated with this index
/// that is processing.
pub fn is_index_processing(&self, index: &str) -> Result<bool> {
let rtxn = self.env.read_txn()?;
let processing_tasks = self.processing_tasks.read().unwrap().processing.clone();
let processing_tasks = self.runtime_tasks.read().unwrap().processing.processing.clone();
let index_tasks = self.queue.tasks.index_tasks(&rtxn, index)?;
let nbr_index_processing_tasks = processing_tasks.intersection_len(&index_tasks);
Ok(nbr_index_processing_tasks > 0)
@@ -647,8 +648,8 @@ impl IndexScheduler {
filters: &meilisearch_auth::AuthFilter,
) -> Result<(Vec<Task>, u64)> {
let rtxn = self.read_txn()?;
let processing = self.processing_tasks.read().unwrap();
self.queue.get_tasks_from_authorized_indexes(&rtxn, query, filters, &processing)
let processing = &self.runtime_tasks.read().unwrap().processing;
self.queue.get_tasks_from_authorized_indexes(&rtxn, query, filters, processing)
}
/// Return the task ids matching the query along with the total number of tasks
@@ -666,18 +667,18 @@ impl IndexScheduler {
filters: &meilisearch_auth::AuthFilter,
) -> Result<(RoaringBitmap, u64)> {
let rtxn = self.read_txn()?;
let processing = self.processing_tasks.read().unwrap();
self.queue.get_task_ids_from_authorized_indexes(&rtxn, query, filters, &processing)
let processing = &self.runtime_tasks.read().unwrap().processing;
self.queue.get_task_ids_from_authorized_indexes(&rtxn, query, filters, processing)
}
pub fn set_task_network(&self, task_id: TaskId, network: TaskNetwork) -> Result<()> {
pub fn set_task_network(&self, task_id: TaskId, network: TaskNetwork) -> Result<Task> {
let mut wtxn = self.env.write_txn()?;
let mut task =
self.queue.tasks.get_task(&wtxn, task_id)?.ok_or(Error::TaskNotFound(task_id))?;
task.network = Some(network);
self.queue.tasks.all_tasks.put(&mut wtxn, &task_id, &task)?;
wtxn.commit()?;
Ok(())
Ok(task)
}
/// Return the batches matching the query from the user's point of view along
@@ -695,8 +696,8 @@ impl IndexScheduler {
filters: &meilisearch_auth::AuthFilter,
) -> Result<(Vec<Batch>, u64)> {
let rtxn = self.read_txn()?;
let processing = self.processing_tasks.read().unwrap();
self.queue.get_batches_from_authorized_indexes(&rtxn, query, filters, &processing)
let processing = &self.runtime_tasks.read().unwrap().processing;
self.queue.get_batches_from_authorized_indexes(&rtxn, query, filters, processing)
}
/// Return the batch ids matching the query along with the total number of batches
@@ -714,8 +715,8 @@ impl IndexScheduler {
filters: &meilisearch_auth::AuthFilter,
) -> Result<(RoaringBitmap, u64)> {
let rtxn = self.read_txn()?;
let processing = self.processing_tasks.read().unwrap();
self.queue.get_batch_ids_from_authorized_indexes(&rtxn, query, filters, &processing)
let processing = &self.runtime_tasks.read().unwrap().processing;
self.queue.get_batch_ids_from_authorized_indexes(&rtxn, query, filters, processing)
}
/// Register a new task in the scheduler.
@@ -742,7 +743,12 @@ impl IndexScheduler {
// we inform the processing tasks to stop (if necessary).
if let KindWithContent::TaskCancelation { tasks, .. } = kind {
let tasks_to_cancel = RoaringBitmap::from_iter(tasks);
if self.processing_tasks.read().unwrap().must_cancel_processing_tasks(&tasks_to_cancel)
if self
.runtime_tasks
.read()
.unwrap()
.processing
.must_cancel_processing_tasks(&tasks_to_cancel)
{
self.scheduler.must_stop_processing.must_stop();
}

22
crates/index-scheduler/src/processing.rs
View File

@@ -6,6 +6,28 @@ use roaring::RoaringBitmap;
use crate::utils::ProcessingBatch;
pub struct RuntimeTasks {
pub processing: ProcessingTasks,
pub enqueued_network: EnqueuedNetworkTasks,
}
impl RuntimeTasks {
pub(crate) fn new() -> Self {
Self { processing: ProcessingTasks::new(), enqueued_network: Default::default() }
}
}
#[derive(Default)]
pub struct EnqueuedNetworkTasks {
tasks: RoaringBitmap,
}
impl EnqueuedNetworkTasks {
pub fn swap(&mut self, mut new: RoaringBitmap) -> RoaringBitmap {
std::mem::swap(&mut self.tasks, &mut new);
new
}
}
#[derive(Clone, Default)]
pub struct ProcessingTasks {
pub batch: Option<Arc<ProcessingBatch>>,

4
crates/index-scheduler/src/queue/batches_test.rs
View File

@@ -26,7 +26,7 @@ fn query_batches_from_and_limit() {
handle.advance_n_successful_batches(3);
snapshot!(snapshot_index_scheduler(&index_scheduler), name: "processed_all_tasks");
let proc = index_scheduler.processing_tasks.read().unwrap().clone();
let proc = index_scheduler.runtime_tasks.read().unwrap().processing.clone();
let rtxn = index_scheduler.env.read_txn().unwrap();
let query = Query { limit: Some(0), ..Default::default() };
let (batches, _) = index_scheduler
@@ -359,7 +359,7 @@ fn query_batches_special_rules() {
handle.advance_till([Start, BatchCreated]);
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = index_scheduler.processing_tasks.read().unwrap().clone();
let proc = index_scheduler.runtime_tasks.read().unwrap().processing.clone();
let query = Query { index_uids: Some(vec!["catto".to_owned()]), ..Default::default() };
let (batches, _) = index_scheduler

6
crates/index-scheduler/src/queue/tasks_test.rs
View File

@@ -27,7 +27,7 @@ fn query_tasks_from_and_limit() {
snapshot!(snapshot_index_scheduler(&index_scheduler), name: "processed_all_tasks");
let rtxn = index_scheduler.env.read_txn().unwrap();
let processing = index_scheduler.processing_tasks.read().unwrap();
let processing = &index_scheduler.runtime_tasks.read().unwrap().processing;
let query = Query { limit: Some(0), ..Default::default() };
let (tasks, _) = index_scheduler
.queue
@@ -317,7 +317,7 @@ fn query_tasks_special_rules() {
handle.advance_till([Start, BatchCreated]);
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = index_scheduler.processing_tasks.read().unwrap();
let proc = &index_scheduler.runtime_tasks.read().unwrap().processing;
let query = Query { index_uids: Some(vec!["catto".to_owned()]), ..Default::default() };
let (tasks, _) = index_scheduler
@@ -414,7 +414,7 @@ fn query_tasks_canceled_by() {
snapshot!(snapshot_index_scheduler(&index_scheduler), name: "start");
let rtxn = index_scheduler.read_txn().unwrap();
let proc = index_scheduler.processing_tasks.read().unwrap();
let proc = &index_scheduler.runtime_tasks.read().unwrap().processing;
let query = Query { canceled_by: Some(vec![task_cancelation.uid]), ..Query::default() };
let (tasks, _) = index_scheduler
.queue

134
crates/index-scheduler/src/queue/test.rs
View File

@@ -203,26 +203,30 @@ fn test_disable_auto_deletion_of_tasks() {
)
.unwrap();
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = index_scheduler.processing_tasks.read().unwrap();
let tasks =
index_scheduler.queue.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc).unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]" }), name: "task_queue_is_full");
drop(rtxn);
drop(proc);
{
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = &index_scheduler.runtime_tasks.read().unwrap().processing;
let tasks = index_scheduler
.queue
.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc)
.unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]" }), name: "task_queue_is_full");
}
// now we're above the max number of tasks
// and if we try to advance in the tick function no new task deletion should be enqueued
handle.advance_till([Start, BatchCreated]);
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = index_scheduler.processing_tasks.read().unwrap();
let tasks =
index_scheduler.queue.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc).unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]", ".**.original_filter" => "[filter]", ".**.query" => "[query]" }), name: "task_deletion_have_not_been_enqueued");
drop(rtxn);
drop(proc);
{
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = &index_scheduler.runtime_tasks.read().unwrap().processing;
let tasks = index_scheduler
.queue
.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc)
.unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]", ".**.original_filter" => "[filter]", ".**.query" => "[query]" }), name: "task_deletion_have_not_been_enqueued");
}
}
#[test]
@@ -267,59 +271,69 @@ fn test_auto_deletion_of_tasks() {
)
.unwrap();
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = index_scheduler.processing_tasks.read().unwrap();
let tasks =
index_scheduler.queue.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc).unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]" }), name: "task_queue_is_full");
drop(rtxn);
drop(proc);
{
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = &index_scheduler.runtime_tasks.read().unwrap().processing;
let tasks = index_scheduler
.queue
.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc)
.unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]" }), name: "task_queue_is_full");
}
// now we're above the max number of tasks
// and if we try to advance in the tick function a new task deletion should be enqueued
handle.advance_till([Start, BatchCreated]);
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = index_scheduler.processing_tasks.read().unwrap();
let tasks =
index_scheduler.queue.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc).unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]", ".**.original_filter" => "[filter]", ".**.query" => "[query]" }), name: "task_deletion_have_been_enqueued");
drop(rtxn);
drop(proc);
{
// now we're above the max number of tasks
// and if we try to advance in the tick function a new task deletion should be enqueued
handle.advance_till([Start, BatchCreated]);
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = &index_scheduler.runtime_tasks.read().unwrap().processing;
let tasks = index_scheduler
.queue
.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc)
.unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]", ".**.original_filter" => "[filter]", ".**.query" => "[query]" }), name: "task_deletion_have_been_enqueued");
}
handle.advance_till([InsideProcessBatch, ProcessBatchSucceeded, AfterProcessing]);
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = index_scheduler.processing_tasks.read().unwrap();
let tasks =
index_scheduler.queue.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc).unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]", ".**.original_filter" => "[filter]", ".**.query" => "[query]" }), name: "task_deletion_have_been_processed");
drop(rtxn);
drop(proc);
{
handle.advance_till([InsideProcessBatch, ProcessBatchSucceeded, AfterProcessing]);
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = &index_scheduler.runtime_tasks.read().unwrap().processing;
let tasks = index_scheduler
.queue
.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc)
.unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]", ".**.original_filter" => "[filter]", ".**.query" => "[query]" }), name: "task_deletion_have_been_processed");
}
handle.advance_one_failed_batch();
// a new task deletion has been enqueued
handle.advance_one_successful_batch();
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = index_scheduler.processing_tasks.read().unwrap();
let tasks =
index_scheduler.queue.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc).unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]", ".**.original_filter" => "[filter]", ".**.query" => "[query]" }), name: "after_the_second_task_deletion");
drop(rtxn);
drop(proc);
{
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = &index_scheduler.runtime_tasks.read().unwrap().processing;
let tasks = index_scheduler
.queue
.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc)
.unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]", ".**.original_filter" => "[filter]", ".**.query" => "[query]" }), name: "after_the_second_task_deletion");
}
handle.advance_one_failed_batch();
handle.advance_one_successful_batch();
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = index_scheduler.processing_tasks.read().unwrap();
let tasks =
index_scheduler.queue.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc).unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]", ".**.original_filter" => "[filter]", ".**.query" => "[query]" }), name: "everything_has_been_processed");
drop(rtxn);
drop(proc);
{
let rtxn = index_scheduler.env.read_txn().unwrap();
let proc = &index_scheduler.runtime_tasks.read().unwrap().processing;
let tasks = index_scheduler
.queue
.get_task_ids(&rtxn, &Query { ..Default::default() }, &proc)
.unwrap();
let tasks = index_scheduler.queue.tasks.get_existing_tasks(&rtxn, tasks).unwrap();
snapshot!(json_string!(tasks, { "[].enqueuedAt" => "[date]", "[].startedAt" => "[date]", "[].finishedAt" => "[date]", ".**.original_filter" => "[filter]", ".**.query" => "[query]" }), name: "everything_has_been_processed");
}
}
#[test]

1
crates/index-scheduler/src/scheduler/autobatcher.rs
View File

@@ -75,6 +75,7 @@ impl From<KindWithContent> for AutobatchKind {
| KindWithContent::DumpCreation { .. }
| KindWithContent::Export { .. }
| KindWithContent::UpgradeDatabase { .. }
| KindWithContent::NetworkTopologyChange(_)
| KindWithContent::SnapshotCreation => {
panic!("The autobatcher should never be called with tasks that don't apply to an index.")
}

171
crates/index-scheduler/src/scheduler/create_batch.rs
View File

@@ -4,6 +4,7 @@ use std::io::ErrorKind;
use meilisearch_types::heed::RoTxn;
use meilisearch_types::milli::update::IndexDocumentsMethod;
use meilisearch_types::settings::{Settings, Unchecked};
use meilisearch_types::tasks::enterprise_edition::network::NetworkTopologyState;
use meilisearch_types::tasks::{BatchStopReason, Kind, KindWithContent, Status, Task};
use roaring::RoaringBitmap;
use uuid::Uuid;
@@ -59,6 +60,13 @@ pub(crate) enum Batch {
index_uid: String,
task: Task,
},
NetworkIndexBatch {
network_task: Task,
inner_batch: Box<Batch>,
},
NetworkWait {
task: Task,
},
}
#[derive(Debug)]
@@ -140,9 +148,14 @@ impl Batch {
..
} => RoaringBitmap::from_iter(tasks.iter().chain(other).map(|task| task.uid)),
},
Batch::IndexSwap { task } => {
Batch::IndexSwap { task } | Batch::NetworkWait { task } => {
RoaringBitmap::from_sorted_iter(std::iter::once(task.uid)).unwrap()
}
Batch::NetworkIndexBatch { network_task, inner_batch } => {
let mut tasks = inner_batch.ids();
tasks.insert(network_task.uid);
tasks
}
}
}
@@ -156,12 +169,14 @@ impl Batch {
| Dump(_)
| Export { .. }
| UpgradeDatabase { .. }
| NetworkWait { .. }
| IndexSwap { .. } => None,
IndexOperation { op, .. } => Some(op.index_uid()),
IndexCreation { index_uid, .. }
| IndexUpdate { index_uid, .. }
| IndexDeletion { index_uid, .. }
| IndexCompaction { index_uid, .. } => Some(index_uid),
NetworkIndexBatch { network_task: _, inner_batch } => inner_batch.index_uid(),
}
}
}
@@ -184,6 +199,8 @@ impl fmt::Display for Batch {
Batch::IndexCompaction { .. } => f.write_str("IndexCompaction")?,
Batch::Export { .. } => f.write_str("Export")?,
Batch::UpgradeDatabase { .. } => f.write_str("UpgradeDatabase")?,
Batch::NetworkIndexBatch { .. } => f.write_str("NetworkTopologyChange")?,
Batch::NetworkWait { .. } => f.write_str("NetworkTopologyChange")?,
};
match index_uid {
Some(name) => f.write_fmt(format_args!(" on {name:?} from tasks: {tasks:?}")),
@@ -466,7 +483,6 @@ impl IndexScheduler {
let mut current_batch = ProcessingBatch::new(batch_id);
let enqueued = &self.queue.tasks.get_status(rtxn, Status::Enqueued)?;
let count_total_enqueued = enqueued.len();
let failed = &self.queue.tasks.get_status(rtxn, Status::Failed)?;
// 0. we get the last task to cancel.
@@ -525,6 +541,14 @@ impl IndexScheduler {
return Ok(Some((Batch::TaskDeletions(tasks), current_batch)));
}
// 3. Check for enqueued network topology changes
let network_changes =
self.queue.tasks.get_kind(rtxn, Kind::NetworkTopologyChange)? & enqueued;
if let Some(task_id) = network_changes.iter().next() {
let task = self.queue.tasks.get_task(rtxn, task_id)?.unwrap();
return self.start_processing_network(rtxn, task, enqueued, current_batch);
}
// 3. we batch the export.
let to_export = self.queue.tasks.get_kind(rtxn, Kind::Export)? & enqueued;
if !to_export.is_empty() {
@@ -559,7 +583,24 @@ impl IndexScheduler {
}
// 6. We make a batch from the unprioritised tasks. Start by taking the next enqueued task.
let task_id = if let Some(task_id) = enqueued.min() { task_id } else { return Ok(None) };
let (batch, current_batch) =
self.create_next_batch_unprioritized(rtxn, &enqueued, current_batch)?;
Ok(batch.map(|batch| (batch, current_batch)))
}
fn create_next_batch_unprioritized(
&self,
rtxn: &RoTxn,
enqueued: &RoaringBitmap,
mut current_batch: ProcessingBatch,
) -> Result<(Option<Batch>, ProcessingBatch)> {
let count_total_enqueued = enqueued.len();
let task_id = if let Some(task_id) = enqueued.min() {
task_id
} else {
return Ok((None, current_batch));
};
let mut task =
self.queue.tasks.get_task(rtxn, task_id)?.ok_or(Error::CorruptedTaskQueue)?;
@@ -576,7 +617,7 @@ impl IndexScheduler {
kind: Kind::IndexSwap,
id: task.uid,
});
return Ok(Some((Batch::IndexSwap { task }, current_batch)));
return Ok((Some(Batch::IndexSwap { task }), current_batch));
};
let index_already_exists = self.index_mapper.exists(rtxn, index_name)?;
@@ -641,7 +682,7 @@ impl IndexScheduler {
autobatcher::autobatch(enqueued, index_already_exists, primary_key.as_deref())
{
current_batch.reason(autobatch_stop_reason.unwrap_or(stop_reason));
return Ok(self
let batch = self
.create_next_batch_index(
rtxn,
index_name.to_string(),
@@ -649,11 +690,127 @@ impl IndexScheduler {
&mut current_batch,
create_index,
)?
.map(|batch| (batch, current_batch)));
.map(|batch| batch);
return Ok((batch, current_batch));
}
// If we found no tasks then we were notified for something that got autobatched
// somehow and there is nothing to do.
Ok(None)
Ok((None, current_batch))
}
fn start_processing_network(
&self,
rtxn: &RoTxn,
mut task: Task,
enqueued: &RoaringBitmap,
current_batch: ProcessingBatch,
) -> Result<Option<(Batch, ProcessingBatch)>> {
let change_version =
task.network.as_ref().map(|network| network.network_version()).unwrap_or_default();
let KindWithContent::NetworkTopologyChange(network_topology_change) = &task.kind else {
panic!("inconsistent kind with content")
};
match network_topology_change.state() {
NetworkTopologyState::WaitingForOlderTasks => {
let mut old_tasks = RoaringBitmap::new();
for task_id in enqueued {
let task = self
.queue
.tasks
.get_task(rtxn, task_id)?
.ok_or(Error::CorruptedTaskQueue)?;
let has_index = task.index_uid().is_some();
if !has_index {
continue;
}
let has_older_network_version = task
.network
.map(|network| network.network_version() <= change_version)
// if there is no version, we never retain the task
.unwrap_or_default();
if has_older_network_version {
old_tasks.push(task_id);
}
}
let res = self.create_next_batch_unprioritized(rtxn, &old_tasks, current_batch);
self.runtime_tasks.write().unwrap().enqueued_network.swap(old_tasks);
let (batch, mut current_batch) = res?;
current_batch.processing(Some(&mut task));
let batch = match batch {
Some(batch) => {
let inner_batch = Box::new(batch);
Batch::NetworkIndexBatch { network_task: task, inner_batch }
}
None => Batch::NetworkWait { task },
};
Ok(Some((batch, current_batch)))
}
NetworkTopologyState::ImportingDocuments => {
let mut import_tasks = RoaringBitmap::new();
for task_id in enqueued {
let task = self
.queue
.tasks
.get_task(rtxn, task_id)?
.ok_or(Error::CorruptedTaskQueue)?;
let has_index = task.index_uid().is_some();
if !has_index {
continue;
}
let is_import_task = task
.network
.map(|network| {
network.network_version() == change_version
&& network.import_data().is_some()
})
// if there is no version, we never retain the task
.unwrap_or_default();
if is_import_task {
import_tasks.push(task_id);
}
}
let res = self.create_next_batch_unprioritized(rtxn, &import_tasks, current_batch);
self.runtime_tasks.write().unwrap().enqueued_network.swap(import_tasks);
let (batch, mut current_batch) = res?;
current_batch.processing(Some(&mut task));
let batch = match batch {
Some(batch) => {
let inner_batch = Box::new(batch);
Batch::NetworkIndexBatch { network_task: task, inner_batch }
}
None => Batch::NetworkWait { task },
};
Ok(Some((batch, current_batch)))
}
NetworkTopologyState::ExportingDocuments | NetworkTopologyState::Finished => {
Ok(Some((Batch::NetworkWait { task }, current_batch)))
}
}
}
}
pub enum BatchOutcome {
NoTaskToProcess,
Batch { batch: Batch, processing: ProcessingBatch },
}

5
crates/index-scheduler/src/scheduler/mod.rs
View File

@@ -167,9 +167,10 @@ impl IndexScheduler {
// We reset the must_stop flag to be sure that we don't stop processing tasks
self.scheduler.must_stop_processing.reset();
let progress = self
.processing_tasks
.runtime_tasks
.write()
.unwrap()
.processing
// We can clone the processing batch here because we don't want its modification to affect the view of the processing batches
.start_processing(processing_batch.clone(), ids.clone());
@@ -420,7 +421,7 @@ impl IndexScheduler {
// We should stop processing AFTER everything is processed and written to disk otherwise, a batch (which only lives in RAM) may appear in the processing task
// and then become « not found » for some time until the commit everything is written and the final commit is made.
self.processing_tasks.write().unwrap().stop_processing();
self.runtime_tasks.write().unwrap().processing.stop_processing();
// Once the tasks are committed, we should delete all the update files associated ASAP to avoid leaking files in case of a restart
tracing::debug!("Deleting the update files");

215
crates/index-scheduler/src/scheduler/process_batch.rs
View File

@@ -1,14 +1,21 @@
use std::collections::{BTreeSet, HashMap, HashSet};
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::io::{Seek, SeekFrom};
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::atomic::Ordering;
use std::time::Duration;
use bumpalo::Bump;
use byte_unit::Byte;
use meilisearch_types::batches::{BatchEnqueuedAt, BatchId};
use meilisearch_types::enterprise_edition::network::Remote;
use meilisearch_types::heed::{RoTxn, RwTxn};
use meilisearch_types::milli::documents::PrimaryKey;
use meilisearch_types::milli::heed::CompactionOption;
use meilisearch_types::milli::progress::{Progress, VariableNameStep};
use meilisearch_types::milli::progress::{EmbedderStats, Progress, VariableNameStep};
use meilisearch_types::milli::update::new::indexer;
use meilisearch_types::milli::update::new::indexer::enterprise_edition::sharding::Shards;
use meilisearch_types::milli::{self, ChannelCongestion};
use meilisearch_types::tasks::enterprise_edition::network::{NetworkTopologyState, Origin};
use meilisearch_types::tasks::{Details, IndexSwap, Kind, KindWithContent, Status, Task};
use meilisearch_types::versioning::{VERSION_MAJOR, VERSION_MINOR, VERSION_PATCH};
use milli::update::Settings as MilliSettings;
@@ -22,6 +29,7 @@ use crate::processing::{
IndexCompaction, InnerSwappingTwoIndexes, SwappingTheIndexes, TaskCancelationProgress,
TaskDeletionProgress, UpdateIndexProgress,
};
use crate::scheduler::process_export::{ExportContext, ExportOptions, TargetInstance};
use crate::utils::{
self, remove_n_tasks_datetime_earlier_than, remove_task_datetime, swap_index_uid_in_task,
ProcessingBatch,
@@ -535,9 +543,210 @@ impl IndexScheduler {
Ok((tasks, ProcessBatchInfo::default()))
}
Batch::NetworkIndexBatch { mut network_task, inner_batch } => {
let (mut tasks, info) =
self.process_batch(*inner_batch, current_batch, progress)?;
let KindWithContent::NetworkTopologyChange(network_topology_change) =
&mut network_task.kind
else {
return Err(Error::CorruptedTaskQueue);
};
for task in &tasks {
let Some(network) = task.network.as_ref() else {
continue;
};
let Some(import) = network.import_data() else {
continue;
};
network_topology_change.process_remote_tasks(
&import.remote_name,
&import.index_name,
import.document_count,
);
}
tasks.push(network_task);
Ok((tasks, info))
}
Batch::NetworkWait { mut task } => {
let KindWithContent::NetworkTopologyChange(network_topology_change) =
&mut task.kind
else {
tracing::error!("network topology change task has the wrong kind with content");
return Err(Error::CorruptedTaskQueue);
};
let Some(task_network) = &task.network else {
tracing::error!("network topology change task has no network");
return Err(Error::CorruptedTaskQueue);
};
let origin;
let origin = match task_network.origin() {
Some(origin) => origin,
None => {
let myself =
network_topology_change.in_name().expect("origin is not the leader");
origin = Origin {
remote_name: myself.to_string(),
task_uid: task.uid,
network_version: task_network.network_version(),
};
&origin
}
};
if let Some((remotes, out_name)) = network_topology_change.export_to_process() {
self.balance_documents(
remotes,
out_name,
network_topology_change.in_name(),
origin,
&progress,
&self.scheduler.must_stop_processing,
)?;
}
network_topology_change.update_state();
if network_topology_change.state() == NetworkTopologyState::Finished {
task.status = Status::Succeeded;
}
Ok((vec![task], Default::default()))
}
}
}
fn balance_documents(
&self,
remotes: &BTreeMap<String, Remote>,
out_name: &str,
in_name: Option<&str>,
network_change_origin: &Origin,
progress: &Progress,
must_stop_processing: &crate::scheduler::MustStopProcessing,
) -> crate::Result<()> {
let new_shards = Shards::from_remotes_local(remotes.keys().map(String::as_str), in_name);
// TECHDEBT: this spawns a `ureq` agent additionally to `reqwest`. We probably want to harmonize all of this.
let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build();
let mut indexer_alloc = Bump::new();
let scheduler_rtxn = self.env.read_txn()?;
let index_count = self.index_mapper.index_count(&scheduler_rtxn)?;
// process by batches of 20MiB. Allow for compression? Don't forget about embeddings
let _: Vec<()> = self.index_mapper.try_for_each_index(
&scheduler_rtxn,
|index_uid, index| -> crate::Result<()> {
indexer_alloc.reset();
let err = |err| Error::from_milli(err, Some(index_uid.to_string()));
let index_rtxn = index.read_txn()?;
let all_docids = index.external_documents_ids();
let mut documents_to_move_to: HashMap<String, RoaringBitmap> = HashMap::new();
let mut documents_to_delete = RoaringBitmap::new();
for res in all_docids.iter(&index_rtxn)? {
let (external_docid, docid) = res?;
match new_shards.processing_shard(external_docid) {
Some(shard) if shard.is_own => continue,
Some(shard) => {
documents_to_move_to
.entry(shard.name.clone())
.or_default()
.insert(docid);
}
None => {
documents_to_delete.insert(docid);
}
}
}
let fields_ids_map = index.fields_ids_map(&index_rtxn)?;
for (remote, documents_to_move) in documents_to_move_to {
/// TODO: justify the unwrap
let remote = remotes.get(&remote).unwrap();
let target = TargetInstance {
base_url: &remote.url,
api_key: remote.write_api_key.as_deref(),
};
let options = ExportOptions {
index_uid,
payload_size: None,
override_settings: false,
/// TODO: index count and max docs nb
export_mode: super::process_export::ExportMode::NetworkBalancing {
index_count,
export_old_remote_name: out_name,
network_change_origin,
},
};
let ctx = ExportContext {
index,
index_rtxn: &index_rtxn,
universe: &documents_to_move,
progress,
agent: &agent,
must_stop_processing,
};
self.export_one_index(target, options, ctx)?;
documents_to_delete |= documents_to_move;
}
if documents_to_delete.is_empty() {
return Ok(());
}
let mut new_fields_ids_map = fields_ids_map.clone();
// candidates not empty => index not empty => a primary key is set
let primary_key = index.primary_key(&index_rtxn)?.unwrap();
let primary_key = PrimaryKey::new_or_insert(primary_key, &mut new_fields_ids_map)
.map_err(milli::Error::from)
.map_err(err)?;
let mut index_wtxn = index.write_txn()?;
let mut indexer = indexer::DocumentDeletion::new();
indexer.delete_documents_by_docids(documents_to_delete);
let document_changes = indexer.into_changes(&indexer_alloc, primary_key);
let embedders = index
.embedding_configs()
.embedding_configs(&index_wtxn)
.map_err(milli::Error::from)
.map_err(err)?;
let embedders = self.embedders(index_uid.to_string(), embedders)?;
let indexer_config = self.index_mapper.indexer_config();
let pool = &indexer_config.thread_pool;
indexer::index(
&mut index_wtxn,
index,
pool,
indexer_config.grenad_parameters(),
&fields_ids_map,
new_fields_ids_map,
None, // document deletion never changes primary key
&document_changes,
embedders,
&|| must_stop_processing.get(),
&progress,
&EmbedderStats::default(),
)
.map_err(err)?;
index_wtxn.commit()?;
Ok(())
},
)?;
Ok(())
}
fn apply_compaction(
&self,
rtxn: &RoTxn,
@@ -706,7 +915,7 @@ impl IndexScheduler {
// 1. Remove from this list the tasks that we are not allowed to delete
let enqueued_tasks = self.queue.tasks.get_status(wtxn, Status::Enqueued)?;
let processing_tasks = &self.processing_tasks.read().unwrap().processing.clone();
let processing_tasks = &self.runtime_tasks.read().unwrap().processing.processing.clone();
let all_task_ids = self.queue.tasks.all_task_ids(wtxn)?;
let mut to_delete_tasks = all_task_ids & matched_tasks;

555
crates/index-scheduler/src/scheduler/process_export.rs
View File

@@ -15,7 +15,9 @@ use meilisearch_types::milli::update::{request_threads, Setting};
use meilisearch_types::milli::vector::parsed_vectors::{ExplicitVectors, VectorOrArrayOfVectors};
use meilisearch_types::milli::{self, obkv_to_json, Filter, InternalError};
use meilisearch_types::settings::{self, SecretPolicy};
use meilisearch_types::tasks::enterprise_edition::network::{headers, ImportData, Origin};
use meilisearch_types::tasks::{DetailsExportIndexSettings, ExportIndexSettings};
use roaring::RoaringBitmap;
use serde::Deserialize;
use ureq::{json, Response};
@@ -50,6 +52,7 @@ impl IndexScheduler {
let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build();
let must_stop_processing = self.scheduler.must_stop_processing.clone();
for (i, (_pattern, uid, export_settings)) in indexes.iter().enumerate() {
let err = |err| Error::from_milli(err, Some(uid.to_string()));
if must_stop_processing.get() {
return Err(Error::AbortedTask);
}
@@ -61,104 +64,31 @@ impl IndexScheduler {
));
let ExportIndexSettings { filter, override_settings } = export_settings;
let index = self.index(uid)?;
let index_rtxn = index.read_txn()?;
let bearer = api_key.map(|api_key| format!("Bearer {api_key}"));
// First, check if the index already exists
let url = format!("{base_url}/indexes/{uid}");
let response = retry(&must_stop_processing, || {
let mut request = agent.get(&url);
if let Some(bearer) = &bearer {
request = request.set("Authorization", bearer);
}
request.send_bytes(Default::default()).map_err(into_backoff_error)
});
let index_exists = match response {
Ok(response) => response.status() == 200,
Err(Error::FromRemoteWhenExporting { code, .. }) if code == "index_not_found" => {
false
}
Err(e) => return Err(e),
};
let primary_key = index
.primary_key(&index_rtxn)
.map_err(|e| Error::from_milli(e.into(), Some(uid.to_string())))?;
// Create the index
if !index_exists {
let url = format!("{base_url}/indexes");
retry(&must_stop_processing, || {
let mut request = agent.post(&url);
if let Some(bearer) = &bearer {
request = request.set("Authorization", bearer);
}
let index_param = json!({ "uid": uid, "primaryKey": primary_key });
request.send_json(&index_param).map_err(into_backoff_error)
})?;
}
// Patch the index primary key
if index_exists && *override_settings {
let url = format!("{base_url}/indexes/{uid}");
retry(&must_stop_processing, || {
let mut request = agent.patch(&url);
if let Some(bearer) = &bearer {
request = request.set("Authorization", bearer);
}
let index_param = json!({ "primaryKey": primary_key });
request.send_json(&index_param).map_err(into_backoff_error)
})?;
}
// Send the index settings
if !index_exists || *override_settings {
let mut settings =
settings::settings(&index, &index_rtxn, SecretPolicy::RevealSecrets)
.map_err(|e| Error::from_milli(e, Some(uid.to_string())))?;
// Remove the experimental chat setting if not enabled
if self.features().check_chat_completions("exporting chat settings").is_err() {
settings.chat = Setting::NotSet;
}
// Retry logic for sending settings
let url = format!("{base_url}/indexes/{uid}/settings");
retry(&must_stop_processing, || {
let mut request = agent.patch(&url);
if let Some(bearer) = bearer.as_ref() {
request = request.set("Authorization", bearer);
}
request.send_json(settings.clone()).map_err(into_backoff_error)
})?;
}
let filter = filter
.as_ref()
.map(Filter::from_json)
.transpose()
.map_err(|e| Error::from_milli(e, Some(uid.to_string())))?
.flatten();
let filter_universe = filter
.map(|f| f.evaluate(&index_rtxn, &index))
.transpose()
.map_err(|e| Error::from_milli(e, Some(uid.to_string())))?;
let whole_universe = index
.documents_ids(&index_rtxn)
.map_err(|e| Error::from_milli(e.into(), Some(uid.to_string())))?;
let filter = filter.as_ref().map(Filter::from_json).transpose().map_err(err)?.flatten();
let filter_universe =
filter.map(|f| f.evaluate(&index_rtxn, &index)).transpose().map_err(err)?;
let whole_universe =
index.documents_ids(&index_rtxn).map_err(milli::Error::from).map_err(err)?;
let universe = filter_universe.unwrap_or(whole_universe);
let fields_ids_map = index.fields_ids_map(&index_rtxn)?;
let all_fields: Vec<_> = fields_ids_map.iter().map(|(id, _)| id).collect();
// We don't need to keep this one alive as we will
// spawn many threads to process the documents
drop(index_rtxn);
let total_documents = universe.len() as u32;
let (step, progress_step) = AtomicDocumentStep::new(total_documents);
progress.update_progress(progress_step);
let target = TargetInstance { base_url, api_key };
let ctx = ExportContext {
index: &index,
index_rtxn: &index_rtxn,
universe: &universe,
progress: &progress,
agent: &agent,
must_stop_processing: &must_stop_processing,
};
let options = ExportOptions {
index_uid: uid,
payload_size,
override_settings: *override_settings,
export_mode: ExportMode::ExportRoute,
};
let total_documents = self.export_one_index(target, options, ctx)?;
output.insert(
IndexUidPattern::new_unchecked(uid.clone()),
@@ -167,155 +97,268 @@ impl IndexScheduler {
matched_documents: Some(total_documents as u64),
},
);
let limit = payload_size.map(|ps| ps.as_u64() as usize).unwrap_or(20 * 1024 * 1024); // defaults to 20 MiB
let documents_url = format!("{base_url}/indexes/{uid}/documents");
let results = request_threads()
.broadcast(|ctx| {
let index_rtxn = index
.read_txn()
.map_err(|e| Error::from_milli(e.into(), Some(uid.to_string())))?;
let mut buffer = Vec::new();
let mut tmp_buffer = Vec::new();
let mut compressed_buffer = Vec::new();
for (i, docid) in universe.iter().enumerate() {
if i % ctx.num_threads() != ctx.index() {
continue;
}
let document = index
.document(&index_rtxn, docid)
.map_err(|e| Error::from_milli(e, Some(uid.to_string())))?;
let mut document = obkv_to_json(&all_fields, &fields_ids_map, document)
.map_err(|e| Error::from_milli(e, Some(uid.to_string())))?;
// TODO definitely factorize this code
'inject_vectors: {
let embeddings = index
.embeddings(&index_rtxn, docid)
.map_err(|e| Error::from_milli(e, Some(uid.to_string())))?;
if embeddings.is_empty() {
break 'inject_vectors;
}
let vectors = document
.entry(RESERVED_VECTORS_FIELD_NAME)
.or_insert(serde_json::Value::Object(Default::default()));
let serde_json::Value::Object(vectors) = vectors else {
return Err(Error::from_milli(
milli::Error::UserError(
milli::UserError::InvalidVectorsMapType {
document_id: {
if let Ok(Some(Ok(index))) = index
.external_id_of(
&index_rtxn,
std::iter::once(docid),
)
.map(|it| it.into_iter().next())
{
index
} else {
format!("internal docid={docid}")
}
},
value: vectors.clone(),
},
),
Some(uid.to_string()),
));
};
for (
embedder_name,
EmbeddingsWithMetadata { embeddings, regenerate, has_fragments },
) in embeddings
{
let embeddings = ExplicitVectors {
embeddings: Some(
VectorOrArrayOfVectors::from_array_of_vectors(embeddings),
),
regenerate: regenerate &&
// Meilisearch does not handle well dumps with fragments, because as the fragments
// are marked as user-provided,
// all embeddings would be regenerated on any settings change or document update.
// To prevent this, we mark embeddings has non regenerate in this case.
!has_fragments,
};
vectors.insert(
embedder_name,
serde_json::to_value(embeddings).unwrap(),
);
}
}
tmp_buffer.clear();
serde_json::to_writer(&mut tmp_buffer, &document)
.map_err(milli::InternalError::from)
.map_err(|e| Error::from_milli(e.into(), Some(uid.to_string())))?;
// Make sure we put at least one document in the buffer even
// though we might go above the buffer limit before sending
if !buffer.is_empty() && buffer.len() + tmp_buffer.len() > limit {
// We compress the documents before sending them
let mut encoder =
GzEncoder::new(&mut compressed_buffer, Compression::default());
encoder
.write_all(&buffer)
.map_err(|e| Error::from_milli(e.into(), Some(uid.clone())))?;
encoder
.finish()
.map_err(|e| Error::from_milli(e.into(), Some(uid.clone())))?;
retry(&must_stop_processing, || {
let mut request = agent.post(&documents_url);
request = request.set("Content-Type", "application/x-ndjson");
request = request.set("Content-Encoding", "gzip");
if let Some(bearer) = &bearer {
request = request.set("Authorization", bearer);
}
request.send_bytes(&compressed_buffer).map_err(into_backoff_error)
})?;
buffer.clear();
compressed_buffer.clear();
}
buffer.extend_from_slice(&tmp_buffer);
if i > 0 && i % 100 == 0 {
step.fetch_add(100, atomic::Ordering::Relaxed);
}
}
retry(&must_stop_processing, || {
let mut request = agent.post(&documents_url);
request = request.set("Content-Type", "application/x-ndjson");
if let Some(bearer) = &bearer {
request = request.set("Authorization", bearer);
}
request.send_bytes(&buffer).map_err(into_backoff_error)
})?;
Ok(())
})
.map_err(|e| {
Error::from_milli(
milli::Error::InternalError(InternalError::PanicInThreadPool(e)),
Some(uid.to_string()),
)
})?;
for result in results {
result?;
}
step.store(total_documents, atomic::Ordering::Relaxed);
}
Ok(output)
}
pub(super) fn export_one_index(
&self,
target: TargetInstance<'_>,
options: ExportOptions<'_>,
ctx: ExportContext<'_>,
) -> Result<u64, Error> {
let err = |err| Error::from_milli(err, Some(options.index_uid.to_string()));
let bearer = target.api_key.map(|api_key| format!("Bearer {api_key}"));
let url = format!(
"{base_url}/indexes/{index_uid}",
base_url = target.base_url,
index_uid = options.index_uid
);
let response = retry(ctx.must_stop_processing, || {
let mut request = ctx.agent.get(&url);
if let Some(bearer) = &bearer {
request = request.set("Authorization", bearer);
}
request.send_bytes(Default::default()).map_err(into_backoff_error)
});
let index_exists = match response {
Ok(response) => response.status() == 200,
Err(Error::FromRemoteWhenExporting { code, .. }) if code == "index_not_found" => false,
Err(e) => return Err(e),
};
let primary_key =
ctx.index.primary_key(&ctx.index_rtxn).map_err(milli::Error::from).map_err(err)?;
if !index_exists {
let url = format!("{base_url}/indexes", base_url = target.base_url);
retry(ctx.must_stop_processing, || {
let mut request = ctx.agent.post(&url);
if let Some(bearer) = &bearer {
request = request.set("Authorization", bearer);
}
let index_param = json!({ "uid": options.index_uid, "primaryKey": primary_key });
request.send_json(&index_param).map_err(into_backoff_error)
})?;
}
if index_exists && options.override_settings {
retry(ctx.must_stop_processing, || {
let mut request = ctx.agent.patch(&url);
if let Some(bearer) = &bearer {
request = request.set("Authorization", bearer);
}
let index_param = json!({ "primaryKey": primary_key });
request.send_json(&index_param).map_err(into_backoff_error)
})?;
}
if !index_exists || options.override_settings {
let mut settings =
settings::settings(&ctx.index, &ctx.index_rtxn, SecretPolicy::RevealSecrets)
.map_err(err)?;
// Remove the experimental chat setting if not enabled
if self.features().check_chat_completions("exporting chat settings").is_err() {
settings.chat = Setting::NotSet;
}
// Retry logic for sending settings
let url = format!(
"{base_url}/indexes/{index_uid}/settings",
base_url = target.base_url,
index_uid = options.index_uid
);
retry(ctx.must_stop_processing, || {
let mut request = ctx.agent.patch(&url);
if let Some(bearer) = bearer.as_ref() {
request = request.set("Authorization", bearer);
}
request.send_json(settings.clone()).map_err(into_backoff_error)
})?;
}
let fields_ids_map = ctx.index.fields_ids_map(&ctx.index_rtxn)?;
let all_fields: Vec<_> = fields_ids_map.iter().map(|(id, _)| id).collect();
let total_documents = ctx.universe.len() as u32;
let (step, progress_step) = AtomicDocumentStep::new(total_documents);
ctx.progress.update_progress(progress_step);
let limit = options.payload_size.map(|ps| ps.as_u64() as usize).unwrap_or(20 * 1024 * 1024);
let documents_url = format!(
"{base_url}/indexes/{index_uid}/documents",
base_url = target.base_url,
index_uid = options.index_uid
);
let results = request_threads()
.broadcast(|broadcast| {
let mut task_network = if let ExportMode::NetworkBalancing {
index_count,
export_old_remote_name,
network_change_origin,
} = options.export_mode
{
Some((
ImportData {
remote_name: export_old_remote_name.to_string(),
index_name: options.index_uid.to_string(),
first_docid: 0,
document_count: 0,
},
network_change_origin.clone(),
))
} else {
None
};
let index_rtxn = ctx.index.read_txn().map_err(milli::Error::from).map_err(err)?;
let mut buffer = Vec::new();
let mut tmp_buffer = Vec::new();
let mut compressed_buffer = Vec::new();
for (i, docid) in ctx.universe.iter().enumerate() {
if i % broadcast.num_threads() != broadcast.index() {
continue;
}
if let Some((import_data, _)) = &mut task_network {
import_data.document_count += 1;
import_data.first_docid = docid;
}
let document = ctx.index.document(&index_rtxn, docid).map_err(err)?;
let mut document =
obkv_to_json(&all_fields, &fields_ids_map, document).map_err(err)?;
// TODO definitely factorize this code
'inject_vectors: {
let embeddings = ctx.index.embeddings(&index_rtxn, docid).map_err(err)?;
if embeddings.is_empty() {
break 'inject_vectors;
}
let vectors = document
.entry(RESERVED_VECTORS_FIELD_NAME)
.or_insert(serde_json::Value::Object(Default::default()));
let serde_json::Value::Object(vectors) = vectors else {
return Err(err(milli::Error::UserError(
milli::UserError::InvalidVectorsMapType {
document_id: {
if let Ok(Some(Ok(index))) = ctx
.index
.external_id_of(&index_rtxn, std::iter::once(docid))
.map(|it| it.into_iter().next())
{
index
} else {
format!("internal docid={docid}")
}
},
value: vectors.clone(),
},
)));
};
for (
embedder_name,
EmbeddingsWithMetadata { embeddings, regenerate, has_fragments },
) in embeddings
{
let embeddings = ExplicitVectors {
embeddings: Some(VectorOrArrayOfVectors::from_array_of_vectors(
embeddings,
)),
regenerate: regenerate &&
// Meilisearch does not handle well dumps with fragments, because as the fragments
// are marked as user-provided,
// all embeddings would be regenerated on any settings change or document update.
// To prevent this, we mark embeddings has non regenerate in this case.
!has_fragments,
};
vectors
.insert(embedder_name, serde_json::to_value(embeddings).unwrap());
}
}
tmp_buffer.clear();
serde_json::to_writer(&mut tmp_buffer, &document)
.map_err(milli::InternalError::from)
.map_err(milli::Error::from)
.map_err(err)?;
// Make sure we put at least one document in the buffer even
// though we might go above the buffer limit before sending
if !buffer.is_empty() && buffer.len() + tmp_buffer.len() > limit {
// We compress the documents before sending them
let mut encoder =
GzEncoder::new(&mut compressed_buffer, Compression::default());
encoder.write_all(&buffer).map_err(milli::Error::from).map_err(err)?;
encoder.finish().map_err(milli::Error::from).map_err(err)?;
retry(ctx.must_stop_processing, || {
let mut request = ctx.agent.post(&documents_url);
request = request.set("Content-Type", "application/x-ndjson");
request = request.set("Content-Encoding", "gzip");
if let Some(bearer) = &bearer {
request = request.set("Authorization", bearer);
}
if let Some((import_data, origin)) = &task_network {
request = set_network_ureq_headers(request, import_data, origin);
}
request.send_bytes(&compressed_buffer).map_err(into_backoff_error)
})?;
buffer.clear();
compressed_buffer.clear();
if let Some((import_data, _)) = &mut task_network {
import_data.document_count = 0;
import_data.first_docid = 0;
}
}
buffer.extend_from_slice(&tmp_buffer);
if i > 0 && i % 100 == 0 {
step.fetch_add(100, atomic::Ordering::Relaxed);
}
}
retry(ctx.must_stop_processing, || {
let mut request = ctx.agent.post(&documents_url);
request = request.set("Content-Type", "application/x-ndjson");
if let Some((import_data, origin)) = &task_network {
request = set_network_ureq_headers(request, import_data, origin);
}
if let Some(bearer) = &bearer {
request = request.set("Authorization", bearer);
}
request.send_bytes(&buffer).map_err(into_backoff_error)
})?;
Ok(())
})
.map_err(|e| err(milli::Error::InternalError(InternalError::PanicInThreadPool(e))))?;
for result in results {
result?;
}
step.store(total_documents, atomic::Ordering::Relaxed);
Ok(total_documents as u64)
}
}
fn set_network_ureq_headers(
request: ureq::Request,
import_data: &ImportData,
origin: &Origin,
) -> ureq::Request {
request
.set(headers::PROXY_ORIGIN_REMOTE_HEADER, &origin.remote_name)
.set(headers::PROXY_ORIGIN_TASK_UID_HEADER, &origin.task_uid.to_string())
.set(
headers::PROXY_ORIGIN_NETWORK_VERSION_HEADER,
&origin.network_version.to_u128_le().to_string(),
)
.set(headers::PROXY_IMPORT_REMOTE_HEADER, &import_data.remote_name)
.set(headers::PROXY_IMPORT_INDEX_HEADER, &import_data.index_name)
.set(headers::PROXY_IMPORT_FIRST_DOC_HEADER, &import_data.first_docid.to_string())
.set(headers::PROXY_IMPORT_DOCS_HEADER, &import_data.document_count.to_string())
}
fn retry<F>(must_stop_processing: &MustStopProcessing, send_request: F) -> Result<ureq::Response>
@@ -374,4 +417,36 @@ fn ureq_error_into_error(error: ureq::Error) -> Error {
}
}
// export_one_index arguments
pub(super) struct TargetInstance<'a> {
pub(super) base_url: &'a str,
pub(super) api_key: Option<&'a str>,
}
pub(super) struct ExportOptions<'a> {
pub(super) index_uid: &'a str,
pub(super) payload_size: Option<&'a Byte>,
pub(super) override_settings: bool,
pub(super) export_mode: ExportMode<'a>,
}
pub(super) struct ExportContext<'a> {
pub(super) index: &'a meilisearch_types::milli::Index,
pub(super) index_rtxn: &'a milli::heed::RoTxn<'a>,
pub(super) universe: &'a RoaringBitmap,
pub(super) progress: &'a Progress,
pub(super) agent: &'a ureq::Agent,
pub(super) must_stop_processing: &'a MustStopProcessing,
}
pub(super) enum ExportMode<'a> {
ExportRoute,
NetworkBalancing {
index_count: u64,
export_old_remote_name: &'a str,
network_change_origin: &'a Origin,
},
}
// progress related
enum ExportIndex {}

8
crates/index-scheduler/src/utils.rs
View File

@@ -286,6 +286,7 @@ pub fn swap_index_uid_in_task(task: &mut Task, swap: (&str, &str)) {
| K::DumpCreation { .. }
| K::Export { .. }
| K::UpgradeDatabase { .. }
| K::NetworkTopologyChange(_)
| K::SnapshotCreation => (),
};
if let Some(Details::IndexSwap { swaps }) = &mut task.details {
@@ -626,6 +627,13 @@ impl crate::IndexScheduler {
} => {
assert_eq!(kind.as_kind(), Kind::IndexCompaction);
}
Details::NetworkTopologyChange {
moved_documents: _,
received_documents: _,
message: _,
} => {
assert_eq!(kind.as_kind(), Kind::NetworkTopologyChange);
}
}
}

19
crates/meilisearch-types/src/enterprise_edition/network.rs
View File

@@ -7,6 +7,7 @@ use std::collections::BTreeMap;
use milli::update::new::indexer::enterprise_edition::sharding::Shards;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Default)]
#[serde(rename_all = "camelCase")]
@@ -16,20 +17,18 @@ pub struct Network {
#[serde(default)]
pub remotes: BTreeMap<String, Remote>,
#[serde(default)]
pub sharding: bool,
pub leader: Option<String>,
#[serde(default)]
pub version: Uuid,
}
impl Network {
pub fn shards(&self) -> Option<Shards> {
if self.sharding {
let this = self.local.as_deref().expect("Inconsistent `sharding` and `self`");
let others = self
.remotes
.keys()
.filter(|name| name.as_str() != this)
.map(|name| name.to_owned())
.collect();
Some(Shards { own: vec![this.to_owned()], others })
if self.leader.is_some() {
Some(Shards::from_remotes_local(
self.remotes.keys().map(String::as_str),
self.local.as_deref(),
))
} else {
None
}

4
crates/meilisearch-types/src/error.rs
View File

@@ -268,9 +268,9 @@ InvalidMultiSearchQueryRankingRules , InvalidRequest , BAD_REQU
InvalidMultiSearchQueryPosition , InvalidRequest , BAD_REQUEST ;
InvalidMultiSearchRemote , InvalidRequest , BAD_REQUEST ;
InvalidMultiSearchWeight , InvalidRequest , BAD_REQUEST ;
InvalidNetworkLeader , InvalidRequest , BAD_REQUEST ;
InvalidNetworkRemotes , InvalidRequest , BAD_REQUEST ;
InvalidNetworkSelf , InvalidRequest , BAD_REQUEST ;
InvalidNetworkSharding , InvalidRequest , BAD_REQUEST ;
InvalidNetworkSearchApiKey , InvalidRequest , BAD_REQUEST ;
InvalidNetworkWriteApiKey , InvalidRequest , BAD_REQUEST ;
InvalidNetworkUrl , InvalidRequest , BAD_REQUEST ;
@@ -374,6 +374,7 @@ MissingSearchHybrid , InvalidRequest , BAD_REQU
MissingSwapIndexes , InvalidRequest , BAD_REQUEST ;
MissingTaskFilters , InvalidRequest , BAD_REQUEST ;
NoSpaceLeftOnDevice , System , UNPROCESSABLE_ENTITY;
NotLeader , InvalidRequest , BAD_REQUEST ;
PayloadTooLarge , InvalidRequest , PAYLOAD_TOO_LARGE ;
RemoteBadResponse , System , BAD_GATEWAY ;
RemoteBadRequest , InvalidRequest , BAD_REQUEST ;
@@ -387,6 +388,7 @@ TaskFileNotFound , InvalidRequest , NOT_FOUN
BatchNotFound , InvalidRequest , NOT_FOUND ;
TooManyOpenFiles , System , UNPROCESSABLE_ENTITY ;
TooManyVectors , InvalidRequest , BAD_REQUEST ;
UnexpectedNetworkPreviousRemotes , InvalidRequest , BAD_REQUEST ;
UnretrievableDocument , Internal , BAD_REQUEST ;
UnretrievableErrorCode , InvalidRequest , BAD_REQUEST ;
UnsupportedMediaType , InvalidRequest , UNSUPPORTED_MEDIA_TYPE ;

26
crates/meilisearch-types/src/task_view.rs
View File

@@ -9,9 +9,9 @@ use utoipa::ToSchema;
use crate::batches::BatchId;
use crate::error::ResponseError;
use crate::settings::{Settings, Unchecked};
use crate::tasks::enterprise_edition::network::TaskNetwork;
use crate::tasks::{
serialize_duration, Details, DetailsExportIndexSettings, IndexSwap, Kind, Status, Task, TaskId,
TaskNetwork,
};
#[derive(Debug, Clone, PartialEq, Serialize, ToSchema)]
@@ -147,6 +147,11 @@ pub struct DetailsView {
pub pre_compaction_size: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub post_compaction_size: Option<String>,
// network topology change
#[serde(skip_serializing_if = "Option::is_none")]
pub moved_documents: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
}
impl DetailsView {
@@ -157,6 +162,17 @@ impl DetailsView {
(None, Some(doc)) | (Some(doc), None) => Some(doc),
(Some(left), Some(right)) => Some(left + right),
},
moved_documents: match (self.moved_documents, other.moved_documents) {
(None, None) => None,
(None, Some(doc)) | (Some(doc), None) => Some(doc),
(Some(left), Some(right)) => Some(left + right),
},
message: match (&mut self.message, &other.message) {
(None, None) => None,
(None, Some(message)) => Some(message.clone()),
(Some(message), None) => Some(std::mem::take(message)),
(Some(message), Some(_)) => Some(std::mem::take(message)),
},
indexed_documents: match (self.indexed_documents, other.indexed_documents) {
(None, None) => None,
(None, Some(None)) | (Some(None), None) | (Some(None), Some(None)) => Some(None),
@@ -447,6 +463,14 @@ impl From<Details> for DetailsView {
..Default::default()
}
}
Details::NetworkTopologyChange { moved_documents, received_documents, message } => {
DetailsView {
moved_documents: Some(moved_documents),
received_documents: Some(received_documents),
message: Some(message),
..Default::default()
}
}
}
}
}

6
crates/meilisearch-types/src/tasks/enterprise_edition/mod.rs Normal file
View File

@@ -0,0 +1,6 @@
// Copyright © 2025 Meilisearch Some Rights Reserved
// This file is part of Meilisearch Enterprise Edition (EE).
// Use of this source code is governed by the Business Source License 1.1,
// as found in the LICENSE-EE file or at <https://mariadb.com/bsl11>
pub mod network;

524
crates/meilisearch-types/src/tasks/enterprise_edition/network.rs Normal file
View File

@@ -0,0 +1,524 @@
// Copyright © 2025 Meilisearch Some Rights Reserved
// This file is part of Meilisearch Enterprise Edition (EE).
// Use of this source code is governed by the Business Source License 1.1,
// as found in the LICENSE-EE file or at <https://mariadb.com/bsl11>
use std::collections::BTreeMap;
use milli::DocumentId;
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
use uuid::Uuid;
use crate::enterprise_edition::network::{Network, Remote};
use crate::error::ResponseError;
use crate::tasks::{Details, TaskId};
#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, ToSchema)]
#[serde(untagged, rename_all = "camelCase")]
pub enum TaskNetwork {
/// Tasks that were duplicated from `origin`
Origin { origin: Origin },
/// Tasks that were duplicated as `remote_tasks`
Remotes {
remote_tasks: BTreeMap<String, RemoteTask>,
#[serde(default)]
network_version: Uuid,
},
/// Document import tasks sent in the context of `network_change`
Import { import_from: ImportData, network_change: Origin },
}
impl TaskNetwork {
pub fn network_version(&self) -> Uuid {
match self {
TaskNetwork::Origin { origin } => origin.network_version,
TaskNetwork::Remotes { remote_tasks: _, network_version } => *network_version,
TaskNetwork::Import { import_from: _, network_change } => {
network_change.network_version
}
}
}
pub fn import_data(&self) -> Option<&ImportData> {
match self {
TaskNetwork::Origin { .. } | TaskNetwork::Remotes { .. } => None,
TaskNetwork::Import { import_from, .. } => Some(import_from),
}
}
pub fn origin(&self) -> Option<&Origin> {
match self {
TaskNetwork::Origin { origin } => Some(origin),
TaskNetwork::Remotes { .. } => None,
TaskNetwork::Import { network_change, .. } => Some(network_change),
}
}
}
#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct Origin {
pub remote_name: String,
pub task_uid: u32,
pub network_version: Uuid,
}
#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct ImportData {
pub remote_name: String,
pub index_name: String,
pub first_docid: DocumentId,
pub document_count: u64,
}
#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct RemoteTask {
#[serde(skip_serializing_if = "Option::is_none")]
task_uid: Option<TaskId>,
error: Option<ResponseError>,
}
impl From<Result<TaskId, ResponseError>> for RemoteTask {
fn from(res: Result<TaskId, ResponseError>) -> RemoteTask {
match res {
Ok(task_uid) => RemoteTask { task_uid: Some(task_uid), error: None },
Err(err) => RemoteTask { task_uid: None, error: Some(err) },
}
}
}
/// Contains the full state of a network topology change.
///
/// A network topology change task is unique in that it can be processed in multiple different batches, as its resolution
/// depends on various document additions tasks being processed.
///
/// A network topology task has 4 states:
///
/// 1. Processing any task that was meant for an earlier version of the network. This is necessary to know that we have the right version of
/// documents.
/// 2. Sending all documents that must be moved to other remotes.
/// 3. Processing any task coming from the remotes.
/// 4. Finished.
///
/// Furthermore, it maintains some stats
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct NetworkTopologyChange {
state: NetworkTopologyState,
// in name, `None` if the node is no longer part of the network
#[serde(skip_serializing_if = "Option::is_none")]
in_name: Option<String>,
// out name, `None` if the node is new to the network
#[serde(skip_serializing_if = "Option::is_none")]
out_name: Option<String>,
out_remotes: BTreeMap<String, Remote>,
in_remotes: BTreeMap<String, InRemote>,
stats: NetworkTopologyStats,
}
impl NetworkTopologyChange {
pub fn new(old_network: Network, new_network: Network) -> Self {
// we use our old name as export name
let out_name = old_network.local;
// we use our new name as import name
let in_name = new_network.local;
// we export to the new network
let mut out_remotes = new_network.remotes;
// don't export to ourselves
if let Some(in_name) = &in_name {
out_remotes.remove(in_name);
}
/// FIXME: doesn't work if the old network is not the same for old nodes
let in_remotes = old_network
.remotes
.into_keys()
// don't await imports from ourselves
.filter(|name| Some(name.as_str()) != out_name.as_deref())
.map(|name| (name, InRemote::new()))
.collect();
Self {
state: NetworkTopologyState::WaitingForOlderTasks,
in_name,
out_name,
out_remotes,
in_remotes,
stats: NetworkTopologyStats { received_documents: 0, moved_documents: 0 },
}
}
pub fn state(&self) -> NetworkTopologyState {
self.state
}
pub fn out_name(&self) -> Option<&str> {
// unwrap: one of out name or in_name must be defined
self.out_name.as_deref()
}
pub fn in_name(&self) -> Option<&str> {
self.in_name.as_deref()
}
pub fn export_to_process(&self) -> Option<(&BTreeMap<String, Remote>, &str)> {
if self.state != NetworkTopologyState::ExportingDocuments {
return None;
}
if self.out_remotes.is_empty() {
return None;
}
let out_name = self.out_name()?;
Some((&self.out_remotes, out_name))
}
/// Compute the next state from the current state of the task.
pub fn update_state(&mut self) {
self.state = match self.state {
NetworkTopologyState::WaitingForOlderTasks => {
// no more older tasks, so finished waiting
NetworkTopologyState::ExportingDocuments
}
NetworkTopologyState::ExportingDocuments => {
// processed all exported documents
NetworkTopologyState::ImportingDocuments
}
NetworkTopologyState::ImportingDocuments => {
if self.is_import_finished() {
NetworkTopologyState::Finished
} else {
NetworkTopologyState::ImportingDocuments
}
}
NetworkTopologyState::Finished => NetworkTopologyState::Finished,
};
}
pub fn receive_remote_task(
&mut self,
remote_name: &str,
index_name: &str,
first_docid: DocumentId,
document_count: u64,
total_indexes: u64,
total_index_documents: u64,
) -> Result<(), ReceiveTaskError> {
let remote = self
.in_remotes
.get_mut(remote_name)
.ok_or_else(|| ReceiveTaskError::UnknownRemote(remote_name.to_string()))?;
remote.import_state = match std::mem::take(&mut remote.import_state) {
ImportState::WaitingForInitialTask => {
let mut import_index_state = BTreeMap::new();
import_index_state.insert(
index_name.to_owned(),
ImportIndexState::Ongoing {
total_documents: total_index_documents,
received_documents: document_count,
first_docids: vec![first_docid],
processed_documents: 0,
},
);
ImportState::Ongoing { import_index_state, total_indexes }
}
ImportState::Ongoing { mut import_index_state, total_indexes } => {
if let Some((index_name, mut index_state)) =
import_index_state.remove_entry(index_name)
{
index_state = match index_state {
ImportIndexState::Ongoing {
total_documents,
received_documents: previously_received,
processed_documents,
mut first_docids,
} => {
if first_docids.contains(&first_docid) {
return Err(ReceiveTaskError::DuplicateTask(first_docid));
}
first_docids.push(first_docid);
ImportIndexState::Ongoing {
total_documents,
received_documents: previously_received + document_count,
processed_documents,
first_docids,
}
}
ImportIndexState::Finished { total_documents } => {
ImportIndexState::Finished { total_documents }
}
};
import_index_state.insert(index_name, index_state);
} else {
let state = ImportIndexState::Ongoing {
total_documents: total_index_documents,
received_documents: document_count,
processed_documents: 0,
first_docids: vec![first_docid],
};
import_index_state.insert(index_name.to_string(), state);
}
ImportState::Ongoing { import_index_state, total_indexes: total_indexes }
}
ImportState::Finished { total_indexes, total_documents } => {
ImportState::Finished { total_indexes, total_documents }
}
};
Ok(())
}
pub fn process_remote_tasks(
&mut self,
remote_name: &str,
index_name: &str,
document_count: u64,
) {
/// FIXME: unwraps and panics
let remote = self.in_remotes.get_mut(remote_name).unwrap();
remote.import_state = match std::mem::take(&mut remote.import_state) {
ImportState::WaitingForInitialTask => panic!("no task received yet one processed"),
ImportState::Ongoing { mut import_index_state, total_indexes } => {
let (index_name, mut index_state) =
import_index_state.remove_entry(index_name).unwrap();
index_state = match index_state {
ImportIndexState::Ongoing {
total_documents,
received_documents,
processed_documents: previously_processed,
first_docids,
} => {
let newly_processed_documents = previously_processed + document_count;
if newly_processed_documents >= total_documents {
ImportIndexState::Finished { total_documents }
} else {
ImportIndexState::Ongoing {
total_documents,
received_documents,
processed_documents: newly_processed_documents,
first_docids,
}
}
}
ImportIndexState::Finished { total_documents } => {
ImportIndexState::Finished { total_documents }
}
};
import_index_state.insert(index_name, index_state);
if import_index_state.len() as u64 == total_indexes
&& import_index_state.values().all(|index| index.is_finished())
{
let total_documents =
import_index_state.values().map(|index| index.total_documents()).sum();
ImportState::Finished { total_indexes, total_documents }
} else {
ImportState::Ongoing { import_index_state, total_indexes }
}
}
ImportState::Finished { total_indexes, total_documents } => {
ImportState::Finished { total_indexes, total_documents }
}
}
}
pub fn to_details(&self) -> Details {
let message = match self.state {
NetworkTopologyState::WaitingForOlderTasks => {
"Waiting for tasks enqueued before the network change to finish processing".into()
}
NetworkTopologyState::ExportingDocuments => "Exporting documents".into(),
NetworkTopologyState::ImportingDocuments => {
let mut finished_count = 0;
let mut first_ongoing = None;
let mut ongoing_total_indexes = 0;
let mut ongoing_processed_documents = 0;
let mut ongoing_missing_documents = 0;
let mut ongoing_total_documents = 0;
let mut other_ongoing_count = 0;
let mut first_waiting = None;
let mut other_waiting_count = 0;
for (remote_name, in_remote) in &self.in_remotes {
match &in_remote.import_state {
ImportState::WaitingForInitialTask => {
first_waiting = match first_waiting {
None => Some(remote_name),
first_waiting => {
other_waiting_count += 1;
first_waiting
}
};
}
ImportState::Ongoing { import_index_state, total_indexes } => {
first_ongoing = match first_ongoing {
None => {
ongoing_total_indexes = *total_indexes;
Some(remote_name)
}
first_ongoing => {
other_ongoing_count += 1;
first_ongoing
}
};
for import_state in import_index_state.values() {
match import_state {
ImportIndexState::Ongoing {
total_documents,
processed_documents,
received_documents,
first_docids: _,
} => {
ongoing_total_documents += total_documents;
ongoing_processed_documents += processed_documents;
ongoing_missing_documents +=
total_documents.saturating_sub(*received_documents);
}
ImportIndexState::Finished { total_documents } => {
ongoing_total_documents += total_documents;
ongoing_processed_documents += total_documents;
}
}
}
}
ImportState::Finished { total_indexes, total_documents } => {
finished_count += 1;
ongoing_total_indexes = *total_indexes;
ongoing_total_documents += *total_documents;
ongoing_processed_documents += *total_documents;
}
}
}
format!(
"Importing documents from {total} remotes{waiting}{ongoing}{finished}",
total = self.in_remotes.len(),
waiting = if let Some(first_waiting) = first_waiting {
&format!(
", waiting on first task from `{}`{others}",
first_waiting,
others = if other_waiting_count > 0 {
&format!(" and {other_waiting_count} other remotes")
} else {
""
}
)
} else {
""
},
ongoing = if let Some(first_ongoing) = first_ongoing {
&format!(", awaiting {ongoing_missing_documents} and processed {ongoing_processed_documents} out of {ongoing_total_documents} documents in {ongoing_total_indexes} indexes from `{first_ongoing}`{others}",
others=if other_ongoing_count > 0 {&format!(" and {other_ongoing_count} other remotes")} else {""})
} else {
""
},
finished = if finished_count >= 0 {
&format!(", {finished_count} remotes finished processing")
} else {
""
}
)
}
NetworkTopologyState::Finished => "Finished".into(),
};
Details::NetworkTopologyChange {
moved_documents: self.stats.moved_documents,
received_documents: self.stats.received_documents,
message,
}
}
fn is_import_finished(&self) -> bool {
self.in_remotes.values().all(|remote| remote.is_finished())
}
}
pub enum ReceiveTaskError {
UnknownRemote(String),
DuplicateTask(DocumentId),
}
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum NetworkTopologyState {
WaitingForOlderTasks,
ExportingDocuments,
ImportingDocuments,
Finished,
}
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct NetworkTopologyStats {
pub received_documents: u64,
pub moved_documents: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct InRemote {
import_state: ImportState,
}
impl InRemote {
pub fn new() -> Self {
Self { import_state: ImportState::WaitingForInitialTask }
}
pub fn is_finished(&self) -> bool {
matches!(self.import_state, ImportState::Finished { .. })
}
}
#[derive(Default, Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
enum ImportState {
/// Initially Meilisearch doesn't know how many documents it should expect from a remote.
/// The first task for each remote contains the information of how many indexes will be imported,
/// and the first task for each index contains the number of documents to import for that index.
#[default]
WaitingForInitialTask,
Ongoing {
import_index_state: BTreeMap<String, ImportIndexState>,
total_indexes: u64,
},
Finished {
total_indexes: u64,
total_documents: u64,
},
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
enum ImportIndexState {
Ongoing {
total_documents: u64,
received_documents: u64,
processed_documents: u64,
first_docids: Vec<DocumentId>,
},
Finished {
total_documents: u64,
},
}
impl ImportIndexState {
pub fn is_finished(&self) -> bool {
matches!(self, ImportIndexState::Finished { .. })
}
fn total_documents(&self) -> u64 {
match *self {
ImportIndexState::Ongoing { total_documents, .. }
| ImportIndexState::Finished { total_documents } => total_documents,
}
}
}
pub mod headers {
pub const PROXY_ORIGIN_REMOTE_HEADER: &str = "Meili-Proxy-Origin-Remote";
pub const PROXY_ORIGIN_TASK_UID_HEADER: &str = "Meili-Proxy-Origin-TaskUid";
pub const PROXY_ORIGIN_NETWORK_VERSION_HEADER: &str = "Meili-Proxy-Origin-Network-Version";
pub const PROXY_IMPORT_REMOTE_HEADER: &str = "Meili-Proxy-Import-Remote";
pub const PROXY_IMPORT_INDEX_HEADER: &str = "Meili-Proxy-Import-Index";
pub const PROXY_IMPORT_FIRST_DOC_HEADER: &str = "Meili-Proxy-Import-First-Doc";
pub const PROXY_IMPORT_DOCS_HEADER: &str = "Meili-Proxy-Import-Docs";
}

65
crates/meilisearch-types/src/tasks.rs → crates/meilisearch-types/src/tasks/mod.rs
View File

@@ -23,6 +23,8 @@ use crate::{versioning, InstanceUid};
pub type TaskId = u32;
pub mod enterprise_edition;
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Task {
@@ -44,7 +46,7 @@ pub struct Task {
pub kind: KindWithContent,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub network: Option<TaskNetwork>,
pub network: Option<enterprise_edition::network::TaskNetwork>,
}
impl Task {
@@ -58,6 +60,7 @@ impl Task {
| TaskDeletion { .. }
| Export { .. }
| UpgradeDatabase { .. }
| NetworkTopologyChange { .. }
| IndexSwap { .. } => None,
DocumentAdditionOrUpdate { index_uid, .. }
| DocumentEdition { index_uid, .. }
@@ -96,6 +99,7 @@ impl Task {
| KindWithContent::SnapshotCreation
| KindWithContent::Export { .. }
| KindWithContent::UpgradeDatabase { .. }
| KindWithContent::NetworkTopologyChange { .. }
| KindWithContent::IndexCompaction { .. } => None,
}
}
@@ -175,6 +179,7 @@ pub enum KindWithContent {
IndexCompaction {
index_uid: String,
},
NetworkTopologyChange(enterprise_edition::network::NetworkTopologyChange),
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
@@ -212,6 +217,7 @@ impl KindWithContent {
KindWithContent::Export { .. } => Kind::Export,
KindWithContent::UpgradeDatabase { .. } => Kind::UpgradeDatabase,
KindWithContent::IndexCompaction { .. } => Kind::IndexCompaction,
KindWithContent::NetworkTopologyChange { .. } => Kind::NetworkTopologyChange,
}
}
@@ -224,6 +230,7 @@ impl KindWithContent {
| TaskCancelation { .. }
| TaskDeletion { .. }
| Export { .. }
| NetworkTopologyChange { .. }
| UpgradeDatabase { .. } => vec![],
DocumentAdditionOrUpdate { index_uid, .. }
| DocumentEdition { index_uid, .. }
@@ -337,6 +344,11 @@ impl KindWithContent {
pre_compaction_size: None,
post_compaction_size: None,
}),
KindWithContent::NetworkTopologyChange { .. } => Some(Details::NetworkTopologyChange {
moved_documents: 0,
received_documents: 0,
message: "processing tasks for previous network versions".into(),
}),
}
}
@@ -389,7 +401,7 @@ impl KindWithContent {
})
}
KindWithContent::IndexSwap { .. } => {
todo!()
unimplemented!("do not call `default_finished_details` for `IndexSwap` tasks")
}
KindWithContent::TaskCancelation { query, tasks } => Some(Details::TaskCancelation {
matched_tasks: tasks.len(),
@@ -424,6 +436,9 @@ impl KindWithContent {
pre_compaction_size: None,
post_compaction_size: None,
}),
KindWithContent::NetworkTopologyChange(network_topology_change) => {
Some(network_topology_change.to_details())
}
}
}
}
@@ -491,6 +506,9 @@ impl From<&KindWithContent> for Option<Details> {
pre_compaction_size: None,
post_compaction_size: None,
}),
KindWithContent::NetworkTopologyChange(network_topology_change) => {
Some(network_topology_change.to_details())
}
}
}
}
@@ -602,6 +620,7 @@ pub enum Kind {
Export,
UpgradeDatabase,
IndexCompaction,
NetworkTopologyChange,
}
impl Kind {
@@ -621,6 +640,7 @@ impl Kind {
| Kind::DumpCreation
| Kind::Export
| Kind::UpgradeDatabase
| Kind::NetworkTopologyChange
| Kind::SnapshotCreation => false,
}
}
@@ -643,6 +663,7 @@ impl Display for Kind {
Kind::Export => write!(f, "export"),
Kind::UpgradeDatabase => write!(f, "upgradeDatabase"),
Kind::IndexCompaction => write!(f, "indexCompaction"),
Kind::NetworkTopologyChange => write!(f, "networkTopologyChange"),
}
}
}
@@ -680,6 +701,8 @@ impl FromStr for Kind {
Ok(Kind::UpgradeDatabase)
} else if kind.eq_ignore_ascii_case("indexCompaction") {
Ok(Kind::IndexCompaction)
} else if kind.eq_ignore_ascii_case("networkTopologyChange") {
Ok(Kind::NetworkTopologyChange)
} else {
Err(ParseTaskKindError(kind.to_owned()))
}
@@ -770,36 +793,11 @@ pub enum Details {
pre_compaction_size: Option<Byte>,
post_compaction_size: Option<Byte>,
},
}
#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, ToSchema)]
#[serde(untagged, rename_all = "camelCase")]
pub enum TaskNetwork {
Origin { origin: Origin },
Remotes { remote_tasks: BTreeMap<String, RemoteTask> },
}
#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct Origin {
pub remote_name: String,
pub task_uid: usize,
}
#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct RemoteTask {
#[serde(skip_serializing_if = "Option::is_none")]
task_uid: Option<TaskId>,
error: Option<ResponseError>,
}
impl From<Result<TaskId, ResponseError>> for RemoteTask {
fn from(res: Result<TaskId, ResponseError>) -> RemoteTask {
match res {
Ok(task_uid) => RemoteTask { task_uid: Some(task_uid), error: None },
Err(err) => RemoteTask { task_uid: None, error: Some(err) },
}
}
NetworkTopologyChange {
moved_documents: u64,
received_documents: u64,
message: String,
},
}
#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, ToSchema)]
@@ -842,6 +840,9 @@ impl Details {
| Self::Export { .. }
| Self::UpgradeDatabase { .. }
| Self::IndexSwap { .. } => (),
Self::NetworkTopologyChange { moved_documents: _, received_documents: _, message } => {
*message = format!("Failed. Previous status: {}", message);
}
}
details

38
crates/meilisearch/src/error.rs
View File

@@ -6,6 +6,10 @@ use meilisearch_types::error::{Code, ErrorCode, ResponseError};
use meilisearch_types::index_uid::{IndexUid, IndexUidFormatError};
use meilisearch_types::milli;
use meilisearch_types::milli::OrderBy;
use meilisearch_types::tasks::enterprise_edition::network::headers::{
PROXY_IMPORT_DOCS_HEADER, PROXY_IMPORT_FIRST_DOC_HEADER, PROXY_IMPORT_INDEX_HEADER,
PROXY_IMPORT_REMOTE_HEADER,
};
use serde_json::Value;
use tokio::task::JoinError;
@@ -91,8 +95,32 @@ pub enum MeilisearchHttpError {
} else { PROXY_ORIGIN_TASK_UID_HEADER }
)]
InconsistentOriginHeaders { is_remote_missing: bool },
#[error("Inconsistent `Import` headers: {remote}: {remote_status}, {index}: {index_status}, {last}: {last_status}, {docs}: {docs_status}.\n - Hint: either all four headers should be provided, or none of them",
remote = PROXY_IMPORT_REMOTE_HEADER,
remote_status = if *is_remote_missing { "missing" } else{ "provided" },
index = PROXY_IMPORT_INDEX_HEADER,
index_status = if *is_index_missing { "missing" } else { "provided" },
last = PROXY_IMPORT_FIRST_DOC_HEADER,
last_status = if *is_last_docs_missing { "missing" } else { "provided" },
docs = PROXY_IMPORT_DOCS_HEADER,
docs_status = if *is_docs_missing { "missing" } else { "provided" }
)]
InconsistentImportHeaders {
is_remote_missing: bool,
is_index_missing: bool,
is_last_docs_missing: bool,
is_docs_missing: bool,
},
#[error(
"Inconsistent headers: `Import` headers are present, but `Origin` headers are missing."
)]
MissingOriginHeaders,
#[error("Invalid value for header {header_name}: {msg}")]
InvalidHeaderValue { header_name: &'static str, msg: String },
#[error("This remote is not the leader of the network.\n - Note: only the leader `{leader}` can receive new tasks.")]
NotLeader { leader: String },
#[error("Unexpected `previousRemotes` in network call.\n - Note: `previousRemote` is reserved for internal use.")]
UnexpectedNetworkPreviousRemotes,
}
impl MeilisearchHttpError {
@@ -140,7 +168,17 @@ impl ErrorCode for MeilisearchHttpError {
MeilisearchHttpError::InconsistentOriginHeaders { .. } => {
Code::InconsistentDocumentChangeHeaders
}
MeilisearchHttpError::InconsistentImportHeaders { .. } => {
Code::InconsistentDocumentChangeHeaders
}
MeilisearchHttpError::MissingOriginHeaders { .. } => {
Code::InconsistentDocumentChangeHeaders
}
MeilisearchHttpError::InvalidHeaderValue { .. } => Code::InvalidHeaderValue,
MeilisearchHttpError::NotLeader { .. } => Code::NotLeader,
MeilisearchHttpError::UnexpectedNetworkPreviousRemotes => {
Code::UnexpectedNetworkPreviousRemotes
}
}
}
}

64
crates/meilisearch/src/routes/indexes/documents.rs
View File

@@ -45,7 +45,9 @@ use crate::extractors::authentication::policies::*;
use crate::extractors::authentication::GuardedData;
use crate::extractors::payload::Payload;
use crate::extractors::sequential_extractor::SeqHandler;
use crate::routes::indexes::enterprise_edition::proxy::{proxy, Body};
use crate::routes::indexes::enterprise_edition::proxy::{
proxy, task_network_and_check_leader, Body,
};
use crate::routes::indexes::search::fix_sort_query_parameters;
use crate::routes::{
get_task_id, is_dry_run, PaginationView, SummarizedTaskView, PAGINATION_DEFAULT_LIMIT,
@@ -340,6 +342,7 @@ pub async fn delete_document(
let DocumentParam { index_uid, document_id } = path.into_inner();
let index_uid = IndexUid::try_from(index_uid)?;
let network = index_scheduler.network();
let task_network = task_network_and_check_leader(&req, &network)?;
analytics.publish(
DocumentsDeletionAggregator {
@@ -362,8 +365,9 @@ pub async fn delete_document(
tokio::task::spawn_blocking(move || index_scheduler.register(task, uid, dry_run)).await??
};
if network.sharding && !dry_run {
proxy(&index_scheduler, &index_uid, &req, network, Body::none(), &task).await?;
if network.leader.is_some() && !dry_run {
proxy(&index_scheduler, Some(&index_uid), &req, task_network, network, Body::none(), &task)
.await?;
}
let task: SummarizedTaskView = task.into();
@@ -946,6 +950,7 @@ async fn document_addition(
) -> Result<SummarizedTaskView, MeilisearchHttpError> {
let mime_type = extract_mime_type(req)?;
let network = index_scheduler.network();
let task_network = task_network_and_check_leader(&req, &network)?;
let format = match (
mime_type.as_ref().map(|m| (m.type_().as_str(), m.subtype().as_str())),
@@ -1075,12 +1080,13 @@ async fn document_addition(
}
};
if network.sharding {
if network.leader.is_some() {
if let Some(file) = file {
proxy(
&index_scheduler,
&index_uid,
Some(&index_uid),
req,
task_network,
network,
Body::with_ndjson_payload(file),
&task,
@@ -1168,6 +1174,7 @@ pub async fn delete_documents_batch(
debug!(parameters = ?body, "Delete documents by batch");
let index_uid = IndexUid::try_from(index_uid.into_inner())?;
let network = index_scheduler.network();
let task_network = task_network_and_check_leader(&req, &network)?;
analytics.publish(
DocumentsDeletionAggregator {
@@ -1193,8 +1200,17 @@ pub async fn delete_documents_batch(
tokio::task::spawn_blocking(move || index_scheduler.register(task, uid, dry_run)).await??
};
if network.sharding && !dry_run {
proxy(&index_scheduler, &index_uid, &req, network, Body::Inline(body), &task).await?;
if network.leader.is_some() && !dry_run {
proxy(
&index_scheduler,
Some(&index_uid),
&req,
task_network,
network,
Body::inline(body),
&task,
)
.await?;
}
let task: SummarizedTaskView = task.into();
@@ -1254,6 +1270,7 @@ pub async fn delete_documents_by_filter(
let index_uid = index_uid.into_inner();
let filter = body.into_inner();
let network = index_scheduler.network();
let task_network = task_network_and_check_leader(&req, &network)?;
analytics.publish(
DocumentsDeletionAggregator {
@@ -1285,8 +1302,17 @@ pub async fn delete_documents_by_filter(
tokio::task::spawn_blocking(move || index_scheduler.register(task, uid, dry_run)).await??
};
if network.sharding && !dry_run {
proxy(&index_scheduler, &index_uid, &req, network, Body::Inline(filter), &task).await?;
if network.leader.is_some() && !dry_run {
proxy(
&index_scheduler,
Some(&index_uid),
&req,
task_network,
network,
Body::inline(filter),
&task,
)
.await?;
}
let task: SummarizedTaskView = task.into();
@@ -1384,6 +1410,7 @@ pub async fn edit_documents_by_function(
.check_edit_documents_by_function("Using the documents edit route")?;
let network = index_scheduler.network();
let task_network = task_network_and_check_leader(&req, &network)?;
let index_uid = IndexUid::try_from(index_uid.into_inner())?;
let index_uid = index_uid.into_inner();
@@ -1435,8 +1462,17 @@ pub async fn edit_documents_by_function(
tokio::task::spawn_blocking(move || index_scheduler.register(task, uid, dry_run)).await??
};
if network.sharding && !dry_run {
proxy(&index_scheduler, &index_uid, &req, network, Body::Inline(params), &task).await?;
if network.leader.is_some() && !dry_run {
proxy(
&index_scheduler,
Some(&index_uid),
&req,
task_network,
network,
Body::inline(params),
&task,
)
.await?;
}
let task: SummarizedTaskView = task.into();
@@ -1483,6 +1519,7 @@ pub async fn clear_all_documents(
) -> Result<HttpResponse, ResponseError> {
let index_uid = IndexUid::try_from(index_uid.into_inner())?;
let network = index_scheduler.network();
let task_network = task_network_and_check_leader(&req, &network)?;
analytics.publish(
DocumentsDeletionAggregator {
@@ -1504,8 +1541,9 @@ pub async fn clear_all_documents(
tokio::task::spawn_blocking(move || index_scheduler.register(task, uid, dry_run)).await??
};
if network.sharding && !dry_run {
proxy(&index_scheduler, &index_uid, &req, network, Body::none(), &task).await?;
if network.leader.is_some() && !dry_run {
proxy(&index_scheduler, Some(&index_uid), &req, task_network, network, Body::none(), &task)
.await?;
}
let task: SummarizedTaskView = task.into();

513
crates/meilisearch/src/routes/indexes/enterprise_edition/proxy.rs
View File

@@ -3,6 +3,7 @@
// Use of this source code is governed by the Business Source License 1.1,
// as found in the LICENSE-EE file or at <https://mariadb.com/bsl11>
use std::borrow::Cow;
use std::collections::BTreeMap;
use std::fs::File;
@@ -10,8 +11,16 @@ use actix_web::http::header::CONTENT_TYPE;
use actix_web::HttpRequest;
use bytes::Bytes;
use index_scheduler::IndexScheduler;
use meilisearch_types::enterprise_edition::network::Remote;
use meilisearch_types::error::ResponseError;
use meilisearch_types::tasks::{Origin, RemoteTask, TaskNetwork};
use meilisearch_types::milli::DocumentId;
use meilisearch_types::tasks::enterprise_edition::network::headers::{
PROXY_IMPORT_DOCS_HEADER, PROXY_IMPORT_FIRST_DOC_HEADER, PROXY_IMPORT_INDEX_HEADER,
PROXY_IMPORT_REMOTE_HEADER, PROXY_ORIGIN_NETWORK_VERSION_HEADER, PROXY_ORIGIN_REMOTE_HEADER,
PROXY_ORIGIN_TASK_UID_HEADER,
};
use meilisearch_types::tasks::enterprise_edition::network::{ImportData, Origin, TaskNetwork};
use meilisearch_types::tasks::Task;
use reqwest::StatusCode;
use serde::de::DeserializeOwned;
use serde_json::Value;
@@ -22,13 +31,18 @@ use crate::routes::indexes::enterprise_edition::proxy::error::{
};
use crate::routes::SummarizedTaskView;
pub enum Body<T: serde::Serialize> {
pub enum Body<T, F>
where
T: serde::Serialize,
F: FnMut(&str, &Remote, &mut T),
{
NdJsonPayload(File),
Inline(T),
Generated(T, F),
None,
}
impl Body<()> {
impl Body<(), fn(&str, &Remote, &mut ())> {
pub fn with_ndjson_payload(file: File) -> Self {
Self::NdJsonPayload(file)
}
@@ -38,7 +52,106 @@ impl Body<()> {
}
}
/// If necessary, proxies the passed request to the network and update the task description.
impl<T> Body<T, fn(&str, &Remote, &mut T)>
where
T: serde::Serialize,
{
pub fn inline(payload: T) -> Self {
Self::Inline(payload)
}
}
impl<T, F> Body<T, F>
where
T: serde::Serialize,
F: FnMut(&str, &Remote, &mut T),
{
pub fn generated(initial: T, f: F) -> Self {
Self::Generated(initial, f)
}
}
impl<T, F> Body<T, F>
where
T: serde::Serialize,
F: FnMut(&str, &Remote, &mut T),
{
pub fn into_bytes_iter(
self,
remotes: impl IntoIterator<Item = (String, Remote)>,
) -> Result<
impl Iterator<Item = (Option<Bytes>, (String, Remote))>,
meilisearch_types::milli::Error,
> {
let bytes = match self {
Body::NdJsonPayload(file) => {
Some(Bytes::from_owner(unsafe { memmap2::Mmap::map(&file)? }))
}
Body::Inline(payload) => {
Some(Bytes::copy_from_slice(&serde_json::to_vec(&payload).unwrap()))
}
Body::None => None,
Body::Generated(mut initial, mut f) => {
return Ok(either::Right(remotes.into_iter().map(move |(name, remote)| {
f(&name, &remote, &mut initial);
let bytes =
Some(Bytes::copy_from_slice(&serde_json::to_vec(&initial).unwrap()));
(bytes, (name, remote))
})));
}
};
Ok(either::Left(std::iter::repeat(bytes).zip(remotes)))
}
}
/// Parses the header to determine if this task is a duplicate and originates with a remote.
///
/// If not, checks whether this remote is the leader and return `MeilisearchHttpError::NotLeader` if not.
///
/// # Errors
///
/// - `MeiliearchHttpError::NotLeader`: if the following are true simultaneously:
/// 1. The task originates with the current node
/// 2. There's a declared `leader`
/// 3. The declared leader is **not** the current node
/// - `MeilisearchHttpError::InvalidHeaderValue`: if only parts of the headers are present, or if they cannot be parsed as a task network.
/// - `MeilisearchHttpError::Inconsistent`
pub fn task_network_and_check_leader(
req: &HttpRequest,
network: &meilisearch_types::enterprise_edition::network::Network,
) -> Result<TaskNetwork, MeilisearchHttpError> {
match (origin_from_req(req)?, import_from_req(req)?) {
(Some(network_change), Some(import_from)) => {
Ok(TaskNetwork::Import { import_from, network_change })
}
(Some(origin), None) => Ok(TaskNetwork::Origin { origin }),
(None, Some(_)) => Err(MeilisearchHttpError::MissingOriginHeaders),
(None, None) => {
match (network.leader.as_deref(), network.local.as_deref()) {
// 1. Always allowed if there is no leader
(None, _) => (),
// 2. Allowed if the leader is self
(Some(leader), Some(this)) if leader == this => (),
// 3. Any other change is disallowed
(Some(leader), _) => {
return Err(
MeilisearchHttpError::NotLeader { leader: leader.to_string() }.into()
)
}
}
Ok(TaskNetwork::Remotes {
remote_tasks: Default::default(),
network_version: network.version,
})
}
}
}
/// Updates the task description and, if necessary, proxies the passed request to the network and update the task description.
///
/// This function reads the custom headers from the request to determine if must proxy the request or if the request
/// has already been proxied.
@@ -48,152 +161,139 @@ impl Body<()> {
/// with the task ids from the task queues of the remotes.
/// - when the request has already been proxied, the custom headers contains information about the remote that created the initial task.
/// This information is copied to the passed task.
pub async fn proxy<T: serde::Serialize>(
///
/// # Returns
///
/// The updated task. The task is read back from the database to avoid erasing concurrent changes.
pub async fn proxy<T, F>(
index_scheduler: &IndexScheduler,
index_uid: &str,
index_uid: Option<&str>,
req: &HttpRequest,
mut task_network: TaskNetwork,
network: meilisearch_types::enterprise_edition::network::Network,
body: Body<T>,
body: Body<T, F>,
task: &meilisearch_types::tasks::Task,
) -> Result<(), MeilisearchHttpError> {
match origin_from_req(req)? {
Some(origin) => {
index_scheduler.set_task_network(task.uid, TaskNetwork::Origin { origin })?
) -> Result<Task, MeilisearchHttpError>
where
T: serde::Serialize,
F: FnMut(&str, &Remote, &mut T),
{
if let TaskNetwork::Remotes { remote_tasks, network_version: _ } = &mut task_network {
let this = network
.local
.as_deref()
.expect("inconsistent `network.sharding` and `network.self`")
.to_owned();
let content_type = match &body {
// for file bodies, force x-ndjson
Body::NdJsonPayload(_) => Some(b"application/x-ndjson".as_slice()),
// otherwise get content type from request
_ => req.headers().get(CONTENT_TYPE).map(|h| h.as_bytes()),
};
let mut in_flight_remote_queries = BTreeMap::new();
let client = reqwest::ClientBuilder::new()
.connect_timeout(std::time::Duration::from_secs(3))
.build()
.unwrap();
let method = from_old_http_method(req.method());
// send payload to all remotes
for (body, (node_name, node)) in body
.into_bytes_iter(network.remotes.into_iter().filter(|(name, _)| name.as_str() != this))
.map_err(|err| {
MeilisearchHttpError::from_milli(err, index_uid.map(ToOwned::to_owned))
})?
{
let client = client.clone();
let api_key = node.write_api_key;
let this = this.clone();
let method = method.clone();
let path_and_query = req.uri().path_and_query().map(|paq| paq.as_str()).unwrap_or("/");
in_flight_remote_queries.insert(
node_name,
tokio::spawn({
let url = format!("{}{}", node.url, path_and_query);
let url_encoded_this = urlencoding::encode(&this).into_owned();
let url_encoded_task_uid = task.uid.to_string(); // it's url encoded i promize
let content_type = content_type.map(|b| b.to_owned());
let backoff = backoff::ExponentialBackoffBuilder::new()
.with_max_elapsed_time(Some(std::time::Duration::from_secs(25)))
.build();
backoff::future::retry(backoff, move || {
let url = url.clone();
let client = client.clone();
let url_encoded_this = url_encoded_this.clone();
let url_encoded_task_uid = url_encoded_task_uid.clone();
let content_type = content_type.clone();
let body = body.clone();
let api_key = api_key.clone();
let method = method.clone();
async move {
try_proxy(
method,
&url,
content_type.as_deref(),
api_key.as_deref(),
&client,
&url_encoded_this,
&url_encoded_task_uid,
body,
)
.await
}
})
}),
);
}
None => {
let this = network
.local
.as_deref()
.expect("inconsistent `network.sharding` and `network.self`")
.to_owned();
let content_type = match &body {
// for file bodies, force x-ndjson
Body::NdJsonPayload(_) => Some(b"application/x-ndjson".as_slice()),
// otherwise get content type from request
_ => req.headers().get(CONTENT_TYPE).map(|h| h.as_bytes()),
};
// wait for all in-flight queries to finish and collect their results
for (node_name, handle) in in_flight_remote_queries {
match handle.await {
Ok(Ok(res)) => {
let task_uid = res.task_uid;
let body = match body {
Body::NdJsonPayload(file) => Some(Bytes::from_owner(unsafe {
memmap2::Mmap::map(&file).map_err(|err| {
MeilisearchHttpError::from_milli(err.into(), Some(index_uid.to_owned()))
})?
})),
Body::Inline(payload) => {
Some(Bytes::copy_from_slice(&serde_json::to_vec(&payload).unwrap()))
remote_tasks.insert(node_name, Ok(task_uid).into());
}
Body::None => None,
};
let mut in_flight_remote_queries = BTreeMap::new();
let client = reqwest::ClientBuilder::new()
.connect_timeout(std::time::Duration::from_secs(3))
.build()
.unwrap();
let method = from_old_http_method(req.method());
// send payload to all remotes
for (node_name, node) in
network.remotes.into_iter().filter(|(name, _)| name.as_str() != this)
{
let body = body.clone();
let client = client.clone();
let api_key = node.write_api_key;
let this = this.clone();
let method = method.clone();
let path_and_query =
req.uri().path_and_query().map(|paq| paq.as_str()).unwrap_or("/");
in_flight_remote_queries.insert(
node_name,
tokio::spawn({
let url = format!("{}{}", node.url, path_and_query);
let url_encoded_this = urlencoding::encode(&this).into_owned();
let url_encoded_task_uid = task.uid.to_string(); // it's url encoded i promize
let content_type = content_type.map(|b| b.to_owned());
let backoff = backoff::ExponentialBackoffBuilder::new()
.with_max_elapsed_time(Some(std::time::Duration::from_secs(25)))
.build();
backoff::future::retry(backoff, move || {
let url = url.clone();
let client = client.clone();
let url_encoded_this = url_encoded_this.clone();
let url_encoded_task_uid = url_encoded_task_uid.clone();
let content_type = content_type.clone();
let body = body.clone();
let api_key = api_key.clone();
let method = method.clone();
async move {
try_proxy(
method,
&url,
content_type.as_deref(),
api_key.as_deref(),
&client,
&url_encoded_this,
&url_encoded_task_uid,
body,
)
.await
}
})
}),
);
}
// wait for all in-flight queries to finish and collect their results
let mut remote_tasks: BTreeMap<String, RemoteTask> = BTreeMap::new();
for (node_name, handle) in in_flight_remote_queries {
match handle.await {
Ok(Ok(res)) => {
let task_uid = res.task_uid;
remote_tasks.insert(node_name, Ok(task_uid).into());
}
Ok(Err(error)) => {
remote_tasks.insert(node_name, Err(error.as_response_error()).into());
}
Err(panic) => match panic.try_into_panic() {
Ok(panic) => {
let msg = match panic.downcast_ref::<&'static str>() {
Some(s) => *s,
None => match panic.downcast_ref::<String>() {
Some(s) => &s[..],
None => "Box<dyn Any>",
},
};
remote_tasks.insert(
node_name,
Err(ResponseError::from_msg(
msg.to_string(),
meilisearch_types::error::Code::Internal,
))
.into(),
);
}
Err(_) => {
tracing::error!("proxy task was unexpectedly cancelled")
}
},
Ok(Err(error)) => {
remote_tasks.insert(node_name, Err(error.as_response_error()).into());
}
Err(panic) => match panic.try_into_panic() {
Ok(panic) => {
let msg = match panic.downcast_ref::<&'static str>() {
Some(s) => *s,
None => match panic.downcast_ref::<String>() {
Some(s) => &s[..],
None => "Box<dyn Any>",
},
};
remote_tasks.insert(
node_name,
Err(ResponseError::from_msg(
msg.to_string(),
meilisearch_types::error::Code::Internal,
))
.into(),
);
}
Err(_) => {
tracing::error!("proxy task was unexpectedly cancelled")
}
},
}
// edit details to contain the return values from the remotes
index_scheduler.set_task_network(task.uid, TaskNetwork::Remotes { remote_tasks })?;
}
}
Ok(())
Ok(index_scheduler.set_task_network(task.uid, task_network)?)
}
fn from_old_http_method(method: &actix_http::Method) -> reqwest::Method {
@@ -375,25 +475,23 @@ mod error {
}
}
pub const PROXY_ORIGIN_REMOTE_HEADER: &str = "Meili-Proxy-Origin-Remote";
pub const PROXY_ORIGIN_TASK_UID_HEADER: &str = "Meili-Proxy-Origin-TaskUid";
pub fn origin_from_req(req: &HttpRequest) -> Result<Option<Origin>, MeilisearchHttpError> {
let (remote_name, task_uid) = match (
let (remote_name, task_uid, network_version) = match (
req.headers().get(PROXY_ORIGIN_REMOTE_HEADER),
req.headers().get(PROXY_ORIGIN_TASK_UID_HEADER),
req.headers().get(PROXY_ORIGIN_NETWORK_VERSION_HEADER),
) {
(None, None) => return Ok(None),
(None, Some(_)) => {
(None, None, _) => return Ok(None),
(None, Some(_), _) => {
return Err(MeilisearchHttpError::InconsistentOriginHeaders { is_remote_missing: true })
}
(Some(_), None) => {
(Some(_), None, _) => {
return Err(MeilisearchHttpError::InconsistentOriginHeaders {
is_remote_missing: false,
})
}
(Some(remote_name), Some(task_uid)) => (
urlencoding::decode(remote_name.to_str().map_err(|err| {
(Some(remote_name), Some(task_uid), network_version) => {
let remote_name = urlencoding::decode(remote_name.to_str().map_err(|err| {
MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_ORIGIN_REMOTE_HEADER,
msg: format!("while parsing remote name as UTF-8: {err}"),
@@ -402,8 +500,8 @@ pub fn origin_from_req(req: &HttpRequest) -> Result<Option<Origin>, MeilisearchH
.map_err(|err| MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_ORIGIN_REMOTE_HEADER,
msg: format!("while URL-decoding remote name: {err}"),
})?,
urlencoding::decode(task_uid.to_str().map_err(|err| {
})?;
let task_uid = urlencoding::decode(task_uid.to_str().map_err(|err| {
MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_ORIGIN_TASK_UID_HEADER,
msg: format!("while parsing task UID as UTF-8: {err}"),
@@ -412,15 +510,126 @@ pub fn origin_from_req(req: &HttpRequest) -> Result<Option<Origin>, MeilisearchH
.map_err(|err| MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_ORIGIN_TASK_UID_HEADER,
msg: format!("while URL-decoding task UID: {err}"),
})?,
),
})?;
let network_version = match network_version {
Some(network_version) => {
urlencoding::decode(network_version.to_str().map_err(|err| {
MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_ORIGIN_NETWORK_VERSION_HEADER,
msg: format!("while parsing network version as UTF-8: {err}"),
}
})?)
.map_err(|err| {
MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_ORIGIN_NETWORK_VERSION_HEADER,
msg: format!("while URL-decoding network version: {err}"),
}
})?
}
None => Cow::Borrowed("0"),
};
(remote_name, task_uid, network_version)
}
};
let task_uid: usize =
let task_uid: u32 =
task_uid.parse().map_err(|err| MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_ORIGIN_TASK_UID_HEADER,
msg: format!("while parsing the task UID as an integer: {err}"),
})?;
Ok(Some(Origin { remote_name: remote_name.into_owned(), task_uid }))
let network_version: u128 =
network_version.parse().map_err(|err| MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_ORIGIN_NETWORK_VERSION_HEADER,
msg: format!("while parsing the network version as a u128: {err}"),
})?;
let network_version = uuid::Uuid::from_u128(network_version);
Ok(Some(Origin { remote_name: remote_name.into_owned(), task_uid, network_version }))
}
pub fn import_from_req(req: &HttpRequest) -> Result<Option<ImportData>, MeilisearchHttpError> {
let (remote_name, index_name, last_documents, documents) = match (
req.headers().get(PROXY_IMPORT_REMOTE_HEADER),
req.headers().get(PROXY_IMPORT_INDEX_HEADER),
req.headers().get(PROXY_IMPORT_FIRST_DOC_HEADER),
req.headers().get(PROXY_IMPORT_DOCS_HEADER),
) {
(None, None, None, None) => return Ok(None),
(Some(remote_name), Some(index_name), Some(last_documents), Some(documents)) => {
let remote_name = urlencoding::decode(remote_name.to_str().map_err(|err| {
MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_IMPORT_REMOTE_HEADER,
msg: format!("while parsing import remote name as UTF-8: {err}"),
}
})?)
.map_err(|err| MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_IMPORT_REMOTE_HEADER,
msg: format!("while URL-decoding import remote name: {err}"),
})?;
let index_name = urlencoding::decode(index_name.to_str().map_err(|err| {
MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_IMPORT_INDEX_HEADER,
msg: format!("while parsing import index name as UTF-8: {err}"),
}
})?)
.map_err(|err| MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_IMPORT_INDEX_HEADER,
msg: format!("while URL-decoding import index name: {err}"),
})?;
let last_documents = urlencoding::decode(last_documents.to_str().map_err(|err| {
MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_IMPORT_FIRST_DOC_HEADER,
msg: format!("while parsing last documents as UTF-8: {err}"),
}
})?)
.map_err(|err| MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_IMPORT_FIRST_DOC_HEADER,
msg: format!("while URL-decoding last documents: {err}"),
})?;
let documents = urlencoding::decode(documents.to_str().map_err(|err| {
MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_IMPORT_DOCS_HEADER,
msg: format!("while parsing documents as UTF-8: {err}"),
}
})?)
.map_err(|err| MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_IMPORT_DOCS_HEADER,
msg: format!("while URL-decoding documents: {err}"),
})?;
(remote_name, index_name, last_documents, documents)
}
// catch-all pattern that has to contain an inconsistency since we already matched (None, None, None) and (Some, Some, Some)
(remote_name, index_name, last_documents, documents) => {
return Err(MeilisearchHttpError::InconsistentImportHeaders {
is_remote_missing: remote_name.is_none(),
is_index_missing: index_name.is_none(),
is_last_docs_missing: last_documents.is_none(),
is_docs_missing: documents.is_none(),
})
}
};
let first_docid: DocumentId =
last_documents.parse().map_err(|err| MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_IMPORT_FIRST_DOC_HEADER,
msg: format!("while parsing the last documents as an integer: {err}"),
})?;
let document_count: u64 =
documents.parse().map_err(|err| MeilisearchHttpError::InvalidHeaderValue {
header_name: PROXY_IMPORT_DOCS_HEADER,
msg: format!("while parsing the documents as an integer: {err}"),
})?;
Ok(Some(ImportData {
remote_name: remote_name.to_string(),
index_name: index_name.to_string(),
first_docid,
document_count,
}))
}

6
crates/meilisearch/src/routes/indexes/mod.rs
View File

@@ -30,7 +30,7 @@ use crate::Opt;
pub mod compact;
pub mod documents;
mod enterprise_edition;
pub mod enterprise_edition;
pub mod facet_search;
pub mod search;
mod search_analytics;
@@ -41,7 +41,9 @@ mod settings_analytics;
pub mod similar;
mod similar_analytics;
pub use enterprise_edition::proxy::{PROXY_ORIGIN_REMOTE_HEADER, PROXY_ORIGIN_TASK_UID_HEADER};
pub use meilisearch_types::tasks::enterprise_edition::network::headers::{
PROXY_ORIGIN_REMOTE_HEADER, PROXY_ORIGIN_TASK_UID_HEADER,
};
#[derive(OpenApi)]
#[openapi(

258
crates/meilisearch/src/routes/network.rs
View File

@@ -9,20 +9,27 @@ use itertools::{EitherOrBoth, Itertools};
use meilisearch_types::deserr::DeserrJsonError;
use meilisearch_types::enterprise_edition::network::{Network as DbNetwork, Remote as DbRemote};
use meilisearch_types::error::deserr_codes::{
InvalidNetworkRemotes, InvalidNetworkSearchApiKey, InvalidNetworkSelf, InvalidNetworkSharding,
InvalidNetworkLeader, InvalidNetworkRemotes, InvalidNetworkSearchApiKey, InvalidNetworkSelf,
InvalidNetworkUrl, InvalidNetworkWriteApiKey,
};
use meilisearch_types::error::ResponseError;
use meilisearch_types::keys::actions;
use meilisearch_types::milli::update::Setting;
use meilisearch_types::tasks::enterprise_edition::network::{
NetworkTopologyChange, Origin, TaskNetwork,
};
use meilisearch_types::tasks::KindWithContent;
use serde::Serialize;
use tracing::debug;
use utoipa::{OpenApi, ToSchema};
use crate::analytics::{Aggregate, Analytics};
use crate::error::MeilisearchHttpError;
use crate::extractors::authentication::policies::ActionPolicy;
use crate::extractors::authentication::GuardedData;
use crate::extractors::sequential_extractor::SeqHandler;
use crate::routes::indexes::enterprise_edition::proxy::{proxy, Body};
use crate::routes::SummarizedTaskView;
#[derive(OpenApi)]
#[openapi(
@@ -83,7 +90,7 @@ async fn get_network(
Ok(HttpResponse::Ok().json(network))
}
#[derive(Debug, Deserr, ToSchema, Serialize)]
#[derive(Clone, Debug, Deserr, ToSchema, Serialize)]
#[deserr(error = DeserrJsonError<InvalidNetworkRemotes>, rename_all = camelCase, deny_unknown_fields)]
#[serde(rename_all = "camelCase")]
#[schema(rename_all = "camelCase")]
@@ -106,12 +113,19 @@ pub struct Remote {
pub write_api_key: Setting<String>,
}
#[derive(Debug, Deserr, ToSchema, Serialize)]
#[derive(Clone, Debug, Deserr, ToSchema, Serialize)]
#[deserr(error = DeserrJsonError, rename_all = camelCase, deny_unknown_fields)]
#[serde(rename_all = "camelCase")]
#[schema(rename_all = "camelCase")]
pub struct Network {
#[schema(value_type = Option<BTreeMap<String, Remote>>, example = json!("http://localhost:7700"))]
#[schema(value_type = Option<BTreeMap<String, Remote>>, example = json!({
"ms-00": {
"url": "http://localhost:7700"
},
"ms-01": {
"url": "http://localhost:7701"
}
}))]
#[deserr(default, error = DeserrJsonError<InvalidNetworkRemotes>)]
#[serde(default)]
pub remotes: Setting<BTreeMap<String, Option<Remote>>>,
@@ -119,10 +133,21 @@ pub struct Network {
#[serde(default, rename = "self")]
#[deserr(default, rename = "self", error = DeserrJsonError<InvalidNetworkSelf>)]
pub local: Setting<String>,
#[schema(value_type = Option<bool>, example = json!(true))]
#[schema(value_type = Option<String>, example = json!("ms-00"))]
#[serde(default)]
#[deserr(default, error = DeserrJsonError<InvalidNetworkSharding>)]
pub sharding: Setting<bool>,
#[deserr(default, error = DeserrJsonError<InvalidNetworkLeader>)]
pub leader: Setting<String>,
#[schema(value_type = Option<BTreeMap<String, Remote>>, example = json!({
"ms-00": {
"url": "http://localhost:7700"
},
"ms-01": {
"url": "http://localhost:7701"
}
}))]
#[deserr(default, error = DeserrJsonError<InvalidNetworkRemotes>)]
#[serde(default)]
pub previous_remotes: Setting<BTreeMap<String, Option<Remote>>>,
}
impl Remote {
@@ -207,29 +232,203 @@ async fn patch_network(
) -> Result<HttpResponse, ResponseError> {
index_scheduler.features().check_network("Using the /network route")?;
match crate::routes::indexes::enterprise_edition::proxy::origin_from_req(&req)? {
Some(origin) => {
patch_network_with_origin(index_scheduler, new_network, req, origin, analytics).await
}
None => patch_network_without_origin(index_scheduler, new_network, req, analytics).await,
}
}
async fn patch_network_without_origin(
index_scheduler: GuardedData<ActionPolicy<{ actions::NETWORK_UPDATE }>, Data<IndexScheduler>>,
new_network: AwebJson<Network, DeserrJsonError>,
req: HttpRequest,
analytics: Data<Analytics>,
) -> Result<HttpResponse, ResponseError> {
let new_network = new_network.0;
let old_network = index_scheduler.network();
debug!(parameters = ?new_network, "Patch network");
if !matches!(new_network.previous_remotes, Setting::NotSet) {
return Err(MeilisearchHttpError::UnexpectedNetworkPreviousRemotes.into());
}
let merged_network = merge_networks(old_network.clone(), new_network)?;
index_scheduler.put_network(merged_network.clone())?;
analytics.publish(
PatchNetworkAnalytics {
network_size: merged_network.remotes.len(),
network_has_self: merged_network.local.is_some(),
},
&req,
);
/// TODO: spawn task only if necessary
let network_topology_change =
NetworkTopologyChange::new(old_network.clone(), merged_network.clone());
let task = KindWithContent::NetworkTopologyChange(network_topology_change);
let task = {
let index_scheduler = index_scheduler.clone();
tokio::task::spawn_blocking(move || index_scheduler.register(task, None, false)).await??
};
let mut proxied_network = Network {
remotes: Setting::Set(to_settings_remotes(&merged_network.remotes)),
local: Setting::NotSet,
leader: Setting::some_or_not_set(merged_network.leader.clone()),
previous_remotes: Setting::Set(to_settings_remotes(&old_network.remotes)),
};
let mut deleted_network = old_network;
let deleted_remotes = &mut deleted_network.remotes;
deleted_remotes.retain(|node, _| !merged_network.remotes.contains_key(node));
// proxy network change to the remaining remotes.
let updated_task = proxy(
&index_scheduler,
None,
&req,
TaskNetwork::Remotes {
remote_tasks: Default::default(),
network_version: merged_network.version,
},
merged_network,
Body::generated(proxied_network.clone(), |name, _remote, network| {
network.local = Setting::Set(name.to_string());
}),
&task,
)
.await?;
// unwrap: network was set by `proxy`
let task_network = updated_task.network.unwrap();
proxied_network.previous_remotes = Setting::NotSet;
// proxy network change to the deleted remotes
proxy(
&index_scheduler,
None,
&req,
task_network,
deleted_network,
Body::generated(proxied_network.clone(), |_name, _remote, network| {
network.local = Setting::Reset;
}),
&task,
)
.await?;
let task: SummarizedTaskView = task.into();
debug!("returns: {:?}", task);
Ok(HttpResponse::Accepted().json(task))
}
async fn patch_network_with_origin(
index_scheduler: GuardedData<ActionPolicy<{ actions::NETWORK_UPDATE }>, Data<IndexScheduler>>,
merged_network: AwebJson<Network, DeserrJsonError>,
req: HttpRequest,
origin: Origin,
analytics: Data<Analytics>,
) -> Result<HttpResponse, ResponseError> {
let merged_network = merged_network.into_inner();
debug!(parameters = ?merged_network, ?origin, "Patch network");
let mut remotes = BTreeMap::new();
let mut old_network = index_scheduler.network();
for (name, remote) in merged_network.remotes.set().into_iter().flat_map(|x| x.into_iter()) {
let Some(remote) = remote else { continue };
let remote = remote.try_into_db_node(&name)?;
remotes.insert(name, remote);
}
let mut previous_remotes = BTreeMap::new();
for (name, remote) in
merged_network.previous_remotes.set().into_iter().flat_map(|x| x.into_iter())
{
let Some(remote) = remote else {
continue;
};
let remote = remote.try_into_db_node(&name)?;
previous_remotes.insert(name, remote);
}
old_network.remotes = previous_remotes;
let new_network = DbNetwork {
local: merged_network.local.set(),
remotes,
leader: merged_network.leader.set(),
version: origin.network_version,
};
index_scheduler.put_network(new_network.clone())?;
analytics.publish(
PatchNetworkAnalytics {
network_size: new_network.remotes.len(),
network_has_self: new_network.local.is_some(),
},
&req,
);
/// TODO: spawn task only if necessary
let network_topology_change = NetworkTopologyChange::new(old_network, new_network);
let task = KindWithContent::NetworkTopologyChange(network_topology_change);
let task = {
let index_scheduler = index_scheduler.clone();
tokio::task::spawn_blocking(move || index_scheduler.register(task, None, false)).await??
};
index_scheduler.set_task_network(task.uid, TaskNetwork::Origin { origin })?;
let task: SummarizedTaskView = task.into();
debug!("returns: {:?}", task);
Ok(HttpResponse::Accepted().json(task))
}
fn to_settings_remotes(
db_remotes: &BTreeMap<String, DbRemote>,
) -> BTreeMap<String, Option<Remote>> {
db_remotes
.iter()
.map(|(name, remote)| {
(
name.clone(),
Some(Remote {
url: Setting::Set(remote.url.clone()),
search_api_key: Setting::some_or_not_set(remote.search_api_key.clone()),
write_api_key: Setting::some_or_not_set(remote.write_api_key.clone()),
}),
)
})
.collect()
}
fn merge_networks(
old_network: DbNetwork,
new_network: Network,
) -> Result<DbNetwork, ResponseError> {
let merged_self = match new_network.local {
Setting::Set(new_self) => Some(new_self),
Setting::Reset => None,
Setting::NotSet => old_network.local,
};
let merged_sharding = match new_network.sharding {
Setting::Set(new_sharding) => new_sharding,
Setting::Reset => false,
Setting::NotSet => old_network.sharding,
let merged_leader = match new_network.leader {
Setting::Set(new_leader) => Some(new_leader),
Setting::Reset => None,
Setting::NotSet => old_network.leader,
};
if merged_sharding && merged_self.is_none() {
return Err(ResponseError::from_msg(
"`.sharding`: enabling the sharding requires `.self` to be set\n - Hint: Disable `sharding` or set `self` to a value.".into(),
meilisearch_types::error::Code::InvalidNetworkSharding,
));
match (merged_leader.as_deref(), merged_self.as_deref()) {
// 1. Always allowed if there is no leader
(None, _) => (),
// 2. Allowed if the leader is self
(Some(leader), Some(this)) if leader == this => (),
// 3. Any other change is disallowed
(Some(leader), _) => {
return Err(MeilisearchHttpError::NotLeader { leader: leader.to_string() }.into())
}
}
let new_version = uuid::Uuid::now_v7();
let merged_remotes = match new_network.remotes {
Setting::Set(new_remotes) => {
let mut merged_remotes = BTreeMap::new();
@@ -301,18 +500,11 @@ async fn patch_network(
Setting::Reset => BTreeMap::new(),
Setting::NotSet => old_network.remotes,
};
analytics.publish(
PatchNetworkAnalytics {
network_size: merged_remotes.len(),
network_has_self: merged_self.is_some(),
},
&req,
);
let merged_network =
DbNetwork { local: merged_self, remotes: merged_remotes, sharding: merged_sharding };
index_scheduler.put_network(merged_network.clone())?;
debug!(returns = ?merged_network, "Patch network");
Ok(HttpResponse::Ok().json(merged_network))
let merged_network = DbNetwork {
local: merged_self,
remotes: merged_remotes,
leader: merged_leader,
version: new_version,
};
Ok(merged_network)
}

32
crates/milli/src/update/new/indexer/enterprise_edition/sharding.rs
View File

@@ -5,18 +5,36 @@
use std::hash::{BuildHasher as _, BuildHasherDefault};
pub struct Shards {
pub own: Vec<String>,
pub others: Vec<String>,
pub struct Shards(pub Vec<Shard>);
pub struct Shard {
pub is_own: bool,
pub name: String,
}
impl Shards {
pub fn from_remotes_local<'a>(
remotes: impl IntoIterator<Item = &'a str>,
local: Option<&str>,
) -> Self {
Shards(
remotes
.into_iter()
.map(|name| Shard { is_own: Some(name) == local, name: name.to_owned() })
.collect(),
)
}
pub fn must_process(&self, docid: &str) -> bool {
self.processing_shard(docid).map(|shard| shard.is_own).unwrap_or_default()
}
pub fn processing_shard<'a>(&'a self, docid: &str) -> Option<&'a Shard> {
let hasher = BuildHasherDefault::<twox_hash::XxHash3_64>::new();
let to_hash = |shard: &String| hasher.hash_one((shard, docid));
let to_hash = |shard: &'a Shard| (shard, hasher.hash_one((&shard.name, docid)));
let max_hash = self.others.iter().map(to_hash).max().unwrap_or_default();
self.own.iter().map(to_hash).any(|hash| hash > max_hash)
let shard =
self.0.iter().map(to_hash).max_by_key(|(_, hash)| *hash).map(|(shard, _)| shard);
shard
}
}