Skip to content

Commit 58f6ab3

Browse files
hjlarryhjlarry
committed
feat: support permanent deletion of monthly workflow archives (#44052)
(cherry picked from commit c11b36c)
1 parent 1970629 commit 58f6ab3

56 files changed

Lines changed: 2799 additions & 149 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎api/controllers/console/__init__.py‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,19 @@
1+
from collections import OrderedDict
12
from http import HTTPStatus
23
from importlib import import_module
4+
from typing import cast
35

46
from flask import Blueprint, current_app, got_request_exception
57
from flask_restx import Namespace
68

79
from libs.external_api import ExternalApi
810
from machinery.errors import ActiveWorkspaceRequiredError
11+
from services.retention.workflow_run.archive_log_service import (
12+
WorkflowRunArchiveDownloadNotReadyError,
13+
WorkflowRunArchiveDownloadTaskNotFoundError,
14+
WorkflowRunArchiveInvalidMonthError,
15+
WorkflowRunArchiveNotFoundError,
16+
)
917

1018
bp = Blueprint("console", __name__, url_prefix="/console/api")
1119

@@ -29,6 +37,46 @@ def _handle_active_workspace_required_error(error: ActiveWorkspaceRequiredError)
2937
}, status.value
3038

3139

40+
@api.errorhandler(WorkflowRunArchiveInvalidMonthError)
41+
def _handle_workflow_run_archive_invalid_month_error(
42+
error: WorkflowRunArchiveInvalidMonthError,
43+
) -> tuple[dict[str, str | int], int]:
44+
got_request_exception.send(current_app, exception=error)
45+
status = HTTPStatus.BAD_REQUEST
46+
return {"code": "bad_request", "message": str(error), "status": status.value}, status.value
47+
48+
49+
@api.errorhandler(WorkflowRunArchiveNotFoundError)
50+
@api.errorhandler(WorkflowRunArchiveDownloadTaskNotFoundError)
51+
def _handle_workflow_run_archive_not_found_error(
52+
error: WorkflowRunArchiveNotFoundError | WorkflowRunArchiveDownloadTaskNotFoundError,
53+
) -> tuple[dict[str, str | int], int]:
54+
got_request_exception.send(current_app, exception=error)
55+
status = HTTPStatus.NOT_FOUND
56+
return {"code": "not_found", "message": str(error), "status": status.value}, status.value
57+
58+
59+
@api.errorhandler(WorkflowRunArchiveDownloadNotReadyError)
60+
def _handle_workflow_run_archive_download_not_ready_error(
61+
error: WorkflowRunArchiveDownloadNotReadyError,
62+
) -> tuple[dict[str, str | int], int]:
63+
got_request_exception.send(current_app, exception=error)
64+
status = HTTPStatus.CONFLICT
65+
return {"code": "conflict", "message": str(error), "status": status.value}, status.value
66+
67+
68+
# Flask-RESTX dispatches in registration order; the shared Exception handler
69+
# must not shadow these Console-specific error mappings.
70+
for error_type in (
71+
ActiveWorkspaceRequiredError,
72+
WorkflowRunArchiveInvalidMonthError,
73+
WorkflowRunArchiveNotFoundError,
74+
WorkflowRunArchiveDownloadTaskNotFoundError,
75+
WorkflowRunArchiveDownloadNotReadyError,
76+
):
77+
cast(OrderedDict[type[Exception], object], api.error_handlers).move_to_end(error_type, last=False)
78+
79+
3280
console_ns = Namespace("console", description="Console management API operations", path="/")
3381

3482
RESOURCE_MODULES = (

‎api/controllers/console/workflow_run_archive.py‎

Lines changed: 66 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,10 @@
11
import datetime
22
from http import HTTPStatus
3+
from typing import Literal
34

45
from flask import redirect
56
from flask_restx import Resource
6-
from pydantic import BaseModel, Field
7-
from werkzeug.exceptions import Conflict, NotFound
7+
from pydantic import BaseModel, Field, ValidationError
88

99
from controllers.common.fields import RedirectResponse
1010
from controllers.common.schema import register_response_schema_models, register_schema_models
@@ -20,13 +20,12 @@
2020
from libs.helper import dump_response
2121
from machinery.context import RequestContext
2222
from models.account import TenantAccountRole
23+
from services.retention.workflow_run.archive_deletion_task import WorkflowRunArchiveDeletionStatus
2324
from services.retention.workflow_run.archive_download_task import (
2425
WorkflowRunArchiveDownloadStatus,
2526
)
2627
from services.retention.workflow_run.archive_log_service import (
27-
WorkflowRunArchiveDownloadNotReadyError,
28-
WorkflowRunArchiveDownloadTaskNotFoundError,
29-
WorkflowRunArchiveNotFoundError,
28+
WorkflowRunArchiveInvalidMonthError,
3029
)
3130

3231
_WORKFLOW_RUN_ARCHIVE_ALLOWED_ROLES = frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN})
@@ -35,7 +34,7 @@
3534
class WorkflowRunArchiveDownloadPayload(BaseModel):
3635
"""Request body for preparing one monthly workflow-run archive download."""
3736

38-
year: int = Field(ge=1)
37+
year: int = Field(ge=1, le=9999)
3938
month: int = Field(ge=1, le=12)
4039

4140

@@ -72,6 +71,21 @@ class WorkflowRunArchiveMonthResponse(ResponseModel):
7271
archive_bytes: int
7372
latest_archived_at: datetime.datetime
7473
download_task: WorkflowRunArchiveDownloadTaskResponse | None = None
74+
deletion_status: (
75+
Literal[
76+
WorkflowRunArchiveDeletionStatus.PENDING,
77+
WorkflowRunArchiveDeletionStatus.PROCESSING,
78+
WorkflowRunArchiveDeletionStatus.FAILED,
79+
]
80+
| None
81+
) = None
82+
83+
84+
class WorkflowRunArchiveDeletionTaskResponse(ResponseModel):
85+
year: int
86+
month: int
87+
status: WorkflowRunArchiveDeletionStatus
88+
error: str | None = None
7589

7690

7791
class WorkflowRunArchiveListResponse(ResponseModel):
@@ -86,6 +100,7 @@ class WorkflowRunArchiveListResponse(ResponseModel):
86100
WorkflowRunArchiveMonthResponse,
87101
WorkflowRunArchiveListResponse,
88102
WorkflowRunArchiveDownloadTaskResponse,
103+
WorkflowRunArchiveDeletionTaskResponse,
89104
RedirectResponse,
90105
)
91106

@@ -105,6 +120,38 @@ def get(self, request_context: RequestContext):
105120
return dump_response(WorkflowRunArchiveListResponse, archives)
106121

107122

123+
@console_ns.route("/workflow-run-archives/<int:year>/<int:month>")
124+
class WorkflowRunArchiveMonthApi(Resource):
125+
@console_ns.doc("delete_workflow_run_archive")
126+
@console_ns.doc(description="Permanently delete this workspace's monthly archive files and archive index")
127+
@console_ns.doc(
128+
params={
129+
"year": {"in": "path", "type": "integer", "minimum": 1, "maximum": 9999},
130+
"month": {"in": "path", "type": "integer", "minimum": 1, "maximum": 12},
131+
}
132+
)
133+
@console_ns.response(
134+
HTTPStatus.ACCEPTED,
135+
"Archive deletion accepted",
136+
console_ns.models[WorkflowRunArchiveDeletionTaskResponse.__name__],
137+
)
138+
@console_account_admission(
139+
editions=frozenset({DeploymentEdition.CLOUD}),
140+
# shortcut: allowed_roles only applies with RBAC disabled; add RBAC checks here when enabling it.
141+
allowed_roles=frozenset({TenantAccountRole.OWNER}),
142+
)
143+
@cloud_edition_billing_paid_plan_required
144+
def delete(self, request_context: RequestContext, year: int, month: int):
145+
try:
146+
archive_month = WorkflowRunArchiveDownloadPayload(year=year, month=month)
147+
except ValidationError as exc:
148+
raise WorkflowRunArchiveInvalidMonthError(str(exc)) from exc
149+
task = application_services().workflow_run_archives.delete_archive(
150+
request_context, year=archive_month.year, month=archive_month.month
151+
)
152+
return dump_response(WorkflowRunArchiveDeletionTaskResponse, task), HTTPStatus.ACCEPTED
153+
154+
108155
@console_ns.route("/workflow-run-archives/downloads")
109156
class WorkflowRunArchiveDownloadsApi(Resource):
110157
@console_ns.doc("create_workflow_run_archive_download")
@@ -122,14 +169,11 @@ class WorkflowRunArchiveDownloadsApi(Resource):
122169
@cloud_edition_billing_paid_plan_required
123170
@model_validate(WorkflowRunArchiveDownloadPayload)
124171
def post(self, req_data: WorkflowRunArchiveDownloadPayload, request_context: RequestContext):
125-
try:
126-
task = application_services().workflow_run_archives.create_download(
127-
request_context,
128-
year=req_data.year,
129-
month=req_data.month,
130-
)
131-
except WorkflowRunArchiveNotFoundError as exc:
132-
raise NotFound(str(exc)) from exc
172+
task = application_services().workflow_run_archives.create_download(
173+
request_context,
174+
year=req_data.year,
175+
month=req_data.month,
176+
)
133177
return dump_response(WorkflowRunArchiveDownloadTaskResponse, task), HTTPStatus.ACCEPTED
134178

135179

@@ -144,13 +188,10 @@ class WorkflowRunArchiveDownloadApi(Resource):
144188
)
145189
@cloud_edition_billing_paid_plan_required
146190
def get(self, request_context: RequestContext, download_id: str):
147-
try:
148-
task = application_services().workflow_run_archives.get_download(
149-
request_context,
150-
download_id=download_id,
151-
)
152-
except WorkflowRunArchiveDownloadTaskNotFoundError as exc:
153-
raise NotFound(str(exc)) from exc
191+
task = application_services().workflow_run_archives.get_download(
192+
request_context,
193+
download_id=download_id,
194+
)
154195
return dump_response(WorkflowRunArchiveDownloadTaskResponse, task)
155196

156197

@@ -170,13 +211,8 @@ class WorkflowRunArchiveDownloadFileApi(Resource):
170211
)
171212
@cloud_edition_billing_paid_plan_required
172213
def get(self, request_context: RequestContext, download_id: str):
173-
try:
174-
presigned_url = application_services().workflow_run_archives.get_download_url(
175-
request_context,
176-
download_id=download_id,
177-
)
178-
except WorkflowRunArchiveDownloadTaskNotFoundError as exc:
179-
raise NotFound(str(exc)) from exc
180-
except WorkflowRunArchiveDownloadNotReadyError as exc:
181-
raise Conflict(str(exc)) from exc
214+
presigned_url = application_services().workflow_run_archives.get_download_url(
215+
request_context,
216+
download_id=download_id,
217+
)
182218
return redirect(presigned_url, code=HTTPStatus.FOUND)

‎api/extensions/ext_application_services.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -178,6 +178,7 @@
178178
from services.recommended_app_query_service import RecommendedAppQueryService
179179
from services.remote_file_service import RemoteFileService
180180
from services.retention.workflow_run.archive_download_adapters import (
181+
dispatch_workflow_run_archive_deletion_task,
181182
dispatch_workflow_run_archive_download_task,
182183
sign_workflow_run_archive_download_url,
183184
)
@@ -717,6 +718,7 @@ def build_application_services(
717718
tasks=WorkflowRunArchiveDownloadTaskCache(redis=redis),
718719
dispatcher=dispatch_workflow_run_archive_download_task,
719720
sign_download_url=sign_workflow_run_archive_download_url,
721+
deletion_dispatcher=dispatch_workflow_run_archive_deletion_task,
720722
),
721723
workflow_runs=WorkflowRunService(
722724
workflow_runs=workflow_run_repository,

‎api/extensions/ext_celery.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -177,6 +177,7 @@ def __call__(self, *args: object, **kwargs: object) -> object:
177177
"tasks.refresh_billing_vector_space_task", # billing vector-space cache refresh
178178
"tasks.app_generate.resume_agent_app_task", # ENG-635: Agent v2 chat ask_human resume
179179
"tasks.workflow_run_archive_download_tasks", # workflow-run archive download preparation
180+
"tasks.workflow_run_archive_deletion_tasks", # permanent monthly archive deletion
180181
]
181182
day = dify_config.CELERY_BEAT_SCHEDULER_TIME
182183

‎api/migrations/versions/2026_10_12_1200-f1d3c5b7a902_add_archive_bundle_deletion_status.py‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
"""add archive bundle deletion status
2+
3+
Revision ID: f1d3c5b7a902
4+
Revises: d8e4a6b1c902
5+
Create Date: 2026-10-12 12:00:00.000000
6+
7+
"""
8+
9+
import sqlalchemy as sa
10+
from alembic import op
11+
12+
revision = "f1d3c5b7a902"
13+
down_revision = "d8e4a6b1c902"
14+
branch_labels = None
15+
depends_on = None
16+
17+
18+
def upgrade() -> None:
19+
op.add_column("workflow_run_archive_bundles", sa.Column("deletion_status", sa.String(16), nullable=True))
20+
21+
22+
def downgrade() -> None:
23+
op.drop_column("workflow_run_archive_bundles", "deletion_status")

‎api/models/workflow.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1516,6 +1516,8 @@ class WorkflowRunArchiveBundle(DefaultFieldsDCMixin, TypeBase):
15161516
row_count: Mapped[int] = mapped_column(sa.BigInteger, nullable=False)
15171517
archive_bytes: Mapped[int] = mapped_column(sa.BigInteger, nullable=False)
15181518
archived_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
1519+
# A non-null value is a durable deletion request, including failed requests that may be retried.
1520+
deletion_status: Mapped[str | None] = mapped_column(String(16), nullable=True, default=None)
15191521

15201522

15211523
class ConversationVariable(TypeBase):

‎api/openapi/markdown/console-openapi.md‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9693,6 +9693,22 @@ Redirect to a prepared workflow-run archive ZIP file
96939693
| 302 | Redirect to pre-signed archive storage URL | **application/json**: [RedirectResponse](#redirectresponse)<br> |
96949694
| 409 | Download task is not ready | |
96959695

9696+
### [DELETE] /workflow-run-archives/{year}/{month}
9697+
Permanently delete this workspace's monthly archive files and archive index
9698+
9699+
#### Parameters
9700+
9701+
| Name | Located in | Description | Required | Schema |
9702+
| ---- | ---------- | ----------- | -------- | ------ |
9703+
| month | path | | Yes | integer |
9704+
| year | path | | Yes | integer |
9705+
9706+
#### Responses
9707+
9708+
| Code | Description | Schema |
9709+
| ---- | ----------- | ------ |
9710+
| 202 | Archive deletion accepted | **application/json**: [WorkflowRunArchiveDeletionTaskResponse](#workflowrunarchivedeletiontaskresponse)<br> |
9711+
96969712
### [GET] /workflow/{workflow_run_id}/events
96979713
**Get workflow execution events stream after resume**
96989714

@@ -24703,6 +24719,21 @@ tenant's default model. The underlying generator never raises — an empty
2470324719
| result | string | | Yes |
2470424720
| updated_at | integer | | Yes |
2470524721

24722+
#### WorkflowRunArchiveDeletionStatus
24723+
24724+
| Name | Type | Description | Required |
24725+
| ---- | ---- | ----------- | -------- |
24726+
| WorkflowRunArchiveDeletionStatus | string | | |
24727+
24728+
#### WorkflowRunArchiveDeletionTaskResponse
24729+
24730+
| Name | Type | Description | Required |
24731+
| ---- | ---- | ----------- | -------- |
24732+
| error | string | | No |
24733+
| month | integer | | Yes |
24734+
| status | [WorkflowRunArchiveDeletionStatus](#workflowrunarchivedeletionstatus) | | Yes |
24735+
| year | integer | | Yes |
24736+
2470624737
#### WorkflowRunArchiveDownloadPayload
2470724738

2470824739
Request body for preparing one monthly workflow-run archive download.
@@ -24752,6 +24783,7 @@ Lifecycle state for an asynchronous archive download request.
2475224783
| ---- | ---- | ----------- | -------- |
2475324784
| archive_bytes | integer | | Yes |
2475424785
| bundle_count | integer | | Yes |
24786+
| deletion_status | string, <br>**Available values:** "failed", "pending", "processing" | | No |
2475524787
| download_task | [WorkflowRunArchiveDownloadTaskResponse](#workflowrunarchivedownloadtaskresponse) | | No |
2475624788
| latest_archived_at | dateTime | | Yes |
2475724789
| month | integer | | Yes |

0 commit comments

Comments
 (0)