Archived
fix(app): 广播桥 —— 转发进来的不再 NOTIFY 回去,治两台后端互桥回弹的广播风暴
「自己发的自己不收」只挡一跳:收到别人的就 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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0167qWjeyBwvfdGovd7FWbCu
This commit is contained in:
@@ -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<Touch>();
|
||||
|
||||
/** 本地 → 库:自己推的顺手 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<Wire>;
|
||||
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:丢这一条,不拖垮连接 */
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user