fix: trois defauts trouves par la revue adverse sur le registre derive

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.

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<Nuri, BootstrapOpen>();
  * 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<void> {
   // 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<void> {
   // 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<void> {
         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<void> {
         },
         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<void> {
       if (standing !== undefined && standing.release !== null) {
         const superseded = standing.release;
         standing.release = null;
+        standing.incarnation = 0;
         try {
           superseded();
         } catch {

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<Nuri, DocFanOut>();
 
+/** 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);
   }

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<typeof makeFakeNgWithSubscribe>);
+    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<never>(() => {}));
+    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<typeof makeFakeNgWithSubscribe>);
+    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"]],