fix: search indexing performance and batching

apps/labrinth/.env.docker-compose

@@ -25,9 +25,12 @@ ELASTICSEARCH_INDEX_PREFIX=labrinth
 ELASTICSEARCH_USERNAME=elastic
 ELASTICSEARCH_PASSWORD=elastic
 SEARCH_INDEX_CHUNK_SIZE=5000
+SEARCH_INCREMENTAL_INDEX_BATCH_DELAY_SECONDS=5
+SEARCH_INCREMENTAL_INDEX_BATCH_MAX_SIZE=1000
 TYPESENSE_URL=http://localhost:8108
 TYPESENSE_API_KEY=modrinth
 TYPESENSE_INDEX_PREFIX=labrinth
+TYPESENSE_IMPORT_BATCH_SIZE=5000
 
 REDIS_URL=redis://labrinth-redis
 REDIS_MIN_CONNECTIONS=0

apps/labrinth/.env.local

@@ -43,9 +43,12 @@ ELASTICSEARCH_USERNAME=
 ELASTICSEARCH_PASSWORD=
 
 SEARCH_INDEX_CHUNK_SIZE=5000
+SEARCH_INCREMENTAL_INDEX_BATCH_DELAY_SECONDS=5
+SEARCH_INCREMENTAL_INDEX_BATCH_MAX_SIZE=1000
 TYPESENSE_URL=http://localhost:8108
 TYPESENSE_API_KEY=modrinth
 TYPESENSE_INDEX_PREFIX=labrinth
+TYPESENSE_IMPORT_BATCH_SIZE=5000
 
 REDIS_URL=redis://localhost
 REDIS_MIN_CONNECTIONS=0

apps/labrinth/src/background_task.rs

@@ -19,7 +19,7 @@ use crate::util::anrok;
 use actix_web::web;
 use clap::ValueEnum;
 use eyre::WrapErr;
-use tracing::info;
+use tracing::{info, instrument};
 
 #[derive(ValueEnum, Debug, Copy, Clone, PartialEq, Eq)]
 #[clap(rename_all = "kebab_case")]
@@ -50,6 +50,7 @@ pub enum BackgroundTask {
 
 impl BackgroundTask {
     #[allow(clippy::too_many_arguments)]
+    #[instrument(skip_all, fields(background_task = ?self))]
     pub async fn run(
         self,
         pool: PgPool,
@@ -176,7 +177,7 @@ pub async fn index_search(
     search_backend: web::Data<dyn SearchBackend>,
 ) -> eyre::Result<()> {
     info!("Indexing local database");
-    search_backend.index_projects(ro_pool, redis_pool).await
+    search_backend.rebuild_index(ro_pool, redis_pool).await
 }
 
 pub async fn release_scheduled(pool: PgPool) -> eyre::Result<()> {

apps/labrinth/src/env.rs

@@ -38,7 +38,7 @@ macro_rules! vars {
                     )]
                     let $field: Option<$ty> = {
                         let mut default = None::<$ty>;
-                        $( default = Some({ $default }.into()); )?
+                        $( default = Some(<$ty>::from({ $default })); )?
 
                         match parse_value::<$ty>(stringify!($field), default) {
                             Ok(value) => Some(value),
@@ -160,9 +160,13 @@ vars! {
     // search
     SEARCH_BACKEND: crate::search::SearchBackendKind = crate::search::SearchBackendKind::Typesense;
     SEARCH_INDEX_CHUNK_SIZE: i64 = 5000i64;
+    SEARCH_INCREMENTAL_INDEX_BATCH_DELAY_SECONDS: u64 = 5u64;
+    SEARCH_INCREMENTAL_INDEX_BATCH_MAX_SIZE: usize = 1000usize;
     TYPESENSE_URL: String = "http://localhost:8108";
     TYPESENSE_API_KEY: String = "modrinth";
     TYPESENSE_INDEX_PREFIX: String = "labrinth";
+    TYPESENSE_IMPORT_BATCH_SIZE: usize = 5000usize;
+    TYPESENSE_DELETE_BATCH_SIZE: usize = 10_000usize;
 
     // storage
     STORAGE_BACKEND: crate::file_hosting::FileHostKind = crate::file_hosting::FileHostKind::Local;

apps/labrinth/src/lib.rs

@@ -131,30 +131,6 @@ pub fn app_setup(
     ));
 
     if enable_background_tasks {
-        // The interval in seconds at which the local database is indexed
-        // for searching.  Defaults to 1 hour if unset.
-        let local_index_interval =
-            Duration::from_secs(ENV.LOCAL_INDEX_INTERVAL);
-        let pool_ref = pool.clone();
-        let redis_pool_ref = redis_pool.clone();
-        let search_backend_ref = search_backend.clone();
-        scheduler.run(local_index_interval, move || {
-            let pool_ref = pool_ref.clone();
-            let redis_pool_ref = redis_pool_ref.clone();
-            let search_backend = search_backend_ref.clone();
-            async move {
-                if let Err(err) = background_task::index_search(
-                    pool_ref,
-                    redis_pool_ref,
-                    search_backend,
-                )
-                .await
-                {
-                    warn!("Failed to index search: {err:?}");
-                }
-            }
-        });
-
         // Changes statuses of scheduled projects/versions
         let pool_ref = pool.clone();
         // TODO: Clear cache when these are run

apps/labrinth/src/queue/server_ping.rs

@@ -181,8 +181,9 @@ impl ServerPingQueue {
                         None,
                         &self.redis,
                     );
-                    let queue_search =
-                        self.incremental_search_queue.push(*project_id);
+                    let queue_search = self
+                        .incremental_search_queue
+                        .push_project_change(*project_id);
 
                     let (clear_cache_result, _) =
                         join(clear_cache, queue_search).await;

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

@@ -8,7 +8,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

@@ -11,7 +11,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,26 +1099,28 @@ pub async fn project_edit_internal(
             }
         }
 
-        Ok(())
+        Ok(true)
     }
 
-    update(
+    let mut reindex_versions = false;
+
+    reindex_versions |= update(
         &mut transaction,
         id,
         new_project.minecraft_server,
         &mut project_item.inner.components.minecraft_server,
         perms,
     )
     .await?;
-    update(
+    reindex_versions |= update(
         &mut transaction,
         id,
         new_project.minecraft_java_server,
         &mut project_item.inner.components.minecraft_java_server,
         perms,
     )
     .await?;
-    update(
+    reindex_versions |= update(
         &mut transaction,
         id,
         new_project.minecraft_bedrock_server,
@@ -1167,31 +1173,41 @@ 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)) = (
         project_item.inner.status.is_searchable(),
         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(""))
@@ -2791,27 +2807,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)