From 61c6df123db6cd4143a49ded46bb8ddc4c38c9ed Mon Sep 17 00:00:00 2001 From: Vaibhav Zope Date: Tue, 1 Sep 2026 00:54:38 +0530 Subject: [PATCH 1/2] Tell a browser to refetch when a server's channel subscription comes back --- CHANGELOG.md | 14 ++ app/src/lib/channels/use-channel-events.ts | 41 ++++- server/src/channels/events.ts | 89 +++++++++- .../tests/channel-events.integration.test.ts | 166 ++++++++++++++++++ 4 files changed, 301 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c851fcd32..be742f238 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,20 @@ Newest first. `Unreleased` is what is on `main` and not yet tagged. ## Unreleased +### A conversation deleted while a server was reconnecting no longer lingers on the screen + +Announcements between servers travel as Postgres notifications, which reach whoever is subscribed at +the moment they are sent and are never replayed. While a server's subscription was down — a database +restart, a failover, a rolling upgrade — every channel deletion, pin and message announced in that +window was lost, and nothing afterwards asked for it again. + +The browser could not notice. Its own connection to the server stayed open throughout, so the +refetch it already does when that connection comes back was never triggered, and the roster went on +showing a conversation that had been deleted until the page was reloaded. + +A server now tells the browsers it is holding to refetch when its subscription is re-established. +Nothing to configure, and no change for a deployment whose database connection never drops. + ### A Bot's shell can no longer reach the embedded database without a password In the all-in-one image the cluster was `trust`-auth on loopback, and the Bot's shell runs in the diff --git a/app/src/lib/channels/use-channel-events.ts b/app/src/lib/channels/use-channel-events.ts index 70ffab505..061246e36 100644 --- a/app/src/lib/channels/use-channel-events.ts +++ b/app/src/lib/channels/use-channel-events.ts @@ -8,7 +8,30 @@ import { type ChannelPage, type ChannelSummary, channelKeys } from "./queries"; * * The query remains the source of truth; socket events only patch its cache. Reconnects refetch the * list to recover events missed while disconnected. + * + * TWO CONNECTIONS CAN DROP, AND ONLY ONE OF THEM IS THIS ONE. `onopen` below covers this socket + * going away. The other is the server's own subscription to Postgres, which carries every event + * before it reaches this socket: while that is away the announcements are lost and never replayed, + * and this socket stays open throughout, so nothing here would ever know. The server sends a resync + * when its subscription is re-established, and it is answered with the same refetch `onopen` makes. + */ + +/** + * The server telling us it may have missed announcements, so the roster we hold may be wrong. + * + * It carries nothing else, because nothing else is knowable: what was lost was per-member and + * Postgres does not keep it. The only answer is to ask again. */ +export type ChannelResyncEvent = { resync: true }; + +/** What arrives on the socket. `resync` is the discriminant; an activity event never carries it. */ +export type ChannelSocketMessage = ChannelActivityEvent | ChannelResyncEvent; + +export function isResync( + message: ChannelSocketMessage, +): message is ChannelResyncEvent { + return (message as ChannelResyncEvent).resync === true; +} export type ChannelActivityEvent = { channelId: string; @@ -132,13 +155,27 @@ export function useChannelEvents() { }; socket.onmessage = (message) => { - let activity: ChannelActivityEvent; + let parsed: ChannelSocketMessage; try { - activity = JSON.parse(message.data as string); + parsed = JSON.parse(message.data as string); } catch { return; } + /* + * The server's subscription came back, so it may have missed announcements while it was + * away. Refetch rather than patch: there is no delta to apply, which is the whole reason + * this message exists rather than a replay of what was lost. + * + * Checked before anything reads `channelId`, because this message has none. + */ + if (isResync(parsed)) { + void queryClient.invalidateQueries({ queryKey: channelKeys.list() }); + return; + } + + const activity = parsed; + /* * The list is paged, so the cache holds pages rather than one array. * diff --git a/server/src/channels/events.ts b/server/src/channels/events.ts index b0bced6f9..08aa1e82f 100644 --- a/server/src/channels/events.ts +++ b/server/src/channels/events.ts @@ -11,6 +11,13 @@ import postgres from "postgres"; * Delivery goes through Postgres rather than an in-process list, because an in-process list is * silently wrong the moment a second server instance exists: the writer is on one and the listener * on the other, and the message is never delivered. + * + * A NOTIFY reaches whoever is subscribed at the time and is never replayed, so an announcement made + * while this server's subscription is down is gone. "Recovers by refetching on reconnect" is the + * client's rule for ITS OWN socket dropping, and that socket is not the one at risk here: the + * browser's connection to this server is untouched while the server's connection to Postgres is + * away, so nothing on the client ever learns it missed anything. `resyncAll` below is that missing + * signal, sent when the subscription is re-established. See `startChannelActivityListener`. */ export const CHANNEL_ACTIVITY_TOPIC = "channel_activity"; @@ -33,6 +40,20 @@ export type ChannelActivityEvent = { pinned?: boolean; }; +/** + * Told to every connection that this server may have missed announcements. + * + * Carries no channel and no delta, because a missed NOTIFY cannot be reconstructed: what was lost is + * per-member and Postgres does not keep it. All this says is "the roster you hold may be wrong", and + * the client answers it by refetching the roster — the same recovery it already runs when its own + * socket reconnects, reached from the one direction that had no way to trigger it. + */ +export type ChannelResyncEvent = { resync: true }; + +const RESYNC_PAYLOAD = JSON.stringify({ + resync: true, +} satisfies ChannelResyncEvent); + type Send = (payload: string) => void; export type ChannelEventHub = { @@ -40,6 +61,13 @@ export type ChannelEventHub = { register(userId: string, send: Send): () => void; /** Fan one event out to this instance's own connections. */ deliver(event: ChannelActivityEvent): void; + /** + * Tell every connection on this instance to refetch, because announcements may have been missed. + * + * Everybody rather than a member list, because what was missed is not known: the events that were + * lost named their own recipients and those events are gone. + */ + resyncAll(): void; connectionCount(userId: string): number; }; @@ -75,6 +103,19 @@ export function createChannelEventHub(): ChannelEventHub { } }, + resyncAll() { + for (const sends of connections.values()) { + for (const send of sends) { + try { + send(RESYNC_PAYLOAD); + } catch { + // A connection that cannot be written to is one that is closing, and its own close + // handler detaches it. Failing here would deny the resync to everybody after it. + } + } + } + }, + connectionCount(userId) { return connections.get(userId)?.size ?? 0; }, @@ -95,14 +136,48 @@ export async function startChannelActivityListener( ): Promise { const connection = postgres(databaseUrl, { max: 1 }); - await connection.listen(CHANNEL_ACTIVITY_TOPIC, (payload) => { - try { - hub.deliver(JSON.parse(payload) as ChannelActivityEvent); - } catch { - // A payload we cannot read is not a reason to tear down the subscription: the roster query is - // still correct, and the next refetch shows whatever this event would have. + /* + * Every establish after the first, which is the moment this server could have missed something. + * + * `onlisten` fires when the driver establishes the subscription and again on every reconnect, so + * this is the same hook the action policy listener uses to re-read its row. What it cannot do here + * is re-read anything: the policy is one row and this is a stream of per-member deltas Postgres + * does not keep. So the recovery is handed to the browsers, which already know how to do it — the + * resync tells them to refetch the roster, exactly as their own `onopen` does. + * + * THE FIRST ESTABLISH IS SKIPPED, and the flag is what keeps the message meaning one thing. A + * resync says "there was a gap, and something announced in it may have been yours". The first + * establish has no gap behind it: there is no earlier subscription for anything to have been + * missed between. Sending one anyway would ask every connection to refetch on a boot where nothing + * was lost, and would make the message mean "possibly a gap", which is not a thing a client can + * act on differently. + * + * The cost is one roster refetch per connected tab per reconnect. That is the same burst the + * deployment already absorbs whenever this server restarts and every browser's own socket + * reconnects at once, so it is a shape the roster query is already sized for, and a Postgres + * reconnect is rarer than a deploy. + */ + let subscribed = false; + const resync = () => { + if (!subscribed) { + subscribed = true; + return; } - }); + hub.resyncAll(); + }; + + await connection.listen( + CHANNEL_ACTIVITY_TOPIC, + (payload) => { + try { + hub.deliver(JSON.parse(payload) as ChannelActivityEvent); + } catch { + // A payload we cannot read is not a reason to tear down the subscription: the roster query + // is still correct, and the next refetch shows whatever this event would have. + } + }, + resync, + ); return { stop: async () => { diff --git a/server/tests/channel-events.integration.test.ts b/server/tests/channel-events.integration.test.ts index 75c195f6c..caa554b72 100644 --- a/server/tests/channel-events.integration.test.ts +++ b/server/tests/channel-events.integration.test.ts @@ -4,6 +4,7 @@ import { eq } from "drizzle-orm"; import { createAgentProfileStore } from "../src/agents/profile-store"; import type { AgentActor } from "../src/agents/profile-types"; import { + CHANNEL_ACTIVITY_TOPIC, type ChannelActivityEvent, type ChannelEventHub, createChannelEventHub, @@ -66,6 +67,49 @@ describe("channel event hub", () => { expect(hub.connectionCount("user-1")).toBe(2); }); + test("a resync reaches every connection, whoever it belongs to", () => { + /* + * Everybody, not a member list. A resync is sent because announcements were missed and those + * announcements are gone, so there is no list of who they were for left to narrow it down with. + */ + const hub = createChannelEventHub(); + const first: string[] = []; + const second: string[] = []; + const other: string[] = []; + hub.register("user-1", (payload) => first.push(payload)); + hub.register("user-1", (payload) => second.push(payload)); + hub.register("user-2", (payload) => other.push(payload)); + + hub.resyncAll(); + + expect(first).toEqual(['{"resync":true}']); + expect(second).toEqual(['{"resync":true}']); + expect(other).toEqual(['{"resync":true}']); + }); + + test("a detached connection receives no resync", () => { + const hub = createChannelEventHub(); + const received: string[] = []; + const detach = hub.register("user-1", (payload) => received.push(payload)); + detach(); + + hub.resyncAll(); + + expect(received).toEqual([]); + }); + + test("one failing connection does not deny the resync to the rest", () => { + const hub = createChannelEventHub(); + const received: string[] = []; + hub.register("user-1", () => { + throw new Error("this connection is closing"); + }); + hub.register("user-1", (payload) => received.push(payload)); + + expect(() => hub.resyncAll()).not.toThrow(); + expect(received).toEqual(['{"resync":true}']); + }); + test("stops delivering once a connection detaches, and forgets the person", () => { const hub = createChannelEventHub(); const received: string[] = []; @@ -461,3 +505,125 @@ describe("channel change delivery", () => { expect(watched.of(other.id)).toEqual([]); }); }); + +/** + * A NOTIFY reaches whoever is subscribed at the time, and Postgres never replays it. + * + * So an announcement made while this server's subscription is down is lost — and the connection that + * dropped is not the browser's. Its socket to this server stays open throughout, so the client's own + * `onopen` recovery in `app/src/lib/channels/use-channel-events.ts` never fires and nothing on + * either side ever finds out. The roster then renders a channel that no longer resolves. + * + * The action policy listener answers the same problem with `onlisten`, which fires on every + * establish including a reconnect; see the sibling test "catches up when its subscription comes + * back" in policy-fanout.integration.test.ts. There is nothing to re-read here, so this half hands + * the recovery to the browsers instead: they are told to refetch, which is what they already do when + * their own socket comes back. + * + * The drop is a real one rather than a simulated one. The subscription's backend is terminated and + * held down until `pg_stat_activity` shows it gone, so the announcement below is published into a + * gap that is known to exist rather than assumed to. + */ +describe("a server whose subscription dropped", () => { + test("tells its connections to refetch when the subscription comes back", async () => { + const hub = createChannelEventHub(); + // The browser's end, registered once and never touched again: its socket does not drop. + const received: string[] = []; + hub.register("user-1", (payload) => received.push(payload)); + + const before = new Set(await listenBackendPids()); + const listener = await startChannelActivityListener(databaseUrl, hub); + try { + await until(async () => (await ourPids(before)).length === 1); + + // Control: while connected, an announcement arrives. Without this the test could pass on a + // hub that was never wired to the database at all. + await announce(event({ channelId: "channel_connected" })); + await until(() => + received.some((raw) => raw.includes("channel_connected")), + ); + expect(received.some((raw) => raw.includes("channel_connected"))).toBe( + true, + ); + + // Terminated in a loop, because the driver reconnects in tens of milliseconds. The publish + // happens only once no LISTEN backend of ours is connected. + let published = false; + const holdUntil = Date.now() + 5_000; + while (Date.now() < holdUntil && !published) { + for (const pid of await ourPids(before)) { + await database.$client`select pg_terminate_backend(${pid})`; + } + if ((await ourPids(before)).length === 0) { + await announce( + event({ channelId: "channel_in_the_gap", deleted: true }), + ); + published = true; + } + } + expect(published).toBe(true); + + // The subscription is live again, proven by an event that arrives after it. + await until(async () => (await ourPids(before)).length === 1); + await until(async () => { + await announce(event({ channelId: "channel_after" })); + return received.some((raw) => raw.includes("channel_after")); + }, 10_000); + expect(received.some((raw) => raw.includes("channel_after"))).toBe(true); + + // The lost announcement is not recoverable and is deliberately not asserted for. What must + // reach the browser is the instruction to ask again. + expect(received.some((raw) => raw.includes("channel_in_the_gap"))).toBe( + false, + ); + expect(received).toContain('{"resync":true}'); + } finally { + await listener.stop().catch(() => undefined); + } + }, 30_000); +}); + +/** + * Backends subscribed to THIS topic, which is how a subscription is found in order to be broken. + * + * Scoped to the topic rather than to `listen%`, because the action policy listener is a subscription + * too and a test that terminates other people's connections is a test that breaks whatever is + * running beside it. `pg_stat_activity.query` holds the statement, which for a subscription is + * `listen "channel_activity"`, so the two are told apart by name rather than by timing. + */ +async function listenBackendPids(): Promise { + const rows = await database.$client<{ pid: number }[]>` + select pid from pg_stat_activity + where query = ${`listen "${CHANNEL_ACTIVITY_TOPIC}"`} and pid <> pg_backend_pid()`; + return rows.map((row) => Number(row.pid)); +} + +/** + * The listener this test started, and nothing else. + * + * Both halves are load-bearing. The topic filter in `listenBackendPids` keeps this off the policy + * listener; `before` keeps it off any channel subscription that was already up when this test began, + * including one left behind by a process sharing the database. + */ +async function ourPids(before: Set): Promise { + return (await listenBackendPids()).filter((pid) => !before.has(pid)); +} + +function announce(activity: ChannelActivityEvent) { + return database.$client`select pg_notify(${CHANNEL_ACTIVITY_TOPIC}, ${JSON.stringify(activity)})`; +} + +/** + * Polled rather than slept, so the test is not a fixed delay that is either flaky or slow. The same + * helper policy-fanout.integration.test.ts uses, for the same reason. + */ +async function until( + condition: () => boolean | Promise, + timeoutMs = 8_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await condition()) return; + await new Promise((resolve) => setTimeout(resolve, 25)); + } +} From 91fa5d5a04e7f9cef3d45548745513db822d007c Mon Sep 17 00:00:00 2001 From: Vaibhav Zope Date: Tue, 1 Sep 2026 01:11:09 +0530 Subject: [PATCH 2/2] Say the resync rule once each, instead of four times over --- app/src/lib/channels/use-channel-events.ts | 24 +++------ server/src/channels/events.ts | 52 +++++-------------- .../tests/channel-events.integration.test.ts | 47 ++++------------- 3 files changed, 29 insertions(+), 94 deletions(-) diff --git a/app/src/lib/channels/use-channel-events.ts b/app/src/lib/channels/use-channel-events.ts index 061246e36..938667ddf 100644 --- a/app/src/lib/channels/use-channel-events.ts +++ b/app/src/lib/channels/use-channel-events.ts @@ -9,19 +9,12 @@ import { type ChannelPage, type ChannelSummary, channelKeys } from "./queries"; * The query remains the source of truth; socket events only patch its cache. Reconnects refetch the * list to recover events missed while disconnected. * - * TWO CONNECTIONS CAN DROP, AND ONLY ONE OF THEM IS THIS ONE. `onopen` below covers this socket - * going away. The other is the server's own subscription to Postgres, which carries every event - * before it reaches this socket: while that is away the announcements are lost and never replayed, - * and this socket stays open throughout, so nothing here would ever know. The server sends a resync - * when its subscription is re-established, and it is answered with the same refetch `onopen` makes. + * Two connections can drop and only one is this one. `onopen` covers this socket. The other is the + * server's subscription to Postgres, which stays invisible here — so the server sends a resync when + * it comes back, answered with the same refetch. */ -/** - * The server telling us it may have missed announcements, so the roster we hold may be wrong. - * - * It carries nothing else, because nothing else is knowable: what was lost was per-member and - * Postgres does not keep it. The only answer is to ask again. - */ +/** The server saying it may have missed announcements, so the roster we hold may be wrong. */ export type ChannelResyncEvent = { resync: true }; /** What arrives on the socket. `resync` is the discriminant; an activity event never carries it. */ @@ -162,13 +155,8 @@ export function useChannelEvents() { return; } - /* - * The server's subscription came back, so it may have missed announcements while it was - * away. Refetch rather than patch: there is no delta to apply, which is the whole reason - * this message exists rather than a replay of what was lost. - * - * Checked before anything reads `channelId`, because this message has none. - */ + // Refetch rather than patch: there is no delta to apply. Checked before anything reads + // `channelId`, because this message has none. if (isResync(parsed)) { void queryClient.invalidateQueries({ queryKey: channelKeys.list() }); return; diff --git a/server/src/channels/events.ts b/server/src/channels/events.ts index 08aa1e82f..5b27c85f7 100644 --- a/server/src/channels/events.ts +++ b/server/src/channels/events.ts @@ -12,12 +12,9 @@ import postgres from "postgres"; * silently wrong the moment a second server instance exists: the writer is on one and the listener * on the other, and the message is never delivered. * - * A NOTIFY reaches whoever is subscribed at the time and is never replayed, so an announcement made - * while this server's subscription is down is gone. "Recovers by refetching on reconnect" is the - * client's rule for ITS OWN socket dropping, and that socket is not the one at risk here: the - * browser's connection to this server is untouched while the server's connection to Postgres is - * away, so nothing on the client ever learns it missed anything. `resyncAll` below is that missing - * signal, sent when the subscription is re-established. See `startChannelActivityListener`. + * A NOTIFY is never replayed, so an announcement made while this subscription is down is gone. The + * client's own "refetch on reconnect" does not cover it: the socket that dropped is this server's, + * not the browser's. `resyncAll` is the signal that closes that gap. */ export const CHANNEL_ACTIVITY_TOPIC = "channel_activity"; @@ -40,14 +37,7 @@ export type ChannelActivityEvent = { pinned?: boolean; }; -/** - * Told to every connection that this server may have missed announcements. - * - * Carries no channel and no delta, because a missed NOTIFY cannot be reconstructed: what was lost is - * per-member and Postgres does not keep it. All this says is "the roster you hold may be wrong", and - * the client answers it by refetching the roster — the same recovery it already runs when its own - * socket reconnects, reached from the one direction that had no way to trigger it. - */ +/** "The roster you hold may be wrong." Carries no delta, because what was lost is not recoverable. */ export type ChannelResyncEvent = { resync: true }; const RESYNC_PAYLOAD = JSON.stringify({ @@ -61,12 +51,7 @@ export type ChannelEventHub = { register(userId: string, send: Send): () => void; /** Fan one event out to this instance's own connections. */ deliver(event: ChannelActivityEvent): void; - /** - * Tell every connection on this instance to refetch, because announcements may have been missed. - * - * Everybody rather than a member list, because what was missed is not known: the events that were - * lost named their own recipients and those events are gone. - */ + /** Tell every connection to refetch. Everybody, because the lost events named their own members. */ resyncAll(): void; connectionCount(userId: string): number; }; @@ -109,8 +94,7 @@ export function createChannelEventHub(): ChannelEventHub { try { send(RESYNC_PAYLOAD); } catch { - // A connection that cannot be written to is one that is closing, and its own close - // handler detaches it. Failing here would deny the resync to everybody after it. + // Closing, and detached by its own close handler. See `deliver`. } } } @@ -137,25 +121,13 @@ export async function startChannelActivityListener( const connection = postgres(databaseUrl, { max: 1 }); /* - * Every establish after the first, which is the moment this server could have missed something. - * - * `onlisten` fires when the driver establishes the subscription and again on every reconnect, so - * this is the same hook the action policy listener uses to re-read its row. What it cannot do here - * is re-read anything: the policy is one row and this is a stream of per-member deltas Postgres - * does not keep. So the recovery is handed to the browsers, which already know how to do it — the - * resync tells them to refetch the roster, exactly as their own `onopen` does. - * - * THE FIRST ESTABLISH IS SKIPPED, and the flag is what keeps the message meaning one thing. A - * resync says "there was a gap, and something announced in it may have been yours". The first - * establish has no gap behind it: there is no earlier subscription for anything to have been - * missed between. Sending one anyway would ask every connection to refetch on a boot where nothing - * was lost, and would make the message mean "possibly a gap", which is not a thing a client can - * act on differently. + * `onlisten` fires on every establish, reconnects included — the same hook `policy-listener.ts` + * uses to re-read its row. There is no row to re-read here, so the browsers are told to refetch + * instead. * - * The cost is one roster refetch per connected tab per reconnect. That is the same burst the - * deployment already absorbs whenever this server restarts and every browser's own socket - * reconnects at once, so it is a shape the roster query is already sized for, and a Postgres - * reconnect is rarer than a deploy. + * The first establish is skipped so the message means one thing. It has no earlier subscription + * behind it, so nothing can have been missed, and a resync there would say "possibly a gap" — not + * something a client can act on differently. */ let subscribed = false; const resync = () => { diff --git a/server/tests/channel-events.integration.test.ts b/server/tests/channel-events.integration.test.ts index caa554b72..aeea0bb3f 100644 --- a/server/tests/channel-events.integration.test.ts +++ b/server/tests/channel-events.integration.test.ts @@ -68,10 +68,7 @@ describe("channel event hub", () => { }); test("a resync reaches every connection, whoever it belongs to", () => { - /* - * Everybody, not a member list. A resync is sent because announcements were missed and those - * announcements are gone, so there is no list of who they were for left to narrow it down with. - */ + // Everybody, not a member list: the events that were lost named their own members. const hub = createChannelEventHub(); const first: string[] = []; const second: string[] = []; @@ -507,22 +504,13 @@ describe("channel change delivery", () => { }); /** - * A NOTIFY reaches whoever is subscribed at the time, and Postgres never replays it. - * - * So an announcement made while this server's subscription is down is lost — and the connection that - * dropped is not the browser's. Its socket to this server stays open throughout, so the client's own - * `onopen` recovery in `app/src/lib/channels/use-channel-events.ts` never fires and nothing on - * either side ever finds out. The roster then renders a channel that no longer resolves. + * An announcement made while this server's subscription is down is lost, and the connection that + * dropped is not the browser's — so the client's own `onopen` recovery never fires and the roster + * goes on rendering a channel that no longer resolves. * - * The action policy listener answers the same problem with `onlisten`, which fires on every - * establish including a reconnect; see the sibling test "catches up when its subscription comes - * back" in policy-fanout.integration.test.ts. There is nothing to re-read here, so this half hands - * the recovery to the browsers instead: they are told to refetch, which is what they already do when - * their own socket comes back. - * - * The drop is a real one rather than a simulated one. The subscription's backend is terminated and - * held down until `pg_stat_activity` shows it gone, so the announcement below is published into a - * gap that is known to exist rather than assumed to. + * The drop is real rather than simulated: the backend is terminated and held down until + * `pg_stat_activity` shows it gone, so the announcement lands in a gap known to exist. Compare + * "catches up when its subscription comes back" in policy-fanout.integration.test.ts. */ describe("a server whose subscription dropped", () => { test("tells its connections to refetch when the subscription comes back", async () => { @@ -584,12 +572,8 @@ describe("a server whose subscription dropped", () => { }); /** - * Backends subscribed to THIS topic, which is how a subscription is found in order to be broken. - * - * Scoped to the topic rather than to `listen%`, because the action policy listener is a subscription - * too and a test that terminates other people's connections is a test that breaks whatever is - * running beside it. `pg_stat_activity.query` holds the statement, which for a subscription is - * `listen "channel_activity"`, so the two are told apart by name rather than by timing. + * Backends subscribed to THIS topic, scoped by name rather than `listen%` so that terminating them + * can never reach the action policy listener's subscription running beside it. */ async function listenBackendPids(): Promise { const rows = await database.$client<{ pid: number }[]>` @@ -598,13 +582,7 @@ async function listenBackendPids(): Promise { return rows.map((row) => Number(row.pid)); } -/** - * The listener this test started, and nothing else. - * - * Both halves are load-bearing. The topic filter in `listenBackendPids` keeps this off the policy - * listener; `before` keeps it off any channel subscription that was already up when this test began, - * including one left behind by a process sharing the database. - */ +/** The listener this test started: the topic filter excludes the policy listener, `before` the rest. */ async function ourPids(before: Set): Promise { return (await listenBackendPids()).filter((pid) => !before.has(pid)); } @@ -613,10 +591,7 @@ function announce(activity: ChannelActivityEvent) { return database.$client`select pg_notify(${CHANNEL_ACTIVITY_TOPIC}, ${JSON.stringify(activity)})`; } -/** - * Polled rather than slept, so the test is not a fixed delay that is either flaky or slow. The same - * helper policy-fanout.integration.test.ts uses, for the same reason. - */ +/** Polled rather than slept, so the test is neither flaky nor slow. As policy-fanout does. */ async function until( condition: () => boolean | Promise, timeoutMs = 8_000,