Merge branch 'feat/schedule-cleanup-handover' into develop

2026-08-20 19:27:32 +08:00
parent e4d010ec76 bc62034241
commit 5c15e68a40
7 changed files with 306 additions and 63 deletions
@@ -6,6 +6,78 @@

 ---

+## 0. 当前交接状态（2026-08-20）
+
+### 0.1 代码与服务
+
+- 基线代码：`ae3b6d6`（`develop`）。本次改动尚未提交，修改范围见下方「0.4」。
+- Docker Compose 项目名：`model-platform-develop14`。
+- 主入口：<http://127.0.0.1:9120>；浏览器通过 web/Nginx 访问后端 API。
+- 当前启用的 `.env` 已配置 MySQL 连接和存储后端。**不要把 `.env` 中的连接串、密码或
+  Token 提交到 Git。**
+
+检查服务：
+
+```powershell
+docker compose -p model-platform-develop14 ps
+Invoke-WebRequest http://127.0.0.1:9120/api/v1/health
+```
+
+完整重建并启动：
+
+```powershell
+docker compose -p model-platform-develop14 up -d --build
+```
+
+仅后端 Python 代码变动时，使用下面命令即可；它不会删除数据库或 `data/`：
+
+```powershell
+docker compose -p model-platform-develop14 build backend
+docker compose -p model-platform-develop14 up -d --force-recreate --no-deps backend
+```
+
+### 0.2 调度模块本次行为
+
+调度页面位于 `frontend/app/features/schedules/`，对应 API 位于
+`backend/src/backend/schedules.py`。
+
+- 删除调度方案：删除该方案的节点、连线、运行记录，以及可删除的运行日志/结果产物；若方案
+  仍有运行中的任务，后端返回 409，避免删到一半。
+- 删除节点：运行中的节点不可删除；有历史运行记录时，前端会提示“该节点有运行日志，是否一并删除？”。
+  用户确认后会删除节点、关联连线、该节点的 `schedule_node_runs` 记录和可删除日志/结果。
+- `StorageObjects.is_immutable=1` 的审计原件受存储层保护，不能物理删除；删除节点时会删除其
+  运行记录引用，使其不再在调度页面展示，但保留原件，避免再次出现
+  `immutable object cannot be deleted` 并导致节点删除回滚。
+- 运行记录列表已加入轮询：有运行中任务时约 1.5 秒刷新一次，空闲时约 5 秒刷新一次；右侧刷新
+  按钮仍可手动刷新。
+
+### 0.3 关键排查位置
+
+| 现象 | 首先查看 |
+|---|---|
+| 页面请求报错 | 浏览器 F12 → Network → Fetch/XHR，查看请求 URL、状态码和响应内容 |
+| API / 数据库异常 | `docker compose -p model-platform-develop14 logs --tail=200 backend` |
+| Cron 未触发 | `docker compose -p model-platform-develop14 logs --tail=200 schedule` |
+| 脚本执行失败 | 调度页面“运行记录”→“查看结果”→ 节点日志 |
+| 容器状态或端口问题 | `docker compose -p model-platform-develop14 ps` |
+
+### 0.4 未提交改动清单
+
+当前工作区包含以下代码改动，提交前应按需执行 typecheck / 后端测试：
+
+| 文件 | 作用 |
+|---|---|
+| `backend/src/backend/schedule_schemas.py` | 节点删除请求支持历史记录确认参数 |
+| `backend/src/backend/schedules.py` | 调度/节点删除、运行中保护、日志清理与不可变产物兼容 |
+| `frontend/app/services/api.ts` | 节点删除 API 参数 |
+| `frontend/app/context/AuthContext.tsx` | API 参数透传 |
+| `frontend/app/features/schedules/state/schedulesStore.ts` | 删除确认及删除后刷新状态 |
+| `frontend/app/features/schedules/SchedulePage.tsx` | 运行记录轮询 |
+
+`data/` 是本地运行数据，不应作为本次代码改动一起提交。
+
+---
+
 ## 1. 最近 8 个 commit（按时间倒序）

 ### `85b2916` — fix: local_storage_base_dir（bucket 标识符语义统一：最终 fix）
@@ -423,4 +495,4 @@ M frontend/vite.config.ts       (4+/2-)
 ```

 `frontend/vite.config.ts` 的改动不在本次 trash / bucket 修复范围内，等下次 commit 一起提。
-`data/` 是 `STORAGE_BACKEND=local` 的对象存储本地根，git ignore 不应跟踪。
+`data/` 是 `STORAGE_BACKEND=local` 的对象存储本地根，git ignore 不应跟踪。
@@ -97,6 +97,12 @@ class WorkflowVersionRequest(StrictModel):
    workflow_version: int = Field(ge=1)


+class DeleteScheduleNodeRequest(WorkflowVersionRequest):
+    """删除节点时可显式确认一并清理其已经结束的运行记录。"""
+
+    delete_execution_history: bool = False
+
+
 class CreateScheduleNodeRequest(StrictModel):
    workflow_version: int = Field(ge=1)
    node_key: str = Field(
@@ -17,13 +17,15 @@ from common.db.models import (
    ScheduleEdges,
    ScheduleNodeRuns,
    ScheduleNodes,
+    ScheduleRuns,
    Schedules,
    Scripts,
+    StorageObjects,
    Versions,
 )
 from common.ids import new_ulid
 from croniter import CroniterBadCronError, croniter
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 from sqlalchemy import delete, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession

@@ -37,14 +39,18 @@ from backend.schedule_schemas import (
    CreateScheduleNodeRequest,
    CreateScheduleRequest,
    CronPreviewRequest,
+    DeleteScheduleNodeRequest,
    UpdateScheduleEdgeRequest,
    UpdateScheduleNodeRequest,
    UpdateScheduleRequest,
    WorkflowVersionRequest,
 )
+from backend.services.storage import soft_delete_object

 router = APIRouter(tags=["schedules"])

+_ACTIVE_RUN_STATUSES = ("queued", "running")
+

 def _iso(value: datetime | None) -> str | None:
    if value is None:
@@ -60,6 +66,48 @@ def _mysql_utc(value: datetime) -> datetime:
    return value.astimezone(UTC).replace(tzinfo=None)


+def _execution_artifact_ids(
+    items: list[ScheduleRuns | ScheduleNodeRuns],
+) -> set[str]:
+    """收集运行日志和结果产物，供删除记录时一并移入回收站。"""
+
+    return {
+        storage_object_id
+        for item in items
+        for storage_object_id in (item.logs_object_id, item.result_object_id)
+        if storage_object_id
+    }
+
+
+async def _delete_execution_artifacts(
+    storage_object_ids: set[str],
+    request: Request,
+    session: AsyncSession,
+) -> None:
+    """软删除可删除的运行产物。
+
+    不可变对象是运行审计原件，存储层不允许移动或删除它们。调用方随后会
+    删除运行记录本身，因此不可变原件不会再通过该调度节点暴露；保留原件也
+    不应阻塞节点或调度方案的删除。
+    """
+
+    if not storage_object_ids:
+        return
+
+    mutable_storage_object_ids = set(
+        (
+            await session.scalars(
+                select(StorageObjects.storage_object_id).where(
+                    StorageObjects.storage_object_id.in_(storage_object_ids),
+                    StorageObjects.is_immutable == 0,
+                )
+            )
+        ).all()
+    )
+    for storage_object_id in sorted(mutable_storage_object_ids):
+        await soft_delete_object(storage_object_id, request, session)
+
+
 def _timezone(value: str) -> ZoneInfo:
    try:
        return ZoneInfo(value)
@@ -740,6 +788,7 @@ async def update_schedule(
 async def delete_schedule(
    schedule_id: str,
    payload: WorkflowVersionRequest,
+    request: Request,
    context: RequestContext = Depends(request_context),
    session: AsyncSession = Depends(database_session),
 ) -> dict[str, Any]:
@@ -750,6 +799,64 @@ async def delete_schedule(
        for_update=True,
    )
    require_workflow_version(item, payload.workflow_version)
+
+    # 正在执行的 DAG 依赖运行快照和日志对象。此时删除会让执行器无法安全收尾，
+    # 因此要求先等待运行结束，避免影响现有运行功能。
+    active_run_id = await session.scalar(
+        select(ScheduleRuns.run_id)
+        .where(
+            ScheduleRuns.schedule_id == schedule_id,
+            ScheduleRuns.run_status.in_(_ACTIVE_RUN_STATUSES),
+        )
+        .limit(1)
+    )
+    if active_run_id is not None:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            "schedule has active runs; wait for completion before deletion",
+        )
+
+    schedule_runs = list(
+        (
+            await session.scalars(
+                select(ScheduleRuns).where(ScheduleRuns.schedule_id == schedule_id)
+            )
+        ).all()
+    )
+    run_ids = [run.run_id for run in schedule_runs]
+    node_runs = (
+        list(
+            (
+                await session.scalars(
+                    select(ScheduleNodeRuns).where(
+                        ScheduleNodeRuns.run_id.in_(run_ids)
+                    )
+                )
+            ).all()
+        )
+        if run_ids
+        else []
+    )
+
+    # 调度删除会清理节点、边和所有运行记录；运行日志/结果文件同时移入回收站。
+    await _delete_execution_artifacts(
+        _execution_artifact_ids([*schedule_runs, *node_runs]),
+        request,
+        session,
+    )
+    if run_ids:
+        await session.execute(
+            delete(ScheduleNodeRuns).where(ScheduleNodeRuns.run_id.in_(run_ids))
+        )
+        await session.execute(
+            delete(ScheduleRuns).where(ScheduleRuns.run_id.in_(run_ids))
+        )
+    await session.execute(
+        delete(ScheduleEdges).where(ScheduleEdges.schedule_id == schedule_id)
+    )
+    await session.execute(
+        delete(ScheduleNodes).where(ScheduleNodes.schedule_id == schedule_id)
+    )
    item.enabled = 0
    item.next_run_at = None
    item.deleted_at = _mysql_utc(datetime.now(UTC))
@@ -884,7 +991,8 @@ async def update_schedule_node(
 async def delete_schedule_node(
    schedule_id: str,
    node_id: str,
-    payload: WorkflowVersionRequest,
+    payload: DeleteScheduleNodeRequest,
+    request: Request,
    context: RequestContext = Depends(request_context),
    session: AsyncSession = Depends(database_session),
 ) -> dict[str, Any]:
@@ -903,15 +1011,44 @@ async def delete_schedule_node(
    )
    if node is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "schedule node not found")
-    has_runs = await session.scalar(
-        select(func.count())
-        .select_from(ScheduleNodeRuns)
-        .where(ScheduleNodeRuns.node_id == node_id)
+    active_run_id = await session.scalar(
+        select(ScheduleRuns.run_id)
+        .join(ScheduleNodeRuns, ScheduleNodeRuns.run_id == ScheduleRuns.run_id)
+        .where(
+            ScheduleRuns.schedule_id == schedule_id,
+            ScheduleRuns.run_status.in_(_ACTIVE_RUN_STATUSES),
+            ScheduleNodeRuns.node_id == node_id,
+        )
+        .limit(1)
    )
-    if int(has_runs or 0):
+    if active_run_id is not None:
        raise HTTPException(
            status.HTTP_409_CONFLICT,
-            "a node with execution history cannot be deleted",
+            "node has active execution; wait for the run to finish before deletion",
+        )
+    node_runs = list(
+        (
+            await session.scalars(
+                select(ScheduleNodeRuns).where(ScheduleNodeRuns.node_id == node_id)
+            )
+        ).all()
+    )
+    if node_runs and not payload.delete_execution_history:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            detail={
+                "code": "node_execution_history_exists",
+                "message": "node has execution history; confirmation required",
+            },
+        )
+    if node_runs:
+        await _delete_execution_artifacts(
+            _execution_artifact_ids(node_runs),
+            request,
+            session,
+        )
+        await session.execute(
+            delete(ScheduleNodeRuns).where(ScheduleNodeRuns.node_id == node_id)
        )
    await session.execute(
        delete(ScheduleEdges).where(
@@ -279,8 +279,14 @@ export function useApi(): WorkspaceBoundApi {
      rawApi.createScheduleNode(workspaceId, scheduleId, input),
    updateScheduleNode: (scheduleId, nodeId, input) =>
      rawApi.updateScheduleNode(workspaceId, scheduleId, nodeId, input),
-    deleteScheduleNode: (scheduleId, nodeId, workflowVersion) =>
-      rawApi.deleteScheduleNode(workspaceId, scheduleId, nodeId, workflowVersion),
+    deleteScheduleNode: (scheduleId, nodeId, workflowVersion, options) =>
+      rawApi.deleteScheduleNode(
+        workspaceId,
+        scheduleId,
+        nodeId,
+        workflowVersion,
+        options,
+      ),
    createScheduleEdge: (scheduleId, input) =>
      rawApi.createScheduleEdge(workspaceId, scheduleId, input),
    deleteScheduleEdge: (scheduleId, edgeId, workflowVersion) =>
@@ -234,9 +234,6 @@ export default function SchedulePage({
    void useSchedulesStore.getState().loadInitial();
  }, [workspaceId, api, setSchedules, setArtifacts, setSchedule]);

-  // 进入调度页或切换调度方案后，主动加载该方案已经存在的运行记录。
-  // 原先只有「手动运行」和运行中的轮询会更新列表，因此从其他页面返回时会
-  // 一直停留在加载状态，直到产生新的运行记录。
  useEffect(() => {
    const scheduleId = schedule?.schedule_id;
    if (!scheduleId || !api) {
@@ -244,35 +241,32 @@ export default function SchedulePage({
      setRunsLoading(false);
      return;
    }
-    void useSchedulesStore.getState().refreshRuns(scheduleId, true);
-  }, [schedule?.schedule_id, api, setRuns, setRunsLoading]);
-
-  useEffect(() => {
-    const scheduleId = schedule?.schedule_id;
-    if (!scheduleId || !api) return;
    let timer: number | undefined;
    let cancelled = false;
-    const tick = (): void => {
+    const tick = async (showLoading = false): Promise<void> => {
      if (cancelled) return;
-      void useSchedulesStore.getState().refreshRuns(scheduleId);
-      // 每次 tick 重新读最新 runs,不再把 runs 放进依赖数组,避免每次 setRuns
-      // 都触发 effect cleanup + 重建 setInterval。
+      // 必须等待请求完成再判断状态；旧实现会在首个请求返回前看到空数组，
+      // 因而错误地停止轮询，导致运行记录必须手工刷新才出现。
+      await useSchedulesStore.getState().refreshRuns(scheduleId, showLoading);
+      if (cancelled || useSchedulesStore.getState().schedule?.schedule_id !== scheduleId) {
+        return;
+      }
      const runs = useSchedulesStore.getState().runs;
      const hasActiveRun = runs.some(
        (item) => item.run_status === "queued" || item.run_status === "running",
      );
-      const isEnabledCron = schedule.trigger_type === "cron" && schedule.enabled;
-      // Cron 会由后端在未来某个整分钟创建新记录。即使当前没有运行中的
-      // 记录，也要持续刷新，才能让新一轮运行自动出现在右侧列表中。
-      if (!hasActiveRun && !isEnabledCron) return;
-      timer = window.setTimeout(tick, hasActiveRun ? 1500 : 3000);
+      // 页面停留期间持续刷新：运行中更快，空闲时较慢，既能自动显示 Cron
+      // 新记录，也不会因频繁请求影响其他调度页面操作。
+      timer = window.setTimeout(() => {
+        void tick();
+      }, hasActiveRun ? 1500 : 5000);
    };
-    tick();
+    void tick(true);
    return () => {
      cancelled = true;
      if (timer !== undefined) window.clearTimeout(timer);
    };
-  }, [schedule?.schedule_id, schedule?.trigger_type, schedule?.enabled, api]);
+  }, [schedule?.schedule_id, api, setRuns, setRunsLoading]);

  // Form submit wrapper for create-schedule modal — keeps the FormEvent flow out of JSX.
  const handleCreateScheduleSubmit = (event: FormEvent<HTMLFormElement>): void => {
@@ -368,6 +368,30 @@ export const useSchedulesStore = create<State & Actions>((set, get) => {
    });
  }

+  function applyServerUpdatedSchedule(serverUpdated: Schedule): void {
+    const validNodeIds = new Set(
+      serverUpdated.nodes.map((node) => node.node_id),
+    );
+    for (const nodeId of [...positionDrafts.keys()]) {
+      if (!validNodeIds.has(nodeId)) positionDrafts.delete(nodeId);
+    }
+    const updated = applyPositionDrafts(serverUpdated);
+    set((state) => ({
+      schedule: updated,
+      schedules: (() => {
+        const summary = { ...updated, nodes: [], edges: [] };
+        const index = state.schedules.findIndex(
+          (item) => item.schedule_id === updated.schedule_id,
+        );
+        if (index < 0) return [summary, ...state.schedules];
+        return state.schedules.map((item) =>
+          item.schedule_id === updated.schedule_id ? summary : item
+        );
+      })(),
+      positionDraftCount: positionDrafts.size,
+    }));
+  }
+
  // Generic mutation wrapper: busy + try/catch + 412 + positionDraft cleanup + schedules list update
  async function withMutation(
    label: string,
@@ -379,29 +403,9 @@ export const useSchedulesStore = create<State & Actions>((set, get) => {
    set({ busy: label });
    try {
      const serverUpdated = await action();
-      const validNodeIds = new Set(
-        serverUpdated.nodes.map((node) => node.node_id),
-      );
-      for (const nodeId of [...positionDrafts.keys()]) {
-        if (!validNodeIds.has(nodeId)) positionDrafts.delete(nodeId);
-      }
-      const updated = applyPositionDrafts(serverUpdated);
-      set((state) => ({
-        schedule: updated,
-        schedules: (() => {
-          const summary = { ...updated, nodes: [], edges: [] };
-          const index = state.schedules.findIndex(
-            (item) => item.schedule_id === updated.schedule_id,
-          );
-          if (index < 0) return [summary, ...state.schedules];
-          return state.schedules.map((item) =>
-            item.schedule_id === updated.schedule_id ? summary : item
-          );
-        })(),
-        positionDraftCount: positionDrafts.size,
-      }));
+      applyServerUpdatedSchedule(serverUpdated);
      notify({ tone: "success", message: successMessage });
-      return updated;
+      return serverUpdated;
    } catch (error) {
      await handleError(error, `${successMessage}失败`);
      return null;
@@ -946,18 +950,37 @@ export const useSchedulesStore = create<State & Actions>((set, get) => {
      if (!schedule || !node || state.busy) return;
      set({ contextMenu: null });
      if (!window.confirm(`确定删除节点"${node.node_name}"吗？`)) return;
-      const updated = await withMutation(
-        "delete-node",
-        () =>
-          api.deleteScheduleNode(
+      set({ busy: "delete-node" });
+      try {
+        let updated: Schedule;
+        try {
+          // 先走普通删除：没有历史记录时不额外打扰用户。
+          updated = await api.deleteScheduleNode(
            schedule.schedule_id,
            node.node_id,
            schedule.workflow_version,
-          ),
-        "节点已删除",
-      );
-      if (updated && selectedNodeId === node.node_id) {
-        set({ selectedNodeId: null });
+          );
+        } catch (error) {
+          const requiresHistoryConfirmation =
+            error instanceof ApiRequestError
+            && error.status === 409
+            && error.code === "node_execution_history_exists";
+          if (!requiresHistoryConfirmation) throw error;
+          if (!window.confirm("该节点有运行日志，是否一并删除？")) return;
+          updated = await api.deleteScheduleNode(
+            schedule.schedule_id,
+            node.node_id,
+            schedule.workflow_version,
+            { delete_execution_history: true },
+          );
+        }
+        applyServerUpdatedSchedule(updated);
+        if (selectedNodeId === node.node_id) set({ selectedNodeId: null });
+        notify({ tone: "success", message: "节点及其运行日志已删除" });
+      } catch (error) {
+        await handleError(error, "删除节点失败");
+      } finally {
+        set({ busy: null });
      }
    },

@@ -1314,10 +1314,14 @@ export async function deleteScheduleNode(
  scheduleId: string,
  nodeId: string,
  workflowVersion: number,
+  options: { delete_execution_history?: boolean } = {},
 ): Promise<Schedule> {
  return apiRequest<Schedule>(
    `/api/v1/schedules/${scheduleId}/nodes/${nodeId}`,
-    { method: "DELETE", body: JSON.stringify({ workflow_version: workflowVersion }) },
+    {
+      method: "DELETE",
+      body: JSON.stringify({ workflow_version: workflowVersion, ...options }),
+    },
    workspaceId,
  );
 }
@@ -1586,6 +1590,7 @@ export type WorkspaceBoundApi = {
    scheduleId: string,
    nodeId: string,
    workflowVersion: number,
+    options?: { delete_execution_history?: boolean },
  ) => Promise<Schedule>;
  createScheduleEdge: (
    scheduleId: string,