/** * DECISIVE real-broker determination: does `doc_subscribe` actually PUSH when a * subscribed document is written? * * This is the reactive-layer coverage whose ABSENCE let a reactivity bug ship: the * app's whole read-model reactivity rests on `subscribeDoc(nuri, cb)` (the polyfill * wrapper over `ng.doc_subscribe`, `src/subscribe.ts`) firing `cb` again on every * commit to the doc. Two pushes are load-bearing in production and were reported as * NOT firing: * (SELF) a session's own `sparqlUpdate` to a doc it subscribes to. * (CROSS) another session writes to a doc the first session subscribes to. * * This runner exercises BOTH against the REAL broker, through the SAME public * surface the app uses — `subscribeDoc` (via the harness's `stateProbe*` bridge, * which passes the raw `AppResponse` straight through the polyfill wrapper), * `docs.docCreate`, and `docs.sparqlUpdate` (`writeTo`). It records EVERY push as a * typed event (`{ typeKey: "State" | "Patch" | "TabInfo" | …, elapsedMs }`) so the * verdict is the ground truth "did the subscription callback fire again", not a * re-read of the document. Each wait is a single event-driven promise+timeout on the * push (NO re-read loop) — a timeout is a DEFINITE "did-not-fire", not a flaky miss. * * Standalone (NOT `bun test`). Run: * bun run e2e/reactivity-doc-subscribe.ts * (or `bun run test:e2e:reactivity` from packages/client) * * It reuses the exact real-broker plumbing of run.ts / broker.ts: the dedicated lib * wallet, the broker iframe, `window.__sdk`. The CROSS case opens a SECOND page on * the SAME persistent wallet context — a second concurrent verifier session on one * shared wallet (as faithfulReconnect does) — and writes from it. */ import type { Frame, Page, BrowserContext } from "playwright"; import { buildBundle, serveHarness, ensureWallet, launchWalletContext, setupBrokerPage, } from "./broker"; type Check = { name: string; ok: boolean; detail?: string }; const results: Check[] = []; function record(name: string, ok: boolean, detail?: string): void { results.push({ name, ok, detail }); console.log(` [${ok ? "PASS" : "FAIL"}] ${name}${detail ? " — " + detail : ""}`); } type Event = { typeKey: string; elapsedMs: number }; // Call a bridge method inside a given iframe. function sdk(frame: Frame, method: string, ...args: unknown[]): Promise { return frame.evaluate( ([m, a]) => (window as any).__sdk[m as string](...(a as unknown[])), [method, args] as const, ) as Promise; } /** * The decisive wait: resolve TRUE as soon as the probe's recorded push count grows * past `base` (the subscription callback fired again), or FALSE on timeout. This is * a promise+timeout on the PUSH itself — it polls only the in-memory event counter * the `subscribeDoc` callback writes, NEVER re-reads the document. A FALSE here is a * definite non-delivery within the window, not a missed re-read. */ async function waitForPush(frame: Frame, base: number, timeoutMs: number): Promise { try { await frame.waitForFunction( (b) => (window as any).__sdk.stateProbeEvents().length > (b as number), base, { timeout: timeoutMs }, ); return true; } catch { return false; // timed out → the callback did NOT fire again within the window } } const seq = (events: Event[]): string => events.length ? events.map((e) => `${e.typeKey}@${e.elapsedMs}ms`).join(" → ") : "(none)"; async function openSession( ctx: BrowserContext, url: string, tag: string, ): Promise<{ page: Page; frame: Frame; sessionId: string }> { const page = await ctx.newPage(); page.on("pageerror", (e) => console.error(`[iframe error:${tag}]`, e.message)); page.on("console", (m) => { const t = m.text(); // Surface the polyfill's own "doc_subscribe FIRE" diagnostic (subscribe.ts) if // access logging happens to be on — an independent confirmation of a push. if (m.type() === "error") console.error(`[iframe console:${tag}]`, t); else if (t.includes("doc_subscribe FIRE")) console.log(`[${tag}] ${t}`); }); const frame = await setupBrokerPage(page, url); await frame.waitForFunction(() => (window as any).__sdk !== undefined, { timeout: 30000 }); await frame.waitForFunction(() => (window as any).__sdk.status() === "connected", { timeout: 60000, }); const info = await sdk<{ session_id: string } | null>(frame, "sessionInfo"); const sessionId = info?.session_id ?? "(none)"; console.log(`[session:${tag}] connected — session_id=${sessionId}`); return { page, frame, sessionId }; } const SELF_TIMEOUT_MS = 10000; const CROSS_TIMEOUT_MS = 15000; const STATE_TIMEOUT_MS = 20000; async function main(): Promise { console.log("[reactivity] building SDK page bundle..."); buildBundle(); console.log("[reactivity] ensuring dedicated lib wallet..."); await ensureWallet(); const { url, close: closeServer } = await serveHarness(); console.log(`[reactivity] harness served at ${url}`); let ctx: BrowserContext | null = null; try { ctx = await launchWalletContext(); // ── Session A (the subscriber for both cases) ──────────────────────────── const A = await openSession(ctx, url, "A"); // ════════════════════════════════════════════════════════════════════════ // CASE 1 — SELF: A subscribes to D, then A itself writes to D. // ════════════════════════════════════════════════════════════════════════ console.log("\n── CASE 1: SELF (single session — own write to own subscribed doc) ──"); { const doc = await sdk(A.frame, "docCreate"); console.log(` [SELF] created doc D = ${doc}`); await sdk(A.frame, "stateProbeSubscribe", doc); // Wait for the initial State (the sync barrier). TabInfo may precede it. const gotState = await (async () => { try { await A.frame.waitForFunction( () => (window as any).__sdk.stateProbeStateCount() >= 1, { timeout: STATE_TIMEOUT_MS }, ); return true; } catch { return false; } })(); const afterSubscribe = await sdk(A.frame, "stateProbeEvents"); console.log(` [SELF] pushes after subscribe: ${seq(afterSubscribe)}`); record( "SELF: initial State push arrives on subscribe (baseline sanity)", gotState && afterSubscribe.some((e) => e.typeKey === "State"), `sequence=${seq(afterSubscribe)}`, ); // Now the decisive write: A's OWN sparqlUpdate to D. const preWrite = afterSubscribe.length; console.log(` [SELF] A writes to D (own sparqlUpdate); waiting ≤${SELF_TIMEOUT_MS}ms for a push…`); await sdk(A.frame, "writeTo", doc, "self-1"); const fired = await waitForPush(A.frame, preWrite, SELF_TIMEOUT_MS); const afterWrite = await sdk(A.frame, "stateProbeEvents"); const newEvents = afterWrite.slice(preWrite); console.log(` [SELF] pushes AFTER own write: ${seq(newEvents)}`); console.log(` [SELF] VERDICT: callback ${fired ? "FIRED" : "did NOT fire"} within ${SELF_TIMEOUT_MS}ms`); record( `SELF: subscription callback fires on the session's OWN write (≤${SELF_TIMEOUT_MS}ms)`, fired, `newPushes=${seq(newEvents)}`, ); await sdk(A.frame, "stateProbeStop"); } // ════════════════════════════════════════════════════════════════════════ // CASE 2 — CROSS-SESSION: A subscribes to D2; a SECOND session B (same shared // wallet, own concurrent verifier session) writes to D2. // ════════════════════════════════════════════════════════════════════════ console.log("\n── CASE 2: CROSS-SESSION (session B writes to a doc session A subscribes to) ──"); let B: { page: Page; frame: Frame; sessionId: string } | null = null; try { B = await openSession(ctx, url, "B"); } catch (e: any) { console.log(` [CROSS] COULD-NOT-TEST: second concurrent session on the shared wallet failed to open: ${String(e?.message ?? e)}`); record( "CROSS: second concurrent session opened on the shared wallet", false, `open failed: ${String(e?.message ?? e)} — see Festipod multibrowser harness as the alternative venue`, ); } if (B) { // NB: `session_id` is a PER-PAGE local verifier counter (each fresh iframe // numbers its first session "1"), so it is NOT a global identifier and cannot // be used to prove distinctness. The REAL proof that A and B are two separate // verifier sessions is behavioural: B's write reaches A only after a broker // round-trip (a delayed Patch), not as an instant same-session echo. console.log( ` [CROSS] both pages connected — A.session=${A.sessionId} B.session=${B.sessionId} (per-page local counter; distinctness shown by the cross-broker propagation below)`, ); record( "CROSS: a second concurrent page/session is open on the same shared wallet", true, `A=${A.sessionId} B=${B.sessionId} (session_id is a per-page counter, not a global id)`, ); // A creates D2 and subscribes. const doc2 = await sdk(A.frame, "docCreate"); console.log(` [CROSS] A created doc D2 = ${doc2}`); await sdk(A.frame, "stateProbeSubscribe", doc2); const gotState2 = await (async () => { try { await A.frame.waitForFunction( () => (window as any).__sdk.stateProbeStateCount() >= 1, { timeout: STATE_TIMEOUT_MS }, ); return true; } catch { return false; } })(); const afterSub2 = await sdk(A.frame, "stateProbeEvents"); console.log(` [CROSS] A pushes after subscribe: ${seq(afterSub2)}`); record( "CROSS: A receives its initial State on D2 (baseline sanity)", gotState2 && afterSub2.some((e) => e.typeKey === "State"), `sequence=${seq(afterSub2)}`, ); // B writes to D2. Capture a write failure (e.g. RepoNotFound) explicitly — // it would mean B cannot reach A's doc, which is itself a determination. const preCross = afterSub2.length; let writeThrew: string | null = null; // Cross-session writes to a doc created by ANOTHER session can be slow: B must // sync/open D2's repo before it can commit. Time it separately so the push // latency is reported relative to when B's write actually LANDED, not to // subscribe time. console.log(` [CROSS] B writes to D2 from its own session…`); const tWriteStart = Date.now(); try { await sdk(B.frame, "writeTo", doc2, "cross-1"); } catch (e: any) { writeThrew = String(e?.message ?? e); console.log(` [CROSS] B's write THREW: ${writeThrew}`); } const writeMs = Date.now() - tWriteStart; record("CROSS: session B's write to D2 did not throw", writeThrew === null, writeThrew ? writeThrew : `landed in ${writeMs}ms`); console.log(` [CROSS] B's write returned in ${writeMs}ms; now waiting ≤${CROSS_TIMEOUT_MS}ms for A's push…`); const tWaitStart = Date.now(); const crossFired = writeThrew ? false : await waitForPush(A.frame, preCross, CROSS_TIMEOUT_MS); const pushAfterWriteMs = Date.now() - tWaitStart; const afterCross = await sdk(A.frame, "stateProbeEvents"); const crossNew = afterCross.slice(preCross); console.log(` [CROSS] A pushes AFTER B's write: ${seq(crossNew)}`); console.log( ` [CROSS] VERDICT: A's callback ${crossFired ? `FIRED (${pushAfterWriteMs}ms after B's write landed)` : "did NOT fire"} within ${CROSS_TIMEOUT_MS}ms${writeThrew ? " (B's write threw first)" : ""}`, ); record( `CROSS: A's subscription callback fires on B's write (≤${CROSS_TIMEOUT_MS}ms after B's write landed)`, crossFired, `newPushes=${seq(crossNew)} (B write took ${writeMs}ms; push ${crossFired ? pushAfterWriteMs + "ms after" : "not seen"})${writeThrew ? ` — B write threw: ${writeThrew}` : ""}`, ); await sdk(A.frame, "stateProbeStop"); } } finally { try { if (ctx) await ctx.close(); } catch { /* ignore */ } closeServer(); } // ── Determination summary (not a pass/fail gate — this is a probe) ────────── console.log("\n══ doc_subscribe delivery determination ══"); for (const r of results) console.log(` [${r.ok ? "PASS" : "FAIL"}] ${r.name}${r.detail ? " — " + r.detail : ""}`); const self = results.find((r) => r.name.startsWith("SELF: subscription callback fires")); const cross = results.find((r) => r.name.startsWith("CROSS: A's subscription callback fires")); console.log("\n SELF →", self ? (self.ok ? "FIRES" : "DOES-NOT-FIRE") : "could-not-test"); console.log(" CROSS →", cross ? (cross.ok ? "FIRES" : "DOES-NOT-FIRE") : "could-not-test"); } main().catch((e) => { console.error("[reactivity] fatal:", e); process.exit(1); });