diff --git a/packages/host/src/live-follow.test.ts b/packages/host/src/live-follow.test.ts index 471f5103b5..9c853ba625 100644 --- a/packages/host/src/live-follow.test.ts +++ b/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"]); +}); diff --git a/packages/host/src/live-follow.ts b/packages/host/src/live-follow.ts index 44a6db8fd9..7a642cec3c 100644 --- a/packages/host/src/live-follow.ts +++ b/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 => - import("@neon/realtime/client").then(({ createRealtimeClient }) => createRealtimeClient({ url })); +export function shareLiveClients( + open: (url: string) => LiveClient | PromiseLike, +): (url: string) => Promise { + const standing = new Map; holders: number }>(); + const forget = (url: string, shared: { readonly client: Promise }) => { + 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 {