feat: ElasticSearch backend (#6903)

* (do not merge) search test branch

* perf

* fix up search pagination

* more parity and perf work

* expand what parity does

* fix author query

* approach parity even without explicit parity enabled

* remove old parity code

* fix shear

* fix

* fmt

apps/labrinth/.env.docker-compose

@@ -17,14 +17,14 @@ DATABASE_URL=postgresql://labrinth:labrinth@labrinth-postgres/labrinth
 DATABASE_MIN_CONNECTIONS=0
 DATABASE_MAX_CONNECTIONS=16
 
-SEARCH_BACKEND=typesense
+SEARCH_BACKEND=elasticsearch
 MEILISEARCH_READ_ADDR=http://localhost:7700
 MEILISEARCH_WRITE_ADDRS=http://localhost:7700
 MEILISEARCH_KEY=modrinth
-ELASTICSEARCH_URL=http://localhost:9200
+ELASTICSEARCH_URL=http://elasticsearch0:9200
 ELASTICSEARCH_INDEX_PREFIX=labrinth
-ELASTICSEARCH_USERNAME=elastic
-ELASTICSEARCH_PASSWORD=elastic
+ELASTICSEARCH_USERNAME=
+ELASTICSEARCH_PASSWORD=
 SEARCH_INDEX_CHUNK_SIZE=5000
 SEARCH_INCREMENTAL_INDEX_BATCH_DELAY_SECONDS=5
 SEARCH_INCREMENTAL_INDEX_BATCH_MAX_SIZE=1000

apps/labrinth/.env.local

@@ -17,7 +17,7 @@ DATABASE_URL=postgresql://labrinth:labrinth@localhost/labrinth
 DATABASE_MIN_CONNECTIONS=0
 DATABASE_MAX_CONNECTIONS=16
 
-SEARCH_BACKEND=typesense
+SEARCH_BACKEND=elasticsearch
 
 # Meilisearch configuration
 MEILISEARCH_READ_ADDR=http://localhost:7700
@@ -32,7 +32,7 @@ ELASTICSEARCH_INDEX_PREFIX=labrinth
 # MEILISEARCH_READ_ADDR=http://localhost:7710
 # MEILISEARCH_WRITE_ADDRS=http://localhost:7700,http://localhost:7701
 
-SEARCH_BACKEND=typesense
+SEARCH_BACKEND=elasticsearch
 
 MEILISEARCH_KEY=modrinth
 MEILISEARCH_META_NAMESPACE=

apps/labrinth/src/env.rs

@@ -237,6 +237,11 @@ vars! {
     SEARCH_TYPESENSE_DEFAULT_BUCKETING: Json<crate::search::backend::typesense::Bucketing> =
         Json(crate::search::backend::typesense::Bucketing::Buckets(5));
     SEARCH_TYPESENSE_DEFAULT_MAX_CANDIDATES: usize = 24usize;
+    ELASTICSEARCH_URL: String = "http://localhost:9200";
+    ELASTICSEARCH_INDEX_PREFIX: String = "labrinth";
+    ELASTICSEARCH_USERNAME: String = "";
+    ELASTICSEARCH_PASSWORD: String = "";
+    ELASTICSEARCH_BULK_BATCH_SIZE: usize = 1000usize;
 
     // storage
     STORAGE_BACKEND: crate::file_hosting::FileHostKind = crate::file_hosting::FileHostKind::Local;

apps/labrinth/src/search/backend/elasticsearch/filter.rs

@@ -0,0 +1,390 @@
+use eyre::{Result, eyre};
+use serde_json::{Value, json};
+
+use crate::search::filter::{
+    FilterComparison, FilterCondition, FilterExpr, FilterLiteral,
+    FilterPredicate,
+};
+use crate::search::indexing::normalize_for_search;
+
+const MAX_DNF_CLAUSES: usize = 64;
+const MAX_FILTER_DEPTH: usize = 64;
+const MAX_FILTER_NODES: usize = 1024;
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum FilterScope {
+    Project,
+    Version,
+    Mixed,
+}
+
+pub(super) struct ElasticsearchFilter {
+    pub query: Value,
+    pub has_version_filter: bool,
+}
+
+pub(super) fn serialize_filter(
+    filter: &FilterExpr,
+) -> Result<ElasticsearchFilter> {
+    let (nodes, depth) = filter_complexity(filter);
+    if nodes > MAX_FILTER_NODES {
+        return Err(eyre!("search filter has too many expressions"));
+    }
+    if depth > MAX_FILTER_DEPTH {
+        return Err(eyre!("search filter is nested too deeply"));
+    }
+
+    let mut inner_hits_index = 0;
+    let query = plan(filter, &mut inner_hits_index)?;
+    Ok(ElasticsearchFilter {
+        query,
+        has_version_filter: inner_hits_index != 0,
+    })
+}
+
+fn plan(filter: &FilterExpr, inner_hits_index: &mut usize) -> Result<Value> {
+    match filter_scope(filter) {
+        FilterScope::Project => lower(filter),
+        FilterScope::Version => version_query(lower(filter)?, inner_hits_index),
+        FilterScope::Mixed => plan_mixed(filter, inner_hits_index),
+    }
+}
+
+fn plan_mixed(
+    filter: &FilterExpr,
+    inner_hits_index: &mut usize,
+) -> Result<Value> {
+    match filter {
+        FilterExpr::Or(expressions) => expressions
+            .iter()
+            .map(|expression| plan(expression, inner_hits_index))
+            .collect::<Result<Vec<_>>>()
+            .map(or_query),
+        FilterExpr::And(expressions)
+            if expressions.iter().all(|expression| {
+                filter_scope(expression) != FilterScope::Mixed
+            }) =>
+        {
+            plan_partitioned_and(expressions, inner_hits_index)
+        }
+        _ => {
+            let clauses = to_dnf(filter)?;
+            clauses
+                .into_iter()
+                .map(|clause| plan_clause(clause, inner_hits_index))
+                .collect::<Result<Vec<_>>>()
+                .map(or_query)
+        }
+    }
+}
+
+fn plan_partitioned_and(
+    expressions: &[FilterExpr],
+    inner_hits_index: &mut usize,
+) -> Result<Value> {
+    let mut project = Vec::new();
+    let mut version = Vec::new();
+    for expression in expressions {
+        match filter_scope(expression) {
+            FilterScope::Project => project.push(lower(expression)?),
+            FilterScope::Version => version.push(lower(expression)?),
+            FilterScope::Mixed => {
+                return Err(eyre!("could not partition mixed search filter"));
+            }
+        }
+    }
+    if !version.is_empty() {
+        project.push(version_query(and_query(version), inner_hits_index)?);
+    }
+    Ok(and_query(project))
+}
+
+fn plan_clause(
+    predicates: Vec<&FilterPredicate>,
+    inner_hits_index: &mut usize,
+) -> Result<Value> {
+    let mut project = Vec::new();
+    let mut version = Vec::new();
+    for predicate in predicates {
+        let query = predicate_query(predicate)?;
+        if is_version_filter_field(predicate.field.as_str()) {
+            version.push(query);
+        } else {
+            project.push(query);
+        }
+    }
+    if !version.is_empty() {
+        project.push(version_query(and_query(version), inner_hits_index)?);
+    }
+    Ok(and_query(project))
+}
+
+fn lower(filter: &FilterExpr) -> Result<Value> {
+    match filter {
+        FilterExpr::And(expressions) => expressions
+            .iter()
+            .map(lower)
+            .collect::<Result<Vec<_>>>()
+            .map(and_query),
+        FilterExpr::Or(expressions) => expressions
+            .iter()
+            .map(lower)
+            .collect::<Result<Vec<_>>>()
+            .map(or_query),
+        FilterExpr::Predicate(predicate) => predicate_query(predicate),
+        FilterExpr::Not(_) => {
+            Err(eyre!("search filter contains an unnormalized negation"))
+        }
+    }
+}
+
+fn predicate_query(predicate: &FilterPredicate) -> Result<Value> {
+    let source_field = predicate.field.as_str();
+    let field = exact_field(source_field);
+    match &predicate.condition {
+        FilterCondition::Compare { comparison, value } => {
+            let value = literal_value(source_field, value)?;
+            Ok(match comparison {
+                FilterComparison::Equal => {
+                    json!({"term": {(field): {"value": value}}})
+                }
+                FilterComparison::NotEqual => not_query(json!({
+                    "term": {(field): {"value": value}}
+                })),
+                FilterComparison::GreaterThan => {
+                    json!({"range": {(field): {"gt": value}}})
+                }
+                FilterComparison::GreaterThanOrEqual => {
+                    json!({"range": {(field): {"gte": value}}})
+                }
+                FilterComparison::LessThan => {
+                    json!({"range": {(field): {"lt": value}}})
+                }
+                FilterComparison::LessThanOrEqual => {
+                    json!({"range": {(field): {"lte": value}}})
+                }
+            })
+        }
+        FilterCondition::In { values, negated } => {
+            let values = values
+                .iter()
+                .map(|value| literal_value(source_field, value))
+                .collect::<Result<Vec<_>>>()?;
+            let query = json!({"terms": {(field): values}});
+            Ok(if *negated { not_query(query) } else { query })
+        }
+        FilterCondition::Exists { negated } => {
+            let query = json!({"exists": {"field": field}});
+            Ok(if *negated { not_query(query) } else { query })
+        }
+    }
+}
+
+fn literal_value(field: &str, literal: &FilterLiteral) -> Result<Value> {
+    match literal {
+        FilterLiteral::String(value) if field == "author" => {
+            Ok(Value::String(normalize_for_search(value)))
+        }
+        FilterLiteral::String(value) => Ok(Value::String(value.clone())),
+        FilterLiteral::Number(value) => serde_json::from_str(value)
+            .map_err(|error| eyre!("invalid numeric filter literal: {error}")),
+        FilterLiteral::Bool(value) => Ok(Value::Bool(*value)),
+    }
+}
+
+fn exact_field(field: &str) -> &str {
+    match field {
+        "name" => "name.keyword",
+        "author" => "indexed_author.keyword",
+        "summary" => "summary.keyword",
+        "slug" => "slug.keyword",
+        _ => field,
+    }
+}
+
+fn version_query(query: Value, inner_hits_index: &mut usize) -> Result<Value> {
+    let name = format!("matching_versions_{}", *inner_hits_index);
+    *inner_hits_index += 1;
+    Ok(json!({
+        "has_child": {
+            "type": "version",
+            "score_mode": "none",
+            "query": query,
+            "inner_hits": {
+                "name": name,
+                "size": 1,
+                "_source": ["version_id", "version_published_timestamp"],
+                "sort": [
+                    {"version_published_timestamp": {"order": "desc"}},
+                    {"version_id": {"order": "desc"}}
+                ]
+            }
+        }
+    }))
+}
+
+fn and_query(queries: Vec<Value>) -> Value {
+    match queries.len() {
+        0 => json!({"match_all": {}}),
+        1 => queries.into_iter().next().unwrap_or_default(),
+        _ => json!({"bool": {"filter": queries}}),
+    }
+}
+
+fn or_query(queries: Vec<Value>) -> Value {
+    match queries.len() {
+        0 => json!({"match_none": {}}),
+        1 => queries.into_iter().next().unwrap_or_default(),
+        _ => json!({
+            "bool": {
+                "should": queries,
+                "minimum_should_match": 1
+            }
+        }),
+    }
+}
+
+fn not_query(query: Value) -> Value {
+    json!({
+        "bool": {
+            "must": [{"match_all": {}}],
+            "must_not": [query]
+        }
+    })
+}
+
+fn filter_scope(filter: &FilterExpr) -> FilterScope {
+    match filter {
+        FilterExpr::Predicate(predicate) => {
+            if is_version_filter_field(predicate.field.as_str()) {
+                FilterScope::Version
+            } else {
+                FilterScope::Project
+            }
+        }
+        FilterExpr::And(expressions) | FilterExpr::Or(expressions) => {
+            let mut scopes = expressions.iter().map(filter_scope);
+            let Some(first) = scopes.next() else {
+                return FilterScope::Project;
+            };
+            if scopes.all(|scope| scope == first) {
+                first
+            } else {
+                FilterScope::Mixed
+            }
+        }
+        FilterExpr::Not(expression) => filter_scope(expression),
+    }
+}
+
+fn is_version_filter_field(field: &str) -> bool {
+    matches!(
+        field,
+        "categories"
+            | "project_types"
+            | "environment"
+            | "game_versions"
+            | "client_side"
+            | "server_side"
+    )
+}
+
+fn to_dnf(filter: &FilterExpr) -> Result<Vec<Vec<&FilterPredicate>>> {
+    match filter {
+        FilterExpr::Predicate(predicate) => Ok(vec![vec![predicate]]),
+        FilterExpr::Or(expressions) => {
+            let mut clauses = Vec::new();
+            for expression in expressions.iter() {
+                clauses.extend(to_dnf(expression)?);
+                if clauses.len() > MAX_DNF_CLAUSES {
+                    return Err(eyre!(
+                        "search filter has too many boolean clauses"
+                    ));
+                }
+            }
+            Ok(clauses)
+        }
+        FilterExpr::And(expressions) => {
+            let mut clauses = vec![Vec::new()];
+            for expression in expressions.iter() {
+                let right = to_dnf(expression)?;
+                if clauses.len().saturating_mul(right.len()) > MAX_DNF_CLAUSES {
+                    return Err(eyre!(
+                        "search filter has too many boolean clauses"
+                    ));
+                }
+                clauses = clauses
+                    .into_iter()
+                    .flat_map(|left| {
+                        right.iter().map(move |right| {
+                            let mut clause = left.clone();
+                            clause.extend(right);
+                            clause
+                        })
+                    })
+                    .collect();
+            }
+            Ok(clauses)
+        }
+        FilterExpr::Not(_) => {
+            Err(eyre!("search filter contains an unnormalized negation"))
+        }
+    }
+}
+
+fn filter_complexity(filter: &FilterExpr) -> (usize, usize) {
+    match filter {
+        FilterExpr::Predicate(_) => (1, 1),
+        FilterExpr::And(expressions) | FilterExpr::Or(expressions) => {
+            expressions.iter().map(filter_complexity).fold(
+                (1, 1),
+                |(nodes, depth), (child_nodes, child_depth)| {
+                    (nodes + child_nodes, depth.max(child_depth + 1))
+                },
+            )
+        }
+        FilterExpr::Not(expression) => {
+            let (nodes, depth) = filter_complexity(expression);
+            (nodes + 1, depth + 1)
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::serialize_filter;
+    use crate::search::filter::{normalize, parse_expression};
+    use serde_json::Value;
+
+    fn serialize(input: &str) -> Value {
+        let filter = normalize(parse_expression(input).unwrap());
+        serialize_filter(&filter).unwrap().query
+    }
+
+    #[test]
+    fn correlated_version_filters_use_one_join() {
+        let query = serialize("categories = fabric AND game_versions = 1.21");
+        assert_eq!(query.to_string().matches("has_child").count(), 1);
+    }
+
+    #[test]
+    fn project_filters_do_not_use_a_join() {
+        let query = serialize("license = MIT");
+        assert_eq!(query.to_string().matches("has_child").count(), 0);
+        assert_eq!(query["term"]["license"]["value"], "MIT");
+    }
+
+    #[test]
+    fn author_filters_use_the_normalized_exact_field() {
+        let query = serialize("author = User");
+        assert_eq!(query["term"]["indexed_author.keyword"]["value"], "user");
+    }
+
+    #[test]
+    fn mixed_boolean_filters_preserve_version_correlation() {
+        let query = serialize(
+            "(license = MIT OR categories = fabric) AND game_versions = 1.21",
+        );
+        assert_eq!(query.to_string().matches("has_child").count(), 2);
+    }
+}

apps/labrinth/src/search/backend/mod.rs

@@ -1,8 +1,10 @@
 mod common;
+pub mod elasticsearch;
 pub mod typesense;
 
 pub use common::{
     ParsedSearchRequest, SearchIndex, SearchSort, combined_search_filters,
     parse_search_index, parse_search_request,
 };
+pub use elasticsearch::{Elasticsearch, ElasticsearchConfig};
 pub use typesense::{Typesense, TypesenseConfig};

apps/labrinth/src/search/indexing.rs

@@ -45,7 +45,7 @@ struct PartialProject {
     components: exp::ProjectSerial,
 }
 
-fn normalize_for_search(s: &str) -> String {
+pub(crate) fn normalize_for_search(s: &str) -> String {
     static SPECIAL_CHARS_RE: LazyLock<Regex> =
         LazyLock::new(|| Regex::new(r"[^a-zA-Z0-9-.\s]").expect("valid regex"));
 

apps/labrinth/src/search/mod.rs

@@ -189,6 +189,7 @@ pub enum TasksCancelFilter {
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum SearchBackendKind {
     Typesense,
+    Elasticsearch,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, strum::EnumIter)]
@@ -224,6 +225,7 @@ impl FromStr for SearchBackendKind {
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         Ok(match s {
             "typesense" => SearchBackendKind::Typesense,
+            "elasticsearch" => SearchBackendKind::Elasticsearch,
             _ => return Err(InvalidSearchBackendKind),
         })
     }
@@ -438,5 +440,9 @@ pub fn backend(meta_namespace: Option<String>) -> Box<dyn SearchBackend> {
             let config = backend::TypesenseConfig::new(meta_namespace);
             Box::new(backend::Typesense::new(config))
         }
+        SearchBackendKind::Elasticsearch => {
+            let config = backend::ElasticsearchConfig::new(meta_namespace);
+            Box::new(backend::Elasticsearch::new(config))
+        }
     }
 }

docker-compose.yml

@@ -32,6 +32,102 @@ services:
       interval: 3s
       timeout: 5s
       retries: 3
+  elasticsearch0:
+    image: docker.elastic.co/elasticsearch/elasticsearch:9.4.4
+    container_name: labrinth-elasticsearch0
+    restart: on-failure
+    networks:
+      - elasticsearch-mesh
+    ports:
+      - '127.0.0.1:9200:9200'
+    volumes:
+      - elasticsearch0-data:/usr/share/elasticsearch/data
+    environment:
+      node.name: elasticsearch0
+      cluster.name: labrinth-elasticsearch
+      discovery.seed_hosts: elasticsearch0,elasticsearch1,elasticsearch2
+      cluster.initial_master_nodes: elasticsearch0,elasticsearch1,elasticsearch2
+      bootstrap.memory_lock: 'true'
+      xpack.security.enabled: 'false'
+      xpack.security.enrollment.enabled: 'false'
+      ES_JAVA_OPTS: -Xms512m -Xmx512m
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+    healthcheck:
+      test:
+        [
+          'CMD-SHELL',
+          'curl --fail http://localhost:9200/_cluster/health?wait_for_status=yellow',
+        ]
+      interval: 5s
+      timeout: 5s
+      retries: 30
+  elasticsearch1:
+    image: docker.elastic.co/elasticsearch/elasticsearch:9.4.4
+    container_name: labrinth-elasticsearch1
+    restart: on-failure
+    networks:
+      - elasticsearch-mesh
+    ports:
+      - '127.0.0.1:9201:9200'
+    volumes:
+      - elasticsearch1-data:/usr/share/elasticsearch/data
+    environment:
+      node.name: elasticsearch1
+      cluster.name: labrinth-elasticsearch
+      discovery.seed_hosts: elasticsearch0,elasticsearch1,elasticsearch2
+      cluster.initial_master_nodes: elasticsearch0,elasticsearch1,elasticsearch2
+      bootstrap.memory_lock: 'true'
+      xpack.security.enabled: 'false'
+      xpack.security.enrollment.enabled: 'false'
+      ES_JAVA_OPTS: -Xms512m -Xmx512m
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+    healthcheck:
+      test:
+        [
+          'CMD-SHELL',
+          'curl --fail http://localhost:9200/_cluster/health?wait_for_status=yellow',
+        ]
+      interval: 5s
+      timeout: 5s
+      retries: 30
+  elasticsearch2:
+    image: docker.elastic.co/elasticsearch/elasticsearch:9.4.4
+    container_name: labrinth-elasticsearch2
+    restart: on-failure
+    networks:
+      - elasticsearch-mesh
+    ports:
+      - '127.0.0.1:9202:9200'
+    volumes:
+      - elasticsearch2-data:/usr/share/elasticsearch/data
+    environment:
+      node.name: elasticsearch2
+      cluster.name: labrinth-elasticsearch
+      discovery.seed_hosts: elasticsearch0,elasticsearch1,elasticsearch2
+      cluster.initial_master_nodes: elasticsearch0,elasticsearch1,elasticsearch2
+      bootstrap.memory_lock: 'true'
+      xpack.security.enabled: 'false'
+      xpack.security.enrollment.enabled: 'false'
+      ES_JAVA_OPTS: -Xms512m -Xmx512m
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+    healthcheck:
+      test:
+        [
+          'CMD-SHELL',
+          'curl --fail http://localhost:9200/_cluster/health?wait_for_status=yellow',
+        ]
+      interval: 5s
+      timeout: 5s
+      retries: 30
   meilisearch0:
     image: getmeili/meilisearch:v1.12.0
     container_name: labrinth-meilisearch0
@@ -398,7 +494,7 @@ services:
     depends_on:
       postgres_db:
         condition: service_healthy
-      meilisearch:
+      meilisearch0:
         condition: service_healthy
       elasticsearch0:
         condition: service_healthy
@@ -481,6 +577,8 @@ services:
     volumes:
       - ./apps/labrinth/nginx/meili-lb.conf:/etc/nginx/conf.d/default.conf:ro
 networks:
+  elasticsearch-mesh:
+    driver: bridge
   meilisearch-mesh:
     driver: bridge
   redis-cluster-mesh: