reintroduce anyhow

This commit is contained in:
marin postma 2021-06-15 17:39:07 +02:00
parent 439db1aae0
commit 02277ec2cf
No known key found for this signature in database
GPG Key ID: 6088B7721C3E39F9
36 changed files with 110 additions and 154 deletions

View File

@ -27,7 +27,7 @@ actix-http = { version = "=3.0.0-beta.6" }
actix-service = "2.0.0" actix-service = "2.0.0"
actix-web = { version = "=4.0.0-beta.6", features = ["rustls"] } actix-web = { version = "=4.0.0-beta.6", features = ["rustls"] }
actix-web-static-files = { git = "https://github.com/MarinPostma/actix-web-static-files.git", rev = "6db8c3e", optional = true } actix-web-static-files = { git = "https://github.com/MarinPostma/actix-web-static-files.git", rev = "6db8c3e", optional = true }
#anyhow = "1.0.36" anyhow = "1.0.36"
async-stream = "0.3.0" async-stream = "0.3.0"
async-trait = "0.1.42" async-trait = "0.1.42"
arc-swap = "1.2.0" arc-swap = "1.2.0"

View File

@ -5,8 +5,7 @@ use sha2::Digest;
use crate::index::{Checked, Settings}; use crate::index::{Checked, Settings};
use crate::index_controller::{ use crate::index_controller::{
DumpInfo, IndexController, IndexMetadata, IndexSettings, IndexStats, Stats, error::Result, DumpInfo, IndexController, IndexMetadata, IndexSettings, IndexStats, Stats,
error::Result
}; };
use crate::option::Opt; use crate::option::Opt;
@ -57,7 +56,7 @@ impl ApiKeys {
} }
impl Data { impl Data {
pub fn new(options: Opt) -> std::result::Result<Data, Box<dyn std::error::Error>> { pub fn new(options: Opt) -> anyhow::Result<Data> {
let path = options.db_path.clone(); let path = options.db_path.clone();
let index_controller = IndexController::new(&path, &options)?; let index_controller = IndexController::new(&path, &options)?;

View File

@ -5,11 +5,7 @@ use crate::index::{SearchQuery, SearchResult};
use crate::index_controller::error::Result; use crate::index_controller::error::Result;
impl Data { impl Data {
pub async fn search( pub async fn search(&self, index: String, search_query: SearchQuery) -> Result<SearchResult> {
&self,
index: String,
search_query: SearchQuery,
) -> Result<SearchResult> {
self.index_controller.search(index, search_query).await self.index_controller.search(index, search_query).await
} }

View File

@ -3,7 +3,7 @@ use milli::update::{IndexDocumentsMethod, UpdateFormat};
use super::Data; use super::Data;
use crate::index::{Checked, Settings}; use crate::index::{Checked, Settings};
use crate::index_controller::{IndexMetadata, IndexSettings, UpdateStatus, error::Result}; use crate::index_controller::{error::Result, IndexMetadata, IndexSettings, UpdateStatus};
impl Data { impl Data {
pub async fn add_documents( pub async fn add_documents(

View File

@ -22,7 +22,7 @@ pub enum AuthenticationError {
impl ErrorCode for AuthenticationError { impl ErrorCode for AuthenticationError {
fn error_code(&self) -> Code { fn error_code(&self) -> Code {
match self { match self {
AuthenticationError ::MissingAuthorizationHeader => Code::MissingAuthorizationHeader, AuthenticationError::MissingAuthorizationHeader => Code::MissingAuthorizationHeader,
AuthenticationError::InvalidToken(_) => Code::InvalidToken, AuthenticationError::InvalidToken(_) => Code::InvalidToken,
} }
} }
@ -62,11 +62,7 @@ macro_rules! response_error {
}; };
} }
response_error!( response_error!(IndexControllerError, AuthenticationError);
IndexControllerError,
AuthenticationError
);
impl Serialize for ResponseError { impl Serialize for ResponseError {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
@ -114,7 +110,8 @@ impl<E: Error> ErrorCode for PayloadError<E> {
} }
impl<E> From<PayloadError<E>> for ResponseError impl<E> From<PayloadError<E>> for ResponseError
where E: Error + Sync + Send + 'static where
E: Error + Sync + Send + 'static,
{ {
fn from(other: PayloadError<E>) -> Self { fn from(other: PayloadError<E>) -> Self {
ResponseError { ResponseError {
@ -124,7 +121,8 @@ where E: Error + Sync + Send + 'static
} }
pub fn payload_error_handler<E>(err: E) -> ResponseError pub fn payload_error_handler<E>(err: E) -> ResponseError
where E: Error + Sync + Send + 'static where
E: Error + Sync + Send + 'static,
{ {
let error = PayloadError(err); let error = PayloadError(err);
error.into() error.into()

View File

@ -9,7 +9,7 @@ use futures::future::{ok, Future, Ready};
use futures::ready; use futures::ready;
use pin_project::pin_project; use pin_project::pin_project;
use crate::error::{ResponseError, AuthenticationError}; use crate::error::{AuthenticationError, ResponseError};
use crate::Data; use crate::Data;
#[derive(Clone, Copy)] #[derive(Clone, Copy)]
@ -117,7 +117,8 @@ where
AuthProj::NoHeader(req) => { AuthProj::NoHeader(req) => {
match req.take() { match req.take() {
Some(req) => { Some(req) => {
let response = ResponseError::from(AuthenticationError::MissingAuthorizationHeader); let response =
ResponseError::from(AuthenticationError::MissingAuthorizationHeader);
let response = response.error_response(); let response = response.error_response();
let response = req.into_response(response); let response = req.into_response(response);
Poll::Ready(Ok(response)) Poll::Ready(Ok(response))
@ -134,7 +135,8 @@ where
.get("X-Meili-API-Key") .get("X-Meili-API-Key")
.map(|h| h.to_str().map(String::from).unwrap_or_default()) .map(|h| h.to_str().map(String::from).unwrap_or_default())
.unwrap_or_default(); .unwrap_or_default();
let response = ResponseError::from(AuthenticationError::InvalidToken(bad_token)); let response =
ResponseError::from(AuthenticationError::InvalidToken(bad_token));
let response = response.error_response(); let response = response.error_response();
let response = req.into_response(response); let response = req.into_response(response);
Poll::Ready(Ok(response)) Poll::Ready(Ok(response))

View File

@ -5,7 +5,7 @@ use std::path::Path;
use flate2::{read::GzDecoder, write::GzEncoder, Compression}; use flate2::{read::GzDecoder, write::GzEncoder, Compression};
use tar::{Archive, Builder}; use tar::{Archive, Builder};
pub fn to_tar_gz(src: impl AsRef<Path>, dest: impl AsRef<Path>) -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> { pub fn to_tar_gz(src: impl AsRef<Path>, dest: impl AsRef<Path>) -> anyhow::Result<()> {
let mut f = File::create(dest)?; let mut f = File::create(dest)?;
let gz_encoder = GzEncoder::new(&mut f, Compression::default()); let gz_encoder = GzEncoder::new(&mut f, Compression::default());
let mut tar_encoder = Builder::new(gz_encoder); let mut tar_encoder = Builder::new(gz_encoder);
@ -16,7 +16,7 @@ pub fn to_tar_gz(src: impl AsRef<Path>, dest: impl AsRef<Path>) -> Result<(), Bo
Ok(()) Ok(())
} }
pub fn from_tar_gz(src: impl AsRef<Path>, dest: impl AsRef<Path>) -> Result<(), Box<dyn std::error::Error>> { pub fn from_tar_gz(src: impl AsRef<Path>, dest: impl AsRef<Path>) -> anyhow::Result<()> {
let f = File::open(&src)?; let f = File::open(&src)?;
let gz = GzDecoder::new(f); let gz = GzDecoder::new(f);
let mut ar = Archive::new(gz); let mut ar = Archive::new(gz);

View File

@ -3,6 +3,7 @@ use std::io::{BufRead, BufReader, Write};
use std::path::Path; use std::path::Path;
use std::sync::Arc; use std::sync::Arc;
use anyhow::Context;
use heed::RoTxn; use heed::RoTxn;
use indexmap::IndexMap; use indexmap::IndexMap;
use milli::update::{IndexDocumentsMethod, UpdateFormat::JsonStream}; use milli::update::{IndexDocumentsMethod, UpdateFormat::JsonStream};
@ -10,8 +11,8 @@ use serde::{Deserialize, Serialize};
use crate::option::IndexerOpts; use crate::option::IndexerOpts;
use super::{update_handler::UpdateHandler, Index, Settings, Unchecked};
use super::error::{IndexError, Result}; use super::error::{IndexError, Result};
use super::{update_handler::UpdateHandler, Index, Settings, Unchecked};
#[derive(Serialize, Deserialize)] #[derive(Serialize, Deserialize)]
struct DumpMeta { struct DumpMeta {
@ -37,7 +38,8 @@ impl Index {
let document_file_path = path.as_ref().join(DATA_FILE_NAME); let document_file_path = path.as_ref().join(DATA_FILE_NAME);
let mut document_file = File::create(&document_file_path)?; let mut document_file = File::create(&document_file_path)?;
let documents = self.all_documents(txn) let documents = self
.all_documents(txn)
.map_err(|e| IndexError::Internal(e.into()))?; .map_err(|e| IndexError::Internal(e.into()))?;
let fields_ids_map = self.fields_ids_map(txn)?; let fields_ids_map = self.fields_ids_map(txn)?;
@ -82,13 +84,12 @@ impl Index {
dst: impl AsRef<Path>, dst: impl AsRef<Path>,
size: usize, size: usize,
indexing_options: &IndexerOpts, indexing_options: &IndexerOpts,
) -> std::result::Result<(), Box<dyn std::error::Error>> { ) -> anyhow::Result<()> {
let dir_name = src let dir_name = src
.as_ref() .as_ref()
.file_name() .file_name()
// TODO: remove .with_context(|| format!("invalid dump index: {}", src.as_ref().display()))?;
//.with_context(|| format!("invalid dump index: {}", src.as_ref().display()))?;
.unwrap();
let dst_dir_path = dst.as_ref().join("indexes").join(dir_name); let dst_dir_path = dst.as_ref().join("indexes").join(dir_name);
create_dir_all(&dst_dir_path)?; create_dir_all(&dst_dir_path)?;

View File

@ -27,12 +27,7 @@ macro_rules! internal_error {
} }
} }
internal_error!( internal_error!(std::io::Error, heed::Error, fst::Error, serde_json::Error);
std::io::Error,
heed::Error,
fst::Error,
serde_json::Error
);
impl ErrorCode for IndexError { impl ErrorCode for IndexError {
fn error_code(&self) -> Code { fn error_code(&self) -> Code {
@ -47,14 +42,13 @@ impl ErrorCode for IndexError {
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum FacetError { pub enum FacetError {
#[error("Invalid facet expression, expected {}, found: {1}", .0.join(", "))] #[error("Invalid facet expression, expected {}, found: {1}", .0.join(", "))]
InvalidExpression(&'static [&'static str], Value) InvalidExpression(&'static [&'static str], Value),
} }
impl ErrorCode for FacetError { impl ErrorCode for FacetError {
fn error_code(&self) -> Code { fn error_code(&self) -> Code {
match self { match self {
FacetError::InvalidExpression(_, _) => Code::Facet, FacetError::InvalidExpression(_, _) => Code::Facet,
} }
} }
} }

View File

@ -21,7 +21,7 @@ pub struct UpdateHandler {
} }
impl UpdateHandler { impl UpdateHandler {
pub fn new(opt: &IndexerOpts) -> std::result::Result<Self, Box<dyn std::error::Error>> { pub fn new(opt: &IndexerOpts) -> anyhow::Result<Self> {
let thread_pool = rayon::ThreadPoolBuilder::new() let thread_pool = rayon::ThreadPoolBuilder::new()
.num_threads(opt.indexing_jobs.unwrap_or(0)) .num_threads(opt.indexing_jobs.unwrap_or(0))
.build()?; .build()?;

View File

@ -1,4 +1,4 @@
use std::collections::{BTreeSet, BTreeMap, HashSet}; use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::io; use std::io;
use std::marker::PhantomData; use std::marker::PhantomData;
use std::num::NonZeroUsize; use std::num::NonZeroUsize;
@ -308,10 +308,9 @@ impl Index {
} }
} }
builder.execute(|indexing_step, update_id| { builder
info!("update {}: {:?}", update_id, indexing_step) .execute(|indexing_step, update_id| info!("update {}: {:?}", update_id, indexing_step))
}) .map_err(|e| IndexError::Internal(e.into()))?;
.map_err(|e| IndexError::Internal(e.into()))?;
Ok(UpdateResult::Other) Ok(UpdateResult::Other)
} }
@ -333,7 +332,8 @@ impl Index {
update_builder: UpdateBuilder, update_builder: UpdateBuilder,
) -> Result<UpdateResult> { ) -> Result<UpdateResult> {
let mut txn = self.write_txn()?; let mut txn = self.write_txn()?;
let mut builder = update_builder.delete_documents(&mut txn, self) let mut builder = update_builder
.delete_documents(&mut txn, self)
.map_err(|e| IndexError::Internal(e.into()))?; .map_err(|e| IndexError::Internal(e.into()))?;
// We ignore unexisting document ids // We ignore unexisting document ids

View File

@ -10,9 +10,9 @@ use tokio::sync::{mpsc, oneshot, RwLock};
use update_actor::UpdateActorHandle; use update_actor::UpdateActorHandle;
use uuid_resolver::UuidResolverHandle; use uuid_resolver::UuidResolverHandle;
use super::error::{DumpActorError, Result};
use super::{DumpInfo, DumpMsg, DumpStatus, DumpTask}; use super::{DumpInfo, DumpMsg, DumpStatus, DumpTask};
use crate::index_controller::{update_actor, uuid_resolver}; use crate::index_controller::{update_actor, uuid_resolver};
use super::error::{DumpActorError, Result};
pub const CONCURRENT_DUMP_MSG: usize = 10; pub const CONCURRENT_DUMP_MSG: usize = 10;

View File

@ -1,6 +1,8 @@
use meilisearch_error::{Code, ErrorCode}; use meilisearch_error::{Code, ErrorCode};
use crate::index_controller::{update_actor::error::UpdateActorError, uuid_resolver::UuidResolverError}; use crate::index_controller::{
update_actor::error::UpdateActorError, uuid_resolver::UuidResolverError,
};
pub type Result<T> = std::result::Result<T, DumpActorError>; pub type Result<T> = std::result::Result<T, DumpActorError>;

View File

@ -3,8 +3,8 @@ use std::path::Path;
use actix_web::web::Bytes; use actix_web::web::Bytes;
use tokio::sync::{mpsc, oneshot}; use tokio::sync::{mpsc, oneshot};
use super::{DumpActor, DumpActorHandle, DumpInfo, DumpMsg};
use super::error::Result; use super::error::Result;
use super::{DumpActor, DumpActorHandle, DumpInfo, DumpMsg};
#[derive(Clone)] #[derive(Clone)]
pub struct DumpActorHandleImpl { pub struct DumpActorHandleImpl {
@ -35,7 +35,7 @@ impl DumpActorHandleImpl {
update: crate::index_controller::update_actor::UpdateActorHandleImpl<Bytes>, update: crate::index_controller::update_actor::UpdateActorHandleImpl<Bytes>,
index_db_size: usize, index_db_size: usize,
update_db_size: usize, update_db_size: usize,
) -> std::result::Result<Self, Box<dyn std::error::Error>> { ) -> anyhow::Result<Self> {
let (sender, receiver) = mpsc::channel(10); let (sender, receiver) = mpsc::channel(10);
let actor = DumpActor::new( let actor = DumpActor::new(
receiver, receiver,

View File

@ -31,7 +31,7 @@ impl MetadataV1 {
dst: impl AsRef<Path>, dst: impl AsRef<Path>,
size: usize, size: usize,
indexer_options: &IndexerOpts, indexer_options: &IndexerOpts,
) -> std::result::Result<(), Box<dyn std::error::Error>> { ) -> anyhow::Result<()> {
info!( info!(
"Loading dump, dump database version: {}, dump version: V1", "Loading dump, dump database version: {}, dump version: V1",
self.db_version self.db_version
@ -83,7 +83,7 @@ fn load_index(
primary_key: Option<&str>, primary_key: Option<&str>,
size: usize, size: usize,
indexer_options: &IndexerOpts, indexer_options: &IndexerOpts,
) -> std::result::Result<(), Box<dyn std::error::Error>> { ) -> anyhow::Result<()> {
let index_path = dst.as_ref().join(&format!("indexes/index-{}", uuid)); let index_path = dst.as_ref().join(&format!("indexes/index-{}", uuid));
create_dir_all(&index_path)?; create_dir_all(&index_path)?;
@ -172,7 +172,7 @@ impl From<Settings> for index_controller::Settings<Unchecked> {
} }
/// Extract Settings from `settings.json` file present at provided `dir_path` /// Extract Settings from `settings.json` file present at provided `dir_path`
fn import_settings(dir_path: impl AsRef<Path>) -> std::result::Result<Settings, Box<dyn std::error::Error>> { fn import_settings(dir_path: impl AsRef<Path>) -> anyhow::Result<Settings> {
let path = dir_path.as_ref().join("settings.json"); let path = dir_path.as_ref().join("settings.json");
let file = File::open(path)?; let file = File::open(path)?;
let reader = std::io::BufReader::new(file); let reader = std::io::BufReader::new(file);

View File

@ -34,7 +34,7 @@ impl MetadataV2 {
index_db_size: usize, index_db_size: usize,
update_db_size: usize, update_db_size: usize,
indexing_options: &IndexerOpts, indexing_options: &IndexerOpts,
) -> std::result::Result<(), Box<dyn std::error::Error>> { ) -> anyhow::Result<()> {
info!( info!(
"Loading dump from {}, dump database version: {}, dump version: V2", "Loading dump from {}, dump database version: {}, dump version: V2",
self.dump_date, self.db_version self.dump_date, self.db_version

View File

@ -1,7 +1,7 @@
use tokio::sync::oneshot; use tokio::sync::oneshot;
use super::DumpInfo;
use super::error::Result; use super::error::Result;
use super::DumpInfo;
pub enum DumpMsg { pub enum DumpMsg {
CreateDump { CreateDump {

View File

@ -1,6 +1,7 @@
use std::fs::File; use std::fs::File;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use anyhow::Context;
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use log::{info, warn}; use log::{info, warn};
#[cfg(test)] #[cfg(test)]
@ -21,10 +22,10 @@ use crate::{helpers::compression, option::IndexerOpts};
use error::Result; use error::Result;
mod actor; mod actor;
pub mod error;
mod handle_impl; mod handle_impl;
mod loaders; mod loaders;
mod message; mod message;
pub mod error;
const META_FILE_NAME: &str = "metadata.json"; const META_FILE_NAME: &str = "metadata.json";
@ -107,7 +108,7 @@ pub fn load_dump(
index_db_size: usize, index_db_size: usize,
update_db_size: usize, update_db_size: usize,
indexer_opts: &IndexerOpts, indexer_opts: &IndexerOpts,
) -> std::result::Result<(), Box<dyn std::error::Error>> { ) -> anyhow::Result<()> {
let tmp_src = tempfile::tempdir_in(".")?; let tmp_src = tempfile::tempdir_in(".")?;
let tmp_src_path = tmp_src.path(); let tmp_src_path = tmp_src.path();
@ -120,9 +121,7 @@ pub fn load_dump(
let dst_dir = dst_path let dst_dir = dst_path
.as_ref() .as_ref()
.parent() .parent()
// TODO .with_context(|| format!("Invalid db path: {}", dst_path.as_ref().display()))?;
//.with_context(|| format!("Invalid db path: {}", dst_path.as_ref().display()))?;
.unwrap();
let tmp_dst = tempfile::tempdir_in(dst_dir)?; let tmp_dst = tempfile::tempdir_in(dst_dir)?;
@ -188,7 +187,7 @@ where
let dump_path = tokio::task::spawn_blocking(move || -> Result<PathBuf> { let dump_path = tokio::task::spawn_blocking(move || -> Result<PathBuf> {
let temp_dump_file = tempfile::NamedTempFile::new_in(&self.path)?; let temp_dump_file = tempfile::NamedTempFile::new_in(&self.path)?;
compression::to_tar_gz(temp_dump_path, temp_dump_file.path()) compression::to_tar_gz(temp_dump_path, temp_dump_file.path())
.map_err(|e| DumpActorError::Internal(e))?; .map_err(|e| DumpActorError::Internal(e.into()))?;
let dump_path = self.path.join(self.uid).with_extension("dump"); let dump_path = self.path.join(self.uid).with_extension("dump");
temp_dump_file.persist(&dump_path)?; temp_dump_file.persist(&dump_path)?;

View File

@ -19,8 +19,8 @@ use crate::index_controller::{
}; };
use crate::option::IndexerOpts; use crate::option::IndexerOpts;
use super::error::{IndexActorError, Result};
use super::{IndexMeta, IndexMsg, IndexSettings, IndexStore}; use super::{IndexMeta, IndexMsg, IndexSettings, IndexStore};
use super::error::{Result, IndexActorError};
pub const CONCURRENT_INDEX_MSG: usize = 10; pub const CONCURRENT_INDEX_MSG: usize = 10;
@ -31,7 +31,7 @@ pub struct IndexActor<S> {
} }
impl<S: IndexStore + Sync + Send> IndexActor<S> { impl<S: IndexStore + Sync + Send> IndexActor<S> {
pub fn new(receiver: mpsc::Receiver<IndexMsg>, store: S) -> std::result::Result<Self, Box<dyn std::error::Error>> { pub fn new(receiver: mpsc::Receiver<IndexMsg>, store: S) -> anyhow::Result<Self> {
let options = IndexerOpts::default(); let options = IndexerOpts::default();
let update_handler = UpdateHandler::new(&options)?; let update_handler = UpdateHandler::new(&options)?;
let update_handler = Arc::new(update_handler); let update_handler = Arc::new(update_handler);
@ -146,7 +146,6 @@ impl<S: IndexStore + Sync + Send> IndexActor<S> {
.ok_or(IndexActorError::UnexistingIndex)?; .ok_or(IndexActorError::UnexistingIndex)?;
let result = spawn_blocking(move || index.perform_search(query)).await??; let result = spawn_blocking(move || index.perform_search(query)).await??;
Ok(result) Ok(result)
} }
async fn handle_create_index( async fn handle_create_index(
@ -269,7 +268,8 @@ impl<S: IndexStore + Sync + Send> IndexActor<S> {
} }
let mut builder = UpdateBuilder::new(0).settings(&mut txn, &index); let mut builder = UpdateBuilder::new(0).settings(&mut txn, &index);
builder.set_primary_key(primary_key); builder.set_primary_key(primary_key);
builder.execute(|_, _| ()) builder
.execute(|_, _| ())
.map_err(|e| IndexActorError::Internal(Box::new(e)))?; .map_err(|e| IndexActorError::Internal(Box::new(e)))?;
let meta = IndexMeta::new_txn(&index, &txn)?; let meta = IndexMeta::new_txn(&index, &txn)?;
txn.commit()?; txn.commit()?;
@ -340,10 +340,12 @@ impl<S: IndexStore + Sync + Send> IndexActor<S> {
Ok(IndexStats { Ok(IndexStats {
size: index.size(), size: index.size(),
number_of_documents: index.number_of_documents(&rtxn) number_of_documents: index
.number_of_documents(&rtxn)
.map_err(|e| IndexActorError::Internal(Box::new(e)))?, .map_err(|e| IndexActorError::Internal(Box::new(e)))?,
is_indexing: None, is_indexing: None,
fields_distribution: index.fields_distribution(&rtxn) fields_distribution: index
.fields_distribution(&rtxn)
.map_err(|e| IndexActorError::Internal(e.into()))?, .map_err(|e| IndexActorError::Internal(e.into()))?,
}) })
}) })

View File

@ -30,11 +30,7 @@ macro_rules! internal_error {
} }
} }
internal_error!( internal_error!(heed::Error, tokio::task::JoinError, std::io::Error);
heed::Error,
tokio::task::JoinError,
std::io::Error
);
impl ErrorCode for IndexActorError { impl ErrorCode for IndexActorError {
fn error_code(&self) -> Code { fn error_code(&self) -> Code {

View File

@ -12,8 +12,8 @@ use crate::{
index_controller::{Failed, Processed}, index_controller::{Failed, Processed},
}; };
use super::{IndexActor, IndexActorHandle, IndexMeta, IndexMsg, MapIndexStore};
use super::error::Result; use super::error::Result;
use super::{IndexActor, IndexActorHandle, IndexMeta, IndexMsg, MapIndexStore};
#[derive(Clone)] #[derive(Clone)]
pub struct IndexActorHandleImpl { pub struct IndexActorHandleImpl {
@ -22,11 +22,7 @@ pub struct IndexActorHandleImpl {
#[async_trait::async_trait] #[async_trait::async_trait]
impl IndexActorHandle for IndexActorHandleImpl { impl IndexActorHandle for IndexActorHandleImpl {
async fn create_index( async fn create_index(&self, uuid: Uuid, primary_key: Option<String>) -> Result<IndexMeta> {
&self,
uuid: Uuid,
primary_key: Option<String>,
) -> Result<IndexMeta> {
let (ret, receiver) = oneshot::channel(); let (ret, receiver) = oneshot::channel();
let msg = IndexMsg::CreateIndex { let msg = IndexMsg::CreateIndex {
ret, ret,
@ -118,11 +114,7 @@ impl IndexActorHandle for IndexActorHandleImpl {
Ok(receiver.await.expect("IndexActor has been killed")?) Ok(receiver.await.expect("IndexActor has been killed")?)
} }
async fn update_index( async fn update_index(&self, uuid: Uuid, index_settings: IndexSettings) -> Result<IndexMeta> {
&self,
uuid: Uuid,
index_settings: IndexSettings,
) -> Result<IndexMeta> {
let (ret, receiver) = oneshot::channel(); let (ret, receiver) = oneshot::channel();
let msg = IndexMsg::UpdateIndex { let msg = IndexMsg::UpdateIndex {
uuid, uuid,
@ -156,7 +148,7 @@ impl IndexActorHandle for IndexActorHandleImpl {
} }
impl IndexActorHandleImpl { impl IndexActorHandleImpl {
pub fn new(path: impl AsRef<Path>, index_size: usize) -> std::result::Result<Self, Box<dyn std::error::Error>> { pub fn new(path: impl AsRef<Path>, index_size: usize) -> anyhow::Result<Self> {
let (sender, receiver) = mpsc::channel(100); let (sender, receiver) = mpsc::channel(100);
let store = MapIndexStore::new(path, index_size); let store = MapIndexStore::new(path, index_size);

View File

@ -3,9 +3,9 @@ use std::path::PathBuf;
use tokio::sync::oneshot; use tokio::sync::oneshot;
use uuid::Uuid; use uuid::Uuid;
use super::error::Result as IndexResult;
use crate::index::{Checked, Document, SearchQuery, SearchResult, Settings}; use crate::index::{Checked, Document, SearchQuery, SearchResult, Settings};
use crate::index_controller::{Failed, IndexStats, Processed, Processing}; use crate::index_controller::{Failed, IndexStats, Processed, Processing};
use super::error::Result as IndexResult;
use super::{IndexMeta, IndexSettings}; use super::{IndexMeta, IndexSettings};

View File

@ -22,10 +22,10 @@ use self::error::IndexActorError;
use super::IndexSettings; use super::IndexSettings;
mod actor; mod actor;
pub mod error;
mod handle_impl; mod handle_impl;
mod message; mod message;
mod store; mod store;
pub mod error;
#[derive(Debug, Serialize, Deserialize, Clone)] #[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")] #[serde(rename_all = "camelCase")]
@ -60,8 +60,7 @@ impl IndexMeta {
#[async_trait::async_trait] #[async_trait::async_trait]
#[cfg_attr(test, automock)] #[cfg_attr(test, automock)]
pub trait IndexActorHandle { pub trait IndexActorHandle {
async fn create_index(&self, uuid: Uuid, primary_key: Option<String>) async fn create_index(&self, uuid: Uuid, primary_key: Option<String>) -> Result<IndexMeta>;
-> Result<IndexMeta>;
async fn update( async fn update(
&self, &self,
uuid: Uuid, uuid: Uuid,
@ -86,11 +85,7 @@ pub trait IndexActorHandle {
) -> Result<Document>; ) -> Result<Document>;
async fn delete(&self, uuid: Uuid) -> Result<()>; async fn delete(&self, uuid: Uuid) -> Result<()>;
async fn get_index_meta(&self, uuid: Uuid) -> Result<IndexMeta>; async fn get_index_meta(&self, uuid: Uuid) -> Result<IndexMeta>;
async fn update_index( async fn update_index(&self, uuid: Uuid, index_settings: IndexSettings) -> Result<IndexMeta>;
&self,
uuid: Uuid,
index_settings: IndexSettings,
) -> Result<IndexMeta>;
async fn snapshot(&self, uuid: Uuid, path: PathBuf) -> Result<()>; async fn snapshot(&self, uuid: Uuid, path: PathBuf) -> Result<()>;
async fn dump(&self, uuid: Uuid, path: PathBuf) -> Result<()>; async fn dump(&self, uuid: Uuid, path: PathBuf) -> Result<()>;
async fn get_index_stats(&self, uuid: Uuid) -> Result<IndexStats>; async fn get_index_stats(&self, uuid: Uuid) -> Result<IndexStats>;
@ -105,11 +100,7 @@ mod test {
#[async_trait::async_trait] #[async_trait::async_trait]
/// Useful for passing around an `Arc<MockIndexActorHandle>` in tests. /// Useful for passing around an `Arc<MockIndexActorHandle>` in tests.
impl IndexActorHandle for Arc<MockIndexActorHandle> { impl IndexActorHandle for Arc<MockIndexActorHandle> {
async fn create_index( async fn create_index(&self, uuid: Uuid, primary_key: Option<String>) -> Result<IndexMeta> {
&self,
uuid: Uuid,
primary_key: Option<String>,
) -> Result<IndexMeta> {
self.as_ref().create_index(uuid, primary_key).await self.as_ref().create_index(uuid, primary_key).await
} }

View File

@ -29,12 +29,12 @@ use self::dump_actor::load_dump;
use self::error::IndexControllerError; use self::error::IndexControllerError;
mod dump_actor; mod dump_actor;
pub mod error;
mod index_actor; mod index_actor;
mod snapshot; mod snapshot;
mod update_actor; mod update_actor;
mod updates; mod updates;
mod uuid_resolver; mod uuid_resolver;
pub mod error;
#[derive(Debug, Serialize, Deserialize, Clone)] #[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")] #[serde(rename_all = "camelCase")]
@ -83,7 +83,7 @@ pub struct Stats {
} }
impl IndexController { impl IndexController {
pub fn new(path: impl AsRef<Path>, options: &Opt) -> std::result::Result<Self, Box<dyn std::error::Error>> { pub fn new(path: impl AsRef<Path>, options: &Opt) -> anyhow::Result<Self> {
let index_size = options.max_mdb_size.get_bytes() as usize; let index_size = options.max_mdb_size.get_bytes() as usize;
let update_store_size = options.max_udb_size.get_bytes() as usize; let update_store_size = options.max_udb_size.get_bytes() as usize;
@ -238,10 +238,7 @@ impl IndexController {
} }
} }
pub async fn create_index( pub async fn create_index(&self, index_settings: IndexSettings) -> Result<IndexMetadata> {
&self,
index_settings: IndexSettings,
) -> Result<IndexMetadata> {
let IndexSettings { uid, primary_key } = index_settings; let IndexSettings { uid, primary_key } = index_settings;
let uid = uid.ok_or(IndexControllerError::MissingUid)?; let uid = uid.ok_or(IndexControllerError::MissingUid)?;
let uuid = Uuid::new_v4(); let uuid = Uuid::new_v4();

View File

@ -52,7 +52,7 @@ where
} }
} }
async fn perform_snapshot(&self) -> std::result::Result<(), Box<dyn std::error::Error + Sync + Send + 'static>> { async fn perform_snapshot(&self) -> anyhow::Result<()> {
info!("Performing snapshot."); info!("Performing snapshot.");
let snapshot_dir = self.snapshot_path.clone(); let snapshot_dir = self.snapshot_path.clone();
@ -77,7 +77,7 @@ where
let snapshot_path = self let snapshot_path = self
.snapshot_path .snapshot_path
.join(format!("{}.snapshot", self.db_name)); .join(format!("{}.snapshot", self.db_name));
let snapshot_path = spawn_blocking(move || -> Result<PathBuf, Box<dyn std::error::Error + Sync + Send + 'static>> { let snapshot_path = spawn_blocking(move || -> anyhow::Result<PathBuf> {
let temp_snapshot_file = tempfile::NamedTempFile::new_in(snapshot_dir)?; let temp_snapshot_file = tempfile::NamedTempFile::new_in(snapshot_dir)?;
let temp_snapshot_file_path = temp_snapshot_file.path().to_owned(); let temp_snapshot_file_path = temp_snapshot_file.path().to_owned();
compression::to_tar_gz(temp_snapshot_path, temp_snapshot_file_path)?; compression::to_tar_gz(temp_snapshot_path, temp_snapshot_file_path)?;
@ -97,7 +97,7 @@ pub fn load_snapshot(
snapshot_path: impl AsRef<Path>, snapshot_path: impl AsRef<Path>,
ignore_snapshot_if_db_exists: bool, ignore_snapshot_if_db_exists: bool,
ignore_missing_snapshot: bool, ignore_missing_snapshot: bool,
) -> std::result::Result<(), Box<dyn std::error::Error>> { ) -> anyhow::Result<()> {
if !db_path.as_ref().exists() && snapshot_path.as_ref().exists() { if !db_path.as_ref().exists() && snapshot_path.as_ref().exists() {
match compression::from_tar_gz(snapshot_path, &db_path) { match compression::from_tar_gz(snapshot_path, &db_path) {
Ok(()) => Ok(()), Ok(()) => Ok(()),

View File

@ -13,8 +13,8 @@ use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use uuid::Uuid; use uuid::Uuid;
use super::{PayloadData, UpdateMsg, UpdateStore, UpdateStoreInfo};
use super::error::{Result, UpdateActorError}; use super::error::{Result, UpdateActorError};
use super::{PayloadData, UpdateMsg, UpdateStore, UpdateStoreInfo};
use crate::index_controller::index_actor::IndexActorHandle; use crate::index_controller::index_actor::IndexActorHandle;
use crate::index_controller::{UpdateMeta, UpdateStatus}; use crate::index_controller::{UpdateMeta, UpdateStatus};
@ -36,7 +36,7 @@ where
inbox: mpsc::Receiver<UpdateMsg<D>>, inbox: mpsc::Receiver<UpdateMsg<D>>,
path: impl AsRef<Path>, path: impl AsRef<Path>,
index_handle: I, index_handle: I,
) -> std::result::Result<Self, Box<dyn std::error::Error>> { ) -> anyhow::Result<Self> {
let path = path.as_ref().join("updates"); let path = path.as_ref().join("updates");
std::fs::create_dir_all(&path)?; std::fs::create_dir_all(&path)?;
@ -201,9 +201,9 @@ where
async fn handle_get_update(&self, uuid: Uuid, id: u64) -> Result<UpdateStatus> { async fn handle_get_update(&self, uuid: Uuid, id: u64) -> Result<UpdateStatus> {
let store = self.store.clone(); let store = self.store.clone();
tokio::task::spawn_blocking(move || { tokio::task::spawn_blocking(move || {
let result = store let result = store
.meta(uuid, id)? .meta(uuid, id)?
.ok_or(UpdateActorError::UnexistingUpdate(id))?; .ok_or(UpdateActorError::UnexistingUpdate(id))?;
Ok(result) Ok(result)
}) })
.await? .await?

View File

@ -45,6 +45,7 @@ impl From<tokio::sync::oneshot::error::RecvError> for UpdateActorError {
} }
internal_error!( internal_error!(
UpdateActorError:
heed::Error, heed::Error,
std::io::Error, std::io::Error,
serde_json::Error, serde_json::Error,

View File

@ -22,7 +22,7 @@ where
index_handle: I, index_handle: I,
path: impl AsRef<Path>, path: impl AsRef<Path>,
update_store_size: usize, update_store_size: usize,
) -> std::result::Result<Self, Box<dyn std::error::Error>> ) -> anyhow::Result<Self>
where where
I: IndexActorHandle + Clone + Send + Sync + 'static, I: IndexActorHandle + Clone + Send + Sync + 'static,
{ {

View File

@ -4,8 +4,8 @@ use std::path::PathBuf;
use tokio::sync::{mpsc, oneshot}; use tokio::sync::{mpsc, oneshot};
use uuid::Uuid; use uuid::Uuid;
use super::{PayloadData, UpdateMeta, UpdateStatus, UpdateStoreInfo};
use super::error::Result; use super::error::Result;
use super::{PayloadData, UpdateMeta, UpdateStatus, UpdateStoreInfo};
pub enum UpdateMsg<D> { pub enum UpdateMsg<D> {
Update { Update {

View File

@ -7,16 +7,16 @@ use uuid::Uuid;
use crate::index_controller::{UpdateMeta, UpdateStatus}; use crate::index_controller::{UpdateMeta, UpdateStatus};
use actor::UpdateActor; use actor::UpdateActor;
use message::UpdateMsg;
use error::Result; use error::Result;
use message::UpdateMsg;
pub use handle_impl::UpdateActorHandleImpl; pub use handle_impl::UpdateActorHandleImpl;
pub use store::{UpdateStore, UpdateStoreInfo}; pub use store::{UpdateStore, UpdateStoreInfo};
mod actor; mod actor;
pub mod error;
mod handle_impl; mod handle_impl;
mod message; mod message;
pub mod error;
pub mod store; pub mod store;
type PayloadData<D> = std::result::Result<D, PayloadError>; type PayloadData<D> = std::result::Result<D, PayloadError>;

View File

@ -9,7 +9,7 @@ use heed::{EnvOpenOptions, RoTxn};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use uuid::Uuid; use uuid::Uuid;
use super::{State, UpdateStore, Result}; use super::{Result, State, UpdateStore};
use crate::index_controller::{ use crate::index_controller::{
index_actor::IndexActorHandle, update_actor::store::update_uuid_to_file_path, Enqueued, index_actor::IndexActorHandle, update_actor::store::update_uuid_to_file_path, Enqueued,
UpdateStatus, UpdateStatus,
@ -125,7 +125,7 @@ impl UpdateStore {
src: impl AsRef<Path>, src: impl AsRef<Path>,
dst: impl AsRef<Path>, dst: impl AsRef<Path>,
db_size: usize, db_size: usize,
) -> std::result::Result<(), Box<dyn std::error::Error>> { ) -> anyhow::Result<()> {
let dst_update_path = dst.as_ref().join("updates/"); let dst_update_path = dst.as_ref().join("updates/");
create_dir_all(&dst_update_path)?; create_dir_all(&dst_update_path)?;

View File

@ -23,10 +23,10 @@ use uuid::Uuid;
use codec::*; use codec::*;
use super::UpdateMeta;
use super::error::Result; use super::error::Result;
use crate::index_controller::{index_actor::CONCURRENT_INDEX_MSG, updates::*, IndexActorHandle}; use super::UpdateMeta;
use crate::helpers::EnvSizer; use crate::helpers::EnvSizer;
use crate::index_controller::{index_actor::CONCURRENT_INDEX_MSG, updates::*, IndexActorHandle};
#[allow(clippy::upper_case_acronyms)] #[allow(clippy::upper_case_acronyms)]
type BEU64 = U64<heed::byteorder::BE>; type BEU64 = U64<heed::byteorder::BE>;
@ -110,7 +110,7 @@ impl UpdateStore {
fn new( fn new(
mut options: EnvOpenOptions, mut options: EnvOpenOptions,
path: impl AsRef<Path>, path: impl AsRef<Path>,
) -> std::result::Result<(Self, mpsc::Receiver<()>), Box<dyn std::error::Error>> { ) -> anyhow::Result<(Self, mpsc::Receiver<()>)> {
options.max_dbs(5); options.max_dbs(5);
let env = options.open(&path)?; let env = options.open(&path)?;
@ -141,7 +141,7 @@ impl UpdateStore {
path: impl AsRef<Path>, path: impl AsRef<Path>,
index_handle: impl IndexActorHandle + Clone + Sync + Send + 'static, index_handle: impl IndexActorHandle + Clone + Sync + Send + 'static,
must_exit: Arc<AtomicBool>, must_exit: Arc<AtomicBool>,
) -> std::result::Result<Arc<Self>, Box<dyn std::error::Error>> { ) -> anyhow::Result<Arc<Self>> {
let (update_store, mut notification_receiver) = Self::new(options, path)?; let (update_store, mut notification_receiver) = Self::new(options, path)?;
let update_store = Arc::new(update_store); let update_store = Arc::new(update_store);
@ -270,11 +270,8 @@ impl UpdateStore {
} }
_ => { _ => {
let _update_id = self.next_update_id_raw(wtxn, index_uuid)?; let _update_id = self.next_update_id_raw(wtxn, index_uuid)?;
self.updates.put( self.updates
wtxn, .put(wtxn, &(index_uuid, update.id()), &update)?;
&(index_uuid, update.id()),
&update,
)?;
} }
} }
Ok(()) Ok(())
@ -283,10 +280,7 @@ impl UpdateStore {
/// Executes the user provided function on the next pending update (the one with the lowest id). /// Executes the user provided function on the next pending update (the one with the lowest id).
/// This is asynchronous as it let the user process the update with a read-only txn and /// This is asynchronous as it let the user process the update with a read-only txn and
/// only writing the result meta to the processed-meta store *after* it has been processed. /// only writing the result meta to the processed-meta store *after* it has been processed.
fn process_pending_update( fn process_pending_update(&self, index_handle: impl IndexActorHandle) -> Result<Option<()>> {
&self,
index_handle: impl IndexActorHandle,
) -> Result<Option<()>> {
// Create a read transaction to be able to retrieve the pending update in order. // Create a read transaction to be able to retrieve the pending update in order.
let rtxn = self.env.read_txn()?; let rtxn = self.env.read_txn()?;
let first_meta = self.pending_queue.first(&rtxn)?; let first_meta = self.pending_queue.first(&rtxn)?;
@ -353,11 +347,8 @@ impl UpdateStore {
Err(res) => res.into(), Err(res) => res.into(),
}; };
self.updates.put( self.updates
&mut wtxn, .put(&mut wtxn, &(index_uuid, update_id), &result)?;
&(index_uuid, update_id),
&result,
)?;
wtxn.commit()?; wtxn.commit()?;
@ -704,18 +695,10 @@ mod test {
let txn = store.env.read_txn().unwrap(); let txn = store.env.read_txn().unwrap();
assert!(store.pending_queue.first(&txn).unwrap().is_none()); assert!(store.pending_queue.first(&txn).unwrap().is_none());
let update = store let update = store.updates.get(&txn, &(uuid, 0)).unwrap().unwrap();
.updates
.get(&txn, &(uuid, 0))
.unwrap()
.unwrap();
assert!(matches!(update, UpdateStatus::Processed(_))); assert!(matches!(update, UpdateStatus::Processed(_)));
let update = store let update = store.updates.get(&txn, &(uuid, 1)).unwrap().unwrap();
.updates
.get(&txn, &(uuid, 1))
.unwrap()
.unwrap();
assert!(matches!(update, UpdateStatus::Failed(_))); assert!(matches!(update, UpdateStatus::Failed(_)));
} }

View File

@ -25,7 +25,7 @@ pub enum UpdateMeta {
}, },
ClearDocuments, ClearDocuments,
DeleteDocuments { DeleteDocuments {
ids: Vec<String> ids: Vec<String>,
}, },
Settings(Settings<Unchecked>), Settings(Settings<Unchecked>),
} }

View File

@ -30,6 +30,7 @@ async fn main() -> Result<(), MainError> {
.into(), .into(),
); );
} }
#[cfg(all(not(debug_assertions), feature = "analytics"))] #[cfg(all(not(debug_assertions), feature = "analytics"))]
if !opt.no_analytics { if !opt.no_analytics {
let logger = let logger =

View File

@ -24,17 +24,19 @@ pub enum UpdateType {
Customs, Customs,
DocumentsAddition { DocumentsAddition {
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
number: Option<usize> number: Option<usize>,
}, },
DocumentsPartial { DocumentsPartial {
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
number: Option<usize> number: Option<usize>,
}, },
DocumentsDeletion { DocumentsDeletion {
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
number: Option<usize> number: Option<usize>,
},
Settings {
settings: Settings<Unchecked>,
}, },
Settings { settings: Settings<Unchecked> },
} }
impl From<&UpdateStatus> for UpdateType { impl From<&UpdateStatus> for UpdateType {
@ -60,9 +62,9 @@ impl From<&UpdateStatus> for UpdateType {
} }
} }
UpdateMeta::ClearDocuments => UpdateType::ClearAll, UpdateMeta::ClearDocuments => UpdateType::ClearAll,
UpdateMeta::DeleteDocuments { ids } => { UpdateMeta::DeleteDocuments { ids } => UpdateType::DocumentsDeletion {
UpdateType::DocumentsDeletion { number: Some(ids.len()) } number: Some(ids.len()),
} },
UpdateMeta::Settings(settings) => UpdateType::Settings { UpdateMeta::Settings(settings) => UpdateType::Settings {
settings: settings.clone(), settings: settings.clone(),
}, },

View File

@ -21,7 +21,7 @@ async fn create_and_get_index() {
assert_eq!(response.as_object().unwrap().len(), 5); assert_eq!(response.as_object().unwrap().len(), 5);
} }
// TODO: partial test since we are testing error, amd error is not yet fully implemented in // TODO: partial test since we are testing error, and error is not yet fully implemented in
// transplant // transplant
#[actix_rt::test] #[actix_rt::test]
async fn get_unexisting_index() { async fn get_unexisting_index() {