From 7b353007237b587c90aa30a80b117dbf05496b36 Mon Sep 17 00:00:00 2001 From: Sylvain Duchesne Date: Mon, 17 Aug 2026 12:17:59 +0200 Subject: [PATCH] fix: un rejet tardif ne ferme plus un canal vivant, un registre illisible ne perd plus toutes les inbox --- .../src/emulated-verifier/branch-registers.ts | 73 +++++++++-- .../src/emulated-verifier/inbox-observer.ts | 60 ++++++---- packages/polyfill/src/surface/subscribe.ts | 79 ++++++++++-- .../test/continuous-inbox-observation.test.ts | 113 ++++++++++++++++-- packages/polyfill/test/subscribe.test.ts | 101 +++++++++++++++- 5 files changed, 364 insertions(+), 62 deletions(-) diff --git a/packages/polyfill/src/emulated-verifier/branch-registers.ts b/packages/polyfill/src/emulated-verifier/branch-registers.ts index 7a4638a..7fd6cf7 100644 --- a/packages/polyfill/src/emulated-verifier/branch-registers.ts +++ b/packages/polyfill/src/emulated-verifier/branch-registers.ts @@ -482,20 +482,73 @@ export async function readInboxCapsFor(doc: Nuri): Promise { */ // @provenance myInboxes kind=aligned level=1 ref=engine/repo/src/types.rs:AddInboxCapV0 — the User branch answers 'which inboxes may I read'. The document-inbox half of this list is declared-not-wired — see `readInboxCapPairs` export async function myInboxes(): Promise { + const { inboxes, incomplete } = await enumerateMyInboxes(); + // The whole list or the failure that stopped it — the contract this function has always + // had, and the one `connect.connectedUser` is built on: not knowing which queues exist is + // the session failing to establish, and a short list would silently leave a delivered + // share un-drained. The original error is re-thrown, not wrapped: its caller reads it. + if (incomplete !== null) throw incomplete.error; + return inboxes; +} + +/** + * What could be listed of {@link myInboxes}, and whether that is ALL of it. + * + * `incomplete` is `null` when the list is whole, and otherwise carries what stopped the rest + * from being listed. Not an error CODE and not a flag on the array: the point is that a + * caller cannot read this answer without meeting the question "was there a failure", which + * is exactly what a short array on its own let everybody skip. + */ +export interface InboxEnumeration { + /** Every inbox this holder may read that COULD be listed — possibly not all of them. */ + inboxes: Nuri[]; + /** What stopped the rest from being listed, or `null` when nothing did. */ + incomplete: { error: unknown } | null; +} + +/** + * {@link myInboxes}, for the caller that can use a PARTIAL answer — the live observation. + * + * The list is built from two independent registers: the account record, which names the + * user's own two store inboxes, and the User branch, which names one per document it opened + * an inbox on. Built in one `try`, one unreachable register discarded BOTH halves — the two + * user inboxes were already in hand when the second read threw, and the throw dropped them + * on the floor. A broker hiccup spanning sign-in therefore left the identity connected with + * ZERO inboxes watched, not with the one register it could not reach missing: every deposit, + * including the ones addressed to the person by name, waited for the next connection. + * + * So each half answers for itself, and what came back is returned WITH the failure rather + * than instead of it. That is the line this package draws everywhere: reaching a register is + * infrastructure and may fail, but a failure must never come back looking like an absence — + * hence {@link InboxEnumeration.incomplete}, which the caller has to look at. + */ +export async function enumerateMyInboxes(): Promise { const holder = getCurrentUser(); - if (holder === null) return []; + if (holder === null) return { inboxes: [], incomplete: null }; const out: Nuri[] = []; - // BOTH of the user's inboxes — public and protected — since upstream a site carries - // one on each of those two store repos (`engine/verifier/src/site.rs:127-152`). - if ((await resolveAccount(holder)) !== null) { - for (const scope of ["public", "protected"] as const) out.push(await userInbox(holder, scope)); + try { + // BOTH of the user's inboxes — public and protected — since upstream a site carries + // one on each of those two store repos (`engine/verifier/src/site.rs:127-152`). + if ((await resolveAccount(holder)) !== null) { + for (const scope of ["public", "protected"] as const) { + out.push(await userInbox(holder, scope)); + } + } + } catch (error) { + // The account record is what the other half reads THROUGH (`readInboxCapPairs` resolves + // the same record to find the private store), so there is no second half to attempt. + return { inboxes: out, incomplete: { error } }; } - for (const { inbox } of await readInboxCapPairs()) { - // The record entitles this holder to read it — see the note above. - fileOwnInbox(holder, inbox); - out.push(inbox); + try { + for (const { inbox } of await readInboxCapPairs()) { + // The record entitles this holder to read it — see the note above. + fileOwnInbox(holder, inbox); + out.push(inbox); + } + } catch (error) { + return { inboxes: out, incomplete: { error } }; } - return out; + return { inboxes: out, incomplete: null }; } /** diff --git a/packages/polyfill/src/emulated-verifier/inbox-observer.ts b/packages/polyfill/src/emulated-verifier/inbox-observer.ts index f339510..0329bee 100644 --- a/packages/polyfill/src/emulated-verifier/inbox-observer.ts +++ b/packages/polyfill/src/emulated-verifier/inbox-observer.ts @@ -57,7 +57,7 @@ import { getCaps, getCurrentUser } from "../shared-wallet/bootstrap"; import { accessLogPrefix, logStage, shortNuri } from "../shared-wallet/access-log"; -import { myInboxes } from "./branch-registers"; +import { enumerateMyInboxes, type InboxEnumeration } from "./branch-registers"; import { drainInboxSerially, drainsSettled } from "./inbox-drain"; import { lookupAccount } from "../shared-wallet/account-registry"; import { processInbox } from "../surface/inbox"; @@ -78,12 +78,6 @@ 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. */ @@ -171,19 +165,35 @@ async function applyWhatArrived(obs: Observation, inbox: Nuri): Promise { * * 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. + * + * ── A list that came back SHORT still gets watched ──────────────────────── + * `enumerateMyInboxes` and not `myInboxes`: the list is read from two independent registers + * and one of them being unreachable used to discard the other's answer too, so a broker + * hiccup spanning sign-in left the identity connected with ZERO inboxes watched rather than + * with one register's worth missing. Whatever came back is watched here and now — including, + * crucially, the user's own two inboxes, which are where a share addressed to a PERSON + * lands, and therefore the channel by which a read-only identity converges at all. + * + * The shortfall is reported, never passed over: an incomplete list is not a complete one, + * and the inboxes it is missing are watched on the next enumeration — which is what the + * register push and the holdings signal are for. Applying what lands in a user inbox is + * itself one of those signals (it files a cap), so the ordinary case repairs itself. */ async function watchTheInboxes(obs: Observation): Promise { if (!current(obs)) return; - let inboxes: Nuri[]; + let listed: InboxEnumeration; try { - inboxes = await myInboxes(); + listed = await enumerateMyInboxes(); } catch (error) { - // Not knowing WHICH inboxes exist is the whole observation failing, not one queue. + // Not knowing WHICH inboxes exist AT ALL 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 (listed.incomplete !== null) { + reportUnobserved("the inboxes to watch could not all be listed", listed.incomplete.error); + } + for (const inbox of listed.inboxes) { if (obs.inboxes.has(inbox)) continue; try { // The push is the signal; the drain is the work. The FIRST push is the initial @@ -212,20 +222,28 @@ async function watchTheInboxes(obs: Observation): Promise { } /** - * The subscription on `inbox` could not be OPENED — report it, forget it, and try once more. + * The subscription on `inbox` could not be OPENED — report it, and forget it. * * **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. + * ── And that is ALL it does: recovery waits for the next SIGNAL ─────────── + * Until 2026-08-17 this also re-enumerated on the spot, once per inbox. That call runs + * synchronously out of the very rejection it is reacting to, so it asks the broker that has + * just refused, in the same turn, with nothing having changed — and spends the one extra + * attempt it was allowed doing it. Deleting it left the whole suite green, which is the + * measure of what it achieved. + * + * What actually repairs this is an event, and this module already listens to the two that + * exist: the register push (`watchTheRegister`) and the held-caps signal + * (`CapRegistry.onChange`), each of which re-enters {@link enumerate} and finds this inbox + * unwatched. Applying anything at all fires the second one, so a session that is doing + * something catches up on its own. A session that is doing nothing waits for its next + * connection — where an unapplied deposit has always waited, since an inbox is not consumed + * by being unread. Polling for it is ruled out here for the same reason it is in + * `inbox-processor.ts`: this regime is push-driven, and a timer is what it replaced. */ function watchFailed(obs: Observation, inbox: Nuri, error: unknown): void { if (!current(obs)) return; @@ -239,9 +257,6 @@ function watchFailed(obs: Observation, inbox: Nuri, error: unknown): void { } } reportUnobserved("this inbox could not be watched: " + shortNuri(inbox), error); - if (obs.reattempted.has(inbox)) return; - obs.reattempted.add(inbox); - track(obs, enumerate(obs)); } /** @@ -337,7 +352,6 @@ 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/subscribe.ts b/packages/polyfill/src/surface/subscribe.ts index bb30d4a..abba2d9 100644 --- a/packages/polyfill/src/surface/subscribe.ts +++ b/packages/polyfill/src/surface/subscribe.ts @@ -175,12 +175,26 @@ interface DocFanOut { * 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 + * ({@link releaseFanOut}), and the CURRENT setup FAILING ({@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; + /** + * WHICH establish this fan-out is waiting on — bumped by {@link beginEstablish} every time + * one is kicked off, and captured by that call for the whole of its life. + * + * Two establishes can be in flight over the SAME entry: {@link resubscribeDocs} re-opens the + * channel of a fan-out whose first attempt has not settled yet, and it re-opens it on the + * entry rather than on a fresh one, precisely so the listeners are kept. So `entry` identity + * — the only currency this module had — answers "is this fan-out still the one for this + * NURI", and cannot answer "is this attempt still the one this fan-out is waiting on". A + * counter can, and that second question is the one a LATE outcome has to ask: the first + * attempt rejecting long after the second SUCCEEDED is not this document failing, it is one + * superseded call finally answering. + */ + attempt: number; /** * 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. @@ -216,6 +230,25 @@ function fanOut(nuri: Nuri, entry: DocFanOut, resp: DocChange, type: DocChangeTy } } +/** Is `attempt` still the establish this fan-out is waiting on? */ +function isCurrentAttempt(nuri: Nuri, entry: DocFanOut, attempt: number): boolean { + return fanOuts.get(nuri) === entry && entry.attempt === attempt; +} + +/** + * Kick off the one real subscription for `entry` — the ONLY way an establish is started. + * + * It stamps the attempt (see {@link DocFanOut.attempt}) as it starts it, so an outcome that + * arrives late can tell whether it is still the one being waited on. Being the only door is + * what makes that true: a second call site that forgot to bump the counter would hand its + * establish the number of the one it just superseded. + */ +function beginEstablish(nuri: Nuri, entry: DocFanOut, ng: { doc_subscribe?: unknown }): void { + entry.establishing = true; + entry.attempt += 1; + void establish(nuri, entry, ng, entry.attempt); +} + /** * The one real subscription could not be OPENED: forget the attempt, and say so. * @@ -226,11 +259,24 @@ function fanOut(nuri: Nuri, entry: DocFanOut, resp: DocChange, type: DocChangeTy * **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. + * + * ── Only for the attempt still being WAITED ON ───────────────────────────── + * Both halves act on the fan-out as it stands NOW, so a SUPERSEDED attempt must do neither. + * A rejection that arrives after {@link resubscribeDocs} has already re-opened the channel is + * not this document failing — the channel is open and pushing. Told anyway, the inbox + * observation dropped its entry and released it, which closed the WORKING subscription, + * printed "this inbox could not be watched" about an inbox that was, and left the identity + * with a silent queue. Clearing `establishing` was the same mistake one level down: it says + * "nothing is running" while the second attempt is, so the next joiner opens a third — and a + * second `doc_subscribe` on a branch EVICTS the one before it. + * + * Superseded, it is therefore LOGGED and nothing else: the call that replaced it owns the + * outcome, and reports its own failure if it has one. */ -function reportSetupFailure(nuri: Nuri, entry: DocFanOut, error: unknown): void { +function reportSetupFailure(nuri: Nuri, entry: DocFanOut, error: unknown, attempt: number): void { console.error("[subscribe] doc_subscribe failed for", nuri, error); + if (!isCurrentAttempt(nuri, entry, attempt)) return; 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; @@ -262,7 +308,12 @@ function releaseFanOut(nuri: Nuri, entry: DocFanOut): void { * 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 { +async function establish( + nuri: Nuri, + entry: DocFanOut, + ng: { doc_subscribe?: unknown }, + attempt: number, +): 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 @@ -287,7 +338,7 @@ async function establish(nuri: Nuri, entry: DocFanOut, ng: { doc_subscribe?: unk } entry.realUnsub = typeof unsub === "function" ? unsub : null; } catch (error) { - reportSetupFailure(nuri, entry, error); + reportSetupFailure(nuri, entry, error, attempt); } } @@ -345,8 +396,7 @@ export function resubscribeDocs(): void { console.error("[subscribe] unsubscribe failed for", nuri, error); } } - entry.establishing = true; - void establish(nuri, entry, ng); + beginEstablish(nuri, entry, ng); } } @@ -422,7 +472,13 @@ export function subscribeDocUnguarded( let entry = fanOuts.get(nuri); if (!entry) { - entry = { listeners: new Set(), realUnsub: null, establishing: false, lastState: null }; + entry = { + listeners: new Set(), + realUnsub: null, + establishing: false, + attempt: 0, + lastState: null, + }; fanOuts.set(nuri, entry); } const joined = entry; @@ -443,8 +499,7 @@ export function subscribeDocUnguarded( } if (!joined.establishing) { - joined.establishing = true; - void establish(nuri, joined, ng as { doc_subscribe?: unknown }); + beginEstablish(nuri, joined, ng as { doc_subscribe?: unknown }); } let stopped = false; diff --git a/packages/polyfill/test/continuous-inbox-observation.test.ts b/packages/polyfill/test/continuous-inbox-observation.test.ts index f41e2ce..c4cd9d6 100644 --- a/packages/polyfill/test/continuous-inbox-observation.test.ts +++ b/packages/polyfill/test/continuous-inbox-observation.test.ts @@ -35,6 +35,7 @@ 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 { enumerateMyInboxes, myInboxes } from "../src/emulated-verifier/branch-registers"; 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"; @@ -353,7 +354,7 @@ 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 () => { + test("because its inbox could not be WATCHED is reported, and the next signal re-attempts it", async () => { const { doc, inTransit } = await bobSharesWithAlice(); const aliceInbox = await userInbox("alice", "protected"); @@ -387,9 +388,14 @@ describe("a deposit that cannot be applied", () => { 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. + // The broker recovers and Alice does something ordinary — which is a SIGNAL, not a + // coincidence: creating anything files caps, and the held-caps channel re-enters the + // enumeration. That is the whole of the repair, and it is deliberately the whole of it: + // a re-attempt fired from the rejection itself asks the broker that has just refused, in + // the same turn, with nothing having changed. The inbox that could not be opened was not + // written off for the session — the failed entry is forgotten, so this enumeration + // subscribes again as if it had never been attempted, and the initial push of that new + // subscription finds the deposit still waiting. refusing = false; await storeRegistry.createEntityDoc("protected"); await converge(); @@ -422,18 +428,30 @@ describe("a deposit that cannot be applied", () => { }); describe("a connection whose own work FAILED", () => { - test("still leaves the identity watched — being connected is what is observed", async () => { + /** + * Aimed at the PRIVATE store, and that is the whole test. + * + * It used to fail reads on `docPublic`, which makes the restore reject and leaves the + * enumeration of the inboxes untouched — so it proved that watching survives a failure that + * was never going to threaten it. The private store is the one the connection restores from + * AND the register that says which inboxes exist, so failing it is the case that actually + * decides: listing the inboxes reads it, and one throw used to discard the two user inboxes + * that had ALREADY been listed before it. The identity was then connected with nothing + * watched at all, and a person who only reads — who never creates anything, so never fires + * a signal — had no way back for the rest of the session. + */ + test("still leaves the identity watched — including when the failing store is the register", async () => { const { doc, inTransit } = await bobSharesWithAlice(); - // The broker cannot answer for one of Alice's own stores, so the RESTORE fails and the + // The broker cannot answer for Alice's private store, 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"); + const store = (await resolveAccount("alice"))?.docPrivate; + if (store === undefined) throw new Error("the fixture did not give Alice a private store"); fake._failReadsOn.add(store); setCurrentUser("alice"); let rejected = false; - await whileWatchingTheLog(async () => { + const reported = await whileWatchingTheLog(async () => { try { await connectedUser(); } catch { @@ -444,17 +462,86 @@ describe("a connection whose own work FAILED", () => { // 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); + // …and so is the log, about the half of the list that could not be read. A short list + // that says nothing is a failure wearing the face of an absence, which is the one thing + // this package will not do — the inboxes it names are watched, the ones it does not are + // owed a next enumeration, and both facts have to be legible. + expect(reported.filter((l) => /could not all be listed/.test(l)).length).toBeGreaterThan(0); + expect(reported.find((l) => /could not all be listed/.test(l))).toContain("[alice][polyfill]"); - // The hiccup passes. Alice never touched the page. + // The hiccup passes. Alice never touched the page — no sign-in, no document created, + // nothing that could stand in for the watching she is owed. 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. + // there converges. Her own two inboxes are where a share addressed to her by NAME lands, + // and they were listed before the register threw; watching them is what makes this + // session behave like every other one. expect(getCaps().capForHolder("alice", doc)).toBeDefined(); expect(await documentsGivenTo("alice")).toContain(doc); }); + +}); + +/** + * The list the observation works from, asked directly. + * + * It is built from two independent registers — the account record, which names the user's own + * two store inboxes, and the User branch, which names one per document it opened an inbox on + * — and the two fail independently. What a caller may do with a half-read list depends + * entirely on being TOLD it is half-read, so both halves of that answer are pinned here + * rather than only through the behaviour above. + */ +describe("listing the inboxes when one of the two registers cannot be read", () => { + test("comes back as what WAS listed plus the failure — never as a short list", async () => { + await signIn("alice"); + // A document inbox: a record on the User branch of the private store, which is the + // register the broker is about to stop answering for. + const note = await storeRegistry.createEntityDoc("public"); + await storeRegistry.openDocumentInbox(note); + await converge(); + const store = (await resolveAccount("alice"))?.docPrivate; + if (store === undefined) throw new Error("the fixture did not give Alice a private store"); + + const whole = await enumerateMyInboxes(); + expect(whole.incomplete).toBeNull(); + expect(whole.inboxes).toContain(inboxOnTheNote(note)); + + fake._failReadsOn.add(store); + const partial = await enumerateMyInboxes(); + fake._failReadsOn.delete(store); + + // Her own two inboxes were in hand before the second register threw. Discarding them + // with it is what left an identity connected with ZERO inboxes watched. + expect(partial.inboxes).toEqual([ + await userInbox("alice", "public"), + await userInbox("alice", "protected"), + ]); + // …and the shortfall travels WITH them: an answer that came back short while looking + // complete is a failure disguised as an absence, which is the fault this package keeps + // closing. The document inbox is missing from the list and that fact is legible. + expect(partial.incomplete).not.toBeNull(); + expect(String(partial.incomplete?.error)).toContain("RepoNotFound"); + expect(partial.inboxes).not.toContain(inboxOnTheNote(note)); + }); + + test("still REJECTS for the caller that cannot use a partial list", async () => { + await signIn("alice"); + await storeRegistry.openDocumentInbox(await storeRegistry.createEntityDoc("public")); + await converge(); + const store = (await resolveAccount("alice"))?.docPrivate; + if (store === undefined) throw new Error("the fixture did not give Alice a private store"); + + fake._failReadsOn.add(store); + try { + // `connect.connectedUser` drains this list, and a queue missing from it is a delivered + // share silently never applied. Not knowing which queues exist is the session failing + // to establish, and that contract is not what the partial answer above relaxes. + await expect(myInboxes()).rejects.toThrow(/RepoNotFound/); + } finally { + fake._failReadsOn.delete(store); + } + }); }); diff --git a/packages/polyfill/test/subscribe.test.ts b/packages/polyfill/test/subscribe.test.ts index 16f891e..f2b534c 100644 --- a/packages/polyfill/test/subscribe.test.ts +++ b/packages/polyfill/test/subscribe.test.ts @@ -1,6 +1,7 @@ import { test, expect, mock, afterAll } from "bun:test"; import { docChangeType, + resubscribeDocs, subscribeDoc, subscribeDocReportingSetupFailure, subscribeDocs, @@ -40,10 +41,19 @@ const SESSION: RegistrySession = { sessionId: "sid-1", privateStoreId: "PRIV" }; * 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()) { +function makeFakeNg(failFor: Set = new Set(), hangFor: Set = new Set()) { const subs = new Map void>(); + // A call the broker has neither answered nor refused yet, so a LATER call can overtake it + // and this one can settle afterwards. Held on an object rather than in a `let` so its + // type survives being written from one closure and read from another. + const hung: { reject: ((error: unknown) => void) | null } = { reject: null }; const doc_subscribe = mock(async (nuri: string, _sid: unknown, cb: (r: unknown) => void) => { if (failFor.has(nuri)) throw new Error(`RepoNotFound: ${nuri}`); + if (hangFor.has(nuri)) { + return await new Promise((_resolve, reject) => { + hung.reject = reject; + }); + } 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. @@ -58,11 +68,17 @@ function makeFakeNg(failFor: Set = new Set()) { subs.get(nuri)?.({ V0: { Patch: { doc: nuri } } }); }; const isSubscribed = (nuri: string): boolean => subs.has(nuri); - return { doc_subscribe, push, isSubscribed, _subs: subs }; + /** The call that was left hanging finally answers — with a refusal. */ + const rejectHung = (): void => { + const reject = hung.reject; + hung.reject = null; + reject?.(new Error("RepoNotFound: late")); + }; + return { doc_subscribe, push, isSubscribed, rejectHung, _subs: subs }; } -function inject(failFor?: Set) { - const ng = makeFakeNg(failFor); +function inject(failFor?: Set, hangFor?: Set) { + const ng = makeFakeNg(failFor, hangFor); configure({ ng: ng as any, useShape: (() => {}) as any }); // Synchronous fake store → no sync lag; disable the anti-fork retry backoff. configureStoreRegistry({ getSession: async () => SESSION }); @@ -321,3 +337,80 @@ test("a caller that asks to be told learns its subscription could not be opened" expect(String(failures[0])).toContain("RepoNotFound"); stop(); }); + +/** + * Two establishes over ONE fan-out, and the first one answering last. + * + * `resubscribeDocs` re-opens the channel of a fan-out whose first `doc_subscribe` has not + * settled yet — that is the whole point of it, since the session it was opened against is + * gone — and it re-opens it on the SAME entry so the listeners are kept. The two calls + * therefore race, and the broker is under no obligation to answer them in order. + */ +async function hangingThenReopened(): Promise<{ + ng: ReturnType; + failures: unknown[]; + seen: unknown[]; + stop: Unsubscribe; +}> { + const hangFor = new Set([A]); + const ng = inject(new Set(), hangFor); + const failures: unknown[] = []; + const seen: unknown[] = []; + const stop = subscribeDocReportingSetupFailure( + A, + (r) => seen.push(r), + (error) => failures.push(error), + ); + // Awaited before the broker is allowed to answer: `establish` resolves the session id + // first, so the call this has to leave hanging has not been placed yet. + await tick(); + hangFor.delete(A); + return { ng, failures, seen, stop }; +} + +test("a SUPERSEDED setup rejecting late is not reported as this document failing", async () => { + const { ng, failures, seen, stop } = await hangingThenReopened(); + expect(ng.isSubscribed(A)).toBe(false); // the first call has not answered + + resubscribeDocs(); // the session rotated: a second establish, on the same fan-out + await tick(); + expect(ng.isSubscribed(A)).toBe(true); + const before = seen.length; + expect(before).toBeGreaterThan(0); // …and it is pushing + + ng.rejectHung(); // …and only now does the first call refuse + await tick(); + + // Nothing failed: the document is subscribed and pushing. Told otherwise, the caller whose + // job depends on this subscription (the inbox observation) releases the entry it holds — + // which closes the WORKING channel and reports an inbox that is watched as unwatchable. + expect(failures).toEqual([]); + expect(ng.isSubscribed(A)).toBe(true); + ng.push(A); + expect(seen.length).toBeGreaterThan(before); + stop(); +}); + +test("a SUPERSEDED setup rejecting late does not let the next joiner evict the live channel", async () => { + const { ng, stop } = await hangingThenReopened(); + resubscribeDocs(); + await tick(); + expect(ng.doc_subscribe).toHaveBeenCalledTimes(2); + + ng.rejectHung(); + await tick(); + + // The second establish still stands, so this joiner must join it. Counting the late + // rejection as "nothing is running any more" opens a THIRD `doc_subscribe` — and a second + // subscribe on a branch evicts the one before it, so the joiner's own call is what silences + // everybody already listening. + const late: unknown[] = []; + const stopLate = subscribeDoc(A, (r) => late.push(r)); + await tick(); + expect(ng.doc_subscribe).toHaveBeenCalledTimes(2); + expect(late.length).toBeGreaterThan(0); // replayed the barrier, as any late joiner is + ng.push(A); + expect(late.length).toBeGreaterThan(1); + stopLate(); + stop(); +});