diff --git a/HANDOVER.md b/HANDOVER.md index 63a5dbf..bb56051 100644 --- a/HANDOVER.md +++ b/HANDOVER.md @@ -6,6 +6,78 @@ --- +## 0. 当前交接状态（2026-08-20） + +### 0.1 代码与服务 + +- 基线代码：`ae3b6d6`（`develop`）。本次改动尚未提交，修改范围见下方「0.4」。 +- Docker Compose 项目名：`model-platform-develop14`。 +- 主入口：；浏览器通过 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 不应跟踪。 \ No newline at end of file +`data/` 是 `STORAGE_BACKEND=local` 的对象存储本地根，git ignore 不应跟踪。 diff --git a/backend/src/backend/schedule_schemas.py b/backend/src/backend/schedule_schemas.py index 1d0d705..0f9f170 100644 --- a/backend/src/backend/schedule_schemas.py +++ b/backend/src/backend/schedule_schemas.py @@ -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( diff --git a/backend/src/backend/schedules.py b/backend/src/backend/schedules.py index d2cc110..bfe6cc7 100644 --- a/backend/src/backend/schedules.py +++ b/backend/src/backend/schedules.py @@ -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( diff --git a/frontend/app/context/AuthContext.tsx b/frontend/app/context/AuthContext.tsx index 2c9efa0..3b01b75 100644 --- a/frontend/app/context/AuthContext.tsx +++ b/frontend/app/context/AuthContext.tsx @@ -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) => diff --git a/frontend/app/features/schedules/SchedulePage.tsx b/frontend/app/features/schedules/SchedulePage.tsx index 1392b76..84b5a02 100644 --- a/frontend/app/features/schedules/SchedulePage.tsx +++ b/frontend/app/features/schedules/SchedulePage.tsx @@ -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 => { 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): void => { diff --git a/frontend/app/features/schedules/state/schedulesStore.ts b/frontend/app/features/schedules/state/schedulesStore.ts index 8cc2b65..e2a2f50 100644 --- a/frontend/app/features/schedules/state/schedulesStore.ts +++ b/frontend/app/features/schedules/state/schedulesStore.ts @@ -368,6 +368,30 @@ export const useSchedulesStore = create((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((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((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 }); } }, diff --git a/frontend/app/services/api.ts b/frontend/app/services/api.ts index 29eb47c..1f12ee2 100644 --- a/frontend/app/services/api.ts +++ b/frontend/app/services/api.ts @@ -1314,10 +1314,14 @@ export async function deleteScheduleNode( scheduleId: string, nodeId: string, workflowVersion: number, + options: { delete_execution_history?: boolean } = {}, ): Promise { return apiRequest( `/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; createScheduleEdge: ( scheduleId: string,