diff --git a/CHANGELOG.md b/CHANGELOG.md index bdd036d..72bcfc8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,20 @@ # Changelog +## 0.14.1 - 2026-08-23 + +- Add `solid-objects/signals`, live signals on actor references + ([#24](https://github.com/cardmagic/solid-objects-js/issues/24)). One + side-effect import enables `reference.live`: read-only signals on the + proposed standard JavaScript signals API (`signal-polyfill`, a new + optional peer dependency) that subscribe through an in-process + `runtime.realtime` session when first watched and unsubscribe after a + linger when the last watcher leaves. Value-broadcast observables feed + named signals; `live.snapshot` re-fetches the authorized snapshot on + each accepted envelope; `live.payloads.` carries personalized + payload projections under their independent revision fences; stale + revisions are fenced. Works in Node and + in the browser runtime. + ## 0.14.0 - 2026-08-23 - Add `solid-objects/database/shared-sqlite-wasm`, the transparent diff --git a/docs/api.md b/docs/api.md index a41cc23..443c0e1 100644 --- a/docs/api.md +++ b/docs/api.md @@ -29,6 +29,8 @@ generic signatures; this index explains the supported role of every export. only its name in the durable envelope when it changes. - `ObservableBroadcast`: the immutable marker type returned by either helper. - `VERSION`: running package version. +- `reference.live`: read-only live signals for an actor, enabled by the + `solid-objects/signals` entry point documented below. - `ActorClass`, `ActorReference`, `ActorMessageSender`, `ActorSnapshot`, `ActorOperationNames`, `ActorQueryNames`, `StagedOperations`, and `ScheduledOperations`: inferred actor-class and fluent-dispatch types. @@ -492,6 +494,67 @@ The transmit wire contract is shared with the Ruby gem ([solid-objects-ruby#49](https://github.com/cardmagic/solid-objects-ruby/pull/49)); `compatibility/transmit-envelopes.json` pins it in both repositories. +## `solid-objects/signals` + +Live signals: the read-side adapter on the proposed standard JavaScript +signals API. One side-effect import enables `reference.live`: + +```typescript +import "solid-objects/signals" + +const counter = Counter.ref("page-hits") +counter.live.count // a read-only signal of the broadcast observable +counter.live.snapshot // a read-only signal of the authorized snapshot +``` + +One property, three tenses: `snapshot.count` is one committed read, +`await counter.count` asks the actor now, and `counter.live.count` stays +current. From Lit, `watch(counter.live.count)` with the `SignalWatcher` +mixin renders it with no further glue; any consumer of the standard +signals API composes the same way. + +- `configureLiveSignals(options)`: tune the lifecycle. + `LiveSignalsConfiguration` carries `lingerMilliseconds` (default one + second): how long a signal with no watchers keeps its subscription + before the session closes; and `retryMilliseconds` (default one + second): how long a still-watched signal waits before it retries a + denied or failed subscription. +- `activeLiveSubscriptionCount()` and `liveEntryCount(runtime)`: open + sessions and live per-actor entries, for diagnostics and leak tests. + The cache holds entries through weak references: one canonical entry + per actor for as long as any proxy or signal for it is reachable, so + two references to one actor can never open competing subscriptions, + and an entry whose signals are all garbage-collected leaves the cache + with them. +- `ActorLiveSignals` and `LiveSignal`: the structural types on + `reference.live`. `LiveSignal` exposes only `get()`, so the package + types never require the optional peer; at runtime every signal is a + standard `Signal.Computed`, read-only by construction. + +Behavior: + +- A signal subscribes its actor through an in-process + `runtime.realtime` session when the first watcher arrives and closes + the session after the linger when the last watcher leaves. The + subscription authorizes through `authorizeSubscription` with an + undefined authorization context. +- Value-broadcast observables set their named signals from each + envelope. Invalidation-only observables stay `undefined` by design; + `live.snapshot` re-fetches the authorized snapshot (coalesced) on + every accepted envelope, so private-value flows read from there. +- Personalized payload projections arrive as + `live.payloads.` signals. A newly watched payload name re-sends + the subscription with the grown name list, and each payload keeps the + independent per-name revision fence the wire protocol gives it. + Payloads evaluate under the live session's authorization context. +- Envelopes apply only on a monotonic revision advance for the same + instance, the same fence the browser client uses. +- `snapshot` and `payloads` are reserved names on `live`; observables + with those names are shadowed. +- `signal-polyfill` is an optional peer dependency. Nothing loads it + until the `solid-objects/signals` entry is imported; `reference.live` + throws a pointer to that import otherwise. + ## `solid-objects/web` - `createDashboard(options)` creates an immutable `SolidObjectsDashboard` with diff --git a/docs/parity.md b/docs/parity.md index 3f6ffaa..0b3b92d 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -143,6 +143,17 @@ complete: replay deduplication, and recovery after an offline period, on SQLite, PostgreSQL, and MySQL. +## JavaScript-only: live signals + +`reference.live` (the `solid-objects/signals` entry) adapts committed +actor state to the proposed standard JavaScript signals API, so +signal-consuming renderers track actors with no manual registration. No +Ruby row exists because the slot it fills is already native in Rails: +the gem's Turbo and Action Cable component surface re-renders partials +from the same committed observables. Each runtime renders with its +ecosystem's primitive; the guarantee — views track committed state under +revision fencing and the same privacy model — is what parity preserves. + ## Shared capability: the transmit family The transmit family is the one part of the browser work that both runtimes diff --git a/package.json b/package.json index 3c31408..a48df50 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "solid-objects", - "version": "0.14.0", + "version": "0.14.1", "description": "Race-free realtime state per application identity, backed by your SQL database", "type": "module", "license": "MIT", @@ -78,6 +78,10 @@ "types": "./dist/browser/tab-host.d.ts", "import": "./dist/browser/tab-host.js" }, + "./signals": { + "types": "./dist/signals.d.ts", + "import": "./dist/signals.js" + }, "./transmit": { "types": "./dist/transmit.d.ts", "import": "./dist/transmit.js" @@ -121,12 +125,14 @@ "pg": "^8.23.0", "prettier": "^3.9.6", "redis": "^6.2.1", + "signal-polyfill": "^0.2.2", "typescript": "^5.9.0", "vitest": "^4.1.10", "ws": "^8.21.3" }, "peerDependencies": { "@sqlite.org/sqlite-wasm": ">=3.50.0-build1", + "signal-polyfill": ">=0.2.2", "mysql2": "^3.23.3", "pg": "^8.23.0", "redis": "^6.2.1" @@ -135,6 +141,9 @@ "@sqlite.org/sqlite-wasm": { "optional": true }, + "signal-polyfill": { + "optional": true + }, "mysql2": { "optional": true }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 86ccf64..74c1829 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -38,6 +38,9 @@ importers: redis: specifier: ^6.2.1 version: 6.2.1 + signal-polyfill: + specifier: ^0.2.2 + version: 0.2.2 typescript: specifier: ^5.9.0 version: 5.9.3 @@ -577,6 +580,9 @@ packages: siginfo@2.0.0: resolution: {integrity: sha512-ybx0WO1/8bSBLEWXZvEd7gMW3Sn3JFlW3TvX1nREbDLRNQNaeNN8WK0meBwPdAaOI7TtRRRJn/Es1zhrrCHu7g==} + signal-polyfill@0.2.2: + resolution: {integrity: sha512-p63Y4Er5/eMQ9RHg0M0Y64NlsQKpiu6MDdhBXpyywRuWiPywhJTpKJ1iB5K2hJEbFZ0BnDS7ZkJ+0AfTuL37Rg==} + source-map-js@1.2.1: resolution: {integrity: sha512-UXWMKhLOwVKb728IUtQPXxfYU+usdybtUrK/8uGE8CQMvrhOpwvzDBwj0QhSL7MQc7vIsISBG8VQ8+IDQxpfQA==} engines: {node: '>=0.10.0'} @@ -1160,6 +1166,8 @@ snapshots: siginfo@2.0.0: {} + signal-polyfill@0.2.2: {} + source-map-js@1.2.1: {} split2@4.2.0: {} diff --git a/scripts/check-browser-imports.mjs b/scripts/check-browser-imports.mjs index b5ac6ca..3c2a7f0 100644 --- a/scripts/check-browser-imports.mjs +++ b/scripts/check-browser-imports.mjs @@ -8,6 +8,7 @@ const browserSafeRoots = [ "src/browser/host.ts", "src/browser/index.ts", "src/browser/tab-host.ts", + "src/signals.ts", "src/transmit.ts", "src/context.ts", "src/database/deadline.ts", diff --git a/scripts/check-documentation.mjs b/scripts/check-documentation.mjs index eea8ff1..6345ba1 100644 --- a/scripts/check-documentation.mjs +++ b/scripts/check-documentation.mjs @@ -64,6 +64,7 @@ const entryPoints = [ "src/browser/host.ts", "src/browser/tab-host.ts", "src/transmit.ts", + "src/signals.ts", "src/web/index.ts", ] diff --git a/src/reference.ts b/src/reference.ts index 3830049..cae7d08 100644 --- a/src/reference.ts +++ b/src/reference.ts @@ -7,6 +7,7 @@ import type { DestroyOptions, InvocationOptions, JsonObject, + JsonValue, MessageStatus, SnapshotOptions, } from "./types.js" @@ -164,6 +165,27 @@ export class ActorMessageSenderCore { } } +export interface LiveSignal { + get(): Value +} + +export type ActorLiveSignals = { + readonly snapshot: LiveSignal | undefined> + readonly payloads: { + readonly [name: string]: LiveSignal | undefined> + } +} & { + readonly [name: string]: LiveSignal | undefined> +} + +export type LiveSignalsFactory = (reference: ActorReferenceCore) => object + +let liveSignalsFactory: LiveSignalsFactory | undefined + +export function installLiveSignals(factory: LiveSignalsFactory): void { + liveSignalsFactory = factory +} + export class ActorReferenceCore { readonly send: ActorMessageSender readonly runtime: SolidObjectsRuntime @@ -194,6 +216,18 @@ export class ActorReferenceCore { return createInvoker(this, options) } + #live: object | undefined + + get live(): ActorLiveSignals { + if (!liveSignalsFactory) { + throw new Error( + 'ref.live is inactive; import "solid-objects/signals" once to enable live signals', + ) + } + this.#live ??= liveSignalsFactory(this as unknown as ActorReferenceCore) + return this.#live as ActorLiveSignals + } + snapshot(options: SnapshotOptions = {}): Promise> { return this.runtime.snapshot(this, options) as Promise> } diff --git a/src/signals.ts b/src/signals.ts new file mode 100644 index 0000000..18fb318 --- /dev/null +++ b/src/signals.ts @@ -0,0 +1,323 @@ +import { Signal } from "signal-polyfill" +import type { RealtimeEnvelope } from "./browser/index.js" +import type { Actor } from "./actor.js" +import { + installLiveSignals, + type ActorReferenceCore, + type ActorSnapshot, + type LiveSignal, +} from "./reference.js" +import type { DeepReadonly, JsonValue } from "./types.js" + +export type { ActorLiveSignals, LiveSignal } from "./reference.js" + +export interface LiveSignalsConfiguration { + lingerMilliseconds?: number + retryMilliseconds?: number +} + +const SNAPSHOT_KEY = "snapshot" +const PAYLOADS_KEY = "payloads" + +let lingerMilliseconds = 1_000 +let retryMilliseconds = 1_000 +const openSessions = new Set() + +export function configureLiveSignals(configuration: LiveSignalsConfiguration): void { + if (configuration.lingerMilliseconds !== undefined) { + if ( + !Number.isFinite(configuration.lingerMilliseconds) || + configuration.lingerMilliseconds < 0 + ) { + throw new TypeError("lingerMilliseconds must be a non-negative number") + } + lingerMilliseconds = configuration.lingerMilliseconds + } + if (configuration.retryMilliseconds !== undefined) { + if (!Number.isFinite(configuration.retryMilliseconds) || configuration.retryMilliseconds < 0) { + throw new TypeError("retryMilliseconds must be a non-negative number") + } + retryMilliseconds = configuration.retryMilliseconds + } +} + +export function liveEntryCount(runtime: object): number { + const entries = entriesByRuntime.get(runtime) + if (!entries) return 0 + let alive = 0 + for (const reference of entries.values()) { + if (reference.deref() !== undefined) alive += 1 + } + return alive +} + +export function activeLiveSubscriptionCount(): number { + return openSessions.size +} + +interface LiveSession { + receive(request: object): Promise + close(): void +} + +class LiveActorEntry { + readonly #reference: ActorReferenceCore + readonly #states = new Map | undefined>>() + readonly #mirrors = new Map | undefined>>() + readonly #snapshotState = new Signal.State | undefined>(undefined) + #snapshotMirror: LiveSignal | undefined> | undefined + readonly #payloadStates = new Map | undefined>>() + readonly #payloadMirrors = new Map | undefined>>() + readonly #payloadWatcherCounts = new Map() + readonly #payloadFences = new Map() + #payloadsProxy: object | undefined + #watcherCount = 0 + #snapshotWatcherCount = 0 + #session: LiveSession | undefined + #linger: ReturnType | undefined + #instanceId: string | undefined + #revision = -1n + #refreshing = false + #refreshQueued = false + #retry: ReturnType | undefined + + constructor(reference: ActorReferenceCore) { + this.#reference = reference + } + + signalFor(name: string): LiveSignal | undefined> { + const existing = this.#mirrors.get(name) + if (existing) return existing + const state = this.stateFor(name) + const mirror = new Signal.Computed(() => state.get(), { + [Signal.subtle.watched]: () => this.retain({ snapshot: false }), + [Signal.subtle.unwatched]: () => this.release({ snapshot: false }), + }) + this.#mirrors.set(name, mirror) + return mirror + } + + payloadsProxy(): object { + this.#payloadsProxy ??= new Proxy( + {}, + { + get: (_target, property) => { + if (typeof property !== "string") return undefined + return this.payloadSignalFor(property) + }, + has: (_target, property) => typeof property === "string", + }, + ) + return this.#payloadsProxy + } + + payloadSignalFor(name: string): LiveSignal | undefined> { + const existing = this.#payloadMirrors.get(name) + if (existing) return existing + const state = this.payloadStateFor(name) + const mirror = new Signal.Computed(() => state.get(), { + [Signal.subtle.watched]: () => { + this.#payloadWatcherCounts.set(name, (this.#payloadWatcherCounts.get(name) ?? 0) + 1) + this.retain({ snapshot: false, payloadName: name }) + }, + [Signal.subtle.unwatched]: () => { + const count = (this.#payloadWatcherCounts.get(name) ?? 1) - 1 + if (count <= 0) this.#payloadWatcherCounts.delete(name) + else this.#payloadWatcherCounts.set(name, count) + this.release({ snapshot: false }) + }, + }) + this.#payloadMirrors.set(name, mirror) + return mirror + } + + private payloadStateFor(name: string) { + let state = this.#payloadStates.get(name) + if (!state) { + state = new Signal.State | undefined>(undefined) + this.#payloadStates.set(name, state) + } + return state + } + + snapshotSignal(): LiveSignal | undefined> { + this.#snapshotMirror ??= new Signal.Computed(() => this.#snapshotState.get(), { + [Signal.subtle.watched]: () => this.retain({ snapshot: true }), + [Signal.subtle.unwatched]: () => this.release({ snapshot: true }), + }) + return this.#snapshotMirror + } + + private stateFor(name: string) { + let state = this.#states.get(name) + if (!state) { + state = new Signal.State | undefined>(undefined) + this.#states.set(name, state) + } + return state + } + + private retain(options: { snapshot: boolean; payloadName?: string }): void { + this.#watcherCount += 1 + if (options.snapshot) this.#snapshotWatcherCount += 1 + if (this.#linger !== undefined) { + clearTimeout(this.#linger) + this.#linger = undefined + } + if (!this.#session) this.open() + else if (options.payloadName !== undefined) this.sendSubscribe(this.#session) + else if (options.snapshot) void this.refreshSnapshot() + } + + private release(options: { snapshot: boolean }): void { + this.#watcherCount -= 1 + if (options.snapshot) this.#snapshotWatcherCount -= 1 + if (this.#watcherCount > 0) return + if (this.#linger !== undefined) clearTimeout(this.#linger) + this.#linger = setTimeout(() => { + this.#linger = undefined + if (this.#watcherCount === 0) this.close() + }, lingerMilliseconds) + } + + private open(): void { + if (this.#retry !== undefined) { + clearTimeout(this.#retry) + this.#retry = undefined + } + const session = this.#reference.runtime.realtime.connect({ + authorizationContext: undefined, + send: (envelope: RealtimeEnvelope) => this.receive(envelope), + }) as LiveSession + this.#session = session + openSessions.add(this) + void this.sendSubscribe(session) + .then(() => this.refreshSnapshot()) + .catch((error: unknown) => { + this.#reference.runtime.settings.logger.warn({ + event: "solid_objects.live_signals.subscribe_failed", + actorType: this.#reference.actorType, + actorId: this.#reference.actorId, + error: error instanceof Error ? error.name : "Error", + }) + if (this.#session !== session) return + this.close() + if (this.#watcherCount === 0) return + this.#retry = setTimeout(() => { + this.#retry = undefined + if (this.#watcherCount > 0 && !this.#session) this.open() + }, retryMilliseconds) + }) + } + + private sendSubscribe(session: LiveSession): Promise { + const payloadNames = [...this.#payloadWatcherCounts.keys()] + return session.receive({ + version: 1, + action: "subscribe", + actorType: this.#reference.actorType, + actorId: this.#reference.actorId, + ...(payloadNames.length > 0 ? { payloads: payloadNames } : {}), + }) + } + + private close(): void { + if (this.#retry !== undefined) { + clearTimeout(this.#retry) + this.#retry = undefined + } + const session = this.#session + if (!session) return + this.#session = undefined + openSessions.delete(this) + session.close() + } + + private receive(envelope: RealtimeEnvelope): void { + if (envelope.kind === "payload") { + this.receivePayload(envelope) + return + } + if (envelope.kind !== "invalidation") return + const revision = BigInt(envelope.revision) + if (this.#instanceId === envelope.instanceId && revision <= this.#revision) return + this.#instanceId = envelope.instanceId + this.#revision = revision + for (const [name, value] of Object.entries(envelope.observables)) { + this.stateFor(name).set(value as DeepReadonly) + } + void this.refreshSnapshot() + } + + private receivePayload(envelope: RealtimeEnvelope & { kind: "payload" }): void { + const revision = BigInt(envelope.revision) + const fence = this.#payloadFences.get(envelope.name) + if (fence && fence.instanceId === envelope.instanceId && revision <= fence.revision) return + this.#payloadFences.set(envelope.name, { instanceId: envelope.instanceId, revision }) + this.payloadStateFor(envelope.name).set(envelope.payload as DeepReadonly) + } + + private async refreshSnapshot(): Promise { + if (this.#snapshotWatcherCount === 0) return + if (this.#refreshing) { + this.#refreshQueued = true + return + } + this.#refreshing = true + try { + const snapshot = await this.#reference.snapshot() + this.#snapshotState.set(snapshot as ActorSnapshot) + } catch (error) { + this.#reference.runtime.settings.logger.warn({ + event: "solid_objects.live_signals.snapshot_failed", + actorType: this.#reference.actorType, + actorId: this.#reference.actorId, + error: error instanceof Error ? error.name : "Error", + }) + } finally { + this.#refreshing = false + if (this.#refreshQueued) { + this.#refreshQueued = false + void this.refreshSnapshot() + } + } + } +} + +const entriesByRuntime = new WeakMap>>() + +const collectedEntries = new FinalizationRegistry( + (held: { entries: Map>; key: string }) => { + if (held.entries.get(held.key)?.deref() === undefined) held.entries.delete(held.key) + }, +) + +installLiveSignals((reference) => { + let entries = entriesByRuntime.get(reference.runtime) + if (!entries) { + entries = new Map() + entriesByRuntime.set(reference.runtime, entries) + } + const key = `${reference.actorType} ${reference.actorId}` + let entry = entries.get(key)?.deref() + if (!entry) { + entry = new LiveActorEntry(reference) + entries.set(key, new WeakRef(entry)) + collectedEntries.register(entry, { entries, key }) + } + const resolved = entry + return new Proxy( + {}, + { + get(_target, property) { + if (typeof property !== "string") return undefined + if (property === SNAPSHOT_KEY) return resolved.snapshotSignal() + if (property === PAYLOADS_KEY) return resolved.payloadsProxy() + return resolved.signalFor(property) + }, + has(_target, property) { + return typeof property === "string" + }, + }, + ) +}) diff --git a/src/version.ts b/src/version.ts index d2a1bcf..cb02bb9 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1 +1 @@ -export const VERSION = "0.14.0" +export const VERSION = "0.14.1" diff --git a/test/browser-server.mjs b/test/browser-server.mjs index fbfad9e..29e012a 100644 --- a/test/browser-server.mjs +++ b/test/browser-server.mjs @@ -7,6 +7,7 @@ import { createDashboard, createNodeDashboardHandler } from "../dist/web/index.j const root = resolve(import.meta.dirname, "../dist") const sqliteWasmRoot = resolve(import.meta.dirname, "../node_modules/@sqlite.org/sqlite-wasm/dist") +const signalPolyfillRoot = resolve(import.meta.dirname, "../node_modules/signal-polyfill/dist") const browserFixtureRoot = resolve(import.meta.dirname, "browser") const contentTypes = { ".js": "text/javascript; charset=utf-8", @@ -113,11 +114,25 @@ const server = createServer(async (request, response) => { pathname === "/runtime-worker.mjs" || pathname === "/tab-host-worker.mjs" || pathname === "/transmit-worker.mjs" || - pathname === "/shared-db-worker.mjs" + pathname === "/shared-db-worker.mjs" || + pathname === "/live-signals-worker.mjs" ) { await serveFile({ response, path: resolve(browserFixtureRoot, pathname.slice(1)) }) return } + if (pathname.startsWith("/vendor/signal-polyfill/")) { + const vendorPath = resolve( + signalPolyfillRoot, + `.${pathname.slice("/vendor/signal-polyfill".length)}`, + ) + if (!vendorPath.startsWith(`${signalPolyfillRoot}/`)) { + response.writeHead(404) + response.end() + return + } + await serveFile({ response, path: vendorPath }) + return + } if (pathname.startsWith("/vendor/sqlite-wasm/")) { const vendorPath = resolve(sqliteWasmRoot, `.${pathname.slice("/vendor/sqlite-wasm".length)}`) if (!vendorPath.startsWith(`${sqliteWasmRoot}/`)) { @@ -154,6 +169,7 @@ async function serveFile({ response, path }) { contents = contents .toString("utf-8") .replaceAll('"@sqlite.org/sqlite-wasm"', '"/vendor/sqlite-wasm/index.mjs"') + .replaceAll('"signal-polyfill"', '"/vendor/signal-polyfill/index.js"') } response.writeHead(200, { "content-type": contentType }) response.end(contents) diff --git a/test/browser/live-signals-worker.mjs b/test/browser/live-signals-worker.mjs new file mode 100644 index 0000000..8238b32 --- /dev/null +++ b/test/browser/live-signals-worker.mjs @@ -0,0 +1,74 @@ +import { Actor, broadcastValue, configure, sqliteWasm } from "/browser/host.js" +import "/signals.js" +import { Signal } from "signal-polyfill" + +class LiveCounter extends Actor { + static actorType = "LiveCounter" + + count = 0 + + increment() { + this.count += 1 + return this.count + } + + observables() { + return { count: broadcastValue(this.count) } + } +} + +self.onmessage = async (event) => { + const { requestId } = event.data + try { + const database = await sqliteWasm({ path: "live-signals.db" }) + const runtime = configure({ + database, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + authorizeSubscription: () => true, + pollingIntervalMilliseconds: 5, + syncPollingIntervalMilliseconds: 5, + }) + runtime.register(LiveCounter) + await runtime.install() + const runAbort = new AbortController() + const running = runtime.run(runAbort.signal) + + const counter = runtime.ref(LiveCounter, "browser-live") + const observed = [] + const watcher = new Signal.subtle.Watcher(async () => { + await 0 + observed.push(counter.live.count.get()) + watcher.watch() + }) + watcher.watch(counter.live.count) + counter.live.count.get() + + await counter.increment() + await counter.increment() + await waitFor(() => counter.live.count.get() === 2) + + watcher.unwatch(counter.live.count) + runAbort.abort() + await running.catch(() => undefined) + await runtime.close() + await database.close() + postMessage({ requestId, ok: true, value: { final: 2, observed } }) + } catch (error) { + postMessage({ + requestId, + ok: false, + message: String((error && error.message) || error), + stack: String((error && error.stack) || ""), + }) + } +} + +async function waitFor(condition) { + for (let attempt = 0; attempt < 200; attempt += 1) { + if (condition()) return + await new Promise((resolve) => setTimeout(resolve, 25)) + } + throw new Error("condition never became true") +} diff --git a/test/browser/live-signals.browser.ts b/test/browser/live-signals.browser.ts new file mode 100644 index 0000000..7f94771 --- /dev/null +++ b/test/browser/live-signals.browser.ts @@ -0,0 +1,26 @@ +import { expect, test } from "@playwright/test" + +test("live signals track a browser runtime actor", async ({ page }) => { + await page.goto("/") + + const report = await page.evaluate( + () => + new Promise((resolve, reject) => { + const worker = new Worker("/live-signals-worker.mjs", { type: "module" }) + worker.onmessage = (event) => { + worker.terminate() + if (event.data.ok) resolve(event.data.value) + else reject(new Error(`${event.data.message}\n${event.data.stack}`)) + } + worker.onerror = (event) => { + worker.terminate() + reject(new Error(event.message)) + } + worker.postMessage({ requestId: crypto.randomUUID() }) + }), + ) + + expect(report).toMatchObject({ final: 2 }) + const observed = (report as { observed: number[] }).observed + expect(observed[observed.length - 1]).toBe(2) +}) diff --git a/test/live-signals-uninstalled.test.ts b/test/live-signals-uninstalled.test.ts new file mode 100644 index 0000000..664262b --- /dev/null +++ b/test/live-signals-uninstalled.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest" +import "../src/platform/node.js" +import { Actor } from "../src/actor.js" +import { createRuntime } from "../src/runtime.js" +import { sqlite } from "../src/database/sqlite.js" + +class PlainCounter extends Actor { + static override readonly actorType = "PlainCounter" + + count = 0 + + increment(): number { + this.count += 1 + return this.count + } +} + +describe("live signals without the signals entry", () => { + it("explains that solid-objects/signals must be imported", () => { + const runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + }) + const counter = runtime.ref(PlainCounter, "plain") + expect(() => counter.live).toThrow("solid-objects/signals") + }) +}) diff --git a/test/live-signals.test.ts b/test/live-signals.test.ts new file mode 100644 index 0000000..0a17444 --- /dev/null +++ b/test/live-signals.test.ts @@ -0,0 +1,290 @@ +import { Signal } from "signal-polyfill" +import { afterEach, describe, expect, it } from "vitest" +import "../src/platform/node.js" +import { + activeLiveSubscriptionCount, + configureLiveSignals, + liveEntryCount, + type LiveSignal, +} from "../src/signals.js" +import { Actor, broadcastInvalidation, broadcastValue } from "../src/actor.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { sqlite } from "../src/database/sqlite.js" + +class LiveCounter extends Actor { + static override readonly actorType = "LiveCounter" + + count = 0 + note = "quiet" + + increment(): number { + this.count += 1 + return this.count + } + + annotate({ note }: { note: string }): void { + this.note = note + } + + override observables(): Record { + return { + count: broadcastValue(this.count), + note: broadcastInvalidation(this.note), + } + } +} + +class LivePayloadCounter extends Actor { + static override readonly actorType = "LivePayloadCounter" + static override readonly payloads = { + personalized: (actor: LivePayloadCounter) => ({ count: actor.count }), + doubled: (actor: LivePayloadCounter) => ({ value: actor.count * 2 }), + } + + count = 0 + + increment(): number { + this.count += 1 + return this.count + } + + override observables(): Record { + return { count: broadcastValue(this.count) } + } +} + +const runtimes: SolidObjectsRuntime[] = [] +const watchers: Array<{ + watcher: InstanceType + signals: LiveSignal[] +}> = [] + +afterEach(async () => { + for (const { watcher, signals } of watchers) { + for (const signal of signals) watcher.unwatch(signal as never) + } + watchers.length = 0 + await Promise.all(runtimes.map((runtime) => runtime.close())) + runtimes.length = 0 + configureLiveSignals({ lingerMilliseconds: 50, retryMilliseconds: 50 }) + await eventually(() => activeLiveSubscriptionCount() === 0) +}) + +async function liveRuntime(): Promise { + const runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + authorizeSubscription: () => true, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + }) + runtime.register(LiveCounter) + runtime.register(LivePayloadCounter) + runtimes.push(runtime) + await runtime.install() + return runtime +} + +function watch(...signals: LiveSignal[]): void { + const watcher = new Signal.subtle.Watcher(() => {}) + for (const signal of signals) watcher.watch(signal as never) + for (const signal of signals) (signal as { get(): unknown }).get() + watchers.push({ watcher, signals }) +} + +async function eventually(condition: () => boolean): Promise { + for (let attempt = 0; attempt < 200; attempt += 1) { + if (condition()) return + await new Promise((resolve) => setTimeout(resolve, 25)) + } + throw new Error("condition never became true") +} + +describe("live signals", () => { + it("replays committed observables into a watched signal", async () => { + const runtime = await liveRuntime() + const counter = runtime.ref(LiveCounter, "replayed") + await counter.increment() + + watch(counter.live.count!) + await eventually(() => counter.live.count!.get() === 1) + }) + + it("follows later commits through the broadcast outbox", async () => { + const runtime = await liveRuntime() + const counter = runtime.ref(LiveCounter, "following") + watch(counter.live.count!) + await eventually(() => activeLiveSubscriptionCount() === 1) + + await counter.increment() + await runtime.testing.drain({ roles: ["actors", "broadcasts"] }) + await eventually(() => counter.live.count!.get() === 1) + + await counter.increment() + await runtime.testing.drain({ roles: ["actors", "broadcasts"] }) + await eventually(() => counter.live.count!.get() === 2) + }) + + it("keeps live.snapshot current for invalidation-only observables", async () => { + const runtime = await liveRuntime() + const counter = runtime.ref(LiveCounter, "noted") + watch(counter.live.snapshot) + await eventually(() => activeLiveSubscriptionCount() === 1) + + await counter.annotate({ note: "loud" }) + await runtime.testing.drain({ roles: ["actors", "broadcasts"] }) + + await eventually(() => { + const snapshot = counter.live.snapshot.get() as { note?: string } | undefined + return snapshot?.note === "loud" + }) + expect(counter.live.note!.get()).toBeUndefined() + }) + + it("ignores envelopes with stale revisions", async () => { + const runtime = await liveRuntime() + const counter = runtime.ref(LiveCounter, "fenced") + watch(counter.live.count!) + await eventually(() => activeLiveSubscriptionCount() === 1) + await counter.increment() + await runtime.testing.drain({ roles: ["actors", "broadcasts"] }) + await eventually(() => counter.live.count!.get() === 1) + + const incarnation = await runtime.snapshotWithIncarnation(counter) + await runtime.realtime.publish({ + actorType: "LiveCounter", + actorId: "fenced", + instanceId: incarnation.instanceId, + revision: "0", + observables: { count: 999 }, + }) + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(counter.live.count!.get()).toBe(1) + }) + + it("subscribes on first watch and releases after the linger", async () => { + configureLiveSignals({ lingerMilliseconds: 50 }) + const runtime = await liveRuntime() + const counter = runtime.ref(LiveCounter, "lifecycled") + expect(activeLiveSubscriptionCount()).toBe(0) + + const watcher = new Signal.subtle.Watcher(() => {}) + watcher.watch(counter.live.count! as never) + counter.live.count!.get() + await eventually(() => activeLiveSubscriptionCount() === 1) + + watcher.unwatch(counter.live.count! as never) + await eventually(() => activeLiveSubscriptionCount() === 0) + + watcher.watch(counter.live.count! as never) + counter.live.count!.get() + await eventually(() => activeLiveSubscriptionCount() === 1) + watcher.unwatch(counter.live.count! as never) + }) + + it("retries after a failed subscription and recovers", async () => { + configureLiveSignals({ lingerMilliseconds: 50, retryMilliseconds: 50 }) + let denials = 2 + const runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + authorizeSubscription: () => { + if (denials > 0) { + denials -= 1 + return false + } + return true + }, + pollingIntervalMilliseconds: 1, + syncPollingIntervalMilliseconds: 1, + }) + runtime.register(LiveCounter) + runtimes.push(runtime) + await runtime.install() + const counter = runtime.ref(LiveCounter, "recovering") + await counter.increment() + + watch(counter.live.count!) + await eventually(() => counter.live.count!.get() === 1) + expect(denials).toBe(0) + }) + + it("keeps one canonical entry across references and eviction cycles", async () => { + configureLiveSignals({ lingerMilliseconds: 50 }) + const runtime = await liveRuntime() + const first = runtime.ref(LiveCounter, "canonical") + + const watcher = new Signal.subtle.Watcher(() => {}) + watcher.watch(first.live.count! as never) + first.live.count!.get() + await eventually(() => activeLiveSubscriptionCount() === 1) + expect(liveEntryCount(runtime)).toBe(1) + + watcher.unwatch(first.live.count! as never) + await eventually(() => activeLiveSubscriptionCount() === 0) + + const second = runtime.ref(LiveCounter, "canonical") + watcher.watch(second.live.count! as never) + second.live.count!.get() + await eventually(() => activeLiveSubscriptionCount() === 1) + + watcher.watch(first.live.count! as never) + first.live.count!.get() + await new Promise((resolve) => setTimeout(resolve, 150)) + expect(activeLiveSubscriptionCount()).toBe(1) + expect(liveEntryCount(runtime)).toBe(1) + expect(first.live.count).toBe(second.live.count) + + await runtime.ref(LiveCounter, "canonical").increment() + await runtime.testing.drain({ roles: ["actors", "broadcasts"] }) + await eventually(() => first.live.count!.get() === 1) + expect(second.live.count!.get()).toBe(1) + + watcher.unwatch(first.live.count! as never) + watcher.unwatch(second.live.count! as never) + }) + + it("delivers personalized payload signals with independent fences", async () => { + const runtime = await liveRuntime() + const counter = runtime.ref(LivePayloadCounter, "payloaded") + watch(counter.live.payloads.personalized!) + await eventually(() => activeLiveSubscriptionCount() === 1) + + await counter.increment() + await runtime.testing.drain({ roles: ["actors", "broadcasts"] }) + await eventually(() => { + const payload = counter.live.payloads.personalized!.get() as { count?: number } | undefined + return payload?.count === 1 + }) + + watch(counter.live.payloads.doubled!) + await counter.increment() + await runtime.testing.drain({ roles: ["actors", "broadcasts"] }) + await eventually(() => { + const doubled = counter.live.payloads.doubled!.get() as { value?: number } | undefined + return doubled?.value === 4 + }) + expect((counter.live.payloads.personalized!.get() as { count?: number }).count).toBe(2) + expect(counter.live.payloads.personalized).toBe(counter.live.payloads.personalized) + }) + + it("exposes read-only signals and a stable proxy", () => { + const runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + }) + runtimes.push(runtime) + runtime.register(LiveCounter) + const counter = runtime.ref(LiveCounter, "readonly") + + expect(counter.live).toBe(counter.live) + expect(counter.live.count).toBe(counter.live.count) + expect("set" in counter.live.count!).toBe(false) + }) +})