opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-08-15 04:59:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(agent): propagate resource collection failures (#40697)

Browse Source
This commit is contained in:
盐粒 Yanli 2026-08-13 06:16:12 +00:00 committed by GitHub
parent 6e80f42f38
commit 8df220c2b3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 186 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
api/services/agent/home_snapshot_service.py
View File

@ -2,8 +2,6 @@
from __future__ import annotations
import logging
from dify_agent.client import Client, DifyAgentNotFoundError
from dify_agent.protocol import CreateHomeSnapshotFromBindingRequest
from sqlalchemy import select
@ -26,8 +24,6 @@ from models.agent import (
from services.agent.errors import AgentBuildSandboxNotFoundError
from services.agent.workspace_service import AgentWorkspaceService, WorkspaceOwnerScope
logger = logging.getLogger(__name__)
class AgentHomeSnapshotUnavailableError(RuntimeError):
"""The requested owner-scoped Home Snapshot cannot be used."""
@ -122,16 +118,6 @@ class AgentHomeSnapshotService:
@classmethod
def collect_retired_home_snapshot(cls, *, tenant_id: str, home_snapshot_id: str) -> None:
try:
cls._collect_retired_home_snapshot(tenant_id=tenant_id, home_snapshot_id=home_snapshot_id)
except Exception:
logger.exception(
"Failed to collect retired Agent Home Snapshot",
extra={"tenant_id": tenant_id, "home_snapshot_id": home_snapshot_id},
)
@classmethod
def _collect_retired_home_snapshot(cls, *, tenant_id: str, home_snapshot_id: str) -> None:
with session_factory.create_session() as session:
snapshot = session.scalar(
select(AgentHomeSnapshot).where(
@ -150,14 +136,7 @@ class AgentHomeSnapshotService:
if referenced is not None:
return
snapshot_ref = snapshot.snapshot_ref
try:
cls.delete(snapshot_ref=snapshot_ref)
except Exception:
logger.exception(
"Failed to collect retired Agent Home Snapshot",
extra={"tenant_id": tenant_id, "home_snapshot_id": home_snapshot_id},
)
return
cls.delete(snapshot_ref=snapshot_ref)
with session_factory.create_session() as session:
snapshot = session.scalar(
select(AgentHomeSnapshot).where(

56
api/services/agent/workspace_service.py
View File

@ -298,16 +298,6 @@ class AgentWorkspaceService:
@classmethod
def collect_retired_binding(cls, *, tenant_id: str, binding_id: str) -> None:
try:
cls._collect_retired_binding(tenant_id=tenant_id, binding_id=binding_id)
except Exception:
logger.exception(
"Failed to collect retired Agent Workspace Binding",
extra={"tenant_id": tenant_id, "binding_id": binding_id},
)
@classmethod
def _collect_retired_binding(cls, *, tenant_id: str, binding_id: str) -> None:
with session_factory.create_session() as session:
binding = session.scalar(
select(AgentWorkspaceBinding).where(
@ -332,20 +322,13 @@ class AgentWorkspaceService:
if workspace_id is not None:
cls.collect_retired_workspace(tenant_id=tenant_id, workspace_id=workspace_id)
return
try:
with cls._client() as client:
client.destroy_execution_binding_sync(
DestroyExecutionBindingRequest(
binding_ref=backend_binding_ref,
destroy_workspace=False,
)
with cls._client() as client:
client.destroy_execution_binding_sync(
DestroyExecutionBindingRequest(
binding_ref=backend_binding_ref,
destroy_workspace=False,
)
except Exception:
logger.exception(
"Failed to collect retired Agent Workspace Binding",
extra={"tenant_id": tenant_id, "binding_id": binding_id},
)
return
with session_factory.create_session() as session:
binding = session.scalar(
select(AgentWorkspaceBinding).where(
@ -360,16 +343,6 @@ class AgentWorkspaceService:
@classmethod
def collect_retired_workspace(cls, *, tenant_id: str, workspace_id: str) -> None:
try:
cls._collect_retired_workspace(tenant_id=tenant_id, workspace_id=workspace_id)
except Exception:
logger.exception(
"Failed to collect retired Agent Workspace",
extra={"tenant_id": tenant_id, "workspace_id": workspace_id},
)
@classmethod
def _collect_retired_workspace(cls, *, tenant_id: str, workspace_id: str) -> None:
with session_factory.create_session() as session:
workspace = session.scalar(
select(AgentWorkspace).where(
@ -400,21 +373,14 @@ class AgentWorkspaceService:
workspace_ref = workspace.backend_workspace_ref
binding_ref = anchor.backend_binding_ref
anchor_id = anchor.id
try:
with cls._client() as client:
client.destroy_execution_binding_sync(
DestroyExecutionBindingRequest(
binding_ref=binding_ref,
workspace_ref=workspace_ref,
destroy_workspace=True,
)
with cls._client() as client:
client.destroy_execution_binding_sync(
DestroyExecutionBindingRequest(
binding_ref=binding_ref,
workspace_ref=workspace_ref,
destroy_workspace=True,
)
except Exception:
logger.exception(
"Failed to collect retired Agent Workspace",
extra={"tenant_id": tenant_id, "workspace_id": workspace_id, "binding_id": anchor_id},
)
return
with session_factory.create_session() as session:
stored_workspace = session.scalar(
select(AgentWorkspace).where(

1
api/tasks/collect_agent_resources_task.py
View File

@ -45,6 +45,7 @@ def collect_agent_resources(
"resource_id": resource_id,
},
)
raise
def enqueue_agent_resource_collection(

74
api/tests/unit_tests/services/agent/test_home_snapshot_service.py
View File

@ -129,27 +129,23 @@ def test_build_apply_fails_fast_without_source_binding() -> None:
)
def test_home_snapshot_collection_database_failure_is_best_effort(monkeypatch: pytest.MonkeyPatch) -> None:
def test_home_snapshot_collection_database_failure_propagates(monkeypatch: pytest.MonkeyPatch) -> None:
context = MagicMock()
session = context.__enter__.return_value
session.scalar.side_effect = RuntimeError("database unavailable")
log_exception = MagicMock()
error = RuntimeError("database unavailable")
session.scalar.side_effect = error
monkeypatch.setattr(
"services.agent.home_snapshot_service.session_factory.create_session",
lambda: context,
)
monkeypatch.setattr("services.agent.home_snapshot_service.logger.exception", log_exception)
AgentHomeSnapshotService.collect_retired_home_snapshot(
tenant_id="tenant-1",
home_snapshot_id="home-1",
)
with pytest.raises(RuntimeError) as exc_info:
AgentHomeSnapshotService.collect_retired_home_snapshot(
tenant_id="tenant-1",
home_snapshot_id="home-1",
)
session.scalar.assert_called_once()
log_exception.assert_called_once_with(
"Failed to collect retired Agent Home Snapshot",
extra={"tenant_id": "tenant-1", "home_snapshot_id": "home-1"},
)
assert exc_info.value is error
@pytest.mark.parametrize(
@ -157,7 +153,7 @@ def test_home_snapshot_collection_database_failure_is_best_effort(monkeypatch: p
[(AgentHomeSnapshot, AgentConfigDraft, AgentConfigSnapshot)],
indirect=True,
)
def test_home_snapshot_collection_final_delete_failure_is_best_effort(
def test_home_snapshot_collection_backend_failure_propagates_and_preserves_retired_snapshot(
monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
) -> None:
snapshot = AgentHomeSnapshot(
@ -169,7 +165,45 @@ def test_home_snapshot_collection_final_delete_failure_is_best_effort(
)
sqlite_session.add(snapshot)
sqlite_session.commit()
commit = MagicMock(side_effect=RuntimeError("database unavailable"))
error = RuntimeError("Agent backend unavailable")
delete = MagicMock(side_effect=error)
monkeypatch.setattr(
"services.agent.home_snapshot_service.session_factory.create_session",
lambda: nullcontext(sqlite_session),
)
monkeypatch.setattr(AgentHomeSnapshotService, "delete", delete)
with pytest.raises(RuntimeError) as exc_info:
AgentHomeSnapshotService.collect_retired_home_snapshot(
tenant_id="tenant-1",
home_snapshot_id=snapshot.id,
)
assert exc_info.value is error
stored_snapshot = sqlite_session.get(AgentHomeSnapshot, snapshot.id)
assert stored_snapshot is not None
assert stored_snapshot.status is AgentWorkingResourceStatus.RETIRED
@pytest.mark.parametrize(
"sqlite_session",
[(AgentHomeSnapshot, AgentConfigDraft, AgentConfigSnapshot)],
indirect=True,
)
def test_home_snapshot_collection_final_delete_failure_propagates(
monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
) -> None:
snapshot = AgentHomeSnapshot(
id="home-1",
tenant_id="tenant-1",
agent_id="agent-1",
snapshot_ref="snapshot-ref-1",
status=AgentWorkingResourceStatus.RETIRED,
)
sqlite_session.add(snapshot)
sqlite_session.commit()
error = RuntimeError("database unavailable")
commit = MagicMock(side_effect=error)
delete = MagicMock()
monkeypatch.setattr(
"services.agent.home_snapshot_service.session_factory.create_session",
@ -178,9 +212,11 @@ def test_home_snapshot_collection_final_delete_failure_is_best_effort(
monkeypatch.setattr(sqlite_session, "commit", commit)
monkeypatch.setattr(AgentHomeSnapshotService, "delete", delete)
AgentHomeSnapshotService.collect_retired_home_snapshot(
tenant_id="tenant-1",
home_snapshot_id="home-1",
)
with pytest.raises(RuntimeError) as exc_info:
AgentHomeSnapshotService.collect_retired_home_snapshot(
tenant_id="tenant-1",
home_snapshot_id="home-1",
)
assert exc_info.value is error
delete.assert_called_once_with(snapshot_ref="snapshot-ref-1")

84
api/tests/unit_tests/services/agent/test_workspace_service.py
View File

@ -431,45 +431,91 @@ def test_collect_workspace_destroys_workspace_then_remaining_bindings(
assert sqlite_session.get(AgentWorkspaceBinding, remaining.id) is None
def test_binding_collection_database_failure_is_best_effort(monkeypatch: pytest.MonkeyPatch) -> None:
def test_binding_collection_database_failure_propagates(monkeypatch: pytest.MonkeyPatch) -> None:
context = MagicMock()
session = context.__enter__.return_value
session.scalar.side_effect = RuntimeError("database unavailable")
log_exception = MagicMock()
error = RuntimeError("database unavailable")
session.scalar.side_effect = error
monkeypatch.setattr("services.agent.workspace_service.session_factory.create_session", lambda: context)
monkeypatch.setattr("services.agent.workspace_service.logger.exception", log_exception)
AgentWorkspaceService.collect_retired_binding(tenant_id="tenant-1", binding_id="binding-1")
with pytest.raises(RuntimeError) as exc_info:
AgentWorkspaceService.collect_retired_binding(tenant_id="tenant-1", binding_id="binding-1")
session.scalar.assert_called_once()
log_exception.assert_called_once_with(
"Failed to collect retired Agent Workspace Binding",
extra={"tenant_id": "tenant-1", "binding_id": "binding-1"},
)
assert exc_info.value is error
@pytest.mark.parametrize("sqlite_session", [(AgentWorkspace, AgentWorkspaceBinding)], indirect=True)
def test_workspace_collection_final_delete_failure_is_best_effort(
def test_binding_collection_backend_failure_propagates_and_preserves_retired_binding(
monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
) -> None:
workspace = _workspace()
binding = _binding(status=AgentWorkingResourceStatus.RETIRED)
sqlite_session.add_all([workspace, binding])
sqlite_session.commit()
error = RuntimeError("Agent backend unavailable")
client = MagicMock()
client.destroy_execution_binding_sync.side_effect = error
monkeypatch.setattr(
"services.agent.workspace_service.session_factory.create_session", lambda: nullcontext(sqlite_session)
)
monkeypatch.setattr(AgentWorkspaceService, "_client", lambda: nullcontext(client))
with pytest.raises(RuntimeError) as exc_info:
AgentWorkspaceService.collect_retired_binding(tenant_id="tenant-1", binding_id=binding.id)
assert exc_info.value is error
stored_binding = sqlite_session.get(AgentWorkspaceBinding, binding.id)
assert stored_binding is not None
assert stored_binding.status is AgentWorkingResourceStatus.RETIRED
@pytest.mark.parametrize("sqlite_session", [(AgentWorkspace, AgentWorkspaceBinding)], indirect=True)
def test_workspace_collection_backend_failure_propagates_and_preserves_retired_resources(
monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
) -> None:
workspace = _workspace(status=AgentWorkingResourceStatus.RETIRED)
anchor = _binding(status=AgentWorkingResourceStatus.RETIRED)
sqlite_session.add_all([workspace, anchor])
sqlite_session.commit()
commit = MagicMock(side_effect=RuntimeError("database unavailable"))
error = RuntimeError("Agent backend unavailable")
client = MagicMock()
client.destroy_execution_binding_sync.side_effect = error
monkeypatch.setattr(
"services.agent.workspace_service.session_factory.create_session", lambda: nullcontext(sqlite_session)
)
monkeypatch.setattr(AgentWorkspaceService, "_client", lambda: nullcontext(client))
with pytest.raises(RuntimeError) as exc_info:
AgentWorkspaceService.collect_retired_workspace(tenant_id="tenant-1", workspace_id=workspace.id)
assert exc_info.value is error
stored_workspace = sqlite_session.get(AgentWorkspace, workspace.id)
stored_anchor = sqlite_session.get(AgentWorkspaceBinding, anchor.id)
assert stored_workspace is not None
assert stored_workspace.status is AgentWorkingResourceStatus.RETIRED
assert stored_anchor is not None
assert stored_anchor.status is AgentWorkingResourceStatus.RETIRED
@pytest.mark.parametrize("sqlite_session", [(AgentWorkspace, AgentWorkspaceBinding)], indirect=True)
def test_workspace_collection_final_delete_failure_propagates(
monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
) -> None:
workspace = _workspace(status=AgentWorkingResourceStatus.RETIRED)
anchor = _binding(status=AgentWorkingResourceStatus.RETIRED)
sqlite_session.add_all([workspace, anchor])
sqlite_session.commit()
error = RuntimeError("database unavailable")
commit = MagicMock(side_effect=error)
client = MagicMock()
log_exception = MagicMock()
monkeypatch.setattr(
"services.agent.workspace_service.session_factory.create_session", lambda: nullcontext(sqlite_session)
)
monkeypatch.setattr(sqlite_session, "commit", commit)
monkeypatch.setattr(AgentWorkspaceService, "_client", lambda: nullcontext(client))
monkeypatch.setattr("services.agent.workspace_service.logger.exception", log_exception)
AgentWorkspaceService.collect_retired_workspace(tenant_id="tenant-1", workspace_id="workspace-1")
with pytest.raises(RuntimeError) as exc_info:
AgentWorkspaceService.collect_retired_workspace(tenant_id="tenant-1", workspace_id="workspace-1")
assert exc_info.value is error
client.destroy_execution_binding_sync.assert_called_once()
log_exception.assert_called_once_with(
"Failed to collect retired Agent Workspace",
extra={"tenant_id": "tenant-1", "workspace_id": "workspace-1"},
)

55
api/tests/unit_tests/tasks/test_collect_agent_resources_task.py
View File

@ -39,14 +39,13 @@ def test_enqueue_deduplicates_ids_and_skips_empty_input(monkeypatch: pytest.Monk
)
def test_collection_continues_in_workspace_binding_snapshot_order(monkeypatch: pytest.MonkeyPatch) -> None:
def test_collection_runs_in_workspace_binding_snapshot_order(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[str] = []
def collect_workspace(**_kwargs: object) -> None:
calls.append("workspace")
raise RuntimeError("workspace failed")
monkeypatch.setattr(AgentWorkspaceService, "collect_retired_workspace", collect_workspace)
monkeypatch.setattr(
AgentWorkspaceService,
"collect_retired_workspace",
lambda **_kwargs: calls.append("workspace"),
)
monkeypatch.setattr(
AgentWorkspaceService,
"collect_retired_binding",
@ -68,6 +67,48 @@ def test_collection_continues_in_workspace_binding_snapshot_order(monkeypatch: p
assert calls == ["workspace", "binding", "home"]
def test_collection_failure_propagates_and_stops_task(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[str] = []
error = RuntimeError("workspace failed")
log_exception = MagicMock()
def collect_workspace(**_kwargs: object) -> None:
calls.append("workspace")
raise error
monkeypatch.setattr(AgentWorkspaceService, "collect_retired_workspace", collect_workspace)
monkeypatch.setattr(
AgentWorkspaceService,
"collect_retired_binding",
lambda **_kwargs: calls.append("binding"),
)
monkeypatch.setattr(
AgentHomeSnapshotService,
"collect_retired_home_snapshot",
lambda **_kwargs: calls.append("home"),
)
monkeypatch.setattr("tasks.collect_agent_resources_task.logger.exception", log_exception)
with pytest.raises(RuntimeError) as exc_info:
collect_agent_resources.run(
tenant_id="tenant-1",
workspace_ids=["workspace-1"],
binding_ids=["binding-1"],
home_snapshot_ids=["home-1"],
)
assert exc_info.value is error
assert calls == ["workspace"]
log_exception.assert_called_once_with(
"Failed to collect retired Agent resource",
extra={
"tenant_id": "tenant-1",
"resource_type": "workspace",
"resource_id": "workspace-1",
},
)
def test_enqueue_failure_is_best_effort(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
collect_agent_resources,

8
dify-agent/docs/dify-agent/concepts/runtime-resources/index.md
View File

@ -115,9 +115,11 @@ Retirement is a database transition from `ACTIVE` to `RETIRED`. It prevents new
product use without performing network I/O inside the caller's transaction.
Product lifecycle paths commit this transition synchronously. After the
transaction commits, one Celery task asks Dify Agent to destroy the physical
resources. A successful collector deletes the corresponding ledger row; a
failed collector logs the failure and leaves the RETIRED row available for a
future retry or reconciler.
resources. A successful collector deletes the corresponding ledger row. If a
collector raises, the task logs the tenant, resource type, and resource ID,
re-raises the exception so collection stops and Celery records the task as
failed, and leaves the RETIRED row intact. No automatic retry or reconciliation
is performed.
The unified `collect_agent_resources` task is registered on normal Celery
workers and explicitly uses the existing `retention` queue. Standard workers