mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-01 16:10:37 +00:00
fix(projects): defer trash cleanup when retention config is unavailable (#7683)
* fix(projects): preserve trash when retention config is unavailable * test(projects): address trash retention review feedback --------- Co-authored-by: jswSpace <198684035+jswSpace@users.noreply.github.com>
This commit is contained in:
parent
0f2f199b3d
commit
bee97022b1
@ -3692,6 +3692,10 @@ origin project, or to a project you pick when the origin is gone or archived;
|
||||
if the target already holds an identical active file, the entries merge.
|
||||
Deleting a project moves its entire shelf to trash in the same step.
|
||||
|
||||
If the retention configuration cannot be read, automatic cleanup is deferred
|
||||
and the trash listing remains available. Cleanup resumes with the current
|
||||
configured retention window once the configuration is readable again.
|
||||
|
||||
Permanently deleting read-only documents also works on Windows. A file that is
|
||||
already missing does not block deletion; other filesystem errors keep the entry
|
||||
in Trash for retry. Permissions on symlink targets are preserved.
|
||||
|
||||
@ -1,5 +1,10 @@
|
||||
# Router contracts
|
||||
|
||||
Trash listing resolves the live retention configuration off-loop before its
|
||||
best-effort sweep. Configuration failures skip automatic cleanup and preserve
|
||||
the listing; never substitute a default policy for an unreadable one. Resolve
|
||||
the policy before consuming the reconciliation throttle window.
|
||||
|
||||
HTTP creation holds `goal_thread_lock` from the idempotency lookup through the
|
||||
initial checkpoint write. Same-id HTTP creates must return the existing row,
|
||||
including in memory mode; retain the native memory store's overwrite contract
|
||||
|
||||
@ -14,6 +14,7 @@ not a server-enforced handshake (§11).
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from typing import Literal
|
||||
@ -103,13 +104,10 @@ class PurgeResponse(BaseModel):
|
||||
|
||||
|
||||
def _projects_config() -> ProjectsConfig:
|
||||
"""Projects config, falling back to defaults when the app config is unavailable."""
|
||||
"""Read the retention policy; an unavailable config must not shorten it."""
|
||||
from deerflow.config.app_config import get_app_config
|
||||
|
||||
try:
|
||||
return get_app_config().projects
|
||||
except Exception:
|
||||
return ProjectsConfig()
|
||||
return get_app_config().projects
|
||||
|
||||
|
||||
def _not_found() -> HTTPException:
|
||||
@ -149,10 +147,11 @@ async def list_trash_documents(
|
||||
repo = get_project_document_repo(request)
|
||||
user_id = get_effective_user_id()
|
||||
try:
|
||||
config = await asyncio.to_thread(_projects_config)
|
||||
await run_trash_retention_sweep(
|
||||
repo,
|
||||
get_paths(),
|
||||
retention_days=_projects_config().trash_retention_days,
|
||||
retention_days=config.trash_retention_days,
|
||||
user_id=user_id,
|
||||
include_reconciliation=_reconciliation_due(user_id),
|
||||
)
|
||||
|
||||
72
backend/tests/blocking_io/test_trash_router_config_read.py
Normal file
72
backend/tests/blocking_io/test_trash_router_config_read.py
Normal file
@ -0,0 +1,72 @@
|
||||
"""Trash listing must resolve its destructive retention policy off the loop."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from blockbuster import BlockingError
|
||||
from fastapi import FastAPI
|
||||
|
||||
from app.gateway.auth_middleware import AuthMiddleware
|
||||
from app.gateway.routers import trash
|
||||
from deerflow.config.app_config import get_app_config, reset_app_config
|
||||
from deerflow.config.extensions_config import reset_extensions_config
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client(tmp_path, monkeypatch):
|
||||
def write_configs():
|
||||
config_path = tmp_path / "config.yaml"
|
||||
config_path.write_text("sandbox:\n use: deerflow.sandbox.local:LocalSandboxProvider\nprojects:\n trash_retention_days: 365\n", encoding="utf-8")
|
||||
extensions_path = tmp_path / "extensions_config.json"
|
||||
extensions_path.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
||||
return config_path, extensions_path
|
||||
|
||||
config_path, extensions_path = await asyncio.to_thread(write_configs)
|
||||
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path))
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
|
||||
monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
|
||||
monkeypatch.delenv("DEER_FLOW_ENV", raising=False)
|
||||
monkeypatch.delenv("ENVIRONMENT", raising=False)
|
||||
reset_app_config()
|
||||
reset_extensions_config()
|
||||
await asyncio.to_thread(get_app_config)
|
||||
|
||||
repo = AsyncMock()
|
||||
repo.list_trashed.return_value = []
|
||||
repo.count_trashed.return_value = 0
|
||||
sweep = AsyncMock()
|
||||
monkeypatch.setattr(trash, "run_trash_retention_sweep", sweep)
|
||||
monkeypatch.setattr(trash, "get_paths", lambda: object())
|
||||
monkeypatch.setattr(trash, "_LAST_RECONCILIATION", {})
|
||||
app = FastAPI()
|
||||
app.add_middleware(AuthMiddleware)
|
||||
app.state.project_document_repo = repo
|
||||
app.include_router(trash.router)
|
||||
try:
|
||||
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://gateway.test") as http_client:
|
||||
yield http_client, sweep
|
||||
finally:
|
||||
reset_app_config()
|
||||
reset_extensions_config()
|
||||
|
||||
|
||||
async def test_trash_listing_reads_configured_retention_off_the_loop(client):
|
||||
http_client, sweep = client
|
||||
|
||||
response = await http_client.get("/api/trash/documents")
|
||||
|
||||
assert response.status_code == 200
|
||||
sweep.assert_awaited_once()
|
||||
assert sweep.await_args.kwargs["retention_days"] == 365
|
||||
|
||||
|
||||
async def test_inline_config_probe_trips_the_blocking_gate(client):
|
||||
with pytest.raises(BlockingError):
|
||||
get_app_config()
|
||||
@ -81,11 +81,16 @@ def _as_user(user_id: str) -> dict[str, str]:
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolate(tmp_path, monkeypatch):
|
||||
import deerflow.config.paths as paths_mod
|
||||
from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
|
||||
|
||||
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path / "home"))
|
||||
monkeypatch.setattr(paths_mod, "_paths", None)
|
||||
set_app_config(AppConfig(sandbox={"use": "deerflow.sandbox.local:LocalSandboxProvider"}))
|
||||
yield
|
||||
anyio.run(close_engine)
|
||||
try:
|
||||
anyio.run(close_engine)
|
||||
finally:
|
||||
reset_app_config()
|
||||
|
||||
|
||||
def _create_project(client: TestClient, name: str = "P", **kwargs) -> dict:
|
||||
@ -179,6 +184,70 @@ def test_long_workspace_document_lifecycle(tmp_path, monkeypatch, filename):
|
||||
|
||||
|
||||
class TestTrashListing:
|
||||
def test_missing_config_preserves_documents_until_retention_can_be_read(self, tmp_path, monkeypatch, caplog):
|
||||
"""A transient config loss must not shorten a 365-day retention to 30 days."""
|
||||
from app.gateway.auth_disabled import AUTH_DISABLED_USER_ID
|
||||
from app.gateway.auth_middleware import AuthMiddleware
|
||||
from deerflow.config.app_config import get_app_config, reset_app_config
|
||||
from deerflow.config.extensions_config import reset_extensions_config
|
||||
|
||||
config_path = tmp_path / "config.yaml"
|
||||
config_text = "sandbox:\n use: deerflow.sandbox.local:LocalSandboxProvider\nprojects:\n trash_retention_days: 365\n"
|
||||
config_path.write_text(config_text, encoding="utf-8")
|
||||
extensions_path = tmp_path / "extensions_config.json"
|
||||
extensions_path.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
||||
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path))
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
|
||||
monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
|
||||
monkeypatch.delenv("DEER_FLOW_ENV", raising=False)
|
||||
monkeypatch.delenv("ENVIRONMENT", raising=False)
|
||||
reconciliation_time = 0.0
|
||||
monkeypatch.setattr(trash, "time", SimpleNamespace(monotonic=lambda: reconciliation_time))
|
||||
monkeypatch.setattr(trash, "_LAST_RECONCILIATION", {})
|
||||
reset_app_config()
|
||||
reset_extensions_config()
|
||||
try:
|
||||
assert get_app_config().projects.trash_retention_days == 365
|
||||
app = _build_app(tmp_path)
|
||||
# Exercise real local-mode authentication and configuration loading,
|
||||
# rather than letting a fixture bypass a request-time config failure.
|
||||
app.user_middleware.clear()
|
||||
app.add_middleware(AuthMiddleware)
|
||||
app.dependency_overrides.pop(get_config)
|
||||
with TestClient(app) as client:
|
||||
project = _create_project(client)
|
||||
doc = _upload(client, project["id"], "protected.txt", b"still recoverable")
|
||||
_trash(client, project["id"], doc["id"])
|
||||
_set_trashed_at(doc["id"], datetime.now(UTC) - timedelta(days=60))
|
||||
row = _get_row(app, doc["id"], user_id=AUTH_DISABLED_USER_ID)
|
||||
original = _original_path(app, row, user_id=AUTH_DISABLED_USER_ID)
|
||||
assert client.get("/api/trash/documents").json()["total"] == 1
|
||||
|
||||
last_reconciliation = trash._LAST_RECONCILIATION[AUTH_DISABLED_USER_ID]
|
||||
# Make reconciliation due again so a premature throttle stamp is observable.
|
||||
reconciliation_time += trash._RECONCILIATION_MIN_INTERVAL_SECONDS
|
||||
config_path.unlink()
|
||||
listing = client.get("/api/trash/documents")
|
||||
|
||||
assert listing.status_code == 200
|
||||
assert trash._LAST_RECONCILIATION[AUTH_DISABLED_USER_ID] == last_reconciliation
|
||||
assert _get_row(app, doc["id"], user_id=AUTH_DISABLED_USER_ID) is not None
|
||||
assert original.read_bytes() == b"still recoverable"
|
||||
assert listing.json()["total"] == 1
|
||||
assert "Lazy trash retention sweep failed" in caplog.text
|
||||
|
||||
config_path.write_text(config_text, encoding="utf-8")
|
||||
assert client.get("/api/trash/documents").json()["total"] == 1
|
||||
assert trash._LAST_RECONCILIATION[AUTH_DISABLED_USER_ID] == reconciliation_time
|
||||
# A successful hot reload still applies an intentionally shorter policy.
|
||||
config_path.write_text(config_text.replace("365", "30"), encoding="utf-8")
|
||||
assert client.get("/api/trash/documents").json()["total"] == 0
|
||||
assert _get_row(app, doc["id"], user_id=AUTH_DISABLED_USER_ID) is None
|
||||
assert not original.exists()
|
||||
finally:
|
||||
reset_app_config()
|
||||
reset_extensions_config()
|
||||
|
||||
def test_list_empty_then_populated_with_origin_display_fields(self, tmp_path):
|
||||
app = _build_app(tmp_path)
|
||||
client = TestClient(app)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user