diff --git a/packages/polyfill/src/emulated-verifier/inbox-observer.ts b/packages/polyfill/src/emulated-verifier/inbox-observer.ts index 0329bee..0504df4 100644 --- a/packages/polyfill/src/emulated-verifier/inbox-observer.ts +++ b/packages/polyfill/src/emulated-verifier/inbox-observer.ts @@ -82,6 +82,17 @@ interface Observation { enumerating: boolean; /** A trigger that arrived mid-cycle: the running one repeats once rather than lose it. */ enumerateAgain: boolean; + /** + * True while the shortfall reported by the last enumeration is still the SAME one — so a + * register that stays unreadable is reported once, not once per cycle. + * + * Every signal this observation listens to runs a full cycle, and a persistent shortfall + * is re-read by each of them, so the condition is reported as many times as the identity + * is nudged: one deposit produced three copies of the same line. That is what buries the + * report that matters under the report that repeats. Cleared the moment a list comes back + * WHOLE, so a shortfall that goes away and returns is a new occurrence and says so. + */ + shortfallReported: boolean; /** Enumerations and applications in flight — what {@link observationSettled} waits on. */ pending: Set>; } @@ -178,6 +189,11 @@ async function applyWhatArrived(obs: Observation, inbox: Nuri): Promise { * 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. + * + * Once per OCCURRENCE, though, not once per enumeration — see + * {@link Observation.shortfallReported}. The retry that repairs it is also what re-reads it, + * so a condition that persists is re-discovered by every signal; reporting each discovery + * says "it happened again" about the single thing that never stopped happening. */ async function watchTheInboxes(obs: Observation): Promise { if (!current(obs)) return; @@ -190,7 +206,10 @@ async function watchTheInboxes(obs: Observation): Promise { return; } if (!current(obs)) return; - if (listed.incomplete !== null) { + if (listed.incomplete === null) { + obs.shortfallReported = false; + } else if (!obs.shortfallReported) { + obs.shortfallReported = true; reportUnobserved("the inboxes to watch could not all be listed", listed.incomplete.error); } for (const inbox of listed.inboxes) { @@ -354,6 +373,7 @@ export async function startObservingInboxes(): Promise { holdings: null, enumerating: false, enumerateAgain: false, + shortfallReported: false, pending: new Set(), }; observation = obs; diff --git a/packages/polyfill/src/surface/subscribe.ts b/packages/polyfill/src/surface/subscribe.ts index abba2d9..72e202c 100644 --- a/packages/polyfill/src/surface/subscribe.ts +++ b/packages/polyfill/src/surface/subscribe.ts @@ -230,7 +230,15 @@ function fanOut(nuri: Nuri, entry: DocFanOut, resp: DocChange, type: DocChangeTy } } -/** Is `attempt` still the establish this fan-out is waiting on? */ +/** + * Is `attempt` still the establish this fan-out is waiting on? + * + * Two independent ways to stop being it, and each is the only one that catches its own case: + * the fan-out was RELEASED (last listener left, or `resetDocSubscriptions`), which drops it + * from the map and leaves the counter untouched — so only the map half sees it; or it was + * SUPERSEDED in place by {@link resubscribeDocs}, which keeps the very same entry in the map + * and bumps the counter — so only the counter half sees it. + */ function isCurrentAttempt(nuri: Nuri, entry: DocFanOut, attempt: number): boolean { return fanOuts.get(nuri) === entry && entry.attempt === attempt; } @@ -307,6 +315,33 @@ function releaseFanOut(nuri: Nuri, entry: DocFanOut): void { * 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. + * + * ── Everything here is gated on the ATTEMPT, never on the entry alone ────── + * A superseded establish must not deliver, must not re-seed the barrier, and must not + * install its unsubscribe: {@link resubscribeDocs} re-opens the channel ON THE SAME entry, + * so entry identity says "this fan-out still stands" and says nothing about which call it is + * waiting on. Gated on identity only, a first attempt that SUCCEEDS after a second one has + * replaced it fans the PREVIOUS session's `State` out to every listener, re-seeds + * `lastState` with it — the exact stale barrier {@link resubscribeDocs} cleared it to + * prevent — and overwrites `realUnsub`, so the eventual teardown releases the dead channel + * and leaves the live one subscribed upstream. + * + * ── A superseded attempt is RELEASED, not dropped ────────────────────────── + * Dropping its unsubscribe would leak a live subscription. Upstream's `CancelFn` closes only + * its OWN `tx` and never touches `branch_subscriptions` (`engine/verifier/src/verifier.rs:479-484`), + * so releasing a loser cannot silence the winner — and cross-session there IS no winner to + * silence: `resubscribeDocs` runs on a session change, the superseded call sits on the + * previous session's verifier, and nothing else will ever close it. + * + * ── Why the guard sits BEFORE the call, and not only after it ────────────── + * It is what makes the sentence above true rather than merely likely. Superseded while the + * session id was still resolving, this opens NO channel at all — so every attempt that owns + * one issued its `doc_subscribe` strictly before the current attempt existed, and by the + * global broker lock held for the whole call (`sdk/rust/src/local_broker.rs:3057`) it was + * served first. Whoever inserts last wins the branch (`verifier.rs:361`), so a superseded + * attempt is always the upstream LOSER, never the live channel. Without this guard the two + * calls could reach the broker in either order, and releasing the loser would be a coin flip + * on whether the document goes silent. */ async function establish( nuri: Nuri, @@ -322,17 +357,25 @@ async function establish( const doc_subscribe = ng.doc_subscribe as (...a: unknown[]) => unknown; try { const sid = await sessionId(); + // Superseded (or torn down) while the session id was resolving: open NOTHING. Cheaper + // than opening a channel to close it, and it is what keeps a superseded attempt from + // ever reaching the broker AFTER the attempt that replaced it. + if (!isCurrentAttempt(nuri, entry, attempt)) return; 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; + // A push that belongs to a torn-down fan-out, or to an establish that has been + // superseded — either way the real unsubscribe has not taken effect yet, and neither + // this fan-out's listeners nor its barrier are this call's to touch any more. + if (!isCurrentAttempt(nuri, entry, attempt)) return; const type = docChangeType(resp); // Remember the barrier for whoever joins next; a later `State` replaces it. if (type === "State") entry.lastState = { 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. + if (!isCurrentAttempt(nuri, entry, attempt)) { + // Everyone left, the fan-out was reset, or a later establish replaced this one — + // release the channel this call just opened, because nothing else holds it. (A + // released fan-out is out of the map, which is why the listener count is not asked + // about separately: the last one out already took the entry with it.) if (typeof unsub === "function") unsub(); return; } diff --git a/packages/polyfill/test/continuous-inbox-observation.test.ts b/packages/polyfill/test/continuous-inbox-observation.test.ts index c4cd9d6..0b69d01 100644 --- a/packages/polyfill/test/continuous-inbox-observation.test.ts +++ b/packages/polyfill/test/continuous-inbox-observation.test.ts @@ -545,3 +545,88 @@ describe("listing the inboxes when one of the two registers cannot be read", () } }); }); + +/** + * Bob shares TWO documents with Alice, each held back separately, so a test has two + * independent arrivals to drive the observation with rather than one. + */ +async function bobSharesTwiceWithAlice(): Promise { + await signIn("alice"); + await storeRegistry.createEntityDoc("protected"); // her first visit — now she exists + await signIn("bob"); + const aliceInbox = await userInbox("alice", "protected"); + const batches: Quad[][] = []; + for (let round = 0; round < 2; round += 1) { + const doc = await storeRegistry.createEntityDoc("protected"); + await inboxSurface.share(doc, "alice"); + batches.push(heldByTheBroker(aliceInbox)); + } + return batches; +} + +describe("a register that stays unreadable", () => { + /** + * The retry that would repair the shortfall is also what re-reads it. + * + * Every signal the observation listens to runs a full cycle, and each cycle re-discovers a + * condition that never stopped holding — so ONE deposit put three copies of the same line + * in the log. A report that repeats on its own is not more information; it is what buries + * the report that means something, and it invites reading a persistent fault as a + * recurring one. + */ + test("is reported ONCE, not once per enumeration", async () => { + const { inTransit } = await bobSharesWithAlice(); + 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); + + const reported = await whileWatchingTheLog(async () => { + try { + await signIn("alice"); + } catch { + // The restore rejects on the unreadable store; she is connected regardless. + } + await converge(); + fake._deliver(inTransit); + await converge(); + await converge(); + }); + fake._failReadsOn.delete(store); + + expect(reported.filter((l) => /could not all be listed/.test(l))).toHaveLength(1); + }); + + /** + * Once per OCCURRENCE, and a second occurrence is a real one. + * + * Silencing the repeat by remembering "already said" and never forgetting it would trade a + * noisy log for a mute one: the register breaking again, after a spell of working, is news + * — and it is exactly the case the retry exists for. + */ + test("is reported AGAIN once the list has come back WHOLE in between", async () => { + const [first, second] = await bobSharesTwiceWithAlice(); + 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); + const reported = await whileWatchingTheLog(async () => { + try { + await signIn("alice"); + } catch { + // As above: the restore rejects, the connection stands. + } + await converge(); // the shortfall, first occurrence + + fake._failReadsOn.delete(store); // the register is readable again… + fake._deliver(first ?? []); // …and an arrival makes the observation re-enumerate + await converge(); // the list comes back WHOLE + + fake._failReadsOn.add(store); // and then it breaks a second time + fake._deliver(second ?? []); + await converge(); // the shortfall, second occurrence + }); + fake._failReadsOn.delete(store); + + expect(reported.filter((l) => /could not all be listed/.test(l))).toHaveLength(2); + }); +}); diff --git a/packages/polyfill/test/subscribe.test.ts b/packages/polyfill/test/subscribe.test.ts index f2b534c..f30db58 100644 --- a/packages/polyfill/test/subscribe.test.ts +++ b/packages/polyfill/test/subscribe.test.ts @@ -46,12 +46,19 @@ function makeFakeNg(failFor: Set = new Set(), hangFor: Set = new // 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 hung: { + reject: ((error: unknown) => void) | null; + resolve: ((unsub: () => void) => void) | null; + cb: ((r: unknown) => void) | null; + live: boolean; + } = { reject: null, resolve: null, cb: null, live: false }; 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) => { + return await new Promise((resolve, reject) => { hung.reject = reject; + hung.resolve = resolve as (unsub: () => void) => void; + hung.cb = cb; }); } subs.set(nuri, cb); // whoever held this branch is dropped, without a word @@ -74,7 +81,47 @@ function makeFakeNg(failFor: Set = new Set(), hangFor: Set = new hung.reject = null; reject?.(new Error("RepoNotFound: late")); }; - return { doc_subscribe, push, isSubscribed, rejectHung, _subs: subs }; + /** + * The call that was left hanging finally answers — SUCCESSFULLY, and from the session it + * was placed against, which by now is the PREVIOUS one. + * + * Modelled apart from `subs` on purpose: a session is a verifier of its own, with its own + * `branch_subscriptions`, so this channel neither evicted nor was evicted by the one the + * new session opened (`engine/verifier/src/verifier.rs:361` inserts into `self`). It is + * genuinely live and genuinely pushing, and nothing but its own unsubscribe closes it — + * which is the whole reason a superseded attempt has to be released rather than dropped. + */ + const resolveHung = (): void => { + const resolve = hung.resolve; + hung.resolve = null; + hung.reject = null; + hung.live = true; + resolve?.(() => { + hung.live = false; + }); + }; + /** + * The PREVIOUS session pushes over the channel it just handed back. + * + * Unconditionally, even once that channel has been released: a cancel does not reach back + * and un-send what the broker already dispatched, which is precisely why the callback + * carries a guard of its own instead of trusting the unsubscribe to have taken effect. + */ + const pushHung = (payload: unknown): void => { + hung.cb?.(payload); + }; + /** Is the previous session's channel still open upstream? */ + const hungIsLive = (): boolean => hung.live; + return { + doc_subscribe, + push, + isSubscribed, + rejectHung, + resolveHung, + pushHung, + hungIsLive, + _subs: subs, + }; } function inject(failFor?: Set, hangFor?: Set) { @@ -414,3 +461,110 @@ test("a SUPERSEDED setup rejecting late does not let the next joiner evict the l stopLate(); stop(); }); + +/** A `State` that could only have come from the session `resubscribeDocs` walked away from. */ +const OLD_SESSION_STATE = { V0: { State: { doc: A, from: "the previous session" } } }; + +test("a SUPERSEDED setup that SUCCEEDS pushes to nobody, re-seeds nothing, and is released", async () => { + const { ng, seen, stop } = await hangingThenReopened(); + resubscribeDocs(); // the session rotated: a second establish, on the SAME fan-out + await tick(); + expect(ng.isSubscribed(A)).toBe(true); + const afterReopen = seen.length; + expect(afterReopen).toBeGreaterThan(0); // the current session is pushing + + // …and only now does the FIRST call answer — successfully, from the session that is gone. + ng.resolveHung(); + await tick(); + ng.pushHung(OLD_SESSION_STATE); + await tick(); + + // Its channel was released as it resolved: nothing else would ever close it, because it + // sits on the previous session's verifier where the new subscription never displaced it. + expect(ng.hungIsLive()).toBe(false); + + // Its push reaches nobody. The listeners are following the establish that replaced it, and + // handing them a verifier nobody is talking to any more is the staleness, not the cure. + expect(seen).not.toContainEqual(OLD_SESSION_STATE); + expect(seen.length).toBe(afterReopen); + + // Nor is the abandoned `State` left as the barrier: a joiner is replayed the CURRENT + // session's, which is what `resubscribeDocs` cleared `lastState` to guarantee. Replayed the + // other one, a bootstrap open calls a repo this session has never synced "synced". + const late: unknown[] = []; + const stopLate = subscribeDoc(A, (r) => late.push(r)); + await tick(); + expect(late.length).toBeGreaterThan(0); + expect(late).not.toContainEqual(OLD_SESSION_STATE); + + // And `realUnsub` is still the live call's, so the last listener out leaves NOTHING + // subscribed upstream — neither session's channel. + stopLate(); + stop(); + await tick(); + expect(ng.isSubscribed(A)).toBe(false); + expect(ng.hungIsLive()).toBe(false); +}); + +test("a SUPERSEDED setup that FAILS is logged, even though it is reported to nobody", async () => { + const { ng, failures, stop } = await hangingThenReopened(); + resubscribeDocs(); + await tick(); + + const logged: string[] = []; + const realError = console.error; + console.error = ((...a: unknown[]) => { + logged.push(a.map(String).join(" ")); + }) as typeof console.error; + try { + ng.rejectHung(); + await tick(); + } finally { + console.error = realError; + } + + // Told to nobody — the call that replaced it owns the outcome — but never passed over in + // silence: a rejection that reaches no caller AND no log is a failure that never happened, + // and this one is the only trace that the abandoned session refused at all. + expect(failures).toEqual([]); + expect(logged.some((l) => l.startsWith("[subscribe] doc_subscribe failed for"))).toBe(true); + stop(); +}); + +test("an establish SUPERSEDED while the session id resolves never reaches the broker at all", async () => { + const ng = makeFakeNg(); + configure({ ng: ng as unknown as Parameters[0]["ng"], useShape: (() => {}) as never }); + // A session lookup this test holds open, so the supersession lands while the first + // establish is between "started" and "placed its call" — the window `resubscribeDocs` + // actually opens, since it fires from a `resetOpenedRepos` that a session read preceded. + const gate: { release: (() => void) | null } = { release: null }; + const held = new Promise((r) => { + gate.release = r; + }); + configureStoreRegistry({ + getSession: async () => { + await held; + return SESSION; + }, + }); + + const seen: unknown[] = []; + const stop = subscribeDoc(A, (r) => seen.push(r)); + await tick(); + expect(ng.doc_subscribe).toHaveBeenCalledTimes(0); // still resolving the session + + resubscribeDocs(); // superseded before it ever placed its call + gate.release?.(); + await tick(); + + // ONE call, and it is the current attempt's. This is what makes "release a superseded + // attempt" safe rather than a coin flip: every attempt that OWNS a channel placed its call + // before the attempt replacing it existed, so the broker — which serialises the whole call + // under one lock — served it first, and whoever inserts last holds the branch. The + // superseded one is therefore always the evicted one, never the live channel. + expect(ng.doc_subscribe).toHaveBeenCalledTimes(1); + expect(ng.isSubscribed(A)).toBe(true); + expect(seen.length).toBeGreaterThan(0); + stop(); + expect(ng.isSubscribed(A)).toBe(false); +});