feat: Add VFolder force delete API

src/ai/backend/manager/api/vfolder.py

@@ -2463,6 +2463,43 @@ async def delete_from_trash_bin(
     return web.Response(status=204)
 
 
+@auth_required
+@server_status_required(ALL_ALLOWED)
+async def force_delete(request: web.Request) -> web.Response:
+    root_ctx: RootContext = request.app["_root.context"]
+
+    piece = request.match_info["name"]
+    try:
+        folder_id = uuid.UUID(piece)
+    except ValueError:
+        log.error(f"Not allowed UUID type value ({piece})")
+        return web.Response(status=400)
+
+    row = (
+        await resolve_vfolder_rows(
+            request, VFolderPermission.OWNER_PERM, folder_id, allow_privileged_access=True
+        )
+    )[0]
+    try:
+        await check_vfolder_status(row, VFolderStatusSet.PURGABLE)
+    except VFolderFilterStatusFailed:
+        await check_vfolder_status(row, VFolderStatusSet.DELETABLE)
+    log.info(
+        "VFOLDER.FORCE_DELETE (email:{}, ak:{}, vf:{})",
+        request["user"]["email"],
+        request["keypair"]["access_key"],
+        folder_id,
+    )
+    await initiate_vfolder_deletion(
+        root_ctx.db,
+        [VFolderDeletionInfo(VFolderID.from_row(row), row["host"])],
+        root_ctx.storage_manager,
+        force=True,
+    )
+
+    return web.Response(status=204)
+
+
 class PurgeRequestModel(BaseModel):
     vfolder_id: uuid.UUID = Field(
         validation_alias=AliasChoices("vfolder_id", "vfolderId", "id"),
@@ -3584,6 +3621,7 @@ def create_app(default_cors_options):
     cors.add(add_route("POST", r"/purge", purge))
     cors.add(add_route("POST", r"/restore-from-trash-bin", restore))
     cors.add(add_route("POST", r"/delete-from-trash-bin", delete_from_trash_bin))
+    cors.add(add_route("DELETE", r"/{name}/force", force_delete))
     cors.add(add_route("GET", r"/invitations/list-sent", list_sent_invitations))
     cors.add(add_route("GET", r"/invitations/list_sent", list_sent_invitations))  # legacy underbar
     cors.add(add_route("POST", r"/invitations/update/{inv_id}", update_invitation))

src/ai/backend/manager/models/vfolder.py

@@ -204,6 +204,18 @@ class VFolderOperationStatus(enum.StrEnum):
     DELETE_COMPLETE = "delete-complete"  # vfolder is deleted permanently, only DB row remains
     DELETE_ERROR = "delete-error"
 
+    def is_deletable(self, force: bool = False) -> bool:
+        if force:
+            return self in {
+                VFolderOperationStatus.READY,
+                VFolderOperationStatus.DELETE_PENDING,
+            }
+        else:
+            return self == VFolderOperationStatus.DELETE_PENDING
+
+    def attach_timestamp(self) -> bool:
+        return self == VFolderOperationStatus.DELETE_ONGOING
+
 
 class VFolderStatusSet(enum.StrEnum):
     """
@@ -1025,6 +1037,7 @@ async def update_vfolder_status(
     vfolder_ids: Sequence[uuid.UUID],
     update_status: VFolderOperationStatus,
     do_log: bool = True,
+    force: bool = False,
 ) -> None:
     vfolder_info_len = len(vfolder_ids)
     cond = vfolders.c.id.in_(vfolder_ids)
@@ -1035,7 +1048,7 @@ async def update_vfolder_status(
 
     now = datetime.now(timezone.utc)
 
-    if update_status == VFolderOperationStatus.DELETE_PENDING:
+    if update_status.is_deletable(force):
         select_stmt = sa.select(VFolderRow).where(VFolderRow.id.in_(vfolder_ids))
         async with engine.begin_readonly_session() as db_session:
             for vf_row in await db_session.scalars(select_stmt):
@@ -1072,7 +1085,7 @@ async def _update() -> None:
                     },
                 ),
             }
-            if update_status == VFolderOperationStatus.DELETE_ONGOING:
+            if update_status.attach_timestamp():
                 values["name"] = VFolderRow.name + f"_deleted_{now.strftime('%Y-%m-%dT%H%M%S%z')}"
             query = sa.update(vfolders).values(**values).where(cond)
             await db_session.execute(query)
@@ -1253,7 +1266,9 @@ async def initiate_vfolder_deletion(
     db_engine: ExtendedAsyncSAEngine,
     requested_vfolders: Sequence[VFolderDeletionInfo],
     storage_manager: StorageSessionManager,
-    storage_ptask_group: aiotools.PersistentTaskGroup,
+    storage_ptask_group: Optional[aiotools.PersistentTaskGroup] = None,
+    *,
+    force: bool = False,
 ) -> int:
     """Purges VFolder content from storage host."""
     vfolder_info_len = len(requested_vfolders)
@@ -1267,7 +1282,11 @@ async def initiate_vfolder_deletion(
     async with db_engine.connect() as db_conn:
         await delete_vfolder_relation_rows(db_conn, db_engine.begin_session, vfolder_ids)
     await update_vfolder_status(
-        db_engine, vfolder_ids, VFolderOperationStatus.DELETE_ONGOING, do_log=False
+        db_engine,
+        vfolder_ids,
+        VFolderOperationStatus.DELETE_ONGOING,
+        do_log=False,
+        force=force,
     )
 
     already_deleted: list[VFolderDeletionInfo] = []