diff --git a/CHANGELOG.md b/CHANGELOG.md index 97932d5aa3..71229300dd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -466,6 +466,17 @@ This release closes that milestone with **301 merged pull requests**. ### Fixed +- **projects:** Reading a shelf document for the first time no longer blocks + every other database write on SQLite while the document converts. Lazy + conversion ran pymupdf/markitdown inside the `BEGIN IMMEDIATE` transaction + that serializes the publish against trash and purge, and on SQLite that lock + is database-wide, so run status, thread metadata and scheduler writes waited + for the whole conversion and failed with `database is locked` after 30 + seconds. Conversion now writes into `.staging/` outside any transaction; the + lock is held only to revalidate the row and atomically rename the output + into place, so a document trashed or purged meanwhile still publishes + nothing. Concurrent first reads of one document share a single conversion + rather than each holding a file-IO worker. ([#6305]) - **gateway:** Per-run reads now return the rows of IM-channel owners. `start_run` stamps run rows and run events with the raw trusted owner id (for example `feishu:owner-777`), but several run-scoped routes filtered by the @@ -7734,3 +7745,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6255]: https://github.com/bytedance/deer-flow/pull/6255 [#6263]: https://github.com/bytedance/deer-flow/pull/6263 [#6282]: https://github.com/bytedance/deer-flow/pull/6282 +[#6305]: https://github.com/bytedance/deer-flow/pull/6305 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 487c16f4aa..b9d42e5bc7 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -423,6 +423,13 @@ ### 修复 +- **项目:** 在 SQLite 上首次读取书架文档时,文档转换期间不再阻塞所有其他数据库 + 写入。此前懒转换在 `BEGIN IMMEDIATE` 事务内运行 pymupdf/markitdown,以便与移入 + 回收站和彻底删除串行化发布;而 SQLite 的这把锁作用于整个数据库,运行状态、线程 + 元数据与调度器的写入都要等待整个转换完成,超过 30 秒即报 `database is locked`。 + 现在转换在任何事务之外写入 `.staging/`,只在重新校验文档行并原子重命名输出时 + 持锁,期间被移入回收站或彻底删除的文档仍不会发布任何内容。同一文档的并发首次读取 + 共享一次转换,不再各自占用文件 IO 工作线程。([#6305]) - **网关:** 单次运行的读取现在能返回 IM 渠道所有者的数据。`start_run` 用原始 受信所有者 ID(例如 `feishu:owner-777`)标记运行行与运行事件,但多个运行级路由 按内部调用方规范化后的 ID 过滤,因此在 SQL 存储上,只要所有者 ID 含有 @@ -6455,3 +6462,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#6255]: https://github.com/bytedance/deer-flow/pull/6255 [#6263]: https://github.com/bytedance/deer-flow/pull/6263 [#6282]: https://github.com/bytedance/deer-flow/pull/6282 +[#6305]: https://github.com/bytedance/deer-flow/pull/6305 diff --git a/backend/docs/ARCHITECTURE.md b/backend/docs/ARCHITECTURE.md index 1d0e68b362..2709862b29 100644 --- a/backend/docs/ARCHITECTURE.md +++ b/backend/docs/ARCHITECTURE.md @@ -102,7 +102,7 @@ The web conversation delete flow first deletes Gateway-managed thread state thro **Role authority.** Project data rides a user-role request message, never the system prompt: framework rules stay in SystemMessages; user-authored project configuration is data. -**Shelf storage (spec §6.1-6.3).** `project_documents` rows carry a server-generated content address (`stored_relpath` relative to `users/{user_id}/projects/`, embedding the sha256 and the row's own document ID) so rows never share files and re-upload after trash lands in a fresh namespace. Layout: `{sha256[:2]}/{sha256}/{document_id}/original/{name}` plus an optional `derived/converted.md` companion written via temp file and atomic rename on first read. Inserts are atomic with the active-project row lock: stage → hash → dedup among active `(project_id, sha256)` → place bytes → `INSERT` (file-before-row); a dedup hit returns the existing row (first name wins). Document delete and project delete are pure row transitions to a trash tier (`trashed_at`, `trash_origin` snapshot) — no filesystem work; trashed rows are invisible to the index, the tools, and the listing APIs. +**Shelf storage (spec §6.1-6.3).** `project_documents` rows carry a server-generated content address (`stored_relpath` relative to `users/{user_id}/projects/`, embedding the sha256 and the row's own document ID) so rows never share files and re-upload after trash lands in a fresh namespace. Layout: `{sha256[:2]}/{sha256}/{document_id}/original/{name}` plus an optional `derived/converted.md` companion created on first read: conversion writes into `.staging/` outside any database transaction, then the document-row lock is taken only to revalidate the live row and atomically rename the output into place (a row trashed or purged meanwhile discards it); concurrent first reads of one document share a single conversion. Inserts are atomic with the active-project row lock: stage → hash → dedup among active `(project_id, sha256)` → place bytes → `INSERT` (file-before-row); a dedup hit returns the existing row (first name wins). Document delete and project delete are pure row transitions to a trash tier (`trashed_at`, `trash_origin` snapshot) — no filesystem work; trashed rows are invisible to the index, the tools, and the listing APIs. **Bounded live reads (spec §7.3).** `list_project_documents` / `read_project_document` (registered only for runs carrying the pinned key) take `project_id` from the pin and read live shelf rows; binaries and conversion-disabled convertibles are declined with an error naming attach-to-thread, and documents trashed mid-run fail with a "no longer on the shelf" error rather than serving stale content. diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index cb697fc59c..aeff416cf2 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/AGENTS.md @@ -5,3 +5,5 @@ Postgres bootstrap owns its session-scoped advisory lock until `pg_advisory_unlo When `database.postgres_schema` is configured, both async ORM connections and the synchronous SQLAlchemy connections used by DB-backed custom agents and managed subagents must use the same `search_path`; preserve this invariant when adding another persistence entry point. Alembic stamp/upgrade workers started inside `bootstrap_schema()` remain owned by the bootstrap critical section until the worker finishes. Drain those `asyncio.to_thread()` calls across host cancellation before releasing the in-process SQLite bootstrap lock or PostgreSQL advisory lock; otherwise another bootstrap can overlap a still-running migration worker. + +On SQLite, `BEGIN IMMEDIATE` takes the database-wide write lock, so every unrelated writer (run status, thread metadata, the scheduler) waits for the transaction and fails with `database is locked` after `busy_timeout` (30s). Do slow work such as document conversion before opening a locked transaction; lock only to revalidate and publish, as `ProjectDocumentRepository.publish_under_live_lock` does. `tests/test_project_document_tools.py::TestConversionSerialization` pins this. diff --git a/backend/packages/harness/deerflow/persistence/projects/sql.py b/backend/packages/harness/deerflow/persistence/projects/sql.py index 212cbe638e..aeef6fdb5d 100644 --- a/backend/packages/harness/deerflow/persistence/projects/sql.py +++ b/backend/packages/harness/deerflow/persistence/projects/sql.py @@ -477,33 +477,37 @@ class ProjectDocumentRepository: await session.commit() return row, staged - async def convert_under_live_lock[T]( + async def publish_under_live_lock[T]( self, document_id: str, *, project_id: str, - convert: Callable[[dict[str, Any]], Awaitable[T]], + publish: Callable[[dict[str, Any]], Awaitable[T]], user_id: str | None | _AutoSentinel = AUTO, ) -> tuple[dict[str, Any], T] | None: - """Run *convert* while holding the owned LIVE document row's lock (§6.3). + """Run *publish* while holding the owned LIVE document row's lock (§6.3). Lazy conversion publishes ``derived/converted.md`` serialized against trash and purge: this takes the document row lock inside one transaction (``BEGIN IMMEDIATE`` on SQLite, ``SELECT … FOR UPDATE`` elsewhere), revalidates ownership, shelf membership and active state - (``trashed_at IS NULL``) AFTER locking, then invokes *convert* — the - offloaded convert-to-temp + atomic ``os.replace`` publish — and - commits. A concurrent trash/purge either commits first (the locked - re-read finds the row gone/trashed and this returns ``None`` without - invoking *convert*, so nothing is published) or blocks until the - publish commits — a derived file is never published after purge - removed the original (§6.3 lock order, §13). Only the document lock - is taken, so no lock-ordering hazard exists against trash's - project→document order. ``None`` means missing/foreign/trashed; the - caller maps it to its not-on-the-shelf / ``content_missing`` - semantics. + (``trashed_at IS NULL``) AFTER locking, then invokes *publish* — the + atomic ``os.replace`` of an already-converted file — and commits. A + concurrent trash/purge either commits first (the locked re-read finds + the row gone/trashed and this returns ``None`` without invoking + *publish*) or blocks until the publish commits — a derived file is + never published after purge removed the original (§6.3 lock order, + §13). Only the document lock is taken, so no lock-ordering hazard + exists against trash's project→document order. ``None`` means + missing/foreign/trashed; the caller maps it to its not-on-the-shelf / + ``content_missing`` semantics. + + *publish* must stay short: on SQLite ``BEGIN IMMEDIATE`` takes the + database-wide write lock, so every unrelated writer (run status, + thread metadata, the scheduler) waits for it. Convert BEFORE calling + this, never inside *publish*. """ - resolved_user_id = resolve_user_id(user_id, method_name="ProjectDocumentRepository.convert_under_live_lock") + resolved_user_id = resolve_user_id(user_id, method_name="ProjectDocumentRepository.publish_under_live_lock") async with self._sf() as session: await self._begin_immediate_if_sqlite(session) locked = (await session.execute(select(ProjectDocumentRow).where(ProjectDocumentRow.id == document_id).with_for_update())).scalar_one_or_none() @@ -511,9 +515,9 @@ class ProjectDocumentRepository: await session.rollback() return None row = self._row_to_dict(locked) - # Drained: a cancelled conversion finishes its worker (and any - # publish) before the document lock unwinds (§6.3). - result = await await_drained(convert(row)) + # Drained: a cancelled publish finishes its worker before the + # document lock unwinds (§6.3). + result = await await_drained(publish(row)) await session.commit() return row, result diff --git a/backend/packages/harness/deerflow/projects/documents.py b/backend/packages/harness/deerflow/projects/documents.py index 1df3e5a194..e3554ff514 100644 --- a/backend/packages/harness/deerflow/projects/documents.py +++ b/backend/packages/harness/deerflow/projects/documents.py @@ -13,8 +13,9 @@ Files are immutable and hash-qualified: ``stored_relpath`` (relative to ``users/{user_id}/projects/``) embeds the content hash and the row's own document ID, so rows never share bytes and a re-upload after trash lands in a fresh namespace (§6.2). Conversion is lazy: a convertible original is turned -into ``derived/converted.md`` on first read, written through a temporary file -and an atomic rename, and only when ``uploads.auto_convert_documents`` is on +into ``derived/converted.md`` on first read, converted into ``.staging/`` +outside any database transaction and then atomically renamed into place under +the document row lock, and only when ``uploads.auto_convert_documents`` is on (§6.4, §7.3). """ @@ -22,13 +23,15 @@ from __future__ import annotations import asyncio import codecs +import concurrent.futures import hashlib import logging import os import shutil +import threading import uuid from collections import OrderedDict -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Awaitable, Callable from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Any @@ -36,7 +39,7 @@ from typing import TYPE_CHECKING, Any from deerflow.config.paths import Paths from deerflow.uploads.manager import is_reserved_upload_filename from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown -from deerflow.utils.file_io import run_file_io +from deerflow.utils.file_io import await_drained, run_file_io from deerflow.utils.text_detection import is_text_file_by_content if TYPE_CHECKING: @@ -136,6 +139,11 @@ class StagedDocument: size_bytes: int +def _new_staging_path(paths: Paths, *, user_id: str, project_id: str, prefix: str = "") -> Path: + """A fresh ``.staging/{prefix}{uuid}`` path for in-flight bytes; the sweep collects leftovers after 24 hours (§8.3).""" + return paths.project_documents_dir(user_id, project_id) / ".staging" / f"{prefix}{uuid.uuid4().hex}" + + def _open_staging(staging_dir: Path, staging_path: Path): """Worker-thread: create the staging directory and open the staging file.""" staging_dir.mkdir(parents=True, exist_ok=True) @@ -160,9 +168,8 @@ async def stage_document_bytes(paths: Paths, *, user_id: str, project_id: str, c (the route maps it to 413, mirroring the uploads router) and leaves no staging file behind. """ - staging_dir = paths.project_documents_dir(user_id, project_id) / ".staging" - staging_path = staging_dir / uuid.uuid4().hex - handle = await run_file_io(_open_staging, staging_dir, staging_path) + staging_path = _new_staging_path(paths, user_id=user_id, project_id=project_id) + handle = await run_file_io(_open_staging, staging_path.parent, staging_path) digest = hashlib.sha256() size = 0 @@ -313,44 +320,96 @@ def _convert_in_thread(original: Path, output_path: Path) -> Path | None: return asyncio.run(convert_file_to_markdown(original, output_path=output_path)) -def _convert_and_publish(original: Path, derived: Path) -> bool: - """Worker-thread body: convert to a temp file, then atomically publish. +#: First-read conversions in flight, keyed by ``(user_id, document_id, +#: project_id)``. A conversion holds a bounded file-IO pool thread for its +#: whole duration, so concurrent first reads of one document share it instead +#: of each pinning a thread. The tool also runs on the isolated subagent loop, +#: so this is a lock-guarded map of ``concurrent.futures.Future`` rather than +#: a per-loop asyncio primitive. A ``None`` result means the leader was +#: cancelled or raised: its followers retry and one of them leads. +_inflight_conversions: dict[tuple[str, str, str], concurrent.futures.Future[tuple[Path | None, str | None] | None]] = {} +_inflight_conversions_lock = threading.Lock() - The temp file lives beside the final path (same filesystem) so the - rename is atomic: readers only ever see a complete ``converted.md`` or - none at all. A conversion failure publishes nothing and removes the temp. + +async def _single_flight_conversion(key: tuple[str, str, str], convert: Callable[[], Awaitable[tuple[Path | None, str | None]]]) -> tuple[Path | None, str | None]: + """Run *convert* once per *key* across concurrent callers on any event loop.""" + while True: + with _inflight_conversions_lock: + shared = _inflight_conversions.get(key) + leading = shared is None + if leading: + shared = _inflight_conversions[key] = concurrent.futures.Future() + if not leading: + # Shielded: a cancelled follower must not cancel the shared future + # the leader still has to resolve. + outcome = await asyncio.shield(asyncio.wrap_future(shared)) + if outcome is not None: + return outcome + continue + outcome = None + try: + outcome = await convert() + return outcome + finally: + # Unregister before resolving, so a follower that retries after a + # ``None`` outcome cannot find this finished future again. + with _inflight_conversions_lock: + del _inflight_conversions[key] + shared.set_result(outcome) + + +def _convert_to_staging(original: Path, staging_path: Path) -> bool: + """Worker-thread body: convert *original* into its ``.staging/`` file. + + Runs outside any database transaction — a conversion can take seconds, + and SQLite's write lock is database-wide. The output stays outside the + document namespace until published, so a purge racing the conversion + never finds, or leaves behind, a partial file there. A conversion failure + returns ``False``; the caller removes the staging file either way. """ - derived.parent.mkdir(parents=True, exist_ok=True) - temp_path = derived.parent / f".{derived.name}.{uuid.uuid4().hex}.tmp" + staging_path.parent.mkdir(parents=True, exist_ok=True) try: - produced = _convert_in_thread(original, temp_path) - if produced is None: - return False - os.replace(produced, derived) - return True + return _convert_in_thread(original, staging_path) is not None except Exception: logger.warning("Failed to convert shelf document %s", original, exc_info=True) - temp_path.unlink(missing_ok=True) return False - finally: - temp_path.unlink(missing_ok=True) + + +def _publish_converted(staging_path: Path, derived: Path) -> bool: + """Worker-thread body: atomically publish a staged conversion as ``derived/converted.md``. + + ``.staging/`` and the namespace share the projects root, so the rename is + atomic: readers only ever see a complete ``converted.md`` or none at all. + """ + try: + derived.parent.mkdir(parents=True, exist_ok=True) + os.replace(staging_path, derived) + return True + except Exception: + logger.warning("Failed to publish converted shelf document %s", derived, exc_info=True) + return False async def ensure_converted_markdown(repo: ProjectDocumentRepository, paths: Paths, *, user_id: str, row: dict, auto_convert: bool) -> tuple[Path | None, str | None]: """Return ``(converted_path, None)`` or ``(None, reason)``, converting on first read. - Conversion runs — and ``derived/converted.md`` is published — while the - repository holds the document row lock, with ownership, shelf membership - and active state revalidated AFTER locking (§6.3): a trash/purge that - committed first makes this decline as ``content_missing`` without - converting or publishing anything, and one arriving during conversion - blocks on the row lock and then proceeds. The temp-file + atomic - ``os.replace`` publish stays inside the lock, so no partial converted - text is ever exposed. The published companion is valid for the lifetime - of the row's immutable original and never needs invalidation (§6.2, - §10.9). ``reason`` is one of ``"conversion_disabled"`` (auto-convert - off, §7.3), ``"content_missing"`` (row no longer live), or ``"binary"`` - (not convertible, or the conversion itself failed). + The conversion runs into ``.staging/`` outside any database transaction; + only the publish holds the document row lock, with ownership, shelf + membership and active state revalidated AFTER locking (§6.3). A + trash/purge that committed before this read declines as + ``content_missing`` without converting; one that commits during the + conversion makes the locked revalidation decline, so the staged output is + discarded and nothing is published; one arriving during the publish + blocks on the row lock and then proceeds. The atomic ``os.replace`` + publish exposes no partial converted text. Concurrent first reads of one + document share a single conversion (``_single_flight_conversion``); one + from another process may still convert again and publish identical + bytes. The published companion is + valid for the lifetime of the row's immutable original and never needs + invalidation (§6.2, §10.9). ``reason`` is one of + ``"conversion_disabled"`` (auto-convert off, §7.3), ``"content_missing"`` + (row no longer live), or ``"binary"`` (not convertible, or the conversion + itself failed). """ if not auto_convert: return None, "conversion_disabled" @@ -359,18 +418,37 @@ async def ensure_converted_markdown(repo: ProjectDocumentRepository, paths: Path return None, "binary" if await run_file_io(derived.is_file): return derived, None + key = (user_id, row["id"], row["project_id"]) + return await _single_flight_conversion(key, lambda: _convert_and_publish(repo, paths, user_id=user_id, row=row, original=original, derived=derived)) - async def _convert(_locked_row: dict) -> bool: - # Re-check under the lock: a conversion that blocked on this row's - # lock may already have been published by the lock's previous holder. - if await run_file_io(derived.is_file): - return True - return await run_file_io(_convert_and_publish, original, derived) - locked = await repo.convert_under_live_lock(row["id"], project_id=row["project_id"], convert=_convert, user_id=user_id) +async def _convert_and_publish(repo: ProjectDocumentRepository, paths: Paths, *, user_id: str, row: dict, original: Path, derived: Path) -> tuple[Path | None, str | None]: + """Convert into ``.staging/`` outside any transaction, then publish under the row lock (§6.3).""" + # Read-only probe: skips converting a row that is already gone. It decides + # nothing — the locked revalidation before the publish does (§15.5). + live = await repo.get(row["id"], user_id=user_id) + if live is None or live["project_id"] != row["project_id"]: + return None, "content_missing" + + staging_path = _new_staging_path(paths, user_id=user_id, project_id=row["project_id"], prefix="convert-") + try: + # Drained: a cancelled read waits for the worker, so the cleanup below + # runs after its last write. + converted = await await_drained(run_file_io(_convert_to_staging, original, staging_path)) + + async def _publish(_locked_row: dict) -> bool: + if not converted: + return False + return await run_file_io(_publish_converted, staging_path, derived) + + # Revalidate even after a failed conversion: a row purged meanwhile is + # ``content_missing``, not ``binary``. + locked = await repo.publish_under_live_lock(row["id"], project_id=row["project_id"], publish=_publish, user_id=user_id) + finally: + await run_file_io(_remove_staging, staging_path) if locked is None: - # Purged/trashed (or foreign) between the caller's read and the - # locked revalidation: decline without publishing anything (§6.3). + # Purged/trashed (or foreign) before the locked revalidation: the + # staged output was discarded, nothing published (§6.3). return None, "content_missing" _locked_row, published = locked return (derived, None) if published else (None, "binary") @@ -558,7 +636,7 @@ async def stage_document_copy_for_attach( missing, foreign, trashed, or not on this project's shelf (the route maps it to 404); :class:`ShelfContentMissingError` maps to 409. """ - staging_path = paths.project_documents_dir(user_id, project_id) / ".staging" / f"attach-{uuid.uuid4().hex}" + staging_path = _new_staging_path(paths, user_id=user_id, project_id=project_id, prefix="attach-") async def _copy(row: dict) -> Path: await run_file_io(_copy_original_under_lock, paths, staging_path, user_id=user_id, row=row) diff --git a/backend/tests/blocking_io/test_project_documents.py b/backend/tests/blocking_io/test_project_documents.py index 727990a7f9..6dc8de42ca 100644 --- a/backend/tests/blocking_io/test_project_documents.py +++ b/backend/tests/blocking_io/test_project_documents.py @@ -2,8 +2,8 @@ Guards the Phase-2 Slice-B filesystem surfaces (spec §13): upload staging and the atomic rename into the document namespace, dedup-hit staging -cleanup, the conversion publish path (temp file + atomic rename of -``derived/converted.md``), and the ``read_project_document`` tool's sampled +cleanup, the conversion publish path (``.staging/`` conversion + atomic +rename into ``derived/converted.md``), and the ``read_project_document`` tool's sampled text detection and content reads. A real ``init_engine`` cannot run under the strict Blockbuster gate @@ -167,9 +167,10 @@ async def test_tool_read_dispatches_off_the_loop(tmp_path, monkeypatch) -> None: async def test_conversion_publish_path_dispatches_off_the_loop(tmp_path, monkeypatch) -> None: """First read of a convertible document validates the original - (``_content_intact``) and runs the whole convert + temp write + atomic - rename (``_convert_and_publish``, inside the locked conversion callback) - through the offload.""" + (``_content_intact``), converts into ``.staging/`` (``_convert_to_staging``, + outside any transaction), publishes with an atomic rename + (``_publish_converted``, inside the locked publish callback) and removes + the staging file (``_remove_staging``) — all through the offload.""" async def fake_convert(file_path, output_path=None): output_path.write_text("# converted markdown", encoding="utf-8") @@ -188,7 +189,9 @@ async def test_conversion_publish_path_dispatches_off_the_loop(tmp_path, monkeyp assert payload["content"] == "# converted markdown" assert "_content_intact" in calls assert "resolve_document_paths" in calls - assert "_convert_and_publish" in calls + assert "_convert_to_staging" in calls + assert "_publish_converted" in calls + assert "_remove_staging" in calls assert "_read_text_window" in calls assert "_cached_char_count" in calls finally: diff --git a/backend/tests/test_project_document_tools.py b/backend/tests/test_project_document_tools.py index 6a9aef9e73..c7139dc265 100644 --- a/backend/tests/test_project_document_tools.py +++ b/backend/tests/test_project_document_tools.py @@ -41,7 +41,7 @@ async def env(tmp_path, monkeypatch): projects = ProjectRepository(sf) docs = ProjectDocumentRepository(sf) project = await projects.create(name="P", user_id=_USER) - yield SimpleNamespace(paths=paths_mod.get_paths(), projects=projects, docs=docs, project=project) + yield SimpleNamespace(paths=paths_mod.get_paths(), sf=sf, projects=projects, docs=docs, project=project) await close_engine() @@ -508,53 +508,166 @@ class TestReadOriginalIntegrity: class TestConversionSerialization: - """§6.3/§13: conversion holds the document row lock through publication, - revalidating active state after locking — a trash/purge either commits - first (conversion declines, nothing published) or blocks until the - publish commits and then proceeds.""" + """§6.3/§13: conversion runs outside any database transaction and stages + its output under ``.staging/``; only the publish — revalidate the live row, + then atomically rename into ``derived/`` — holds the document row lock. A + trash/purge that commits before the publish makes conversion decline with + nothing published; one arriving during the publish blocks until it + commits. On SQLite that lock is database-wide, so a conversion holding it + would stall every unrelated writer.""" @staticmethod def _no_temp_left(env, row: dict) -> bool: - namespace = env.paths.project_document_path(_USER, row["stored_relpath"]) - return not namespace.exists() or list(namespace.rglob("*.tmp")) == [] + staging = env.paths.project_documents_dir(_USER, row["project_id"]) / ".staging" + return not staging.exists() or list(staging.iterdir()) == [] - async def test_conversion_first_purge_blocks_then_removes_everything(self, env, monkeypatch): + @staticmethod + def _gated_convert(started: threading.Event, finish: threading.Event): + async def fake_convert(file_path, output_path=None): + # Stall after the output exists, so every cleanup path has a file to remove. + output_path.write_text("# converted markdown", encoding="utf-8") + started.set() + assert finish.wait(10) + return output_path + + return fake_convert + + async def test_conversion_does_not_block_unrelated_database_writes(self, env, monkeypatch): + from deerflow.persistence.thread_meta.sql import ThreadMetaRepository + from deerflow.projects import documents as documents_mod + + started = threading.Event() + finish = threading.Event() + monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._gated_convert(started, finish)) + monkeypatch.setattr("deerflow.projects.tools._resolve_auto_convert", lambda: True) + row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes") + + # Driven through the agent tool, the path a first read actually takes. + read = asyncio.create_task(_read_project_document_impl(_runtime(project_id=env.project["id"]), document_id=row["id"], offset=0, limit=100)) + try: + assert await asyncio.to_thread(started.wait, 10) + # Another user's thread-metadata write lands while the conversion is still running. + await asyncio.wait_for(ThreadMetaRepository(env.sf).create("t-unrelated", user_id="u2"), 5) + finally: + finish.set() + assert json.loads(await read)["content"] == "# converted markdown" + assert converted_markdown_path(env.paths, user_id=_USER, row=row).is_file() + assert self._no_temp_left(env, row) + + async def test_restored_document_converts_from_its_new_project_into_its_original_namespace(self, env, monkeypatch): + from deerflow.projects import documents as documents_mod + + async def fake_convert(file_path, output_path=None): + output_path.write_text("# converted markdown", encoding="utf-8") + return output_path + + monkeypatch.setattr(documents_mod, "convert_file_to_markdown", fake_convert) + monkeypatch.setattr("deerflow.projects.tools._resolve_auto_convert", lambda: True) + row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes") + target = await env.projects.create(name="T", user_id=_USER) + assert await env.docs.trash(row["id"], user_id=_USER) is True + outcome, restored = await env.docs.restore(row["id"], target_project_id=target["id"], user_id=_USER) + assert outcome == "restored" + # Restore re-points the row without moving bytes: the namespace stays + # under the source project while staging happens under the target. + assert restored["project_id"] == target["id"] + assert restored["stored_relpath"].startswith(f"{env.project['id']}/") + + result = json.loads(await _read_project_document_impl(_runtime(project_id=target["id"]), document_id=row["id"], offset=0, limit=100)) + assert result["content"] == "# converted markdown" + assert converted_markdown_path(env.paths, user_id=_USER, row=restored).is_file() + assert self._no_temp_left(env, restored) + + async def test_purge_during_conversion_proceeds_and_conversion_publishes_nothing(self, env, monkeypatch): + from deerflow.projects import documents as documents_mod + + started = threading.Event() + finish = threading.Event() + monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._gated_convert(started, finish)) + row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes") + + conversion = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True)) + try: + assert await asyncio.to_thread(started.wait, 10) + assert await asyncio.wait_for(env.docs.trash(row["id"], user_id=_USER), 5) is True + assert await asyncio.wait_for(env.docs.purge(row["id"], remove_files=make_purge_file_remover(env.paths, user_id=_USER), user_id=_USER), 5) is True + finally: + finish.set() + served, reason = await conversion + # The locked revalidation finds the row gone, so the staged output is + # discarded instead of published after the purge. + assert (served, reason) == (None, "content_missing") + assert await env.docs.get(row["id"], include_trashed=True, user_id=_USER) is None + # The output never lived in the namespace, so purge removed all of it. + assert not env.paths.project_document_path(_USER, row["stored_relpath"]).exists() + assert self._no_temp_left(env, row) + + async def test_trash_during_conversion_proceeds_and_conversion_publishes_nothing(self, env, monkeypatch): + from deerflow.projects import documents as documents_mod + + started = threading.Event() + finish = threading.Event() + monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._gated_convert(started, finish)) + row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes") + derived = converted_markdown_path(env.paths, user_id=_USER, row=row) + + conversion = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True)) + try: + assert await asyncio.to_thread(started.wait, 10) + assert await asyncio.wait_for(env.docs.trash(row["id"], user_id=_USER), 5) is True + finally: + finish.set() + served, reason = await conversion + assert (served, reason) == (None, "content_missing") + assert not derived.exists() + assert self._no_temp_left(env, row) + + async def test_failed_conversion_of_a_row_purged_meanwhile_reports_content_missing(self, env, monkeypatch): from deerflow.projects import documents as documents_mod started = threading.Event() finish = threading.Event() - async def fake_convert(file_path, output_path=None): + async def failing_convert(file_path, output_path=None): started.set() assert finish.wait(10) - output_path.write_text("# converted markdown", encoding="utf-8") - return output_path + return None - monkeypatch.setattr(documents_mod, "convert_file_to_markdown", fake_convert) + monkeypatch.setattr(documents_mod, "convert_file_to_markdown", failing_convert) + row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes") + + conversion = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True)) + try: + assert await asyncio.to_thread(started.wait, 10) + assert await asyncio.wait_for(env.docs.trash(row["id"], user_id=_USER), 5) is True + assert await asyncio.wait_for(env.docs.purge(row["id"], remove_files=make_purge_file_remover(env.paths, user_id=_USER), user_id=_USER), 5) is True + finally: + finish.set() + # The failure is revalidated under the lock: the row is gone, so the + # reason is content_missing rather than binary. + assert await conversion == (None, "content_missing") + assert self._no_temp_left(env, row) + + async def test_cancelled_conversion_drains_and_leaves_no_staged_output(self, env, monkeypatch): + from deerflow.projects import documents as documents_mod + + started = threading.Event() + finish = threading.Event() + monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._gated_convert(started, finish)) row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes") - original = original_file_path(env.paths, user_id=_USER, row=row) derived = converted_markdown_path(env.paths, user_id=_USER, row=row) conversion = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True)) assert await asyncio.to_thread(started.wait, 10) - - async def trash_and_purge() -> bool: - assert await env.docs.trash(row["id"], user_id=_USER) is True - return await env.docs.purge(row["id"], remove_files=make_purge_file_remover(env.paths, user_id=_USER), user_id=_USER) - - purge_task = asyncio.create_task(trash_and_purge()) + conversion.cancel() await asyncio.sleep(0.2) - # Trash+purge is blocked on the row lock the conversion holds. - assert not purge_task.done() + # The worker cannot be interrupted: the cancellation waits for it, so + # the staged output is removed after the worker's last write. + assert not conversion.done() finish.set() - (served, reason), purged = await asyncio.gather(conversion, purge_task) - assert reason is None and served == derived - assert purged is True - # Purge then removed original, derived, and the row; no partial - # conversion temp file remains. - assert not original.exists() + with pytest.raises(asyncio.CancelledError): + await conversion assert not derived.exists() - assert await env.docs.get(row["id"], include_trashed=True, user_id=_USER) is None assert self._no_temp_left(env, row) async def test_purge_committed_first_conversion_publishes_nothing(self, env, monkeypatch): @@ -601,3 +714,102 @@ class TestConversionSerialization: assert called is False assert not derived.exists() assert self._no_temp_left(env, row) + + +class TestConversionSingleFlight: + """Concurrent first reads of one document share a single conversion. + + Conversion occupies a bounded file-IO pool thread for its whole duration, + so duplicates would starve unrelated offloaded file work. Followers wait + on the leader without holding a pool thread — from any event loop, since + the tool also runs on the isolated subagent loop — and a cancelled leader + hands the work to a follower instead of failing it.""" + + @staticmethod + def _counting_gated_convert(started: threading.Event, finish: threading.Event, calls: list[str]): + async def fake_convert(file_path, output_path=None): + calls.append(str(file_path)) + started.set() + assert finish.wait(10) + output_path.write_text("# converted markdown", encoding="utf-8") + return output_path + + return fake_convert + + async def test_concurrent_first_reads_share_one_conversion_across_event_loops(self, env, monkeypatch): + from deerflow.projects import documents as documents_mod + + started = threading.Event() + finish = threading.Event() + calls: list[str] = [] + monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._counting_gated_convert(started, finish, calls)) + row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes") + derived = converted_markdown_path(env.paths, user_id=_USER, row=row) + + def read() -> asyncio.Task: + return asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True)) + + leader = read() + try: + assert await asyncio.to_thread(started.wait, 10) + followers = [read(), read()] + # A follower on another event loop, like the isolated subagent loop. + other_loop = asyncio.create_task(asyncio.to_thread(asyncio.run, ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True))) + await asyncio.sleep(0.2) + assert len(calls) == 1 + finally: + finish.set() + results = await asyncio.gather(leader, *followers, other_loop) + assert results == [(derived, None)] * 4 + assert len(calls) == 1 + assert documents_mod._inflight_conversions == {} + + async def test_cancelled_follower_leaves_the_shared_conversion_running(self, env, monkeypatch): + from deerflow.projects import documents as documents_mod + + started = threading.Event() + finish = threading.Event() + calls: list[str] = [] + monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._counting_gated_convert(started, finish, calls)) + row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes") + derived = converted_markdown_path(env.paths, user_id=_USER, row=row) + + leader = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True)) + try: + assert await asyncio.to_thread(started.wait, 10) + follower = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True)) + await asyncio.sleep(0.2) + follower.cancel() + with pytest.raises(asyncio.CancelledError): + await follower + finally: + finish.set() + assert await leader == (derived, None) + assert len(calls) == 1 + assert documents_mod._inflight_conversions == {} + + async def test_follower_takes_over_when_the_leader_is_cancelled(self, env, monkeypatch): + from deerflow.projects import documents as documents_mod + + started = threading.Event() + finish = threading.Event() + calls: list[str] = [] + monkeypatch.setattr(documents_mod, "convert_file_to_markdown", self._counting_gated_convert(started, finish, calls)) + row = await _shelve(env, name="report.docx", data=b"\x00docx-bytes") + derived = converted_markdown_path(env.paths, user_id=_USER, row=row) + + leader = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True)) + try: + assert await asyncio.to_thread(started.wait, 10) + follower = asyncio.create_task(ensure_converted_markdown(env.docs, env.paths, user_id=_USER, row=row, auto_convert=True)) + await asyncio.sleep(0.2) + leader.cancel() + finally: + finish.set() + with pytest.raises(asyncio.CancelledError): + await leader + # The cancelled leader published nothing; the follower converted again + # instead of inheriting the cancellation. + assert await follower == (derived, None) + assert len(calls) == 2 + assert documents_mod._inflight_conversions == {} diff --git a/backend/tests/test_project_documents_repo.py b/backend/tests/test_project_documents_repo.py index 715dbd8f9c..2f4f28999c 100644 --- a/backend/tests/test_project_documents_repo.py +++ b/backend/tests/test_project_documents_repo.py @@ -611,18 +611,18 @@ class TestPurge: assert await docs.get(row["id"], user_id="u1") is not None -class TestConvertUnderLiveLock: - async def test_runs_convert_under_the_lock_and_returns_the_row(self, repos): +class TestPublishUnderLiveLock: + async def test_runs_publish_under_the_lock_and_returns_the_row(self, repos): projects, docs = repos project = await _project(projects) row = await _insert(docs, project["id"], sha="cf" * 32, doc_id="cv-ok") seen: list[str] = [] - async def convert(r: dict) -> str: + async def publish(r: dict) -> str: seen.append(r["id"]) return "published" - result = await docs.convert_under_live_lock(row["id"], project_id=project["id"], convert=convert, user_id="u1") + result = await docs.publish_under_live_lock(row["id"], project_id=project["id"], publish=publish, user_id="u1") assert result is not None locked_row, value = result assert locked_row["id"] == row["id"] @@ -637,21 +637,21 @@ class TestConvertUnderLiveLock: trashed_project, trashed = await _trashed(docs, projects, sha="d1" * 32, doc_id="cv-trashed") calls: list[str] = [] - async def convert(r: dict) -> None: + async def publish(r: dict) -> None: calls.append(r["id"]) - assert await docs.convert_under_live_lock("missing", project_id=project["id"], convert=convert, user_id="u1") is None + assert await docs.publish_under_live_lock("missing", project_id=project["id"], publish=publish, user_id="u1") is None # Foreign owner, sibling project, and trashed rows all decline without - # ever invoking the conversion callback. - assert await docs.convert_under_live_lock(row["id"], project_id=project["id"], convert=convert, user_id="u2") is None - assert await docs.convert_under_live_lock(row["id"], project_id=other["id"], convert=convert, user_id="u1") is None - assert await docs.convert_under_live_lock(trashed["id"], project_id=trashed_project["id"], convert=convert, user_id="u1") is None + # ever invoking the publish callback. + assert await docs.publish_under_live_lock(row["id"], project_id=project["id"], publish=publish, user_id="u2") is None + assert await docs.publish_under_live_lock(row["id"], project_id=other["id"], publish=publish, user_id="u1") is None + assert await docs.publish_under_live_lock(trashed["id"], project_id=trashed_project["id"], publish=publish, user_id="u1") is None assert calls == [] - async def test_concurrent_conversion_vs_purge_purge_commits_first(self, repos): + async def test_concurrent_publish_vs_purge_purge_commits_first(self, repos): """§13: purge holds its document lock across unlink and commit; a - conversion that blocked on the row lock then finds the row gone and - converts/publishes nothing.""" + publish that blocked on the row lock then finds the row gone and + publishes nothing.""" projects, docs = repos project = await _project(projects) row = await _insert(docs, project["id"], sha="e0" * 32, doc_id="cv-purge-first") @@ -664,26 +664,26 @@ class TestConvertUnderLiveLock: entered.set() await asyncio.wait_for(release.wait(), 5) - async def convert(r: dict) -> str: + async def publish(r: dict) -> str: calls.append(r["id"]) return "published" - async def run_conversion() -> tuple[dict, str] | None: + async def run_publish() -> tuple[dict, str] | None: await entered.wait() - task = asyncio.create_task(docs.convert_under_live_lock(row["id"], project_id=project["id"], convert=convert, user_id="u1")) + task = asyncio.create_task(docs.publish_under_live_lock(row["id"], project_id=project["id"], publish=publish, user_id="u1")) release.set() return await task purged, result = await asyncio.gather( docs.purge(row["id"], remove_files=remove_files, user_id="u1"), - run_conversion(), + run_publish(), ) assert purged is True assert result is None assert calls == [] - async def test_concurrent_conversion_vs_trash_conversion_commits_first(self, repos): - """§13: conversion holds the document lock through the publish; a + async def test_concurrent_publish_vs_trash_publish_commits_first(self, repos): + """§13: the publish holds the document lock until it commits; a trash arriving meanwhile blocks and then proceeds.""" projects, docs = repos project = await _project(projects) @@ -691,7 +691,7 @@ class TestConvertUnderLiveLock: entered = asyncio.Event() release = asyncio.Event() - async def convert(r: dict) -> str: + async def publish(r: dict) -> str: entered.set() await asyncio.wait_for(release.wait(), 5) return "published" @@ -703,7 +703,7 @@ class TestConvertUnderLiveLock: return await task result, trashed = await asyncio.gather( - docs.convert_under_live_lock(row["id"], project_id=project["id"], convert=convert, user_id="u1"), + docs.publish_under_live_lock(row["id"], project_id=project["id"], publish=publish, user_id="u1"), run_trash(), ) assert result is not None and result[1] == "published" @@ -765,38 +765,38 @@ class TestLockedOffloadDrain: still = await docs.get(row["id"], include_trashed=True, user_id="u1") assert still is not None and still["trashed_at"] is not None - async def test_cancelled_conversion_drains_the_worker_before_unlocking(self, repos): + async def test_cancelled_publish_drains_the_worker_before_unlocking(self, repos): import threading projects, docs = repos project = await _project(projects) - row = await _insert(docs, project["id"], sha="e7" * 32, doc_id="convert-cancel") + row = await _insert(docs, project["id"], sha="e7" * 32, doc_id="publish-cancel") entered = threading.Event() finish = threading.Event() state = {"published": False} - def slow_convert() -> None: + def slow_publish() -> None: entered.set() assert finish.wait(10) state["published"] = True - async def convert(r: dict) -> bool: - await asyncio.to_thread(slow_convert) + async def publish(r: dict) -> bool: + await asyncio.to_thread(slow_publish) return True - convert_task = asyncio.create_task(docs.convert_under_live_lock(row["id"], project_id=project["id"], convert=convert, user_id="u1")) + publish_task = asyncio.create_task(docs.publish_under_live_lock(row["id"], project_id=project["id"], publish=publish, user_id="u1")) await asyncio.to_thread(entered.wait, 10) - convert_task.cancel() - # The document lock stays held while the conversion worker drains: a + publish_task.cancel() + # The document lock stays held while the publish worker drains: a # trash blocks on it, so the publish never lands after the lock is # gone (the round-2 serialization survives cancellation). trash_task = asyncio.create_task(docs.trash(row["id"], user_id="u1")) await asyncio.sleep(0.3) - assert not convert_task.done() + assert not publish_task.done() assert not trash_task.done() finish.set() with pytest.raises(asyncio.CancelledError): - await convert_task + await publish_task assert state["published"] assert await trash_task is True