Fix the update system to always consume updates even if failing

meilidb-core/Cargo.toml

@@ -12,7 +12,7 @@ crossbeam-channel = "0.3.9"
 deunicode = "1.0.0"
 env_logger = "0.7.0"
 hashbrown = { version = "0.6.0", features = ["serde"] }
-heed = "0.3.0"
+heed = "0.3.1"
 log = "0.4.8"
 meilidb-schema = { path = "../meilidb-schema", version = "0.6.0" }
 meilidb-tokenizer = { path = "../meilidb-tokenizer", version = "0.6.0" }

meilidb-core/src/database.rs

@@ -28,34 +28,63 @@ fn update_awaiter(receiver: Receiver<()>, env: heed::Env, update_fn: Arc<ArcSwap
             let mut writer = match env.write_txn() {
                 Ok(writer) => writer,
                 Err(e) => {
-                    error!("LMDB writer transaction begin failed: {}", e);
+                    error!("LMDB write transaction begin failed: {}", e);
                     break;
                 }
             };
 
-            match update::update_task(&mut writer, index.clone()) {
-                Ok(Some(status)) => {
-                    match status.result {
-                        Ok(_) => {
-                            if let Err(e) = writer.commit() {
-                                error!("update transaction failed: {}", e)
-                            }
-                        }
-                        Err(_) => writer.abort(),
-                    }
-
-                    if let Some(ref callback) = *update_fn.load() {
-                        (callback)(status);
-                    }
-                }
-                // no more updates to handle for now
+            let (update_id, update) = match index.updates.pop_front(&mut writer) {
+                Ok(Some(value)) => value,
                 Ok(None) => {
                     debug!("no more updates");
                     writer.abort();
                     break;
                 }
+                Err(e) => {
+                    error!("pop front update failed: {}", e);
+                    break;
+                }
+            };
+
+            let mut nested_writer = match unsafe { env.nested_write_txn(&writer) } {
+                Ok(writer) => writer,
+                Err(e) => {
+                    error!("LMDB nested write transaction begin failed: {}", e);
+                    break;
+                }
+            };
+
+            match update::update_task(&mut nested_writer, index.clone(), update_id, update) {
+                Ok(status) => {
+                    match &status.result {
+                        Ok(_) => {
+                            if let Err(e) = nested_writer.commit() {
+                                error!("update nested transaction failed: {}", e);
+                            }
+                        }
+                        Err(_) => nested_writer.abort(),
+                    }
+
+                    let result =
+                        index
+                            .updates_results
+                            .put_update_result(&mut writer, update_id, &status);
+
+                    if let Err(e) = result {
+                        error!("update result store commit failed: {}", e);
+                    }
+
+                    if let Err(e) = writer.commit() {
+                        error!("update parent transaction failed: {}", e);
+                    }
+
+                    if let Some(ref callback) = *update_fn.load() {
+                        (callback)(status);
+                    }
+                }
                 Err(e) => {
                     error!("update task failed: {}", e);
+                    nested_writer.abort();
                     writer.abort()
                 }
             }

meilidb-core/src/update/mod.rs

@@ -152,12 +152,9 @@ pub fn next_update_id(
 pub fn update_task(
     writer: &mut heed::RwTxn,
     index: store::Index,
-) -> MResult<Option<ProcessedUpdateResult>> {
-    let (update_id, update) = match index.updates.pop_front(writer)? {
-        Some(value) => value,
-        None => return Ok(None),
-    };
-
+    update_id: u64,
+    update: Update,
+) -> MResult<ProcessedUpdateResult> {
     debug!("Processing update number {}", update_id);
 
     let (update_type, result, duration) = match update {
@@ -308,9 +305,5 @@ pub fn update_task(
         detailed_duration,
     };
 
-    index
-        .updates_results
-        .put_update_result(writer, update_id, &status)?;
-
-    Ok(Some(status))
+    Ok(status)
 }

meilidb-http/Cargo.toml

@@ -13,7 +13,7 @@ chrono = { version = "0.4.9", features = ["serde"] }
 crossbeam-channel = "0.3.9"
 envconfig = "0.5.1"
 envconfig_derive = "0.5.1"
-heed = "0.3.0"
+heed = "0.3.1"
 http = "0.1.19"
 indexmap = { version = "1.3.0", features = ["serde-1"] }
 jemallocator = "0.3.2"