From 33212a8b00b15f30a5ab446c6a453bed29327f01 Mon Sep 17 00:00:00 2001 From: Sylvain Duchesne Date: Mon, 17 Aug 2026 11:47:21 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20un=20abonnement=20en=20=C3=A9chec=20n'em?= =?UTF-8?q?poisonne=20plus=20le=20document,=20et=20un=20d=C3=A9p=C3=B4t=20?= =?UTF-8?q?ne=20change=20plus=20de=20destinataire?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .project/concepts/app-contract/_debt.md | 1 + .../polyfill/src/emulated-verifier/connect.ts | 65 +++++--- .../src/emulated-verifier/inbox-observer.ts | 85 +++++++++-- packages/polyfill/src/surface/inbox.ts | 33 ++++- packages/polyfill/src/surface/subscribe.ts | 139 ++++++++++++++++-- .../test/continuous-inbox-observation.test.ts | 120 +++++++++++++++ packages/polyfill/test/subscribe.test.ts | 98 +++++++++++- 7 files changed, 493 insertions(+), 48 deletions(-) diff --git a/.project/concepts/app-contract/_debt.md b/.project/concepts/app-contract/_debt.md index b70383b..7fcded1 100644 --- a/.project/concepts/app-contract/_debt.md +++ b/.project/concepts/app-contract/_debt.md @@ -5,3 +5,4 @@ ## Raw markers (consolidate into blocks, then delete) - TOUCHED packages/polyfill/src/surface/subscribe.ts @2026-08-17 (session f93872b5-293a-4916-a353-181409a96d42) +- TOUCHED packages/polyfill/src/surface/inbox.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 5dd30b5..5b6b672 100644 --- a/packages/polyfill/src/emulated-verifier/connect.ts +++ b/packages/polyfill/src/emulated-verifier/connect.ts @@ -18,7 +18,9 @@ * 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. + * this is the regime. It runs whatever became of the first two, including a step 1 that + * REJECTED: what is watched is *being connected*, and `setCurrentUser` has already made + * that true by the time any of this runs. * * Restoring first means a reconnecting user can read its documents immediately, without * waiting on the inbox round-trip; watching last means the backlog is applied before the @@ -164,7 +166,7 @@ export async function connectedUser(): Promise { // Captured with the identity, handed back at filing time — see `caps.holderKey`. const holderKey = getCaps().holderKey(); - const run = (async (): Promise => { + const restoreAndDrain = async (): Promise => { // Connecting must not PROVISION. `ensureAccount` would create the user on // first sight, so connecting an identity that does not exist yet would // silently mint its stores and their caps — arming the whole emulation as a @@ -186,8 +188,7 @@ export async function connectedUser(): Promise { // 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(); + // something. (Started below, for every outcome of this function alike.) return; } if (!stillConnected()) return; @@ -226,24 +227,44 @@ 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(); + }; + + const run = (async (): Promise => { + try { + await restoreAndDrain(); + } finally { + // 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 restore and the drain, not before: connecting owes the backlog first, + // and the observation subscribes to the same inboxes that loop just read. + // + // ── Unconditional on how they WENT, and that is the whole point ── + // 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. + // + // The same holds one step up, and until 2026-08-17 it did not: a restore that + // REJECTED skipped this line, and the identity was left CONNECTED — `setCurrentUser` + // is synchronous and had already taken effect — with nothing watching its inboxes + // for the rest of the session. `startObservingInboxes` has no other caller, so one + // broker hiccup at sign-in cost that person every deposit made from then on, in + // silence, long after the broker had recovered. + // + // It does not blur the rule this function is built on. The rule is about what the + // CALLER is told — failing to reach the queues rejects, failing to apply one is + // reported — and rejecting is exactly what still happens: the error raised above + // propagates through this `finally` untouched. What changes is that being connected + // now means being watched, whatever the connection made of its own work. + // + // 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()) await startObservingInboxes(); + } })(); inFlight.set(holder, run); diff --git a/packages/polyfill/src/emulated-verifier/inbox-observer.ts b/packages/polyfill/src/emulated-verifier/inbox-observer.ts index 36e1f3c..f339510 100644 --- a/packages/polyfill/src/emulated-verifier/inbox-observer.ts +++ b/packages/polyfill/src/emulated-verifier/inbox-observer.ts @@ -61,7 +61,7 @@ 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 { subscribeDoc, subscribeDocReportingSetupFailure, type Unsubscribe } from "../surface/subscribe"; import type { Nuri, PrincipalId } from "../model/types"; /** @@ -78,6 +78,12 @@ interface Observation { register: Unsubscribe | null; /** Unsubscribe from the held-caps change signal — the second "which inboxes" channel. */ holdings: Unsubscribe | null; + /** + * The inboxes whose subscription has already been re-attempted once after failing to be + * OPENED. What bounds the repair below to one extra try per inbox, so a document the broker + * genuinely cannot serve costs two calls rather than a loop. + */ + reattempted: Set; /** 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. */ @@ -128,21 +134,30 @@ 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. + // identity can have moved while it waited. Running it for the wrong holder is not a + // near-miss — it reads someone else's registers and files into someone else's ring. + // + // This check is NECESSARY and it is not SUFFICIENT, and until 2026-08-17 this comment + // claimed it was. It cannot be: the identity can move after it passes, while + // `processInbox` is mid-read. The claim was that `processInbox` "resolves the holder at + // each step, so the new holder is refused an inbox that is not theirs somewhere in the + // middle" — but its ownership guard runs ONCE, at entry, and a switch after that reached + // the filing with nobody left to refuse it. What it filed was the previous holder's cap, + // into the new holder's ring, durably. The guard that makes this run safe is therefore + // the one INSIDE `processInbox`, which captures the holder its guard authorised; this one + // only spares the work when the switch is already visible. 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`. + // An identity that moved MID-DRAIN can throw here too — the switch may land before + // `processInbox`'s own ownership guard, which then refuses the new holder an inbox that + // is not theirs. (Landing after it, the run ABANDONS quietly instead and returns; both + // leave the deposit where it is.) Abandoning is what 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; } @@ -175,9 +190,19 @@ async function watchTheInboxes(obs: Observation): Promise { // `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. + // + // The failure channel is not decoration: `subscribeDoc` returns SYNCHRONOUSLY and its + // one real `doc_subscribe` is opened afterwards, so a rejection there reaches nobody. + // Without it, this map held an entry for an inbox that was never watched, the loop + // above skipped it at every later enumeration, and the only observable difference from + // a healthy session was that shares stopped arriving. obs.inboxes.set( inbox, - subscribeDoc(inbox, () => track(obs, applyWhatArrived(obs, inbox))), + subscribeDocReportingSetupFailure( + inbox, + () => track(obs, applyWhatArrived(obs, inbox)), + (error) => watchFailed(obs, inbox, error), + ), ); logStage("OBSERVING " + shortNuri(inbox) + " for " + obs.holder); } catch (error) { @@ -186,6 +211,39 @@ async function watchTheInboxes(obs: Observation): Promise { } } +/** + * The subscription on `inbox` could not be OPENED — report it, forget it, and try once more. + * + * **Forget it**, because the entry this observation holds is the whole record of "already + * watched": leaving a dead one in place is how one rejection at connection turned into a + * session-long silence. Removed, the next enumeration subscribes again as if it had never + * been attempted. + * + * **Try once more**, because the usual cause is a repo the verifier has not synced yet, and + * the whole cost of finding out is one call. Exactly one extra attempt per inbox, tracked in + * {@link Observation.reattempted}: a broker that genuinely cannot serve this document must + * not be asked again and again by a loop that has no reason to stop. If the second attempt + * fails too, the failure has been reported twice and the inbox waits for the next + * enumeration signal — or for the next connection, where an unapplied deposit has always + * waited. + */ +function watchFailed(obs: Observation, inbox: Nuri, error: unknown): void { + if (!current(obs)) return; + const stale = obs.inboxes.get(inbox); + obs.inboxes.delete(inbox); + if (stale) { + try { + stale(); + } catch (thrown) { + console.error(accessLogPrefix() + " releasing a failed inbox observation failed:", thrown); + } + } + reportUnobserved("this inbox could not be watched: " + shortNuri(inbox), error); + if (obs.reattempted.has(inbox)) return; + obs.reattempted.add(inbox); + track(obs, enumerate(obs)); +} + /** * 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 @@ -279,6 +337,7 @@ export async function startObservingInboxes(): Promise { inboxes: new Map(), register: null, holdings: null, + reattempted: new Set(), enumerating: false, enumerateAgain: false, pending: new Set(), diff --git a/packages/polyfill/src/surface/inbox.ts b/packages/polyfill/src/surface/inbox.ts index 4357ead..801cba2 100644 --- a/packages/polyfill/src/surface/inbox.ts +++ b/packages/polyfill/src/surface/inbox.ts @@ -669,16 +669,47 @@ export async function readSyncedForDocument(docLike: NuriLike): Promise { const targetInbox = toNuri(targetInboxLike, "inbox.processInbox"); + // WHO this processing is for, captured before the read that authorises it — see above. + const holder = getCurrentUser(); const deposits = await readSynced(targetInbox); // `readSynced` already put every Link in memory for this session; now make // them durable. Reading the raw deposits again would mean re-parsing, so the caps // are taken from what the read just observed. - for (const cap of capsSeenIn(targetInbox)) await addLink(cap); + const seen = capsSeenIn(targetInbox); seenByInbox.delete(targetInbox); + for (const cap of seen) { + // Re-checked per cap, not once: `addLink` reads and writes, so the identity can move + // between two of them just as easily as during the read. + if (getCurrentUser() !== holder) { + logStage( + "ABANDONED " + shortNuri(targetInbox) + " — the identity changed while it was being " + + "processed; its deposits stay for their owner", + ); + return deposits; + } + await addLink(cap, holder ?? undefined); + } return deposits; } diff --git a/packages/polyfill/src/surface/subscribe.ts b/packages/polyfill/src/surface/subscribe.ts index 684f003..bb30d4a 100644 --- a/packages/polyfill/src/surface/subscribe.ts +++ b/packages/polyfill/src/surface/subscribe.ts @@ -61,6 +61,19 @@ * 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. + * + * ── What SHARING one subscription must not cost a caller ─────────────────── + * Three things a per-caller `doc_subscribe` gave for free, and that the fan-out has to give + * back deliberately — each of them was lost when it was introduced on 2026-08-17: + * + * - a caller is ITSELF, not its handler. One record per CALL, so passing the same function + * twice is two subscriptions and the first unsubscribe does not silence the second; + * - a caller that has LEFT hears nothing more, including from the push during which it + * left — a handler may tear another one down, and the pushes are re-checked against the + * live set rather than a copy taken before the first handler ran; + * - a setup that FAILED is not the end of the document. The attempt is forgotten so the + * next joiner makes its own (upstream, each caller's own `doc_subscribe` retried), and + * it is reported rather than left as silence — see {@link reportSetupFailure}. */ import { getConfig, getStoreRegistryDeps } from "../shared-wallet/bootstrap"; @@ -122,16 +135,51 @@ async function sessionId(): Promise { /** What a listener is handed on every push. */ type Listener = (r: DocChange, type: DocChangeType) => void; +/** + * ONE call to {@link subscribeDocUnguarded}, and what that caller asked for. + * + * Identified by this record and never by the `onChange` function: two callers may + * legitimately pass the SAME function — a module-level handler, a bound method, an arrow + * that closes over nothing — and they are two subscriptions with two independent lifetimes. + * Keyed by the function, the second call was swallowed by the set and the FIRST caller's + * unsubscribe silenced the second, which had never asked to leave. + */ +interface DocListener { + /** What this caller is handed on every push. */ + onChange: Listener; + /** + * Told when the one real `doc_subscribe` behind this listener could not be OPENED. + * + * Upstream `doc_subscribe` is async and REJECTS on a setup failure, so its caller learns. + * This wrapper returns synchronously, so without this channel a caller cannot tell "this + * document is quiet" from "this document is not subscribed at all" — the failure-as-absence + * this package keeps closing. `null` for a caller that did not ask (the published + * {@link subscribeDoc}, whose contract has no such argument); the failure is logged either + * way. + */ + onSetupFailed: ((error: unknown) => void) | null; +} + /** * 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; + 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. */ + /** + * True from the moment a setup is kicked off — a later joiner must not kick off a second, + * because a second `doc_subscribe` on a branch EVICTS the first (see the module header). + * It therefore stays true once the setup has SUCCEEDED: the attempt still stands. + * + * Cleared on the two ways the attempt stops standing: the last listener leaving + * ({@link releaseFanOut}), and a setup that FAILED ({@link reportSetupFailure}). Left true + * on failure it stopped meaning "one is already running" and started meaning "this NURI is + * finished" — no later joiner ever attempted it again, for the whole session, over one + * transient rejection. + */ establishing: boolean; /** * The most recent `State` push, replayed to a listener that joins later. @@ -143,14 +191,57 @@ interface DocFanOut { 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 { +function deliver(nuri: Nuri, listener: DocListener, resp: DocChange, type: DocChangeType): void { try { - listener(resp, type); + listener.onChange(resp, type); } catch (error) { console.error("[subscribe] onChange handler threw for", nuri, error); } } +/** + * Hand one push to every listener of `entry` — the fan-out itself. + * + * Over a COPY, because a handler may unsubscribe itself or another from inside the push; and + * re-checking each listener against the live set, because a copy alone only stops the walk + * from breaking — it still delivers to whoever left DURING it. That contradicted what + * {@link subscribeDoc} publishes ("no further `onChange` fires after unsubscribe"), on the + * one ordering an application cannot control: whether its handler runs before or after the + * one that tore it down. + */ +function fanOut(nuri: Nuri, entry: DocFanOut, resp: DocChange, type: DocChangeType): void { + for (const listener of [...entry.listeners]) { + if (!entry.listeners.has(listener)) continue; + deliver(nuri, listener, resp, type); + } +} + +/** + * The one real subscription could not be OPENED: forget the attempt, and say so. + * + * Two halves, and the shape of every "failure disguised as an absence" this package has + * closed. **Retryable** — `establishing` goes back to false, so the next `subscribeDoc` on + * this NURI attempts it again instead of joining a fan-out that will never push. Left true, + * one transient rejection made the document silent for every later joiner in the session. + * **Visible** — the log always, plus the callers who asked to be told, so a caller whose job + * depends on the subscription (the inbox observation) can report it and try again rather + * than sit on a dead entry. + */ +function reportSetupFailure(nuri: Nuri, entry: DocFanOut, error: unknown): void { + console.error("[subscribe] doc_subscribe failed for", nuri, error); + entry.establishing = false; + if (fanOuts.get(nuri) !== entry) return; + for (const listener of [...entry.listeners]) { + const tell = listener.onSetupFailed; + if (tell === null || !entry.listeners.has(listener)) continue; + try { + tell(error); + } catch (thrown) { + console.error("[subscribe] onSetupFailed handler threw for", nuri, thrown); + } + } +} + /** 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); @@ -187,8 +278,7 @@ async function establish(nuri: Nuri, entry: DocFanOut, ng: { doc_subscribe?: unk 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); + fanOut(nuri, entry, 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. @@ -197,7 +287,7 @@ async function establish(nuri: Nuri, entry: DocFanOut, ng: { doc_subscribe?: unk } entry.realUnsub = typeof unsub === "function" ? unsub : null; } catch (error) { - console.error("[subscribe] doc_subscribe failed for", nuri, error); + reportSetupFailure(nuri, entry, error); } } @@ -291,6 +381,30 @@ export function subscribeDoc( return subscribeDocUnguarded(nuri, onChange); } +/** + * {@link subscribeDoc}, for a caller that must be TOLD when the subscription could not be + * opened. Same guard, same fan-out; the only difference is that a setup failure reaches + * `onSetupFailed` instead of only the log. + * + * Internal, and deliberately not the published shape: upstream `doc_subscribe` is async and + * rejects, so a failure has a caller to reach; ours returns synchronously, and the published + * signature has no room for it (`.project/concepts/app-contract/polyfill-surface/`). An + * application uses a subscription as a change SIGNAL and has nothing to do with the answer, + * so it keeps the two-argument call. The inbox observation does have something to do with it + * — an inbox it believes it is watching and is not leaves every deposit unapplied for the + * session — so it asks. + */ +// @provenance subscribeDocReportingSetupFailure kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — restores to an internal caller what upstream's async `doc_subscribe` gives every caller: the setup failure. One delta remains, the synchronous unsubscribe +export function subscribeDocReportingSetupFailure( + nuriLike: NuriLike, + onChange: (r: DocChange, type: DocChangeType) => void, + onSetupFailed: (error: unknown) => void, +): Unsubscribe { + const nuri = toNuri(nuriLike, "subscribeDoc"); + assertMayReach(nuri, "subscribeDoc"); + return subscribeDocUnguarded(nuri, onChange, onSetupFailed); +} + /** * The unguarded core. Exported for ONE importer — `shared-wallet/physical.ts`, which * owns the machinery's entire privileged door — and for nobody else. It is not @@ -300,6 +414,7 @@ export function subscribeDoc( export function subscribeDocUnguarded( nuri: Nuri, onChange: (r: DocChange, type: DocChangeType) => void, + onSetupFailed?: (error: unknown) => void, ): Unsubscribe { // Resolved here, synchronously, so calling this before `configure()` still throws at // the call rather than inside a background task nobody awaits. @@ -311,7 +426,9 @@ export function subscribeDocUnguarded( fanOuts.set(nuri, entry); } const joined = entry; - joined.listeners.add(onChange); + // THIS call's subscription — see {@link DocListener} for why it is not the function. + const listener: DocListener = { onChange, onSetupFailed: onSetupFailed ?? null }; + joined.listeners.add(listener); // 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 @@ -319,8 +436,8 @@ export function subscribeDocUnguarded( if (joined.lastState) { const { resp, type } = joined.lastState; queueMicrotask(() => { - if (fanOuts.get(nuri) === joined && joined.listeners.has(onChange)) { - deliver(nuri, onChange, resp, type); + if (fanOuts.get(nuri) === joined && joined.listeners.has(listener)) { + deliver(nuri, listener, resp, type); } }); } @@ -334,7 +451,7 @@ export function subscribeDocUnguarded( return () => { if (stopped) return; stopped = true; - joined.listeners.delete(onChange); + joined.listeners.delete(listener); // 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). diff --git a/packages/polyfill/test/continuous-inbox-observation.test.ts b/packages/polyfill/test/continuous-inbox-observation.test.ts index 3ddecfa..f41e2ce 100644 --- a/packages/polyfill/test/continuous-inbox-observation.test.ts +++ b/packages/polyfill/test/continuous-inbox-observation.test.ts @@ -34,6 +34,8 @@ 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 { connectedUser } from "../src/emulated-verifier/connect"; +import { setOpenTimeoutForTests } from "../src/emulated-verifier/open-repo"; import { bootPage, forgetEverything, signIn, type FakeWallet, type Quad } from "./wallet-fake"; import type { Nuri } from "../src/model/types"; @@ -268,6 +270,42 @@ describe("switching identity", () => { expect(await documentsGivenTo("alice")).toEqual([]); }); + test("MID-DRAIN files nothing for the identity that arrives", async () => { + const { doc, inTransit } = await bobSharesWithAlice(); + // Test-side inspection only: the address is used to recognise the read in flight, and + // is never handed to an actor. + const aliceInbox = await userInbox("alice", "protected"); + await signIn("alice"); + + // The broker takes its time over the deposits read, and the page switches user INSIDE + // that window — a person clicking "sign in as Bob" while a push is being applied. The + // ownership guard has already passed by then; it ran at the start of the read. + const answering = fake.sparql_query.getMockImplementation()!; + let switched = false; + fake.sparql_query.mockImplementation(async (...args: unknown[]) => { + const answer = await answering(...args); + if (!switched && args[3] === aliceInbox && String(args[1]).includes(`${INBOX}:payload`)) { + switched = true; + setCurrentUser("bob"); + } + return answer; + }); + + fake._deliver(inTransit); + await converge(); + + // Bob was GIVEN nothing. (His own cap on the document is not evidence either way — he + // made it; what would be evidence is a Link, and there must be none.) Filed here, the + // cap addressed to Alice becomes a capability Bob holds at his next sign-in, and its + // real recipient is left with nothing at all. + expect(await documentsGivenTo("bob")).toEqual([]); + // …and Alice has lost nothing: an inbox is not consumed by a drain that abandoned it, + // so what was deposited for her is still there when she is the one connected. + await signIn("alice"); + await converge(); + expect(getCaps().capForHolder("alice", doc)).toBeDefined(); + }); + test("and Alice coming back finds the deposit still there to apply", async () => { const { doc, inTransit } = await bobSharesWithAlice(); await signIn("alice"); @@ -315,6 +353,50 @@ describe("a deposit that cannot be applied", () => { expect(getCaps().capForHolder("alice", first.doc)).toBeDefined(); }); + test("because its inbox could not be WATCHED is reported, and attempted again", async () => { + const { doc, inTransit } = await bobSharesWithAlice(); + const aliceInbox = await userInbox("alice", "protected"); + + // A repo that never pushes its initial `State` is what a refused subscription looks like + // from the bootstrap open's side, and it waits out its bounded fallback before giving up. + // Eight seconds of it, twice, is the production wait and not a unit test's. + setOpenTimeoutForTests(20); + + // The broker will not open a channel on Alice's inbox. Everything else about her session + // works — which is the point: the only symptom of a watch that was never established is + // that shares stop arriving. + const opening = fake.doc_subscribe!.getMockImplementation()!; + let refusing = true; + fake.doc_subscribe!.mockImplementation(async (...args: unknown[]) => { + if (refusing && args[0] === aliceInbox) throw new Error(`RepoNotFound: ${String(args[0])}`); + return opening(...args); + }); + + const reported = await whileWatchingTheLog(async () => { + await signIn("alice"); + await converge(); + }); + + // Said out loud, in this package's own words and under the connected identity — not left + // as the absence of a push. + expect(reported.filter((l) => /could not be watched/.test(l)).length).toBeGreaterThan(0); + expect(reported.find((l) => /could not be watched/.test(l))).toContain("[alice][polyfill]"); + + // Nothing is watching, so the deposit that lands now cannot be applied — and is not. + fake._deliver(inTransit); + await converge(); + expect(getCaps().capForHolder("alice", doc)).toBeUndefined(); + + // The broker recovers and Alice does something ordinary. The inbox that could not be + // opened was not written off for the session: it is subscribed to on the next + // enumeration, and its initial push finds the deposit still waiting. + refusing = false; + await storeRegistry.createEntityDoc("protected"); + await converge(); + expect(getCaps().capForHolder("alice", doc)).toBeDefined(); + expect(await documentsGivenTo("alice")).toContain(doc); + }); + test("never rejects into the application — nobody asked for this work", async () => { const { inTransit } = await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); @@ -338,3 +420,41 @@ describe("a deposit that cannot be applied", () => { expect(unhandled).toEqual([]); }); }); + +describe("a connection whose own work FAILED", () => { + test("still leaves the identity watched — being connected is what is observed", async () => { + const { doc, inTransit } = await bobSharesWithAlice(); + // The broker cannot answer for one of Alice's own stores, so the RESTORE fails and the + // connection rejects. She is connected regardless: `setCurrentUser` is synchronous and + // took effect before any of this ran, and nothing signs her back out. + const store = (await resolveAccount("alice"))?.docPublic; + if (store === undefined) throw new Error("the fixture did not give Alice a public store"); + fake._failReadsOn.add(store); + + setCurrentUser("alice"); + let rejected = false; + await whileWatchingTheLog(async () => { + try { + await connectedUser(); + } catch { + rejected = true; + } + await converge(); + }); + // The caller is still TOLD, and that rule is not what changes here: failing to reach the + // registers rejects, exactly as before. + expect(rejected).toBe(true); + + // The hiccup passes. Alice never touched the page. + fake._failReadsOn.delete(store); + fake._deliver(inTransit); + await converge(); + + // What she is owed is not the restore she lost — it is that a deposit made while she sits + // there converges. Watching used to be the LAST line of the connection work, so a restore + // that rejected skipped it and left her connected with nothing observing her inboxes: one + // hiccup at sign-in, and every share made afterwards was lost to her for the session. + expect(getCaps().capForHolder("alice", doc)).toBeDefined(); + expect(await documentsGivenTo("alice")).toContain(doc); + }); +}); diff --git a/packages/polyfill/test/subscribe.test.ts b/packages/polyfill/test/subscribe.test.ts index 29a681d..16f891e 100644 --- a/packages/polyfill/test/subscribe.test.ts +++ b/packages/polyfill/test/subscribe.test.ts @@ -1,5 +1,11 @@ import { test, expect, mock, afterAll } from "bun:test"; -import { docChangeType, subscribeDoc, subscribeDocs } from "../src/surface/subscribe"; +import { + docChangeType, + subscribeDoc, + subscribeDocReportingSetupFailure, + subscribeDocs, + type Unsubscribe, +} from "../src/surface/subscribe"; import { configure } from "../src/index"; import { configureStoreRegistry } from "../src/shared-wallet/bootstrap"; import { resetConfig, resetStoreRegistry } from "../src/shared-wallet/bootstrap"; @@ -225,3 +231,93 @@ test("the real subscription is released only when the LAST subscriber leaves", a stopSecond(); expect(ng.isSubscribed(A)).toBe(false); // now nobody is }); + +// --- what sharing one subscription must not COST a caller -------------------- +// +// Three things a per-caller `doc_subscribe` gave for free. The fan-out took each of them +// away when it was introduced, and each is invisible from the caller's side: nothing +// rejects, nothing logs, the document simply stops speaking to somebody. + +test("two subscriptions with the SAME handler are two subscriptions", async () => { + const ng = inject(); + const seen: unknown[] = []; + // One function, two callers. A module-level handler, a bound method or a shared arrow + // makes this ordinary rather than exotic — and keyed by the function, the second caller + // was never registered at all, so the first one's departure took it with it. + const handler = (r: unknown): void => { + seen.push(r); + }; + const stopFirst = subscribeDoc(A, handler); + const stopSecond = subscribeDoc(A, handler); + await tick(); + const afterInitial = seen.length; + + stopFirst(); // only the FIRST caller has left + expect(ng.isSubscribed(A)).toBe(true); + ng.push(A); + expect(seen.length).toBeGreaterThan(afterInitial); // the second one is still listening + + stopSecond(); + expect(ng.isSubscribed(A)).toBe(false); // …and now the last one has gone +}); + +test("a subscriber torn down inside another's handler does not receive that push", async () => { + const ng = inject(); + const seen: unknown[] = []; + let stopSecond: Unsubscribe | null = null; + let armed = false; + // The first handler tears the second one down mid-push. Which of the two runs first is an + // ordering no application controls, and `subscribeDoc` publishes that no `onChange` fires + // after unsubscribe — so the answer must not depend on it. + const stopFirst = subscribeDoc(A, () => { + if (armed) stopSecond?.(); + }); + stopSecond = subscribeDoc(A, (r) => seen.push(r)); + await tick(); + seen.length = 0; + + armed = true; + ng.push(A); + + expect(seen).toEqual([]); + stopFirst(); +}); + +test("a setup that FAILED does not poison the document for the next subscriber", async () => { + const failFor = new Set([A]); + const ng = inject(failFor); + const stopFirst = subscribeDoc(A, () => {}); + await tick(); + expect(ng.doc_subscribe).toHaveBeenCalledTimes(1); // …and it rejected + + // The document syncs; the broker can serve it now. A caller arriving after a transient + // rejection must get a real subscription, exactly as its own `doc_subscribe` would have. + failFor.delete(A); + const seen: unknown[] = []; + const stopSecond = subscribeDoc(A, (r) => seen.push(r)); + await tick(); + + expect(ng.doc_subscribe).toHaveBeenCalledTimes(2); // attempted again, not joined to a corpse + expect(seen.length).toBeGreaterThan(0); // its initial State + ng.push(A); + expect(seen.length).toBeGreaterThan(1); // and the changes that follow + stopFirst(); + stopSecond(); +}); + +test("a caller that asks to be told learns its subscription could not be opened", async () => { + inject(new Set([A])); + const failures: unknown[] = []; + const stop = subscribeDocReportingSetupFailure( + A, + () => {}, + (error) => failures.push(error), + ); + await tick(); + // Upstream `doc_subscribe` is async and rejects, so its caller learns. This wrapper + // returns synchronously, and a caller whose job depends on the subscription (the inbox + // observation) cannot otherwise tell "quiet" from "never opened". + expect(failures).toHaveLength(1); + expect(String(failures[0])).toContain("RepoNotFound"); + stop(); +});