From 9c2dc6e392dad8052c8efc23f30088948f4682ac Mon Sep 17 00:00:00 2001 From: NanPan <111261006+poijygfdyy@users.noreply.github.com> Date: Sat, 3 Oct 2026 08:43:10 +0800 Subject: [PATCH] fix(channels): keep runtime config commit owned across cancellation (#6198) * fix(channels): own runtime config commit across cancellation * test(channels): pin cancelled runtime config commit * docs(channels): record runtime config commit ownership * style(tests): fix channel connection router lint --------- Co-authored-by: Willem Jiang --- backend/app/channels/AGENTS.md | 1 + .../gateway/routers/channel_connections.py | 50 +++++++----- .../tests/test_channel_connections_router.py | 77 +++++++++++++++++++ 3 files changed, 110 insertions(+), 18 deletions(-) diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index e09e0afb84..d358dede6d 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -63,6 +63,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_ - WeCom outbound content is capped at the protocol's 20480 UTF-8 bytes: stream replies clip on a character boundary with a truncation marker (one stream carries the whole reply and cannot split mid-way), while proactive pushes split into at most 10 sequential markdown messages per push, with the remaining tail clipped and marker-terminated. Both paths measure bytes, not characters. A per-chat send lock serializes each split batch end to end, because manager workers run concurrently and two long pushes to the same chat would otherwise interleave chunks; locks are reference-counted and reclaimed once no sender holds or waits on them, so the registry does not grow for the life of the Gateway. - Frontend APIs: `GET /api/channels/providers`, `GET /api/channels/connections`, `POST /api/channels/{provider}/connect`, and `DELETE /api/channels/connections/{connection_id}`. - Browser APIs remain protected by normal Gateway auth/CSRF. Provider messages arrive through the already-configured channel workers. +- Runtime credential configure/disconnect treats the runtime-config file write and `app.state.channels_config` reconciliation as one cancellation-owned commit; do not let caller cancellation escape between them. - Provider-level `connection_status` reflects the user's newest connection row. With no binding it is `not_connected`, except in auth-disabled local mode where a configured running channel reports `connected` because all channel messages already route to the default user. - Slack replies use the configured operator bot token from `channels.slack` unless per-connection credentials are present; unreadable or corrupt stored credentials are treated as unavailable. - Telegram, Slack, Discord, Feishu/Lark, DingTalk, WeChat, and WeCom workers resolve incoming platform identities to connection records before reaching `ChannelManager`. diff --git a/backend/app/gateway/routers/channel_connections.py b/backend/app/gateway/routers/channel_connections.py index ef25aa4545..ed137b6ab7 100644 --- a/backend/app/gateway/routers/channel_connections.py +++ b/backend/app/gateway/routers/channel_connections.py @@ -19,9 +19,11 @@ from app.channels.runtime_config_store import ( ) from app.channels.wechat_qr_login import QRLoginError, WechatQRLogin from app.gateway.deps import require_admin_user +from app.gateway.persistent_writes import run_drained_write from deerflow.config.channel_connections_config import ChannelConnectionsConfig from deerflow.persistence.channel_connections import ChannelConnectionRepository from deerflow.persistence.engine import get_session_factory +from deerflow.utils.file_io import await_drained router = APIRouter(prefix="/api/channels", tags=["channel-connections"]) logger = logging.getLogger(__name__) @@ -623,15 +625,7 @@ async def _disconnect_channel_provider_runtime(provider: str, request: Request) if repo is not None: await repo.disconnect_provider_connections(provider=provider) - store = await _get_runtime_config_store(request) - await asyncio.to_thread(store.set_provider_disconnected, provider) - - # Re-read the live cached config and drop only this provider so a concurrent - # mutation for a different provider is not clobbered. No await may occur - # between this read and the reassignment. - live_channels_config = await _get_channels_config(request) - live_channels_config.pop(provider, None) - request.app.state.channels_config = live_channels_config + live_channels_config = await _commit_runtime_channel_config(request, provider, None) return _provider_response(config, live_channels_config, provider, _PROVIDER_META[provider]) @@ -712,21 +706,41 @@ async def _configure_channel_provider_runtime(provider: str, body: ChannelRuntim return await _apply_runtime_channel_config(request, config, provider, runtime_config) +async def _commit_runtime_channel_config( + request: Request, + provider: str, + runtime_config: dict[str, Any] | None, +) -> dict[str, Any]: + """Persist one runtime config change and reconcile the live cache atomically to cancellation.""" + + async def _commit() -> dict[str, Any]: + store = await _get_runtime_config_store(request) + if runtime_config is None: + await run_drained_write("Disconnect channel runtime config", store.set_provider_disconnected, (), provider) + else: + await run_drained_write("Save channel runtime config", store.set_provider_config, (), provider, runtime_config) + + # Re-read the live cached config after the durable write so concurrent + # mutations for sibling providers are preserved. The outer drain keeps + # cancellation owned until this in-memory reconciliation also lands. + live_channels_config = await _get_channels_config(request) + if runtime_config is None: + live_channels_config.pop(provider, None) + else: + live_channels_config[provider] = runtime_config + request.app.state.channels_config = live_channels_config + return live_channels_config + + return await await_drained(_commit()) + + async def _apply_runtime_channel_config(request: Request, config: ChannelConnectionsConfig, provider: str, runtime_config: dict[str, Any]) -> ChannelProviderResponse: started = await _restart_runtime_channel_if_available(provider, runtime_config) if started is False: display_name = _PROVIDER_META[provider]["display_name"] raise HTTPException(status_code=400, detail=f"Failed to start {display_name} channel. Check the values and try again.") - store = await _get_runtime_config_store(request) - await asyncio.to_thread(store.set_provider_config, provider, runtime_config) - - # Re-read the live cached config and apply only this provider's change so a - # concurrent mutation for a different provider is not clobbered. No await - # may occur between this read and the reassignment. - live_channels_config = await _get_channels_config(request) - live_channels_config[provider] = runtime_config - request.app.state.channels_config = live_channels_config + live_channels_config = await _commit_runtime_channel_config(request, provider, runtime_config) return _provider_response(config, live_channels_config, provider, _PROVIDER_META[provider]) diff --git a/backend/tests/test_channel_connections_router.py b/backend/tests/test_channel_connections_router.py index e48f008083..29a6e95920 100644 --- a/backend/tests/test_channel_connections_router.py +++ b/backend/tests/test_channel_connections_router.py @@ -2,6 +2,8 @@ from __future__ import annotations +import asyncio +import threading from tempfile import TemporaryDirectory from types import SimpleNamespace from unittest.mock import AsyncMock @@ -878,6 +880,81 @@ def test_configure_provider_runtime_credentials_preserves_masked_secrets(tmp_pat anyio.run(repo.close) +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("runtime_config", "expected_persisted"), + [ + ({"enabled": True, "bot_token": "new-token"}, {"enabled": True, "bot_token": "new-token"}), + (None, {"enabled": False, "_runtime_disabled": True}), + ], +) +async def test_runtime_config_commit_drains_persistence_and_cache_across_cancellation( + monkeypatch, + runtime_config, + expected_persisted, +): + started = threading.Event() + release = threading.Event() + persisted: dict[str, dict] = {} + live_channels_config = { + "telegram": {"enabled": True, "bot_token": "sibling"}, + "slack": {"enabled": True, "bot_token": "old-token"}, + } + + def set_provider_config(provider, value): + started.set() + assert release.wait(timeout=5) + persisted[provider] = dict(value) + + def set_provider_disconnected(provider): + started.set() + assert release.wait(timeout=5) + persisted[provider] = {"enabled": False, "_runtime_disabled": True} + + store = SimpleNamespace( + set_provider_config=set_provider_config, + set_provider_disconnected=set_provider_disconnected, + ) + + async def get_store(_request): + return store + + async def get_channels_config(_request): + return live_channels_config + + monkeypatch.setattr(channel_connections, "_get_runtime_config_store", get_store) + monkeypatch.setattr(channel_connections, "_get_channels_config", get_channels_config) + request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(channels_config=live_channels_config))) + + task = asyncio.create_task(channel_connections._commit_runtime_channel_config(request, "slack", runtime_config)) + try: + assert await asyncio.to_thread(started.wait, 5) + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + + release.set() + with pytest.raises(asyncio.CancelledError): + await task + finally: + release.set() + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + assert persisted["slack"] == expected_persisted + assert request.app.state.channels_config["telegram"] == { + "enabled": True, + "bot_token": "sibling", + } + if runtime_config is None: + assert "slack" not in request.app.state.channels_config + else: + assert request.app.state.channels_config["slack"] == runtime_config + + def test_disconnect_provider_runtime_config_clears_connected_state(tmp_path): import anyio