feat(tasks): retry failed project tasks
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been cancelled
Deploy / Deploy Staging (push) Has been cancelled
Deploy / Build Production Runtime Images (push) Failing after 2022h11m24s
Deploy / Production Browser E2E (push) Failing after 2022h1m28s
Deploy / Deploy Production (push) Failing after 2022h2m14s

This commit is contained in:
Xiaoxia AI
2026-06-24 09:19:23 +08:00
parent 9ce7098007
commit d31e30c494
3 changed files with 78 additions and 4 deletions
+59
View File
@@ -9,8 +9,10 @@ from app.dependencies import (
get_workspace_member_repository,
)
from app.schemas.task_center import ListProjectTasksResponse, ProjectTaskResponse
from app.core.celery_app import celery_app
from fastapi import APIRouter, Depends, HTTPException
from packages.application import CreateGenerationTaskCommand, CreateGenerationTaskUseCase, SubmitIngestJobCommand, SubmitIngestJobUseCase
from packages.ports.workspace_member_repository import WorkspaceMemberRepository
router = APIRouter()
@@ -103,3 +105,60 @@ def list_project_tasks(
))
items.sort(key=lambda item: item.updated_at or item.created_at or "", reverse=True)
return ListProjectTasksResponse(items=items)
@router.post("/tasks/{task_type}/{source_id}/retry", response_model=ProjectTaskResponse)
def retry_project_task(
task_type: str,
source_id: str,
authenticated_user: AuthenticatedUser = Depends(get_current_user),
ingest_job_repository: Any = Depends(get_ingest_job_repository),
generation_task_repository: Any = Depends(get_generation_task_repository),
workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
) -> ProjectTaskResponse:
if task_type == "generation":
task = generation_task_repository.get(source_id)
if task is None:
raise HTTPException(status_code=404, detail="Generation task not found")
require_workspace_member(task.workspace_id, authenticated_user, workspace_member_repository)
if task.status.value != "failed":
raise HTTPException(status_code=409, detail="Only failed tasks can be retried")
use_case = CreateGenerationTaskUseCase(generation_task_repository)
retried = use_case.execute(CreateGenerationTaskCommand(
workspace_id=task.workspace_id,
project_id=task.project_id,
asset_library_id=task.asset_library_id,
strategy_id=task.strategy_id,
voice_library_id=task.voice_library_id,
created_by_user_id=authenticated_user.user.id,
))
celery_app.send_task("worker.generate_video", args=[retried.id])
return ProjectTaskResponse(
id=f"generation:{retried.id}", task_type="generation", workspace_id=retried.workspace_id,
project_id=retried.project_id, status=retried.status.value, progress=retried.progress,
current_step=_generation_step(retried), source_id=retried.id, created_at=retried.created_at,
updated_at=retried.created_at,
)
if task_type == "ingest":
job = ingest_job_repository.get(source_id)
if job is None:
raise HTTPException(status_code=404, detail="Ingest job not found")
require_workspace_member(job.workspace_id, authenticated_user, workspace_member_repository)
if job.status.value != "failed":
raise HTTPException(status_code=409, detail="Only failed tasks can be retried")
use_case = SubmitIngestJobUseCase(ingest_job_repository)
retried = use_case.execute(SubmitIngestJobCommand(
workspace_id=job.workspace_id,
project_id=job.project_id,
library_id=job.library_id,
storage_key=job.storage_key,
))
celery_app.send_task("worker.ingest_asset", args=[retried.id])
return ProjectTaskResponse(
id=f"ingest:{retried.id}", task_type="ingest", workspace_id=retried.workspace_id,
project_id=retried.project_id, status=retried.status.value, progress=0,
current_step=_ingest_step(retried), source_id=retried.id, created_at=retried.created_at,
updated_at=retried.updated_at,
)
raise HTTPException(status_code=400, detail="Unsupported task type")
+5
View File
@@ -20,3 +20,8 @@ export const getProjectTasks = async (projectId: string): Promise<ProjectTaskIte
const response = await apiClient.get(`/projects/${projectId}/tasks`);
return response.data.items;
};
export const retryProjectTask = async (taskType: string, sourceId: string): Promise<ProjectTaskItem> => {
const response = await apiClient.post(`/tasks/${taskType}/${sourceId}/retry`);
return response.data;
};
+14 -4
View File
@@ -1,8 +1,8 @@
import React from 'react';
import { Alert, Button, Card, Empty, List, Progress, Space, Tag, Typography } from 'antd';
import { useQuery } from '@tanstack/react-query';
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
import { useParams } from 'react-router-dom';
import { getProjectTasks } from '@/api/tasks';
import { getProjectTasks, retryProjectTask } from '@/api/tasks';
const statusMap: Record<string, { label: string; color: string }> = {
pending: { label: '排队中', color: 'default' },
@@ -21,6 +21,7 @@ const taskTypeLabels: Record<string, string> = {
const ProjectTasks: React.FC = () => {
const { id } = useParams<{ id: string }>();
const projectId = id || '';
const queryClient = useQueryClient();
const tasksQuery = useQuery({
queryKey: ['project-tasks', projectId],
@@ -29,6 +30,11 @@ const ProjectTasks: React.FC = () => {
refetchInterval: 5000,
});
const retryMutation = useMutation({
mutationFn: ({ taskType, sourceId }: { taskType: string; sourceId: string }) => retryProjectTask(taskType, sourceId),
onSuccess: () => queryClient.invalidateQueries({ queryKey: ['project-tasks', projectId] }),
});
return (
<div style={{ padding: 24 }}>
<Card title="项目任务中心" extra={<Button onClick={() => tasksQuery.refetch()}>刷新</Button>}>
@@ -39,7 +45,7 @@ const ProjectTasks: React.FC = () => {
type="warning"
showIcon
message="部分任务可重试"
description="重试入口将在任务执行器安全幂等化后开放;当前先展示失败原因,避免假重试。"
description="失败任务可通过“重试”重新排队;系统会保留原始失败原因方便排查。"
/>
)}
<List
@@ -56,7 +62,11 @@ const ProjectTasks: React.FC = () => {
<Typography.Text>{taskTypeLabels[task.task_type] || task.task_type}</Typography.Text>
<Tag color={status.color}>{status.label}</Tag>
<Tag>{task.current_step}</Tag>
{task.retryable ? <Tag color="orange">可重试(暂未开放)</Tag> : null}
{task.retryable ? (
<Button size="small" loading={retryMutation.isPending} onClick={() => retryMutation.mutate({ taskType: task.task_type, sourceId: task.source_id })}>
重试
</Button>
) : null}
</Space>
}
description={