feat(host): every read the host follows over Realtime shares one socket (P1)

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

@@ -3,12 +3,14 @@ import { it } from "@effect/vitest";
 import { LIVE_POLL_URL, 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";
 
 /**
@@ -235,3 +237,108 @@ it.effect(
       yield* Fiber.interrupt(following);
     }),
 );
+
+/**
+ * Every read the host follows at one address shares one Realtime client and
+ * its one socket: a second read opens nothing new, letting one go leaves the
+ * other's subscriptions standing, the socket closes once the last is let go,
+ * a handle let go twice counts once, and the next read opens a fresh one.
+ * Reads that wait on a client still loading share it too, and a client that
+ * failed to load is asked for afresh by the next read.
+ *
+ * Realtime's client is a script that counts what it opened and closed.
+ */
+
+function scriptedClients() {
+  const opened: string[] = [];
+  const closed: string[] = [];
+  const subscribed: string[] = [];
+  const listeners: ((snapshot: { readonly status: string; readonly data: undefined }) => void)[] =
+    [];
+  const open = (url: string): LiveClient => {
+    opened.push(url);
+    return {
+      subscribe: (query) => {
+        subscribed.push(query.capability);
+        return {
+          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, subscribed, fail };
+}
+
+const sealedAs = (capability: string) => ({
+  capability,
+  queryFingerprint: capability,
+  expiresAt: 0,
+});
+
+test("reads at one address share one client, closed once the last lets go", async () => {
+  const clients = scriptedClients();
+  const open = shareLiveClients(clients.open);
+
+  const [inbox, plan] = await Promise.all([
+    open("wss://realtime.example/v1"),
+    open("wss://realtime.example/v1"),
+  ]);
+  inbox.subscribe(sealedAs("inbox"));
+  plan.subscribe(sealedAs("plan"));
+  assert.deepEqual(clients.opened, ["wss://realtime.example/v1"]);
+  assert.deepEqual(clients.subscribed, ["inbox", "plan"]);
+
+  inbox.close();
+  inbox.close();
+  assert.deepEqual(clients.closed, [], "a handle let go twice counts once, and the plan's stands");
+  plan.close();
+  assert.deepEqual(clients.closed, ["wss://realtime.example/v1"]);
+
+  await open("wss://realtime.example/v1");
+  assert.equal(clients.opened.length, 2, "the next read opens a fresh client");
+});
+
+test("a client a subscription saw fail is replaced for the next read, and closed once its holders let go", async () => {
+  const clients = scriptedClients();
+  const open = shareLiveClients(clients.open);
+  const inbox = await open("wss://realtime.example/v1");
+  inbox.subscribe(sealedAs("inbox"));
+  clients.fail();
+
+  const plan = await open("wss://realtime.example/v1");
+  assert.equal(clients.opened.length, 2, "the next read opens a new client");
+  inbox.close();
+  assert.deepEqual(clients.closed, ["wss://realtime.example/v1"], "the failed one closes");
+  plan.close();
+  assert.equal(clients.closed.length, 2);
+});
+
+test("reads at two addresses each have their own client", async () => {
+  const clients = scriptedClients();
+  const open = shareLiveClients(clients.open);
+  await open("wss://one.example/v1");
+  await open("wss://two.example/v1");
+  assert.deepEqual(clients.opened, ["wss://one.example/v1", "wss://two.example/v1"]);
+});
+
+test("a client that failed to load is opened afresh by the next read", async () => {
+  const clients = scriptedClients();
+  let loads = 0;
+  const open = shareLiveClients((url) => {
+    loads += 1;
+    return loads === 1 ? Promise.reject(new Error("offline")) : clients.open(url);
+  });
+  await assert.rejects(open("wss://realtime.example/v1"));
+  await open("wss://realtime.example/v1");
+  assert.deepEqual(clients.opened, ["wss://realtime.example/v1"]);
+});

packages/host/src/live-follow.ts

@@ -4,8 +4,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
+ * following 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
@@ -59,12 +61,72 @@ 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's socket closes once the last handle is
+ * let go. Neon's client multiplexes every subscription over its one socket,
+ * so the host holds one socket however many reads it follows, rather than
+ * one per read. The client is opened once per address, however many reads
+ * wait on it while it loads; one that failed to open is forgotten, so the
+ * next read opens it afresh. So is one any of whose subscriptions moved to
+ * `error`: Neon's guide says a client whose connection failed cannot
+ * reconnect ("Create a new client with `createRealtimeClient()` before you
+ * subscribe again"), so the next read opens a new one, and the old one
+ * closes once its last holder lets go.
  */
-export const openRealtimeClient = (url: string): Promise<LiveClient> =>
-  import("@neon/realtime/client").then(({ createRealtimeClient }) => createRealtimeClient({ url }));
+export function shareLiveClients(
+  open: (url: string) => LiveClient | PromiseLike<LiveClient>,
+): (url: string) => Promise<LiveClient> {
+  const standing = new Map<string, { readonly client: Promise<LiveClient>; holders: number }>();
+  const forget = (url: string, shared: { readonly client: Promise<LiveClient> }) => {
+    if (standing.get(url) === shared) standing.delete(url);
+  };
+  return async (url) => {
+    let shared = standing.get(url);
+    if (shared === undefined) {
+      const opening = { 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);
+        // A subscription in error may be its connection's, which Neon's guide says never recovers: the client is replaced for the next read rather than shared on.
+        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();
+      },
+    };
+  };
+}
+
+/**
+ * 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 = shareLiveClients((url) =>
+  import("@neon/realtime/client").then(({ createRealtimeClient }) => createRealtimeClient({ url })),
+);
 
 /** A read sealed for following: where to subscribe, and each part's sealed statement. */
 export interface SealedRead<Part extends string> {