From dcf3814446e6997e02c5c3409de85f465fb43f23 Mon Sep 17 00:00:00 2001 From: Sylvain Duchesne Date: Thu, 24 Sep 2026 13:17:10 +0200 Subject: [PATCH] fix: trois defauts trouves par la revue adverse sur le registre derive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Chacun reproduit avant d'etre corrige, chacun epingle par un test qui echoue si la correction est retiree. 1. Une barriere pouvait survivre a la tentative qui l'avait produite. `getSyncState` repondait `"synced"` sur le seul `barrierCrossed`, alors que son propre commentaire disait « sur le canal qui tient MAINTENANT » : `reportSetupFailure` efface `establishing` et laisse `lastState`, donc un `doc_subscribe` qui poussait un `State` puis echouait laissait une autorite - et `noRowsMeansAbsent` prend une lecture a zero ligne pour une absence definitive. Le registre precedent repondait `"timed-out"` la ; strict est la lecture de cette famille, son cout est la latence, jamais la perte. 2. `release` ne repondait pas « est-ce que j'ecoute encore ». `resetDocSubscriptions()` largue les auditeurs et supprime l'entree sans rien dire, donc le handle continuait de paraitre vivant ; si autre chose souscrivait d'abord au meme document, le court-circuit sautait l'ouverture due et le document restait tenu par un auditeur libre de partir. Le fan-out numerote desormais ses incarnations, et l'ouverture compare. 3. Une tentative sans reponse TIENT quand meme. Elle ne comptait pas, si bien que chaque lecture d'un document dont le `doc_subscribe` ne repond jamais repayait le repli borne - 8 s en production, sur le chemin de la lecture ancree. Un second `doc_subscribe` est de toute facon impossible tant qu'un est en vol, donc la tentative fraiche n'achetait rien. Les trois questions sont maintenant une fonction nommee, `stillStanding`, et chacune est posee a qui detient la reponse. --- .../src/emulated-verifier/open-repo.ts | 80 ++++++++++++--- packages/polyfill/src/surface/subscribe.ts | 28 +++++- packages/polyfill/test/open-repo.test.ts | 98 +++++++++++++++++++ 3 files changed, 191 insertions(+), 15 deletions(-) diff --git a/packages/polyfill/src/emulated-verifier/open-repo.ts b/packages/polyfill/src/emulated-verifier/open-repo.ts index f0b6d21..7364afb 100644 --- a/packages/polyfill/src/emulated-verifier/open-repo.ts +++ b/packages/polyfill/src/emulated-verifier/open-repo.ts @@ -148,6 +148,17 @@ interface BootstrapOpen { * (`resetDocSubscriptions`) and nothing here would be told. */ release: Unsubscribe | null; + /** + * WHICH fan-out {@link release}'s listener was added to (`surface/subscribe.ts`, + * {@link DocChannel.incarnation}). + * + * `release` alone does NOT answer "am I still listening", and believing it did was a defect of + * this repair's first draft: `resetDocSubscriptions()` - which the PUBLISHED `configure()` + * calls - drops every listener and deletes the entry, and nothing here is told. Held a + * non-null `release` against a fan-out somebody else had since created, this module skipped an + * open it owed and the document was left held by a listener free to leave. + */ + incarnation: number; /** * How to END the wait from outside it - {@link resetOpenedRepos}, the only caller; `null` * once the wait is over. An open normally settles on the first `State` or on its own @@ -173,12 +184,46 @@ const opens = new Map(); * writes nothing - object identity doing what a counter used to. */ function disowned(): BootstrapOpen { - return { done: Promise.resolve(), release: null, settle: null, waiting: false }; + return { done: Promise.resolve(), release: null, incarnation: 0, settle: null, waiting: false }; } /** The session id the current {@link opens} entries belong to. A change means a * new verifier session (fresh `self.repos`) → the registry must be invalidated. */ let boundSessionId: string | number | null = null; +/** + * Is this module's own bootstrap open still STANDING for `nuri` - so that a caller asking for an + * open has nothing to do? + * + * Three questions, and each catches a case the others cannot. Every one of them is asked of the + * thing that owns the answer rather than remembered here, which is the whole of this module's + * invariant. + * + * - **do I hold a listener at all?** `release` is nulled by a refused subscribe and by + * {@link resetOpenedRepos}. Without this a document whose subscription a reset took away + * would never be re-held for the new session. + * - **is that listener still in the fan-out that stands NOW?** `resetDocSubscriptions()` drops + * every listener and deletes the entry without telling anybody, so `release` goes on looking + * live; if something else has since subscribed to the same document, the channel is `"open"` + * and the two questions above both pass while this module holds nothing. The document is then + * kept open only by a listener that is free to leave. + * - **does an attempt stand behind it?** A fan-out whose establish was REFUSED keeps its + * listeners and stands for nothing; the next caller must subscribe again. + * + * An attempt that has not ANSWERED yet (`"opening"`) counts as standing, deliberately. A second + * `doc_subscribe` cannot be placed while one is in flight - the fan-out dedupes it, and upstream + * it would EVICT the first - so a fresh attempt here buys nothing and costs the caller the whole + * bounded fallback again, 8s in production, on every read of a document whose subscribe never + * answers. Asking live is what makes that safe: the moment the attempt stops standing this + * returns false, where a RECORD of openness written at the same moment would have stood for the + * session. + */ +function stillStanding(nuri: Nuri, open: BootstrapOpen | undefined): boolean { + if (open === undefined || open.release === null) return false; + const channel = docChannel(nuri); + if (channel.incarnation !== open.incarnation) return false; + return channel.state !== "none"; +} + /** * Max wait (ms) for the initial-state push before proceeding with the read anyway. * The push normally lands quickly once the repo loads; the timeout only guards the @@ -255,12 +300,19 @@ export function resetOpenedRepos(): void { // @provenance getSyncState kind=aligned level=1 ref=engine/net/src/app_protocol.rs:AppResponseV0 — reports whether the first `State` arrived; upstream this is not observable from JS export function getSyncState(nuri: Nuri): SyncState | "unknown" { const channel = docChannel(nuri); - // The first `State` on the channel standing NOW. Asked first and asked of the fan-out, - // because that is the one record of it: a barrier is forgotten when its channel is released - // or its session changes, so this cannot report one that has gone. Answered even for a + // The first `State` on the channel STANDING NOW - both halves, and the second was missing in + // this repair's first draft. `barrierCrossed` alone said `"synced"` for a fan-out whose + // establish had been REFUSED after pushing a `State`: `reportSetupFailure` clears + // `establishing` and leaves `lastState`, so the barrier outlived the attempt that produced it, + // and `"synced"` is AUTHORITY - `index-deposit.ts` `noRowsMeansAbsent` then calls a zero-row + // read a definitive absence. The previous registry answered `"timed-out"` there, and strict is + // the reading this family is built on: the cost of it is latency, never loss. + // + // Asked of the fan-out rather than remembered, because that is the one record of it: a barrier + // is forgotten when its channel is released and when its session changes. Answered even for a // document this module never opened - an application's own subscription crosses the same // barrier, and `"synced"` is then simply true. - if (channel.barrierCrossed) return "synced"; + if (channel.state !== "none" && channel.barrierCrossed) return "synced"; const open = opens.get(nuri); // Never opened here, or opened on the no-`doc_subscribe` path, which records nothing // because there is no barrier to have missed. @@ -347,13 +399,9 @@ export async function openRepoUnguarded(nuri: Nuri): Promise { // A caller arriving while the bounded wait is still running JOINS it, and waits on the same // barrier rather than stepping past it. if (standing?.waiting === true) return standing.done; - // A caller arriving when this module's own listener is still in place AND the channel behind - // it has ANSWERED has nothing to do. Both halves, and each catches a defect the other - // cannot: without the listener half, a document whose subscription a reset took away would - // never be re-held for the new session; without the channel half - which is the whole of - // this task - a document whose `doc_subscribe` has not answered, or whose fan-out was torn - // down by a `configure()`, reads as open for ever and nothing re-subscribes it. - if (standing !== undefined && standing.release !== null && docChannel(nuri).state === "open") return; + // A caller arriving when this module's bootstrap open still STANDS has nothing to do - three + // questions, all asked live, enumerated on {@link stillStanding}. + if (stillStanding(nuri, standing)) return; // No reactive primitive on the injected ng (fake-ng unit suite): nothing to open, // no `State` to await → preserve the old immediate-resolve behaviour so `bun test` @@ -375,7 +423,7 @@ export async function openRepoUnguarded(nuri: Nuri): Promise { // Published BEFORE the subscribe, so a concurrent caller joins this wait instead of opening // a second time - and carrying nothing but the wait, because at this instant nothing is // true yet. What the previous registry put here was a claim of openness. - const mine: BootstrapOpen = { done: Promise.resolve(), release: null, settle: null, waiting: true }; + const mine: BootstrapOpen = { done: Promise.resolve(), release: null, incarnation: 0, settle: null, waiting: true }; opens.set(nuri, mine); mine.done = (async () => { @@ -425,6 +473,7 @@ export async function openRepoUnguarded(nuri: Nuri): Promise { console.error(accessLogPrefix() + " ensureRepoOpen: subscribe failed for " + shortNuri(nuri) + ":", error); const listening = mine.release; mine.release = null; + mine.incarnation = 0; try { listening?.(); } catch { @@ -449,6 +498,10 @@ export async function openRepoUnguarded(nuri: Nuri): Promise { }, onSetupFailed, ); + // WHICH fan-out that listener went into, so this open can later tell whether it is still + // in the one that stands. Read straight back rather than assumed: the call above either + // joined an existing fan-out or created one. + mine.incarnation = docChannel(nuri).incarnation; // A record this open REPLACED still holds a listener, and its wait is over - so nothing // will ever release it. Done after the new listener has joined, so the fan-out keeps its // entry and its establish, and this does not turn into a second `doc_subscribe` on a @@ -456,6 +509,7 @@ export async function openRepoUnguarded(nuri: Nuri): Promise { if (standing !== undefined && standing.release !== null) { const superseded = standing.release; standing.release = null; + standing.incarnation = 0; try { superseded(); } catch { diff --git a/packages/polyfill/src/surface/subscribe.ts b/packages/polyfill/src/surface/subscribe.ts index 8c3d93a..df57f76 100644 --- a/packages/polyfill/src/surface/subscribe.ts +++ b/packages/polyfill/src/surface/subscribe.ts @@ -209,6 +209,18 @@ interface DocFanOut { * Its own `doc_subscribe` would have pushed one; the fan-out owes it the same. */ lastState: { resp: DocChange; type: DocChangeType } | null; + /** + * WHICH fan-out this is for this NURI - a new number for every entry ever created, so a + * caller holding a listener can tell whether the entry it joined is still the one in the map. + * + * It exists because {@link resetDocSubscriptions} drops every listener WITHOUT telling the + * callers who took them out, and a caller cannot see that from its own unsubscribe function: + * `emulated-verifier/open-repo.ts` held a listener handle that went on reading as "I am still + * listening" after the entry it belonged to had been deleted, and skipped an open it owed. The + * entry OBJECT already answers this inside the module ({@link isCurrentAttempt}); a number is + * what can leave it. + */ + readonly incarnation: number; } /** @@ -222,6 +234,9 @@ interface DocFanOut { */ const fanOuts = new Map(); +/** How many fan-outs have ever been created - see {@link DocFanOut.incarnation}. */ +let fanOutsEverCreated = 0; + /** * What the one real subscription for a document IS, right now. * @@ -259,6 +274,14 @@ export interface DocChannel { * channel being released, and the session changing under it ({@link resubscribeDocs}). */ readonly barrierCrossed: boolean; + /** + * WHICH fan-out is standing for this document, or `0` when none is. + * + * A caller that took a listener out compares this with what it saw then: a different number + * means its listener was dropped from under it and the entry replaced, which + * {@link resetDocSubscriptions} does without a word. See {@link DocFanOut.incarnation}. + */ + readonly incarnation: number; } /** @@ -270,13 +293,13 @@ export interface DocChannel { // @provenance docChannel kind=invention level=none ref=none — a question about this module's own fan-out, which exists because a branch upstream holds a single subscriber; there is nothing to align it with export function docChannel(nuri: Nuri): DocChannel { const entry = fanOuts.get(nuri); - if (!entry) return { state: "none", barrierCrossed: false }; + if (!entry) return { state: "none", barrierCrossed: false, incarnation: 0 }; // `establishing` says an attempt STANDS (it is left true once one has succeeded, see // {@link DocFanOut.establishing}); `realUnsub` says that attempt's call has answered. The // two together are the only thing that separates "asked" from "open" - and recording the // first as if it were the second is the defect this accessor exists to make impossible. const state: DocChannelState = !entry.establishing ? "none" : entry.realUnsub !== null ? "open" : "opening"; - return { state, barrierCrossed: entry.lastState !== null }; + return { state, barrierCrossed: entry.lastState !== null, incarnation: entry.incarnation }; } /** Hand one push to one listener, isolating a throwing handler from the others. */ @@ -633,6 +656,7 @@ export function subscribeDocUnguarded( establishing: false, attempt: 0, lastState: null, + incarnation: ++fanOutsEverCreated, }; fanOuts.set(nuri, entry); } diff --git a/packages/polyfill/test/open-repo.test.ts b/packages/polyfill/test/open-repo.test.ts index d669f5d..fdd1ea0 100644 --- a/packages/polyfill/test/open-repo.test.ts +++ b/packages/polyfill/test/open-repo.test.ts @@ -367,6 +367,104 @@ describe("ensureRepoOpen", () => { expect(getSyncState(doc)).toBe("synced"); }); + // A barrier must not outlive the attempt that produced it. + // + // `subscribe.ts` clears `establishing` on a refused establish and LEAVES the remembered + // `State`, so a call that pushed the barrier and then failed leaves the fan-out with a + // barrier and no channel. Reading `"synced"` off that is AUTHORITY - `index-deposit.ts` + // `noRowsMeansAbsent` calls a zero-row read over it a definitive absence, which destroys a + // deposit - and the registry this replaced answered `"timed-out"` there. Strict is the + // reading: the cost of it is latency, never loss. + it("a barrier pushed by a subscribe that then FAILED is not authority", async () => { + const doc_subscribe = mock(async (_n: string, _s: string, cb: (r: unknown) => void) => { + cb({ V0: { State: {} } }); // the barrier arrives... + throw new Error("RepoNotFound: after the push"); // ...and the call then fails + }); + const ng = { + doc_subscribe, + sparql_query: mock(async () => ({ results: { bindings: [] } })), + doc_create: mock(async () => "did:ng:o:new"), + sparql_update: mock(async () => undefined), + }; + inject(ng as unknown as ReturnType); + setOpenTimeoutForTests(20); + const doc = "did:ng:o:pushed-then-failed"; + + // Something else holds the fan-out, so the refusal does not take the entry with it and the + // remembered barrier survives the attempt that pushed it. + const other = subscribeDoc(doc, () => {}); + await ensureRepoOpen(doc); + await new Promise((r) => setTimeout(r, 10)); + + expect(docChannel(doc).barrierCrossed).toBe(true); // the barrier IS remembered... + expect(docChannel(doc).state).toBe("none"); // ...and nothing stands behind it + expect(getSyncState(doc)).toBe("timed-out"); + other(); + }); + + // `release` does not answer "am I still listening", and treating it as though it did leaves the + // document held by somebody else's listener. + // + // `resetDocSubscriptions()` - which the PUBLISHED `configure()` calls - drops every listener + // and deletes the entry without telling this module, so its handle goes on looking live. If + // anything else subscribes to the same document first, the channel is `"open"` again and a + // short-circuit that asks only "do I hold a handle" skips the open it owes. + it("a second configure() re-holds the document even when something else subscribed first", async () => { + const ngA = makeFakeNgWithSubscribe({}); + inject(ngA); + const doc = "did:ng:o:foreign-first"; + await ensureRepoOpen(doc); + expect(getSyncState(doc)).toBe("synced"); + + const ngB = makeFakeNgWithSubscribe({}); + inject(ngB); // resetDocSubscriptions(): every listener dropped, this module not told + const foreign = subscribeDoc(doc, () => {}); // ...and something else gets there first + await new Promise((r) => setTimeout(r, 10)); + expect(docChannel(doc).state).toBe("open"); + + await ensureRepoOpen(doc); + await new Promise((r) => setTimeout(r, 10)); + + // The document must survive the foreign listener leaving, which it can only do if the open + // above took a listener of its own. + foreign(); + expect(docChannel(doc).state).toBe("open"); + expect(getSyncState(doc)).toBe("synced"); + }); + + // An attempt that has not ANSWERED still STANDS, so a repeat open costs nothing. + // + // A second `doc_subscribe` cannot be placed while one is in flight - the fan-out dedupes it, + // and upstream it would EVICT the first - so making a fresh attempt buys nothing and costs the + // caller the whole bounded fallback again: 8s in production, on every read of a document whose + // subscribe never answers, and the anchored read path awaits it. + it("a subscribe that never answers costs the fallback ONCE, not once per read", async () => { + const doc_subscribe = mock(async () => await new Promise(() => {})); + const ng = { + doc_subscribe, + sparql_query: mock(async () => ({ results: { bindings: [] } })), + doc_create: mock(async () => "did:ng:o:new"), + sparql_update: mock(async () => undefined), + }; + inject(ng as unknown as ReturnType); + setOpenTimeoutForTests(60); + const doc = "did:ng:o:never-answers"; + + let at = Date.now(); + await ensureRepoOpen(doc); + const paid = Date.now() - at; + at = Date.now(); + await ensureRepoOpen(doc); + const again = Date.now() - at; + + expect(paid).toBeGreaterThanOrEqual(50); // the first read waits the barrier out + expect(again).toBeLessThan(25); // every later one does not + expect(doc_subscribe).toHaveBeenCalledTimes(1); // and no second call was ever placed + // Not authority either way: nothing has been heard from this document. + expect(getSyncState(doc)).toBe("timed-out"); + resetOpenedRepos(); + }); + it("the anchored document read triggers doc_subscribe then returns data (integration path)", async () => { const ng = makeFakeNgWithSubscribe({ "did:ng:o:p": [[TYPE, `${FP}Participation`], [`${FP}event`, "did:ng:o:e"]],