fix(integrations): 「接收中」跨进程一致,不再跟着 gunicorn 进程抖动
CI / backend (push) Canceled after 0s

上一版用 `manager.polling()` 判"有没有人在收消息",那是进程内状态:请求落到持有者
回答 true、落到备用进程回答 false,页面上就会每两秒闪一次"已停止"。

改成查数据库里的 advisory lock(`pg_locks`,objsubid=2 的两段式锁),谁抢到租约所有
进程都看得见。`lease_held()` 自己开一条短会话查,不复用请求会话——接口里常在
`await session.commit()` 之后才拼返回值,那时原事务已经关了。
This commit is contained in:
Steven
2026-11-05 23:39:33 +08:00
parent 2170a0664d
commit f6e7616756
3 changed files with 43 additions and 8 deletions
+8 -8
View File
@@ -33,7 +33,7 @@ from ..errors import AppError, ErrorCode
from ..models_integrations import BotChannel, BotMessage from ..models_integrations import BotChannel, BotMessage
from ..schemas import ok from ..schemas import ok
from ..services import wechat_ilink from ..services import wechat_ilink
from ..services.wechat_bot import manager from ..services.wechat_bot import lease_held, manager
from ..stores import get_store from ..stores import get_store
admin_router = APIRouter(prefix="/workspaces", tags=["外部接入"]) 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}" 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 { return {
"id": str(channel.id), "id": str(channel.id),
"kind": channel.kind, "kind": channel.kind,
"name": channel.name, "name": channel.name,
"status": channel.status, "status": channel.status,
"connected": bool(channel.bot_token_enc) and channel.status == "active", "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, "ilink_user_id": channel.ilink_user_id,
"last_seen_at": channel.last_seen_at, "last_seen_at": channel.last_seen_at,
"last_error": channel.last_error, "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) .where(BotChannel.workspace_id == ws_id)
.order_by(BotChannel.created_at.desc()) .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") @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 get_store().delete(_qr_key(ws_id))
await manager.start(channel.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}") @admin_router.patch("/{ws_id}/integrations/{channel_id}")
@@ -193,7 +193,7 @@ async def update_integration(
await manager.start(channel.id) await manager.start(channel.id)
else: else:
await manager.stop(channel.id) 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}") @admin_router.delete("/{ws_id}/integrations/{channel_id}")
@@ -211,7 +211,7 @@ async def disconnect_integration(
channel.status = "pending" channel.status = "pending"
channel.last_error = None channel.last_error = None
await session.commit() 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") @admin_router.get("/{ws_id}/integrations/{channel_id}/messages")
+21
View File
@@ -54,6 +54,27 @@ def site_base() -> str:
return "" 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]: async def active_channel_ids() -> list[uuid.UUID]:
"""列出"已扫码登录、该收消息"的通道。 """列出"已扫码登录、该收消息"的通道。
+14
View File
@@ -377,3 +377,17 @@ async def test_startup_sees_logged_in_channels(client: AsyncClient) -> None:
ids = await wechat_bot.active_channel_ids() ids = await wechat_bot.active_channel_ids()
assert channel_id in 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