From 935cce4d7b2286184055966997b352f6f38cc05e Mon Sep 17 00:00:00 2001 From: Sylvain Duchesne Date: Mon, 17 Aug 2026 11:11:04 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20un=20second=20doc=5Fsubscribe=20tuait=20?= =?UTF-8?q?le=20premier,=20et=20les=20inbox=20n'=C3=A9taient=20lues=20qu'?= =?UTF-8?q?=C3=A0=20la=20connexion?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 3 + .project/concepts/e2e-harness/_debt.md | 10 + .../polyfill/src/emulated-verifier/connect.ts | 50 ++- .../src/emulated-verifier/inbox-drain.ts | 83 +++++ .../src/emulated-verifier/inbox-observer.ts | 346 ++++++++++++++++++ .../src/emulated-verifier/inbox-processor.ts | 70 ++-- .../src/emulated-verifier/open-repo.ts | 11 +- .../polyfill/src/shared-wallet/bootstrap.ts | 22 ++ packages/polyfill/src/surface/subscribe.ts | 241 ++++++++++-- .../test/continuous-inbox-observation.test.ts | 340 +++++++++++++++++ .../test/deferred-inbox-processing.test.ts | 5 +- packages/polyfill/test/public-store.test.ts | 17 +- packages/polyfill/test/subscribe.test.ts | 116 +++++- packages/polyfill/test/wallet-fake.ts | 85 ++++- 14 files changed, 1284 insertions(+), 115 deletions(-) create mode 100644 .project/concepts/e2e-harness/_debt.md create mode 100644 packages/polyfill/src/emulated-verifier/inbox-drain.ts create mode 100644 packages/polyfill/src/emulated-verifier/inbox-observer.ts create mode 100644 packages/polyfill/test/continuous-inbox-observation.test.ts diff --git a/.gitignore b/.gitignore index ed006c6..41db391 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,6 @@ bun.lockb # SDK e2e real-broker harness — dedicated wallet profile + bundle (never committed) packages/polyfill/e2e/.dist/ *.ngw + +# Per-developer contract access map — canonical identities are committed, local paths are not +.project/contracts.local.yaml diff --git a/.project/concepts/e2e-harness/_debt.md b/.project/concepts/e2e-harness/_debt.md new file mode 100644 index 0000000..05a91b5 --- /dev/null +++ b/.project/concepts/e2e-harness/_debt.md @@ -0,0 +1,10 @@ +# Doc-debt — e2e-harness + +> Presence of a block = doc to update. Processed → delete the block; no blocks left → delete this file. +> One block = one "big change": `why` + `files` + `verify` (leaves to review). + +## Raw markers (consolidate into blocks, then delete) +- TOUCHED packages/polyfill/e2e/reactivity-doc-subscribe.ts @2026-08-17 (session f93872b5-293a-4916-a353-181409a96d42) +- TOUCHED packages/polyfill/e2e/repro-fresh-wallet.ts @2026-08-17 (session f93872b5-293a-4916-a353-181409a96d42) +- TOUCHED packages/polyfill/e2e/run.ts @2026-08-17 (session f93872b5-293a-4916-a353-181409a96d42) +- TOUCHED packages/polyfill/e2e/notebook.ts @2026-08-17 (session f93872b5-293a-4916-a353-181409a96d42) diff --git a/packages/polyfill/src/emulated-verifier/connect.ts b/packages/polyfill/src/emulated-verifier/connect.ts index e07aa91..5dd30b5 100644 --- a/packages/polyfill/src/emulated-verifier/connect.ts +++ b/packages/polyfill/src/emulated-verifier/connect.ts @@ -7,7 +7,7 @@ * like "the share did not work" rather than "nobody processed the queue". So the * moment an identity is connected ({@link setCurrentUser}), this runs. * - * Two steps, in order, and the order matters: + * Three steps, in order, and the order matters: * * 1. **Restore** — replay the durable registers back into what this user holds. All of * them: the Store branches of its three stores (the emulated `AddRepo { read_cap }` — @@ -16,9 +16,13 @@ * a handful of reads, no inbox needed. * 2. **Process** — drain the user's inbox (`inbox.processInbox`), which files any * new Link durably and puts it among what the user holds. + * 3. **Keep applying** — start watching those same inboxes, for as long as this identity + * stays connected (`emulated-verifier/inbox-observer.ts`). Step 2 is the backlog; + * this is the regime. * * Restoring first means a reconnecting user can read its documents immediately, without - * waiting on the inbox round-trip. + * waiting on the inbox round-trip; watching last means the backlog is applied before the + * first push arrives to apply it again. * * ── Restoring means ALL the registers, and that is a scar ───────────────── * Until 2026-08-16 this step read the Links and nothing else, so a fresh page put back the @@ -46,12 +50,21 @@ * both are answered by the same place — `AddInboxCap` records on the User branch * (`engine/repo/src/types.rs:1969`) — so `storeRegistry.myInboxes()` enumerates * them and this drains each in turn. + * + * ── And then it KEEPS going ─────────────────────────────────────────────── + * Step 2 drains the backlog; it does not end the obligation. Upstream the backlog is the + * exception (`from_queue`) and the rule is that a session applies each message as it + * arrives, so connecting ends by starting the continuous observation of the same inboxes + * (`emulated-verifier/inbox-observer.ts`), which lasts as long as this identity is + * connected. Until 2026-08-17 there was no step 3, and a deposit made while its recipient + * sat connected in front of the application converged only when that person reloaded. */ import { getCaps, getCurrentUser } from "../shared-wallet/bootstrap"; import { lookupAccount } from "../shared-wallet/account-registry"; import { accessLogPrefix, shortNuri } from "../shared-wallet/access-log"; import { myInboxes, readLinks, restoreOwnCaps } from "./branch-registers"; +import { startObservingInboxes } from "./inbox-observer"; import { processInbox } from "../surface/inbox"; import type { Nuri } from "../model/types"; @@ -126,7 +139,7 @@ const inFlight = new Map>(); * Not fixed with a retry, a timeout or a flag on purpose: deciding *what to do* about a * broker that cannot answer belongs to the caller, and it can only decide if it is told. */ -// @provenance connectedUser kind=aligned level=1 ref=engine/verifier/src/verifier.rs:inbox — upstream the verifier restores and processes with nothing for a caller to await; this is the awaitable form, for a deterministic start +// @provenance connectedUser kind=aligned level=1 ref=engine/verifier/src/verifier.rs:inbox — upstream the verifier restores, applies the queue handed over at connection and keeps applying what follows, with nothing for a caller to await; this is the awaitable form of the first two, for a deterministic start export async function connectedUser(): Promise { const holder = getCurrentUser(); if (holder === null) return; @@ -165,7 +178,18 @@ export async function connectedUser(): Promise { // success. That is how sharing broke once (`.project/concepts/sign-in/` // `knowledge_settling-is-not-connecting`), and conflating absence with ignorance is // the same fault `inbox.share` was fixed for on 2026-08-10. - if ((await lookupAccount(holder)) === null) return; + if ((await lookupAccount(holder)) === null) { + // Nothing to restore and no queue to drain — but this identity is CONNECTED, and it + // will acquire both during the session: provisioning is lazy here, so a person's very + // first visit connects with no stores at all and gets its account the moment the + // application creates anything. Watching from now on provisions nothing (`myInboxes` + // answers `[]` for an identity with no account) and is what makes that first session + // behave like every other one — the alternative was a brand-new user watched from + // their SECOND visit onwards, which is precisely the person most likely to be sent + // something. + if (stillConnected()) await startObservingInboxes(); + return; + } if (!stillConnected()) return; // 1. Durable first, and ALL of it — see the header. The documents this user MADE, from // the Store branches of its own stores, and then the caps it was GIVEN, from the @@ -202,6 +226,24 @@ export async function connectedUser(): Promise { reportUndrained(inbox, error); } } + // 3. …and from here on, KEEP applying. The backlog above is the special case, not the + // rule: upstream a session is handed each inbox message as it arrives and applies it + // inline, and only the messages waiting at connection are the "queue" + // (`from_queue`). Draining once and stopping made a deposit wait for the recipient + // to reload the page. See `emulated-verifier/inbox-observer.ts`. + // + // After the drain, not before: connecting owes the backlog first, and the + // observation subscribes to the same inboxes this loop just read. + // + // Unconditional on the loop's outcome, deliberately. An inbox that could not be + // drained is reported and denies nobody their session, and it must not deny them the + // observation of the OTHER inboxes either — nor of itself, since the next push is a + // fresh attempt at exactly the deposit that failed. + // Awaited, and it never rejects: what connecting starts, connecting finishes, so a + // caller that got its promise back knows the watching is in place — not merely + // requested. (It does not wait for the applying that watching then triggers.) + if (!stillConnected()) return; + await startObservingInboxes(); })(); inFlight.set(holder, run); diff --git a/packages/polyfill/src/emulated-verifier/inbox-drain.ts b/packages/polyfill/src/emulated-verifier/inbox-drain.ts new file mode 100644 index 0000000..17fea16 --- /dev/null +++ b/packages/polyfill/src/emulated-verifier/inbox-drain.ts @@ -0,0 +1,83 @@ +/** + * ONE drain at a time per inbox — the serializer both things that drain an inbox go through. + * + * Two of them exist in this package, and they are not coordinated by anything else: the + * CONTINUOUS observation of the connected identity's own inboxes + * (`emulated-verifier/inbox-observer.ts`) and the deferred timer that stands in for an + * ABSENT owner (`emulated-verifier/inbox-processor.ts`). Both can be pointed at the same + * inbox in the same page — a deposit into an inbox whose owner is also connected here arms + * the timer AND pushes to the observation. + * + * Draining writes what it applies (`branch-registers.addLink`, on the owner's User branch), + * so two passes over one queue interleave their reads and writes and apply the same records + * twice. Sequencing them is the same reasoning `connect.connectedUser` drains its inboxes + * one after another for, and it belongs HERE rather than in either caller: an invariant that + * holds only while both callers remember it is not an invariant. + * + * It is a queue, not a lock: a run that arrives while another is going is not dropped, it + * follows. Dropping would be wrong — the second run exists because something new landed. + */ + +import { accessLogPrefix, shortNuri } from "../shared-wallet/access-log"; +import type { Nuri } from "../model/types"; + +/** Drains in flight or queued, per inbox — the chain a new run appends itself to. */ +const running = new Map>(); + +/** + * Where a drain's failure goes. + * + * It cannot throw: nothing awaits these — neither the timer nor a push is a caller — and an + * unhandled rejection would take down whatever runtime it is in over work the application + * never requested. But "nowhere" is not the alternative. A drain that swallows its failure + * is indistinguishable from one that succeeded, which is precisely the defect family removed + * from this package on 2026-08-13, and a CONTINUOUS drain that swallowed would repeat that + * silence for as long as the session lasts. So it goes to the package's own log stream — the + * same `console.error` with the same `[][polyfill]` prefix `startConnect` and + * `inbox.watch` use — and it is NOT gated by the access-log flag: a diagnostic may be + * opt-in, a failure may not. + * + * One failed item denies nothing, here as at connection: the inbox is not consumed by + * failing, the deposit stays in it, and the next push or the next connection applies it. + */ +function reportFailure(inbox: Nuri, error: unknown): void { + console.error( + accessLogPrefix() + " could not apply what is in this inbox — its deposits stay in it " + + "and the next attempt will try again: " + shortNuri(inbox) + ":", + error, + ); +} + +/** + * Run `process` for `inbox`, behind any run of the same inbox still in flight. + * + * Resolves when THIS run has finished (or failed, having reported itself), so a caller + * that wants to wait can — while a caller that does not simply drops the promise. + */ +// @provenance drainInboxSerially kind=aligned level=1 ref=engine/verifier/src/verifier.rs:inbox — upstream one verifier owns an inbox and applies its messages one at a time as they arrive; this is that serialization, for the two paths this package drains from +export function drainInboxSerially(inbox: Nuri, process: () => Promise): Promise { + const previous = running.get(inbox) ?? Promise.resolve(); + const next = previous + // A previous run that FAILED must not cancel this one: its failure was reported where + // failures go, and the deposits it did not apply are exactly what this run is for. + .catch(() => undefined) + .then(process) + .catch((error: unknown) => reportFailure(inbox, error)) + .then(() => { + if (running.get(inbox) === next) running.delete(inbox); + }); + running.set(inbox, next); + return next; +} + +/** + * Wait until no drain is in flight. The seam a suite needs to observe convergence without + * waiting for it — and the one `runScheduledInboxProcessingNow` awaits after firing its + * windows. Loops, because a drain can be chained behind the one being awaited. + */ +// @provenance drainsSettled kind=invention level=none ref=none — a test/lifecycle seam over the queue above; upstream nothing exposes "is the verifier done applying" +export async function drainsSettled(): Promise { + while (running.size > 0) { + await Promise.all([...running.values()]); + } +} diff --git a/packages/polyfill/src/emulated-verifier/inbox-observer.ts b/packages/polyfill/src/emulated-verifier/inbox-observer.ts new file mode 100644 index 0000000..36e1f3c --- /dev/null +++ b/packages/polyfill/src/emulated-verifier/inbox-observer.ts @@ -0,0 +1,346 @@ +/** + * While an identity is connected, its inboxes are WATCHED and what arrives is APPLIED. + * + * ── The regime this restores, and the one it replaces ───────────────────── + * Upstream, applying an inbox is not something that happens at connection — it is what a + * session DOES. A sealed message reaches the recipient's own verifier as it arrives + * (`LocalBrokerMessage::Inbox` → `session.verifier.inbox(msg, from_queue)`, + * `sdk/rust/src/local_broker.rs`), which unseals it and applies it inline + * (`engine/verifier/src/inbox_processor.rs`); `from_queue` distinguishes the backlog handed + * over at connection from the messages that follow, and BOTH go through the same door. So + * the backlog is the special case, and the continuity is the rule. + * + * This package had emulated only the backlog. `inbox.processInbox` was called from exactly + * one place — `connect.connectedUser`, at connection — and nothing anywhere applied a + * deposit after that. A share deposited into an inbox whose owner was sitting connected in + * front of it converged only when that person reloaded the page, and the deposit's own + * session covered the gap by usurping the owner's identity on a twenty-second timer + * (`emulated-verifier/inbox-processor.ts`) — which does nothing at all if the depositor + * closes their tab, and tells the connected owner nothing either way. + * + * So: for as long as an identity is connected, every inbox it may read is subscribed to, + * and every push over one of them applies what is in it. + * + * ── Watching is not applying ────────────────────────────────────────────── + * `inbox.watch` looks like this and is not: it NOTIFIES an application that made a call and + * named one inbox, and it applies nothing. This applies, on the whole set, without anybody + * asking — because processing an inbox is the library's job and not the app's, which is the + * same ruling `connect.ts` opens with. + * + * ── Subscription, never a poll ──────────────────────────────────────────── + * Every half is push-driven. The inboxes themselves are subscribed to individually. WHICH + * inboxes exist is itself a subscription, on two channels that do not overlap: + * + * - the REGISTER — an inbox opened mid-session appends an `AddInboxCap` record to the + * User branch of the private store (`branch-registers.openDocumentInbox`), so a push on + * that document re-enumerates `myInboxes()`; + * - WHAT THIS IDENTITY HOLDS (`CapRegistry.onChange`) — the same signal `watchShape` + * re-reads on. It answers the case the register cannot: an identity that connects + * before it has an account at all. Provisioning is lazy here, so a person's FIRST visit + * connects with no stores, no private store to follow and no inbox to watch; the + * account appears later, the moment the application creates anything, and filing its + * caps is what says so. Without this channel a brand-new user was watched from their + * SECOND visit onwards — which is exactly the person most likely to be sent something. + * + * So a document whose inbox this identity opened five minutes into the session is watched + * like the rest, and so is the very first inbox of a person who had none when they arrived. + * + * ── It belongs to ONE holder, and stops when that holder does ───────────── + * Everything below resolves the CURRENT holder when it runs — `myInboxes`, `processInbox` + * and everything under them ask `getCurrentUser()` at the moment they execute. So an + * observation started for Alice must not still be firing when Bob is connected: it would + * read Alice's registers under Bob, or file Alice's capabilities into Bob's ring. The + * observation is therefore stamped with its holder, every step re-checks it, and switching + * identity (or disconnecting) tears it down — the same rule, and the same reason, + * `connect.connectedUser` carries `stillConnected()` for. + */ + +import { getCaps, getCurrentUser } from "../shared-wallet/bootstrap"; +import { accessLogPrefix, logStage, shortNuri } from "../shared-wallet/access-log"; +import { myInboxes } from "./branch-registers"; +import { drainInboxSerially, drainsSettled } from "./inbox-drain"; +import { lookupAccount } from "../shared-wallet/account-registry"; +import { processInbox } from "../surface/inbox"; +import { subscribeDoc, type Unsubscribe } from "../surface/subscribe"; +import type { Nuri, PrincipalId } from "../model/types"; + +/** + * One identity's live observation. Held whole rather than as loose module variables so + * that a step started under it can ask "am I still the current one" by IDENTITY of the + * object, not by comparing a name that a re-connection may have restored in between. + */ +interface Observation { + /** WHO this observation belongs to. Every step re-checks it before acting. */ + holder: PrincipalId; + /** One subscription per inbox observed, keyed by it. */ + inboxes: Map; + /** The subscription on the register that says WHICH inboxes exist. */ + register: Unsubscribe | null; + /** Unsubscribe from the held-caps change signal — the second "which inboxes" channel. */ + holdings: Unsubscribe | null; + /** True while a re-enumeration is running, so its own effects do not restart it. */ + enumerating: boolean; + /** A trigger that arrived mid-cycle: the running one repeats once rather than lose it. */ + enumerateAgain: boolean; + /** Enumerations and applications in flight — what {@link observationSettled} waits on. */ + pending: Set>; +} + +let observation: Observation | null = null; + +/** + * Where a failure to WATCH goes — distinct from a failure to APPLY, which + * `inbox-drain.ts` reports. + * + * Not a rejection: nothing awaits this work, exactly as at connection. And not silence, for + * the same reason as everywhere else in this package — an observation that cannot be + * established leaves deposits unapplied for the whole session, and the only thing a person + * would notice is that a share never arrives. + */ +function reportUnobserved(what: string, error: unknown): void { + console.error( + accessLogPrefix() + " connected, but " + what + " — deposits made from now on may not be " + + "applied until the next connection:", + error, + ); +} + +/** Is `obs` still the live observation, for the identity it belongs to? */ +function current(obs: Observation): boolean { + return observation === obs && getCurrentUser() === obs.holder; +} + +/** Track a piece of in-flight work so {@link observationSettled} can wait for it. */ +function track(obs: Observation, work: Promise): void { + obs.pending.add(work); + void work.finally(() => obs.pending.delete(work)); +} + +/** + * Apply whatever is in `inbox`, for the holder this observation belongs to. + * + * Through the shared per-inbox queue, so this cannot interleave with a drain the deferred + * timer started on the same inbox (nor with another push of its own): two passes over one + * queue apply the same records twice. A failure is reported there and stops nothing — the + * next push over this inbox tries again, and the other inboxes were never involved. + */ +async function applyWhatArrived(obs: Observation, inbox: Nuri): Promise { + if (!current(obs)) return; + await drainInboxSerially(inbox, async () => { + // Re-checked INSIDE the queue: this run may have waited behind another one, and the + // identity can have moved while it waited. `processInbox` resolves the holder at each + // of its steps, so running it for the wrong one is not a near-miss — it reads someone + // else's registers and files into someone else's ring. + if (!current(obs)) return; + try { + await processInbox(inbox); + } catch (error) { + // An identity that moved MID-DRAIN throws here too, and it is the ordinary case rather + // than an exotic one: `processInbox` resolves the holder at each step, so the new + // holder is refused an inbox that is not theirs somewhere in the middle. That is + // ABANDONING — which this package has always called "not a failure" — and reporting it + // would put a broker-looking error in the log every time a page switches user. + // Nothing is lost: an inbox is not consumed by being abandoned, so the deposit is + // still there for its owner's next connection. Same rule, same words, as the drain + // loop in `connect.connectedUser`. + if (!current(obs)) return; + throw error; + } + }); +} + +/** + * (Re-)enumerate the inboxes this identity may read and subscribe to the ones not yet + * watched. Idempotent: an inbox already observed is left alone, so a re-enumeration costs + * one read and nothing else. + * + * Never removes: the list only grows within a session (an `AddInboxCap` record is durable), + * and dropping a subscription on a guess would silently stop applying an inbox. + */ +async function watchTheInboxes(obs: Observation): Promise { + if (!current(obs)) return; + let inboxes: Nuri[]; + try { + inboxes = await myInboxes(); + } catch (error) { + // Not knowing WHICH inboxes exist is the whole observation failing, not one queue. + reportUnobserved("the inboxes to watch could not be listed", error); + return; + } + if (!current(obs)) return; + for (const inbox of inboxes) { + if (obs.inboxes.has(inbox)) continue; + try { + // The push is the signal; the drain is the work. The FIRST push is the initial + // `State`, which drains an inbox connection has usually just drained — idempotent, + // and the alternative (skip the first) would lose a deposit that landed in the gap + // between the two. + obs.inboxes.set( + inbox, + subscribeDoc(inbox, () => track(obs, applyWhatArrived(obs, inbox))), + ); + logStage("OBSERVING " + shortNuri(inbox) + " for " + obs.holder); + } catch (error) { + reportUnobserved("this inbox could not be watched: " + shortNuri(inbox), error); + } + } +} + +/** + * Follow the register that says which inboxes exist, so one opened MID-SESSION is picked + * up. See the module header: `openDocumentInbox` appends its record to the User branch of + * the private store, and a same-session write pushes to its own subscribers. + * + * Idempotent, and RETRIED rather than given up on: an identity that connects before it has + * an account has no private store to follow yet, and acquires one the first time the + * application creates anything. Re-entered from the holdings signal, it catches up then. + */ +async function watchTheRegister(obs: Observation): Promise { + if (!current(obs) || obs.register !== null) return; + let store: Nuri | undefined; + try { + // `lookupAccount`, not `resolveAccount`: the tolerant form answers `null` for a read + // that FAILED exactly as for one that found nothing, and the two could not mean more + // different things here — "this identity has no account yet" is a state to wait + // quietly through, "the broker did not answer" is a failure to report. The same + // distinction `connect.connectedUser` and `inbox.share` are built on. + store = (await lookupAccount(obs.holder))?.docPrivate; + } catch (error) { + reportUnobserved("the register of your inboxes could not be reached", error); + return; + } + // No account yet — nothing to follow, and asking for one would PROVISION it, which + // connecting deliberately does not do (`connect.connectedUser`). The holdings signal + // brings us back here the moment there is something to follow. + if (!store || !current(obs) || obs.register !== null) return; + try { + obs.register = subscribeDoc(store, () => track(obs, enumerate(obs))); + } catch (error) { + reportUnobserved("an inbox opened later in this session will not be watched", error); + } +} + +/** + * The whole "which inboxes am I watching" cycle: catch up on the register, then on the + * inboxes. Both halves are idempotent, so this is what every trigger runs. + * + * A trigger that arrives while a cycle is running does not start a second one — the cycle + * FILES caps for the inboxes it lists (`myInboxes` → `fileOwnInbox`), which is one of the + * signals that runs it, so overlapping runs would pile up on their own effects. It is + * REMEMBERED instead of dropped: the running cycle repeats once more at the end, because a + * signal that arrived mid-cycle may be about something that cycle had already read past. + * + * That terminates rather than ping-ponging: filing a cap this identity already holds fires + * nothing (`caps.file` notifies only on a genuinely new one), so the repeat that finds + * nothing new sets no flag and the loop ends. + */ +async function enumerate(obs: Observation): Promise { + if (!current(obs)) return; + if (obs.enumerating) { + obs.enumerateAgain = true; + return; + } + obs.enumerating = true; + try { + do { + obs.enumerateAgain = false; + await watchTheRegister(obs); + await watchTheInboxes(obs); + } while (obs.enumerateAgain && current(obs)); + } finally { + obs.enumerating = false; + } +} + +/** + * Start observing the connected identity's inboxes. Called at the end of the connection + * work (`connect.connectedUser`), and idempotent: calling it again for the identity already + * being observed changes nothing. + * + * ── It resolves when the WATCHING is in place, not when the applying is done ── + * Connecting publishes one guarantee — *the work it starts, it finishes* — and "your + * inboxes are now being watched" is part of that work, so this is awaited rather than + * fired. What it does NOT wait for is the applying: subscribing pushes an initial `State`, + * which drains each inbox again, and making a person's sign-in wait on a second pass over + * a backlog `connectedUser` has just applied would buy nothing. + * + * It never rejects. Failing to WATCH is reported where failures go and denies nobody their + * session — the same rule the drain loop above it follows, and for the same reason: what + * has not been applied stays in its queue for the next attempt. + */ +// @provenance startObservingInboxes kind=aligned level=1 ref=engine/verifier/src/verifier.rs:inbox — upstream a connected verifier is handed each inbox message as it arrives and applies it; here the arrival signal is a document push, and the set of inboxes is the User branch's `AddInboxCap` records +export async function startObservingInboxes(): Promise { + const holder = getCurrentUser(); + if (holder === null) return; + if (observation !== null && observation.holder === holder) return; + stopObservingInboxes(); + const obs: Observation = { + holder, + inboxes: new Map(), + register: null, + holdings: null, + enumerating: false, + enumerateAgain: false, + pending: new Set(), + }; + observation = obs; + // The second "which inboxes exist" channel — see the module header. It is what makes a + // person's FIRST session watched at all: they connect holding nothing, and their account + // (with its inboxes) appears the moment the application creates something. + obs.holdings = getCaps().onChange(() => track(obs, enumerate(obs))); + const setup = enumerate(obs); + track(obs, setup); + await setup; +} + +/** + * Stop observing — on an identity change, on disconnection, and when the library is + * un-configured. + * + * Work already inside the drain queue is not cancelled (nothing can un-issue a write + * already sent); it is neutralised instead, because every step re-checks that its + * observation is still the current one and returns rather than filing for the wrong holder. + */ +// @provenance stopObservingInboxes kind=aligned level=1 ref=engine/verifier/src/verifier.rs:Verifier — a verifier's inbox processing lives and dies with its session; switching identity is another session, so nothing of the previous one keeps applying +export function stopObservingInboxes(): void { + const obs = observation; + if (obs === null) return; + observation = null; + for (const unsub of obs.inboxes.values()) { + try { + unsub(); + } catch (error) { + console.error(accessLogPrefix() + " stopping an inbox observation failed:", error); + } + } + obs.inboxes.clear(); + for (const unsub of [obs.register, obs.holdings]) { + if (!unsub) continue; + try { + unsub(); + } catch (error) { + console.error(accessLogPrefix() + " stopping an inbox observation failed:", error); + } + } + obs.register = null; + obs.holdings = null; +} + +/** + * Wait until the observation has nothing in flight — the enumerations it is running and the + * drains they started. + * + * A seam for a suite that has to observe convergence without waiting for it, in the same + * spirit as `runScheduledInboxProcessingNow`: it names no inbox and starts no work, so it + * can only wait for what a push has already caused. Never exported from the package. + */ +// @provenance observationSettled kind=invention level=none ref=none — a test seam over the observation; upstream nothing exposes "has the verifier finished applying" +export async function observationSettled(): Promise { + for (;;) { + const obs = observation; + const inFlight = obs ? [...obs.pending] : []; + if (inFlight.length === 0) break; + await Promise.all(inFlight); + } + await drainsSettled(); +} diff --git a/packages/polyfill/src/emulated-verifier/inbox-processor.ts b/packages/polyfill/src/emulated-verifier/inbox-processor.ts index 03ed1aa..d16ae3f 100644 --- a/packages/polyfill/src/emulated-verifier/inbox-processor.ts +++ b/packages/polyfill/src/emulated-verifier/inbox-processor.ts @@ -1,18 +1,26 @@ /** - * The inbox-processing SERVICE this deployment has not got — emulated by a timer. + * A stand-in for an inbox whose owner is NOT HERE — emulated by a timer. * * ── What is missing, and what stands in for it ──────────────────────────── * Upstream a deposit does not wait for anybody: the broker routes the sealed message * (`inboxes: PubKey → RepoId`, `engine/verifier/src/verifier.rs:105,1677`) and the - * recipient's verifier applies it when it processes the inbox - * (`engine/verifier/src/inbox_processor.rs`). A real NextGraph deployment will have - * something running continuously that does this; today, in this library, the ONLY thing - * that drains an inbox is its owner connecting (`connect.connectedUser` → - * `inbox.processInbox`), which can be hours away. + * RECIPIENT's own verifier applies it as it arrives, for as long as that recipient has a + * session (`LocalBrokerMessage::Inbox` → `session.verifier.inbox(…)`, + * `sdk/rust/src/local_broker.rs`). Two different things follow from that, and this module + * is only one of them: + * + * - a connected identity's own inboxes are observed and applied continuously, which is + * `emulated-verifier/inbox-observer.ts` and needs no timer at all; + * - an identity that is NOT connected has nobody to apply anything for it. Upstream the + * broker queues the message until it comes back; here, on ONE shared wallet, the + * depositor's own page can stand in for it. * * So: a successful deposit arms a ONE-SHOT timer that processes the TARGET's inbox, twenty * seconds later. Alice deposits into Bob's inbox; twenty seconds later this package drains - * Bob's inbox, while the connected session is still Alice's. + * Bob's inbox, while the connected session is still Alice's. When Bob IS connected in this + * page, the observation has already applied it long before the window closes and the timer + * finds nothing left to do — harmless (draining is idempotent) and not worth a special + * case: the two answer different questions, and only this one answers the absent owner. * * ── That is identity usurpation, and it is deliberate ───────────────────── * It is only possible because one shared wallet holds every virtual identity, and it is @@ -32,7 +40,8 @@ * the real path, not a repair. */ -import { accessLogPrefix, logStage, shortNuri } from "../shared-wallet/access-log"; +import { logStage, shortNuri } from "../shared-wallet/access-log"; +import { drainInboxSerially, drainsSettled } from "./inbox-drain"; import type { Nuri } from "../model/types"; /** @@ -62,13 +71,6 @@ interface Scheduled { */ const scheduled = new Map(); -/** - * Drains in flight, per inbox — so a run that starts while another is still going CHAINS - * behind it instead of interleaving with it. Same reason `connect.connectedUser` drains its - * inboxes sequentially: two concurrent passes over one queue apply the same records twice. - */ -const running = new Map>(); - /** * Keep a pending drain from holding a runtime open. * @@ -83,38 +85,14 @@ function releaseFromTheEventLoop(handle: ReturnType): void { } /** - * Where a deferred drain's failure goes. - * - * It cannot throw: nothing awaits it — the application never asked for this work, and an - * unhandled rejection would take down whatever runtime it is in over a drain it did not - * request. But "nowhere" is not the alternative. A drain that swallows its failure is - * indistinguishable from one that succeeded, which is precisely the defect family removed - * from this package on 2026-08-13. So it goes to the package's own log stream — the same - * `console.error` with the same `[][polyfill]` prefix that `startConnect` and - * `inbox.watch` use for their un-awaitable failures, and it is NOT gated by the access-log - * flag: a diagnostic may be opt-in, a failure may not. + * Run the drain of `inbox`, behind any run of the same inbox still in flight — including a + * run the CONTINUOUS observation started, which is why the queue lives in + * `emulated-verifier/inbox-drain.ts` rather than here. That module also owns where a + * drain's failure goes; it cannot travel back to a caller, because a timer has none. */ -function reportFailure(inbox: Nuri, error: unknown): void { - console.error( - accessLogPrefix() + " deferred inbox processing failed for " + shortNuri(inbox) + ":", - error, - ); -} - -/** Run the drain of `inbox`, behind any run of the same inbox still in flight. */ function fire(inbox: Nuri, process: () => Promise): void { scheduled.delete(inbox); - const previous = running.get(inbox) ?? Promise.resolve(); - const next = previous - // A previous run that FAILED must not cancel this one: its failure was reported where - // failures go, and the deposits it did not apply are exactly what this run is for. - .catch(() => undefined) - .then(process) - .catch((error: unknown) => reportFailure(inbox, error)) - .then(() => { - if (running.get(inbox) === next) running.delete(inbox); - }); - running.set(inbox, next); + void drainInboxSerially(inbox, process); } /** @@ -126,7 +104,7 @@ function fire(inbox: Nuri, process: () => Promise): void { * * Idempotent inside the window: a second deposit into the same inbox joins the pending run. */ -// @provenance scheduleInboxProcessing kind=divergent level=1 ref=engine/verifier/src/verifier.rs:inbox — upstream the RECIPIENT's own verifier applies its inbox; here another identity's session drains it on a timer. Deliberate, unpublished, and only possible on one shared wallet +// @provenance scheduleInboxProcessing kind=divergent level=1 ref=engine/verifier/src/verifier.rs:inbox — upstream the RECIPIENT's own verifier applies its inbox and the broker queues for it while it is away; here another identity's session drains it on a timer, for a recipient that is not connected in this page (one that IS has its own observation). Deliberate, unpublished, and only possible on one shared wallet export function scheduleInboxProcessing(inbox: Nuri, process: () => Promise): void { if (scheduled.has(inbox)) return; const handle = setTimeout(() => fire(inbox, process), PROCESSING_DELAY_MS); @@ -152,7 +130,7 @@ export async function runScheduledInboxProcessingNow(): Promise { clearTimeout(entry.handle); fire(inbox, entry.process); } - await Promise.all([...running.values()]); + await drainsSettled(); } /** diff --git a/packages/polyfill/src/emulated-verifier/open-repo.ts b/packages/polyfill/src/emulated-verifier/open-repo.ts index 398d7f6..476c8d1 100644 --- a/packages/polyfill/src/emulated-verifier/open-repo.ts +++ b/packages/polyfill/src/emulated-verifier/open-repo.ts @@ -74,7 +74,7 @@ import { mustNotAttempt } from "./reach"; import { fetchReadCap } from "./public-store"; import { getConfig, getStoreRegistryDeps } from "../shared-wallet/bootstrap"; -import { subscribeDocUnguarded, type Unsubscribe } from "../surface/subscribe"; +import { resubscribeDocs, subscribeDocUnguarded, type Unsubscribe } from "../surface/subscribe"; import { logStage, shortNuri } from "../shared-wallet/access-log"; import type { Nuri } from "../model/types"; @@ -137,6 +137,15 @@ export function resetOpenedRepos(): void { syncState.clear(); boundSessionId = null; OPEN_TIMEOUT_MS = 8000; + // Whatever ELSE was still subscribed to a document — a `watchShape` following a scope, + // an inbox observation — belongs to the session that just went away too. One real + // subscription now serves all of them (`surface/subscribe.ts`), so dropping the bootstrap + // ones above no longer closes those channels; they are re-opened against the new session, + // with their barrier forgotten. Without this a bootstrap open would join a surviving + // fan-out, be handed the PREVIOUS session's `State`, and call a repo the new session has + // never synced "synced". After the held teardown, so a document nobody else follows is + // simply released. + resubscribeDocs(); } /** diff --git a/packages/polyfill/src/shared-wallet/bootstrap.ts b/packages/polyfill/src/shared-wallet/bootstrap.ts index ded013b..5b4ce81 100644 --- a/packages/polyfill/src/shared-wallet/bootstrap.ts +++ b/packages/polyfill/src/shared-wallet/bootstrap.ts @@ -25,6 +25,8 @@ import { setAccessLog } from "./access-log"; import { inspectOutbox } from "./outbox-log"; import { startConnect } from "../emulated-verifier/connect"; import { cancelScheduledInboxProcessing } from "../emulated-verifier/inbox-processor"; +import { stopObservingInboxes } from "../emulated-verifier/inbox-observer"; +import { resetDocSubscriptions } from "../surface/subscribe"; import { resetSharedWalletSession, sharedWalletSession } from "./session"; /** @@ -191,6 +193,13 @@ let caps = new CapRegistry(capsHolder); // @provenance configure kind=invention level=none ref=none — same — the whole call disappears at migration, when the consumer initialises the real SDK directly export function configure(c: EventuallyConfig): void { + // Whatever the PREVIOUS SDK was driving stops here. A document subscription belongs to + // the `ng` that opened it and an inbox observation to the identity that was connected + // through it, so carrying either into a fresh configuration would leave listeners + // attached to a verifier nobody is talking to any more — and, on the observation, work + // that files for an identity this call is about to clear. + resetDocSubscriptions(); + stopObservingInboxes(); cfg = c; // Not taken from the config: an application never supplies its own identity — upstream // it comes FROM the wallet a person opened. Accepting one here would also let a caller @@ -224,6 +233,12 @@ export function resetConfig(): void { // no longer has an `ng` to read or write with, and report a failure about a session that // no longer exists. See `emulated-verifier/inbox-processor.ts`. cancelScheduledInboxProcessing(); + // …and so does the continuous observation, for the same reason and one more: it is a set + // of live document subscriptions, which the un-configured library has no `ng` to cancel + // through. Both go before the subscriptions themselves, so nothing is left holding a + // torn-down fan-out. + stopObservingInboxes(); + resetDocSubscriptions(); // The hand-over goes with it, for the same reason: it is made of the config's injected // `init`, so leaving it behind would let a revived barrier delegate to the PREVIOUS // application's SDK. @@ -350,6 +365,13 @@ export function adoptCurrentUser(id: PrincipalId | null): boolean { */ export function setCurrentUser(id: PrincipalId | null): void { const changed = adoptCurrentUser(id); + // The PREVIOUS identity stops being watched the moment it stops being connected — + // including on `null`, which is a disconnection and not merely "no new work". An + // observation applies what arrives by resolving the current holder at each step, so one + // left running past a switch would read the old identity's registers under the new one + // and file the old one's capabilities into the new one's ring. Whoever connects next + // starts their own. See `emulated-verifier/inbox-observer.ts`. + if (changed) stopObservingInboxes(); // Connecting a user is what triggers inbox processing — the library's job, not // the app's. Fire-and-forget: this setter is synchronous and every consumer calls // it from synchronous code, so the work announces itself through the cap diff --git a/packages/polyfill/src/surface/subscribe.ts b/packages/polyfill/src/surface/subscribe.ts index 4fc4e95..684f003 100644 --- a/packages/polyfill/src/surface/subscribe.ts +++ b/packages/polyfill/src/surface/subscribe.ts @@ -31,6 +31,36 @@ * 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. + * + * ── ONE real subscription per document, fanned out here ──────────────────── + * A branch has room for exactly ONE subscriber upstream, and a second `doc_subscribe` + * does not join it — it EVICTS the first. `branch_subscriptions: HashMap>` holds one sender per branch, and `create_branch_subscription` + * inserts into it and closes whatever it displaced + * (`engine/verifier/src/verifier.rs:create_branch_subscription`), for the document's own + * branch AND for its Header branch. + * + * Verified against the real broker on 2026-08-17: with two `subscribeDoc` calls on one + * document, a write fired the SECOND callback (`Patch`) and the first stopped firing + * entirely — after having fired on a write moments earlier. Silently: nothing rejects, + * nothing logs, the stale unsubscribe still "works". + * + * That made every internal subscriber a hazard to every other one. `ensureRepoOpen` + * holds a bootstrap subscription per document for the whole session, `watchShape` + * subscribes to the documents of a scope, `inbox.watch` to an inbox — so opening a + * repo killed the watch on it, and the app-visible result was a view that never + * re-read and an inbox that never notified. Both were reported as "the layer does not + * notify of its own actions", which is NOT what happens: a same-session write DOES + * push (verified the same day — `Patch@69ms` on the writer's own `sparqlUpdate`, + * `e2e/reactivity-doc-subscribe.ts`). The push arrived; there was no longer anybody + * on the other end. + * + * So this module keeps at most one real subscription per NURI and fans its pushes out + * to every local listener. A listener that joins after the initial `State` is replayed + * the remembered one, because that is what its own `doc_subscribe` would have handed + * it — without the replay, joining a document somebody else already opened would never + * fire, and `inbox.watch`'s "fires once immediately" would silently stop being true. + * The real subscription is torn down when the LAST listener leaves. */ import { getConfig, getStoreRegistryDeps } from "../shared-wallet/bootstrap"; @@ -87,6 +117,149 @@ async function sessionId(): Promise { return (await getStoreRegistryDeps().getSession()).sessionId; } +// --- one real subscription per document ----------------------------------- + +/** What a listener is handed on every push. */ +type Listener = (r: DocChange, type: DocChangeType) => void; + +/** + * The single real `doc_subscribe` behind every local listener on one document. + * See the module header for why there can only be one. + */ +interface DocFanOut { + /** Every local listener on this document. The last one to leave tears it down. */ + listeners: Set; + /** The platform's unsubscribe, once the async setup resolved. */ + realUnsub: (() => void) | null; + /** True from the moment setup is kicked off — a later joiner must not kick off a second. */ + establishing: boolean; + /** + * The most recent `State` push, replayed to a listener that joins later. + * Its own `doc_subscribe` would have pushed one; the fan-out owes it the same. + */ + lastState: { resp: DocChange; type: DocChangeType } | null; +} + +const fanOuts = new Map(); + +/** Hand one push to one listener, isolating a throwing handler from the others. */ +function deliver(nuri: Nuri, listener: Listener, resp: DocChange, type: DocChangeType): void { + try { + listener(resp, type); + } catch (error) { + console.error("[subscribe] onChange handler threw for", nuri, error); + } +} + +/** Drop a fan-out: forget the remembered state and release the real subscription. */ +function releaseFanOut(nuri: Nuri, entry: DocFanOut): void { + if (fanOuts.get(nuri) === entry) fanOuts.delete(nuri); + entry.lastState = null; + entry.establishing = false; + const unsub = entry.realUnsub; + entry.realUnsub = null; + if (!unsub) return; + try { + unsub(); + } catch (error) { + console.error("[subscribe] unsubscribe failed for", nuri, error); + } +} + +/** + * Open the one real subscription for `entry` and route its pushes to every listener. + * Errors are isolated to this document (they never reject a shared batch — see + * {@link subscribeDocs}); a failed setup simply leaves the document silent. + */ +async function establish(nuri: Nuri, entry: DocFanOut, ng: { doc_subscribe?: unknown }): Promise { + // No reactive primitive on the injected `ng` (the fake in the unit suite): there is + // nothing to call, so this document simply never pushes — the same documented no-op + // `openRepoUnguarded` takes for the same injection, and not a failure to report. A real + // `@ng-org/web` always exposes it. + if (typeof ng.doc_subscribe !== "function") return; + const doc_subscribe = ng.doc_subscribe as (...a: unknown[]) => unknown; + try { + const sid = await sessionId(); + const unsub = (await doc_subscribe(nuri, sid, (resp: DocChange): void => { + // A push that arrives after the last listener left belongs to a torn-down + // fan-out — the real unsubscribe may not have taken effect yet. + if (fanOuts.get(nuri) !== entry) return; + const type = docChangeType(resp); + // Remember the barrier for whoever joins next; a later `State` replaces it. + if (type === "State") entry.lastState = { resp, type }; + // A copy: a handler may unsubscribe itself (or another) from inside the push. + for (const listener of [...entry.listeners]) deliver(nuri, listener, resp, type); + })) as (() => void) | undefined; + if (fanOuts.get(nuri) !== entry || entry.listeners.size === 0) { + // Everyone left (or the fan-out was reset) before setup resolved — cancel now. + if (typeof unsub === "function") unsub(); + return; + } + entry.realUnsub = typeof unsub === "function" ? unsub : null; + } catch (error) { + console.error("[subscribe] doc_subscribe failed for", nuri, error); + } +} + +/** + * Drop every fan-out without notice — what a page that re-`configure()`s does. + * + * A subscription belongs to the injected SDK that opened it, so when that is replaced + * (or removed) its live subscriptions are void: their pushes would come from a verifier + * nobody is talking to any more. Called by `configure`/`resetConfig`, and by nothing on + * the reactive path — a listener is never dropped while its SDK is still there. + */ +// @provenance resetDocSubscriptions kind=invention level=none ref=none — per-session bookkeeping reset, like `resetOpenedRepos`; upstream a verifier owns its own subscriptions and nothing resets them from outside +export function resetDocSubscriptions(): void { + for (const [nuri, entry] of [...fanOuts]) { + entry.listeners.clear(); + releaseFanOut(nuri, entry); + } + fanOuts.clear(); +} + +/** + * Re-open every live subscription against the CURRENT session, keeping its listeners. + * + * Called when the session id changes under the page (`open-repo.resetOpenedRepos`), and + * this is the one thing the fan-out owes that a per-caller subscription got for free. A new + * session is a new verifier with an empty `self.repos`, so the channel behind a fan-out is + * dead — and its remembered `State` is the old verifier's, which would let a bootstrap open + * JOIN it, be handed that stale barrier at once, and read a repo the new session has never + * synced. Zero rows, "synced", no error: the silent staleness this module exists to remove. + * + * So the barrier is forgotten and the channel re-opened, rather than the listeners dropped: + * a `watchShape` that was following a document keeps following it across the change. + */ +// @provenance resubscribeDocs kind=invention level=none ref=none — per-session bookkeeping, like `resetOpenedRepos`; upstream a session's subscriptions die with it and nothing carries them over +export function resubscribeDocs(): void { + // Un-configured (a torn-down suite): there is no SDK to re-open anything against, and + // `resetDocSubscriptions` has already emptied this map on that path. + let ng: { doc_subscribe?: unknown }; + try { + ng = getConfig().ng as { doc_subscribe?: unknown }; + } catch { + return; + } + for (const [nuri, entry] of [...fanOuts]) { + if (entry.listeners.size === 0) continue; + const stale = entry.realUnsub; + entry.realUnsub = null; + // The barrier belonged to the session that is gone. Whoever joins next waits for a + // real one, exactly as they would have on a fresh page. + entry.lastState = null; + if (stale) { + try { + stale(); + } catch (error) { + console.error("[subscribe] unsubscribe failed for", nuri, error); + } + } + entry.establishing = true; + void establish(nuri, entry, ng); + } +} + /** * 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 @@ -106,7 +279,7 @@ async function sessionId(): Promise { * * Calls the REAL injected `ng.doc_subscribe` directly (never `makeNg`). */ -// @provenance subscribeDoc kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — two deliberate deltas: the unsubscribe is returned synchronously, and the callback gets a pre-extracted `type` +// @provenance subscribeDoc kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — three deliberate deltas: the unsubscribe is returned synchronously, the callback gets a pre-extracted `type`, and N callers of ONE document share ONE upstream subscription because a branch holds a single sender and a second subscribe evicts the first export function subscribeDoc( nuriLike: NuriLike, onChange: (r: DocChange, type: DocChangeType) => void, @@ -123,53 +296,49 @@ export function subscribeDoc( * owns the machinery's entire privileged door — and for nobody else. It is not * re-exported by either entry point; the `Unguarded` suffix is the warning. */ -// @provenance subscribeDocUnguarded kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — the same call without the reach guard; internal, never published +// @provenance subscribeDocUnguarded kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — the same call without the reach guard, and the place the one-subscription-per-document fan-out is kept; internal, never published export function subscribeDocUnguarded( nuri: Nuri, onChange: (r: DocChange, type: DocChangeType) => void, ): Unsubscribe { + // Resolved here, synchronously, so calling this before `configure()` still throws at + // the call rather than inside a background task nobody awaits. const { ng } = getConfig(); - let stopped = false; - let realUnsub: (() => void) | null = null; - const cb = (resp: DocChange): void => { - if (stopped) return; - try { - onChange(resp, docChangeType(resp)); - } catch (error) { - console.error("[subscribe] onChange handler threw for", nuri, error); - } - }; + let entry = fanOuts.get(nuri); + if (!entry) { + entry = { listeners: new Set(), realUnsub: null, establishing: false, lastState: null }; + fanOuts.set(nuri, entry); + } + const joined = entry; + joined.listeners.add(onChange); - // 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; + // Joining a document somebody else already opened: hand this listener the `State` its + // own `doc_subscribe` would have pushed it. Asynchronously, like the real push, so a + // caller that sets itself up after this call still sees it. + if (joined.lastState) { + const { resp, type } = joined.lastState; + queueMicrotask(() => { + if (fanOuts.get(nuri) === joined && joined.listeners.has(onChange)) { + deliver(nuri, onChange, resp, type); } - realUnsub = typeof unsub === "function" ? unsub : null; - } catch (error) { - console.error("[subscribe] doc_subscribe failed for", nuri, error); - } - })(); + }); + } + if (!joined.establishing) { + joined.establishing = true; + void establish(nuri, joined, ng as { doc_subscribe?: unknown }); + } + + let stopped = false; return () => { if (stopped) return; stopped = true; - if (realUnsub) { - try { - realUnsub(); - } catch (error) { - console.error("[subscribe] unsubscribe failed for", nuri, error); - } - realUnsub = null; - } + joined.listeners.delete(onChange); + // The LAST one out releases the real subscription — while anybody is still + // listening, tearing it down would silence them (and re-opening it later is not + // free: a second `doc_subscribe` evicts whoever else has the branch by then). + if (joined.listeners.size === 0) releaseFanOut(nuri, joined); }; } diff --git a/packages/polyfill/test/continuous-inbox-observation.test.ts b/packages/polyfill/test/continuous-inbox-observation.test.ts new file mode 100644 index 0000000..3ddecfa --- /dev/null +++ b/packages/polyfill/test/continuous-inbox-observation.test.ts @@ -0,0 +1,340 @@ +/** + * While I am connected, what lands in my inboxes is applied — no reload, nobody asked. + * + * ── The regime, and the one this replaces ───────────────────────────────── + * Upstream, applying an inbox is what a SESSION does: a sealed message reaches the + * recipient's own verifier as it arrives and is applied inline, and only the backlog handed + * over at connection is marked apart (`from_queue`). This package had emulated the backlog + * and nothing else — `inbox.processInbox` was called from exactly one place, at connection — + * so a share deposited while its recipient sat connected in front of the application + * converged when that person next RELOADED the page. Its stand-in was a twenty-second timer + * in the DEPOSITOR's session, which does nothing if that tab closes and tells the connected + * owner nothing either way. + * + * ── How a deposit gets here without a reconnection ──────────────────────── + * Bob makes his deposit through the published surface, under his own identity, naming a + * PERSON (`inbox.share(doc, "alice")`) — he is handed no address, and the test hands him + * none. What the test then does is what a broker does: it HOLDS what he wrote and delivers + * it to this page later, once Alice is the one connected (`wallet-fake._deliver`). The + * quads delivered are the ones the library itself produced under Bob; nothing is composed + * by hand. That is the cross-session case verified against the real broker + * (`e2e/reactivity-doc-subscribe.ts`): a second session's write reaches the first session's + * subscription as a `Patch`. + * + * ── A deposit by the owner is a REAL case, not a shortcut ───────────────── + * Two tests below have Alice deposit into her own document's inbox while she is connected. + * That is not a stand-in for a stranger: it is the case a consuming application reported, + * and a same-session write is pushed to that session's own subscription — verified against + * the real broker the same day (`Patch@69ms` on the writer's own `sparqlUpdate`). + */ + +import { test, expect, describe, afterAll, beforeEach } from "bun:test"; +import { inbox as inboxSurface, storeRegistry } from "../src/index"; +import { getCaps, setCurrentUser } from "../src/shared-wallet/bootstrap"; +import { resolveAccount, userInbox } from "../src/shared-wallet/account-registry"; +import { observationSettled } from "../src/emulated-verifier/inbox-observer"; +import { cancelScheduledInboxProcessing } from "../src/emulated-verifier/inbox-processor"; +import { bootPage, forgetEverything, signIn, type FakeWallet, type Quad } from "./wallet-fake"; +import type { Nuri } from "../src/model/types"; + +const SHIM = "urn:ng-eventually:shim"; +const INBOX = "urn:ng-eventually:inbox"; + +/** The reactive fake: a repo answers nothing until subscribed, and a commit pushes. */ +const COLD = { unsyncedUntilSubscribed: true } as const; + +let quads: Quad[]; +let fake: FakeWallet; + +function boot(): void { + quads = []; + fake = bootPage(quads, COLD); +} + +/** + * Let the page's pushes land, and everything they set off finish. + * + * Not a sleep with a number on it: each round YIELDS so the fake can deliver the push it + * queued (on a macrotask, as the real RPC does), then WAITS on the work that push actually + * started (`observationSettled` — the enumerations in flight and the drains behind them). + * Several rounds because applying one push can queue the next: a Link filed on the private + * store pushes to the register subscription, which re-enumerates. + */ +async function converge(): Promise { + for (let round = 0; round < 5; round += 1) { + await new Promise((resolve) => setTimeout(resolve, 0)); + await observationSettled(); + } +} + +/** + * Take what has been written into `graph` OUT of this page's wallet and hand it back — the + * broker holding a commit it has not delivered yet. Delivering it later + * (`fake._deliver`) is the only way a deposit made in another session can arrive here + * while Alice, and not its author, is the connected identity. + */ +function heldByTheBroker(graph: Nuri): Quad[] { + const held: Quad[] = []; + for (let i = quads.length - 1; i >= 0; i -= 1) { + if (quads[i]!.g === graph) held.unshift(...quads.splice(i, 1)); + } + return held; +} + +/** + * The documents a user has durably been GIVEN — the emulated `AddLink` records on their + * User branch, named by the document each one opens. The record holds a `ReadCap` (the + * reference plus its secret); the tests are about WHICH document arrived, so the secret is + * dropped here rather than pinned to the stand-in value the emulation currently mints. + */ +async function documentsGivenTo(id: string): Promise { + const store = (await resolveAccount(id))?.docPrivate; + if (!store) return []; + return quads + .filter((q) => q.g === store && q.p === `${SHIM}:link`) + .map((q) => q.o.split(":r:")[0]!); +} + +/** How many times this inbox's deposits have been read — i.e. how often it was processed. */ +function depositReadsOf(inbox: Nuri): number { + return fake.sparql_query.mock.calls.filter( + (c) => c[3] === inbox && String(c[1]).includes(`${INBOX}:payload`), + ).length; +} + +/** The inbox recorded for a note, read off the WALLET — no application can ask the package. */ +function inboxOnTheNote(note: Nuri): Nuri { + const record = quads.find((q) => q.p === `${SHIM}:inboxCap` && q.o.startsWith(note + " ")); + if (!record) throw new Error("no AddInboxCap record was written for the note"); + return record.o.split(" ")[1] as Nuri; +} + +/** Capture what `console.error` is told while `body` runs. */ +async function whileWatchingTheLog(body: () => Promise): Promise { + const lines: string[] = []; + const real = console.error; + console.error = ((...args: unknown[]) => { + lines.push(args.map((a) => String(a)).join(" ")); + }) as typeof console.error; + try { + await body(); + } finally { + console.error = real; + } + return lines; +} + +/** + * Bob makes a document and shares it with Alice by NAME, then the broker holds his deposit + * back. Returns the document he shared and the quads still in transit. + * + * Alice has to have been here before, and to have DONE something: `inbox.share` refuses a + * recipient nobody has ever been, and connecting does not provision — an identity acquires + * its account the first time it creates anything. That is the model and not a fixture + * detail: upstream a deposit is sealed to an inbox key somebody had to hand you, so you + * cannot address a name you invented. Her first visit here is not the one under test; every + * test below connects her again afterwards, and the deposit arrives strictly after that. + */ +async function bobSharesWithAlice(): Promise<{ doc: Nuri; inTransit: Quad[] }> { + await signIn("alice"); + await storeRegistry.createEntityDoc("protected"); // her first visit — now she exists + await signIn("bob"); + const doc = await storeRegistry.createEntityDoc("protected"); + await inboxSurface.share(doc, "alice"); + // Test-side inspection only — the address is read off the wallet to say WHICH document + // is in transit, and is never handed to an actor. + const aliceInbox = await userInbox("alice", "protected"); + return { doc, inTransit: heldByTheBroker(aliceInbox) }; +} + +beforeEach(() => { + forgetEverything(); + boot(); +}); + +afterAll(() => { + cancelScheduledInboxProcessing(); + forgetEverything(); +}); + +describe("a deposit that arrives while its recipient is connected", () => { + test("is applied, without anyone reconnecting", async () => { + const { doc, inTransit } = await bobSharesWithAlice(); + + await signIn("alice"); + // The honest baseline: as far as this page is concerned Alice's inbox is empty, so + // connecting applied nothing. Whatever the next lines prove, they do not prove it twice. + expect(getCaps().capForHolder("alice", doc)).toBeUndefined(); + + fake._deliver(inTransit); + await converge(); + + expect(getCaps().capForHolder("alice", doc)).toBeDefined(); + expect(await documentsGivenTo("alice")).toContain(doc); + }); + + test("is applied DURABLY — the same as if she had reconnected to find it", async () => { + const { doc, inTransit } = await bobSharesWithAlice(); + await signIn("alice"); + fake._deliver(inTransit); + await converge(); + + // A cap held only in memory is a cap lost at the next reload, and the whole point of + // applying rather than merely reading is that it survives. + expect(await documentsGivenTo("alice")).toEqual([doc]); + }); + + test("does not need the depositor's tab to stay open — no timer is involved", async () => { + const { doc, inTransit } = await bobSharesWithAlice(); + await signIn("alice"); + // Whatever the deposit armed in Bob's session is dropped here, exactly as a closed tab + // drops it. What follows is the connected owner's own doing, or it does not happen. + cancelScheduledInboxProcessing(); + + fake._deliver(inTransit); + await converge(); + + expect(getCaps().capForHolder("alice", doc)).toBeDefined(); + }); +}); + +describe("an inbox opened in the middle of a session", () => { + test("is watched too — what lands in it is processed without reconnecting", async () => { + await signIn("alice"); + // The note and its inbox come into existence AFTER connecting, so nothing the + // connection enumerated could have included them. + const note = await storeRegistry.createEntityDoc("public"); + await storeRegistry.openDocumentInbox(note); + await converge(); + + const inbox = inboxOnTheNote(note); + const readsBefore = depositReadsOf(inbox); + + // A message is left on the NOTE — the depositor names the document, never an address. + await inboxSurface.postToDocument(note, { payload: { text: "j'apporte le café" }, from: null, ts: 1 }); + await converge(); + + // Processing an inbox IS reading its queue and applying what is this library's to + // apply; for a document inbox nothing is (a Link only ever reaches a person's inbox), + // so the read is the whole of the consequence — and it can come from nowhere else: + // depositing reads the shim, not the queue, and the deferred window has not closed. + expect(depositReadsOf(inbox)).toBeGreaterThan(readsBefore); + }); + + test("the messages left on it are readable, on the document its owner named", async () => { + await signIn("alice"); + const note = await storeRegistry.createEntityDoc("public"); + await storeRegistry.openDocumentInbox(note); + await converge(); + + await inboxSurface.postToDocument(note, { payload: { text: "à demain" }, from: null, ts: 2 }); + await converge(); + + const left = await inboxSurface.readForDocument(note); + expect(left.map((d) => (d.payload as { text: string }).text)).toEqual(["à demain"]); + }); +}); + +describe("switching identity", () => { + test("stops the observation — the previous identity's inbox is no longer applied", async () => { + const { doc, inTransit } = await bobSharesWithAlice(); + await signIn("alice"); + + // Alice steps away and Bob takes the page. A session belongs to one person. + await signIn("bob"); + + fake._deliver(inTransit); + await converge(); + + // Nothing was applied for Alice — she is not connected, and her queue keeps its + // deposit for the next time she is. + expect(getCaps().capForHolder("alice", doc)).toBeUndefined(); + expect(await documentsGivenTo("alice")).toEqual([]); + // …and emphatically nothing was filed for Bob either: work started for one holder must + // never file for another. (Bob's own cap on the document is not evidence of that — he + // made it. What would be evidence is a Link, and there is none.) + expect(await documentsGivenTo("bob")).toEqual([]); + }); + + test("and disconnecting stops it too", async () => { + const { doc, inTransit } = await bobSharesWithAlice(); + await signIn("alice"); + + setCurrentUser(null); // no identity is acting — anonymous holds nothing and owns no inbox + + fake._deliver(inTransit); + await converge(); + + expect(await documentsGivenTo("alice")).toEqual([]); + }); + + test("and Alice coming back finds the deposit still there to apply", async () => { + const { doc, inTransit } = await bobSharesWithAlice(); + await signIn("alice"); + await signIn("bob"); + fake._deliver(inTransit); + await converge(); + + // An inbox is not consumed by being ignored: connecting again drains what was left. + await signIn("alice"); + await converge(); + + expect(getCaps().capForHolder("alice", doc)).toBeDefined(); + }); +}); + +describe("a deposit that cannot be applied", () => { + test("is reported, and stops neither the observation nor the next deposit", async () => { + const first = await bobSharesWithAlice(); + const second = await bobSharesWithAlice(); + const aliceInbox = await userInbox("alice", "protected"); + + await signIn("alice"); + + // The broker cannot answer for Alice's inbox — the deposit lands, applying it does not. + fake._failReadsOn.add(aliceInbox); + const reported = await whileWatchingTheLog(async () => { + fake._deliver(first.inTransit); + await converge(); + }); + + expect(reported.filter((l) => /could not apply what is in this inbox/.test(l)).length) + .toBeGreaterThan(0); + // Prefixed by the connected identity, like every other polyfill-layer line. + expect(reported.find((l) => /could not apply what is in this inbox/.test(l))) + .toContain("[alice][polyfill]"); + expect(getCaps().capForHolder("alice", first.doc)).toBeUndefined(); + + // The broker recovers. The observation is still running — one unapplicable item denies + // nothing — and the deposit that failed was never consumed, so both land now. + fake._failReadsOn.delete(aliceInbox); + fake._deliver(second.inTransit); + await converge(); + + expect(getCaps().capForHolder("alice", second.doc)).toBeDefined(); + expect(getCaps().capForHolder("alice", first.doc)).toBeDefined(); + }); + + test("never rejects into the application — nobody asked for this work", async () => { + const { inTransit } = await bobSharesWithAlice(); + const aliceInbox = await userInbox("alice", "protected"); + await signIn("alice"); + fake._failReadsOn.add(aliceInbox); + + const unhandled: unknown[] = []; + const onUnhandled = (e: unknown): void => { + unhandled.push(e); + }; + process.on("unhandledRejection", onUnhandled); + try { + await whileWatchingTheLog(async () => { + fake._deliver(inTransit); + await converge(); + }); + } finally { + process.off("unhandledRejection", onUnhandled); + fake._failReadsOn.delete(aliceInbox); + } + expect(unhandled).toEqual([]); + }); +}); diff --git a/packages/polyfill/test/deferred-inbox-processing.test.ts b/packages/polyfill/test/deferred-inbox-processing.test.ts index 166ce42..655db15 100644 --- a/packages/polyfill/test/deferred-inbox-processing.test.ts +++ b/packages/polyfill/test/deferred-inbox-processing.test.ts @@ -434,7 +434,10 @@ test("a drain that fails says so in the package's log, and rejects into nobody", console.error = realError; } - const reported = errors.filter((line) => /deferred inbox processing failed/.test(line)); + // The wording is the SHARED reporter's since 2026-08-17 (`emulated-verifier/inbox-drain.ts`): + // the timer is no longer the only thing that drains an inbox — the continuous observation + // does too, through the same queue — so the line names the act and not the schedule. + const reported = errors.filter((line) => /could not apply what is in this inbox/.test(line)); expect(reported.length).toBe(1); // Prefixed by the CONNECTED identity, like every other polyfill-layer line — which is // what makes a drain running under someone else's session legible in a live trace. diff --git a/packages/polyfill/test/public-store.test.ts b/packages/polyfill/test/public-store.test.ts index cb93f34..4f553d4 100644 --- a/packages/polyfill/test/public-store.test.ts +++ b/packages/polyfill/test/public-store.test.ts @@ -145,21 +145,30 @@ test("a public store serves every asker, not only the first", async () => { test("asked once per document: the outcome is memoised, in both directions", async () => { const { sparql_query } = inject(); + // Counted PER DOCUMENT (the read is anchored on the one being asked about), not over + // every read the process makes. `setCurrentUser` fires the connection work, which reads + // on its own account in the background — none of it about this document — so a total + // makes the memo's arithmetic depend on whatever else happens to be in flight. This is + // the same assertion, about the read it is actually about. + const asks = (nuri: string): number => + sparql_query.mock.calls.filter((c) => c[3] === nuri).length; + setCurrentUser("alice"); await aliceExposesHerNote(); armEmulation(); setCurrentUser("bob"); await fetchReadCap(PUB); - const afterHit = sparql_query.mock.calls.length; + const afterHit = asks(PUB); await fetchReadCap(PUB); // held now → not even the memo is consulted - expect(sparql_query.mock.calls.length).toBe(afterHit); + expect(asks(PUB)).toBe(afterHit); const absent = "did:ng:o:nothing-here" as Nuri; await fetchReadCap(absent); - const afterMiss = sparql_query.mock.calls.length; + const afterMiss = asks(absent); + expect(afterMiss).toBe(1); // it WAS asked once — a memo over nothing proves nothing await fetchReadCap(absent); // a miss is remembered too - expect(sparql_query.mock.calls.length).toBe(afterMiss); + expect(asks(absent)).toBe(afterMiss); }); test("resetting the caps forgets the memo — a stale yes would hand back what is no longer held", async () => { diff --git a/packages/polyfill/test/subscribe.test.ts b/packages/polyfill/test/subscribe.test.ts index d73145d..29a681d 100644 --- a/packages/polyfill/test/subscribe.test.ts +++ b/packages/polyfill/test/subscribe.test.ts @@ -1,5 +1,5 @@ import { test, expect, mock, afterAll } from "bun:test"; -import { subscribeDoc, subscribeDocs } from "../src/surface/subscribe"; +import { docChangeType, subscribeDoc, subscribeDocs } from "../src/surface/subscribe"; import { configure } from "../src/index"; import { configureStoreRegistry } from "../src/shared-wallet/bootstrap"; import { resetConfig, resetStoreRegistry } from "../src/shared-wallet/bootstrap"; @@ -17,29 +17,41 @@ afterAll(() => { 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). + * A fake reactive `ng`: `doc_subscribe(nuri, sid, cb)` takes over `nuri`, fires `cb` once + * (initial State push), and returns an unsubscribe. `push(nuri)` drives a later change to + * that doc's subscriber. A per-doc `failFor` set makes `doc_subscribe` reject for chosen + * NURIs (a not-yet-synced doc). + * + * ── ONE subscriber per document, and a second one EVICTS it ─────────────── + * Not a simplification — it is what the broker does. A branch holds a single sender + * (`branch_subscriptions: HashMap>`) and + * `create_branch_subscription` closes whatever it displaces, silently: the evicted + * unsubscribe still returns cleanly and nothing anywhere errors. Confirmed against the real + * broker on 2026-08-17 — with two subscriptions on one document, a write fired the second + * callback and the first, which had been firing moments earlier, went quiet for good. + * + * A Set of callbacks here would model a world where every subscriber coexists. It is + * exactly the assumption that cost this package a view that never re-read and an inbox that + * never notified, and a fake that holds it cannot fail on either. */ function makeFakeNg(failFor: Set = new Set()) { - const subs = new Map void>>(); + 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); + subs.set(nuri, cb); // whoever held this branch is dropped, without a word + // Initial State push, delivered async (as the real RPC does) — and only while this + // callback still holds the branch. + queueMicrotask(() => { + if (subs.get(nuri) === cb) cb({ V0: { State: { doc: nuri } } }); + }); + return () => { + if (subs.get(nuri) === cb) subs.delete(nuri); + }; }); const push = (nuri: string): void => { - for (const cb of subs.get(nuri) ?? []) cb({ V0: { Patch: { doc: nuri } } }); + subs.get(nuri)?.({ V0: { Patch: { doc: nuri } } }); }; - const isSubscribed = (nuri: string): boolean => (subs.get(nuri)?.size ?? 0) > 0; + const isSubscribed = (nuri: string): boolean => subs.has(nuri); return { doc_subscribe, push, isSubscribed, _subs: subs }; } @@ -143,3 +155,73 @@ test("subscribeDocs deduplicates repeated NURIs", async () => { await tick(); expect(ng.doc_subscribe).toHaveBeenCalledTimes(1); }); + +// --- two subscribers on ONE document --------------------------------------- +// +// A branch has room for exactly one subscriber and a second `doc_subscribe` evicts the +// first (see `makeFakeNg`). Everything inside this package subscribes — `ensureRepoOpen` +// holds a bootstrap subscription per document for the session, `watchShape` follows the +// documents of a scope, `inbox.watch` follows an inbox, and the inbox observation follows +// every inbox — so any two of them on the same document used to silence one another. +// Nothing rejected and nothing logged; the view simply stopped re-reading. +// +// So the package opens ONE real subscription per document and fans it out. These are the +// tests that say so. + +test("two subscribers on one document BOTH keep firing", async () => { + const ng = inject(); + const first: unknown[] = []; + const second: unknown[] = []; + subscribeDoc(A, (r) => first.push(r)); + await tick(); + expect(first).toHaveLength(1); // its initial State + + subscribeDoc(A, (r) => second.push(r)); + await tick(); + + ng.push(A); + // The one that was there first is not silenced by the one that came second. + expect(first).toHaveLength(2); + expect(second.length).toBeGreaterThanOrEqual(1); +}); + +test("one real doc_subscribe serves every subscriber of a document", async () => { + const ng = inject(); + subscribeDoc(A, () => {}); + subscribeDoc(A, () => {}); + subscribeDoc(A, () => {}); + await tick(); + // Three callers, one branch taken. A second call would have evicted the first caller. + expect(ng.doc_subscribe).toHaveBeenCalledTimes(1); +}); + +test("a subscriber that joins LATE still gets its initial State", async () => { + inject(); + subscribeDoc(A, () => {}); + await tick(); // the initial State has come and gone + + const late: unknown[] = []; + subscribeDoc(A, (r) => late.push(r)); + await tick(); + + // Its own `doc_subscribe` would have pushed it a State; joining an open one owes it the + // same, or "fires once immediately" quietly stops being true for whoever arrives second. + expect(late).toHaveLength(1); + expect(docChangeType(late[0])).toBe("State"); +}); + +test("the real subscription is released only when the LAST subscriber leaves", async () => { + const ng = inject(); + const seen: unknown[] = []; + const stopFirst = subscribeDoc(A, () => {}); + const stopSecond = subscribeDoc(A, (r) => seen.push(r)); + await tick(); + + stopFirst(); + expect(ng.isSubscribed(A)).toBe(true); // somebody is still listening + ng.push(A); + expect(seen.length).toBeGreaterThanOrEqual(2); // and still hearing + + stopSecond(); + expect(ng.isSubscribed(A)).toBe(false); // now nobody is +}); diff --git a/packages/polyfill/test/wallet-fake.ts b/packages/polyfill/test/wallet-fake.ts index c08290a..90139a6 100644 --- a/packages/polyfill/test/wallet-fake.ts +++ b/packages/polyfill/test/wallet-fake.ts @@ -125,6 +125,23 @@ export interface FakeWallet { /** Present only under {@link WalletOptions.unsyncedUntilSubscribed}. */ doc_subscribe?: ReturnType; _quads: Quad[]; + /** + * A commit made in ANOTHER session, reaching this page now — the broker delivering what + * it was holding. The quads land in the wallet and each document they touch pushes to + * its subscriber, which is what a remote write does here: verified against the real + * broker, a second session's write reached the first session's subscription as a `Patch` + * 12ms after it landed (`e2e/reactivity-doc-subscribe.ts`, CROSS). + * + * It delivers; it does not INVENT. A caller hands it quads the library itself produced + * under the other actor's identity — never a shape a test wrote by hand. + */ + _deliver: (arriving: Quad[]) => void; + /** + * Anchors whose anchored READ throws, as an unreachable repo does. Mutable after boot, + * so a suite builds a healthy world first and breaks only the one call it is about — + * the fault is the broker's, never a reach into the library to make it reject. + */ + _failReadsOn: Set; } export interface WalletOptions { @@ -168,6 +185,38 @@ export function makeWallet(quads: Quad[], options: WalletOptions = {}): FakeWall const synced = new Set(); const cold = options.unsyncedUntilSubscribed === true; + /** + * The ONE subscriber a document can have. + * + * Not a convenience — it is what the broker does. A branch holds exactly one sender + * (`branch_subscriptions: HashMap>`) and + * `create_branch_subscription` closes whatever it displaces, so a second + * `doc_subscribe` on a document does not join the first, it EVICTS it — silently, with + * the evicted unsubscribe still callable and no error anywhere. Confirmed against the + * real broker on 2026-08-17: with two subscriptions on one document, a write fired the + * second callback and the first, which had been firing moments before, went quiet. + * + * A Set here would fabricate a world where every subscriber coexists — precisely the + * assumption whose falseness cost this package a view that never re-read and an inbox + * that never notified. + */ + const subscriber = new Map void>(); + + /** See {@link FakeWallet._failReadsOn}. */ + const failReadsOn = new Set(); + + /** A commit on `g` pushes a `Patch` to that document's subscriber — the SESSION THAT + * WROTE IT INCLUDED. Verified against the real broker the same day: a session's own + * `sparqlUpdate` to a document it subscribes to pushed `Patch@69ms`. The engine keys + * its senders by branch and knows nothing about who issued the write. */ + const commit = (g: string): void => { + const cb = subscriber.get(g); + if (!cb) return; + setTimeout(() => { + if (subscriber.get(g) === cb) cb({ V0: { Patch: {} } }); + }, 0); + }; + const doc_create = mock(async () => { const nuri = `did:ng:o:doc${++minted}`; // Created here: nothing remote to wait for. This is why the session that wrote the @@ -180,11 +229,19 @@ export function makeWallet(quads: Quad[], options: WalletOptions = {}): FakeWall const nuri = a[0] as string; const onChange = a[2] as (r: unknown) => void; synced.add(nuri); + subscriber.set(nuri, onChange); // `TabInfo` first, then the initial `State` — the platform's own order, so a waiter // that resolved on "the first push of any kind" would return BEFORE the barrier. - setTimeout(() => onChange({ V0: { TabInfo: {} } }), 0); - setTimeout(() => onChange({ V0: { State: {} } }), 0); - return () => {}; + // Only while this callback still holds the branch: an evicted subscriber hears nothing. + setTimeout(() => { + if (subscriber.get(nuri) === onChange) onChange({ V0: { TabInfo: {} } }); + }, 0); + setTimeout(() => { + if (subscriber.get(nuri) === onChange) onChange({ V0: { State: {} } }); + }, 0); + return () => { + if (subscriber.get(nuri) === onChange) subscriber.delete(nuri); + }; }); const sparql_update = mock(async (...a: unknown[]) => { @@ -199,6 +256,7 @@ export function makeWallet(quads: Quad[], options: WalletOptions = {}): FakeWall const q = quads[i]!; if (q.g === anchor && q.s === pattern[1] && q.p === pattern[2]) quads.splice(i, 1); } + commit(anchor); } return undefined; } @@ -210,6 +268,7 @@ export function makeWallet(quads: Quad[], options: WalletOptions = {}): FakeWall ? wrapped[2]! : query.replace(/^[\s\S]*?INSERT\s+DATA\s*\{/i, "").replace(/\}\s*$/, ""); for (const t of parseTriples(body)) quads.push({ g, ...t }); + commit(g); return undefined; }); @@ -221,6 +280,12 @@ export function makeWallet(quads: Quad[], options: WalletOptions = {}): FakeWall // The repo the verifier resolves the read against — the anchor when there is one, // otherwise the graph named in the query. const target = anchor ?? g; + // The repo this broker cannot answer for. Rejects, as `resolve_target_for_sparql` + // does on a repo the verifier does not have — never 0 rows, which would be the + // altogether different (and silent) cold-start state modelled below. + if (target !== undefined && failReadsOn.has(target)) { + throw new Error(`RepoNotFound: ${target}`); + } // COLD: present but unsynced. No error, no rows — which is exactly why it is dangerous. if (cold && target !== undefined && !synced.has(target)) return { results: { bindings: [] } }; const inGraph = quads.filter((q) => q.g === g); @@ -299,9 +364,17 @@ export function makeWallet(quads: Quad[], options: WalletOptions = {}): FakeWall return { results: { bindings: [] } }; }); - return cold - ? { doc_create, doc_subscribe, sparql_update, sparql_query, _quads: quads } - : { doc_create, sparql_update, sparql_query, _quads: quads }; + const _deliver = (arriving: Quad[]): void => { + const touched = new Set(); + for (const q of arriving) { + quads.push(q); + touched.add(q.g); + } + for (const g of touched) commit(g); + }; + + const common = { doc_create, sparql_update, sparql_query, _quads: quads, _deliver, _failReadsOn: failReadsOn }; + return cold ? { ...common, doc_subscribe } : common; } /** Wire the library onto `quads` — what a page load does. */