From 2170a0664de90a7d1e4ee5f543f3e16361f8a0bb Mon Sep 17 00:00:00 2001 From: Steven Date: Thu, 5 Nov 2026 23:33:44 +0800 Subject: [PATCH] =?UTF-8?q?fix(integrations):=20=E9=87=8D=E5=90=AF?= =?UTF-8?q?=E5=90=8E=E6=9C=BA=E5=99=A8=E4=BA=BA=E8=A6=81=E8=83=BD=E8=87=AA?= =?UTF-8?q?=E5=B7=B1=E6=8E=A5=E5=9B=9E=E6=9D=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 上一版把双进程抢会话修了,但生产重启后仍然不收消息:`start_all()` 是进程启动时的 查询,那一刻没有工作区上下文,`bot_channel` 的 RLS 把行全藏起来,一行都查不到, 于是"页面上显示已连接,实际上没人在拉消息"。 - 迁移 0016 补一条只读策略:`current_setting('app.bot_list') = 'on'` 时允许 SELECT (只开 SELECT,防止顺手拿它改别人的通道); - `start_all()` 走 `active_channel_ids()`,先设这个 GUC 再列通道; - 补测试 `test_startup_sees_logged_in_channels`。 --- 2.Design/ROADMAP-M0-M1.md | 7 +++ .../0016_bot_channel_startup_policy.py | 43 +++++++++++++++++++ backend/app/services/wechat_bot.py | 26 +++++++---- backend/tests/test_wechat_bot.py | 23 ++++++++++ 4 files changed, 91 insertions(+), 8 deletions(-) create mode 100644 backend/alembic/versions/0016_bot_channel_startup_policy.py diff --git a/2.Design/ROADMAP-M0-M1.md b/2.Design/ROADMAP-M0-M1.md index 1148200..87ce332 100644 --- a/2.Design/ROADMAP-M0-M1.md +++ b/2.Design/ROADMAP-M0-M1.md @@ -824,6 +824,13 @@ 2. **回话别再带站点链接**:`PUBLIC_BASE_URL` 是内网地址,手机上打不开; 现在只回「已保存任务:…」/「已保存知识草稿:…」。 3. **栏目名从「接入」改成「机器人」**:用户要的是「我和这个机器人说话」,不是「配置接入」。 +4. **重启之后机器人静默不收消息**:生产是 `gunicorn -w 2`,两个进程各跑一遍 lifespan, + 同一个通道会被起两条长轮询(消息被一个消费、回复从另一个发出,游标互相覆盖); + 而且 `start_all()` 是"没有工作区上下文"的查询,被 RLS 全藏起来,一行都查不到。 + - 租约:会话级 `pg_try_advisory_lock`,抢到的进程收,另一个每 10 秒重试;进程挂了连接断、 + 锁自动释放,备用进程接上。页面「接收中」只认抢到租约的那个。 + - 迁移 `0016` 补一条只读策略:`current_setting('app.bot_list') = 'on'` 时允许 SELECT, + `start_all()` 先设这个 GUC 再列通道。 ### 13.7 教训:第一版是猜的 diff --git a/backend/alembic/versions/0016_bot_channel_startup_policy.py b/backend/alembic/versions/0016_bot_channel_startup_policy.py new file mode 100644 index 0000000..6249129 --- /dev/null +++ b/backend/alembic/versions/0016_bot_channel_startup_policy.py @@ -0,0 +1,43 @@ +"""启动时列出要接管的通道,也需要一条 RLS 口子 + +Revision ID: 0016 +Revises: 0015 +Create Date: 2026-11-05 + +`BotManager.start_all()` 要在进程启动时问一句"现在有哪些通道已登录、该接着收消息", +但 `bot_channel` 的隔离策略要求 `workspace_id = app_ws_id()`,启动那一刻**没有工作区**, +结果一行都查不到——表现是"重启之后机器人静默不收消息,页面上还显示已连接"。 + +0014 那条口子是"知道 id、只看自己那一行",这里补的是"允许列出还没开始干的活": +只有当次事务显式设了 `app.bot_list = 'on'` 时才放行,而且只开 **SELECT**, +防止有人顺手拿它改别人的通道。 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +from app.roles import APP_ROLE + +revision: str = "0016" +down_revision: str = "0015" +branch_labels: Sequence[str] | None = None +depends_on: Sequence[str] | None = None + +CREATE_POLICY = """ +CREATE POLICY bot_channel_startup ON bot_channel + FOR SELECT + USING (current_setting('app.bot_list', true) = 'on') +""" + + +def upgrade() -> None: + op.execute("DROP POLICY IF EXISTS bot_channel_startup ON bot_channel") + op.execute(CREATE_POLICY) + op.execute(f"GRANT SELECT ON bot_channel TO {APP_ROLE}") + + +def downgrade() -> None: + op.execute("DROP POLICY IF EXISTS bot_channel_startup ON bot_channel") diff --git a/backend/app/services/wechat_bot.py b/backend/app/services/wechat_bot.py index 5a15277..1e3b7dd 100644 --- a/backend/app/services/wechat_bot.py +++ b/backend/app/services/wechat_bot.py @@ -54,6 +54,23 @@ def site_base() -> str: return "" +async def active_channel_ids() -> list[uuid.UUID]: + """列出"已扫码登录、该收消息"的通道。 + + 启动时还没有工作区上下文,`bot_channel` 的 RLS 会把行全藏起来,所以先设 + `app.bot_list='on'` 走迁移 0016 那条只读口子——不设的话重启后机器人会静默失联。 + """ + async with SessionFactory() as session, session.begin(): + await session.execute(sa_text("SELECT set_config('app.bot_list', 'on', true)")) + rows = await session.execute( + select(BotChannel.id).where( + BotChannel.status == "active", + BotChannel.bot_token_enc.is_not(None), + ) + ) + return [row[0] for row in rows] + + async def acquire_lease(channel_id: uuid.UUID): """抢"这个通道由我收消息"的租约;抢到返回连接(关掉就是释放),抢不到返回 None。 @@ -103,14 +120,7 @@ class BotManager: async def start_all(self) -> None: """进程启动时把已登录的通道都拉起来,不然重启就静默不收消息了。""" - async with SessionFactory() as session: - rows = await session.execute( - select(BotChannel.id).where( - BotChannel.status == "active", - BotChannel.bot_token_enc.is_not(None), - ) - ) - ids = [row[0] for row in rows] + ids = await active_channel_ids() for channel_id in ids: await self.start(channel_id) if ids: diff --git a/backend/tests/test_wechat_bot.py b/backend/tests/test_wechat_bot.py index a616a99..9016c62 100644 --- a/backend/tests/test_wechat_bot.py +++ b/backend/tests/test_wechat_bot.py @@ -354,3 +354,26 @@ async def test_only_one_process_holds_the_polling_lease() -> None: third = await wechat_bot.acquire_lease(channel_id) assert third is not None, "前一个进程放手后,备用进程要能接上" await third.close() + + +async def test_startup_sees_logged_in_channels(client: AsyncClient) -> None: + """重启后要能把已登录的通道接回来:没有工作区上下文时也得看得见(迁移 0016)。""" + user = await login(client) + ws = await create_workspace(client, user, "重启接管") + ws_id = uuid.UUID(ws["id"]) + + async with SessionFactory() as session: + await bind_tenant(session, ws_id, uuid.UUID(user["user_id"])) + channel = BotChannel( + workspace_id=ws_id, + kind="wechat_clawbot", + name="微信", + status="active", + bot_token_enc=wechat_ilink.encrypt_token("bot-token-x"), + ) + session.add(channel) + await session.commit() + channel_id = channel.id + + ids = await wechat_bot.active_channel_ids() + assert channel_id in ids, "启动时看不见已登录通道,重启后机器人就静默失联了"