add validate_mongo_builds method and API endpoint

Author: DylanWelzel

biothings/hub/__init__.py

@@ -1153,6 +1153,7 @@ def configure_commands(self):
         if self.managers.get("mongo_build_cleanup_manager"):
             self.commands["list_mongo_builds"] = self.managers["mongo_build_cleanup_manager"].list_mongo_builds
             self.commands["delete_mongo_builds"] = self.managers["mongo_build_cleanup_manager"].delete_mongo_builds
+            self.commands["validate_mongo_builds"] = self.managers["mongo_build_cleanup_manager"].validate_mongo_builds
         # data release commands
         if self.managers.get("release_manager"):
             self.commands["create_release_note"] = self.managers["release_manager"].create_release_note
@@ -1525,6 +1526,10 @@ def configure_api_endpoints(self):
             self.api_endpoints["mongo_builds/delete"] = EndpointDefinition(
                 name="delete_mongo_builds", method="put", force_bodyargs=True
             )
+        if "validate_mongo_builds" in cmdnames:
+            self.api_endpoints["mongo_builds/validate"] = EndpointDefinition(
+                name="validate_mongo_builds", method="post"
+            )
         if "sync" in cmdnames:
             self.api_endpoints["sync"] = EndpointDefinition(name="sync", method="post", force_bodyargs=True)
         if "whatsnew" in cmdnames:

biothings/hub/dataindex/mongo_build_cleanup.py

@@ -10,14 +10,22 @@ class MongoBuildCleaner:
     def __init__(self, job_manager):
         self.job_manager = job_manager
 
-    def list_builds(self, build_config=None, build_name=None):
+    def list_builds(self, build_config=None, build_name=None, year=None):
         collection = get_src_build()
 
         filters = {}
         if build_config:
             filters["build_config._id"] = build_config
         if build_name:
             filters["_id"] = build_name
+        if year:
+            from datetime import datetime
+
+            year = int(year)
+            filters["started_at"] = {
+                "$gte": datetime(year, 1, 1),
+                "$lt": datetime(year + 1, 1, 1),
+            }
 
         projection = {
             "_id": 1,
@@ -81,6 +89,51 @@ async def delete_builds(self, build_ids):
         finally:
             await conn.close()
 
+    async def validate_builds(self):
+        """Validate that target collections exist for each build record.
+
+        Checks every build in src_build to see if its target collection still
+        exists in the target database.  Build records whose target collections
+        have been removed are deleted, keeping the database in sync with the
+        actual data.
+
+        Returns a dict with ``builds_removed`` (count) and ``builds_removed_names``.
+        """
+        from biothings.utils import mongo
+
+        logging.info("Starting validation of MongoDB builds...")
+        conn = mongo.get_hub_db_async_conn()
+        try:
+            src_build = mongo.get_src_build_async(conn)
+            target_db = conn[btconfig.DATA_TARGET_DATABASE]
+
+            existing_collections = set(await target_db.list_collection_names())
+
+            orphaned_ids = []
+            async for doc in src_build.find({}, {"_id": 1, "target_name": 1}):
+                build_id = doc["_id"]
+                target_name = doc.get("target_name") or build_id
+                if target_name not in existing_collections:
+                    orphaned_ids.append(build_id)
+
+            if orphaned_ids:
+                result = await src_build.delete_many({"_id": {"$in": orphaned_ids}})
+                deleted_count = result.deleted_count
+            else:
+                deleted_count = 0
+
+            logging.info(
+                "Build validation complete: removed %d orphaned build record(s)",
+                deleted_count,
+                extra={"notify": True},
+            )
+            return {
+                "builds_removed": deleted_count,
+                "builds_removed_names": sorted(orphaned_ids),
+            }
+        finally:
+            await conn.close()
+
     def done(self, future):
         try:
             result = future.result()
@@ -93,14 +146,25 @@ def done(self, future):
         except Exception as exc:
             logging.exception("Failed to delete MongoDB builds: %s", exc, extra={"notify": True})
 
+    def validate_done(self, future):
+        try:
+            result = future.result()
+            logging.info(
+                "Build validation complete: removed %d orphaned build record(s)",
+                result.get("builds_removed", 0),
+                extra={"notify": True},
+            )
+        except Exception as exc:
+            logging.exception("Failed to validate MongoDB builds: %s", exc, extra={"notify": True})
+
 
 class MongoBuildCleanupManager(BaseManager):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.cleaner = MongoBuildCleaner(self.job_manager)
 
-    def list_mongo_builds(self, build_config=None, build_name=None):
-        return self.cleaner.list_builds(build_config=build_config, build_name=build_name)
+    def list_mongo_builds(self, build_config=None, build_name=None, year=None):
+        return self.cleaner.list_builds(build_config=build_config, build_name=build_name, year=year)
 
     def delete_mongo_builds(self, build_ids):
         try:
@@ -110,3 +174,12 @@ def delete_mongo_builds(self, build_ids):
             logging.exception("Error while submitting MongoDB build deletion job: %s", ex, extra={"notify": True})
             raise
         return job
+
+    def validate_mongo_builds(self):
+        try:
+            job = self.job_manager.submit(partial(self.cleaner.validate_builds))
+            job.add_done_callback(self.cleaner.validate_done)
+        except Exception as ex:
+            logging.exception("Error while submitting MongoDB build validation job: %s", ex, extra={"notify": True})
+            raise
+        return job