Sylvain Duchesne
276 lines
14 KiB
TypeScript
276 lines
14 KiB
TypeScript
/**
|
|
* 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/polyfill)
|
|
*
|
|
* It reuses the exact real-broker plumbing of run.ts (`ng-e2e-helpers`): 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 { launchWatchedContext, setupBrokerPage, type RunProfile, type Frame, type Page, type BrowserContext } from "ng-e2e-helpers";
|
|
import { WALLET, buildBundle, mintBatchWallet, serveHarness } from "./harness-page";
|
|
|
|
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<T>(frame: Frame, method: string, ...args: unknown[]): Promise<T> {
|
|
return frame.evaluate(
|
|
([m, a]) => (window as any).__sdk[m as string](...(a as unknown[])),
|
|
[method, args] as const,
|
|
) as Promise<T>;
|
|
}
|
|
|
|
/**
|
|
* 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<boolean> {
|
|
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, WALLET.password);
|
|
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<void> {
|
|
console.log("[reactivity] building SDK page bundle...");
|
|
buildBundle();
|
|
console.log("[reactivity] minting this run's wallet...");
|
|
const wallet: RunProfile = await mintBatchWallet("the reactivity suite (e2e/reactivity-doc-subscribe.ts)");
|
|
const { url, close: closeServer } = await serveHarness();
|
|
console.log(`[reactivity] harness served at ${url}`);
|
|
|
|
let ctx: BrowserContext | null = null;
|
|
try {
|
|
ctx = await launchWatchedContext("reactivity", wallet.dir);
|
|
|
|
// ── 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<string>(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<Event[]>(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<Event[]>(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<string>(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<Event[]>(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<Event[]>(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();
|
|
wallet.discard();
|
|
}
|
|
|
|
// ── 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);
|
|
});
|