From a025c9dc48d6fefec58fb75c98207d069db0621e Mon Sep 17 00:00:00 2001 From: Charles Pan <30245070+charleslpan@users.noreply.github.com> Date: Wed, 21 Oct 2026 18:23:27 +0000 Subject: [PATCH] Share one Neon Realtime client per address across the host's live reads Every liveRows read at one Realtime URL now shares one client and its socket, as Neon's guide prescribes. A client whose subscription moves to error is replaced for the next read and closed once its last read lets go; a sign-out or account switch drops the shared client. Co-authored-by: luke-agent[bot] <339870089+luke-agent[bot]@users.noreply.github.com> --- packages/host/src/compose-account.ts | 6 +- packages/host/src/live-follow.test.ts | 97 +++++++++++++++++++++++++++ packages/host/src/live-follow.ts | 86 ++++++++++++++++++++++-- 3 files changed, 181 insertions(+), 8 deletions(-) diff --git a/packages/host/src/compose-account.ts b/packages/host/src/compose-account.ts index 85a5f46979..fcb14782c3 100644 --- a/packages/host/src/compose-account.ts +++ b/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)); diff --git a/packages/host/src/live-follow.test.ts b/packages/host/src/live-follow.test.ts index e5be7e67d6..59eea44826 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, 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]); +}); diff --git a/packages/host/src/live-follow.ts b/packages/host/src/live-follow.ts index 954e5dcd63..a8dcc00930 100644 --- a/packages/host/src/live-follow.ts +++ b/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 => - import("@neon/realtime/client").then(({ createRealtimeClient }) => createRealtimeClient({ url })); +export function shareLiveClients(open: (url: string) => LiveClient | PromiseLike): { + readonly open: (url: string) => Promise; + readonly drop: () => void; +} { + interface Shared { + readonly client: Promise; + holders: number; + } + const standing = new Map(); + 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 {