From 9385d33cae699c03ad2e32bdc04fe7b4e9c00f32 Mon Sep 17 00:00:00 2001 From: Charles Pan <30245070+charleslpan@users.noreply.github.com> Date: Mon, 19 Oct 2026 20:28:35 +0000 Subject: [PATCH 1/3] feat(service,host): poll where Neon Realtime is not configured A deployment without NEON_REALTIME_URL and NEON_REALTIME_SECRET (a Preview, local dev) now polls behind the same interfaces rather than hearing nothing: the service's LiveWake wakes every held read about once a second, its sealer reads each live statement on the database client and answers the rows inline at LIVE_POLL_URL, and the host's client at that address hands those rows on, sealing again every three seconds and skipping a poll whose rows did not change. Co-authored-by: luke-agent[bot] <339870089+luke-agent[bot]@users.noreply.github.com> --- ...ive-wire-accountInboxLiveAnswerSchema.json | 21 ++ .../live-wire-scopeInboxLiveAnswerSchema.json | 3 + .../live-wire-sealedLiveQuerySchema.json | 3 + packages/features/src/live-wire.ts | 15 ++ packages/host/src/live-follow.test.ts | 67 +++++++ packages/host/src/live-follow.ts | 68 ++++++- packages/service/src/http/held-read.ts | 7 +- packages/service/src/live/live-app.test.ts | 66 ++++++- packages/service/src/live/live-app.ts | 17 +- packages/service/src/live/live-wake.test.ts | 18 +- packages/service/src/live/live-wake.ts | 37 ++-- packages/service/src/live/realtime.test.ts | 179 ++++++++++-------- packages/service/src/live/realtime.ts | 52 ++++- packages/service/src/runtime.ts | 13 +- 14 files changed, 431 insertions(+), 135 deletions(-) create mode 100644 packages/host/src/live-follow.test.ts diff --git a/packages/features/fixtures/json-schema/live-wire-accountInboxLiveAnswerSchema.json b/packages/features/fixtures/json-schema/live-wire-accountInboxLiveAnswerSchema.json index 4e0ea08ab6..272df686ec 100644 --- a/packages/features/fixtures/json-schema/live-wire-accountInboxLiveAnswerSchema.json +++ b/packages/features/fixtures/json-schema/live-wire-accountInboxLiveAnswerSchema.json @@ -19,6 +19,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ @@ -40,6 +43,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ @@ -61,6 +67,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ @@ -82,6 +91,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ @@ -103,6 +115,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ @@ -124,6 +139,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ @@ -145,6 +163,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ diff --git a/packages/features/fixtures/json-schema/live-wire-scopeInboxLiveAnswerSchema.json b/packages/features/fixtures/json-schema/live-wire-scopeInboxLiveAnswerSchema.json index 4b7fd8ee50..94da7cdb4a 100644 --- a/packages/features/fixtures/json-schema/live-wire-scopeInboxLiveAnswerSchema.json +++ b/packages/features/fixtures/json-schema/live-wire-scopeInboxLiveAnswerSchema.json @@ -19,6 +19,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ diff --git a/packages/features/fixtures/json-schema/live-wire-sealedLiveQuerySchema.json b/packages/features/fixtures/json-schema/live-wire-sealedLiveQuerySchema.json index 8e99553fce..1929f06b03 100644 --- a/packages/features/fixtures/json-schema/live-wire-sealedLiveQuerySchema.json +++ b/packages/features/fixtures/json-schema/live-wire-sealedLiveQuerySchema.json @@ -10,6 +10,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ diff --git a/packages/features/src/live-wire.ts b/packages/features/src/live-wire.ts index aea26f44a4..2993e1b297 100644 --- a/packages/features/src/live-wire.ts +++ b/packages/features/src/live-wire.ts @@ -44,14 +44,29 @@ import { countedNumber } from "./service-wire.js"; * statement to select every column under the key the wire reads it by and * each instant as epoch milliseconds. Realtime keeps no order, so the * reader orders them as the held read does. + * + * A deployment without Realtime (a Preview, or local dev) answers the same + * shape at `LIVE_POLL_URL` instead, each statement already run and its + * `rows` inline, and the desktop polls by sealing again. */ +/** The address a deployment without Realtime answers, which tells the desktop to poll rather than subscribe. */ +export const LIVE_POLL_URL = "luke-poll:"; + /** One sealed statement as the service answers it, which Realtime's client subscribes to and renews with. */ export const sealedLiveQuerySchema = EffectSchema.Struct({ capability: EffectSchema.String, queryFingerprint: EffectSchema.String, /** When the seal expires, in epoch milliseconds; the reader renews it before. */ expiresAt: countedNumber, + /** + * The statement's rows as they stand, answered only at `LIVE_POLL_URL`, + * where nothing subscribes. They travel as JSON text, since a row has no + * wire form until its reader decodes it. + */ + rows: EffectSchema.optionalKey( + EffectSchema.fromJsonString(EffectSchema.Array(EffectSchema.Unknown)), + ), }); export type SealedLiveQueryWire = typeof sealedLiveQuerySchema.Type; diff --git a/packages/host/src/live-follow.test.ts b/packages/host/src/live-follow.test.ts new file mode 100644 index 0000000000..f0d308b2fa --- /dev/null +++ b/packages/host/src/live-follow.test.ts @@ -0,0 +1,67 @@ +import assert from "node:assert/strict"; +import { it } from "@effect/vitest"; +import { LIVE_POLL_URL, type SealedLiveQueryWire } from "@luke/features/live-wire"; +import { Effect, Fiber, Stream } from "effect"; +import { TestClock } from "effect/testing"; +import { liveRows, openRealtimeClient, type SealedRead } from "./live-follow.js"; + +/** + * A read the service answered at the poll address, where it has no Realtime: + * the rows its seal carried are the read's first rows, and each seal after, + * a few seconds apart, hands on its rows only where they changed. + * + * Rows are synthetic; the service is a script of the seals it answers. + */ + +/** A seal as the polling service answers one: already expired, its rows inline. */ +const polled = (rows: readonly unknown[]): SealedRead<"items"> => ({ + url: LIVE_POLL_URL, + queries: { + items: { + capability: LIVE_POLL_URL, + queryFingerprint: LIVE_POLL_URL, + expiresAt: 0, + rows, + } satisfies SealedLiveQueryWire, + }, +}); + +/** Every step the read has to take, taken. */ +const settled = Effect.repeat(Effect.yieldNow, { times: 50 }); + +it.effect( + "a polled read hands on its first rows, then each seal's rows a few seconds apart, skipping a seal whose rows did not change", + () => + Effect.gen(function* () { + const answers = [[{ id: "a" }], [{ id: "a" }, { id: "b" }], [{ id: "b" }]]; + const heard: unknown[][] = []; + const reading = yield* Effect.forkChild( + Stream.runForEach( + liveRows( + { + openLive: openRealtimeClient, + reseal: () => Effect.sync(() => polled(answers.shift() ?? [])), + now: () => 0, + }, + polled([{ id: "a" }]), + ), + (rows) => Effect.sync(() => void heard.push([...rows.items])), + ), + ); + yield* settled; + assert.deepEqual(heard, [[{ id: "a" }]]); + + yield* TestClock.adjust("3 seconds"); + yield* settled; + assert.deepEqual(heard, [[{ id: "a" }]], "an unchanged poll is no change"); + + yield* TestClock.adjust("3 seconds"); + yield* settled; + assert.deepEqual(heard, [[{ id: "a" }], [{ id: "a" }, { id: "b" }]]); + + yield* TestClock.adjust("3 seconds"); + yield* settled; + assert.deepEqual(heard.at(-1), [{ id: "b" }]); + yield* Fiber.interrupt(reading); + }), +); diff --git a/packages/host/src/live-follow.ts b/packages/host/src/live-follow.ts index 69faaaaba4..723a622fb3 100644 --- a/packages/host/src/live-follow.ts +++ b/packages/host/src/live-follow.ts @@ -1,4 +1,4 @@ -import type { SealedLiveQueryWire } from "@luke/features/live-wire"; +import { LIVE_POLL_URL, type SealedLiveQueryWire } from "@luke/features/live-wire"; import { Cause, Data, Duration, Effect, Queue, Stream } from "effect"; /** @@ -12,12 +12,19 @@ import { Cause, Data, Duration, Effect, Queue, Stream } from "effect"; * later, as Neon's own bindings renew ("Renew the sealed query"). The sealed * statements stay here, where the account's own bearer is; only what the * caller builds from the rows leaves. + * + * Where the service has no Realtime it answers `LIVE_POLL_URL` with each + * statement's rows inline, and the read polls instead: the same client + * shape hands those rows on (`pollingClient`), and the read is sealed again + * every `LIVE_POLL_EVERY` rather than before an expiry. */ /** How long before a sealed statement expires it is renewed, as Neon's own bindings renew it. */ const RENEW_EARLY = Duration.seconds(10); /** How soon a renewal whose fetch failed is asked again, as Neon's own bindings ask. */ const RENEW_RETRY = Duration.seconds(1); +/** How often a read the service answered at `LIVE_POLL_URL` is sealed again, which is how often it polls. */ +const LIVE_POLL_EVERY = Duration.seconds(3); /** A subscription that stopped for good: Realtime's client did not load, Realtime said so, a renewal was refused, or a row did not decode. */ export class LiveBroken extends Data.TaggedError("LiveBroken")<{ readonly reason: string }> {} @@ -40,13 +47,53 @@ export interface LiveClient { close(): void; } +/** One snapshot of a part, as Realtime's client reports it. */ +type LiveSnapshot = Parameters[0]>[0]; + /** - * Realtime's own client, at the address the service named, loaded only when + * The client where the service polls: each part's rows are the ones its + * seal carried, handed on as each listener joins and again as each renewal + * brings rows that differ from the last, so an unchanged poll is no change. + */ +function pollingClient(): LiveClient { + return { + subscribe: (query) => { + let rows = query.rows; + let said = JSON.stringify(rows); + const listeners = new Set<(snapshot: LiveSnapshot) => void>(); + return { + onChange: (listener) => { + listeners.add(listener); + listener({ status: "ready", data: rows }); + return () => void listeners.delete(listener); + }, + renew: (renewed) => { + const saying = JSON.stringify(renewed.rows); + if (saying !== said) { + rows = renewed.rows; + said = saying; + for (const listener of listeners) listener({ status: "ready", data: rows }); + } + return Promise.resolve(); + }, + unsubscribe: () => listeners.clear(), + }; + }, + close: () => undefined, + }; +} + +/** + * 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. + * web app's first load; or the polling one where it named `LIVE_POLL_URL`. */ export const openRealtimeClient = (url: string): Promise => - import("@neon/realtime/client").then(({ createRealtimeClient }) => createRealtimeClient({ url })); + url === LIVE_POLL_URL + ? Promise.resolve(pollingClient()) + : 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 { @@ -70,6 +117,12 @@ function earliestExpiry(read: SealedRead): number { ); } +/** How long until the read is sealed again: the poll's own beat where it polls, and otherwise just before the earliest seal expires. */ +function untilRenewal(read: SealedRead, now: number): Duration.Duration { + if (read.url === LIVE_POLL_URL) return LIVE_POLL_EVERY; + return Duration.millis(Math.max(0, earliestExpiry(read) - Duration.toMillis(RENEW_EARLY) - now)); +} + /** * Every part's rows, whole and undecoded, each time any part changes once * every part has its first result, for as long as the subscriptions stand. @@ -116,10 +169,9 @@ export function liveRows( for (const [, subscription] of standing) subscription.unsubscribe(); }), ); - let expiresAt = earliestExpiry(first); + let latest: SealedRead = first; const renewal = Effect.gen(function* () { - const wait = expiresAt - Duration.toMillis(RENEW_EARLY) - dependencies.now(); - yield* Effect.sleep(Duration.millis(Math.max(0, wait))); + yield* Effect.sleep(untilRenewal(latest, dependencies.now())); let next = yield* dependencies.reseal(); while ("failure" in next) { yield* Effect.sleep(RENEW_RETRY); @@ -135,7 +187,7 @@ export function liveRows( }), { discard: true }, ); - expiresAt = earliestExpiry(renewed); + latest = renewed; }); yield* Effect.forkScoped( Effect.forever(renewal).pipe( diff --git a/packages/service/src/http/held-read.ts b/packages/service/src/http/held-read.ts index 9d98aa6944..8338b53dea 100644 --- a/packages/service/src/http/held-read.ts +++ b/packages/service/src/http/held-read.ts @@ -22,7 +22,12 @@ import { LiveWake } from "../live/live-wake.js"; * way (`lookedUntil`). */ -/** How often a watched wait looks though nothing told it to: what a change the watch missed waits at most. */ +/** + * How often a watched wait looks though nothing told it to: what a change the + * watch missed waits at most. Note that it is the only recovery from a + * Realtime subscription that was refused or dropped, which the wake never + * subscribes again, so it stays even where Realtime is configured. + */ const LIVE_SAFETY_POLL = Duration.seconds(5); /** How many characters of the digest a cursor carries: far more than any two readings of one plan's rows could share by chance. */ diff --git a/packages/service/src/live/live-app.test.ts b/packages/service/src/live/live-app.test.ts index 48b1f45d5c..f59050f218 100644 --- a/packages/service/src/live/live-app.test.ts +++ b/packages/service/src/live/live-app.test.ts @@ -6,20 +6,29 @@ import { Database } from "@luke/database/core/query"; import { createCodingAgent } from "@luke/database/models/coding-agent-store"; import { user } from "@luke/database/schemas/auth"; import { getTestDB, startTask } from "@luke/database/testing"; -import { accountInboxLiveAnswerSchema, scopeInboxLiveAnswerSchema } from "@luke/features/live-wire"; +import { INBOX_RESPONSE } from "@luke/features/inbox-wire"; +import { + accountInboxLiveAnswerSchema, + inboxOfLiveRows, + LIVE_POLL_URL, + scopeInboxLiveAnswerSchema, + scopeInboxLiveRowsSchema, +} from "@luke/features/live-wire"; import type { WireBoundaryInput } from "@luke/wire"; -import { Effect, Layer, Option, Schema } from "effect"; +import { Effect, Layer, Option, Result, Schema } from "effect"; import { HttpRouter } from "effect/http"; import { SqlClient } from "effect/sql"; import { HOSTED_API_ERROR, HOSTED_HTTP_STATUS } from "../http/http.js"; +import { postInboxItem, taskInboxItems } from "../inbox/inbox-store.js"; import { liveApp } from "./live-app.js"; -import { type LiveSealer, LiveSealing, SealRefused } from "./realtime.js"; +import { type LiveSealer, LiveSealing, pollSealer, SealRefused } from "./realtime.js"; /** * The reads the desktop follows over Neon Realtime, sealed by the route: * refused without a bearer, unavailable where the deployment has no * Realtime, and otherwise every part of the account inbox sealed for the - * bearer's own account, with the address to subscribe at. + * bearer's own account, with the address to subscribe at; where the + * deployment polls, each statement's rows are answered inline. * * Synthetic accounts and bearers throughout; the secret is a script that * seals a statement as the values it was bound to, so a test can read whose @@ -194,4 +203,53 @@ it.layer(getTestDB())("the live reads", (it) => { } }), ); + + it.effect( + "where the deployment polls, a plan's inbox is answered at the poll address with its rows inline, which read as the held read reads the inbox", + () => + Effect.gen(function* () { + const owner = yield* openUser; + const task = yield* startTask(owner, { name: "Teammate invitations" }); + const started = yield* createCodingAgent(owner, { + taskId: task.id, + idempotencyKey: randomUUID(), + model: "anthropic/claude-opus-5.5", + effort: "high", + repository: "acme/relay", + planSnapshot: "# Teammate invitations\n", + prompt: null, + }); + assert.ok(Option.isSome(started)); + const posted = yield* postInboxItem( + { userId: owner, taskId: task.id, agentId: started.value.agent.id }, + { + key: "branch-protection", + title: "Require a review on main", + response: { type: INBOX_RESPONSE.TODO }, + needsReply: false, + }, + ); + assert.ok(Result.isSuccess(posted)); + + const polled = yield* ask({ + bearer: owner, + path: `/api/live/task-inbox?id=${task.id}`, + sealing: Option.some(pollSealer(yield* SqlClient.SqlClient)), + }); + + assert.equal(polled.status, HOSTED_HTTP_STATUS.OK); + const answer = Schema.decodeUnknownSync(scopeInboxLiveAnswerSchema)(polled.body); + assert.equal(answer.url, LIVE_POLL_URL); + const rows = Schema.decodeUnknownSync(scopeInboxLiveRowsSchema)({ + items: answer.queries.items.rows, + messages: answer.queries.messages.rows, + files: answer.queries.files.rows, + }); + assert.deepEqual(inboxOfLiveRows(rows), yield* taskInboxItems(owner, task.id)); + assert.deepEqual( + inboxOfLiveRows(rows).map((item) => item.title), + ["Require a review on main"], + ); + }), + ); }); diff --git a/packages/service/src/live/live-app.ts b/packages/service/src/live/live-app.ts index 2155b5b9a0..807fcae587 100644 --- a/packages/service/src/live/live-app.ts +++ b/packages/service/src/live/live-app.ts @@ -2,7 +2,8 @@ import type { DatabaseServices } from "@luke/database/core/query"; import { ownFailedRunsLiveStatement } from "@luke/database/models/automation-store"; import { readTask } from "@luke/database/models/task-store"; import { HOSTED_SERVICE_PATH } from "@luke/features"; -import { Effect, Layer, Option } from "effect"; +import { accountInboxLiveAnswerSchema, scopeInboxLiveAnswerSchema } from "@luke/features/live-wire"; +import { Effect, Layer, Option, Schema } from "effect"; import { HttpRouter, HttpServerRequest, type HttpServerResponse } from "effect/http"; import { reachedAgent, resolvedUserId, uuidOf } from "../agents/agent-reach.js"; import { HOSTED_HTTP_STATUS } from "../http/http.js"; @@ -42,7 +43,9 @@ import { type LiveSealer, LiveSealing } from "./realtime.js"; * latest settled ones as one statement, their threads and files as two * more, after the plan or the agent is read as the account, * so one it does not reach is not found, as its held read answers. - * Nothing else here reads a row: the store is read by Realtime. + * Nothing else here reads a row: the store is read by Realtime, or, where + * the deployment has none, by the sealer itself (`pollSealer`), whose rows + * are answered inline, behind the same bearer. */ const HTTP_METHOD = { GET: "GET" } as const; @@ -88,7 +91,10 @@ const accountInboxLiveEndpoint = /* @__PURE__ */ Effect.fn("web/accountInboxLive chatAsks: seal(statements.value.chatAsks), chatNotices: seal(statements.value.chatNotices), }).pipe(Effect.mapError(() => HOSTED_REFUSAL.UNAVAILABLE)); - return hostedJsonResponse(HOSTED_HTTP_STATUS.OK, { url, queries }); + const answer = yield* Effect.orDie( + Schema.encodeEffect(accountInboxLiveAnswerSchema)({ url, queries }), + ); + return hostedJsonResponse(HOSTED_HTTP_STATUS.OK, answer); }, ); @@ -107,7 +113,10 @@ const sealedScope = ( messages: seal(statements.value.messages), files: seal(statements.value.files), }).pipe(Effect.mapError(() => HOSTED_REFUSAL.UNAVAILABLE)); - return hostedJsonResponse(HOSTED_HTTP_STATUS.OK, { url, queries }); + const answer = yield* Effect.orDie( + Schema.encodeEffect(scopeInboxLiveAnswerSchema)({ url, queries }), + ); + return hostedJsonResponse(HOSTED_HTTP_STATUS.OK, answer); }); /** The GET a scope's endpoint takes, as the web request it is. */ diff --git a/packages/service/src/live/live-wake.test.ts b/packages/service/src/live/live-wake.test.ts index c61e83a84d..1ae349bb89 100644 --- a/packages/service/src/live/live-wake.test.ts +++ b/packages/service/src/live/live-wake.test.ts @@ -1,6 +1,6 @@ import assert from "node:assert/strict"; import { it } from "@effect/vitest"; -import { Effect, Exit, Fiber, Option } from "effect"; +import { Effect, Fiber, Option } from "effect"; import { TestClock } from "effect/testing"; import { lookedUntil } from "../http/held-read.js"; import { @@ -13,8 +13,8 @@ import { /** * A wait on a row, woken by Neon Realtime: it looks again as soon as the - * rows change, and on a safety beat beside; with no wake provided it fails - * rather than polling; and the wake over Realtime's direct server subscribes + * rows change, and on a safety beat beside; with no wake provided it polls + * about once a second; and the wake over Realtime's direct server subscribes * to each statement for the length of the wait, subscribes again where one * moves to error, on a new server where the connection failed, and lets go * after. @@ -57,12 +57,16 @@ function scriptedWatch() { }; } -it.effect("with no wake provided a wait fails rather than looking on a clock", () => +it.effect("where Realtime is not configured, a wait looks again about once a second", () => Effect.gen(function* () { const script = scriptedLook(); - const exit = yield* Effect.exit(lookedUntil(script.look, "10 seconds", [STATEMENT])); - assert.ok(Exit.isFailure(exit)); - assert.equal(script.looks(), 0); + const waiting = yield* Effect.forkChild(lookedUntil(script.look, "10 seconds", [STATEMENT])); + yield* settled; + assert.equal(script.looks(), 1); + script.find("settled"); + yield* TestClock.adjust("1 second"); + assert.deepEqual(yield* Fiber.join(waiting), Option.some("settled")); + assert.equal(script.looks(), 2, "it looked on the poll, well before the safety beat"); }), ); diff --git a/packages/service/src/live/live-wake.ts b/packages/service/src/live/live-wake.ts index 731a17f295..d05372b8f2 100644 --- a/packages/service/src/live/live-wake.ts +++ b/packages/service/src/live/live-wake.ts @@ -1,6 +1,6 @@ import type { LiveStatement } from "@luke/database/core/live-statement"; import { createRealtime, type RawSqlQuery, rawSql } from "@neon/realtime/server"; -import { Config, Context, Effect, Layer, Option, Redacted, type Scope } from "effect"; +import { Config, Context, Duration, Effect, Layer, Option, Redacted, type Scope } from "effect"; import { databaseUrl } from "../db/index.js"; import { databaseNameOf, REALTIME_ENVIRONMENT } from "./realtime.js"; @@ -12,6 +12,8 @@ import { databaseNameOf, REALTIME_ENVIRONMENT } from "./realtime.js"; * statement, so what the desktop is answered is what the store says and * nothing the subscription carried. A watch that misses a change, or one * Realtime refused or stopped, costs the read a look on its safety beat. + * Where Realtime is not configured, every watch wakes on a short clock + * instead, so a held read polls. */ /** What a held read waits on, for the length of one hold. */ @@ -30,19 +32,25 @@ export interface LiveWakeShape { /** The watch that hears nothing: a wait on no statements, or one Realtime refused, looks on its safety beat alone. */ const QUIET: LiveWatch = { changed: Effect.never }; -/** The wake where Realtime is not configured, which refuses every watch loudly rather than looking on a clock. */ -const UNCONFIGURED: LiveWakeShape = { - watch: () => Effect.die(new Error("live-wake: Neon Realtime is not configured")), +/** How often a wait looks where Realtime is not configured: each hold's own look, about once a second. */ +const LIVE_WAKE_POLL = Duration.seconds(1); + +/** + * The wake where Realtime is not configured (a Preview, or local dev): every + * watch says the rows changed once a second, so a held read polls its own + * statement rather than hearing nothing. + */ +const pollingWake: LiveWakeShape = { + watch: () => Effect.succeed({ changed: Effect.sleep(LIVE_WAKE_POLL) }), }; /** * The wake a held read waits on: Realtime, where the web runtime provides - * `liveWakeFromEnvironment`, and a scripted one in a test. With none - * provided, every watch dies, so a missing wake is heard rather than - * polled around. + * `liveWakeFromEnvironment` and the deployment configures it, and a scripted + * one in a test. With none provided, every watch polls. */ export const LiveWake = Context.Reference("@luke/service/LiveWake", { - defaultValue: () => UNCONFIGURED, + defaultValue: () => pollingWake, }); /** One statement's subscription, the slice of Realtime's raw subscription a watch reads. */ @@ -213,8 +221,8 @@ export const realtimeServers = ( * The wake as the deployment's environment configures it: Realtime, over one * direct server and one socket the instance's every held read shares, closed * with the runtime and replaced where its connection failed. Where a variable or the database name is missing, or - * Realtime refuses its configuration, it is said in the log and every watch - * dies, since a held read has nothing else to wait on. + * Realtime refuses its configuration, every watch polls instead, said once in + * the log where Realtime refused. */ export const liveWakeFromEnvironment = Layer.effect( LiveWake, @@ -227,16 +235,13 @@ export const liveWakeFromEnvironment = Layer.effect( const url = Option.getOrUndefined(read.url)?.trim(); const secret = Option.getOrUndefined(read.secret); const database = Option.flatMap(read.database, databaseNameOf); - if (!url || secret === undefined || Option.isNone(database)) { - yield* Effect.logError("live-wake: Neon Realtime is not configured; held reads fail"); - return UNCONFIGURED; - } + if (!url || secret === undefined || Option.isNone(database)) return pollingWake; const servers = yield* realtimeServers(() => createRealtime({ url, secret: Redacted.value(secret).trim(), db: database.value }), ).pipe(Effect.option); if (Option.isNone(servers)) { - yield* Effect.logError("live-wake: Realtime refused its configuration; held reads fail"); - return UNCONFIGURED; + yield* Effect.logError("live-wake: Realtime refused its configuration; held reads poll"); + return pollingWake; } return realtimeWake(servers.value); }), diff --git a/packages/service/src/live/realtime.test.ts b/packages/service/src/live/realtime.test.ts index 63382c8978..46f88808bd 100644 --- a/packages/service/src/live/realtime.test.ts +++ b/packages/service/src/live/realtime.test.ts @@ -3,13 +3,16 @@ import { it } from "@effect/vitest"; import { liveSelection, liveStatementOf } from "@luke/database/core/live-statement"; import { queryBuilder } from "@luke/database/core/query"; import { inboxItem } from "@luke/database/schemas/inbox"; +import { getTestDB } from "@luke/database/testing"; +import { LIVE_POLL_URL } from "@luke/features/live-wire"; import { eq } from "drizzle-orm"; import { ConfigProvider, Effect, Layer, Option, Result } from "effect"; +import { SqlClient } from "effect/sql"; import { LiveSealing, liveSealingFromEnvironment, REALTIME_ENVIRONMENT } from "./realtime.js"; /** - * The deployment's Neon Realtime as its environment configures it: off - * without both of its variables or with a secret Realtime refuses, and on + * The deployment's Neon Realtime as its environment configures it: polled + * without both of its variables, off with a secret Realtime refuses, and on * otherwise, sealing a statement locally under Realtime's own SDK for the * database the connection string names, with the address the desktop * subscribes at. @@ -23,91 +26,107 @@ const SYNTHETIC_SECRET = `nrt_live_1${Buffer.alloc(32).toString("base64url")}`; const REALTIME_URL = "wss://realtime.example/v1"; const DATABASE_URL = "postgresql://luke:synthetic@db.example/neondb?sslmode=require"; -/** The sealing the layer builds over an environment holding just these variables. */ +/** The sealing the layer builds over an environment holding just these variables, on the test's database. */ const sealingOver = (environment: Readonly>) => Effect.gen(function* () { - return yield* LiveSealing; - }).pipe( - Effect.provide( + const sql = yield* SqlClient.SqlClient; + return yield* Effect.provide( + Effect.gen(function* () { + return yield* LiveSealing; + }), liveSealingFromEnvironment.pipe( - Layer.provide(ConfigProvider.layer(ConfigProvider.fromUnknown(environment))), + Layer.provide( + Layer.merge( + ConfigProvider.layer(ConfigProvider.fromUnknown(environment)), + Layer.succeed(SqlClient.SqlClient, sql), + ), + ), ), - ), + ); + }); + +it.layer(getTestDB())("the deployment's Realtime", (it) => { + it.effect( + "without both of its variables Realtime is polled, each statement read and its rows answered; with a secret Realtime refuses it is off", + () => + Effect.gen(function* () { + for (const environment of [ + { DATABASE_URL }, + { DATABASE_URL, [REALTIME_ENVIRONMENT.URL]: REALTIME_URL }, + { + DATABASE_URL, + [REALTIME_ENVIRONMENT.URL]: REALTIME_URL, + [REALTIME_ENVIRONMENT.SECRET]: " ", + }, + ]) { + const sealing = yield* sealingOver(environment); + assert.ok(Option.isSome(sealing)); + assert.equal(sealing.value.url, LIVE_POLL_URL); + const sealed = yield* sealing.value.seal({ + sql: 'select $1::text as "who", 2::int as "count"', + params: ["user-dana"], + }); + assert.deepEqual(sealed.rows, [{ who: "user-dana", count: 2 }]); + } + assert.ok( + Option.isNone( + yield* sealingOver({ + DATABASE_URL, + [REALTIME_ENVIRONMENT.URL]: REALTIME_URL, + [REALTIME_ENVIRONMENT.SECRET]: "not a secret", + }), + ), + ); + }), ); -it.effect("without both of its variables, or with a secret Realtime refuses, Realtime is off", () => - Effect.gen(function* () { - assert.ok(Option.isNone(yield* sealingOver({ DATABASE_URL }))); - assert.ok( - Option.isNone(yield* sealingOver({ DATABASE_URL, [REALTIME_ENVIRONMENT.URL]: REALTIME_URL })), - ); - assert.ok( - Option.isNone( - yield* sealingOver({ + it.effect( + "with both, a statement is sealed locally, its words unreadable in the seal, with the address the desktop subscribes at", + () => + Effect.gen(function* () { + const sealing = yield* sealingOver({ DATABASE_URL, - [REALTIME_ENVIRONMENT.URL]: REALTIME_URL, - [REALTIME_ENVIRONMENT.SECRET]: " ", - }), - ), - ); - assert.ok( - Option.isNone( - yield* sealingOver({ + [REALTIME_ENVIRONMENT.URL]: ` ${REALTIME_URL} `, + [REALTIME_ENVIRONMENT.SECRET]: SYNTHETIC_SECRET, + }); + assert.ok(Option.isSome(sealing)); + assert.equal(sealing.value.url, REALTIME_URL); + const statement = { + sql: 'select "inbox_item"."id" as "id" from "inbox_item" where "inbox_item"."user_id" = $1', + params: ["user-dana"], + }; + const sealed = yield* sealing.value.seal(statement); + const again = yield* sealing.value.seal(statement); + assert.ok(!sealed.capability.includes("user-dana")); + assert.ok(!sealed.capability.includes("inbox_item")); + assert.equal(sealed.queryFingerprint, again.queryFingerprint); + const other = yield* sealing.value.seal({ ...statement, params: ["user-eli"] }); + assert.notEqual(other.queryFingerprint, sealed.queryFingerprint); + }), + ); + + it.effect( + "a Drizzle select is sealed through Realtime's Drizzle adapter, which refuses one whose key is not the name Postgres answers it under", + () => + Effect.gen(function* () { + const sealing = yield* sealingOver({ DATABASE_URL, [REALTIME_ENVIRONMENT.URL]: REALTIME_URL, - [REALTIME_ENVIRONMENT.SECRET]: "not a secret", - }), - ), - ); - }), -); - -it.effect( - "with both, a statement is sealed locally, its words unreadable in the seal, with the address the desktop subscribes at", - () => - Effect.gen(function* () { - const sealing = yield* sealingOver({ - DATABASE_URL, - [REALTIME_ENVIRONMENT.URL]: ` ${REALTIME_URL} `, - [REALTIME_ENVIRONMENT.SECRET]: SYNTHETIC_SECRET, - }); - assert.ok(Option.isSome(sealing)); - assert.equal(sealing.value.url, REALTIME_URL); - const statement = { - sql: 'select "inbox_item"."id" as "id" from "inbox_item" where "inbox_item"."user_id" = $1', - params: ["user-dana"], - }; - const sealed = yield* sealing.value.seal(statement); - const again = yield* sealing.value.seal(statement); - assert.ok(!sealed.capability.includes("user-dana")); - assert.ok(!sealed.capability.includes("inbox_item")); - assert.equal(sealed.queryFingerprint, again.queryFingerprint); - const other = yield* sealing.value.seal({ ...statement, params: ["user-eli"] }); - assert.notEqual(other.queryFingerprint, sealed.queryFingerprint); - }), -); - -it.effect( - "a Drizzle select is sealed through Realtime's Drizzle adapter, which refuses one whose key is not the name Postgres answers it under", - () => - Effect.gen(function* () { - const sealing = yield* sealingOver({ - DATABASE_URL, - [REALTIME_ENVIRONMENT.URL]: REALTIME_URL, - [REALTIME_ENVIRONMENT.SECRET]: SYNTHETIC_SECRET, - }); - assert.ok(Option.isSome(sealing)); - const select = (columns: Parameters[0]) => - Option.getOrThrow( - liveStatementOf( - queryBuilder.select(columns).from(inboxItem).where(eq(inboxItem.userId, "user-dana")), - ), + [REALTIME_ENVIRONMENT.SECRET]: SYNTHETIC_SECRET, + }); + assert.ok(Option.isSome(sealing)); + const select = (columns: Parameters[0]) => + Option.getOrThrow( + liveStatementOf( + queryBuilder.select(columns).from(inboxItem).where(eq(inboxItem.userId, "user-dana")), + ), + ); + const aliased = yield* Effect.result( + sealing.value.seal(select(liveSelection({ itemId: inboxItem.id }))), ); - const aliased = yield* Effect.result( - sealing.value.seal(select(liveSelection({ itemId: inboxItem.id }))), - ); - assert.ok(Result.isSuccess(aliased)); - const renamed = yield* Effect.result(sealing.value.seal(select({ itemId: inboxItem.id }))); - assert.ok(Result.isFailure(renamed)); - }), -); + assert.ok(Result.isSuccess(aliased)); + const renamed = yield* Effect.result(sealing.value.seal(select({ itemId: inboxItem.id }))); + assert.ok(Result.isFailure(renamed)); + }), + ); +}); diff --git a/packages/service/src/live/realtime.ts b/packages/service/src/live/realtime.ts index e32236063b..c32a2f9cee 100644 --- a/packages/service/src/live/realtime.ts +++ b/packages/service/src/live/realtime.ts @@ -1,8 +1,9 @@ import type { LiveStatement } from "@luke/database/core/live-statement"; -import type { SealedLiveQueryWire } from "@luke/features/live-wire"; +import { LIVE_POLL_URL, type SealedLiveQueryWire } from "@luke/features/live-wire"; import { createRealtime, rawSql } from "@neon/realtime/server"; import { drizzleAdapter } from "@neon/realtime-drizzle"; -import { Config, Context, Data, Effect, Layer, Option, Redacted } from "effect"; +import { Clock, Config, Context, Data, Effect, Layer, Option, Redacted } from "effect"; +import { SqlClient } from "effect/sql"; import { databaseUrl } from "../db/index.js"; /** @@ -16,6 +17,11 @@ import { databaseUrl } from "../db/index.js"; * desktop can subscribe to it and can neither read nor change it. The * secret never leaves this service. A sealed read is itself a short-lived * bearer, so it travels in an answer body alone, never a URL or a log. + * + * A deployment without Realtime (a Preview, or local dev) seals by reading + * instead (`pollSealer`): the statement is run now and its rows answered + * inline at `LIVE_POLL_URL`, behind the same bearer, and the desktop polls + * by sealing again. */ /** The variables a deployment turns Realtime on with, both from its branch's Realtime page in Neon's console. */ @@ -35,9 +41,10 @@ export interface LiveSealer { } /** - * The deployment's Realtime, or none where it is misconfigured, which every - * live endpoint answers as unavailable. None by default, so a test that - * follows nothing live need provide nothing. + * The deployment's Realtime, or its poll sealer where Realtime is not + * configured, or none where Realtime refused its secret, which every live + * endpoint answers as unavailable. None by default, so a test that follows + * nothing live need provide nothing. */ export const LiveSealing = Context.Reference>( "@luke/service/LiveSealing", @@ -90,11 +97,36 @@ function liveSealer(options: { ); } +/** + * The sealer where Realtime is not configured: each statement is run on the + * service's own database client and answered with its rows, already + * expired, since nothing renews it but a fresh read. Note that no secret + * stands here, so the capability and fingerprint say only where to poll. + */ +export function pollSealer(sql: SqlClient.SqlClient): LiveSealer { + return { + url: LIVE_POLL_URL, + seal: (statement) => + Effect.all({ + rows: sql.unsafe(statement.sql, statement.params), + now: Clock.currentTimeMillis, + }).pipe( + Effect.map(({ rows, now }) => ({ + capability: LIVE_POLL_URL, + queryFingerprint: LIVE_POLL_URL, + expiresAt: now, + rows, + })), + Effect.mapError(() => new SealRefused({ reason: "the statement could not be read" })), + ), + }; +} + /** * Realtime as the deployment's environment configures it: on where both of - * its variables and the database name stand, off where any is missing or - * blank, so a deployment turns it on by setting them and off by removing - * them. A secret Realtime refuses leaves it off, said once in the log. + * its variables and the database name stand, polled where any is missing or + * blank, so a deployment turns it on by setting them and back to polling by + * removing them. A secret Realtime refuses leaves it off, said once in the log. */ export const liveSealingFromEnvironment = Layer.effect( LiveSealing, @@ -108,7 +140,9 @@ export const liveSealingFromEnvironment = Layer.effect( const secret = Option.getOrUndefined(read.secret); const database = Option.flatMap(read.database, databaseNameOf); const secretText = secret === undefined ? "" : Redacted.value(secret).trim(); - if (!url || secretText === "" || Option.isNone(database)) return Option.none(); + if (!url || secretText === "" || Option.isNone(database)) { + return Option.some(pollSealer(yield* SqlClient.SqlClient)); + } const sealer = yield* Effect.option( liveSealer({ url, secret: Redacted.make(secretText), database: database.value }), ); diff --git a/packages/service/src/runtime.ts b/packages/service/src/runtime.ts index ef5a77d1b2..ff67568b1d 100644 --- a/packages/service/src/runtime.ts +++ b/packages/service/src/runtime.ts @@ -64,11 +64,12 @@ const webLogger = Logger.layer([Logger.consoleJson, Logger.tracerLogger, sentryL * * `LiveSealing` is the one place Neon Realtime's address and secret are read, * so a live read is sealed under the same secret on every route - * (`live/realtime.ts`); without both it stands as none, and every live read - * answers unavailable. `LiveWake` is the same Realtime seen from the service's - * side: one socket per instance, which every held read and every wait on a - * row watches its rows over rather than looking on a clock - * (`live/live-wake.ts`); without Realtime each fails. + * (`live/realtime.ts`); without both it polls, reading each live statement + * on the same database client (which layers memoize, so it is the one pool) + * and answering its rows inline. `LiveWake` is the same Realtime seen from + * the service's side: one socket per instance, which every held read and + * every wait on a row watches its rows over rather than looking on a clock + * (`live/live-wake.ts`); without Realtime each looks on a short clock. * * `ModelCatalog` is here for the same reason: its read of AI Gateway's * public catalog is cached on the service, so one instance reads the catalog @@ -89,7 +90,7 @@ export const webServices = Layer.mergeAll( integrationConnectFromEnvironment, workspaceLinearConnectFromEnvironment, webLogger, - liveSealingFromEnvironment, + liveSealingFromEnvironment.pipe(Layer.provide(webSqlClient)), liveWakeFromEnvironment, modelCatalogLayer.pipe(Layer.provide(Layer.merge(FetchHttpClient.layer, hostedEnvironment))), productAnalyticsFromEnvironment.pipe( From 7c1266c9ff29e9b9c6d3b048aedb89b0f33aeaf2 Mon Sep 17 00:00:00 2001 From: Charles Pan <30245070+charleslpan@users.noreply.github.com> Date: Tue, 20 Oct 2026 01:42:11 +0000 Subject: [PATCH 2/3] test(features): the scope inbox's live answer records its polled rows Co-authored-by: luke-agent[bot] <339870089+luke-agent[bot]@users.noreply.github.com> --- .../json-schema/live-wire-scopeInboxLiveAnswerSchema.json | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/features/fixtures/json-schema/live-wire-scopeInboxLiveAnswerSchema.json b/packages/features/fixtures/json-schema/live-wire-scopeInboxLiveAnswerSchema.json index 94da7cdb4a..213aa6abc8 100644 --- a/packages/features/fixtures/json-schema/live-wire-scopeInboxLiveAnswerSchema.json +++ b/packages/features/fixtures/json-schema/live-wire-scopeInboxLiveAnswerSchema.json @@ -43,6 +43,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ @@ -64,6 +67,9 @@ "expiresAt": { "type": "number", "minimum": 0 + }, + "rows": { + "type": "string" } }, "required": [ From 8acbd2110ae44db7326115ef177f02e4ed6e608a Mon Sep 17 00:00:00 2001 From: Charles Pan <30245070+charleslpan@users.noreply.github.com> Date: Tue, 20 Oct 2026 02:44:41 +0000 Subject: [PATCH 3/3] feat(service): production without Neon Realtime says so once in the log as an error Previews and local dev poll quietly; production should never poll, so an instance there that finds the Realtime variables missing logs an error at start instead of polling silently. Co-authored-by: luke-agent[bot] <339870089+luke-agent[bot]@users.noreply.github.com> --- packages/service/src/live/realtime.test.ts | 23 +++++++++++++++++++++- packages/service/src/live/realtime.ts | 12 ++++++++++- 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/packages/service/src/live/realtime.test.ts b/packages/service/src/live/realtime.test.ts index 46f88808bd..5013ee8523 100644 --- a/packages/service/src/live/realtime.test.ts +++ b/packages/service/src/live/realtime.test.ts @@ -6,7 +6,7 @@ import { inboxItem } from "@luke/database/schemas/inbox"; import { getTestDB } from "@luke/database/testing"; import { LIVE_POLL_URL } from "@luke/features/live-wire"; import { eq } from "drizzle-orm"; -import { ConfigProvider, Effect, Layer, Option, Result } from "effect"; +import { ConfigProvider, Effect, Layer, Logger, Option, Result } from "effect"; import { SqlClient } from "effect/sql"; import { LiveSealing, liveSealingFromEnvironment, REALTIME_ENVIRONMENT } from "./realtime.js"; @@ -129,4 +129,25 @@ it.layer(getTestDB())("the deployment's Realtime", (it) => { assert.ok(Result.isFailure(renamed)); }), ); + + it.effect( + "a production deployment without Realtime says so once in the log as an error, and a preview polls quietly", + () => + Effect.gen(function* () { + const errors: string[] = []; + const recorder = Logger.make(({ logLevel, message }) => { + if (logLevel === "Error") errors.push(JSON.stringify(message)); + }); + const quietly = Effect.provide(Logger.layer([recorder])); + yield* sealingOver({ DATABASE_URL, VERCEL_ENV: "preview" }).pipe(quietly); + assert.deepEqual(errors, []); + const sealing = yield* sealingOver({ DATABASE_URL, VERCEL_ENV: "production" }).pipe( + quietly, + ); + assert.ok(Option.isSome(sealing)); + assert.equal(sealing.value.url, LIVE_POLL_URL); + assert.equal(errors.length, 1); + assert.match(errors[0] ?? "", /production has no Neon Realtime configured/); + }), + ); }); diff --git a/packages/service/src/live/realtime.ts b/packages/service/src/live/realtime.ts index c32a2f9cee..19c3fdc756 100644 --- a/packages/service/src/live/realtime.ts +++ b/packages/service/src/live/realtime.ts @@ -30,6 +30,9 @@ export const REALTIME_ENVIRONMENT = { SECRET: "NEON_REALTIME_SECRET", } as const; +/** What Vercel names the deployment a function runs in, production being the one that should never poll. */ +const VERCEL_DEPLOYMENT = { VARIABLE: "VERCEL_ENV", PRODUCTION: "production" } as const; + /** Why a statement could not be sealed: the secret refused it. */ export class SealRefused extends Data.TaggedError("SealRefused")<{ readonly reason: string }> {} @@ -125,7 +128,8 @@ export function pollSealer(sql: SqlClient.SqlClient): LiveSealer { /** * Realtime as the deployment's environment configures it: on where both of * its variables and the database name stand, polled where any is missing or - * blank, so a deployment turns it on by setting them and back to polling by + * blank (said once in the log as an error in production, which should never + * poll), so a deployment turns it on by setting them and back to polling by * removing them. A secret Realtime refuses leaves it off, said once in the log. */ export const liveSealingFromEnvironment = Layer.effect( @@ -135,12 +139,18 @@ export const liveSealingFromEnvironment = Layer.effect( url: Config.option(Config.String(REALTIME_ENVIRONMENT.URL)), secret: Config.option(Config.Redacted(REALTIME_ENVIRONMENT.SECRET)), database: Config.option(databaseUrl), + deployment: Config.option(Config.String(VERCEL_DEPLOYMENT.VARIABLE)), }); const url = Option.getOrUndefined(read.url)?.trim(); const secret = Option.getOrUndefined(read.secret); const database = Option.flatMap(read.database, databaseNameOf); const secretText = secret === undefined ? "" : Redacted.value(secret).trim(); if (!url || secretText === "" || Option.isNone(database)) { + if (Option.getOrUndefined(read.deployment)?.trim() === VERCEL_DEPLOYMENT.PRODUCTION) { + yield* Effect.logError( + "realtime: production has no Neon Realtime configured; live reads poll instead", + ); + } return Option.some(pollSealer(yield* SqlClient.SqlClient)); } const sealer = yield* Effect.option(