Tell a browser to refetch when a server's channel subscription comes back

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

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.
          *

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,13 +45,22 @@ 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 = {
   /** Attach a connection for a person. Returns the detach. */
   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<ChannelActivityListener> {
   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 () => {

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<number[]> {
+  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<number>): Promise<number[]> {
+  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<boolean>,
+  timeoutMs = 8_000,
+): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (await condition()) return;
+    await new Promise((resolve) => setTimeout(resolve, 25));
+  }
+}