diff --git a/apps/labrinth/src/search/backend/typesense/mod.rs b/apps/labrinth/src/search/backend/typesense/mod.rs index 6bbb141bf7..42beb818a5 100644 --- a/apps/labrinth/src/search/backend/typesense/mod.rs +++ b/apps/labrinth/src/search/backend/typesense/mod.rs @@ -7,7 +7,7 @@ use regex::Regex; use reqwest::Method; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; -use tracing::{info, warn}; +use tracing::{debug, info, warn}; use crate::database::PgPool; use crate::database::redis::RedisPool; @@ -1038,21 +1038,35 @@ impl SearchBackend for Typesense { self.config.get_alias_name("projects"), self.config.get_alias_name("projects_filtered"), ] { + debug!("Performing removal on alias {alias:?}"); + let live = self.client.get_alias(&alias).await?; + debug!("Got live alias {live:?}"); + let shadow_alt = self.config.get_next_collection_name(&alias, true); + debug!("Got shadow alt {shadow_alt:?}"); + let shadow_current = 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? { + debug!( + filter_len = filter.len(), + "Collection exists, deleting by filter" + ); self.client .delete_documents_by_filter(&collection, &filter) .await?; } } } + + debug!("Done"); Ok(()) } diff --git a/apps/labrinth/src/search/incremental/consume.rs b/apps/labrinth/src/search/incremental/consume.rs index f74b0dd5c1..58e8a6456a 100644 --- a/apps/labrinth/src/search/incremental/consume.rs +++ b/apps/labrinth/src/search/incremental/consume.rs @@ -1,6 +1,5 @@ use actix_web::web; use eyre::WrapErr; -use futures::never::Never; use rdkafka::{ Message, consumer::{CommitMode, Consumer, StreamConsumer}, @@ -215,8 +214,10 @@ pub async fn reindex_projects( search_backend: &dyn SearchBackend, project_ids: &[ProjectId], ) -> eyre::Result<()> { + info!("Removing documents for batch"); search_backend.remove_project_documents(project_ids).await?; + info!("Creating project documents"); let documents = index_project_documents(ro_pool, redis_pool, project_ids) .instrument(info_span!("index", batch_size = project_ids.len())) .await diff --git a/scripts/__pycache__/create-dummy-projects.cpython-314.pyc b/scripts/__pycache__/create-dummy-projects.cpython-314.pyc new file mode 100644 index 0000000000..5100c7d310 Binary files /dev/null and b/scripts/__pycache__/create-dummy-projects.cpython-314.pyc differ diff --git a/scripts/create-dummy-projects.py b/scripts/create-dummy-projects.py new file mode 100755 index 0000000000..c55ea25d2e --- /dev/null +++ b/scripts/create-dummy-projects.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +import argparse +import json +import time +import urllib.error +import urllib.request +from concurrent.futures import ThreadPoolExecutor, as_completed +from uuid import uuid4 + + +def make_body(boundary, slug, index): + data = { + "name": f"Dummy Load {index:04d}", + "slug": slug, + "summary": "A dummy project for local load testing.", + "description": "This project was generated locally for batch indexing tests.", + "initial_versions": [], + "is_draft": True, + "categories": [], + "license_id": "MIT", + } + + payload = json.dumps(data, separators=(",", ":")) + return ( + f"--{boundary}\r\n" + 'Content-Disposition: form-data; name="data"\r\n' + "Content-Type: application/json\r\n\r\n" + f"{payload}\r\n" + f"--{boundary}--\r\n" + ).encode() + + +def create_project(base_url, token, boundary, prefix, index, retries): + slug = f"{prefix}-{index:04d}" + body = make_body(boundary, slug, index) + headers = { + "Authorization": f"Bearer {token}", + "Content-Type": f"multipart/form-data; boundary={boundary}", + } + + for attempt in range(retries + 1): + req = urllib.request.Request( + f"{base_url}/v3/project", + data=body, + headers=headers, + method="POST", + ) + + try: + with urllib.request.urlopen(req, timeout=60) as resp: + resp.read() + return True, slug, resp.status, "" + except urllib.error.HTTPError as err: + text = err.read().decode("utf-8", errors="replace") + if err.code < 500 or attempt == retries: + return False, slug, err.code, text + except Exception as err: + if attempt == retries: + return False, slug, "error", repr(err) + + time.sleep(min(2**attempt, 10)) + + raise RuntimeError("unreachable") + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--base-url", default="http://localhost:8000") + parser.add_argument("--token", default="mra_admin") + parser.add_argument("--count", type=int, default=1000) + parser.add_argument("--concurrency", type=int, default=2) + parser.add_argument("--retries", type=int, default=5) + args = parser.parse_args() + + boundary = "----modrinth-dummy-project-boundary" + prefix = f"dummy-load-{int(time.time())}-{uuid4().hex[:6]}" + + ok = 0 + failures = [] + + with ThreadPoolExecutor(max_workers=args.concurrency) as executor: + futures = [ + executor.submit( + create_project, + args.base_url, + args.token, + boundary, + prefix, + index, + args.retries, + ) + for index in range(args.count) + ] + + for completed, future in enumerate(as_completed(futures), 1): + success, slug, status, text = future.result() + if success: + ok += 1 + else: + failures.append((slug, status, text[:500])) + + if completed % 50 == 0: + print( + f"completed={completed} created={ok} failed={len(failures)}", + flush=True, + ) + + print( + json.dumps( + { + "prefix": prefix, + "attempted": args.count, + "created": ok, + "failed": len(failures), + "failures": failures[:20], + }, + indent=2, + ) + ) + + +if __name__ == "__main__": + main()