fix: change incremental search indexing operation set (#7179)

* use ro_pool for search indexing again

* adjust incremental indexing operations

apps/labrinth/src/background_task.rs

@@ -100,7 +100,7 @@ impl BackgroundTask {
             }
             IncrementalIndexSearch => {
                 crate::search::incremental::consume::run(
-                    pool,
+                    ro_pool,
                     redis_pool,
                     search_backend,
                     kafka_client,

apps/labrinth/src/models/v2/search.rs

@@ -152,7 +152,9 @@ impl LegacyResultSearchProject {
             server_side,
             environment: environments,
             versions,
-            latest_version: result_search_project.version_id,
+            latest_version: result_search_project
+                .version_id
+                .unwrap_or_default(),
             categories,
 
             project_id: result_search_project.project_id,

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

@@ -1207,6 +1207,14 @@ pub async fn project_edit_internal(
 
     let mut reindex_versions = new_project.categories.is_some()
         || new_project.additional_categories.is_some();
+    let became_searchable = !project_item.inner.status.is_searchable()
+        && new_project
+            .status
+            .is_some_and(|status| status.is_searchable());
+    let became_unsearchable = project_item.inner.status.is_searchable()
+        && new_project
+            .status
+            .is_some_and(|status| !status.is_searchable());
 
     reindex_versions |= update(
         &mut transaction,
@@ -1284,7 +1292,7 @@ pub async fn project_edit_internal(
         .await
         .wrap_internal_err("committing database transaction")?;
 
-    if reindex_versions {
+    if became_unsearchable {
         db_models::DBProject::clear_cache(
             project_item.inner.id,
             project_item.inner.slug,
@@ -1295,10 +1303,20 @@ pub async fn project_edit_internal(
         .wrap_internal_err("clearing cached data from Redis")?;
         search_state
             .queue
-            .push_version_changes(
+            .push_project_removal(project_item.inner.id.into())
-                project_item.inner.id.into(),
+            .await;
-                project_item.versions.iter().copied().map(VersionId::from),
+    } else if reindex_versions || became_searchable {
+        db_models::DBProject::clear_cache(
+            project_item.inner.id,
+            project_item.inner.slug,
+            None,
+            &redis,
         )
+        .await
+        .wrap_internal_err("clearing cached data from Redis")?;
+        search_state
+            .queue
+            .push_project_with_all_versions_change(project_item.inner.id.into())
             .await;
     } else {
         clear_project_cache_and_queue_search(
@@ -1312,17 +1330,6 @@ pub async fn project_edit_internal(
         .wrap_api_err("executing `clear_project_cache_and_queue_search`")?;
     }
 
-    // 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
-            .queue
-            .push_project_removal(project_item.inner.id.into())
-            .await;
-    }
-
     Ok(HttpResponse::NoContent().body(""))
 }
 
@@ -1890,7 +1897,6 @@ pub async fn projects_edit(
         changed_projects.push((
             project.inner.id,
             project.inner.slug,
-            project.versions,
             reindex_versions,
         ));
     }
@@ -1900,17 +1906,14 @@ pub async fn projects_edit(
         .await
         .wrap_internal_err("committing database transaction")?;
 
-    for (project_id, slug, versions, reindex_versions) in changed_projects {
+    for (project_id, slug, reindex_versions) in changed_projects {
         if reindex_versions {
             db_models::DBProject::clear_cache(project_id, slug, None, &redis)
                 .await
                 .wrap_internal_err("clearing cached data from Redis")?;
             search_state
                 .queue
-                .push_version_changes(
+                .push_project_with_all_versions_change(project_id.into())
-                    project_id.into(),
-                    versions.into_iter().map(VersionId::from),
-                )
                 .await;
         } else {
             clear_project_cache_and_queue_search(

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

@@ -910,6 +910,7 @@ impl Elasticsearch {
                 let mut document = hit["_source"].clone();
                 let object = document.as_object_mut()?;
                 object.remove("document_type");
+                object.remove("_search_tokens");
                 if filter
                     .as_ref()
                     .is_some_and(|filter| filter.has_version_filter)

apps/labrinth/src/search/incremental.rs

@@ -17,7 +17,7 @@ use crate::{
 };
 
 pub const SEARCH_PROJECT_INDEX_QUEUE_TOPIC: &str =
-    "public.labrinth.search-project-index-queue.v1";
+    "public.labrinth.search-project-index-queue.v2";
 const QUEUE_FLUSH_INTERVAL: Duration = Duration::from_secs(10);
 
 #[derive(Clone)]
@@ -40,6 +40,16 @@ impl IncrementalSearchQueue {
         self.operations.lock().await.push_project_change(project_id);
     }
 
+    pub async fn push_project_with_all_versions_change(
+        &self,
+        project_id: ProjectId,
+    ) {
+        self.operations
+            .lock()
+            .await
+            .push_project_with_all_versions_change(project_id);
+    }
+
     pub async fn push_version_changes(
         &self,
         project_id: ProjectId,
@@ -116,6 +126,7 @@ impl IncrementalSearchQueue {
 #[derive(Default)]
 struct PendingSearchIndexOperations {
     changed_project_ids: HashSet<ProjectId>,
+    changed_project_ids_with_all_versions: HashSet<ProjectId>,
     changed_project_versions: HashMap<ProjectId, HashSet<VersionId>>,
     removed_project_ids: HashSet<ProjectId>,
 }
@@ -123,25 +134,47 @@ struct PendingSearchIndexOperations {
 impl PendingSearchIndexOperations {
     fn is_empty(&self) -> bool {
         self.changed_project_ids.is_empty()
+            && self.changed_project_ids_with_all_versions.is_empty()
             && self.changed_project_versions.is_empty()
             && self.removed_project_ids.is_empty()
     }
 
     fn push_project_change(&mut self, project_id: ProjectId) {
-        if !self.removed_project_ids.contains(&project_id) {
+        if !self.removed_project_ids.contains(&project_id)
+            && !self
+                .changed_project_ids_with_all_versions
+                .contains(&project_id)
+            && !self.changed_project_versions.contains_key(&project_id)
+        {
             self.changed_project_ids.insert(project_id);
         }
     }
 
+    fn push_project_with_all_versions_change(&mut self, project_id: ProjectId) {
+        if self.removed_project_ids.contains(&project_id) {
+            return;
+        }
+
+        self.changed_project_ids.remove(&project_id);
+        self.changed_project_versions.remove(&project_id);
+        self.changed_project_ids_with_all_versions
+            .insert(project_id);
+    }
+
     fn push_version_change(
         &mut self,
         project_id: ProjectId,
         version_ids: impl IntoIterator<Item = VersionId>,
     ) {
-        if self.removed_project_ids.contains(&project_id) {
+        if self.removed_project_ids.contains(&project_id)
+            || self
+                .changed_project_ids_with_all_versions
+                .contains(&project_id)
+        {
             return;
         }
 
+        self.changed_project_ids.remove(&project_id);
         let version_ids = version_ids.into_iter().collect::<HashSet<_>>();
         if !version_ids.is_empty() {
             self.changed_project_versions
@@ -153,6 +186,8 @@ impl PendingSearchIndexOperations {
 
     fn push_project_removal(&mut self, project_id: ProjectId) {
         self.changed_project_ids.remove(&project_id);
+        self.changed_project_ids_with_all_versions
+            .remove(&project_id);
         self.changed_project_versions.remove(&project_id);
         self.removed_project_ids.insert(project_id);
     }
@@ -162,6 +197,9 @@ impl PendingSearchIndexOperations {
             SearchProjectIndexQueueEventData::Change { project_id } => {
                 self.push_project_change(project_id)
             }
+            SearchProjectIndexQueueEventData::ChangeWithAllVersions {
+                project_id,
+            } => self.push_project_with_all_versions_change(project_id),
             SearchProjectIndexQueueEventData::VersionChange {
                 project_id,
                 version_ids,
@@ -175,6 +213,7 @@ impl PendingSearchIndexOperations {
     fn into_events(self) -> Vec<SearchProjectIndexQueueEventData> {
         let mut events = Vec::with_capacity(
             self.changed_project_ids.len()
+                + self.changed_project_ids_with_all_versions.len()
                 + self.changed_project_versions.len()
                 + self.removed_project_ids.len(),
         );
@@ -185,6 +224,15 @@ impl PendingSearchIndexOperations {
         events.extend(self.changed_project_ids.into_iter().map(|project_id| {
             SearchProjectIndexQueueEventData::Change { project_id }
         }));
+        events.extend(
+            self.changed_project_ids_with_all_versions.into_iter().map(
+                |project_id| {
+                    SearchProjectIndexQueueEventData::ChangeWithAllVersions {
+                        project_id,
+                    }
+                },
+            ),
+        );
         events.extend(self.changed_project_versions.into_iter().map(
             |(project_id, version_ids)| {
                 SearchProjectIndexQueueEventData::VersionChange {
@@ -202,6 +250,8 @@ impl PendingSearchIndexOperations {
 pub enum SearchProjectIndexQueueEventData {
     #[serde(rename = "project_change")]
     Change { project_id: ProjectId },
+    #[serde(rename = "project_change_with_all_versions")]
+    ChangeWithAllVersions { project_id: ProjectId },
     #[serde(rename = "project_version_change")]
     VersionChange {
         project_id: ProjectId,

apps/labrinth/src/search/incremental/consume.rs

@@ -21,7 +21,10 @@ use crate::{
         SearchBackend, SearchDocumentBatch, SearchIndexUpdate,
         UploadSearchProject,
         incremental::SEARCH_PROJECT_INDEX_QUEUE_TOPIC,
-        indexing::{build_project_documents, build_version_change_documents},
+        indexing::{
+            build_project_and_all_version_documents, build_project_documents,
+            build_version_change_documents,
+        },
     },
     util::kafka::{
         INCREMENTAL_INDEX_SEARCH_TASK, KAFKA_OPERATION_INTERVAL,
@@ -132,6 +135,7 @@ async fn consume_batch(
     let start = Instant::now();
 
     let mut project_ids_to_change = HashSet::new();
+    let mut project_ids_with_all_versions_to_change = HashSet::new();
     let mut project_ids_with_version_changes = HashSet::new();
     let mut project_ids_to_remove = HashSet::new();
     let mut version_ids_to_change = HashSet::new();
@@ -180,6 +184,11 @@ async fn consume_batch(
             SearchProjectIndexQueueEventData::Change { project_id } => {
                 project_ids_to_change.insert(project_id);
             }
+            SearchProjectIndexQueueEventData::ChangeWithAllVersions {
+                project_id,
+            } => {
+                project_ids_with_all_versions_to_change.insert(project_id);
+            }
             SearchProjectIndexQueueEventData::VersionChange {
                 project_id,
                 version_ids,
@@ -198,16 +207,26 @@ async fn consume_batch(
 
     project_ids_to_change
         .retain(|project_id| !project_ids_to_remove.contains(project_id));
+    project_ids_with_all_versions_to_change
+        .retain(|project_id| !project_ids_to_remove.contains(project_id));
     project_ids_with_version_changes
         .retain(|project_id| !project_ids_to_remove.contains(project_id));
+    project_ids_with_version_changes.retain(|project_id| {
+        !project_ids_with_all_versions_to_change.contains(project_id)
+    });
     project_ids_to_change.retain(|project_id| {
         !project_ids_with_version_changes.contains(project_id)
+            && !project_ids_with_all_versions_to_change.contains(project_id)
     });
     let project_ids_to_change =
         project_ids_to_change.into_iter().collect::<Vec<_>>();
     let project_ids_with_version_changes = project_ids_with_version_changes
         .into_iter()
         .collect::<Vec<_>>();
+    let project_ids_with_all_versions_to_change =
+        project_ids_with_all_versions_to_change
+            .into_iter()
+            .collect::<Vec<_>>();
     let mut project_ids_to_remove =
         project_ids_to_remove.into_iter().collect::<Vec<_>>();
     let version_ids_to_change =
@@ -215,9 +234,10 @@ async fn consume_batch(
 
     info!(
         kafka.message_count = messages_to_commit.len(),
-        "Read all Kafka messages in {:.2?}, found {} projects to change, {} projects with {} version changes, and {} projects to remove",
+        "Read all Kafka messages in {:.2?}, found {} projects to change, {} projects with all versions to change, {} projects with {} version changes, and {} projects to remove",
         start.elapsed(),
         project_ids_to_change.len(),
+        project_ids_with_all_versions_to_change.len(),
         project_ids_with_version_changes.len(),
         version_ids_to_change.len(),
         project_ids_to_remove.len(),
@@ -225,6 +245,35 @@ async fn consume_batch(
     let start = Instant::now();
     let mut documents = SearchDocumentBatch::default();
 
+    if !project_ids_with_all_versions_to_change.is_empty() {
+        let operation_start = Instant::now();
+        let changed_documents = build_project_and_all_version_documents(
+            ro_pool,
+            redis_pool,
+            &project_ids_with_all_versions_to_change,
+        )
+        .instrument(info_span!(
+            "index",
+            batch_size = project_ids_with_all_versions_to_change.len()
+        ))
+        .await
+        .wrap_err_with(|| {
+            format!(
+                "failed to build search documents for {} projects and all their versions",
+                project_ids_with_all_versions_to_change.len()
+            )
+        })?;
+        project_ids_to_remove
+            .extend(project_ids_with_all_versions_to_change.iter().copied());
+        documents.projects.extend(changed_documents.projects);
+        documents.versions.extend(changed_documents.versions);
+        info!(
+            project_count = project_ids_with_all_versions_to_change.len(),
+            "Built changed projects and all their versions in {:.2?}",
+            operation_start.elapsed()
+        );
+    }
+
     if !project_ids_with_version_changes.is_empty() {
         let operation_start = Instant::now();
         let changed_documents = build_version_change_documents(
@@ -402,6 +451,8 @@ enum SearchProjectIndexQueueEvent {
 enum SearchProjectIndexQueueEventData {
     #[serde(rename = "project_change")]
     Change { project_id: ProjectId },
+    #[serde(rename = "project_change_with_all_versions")]
+    ChangeWithAllVersions { project_id: ProjectId },
     #[serde(rename = "project_version_change")]
     VersionChange {
         project_id: ProjectId,

apps/labrinth/src/search/indexing.rs

@@ -129,11 +129,22 @@ pub async fn build_project_documents(
     project_ids: &[ProjectId],
 ) -> eyre::Result<Vec<UploadSearchProject>> {
     let version_ids = HashSet::new();
-    Ok(
+    Ok(build_search_document_batch(
-        build_search_document_batch(pool, redis, project_ids, &version_ids)
+        pool,
-            .await?
+        redis,
-            .projects,
+        project_ids,
+        Some(&version_ids),
     )
+    .await?
+    .projects)
+}
+
+pub async fn build_project_and_all_version_documents(
+    pool: &PgPool,
+    redis: &RedisPool,
+    project_ids: &[ProjectId],
+) -> eyre::Result<SearchDocumentBatch> {
+    build_search_document_batch(pool, redis, project_ids, None).await
 }
 
 pub async fn build_version_change_documents(
@@ -147,14 +158,15 @@ pub async fn build_version_change_documents(
         .copied()
         .map(DBVersionId::from)
         .collect::<HashSet<_>>();
-    build_search_document_batch(pool, redis, project_ids, &version_ids).await
+    build_search_document_batch(pool, redis, project_ids, Some(&version_ids))
+        .await
 }
 
 async fn build_search_document_batch(
     pool: &PgPool,
     redis: &RedisPool,
     project_ids: &[ProjectId],
-    version_ids: &HashSet<DBVersionId>,
+    version_ids: Option<&HashSet<DBVersionId>>,
 ) -> eyre::Result<SearchDocumentBatch> {
     let searchable_statuses = searchable_statuses();
     let project_ids = project_ids
@@ -194,7 +206,7 @@ async fn build_search_document_batch(
     .await
     .wrap_err("failed to fetch project")?;
 
-    build_search_documents(pool, redis, db_projects, Some(version_ids)).await
+    build_search_documents(pool, redis, db_projects, version_ids).await
 }
 
 async fn build_search_documents(
@@ -612,14 +624,12 @@ async fn build_search_documents(
             .map(|dependency| dependency.project_id.clone())
             .collect::<Vec<_>>();
 
-        if let Some(versions) = versions.remove(&project.id) {
+        let versions = versions.remove(&project.id).unwrap_or_default();
-            let Some(latest_version) = versions.iter().max_by(|a, b| {
+        let latest_version = versions.iter().max_by(|a, b| {
             a.date_published
                 .cmp(&b.date_published)
                 .then_with(|| a.id.0.cmp(&b.id.0))
-            }) else {
+        });
-                continue;
-            };
 
         let project_version_fields = versions
             .iter()
@@ -661,9 +671,9 @@ async fn build_search_documents(
 
         let project_id = ProjectId::from(project.id).to_string();
         version_uploads.extend(versions.iter().filter_map(|version| {
-                if version_ids.is_some_and(|version_ids| {
+            if version_ids
-                    !version_ids.contains(&version.id)
+                .is_some_and(|version_ids| !version_ids.contains(&version.id))
-                }) {
+            {
                 return None;
             }
 
@@ -676,10 +686,7 @@ async fn build_search_documents(
             let unvectorized_loader_fields = version_fields
                 .iter()
                 .map(|field| {
-                        (
+                    (field.field_name.clone(), field.value.serialize_internal())
-                            field.field_name.clone(),
-                            field.value.serialize_internal(),
-                        )
                 })
                 .collect();
             let mut fields = from_duplicate_version_fields(version_fields);
@@ -738,9 +745,7 @@ async fn build_search_documents(
                 project_id: project_id.clone(),
                 categories: version_categories,
                 project_types: version_project_types,
-                    version_published_timestamp: version
+                version_published_timestamp: version.date_published.timestamp(),
-                        .date_published
-                        .timestamp(),
                 loader_fields: fields,
             })
         }));
@@ -770,12 +775,10 @@ async fn build_search_documents(
             );
 
         if let Ok(client_side) = serde_json::to_value(client_side) {
-                loader_fields
+            loader_fields.insert("client_side".to_string(), vec![client_side]);
-                    .insert("client_side".to_string(), vec![client_side]);
         }
         if let Ok(server_side) = serde_json::to_value(server_side) {
-                loader_fields
+            loader_fields.insert("server_side".to_string(), vec![server_side]);
-                    .insert("server_side".to_string(), vec![server_side]);
         }
 
         let components = project
@@ -786,10 +789,9 @@ async fn build_search_documents(
         let indexed_name = normalize_for_search(&project.name);
 
         project_uploads.push(UploadSearchProject {
-                version_id: crate::models::ids::VersionId::from(
+            version_id: latest_version.map(|version| {
-                    latest_version.id,
+                crate::models::ids::VersionId::from(version.id).to_string()
-                )
+            }),
-                .to_string(),
             project_id,
             name: project.name,
             indexed_name,
@@ -813,8 +815,7 @@ async fn build_search_documents(
             date_modified: project.updated,
             modified_timestamp: project.updated.timestamp(),
             version_published_timestamp: latest_version
-                    .date_published
+                .map(|version| version.date_published.timestamp()),
-                    .timestamp(),
             license,
             slug: project.slug,
             project_types: project_types.clone(),
@@ -837,7 +838,6 @@ async fn build_search_documents(
             components,
         });
     }
-    }
 
     Ok(SearchDocumentBatch {
         projects: project_uploads,

apps/labrinth/src/search/mod.rs

@@ -247,7 +247,8 @@ impl FromStr for SearchBackendKind {
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct UploadSearchProject {
     /// ID of the most recently published version.
-    pub version_id: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub version_id: Option<String>,
     pub project_id: String,
     //
     pub project_types: Vec<String>,
@@ -285,7 +286,8 @@ pub struct UploadSearchProject {
     /// Unix timestamp of the last major modification
     pub modified_timestamp: i64,
     /// Unix timestamp of the most recently published version.
-    pub version_published_timestamp: i64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub version_published_timestamp: Option<i64>,
     pub open_source: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub color: Option<u32>,
@@ -369,7 +371,8 @@ pub struct SearchResults {
 #[derive(Serialize, Deserialize, Debug, Clone, ToSchema)]
 pub struct ResultSearchProject {
     /// ID of the most recently published version.
-    pub version_id: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub version_id: Option<String>,
     pub project_id: String,
     pub project_types: Vec<String>,
     #[serde(default)]

docker-compose.yml

@@ -578,6 +578,8 @@ services:
     volumes:
       - ./apps/labrinth/nginx/meili-lb.conf:/etc/nginx/conf.d/default.conf:ro
 networks:
+  default:
+    driver: bridge
   elasticsearch-mesh:
     driver: bridge
   meilisearch-mesh:

packages/api-client/src/modules/labrinth/types.ts

@@ -1899,7 +1899,7 @@ export namespace Labrinth {
 
 		export namespace v3 {
 			export interface ResultSearchProject {
-				version_id: string
+				version_id?: string
 				project_id: string
 				project_types: string[]
 				all_project_types: string[]