fix: search indexing performance and batching (#6521)

* Add consume batching delay

* maybe fix

* max batch size

* more logging

* parallelize remove tasks

* delete/upsert project/version messages

* prepare

* more logging

* log number of docs

* try more targeted, homogenous version change ops

* ensure only necessary fields are serialized into typesense

* disable index background task

* wip: script changes

* don't facet by project id

* fix

* fix clippy

* batch by document and dedup loaders

* fix test

* wip: projects/versions collections

* clean up SearchBackend interface

* cleanup pass

* cleanup pass 2

* standardise fn names

* cleanup pass

* fix compile

* factor out filter rewriting

* query perf

* put categories into the project doc

* wip: search filter AST

* doc comment

* more AST normalization

* (temp) convert search request errors into internal errors

* implement unary NOT

* tombi fmt

* fix tests

* revert request error

* try increase stack size

apps/labrinth/src/routes/internal/admin.rs

@@ -7,7 +7,7 @@ use crate::queue::analytics::AnalyticsQueue;
 use crate::queue::session::AuthQueue;
 use crate::routes::ApiError;
 use crate::search::SearchBackend;
-use crate::search::incremental::consume::reindex_project;
+use crate::search::incremental::consume::reindex_project_document;
 use crate::util::date::get_current_tenths_of_ms;
 use crate::util::error::Context;
 use crate::util::guards::admin_key_guard;
@@ -330,7 +330,7 @@ pub async fn force_reindex(
 ) -> Result<HttpResponse, ApiError> {
     let redis = redis.get_ref();
     search_backend
-        .index_projects(pool.as_ref().clone(), redis.clone())
+        .rebuild_index(pool.as_ref().clone(), redis.clone())
         .await
         .wrap_internal_err("failed to index projects")?;
     Ok(HttpResponse::NoContent().finish())
@@ -355,7 +355,7 @@ pub async fn force_reindex_project(
     search_backend: web::Data<dyn SearchBackend>,
 ) -> Result<HttpResponse, ApiError> {
     let (project_id,) = path.into_inner();
-    reindex_project(
+    reindex_project_document(
         pool.as_ref(),
         redis.as_ref(),
         search_backend.as_ref(),

apps/labrinth/src/routes/v2/versions.rs

@@ -10,7 +10,7 @@ use crate::models::projects::{
 use crate::models::v2::projects::LegacyVersion;
 use crate::queue::session::AuthQueue;
 use crate::routes::{v2_reroute, v3};
-use crate::search::{SearchBackend, SearchState};
+use crate::search::SearchState;
 use actix_web::{HttpRequest, HttpResponse, delete, get, patch, web};
 use serde::{Deserialize, Serialize};
 use validator::Validate;
@@ -488,7 +488,6 @@ pub async fn version_delete(
     pool: web::Data<PgPool>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
-    search_backend: web::Data<dyn SearchBackend>,
     search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     // Returns NoContent, so we don't need to convert the response
@@ -498,7 +497,6 @@ pub async fn version_delete(
         pool,
         redis,
         session_queue,
-        search_backend,
         search_state,
     )
     .await

apps/labrinth/src/routes/v3/projects.rs

@@ -85,7 +85,11 @@ pub async fn clear_project_cache_and_queue_search(
         redis,
     )
     .await?;
-    search_state.queue.push(project_id.into()).await;
+
+    search_state
+        .queue
+        .push_project_change(project_id.into())
+        .await;
 
     Ok(())
 }
@@ -1053,10 +1057,10 @@ pub async fn project_edit_internal(
         edit: Option<Option<E>>,
         mut component: &mut Option<E::Component>,
         perms: ProjectPermissions,
-    ) -> Result<(), ApiError> {
+    ) -> Result<bool, ApiError> {
         let Some(edit) = edit else {
             // component is not specified in the input JSON - leave alone
-            return Ok(());
+            return Ok(false);
         };
 
         if !perms.contains(ProjectPermissions::EDIT_DETAILS) {
@@ -1095,10 +1099,13 @@ pub async fn project_edit_internal(
             }
         }
 
-        Ok(())
+        Ok(true)
     }
 
-    update(
+    let mut reindex_versions = new_project.categories.is_some()
+        || new_project.additional_categories.is_some();
+
+    reindex_versions |= update(
         &mut transaction,
         id,
         new_project.minecraft_server,
@@ -1106,7 +1113,7 @@ pub async fn project_edit_internal(
         perms,
     )
     .await?;
-    update(
+    reindex_versions |= update(
         &mut transaction,
         id,
         new_project.minecraft_java_server,
@@ -1114,7 +1121,7 @@ pub async fn project_edit_internal(
         perms,
     )
     .await?;
-    update(
+    reindex_versions |= update(
         &mut transaction,
         id,
         new_project.minecraft_bedrock_server,
@@ -1167,14 +1174,31 @@ pub async fn project_edit_internal(
 
     transaction.commit().await?;
 
-    clear_project_cache_and_queue_search(
-        &redis,
-        &search_state,
-        project_item.inner.id,
-        project_item.inner.slug,
-        None,
-    )
-    .await?;
+    if reindex_versions {
+        db_models::DBProject::clear_cache(
+            project_item.inner.id,
+            project_item.inner.slug,
+            None,
+            &redis,
+        )
+        .await?;
+        search_state
+            .queue
+            .push_version_changes(
+                project_item.inner.id.into(),
+                project_item.versions.iter().copied().map(VersionId::from),
+            )
+            .await;
+    } else {
+        clear_project_cache_and_queue_search(
+            &redis,
+            &search_state,
+            project_item.inner.id,
+            project_item.inner.slug,
+            None,
+        )
+        .await?;
+    }
 
     // Remove no longer searchable projects from search index
     if let (true, Some(false)) = (
@@ -1182,16 +1206,9 @@ pub async fn project_edit_internal(
         new_project.status.map(|status| status.is_searchable()),
     ) {
         search_state
-            .backend
-            .remove_documents(
-                &project_item
-                    .versions
-                    .into_iter()
-                    .map(|x| x.into())
-                    .collect::<Vec<_>>(),
-            )
-            .await
-            .wrap_internal_err("failed to remove documents")?;
+            .queue
+            .push_project_removal(project_item.inner.id.into())
+            .await;
     }
 
     Ok(HttpResponse::NoContent().body(""))
@@ -1654,7 +1671,7 @@ pub async fn projects_edit(
             };
         }
 
-        bulk_edit_project_categories(
+        let mut reindex_versions = bulk_edit_project_categories(
             &categories,
             &project.categories,
             project.inner.id as db_ids::DBProjectId,
@@ -1669,7 +1686,7 @@ pub async fn projects_edit(
         )
         .await?;
 
-        bulk_edit_project_categories(
+        reindex_versions |= bulk_edit_project_categories(
             &categories,
             &project.additional_categories,
             project.inner.id as db_ids::DBProjectId,
@@ -1728,20 +1745,37 @@ pub async fn projects_edit(
             }
         }
 
-        changed_projects.push((project.inner.id, project.inner.slug));
+        changed_projects.push((
+            project.inner.id,
+            project.inner.slug,
+            project.versions,
+            reindex_versions,
+        ));
     }
 
     transaction.commit().await?;
 
-    for (project_id, slug) in changed_projects {
-        clear_project_cache_and_queue_search(
-            &redis,
-            &search_state,
-            project_id,
-            slug,
-            None,
-        )
-        .await?;
+    for (project_id, slug, versions, reindex_versions) in changed_projects {
+        if reindex_versions {
+            db_models::DBProject::clear_cache(project_id, slug, None, &redis)
+                .await?;
+            search_state
+                .queue
+                .push_version_changes(
+                    project_id.into(),
+                    versions.into_iter().map(VersionId::from),
+                )
+                .await;
+        } else {
+            clear_project_cache_and_queue_search(
+                &redis,
+                &search_state,
+                project_id,
+                slug,
+                None,
+            )
+            .await?;
+        }
     }
 
     Ok(HttpResponse::NoContent().body(""))
@@ -1755,7 +1789,7 @@ pub async fn bulk_edit_project_categories(
     max_num_categories: usize,
     is_additional: bool,
     transaction: &mut PgTransaction<'_>,
-) -> Result<(), ApiError> {
+) -> Result<bool, ApiError> {
     let mut set_categories =
         if let Some(categories) = bulk_changes.categories.clone() {
             categories
@@ -1782,7 +1816,8 @@ pub async fn bulk_edit_project_categories(
         }
     }
 
-    if &set_categories != project_categories {
+    let changed = &set_categories != project_categories;
+    if changed {
         sqlx::query!(
             "
             DELETE FROM mods_categories
@@ -1815,7 +1850,7 @@ pub async fn bulk_edit_project_categories(
         DBModCategory::insert_many(mod_categories, &mut *transaction).await?;
     }
 
-    Ok(())
+    Ok(changed)
 }
 
 #[derive(Serialize, Deserialize)]
@@ -2791,27 +2826,18 @@ pub async fn project_delete_internal(
         .await
         .wrap_internal_err("failed to commit transaction")?;
 
-    search_state
-        .backend
-        .remove_documents(
-            &project
-                .versions
-                .into_iter()
-                .map(|x| x.into())
-                .collect::<Vec<_>>(),
-        )
-        .await
-        .wrap_internal_err("failed to remove project version documents")?;
-
     if result.is_some() {
-        clear_project_cache_and_queue_search(
-            &redis,
-            &search_state,
+        db_models::DBProject::clear_cache(
             project.inner.id,
             project.inner.slug,
             None,
+            &redis,
         )
         .await?;
+        search_state
+            .queue
+            .push_project_removal(project.inner.id.into())
+            .await;
         Ok(())
     } else {
         Err(ApiError::NotFound)

apps/labrinth/src/routes/v3/version_creation.rs

@@ -184,19 +184,20 @@ pub async fn version_create(
         if let Err(e) = rollback_result {
             return Err(e.into());
         }
-    } else if let Ok((_, project_id)) = &result {
+    } else if let Ok((_, project_id, version_id)) = &result {
         transaction.commit().await?;
-        super::projects::clear_project_cache_and_queue_search(
-            &redis,
-            &search_state,
-            *project_id,
-            None,
-            Some(true),
-        )
-        .await?;
+        models::DBProject::clear_cache(*project_id, None, Some(true), &redis)
+            .await?;
+        search_state
+            .queue
+            .push_version_changes(
+                (*project_id).into(),
+                [VersionId::from(*version_id)],
+            )
+            .await;
     }
 
-    result.map(|(response, _)| response)
+    result.map(|(response, _, _)| response)
 }
 
 #[allow(clippy::too_many_arguments)]
@@ -210,7 +211,8 @@ async fn version_create_inner(
     pool: &PgPool,
     session_queue: &AuthQueue,
     http: &reqwest::Client,
-) -> Result<(HttpResponse, models::DBProjectId), CreateError> {
+) -> Result<(HttpResponse, models::DBProjectId, models::DBVersionId), CreateError>
+{
     let mut initial_version_data = None;
     let mut version_builder = None;
     let mut selected_loaders = None;
@@ -568,7 +570,11 @@ async fn version_create_inner(
         }
     }
 
-    Ok((HttpResponse::Ok().json(response), project_id))
+    Ok((
+        HttpResponse::Ok().json(response),
+        project_id,
+        models::DBVersionId::from(version_id),
+    ))
 }
 
 /// Add files to an existing version.
@@ -635,17 +641,18 @@ pub async fn upload_file_to_version(
     let mut transaction = client.begin().await?;
     let mut uploaded_files = Vec::new();
 
-    let version_id = models::DBVersionId::from(url_data.into_inner().0);
+    let version_id = url_data.into_inner().0;
+    let db_version_id = models::DBVersionId::from(version_id);
 
     let result = upload_file_to_version_inner(
         req,
         &mut payload,
-        client,
+        client.clone(),
         &mut transaction,
         redis.clone(),
         &**file_host,
         &mut uploaded_files,
-        version_id,
+        db_version_id,
         &session_queue,
         &http,
     )
@@ -665,14 +672,12 @@ pub async fn upload_file_to_version(
         }
     } else if let Ok((_, project_id)) = &result {
         transaction.commit().await?;
-        super::projects::clear_project_cache_and_queue_search(
-            &redis,
-            &search_state,
-            *project_id,
-            None,
-            Some(true),
-        )
-        .await?;
+        models::DBProject::clear_cache(*project_id, None, Some(true), &redis)
+            .await?;
+        search_state
+            .queue
+            .push_version_changes((*project_id).into(), [version_id])
+            .await;
     }
 
     result.map(|(response, _)| response)

apps/labrinth/src/routes/v3/versions.rs

@@ -26,8 +26,7 @@ use crate::models::teams::ProjectPermissions;
 use crate::queue::file_scan::get_files_missing_attribution;
 use crate::queue::session::AuthQueue;
 use crate::routes::internal::delphi;
-use crate::search::{SearchBackend, SearchState};
-use crate::util::error::Context;
+use crate::search::SearchState;
 use crate::util::img;
 use crate::util::validate::validation_errors_to_string;
 use actix_web::{HttpRequest, HttpResponse, delete, get, patch, web};
@@ -853,14 +852,20 @@ pub async fn version_edit_helper(
             transaction.commit().await?;
             database::models::DBVersion::clear_cache(&version_item, &redis)
                 .await?;
-            super::projects::clear_project_cache_and_queue_search(
-                &redis,
-                &search_state,
+            database::models::DBProject::clear_cache(
                 version_item.inner.project_id,
                 None,
                 Some(true),
+                &redis,
             )
             .await?;
+            search_state
+                .queue
+                .push_version_changes(
+                    version_item.inner.project_id.into(),
+                    [VersionId::from(version_item.inner.id)],
+                )
+                .await;
             Ok(HttpResponse::NoContent().body(""))
         } else {
             Err(ApiError::CustomAuthentication(
@@ -1129,19 +1134,9 @@ pub async fn version_delete_route(
     pool: web::Data<PgPool>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
-    search_backend: web::Data<dyn SearchBackend>,
     search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
-    version_delete(
-        req,
-        info,
-        pool,
-        redis,
-        session_queue,
-        search_backend,
-        search_state,
-    )
-    .await
+    version_delete(req, info, pool, redis, session_queue, search_state).await
 }
 
 pub async fn version_delete(
@@ -1150,7 +1145,6 @@ pub async fn version_delete(
     pool: web::Data<PgPool>,
     redis: web::Data<RedisPool>,
     session_queue: web::Data<AuthQueue>,
-    search_backend: web::Data<dyn SearchBackend>,
     search_state: web::Data<SearchState>,
 ) -> Result<HttpResponse, ApiError> {
     let user = get_user_from_headers(
@@ -1246,18 +1240,20 @@ pub async fn version_delete(
 
     transaction.commit().await?;
 
-    super::projects::clear_project_cache_and_queue_search(
-        &redis,
-        &search_state,
+    database::models::DBProject::clear_cache(
         version.inner.project_id,
         None,
         Some(true),
+        &redis,
     )
     .await?;
-    search_backend
-        .remove_documents(&[version.inner.id.into()])
-        .await
-        .wrap_internal_err("failed to remove documents")?;
+    search_state
+        .queue
+        .push_version_changes(
+            version.inner.project_id.into(),
+            [VersionId::from(version.inner.id)],
+        )
+        .await;
     if result.is_some() {
         Ok(HttpResponse::NoContent().body(""))
     } else {