WIP: dump

meilisearch-http/src/index_controller/update_actor/actor.rs

@@ -91,7 +91,6 @@ where
         meta: UpdateMeta,
         mut payload: mpsc::Receiver<PayloadData<D>>,
     ) -> Result<UpdateStatus> {
-
         let file_path = match meta {
             UpdateMeta::DocumentsAddition { .. }
             | UpdateMeta::DeleteDocuments => {

meilisearch-http/src/index_controller/update_actor/mod.rs

@@ -13,9 +13,8 @@ use crate::index_controller::{UpdateMeta, UpdateStatus};
 
 use actor::UpdateActor;
 use message::UpdateMsg;
-use update_store::UpdateStore;
-pub use update_store::UpdateStoreInfo;
 
+pub use update_store::{UpdateStore, UpdateStoreInfo};
 pub use handle_impl::UpdateActorHandleImpl;
 
 pub type Result<T> = std::result::Result<T, UpdateError>;

meilisearch-http/src/index_controller/update_actor/update_store.rs

@@ -177,11 +177,7 @@ pub struct UpdateStore {
 }
 
 impl UpdateStore {
-    pub fn open(
-        mut options: EnvOpenOptions,
-        path: impl AsRef<Path>,
-        index_handle: impl IndexActorHandle + Clone + Sync + Send + 'static,
-    ) -> anyhow::Result<Arc<Self>> {
+    pub fn create(mut options: EnvOpenOptions, path: impl AsRef<Path>) -> anyhow::Result<(Self, mpsc::Receiver<()>)> {
         options.max_dbs(5);
 
         let env = options.open(path)?;
@@ -189,21 +185,30 @@ impl UpdateStore {
         let next_update_id = env.create_database(Some("next-update-id"))?;
         let updates = env.create_database(Some("updates"))?;
 
-        let (notification_sender, mut notification_receiver) = mpsc::channel(10);
+        let state = Arc::new(StateLock::from_state(State::Idle));
+
+        let (notification_sender, notification_receiver) = mpsc::channel(10);
         // Send a first notification to trigger the process.
         let _ = notification_sender.send(());
 
-        let state = Arc::new(StateLock::from_state(State::Idle));
+        Ok((Self { env, pending_queue, next_update_id, updates, state, notification_sender }, notification_receiver))
+    }
+
+    pub fn open(
+        options: EnvOpenOptions,
+        path: impl AsRef<Path>,
+        index_handle: impl IndexActorHandle + Clone + Sync + Send + 'static,
+    ) -> anyhow::Result<Arc<Self>> {
+        let (update_store, mut notification_receiver) = Self::create(options, path)?;
+        let update_store = Arc::new(update_store);
 
         // Init update loop to perform any pending updates at launch.
         // Since we just launched the update store, and we still own the receiving end of the
         // channel, this call is guaranteed to succeed.
-        notification_sender
+        update_store.notification_sender
             .try_send(())
             .expect("Failed to init update store");
 
-        let update_store = Arc::new(UpdateStore { env, pending_queue, next_update_id, updates, state, notification_sender });
-
         // We need a weak reference so we can take ownership on the arc later when we
         // want to close the index.
         let update_store_weak = Arc::downgrade(&update_store);
@@ -283,6 +288,18 @@ impl UpdateStore {
         Ok(meta)
     }
 
+    /// Push already processed updates in the UpdateStore. This is useful for the dumps
+    pub fn register_already_processed_update (
+        &self,
+        result: UpdateStatus,
+        index_uuid: Uuid,
+    ) -> heed::Result<()> {
+        let mut wtxn = self.env.write_txn()?;
+        let (_global_id, update_id) = self.next_update_id(&mut wtxn, index_uuid)?;
+        self.updates.remap_key_type::<UpdateKeyCodec>().put(&mut wtxn, &(index_uuid, update_id), &result)?;
+        wtxn.commit()
+    }
+
     /// 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
     /// only writing the result meta to the processed-meta store *after* it has been processed.