Delete background jobs on org delete + ensure org delete happens in background job

backend/btrixcloud/background_jobs.py

@@ -519,8 +519,13 @@ async def job_finished(
                 await self._send_bg_job_failure_email(cleanup_job, finished)
             return
 
+        # If org has been successfully deleted in job, delete k8s resources
+        # associated with this org now that no jobs are running
+        if job_type == BgJobType.DELETE_ORG and oid and success:
+            await self.crawl_manager.delete_all_k8s_resources_for_org(str(oid))
+
         job = await self.get_background_job(job_id)
-        if job.finished:
+        if not job or job.finished:
             return
 
         if job.type != job_type:

backend/btrixcloud/db.py

@@ -44,7 +44,7 @@
     ) = object
 
 
-CURR_DB_VERSION = "0056"
+CURR_DB_VERSION = "0057"
 
 MIN_DB_VERSION = 7.0
 

backend/btrixcloud/migrations/migration_0057_deleted_org_bg_job_cleanup.py

@@ -0,0 +1,60 @@
+"""
+Migration 0057 - Remove background jobs for deleted orgs from db
+"""
+
+from uuid import UUID
+
+from motor.motor_asyncio import AsyncIOMotorDatabase
+
+from btrixcloud.migrations import BaseMigration
+from btrixcloud.models import BgJobType
+
+MIGRATION_VERSION = "0057"
+
+
+class Migration(BaseMigration):
+    """Migration class."""
+
+    # pylint: disable=unused-argument
+    def __init__(self, mdb: AsyncIOMotorDatabase, **kwargs):
+        super().__init__(mdb, migration_version=MIGRATION_VERSION)
+
+    async def migrate_up(self):
+        """Perform migration up.
+
+        Delete background jobs from deleted orgs from the database.
+        """
+        # pylint: disable=duplicate-code
+        jobs_mdb = self.mdb["jobs"]
+        orgs_mdb = self.mdb["organizations"]
+
+        job_orgs_to_delete: list[UUID] = []
+
+        job_oids = await jobs_mdb.distinct("oid", {})
+
+        for oid in job_oids:
+            res = await orgs_mdb.find_one({"_id": oid})
+            if res is None:
+                job_orgs_to_delete.append(oid)
+
+        if job_orgs_to_delete:
+            del_count = len(job_orgs_to_delete)
+            print(
+                f"Deleting background jobs for {del_count} deleted orgs",
+                flush=True,
+            )
+
+            try:
+                res = await jobs_mdb.delete_many(
+                    {
+                        "oid": {"$in": job_orgs_to_delete},
+                        # Maintain consistency with behavior moving forward, to
+                        # retain only the one org deletion background job from
+                        # deleted orgs
+                        "type": {"$ne": BgJobType.DELETE_ORG},
+                    }
+                )
+                print(f"Deleted {res.deleted_count} jobs from database", flush=True)
+            # pylint: disable=broad-exception-caught
+            except Exception as err:
+                print(f"Error deleting jobs from deleted orgs: {err}", flush=True)

backend/btrixcloud/orgs.py

@@ -101,6 +101,7 @@
     OrgPublicProfileUpdate,
     MAX_BROWSER_WINDOWS,
     MAX_CRAWL_SCALE,
+    BgJobType,
 )
 from .pagination import DEFAULT_PAGE_SIZE, paginated_format
 from .utils import (
@@ -228,6 +229,7 @@ def __init__(
         self.pages_db = mdb["pages"]
         self.version_db = mdb["version"]
         self.invites_db = mdb["invites"]
+        self.jobs_db = mdb["jobs"]
 
         self.router = None
         self.org_viewer_dep = None
@@ -1494,7 +1496,13 @@ async def import_org(
     async def delete_org_and_data(
         self, org: Organization, user_manager: UserManager
     ) -> None:
-        """Delete org and all of its associated data."""
+        """Delete org and all of its associated data.
+
+        This method should only be run in a background job. The operator
+        will delete associated k8s resources for this org when the job
+        successfully completes to prevent deleting this job before it
+        completes.
+        """
         print(f"Deleting org: {org.slug} {org.name} {org.id}")
 
         # Delete archived items
@@ -1549,8 +1557,12 @@ async def delete_org_and_data(
         # Delete org
         await self.orgs.delete_one({"_id": org.id})
 
-        # Delete related k8s objects
-        await self.crawl_manager.delete_all_k8s_resources_for_org(str(org.id))
+        # Delete all background jobs except this one from database,
+        # so that we are left with some record of the org having
+        # existed and successfully deleted
+        await self.orgs.delete_one(
+            {"_id": org.id, "type": {"$ne": BgJobType.DELETE_ORG}}
+        )
 
     async def recalculate_storage(self, org: Organization) -> dict[str, bool]:
         """Recalculate org storage use"""

backend/btrixcloud/subs.py

@@ -198,7 +198,7 @@ async def cancel_subscription(self, cancel: SubscriptionCancel) -> dict[str, boo
         )
 
         if not org.subscription.readOnlyOnCancel:
-            await self.org_ops.delete_org_and_data(org, self.user_manager)
+            await self.org_ops.background_job_ops.create_delete_org_job(org)
             deleted = True
 
         await self.add_sub_event("cancel", cancel, org.id)

backend/test/test_org_subs.py

@@ -1,15 +1,17 @@
 import requests
-
-from .conftest import API_PREFIX
+import time
 from uuid import uuid4
 
+from .conftest import API_PREFIX
 
 new_subs_oid = None
 new_subs_oid_2 = None
 
 new_user_invite_token = None
 existing_user_invite_token = None
 
+MAX_ATTEMPTS = 24
+
 VALID_PASSWORD = "ValidPassW0rd!"
 
 invite_email = "test-User@EXample.com"
@@ -357,9 +359,21 @@ def test_cancel_sub_and_delete_org(admin_auth_headers):
     assert r.status_code == 200
     assert r.json() == {"canceled": True, "deleted": True}
 
-    r = requests.get(f"{API_PREFIX}/orgs/{new_subs_oid}", headers=admin_auth_headers)
-    assert r.status_code == 404
-    assert r.json()["detail"] == "org_not_found"
+    # Wait for org to be deleted
+    count = 0
+    while count < MAX_ATTEMPTS:
+        r = requests.get(
+            f"{API_PREFIX}/orgs/{new_subs_oid}", headers=admin_auth_headers
+        )
+        if r.status_code == 404:
+            assert r.json().get("detail") == "org_not_found"
+            break
+
+        if count + 1 == MAX_ATTEMPTS:
+            assert False
+
+        time.sleep(10)
+        count += 1
 
 
 def test_cancel_sub_and_no_delete_org(admin_auth_headers):