fix auto-synchronization with zk

Cargo.lock

@@ -303,6 +303,12 @@ dependencies = [
  "alloc-no-stdlib",
 ]
 
+[[package]]
+name = "allocator-api2"
+version = "0.2.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0942ffc6dcaadf03badf6e6a2d0228460359d5e34b57ccdc720b7382dfbd5ec5"
+
 [[package]]
 name = "anes"
 version = "0.1.6"
@@ -664,6 +670,15 @@ version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"
 
+[[package]]
+name = "castaway"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8a17ed5635fc8536268e5d4de1e22e81ac34419e5f052d4d51f4e01dcc263fcc"
+dependencies = [
+ "rustversion",
+]
+
 [[package]]
 name = "cc"
 version = "1.0.79"
@@ -791,7 +806,7 @@ version = "4.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "191d9573962933b4027f932c600cd252ce27a8ad5979418fe78e43c07996f27b"
 dependencies = [
- "heck",
+ "heck 0.4.1",
  "proc-macro2",
  "quote",
  "syn 2.0.28",
@@ -809,6 +824,17 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "acbf1af155f9b9ef647e42cdc158db4b64a1b61f743629225fde6f3e0be2a7c7"
 
+[[package]]
+name = "compact_str"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e33b5c3ee2b4ffa00ac2b00d1645cd9229ade668139bccf95f15fadcf374127b"
+dependencies = [
+ "castaway",
+ "itoa",
+ "ryu",
+]
+
 [[package]]
 name = "concat-arrays"
 version = "0.1.2"
@@ -832,6 +858,26 @@ dependencies = [
  "windows-sys 0.45.0",
 ]
 
+[[package]]
+name = "const_format"
+version = "0.2.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c990efc7a285731f9a4378d81aff2f0e85a2c8781a05ef0f8baa8dac54d0ff48"
+dependencies = [
+ "const_format_proc_macros",
+]
+
+[[package]]
+name = "const_format_proc_macros"
+version = "0.2.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e026b6ce194a874cb9cf32cd5772d1ef9767cc8fcb5765948d74f37a9d8b2bf6"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-xid",
+]
+
 [[package]]
 name = "constant_time_eq"
 version = "0.1.5"
@@ -1224,9 +1270,9 @@ dependencies = [
 
 [[package]]
 name = "either"
-version = "1.8.1"
+version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7fcaabb2fef8c910e7f4c7ce9f67a1283a1715879a7c230ca9d6d1ae31f16d91"
+checksum = "a26ae43d7bcc3b814de94796a5e736d4029efb0ee900c12e2d54c993ad1a1e07"
 dependencies = [
  "serde",
 ]
@@ -1715,6 +1761,19 @@ name = "hashbrown"
 version = "0.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2c6201b9ff9fd90a5a3bac2e56a830d0caa509576f0e503818ee82c181b3437a"
+dependencies = [
+ "ahash 0.8.3",
+ "allocator-api2",
+]
+
+[[package]]
+name = "hashlink"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "312f66718a2d7789ffef4f4b7b213138ed9f1eb3aa1d0d82fc99f88fb3ffd26f"
+dependencies = [
+ "hashbrown 0.14.0",
+]
 
 [[package]]
 name = "heapless"
@@ -1729,6 +1788,15 @@ dependencies = [
  "stable_deref_trait",
 ]
 
+[[package]]
+name = "heck"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6d621efb26863f0e9924c6ac577e8275e5e6b77455db64ffa6c65c904e9e132c"
+dependencies = [
+ "unicode-segmentation",
+]
+
 [[package]]
 name = "heck"
 version = "0.4.1"
@@ -1892,6 +1960,12 @@ dependencies = [
  "unicode-normalization",
 ]
 
+[[package]]
+name = "ignore-result"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "665ff4dce8edd10d490641ccb78949832f1ddbff02c584fb1f85ab888fe0e50c"
+
 [[package]]
 name = "index-scheduler"
 version = "1.3.0"
@@ -2596,6 +2670,7 @@ dependencies = [
  "walkdir",
  "yaup",
  "zip",
+ "zookeeper-client",
 ]
 
 [[package]]
@@ -2605,6 +2680,7 @@ dependencies = [
  "base64 0.21.2",
  "enum-iterator",
  "hmac",
+ "log",
  "maplit",
  "meilisearch-types",
  "rand",
@@ -2614,7 +2690,9 @@ dependencies = [
  "sha2",
  "thiserror",
  "time",
+ "tokio",
  "uuid 1.3.3",
+ "zookeeper-client",
 ]
 
 [[package]]
@@ -2864,6 +2942,27 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "num_enum"
+version = "0.5.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1f646caf906c20226733ed5b1374287eb97e3c2a5c227ce668c1f2ce20ae57c9"
+dependencies = [
+ "num_enum_derive",
+]
+
+[[package]]
+name = "num_enum_derive"
+version = "0.5.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dcbff9bc912032c62bf65ef1d5aea88983b420f4f839db1e9b0c281a25c9c799"
+dependencies = [
+ "proc-macro-crate",
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "object"
 version = "0.30.3"
@@ -3159,6 +3258,16 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"
 
+[[package]]
+name = "proc-macro-crate"
+version = "1.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f4c021e1093a56626774e81216a4ce732a735e5bad4868a03f3ed65ca0c3919"
+dependencies = [
+ "once_cell",
+ "toml_edit",
+]
+
 [[package]]
 name = "proc-macro-error"
 version = "1.0.4"
@@ -3853,6 +3962,28 @@ version = "0.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
 
+[[package]]
+name = "strum"
+version = "0.23.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cae14b91c7d11c9a851d3fbc80a963198998c2a64eec840477fa92d8ce9b70bb"
+dependencies = [
+ "strum_macros",
+]
+
+[[package]]
+name = "strum_macros"
+version = "0.23.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5bb0dc7ee9c15cea6199cde9a127fa16a4c5819af85395457ad72d68edc85a38"
+dependencies = [
+ "heck 0.3.3",
+ "proc-macro2",
+ "quote",
+ "rustversion",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "subtle"
 version = "2.5.0"
@@ -4732,6 +4863,27 @@ dependencies = [
  "zstd 0.11.2+zstd.1.5.2",
 ]
 
+[[package]]
+name = "zookeeper-client"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ade019937af9b1e2160582cb8d905bcff5d4b71da82444b7bf2ccc9c93f344f2"
+dependencies = [
+ "bytes",
+ "compact_str",
+ "const_format",
+ "either",
+ "hashbrown 0.12.3",
+ "hashlink",
+ "ignore-result",
+ "log",
+ "num_enum",
+ "static_assertions",
+ "strum",
+ "thiserror",
+ "tokio",
+]
+
 [[package]]
 name = "zstd"
 version = "0.11.2+zstd.1.5.2"

docker-compose.yml

@@ -0,0 +1,37 @@
+version: "3.9"
+services:
+  zk1:
+    container_name: zk1
+    hostname: zk1
+    image: bitnami/zookeeper:3.7.1
+    ports:
+      - 21811:2181
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+      - ZOO_SERVER_ID=1
+      - ZOO_SERVERS=0.0.0.0:2888:3888,zk2:2888:3888,zk3:2888:3888
+  zk2:
+    container_name: zk2
+    hostname: zk2
+    image: bitnami/zookeeper:3.7.1
+    ports:
+      - 21812:2181
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+      - ZOO_SERVER_ID=2
+      - ZOO_SERVERS=zk1:2888:3888,0.0.0.0:2888:3888,zk3:2888:3888
+  zk3:
+    container_name: zk3
+    hostname: zk3
+    image: bitnami/zookeeper:3.7.1
+    ports:
+      - 21813:2181
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+      - ZOO_SERVER_ID=3
+      - ZOO_SERVERS=zk1:2888:3888,zk2:2888:3888,0.0.0.0:2888:3888
+  zoonavigator:
+    container_name: zoonavigator
+    image: elkozmon/zoonavigator
+    ports:
+      - 9000:9000

meilisearch-auth/Cargo.toml

@@ -14,6 +14,7 @@ license.workspace = true
 base64 = "0.21.0"
 enum-iterator = "1.4.0"
 hmac = "0.12.1"
+log = "0.4.19"
 maplit = "1.0.2"
 meilisearch-types = { path = "../meilisearch-types" }
 rand = "0.8.5"
@@ -23,4 +24,6 @@ serde_json = { version = "1.0.95", features = ["preserve_order"] }
 sha2 = "0.10.6"
 thiserror = "1.0.40"
 time = { version = "0.3.20", features = ["serde-well-known", "formatting", "parsing", "macros"] }
+tokio = { version = "1.27.0", features = ["full"] }
 uuid = { version = "1.3.1", features = ["serde", "v4"] }
+zookeeper-client = "0.4.0"

meilisearch-auth/src/error.rs

@@ -2,6 +2,7 @@ use std::error::Error;
 
 use meilisearch_types::error::{Code, ErrorCode};
 use meilisearch_types::internal_error;
+use zookeeper_client as zk;
 
 pub type Result<T> = std::result::Result<T, AuthControllerError>;
 
@@ -19,6 +20,8 @@ internal_error!(
     AuthControllerError: meilisearch_types::milli::heed::Error,
     std::io::Error,
     serde_json::Error,
+    tokio::task::JoinError,
+    zk::Error,
     std::str::Utf8Error
 );
 

meilisearch-auth/src/lib.rs

@@ -16,22 +16,119 @@ pub use store::open_auth_store_env;
 use store::{generate_key_as_hexa, HeedAuthStore};
 use time::OffsetDateTime;
 use uuid::Uuid;
+use zookeeper_client as zk;
 
 #[derive(Clone)]
 pub struct AuthController {
     store: Arc<HeedAuthStore>,
     master_key: Option<String>,
+    zk: Option<zk::Client>,
 }
 
 impl AuthController {
-    pub fn new(db_path: impl AsRef<Path>, master_key: &Option<String>) -> Result<Self> {
+    pub async fn new(
+        db_path: impl AsRef<Path>,
+        master_key: &Option<String>,
+        zk: Option<zk::Client>,
+    ) -> Result<Self> {
         let store = HeedAuthStore::new(db_path)?;
+        let controller = Self { store: Arc::new(store), master_key: master_key.clone(), zk };
 
-        if store.is_empty()? {
-            generate_default_keys(&store)?;
+        match controller.zk {
+            // setup the auth zk environment, the `auth` node
+            Some(ref zk) => {
+                let options =
+                    zk::CreateOptions::new(zk::CreateMode::Persistent, zk::Acl::anyone_all());
+                // TODO: we should catch the potential unexpected errors here https://docs.rs/zookeeper-client/latest/zookeeper_client/struct.Client.html#method.create
+                // for the moment we consider that `create` only returns Error::NodeExists.
+                match zk.create("/auth", &[], &options).await {
+                    // If the store is empty, we must generate and push the default api-keys.
+                    Ok(_) => generate_default_keys(&controller).await?,
+                    // If the node exist we should clear our DB and download all the existing api-keys
+                    Err(zk::Error::NodeExists) => {
+                        log::warn!("Auth directory already exists, we need to clear our keys + import the one in zookeeper");
+
+                        let store = controller.store.clone();
+                        tokio::task::spawn_blocking(move || store.delete_all_keys()).await??;
+                        let children = zk
+                            .list_children("/auth")
+                            .await
+                            .expect("Internal, the auth directory was deleted during execution.");
+
+                        log::info!("Importing {} api-keys", children.len());
+                        for path in children {
+                            log::info!("  Importing {}", path);
+                            match zk.get_data(&format!("/auth/{}", &path)).await {
+                                Ok((key, _stat)) => {
+                                let key = serde_json::from_slice(&key).unwrap();
+                                let store = controller.store.clone();
+                                tokio::task::spawn_blocking(move || store.put_api_key(key))
+                                    .await??;
+
+                                },
+                                Err(e) => panic!("{e}")
+                            }
+                            // else the file was deleted while we were inserting the key. We ignore it.
+                            // TODO: What happens if someone updates the files before we have the time
+                            //       to setup the watcher
+                        }
+                    }
+                    e @ Err(
+                        zk::Error::NoNode
+                        | zk::Error::NoChildrenForEphemerals
+                        | zk::Error::InvalidAcl,
+                    ) => unreachable!("{e:?}"),
+                    Err(e) => {
+                        panic!("{e}")
+                    }
+                }
+                // TODO: Race condition above:
+                //       What happens if two node join exactly at the same moment:
+                //       One will create the dir
+                //       The second one will delete its DB, load nothing and install a watcher
+                //       The first one will push its keys and should wake up and update the second one.
+                //     / BUT, if the second one delete its DB and the first one push its files before the second one install the watcher we're fucked
+
+                // Zookeeper Event listener loop
+                let controller_clone = controller.clone();
+                let mut watcher = zk.watch("/auth", zk::AddWatchMode::PersistentRecursive).await?;
+                let czk = zk.clone();
+                tokio::spawn(async move {
+                    let zk = czk;
+                    loop {
+                        let zk::WatchedEvent { event_type, session_state, path } =
+                            dbg!(watcher.changed().await);
+
+                        match event_type {
+                            zk::EventType::Session => panic!("Session error {:?}", session_state),
+                            // a key is deleted from zk
+                            zk::EventType::NodeDeleted => {
+                                // TODO: ugly unwraps
+                                let uuid = path.strip_prefix("/auth/").unwrap();
+                                let uuid = Uuid::parse_str(&uuid).unwrap();
+                                log::info!("The key {} has been deleted", uuid);
+                                dbg!(controller_clone.store.delete_api_key(uuid).unwrap());
+                            }
+                            zk::EventType::NodeCreated | zk::EventType::NodeDataChanged => {
+                                let (key, stat) = zk.get_data(&path).await.unwrap();
+                                let key: Key = serde_json::from_slice(&key).unwrap();
+                                log::info!("The key {} has been deleted", key.uid);
+                                
+                                dbg!(controller_clone.store.put_api_key(key).unwrap());
+                            }
+                            zk::EventType::NodeChildrenChanged => panic!("Got the unexpected NodeChildrenChanged event, what is it used for?"),
+                        }
+                    }
+                });
+            }
+            None => {
+                if controller.store.is_empty()? {
+                    generate_default_keys(&controller).await?;
+                }
+            }
         }
 
-        Ok(Self { store: Arc::new(store), master_key: master_key.clone() })
+        Ok(controller)
     }
 
     /// Return `Ok(())` if the auth controller is able to access one of its database.
@@ -50,14 +147,27 @@ impl AuthController {
         self.store.used_size()
     }
 
-    pub fn create_key(&self, create_key: CreateApiKey) -> Result<Key> {
+    pub async fn create_key(&self, create_key: CreateApiKey) -> Result<Key> {
         match self.store.get_api_key(create_key.uid)? {
             Some(_) => Err(AuthControllerError::ApiKeyAlreadyExists(create_key.uid.to_string())),
-            None => self.store.put_api_key(create_key.to_key()),
+            None => self.put_key(create_key.to_key()).await,
         }
     }
 
-    pub fn update_key(&self, uid: Uuid, patch: PatchApiKey) -> Result<Key> {
+    pub async fn put_key(&self, key: Key) -> Result<Key> {
+        let store = self.store.clone();
+        // TODO: we may commit only after zk persisted the keys
+        let key = tokio::task::spawn_blocking(move || store.put_api_key(key)).await??;
+        if let Some(ref zk) = self.zk {
+            let options = zk::CreateOptions::new(zk::CreateMode::Persistent, zk::Acl::anyone_all());
+
+            zk.create(&format!("/auth/{}", key.uid), &serde_json::to_vec_pretty(&key)?, &options)
+                .await?;
+        }
+        Ok(key)
+    }
+
+    pub async fn update_key(&self, uid: Uuid, patch: PatchApiKey) -> Result<Key> {
         let mut key = self.get_key(uid)?;
         match patch.description {
             Setting::NotSet => (),
@@ -68,7 +178,14 @@ impl AuthController {
             name => key.name = name.set(),
         };
         key.updated_at = OffsetDateTime::now_utc();
-        self.store.put_api_key(key)
+        let store = self.store.clone();
+        // TODO: we may commit only after zk persisted the keys
+        let key = tokio::task::spawn_blocking(move || store.put_api_key(key)).await??;
+        if let Some(ref zk) = self.zk {
+            zk.set_data(&format!("/auth/{}", key.uid), &serde_json::to_vec_pretty(&key)?, None)
+                .await?;
+        }
+        Ok(key)
     }
 
     pub fn get_key(&self, uid: Uuid) -> Result<Key> {
@@ -109,8 +226,13 @@ impl AuthController {
         self.store.list_api_keys()
     }
 
-    pub fn delete_key(&self, uid: Uuid) -> Result<()> {
-        if self.store.delete_api_key(uid)? {
+    pub async fn delete_key(&self, uid: Uuid) -> Result<()> {
+        let store = self.store.clone();
+        let deleted = tokio::task::spawn_blocking(move || store.delete_api_key(uid)).await??;
+        if deleted {
+            if let Some(ref zk) = self.zk {
+                zk.delete(&format!("/auth/{}", uid), None).await?;
+            }
             Ok(())
         } else {
             Err(AuthControllerError::ApiKeyNotFound(uid.to_string()))
@@ -159,7 +281,7 @@ impl AuthController {
         self.store.delete_all_keys()
     }
 
-    /// Delete all the keys in the DB.
+    /// Insert a key in the DB without any check on its validity
     pub fn raw_insert_key(&mut self, key: Key) -> Result<()> {
         self.store.put_api_key(key)?;
         Ok(())
@@ -304,9 +426,9 @@ pub struct IndexSearchRules {
     pub filter: Option<serde_json::Value>,
 }
 
-fn generate_default_keys(store: &HeedAuthStore) -> Result<()> {
-    store.put_api_key(Key::default_admin())?;
-    store.put_api_key(Key::default_search())?;
+async fn generate_default_keys(controller: &AuthController) -> Result<()> {
+    controller.put_key(Key::default_admin()).await?;
+    controller.put_key(Key::default_search()).await?;
 
     Ok(())
 }

meilisearch/Cargo.toml

@@ -105,6 +105,7 @@ walkdir = "2.3.3"
 yaup = "0.2.1"
 serde_urlencoded = "0.7.1"
 termcolor = "1.2.0"
+zookeeper-client = "0.4.0"
 
 [dev-dependencies]
 actix-rt = "2.8.0"

meilisearch/src/analytics/segment_analytics.rs

@@ -312,6 +312,7 @@ impl From<Opt> for Infos {
             config_file_path,
             #[cfg(all(not(debug_assertions), feature = "analytics"))]
                 no_analytics: _,
+            zk_url: _,
         } = options;
 
         let schedule_snapshot = match schedule_snapshot {

meilisearch/src/lib.rs

@@ -39,6 +39,7 @@ use meilisearch_types::versioning::{check_version_file, create_version_file};
 use meilisearch_types::{compression, milli, VERSION_FILE_NAME};
 pub use option::Opt;
 use option::ScheduleSnapshot;
+use zookeeper_client as zk;
 
 use crate::error::MeilisearchHttpError;
 
@@ -136,14 +137,17 @@ enum OnFailure {
     KeepDb,
 }
 
-pub fn setup_meilisearch(opt: &Opt) -> anyhow::Result<(Arc<IndexScheduler>, Arc<AuthController>)> {
+pub async fn setup_meilisearch(
+    opt: &Opt,
+    zk: Option<zk::Client>,
+) -> anyhow::Result<(Arc<IndexScheduler>, Arc<AuthController>)> {
     let empty_db = is_empty_db(&opt.db_path);
     let (index_scheduler, auth_controller) = if let Some(ref snapshot_path) = opt.import_snapshot {
         let snapshot_path_exists = snapshot_path.exists();
         // the db is empty and the snapshot exists, import it
         if empty_db && snapshot_path_exists {
             match compression::from_tar_gz(snapshot_path, &opt.db_path) {
-                Ok(()) => open_or_create_database_unchecked(opt, OnFailure::RemoveDb)?,
+                Ok(()) => open_or_create_database_unchecked(opt, OnFailure::RemoveDb, zk).await?,
                 Err(e) => {
                     std::fs::remove_dir_all(&opt.db_path)?;
                     return Err(e);
@@ -160,14 +164,14 @@ pub fn setup_meilisearch(opt: &Opt) -> anyhow::Result<(Arc<IndexScheduler>, Arc<
             bail!("snapshot doesn't exist at {}", snapshot_path.display())
         // the snapshot and the db exist, and we can ignore the snapshot because of the ignore_snapshot_if_db_exists flag
         } else {
-            open_or_create_database(opt, empty_db)?
+            open_or_create_database(opt, empty_db, zk).await?
         }
     } else if let Some(ref path) = opt.import_dump {
         let src_path_exists = path.exists();
         // the db is empty and the dump exists, import it
         if empty_db && src_path_exists {
             let (mut index_scheduler, mut auth_controller) =
-                open_or_create_database_unchecked(opt, OnFailure::RemoveDb)?;
+                open_or_create_database_unchecked(opt, OnFailure::RemoveDb, zk).await?;
             match import_dump(&opt.db_path, path, &mut index_scheduler, &mut auth_controller) {
                 Ok(()) => (index_scheduler, auth_controller),
                 Err(e) => {
@@ -187,10 +191,10 @@ pub fn setup_meilisearch(opt: &Opt) -> anyhow::Result<(Arc<IndexScheduler>, Arc<
         // the dump and the db exist and we can ignore the dump because of the ignore_dump_if_db_exists flag
         // or, the dump is missing but we can ignore that because of the ignore_missing_dump flag
         } else {
-            open_or_create_database(opt, empty_db)?
+            open_or_create_database(opt, empty_db, zk).await?
         }
     } else {
-        open_or_create_database(opt, empty_db)?
+        open_or_create_database(opt, empty_db, zk).await?
     };
 
     // We create a loop in a thread that registers snapshotCreation tasks
@@ -214,13 +218,14 @@ pub fn setup_meilisearch(opt: &Opt) -> anyhow::Result<(Arc<IndexScheduler>, Arc<
 }
 
 /// Try to start the IndexScheduler and AuthController without checking the VERSION file or anything.
-fn open_or_create_database_unchecked(
+async fn open_or_create_database_unchecked(
     opt: &Opt,
     on_failure: OnFailure,
+    zk: Option<zk::Client>,
 ) -> anyhow::Result<(IndexScheduler, AuthController)> {
     // we don't want to create anything in the data.ms yet, thus we
     // wrap our two builders in a closure that'll be executed later.
-    let auth_controller = AuthController::new(&opt.db_path, &opt.master_key);
+    let auth_controller = AuthController::new(&opt.db_path, &opt.master_key, zk);
     let instance_features = opt.to_instance_features();
     let index_scheduler_builder = || -> anyhow::Result<_> {
         Ok(IndexScheduler::new(IndexSchedulerOptions {
@@ -245,7 +250,7 @@ fn open_or_create_database_unchecked(
 
     match (
         index_scheduler_builder(),
-        auth_controller.map_err(anyhow::Error::from),
+        auth_controller.await.map_err(anyhow::Error::from),
         create_version_file(&opt.db_path).map_err(anyhow::Error::from),
     ) {
         (Ok(i), Ok(a), Ok(())) => Ok((i, a)),
@@ -259,15 +264,16 @@ fn open_or_create_database_unchecked(
 }
 
 /// Ensure you're in a valid state and open the IndexScheduler + AuthController for you.
-fn open_or_create_database(
+async fn open_or_create_database(
     opt: &Opt,
     empty_db: bool,
+    zk: Option<zk::Client>,
 ) -> anyhow::Result<(IndexScheduler, AuthController)> {
     if !empty_db {
         check_version_file(&opt.db_path)?;
     }
 
-    open_or_create_database_unchecked(opt, OnFailure::KeepDb)
+    open_or_create_database_unchecked(opt, OnFailure::KeepDb, zk).await
 }
 
 fn import_dump(

meilisearch/src/main.rs

@@ -12,6 +12,7 @@ use meilisearch::analytics::Analytics;
 use meilisearch::{analytics, create_app, prototype_name, setup_meilisearch, Opt};
 use meilisearch_auth::{generate_master_key, AuthController, MASTER_KEY_MIN_SIZE};
 use termcolor::{Color, ColorChoice, ColorSpec, StandardStream, WriteColor};
+use zookeeper_client as zk;
 
 #[global_allocator]
 static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
@@ -63,7 +64,11 @@ async fn main() -> anyhow::Result<()> {
         _ => (),
     }
 
-    let (index_scheduler, auth_controller) = setup_meilisearch(&opt)?;
+    let zk = match opt.zk_url {
+        Some(ref url) => Some(zk::Client::connect(url).await.unwrap()),
+        None => None,
+    };
+    let (index_scheduler, auth_controller) = setup_meilisearch(&opt, zk).await?;
 
     #[cfg(all(not(debug_assertions), feature = "analytics"))]
     let analytics = if !opt.no_analytics {

meilisearch/src/option.rs

@@ -28,6 +28,7 @@ const MEILI_DB_PATH: &str = "MEILI_DB_PATH";
 const MEILI_HTTP_ADDR: &str = "MEILI_HTTP_ADDR";
 const MEILI_MASTER_KEY: &str = "MEILI_MASTER_KEY";
 const MEILI_ENV: &str = "MEILI_ENV";
+const MEILI_ZK_URL: &str = "MEILI_ZK_URL";
 #[cfg(all(not(debug_assertions), feature = "analytics"))]
 const MEILI_NO_ANALYTICS: &str = "MEILI_NO_ANALYTICS";
 const MEILI_HTTP_PAYLOAD_SIZE_LIMIT: &str = "MEILI_HTTP_PAYLOAD_SIZE_LIMIT";
@@ -154,6 +155,11 @@ pub struct Opt {
     #[serde(default = "default_env")]
     pub env: String,
 
+    /// Sets the HTTP address and port used to communicate with the zookeeper cluster.
+    /// If ran locally, the default url is `http://localhost:2181/`.
+    #[clap(long, env = MEILI_ZK_URL)]
+    pub zk_url: Option<String>,
+
     /// Deactivates Meilisearch's built-in telemetry when provided.
     ///
     /// Meilisearch automatically collects data from all instances that do not opt out using this flag.
@@ -368,6 +374,7 @@ impl Opt {
             http_addr,
             master_key,
             env,
+            zk_url,
             max_index_size: _,
             max_task_db_size: _,
             http_payload_size_limit,
@@ -401,6 +408,9 @@ impl Opt {
             export_to_env_if_not_present(MEILI_MASTER_KEY, master_key);
         }
         export_to_env_if_not_present(MEILI_ENV, env);
+        if let Some(zk_url) = zk_url {
+            export_to_env_if_not_present(MEILI_ZK_URL, zk_url);
+        }
         #[cfg(all(not(debug_assertions), feature = "analytics"))]
         {
             export_to_env_if_not_present(MEILI_NO_ANALYTICS, no_analytics.to_string());
@@ -715,6 +725,10 @@ fn default_env() -> String {
     DEFAULT_ENV.to_string()
 }
 
+pub fn default_zk_url() -> String {
+    DEFAULT_HTTP_ADDR.to_string()
+}
+
 fn default_max_index_size() -> Byte {
     Byte::from_bytes(INDEX_SIZE)
 }

meilisearch/src/routes/api_key.rs

@@ -41,14 +41,10 @@ pub async fn create_api_key(
     _req: HttpRequest,
 ) -> Result<HttpResponse, ResponseError> {
     let v = body.into_inner();
-    let res = tokio::task::spawn_blocking(move || -> Result<_, AuthControllerError> {
-        let key = auth_controller.create_key(v)?;
-        Ok(KeyView::from_key(key, &auth_controller))
-    })
-    .await
-    .map_err(|e| ResponseError::from_msg(e.to_string(), Code::Internal))??;
+    let key = auth_controller.create_key(v).await?;
+    let key = KeyView::from_key(key, &auth_controller);
 
-    Ok(HttpResponse::Created().json(res))
+    Ok(HttpResponse::Created().json(key))
 }
 
 #[derive(Deserr, Debug, Clone, Copy)]
@@ -110,17 +106,11 @@ pub async fn patch_api_key(
 ) -> Result<HttpResponse, ResponseError> {
     let key = path.into_inner().key;
     let patch_api_key = body.into_inner();
-    let res = tokio::task::spawn_blocking(move || -> Result<_, AuthControllerError> {
-        let uid =
-            Uuid::parse_str(&key).or_else(|_| auth_controller.get_uid_from_encoded_key(&key))?;
-        let key = auth_controller.update_key(uid, patch_api_key)?;
+    let uid = Uuid::parse_str(&key).or_else(|_| auth_controller.get_uid_from_encoded_key(&key))?;
+    let key = auth_controller.update_key(uid, patch_api_key).await?;
+    let key = KeyView::from_key(key, &auth_controller);
 
-        Ok(KeyView::from_key(key, &auth_controller))
-    })
-    .await
-    .map_err(|e| ResponseError::from_msg(e.to_string(), Code::Internal))??;
-
-    Ok(HttpResponse::Ok().json(res))
+    Ok(HttpResponse::Ok().json(key))
 }
 
 pub async fn delete_api_key(
@@ -128,13 +118,8 @@ pub async fn delete_api_key(
     path: web::Path<AuthParam>,
 ) -> Result<HttpResponse, ResponseError> {
     let key = path.into_inner().key;
-    tokio::task::spawn_blocking(move || {
-        let uid =
-            Uuid::parse_str(&key).or_else(|_| auth_controller.get_uid_from_encoded_key(&key))?;
-        auth_controller.delete_key(uid)
-    })
-    .await
-    .map_err(|e| ResponseError::from_msg(e.to_string(), Code::Internal))??;
+    let uid = Uuid::parse_str(&key).or_else(|_| auth_controller.get_uid_from_encoded_key(&key))?;
+    auth_controller.delete_key(uid).await?;
 
     Ok(HttpResponse::NoContent().finish())
 }

meilisearch/tests/common/server.rs

@@ -39,7 +39,7 @@ impl Server {
 
         let options = default_settings(dir.path());
 
-        let (index_scheduler, auth) = setup_meilisearch(&options).unwrap();
+        let (index_scheduler, auth) = setup_meilisearch(&options, None).await.unwrap();
         let service = Service { index_scheduler, auth, options, api_key: None };
 
         Server { service, _dir: Some(dir) }
@@ -54,7 +54,7 @@ impl Server {
 
         options.master_key = Some("MASTER_KEY".to_string());
 
-        let (index_scheduler, auth) = setup_meilisearch(&options).unwrap();
+        let (index_scheduler, auth) = setup_meilisearch(&options, None).await.unwrap();
         let service = Service { index_scheduler, auth, options, api_key: None };
 
         Server { service, _dir: Some(dir) }
@@ -67,7 +67,7 @@ impl Server {
     }
 
     pub async fn new_with_options(options: Opt) -> Result<Self, anyhow::Error> {
-        let (index_scheduler, auth) = setup_meilisearch(&options)?;
+        let (index_scheduler, auth) = setup_meilisearch(&options, None).await?;
         let service = Service { index_scheduler, auth, options, api_key: None };
 
         Ok(Server { service, _dir: None })