From b600c6810bf4fa923fb8584c233c9adcd86e80c6 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Thu, 20 Aug 2026 12:07:52 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20P0-5=20=E2=80=94=20upload-status=20rollb?= =?UTF-8?q?ack,=20streaming=20copy=20(LOCAL=20only),=20user=20re-verify,?= =?UTF-8?q?=20honest=20lock?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit B1: `_mark_upload_failed_and_raise` now commits on a separate session - Helper takes `request + upload_id`, opens a fresh session from `request.app.state.session_factory` and commits there before raising. - Closes the named-lock connection-pool leak Codex flagged: the old "commit-on-the-same-session" implementation could return the GET_LOCK connection to the pool before the enclosing `finally: release_named_lock` ran, leaking `mp:` for up to `pool_recycle` and re-opening the same-key upload race. - Same helper now used by `create_server_object_payload`'s put-failure branch — two failure paths have identical semantics. B2: streaming copy for soft-delete + restore (`get_stream() + put()`) - LOCAL backend: zero-copy (aiofiles stream write). OOM fixed. - S3 backend: still OOMs on multi-GB objects — `put()` materializes the async iter via `b"".join(chunks)`. Multipart `put` is a follow-up; do NOT claim "OOM fixed on production" since production defaults to S3. C1: worker re-verifies `Users.status='active' AND is_deleted=0` - `_assert_user_active` called from `_execution_context` after resolving `triggered_by`; skips `SYSTEM_CRON_USER_ID`. - `USER_DISABLED` error_code goes into the `NODE_FINISHED_EVENT` outbox payload — `schedule_node_runs` has no `error_code` column, the row only carries the `message` text. Docstrings corrected to say so explicitly (previous docstring falsely promised row-level observability). F1: honest browser-local file lock - `api.ts` `acquireFileLock/heartbeatFileLock/releaseFileLock/ releaseFileLockOnUnload` are now no-ops with comments stating they never call the network. - `scriptWorkspaceStore` dropped `tickHeartbeats`; `tickCleanup` simplified to just clear cache. - `useEditSessionLifecycle` dropped its 15s heartbeat `setInterval`. - `ScriptWorkspace.tsx` renders `.local-lock-banner` info bar when `isEditing`. Two tabs may still silently last-write — banner is the only guard (acceptable disclosure-only tradeoff). Dead code: deleted the duplicate `upload_bytes_to_session` in `backend/src/backend/storage_api.py`. The `services.storage` import is now the only source of the function; `create_upload_record`'s docstring updated to point at `backend.resources`. --- backend/src/backend/services/storage.py | 72 +++++++-- backend/src/backend/storage_api.py | 115 +------------- backend/tests/test_scripts.py | 57 +++++++ backend/tests/test_storage_upload_status.py | 149 ++++++++++++++++++ .../app/features/platform/ScriptWorkspace.tsx | 12 ++ .../platform/hooks/useEditSessionLifecycle.ts | 18 +-- .../platform/state/scriptWorkspaceStore.ts | 63 +------- frontend/app/services/api.ts | 24 +-- frontend/app/styles/platform.css | 21 +++ schedule/src/schedule/worker.py | 61 ++++++- schedule/tests/test_worker.py | 51 +++++- 11 files changed, 437 insertions(+), 206 deletions(-) create mode 100644 backend/tests/test_storage_upload_status.py diff --git a/backend/src/backend/services/storage.py b/backend/src/backend/services/storage.py index 760c6e6..65ed927 100644 --- a/backend/src/backend/services/storage.py +++ b/backend/src/backend/services/storage.py @@ -151,6 +151,47 @@ async def release_named_lock(session: AsyncSession, lock_name: str) -> None: ) +async def _mark_upload_failed_and_raise( + request: Request, + upload_id: str, + http_status: int, + detail: str, +) -> None: + """Flip ``upload_status`` to a terminal state and raise, surviving the + surrounding ``session_scope`` rollback AND keeping the named-lock + connection pristine. + + Why a separate session: this helper is invoked from inside the + ``acquire_named_lock`` critical section in + :func:`upload_bytes_to_session` (size / hash / storage-put failure + branches). Committing on the lock-bound session could return its + connection to the pool, after which the enclosing + ``finally: release_named_lock`` would check out a different + connection and leak the ``mp:`` lock for up to ``pool_recycle`` + seconds — re-opening the same-key upload race the lock exists to + close. Opening a fresh session from the factory commits the status + flip independently and leaves the lock connection untouched so + ``release_named_lock`` runs on the same connection that ran + ``GET_LOCK``. + + Why this helper exists at all: the route handler wraps every request + in ``session_scope`` (see ``backend.dependencies.database_session``), + which rolls back on exception. Without this helper, a naive + ``upload.upload_status = "failed"; raise HTTPException(...)`` would + lose the status flip and leave the row stuck in ``created``/``uploading`` + forever. + """ + session_factory = request.app.state.session_factory + async with session_factory() as session: + upload = await session.scalar( + select(UploadSessions).where(UploadSessions.upload_id == upload_id) + ) + if upload is not None: + upload.upload_status = "failed" + await session.commit() + raise HTTPException(http_status, detail) + + async def _resolve_unique_object_key( session: AsyncSession, object_key: str, @@ -401,7 +442,10 @@ async def upload_bytes_to_session( f"upload cannot continue from status {upload.upload_status}", ) if upload.expires_at < _utcnow_naive(): + # 同样需要事务外 commit,否则 expired 状态会被外层 session_scope + # 回滚,客户端重试永远得到 created。 upload.upload_status = "expired" + await session.commit() raise HTTPException(status.HTTP_409_CONFLICT, "upload expired") # 按 object_key 串行化「PUT + INSERT」临界区:否则两个同 key 并发上传 @@ -434,16 +478,16 @@ async def upload_bytes_to_session( upload.expected_size_bytes is not None and actual_size != upload.expected_size_bytes ): - upload.upload_status = "failed" - raise HTTPException( + await _mark_upload_failed_and_raise( + request, upload.upload_id, status.HTTP_409_CONFLICT, "uploaded bytes size does not match expected_size_bytes", ) actual_hash = hashlib.sha256(content).hexdigest() if content else "" if upload.expected_hash and actual_hash != upload.expected_hash: - upload.upload_status = "failed" - raise HTTPException( + await _mark_upload_failed_and_raise( + request, upload.upload_id, status.HTTP_409_CONFLICT, "uploaded bytes hash does not match expected_hash", ) @@ -460,11 +504,11 @@ async def upload_bytes_to_session( metadata=s3_metadata or None, ) except Exception as exc: - upload.upload_status = "failed" - raise HTTPException( + await _mark_upload_failed_and_raise( + request, upload.upload_id, status.HTTP_503_SERVICE_UNAVAILABLE, f"failed to write object to storage: {exc}", - ) from exc + ) item = _build_storage_object( upload=upload, @@ -571,11 +615,11 @@ async def create_server_object_payload( metadata={"sha256": content_hash}, ) except Exception as exc: - upload.upload_status = "failed" - raise HTTPException( + await _mark_upload_failed_and_raise( + request, upload.upload_id, status.HTTP_503_SERVICE_UNAVAILABLE, f"failed to write object to storage: {exc}", - ) from exc + ) item = _build_storage_object( upload=upload, @@ -687,8 +731,12 @@ async def soft_delete_object( trash_bucket = actual_bucket_name("trash") try: object_stores = request.app.state.object_stores - data = await object_stores[item.bucket_name].get(item.object_key) - await object_stores[trash_bucket].put(trash_key, data) + # P0-5 / B2: 流式迁移,避免 get() 全量加载导致 10G 对象 OOM。 + # LOCAL 后端的 put 是 aiofiles 流式写入,get_stream + put 零字节驻留; + # S3 后端目前 put 仍会 ``b"".join(chunks)`` 物化到内存,S3 大对象 + # 的 OOM 修复需要把 put 改成 multipart upload —— 后续工单。 + stream = object_stores[item.bucket_name].get_stream(item.object_key) + await object_stores[trash_bucket].put(trash_key, stream) await object_stores[item.bucket_name].delete(item.object_key) except Exception as exc: raise HTTPException( diff --git a/backend/src/backend/storage_api.py b/backend/src/backend/storage_api.py index 41c6d8b..843c5c2 100644 --- a/backend/src/backend/storage_api.py +++ b/backend/src/backend/storage_api.py @@ -253,7 +253,8 @@ async def create_upload_record( # Two-step server-proxied upload: the caller PUTs the raw bytes to # ``upload_path`` after this response, which routes through - # ``upload_bytes_to_session`` below. + # ``backend.resources.upload_bytes_to_session`` (the canonical helper + # in ``services.storage``). return { "upload_id": upload.upload_id, "status": upload.upload_status, @@ -284,111 +285,6 @@ def _public_base_url(request: Request) -> str: return f"{scheme}://{host}" -async def upload_bytes_to_session( - upload_id: str, session: AsyncSession, request: Request -) -> StorageObjects: - """Server-proxied upload: read raw bytes from the request body, validate - against the ``UploadSessions`` expectations, call ``backend.put``, and - create the ``StorageObjects`` row. - - Replaces the old presign-PUT + head-validate flow. - """ - upload = await session.scalar( - select(UploadSessions) - .where(UploadSessions.upload_id == upload_id) - .with_for_update() - ) - if upload is None: - raise HTTPException(status.HTTP_404_NOT_FOUND, "upload not found") - if upload.upload_status == "completed" and upload.storage_object_id: - item = await session.get(StorageObjects, upload.storage_object_id) - if item is None or item.object_status != "available": - # Linked object was deleted; allow re-upload with the same id. - upload.storage_object_id = None - upload.upload_status = "created" - else: - return item - if upload.upload_status not in {"created", "uploading"}: - raise HTTPException( - status.HTTP_409_CONFLICT, - f"upload cannot continue from status {upload.upload_status}", - ) - if upload.expires_at < utcnow(): - upload.upload_status = "expired" - raise HTTPException(status.HTTP_409_CONFLICT, "upload expired") - - content = await request.body() - actual_size = len(content) - - if ( - upload.expected_size_bytes is not None - and actual_size != upload.expected_size_bytes - ): - upload.upload_status = "failed" - raise HTTPException( - status.HTTP_409_CONFLICT, - "uploaded bytes size does not match expected_size_bytes", - ) - - actual_hash = hashlib.sha256(content).hexdigest() if content else "" - if upload.expected_hash and actual_hash != upload.expected_hash: - upload.upload_status = "failed" - raise HTTPException( - status.HTTP_409_CONFLICT, "uploaded bytes hash does not match expected_hash" - ) - - # Round-trip content_type + sha256 metadata through the storage backend - # so the next head() (or our own put signature) can recover them. - s3_metadata: dict[str, str] = {} - if actual_hash: - s3_metadata["sha256"] = actual_hash - - try: - await request.app.state.object_stores[upload.bucket_name].put( - upload.object_key, - content, - content_type=upload.content_type, - metadata=s3_metadata or None, - ) - except Exception as exc: - upload.upload_status = "failed" - raise HTTPException( - status.HTTP_503_SERVICE_UNAVAILABLE, - f"failed to write object to storage: {exc}", - ) from exc - - file_name = safe_file_name(upload.file_name_hint or "upload.bin") - item = StorageObjects( - storage_object_id=new_ulid(), - workspace_id=upload.workspace_id, - owner_user_id=upload.user_id, - object_type="file", - usage_type=upload.usage_type, - storage_backend=settings.storage_backend, - bucket_name=upload.bucket_name, - object_key=upload.object_key, - object_key_hash=upload.object_key_hash, - storage_uri=build_storage_uri(upload.bucket_name, upload.object_key), - file_name=file_name, - file_extension=PurePosixPath(file_name).suffix.lower() or None, - mime_type=upload.content_type, - size_bytes=actual_size, - content_hash=actual_hash or None, - object_etag=None, - visibility=upload.visibility, - is_immutable=int(upload.is_immutable), - object_status="available", - created_by=upload.user_id, - ) - session.add(item) - await session.flush() - await session.refresh(item) - upload.storage_object_id = item.storage_object_id - upload.upload_status = "completed" - upload.completed_at = utcnow() - return item - - @router.post( "/v1/objects", dependencies=[Depends(require_internal_service)], @@ -438,9 +334,9 @@ async def restore_object( status.HTTP_409_CONFLICT, "object has no trash pointer; cannot restore" ) try: - # Cross-backend copy: get from trash, put back to source bucket. + # P0-5 / B2: 与 soft_delete_object 对称,从回收站恢复也走流式, + # 不再把整个对象加载进内存。S3 put 物化限制同 soft_delete_object。 object_stores = request.app.state.object_stores - data = await object_stores[actual_bucket_name("trash")].get(item.object_key) source_purpose, _, trash_tail = item.object_key.partition("/") if not source_purpose: source_purpose = "workspace" @@ -453,7 +349,8 @@ async def restore_object( else trash_tail ) target_bucket = actual_bucket_name(source_purpose) - await object_stores[target_bucket].put(source_key, data) + stream = object_stores[actual_bucket_name("trash")].get_stream(item.object_key) + await object_stores[target_bucket].put(source_key, stream) except Exception as exc: raise HTTPException( status.HTTP_502_BAD_GATEWAY, diff --git a/backend/tests/test_scripts.py b/backend/tests/test_scripts.py index cc97c79..8b6ebeb 100644 --- a/backend/tests/test_scripts.py +++ b/backend/tests/test_scripts.py @@ -498,6 +498,63 @@ async def test_soft_delete_object_sets_is_deleted() -> None: assert result["data"]["object_status"] == "deleted" +@pytest.mark.asyncio +async def test_soft_delete_object_streams_via_get_stream() -> None: + """P0-5 / B2: 必须走 ``get_stream()`` 流式迁移,不能调 ``get()`` + 把整个对象加载到内存(10G 对象会 OOM)。""" + from backend.services.storage import soft_delete_object + + item = _storage_object_row() + # 让 storage_backend 与 settings 一致,确保走"移动到 trash"分支 + item.storage_backend = settings.storage_backend + item.object_status = "available" + item.bucket_name = "workspace" + item.object_key = "ws/u/file.bin" + item.usage_type = "working_copy" + + source_store = MagicMock() + # 模拟一个 async generator 作为 get_stream 的返回值 + async def _fake_stream(_key, _chunk_size=65536): + yield b"chunk-1" + yield b"chunk-2" + source_store.get_stream.side_effect = _fake_stream + source_store.put = AsyncMock() + source_store.delete = AsyncMock() + + trash_store = MagicMock() + trash_store.put = AsyncMock() + trash_store.delete = AsyncMock() + + request = MagicMock() + request.app.state.object_stores = { + item.bucket_name: source_store, + "trash": trash_store, + } + + session = AsyncMock() + session.scalar = AsyncMock(return_value=item) + + # 捕获原 key —— helper 在移动后会把 item.object_key 重写成 trash_key, + # 不捕获的话下面断言会拿不到原值。 + original_object_key = item.object_key + + await soft_delete_object( + storage_object_id=item.storage_object_id, + request=request, + session=session, + ) + + # get_stream 必须被调用;get() 不应被调用 —— 否则仍是全量加载路径 + source_store.get_stream.assert_called_once_with(original_object_key) + source_store.get.assert_not_called() + # 源对象删除 —— 同样用原始 key 断言(item.object_key 已被 helper 改写) + source_store.delete.assert_awaited_once_with(original_object_key) + # 目标桶写入走 put(接受 async iter),trash_key 尾部拼 storage_object_id + trash_store.put.assert_awaited_once() + put_args, _ = trash_store.put.call_args + assert put_args[0] == f"workspace/{original_object_key}-{item.storage_object_id}" + + @pytest.mark.asyncio async def test_jupyter_check_notebook_lock_ignores_deleted_scripts() -> None: """``is_deleted == 0`` filter must hide deleted notebooks from Jupyter checks.""" diff --git a/backend/tests/test_storage_upload_status.py b/backend/tests/test_storage_upload_status.py new file mode 100644 index 0000000..a5a46f1 --- /dev/null +++ b/backend/tests/test_storage_upload_status.py @@ -0,0 +1,149 @@ +"""Unit tests for ``upload_bytes_to_session`` failure-path status persistence. + +P0-5 / B1: the route handler wraps every request in ``session_scope`` +(``backend.dependencies.database_session``), which rolls back on +exception. A naive ``upload.upload_status = "failed"; raise HTTPException(...)`` +loses the status flip and leaves the row stuck in ``created``/``uploading`` +forever. The fix is ``_mark_upload_failed_and_raise`` which opens a +*separate* session from ``request.app.state.session_factory`` and commits +the status change on it before raising. The outer rollback is then a +no-op on the caller's session, AND the named-lock connection (acquired +inside the helper's caller for the size/hash/storage failure branches) +stays pristine so ``release_named_lock`` runs on the same connection that +ran ``GET_LOCK`` — closing the lock-leak window Codex flagged. + +These tests exercise the helper directly + the expired-status branch of +``upload_bytes_to_session`` with a mocked session. +""" +from __future__ import annotations + +from datetime import datetime, timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi import HTTPException + +from backend.services.storage import ( + _mark_upload_failed_and_raise, + upload_bytes_to_session, +) + + +def _make_upload(upload_id: str = "01UPL0000000000000000000A") -> SimpleNamespace: + """Minimal UploadSessions row with the columns the helper touches.""" + return SimpleNamespace( + upload_id=upload_id, + upload_status="uploading", + expires_at=None, + expected_size_bytes=None, + expected_hash=None, + bucket_name="workspace", + object_key="ws/user/file.bin", + object_key_hash=b"\x00" * 32, + ) + + +class _AsyncContextManager: + """Async context manager that yields ``session`` on enter. + + Used to mock the result of ``session_factory()`` without pulling in + a real SQLAlchemy engine. + """ + + def __init__(self, session: object) -> None: + self._session = session + + async def __aenter__(self) -> object: + return self._session + + async def __aexit__(self, *_args: object) -> None: + return None + + +async def test_mark_upload_failed_and_raise_persists_status() -> None: + """The helper must open a fresh session, mark 'failed', commit, then + raise — surviving both the surrounding ``session_scope`` rollback and + the named-lock connection-pool leak the original implementation opened + up by committing on the lock-bound session. + """ + upload = _make_upload() + fresh_session = SimpleNamespace( + scalar=AsyncMock(return_value=upload), + commit=AsyncMock(), + ) + session_factory = MagicMock(return_value=_AsyncContextManager(fresh_session)) + request = SimpleNamespace( + app=SimpleNamespace(state=SimpleNamespace(session_factory=session_factory)), + ) + + with pytest.raises(HTTPException) as excinfo: + await _mark_upload_failed_and_raise( + request, upload.upload_id, 409, "size mismatch", + ) + + assert excinfo.value.status_code == 409 + assert excinfo.value.detail == "size mismatch" + assert upload.upload_status == "failed" + # The fresh session — not the caller's — must have committed. + fresh_session.scalar.assert_awaited_once() + fresh_session.commit.assert_awaited_once() + session_factory.assert_called_once() + + +async def test_mark_upload_failed_and_raise_raises_even_if_upload_missing() -> None: + """If the row has been hard-deleted between the caller's lookup and the + helper's separate-session write, the helper still raises HTTPException + with the requested status/detail — it just skips the commit. The caller + still gets the same error contract. + """ + fresh_session = SimpleNamespace( + scalar=AsyncMock(return_value=None), + commit=AsyncMock(), + ) + session_factory = MagicMock(return_value=_AsyncContextManager(fresh_session)) + request = SimpleNamespace( + app=SimpleNamespace(state=SimpleNamespace(session_factory=session_factory)), + ) + + with pytest.raises(HTTPException) as excinfo: + await _mark_upload_failed_and_raise( + request, "01UPL0000000000000000000A", 503, "boom", + ) + + assert excinfo.value.status_code == 503 + assert excinfo.value.detail == "boom" + # No row to update, so commit must NOT have been called. + fresh_session.commit.assert_not_awaited() + + +async def test_upload_bytes_to_session_expired_commits_status() -> None: + """The early ``if upload.expires_at < now`` branch must also + commit the expired status, otherwise a client retrying after the + expiry window would see ``created`` again and re-upload. + + The expired branch is OUTSIDE the named-lock critical section, so it + can still commit on the caller's session safely — no separate-session + detour is needed. + """ + expired_upload = _make_upload() + expired_upload.upload_status = "created" + # expires_at < utcnow_naive() triggers the expired branch. + expired_upload.expires_at = datetime.utcnow() - timedelta(minutes=1) + + session = SimpleNamespace( + scalar=AsyncMock(return_value=expired_upload), + commit=AsyncMock(), + get=AsyncMock(return_value=None), + ) + request = SimpleNamespace( + app=SimpleNamespace(state=SimpleNamespace(object_stores={})), + ) + + with pytest.raises(HTTPException) as excinfo: + await upload_bytes_to_session("01UPL0000000000000000000A", session, request) + + assert excinfo.value.status_code == 409 + assert "expired" in excinfo.value.detail + assert expired_upload.upload_status == "expired" + session.commit.assert_awaited() \ No newline at end of file diff --git a/frontend/app/features/platform/ScriptWorkspace.tsx b/frontend/app/features/platform/ScriptWorkspace.tsx index 7f41889..eac3b86 100644 --- a/frontend/app/features/platform/ScriptWorkspace.tsx +++ b/frontend/app/features/platform/ScriptWorkspace.tsx @@ -478,6 +478,18 @@ export function ScriptWorkspace({ )} + {/* 本地编辑锁提示——只在非 Python 编辑时显示， + 因为 Python 编辑走 pythonEditorBuffers，不走文件锁。 + 这个锁只在本浏览器当前 tab 内有效，不能阻止隐身模式 / 其它浏览器同时编辑。 */} + {isEditing && ( +
+ + + 本地编辑锁——关闭标签页、刷新页面或换浏览器后失效，不会阻止他人同时编辑。 + +
+ )} + {/* 编辑器画布区域 - 始终渲染，保证 iframe 不重新加载 */}
{/* 只读模式内容 - 用 CSS 控制显示/隐藏 */} diff --git a/frontend/app/features/platform/hooks/useEditSessionLifecycle.ts b/frontend/app/features/platform/hooks/useEditSessionLifecycle.ts index 3007335..0154e29 100644 --- a/frontend/app/features/platform/hooks/useEditSessionLifecycle.ts +++ b/frontend/app/features/platform/hooks/useEditSessionLifecycle.ts @@ -9,8 +9,9 @@ type ActivePage = "home" | "scripts" | "schedules" | "system"; /** * 必须在 layout 层挂载，不能放在 ScriptsPage。 - * 原因：心跳 / cleanup / beforeunload 需要在用户切到 /schedules 时仍运行， - * 否则其他 tab 中的编辑会话锁会过期。 + * 原因：cleanup / beforeunload 需要在用户切到 /schedules 时仍运行，否则跨页 + * 浏览 10 分钟以上再回来时，sessionCache 里塞的全是陈旧的本地锁。 + * 心跳已删除——本地锁没有过期概念，存不存在 15s 定时器都一样。 */ export function useEditSessionLifecycle({ activePage, @@ -26,17 +27,9 @@ export function useEditSessionLifecycle({ void useScriptWorkspaceStore.getState().endEditing(true, false); }, [activePage]); - // 2) 心跳 + cleanup 定时器（15s 心跳，60s 检查 cleanup） + // 2) 本地缓存回收（60s 检查一次，10 分钟无活动的本地锁清掉） useEffect(() => { - let heartbeatRunning = false; let cleanupRunning = false; - const heartbeatTimer = window.setInterval(() => { - if (heartbeatRunning) return; - heartbeatRunning = true; - void useScriptWorkspaceStore.getState().tickHeartbeats().finally(() => { - heartbeatRunning = false; - }); - }, 15 * 1000); const cleanupTimer = window.setInterval(() => { if (cleanupRunning) return; cleanupRunning = true; @@ -47,12 +40,11 @@ export function useEditSessionLifecycle({ } }, 60 * 1000); return () => { - window.clearInterval(heartbeatTimer); window.clearInterval(cleanupTimer); }; }, []); - // 3) 卸载前释放编辑锁 + // 3) 卸载前清理本地锁引用（详见 store.releaseActiveOnUnload） useEffect(() => { const handleUnload = () => { useScriptWorkspaceStore.getState().releaseActiveOnUnload(); diff --git a/frontend/app/features/platform/state/scriptWorkspaceStore.ts b/frontend/app/features/platform/state/scriptWorkspaceStore.ts index c95fef3..2236de8 100644 --- a/frontend/app/features/platform/state/scriptWorkspaceStore.ts +++ b/frontend/app/features/platform/state/scriptWorkspaceStore.ts @@ -128,7 +128,8 @@ type State = { refreshReadOnlyContent: () => void; // 刷新只读内容 // 生命周期 tick（由 layout 层 useEditSessionLifecycle 周期性调用） - tickHeartbeats: () => Promise; + // 本地锁无服务端心跳——tickHeartbeats 已删除，保留 tickCleanup 仅用于回收 + // 闲置 10 分钟以上的本地缓存。 tickCleanup: () => void; releaseActiveOnUnload: () => void; }; @@ -1077,57 +1078,9 @@ export const useScriptWorkspaceStore = create((set, get) => { } }, - tickHeartbeats: async () => { - if (!_api) return; - // 1) 当前 active session 心跳 - const active = _editSession; - if (active) { - try { - const updated = await _api.heartbeatFileLock(active); - if ( - _editSession - && _editSession.edit_session_id === updated.edit_session_id - ) { - const merged = { - ..._editSession, - session_status: updated.session_status, - expires_at: updated.expires_at, - }; - _editSession = merged; - editSessionHandle.current = merged; - set({ editSession: merged }); - } - } catch (error) { - _editSession = null; - editSessionHandle.current = null; - set({ editSession: null, embeddedJupyterUrl: null }); - pushToast( - "error", - `编辑锁心跳已中断：${ - error instanceof Error ? error.message : "请重新打开文件" - }`, - ); - } - } - // 2) 缓存会话心跳（不更新 React state，只更新缓存对象本身） - const promises: Promise[] = []; - for (const cached of sessionCache.values()) { - promises.push( - _api.heartbeatFileLock(cached.session) - .then((updated) => { - cached.session.session_status = updated.session_status; - cached.session.expires_at = updated.expires_at; - }) - .catch(() => { - // 静默失败：等用户切回来时再处理 - }), - ); - } - await Promise.allSettled(promises); - }, - tickCleanup: () => { - if (!_api) return; + // 本地锁：清缓存即可，不要发 releaseFileLock 请求。 + // 接口是异步的，但在本地实现里等价于无操作，promise 没人 await。 const TEN_MINUTES = 10 * 60 * 1000; const now = Date.now(); const toCleanup: string[] = []; @@ -1141,15 +1094,11 @@ export const useScriptWorkspaceStore = create((set, get) => { } if (toCleanup.length === 0) return; for (const scriptId of toCleanup) { - const cached = sessionCache.get(scriptId); - if (cached) { - _api.releaseFileLock(cached.session).catch(console.warn); - sessionCache.delete(scriptId); - } + sessionCache.delete(scriptId); } pushToast( "info", - `已清理 ${toCleanup.length} 个长时间未活动的编辑会话`, + `已清理 ${toCleanup.length} 个长时间未活动的本地编辑会话`, ); }, diff --git a/frontend/app/services/api.ts b/frontend/app/services/api.ts index 6e5c47e..29eb47c 100644 --- a/frontend/app/services/api.ts +++ b/frontend/app/services/api.ts @@ -678,10 +678,12 @@ export type StableVersion = { created_at: string; }; -// The current backend authorizes Jupyter through the session cookie and -// deliberately has no persisted file-lock or access-ticket endpoints. -// Keep the editor's session-shaped UI contract locally while opening the -// existing authenticated Jupyter proxy directly. +// 本地浏览器级“编辑锁”——后端没有 acquire/heartbeat/release/edit-session 表。 +// 这里的四个函数全部是占位：返回结构是为了让上层 store 的 +// _editSession / sessionCache 继续按“session”接口工作，但锁的实际作用域 +// 仅限当前 tab。关闭 tab、刷新页面、用隐身模式打开、或换浏览器，锁即失效。 +// 不要把这些函数当作鉴权或并发控制用——它们什么都不查、什么都不写。 +// 真实并发控制需要后端 edit_sessions 表 + Nginx auth_request 联动，是后续工单。 export async function acquireFileLock( workspaceId: string, @@ -698,9 +700,9 @@ export async function acquireFileLock( heartbeat_interval_seconds: 300, expires_at: new Date(now + 3600_000).toISOString(), runtime_id: workspaceId, - jupyter_session_id: "unlocked-session", + jupyter_session_id: "local", relative_path: script.relative_path, - lock_token: "unlocked-session", + lock_token: "local", script_id: script.script_id, script_name: script.script_name, jupyter_path: script.jupyter_path, @@ -711,10 +713,9 @@ export async function heartbeatFileLock( _workspaceId: string, session: ActiveEditSession, ): Promise { - return { - ...session, - expires_at: new Date(Date.now() + 3600_000).toISOString(), - }; + // 本地锁不存在过期概念；只是把 expires_at 推后让 UI 看着还活着。 + // 该字段当前没有任何消费者，保留只是为了不破坏契约。 + return session; } export async function releaseFileLock( @@ -728,7 +729,8 @@ export function releaseFileLockOnUnload( _workspaceId: string, _session: ActiveEditSession, ): void { - // No backend lock is created in compatibility mode. + // 本地锁随 tab 生命周期结束。beforeunload 调到这里只是让 store 端 + // 清理模块级引用，避免下一个 tab 复用时看到陈旧 _editSession。 } async function waitForJupyterReady(jupyterUrl: string): Promise { diff --git a/frontend/app/styles/platform.css b/frontend/app/styles/platform.css index c9a471c..0d886af 100644 --- a/frontend/app/styles/platform.css +++ b/frontend/app/styles/platform.css @@ -2090,6 +2090,27 @@ button { flex-shrink: 0; } +/* 本地编辑锁提示：放在 editor-toolbar 下方、editor-canvas 上方。 + 蓝色 info 调，与 readonly-editor-banner（黄色 hard-block）区分。 + flex-shrink: 0 让 banner 不会被 canvas 高度挤压。 */ +.local-lock-banner { + display: flex; + align-items: center; + justify-content: center; + gap: 8px; + padding: 6px 16px; + background: #e7f1ff; + border-bottom: 1px solid #c9def9; + color: #1f4e8a; + font-size: 12px; + font-weight: 500; + flex-shrink: 0; +} + +.local-lock-banner svg { + flex-shrink: 0; +} + .readonly-editor-content { flex: 1; overflow-y: auto; diff --git a/schedule/src/schedule/worker.py b/schedule/src/schedule/worker.py index 759b7dd..436d148 100644 --- a/schedule/src/schedule/worker.py +++ b/schedule/src/schedule/worker.py @@ -27,6 +27,7 @@ from common.db.models import ( ScheduleRuns, Schedules, StorageObjects, + Users, Versions, Workspaces, ) @@ -36,6 +37,7 @@ from common.eventing import ( schedule_event_type, utcnow, ) +from common.scheduler.trigger import SYSTEM_CRON_USER_ID from loguru import logger from sqlalchemy import select @@ -48,6 +50,13 @@ NODE_FINISHED_EVENT = schedule_event_type("job.node.finished") class NodeExecutor: """Owns the actual execution of one schedule node (notebook / python).""" + # P0-5 / C1: distinct error_code for runs blocked because the originating + # user was disabled or soft-deleted between queue time and worker pickup. + # The value lands in the NODE_FINISHED_EVENT outbox payload (the + # ``error_code`` field) — schedule_node_runs has no such column; the row + # only carries the message text. Operators grep the outbox stream. + USER_DISABLED_ERROR_CODE = "USER_DISABLED" + def __init__( self, *, @@ -123,18 +132,26 @@ class NodeExecutor: ) except Exception as exc: trace = traceback.format_exc() + # P0-5 / C1: 让 _assert_user_active 抛的 ValueError 透传成单独的 + # error_code,便于运维 grep 区分"用户被禁用"和"代码崩溃"。 + exc_message = str(exc) + error_code = ( + self.USER_DISABLED_ERROR_CODE + if exc_message.startswith("USER_DISABLED:") + else "WORKER_EXECUTION_FAILED" + ) result = ExecutionResult( status="failed", exit_code=1, logs=trace.encode("utf-8", errors="replace"), result=json.dumps( - {"status": "failed", "error": str(exc)}, + {"status": "failed", "error": exc_message}, ensure_ascii=False, ).encode("utf-8"), result_file_name=f"{payload['node_run_id']}-result.json", result_content_type="application/json", - error_code="WORKER_EXECUTION_FAILED", - error_message=str(exc)[:2000], + error_code=error_code, + error_message=exc_message[:2000], ) if context is None: context = await self._fallback_execution_context(payload) @@ -299,6 +316,37 @@ class NodeExecutor: return "3.12" return row[0] + async def _assert_user_active( + self, + session: AsyncSession, + user_id: str, + ) -> None: + """P0-5 / C1: re-verify the user is still ``status='active'`` and + ``is_deleted=0`` before executing a run they originated. + + Raises :class:`ValueError` whose message starts with + ``USER_DISABLED:`` when the user has been disabled or soft-deleted + between run creation and worker pickup. The outer + :meth:`handle_node_execute` parses that prefix and routes the + resulting ``error_code="USER_DISABLED"`` into the + ``NODE_FINISHED_EVENT`` outbox payload (the + ``schedule_node_runs`` row has no ``error_code`` column, only a + ``message`` text field). + """ + user = await session.scalar( + select(Users.status, Users.is_deleted).where(Users.user_id == user_id) + ) + if user is None: + raise ValueError( + f"USER_DISABLED: originating user {user_id} no longer exists" + ) + status_value, is_deleted = user + if status_value != "active" or is_deleted != 0: + raise ValueError( + f"USER_DISABLED: originating user {user_id} is " + f"status={status_value!r} is_deleted={is_deleted}" + ) + async def _execution_context( self, payload: dict[str, Any], @@ -352,6 +400,13 @@ class NodeExecutor: if not storage.bucket_name or not storage.object_key: raise ValueError("stable version artifact location is incomplete") user_id = run.triggered_by or schedule.created_by + # P0-5 / C1: re-verify the user is still active. ``create_scheduled_run`` + # checked membership when the run was queued, but the user may + # have been disabled or soft-deleted in the meantime (admin + # action, offboarding). Skip the check for the synthetic SYSTEM_CRON + # user — that row is a fixed admin baseline and never goes inactive. + if user_id != SYSTEM_CRON_USER_ID: + await self._assert_user_active(session, user_id) context = { "node_status": node_run.node_status, "workspace_id": run.workspace_id, diff --git a/schedule/tests/test_worker.py b/schedule/tests/test_worker.py index 71c6f09..1fac0d1 100644 --- a/schedule/tests/test_worker.py +++ b/schedule/tests/test_worker.py @@ -1,4 +1,5 @@ -"""Tests for NodeExecutor bucket-aware artifact download (P0-3). +"""Tests for NodeExecutor bucket-aware artifact download (P0-3) and +user-status re-verification (P0-5 / C1). Pre-fix the worker's ``object_store`` was bound to the global version bucket at startup, so a workspace whose ``Workspaces.artifact_bucket`` @@ -6,6 +7,11 @@ points at a custom S3 bucket would always 404 on download. The fix introduces a per-bucket store cache so the worker reads from whatever bucket the artifact actually lives in. +P0-5 / C1: the worker previously trusted the ``triggered_by`` user_id +without re-checking ``Users.status`` / ``is_deleted``. A user disabled +after a run was created would still have their schedules execute. +The fix adds ``_assert_user_active`` to the execution-context path. + These tests cover only the routing logic — no live S3 / MySQL. """ @@ -152,3 +158,46 @@ async def test_download_artifact_rejects_hash_mismatch() -> None: # Reads still went through the bucket-bound store, not the default. custom_store.get.assert_awaited_once() default_store.get.assert_not_called() + + +# ── P0-5 / C1: user.status re-verification ──────────────────────────────── + + +async def test_assert_user_active_passes_for_active_user() -> None: + """The active path must not raise — the run is allowed to execute.""" + executor, _, _ = _make_executor() + session = MagicMock() + session.scalar = AsyncMock(return_value=("active", 0)) + + await executor._assert_user_active(session, "01USR0000000000000000000A") + + +async def test_assert_user_active_blocks_inactive_user() -> None: + """A user whose status flipped to ``disabled`` after the run was + queued must be rejected with a USER_DISABLED error_code-shaped prefix.""" + executor, _, _ = _make_executor() + session = MagicMock() + session.scalar = AsyncMock(return_value=("disabled", 0)) + + with pytest.raises(ValueError, match="USER_DISABLED"): + await executor._assert_user_active(session, "01USR0000000000000000000A") + + +async def test_assert_user_active_blocks_soft_deleted_user() -> None: + """``is_deleted=1`` (soft delete) must also block execution.""" + executor, _, _ = _make_executor() + session = MagicMock() + session.scalar = AsyncMock(return_value=("active", 1)) + + with pytest.raises(ValueError, match="USER_DISABLED"): + await executor._assert_user_active(session, "01USR0000000000000000000A") + + +async def test_assert_user_active_blocks_missing_user() -> None: + """Hard-deleted user (row gone) — also blocked, same prefix.""" + executor, _, _ = _make_executor() + session = MagicMock() + session.scalar = AsyncMock(return_value=None) + + with pytest.raises(ValueError, match="USER_DISABLED"): + await executor._assert_user_active(session, "01USR0000000000000000000A")