From d67f744f598ffecd1713d98fb8fa943fefad701f Mon Sep 17 00:00:00 2001 From: xcs Date: Thu, 27 Aug 2026 11:44:35 +0800 Subject: [PATCH] =?UTF-8?q?fix(app):=20=E5=B9=BF=E6=92=AD=E6=A1=A5=20?= =?UTF-8?q?=E2=80=94=E2=80=94=20=E8=BD=AC=E5=8F=91=E8=BF=9B=E6=9D=A5?= =?UTF-8?q?=E7=9A=84=E4=B8=8D=E5=86=8D=20NOTIFY=20=E5=9B=9E=E5=8E=BB,?= =?UTF-8?q?=E6=B2=BB=E4=B8=A4=E5=8F=B0=E5=90=8E=E7=AB=AF=E4=BA=92=E6=A1=A5?= =?UTF-8?q?=E5=9B=9E=E5=BC=B9=E7=9A=84=E5=B9=BF=E6=92=AD=E9=A3=8E=E6=9A=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 「自己发的自己不收」只挡一跳:收到别人的就 publish 到本地,而桥自己也订阅着这个 broadcaster, 那条转发被当成「自己推的」再 NOTIFY 一遍(from 换成自己)→ 对面不是它发的、再转发……两台接力 就是永动机。实测两台后端连同一个库,一次 POST /api/gov/process/validate 引出 45 秒 19 万条广播, 工作台三把查询被打成连续重取(用户看到的是「疯狂轮询三个接口」)。 转发进来的那条打记号(WeakSet,publish 同步派发),「本地 → 库」那一侧认出记号就不发; `bridgeBroadcaster` 第三参 `self` 可注入(同进程两座桥才模拟得出两台机器)。 2 测:转发不再 NOTIFY 出去 · 两台互桥一条广播各到一次就停(假频道封顶 50 条,让用例失败而不是撑爆进程)。 2026-08-26 修的,代码一直留在工作树没入库;CLAUDE.md 快照已记。 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_0167qWjeyBwvfdGovd7FWbCu --- .../app/__tests__/broadcast-bridge.test.ts | 55 +++++++++++++++++++ apps/api/src/app/broadcast-bridge.ts | 20 +++++-- 2 files changed, 70 insertions(+), 5 deletions(-) diff --git a/apps/api/src/app/__tests__/broadcast-bridge.test.ts b/apps/api/src/app/__tests__/broadcast-bridge.test.ts index a5950c9..d9cb550 100644 --- a/apps/api/src/app/__tests__/broadcast-bridge.test.ts +++ b/apps/api/src/app/__tests__/broadcast-bridge.test.ts @@ -87,6 +87,61 @@ describe('bridgeBroadcaster', () => { expect(got).toEqual([]); }); + it('**转发别人的那一条不再 NOTIFY 出去** —— 转发不是「自己推的」', async () => { + const fake = fakePool(); + const bus = createBroadcaster(); + bridgeBroadcaster(fake.pool as never, bus); + await settle(); + + fake.deliver({ kind: 'system', id: 'gov', reason: 'POST /api/gov/process/validate', from: '别的机器' }); + await settle(); + /* + * 从前这里会 NOTIFY 一条(from 换成自己):对面收到不是它发的、再转发、再 NOTIFY…… + * 两台后端连同一个库时一条广播就此永动 —— 实测 45 秒 19 万条,工作台三把查询被打成连续重取。 + */ + expect(fake.notified).toEqual([]); + }); + + it('两台后端互桥,一条广播各到一次就停 —— 不来回弹', async () => { + /** 两个假池共用一条「频道」:谁 NOTIFY 了,另一个当场收到(自己也收到,和真库一样) */ + const a = fakePool(); + const b = fakePool(); + const busA = createBroadcaster(); + const busB = createBroadcaster(); + const relay = (from: typeof a, to: typeof b) => { + const origQuery = from.pool.query; + from.pool.query = async (sql: string, params: unknown[]) => { + const r = await origQuery(sql, params); + // 回弹起来是无穷的:封顶 50 条,让用例「失败」而不是把测试进程撑爆 + if (from.notified.length > 50) return r; + // 真库会把这条送给所有 LISTEN 的连接,包括发它的那个进程自己 + const wire = JSON.parse(String(params[1])) as Record; + from.deliver(wire); + to.deliver(wire); + return r; + }; + }; + // 两台机器两个身份(真环境里 SELF 按进程生成;测试里显式给,不然同进程两座桥同名) + bridgeBroadcaster(a.pool as never, busA, 'A'); + bridgeBroadcaster(b.pool as never, busB, 'B'); + await settle(); + relay(a, b); + relay(b, a); + + const seenA: string[] = []; + const seenB: string[] = []; + busA.subscribe((t) => seenA.push(t.reason)); + busB.subscribe((t) => seenB.push(t.reason)); + busA.publish({ kind: 'system', id: 'gov', reason: 'x' }); + for (let i = 0; i < 20; i += 1) await settle(); + + // A 本地一次、B 转发一次;NOTIFY 只有 A 发的那一条 —— B 转发的不再发回去 + expect(seenA).toEqual(['x']); + expect(seenB).toEqual(['x']); + expect(a.notified).toHaveLength(1); + expect(b.notified).toHaveLength(0); + }); + it('载荷坏了只丢这一条,连接照旧活着', async () => { const fake = fakePool(); const bus = createBroadcaster(); diff --git a/apps/api/src/app/broadcast-bridge.ts b/apps/api/src/app/broadcast-bridge.ts index 7d009ce..2e577e2 100644 --- a/apps/api/src/app/broadcast-bridge.ts +++ b/apps/api/src/app/broadcast-bridge.ts @@ -12,6 +12,12 @@ * · 本进程 publish 一条 → 顺手 `NOTIFY`(带上进程标识); * · 一条**专用连接**常年 `LISTEN`,收到别的进程发来的就在本地 publish 一次。 * **自己发的自己不再收**(靠进程标识认),不然一条推送在两台之间来回弹。 + * **转发进来的不再发出去**(2026-08-26):从前收到别人的就 `publish` 到本地 broadcaster, + * 而本桥自己也订阅着它 —— 那条转发被当成「自己推的」再 NOTIFY 一遍(from 换成自己), + * 对面收到不是它发的、再转发、再 NOTIFY……只挡「自己发的自己不收」这一跳,两台接力就是永动机: + * 实测两台后端连同一个库,一次 `POST /api/gov/process/validate` 引出 45 秒 19 万条广播, + * 工作台三把查询被打成连续重取(用户看到的就是「疯狂轮询三个接口」)。 + * 转发的那一条在本地 publish 前打上记号,「本地 → 库」那一侧认出记号就不再发。 * * 桥断了不影响业务:推送本就是「告诉在线浏览器去重取」的提示,丢一条最坏是 * 那一屏晚两秒(前端还有轮询兜底)。所以这里所有错误都只记一行,不往上抛。 @@ -33,14 +39,16 @@ type Wire = Touch & { readonly from: string }; * `notify` 用参数化的 `pg_notify` 而不是拼 `NOTIFY` 语句:载荷里有中文与引号, * 拼字符串迟早出事(而这条路上的字符串全来自业务)。 */ -export function bridgeBroadcaster(pool: Pool, broadcaster: Broadcaster): { stop(): void } { +export function bridgeBroadcaster(pool: Pool, broadcaster: Broadcaster, self: string = SELF): { stop(): void } { let listener: PoolClient | undefined; let stopped = false; + /** 从库上转发进来的那几条 —— 它们不是本进程推的,不许再 NOTIFY 回去(按对象身份认,publish 同步派发) */ + const relayed = new WeakSet(); /** 本地 → 库:自己推的顺手 NOTIFY 一条(不 await,广播不该拖慢业务) */ const unsubscribe = broadcaster.subscribe((touch) => { - if (stopped) return; - const wire: Wire = { ...touch, from: SELF }; + if (stopped || relayed.has(touch)) return; + const wire: Wire = { ...touch, from: self }; void pool .query('select pg_notify($1, $2)', [TOUCH_CHANNEL, JSON.stringify(wire)]) .catch((error: unknown) => { @@ -68,8 +76,10 @@ export function bridgeBroadcaster(pool: Pool, broadcaster: Broadcaster): { stop( const wire = parsed as Partial; if (typeof wire.kind !== 'string' || typeof wire.id !== 'string') return; // **自己发的自己不收** —— 否则一条推送在进程之间来回弹 - if (wire.from === SELF) return; - broadcaster.publish({ kind: wire.kind as Touch['kind'], id: wire.id, reason: wire.reason ?? 'remote' }); + if (wire.from === self) return; + const touch: Touch = { kind: wire.kind as Touch['kind'], id: wire.id, reason: wire.reason ?? 'remote' }; + relayed.add(touch); + broadcaster.publish(touch); } catch { /* 载荷根本不是 JSON:丢这一条,不拖垮连接 */ }