diff --git a/CHANGELOG.md b/CHANGELOG.md index da9c37430..7a28adc41 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,19 @@ 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 answer comes back to the conversation that asked **This reverses what 0.0.5 shipped.** The 0.0.5 notes below say the asking Bot does not relay text diff --git a/app/src/lib/channels/use-channel-events.ts b/app/src/lib/channels/use-channel-events.ts index 8bd1df464..87a87c41e 100644 --- a/app/src/lib/channels/use-channel-events.ts +++ b/app/src/lib/channels/use-channel-events.ts @@ -8,8 +8,24 @@ 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 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 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. */ +export type ChannelSocketMessage = ChannelActivityEvent | ChannelResyncEvent; + +export function isResync( + message: ChannelSocketMessage, +): message is ChannelResyncEvent { + return (message as ChannelResyncEvent).resync === true; +} + export type ChannelActivityEvent = { channelId: string; lastMessage: string | null; @@ -155,13 +171,22 @@ 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; } + // 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; + } + + 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 9f310baba..586efa701 100644 --- a/server/src/channels/events.ts +++ b/server/src/channels/events.ts @@ -11,6 +11,10 @@ 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 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"; @@ -41,6 +45,13 @@ export type ChannelActivityEvent = { busy?: boolean; }; +/** "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({ + resync: true, +} satisfies ChannelResyncEvent); + type Send = (payload: string) => void; export type ChannelEventHub = { @@ -48,6 +59,8 @@ 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 to refetch. Everybody, because the lost events named their own members. */ + resyncAll(): void; connectionCount(userId: string): number; }; @@ -83,6 +96,18 @@ export function createChannelEventHub(): ChannelEventHub { } }, + resyncAll() { + for (const sends of connections.values()) { + for (const send of sends) { + try { + send(RESYNC_PAYLOAD); + } catch { + // Closing, and detached by its own close handler. See `deliver`. + } + } + } + }, + connectionCount(userId) { return connections.get(userId)?.size ?? 0; }, @@ -103,14 +128,36 @@ 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. + /* + * `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 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 = () => { + 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..aeea0bb3f 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,46 @@ 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: the events that were lost named their own members. + 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 +502,103 @@ describe("channel change delivery", () => { expect(watched.of(other.id)).toEqual([]); }); }); + +/** + * 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 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 () => { + 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, 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 }[]>` + 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: the topic filter excludes the policy listener, `before` the rest. */ +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 neither flaky nor slow. As policy-fanout does. */ +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)); + } +}