Convert branch deletion to celery job

Author: mxsrc

src/api/organization/project/branch/__init__.py

@@ -73,7 +73,7 @@
 )
 from .....models.resources import ResourceLimitsPublic, ResourceType
 from ...._util import Conflict, Forbidden, NotFound, Unauthenticated, url_path_for
-from ...._util.backups import copy_branch_backup_schedules, delete_branch_backups, ensure_branch_pitr_schedule
+from ...._util.backups import copy_branch_backup_schedules, ensure_branch_pitr_schedule
 from ...._util.resourcelimit import (
     apply_branch_resource_allocation,
     check_available_resources_limits,
@@ -94,6 +94,7 @@
 from ....settings import get_settings as get_api_settings
 from .api_keys import api as api_key_api
 from .auth import api as auth_api
+from .delete_tasks import dispatch_delete
 from .resize_tasks import dispatch_resize
 from .tasks import task_api
 
@@ -1653,31 +1654,34 @@ async def update(
 @instance_api.delete(
     "/",
     name="organizations:projects:branch:delete",
-    status_code=204,
-    responses={401: Unauthenticated, 403: Forbidden, 404: NotFound},
+    status_code=202,
+    responses={401: Unauthenticated, 403: Forbidden, 404: NotFound, 400: {"description": "Delete already in progress"}},
 )
 async def delete(
     session: SessionDep,
-    _organization: OrganizationDep,
-    _project: ProjectDep,
+    request: Request,
+    organization: OrganizationDep,
+    project: ProjectDep,
     branch: BranchDep,
 ):
-    branch_id = branch.id
-    await _set_branch_status(session, branch, BranchServiceStatus.DELETING)
-    await delete_deployment(branch_id)
-    try:
-        await realm_admin("master").a_delete_realm(str(branch_id))
-    except KeycloakError as exc:
-        if getattr(exc, "response_code", None) == 404:
-            logger.error("Keycloak realm not found for branch %s during delete; continuing.", branch_id, exc_info=True)
-        else:
-            raise
-    await delete_branch_provisioning(session, branch_id)
-    await delete_branch_backups(session, branch_id)
-    await session.delete(branch)
+    if branch.delete_task_id is not None:
+        raise HTTPException(status_code=400, detail="A delete operation is already in progress")
+
+    task_id = dispatch_delete(str(branch.id))
+
+    branch.set_status(BranchServiceStatus.DELETING)
+    branch.delete_task_id = task_id
     await session.commit()
 
-    return Response(status_code=204)
+    task_url = url_path_for(
+        request,
+        "organizations:projects:branch:tasks:detail",
+        organization_id=await organization.awaitable_attrs.id,
+        project_id=await project.awaitable_attrs.id,
+        branch_id=await branch.awaitable_attrs.id,
+        task_id=task_id,
+    )
+    return Response(status_code=202, headers={"Location": task_url})
 
 
 @instance_api.post(

src/api/organization/project/branch/delete_tasks.py

@@ -0,0 +1,61 @@
+"""Celery chord tasks for async branch deletion.
+
+The chord dispatches three sub-tasks in parallel (K8s, Keycloak, backups) and
+calls ``finalize_delete`` when all sub-tasks have settled.  With
+``task_chord_propagates = False`` the callback always fires, so the branch DB
+record is removed regardless of individual sub-task outcomes (errors are logged
+but do not abort the delete).
+"""
+
+import logging
+from uuid import UUID
+
+from asgiref.sync import async_to_sync
+from celery import chord
+from ulid import ULID
+
+from .....database import AsyncSessionLocal
+from .....deployment.delete import delete_backup_snapshots, delete_k8s_deployment, delete_keycloak_realm
+from .....models.branch import Branch
+from .....worker import app
+from ...._util.resourcelimit import delete_branch_provisioning
+
+logger = logging.getLogger(__name__)
+
+
+async def _async_finalize_delete(job_results: list, branch_id: str) -> dict:
+    errors: list[str] = []
+    for result in job_results:
+        if isinstance(result, Exception):
+            errors.append(str(result))
+            logger.error("Delete sub-task error for branch %s: %s", branch_id, result)
+
+    branch_ulid = ULID.from_str(branch_id)
+    async with AsyncSessionLocal() as session:
+        branch = await session.get(Branch, branch_ulid)
+        if branch is None:
+            logger.warning("Branch %s not found in finalize_delete; already deleted.", branch_id)
+            return {"errors": errors}
+
+        await delete_branch_provisioning(session, branch_ulid, commit=False)
+        await session.delete(branch)
+        await session.commit()
+
+    return {"errors": errors}
+
+
+@app.task(name="simplyblock.vela.branch.delete.finalize")
+def finalize_delete(job_results: list, branch_id: str) -> dict:
+    """Chord callback: delete provisioning records and the branch DB row."""
+    return async_to_sync(_async_finalize_delete)(job_results, branch_id)
+
+
+def dispatch_delete(branch_id: str) -> UUID:
+    """Dispatch the delete chord; return the chord result UUID."""
+    jobs = [
+        delete_k8s_deployment.s(branch_id),
+        delete_keycloak_realm.s(branch_id),
+        delete_backup_snapshots.s(branch_id),
+    ]
+    result = chord(jobs)(finalize_delete.s(branch_id=branch_id))
+    return UUID(result.id)

src/api/organization/project/branch/tasks.py

@@ -1,6 +1,6 @@
 """Branch task list/detail endpoints.
 
-Exposes Celery task state (currently resize only) under:
+Exposes Celery task state (resize and delete) under:
   GET .../branches/{branch_id}/tasks
   GET .../branches/{branch_id}/tasks/{task_id}
 """
@@ -14,6 +14,7 @@
 
 from ...._util import Forbidden, NotFound, Unauthenticated
 from ....dependencies import BranchDep, OrganizationDep, ProjectDep
+from .delete_tasks import finalize_delete
 from .resize_tasks import finalize_resize
 
 task_api = APIRouter(tags=["branch"])
@@ -37,16 +38,20 @@ class BranchTaskPublic(BaseModel):
     date_done: datetime | None
 
 
-def _build_task_public(task_id: UUID) -> BranchTaskPublic:
-    result = finalize_resize.AsyncResult(str(task_id))
+def _build_task_public(task_id: UUID, task_type: str) -> BranchTaskPublic:
+    if task_type == "delete":
+        result = finalize_delete.AsyncResult(str(task_id))
+    else:
+        result = finalize_resize.AsyncResult(str(task_id))
     state = result.state
     status = _CELERY_STATE_TO_STATUS.get(state, state)
     kwargs: dict = result.kwargs or {}
+    parameters = kwargs.get("effective_parameters", {}) if task_type == "resize" else {}
     return BranchTaskPublic(
         id=task_id,
-        task_type="resize",
+        task_type=task_type,
         status=status,
-        parameters=kwargs.get("effective_parameters", {}),
+        parameters=parameters,
         result=result.result if state == "SUCCESS" else None,
         error=str(result.traceback) if state == "FAILURE" and result.traceback else None,
         date_done=result.date_done,
@@ -64,9 +69,12 @@ async def list_tasks(
     _project: ProjectDep,
     branch: BranchDep,
 ) -> list[BranchTaskPublic]:
-    if branch.resize_task_id is None:
-        return []
-    return [_build_task_public(branch.resize_task_id)]
+    tasks = []
+    if branch.resize_task_id is not None:
+        tasks.append(_build_task_public(branch.resize_task_id, "resize"))
+    if branch.delete_task_id is not None:
+        tasks.append(_build_task_public(branch.delete_task_id, "delete"))
+    return tasks
 
 
 @task_api.get(
@@ -81,6 +89,8 @@ async def get_task(
     branch: BranchDep,
     task_id: UUID,
 ) -> BranchTaskPublic:
-    if branch.resize_task_id != task_id:
-        raise HTTPException(status_code=404, detail="Task not found")
-    return _build_task_public(task_id)
+    if branch.resize_task_id == task_id:
+        return _build_task_public(task_id, "resize")
+    if branch.delete_task_id == task_id:
+        return _build_task_public(task_id, "delete")
+    raise HTTPException(status_code=404, detail="Task not found")

src/deployment/delete.py

@@ -0,0 +1,54 @@
+"""Celery sub-tasks for async branch deletion (deployment layer).
+
+These tasks handle the infrastructure-level teardown steps that can run in
+parallel: K8s namespace/VM deletion, Keycloak realm removal, and backup
+snapshot cleanup.  The API layer's ``delete_tasks.finalize_delete`` chord
+callback then removes the DB record after all sub-tasks have settled.
+"""
+
+import logging
+
+from asgiref.sync import async_to_sync
+from keycloak.exceptions import KeycloakError
+from ulid import ULID
+
+from ..api._util.backups import delete_branch_backups
+from ..api.keycloak import realm_admin
+from ..database import AsyncSessionLocal
+from ..worker import app
+from . import delete_deployment
+
+logger = logging.getLogger(__name__)
+
+
+async def _delete_keycloak(branch_id: ULID) -> None:
+    try:
+        await realm_admin("master").a_delete_realm(str(branch_id))
+    except KeycloakError as exc:
+        if getattr(exc, "response_code", None) == 404:
+            logger.warning("Keycloak realm not found for branch %s; skipping.", branch_id)
+        else:
+            raise
+
+
+async def _delete_snapshots(branch_id: ULID) -> None:
+    async with AsyncSessionLocal() as session:
+        await delete_branch_backups(session, branch_id)
+
+
+@app.task(name="simplyblock.vela.deployment.delete.k8s")
+def delete_k8s_deployment(branch_id: str) -> None:
+    """Delete the K8s namespace and associated VM for a branch."""
+    async_to_sync(delete_deployment)(ULID.from_str(branch_id))
+
+
+@app.task(name="simplyblock.vela.deployment.delete.keycloak")
+def delete_keycloak_realm(branch_id: str) -> None:
+    """Delete the Keycloak realm for a branch (404 is treated as success)."""
+    async_to_sync(_delete_keycloak)(ULID.from_str(branch_id))
+
+
+@app.task(name="simplyblock.vela.deployment.delete.backups")
+def delete_backup_snapshots(branch_id: str) -> None:
+    """Delete K8s volume snapshots for all backups belonging to a branch."""
+    async_to_sync(_delete_snapshots)(ULID.from_str(branch_id))

src/models/branch.py

@@ -3,7 +3,6 @@
 from enum import StrEnum
 from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Literal, Optional
 
-import sqlalchemy as sa
 from pydantic import BaseModel, model_validator
 from pydantic import Field as PydanticField
 from sqlalchemy import BigInteger, Boolean, Column, String, Text, UniqueConstraint, text
@@ -109,7 +108,8 @@ class Branch(AsyncAttrs, Model, table=True):
     )
     pitr_enabled: bool = Field(default=False, sa_column=Column(Boolean, nullable=False, server_default=text("false")))
     resize_task_id: uuid.UUID | None = Field(default=None, nullable=True)
-    db_port: int | None = Field(default=None, sa_column=Column(sa.Integer, nullable=True))
+    delete_task_id: uuid.UUID | None = None
+    db_port: int | None = None
 
     __table_args__ = (UniqueConstraint("project_id", "name", name="unique_branch_name_per_project"),)
 

src/models/migrations/versions/b8e4c9f12d35_add_branch_delete_task_id.py

@@ -0,0 +1,28 @@
+"""Add branch delete_task_id
+
+Revision ID: b8e4c9f12d35
+Revises: a1b2c3d4e5f6
+Create Date: 2026-04-01 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'b8e4c9f12d35'
+down_revision: Union[str, Sequence[str], None] = 'a1b2c3d4e5f6'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.add_column('branch', sa.Column('delete_task_id', sa.Uuid(), nullable=True))
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_column('branch', 'delete_task_id')

src/worker/__init__.py

@@ -20,5 +20,7 @@ class Settings(BaseSettings):
 app.conf.task_chord_propagates = False
 
 # Register tasks — must be imported after `app` is defined.
+from ..api.organization.project.branch import delete_tasks as _api_delete_tasks  # noqa: E402, F401
 from ..api.organization.project.branch import resize_tasks as _api_resize_tasks  # noqa: E402, F401
+from ..deployment import delete as _deployment_delete  # noqa: E402, F401
 from ..deployment import resize as _deployment_resize  # noqa: E402, F401

tests/branches/test_delete.py

@@ -0,0 +1,85 @@
+"""Integration tests for async branch deletion.
+
+These tests verify:
+  - DELETE /branches/{id}/ returns 202 Accepted with a Location header
+  - A concurrent DELETE returns 400 while the first delete is in progress
+  - The branch is eventually gone (GET returns 404) after the task completes
+"""
+
+import time
+
+import pytest
+from conftest import BRANCH_TIMEOUT_SEC, wait_for_deletion
+
+pytestmark = pytest.mark.branch
+
+_POLL_INTERVAL = 10
+
+_state: dict = {}
+
+
+@pytest.fixture(scope="module")
+def org(make_org):
+    return make_org("test-org-delete")
+
+
+@pytest.fixture(scope="module")
+def project(make_project, org):
+    return make_project(org, "test-project-delete")
+
+
+@pytest.fixture(scope="module")
+def branch_id(make_branch, org, project):
+    return make_branch(org, project, "test-branch-delete")
+
+
+def test_delete_returns_202_with_location(client, org, project, branch_id):
+    r = client.delete(f"organizations/{org}/projects/{project}/branches/{branch_id}/")
+    assert r.status_code == 202
+    assert "Location" in r.headers
+    _state["task_url"] = r.headers["Location"]
+
+
+def test_task_listed_while_deleting(client, org, project, branch_id):
+    r = client.get(f"organizations/{org}/projects/{project}/branches/{branch_id}/tasks/")
+    if r.status_code == 404:
+        pytest.skip("Branch already deleted before task listing test could run")
+    assert r.status_code == 200
+    tasks = r.json()
+    assert any(t["task_type"] == "delete" for t in tasks)
+
+
+def test_concurrent_delete_returns_400(client, org, project, branch_id):
+    # While the branch has delete_task_id set the endpoint must reject a second delete.
+    # The branch may have been deleted already in fast CI environments (404 is also acceptable).
+    r = client.delete(f"organizations/{org}/projects/{project}/branches/{branch_id}/")
+    assert r.status_code in (400, 404)
+
+
+def test_branch_gone_after_delete(client, org, project, branch_id):
+    # Poll until the branch returns 404 — the Celery task has completed and deleted it.
+    wait_for_deletion(
+        client,
+        f"organizations/{org}/projects/{project}/branches/{branch_id}/",
+        BRANCH_TIMEOUT_SEC,
+    )
+    r = client.get(f"organizations/{org}/projects/{project}/branches/{branch_id}/")
+    assert r.status_code == 404
+
+
+def test_task_detail_unavailable_after_deletion(client):
+    task_url = _state.get("task_url")
+    if not task_url:
+        pytest.skip("No task URL stored from delete test")
+    # Once the branch is gone the task detail endpoint returns 404 (BranchDep lookup fails).
+    deadline = time.monotonic() + BRANCH_TIMEOUT_SEC
+    while True:
+        r = client.get(task_url, timeout=30)
+        if r.status_code == 404:
+            return
+        status = r.json().get("status") if r.status_code == 200 else None
+        if status in ("COMPLETED", "FAILED"):
+            return
+        if time.monotonic() >= deadline:
+            raise TimeoutError(f"Timed out waiting for task to complete; last status={status!r}")
+        time.sleep(_POLL_INTERVAL)