Merge branch 'main' into trinity/increase-stack-size-coverage

.github/workflows/coverage.yml

@@ -162,3 +162,32 @@ jobs:
         with:
           token: ${{ secrets.CODECOV_TOKEN }} # not required for public repos
           files: ./quickwit/lcov.info
+
+  on-failure:
+    if: ${{ github.repository_owner == 'quickwit-oss' && failure() }}
+    name: On Failure
+    needs: [test]
+    runs-on: ubuntu-latest
+    steps:
+      - name: Send Message
+        uses: sarisia/actions-status-discord@v1
+        with:
+          webhook: ${{ secrets.DISCORD_WEBHOOK }}
+          nodetail: true
+          color: "#FF0000"
+          title: ""
+          description: |
+            ### ❌ [${{ github.event.pull_request.title }}](${{ github.event.pull_request.html_url }})
+
+            @${{ github.actor }} quickwit coverage CI failed on your PR.
+
+            Coverage CI contains tests that are not running in the regular CI because they are too lengthy.
+            For this reason it is possible for it to break even if the tests were passing on your PR.
+            This is not a catastrophy, but you are responsible for fixing it!
+
+            You can run the full test suite locally with `make test-all`.
+
+            Please report in this channel that you are working on it/fixed it/or if it is a flaky test/
+            or if you need help.
+
+            **[View logs](https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }})**

docker-compose.yml

@@ -77,7 +77,7 @@ services:
       - "${MAP_HOST_PULSAR:-127.0.0.1}:6650:6650"
       - "${MAP_HOST_PULSAR:-127.0.0.1}:8081:8080"
     environment:
-      PULSAR_MEM: "-Xms256M -Xmx256M"
+      PULSAR_MEM: "-Xms384M -Xmx384M"
     profiles:
       - all
       - pulsar

quickwit/Cargo.toml

@@ -155,7 +155,6 @@ mockall = "0.11"
 mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "306c0a7" }
 new_string_template = "1.5.1"
 nom = "7.1.3"
-num_cpus = "1"
 numfmt = "1.1.1"
 once_cell = "1"
 oneshot = "0.1.5"
@@ -203,8 +202,8 @@ reqwest = { version = "0.11", default-features = false, features = [
   "rustls-tls",
 ] }
 rust-embed = "6.8.1"
-sea-query = { version = "0" }
-sea-query-binder = { version = "0", features = [
+sea-query = { version = "0.30" }
+sea-query-binder = { version = "0.5", features = [
   "runtime-tokio-rustls",
   "sqlx-postgres",
 ] }
@@ -322,7 +321,7 @@ quickwit-serve = { path = "quickwit-serve" }
 quickwit-storage = { path = "quickwit-storage" }
 quickwit-telemetry = { path = "quickwit-telemetry" }
 
-tantivy = { git = "https://github.com/quickwit-oss/tantivy/", rev = "92b5526", default-features = false, features = [
+tantivy = { git = "https://github.com/quickwit-oss/tantivy/", rev = "1ee5f907", default-features = false, features = [
   "lz4-compression",
   "mmap",
   "quickwit",

quickwit/quickwit-cluster/src/cluster.rs

@@ -418,15 +418,11 @@ impl Cluster {
     /// - value: Number of indexing tasks in the group.
     /// Keys present in chitchat state but not in the given `indexing_tasks` are marked for
     /// deletion.
-    pub async fn update_self_node_indexing_tasks(
-        &self,
-        indexing_tasks: &[IndexingTask],
-    ) -> anyhow::Result<()> {
+    pub async fn update_self_node_indexing_tasks(&self, indexing_tasks: &[IndexingTask]) {
         let chitchat = self.chitchat().await;
         let mut chitchat_guard = chitchat.lock().await;
         let node_state = chitchat_guard.self_node_state();
         set_indexing_tasks_in_node_state(indexing_tasks, node_state);
-        Ok(())
     }
 
     pub async fn chitchat(&self) -> Arc<Mutex<Chitchat>> {
@@ -961,8 +957,7 @@ mod tests {
             .await;
         cluster2
             .update_self_node_indexing_tasks(&[indexing_task1.clone(), indexing_task2.clone()])
-            .await
-            .unwrap();
+            .await;
         cluster1
             .wait_for_ready_members(|members| members.len() == 2, Duration::from_secs(30))
             .await
@@ -1042,8 +1037,7 @@ mod tests {
             .collect_vec();
         cluster1
             .update_self_node_indexing_tasks(&indexing_tasks)
-            .await
-            .unwrap();
+            .await;
         for cluster in [&cluster2, &cluster3] {
             let cluster_clone = cluster.clone();
             let indexing_tasks_clone = indexing_tasks.clone();
@@ -1063,7 +1057,7 @@ mod tests {
         }
 
         // Mark tasks for deletion.
-        cluster1.update_self_node_indexing_tasks(&[]).await.unwrap();
+        cluster1.update_self_node_indexing_tasks(&[]).await;
         for cluster in [&cluster2, &cluster3] {
             let cluster_clone = cluster.clone();
             wait_until_predicate(
@@ -1084,8 +1078,7 @@ mod tests {
         // Re-add tasks.
         cluster1
             .update_self_node_indexing_tasks(&indexing_tasks)
-            .await
-            .unwrap();
+            .await;
         for cluster in [&cluster2, &cluster3] {
             let cluster_clone = cluster.clone();
             let indexing_tasks_clone = indexing_tasks.clone();

quickwit/quickwit-common/Cargo.toml

@@ -25,7 +25,6 @@ hostname = { workspace = true }
 http = { workspace = true }
 hyper = { workspace = true }
 itertools = { workspace = true }
-num_cpus = { workspace = true }
 once_cell = { workspace = true }
 pin-project = { workspace = true }
 pnet = { workspace = true }

quickwit/quickwit-common/src/lib.rs

@@ -82,13 +82,13 @@ pub fn split_file(split_id: impl Display) -> String {
 pub fn get_from_env<T: FromStr + Debug>(key: &str, default_value: T) -> T {
     if let Ok(value_str) = std::env::var(key) {
         if let Ok(value) = T::from_str(&value_str) {
-            info!(value=?value, "Setting `{}` from environment", key);
+            info!(value=?value, "setting `{}` from environment", key);
             return value;
         } else {
-            error!(value_str=%value_str, "Failed to parse `{}` from environment", key);
+            error!(value_str=%value_str, "failed to parse `{}` from environment", key);
         }
     }
-    info!(value=?default_value, "Setting `{}` from default", key);
+    info!(value=?default_value, "setting `{}` from default", key);
     default_value
 }
 
@@ -173,6 +173,18 @@ pub const fn div_ceil(lhs: i64, rhs: i64) -> i64 {
     }
 }
 
+/// Return the number of vCPU/hyperthreads available.
+/// This number is usually not equal to the number of cpu cores
+pub fn num_cpus() -> usize {
+    match std::thread::available_parallelism() {
+        Ok(num_cpus) => num_cpus.get(),
+        Err(io_error) => {
+            error!(errror=?io_error, "failed to detect the number of threads available: arbitrarily returning 2");
+            2
+        }
+    }
+}
+
 // The following are helpers to build named tasks.
 //
 // Named tasks require the tokio feature `tracing` to be enabled.