parallelize remove tasks

apps/labrinth/src/env.rs

@@ -165,6 +165,7 @@ vars! {
     TYPESENSE_URL: String = "http://localhost:8108";
     TYPESENSE_API_KEY: String = "modrinth";
     TYPESENSE_INDEX_PREFIX: String = "labrinth";
+    TYPESENSE_DELETE_BATCH_SIZE: usize = 10_000usize;
 
     // storage
     STORAGE_BACKEND: crate::file_hosting::FileHostKind = crate::file_hosting::FileHostKind::Local;

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

@@ -32,6 +32,7 @@ pub struct TypesenseConfig {
     pub index_prefix: String,
     pub meta_namespace: String,
     pub index_chunk_size: i64,
+    pub delete_batch_size: usize,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -160,6 +161,7 @@ impl TypesenseConfig {
             index_prefix: ENV.TYPESENSE_INDEX_PREFIX.clone(),
             meta_namespace: meta_namespace.unwrap_or_default(),
             index_chunk_size: ENV.SEARCH_INDEX_CHUNK_SIZE,
+            delete_batch_size: ENV.TYPESENSE_DELETE_BATCH_SIZE,
         }
     }
 
@@ -323,12 +325,13 @@ impl TypesenseClient {
         &self,
         collection: &str,
         filter_by: &str,
+        batch_size: usize,
     ) -> Result<()> {
         let resp = self
             .request(
                 Method::DELETE,
                 &format!(
-					"/collections/{collection}/documents?filter_by={}&batch_size=1000",
+					"/collections/{collection}/documents?filter_by={}&batch_size={batch_size}",
 					urlencoding::encode(filter_by)
 				),
             )
@@ -721,6 +724,24 @@ impl Typesense {
         self.client.upsert_alias(alias, &name).await?;
         Ok(())
     }
+
+    async fn delete_documents_by_filter_if_exists(
+        &self,
+        collection: &str,
+        filter: &str,
+    ) -> Result<()> {
+        if self.client.collection_exists(collection).await? {
+            self.client
+                .delete_documents_by_filter(
+                    collection,
+                    filter,
+                    self.config.delete_batch_size,
+                )
+                .await?;
+        }
+
+        Ok(())
+    }
 }
 
 #[async_trait]
@@ -1050,20 +1071,53 @@ impl SearchBackend for Typesense {
                 self.config.get_next_collection_name(&alias, false);
             debug!("Got shadow current {shadow_current:?}");
 
-            for collection in
-                live.into_iter().chain([shadow_alt, shadow_current])
-            {
-                debug!("Working on collection {collection:?}");
-                if self.client.collection_exists(&collection).await? {
+            let delete_live = async {
+                if let Some(collection) = live.as_deref() {
+                    debug!("Working on collection {collection:?}");
                     debug!(
                         filter_len = filter.len(),
                         "Collection exists, deleting by filter"
                     );
-                    self.client
-                        .delete_documents_by_filter(&collection, &filter)
-                        .await?;
+                    self.delete_documents_by_filter_if_exists(
+                        collection, &filter,
+                    )
+                    .await?;
                 }
-            }
+
+                Ok::<(), eyre::Report>(())
+            };
+            let delete_shadow_alt = async {
+                if live.as_deref() != Some(shadow_alt.as_str()) {
+                    debug!("Working on collection {shadow_alt:?}");
+                    self.delete_documents_by_filter_if_exists(
+                        &shadow_alt,
+                        &filter,
+                    )
+                    .await?;
+                }
+
+                Ok::<(), eyre::Report>(())
+            };
+            let delete_shadow_current = async {
+                if live.as_deref() != Some(shadow_current.as_str()) {
+                    debug!("Working on collection {shadow_current:?}");
+                    self.delete_documents_by_filter_if_exists(
+                        &shadow_current,
+                        &filter,
+                    )
+                    .await?;
+                }
+
+                Ok::<(), eyre::Report>(())
+            };
+            let (live_result, shadow_alt_result, shadow_current_result) = tokio::join!(
+                delete_live,
+                delete_shadow_alt,
+                delete_shadow_current
+            );
+            live_result?;
+            shadow_alt_result?;
+            shadow_current_result?;
         }
 
         debug!("Done");
@@ -1092,15 +1146,46 @@ impl SearchBackend for Typesense {
             let shadow_current =
                 self.config.get_next_collection_name(&alias, false);
 
-            for collection in
-                live.into_iter().chain([shadow_alt, shadow_current])
-            {
-                if self.client.collection_exists(&collection).await? {
-                    self.client
-                        .delete_documents_by_filter(&collection, &filter)
-                        .await?;
+            let delete_live = async {
+                if let Some(collection) = live.as_deref() {
+                    self.delete_documents_by_filter_if_exists(
+                        collection, &filter,
+                    )
+                    .await?;
                 }
-            }
+
+                Ok::<(), eyre::Report>(())
+            };
+            let delete_shadow_alt = async {
+                if live.as_deref() != Some(shadow_alt.as_str()) {
+                    self.delete_documents_by_filter_if_exists(
+                        &shadow_alt,
+                        &filter,
+                    )
+                    .await?;
+                }
+
+                Ok::<(), eyre::Report>(())
+            };
+            let delete_shadow_current = async {
+                if live.as_deref() != Some(shadow_current.as_str()) {
+                    self.delete_documents_by_filter_if_exists(
+                        &shadow_current,
+                        &filter,
+                    )
+                    .await?;
+                }
+
+                Ok::<(), eyre::Report>(())
+            };
+            let (live_result, shadow_alt_result, shadow_current_result) = tokio::join!(
+                delete_live,
+                delete_shadow_alt,
+                delete_shadow_current
+            );
+            live_result?;
+            shadow_alt_result?;
+            shadow_current_result?;
         }
         Ok(())
     }