From c0498a6ebc6594d415ddaf92383f3fcdadcdc12f Mon Sep 17 00:00:00 2001 From: Sylvain Duchesne Date: Mon, 6 Jul 2026 23:04:32 +0200 Subject: [PATCH] feat(client): per-document reactive subscription (doc_subscribe), drop polling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expose subscribeDoc(nuri, onChange) / subscribeDocs(nuris, onChange) wrapping the real ng.doc_subscribe — per-document, event-driven (initial State + a Patch per commit, local or broker-synced from a remote peer), returning a sync unsubscribe. subscribeDocs isolates per doc (a failing/unsynced doc never aborts the others), so it sidesteps the ORM fan-out hang (never orm_start_graph(graphs:[…])). Replace the setInterval polling in inbox.watch() and discovery.watchIndex() with doc_subscribe — same public contract, now push not poll. NextGraph is subscription- first; no polling remains. Verified against the real broker (@data harness spike): the doc_subscribe callback marshals across the iframe RPC (@ng-org/web strips the callback and drives it via a MessagePort — no DataCloneError) and fires on the initial push and on a real write. Lib bun test 91 pass, tsc clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/client/src/discovery.ts | 41 +++++-- packages/client/src/inbox.ts | 33 ++++-- packages/client/src/index.ts | 2 + packages/client/src/subscribe.ts | 149 +++++++++++++++++++++++++ packages/client/test/discovery.test.ts | 22 +++- packages/client/test/inbox.test.ts | 24 +++- packages/client/test/subscribe.test.ts | 147 ++++++++++++++++++++++++ 7 files changed, 396 insertions(+), 22 deletions(-) create mode 100644 packages/client/src/subscribe.ts create mode 100644 packages/client/test/subscribe.test.ts diff --git a/packages/client/src/discovery.ts b/packages/client/src/discovery.ts index 5e1c84b..1f20439 100644 --- a/packages/client/src/discovery.ts +++ b/packages/client/src/discovery.ts @@ -38,6 +38,7 @@ */ import * as inbox from "./inbox"; +import { subscribeDoc } from "./subscribe"; import { ensureAccount, reservedAccount } from "./store-registry"; import { getCaps } from "./polyfill"; import type { Nuri, PrincipalId } from "./types"; @@ -156,18 +157,25 @@ export async function readIndex(): Promise { } /** - * Watch the discovery index. Polls {@link readIndex} and fires `onEntries` - * whenever the index changes. Returns an unsubscribe. Fires once immediately. - * (Deduplication is applied on each read.) + * Watch the discovery index — **event-driven, not polled**. Subscribes to the + * index document via {@link subscribeDoc} (the platform's `doc_subscribe` push): + * `onEntries` fires once on the initial state push and again on every subsequent + * change to the index document — a local submission OR a broker-synced remote one. + * Returns an unsubscribe. (Deduplication is applied on each read.) + * + * The `intervalMs` option is accepted for signature compatibility but IGNORED: + * there is no polling. The index is a single document, so this is immune to the + * ORM fan-out hang (see {@link subscribeDoc}). */ export function watchIndex( onEntries: (entries: IndexEntry[]) => void, - opts?: { intervalMs?: number }, + _opts?: { intervalMs?: number }, ): () => void { let stopped = false; let lastCount = -1; - const intervalMs = opts?.intervalMs ?? 1000; - const tick = async (): Promise => { + let unsubscribe: (() => void) | null = null; + + const refresh = async (): Promise => { if (stopped) return; try { const entries = await readIndex(); @@ -179,10 +187,25 @@ export function watchIndex( console.error("[discovery] watchIndex read failed:", error); } }; - void tick(); - const handle = setInterval(() => void tick(), intervalMs); + + // The index document NURI is resolved async (ensureAccount); subscribe once it + // is known. The initial State push fires the first read (onEntries fires once), + // each later Patch a re-read. + void (async () => { + try { + const anchor = await indexInboxNuri(); + if (stopped) return; + unsubscribe = subscribeDoc(anchor, () => void refresh()); + } catch (error) { + console.error("[discovery] watchIndex subscribe failed:", error); + } + })(); + return () => { stopped = true; - clearInterval(handle); + if (unsubscribe) { + unsubscribe(); + unsubscribe = null; + } }; } diff --git a/packages/client/src/inbox.ts b/packages/client/src/inbox.ts index d1a7b1f..18b0952 100644 --- a/packages/client/src/inbox.ts +++ b/packages/client/src/inbox.ts @@ -25,6 +25,7 @@ */ import { sparqlUpdate, sparqlQuery } from "./docs"; +import { subscribeDoc } from "./subscribe"; import { getCurrentUser, getStoreRegistryDeps } from "./polyfill"; import { escapeLiteral } from "./sparql"; import type { Nuri, PrincipalId } from "./types"; @@ -186,22 +187,31 @@ export async function read(targetInbox: Nuri): Promise { export const materialize = read; /** - * Subscription over an inbox. Polls {@link read} on an interval and invokes - * `onDeposits` with the full current deposit list whenever it changes (grows). - * Returns an unsubscribe function. The polyfill has no reactive inbox - * subscription, so this polls; against real NextGraph it follows the recipient's - * own inbox processing. `onDeposits` fires once immediately. + * Subscription over an inbox — **event-driven, not polled**. Subscribes to the + * inbox document via {@link subscribeDoc} (the platform's `doc_subscribe` push): + * `onDeposits` fires once on the initial state push and again on every subsequent + * change to the inbox document — a local deposit OR a broker-synced remote one. + * Returns an unsubscribe function. + * + * On each push it re-reads the full deposit list ({@link read}) and invokes + * `onDeposits` only when the deposit count changed (grew), keeping the same + * "fires on change" contract the polling watcher had — same callback signature + * and same behaviour, just event-driven instead of `setInterval`. + * + * The `intervalMs` option is accepted for signature compatibility but IGNORED: + * there is no polling. (The inbox document is a single doc, so this is immune to + * the ORM fan-out hang — see {@link subscribeDoc}.) */ export function watch( targetInbox: Nuri, onDeposits: (deposits: Deposit[]) => void, - opts?: { intervalMs?: number }, + _opts?: { intervalMs?: number }, ): () => void { let stopped = false; let lastCount = -1; - const intervalMs = opts?.intervalMs ?? 1000; - const tick = async (): Promise => { + // Re-read on every push; fire onDeposits only when the set changed (grew). + const refresh = async (): Promise => { if (stopped) return; try { const deposits = await read(targetInbox); @@ -214,10 +224,11 @@ export function watch( } }; - void tick(); // fire once immediately - const handle = setInterval(() => void tick(), intervalMs); + // Subscribe to the inbox document: the initial State push fires the first read + // (so onDeposits fires once immediately, as before), each later Patch a re-read. + const unsubscribe = subscribeDoc(targetInbox, () => void refresh()); return () => { stopped = true; - clearInterval(handle); + unsubscribe(); }; } diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index 9e0d225..70a8a0d 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -18,6 +18,8 @@ export * as inbox from "./inbox"; export * as discovery from "./discovery"; export type { IndexEntry, SubmitOptions } from "./discovery"; export * as docs from "./docs"; +export { subscribeDoc, subscribeDocs } from "./subscribe"; +export type { DocChange, Unsubscribe } from "./subscribe"; export * as readModel from "./read-model"; export type { UnionSubject } from "./read-model"; export * as storeRegistry from "./store-registry"; diff --git a/packages/client/src/subscribe.ts b/packages/client/src/subscribe.ts new file mode 100644 index 0000000..59abaa5 --- /dev/null +++ b/packages/client/src/subscribe.ts @@ -0,0 +1,149 @@ +/** + * Reactive single-document subscription — the polyfill's typed wrapper over the + * platform's `doc_subscribe` primitive. This is the canonical NextGraph reactive + * read at the document granularity: subscribe once, get the initial state pushed, + * then a push on every subsequent commit to that document — whether the write was + * local (this session) or a broker-synced remote change. NO POLLING. + * + * ── Why call the REAL injected `ng` directly (never `makeNg`) ────────────── + * Same hard constraint as `docs.ts`: the public `ng` is a JS `Proxy` over + * `@ng-org/web`'s iframe-RPC proxy. `doc_subscribe` is a STREAMED method — the + * `@ng-org/web` RPC strips the callback (by positional index) BEFORE it posts to + * the iframe and drives it locally via a `MessageChannel` port (the function is + * never structured-cloned, so no `DataCloneError`). Layering our own Proxy on top + * risks re-wrapping that surface; reaching the real `ng` held in the config avoids + * the double-proxy exactly as the raw `docs` primitives do. Do not import from + * `./ng-proxy`. + * + * ── The primitive shape (verified against nextgraph-rs) ──────────────────── + * `ng.doc_subscribe(repo_o: string, session_id, callback)` + * (`sdk/js/lib-wasm/src/lib.rs:1907`) is **per-document** — one repo NURI, one + * callback. It is `async`, resolving to a JS **unsubscribe function**. The + * callback is invoked `callback(appResponse)` with a serialized `AppResponse`: + * `{ V0: { State | Patch | TabInfo | ... } }`. It pushes an initial `State` + * (plus a `TabInfo`) on subscribe, then a `Patch` per verified commit on the + * branch. Returning `true` from the callback also cancels; we cancel by calling + * the returned unsubscribe fn. + * + * ── Why per-document, never `orm_start_graph(graphs:[…])` ────────────────── + * A single not-yet-synced repo in an ORM graph fan-out makes `RepoNotFound` abort + * the WHOLE subscription (`initialize.rs:125-128`), so the readyPromise never + * resolves → the ~75s hang. `doc_subscribe` is per-branch/per-doc and has no + * fan-out: an absent doc breaks only its own subscription. {@link subscribeDocs} + * builds a set of these with per-doc error isolation to preserve that property. + */ + +import { getConfig, getStoreRegistryDeps } from "./polyfill"; +import type { Nuri } from "./types"; + +/** + * A push from the platform to a document subscriber. Loosely typed: the raw + * serialized `AppResponse` (`{ V0: { State | Patch | TabInfo | ... } }`). The + * consumer typically ignores the payload and uses the push purely as a + * change SIGNAL (re-query on change — the read-model pattern), so this stays + * permissive rather than modelling every AppResponse variant. + */ +export type DocChange = unknown; + +/** An unsubscribe function — idempotent (calling it twice is a no-op). */ +export type Unsubscribe = () => void; + +async function sessionId(): Promise { + return (await getStoreRegistryDeps().getSession()).sessionId; +} + +/** + * Subscribe to ONE document. `onChange` fires on the initial state push and on + * every subsequent change to that doc (local write OR broker-synced remote + * change). Returns an unsubscribe function. + * + * The wrapper is synchronous-returning (an unsubscribe fn) even though the + * underlying `ng.doc_subscribe` is async: the real unsubscribe is captured when + * the promise resolves; if the caller unsubscribes before setup completes, the + * cancellation is honoured as soon as the real unsubscribe is available (and no + * further `onChange` fires after unsubscribe). + * + * Calls the REAL injected `ng.doc_subscribe` directly (never `makeNg`). + */ +export function subscribeDoc(nuri: Nuri, onChange: (r: DocChange) => void): Unsubscribe { + const { ng } = getConfig(); + let stopped = false; + let realUnsub: (() => void) | null = null; + + const cb = (resp: DocChange): void => { + if (stopped) return; + try { + onChange(resp); + } catch (error) { + console.error("[subscribe] onChange handler threw for", nuri, error); + } + }; + + // Kick off the async subscription. Errors are isolated to this doc (they never + // reject a shared batch — see subscribeDocs). If setup fails, this doc simply + // never fires; the caller's unsubscribe stays a safe no-op. + void (async () => { + try { + const sid = await sessionId(); + const unsub = (await ng.doc_subscribe(nuri, sid, cb)) as (() => void) | undefined; + if (stopped) { + // Unsubscribed before setup resolved — cancel immediately. + if (typeof unsub === "function") unsub(); + return; + } + realUnsub = typeof unsub === "function" ? unsub : null; + } catch (error) { + console.error("[subscribe] doc_subscribe failed for", nuri, error); + } + })(); + + return () => { + if (stopped) return; + stopped = true; + if (realUnsub) { + try { + realUnsub(); + } catch (error) { + console.error("[subscribe] unsubscribe failed for", nuri, error); + } + realUnsub = null; + } + }; +} + +/** + * Subscribe to a SET of documents, one {@link subscribeDoc} per NURI, with + * PER-DOC error isolation. `onChange(nuri, r)` fires for whichever doc changed. + * Returns a single unsubscribe that tears down all of them. + * + * The per-doc isolation is the point: a bad / not-yet-synced doc breaks only its + * own subscription and NEVER aborts the others (this is precisely what avoids the + * ORM fan-out hang — do NOT replace this with `orm_start_graph(graphs:[…])`). The + * set is deduplicated; an empty set returns a no-op unsubscribe. + */ +export function subscribeDocs( + nuris: Nuri[], + onChange: (nuri: Nuri, r: DocChange) => void, +): Unsubscribe { + const unique = [...new Set(nuris.filter(Boolean))]; + const unsubs = unique.map((nuri) => { + // Each subscription is independent: subscribeDoc already isolates its own + // async setup failure (logged, never thrown), so one bad doc cannot abort the + // construction of the others here. + try { + return subscribeDoc(nuri, (r) => onChange(nuri, r)); + } catch (error) { + console.error("[subscribe] subscribeDocs: failed to subscribe", nuri, error); + return () => {}; + } + }); + return () => { + for (const u of unsubs) { + try { + u(); + } catch (error) { + console.error("[subscribe] subscribeDocs: unsubscribe failed", error); + } + } + }; +} diff --git a/packages/client/test/discovery.test.ts b/packages/client/test/discovery.test.ts index 9b93194..dc2272b 100644 --- a/packages/client/test/discovery.test.ts +++ b/packages/client/test/discovery.test.ts @@ -57,6 +57,25 @@ function makeFakeNg() { const quads: Quad[] = []; let docCounter = 0; + // Reactive subscriptions (see inbox.test.ts): doc_subscribe registers a + // callback per anchor + fires an initial push; sparql_update pushes a Patch to + // that anchor's subscribers, so discovery.watchIndex (now event-driven) works + // without a timer. + const subs = new Map void>>(); + const doc_subscribe = mock(async (nuri: string, _sid: unknown, cb: (r: unknown) => void) => { + let set = subs.get(nuri); + if (!set) { + set = new Set(); + subs.set(nuri, set); + } + set.add(cb); + queueMicrotask(() => cb({ V0: { State: { doc: nuri } } })); + return () => set!.delete(cb); + }); + const pushTo = (anchor: string): void => { + for (const cb of subs.get(anchor) ?? []) cb({ V0: { Patch: { doc: anchor } } }); + }; + const doc_create = mock(async (..._a: unknown[]) => `did:ng:o:doc${++docCounter}`); const sparql_update = mock(async (...a: unknown[]) => { @@ -93,6 +112,7 @@ function makeFakeNg() { const o = m[2] !== undefined ? unescapeLiteral(m[2]) : (m[3] ?? ""); quads.push({ g, s, p, o }); } + pushTo(g); // local-push to the written graph's subscribers return undefined; }); @@ -157,7 +177,7 @@ function makeFakeNg() { return { results: { bindings: [] } }; }); - return { doc_create, sparql_update, sparql_query, _quads: quads }; + return { doc_create, doc_subscribe, sparql_update, sparql_query, _quads: quads }; } const SESSION: RegistrySession = { sessionId: "sid-1", privateStoreId: "PRIV" }; diff --git a/packages/client/test/inbox.test.ts b/packages/client/test/inbox.test.ts index 182baa7..ce47feb 100644 --- a/packages/client/test/inbox.test.ts +++ b/packages/client/test/inbox.test.ts @@ -57,6 +57,25 @@ function unescapeLiteral(s: string): string { function makeFakeNg() { const quads: Quad[] = []; + // Reactive subscriptions: doc_subscribe registers a callback per anchor and + // fires an initial State push; a matching sparql_update pushes a Patch to that + // anchor's subscribers. This mirrors the real broker's local-push behaviour so + // inbox.watch (now event-driven, no polling) can be tested without a timer. + const subs = new Map void>>(); + const doc_subscribe = mock(async (nuri: string, _sid: unknown, cb: (r: unknown) => void) => { + let set = subs.get(nuri); + if (!set) { + set = new Set(); + subs.set(nuri, set); + } + set.add(cb); + queueMicrotask(() => cb({ V0: { State: { doc: nuri } } })); // initial push + return () => set!.delete(cb); + }); + const pushTo = (anchor: string): void => { + for (const cb of subs.get(anchor) ?? []) cb({ V0: { Patch: { doc: anchor } } }); + }; + const doc_create = mock(async (..._a: unknown[]) => "did:ng:o:new"); // Parses one deposit: ` a ; "..." ; "..." [; "..."] .` @@ -89,6 +108,9 @@ function makeFakeNg() { const o = rawLit !== undefined ? unescapeLiteral(rawLit) : (m[3] ?? ""); quads.push({ g, s, p, o }); } + // A write to `g` (the anchored default graph) pushes a Patch to that doc's + // subscribers — the local-push the real broker performs on a verified commit. + pushTo(g); return undefined; }); @@ -121,7 +143,7 @@ function makeFakeNg() { return { results: { bindings } }; }); - return { doc_create, sparql_update, sparql_query, _quads: quads }; + return { doc_create, doc_subscribe, sparql_update, sparql_query, _quads: quads }; } const SESSION: RegistrySession = { sessionId: "sid-1", privateStoreId: "PRIV" }; diff --git a/packages/client/test/subscribe.test.ts b/packages/client/test/subscribe.test.ts new file mode 100644 index 0000000..ada884c --- /dev/null +++ b/packages/client/test/subscribe.test.ts @@ -0,0 +1,147 @@ +import { test, expect, mock, afterAll } from "bun:test"; +import { subscribeDoc, subscribeDocs } from "../src/subscribe"; +import { + configure, + configureStoreRegistry, + resetConfig, + resetStoreRegistry, +} from "../src/polyfill"; +import type { RegistrySession } from "../src/store-registry"; + +// subscribeDoc/subscribeDocs wrap the REAL injected `ng.doc_subscribe`. This +// suite injects a fake `ng` whose `doc_subscribe` records the callback per doc +// and hands back an unsubscribe, so we can assert routing + isolation without a +// broker. Restore the un-configured state at the end. +afterAll(() => { + resetConfig(); + resetStoreRegistry(); +}); + +const SESSION: RegistrySession = { sessionId: "sid-1", privateStoreId: "PRIV" }; + +/** + * A fake reactive `ng`: `doc_subscribe(nuri, sid, cb)` registers `cb` for `nuri`, + * fires it once (initial State push), and returns an unsubscribe. `push(nuri)` + * drives a later change to that doc's subscribers. A per-doc `failFor` set makes + * `doc_subscribe` reject for chosen NURIs (a not-yet-synced doc). + */ +function makeFakeNg(failFor: Set = new Set()) { + const subs = new Map void>>(); + const doc_subscribe = mock(async (nuri: string, _sid: unknown, cb: (r: unknown) => void) => { + if (failFor.has(nuri)) throw new Error(`RepoNotFound: ${nuri}`); + let set = subs.get(nuri); + if (!set) { + set = new Set(); + subs.set(nuri, set); + } + set.add(cb); + // Initial State push, delivered async (as the real RPC does). + queueMicrotask(() => cb({ V0: { State: { doc: nuri } } })); + return () => set!.delete(cb); + }); + const push = (nuri: string): void => { + for (const cb of subs.get(nuri) ?? []) cb({ V0: { Patch: { doc: nuri } } }); + }; + const isSubscribed = (nuri: string): boolean => (subs.get(nuri)?.size ?? 0) > 0; + return { doc_subscribe, push, isSubscribed, _subs: subs }; +} + +function inject(failFor?: Set) { + const ng = makeFakeNg(failFor); + configure({ ng: ng as any, useShape: (() => {}) as any }); + configureStoreRegistry({ getSession: async () => SESSION }); + return ng; +} + +const A = "did:ng:o:docA"; +const B = "did:ng:o:docB"; + +const tick = () => new Promise((r) => setTimeout(r, 5)); + +test("subscribeDoc calls ng.doc_subscribe with (nuri, sessionId, callback)", async () => { + const ng = inject(); + const onChange = mock(() => {}); + subscribeDoc(A, onChange); + await tick(); + expect(ng.doc_subscribe).toHaveBeenCalledTimes(1); + const call = ng.doc_subscribe.mock.calls[0]!; + expect(call[0]).toBe(A); + expect(call[1]).toBe("sid-1"); // sessionId from the injected session + expect(typeof call[2]).toBe("function"); // the callback +}); + +test("subscribeDoc routes the initial State push and every later change", async () => { + const ng = inject(); + const seen: unknown[] = []; + subscribeDoc(A, (r) => seen.push(r)); + await tick(); + expect(seen).toHaveLength(1); // initial State push + ng.push(A); + ng.push(A); + expect(seen).toHaveLength(3); // + 2 patches +}); + +test("subscribeDoc unsubscribe stops further callbacks", async () => { + const ng = inject(); + const seen: unknown[] = []; + const stop = subscribeDoc(A, (r) => seen.push(r)); + await tick(); + expect(seen).toHaveLength(1); + stop(); + expect(ng.isSubscribed(A)).toBe(false); // real unsubscribe was invoked + ng.push(A); // ignored — no subscriber + expect(seen).toHaveLength(1); +}); + +test("subscribeDoc unsubscribe BEFORE async setup resolves cancels cleanly", async () => { + const ng = inject(); + const seen: unknown[] = []; + const stop = subscribeDoc(A, (r) => seen.push(r)); + stop(); // before the microtask/promise setup resolved + await tick(); + // The subscription was cancelled the moment setup resolved: no callbacks, and + // no lingering subscriber. + expect(seen).toHaveLength(0); + expect(ng.isSubscribed(A)).toBe(false); +}); + +test("subscribeDocs fans out one subscription per doc and reports the source nuri", async () => { + const ng = inject(); + const seen: Array<[string, unknown]> = []; + subscribeDocs([A, B], (nuri, r) => seen.push([nuri, r])); + await tick(); + // Two initial pushes, one per doc. + expect(seen.map((s) => s[0]).sort()).toEqual([A, B]); + ng.push(B); + expect(seen.filter((s) => s[0] === B)).toHaveLength(2); // initial + patch + expect(seen.filter((s) => s[0] === A)).toHaveLength(1); // isolated: A didn't fire +}); + +test("subscribeDocs isolates a failing doc — the others still fire", async () => { + const ng = inject(new Set([A])); // A's subscription throws (RepoNotFound) + const seen: Array<[string, unknown]> = []; + subscribeDocs([A, B], (nuri, r) => seen.push([nuri, r])); + await tick(); + // A failed to subscribe (logged, not thrown); B is unaffected and fired. + expect(seen.map((s) => s[0])).toEqual([B]); + ng.push(B); + expect(seen.filter((s) => s[0] === B)).toHaveLength(2); +}); + +test("subscribeDocs unsubscribe tears down all subscriptions", async () => { + const ng = inject(); + const stop = subscribeDocs([A, B], () => {}); + await tick(); + expect(ng.isSubscribed(A)).toBe(true); + expect(ng.isSubscribed(B)).toBe(true); + stop(); + expect(ng.isSubscribed(A)).toBe(false); + expect(ng.isSubscribed(B)).toBe(false); +}); + +test("subscribeDocs deduplicates repeated NURIs", async () => { + const ng = inject(); + subscribeDocs([A, A, A], () => {}); + await tick(); + expect(ng.doc_subscribe).toHaveBeenCalledTimes(1); +});