diff --git a/backend/app/api/integrations.py b/backend/app/api/integrations.py index c969227..7708dc9 100644 --- a/backend/app/api/integrations.py +++ b/backend/app/api/integrations.py @@ -33,7 +33,7 @@ from ..errors import AppError, ErrorCode from ..models_integrations import BotChannel, BotMessage from ..schemas import ok from ..services import wechat_ilink -from ..services.wechat_bot import manager +from ..services.wechat_bot import lease_held, manager from ..stores import get_store admin_router = APIRouter(prefix="/workspaces", tags=["外部接入"]) @@ -59,15 +59,15 @@ def render_qr_svg(payload: str) -> str: return f"data:image/svg+xml;base64,{encoded}" -def _channel_out(channel: BotChannel) -> dict[str, Any]: +async def _channel_out(channel: BotChannel) -> dict[str, Any]: return { "id": str(channel.id), "kind": channel.kind, "name": channel.name, "status": channel.status, "connected": bool(channel.bot_token_enc) and channel.status == "active", - # 只有抢到租约、真在拉消息的进程才算"接收中";备用进程的任务在等键,不算。 - "polling": manager.polling(channel.id), + # 数据库级租约:谁在真收消息所有进程都认,页面不会跟着进程抖动。 + "polling": await lease_held(channel.id), "ilink_user_id": channel.ilink_user_id, "last_seen_at": channel.last_seen_at, "last_error": channel.last_error, @@ -91,7 +91,7 @@ async def list_integrations(ws_id: uuid.UUID, ctx: AdminDep, session: SessionDep .where(BotChannel.workspace_id == ws_id) .order_by(BotChannel.created_at.desc()) ) - return ok({"items": [_channel_out(c) for c in rows.scalars().all()]}) + return ok({"items": [await _channel_out(c) for c in rows.scalars().all()]}) @admin_router.post("/{ws_id}/integrations/wechat/qr") @@ -168,7 +168,7 @@ async def poll_wechat_login(ws_id: uuid.UUID, qrcode_id: str, ctx: AdminDep, ses await get_store().delete(_qr_key(ws_id)) await manager.start(channel.id) - return ok({"status": "confirmed", "channel": _channel_out(channel)}) + return ok({"status": "confirmed", "channel": await _channel_out(channel)}) @admin_router.patch("/{ws_id}/integrations/{channel_id}") @@ -193,7 +193,7 @@ async def update_integration( await manager.start(channel.id) else: await manager.stop(channel.id) - return ok({"channel": _channel_out(channel)}) + return ok({"channel": await _channel_out(channel)}) @admin_router.delete("/{ws_id}/integrations/{channel_id}") @@ -211,7 +211,7 @@ async def disconnect_integration( channel.status = "pending" channel.last_error = None await session.commit() - return ok({"channel": _channel_out(channel)}) + return ok({"channel": await _channel_out(channel)}) @admin_router.get("/{ws_id}/integrations/{channel_id}/messages") diff --git a/backend/app/services/wechat_bot.py b/backend/app/services/wechat_bot.py index 1e3b7dd..3425efb 100644 --- a/backend/app/services/wechat_bot.py +++ b/backend/app/services/wechat_bot.py @@ -54,6 +54,27 @@ def site_base() -> str: return "" +async def lease_held(channel_id: uuid.UUID) -> bool: + """这个通道现在有没有进程在收消息。 + + 不能看 `manager.polling()`:生产是 gunicorn 多进程,每个进程只知道自己那把锁, + 请求落到哪个进程就回哪个答案,页面会在"接收中"和"已停止"之间闪。 + advisory lock 是数据库级的,`pg_locks` 里看得见——谁抢到了,所有进程都认。 + """ + async with SessionFactory() as session: + result = await session.execute( + sa_text( + "SELECT 1 FROM pg_locks" + " WHERE locktype = 'advisory' AND granted AND objsubid = 2" + " AND classid::bigint = (hashtext(:ns)::bigint & 4294967295)" + " AND objid::bigint = (hashtext(:cid)::bigint & 4294967295)" + " LIMIT 1" + ), + {"ns": LEASE_NAMESPACE, "cid": str(channel_id)}, + ) + return result.first() is not None + + async def active_channel_ids() -> list[uuid.UUID]: """列出"已扫码登录、该收消息"的通道。 diff --git a/backend/tests/test_wechat_bot.py b/backend/tests/test_wechat_bot.py index 9016c62..9885d78 100644 --- a/backend/tests/test_wechat_bot.py +++ b/backend/tests/test_wechat_bot.py @@ -377,3 +377,17 @@ async def test_startup_sees_logged_in_channels(client: AsyncClient) -> None: ids = await wechat_bot.active_channel_ids() assert channel_id in ids, "启动时看不见已登录通道,重启后机器人就静默失联了" + + +async def test_polling_flag_is_visible_to_every_process() -> None: + """页面的"接收中"要跨进程一致:看数据库里的锁,而不是本进程的任务表。""" + channel_id = uuid.uuid4() + assert await wechat_bot.lease_held(channel_id) is False + + lease = await wechat_bot.acquire_lease(channel_id) + assert lease is not None + try: + assert await wechat_bot.lease_held(channel_id) is True + finally: + await lease.close() + assert await wechat_bot.lease_held(channel_id) is False