Host: one shared Neon Realtime client per address for every live read

packages/host/src/compose-account.ts

@@ -33,6 +33,7 @@ import { agentTraceDirectory } from "./devtrace/trace-directory.js";
 import { AgentTraceWriter } from "./devtrace/trace-writer.js";
 import { HostKernelTag, lateService } from "./effect/kernel.js";
 import { AccountSource, AppIdentity, Environment, ServiceOrigin } from "./effect/seams.js";
+import { dropRealtimeClients } from "./live-follow.js";
 
 /** What a refused account step says, worded by the step alone. */
 const SIGN_IN_REFUSED = "the sign-in did not finish";
@@ -180,7 +181,10 @@ export const composeAccount = /* @__PURE__ */ Effect.fn("host/composeAccount")(f
       previousAccount.status === ACCOUNT_STATUS.SIGNED_IN ? previousAccount.email : undefined;
     const nextAccountKey = signedIn ? next.email : undefined;
     previousAccount = next;
-    if (previousAccountKey !== nextAccountKey) settings.forgetAccountPreferenceHydration();
+    if (previousAccountKey !== nextAccountKey) {
+      settings.forgetAccountPreferenceHydration();
+      dropRealtimeClients();
+    }
     // A deletion stands recording down until the next sign-in, which starts it again.
     if (signedIn && !wasSignedIn) sessionReplayEndedByDeletion = false;
     kernel.emit(GATEWAY_EVENT.ACCOUNT_CHANGED, carried(next));

packages/host/src/live-follow.test.ts

@@ -3,12 +3,14 @@ import { it } from "@effect/vitest";
 import { LIVE_POLL_URL, LIVE_SIGNED_OUT, type SealedLiveQueryWire } from "@luke/features/live-wire";
 import { Clock, Duration, Effect, Exit, Fiber, Stream } from "effect";
 import { TestClock } from "effect/testing";
+import { test } from "vitest";
 import {
   LiveBroken,
   type LiveClient,
   liveRows,
   openRealtimeClient,
   type SealedRead,
+  shareLiveClients,
 } from "./live-follow.js";
 
 /**
@@ -341,3 +343,98 @@ it.effect("a read whose session the service refuses as signed out ends, and is a
     assert.ok(Exit.isFailure(exit));
   }),
 );
+
+/**
+ * Every read at one address shares one Realtime client: a second read opens
+ * nothing new, the client closes once the last read lets go, a connection
+ * error makes the next read open a new client, and a sign-out drops it.
+ * Realtime's client is a script that counts what it opened and closed.
+ */
+
+function scriptedClients() {
+  const opened: string[] = [];
+  const closed: string[] = [];
+  const listeners: ((snapshot: { readonly status: string; readonly data: undefined }) => void)[] =
+    [];
+  const open = (url: string): LiveClient => {
+    opened.push(url);
+    return {
+      subscribe: () => ({
+        onChange: (listener) => {
+          listeners.push(listener);
+          return () => undefined;
+        },
+        renew: () => Promise.resolve(),
+        unsubscribe: () => undefined,
+      }),
+      close: () => void closed.push(url),
+    };
+  };
+  /** Every subscription so far moves to `error`, as a failed connection moves them. */
+  const fail = () => {
+    for (const listener of listeners) listener({ status: "error", data: undefined });
+  };
+  return { open, opened, closed, fail };
+}
+
+const ADDRESS = "wss://realtime.example/v1";
+const sealedAs = (capability: string) => ({
+  capability,
+  queryFingerprint: capability,
+  expiresAt: 0,
+});
+
+test("two reads at one address share one client, closed once the last lets go", async () => {
+  const clients = scriptedClients();
+  const shared = shareLiveClients(clients.open);
+  const [inbox, plan] = await Promise.all([shared.open(ADDRESS), shared.open(ADDRESS)]);
+  inbox.subscribe(sealedAs("inbox"));
+  plan.subscribe(sealedAs("plan"));
+  assert.deepEqual(clients.opened, [ADDRESS]);
+
+  inbox.close();
+  inbox.close();
+  assert.deepEqual(clients.closed, [], "a handle let go twice counts once");
+  plan.close();
+  assert.deepEqual(clients.closed, [ADDRESS]);
+});
+
+test("a connection error makes the next read open a new client; the old closes once let go", async () => {
+  const clients = scriptedClients();
+  const shared = shareLiveClients(clients.open);
+  const inbox = await shared.open(ADDRESS);
+  inbox.subscribe(sealedAs("inbox"));
+  clients.fail();
+
+  const plan = await shared.open(ADDRESS);
+  assert.equal(clients.opened.length, 2, "the next read opens a new client");
+  inbox.close();
+  assert.equal(clients.closed.length, 1, "the failed one closes");
+  plan.close();
+  assert.equal(clients.closed.length, 2);
+});
+
+test("a sign-out drops the shared client", async () => {
+  const clients = scriptedClients();
+  const shared = shareLiveClients(clients.open);
+  const before = await shared.open(ADDRESS);
+  shared.drop();
+  const after = await shared.open(ADDRESS);
+  assert.equal(clients.opened.length, 2, "the next read opens a new client");
+  before.close();
+  assert.equal(clients.closed.length, 1, "the dropped one closes once its read lets go");
+  after.close();
+  assert.equal(clients.closed.length, 2);
+});
+
+test("a client that failed to load is loaded afresh by the next read", async () => {
+  const clients = scriptedClients();
+  let loads = 0;
+  const shared = shareLiveClients((url) => {
+    loads += 1;
+    return loads === 1 ? Promise.reject(new Error("offline")) : clients.open(url);
+  });
+  await assert.rejects(shared.open(ADDRESS));
+  await shared.open(ADDRESS);
+  assert.deepEqual(clients.opened, [ADDRESS]);
+});

packages/host/src/live-follow.ts

@@ -9,8 +9,10 @@ import { Cause, Data, Duration, Effect, Fiber, Queue, Schedule, Stream } from "e
 /**
  * live-follow.ts -- a read the host follows over Neon Realtime, as Neon's guide says to: sealed statements subscribed to at the address the service names, and renewed before they expire.
  *
- * Each read is one or more parts, one sealed statement each, all on one
- * socket ("Subscribe to the sealed query"). Once every part has its first
+ * Each read is one or more parts, one sealed statement each, and every read
+ * at one address shares one client and its one socket ("Create one client
+ * for the whole app, because every subscription shares its connection").
+ * Once every part has its first
  * result, every change of any part is the parts' rows whole, undecoded, for
  * the caller to decode at this boundary. Every seal is renewed ten seconds
  * before the earliest expires, a fetch that failed asked again a second
@@ -82,12 +84,82 @@ export interface LiveClient {
 }
 
 /**
- * Realtime's own client, at the address the service named, loaded only when
- * a read is first followed, so its protocol's validators stay out of the
- * web app's first load.
+ * One client per address, shared by every read following there: each open
+ * takes a handle, and the client closes once the last handle is let go. The
+ * client is opened once however many reads wait on it while it loads; one
+ * that failed to load is forgotten, so the next read loads it afresh. So is
+ * one any of whose subscriptions moved to `error`: where its connection
+ * failed, every subscription on it is in `error` and Neon's guide says to
+ * create a new client before subscribing again, so the next read opens a
+ * new one and the old one closes once its last holder lets go. `drop`
+ * forgets every client the same way, for a sign-out or an account switch.
  */
-export const openRealtimeClient = (url: string): Promise<LiveClient> =>
-  import("@neon/realtime/client").then(({ createRealtimeClient }) => createRealtimeClient({ url }));
+export function shareLiveClients(open: (url: string) => LiveClient | PromiseLike<LiveClient>): {
+  readonly open: (url: string) => Promise<LiveClient>;
+  readonly drop: () => void;
+} {
+  interface Shared {
+    readonly client: Promise<LiveClient>;
+    holders: number;
+  }
+  const standing = new Map<string, Shared>();
+  const forget = (url: string, shared: Shared) => {
+    if (standing.get(url) === shared) standing.delete(url);
+  };
+  return {
+    open: async (url) => {
+      let shared = standing.get(url);
+      if (shared === undefined) {
+        const opening: Shared = { client: Promise.resolve(url).then(open), holders: 0 };
+        opening.client.catch(() => forget(url, opening));
+        standing.set(url, opening);
+        shared = opening;
+      }
+      const held = shared;
+      held.holders += 1;
+      let client: LiveClient;
+      try {
+        client = await held.client;
+      } catch (error) {
+        held.holders -= 1;
+        throw error;
+      }
+      let holding = true;
+      return {
+        subscribe: (query) => {
+          const subscription = client.subscribe(query);
+          subscription.onChange((snapshot) => {
+            if (snapshot.status === "error") forget(url, held);
+          });
+          return subscription;
+        },
+        close: () => {
+          if (!holding) return;
+          holding = false;
+          held.holders -= 1;
+          if (held.holders > 0) return;
+          forget(url, held);
+          client.close();
+        },
+      };
+    },
+    drop: () => standing.clear(),
+  };
+}
+
+const realtimeClients = shareLiveClients((url) =>
+  import("@neon/realtime/client").then(({ createRealtimeClient }) => createRealtimeClient({ url })),
+);
+
+/**
+ * Realtime's own client at the address the service named, shared by every
+ * read following there, and loaded only when a read is first followed, so
+ * its protocol's validators stay out of the web app's first load.
+ */
+export const openRealtimeClient = realtimeClients.open;
+
+/** Forgets the shared Realtime clients, on a sign-out or an account switch: the next read opens a new one, and each old one closes once its reads let go. */
+export const dropRealtimeClients = realtimeClients.drop;
 
 /** A read sealed for following: where to subscribe, and each part's sealed statement. */
 export interface SealedRead<Part extends string> {